| 1 | package dbreads |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "io" |
| 8 | "net" |
| 9 | "os" |
| 10 | "strings" |
| 11 | "sync/atomic" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/benbjohnson/litestream" |
| 16 | lsfile "github.com/benbjohnson/litestream/file" |
| 17 | "github.com/superfly/ltx" |
| 18 | |
| 19 | "github.com/carlosframework/platform/internal/dbdoor" |
| 20 | "github.com/carlosframework/platform/internal/dbdoor/dbdoortest" |
| 21 | "github.com/carlosframework/platform/internal/dbquery" |
| 22 | ) |
| 23 | |
| 24 | // workerPipe starts ServeWorker on one end of a pipe and returns the other. |
| 25 | // Not acceptance: the units are Task 15's. |
| 26 | func workerPipe(t *testing.T, opt WorkerOptions) (net.Conn, <-chan error) { |
| 27 | t.Helper() |
| 28 | agent, worker := net.Pipe() |
| 29 | if opt.Scratch == "" { |
| 30 | opt.Scratch = t.TempDir() |
| 31 | } |
| 32 | if opt.Exit == nil { |
| 33 | opt.Exit = func(int) { t.Error("the worker asked to exit the process") } |
| 34 | } |
| 35 | done := make(chan error, 1) |
| 36 | go func() { |
| 37 | err := ServeWorker(worker, opt) |
| 38 | worker.Close() |
| 39 | done <- err |
| 40 | }() |
| 41 | t.Cleanup(func() { agent.Close() }) |
| 42 | return agent, done |
| 43 | } |
| 44 | |
| 45 | // sendPlan sends a header and the planned files of dir's replica. |
| 46 | func sendPlan(t *testing.T, conn net.Conn, dir string) dbdoor.RestorePlan { |
| 47 | t.Helper() |
| 48 | rd := lsfile.NewReplicaClient(dir) |
| 49 | plan, err := dbdoor.PlanFrom(context.Background(), rd, time.Time{}) |
| 50 | if err != nil { |
| 51 | t.Fatal(err) |
| 52 | } |
| 53 | if err := writeJSONFrame(conn, frameHeader, planHeader{V: ProtocolVersion, Op: OpQuery, Files: plan.Planned, Target: plan.TXID, MaxRestored: DefaultMaxRestored}); err != nil { |
| 54 | t.Fatal(err) |
| 55 | } |
| 56 | for _, f := range plan.Planned { |
| 57 | rc, err := rd.OpenLTXFile(context.Background(), f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0) |
| 58 | if err != nil { |
| 59 | t.Fatal(err) |
| 60 | } |
| 61 | body, _ := io.ReadAll(rc) |
| 62 | rc.Close() |
| 63 | if err := writeFrame(conn, frameFile, body); err != nil { |
| 64 | t.Fatal(err) |
| 65 | } |
| 66 | } |
| 67 | return plan |
| 68 | } |
| 69 | |
| 70 | func readTyped(t *testing.T, conn net.Conn, want byte, v any) { |
| 71 | t.Helper() |
| 72 | typ, p, err := readFrame(conn) |
| 73 | if err != nil || typ != want { |
| 74 | t.Fatalf("frame %q (%s), want %q: %v", typ, p, want, err) |
| 75 | } |
| 76 | if v != nil { |
| 77 | if err := json.Unmarshal(p, v); err != nil { |
| 78 | t.Fatal(err) |
| 79 | } |
| 80 | } |
| 81 | } |
| 82 | |
| 83 | func TestWorkerAnswersHelloWithItsSelfView(t *testing.T) { |
| 84 | t.Parallel() |
| 85 | conn, done := workerPipe(t, WorkerOptions{Self: func() HelloReply { |
| 86 | return HelloReply{V: ProtocolVersion, Interfaces: []string{"lo"}} |
| 87 | }}) |
| 88 | if err := writeJSONFrame(conn, frameHello, hello{V: ProtocolVersion}); err != nil { |
| 89 | t.Fatal(err) |
| 90 | } |
| 91 | var h HelloReply |
| 92 | readTyped(t, conn, frameHelloReply, &h) |
| 93 | if h.V != ProtocolVersion || len(h.Interfaces) != 1 { |
| 94 | t.Fatalf("hello = %+v", h) |
| 95 | } |
| 96 | if err := <-done; err != nil { |
| 97 | t.Fatal(err) |
| 98 | } |
| 99 | } |
| 100 | |
| 101 | // The default self-view reads /proc/self/net/dev, which needs no socket |
| 102 | // (plan review finding 1). Mutation: SelfView back on net.Interfaces makes |
| 103 | // TestRootDBReadReadiness's baseline red inside the installed unit. |
| 104 | func TestParseNetDev(t *testing.T) { |
| 105 | t.Parallel() |
| 106 | only := "Inter-| Receive\n face |bytes\n lo: 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0\n" |
| 107 | if got := parseNetDev(only); len(got) != 1 || got[0] != "lo" { |
| 108 | t.Fatalf("%q", got) |
| 109 | } |
| 110 | two := only + " eth0: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16\n" |
| 111 | if got := parseNetDev(two); len(got) != 2 || got[1] != "eth0" { |
| 112 | t.Fatalf("%q", got) |
| 113 | } |
| 114 | if got := SelfView(); len(got.Errors) != 0 && !strings.Contains(strings.Join(got.Errors, " "), "/proc") { |
| 115 | t.Fatalf("SelfView on this machine: %+v", got) |
| 116 | } |
| 117 | } |
| 118 | |
| 119 | func TestWorkerRestoresThenRunsOnlyWhenTold(t *testing.T) { |
| 120 | t.Parallel() |
| 121 | dir := t.TempDir() |
| 122 | dbdoortest.Replicate(t, dir, 3) |
| 123 | conn, done := workerPipe(t, WorkerOptions{}) |
| 124 | plan := sendPlan(t, conn, dir) |
| 125 | var ready readyMsg |
| 126 | readTyped(t, conn, frameReady, &ready) |
| 127 | if ready.TXID != plan.TXID { |
| 128 | t.Fatalf("ready at %d, planned %d", ready.TXID, plan.TXID) |
| 129 | } |
| 130 | if err := writeJSONFrame(conn, frameRun, runMsg{SQL: "SELECT count(*) AS n FROM t"}); err != nil { |
| 131 | t.Fatal(err) |
| 132 | } |
| 133 | var out bytes.Buffer |
| 134 | for { |
| 135 | typ, p, err := readFrame(conn) |
| 136 | if err != nil { |
| 137 | t.Fatal(err) |
| 138 | } |
| 139 | if typ == frameOutput { |
| 140 | out.Write(p) |
| 141 | continue |
| 142 | } |
| 143 | var end endMsg |
| 144 | if typ != frameEnd || json.Unmarshal(p, &end) != nil || end.Code != "" { |
| 145 | t.Fatalf("end %q %s", typ, p) |
| 146 | } |
| 147 | break |
| 148 | } |
| 149 | if out.String() != "n\n30\n" { |
| 150 | t.Fatalf("output %q", out.String()) |
| 151 | } |
| 152 | if err := <-done; err != nil { |
| 153 | t.Fatal(err) |
| 154 | } |
| 155 | } |
| 156 | |
| 157 | func TestWorkerStopRunsNothing(t *testing.T) { |
| 158 | t.Parallel() |
| 159 | dir := t.TempDir() |
| 160 | dbdoortest.Replicate(t, dir, 1) |
| 161 | conn, done := workerPipe(t, WorkerOptions{}) |
| 162 | sendPlan(t, conn, dir) |
| 163 | readTyped(t, conn, frameReady, nil) |
| 164 | if err := writeFrame(conn, frameStop, nil); err != nil { |
| 165 | t.Fatal(err) |
| 166 | } |
| 167 | if err := <-done; err != nil { |
| 168 | t.Fatal(err) |
| 169 | } |
| 170 | } |
| 171 | |
| 172 | // The worker refuses a file larger than its header said (spec §8.3). |
| 173 | func TestWorkerRefusesAFileLargerThanItsHeader(t *testing.T) { |
| 174 | t.Parallel() |
| 175 | conn, done := workerPipe(t, WorkerOptions{}) |
| 176 | files := []dbdoor.LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 1, Size: 4}} |
| 177 | if err := writeJSONFrame(conn, frameHeader, planHeader{V: ProtocolVersion, Op: OpQuery, Files: files, Target: 1, MaxRestored: DefaultMaxRestored}); err != nil { |
| 178 | t.Fatal(err) |
| 179 | } |
| 180 | go writeFrame(conn, frameFile, []byte("12345")) |
| 181 | var end endMsg |
| 182 | readTyped(t, conn, frameEnd, &end) |
| 183 | if end.Code != CodeCheckFailed { |
| 184 | t.Fatalf("end = %+v", end) |
| 185 | } |
| 186 | <-done |
| 187 | } |
| 188 | |
| 189 | func TestWorkerSQLErrorIsSQLitesText(t *testing.T) { |
| 190 | t.Parallel() |
| 191 | dir := t.TempDir() |
| 192 | dbdoortest.Replicate(t, dir, 1) |
| 193 | conn, done := workerPipe(t, WorkerOptions{}) |
| 194 | sendPlan(t, conn, dir) |
| 195 | readTyped(t, conn, frameReady, nil) |
| 196 | writeJSONFrame(conn, frameRun, runMsg{SQL: "SELECT * FROM nope"}) |
| 197 | var end endMsg |
| 198 | readTyped(t, conn, frameEnd, &end) |
| 199 | if end.Code != CodeSQLError || end.Error == "" { |
| 200 | t.Fatalf("end = %+v", end) |
| 201 | } |
| 202 | if err := dbquery.CheckSQL("SELECT * FROM nope"); err != nil { |
| 203 | t.Fatal(err) |
| 204 | } |
| 205 | <-done |
| 206 | } |
| 207 | |
| 208 | func TestWorkerQueryTimeout(t *testing.T) { |
| 209 | t.Parallel() |
| 210 | dir := t.TempDir() |
| 211 | dbdoortest.Replicate(t, dir, 1) |
| 212 | conn, done := workerPipe(t, WorkerOptions{QueryTimeout: 50 * time.Millisecond}) |
| 213 | sendPlan(t, conn, dir) |
| 214 | readTyped(t, conn, frameReady, nil) |
| 215 | writeJSONFrame(conn, frameRun, runMsg{SQL: "WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x+1 FROM c) SELECT count(*) FROM c"}) |
| 216 | for { |
| 217 | typ, p, err := readFrame(conn) |
| 218 | if err != nil { |
| 219 | t.Fatal(err) |
| 220 | } |
| 221 | if typ == frameEnd { |
| 222 | var end endMsg |
| 223 | json.Unmarshal(p, &end) |
| 224 | if end.Code != CodeTimeout { |
| 225 | t.Fatalf("end = %+v", end) |
| 226 | } |
| 227 | break |
| 228 | } |
| 229 | } |
| 230 | <-done |
| 231 | } |
| 232 | |
| 233 | // The worker stops when its agent goes (spec §8.3, codex 8): EOF cancels the |
| 234 | // context, and a step that ignores the context is ended by the exit after |
| 235 | // the grace. |
| 236 | func TestWorkerExitsWhenTheAgentGoesMidRestore(t *testing.T) { |
| 237 | t.Parallel() |
| 238 | dir := t.TempDir() |
| 239 | dbdoortest.Replicate(t, dir, 1) |
| 240 | var exited atomic.Int32 |
| 241 | hang := make(chan struct{}) |
| 242 | t.Cleanup(func() { close(hang) }) |
| 243 | conn, _ := workerPipe(t, WorkerOptions{ |
| 244 | Grace: 50 * time.Millisecond, |
| 245 | Exit: func(code int) { exited.Store(int32(code)) }, |
| 246 | Restore: func(context.Context, litestream.ReplicaClient, string, uint64) error { |
| 247 | <-hang // a compaction that ignores its context |
| 248 | return nil |
| 249 | }, |
| 250 | }) |
| 251 | sendPlan(t, conn, dir) |
| 252 | conn.Close() |
| 253 | deadline := time.Now().Add(5 * time.Second) |
| 254 | for exited.Load() == 0 && time.Now().Before(deadline) { |
| 255 | time.Sleep(10 * time.Millisecond) |
| 256 | } |
| 257 | if exited.Load() == 0 { |
| 258 | t.Fatal("the worker outlived its agent past the grace") |
| 259 | } |
| 260 | } |
| 261 | |
| 262 | func TestWorkerCheckFailsAMissingFile(t *testing.T) { |
| 263 | t.Parallel() |
| 264 | conn, done := workerPipe(t, WorkerOptions{}) |
| 265 | if err := writeJSONFrame(conn, frameHeader, planHeader{V: ProtocolVersion, Op: OpQuery, Files: nil, Target: 9, MaxRestored: DefaultMaxRestored}); err != nil { |
| 266 | t.Fatal(err) |
| 267 | } |
| 268 | var end endMsg |
| 269 | readTyped(t, conn, frameEnd, &end) |
| 270 | if end.Code != CodeCheckFailed { |
| 271 | t.Fatalf("an empty plan = %+v", end) |
| 272 | } |
| 273 | <-done |
| 274 | } |
| 275 | |
| 276 | // The worker bounds the bytes it will write across ALL files, not just each |
| 277 | // file (Astra round 1, finding 1): a header whose files sum past |
| 278 | // PlanBytesFactor times its restored-size cap, or whose cap is above the |
| 279 | // worker's own, is refused before anything is written. LimitFSIZE bounds |
| 280 | // one file only. |
| 281 | func TestWorkerRefusesAPlanOverItsTotal(t *testing.T) { |
| 282 | t.Parallel() |
| 283 | for name, h := range map[string]planHeader{ |
| 284 | "files past twice the cap": {V: ProtocolVersion, Op: OpQuery, Target: 3, MaxRestored: 1000, |
| 285 | Files: []dbdoor.LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 1, Size: 800}, {Level: 0, MinTXID: 2, MaxTXID: 2, Size: 800}, {Level: 0, MinTXID: 3, MaxTXID: 3, Size: 800}}}, |
| 286 | "a cap above the worker's": {V: ProtocolVersion, Op: OpQuery, Target: 1, MaxRestored: 5000, |
| 287 | Files: []dbdoor.LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 1, Size: 10}}}, |
| 288 | "a negative size": {V: ProtocolVersion, Op: OpQuery, Target: 2, MaxRestored: 1000, |
| 289 | Files: []dbdoor.LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 1, Size: -5}, {Level: 0, MinTXID: 2, MaxTXID: 2, Size: 10}}}, |
| 290 | } { |
| 291 | t.Run(name, func(t *testing.T) { |
| 292 | t.Parallel() |
| 293 | scratch := t.TempDir() |
| 294 | conn, done := workerPipe(t, WorkerOptions{Scratch: scratch, MaxRestored: 1000}) |
| 295 | if err := writeJSONFrame(conn, frameHeader, h); err != nil { |
| 296 | t.Fatal(err) |
| 297 | } |
| 298 | var end endMsg |
| 299 | readTyped(t, conn, frameEnd, &end) |
| 300 | if end.Code != CodeCheckFailed { |
| 301 | t.Fatalf("end = %+v", end) |
| 302 | } |
| 303 | <-done |
| 304 | if ents, _ := os.ReadDir(scratch); len(ents) != 0 { |
| 305 | t.Fatalf("the worker wrote %d entries for a refused plan", len(ents)) |
| 306 | } |
| 307 | }) |
| 308 | } |
| 309 | } |
| 310 | |
amadan