dcache: batch job and target tables with a sliding lease, sweep and byte budget

Add the batch job store to internal/dcache (storage only; no HTTP, no worker).
New internal/dcache/batch.go holds the BatchJob/BatchTarget types and all
accessors; Open gains the batch_job/batch_target tables and indices, and DB
gains BatchH/BatchM counters.

Sliding lease: writes (UpdateBatchTarget, SetBatchJobStatus, TouchBatchJob,
MarkInFlightFailed) bump the job lease to now; reads never write. BatchJobByIDem
returns the newest (caller, idem_key) job with lease+ttl>now and status
queued/running/done/partial. UpdateBatchTarget counts targets_done exactly once
per target entering done/failed and adds len(result) to nbytes. SweepBatch
removes by lease; EvictBatchOldest evicts whole non-live jobs oldest-lease-first;
MarkInFlightFailed reconciles in-flight jobs on restart. Batch nbytes are kept out
of Size() and the content evict union.

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

docs/implementer-log.md

@@ -17,6 +17,7 @@ One row per task session that ends in a commit. Newest last.
 | p5b/03-api | 2026-09-26 | done | 1 | gate: ok | none | The digest now answers long documents itself (task 02), so the API's refusal limit is only a safety cap: `DefaultPDFMaxQuestionChunks` 200→5000 in `internal/api/pdf.go` (the 422 message text unchanged, still "document too long for question (N chunks); use grep or pages"). In `v1fetch` (`internal/api/api.go`) the digest-meta map gains `"retrieval": dr.Retrieval` and `"embedded": dr.Embedded` next to `chunks_used`/`chunks_total`, only when the digest ran without a fallback. Copied the two given tests byte-identical from the plan — the replacement `pdf_test.go` is now the protected one and asserts `DefaultPDFMaxQuestionChunks == 5000`; `pdf_retrieval_test.go` asserts the first long-doc question returns `retrieval=bm25`, `embedded=0`. README PDF section: a long-doc question is answered first from BM25 (`retrieval: "bm25"`) while its chunks embed in the background and hybrid on later questions, plus the 5000-chunk cap. One sentence added to `skill/SKILL.md`. All api tests pass under `-race -count=3`; make gate ok; given files byte-identical. |
 | 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. |
 
 ## P5a review (owner, 2026-09-26)
 

internal/dcache/batch.go

@@ -0,0 +1,314 @@
+package dcache
+
+import (
+	"database/sql"
+	"fmt"
+	"time"
+)
+
+// terminal statuses mark a target as finished; a target already in one is not
+// counted again on a re-write.
+func isTerminal(status string) bool {
+	return status == "done" || status == "failed"
+}
+
+type BatchJob struct {
+	ID           string
+	Caller       string
+	CallerClass  string
+	IDemKey      string
+	Status       string
+	Error        string
+	TargetsTotal int
+	TargetsDone  int
+	CreatedAt    int64
+	Lease        int64
+	NBytes       int64
+}
+
+type BatchTarget struct {
+	JobID   string
+	Idx     int
+	URL     string
+	Options string
+	Status  string
+	Result  *string
+	Error   *string
+	Hint    *string
+}
+
+func nullString(s *string) sql.NullString {
+	if s == nil {
+		return sql.NullString{Valid: false}
+	}
+	return sql.NullString{String: *s, Valid: true}
+}
+
+func nullPtr(n sql.NullString) *string {
+	if !n.Valid {
+		return nil
+	}
+	s := n.String
+	return &s
+}
+
+func (d *DB) CreateBatchJob(j BatchJob) error {
+	if d == nil {
+		return fmt.Errorf("dcache: nil db")
+	}
+	_, err := d.sql.Exec(`
+		INSERT INTO batch_job
+			(id, caller, caller_class, idem_key, status, error, targets_total, targets_done, created_at, lease, nbytes)
+		VALUES(?,?,?,?,?,?,?,?,?,?,?)
+	`, j.ID, j.Caller, j.CallerClass, j.IDemKey, j.Status, j.Error, j.TargetsTotal, j.TargetsDone, j.CreatedAt, j.Lease, j.NBytes)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (d *DB) BatchJob(id string) (BatchJob, bool) {
+	if d == nil {
+		return BatchJob{}, false
+	}
+	var j BatchJob
+	err := d.sql.QueryRow(`
+		SELECT id, caller, caller_class, idem_key, status, error,
+		       targets_total, targets_done, created_at, lease, nbytes
+		FROM batch_job WHERE id = ?
+	`, id).Scan(&j.ID, &j.Caller, &j.CallerClass, &j.IDemKey, &j.Status, &j.Error,
+		&j.TargetsTotal, &j.TargetsDone, &j.CreatedAt, &j.Lease, &j.NBytes)
+	if err != nil {
+		d.mu.Lock()
+		d.BatchM++
+		d.mu.Unlock()
+		return BatchJob{}, false
+	}
+	d.mu.Lock()
+	d.BatchH++
+	d.mu.Unlock()
+	return j, true
+}
+
+// BatchJobByIDem returns the id of the newest job for (caller, idemKey) that is
+// still within its lease and not failed/cancelled. Reads never write.
+func (d *DB) BatchJobByIDem(caller, idemKey string, now int64, ttl time.Duration) (string, bool) {
+	if d == nil {
+		return "", false
+	}
+	ttlSec := int64(ttl.Seconds())
+	var id string
+	err := d.sql.QueryRow(`
+		SELECT id FROM batch_job
+		WHERE caller = ? AND idem_key = ? AND lease + ? > ?
+			AND status IN ('queued','running','done','partial')
+		ORDER BY created_at DESC LIMIT 1
+	`, caller, idemKey, ttlSec, now).Scan(&id)
+	if err != nil {
+		return "", false
+	}
+	return id, true
+}
+
+func (d *DB) LiveBatchJobs(caller string) int {
+	if d == nil {
+		return 0
+	}
+	var n int
+	_ = d.sql.QueryRow(`
+		SELECT COUNT(*) FROM batch_job WHERE caller = ? AND status IN ('queued','running')
+	`, caller).Scan(&n)
+	return n
+}
+
+func (d *DB) LiveBatchJobsTotal() int {
+	if d == nil {
+		return 0
+	}
+	var n int
+	_ = d.sql.QueryRow(`SELECT COUNT(*) FROM batch_job WHERE status IN ('queued','running')`).Scan(&n)
+	return n
+}
+
+func (d *DB) PutBatchTargets(jobID string, targets []BatchTarget) error {
+	if d == nil {
+		return fmt.Errorf("dcache: nil db")
+	}
+	for _, t := range targets {
+		if _, err := d.sql.Exec(`
+			INSERT INTO batch_target(job_id, idx, url, options, status, result, error, hint)
+			VALUES(?,?,?,?,?,?,?,?)
+		`, t.JobID, t.Idx, t.URL, t.Options, t.Status,
+			nullString(t.Result), nullString(t.Error), nullString(t.Hint)); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (d *DB) BatchTargets(jobID string) ([]BatchTarget, bool) {
+	if d == nil {
+		return nil, false
+	}
+	rows, err := d.sql.Query(`
+		SELECT job_id, idx, url, options, status, result, error, hint
+		FROM batch_target WHERE job_id = ? ORDER BY idx ASC
+	`, jobID)
+	if err != nil {
+		d.mu.Lock()
+		d.BatchM++
+		d.mu.Unlock()
+		return nil, false
+	}
+	defer rows.Close()
+	var out []BatchTarget
+	for rows.Next() {
+		var (
+			r, e, h sql.NullString
+			t       BatchTarget
+		)
+		if err := rows.Scan(&t.JobID, &t.Idx, &t.URL, &t.Options, &t.Status, &r, &e, &h); err != nil {
+			d.mu.Lock()
+			d.BatchM++
+			d.mu.Unlock()
+			return nil, false
+		}
+		t.Result = nullPtr(r)
+		t.Error = nullPtr(e)
+		t.Hint = nullPtr(h)
+		out = append(out, t)
+	}
+	if err := rows.Err(); err != nil {
+		d.mu.Lock()
+		d.BatchM++
+		d.mu.Unlock()
+		return nil, false
+	}
+	if len(out) == 0 {
+		return nil, false
+	}
+	return out, true
+}
+
+func (d *DB) UpdateBatchTarget(jobID string, idx int, now int64, status string, result, errMsg, hint *string) error {
+	if d == nil {
+		return fmt.Errorf("dcache: nil db")
+	}
+	tx, err := d.sql.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	var prev string
+	err = tx.QueryRow(`SELECT status FROM batch_target WHERE job_id = ? AND idx = ?`, jobID, idx).Scan(&prev)
+	if err != nil {
+		return err
+	}
+
+	var resBytes int64
+	if result != nil {
+		resBytes = int64(len(*result))
+	}
+	deltaDone := 0
+	if isTerminal(status) && !isTerminal(prev) {
+		deltaDone = 1
+	}
+
+	if _, err := tx.Exec(`
+		UPDATE batch_target SET status = ?, result = ?, error = ?, hint = ? WHERE job_id = ? AND idx = ?
+	`, status, nullString(result), nullString(errMsg), nullString(hint), jobID, idx); err != nil {
+		return err
+	}
+	if _, err := tx.Exec(`
+		UPDATE batch_job SET lease = ?, targets_done = targets_done + ?, nbytes = nbytes + ? WHERE id = ?
+	`, now, deltaDone, resBytes, jobID); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
+func (d *DB) SetBatchJobStatus(id, status, errMsg string, now int64) error {
+	if d == nil {
+		return fmt.Errorf("dcache: nil db")
+	}
+	_, err := d.sql.Exec(`UPDATE batch_job SET status = ?, error = ?, lease = ? WHERE id = ?`, status, errMsg, now, id)
+	return err
+}
+
+func (d *DB) TouchBatchJob(id string, now int64) error {
+	if d == nil {
+		return fmt.Errorf("dcache: nil db")
+	}
+	_, err := d.sql.Exec(`UPDATE batch_job SET lease = ? WHERE id = ?`, now, id)
+	return err
+}
+
+func (d *DB) SweepBatch(now int64, ttl time.Duration) int {
+	if d == nil {
+		return 0
+	}
+	ttlSec := int64(ttl.Seconds())
+	if _, err := d.sql.Exec(`
+		DELETE FROM batch_target WHERE job_id IN
+			(SELECT id FROM batch_job WHERE lease + ? <= ?)
+	`, ttlSec, now); err != nil {
+		return 0
+	}
+	res, err := d.sql.Exec(`DELETE FROM batch_job WHERE lease + ? <= ?`, ttlSec, now)
+	if err != nil {
+		return 0
+	}
+	n, _ := res.RowsAffected()
+	return int(n)
+}
+
+func (d *DB) EvictBatchOldest(maxBytes int64) int {
+	if d == nil {
+		return 0
+	}
+	evicted := 0
+	for d.BatchNbytes() > maxBytes {
+		var id string
+		err := d.sql.QueryRow(`
+			SELECT id FROM batch_job
+			WHERE status NOT IN ('queued','running')
+			ORDER BY lease ASC, id ASC
+			LIMIT 1
+		`).Scan(&id)
+		if err != nil {
+			break
+		}
+		if _, err := d.sql.Exec(`DELETE FROM batch_target WHERE job_id = ?`, id); err != nil {
+			break
+		}
+		if _, err := d.sql.Exec(`DELETE FROM batch_job WHERE id = ?`, id); err != nil {
+			break
+		}
+		evicted++
+	}
+	return evicted
+}
+
+func (d *DB) MarkInFlightFailed(reason string, now int64) int {
+	if d == nil {
+		return 0
+	}
+	res, err := d.sql.Exec(`
+		UPDATE batch_job SET status = 'failed', error = ?, lease = ? WHERE status IN ('queued','running')
+	`, reason, now)
+	if err != nil {
+		return 0
+	}
+	n, _ := res.RowsAffected()
+	return int(n)
+}
+
+func (d *DB) BatchNbytes() int64 {
+	if d == nil {
+		return 0
+	}
+	var n sql.NullInt64
+	_ = d.sql.QueryRow(`SELECT IFNULL(SUM(nbytes),0) FROM batch_job`).Scan(&n)
+	return n.Int64
+}

internal/dcache/batch_test.go

@@ -0,0 +1,254 @@
+package dcache
+
+import (
+	"io"
+	"log/slog"
+	"strings"
+	"testing"
+	"time"
+)
+
+func newBatchDB(t *testing.T) *DB {
+	t.Helper()
+	db, err := Open("", 1<<20, slog.New(slog.NewTextHandler(io.Discard, nil)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	return db
+}
+
+func strp(s string) *string { return &s }
+
+func TestBatchJobRoundTrip(t *testing.T) {
+	db := newBatchDB(t)
+	j := BatchJob{
+		ID: "j1", Caller: "alice", CallerClass: ClassFleet, IDemKey: "k1",
+		Status: "queued", TargetsTotal: 3, CreatedAt: 100, Lease: 100,
+	}
+	if err := db.CreateBatchJob(j); err != nil {
+		t.Fatal(err)
+	}
+	got, ok := db.BatchJob("j1")
+	if !ok {
+		t.Fatal("missing job")
+	}
+	if got.ID != j.ID || got.Caller != "alice" || got.CallerClass != ClassFleet ||
+		got.Status != "queued" || got.TargetsTotal != 3 || got.Lease != 100 || got.CreatedAt != 100 {
+		t.Fatalf("round trip mismatch: %+v", got)
+	}
+	if _, ok := db.BatchJob("nope"); ok {
+		t.Fatal("unknown id must be ok=false")
+	}
+}
+
+func TestBatchIDemLookup(t *testing.T) {
+	db := newBatchDB(t)
+	if err := db.CreateBatchJob(BatchJob{ID: "j1", Caller: "a", CallerClass: ClassFleet, IDemKey: "k", Status: "running", TargetsTotal: 1, CreatedAt: 100, Lease: 1000}); err != nil {
+		t.Fatal(err)
+	}
+	if id, ok := db.BatchJobByIDem("a", "k", 2000, time.Hour); !ok || id != "j1" {
+		t.Fatalf("live idem: %q %v", id, ok)
+	}
+	if _, ok := db.BatchJobByIDem("a", "k", 5000, time.Hour); ok {
+		t.Fatal("lease-expired job must not match")
+	}
+	if _, ok := db.BatchJobByIDem("b", "k", 2000, time.Hour); ok {
+		t.Fatal("other caller must not match")
+	}
+	// Only a job that is live or has results is reused. A failed (e.g. killed by a restart)
+	// or cancelled job must not be handed back to a resubmit for the rest of its lease.
+	for _, c := range []struct {
+		status string
+		reuse  bool
+	}{{"queued", true}, {"done", true}, {"partial", true}, {"failed", false}, {"cancelled", false}} {
+		key := "k-" + c.status
+		if err := db.CreateBatchJob(BatchJob{ID: "j-" + c.status, Caller: "a", CallerClass: ClassFleet, IDemKey: key, Status: c.status, TargetsTotal: 1, CreatedAt: 100, Lease: 1000}); err != nil {
+			t.Fatal(err)
+		}
+		if _, ok := db.BatchJobByIDem("a", key, 2000, time.Hour); ok != c.reuse {
+			t.Errorf("status %s: reused=%v, want %v", c.status, ok, c.reuse)
+		}
+	}
+}
+
+func TestBatchLiveCounts(t *testing.T) {
+	db := newBatchDB(t)
+	mk := func(id, caller, status string) {
+		if err := db.CreateBatchJob(BatchJob{ID: id, Caller: caller, CallerClass: ClassFleet, IDemKey: id, Status: status, TargetsTotal: 1, CreatedAt: 1, Lease: 1}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	mk("j1", "a", "queued")
+	mk("j2", "a", "running")
+	mk("j3", "a", "done")
+	mk("j4", "b", "running")
+	if n := db.LiveBatchJobs("a"); n != 2 {
+		t.Fatalf("caller a live=%d want 2", n)
+	}
+	if n := db.LiveBatchJobsTotal(); n != 3 {
+		t.Fatalf("total live=%d want 3", n)
+	}
+}
+
+func TestBatchTargetsOrderedAndUpdate(t *testing.T) {
+	db := newBatchDB(t)
+	must := func(err error) {
+		t.Helper()
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	must(db.CreateBatchJob(BatchJob{ID: "j1", Caller: "a", CallerClass: ClassFleet, IDemKey: "k", Status: "running", TargetsTotal: 3, CreatedAt: 100, Lease: 100}))
+	must(db.PutBatchTargets("j1", []BatchTarget{
+		{JobID: "j1", Idx: 0, URL: "https://a.example/", Options: `{"mode":"clean"}`, Status: "pending"},
+		{JobID: "j1", Idx: 1, URL: "https://b.example/", Status: "pending"},
+		{JobID: "j1", Idx: 2, URL: "https://c.example/", Status: "pending"},
+	}))
+	got, ok := db.BatchTargets("j1")
+	if !ok || len(got) != 3 {
+		t.Fatalf("targets len=%d ok=%v", len(got), ok)
+	}
+	for i := range got {
+		if got[i].Idx != i {
+			t.Fatalf("order broken: got[%d].Idx=%d", i, got[i].Idx)
+		}
+	}
+
+	// A done write bumps the lease to now and counts the target once.
+	res := strp(`{"content":"ok"}`)
+	must(db.UpdateBatchTarget("j1", 0, 200, "done", res, nil, nil))
+	j, _ := db.BatchJob("j1")
+	if j.Lease != 200 || j.TargetsDone != 1 {
+		t.Fatalf("after first write lease=%d done=%d", j.Lease, j.TargetsDone)
+	}
+	// Re-writing the same target (still a done transition) must not double-count.
+	must(db.UpdateBatchTarget("j1", 0, 201, "done", res, nil, nil))
+	j, _ = db.BatchJob("j1")
+	if j.TargetsDone != 1 {
+		t.Fatalf("double-counted: done=%d", j.TargetsDone)
+	}
+	// A failed write counts too.
+	must(db.UpdateBatchTarget("j1", 1, 202, "failed", nil, strp("upstream 500"), strp("browser")))
+	j, _ = db.BatchJob("j1")
+	if j.TargetsDone != 2 || j.Lease != 202 {
+		t.Fatalf("after failed write lease=%d done=%d", j.Lease, j.TargetsDone)
+	}
+}
+
+func TestBatchTouchAndStatus(t *testing.T) {
+	db := newBatchDB(t)
+	if err := db.CreateBatchJob(BatchJob{ID: "j1", Caller: "a", CallerClass: ClassFleet, IDemKey: "k", Status: "running", TargetsTotal: 1, CreatedAt: 100, Lease: 100}); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.TouchBatchJob("j1", 500); err != nil {
+		t.Fatal(err)
+	}
+	if j, _ := db.BatchJob("j1"); j.Lease != 500 {
+		t.Fatalf("touch lease=%d want 500", j.Lease)
+	}
+	if err := db.SetBatchJobStatus("j1", "cancelled", "", 600); err != nil {
+		t.Fatal(err)
+	}
+	if j, _ := db.BatchJob("j1"); j.Status != "cancelled" || j.Lease != 600 {
+		t.Fatalf("status=%q lease=%d", j.Status, j.Lease)
+	}
+}
+
+func TestSweepBatch(t *testing.T) {
+	db := newBatchDB(t)
+	mk := func(id string, lease int64) {
+		if err := db.CreateBatchJob(BatchJob{ID: id, Caller: "a", CallerClass: ClassFleet, IDemKey: id, Status: "done", TargetsTotal: 1, CreatedAt: 1, Lease: lease}); err != nil {
+			t.Fatal(err)
+		}
+		if err := db.PutBatchTargets(id, []BatchTarget{{JobID: id, Idx: 0, URL: "https://x.example/", Status: "done", Result: strp("x")}}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	mk("old", 100)
+	mk("new", 1000)
+	n := db.SweepBatch(300, 100*time.Second) // old: 100+100=200<=300 expired; new: 1100>300 kept
+	if n != 1 {
+		t.Fatalf("swept %d want 1", n)
+	}
+	if _, ok := db.BatchJob("old"); ok {
+		t.Fatal("old job must be gone")
+	}
+	if _, ok := db.BatchJob("new"); !ok {
+		t.Fatal("new job must remain")
+	}
+	if _, ok := db.BatchTargets("old"); ok {
+		t.Fatal("old targets must be gone")
+	}
+}
+
+func TestEvictBatchOldest(t *testing.T) {
+	db := newBatchDB(t)
+	done := func(id string, lease int64, size int) {
+		if err := db.CreateBatchJob(BatchJob{ID: id, Caller: "a", CallerClass: ClassFleet, IDemKey: id, Status: "done", TargetsTotal: 1, CreatedAt: 1, Lease: lease}); err != nil {
+			t.Fatal(err)
+		}
+		if err := db.PutBatchTargets(id, []BatchTarget{{JobID: id, Idx: 0, URL: "https://x.example/", Status: "pending"}}); err != nil {
+			t.Fatal(err)
+		}
+		if err := db.UpdateBatchTarget(id, 0, lease, "done", strp(strings.Repeat("x", size)), nil, nil); err != nil {
+			t.Fatal(err)
+		}
+	}
+	done("old1", 100, 100)
+	done("old2", 200, 200)
+	done("new", 300, 300)
+
+	// A running job with one completed target must never be byte-evicted.
+	if err := db.CreateBatchJob(BatchJob{ID: "run", Caller: "a", CallerClass: ClassFleet, IDemKey: "run", Status: "running", TargetsTotal: 2, TargetsDone: 1, CreatedAt: 1, Lease: 400}); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.PutBatchTargets("run", []BatchTarget{
+		{JobID: "run", Idx: 0, URL: "https://x.example/", Status: "pending"},
+		{JobID: "run", Idx: 1, URL: "https://y.example/", Status: "pending"},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.UpdateBatchTarget("run", 0, 400, "done", strp(strings.Repeat("x", 500)), nil, nil); err != nil {
+		t.Fatal(err)
+	}
+
+	// total nbytes = 100+200+300+500 = 1100. Evict oldest done until <= 250:
+	// old1(100), old2(200), new(300) all evicted (oldest first); run stays (live).
+	db.EvictBatchOldest(250)
+	for _, id := range []string{"old1", "old2", "new"} {
+		if _, ok := db.BatchJob(id); ok {
+			t.Fatalf("%s should be byte-evicted", id)
+		}
+	}
+	if _, ok := db.BatchJob("run"); !ok {
+		t.Fatal("running job must not be byte-evicted")
+	}
+	if n := db.BatchNbytes(); n != 500 {
+		t.Fatalf("BatchNbytes=%d want 500", n)
+	}
+}
+
+func TestMarkInFlightFailed(t *testing.T) {
+	db := newBatchDB(t)
+	mk := func(id, status string) {
+		if err := db.CreateBatchJob(BatchJob{ID: id, Caller: "a", CallerClass: ClassFleet, IDemKey: id, Status: status, TargetsTotal: 1, CreatedAt: 1, Lease: 1}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	mk("q", "queued")
+	mk("r", "running")
+	mk("d", "done")
+	if n := db.MarkInFlightFailed("restart", 999); n != 2 {
+		t.Fatalf("marked %d want 2", n)
+	}
+	if j, _ := db.BatchJob("q"); j.Status != "failed" || j.Error != "restart" || j.Lease != 999 {
+		t.Fatalf("q %+v", j)
+	}
+	if j, _ := db.BatchJob("r"); j.Status != "failed" || j.Error != "restart" {
+		t.Fatalf("r %+v", j)
+	}
+	if j, _ := db.BatchJob("d"); j.Status != "done" {
+		t.Fatalf("done job changed: %+v", j)
+	}
+}

internal/dcache/dcache.go

@@ -47,6 +47,8 @@ type DB struct {
 	SearchM   uint64
 	PDFH      uint64
 	PDFM      uint64
+	BatchH    uint64
+	BatchM    uint64
 	Evictions uint64
 }
 
@@ -110,6 +112,32 @@ func Open(path string, maxBytes int64, log *slog.Logger) (*DB, error) {
 			text TEXT,
 			PRIMARY KEY (k, page)
 		);
+		CREATE TABLE IF NOT EXISTS batch_job (
+			id            TEXT PRIMARY KEY,
+			caller        TEXT NOT NULL,
+			caller_class  TEXT NOT NULL,
+			idem_key      TEXT NOT NULL,
+			status        TEXT NOT NULL,
+			error         TEXT NOT NULL DEFAULT '',
+			targets_total INTEGER NOT NULL,
+			targets_done  INTEGER NOT NULL DEFAULT 0,
+			created_at    INTEGER NOT NULL,
+			lease         INTEGER NOT NULL,
+			nbytes        INTEGER NOT NULL DEFAULT 0
+		);
+		CREATE TABLE IF NOT EXISTS batch_target (
+			job_id  TEXT NOT NULL,
+			idx     INTEGER NOT NULL,
+			url     TEXT NOT NULL,
+			options TEXT NOT NULL,
+			status  TEXT NOT NULL,
+			result  TEXT,
+			error   TEXT,
+			hint    TEXT,
+			PRIMARY KEY (job_id, idx)
+		);
+		CREATE INDEX IF NOT EXISTS batch_job_idem ON batch_job(caller, idem_key);
+		CREATE INDEX IF NOT EXISTS batch_job_lease ON batch_job(lease);
 	`); err != nil {
 		_ = sq.Close()
 		return nil, err