diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 2ac42a7..16658ac 100644 --- a/docs/implementer-log.md +++ b/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 ` 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) diff --git a/internal/dcache/batch.go b/internal/dcache/batch.go new file mode 100644 index 0000000..928113b --- /dev/null +++ b/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 +} diff --git a/internal/dcache/batch_test.go b/internal/dcache/batch_test.go new file mode 100644 index 0000000..c9afc19 --- /dev/null +++ b/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) + } +} diff --git a/internal/dcache/dcache.go b/internal/dcache/dcache.go index ebff459..579cb13 100644 --- a/internal/dcache/dcache.go +++ b/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