| 1 | package dbreads |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "fmt" |
| 8 | "io" |
| 9 | "strings" |
| 10 | "sync" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | lsfile "github.com/benbjohnson/litestream/file" |
| 15 | "github.com/superfly/ltx" |
| 16 | |
| 17 | "github.com/carlosframework/platform/internal/dbdoor" |
| 18 | "github.com/carlosframework/platform/internal/dbdoor/dbdoortest" |
| 19 | ) |
| 20 | |
| 21 | // sendPullPlan sends a pull header and the planned files; it returns the |
| 22 | // files' bytes concatenated, which is what the worker must send back. |
| 23 | // net.Pipe to the real worker code: not acceptance (Task 5's root-runner |
| 24 | // test is). |
| 25 | func sendPullPlan(t *testing.T, w io.Writer, dir string, corrupt bool) (dbdoor.RestorePlan, []byte) { |
| 26 | t.Helper() |
| 27 | rd := lsfile.NewReplicaClient(dir) |
| 28 | plan, err := dbdoor.PlanFrom(context.Background(), rd, time.Time{}) |
| 29 | if err != nil { |
| 30 | t.Fatal(err) |
| 31 | } |
| 32 | if err := writeJSONFrame(w, frameHeader, planHeader{V: ProtocolVersion, Op: OpPull, Files: plan.Planned, Target: plan.TXID, MaxRestored: DefaultMaxRestored}); err != nil { |
| 33 | t.Fatal(err) |
| 34 | } |
| 35 | var all bytes.Buffer |
| 36 | for i, f := range plan.Planned { |
| 37 | rc, err := rd.OpenLTXFile(context.Background(), f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0) |
| 38 | if err != nil { |
| 39 | t.Fatal(err) |
| 40 | } |
| 41 | b, _ := io.ReadAll(rc) |
| 42 | rc.Close() |
| 43 | if corrupt && i == 0 { |
| 44 | b[len(b)/2] ^= 0xff |
| 45 | } |
| 46 | all.Write(b) |
| 47 | if err := writeFrame(w, frameFile, b); err != nil { |
| 48 | t.Fatal(err) |
| 49 | } |
| 50 | } |
| 51 | return plan, all.Bytes() |
| 52 | } |
| 53 | |
| 54 | func TestWorkerHoldsThenSendsThePlannedFiles(t *testing.T) { |
| 55 | t.Parallel() |
| 56 | dir := t.TempDir() |
| 57 | dbdoortest.Replicate(t, dir, 3) |
| 58 | conn, done := workerPipe(t, WorkerOptions{}) |
| 59 | plan, want := sendPullPlan(t, conn, dir, false) |
| 60 | var ready readyMsg |
| 61 | readTyped(t, conn, frameReady, &ready) |
| 62 | if ready.TXID != plan.TXID { |
| 63 | t.Fatalf("ready at %d", ready.TXID) |
| 64 | } |
| 65 | if err := writeJSONFrame(conn, frameRun, runMsg{}); err != nil { |
| 66 | t.Fatal(err) |
| 67 | } |
| 68 | var got bytes.Buffer |
| 69 | for { |
| 70 | typ, p, err := readFrame(conn) |
| 71 | if err != nil { |
| 72 | t.Fatal(err) |
| 73 | } |
| 74 | if typ == frameOutput { |
| 75 | got.Write(p) |
| 76 | continue |
| 77 | } |
| 78 | var end endMsg |
| 79 | if typ != frameEnd || json.Unmarshal(p, &end) != nil || end.Code != "" { |
| 80 | t.Fatalf("end %q %s", typ, p) |
| 81 | } |
| 82 | break |
| 83 | } |
| 84 | if !bytes.Equal(got.Bytes(), want) { |
| 85 | t.Fatalf("sent back %d bytes, held %d", got.Len(), len(want)) |
| 86 | } |
| 87 | if err := <-done; err != nil { |
| 88 | t.Fatal(err) |
| 89 | } |
| 90 | } |
| 91 | |
| 92 | // Review Focus 2: a corrupt file is refused before ready, never shipped. |
| 93 | // Mutation: holdAndSend skips verifyLTX (red: ready, not check-failed). |
| 94 | func TestWorkerHoldRefusesACorruptFile(t *testing.T) { |
| 95 | t.Parallel() |
| 96 | dir := t.TempDir() |
| 97 | dbdoortest.Replicate(t, dir, 2) |
| 98 | conn, done := workerPipe(t, WorkerOptions{}) |
| 99 | sendPullPlan(t, conn, dir, true) |
| 100 | var end endMsg |
| 101 | readTyped(t, conn, frameEnd, &end) |
| 102 | if end.Code != CodeCheckFailed { |
| 103 | t.Fatalf("end = %+v", end) |
| 104 | } |
| 105 | <-done |
| 106 | } |
| 107 | |
| 108 | func TestWorkerHoldStopsWithoutSending(t *testing.T) { |
| 109 | t.Parallel() |
| 110 | dir := t.TempDir() |
| 111 | dbdoortest.Replicate(t, dir, 1) |
| 112 | conn, done := workerPipe(t, WorkerOptions{}) |
| 113 | sendPullPlan(t, conn, dir, false) |
| 114 | readTyped(t, conn, frameReady, nil) |
| 115 | if err := writeFrame(conn, frameStop, nil); err != nil { |
| 116 | t.Fatal(err) |
| 117 | } |
| 118 | if err := <-done; err != nil { |
| 119 | t.Fatal(err) |
| 120 | } |
| 121 | } |
| 122 | |
| 123 | func (fx *fixture) pullRequest(at string) Request { |
| 124 | req := fx.request("", at) |
| 125 | req.Op = OpPull |
| 126 | return req |
| 127 | } |
| 128 | |
| 129 | // answerPull opens a pull's chunks as the CLI does: the ltx stream. |
| 130 | func (fx *fixture) answerPull(id string) ([]byte, Summary, Result) { |
| 131 | fx.t.Helper() |
| 132 | raw, _, err := fx.store.Get(context.Background(), ResultKey(id)) |
| 133 | if err != nil { |
| 134 | fx.t.Fatal(err) |
| 135 | } |
| 136 | var r Result |
| 137 | if err := json.Unmarshal(raw, &r); err != nil { |
| 138 | fx.t.Fatal(err) |
| 139 | } |
| 140 | s, err := OpenSummary(fx.reply, id, r.Sealed) |
| 141 | if err != nil { |
| 142 | fx.t.Fatal(err) |
| 143 | } |
| 144 | var out bytes.Buffer |
| 145 | for seq := int64(1); seq <= r.LastSeq; seq++ { |
| 146 | sealed, _, _ := fx.store.Get(context.Background(), ChunkKey(id, seq)) |
| 147 | plain, err := OpenChunk(fx.reply, id, seq, StreamLTX, seq == r.LastSeq, sealed) |
| 148 | if err != nil { |
| 149 | fx.t.Fatalf("chunk %d: %v", seq, err) |
| 150 | } |
| 151 | out.Write(plain) |
| 152 | } |
| 153 | return out.Bytes(), s, r |
| 154 | } |
| 155 | |
| 156 | // plannedBytes is dir's planned files, concatenated in plan order. |
| 157 | func plannedBytes(t *testing.T, dir string) (dbdoor.RestorePlan, []byte) { |
| 158 | t.Helper() |
| 159 | rd := lsfile.NewReplicaClient(dir) |
| 160 | plan, err := dbdoor.PlanFrom(context.Background(), rd, time.Time{}) |
| 161 | if err != nil { |
| 162 | t.Fatal(err) |
| 163 | } |
| 164 | var want bytes.Buffer |
| 165 | for _, f := range plan.Planned { |
| 166 | rc, err := rd.OpenLTXFile(context.Background(), f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0) |
| 167 | if err != nil { |
| 168 | t.Fatal(err) |
| 169 | } |
| 170 | io.Copy(&want, rc) |
| 171 | rc.Close() |
| 172 | } |
| 173 | return plan, want.Bytes() |
| 174 | } |
| 175 | |
| 176 | // The agent's pull ships the planned files whole, in plan order, as the |
| 177 | // ltx stream in chunks of at most PullChunkBytes, with the file list in the |
| 178 | // summary, and logs plan 1's two timing lines (plan review finding 26). |
| 179 | // Mutation: drop the timing lines (red). |
| 180 | func TestPullShipsThePlannedFiles(t *testing.T) { |
| 181 | t.Parallel() |
| 182 | fx := newFixture(t, 4) |
| 183 | var logMu sync.Mutex |
| 184 | var logs []string |
| 185 | a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) { |
| 186 | c.Logf = func(format string, args ...any) { |
| 187 | logMu.Lock() |
| 188 | defer logMu.Unlock() |
| 189 | logs = append(logs, fmt.Sprintf(format, args...)) |
| 190 | } |
| 191 | }) |
| 192 | req := fx.pullRequest("") |
| 193 | fx.submit(req) |
| 194 | a.Pass(context.Background()) |
| 195 | a.Wait() |
| 196 | got, s, r := fx.answerPull(req.ID) |
| 197 | plan, want := plannedBytes(t, fx.replica) |
| 198 | if r.State != StateDone || !bytes.Equal(got, want) || s.TXID != plan.TXID || len(s.Files) != len(plan.Planned) || s.Bytes != plan.LTXBytes { |
| 199 | t.Fatalf("result %+v summary %+v: %d bytes, want %d", r, s, len(got), len(want)) |
| 200 | } |
| 201 | for i, f := range s.Files { |
| 202 | if f.Level != plan.Planned[i].Level || f.MinTXID != plan.Planned[i].MinTXID || f.MaxTXID != plan.Planned[i].MaxTXID || f.Size != plan.Planned[i].Size { |
| 203 | t.Fatalf("file %d is %+v, planned %+v", i, f, plan.Planned[i]) |
| 204 | } |
| 205 | } |
| 206 | for seq := int64(1); seq < r.LastSeq; seq++ { |
| 207 | sealed, _, _ := fx.store.Get(context.Background(), ChunkKey(req.ID, seq)) |
| 208 | if len(sealed)-SealOverhead > PullChunkBytes { |
| 209 | t.Fatalf("chunk %d is %d bytes", seq, len(sealed)) |
| 210 | } |
| 211 | } |
| 212 | // Task 1's measurement reads these two lines (plan review finding 26). |
| 213 | logMu.Lock() |
| 214 | all := strings.Join(logs, "\n") |
| 215 | logMu.Unlock() |
| 216 | if !strings.Contains(all, "and held through") || !strings.Contains(all, "confirmed chunks in") { |
| 217 | t.Fatalf("the pull's timing lines are missing:\n%s", all) |
| 218 | } |
| 219 | } |
| 220 | |
| 221 | // A pull over the agent's cap ends too-large with the planned bytes and the |
| 222 | // cap, and a record change after the last byte left the bucket (T9 for a |
| 223 | // pull) writes no chunk and never sends the worker run. Mutation: the |
| 224 | // agent's pull sends run before check B (red: the recorded frames hold a |
| 225 | // run). |
| 226 | func TestPullTooLargeAndNoChunkBeforeCheckB(t *testing.T) { |
| 227 | t.Parallel() |
| 228 | fx := newFixture(t, 2) |
| 229 | a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) { c.MaxPull = 100 }) |
| 230 | req := fx.pullRequest("") |
| 231 | fx.submit(req) |
| 232 | a.Pass(context.Background()) |
| 233 | a.Wait() |
| 234 | _, s, r := fx.answerPull(req.ID) |
| 235 | if r.Code != CodeTooLarge || s.Cap != 100 || s.Restored <= 100 { |
| 236 | t.Fatalf("%+v %+v", r, s) |
| 237 | } |
| 238 | fx2 := newFixture(t, 2) |
| 239 | plan, _ := plannedBytes(t, fx2.replica) |
| 240 | last := plan.Planned[len(plan.Planned)-1] |
| 241 | hook := &stageHook{ReplicaReader: lsfile.NewReplicaClient(fx2.replica), stage: "after the last byte", |
| 242 | lastSpan: dbdoor.Span(last.MinTXID, last.MaxTXID), change: func() { r := fx2.record(); r.Region = "ie"; fx2.putRecord(r) }} |
| 243 | b, w := fx2.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) { |
| 244 | c.OpenReplica = func(context.Context, string) (dbdoor.ReplicaReader, error) { return hook, nil } |
| 245 | }) |
| 246 | req2 := fx2.pullRequest("") |
| 247 | fx2.submit(req2) |
| 248 | b.Pass(context.Background()) |
| 249 | b.Wait() |
| 250 | if _, _, r := fx2.answerPull(req2.ID); r.Code != CodeRecordChanged || len(fx2.chunkKeys(req2.ID)) != 0 { |
| 251 | t.Fatalf("%+v, chunks %v", r, fx2.chunkKeys(req2.ID)) |
| 252 | } |
| 253 | // The worker never heard run: check B failed, so the agent said stop |
| 254 | // (plan review finding 24). Without this, an agent that sent run first |
| 255 | // and refused the files afterwards would pass the two checks above. |
| 256 | sent := w.sentFrames() |
| 257 | if !bytes.Contains(sent, []byte{frameStop}) { |
| 258 | t.Fatalf("the agent never said stop: %q", sent) |
| 259 | } |
| 260 | if bytes.Contains(sent, []byte{frameRun}) { |
| 261 | t.Fatalf("the agent sent run before check B passed: %q", sent) |
| 262 | } |
| 263 | } |
| 264 | |
| 265 | // Plan review finding 24: NewAgent itself clamps the pull's cap, whatever a |
| 266 | // box's settings hand it. Mutation: NewAgent keeps a MaxPull above |
| 267 | // MaxPullBytes (red). |
| 268 | func TestNewAgentClampsMaxPull(t *testing.T) { |
| 269 | t.Parallel() |
| 270 | fx := newFixture(t, 1) |
| 271 | for give, want := range map[int64]int64{0: MaxPullBytes, -1: MaxPullBytes, 1 << 20: 1 << 20, MaxPullBytes + 1: MaxPullBytes} { |
| 272 | a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) { c.MaxPull = give }) |
| 273 | if a.cfg.MaxPull != want { |
| 274 | t.Errorf("MaxPull %d became %d, want %d", give, a.cfg.MaxPull, want) |
| 275 | } |
| 276 | } |
| 277 | // Never more than the worker holds: PlanBytesFactor times MaxRestored. |
| 278 | a, _ := fx.agent("uk-1", func(c *AgentConfig, _ *inProcWorkers) { c.MaxRestored = 256 << 20 }) |
| 279 | if a.cfg.MaxPull != PlanBytesFactor*(256<<20) { |
| 280 | t.Errorf("MaxPull %d with a 256 MiB restored cap", a.cfg.MaxPull) |
| 281 | } |
| 282 | } |
| 283 | |
| 284 | // relabelled is the bytes of dir's first planned file, a valid snapshot at |
| 285 | // transaction 1, under the coordinates of a file that ends at 2. |
| 286 | func relabelled(t *testing.T, dir string) (dbdoor.LTXFile, []byte) { |
| 287 | t.Helper() |
| 288 | rd := lsfile.NewReplicaClient(dir) |
| 289 | rc, err := rd.OpenLTXFile(context.Background(), 0, 1, 1, 0, 0) |
| 290 | if err != nil { |
| 291 | t.Fatal(err) |
| 292 | } |
| 293 | b, _ := io.ReadAll(rc) |
| 294 | rc.Close() |
| 295 | return dbdoor.LTXFile{Level: 0, MinTXID: 1, MaxTXID: 2, Size: int64(len(b))}, b |
| 296 | } |
| 297 | |
| 298 | // Astra round 1 on plan 3, finding 1: a valid snapshot under another file's |
| 299 | // coordinates is refused before ready, for a pull and for a query alike: the |
| 300 | // worker checks each file's LTX header against its plan entry, so a copy |
| 301 | // can never be reported at a transaction it does not hold. Mutation: drop |
| 302 | // the header comparison (red: the query restores and answers ready at 2; |
| 303 | // the pull verifies and answers ready). |
| 304 | func TestWorkerRefusesARelabelledSnapshot(t *testing.T) { |
| 305 | t.Parallel() |
| 306 | dir := t.TempDir() |
| 307 | dbdoortest.Replicate(t, dir, 2) |
| 308 | for _, op := range []string{OpQuery, OpPull} { |
| 309 | lf, b := relabelled(t, dir) |
| 310 | conn, done := workerPipe(t, WorkerOptions{}) |
| 311 | if err := writeJSONFrame(conn, frameHeader, planHeader{V: ProtocolVersion, Op: op, Files: []dbdoor.LTXFile{lf}, Target: 2, MaxRestored: DefaultMaxRestored}); err != nil { |
| 312 | t.Fatal(err) |
| 313 | } |
| 314 | if err := writeFrame(conn, frameFile, b); err != nil { |
| 315 | t.Fatal(err) |
| 316 | } |
| 317 | typ, p, err := readFrame(conn) |
| 318 | var end endMsg |
| 319 | if err != nil || typ != frameEnd || json.Unmarshal(p, &end) != nil || end.Code != CodeCheckFailed { |
| 320 | t.Errorf("%s: a relabelled snapshot answered %q %s %v", op, typ, p, err) |
| 321 | writeFrame(conn, frameStop, nil) |
| 322 | } |
| 323 | <-done |
| 324 | } |
| 325 | } |
| 326 | |
amadan