sift/docs/plans/p6/06-batch-wiring.md

P6 task 06: the runner wired to fetchTarget and the metrics

Branch: p6 (git switch p6; git status --short must be empty, otherwise stop) Commit subject: api: NewBatchRunner — the batch fetch adapter, retry on busy, outcome metrics

Goal

Build the production runner: its FetchFunc is an adapter over fetchTarget (task 05), and its hooks feed the sift_batch_* metrics (task 04). No HTTP endpoints yet (task 07).

Files

• Copy: docs/plans/p6/_files/api/batchrun_test.go → internal/api/batchrun_test.go
• Create: internal/api/batch.go (NewBatchRunner, batchFetch, batchOutcome)
• Modify:
  • internal/api/api.go: only the Batch field.
  • internal/api/pdf.go: only the pool caller name (rule 3).
  • docs/implementer-log.md

What already exists

// internal/batch (task 02)
type Options struct{ Mode, Question, Pages, Grep, JQ string; Context *bool; Recover bool }
type Target struct{ URL string; Options Options }
type Result struct{ Content string; Meta map[string]any }
type TargetError struct{ Status int; Msg, Hint string }
type FetchFunc func(ctx context.Context, c *auth.Caller, t Target) (Result, error)
type Job struct{ ID, Caller, Status, Error string; TargetsTotal, TargetsDone int; CreatedAt int64 }
type TargetStatus struct{ Idx int; URL, Status string; Result *Result; Error, Hint string }
type Hooks struct {
	Job        func(outcome string)
	Target     func(err error, d time.Duration, resultBytes int)
	JobSeconds func(d time.Duration)
	Depth      func(live int)
	Inflight   func(n int)
}
type Config struct{ TTL time.Duration; TargetsPerJob, JobsPerCaller, JobsTotal, Workers int; MaxBytes int64; RetryWait, RetryFor time.Duration; Hooks Hooks }
var ErrNoTargets, ErrTooManyTargets, ErrCallerBusy, ErrBusy, ErrRetryLater error
func New(db *dcache.DB, fetch FetchFunc, cfg Config, log *slog.Logger) *Runner
func (r *Runner) Submit(ctx context.Context, c *auth.Caller, callerClass string, targets []Target) (string, bool, error)
func (r *Runner) Cancel(id string) bool
func (r *Runner) Status(id string) (Job, []TargetStatus, bool)
func (r *Runner) Touch(id string)
func (r *Runner) Close()

// internal/metrics (task 04), on s.Prom
ObserveBatchJob(outcome string); ObserveBatchTarget(outcome string)
SetBatchQueueDepth(n int); SetBatchTargetsInflight(n int)
ObserveBatchJobSeconds(d time.Duration); ObserveBatchTargetSeconds(d time.Duration); ObserveBatchResultBytes(n int)

// internal/api (task 05 and before)
func (s *Server) fetchTarget(ctx context.Context, c *auth.Caller, t batch.Target) (batch.Result, error)
func (s *Server) caller(w http.ResponseWriter, r *http.Request) *auth.Caller // nil after writing 401/403
func (s *Server) callerClass(c *auth.Caller) string
func (s *Server) logErr(c *auth.Caller, err error) any
// requestDeadline (210 s), maxReqBody (64 KiB), writeErr, writeJSON

Interfaces added here

// in Server (after pdfPoolOnce):
	Batch *batch.Runner // nil: the /v1/batch endpoints (task 07) answer 503 "batch disabled"

// NewBatchRunner builds the runner that serve (task 08) and the tests use: its FetchFunc is
// batchFetch and its Hooks feed s.Prom. cfg.Hooks is overwritten.
func (s *Server) NewBatchRunner(db *dcache.DB, cfg batch.Config) *batch.Runner

func (s *Server) batchFetch(ctx context.Context, c *auth.Caller, t batch.Target) (batch.Result, error)
func batchOutcome(err error) string

Rules

1. batchFetch, the runner's FetchFunc:
   • context.WithTimeout(ctx, requestDeadline), mark the context as a batch fetch (an unexported context key), and call s.fetchTarget.
   • Map its error:
     • a *fetch.Error with status 429 or 503 → fmt.Errorf("%s: %w", fe.Msg, batch.ErrRetryLater), so the runner retries it;
     • any other *fetch.Error → &batch.TargetError{Status, Msg, Hint}, with 502 rewritten to 500 internal error as writeFetchErr does, and s.blockN.Add(1) for a 403;
     • anything else → &batch.TargetError{Status: 500, Msg: "internal error"}, logged Error with s.logErr(c, err).
2. Hooks (in NewBatchRunner):
   • Job → ObserveBatchJob;
   • Target(err, d, n) → ObserveBatchTarget(batchOutcome(err)), ObserveBatchTargetSeconds(d), and ObserveBatchResultBytes(n) when err == nil;
   • JobSeconds, Depth, Inflight → the matching s.Prom methods.
   • batchOutcome(err) is:
     • ok for nil;
     • busy for an error wrapping ErrRetryLater (the runner gave up retrying);
     • otherwise, by TargetError.Status: 403 blocked, 429 busy, 504 timeout, 415 unsupported, 422 whose message starts document too long too_long, other 4xx upstream_4xx, 5xx upstream_5xx; any other error upstream_5xx.
3. The PDF pool under a batch: in extractPDF, the pool's caller argument is c.Name+"/batch" when the context carries the batch mark, else c.Name as today. A batch extraction then never takes the caller's own interactive slot. The single-flight key is unchanged, so a batch and an interactive extraction of the same PDF still share one child.
4. Log lines added here carry the caller and the job id, never a target URL or hostname. The per-target fetch line inside fetchTarget is unchanged.

Facts

• The given api test builds the server through s.NewBatchRunner over an in-memory dcache (newBatchServer). Targets are local httptest servers, or internal.test, which the test's LookupIP resolves to 10.0.0.5.
• TestBatchFetchErrors fills the 8-slot global gate with s.acquire so the next fetch is refused busy after acquire's 5 s wait.
• batchrun_test.go also holds the helpers task 07's test uses; it is a given file like the others.

Steps

• 1. git switch p6; copy the given test. 2. See it fail.
• 3. batch.go. 4. The Batch field and the pool name.
• 5. gofmt -w; go test -race -count=3 ./internal/api/ → ok. 6. make gate.
• 7. Row p6/06-batch-wiring; commit.

git add internal/api docs/implementer-log.md
git commit -F .state/commit-msg.txt

Done when

• The given test and every existing api test pass under -race -count=3; make gate ok; the given file is byte-identical.