amadan

carlos / platform Public

db-region-forward: plan 3, pull and pull --at; plan 1 frees a lost election's reservation at once

Plan 3, ten tasks: measure on dev and size the pull's cap inside the
22-minute bound, the pull rows' review, pull in the wire, the worker's
hold mode (verify, hold, send back after check B), the agent's pull, the
console forwarding it, the CLI's spooled pull with a local restore,
DBF-23, acceptance, and the census and roll.

Plan 1: a read the box admits and then loses in the election started no
worker, so admission.abandon gives its floor back at once instead of
holding it until every worker unit has stopped.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 7e5e06fc484906d18d7b117c2f36a2e22d78cc46 parent 2c3e071
2 files changed, +1525 −3
  • docs/superpowers/plans/2026-10-22-db-region-forward-plan-1-box.md +35 −3
  • docs/superpowers/plans/2026-10-22-db-region-forward-plan-3-pull.md +1490 −0
diff --git a/docs/superpowers/plans/2026-10-22-db-region-forward-plan-1-box.md b/docs/superpowers/plans/2026-10-22-db-region-forward-plan-1-box.md
index 2fc56420a..ee4c6e679 100644
--- a/docs/superpowers/plans/2026-10-22-db-region-forward-plan-1-box.md
+++ b/docs/superpowers/plans/2026-10-22-db-region-forward-plan-1-box.md
@@ -2959,7 +2959,7 @@ git commit -m "carlos system db-worker: the stdin socket and peer checks, Linux
- Create: `internal/dbreads/admission_test.go`
**Interfaces:**
-- Produces: `newAdmission(units func(context.Context) (int, error), free func() (avail, total uint64, err error)) *admission`; methods `admitRead(ctx, id) error`, `grow(id string, need int64) error`, `finish(id string)`, `admitProbe(ctx) bool`, `probeDone()`, `opened() bool`, `closed()`; errors `errBusy`, `errOrphans`, `errNoRoom`.
+- Produces: `newAdmission(units func(context.Context) (int, error), free func() (avail, total uint64, err error)) *admission`; methods `admitRead(ctx, id) error`, `grow(id string, need int64) error`, `finish(id string)`, `abandon(id string)`, `admitProbe(ctx) bool`, `probeDone()`, `opened() bool`, `closed()`; errors `errBusy`, `errOrphans`, `errNoRoom`.
- [ ] **Step 1: Write the failing test**
@@ -3086,6 +3086,28 @@ func TestAdmissionFailsClosedWhenUnitsCannotBeListed(t *testing.T) {
}
}
+// A read admitted and then lost in the election started no worker: its
+// floor goes at once, even while another read's unit is running.
+func TestAdmissionAbandonFreesAtOnce(t *testing.T) {
+ t.Parallel()
+ u := &unitsSeam{}
+ // 17 GiB of 20: room for two floors (17-10 = 7), not three.
+ a := newAdmission(u.count, func() (uint64, uint64, error) { return gib(17), gib(20), nil })
+ ctx := context.Background()
+ if err := a.admitRead(ctx, "running"); err != nil {
+ t.Fatal(err)
+ }
+ a.opened()
+ u.n = 1 // its worker unit is up, so finished reservations stay held
+ if err := a.admitRead(ctx, "lost"); err != nil {
+ t.Fatal(err)
+ }
+ a.abandon("lost") // it lost the election; no worker started
+ if err := a.admitRead(ctx, "next"); err != nil {
+ t.Fatalf("the lost read's floor is still held: %v", err)
+ }
+}
+
func TestAdmissionFailsClosedWhenTheDiskCannotBeRead(t *testing.T) {
t.Parallel()
a := newAdmission((&unitsSeam{}).count, func() (uint64, uint64, error) { return 0, 0, errors.New("statfs") })
@@ -3216,6 +3238,16 @@ func (a *admission) finish(id string) {
delete(a.running, id)
}
+// abandon gives back a read this agent admitted and then did not run (it
+// lost the election, or stopped before it): no worker was started for it,
+// so its reservation goes at once rather than waiting for the units to drain.
+func (a *admission) abandon(id string) {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ delete(a.running, id)
+ delete(a.held, id)
+}
+
// admitProbe admits the readiness probe: no orphan and no other probe. The
// probe's connection is the third (opened checks the count).
func (a *admission) admitProbe(ctx context.Context) bool {
@@ -3983,7 +4015,7 @@ func (a *Agent) consider(ctx context.Context, e QueueEntry) {
if d := a.cfg.Jitter(); d > 0 {
select {
case <-ctx.Done():
- a.adm.finish(e.ID)
+ a.adm.abandon(e.ID)
return
case <-time.After(d):
}
@@ -3991,7 +4023,7 @@ func (a *Agent) consider(ctx context.Context, e QueueEntry) {
p, owned := a.elect(ctx, e.ID)
a.decide(e.ID)
if !owned {
- a.adm.finish(e.ID)
+ a.adm.abandon(e.ID)
return
}
a.reads.Add(1)
diff --git a/docs/superpowers/plans/2026-10-22-db-region-forward-plan-3-pull.md b/docs/superpowers/plans/2026-10-22-db-region-forward-plan-3-pull.md
new file mode 100644
index 000000000..387c3c2fc
--- /dev/null
+++ b/docs/superpowers/plans/2026-10-22-db-region-forward-plan-3-pull.md
@@ -0,0 +1,1490 @@
+# `carlos db` region forward, plan 3: `pull` and `pull --at` Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Let a member run `carlos db pull --out <path>` and `carlos db pull --at <time> --out <path>` against an instance whose data lives in another region: the box ships the planned LTX files, sealed, inside the same single request, and the CLI checks them, restores them locally with today's code and saves the copy, exactly as an in-region pull does.
+
+**Architecture:** Plans 1 and 2's path with `op: "pull"`. The box never restores for a pull: its worker receives the planned files, verifies each one's LTX checksums and holds them in its private scratch; after check B the agent asks for them back and seals them as the `ltx` stream in larger chunks. The sealed summary lists the files in order. The console forwards the new op and audits it. The CLI spools the sealed chunks to a private directory rather than memory, checks the whole answer against the summary, writes the files as a litestream file replica, restores them with `dbdoor.RestoreLocal` (today's restore and `lease.CheckDBContext`), and hands the copy to today's `copyOut` and pull lines. The pull's size cap and chunk size are chosen from a measurement on dev before anything is built (Task 1), inside the spec's 22-minute request bound.
+
+**Tech Stack:** Go 1.26, plans 1 and 2's `internal/dbreads`, `internal/dbdoor`, console and CLI code, superfly/ltx's decoder (`Verify`), litestream's file replica client.
+
+**Spec:** `docs/superpowers/specs/2026-10-22-db-region-forward-design.md` (rev 5; Paul approved building it on 2026-10-22 with the rulings in §16, among them ruling 1: pull goes through the deployment bucket as designed, with the true storage statement, and a per-region bucket is plan 5). Read §2, §5.5 to §5.7, §6, §14 and §16. Plans 1 and 2 must have merged and plan 2 must have rolled to dev.
+
+**Branch:** `db-region-forward-3-pull`, cut from amadan `main` after plan 2 merged.
+
+## Global Constraints
+
+Every task's requirements include this section.
+
+- **Production-wired acceptance.** As plans 1 and 2: the real console server, the real CLI binary through `carlosTestEnv` with a scratch `cmd.Dir`, real stores (MinIO in CI), real verify, seal and restore code. The in-process box (`dbreadstest`) is not acceptance for the units; plan 1's root-runner tests are, and Task 4 extends them for the pull's worker mode. Scripted boxes are stand-ins and say so.
+- **Targeted local runs; CI runs the full suite.** Never `make test` or a whole console, host or root package locally; build, vet, gofmt and the cross-compiles run locally.
+- **Cross-compile gate.** `GOOS=darwin GOARCH=arm64 go build ./...` and `GOOS=windows GOARCH=amd64 go build ./...` before every push. The CLI's pull path is portable.
+- **Copy needs review before it ships, with no em dashes,** lower case after a full stop (ruling 4). Task 2 reviews the pull rows; no pull string reaches code first.
+- **No tenant names.** `shop`, `shop.example.com`, `far.example.com`, `member@example.com`, regions `ie` and `uk`.
+- **Go cache** at its defaults; the sandbox off for go commands if the cache is read-only, and for socket tests and binary builds.
+- **Census overlays per branch, loaded LAST** after whatever main loads last at the merge (plan 2's `db-region-forward-2` at the earliest), previous seal first; rows `rev3=PROPOSED-db-region-forward-3`; ruling asked on #280.
+- **Merge main first** before review and before merging; merge only behind `git merge-base --is-ancestor origin/main HEAD`.
+- **Test weights** for new console or root tests of a second or more.
+- **No CI-only scripts under `scripts/`;** MinIO additions go in `.amadan/ci.d/30-minio-contract`.
+- **New host tests call `t.Parallel()`** unless they touch env, cwd, umask or package vars.
+- **Scratch** under `$TMPDIR` or `t.TempDir()`, removed by name. The CLI's own spool directory is under `os.TempDir()`, 0700, removed on every path, as today's restore directory is.
+- **Ask before anything large.** A test or measurement that writes more than a few GB (a full-size pull) runs only on Paul's say-so (machine rule 5, the CPU budget note).
+- **Wire shapes are additive.** The forward marker gains `"pull"` in `forward_ops`; a CLI from plan 2 still prints DBF-8 for a pull.
+- **Spec numbers, verbatim,** as plans 1 and 2, and: the request at most 22 minutes, the CLI's context 23; the pull's cap and chunk size are Task 1's, recorded in the spec before use.
+
+## Review Focus
+
+The five input classes the spec implies but no gate test names, most likely first. Each has its test in the owning task.
+
+1. **The member's disk fills mid-pull.** A pull spools up to the cap of ciphertext, then writes the files and the restored copy: about three times the cap under `os.TempDir()`. A person expects a clean failure, nothing at `--out`, and the spool directory gone. Test: Task 7, `TestForwardedPullLeavesNothingWhenTheSpoolFails`.
+2. **A planned file that is corrupt in the bucket.** A person expects the box's worker to refuse it (`check-failed`, today's sentence), never to ship it, and never a half-written `--out`. Test: Task 4, `TestWorkerHoldRefusesACorruptFile`.
+3. **Files that do not add up.** A summary whose file sizes do not sum to its byte count, or a file list that does not end at the summary's transaction. A person expects DBF-13 and nothing at `--out`. Test: Task 7, `TestForwardedPullRefusesFilesThatDoNotAddUp`.
+4. **Ctrl-C during a long pull.** A person expects `stopped`, nothing at `--out`, and the spool directory gone. Test: Task 7, `TestForwardedPullCtrlCLeavesNothing`.
+5. **A box whose cap is raised past the CLI's bound.** `CARLOS_DB_PULL_MAX_BYTES` above the built-in cap would ship more than any CLI accepts. A person expects the box to clamp to the built-in cap. Test: Task 5, `TestPullCapOnlyLowers`.
+
+## Census impact
+
+- **Plan 14 root-read census:** one overlay, `plan14-root-reads-db-region-forward-3.tsv`, `rev3=PROPOSED-db-region-forward-3`: the worker's pull units (receive with LTX verification, send back) at J4 as the db-read worker; the agent's changed read units re-frozen at plan 1's dispositions with the pull appended; the console's changed reads-route units re-frozen at plan 2's; new CLI units for the pull (the spool directory's writes and reads, the file replica it writes, the restore) at J4 as the carlos CLI; envelopes `.`, `internal/dbreads`, `internal/console`, `internal/host` (for the root-runner dispatch, tests only, so only if the census names it). The seal moves.
+- **9B8 record-write census:** no new write site is expected (the pull reuses plan 1's chunk writer and plan 2's request writes); a changed digest of an existing `nonrecord` or `key-site` entry is re-pinned at its class.
+- **Claim routes:** unchanged.
+- **go.mod:** unchanged.
+
+## Copy strings
+
+Drafted here for Task 2's review; none reaches code before Paul approves it:
+
+| id | file | text |
+|---|---|---|
+| DBF-17 | dbforward.go | <host>'s database is <size> of replica files. carlos db pull can fetch at most <cap> from region <region>. |
+| DBF-18 | dbforward.go | fetching a copy of <host>'s database from region <region>. |
+| DBF-19 | internal/console/apidbforward.go | forwarded a database pull to region <region> (cli) |
+| DBF-20 | internal/console/apidbforward.go | forwarded a database pull to region <region>, as of <time> (cli) |
+| DBF-23 | docs/site/cli.md | If the app has more than one instance, name one with `--instance`. Only instances that sleep have a replica. An instance on your own fleet or in a pool cannot be read this way, and the command says which.\n\nWhen an instance keeps its data in another region than the console's, `carlos db query` runs in that region. A server there restores the copy, runs your query and sends back only the rows. `carlos db pull` fetches the copy's replica files from there and restores them on your machine. Both are encrypted for your terminal on the way, so nothing in between can read them, and nothing is printed or saved until all of it has arrived and checked out. There, a query's copy can be at most 2 GB once restored, the query runs for at most 5 minutes, and the rows can come to at most 64 MB. A pull can fetch at most <pull cap> of replica files. |
+| DBF-24 | dbforward.go | the server in region <region> does not have room to prepare a copy of <host>'s database right now. try again later. |
+
+`<pull cap>` in DBF-23 is Task 1's number, written in before the review (for example `2 GB`). DBF-23 replaces DBF-30 in the CLI page; DBF-30 stays in the record, approved, and leaves the built list. DBF-8 stays built: a console from plan 2 does not forward pulls, and a new CLI still says DBF-8 for it. Reused unchanged: today's three pull lines ("Saved a copy of ...", `dbPullHolds`, "It is a copy. ..."), the `dbRestoreError` sentences, `stopped`, and plan 2's DBF-3, DBF-12, DBF-13, DBF-15 and DBF-16.
+
+## File structure
+
+| File | Responsibility |
+|---|---|
+| `internal/dbreads/limits.go`, `envelope.go`, `lines.go`, `chunks.go` (modify) | `MaxPullBytes`, `PullChunkBytes`, `MaxLineBytes`; `pull` accepted; the chunk size a parameter. |
+| `internal/dbreads/worker.go` (modify) | The worker's hold mode: receive, verify, hold, send back. |
+| `internal/dbreads/read.go`, `agent.go` (modify) | The agent's pull: the cap, the reservation, the `ltx` stream, the summary's files. |
+| `internal/host/dbreads.go` (modify) | `CARLOS_DB_PULL_MAX_BYTES`, lowering only. |
+| `internal/console/apidbforward.go` (modify) | `forward_ops` gains `pull`; the route admits it; DBF-19, DBF-20. |
+| `dbforward.go`, `db.go` (modify) | The CLI's spool, its pull opener, `dbPullForward`; DBF-17, DBF-18, DBF-24. |
+| `docs/site/cli.md` (modify) | DBF-23. |
+| Tests: `internal/dbreads/pull_test.go`, `internal/console/apidbforward_test.go`, `internal/console/dbforward_minio_test.go`, `dbforward_test.go`, `dbforward_e2e_test.go`, `internal/host/dbreads_rootrunner_linux_test.go` (create or modify) | The pull's unit, acceptance and root-runner tests. |
+
+---
+
+### Task 1: Measure, then choose the pull's cap and chunk size
+
+Spec §14: "A 2 GiB pull in one request is the case most likely to need a longer request bound; plan 3 measures it before choosing." The choice rule here keeps the request bound (22 minutes, J1) and sizes the pull to fit it, because a longer request would reopen every timing argument of spec §5.2.
+
+**Files:**
+- Modify: `docs/superpowers/specs/2026-10-22-db-region-forward-design.md` (§5.7's limits table and a short rev 6 note)
+
+- [ ] **Step 1: Measure the forwarded path's throughput on dev**
+
+On dev, after plan 2 has rolled and `CARLOS_DB_FORWARD=1` is set on the dev console, from a laptop signed in to it, against an instance in dev's uk region:
+
+```bash
+time carlos db query --app <dev app> --instance <its uk instance> \
+ --sql "WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x+1 FROM c LIMIT 32000) SELECT hex(randomblob(1000)) FROM c" > "$TMPDIR/dbf-measure.out"
+wc -c "$TMPDIR/dbf-measure.out"; rm "$TMPDIR/dbf-measure.out"
+```
+
+That is about 64 MB of rows (the output cap; the command ends with DBF-6, which is expected). Run it three times. From the box's journal (`journalctl -u carlos-edge` on the uk box, read through the operator's usual access) take the time between the read's `picked` and its result write, and subtract the restore and query time the worker reports; from the console's journal take the same read's request duration. The relay rate R is the output bytes over the slower of the two streaming times, in MB/s. Record all three runs.
+
+- [ ] **Step 2: Apply the rule**
+
+- The pull's transfer must fit the owner's deadline with room for planning: `cap = floor(R × 15 minutes / 256 MiB) × 256 MiB`, at most 2 GiB (ruling 2's restored-size cap: a pull never needs more replica bytes than the database it restores, for a compacted replica).
+- The chunk size: 1 MiB (`PullChunkBytes`), so a 2 GiB pull is 2,048 objects, not 8,192. Each object is a PUT on the box and a GET on the console, and their per-request latency, not their bytes, bounds a small chunk's rate; a larger chunk only helps. Keep 1 MiB.
+- `dbreads.MaxLineBytes` becomes 2 MiB, which a 1 MiB chunk's sealed, base64 line (about 1.4 MB) fits.
+
+- [ ] **Step 3: Record it in the spec**
+
+In §5.7's limits table add, under "Output":
+
+```markdown
+| Pull | at most <cap> of planned LTX files (measured on dev, <date>: relay rate <R> MB/s); `CARLOS_DB_PULL_MAX_BYTES` on a box can only lower it; the read ends `too-large` | agent, from the plan, before any byte moves; the CLI's spool is bounded by the same number |
+| Pull chunk | 1 MiB or 1 s, whichever comes first | agent |
+```
+
+and a line in the status header: "Rev 6 (plan 3, <date>): the pull's cap and chunk size, from a measurement on dev; the request bound is unchanged." Ask Paul to confirm the cap as a limit (ruling 2 covers the others), and write his answer into §16's ruling 2. Commit:
+
+```bash
+git add docs/superpowers/specs/2026-10-22-db-region-forward-design.md
+git commit -m "db-region-forward spec rev 6: the pull's cap and chunk size, measured on dev"
+```
+
+Every later task uses the recorded numbers. Where this plan writes `2 << 30` for the cap, use Task 1's value if it is lower.
+
+---
+
+### Task 2: The pull rows: draft, review, record
+
+**Files:**
+- Modify: `docs/superpowers/plans/2026-10-22-db-region-forward-copy.md`
+- Modify: `approved_db_forward_copy_test.go`
+
+- [ ] **Step 1: Add the rows, pending**
+
+Add a "Pull (plan 3)" table to the record with DBF-17, DBF-18, DBF-19, DBF-20, DBF-23 and DBF-24 as drafted in this plan's copy strings, each `pending`, with Task 1's cap written into DBF-23. In the gate, extend `dbForwardCopyIDs`:
+
+```go
+var dbForwardCopyIDs = func() []string {
+ ids := make([]string, 0, 26)
+ for i := 1; i <= 24; i++ {
+ if i == 21 || i == 22 {
+ continue // appended below, in the record's order
+ }
+ ids = append(ids, "DBF-"+strconv.Itoa(i))
+ }
+ return append(ids, "DBF-21", "DBF-22", "DBF-30", "DBF-31")
+}()
+```
+
+The list is DBF-1 to DBF-20, DBF-23, DBF-24, then DBF-21, DBF-22, DBF-30 and DBF-31; `checkMovesInventory` compares the sets. Run `go test . -run 'TestDBForwardCopy' -count=1`. Expected: PASS.
+
+- [ ] **Step 2: Review**
+
+Invoke the `copy-review` skill with the six rows, with context (DBF-17: a pull refused as too big; DBF-18: stderr while a pull waits; DBF-19, DBF-20: the audit row; DBF-23: the CLI page's paragraph, replacing DBF-30; DBF-24: a pull refused for want of disk on the box). Apply Paul's text verbatim, mark the rows approved, and run the gate again.
+
+- [ ] **Step 3: Commit**
+
+```bash
+git add docs/superpowers/plans/2026-10-22-db-region-forward-copy.md approved_db_forward_copy_test.go
+git commit -m "db forward: the pull rows, approved"
+```
+
+---
+
+### Task 3: The wire: `pull` in the envelope, the pull's limits, the chunk size
+
+**Files:**
+- Modify: `internal/dbreads/limits.go`, `internal/dbreads/envelope.go:39` (`ops`) and `valid()`, `internal/dbreads/lines.go` (`MaxLineBytes`), `internal/dbreads/chunks.go` (`newChunkWriter` takes a chunk size), `internal/dbreads/read.go` (its one `newChunkWriter` call)
+- Modify: `internal/dbreads/envelope_test.go`, `internal/dbreads/read_test.go` (the `newChunkWriter` call in `TestChunkWriterCapIsExact`)
+
+**Interfaces:**
+- Produces: `dbreads.MaxPullBytes int64 = 2 << 30` (or Task 1's cap), `dbreads.PullChunkBytes = 1 << 20`, `dbreads.MaxLineBytes = 2 << 20`; `newChunkWriter(ctx, s, id, pub, stream string, every time.Duration, max int64, size int) *chunkWriter`; `Request.valid` accepts `op: "pull"` with an empty `sql`.
+
+- [ ] **Step 1: Write the failing tests**
+
+Append to `internal/dbreads/envelope_test.go`:
+
+```go
+func TestEnvelopeCarriesAPull(t *testing.T) {
+ t.Parallel()
+ pub, priv := rootKeys(t)
+ req := testRequest(t)
+ req.Op, req.SQL = OpPull, ""
+ wire, err := Sign(release.RootSigner(priv), req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got, err := Verify(pub, wire, req.ID, "uk", testNow); err != nil || got.Op != OpPull {
+ t.Fatalf("%+v %v", got, err)
+ }
+ req.SQL = "SELECT 1"
+ if _, err := Sign(release.RootSigner(priv), req); !errors.Is(err, ErrEnvelope) {
+ t.Fatalf("a pull carrying SQL signed: %v", err)
+ }
+}
+```
+
+and to `internal/dbreads/lines_test.go`:
+
+```go
+func TestAFullPullChunkLineFits(t *testing.T) {
+ t.Parallel()
+ full, _ := json.Marshal(Line{Chunk: 1, Data: bytes.Repeat([]byte{0xff}, PullChunkBytes+SealOverhead)})
+ if len(full)+1 > MaxLineBytes {
+ t.Fatalf("a full pull chunk line is %d bytes", len(full))
+ }
+}
+```
+
+- [ ] **Step 2: Run them to verify they fail**
+
+Run: `go test ./internal/dbreads/ -run 'TestEnvelopeCarriesAPull|TestAFullPullChunkLineFits' -count=1`
+Expected: FAIL (`PullChunkBytes` undefined; a pull refused).
+
+- [ ] **Step 3: Make the changes**
+
+In `limits.go` add:
+
+```go
+ // MaxPullBytes is a pull's cap on its planned LTX files (spec §5.7, rev
+ // 6). CARLOS_DB_PULL_MAX_BYTES on a box can only lower it, so no box
+ // ships more than a CLI accepts.
+ MaxPullBytes int64 = 2 << 30
+ // PullChunkBytes is a pull's chunk: larger than a query's, so a full
+ // pull is about two thousand objects.
+ PullChunkBytes = 1 << 20
+```
+
+In `lines.go`, `MaxLineBytes = 2 << 20` with its comment saying a pull chunk's line is about 1.4 MB. In `envelope.go`, `var ops = map[string]bool{OpQuery: true, OpPull: true}` and in `valid()`:
+
+```go
+ switch r.Op {
+ case OpQuery:
+ if r.SQL == "" || len(r.SQL) > MaxSQLBytes || !utf8.ValidString(r.SQL) {
+ return false
+ }
+ case OpPull:
+ if r.SQL != "" {
+ return false
+ }
+ }
+```
+
+replacing the query-only SQL check. In `chunks.go`, add a `size int` field and parameter to `chunkWriter`/`newChunkWriter` and use `c.size` where it used `ChunkBytes`; plan 1's query call passes `ChunkBytes`, and `TestChunkWriterCapIsExact` passes `ChunkBytes`.
+
+- [ ] **Step 4: Run the package**
+
+Run: `go test ./internal/dbreads/ -count=1 -race`
+Expected: PASS.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add internal/dbreads
+git commit -m "dbreads: pull in the envelope, the pull's cap and chunk size"
+```
+
+---
+
+### Task 4: The worker's hold mode
+
+**Files:**
+- Modify: `internal/dbreads/worker.go`
+- Create: `internal/dbreads/pull_test.go` (this task's tests)
+- Modify: `internal/host/dbreads_rootrunner_linux_test.go` (a pull through the installed units)
+
+**Interfaces:**
+- Consumes: `receiveFile`, `frameWriter`, the frames (plan 1 Task 7); `ltx.NewDecoder`, `(*ltx.Decoder).Verify`.
+- Produces: `ServeWorker` serves `planHeader{Op: OpPull}`: receive every file, verify it, answer `ready`, then on `run` send every file back in the header's order as output frames and end; on `stop`, nothing.
+
+- [ ] **Step 1: Write the failing tests**
+
+Create `internal/dbreads/pull_test.go` (net.Pipe to the real worker code: not acceptance):
+
+```go
+package dbreads
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "io"
+ "testing"
+ "time"
+
+ lsfile "github.com/benbjohnson/litestream/file"
+ "github.com/superfly/ltx"
+
+ "github.com/carlosframework/platform/internal/dbdoor"
+ "github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
+)
+
+// sendPullPlan sends a pull header and the planned files; it returns the
+// files' bytes concatenated, which is what the worker must send back.
+func sendPullPlan(t *testing.T, w io.Writer, dir string, corrupt bool) (dbdoor.RestorePlan, []byte) {
+ t.Helper()
+ rd := lsfile.NewReplicaClient(dir)
+ plan, err := dbdoor.PlanFrom(context.Background(), rd, time.Time{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := writeJSONFrame(w, frameHeader, planHeader{V: ProtocolVersion, Op: OpPull, Files: plan.Planned, Target: plan.TXID, MaxRestored: DefaultMaxRestored}); err != nil {
+ t.Fatal(err)
+ }
+ var all bytes.Buffer
+ for i, f := range plan.Planned {
+ rc, _ := rd.OpenLTXFile(context.Background(), f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0)
+ b, _ := io.ReadAll(rc)
+ rc.Close()
+ if corrupt && i == 0 {
+ b[len(b)/2] ^= 0xff
+ }
+ all.Write(b)
+ writeFrame(w, frameFile, b)
+ }
+ return plan, all.Bytes()
+}
+
+func TestWorkerHoldsThenSendsThePlannedFiles(t *testing.T) {
+ t.Parallel()
+ dir := t.TempDir()
+ dbdoortest.Replicate(t, dir, 3)
+ conn, done := workerPipe(t, WorkerOptions{})
+ plan, want := sendPullPlan(t, conn, dir, false)
+ var ready readyMsg
+ readTyped(t, conn, frameReady, &ready)
+ if ready.TXID != plan.TXID {
+ t.Fatalf("ready at %d", ready.TXID)
+ }
+ writeJSONFrame(conn, frameRun, runMsg{})
+ var got bytes.Buffer
+ for {
+ typ, p, err := readFrame(conn)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if typ == frameOutput {
+ got.Write(p)
+ continue
+ }
+ var end endMsg
+ if typ != frameEnd || json.Unmarshal(p, &end) != nil || end.Code != "" {
+ t.Fatalf("end %q %s", typ, p)
+ }
+ break
+ }
+ if !bytes.Equal(got.Bytes(), want) {
+ t.Fatalf("sent back %d bytes, held %d", got.Len(), len(want))
+ }
+ if err := <-done; err != nil {
+ t.Fatal(err)
+ }
+}
+
+// Review Focus 2: a corrupt file is refused before ready, never shipped.
+func TestWorkerHoldRefusesACorruptFile(t *testing.T) {
+ t.Parallel()
+ dir := t.TempDir()
+ dbdoortest.Replicate(t, dir, 2)
+ conn, done := workerPipe(t, WorkerOptions{})
+ sendPullPlan(t, conn, dir, true)
+ var end endMsg
+ readTyped(t, conn, frameEnd, &end)
+ if end.Code != CodeCheckFailed {
+ t.Fatalf("end = %+v", end)
+ }
+ <-done
+}
+
+func TestWorkerHoldStopsWithoutSending(t *testing.T) {
+ t.Parallel()
+ dir := t.TempDir()
+ dbdoortest.Replicate(t, dir, 1)
+ conn, done := workerPipe(t, WorkerOptions{})
+ sendPullPlan(t, conn, dir, false)
+ readTyped(t, conn, frameReady, nil)
+ writeFrame(conn, frameStop, nil)
+ if err := <-done; err != nil {
+ t.Fatal(err)
+ }
+}
+```
+
+- [ ] **Step 2: Run them to verify they fail**
+
+Run: `go test ./internal/dbreads/ -run 'TestWorkerHold' -count=1`
+Expected: FAIL: the worker refuses `op: pull` as a protocol error.
+
+- [ ] **Step 3: Write the hold mode**
+
+In `ServeWorker`, accept both ops (`h.Op != OpQuery && h.Op != OpPull` is the protocol error), and after the files are received branch before the restore:
+
+```go
+ if h.Op == OpPull {
+ return holdAndSend(ctx, conn, next, end, root, h)
+ }
+```
+
+and add:
+
+```go
+// holdAndSend is a pull's whole worker job (spec §6: the box never restores
+// for a pull). Each planned file's LTX checksums are verified (the decode
+// is the worker's, as every decode of tenant bytes is), the files are held
+// in this unit's private scratch, and after the agent's check B they go
+// back in the header's order, whole, as output.
+func holdAndSend(ctx context.Context, conn io.Writer, next func() (frame, error), end func(string, error) error, root string, h planHeader) error {
+ paths := make([]string, len(h.Files))
+ for i, lf := range h.Files {
+ paths[i] = litestream.LTXFilePath(root, lf.Level, ltx.TXID(lf.MinTXID), ltx.TXID(lf.MaxTXID))
+ if err := verifyLTX(paths[i]); err != nil {
+ return end(CodeCheckFailed, err)
+ }
+ }
+ if len(h.Files) == 0 || h.Files[len(h.Files)-1].MaxTXID != h.Target {
+ return end(CodeCheckFailed, errors.New("the plan does not end at its target"))
+ }
+ if err := writeJSONFrame(conn, frameReady, readyMsg{TXID: h.Target}); err != nil {
+ return err
+ }
+ f, err := next()
+ if err != nil {
+ return err
+ }
+ switch f.typ {
+ case frameStop:
+ return nil
+ case frameRun:
+ default:
+ return errProtocol
+ }
+ out := &frameWriter{w: conn}
+ for _, p := range paths {
+ if ctx.Err() != nil {
+ return errAgentGone
+ }
+ in, err := os.Open(p)
+ if err != nil {
+ return end(CodeCheckFailed, err)
+ }
+ _, err = io.Copy(out, in)
+ in.Close()
+ if err != nil {
+ return err
+ }
+ }
+ if err := out.flush(); err != nil {
+ return err
+ }
+ return end("", nil)
+}
+
+// verifyLTX decodes one held file whole and checks its checksums.
+func verifyLTX(path string) error {
+ f, err := os.Open(path)
+ if err != nil {
+ return err
+ }
+ defer f.Close()
+ return ltx.NewDecoder(f).Verify()
+}
+```
+
+- [ ] **Step 4: Add the pull to the root-runner tests**
+
+In `internal/host/dbreads_rootrunner_linux_test.go`, add `TestRootDBReadPullThroughTheInstalledUnits`: as `TestRootDBReadServesThroughTheInstalledUnits`, but `r.submit` builds a pull request (`Op: dbreads.OpPull`, empty `SQL`; give `submit` an `op` parameter and pass `dbreads.OpQuery` at every existing call), and the answer is checked by opening every chunk with `dbreads.StreamLTX`, comparing the concatenation's SHA-256 with the summary's, and comparing it byte for byte with the planned files read from `r.dir/replica/instances/shop.example.com` in plan order. After it, `privateTmpGone(t, 10*time.Second)`.
+
+- [ ] **Step 5: Run the unit tests and build the root-runner binary**
+
+Run (sandbox off): `go test ./internal/dbreads/ -run 'TestWorker' -count=1 -race && GOOS=linux GOARCH=arm64 go test -c -tags rootrunner -o "$TMPDIR/dbf3-rr.test" ./internal/host && rm "$TMPDIR/dbf3-rr.test"`
+Expected: PASS and a clean build. Run `TestRootDBReadPullThroughTheInstalledUnits` on the root runner (plan 1 Task 15 Step 4's command) before the branch is reviewed, and record the run id.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add internal/dbreads internal/host/dbreads_rootrunner_linux_test.go
+git commit -m "dbreads: the worker's hold mode for a pull: verify, hold, send back after check B"
+```
+
+---
+
+### Task 5: The agent's pull
+
+**Files:**
+- Modify: `internal/dbreads/read.go` (`attempt` branches on the op), `internal/dbreads/agent.go` (`AgentConfig.MaxPull`)
+- Modify: `internal/host/dbreads.go` (`CARLOS_DB_PULL_MAX_BYTES`)
+- Modify: `internal/dbreads/pull_test.go`, `internal/host/dbreads_test.go`
+
+**Interfaces:**
+- Consumes: `attempt`, `newChunkWriter` with a size (Task 3), `relay`, `ended`, the worker's hold mode (Task 4).
+- Produces: `AgentConfig.MaxPull int64` (`MaxPullBytes` when zero or larger); `dbReadsEnv.maxPull`; a pull's result: `done` with chunks of the `ltx` stream and a summary carrying `TXID`, `At`, `Files` (the plan, in order), `Bytes` and `SHA256`; `too-large` with `Restored` (the plan's LTX bytes) and `Cap`; `no-room`.
+
+- [ ] **Step 1: Write the failing tests**
+
+Append to `internal/dbreads/pull_test.go`:
+
+```go
+func (fx *fixture) pullRequest(at string) Request {
+ req := fx.request("", at)
+ req.Op = OpPull
+ return req
+}
+
+// answerPull opens a pull's chunks as the CLI does: the ltx stream.
+func (fx *fixture) answerPull(id string) ([]byte, Summary, Result) {
+ fx.t.Helper()
+ raw, _, err := fx.store.Get(context.Background(), ResultKey(id))
+ if err != nil {
+ fx.t.Fatal(err)
+ }
+ var r Result
+ json.Unmarshal(raw, &r)
+ s, err := OpenSummary(fx.reply, id, r.Sealed)
+ if err != nil {
+ fx.t.Fatal(err)
+ }
+ var out bytes.Buffer
+ for seq := int64(1); seq <= r.LastSeq; seq++ {
+ sealed, _, _ := fx.store.Get(context.Background(), ChunkKey(id, seq))
+ plain, err := OpenChunk(fx.reply, id, seq, StreamLTX, seq == r.LastSeq, sealed)
+ if err != nil {
+ fx.t.Fatalf("chunk %d: %v", seq, err)
+ }
+ out.Write(plain)
+ }
+ return out.Bytes(), s, r
+}
+
+func TestPullShipsThePlannedFiles(t *testing.T) {
+ t.Parallel()
+ fx := newFixture(t, 4)
+ a, _ := fx.agent("uk-1")
+ req := fx.pullRequest("")
+ fx.submit(req)
+ a.Pass(context.Background())
+ a.Wait()
+ got, s, r := fx.answerPull(req.ID)
+ plan, _ := dbdoor.PlanFrom(context.Background(), lsfile.NewReplicaClient(fx.replica), time.Time{})
+ var want bytes.Buffer
+ for _, f := range plan.Planned {
+ rc, _ := lsfile.NewReplicaClient(fx.replica).OpenLTXFile(context.Background(), f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0)
+ io.Copy(&want, rc)
+ rc.Close()
+ }
+ if r.State != StateDone || !bytes.Equal(got, want.Bytes()) || s.TXID != plan.TXID || len(s.Files) != len(plan.Planned) || s.Bytes != plan.LTXBytes {
+ t.Fatalf("result %+v summary %+v: %d bytes, want %d", r, s, len(got), want.Len())
+ }
+ for seq := int64(1); seq < r.LastSeq; seq++ {
+ sealed, _, _ := fx.store.Get(context.Background(), ChunkKey(req.ID, seq))
+ if len(sealed)-SealOverhead > PullChunkBytes {
+ t.Fatalf("chunk %d is %d bytes", seq, len(sealed))
+ }
+ }
+}
+
+func TestPullTooLargeAndNoChunkBeforeCheckB(t *testing.T) {
+ t.Parallel()
+ fx := newFixture(t, 2)
+ a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) { c.MaxPull = 100 })
+ req := fx.pullRequest("")
+ fx.submit(req)
+ a.Pass(context.Background())
+ a.Wait()
+ _, s, r := fx.answerPull(req.ID)
+ if r.Code != CodeTooLarge || s.Cap != 100 || s.Restored <= 100 {
+ t.Fatalf("%+v %+v", r, s)
+ }
+ // T9 for a pull: a record change after the last byte left the bucket.
+ fx2 := newFixture(t, 2)
+ plan, _ := dbdoor.PlanFrom(context.Background(), lsfile.NewReplicaClient(fx2.replica), time.Time{})
+ last := plan.Planned[len(plan.Planned)-1]
+ hook := &stageHook{ReplicaReader: lsfile.NewReplicaClient(fx2.replica), stage: "after the last byte",
+ lastSpan: dbdoor.Span(last.MinTXID, last.MaxTXID), change: func() { r := fx2.record(); r.Region = "ie"; fx2.putRecord(r) }}
+ b, _ := fx2.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) {
+ c.OpenReplica = func(context.Context, string) (dbdoor.ReplicaReader, error) { return hook, nil }
+ })
+ req2 := fx2.pullRequest("")
+ fx2.submit(req2)
+ b.Pass(context.Background())
+ b.Wait()
+ if _, _, r := fx2.answerPull(req2.ID); r.Code != CodeRecordChanged || len(fx2.chunkKeys(req2.ID)) != 0 {
+ t.Fatalf("%+v, chunks %v", r, fx2.chunkKeys(req2.ID))
+ }
+}
+```
+
+Append to `internal/host/dbreads_test.go`:
+
+```go
+// Review Focus 5: the pull's cap can only be lowered on a box.
+func TestPullCapOnlyLowers(t *testing.T) {
+ t.Parallel()
+ pub, _, _ := ed25519.GenerateKey(rand.Reader)
+ base := map[string]string{"CARLOS_HIBERNATE_BUCKET": "d", "CARLOS_REGION": "uk", "CARLOS_RELEASE_PUBKEY": base64.StdEncoding.EncodeToString(pub)}
+ get := func(v string) func(string) string {
+ return func(k string) string {
+ if k == "CARLOS_DB_PULL_MAX_BYTES" {
+ return v
+ }
+ return base[k]
+ }
+ }
+ if env, _ := dbReadsSettings(get("")); env.maxPull != dbreads.MaxPullBytes {
+ t.Fatalf("default %d", env.maxPull)
+ }
+ if env, _ := dbReadsSettings(get("1048576")); env.maxPull != 1<<20 {
+ t.Fatalf("lowered %d", env.maxPull)
+ }
+ if env, _ := dbReadsSettings(get("99999999999999")); env.maxPull != dbreads.MaxPullBytes {
+ t.Fatalf("raised past the built-in cap: %d", env.maxPull)
+ }
+}
+```
+
+- [ ] **Step 2: Run them to verify they fail**
+
+Run (sandbox off): `go test ./internal/dbreads/ -run 'TestPull' -count=1 && go test ./internal/host/ -run 'TestPullCapOnlyLowers' -count=1`
+Expected: FAIL.
+
+- [ ] **Step 3: Write the agent's pull**
+
+In `agent.go`, add `MaxPull int64` to `AgentConfig` (test seam and box setting) and in `NewAgent`:
+
+```go
+ if cfg.MaxPull <= 0 || cfg.MaxPull > MaxPullBytes {
+ cfg.MaxPull = MaxPullBytes
+ }
+```
+
+In `read.go`'s `attempt`, replace plan 1's restored-size check and its `a.adm.grow` call (the two statements right after `PlanFrom`'s error switch) with a branch on the op for the cap and the reservation:
+
+```go
+ pull := req.Op == OpPull
+ switch {
+ case pull && plan.LTXBytes > a.cfg.MaxPull:
+ return fail(CodeTooLarge, Summary{Restored: plan.LTXBytes, Cap: a.cfg.MaxPull})
+ case !pull && plan.RestoredBytes > a.cfg.MaxRestored:
+ return fail(CodeTooLarge, Summary{Restored: plan.RestoredBytes, Cap: a.cfg.MaxRestored})
+ }
+ need := plan.LTXBytes + plan.RestoredBytes
+ if pull {
+ need = plan.LTXBytes // the worker holds the files and restores nothing
+ }
+ if err := a.adm.grow(req.ID, need); err != nil {
+ return fail(CodeNoRoom, Summary{})
+ }
+```
+
+and replace everything from plan 1's `stopPlanBound()` after check B through its `s := Summary{...}` line with this, which chooses the stream, the chunk size, the bound and the summary:
+
+```go
+ stopPlanBound() // the run is bounded by the worker and the owner's deadline
+ stream, size, max := StreamRows, ChunkBytes, a.cfg.MaxOutput
+ if pull {
+ stream, size, max = StreamLTX, PullChunkBytes, plan.LTXBytes
+ }
+ if err := writeJSONFrame(conn, frameRun, runMsg{SQL: req.SQL}); err != nil {
+ return fail(CodeInterrupted, Summary{})
+ }
+ cw := newChunkWriter(ctx, a.cfg.Store, req.ID, pub, stream, a.cfg.ChunkEvery, max, size)
+ endCode, sqlErr, rerr := relay(conn, cw)
+ last, n, sum, werr := cw.close()
+ s := Summary{TXID: plan.TXID, At: plan.At, Bytes: n, SHA256: sum, Error: sqlErr}
+ if pull {
+ s.Files = plan.Planned
+ // A worker that sends more or less than it held is a check failure,
+ // never an output cap.
+ if werr == nil && rerr == nil && endCode == "" && (cw.capped() || n != plan.LTXBytes) {
+ return a.ended(pub, req.ID, StateFailed, CodeCheckFailed, Summary{}, last), false
+ }
+ }
+```
+
+The existing `switch` on `werr`, `cw.capped()`, `rerr` and `endCode` follows unchanged. For a pull, `cw.capped()` is handled above, so its `output-cap` branch is unreachable. The worker's `end` for a pull is `""` or `check-failed`, which the existing frame-end branch before check B already maps.
+
+In `internal/host/dbreads.go`, add `maxPull int64` to `dbReadsEnv`, parse `CARLOS_DB_PULL_MAX_BYTES` like `CARLOS_DB_READ_MAX_BYTES` (a whole number of bytes, else not armed), keep it only when it is below `dbreads.MaxPullBytes`, default `dbreads.MaxPullBytes`, and pass `MaxPull: env.maxPull` in `armDBReads`.
+
+- [ ] **Step 4: Run the tests**
+
+Run (sandbox off): `go test ./internal/dbreads/ -count=1 -race && go test ./internal/host/ -run 'TestDBReads|TestPullCap' -count=1`
+Expected: PASS.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add internal/dbreads internal/host/dbreads.go internal/host/dbreads_test.go
+git commit -m "dbreads: the agent's pull: its cap, its reservation, the ltx stream and the file list"
+```
+
+---
+
+### Task 6: The console forwards a pull
+
+Waits for Task 2 (DBF-19, DBF-20 approved).
+
+**Files:**
+- Modify: `internal/console/apidbforward.go` (`dbResidency`'s `ForwardOps`, `admitDBRead`'s op check, `dbForwardAuditDetail`, two copy constants)
+- Modify: `internal/console/apidbforward_test.go`
+- Modify: `approved_db_forward_copy_test.go` (DBF-19, DBF-20)
+
+**Interfaces:**
+- Produces: `ForwardOps: []string{dbreads.OpQuery, dbreads.OpPull}`; `dbForwardAuditDetail(op, region, sqlText string, at time.Time) string`; `dbfAuditPull`, `dbfAuditPullAt`.
+
+- [ ] **Step 1: Write the failing tests**
+
+In `internal/console/apidbforward_test.go`, change `TestListDoorMarksACapableRegion`'s expected ops to `[]string{dbreads.OpQuery, dbreads.OpPull}`, change `TestDBReadRouteRefusesBeforeAdmitting`'s "an op this console does not forward" case to `op: "shell"`, add a case `"a pull carrying SQL"` (`Op: pull`, `SQL: "SELECT 1"`, 400), and add:
+
+```go
+func TestDBReadPullStreamsTheFiles(t *testing.T) {
+ fx := newDBFFixture(t, true)
+ dir := t.TempDir()
+ dbdoortest.Replicate(t, dir, 3)
+ fx.startBox(t, dir)
+ key, _ := dbreads.NewReplyKey()
+ res := fx.openRead(t, dbfHost, fx.member, dbreads.ReadBody{Op: dbreads.OpPull, ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())})
+ lines := readLines(t, res.Body)
+ res.Body.Close()
+ last := lines[len(lines)-1].Result
+ if last == nil || last.State != dbreads.StateDone {
+ t.Fatalf("lines %+v", lines)
+ }
+ s, err := dbreads.OpenSummary(key, lines[0].Admitted, last.Sealed)
+ if err != nil || len(s.Files) == 0 {
+ t.Fatalf("summary %+v %v", s, err)
+ }
+ evs, _ := fx.st.Events(OpsAccountID, "shop", 10)
+ want := fmt.Sprintf(dbfAuditPull, "uk")
+ for _, e := range evs {
+ if e.Action == "db-read" && e.Detail == want {
+ return
+ }
+ }
+ t.Fatalf("no %q audit row in %+v", want, evs)
+}
+```
+
+- [ ] **Step 2: Run them to verify they fail**
+
+Run: `go test ./internal/console/ -run 'TestListDoorMarks|TestDBReadRouteRefuses|TestDBReadPull' -count=1`
+Expected: FAIL.
+
+- [ ] **Step 3: Make the changes**
+
+In `apidbforward.go`:
+
+```go
+ dbfAuditPull = "forwarded a database pull to region %s (cli)" // DBF-19
+ dbfAuditPullAt = "forwarded a database pull to region %s, as of %s (cli)" // DBF-20
+```
+
+`dbResidency` writes `ForwardOps: []string{dbreads.OpQuery, dbreads.OpPull}`. `admitDBRead`'s op check becomes:
+
+```go
+ switch body.Op {
+ case dbreads.OpQuery:
+ if err := dbquery.CheckSQL(body.SQL); err != nil || body.SQL == "" {
+ msg := http.StatusText(http.StatusBadRequest)
+ if err != nil {
+ msg = err.Error()
+ }
+ jsonError(w, http.StatusBadRequest, msg)
+ return none, "", false
+ }
+ case dbreads.OpPull:
+ if body.SQL != "" {
+ http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+ return none, "", false
+ }
+ default:
+ http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+ return none, "", false
+ }
+```
+
+and `dbForwardAuditDetail` takes the op first:
+
+```go
+func dbForwardAuditDetail(op, region, sqlText string, at time.Time) string {
+ if op == dbreads.OpPull {
+ if at.IsZero() {
+ return fmt.Sprintf(dbfAuditPull, region)
+ }
+ return fmt.Sprintf(dbfAuditPullAt, region, at.UTC().Format(time.RFC3339))
+ }
+ sum := sha256.Sum256([]byte(sqlText))
+ sha := hex.EncodeToString(sum[:])[:16]
+ if at.IsZero() {
+ return fmt.Sprintf(dbfAudit, region, len(sqlText), sha)
+ }
+ return fmt.Sprintf(dbfAuditAt, region, at.UTC().Format(time.RFC3339), len(sqlText), sha)
+}
+```
+
+with its caller passing `body.Op` and `TestDBReadQueuesASignedRequestAndAudits` updated to pass `dbreads.OpQuery`. Add DBF-19 and DBF-20 to `dbForwardCopyBuilt`.
+
+- [ ] **Step 4: Run the tests**
+
+Run: `go test ./internal/console/ -run 'TestDBRead|TestListDoor|TestDBForward' -count=1 && go test . -run 'TestDBForward' -count=1`
+Expected: PASS.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add internal/console approved_db_forward_copy_test.go
+git commit -m "console: forward a pull, and audit it (DBF-19, DBF-20)"
+```
+
+---
+
+### Task 7: The CLI's forwarded pull
+
+Waits for Task 2 (DBF-17, DBF-18, DBF-24 approved).
+
+**Files:**
+- Modify: `dbforward.go` (the chunk sink, `readDBForward` generalised, the pull opener, `dbPullForward`, three constants)
+- Modify: `db.go` (`dbPull` uses the forwarded pull when the marker offers it)
+- Modify: `dbforward_test.go`
+- Modify: `approved_db_forward_copy_test.go` (DBF-17, DBF-18, DBF-24)
+
+**Interfaces:**
+- Consumes: plan 2's `readDBForward`, `openDBForward`, `errDBForward`; `dbdoor.RestoreLocal`, `dbdoor.Restored`, `lsfile.NewReplicaClient`, `litestream.LTXFilePath`; `copyOut`, `dbPullHolds`, `dbRestoreError`, `humanBytes`.
+- Produces: `type dbChunkSink interface { add([]byte) error; count() int; chunk(i int) ([]byte, error) }`; `memSink`, `fileSink{dir string}`; `readDBForward(ctx, body io.Reader, key *ecdh.PrivateKey, host, region, reading string, status io.Writer, sink dbChunkSink, max int64) (string, *dbreads.Result, error)` (the id and the result; `openDBForward` is called by the query path); `openDBPull(ctx, id string, key *ecdh.PrivateKey, sink dbChunkSink, res dbreads.Result, host, region, dir string) (*dbdoor.Restored, error)`; `dbPullForward(ctx context.Context, t dbTarget, fwd dbdoor.Forward) (*dbdoor.Restored, error)`; `dbfPullTooLarge`, `dbfFetching`, `dbfPullNoRoom`.
+
+- [ ] **Step 1: Write the failing tests**
+
+Append to `dbforward_test.go` (crafted streams with real sealing: unit tests; Task 9 runs the binary):
+
+```go
+// pullStream is a pull's body for the files of dir's replica.
+func pullStream(t *testing.T, key *ecdh.PrivateKey, dir string, mutate func(s *dbreads.Summary, files [][]byte)) *stream {
+ t.Helper()
+ rd := lsfile.NewReplicaClient(dir)
+ plan, err := dbdoor.PlanFrom(context.Background(), rd, time.Time{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ var files [][]byte
+ var all bytes.Buffer
+ for _, f := range plan.Planned {
+ rc, _ := rd.OpenLTXFile(context.Background(), f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0)
+ b, _ := io.ReadAll(rc)
+ rc.Close()
+ files = append(files, b)
+ all.Write(b)
+ }
+ sum := sha256.Sum256(all.Bytes())
+ s := dbreads.Summary{TXID: plan.TXID, At: plan.At, Files: plan.Planned, Bytes: int64(all.Len()), SHA256: hex.EncodeToString(sum[:])}
+ if mutate != nil {
+ mutate(&s, files)
+ }
+ st := newStream(t, key)
+ sealed, _ := dbreads.SealChunk(key.PublicKey(), fwdID, 1, dbreads.StreamLTX, true, all.Bytes())
+ st.line(dbreads.Line{Chunk: 1, Data: sealed})
+ st.result(dbreads.StateDone, "", 1, s)
+ return st
+}
+
+func readPull(t *testing.T, st *stream, key *ecdh.PrivateKey) (*dbdoor.Restored, error) {
+ t.Helper()
+ dir := t.TempDir()
+ sink := &fileSink{dir: filepath.Join(dir, "sealed")}
+ id, res, err := readDBForward(context.Background(), &st.buf, key, "shop.example.com", "uk", dbfFetching, io.Discard, sink, dbreads.MaxPullBytes)
+ if err != nil {
+ return nil, err
+ }
+ return openDBPull(context.Background(), id, key, sink, *res, "shop.example.com", "uk", dir)
+}
+
+func TestForwardedPullRestoresTheFiles(t *testing.T) {
+ t.Parallel()
+ dir := t.TempDir()
+ dbdoortest.Replicate(t, dir, 3)
+ key, _ := dbreads.NewReplyKey()
+ r, err := readPull(t, pullStream(t, key, dir, nil), key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ db, _ := sql.Open("sqlite", "file:"+r.Path+"?mode=ro")
+ defer db.Close()
+ var n int
+ if err := db.QueryRow("SELECT count(*) FROM t").Scan(&n); err != nil || n != 30 {
+ t.Fatalf("%d rows, %v", n, err)
+ }
+}
+
+// Review Focus 3: files that do not add up are DBF-13, with nothing
+// restored.
+func TestForwardedPullRefusesFilesThatDoNotAddUp(t *testing.T) {
+ t.Parallel()
+ dir := t.TempDir()
+ dbdoortest.Replicate(t, dir, 3)
+ for name, mutate := range map[string]func(s *dbreads.Summary, files [][]byte){
+ "sizes that do not sum": func(s *dbreads.Summary, _ [][]byte) { s.Files[0].Size++ },
+ "a list that ends elsewhere": func(s *dbreads.Summary, _ [][]byte) { s.TXID++ },
+ "no files": func(s *dbreads.Summary, _ [][]byte) { s.Files = nil },
+ } {
+ key, _ := dbreads.NewReplyKey()
+ if _, err := readPull(t, pullStream(t, key, dir, mutate), key); err == nil || err.Error() != fmt.Sprintf(dbfUnchecked, "uk") {
+ t.Errorf("%s: %v", name, err)
+ }
+ }
+}
+
+// Review Focus 1: a spool that cannot be written fails cleanly and leaves
+// nothing behind.
+func TestForwardedPullLeavesNothingWhenTheSpoolFails(t *testing.T) {
+ t.Parallel()
+ dir := t.TempDir()
+ dbdoortest.Replicate(t, dir, 1)
+ key, _ := dbreads.NewReplyKey()
+ st := pullStream(t, key, dir, nil)
+ blocked := filepath.Join(t.TempDir(), "sealed")
+ if err := os.WriteFile(blocked, nil, 0o600); err != nil { // a file where the spool's directory should be
+ t.Fatal(err)
+ }
+ _, _, err := readDBForward(context.Background(), &st.buf, key, "shop.example.com", "uk", dbfFetching, io.Discard, &fileSink{dir: blocked}, dbreads.MaxPullBytes)
+ if err == nil {
+ t.Fatal("a spool that cannot be written was accepted")
+ }
+}
+
+// Review Focus 4: Ctrl-C during a pull is "stopped", and dbPullForward's
+// spool directory is removed on that path (its deferred removal; checked
+// end to end in Task 9's binary test).
+func TestForwardedPullCtrlCLeavesNothing(t *testing.T) {
+ t.Parallel()
+ key, _ := dbreads.NewReplyKey()
+ st := newStream(t, key)
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ _, _, err := readDBForward(ctx, &cutReader{r: &st.buf}, key, "shop.example.com", "uk", dbfFetching, io.Discard, &memSink{}, dbreads.MaxPullBytes)
+ if err == nil || err.Error() != "stopped" {
+ t.Fatalf("%v", err)
+ }
+}
+```
+
+Update plan 2's query tests to the generalised signature: `readFwd` becomes
+
+```go
+func readFwd(t *testing.T, s *stream, key *ecdh.PrivateKey) (dbForwardAnswer, string, error) {
+ t.Helper()
+ var status bytes.Buffer
+ sink := &memSink{}
+ id, res, err := readDBForward(context.Background(), &s.buf, key, "shop.example.com", "uk", dbfReading, &status, sink, dbreads.MaxOutputBytes)
+ if err != nil {
+ return dbForwardAnswer{}, status.String(), err
+ }
+ ans, err := openDBForward(id, key, sink, *res, "shop.example.com", "uk")
+ return ans, status.String(), err
+}
+```
+
+and `TestForwardedQueryCtrlCPrintsNothing` calls `readDBForward` with `dbfReading, &status, &memSink{}, dbreads.MaxOutputBytes` and checks only its error. Add imports `"database/sql"`, `"io"`, `"path/filepath"`, `lsfile`, `ltx`, `dbdoortest`.
+
+- [ ] **Step 2: Run them to verify they fail**
+
+Run: `go test . -run 'TestForwardedPull|TestForwardedQuery' -count=1`
+Expected: FAIL: undefined names.
+
+- [ ] **Step 3: Write the code**
+
+In `dbforward.go` add the constants:
+
+```go
+ dbfPullTooLarge = "%s's database is %s of replica files. carlos db pull can fetch at most %s from region %s." // DBF-17
+ dbfFetching = "fetching a copy of %s's database from region %s." // DBF-18
+ dbfPullNoRoom = "the server in region %s does not have room to prepare a copy of %s's database right now. try again later." // DBF-24
+```
+
+the sinks:
+
+```go
+// dbChunkSink holds a read's sealed chunks until its result arrives: in
+// memory for a query (at most 64 MiB), in the pull's private directory for a
+// pull (up to its cap).
+type dbChunkSink interface {
+ add(sealed []byte) error
+ count() int
+ chunk(i int) ([]byte, error) // 0-based
+}
+
+type memSink struct{ chunks [][]byte }
+
+func (m *memSink) add(b []byte) error { m.chunks = append(m.chunks, b); return nil }
+func (m *memSink) count() int { return len(m.chunks) }
+func (m *memSink) chunk(i int) ([]byte, error) { return m.chunks[i], nil }
+
+// fileSink spools sealed chunks to dir, one file each, 0600 in a 0700
+// directory the caller removes.
+type fileSink struct {
+ dir string
+ n int
+}
+
+func (f *fileSink) add(b []byte) error {
+ if f.n == 0 {
+ if err := os.MkdirAll(f.dir, 0o700); err != nil {
+ return err
+ }
+ }
+ f.n++
+ return os.WriteFile(filepath.Join(f.dir, strconv.Itoa(f.n)), b, 0o600)
+}
+
+func (f *fileSink) count() int { return f.n }
+
+func (f *fileSink) chunk(i int) ([]byte, error) {
+ return os.ReadFile(filepath.Join(f.dir, strconv.Itoa(i+1)))
+}
+```
+
+`readDBForward` takes the waiting sentence, the sink and the byte bound, and returns the id and the result instead of opening it:
+
+```go
+func readDBForward(ctx context.Context, body io.Reader, key *ecdh.PrivateKey, host, region, reading string, status io.Writer, sink dbChunkSink, max int64) (string, *dbreads.Result, error) {
+ unchecked := fmt.Errorf(dbfUnchecked, region)
+ sc := bufio.NewScanner(body)
+ sc.Buffer(make([]byte, 64<<10), dbreads.MaxLineBytes)
+ var id, ended string
+ var total int64
+ for sc.Scan() {
+ var l dbreads.Line
+ if err := json.Unmarshal(sc.Bytes(), &l); err != nil {
+ return "", nil, unchecked
+ }
+ switch {
+ case l.Admitted != "":
+ if id != "" || !dbreads.ValidID(l.Admitted) {
+ return "", nil, unchecked
+ }
+ id = l.Admitted
+ fmt.Fprintln(status, fmt.Sprintf(reading, host, region))
+ case l.Chunk != 0:
+ total += int64(len(l.Data))
+ if id == "" || l.Chunk != int64(sink.count())+1 || total > max+int64(sink.count()+1)*int64(dbreads.SealOverhead) {
+ return "", nil, unchecked
+ }
+ if err := sink.add(l.Data); err != nil {
+ return "", nil, err
+ }
+ case l.Result != nil:
+ if id == "" {
+ return "", nil, unchecked
+ }
+ return id, l.Result, nil
+ case l.Ended != "":
+ ended = l.Ended
+ }
+ }
+ if errors.Is(ctx.Err(), context.Canceled) {
+ return "", nil, errors.New("stopped")
+ }
+ if ended == dbreads.EndedNotPicked {
+ return "", nil, fmt.Errorf(dbfNotPicked, region)
+ }
+ return "", nil, fmt.Errorf(dbfUnfinished, host, region)
+}
+```
+
+`openDBForward` reads its chunks from the sink: its signature becomes `openDBForward(id string, key *ecdh.PrivateKey, sink dbChunkSink, res dbreads.Result, host, region string)`, with `len(sealed)` replaced by `sink.count()` and the loop reading `sink.chunk(i)` (an error from it is `unchecked`). `dbQueryForward` calls `readDBForward(ctx, res.Body, key, t.host, fwd.Region, dbfReading, os.Stderr, &memSink{}, dbreads.MaxOutputBytes)` and then `openDBForward`.
+
+The pull's opener and verb:
+
+```go
+// openDBPull opens a pull's result (spec §6, phase 2): a done one's chunks,
+// opened in order, are the planned files concatenated; they are checked
+// against the summary's count and hash and split by its file list into a
+// litestream file replica, which today's restore and check turn into the
+// copy. Any other state maps to its sentence.
+func openDBPull(ctx context.Context, id string, key *ecdh.PrivateKey, sink dbChunkSink, res dbreads.Result, host, region, dir string) (*dbdoor.Restored, error) {
+ unchecked := fmt.Errorf(dbfUnchecked, region)
+ if res.Valid() != nil {
+ return nil, unchecked
+ }
+ s, err := dbreads.OpenSummary(key, id, res.Sealed)
+ if err != nil {
+ return nil, unchecked
+ }
+ if res.State != dbreads.StateDone {
+ switch res.Code {
+ case dbreads.CodeTooLarge:
+ return nil, fmt.Errorf(dbfPullTooLarge, host, humanBytes(&s.Restored), humanBytes(&s.Cap), region)
+ case dbreads.CodeNoRoom:
+ return nil, fmt.Errorf(dbfPullNoRoom, region, host)
+ }
+ _, err := openDBForward(id, key, sink, res, host, region)
+ return nil, err
+ }
+ var sizes int64
+ for _, f := range s.Files {
+ sizes += f.Size
+ }
+ if res.LastSeq != int64(sink.count()) || len(s.Files) == 0 || sizes != s.Bytes || s.Files[len(s.Files)-1].MaxTXID != s.TXID {
+ return nil, unchecked
+ }
+ root := filepath.Join(dir, "replica")
+ h := sha256.New()
+ fi, written := 0, int64(0)
+ var out *os.File
+ defer func() {
+ if out != nil {
+ out.Close()
+ }
+ }()
+ for i := 0; i < sink.count(); i++ {
+ sealed, err := sink.chunk(i)
+ if err != nil {
+ return nil, unchecked
+ }
+ seq := int64(i + 1)
+ plain, err := dbreads.OpenChunk(key, id, seq, dbreads.StreamLTX, seq == res.LastSeq, sealed)
+ if err != nil {
+ return nil, unchecked
+ }
+ h.Write(plain)
+ for len(plain) > 0 {
+ if fi >= len(s.Files) {
+ return nil, unchecked
+ }
+ f := s.Files[fi]
+ if out == nil {
+ p := litestream.LTXFilePath(root, f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID))
+ if err := os.MkdirAll(filepath.Dir(p), 0o700); err != nil {
+ return nil, err
+ }
+ if out, err = os.OpenFile(p, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600); err != nil {
+ return nil, err
+ }
+ }
+ n := min(int64(len(plain)), f.Size-written)
+ if _, err := out.Write(plain[:n]); err != nil {
+ return nil, err
+ }
+ plain, written = plain[n:], written+n
+ if written == f.Size {
+ if err := out.Close(); err != nil {
+ return nil, err
+ }
+ out, fi, written = nil, fi+1, 0
+ }
+ }
+ }
+ if fi != len(s.Files) || hex.EncodeToString(h.Sum(nil)) != s.SHA256 {
+ return nil, unchecked
+ }
+ db := filepath.Join(dir, "replica.db")
+ if err := dbdoor.RestoreLocal(ctx, lsfile.NewReplicaClient(root), db, s.TXID); err != nil {
+ return nil, dbRestoreError(host, err)
+ }
+ os.RemoveAll(root)
+ return &dbdoor.Restored{Path: db, Dir: dir, TXID: s.TXID, At: s.At}, nil
+}
+
+// dbPullForward is a pull through the forwarded read. The spool, the files
+// and the copy live in one private 0700 directory under the system temp
+// dir, removed on every path but success, where the caller removes it after
+// copyOut, as it does today's restore directory.
+func dbPullForward(ctx context.Context, t dbTarget, fwd dbdoor.Forward) (*dbdoor.Restored, error) {
+ ctx, cancel := context.WithTimeout(ctx, dbreads.CLIBound)
+ defer cancel()
+ key, err := dbreads.NewReplyKey()
+ if err != nil {
+ return nil, err
+ }
+ at := ""
+ if !t.at.IsZero() {
+ at = t.at.UTC().Format(time.RFC3339Nano)
+ }
+ body, err := json.Marshal(dbreads.ReadBody{Op: dbreads.OpPull, At: at, ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())})
+ if err != nil {
+ return nil, err
+ }
+ res, err := t.sess.c.doStream(ctx, http.MethodPost, apiPath(t.acct, t.app, "/db/"+url.PathEscape(t.host)+"/reads"),
+ bytes.NewReader(body), int64(len(body)), http.Header{"Content-Type": {"application/json"}})
+ if err != nil {
+ if errors.Is(ctx.Err(), context.Canceled) {
+ return nil, errors.New("stopped")
+ }
+ return nil, fmt.Errorf(dbfUnfinished, t.host, fwd.Region)
+ }
+ defer res.Body.Close()
+ if res.StatusCode != http.StatusOK {
+ raw, _ := io.ReadAll(io.LimitReader(res.Body, 64<<10))
+ return nil, apiErrorFromBody(res.StatusCode, raw)
+ }
+ dir, err := os.MkdirTemp("", "carlos-db-")
+ if err != nil {
+ return nil, err
+ }
+ if err := os.Chmod(dir, 0o700); err != nil {
+ os.RemoveAll(dir)
+ return nil, err
+ }
+ done := false
+ defer func() {
+ if !done {
+ os.RemoveAll(dir)
+ }
+ }()
+ sink := &fileSink{dir: filepath.Join(dir, "sealed")}
+ id, result, err := readDBForward(ctx, res.Body, key, t.host, fwd.Region, dbfFetching, os.Stderr, sink, dbreads.MaxPullBytes)
+ if err != nil {
+ return nil, err
+ }
+ r, err := openDBPull(ctx, id, key, sink, *result, t.host, fwd.Region, dir)
+ if err != nil {
+ return nil, err
+ }
+ os.RemoveAll(sink.dir)
+ done = true
+ return r, nil
+}
+```
+
+In `db.go`, `dbPull` becomes:
+
+```go
+func dbPull(ctx context.Context, t dbTarget, out string) error {
+ r, err := dbRestore(ctx, t)
+ var fwd *errDBForward
+ if errors.As(err, &fwd) {
+ if !slices.Contains(fwd.door.Forward.Ops, dbreads.OpPull) {
+ // A console that forwards queries and not pulls (plan 2's).
+ return fmt.Errorf(dbfNoPull, t.host, fwd.door.Forward.Region)
+ }
+ r, err = dbPullForward(ctx, t, *fwd.door.Forward)
+ }
+ if err != nil {
+ return err
+ }
+ defer os.RemoveAll(r.Dir)
+ if err := copyOut(ctx, r.Path, out); err != nil {
+ return err
+ }
+ fmt.Fprintf(stdout, "Saved a copy of %s's database to %s.\n", t.host, out)
+ fmt.Fprint(stdout, dbPullHolds(r, t.at))
+ fmt.Fprintf(stdout, "It is a copy. Nothing you do to it reaches %s.\n", t.host)
+ return nil
+}
+```
+
+Add `strconv`, `litestream`, `lsfile` and `ltx` imports to `dbforward.go`. Add DBF-17, DBF-18 and DBF-24 to `dbForwardCopyBuilt`.
+
+- [ ] **Step 4: Run the tests**
+
+Run: `go test . -run 'TestForwardedPull|TestForwardedQuery|TestDBForward|TestDBPull|TestCopyOut' -count=1 && go vet . && GOOS=windows GOARCH=amd64 go build . && GOOS=darwin GOARCH=arm64 go build .`
+Expected: PASS and clean builds.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add dbforward.go dbforward_test.go db.go approved_db_forward_copy_test.go
+git commit -m "carlos db pull: the forwarded pull, spooled, checked, restored locally (DBF-17, DBF-18, DBF-24)"
+```
+
+---
+
+### Task 8: The CLI page (DBF-23)
+
+Waits for Task 2.
+
+**Files:**
+- Modify: `docs/site/cli.md` (the two paragraphs DBF-30 wrote)
+- Modify: `approved_db_forward_copy_test.go` (DBF-23 in, DBF-30 out of `dbForwardCopyBuilt`)
+
+- [ ] **Step 1: Swap the ids and run the gate**
+
+In `dbForwardCopyBuilt`, replace `"DBF-30"` with `"DBF-23"`. Run: `go test . -run 'TestDBForwardBuiltCopyIsApplied' -count=1`. Expected: FAIL: the page still carries DBF-30.
+
+- [ ] **Step 2: Replace the paragraphs**
+
+Replace DBF-30's two one-line paragraphs in the `carlos db` section of `docs/site/cli.md` with DBF-23's approved two paragraphs, one line each, a blank line between them.
+
+- [ ] **Step 3: Run the gate and the docs check**
+
+Run: `go test . -run 'TestDBForward' -count=1 && go test ./internal/docsite/ -count=1`
+Expected: PASS.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add docs/site/cli.md approved_db_forward_copy_test.go
+git commit -m "docs: carlos db pull reads another region's instances (DBF-23)"
+```
+
+---
+
+### Task 9: Acceptance: the pull through the real CLI, MinIO, and size
+
+**Files:**
+- Modify: `dbforward_e2e_test.go`
+- Modify: `internal/console/dbforward_minio_test.go`
+
+**Interfaces:**
+- Consumes: plan 2's `dbfE2E`, `newDBFE2E`, `run`, `betweenTheSyncs`, `scriptBox`, `e2eScript`, `TestDBForwardMinio`'s helpers.
+
+- [ ] **Step 1: Write the tests**
+
+Append to `dbforward_e2e_test.go`:
+
+```go
+func sha256File(t *testing.T, path string) string {
+ t.Helper()
+ b, err := os.ReadFile(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ sum := sha256.Sum256(b)
+ return hex.EncodeToString(sum[:])
+}
+
+// The pull's parity: a forwarded pull and an in-region pull of the same
+// replica save byte-identical copies, for the newest copy and as of a time,
+// and print the same lines but for the host.
+func TestDBForwardPullParityThroughTheCLI(t *testing.T) {
+ e := newDBFE2E(t, dbfE2EOptions{forward: true})
+ for _, at := range []string{"", e.betweenTheSyncs(t)} {
+ dir := t.TempDir()
+ args := []string{"db", "pull", "--app", "shop"}
+ if at != "" {
+ args = append(args, "--at", at)
+ }
+ here, hereErr, err := e.run(t, append(args, "--instance", dbfE2EHere, "--out", filepath.Join(dir, "here.db"))...)
+ if err != nil {
+ t.Fatalf("in region: %v\n%s", err, hereErr)
+ }
+ far, farErr, err := e.run(t, append(args, "--instance", dbfE2EFar, "--out", filepath.Join(dir, "far.db"))...)
+ if err != nil {
+ t.Fatalf("forwarded: %v\n%s", err, farErr)
+ }
+ if sha256File(t, filepath.Join(dir, "here.db")) != sha256File(t, filepath.Join(dir, "far.db")) {
+ t.Fatalf("at %q the copies differ", at)
+ }
+ if strings.ReplaceAll(far, dbfE2EFar, dbfE2EHere) != strings.ReplaceAll(here, filepath.Join(dir, "here.db"), filepath.Join(dir, "far.db")) {
+ t.Fatalf("at %q: %q vs %q", at, here, far)
+ }
+ if !strings.Contains(farErr, fmt.Sprintf(dbfFetching, dbfE2EFar, "uk")) {
+ t.Fatalf("no DBF-18: %s", farErr)
+ }
+ }
+}
+
+// T11 for a pull: a hash mismatch saves nothing at --out.
+func TestDBForwardPullSavesNothingForABadAnswer(t *testing.T) {
+ e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true})
+ stop := e.scriptBox(t, func(b *e2eScript) {
+ b.picked(time.Now().Add(time.Minute))
+ sealed, _ := dbreads.SealChunk(b.pub, b.id, 1, dbreads.StreamLTX, true, []byte("not the files"))
+ b.store.Put(context.Background(), dbreads.ChunkKey(b.id, 1), sealed, "")
+ summary, _ := dbreads.SealSummary(b.pub, b.id, dbreads.Summary{TXID: 1, Bytes: 13, SHA256: "00",
+ Files: []dbdoor.LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 1, Size: 13}}})
+ r, _ := json.Marshal(dbreads.Result{State: dbreads.StateDone, LastSeq: 1, Box: "uk-1", Sealed: summary})
+ b.store.Put(context.Background(), dbreads.ResultKey(b.id), r, "")
+ })
+ defer stop()
+ out := filepath.Join(t.TempDir(), "far.db")
+ _, stderr, err := e.run(t, "db", "pull", "--app", "shop", "--instance", dbfE2EFar, "--out", out)
+ if err == nil || !strings.Contains(stderr, "did not check out") {
+ t.Fatalf("%q %v", stderr, err)
+ }
+ if _, err := os.Stat(out); !os.IsNotExist(err) {
+ t.Fatalf("something was saved at --out: %v", err)
+ }
+}
+
+// A console from plan 2 forwards queries and not pulls. Its marker is
+// played here by a wrapper that drops "pull" from forward_ops on the way
+// out: the new CLI says DBF-8 and saves nothing.
+func TestDBForwardPullOnAQueryOnlyConsole(t *testing.T) {
+ queryOnly := func(t *testing.T, h http.Handler) (string, []string) {
+ ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ rec := httptest.NewRecorder()
+ h.ServeHTTP(rec, r)
+ body := rec.Body.Bytes()
+ if rec.Code == http.StatusConflict {
+ body = bytes.Replace(body, []byte(`"forward_ops":["query","pull"]`), []byte(`"forward_ops":["query"]`), 1)
+ }
+ for k, vs := range rec.Header() {
+ w.Header()[k] = vs
+ }
+ w.Header().Del("Content-Length")
+ w.WriteHeader(rec.Code)
+ w.Write(body)
+ }))
+ t.Cleanup(ts.Close)
+ return ts.URL, nil
+ }
+ e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true, serve: queryOnly})
+ out := filepath.Join(t.TempDir(), "far.db")
+ _, stderr, err := e.run(t, "db", "pull", "--app", "shop", "--instance", dbfE2EFar, "--out", out)
+ if err == nil || !strings.Contains(stderr, fmt.Sprintf(dbfNoPull, dbfE2EFar, "uk")) {
+ t.Fatalf("%q %v", stderr, err)
+ }
+ if _, err := os.Stat(out); !os.IsNotExist(err) {
+ t.Fatalf("something was saved at --out: %v", err)
+ }
+}
+```
+
+The wrapper buffers each response, so it is used only where no response streams (the list door's 409 is all this test reads).
+
+In `internal/console/dbforward_minio_test.go`, right after T1's query loop, add a pull of `shop` through MinIO:
+
+```go
+ // A pull through MinIO: the planned files, sealed as the ltx stream,
+ // adding up to the summary.
+ pkey, _ := dbreads.NewReplyKey()
+ pres := dfx.openRead(t, shop, member, dbreads.ReadBody{Op: dbreads.OpPull, ReplyKey: dbreads.EncodeReplyKey(pkey.PublicKey())})
+ plines := readLines(t, pres.Body)
+ pres.Body.Close()
+ presult := plines[len(plines)-1].Result
+ if presult == nil || presult.State != dbreads.StateDone {
+ t.Fatalf("pull: %+v", plines[len(plines)-1])
+ }
+ psum, err := dbreads.OpenSummary(pkey, plines[0].Admitted, presult.Sealed)
+ if err != nil || len(psum.Files) == 0 || psum.TXID != psum.Files[len(psum.Files)-1].MaxTXID {
+ t.Fatalf("pull summary %+v %v", psum, err)
+ }
+ h := sha256.New()
+ for _, l := range plines {
+ if l.Chunk == 0 {
+ continue
+ }
+ plain, err := dbreads.OpenChunk(pkey, plines[0].Admitted, l.Chunk, dbreads.StreamLTX, l.Chunk == presult.LastSeq, l.Data)
+ if err != nil {
+ t.Fatalf("pull chunk %d: %v", l.Chunk, err)
+ }
+ h.Write(plain)
+ }
+ if hex.EncodeToString(h.Sum(nil)) != psum.SHA256 {
+ t.Fatal("the pulled bytes do not add up to the summary")
+ }
+```
+
+with `"crypto/sha256"` and `"encoding/hex"` added to that file's imports.
+
+- [ ] **Step 2: Run them**
+
+Run (sandbox off): `go test . -run 'TestDBForwardPull' -count=1 -v`
+Expected: PASS. The MinIO subtest runs in CI's MinIO step.
+
+- [ ] **Step 3: The size run (only on Paul's say-so)**
+
+The spec's long-request concern for a pull is a full-size one. Ask Paul before running it: it writes a database of the cap's size (Task 1's number, up to 2 GiB) and its replica, about three times that on disk. With his go, on dev (not DOCS-T1): an instance in dev's uk region whose database is near the cap (seed it by a `carlos run` of a script that inserts `randomblob` rows until the database is the size wanted, then let it sleep so its replica syncs), then `time carlos db pull --app <app> --instance <host> --out "$TMPDIR/dbf-big.db"` from a laptop through the dev console. Record the time, the CLI's peak spool size, and that it finished inside 22 minutes; remove the file and the instance after. Report on the branch's discussion.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add dbforward_e2e_test.go internal/console/dbforward_minio_test.go
+git commit -m "db forward: the pull's parity, bad answer and MinIO tests"
+```
+
+---
+
+### Task 10: Mutation checks, census, weights, merge, CI, roll
+
+**Files:**
+- Create: `internal/provision/testdata/plan14-root-reads-db-region-forward-3.tsv`
+- Modify: `internal/provision/plan14_root_reads_test.go`
+- Create: `docs/superpowers/notes/census-proposal-db-region-forward-3.md`
+- Modify: `.amadan/test-weights.tsv`
+
+- [ ] **Step 1: Mutation checks**
+
+In a tree copy under `$TMPDIR`, as plans 1 and 2 did; every row must turn red:
+
+| Mutation | Test that must fail |
+|---|---|
+| `holdAndSend` skips `verifyLTX` | `TestWorkerHoldRefusesACorruptFile` |
+| the agent's pull sends `run` before check B | `TestPullTooLargeAndNoChunkBeforeCheckB` |
+| `openDBPull` skips the hash comparison | `TestDBForwardPullSavesNothingForABadAnswer` |
+| `openDBPull` skips the file-size sum | `TestForwardedPullRefusesFilesThatDoNotAddUp` |
+| `NewAgent` keeps a `MaxPull` above `MaxPullBytes` | `TestPullCapOnlyLowers` (through `dbReadsSettings`) |
+
+- [ ] **Step 2: Merge main, census, 9B8**
+
+Merge main; run plan 1 Task 17's dump with the prefixes `dbforward.go`, `db.go`, `internal/dbreads/`, `internal/console/apidbforward`, `internal/host/dbreads`; write `plan14-root-reads-db-region-forward-3.tsv` (`rev3=PROPOSED-db-region-forward-3`) with plans 1 and 2's basis templates: worker units as the db-read worker at J4 ("verifies and holds the planned files in its private scratch and sends them back after the agent's check B"), agent and console units re-frozen at their plan 1 and plan 2 dispositions with the pull appended, CLI units as the carlos CLI at J4 ("spools sealed chunks to a 0700 directory under the system temp dir, writes the opened files as a litestream file replica there, restores and checks them with dbdoor.RestoreLocal, and removes the directory on every path"); envelopes re-pinned. Load it last with the previous seal checked first; set the new seal. Run `TestPlan9B8RecordWriteCensus`; re-pin any changed entry at its class in a `dbRegionForward3WriteCensusInventory` layered on plan 2's. Write the census proposal and post it on #280.
+
+- [ ] **Step 3: Test weights**
+
+Measure the new root and console tests with targeted `-json` runs (plan 2 Task 14 Step 6) and add rows for those of a second or more.
+
+- [ ] **Step 4: Push, describe, CI, review, merge**
+
+As plan 2 Task 14 Steps 7 and 8: final local checks, push, the branch description (what plan 3 adds; Task 1's measurement and Paul's confirmation of the cap; the mutation table; the root-runner run id of `TestRootDBReadPullThroughTheInstalledUnits`; whether the size run was done and its numbers; the census proposal; the copy state), green CI on the tip, the whole-branch review, merge main again, squash, merge behind `git merge-base --is-ancestor origin/main HEAD`.
+
+- [ ] **Step 5: The roll, each live step on Paul's go**
+
+Dev boxes, dev console, check (a forwarded `carlos db pull` and `carlos db pull --at` of a dev uk instance from a laptop: both save a copy whose `PRAGMA integrity_check` is `ok`, with today's three lines on stdout and DBF-18 on stderr), then the flagship boxes and console. The lifecycle rule is already in place from plan 2. The per-region bucket that keeps a pull's ciphertext in the data's region is plan 5 (spec §14, ruling 1), not this plan.