sift/docs/plans/p6/03-runner-workers.md

P6 task 03: the batch runner's workers

Branch: p6 (git switch p6; git status --short must be empty, otherwise stop) Commit subject: batch: workers — round-robin, serial per job, busy retry, cancel in flight, hooks

Goal

Task 02 wrote the runner's types, Submit, Status, Touch and Cancel; every job stays queued. This task adds the workers that drain jobs: one target at a time per job, round-robin across jobs, with a retry for a target that was only busy.

Files

• Copy: docs/plans/p6/_files/batch/runner_test.go → internal/batch/runner_test.go
• Create: internal/batch/worker.go (the workers, the rotation, one target, the terminal status)
• Modify: internal/batch/batch.go (New starts the workers; Close stops them; Submit wakes one; Cancel handles a job with a target in flight), docs/implementer-log.md

Work in small steps, in files. Write worker.go with the worker loop and a stub for one target, and make it compile. Then fill it in, running the tests as you go. Do not write the implementation out in your reply before putting it in a file.

What already exists (task 02, in internal/batch/batch.go)

All the types in the task 02 file: Options, Target, Result, TargetError, FetchFunc, Job, TargetStatus, Hooks, Config, the errors (including ErrRetryLater), Runner, New, Submit, Status, Touch, Cancel, Close. Read batch.go for how the runner keeps its jobs and each job's caller; the dcache API it uses is listed in the task 02 file's "What already exists".

Design (follow it; do not redesign)

The concurrency is decided here so you can write it straight into worker.go. Put the design into code, not into your reply.

State, added to Runner (keep task 02's fields):

	mu       sync.Mutex    // guards everything below
	cond     *sync.Cond    // on mu
	live     []*liveJob    // the rotation, in submit order
	cursor   int           // index in live where the next scan starts
	closed   bool
	ctx      context.Context // the FetchFunc context; Close cancels it
	stop     context.CancelFunc
	wg       sync.WaitGroup  // the workers
	inflight int

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
}

Submit (task 02's code, plus): after storing, lock mu, append a liveJob, Hooks.Depth(len(live)), cond.Broadcast(), unlock.

A worker (New starts cfg.Workers of them with wg.Add):

loop:
  lock mu
  j := pick(now)        // scan live from cursor, wrapping once: the first job with !running,
                        // !cancelled, next < len(targets), and now >= retryAt; set cursor to
                        // the index after it
  if j == nil:
      if closed: unlock; return
      if some live job has a future retryAt: time.AfterFunc(until the earliest, cond.Broadcast)
      cond.Wait(); unlock-free continue (Wait re-locks; go back to loop start with mu held)
  j.running = true; inflight++; Hooks.Inflight(inflight)
  if !j.started: SetBatchJobStatus(j.id, "running", "", now); j.started = true
  unlock
  res, err := call(ctx, j.caller, j.targets[j.next])   // recovers a panic into "internal error"
  lock mu
  j.running = false; inflight--; Hooks.Inflight(inflight)
  if errors.Is(err, ErrRetryLater):
      if j.busySince is zero: j.busySince = now
      if now − busySince <= RetryFor: j.retryAt = now + RetryWait; Broadcast; unlock; continue
      // else fall through and store it failed with err.Error()
  store target j.next (rule 2); j.next++; j.busySince, j.retryAt = zero; count done/failed
  Hooks.Target(...); EvictBatchOldest(MaxBytes)
  if j.cancelled or j.next == len(targets):
      if !j.cancelled: set the terminal status (rule 3)
      remove j from live (fix cursor), forget its caller, Hooks.Depth(len(live))
  Broadcast; unlock

Store calls happen under mu. They are short (one SQLite connection), and holding mu keeps the store consistent with the in-memory state.

Cancel: lock mu. If the job is in live: set j.cancelled, SetBatchJobStatus(cancelled), and Hooks.Job("cancelled"). If !j.running, remove it from live now; a running one is removed by its worker after the store (above), which must not overwrite cancelled. Broadcast, unlock. A job not in live keeps task 02's behaviour.

Close: lock; closed = true; stop(); Broadcast; unlock; wg.Wait(). Safe to call twice.

A target stays pending in the store while it is in flight. The store has no running target status write, and no test needs one.

Rules

1. Workers. Workers goroutines share one rotation of live jobs in submit order. A worker takes the next job in the rotation that has a pending target and no target in flight. It sets the job to running on its first target, fetches that one target, stores it, and moves on to the next job in the rotation. So:
   • each job is serial: its target i+1 never starts before i is stored;
   • with one worker and two live jobs, their targets alternate (a1 b1 a2 b2 …);
   • a job that a worker has not reached yet stays queued.
2. One target:
   • Call the FetchFunc with the job's caller, under a context that Close cancels. Recover a panic into the error internal error (never echo the panic value).
   • If the error wraps ErrRetryLater, the target stays pending: the worker goes on with the rotation, and the target is tried again after RetryWait. A target that has been coming back busy for longer than RetryFor is stored failed with the error's text.
   • Otherwise store it: success → done with the JSON of the Result; error → failed, with TargetError's Msg/Hint or err.Error() and no hint.
   • Then call Hooks.Target(err, time, len(result JSON)), then EvictBatchOldest(MaxBytes).
3. Terminal status. When a job's last target is stored: all done → done; all failed → failed; a mix → partial. SetBatchJobStatus, then Hooks.Job(status) and Hooks.JobSeconds(now − created), exactly once per job.
4. Cancel(id) returns false for an unknown id or a job already terminal. Otherwise it sets the job cancelled at once. The in-flight target, if any, finishes and is stored; no further target starts; later targets stay pending. Storing that last target must not overwrite cancelled. Hooks.Job("cancelled") once.
5. Hooks.Depth gets the live-job count and Hooks.Inflight the in-flight count whenever either changes.

Facts

• The given test uses FetchFuncs that hold on channels, fail, return ErrRetryLater, or panic. It calls Close before the store closes.
• TestRoundRobin holds the first target until the second job is submitted, then expects a1 b1 a2 b2 a3 b3 from one worker.
• Task 02's tests (runner_core_test.go) must still pass: with workers, a job there may already be running or done where it was queued; those tests allow for that.

Steps

• 1. git switch p6; copy the given test. 2. See it fail (the jobs never finish).
• 3. worker.go. 4. The batch.go changes.
• 5. gofmt -w; go test -race -count=3 ./internal/batch/ → ok. 6. make gate.
• 7. Row p6/03-runner-workers; commit.

git add internal/batch docs/implementer-log.md
git commit -F .state/commit-msg.txt

Done when

• Both given tests pass under -race -count=3; make gate ok; the given files are byte-identical.

Stop and report if

• A given test's timing can only be met by sleeping longer than RetryWait inside a worker.