P6 during the run: split the runner task (02 core, 03 workers); guard.PreCheck as an owner commit; AGENTS.md: write code into files early
Task 02 (the whole runner) ended at 21:08 with no file and no commit: Ornith wrote the
implementation out in its reply (~700 lines), OpenCode compacted the session, and the model
then emitted a malformed tool call. It had also looked for the given test at a wrong path
("_files/…" relative to the plan: my ambiguity).
- 02-runner → 02-runner-core (types, Submit, Status, Touch, Cancel; no goroutines) and
03-runner-workers; the given test split to match (runner_core_test.go gains three tests:
queued storage, cancel while queued, Status never moves the lease); later tasks renumbered
04-09 and every cross-reference checked.
- Copy paths spelled from the repo root.
- guard.PreCheck (+ its test) is committed here by the owner: the old task 05 told the
implementer to edit internal/guard/, which AGENTS.md forbids. My contradiction.
- AGENTS.md rule 6: write code into files early, in small steps.
Both runner test files compile against a panic-only skeleton (the core file also alone) and
fail there on panic: todo.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
+15
-7
@@ -242,13 +242,14 @@ The design stands; these were implementation-blocking or would have hurt agents:
|
||||
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
|
||||
of the allow list. New `Guard.PreCheck` is `CheckURL` without resolution (an owner commit on
|
||||
master, since `AGENTS.md` keeps the implementer out of `internal/guard/`). 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`.
|
||||
9. **Task shape.** 5 tasks → 9 (the last split came during the run, below), one package each where possible:
|
||||
01 dcache, 02 runner core, 03 runner workers, 04 metrics, 05 the `fetchTarget` refactor alone,
|
||||
06 runner wiring, 07 endpoints, 08 serve/config, 09 CLI/docs. 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;
|
||||
@@ -256,5 +257,12 @@ The design stands; these were implementation-blocking or would have hurt agents:
|
||||
- `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
|
||||
`retrieval`/`embedded`), which is the code task 05 moves. Recheck the api given tests against
|
||||
a skeleton built on post-P5c master before the run.
|
||||
|
||||
**During the run (2026-09-26 21:08).** Task 02 (the whole runner) ended with no file: the model
|
||||
wrote the implementation out in its reply, the session was compacted, and it then emitted a
|
||||
malformed tool call. The runner is now two tasks (02 core, 03 workers), each with its own given
|
||||
test. `AGENTS.md` gains a rule to write code into files early. Copy paths are spelled from the
|
||||
repo root. Details in `docs/plans/p6/README.md`, "Changes during the run".
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@
|
||||
|
||||
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 05) call them.
|
||||
their accessors live here; the runner (task 02) and the API (tasks 06–07) call them.
|
||||
|
||||
## Context
|
||||
|
||||
@@ -102,9 +102,9 @@ caller decides the clock, so the given test asserts exact lease values.
|
||||
`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.
|
||||
a job alive because the API calls `TouchBatchJob` after it (task 07), 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 05). The store stores what it is told and returns
|
||||
caller filter is the API layer's job (task 07). The store stores what it is told and returns
|
||||
what is asked.
|
||||
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`,
|
||||
@@ -123,7 +123,7 @@ caller decides the clock, so the given test asserts exact lease values.
|
||||
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; `serve` calls it on startup (task 06).
|
||||
reconciliation hook; `serve` calls it on startup (task 08).
|
||||
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.
|
||||
9. Accessors follow the existing `dcache` style: return `(value, bool)` or `error`, never panic,
|
||||
@@ -132,7 +132,7 @@ caller decides the clock, so the given test asserts exact lease values.
|
||||
|
||||
## Given
|
||||
|
||||
Copy `_files/dcache/batch_test.go` → `internal/dcache/batch_test.go`. It opens an in-memory `DB`
|
||||
Copy `docs/plans/p6/_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 and the
|
||||
status filter (rule 3), live-job
|
||||
counts, ordered targets, the exactly-once `targets_done`/lease-bump rules, `SweepBatch`,
|
||||
|
||||
@@ -1,24 +1,35 @@
|
||||
# P6 task 02: the batch runner in `internal/batch`
|
||||
# P6 task 02: the batch runner's core in `internal/batch` (no workers yet)
|
||||
|
||||
**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`
|
||||
**Commit subject:** `batch: runner core — types, Submit with idempotency and caps, Status, Touch, Cancel`
|
||||
|
||||
## Goal
|
||||
|
||||
A new package `internal/batch`: the job lifecycle. It creates a job, lets a small pool of
|
||||
workers drain its targets one at a time, writes each outcome to the store, handles cancellation,
|
||||
retries a target whose fetch was only busy, and enforces the caps and idempotency.
|
||||
A new package `internal/batch`. This task writes the whole public surface, and the part of it
|
||||
that needs no goroutines:
|
||||
- `Submit` (idempotency, caps, storing the job);
|
||||
- `Status`, `Touch`;
|
||||
- `Cancel` for a job no worker holds.
|
||||
|
||||
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.
|
||||
**Task 03 adds the workers.** Until then every job stays `queued`: `New` starts nothing, and
|
||||
`Close` has nothing to stop.
|
||||
|
||||
The runner knows **nothing** about HTTP, the guard, or the real fetch pipeline. The per-target
|
||||
work is an injected `FetchFunc` (unused until task 03), 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)
|
||||
- Copy: `docs/plans/p6/_files/batch/runner_core_test.go` → `internal/batch/runner_core_test.go`
|
||||
- Create: `internal/batch/batch.go`: every type, variable and function in Interfaces below.
|
||||
Task 03 adds `worker.go`.
|
||||
- Modify: `docs/implementer-log.md`
|
||||
|
||||
**Work in small steps, in files.** First write `batch.go` with the types and function
|
||||
signatures, and make it compile (`go vet ./internal/batch/`). Then fill in one function at a
|
||||
time, running the tests as you go. Do not write the implementation out in your reply before
|
||||
putting it in a file.
|
||||
|
||||
## What already exists (do not re-read these packages; this is their API)
|
||||
|
||||
```go
|
||||
@@ -151,50 +162,32 @@ func (r *Runner) Close() // stops the work
|
||||
- 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
|
||||
- `Hooks.Job("created")`; `SweepBatch(now, TTL)`. (Task 03 also wakes a worker here.)
|
||||
2. **`New`** stores its arguments and applies the `Config` defaults. It starts no goroutine
|
||||
(task 03 starts the workers there). **`Close`** is safe to call more than once.
|
||||
3. **Cancel(id)** returns false for an unknown id or a job already terminal (`done`, `partial`,
|
||||
`failed`, `cancelled`). Otherwise it sets the job `cancelled` at once (its pending targets
|
||||
stay `pending`) and calls `Hooks.Job("cancelled")`. Keep the rest of `Cancel` simple;
|
||||
task 03 adds the in-flight target case.
|
||||
4. **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
|
||||
5. Library code never panics; a store error is logged (`Warn`, with the job id, never a URL) and
|
||||
the job goes on. Log lines carry the job id, counts and statuses, **never a URL or host**.
|
||||
|
||||
|
||||
## Facts
|
||||
|
||||
- The given test uses an in-memory `dcache` and `FetchFunc`s that hold on channels, fail, return
|
||||
`ErrRetryLater`, or panic. It calls `Close` before the store closes.
|
||||
- `TestRoundRobin` holds the first target until the second job is submitted, then expects
|
||||
`a1 b1 a2 b2 a3 b3` from one worker.
|
||||
- `runner_core_test.go` holds the helpers that task 03's test also uses (`newRunner`,
|
||||
`waitStatus`, `urls`, and the callers `alice`, `bob`, `carol`).
|
||||
- Without workers, a submitted job stays `queued`, so it counts as live for the caps.
|
||||
|
||||
## Steps
|
||||
|
||||
- [ ] **1.** `git switch p6`; copy the given test. **2. See it fail** (does not compile).
|
||||
- [ ] **3.** `batch.go`, **4.** `worker.go`.
|
||||
- [ ] **5.** `gofmt -w`; `go test -race -count=3 ./internal/batch/` → `ok`. **6.** `make gate`.
|
||||
- [ ] **7.** Row `p6/02-runner`; commit.
|
||||
- [ ] **3.** `batch.go`: types and signatures first, compile; then `Submit`, `Status`, `Touch`,
|
||||
`Cancel`.
|
||||
- [ ] **4.** `gofmt -w`; `go test -race -count=3 ./internal/batch/` → `ok`. **5.** `make gate`.
|
||||
- [ ] **6.** Row `p6/02-runner-core`; commit.
|
||||
|
||||
```sh
|
||||
git add internal/batch docs/implementer-log.md
|
||||
@@ -204,7 +197,3 @@ 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.
|
||||
@@ -0,0 +1,86 @@
|
||||
# P6 task 03: the batch runner's workers
|
||||
|
||||
**Branch:** `p6` (`git switch p6`; `git status --short` must be empty, otherwise stop)
|
||||
**Commit subject:** `batch: workers — round-robin, serial per job, busy retry, cancel in flight, hooks`
|
||||
|
||||
## Goal
|
||||
|
||||
Task 02 wrote the runner's types, `Submit`, `Status`, `Touch` and `Cancel`; every job stays
|
||||
`queued`. This task adds the workers that drain jobs: one target at a time per job, round-robin
|
||||
across jobs, with a retry for a target that was only busy.
|
||||
|
||||
## Files
|
||||
|
||||
- Copy: `docs/plans/p6/_files/batch/runner_test.go` → `internal/batch/runner_test.go`
|
||||
- Create: `internal/batch/worker.go` (the workers, the rotation, one target, the terminal status)
|
||||
- Modify: `internal/batch/batch.go` (`New` starts the workers; `Close` stops them; `Submit` wakes
|
||||
one; `Cancel` handles a job with a target in flight), `docs/implementer-log.md`
|
||||
|
||||
**Work in small steps, in files.** Write `worker.go` with the worker loop and a stub for one
|
||||
target, and make it compile. Then fill it in, running the tests as you go. Do not write the
|
||||
implementation out in your reply before putting it in a file.
|
||||
|
||||
## What already exists (task 02, in `internal/batch/batch.go`)
|
||||
|
||||
All the types in the task 02 file: `Options`, `Target`, `Result`, `TargetError`, `FetchFunc`,
|
||||
`Job`, `TargetStatus`, `Hooks`, `Config`, the errors (including `ErrRetryLater`), `Runner`,
|
||||
`New`, `Submit`, `Status`, `Touch`, `Cancel`, `Close`. Read `batch.go` for how the runner keeps
|
||||
its jobs and each job's caller; the dcache API it uses is listed in the task 02 file's
|
||||
"What already exists".
|
||||
|
||||
## Rules
|
||||
|
||||
1. **Workers.** `Workers` goroutines share one rotation of live jobs in submit order. A worker
|
||||
takes the next job in the rotation that has a pending target **and no target in flight**. It
|
||||
sets the job to `running` on its first target, fetches that one target, stores it, and moves
|
||||
on to the next job in the rotation. So:
|
||||
- each job is serial: its target `i+1` never starts before `i` is stored;
|
||||
- with one worker and two live jobs, their targets alternate (`a1 b1 a2 b2 …`);
|
||||
- a job that a worker has not reached yet stays `queued`.
|
||||
2. **One target:**
|
||||
- Call the `FetchFunc` with the job's caller, under a context that `Close` cancels. Recover
|
||||
a panic into the error `internal error` (never echo the panic value).
|
||||
- If the error wraps `ErrRetryLater`, the target stays `pending`: the worker goes on with the
|
||||
rotation, and the target is tried again after `RetryWait`. A target that has been coming
|
||||
back busy for longer than `RetryFor` is stored `failed` with the error's text.
|
||||
- Otherwise store it: success → `done` with the JSON of the `Result`; error → `failed`, with
|
||||
`TargetError`'s `Msg`/`Hint` or `err.Error()` and no hint.
|
||||
- Then call `Hooks.Target(err, time, len(result JSON))`, then `EvictBatchOldest(MaxBytes)`.
|
||||
3. **Terminal status.** When a job's last target is stored: all `done` → `done`; all `failed` →
|
||||
`failed`; a mix → `partial`. `SetBatchJobStatus`, then `Hooks.Job(status)` and
|
||||
`Hooks.JobSeconds(now − created)`, exactly once per job.
|
||||
4. **Cancel(id)** returns false for an unknown id or a job already terminal. Otherwise it sets the job
|
||||
`cancelled` at once. The in-flight target, if any, finishes and is stored; no further target
|
||||
starts; later targets stay `pending`. Storing that last target must not overwrite `cancelled`.
|
||||
`Hooks.Job("cancelled")` once.
|
||||
5. **Hooks.Depth** gets the live-job count and **Hooks.Inflight** the in-flight count whenever
|
||||
either changes.
|
||||
|
||||
## Facts
|
||||
|
||||
- The given test uses `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.
|
||||
- Task 02's tests (`runner_core_test.go`) must still pass: with workers, a job there may
|
||||
already be `running` or `done` where it was `queued`; those tests allow for that.
|
||||
|
||||
## Steps
|
||||
|
||||
- [ ] **1.** `git switch p6`; copy the given test. **2. See it fail** (the jobs never finish).
|
||||
- [ ] **3.** `worker.go`. **4.** The `batch.go` changes.
|
||||
- [ ] **5.** `gofmt -w`; `go test -race -count=3 ./internal/batch/` → `ok`. **6.** `make gate`.
|
||||
- [ ] **7.** Row `p6/03-runner-workers`; commit.
|
||||
|
||||
```sh
|
||||
git add internal/batch docs/implementer-log.md
|
||||
git commit -F .state/commit-msg.txt
|
||||
```
|
||||
|
||||
## Done when
|
||||
|
||||
- Both given tests pass under `-race -count=3`; `make gate` ok; the given files are byte-identical.
|
||||
|
||||
## Stop and report if
|
||||
|
||||
- A given test's timing can only be met by sleeping longer than `RetryWait` inside a worker.
|
||||
@@ -1,11 +1,11 @@
|
||||
# P6 task 03: `sift_batch_*` series in `internal/metrics`
|
||||
# P6 task 04: `sift_batch_*` series in `internal/metrics`
|
||||
|
||||
**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`.
|
||||
`internal/metrics`**: task 06 wires these methods to the runner's `batch.Hooks`.
|
||||
|
||||
## Series (names and label values exact — the given test asserts these)
|
||||
|
||||
@@ -19,7 +19,7 @@ and the byte budget can be re-sized from observed data. **This task touches only
|
||||
| `sift_batch_target_seconds` | histogram | wall time per target |
|
||||
| `sift_batch_result_bytes` | histogram | stored bytes per target result |
|
||||
|
||||
Classifying an error into a target outcome is task 05's job (it has the `*fetch.Error`); this
|
||||
Classifying an error into a target outcome is task 06'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)
|
||||
@@ -62,7 +62,7 @@ upstream_5xx, unsupported, too_long, cancel`.
|
||||
|
||||
## Given
|
||||
|
||||
Copy `_files/metrics/batch_test.go` → `internal/metrics/batch_test.go`. It drives the `Observe…`
|
||||
Copy `docs/plans/p6/_files/metrics/batch_test.go` → `internal/metrics/batch_test.go`. It drives the `Observe…`
|
||||
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`.
|
||||
|
||||
@@ -71,7 +71,7 @@ label values, gauge values, histogram counts, and that a raw outcome string is f
|
||||
- [ ] **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.
|
||||
- [ ] **6.** Row `p6/04-metrics`; commit.
|
||||
|
||||
```sh
|
||||
git add internal/metrics docs/implementer-log.md
|
||||
@@ -1,4 +1,4 @@
|
||||
# P6 task 04: extract `v1fetch`'s pipeline into `fetchTarget`
|
||||
# P6 task 05: 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)`
|
||||
@@ -9,7 +9,7 @@
|
||||
## 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
|
||||
question limit → digest → context → log line → envelope. The batch worker (task 06) needs the
|
||||
same pipeline for one target. Move it into one method that both call:
|
||||
|
||||
```go
|
||||
@@ -21,7 +21,7 @@ func (s *Server) fetchTarget(ctx context.Context, c *auth.Caller, t batch.Target
|
||||
|
||||
## Files
|
||||
|
||||
- Copy: `_files/api/fetchtarget_test.go` → `internal/api/fetchtarget_test.go`
|
||||
- Copy: `docs/plans/p6/_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`
|
||||
@@ -69,7 +69,7 @@ already turns into a response.
|
||||
- [ ] **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.
|
||||
- [ ] **8.** Row `p6/05-fetchtarget`; commit.
|
||||
|
||||
```sh
|
||||
git add internal/api docs/implementer-log.md
|
||||
@@ -1,42 +1,22 @@
|
||||
# P6 task 05: the runner wired to `fetchTarget` and the metrics
|
||||
# P6 task 06: 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`
|
||||
**Commit subject:** `api: NewBatchRunner — the batch fetch adapter, retry on busy, outcome metrics`
|
||||
|
||||
## 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.
|
||||
Build the production runner: its `FetchFunc` is an adapter over `fetchTarget` (task 05), and its
|
||||
hooks feed the `sift_batch_*` metrics (task 04). **No HTTP endpoints yet** (task 07).
|
||||
|
||||
## Files
|
||||
|
||||
- Copy: `_files/api/batchrun_test.go` → `internal/api/batchrun_test.go`, and
|
||||
`_files/guard/precheck_test.go` → `internal/guard/precheck_test.go`
|
||||
- Copy: `docs/plans/p6/_files/api/batchrun_test.go` → `internal/api/batchrun_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
|
||||
@@ -64,12 +44,12 @@ 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
|
||||
// internal/metrics (task 04), 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)
|
||||
// internal/api (task 05 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
|
||||
@@ -81,9 +61,9 @@ func (s *Server) logErr(c *auth.Caller, err error) any
|
||||
|
||||
```go
|
||||
// in Server (after pdfPoolOnce):
|
||||
Batch *batch.Runner // nil: the /v1/batch endpoints (task 06) answer 503 "batch disabled"
|
||||
Batch *batch.Runner // nil: the /v1/batch endpoints (task 07) answer 503 "batch disabled"
|
||||
|
||||
// NewBatchRunner builds the runner that serve (task 07) and the tests use: its FetchFunc is
|
||||
// NewBatchRunner builds the runner that serve (task 08) 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
|
||||
|
||||
@@ -128,22 +108,22 @@ func batchOutcome(err error) string
|
||||
`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.
|
||||
- `batchrun_test.go` also holds the helpers task 07'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.
|
||||
- [ ] **1.** `git switch p6`; copy the given test. **2. See it fail.**
|
||||
- [ ] **3.** `batch.go`. **4.** The `Batch` field and the pool name.
|
||||
- [ ] **5.** `gofmt -w`; `go test -race -count=3 ./internal/api/` → `ok`.
|
||||
**6.** `make gate`.
|
||||
- [ ] **7.** Row `p6/06-batch-wiring`; commit.
|
||||
|
||||
```sh
|
||||
git add internal/api internal/guard/guard.go internal/guard/precheck_test.go docs/implementer-log.md
|
||||
git add internal/api 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.
|
||||
- The given test and every existing api test pass under `-race -count=3`; `make gate` ok; the
|
||||
given file is byte-identical.
|
||||
@@ -1,16 +1,16 @@
|
||||
# P6 task 06: the `/v1/batch` endpoints
|
||||
# P6 task 07: 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
|
||||
The HTTP surface over `s.Batch` (task 06): 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`
|
||||
- Copy: `docs/plans/p6/_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`
|
||||
|
||||
@@ -29,11 +29,11 @@ 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)
|
||||
// internal/guard (on master)
|
||||
func (g *Guard) PreCheck(raw string) error // "blocked: …" errors
|
||||
|
||||
// internal/api
|
||||
Batch *batch.Runner // in Server (task 05); nil → 503 "batch disabled"
|
||||
Batch *batch.Runner // in Server (task 06); 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)
|
||||
@@ -85,7 +85,7 @@ Add `const batchMaxBody = 256 << 10`.
|
||||
|
||||
## Facts
|
||||
|
||||
- `newBatchServer`, `pageServer` and `fakeDNS` come from `batchrun_test.go` (task 05). Targets
|
||||
- `newBatchServer`, `pageServer` and `fakeDNS` come from `batchrun_test.go` (task 06). 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`.
|
||||
@@ -95,7 +95,7 @@ Add `const batchMaxBody = 256 << 10`.
|
||||
- [ ] **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.
|
||||
- [ ] **7.** Row `p6/07-api-endpoints`; commit.
|
||||
|
||||
```sh
|
||||
git add internal/api docs/implementer-log.md
|
||||
@@ -1,4 +1,4 @@
|
||||
# P6 task 07: the `[batch]` config and `serve` wiring
|
||||
# P6 task 08: 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`
|
||||
@@ -10,7 +10,7 @@ killed as failed (they are never resumed). On shutdown it stops the runner befor
|
||||
|
||||
## Files
|
||||
|
||||
- Copy: `_files/config/batch_test.go` → `internal/config/batch_test.go`
|
||||
- Copy: `docs/plans/p6/_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`
|
||||
|
||||
@@ -21,10 +21,10 @@ killed as failed (they are never resumed). On shutdown it stops the runner befor
|
||||
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)
|
||||
// internal/api (task 06)
|
||||
func (s *Server) NewBatchRunner(db *dcache.DB, cfg batch.Config) *batch.Runner
|
||||
Batch *batch.Runner // in Server
|
||||
// s.Prom.ObserveBatchJob(outcome string) (task 03)
|
||||
// s.Prom.ObserveBatchJob(outcome string) (task 04)
|
||||
|
||||
// internal/dcache (task 01)
|
||||
func (d *DB) MarkInFlightFailed(reason string, now int64) int
|
||||
@@ -88,7 +88,7 @@ func ParseSize(s string, def int64) int64
|
||||
- [ ] **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.
|
||||
- [ ] **8.** Row `p6/08-serve-config`; commit.
|
||||
|
||||
```sh
|
||||
git add internal/config cmd/sift/serve.go dist/sift.toml.example docs/implementer-log.md
|
||||
@@ -1,4 +1,4 @@
|
||||
# P6 task 08: `sift batch`, README and skill
|
||||
# P6 task 09: `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`
|
||||
@@ -10,7 +10,7 @@ Hermes plugin.
|
||||
|
||||
## Files
|
||||
|
||||
- Copy: `_files/cmd_sift/batch_test.go` → `cmd/sift/batch_test.go`
|
||||
- Copy: `docs/plans/p6/_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`
|
||||
@@ -20,7 +20,7 @@ Hermes plugin.
|
||||
- `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):
|
||||
- The endpoints (task 07):
|
||||
- `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);
|
||||
@@ -76,7 +76,7 @@ sift batch cancel <job_id> # prints {"success", "status"}
|
||||
- [ ] **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.
|
||||
- [ ] **8.** Row `p6/09-cli-docs`; commit.
|
||||
|
||||
```sh
|
||||
git add cmd/sift README.md skill/SKILL.md docs/implementer-log.md
|
||||
+30
-7
@@ -39,13 +39,17 @@ Decisions (settled with the owner, 2026-09-26; revised after review the same eve
|
||||
| Task | Package | Given |
|
||||
|---|---|---|
|
||||
| 01-batchstore | `internal/dcache` | `dcache/batch_test.go` |
|
||||
| 02-runner | `internal/batch` | `batch/runner_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` |
|
||||
| 02-runner-core | `internal/batch` (types, Submit, Status, Touch, Cancel; no workers) | `batch/runner_core_test.go` |
|
||||
| 03-runner-workers | `internal/batch` (workers, retry, cancel in flight, hooks) | `batch/runner_test.go` |
|
||||
| 04-metrics | `internal/metrics` | `metrics/batch_test.go` |
|
||||
| 05-fetchtarget | `internal/api` (the one refactor of working code) | `api/fetchtarget_test.go` |
|
||||
| 06-batch-wiring | `internal/api` | `api/batchrun_test.go` |
|
||||
| 07-api-endpoints | `internal/api` | `api/batch_test.go` |
|
||||
| 08-serve-config | `internal/config`, `cmd/sift/serve.go` | `config/batch_test.go` |
|
||||
| 09-cli-docs | `cmd/sift`, README, skill | `cmd_sift/batch_test.go` |
|
||||
|
||||
`guard.PreCheck` and its test are on master (owner commit): `AGENTS.md` forbids the implementer
|
||||
to edit `internal/guard/`, the security boundary.
|
||||
|
||||
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
|
||||
@@ -56,3 +60,22 @@ work) on top of the P5 pool (`internal/pdfpool`), cache (`internal/dcache`), and
|
||||
(`global`/`per` in `internal/api`) infrastructure. Acceptance tests first: the given tests are
|
||||
authored against this spec and compiled against a panic-only skeleton so they fail for the
|
||||
intended reasons before each task is implemented.
|
||||
|
||||
## Changes during the run
|
||||
|
||||
- **Task 02 (runner), first attempt, 20:39–21:08. Model and task size, in about equal parts.**
|
||||
- The session wrote the whole runner out as ~700 lines of reasoning in its reply, never in a
|
||||
file. OpenCode compacted the context, and the model then emitted a malformed tool call
|
||||
(`</tool_call>`) and ended with no commit and no file.
|
||||
- It had also looked for the given test at a wrong path. The task said `_files/…` relative
|
||||
to the plan; that is my ambiguity.
|
||||
- The owner changes:
|
||||
- task 02 split into **02-runner-core** (no goroutines) and **03-runner-workers**, with the
|
||||
given test split to match (`runner_core_test.go` + `runner_test.go`); later tasks
|
||||
renumbered 04–09;
|
||||
- every copy path is spelled from the repo root (`docs/plans/p6/_files/…`);
|
||||
- `AGENTS.md` rule 6: write code into files early, in small steps;
|
||||
- `guard.PreCheck` moved to an owner commit on master. The old task 05 (now 06) told the
|
||||
implementer to edit `internal/guard/`, which `AGENTS.md` forbids: my contradiction, found
|
||||
while fixing the above.
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
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.
|
||||
// P6 task 06: the runner wired to fetchTarget and the metrics, with no HTTP yet. The helpers
|
||||
// here (newBatchServer, pageServer, fakeDNS) are shared with task 07's batch_test.go.
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
package api
|
||||
|
||||
// P6 task 04: v1fetch's per-target pipeline is extracted into s.fetchTarget, which the batch
|
||||
// P6 task 05: 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.
|
||||
|
||||
|
||||
@@ -0,0 +1,182 @@
|
||||
package batch
|
||||
|
||||
// P6 task 02: the runner's core — types, Submit (idempotency, caps, storage), Status, Touch and
|
||||
// Cancel — with no workers yet: every job stays queued. The helpers here are shared with task
|
||||
// 03's runner_test.go.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
"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)))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
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(5 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
j, _, ok := r.Status(id)
|
||||
if ok && j.Status == want {
|
||||
return
|
||||
}
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
}
|
||||
j, _, _ := r.Status(id)
|
||||
t.Fatalf("job %s never reached %q (now %q)", id, want, j.Status)
|
||||
}
|
||||
|
||||
func urls(ss ...string) []Target {
|
||||
var out []Target
|
||||
for _, s := range ss {
|
||||
out = append(out, Target{URL: "https://" + s + ".example/"})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Submit stores the job queued with its targets pending, before it returns.
|
||||
func TestSubmitStoresQueued(t *testing.T) {
|
||||
r := newRunner(t, func(context.Context, *auth.Caller, Target) (Result, error) {
|
||||
return Result{Content: "ok"}, nil
|
||||
}, Config{Workers: 1})
|
||||
id, reused, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
|
||||
if err != nil || reused || len(id) != 32 {
|
||||
t.Fatalf("submit id=%q reused=%v err=%v, want a 32-hex-char id", id, reused, err)
|
||||
}
|
||||
j, ts, ok := r.Status(id)
|
||||
if !ok || j.ID != id || j.Caller != "alice" || j.TargetsTotal != 2 || j.CreatedAt == 0 {
|
||||
t.Fatalf("job %+v ok=%v", j, ok)
|
||||
}
|
||||
if j.Status != "queued" && j.Status != "running" && j.Status != "done" {
|
||||
t.Fatalf("status %q", j.Status)
|
||||
}
|
||||
if len(ts) != 2 || ts[0].Idx != 0 || ts[0].URL != "https://a.example/" || ts[1].URL != "https://b.example/" {
|
||||
t.Fatalf("targets %+v", ts)
|
||||
}
|
||||
}
|
||||
|
||||
// Cancelling a job no worker has reached: it is cancelled at once, its targets stay pending,
|
||||
// and the same list then starts a new job.
|
||||
func TestCancelQueued(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
defer close(release)
|
||||
fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
|
||||
<-release
|
||||
return Result{Content: "ok"}, nil
|
||||
}
|
||||
r := newRunner(t, fetch, Config{Workers: 1})
|
||||
busy, _, err := r.Submit(context.Background(), bob, dcache.ClassFleet, urls("x")) // occupies the only worker once there is one
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = busy
|
||||
id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !r.Cancel(id) {
|
||||
t.Fatal("cancel returned false")
|
||||
}
|
||||
j, ts, _ := r.Status(id)
|
||||
if j.Status != "cancelled" || ts[0].Status != "pending" || ts[1].Status != "pending" {
|
||||
t.Fatalf("after cancel: %+v %+v", j, ts)
|
||||
}
|
||||
if r.Cancel(id) || r.Cancel("no-such-job") {
|
||||
t.Fatal("cancel of a finished or unknown job must return false")
|
||||
}
|
||||
id2, reused, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a", "b"))
|
||||
if err != nil || reused || id2 == id {
|
||||
t.Fatalf("resubmit after cancel id=%q reused=%v err=%v", id2, reused, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Touch keeps a job alive; Status never does.
|
||||
func TestTouchAndStatusLease(t *testing.T) {
|
||||
db, err := dcache.Open("", 1<<20, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
release := make(chan struct{})
|
||||
fetch := func(context.Context, *auth.Caller, Target) (Result, error) {
|
||||
<-release
|
||||
return Result{Content: "ok"}, nil
|
||||
}
|
||||
r := New(db, fetch, Config{Workers: 1}, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
t.Cleanup(r.Close)
|
||||
defer close(release) // runs before the cleanups
|
||||
id, _, err := r.Submit(context.Background(), alice, dcache.ClassFleet, urls("a"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
time.Sleep(50 * time.Millisecond) // let a worker (task 03) take the target; it then blocks
|
||||
before, _ := db.BatchJob(id)
|
||||
time.Sleep(1100 * time.Millisecond) // the lease is in whole seconds
|
||||
r.Status(id)
|
||||
if j, _ := db.BatchJob(id); j.Lease != before.Lease {
|
||||
t.Fatalf("Status moved the lease %d → %d; a read must not write", before.Lease, j.Lease)
|
||||
}
|
||||
r.Touch(id)
|
||||
if j, _ := db.BatchJob(id); j.Lease <= before.Lease {
|
||||
t.Fatalf("Touch did not move the lease: %d → %d", before.Lease, j.Lease)
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
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")
|
||||
}
|
||||
}
|
||||
@@ -4,8 +4,6 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
@@ -15,46 +13,6 @@ import (
|
||||
"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)))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
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(5 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
j, _, ok := r.Status(id)
|
||||
if ok && j.Status == want {
|
||||
return
|
||||
}
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
}
|
||||
j, _, _ := r.Status(id)
|
||||
t.Fatalf("job %s never reached %q (now %q)", id, want, j.Status)
|
||||
}
|
||||
|
||||
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
|
||||
@@ -235,36 +193,6 @@ func TestResubmitAfterCancelIsNew(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
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{})
|
||||
@@ -443,10 +371,3 @@ func TestHooks(t *testing.T) {
|
||||
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")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
package config
|
||||
|
||||
// P6 task 07: the [batch] block and ParseSize.
|
||||
// P6 task 08: the [batch] block and ParseSize.
|
||||
|
||||
import (
|
||||
"os"
|
||||
|
||||
@@ -1,53 +0,0 @@
|
||||
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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user