From a452aa16a1e532cc8eb28529259bfa27ab7fd6b3 Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Sat, 26 Sep 2026 23:27:47 -0700 Subject: [PATCH] P6 review: runner shutdown, cancel-while-busy, concurrent submits, batch-off 503, transactional store, CLI 202 Owner review of p6 (Ornith tasks 01-09), with an independent reviewer pass whose findings I confirmed against the code. Every fix has a regression test that fails on the old code. internal/batch - Close let the workers keep draining every live job under the cancelled context, failing every target; they now stop at once, store nothing on a failure during shutdown, and leave the job live for MarkInFlightFailed("restart"). - A job cancelled while its in-flight target came back busy stayed in the rotation forever (depth gauge stuck, job held in memory); it now leaves. The task 03 design had this gap: mine. - Submit's idempotency lookup, caps and create ran unlocked: 20 concurrent submits by one caller made 8 jobs. Serialised with submitMu. - A failed target insert left a queued job no worker would run (holding the caller's cap); it is marked failed. - A second concurrent cancel returned true and counted twice. - Config: zero OR negative values take the defaults (negative Workers panicked the WaitGroup; a negative TTL or byte budget would sweep or evict everything). - Retry timing used the pre-fetch time (a 5 s busy answer made RetryWait moot). - The rotation cursor is kept within [0, len(live)] (it was absolute and drifted on removal). - The unused in-memory callers map is gone (Submit racing Close wrote to a nil map). internal/dcache - PutBatchTargets, SweepBatch and eviction are transactional; the sweep never removes a queued or running job. internal/api - Every /v1/batch endpoint answers 503 when batch is off; POST panicked on the nil runner. - GET /{id}/{idx} bounds idx by the rows actually returned. - batchFetch copies a *fetch.Error instead of mutating one the PDF pool may share. internal/config - ParseSize no longer overflows to a negative size. cmd/sift - sift batch treated 202 Accepted (submit) as an error and exited 1 with no message. Co-Authored-By: Claude Opus 5.5 --- cmd/sift/batch_review_test.go | 52 ++++++++ cmd/sift/batchcmd.go | 2 +- internal/api/batch.go | 18 ++- internal/api/batch_review_test.go | 29 +++++ internal/batch/batch.go | 62 +++++----- internal/batch/review_test.go | 177 +++++++++++++++++++++++++++ internal/batch/worker.go | 41 +++++-- internal/config/batch_review_test.go | 10 ++ internal/config/config.go | 3 +- internal/dcache/batch.go | 50 ++++++-- internal/dcache/batch_review_test.go | 25 ++++ 11 files changed, 406 insertions(+), 63 deletions(-) create mode 100644 cmd/sift/batch_review_test.go create mode 100644 internal/api/batch_review_test.go create mode 100644 internal/batch/review_test.go create mode 100644 internal/config/batch_review_test.go create mode 100644 internal/dcache/batch_review_test.go diff --git a/cmd/sift/batch_review_test.go b/cmd/sift/batch_review_test.go new file mode 100644 index 0000000..aeb8a9a --- /dev/null +++ b/cmd/sift/batch_review_test.go @@ -0,0 +1,52 @@ +package main + +// P6 owner review: the batch CLI treats every 2xx as success. POST /v1/batch answers 202, and +// batchDo exited 1 on it with an empty message. + +import ( + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestBatchDoAccepts202(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer k" { + w.WriteHeader(401) + _, _ = io.WriteString(w, `{"error":"unauthorized"}`) + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + _, _ = io.WriteString(w, `{"success":true,"job_id":"j1","status":"queued","targets":1,"reused":false}`) + })) + defer srv.Close() + dir := t.TempDir() + keyPath := filepath.Join(dir, "key") + if err := os.WriteFile(keyPath, []byte("k\n"), 0o600); err != nil { + t.Fatal(err) + } + cfg := filepath.Join(dir, "client.toml") + if err := os.WriteFile(cfg, []byte("url = \""+srv.URL+"\"\nkey_file = \""+keyPath+"\"\n"), 0o600); err != nil { + t.Fatal(err) + } + t.Setenv("SIFT_CLIENT", cfg) + + r, w, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + old := os.Stdout + os.Stdout = w + batchDo(http.MethodPost, "/v1/batch", strings.NewReader(`{"targets":[{"url":"https://example.com/"}]}`)) + os.Stdout = old + _ = w.Close() + out, _ := io.ReadAll(r) + if !strings.Contains(string(out), `"job_id": "j1"`) { + t.Fatalf("stdout %q, want the 202 body printed", out) + } +} diff --git a/cmd/sift/batchcmd.go b/cmd/sift/batchcmd.go index ef2986b..2c56d2e 100644 --- a/cmd/sift/batchcmd.go +++ b/cmd/sift/batchcmd.go @@ -189,7 +189,7 @@ func batchDo(method, path string, body io.Reader) { fmt.Fprintln(os.Stderr, err) os.Exit(5) } - if resp.StatusCode != 200 { + if resp.StatusCode < 200 || resp.StatusCode > 299 { // submit answers 202 Accepted msg, _ := env["error"].(string) fmt.Fprintln(os.Stderr, msg) os.Exit(exitFor(resp.StatusCode, msg)) diff --git a/internal/api/batch.go b/internal/api/batch.go index 1945972..50f7aac 100644 --- a/internal/api/batch.go +++ b/internal/api/batch.go @@ -66,14 +66,16 @@ func (s *Server) batchFetch(ctx context.Context, c *auth.Caller, t batch.Target) case 429, 503: return batch.Result{}, fmt.Errorf("%s: %w", fe.Msg, batch.ErrRetryLater) } - if fe.Status == 502 { - fe.Status = 500 - fe.Msg = "internal error" + // Copy, never mutate: the PDF pool's single-flight may share this error with an + // interactive request. + status, msg := fe.Status, fe.Msg + if status == 502 { + status, msg = 500, "internal error" } - if fe.Status == 403 { + if status == 403 { s.blockN.Add(1) } - return batch.Result{}, &batch.TargetError{Status: fe.Status, Msg: fe.Msg, Hint: fe.Hint} + return batch.Result{}, &batch.TargetError{Status: status, Msg: msg, Hint: fe.Hint} } s.Log.Error("internal", "caller", c.Name, "host", "-", "err", s.logErr(c, err)) return batch.Result{}, &batch.TargetError{Status: 500, Msg: "internal error"} @@ -158,6 +160,10 @@ func (s *Server) batchSubmit(w http.ResponseWriter, r *http.Request) { if c == nil { return } + if s.Batch == nil { // [batch] disabled, or the store did not open + writeErr(w, 503, "batch disabled", "") + return + } var req struct { Targets []v1Req `json:"targets"` } @@ -264,7 +270,7 @@ func (s *Server) batchResult(w http.ResponseWriter, r *http.Request) { return } idx, err := strconv.Atoi(r.PathValue("idx")) - if err != nil || idx < 0 || idx >= j.TargetsTotal { + if err != nil || idx < 0 || idx >= j.TargetsTotal || idx >= len(ts) { writeErr(w, 404, "not found", "") return } diff --git a/internal/api/batch_review_test.go b/internal/api/batch_review_test.go new file mode 100644 index 0000000..0e526be --- /dev/null +++ b/internal/api/batch_review_test.go @@ -0,0 +1,29 @@ +package api + +// P6 owner review: with batch off (disabled, or the store did not open) every batch endpoint +// answers 503; none of them panics on the nil runner. + +import ( + "net/http" + "testing" +) + +func TestBatchDisabled(t *testing.T) { + s, keys := newBatchServer(t, "test") + s.Batch.Close() + s.Batch = nil + for _, c := range []struct{ method, path string }{ + {http.MethodPost, "/v1/batch"}, + {http.MethodGet, "/v1/batch/x"}, + {http.MethodGet, "/v1/batch/x/0"}, + {http.MethodDelete, "/v1/batch/x"}, + } { + var body any + if c.method == http.MethodPost { + body = map[string]any{"targets": []any{map[string]any{"url": "http://127.0.0.1/"}}} + } + if rec := batchReq(t, s, c.method, c.path, keys["test"], body); rec.Code != 503 { + t.Errorf("%s %s: %d %s, want 503", c.method, c.path, rec.Code, rec.Body.String()) + } + } +} diff --git a/internal/batch/batch.go b/internal/batch/batch.go index bddc3d2..49677b0 100644 --- a/internal/batch/batch.go +++ b/internal/batch/batch.go @@ -118,14 +118,13 @@ type Runner struct { fetch FetchFunc log *slog.Logger cfg Config - callers map[string]*auth.Caller // job id → submitting caller, held in memory only - callMu sync.Mutex closeOnce sync.Once + submitMu sync.Mutex // serialises Submit: idempotency lookup, caps and create are one step 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 + cursor int // where the next scan starts, within [0, len(live)]; read as cursor % len(live) closed bool ctx context.Context // the FetchFunc context; Close cancels it stop context.CancelFunc @@ -134,42 +133,43 @@ type Runner struct { } func New(db *dcache.DB, fetch FetchFunc, cfg Config, log *slog.Logger) *Runner { - if cfg.TTL == 0 { + // Zero or negative means the default: a negative TTL or byte budget would make the sweep or + // the eviction delete everything, and negative workers would panic the WaitGroup. + if cfg.TTL <= 0 { cfg.TTL = 24 * time.Hour } - if cfg.TargetsPerJob == 0 { + if cfg.TargetsPerJob <= 0 { cfg.TargetsPerJob = 32 } - if cfg.JobsPerCaller == 0 { + if cfg.JobsPerCaller <= 0 { cfg.JobsPerCaller = 1 } - if cfg.JobsTotal == 0 { + if cfg.JobsTotal <= 0 { cfg.JobsTotal = 8 } - if cfg.Workers == 0 { + if cfg.Workers <= 0 { cfg.Workers = 1 } if cfg.Workers > 2 { cfg.Workers = 2 } - if cfg.MaxBytes == 0 { + if cfg.MaxBytes <= 0 { cfg.MaxBytes = 64 << 20 } - if cfg.RetryWait == 0 { + if cfg.RetryWait <= 0 { cfg.RetryWait = 2 * time.Second } - if cfg.RetryFor == 0 { + if cfg.RetryFor <= 0 { cfg.RetryFor = 10 * time.Minute } 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, + db: db, + fetch: fetch, + log: log, + cfg: cfg, + ctx: ctx, + stop: stop, } r.cond = sync.NewCond(&r.mu) r.wg.Add(cfg.Workers) @@ -193,6 +193,11 @@ func (r *Runner) Submit(ctx context.Context, c *auth.Caller, callerClass string, sum := sha256.Sum256([]byte(c.Name + "\x00" + string(raw))) idemKey := hex.EncodeToString(sum[:]) + // The lookup, the caps and the create must be one step, or concurrent submits all pass the + // checks before any of them is stored. + r.submitMu.Lock() + defer r.submitMu.Unlock() + now := time.Now().Unix() if existing, ok := r.db.BatchJobByIDem(c.Name, idemKey, now, r.cfg.TTL); ok { r.callJob("reused") @@ -241,13 +246,11 @@ func (r *Runner) Submit(ctx context.Context, c *auth.Caller, callerClass string, } if err := r.db.PutBatchTargets(id, btargets); err != nil { r.log.Warn("batch: store targets", "job", id, "err", err) + // Never leave a queued job that no worker will run: it would hold the caller's cap. + _ = r.db.SetBatchJobStatus(id, "failed", "store", now) return "", false, fmt.Errorf("batch: store targets: %w", err) } - r.callMu.Lock() - r.callers[id] = c - r.callMu.Unlock() - r.mu.Lock() r.live = append(r.live, &liveJob{ id: id, @@ -278,6 +281,10 @@ func (r *Runner) Cancel(id string) bool { } } if j != nil { + if j.cancelled { // a concurrent DELETE got here first + r.mu.Unlock() + return false + } 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) @@ -285,7 +292,6 @@ func (r *Runner) Cancel(id string) bool { r.callJob("cancelled") if !j.running { r.removeLocked(j) - r.forgetCaller(id) r.callDepth(len(r.live)) } r.cond.Broadcast() @@ -302,9 +308,6 @@ func (r *Runner) Cancel(id string) bool { r.log.Warn("batch: cancel job", "job", id, "err", err) return false } - r.callMu.Lock() - delete(r.callers, id) - r.callMu.Unlock() r.callJob("cancelled") return true } @@ -356,13 +359,10 @@ func (r *Runner) Touch(id string) { } // 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. +// Safe to call more than once. Jobs still live stay queued or running in the store, for +// MarkInFlightFailed("restart") on the next start. 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() diff --git a/internal/batch/review_test.go b/internal/batch/review_test.go new file mode 100644 index 0000000..2712e3a --- /dev/null +++ b/internal/batch/review_test.go @@ -0,0 +1,177 @@ +package batch + +// P6 owner review: regression tests for the runner (findings from the independent review). + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "git.wntrmute.dev/kyle/sift/internal/auth" + "git.wntrmute.dev/kyle/sift/internal/dcache" +) + +// A job cancelled while its in-flight target comes back busy leaves the rotation. +func TestCancelWhileBusyLeavesRotation(t *testing.T) { + started := make(chan struct{}, 1) + release := make(chan struct{}) + var depth atomic.Int64 + fetch := func(context.Context, *auth.Caller, Target) (Result, error) { + select { + case started <- struct{}{}: + default: + } + <-release + return Result{}, fmt.Errorf("busy: %w", ErrRetryLater) + } + r := newRunner(t, fetch, Config{RetryWait: time.Millisecond, Hooks: Hooks{Depth: func(n int) { depth.Store(int64(n)) }}}) + id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b")) + if err != nil { + t.Fatal(err) + } + <-started + if !r.Cancel(id) { + t.Fatal("cancel returned false") + } + close(release) + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + r.mu.Lock() + n := len(r.live) + r.mu.Unlock() + if n == 0 && depth.Load() == 0 { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("cancelled job still in the rotation (depth %d)", depth.Load()) +} + +// Close stops the workers: no further target is fetched, and the job is left live in the +// store for MarkInFlightFailed("restart") to reap, not failed target by target. +func TestCloseStopsDraining(t *testing.T) { + started := make(chan struct{}, 1) + var calls atomic.Int64 + fetch := func(ctx context.Context, _ *auth.Caller, _ Target) (Result, error) { + calls.Add(1) + select { + case started <- struct{}{}: + default: + } + <-ctx.Done() + return Result{}, ctx.Err() + } + r := newRunner(t, fetch, Config{}) + id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b", "c", "d")) + if err != nil { + t.Fatal(err) + } + <-started + r.Close() + if n := calls.Load(); n != 1 { + t.Fatalf("%d fetches after Close, want only the one in flight", n) + } + j, ts, _ := r.Status(id) + if j.Status != "running" && j.Status != "queued" { + t.Fatalf("job %q after Close, want it left live for the restart reaper", j.Status) + } + for _, x := range ts { + if x.Status != "pending" { + t.Fatalf("target %d is %q after Close, want pending", x.Idx, x.Status) + } + } +} + +// Concurrent submits respect the caps and idempotency: one job per caller, and identical +// lists all get the same job. +func TestConcurrentSubmits(t *testing.T) { + release := make(chan struct{}) + defer close(release) + fetch := func(context.Context, *auth.Caller, Target) (Result, error) { + <-release + return Result{Content: "ok"}, nil + } + r := newRunner(t, fetch, Config{}) + var wg sync.WaitGroup + var mu sync.Mutex + ids := map[string]bool{} + for i := 0; i < 20; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls(fmt.Sprintf("u%d", i))) + if err == nil { + mu.Lock() + ids[id] = true + mu.Unlock() + } + }(i) + } + wg.Wait() + if len(ids) != 1 { + t.Fatalf("20 concurrent submits by one caller (cap 1) created %d jobs", len(ids)) + } + same := map[string]bool{} + for i := 0; i < 20; i++ { + wg.Add(1) + go func() { + defer wg.Done() + id, _, err := r.Submit(context.Background(), bob, dcache.ClassFleet, urls("same")) + if err == nil { + mu.Lock() + same[id] = true + mu.Unlock() + } + }() + } + wg.Wait() + if len(same) != 1 { + t.Fatalf("20 concurrent identical submits created %d jobs, want 1", len(same)) + } +} + +// A second cancel of a job with a target in flight returns false and counts once. +func TestDoubleCancel(t *testing.T) { + started := make(chan struct{}, 1) + release := make(chan struct{}) + var cancels atomic.Int64 + fetch := func(context.Context, *auth.Caller, Target) (Result, error) { + select { + case started <- struct{}{}: + default: + } + <-release + return Result{Content: "ok"}, nil + } + r := newRunner(t, fetch, Config{Hooks: Hooks{Job: func(o string) { + if o == "cancelled" { + cancels.Add(1) + } + }}}) + defer close(release) + id, _, _ := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b")) + <-started + if !r.Cancel(id) { + t.Fatal("first cancel returned false") + } + if r.Cancel(id) { + t.Fatal("second cancel returned true") + } + if n := cancels.Load(); n != 1 { + t.Fatalf("cancelled counted %d times", n) + } +} + +// Negative or zero durations and sizes fall back to the defaults. +func TestConfigDefaults(t *testing.T) { + r := newRunner(t, func(context.Context, *auth.Caller, Target) (Result, error) { return Result{}, nil }, + Config{TTL: -time.Hour, MaxBytes: -1, RetryWait: -1, RetryFor: -1, TargetsPerJob: -1, JobsPerCaller: -1, JobsTotal: -1, Workers: -1}) + c := r.cfg + if c.TTL != 24*time.Hour || c.MaxBytes != 64<<20 || c.RetryWait != 2*time.Second || c.RetryFor != 10*time.Minute || + c.TargetsPerJob != 32 || c.JobsPerCaller != 1 || c.JobsTotal != 8 || c.Workers != 1 { + t.Fatalf("config %+v", c) + } +} diff --git a/internal/batch/worker.go b/internal/batch/worker.go index 5cb64df..caf223a 100644 --- a/internal/batch/worker.go +++ b/internal/batch/worker.go @@ -36,13 +36,13 @@ func (r *Runner) worker() { for { r.mu.Lock() + if r.closed { // stop taking targets at once; live jobs are reaped on the next start + r.mu.Unlock() + return + } 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() @@ -68,10 +68,28 @@ func (r *Runner) worker() { res, err := r.runTarget(r.ctx, j, idx, target) r.mu.Lock() + now = time.Now() // after the fetch: a busy answer can take seconds to come back j.running = false r.inflight-- r.callInflight(r.inflight) + // Shutting down: a failure now is most likely the cancelled context. Store nothing; the + // job stays live in the store for the restart reaper. + if r.closed && err != nil { + r.cond.Broadcast() + r.mu.Unlock() + continue + } + + // A cancelled job whose in-flight target was only busy is finished: nothing more runs. + if errors.Is(err, ErrRetryLater) && j.cancelled { + r.removeLocked(j) + r.callDepth(len(r.live)) + r.cond.Broadcast() + r.mu.Unlock() + continue + } + // 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) { @@ -102,7 +120,6 @@ func (r *Runner) worker() { r.setTerminal(j) } r.removeLocked(j) - r.forgetCaller(j.id) r.callDepth(len(r.live)) } r.cond.Broadcast() @@ -124,7 +141,9 @@ func (r *Runner) pick(now time.Time) *liveJob { 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 + // The job after this one; n (one past the end) means "whatever is appended next, + // else wrap to 0" — reads take cursor % n. + r.cursor = (r.cursor+i)%n + 1 return j } } @@ -168,6 +187,9 @@ func (r *Runner) removeLocked(j *liveJob) { if r.cursor > idx { r.cursor-- } + if r.cursor > len(r.live) { + r.cursor = len(r.live) + } } // setTerminal writes the job's finished status once: all done → done, all failed → failed, a mix @@ -239,13 +261,6 @@ func (r *Runner) storeTarget(j *liveJob, idx int, res Result, err error) int { 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) diff --git a/internal/config/batch_review_test.go b/internal/config/batch_review_test.go new file mode 100644 index 0000000..56ff0bb --- /dev/null +++ b/internal/config/batch_review_test.go @@ -0,0 +1,10 @@ +package config + +import "testing" + +// P6 owner review: a size too large for int64 falls back to the default instead of wrapping negative. +func TestParseSizeOverflow(t *testing.T) { + if got := ParseSize("9223372036854775807GiB", 7); got != 7 { + t.Fatalf("ParseSize(huge) = %d, want the default", got) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 796007c..4dc5299 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1,6 +1,7 @@ package config import ( + "math" "os" "strconv" "strings" @@ -145,7 +146,7 @@ func ParseSize(s string, def int64) int64 { mult = 1 } n, err := strconv.ParseInt(s, 10, 64) - if err != nil || n <= 0 { + if err != nil || n <= 0 || n > math.MaxInt64/mult { // the last: n*mult would overflow return def } return n * mult diff --git a/internal/dcache/batch.go b/internal/dcache/batch.go index 928113b..8a44759 100644 --- a/internal/dcache/batch.go +++ b/internal/dcache/batch.go @@ -134,8 +134,14 @@ func (d *DB) PutBatchTargets(jobID string, targets []BatchTarget) error { if d == nil { return fmt.Errorf("dcache: nil db") } + // One transaction: a job never has only some of its targets. + tx, err := d.sql.Begin() + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() for _, t := range targets { - if _, err := d.sql.Exec(` + if _, err := tx.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, @@ -143,7 +149,7 @@ func (d *DB) PutBatchTargets(jobID string, targets []BatchTarget) error { return err } } - return nil + return tx.Commit() } func (d *DB) BatchTargets(jobID string) ([]BatchTarget, bool) { @@ -249,16 +255,25 @@ func (d *DB) SweepBatch(now int64, ttl time.Duration) int { 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 { + // A queued or running job is never swept: it is still in a worker's rotation. The restart + // reaper marks such jobs failed, and they are swept after that. Both deletes are one + // transaction, so a job never loses its targets but keeps its row. + tx, err := d.sql.Begin() + if err != nil { return 0 } - res, err := d.sql.Exec(`DELETE FROM batch_job WHERE lease + ? <= ?`, ttlSec, now) + defer func() { _ = tx.Rollback() }() + const expired = `SELECT id FROM batch_job WHERE lease + ? <= ? AND status NOT IN ('queued','running')` + if _, err := tx.Exec(`DELETE FROM batch_target WHERE job_id IN (`+expired+`)`, ttlSec, now); err != nil { + return 0 + } + res, err := tx.Exec(`DELETE FROM batch_job WHERE id IN (`+expired+`)`, ttlSec, now) if err != nil { return 0 } + if err := tx.Commit(); err != nil { + return 0 + } n, _ := res.RowsAffected() return int(n) } @@ -279,10 +294,7 @@ func (d *DB) EvictBatchOldest(maxBytes int64) int { 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 { + if err := d.deleteBatchJob(id); err != nil { break } evicted++ @@ -312,3 +324,19 @@ func (d *DB) BatchNbytes() int64 { _ = d.sql.QueryRow(`SELECT IFNULL(SUM(nbytes),0) FROM batch_job`).Scan(&n) return n.Int64 } + +// deleteBatchJob removes a job and its targets in one transaction. +func (d *DB) deleteBatchJob(id string) error { + tx, err := d.sql.Begin() + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + if _, err := tx.Exec(`DELETE FROM batch_target WHERE job_id = ?`, id); err != nil { + return err + } + if _, err := tx.Exec(`DELETE FROM batch_job WHERE id = ?`, id); err != nil { + return err + } + return tx.Commit() +} diff --git a/internal/dcache/batch_review_test.go b/internal/dcache/batch_review_test.go new file mode 100644 index 0000000..841becf --- /dev/null +++ b/internal/dcache/batch_review_test.go @@ -0,0 +1,25 @@ +package dcache + +// P6 owner review: the sweep never removes a live job. + +import ( + "testing" + "time" +) + +func TestSweepSparesLiveJobs(t *testing.T) { + db := newBatchDB(t) + for _, st := range []string{"queued", "running", "done"} { + if err := db.CreateBatchJob(BatchJob{ID: st, Caller: "a", CallerClass: ClassFleet, IDemKey: st, Status: st, TargetsTotal: 1, CreatedAt: 1, Lease: 1}); err != nil { + t.Fatal(err) + } + } + if n := db.SweepBatch(1_000_000, time.Hour); n != 1 { + t.Fatalf("swept %d, want only the done job", n) + } + for _, id := range []string{"queued", "running"} { + if _, ok := db.BatchJob(id); !ok { + t.Fatalf("live job %s was swept", id) + } + } +}