| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "bytes" |
| 6 | "context" |
| 7 | "crypto/ecdh" |
| 8 | "crypto/ed25519" |
| 9 | "crypto/rand" |
| 10 | "crypto/sha256" |
| 11 | "encoding/base64" |
| 12 | "encoding/hex" |
| 13 | "encoding/json" |
| 14 | "fmt" |
| 15 | "io" |
| 16 | "net" |
| 17 | "net/http" |
| 18 | "net/http/httptest" |
| 19 | "path/filepath" |
| 20 | "slices" |
| 21 | "strings" |
| 22 | "sync" |
| 23 | "testing" |
| 24 | "time" |
| 25 | |
| 26 | lsfile "github.com/benbjohnson/litestream/file" |
| 27 | |
| 28 | "github.com/carlosframework/platform/internal/dbdoor" |
| 29 | "github.com/carlosframework/platform/internal/dbdoor/dbdoortest" |
| 30 | "github.com/carlosframework/platform/internal/dbquery" |
| 31 | "github.com/carlosframework/platform/internal/dbreads" |
| 32 | "github.com/carlosframework/platform/internal/dbreads/dbreadstest" |
| 33 | "github.com/carlosframework/platform/internal/provision" |
| 34 | "github.com/carlosframework/platform/internal/release" |
| 35 | ) |
| 36 | |
| 37 | // The console forwards only with the switch on, a root key whose public |
| 38 | // half is the deployment's, and no host claims (spec §5.5, §9.1, §11.1). |
| 39 | func TestDBForwardSignerNeedsTheSwitchAndTheRootKey(t *testing.T) { |
| 40 | t.Parallel() |
| 41 | pub, priv, _ := ed25519.GenerateKey(rand.Reader) |
| 42 | _, other, _ := ed25519.GenerateKey(rand.Reader) |
| 43 | for name, c := range map[string]struct { |
| 44 | cfg Config |
| 45 | want bool |
| 46 | }{ |
| 47 | "on, root key": {Config{DBForward: true, Key: priv, ReleasePub: pub}, true}, |
| 48 | "off": {Config{Key: priv, ReleasePub: pub}, false}, |
| 49 | "no key": {Config{DBForward: true, ReleasePub: pub}, false}, |
| 50 | "not the root's key": {Config{DBForward: true, Key: other, ReleasePub: pub}, false}, |
| 51 | "host claims": {Config{DBForward: true, Key: priv, ReleasePub: pub, HostClaim: true}, false}, |
| 52 | } { |
| 53 | s := &Server{cfg: c.cfg} |
| 54 | if _, ok := s.dbForwardSigner(); ok != c.want { |
| 55 | t.Errorf("%s: %v", name, ok) |
| 56 | } |
| 57 | } |
| 58 | } |
| 59 | |
| 60 | func TestNewHTTPServerKeepsTheConsolesTimeouts(t *testing.T) { |
| 61 | t.Parallel() |
| 62 | srv := NewHTTPServer(nil) |
| 63 | if srv.ReadHeaderTimeout != 10*time.Second || srv.ReadTimeout != 5*time.Minute || srv.WriteTimeout != 0 { |
| 64 | t.Fatalf("%+v", srv) |
| 65 | } |
| 66 | } |
| 67 | |
| 68 | // putFixtureObject creates key, or replaces it under its current etag: a |
| 69 | // fixture may republish a box's files, as a box does every pass. Protocol |
| 70 | // writes stay create-only; this helper is for fixtures only (plan review |
| 71 | // finding 6: DirStore.Put refuses a create over an existing key). |
| 72 | func putFixtureObject(t *testing.T, s release.Store, key string, body []byte) { |
| 73 | t.Helper() |
| 74 | etag := "" |
| 75 | if _, cur, err := s.Get(context.Background(), key); err == nil { |
| 76 | etag = cur |
| 77 | } |
| 78 | if _, err := s.Put(context.Background(), key, body, etag); err != nil { |
| 79 | t.Fatal(err) |
| 80 | } |
| 81 | } |
| 82 | |
| 83 | // putEdgesFile writes control/edges/<box>.json as a box publishes it. |
| 84 | func putEdgesFile(t *testing.T, s release.Store, box, placement string, caps []string, draining bool) { |
| 85 | t.Helper() |
| 86 | body, _ := json.Marshal(map[string]any{"v": 1, "box": box, "eip": "203.0.113.10", "placement": placement, |
| 87 | "capabilities": caps, "draining": draining, "hosts": map[string]any{}}) |
| 88 | putFixtureObject(t, s, "control/edges/"+box+".json", body) |
| 89 | } |
| 90 | |
| 91 | // putBoxStatus writes status/boxes/<box>.json generated at at. |
| 92 | func putBoxStatus(t *testing.T, s release.Store, box string, at time.Time) { |
| 93 | t.Helper() |
| 94 | body, _ := json.Marshal(map[string]any{"generated_at": at.UTC(), "label": box}) |
| 95 | putFixtureObject(t, s, "status/boxes/"+box+".json", body) |
| 96 | } |
| 97 | |
| 98 | // T14 (codex 10): only complete evidence, both flags, refuses up front. |
| 99 | // Incomplete evidence of any kind admits the read and lets the pickup |
| 100 | // window decide. Mutation: check Complete and not BoxesComplete (red: the |
| 101 | // missing, unmatched and unreadable status cases refuse). |
| 102 | func TestDBForwardEvidenceRefusesOnlyOnCompleteEvidence(t *testing.T) { |
| 103 | t.Parallel() |
| 104 | now := time.Now() |
| 105 | capable := []string{dbreads.Capability} |
| 106 | for name, c := range map[string]struct { |
| 107 | setup func(t *testing.T, s release.Store) |
| 108 | want string |
| 109 | }{ |
| 110 | "a capable live box": {func(t *testing.T, s release.Store) { |
| 111 | putEdgesFile(t, s, "uk-1", "uk", capable, false) |
| 112 | putBoxStatus(t, s, "uk-1", now) |
| 113 | }, ""}, |
| 114 | "no capable box": {func(t *testing.T, s release.Store) { |
| 115 | putEdgesFile(t, s, "uk-1", "uk", nil, false) |
| 116 | putBoxStatus(t, s, "uk-1", now) |
| 117 | }, dbForwardNone}, |
| 118 | "a capable box that is draining": {func(t *testing.T, s release.Store) { |
| 119 | putEdgesFile(t, s, "uk-1", "uk", capable, true) |
| 120 | putBoxStatus(t, s, "uk-1", now) |
| 121 | }, dbForwardNone}, |
| 122 | "capable but stale": {func(t *testing.T, s release.Store) { |
| 123 | putEdgesFile(t, s, "uk-1", "uk", capable, false) |
| 124 | putBoxStatus(t, s, "uk-1", now.Add(-time.Hour)) |
| 125 | }, dbForwardDead}, |
| 126 | "capable in another region only": {func(t *testing.T, s release.Store) { |
| 127 | putEdgesFile(t, s, "ie-1", "ie", capable, false) |
| 128 | putBoxStatus(t, s, "ie-1", now) |
| 129 | }, dbForwardNone}, |
| 130 | "a status snapshot missing": {func(t *testing.T, s release.Store) { |
| 131 | putEdgesFile(t, s, "uk-1", "uk", nil, false) |
| 132 | }, ""}, |
| 133 | "an edges file unreadable": {func(t *testing.T, s release.Store) { |
| 134 | s.Put(context.Background(), "control/edges/uk-1.json", []byte("{not json"), "") |
| 135 | putBoxStatus(t, s, "uk-1", now) |
| 136 | }, ""}, |
| 137 | "a fresh snapshot with no edges file": {func(t *testing.T, s release.Store) { |
| 138 | putEdgesFile(t, s, "uk-1", "uk", nil, false) |
| 139 | putBoxStatus(t, s, "uk-1", now) |
| 140 | putBoxStatus(t, s, "uk-2", now) |
| 141 | }, ""}, |
| 142 | "status snapshots unreadable": {func(t *testing.T, s release.Store) { |
| 143 | putEdgesFile(t, s, "uk-1", "uk", nil, false) |
| 144 | s.Put(context.Background(), "status/boxes/uk-1.json", []byte("{not json"), "") |
| 145 | }, ""}, |
| 146 | } { |
| 147 | t.Run(name, func(t *testing.T) { |
| 148 | t.Parallel() |
| 149 | s := release.NewDirStore(t.TempDir()) |
| 150 | c.setup(t, s) |
| 151 | snap, err := readLiveEdgesSnapshot(context.Background(), s, now) |
| 152 | if got := dbForwardEvidence(snap, err, "uk"); got != c.want { |
| 153 | t.Fatalf("evidence %q, want %q (complete %v, boxes %v)", got, c.want, snap.Complete, snap.BoxesComplete) |
| 154 | } |
| 155 | }) |
| 156 | } |
| 157 | } |
| 158 | |
| 159 | const dbfHost = "shop.example.com" |
| 160 | |
| 161 | // dbfFixture is a console in ie with forwarding on or off, a member, and |
| 162 | // shop.example.com's record in uk with a capable, live uk box published. |
| 163 | type dbfFixture struct { |
| 164 | *consoleFixture |
| 165 | member string |
| 166 | } |
| 167 | |
| 168 | func newDBFFixture(t *testing.T, forward bool) *dbfFixture { |
| 169 | t.Helper() |
| 170 | fx := newConsoleFixture(t) |
| 171 | fx.srv.cfg.DBForward = forward |
| 172 | fx.srv.cfg.Region = "ie" |
| 173 | fx.srv.cfg.HibernateBucket = "data-ie" |
| 174 | if _, err := fx.st.CreateAppIn(OpsAccountID, "shop", "paul@keymail.dev", time.Now()); err != nil { |
| 175 | t.Fatal(err) |
| 176 | } |
| 177 | if err := fx.st.AddMember(OpsAccountID, "member@example.com", RoleMember, time.Now()); err != nil { |
| 178 | t.Fatal(err) |
| 179 | } |
| 180 | member := cliToken(t, fx, "member@example.com", "code-dbf") |
| 181 | putShopRecord(t, fx, provision.Record{Host: dbfHost, Region: "uk", CreatedAt: time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)}) |
| 182 | putEdgesFile(t, fx.rel, "uk-1", "uk", []string{dbreads.Capability}, false) |
| 183 | putBoxStatus(t, fx.rel, "uk-1", time.Now()) |
| 184 | return &dbfFixture{consoleFixture: fx, member: member} |
| 185 | } |
| 186 | |
| 187 | func (fx *dbfFixture) listURL() string { |
| 188 | return fx.ts.URL + "/api/cli/apps/" + release.OpsAccount + "/shop/db/" + dbfHost + "/ltx/9" |
| 189 | } |
| 190 | |
| 191 | // T8, the console's half: forwarding off is main's answer byte for byte. |
| 192 | // The want is main's dbResidencyRefusal for this fixture, confirmed against |
| 193 | // main's code before this branch changed dbGate. |
| 194 | func TestListDoorWithForwardingOffIsMains(t *testing.T) { |
| 195 | fx := newDBFFixture(t, false) |
| 196 | status, _, body := rawReq(t, http.MethodGet, fx.listURL(), fx.member, "") |
| 197 | want := `{"error":"shop.example.com keeps its data in region uk. This console can only read data kept in region ie."}` |
| 198 | if status != http.StatusConflict || strings.TrimSpace(body) != want { |
| 199 | t.Fatalf("%d %s", status, body) |
| 200 | } |
| 201 | } |
| 202 | |
| 203 | func TestListDoorMarksACapableRegion(t *testing.T) { |
| 204 | fx := newDBFFixture(t, true) |
| 205 | status, _, body := rawReq(t, http.MethodGet, fx.listURL(), fx.member, "") |
| 206 | var got dbdoor.ForwardBody |
| 207 | json.Unmarshal([]byte(body), &got) |
| 208 | if status != http.StatusConflict || got.Forward != 1 || got.Region != "uk" || |
| 209 | !slices.Equal(got.ForwardOps, []string{dbreads.OpQuery, dbreads.OpPull}) || |
| 210 | got.Error != fmt.Sprintf(dbfUpdateCarlos, dbfHost, "uk") { |
| 211 | t.Fatalf("%d %s", status, body) |
| 212 | } |
| 213 | } |
| 214 | |
| 215 | func TestListDoorRefusesOnCompleteEvidence(t *testing.T) { |
| 216 | fx := newDBFFixture(t, true) |
| 217 | putEdgesFile(t, fx.rel, "uk-1", "uk", nil, false) |
| 218 | status, _, body := rawReq(t, http.MethodGet, fx.listURL(), fx.member, "") |
| 219 | if status != http.StatusConflict || !strings.Contains(body, fmt.Sprintf(dbfNoCapable, dbfHost, "uk")) || strings.Contains(body, `"forward"`) { |
| 220 | t.Fatalf("no capable box: %d %s", status, body) |
| 221 | } |
| 222 | putEdgesFile(t, fx.rel, "uk-2", "uk", []string{dbreads.Capability}, false) |
| 223 | putBoxStatus(t, fx.rel, "uk-2", time.Now().Add(-time.Hour)) |
| 224 | status, _, body = rawReq(t, http.MethodGet, fx.listURL(), fx.member, "") |
| 225 | if status != http.StatusConflict || !strings.Contains(body, fmt.Sprintf(dbfNoneLive, dbfHost, "uk")) { |
| 226 | t.Fatalf("none live: %d %s", status, body) |
| 227 | } |
| 228 | } |
| 229 | |
| 230 | // Review Focus 5: a home-region record seen from a regional console keeps |
| 231 | // main's sentence; there is no region to forward to by name. |
| 232 | func TestListDoorKeepsMainsSentenceForTheHomeRegion(t *testing.T) { |
| 233 | fx := newDBFFixture(t, true) |
| 234 | putShopRecord(t, fx.consoleFixture, provision.Record{Host: "home.example.com", CreatedAt: time.Now().UTC()}) |
| 235 | u := fx.ts.URL + "/api/cli/apps/" + release.OpsAccount + "/shop/db/home.example.com/ltx/9" |
| 236 | status, _, body := rawReq(t, http.MethodGet, u, fx.member, "") |
| 237 | if status != http.StatusConflict || strings.Contains(body, `"forward"`) || !strings.Contains(body, "This console can only read data kept in") { |
| 238 | t.Fatalf("%d %s", status, body) |
| 239 | } |
| 240 | } |
| 241 | |
| 242 | func (fx *dbfFixture) readURL(host string) string { |
| 243 | return fx.ts.URL + "/api/cli/apps/" + release.OpsAccount + "/shop/db/" + host + "/reads" |
| 244 | } |
| 245 | |
| 246 | func queryBody(t *testing.T, sqlText string) (dbreads.ReadBody, *ecdh.PrivateKey) { |
| 247 | t.Helper() |
| 248 | key, err := dbreads.NewReplyKey() |
| 249 | if err != nil { |
| 250 | t.Fatal(err) |
| 251 | } |
| 252 | return dbreads.ReadBody{Op: dbreads.OpQuery, SQL: sqlText, ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())}, key |
| 253 | } |
| 254 | |
| 255 | // openRead posts a read and returns the open response; the caller reads its |
| 256 | // lines or closes it. |
| 257 | func (fx *dbfFixture) openRead(t *testing.T, host, bearer string, body dbreads.ReadBody) *http.Response { |
| 258 | t.Helper() |
| 259 | raw, _ := json.Marshal(body) |
| 260 | req, _ := http.NewRequest(http.MethodPost, fx.readURL(host), bytes.NewReader(raw)) |
| 261 | req.Header.Set("Content-Type", "application/json") |
| 262 | req.Header.Set("Authorization", "Bearer "+bearer) |
| 263 | res, err := http.DefaultClient.Do(req) |
| 264 | if err != nil { |
| 265 | t.Fatal(err) |
| 266 | } |
| 267 | return res |
| 268 | } |
| 269 | |
| 270 | func readLines(t *testing.T, r io.Reader) []dbreads.Line { |
| 271 | t.Helper() |
| 272 | sc := bufio.NewScanner(r) |
| 273 | sc.Buffer(make([]byte, 64<<10), dbreads.MaxLineBytes) |
| 274 | var out []dbreads.Line |
| 275 | for sc.Scan() { |
| 276 | var l dbreads.Line |
| 277 | if err := json.Unmarshal(sc.Bytes(), &l); err != nil { |
| 278 | t.Fatalf("not a line: %s", sc.Bytes()) |
| 279 | } |
| 280 | out = append(out, l) |
| 281 | } |
| 282 | return out |
| 283 | } |
| 284 | |
| 285 | func refusal(t *testing.T, res *http.Response) (int, string) { |
| 286 | t.Helper() |
| 287 | defer res.Body.Close() |
| 288 | b, _ := io.ReadAll(res.Body) |
| 289 | return res.StatusCode, string(b) |
| 290 | } |
| 291 | |
| 292 | func TestDBReadRouteRefusesBeforeAdmitting(t *testing.T) { |
| 293 | fx := newDBFFixture(t, true) |
| 294 | good, _ := queryBody(t, "SELECT 1") |
| 295 | putShopRecord(t, fx.consoleFixture, provision.Record{Host: "here.example.com", Region: "ie", CreatedAt: time.Now().UTC()}) |
| 296 | putShopRecord(t, fx.consoleFixture, provision.Record{Host: "moving.example.com", Region: "uk", CreatedAt: time.Now().UTC(), |
| 297 | Move: &provision.MoveState{ID: "mv_x", To: "ie", State: "requested"}}) |
| 298 | two := good |
| 299 | two.SQL = "SELECT 1; SELECT 2" |
| 300 | shell := good |
| 301 | shell.Op = "shell" |
| 302 | pullSQL := good |
| 303 | pullSQL.Op = dbreads.OpPull |
| 304 | badKey := good |
| 305 | badKey.ReplyKey = "AAAA" |
| 306 | future := good |
| 307 | future.At = time.Now().Add(time.Hour).UTC().Format(time.RFC3339) |
| 308 | for name, c := range map[string]struct { |
| 309 | host string |
| 310 | body dbreads.ReadBody |
| 311 | status int |
| 312 | want string |
| 313 | }{ |
| 314 | "an in-region instance": {"here.example.com", good, http.StatusNotFound, ""}, |
| 315 | "no such instance": {"nope.example.com", good, http.StatusNotFound, ""}, |
| 316 | "moving": {"moving.example.com", good, http.StatusConflict, fmt.Sprintf(dbfMoving, "moving.example.com")}, |
| 317 | "two statements": {dbfHost, two, http.StatusBadRequest, dbquery.ErrOneStatement.Error()}, |
| 318 | "an op this console does not forward": {dbfHost, shell, http.StatusBadRequest, ""}, |
| 319 | "a pull carrying SQL": {dbfHost, pullSQL, http.StatusBadRequest, ""}, |
| 320 | "a reply key that is not a key": {dbfHost, badKey, http.StatusBadRequest, ""}, |
| 321 | "a time in the future": {dbfHost, future, http.StatusBadRequest, ""}, |
| 322 | } { |
| 323 | status, body := refusal(t, fx.openRead(t, c.host, fx.member, c.body)) |
| 324 | if status != c.status || !strings.Contains(body, c.want) { |
| 325 | t.Errorf("%s: %d %s", name, status, body) |
| 326 | } |
| 327 | } |
| 328 | // SQL that JSON decoding would have rewritten (plan review finding 36): |
| 329 | // a byte that is not UTF-8, and a lone surrogate escape. |
| 330 | for name, sqlJSON := range map[string]string{ |
| 331 | "a byte that is not UTF-8": "\"SELECT '\xff'\"", |
| 332 | "a lone surrogate": `"SELECT '\ud800'"`, |
| 333 | } { |
| 334 | raw := `{"op":"query","sql":` + sqlJSON + `,"reply_key":"` + good.ReplyKey + `"}` |
| 335 | req, _ := http.NewRequest(http.MethodPost, fx.readURL(dbfHost), strings.NewReader(raw)) |
| 336 | req.Header.Set("Content-Type", "application/json") |
| 337 | req.Header.Set("Authorization", "Bearer "+fx.member) |
| 338 | res, err := http.DefaultClient.Do(req) |
| 339 | if err != nil { |
| 340 | t.Fatal(err) |
| 341 | } |
| 342 | if status, body := refusal(t, res); status != http.StatusBadRequest { |
| 343 | t.Errorf("%s: %d %s", name, status, body) |
| 344 | } |
| 345 | } |
| 346 | if keys, _ := fx.rel.List(context.Background(), dbreads.Prefix); len(keys) != 0 { |
| 347 | t.Fatalf("a refused read wrote %v", keys) |
| 348 | } |
| 349 | fx.srv.cfg.DBForward = false |
| 350 | status, body := refusal(t, fx.openRead(t, dbfHost, fx.member, good)) |
| 351 | if status != http.StatusConflict || !strings.Contains(body, "This console can only read data kept in region ie.") { |
| 352 | t.Fatalf("forwarding off: %d %s", status, body) |
| 353 | } |
| 354 | fx.srv.cfg.DBForward, fx.srv.cfg.HostClaim = true, true |
| 355 | status, body = refusal(t, fx.openRead(t, dbfHost, fx.member, good)) |
| 356 | if status != http.StatusConflict || !strings.Contains(body, "This console can only read data kept in") { |
| 357 | t.Fatalf("a host-claim console: %d %s", status, body) |
| 358 | } |
| 359 | } |
| 360 | |
| 361 | func TestDBReadRouteRefusesOnCompleteEvidence(t *testing.T) { |
| 362 | fx := newDBFFixture(t, true) |
| 363 | putEdgesFile(t, fx.rel, "uk-1", "uk", nil, false) |
| 364 | good, _ := queryBody(t, "SELECT 1") |
| 365 | status, body := refusal(t, fx.openRead(t, dbfHost, fx.member, good)) |
| 366 | if status != http.StatusConflict || !strings.Contains(body, fmt.Sprintf(dbfNoCapable, dbfHost, "uk")) { |
| 367 | t.Fatalf("%d %s", status, body) |
| 368 | } |
| 369 | } |
| 370 | |
| 371 | // Admission, then the request, the queue entry, the serial and the audit |
| 372 | // row, all before the 200. |
| 373 | func TestDBReadQueuesASignedRequestAndAudits(t *testing.T) { |
| 374 | fx := newDBFFixture(t, true) |
| 375 | fx.srv.dbf.pickup = 200 * time.Millisecond |
| 376 | body, _ := queryBody(t, "SELECT count(*) FROM orders") |
| 377 | res := fx.openRead(t, dbfHost, fx.member, body) |
| 378 | lines := readLines(t, res.Body) |
| 379 | res.Body.Close() |
| 380 | if res.StatusCode != http.StatusOK || len(lines) == 0 || lines[0].Admitted == "" { |
| 381 | t.Fatalf("%d %+v", res.StatusCode, lines) |
| 382 | } |
| 383 | id := lines[0].Admitted |
| 384 | raw, _, err := fx.rel.Get(context.Background(), dbreads.RequestKey(id)) |
| 385 | if err != nil { |
| 386 | t.Fatal(err) |
| 387 | } |
| 388 | req, err := dbreads.Verify(fx.pub, raw, id, "uk", time.Now()) |
| 389 | if err != nil || req.SQL != body.SQL || req.Host != dbfHost || req.By != "member@example.com" || req.Account != release.OpsAccount { |
| 390 | t.Fatalf("request %+v: %v", req, err) |
| 391 | } |
| 392 | issued, _ := req.Issued() |
| 393 | if _, _, err := fx.rel.Get(context.Background(), dbreads.QueueKey("uk", issued, id)); err != nil { |
| 394 | t.Fatalf("no queue entry: %v", err) |
| 395 | } |
| 396 | if _, _, err := fx.rel.Get(context.Background(), release.SerialKey); err != nil { |
| 397 | t.Fatalf("the serial was not bumped: %v", err) |
| 398 | } |
| 399 | evs, _ := fx.st.Events(OpsAccountID, "shop", 10) |
| 400 | want := dbForwardAuditDetail(dbreads.OpQuery, "uk", body.SQL, time.Time{}) |
| 401 | found := false |
| 402 | for _, e := range evs { |
| 403 | found = found || (e.Action == "db-read" && e.Target == dbfHost && e.Detail == want) |
| 404 | } |
| 405 | if !found || strings.Contains(want, "orders") { |
| 406 | t.Fatalf("audit %q in %+v", want, evs) |
| 407 | } |
| 408 | } |
| 409 | |
| 410 | // dbfPutHook runs hook after each create of a key with suffix, once the |
| 411 | // create has landed. |
| 412 | type dbfPutHook struct { |
| 413 | release.Store |
| 414 | suffix string |
| 415 | hook func() |
| 416 | } |
| 417 | |
| 418 | func (h *dbfPutHook) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| 419 | etag, err := h.Store.Put(ctx, key, body, ifMatch) |
| 420 | if strings.HasSuffix(key, h.suffix) || (h.suffix == "queue" && strings.HasPrefix(key, dbreads.Prefix+"queue/")) { |
| 421 | h.hook() |
| 422 | } |
| 423 | return etag, err |
| 424 | } |
| 425 | |
| 426 | // dbReadAudited is safe from a handler's goroutine: it never calls t.Fatal. |
| 427 | func dbReadAudited(t *testing.T, st *Store) bool { |
| 428 | t.Helper() |
| 429 | evs, err := st.Events(OpsAccountID, "shop", 10) |
| 430 | if err != nil { |
| 431 | t.Error(err) |
| 432 | return false |
| 433 | } |
| 434 | for _, e := range evs { |
| 435 | if e.Action == "db-read" && e.Target == dbfHost { |
| 436 | return true |
| 437 | } |
| 438 | } |
| 439 | return false |
| 440 | } |
| 441 | |
| 442 | // dbfServe drives the handler in process with ctx as the request's own |
| 443 | // context, so a test can end the request at an exact step. |
| 444 | func (fx *dbfFixture) dbfServe(t *testing.T, ctx context.Context, sqlText string) *httptest.ResponseRecorder { |
| 445 | t.Helper() |
| 446 | body, _ := queryBody(t, sqlText) |
| 447 | raw, _ := json.Marshal(body) |
| 448 | req := httptest.NewRequest(http.MethodPost, fx.readURL(dbfHost), bytes.NewReader(raw)).WithContext(ctx) |
| 449 | req.Header.Set("Content-Type", "application/json") |
| 450 | req.Header.Set("Authorization", "Bearer "+fx.member) |
| 451 | rec := httptest.NewRecorder() |
| 452 | fx.srv.ServeHTTP(rec, req) |
| 453 | return rec |
| 454 | } |
| 455 | |
| 456 | // dbfBucket is what a read left under db-reads/: request objects, queue |
| 457 | // entries and cancels, by count. |
| 458 | func dbfBucket(t *testing.T, s release.Store) (requests, queued, cancels int) { |
| 459 | t.Helper() |
| 460 | keys, _ := s.List(context.Background(), dbreads.Prefix) |
| 461 | for _, k := range keys { |
| 462 | switch { |
| 463 | case strings.HasPrefix(k, dbreads.Prefix+"queue/"): |
| 464 | queued++ |
| 465 | case strings.HasSuffix(k, "/request.json"): |
| 466 | requests++ |
| 467 | case strings.HasSuffix(k, "/cancel"): |
| 468 | cancels++ |
| 469 | } |
| 470 | } |
| 471 | return |
| 472 | } |
| 473 | |
| 474 | // Plan review findings 11, 28 and 33 (J1, spec §5.5 step 3 and §9.3 rev 6): |
| 475 | // the request's context is checked immediately before every write of |
| 476 | // admission. Ended before the audit row: nothing at all. Before |
| 477 | // request.json: the row only. Before the queue entry: the row, request.json |
| 478 | // and the cancel. Before the serial: the row, both objects and the cancel. |
| 479 | // Never a serial bump, never a 200. And the bump itself runs on the |
| 480 | // request's context, so a request that ends while it reads the serial does |
| 481 | // not write it. |
| 482 | // Mutations, each red in its own case: drop any one of the four ended() |
| 483 | // checks; drop the cancel on a published step; write the audit row after |
| 484 | // the creates ("the audit row is down before the request"); call |
| 485 | // s.bumpSerial instead of announceDBRead ("ended during the serial read": |
| 486 | // the detached write lands). |
| 487 | func TestDBReadEndedDuringAdmissionWritesOnlyTheCancel(t *testing.T) { |
| 488 | t.Run("the audit row is down before the request", func(t *testing.T) { |
| 489 | fx := newDBFFixture(t, true) |
| 490 | auditedFirst := false |
| 491 | fx.srv.cfg.Releases = &dbfPutHook{Store: fx.rel, suffix: "/request.json", hook: func() { |
| 492 | auditedFirst = dbReadAudited(t, fx.st) |
| 493 | }} |
| 494 | fx.srv.dbf.pickup = 100 * time.Millisecond |
| 495 | body, _ := queryBody(t, "SELECT 1") |
| 496 | res := fx.openRead(t, dbfHost, fx.member, body) |
| 497 | io.Copy(io.Discard, res.Body) |
| 498 | res.Body.Close() |
| 499 | if !auditedFirst { |
| 500 | t.Fatal("request.json was written before the audit row") |
| 501 | } |
| 502 | }) |
| 503 | for _, c := range []struct { |
| 504 | step string |
| 505 | audited bool |
| 506 | requests, queued, cancels int |
| 507 | }{ |
| 508 | {"audit", false, 0, 0, 0}, |
| 509 | {"request", true, 0, 0, 0}, |
| 510 | {"queue", true, 1, 0, 1}, |
| 511 | {"serial", true, 1, 1, 1}, |
| 512 | } { |
| 513 | t.Run("ended before "+c.step, func(t *testing.T) { |
| 514 | fx := newDBFFixture(t, true) |
| 515 | _, serialBefore, _ := fx.rel.Get(context.Background(), release.SerialKey) |
| 516 | ctx, cancel := context.WithCancel(context.Background()) |
| 517 | defer cancel() |
| 518 | fx.srv.dbf.beforeWrite = func(step string) { |
| 519 | if step == c.step { |
| 520 | cancel() |
| 521 | } |
| 522 | } |
| 523 | rec := fx.dbfServe(t, ctx, "SELECT 1") |
| 524 | // The ended path writes no response (nobody is there to read |
| 525 | // it), and a recorder never written reports 200: read the body. |
| 526 | if strings.Contains(rec.Body.String(), "admitted") { |
| 527 | t.Fatalf("an ended request was admitted: %s", rec.Body) |
| 528 | } |
| 529 | if _, serialAfter, _ := fx.rel.Get(context.Background(), release.SerialKey); serialAfter != serialBefore { |
| 530 | t.Fatal("the serial was bumped after the request ended") |
| 531 | } |
| 532 | requests, queued, cancels := dbfBucket(t, fx.rel) |
| 533 | if got := dbReadAudited(t, fx.st); got != c.audited || requests != c.requests || queued != c.queued || cancels != c.cancels { |
| 534 | t.Fatalf("audited %v, requests %d, queue entries %d, cancels %d", got, requests, queued, cancels) |
| 535 | } |
| 536 | }) |
| 537 | } |
| 538 | // Over the real DirStore, whose Put ignores its context (Astra round 1, |
| 539 | // P2): only the read of the serial is hooked, so nothing but the bump's |
| 540 | // own check can stop the write. The request is then re-checked after the |
| 541 | // bump: the read was published, so it gets its cancel and no 200. |
| 542 | t.Run("ended during the serial read", func(t *testing.T) { |
| 543 | fx := newDBFFixture(t, true) |
| 544 | _, serialBefore, _ := fx.rel.Get(context.Background(), release.SerialKey) |
| 545 | ctx, cancel := context.WithCancel(context.Background()) |
| 546 | defer cancel() |
| 547 | fx.srv.cfg.Releases = &dbfGetHook{Store: fx.rel, key: release.SerialKey, onGet: cancel} |
| 548 | rec := fx.dbfServe(t, ctx, "SELECT 1") |
| 549 | if _, serialAfter, _ := fx.rel.Get(context.Background(), release.SerialKey); serialAfter != serialBefore { |
| 550 | t.Fatal("the serial was written after the request ended during the bump") |
| 551 | } |
| 552 | if strings.Contains(rec.Body.String(), "admitted") { |
| 553 | t.Fatalf("a request that ended during the bump was admitted: %s", rec.Body) |
| 554 | } |
| 555 | if requests, queued, cancels := dbfBucket(t, fx.rel); requests != 1 || queued != 1 || cancels != 1 { |
| 556 | t.Fatalf("requests %d, queue entries %d, cancels %d", requests, queued, cancels) |
| 557 | } |
| 558 | }) |
| 559 | } |
| 560 | |
| 561 | // dbfGetHook runs onGet once on a read of key and is otherwise the store |
| 562 | // beneath it, its context handling included: a DirStore's Put ignores its |
| 563 | // context, as CARLOS_DEPLOYMENT_DIR's store does in production. |
| 564 | type dbfGetHook struct { |
| 565 | release.Store |
| 566 | key string |
| 567 | onGet func() |
| 568 | once sync.Once |
| 569 | } |
| 570 | |
| 571 | func (h *dbfGetHook) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 572 | if key == h.key { |
| 573 | h.once.Do(h.onGet) |
| 574 | } |
| 575 | return h.Store.Get(ctx, key) |
| 576 | } |
| 577 | |
| 578 | // Plan review finding 13: a body delivered slowly is cut at s.dbf.preHeader, |
| 579 | // well inside backhaul's 30 s header wait, and admits nothing. |
| 580 | // Mutation: drop the preHeader read deadline, keeping only the bound's (red: |
| 581 | // the connection is still open at 3 s). |
| 582 | func TestDBReadStalledBodyIsCutAtThePreHeaderBound(t *testing.T) { |
| 583 | fx := newDBFFixture(t, true) |
| 584 | fx.srv.dbf.preHeader = 300 * time.Millisecond |
| 585 | conn, err := net.Dial("tcp", strings.TrimPrefix(fx.ts.URL, "http://")) |
| 586 | if err != nil { |
| 587 | t.Fatal(err) |
| 588 | } |
| 589 | defer conn.Close() |
| 590 | path := strings.TrimPrefix(fx.readURL(dbfHost), fx.ts.URL) |
| 591 | fmt.Fprintf(conn, "POST %s HTTP/1.1\r\nHost: console\r\nAuthorization: Bearer %s\r\n"+ |
| 592 | "Content-Type: application/json\r\nContent-Length: 4096\r\n\r\n{\"op\":", path, fx.member) |
| 593 | conn.SetReadDeadline(time.Now().Add(3 * time.Second)) |
| 594 | start := time.Now() |
| 595 | _, err = io.ReadAll(conn) // a refusal, then the close; or the close alone |
| 596 | if ne, ok := err.(net.Error); ok && ne.Timeout() { |
| 597 | t.Fatalf("the stalled body still held the request after %s", time.Since(start)) |
| 598 | } |
| 599 | if keys, _ := fx.rel.List(context.Background(), dbreads.Prefix); len(keys) != 0 { |
| 600 | t.Fatalf("a stalled body wrote %v", keys) |
| 601 | } |
| 602 | } |
| 603 | |
| 604 | // startBox runs plan 1's real agent and worker code in process for uk, |
| 605 | // passing every 20 ms: a stand-in for the box's serial poll, not acceptance. |
| 606 | func (fx *dbfFixture) startBox(t *testing.T, replicaDir string) *dbreadstest.Box { |
| 607 | t.Helper() |
| 608 | box := dbreadstest.Start(t, dbreadstest.Options{Store: fx.rel, Region: "uk", Box: "uk-1", Pub: fx.pub, HibernatePrefix: "instances", |
| 609 | OpenReplica: func(context.Context, string) (dbdoor.ReplicaReader, error) { |
| 610 | return lsfile.NewReplicaClient(replicaDir), nil |
| 611 | }}) |
| 612 | runBox(t, box) |
| 613 | return box |
| 614 | } |
| 615 | |
| 616 | func runBox(t *testing.T, box *dbreadstest.Box) { |
| 617 | ctx, cancel := context.WithCancel(context.Background()) |
| 618 | done := make(chan struct{}) |
| 619 | go func() { |
| 620 | defer close(done) |
| 621 | for ctx.Err() == nil { |
| 622 | box.Agent.Pass(ctx) |
| 623 | time.Sleep(20 * time.Millisecond) |
| 624 | } |
| 625 | }() |
| 626 | t.Cleanup(func() { cancel(); <-done; box.Agent.Wait() }) |
| 627 | } |
| 628 | |
| 629 | // openAnswer opens a done result's chunks and summary with the CLI's key. |
| 630 | func openAnswer(t *testing.T, key *ecdh.PrivateKey, lines []dbreads.Line) ([]byte, dbreads.Summary) { |
| 631 | t.Helper() |
| 632 | id := lines[0].Admitted |
| 633 | res := lines[len(lines)-1].Result |
| 634 | if res == nil || res.State != dbreads.StateDone { |
| 635 | t.Fatalf("no done result: %+v", lines[len(lines)-1]) |
| 636 | } |
| 637 | var rows bytes.Buffer |
| 638 | for _, l := range lines { |
| 639 | if l.Chunk == 0 { |
| 640 | continue |
| 641 | } |
| 642 | plain, err := dbreads.OpenChunk(key, id, l.Chunk, dbreads.StreamRows, l.Chunk == res.LastSeq, l.Data) |
| 643 | if err != nil { |
| 644 | t.Fatalf("chunk %d: %v", l.Chunk, err) |
| 645 | } |
| 646 | rows.Write(plain) |
| 647 | } |
| 648 | s, err := dbreads.OpenSummary(key, id, res.Sealed) |
| 649 | if err != nil { |
| 650 | t.Fatal(err) |
| 651 | } |
| 652 | return rows.Bytes(), s |
| 653 | } |
| 654 | |
| 655 | // A pull through the console: the route admits it, relays the box's ltx |
| 656 | // stream and its summary's file list, and audits it with DBF-19; with --at, |
| 657 | // DBF-20. |
| 658 | func TestDBReadPullStreamsTheFiles(t *testing.T) { |
| 659 | fx := newDBFFixture(t, true) |
| 660 | dir := t.TempDir() |
| 661 | dbdoortest.Replicate(t, dir, 3) |
| 662 | fx.startBox(t, dir) |
| 663 | key, _ := dbreads.NewReplyKey() |
| 664 | res := fx.openRead(t, dbfHost, fx.member, dbreads.ReadBody{Op: dbreads.OpPull, ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())}) |
| 665 | lines := readLines(t, res.Body) |
| 666 | res.Body.Close() |
| 667 | last := lines[len(lines)-1].Result |
| 668 | if last == nil || last.State != dbreads.StateDone { |
| 669 | t.Fatalf("lines %+v", lines) |
| 670 | } |
| 671 | s, err := dbreads.OpenSummary(key, lines[0].Admitted, last.Sealed) |
| 672 | if err != nil || len(s.Files) == 0 || s.TXID != s.Files[len(s.Files)-1].MaxTXID { |
| 673 | t.Fatalf("summary %+v %v", s, err) |
| 674 | } |
| 675 | h := sha256.New() |
| 676 | for _, l := range lines { |
| 677 | if l.Chunk == 0 { |
| 678 | continue |
| 679 | } |
| 680 | plain, err := dbreads.OpenChunk(key, lines[0].Admitted, l.Chunk, dbreads.StreamLTX, l.Chunk == last.LastSeq, l.Data) |
| 681 | if err != nil { |
| 682 | t.Fatalf("chunk %d: %v", l.Chunk, err) |
| 683 | } |
| 684 | h.Write(plain) |
| 685 | } |
| 686 | if hex.EncodeToString(h.Sum(nil)) != s.SHA256 { |
| 687 | t.Fatal("the pulled bytes do not add up to the summary") |
| 688 | } |
| 689 | at := time.Now().Add(-time.Second).UTC().Truncate(time.Second) |
| 690 | res = fx.openRead(t, dbfHost, fx.member, dbreads.ReadBody{Op: dbreads.OpPull, At: at.Format(time.RFC3339), |
| 691 | ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())}) |
| 692 | readLines(t, res.Body) |
| 693 | res.Body.Close() |
| 694 | evs, _ := fx.st.Events(OpsAccountID, "shop", 10) |
| 695 | for _, want := range []string{fmt.Sprintf(dbfAuditPull, "uk"), fmt.Sprintf(dbfAuditPullAt, "uk", at.Format(time.RFC3339))} { |
| 696 | found := false |
| 697 | for _, e := range evs { |
| 698 | found = found || (e.Action == "db-read" && e.Target == dbfHost && e.Detail == want) |
| 699 | } |
| 700 | if !found { |
| 701 | t.Errorf("no %q audit row in %+v", want, evs) |
| 702 | } |
| 703 | } |
| 704 | } |
| 705 | |
| 706 | func TestDBReadStreamsAQueryEndToEnd(t *testing.T) { |
| 707 | fx := newDBFFixture(t, true) |
| 708 | dir := t.TempDir() |
| 709 | dbdoortest.Replicate(t, dir, 3) |
| 710 | fx.startBox(t, dir) |
| 711 | body, key := queryBody(t, "SELECT count(*) AS n FROM t") |
| 712 | res := fx.openRead(t, dbfHost, fx.member, body) |
| 713 | defer res.Body.Close() |
| 714 | if res.StatusCode != http.StatusOK || res.Header.Get("Content-Type") != "application/x-ndjson" || |
| 715 | res.Header.Get("Cache-Control") != "no-store" || res.ContentLength != -1 { |
| 716 | t.Fatalf("%d %v", res.StatusCode, res.Header) |
| 717 | } |
| 718 | lines := readLines(t, res.Body) |
| 719 | if len(lines) < 2 || lines[0].Admitted == "" || lines[1].Picked != "uk-1" { |
| 720 | t.Fatalf("lines %+v", lines) |
| 721 | } |
| 722 | rows, _ := openAnswer(t, key, lines) |
| 723 | if string(rows) != "n\n30\n" { |
| 724 | t.Fatalf("rows %q", rows) |
| 725 | } |
| 726 | } |
| 727 | |
| 728 | // T1's second tenant, locally (spec §11.4: the host or prefix taken from |
| 729 | // anywhere but the record): two instances in uk, each with its own replica |
| 730 | // at the prefix the box derives for it, served in turn by one box. Each read |
| 731 | // gets its own rows. Mutation: the agent opens the first host it ever |
| 732 | // checked (red: the second read gets the first instance's rows). |
| 733 | // TestDBForwardMinio makes the same check against real S3 with per-region |
| 734 | // credentials, in CI's MinIO step. |
| 735 | func TestDBReadEachTenantGetsItsOwnRows(t *testing.T) { |
| 736 | fx := newDBFFixture(t, true) |
| 737 | const other = "other.example.com" |
| 738 | putShopRecord(t, fx.consoleFixture, provision.Record{Host: other, Region: "uk", CreatedAt: time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)}) |
| 739 | root := t.TempDir() |
| 740 | dbdoortest.Replicate(t, filepath.Join(root, "instances", dbfHost), 3) |
| 741 | dbdoortest.Replicate(t, filepath.Join(root, "instances", other), 1) |
| 742 | box := dbreadstest.Start(t, dbreadstest.Options{Store: fx.rel, Region: "uk", Box: "uk-1", Pub: fx.pub, HibernatePrefix: "instances", |
| 743 | OpenReplica: func(_ context.Context, prefix string) (dbdoor.ReplicaReader, error) { |
| 744 | return lsfile.NewReplicaClient(filepath.Join(root, prefix)), nil |
| 745 | }}) |
| 746 | runBox(t, box) |
| 747 | for _, c := range []struct{ host, want string }{{dbfHost, "n\n30\n"}, {other, "n\n10\n"}} { |
| 748 | body, key := queryBody(t, "SELECT count(*) AS n FROM t") |
| 749 | res := fx.openRead(t, c.host, fx.member, body) |
| 750 | lines := readLines(t, res.Body) |
| 751 | res.Body.Close() |
| 752 | if rows, _ := openAnswer(t, key, lines); string(rows) != c.want { |
| 753 | t.Fatalf("%s: %q, want %q", c.host, rows, c.want) |
| 754 | } |
| 755 | } |
| 756 | } |
| 757 | |
| 758 | // No box: wait lines while the console waits, then ended not-picked after |
| 759 | // the pickup window, a cancel, and the slot free again. |
| 760 | func TestDBReadEndsNotPickedAndCancels(t *testing.T) { |
| 761 | fx := newDBFFixture(t, true) |
| 762 | // issued_at is truncated to the second (production's semantics), so the |
| 763 | // pickup window seen from admission is pickup minus up to a second: 1.4 s |
| 764 | // leaves at least 400 ms, eight wait periods (plan review finding 30). |
| 765 | fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.pickup = 10*time.Millisecond, 50*time.Millisecond, 1400*time.Millisecond |
| 766 | body, _ := queryBody(t, "SELECT 1") |
| 767 | res := fx.openRead(t, dbfHost, fx.member, body) |
| 768 | lines := readLines(t, res.Body) |
| 769 | res.Body.Close() |
| 770 | waits := 0 |
| 771 | for _, l := range lines { |
| 772 | if l.Wait == dbreads.WaitPickup { |
| 773 | waits++ |
| 774 | } |
| 775 | } |
| 776 | last := lines[len(lines)-1] |
| 777 | if waits < 3 || last.Ended != dbreads.EndedNotPicked || last.Result != nil { |
| 778 | t.Fatalf("lines %+v", lines) |
| 779 | } |
| 780 | if _, _, err := fx.rel.Get(context.Background(), dbreads.CancelKey(lines[0].Admitted)); err != nil { |
| 781 | t.Fatalf("no cancel: %v", err) |
| 782 | } |
| 783 | } |
| 784 | |
| 785 | // T7, the console's half: admitted leaves with the headers, and every line |
| 786 | // is flushed as it is written. |
| 787 | func TestDBReadLinesAreFlushedAsWritten(t *testing.T) { |
| 788 | fx := newDBFFixture(t, true) |
| 789 | // At least a second of pickup is left after issued_at's truncation: room |
| 790 | // for the five lines below at 100 ms each (plan review finding 30). |
| 791 | fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.pickup = 10*time.Millisecond, 100*time.Millisecond, 2*time.Second |
| 792 | body, _ := queryBody(t, "SELECT 1") |
| 793 | start := time.Now() |
| 794 | res := fx.openRead(t, dbfHost, fx.member, body) |
| 795 | defer res.Body.Close() |
| 796 | br := bufio.NewReader(res.Body) |
| 797 | prev := start |
| 798 | for i := 0; i < 5; i++ { |
| 799 | if _, err := br.ReadBytes('\n'); err != nil { |
| 800 | t.Fatal(err) |
| 801 | } |
| 802 | if gap := time.Since(prev); gap > 400*time.Millisecond { |
| 803 | t.Fatalf("line %d arrived %s after the last: not flushed", i, gap) |
| 804 | } |
| 805 | prev = time.Now() |
| 806 | } |
| 807 | } |
| 808 | |
| 809 | // dbfOpen is one open read: its response, a reader over its lines, and its |
| 810 | // id ("" when it was refused). |
| 811 | type dbfOpen struct { |
| 812 | res *http.Response |
| 813 | br *bufio.Reader |
| 814 | id string |
| 815 | } |
| 816 | |
| 817 | func (fx *dbfFixture) openHeld(t *testing.T, base, bearer string) dbfOpen { |
| 818 | t.Helper() |
| 819 | body, _ := queryBody(t, "SELECT 1") |
| 820 | raw, _ := json.Marshal(body) |
| 821 | req, _ := http.NewRequest(http.MethodPost, base+"/api/cli/apps/"+release.OpsAccount+"/shop/db/"+dbfHost+"/reads", bytes.NewReader(raw)) |
| 822 | req.Header.Set("Content-Type", "application/json") |
| 823 | req.Header.Set("Authorization", "Bearer "+bearer) |
| 824 | res, err := http.DefaultClient.Do(req) |
| 825 | if err != nil { |
| 826 | t.Fatal(err) |
| 827 | } |
| 828 | o := dbfOpen{res: res, br: bufio.NewReader(res.Body)} |
| 829 | if res.StatusCode == http.StatusOK { |
| 830 | line, _ := o.br.ReadBytes('\n') |
| 831 | var l dbreads.Line |
| 832 | json.Unmarshal(line, &l) |
| 833 | o.id = l.Admitted |
| 834 | } |
| 835 | return o |
| 836 | } |
| 837 | |
| 838 | // untilPicked reads lines until the console has written picked, so the |
| 839 | // handler is in the owner's phase. |
| 840 | func (o dbfOpen) untilPicked(t *testing.T) { |
| 841 | t.Helper() |
| 842 | for { |
| 843 | line, err := o.br.ReadBytes('\n') |
| 844 | if err != nil { |
| 845 | t.Fatalf("%s ended before picked: %v", o.id, err) |
| 846 | } |
| 847 | var l dbreads.Line |
| 848 | json.Unmarshal(line, &l) |
| 849 | if l.Picked != "" { |
| 850 | return |
| 851 | } |
| 852 | } |
| 853 | } |
| 854 | |
| 855 | func refusedTooMany(t *testing.T, o dbfOpen, what string) { |
| 856 | t.Helper() |
| 857 | status, body := refusal(t, o.res) |
| 858 | if status != http.StatusTooManyRequests || o.res.Header.Get("Retry-After") == "" || !strings.Contains(body, dbfTooMany) { |
| 859 | t.Fatalf("%s: %d %s", what, status, body) |
| 860 | } |
| 861 | } |
| 862 | |
| 863 | // admitWithin retries until a read is admitted: a handler frees its slot when |
| 864 | // it notices its client went, which takes up to a poll. |
| 865 | func (fx *dbfFixture) admitWithin(t *testing.T, base, bearer string) dbfOpen { |
| 866 | t.Helper() |
| 867 | deadline := time.Now().Add(5 * time.Second) |
| 868 | for { |
| 869 | o := fx.openHeld(t, base, bearer) |
| 870 | if o.id != "" { |
| 871 | return o |
| 872 | } |
| 873 | o.res.Body.Close() |
| 874 | if time.Now().After(deadline) { |
| 875 | t.Fatal("a disconnected read still holds its slot") |
| 876 | } |
| 877 | time.Sleep(20 * time.Millisecond) |
| 878 | } |
| 879 | } |
| 880 | |
| 881 | // DBF-9 and T15 (J5; plan review findings 7 and 18): two open reads per |
| 882 | // person and sixteen per console, held by the open requests. Both of a |
| 883 | // person's slots come back when both CLIs leave before pickup, and again |
| 884 | // when both leave after it; the seventeenth read across nine people is |
| 885 | // refused; each disconnected read writes cancel; and a console restarted |
| 886 | // under open reads recovers nothing and starts with empty counts. |
| 887 | // Mutations: drop handleAPIDBRead's defer release (red: the second pair is |
| 888 | // never admitted); release at the end of admitDBRead (red: the third read of |
| 889 | // a pair is admitted); skip cancelDBRead on a client that went (red: no |
| 890 | // cancel); a startup pass that writes under db-reads/ (red: the listing |
| 891 | // changes across NewServer). |
| 892 | func TestDBReadAdmissionBounds(t *testing.T) { |
| 893 | fx := newDBFFixture(t, true) |
| 894 | fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.pickup = 10*time.Millisecond, 50*time.Millisecond, time.Minute |
| 895 | base := fx.ts.URL |
| 896 | var left []string |
| 897 | |
| 898 | // Before pickup. |
| 899 | a, b := fx.openHeld(t, base, fx.member), fx.openHeld(t, base, fx.member) |
| 900 | if a.id == "" || b.id == "" { |
| 901 | t.Fatal("the first two reads were refused") |
| 902 | } |
| 903 | refusedTooMany(t, fx.openHeld(t, base, fx.member), "a third read") |
| 904 | a.res.Body.Close() |
| 905 | b.res.Body.Close() |
| 906 | left = append(left, a.id, b.id) |
| 907 | |
| 908 | // Both slots come back; the pair is picked, then both CLIs leave. |
| 909 | c, d := fx.admitWithin(t, base, fx.member), fx.admitWithin(t, base, fx.member) |
| 910 | refusedTooMany(t, fx.openHeld(t, base, fx.member), "a third read after the first pair left") |
| 911 | for _, o := range []dbfOpen{c, d} { |
| 912 | scriptFor(t, fx.rel, o.id).picked(time.Now().Add(time.Minute)) |
| 913 | o.untilPicked(t) |
| 914 | o.res.Body.Close() |
| 915 | left = append(left, o.id) |
| 916 | } |
| 917 | e, f := fx.admitWithin(t, base, fx.member), fx.admitWithin(t, base, fx.member) |
| 918 | refusedTooMany(t, fx.openHeld(t, base, fx.member), "a third read after the picked pair left") |
| 919 | e.res.Body.Close() |
| 920 | f.res.Body.Close() |
| 921 | left = append(left, e.id, f.id) |
| 922 | for _, id := range left { |
| 923 | deadline := time.Now().Add(5 * time.Second) |
| 924 | for { |
| 925 | if _, _, err := fx.rel.Get(context.Background(), dbreads.CancelKey(id)); err == nil { |
| 926 | break |
| 927 | } |
| 928 | if time.Now().After(deadline) { |
| 929 | t.Fatalf("%s: no cancel after its CLI left", id) |
| 930 | } |
| 931 | time.Sleep(20 * time.Millisecond) |
| 932 | } |
| 933 | } |
| 934 | |
| 935 | // Sixteen per console, across people. |
| 936 | var held []dbfOpen |
| 937 | people := dbreads.OpenPerConsole/dbreads.OpenPerPerson + 1 |
| 938 | for i := 0; i < people; i++ { |
| 939 | tok := fx.member |
| 940 | if i > 0 { |
| 941 | addr := fmt.Sprintf("member%d@example.com", i) |
| 942 | if err := fx.st.AddMember(OpsAccountID, addr, RoleMember, time.Now()); err != nil { |
| 943 | t.Fatal(err) |
| 944 | } |
| 945 | tok = cliToken(t, fx.consoleFixture, addr, fmt.Sprintf("code-dbf-%d", i)) |
| 946 | } |
| 947 | for j := 0; j < dbreads.OpenPerPerson; j++ { |
| 948 | if len(held) == dbreads.OpenPerConsole { |
| 949 | refusedTooMany(t, fx.openHeld(t, base, tok), "the seventeenth read on the console") |
| 950 | continue |
| 951 | } |
| 952 | o := fx.admitWithin(t, base, tok) |
| 953 | held = append(held, o) |
| 954 | } |
| 955 | } |
| 956 | if len(held) != dbreads.OpenPerConsole { |
| 957 | t.Fatalf("held %d reads", len(held)) |
| 958 | } |
| 959 | |
| 960 | // A restart under open reads: nothing to recover, nothing written, and |
| 961 | // the new console's counts start empty. |
| 962 | before, _ := fx.rel.List(context.Background(), dbreads.Prefix) |
| 963 | srv2, err := NewServer(fx.srv.cfg, fx.st) |
| 964 | if err != nil { |
| 965 | t.Fatal(err) |
| 966 | } |
| 967 | srv2.dbf = fx.srv.dbf |
| 968 | ts2 := httptest.NewServer(srv2) |
| 969 | defer ts2.Close() |
| 970 | after, _ := fx.rel.List(context.Background(), dbreads.Prefix) |
| 971 | if !slices.Equal(before, after) { |
| 972 | t.Fatalf("starting a console wrote under db-reads/: %v", after) |
| 973 | } |
| 974 | g := fx.openHeld(t, ts2.URL, fx.member) |
| 975 | if g.id == "" { |
| 976 | status, body := refusal(t, g.res) |
| 977 | t.Fatalf("the restarted console refused a read: %d %s", status, body) |
| 978 | } |
| 979 | g.res.Body.Close() |
| 980 | for _, o := range held { |
| 981 | o.res.Body.Close() |
| 982 | } |
| 983 | } |
| 984 | |
| 985 | // slowStore holds every call whose key contains match until free closes, |
| 986 | // ignoring its context: the worst store, for plan review finding 12. |
| 987 | type slowStore struct { |
| 988 | release.Store |
| 989 | match string |
| 990 | delay time.Duration // 0: until free closes |
| 991 | free chan struct{} |
| 992 | } |
| 993 | |
| 994 | func (s *slowStore) hold(key string) { |
| 995 | if !strings.Contains(key, s.match) { |
| 996 | return |
| 997 | } |
| 998 | if s.delay > 0 { |
| 999 | time.Sleep(s.delay) |
| 1000 | return |
| 1001 | } |
| 1002 | <-s.free |
| 1003 | } |
| 1004 | |
| 1005 | func (s *slowStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 1006 | s.hold(key) |
| 1007 | return s.Store.Get(ctx, key) |
| 1008 | } |
| 1009 | |
| 1010 | func (s *slowStore) List(ctx context.Context, prefix string) ([]string, error) { |
| 1011 | s.hold(prefix) |
| 1012 | return s.Store.List(ctx, prefix) |
| 1013 | } |
| 1014 | |
| 1015 | // Plan review finding 12: a store that stops answering, in either phase, |
| 1016 | // neither silences the stream nor holds it past the phase's deadline. |
| 1017 | // Mutations: call the store on r.Context() without dbfCall (red: no wait |
| 1018 | // line and no end until free closes); drop the pickup deadline context (red: |
| 1019 | // same); drop the owner's deadline context (red: same in the owner case). |
| 1020 | func TestDBReadStalledStoreKeepsTheStreamAndItsBound(t *testing.T) { |
| 1021 | for _, phase := range []string{"pickup", "owner"} { |
| 1022 | t.Run(phase, func(t *testing.T) { |
| 1023 | fx := newDBFFixture(t, true) |
| 1024 | fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.call = 10*time.Millisecond, 100*time.Millisecond, time.Minute |
| 1025 | fx.srv.dbf.pickup, fx.srv.dbf.grace = 1500*time.Millisecond, 0 |
| 1026 | ss := &slowStore{Store: fx.rel, match: "/picked", free: make(chan struct{})} |
| 1027 | want := dbreads.EndedNotPicked |
| 1028 | if phase == "owner" { |
| 1029 | ss.match, want = "/result.json", dbreads.EndedNoResult |
| 1030 | } |
| 1031 | defer close(ss.free) |
| 1032 | fx.srv.cfg.Releases = ss |
| 1033 | o := fx.openHeld(t, fx.ts.URL, fx.member) |
| 1034 | defer o.res.Body.Close() |
| 1035 | if phase == "owner" { |
| 1036 | scriptFor(t, fx.rel, o.id).picked(time.Now().Add(time.Second)) |
| 1037 | } |
| 1038 | start, prev := time.Now(), time.Now() |
| 1039 | var last dbreads.Line |
| 1040 | for { |
| 1041 | line, err := o.br.ReadBytes('\n') |
| 1042 | if err != nil { |
| 1043 | break |
| 1044 | } |
| 1045 | if gap := time.Since(prev); gap > 500*time.Millisecond { |
| 1046 | t.Fatalf("a %s gap with the store stalled", gap) |
| 1047 | } |
| 1048 | prev = time.Now() |
| 1049 | json.Unmarshal(line, &last) |
| 1050 | if time.Since(start) > 5*time.Second { |
| 1051 | t.Fatal("the stalled store held the request past its deadline") |
| 1052 | } |
| 1053 | } |
| 1054 | if last.Ended != want { |
| 1055 | t.Fatalf("last line %+v, want %s", last, want) |
| 1056 | } |
| 1057 | }) |
| 1058 | } |
| 1059 | } |
| 1060 | |
| 1061 | // Plan review finding 12: a relay slower than the time left ends at the |
| 1062 | // request's bound, without the rest of the chunks and without the result, |
| 1063 | // though both are in the bucket. issued_at is truncated to the second, so |
| 1064 | // the bound is 2 to 3 s after the request; relaying all forty chunks takes 6. |
| 1065 | // Mutation: relay without the owner's deadline, the chunk Gets on |
| 1066 | // r.Context() and no ownCtx check after them (red: all forty are relayed |
| 1067 | // and the request lasts 6 s). |
| 1068 | // Not witnessed alone: the ownCtx check before the result line; dbfCall |
| 1069 | // already cuts every call after the deadline, so that check guards only a |
| 1070 | // line write that outlasts it, which TestDBReadResultAfterTheDeadlineIsNotRelayed |
| 1071 | // forces with a held writer. |
| 1072 | func TestDBReadSlowRelayEndsAtTheBound(t *testing.T) { |
| 1073 | fx := newDBFFixture(t, true) |
| 1074 | fx.srv.dbf.poll, fx.srv.dbf.bound, fx.srv.dbf.grace = 10*time.Millisecond, 3*time.Second, 0 |
| 1075 | fx.srv.cfg.Releases = &slowStore{Store: fx.rel, match: "/out/", delay: 150 * time.Millisecond} |
| 1076 | start := time.Now() |
| 1077 | o := fx.openHeld(t, fx.ts.URL, fx.member) |
| 1078 | defer o.res.Body.Close() |
| 1079 | box := scriptFor(t, fx.rel, o.id) |
| 1080 | const n = 40 |
| 1081 | for seq := int64(1); seq <= n; seq++ { |
| 1082 | box.chunk(seq, seq == n, "row\n") |
| 1083 | } |
| 1084 | box.result(n, strings.Repeat("row\n", n)) |
| 1085 | box.picked(time.Now().Add(time.Minute)) |
| 1086 | lines := readLines(t, o.br) |
| 1087 | chunks := 0 |
| 1088 | for _, l := range lines { |
| 1089 | if l.Result != nil { |
| 1090 | t.Fatalf("a result relayed past the bound: %+v", l.Result) |
| 1091 | } |
| 1092 | if l.Chunk != 0 { |
| 1093 | chunks++ |
| 1094 | } |
| 1095 | } |
| 1096 | if took := time.Since(start); took > 4500*time.Millisecond || chunks >= n { |
| 1097 | t.Fatalf("relayed %d chunks in %s", chunks, took) |
| 1098 | } |
| 1099 | if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult { |
| 1100 | t.Fatalf("last line %+v", last) |
| 1101 | } |
| 1102 | } |
| 1103 | |
| 1104 | // scriptedBox writes a read's objects itself: a stand-in for a box, for the |
| 1105 | // orderings T18 needs. It seals real chunks to the request's reply key. |
| 1106 | type scriptedBox struct { |
| 1107 | t *testing.T |
| 1108 | s release.Store |
| 1109 | pub *ecdh.PublicKey |
| 1110 | id string |
| 1111 | } |
| 1112 | |
| 1113 | func scriptFor(t *testing.T, s release.Store, id string) *scriptedBox { |
| 1114 | t.Helper() |
| 1115 | raw, _, err := s.Get(context.Background(), dbreads.RequestKey(id)) |
| 1116 | if err != nil { |
| 1117 | t.Fatal(err) |
| 1118 | } |
| 1119 | var w struct{ Payload string } |
| 1120 | json.Unmarshal(raw, &w) |
| 1121 | payload, _ := base64.StdEncoding.DecodeString(w.Payload) |
| 1122 | var req dbreads.Request |
| 1123 | json.Unmarshal(payload, &req) |
| 1124 | pub, err := dbreads.ParseReplyKey(req.ReplyKey) |
| 1125 | if err != nil { |
| 1126 | t.Fatal(err) |
| 1127 | } |
| 1128 | return &scriptedBox{t: t, s: s, pub: pub, id: id} |
| 1129 | } |
| 1130 | |
| 1131 | func (b *scriptedBox) picked(deadline time.Time) { |
| 1132 | p, _ := json.Marshal(dbreads.Picked{Box: "uk-1", Nonce: "n", Deadline: deadline}) |
| 1133 | b.s.Put(context.Background(), dbreads.PickedKey(b.id), p, "") |
| 1134 | } |
| 1135 | |
| 1136 | func (b *scriptedBox) chunk(seq int64, last bool, plain string) { |
| 1137 | sealed, _ := dbreads.SealChunk(b.pub, b.id, seq, dbreads.StreamRows, last, []byte(plain)) |
| 1138 | b.s.Put(context.Background(), dbreads.ChunkKey(b.id, seq), sealed, "") |
| 1139 | } |
| 1140 | |
| 1141 | func (b *scriptedBox) result(last int64, rows string) { |
| 1142 | sum := sha256.Sum256([]byte(rows)) |
| 1143 | sealed, _ := dbreads.SealSummary(b.pub, b.id, dbreads.Summary{TXID: 9, Bytes: int64(len(rows)), SHA256: hex.EncodeToString(sum[:])}) |
| 1144 | r, _ := json.Marshal(dbreads.Result{State: dbreads.StateDone, LastSeq: last, Box: "uk-1", Sealed: sealed}) |
| 1145 | b.s.Put(context.Background(), dbreads.ResultKey(b.id), r, "") |
| 1146 | } |
| 1147 | |
| 1148 | // dbReadHookStore runs once on the console's first listing of a read's |
| 1149 | // chunks, after the listing is taken: a box committing between the console's |
| 1150 | // listing and its next result read. It also records, in order, each result |
| 1151 | // read and each chunk listing. (hookStore is temphook_test.go's: plan review |
| 1152 | // finding 2.) |
| 1153 | type dbReadHookStore struct { |
| 1154 | release.Store |
| 1155 | once sync.Once |
| 1156 | hook func() |
| 1157 | mu sync.Mutex |
| 1158 | ops []string |
| 1159 | } |
| 1160 | |
| 1161 | func (h *dbReadHookStore) record(op string) { |
| 1162 | h.mu.Lock() |
| 1163 | defer h.mu.Unlock() |
| 1164 | h.ops = append(h.ops, op) |
| 1165 | } |
| 1166 | |
| 1167 | func (h *dbReadHookStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 1168 | if strings.HasSuffix(key, "/result.json") { |
| 1169 | h.record("result") |
| 1170 | } |
| 1171 | return h.Store.Get(ctx, key) |
| 1172 | } |
| 1173 | |
| 1174 | func (h *dbReadHookStore) List(ctx context.Context, prefix string) ([]string, error) { |
| 1175 | keys, err := h.Store.List(ctx, prefix) |
| 1176 | if strings.HasSuffix(prefix, "/out/") { |
| 1177 | h.record("list") |
| 1178 | h.once.Do(h.hook) |
| 1179 | } |
| 1180 | return keys, err |
| 1181 | } |
| 1182 | |
| 1183 | // T18 (codex 20): the box commits its last chunk and result.json between the |
| 1184 | // console's listing and its result read, in both orders. The result line |
| 1185 | // never comes before every chunk through last_seq, the read succeeds, and in |
| 1186 | // every round the console reads result.json before it lists the chunks. |
| 1187 | // Mutations, each red on its own (plan review finding 20): list out/ before |
| 1188 | // reading result.json (the operation-order check), or write the result line |
| 1189 | // without waiting for next > last_seq (the chunk-count check). |
| 1190 | func TestDBReadResultWaitsForItsLastChunk(t *testing.T) { |
| 1191 | for _, resultFirst := range []bool{false, true} { |
| 1192 | fx := newDBFFixture(t, true) |
| 1193 | fx.srv.dbf.poll = 10 * time.Millisecond |
| 1194 | hs := &dbReadHookStore{Store: fx.rel} |
| 1195 | fx.srv.cfg.Releases = hs |
| 1196 | body, key := queryBody(t, "SELECT 1") |
| 1197 | res := fx.openRead(t, dbfHost, fx.member, body) |
| 1198 | br := bufio.NewReader(res.Body) |
| 1199 | first, _ := br.ReadBytes('\n') |
| 1200 | var adm dbreads.Line |
| 1201 | json.Unmarshal(first, &adm) |
| 1202 | box := scriptFor(t, fx.rel, adm.Admitted) |
| 1203 | hs.hook = func() { |
| 1204 | if resultFirst { |
| 1205 | box.result(2, "a\nb\n") |
| 1206 | box.chunk(2, true, "b\n") |
| 1207 | } else { |
| 1208 | box.chunk(2, true, "b\n") |
| 1209 | box.result(2, "a\nb\n") |
| 1210 | } |
| 1211 | } |
| 1212 | box.chunk(1, false, "a\n") |
| 1213 | box.picked(time.Now().Add(time.Minute)) |
| 1214 | lines := append([]dbreads.Line{adm}, readLines(t, br)...) |
| 1215 | res.Body.Close() |
| 1216 | chunks := 0 |
| 1217 | for _, l := range lines { |
| 1218 | if l.Chunk != 0 { |
| 1219 | chunks++ |
| 1220 | } |
| 1221 | if l.Result != nil && chunks != 2 { |
| 1222 | t.Fatalf("result-first %v: the result line came after %d chunks", resultFirst, chunks) |
| 1223 | } |
| 1224 | } |
| 1225 | if rows, _ := openAnswer(t, key, lines); string(rows) != "a\nb\n" { |
| 1226 | t.Fatalf("rows %q", rows) |
| 1227 | } |
| 1228 | hs.mu.Lock() |
| 1229 | ops := append([]string(nil), hs.ops...) |
| 1230 | hs.mu.Unlock() |
| 1231 | for i, op := range ops { |
| 1232 | if op == "list" && (i == 0 || ops[i-1] != "result") { |
| 1233 | t.Fatalf("result-first %v: a chunk listing not preceded by its round's result read: %v", resultFirst, ops) |
| 1234 | } |
| 1235 | } |
| 1236 | } |
| 1237 | } |
| 1238 | |
| 1239 | // dbfHideStore leaves chunk seq out of the first hide listings of a read's |
| 1240 | // chunks: a listing that lags the box's writes. |
| 1241 | type dbfHideStore struct { |
| 1242 | release.Store |
| 1243 | seq int64 |
| 1244 | mu sync.Mutex |
| 1245 | hide int |
| 1246 | } |
| 1247 | |
| 1248 | func (h *dbfHideStore) List(ctx context.Context, prefix string) ([]string, error) { |
| 1249 | keys, err := h.Store.List(ctx, prefix) |
| 1250 | if !strings.HasSuffix(prefix, "/out/") { |
| 1251 | return keys, err |
| 1252 | } |
| 1253 | h.mu.Lock() |
| 1254 | defer h.mu.Unlock() |
| 1255 | if h.hide == 0 { |
| 1256 | return keys, err |
| 1257 | } |
| 1258 | h.hide-- |
| 1259 | var out []string |
| 1260 | for _, k := range keys { |
| 1261 | if seq, ok := dbreads.ParseChunkKey(k, prefix); !ok || seq != h.seq { |
| 1262 | out = append(out, k) |
| 1263 | } |
| 1264 | } |
| 1265 | return out, err |
| 1266 | } |
| 1267 | |
| 1268 | // T18's completeness guard on its own (plan review finding 31): result.json |
| 1269 | // says last_seq 2 from the first round, and the first three listings miss |
| 1270 | // chunk 2. The result line waits until chunk 2 is relayed. Mutation: write |
| 1271 | // the result line without waiting for next > res.LastSeq (red: the result |
| 1272 | // comes after one chunk). |
| 1273 | func TestDBReadResultWaitsForAChunkTheListingMissed(t *testing.T) { |
| 1274 | fx := newDBFFixture(t, true) |
| 1275 | fx.srv.dbf.poll = 10 * time.Millisecond |
| 1276 | fx.srv.cfg.Releases = &dbfHideStore{Store: fx.rel, seq: 2, hide: 3} |
| 1277 | body, key := queryBody(t, "SELECT 1") |
| 1278 | res := fx.openRead(t, dbfHost, fx.member, body) |
| 1279 | defer res.Body.Close() |
| 1280 | br := bufio.NewReader(res.Body) |
| 1281 | first, _ := br.ReadBytes('\n') |
| 1282 | var adm dbreads.Line |
| 1283 | json.Unmarshal(first, &adm) |
| 1284 | box := scriptFor(t, fx.rel, adm.Admitted) |
| 1285 | box.chunk(1, false, "a\n") |
| 1286 | box.chunk(2, true, "b\n") |
| 1287 | box.result(2, "a\nb\n") |
| 1288 | box.picked(time.Now().Add(time.Minute)) |
| 1289 | lines := append([]dbreads.Line{adm}, readLines(t, br)...) |
| 1290 | chunks := 0 |
| 1291 | for _, l := range lines { |
| 1292 | if l.Chunk != 0 { |
| 1293 | chunks++ |
| 1294 | } |
| 1295 | if l.Result != nil && chunks != 2 { |
| 1296 | t.Fatalf("the result line came after %d chunks", chunks) |
| 1297 | } |
| 1298 | } |
| 1299 | if rows, _ := openAnswer(t, key, lines); string(rows) != "a\nb\n" { |
| 1300 | t.Fatalf("rows %q", rows) |
| 1301 | } |
| 1302 | } |
| 1303 | |
| 1304 | // dbfHeldWriter is a ResponseWriter that holds the write of a line |
| 1305 | // containing holdOn until until: a client that takes one line slowly. A |
| 1306 | // stand-in for a network client, in a unit test. |
| 1307 | type dbfHeldWriter struct { |
| 1308 | header http.Header |
| 1309 | holdOn string |
| 1310 | until time.Time |
| 1311 | mu sync.Mutex |
| 1312 | buf bytes.Buffer |
| 1313 | } |
| 1314 | |
| 1315 | func (w *dbfHeldWriter) Header() http.Header { return w.header } |
| 1316 | func (w *dbfHeldWriter) WriteHeader(int) {} |
| 1317 | func (w *dbfHeldWriter) Flush() {} |
| 1318 | |
| 1319 | func (w *dbfHeldWriter) Write(p []byte) (int, error) { |
| 1320 | if bytes.Contains(p, []byte(w.holdOn)) { |
| 1321 | time.Sleep(time.Until(w.until)) |
| 1322 | } |
| 1323 | w.mu.Lock() |
| 1324 | defer w.mu.Unlock() |
| 1325 | return w.buf.Write(p) |
| 1326 | } |
| 1327 | |
| 1328 | // Plan review finding 35: the owner's deadline holds for the result line |
| 1329 | // too. Every chunk and result.json are in the bucket, but the client takes |
| 1330 | // the last chunk's line until past the deadline: the read ends no-result, |
| 1331 | // and the result is not relayed. Mutation: drop the ownCtx check before the |
| 1332 | // result line (red: the result is relayed). |
| 1333 | func TestDBReadResultAfterTheDeadlineIsNotRelayed(t *testing.T) { |
| 1334 | fx := newDBFFixture(t, true) |
| 1335 | fx.srv.dbf.poll, fx.srv.dbf.grace = 10*time.Millisecond, 0 |
| 1336 | key, err := dbreads.NewReplyKey() |
| 1337 | if err != nil { |
| 1338 | t.Fatal(err) |
| 1339 | } |
| 1340 | id, err := dbreads.MintID() |
| 1341 | if err != nil { |
| 1342 | t.Fatal(err) |
| 1343 | } |
| 1344 | box := &scriptedBox{t: t, s: fx.rel, pub: key.PublicKey(), id: id} |
| 1345 | deadline := time.Now().Add(800 * time.Millisecond) |
| 1346 | box.chunk(1, true, "a\n") |
| 1347 | box.result(1, "a\n") |
| 1348 | box.picked(deadline) |
| 1349 | w := &dbfHeldWriter{header: http.Header{}, holdOn: `"chunk":1,`, until: deadline.Add(200 * time.Millisecond)} |
| 1350 | r := httptest.NewRequest(http.MethodPost, "/", nil) |
| 1351 | fx.srv.streamDBRead(w, r, http.NewResponseController(w), dbReadAdmitted{id: id, region: "uk", issued: time.Now().UTC().Truncate(time.Second)}) |
| 1352 | w.mu.Lock() |
| 1353 | lines := readLines(t, bytes.NewReader(w.buf.Bytes())) |
| 1354 | w.mu.Unlock() |
| 1355 | chunks := 0 |
| 1356 | for _, l := range lines { |
| 1357 | if l.Result != nil { |
| 1358 | t.Fatalf("a result relayed past the owner's deadline: %+v", l.Result) |
| 1359 | } |
| 1360 | if l.Chunk != 0 { |
| 1361 | chunks++ |
| 1362 | } |
| 1363 | } |
| 1364 | if last := lines[len(lines)-1]; chunks != 1 || last.Ended != dbreads.EndedNoResult { |
| 1365 | t.Fatalf("%d chunks, last line %+v", chunks, last) |
| 1366 | } |
| 1367 | } |
| 1368 | |
| 1369 | // dbfDeadlineListener records, for every read deadline the server sets on |
| 1370 | // its connections, how far ahead it was set. |
| 1371 | type dbfDeadlineListener struct { |
| 1372 | net.Listener |
| 1373 | mu sync.Mutex |
| 1374 | ahead []time.Duration |
| 1375 | } |
| 1376 | |
| 1377 | type dbfDeadlineConn struct { |
| 1378 | net.Conn |
| 1379 | l *dbfDeadlineListener |
| 1380 | } |
| 1381 | |
| 1382 | func (l *dbfDeadlineListener) Accept() (net.Conn, error) { |
| 1383 | c, err := l.Listener.Accept() |
| 1384 | if err != nil { |
| 1385 | return nil, err |
| 1386 | } |
| 1387 | return dbfDeadlineConn{Conn: c, l: l}, nil |
| 1388 | } |
| 1389 | |
| 1390 | func (c dbfDeadlineConn) SetReadDeadline(t time.Time) error { |
| 1391 | if !t.IsZero() { |
| 1392 | c.l.mu.Lock() |
| 1393 | c.l.ahead = append(c.l.ahead, time.Until(t)) |
| 1394 | c.l.mu.Unlock() |
| 1395 | } |
| 1396 | return c.Conn.SetReadDeadline(t) |
| 1397 | } |
| 1398 | |
| 1399 | // Astra round 1, P1: complete query JSON followed by padding, with the rest |
| 1400 | // of the declared body withheld. A decoder that stops after one JSON value |
| 1401 | // would admit the read, and the server would later drain the tail under the |
| 1402 | // extended read deadline. The whole body is read under s.dbf.preHeader |
| 1403 | // instead, so the request is cut there and nothing is published. |
| 1404 | // Mutation: decode one value without reading to EOF (red: the read is |
| 1405 | // queued and the connection is still held at 3 s). |
| 1406 | func TestDBReadWithheldBodyTailIsCutAtThePreHeaderBound(t *testing.T) { |
| 1407 | fx := newDBFFixture(t, true) |
| 1408 | fx.srv.dbf.preHeader = 300 * time.Millisecond |
| 1409 | body, _ := queryBody(t, "SELECT 1") |
| 1410 | raw, _ := json.Marshal(body) |
| 1411 | conn, err := net.Dial("tcp", strings.TrimPrefix(fx.ts.URL, "http://")) |
| 1412 | if err != nil { |
| 1413 | t.Fatal(err) |
| 1414 | } |
| 1415 | defer conn.Close() |
| 1416 | path := strings.TrimPrefix(fx.readURL(dbfHost), fx.ts.URL) |
| 1417 | fmt.Fprintf(conn, "POST %s HTTP/1.1\r\nHost: console\r\nAuthorization: Bearer %s\r\n"+ |
| 1418 | "Content-Type: application/json\r\nContent-Length: 16384\r\n\r\n%s%s", path, fx.member, raw, strings.Repeat(" ", 64)) |
| 1419 | conn.SetReadDeadline(time.Now().Add(3 * time.Second)) |
| 1420 | start := time.Now() |
| 1421 | _, err = io.ReadAll(conn) |
| 1422 | if ne, ok := err.(net.Error); ok && ne.Timeout() { |
| 1423 | t.Fatalf("the withheld body tail still held the request after %s", time.Since(start)) |
| 1424 | } |
| 1425 | if keys, _ := fx.rel.List(context.Background(), dbreads.Prefix); len(keys) != 0 { |
| 1426 | t.Fatalf("a request with a withheld body tail wrote %v", keys) |
| 1427 | } |
| 1428 | } |
| 1429 | |
| 1430 | // A body holding more than one JSON value is refused, not half-read. |
| 1431 | func TestDBReadRefusesTrailingJSON(t *testing.T) { |
| 1432 | fx := newDBFFixture(t, true) |
| 1433 | body, _ := queryBody(t, "SELECT 1") |
| 1434 | raw, _ := json.Marshal(body) |
| 1435 | req, _ := http.NewRequest(http.MethodPost, fx.readURL(dbfHost), strings.NewReader(string(raw)+`{"op":"query"}`)) |
| 1436 | req.Header.Set("Content-Type", "application/json") |
| 1437 | req.Header.Set("Authorization", "Bearer "+fx.member) |
| 1438 | res, err := http.DefaultClient.Do(req) |
| 1439 | if err != nil { |
| 1440 | t.Fatal(err) |
| 1441 | } |
| 1442 | if status, out := refusal(t, res); status != http.StatusBadRequest { |
| 1443 | t.Fatalf("%d %s", status, out) |
| 1444 | } |
| 1445 | if keys, _ := fx.rel.List(context.Background(), dbreads.Prefix); len(keys) != 0 { |
| 1446 | t.Fatalf("a body with trailing JSON wrote %v", keys) |
| 1447 | } |
| 1448 | } |
| 1449 | |
| 1450 | // Plan review findings 13 and 38: admission reads under s.dbf.preHeader, |
| 1451 | // and once admitted the read deadline moves out to the read's bound plus a |
| 1452 | // minute. Observed on the connection itself: once a body has been read to |
| 1453 | // its end, net/http's background read clears the read deadline on its own, |
| 1454 | // so a stream that outlives ReadTimeout does not show the extension. |
| 1455 | // Mutations: drop the extension (red: no deadline past 22 minutes); drop the |
| 1456 | // preHeader deadline (red: none between 15 and 20 s; the server's own are |
| 1457 | // 10 s and 5 minutes). |
| 1458 | func TestDBReadSetsBothReadDeadlines(t *testing.T) { |
| 1459 | fx := newDBFFixture(t, true) |
| 1460 | fx.srv.dbf.poll, fx.srv.dbf.pickup = 10*time.Millisecond, 1500*time.Millisecond |
| 1461 | raw, err := net.Listen("tcp", "127.0.0.1:0") |
| 1462 | if err != nil { |
| 1463 | t.Fatal(err) |
| 1464 | } |
| 1465 | ln := &dbfDeadlineListener{Listener: raw} |
| 1466 | hs := NewHTTPServer(fx.srv) |
| 1467 | go hs.Serve(ln) |
| 1468 | t.Cleanup(func() { hs.Close() }) |
| 1469 | body, _ := queryBody(t, "SELECT 1") |
| 1470 | rawBody, _ := json.Marshal(body) |
| 1471 | req, _ := http.NewRequest(http.MethodPost, "http://"+raw.Addr().String()+strings.TrimPrefix(fx.readURL(dbfHost), fx.ts.URL), bytes.NewReader(rawBody)) |
| 1472 | req.Header.Set("Content-Type", "application/json") |
| 1473 | req.Header.Set("Authorization", "Bearer "+fx.member) |
| 1474 | res, err := http.DefaultClient.Do(req) |
| 1475 | if err != nil { |
| 1476 | t.Fatal(err) |
| 1477 | } |
| 1478 | io.Copy(io.Discard, res.Body) |
| 1479 | res.Body.Close() |
| 1480 | ln.mu.Lock() |
| 1481 | defer ln.mu.Unlock() |
| 1482 | var pre, extended bool |
| 1483 | for _, d := range ln.ahead { |
| 1484 | pre = pre || (d > 15*time.Second && d <= 20*time.Second) |
| 1485 | extended = extended || d > 22*time.Minute |
| 1486 | } |
| 1487 | if !pre || !extended { |
| 1488 | t.Fatalf("read deadlines set this far ahead: %v", ln.ahead) |
| 1489 | } |
| 1490 | } |
| 1491 | |
| 1492 | // Review Focus 3: a picked deadline far in the future does not hold the |
| 1493 | // request past its own bound. |
| 1494 | func TestDBReadRequestBoundClampsAForgedDeadline(t *testing.T) { |
| 1495 | fx := newDBFFixture(t, true) |
| 1496 | fx.srv.dbf.poll, fx.srv.dbf.bound = 10*time.Millisecond, 1500*time.Millisecond |
| 1497 | body, _ := queryBody(t, "SELECT 1") |
| 1498 | res := fx.openRead(t, dbfHost, fx.member, body) |
| 1499 | br := bufio.NewReader(res.Body) |
| 1500 | first, _ := br.ReadBytes('\n') |
| 1501 | var adm dbreads.Line |
| 1502 | json.Unmarshal(first, &adm) |
| 1503 | scriptFor(t, fx.rel, adm.Admitted).picked(time.Date(2100, 1, 1, 0, 0, 0, 0, time.UTC)) |
| 1504 | done := make(chan []dbreads.Line) |
| 1505 | go func() { done <- readLines(t, br) }() |
| 1506 | select { |
| 1507 | case lines := <-done: |
| 1508 | if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult { |
| 1509 | t.Fatalf("last line %+v", last) |
| 1510 | } |
| 1511 | case <-time.After(5 * time.Second): |
| 1512 | t.Fatal("the request outlived its bound") |
| 1513 | } |
| 1514 | res.Body.Close() |
| 1515 | } |
| 1516 | |
| 1517 | // Review Focus 4: a result.json that is not a result is never relayed. |
| 1518 | func TestDBReadIgnoresAMalformedResult(t *testing.T) { |
| 1519 | fx := newDBFFixture(t, true) |
| 1520 | fx.srv.dbf.poll, fx.srv.dbf.grace = 10*time.Millisecond, 0 |
| 1521 | body, _ := queryBody(t, "SELECT 1") |
| 1522 | res := fx.openRead(t, dbfHost, fx.member, body) |
| 1523 | br := bufio.NewReader(res.Body) |
| 1524 | first, _ := br.ReadBytes('\n') |
| 1525 | var adm dbreads.Line |
| 1526 | json.Unmarshal(first, &adm) |
| 1527 | box := scriptFor(t, fx.rel, adm.Admitted) |
| 1528 | box.picked(time.Now().Add(500 * time.Millisecond)) |
| 1529 | fx.rel.Put(context.Background(), dbreads.ResultKey(adm.Admitted), []byte(`{"state":"maybe","last_seq":-1}`), "") |
| 1530 | lines := readLines(t, br) |
| 1531 | res.Body.Close() |
| 1532 | for _, l := range lines { |
| 1533 | if l.Result != nil { |
| 1534 | t.Fatalf("a malformed result was relayed: %+v", l.Result) |
| 1535 | } |
| 1536 | } |
| 1537 | if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult { |
| 1538 | t.Fatalf("last line %+v", last) |
| 1539 | } |
| 1540 | } |
| 1541 | |
| 1542 | // strictStore fails the test on any delete or replacing write under |
| 1543 | // db-reads/ (T16, the console's half). |
| 1544 | type strictStore struct { |
| 1545 | release.Store |
| 1546 | t *testing.T |
| 1547 | } |
| 1548 | |
| 1549 | func (s strictStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| 1550 | if strings.HasPrefix(key, dbreads.Prefix) && ifMatch != "" { |
| 1551 | s.t.Errorf("a replacing write of %s", key) |
| 1552 | } |
| 1553 | return s.Store.Put(ctx, key, body, ifMatch) |
| 1554 | } |
| 1555 | |
| 1556 | func (s strictStore) Delete(ctx context.Context, key string) error { |
| 1557 | if strings.HasPrefix(key, dbreads.Prefix) { |
| 1558 | s.t.Errorf("a delete of %s", key) |
| 1559 | } |
| 1560 | return s.Store.Delete(ctx, key) |
| 1561 | } |
| 1562 | |
| 1563 | // T16 (codex 4, 13, 14): a box that writes after the request ended changes |
| 1564 | // nothing any reader sees, and the console never deletes or replaces. |
| 1565 | func TestDBReadLateWritesChangeNothing(t *testing.T) { |
| 1566 | fx := newDBFFixture(t, true) |
| 1567 | fx.srv.cfg.Releases = strictStore{Store: fx.rel, t: t} |
| 1568 | fx.srv.dbf.poll, fx.srv.dbf.grace = 10*time.Millisecond, 0 |
| 1569 | body, _ := queryBody(t, "SELECT 1") |
| 1570 | res := fx.openRead(t, dbfHost, fx.member, body) |
| 1571 | br := bufio.NewReader(res.Body) |
| 1572 | first, _ := br.ReadBytes('\n') |
| 1573 | var adm dbreads.Line |
| 1574 | json.Unmarshal(first, &adm) |
| 1575 | box := scriptFor(t, fx.rel, adm.Admitted) |
| 1576 | box.picked(time.Now().Add(200 * time.Millisecond)) |
| 1577 | lines := readLines(t, br) // ends no-result at the deadline |
| 1578 | res.Body.Close() |
| 1579 | box.chunk(1, true, "late\n") |
| 1580 | box.result(1, "late\n") |
| 1581 | if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult { |
| 1582 | t.Fatalf("last line %+v", last) |
| 1583 | } |
| 1584 | keys, _ := fx.rel.List(context.Background(), dbreads.Prefix+adm.Admitted+"/") |
| 1585 | for _, k := range keys { |
| 1586 | if strings.HasSuffix(k, "/cancel") { |
| 1587 | return |
| 1588 | } |
| 1589 | } |
| 1590 | t.Fatalf("no cancel among %v", keys) |
| 1591 | } |
| 1592 | |
amadan