batch: workers — round-robin, serial per job, busy retry, cancel in flight, hooks

Start cfg.Workers goroutines in New that drain one rotation of live jobs in
submit order, each taking the next job with a pending target and none in flight,
so targets alternate across jobs and each job stays serial. One target per fetch
under the Close-cancelled context: panic recovers into ErrInternal (never
echoed), ErrRetryLater retries after RetryWait (bounded by RetryFor), otherwise
the target is stored done/failed and the hooks fire. Terminal status is set once
per job; cancelling a running job stores the in-flight target without overwriting
cancelled and leaves later targets pending.

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

docs/implementer-log.md

@@ -18,7 +18,8 @@ One row per task session that ends in a commit. Newest last.
 | p5c/01-extract-api | 2026-09-26 | done | 1 | gate: ok | none | `POST /v1/extract` answers a PDF posted in the body like `/v1/fetch` answers the same PDF by URL, nothing fetched. `internal/api/extract.go`: auth via `caller`; `io.ReadAll` of the body (a `*http.MaxBytesError`→413 "too large (>N MB)", empty→400); `clean.Kind` must be "pdf" else 415; query `pages`/`grep`/`question`/`mode` (digest when a question is set, else clean; other→400). Content-addressed cache: `hash="sha256:"+HashBody(body)`, `key=PDFKey("upload:"+hash, callerClass)`; a `GetPDF` hit rebuilds the doc via `pdfFromEntry` with `cached:true` + `PDFCacheHit` and no child (bypasses the TTL/revalidation that `pdfFromCache` does, so identical content is extracted once); a miss builds a synthetic `fetch.Resource{URL:"upload",...}` and runs `extractPDF`+`storePDF`. Bad pages/grep→400 via `writeFetchErr`, the chunk limit→the same 422 as `v1fetch`, extraction errors→`pdfError` via `writeFetchErr`. Envelope mirrors the PDF fetch path (`content`/`mode`/`untrusted`/`warnings`/`sizes`/`timings`/`pdf`/`digest`); `source` is `{"kind":"upload","hash","bytes","cached","fetched_at"}` with no `url` key; no Honcho context for uploads; one log line "extract" with caller/bytes/pdf_pages/mode/cached. `internal/api/api.go` scoped to the rule-1 changes: `DefaultPDFMaxUpload` const (100 MiB), `PDFMaxUpload int64` Server field, the `POST /v1/extract` route in `New`, and `ServeHTTP` (per-path upload cap instead of the 64 KiB JSON limit, and `/v1/extract` added to the `requestDeadline` switch). The `pdfMaxUpload()` field-or-default resolver lives in extract.go. Given test byte-identical; all api tests pass under `-race -count=3`; make gate ok. |
 | p5c/02-extract-cli | 2026-09-26 | done | 1 | gate: ok | none | `sift extract <file.pdf>` uploads a local PDF to `/v1/extract` and prints the answer exactly as `sift fetch` does. `cmd/sift/extractcmd.go`: `extractOpts` + `parseExtractOpts` (reuses `splitAroundPositional` with `extractValueFlags` mode/pages/grep/question, one positional file, `--json`), `extractURL` builds `base+"/v1/extract"` with `url.Values` set only for the options that are set and `mode=digest` when a question is given and no mode — values are URL-encoded so a grep like `a b&c` survives round-trip; `extractMain` reuses `loadClient`, `stat`s the file and refuses >100 MiB before uploading (exit 3, matching the server's 413→exitFor), then `http.NewRequest` with the `*os.File` as the streaming body and `ContentLength` set (no second in-memory copy), `Content-Type: application/pdf` + bearer, and the same error/exitFor path as fetch; success prints `content` (or the `--json` envelope). `main.go`: `case "extract"` and one help line. README: extract line in the CLI block plus a sentence; SKILL.md: a PDF-on-disk bullet in the PDFs section. Given test byte-identical; all cmd/sift tests pass under `-race -count=3`; make gate ok; `bin/sift extract` prints the usage line. |
 | p6/01-batchstore | 2026-09-26 | done | 1 | gate: ok | none | Storage-only batch store in `internal/dcache` (new `batch.go`; schema added to `Open`; `BatchH`/`BatchM` counters added to `DB`). `CreateBatchJob` stores as given. Reads never write the lease: `BatchJob`, `BatchTargets` and the counts are pure, so exact lease values hold after a read. `BatchJobByIDem` returns the newest `(caller,idemKey)` job with `lease+ttl>now` and status queued/running/done/partial; failed/cancelled never reused. `PutBatchTargets` inserts in order; `BatchTargets` returns idx order with NULLs as nil pointers. `UpdateBatchTarget` bumps lease to `now`, increments `targets_done` exactly once per target entering done/failed (previous-status check prevents re-write double-count), and adds `len(result)` to `nbytes` when a result is written — all in one tx. `SetBatchJobStatus`/`TouchBatchJob` set lease; `SweepBatch` deletes by `lease+ttl<=now` with its targets and returns the job count; `EvictBatchOldest` evicts whole non-live jobs oldest-lease-first (never a queued/running one) until `BatchNbytes()<=maxBytes`; `MarkInFlightFailed` fails queued/running on restart; `BatchNbytes` sums job nbytes, kept out of `Size()`. Accessors return `(v,bool)`/`error`, count misses in `BatchM`, reuse `ClassFleet`/`ClassRestricted`. A first pass failed `TestBatchIDemLookup` because I forgot `.Scan(&id)` on the `QueryRow` (the un-scanned `*sql.Row` was always non-nil, so the method always missed); fixed and verified. All dcache tests pass under `-race -count=3`; make gate ok; given file byte-identical.
-| p6/02-runner-core | 2026-09-26 | done | 1 | gate: ok | caller-cleanup hand-off to task 03 | Runner core in new internal/batch with no workers (task 03 adds them; jobs stay queued, so a queued job counts as live for the caps). Public surface in batch.go: Options/Target/Result/TargetError, FetchFunc, Job/TargetStatus, Hooks and Config with defaults (TTL 24h, 32 targets/job, 1 job/caller, 8 total, worker clamped to 2, 64 MiB, 2 s/10 m retry). Submit runs the given order — empty→ErrNoTargets, idempotency via hex SHA-256 of `name+NUL+json.Marshal(targets)` (a live hit returns the id reused, touching no caps), then TargetsPerJob/JobsPerCaller/JobsTotal caps, then stores the job queued (random 16-byte hex id, CreatedAt=Lease=now) and its targets (pending, Options=json of t.Options) before returning so a queued job is live; the submitting caller is held in memory by job id, never persisted. Status maps rows and decodes Result without writing; Touch moves the lease; Cancel sets a queued job cancelled at once and leaves its targets pending; Close is idempotent. Store errors are Warn-logged with the job id, never a URL/host. | |
+| p6/02-runner-core | 2026-09-26 | done | 1 | gate: ok | caller-cleanup hand-off to task 03 | Runner core in new internal/batch with no workers (task 03 adds them; jobs stay queued, so a queued job counts as live for the caps). Public surface in batch.go: Options/Target/Result/TargetError, FetchFunc, Job/TargetStatus, Hooks and Config with defaults (TTL 24h, 32 targets/job, 1 job/caller, 8 total, worker clamped to 2, 64 MiB, 2 s/10 m retry). Submit runs the given order — empty→ErrNoTargets, idempotency via hex SHA-256 of `name+NUL+json.Marshal(targets)` (a live hit returns the id reused, touching no caps), then TargetsPerJob/JobsPerCaller/JobsTotal caps, then stores the job queued (random 16-byte hex id, CreatedAt=Lease=now) and its targets (pending, Options=json of t.Options) before returning so a queued job is live; the submitting caller is held in memory by job id, never persisted. Status maps rows and decodes Result without writing; Touch moves the lease; Cancel sets a queued job cancelled at once and leaves its targets pending; Close is idempotent. Store errors are Warn-logged with the job id, never a URL/host. |
+| 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. | |
 
 ## P5a review (owner, 2026-09-26)
 

internal/batch/batch.go

@@ -121,6 +121,16 @@ type Runner struct {
 	callers   map[string]*auth.Caller // job id → submitting caller, held in memory only
 	callMu    sync.Mutex
 	closeOnce sync.Once
+
+	mu       sync.Mutex // guards everything below
+	cond     *sync.Cond // on mu
+	live     []*liveJob // the rotation, in submit order
+	cursor   int        // absolute rotation position; wraps at access, so it survives jobs being added
+	closed   bool
+	ctx      context.Context // the FetchFunc context; Close cancels it
+	stop     context.CancelFunc
+	wg       sync.WaitGroup // the workers
+	inflight int
 }
 
 func New(db *dcache.DB, fetch FetchFunc, cfg Config, log *slog.Logger) *Runner {
@@ -151,13 +161,22 @@ func New(db *dcache.DB, fetch FetchFunc, cfg Config, log *slog.Logger) *Runner {
 	if cfg.RetryFor == 0 {
 		cfg.RetryFor = 10 * time.Minute
 	}
-	return &Runner{
+	ctx, stop := context.WithCancel(context.Background())
+	r := &Runner{
 		db:      db,
 		fetch:   fetch,
 		log:     log,
 		cfg:     cfg,
 		callers: make(map[string]*auth.Caller),
+		ctx:     ctx,
+		stop:    stop,
 	}
+	r.cond = sync.NewCond(&r.mu)
+	r.wg.Add(cfg.Workers)
+	for i := 0; i < cfg.Workers; i++ {
+		go r.worker()
+	}
+	return r
 }
 
 // Submit stores the job and its targets before it returns. The order matters: an empty
@@ -229,14 +248,52 @@ func (r *Runner) Submit(ctx context.Context, c *auth.Caller, callerClass string,
 	r.callers[id] = c
 	r.callMu.Unlock()
 
+	r.mu.Lock()
+	r.live = append(r.live, &liveJob{
+		id:      id,
+		caller:  c,
+		targets: targets,
+		created: time.Unix(now, 0),
+	})
+	r.callDepth(len(r.live))
+	r.cond.Broadcast()
+	r.mu.Unlock()
+
 	r.callJob("created")
 	r.db.SweepBatch(now, r.cfg.TTL)
 	return id, false, nil
 }
 
-// Cancel sets a queued job cancelled at once; its pending targets stay pending. It is
+// Cancel cancels a job. A job still in the rotation (queued or with a target in flight) is marked
-// false for an unknown id or a job already terminal.
+// cancelled at once: its in-flight target, if any, finishes and is stored without overwriting
+// cancelled, and no further target starts. A job not in the rotation (unknown or already terminal)
+// keeps task 02's behaviour.
 func (r *Runner) Cancel(id string) bool {
+	r.mu.Lock()
+	var j *liveJob
+	for _, lj := range r.live {
+		if lj.id == id {
+			j = lj
+			break
+		}
+	}
+	if j != nil {
+		j.cancelled = true
+		if err := r.db.SetBatchJobStatus(id, "cancelled", "", time.Now().Unix()); err != nil {
+			r.log.Warn("batch: cancel job", "job", id, "err", err)
+		}
+		r.callJob("cancelled")
+		if !j.running {
+			r.removeLocked(j)
+			r.forgetCaller(id)
+			r.callDepth(len(r.live))
+		}
+		r.cond.Broadcast()
+		r.mu.Unlock()
+		return true
+	}
+	r.mu.Unlock()
+
 	jb, ok := r.db.BatchJob(id)
 	if !ok || isJobTerminal(jb.Status) {
 		return false
@@ -298,13 +355,20 @@ func (r *Runner) Touch(id string) {
 	}
 }
 
-// Close stops the workers and waits for an in-flight fetch to return. Safe to call more than
+// Close stops the workers, cancels their context, and waits for an in-flight fetch to return.
-// once. With no workers (task 02) there is nothing to stop; the in-memory callers are dropped.
+// Safe to call more than once. The in-memory callers are dropped.
 func (r *Runner) Close() {
 	r.closeOnce.Do(func() {
 		r.callMu.Lock()
 		r.callers = nil
 		r.callMu.Unlock()
+
+		r.mu.Lock()
+		r.closed = true
+		r.stop()
+		r.cond.Broadcast()
+		r.mu.Unlock()
+		r.wg.Wait()
 	})
 }
 

internal/batch/runner_test.go

@@ -0,0 +1,373 @@
+package batch
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"git.wntrmute.dev/kyle/sift/internal/auth"
+	"git.wntrmute.dev/kyle/sift/internal/dcache"
+)
+
+// Targets run one at a time, in order, and each fetch is given the submitting caller.
+func TestSerialOrderAndCaller(t *testing.T) {
+	var mu sync.Mutex
+	var calls, names []string
+	inflight, peak := 0, 0
+	fetch := func(_ context.Context, c *auth.Caller, tgt Target) (Result, error) {
+		mu.Lock()
+		inflight++
+		if inflight > peak {
+			peak = inflight
+		}
+		calls = append(calls, tgt.URL)
+		names = append(names, c.Name)
+		mu.Unlock()
+		time.Sleep(time.Millisecond)
+		mu.Lock()
+		inflight--
+		mu.Unlock()
+		return Result{Content: "ok " + tgt.URL}, nil
+	}
+	r := newRunner(t, fetch, Config{})
+	id, reused, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b", "c"))
+	if err != nil || reused {
+		t.Fatalf("submit err=%v reused=%v", err, reused)
+	}
+	waitStatus(t, r, id, "done")
+	mu.Lock()
+	defer mu.Unlock()
+	if strings.Join(calls, " ") != "https://a.example/ https://b.example/ https://c.example/" {
+		t.Fatalf("order %v", calls)
+	}
+	if peak != 1 {
+		t.Fatalf("peak inflight %d, want 1 (serial)", peak)
+	}
+	for _, n := range names {
+		if n != "alice" {
+			t.Fatalf("fetch saw caller %q, want alice", n)
+		}
+	}
+	j, _, _ := r.Status(id)
+	if j.Caller != "alice" || j.TargetsTotal != 3 || j.TargetsDone != 3 || j.CreatedAt == 0 {
+		t.Fatalf("job %+v", j)
+	}
+}
+
+// With one worker, two live jobs take turns target by target.
+func TestRoundRobin(t *testing.T) {
+	first := make(chan struct{})
+	release := make(chan struct{})
+	var once sync.Once
+	var mu sync.Mutex
+	var calls []string
+	fetch := func(_ context.Context, _ *auth.Caller, tgt Target) (Result, error) {
+		mu.Lock()
+		calls = append(calls, strings.TrimSuffix(strings.TrimPrefix(tgt.URL, "https://"), ".example/"))
+		mu.Unlock()
+		once.Do(func() { close(first); <-release })
+		return Result{Content: "ok"}, nil
+	}
+	r := newRunner(t, fetch, Config{Workers: 1})
+	a, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a1", "a2", "a3"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	<-first // a1 is in flight
+	b, _, err := r.Submit(context.Background(), bob, dcache.ClassFleet, urls("b1", "b2", "b3"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	close(release)
+	waitStatus(t, r, a, "done")
+	waitStatus(t, r, b, "done")
+	mu.Lock()
+	defer mu.Unlock()
+	if got := strings.Join(calls, " "); got != "a1 b1 a2 b2 a3 b3" {
+		t.Fatalf("order %q, want round-robin a1 b1 a2 b2 a3 b3", got)
+	}
+}
+
+func TestFailureIsolation(t *testing.T) {
+	fetch := func(_ context.Context, _ *auth.Caller, tgt Target) (Result, error) {
+		if tgt.URL == "https://b.example/" {
+			return Result{}, &TargetError{Msg: "upstream 500", Hint: "browser"}
+		}
+		return Result{Content: "ok " + tgt.URL}, nil
+	}
+	r := newRunner(t, fetch, Config{})
+	id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b", "c"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	waitStatus(t, r, id, "partial")
+	j, ts, ok := r.Status(id)
+	if !ok || j.TargetsDone != 3 {
+		t.Fatalf("job %+v ok=%v", j, ok)
+	}
+	if ts[0].Status != "done" || ts[1].Status != "failed" || ts[2].Status != "done" {
+		t.Fatalf("statuses %+v", ts)
+	}
+	if ts[1].Error != "upstream 500" || ts[1].Hint != "browser" {
+		t.Fatalf("failed target error=%q hint=%q", ts[1].Error, ts[1].Hint)
+	}
+	if ts[0].Result == nil || ts[0].Result.Content != "ok https://a.example/" {
+		t.Fatalf("result %+v", ts[0].Result)
+	}
+}
+
+func TestAllFailedIsFailed(t *testing.T) {
+	fetch := func(context.Context, *auth.Caller, Target) (Result, error) { return Result{}, errors.New("nope") }
+	r := newRunner(t, fetch, Config{})
+	id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	waitStatus(t, r, id, "failed")
+	_, ts, _ := r.Status(id)
+	if ts[0].Error != "nope" || ts[0].Hint != "" {
+		t.Fatalf("plain error stored as error=%q hint=%q", ts[0].Error, ts[0].Hint)
+	}
+}
+
+func TestIdempotentResubmit(t *testing.T) {
+	var mu sync.Mutex
+	n := 0
+	fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
+		mu.Lock()
+		n++
+		mu.Unlock()
+		return Result{Content: "ok"}, nil
+	}
+	r := newRunner(t, fetch, Config{})
+	id1, reused1, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
+	if err != nil || reused1 {
+		t.Fatalf("first submit err=%v reused=%v", err, reused1)
+	}
+	id2, reused2, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
+	if err != nil || !reused2 || id2 != id1 {
+		t.Fatalf("resubmit id=%q reused=%v err=%v", id2, reused2, err)
+	}
+	waitStatus(t, r, id1, "done")
+	// Still reused once done (within the lease), and never re-fetched.
+	if id3, reused3, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b")); err != nil || !reused3 || id3 != id1 {
+		t.Fatalf("resubmit after done id=%q reused=%v err=%v", id3, reused3, err)
+	}
+	// Another caller with the same list gets its own job.
+	if id4, reused4, err := r.Submit(context.Background(), bob, dcache.ClassFleet, urls("a", "b")); err != nil || reused4 || id4 == id1 {
+		t.Fatalf("other caller id=%q reused=%v err=%v", id4, reused4, err)
+	}
+	time.Sleep(50 * time.Millisecond)
+	mu.Lock()
+	defer mu.Unlock()
+	if n != 4 {
+		t.Fatalf("fetched %d times, want 4 (2 for alice's one job, 2 for bob's)", n)
+	}
+}
+
+// A cancelled job is not handed back to a resubmit: the same list starts a new job.
+func TestResubmitAfterCancelIsNew(t *testing.T) {
+	release := make(chan struct{})
+	fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
+		<-release
+		return Result{Content: "ok"}, nil
+	}
+	r := newRunner(t, fetch, Config{})
+	id1, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	waitStatus(t, r, id1, "running")
+	if !r.Cancel(id1) {
+		t.Fatal("cancel returned false")
+	}
+	close(release)
+	waitStatus(t, r, id1, "cancelled")
+	id2, reused, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
+	if err != nil || reused || id2 == id1 {
+		t.Fatalf("resubmit after cancel id=%q reused=%v err=%v", id2, reused, err)
+	}
+}
+
+// Submit returns with the job stored "queued"; it becomes "running" when a worker takes it.
+func TestQueuedThenRunning(t *testing.T) {
+	release := make(chan struct{})
+	fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
+		<-release
+		return Result{Content: "ok"}, nil
+	}
+	r := newRunner(t, fetch, Config{Workers: 1})
+	defer close(release)
+	a, _, _ := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a"))
+	waitStatus(t, r, a, "running")
+	b, _, err := r.Submit(context.Background(), bob, dcache.ClassFleet, urls("b"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	// The only worker is busy with alice's target, so bob's job waits in the queue.
+	time.Sleep(20 * time.Millisecond)
+	if j, _, _ := r.Status(b); j.Status != "queued" {
+		t.Fatalf("bob's job is %q while the only worker is busy, want queued", j.Status)
+	}
+}
+
+func TestCancel(t *testing.T) {
+	started := make(chan struct{}, 1)
+	release := make(chan struct{})
+	var mu sync.Mutex
+	var calls []string
+	fetch := func(_ context.Context, _ *auth.Caller, tgt Target) (Result, error) {
+		mu.Lock()
+		calls = append(calls, tgt.URL)
+		mu.Unlock()
+		select {
+		case started <- struct{}{}:
+		default:
+		}
+		<-release
+		return Result{Content: "ok"}, nil
+	}
+	r := newRunner(t, fetch, Config{})
+	id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b", "c"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	<-started // the first target is in flight
+	if !r.Cancel(id) {
+		t.Fatal("cancel returned false")
+	}
+	close(release) // let the in-flight target finish
+	waitStatus(t, r, id, "cancelled")
+	time.Sleep(20 * time.Millisecond)
+	mu.Lock()
+	n := len(calls)
+	mu.Unlock()
+	if n != 1 {
+		t.Fatalf("cancelled after %d fetches, want 1", n)
+	}
+	j, ts, ok := r.Status(id)
+	if !ok || j.Status != "cancelled" {
+		t.Fatalf("job %+v: the in-flight target's write must not overwrite cancelled", j)
+	}
+	if ts[0].Status != "done" {
+		t.Fatalf("in-flight target should be stored done: %+v", ts[0])
+	}
+	if ts[1].Status != "pending" || ts[2].Status != "pending" {
+		t.Fatalf("later targets must stay pending: %+v", ts[1:])
+	}
+	if r.Cancel(id) {
+		t.Fatal("cancelling a terminal job must return false")
+	}
+	if r.Cancel("no-such-job") {
+		t.Fatal("cancelling an unknown job must return false")
+	}
+}
+
+// A target that reports ErrRetryLater (the fetch gate was busy) is retried, not failed.
+func TestBusyIsRetried(t *testing.T) {
+	var mu sync.Mutex
+	n := 0
+	fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
+		mu.Lock()
+		defer mu.Unlock()
+		n++
+		if n <= 2 {
+			return Result{}, fmt.Errorf("busy: %w", ErrRetryLater)
+		}
+		return Result{Content: "ok"}, nil
+	}
+	r := newRunner(t, fetch, Config{RetryWait: time.Millisecond})
+	id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	waitStatus(t, r, id, "done")
+	mu.Lock()
+	defer mu.Unlock()
+	if n != 3 {
+		t.Fatalf("fetched %d times, want 3 (two busy, then ok)", n)
+	}
+}
+
+// Retrying is bounded: a target busy for longer than RetryFor fails with a busy error.
+func TestBusyGivesUp(t *testing.T) {
+	fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
+		return Result{}, fmt.Errorf("busy: %w", ErrRetryLater)
+	}
+	r := newRunner(t, fetch, Config{RetryWait: time.Millisecond, RetryFor: 30 * time.Millisecond})
+	id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	waitStatus(t, r, id, "failed")
+	_, ts, _ := r.Status(id)
+	if ts[0].Status != "failed" || !strings.Contains(ts[0].Error, "busy") {
+		t.Fatalf("target %+v, want failed with a busy error", ts[0])
+	}
+}
+
+func TestPanicRecovered(t *testing.T) {
+	fetch := func(_ context.Context, _ *auth.Caller, tgt Target) (Result, error) {
+		if tgt.URL == "https://b.example/" {
+			panic("boom")
+		}
+		return Result{Content: "ok"}, nil
+	}
+	r := newRunner(t, fetch, Config{})
+	id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b", "c"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	waitStatus(t, r, id, "partial")
+	_, ts, _ := r.Status(id)
+	if ts[0].Status != "done" || ts[1].Status != "failed" || ts[2].Status != "done" {
+		t.Fatalf("statuses %+v", ts)
+	}
+	if ts[1].Error == "" || strings.Contains(ts[1].Error, "boom") {
+		t.Fatalf("panic target error %q: must be set, and must not echo the panic value", ts[1].Error)
+	}
+}
+
+// The hooks see every job outcome once and every stored target once.
+func TestHooks(t *testing.T) {
+	var mu sync.Mutex
+	jobs := map[string]int{}
+	var targets, jobSecs int
+	var bytes int
+	h := Hooks{
+		Job: func(o string) { mu.Lock(); jobs[o]++; mu.Unlock() },
+		Target: func(err error, _ time.Duration, n int) {
+			mu.Lock()
+			targets++
+			bytes += n
+			mu.Unlock()
+		},
+		JobSeconds: func(time.Duration) { mu.Lock(); jobSecs++; mu.Unlock() },
+	}
+	fetch := func(_ context.Context, _ *auth.Caller, tgt Target) (Result, error) {
+		if tgt.URL == "https://b.example/" {
+			return Result{}, errors.New("nope")
+		}
+		return Result{Content: "ok"}, nil
+	}
+	r := newRunner(t, fetch, Config{Hooks: h})
+	id, _, _ := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
+	waitStatus(t, r, id, "partial")
+	_, _, _ = r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b")) // reused
+	time.Sleep(20 * time.Millisecond)
+	mu.Lock()
+	defer mu.Unlock()
+	if jobs["created"] != 1 || jobs["partial"] != 1 || jobs["reused"] != 1 || len(jobs) != 3 {
+		t.Fatalf("job outcomes %v", jobs)
+	}
+	if targets != 2 || jobSecs != 1 {
+		t.Fatalf("targets=%d jobSeconds=%d, want 2 and 1", targets, jobSecs)
+	}
+	if bytes <= 0 {
+		t.Fatalf("stored result bytes %d, want > 0 for the done target", bytes)
+	}
+}

internal/batch/worker.go

@@ -0,0 +1,271 @@
+package batch
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"git.wntrmute.dev/kyle/sift/internal/auth"
+)
+
+// ErrInternal is stored for a target whose fetch panicked; the panic value is never echoed.
+var ErrInternal = errors.New("internal error")
+
+// liveJob is a submitted job that the workers are draining. It lives in Runner.live until the
+// last target is stored or the job is cancelled.
+type liveJob struct {
+	id           string
+	caller       *auth.Caller
+	targets      []Target // as submitted
+	created      time.Time
+	next         int  // index of the next target to fetch
+	running      bool // a target of this job is in flight
+	started      bool // set to "running" in the store already
+	cancelled    bool
+	retryAt      time.Time // zero, or when the busy target may be tried again
+	busySince    time.Time // zero, or when the current target first came back busy
+	done, failed int
+}
+
+// worker drains the rotation: it takes the next live job with a pending target and no target in
+// flight, fetches and stores that one target, and moves on. Each job is serial — its target i+1
+// never starts before i is stored.
+func (r *Runner) worker() {
+	defer r.wg.Done()
+
+	for {
+		r.mu.Lock()
+		now := time.Now()
+		j := r.pick(now)
+		if j == nil {
+			if r.closed {
+				r.mu.Unlock()
+				return
+			}
+			r.scheduleRetryLocked(now)
+			r.cond.Wait() // re-locks on wake; release before the next top-of-loop Lock
+			r.mu.Unlock()
+			continue
+		}
+
+		// Take the target: mark the job running and bump the in-flight count under the lock,
+		// then fetch outside it.
+		j.running = true
+		r.inflight++
+		r.callInflight(r.inflight)
+		if !j.started {
+			if err := r.db.SetBatchJobStatus(j.id, "running", "", time.Now().Unix()); err != nil {
+				r.log.Warn("batch: set running", "job", j.id, "err", err)
+			}
+			j.started = true
+		}
+		idx := j.next
+		target := j.targets[idx]
+		startedAt := time.Now()
+		r.mu.Unlock()
+
+		res, err := r.runTarget(r.ctx, j, idx, target)
+
+		r.mu.Lock()
+		j.running = false
+		r.inflight--
+		r.callInflight(r.inflight)
+
+		// A target that was only busy stays pending and is retried after RetryWait, unless it
+		// has been busy for longer than RetryFor, in which case it is stored failed below.
+		if errors.Is(err, ErrRetryLater) {
+			if j.busySince.IsZero() {
+				j.busySince = now
+			}
+			if now.Sub(j.busySince) <= r.cfg.RetryFor {
+				j.retryAt = now.Add(r.cfg.RetryWait)
+				r.cond.Broadcast()
+				r.mu.Unlock()
+				continue
+			}
+		}
+
+		resultBytes := r.storeTarget(j, idx, res, err)
+		j.next++
+		j.busySince, j.retryAt = time.Time{}, time.Time{}
+		if err == nil {
+			j.done++
+		} else {
+			j.failed++
+		}
+		r.callTarget(err, time.Since(startedAt), resultBytes)
+		r.db.EvictBatchOldest(r.cfg.MaxBytes)
+
+		if j.cancelled || j.next >= len(j.targets) {
+			if !j.cancelled {
+				r.setTerminal(j)
+			}
+			r.removeLocked(j)
+			r.forgetCaller(j.id)
+			r.callDepth(len(r.live))
+		}
+		r.cond.Broadcast()
+		r.mu.Unlock()
+	}
+}
+
+// pick returns the first live job, scanning from cursor and wrapping once, that has a pending
+// target that is not running, not cancelled, and ready (now past any retry). It advances cursor to
+// the index after the job, so the rotation is round-robin across jobs in submit order.
+func (r *Runner) pick(now time.Time) *liveJob {
+	n := len(r.live)
+	if n == 0 {
+		return nil
+	}
+	for i := 0; i < n; i++ {
+		j := r.live[(r.cursor+i)%n]
+		if j.running || j.cancelled || j.next >= len(j.targets) {
+			continue
+		}
+		if !now.Before(j.retryAt) { // now >= retryAt; a zero retryAt is always ready
+			r.cursor += i + 1 // absolute; wraps at access, so the rotation survives jobs being added
+			return j
+		}
+	}
+	return nil
+}
+
+// scheduleRetryLocked arms a timer that wakes the workers at the earliest future retryAt, so they
+// do not sleep past a busy target that is ready to be tried again.
+func (r *Runner) scheduleRetryLocked(now time.Time) {
+	var earliest time.Time
+	for _, j := range r.live {
+		if j.cancelled || j.next >= len(j.targets) || !j.retryAt.After(now) {
+			continue
+		}
+		if earliest.IsZero() || j.retryAt.Before(earliest) {
+			earliest = j.retryAt
+		}
+	}
+	if !earliest.IsZero() {
+		d := time.Until(earliest)
+		if d < 0 {
+			d = 0
+		}
+		time.AfterFunc(d, r.cond.Broadcast)
+	}
+}
+
+// removeLocked drops j from live and fixes cursor so the rotation still points at the next job.
+func (r *Runner) removeLocked(j *liveJob) {
+	idx := -1
+	for i := range r.live {
+		if r.live[i] == j {
+			idx = i
+			break
+		}
+	}
+	if idx < 0 {
+		return
+	}
+	r.live = append(r.live[:idx], r.live[idx+1:]...)
+	if r.cursor > idx {
+		r.cursor--
+	}
+}
+
+// setTerminal writes the job's finished status once: all done → done, all failed → failed, a mix
+// → partial.
+func (r *Runner) setTerminal(j *liveJob) {
+	var status string
+	switch {
+	case j.failed == 0:
+		status = "done"
+	case j.done == 0:
+		status = "failed"
+	default:
+		status = "partial"
+	}
+	if err := r.db.SetBatchJobStatus(j.id, status, "", time.Now().Unix()); err != nil {
+		r.log.Warn("batch: set terminal status", "job", j.id, "err", err)
+	}
+	r.callJob(status)
+	r.callJobSeconds(time.Since(j.created))
+}
+
+// runTarget fetches one target, recovering a panic into ErrInternal (never echoing the value).
+func (r *Runner) runTarget(ctx context.Context, j *liveJob, idx int, t Target) (res Result, err error) {
+	defer func() {
+		if rec := recover(); rec != nil {
+			r.log.Warn("batch: recover fetch panic", "job", j.id, "idx", idx)
+			res = Result{}
+			err = ErrInternal
+		}
+	}()
+	return r.fetch(ctx, j.caller, t)
+}
+
+// storeTarget stores one target: success → done with the result JSON, error → failed with the
+// TargetError's Msg/Hint or the error's text. It returns the number of result bytes stored.
+func (r *Runner) storeTarget(j *liveJob, idx int, res Result, err error) int {
+	now := time.Now().Unix()
+	status := "failed"
+	var resultStr, errMsg, hint *string
+	if err == nil {
+		status = "done"
+		b, merr := json.Marshal(res)
+		if merr != nil {
+			r.log.Warn("batch: marshal result", "job", j.id, "idx", idx, "err", merr)
+		} else {
+			s := string(b)
+			resultStr = &s
+		}
+	} else {
+		var terr *TargetError
+		if errors.As(err, &terr) {
+			m := terr.Msg
+			errMsg = &m
+			if terr.Hint != "" {
+				h := terr.Hint
+				hint = &h
+			}
+		} else {
+			e := err.Error()
+			errMsg = &e
+		}
+	}
+	if serr := r.db.UpdateBatchTarget(j.id, idx, now, status, resultStr, errMsg, hint); serr != nil {
+		r.log.Warn("batch: store target", "job", j.id, "idx", idx, "err", serr)
+	}
+	if resultStr != nil {
+		return len(*resultStr)
+	}
+	return 0
+}
+
+// forgetCaller drops the in-memory caller for a finished job.
+func (r *Runner) forgetCaller(id string) {
+	r.callMu.Lock()
+	delete(r.callers, id)
+	r.callMu.Unlock()
+}
+
+func (r *Runner) callDepth(live int) {
+	if r.cfg.Hooks.Depth != nil {
+		r.cfg.Hooks.Depth(live)
+	}
+}
+
+func (r *Runner) callInflight(n int) {
+	if r.cfg.Hooks.Inflight != nil {
+		r.cfg.Hooks.Inflight(n)
+	}
+}
+
+func (r *Runner) callTarget(err error, d time.Duration, resultBytes int) {
+	if r.cfg.Hooks.Target != nil {
+		r.cfg.Hooks.Target(err, d, resultBytes)
+	}
+}
+
+func (r *Runner) callJobSeconds(d time.Duration) {
+	if r.cfg.Hooks.JobSeconds != nil {
+		r.cfg.Hooks.JobSeconds(d)
+	}
+}