# P6 task 02: the batch runner in `internal/batch`

**Branch:** `p6` (`git switch p6`; `git status --short` must be empty, otherwise stop)
**Commit subject:** `batch: store-backed job runner — serial per job, round-robin workers, caps, idempotency, cancel, retry`

## Goal

A new package `internal/batch`: the job lifecycle. It creates a job, lets a small pool of
workers drain its targets one at a time, writes each outcome to the store, handles cancellation,
retries a target whose fetch was only busy, and enforces the caps and idempotency.

It knows **nothing** about HTTP, the guard, or the real fetch pipeline: the per-target work is an
injected `FetchFunc`, and persistence is the `*dcache.DB` from task 01.

## Files

- Copy: `_files/batch/runner_test.go` → `internal/batch/runner_test.go`
- Create: `internal/batch/batch.go` (types, `New`, `Submit`, `Cancel`, `Status`, `Touch`, `Close`),
  `internal/batch/worker.go` (the workers and the drain)
- Modify: `docs/implementer-log.md`

## What already exists (do not re-read these packages; this is their API)

```go
// internal/dcache (task 01). Every time argument is unix seconds.
type BatchJob struct{ ID, Caller, CallerClass, IDemKey, Status, Error string; TargetsTotal, TargetsDone int; CreatedAt, Lease, NBytes int64 }
type BatchTarget struct{ JobID string; Idx int; URL, Options, Status string; Result, Error, Hint *string }
func (d *DB) CreateBatchJob(j BatchJob) error
func (d *DB) BatchJob(id string) (BatchJob, bool)                          // a pure read
func (d *DB) BatchJobByIDem(caller, idemKey string, now int64, ttl time.Duration) (string, bool) // queued/running/done/partial only
func (d *DB) LiveBatchJobs(caller string) int                             // queued + running
func (d *DB) LiveBatchJobsTotal() int
func (d *DB) PutBatchTargets(jobID string, targets []BatchTarget) error
func (d *DB) BatchTargets(jobID string) ([]BatchTarget, bool)             // in idx order
func (d *DB) UpdateBatchTarget(jobID string, idx int, now int64, status string, result, errMsg, hint *string) error
func (d *DB) SetBatchJobStatus(id, status, errMsg string, now int64) error
func (d *DB) TouchBatchJob(id string, now int64) error
func (d *DB) SweepBatch(now int64, ttl time.Duration) int
func (d *DB) EvictBatchOldest(maxBytes int64) int
const ClassFleet, ClassRestricted = "fleet", "restricted"

// internal/auth
type Caller struct{ Name, Workspace, KeySHA256 string }
```

## Interfaces (exact: the given test compiles against them)

```go
package batch

type Options struct {
	Mode     string `json:"mode,omitempty"`
	Question string `json:"question,omitempty"`
	Pages    string `json:"pages,omitempty"`
	Grep     string `json:"grep,omitempty"`
	JQ       string `json:"jq,omitempty"`
	Context  *bool  `json:"context,omitempty"` // nil = /v1/fetch's default (on when there is a question)
	Recover  bool   `json:"recover,omitempty"`
}

type Target struct {
	URL     string  `json:"url"`
	Options Options `json:"options"`
}

// Result is a finished target: the /v1/fetch envelope's content, and every other envelope field
// in Meta. It is stored JSON-encoded in the target's result column.
type Result struct {
	Content string         `json:"content"`
	Meta    map[string]any `json:"meta,omitempty"`
}

// TargetError is a failure with an agent-facing hint. Status is the HTTP status the same
// failure gets from /v1/fetch; the runner ignores it (the metrics hook reads it). Any other
// error is stored with its Error() text and no hint.
type TargetError struct {
	Status    int
	Msg, Hint string
}

func (e *TargetError) Error() string { return e.Msg }

// FetchFunc does one target for the caller that submitted the job.
type FetchFunc func(ctx context.Context, c *auth.Caller, t Target) (Result, error)

type Job struct {
	ID           string
	Caller       string
	Status       string // queued | running | done | partial | failed | cancelled
	Error        string
	TargetsTotal int
	TargetsDone  int
	CreatedAt    int64 // unix seconds
}

type TargetStatus struct {
	Idx    int
	URL    string
	Status string // pending | running | done | failed
	Result *Result
	Error  string
	Hint   string
}

// Hooks observe the runner for metrics. Any nil field is skipped.
type Hooks struct {
	Job        func(outcome string)                              // created, reused, done, partial, failed, cancelled
	Target     func(err error, d time.Duration, resultBytes int) // once per stored target; err nil on success
	JobSeconds func(d time.Duration)                             // submit → terminal, once per job
	Depth      func(live int)                                    // live jobs, after every change
	Inflight   func(n int)                                       // targets being fetched, after every change
}

type Config struct {
	TTL           time.Duration // 0 → 24h
	TargetsPerJob int           // 0 → 32
	JobsPerCaller int           // 0 → 1
	JobsTotal     int           // 0 → 8
	Workers       int           // 0 → 1; values above 2 are clamped to 2
	MaxBytes      int64         // 0 → 64 MiB
	RetryWait     time.Duration // 0 → 2s: the pause before a busy target is tried again
	RetryFor      time.Duration // 0 → 10m: how long one target may keep coming back busy
	Hooks         Hooks
}

var (
	ErrNoTargets      = errors.New("no targets")
	ErrTooManyTargets = errors.New("too many targets")
	ErrCallerBusy     = errors.New("busy")
	ErrBusy           = errors.New("jobs busy")
	ErrRetryLater     = errors.New("retry later") // a FetchFunc wraps this when the target was only busy
)

func New(db *dcache.DB, fetch FetchFunc, cfg Config, log *slog.Logger) *Runner // starts the workers
func (r *Runner) Submit(ctx context.Context, c *auth.Caller, callerClass string, targets []Target) (id string, reused bool, err error)
func (r *Runner) Cancel(id string) bool
func (r *Runner) Status(id string) (Job, []TargetStatus, bool) // a pure read
func (r *Runner) Touch(id string)                              // bumps the lease to now
func (r *Runner) Close()                                       // stops the workers; waits for an in-flight fetch to return
```

## Rules

1. **Submit, in this order:**
   - `len(targets) == 0` → `ErrNoTargets`.
   - **Idempotency.** `idem_key` = hex SHA-256 of `c.Name + "\x00" + json.Marshal(targets)`.
     If `BatchJobByIDem(c.Name, key, now, TTL)` finds a job, return its id with `reused = true`:
     no new job, no caps, `Hooks.Job("reused")`.
   - **Caps:** more than `TargetsPerJob` → `ErrTooManyTargets`; `LiveBatchJobs(c.Name) >= JobsPerCaller`
     → `ErrCallerBusy`; `LiveBatchJobsTotal() >= JobsTotal` → `ErrBusy`.
   - Store the job **`queued`** (random 16-byte hex id, `CreatedAt` = `Lease` = now) and its targets
     (`pending`, `Options` = the JSON of `t.Options`), **before returning**. Keep `c` in memory
     for this job: jobs never resume after a restart, so the caller is never persisted.
   - `Hooks.Job("created")`; `SweepBatch(now, TTL)`; wake a worker.
2. **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`.
3. **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)`.
4. **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.
5. **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.
6. **Status(id)** maps the store rows; `Result` is decoded from the result column. It writes
   nothing. **Touch(id)** is `TouchBatchJob(id, now)`; the API calls it after an owner's read.
7. **Hooks.Depth** gets the live-job count and **Hooks.Inflight** the in-flight count whenever
   either changes.
8. Library code never panics; a store error is logged (`Warn`, with the job id, never a URL) and
   the job goes on. Log lines carry the job id, counts and statuses, **never a URL or host**.

## Facts

- The given test uses an in-memory `dcache` and `FetchFunc`s 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.

## Steps

- [ ] **1.** `git switch p6`; copy the given test. **2. See it fail** (does not compile).
- [ ] **3.** `batch.go`, **4.** `worker.go`.
- [ ] **5.** `gofmt -w`; `go test -race -count=3 ./internal/batch/` → `ok`. **6.** `make gate`.
- [ ] **7.** Row `p6/02-runner`; commit.

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

## Done when

- The given test passes under `-race -count=3`; `make gate` ok; the given file is byte-identical.

## Stop and report if

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