From a804f5ac939c72973f5292c635d69af15a252aa0 Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Sat, 26 Sep 2026 19:56:49 -0700 Subject: [PATCH] P5c review: bound concurrent uploads; warnings is a list; sift extract keeps a base-URL path Owner review of p5c (tasks 01-02 by Ornith): - extract.go: an upload (up to 100 MiB) is read whole before the PDF pool's limits apply, so any number of concurrent uploads could buffer at once in a 1 GiB service. At most 2 bodies are read at a time now; a third waits up to 5 s, then 429 busy with Retry-After 10. - extract.go: `warnings` was null when empty; it is [] as in /v1/fetch. - extractcmd.go: extractURL replaced the client base URL's path; it now appends /v1/extract like fetch and search do. Regression tests in extract_review_test.go (api and cmd/sift). End-to-end (local serve -> Dixie; sift extract with SIFT_CLIENT): Llama 2 paper, 13.6 MB / 77 pages: --pages 1 ok, repeat served from the content-hash cache, --grep 2 hits, --question answered with page citations in 16.8 s. Log lines carry bytes/pages/mode/cached only. Co-Authored-By: Claude Opus 5.5 --- cmd/sift/extract_review_test.go | 11 ++++++++ cmd/sift/extractcmd.go | 10 +++---- internal/api/api.go | 5 ++++ internal/api/extract.go | 21 ++++++++++++++- internal/api/extract_review_test.go | 42 +++++++++++++++++++++++++++++ 5 files changed, 82 insertions(+), 7 deletions(-) create mode 100644 cmd/sift/extract_review_test.go create mode 100644 internal/api/extract_review_test.go diff --git a/cmd/sift/extract_review_test.go b/cmd/sift/extract_review_test.go new file mode 100644 index 0000000..3620848 --- /dev/null +++ b/cmd/sift/extract_review_test.go @@ -0,0 +1,11 @@ +package main + +import "testing" + +// P5c owner review: a client base URL with a path prefix keeps it, as fetch and search do. +func TestExtractURLKeepsBasePath(t *testing.T) { + got := extractURL("https://sift.example:8443/prefix/", extractOpts{File: "a.pdf", Pages: "2"}) + if want := "https://sift.example:8443/prefix/v1/extract?pages=2"; got != want { + t.Fatalf("extractURL = %q, want %q", got, want) + } +} diff --git a/cmd/sift/extractcmd.go b/cmd/sift/extractcmd.go index eeaad0a..a2dea71 100644 --- a/cmd/sift/extractcmd.go +++ b/cmd/sift/extractcmd.go @@ -79,13 +79,11 @@ func extractURL(base string, o extractOpts) string { if mode != "" { v.Set("mode", mode) } - u, err := url.Parse(base) - if err != nil { - return base + "/v1/extract?" + v.Encode() + u := strings.TrimRight(base, "/") + "/v1/extract" + if q := v.Encode(); q != "" { + u += "?" + q } - u.Path = "/v1/extract" - u.RawQuery = v.Encode() - return u.String() + return u } func extractMain() { diff --git a/internal/api/api.go b/internal/api/api.go index a29338b..9c7d4b0 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -74,6 +74,10 @@ type Server struct { reqN, blockN, scrapeN atomic.Uint64 pdfPoolOnce sync.Once + + // uploads bounds how many /v1/extract bodies (up to PDFMaxUpload each) are read into + // memory at once, ahead of the PDF pool's own limits. + uploads chan struct{} } func New(version string, g *guard.Guard, f *fetch.Fetcher, keys *auth.Store, log *slog.Logger) *Server { @@ -90,6 +94,7 @@ func New(version string, g *guard.Guard, f *fetch.Fetcher, keys *auth.Store, log OutCap: 200_000, global: make(chan struct{}, 8), per: map[string]chan struct{}{}, + uploads: make(chan struct{}, maxUploads), } s.Prom = metrics.New(s) s.Mux.HandleFunc("GET /healthz", s.healthz) diff --git a/internal/api/extract.go b/internal/api/extract.go index 19864d3..f18ecf6 100644 --- a/internal/api/extract.go +++ b/internal/api/extract.go @@ -27,6 +27,17 @@ func (s *Server) v1extract(w http.ResponseWriter, r *http.Request) { return } + select { + case s.uploads <- struct{}{}: + defer func() { <-s.uploads }() + case <-time.After(uploadWait): + w.Header().Set("Retry-After", "10") + writeErr(w, 429, "busy", "") + return + case <-r.Context().Done(): + return + } + body, rerr := io.ReadAll(r.Body) if rerr != nil { var me *http.MaxBytesError @@ -105,7 +116,7 @@ func (s *Server) v1extract(w http.ResponseWriter, r *http.Request) { content = full } - var warn []string + warn := []string{} var digMeta map[string]any var digestMS int64 if mode == "digest" { @@ -192,6 +203,14 @@ func (s *Server) v1extract(w http.ResponseWriter, r *http.Request) { writeJSON(w, 200, out) } +// maxUploads is how many uploads may be in memory at once, and uploadWait how long a further +// one waits for a slot before a 429: at 100 MiB each, a few are all a 1 GiB service can hold +// beside the extraction children. +const ( + maxUploads = 2 + uploadWait = 5 * time.Second +) + // pdfMaxUpload is the /v1/extract upload cap: the configured value, or the default when unset. func (s *Server) pdfMaxUpload() int64 { if s.PDFMaxUpload > 0 { diff --git a/internal/api/extract_review_test.go b/internal/api/extract_review_test.go new file mode 100644 index 0000000..b234df9 --- /dev/null +++ b/internal/api/extract_review_test.go @@ -0,0 +1,42 @@ +package api + +// P5c owner review: regression tests for /v1/extract. + +import ( + "net/http" + "testing" + "time" +) + +// Uploads are buffered whole (up to 100 MiB) before the PDF pool sees them, so the number read +// at once is bounded: a third concurrent upload waits, then is refused busy. +func TestExtractUploadsBounded(t *testing.T) { + r := newPDFRig(t) + for i := 0; i < cap(r.s.uploads); i++ { + r.s.uploads <- struct{}{} + } + start := time.Now() + code, out := r.extract(t, r.fleet, "", threePDF(), "application/pdf") + if code != http.StatusTooManyRequests { + t.Fatalf("upload with every slot taken: %d %v, want 429", code, out) + } + if d := time.Since(start); d < uploadWait-100*time.Millisecond { + t.Fatalf("refused after %v, want a wait of about %v first", d, uploadWait) + } + <-r.s.uploads + if code, out := r.extract(t, r.fleet, "", threePDF(), "application/pdf"); code != 200 { + t.Fatalf("upload with a free slot: %d %v", code, out) + } +} + +// The envelope's warnings are a list, never null, as in /v1/fetch. +func TestExtractWarningsIsAList(t *testing.T) { + r := newPDFRig(t) + code, out := r.extract(t, r.fleet, "", threePDF(), "application/pdf") + if code != 200 { + t.Fatalf("%d %v", code, out) + } + if _, ok := out["warnings"].([]any); !ok { + t.Fatalf("warnings = %#v, want a list", out["warnings"]) + } +}