From 816a76fe5b679c29afcfb9e5c61d1cdb1c3dfc43 Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Sat, 26 Sep 2026 22:18:35 -0700 Subject: [PATCH] =?UTF-8?q?batch:=20workers=20=E2=80=94=20round-robin,=20s?= =?UTF-8?q?erial=20per=20job,=20busy=20retry,=20cancel=20in=20flight,=20ho?= =?UTF-8?q?oks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- docs/implementer-log.md | 3 +- internal/batch/batch.go | 74 ++++++- internal/batch/runner_test.go | 373 ++++++++++++++++++++++++++++++++++ internal/batch/worker.go | 271 ++++++++++++++++++++++++ 4 files changed, 715 insertions(+), 6 deletions(-) create mode 100644 internal/batch/runner_test.go create mode 100644 internal/batch/worker.go diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 12b81fd..105bd40 100644 --- a/docs/implementer-log.md +++ b/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 ` 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) diff --git a/internal/batch/batch.go b/internal/batch/batch.go index 51ed5d5..bddc3d2 100644 --- a/internal/batch/batch.go +++ b/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 -// false for an unknown id or a job already terminal. +// Cancel cancels a job. A job still in the rotation (queued or with a target in flight) is marked +// 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 -// once. With no workers (task 02) there is nothing to stop; the in-memory callers are dropped. +// Close stops the workers, cancels their context, and waits for an in-flight fetch to return. +// 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() }) } diff --git a/internal/batch/runner_test.go b/internal/batch/runner_test.go new file mode 100644 index 0000000..7d71b21 --- /dev/null +++ b/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) + } +} diff --git a/internal/batch/worker.go b/internal/batch/worker.go new file mode 100644 index 0000000..5cb64df --- /dev/null +++ b/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) + } +}