From cea78c5d4b7e91ec2d840025c571c53aa85cd5ce Mon Sep 17 00:00:00 2001 From: Andrei Marinescu Date: Mon, 13 Jul 2026 18:09:13 +0300 Subject: [PATCH 1/3] crawler: enforce Crawl-delay, back off on 429/503, surface politeness counters Two field incidents drove this: a temporary IP ban from news.ycombinator.com (robots asks Crawl-delay: 30; we fetched at 1/s regardless) and a 424-response 429 storm from huggingface.co (an 823-item feed crawled with no backoff). - Crawl-delay was parsed but effectively unenforced: the delay was slept per-worker AFTER the host-gate slot was reserved, so concurrent workers still received slots at per_host_delay_ms spacing and the per-host request rate ignored robots entirely. robots.txt is now consulted before the gate and the delay widens the slot reservation itself (hostGate.WaitFor). Fractional Crawl-delay values parse, and a new crawler.max_crawl_delay_ms (default 2 min) clamps hostile/misconfigured values. - 429 (and 503 with Retry-After) no longer error the frontier entry: the host's next slot is deferred by Retry-After (or an exponential fallback, capped at 5 min) and the URL is requeued up to 3 times before failing. Rate-limited responses stay out of hostStats so the success-ratio blacklist can't be poisoned by a healthy-but-busy server. - Claim-time allowlist drops were invisible (imports report success while every item is discarded): now counted, logged once per host, and exposed together with deferral counts via /stats (crawl_dropped_disallowed, crawl_rate_limited_deferrals) and /metrics. - /admin/crawl-enqueue accepts an optional "lane" field (empty keeps the historical discovered default) so bulk feeds don't compete with discovery. - Tests: gate WaitFor/Defer timing under concurrency, fractional/negative Crawl-delay parsing, first fetchRSS coverage (RSS2/Atom/failure paths), 429 requeue-recover and give-up flows, enqueue lane routing. --- cmd/cosift/enqueue_lane_test.go | 48 ++++++++ cmd/cosift/serve_crawl.go | 13 +- cmd/cosift/serve_setup.go | 9 ++ cmd/cosift/serve_stats.go | 14 +++ internal/config/config.go | 6 + internal/crawler/crawler.go | 190 +++++++++++++++++++++++++---- internal/crawler/gate.go | 28 +++++ internal/crawler/gate_test.go | 67 ++++++++++ internal/crawler/ratelimit_test.go | 106 ++++++++++++++++ internal/crawler/robots.go | 5 +- internal/crawler/robots_test.go | 17 +++ internal/crawler/rss_test.go | 81 ++++++++++++ 12 files changed, 559 insertions(+), 25 deletions(-) create mode 100644 cmd/cosift/enqueue_lane_test.go create mode 100644 internal/crawler/ratelimit_test.go create mode 100644 internal/crawler/rss_test.go diff --git a/cmd/cosift/enqueue_lane_test.go b/cmd/cosift/enqueue_lane_test.go new file mode 100644 index 0000000..a8d7d89 --- /dev/null +++ b/cmd/cosift/enqueue_lane_test.go @@ -0,0 +1,48 @@ +package main + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// The lane field must be optional and backward compatible: an omitted/empty +// lane keeps the historical default (Seed → discovered), never the +// parseLaneName("") default (submitted). +func TestHandleCrawlEnqueueLane(t *testing.T) { + var seedCalls, laneCalls int + var gotLane byte + s := &pebbleHTTP{ + crawlSeed: func(string) error { seedCalls++; return nil }, + crawlSeedLane: func(_ string, lane byte) error { laneCalls++; gotLane = lane; return nil }, + } + + post := func(body string) int { + req := httptest.NewRequest(http.MethodPost, "/admin/crawl-enqueue", strings.NewReader(body)) + rec := httptest.NewRecorder() + s.handleCrawlEnqueue(rec, req) + return rec.Code + } + + if code := post(`{"url":"https://example.com/a"}`); code != http.StatusOK { + t.Fatalf("no lane: code %d", code) + } + if seedCalls != 1 || laneCalls != 0 { + t.Errorf("empty lane must use crawlSeed: seed=%d lane=%d", seedCalls, laneCalls) + } + + if code := post(`{"url":"https://example.com/b","lane":"bulk"}`); code != http.StatusOK { + t.Fatalf("bulk lane: code %d", code) + } + if laneCalls != 1 || gotLane != 3 { + t.Errorf("lane=bulk: laneCalls=%d gotLane=%d, want 1/3", laneCalls, gotLane) + } + + if code := post(`{"url":"https://example.com/c","lane":"refresh"}`); code != http.StatusOK { + t.Fatalf("refresh lane: code %d", code) + } + if gotLane != 1 { + t.Errorf("lane=refresh: gotLane=%d, want 1", gotLane) + } +} diff --git a/cmd/cosift/serve_crawl.go b/cmd/cosift/serve_crawl.go index 5039f13..1509380 100644 --- a/cmd/cosift/serve_crawl.go +++ b/cmd/cosift/serve_crawl.go @@ -21,6 +21,11 @@ import ( // PeerAuthToken Bearer header. type crawlEnqueueReq struct { URL string `json:"url"` + // Lane optionally targets a frontier lane ("submitted", "refresh", + // "discovered", "bulk"). Empty keeps the historical default + // (discovered) — note parseLaneName("") would mean submitted, so the + // empty case must not be routed through it. + Lane string `json:"lane,omitempty"` } func (s *pebbleHTTP) handleCrawlEnqueue(w http.ResponseWriter, r *http.Request) { @@ -42,7 +47,13 @@ func (s *pebbleHTTP) handleCrawlEnqueue(w http.ResponseWriter, r *http.Request) writeProblem(w, http.StatusBadRequest, "expected {\"url\": \"...\"}") return } - if err := s.crawlSeed(req.URL); err != nil { + var err error + if req.Lane != "" && s.crawlSeedLane != nil { + err = s.crawlSeedLane(req.URL, parseLaneName(req.Lane)) + } else { + err = s.crawlSeed(req.URL) + } + if err != nil { writeProblem(w, http.StatusInternalServerError, err.Error()) return } diff --git a/cmd/cosift/serve_setup.go b/cmd/cosift/serve_setup.go index 25369a1..aa33aa6 100644 --- a/cmd/cosift/serve_setup.go +++ b/cmd/cosift/serve_setup.go @@ -792,6 +792,10 @@ func (s *pebbleHTTP) startInProcessCrawl(ctx context.Context, ps *store.PebbleSt } // Expose Seed so /admin/crawl-enqueue can hand off forwarded URLs. s.crawlSeed = c.Seed + // expose SeedLane so /admin/crawl-enqueue can honor an explicit lane + // (harvester bulk feeds shouldn't compete with discovery). + s.crawlSeedLane = c.SeedLane + s.crawlPoliteness = c.PolitenessStats // expose SeedSitemap so /admin/sitemap-import can push // sitemap-discovered URLs into the live frontier. s.crawlSeedSitemap = c.SeedSitemap @@ -1027,6 +1031,11 @@ type pebbleHTTP struct { // allowlist (used by /admin/allow-domain for organic HN/Reddit growth). crawlAllowDomain func(domain string) error crawlRecrawl func(ctx context.Context, url string) error + // crawlSeedLane backs the optional "lane" field on /admin/crawl-enqueue. + crawlSeedLane func(url string, lane byte) error + // crawlPoliteness feeds the allowlist-drop / rate-limit-deferral + // counters into /stats and /metrics. + crawlPoliteness func() (droppedDisallowed, rateDeferrals int64) // doc count at startup so /stats can report crawl rate // without persistent counter tables. docs_added = current - startup, diff --git a/cmd/cosift/serve_stats.go b/cmd/cosift/serve_stats.go index b3b3e4b..755dbfb 100644 --- a/cmd/cosift/serve_stats.go +++ b/cmd/cosift/serve_stats.go @@ -455,6 +455,11 @@ func (s *pebbleHTTP) buildStatsBody(ctx context.Context) ([]byte, error) { out["crawl_active"] = true out["docs_added_since_start"] = added out["docs_per_minute"] = rate + if s.crawlPoliteness != nil { + dropped, deferred := s.crawlPoliteness() + out["crawl_dropped_disallowed"] = dropped + out["crawl_rate_limited_deferrals"] = deferred + } } return json.Marshal(out) } @@ -515,6 +520,15 @@ func (s *pebbleHTTP) handleMetrics(w http.ResponseWriter, r *http.Request) { fmt.Fprintf(w, "# HELP cosift_crawl_docs_per_minute Recent crawl rate (docs added per minute, averaged since process start).\n") fmt.Fprintf(w, "# TYPE cosift_crawl_docs_per_minute gauge\n") fmt.Fprintf(w, "cosift_crawl_docs_per_minute %.2f\n", rate) + if s.crawlPoliteness != nil { + dropped, deferred := s.crawlPoliteness() + fmt.Fprintf(w, "# HELP cosift_crawl_dropped_disallowed_total Frontier claims skipped because the domain is not allowlisted.\n") + fmt.Fprintf(w, "# TYPE cosift_crawl_dropped_disallowed_total counter\n") + fmt.Fprintf(w, "cosift_crawl_dropped_disallowed_total %d\n", dropped) + fmt.Fprintf(w, "# HELP cosift_crawl_rate_limited_deferrals_total Host backoffs triggered by 429/503 rate-limit responses.\n") + fmt.Fprintf(w, "# TYPE cosift_crawl_rate_limited_deferrals_total counter\n") + fmt.Fprintf(w, "cosift_crawl_rate_limited_deferrals_total %d\n", deferred) + } } // HyDE cache effectiveness. Hits/misses both monotonic so // Prometheus rate() over these gives cache pressure under load. diff --git a/internal/config/config.go b/internal/config/config.go index ebf2ad2..6e03096 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -118,6 +118,12 @@ type Crawler struct { // "per_host_overrides": {"slow-server.example.com": 5000, "fast-cdn.com": 50} PerHostOverrides map[string]int `json:"per_host_overrides,omitempty"` + // MaxCrawlDelayMs caps the robots.txt Crawl-delay we honor. Without a + // ceiling, a hostile or misconfigured robots.txt ("Crawl-delay: 86400") + // would wedge its host for the whole 24h robots-cache TTL. 0 = default + // (120000 = 2 min). + MaxCrawlDelayMs int `json:"max_crawl_delay_ms,omitempty"` + // MaxBodyBytes caps the response size. Bigger pages get truncated. MaxBodyBytes int64 `json:"max_body_bytes"` diff --git a/internal/crawler/crawler.go b/internal/crawler/crawler.go index c9709ec..bed596f 100644 --- a/internal/crawler/crawler.go +++ b/internal/crawler/crawler.go @@ -104,6 +104,22 @@ type Crawler struct { // re-enqueuing the same dead URLs the sweeper just purged. autoBlocked sync.Map // host (string) → struct{} + // Rate-limit bookkeeping. consec429 counts consecutive 429/503 + // responses per host (reset on any success) and drives the exponential + // backoff when the server sends no Retry-After. rateRequeues bounds the + // retry loop per URL — a host-level bound would fail every URL claimed + // during one sustained throttle burst, even though the host recovers; + // entries are removed on success or give-up, so the map only holds + // currently-throttled URLs. rateDeferrals and droppedDisallowed feed + // /stats; without the drop counter, allowlist fast-skips are invisible + // to operators (imports appear to succeed while every item is + // discarded at claim time). + consec429 sync.Map // host (string) → *atomic.Int32 + rateRequeues sync.Map // url (string) → *atomic.Int32 + rateDeferrals atomic.Int64 + droppedDisallowed atomic.Int64 + dropLoggedHosts sync.Map // host (string) → struct{}, log once per host + // Dynamic allowlist: runtime-promoted domains that allowedDomain accepts // IN ADDITION to the static cfg.IncludeDomains. Lets curated sources // (HN/Reddit harvesters) grow the crawlable set organically — a domain @@ -532,6 +548,13 @@ func (c *Crawler) SeedLane(rawURL string, lane byte) error { return c.store.PushFrontierLane(context.Background(), canon, 0, lane, 1.0) } +// PolitenessStats reports claim-time allowlist drops and rate-limit host +// deferrals since start. Surfaced via /stats and /metrics — allowlist drops +// are otherwise invisible (imports report success while items vanish). +func (c *Crawler) PolitenessStats() (droppedDisallowed, rateDeferrals int64) { + return c.droppedDisallowed.Load(), c.rateDeferrals.Load() +} + // Recrawl re-enqueues a URL even if it was previously crawled. Status flips // to 'queued', attempts resets. Combined with the content-hash dedup in // processClaimed, an unchanged page costs one HTTP request and zero embedding @@ -812,8 +835,48 @@ func (c *Crawler) worker(ctx context.Context, wg *sync.WaitGroup, gate *hostGate continue } err = c.processClaimed(ctx, item, gate) - if u, perr := url.Parse(item.URL); perr == nil && u.Host != "" { - c.recordHostResult(u.Host, err == nil) + host := "" + if u, perr := url.Parse(item.URL); perr == nil { + host = u.Host + } + var rle *rateLimitedError + if errors.As(err, &rle) && host != "" { + // A throttling host is not a failing host. Back the whole host + // off and requeue the URL (bounded per URL) instead of erroring + // it — errored entries are terminal, and it stays out of + // hostStats so the success-ratio blacklist can't be poisoned by + // a healthy-but-busy server. + d := rle.retryAfter + if d == 0 { + d = c.backoffDelay(c.bumpConsec429(host)) + } else { + c.bumpConsec429(host) + } + if d > maxRateLimitDefer { + d = maxRateLimitDefer + } + gate.Defer(host, d) + c.rateDeferrals.Add(1) + rv, _ := c.rateRequeues.LoadOrStore(item.URL, &atomic.Int32{}) + if n := rv.(*atomic.Int32).Add(1); n <= maxRateLimitRequeues { + log.Printf("crawl %s: %v — host deferred %s, requeued (%d/%d)", + item.URL, err, d, n, maxRateLimitRequeues) + _ = c.store.RecrawlURL(ctx, item.URL) + } else { + log.Printf("crawl %s: %v — giving up after %d requeues", item.URL, err, maxRateLimitRequeues) + c.rateRequeues.Delete(item.URL) + _ = c.store.FailFrontier(ctx, item.URL, err.Error()) + } + continue + } + // Any non-throttled outcome (success or terminal error) ends the + // URL's requeue budget tracking. + c.rateRequeues.Delete(item.URL) + if host != "" { + c.recordHostResult(host, err == nil) + if err == nil { + c.resetConsec429(host) + } } if err != nil { log.Printf("crawl %s: %v", item.URL, err) @@ -990,6 +1053,40 @@ func (c *Crawler) recordHostResult(host string, success bool) { } } +const ( + // maxRateLimitRequeues bounds how many times one URL rides the + // 429-requeue loop before it errors out like any other failure. + maxRateLimitRequeues = 3 + // maxRateLimitDefer caps a host backoff regardless of what + // Retry-After asks for — a worker slot blocked behind a deferred + // host must not stall for hours. + maxRateLimitDefer = 5 * time.Minute +) + +func (c *Crawler) bumpConsec429(host string) int32 { + v, _ := c.consec429.LoadOrStore(host, &atomic.Int32{}) + return v.(*atomic.Int32).Add(1) +} + +func (c *Crawler) resetConsec429(host string) { + if v, ok := c.consec429.Load(host); ok { + v.(*atomic.Int32).Store(0) + } +} + +// backoffDelay is the no-Retry-After fallback: per-host delay doubled per +// consecutive 429, from a 1s floor (tests run with PerHostDelayMs=0). +func (c *Crawler) backoffDelay(n int32) time.Duration { + base := time.Duration(c.cfg.PerHostDelayMs) * time.Millisecond + if base < time.Second { + base = time.Second + } + if n > 8 { + n = 8 + } + return base << uint(n) +} + // terminator cancels runCtx once the frontier has stayed empty (no queued, no // in-flight) across several polls. Three empties at 500ms = ~1.5s of quiet. func (c *Crawler) terminator(ctx context.Context, cancel context.CancelFunc) { @@ -1053,6 +1150,12 @@ func (c *Crawler) processClaimed(ctx context.Context, item store.FrontierItem, g // the cursor to reach quality hosts. Returning nil signals success // to the worker so the frontier entry transitions queued → done. if !c.allowedDomain(item.URL) { + c.droppedDisallowed.Add(1) + if u != nil { + if _, logged := c.dropLoggedHosts.LoadOrStore(u.Host, struct{}{}); !logged { + log.Printf("crawl: dropping %s (domain not allowlisted; further drops for this host not logged)", item.URL) + } + } return nil } // After we've tried a host N times @@ -1062,33 +1165,35 @@ func (c *Crawler) processClaimed(ctx context.Context, item store.FrontierItem, g if u != nil && c.isHostBlacklisted(u.Host) { return nil } - if u != nil { - // first time we see a host, fire-and-forget a sitemap - // discovery. Compounds: each new host typically brings hundreds- - // to-thousands of URLs in one batch. - c.maybeAutoSitemap(ctx, u) - if err := gate.Wait(ctx, u.Host); err != nil { - return err - } - } + // robots.txt before the gate: the Crawl-delay verdict must be known when + // the slot is reserved so it can widen the reservation itself. Sleeping + // after Wait (the previous approach) only shifted each worker's fetch by + // a constant while slots kept being granted at the configured interval — + // the effective per-host rate ignored Crawl-delay entirely. Cache hits + // are a map read; a miss fetches robots.txt once per host per TTL. + var robotsDelay time.Duration if c.robots != nil { - allowed, robotsDelay, err := c.robots.Allowed(ctx, item.URL) + allowed, d, err := c.robots.Allowed(ctx, item.URL) if err != nil { return fmt.Errorf("robots: %w", err) } if !allowed { return errors.New("blocked by robots.txt") } - // Honor Crawl-delay if it exceeds our per-host gate's interval. - // use the effective per-host delay (gate's override or - // default), not the global default — operators who set a longer - // override for a host should have that override compared against - // robots.txt's Crawl-delay, not the global setting. - if robotsDelay > 0 && u != nil { - gateDelay := gate.delayFor(u.Host) - if robotsDelay > gateDelay { - sleepCtx(ctx, robotsDelay-gateDelay) - } + // Clamp: a misconfigured "Crawl-delay: 86400" must not wedge the + // host for the whole robots-cache TTL. + if maxDelay := c.maxCrawlDelay(); d > maxDelay { + d = maxDelay + } + robotsDelay = d + } + if u != nil { + // first time we see a host, fire-and-forget a sitemap + // discovery. Compounds: each new host typically brings hundreds- + // to-thousands of URLs in one batch. + c.maybeAutoSitemap(ctx, u) + if err := gate.WaitFor(ctx, u.Host, robotsDelay); err != nil { + return err } } @@ -1566,6 +1671,42 @@ func (c *Crawler) allowedDomain(rawURL string) bool { return false } +func (c *Crawler) maxCrawlDelay() time.Duration { + if c.cfg.MaxCrawlDelayMs > 0 { + return time.Duration(c.cfg.MaxCrawlDelayMs) * time.Millisecond + } + return 2 * time.Minute +} + +// rateLimitedError marks a 429 (or 503 with Retry-After) so the worker can +// back the host off and requeue instead of erroring the frontier entry — +// errored entries are terminal, and treating throttles as failures is what +// turns them into IP bans. retryAfter is 0 when the server sent no header. +type rateLimitedError struct { + code int + retryAfter time.Duration +} + +func (e *rateLimitedError) Error() string { return fmt.Sprintf("http %d (rate limited)", e.code) } + +// parseRetryAfter handles both forms the header allows: delta-seconds and +// HTTP-date. Returns 0 for anything unparseable or in the past. +func parseRetryAfter(v string) time.Duration { + v = strings.TrimSpace(v) + if v == "" { + return 0 + } + if secs, err := strconv.Atoi(v); err == nil && secs > 0 { + return time.Duration(secs) * time.Second + } + if t, err := http.ParseTime(v); err == nil { + if d := time.Until(t); d > 0 { + return d + } + } + return 0 +} + // fetchResult bundles the fetch outcome so processClaimed can distinguish // "fresh body to parse" from "server said 304, nothing changed". type fetchResult struct { @@ -1612,6 +1753,11 @@ func (c *Crawler) fetch(ctx context.Context, u string, prior *store.Document) (* etag: resp.Header.Get("ETag"), lastModified: resp.Header.Get("Last-Modified")}, nil } if resp.StatusCode >= 400 { + retryAfter := parseRetryAfter(resp.Header.Get("Retry-After")) + if resp.StatusCode == http.StatusTooManyRequests || + (resp.StatusCode == http.StatusServiceUnavailable && retryAfter > 0) { + return nil, &rateLimitedError{code: resp.StatusCode, retryAfter: retryAfter} + } return nil, fmt.Errorf("http %d", resp.StatusCode) } ct := resp.Header.Get("Content-Type") diff --git a/internal/crawler/gate.go b/internal/crawler/gate.go index 2d78187..602dc02 100644 --- a/internal/crawler/gate.go +++ b/internal/crawler/gate.go @@ -52,7 +52,20 @@ func (g *hostGate) delayFor(host string) time.Duration { // Wait blocks until this worker's reserved slot for the host arrives. // Returns ctx.Err() if ctx fires first. func (g *hostGate) Wait(ctx context.Context, host string) error { + return g.WaitFor(ctx, host, 0) +} + +// WaitFor is Wait with a per-call minimum delay: the slot is reserved using +// max(delayFor(host), min). This is how robots.txt Crawl-delay actually gets +// enforced — a sleep after Wait would only shift each worker's fetch by a +// constant while slots keep being granted at the configured interval, so the +// effective request rate would be unchanged. The delay has to widen the +// reservation itself. +func (g *hostGate) WaitFor(ctx context.Context, host string, min time.Duration) error { d := g.delayFor(host) + if min > d { + d = min + } g.mu.Lock() now := time.Now() slot, ok := g.next[host] @@ -77,3 +90,18 @@ func (g *hostGate) Wait(ctx context.Context, host string) error { return nil } } + +// Defer pushes the host's next slot at least d into the future. Used when a +// host answers 429/503: already-reserved slots keep their spacing, but no new +// fetch starts before the server-requested backoff has passed. +func (g *hostGate) Defer(host string, d time.Duration) { + if d <= 0 { + return + } + g.mu.Lock() + earliest := time.Now().Add(d) + if g.next[host].Before(earliest) { + g.next[host] = earliest + } + g.mu.Unlock() +} diff --git a/internal/crawler/gate_test.go b/internal/crawler/gate_test.go index 1dbc977..25f67f0 100644 --- a/internal/crawler/gate_test.go +++ b/internal/crawler/gate_test.go @@ -109,6 +109,73 @@ func TestHostGateOverrideSerializesAtOverrideRate(t *testing.T) { } } +func TestHostGateWaitForWidensReservation(t *testing.T) { + // The HN failure mode: gate at 10ms, robots Crawl-delay at 100ms, several + // concurrent workers. WaitFor must space same-host grants at the robots + // interval — a post-Wait sleep would leave grants at the 10ms gate rate. + g := newHostGate(10*time.Millisecond, nil) + ctx := context.Background() + + start := time.Now() + var wg sync.WaitGroup + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _ = g.WaitFor(ctx, "throttled.com", 100*time.Millisecond) + }() + } + wg.Wait() + elapsed := time.Since(start) + + // 4 grants at 100ms spacing: slot 0 immediate, slot 3 at +300ms. + if elapsed < 290*time.Millisecond { + t.Errorf("Crawl-delay not enforced under concurrency: 4 grants in %v, want ≥290ms", elapsed) + } +} + +func TestHostGateWaitForZeroMinIsWait(t *testing.T) { + g := newHostGate(20*time.Millisecond, nil) + ctx := context.Background() + + start := time.Now() + for i := 0; i < 3; i++ { + _ = g.WaitFor(ctx, "plain.com", 0) + } + if elapsed := time.Since(start); elapsed > 150*time.Millisecond { + t.Errorf("min=0 should behave like Wait (~40ms), took %v", elapsed) + } +} + +func TestHostGateDeferPushesNextSlot(t *testing.T) { + g := newHostGate(5*time.Millisecond, nil) + ctx := context.Background() + + _ = g.Wait(ctx, "backoff.com") // burn the free first slot + g.Defer("backoff.com", 150*time.Millisecond) + + start := time.Now() + _ = g.Wait(ctx, "backoff.com") + if elapsed := time.Since(start); elapsed < 120*time.Millisecond { + t.Errorf("Defer ignored: next grant after %v, want ≥120ms", elapsed) + } +} + +func TestHostGateDeferNeverShortens(t *testing.T) { + g := newHostGate(200*time.Millisecond, nil) + ctx := context.Background() + + _ = g.Wait(ctx, "h.com") + _ = g.Wait(ctx, "h.com") // next slot now ~200ms out + g.Defer("h.com", 10*time.Millisecond) + + start := time.Now() + _ = g.Wait(ctx, "h.com") + if elapsed := time.Since(start); elapsed < 150*time.Millisecond { + t.Errorf("Defer shortened an existing reservation: %v", elapsed) + } +} + func TestHostGateNilOverridesIsSafe(t *testing.T) { // Constructing with nil overrides should work exactly like the single-arg // constructor — every host uses the default. This is the path the diff --git a/internal/crawler/ratelimit_test.go b/internal/crawler/ratelimit_test.go new file mode 100644 index 0000000..a1e7e58 --- /dev/null +++ b/internal/crawler/ratelimit_test.go @@ -0,0 +1,106 @@ +package crawler + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/pilot-protocol/cosift/internal/config" +) + +const rateLimitSampleHTML = `Recovered +

Served after the throttle lifted. Enough text to index.

` + +func rateLimitedCrawlerT(t *testing.T, fail429 int64) (*Crawler, *httptest.Server, *atomic.Int64) { + t.Helper() + var hits atomic.Int64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if hits.Add(1) <= fail429 { + w.Header().Set("Retry-After", "1") + w.WriteHeader(http.StatusTooManyRequests) + return + } + w.Header().Set("Content-Type", "text/html") + _, _ = w.Write([]byte(rateLimitSampleHTML)) + })) + t.Cleanup(srv.Close) + + cfg := config.Default().Crawler + cfg.MaxDepth = 0 + cfg.PerHostDelayMs = 0 + cfg.MaxConcurrent = 1 + cfg.RespectRobots = false + return New(cfg, newStoreT(t)).WithEmbedder(&stubEmbedder{dim: 8}), srv, &hits +} + +func TestRateLimitedURLRequeuesAndRecovers(t *testing.T) { + c, srv, hits := rateLimitedCrawlerT(t, 2) + + if err := c.Seed(srv.URL); err != nil { + t.Fatalf("seed: %v", err) + } + if err := c.Run(context.Background()); err != nil { + t.Fatalf("run: %v", err) + } + + doc, err := c.store.GetDocByURL(context.Background(), srv.URL) + if err != nil || doc == nil { + t.Fatalf("doc not indexed after 429 recovery: %v", err) + } + if doc.Title != "Recovered" { + t.Errorf("title: got %q", doc.Title) + } + if got := hits.Load(); got != 3 { + t.Errorf("server hits: got %d want 3 (two 429s, one success)", got) + } + if _, deferred := c.PolitenessStats(); deferred != 2 { + t.Errorf("rate_limited_deferrals: got %d want 2", deferred) + } + // Throttling must not poison the success-ratio blacklist: the two 429s + // stay out of hostStats entirely, only the final success is recorded. + host := strings.TrimPrefix(srv.URL, "http://") + if v, ok := c.hostStats.Load(host); ok { + s := v.(*hostFetchStats) + if s.attempts.Load() != 1 || s.successes.Load() != 1 { + t.Errorf("hostStats polluted by 429s: attempts=%d successes=%d", + s.attempts.Load(), s.successes.Load()) + } + } +} + +func TestRateLimitedURLGivesUpAfterMaxRequeues(t *testing.T) { + c, srv, hits := rateLimitedCrawlerT(t, 1<<30) // never recovers + + if err := c.Seed(srv.URL); err != nil { + t.Fatalf("seed: %v", err) + } + if err := c.Run(context.Background()); err != nil { + t.Fatalf("run: %v", err) + } + + // Initial attempt + maxRateLimitRequeues requeues, then FailFrontier. + if got := hits.Load(); got != maxRateLimitRequeues+1 { + t.Errorf("server hits: got %d want %d", got, maxRateLimitRequeues+1) + } + if doc, _ := c.store.GetDocByURL(context.Background(), srv.URL); doc != nil { + t.Error("doc should not be indexed for a permanently throttled URL") + } +} + +func TestParseRetryAfter(t *testing.T) { + if d := parseRetryAfter("7"); d != 7*time.Second { + t.Errorf("delta-seconds: got %v", d) + } + if d := parseRetryAfter(time.Now().Add(30 * time.Second).UTC().Format(http.TimeFormat)); d < 25*time.Second || d > 30*time.Second { + t.Errorf("http-date: got %v", d) + } + for _, v := range []string{"", "garbage", "-5", time.Now().Add(-time.Minute).UTC().Format(http.TimeFormat)} { + if d := parseRetryAfter(v); d != 0 { + t.Errorf("parseRetryAfter(%q): got %v want 0", v, d) + } + } +} diff --git a/internal/crawler/robots.go b/internal/crawler/robots.go index 7c68a95..69fa637 100644 --- a/internal/crawler/robots.go +++ b/internal/crawler/robots.go @@ -195,8 +195,9 @@ func parseRobots(body string) *robotsRules { if current == nil { continue } - if d, err := strconv.Atoi(val); err == nil && d > 0 { - current.crawlDelay = time.Duration(d) * time.Second + // Fractional values ("Crawl-delay: 0.5") are common in the wild. + if d, err := strconv.ParseFloat(val, 64); err == nil && d > 0 { + current.crawlDelay = time.Duration(d * float64(time.Second)) } case "sitemap": // Sitemap directive is group-independent per the diff --git a/internal/crawler/robots_test.go b/internal/crawler/robots_test.go index f830825..fa8a93c 100644 --- a/internal/crawler/robots_test.go +++ b/internal/crawler/robots_test.go @@ -29,6 +29,23 @@ Allow: / } } +func TestParseRobotsFractionalCrawlDelay(t *testing.T) { + // "Crawl-delay: 0.5" is common in the wild; the previous integer-only + // parse silently dropped it. + r := parseRobots("User-agent: *\nCrawl-delay: 0.5\n") + if got := r.groups[0].crawlDelay; got != 500*time.Millisecond { + t.Errorf("fractional delay: got %v want 500ms", got) + } + r = parseRobots("User-agent: *\nCrawl-delay: 30\n") + if got := r.groups[0].crawlDelay; got != 30*time.Second { + t.Errorf("integer delay: got %v want 30s", got) + } + r = parseRobots("User-agent: *\nCrawl-delay: -3\n") + if got := r.groups[0].crawlDelay; got != 0 { + t.Errorf("negative delay should be ignored: got %v", got) + } +} + func TestRobotsAllowAndDisallow(t *testing.T) { body := `User-agent: * Disallow: /admin/ diff --git a/internal/crawler/rss_test.go b/internal/crawler/rss_test.go new file mode 100644 index 0000000..97ee919 --- /dev/null +++ b/internal/crawler/rss_test.go @@ -0,0 +1,81 @@ +package crawler + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/pilot-protocol/cosift/internal/config" +) + +func feedCrawlerT(t *testing.T, body string) (*Crawler, string) { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/xml") + _, _ = w.Write([]byte(body)) + })) + t.Cleanup(srv.Close) + + cfg := config.Default().Crawler + cfg.PerHostDelayMs = 0 + cfg.RespectRobots = false + return New(cfg, newStoreT(t)), srv.URL +} + +func TestFetchRSS2ItemLinks(t *testing.T) { + c, feed := feedCrawlerT(t, ` +t +a https://example.com/a +bhttps://example.com/b +`) + + urls, err := c.fetchRSS(t.Context(), feed) + if err != nil { + t.Fatalf("fetchRSS: %v", err) + } + want := []string{"https://example.com/a", "https://example.com/b"} + if len(urls) != 2 || urls[0] != want[0] || urls[1] != want[1] { + t.Errorf("got %v want %v", urls, want) + } +} + +func TestFetchRSSAtomPrefersAlternate(t *testing.T) { + c, feed := feedCrawlerT(t, ` +t + + + + + +`) + + urls, err := c.fetchRSS(t.Context(), feed) + if err != nil { + t.Fatalf("fetchRSS: %v", err) + } + want := []string{"https://example.com/post-1", "https://example.com/post-2"} + if len(urls) != 2 || urls[0] != want[0] || urls[1] != want[1] { + t.Errorf("got %v want %v", urls, want) + } +} + +func TestFetchRSSNeitherFormat(t *testing.T) { + c, feed := feedCrawlerT(t, `not a feed`) + + if _, err := c.fetchRSS(t.Context(), feed); err == nil { + t.Error("expected error for non-feed body") + } +} + +func TestFetchRSSHTTPError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusTeapot) + })) + defer srv.Close() + + cfg := config.Default().Crawler + c := New(cfg, newStoreT(t)) + if _, err := c.fetchRSS(t.Context(), srv.URL); err == nil { + t.Error("expected error for HTTP 418 feed") + } +} From 347d5e6b4b777baf79ff3816af4ee348c77613b3 Mon Sep 17 00:00:00 2001 From: Andrei Marinescu Date: Mon, 13 Jul 2026 20:48:57 +0300 Subject: [PATCH 2/3] test: cover the politeness paths codecov flagged - robots Crawl-delay pacing through the gate under concurrent workers, fractional value in anger, and the max_crawl_delay_ms clamp - exponential fallback when a 429 carries no Retry-After; 503 semantics (throttle only with the header) - claim-time drop counter via frontier entries that predate an allowlist change - /stats and /metrics politeness counters (present with the hook, absent without); gate.Defer no-op guard --- cmd/cosift/stats_politeness_test.go | 61 ++++++++++++++++ internal/crawler/gate_test.go | 1 + internal/crawler/politeness_test.go | 105 ++++++++++++++++++++++++++++ internal/crawler/ratelimit_test.go | 74 ++++++++++++++++++++ 4 files changed, 241 insertions(+) create mode 100644 cmd/cosift/stats_politeness_test.go create mode 100644 internal/crawler/politeness_test.go diff --git a/cmd/cosift/stats_politeness_test.go b/cmd/cosift/stats_politeness_test.go new file mode 100644 index 0000000..79281dd --- /dev/null +++ b/cmd/cosift/stats_politeness_test.go @@ -0,0 +1,61 @@ +package main + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestStatsAndMetricsExposePolitenessCounters(t *testing.T) { + f := populatedPebbleStore(t) + srv := f.makeServer(nil) + srv.crawlActive = true + srv.crawlPoliteness = func() (int64, int64) { return 7, 3 } + + body, err := srv.buildStatsBody(context.Background()) + if err != nil { + t.Fatalf("buildStatsBody: %v", err) + } + var out map[string]any + if err := json.Unmarshal(body, &out); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if got := out["crawl_dropped_disallowed"]; got != float64(7) { + t.Errorf("crawl_dropped_disallowed: got %v want 7", got) + } + if got := out["crawl_rate_limited_deferrals"]; got != float64(3) { + t.Errorf("crawl_rate_limited_deferrals: got %v want 3", got) + } + + rec := httptest.NewRecorder() + srv.handleMetrics(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("/metrics: code %d", rec.Code) + } + metrics := rec.Body.String() + for _, want := range []string{ + "cosift_crawl_dropped_disallowed_total 7", + "cosift_crawl_rate_limited_deferrals_total 3", + } { + if !strings.Contains(metrics, want) { + t.Errorf("/metrics missing %q", want) + } + } +} + +func TestStatsOmitsPolitenessWithoutHook(t *testing.T) { + f := populatedPebbleStore(t) + srv := f.makeServer(nil) + srv.crawlActive = true + + body, err := srv.buildStatsBody(context.Background()) + if err != nil { + t.Fatalf("buildStatsBody: %v", err) + } + if strings.Contains(string(body), "crawl_dropped_disallowed") { + t.Error("politeness keys present without a crawlPoliteness hook") + } +} diff --git a/internal/crawler/gate_test.go b/internal/crawler/gate_test.go index 25f67f0..4661bec 100644 --- a/internal/crawler/gate_test.go +++ b/internal/crawler/gate_test.go @@ -152,6 +152,7 @@ func TestHostGateDeferPushesNextSlot(t *testing.T) { ctx := context.Background() _ = g.Wait(ctx, "backoff.com") // burn the free first slot + g.Defer("backoff.com", 0) // no-op g.Defer("backoff.com", 150*time.Millisecond) start := time.Now() diff --git a/internal/crawler/politeness_test.go b/internal/crawler/politeness_test.go new file mode 100644 index 0000000..bc4a0d6 --- /dev/null +++ b/internal/crawler/politeness_test.go @@ -0,0 +1,105 @@ +package crawler + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/pilot-protocol/cosift/internal/config" +) + +func robotsCrawlerT(t *testing.T, crawlDelay string, maxDelayMs int) (*Crawler, *httptest.Server) { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/robots.txt" { + fmt.Fprintf(w, "User-agent: *\nCrawl-delay: %s\n", crawlDelay) + return + } + w.Header().Set("Content-Type", "text/html") + fmt.Fprintf(w, "p

page %s body text

", r.URL.Path) + })) + t.Cleanup(srv.Close) + + cfg := config.Default().Crawler + cfg.MaxDepth = 0 + cfg.PerHostDelayMs = 0 + cfg.MaxConcurrent = 2 + cfg.RespectRobots = true + cfg.MaxCrawlDelayMs = maxDelayMs + return New(cfg, newStoreT(t)).WithEmbedder(&stubEmbedder{dim: 8}), srv +} + +func TestRobotsCrawlDelayPacesConcurrentWorkers(t *testing.T) { + c, srv := robotsCrawlerT(t, "0.5", 0) + for _, p := range []string{"/a", "/b", "/c"} { + if err := c.Seed(srv.URL + p); err != nil { + t.Fatalf("seed %s: %v", p, err) + } + } + + start := time.Now() + if err := c.Run(context.Background()); err != nil { + t.Fatalf("run: %v", err) + } + // Three same-host fetches at 0.5s Crawl-delay: slots at 0 / 0.5s / 1s. + if elapsed := time.Since(start); elapsed < 900*time.Millisecond { + t.Errorf("Crawl-delay not paced through the gate: 3 fetches in %v, want ≥900ms", elapsed) + } + for _, p := range []string{"/a", "/b", "/c"} { + if doc, _ := c.store.GetDocByURL(context.Background(), srv.URL+p); doc == nil { + t.Errorf("%s not indexed", p) + } + } +} + +func TestRobotsCrawlDelayClamped(t *testing.T) { + c, srv := robotsCrawlerT(t, "30", 200) // hostile 30s delay, clamped to 200ms + for _, p := range []string{"/a", "/b"} { + if err := c.Seed(srv.URL + p); err != nil { + t.Fatalf("seed %s: %v", p, err) + } + } + + start := time.Now() + if err := c.Run(context.Background()); err != nil { + t.Fatalf("run: %v", err) + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Errorf("clamp not applied: 2 fetches took %v with a 30s Crawl-delay", elapsed) + } +} + +func TestClaimTimeDropCountedAndTerminal(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Errorf("disallowed URL was fetched: %s", r.URL) + })) + defer srv.Close() + + cfg := config.Default().Crawler + cfg.MaxDepth = 0 + cfg.PerHostDelayMs = 0 + cfg.MaxConcurrent = 1 + cfg.RespectRobots = false + cfg.IncludeDomains = []string{"allowed.example"} + c := New(cfg, newStoreT(t)).WithEmbedder(&stubEmbedder{dim: 8}) + + // Push directly: Seed rejects disallowed domains up front, but frontier + // entries can predate an allowlist change — the claim-time check is + // what handles those. + for _, p := range []string{"/one", "/two"} { + if err := c.store.PushFrontierLane(context.Background(), srv.URL+p, 0, 2, 1.0); err != nil { + t.Fatalf("push: %v", err) + } + } + if err := c.Run(context.Background()); err != nil { + t.Fatalf("run: %v", err) + } + + dropped, _ := c.PolitenessStats() + if dropped != 2 { + t.Errorf("dropped_disallowed: got %d want 2", dropped) + } +} diff --git a/internal/crawler/ratelimit_test.go b/internal/crawler/ratelimit_test.go index a1e7e58..c31efa7 100644 --- a/internal/crawler/ratelimit_test.go +++ b/internal/crawler/ratelimit_test.go @@ -2,6 +2,7 @@ package crawler import ( "context" + "errors" "net/http" "net/http/httptest" "strings" @@ -91,6 +92,79 @@ func TestRateLimitedURLGivesUpAfterMaxRequeues(t *testing.T) { } } +func TestRateLimitedBackoffWithoutRetryAfter(t *testing.T) { + // No Retry-After header → the exponential fallback paces the host. + var hits atomic.Int64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if hits.Add(1) == 1 { + w.WriteHeader(http.StatusTooManyRequests) + return + } + w.Header().Set("Content-Type", "text/html") + _, _ = w.Write([]byte(rateLimitSampleHTML)) + })) + defer srv.Close() + + cfg := config.Default().Crawler + cfg.MaxDepth = 0 + cfg.PerHostDelayMs = 0 + cfg.MaxConcurrent = 1 + cfg.RespectRobots = false + c := New(cfg, newStoreT(t)).WithEmbedder(&stubEmbedder{dim: 8}) + + if err := c.Seed(srv.URL); err != nil { + t.Fatalf("seed: %v", err) + } + start := time.Now() + if err := c.Run(context.Background()); err != nil { + t.Fatalf("run: %v", err) + } + // backoffDelay floors at 1s; first 429 defers the host 1s<<1 = 2s. + if elapsed := time.Since(start); elapsed < 2*time.Second { + t.Errorf("fallback backoff not applied: recovered in %v, want ≥2s", elapsed) + } + if doc, _ := c.store.GetDocByURL(context.Background(), srv.URL); doc == nil { + t.Error("doc not indexed after fallback backoff") + } +} + +func TestFetch503RetryAfterSemantics(t *testing.T) { + // 503 WITH Retry-After is a throttle; without the header it stays a + // generic failure. + for _, withHeader := range []bool{true, false} { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if withHeader { + w.Header().Set("Retry-After", "3") + } + w.WriteHeader(http.StatusServiceUnavailable) + })) + cfg := config.Default().Crawler + c := New(cfg, newStoreT(t)) + _, err := c.fetch(context.Background(), srv.URL, nil) + var rle *rateLimitedError + if got := errors.As(err, &rle); got != withHeader { + t.Errorf("503 withHeader=%v: rateLimitedError=%v (err %v)", withHeader, got, err) + } + if withHeader && rle.retryAfter != 3*time.Second { + t.Errorf("retryAfter: got %v want 3s", rle.retryAfter) + } + srv.Close() + } +} + +func TestMaxCrawlDelayDefaultAndOverride(t *testing.T) { + cfg := config.Default().Crawler + c := New(cfg, newStoreT(t)) + if d := c.maxCrawlDelay(); d != 2*time.Minute { + t.Errorf("default clamp: got %v want 2m", d) + } + cfg.MaxCrawlDelayMs = 500 + c = New(cfg, newStoreT(t)) + if d := c.maxCrawlDelay(); d != 500*time.Millisecond { + t.Errorf("configured clamp: got %v want 500ms", d) + } +} + func TestParseRetryAfter(t *testing.T) { if d := parseRetryAfter("7"); d != 7*time.Second { t.Errorf("delta-seconds: got %v", d) From 81e71f36ba5bcd90ae9f42c4b719b5b539be0c07 Mon Sep 17 00:00:00 2001 From: Andrei Marinescu Date: Mon, 13 Jul 2026 21:01:53 +0300 Subject: [PATCH 3/3] comments: trim to essentials --- cmd/cosift/serve_crawl.go | 6 ++-- cmd/cosift/serve_setup.go | 6 ++-- internal/config/config.go | 6 ++-- internal/crawler/crawler.go | 57 +++++++++++++------------------------ internal/crawler/gate.go | 12 +++----- 5 files changed, 30 insertions(+), 57 deletions(-) diff --git a/cmd/cosift/serve_crawl.go b/cmd/cosift/serve_crawl.go index 1509380..fbfb871 100644 --- a/cmd/cosift/serve_crawl.go +++ b/cmd/cosift/serve_crawl.go @@ -21,10 +21,8 @@ import ( // PeerAuthToken Bearer header. type crawlEnqueueReq struct { URL string `json:"url"` - // Lane optionally targets a frontier lane ("submitted", "refresh", - // "discovered", "bulk"). Empty keeps the historical default - // (discovered) — note parseLaneName("") would mean submitted, so the - // empty case must not be routed through it. + // Lane optionally targets a frontier lane. Empty keeps the historical + // default (discovered) — parseLaneName("") would mean submitted. Lane string `json:"lane,omitempty"` } diff --git a/cmd/cosift/serve_setup.go b/cmd/cosift/serve_setup.go index aa33aa6..2a8af64 100644 --- a/cmd/cosift/serve_setup.go +++ b/cmd/cosift/serve_setup.go @@ -792,8 +792,7 @@ func (s *pebbleHTTP) startInProcessCrawl(ctx context.Context, ps *store.PebbleSt } // Expose Seed so /admin/crawl-enqueue can hand off forwarded URLs. s.crawlSeed = c.Seed - // expose SeedLane so /admin/crawl-enqueue can honor an explicit lane - // (harvester bulk feeds shouldn't compete with discovery). + // expose SeedLane so /admin/crawl-enqueue can honor an explicit lane. s.crawlSeedLane = c.SeedLane s.crawlPoliteness = c.PolitenessStats // expose SeedSitemap so /admin/sitemap-import can push @@ -1033,8 +1032,7 @@ type pebbleHTTP struct { crawlRecrawl func(ctx context.Context, url string) error // crawlSeedLane backs the optional "lane" field on /admin/crawl-enqueue. crawlSeedLane func(url string, lane byte) error - // crawlPoliteness feeds the allowlist-drop / rate-limit-deferral - // counters into /stats and /metrics. + // crawlPoliteness feeds the politeness counters into /stats + /metrics. crawlPoliteness func() (droppedDisallowed, rateDeferrals int64) // doc count at startup so /stats can report crawl rate diff --git a/internal/config/config.go b/internal/config/config.go index 6e03096..5dc5b02 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -118,10 +118,8 @@ type Crawler struct { // "per_host_overrides": {"slow-server.example.com": 5000, "fast-cdn.com": 50} PerHostOverrides map[string]int `json:"per_host_overrides,omitempty"` - // MaxCrawlDelayMs caps the robots.txt Crawl-delay we honor. Without a - // ceiling, a hostile or misconfigured robots.txt ("Crawl-delay: 86400") - // would wedge its host for the whole 24h robots-cache TTL. 0 = default - // (120000 = 2 min). + // MaxCrawlDelayMs caps the robots.txt Crawl-delay we honor, so a + // misconfigured robots.txt can't wedge its host. 0 = default (2 min). MaxCrawlDelayMs int `json:"max_crawl_delay_ms,omitempty"` // MaxBodyBytes caps the response size. Bigger pages get truncated. diff --git a/internal/crawler/crawler.go b/internal/crawler/crawler.go index bed596f..2eac2e4 100644 --- a/internal/crawler/crawler.go +++ b/internal/crawler/crawler.go @@ -104,16 +104,11 @@ type Crawler struct { // re-enqueuing the same dead URLs the sweeper just purged. autoBlocked sync.Map // host (string) → struct{} - // Rate-limit bookkeeping. consec429 counts consecutive 429/503 - // responses per host (reset on any success) and drives the exponential - // backoff when the server sends no Retry-After. rateRequeues bounds the - // retry loop per URL — a host-level bound would fail every URL claimed - // during one sustained throttle burst, even though the host recovers; - // entries are removed on success or give-up, so the map only holds - // currently-throttled URLs. rateDeferrals and droppedDisallowed feed - // /stats; without the drop counter, allowlist fast-skips are invisible - // to operators (imports appear to succeed while every item is - // discarded at claim time). + // Rate-limit bookkeeping. consec429 (per host, reset on success) drives + // the no-Retry-After exponential backoff. rateRequeues bounds retries + // per URL — a host-level bound would fail every URL claimed during one + // sustained throttle burst; entries drop on any terminal outcome, so + // the map only holds currently-throttled URLs. consec429 sync.Map // host (string) → *atomic.Int32 rateRequeues sync.Map // url (string) → *atomic.Int32 rateDeferrals atomic.Int64 @@ -549,8 +544,7 @@ func (c *Crawler) SeedLane(rawURL string, lane byte) error { } // PolitenessStats reports claim-time allowlist drops and rate-limit host -// deferrals since start. Surfaced via /stats and /metrics — allowlist drops -// are otherwise invisible (imports report success while items vanish). +// deferrals since start, for /stats and /metrics. func (c *Crawler) PolitenessStats() (droppedDisallowed, rateDeferrals int64) { return c.droppedDisallowed.Load(), c.rateDeferrals.Load() } @@ -841,11 +835,9 @@ func (c *Crawler) worker(ctx context.Context, wg *sync.WaitGroup, gate *hostGate } var rle *rateLimitedError if errors.As(err, &rle) && host != "" { - // A throttling host is not a failing host. Back the whole host - // off and requeue the URL (bounded per URL) instead of erroring - // it — errored entries are terminal, and it stays out of - // hostStats so the success-ratio blacklist can't be poisoned by - // a healthy-but-busy server. + // Defer the host and requeue instead of erroring: errored + // entries are terminal, and throttles stay out of hostStats so + // the success-ratio blacklist isn't poisoned by a busy server. d := rle.retryAfter if d == 0 { d = c.backoffDelay(c.bumpConsec429(host)) @@ -869,8 +861,7 @@ func (c *Crawler) worker(ctx context.Context, wg *sync.WaitGroup, gate *hostGate } continue } - // Any non-throttled outcome (success or terminal error) ends the - // URL's requeue budget tracking. + // Any non-throttled outcome ends the URL's requeue budget. c.rateRequeues.Delete(item.URL) if host != "" { c.recordHostResult(host, err == nil) @@ -1054,12 +1045,9 @@ func (c *Crawler) recordHostResult(host string, success bool) { } const ( - // maxRateLimitRequeues bounds how many times one URL rides the - // 429-requeue loop before it errors out like any other failure. + // maxRateLimitRequeues bounds one URL's trips through the 429 loop. maxRateLimitRequeues = 3 - // maxRateLimitDefer caps a host backoff regardless of what - // Retry-After asks for — a worker slot blocked behind a deferred - // host must not stall for hours. + // maxRateLimitDefer caps a host backoff regardless of Retry-After. maxRateLimitDefer = 5 * time.Minute ) @@ -1075,7 +1063,7 @@ func (c *Crawler) resetConsec429(host string) { } // backoffDelay is the no-Retry-After fallback: per-host delay doubled per -// consecutive 429, from a 1s floor (tests run with PerHostDelayMs=0). +// consecutive 429, from a 1s floor. func (c *Crawler) backoffDelay(n int32) time.Duration { base := time.Duration(c.cfg.PerHostDelayMs) * time.Millisecond if base < time.Second { @@ -1165,12 +1153,9 @@ func (c *Crawler) processClaimed(ctx context.Context, item store.FrontierItem, g if u != nil && c.isHostBlacklisted(u.Host) { return nil } - // robots.txt before the gate: the Crawl-delay verdict must be known when - // the slot is reserved so it can widen the reservation itself. Sleeping - // after Wait (the previous approach) only shifted each worker's fetch by - // a constant while slots kept being granted at the configured interval — - // the effective per-host rate ignored Crawl-delay entirely. Cache hits - // are a map read; a miss fetches robots.txt once per host per TTL. + // robots.txt before the gate: Crawl-delay must widen the slot + // reservation itself — sleeping after Wait leaves the effective + // per-host rate at the gate interval. Cache hits are a map read. var robotsDelay time.Duration if c.robots != nil { allowed, d, err := c.robots.Allowed(ctx, item.URL) @@ -1180,8 +1165,7 @@ func (c *Crawler) processClaimed(ctx context.Context, item store.FrontierItem, g if !allowed { return errors.New("blocked by robots.txt") } - // Clamp: a misconfigured "Crawl-delay: 86400" must not wedge the - // host for the whole robots-cache TTL. + // A misconfigured "Crawl-delay: 86400" must not wedge the host. if maxDelay := c.maxCrawlDelay(); d > maxDelay { d = maxDelay } @@ -1678,10 +1662,9 @@ func (c *Crawler) maxCrawlDelay() time.Duration { return 2 * time.Minute } -// rateLimitedError marks a 429 (or 503 with Retry-After) so the worker can -// back the host off and requeue instead of erroring the frontier entry — -// errored entries are terminal, and treating throttles as failures is what -// turns them into IP bans. retryAfter is 0 when the server sent no header. +// rateLimitedError marks a 429 (or 503 with Retry-After) so the worker backs +// the host off and requeues instead of erroring the frontier entry. +// retryAfter is 0 when the server sent no header. type rateLimitedError struct { code int retryAfter time.Duration diff --git a/internal/crawler/gate.go b/internal/crawler/gate.go index 602dc02..88092e9 100644 --- a/internal/crawler/gate.go +++ b/internal/crawler/gate.go @@ -56,11 +56,8 @@ func (g *hostGate) Wait(ctx context.Context, host string) error { } // WaitFor is Wait with a per-call minimum delay: the slot is reserved using -// max(delayFor(host), min). This is how robots.txt Crawl-delay actually gets -// enforced — a sleep after Wait would only shift each worker's fetch by a -// constant while slots keep being granted at the configured interval, so the -// effective request rate would be unchanged. The delay has to widen the -// reservation itself. +// max(delayFor(host), min). robots.txt Crawl-delay is enforced here — a sleep +// after Wait would leave the effective per-host rate at the gate interval. func (g *hostGate) WaitFor(ctx context.Context, host string, min time.Duration) error { d := g.delayFor(host) if min > d { @@ -91,9 +88,8 @@ func (g *hostGate) WaitFor(ctx context.Context, host string, min time.Duration) } } -// Defer pushes the host's next slot at least d into the future. Used when a -// host answers 429/503: already-reserved slots keep their spacing, but no new -// fetch starts before the server-requested backoff has passed. +// Defer pushes the host's next slot at least d into the future — the 429/503 +// backoff hook. It never shortens an existing reservation. func (g *hostGate) Defer(host string, d time.Duration) { if d <= 0 { return