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 <noreply@anthropic.com>
This commit is contained in:
@@ -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)
|
||||
|
||||
+20
-1
@@ -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 {
|
||||
|
||||
@@ -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"])
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user