P6 plan, owner review folded in: caller travels with the job, 1–2 round-robin workers, busy retry, summary reads, local-only tests, 8 tasks

Kyle: "Fold them into the plan." What changed and why is in docs/P6.md §14. In short:
- FetchFunc gets the *auth.Caller (the given api test could not compile, and a fetch with no
  caller would bypass per-caller gates, the pool limit, the cache class and redaction);
- store reads never bump the lease (task 01's rule contradicted its own test); the API Touches;
- 1-2 workers round-robin over live jobs, so a batch holds at most 2 of the 8 global slots;
  busy targets retry in the runner (bounded); batch PDFs use their own pool name; 210 s deadline;
- the api tests no longer touch the network or race the cancel; private names via fake DNS;
- failed/cancelled jobs are not reused; GET is a summary plus GET …/{idx} and ?results=1;
- the 256 KiB body cap moves into ServeHTTP; a fixed metrics label for /v1/batch/{id};
- guard.PreCheck (given verbatim) replaces guard.ParseURL at submit;
- 5 tasks became 8 (refactor, wiring, endpoints and serve/config split), each restating the
  APIs it uses; every given test compiles against a panic-only skeleton and fails on it.

P6 starts after P5c: task 04 moves v1fetch, which P5b changes. Recheck the api given tests
against post-P5c master before the run.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-09-26 17:55:50 -07:00
co-authored by Claude Opus 5.5
parent 63e70e19fc
commit 218eaa8b53
20 changed files with 1854 additions and 617 deletions
+108 -31
View File
@@ -32,8 +32,11 @@ list, walk away, and collect once. It builds on the P5 infrastructure that alrea
| Method | Path | Result |
|---|---|---|
| `POST /v1/batch` | submit a list of targets | `202` created/reused, `400` malformed, `422` refused up front or over a cap, `429` caller/jobs busy |
| `GET /v1/batch/{id}` | status + per-target results (inline) | `200`, `401`/`404` |
| `DELETE /v1/batch/{id}` | cancel | `200`, `401`/`404` |
| `GET /v1/batch/{id}` | status summary; `?results=1` inlines results | `200`, `401`/`404` |
| `GET /v1/batch/{id}/{idx}` | one target's status and result | `200`, `401`/`404` |
| `DELETE /v1/batch/{id}` | cancel | `200`, `409` already finished, `401`/`404` |
With `[batch] disabled = true` every batch endpoint answers `503 batch disabled`.
`POST /v1/batch` request — each target is the `/v1/fetch` field set minus the outer URL:
@@ -47,17 +50,20 @@ Response `202`:
{"success": true, "job_id": "…", "status": "queued", "targets": 3, "reused": false}
```
`GET /v1/batch/{id}`:
`GET /v1/batch/{id}` (a summary: no results, since 32 inline results can be ~6 MB in one agent
tool call):
```json
{"job_id": "…", "status": "done", "targets_total": 3, "targets_done": 3,
"created_at": "…", "targets": [
{"idx": 0, "url": "…", "status": "done", "result": { …/v1/fetch envelope… }},
{"job_id": "…", "status": "partial", "targets_total": 3, "targets_done": 3,
"created_at": "2026-09-27T01:02:03Z", "targets": [
{"idx": 0, "url": "…", "status": "done"},
{"idx": 1, "url": "…", "status": "failed", "error": "blocked: private address …", "hint": "browser"}
]}
```
Results are inline regardless of job status, once a target is finished. A `pending` target has
`status: "pending"` and no `result`/`error`. `DELETE` → `200 {"success": true, "status":
`?results=1` adds `"result"` (the `/v1/fetch` envelope) to every `done` target.
`GET /v1/batch/{id}/{idx}` returns `{"job_id", "idx", "url", "status", "result"?, "error"?,
"hint"?}`. A `pending` target has `status: "pending"` and no `result`/`error`. A `202` for a new
job says `"status": "queued"`; a reused one says the job's current status. `DELETE` → `200 {"success": true, "status":
"cancelled"}`.
Job status vocabulary: `queued → running → done | partial | failed | cancelled`. Target status:
@@ -87,7 +93,15 @@ the only injected behavior is a `FetchFunc func(ctx, Target) (Result, error)`.
Idempotency is lease-scoped, so it is "reuse within TTL, fresh after".
- **Caps.** targets/job 32; live jobs/caller 1; live jobs total 8. "Live" is by status
(`queued`/`running`), not lease. Exceeding a cap is a typed error the API maps to `422`/`429`.
- **The caller.** `Submit` takes the `*auth.Caller`; the runner keeps it in memory with the job
and passes it to every `FetchFunc` call (`func(ctx, *auth.Caller, Target)`), so the fetch runs
under the caller's gates, pool slot, cache class and redaction. Only its name and class are
stored.
- **Workers.** `workers` (default 1, at most 2) share one round-robin over the live jobs, one
target at a time. A job is `queued` until a worker first takes it, then `running`.
- **Serial.** Targets drain in `idx` order; target `i+1` does not start until `i` is written.
- **Busy.** A `FetchFunc` error wrapping `ErrRetryLater` (the API maps fetch `429`/`503` to it)
leaves the target pending; it is tried again after 2 s, for up to 10 min, then failed.
- **Failure isolation.** One target's error is stored per-target and the job continues; the job
status is `done` (all ok), `failed` (all failed), or `partial` (mixed).
- **Cancel.** `DELETE` flips the job `cancelled`, stops before the next target, and lets the
@@ -95,15 +109,26 @@ the only injected behavior is a `FetchFunc func(ctx, Target) (Result, error)`.
## 6. Scheduling and fairness
The worker runs in the background, and every target still goes through the normal `acquire` gate
(8 global, 2 per-host). On `429 busy` the worker **waits and retries** instead of failing — nobody
is holding a socket open. Interactive requests are served by HTTP handlers that grab slots
immediately, so they win by construction; the batch is lower priority for free.
The workers run in the background, and every target still goes through the normal `acquire` gate
(8 global, 2 per-host). With 1–2 workers the batch holds at most 1–2 of the 8 global slots, so
interactive requests always have room. (The first draft let 8 live jobs each run a target at
once, which could fill the whole gate.) On `429 busy` the worker **waits and retries** instead of
failing: nobody is holding a socket open.
Two shared resources get the same treatment:
- **The PDF pool.** A batch extraction uses the pool caller name `<caller>/batch`, so it never
takes the caller's own interactive slot. The single-flight key is unchanged, so a batch and an
interactive request for the same PDF still share one child.
- **The digest gate.** A batch question queues on it like any other digest. With one worker
that is at most one batch digest at a time.
Each batch fetch runs under the same 210 s deadline as a `/v1/fetch` request.
## 7. Retention and TTL
A job carries a single sliding **lease**, bumped on submit, on every target completion, and on
every status/collect read. All targets therefore share one expiry (the last completion's), and a
every read by its owner (the API calls `Touch` after the ownership check; the store's reads
never write). All targets therefore share one expiry (the last completion's), and a
slow consumer's reads keep a finished batch alive. GC deletes a job and its targets once the lease
is older than the TTL (24 h). Completed results survive a restart and stay readable until the lease
lapses; `queued`/`running` jobs are marked `failed` (`restart`) on startup and never resumed.
@@ -134,11 +159,13 @@ becomes a label. `result_bytes` is what sizes the byte budget (64 MiB).
## 10. CLI and skill
- `sift batch submit <targets.json>` (`-` reads stdin; prints `job_id` and `reused`).
- `sift batch status <job_id>` (prints the full status body).
- `sift batch status [--results] <job_id>` (the summary; `--results` inlines every result).
- `sift batch get <job_id> <idx>` (one result).
- `sift batch cancel <job_id>`.
- `skill/SKILL.md`: use `sift batch submit` for a list of targets instead of a loop of
`sift fetch`; `sift batch status` to collect.
- `[batch]` config block: `targets_per_job`, `jobs_per_caller`, `jobs_total`, `ttl`, `max_bytes`.
- `[batch]` config block: `disabled`, `targets_per_job`, `jobs_per_caller`, `jobs_total`,
`workers`, `ttl`, `max_bytes`.
The `/v1/batch` body cap (256 KiB, vs the 64 KiB global) is a constant, not config.
## 11. Refactor of existing code (explicit)
@@ -146,28 +173,39 @@ becomes a label. `result_bytes` is what sizes the byte budget (64 MiB).
`v1fetch` today inlines the whole per-target pipeline (guard → `doFetch` → `recover` →
render/digest → context → envelope). The batch worker needs the same pipeline for one target. It is
extracted into one shared method, `func (s *Server) fetchTarget(ctx, *auth.Caller, batch.Target)
(batch.Result, error)`, that both `v1fetch` and the runner's `FetchFunc` call. `v1fetch` keeps its
(batch.Result, error)`, that both `v1fetch` and the runner's adapter (`batchFetch`) call. It is its
own task (04), done before any batch endpoint exists, so the diff is a pure move. `v1fetch` keeps its
exact response shape and behavior — its existing tests must pass unchanged. This is the one change
to working code in P6 and is called out for reviewer attention.
## 12. Acceptance (contract) tests
Acceptance-first, as in P5a: given tests compiled against a panic-only skeleton, failing for the
intended reasons. See `docs/plans/p6/_files/`.
intended reasons. See `docs/plans/p6/_files/` and the task table in `docs/plans/p6/README.md`.
1. `dcache`: create/read round trip, idempotency lookup (lease expiry + caller scope), live-job
counts, ordered targets, exactly-once `targets_done`/lease bump, sweep, byte-eviction
(oldest-first, never a live job), `MarkInFlightFailed`, `BatchNbytes`.
2. `batch` runner: serial order (peak in-flight 1), failure isolation (`partial`), idempotent
resubmit (`reused`, no double-drain), the three caps, cancel (in-flight stored, later targets
`pending`), panic recovery.
3. `api`: submit `202` + idempotent resubmit; a guard-pre-check rejection (non-80/443 port) is
`422` at submit; 33 targets `422`; a private-address target (passes pre-check, blocked at
connect) is a per-target `failed`, aggregate `partial`; ownership (`GET` by another caller →
`404`); cancel → `cancelled`.
4. `metrics`: exact series names, label values, gauge values, histogram counts, and raw-outcome
folding to `other`.
5. `cmd/sift`: `parseBatchArgs` verb/arg pairs and error cases.
1. `dcache`: round trip, idempotency lookup (lease expiry, caller scope, status filter), live
counts, ordered targets, exactly-once `targets_done` and lease bumps, sweep, byte eviction
(oldest first, never a live job), `MarkInFlightFailed`, `BatchNbytes`.
2. `batch` runner:
- serial order (peak in flight 1), with the caller passed to every fetch;
- round-robin between jobs, and `queued` → `running`;
- failure isolation (`partial`, `failed`);
- idempotent resubmit, including after `done`, but a new job after a cancel;
- all three caps and `ErrNoTargets`; cancel (the in-flight target stored, later ones `pending`);
- busy retry and its limit; panic recovery (the panic value not echoed); the hooks.
3. `metrics`: series names, labels, folding to `other`.
4. `api` `fetchTarget`: the same content and envelope as `/v1/fetch`; its errors as `*fetch.Error`.
5. `api` wiring: the runner through `fetchTarget`, a private-resolving name blocked per target,
metrics through the hooks, outcome classes, busy → `ErrRetryLater`.
6. `guard.PreCheck`: no resolution; the port, scheme, userinfo and literal rules; the allow list.
7. `api` endpoints:
- submit `202` and idempotent resubmit;
- `400`/`422` refusals (the guard pre-check naming the index, 33 targets);
- the 256 KiB body; the per-target `blocked` failure giving `partial`;
- summary, one result, `?results=1`;
- ownership `404` (GET, one result, DELETE); cancel and `409`;
- no job id in `/metrics`.
8. `config`: the `[batch]` block, `ParseSize`. `cmd/sift`: `parseBatchArgs`.
## 13. Rollout
@@ -180,4 +218,43 @@ intended reasons. See `docs/plans/p6/_files/`.
## 14. Changes after the plan was written
(none yet)
**2026-09-26 evening, owner review (Claude), at Kyle's direction ("fold them into the plan").**
The design stands; these were implementation-blocking or would have hurt agents:
1. **The caller was lost.** `FetchFunc` had no caller and the runner was built once, so the given
api test (`batch.New(db, s.fetchTarget, …)`) could not compile against the specified
`fetchTarget(ctx, *auth.Caller, Target)`. Made to compile, it would have fetched every target
with no caller: no per-caller gates, pool limit, cache class or redaction. Now `Submit` takes
the `*auth.Caller` and `FetchFunc` receives it (§5).
2. **Task 01 rule 1 contradicted its test.** Store reads no longer bump the lease; the API calls
`Touch` after an owner's read (§7).
3. **Batch could starve interactive fetches.** 8 live jobs × 1 target filled the 8-slot global
gate. The busy-retry of §6 was in no task. Now there are 1–2 round-robin workers, busy-retry
is in the runner, a batch PDF uses its own pool name, and a per-target deadline applies (§5, §6).
4. **Flaky, networked API tests.** Two tests fetched `https://example.com/`, and the cancel test
raced the fetch. Every api test is now local, using held `httptest` servers and a fake DNS
name for the private-address case.
5. **A resubmit could return a dead job.** A failed or cancelled job was reused for 24 h, and
polling it kept extending that. Now only `queued`/`running`/`done`/`partial` jobs are reused.
6. **Reads were too large.** `GET` inlined every result (up to ~6 MB). It is now a summary, with
`GET …/{idx}` and `?results=1` (§3). The CLI gains `get` and `status --results`.
7. **Body cap.** The global 64 KiB `MaxBytesReader` runs in `ServeHTTP`, so a handler cannot
raise it. `ServeHTTP` now picks 256 KiB for `/v1/batch`. Request metrics use a fixed
`/v1/batch/{id}` label, so a job id never becomes a label value.
8. **Submit pre-check.** `guard.ParseURL` rejects the test's own loopback port and knows nothing
of the allow list. New `Guard.PreCheck` is `CheckURL` without resolution (given verbatim in
task 05). A private IP literal is now refused at submit; a *name* that resolves privately
still fails per target at fetch time.
9. **Task shape.** 5 tasks → 8, one package each where possible. The old task 03 (refactor +
endpoints + main) is now 04 (refactor only), 05 (runner wiring), 06 (endpoints) and 07
(serve/config). Every task restates the APIs it uses from earlier tasks. Every given test
compiles against a panic-only skeleton and fails there on `panic: todo`.
10. **Also fixed:**
- the `ErrBusy` (jobs total) cap now has a test;
- the `202` status is `queued`, consistent with the workers;
- the `restart` outcome is counted by `serve`;
- `Options.Context` is `*bool`, keeping `/v1/fetch`'s default.
**Sequencing.** P6 starts after P5c merges. P5b/P5c change `v1fetch` (the digest meta gains
`retrieval`/`embedded`), which is the code task 04 moves. Recheck the api given tests against
a skeleton built on post-P5c master before the run.
+42 -20
View File
@@ -1,8 +1,11 @@
# 01-batchstore — batch tables, lease, GC, byte budget in `internal/dcache`
# P6 task 01: batch tables, lease, GC, byte budget in `internal/dcache`
**Branch:** `p6` (`git switch -c p6 master` if it does not exist, else `git switch p6`; `git status --short` must be empty, otherwise stop)
**Commit subject:** `dcache: batch job and target tables with a sliding lease, sweep and byte budget`
Add the batch job store to the existing `internal/dcache` SQLite database. No HTTP, no worker —
this task is storage only. `dcache` owns the single SQLite connection, so the batch tables and
their accessors live here; the runner (task 02) and API (task 03) call them.
their accessors live here; the runner (task 02) and API (task 05) call them.
## Context
@@ -43,7 +46,7 @@ CREATE INDEX IF NOT EXISTS batch_job_lease ON batch_job(lease);
```
`result`/`error`/`hint` are `NULL` until the target finishes. The `result` JSON is the
`/v1/fetch`-shaped envelope, bounded by the output cap at write time (task 03 enforces the cap;
`/v1/fetch`-shaped envelope, bounded by the output cap at write time (the render step enforces the cap;
the store just stores what it is given and counts its bytes into `nbytes`).
## Public surface (exact names — the given test compiles against them)
@@ -96,38 +99,57 @@ caller decides the clock, so the given test asserts exact lease values.
## Rules
1. **Sliding lease.** `CreateBatchJob` stores the job as given (the caller sets `Lease`/`CreatedAt`).
`UpdateBatchTarget`, `SetBatchJobStatus`, `TouchBatchJob`, and `MarkInFlightFailed` bump the job
lease to their `now`. `BatchJob` (a read) also bumps the lease to the current wall-clock time so
a slow consumer's reads keep the job alive.
`UpdateBatchTarget`, `SetBatchJobStatus`, `TouchBatchJob`, and `MarkInFlightFailed` set the job
lease to their `now`. **Reads never write:** `BatchJob`, `BatchTargets` and the counts change
nothing, so the given test can assert exact lease values after reading. A consumer's read keeps
a job alive because the API calls `TouchBatchJob` after it (task 05), not because the read does.
2. **No partial-job reads leak a competing caller.** Every accessor is keyed by `job_id` only; the
caller filter is the API layer's job (task 03). The store stores what it is told and returns
caller filter is the API layer's job (task 05). The store stores what it is told and returns
what is asked.
3. `BatchTargets` returns targets in `idx` order. `UpdateBatchTarget` increments `targets_done`
3. **Idempotency lookup.** `BatchJobByIDem(caller, idemKey, now, ttl)` returns the id of a job with
that `(caller, idem_key)` whose `lease + ttl > now` **and whose status is `queued`, `running`,
`done` or `partial`**. A `failed` job (including one killed by a restart) or a `cancelled` one
is never returned: resubmitting the same list after either starts a new job. If several match,
return the newest by `created_at`.
4. `BatchTargets` returns targets in `idx` order. `UpdateBatchTarget` increments `targets_done`
exactly once per target that transitions *into* `done`/`failed` (a re-write of the same target
does not double-count), bumps the job lease to `now`, and adds `len(result)` to `nbytes` when a
result is written.
4. `SweepBatch(now, ttl)` deletes every job whose `lease + ttl <= now`, and its target rows, and
returns the number of jobs deleted. It is the only thing that removes a finished job by time.
5. `EvictBatchOldest(maxBytes)` evicts **whole jobs** (oldest `lease` first) until `BatchNbytes()`
5. `SweepBatch(now, ttl)` deletes every job whose `lease + ttl <= now`, and its target rows, and
returns the number of jobs deleted. It is the only thing that removes a finished job by time; the runner calls it (task 02).
6. `EvictBatchOldest(maxBytes)` evicts **whole jobs** (oldest `lease` first) until `BatchNbytes()`
is `<= maxBytes`, and returns the number evicted. A job in `queued`/`running` is never
byte-evicted (only lease-GC, cancel, or restart removes an active job). It runs after a target
result is written and is separate from the content-cache `evict()`.
6. `MarkInFlightFailed(reason, now)` marks every `queued`/`running` job `failed` with
byte-evicted (only lease-GC, cancel, or restart removes an active job). The runner (task 02) calls it
after each target result is written; it is separate from the content-cache `evict()`.
7. `MarkInFlightFailed(reason, now)` marks every `queued`/`running` job `failed` with
`error=reason`, bumps their lease to `now`, and returns the count. This is the restart
reconciliation hook; `main` calls it on startup (task 03).
7. `BatchNbytes()` sums `nbytes` over `batch_job`. Batch bytes are **not** included in `Size()` or
reconciliation hook; `serve` calls it on startup (task 06).
8. `BatchNbytes()` sums `nbytes` over `batch_job`. Batch bytes are **not** included in `Size()` or
the content `evict()` union; batch is a coherent unit with its own budget.
8. Accessors follow the existing `dcache` style: return `(value, bool)` or `error`, never panic,
9. Accessors follow the existing `dcache` style: return `(value, bool)` or `error`, never panic,
count a miss in the appropriate `*M` counter. Reuse the existing `ClassFleet`/`ClassRestricted`
constants for `caller_class`.
## Given
Copy `_files/dcache/batch_test.go` → `internal/dcache/batch_test.go`. It opens an in-memory `DB`
and asserts the create/read round trip, `BatchJobByIDem` lease expiry + caller scoping, live-job
and asserts the create/read round trip, `BatchJobByIDem` lease expiry, caller scoping and the
status filter (rule 3), live-job
counts, ordered targets, the exactly-once `targets_done`/lease-bump rules, `SweepBatch`,
`EvictBatchOldest` (oldest-first, never a live job), `MarkInFlightFailed`, and `BatchNbytes`.
## Gate
## Steps
`make gate` green; `go test ./internal/dcache/` green under `-race`.
- [ ] **1.** Branch; copy the given test. **2. See it fail** (`go test ./internal/dcache/` does not compile).
- [ ] **3.** Schema in `Open`, the types, the methods, in a new `internal/dcache/batch.go`.
- [ ] **4.** `gofmt -w`; `go test -race -count=3 ./internal/dcache/` → `ok` (every existing test too).
- [ ] **5.** `make gate`. **6.** Row `p6/01-batchstore` in `docs/implementer-log.md`; commit.
```sh
git add internal/dcache docs/implementer-log.md
git commit -F .state/commit-msg.txt
```
## Done when
- The given and existing dcache tests pass under `-race -count=3`; `make gate` ok; the given file is byte-identical.
+146 -62
View File
@@ -1,18 +1,49 @@
# 02-runner — serial job runner in `internal/batch`
# P6 task 02: the batch runner in `internal/batch`
Add a new package `internal/batch`: the job lifecycle and the serial state machine. It knows how
to create a job, drain its targets one at a time, write each target's outcome, bump the lease,
handle cancellation, and enforce the caps and idempotency. It knows **nothing** about HTTP, the
guard, or the real fetch pipeline: the per-target work is an injected function, and persistence is
the `*dcache.DB` from task 01.
**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`
## Context
## Goal
`internal/pdfpool` is the wrong shape (in-memory, coalescing, forgets results). The runner is the
opposite: store-backed, persistent, one job = many distinct targets, results written to the store
as they complete. It is deliberately serial.
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.
## Public surface (exact names — the given test compiles against them)
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
@@ -23,104 +54,157 @@ type Options struct {
Pages string `json:"pages,omitempty"`
Grep string `json:"grep,omitempty"`
JQ string `json:"jq,omitempty"`
Context bool `json:"context,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,omitempty"`
Options Options `json:"options"`
}
// Result is a completed target's outcome. The runner JSON-encodes it into the
// store's target.result column and decodes it back in Status.
// 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"`
}
type FetchFunc func(ctx context.Context, t Target) (Result, error)
// 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)
// Status views returned by Status (mapped from the store rows).
type Job struct {
ID string
Status 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
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
func (r *Runner) Submit(ctx context.Context, caller, callerClass string, targets []Target) (id string, reused bool, err error)
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)
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
```
The runner depends on `*dcache.DB` (task 01) for persistence; the `FetchFunc` is the only injected
behavior. Zero `Config` fields fall back to the defaults above.
## Rules
1. **Idempotency first.** `Submit` derives `idem_key = SHA256(caller + "\x00" + canonical(targets))`
where `canonical` is `json.Marshal` of the decoded targets (so field order and whitespace don't
matter). If a job whose lease has not expired (`BatchJobByIDem`) exists for the same
`(caller, idem_key)` — any status — return it with `reused=true` and do **not** create a new job
or enforce the caps. Idempotency is lease-scoped, so it is "reuse within TTL, fresh after".
2. **Caps second.** targets/job > `TargetsPerJob` → `ErrTooManyTargets`; a live (`queued`/`running`)
job already exists for the caller → `ErrCallerBusy`; live jobs total >= `JobsTotal` → `ErrBusy`.
"Live" is by status, not lease — read from `LiveBatchJobs`/`LiveBatchJobsTotal`. The typed errors
map to `422`/`429` in the API layer.
3. **Submit transitions to `running` synchronously** (before the drain goroutine starts) so a
caller-busy check against a just-submitted job is deterministic. The job is created `queued`
then immediately `running`; `queued` is only ever persisted as the crash-window state that
`MarkInFlightFailed` reaps.
4. **Serial drain.** The worker iterates targets in `idx` order, one at a time, and does not start
target `i+1` until `i` is written to the store. Each target's `FetchFunc` result or error is
written immediately (JSON-encoded); a target failure does not stop the job.
5. **Lease on completion.** Each target write bumps the job lease (task 01); the first and last
target therefore share one expiry.
6. **Cancellation.** `Cancel` flips the job to `cancelled` and signals the worker; the worker
finishes the in-flight target, stores it, and stops before the next. A `cancelled` job's
completed targets stay readable until the lease lapses. `Cancel` returns false for an unknown id
or a job already in a terminal state.
7. **Terminal statuses.** When the last target is written: all `done` → `done`; all `failed` →
`failed`; a mix → `partial`. Set exactly once.
8. **No restart concept in the runner.** It is created fresh each process start; `main` (task 03)
calls `dcache.MarkInFlightFailed("restart", now)` before constructing the runner. Completed
targets on a restart-killed job stay readable; the job is never resumed.
9. **No panic on input.** Library code never panics: bad ids, missing targets, a `FetchFunc` panic
is recovered into a target error and the job continues. The drain runs under
`context.WithoutCancel` of a background context; only the per-job cancellation stops it.
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**.
## Given
## Facts
Copy `_files/batch/runner_test.go` → `internal/batch/runner_test.go`. It uses an in-memory `dcache`
plus a `FetchFunc` that records call order and can fail or panic. Asserts serial order (peak
in-flight 1), failure isolation (`partial`), idempotent resubmit (`reused`, no double-drain), the
three caps, cancel (in-flight stored, later targets `pending`), and panic recovery.
- 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.
## Gate
## Steps
`make gate` green; `go test ./internal/batch/` green under `-race`.
- [ ] **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.
-110
View File
@@ -1,110 +0,0 @@
# 03-api — `/v1/batch` endpoints in `internal/api`
Wire the batch runner into the API server. This is the HTTP surface and the only place the guard,
auth, caps-as-HTTP-statuses, and restricted-caller redaction meet the runner. New code goes in a
new file `internal/api/batch.go`; do **not** grow `api.go`.
## Endpoints
| Method | Path | Result |
|---|---|---|
| `POST /v1/batch` | submit a list of targets | `202` job created/reused, `400` malformed, `422` a target refused up front or over a cap, `429` caller/jobs busy |
| `GET /v1/batch/{id}` | status + per-target results | `200` (results inline), `401`/`404` |
| `DELETE /v1/batch/{id}` | cancel | `200` cancelled, `401`/`404` |
Register them in `New` and add `/v1/batch`, `/v1/batch/` to the `ServeHTTP` fast-path switch
(no `requestDeadline` — submit/poll are quick; the worker is not under any request context).
## Request/response shapes (exact JSON — the given test compares these)
`POST /v1/batch`:
```json
{"targets": [{"url": "…", "mode": "clean", "question": "…", "pages": "…", "grep": "…",
"jq": "…", "context": true, "recover": false}, …]}
```
Each target carries the same fields as `/v1/fetch` minus the outer `url` semantics (each target is
its own URL). A `targets` list is required and non-empty; otherwise `400`.
Response `202`:
```json
{"success": true, "job_id": "…", "status": "queued", "targets": 3, "reused": false}
```
`GET /v1/batch/{id}`:
```json
{"job_id": "…", "status": "done", "targets_total": 3, "targets_done": 3,
"created_at": "…", "targets": [
{"idx": 0, "url": "…", "status": "done", "result": { …/v1/fetch envelope… }},
{"idx": 1, "url": "…", "status": "failed", "error": "blocked: private address …", "hint": "browser"}
]}
```
`result` is the `/v1/fetch`-shaped envelope (content, mode, untrusted, source, warnings, sizes,
timings, and pdf/digest/context when present), written by the runner's `FetchFunc` adapter. A
`failed` target carries `error` and optional `hint` instead of `result`. Results are inline
regardless of status, once a target is finished; a pending target has `status: "pending"` and no
`result`/`error`.
`DELETE /v1/batch/{id}` → `200 {"success": true, "status": "cancelled"}`.
## Refactor of existing code (explicit — this is the one task that edits working code)
`v1fetch` today inlines the whole per-target pipeline: guard → `doFetch` → `recover` →
render/digest → context → envelope. The batch worker needs the *same* pipeline for one target. Do
**not** copy-paste that pipeline into the batch adapter — extract it into one shared method that
both `v1fetch` and the runner's `FetchFunc` call:
```go
func (s *Server) fetchTarget(ctx context.Context, c *auth.Caller, t batch.Target) (batch.Result, error)
```
`v1fetch` is refactored to call it, keeping its exact response shape and behavior — the existing
`v1fetch` tests must pass unchanged. The runner is wired as a new field `s.Batch *batch.Runner`,
and its `FetchFunc` is a closure over `s.fetchTarget` bound to the authenticated caller. This
refactor is called out explicitly for reviewer attention: flag any behavioral drift between the
two callers. The per-target result is capped by the output cap before it is stored.
## Rules
1. **Guard pre-check at submit.** Run `guard.ParseURL` (the no-resolution pre-check) on every
target URL before accepting. If any target fails, refuse the **whole** submit with `422` and
the offending index + reason (`blocked: …`). This is the "refused up front" gate — the
connect-time guard (private IP, rebinding) cannot run until fetch time and surfaces as a
per-target `failed`, not a submit refusal.
2. **Body cap.** The global `MaxBytesReader` caps bodies at `maxReqBody` (64 KiB); `/v1/batch`
gets its own larger limit (256 KiB) so 32 targets with questions fit. Apply it in the handler.
3. **Caps → statuses.** Map the runner's typed errors: `ErrTooManyTargets` → `422`; `ErrCallerBusy`
and `ErrBusy` → `429` with a `Retry-After` when sensible. Idempotent reuse returns `202` with
`reused: true`.
4. **Ownership.** `GET`/`DELETE` resolve the job by id, then require `job.caller == c.Name`; else
`404` (not `403` — do not reveal that another caller's job exists). The job id is the random
opaque token; `idem_key` is never returned or logged.
5. **Restricted redaction.** Pass the caller (not just its name) into the runner so each target's
fetch goes through `doFetch` with the real `*auth.Caller`; the existing `logHost`/`logReason`/
`logErr` then blank host, reason, and error detail automatically. The batch handler's own log
line logs `caller`, `job_id`, target count, and status — **never** a target URL or hostname, for
any caller.
6. **The adapter.** The runner's `FetchFunc` is the shared `s.fetchTarget` (see the refactor
section above), bound to the authenticated `*auth.Caller`. It wraps `doFetch` + `recover` +
digest + context + envelope, keyed to one `batch.Target`, returning `batch.Result` (whose `Meta`
carries the envelope's source/digest/pdf/context/warnings/sizes/timings). The per-target result
is capped by the output cap before it is stored.
7. **Job status on the wire** uses the runner's vocabulary (`queued|running|done|partial|failed|
cancelled`); unknown id → `404`. All three endpoints require a bearer key via the existing
`caller` helper (`MetricsOnly` → `403`).
## Given
Copy `_files/api/batch_test.go` → `internal/api/batch_test.go`. It builds a Server with `s.Batch`
wired over an in-memory `dcache` (pinning the `s.Batch *batch.Runner` field and the
`s.fetchTarget` adapter), then drives the endpoints. Asserts, at least:
- submit → `202` with a `job_id`; resubmit of the same body → same id, `reused: true`;
- a target the guard pre-check rejects (non-80/443 port) → `422` and no job created;
- 33 targets → `422`;
- a private-address target (passes pre-check, blocked at connect) → per-target `failed` with
`error` containing `blocked`, job aggregate `partial`;
- `GET` by another caller's key → `404`; `DELETE` by the owner → `200`, then `GET` → `cancelled`;
## Gate
`make gate` green; `go test ./internal/api/` green under `-race`.
@@ -1,7 +1,11 @@
# 04-metrics — `sift_batch_*` series in `internal/metrics`
# P6 task 03: `sift_batch_*` series in `internal/metrics`
Add the batch series and wire them from the runner and API, so the caps (32 targets, 1 job/caller,
8 jobs total) and the byte budget can be re-sized from observed data.
**Branch:** `p6` (`git switch p6`; `git status --short` must be empty, otherwise stop)
**Commit subject:** `metrics: sift_batch_* series (jobs, targets, depth, in-flight, durations, bytes)`
Add the batch series to `internal/metrics` so the caps (32 targets, 1 job/caller, 8 jobs total)
and the byte budget can be re-sized from observed data. **This task touches only
`internal/metrics`**: task 05 wires these methods to the runner's `batch.Hooks`.
## Series (names and label values exact — the given test asserts these)
@@ -15,10 +19,8 @@ Add the batch series and wire them from the runner and API, so the caps (32 targ
| `sift_batch_target_seconds` | histogram | wall time per target |
| `sift_batch_result_bytes` | histogram | stored bytes per target result |
`outcome` labels on `sift_batch_targets_total` are derived from the fetch error classification the
same way `coarseReason` does (`blocked` for a guard block, `busy` for `429`, `timeout` for `504`,
`unsupported` for `415`, `too_long` for the question-too-long `422`, `upstream_4xx`/`upstream_5xx`
otherwise; `ok` on success; `cancel` when the target was skipped by a cancel).
Classifying an error into a target outcome is task 05's job (it has the `*fetch.Error`); this
package only records the label it is given.
## Public surface (exact names — the given test compiles against them)
@@ -49,18 +51,14 @@ upstream_5xx, unsupported, too_long, cancel`.
## Rules
1. Follow the existing `internal/metrics` pattern: a new `internal/metrics/batch.go` that adds the
series to `metrics.Set`, with `Observe…` methods mirroring `ObservePDF`/`ObserveDigest`.
2. `sift_batch_jobs_total{outcome}` is incremented exactly once per job, at its terminal
transition (or `created`/`reused` at submit, `restart` at reconciliation).
3. `sift_batch_targets_total{outcome}` is incremented exactly once per target.
4. Gauges reflect the runner's live state: `queue_depth` = live job count; `targets_inflight` = the
one in-flight target across jobs (0 or 1 while a job is running, summed over jobs).
5. Histograms record durations/bytes regardless of outcome; `result_bytes` is the stored target
size, which is what sizes the byte budget.
6. The metrics are exposed through the existing `/metrics` and `/metrics.json` paths — no new
endpoint. The API server's `Snapshot`/`writeMetricsJSON` do not need to enumerate them (the
Prometheus text handler covers them), but the given test reads the `metrics.Set` values directly.
1. A new file `internal/metrics/batch.go` adds the series to `metrics.Set` (registered in `New`),
with `Observe…`/`Set…` methods mirroring `ObservePDF`/`ObserveDigest`, nil-receiver safe.
2. Both `outcome` counters fold any value outside their known set to `other`: a raw string
never becomes a label.
3. The series are exposed through the existing `/metrics` and `/metrics.json` paths: no new
endpoint. `Snapshot`/`writeMetricsJSON` do not need to list them.
4. Histogram buckets: seconds as the existing request histograms; bytes
`1K, 4K, 16K, 64K, 256K, 1M`.
## Given
@@ -68,6 +66,18 @@ Copy `_files/metrics/batch_test.go` → `internal/metrics/batch_test.go`. It dri
methods, serves the `/metrics` handler (as `pdf_test.go` does), and asserts the exact series names,
label values, gauge values, histogram counts, and that a raw outcome string is folded to `other`.
## Gate
## Steps
`make gate` green; `go test ./internal/metrics/` green under `-race`.
- [ ] **1.** `git switch p6`; copy the given test. **2. See it fail.**
- [ ] **3.** `internal/metrics/batch.go` and the registration in `New`.
- [ ] **4.** `gofmt -w`; `go test -race -count=3 ./internal/metrics/` → `ok`. **5.** `make gate`.
- [ ] **6.** Row `p6/03-metrics`; commit.
```sh
git add internal/metrics docs/implementer-log.md
git commit -F .state/commit-msg.txt
```
## Done when
- The given and existing metrics tests pass under `-race -count=3`; `make gate` ok; the given file is byte-identical.
+86
View File
@@ -0,0 +1,86 @@
# P6 task 04: extract `v1fetch`'s pipeline into `fetchTarget`
**Branch:** `p6` (`git switch p6`; `git status --short` must be empty, otherwise stop)
**Commit subject:** `api: v1fetch's per-target pipeline becomes fetchTarget (no behaviour change)`
> This is the one task in P6 that edits working code. It is a **move, not a rewrite**: the
> reviewer will diff the old `v1fetch` body against `fetchTarget` line by line.
## Goal
`v1fetch` today inlines the whole per-target pipeline: fetch → recover → PDF render or clean →
question limit → digest → context → log line → envelope. The batch worker (task 05) needs the
same pipeline for one target. Move it into one method that both call:
```go
// internal/api/target.go
func (s *Server) fetchTarget(ctx context.Context, c *auth.Caller, t batch.Target) (batch.Result, error)
```
`v1fetch` keeps its exact responses. **Every existing api test must pass unchanged.**
## Files
- Copy: `_files/api/fetchtarget_test.go` → `internal/api/fetchtarget_test.go`
- Create: `internal/api/target.go` (`fetchTarget`)
- Modify: `internal/api/api.go` (`v1fetch` shrinks; `doFetch` takes a `context.Context` in place of
the `*http.Request`, and its call sites pass `r.Context()`), `docs/implementer-log.md`
## What already exists
```go
// internal/batch (task 02)
type Options struct{ Mode, Question, Pages, Grep, JQ string; Context *bool; Recover bool }
type Target struct{ URL string; Options Options }
type Result struct{ Content string; Meta map[string]any }
```
`fetch.Error{Status int; Msg, Hint string; RetryAfter int}` is the error type `writeFetchErr`
already turns into a response.
## Rules
1. **What stays in `v1fetch`:** `s.reqN.Add(1)`, the caller check, decoding `v1Req`, and the
`400 invalid url` for a bad body or an empty URL. Then it builds a `batch.Target` from the
request (`URL`, and `Mode`, `Question`, `Pages`, `Grep`, `JQ`, `Context`, `Recover` into
`Options`), calls `s.fetchTarget(r.Context(), c, t)`, and:
- on an error, `s.writeFetchErr(w, c, req.URL, err)`;
- on success, `writeJSON(w, 200, env)` where `env` is `res.Meta` plus `"content": res.Content`.
2. **What moves into `fetchTarget`, in the same order and unchanged:** the question truncation
to 500 runes, the mode default and check, `wantCtx`, `doFetch`, the recover path
(`recover.Try(ctx, …)`), the PDF branch (render, `pages`/`grep` errors, the question-chunk
limit), the clean/raw branch, the digest, the context search, the `fetch` log line, and the
envelope.
3. **Errors that `v1fetch` wrote directly with `writeErr` become `*fetch.Error` values with the same
status, message and hint:** `400 invalid url` for a bad mode, `400 raw mode does not apply to
PDFs`. `writeFetchErr` then writes them exactly as before. The only visible difference is its
`fetch failed` log line, which is fine.
4. **The envelope:** `fetchTarget` returns `Result{Content: content, Meta: m}` where `m` holds every
other key of today's `out` map (`mode`, `untrusted`, `source`, `warnings`, `sizes`, `timings`,
and `pdf`, `digest`, `context` when present). Never put `content` in `Meta`.
5. **`doFetch(ctx, c, rawURL, jq, onlyMain)`:** `r.Context()` becomes `ctx`. That is the only change
in `doFetch`. `scrape` (the other caller) passes `r.Context()`.
6. No new behaviour, no new log fields, no renamed JSON keys. If something cannot move without
changing behaviour, stop and report it.
## Steps
- [ ] **1.** `git switch p6`; copy the given test. **2. See it fail** (`fetchTarget` undefined).
- [ ] **3.** `doFetch(ctx, …)` and its call sites. **4.** `target.go`. **5.** Shrink `v1fetch`.
- [ ] **6.** `gofmt -w`; `go test -race -count=3 ./internal/api/` → `ok`, every existing test
included. **7.** `make gate`.
- [ ] **8.** Row `p6/04-fetchtarget`; commit.
```sh
git add internal/api docs/implementer-log.md
git commit -F .state/commit-msg.txt
```
## Done when
- The two given tests and **every existing api test** pass under `-race -count=3`, with no
existing test file modified; `make gate` ok; the given file is byte-identical.
## Stop and report if
- An existing api test fails and fixing it would need a change to that test.
+149
View File
@@ -0,0 +1,149 @@
# P6 task 05: the runner wired to `fetchTarget` and the metrics
**Branch:** `p6` (`git switch p6`; `git status --short` must be empty, otherwise stop)
**Commit subject:** `api: NewBatchRunner — the batch fetch adapter, retry on busy, outcome metrics; guard.PreCheck`
## Goal
Build the production runner: its `FetchFunc` is an adapter over `fetchTarget` (task 04), and its
hooks feed the `sift_batch_*` metrics (task 03). **No HTTP endpoints yet** (task 06). Also add
`guard.PreCheck`, which task 06 uses to refuse a target at submit time.
## Files
- Copy: `_files/api/batchrun_test.go` → `internal/api/batchrun_test.go`, and
`_files/guard/precheck_test.go` → `internal/guard/precheck_test.go`
- Create: `internal/api/batch.go` (`NewBatchRunner`, `batchFetch`, `batchOutcome`)
- Modify:
- `internal/api/api.go`: only the `Batch` field.
- `internal/api/pdf.go`: only the pool caller name (rule 3).
- `internal/guard/guard.go`: add **exactly** this method, verbatim. It is the only change allowed
in the guard package.
- `docs/implementer-log.md`
```go
// PreCheck is CheckURL without name resolution: scheme, userinfo, host and port, and an IP
// literal checked against the address rules. A hostname is not resolved here; the fetch-time
// CheckURL and the dial-time Control still apply to it.
func (g *Guard) PreCheck(raw string) error {
_, literal, err := ParseURL(raw)
if err != nil && !(literal.IsValid() && g.allows(literal)) {
return err
}
if literal.IsValid() {
return g.CheckAddr(literal)
}
return nil
}
```
## What already exists
```go
// internal/batch (task 02)
type Options struct{ Mode, Question, Pages, Grep, JQ string; Context *bool; Recover bool }
type Target struct{ URL string; Options Options }
type Result struct{ Content string; Meta map[string]any }
type TargetError struct{ Status int; Msg, Hint string }
type FetchFunc func(ctx context.Context, c *auth.Caller, t Target) (Result, error)
type Job struct{ ID, Caller, Status, Error string; TargetsTotal, TargetsDone int; CreatedAt int64 }
type TargetStatus struct{ Idx int; URL, Status string; Result *Result; Error, Hint string }
type Hooks struct {
Job func(outcome string)
Target func(err error, d time.Duration, resultBytes int)
JobSeconds func(d time.Duration)
Depth func(live int)
Inflight func(n int)
}
type Config struct{ TTL time.Duration; TargetsPerJob, JobsPerCaller, JobsTotal, Workers int; MaxBytes int64; RetryWait, RetryFor time.Duration; Hooks Hooks }
var ErrNoTargets, ErrTooManyTargets, ErrCallerBusy, ErrBusy, ErrRetryLater error
func New(db *dcache.DB, fetch FetchFunc, cfg Config, log *slog.Logger) *Runner
func (r *Runner) Submit(ctx context.Context, c *auth.Caller, callerClass string, targets []Target) (string, bool, error)
func (r *Runner) Cancel(id string) bool
func (r *Runner) Status(id string) (Job, []TargetStatus, bool)
func (r *Runner) Touch(id string)
func (r *Runner) Close()
// internal/metrics (task 03), on s.Prom
ObserveBatchJob(outcome string); ObserveBatchTarget(outcome string)
SetBatchQueueDepth(n int); SetBatchTargetsInflight(n int)
ObserveBatchJobSeconds(d time.Duration); ObserveBatchTargetSeconds(d time.Duration); ObserveBatchResultBytes(n int)
// internal/api (task 04 and before)
func (s *Server) fetchTarget(ctx context.Context, c *auth.Caller, t batch.Target) (batch.Result, error)
func (s *Server) caller(w http.ResponseWriter, r *http.Request) *auth.Caller // nil after writing 401/403
func (s *Server) callerClass(c *auth.Caller) string
func (s *Server) logErr(c *auth.Caller, err error) any
// requestDeadline (210 s), maxReqBody (64 KiB), writeErr, writeJSON
```
## Interfaces added here
```go
// in Server (after pdfPoolOnce):
Batch *batch.Runner // nil: the /v1/batch endpoints (task 06) answer 503 "batch disabled"
// NewBatchRunner builds the runner that serve (task 07) and the tests use: its FetchFunc is
// batchFetch and its Hooks feed s.Prom. cfg.Hooks is overwritten.
func (s *Server) NewBatchRunner(db *dcache.DB, cfg batch.Config) *batch.Runner
func (s *Server) batchFetch(ctx context.Context, c *auth.Caller, t batch.Target) (batch.Result, error)
func batchOutcome(err error) string
```
## Rules
1. **`batchFetch`**, the runner's `FetchFunc`:
- `context.WithTimeout(ctx, requestDeadline)`, mark the context as a batch fetch (an unexported
context key), and call `s.fetchTarget`.
- Map its error:
- a `*fetch.Error` with status `429` or `503` → `fmt.Errorf("%s: %w", fe.Msg, batch.ErrRetryLater)`,
so the runner retries it;
- any other `*fetch.Error` → `&batch.TargetError{Status, Msg, Hint}`, with `502` rewritten to
`500 internal error` as `writeFetchErr` does, and `s.blockN.Add(1)` for a `403`;
- anything else → `&batch.TargetError{Status: 500, Msg: "internal error"}`, logged `Error`
with `s.logErr(c, err)`.
2. **Hooks** (in `NewBatchRunner`):
- `Job` → `ObserveBatchJob`;
- `Target(err, d, n)` → `ObserveBatchTarget(batchOutcome(err))`, `ObserveBatchTargetSeconds(d)`, and
`ObserveBatchResultBytes(n)` when `err == nil`;
- `JobSeconds`, `Depth`, `Inflight` → the matching `s.Prom` methods.
- `batchOutcome(err)` is:
- `ok` for nil;
- `busy` for an error wrapping `ErrRetryLater` (the runner gave up retrying);
- otherwise, by `TargetError.Status`: `403` `blocked`, `429` `busy`, `504` `timeout`,
`415` `unsupported`, `422` whose message starts `document too long` `too_long`, other
4xx `upstream_4xx`, 5xx `upstream_5xx`; any other error `upstream_5xx`.
3. **The PDF pool under a batch:** in `extractPDF`, the pool's caller argument is `c.Name+"/batch"`
when the context carries the batch mark, else `c.Name` as today. A batch extraction then
never takes the caller's own interactive slot. The single-flight key is unchanged, so a batch
and an interactive extraction of the same PDF still share one child.
4. Log lines added here carry the caller and the job id, **never a target URL or hostname**. The
per-target `fetch` line inside `fetchTarget` is unchanged.
## Facts
- The given api test builds the server through `s.NewBatchRunner` over an in-memory `dcache`
(`newBatchServer`). Targets are local `httptest` servers, or `internal.test`, which the test's
`LookupIP` resolves to `10.0.0.5`.
- `TestBatchFetchErrors` fills the 8-slot global gate with `s.acquire` so the next fetch is
refused busy after `acquire`'s 5 s wait.
- `batchrun_test.go` also holds the helpers task 06's test uses; it is a given file like the others.
## Steps
- [ ] **1.** `git switch p6`; copy the two given tests. **2. See them fail.**
- [ ] **3.** `PreCheck` (verbatim). **4.** `batch.go`. **5.** The `Batch` field and the pool name.
- [ ] **6.** `gofmt -w`; `go test -race -count=3 ./internal/api/ ./internal/guard/` → `ok`.
**7.** `make gate`.
- [ ] **8.** Row `p6/05-batch-wiring`; commit.
```sh
git add internal/api internal/guard/guard.go internal/guard/precheck_test.go docs/implementer-log.md
git commit -F .state/commit-msg.txt
```
## Done when
- The given tests and every existing api and guard test pass under `-race -count=3`; `make gate`
ok; the given files are byte-identical.
-70
View File
@@ -1,70 +0,0 @@
# 05-cli-docs — `sift batch`, skill, README, config
Expose the batch endpoint from the CLI and document it for agents. No Hermes plugin — the CLI is
the agent surface, same rationale as `sift fetch`/`sift search`.
## CLI
Add a `sift batch` subcommand with three verbs. It reads `url` and `key_file` from `client.toml`
exactly as `fetch`/`search` do, and never puts a key on the command line.
```
sift batch submit <targets.json> # '-' reads stdin; prints the job_id (and reused) as JSON
sift batch status <job_id> # prints status + per-target results as JSON
sift batch cancel <job_id> # prints the cancelled status as JSON
```
`submit` takes the same JSON as `POST /v1/batch` (a `{"targets":[…]}` object) and prints
`{"job_id": …, "reused": …}`. `status` prints the full `GET /v1/batch/{id}` body. All three
propagate the server's error envelope and exit non-zero on a non-2xx.
The dispatch/validation is a pure function pinned by the given test:
```go
func parseBatchArgs(args []string) (verb, arg string, err error)
```
`verb` is `submit`/`status`/`cancel`; `arg` is the targets file (submit, `-` for stdin) or the
`job_id` (status/cancel). No verb, a missing arg, an extra arg, or an unknown verb is an error.
## Config
Add a `[batch]` block to `internal/config` (`File`), with:
```toml
[batch]
targets_per_job = 32
jobs_per_caller = 1
jobs_total = 8
ttl = "24h"
max_bytes = "64MiB"
```
The runner and API read these instead of hard-coding; the defaults match the settled caps. The
body-cap knob for `/v1/batch` (256 KiB) is a constant, not config.
## Docs
- `README.md`: add a short "Batch" section (the three endpoints, the serial/background model, caps,
idempotency, lease/TTL semantics, restricted-caller redaction, and the `sift_batch_*` metrics).
- `skill/SKILL.md`: add a batch section telling agents to use `sift batch submit` for a list of
targets rather than a loop of `sift fetch`, and to `sift batch status` to collect.
- `dist/sift.toml.example`: add the `[batch]` block with its defaults and a comment.
## Rules
1. The CLI reuses the existing client plumbing (`client.toml`, bearer header, JSON envelope
handling) — no new HTTP client.
2. `submit` validates its input locally (a non-empty `targets` array) before calling the server.
3. `status`/`cancel` require exactly one `job_id` argument.
4. Keep each verb a small function; do not grow `main.go` — add `cmd/sift/batchcmd.go`.
## Given
Copy `_files/cmd_sift/batch_test.go` → `cmd/sift/batch_test.go`. It pins `parseBatchArgs` — the
valid verb+arg pairs (including `-` as a stdin target file) and the error cases (no verb, missing
arg, extra arg, unknown verb).
## Gate
`make gate` green; `go test ./cmd/sift/` green under `-race`; `go test ./internal/config/` green.
+112
View File
@@ -0,0 +1,112 @@
# P6 task 06: the `/v1/batch` endpoints
**Branch:** `p6` (`git switch p6`; `git status --short` must be empty, otherwise stop)
**Commit subject:** `api: /v1/batch — submit, status summary, one result, cancel`
## Goal
The HTTP surface over `s.Batch` (task 05): submit, a status summary, one target's result, and
cancel, with the submit-time guard pre-check, the caps as HTTP statuses, and ownership.
## Files
- Copy: `_files/api/batch_test.go` → `internal/api/batch_test.go`
- Modify: `internal/api/batch.go` (the four handlers), `internal/api/api.go` (only the routes in
`New` and `ServeHTTP`, rule 2), `docs/implementer-log.md`
## What already exists
```go
// internal/batch (task 02)
type Target struct{ URL string; Options Options }
type Options struct{ Mode, Question, Pages, Grep, JQ string; Context *bool; Recover bool }
type Result struct{ Content string; Meta map[string]any }
type Job struct{ ID, Caller, Status, Error string; TargetsTotal, TargetsDone int; CreatedAt int64 }
type TargetStatus struct{ Idx int; URL, Status string; Result *Result; Error, Hint string }
var ErrNoTargets, ErrTooManyTargets, ErrCallerBusy, ErrBusy error
func (r *Runner) Submit(ctx context.Context, c *auth.Caller, callerClass string, targets []Target) (string, bool, 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
// internal/guard (task 05)
func (g *Guard) PreCheck(raw string) error // "blocked: …" errors
// internal/api
Batch *batch.Runner // in Server (task 05); nil → 503 "batch disabled"
func (s *Server) caller(w http.ResponseWriter, r *http.Request) *auth.Caller // nil after writing 401/403
func (s *Server) callerClass(c *auth.Caller) string
// maxReqBody (64 KiB), writeErr(w, status, msg, hint), writeJSON(w, status, v), s.Prom.Observe(path, d, code)
```
Add `const batchMaxBody = 256 << 10`.
## Rules
1. **Routes** (in `New`): `POST /v1/batch`, `GET /v1/batch/{id}`, `GET /v1/batch/{id}/{idx}`,
`DELETE /v1/batch/{id}`. Each starts with `s.caller(w, r)`.
2. **`ServeHTTP`:**
- The body cap is `batchMaxBody` for `/v1/batch` and `maxReqBody` for every other path. The cap
is applied once, here: a handler cannot raise a cap already set.
- Batch paths are observed with a fixed label: `/v1/batch` or `/v1/batch/{id}`. A job id must
never become a label value.
3. **Submit** (`POST /v1/batch`), checked in this order:
- The body is `{"targets": [{"url", "mode", "question", "pages", "grep", "jq", "context",
"recover"}, …]}`, each target flat with `/v1/fetch`'s field names. Bad JSON, a missing or
empty list, an empty `url`, or a `mode` other than empty, `clean`, `raw` or `digest` →
`400 invalid batch: …`.
- `s.Guard.PreCheck(url)` for every target. The first failure → `422` with the message
`blocked: target <idx>: <reason>`, and no job is created.
- `s.Batch.Submit(r.Context(), c, s.callerClass(c), targets)`:
- `ErrTooManyTargets` → `422 too many targets (max 32)` (the configured cap);
- `ErrCallerBusy` → `429 busy: a batch of yours is still running`, `Retry-After: 30`;
- `ErrBusy` → `429 batch jobs busy`, `Retry-After: 60`;
- `ErrNoTargets` → `400`.
- `202 {"success": true, "job_id", "status", "targets": n, "reused"}`. `status` is `"queued"`
for a new job, and the job's current status for a reused one.
4. **Ownership:** a job whose `Caller` is not `c.Name`, or an unknown id, is `404 not found` for
every GET and DELETE. It is never `403`: a caller cannot probe for another caller's jobs.
5. **Status** (`GET /v1/batch/{id}`): after the ownership check, `s.Batch.Touch(id)`, then
`200 {"job_id", "status", "error" (only when set), "targets_total", "targets_done",
"created_at" (RFC 3339, UTC), "targets": [{"idx", "url", "status", "error"?, "hint"?}]}`.
- This is a summary: **no results**, because 32 inline results can be 6 MB, too much for one
agent tool call.
- With `?results=1`, each `done` target also carries `"result"`: the envelope, i.e. `Meta`
plus `"content"`.
6. **One result** (`GET /v1/batch/{id}/{idx}`): after ownership and `Touch`, `200 {"job_id", "idx",
"url", "status", "result"?, "error"?, "hint"?}`. An `idx` that is not a number in
`0..targets_total-1` → `404`. **Cancel** (`DELETE`): after ownership, a terminal job →
`409 job already finished`; otherwise `Cancel` → `200 {"success": true, "status": "cancelled"}`.
7. **Logging:** each batch handler logs one line: `batch` with `caller`, `job_id`, `targets` and
`status` (or `reused`). **Never a target URL or hostname, for any caller**, and never the
`idem_key`. The per-target `fetch` line inside `fetchTarget` is unchanged: it is redacted
for restricted callers exactly as for `/v1/fetch`.
## Facts
- `newBatchServer`, `pageServer` and `fakeDNS` come from `batchrun_test.go` (task 05). Targets
are local `httptest` servers, some held on a channel, or `internal.test` (resolves to `10.0.0.5`).
- `TestBatchBodyCap` posts 32 targets with questions, about 66 KiB: over `maxReqBody`, under
`batchMaxBody`.
## Steps
- [ ] **1.** `git switch p6`; copy the given test. **2. See it fail.**
- [ ] **3.** The handlers. **4.** Routes and `ServeHTTP`.
- [ ] **5.** `gofmt -w`; `go test -race -count=3 ./internal/api/` → `ok`. **6.** `make gate`.
- [ ] **7.** Row `p6/06-api-endpoints`; commit.
```sh
git add internal/api docs/implementer-log.md
git commit -F .state/commit-msg.txt
```
## Done when
- The given test and every existing api test pass under `-race -count=3`; `make gate` ok; the
given file is byte-identical.
## Stop and report if
- Any given test needs a URL that is not local, or a change to an existing test.
+101
View File
@@ -0,0 +1,101 @@
# P6 task 07: the `[batch]` config and `serve` wiring
**Branch:** `p6` (`git switch p6`; `git status --short` must be empty, otherwise stop)
**Commit subject:** `serve: [batch] config; open the store, reap jobs killed by a restart, start the runner`
## Goal
`sift serve` runs the batch runner from `[batch]` config. On startup it marks jobs that a restart
killed as failed (they are never resumed). On shutdown it stops the runner before closing the store.
## Files
- Copy: `_files/config/batch_test.go` → `internal/config/batch_test.go`
- Modify: `internal/config/config.go` (the `Batch` type and field, `ParseSize`),
`cmd/sift/serve.go` (rule 3), `dist/sift.toml.example` (rule 4), `docs/implementer-log.md`
## What already exists
```go
// internal/batch (task 02)
type Config struct{ TTL time.Duration; TargetsPerJob, JobsPerCaller, JobsTotal, Workers int; MaxBytes int64; RetryWait, RetryFor time.Duration; Hooks Hooks }
func (r *Runner) Close()
// internal/api (task 05)
func (s *Server) NewBatchRunner(db *dcache.DB, cfg batch.Config) *batch.Runner
Batch *batch.Runner // in Server
// s.Prom.ObserveBatchJob(outcome string) (task 03)
// internal/dcache (task 01)
func (d *DB) MarkInFlightFailed(reason string, now int64) int
// internal/config
func ParseDuration(s string, def time.Duration) time.Duration
```
## Interfaces added here
```go
package config
type Batch struct {
Disabled bool `toml:"disabled"`
TargetsPerJob int `toml:"targets_per_job"`
JobsPerCaller int `toml:"jobs_per_caller"`
JobsTotal int `toml:"jobs_total"`
Workers int `toml:"workers"`
TTL string `toml:"ttl"`
MaxBytes string `toml:"max_bytes"`
}
// in File, after Search:
Batch Batch `toml:"batch"`
// ParseSize parses a positive byte count: a plain integer, or one with KiB, MiB or GiB.
// Anything else, zero or negative returns def.
func ParseSize(s string, def int64) int64
```
## Rules
1. Batch is **on by default**: no `[batch]` block means the runner with its own defaults.
`disabled = true` turns it off, and the endpoints then answer 503.
2. `ParseSize` trims spaces and accepts only the binary suffixes `KiB`, `MiB`, `GiB`, or none (bytes).
3. **`serve.go`:**
- Open the `dcache` store when batch is on, as well as in today's cases: same path, same call.
If the store cannot be opened, log `Warn` and run without batch.
- Then, **before** building the runner: `n := cache.MarkInFlightFailed("restart", now)`, and call
`srv.Prom.ObserveBatchJob("restart")` `n` times. Log `batch reaped` with `n` if `n > 0`.
- `srv.Batch = srv.NewBatchRunner(cache, batch.Config{...})`, with the config fields: `TTL` via
`ParseDuration(…, 0)` and `MaxBytes` via `ParseSize(…, 0)`. The runner's defaults fill the zeros.
- On shutdown (both branches of the final `select`), `srv.Batch.Close()` before the process
exits, when it is non-nil.
4. **`dist/sift.toml.example`:** a commented `[batch]` block with the defaults, in the style of the
file:
```toml
# [batch]
# disabled = false
# targets_per_job = 32
# jobs_per_caller = 1
# jobs_total = 8
# workers = 1 # at most 2: each worker fetches one target at a time
# ttl = "24h" # a job's results live this long after its last read or write
# max_bytes = "64MiB" # stored results; the oldest finished jobs go first
```
## Steps
- [ ] **1.** `git switch p6`; copy the given test. **2. See it fail.**
- [ ] **3.** `config.go`. **4.** `serve.go`. **5.** The example config.
- [ ] **6.** `gofmt -w`; `go test -race -count=3 ./internal/config/ ./cmd/sift/` → `ok`. **7.** `make gate`.
- [ ] **8.** Row `p6/07-serve-config`; commit.
```sh
git add internal/config cmd/sift/serve.go dist/sift.toml.example docs/implementer-log.md
git commit -F .state/commit-msg.txt
```
## Done when
- The given test and every existing config and cmd test pass under `-race -count=3`; `make gate`
ok; the given file is byte-identical.
+89
View File
@@ -0,0 +1,89 @@
# P6 task 08: `sift batch`, README and skill
**Branch:** `p6` (`git switch p6`; `git status --short` must be empty, otherwise stop)
**Commit subject:** `cli: sift batch submit|status|get|cancel; README and skill`
## Goal
Agents reach the batch endpoint through the CLI, as they do `fetch` and `search`. There is no
Hermes plugin.
## Files
- Copy: `_files/cmd_sift/batch_test.go` → `cmd/sift/batch_test.go`
- Create: `cmd/sift/batchcmd.go` (`batchMain`, `parseBatchArgs`, one small function per verb)
- Modify: `cmd/sift/main.go` (one `case "batch":` line and the usage text), `README.md`,
`skill/SKILL.md`, `docs/implementer-log.md`
## What already exists
- `cmd/sift`: `loadClient() (base, key string, err error)` reads `client.toml`; `searchcmd.go` is the
pattern for a verb (build the request with the bearer header, decode the JSON envelope, exit
non-zero on a non-2xx). Never put a key on the command line.
- The endpoints (task 06):
- `POST /v1/batch` `{"targets":[…]}` → `202 {"success","job_id","status","targets","reused"}`;
- `GET /v1/batch/{id}` (summary; `?results=1` inlines results);
- `GET /v1/batch/{id}/{idx}` (one result);
- `DELETE /v1/batch/{id}`.
## Interfaces added here
```go
// parseBatchArgs validates a `sift batch` command line (without "batch").
func parseBatchArgs(args []string) (verb string, rest []string, err error)
```
```
sift batch submit <targets.json|-> # prints {"job_id", "status", "reused"}
sift batch status [--results] <job_id> # prints the status body (a summary unless --results)
sift batch get <job_id> <idx> # prints one target's result, as `sift fetch --json` would
sift batch cancel <job_id> # prints {"success", "status"}
```
## Rules
1. `parseBatchArgs` accepts exactly the forms above:
- `submit` with one argument (`-` means stdin);
- `status` with a job id, optionally preceded by `--results`;
- `get` with a job id and an `idx` that is a non-negative integer;
- `cancel` with a job id.
Anything else (no verb, a missing or extra argument, an unknown verb, a bad `idx`) is an
error. `rest` is the arguments after the verb, unchanged.
2. `submit` reads the file and checks it locally before calling the server: it must be a JSON
object with a non-empty `targets` array. It may also be a bare array of targets or of URL
strings; wrap either into `{"targets": […]}`, turning each string into `{"url": s}`.
3. Every verb prints the server's JSON on stdout, and on a non-2xx prints the error envelope to
stderr and exits with `exitFor(status, msg)`, as `fetch` does.
4. **`README.md`**, a short "Batch" section:
- the four endpoints;
- the model: serial per job, one worker by default, round-robin across jobs;
- the caps (32 targets, 1 live job per caller, 8 in all) and the 256 KiB body;
- idempotency: the same list from the same caller returns the same job while it is live or
has results;
- the lease/TTL and restart semantics;
- the summary-vs-results reads;
- restricted-caller redaction;
- the `sift_batch_*` metrics.
5. **`skill/SKILL.md`**, a short section:
- for several URLs you will read later, `sift batch submit` them once, rather than a loop of
`sift fetch`;
- poll with `sift batch status`, and read each result with `sift batch get <job> <idx>`;
- don't use `--results` unless you need every result at once (it can be large);
- a batch is slow on purpose: for one URL you need now, use `sift fetch`.
## Steps
- [ ] **1.** `git switch p6`; copy the given test. **2. See it fail.**
- [ ] **3.** `batchcmd.go`. **4.** `main.go`. **5.** README and skill.
- [ ] **6.** `gofmt -w`; `go test -race -count=3 ./cmd/sift/` → `ok`. **7.** `make gate`.
- [ ] **8.** Row `p6/08-cli-docs`; commit.
```sh
git add cmd/sift README.md skill/SKILL.md docs/implementer-log.md
git commit -F .state/commit-msg.txt
```
## Done when
- The given test and every existing cmd test pass under `-race -count=3`; `make gate` ok; the
given file is byte-identical.
+36 -24
View File
@@ -8,36 +8,48 @@ holding a request open per target submits them as one batch, gets a job ID back
polls for per-target results later. A batch is **deliberate and slow** — serial, one target at a
time — so it is the tool for research threads, not for interactive fetches.
Decisions (settled with the owner, 2026-09-26):
Decisions (settled with the owner, 2026-09-26; revised after review the same evening, see
`docs/P6.md` §14):
- **Serial.** One target drains at a time per job, in the background. Each target still goes
through the normal `acquire` gate (8 global / 2 per-host); on `429 busy` the worker waits and
retries rather than failing, because nobody is holding a socket open. Interactive requests win
by construction: they are served by HTTP handlers that grab slots immediately.
- **Caps.** targets/job 32; live jobs/caller 1; live jobs total 8; per-target result bounded by
the output cap. All instrumented (`sift_batch_*`) so the numbers can be re-sized empirically.
- **Idempotency in v1.** A job has a random opaque `id` plus a derived `idem_key`; resubmitting
the same (caller, targets) while the job is live returns the existing job instead of a new one.
- **Storage.** `batch_job` + `batch_target` tables in the existing `dcache` SQLite.
- **Retention / TTL.** A job carries a single sliding **lease**, bumped on submit, on every target
completion, and on every status/collect read. All targets therefore share one expiry, and a slow
consumer's reads keep a finished batch alive. GC deletes a job (and its targets) once its lease
is older than the TTL. Batch storage is a coherent whole: it is bounded by the lease GC plus a
separate byte budget (oldest-lease jobs evicted whole), not by the content cache's LRU.
- **Restart.** Completed results survive restart and stay readable until the lease lapses.
`queued`/`running` jobs are marked `failed` (reason `restart`) on startup; never resumed.
- **Restricted callers.** Compartmentalized-access systems: their usage is redacted as much as
possible (host, error reason, error detail — the existing `logHost`/`logReason`/`logErr`
machinery), and no other caller may read their jobs or results. The worker passes the caller
through, so this happens automatically; batch log lines never carry a target URL or hostname.
- **Serial per job, few workers.** One worker by default (at most 2) drains all live jobs,
round-robin, one target at a time. The batch therefore holds at most 1–2 of the 8 global fetch
slots, so interactive requests are never starved. A target refused busy (fetch gate or PDF
pool) is retried after a pause, for up to 10 minutes, then failed.
- **The caller travels with the job.** The runner keeps each job's `*auth.Caller` in memory and
hands it to every fetch, so per-caller gates, the PDF pool, the digest cache class and
restricted-caller redaction all apply. Nothing about the caller is persisted beyond its name
and class: jobs never resume after a restart.
- **Caps.** targets/job 32; live jobs/caller 1; live jobs total 8; body 256 KiB. All instrumented
(`sift_batch_*`).
- **Idempotency in v1.** The same (caller, target list) returns the existing job while it is live or
has results (`queued`/`running`/`done`/`partial`) and its lease holds. A failed or cancelled job
is not reused.
- **Storage.** `batch_job` + `batch_target` tables in the existing `dcache` SQLite, with its own
byte budget (64 MiB), outside the content cache's LRU.
- **Retention / TTL.** One sliding lease per job, bumped on submit, on every target write, and on
every owner read (the API calls `Touch`; store reads never write). GC deletes a job when its
lease is older than the TTL (24 h).
- **Reads.** `GET /v1/batch/{id}` is a summary (statuses, errors). A result is read one at a time
(`GET /v1/batch/{id}/{idx}`), or all at once with `?results=1`. 32 inline results can be 6 MB.
- **Restart.** `queued`/`running` jobs are marked `failed` (`restart`) on startup; never resumed.
- **Restricted callers.** Their fetches go through the same `fetchTarget` with the real caller,
so the existing redaction applies. No caller can see another's job (404, never 403). Batch
log lines never carry a target URL or hostname.
| Task | Package | Given |
|---|---|---|
| 01-batchstore | `internal/dcache` | `dcache/batch_test.go` |
| 02-runner | `internal/batch` | `batch/runner_test.go` |
| 03-api | `internal/api` | `api/batch_test.go` |
| 04-metrics | `internal/metrics` | `metrics/batch_test.go` |
| 05-cli-docs | `cmd/sift`, `README.md`, `skill/SKILL.md`, config | `cmd/sift/batch_test.go` |
| 03-metrics | `internal/metrics` | `metrics/batch_test.go` |
| 04-fetchtarget | `internal/api` (the one refactor of working code) | `api/fetchtarget_test.go` |
| 05-batch-wiring | `internal/api` (+ `guard.PreCheck`, given verbatim) | `api/batchrun_test.go`, `guard/precheck_test.go` |
| 06-api-endpoints | `internal/api` | `api/batch_test.go` |
| 07-serve-config | `internal/config`, `cmd/sift/serve.go` | `config/batch_test.go` |
| 08-cli-docs | `cmd/sift`, README, skill | `cmd_sift/batch_test.go` |
Given files live under `_files/<package>/` (`cmd_sift` is `cmd/sift`); each task says where to
copy its file. Every given test was compiled against a panic-only skeleton of the whole plan and
fails there on `panic: todo`, not on a compile error.
**How this plan was made:** design worked through with the owner from `docs/PLAN.md` §13 (future
work) on top of the P5 pool (`internal/pdfpool`), cache (`internal/dcache`), and gate
+202 -176
View File
@@ -2,253 +2,279 @@ package api
import (
"encoding/json"
"fmt"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"net/netip"
"os"
"path/filepath"
"strings"
"testing"
"time"
"git.wntrmute.dev/kyle/sift/internal/auth"
"git.wntrmute.dev/kyle/sift/internal/batch"
"git.wntrmute.dev/kyle/sift/internal/dcache"
"git.wntrmute.dev/kyle/sift/internal/fetch"
"git.wntrmute.dev/kyle/sift/internal/guard"
)
// newBatchServer builds a Server with the batch runner wired over an in-memory
// dcache, exactly as serve does in production. callers are the caller names to
// provision (each with a bearer key).
func newBatchServer(t *testing.T, g *guard.Guard, callers ...string) (*Server, map[string]string) {
t.Helper()
dir := t.TempDir()
keysPath := filepath.Join(dir, "keys.toml")
keys := map[string]string{}
for _, name := range callers {
out := filepath.Join(dir, name)
if err := auth.Add(keysPath, name, "hermes-fleet", out); err != nil {
t.Fatal(err)
}
b, err := os.ReadFile(out)
if err != nil {
t.Fatal(err)
}
keys[name] = strings.TrimSpace(string(b))
}
store, err := auth.Load(keysPath)
if err != nil {
t.Fatal(err)
}
if g == nil {
g = &guard.Guard{}
}
f := &fetch.Fetcher{Guard: g, UA: "sift-test"}
s := New("test", g, f, store, slog.New(slog.NewTextHandler(io.Discard, nil)))
db, err := dcache.Open("", 1<<20, slog.New(slog.NewTextHandler(io.Discard, nil)))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = db.Close() })
s.Batch = batch.New(db, s.fetchTarget, batch.Config{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
return s, keys
}
// Every test fetches only from local httptest servers or internal.test (fake DNS): no test
// touches the network. newBatchServer, pageServer and fakeDNS come from batchrun_test.go.
func submitBatch(t *testing.T, s *Server, key string, targets []map[string]string) *httptest.ResponseRecorder {
func batchReq(t *testing.T, s *Server, method, path, key string, body any) *httptest.ResponseRecorder {
t.Helper()
return postJSON(t, s, "/v1/batch", key, map[string]any{"targets": targets})
}
func getBatch(t *testing.T, s *Server, key, id string) *httptest.ResponseRecorder {
t.Helper()
req := httptest.NewRequest(http.MethodGet, "/v1/batch/"+id, nil)
var rd io.Reader
if body != nil {
b, err := json.Marshal(body)
if err != nil {
t.Fatal(err)
}
rd = strings.NewReader(string(b))
}
req := httptest.NewRequest(method, path, rd)
req.Header.Set("Authorization", "Bearer "+key)
if body != nil {
req.Header.Set("Content-Type", "application/json")
}
rec := httptest.NewRecorder()
s.ServeHTTP(rec, req)
return rec
}
func batchStatus(t *testing.T, s *Server, key, id string) map[string]any {
func batchDecode(t *testing.T, rec *httptest.ResponseRecorder) map[string]any {
t.Helper()
rec := getBatch(t, s, key, id)
var env map[string]any
if err := json.Unmarshal(rec.Body.Bytes(), &env); err != nil {
t.Fatalf("status decode: %v (%s)", err, rec.Body.String())
var m map[string]any
if err := json.Unmarshal(rec.Body.Bytes(), &m); err != nil {
t.Fatalf("decode: %v (%d %s)", err, rec.Code, rec.Body.String())
}
return env
return m
}
// waitBatchTerminal polls until the job reaches a terminal status.
func waitBatchTerminal(t *testing.T, s *Server, key, id string) map[string]any {
func submit(t *testing.T, s *Server, key string, targets ...map[string]any) map[string]any {
t.Helper()
deadline := time.Now().Add(3 * time.Second)
rec := batchReq(t, s, http.MethodPost, "/v1/batch", key, map[string]any{"targets": targets})
if rec.Code != 202 {
t.Fatalf("submit: %d %s", rec.Code, rec.Body.String())
}
return batchDecode(t, rec)
}
func waitTerminal(t *testing.T, s *Server, key, id string) map[string]any {
t.Helper()
deadline := time.Now().Add(10 * time.Second)
for time.Now().Before(deadline) {
env := batchStatus(t, s, key, id)
if st, _ := env["status"].(string); st == "done" || st == "partial" || st == "failed" || st == "cancelled" {
env := batchDecode(t, batchReq(t, s, http.MethodGet, "/v1/batch/"+id, key, nil))
switch env["status"] {
case "done", "partial", "failed", "cancelled":
return env
}
time.Sleep(time.Millisecond)
time.Sleep(5 * time.Millisecond)
}
t.Fatal("batch never reached a terminal status")
return nil
}
func TestBatchSubmitAndCollect(t *testing.T) {
g := &guard.Guard{AllowPrefixes: []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")}}
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "text/html; charset=utf-8")
_, _ = w.Write([]byte(`<html><head><title>T</title></head><body><article><p>hello batch</p></article></body></html>`))
}))
defer ts.Close()
s, keys := newBatchServer(t, g, "test")
rec := submitBatch(t, s, keys["test"], []map[string]string{{"url": ts.URL}, {"url": ts.URL + "/2"}})
if rec.Code != 202 {
t.Fatalf("submit %d %s", rec.Code, rec.Body.String())
}
var sub map[string]any
if err := json.Unmarshal(rec.Body.Bytes(), &sub); err != nil {
t.Fatal(err)
}
if sub["success"] != true {
t.Fatalf("success %v", sub["success"])
}
ts := pageServer(t, nil)
s, keys := newBatchServer(t, "test")
sub := submit(t, s, keys["test"], map[string]any{"url": ts.URL + "/one"}, map[string]any{"url": ts.URL + "/two"})
id, _ := sub["job_id"].(string)
if id == "" {
t.Fatal("missing job_id")
if sub["success"] != true || id == "" || sub["reused"] != false || sub["targets"] != float64(2) || sub["status"] != "queued" {
t.Fatalf("submit body %v", sub)
}
if reused, _ := sub["reused"].(bool); reused {
t.Fatal("first submit must not be reused")
// The status body is a summary: statuses, no results.
env := waitTerminal(t, s, keys["test"], id)
if env["status"] != "done" || env["targets_total"] != float64(2) || env["targets_done"] != float64(2) {
t.Fatalf("status %v", env)
}
env := waitBatchTerminal(t, s, keys["test"], id)
if env["status"] != "done" {
t.Fatalf("status %v", env["status"])
if _, err := time.Parse(time.RFC3339, fmt.Sprint(env["created_at"])); err != nil {
t.Fatalf("created_at %v is not RFC 3339", env["created_at"])
}
targets, _ := env["targets"].([]any)
targets := env["targets"].([]any)
if len(targets) != 2 {
t.Fatalf("targets %d", len(targets))
}
first := targets[0].(map[string]any)
if first["status"] != "done" {
t.Fatalf("target0 %v", first["status"])
for i, x := range targets {
tg := x.(map[string]any)
if tg["idx"] != float64(i) || tg["status"] != "done" {
t.Fatalf("target %d: %v", i, tg)
}
res, _ := first["result"].(map[string]any)
if res == nil || !strings.Contains(res["content"].(string), "hello batch") {
t.Fatalf("target0 result %v", res)
if _, has := tg["result"]; has {
t.Fatalf("summary must not carry results: target %d has one", i)
}
}
// One target's result: the /v1/fetch envelope.
rec := batchReq(t, s, http.MethodGet, "/v1/batch/"+id+"/1", keys["test"], nil)
if rec.Code != 200 {
t.Fatalf("GET idx: %d %s", rec.Code, rec.Body.String())
}
one := batchDecode(t, rec)
res, _ := one["result"].(map[string]any)
if one["idx"] != float64(1) || one["status"] != "done" || res == nil {
t.Fatalf("GET idx body %v", one)
}
if c, _ := res["content"].(string); !strings.Contains(c, "hello batch /two") {
t.Fatalf("result content %q", c)
}
if res["untrusted"] != true || res["mode"] != "clean" {
t.Fatalf("result is not the /v1/fetch envelope: %v", res)
}
if src, _ := res["source"].(map[string]any); src == nil || src["final_url"] != ts.URL+"/two" {
t.Fatalf("result source %v", res["source"])
}
// ?results=1 inlines every result.
full := batchDecode(t, batchReq(t, s, http.MethodGet, "/v1/batch/"+id+"?results=1", keys["test"], nil))
for i, x := range full["targets"].([]any) {
if _, has := x.(map[string]any)["result"]; !has {
t.Fatalf("results=1: target %d has no result", i)
}
}
if got := batchReq(t, s, http.MethodGet, "/v1/batch/"+id+"/2", keys["test"], nil).Code; got != 404 {
t.Fatalf("GET idx out of range: %d, want 404", got)
}
if got := batchReq(t, s, http.MethodGet, "/v1/batch/"+id+"/x", keys["test"], nil).Code; got != 404 {
t.Fatalf("GET idx not a number: %d, want 404", got)
}
}
func TestBatchIdempotentResubmit(t *testing.T) {
g := &guard.Guard{AllowPrefixes: []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")}}
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "text/plain")
_, _ = w.Write([]byte("ok"))
}))
defer ts.Close()
s, keys := newBatchServer(t, g, "test")
targets := []map[string]string{{"url": ts.URL}}
first := submitBatch(t, s, keys["test"], targets)
second := submitBatch(t, s, keys["test"], targets)
var a, b map[string]any
_ = json.Unmarshal(first.Body.Bytes(), &a)
_ = json.Unmarshal(second.Body.Bytes(), &b)
if a["job_id"] != b["job_id"] {
t.Fatalf("ids differ: %v vs %v", a["job_id"], b["job_id"])
}
if reused, _ := b["reused"].(bool); !reused {
t.Fatal("resubmit must be reused=true")
ts := pageServer(t, nil)
s, keys := newBatchServer(t, "test")
tg := map[string]any{"url": ts.URL, "mode": "clean"}
a := submit(t, s, keys["test"], tg)
b := submit(t, s, keys["test"], tg)
if a["job_id"] != b["job_id"] || b["reused"] != true {
t.Fatalf("resubmit: %v then %v", a, b)
}
}
func TestBatchGuardRefusedUpFront(t *testing.T) {
s, keys := newBatchServer(t, nil, "test")
rec := submitBatch(t, s, keys["test"], []map[string]string{{"url": "http://example.com:8080/"}})
if rec.Code != 422 {
t.Fatalf("want 422, got %d %s", rec.Code, rec.Body.String())
func TestBatchRefusedUpFront(t *testing.T) {
s, keys := newBatchServer(t, "test")
cases := []struct {
name string
body any
code int
want string
}{
{"guard pre-check (port)", map[string]any{"targets": []any{map[string]any{"url": "http://127.0.0.1/"}, map[string]any{"url": "http://example.com:8080/"}}}, 422, "blocked"},
{"guard pre-check (private literal)", map[string]any{"targets": []any{map[string]any{"url": "http://10.0.0.5/"}}}, 422, "blocked"},
{"no targets", map[string]any{"targets": []any{}}, 400, ""},
{"not json", "nope", 400, ""},
{"empty url", map[string]any{"targets": []any{map[string]any{"url": ""}}}, 400, ""},
{"bad mode", map[string]any{"targets": []any{map[string]any{"url": "http://127.0.0.1/", "mode": "shout"}}}, 400, ""},
}
if !strings.Contains(rec.Body.String(), "blocked") {
t.Fatalf("body %s", rec.Body.String())
for _, c := range cases {
rec := batchReq(t, s, http.MethodPost, "/v1/batch", keys["test"], c.body)
if rec.Code != c.code || !strings.Contains(rec.Body.String(), c.want) {
t.Errorf("%s: %d %s, want %d containing %q", c.name, rec.Code, rec.Body.String(), c.code, c.want)
}
}
// The guard refusal names the offending index.
rec := batchReq(t, s, http.MethodPost, "/v1/batch", keys["test"], cases[0].body)
if !strings.Contains(rec.Body.String(), "1") {
t.Errorf("guard refusal does not name target 1: %s", rec.Body.String())
}
}
func TestBatchTooManyTargets(t *testing.T) {
s, keys := newBatchServer(t, nil, "test")
var targets []map[string]string
var many []any
for i := 0; i < 33; i++ {
targets = append(targets, map[string]string{"url": "https://example.com/"})
many = append(many, map[string]any{"url": fmt.Sprintf("http://127.0.0.1/%d", i)})
}
rec := submitBatch(t, s, keys["test"], targets)
if rec.Code != 422 {
t.Fatalf("want 422, got %d %s", rec.Code, rec.Body.String())
if rec := batchReq(t, s, http.MethodPost, "/v1/batch", keys["test"], map[string]any{"targets": many}); rec.Code != 422 {
t.Errorf("33 targets: %d, want 422", rec.Code)
}
}
// 32 targets, each with a 500-rune question, fit the batch body cap even though they are well
// over the 64 KiB cap of the other endpoints.
func TestBatchBodyCap(t *testing.T) {
hold := make(chan struct{})
defer close(hold)
ts := pageServer(t, hold)
s, keys := newBatchServer(t, "test")
q := strings.Repeat("q", 500)
var many []any
for i := 0; i < 32; i++ {
many = append(many, map[string]any{"url": fmt.Sprintf("%s/%d?pad=%s", ts.URL, i, strings.Repeat("p", 1500)), "question": q})
}
if rec := batchReq(t, s, http.MethodPost, "/v1/batch", keys["test"], map[string]any{"targets": many}); rec.Code != 202 {
t.Fatalf("32 targets with questions (~%d KiB): %d %s", 32*2100/1024, rec.Code, rec.Body.String())
}
}
func TestBatchPartialFailure(t *testing.T) {
g := &guard.Guard{AllowPrefixes: []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")}}
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "text/plain")
_, _ = w.Write([]byte("ok"))
}))
defer ts.Close()
s, keys := newBatchServer(t, g, "test")
// target 1 is a private address: passes the submit pre-check (valid URL) but
// is blocked by the connect-time guard, surfacing as a per-target failure.
rec := submitBatch(t, s, keys["test"], []map[string]string{{"url": ts.URL}, {"url": "http://10.0.0.5/"}})
if rec.Code != 202 {
t.Fatalf("submit %d %s", rec.Code, rec.Body.String())
}
var sub map[string]any
_ = json.Unmarshal(rec.Body.Bytes(), &sub)
id := sub["job_id"].(string)
env := waitBatchTerminal(t, s, keys["test"], id)
ts := pageServer(t, nil)
s, keys := newBatchServer(t, "test")
// Target 1 is a name that resolves to a private address. Names are not resolved at submit,
// so it passes the pre-check; the fetch-time guard blocks it: a per-target failure.
sub := submit(t, s, keys["test"], map[string]any{"url": ts.URL}, map[string]any{"url": "http://internal.test/"})
env := waitTerminal(t, s, keys["test"], sub["job_id"].(string))
if env["status"] != "partial" {
t.Fatalf("status %v", env["status"])
}
targets := env["targets"].([]any)
second := targets[1].(map[string]any)
second := env["targets"].([]any)[1].(map[string]any)
if second["status"] != "failed" {
t.Fatalf("target1 status %v", second["status"])
t.Fatalf("target 1: %v", second)
}
if err, _ := second["error"].(string); !strings.Contains(err, "blocked") {
t.Fatalf("target1 error %q", err)
if e, _ := second["error"].(string); !strings.Contains(e, "blocked") {
t.Fatalf("target 1 error %q", e)
}
}
func TestBatchOwnership404(t *testing.T) {
s, keys := newBatchServer(t, nil, "test", "other")
rec := submitBatch(t, s, keys["test"], []map[string]string{{"url": "https://example.com/"}})
if rec.Code != 202 {
t.Fatalf("submit %d", rec.Code)
hold := make(chan struct{})
defer close(hold)
ts := pageServer(t, hold)
s, keys := newBatchServer(t, "test", "other")
id := submit(t, s, keys["test"], map[string]any{"url": ts.URL})["job_id"].(string)
for _, p := range []string{"/v1/batch/" + id, "/v1/batch/" + id + "/0"} {
if got := batchReq(t, s, http.MethodGet, p, keys["other"], nil).Code; got != 404 {
t.Fatalf("other caller GET %s: %d (must not reveal the job)", p, got)
}
var sub map[string]any
_ = json.Unmarshal(rec.Body.Bytes(), &sub)
id := sub["job_id"].(string)
if got := getBatch(t, s, keys["other"], id); got.Code != 404 {
t.Fatalf("other caller GET: %d (must not reveal the job)", got.Code)
}
if got := getBatch(t, s, keys["test"], id); got.Code != 200 {
t.Fatalf("owner GET: %d", got.Code)
if got := batchReq(t, s, http.MethodDelete, "/v1/batch/"+id, keys["other"], nil).Code; got != 404 {
t.Fatalf("other caller DELETE: %d", got)
}
if got := batchReq(t, s, http.MethodGet, "/v1/batch/"+id, keys["test"], nil).Code; got != 200 {
t.Fatalf("owner GET: %d", got)
}
if got := batchReq(t, s, http.MethodGet, "/v1/batch/no-such-job", keys["test"], nil).Code; got != 404 {
t.Fatalf("unknown job: %d", got)
}
if got := batchReq(t, s, http.MethodGet, "/v1/batch/"+id, "not-a-key", nil).Code; got != 401 {
t.Fatalf("bad key: %d", got)
}
}
func TestBatchCancel(t *testing.T) {
g := &guard.Guard{AllowPrefixes: []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")}}
s, keys := newBatchServer(t, g, "test")
rec := submitBatch(t, s, keys["test"], []map[string]string{{"url": "https://example.com/"}})
var sub map[string]any
_ = json.Unmarshal(rec.Body.Bytes(), &sub)
id := sub["job_id"].(string)
req := httptest.NewRequest(http.MethodDelete, "/v1/batch/"+id, nil)
req.Header.Set("Authorization", "Bearer "+keys["test"])
drec := httptest.NewRecorder()
s.ServeHTTP(drec, req)
if drec.Code != 200 {
t.Fatalf("cancel %d %s", drec.Code, drec.Body.String())
hold := make(chan struct{})
ts := pageServer(t, hold)
s, keys := newBatchServer(t, "test")
id := submit(t, s, keys["test"], map[string]any{"url": ts.URL + "/a"}, map[string]any{"url": ts.URL + "/b"})["job_id"].(string)
rec := batchReq(t, s, http.MethodDelete, "/v1/batch/"+id, keys["test"], nil)
if rec.Code != 200 {
t.Fatalf("cancel: %d %s", rec.Code, rec.Body.String())
}
env := waitBatchTerminal(t, s, keys["test"], id)
if b := batchDecode(t, rec); b["success"] != true || b["status"] != "cancelled" {
t.Fatalf("cancel body %v", b)
}
close(hold)
env := waitTerminal(t, s, keys["test"], id)
if env["status"] != "cancelled" {
t.Fatalf("status %v", env["status"])
}
if tg := env["targets"].([]any)[1].(map[string]any); tg["status"] != "pending" {
t.Fatalf("target 1 after cancel: %v", tg)
}
if got := batchReq(t, s, http.MethodDelete, "/v1/batch/"+id, keys["test"], nil).Code; got != 409 {
t.Fatalf("cancelling a finished job: %d, want 409", got)
}
}
// The request metrics never get a job id as a label value.
func TestBatchMetricsPathLabel(t *testing.T) {
ts := pageServer(t, nil)
s, keys := newBatchServer(t, "test")
id := submit(t, s, keys["test"], map[string]any{"url": ts.URL})["job_id"].(string)
waitTerminal(t, s, keys["test"], id)
batchReq(t, s, http.MethodGet, "/v1/batch/"+id+"/0", keys["test"], nil)
if strings.Contains(promText(t, s), id) {
t.Fatal("a job id appears in /metrics")
}
}
+202
View File
@@ -0,0 +1,202 @@
package api
// P6 task 05: the runner wired to fetchTarget and the metrics, with no HTTP yet. The helpers
// here (newBatchServer, pageServer, fakeDNS) are shared with task 06's batch_test.go.
import (
"context"
"errors"
"fmt"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"net/netip"
"os"
"path/filepath"
"strings"
"testing"
"time"
"git.wntrmute.dev/kyle/sift/internal/auth"
"git.wntrmute.dev/kyle/sift/internal/batch"
"git.wntrmute.dev/kyle/sift/internal/dcache"
"git.wntrmute.dev/kyle/sift/internal/fetch"
"git.wntrmute.dev/kyle/sift/internal/guard"
)
var loopback = []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")}
// fakeDNS resolves internal.test to a private address; every other name fails to resolve.
func fakeDNS(_ context.Context, host string) ([]netip.Addr, error) {
if host == "internal.test" {
return []netip.Addr{netip.MustParseAddr("10.0.0.5")}, nil
}
return nil, fmt.Errorf("no such host %s", host)
}
// newBatchServer builds a Server with the batch runner wired over an in-memory dcache through
// s.NewBatchRunner, exactly as serve does. callers are the names to provision (each gets a key).
func newBatchServer(t *testing.T, callers ...string) (*Server, map[string]string) {
t.Helper()
dir := t.TempDir()
keysPath := filepath.Join(dir, "keys.toml")
keys := map[string]string{}
for _, name := range callers {
out := filepath.Join(dir, name)
if err := auth.Add(keysPath, name, "hermes-fleet", out); err != nil {
t.Fatal(err)
}
b, err := os.ReadFile(out)
if err != nil {
t.Fatal(err)
}
keys[name] = strings.TrimSpace(string(b))
}
store, err := auth.Load(keysPath)
if err != nil {
t.Fatal(err)
}
g := &guard.Guard{AllowPrefixes: loopback, LookupIP: fakeDNS}
f := &fetch.Fetcher{Guard: g, UA: "sift-test"}
log := slog.New(slog.NewTextHandler(io.Discard, nil))
s := New("test", g, f, store, log)
db, err := dcache.Open("", 1<<20, log)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = db.Close() })
s.Batch = s.NewBatchRunner(db, batch.Config{})
t.Cleanup(s.Batch.Close) // before db.Close
return s, keys
}
// pageServer serves a small HTML page at every path; hold, when non-nil, blocks each request
// until it is closed.
func pageServer(t *testing.T, hold chan struct{}) *httptest.Server {
t.Helper()
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if hold != nil {
select {
case <-hold:
case <-r.Context().Done():
return
}
}
w.Header().Set("Content-Type", "text/html; charset=utf-8")
fmt.Fprintf(w, `<html><head><title>T</title></head><body><article><p>hello batch %s</p></article></body></html>`, r.URL.Path)
}))
t.Cleanup(ts.Close)
return ts
}
func waitJob(t *testing.T, r *batch.Runner, id string) batch.Job {
t.Helper()
deadline := time.Now().Add(10 * time.Second)
for time.Now().Before(deadline) {
j, _, ok := r.Status(id)
if ok && (j.Status == "done" || j.Status == "partial" || j.Status == "failed" || j.Status == "cancelled") {
return j
}
time.Sleep(5 * time.Millisecond)
}
t.Fatal("job never finished")
return batch.Job{}
}
func promText(t *testing.T, s *Server) string {
t.Helper()
req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
req.RemoteAddr, req.Host = "127.0.0.1:9", "127.0.0.1:8890"
rec := httptest.NewRecorder()
s.ServeHTTP(rec, req)
return rec.Body.String()
}
// The runner fetches through fetchTarget as the submitting caller; a name that resolves to a
// private address is a per-target "blocked" failure; the hooks reach the metrics.
func TestBatchRunnerWired(t *testing.T) {
ts := pageServer(t, nil)
s, _ := newBatchServer(t, "test")
c := &auth.Caller{Name: "test", Workspace: "hermes-fleet"}
id, reused, err := s.Batch.Submit(context.Background(), c, dcache.ClassFleet, []batch.Target{
{URL: ts.URL + "/one"}, {URL: "http://internal.test/"},
})
if err != nil || reused {
t.Fatalf("submit: %v %v", err, reused)
}
if j := waitJob(t, s.Batch, id); j.Status != "partial" {
t.Fatalf("job %+v", j)
}
_, ts2, _ := s.Batch.Status(id)
if ts2[0].Status != "done" || ts2[0].Result == nil || !strings.Contains(ts2[0].Result.Content, "hello batch /one") {
t.Fatalf("target 0: %+v", ts2[0])
}
if ts2[0].Result.Meta["mode"] != "clean" || ts2[0].Result.Meta["untrusted"] != true {
t.Fatalf("target 0 meta %v: not the fetchTarget envelope", ts2[0].Result.Meta)
}
if ts2[1].Status != "failed" || !strings.Contains(ts2[1].Error, "blocked") {
t.Fatalf("target 1: %+v", ts2[1])
}
out := promText(t, s)
for _, want := range []string{
`sift_batch_jobs_total{outcome="created"} 1`,
`sift_batch_jobs_total{outcome="partial"} 1`,
`sift_batch_targets_total{outcome="ok"} 1`,
`sift_batch_targets_total{outcome="blocked"} 1`,
`sift_batch_result_bytes_count 1`,
`sift_batch_queue_depth 0`,
} {
if !strings.Contains(out, want) {
t.Errorf("metrics lack %q", want)
}
}
}
// The adapter turns fetch errors into what the runner and the metrics need.
func TestBatchOutcomeClasses(t *testing.T) {
cases := []struct {
err error
want string
}{
{nil, "ok"},
{fmt.Errorf("busy: %w", batch.ErrRetryLater), "busy"},
{&batch.TargetError{Status: 403, Msg: "blocked: private address"}, "blocked"},
{&batch.TargetError{Status: 429, Msg: "busy"}, "busy"},
{&batch.TargetError{Status: 504, Msg: "timeout"}, "timeout"},
{&batch.TargetError{Status: 415, Msg: "unsupported content type"}, "unsupported"},
{&batch.TargetError{Status: 422, Msg: "document too long for question (6001 chunks); use grep or pages"}, "too_long"},
{&batch.TargetError{Status: 422, Msg: "unreadable PDF"}, "upstream_4xx"},
{&batch.TargetError{Status: 404, Msg: "not found"}, "upstream_4xx"},
{&batch.TargetError{Status: 500, Msg: "internal error"}, "upstream_5xx"},
{errors.New("something else"), "upstream_5xx"},
}
for _, c := range cases {
if got := batchOutcome(c.err); got != c.want {
t.Errorf("batchOutcome(%v) = %q, want %q", c.err, got, c.want)
}
}
}
// batchFetch maps a busy fetch to ErrRetryLater (the runner retries it) and every other
// failure to a TargetError with the /v1/fetch status.
func TestBatchFetchErrors(t *testing.T) {
s, _ := newBatchServer(t, "test")
c := &auth.Caller{Name: "test", Workspace: "hermes-fleet"}
_, err := s.batchFetch(context.Background(), c, batch.Target{URL: "http://internal.test/"})
var te *batch.TargetError
if !errors.As(err, &te) || te.Status != 403 || !strings.Contains(te.Msg, "blocked") {
t.Fatalf("private name: %v, want a TargetError 403 blocked", err)
}
// Fill the global fetch gate so the next fetch is refused busy.
for i := 0; i < 8; i++ {
if err := s.acquire(fmt.Sprintf("filler%d", i), "http://filler.test/"); err != nil {
t.Fatal(err)
}
}
ts := pageServer(t, nil)
_, err = s.batchFetch(context.Background(), c, batch.Target{URL: ts.URL})
if !errors.Is(err, batch.ErrRetryLater) {
t.Fatalf("busy gate: %v, want an error wrapping batch.ErrRetryLater", err)
}
}
@@ -0,0 +1,94 @@
package api
// P6 task 04: v1fetch's per-target pipeline is extracted into s.fetchTarget, which the batch
// runner also calls. These tests pin that the two agree; every existing v1fetch test must also
// pass unchanged.
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"net/netip"
"strings"
"testing"
"git.wntrmute.dev/kyle/sift/internal/auth"
"git.wntrmute.dev/kyle/sift/internal/batch"
"git.wntrmute.dev/kyle/sift/internal/fetch"
"git.wntrmute.dev/kyle/sift/internal/guard"
)
// newTargetServer is setupAPI with loopback allowed; its key belongs to caller "test".
func newTargetServer(t *testing.T) (*Server, string) {
t.Helper()
return setupAPI(t, &guard.Guard{AllowPrefixes: []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")}})
}
func TestFetchTargetMatchesV1Fetch(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "text/html; charset=utf-8")
_, _ = w.Write([]byte(`<html><head><title>Same</title></head><body><article><h1>Same</h1><p>one pipeline, two callers</p></article></body></html>`))
}))
defer ts.Close()
s, key := newTargetServer(t)
c := &auth.Caller{Name: "test", Workspace: "hermes-fleet"}
res, err := s.fetchTarget(context.Background(), c, batch.Target{URL: ts.URL})
if err != nil {
t.Fatal(err)
}
if !strings.Contains(res.Content, "one pipeline, two callers") {
t.Fatalf("content %q", res.Content)
}
if _, has := res.Meta["content"]; has {
t.Fatal("Meta must not repeat the content")
}
for _, k := range []string{"mode", "untrusted", "source", "warnings", "sizes", "timings"} {
if _, has := res.Meta[k]; !has {
t.Errorf("Meta lacks %q (it carries every envelope field except content)", k)
}
}
if res.Meta["mode"] != "clean" || res.Meta["untrusted"] != true {
t.Fatalf("meta %v", res.Meta)
}
// The same URL through /v1/fetch: identical content and mode, and the envelope has exactly
// the fields of Meta plus content.
rec := postJSON(t, s, "/v1/fetch", key, map[string]any{"url": ts.URL})
if rec.Code != 200 {
t.Fatalf("v1fetch %d %s", rec.Code, rec.Body.String())
}
var env map[string]any
if err := json.Unmarshal(rec.Body.Bytes(), &env); err != nil {
t.Fatal(err)
}
if env["content"] != res.Content || env["mode"] != res.Meta["mode"] {
t.Fatalf("v1fetch and fetchTarget disagree:\n%v\n%+v", env, res)
}
if len(env) != len(res.Meta)+1 {
t.Fatalf("envelope keys %d, Meta keys %d: want Meta + content", len(env), len(res.Meta))
}
}
func TestFetchTargetErrors(t *testing.T) {
s, _ := newTargetServer(t)
c := &auth.Caller{Name: "test", Workspace: "hermes-fleet"}
cases := []struct {
name string
target batch.Target
status int
msg string
}{
{"blocked", batch.Target{URL: "http://10.0.0.5/"}, 403, "blocked"},
{"bad mode", batch.Target{URL: "http://127.0.0.1/", Options: batch.Options{Mode: "shout"}}, 400, ""},
}
for _, tc := range cases {
_, err := s.fetchTarget(context.Background(), c, tc.target)
var fe *fetch.Error
if !errors.As(err, &fe) || fe.Status != tc.status || !strings.Contains(fe.Msg, tc.msg) {
t.Errorf("%s: err %v, want a *fetch.Error %d containing %q", tc.name, err, tc.status, tc.msg)
}
}
}
+284 -66
View File
@@ -3,15 +3,24 @@ package batch
import (
"context"
"errors"
"fmt"
"io"
"log/slog"
"strings"
"sync"
"testing"
"time"
"git.wntrmute.dev/kyle/sift/internal/auth"
"git.wntrmute.dev/kyle/sift/internal/dcache"
)
var (
alice = &auth.Caller{Name: "alice", Workspace: "hermes-fleet"}
bob = &auth.Caller{Name: "bob", Workspace: "hermes-fleet"}
carol = &auth.Caller{Name: "carol", Workspace: "hermes-fleet"}
)
func newRunner(t *testing.T, fetch FetchFunc, cfg Config) *Runner {
t.Helper()
db, err := dcache.Open("", 1<<20, slog.New(slog.NewTextHandler(io.Discard, nil)))
@@ -19,34 +28,46 @@ func newRunner(t *testing.T, fetch FetchFunc, cfg Config) *Runner {
t.Fatal(err)
}
t.Cleanup(func() { _ = db.Close() })
return New(db, fetch, cfg, slog.New(slog.NewTextHandler(io.Discard, nil)))
r := New(db, fetch, cfg, slog.New(slog.NewTextHandler(io.Discard, nil)))
t.Cleanup(r.Close) // runs before db.Close (cleanups are LIFO)
return r
}
func waitStatus(t *testing.T, r *Runner, id, want string) {
t.Helper()
deadline := time.Now().Add(3 * time.Second)
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) {
j, _, ok := r.Status(id)
if ok && j.Status == want {
return
}
time.Sleep(time.Millisecond)
time.Sleep(2 * time.Millisecond)
}
j, _, _ := r.Status(id)
t.Fatalf("job %s never reached %q (now %q)", id, want, j.Status)
}
func TestSerialOrder(t *testing.T) {
func urls(ss ...string) []Target {
var out []Target
for _, s := range ss {
out = append(out, Target{URL: "https://" + s + ".example/"})
}
return out
}
// 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 []string
var calls, names []string
inflight, peak := 0, 0
fetch := func(_ context.Context, tgt Target) (Result, error) {
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()
@@ -55,111 +76,215 @@ func TestSerialOrder(t *testing.T) {
return Result{Content: "ok " + tgt.URL}, nil
}
r := newRunner(t, fetch, Config{})
id, reused, err := r.Submit(context.Background(), "alice", "fleet", []Target{
{URL: "https://a.example/"}, {URL: "https://b.example/"}, {URL: "https://c.example/"},
})
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()
order := append([]string(nil), calls...)
p := peak
mu.Unlock()
if len(order) != 3 || order[0] != "https://a.example/" || order[1] != "https://b.example/" || order[2] != "https://c.example/" {
t.Fatalf("order %v", order)
defer mu.Unlock()
if strings.Join(calls, " ") != "https://a.example/ https://b.example/ https://c.example/" {
t.Fatalf("order %v", calls)
}
if p != 1 {
t.Fatalf("peak inflight %d, want 1 (serial)", p)
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, tgt Target) (Result, error) {
fetch := func(_ context.Context, _ *auth.Caller, tgt Target) (Result, error) {
if tgt.URL == "https://b.example/" {
return Result{}, errors.New("upstream 500")
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", "fleet", []Target{
{URL: "https://a.example/"}, {URL: "https://b.example/"}, {URL: "https://c.example/"},
})
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 {
t.Fatal("no job")
}
if j.Status != "partial" || j.TargetsDone != 3 {
t.Fatalf("job %+v", j)
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 == "" {
t.Fatal("failed target missing error")
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, tgt Target) (Result, error) {
fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
mu.Lock()
n++
mu.Unlock()
return Result{Content: "ok"}, nil
}
r := newRunner(t, fetch, Config{})
targets := []Target{{URL: "https://a.example/"}, {URL: "https://b.example/"}}
id1, reused1, err := r.Submit(context.Background(), "alice", "fleet", targets)
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", "fleet", targets)
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()
calls := n
mu.Unlock()
if calls != 2 {
t.Fatalf("fetched %d times, want 2 (idempotent)", calls)
defer mu.Unlock()
if n != 4 {
t.Fatalf("fetched %d times, want 4 (2 for alice's one job, 2 for bob's)", n)
}
}
func TestTooManyTargets(t *testing.T) {
fetch := func(_ context.Context, tgt Target) (Result, error) { return Result{Content: "ok"}, nil }
r := newRunner(t, fetch, Config{TargetsPerJob: 2})
three := []Target{{URL: "https://a.example/"}, {URL: "https://b.example/"}, {URL: "https://c.example/"}}
if _, _, err := r.Submit(context.Background(), "alice", "fleet", three); !errors.Is(err, ErrTooManyTargets) {
t.Fatalf("want ErrTooManyTargets, got %v", err)
}
}
func TestCallerBusy(t *testing.T) {
// 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, tgt Target) (Result, error) {
<-release // hold the job in the running state
fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
<-release
return Result{Content: "ok"}, nil
}
r := newRunner(t, fetch, Config{})
id, _, err := r.Submit(context.Background(), "alice", "fleet", []Target{{URL: "https://a.example/"}})
id1, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
if err != nil {
t.Fatal(err)
}
waitStatus(t, r, id, "running")
if _, _, err := r.Submit(context.Background(), "alice", "fleet", []Target{{URL: "https://b.example/"}}); !errors.Is(err, ErrCallerBusy) {
t.Fatalf("want ErrCallerBusy, got %v", 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)
}
}
func TestCaps(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{TargetsPerJob: 2, JobsTotal: 2})
defer close(release)
if _, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b", "c")); !errors.Is(err, ErrTooManyTargets) {
t.Fatalf("3 targets over a cap of 2: want ErrTooManyTargets, got %v", err)
}
if _, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, nil); !errors.Is(err, ErrNoTargets) {
t.Fatalf("no targets: want ErrNoTargets, got %v", err)
}
if _, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a")); err != nil {
t.Fatal(err)
}
// alice's job is live (queued or running): a second, different list is refused.
if _, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("z")); !errors.Is(err, ErrCallerBusy) {
t.Fatalf("want ErrCallerBusy, got %v", err)
}
if _, _, err := r.Submit(context.Background(), bob, dcache.ClassFleet, urls("b")); err != nil {
t.Fatal(err)
}
// Two live jobs in total: a third caller is refused.
if _, _, err := r.Submit(context.Background(), carol, dcache.ClassFleet, urls("c")); !errors.Is(err, ErrBusy) {
t.Fatalf("want ErrBusy, got %v", 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) {
@@ -167,7 +292,7 @@ func TestCancel(t *testing.T) {
release := make(chan struct{})
var mu sync.Mutex
var calls []string
fetch := func(_ context.Context, tgt Target) (Result, error) {
fetch := func(_ context.Context, _ *auth.Caller, tgt Target) (Result, error) {
mu.Lock()
calls = append(calls, tgt.URL)
mu.Unlock()
@@ -179,27 +304,26 @@ func TestCancel(t *testing.T) {
return Result{Content: "ok"}, nil
}
r := newRunner(t, fetch, Config{})
id, _, err := r.Submit(context.Background(), "alice", "fleet", []Target{
{URL: "https://a.example/"}, {URL: "https://b.example/"}, {URL: "https://c.example/"},
})
id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b", "c"))
if err != nil {
t.Fatal(err)
}
<-started // first target is in flight
<-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)
}
_, ts, ok := r.Status(id)
if !ok {
t.Fatal("no job")
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])
@@ -207,19 +331,66 @@ func TestCancel(t *testing.T) {
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, tgt Target) (Result, error) {
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", "fleet", []Target{
{URL: "https://a.example/"}, {URL: "https://b.example/"}, {URL: "https://c.example/"},
})
id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b", "c"))
if err != nil {
t.Fatal(err)
}
@@ -228,7 +399,54 @@ func TestPanicRecovered(t *testing.T) {
if ts[0].Status != "done" || ts[1].Status != "failed" || ts[2].Status != "done" {
t.Fatalf("statuses %+v", ts)
}
if ts[1].Error == "" {
t.Fatal("panic target missing error")
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)
}
}
func TestStatusUnknown(t *testing.T) {
r := newRunner(t, func(context.Context, *auth.Caller, Target) (Result, error) { return Result{}, nil }, Config{})
if _, _, ok := r.Status("no-such-job"); ok {
t.Fatal("unknown id must be ok=false")
}
}
+27 -15
View File
@@ -1,34 +1,46 @@
package main
import "testing"
import (
"slices"
"testing"
)
func TestParseBatchArgs(t *testing.T) {
cases := []struct {
args []string
verb string
arg string
rest []string
ok bool
}{
{[]string{"submit", "targets.json"}, "submit", "targets.json", true},
{[]string{"submit", "-"}, "submit", "-", true},
{[]string{"status", "job-123"}, "status", "job-123", true},
{[]string{"cancel", "job-123"}, "cancel", "job-123", true},
{[]string{}, "", "", false},
{[]string{"status"}, "", "", false},
{[]string{"status", "a", "b"}, "", "", false},
{[]string{"bogus", "x"}, "", "", false},
{[]string{"submit", "targets.json"}, "submit", []string{"targets.json"}, true},
{[]string{"submit", "-"}, "submit", []string{"-"}, true},
{[]string{"status", "job-123"}, "status", []string{"job-123"}, true},
{[]string{"status", "--results", "job-123"}, "status", []string{"--results", "job-123"}, true},
{[]string{"get", "job-123", "4"}, "get", []string{"job-123", "4"}, true},
{[]string{"cancel", "job-123"}, "cancel", []string{"job-123"}, true},
{[]string{}, "", nil, false},
{[]string{"submit"}, "", nil, false},
{[]string{"status"}, "", nil, false},
{[]string{"status", "a", "b"}, "", nil, false},
{[]string{"status", "--results"}, "", nil, false},
{[]string{"get", "job-123"}, "", nil, false},
{[]string{"get", "job-123", "x"}, "", nil, false},
{[]string{"get", "job-123", "-1"}, "", nil, false},
{[]string{"cancel"}, "", nil, false},
{[]string{"bogus", "x"}, "", nil, false},
}
for _, c := range cases {
verb, arg, err := parseBatchArgs(c.args)
verb, rest, err := parseBatchArgs(c.args)
if c.ok {
if err != nil {
t.Fatalf("%v: unexpected error %v", c.args, err)
t.Errorf("%v: unexpected error %v", c.args, err)
continue
}
if verb != c.verb || arg != c.arg {
t.Fatalf("%v: got verb=%q arg=%q", c.args, verb, arg)
if verb != c.verb || !slices.Equal(rest, c.rest) {
t.Errorf("%v: got verb=%q rest=%q", c.args, verb, rest)
}
} else if err == nil {
t.Fatalf("%v: want error", c.args)
t.Errorf("%v: want an error", c.args)
}
}
}
+56
View File
@@ -0,0 +1,56 @@
package config
// P6 task 07: the [batch] block and ParseSize.
import (
"os"
"path/filepath"
"testing"
)
func TestBatchBlock(t *testing.T) {
p := filepath.Join(t.TempDir(), "sift.toml")
body := "[batch]\ndisabled = false\ntargets_per_job = 16\njobs_per_caller = 2\njobs_total = 4\nworkers = 2\nttl = \"12h\"\nmax_bytes = \"32MiB\"\n"
if err := os.WriteFile(p, []byte(body), 0o600); err != nil {
t.Fatal(err)
}
f, err := Load(p)
if err != nil {
t.Fatal(err)
}
b := f.Batch
if b.Disabled || b.TargetsPerJob != 16 || b.JobsPerCaller != 2 || b.JobsTotal != 4 || b.Workers != 2 || b.TTL != "12h" || b.MaxBytes != "32MiB" {
t.Fatalf("batch block %+v", b)
}
// No block: zero values, and batch is on (the runner applies its own defaults).
empty := filepath.Join(t.TempDir(), "empty.toml")
if err := os.WriteFile(empty, []byte("listen = \":8443\"\n"), 0o600); err != nil {
t.Fatal(err)
}
if f, err := Load(empty); err != nil || f.Batch != (Batch{}) {
t.Fatalf("no [batch] block: %+v %v", f.Batch, err)
}
}
func TestParseSize(t *testing.T) {
const def = 7
for _, c := range []struct {
in string
want int64
}{
{"64MiB", 64 << 20},
{"512KiB", 512 << 10},
{"1GiB", 1 << 30},
{"1000", 1000},
{" 2MiB ", 2 << 20},
{"", def},
{"junk", def},
{"-1MiB", def},
{"0", def},
{"12MB", def}, // decimal units are not accepted: say MiB
} {
if got := ParseSize(c.in, def); got != c.want {
t.Errorf("ParseSize(%q) = %d, want %d", c.in, got, c.want)
}
}
}
+14
View File
@@ -56,6 +56,20 @@ func TestBatchIDemLookup(t *testing.T) {
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) {
@@ -0,0 +1,53 @@
package guard
// P6 task 05: PreCheck is CheckURL without name resolution, for refusing a batch target at
// submit time.
import (
"context"
"net/netip"
"strings"
"testing"
)
func TestPreCheck(t *testing.T) {
g := &Guard{
AllowPrefixes: []netip.Prefix{netip.MustParsePrefix("127.0.0.0/8")},
LookupIP: func(context.Context, string) ([]netip.Addr, error) {
t.Fatal("PreCheck resolved a name")
return nil, nil
},
}
for _, u := range []string{
"https://example.com/",
"http://example.com/path?q=1",
"https://internal.test/", // a name: not resolved here, so not refused here
"http://127.0.0.1:8080/", // an allowed literal is exempt from the port rule, as in CheckURL
"http://93.184.215.14/", // a public literal
"http://0x7f000001:8080/", // inet_aton form of 127.0.0.1: an allowed literal
} {
if err := g.PreCheck(u); err != nil {
t.Errorf("PreCheck(%q) = %v, want nil", u, err)
}
}
for _, u := range []string{
"",
"ftp://example.com/",
"http://user@example.com/",
"http://example.com:8080/",
"http://10.0.0.5/",
"http://169.254.169.254/latest/meta-data/",
"http://[::1]/",
"http://0x0a000005/", // inet_aton form of 10.0.0.5
} {
err := g.PreCheck(u)
if err == nil || !strings.HasPrefix(err.Error(), "blocked") {
t.Errorf("PreCheck(%q) = %v, want a blocked error", u, err)
}
}
// With no allow list, loopback is refused like any other private literal.
var zero Guard
if err := zero.PreCheck("http://127.0.0.1/"); err == nil {
t.Error("a zero Guard allowed loopback")
}
}