| index 3c71fa9d1..764f0e1bb 100644 |
| --- a/docs/superpowers/plans/2026-10-22-db-region-forward-plan-2-query.md |
| +++ b/docs/superpowers/plans/2026-10-22-db-region-forward-plan-2-query.md |
| @@ -61,6 +61,23 @@ Astra's findings (`docs/superpowers/specs/CODEX-FINDINGS-DB-REGION-FORWARD-PLANS |
| |
| - **11, in part: making the audit insert itself cancellation-aware.** `RecordActivityContext` takes a context, but its claim plumbing ends in `claimRawRecordActivity`'s context-free `Exec`, shared by every audited route (`internal/console/store.go:2588`, `2668`). Changing that is a change to the claim SQL layer and its census, for one local SQLite insert. The order makes it unnecessary: the row is written while the request is open (checked just before) and before any object a box can see, so a request that ends during the insert leaves the row of an admission and no read. The serial bump, the one bucket write, is request-bound. |
| |
| +## Plan review, round 2 |
| + |
| +From `CODEX-FINDINGS-DB-REGION-FORWARD-PLANS-R2.md`: |
| + |
| +- **25** (`dbStream` is `internal/console/apidb.go:475`'s function): the relay type is `dbReadRelay`. Every new identifier in the three plans was then checked against `origin/main`; this was the only collision. |
| +- **28** (admission could still publish after cancellation): the request's context is checked immediately before each of admission's four writes (audit row, `request.json`, queue entry, serial), with `cancel` once anything is published. Test: `TestDBReadEndedDuringAdmissionWritesOnlyTheCancel`, one case per step, through the `s.dbf.beforeWrite` seam. Not closed by the checks alone, and not closable without a transactional store: a request that ends between a check and its write still sees that write land; the next check then writes `cancel`, and a box that has already elected stops at its next `cancel` read (every 5 s). The audit insert's own cancellation stays declined (round 1). |
| +- **30** (pickup windows shorter than `issued_at`'s truncation): every accelerated window leaves the observation it needs after up to a second of truncation; production's timestamps are unchanged. |
| +- **31** (the completeness guard unwitnessed): `TestDBReadResultWaitsForAChunkTheListingMissed`. |
| +- **33** (the serial mutation never ran): "ended during the serial read" cancels inside the bump, with a store that refuses a cancelled write as S3 does. |
| +- **34** (T7 lacked the real CLI and console through backhaul): Task 11 runs both edges in package `edge`'s test binary as a helper process and the real CLI and console in the root package; `checkProgressAtTheCLI` watches the CLI's stderr and socket, through the local edge as well. |
| +- **35** (the result-line deadline check has a witness): `TestDBReadResultAfterTheDeadlineIsNotRelayed`, with a `ResponseWriter` that holds the last chunk's line. |
| +- **36** (SQL JSON would rewrite): the CLI refuses it with DBF-32 before sending, the console refuses U+FFFD with 400. Tests at the console, the CLI and the binary. |
| +- **37** (unbounded chunk count): `dbfChunkBound` caps the count and refuses a short chunk, or anything after an empty one, before it is kept. |
| +- **38** (the extension's mutation stayed green): `TestDBReadSetsBothReadDeadlines` watches the connection's read deadlines. |
| + |
| +None declined beyond round 1's audit-insert item. |
| + |
| ## Census impact |
| |
| - **Plan 14 root-read census:** one overlay, `plan14-root-reads-db-region-forward-2.tsv`, `rev3=PROPOSED-db-region-forward-2`: |
| @@ -96,6 +113,7 @@ Reviewed in Task 1 and built by the tasks named: |
| | DBF-16 | dbforward.go | <host> is moving between regions. read its database once the move has finished. | 8 | |
| | DBF-21 | internal/console/apidbforward.go | forwarded a database query to region <region>. the query was <n> bytes, sha256 <sha> (cli) | 6 | |
| | DBF-22 | internal/console/apidbforward.go | forwarded a database query to region <region>, as of <time>. the query was <n> bytes, sha256 <sha> (cli) | 6 | |
| +| DBF-32 | dbforward.go | this query cannot be sent to region <region>, because it holds bytes that are not UTF-8 text or the character U+FFFD. fix the query and run it again. | 8 | |
| | DBF-30 | docs/site/cli.md | (two paragraphs; spec §10) | 9 | |
| |
| Unchanged and reused (spec §10): `dbQueryStatus`, the `dbRestoreError` sentences, SQLite's own text for `sql-error`, `stopped`, the `checkQuerySQL` refusals, the residency sentence when forwarding is off. |
| @@ -117,7 +135,7 @@ Unchanged and reused (spec §10): `dbQueryStatus`, the `dbRestoreError` sentence |
| | `infra/modules/deployment-buckets/main.tf` (modify), `tests/db_reads.tftest.hcl` (create) | The lifecycle rule (J4), T17. | |
| | `docs/superpowers/notes/db-reads-lifecycle-approval.md` (create) | The scoped approval plan for applying it. | |
| | `.amadan/ci.d/30-minio-contract` (modify) | Buckets, users and the PASS check for T1 and T16. | |
| -| Tests: `internal/console/apidbforward_test.go`, `internal/console/dbforward_minio_test.go`, `dbforward_test.go`, `dbforward_e2e_test.go`, `internal/edge/dbforward_backhaul_test.go` (create) | T1, T6, T7, T8, T11, T14, T15, T16, T18. | |
| +| Tests: `internal/console/apidbforward_test.go`, `internal/console/dbforward_minio_test.go`, `dbforward_test.go`, `dbforward_e2e_test.go`, `internal/edge/dbforward_backhaul_helper_test.go` (create) | T1, T6, T7, T8, T11, T14, T15, T16, T18. | |
| |
| --- |
| |
| @@ -131,7 +149,7 @@ Unchanged and reused (spec §10): `dbQueryStatus`, the `dbRestoreError` sentence |
| |
| - [ ] **Step 1: Run the review** |
| |
| -Invoke the `copy-review` skill with rows DBF-1 to DBF-16, DBF-21, DBF-22 and DBF-30, each with its context from spec §10 (where it appears and when), neutral example values (`shop.example.com`, region `uk`, `2.0GB`, `64 MB`). Read `result.json` before relaunching the review server: a relaunch has destroyed a submitted review before. |
| +Invoke the `copy-review` skill with rows DBF-1 to DBF-16, DBF-21, DBF-22, DBF-30 and DBF-32, each with its context from spec §10 (where it appears and when), neutral example values (`shop.example.com`, region `uk`, `2.0GB`, `64 MB`). Read `result.json` before relaunching the review server: a relaunch has destroyed a submitted review before. |
| |
| - [ ] **Step 2: Apply Paul's edits verbatim** |
| |
| @@ -376,7 +394,7 @@ git commit -m "db forward: the reads route's lines and the list door's additive |
| |
| **Interfaces:** |
| - Consumes: `release.RootSigner`, `newDBStreamLimits` (`apidb.go`), `dbreads.OpenPerConsole`, `dbreads.OpenPerPerson`. |
| -- Produces: `console.NewHTTPServer(h http.Handler) *http.Server`; `Config.DBForward bool`; `(*Server).dbForwardSigner() (release.Signer, bool)`; `Server.dbReads *dbStreamLimits`; `Server.dbf dbForwardTimes` with fields `poll, wait, stall, preHeader, pickup, grace, bound, call time.Duration`. |
| +- Produces: `console.NewHTTPServer(h http.Handler) *http.Server`; `Config.DBForward bool`; `(*Server).dbForwardSigner() (release.Signer, bool)`; `Server.dbReads *dbStreamLimits`; `Server.dbf dbForwardTimes` with fields `poll, wait, stall, preHeader, pickup, grace, bound, call time.Duration` and the test seam `beforeWrite func(step string)`. |
| |
| - [ ] **Step 1: Write the failing tests** |
| |
| @@ -533,6 +551,11 @@ type dbForwardTimes struct { |
| grace time.Duration // after the owner's deadline |
| bound time.Duration // the request's own bound after issued_at |
| call time.Duration // one store call inside the stream (plan review finding 12) |
| + |
| + // beforeWrite, nil in production, runs just before admission checks the |
| + // request's context ahead of each write ("audit", "request", "queue", |
| + // "serial"): a test ends the request there (plan review finding 28). |
| + beforeWrite func(step string) |
| } |
| |
| func defaultDBForwardTimes() dbForwardTimes { |
| @@ -932,7 +955,7 @@ Waits for Task 1 (DBF-9, DBF-10, DBF-21, DBF-22 approved). |
| |
| **Interfaces:** |
| - Consumes: `dbForwardSigner`, `forwardable`, `dbForwardEvidence`, `s.dbReads`, `s.dbf` (Tasks 3 to 5); `resolveDBSource`, `dbClassRefusal`, `dbResidencyRefusal`, `apiAccountApp`, `principalFrom`, `s.cli.db.allow`, `hashToken`, `bearer`, `readJSONLimit`, `jsonError`, `s.store.RecordActivityContext`, `release.BumpSerial`, `release.SerialKey`, `dbAuditAction` (existing console code); `dbreads.Sign`, `MintID`, `RequestKey`, `QueueKey`, `ParseReplyKey`, `ReadBody`; `dbquery.CheckSQL`; `hostclaim.Incarnation`. |
| -- Produces: `(*Server).handleAPIDBRead(w, r)`; `(*Server).admitDBRead(w, r) (dbReadAdmitted, string, bool)`; `type dbReadAdmitted struct{ id, region string; issued time.Time }`; `dbForwardAuditDetail(region, sqlText string, at time.Time) string`; `(*Server).cancelDBRead(id)`; `(*Server).announceDBRead(ctx)`; `dbfTooMany`, `dbfMoving`, `dbfAudit`, `dbfAuditAt`; the stream half (`streamDBRead`) is Task 7's; test helpers `(*dbfFixture).readURL(host)`, `queryBody(t, sql)`, `(*dbfFixture).openRead(t, host, bearer, body) *http.Response`, `readLines(t, body io.Reader) []dbreads.Line`, `refusal(t, res)`, `dbfPutHook`, `dbReadAudited(t, st)`. |
| +- Produces: `(*Server).handleAPIDBRead(w, r)`; `(*Server).admitDBRead(w, r) (dbReadAdmitted, string, bool)`; `type dbReadAdmitted struct{ id, region string; issued time.Time }`; `dbForwardAuditDetail(region, sqlText string, at time.Time) string`; `(*Server).cancelDBRead(id)`; `(*Server).announceDBRead(ctx)`; `dbfTooMany`, `dbfMoving`, `dbfAudit`, `dbfAuditAt`; the stream half (`streamDBRead`) is Task 7's; test helpers `(*dbfFixture).readURL(host)`, `queryBody(t, sql)`, `(*dbfFixture).openRead(t, host, bearer, body) *http.Response`, `readLines(t, body io.Reader) []dbreads.Line`, `refusal(t, res)`, `dbfPutHook`, `dbReadAudited(t, st)`, `(*dbfFixture).dbfServe(t, ctx, sql)`, `dbfBucket(t, s)`, `dbfCtxStore`; the seam `s.dbf.beforeWrite`. |
| |
| - [ ] **Step 1: Write the failing tests** |
| |
| @@ -1022,6 +1045,24 @@ func TestDBReadRouteRefusesBeforeAdmitting(t *testing.T) { |
| t.Errorf("%s: %d %s", name, status, body) |
| } |
| } |
| + // SQL that JSON decoding would have rewritten (plan review finding 36): |
| + // a byte that is not UTF-8, and a lone surrogate escape. |
| + for name, sqlJSON := range map[string]string{ |
| + "a byte that is not UTF-8": "\"SELECT '\xff'\"", |
| + "a lone surrogate": `"SELECT '\ud800'"`, |
| + } { |
| + raw := `{"op":"query","sql":` + sqlJSON + `,"reply_key":"` + good.ReplyKey + `"}` |
| + req, _ := http.NewRequest(http.MethodPost, fx.readURL(dbfHost), strings.NewReader(raw)) |
| + req.Header.Set("Content-Type", "application/json") |
| + req.Header.Set("Authorization", "Bearer "+fx.member) |
| + res, err := http.DefaultClient.Do(req) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if status, body := refusal(t, res); status != http.StatusBadRequest { |
| + t.Errorf("%s: %d %s", name, status, body) |
| + } |
| + } |
| if keys, _ := fx.rel.List(context.Background(), dbreads.Prefix); len(keys) != 0 { |
| t.Fatalf("a refused read wrote %v", keys) |
| } |
| @@ -1118,13 +1159,74 @@ func dbReadAudited(t *testing.T, st *Store) bool { |
| return false |
| } |
| |
| -// Plan review finding 11 (J1, spec §9.3 rev 6): the audit row is written |
| -// before any box can see the read, and a request that ends once the queue |
| -// entry is down writes the cancel and nothing else: no serial bump. |
| -// Mutations: move the audit after the creates (red: no row when request.json |
| -// lands); call s.bumpSerial instead of announceDBRead, or drop the ctx check |
| -// after the queue create (red: the serial moves); drop the cancel on that |
| -// path (red: no cancel object). |
| +// dbfServe drives the handler in process with ctx as the request's own |
| +// context, so a test can end the request at an exact step. |
| +func (fx *dbfFixture) dbfServe(t *testing.T, ctx context.Context, sqlText string) *httptest.ResponseRecorder { |
| + t.Helper() |
| + body, _ := queryBody(t, sqlText) |
| + raw, _ := json.Marshal(body) |
| + req := httptest.NewRequest(http.MethodPost, fx.readURL(dbfHost), bytes.NewReader(raw)).WithContext(ctx) |
| + req.Header.Set("Content-Type", "application/json") |
| + req.Header.Set("Authorization", "Bearer "+fx.member) |
| + rec := httptest.NewRecorder() |
| + fx.srv.ServeHTTP(rec, req) |
| + return rec |
| +} |
| + |
| +// dbfBucket is what a read left under db-reads/: request objects, queue |
| +// entries and cancels, by count. |
| +func dbfBucket(t *testing.T, s release.Store) (requests, queued, cancels int) { |
| + t.Helper() |
| + keys, _ := s.List(context.Background(), dbreads.Prefix) |
| + for _, k := range keys { |
| + switch { |
| + case strings.HasPrefix(k, dbreads.Prefix+"queue/"): |
| + queued++ |
| + case strings.HasSuffix(k, "/request.json"): |
| + requests++ |
| + case strings.HasSuffix(k, "/cancel"): |
| + cancels++ |
| + } |
| + } |
| + return |
| +} |
| + |
| +// dbfCtxStore refuses a write whose context has ended, as the S3 store |
| +// does and DirStore does not, and runs onGet once on a read of key. |
| +type dbfCtxStore struct { |
| + release.Store |
| + key string |
| + onGet func() |
| + once sync.Once |
| +} |
| + |
| +func (c *dbfCtxStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if key == c.key { |
| + c.once.Do(c.onGet) |
| + } |
| + return c.Store.Get(ctx, key) |
| +} |
| + |
| +func (c *dbfCtxStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + if err := ctx.Err(); err != nil { |
| + return "", err |
| + } |
| + return c.Store.Put(ctx, key, body, ifMatch) |
| +} |
| + |
| +// Plan review findings 11, 28 and 33 (J1, spec §5.5 step 3 and §9.3 rev 6): |
| +// the request's context is checked immediately before every write of |
| +// admission. Ended before the audit row: nothing at all. Before |
| +// request.json: the row only. Before the queue entry: the row, request.json |
| +// and the cancel. Before the serial: the row, both objects and the cancel. |
| +// Never a serial bump, never a 200. And the bump itself runs on the |
| +// request's context, so a request that ends while it reads the serial does |
| +// not write it. |
| +// Mutations, each red in its own case: drop any one of the four ended() |
| +// checks; drop the cancel on a published step; write the audit row after |
| +// the creates ("the audit row is down before the request"); call |
| +// s.bumpSerial instead of announceDBRead ("ended during the serial read": |
| +// the detached write lands). |
| func TestDBReadEndedDuringAdmissionWritesOnlyTheCancel(t *testing.T) { |
| t.Run("the audit row is down before the request", func(t *testing.T) { |
| fx := newDBFFixture(t, true) |
| @@ -1141,39 +1243,50 @@ func TestDBReadEndedDuringAdmissionWritesOnlyTheCancel(t *testing.T) { |
| t.Fatal("request.json was written before the audit row") |
| } |
| }) |
| - t.Run("ended after the queue entry", func(t *testing.T) { |
| + for _, c := range []struct { |
| + step string |
| + audited bool |
| + requests, queued, cancels int |
| + }{ |
| + {"audit", false, 0, 0, 0}, |
| + {"request", true, 0, 0, 0}, |
| + {"queue", true, 1, 0, 1}, |
| + {"serial", true, 1, 1, 1}, |
| + } { |
| + t.Run("ended before "+c.step, func(t *testing.T) { |
| + fx := newDBFFixture(t, true) |
| + _, serialBefore, _ := fx.rel.Get(context.Background(), release.SerialKey) |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + defer cancel() |
| + fx.srv.dbf.beforeWrite = func(step string) { |
| + if step == c.step { |
| + cancel() |
| + } |
| + } |
| + rec := fx.dbfServe(t, ctx, "SELECT 1") |
| + // The ended path writes no response (nobody is there to read |
| + // it), and a recorder never written reports 200: read the body. |
| + if strings.Contains(rec.Body.String(), "admitted") { |
| + t.Fatalf("an ended request was admitted: %s", rec.Body) |
| + } |
| + if _, serialAfter, _ := fx.rel.Get(context.Background(), release.SerialKey); serialAfter != serialBefore { |
| + t.Fatal("the serial was bumped after the request ended") |
| + } |
| + requests, queued, cancels := dbfBucket(t, fx.rel) |
| + if got := dbReadAudited(t, fx.st); got != c.audited || requests != c.requests || queued != c.queued || cancels != c.cancels { |
| + t.Fatalf("audited %v, requests %d, queue entries %d, cancels %d", got, requests, queued, cancels) |
| + } |
| + }) |
| + } |
| + t.Run("ended during the serial read", func(t *testing.T) { |
| fx := newDBFFixture(t, true) |
| _, serialBefore, _ := fx.rel.Get(context.Background(), release.SerialKey) |
| ctx, cancel := context.WithCancel(context.Background()) |
| defer cancel() |
| - fx.srv.cfg.Releases = &dbfPutHook{Store: fx.rel, suffix: "queue", hook: cancel} |
| - body, _ := queryBody(t, "SELECT 2") |
| - raw, _ := json.Marshal(body) |
| - req := httptest.NewRequest(http.MethodPost, fx.readURL(dbfHost), bytes.NewReader(raw)).WithContext(ctx) |
| - req.Header.Set("Content-Type", "application/json") |
| - req.Header.Set("Authorization", "Bearer "+fx.member) |
| - rec := httptest.NewRecorder() |
| - fx.srv.ServeHTTP(rec, req) |
| - // The ended path writes no response (nobody is there to read it), and |
| - // a recorder that was never written reports 200: read the body. |
| - if strings.Contains(rec.Body.String(), "admitted") { |
| - t.Fatalf("an ended request was admitted: %s", rec.Body) |
| - } |
| + fx.srv.cfg.Releases = &dbfCtxStore{Store: fx.rel, key: release.SerialKey, onGet: cancel} |
| + fx.dbfServe(t, ctx, "SELECT 1") |
| if _, serialAfter, _ := fx.rel.Get(context.Background(), release.SerialKey); serialAfter != serialBefore { |
| - t.Fatal("the serial was bumped after the request ended") |
| - } |
| - keys, _ := fx.rel.List(context.Background(), dbreads.Prefix) |
| - var queued, cancelled int |
| - for _, k := range keys { |
| - switch { |
| - case strings.HasPrefix(k, dbreads.Prefix+"queue/"): |
| - queued++ |
| - case strings.HasSuffix(k, "/cancel"): |
| - cancelled++ |
| - } |
| - } |
| - if queued != 1 || cancelled != 1 || !dbReadAudited(t, fx.st) { |
| - t.Fatalf("queue entries %d, cancels %d, audited %v in %v", queued, cancelled, dbReadAudited(t, fx.st), keys) |
| + t.Fatal("the serial was written after the request ended during the bump") |
| } |
| }) |
| } |
| @@ -1205,7 +1318,7 @@ func TestDBReadStalledBodyIsCutAtThePreHeaderBound(t *testing.T) { |
| } |
| ``` |
| |
| -Add `"bufio"`, `"bytes"`, `"crypto/ecdh"`, `"io"`, `"net"`, `"net/http"`, `"net/http/httptest"` and the `dbquery` import. |
| +Add `"bufio"`, `"bytes"`, `"crypto/ecdh"`, `"io"`, `"net"`, `"net/http"`, `"net/http/httptest"`, `"sync"` and the `dbquery` import. |
| |
| The admission bounds test (DBF-9, T15) is Task 7's: it needs the stream to hold its requests open (plan review finding 7). |
| |
| @@ -1370,6 +1483,16 @@ func (s *Server) admitDBRead(w http.ResponseWriter, r *http.Request) (dbReadAdmi |
| http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest) |
| return none, "", false |
| } |
| + // encoding/json turns bytes that are not UTF-8, and a lone surrogate |
| + // escape, into U+FFFD without a word, so the SQL decoded here may not be |
| + // the SQL sent. The box must run exactly the bytes typed or be told no |
| + // (plan 1's Review Focus 1), so any U+FFFD is refused; a query that |
| + // meant one literally is refused too, and says so the same way (plan |
| + // review finding 36). The CLI refuses first, with DBF-32. |
| + if strings.ContainsRune(body.SQL, utf8.RuneError) { |
| + http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest) |
| + return none, "", false |
| + } |
| if err := dbquery.CheckSQL(body.SQL); err != nil || body.SQL == "" { |
| msg := http.StatusText(http.StatusBadRequest) |
| if err != nil { |
| @@ -1435,40 +1558,65 @@ func (s *Server) admitDBRead(w http.ResponseWriter, r *http.Request) (dbReadAdmi |
| http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest) |
| return none, "", false |
| } |
| - // The audit row comes first, while the request is open and before any box |
| - // can see the read: a read a box runs always has its row, and a request |
| - // that ends here leaves the row of an admission and nothing else (spec |
| - // §9.3 rev 6; plan review finding 11). A row the store cannot write |
| - // refuses the read rather than run it unaudited. |
| - if err := ctx.Err(); err != nil { |
| + // J1 at every write (plan review finding 28): the request's context is |
| + // checked immediately before each one, because the stores do not stop a |
| + // write for it (DirStore.Put ignores its context, |
| + // internal/release/store.go:414). The audit row comes first, while the |
| + // request is open and before any box can see the read, so a read a box |
| + // runs always has its row (spec §9.3 rev 6). A request that ends before |
| + // request.json is written leaves at most that row; one that ends after |
| + // leaves request.json, perhaps the queue entry, and the cancel, never a |
| + // serial bump. A row the store cannot write refuses the read rather than |
| + // run it unaudited. |
| + ended := func(step string, published bool) bool { |
| + if s.dbf.beforeWrite != nil { |
| + s.dbf.beforeWrite(step) |
| + } |
| + if ctx.Err() == nil { |
| + return false |
| + } |
| + if published { |
| + s.cancelDBRead(id) |
| + } |
| + return true |
| + } |
| + if ended("audit", false) { |
| return none, "", false |
| } |
| if err := s.store.RecordActivityContext(ctx, acct.ID, app, actor, Activity{Action: dbAuditAction, |
| TargetType: "instance", Target: host, Detail: dbForwardAuditDetail(rec.Region, body.SQL, at)}, time.Now()); err != nil { |
| return fail("audit", err) |
| } |
| + if ended("request", false) { |
| + return none, "", false |
| + } |
| if _, err := s.cfg.Releases.Put(ctx, dbreads.RequestKey(id), wire, ""); err != nil { |
| + // An ambiguous create may have landed: cancel, as for the queue. |
| + s.cancelDBRead(id) |
| return fail("write the request", err) |
| } |
| + if ended("queue", true) { |
| + return none, "", false |
| + } |
| if _, err := s.cfg.Releases.Put(ctx, dbreads.QueueKey(rec.Region, issued, id), nil, ""); err != nil { |
| // The queue entry may have landed on an ambiguous answer: cancel, so |
| // a box that lists it skips it (J1's one write). |
| s.cancelDBRead(id) |
| return fail("write the queue entry", err) |
| } |
| - // J1: once the request has ended, nothing about the read is written but |
| - // the cancel. The serial bump runs on the request's context, not |
| - // bumpSerial's detached one. |
| - if err := ctx.Err(); err != nil { |
| - s.cancelDBRead(id) |
| + if ended("serial", true) { |
| return none, "", false |
| } |
| + // The bump runs on the request's context, not bumpSerial's detached one, |
| + // so a request that ends while it runs does not announce (finding 33). |
| s.announceDBRead(ctx) |
| admitted = true |
| return dbReadAdmitted{id: id, region: rec.Region, issued: issued}, actor, true |
| } |
| ``` |
| |
| +Add `"strings"` and `"unicode/utf8"` to `apidbforward.go`'s imports if they are not there yet. |
| + |
| Until Task 7, add a stub so this task builds, and delete it in Task 7: |
| |
| ```go |
| @@ -1515,7 +1663,7 @@ git commit -m "console: the reads route's gate and admission (DBF-9, DBF-10, DBF |
| **Interfaces:** |
| - Consumes: `dbReadAdmitted`, `s.dbf` (Tasks 3, 6); `dbreads.Line`, `Picked`, `Result`, `PickedKey`, `ResultKey`, `OutPrefix`, `ParseChunkKey`, `ChunkKey`, `CancelKey`, the wait and ended words; `dbreadstest.Start`, `Box.Agent` (plan 1). |
| - Consumes also: `(*Server).cancelDBRead` (Task 6), `s.dbf.call` (Task 3). |
| -- Produces: `(*Server).streamDBRead(w, r, rc, a)`; `type dbLineWriter` (with `until`), `(*dbLineWriter).tick(ctx, phase, poll, every) error`; `errDBClientGone`; `dbfEndSlack`; `type dbStream`; `dbfCall[T](st, ctx, phase, op) (T, error)`; `(*Server).readDBResult(ctx, id) *dbreads.Result`; test helpers `runBox(t, box)`, `(*dbfFixture).startBox(t, dir) *dbreadstest.Box`, `dbfOpen`, `(*dbfFixture).openHeld`, `(*dbfFixture).admitWithin`, `refusedTooMany`, `slowStore`, `scriptedBox`, `dbReadHookStore`, `strictStore`. |
| +- Produces: `(*Server).streamDBRead(w, r, rc, a)`; `type dbLineWriter` (with `until`), `(*dbLineWriter).tick(ctx, phase, poll, every) error`; `errDBClientGone`; `dbfEndSlack`; `type dbReadRelay`; `dbfCall[T](st, ctx, phase, op) (T, error)`; `(*Server).readDBResult(ctx, id) *dbreads.Result`; test helpers `runBox(t, box)`, `(*dbfFixture).startBox(t, dir) *dbreadstest.Box`, `dbfOpen`, `(*dbfFixture).openHeld`, `(*dbfFixture).admitWithin`, `refusedTooMany`, `slowStore`, `scriptedBox`, `dbReadHookStore`, `dbfHideStore`, `dbfHeldWriter`, `dbfDeadlineListener`, `dbfDeadlineConn`, `strictStore`. |
| |
| - [ ] **Step 1: Write the failing tests** |
| |
| @@ -1597,7 +1745,10 @@ func TestDBReadStreamsAQueryEndToEnd(t *testing.T) { |
| // the pickup window, a cancel, and the slot free again. |
| func TestDBReadEndsNotPickedAndCancels(t *testing.T) { |
| fx := newDBFFixture(t, true) |
| - fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.pickup = 10*time.Millisecond, 50*time.Millisecond, 400*time.Millisecond |
| + // issued_at is truncated to the second (production's semantics), so the |
| + // pickup window seen from admission is pickup minus up to a second: 1.4 s |
| + // leaves at least 400 ms, eight wait periods (plan review finding 30). |
| + fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.pickup = 10*time.Millisecond, 50*time.Millisecond, 1400*time.Millisecond |
| body, _ := queryBody(t, "SELECT 1") |
| res := fx.openRead(t, dbfHost, fx.member, body) |
| lines := readLines(t, res.Body) |
| @@ -1621,7 +1772,9 @@ func TestDBReadEndsNotPickedAndCancels(t *testing.T) { |
| // is flushed as it is written. |
| func TestDBReadLinesAreFlushedAsWritten(t *testing.T) { |
| fx := newDBFFixture(t, true) |
| - fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.pickup = 10*time.Millisecond, 100*time.Millisecond, time.Second |
| + // At least a second of pickup is left after issued_at's truncation: room |
| + // for the five lines below at 100 ms each (plan review finding 30). |
| + fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.pickup = 10*time.Millisecond, 100*time.Millisecond, 2*time.Second |
| body, _ := queryBody(t, "SELECT 1") |
| start := time.Now() |
| res := fx.openRead(t, dbfHost, fx.member, body) |
| @@ -2068,6 +2221,208 @@ func TestDBReadResultWaitsForItsLastChunk(t *testing.T) { |
| } |
| } |
| |
| +// dbfHideStore leaves chunk seq out of the first hide listings of a read's |
| +// chunks: a listing that lags the box's writes. |
| +type dbfHideStore struct { |
| + release.Store |
| + seq int64 |
| + mu sync.Mutex |
| + hide int |
| +} |
| + |
| +func (h *dbfHideStore) List(ctx context.Context, prefix string) ([]string, error) { |
| + keys, err := h.Store.List(ctx, prefix) |
| + if !strings.HasSuffix(prefix, "/out/") { |
| + return keys, err |
| + } |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + if h.hide == 0 { |
| + return keys, err |
| + } |
| + h.hide-- |
| + var out []string |
| + for _, k := range keys { |
| + if seq, ok := dbreads.ParseChunkKey(k, prefix); !ok || seq != h.seq { |
| + out = append(out, k) |
| + } |
| + } |
| + return out, err |
| +} |
| + |
| +// T18's completeness guard on its own (plan review finding 31): result.json |
| +// says last_seq 2 from the first round, and the first three listings miss |
| +// chunk 2. The result line waits until chunk 2 is relayed. Mutation: write |
| +// the result line without waiting for next > res.LastSeq (red: the result |
| +// comes after one chunk). |
| +func TestDBReadResultWaitsForAChunkTheListingMissed(t *testing.T) { |
| + fx := newDBFFixture(t, true) |
| + fx.srv.dbf.poll = 10 * time.Millisecond |
| + fx.srv.cfg.Releases = &dbfHideStore{Store: fx.rel, seq: 2, hide: 3} |
| + body, key := queryBody(t, "SELECT 1") |
| + res := fx.openRead(t, dbfHost, fx.member, body) |
| + defer res.Body.Close() |
| + br := bufio.NewReader(res.Body) |
| + first, _ := br.ReadBytes('\n') |
| + var adm dbreads.Line |
| + json.Unmarshal(first, &adm) |
| + box := scriptFor(t, fx.rel, adm.Admitted) |
| + box.chunk(1, false, "a\n") |
| + box.chunk(2, true, "b\n") |
| + box.result(2, "a\nb\n") |
| + box.picked(time.Now().Add(time.Minute)) |
| + lines := append([]dbreads.Line{adm}, readLines(t, br)...) |
| + chunks := 0 |
| + for _, l := range lines { |
| + if l.Chunk != 0 { |
| + chunks++ |
| + } |
| + if l.Result != nil && chunks != 2 { |
| + t.Fatalf("the result line came after %d chunks", chunks) |
| + } |
| + } |
| + if rows, _ := openAnswer(t, key, lines); string(rows) != "a\nb\n" { |
| + t.Fatalf("rows %q", rows) |
| + } |
| +} |
| + |
| +// dbfHeldWriter is a ResponseWriter that holds the write of a line |
| +// containing holdOn until until: a client that takes one line slowly. A |
| +// stand-in for a network client, in a unit test. |
| +type dbfHeldWriter struct { |
| + header http.Header |
| + holdOn string |
| + until time.Time |
| + mu sync.Mutex |
| + buf bytes.Buffer |
| +} |
| + |
| +func (w *dbfHeldWriter) Header() http.Header { return w.header } |
| +func (w *dbfHeldWriter) WriteHeader(int) {} |
| +func (w *dbfHeldWriter) Flush() {} |
| + |
| +func (w *dbfHeldWriter) Write(p []byte) (int, error) { |
| + if bytes.Contains(p, []byte(w.holdOn)) { |
| + time.Sleep(time.Until(w.until)) |
| + } |
| + w.mu.Lock() |
| + defer w.mu.Unlock() |
| + return w.buf.Write(p) |
| +} |
| + |
| +// Plan review finding 35: the owner's deadline holds for the result line |
| +// too. Every chunk and result.json are in the bucket, but the client takes |
| +// the last chunk's line until past the deadline: the read ends no-result, |
| +// and the result is not relayed. Mutation: drop the ownCtx check before the |
| +// result line (red: the result is relayed). |
| +func TestDBReadResultAfterTheDeadlineIsNotRelayed(t *testing.T) { |
| + fx := newDBFFixture(t, true) |
| + fx.srv.dbf.poll, fx.srv.dbf.grace = 10*time.Millisecond, 0 |
| + key, err := dbreads.NewReplyKey() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + id, err := dbreads.MintID() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + box := &scriptedBox{t: t, s: fx.rel, pub: key.PublicKey(), id: id} |
| + deadline := time.Now().Add(800 * time.Millisecond) |
| + box.chunk(1, true, "a\n") |
| + box.result(1, "a\n") |
| + box.picked(deadline) |
| + w := &dbfHeldWriter{header: http.Header{}, holdOn: `"chunk":1,`, until: deadline.Add(200 * time.Millisecond)} |
| + r := httptest.NewRequest(http.MethodPost, "/", nil) |
| + fx.srv.streamDBRead(w, r, http.NewResponseController(w), dbReadAdmitted{id: id, region: "uk", issued: time.Now().UTC().Truncate(time.Second)}) |
| + w.mu.Lock() |
| + lines := readLines(t, bytes.NewReader(w.buf.Bytes())) |
| + w.mu.Unlock() |
| + chunks := 0 |
| + for _, l := range lines { |
| + if l.Result != nil { |
| + t.Fatalf("a result relayed past the owner's deadline: %+v", l.Result) |
| + } |
| + if l.Chunk != 0 { |
| + chunks++ |
| + } |
| + } |
| + if last := lines[len(lines)-1]; chunks != 1 || last.Ended != dbreads.EndedNoResult { |
| + t.Fatalf("%d chunks, last line %+v", chunks, last) |
| + } |
| +} |
| + |
| +// dbfDeadlineListener records, for every read deadline the server sets on |
| +// its connections, how far ahead it was set. |
| +type dbfDeadlineListener struct { |
| + net.Listener |
| + mu sync.Mutex |
| + ahead []time.Duration |
| +} |
| + |
| +type dbfDeadlineConn struct { |
| + net.Conn |
| + l *dbfDeadlineListener |
| +} |
| + |
| +func (l *dbfDeadlineListener) Accept() (net.Conn, error) { |
| + c, err := l.Listener.Accept() |
| + if err != nil { |
| + return nil, err |
| + } |
| + return dbfDeadlineConn{Conn: c, l: l}, nil |
| +} |
| + |
| +func (c dbfDeadlineConn) SetReadDeadline(t time.Time) error { |
| + if !t.IsZero() { |
| + c.l.mu.Lock() |
| + c.l.ahead = append(c.l.ahead, time.Until(t)) |
| + c.l.mu.Unlock() |
| + } |
| + return c.Conn.SetReadDeadline(t) |
| +} |
| + |
| +// Plan review findings 13 and 38: admission reads under s.dbf.preHeader, |
| +// and once admitted the read deadline moves out to the read's bound plus a |
| +// minute. Observed on the connection itself: once a body has been read to |
| +// its end, net/http's background read clears the read deadline on its own, |
| +// so a stream that outlives ReadTimeout does not show the extension. |
| +// Mutations: drop the extension (red: no deadline past 22 minutes); drop the |
| +// preHeader deadline (red: none between 15 and 20 s; the server's own are |
| +// 10 s and 5 minutes). |
| +func TestDBReadSetsBothReadDeadlines(t *testing.T) { |
| + fx := newDBFFixture(t, true) |
| + fx.srv.dbf.poll, fx.srv.dbf.pickup = 10*time.Millisecond, 1500*time.Millisecond |
| + raw, err := net.Listen("tcp", "127.0.0.1:0") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + ln := &dbfDeadlineListener{Listener: raw} |
| + hs := NewHTTPServer(fx.srv) |
| + go hs.Serve(ln) |
| + t.Cleanup(func() { hs.Close() }) |
| + body, _ := queryBody(t, "SELECT 1") |
| + rawBody, _ := json.Marshal(body) |
| + req, _ := http.NewRequest(http.MethodPost, "http://"+raw.Addr().String()+strings.TrimPrefix(fx.readURL(dbfHost), fx.ts.URL), bytes.NewReader(rawBody)) |
| + req.Header.Set("Content-Type", "application/json") |
| + req.Header.Set("Authorization", "Bearer "+fx.member) |
| + res, err := http.DefaultClient.Do(req) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + io.Copy(io.Discard, res.Body) |
| + res.Body.Close() |
| + ln.mu.Lock() |
| + defer ln.mu.Unlock() |
| + var pre, extended bool |
| + for _, d := range ln.ahead { |
| + pre = pre || (d > 15*time.Second && d <= 20*time.Second) |
| + extended = extended || d > 22*time.Minute |
| + } |
| + if !pre || !extended { |
| + t.Fatalf("read deadlines set this far ahead: %v", ln.ahead) |
| + } |
| +} |
| + |
| // Review Focus 3: a picked deadline far in the future does not hold the |
| // request past its own bound. |
| func TestDBReadRequestBoundClampsAForgedDeadline(t *testing.T) { |
| @@ -2170,11 +2525,11 @@ func TestDBReadLateWritesChangeNothing(t *testing.T) { |
| } |
| ``` |
| |
| -Add imports `"crypto/sha256"`, `"encoding/base64"`, `"encoding/hex"`, `"slices"`, `"sync"`, `lsfile "github.com/benbjohnson/litestream/file"`, `dbdoortest`, `dbreadstest`. |
| +Add imports `"crypto/sha256"`, `"encoding/base64"`, `"encoding/hex"`, `"slices"`, `lsfile "github.com/benbjohnson/litestream/file"`, `dbdoortest`, `dbreadstest`. |
| |
| - [ ] **Step 2: Run them to verify they fail** |
| |
| -Run: `go test ./internal/console/ -run 'TestDBReadStreams|TestDBReadEnds|TestDBReadLines|TestDBReadAdmissionBounds|TestDBReadStalledStore|TestDBReadSlowRelay|TestDBReadResultWaits|TestDBReadRequestBound|TestDBReadIgnores|TestDBReadLateWrites' -count=1` |
| +Run: `go test ./internal/console/ -run 'TestDBReadStreams|TestDBReadEnds|TestDBReadLines|TestDBReadAdmissionBounds|TestDBReadStalledStore|TestDBReadSlowRelay|TestDBReadResultWaits|TestDBReadResultAfterTheDeadline|TestDBReadSetsBothReadDeadlines|TestDBReadRequestBound|TestDBReadIgnores|TestDBReadLateWrites' -count=1` |
| Expected: FAIL: the stub writes only `admitted`. |
| |
| - [ ] **Step 3: Write the stream** |
| @@ -2239,8 +2594,9 @@ func (l *dbLineWriter) tick(ctx context.Context, phase string, poll, every time. |
| return nil |
| } |
| |
| -// dbStream is one admitted read's relay, with what its last log line counts. |
| -type dbStream struct { |
| +// dbReadRelay is one admitted read's relay, with what its last log line |
| +// counts. (dbStream is apidb.go's streaming function: plan review finding 25.) |
| +type dbReadRelay struct { |
| s *Server |
| lw *dbLineWriter |
| chunks int |
| @@ -2252,7 +2608,7 @@ type dbStream struct { |
| // the request past its phase nor silences the stream (plan review finding |
| // 12). A store that ignores its context is left behind: its goroutine ends |
| // when the call does, into a buffered channel nobody reads. |
| -func dbfCall[T any](st *dbStream, ctx context.Context, phase string, op func(context.Context) (T, error)) (T, error) { |
| +func dbfCall[T any](st *dbReadRelay, ctx context.Context, phase string, op func(context.Context) (T, error)) (T, error) { |
| var zero T |
| cctx, cancel := context.WithTimeout(ctx, st.s.dbf.call) |
| defer cancel() |
| @@ -2312,7 +2668,7 @@ func (s *Server) streamDBRead(w http.ResponseWriter, r *http.Request, rc *http.R |
| w.Header().Set("Content-Type", "application/x-ndjson") |
| w.Header().Set("Cache-Control", "no-store") |
| w.WriteHeader(http.StatusOK) |
| - st := &dbStream{s: s, lw: &dbLineWriter{w: w, rc: rc, stall: s.dbf.stall, until: bound.Add(dbfEndSlack)}} |
| + st := &dbReadRelay{s: s, lw: &dbLineWriter{w: w, rc: rc, stall: s.dbf.stall, until: bound.Add(dbfEndSlack)}} |
| start, why := time.Now(), "client gone" |
| defer func() { |
| log.Printf("console: db read %s: relayed %d chunks (%d bytes) in %s, ended %s", |
| @@ -2458,11 +2814,11 @@ Waits for Task 1 (the CLI rows approved). |
| - Create: `dbforward.go` |
| - Create: `dbforward_test.go` |
| - Modify: `db.go:187-195` (`dbRestore` hands a forward marker back), `db.go:277-290` (`dbPull`), `db.go:374-382` (`dbQuery`) |
| -- Modify: `approved_db_forward_copy_test.go` (DBF-3 to DBF-8, DBF-11 to DBF-13, DBF-15, DBF-16) |
| +- Modify: `approved_db_forward_copy_test.go` (DBF-3 to DBF-8, DBF-11 to DBF-13, DBF-15, DBF-16, DBF-32) |
| |
| **Interfaces:** |
| - Consumes: `dbdoor.DoorError.Forward`, `dbdoor.Forward` (Task 2); `dbreads.NewReplyKey`, `EncodeReplyKey`, `ReadBody`, `Line`, `OpenChunk`, `OpenSummary`, `MaxLineBytes`, `CLIBound`, `StreamRows`, the state and code constants; `consoleClient.doStream`, `apiPath`, `apiErrorFromBody`, `dbQueryStatus`, `dbRestoreError`, `humanBytes`, `checkQuerySQL`, `stdout` (existing CLI code). |
| -- Produces: `errDBForward` (an error type carrying `dbdoor.Forward`); `dbQueryForward(ctx context.Context, t dbTarget, sqlText string, fwd dbdoor.Forward) error`; `readDBForward(ctx context.Context, body io.Reader, key *ecdh.PrivateKey, host, region string, status io.Writer) (dbForwardAnswer, error)`; `openDBForward(id string, key *ecdh.PrivateKey, sealed [][]byte, res dbreads.Result, host, region string) (dbForwardAnswer, error)`; `type dbForwardAnswer struct{ rows []byte; summary dbreads.Summary; capped bool }`; the `dbf*` CLI constants. |
| +- Produces: `errDBForward` (an error type carrying `dbdoor.Forward`); `dbQueryForward(ctx context.Context, t dbTarget, sqlText string, fwd dbdoor.Forward) error`; `readDBForward(ctx context.Context, body io.Reader, key *ecdh.PrivateKey, host, region string, status io.Writer) (dbForwardAnswer, error)`; `type dbfChunkBound`, `newDBFChunkBound(maxBytes, chunkBytes int64) *dbfChunkBound`, `(*dbfChunkBound).admit(sealed []byte) bool`; `openDBForward(id string, key *ecdh.PrivateKey, sealed [][]byte, res dbreads.Result, host, region string) (dbForwardAnswer, error)`; `type dbForwardAnswer struct{ rows []byte; summary dbreads.Summary; capped bool }`; the `dbf*` CLI constants. |
| |
| - [ ] **Step 1: Write the failing tests** |
| |
| @@ -2479,8 +2835,11 @@ import ( |
| "encoding/hex" |
| "encoding/json" |
| "fmt" |
| + "net/http" |
| + "net/http/httptest" |
| "os" |
| "strings" |
| + "sync/atomic" |
| "testing" |
| "time" |
| |
| @@ -2677,6 +3036,66 @@ func TestForwardedQueryCtrlCPrintsNothing(t *testing.T) { |
| } |
| } |
| |
| +// Plan review finding 37: what the CLI holds before a result is bounded by |
| +// count as well as bytes, and a malformed chunk is refused before it is |
| +// kept. Each stream is otherwise one the CLI would accept, so each case is |
| +// red under its own mutation of dbfChunkBound.admit: drop the size floor |
| +// ("a chunk shorter than its seal" then ends DBF-12, the stream having no |
| +// result, instead of DBF-13 at once); drop the empty-chunk rule or the |
| +// count bound (the stream then opens and prints). |
| +func TestForwardedQueryBoundsChunkCountAndSize(t *testing.T) { |
| + t.Parallel() |
| + key, _ := dbreads.NewReplyKey() |
| + unchecked := fmt.Sprintf(dbfUnchecked, "uk") |
| + many := newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes).maxChunks + 1 |
| + for name, write := range map[string]func(s *stream){ |
| + "a chunk shorter than its seal": func(s *stream) { |
| + s.line(dbreads.Line{Chunk: 1, Data: []byte("short")}) |
| + }, |
| + "a chunk after an empty one": func(s *stream) { |
| + s.chunk(1, false, nil) |
| + s.chunk(2, true, []byte("n\n")) |
| + s.result(dbreads.StateDone, "", 2, summaryOf("n\n")) |
| + }, |
| + "more chunks than a read can make": func(s *stream) { |
| + for seq := int64(1); seq <= many; seq++ { |
| + s.chunk(seq, seq == many, []byte("x")) |
| + } |
| + s.result(dbreads.StateDone, "", many, summaryOf(strings.Repeat("x", int(many)))) |
| + }, |
| + } { |
| + s := newStream(t, key) |
| + write(s) |
| + var status bytes.Buffer |
| + if _, err := readDBForward(context.Background(), &s.buf, key, "shop.example.com", "uk", &status); err == nil || err.Error() != unchecked { |
| + t.Errorf("%s: %v", name, err) |
| + } |
| + } |
| +} |
| + |
| +// Plan review finding 36: SQL that JSON would rewrite is refused before |
| +// anything is sent. Mutation: drop the check (red: the request goes out). |
| +func TestForwardedQueryRefusesSQLThatIsNotUTF8(t *testing.T) { |
| + t.Parallel() |
| + var sent atomic.Bool |
| + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + sent.Store(true) |
| + http.Error(w, "no", http.StatusTeapot) |
| + })) |
| + defer ts.Close() |
| + target := dbTarget{sess: &apiSession{c: newConsoleClient(ts.URL, "tok"), console: ts.URL}, acct: apiAcct{Seg: "acme"}, |
| + app: "shop", host: "shop.example.com"} |
| + for _, q := range []string{"SELECT '\xff'", "SELECT '\uFFFD'"} { |
| + err := dbQueryForward(context.Background(), target, q, dbdoor.Forward{Region: "uk", Ops: []string{dbreads.OpQuery}}) |
| + if err == nil || err.Error() != fmt.Sprintf(dbfNotUTF8, "uk") { |
| + t.Errorf("%q: %v", q, err) |
| + } |
| + } |
| + if sent.Load() { |
| + t.Fatal("a query JSON would rewrite was sent") |
| + } |
| +} |
| + |
| // cutReader ends like a connection the context closed. |
| type cutReader struct{ r *bytes.Buffer } |
| |
| @@ -2742,6 +3161,7 @@ const ( |
| dbfUnchecked = "the answer from region %s did not check out, so carlos db showed none of it. try again." // DBF-13 |
| dbfChanged = "%s changed while the read was waiting. try again." // DBF-15 |
| dbfMovingCLI = "%s is moving between regions. read its database once the move has finished." // DBF-16 |
| + dbfNotUTF8 = "this query cannot be sent to region %s, because it holds bytes that are not UTF-8 text or the character U+FFFD. fix the query and run it again." // DBF-32 |
| ) |
| |
| // errDBForward is the db door's forward marker, handed back by dbRestore so |
| @@ -2760,6 +3180,13 @@ type dbForwardAnswer struct { |
| |
| // dbQueryForward is spec §5.5 steps 2 and 8 for a query. |
| func dbQueryForward(ctx context.Context, t dbTarget, sqlText string, fwd dbdoor.Forward) error { |
| + // json.Marshal would send U+FFFD for bytes that are not UTF-8, and the |
| + // box would run SQL nobody typed; the console refuses U+FFFD for the |
| + // same reason (plan review finding 36). First, so no other check can |
| + // read the bytes differently. |
| + if !utf8.ValidString(sqlText) || strings.ContainsRune(sqlText, utf8.RuneError) { |
| + return fmt.Errorf(dbfNotUTF8, fwd.Region) |
| + } |
| if err := checkQuerySQL(sqlText); err != nil { |
| return err |
| } |
| @@ -2804,6 +3231,35 @@ func dbQueryForward(ctx context.Context, t dbTarget, sqlText string, fwd dbdoor. |
| return nil |
| } |
| |
| +// dbfChunkBound is what a reader keeps before a result arrives (plan review |
| +// finding 37). Every chunk is at least a sealed empty one; only the last |
| +// may be empty, so nothing may follow an empty chunk; there are at most as |
| +// many chunks as a read can make (full ones for the payload, plus one per |
| +// flush interval over the owner's deadline, plus the last); and their |
| +// plaintext, ciphertext less each seal, stays within the payload cap. |
| +type dbfChunkBound struct { |
| + maxBytes, maxChunks int64 |
| + n, plain int64 |
| + sawEmpty bool |
| +} |
| + |
| +func newDBFChunkBound(maxBytes, chunkBytes int64) *dbfChunkBound { |
| + return &dbfChunkBound{maxBytes: maxBytes, |
| + maxChunks: (maxBytes+chunkBytes-1)/chunkBytes + int64(dbreads.OwnerDeadline/dbreads.ChunkEvery) + 1} |
| +} |
| + |
| +// admit says whether a chunk may be kept; false is DBF-13. |
| +func (b *dbfChunkBound) admit(sealed []byte) bool { |
| + size := int64(len(sealed)) - int64(dbreads.SealOverhead) |
| + if size < 0 || b.sawEmpty || b.n >= b.maxChunks || b.plain+size > b.maxBytes { |
| + return false |
| + } |
| + b.n++ |
| + b.plain += size |
| + b.sawEmpty = size == 0 |
| + return true |
| +} |
| + |
| // readDBForward reads the reads route's lines (spec §5.5 steps 6 to 8). It |
| // says DBF-11 once admitted, buffers every chunk in sequence, and opens and |
| // checks the answer only when the result arrives. |
| @@ -2813,7 +3269,7 @@ func readDBForward(ctx context.Context, body io.Reader, key *ecdh.PrivateKey, ho |
| sc.Buffer(make([]byte, 64<<10), dbreads.MaxLineBytes) |
| var id, ended string |
| var sealed [][]byte |
| - var total int64 |
| + bound := newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes) |
| for sc.Scan() { |
| var l dbreads.Line |
| if err := json.Unmarshal(sc.Bytes(), &l); err != nil { |
| @@ -2827,8 +3283,7 @@ func readDBForward(ctx context.Context, body io.Reader, key *ecdh.PrivateKey, ho |
| id = l.Admitted |
| fmt.Fprintln(status, fmt.Sprintf(dbfReading, host, region)) |
| case l.Chunk != 0: |
| - total += int64(len(l.Data)) |
| - if id == "" || l.Chunk != int64(len(sealed))+1 || total > dbreads.MaxOutputBytes+int64(len(sealed)+1)*int64(dbreads.SealOverhead) { |
| + if id == "" || l.Chunk != int64(len(sealed))+1 || !bound.admit(l.Data) { |
| return dbForwardAnswer{}, unchecked |
| } |
| sealed = append(sealed, l.Data) |
| @@ -2953,7 +3408,7 @@ and in `dbPull`: |
| } |
| ``` |
| |
| -Add `"slices"` and the `dbreads` import to `db.go`. Add the CLI ids (DBF-3 to DBF-8, DBF-11, DBF-12, DBF-13, DBF-15, DBF-16) to `dbForwardCopyBuilt`. |
| +Add `"slices"` and the `dbreads` import to `db.go`. Add the CLI ids (DBF-3 to DBF-8, DBF-11, DBF-12, DBF-13, DBF-15, DBF-16, DBF-32) to `dbForwardCopyBuilt`, and `"unicode/utf8"` to `dbforward.go`'s imports (and `"strings"` if it is not there). |
| |
| - [ ] **Step 4: Run the tests and the gate** |
| |
| @@ -3015,7 +3470,7 @@ git commit -m "docs: carlos db reads another region's instances (DBF-30)" |
| |
| **Interfaces:** |
| - Consumes: `buildCarlos`, `useTempCredentials`, `clearDeploymentEnv`, `carlosTestEnv`, `e2eDeviceLogin`, `loadCredentials`, `credential`, `replicaS3` (`db_e2e_test.go`), `dbdoortest.Sync`, `console.Open`, `console.NewServer`, `console.NewHTTPServer`, `console.OpsAccountID`, `console.RoleOwner`, `edge.New`, `registry.Open`, `registry.Route`, `dbreadstest.Start`, `dbreads.*`, `dbdoor.RestoreLocal`, `dbdoor.NewClient`, `dbdoor.Base`, `lease.ReadReplicaClient`. |
| -- Produces: the harness `dbfE2E` and `newDBFE2E(t *testing.T, o dbfE2EOptions) *dbfE2E`; `(*dbfE2E).run(t, args...) (stdout, stderr string, err error)`; `(*dbfE2E).betweenTheSyncs(t) string`; `(*dbfE2E).doer(host string) dbdoor.Doer`; `(*dbfE2E).scriptBox(t, script func(*e2eScript)) (stop func())`; `e2eScript` with `picked(deadline)`, `chunk(seq, last, plain)`, `result(last, rows)`; `throughTheEdge(readTimeout time.Duration, edgeURL *string)`; `slowRestore(d time.Duration) dbreads.WorkerOptions`; and the tests below. |
| +- Produces: the harness `dbfE2E` and `newDBFE2E(t *testing.T, o dbfE2EOptions) *dbfE2E`; `(*dbfE2E).run(t, args...) (stdout, stderr string, err error)`; `(*dbfE2E).betweenTheSyncs(t) string`; `(*dbfE2E).doer(host string) dbdoor.Doer`; `(*dbfE2E).scriptBox(t, script func(*e2eScript)) (stop func())`; `e2eScript` with `picked(deadline)`, `chunk(seq, last, plain)`, `result(last, rows)`; `throughTheEdge(readTimeout time.Duration, edgeURL *string, tap **dbfTap)`; `dbfTap`, `newDBFTap(t, target string, connect bool) *dbfTap`, `(*dbfTap).longestGap(start, end time.Time) time.Duration`; `checkProgressAtTheCLI(t, e, tap, atLeast time.Duration)`; `slowRestore(d time.Duration) dbreads.WorkerOptions`; and the tests below. |
| |
| - [ ] **Step 1: Write the harness** |
| |
| @@ -3052,6 +3507,7 @@ import ( |
| "os/exec" |
| "path/filepath" |
| "strings" |
| + "sync" |
| "testing" |
| "time" |
| |
| @@ -3355,7 +3811,7 @@ func (e *dbfE2E) scriptBox(t *testing.T, script func(*e2eScript)) func() { |
| // 127.0.0.1, its ReadTimeout set to readTimeout, behind a real edge route |
| // for console.test. The CLI reaches it by that name through HTTP_PROXY, |
| // which stands in for DNS. |
| -func throughTheEdge(readTimeout time.Duration, edgeURL *string) func(t *testing.T, h http.Handler) (string, []string) { |
| +func throughTheEdge(readTimeout time.Duration, edgeURL *string, tap **dbfTap) func(t *testing.T, h http.Handler) (string, []string) { |
| return func(t *testing.T, h http.Handler) (string, []string) { |
| ln, err := net.Listen("tcp", "127.0.0.1:0") |
| if err != nil { |
| @@ -3382,7 +3838,146 @@ func throughTheEdge(readTimeout time.Duration, edgeURL *string) func(t *testing. |
| es := httptest.NewServer(ed) |
| t.Cleanup(es.Close) |
| *edgeURL = es.URL |
| - return "http://console.test", []string{"HTTP_PROXY=" + es.URL, "NO_PROXY="} |
| + // The CLI's proxy is a tap in front of the edge, so the test sees |
| + // what reaches the CLI's own socket (plan review finding 34). |
| + tp := newDBFTap(t, es.Listener.Addr().String(), false) |
| + *tap = tp |
| + return "http://console.test", []string{"HTTP_PROXY=http://" + tp.addr, "NO_PROXY="} |
| + } |
| +} |
| + |
| +// dbfTap stands between the CLI and the first edge, forwarding bytes both |
| +// ways and noting when each read from the edge's side was passed to the |
| +// CLI's socket: what the CLI process received, and when. With connect it is |
| +// an HTTPS proxy (CONNECT), which is how the CLI reaches a TLS edge by name |
| +// here; without, a plain forwarder for HTTP_PROXY. |
| +type dbfTap struct { |
| + addr string |
| + mu sync.Mutex |
| + down []time.Time |
| +} |
| + |
| +func newDBFTap(t *testing.T, target string, connect bool) *dbfTap { |
| + t.Helper() |
| + ln, err := net.Listen("tcp", "127.0.0.1:0") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { ln.Close() }) |
| + tp := &dbfTap{addr: ln.Addr().String()} |
| + go func() { |
| + for { |
| + c, err := ln.Accept() |
| + if err != nil { |
| + return |
| + } |
| + go tp.serve(c, target, connect) |
| + } |
| + }() |
| + return tp |
| +} |
| + |
| +func (tp *dbfTap) serve(c net.Conn, target string, connect bool) { |
| + defer c.Close() |
| + var in io.Reader = c |
| + if connect { |
| + br := bufio.NewReader(c) |
| + req, err := http.ReadRequest(br) |
| + if err != nil || req.Method != http.MethodConnect { |
| + return |
| + } |
| + if _, err := io.WriteString(c, "HTTP/1.1 200 Connection established\r\n\r\n"); err != nil { |
| + return |
| + } |
| + in = br |
| + } |
| + up, err := net.Dial("tcp", target) |
| + if err != nil { |
| + return |
| + } |
| + defer up.Close() |
| + go func() { |
| + io.Copy(up, in) |
| + up.(*net.TCPConn).CloseWrite() |
| + }() |
| + buf := make([]byte, 32<<10) |
| + for { |
| + n, err := up.Read(buf) |
| + if n > 0 { |
| + tp.mu.Lock() |
| + tp.down = append(tp.down, time.Now()) |
| + tp.mu.Unlock() |
| + if _, werr := c.Write(buf[:n]); werr != nil { |
| + return |
| + } |
| + } |
| + if err != nil { |
| + return |
| + } |
| + } |
| +} |
| + |
| +// longestGap is the longest silence at the CLI's socket between start and |
| +// end, both of which count as arrivals. |
| +func (tp *dbfTap) longestGap(start, end time.Time) time.Duration { |
| + tp.mu.Lock() |
| + defer tp.mu.Unlock() |
| + prev, longest := start, time.Duration(0) |
| + for _, at := range append(append([]time.Time(nil), tp.down...), end) { |
| + if at.Before(start) || at.After(end) { |
| + continue |
| + } |
| + if d := at.Sub(prev); d > longest { |
| + longest = d |
| + } |
| + prev = at |
| + } |
| + return longest |
| +} |
| + |
| +// checkProgressAtTheCLI is T7's witness at the CLI process itself (plan |
| +// review finding 34): it runs the real CLI and needs DBF-11 on its stderr |
| +// within a second of the process starting (stricter than the spec's "of the |
| +// request": the process starts before it sends anything), no silence over |
| +// 16 s (15 s and a poll) at its socket, the read lasting at least atLeast, and |
| +// its rows. atLeast is the quiet restore's length. |
| +func checkProgressAtTheCLI(t *testing.T, e *dbfE2E, tap *dbfTap, atLeast time.Duration) { |
| + t.Helper() |
| + cmd := exec.Command(e.bin, "db", "query", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT count(*) AS n FROM orders") |
| + cmd.Env = carlosTestEnv(t, append([]string{"CARLOS_CREDENTIALS=" + e.creds}, e.env...)...) |
| + cmd.Dir = t.TempDir() |
| + var out bytes.Buffer |
| + cmd.Stdout = &out |
| + pipe, err := cmd.StderrPipe() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + start := time.Now() |
| + if err := cmd.Start(); err != nil { |
| + t.Fatal(err) |
| + } |
| + var stderr []string |
| + var reading time.Time |
| + sc := bufio.NewScanner(pipe) |
| + for sc.Scan() { |
| + if reading.IsZero() && strings.Contains(sc.Text(), fmt.Sprintf(dbfReading, dbfE2EFar, "uk")) { |
| + reading = time.Now() |
| + } |
| + stderr = append(stderr, sc.Text()) |
| + } |
| + err = cmd.Wait() |
| + end := time.Now() |
| + if err != nil || out.String() != "n\n3\n" { |
| + t.Fatalf("%q %v\n%s", out.String(), err, strings.Join(stderr, "\n")) |
| + } |
| + if reading.IsZero() || reading.Sub(start) > time.Second { |
| + t.Fatalf("DBF-11 after %s: %q", reading.Sub(start), stderr) |
| + } |
| + if gap := tap.longestGap(start, end); gap > 16*time.Second { |
| + t.Fatalf("%s of silence at the CLI's socket", gap) |
| + } |
| + if took := end.Sub(start); took < atLeast { |
| + t.Fatalf("the read took %s, under %s: it did not hold the request", took, atLeast) |
| } |
| } |
| |
| @@ -3441,7 +4036,8 @@ func TestDBForwardParityThroughTheCLI(t *testing.T) { |
| // gets its rows through the same path. |
| func TestDBForwardLongRequestThroughTheEdge(t *testing.T) { |
| var edgeURL string |
| - e := newDBFE2E(t, dbfE2EOptions{forward: true, serve: throughTheEdge(2*time.Second, &edgeURL), worker: slowRestore(20 * time.Second)}) |
| + var tap *dbfTap |
| + e := newDBFE2E(t, dbfE2EOptions{forward: true, serve: throughTheEdge(2*time.Second, &edgeURL, &tap), worker: slowRestore(20 * time.Second)}) |
| key, _ := dbreads.NewReplyKey() |
| body, _ := json.Marshal(dbreads.ReadBody{Op: dbreads.OpQuery, SQL: "SELECT count(*) AS n FROM orders", ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())}) |
| req, _ := http.NewRequest(http.MethodPost, edgeURL+"/api/cli/apps/"+release.OpsAccount+"/shop/db/"+dbfE2EFar+"/reads", bytes.NewReader(body)) |
| @@ -3477,10 +4073,9 @@ func TestDBForwardLongRequestThroughTheEdge(t *testing.T) { |
| if !sawResult || time.Since(start) < 20*time.Second { |
| t.Fatalf("result %v after %s", sawResult, time.Since(start)) |
| } |
| - out, stderr, err := e.run(t, "db", "query", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT count(*) AS n FROM orders") |
| - if err != nil || out != "n\n3\n" { |
| - t.Fatalf("%q %v\n%s", out, err, stderr) |
| - } |
| + // The same through the real CLI process, observed at its own stderr and |
| + // socket (plan review finding 34). |
| + checkProgressAtTheCLI(t, e, tap, 20*time.Second) |
| } |
| |
| // T7, unscaled: a read that holds the request for 21 minutes through the |
| @@ -3494,7 +4089,8 @@ func TestDBForwardTwentyOneMinuteRead(t *testing.T) { |
| t.Skip("set CARLOS_DBF_LONG=1 to run the unscaled 21-minute read, once, before handoff") |
| } |
| var edgeURL string |
| - e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true, serve: throughTheEdge(5*time.Minute, &edgeURL)}) |
| + var tap *dbfTap |
| + e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true, serve: throughTheEdge(5*time.Minute, &edgeURL, &tap)}) |
| stop := e.scriptBox(t, func(b *e2eScript) { |
| time.Sleep(110 * time.Second) |
| b.picked(time.Now().Add(20 * time.Minute)) |
| @@ -3569,13 +4165,28 @@ func TestDBForwardPrintsNothingForABadAnswer(t *testing.T) { |
| }) |
| } |
| } |
| + |
| +// Plan review finding 36, through the binary: a query holding a byte that |
| +// is not UTF-8 is refused with DBF-32 before anything is sent, so no altered |
| +// query reaches the bucket. (An argv may carry any bytes on Linux, where |
| +// this runs.) |
| +func TestDBForwardRefusesSQLThatIsNotUTF8(t *testing.T) { |
| + e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true}) |
| + out, stderr, err := e.run(t, "db", "query", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT 'caf\xe9'") |
| + if err == nil || out != "" || !strings.Contains(stderr, fmt.Sprintf(dbfNotUTF8, "uk")) { |
| + t.Fatalf("%q %q %v", out, stderr, err) |
| + } |
| + if keys, _ := e.store.List(context.Background(), dbreads.Prefix); len(keys) != 0 { |
| + t.Fatalf("a query JSON would rewrite reached the bucket: %v", keys) |
| + } |
| +} |
| ``` |
| |
| - [ ] **Step 3: Run them** |
| |
| Run (sandbox off; they build the CLI binary): |
| ```bash |
| -go test . -run 'TestDBForwardParityThroughTheCLI|TestDBForwardMixedVersions|TestDBForwardPrintsNothingForABadAnswer|TestDBForwardLongRequestThroughTheEdge' -count=1 -v |
| +go test . -run 'TestDBForwardParityThroughTheCLI|TestDBForwardMixedVersions|TestDBForwardPrintsNothingForABadAnswer|TestDBForwardRefusesSQLThatIsNotUTF8|TestDBForwardLongRequestThroughTheEdge' -count=1 -v |
| ``` |
| Expected: PASS; `TestDBForwardTwentyOneMinuteRead` skips. If the edge answers the CLI with 404, the request reached it with a host other than `console.test`: print `r.Host` in a wrapper around `ed` to see what Go's proxy client sent. |
| |
| @@ -3588,63 +4199,54 @@ git commit -m "db forward: T6, T7, T8 and T11 through the real CLI binary and co |
| |
| --- |
| |
| -### Task 11: T7's backhaul half |
| +### Task 11: T7's backhaul half, through the real CLI and console |
| |
| **Files:** |
| -- Create: `internal/edge/dbforward_backhaul_test.go` |
| +- Create: `internal/edge/dbforward_backhaul_helper_test.go` |
| +- Modify: `dbforward_e2e_test.go` |
| |
| **Interfaces:** |
| -- Consumes (package `edge` test helpers in `backhaul_e2e_test.go`): `newE2ECA`, `(*e2eCA).publish`, `newBackhaulEdge`, `refreshHomeMap`, `MintMeshKey`, `putJSON`, `e2eQuietLogs`, `backhaulPort`, `backhaulRootCAs`, `backhaulHeaderTimeout`. |
| +- Consumes (package `edge` test helpers in `backhaul_e2e_test.go`): `newE2ECA`, `(*e2eCA).publish`, the `e2eCA.root` field, `newBackhaulEdge`, `(*backhaulEdge).tlsAddr`, `.mirror`, `.homes`, `refreshHomeMap`, `MintMeshKey`, `putJSON`, `e2eQuietLogs`, the seams `backhaulPort` and `backhaulRootCAs`. From Task 10: `newDBFE2E`, `slowRestore`, `dbfTap`, `newDBFTap`, `checkProgressAtTheCLI`. |
| +- Produces: `TestDBForwardBackhaulPairHelper` (a helper process, skipped unless `DBF_BACKHAUL_HELPER=1`); `throughBackhaul(readTimeout time.Duration, tap **dbfTap)`; `TestDBForwardThroughTheBackhaulHop`. |
| + |
| +Spec T7 needs the real CLI and the real console through the backhaul hop (plan review finding 34). The console cannot be imported by package `edge`'s tests (the console imports the edge, a cycle), and the backhaul harness is package `edge` test code that nothing else can import. So the acceptance test lives in the root package with the real CLI and console, and the two edges run in package `edge`'s own test binary as a helper process, built by the root test with `go test -c`. |
| |
| -The console cannot be imported here (the console imports the edge, so a package `edge` test importing it is a cycle). The upstream is a stand-in that streams the forwarded read's shape and timing: headers and `admitted` at once, then a line every 5 s for longer than the backhaul hop's 30 s header bound. The real console's flushing is T7a's (Task 10) and Task 7's. |
| +- [ ] **Step 1: Write the helper** |
| |
| -- [ ] **Step 1: Write the test** |
| +Create `internal/edge/dbforward_backhaul_helper_test.go`: |
| |
| ```go |
| package edge |
| |
| import ( |
| - "bufio" |
| "context" |
| + "encoding/pem" |
| "fmt" |
| + "io" |
| "net" |
| - "net/http" |
| - "net/http/httptest" |
| - "strings" |
| + "os" |
| + "path/filepath" |
| "testing" |
| - "time" |
| |
| "github.com/carlosframework/platform/internal/release" |
| ) |
| |
| const dbfBackhaulHost = "console.backhaul.test" |
| |
| -// dbfStream is a stand-in for the console's reads route: not acceptance. It |
| -// writes and flushes admitted with the headers, then a wait line every |
| -// 5 s, and a result after total. |
| -func dbfStream(total time.Duration) *httptest.Server { |
| - return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| - w.Header().Set("Content-Type", "application/x-ndjson") |
| - w.Header().Set("Cache-Control", "no-store") |
| - w.WriteHeader(http.StatusOK) |
| - rc := http.NewResponseController(w) |
| - fmt.Fprintln(w, `{"admitted":"dbr_aaaaaaaaaaaaaaaaaaaaaaaaaa"}`) |
| - rc.Flush() |
| - end := time.Now().Add(total) |
| - for time.Now().Before(end) { |
| - time.Sleep(5 * time.Second) |
| - fmt.Fprintln(w, `{"wait":"reading"}`) |
| - rc.Flush() |
| - } |
| - fmt.Fprintln(w, `{"result":{"state":"done","code":"","box":"uk-1","last_seq":1,"sealed":""}}`) |
| - })) |
| -} |
| - |
| -// T7, the backhaul half (codex 19): a forwarded read that reaches the |
| -// console through another region's edge survives the hop's 30 s header |
| -// bound, because the headers leave at once, and every line arrives as it is |
| -// written. |
| -func TestDBForwardStreamSurvivesTheBackhaulHop(t *testing.T) { |
| +// TestDBForwardBackhaulPairHelper is not a test of its own: the root |
| +// package's TestDBForwardThroughTheBackhaulHop builds this package's test |
| +// binary and runs this as a helper process. It stands up two real edges |
| +// over one bucket, as TestBackhaulE2EAcrossTwoEdges does: annie, home of |
| +// console.backhaul.test, routed to DBF_BACKHAUL_UPSTREAM (the real console |
| +// the parent serves), and betty, which serves no host and so backhauls every |
| +// request for it to annie. It writes the test CA to DBF_BACKHAUL_DIR/ca.pem, |
| +// prints betty's TLS address on a line starting "DBF-BACKHAUL ", and serves |
| +// until its stdin closes. |
| +func TestDBForwardBackhaulPairHelper(t *testing.T) { |
| + if os.Getenv("DBF_BACKHAUL_HELPER") != "1" { |
| + t.Skip("a helper process for the root package's TestDBForwardThroughTheBackhaulHop") |
| + } |
| + upstream, dir := os.Getenv("DBF_BACKHAUL_UPSTREAM"), os.Getenv("DBF_BACKHAUL_DIR") |
| e2eQuietLogs(t) |
| bucket := t.TempDir() |
| seed := release.NewDirStore(bucket) |
| @@ -3653,16 +4255,20 @@ func TestDBForwardStreamSurvivesTheBackhaulHop(t *testing.T) { |
| } |
| ca := newE2ECA(t) |
| ca.publish(t, seed, dbfBackhaulHost) |
| - upstream := dbfStream(backhaulHeaderTimeout + 10*time.Second) |
| - t.Cleanup(upstream.Close) |
| - home := newBackhaulEdge(t, bucket, "annie", ca, nil, map[string]string{dbfBackhaulHost: upstream.Listener.Addr().String()}) |
| + home := newBackhaulEdge(t, bucket, "annie", ca, nil, map[string]string{dbfBackhaulHost: upstream}) |
| sender := newBackhaulEdge(t, bucket, "betty", ca, nil, nil) |
| - eip, port, _ := net.SplitHostPort(home.tlsAddr) |
| + eip, port, err := net.SplitHostPort(home.tlsAddr) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| oldPort, oldCAs := backhaulPort, backhaulRootCAs |
| backhaulPort, backhaulRootCAs = port, ca.pool |
| t.Cleanup(func() { backhaulPort, backhaulRootCAs = oldPort, oldCAs }) |
| putJSON(t, seed, "control/edges/annie.json", fmt.Sprintf(`{"v":1,"box":"annie","eip":%q,"hosts":{%q:{"role":"home"}}}`, eip, dbfBackhaulHost)) |
| - bEIP, _, _ := net.SplitHostPort(sender.tlsAddr) |
| + bEIP, _, err := net.SplitHostPort(sender.tlsAddr) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| putJSON(t, seed, "control/edges/betty.json", fmt.Sprintf(`{"v":1,"box":"betty","eip":%q,"hosts":{}}`, bEIP)) |
| for _, be := range []*backhaulEdge{home, sender} { |
| if err := be.mirror.Refresh(context.Background()); err != nil { |
| @@ -3670,48 +4276,113 @@ func TestDBForwardStreamSurvivesTheBackhaulHop(t *testing.T) { |
| } |
| refreshHomeMap(t, be.homes) |
| } |
| - client := sender.client |
| - client.Timeout = 0 // the stream outlives the harness's 30 s client bound |
| - start := time.Now() |
| - res, err := client.Post("https://"+dbfBackhaulHost+"/api/cli/apps/bab/shop/db/shop.example.com/reads", "application/json", strings.NewReader(`{}`)) |
| - if err != nil { |
| + caPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: ca.root.Raw}) |
| + if err := os.WriteFile(filepath.Join(dir, "ca.pem"), caPEM, 0o600); err != nil { |
| t.Fatal(err) |
| } |
| - defer res.Body.Close() |
| - if time.Since(start) > 2*time.Second || res.StatusCode != http.StatusOK { |
| - t.Fatalf("headers after %s: %d", time.Since(start), res.StatusCode) |
| - } |
| - br := bufio.NewReader(res.Body) |
| - prev, lines := time.Now(), 0 |
| - for { |
| - line, err := br.ReadString('\n') |
| + fmt.Println("DBF-BACKHAUL " + sender.tlsAddr) |
| + io.Copy(io.Discard, os.Stdin) |
| +} |
| +``` |
| + |
| +Run (sandbox off): `go test ./internal/edge/ -run 'TestDBForwardBackhaulPairHelper' -count=1` |
| +Expected: `ok` with the test skipped. |
| + |
| +- [ ] **Step 2: Write the acceptance test** |
| + |
| +Append to `dbforward_e2e_test.go`: |
| + |
| +```go |
| +// throughBackhaul serves the console's handler with console.NewHTTPServer |
| +// (its ReadTimeout set to readTimeout) on 127.0.0.1, behind two real edges |
| +// that package edge's test binary runs as a helper process: the CLI reaches |
| +// betty, which backhauls to annie, the console's home. The CLI finds betty |
| +// through a CONNECT proxy (a dbfTap, standing in for DNS) and trusts the |
| +// edges' test CA through SSL_CERT_FILE. |
| +func throughBackhaul(readTimeout time.Duration, tap **dbfTap) func(t *testing.T, h http.Handler) (string, []string) { |
| + return func(t *testing.T, h http.Handler) (string, []string) { |
| + ln, err := net.Listen("tcp", "127.0.0.1:0") |
| if err != nil { |
| - break |
| + t.Fatal(err) |
| + } |
| + hs := console.NewHTTPServer(h) |
| + hs.ReadTimeout = readTimeout |
| + go hs.Serve(ln) |
| + t.Cleanup(func() { hs.Close() }) |
| + dir := t.TempDir() |
| + bin := filepath.Join(dir, "edge.test") |
| + if out, err := exec.Command("go", "test", "-c", "-o", bin, "./internal/edge").CombinedOutput(); err != nil { |
| + t.Fatalf("build the edge test binary: %v\n%s", err, out) |
| } |
| - if gap := time.Since(prev); gap > 7*time.Second { |
| - t.Fatalf("line %d arrived %s after the last: held by the hop", lines, gap) |
| + cmd := exec.Command(bin, "-test.run", "^TestDBForwardBackhaulPairHelper$", "-test.v") |
| + cmd.Env = append(os.Environ(), "DBF_BACKHAUL_HELPER=1", "DBF_BACKHAUL_UPSTREAM="+ln.Addr().String(), "DBF_BACKHAUL_DIR="+dir) |
| + stdin, err := cmd.StdinPipe() |
| + if err != nil { |
| + t.Fatal(err) |
| } |
| - prev, lines = time.Now(), lines+1 |
| - if strings.Contains(line, `"result"`) { |
| - break |
| + pr, pw := io.Pipe() |
| + cmd.Stdout = pw |
| + var helperErr bytes.Buffer |
| + cmd.Stderr = &helperErr |
| + if err := cmd.Start(); err != nil { |
| + t.Fatal(err) |
| } |
| - } |
| - if elapsed := time.Since(start); elapsed < backhaulHeaderTimeout || lines < 8 { |
| - t.Fatalf("%d lines over %s", lines, elapsed) |
| - } |
| + t.Cleanup(func() { |
| + stdin.Close() |
| + cmd.Wait() |
| + pw.Close() |
| + if t.Failed() { |
| + t.Logf("backhaul helper stderr:\n%s", helperErr.String()) |
| + } |
| + }) |
| + lines := bufio.NewScanner(pr) |
| + sender := "" |
| + for lines.Scan() { |
| + if addr, ok := strings.CutPrefix(lines.Text(), "DBF-BACKHAUL "); ok { |
| + sender = addr |
| + break |
| + } |
| + } |
| + if sender == "" { |
| + t.Fatalf("the backhaul helper never came up:\n%s", helperErr.String()) |
| + } |
| + go func() { |
| + for lines.Scan() { |
| + } |
| + }() |
| + tp := newDBFTap(t, sender, true) |
| + *tap = tp |
| + return "https://console.backhaul.test", []string{"HTTPS_PROXY=http://" + tp.addr, "NO_PROXY=", "SSL_CERT_FILE=" + filepath.Join(dir, "ca.pem")} |
| + } |
| +} |
| + |
| +// T7, the backhaul half, end to end (plan review finding 34): the real CLI |
| +// binary reaches the real console through two real edges, the second hop a |
| +// backhaul with its 30 s header bound, during a 40 s quiet restore. At the |
| +// CLI's own stderr and socket: DBF-11 within a second, no silence over 16 s, |
| +// and its rows after more than the hop's 30 s. The console's real |
| +// http.Server runs with ReadTimeout 2 s, as in |
| +// TestDBForwardLongRequestThroughTheEdge. |
| +// Mutations: drop the stream's flush (red: the wait lines sit in the |
| +// console's buffer and the CLI's socket is silent); stream nothing before |
| +// the box's first chunk (red: no headers inside the hop's 30 s). |
| +func TestDBForwardThroughTheBackhaulHop(t *testing.T) { |
| + var tap *dbfTap |
| + e := newDBFE2E(t, dbfE2EOptions{forward: true, serve: throughBackhaul(2*time.Second, &tap), worker: slowRestore(40 * time.Second)}) |
| + checkProgressAtTheCLI(t, e, tap, 40*time.Second) |
| } |
| ``` |
| |
| -- [ ] **Step 2: Run it** |
| +- [ ] **Step 3: Run it** |
| |
| -Run (sandbox off): `go test ./internal/edge/ -run 'TestDBForwardStreamSurvivesTheBackhaulHop' -count=1 -v` |
| -Expected: PASS after about 40 s. If the sender answers 404 instead of backhauling, the home map did not see annie's edges file: compare with `newBackhaulScene`'s order (mesh key, certificates, edges, `converge`) and follow it. |
| +Run (sandbox off; it builds the CLI and the edge test binary): `go test . -run 'TestDBForwardThroughTheBackhaulHop' -count=1 -v` |
| +Expected: PASS after about 45 s. If the helper never comes up, its stderr is in the log. If the CLI gets 404 from betty instead of a backhaul, betty's home map did not see annie's edges file: compare with `newBackhaulScene`'s order (mesh key, certificates, edges files, then each edge's mirror and home map) and follow it. If the CLI fails its TLS handshake, check that `SSL_CERT_FILE` reached it (`carlosTestEnv` appends the harness's extras last). |
| |
| -- [ ] **Step 3: Commit** |
| +- [ ] **Step 4: Commit** |
| |
| ```bash |
| -git add internal/edge/dbforward_backhaul_test.go |
| -git commit -m "edge: a forwarded read's stream survives the backhaul hop (T7)" |
| +git add internal/edge/dbforward_backhaul_helper_test.go dbforward_e2e_test.go |
| +git commit -m "db forward: T7 through the backhaul hop with the real CLI and console" |
| ``` |
| |
| --- |
| @@ -4315,13 +4986,18 @@ In a copy of the tree (`cp -a . "$TMPDIR/dbf2-mut"`), make each mutation, run it |
| | in `dbreads` `attempt`, open the replica at a prefix from anything but `rec.Host` (for example the first host the agent ever checked) | `TestDBForwardMinio` (the second tenant's rows), in the MinIO step; `TestDBReadsParityWithTheInRegionDoor` | |
| | `openDBForward` skips the byte count and SHA-256 comparison | `TestForwardedQueryRefusesAnAnswerThatDoesNotCheckOut`, `TestDBForwardPrintsNothingForABadAnswer` | |
| | `admitDBRead` releases the counts before returning, or `handleAPIDBRead` drops its `defer` | `TestDBReadAdmissionBounds` | |
| -| `handleAPIDBRead` drops `SetReadDeadline` | `TestDBForwardLongRequestThroughTheEdge` | |
| -| `dbLineWriter.write` drops `Flush` | `TestDBReadLinesAreFlushedAsWritten`, `TestDBForwardLongRequestThroughTheEdge` | |
| -| `streamDBRead` writes the result line without waiting for `next > res.LastSeq` | `TestDBReadResultWaitsForItsLastChunk` (the chunk-count check) | |
| +| `handleAPIDBRead` drops the post-admission `SetReadDeadline` (plan review finding 38) | `TestDBReadSetsBothReadDeadlines` (no deadline past 22 minutes); a finished body lets net/http clear the read deadline itself, so `TestDBForwardLongRequestThroughTheEdge` is not its witness | |
| +| `dbLineWriter.write` drops `Flush` | `TestDBReadLinesAreFlushedAsWritten`, `TestDBForwardLongRequestThroughTheEdge`, `TestDBForwardThroughTheBackhaulHop` | |
| +| `streamDBRead` writes the result line without waiting for `next > res.LastSeq` (plan review finding 31) | `TestDBReadResultWaitsForAChunkTheListingMissed` | |
| | `streamDBRead` lists chunks before reading `result.json`, the wait left in place (plan review finding 20) | `TestDBReadResultWaitsForItsLastChunk` (the operation-order check: a `list` not preceded by its round's `result`) | |
| | `admitDBRead` writes the audit row after the creates (plan review finding 11) | `TestDBReadEndedDuringAdmissionWritesOnlyTheCancel` ("the audit row is down before the request") | |
| -| `admitDBRead` calls `s.bumpSerial` instead of `announceDBRead`, or drops the `ctx.Err()` check after the queue create | `TestDBReadEndedDuringAdmissionWritesOnlyTheCancel` ("ended after the queue entry": the serial moves) | |
| -| `admitDBRead` drops `cancelDBRead` on the ended path | `TestDBReadEndedDuringAdmissionWritesOnlyTheCancel` (no cancel) | |
| +| `admitDBRead` drops any one of its four `ended()` checks (plan review finding 28) | `TestDBReadEndedDuringAdmissionWritesOnlyTheCancel`, the "ended before <step>" case for that step | |
| +| `admitDBRead` drops `cancelDBRead` on a published step | `TestDBReadEndedDuringAdmissionWritesOnlyTheCancel` ("ended before queue", "ended before serial": no cancel) | |
| +| `admitDBRead` calls `s.bumpSerial` instead of `announceDBRead` (plan review finding 33) | `TestDBReadEndedDuringAdmissionWritesOnlyTheCancel` ("ended during the serial read": the detached write lands) | |
| +| `admitDBRead` drops the U+FFFD refusal (plan review finding 36) | `TestDBReadRouteRefusesBeforeAdmitting` (the two raw bodies) | |
| +| `dbQueryForward` drops the UTF-8 check | `TestForwardedQueryRefusesSQLThatIsNotUTF8`, `TestDBForwardRefusesSQLThatIsNotUTF8` | |
| +| `dbfChunkBound.admit` drops its size floor, its empty-chunk rule or its count bound (plan review finding 37) | `TestForwardedQueryBoundsChunkCountAndSize`, the matching case | |
| +| `streamDBRead` drops the `ownCtx` check before the result line (plan review finding 35) | `TestDBReadResultAfterTheDeadlineIsNotRelayed` | |
| | `handleAPIDBRead` drops the `preHeader` read deadline (plan review finding 13) | `TestDBReadStalledBodyIsCutAtThePreHeaderBound` | |
| | `streamDBRead` calls the store on `r.Context()` without `dbfCall`, or drops either phase's deadline context (plan review finding 12) | `TestDBReadStalledStoreKeepsTheStreamAndItsBound` | |
| | `streamDBRead` relays without the owner's deadline (chunk Gets on `r.Context()`, no `ownCtx` check after them) | `TestDBReadSlowRelayEndsAtTheBound` | |
| @@ -4331,8 +5007,6 @@ In a copy of the tree (`cp -a . "$TMPDIR/dbf2-mut"`), make each mutation, run it |
| |
| Record the table with every row red in the branch description. |
| |
| -One check is not in the table on purpose: the `ownCtx.Err()` check just before the result line. `dbfCall` already cuts every store call once the owner's deadline passes, so that check guards only a line write that outlasts the deadline, which no in-process client can force. It stays as defence in depth, and the branch description says it has no witness of its own. |
| - |
| - [ ] **Step 2: Merge main and run the census** |
| |
| ```bash |
amadan