| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "crypto/ecdh" |
| 7 | "crypto/sha256" |
| 8 | "database/sql" |
| 9 | "encoding/hex" |
| 10 | "encoding/json" |
| 11 | "errors" |
| 12 | "fmt" |
| 13 | "io" |
| 14 | "net/http" |
| 15 | "net/http/httptest" |
| 16 | "os" |
| 17 | "path/filepath" |
| 18 | "strings" |
| 19 | "sync/atomic" |
| 20 | "testing" |
| 21 | "time" |
| 22 | |
| 23 | lsfile "github.com/benbjohnson/litestream/file" |
| 24 | "github.com/superfly/ltx" |
| 25 | |
| 26 | "github.com/carlosframework/platform/internal/dbdoor" |
| 27 | "github.com/carlosframework/platform/internal/dbdoor/dbdoortest" |
| 28 | "github.com/carlosframework/platform/internal/dbreads" |
| 29 | ) |
| 30 | |
| 31 | // These feed readDBForward crafted line streams, sealed with real keys: |
| 32 | // unit tests of the CLI's half (dbforward_e2e_test.go runs the binary). |
| 33 | |
| 34 | const fwdID = "dbr_aaaaaaaaaaaaaaaaaaaaaaaaaa" |
| 35 | |
| 36 | // stream builds a reads-route body: admitted, picked, the chunks, then end. |
| 37 | type stream struct { |
| 38 | t *testing.T |
| 39 | pub *ecdh.PublicKey |
| 40 | buf bytes.Buffer |
| 41 | } |
| 42 | |
| 43 | func newStream(t *testing.T, key *ecdh.PrivateKey) *stream { |
| 44 | s := &stream{t: t, pub: key.PublicKey()} |
| 45 | s.line(dbreads.Line{Admitted: fwdID}) |
| 46 | s.line(dbreads.Line{Picked: "uk-1"}) |
| 47 | return s |
| 48 | } |
| 49 | |
| 50 | func (s *stream) line(l dbreads.Line) { |
| 51 | b, _ := json.Marshal(l) |
| 52 | s.buf.Write(append(b, '\n')) |
| 53 | } |
| 54 | |
| 55 | func (s *stream) chunk(seq int64, last bool, plain []byte) { |
| 56 | sealed, err := dbreads.SealChunk(s.pub, fwdID, seq, dbreads.StreamRows, last, plain) |
| 57 | if err != nil { |
| 58 | s.t.Fatal(err) |
| 59 | } |
| 60 | s.line(dbreads.Line{Chunk: seq, Data: sealed}) |
| 61 | } |
| 62 | |
| 63 | func (s *stream) result(state, code string, last int64, sum dbreads.Summary) { |
| 64 | sealed, _ := dbreads.SealSummary(s.pub, fwdID, sum) |
| 65 | s.line(dbreads.Line{Result: &dbreads.Result{State: state, Code: code, LastSeq: last, Sealed: sealed}}) |
| 66 | } |
| 67 | |
| 68 | func summaryOf(rows string) dbreads.Summary { |
| 69 | h := sha256.Sum256([]byte(rows)) |
| 70 | return dbreads.Summary{TXID: 7, At: time.Date(2026, 10, 7, 14, 0, 0, 0, time.UTC), Bytes: int64(len(rows)), SHA256: hex.EncodeToString(h[:])} |
| 71 | } |
| 72 | |
| 73 | func readFwd(t *testing.T, s *stream, key *ecdh.PrivateKey) (dbForwardAnswer, string, error) { |
| 74 | t.Helper() |
| 75 | var status bytes.Buffer |
| 76 | sink := &memSink{} |
| 77 | id, res, err := readDBForward(context.Background(), &s.buf, key, "shop.example.com", "uk", dbfReading, &status, sink, newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes)) |
| 78 | if err != nil { |
| 79 | return dbForwardAnswer{}, status.String(), err |
| 80 | } |
| 81 | ans, err := openDBForward(id, key, sink, *res, "shop.example.com", "uk") |
| 82 | return ans, status.String(), err |
| 83 | } |
| 84 | |
| 85 | func TestForwardedQueryPrintsOnlyACheckedAnswer(t *testing.T) { |
| 86 | t.Parallel() |
| 87 | key, _ := dbreads.NewReplyKey() |
| 88 | s := newStream(t, key) |
| 89 | s.chunk(1, false, []byte("n\n")) |
| 90 | s.chunk(2, true, []byte("30\n")) |
| 91 | s.result(dbreads.StateDone, "", 2, summaryOf("n\n30\n")) |
| 92 | ans, status, err := readFwd(t, s, key) |
| 93 | if err != nil || string(ans.rows) != "n\n30\n" || ans.summary.TXID != 7 { |
| 94 | t.Fatalf("%q %v", ans.rows, err) |
| 95 | } |
| 96 | if status != fmt.Sprintf(dbfReading, "shop.example.com", "uk")+"\n" { |
| 97 | t.Fatalf("status %q", status) |
| 98 | } |
| 99 | } |
| 100 | |
| 101 | // T11's unit half (codex 6): a dropped or bad last chunk and a summary that |
| 102 | // does not add up each print nothing. |
| 103 | func TestForwardedQueryRefusesAnAnswerThatDoesNotCheckOut(t *testing.T) { |
| 104 | t.Parallel() |
| 105 | unchecked := fmt.Sprintf(dbfUnchecked, "uk") |
| 106 | for name, build := range map[string]func(s *stream){ |
| 107 | "the last chunk dropped": func(s *stream) { |
| 108 | s.chunk(1, false, []byte("n\n")) |
| 109 | s.result(dbreads.StateDone, "", 2, summaryOf("n\n30\n")) |
| 110 | }, |
| 111 | "the last chunk not sealed as last": func(s *stream) { |
| 112 | s.chunk(1, false, []byte("n\n")) |
| 113 | s.chunk(2, false, []byte("30\n")) |
| 114 | s.result(dbreads.StateDone, "", 2, summaryOf("n\n30\n")) |
| 115 | }, |
| 116 | "a hash mismatch": func(s *stream) { |
| 117 | s.chunk(1, true, []byte("n\n31\n")) |
| 118 | s.result(dbreads.StateDone, "", 1, summaryOf("n\n30\n")) |
| 119 | }, |
| 120 | "a chunk out of order": func(s *stream) { |
| 121 | s.chunk(2, true, []byte("30\n")) |
| 122 | s.result(dbreads.StateDone, "", 1, summaryOf("30\n")) |
| 123 | }, |
| 124 | } { |
| 125 | key, _ := dbreads.NewReplyKey() |
| 126 | s := newStream(t, key) |
| 127 | build(s) |
| 128 | ans, _, err := readFwd(t, s, key) |
| 129 | if err == nil || err.Error() != unchecked || ans.rows != nil { |
| 130 | t.Errorf("%s: %q, %v", name, ans.rows, err) |
| 131 | } |
| 132 | } |
| 133 | } |
| 134 | |
| 135 | func TestForwardedQueryEndsWithoutAResult(t *testing.T) { |
| 136 | t.Parallel() |
| 137 | key, _ := dbreads.NewReplyKey() |
| 138 | s := newStream(t, key) |
| 139 | s.chunk(1, false, []byte("n\n")) |
| 140 | if _, _, err := readFwd(t, s, key); err == nil || err.Error() != fmt.Sprintf(dbfUnfinished, "shop.example.com", "uk") { |
| 141 | t.Fatalf("no result: %v", err) |
| 142 | } |
| 143 | key2, _ := dbreads.NewReplyKey() |
| 144 | s2 := &stream{t: t, pub: key2.PublicKey()} |
| 145 | s2.line(dbreads.Line{Admitted: fwdID}) |
| 146 | s2.line(dbreads.Line{Wait: dbreads.WaitPickup}) |
| 147 | s2.line(dbreads.Line{Ended: dbreads.EndedNotPicked}) |
| 148 | if _, _, err := readFwd(t, s2, key2); err == nil || err.Error() != fmt.Sprintf(dbfNotPicked, "uk") { |
| 149 | t.Fatalf("not picked: %v", err) |
| 150 | } |
| 151 | } |
| 152 | |
| 153 | func TestForwardedQueryMapsEveryCode(t *testing.T) { |
| 154 | t.Parallel() |
| 155 | host := "shop.example.com" |
| 156 | for code, want := range map[string]string{ |
| 157 | dbreads.CodeTooLarge: fmt.Sprintf(dbfTooLarge, host, "3.0GB", "2.0GB", "uk"), |
| 158 | dbreads.CodeNoRoom: fmt.Sprintf(dbfNoRoom, "uk", host), |
| 159 | dbreads.CodeTimeout: dbfTimeout, |
| 160 | dbreads.CodeSQLError: "no such table: nope", |
| 161 | dbreads.CodeNothingReplicated: dbRestoreError(host, dbdoorErrNothing()).Error(), |
| 162 | dbreads.CodeChangedTwice: dbRestoreError(host, errVanished()).Error(), |
| 163 | dbreads.CodeInterrupted: fmt.Sprintf(dbfUnfinished, host, "uk"), |
| 164 | } { |
| 165 | key, _ := dbreads.NewReplyKey() |
| 166 | s := newStream(t, key) |
| 167 | sum := dbreads.Summary{Restored: 3 << 30, Cap: 2 << 30, Error: "no such table: nope"} |
| 168 | s.result(dbreads.StateFailed, code, 0, sum) |
| 169 | if _, _, err := readFwd(t, s, key); err == nil || err.Error() != want { |
| 170 | t.Errorf("%s: %v, want %q", code, err, want) |
| 171 | } |
| 172 | } |
| 173 | for code, want := range map[string]string{ |
| 174 | dbreads.CodeRecordChanged: fmt.Sprintf(dbfChanged, host), |
| 175 | dbreads.CodeMoving: fmt.Sprintf(dbfMovingCLI, host), |
| 176 | } { |
| 177 | key, _ := dbreads.NewReplyKey() |
| 178 | s := newStream(t, key) |
| 179 | s.result(dbreads.StateRefused, code, 0, dbreads.Summary{}) |
| 180 | if _, _, err := readFwd(t, s, key); err == nil || err.Error() != want { |
| 181 | t.Errorf("%s: %v", code, err) |
| 182 | } |
| 183 | } |
| 184 | } |
| 185 | |
| 186 | func TestForwardedQueryOutputCapPrintsTheRows(t *testing.T) { |
| 187 | t.Parallel() |
| 188 | key, _ := dbreads.NewReplyKey() |
| 189 | s := newStream(t, key) |
| 190 | s.chunk(1, true, []byte("cut")) |
| 191 | s.result(dbreads.StateDone, dbreads.CodeOutputCap, 1, summaryOf("cut")) |
| 192 | ans, _, err := readFwd(t, s, key) |
| 193 | if err != nil || !ans.capped || string(ans.rows) != "cut" { |
| 194 | t.Fatalf("%+v %v", ans, err) |
| 195 | } |
| 196 | } |
| 197 | |
| 198 | // Review Focus 2: a full-size chunk line, about 350 KiB, is read. |
| 199 | func TestForwardedQueryReadsAFullSizeChunkLine(t *testing.T) { |
| 200 | t.Parallel() |
| 201 | key, _ := dbreads.NewReplyKey() |
| 202 | rows := strings.Repeat("x", dbreads.ChunkBytes) |
| 203 | s := newStream(t, key) |
| 204 | s.chunk(1, false, []byte(rows)) |
| 205 | s.chunk(2, true, nil) |
| 206 | s.result(dbreads.StateDone, "", 2, summaryOf(rows)) |
| 207 | ans, _, err := readFwd(t, s, key) |
| 208 | if err != nil || len(ans.rows) != dbreads.ChunkBytes { |
| 209 | t.Fatalf("%d bytes, %v", len(ans.rows), err) |
| 210 | } |
| 211 | } |
| 212 | |
| 213 | // Review Focus 1: Ctrl-C mid-stream is "stopped", with nothing printed. |
| 214 | func TestForwardedQueryCtrlCPrintsNothing(t *testing.T) { |
| 215 | t.Parallel() |
| 216 | key, _ := dbreads.NewReplyKey() |
| 217 | s := newStream(t, key) |
| 218 | s.chunk(1, false, []byte("n\n")) |
| 219 | ctx, cancel := context.WithCancel(context.Background()) |
| 220 | cancel() |
| 221 | var status bytes.Buffer |
| 222 | _, _, err := readDBForward(ctx, &cutReader{r: &s.buf}, key, "shop.example.com", "uk", dbfReading, &status, &memSink{}, newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes)) |
| 223 | if err == nil || err.Error() != "stopped" { |
| 224 | t.Fatalf("%v", err) |
| 225 | } |
| 226 | } |
| 227 | |
| 228 | // Plan review finding 37: what the CLI holds before a result is bounded by |
| 229 | // count as well as bytes, and a malformed chunk is refused before it is |
| 230 | // kept. Each stream is otherwise one the CLI would accept, so each case is |
| 231 | // red under its own mutation of dbfChunkBound.admit: drop the size floor |
| 232 | // ("a chunk shorter than its seal" then ends DBF-12, the stream having no |
| 233 | // result, instead of DBF-13 at once); drop the empty-chunk rule or the |
| 234 | // count bound (the stream then opens and prints). |
| 235 | func TestForwardedQueryBoundsChunkCountAndSize(t *testing.T) { |
| 236 | t.Parallel() |
| 237 | key, _ := dbreads.NewReplyKey() |
| 238 | unchecked := fmt.Sprintf(dbfUnchecked, "uk") |
| 239 | many := newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes).maxChunks + 1 |
| 240 | for name, write := range map[string]func(s *stream){ |
| 241 | "a chunk shorter than its seal": func(s *stream) { |
| 242 | s.line(dbreads.Line{Chunk: 1, Data: []byte("short")}) |
| 243 | }, |
| 244 | "a chunk after an empty one": func(s *stream) { |
| 245 | s.chunk(1, false, nil) |
| 246 | s.chunk(2, true, []byte("n\n")) |
| 247 | s.result(dbreads.StateDone, "", 2, summaryOf("n\n")) |
| 248 | }, |
| 249 | "more chunks than a read can make": func(s *stream) { |
| 250 | for seq := int64(1); seq <= many; seq++ { |
| 251 | s.chunk(seq, seq == many, []byte("x")) |
| 252 | } |
| 253 | s.result(dbreads.StateDone, "", many, summaryOf(strings.Repeat("x", int(many)))) |
| 254 | }, |
| 255 | } { |
| 256 | s := newStream(t, key) |
| 257 | write(s) |
| 258 | var status bytes.Buffer |
| 259 | sink := &memSink{} |
| 260 | id, res, err := readDBForward(context.Background(), &s.buf, key, "shop.example.com", "uk", dbfReading, &status, sink, newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes)) |
| 261 | if err == nil { |
| 262 | _, err = openDBForward(id, key, sink, *res, "shop.example.com", "uk") |
| 263 | } |
| 264 | if err == nil || err.Error() != unchecked { |
| 265 | t.Errorf("%s: %v", name, err) |
| 266 | } |
| 267 | } |
| 268 | } |
| 269 | |
| 270 | // Plan review finding 36: SQL that JSON would rewrite is refused before |
| 271 | // anything is sent. Mutation: drop the check (red: the request goes out). |
| 272 | func TestForwardedQueryRefusesSQLThatIsNotUTF8(t *testing.T) { |
| 273 | t.Parallel() |
| 274 | var sent atomic.Bool |
| 275 | ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 276 | sent.Store(true) |
| 277 | http.Error(w, "no", http.StatusTeapot) |
| 278 | })) |
| 279 | defer ts.Close() |
| 280 | target := dbTarget{sess: &apiSession{c: newConsoleClient(ts.URL, "tok"), console: ts.URL}, acct: apiAcct{Seg: "acme"}, |
| 281 | app: "shop", host: "shop.example.com"} |
| 282 | for _, q := range []string{"SELECT '\xff'", "SELECT '�'"} { |
| 283 | err := dbQueryForward(context.Background(), target, q, dbdoor.Forward{Region: "uk", Ops: []string{dbreads.OpQuery}}) |
| 284 | if err == nil || err.Error() != fmt.Sprintf(dbfNotUTF8, "uk") { |
| 285 | t.Errorf("%q: %v", q, err) |
| 286 | } |
| 287 | } |
| 288 | if sent.Load() { |
| 289 | t.Fatal("a query JSON would rewrite was sent") |
| 290 | } |
| 291 | } |
| 292 | |
| 293 | // cutReader ends like a connection the context closed. |
| 294 | type cutReader struct{ r *bytes.Buffer } |
| 295 | |
| 296 | func (c *cutReader) Read(p []byte) (int, error) { |
| 297 | if c.r.Len() == 0 { |
| 298 | return 0, context.Canceled |
| 299 | } |
| 300 | return c.r.Read(p) |
| 301 | } |
| 302 | |
| 303 | // The errors the in-region restore answers with, for the mapping test. |
| 304 | func dbdoorErrNothing() error { return dbdoor.ErrNothingReplicated } |
| 305 | |
| 306 | func errVanished() error { return os.ErrNotExist } |
| 307 | |
| 308 | // pullStream is a pull's body for the files of dir's replica, in one chunk. |
| 309 | func pullStream(t *testing.T, key *ecdh.PrivateKey, dir string, mutate func(s *dbreads.Summary, files [][]byte)) *stream { |
| 310 | t.Helper() |
| 311 | return pullStreamN(t, key, dir, 1, mutate) |
| 312 | } |
| 313 | |
| 314 | // pullStreamN is pullStream with the files' bytes split across n chunks. |
| 315 | func pullStreamN(t *testing.T, key *ecdh.PrivateKey, dir string, n int, mutate func(s *dbreads.Summary, files [][]byte)) *stream { |
| 316 | t.Helper() |
| 317 | rd := lsfile.NewReplicaClient(dir) |
| 318 | plan, err := dbdoor.PlanFrom(context.Background(), rd, time.Time{}) |
| 319 | if err != nil { |
| 320 | t.Fatal(err) |
| 321 | } |
| 322 | var files [][]byte |
| 323 | var all bytes.Buffer |
| 324 | for _, f := range plan.Planned { |
| 325 | rc, err := rd.OpenLTXFile(context.Background(), f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0) |
| 326 | if err != nil { |
| 327 | t.Fatal(err) |
| 328 | } |
| 329 | b, _ := io.ReadAll(rc) |
| 330 | rc.Close() |
| 331 | files = append(files, b) |
| 332 | all.Write(b) |
| 333 | } |
| 334 | sum := sha256.Sum256(all.Bytes()) |
| 335 | s := dbreads.Summary{TXID: plan.TXID, At: plan.At, Files: plan.Planned, Bytes: int64(all.Len()), SHA256: hex.EncodeToString(sum[:])} |
| 336 | if mutate != nil { |
| 337 | mutate(&s, files) |
| 338 | } |
| 339 | st := newStream(t, key) |
| 340 | b := all.Bytes() |
| 341 | per := (len(b) + n - 1) / n |
| 342 | for seq := 1; seq <= n; seq++ { |
| 343 | part := b[min((seq-1)*per, len(b)):min(seq*per, len(b))] |
| 344 | sealed, err := dbreads.SealChunk(key.PublicKey(), fwdID, int64(seq), dbreads.StreamLTX, seq == n, part) |
| 345 | if err != nil { |
| 346 | t.Fatal(err) |
| 347 | } |
| 348 | st.line(dbreads.Line{Chunk: int64(seq), Data: sealed}) |
| 349 | } |
| 350 | st.result(dbreads.StateDone, "", int64(n), s) |
| 351 | return st |
| 352 | } |
| 353 | |
| 354 | func readPull(t *testing.T, st *stream, key *ecdh.PrivateKey) (*dbdoor.Restored, error) { |
| 355 | t.Helper() |
| 356 | return readPullIn(t, st, key, t.TempDir()) |
| 357 | } |
| 358 | |
| 359 | func readPullIn(t *testing.T, st *stream, key *ecdh.PrivateKey, dir string) (*dbdoor.Restored, error) { |
| 360 | t.Helper() |
| 361 | sink := &fileSink{dir: filepath.Join(dir, "sealed")} |
| 362 | id, res, err := readDBForward(context.Background(), &st.buf, key, "shop.example.com", "uk", dbfFetching, io.Discard, sink, newDBFChunkBound(dbreads.MaxPullBytes, dbreads.PullChunkBytes)) |
| 363 | if err != nil { |
| 364 | return nil, err |
| 365 | } |
| 366 | return openDBPull(context.Background(), id, key, sink, *res, "shop.example.com", "uk", dir) |
| 367 | } |
| 368 | |
| 369 | func TestForwardedPullRestoresTheFiles(t *testing.T) { |
| 370 | t.Parallel() |
| 371 | dir := t.TempDir() |
| 372 | dbdoortest.Replicate(t, dir, 3) |
| 373 | for _, n := range []int{1, 4} { |
| 374 | key, _ := dbreads.NewReplyKey() |
| 375 | out := t.TempDir() |
| 376 | r, err := readPullIn(t, pullStreamN(t, key, dir, n, nil), key, out) |
| 377 | if err != nil { |
| 378 | t.Fatalf("%d chunks: %v", n, err) |
| 379 | } |
| 380 | db, _ := sql.Open("sqlite", "file:"+r.Path+"?mode=ro") |
| 381 | var rows int |
| 382 | err = db.QueryRow("SELECT count(*) FROM t").Scan(&rows) |
| 383 | db.Close() |
| 384 | if err != nil || rows != 30 { |
| 385 | t.Fatalf("%d chunks: %d rows, %v", n, rows, err) |
| 386 | } |
| 387 | // Only the copy is left: each spooled chunk went once its bytes were |
| 388 | // in the files (so the two are never both whole on the disk), and |
| 389 | // the files went once the copy was restored. dbPullForward removes |
| 390 | // the empty spool directory. |
| 391 | if left, _ := os.ReadDir(filepath.Join(out, "sealed")); len(left) != 0 { |
| 392 | t.Errorf("%d chunks: %d spooled chunks are still there", n, len(left)) |
| 393 | } |
| 394 | if _, err := os.Stat(filepath.Join(out, "replica")); !os.IsNotExist(err) { |
| 395 | t.Errorf("%d chunks: the files are still there (%v)", n, err) |
| 396 | } |
| 397 | } |
| 398 | } |
| 399 | |
| 400 | // Review Focus 3 and plan review findings 22 and 24: a file list that does |
| 401 | // not add up is DBF-13, never a panic, and no file is created before the |
| 402 | // list is checked whole. Mutations: drop checkPullManifest's per-file size |
| 403 | // bounds (red: "sizes -1 and N+1" panics or creates files); drop its sum |
| 404 | // (red: "sizes that do not sum" creates the first file before failing); |
| 405 | // drop the duplicate check (red: "a file twice" creates a file); drop the |
| 406 | // level bound (red: "level 12"). |
| 407 | func TestForwardedPullRefusesFilesThatDoNotAddUp(t *testing.T) { |
| 408 | t.Parallel() |
| 409 | dir := t.TempDir() |
| 410 | dbdoortest.Replicate(t, dir, 3) |
| 411 | for name, mutate := range map[string]func(s *dbreads.Summary, files [][]byte){ |
| 412 | "sizes that do not sum": func(s *dbreads.Summary, _ [][]byte) { s.Files[0].Size++ }, |
| 413 | "a list that ends elsewhere": func(s *dbreads.Summary, _ [][]byte) { s.TXID++ }, |
| 414 | "no files": func(s *dbreads.Summary, _ [][]byte) { s.Files = nil }, |
| 415 | "sizes -1 and N+1": func(s *dbreads.Summary, _ [][]byte) { |
| 416 | s.Files[1].Size += s.Files[0].Size + 1 |
| 417 | s.Files[0].Size = -1 |
| 418 | }, |
| 419 | "a file twice": func(s *dbreads.Summary, _ [][]byte) { |
| 420 | s.Files[1] = s.Files[0] |
| 421 | s.Bytes = 0 |
| 422 | for _, f := range s.Files { |
| 423 | s.Bytes += f.Size |
| 424 | } |
| 425 | }, |
| 426 | "level 12": func(s *dbreads.Summary, _ [][]byte) { s.Files[0].Level = 12 }, |
| 427 | "a range that runs back": func(s *dbreads.Summary, _ [][]byte) { s.Files[0].MinTXID = s.Files[0].MaxTXID + 1 }, |
| 428 | "a size past the pull's cap": func(s *dbreads.Summary, _ [][]byte) { s.Files[0].Size = dbreads.MaxPullBytes + 1 }, |
| 429 | } { |
| 430 | key, _ := dbreads.NewReplyKey() |
| 431 | out := t.TempDir() |
| 432 | func() { |
| 433 | defer func() { |
| 434 | if p := recover(); p != nil { |
| 435 | t.Errorf("%s: panic %v", name, p) |
| 436 | } |
| 437 | }() |
| 438 | if _, err := readPullIn(t, pullStream(t, key, dir, mutate), key, out); err == nil || err.Error() != fmt.Sprintf(dbfUnchecked, "uk") { |
| 439 | t.Errorf("%s: %v", name, err) |
| 440 | } |
| 441 | }() |
| 442 | if _, err := os.Stat(filepath.Join(out, "replica")); !os.IsNotExist(err) { |
| 443 | t.Errorf("%s: a file was created before the list was checked (%v)", name, err) |
| 444 | } |
| 445 | } |
| 446 | } |
| 447 | |
| 448 | // The other endings of a pull: too-large is DBF-17 with the planned bytes |
| 449 | // and the cap, no-room DBF-24, and the rest the query's mapping. |
| 450 | func TestForwardedPullEndings(t *testing.T) { |
| 451 | t.Parallel() |
| 452 | for name, c := range map[string]struct { |
| 453 | code string |
| 454 | sum dbreads.Summary |
| 455 | want string |
| 456 | }{ |
| 457 | "too large": {dbreads.CodeTooLarge, dbreads.Summary{Restored: 3 << 30, Cap: 2 << 30}, |
| 458 | fmt.Sprintf(dbfPullTooLarge, "shop.example.com", humanBytes(ptr(int64(3<<30))), humanBytes(ptr(int64(2<<30))), "uk")}, |
| 459 | "no room": {dbreads.CodeNoRoom, dbreads.Summary{}, fmt.Sprintf(dbfPullNoRoom, "uk", "shop.example.com")}, |
| 460 | "nothing replicated": {dbreads.CodeNothingReplicated, dbreads.Summary{}, dbRestoreError("shop.example.com", dbdoor.ErrNothingReplicated).Error()}, |
| 461 | } { |
| 462 | key, _ := dbreads.NewReplyKey() |
| 463 | st := newStream(t, key) |
| 464 | st.result(dbreads.StateFailed, c.code, 0, c.sum) |
| 465 | if _, err := readPull(t, st, key); err == nil || err.Error() != c.want { |
| 466 | t.Errorf("%s: %v, want %q", name, err, c.want) |
| 467 | } |
| 468 | } |
| 469 | } |
| 470 | |
| 471 | func ptr[T any](v T) *T { return &v } |
| 472 | |
| 473 | // Plan review finding 37, at the spool: a chunk shorter than a sealed |
| 474 | // empty one creates no spool file, and a stream of tiny chunks stops at the |
| 475 | // bound's count, so the spool never holds more files than a read can make. |
| 476 | // (TestForwardedQueryBoundsChunkCountAndSize pins the count's derivation; a |
| 477 | // bound of 10 keeps this test small.) Mutation: add to the sink before |
| 478 | // admit (red: a file for the short chunk, eleven files for the tiny ones). |
| 479 | func TestForwardedPullSpoolsNoMalformedOrExcessChunk(t *testing.T) { |
| 480 | t.Parallel() |
| 481 | key, _ := dbreads.NewReplyKey() |
| 482 | for name, write := range map[string]func(s *stream) int{ |
| 483 | "a chunk shorter than its seal": func(s *stream) int { |
| 484 | s.line(dbreads.Line{Chunk: 1, Data: []byte("short")}) |
| 485 | return 0 |
| 486 | }, |
| 487 | "tiny chunks past the count": func(s *stream) int { |
| 488 | for seq := int64(1); seq <= 11; seq++ { |
| 489 | sealed, _ := dbreads.SealChunk(key.PublicKey(), fwdID, seq, dbreads.StreamLTX, false, []byte("x")) |
| 490 | s.line(dbreads.Line{Chunk: seq, Data: sealed}) |
| 491 | } |
| 492 | return 10 |
| 493 | }, |
| 494 | } { |
| 495 | s := newStream(t, key) |
| 496 | want := write(s) |
| 497 | dir := filepath.Join(t.TempDir(), "sealed") |
| 498 | sink := &fileSink{dir: dir} |
| 499 | _, _, err := readDBForward(context.Background(), &s.buf, key, "shop.example.com", "uk", dbfFetching, io.Discard, sink, |
| 500 | &dbfChunkBound{maxBytes: dbreads.MaxPullBytes, maxChunks: 10}) |
| 501 | files, _ := os.ReadDir(dir) |
| 502 | if err == nil || err.Error() != fmt.Sprintf(dbfUnchecked, "uk") || len(files) != want { |
| 503 | t.Errorf("%s: %v, %d spool files, want %d", name, err, len(files), want) |
| 504 | } |
| 505 | } |
| 506 | } |
| 507 | |
| 508 | // standInConsole answers the list door with the forward marker offering a |
| 509 | // pull, and the reads route with body (then holds the request open until |
| 510 | // the client goes when hold is set). It is a stand-in for the console, so |
| 511 | // that these tests run dbPull itself. |
| 512 | func standInConsole(t *testing.T, body []byte, hold bool) dbTarget { |
| 513 | t.Helper() |
| 514 | ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 515 | if !strings.HasSuffix(r.URL.Path, "/reads") { |
| 516 | w.Header().Set("Content-Type", "application/json") |
| 517 | w.WriteHeader(http.StatusConflict) |
| 518 | json.NewEncoder(w).Encode(dbdoor.ForwardBody{Error: "shop.example.com keeps its data in region uk.", |
| 519 | Forward: 1, Region: "uk", ForwardOps: []string{dbreads.OpQuery, dbreads.OpPull}}) |
| 520 | return |
| 521 | } |
| 522 | io.Copy(io.Discard, r.Body) |
| 523 | w.Header().Set("Content-Type", "application/x-ndjson") |
| 524 | w.WriteHeader(http.StatusOK) |
| 525 | w.Write(body) |
| 526 | w.(http.Flusher).Flush() |
| 527 | if hold { |
| 528 | <-r.Context().Done() |
| 529 | } |
| 530 | })) |
| 531 | t.Cleanup(ts.Close) |
| 532 | return dbTarget{sess: &apiSession{c: newConsoleClient(ts.URL, "tok"), console: ts.URL}, acct: apiAcct{Seg: "acme"}, |
| 533 | app: "shop", host: "shop.example.com"} |
| 534 | } |
| 535 | |
| 536 | // leftBehind is what a pull left in its temp dir and at --out. |
| 537 | func leftBehind(t *testing.T, tmp, out string) { |
| 538 | t.Helper() |
| 539 | if _, err := os.Stat(out); !os.IsNotExist(err) { |
| 540 | t.Errorf("something is at --out: %v", err) |
| 541 | } |
| 542 | if left, _ := filepath.Glob(filepath.Join(tmp, "carlos-db-*")); len(left) != 0 { |
| 543 | t.Errorf("the pull's directory is still there: %v", left) |
| 544 | } |
| 545 | } |
| 546 | |
| 547 | // Review Focus 1 and plan review finding 23: a spool that fails part way, |
| 548 | // through dbPull itself, leaves nothing at --out and removes its directory. |
| 549 | // Not parallel: it sets TMPDIR and dbPullSpoolWrite. Mutation: drop |
| 550 | // dbPullForward's deferred RemoveAll (red: carlos-db-* is left). |
| 551 | func TestForwardedPullLeavesNothingWhenTheSpoolFails(t *testing.T) { |
| 552 | tmp := t.TempDir() |
| 553 | t.Setenv("TMPDIR", tmp) |
| 554 | writes := 0 |
| 555 | dbPullSpoolWrite = func(name string, b []byte, mode os.FileMode) error { |
| 556 | if writes++; writes == 3 { |
| 557 | return errors.New("no space left on device") |
| 558 | } |
| 559 | return os.WriteFile(name, b, mode) |
| 560 | } |
| 561 | t.Cleanup(func() { dbPullSpoolWrite = os.WriteFile }) |
| 562 | dir := t.TempDir() |
| 563 | dbdoortest.Replicate(t, dir, 3) |
| 564 | // The stand-in's chunks are sealed to a key the CLI does not hold; the |
| 565 | // pull fails at the spool, before anything is opened. |
| 566 | key, _ := dbreads.NewReplyKey() |
| 567 | st := pullStreamN(t, key, dir, 4, nil) |
| 568 | out := filepath.Join(t.TempDir(), "copy.db") |
| 569 | if err := dbPull(context.Background(), standInConsole(t, st.buf.Bytes(), false), out); err == nil { |
| 570 | t.Fatal("a spool that failed was accepted") |
| 571 | } |
| 572 | if writes != 3 { |
| 573 | t.Fatalf("%d spool writes: the failure was not part way", writes) |
| 574 | } |
| 575 | leftBehind(t, tmp, out) |
| 576 | } |
| 577 | |
| 578 | // Review Focus 4 and plan review finding 23: Ctrl-C while chunks are |
| 579 | // spooling, through dbPull itself, is "stopped", with nothing at --out and |
| 580 | // the directory removed. Not parallel: it sets TMPDIR and dbPullSpoolWrite. |
| 581 | // Mutation: drop dbPullForward's deferred RemoveAll (red). The binary's |
| 582 | // SIGINT is Task 9's TestDBForwardPullInterruptedLeavesNothing. |
| 583 | func TestForwardedPullCtrlCLeavesNothing(t *testing.T) { |
| 584 | tmp := t.TempDir() |
| 585 | t.Setenv("TMPDIR", tmp) |
| 586 | ctx, cancel := context.WithCancel(context.Background()) |
| 587 | defer cancel() |
| 588 | dbPullSpoolWrite = func(name string, b []byte, mode os.FileMode) error { |
| 589 | err := os.WriteFile(name, b, mode) |
| 590 | cancel() // Ctrl-C once the first chunk is on disk |
| 591 | return err |
| 592 | } |
| 593 | t.Cleanup(func() { dbPullSpoolWrite = os.WriteFile }) |
| 594 | dir := t.TempDir() |
| 595 | dbdoortest.Replicate(t, dir, 3) |
| 596 | key, _ := dbreads.NewReplyKey() |
| 597 | st := pullStreamN(t, key, dir, 4, nil) |
| 598 | // Admitted, picked and the first chunk only (newStream writes admitted |
| 599 | // and picked first), then the stand-in holds the request. |
| 600 | lines := bytes.SplitAfter(st.buf.Bytes(), []byte("\n")) |
| 601 | out := filepath.Join(t.TempDir(), "copy.db") |
| 602 | err := dbPull(ctx, standInConsole(t, bytes.Join(lines[:3], nil), true), out) |
| 603 | if err == nil || err.Error() != "stopped" { |
| 604 | t.Fatalf("%v", err) |
| 605 | } |
| 606 | leftBehind(t, tmp, out) |
| 607 | } |
| 608 | |
| 609 | // pullStreamOf is a pull's body for the given bytes and summary, one chunk. |
| 610 | func pullStreamOf(t *testing.T, key *ecdh.PrivateKey, b []byte, s dbreads.Summary) *stream { |
| 611 | t.Helper() |
| 612 | st := newStream(t, key) |
| 613 | sealed, err := dbreads.SealChunk(key.PublicKey(), fwdID, 1, dbreads.StreamLTX, true, b) |
| 614 | if err != nil { |
| 615 | t.Fatal(err) |
| 616 | } |
| 617 | st.line(dbreads.Line{Chunk: 1, Data: sealed}) |
| 618 | st.result(dbreads.StateDone, "", 1, s) |
| 619 | return st |
| 620 | } |
| 621 | |
| 622 | // Astra round 1 on plan 3, finding 1, at the CLI: a valid snapshot of |
| 623 | // transaction 1, listed as [1,2] with a summary at 2 that adds up in every |
| 624 | // other way, is DBF-13 and restores nothing: the CLI checks each rebuilt |
| 625 | // file's LTX header against its manifest entry. Mutation: drop the header |
| 626 | // comparison (red: it restores and reports transaction 2). |
| 627 | func TestForwardedPullRefusesARelabelledSnapshot(t *testing.T) { |
| 628 | t.Parallel() |
| 629 | dir := t.TempDir() |
| 630 | dbdoortest.Replicate(t, dir, 2) |
| 631 | rc, err := lsfile.NewReplicaClient(dir).OpenLTXFile(context.Background(), 0, 1, 1, 0, 0) |
| 632 | if err != nil { |
| 633 | t.Fatal(err) |
| 634 | } |
| 635 | b, _ := io.ReadAll(rc) |
| 636 | rc.Close() |
| 637 | sum := sha256.Sum256(b) |
| 638 | s := dbreads.Summary{TXID: 2, Files: []dbdoor.LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 2, Size: int64(len(b))}}, |
| 639 | Bytes: int64(len(b)), SHA256: hex.EncodeToString(sum[:])} |
| 640 | key, _ := dbreads.NewReplyKey() |
| 641 | out := t.TempDir() |
| 642 | r, err := readPullIn(t, pullStreamOf(t, key, b, s), key, out) |
| 643 | if err == nil || err.Error() != fmt.Sprintf(dbfUnchecked, "uk") { |
| 644 | t.Fatalf("a relabelled snapshot gave %+v, %v", r, err) |
| 645 | } |
| 646 | if _, err := os.Stat(filepath.Join(out, "replica.db")); !os.IsNotExist(err) { |
| 647 | t.Fatalf("a copy was restored: %v", err) |
| 648 | } |
| 649 | } |
| 650 | |
| 651 | // Astra round 1 on plan 3, finding 2, after the download: a pull cancelled |
| 652 | // once every chunk has arrived is "stopped" and rebuilds no file. Mutation: |
| 653 | // drop the rebuild loop's context check (red: the files are written and |
| 654 | // restored before the cancel is seen). |
| 655 | func TestForwardedPullStopsWhenCancelledAfterTheDownload(t *testing.T) { |
| 656 | t.Parallel() |
| 657 | dir := t.TempDir() |
| 658 | dbdoortest.Replicate(t, dir, 3) |
| 659 | key, _ := dbreads.NewReplyKey() |
| 660 | st := pullStreamN(t, key, dir, 4, nil) |
| 661 | out := t.TempDir() |
| 662 | sink := &fileSink{dir: filepath.Join(out, "sealed")} |
| 663 | id, res, err := readDBForward(context.Background(), &st.buf, key, "shop.example.com", "uk", dbfFetching, io.Discard, sink, newDBFChunkBound(dbreads.MaxPullBytes, dbreads.PullChunkBytes)) |
| 664 | if err != nil { |
| 665 | t.Fatal(err) |
| 666 | } |
| 667 | ctx, cancel := context.WithCancel(context.Background()) |
| 668 | cancel() |
| 669 | if _, err := openDBPull(ctx, id, key, sink, *res, "shop.example.com", "uk", out); err == nil || err.Error() != "stopped" { |
| 670 | t.Fatalf("a cancelled pull answered %v", err) |
| 671 | } |
| 672 | for _, gone := range []string{"replica", "replica.db"} { |
| 673 | if _, err := os.Stat(filepath.Join(out, gone)); !os.IsNotExist(err) { |
| 674 | t.Errorf("%s was written after the cancel (%v)", gone, err) |
| 675 | } |
| 676 | } |
| 677 | } |
| 678 | |
amadan