From 72e0828360946ae628c8b786fb2b415c73737733 Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Sat, 26 Sep 2026 22:43:28 -0700 Subject: [PATCH] =?UTF-8?q?api:=20NewBatchRunner=20=E2=80=94=20the=20batch?= =?UTF-8?q?=20fetch=20adapter,=20retry=20on=20busy,=20outcome=20metrics?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the batch runner to fetchTarget and feed the sift_batch_* metrics. New internal/api/batch.go: NewBatchRunner overwrites cfg.Hooks (Job, Target, JobSeconds, Depth, Inflight → s.Prom) and sets the runner's FetchFunc to batchFetch, which tags the context as a batch fetch, runs under requestDeadline, and maps fetchTarget's errors to what the runner retries (429/503 → ErrRetryLater) and what the metrics classify (batchOutcome). Added the Batch field to Server (nil → task 07 answers 503 "batch disabled") and, in extractPDF, charged the PDF pool to c.Name+"/batch" under the batch mark so a batch never takes a caller's interactive slot while the single-flight key is unchanged. No HTTP endpoints. Co-Authored-By: ornith-1.5-35b-a3b --- docs/implementer-log.md | 1 + internal/api/api.go | 4 + internal/api/batch.go | 113 +++++++++++++++++++ internal/api/batchrun_test.go | 202 ++++++++++++++++++++++++++++++++++ internal/api/pdf.go | 6 +- 5 files changed, 325 insertions(+), 1 deletion(-) create mode 100644 internal/api/batch.go create mode 100644 internal/api/batchrun_test.go diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 9dcdbfd..1854bed 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -22,6 +22,7 @@ One row per task session that ends in a commit. Newest last. | p6/03-runner-workers | 2026-09-26 | done | 1 | gate: ok | none | Workers in new internal/batch/worker.go: one rotation of live jobs in submit order, round-robin via an absolute cursor (advances in pick, wraps at access, so it survives a second job being added while the first is in flight — the mod-n cursor from the design stuck at 0 for one job and re-picked it), serial per job (target i+1 never starts before i is stored), one target per fetch under the Close-cancelled context (panic→ErrInternal, never echoed; ErrRetryLater retried after RetryWait, bounded by RetryFor then stored failed with the wrapped text). Terminal status once per job: all done→done, all failed→failed, mix→partial. Cancel of a running job sets cancelled, stores the in-flight target without overwriting cancelled, leaves later targets pending. Close sets closed, cancels the context, broadcasts, waits the workers, idempotent. batch.go: New starts cfg.Workers, Submit appends the liveJob and wakes, Cancel handles the in-flight case, Close stops. Absolute cursor is the one deviation from the design's "index in live" comment. All race tests pass -count=3. | | | p6/04-metrics | 2026-09-26 | done | 1 | gate: ok | none | Batch series in new internal/metrics/batch.go, registered in New: `BatchJobsTotal`/`BatchTargetsTotal` CounterVecs (label "outcome", any value outside the known set folded to "other" via `batchJobOutcomes`/`batchTargetOutcomes` maps so a raw string never becomes a label), `BatchQueueDepth`/`BatchTargetsInflight` gauges, and `BatchJobSeconds`/`BatchTargetSeconds` histograms (durationBuckets) plus `BatchResultBytes` (1K,4K,16K,64K,256K,1M). Nil-receiver-safe `Observe…`/`Set…` methods mirroring ObservePDF/ObserveDigest. Series served through the existing /metrics handler; no new endpoint, no Snapshot change. Given test byte-identical; metrics tests pass under -race -count=3; make gate ok. | | p6/05-fetchtarget | 2026-09-26 | done | 1 | gate: ok | none | Extracted v1fetch's per-target pipeline into new `internal/api/target.go`: `fetchTarget(ctx, c, t batch.Target) (batch.Result, error)`. Body moved unchanged in order — question truncation to 500 runes, mode default/check, wantCtx, doFetch, recover path, PDF branch (render, pages/grep errors, question-chunk limit), clean/raw branch, digest, context search, the fetch log line, and the envelope. `content` is returned as `Result.Content`; every other envelope key (`mode`,`untrusted`,`source`,`warnings`,`sizes`,`timings`, and `pdf`/`digest`/`context` when present) goes in `Result.Meta`, never content. `v1fetch` keeps reqN/caller/decode/400, builds a `batch.Target` (`URL` + `Mode`/`Question`/`Pages`/`Grep`/`JQ`/`Context`/`Recover` into `Options`), calls `fetchTarget`, `writeFetchErr` on error and `writeJSON(w,200,env)` on success where `env` is a fresh copy of `Meta` plus `"content"` (a copy, so `len(env)==len(Meta)+1` holds and Meta is not mutated). The two `writeErr` failures become `*fetch.Error` with the same status/msg/hint: 400 "invalid url" (bad mode), 400 "raw mode does not apply to PDFs". `doFetch` takes `context.Context` instead of `*http.Request` (its three `r.Context()` → `ctx`); `scrape` passes `r.Context()`. Removed now-unused `fmt`/`unicode/utf8`/`recover` imports, added `batch`. All api tests pass under -race -count=3; make gate ok; given test byte-identical. | +| p6/06-batch-wiring | 2026-09-26 | done | 1 | gate: ok | none | New `internal/api/batch.go`: `NewBatchRunner` overwrites `cfg.Hooks` so Job→ObserveBatchJob, Target→ObserveBatchTarget(batchOutcome)+ObserveBatchTargetSeconds (+ObserveBatchResultBytes only when err==nil), and JobSeconds/Depth/Inflight→the matching `s.Prom` methods; the runner's FetchFunc is `batchFetch`, which tags ctx with an unexported `batchCtxKey`, runs under `requestDeadline`, and maps fetchTarget's errors (a *fetch.Error 429/503 → `fmt.Errorf("%s: %w", fe.Msg, batch.ErrRetryLater)` so the runner retries; any other *fetch.Error → TargetError with 502 rewritten to 500 "internal error" and `s.blockN.Add(1)` for 403; anything else → TargetError 500 logged Error with `s.logErr`). `batchOutcome` classifies nil→ok, ErrRetryLater-wrapped→busy, then by TargetError.Status (403 blocked, 429 busy, 504 timeout, 415 unsupported, 422 whose Msg starts "document too long"→too_long, other 4xx→upstream_4xx, 5xx→upstream_5xx, any other error→upstream_5xx). Added `Batch *batch.Runner` Server field (nil → the /v1/batch endpoints answer 503 "batch disabled"). `pdf.go` extractPDF charges the PDF pool to `c.Name+"/batch"` when the ctx carries the batch mark, single-flight key unchanged so a batch and an interactive extract of the same PDF still share one child; the only new log line carries caller and host="-" (never a URL/host). All api tests pass under -race -count=3; make gate ok; given file byte-identical. | ## P5a review (owner, 2026-09-26) diff --git a/internal/api/api.go b/internal/api/api.go index 83c0769..3818f9b 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -73,6 +73,10 @@ type Server struct { pdfPoolOnce sync.Once + // Batch is the batch runner; nil means the /v1/batch endpoints (task 07) answer 503 + // "batch disabled". + Batch *batch.Runner + // 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{} diff --git a/internal/api/batch.go b/internal/api/batch.go new file mode 100644 index 0000000..7dc7c9e --- /dev/null +++ b/internal/api/batch.go @@ -0,0 +1,113 @@ +package api + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + "git.wntrmute.dev/kyle/sift/internal/auth" + "git.wntrmute.dev/kyle/sift/internal/batch" + "git.wntrmute.dev/kyle/sift/internal/dcache" + "git.wntrmute.dev/kyle/sift/internal/fetch" +) + +// batchCtxKey marks a context as a batch fetch so extractPDF charges the PDF pool to the +// caller's "/batch" slot instead of its interactive one. +type batchCtxKey struct{} + +func isBatchFetch(ctx context.Context) bool { + ok, _ := ctx.Value(batchCtxKey{}).(bool) + return ok +} + +// NewBatchRunner builds the runner that serve (task 07) and the tests use. Its FetchFunc is +// s.batchFetch and its Hooks feed s.Prom; any cfg.Hooks from the caller is overwritten. +func (s *Server) NewBatchRunner(db *dcache.DB, cfg batch.Config) *batch.Runner { + cfg.Hooks = batch.Hooks{ + Job: func(outcome string) { + s.Prom.ObserveBatchJob(outcome) + }, + Target: func(err error, d time.Duration, n int) { + s.Prom.ObserveBatchTarget(batchOutcome(err)) + s.Prom.ObserveBatchTargetSeconds(d) + if err == nil { + s.Prom.ObserveBatchResultBytes(n) + } + }, + JobSeconds: func(d time.Duration) { + s.Prom.ObserveBatchJobSeconds(d) + }, + Depth: func(live int) { + s.Prom.SetBatchQueueDepth(live) + }, + Inflight: func(n int) { + s.Prom.SetBatchTargetsInflight(n) + }, + } + return batch.New(db, s.batchFetch, cfg, s.Log) +} + +// batchFetch is the runner's FetchFunc: one target through s.fetchTarget under the request +// deadline, tagged as a batch fetch. Its errors map to what the runner and the metrics need. +func (s *Server) batchFetch(ctx context.Context, c *auth.Caller, t batch.Target) (batch.Result, error) { + ctx, cancel := context.WithTimeout(ctx, requestDeadline) + defer cancel() + ctx = context.WithValue(ctx, batchCtxKey{}, true) + res, err := s.fetchTarget(ctx, c, t) + if err != nil { + var fe *fetch.Error + if errors.As(err, &fe) { + switch fe.Status { + case 429, 503: + return batch.Result{}, fmt.Errorf("%s: %w", fe.Msg, batch.ErrRetryLater) + } + if fe.Status == 502 { + fe.Status = 500 + fe.Msg = "internal error" + } + if fe.Status == 403 { + s.blockN.Add(1) + } + return batch.Result{}, &batch.TargetError{Status: fe.Status, Msg: fe.Msg, Hint: fe.Hint} + } + s.Log.Error("internal", "caller", c.Name, "host", "-", "err", s.logErr(c, err)) + return batch.Result{}, &batch.TargetError{Status: 500, Msg: "internal error"} + } + return res, nil +} + +// batchOutcome classifies a finished target's error into the sift_batch_targets_total label: +// a nil error is ok, a retry-later the runner gave up on is busy, a TargetError by its status, +// and anything else upstream_5xx so a raw string never becomes a label. +func batchOutcome(err error) string { + switch { + case err == nil: + return "ok" + case errors.Is(err, batch.ErrRetryLater): + return "busy" + } + var te *batch.TargetError + if errors.As(err, &te) { + switch te.Status { + case 403: + return "blocked" + case 429: + return "busy" + case 504: + return "timeout" + case 415: + return "unsupported" + case 422: + if strings.HasPrefix(te.Msg, "document too long") { + return "too_long" + } + return "upstream_4xx" + } + if te.Status >= 400 && te.Status < 500 { + return "upstream_4xx" + } + } + return "upstream_5xx" +} diff --git a/internal/api/batchrun_test.go b/internal/api/batchrun_test.go new file mode 100644 index 0000000..5e5c14c --- /dev/null +++ b/internal/api/batchrun_test.go @@ -0,0 +1,202 @@ +package api + +// P6 task 06: the runner wired to fetchTarget and the metrics, with no HTTP yet. The helpers +// here (newBatchServer, pageServer, fakeDNS) are shared with task 07's batch_test.go. + +import ( + "context" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "net/netip" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "git.wntrmute.dev/kyle/sift/internal/auth" + "git.wntrmute.dev/kyle/sift/internal/batch" + "git.wntrmute.dev/kyle/sift/internal/dcache" + "git.wntrmute.dev/kyle/sift/internal/fetch" + "git.wntrmute.dev/kyle/sift/internal/guard" +) + +var loopback = []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")} + +// fakeDNS resolves internal.test to a private address; every other name fails to resolve. +func fakeDNS(_ context.Context, host string) ([]netip.Addr, error) { + if host == "internal.test" { + return []netip.Addr{netip.MustParseAddr("10.0.0.5")}, nil + } + return nil, fmt.Errorf("no such host %s", host) +} + +// newBatchServer builds a Server with the batch runner wired over an in-memory dcache through +// s.NewBatchRunner, exactly as serve does. callers are the names to provision (each gets a key). +func newBatchServer(t *testing.T, callers ...string) (*Server, map[string]string) { + t.Helper() + dir := t.TempDir() + keysPath := filepath.Join(dir, "keys.toml") + keys := map[string]string{} + for _, name := range callers { + out := filepath.Join(dir, name) + if err := auth.Add(keysPath, name, "hermes-fleet", out); err != nil { + t.Fatal(err) + } + b, err := os.ReadFile(out) + if err != nil { + t.Fatal(err) + } + keys[name] = strings.TrimSpace(string(b)) + } + store, err := auth.Load(keysPath) + if err != nil { + t.Fatal(err) + } + g := &guard.Guard{AllowPrefixes: loopback, LookupIP: fakeDNS} + f := &fetch.Fetcher{Guard: g, UA: "sift-test"} + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + s := New("test", g, f, store, log) + db, err := dcache.Open("", 1<<20, log) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + s.Batch = s.NewBatchRunner(db, batch.Config{}) + t.Cleanup(s.Batch.Close) // before db.Close + return s, keys +} + +// pageServer serves a small HTML page at every path; hold, when non-nil, blocks each request +// until it is closed. +func pageServer(t *testing.T, hold chan struct{}) *httptest.Server { + t.Helper() + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if hold != nil { + select { + case <-hold: + case <-r.Context().Done(): + return + } + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + fmt.Fprintf(w, `T

hello batch %s

`, r.URL.Path) + })) + t.Cleanup(ts.Close) + return ts +} + +func waitJob(t *testing.T, r *batch.Runner, id string) batch.Job { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + j, _, ok := r.Status(id) + if ok && (j.Status == "done" || j.Status == "partial" || j.Status == "failed" || j.Status == "cancelled") { + return j + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("job never finished") + return batch.Job{} +} + +func promText(t *testing.T, s *Server) string { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/metrics", nil) + req.RemoteAddr, req.Host = "127.0.0.1:9", "127.0.0.1:8890" + rec := httptest.NewRecorder() + s.ServeHTTP(rec, req) + return rec.Body.String() +} + +// The runner fetches through fetchTarget as the submitting caller; a name that resolves to a +// private address is a per-target "blocked" failure; the hooks reach the metrics. +func TestBatchRunnerWired(t *testing.T) { + ts := pageServer(t, nil) + s, _ := newBatchServer(t, "test") + c := &auth.Caller{Name: "test", Workspace: "hermes-fleet"} + id, reused, err := s.Batch.Submit(context.Background(), c, dcache.ClassFleet, []batch.Target{ + {URL: ts.URL + "/one"}, {URL: "http://internal.test/"}, + }) + if err != nil || reused { + t.Fatalf("submit: %v %v", err, reused) + } + if j := waitJob(t, s.Batch, id); j.Status != "partial" { + t.Fatalf("job %+v", j) + } + _, ts2, _ := s.Batch.Status(id) + if ts2[0].Status != "done" || ts2[0].Result == nil || !strings.Contains(ts2[0].Result.Content, "hello batch /one") { + t.Fatalf("target 0: %+v", ts2[0]) + } + if ts2[0].Result.Meta["mode"] != "clean" || ts2[0].Result.Meta["untrusted"] != true { + t.Fatalf("target 0 meta %v: not the fetchTarget envelope", ts2[0].Result.Meta) + } + if ts2[1].Status != "failed" || !strings.Contains(ts2[1].Error, "blocked") { + t.Fatalf("target 1: %+v", ts2[1]) + } + out := promText(t, s) + for _, want := range []string{ + `sift_batch_jobs_total{outcome="created"} 1`, + `sift_batch_jobs_total{outcome="partial"} 1`, + `sift_batch_targets_total{outcome="ok"} 1`, + `sift_batch_targets_total{outcome="blocked"} 1`, + `sift_batch_result_bytes_count 1`, + `sift_batch_queue_depth 0`, + } { + if !strings.Contains(out, want) { + t.Errorf("metrics lack %q", want) + } + } +} + +// The adapter turns fetch errors into what the runner and the metrics need. +func TestBatchOutcomeClasses(t *testing.T) { + cases := []struct { + err error + want string + }{ + {nil, "ok"}, + {fmt.Errorf("busy: %w", batch.ErrRetryLater), "busy"}, + {&batch.TargetError{Status: 403, Msg: "blocked: private address"}, "blocked"}, + {&batch.TargetError{Status: 429, Msg: "busy"}, "busy"}, + {&batch.TargetError{Status: 504, Msg: "timeout"}, "timeout"}, + {&batch.TargetError{Status: 415, Msg: "unsupported content type"}, "unsupported"}, + {&batch.TargetError{Status: 422, Msg: "document too long for question (6001 chunks); use grep or pages"}, "too_long"}, + {&batch.TargetError{Status: 422, Msg: "unreadable PDF"}, "upstream_4xx"}, + {&batch.TargetError{Status: 404, Msg: "not found"}, "upstream_4xx"}, + {&batch.TargetError{Status: 500, Msg: "internal error"}, "upstream_5xx"}, + {errors.New("something else"), "upstream_5xx"}, + } + for _, c := range cases { + if got := batchOutcome(c.err); got != c.want { + t.Errorf("batchOutcome(%v) = %q, want %q", c.err, got, c.want) + } + } +} + +// batchFetch maps a busy fetch to ErrRetryLater (the runner retries it) and every other +// failure to a TargetError with the /v1/fetch status. +func TestBatchFetchErrors(t *testing.T) { + s, _ := newBatchServer(t, "test") + c := &auth.Caller{Name: "test", Workspace: "hermes-fleet"} + _, err := s.batchFetch(context.Background(), c, batch.Target{URL: "http://internal.test/"}) + var te *batch.TargetError + if !errors.As(err, &te) || te.Status != 403 || !strings.Contains(te.Msg, "blocked") { + t.Fatalf("private name: %v, want a TargetError 403 blocked", err) + } + // Fill the global fetch gate so the next fetch is refused busy. + for i := 0; i < 8; i++ { + if err := s.acquire(fmt.Sprintf("filler%d", i), "http://filler.test/"); err != nil { + t.Fatal(err) + } + } + ts := pageServer(t, nil) + _, err = s.batchFetch(context.Background(), c, batch.Target{URL: ts.URL}) + if !errors.Is(err, batch.ErrRetryLater) { + t.Fatalf("busy gate: %v, want an error wrapping batch.ErrRetryLater", err) + } +} diff --git a/internal/api/pdf.go b/internal/api/pdf.go index d13e313..c141247 100644 --- a/internal/api/pdf.go +++ b/internal/api/pdf.go @@ -58,7 +58,11 @@ func (s *Server) extractPDF(ctx context.Context, c *auth.Caller, res *fetch.Reso return v, err } start := time.Now() - v, err := s.pool().Do(ctx, s.callerClass(c)+"|"+res.SHA256(), c.Name, fn) + caller := c.Name + if isBatchFetch(ctx) { + caller = c.Name + "/batch" + } + v, err := s.pool().Do(ctx, s.callerClass(c)+"|"+res.SHA256(), caller, fn) total := time.Since(start) // The job's own extraction time is what this call measured it to be; the // rest of Do is time spent waiting for a slot.