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..fbfb871 100644 --- a/cmd/cosift/serve_crawl.go +++ b/cmd/cosift/serve_crawl.go @@ -21,6 +21,9 @@ import ( // PeerAuthToken Bearer header. type crawlEnqueueReq struct { URL string `json:"url"` + // Lane optionally targets a frontier lane. Empty keeps the historical + // default (discovered) — parseLaneName("") would mean submitted. + Lane string `json:"lane,omitempty"` } func (s *pebbleHTTP) handleCrawlEnqueue(w http.ResponseWriter, r *http.Request) { @@ -42,7 +45,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..2a8af64 100644 --- a/cmd/cosift/serve_setup.go +++ b/cmd/cosift/serve_setup.go @@ -792,6 +792,9 @@ 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. + 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 +1030,10 @@ 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 politeness counters into /stats + /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/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/config/config.go b/internal/config/config.go index ebf2ad2..5dc5b02 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -118,6 +118,10 @@ 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, 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. MaxBodyBytes int64 `json:"max_body_bytes"` diff --git a/internal/crawler/crawler.go b/internal/crawler/crawler.go index c9709ec..2eac2e4 100644 --- a/internal/crawler/crawler.go +++ b/internal/crawler/crawler.go @@ -104,6 +104,17 @@ type Crawler struct { // re-enqueuing the same dead URLs the sweeper just purged. autoBlocked sync.Map // host (string) → struct{} + // 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 + 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 +543,12 @@ 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, for /stats and /metrics. +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 +829,45 @@ 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 != "" { + // 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)) + } 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 ends the URL's requeue budget. + 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 +1044,37 @@ func (c *Crawler) recordHostResult(host string, success bool) { } } +const ( + // maxRateLimitRequeues bounds one URL's trips through the 429 loop. + maxRateLimitRequeues = 3 + // maxRateLimitDefer caps a host backoff regardless of Retry-After. + 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. +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 +1138,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 +1153,31 @@ 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: 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, 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) - } + // A misconfigured "Crawl-delay: 86400" must not wedge the host. + 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 +1655,41 @@ 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 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 +} + +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 +1736,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..88092e9 100644 --- a/internal/crawler/gate.go +++ b/internal/crawler/gate.go @@ -52,7 +52,17 @@ 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). 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 { + d = min + } g.mu.Lock() now := time.Now() slot, ok := g.next[host] @@ -77,3 +87,17 @@ 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 — the 429/503 +// backoff hook. It never shortens an existing reservation. +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..4661bec 100644 --- a/internal/crawler/gate_test.go +++ b/internal/crawler/gate_test.go @@ -109,6 +109,74 @@ 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", 0) // no-op + 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/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 new file mode 100644 index 0000000..c31efa7 --- /dev/null +++ b/internal/crawler/ratelimit_test.go @@ -0,0 +1,180 @@ +package crawler + +import ( + "context" + "errors" + "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 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) + } + 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") + } +}