diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 1ebeb42..b709470 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -9,3 +9,4 @@ One row per task session that ends in a commit. Newest last. | p5a/03-dcache | 2026-09-25 | done | 1 | gate: ok | evict() tie-break | Added the pdf_doc/pdf_page tables, `PDFEntry`/`PDFKey`/`PutPDF`/`GetPDF`/`TouchPDF` (one-tx write deleting old pages first, page-ordered read, 304 touch moves only fetched_at), PDFH/PDFM counters, `Size()` including pdf_doc nbytes. One deviation: extended the eviction union with `pdf_doc` and added `rowid ASC` to the ORDER BY so same-second puts evict in insertion order (deterministic whole-document eviction); without it TestPDFEviction is order-dependent. Left Stats() signature unchanged for task 07. Given test byte-identical. | | p5a/04-pdfpool | 2026-09-25 | done | 1 | gate: ok | none | Implemented the generic scheduler. `Pool` holds one mutex, a `jobs` map (key→in-flight job), a `callerJobs` map (caller→their job), and a FIFO `waiters` line. `Do` applies the six rules in order: join (counted in `Waiting`, no slot, allowed while the caller holds another job, `fn` may be nil), caller-busy, then slot/queue/refuse. A job's result is broadcast to every sharer via a `done` channel; results are not cached (key forgotten on finish). `fn` runs under `context.WithoutCancel` so the flight outlives a caller that leaves. Panic in `fn` is recovered into an error, freed the slot and caller. One implementation finding (not a plan deviation): queued jobs must not be counted in `Running()`, so `startLocked` only registers a job and a separate `takeSlotLocked`/`grant` increments the slot count — abandoning a queued starter (`abandon`) releases sharers without touching the slot. Given test byte-identical. | | p5a/05-pdfdoc | 2026-09-25 | done | 1 | gate: ok | none | Implemented the render package with no I/O on a `*pdftext.Doc`. `Marker`/`Full` print each extracted page as `--- page N ---`, text trimmed, joined by a blank line. `Render` dispatches grep → page range → whole. Whole fits in OutCap runes, else a map: title (or `Untitled PDF`)/pages/`~tokens` line with `pages 1-E extracted` only when fewer extracted than `Pages`, a cuttable `## Outline` (two-space indent per level below 1, cut before the header/instruction), the instruction line, and whole leading pages within OutCap/4 (at least page 1, cut to fit). Range parses `a-b`/`a`/`a-`, wraps out-of-range in `ErrBadPages` with `pages out of range (1-E)`, clamps `b`, caps at 100 pages and OutCap runes, and notes the rest as `pages="k-b"`. Grep is RE2 with smart-case `(?i)` prefix, 200-byte cap, per-line matches with previous/next page context in `p. P-`/`p. P:` form, first 100 shown, total counted; RE2 keeps the `(a+)+$` test linear. Every length is in runes. Given test byte-identical; make gate ok. | +| p5a/06-api-pdf | 2026-09-26 | done | resumed | gate: ok (7 of 10 PDF tests; 3 are task 07's) | none | Served PDFs through `/v1/fetch` (clean) and `/v2/scrape`. Added `RetryAfter int` to `fetch.Error` and a `Retry-After` header line guarded by `> 0`. Added `PDFRunner`/`PDFPool`/`PDFTTL`/`PDFMaxQuestionChunks` Server fields (PDFPool built once via `sync.Once` only when nil, so the test's direct override is respected) and `pdfPoolOnce`. Routed `clean.Kind` in `doFetch`/`cleanFetched` (html/json/text switched, "other"→415, "pdf"→extractPDF); `doFetch`'s switch routes "text" explicitly so a PDF never runs `clean.Text` first. New `internal/api/pdf.go`: `DefaultPDFTTL`/`DefaultPDFMaxQuestionChunks` (task 07), `runner()`/`pool()` defaults, `extractPDF` (pool key `callerClass|SHA256`, two-value assert, `fmt.Errorf("pdf: bad result %T")`), `pdfError` (422 for the five pdftext errors + panic default, 503+RetryAfter 10 for ErrBusy, 429 for ErrCallerBusy), `renderPDF` (`pdfdoc.Render`, whole text or map). `doFetch` returns the 4th `doc *pdftext.Doc`; digest uses `pdfdoc.Full(doc)` (computed once) as content, clean uses `renderPDF.Content`, envelope `pdf` object has pages/title/outline/extracted/first/last/map/grep_hits/note, raw mode→400. Scrape markdown=`renderPDF.Content`, html="", `metadata.numPages`=doc.Pages. 7 of 10 given tests pass under -race -count=3; the 3 failures (`TestPDFPagesAndGrep`, `TestPDFCache`, `TestPDFQuestionTooLong`) are task 07 tests that reference fields not in this task, exactly as the task doc (lines 145, 157-158) predicts; nothing else fails. Given files byte-identical. | diff --git a/internal/api/api.go b/internal/api/api.go index b0960c2..953759d 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -10,10 +10,12 @@ import ( "log/slog" "net/http" "net/url" + "strconv" "strings" "sync" "sync/atomic" "time" + "unicode/utf8" "git.wntrmute.dev/kyle/sift/internal/auth" "git.wntrmute.dev/kyle/sift/internal/clean" @@ -23,6 +25,9 @@ import ( "git.wntrmute.dev/kyle/sift/internal/fleetctx" "git.wntrmute.dev/kyle/sift/internal/guard" "git.wntrmute.dev/kyle/sift/internal/metrics" + "git.wntrmute.dev/kyle/sift/internal/pdfdoc" + "git.wntrmute.dev/kyle/sift/internal/pdfpool" + "git.wntrmute.dev/kyle/sift/internal/pdftext" "git.wntrmute.dev/kyle/sift/internal/recover" "git.wntrmute.dev/kyle/sift/internal/search" "git.wntrmute.dev/kyle/sift/internal/selfip" @@ -44,8 +49,13 @@ type Server struct { OutCap int SelfIP *selfip.Discoverer Digest *digest.Engine - Search *search.Engine - Ctx *fleetctx.Client + + PDFRunner *pdftext.Runner + PDFPool *pdfpool.Pool + PDFTTL time.Duration + PDFMaxQuestionChunks int + Search *search.Engine + Ctx *fleetctx.Client // PermitRestricted is context.permit_restricted: those callers never // have a fetch host written to the log (host=-). Independent of Honcho. PermitRestricted []string @@ -56,6 +66,8 @@ type Server struct { per map[string]chan struct{} reqN, blockN, scrapeN atomic.Uint64 + + pdfPoolOnce sync.Once } func New(version string, g *guard.Guard, f *fetch.Fetcher, keys *auth.Store, log *slog.Logger) *Server { @@ -300,6 +312,10 @@ type v1Req struct { Context *bool `json:"context"` JQ string `json:"jq"` Recover bool `json:"recover"` + + // Pages narrows a PDF digest to "2-3"; Grep matches lines in it. Optional; task-07. + Pages string `json:"pages"` + Grep string `json:"grep"` } func (s *Server) caller(w http.ResponseWriter, r *http.Request) *auth.Caller { @@ -357,12 +373,23 @@ func (s *Server) scrape(w http.ResponseWriter, r *http.Request) { } } s.scrapeN.Add(1) - res, cl, ferr := s.doFetch(r, c, req.URL, "", onlyMain) + res, cl, doc, ferr := s.doFetch(r, c, req.URL, "", onlyMain) if ferr != nil { s.writeFetchErr(w, c, req.URL, ferr) return } md, htmlOut := cl.Markdown, "" + pages := 0 + if doc != nil { + rp, err := s.renderPDF(doc) + if err != nil { + s.writeFetchErr(w, c, req.URL, err) + return + } + md = rp.Content + htmlOut = "" + pages = doc.Pages + } if wantHTML { htmlOut = cl.HTML } @@ -371,7 +398,7 @@ func (s *Server) scrape(w http.ResponseWriter, r *http.Request) { } rawTok := len(res.Body) / 4 retTok := len(md) / 4 - s.Log.Info("scrape", "caller", c.Name, "host", s.logHost(c, res.FinalURL), "status", res.Status, "bytes", len(res.Body)) + s.Log.Info("scrape", "caller", c.Name, "host", s.logHost(c, res.FinalURL), "status", res.Status, "bytes", len(res.Body), "pdf_pages", pages) writeJSON(w, 200, map[string]any{ "success": true, "data": map[string]any{ @@ -382,6 +409,7 @@ func (s *Server) scrape(w http.ResponseWriter, r *http.Request) { "sourceURL": res.FinalURL, "statusCode": res.Status, "contentType": res.ContentType, + "numPages": pages, "sift": map[string]any{ "hash": res.SHA256(), "fetchedAt": res.FetchedAt.Format(time.RFC3339), @@ -429,12 +457,12 @@ func (s *Server) v1fetch(w http.ResponseWriter, r *http.Request) { t0 := time.Now() onlyMain := true - res, cl, ferr := s.doFetch(r, c, req.URL, req.JQ, onlyMain) + res, cl, doc, ferr := s.doFetch(r, c, req.URL, req.JQ, onlyMain) var snap *recover.Result if ferr != nil && req.Recover && recoverWorthy(ferr) { sr, rerr := recover.Try(r.Context(), s.Fetch, req.URL) if rerr == nil && sr != nil && sr.Res != nil { - res, cl, ferr = s.cleanFetched(sr.Res, req.JQ, onlyMain) + res, cl, doc, ferr = s.cleanFetched(sr.Res, req.JQ, onlyMain) snap = sr } } @@ -444,16 +472,50 @@ func (s *Server) v1fetch(w http.ResponseWriter, r *http.Request) { } fetchMS := time.Since(t0).Milliseconds() - content := cl.Markdown - if cl.Kind == "json" { - content = cl.HTML - } - if mode == "raw" { - content = string(res.Body) - if len(content) > s.OutCap { - content = clean.TruncateUTF8(content, s.OutCap) - cl.Trunc = true - cl.Warning = "truncated to output cap" + var docLen, pdfPages int + var content string + var pdfMap map[string]any + if doc != nil { + pdfPages = doc.Pages + if mode == "raw" { + writeErr(w, 400, "raw mode does not apply to PDFs", "") + return + } + rp, err := s.renderPDF(doc) + if err != nil { + s.writeFetchErr(w, c, req.URL, err) + return + } + full := pdfdoc.Full(doc) + docLen = utf8.RuneCountInString(full) + if mode == "digest" { + content = full + } else { + content = rp.Content + } + pdfMap = map[string]any{ + "pages": doc.Pages, + "title": doc.Title, + "outline": doc.Outline, + "extracted": len(doc.Text), + "first": rp.First, + "last": rp.Last, + "map": rp.Map, + "grep_hits": rp.GrepHits, + "note": rp.Note, + } + } else { + content = cl.Markdown + if cl.Kind == "json" { + content = cl.HTML + } + if mode == "raw" { + content = string(res.Body) + if len(content) > s.OutCap { + content = clean.TruncateUTF8(content, s.OutCap) + cl.Trunc = true + cl.Warning = "truncated to output cap" + } } } warn := []string{} @@ -518,7 +580,7 @@ func (s *Server) v1fetch(w http.ResponseWriter, r *http.Request) { } s.Log.Info("fetch", "caller", c.Name, "host", s.logHost(c, res.FinalURL), "mode", mode, "q_len", qLen, "raw_tokens", len(res.Body)/4, "returned_tokens", len(content)/4, - "digest_ms", digestMS, "fallback", fb) + "pdf_pages", pdfPages, "digest_ms", digestMS, "fallback", fb) src := map[string]any{ "url": res.URL, @@ -549,12 +611,15 @@ func (s *Server) v1fetch(w http.ResponseWriter, r *http.Request) { "source": src, "warnings": warn, "sizes": map[string]int{ - "raw_tokens": len(res.Body) / 4, + "raw_tokens": docLen / 4, "clean_tokens": digest.EstTokens(cl.Markdown), "returned_tokens": digest.EstTokens(content), }, "timings": map[string]int64{"fetch_ms": fetchMS, "digest_ms": digestMS, "context_ms": ctxMS}, } + if pdfMap != nil { + out["pdf"] = pdfMap + } if digMeta != nil { out["digest"] = digMeta } @@ -585,13 +650,11 @@ func headings(md string) []string { return h } -func (s *Server) cleanFetched(res *fetch.Resource, jq string, onlyMain bool) (*fetch.Resource, *clean.Result, error) { - kind := clean.MediaKind(res.ContentType) - if kind == "sniff" { - kind = clean.Sniff(res.Body) - } - if kind == "other" { - return nil, nil, &fetch.Error{Status: 415, Hint: "browser", Msg: "unsupported content type"} +func (s *Server) cleanFetched(res *fetch.Resource, jq string, onlyMain bool) (*fetch.Resource, *clean.Result, *pdftext.Doc, error) { + kind := clean.Kind(res.ContentType, res.Body) + if kind == "other" || kind == "pdf" { + // The recover path (archive snapshots) never serves a PDF. + return nil, nil, nil, &fetch.Error{Status: 415, Hint: "browser", Msg: "unsupported content type"} } var cl *clean.Result var cerr error @@ -604,31 +667,28 @@ func (s *Server) cleanFetched(res *fetch.Resource, jq string, onlyMain bool) (*f cl, cerr = clean.Text(res, s.OutCap) } if cerr != nil { - return nil, nil, &fetch.Error{Status: 422, Msg: cerr.Error()} + return nil, nil, nil, &fetch.Error{Status: 422, Msg: cerr.Error()} } - return res, cl, nil + return res, cl, nil, nil } -func (s *Server) doFetch(r *http.Request, c *auth.Caller, rawURL, jq string, onlyMain bool) (*fetch.Resource, *clean.Result, error) { +func (s *Server) doFetch(r *http.Request, c *auth.Caller, rawURL, jq string, onlyMain bool) (*fetch.Resource, *clean.Result, *pdftext.Doc, error) { if err := s.acquire(c.Name, rawURL); err != nil { - return nil, nil, err + return nil, nil, nil, err } defer s.release(c.Name, rawURL) res, err := s.Fetch.Fetch(r.Context(), rawURL) if err != nil { - return nil, nil, err - } - kind := clean.MediaKind(res.ContentType) - if kind == "sniff" { - kind = clean.Sniff(res.Body) + return nil, nil, nil, err } + kind := clean.Kind(res.ContentType, res.Body) if kind == "other" { ct := res.ContentType if ct == "" { ct = "unknown" } - return nil, nil, &fetch.Error{Status: 415, Hint: "browser", Msg: "unsupported content type " + ct + "; hint: browser"} + return nil, nil, nil, &fetch.Error{Status: 415, Hint: "browser", Msg: "unsupported content type " + ct + "; hint: browser"} } var cl *clean.Result var cerr error @@ -637,17 +697,25 @@ func (s *Server) doFetch(r *http.Request, c *auth.Caller, rawURL, jq string, onl cl, cerr = clean.HTML(res, onlyMain, s.OutCap) case "json": cl, cerr = clean.JSON(res, jq, s.OutCap) - default: + case "text": cl, cerr = clean.Text(res, s.OutCap) } if cerr != nil { var tl *clean.TooLargeError if errors.As(cerr, &tl) { - return nil, nil, &fetch.Error{Status: 400, Msg: tl.Error()} + return nil, nil, nil, &fetch.Error{Status: 400, Msg: tl.Error()} } - return nil, nil, &fetch.Error{Status: 422, Msg: cerr.Error()} + return nil, nil, nil, &fetch.Error{Status: 422, Msg: cerr.Error()} } - return res, cl, nil + var doc *pdftext.Doc + if kind == "pdf" { + doc, err = s.extractPDF(r.Context(), c, res) + if err != nil { + return nil, nil, nil, err + } + cl = &clean.Result{Kind: "pdf", Title: doc.Title} + } + return res, cl, doc, nil } func (s *Server) writeFetchErr(w http.ResponseWriter, c *auth.Caller, rawURL string, err error) { @@ -670,6 +738,9 @@ func (s *Server) writeFetchErr(w http.ResponseWriter, c *auth.Caller, rawURL str } else { s.Log.Warn("fetch failed", "caller", name, "host", host, "reason", s.logReason(c, fe), "status", fe.Status) } + if fe.RetryAfter > 0 { + w.Header().Set("Retry-After", strconv.Itoa(fe.RetryAfter)) + } writeErr(w, fe.Status, fe.Msg, fe.Hint) return } diff --git a/internal/api/pdf.go b/internal/api/pdf.go new file mode 100644 index 0000000..7efa38a --- /dev/null +++ b/internal/api/pdf.go @@ -0,0 +1,86 @@ +package api + +import ( + "context" + "errors" + "fmt" + "time" + + "git.wntrmute.dev/kyle/sift/internal/auth" + "git.wntrmute.dev/kyle/sift/internal/fetch" + "git.wntrmute.dev/kyle/sift/internal/pdfdoc" + "git.wntrmute.dev/kyle/sift/internal/pdfpool" + "git.wntrmute.dev/kyle/sift/internal/pdftext" +) + +const ( + // DefaultPDFTTL bounds a cached PDF's life. Task 07; declared now so the test compiles. + DefaultPDFTTL = 24 * time.Hour + // DefaultPDFMaxQuestionChunks caps a digest question's length. Task 07; declared now. + DefaultPDFMaxQuestionChunks = 200 +) + +// runner returns the configured extraction child, or a zero Runner that runs +// /proc/self/exe pdftext. +func (s *Server) runner() *pdftext.Runner { + if s.PDFRunner != nil { + return s.PDFRunner + } + return &pdftext.Runner{} +} + +// pool returns the extraction pool, built once and only when none is configured. +func (s *Server) pool() *pdfpool.Pool { + if s.PDFPool == nil { + s.pdfPoolOnce.Do(func() { + s.PDFPool = pdfpool.New(2, 4, 20*time.Second) + }) + } + return s.PDFPool +} + +// extractPDF runs one extraction under the pool, coalesced by document hash and +// bounded to one per caller. +func (s *Server) extractPDF(ctx context.Context, c *auth.Caller, res *fetch.Resource) (*pdftext.Doc, error) { + fn := func(jobCtx context.Context) (any, error) { + return s.runner().Run(jobCtx, res.Body) + } + v, err := s.pool().Do(ctx, s.callerClass(c)+"|"+res.SHA256(), c.Name, fn) + if err != nil { + return nil, pdfError(err) + } + doc, ok := v.(*pdftext.Doc) + if !ok { + return nil, fmt.Errorf("pdf: bad result %T", v) + } + return doc, nil +} + +// pdfError maps every PDF-path error to the status sift returns to its caller; +// a PDF never produces a 500. +func pdfError(err error) error { + switch { + case errors.Is(err, pdftext.ErrNoText): + return &fetch.Error{Status: 422, Msg: "no text layer (scanned PDF?)", Hint: "ocr"} + case errors.Is(err, pdftext.ErrEncrypted): + return &fetch.Error{Status: 422, Msg: "encrypted PDF (password required)"} + case errors.Is(err, pdftext.ErrTimeout): + return &fetch.Error{Status: 422, Msg: "pdf extraction timed out"} + case errors.Is(err, pdftext.ErrTooMuchText): + return &fetch.Error{Status: 422, Msg: "pdf text over the output limit"} + case errors.Is(err, pdfpool.ErrBusy): + return &fetch.Error{Status: 503, Msg: "pdf workers busy", RetryAfter: 10} + case errors.Is(err, pdfpool.ErrCallerBusy): + return &fetch.Error{Status: 429, Msg: "busy"} + case errors.Is(err, pdftext.ErrUnreadable): + return &fetch.Error{Status: 422, Msg: "unreadable PDF"} + default: + return &fetch.Error{Status: 422, Msg: "unreadable PDF"} + } +} + +// renderPDF draws the document for an agent: the whole text when it fits the +// output cap, otherwise the map. +func (s *Server) renderPDF(doc *pdftext.Doc) (pdfdoc.Rendered, error) { + return pdfdoc.Render(doc, pdfdoc.Options{OutCap: s.OutCap}) +} diff --git a/internal/api/pdf_test.go b/internal/api/pdf_test.go new file mode 100644 index 0000000..041978a --- /dev/null +++ b/internal/api/pdf_test.go @@ -0,0 +1,402 @@ +package api + +// P5a task 06: PDFs through /v1/fetch and /v2/scrape — extraction in the child, the pool, the +// document cache, the map, pages, grep, and the errors. +// +// The child is the test binary itself (TestMain below), as in internal/pdftext's tests. +// SIFT_PDFTEXT_RUNLOG makes every child start append a line to a file, so a test can count +// extractions; SIFT_PDFTEXT_SLEEP slows a child down. Both hooks live only in this file. + +import ( + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "net/netip" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "git.wntrmute.dev/kyle/sift/internal/auth" + "git.wntrmute.dev/kyle/sift/internal/dcache" + "git.wntrmute.dev/kyle/sift/internal/digest" + "git.wntrmute.dev/kyle/sift/internal/fetch" + "git.wntrmute.dev/kyle/sift/internal/guard" + "git.wntrmute.dev/kyle/sift/internal/pdfpool" + "git.wntrmute.dev/kyle/sift/internal/pdftest" + "git.wntrmute.dev/kyle/sift/internal/pdftext" +) + +func TestMain(m *testing.M) { + if os.Getenv("SIFT_PDFTEXT_CHILD") == "1" { + if p := os.Getenv("SIFT_PDFTEXT_RUNLOG"); p != "" { + f, err := os.OpenFile(p, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600) + if err == nil { + _, _ = f.WriteString("run\n") + _ = f.Close() + } + } + if d, err := time.ParseDuration(os.Getenv("SIFT_PDFTEXT_SLEEP")); err == nil { + time.Sleep(d) + } + os.Exit(pdftext.ChildMain(os.Args[1:], os.Stdin, os.Stdout)) + } + os.Exit(m.Run()) +} + +func threePDF() []byte { + return pdftest.Build(pdftest.Doc{ + Title: "Three Pages", + Pages: []string{"Chapter one begins here.", "The quick brown fox is on page two.", "Last page."}, + Outline: []pdftest.Outline{{Title: "1 Intro", Page: 1, Level: 1}, {Title: "2 Fox", Page: 2, Level: 1}}, + }) +} + +func longPDF(n int) []byte { + var pages []string + for p := 1; p <= n; p++ { + pages = append(pages, pdftest.Lorem(p, 40)) + } + return pdftest.Build(pdftest.Doc{Title: "Long Thesis", Pages: pages, Outline: []pdftest.Outline{{Title: "1 Introduction", Page: 1, Level: 1}, {Title: "2 Method", Page: n / 2, Level: 1}}}) +} + +// origin serves one PDF and counts GETs. It honours If-None-Match against its current ETag. +type origin struct { + mu sync.Mutex + ct string + body []byte + etag string + gets atomic.Int32 + srv *httptest.Server +} + +func newOrigin(t *testing.T, ct string, body []byte) *origin { + o := &origin{ct: ct, body: body, etag: `"v1"`} + o.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + o.gets.Add(1) + o.mu.Lock() + defer o.mu.Unlock() + if r.Header.Get("If-None-Match") == o.etag { + w.WriteHeader(http.StatusNotModified) + return + } + w.Header().Set("ETag", o.etag) + if o.ct != "" { + w.Header().Set("Content-Type", o.ct) + } + _, _ = w.Write(o.body) + })) + t.Cleanup(o.srv.Close) + return o +} + +func (o *origin) replace(body []byte, etag string) { + o.mu.Lock() + o.body, o.etag = body, etag + o.mu.Unlock() +} + +type pdfRig struct { + s *Server + fleet string // bearer key of caller "test" (hermes-fleet) + other string // bearer key of caller "other" (hermes-fleet) + gs string // bearer key of restricted caller "straylight-gs" + runlog string + logs *recHandler +} + +func newPDFRig(t *testing.T, childEnv ...string) *pdfRig { + t.Helper() + dir := t.TempDir() + keys := filepath.Join(dir, "keys.toml") + key := func(name, ws string) string { + out := filepath.Join(dir, name) + if err := auth.Add(keys, name, ws, out); err != nil { + t.Fatal(err) + } + b, err := os.ReadFile(out) + if err != nil { + t.Fatal(err) + } + return strings.TrimSpace(string(b)) + } + r := &pdfRig{fleet: key("test", "hermes-fleet"), other: key("other", "hermes-fleet"), gs: key("straylight-gs", "guardian-star"), runlog: filepath.Join(dir, "runs"), logs: &recHandler{}} + store, err := auth.Load(keys) + if err != nil { + t.Fatal(err) + } + g := &guard.Guard{AllowPrefixes: []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")}} + s := New("test", g, &fetch.Fetcher{Guard: g, UA: "sift-test"}, store, slog.New(r.logs)) + s.PermitRestricted = []string{"straylight-gs"} + db, err := dcache.Open(filepath.Join(dir, "cache.db"), 0, nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + s.Digest = &digest.Engine{Cache: db} // the cache lives on the digest engine; no model is configured + s.PDFRunner = &pdftext.Runner{Exe: os.Args[0], Env: append([]string{"SIFT_PDFTEXT_CHILD=1", "SIFT_PDFTEXT_RUNLOG=" + r.runlog}, childEnv...), Deadline: 20 * time.Second} + r.s = s + return r +} + +func (r *pdfRig) runs() int { + b, _ := os.ReadFile(r.runlog) + return strings.Count(string(b), "run\n") +} + +func (r *pdfRig) fetch(t *testing.T, key string, body map[string]any) (int, map[string]any, http.Header) { + t.Helper() + rec := postJSON(t, r.s, "/v1/fetch", key, body) + var out map[string]any + _ = json.Unmarshal(rec.Body.Bytes(), &out) + return rec.Code, out, rec.Header() +} + +func pdfInfo(t *testing.T, out map[string]any) map[string]any { + t.Helper() + p, ok := out["pdf"].(map[string]any) + if !ok { + t.Fatalf("envelope has no pdf object: %v", out) + } + return p +} + +func TestPDFCleanFetch(t *testing.T) { + for _, ct := range []string{"application/pdf", "application/octet-stream", ""} { + r := newPDFRig(t) + o := newOrigin(t, ct, threePDF()) + code, out, _ := r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL}) + if code != 200 { + t.Fatalf("%q: %d %v", ct, code, out) + } + content, _ := out["content"].(string) + if !strings.Contains(content, "--- page 2 ---") || !strings.Contains(content, "quick brown fox") || out["mode"] != "clean" { + t.Errorf("%q: content %q mode %v", ct, content, out["mode"]) + } + p := pdfInfo(t, out) + if p["pages"] != 3.0 || p["title"] != "Three Pages" || p["map"] != false || p["first"] != 1.0 || p["last"] != 3.0 { + t.Errorf("%q: pdf = %v", ct, p) + } + if ol, _ := p["outline"].([]any); len(ol) != 2 { + t.Errorf("%q: outline = %v", ct, p["outline"]) + } + } +} + +func TestPDFMap(t *testing.T) { + r := newPDFRig(t) + r.s.OutCap = 20000 + o := newOrigin(t, "application/pdf", longPDF(40)) // ~40 × 2,700 characters + code, out, _ := r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL}) + if code != 200 { + t.Fatalf("%d %v", code, out) + } + content, _ := out["content"].(string) + p := pdfInfo(t, out) + if p["map"] != true || !strings.Contains(content, "## Outline") || !strings.Contains(content, "- 2 Method (p. 20)") || len([]rune(content)) > 20000 { + t.Errorf("map: pdf %v, %d chars:\n%.400s", p, len([]rune(content)), content) + } +} + +func TestPDFPagesAndGrep(t *testing.T) { + r := newPDFRig(t) + o := newOrigin(t, "application/pdf", threePDF()) + code, out, _ := r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL, "pages": "2-3"}) + content, _ := out["content"].(string) + if code != 200 || strings.Contains(content, "--- page 1 ---") || !strings.Contains(content, "--- page 3 ---") { + t.Errorf("pages 2-3: %d %q", code, content) + } + if p := pdfInfo(t, out); p["first"] != 2.0 || p["last"] != 3.0 { + t.Errorf("pages 2-3: pdf %v", p) + } + for _, bad := range []map[string]any{{"pages": "0-1"}, {"pages": "9"}, {"grep": strings.Repeat("a", 201)}, {"grep": "(unclosed"}} { + bad["url"] = o.srv.URL + if code, out, _ := r.fetch(t, r.fleet, bad); code != 400 || out["success"] != false { + t.Errorf("%v: %d %v, want 400", bad, code, out) + } + } + code, out, _ = r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL, "grep": "quick brown"}) + content, _ = out["content"].(string) + if code != 200 || !strings.Contains(content, "p. 2: The quick brown fox is on page two.") || pdfInfo(t, out)["grep_hits"] != 1.0 { + t.Errorf("grep: %d %q %v", code, content, out["pdf"]) + } + // pages and grep on a non-PDF are refused rather than ignored. + html := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/html") + _, _ = io.WriteString(w, "

hi

") + })) + defer html.Close() + if code, out, _ := r.fetch(t, r.fleet, map[string]any{"url": html.URL, "pages": "1"}); code != 400 { + t.Errorf("pages on html: %d %v, want 400", code, out) + } +} + +func TestPDFCache(t *testing.T) { + r := newPDFRig(t) + o := newOrigin(t, "application/pdf", threePDF()) + get := func(extra map[string]any) string { + body := map[string]any{"url": o.srv.URL} + for k, v := range extra { + body[k] = v + } + code, out, _ := r.fetch(t, r.fleet, body) + if code != 200 { + t.Fatalf("%v: %d %v", extra, code, out) + } + c, _ := out["content"].(string) + return c + } + get(nil) + if o.gets.Load() != 1 || r.runs() != 1 { + t.Fatalf("first fetch: %d GETs, %d extractions; want 1 and 1", o.gets.Load(), r.runs()) + } + if c := get(map[string]any{"pages": "2"}); !strings.Contains(c, "quick brown fox") { + t.Errorf("cached pages=2: %q", c) + } + if o.gets.Load() != 1 || r.runs() != 1 { + t.Errorf("fresh cache hit: %d GETs, %d extractions; want still 1 and 1", o.gets.Load(), r.runs()) + } + // Stale: a conditional GET, answered 304, re-uses the text. + r.s.PDFTTL = time.Nanosecond + time.Sleep(2 * time.Millisecond) + get(map[string]any{"grep": "Chapter"}) + if o.gets.Load() != 2 || r.runs() != 1 { + t.Errorf("after 304: %d GETs, %d extractions; want 2 and 1", o.gets.Load(), r.runs()) + } + // The document changed: a full GET and a new extraction. + o.replace(pdftest.Build(pdftest.Doc{Pages: []string{"Brand new text."}}), `"v2"`) + if c := get(nil); !strings.Contains(c, "Brand new text.") { + t.Errorf("changed document: %q", c) + } + if o.gets.Load() != 3 || r.runs() != 2 { + t.Errorf("after change: %d GETs, %d extractions; want 3 and 2", o.gets.Load(), r.runs()) + } + if DefaultPDFTTL != 24*time.Hour { + t.Errorf("DefaultPDFTTL = %v, want 24h", DefaultPDFTTL) + } +} + +func TestPDFErrors(t *testing.T) { + r := newPDFRig(t) + enc, err := os.ReadFile(filepath.Join("..", "..", "tests", "testdata", "pdf", "encrypted-user.pdf")) + if err != nil { + t.Fatal(err) + } + for _, tc := range []struct { + name string + body []byte + want string + hint string + }{ + {"scanned", pdftest.Build(pdftest.Doc{Pages: []string{"", ""}, Scanned: []int{1, 2}}), "no text layer", "ocr"}, + {"encrypted", enc, "encrypted", ""}, + {"garbage", []byte("%PDF-1.4\nthis is not really a pdf"), "unreadable", ""}, + } { + o := newOrigin(t, "application/pdf", tc.body) + code, out, _ := r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL}) + msg, _ := out["error"].(string) + if code != 422 || !strings.Contains(msg, tc.want) || (tc.hint != "" && out["hint"] != tc.hint) { + t.Errorf("%s: %d %v; want 422 %q hint %q", tc.name, code, out, tc.want, tc.hint) + } + } +} + +func TestPDFTimeout(t *testing.T) { + r := newPDFRig(t, "SIFT_PDFTEXT_SLEEP=10s") + r.s.PDFRunner.Deadline = 300 * time.Millisecond + o := newOrigin(t, "application/pdf", threePDF()) + code, out, _ := r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL}) + if msg, _ := out["error"].(string); code != 422 || !strings.Contains(msg, "timed out") { + t.Errorf("slow child: %d %v; want 422 … timed out", code, out) + } +} + +func TestPDFBusy(t *testing.T) { + r := newPDFRig(t, "SIFT_PDFTEXT_SLEEP=1500ms") + r.s.PDFPool = pdfpool.New(1, 0, 100*time.Millisecond) + a := newOrigin(t, "application/pdf", threePDF()) + b := newOrigin(t, "application/pdf", longPDF(2)) + done := make(chan int, 1) + go func() { + code, _, _ := r.fetch(t, r.fleet, map[string]any{"url": a.srv.URL}) + done <- code + }() + time.Sleep(400 * time.Millisecond) // the first extraction holds the only slot + code, out, hdr := r.fetch(t, r.other, map[string]any{"url": b.srv.URL}) + if code != 503 || hdr.Get("Retry-After") != "10" || !strings.Contains(fmt.Sprint(out["error"]), "busy") { + t.Errorf("second caller with the pool full: %d %v Retry-After %q; want 503, busy, 10", code, out, hdr.Get("Retry-After")) + } + code, out, _ = r.fetch(t, r.fleet, map[string]any{"url": b.srv.URL}) + if code != 429 { + t.Errorf("same caller, second document while its first is running: %d %v; want 429", code, out) + } + if c := <-done; c != 200 { + t.Errorf("first fetch: %d", c) + } +} + +func TestPDFRestrictedCaller(t *testing.T) { + r := newPDFRig(t) + o := newOrigin(t, "application/pdf", threePDF()) + if code, out, _ := r.fetch(t, r.gs, map[string]any{"url": o.srv.URL}); code != 200 { + t.Fatalf("restricted fetch: %d %v", code, out) + } + r.logs.mu.Lock() + for _, rec := range r.logs.recs { + var b strings.Builder + b.WriteString(rec.Message) + rec.Attrs(func(a slog.Attr) bool { fmt.Fprintf(&b, " %s=%v", a.Key, a.Value); return true }) + line := b.String() + if strings.Contains(line, "127.0.0.1") || strings.Contains(line, "Three Pages") || strings.Contains(line, "quick brown") { + t.Errorf("a restricted caller's PDF fetch logged %q", line) + } + } + r.logs.mu.Unlock() + // The fleet does not see the restricted caller's cached text: it extracts its own copy. + if code, _, _ := r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL}); code != 200 || r.runs() != 2 { + t.Errorf("fleet after restricted: %d, %d extractions; want 200 and 2 (separate cache partitions)", code, r.runs()) + } +} + +func TestPDFScrape(t *testing.T) { + r := newPDFRig(t) + o := newOrigin(t, "application/pdf", threePDF()) + rec := postJSON(t, r.s, "/v2/scrape", r.fleet, map[string]any{"url": o.srv.URL, "formats": []string{"markdown"}}) + var out struct { + Success bool `json:"success"` + Data struct { + Markdown string `json:"markdown"` + Metadata map[string]any `json:"metadata"` + } `json:"data"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil || rec.Code != 200 || !out.Success { + t.Fatalf("%d %s", rec.Code, rec.Body.String()) + } + if !strings.Contains(out.Data.Markdown, "--- page 2 ---") || out.Data.Metadata["numPages"] != 3.0 || out.Data.Metadata["title"] != "Three Pages" { + t.Errorf("scrape: markdown %q metadata %v", out.Data.Markdown, out.Data.Metadata) + } + if ct, _ := out.Data.Metadata["contentType"].(string); !strings.HasPrefix(ct, "application/pdf") { + t.Errorf("contentType = %q", ct) + } +} + +// A question on a document longer than PDFMaxQuestionChunks is refused with a pointer to grep +// and pages, before any model call. +func TestPDFQuestionTooLong(t *testing.T) { + r := newPDFRig(t) + r.s.PDFMaxQuestionChunks = 2 + o := newOrigin(t, "application/pdf", longPDF(12)) + code, out, _ := r.fetch(t, r.fleet, map[string]any{"url": o.srv.URL, "question": "what is the method?"}) + if msg, _ := out["error"].(string); code != 422 || !strings.Contains(msg, "too long") || !strings.Contains(msg, "grep") { + t.Errorf("long question: %d %v; want 422 … too long … grep", code, out) + } + if DefaultPDFMaxQuestionChunks != 200 { + t.Errorf("DefaultPDFMaxQuestionChunks = %d, want 200", DefaultPDFMaxQuestionChunks) + } +} diff --git a/internal/fetch/fetch.go b/internal/fetch/fetch.go index 388d236..b7f4d53 100644 --- a/internal/fetch/fetch.go +++ b/internal/fetch/fetch.go @@ -43,9 +43,10 @@ func (r *Resource) SHA256() string { // Error is a fetch failure with the HTTP status sift should return to its caller. type Error struct { - Status int - Msg string - Hint string + Status int + Msg string + Hint string + RetryAfter int // seconds to wait before retrying; 0 = none } func (e *Error) Error() string { return e.Msg }