| 1 | // Acceptance for the forwarded `carlos db query` through the real CLI |
| 2 | // binary and the real console server: T6 (parity with the in-region door, |
| 3 | // through the fake S3 the db door's own e2e uses), T7 (the long request |
| 4 | // through the real edge and the console's real http.Server), T8 (mixed |
| 5 | // versions) and T11 (nothing printed for an answer that does not check out). |
| 6 | // The box is plan 1's real agent and worker code joined in process |
| 7 | // (dbreadstest): not acceptance for the units, which plan 1's root-runner |
| 8 | // tests are. A scripted box (e2eScript) is a stand-in and says so. |
| 9 | package main |
| 10 | |
| 11 | import ( |
| 12 | "bufio" |
| 13 | "bytes" |
| 14 | "context" |
| 15 | "crypto/ecdh" |
| 16 | "crypto/ed25519" |
| 17 | "crypto/rand" |
| 18 | "crypto/sha256" |
| 19 | "database/sql" |
| 20 | "encoding/hex" |
| 21 | "encoding/json" |
| 22 | "fmt" |
| 23 | "io" |
| 24 | "net" |
| 25 | "net/http" |
| 26 | "net/http/httptest" |
| 27 | "os" |
| 28 | "os/exec" |
| 29 | "path/filepath" |
| 30 | "strings" |
| 31 | "sync" |
| 32 | "testing" |
| 33 | "time" |
| 34 | |
| 35 | "github.com/benbjohnson/litestream" |
| 36 | lsfile "github.com/benbjohnson/litestream/file" |
| 37 | "github.com/superfly/ltx" |
| 38 | |
| 39 | "github.com/carlosframework/platform/internal/console" |
| 40 | "github.com/carlosframework/platform/internal/dbdoor" |
| 41 | "github.com/carlosframework/platform/internal/dbdoor/dbdoortest" |
| 42 | "github.com/carlosframework/platform/internal/dbreads" |
| 43 | "github.com/carlosframework/platform/internal/dbreads/dbreadstest" |
| 44 | "github.com/carlosframework/platform/internal/edge" |
| 45 | "github.com/carlosframework/platform/internal/lease" |
| 46 | "github.com/carlosframework/platform/internal/provision" |
| 47 | "github.com/carlosframework/platform/internal/registry" |
| 48 | "github.com/carlosframework/platform/internal/release" |
| 49 | ) |
| 50 | |
| 51 | const ( |
| 52 | dbfE2EHere = "shop.example.com" // in the console's region, ie |
| 53 | dbfE2EFar = "far.example.com" // in uk, with the same database bytes |
| 54 | ) |
| 55 | |
| 56 | type dbfE2EOptions struct { |
| 57 | forward bool |
| 58 | // serve serves the console's handler and returns the URL the CLI uses |
| 59 | // and any extra environment it needs; nil is the httptest server. |
| 60 | serve func(t *testing.T, h http.Handler) (consoleURL string, env []string) |
| 61 | // worker is the in-process box's worker options (a slow restore for T7). |
| 62 | worker dbreads.WorkerOptions |
| 63 | // noBox leaves the uk box out; the test scripts its own. |
| 64 | noBox bool |
| 65 | } |
| 66 | |
| 67 | type dbfE2E struct { |
| 68 | bin, creds, url, tok, root string |
| 69 | env []string |
| 70 | store release.Store |
| 71 | pub ed25519.PublicKey |
| 72 | priv ed25519.PrivateKey |
| 73 | } |
| 74 | |
| 75 | // newDBFE2E is a console in ie over a DirStore and the fake S3: shop in ie |
| 76 | // and far in uk with byte-identical replicas (the same files with the same |
| 77 | // modification times, so the fake S3 lists the same LastModified for both), |
| 78 | // a capable live uk box, and a signed-in CLI. |
| 79 | func newDBFE2E(t *testing.T, o dbfE2EOptions) *dbfE2E { |
| 80 | t.Helper() |
| 81 | creds := useTempCredentials(t) |
| 82 | clearDeploymentEnv(t) |
| 83 | tmp := t.TempDir() |
| 84 | e := &dbfE2E{bin: buildCarlos(t), creds: creds, store: release.NewDirStore(filepath.Join(tmp, "store")), root: filepath.Join(tmp, "data")} |
| 85 | var err error |
| 86 | if e.pub, e.priv, err = ed25519.GenerateKey(rand.Reader); err != nil { |
| 87 | t.Fatal(err) |
| 88 | } |
| 89 | here := filepath.Join(e.root, "instances", dbfE2EHere) |
| 90 | dbfE2EFixture(t, here) |
| 91 | copyTreeKeepingTimes(t, here, filepath.Join(e.root, "instances", dbfE2EFar)) |
| 92 | s3 := httptest.NewServer(&replicaS3{t: t, bucket: "data", root: e.root}) |
| 93 | t.Cleanup(s3.Close) |
| 94 | t.Setenv("CARLOS_S3_ENDPOINT", s3.URL) |
| 95 | t.Setenv("AWS_ACCESS_KEY_ID", "test") |
| 96 | t.Setenv("AWS_SECRET_ACCESS_KEY", "test") |
| 97 | t.Setenv("AWS_EC2_METADATA_DISABLED", "true") |
| 98 | |
| 99 | cst, err := console.Open(filepath.Join(tmp, "console.db")) |
| 100 | if err != nil { |
| 101 | t.Fatal(err) |
| 102 | } |
| 103 | t.Cleanup(func() { cst.Close() }) |
| 104 | var srv *console.Server |
| 105 | h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { srv.ServeHTTP(w, r) }) |
| 106 | direct := httptest.NewServer(h) |
| 107 | t.Cleanup(direct.Close) |
| 108 | e.url = direct.URL |
| 109 | if o.serve != nil { |
| 110 | e.url, e.env = o.serve(t, h) |
| 111 | } |
| 112 | srv, err = console.NewServer(console.Config{ |
| 113 | Origin: direct.URL, Keymail: "http://keymail.invalid", Registry: filepath.Join(tmp, "registry.db"), |
| 114 | Releases: e.store, ReleasePub: e.pub, Key: e.priv, AppScheme: "http", |
| 115 | Region: "ie", HibernateBucket: "data", HibernateRegion: "us-east-1", HibernatePrefix: "instances", |
| 116 | DBForward: o.forward, |
| 117 | }, cst) |
| 118 | if err != nil { |
| 119 | t.Fatal(err) |
| 120 | } |
| 121 | const addr = "member@example.com" |
| 122 | if err := cst.AddMember(console.OpsAccountID, addr, console.RoleOwner, time.Now()); err != nil { |
| 123 | t.Fatal(err) |
| 124 | } |
| 125 | if _, err := cst.CreateAppIn(console.OpsAccountID, "shop", addr, time.Now()); err != nil { |
| 126 | t.Fatal(err) |
| 127 | } |
| 128 | created := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) |
| 129 | for host, region := range map[string]string{dbfE2EHere: "ie", dbfE2EFar: "uk"} { |
| 130 | yes := true |
| 131 | body, _ := json.Marshal(provision.Record{Host: host, App: "shop", Account: release.OpsAccount, Region: region, Hibernate: &yes, CreatedAt: created}) |
| 132 | if _, err := e.store.Put(context.Background(), provision.RecordKey(release.OpsAccount, "shop", host), body, ""); err != nil { |
| 133 | t.Fatal(err) |
| 134 | } |
| 135 | } |
| 136 | edges, _ := json.Marshal(map[string]any{"v": 1, "box": "uk-1", "eip": "203.0.113.10", "placement": "uk", |
| 137 | "capabilities": []string{dbreads.Capability}, "hosts": map[string]any{}}) |
| 138 | e.store.Put(context.Background(), "control/edges/uk-1.json", edges, "") |
| 139 | status, _ := json.Marshal(map[string]any{"generated_at": time.Now().UTC(), "label": "uk-1"}) |
| 140 | e.store.Put(context.Background(), "status/boxes/uk-1.json", status, "") |
| 141 | token, expiresAt := e2eDeviceLogin(t, direct.URL, cst, addr, "e2e@laptop") |
| 142 | e.tok = token |
| 143 | cf, err := loadCredentials() |
| 144 | if err != nil { |
| 145 | t.Fatal(err) |
| 146 | } |
| 147 | cf.put(credential{Console: e.url, Address: addr, Token: token, ExpiresAt: expiresAt}) |
| 148 | if err := cf.save(); err != nil { |
| 149 | t.Fatal(err) |
| 150 | } |
| 151 | if !o.noBox { |
| 152 | box := dbreadstest.Start(t, dbreadstest.Options{Store: e.store, Region: "uk", Box: "uk-1", Pub: e.pub, |
| 153 | HibernatePrefix: "instances", Worker: o.worker, |
| 154 | OpenReplica: func(ctx context.Context, prefix string) (dbdoor.ReplicaReader, error) { |
| 155 | return lease.ReadReplicaClient(ctx, "data", prefix, "us-east-1") |
| 156 | }}) |
| 157 | ctx, cancel := context.WithCancel(context.Background()) |
| 158 | done := make(chan struct{}) |
| 159 | go func() { |
| 160 | defer close(done) |
| 161 | for ctx.Err() == nil { |
| 162 | box.Agent.Pass(ctx) |
| 163 | time.Sleep(50 * time.Millisecond) |
| 164 | } |
| 165 | }() |
| 166 | t.Cleanup(func() { cancel(); <-done; box.Agent.Wait() }) |
| 167 | } |
| 168 | return e |
| 169 | } |
| 170 | |
| 171 | // dbfE2EFixture is T6's database (NULL, a tab and a newline in one value, a |
| 172 | // blob, a time) in two syncs a whole second apart. |
| 173 | func dbfE2EFixture(t *testing.T, dir string) { |
| 174 | t.Helper() |
| 175 | src := filepath.Join(t.TempDir(), "shop.db") |
| 176 | db, err := sql.Open("sqlite", src) |
| 177 | if err != nil { |
| 178 | t.Fatal(err) |
| 179 | } |
| 180 | defer db.Close() |
| 181 | for _, q := range []string{ |
| 182 | `PRAGMA journal_mode=WAL`, |
| 183 | `CREATE TABLE orders (id INTEGER PRIMARY KEY, note TEXT, data BLOB, at DATETIME)`, |
| 184 | } { |
| 185 | if _, err := db.Exec(q); err != nil { |
| 186 | t.Fatal(err) |
| 187 | } |
| 188 | } |
| 189 | if _, err := db.Exec(`INSERT INTO orders (note, data, at) VALUES (?, ?, ?), (NULL, NULL, NULL)`, |
| 190 | "a\ttab\nline", []byte{0, 0xff}, time.Date(2026, 10, 7, 14, 0, 0, 0, time.UTC)); err != nil { |
| 191 | t.Fatal(err) |
| 192 | } |
| 193 | dbdoortest.Sync(t, src, dir) |
| 194 | time.Sleep(time.Until(time.Now().Truncate(time.Second).Add(1100 * time.Millisecond))) |
| 195 | if _, err := db.Exec(`INSERT INTO orders (note) VALUES ('later')`); err != nil { |
| 196 | t.Fatal(err) |
| 197 | } |
| 198 | dbdoortest.Sync(t, src, dir) |
| 199 | } |
| 200 | |
| 201 | func copyTreeKeepingTimes(t *testing.T, from, to string) { |
| 202 | t.Helper() |
| 203 | err := filepath.Walk(from, func(p string, info os.FileInfo, err error) error { |
| 204 | if err != nil { |
| 205 | return err |
| 206 | } |
| 207 | rel, _ := filepath.Rel(from, p) |
| 208 | dst := filepath.Join(to, rel) |
| 209 | if info.IsDir() { |
| 210 | return os.MkdirAll(dst, 0o755) |
| 211 | } |
| 212 | b, err := os.ReadFile(p) |
| 213 | if err != nil { |
| 214 | return err |
| 215 | } |
| 216 | if err := os.WriteFile(dst, b, 0o644); err != nil { |
| 217 | return err |
| 218 | } |
| 219 | return os.Chtimes(dst, info.ModTime(), info.ModTime()) |
| 220 | }) |
| 221 | if err != nil { |
| 222 | t.Fatal(err) |
| 223 | } |
| 224 | } |
| 225 | |
| 226 | // run runs the real CLI. Callers name the account (--account ops), as the |
| 227 | // db door's own e2e does: a defaulted account prints its note on stdout, |
| 228 | // which is main's behaviour for every db verb and not this plan's to change. |
| 229 | func (e *dbfE2E) run(t *testing.T, args ...string) (string, string, error) { |
| 230 | t.Helper() |
| 231 | cmd := exec.Command(e.bin, args...) |
| 232 | cmd.Env = carlosTestEnv(t, append([]string{"CARLOS_CREDENTIALS=" + e.creds}, e.env...)...) |
| 233 | cmd.Dir = t.TempDir() |
| 234 | var stdout, stderr bytes.Buffer |
| 235 | cmd.Stdout, cmd.Stderr = &stdout, &stderr |
| 236 | err := cmd.Run() |
| 237 | return stdout.String(), stderr.String(), err |
| 238 | } |
| 239 | |
| 240 | // betweenTheSyncs is an --at that holds the first sync and not the second: |
| 241 | // half a second into the second after the first L0 file was written. The |
| 242 | // second sync is at least 1.1 s after the first (dbfE2EFixture), so its |
| 243 | // listed second is never before this time's. |
| 244 | func (e *dbfE2E) betweenTheSyncs(t *testing.T) string { |
| 245 | t.Helper() |
| 246 | ents, err := os.ReadDir(filepath.Join(e.root, "instances", dbfE2EHere, "ltx", "0")) |
| 247 | if err != nil || len(ents) < 2 { |
| 248 | t.Fatalf("the fixture has %d L0 files: %v", len(ents), err) |
| 249 | } |
| 250 | var first time.Time |
| 251 | for _, ent := range ents { |
| 252 | info, err := ent.Info() |
| 253 | if err != nil { |
| 254 | t.Fatal(err) |
| 255 | } |
| 256 | if first.IsZero() || info.ModTime().Before(first) { |
| 257 | first = info.ModTime() |
| 258 | } |
| 259 | } |
| 260 | return first.Truncate(time.Second).Add(1500 * time.Millisecond).UTC().Format(time.RFC3339Nano) |
| 261 | } |
| 262 | |
| 263 | // doer is an old CLI's view of the db door: GETs with the saved token. |
| 264 | func (e *dbfE2E) doer(host string) dbdoor.Doer { |
| 265 | return func(ctx context.Context, tail string, h http.Header) (*http.Response, error) { |
| 266 | req, err := http.NewRequestWithContext(ctx, http.MethodGet, e.url+dbdoor.Base(release.OpsAccount, "shop", host)+tail, nil) |
| 267 | if err != nil { |
| 268 | return nil, err |
| 269 | } |
| 270 | for k, vs := range h { |
| 271 | req.Header[k] = vs |
| 272 | } |
| 273 | req.Header.Set("Authorization", "Bearer "+e.tok) |
| 274 | return http.DefaultClient.Do(req) |
| 275 | } |
| 276 | } |
| 277 | |
| 278 | // e2eScript writes one read's objects itself: a stand-in for a box, sealing |
| 279 | // real chunks and a real summary to the request's reply key. |
| 280 | type e2eScript struct { |
| 281 | store release.Store |
| 282 | id string |
| 283 | pub *ecdh.PublicKey |
| 284 | } |
| 285 | |
| 286 | func (b *e2eScript) picked(deadline time.Time) { |
| 287 | p, _ := json.Marshal(dbreads.Picked{Box: "uk-1", Nonce: "scripted", Deadline: deadline}) |
| 288 | b.store.Put(context.Background(), dbreads.PickedKey(b.id), p, "") |
| 289 | } |
| 290 | |
| 291 | func (b *e2eScript) chunk(seq int64, last bool, plain string) { |
| 292 | sealed, _ := dbreads.SealChunk(b.pub, b.id, seq, dbreads.StreamRows, last, []byte(plain)) |
| 293 | b.store.Put(context.Background(), dbreads.ChunkKey(b.id, seq), sealed, "") |
| 294 | } |
| 295 | |
| 296 | func (b *e2eScript) result(last int64, rows string) { |
| 297 | sum := sha256.Sum256([]byte(rows)) |
| 298 | sealed, _ := dbreads.SealSummary(b.pub, b.id, dbreads.Summary{TXID: 9, Bytes: int64(len(rows)), SHA256: hex.EncodeToString(sum[:])}) |
| 299 | r, _ := json.Marshal(dbreads.Result{State: dbreads.StateDone, LastSeq: last, Box: "uk-1", Sealed: sealed}) |
| 300 | b.store.Put(context.Background(), dbreads.ResultKey(b.id), r, "") |
| 301 | } |
| 302 | |
| 303 | // scriptBox runs script once for each new request in uk's queue, after |
| 304 | // verifying it as a box would. |
| 305 | func (e *dbfE2E) scriptBox(t *testing.T, script func(*e2eScript)) func() { |
| 306 | ctx, cancel := context.WithCancel(context.Background()) |
| 307 | done := make(chan struct{}) |
| 308 | go func() { |
| 309 | defer close(done) |
| 310 | seen := map[string]bool{} |
| 311 | for ctx.Err() == nil { |
| 312 | keys, _ := e.store.List(ctx, dbreads.QueueRegionPrefix("uk")) |
| 313 | for _, k := range keys { |
| 314 | q, ok := dbreads.ParseQueueKey(k) |
| 315 | if !ok || seen[q.ID] { |
| 316 | continue |
| 317 | } |
| 318 | seen[q.ID] = true |
| 319 | raw, _, err := e.store.Get(ctx, dbreads.RequestKey(q.ID)) |
| 320 | if err != nil { |
| 321 | continue |
| 322 | } |
| 323 | req, err := dbreads.Verify(e.pub, raw, q.ID, "uk", time.Now()) |
| 324 | if err != nil { |
| 325 | t.Errorf("the scripted box refused %s: %v", q.ID, err) |
| 326 | continue |
| 327 | } |
| 328 | pub, _ := dbreads.ParseReplyKey(req.ReplyKey) |
| 329 | go script(&e2eScript{store: e.store, id: q.ID, pub: pub}) |
| 330 | } |
| 331 | time.Sleep(50 * time.Millisecond) |
| 332 | } |
| 333 | }() |
| 334 | return func() { cancel(); <-done } |
| 335 | } |
| 336 | |
| 337 | // throughTheEdge serves the console's handler with console.NewHTTPServer on |
| 338 | // 127.0.0.1, its ReadTimeout set to readTimeout, behind a real edge route |
| 339 | // for console.test. The CLI reaches it by that name through HTTP_PROXY, |
| 340 | // which stands in for DNS. |
| 341 | func throughTheEdge(readTimeout time.Duration, edgeURL *string, tap **dbfTap) func(t *testing.T, h http.Handler) (string, []string) { |
| 342 | return func(t *testing.T, h http.Handler) (string, []string) { |
| 343 | ln, err := net.Listen("tcp", "127.0.0.1:0") |
| 344 | if err != nil { |
| 345 | t.Fatal(err) |
| 346 | } |
| 347 | hs := console.NewHTTPServer(h) |
| 348 | hs.ReadTimeout = readTimeout |
| 349 | go hs.Serve(ln) |
| 350 | t.Cleanup(func() { hs.Close() }) |
| 351 | reg, err := registry.Open(filepath.Join(t.TempDir(), "edge.db")) |
| 352 | if err != nil { |
| 353 | t.Fatal(err) |
| 354 | } |
| 355 | if err := reg.Upsert(registry.Route{Host: "console.test", Kind: "service", Addr: ln.Addr().String(), Status: "ready"}); err != nil { |
| 356 | t.Fatal(err) |
| 357 | } |
| 358 | ed, err := edge.New(reg, nil) |
| 359 | if err != nil { |
| 360 | t.Fatal(err) |
| 361 | } |
| 362 | if err := ed.Reload(); err != nil { |
| 363 | t.Fatal(err) |
| 364 | } |
| 365 | es := httptest.NewServer(ed) |
| 366 | t.Cleanup(es.Close) |
| 367 | *edgeURL = es.URL |
| 368 | // The CLI's proxy is a tap in front of the edge, so the test sees |
| 369 | // what reaches the CLI's own socket (plan review finding 34). |
| 370 | tp := newDBFTap(t, es.Listener.Addr().String(), false) |
| 371 | *tap = tp |
| 372 | return "http://console.test", []string{"HTTP_PROXY=http://" + tp.addr, "NO_PROXY="} |
| 373 | } |
| 374 | } |
| 375 | |
| 376 | // dbfTap stands between the CLI and the first edge, forwarding bytes both |
| 377 | // ways and noting when each read from the edge's side was passed to the |
| 378 | // CLI's socket: what the CLI process received, and when. With connect it is |
| 379 | // an HTTPS proxy (CONNECT), which is how the CLI reaches a TLS edge by name |
| 380 | // here; without, a plain forwarder for HTTP_PROXY. |
| 381 | type dbfTap struct { |
| 382 | addr string |
| 383 | mu sync.Mutex |
| 384 | down []time.Time |
| 385 | } |
| 386 | |
| 387 | func newDBFTap(t *testing.T, target string, connect bool) *dbfTap { |
| 388 | t.Helper() |
| 389 | ln, err := net.Listen("tcp", "127.0.0.1:0") |
| 390 | if err != nil { |
| 391 | t.Fatal(err) |
| 392 | } |
| 393 | t.Cleanup(func() { ln.Close() }) |
| 394 | tp := &dbfTap{addr: ln.Addr().String()} |
| 395 | go func() { |
| 396 | for { |
| 397 | c, err := ln.Accept() |
| 398 | if err != nil { |
| 399 | return |
| 400 | } |
| 401 | go tp.serve(c, target, connect) |
| 402 | } |
| 403 | }() |
| 404 | return tp |
| 405 | } |
| 406 | |
| 407 | func (tp *dbfTap) serve(c net.Conn, target string, connect bool) { |
| 408 | defer c.Close() |
| 409 | var in io.Reader = c |
| 410 | if connect { |
| 411 | br := bufio.NewReader(c) |
| 412 | req, err := http.ReadRequest(br) |
| 413 | if err != nil || req.Method != http.MethodConnect { |
| 414 | return |
| 415 | } |
| 416 | if _, err := io.WriteString(c, "HTTP/1.1 200 Connection established\r\n\r\n"); err != nil { |
| 417 | return |
| 418 | } |
| 419 | in = br |
| 420 | } |
| 421 | up, err := net.Dial("tcp", target) |
| 422 | if err != nil { |
| 423 | return |
| 424 | } |
| 425 | defer up.Close() |
| 426 | go func() { |
| 427 | io.Copy(up, in) |
| 428 | up.(*net.TCPConn).CloseWrite() |
| 429 | }() |
| 430 | buf := make([]byte, 32<<10) |
| 431 | for { |
| 432 | n, err := up.Read(buf) |
| 433 | if n > 0 { |
| 434 | tp.mu.Lock() |
| 435 | tp.down = append(tp.down, time.Now()) |
| 436 | tp.mu.Unlock() |
| 437 | if _, werr := c.Write(buf[:n]); werr != nil { |
| 438 | return |
| 439 | } |
| 440 | } |
| 441 | if err != nil { |
| 442 | return |
| 443 | } |
| 444 | } |
| 445 | } |
| 446 | |
| 447 | // longestGap is the longest silence at the CLI's socket between start and |
| 448 | // end, both of which count as arrivals. |
| 449 | func (tp *dbfTap) longestGap(start, end time.Time) time.Duration { |
| 450 | tp.mu.Lock() |
| 451 | defer tp.mu.Unlock() |
| 452 | prev, longest := start, time.Duration(0) |
| 453 | for _, at := range append(append([]time.Time(nil), tp.down...), end) { |
| 454 | if at.Before(start) || at.After(end) { |
| 455 | continue |
| 456 | } |
| 457 | if d := at.Sub(prev); d > longest { |
| 458 | longest = d |
| 459 | } |
| 460 | prev = at |
| 461 | } |
| 462 | return longest |
| 463 | } |
| 464 | |
| 465 | // checkProgressAtTheCLI is T7's witness at the CLI process itself (plan |
| 466 | // review finding 34): it runs the real CLI and needs DBF-11 on its stderr |
| 467 | // within a second of the process starting (stricter than the spec's "of the |
| 468 | // request": the process starts before it sends anything), no silence over |
| 469 | // 16 s (15 s and a poll) at its socket, the read lasting at least atLeast, and |
| 470 | // its rows. atLeast is the quiet restore's length. |
| 471 | func checkProgressAtTheCLI(t *testing.T, e *dbfE2E, tap *dbfTap, atLeast time.Duration) { |
| 472 | t.Helper() |
| 473 | cmd := exec.Command(e.bin, "db", "query", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT count(*) AS n FROM orders") |
| 474 | cmd.Env = carlosTestEnv(t, append([]string{"CARLOS_CREDENTIALS=" + e.creds}, e.env...)...) |
| 475 | cmd.Dir = t.TempDir() |
| 476 | var out bytes.Buffer |
| 477 | cmd.Stdout = &out |
| 478 | pipe, err := cmd.StderrPipe() |
| 479 | if err != nil { |
| 480 | t.Fatal(err) |
| 481 | } |
| 482 | start := time.Now() |
| 483 | if err := cmd.Start(); err != nil { |
| 484 | t.Fatal(err) |
| 485 | } |
| 486 | var stderr []string |
| 487 | var reading time.Time |
| 488 | sc := bufio.NewScanner(pipe) |
| 489 | for sc.Scan() { |
| 490 | if reading.IsZero() && strings.Contains(sc.Text(), fmt.Sprintf(dbfReading, dbfE2EFar, "uk")) { |
| 491 | reading = time.Now() |
| 492 | } |
| 493 | stderr = append(stderr, sc.Text()) |
| 494 | } |
| 495 | err = cmd.Wait() |
| 496 | end := time.Now() |
| 497 | if err != nil || out.String() != "n\n3\n" { |
| 498 | t.Fatalf("%q %v\n%s", out.String(), err, strings.Join(stderr, "\n")) |
| 499 | } |
| 500 | if reading.IsZero() || reading.Sub(start) > time.Second { |
| 501 | t.Fatalf("DBF-11 after %s: %q", reading.Sub(start), stderr) |
| 502 | } |
| 503 | if gap := tap.longestGap(start, end); gap > 16*time.Second { |
| 504 | t.Fatalf("%s of silence at the CLI's socket", gap) |
| 505 | } |
| 506 | if took := end.Sub(start); took < atLeast { |
| 507 | t.Fatalf("the read took %s, under %s: it did not hold the request", took, atLeast) |
| 508 | } |
| 509 | } |
| 510 | |
| 511 | // slowRestore holds the box's restore for d before restoring for real. |
| 512 | func slowRestore(d time.Duration) dbreads.WorkerOptions { |
| 513 | return dbreads.WorkerOptions{Restore: func(ctx context.Context, c litestream.ReplicaClient, out string, target uint64) error { |
| 514 | select { |
| 515 | case <-ctx.Done(): |
| 516 | return ctx.Err() |
| 517 | case <-time.After(d): |
| 518 | } |
| 519 | return dbdoor.RestoreLocal(ctx, c, out, target) |
| 520 | }} |
| 521 | } |
| 522 | |
| 523 | // T6: the forwarded query and the in-region query print byte-identical |
| 524 | // stdout, for the newest copy and as of a time between the two syncs. |
| 525 | func TestDBForwardParityThroughTheCLI(t *testing.T) { |
| 526 | e := newDBFE2E(t, dbfE2EOptions{forward: true}) |
| 527 | const q = "SELECT id, note, data, at FROM orders ORDER BY id" |
| 528 | for _, at := range []string{"", e.betweenTheSyncs(t)} { |
| 529 | args := []string{"db", "query", "--account", "ops", "--app", "shop", "--sql", q} |
| 530 | if at != "" { |
| 531 | args = append(args, "--at", at) |
| 532 | } |
| 533 | here, hereErr, err := e.run(t, append(args, "--instance", dbfE2EHere)...) |
| 534 | if err != nil { |
| 535 | t.Fatalf("in region: %v\n%s", err, hereErr) |
| 536 | } |
| 537 | far, farErr, err := e.run(t, append(args, "--instance", dbfE2EFar)...) |
| 538 | if err != nil { |
| 539 | t.Fatalf("forwarded: %v\n%s", err, farErr) |
| 540 | } |
| 541 | if here != far || here == "" { |
| 542 | t.Fatalf("at %q: in region %q, forwarded %q", at, here, far) |
| 543 | } |
| 544 | if !strings.Contains(farErr, fmt.Sprintf(dbfReading, dbfE2EFar, "uk")) { |
| 545 | t.Fatalf("no DBF-11 on stderr: %s", farErr) |
| 546 | } |
| 547 | if at != "" && strings.Contains(far, "later") { |
| 548 | t.Fatalf("as of %s the copy holds the second sync: %q", at, far) |
| 549 | } |
| 550 | } |
| 551 | } |
| 552 | |
| 553 | // T7, scaled: the console's real http.Server with ReadTimeout 2 s instead of |
| 554 | // 5 minutes, behind the real edge, and a read held 20 s by a slow restore. |
| 555 | // The connection outlives ReadTimeout tenfold, admitted arrives within a |
| 556 | // second, a line arrives at least every 15 s (plus a poll), and the real CLI |
| 557 | // gets its rows through the same path. |
| 558 | func TestDBForwardLongRequestThroughTheEdge(t *testing.T) { |
| 559 | var edgeURL string |
| 560 | var tap *dbfTap |
| 561 | e := newDBFE2E(t, dbfE2EOptions{forward: true, serve: throughTheEdge(2*time.Second, &edgeURL, &tap), worker: slowRestore(20 * time.Second)}) |
| 562 | key, _ := dbreads.NewReplyKey() |
| 563 | body, _ := json.Marshal(dbreads.ReadBody{Op: dbreads.OpQuery, SQL: "SELECT count(*) AS n FROM orders", ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())}) |
| 564 | req, _ := http.NewRequest(http.MethodPost, edgeURL+"/api/cli/apps/"+release.OpsAccount+"/shop/db/"+dbfE2EFar+"/reads", bytes.NewReader(body)) |
| 565 | req.Host = "console.test" |
| 566 | req.Header.Set("Content-Type", "application/json") |
| 567 | req.Header.Set("Authorization", "Bearer "+e.tok) |
| 568 | start := time.Now() |
| 569 | res, err := http.DefaultClient.Do(req) |
| 570 | if err != nil { |
| 571 | t.Fatal(err) |
| 572 | } |
| 573 | defer res.Body.Close() |
| 574 | br := bufio.NewReader(res.Body) |
| 575 | prev, first, sawResult := start, true, false |
| 576 | for { |
| 577 | line, err := br.ReadBytes('\n') |
| 578 | if err == io.EOF { |
| 579 | break |
| 580 | } |
| 581 | if err != nil { |
| 582 | t.Fatal(err) |
| 583 | } |
| 584 | gap := time.Since(prev) |
| 585 | if first && gap > time.Second { |
| 586 | t.Fatalf("admitted took %s", gap) |
| 587 | } |
| 588 | if gap > 16*time.Second { |
| 589 | t.Fatalf("%s with no line", gap) |
| 590 | } |
| 591 | prev, first = time.Now(), false |
| 592 | sawResult = sawResult || bytes.Contains(line, []byte(`"result"`)) |
| 593 | } |
| 594 | if !sawResult || time.Since(start) < 20*time.Second { |
| 595 | t.Fatalf("result %v after %s", sawResult, time.Since(start)) |
| 596 | } |
| 597 | // The same through the real CLI process, observed at its own stderr and |
| 598 | // socket (plan review finding 34). |
| 599 | checkProgressAtTheCLI(t, e, tap, 20*time.Second) |
| 600 | } |
| 601 | |
| 602 | // T7, unscaled: a read that holds the request for 21 minutes through the |
| 603 | // real edge and the console's unmodified http.Server (5-minute ReadTimeout), |
| 604 | // inside the CLI's 23-minute context. The box is scripted (a stand-in): |
| 605 | // picked at 1 minute 50 seconds with a 20-minute deadline, one chunk at 10 |
| 606 | // minutes, the last and the result at 21. It waits; it does not compute. |
| 607 | // Run once before handoff with CARLOS_DBF_LONG=1. |
| 608 | func TestDBForwardTwentyOneMinuteRead(t *testing.T) { |
| 609 | if os.Getenv("CARLOS_DBF_LONG") != "1" { |
| 610 | t.Skip("set CARLOS_DBF_LONG=1 to run the unscaled 21-minute read, once, before handoff") |
| 611 | } |
| 612 | var edgeURL string |
| 613 | var tap *dbfTap |
| 614 | e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true, serve: throughTheEdge(5*time.Minute, &edgeURL, &tap)}) |
| 615 | stop := e.scriptBox(t, func(b *e2eScript) { |
| 616 | time.Sleep(110 * time.Second) |
| 617 | b.picked(time.Now().Add(20 * time.Minute)) |
| 618 | time.Sleep(8 * time.Minute) |
| 619 | b.chunk(1, false, "n\n") |
| 620 | time.Sleep(11*time.Minute + 10*time.Second) |
| 621 | b.chunk(2, true, "3\n") |
| 622 | b.result(2, "n\n3\n") |
| 623 | }) |
| 624 | defer stop() |
| 625 | start := time.Now() |
| 626 | out, stderr, err := e.run(t, "db", "query", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT count(*) AS n FROM orders") |
| 627 | if err != nil || out != "n\n3\n" || time.Since(start) < 21*time.Minute { |
| 628 | t.Fatalf("after %s: %q %v\n%s", time.Since(start), out, err, stderr) |
| 629 | } |
| 630 | } |
| 631 | |
| 632 | // T8: the rows of spec §9.2 the CLI sees. |
| 633 | func TestDBForwardMixedVersions(t *testing.T) { |
| 634 | off := newDBFE2E(t, dbfE2EOptions{forward: false, noBox: true}) |
| 635 | // A new CLI and a console with forwarding off (or an old console): the |
| 636 | // residency sentence as it is, and nothing on stdout. |
| 637 | out, stderr, err := off.run(t, "db", "query", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT 1") |
| 638 | if err == nil || out != "" || !strings.Contains(stderr, "far.example.com keeps its data in region uk. This console can only read data kept in region ie.") { |
| 639 | t.Fatalf("new CLI, forwarding off: %q %q %v", out, stderr, err) |
| 640 | } |
| 641 | on := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true}) |
| 642 | // An old CLI and a console with forwarding on: every CLI from before |
| 643 | // this plan runs dbRestoreError on a door refusal, which prints the |
| 644 | // door's sentence. That sentence is DBF-14. |
| 645 | _, derr := dbdoor.NewClient(on.doer(dbfE2EFar)).LTXFiles(context.Background(), 9, 0, false) |
| 646 | want := "far.example.com keeps its data in region uk. this version of carlos cannot read it from there. run carlos update, then try again." |
| 647 | if got := dbRestoreError(dbfE2EFar, derr).Error(); got != want { |
| 648 | t.Fatalf("old CLI, new console: %q", got) |
| 649 | } |
| 650 | } |
| 651 | |
| 652 | // T11 (codex 6): the real CLI prints nothing for a dropped last chunk, a |
| 653 | // summary hash mismatch, or a response that ends without a result. The box |
| 654 | // is scripted: a stand-in that writes picked, chunks and result itself. |
| 655 | func TestDBForwardPrintsNothingForABadAnswer(t *testing.T) { |
| 656 | for name, c := range map[string]struct { |
| 657 | script func(b *e2eScript) |
| 658 | want string |
| 659 | }{ |
| 660 | // The bucket lost the real last chunk: the result names chunk 1, |
| 661 | // which was sealed as not the last. |
| 662 | "the last chunk dropped": {func(b *e2eScript) { |
| 663 | b.picked(time.Now().Add(time.Minute)) |
| 664 | b.chunk(1, false, "n\n") |
| 665 | b.result(1, "n\n") |
| 666 | }, "was invalid, so carlos db printed nothing"}, |
| 667 | "a summary hash mismatch": {func(b *e2eScript) { |
| 668 | b.picked(time.Now().Add(time.Minute)) |
| 669 | b.chunk(1, true, "n\n31\n") |
| 670 | b.result(1, "n\n30\n") |
| 671 | }, "was invalid, so carlos db printed nothing"}, |
| 672 | // The console waits the owner's deadline plus its 1-minute grace. |
| 673 | "no result": {func(b *e2eScript) { |
| 674 | b.picked(time.Now().Add(2 * time.Second)) |
| 675 | b.chunk(1, false, "n\n") |
| 676 | }, "did not finish, so nothing was printed"}, |
| 677 | } { |
| 678 | t.Run(name, func(t *testing.T) { |
| 679 | e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true}) |
| 680 | stop := e.scriptBox(t, c.script) |
| 681 | defer stop() |
| 682 | out, stderr, err := e.run(t, "db", "query", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT count(*) AS n FROM orders") |
| 683 | if err == nil || out != "" || !strings.Contains(stderr, c.want) { |
| 684 | t.Fatalf("%q %q %v", out, stderr, err) |
| 685 | } |
| 686 | }) |
| 687 | } |
| 688 | } |
| 689 | |
| 690 | // Plan review finding 36, through the binary: a query holding a byte that |
| 691 | // is not UTF-8 is refused with DBF-32 before anything is sent, so no altered |
| 692 | // query reaches the bucket. (An argv may carry any bytes on Linux, where |
| 693 | // this runs.) |
| 694 | func TestDBForwardRefusesSQLThatIsNotUTF8(t *testing.T) { |
| 695 | e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true}) |
| 696 | out, stderr, err := e.run(t, "db", "query", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT 'caf\xe9'") |
| 697 | if err == nil || out != "" || !strings.Contains(stderr, fmt.Sprintf(dbfNotUTF8, "uk")) { |
| 698 | t.Fatalf("%q %q %v", out, stderr, err) |
| 699 | } |
| 700 | if keys, _ := e.store.List(context.Background(), dbreads.Prefix); len(keys) != 0 { |
| 701 | t.Fatalf("a query JSON would rewrite reached the bucket: %v", keys) |
| 702 | } |
| 703 | } |
| 704 | |
| 705 | // throughBackhaul serves the console's handler with console.NewHTTPServer |
| 706 | // (its ReadTimeout set to readTimeout) on 127.0.0.1, behind two real edges |
| 707 | // that package edge's test binary runs as a helper process: the CLI reaches |
| 708 | // betty, which backhauls to annie, the console's home. The CLI finds betty |
| 709 | // through a CONNECT proxy (a dbfTap, standing in for DNS) and trusts the |
| 710 | // edges' test CA through SSL_CERT_FILE. Building the helper needs the go |
| 711 | // command on PATH, as every CI step has. |
| 712 | func throughBackhaul(readTimeout time.Duration, tap **dbfTap) func(t *testing.T, h http.Handler) (string, []string) { |
| 713 | return func(t *testing.T, h http.Handler) (string, []string) { |
| 714 | ln, err := net.Listen("tcp", "127.0.0.1:0") |
| 715 | if err != nil { |
| 716 | t.Fatal(err) |
| 717 | } |
| 718 | hs := console.NewHTTPServer(h) |
| 719 | hs.ReadTimeout = readTimeout |
| 720 | go hs.Serve(ln) |
| 721 | t.Cleanup(func() { hs.Close() }) |
| 722 | dir := t.TempDir() |
| 723 | bin := filepath.Join(dir, "edge.test") |
| 724 | // From the module root with the environment TestMain kept for the |
| 725 | // carlos build (the real HOME and Go cache), as sharedCarlos builds. |
| 726 | build := exec.Command("go", "test", "-c", "-o", bin, "./internal/edge") |
| 727 | build.Dir, build.Env = moduleRoot, carlosBuildEnv |
| 728 | if out, err := build.CombinedOutput(); err != nil { |
| 729 | t.Fatalf("build the edge test binary: %v\n%s", err, out) |
| 730 | } |
| 731 | cmd := exec.Command(bin, "-test.run", "^TestDBForwardBackhaulPairHelper$", "-test.v") |
| 732 | cmd.Env = append(os.Environ(), "DBF_BACKHAUL_HELPER=1", "DBF_BACKHAUL_UPSTREAM="+ln.Addr().String(), "DBF_BACKHAUL_DIR="+dir) |
| 733 | stdin, err := cmd.StdinPipe() |
| 734 | if err != nil { |
| 735 | t.Fatal(err) |
| 736 | } |
| 737 | pr, pw := io.Pipe() |
| 738 | cmd.Stdout = pw |
| 739 | var helperErr bytes.Buffer |
| 740 | cmd.Stderr = &helperErr |
| 741 | if err := cmd.Start(); err != nil { |
| 742 | t.Fatal(err) |
| 743 | } |
| 744 | t.Cleanup(func() { |
| 745 | stdin.Close() |
| 746 | cmd.Wait() |
| 747 | pw.Close() |
| 748 | if t.Failed() { |
| 749 | t.Logf("backhaul helper stderr:\n%s", helperErr.String()) |
| 750 | } |
| 751 | }) |
| 752 | lines := bufio.NewScanner(pr) |
| 753 | sender := "" |
| 754 | for lines.Scan() { |
| 755 | if addr, ok := strings.CutPrefix(lines.Text(), "DBF-BACKHAUL "); ok { |
| 756 | sender = addr |
| 757 | break |
| 758 | } |
| 759 | } |
| 760 | if sender == "" { |
| 761 | t.Fatalf("the backhaul helper never came up:\n%s", helperErr.String()) |
| 762 | } |
| 763 | go func() { |
| 764 | for lines.Scan() { |
| 765 | } |
| 766 | }() |
| 767 | tp := newDBFTap(t, sender, true) |
| 768 | *tap = tp |
| 769 | return "https://console.backhaul.test", []string{"HTTPS_PROXY=http://" + tp.addr, "NO_PROXY=", "SSL_CERT_FILE=" + filepath.Join(dir, "ca.pem")} |
| 770 | } |
| 771 | } |
| 772 | |
| 773 | // T7, the backhaul half, end to end (plan review finding 34): the real CLI |
| 774 | // binary reaches the real console through two real edges, the second hop a |
| 775 | // backhaul with its 30 s header bound, during a 40 s quiet restore. At the |
| 776 | // CLI's own stderr and socket: DBF-11 within a second, no silence over 16 s, |
| 777 | // and its rows after more than the hop's 30 s. The console's real |
| 778 | // http.Server runs with ReadTimeout 2 s, as in |
| 779 | // TestDBForwardLongRequestThroughTheEdge. |
| 780 | // Mutations: drop the stream's flush (red: the wait lines sit in the |
| 781 | // console's buffer and the CLI's socket is silent); stream nothing before |
| 782 | // the box's first chunk (red: no headers inside the hop's 30 s). |
| 783 | func TestDBForwardThroughTheBackhaulHop(t *testing.T) { |
| 784 | var tap *dbfTap |
| 785 | e := newDBFE2E(t, dbfE2EOptions{forward: true, serve: throughBackhaul(2*time.Second, &tap), worker: slowRestore(40 * time.Second)}) |
| 786 | checkProgressAtTheCLI(t, e, tap, 40*time.Second) |
| 787 | } |
| 788 | |
| 789 | func sha256File(t *testing.T, path string) string { |
| 790 | t.Helper() |
| 791 | b, err := os.ReadFile(path) |
| 792 | if err != nil { |
| 793 | t.Fatal(err) |
| 794 | } |
| 795 | sum := sha256.Sum256(b) |
| 796 | return hex.EncodeToString(sum[:]) |
| 797 | } |
| 798 | |
| 799 | // The pull's parity (plan 3): a forwarded pull and an in-region pull of the |
| 800 | // same replica save byte-identical copies, for the newest copy and as of a |
| 801 | // time, and print the same lines but for the host and the path. |
| 802 | func TestDBForwardPullParityThroughTheCLI(t *testing.T) { |
| 803 | e := newDBFE2E(t, dbfE2EOptions{forward: true}) |
| 804 | for _, at := range []string{"", e.betweenTheSyncs(t)} { |
| 805 | dir := t.TempDir() |
| 806 | args := []string{"db", "pull", "--account", "ops", "--app", "shop"} |
| 807 | if at != "" { |
| 808 | args = append(args, "--at", at) |
| 809 | } |
| 810 | hereOut, farOut := filepath.Join(dir, "here.db"), filepath.Join(dir, "far.db") |
| 811 | here, hereErr, err := e.run(t, append(args, "--instance", dbfE2EHere, "--out", hereOut)...) |
| 812 | if err != nil { |
| 813 | t.Fatalf("in region: %v\n%s", err, hereErr) |
| 814 | } |
| 815 | far, farErr, err := e.run(t, append(args, "--instance", dbfE2EFar, "--out", farOut)...) |
| 816 | if err != nil { |
| 817 | t.Fatalf("forwarded: %v\n%s", err, farErr) |
| 818 | } |
| 819 | if sha256File(t, hereOut) != sha256File(t, farOut) { |
| 820 | t.Fatalf("at %q the copies differ", at) |
| 821 | } |
| 822 | if here == "" || strings.ReplaceAll(strings.ReplaceAll(far, dbfE2EFar, dbfE2EHere), farOut, hereOut) != here { |
| 823 | t.Fatalf("at %q: in region %q, forwarded %q", at, here, far) |
| 824 | } |
| 825 | if !strings.Contains(farErr, fmt.Sprintf(dbfFetching, dbfE2EFar, "uk")) { |
| 826 | t.Fatalf("no DBF-18: %s", farErr) |
| 827 | } |
| 828 | db, _ := sql.Open("sqlite", "file:"+farOut+"?mode=ro") |
| 829 | var later int |
| 830 | err = db.QueryRow(`SELECT count(*) FROM orders WHERE note = 'later'`).Scan(&later) |
| 831 | db.Close() |
| 832 | if err != nil || (at == "") != (later == 1) { |
| 833 | t.Fatalf("at %q the copy holds %d of the second sync's rows (%v)", at, later, err) |
| 834 | } |
| 835 | } |
| 836 | } |
| 837 | |
| 838 | // T11 for a pull: a hash mismatch saves nothing at --out. Everything else |
| 839 | // is a pull the CLI would accept: far's real planned files, whole and |
| 840 | // valid, a manifest that passes checkPullManifest, and a summary whose only |
| 841 | // fault is its SHA-256 (plan review finding 32). Mutation: openDBPull skips |
| 842 | // the hash comparison (red: the files restore and a copy is saved). |
| 843 | func TestDBForwardPullSavesNothingForABadAnswer(t *testing.T) { |
| 844 | e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true}) |
| 845 | stop := e.scriptBox(t, func(b *e2eScript) { |
| 846 | ctx := context.Background() |
| 847 | rd := lsfile.NewReplicaClient(filepath.Join(e.root, "instances", dbfE2EFar)) |
| 848 | plan, err := dbdoor.PlanFrom(ctx, rd, time.Time{}) |
| 849 | if err != nil { |
| 850 | t.Errorf("plan far's replica: %v", err) |
| 851 | return |
| 852 | } |
| 853 | var all bytes.Buffer |
| 854 | for _, f := range plan.Planned { |
| 855 | rc, err := rd.OpenLTXFile(ctx, f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0) |
| 856 | if err != nil { |
| 857 | t.Errorf("open %+v: %v", f, err) |
| 858 | return |
| 859 | } |
| 860 | io.Copy(&all, rc) |
| 861 | rc.Close() |
| 862 | } |
| 863 | b.picked(time.Now().Add(time.Minute)) |
| 864 | sealed, _ := dbreads.SealChunk(b.pub, b.id, 1, dbreads.StreamLTX, true, all.Bytes()) |
| 865 | b.store.Put(ctx, dbreads.ChunkKey(b.id, 1), sealed, "") |
| 866 | wrong := sha256.Sum256([]byte("not these bytes")) |
| 867 | summary, _ := dbreads.SealSummary(b.pub, b.id, dbreads.Summary{TXID: plan.TXID, At: plan.At, Files: plan.Planned, |
| 868 | Bytes: int64(all.Len()), SHA256: hex.EncodeToString(wrong[:])}) |
| 869 | r, _ := json.Marshal(dbreads.Result{State: dbreads.StateDone, LastSeq: 1, Box: "uk-1", Sealed: summary}) |
| 870 | b.store.Put(ctx, dbreads.ResultKey(b.id), r, "") |
| 871 | }) |
| 872 | defer stop() |
| 873 | out := filepath.Join(t.TempDir(), "far.db") |
| 874 | _, stderr, err := e.run(t, "db", "pull", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--out", out) |
| 875 | if err == nil || !strings.Contains(stderr, fmt.Sprintf(dbfUnchecked, "uk")) { |
| 876 | t.Fatalf("%q %v", stderr, err) |
| 877 | } |
| 878 | if _, err := os.Stat(out); !os.IsNotExist(err) { |
| 879 | t.Fatalf("something was saved at --out: %v", err) |
| 880 | } |
| 881 | } |
| 882 | |
| 883 | // Review Focus 4 and plan review finding 23, through the binary: SIGINT |
| 884 | // while the CLI is spooling a pull prints "stopped", saves nothing at --out |
| 885 | // and removes the pull's directory. The scripted box (a stand-in) writes two |
| 886 | // chunks and never a result, so the CLI is mid-pull when the signal lands. |
| 887 | // Mutation: drop dbPullForward's deferred RemoveAll (red: carlos-db-* is |
| 888 | // left in the CLI's TMPDIR). |
| 889 | func TestDBForwardPullInterruptedLeavesNothing(t *testing.T) { |
| 890 | e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true}) |
| 891 | stop := e.scriptBox(t, func(b *e2eScript) { |
| 892 | b.picked(time.Now().Add(time.Minute)) |
| 893 | for seq := int64(1); seq <= 2; seq++ { |
| 894 | sealed, _ := dbreads.SealChunk(b.pub, b.id, seq, dbreads.StreamLTX, false, bytes.Repeat([]byte{'x'}, 4096)) |
| 895 | b.store.Put(context.Background(), dbreads.ChunkKey(b.id, seq), sealed, "") |
| 896 | } |
| 897 | }) |
| 898 | defer stop() |
| 899 | spool := t.TempDir() |
| 900 | out := filepath.Join(t.TempDir(), "far.db") |
| 901 | cmd := exec.Command(e.bin, "db", "pull", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--out", out) |
| 902 | // The last TMPDIR wins (os/exec keeps the last of a duplicated key). |
| 903 | cmd.Env = append(carlosTestEnv(t, append([]string{"CARLOS_CREDENTIALS=" + e.creds}, e.env...)...), "TMPDIR="+spool) |
| 904 | cmd.Dir = t.TempDir() |
| 905 | var stderr bytes.Buffer |
| 906 | cmd.Stderr = &stderr |
| 907 | if err := cmd.Start(); err != nil { |
| 908 | t.Fatal(err) |
| 909 | } |
| 910 | deadline := time.Now().Add(30 * time.Second) |
| 911 | for { |
| 912 | if got, _ := filepath.Glob(filepath.Join(spool, "carlos-db-*", "sealed", "2")); len(got) == 1 { |
| 913 | break |
| 914 | } |
| 915 | if time.Now().After(deadline) { |
| 916 | cmd.Process.Kill() |
| 917 | cmd.Wait() |
| 918 | t.Fatalf("the CLI never spooled two chunks: %s", stderr.String()) |
| 919 | } |
| 920 | time.Sleep(50 * time.Millisecond) |
| 921 | } |
| 922 | if err := cmd.Process.Signal(os.Interrupt); err != nil { |
| 923 | t.Fatal(err) |
| 924 | } |
| 925 | if err := cmd.Wait(); err == nil || !strings.Contains(stderr.String(), "stopped") { |
| 926 | t.Fatalf("%v: %s", err, stderr.String()) |
| 927 | } |
| 928 | if _, err := os.Stat(out); !os.IsNotExist(err) { |
| 929 | t.Fatalf("something was saved at --out: %v", err) |
| 930 | } |
| 931 | if left, _ := filepath.Glob(filepath.Join(spool, "carlos-db-*")); len(left) != 0 { |
| 932 | t.Fatalf("the pull's directory is still there: %v", left) |
| 933 | } |
| 934 | } |
| 935 | |
| 936 | // A console from plan 2 forwards queries and not pulls. Its marker is |
| 937 | // played here by a wrapper that drops "pull" from forward_ops on the way |
| 938 | // out: the new CLI says DBF-8 and saves nothing. The wrapper buffers each |
| 939 | // response, so it serves only where no response streams (the list door's |
| 940 | // 409 is all this test reads). |
| 941 | func TestDBForwardPullOnAQueryOnlyConsole(t *testing.T) { |
| 942 | queryOnly := func(t *testing.T, h http.Handler) (string, []string) { |
| 943 | ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 944 | rec := httptest.NewRecorder() |
| 945 | h.ServeHTTP(rec, r) |
| 946 | body := rec.Body.Bytes() |
| 947 | if rec.Code == http.StatusConflict { |
| 948 | body = bytes.Replace(body, []byte(`"forward_ops":["query","pull"]`), []byte(`"forward_ops":["query"]`), 1) |
| 949 | } |
| 950 | for k, vs := range rec.Header() { |
| 951 | w.Header()[k] = vs |
| 952 | } |
| 953 | w.Header().Del("Content-Length") |
| 954 | w.WriteHeader(rec.Code) |
| 955 | w.Write(body) |
| 956 | })) |
| 957 | t.Cleanup(ts.Close) |
| 958 | return ts.URL, nil |
| 959 | } |
| 960 | e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true, serve: queryOnly}) |
| 961 | out := filepath.Join(t.TempDir(), "far.db") |
| 962 | _, stderr, err := e.run(t, "db", "pull", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--out", out) |
| 963 | if err == nil || !strings.Contains(stderr, fmt.Sprintf(dbfNoPull, dbfE2EFar, "uk")) { |
| 964 | t.Fatalf("%q %v", stderr, err) |
| 965 | } |
| 966 | if _, err := os.Stat(out); !os.IsNotExist(err) { |
| 967 | t.Fatalf("something was saved at --out: %v", err) |
| 968 | } |
| 969 | } |
| 970 | |
amadan