serve: [batch] config; open the store, reap jobs killed by a restart, start the runner

Add the [batch] config block (Batch type + File.Batch) and ParseSize (binary
KiB/MiB/GiB byte counts) in internal/config. serve now opens the dcache store
when batch is on, marks the queued/running jobs a restart killed as failed
(never resumed) and observes them, builds the runner from the config with the
runner's defaults filling zeros, and closes the runner on both shutdown paths.
Batch is on by default; disabled = true leaves srv.Batch nil so the endpoints
answer 503.

Co-Authored-By: ornith-1.5-35b-a3b <ornith-1.5-35b-a3b@llama.cpp.invalid>

cmd/sift/serve.go

@@ -10,9 +10,11 @@ import (
 	"os/signal"
 	"strings"
 	"syscall"
+	"time"
 
 	"git.wntrmute.dev/kyle/sift/internal/api"
 	"git.wntrmute.dev/kyle/sift/internal/auth"
+	"git.wntrmute.dev/kyle/sift/internal/batch"
 	"git.wntrmute.dev/kyle/sift/internal/config"
 	"git.wntrmute.dev/kyle/sift/internal/dcache"
 	"git.wntrmute.dev/kyle/sift/internal/digest"
@@ -119,7 +121,7 @@ func serveMain() {
 		cachePath = "/var/lib/sift/cache.db"
 	}
 	var cache *dcache.DB
-	if cfg.Dixie.Chat != "" || cfg.Search.AnyEnabled() {
+	if cfg.Dixie.Chat != "" || cfg.Search.AnyEnabled() || !cfg.Batch.Disabled {
 		c, err := dcache.Open(cachePath, 500<<20, log)
 		if err != nil {
 			log.Warn("cache", "err", err)
@@ -127,6 +129,23 @@ func serveMain() {
 			cache = c
 		}
 	}
+	if !cfg.Batch.Disabled && cache != nil {
+		now := time.Now().Unix()
+		if n := cache.MarkInFlightFailed("restart", now); n > 0 {
+			for i := 0; i < n; i++ {
+				srv.Prom.ObserveBatchJob("restart")
+			}
+			log.Info("batch reaped", "jobs", n)
+		}
+		srv.Batch = srv.NewBatchRunner(cache, batch.Config{
+			TTL:           config.ParseDuration(cfg.Batch.TTL, 0),
+			TargetsPerJob: cfg.Batch.TargetsPerJob,
+			JobsPerCaller: cfg.Batch.JobsPerCaller,
+			JobsTotal:     cfg.Batch.JobsTotal,
+			Workers:       cfg.Batch.Workers,
+			MaxBytes:      config.ParseSize(cfg.Batch.MaxBytes, 0),
+		})
+	}
 	if cfg.Dixie.Chat != "" {
 		srv.Digest = &digest.Engine{
 			ChatURL:    cfg.Dixie.Chat,
@@ -213,8 +232,14 @@ func serveMain() {
 	case <-sig:
 		stop()
 		_ = hs.Close()
+		if srv.Batch != nil {
+			srv.Batch.Close()
+		}
 	case err := <-errCh:
 		stop()
+		if srv.Batch != nil {
+			srv.Batch.Close()
+		}
 		if err != nil {
 			log.Error("serve", "err", err)
 			os.Exit(1)

dist/sift.toml.example

@@ -54,3 +54,12 @@ listen = "127.0.0.1:8890"
 # url = "https://api.tavily.com/search"
 # daily_credit_cap = 500
 # no_rerank = true
+
+# [batch]
+# disabled = false
+# targets_per_job = 32
+# jobs_per_caller = 1
+# jobs_total = 8
+# workers = 1          # at most 2: each worker fetches one target at a time
+# ttl = "24h"          # a job's results live this long after its last read or write
+# max_bytes = "64MiB"  # stored results; the oldest finished jobs go first

docs/implementer-log.md

@@ -24,6 +24,7 @@ One row per task session that ends in a commit. Newest last.
 | 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. |
 | p6/07-api-endpoints | 2026-09-26 | done | 1 | gate: ok | none | The four /v1/batch handlers in new `internal/api/batch.go`: `batchSubmit` (POST) decodes `{"targets":[{url,mode,question,pages,grep,jq,context,recover}]}` flat like `/v1/fetch`, 400 "invalid batch" on bad json/empty list/empty url/bad mode (only empty,clean,raw,digest), then `PreCheck` per target → 422 "blocked: target <idx>: <reason>" with no job, then `Submit` mapping ErrTooManyTargets→422 "too many targets (max 32)", ErrCallerBusy→429+Retry-After 30, ErrBusy→429+Retry-After 60, ErrNoTargets→400, and 202 {success,job_id,status,targets,reused} (status "queued" new, current status reused). `batchStatus` (GET) and `batchResult` (GET {id}/{idx}) and `batchCancel` (DELETE) share `ownership` (unknown or not `c.Name` → 404, never 403; nil batch → 503); status is a summary with no results unless `?results=1` inlines the envelope (Meta+content) on done targets, idx validated 0..targets_total-1 else 404, cancel of a terminal job → 409 "job already finished" else 200 {success,status:cancelled}. `api.go` scoped to the rule-2 changes: `batchMaxBody` (256 KiB) const, the four routes in `New`, and `ServeHTTP` applying the per-path cap once (`/v1/batch`→256 KiB, `/v1/extract`→pdfMaxUpload, else 64 KiB) and collapsing every batch path to a fixed metrics label via `batchObservePath` (`/v1/batch` or `/v1/batch/{id}`, never a job id). Each handler logs one `batch` line with caller/job_id/targets/status-or-reused and never a URL/hostname/idem_key. Given test byte-identical; all api tests pass under -race -count=3; make gate ok. |
+| p6/08-serve-config | 2026-09-26 | done | 1 | gate: ok | none | `[batch]` config in `internal/config`: the `Batch` type (disabled/targets_per_job/jobs_per_caller/jobs_total/workers/ttl/max_bytes) and `File.Batch`, plus `ParseSize` (trims spaces, accepts only binary KiB/MiB/GiB suffixes or a plain byte count, zero/negative/other → def). serve.go opens the dcache store when batch is on (`!cfg.Batch.Disabled`) alongside dixie/search — same path, same call, Warn on failure — then, before building the runner, reaps in-flight jobs a restart killed via `MarkInFlightFailed("restart", now)` with one `ObserveBatchJob("restart")` per reaped job and a `batch reaped` info log when n>0; builds `srv.Batch` from the config (TTL via `ParseDuration`, MaxBytes via `ParseSize`, runner defaults fill the zeros) only when enabled and the store opened; closes `srv.Batch` in both shutdown branches when non-nil. Added a commented `[batch]` block to dist/sift.toml.example. Given test byte-identical; all config and cmd tests pass under -race -count=3. |
 
 ## P5a review (owner, 2026-09-26)
 

internal/config/batch_test.go

@@ -0,0 +1,56 @@
+package config
+
+// P6 task 08: the [batch] block and ParseSize.
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestBatchBlock(t *testing.T) {
+	p := filepath.Join(t.TempDir(), "sift.toml")
+	body := "[batch]\ndisabled = false\ntargets_per_job = 16\njobs_per_caller = 2\njobs_total = 4\nworkers = 2\nttl = \"12h\"\nmax_bytes = \"32MiB\"\n"
+	if err := os.WriteFile(p, []byte(body), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	f, err := Load(p)
+	if err != nil {
+		t.Fatal(err)
+	}
+	b := f.Batch
+	if b.Disabled || b.TargetsPerJob != 16 || b.JobsPerCaller != 2 || b.JobsTotal != 4 || b.Workers != 2 || b.TTL != "12h" || b.MaxBytes != "32MiB" {
+		t.Fatalf("batch block %+v", b)
+	}
+	// No block: zero values, and batch is on (the runner applies its own defaults).
+	empty := filepath.Join(t.TempDir(), "empty.toml")
+	if err := os.WriteFile(empty, []byte("listen = \":8443\"\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if f, err := Load(empty); err != nil || f.Batch != (Batch{}) {
+		t.Fatalf("no [batch] block: %+v %v", f.Batch, err)
+	}
+}
+
+func TestParseSize(t *testing.T) {
+	const def = 7
+	for _, c := range []struct {
+		in   string
+		want int64
+	}{
+		{"64MiB", 64 << 20},
+		{"512KiB", 512 << 10},
+		{"1GiB", 1 << 30},
+		{"1000", 1000},
+		{" 2MiB ", 2 << 20},
+		{"", def},
+		{"junk", def},
+		{"-1MiB", def},
+		{"0", def},
+		{"12MB", def}, // decimal units are not accepted: say MiB
+	} {
+		if got := ParseSize(c.in, def); got != c.want {
+			t.Errorf("ParseSize(%q) = %d, want %d", c.in, got, c.want)
+		}
+	}
+}

internal/config/config.go

@@ -2,6 +2,8 @@ package config
 
 import (
 	"os"
+	"strconv"
+	"strings"
 	"time"
 
 	"github.com/pelletier/go-toml/v2"
@@ -17,6 +19,19 @@ type File struct {
 	Digest       Digest   `toml:"digest"`
 	Context      Context  `toml:"context"`
 	Search       Search   `toml:"search"`
+	Batch        Batch    `toml:"batch"`
+}
+
+// Batch is the [batch] block. Batch is on by default: no block means the runner runs with its own
+// defaults. disabled = true turns it off and the endpoints answer 503.
+type Batch struct {
+	Disabled      bool   `toml:"disabled"`
+	TargetsPerJob int    `toml:"targets_per_job"`
+	JobsPerCaller int    `toml:"jobs_per_caller"`
+	JobsTotal     int    `toml:"jobs_total"`
+	Workers       int    `toml:"workers"`
+	TTL           string `toml:"ttl"`
+	MaxBytes      string `toml:"max_bytes"`
 }
 
 type Dixie struct {
@@ -110,3 +125,28 @@ func ParseDuration(s string, def time.Duration) time.Duration {
 	}
 	return d
 }
+
+// ParseSize parses a positive byte count: a plain integer, or one with a binary suffix KiB, MiB or
+// GiB. Anything else, zero or negative returns def.
+func ParseSize(s string, def int64) int64 {
+	s = strings.TrimSpace(s)
+	if s == "" {
+		return def
+	}
+	var mult int64
+	switch {
+	case strings.HasSuffix(s, "KiB"):
+		mult, s = 1<<10, s[:len(s)-3]
+	case strings.HasSuffix(s, "MiB"):
+		mult, s = 1<<20, s[:len(s)-3]
+	case strings.HasSuffix(s, "GiB"):
+		mult, s = 1<<30, s[:len(s)-3]
+	default:
+		mult = 1
+	}
+	n, err := strconv.ParseInt(s, 10, 64)
+	if err != nil || n <= 0 {
+		return def
+	}
+	return n * mult
+}