amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1//go:build linux
2
3package host
4
5import (
6 "bytes"
7 "context"
8 "crypto/ecdh"
9 "crypto/sha256"
10 "database/sql"
11 "encoding/hex"
12 "encoding/json"
13 "errors"
14 "fmt"
15 "net/http/httptest"
16 "os"
17 "path/filepath"
18 "strings"
19 "syscall"
20 "testing"
21 "time"
22
23 "github.com/benbjohnson/litestream"
24 "github.com/superfly/ltx"
25
26 "github.com/carlosframework/platform/internal/activator"
27 "github.com/carlosframework/platform/internal/appconfig"
28 "github.com/carlosframework/platform/internal/instancekey"
29 "github.com/carlosframework/platform/internal/layout"
30 "github.com/carlosframework/platform/internal/lease"
31 "github.com/carlosframework/platform/internal/lease/leasetest"
32 "github.com/carlosframework/platform/internal/moves"
33 "github.com/carlosframework/platform/internal/moves/archive"
34 "github.com/carlosframework/platform/internal/provision"
35 "github.com/carlosframework/platform/internal/registry"
36 "github.com/carlosframework/platform/internal/release"
37)
38
39// The sealer (Task 4 of docs/superpowers/plans/2026-10-20-offline-operator-
40// move.md), tested through offlineRound, the moves pass's dispatch, over
41// the S3 wire protocol (leasetest.NewS3Server, with buckets data-ie and
42// data-uk behind one endpoint), real litestream replicas written by
43// lease.ReplicateWithLease, the real lease package, a real registry and
44// activator per box, and the production edges-file writer for the target's
45// coordinates. The deployment store is T3's DirStore world.
46
47const (
48 sealIE = "data-ie"
49 sealUK = "data-uk"
50 sealIERegion = "eu-west-1"
51 sealUKRegion = "eu-west-2"
52 sealPrefix = "instances"
53 sealVersion = "v42"
54)
55
56type sealWorld struct {
57 *offWorld
58 fake *leasetest.VersionedBucket
59 keyB64 string
60 priv *ecdh.PrivateKey
61 ie, uk *offBox
62 pages []string
63 bodies []string // each page's body, beside pages
64}
65
66func newSealWorld(t *testing.T) *sealWorld {
67 t.Helper()
68 fake := leasetest.NewVersionedBucket()
69 fake.AddBucket(sealIE)
70 fake.AddBucket(sealUK)
71 h, _ := leasetest.NewS3Server(fake)
72 srv := httptest.NewServer(h) // 127.0.0.1 only
73 t.Cleanup(srv.Close)
74 t.Setenv("AWS_ACCESS_KEY_ID", "test-key")
75 t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret")
76 t.Setenv("AWS_SESSION_TOKEN", "")
77 t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
78 t.Setenv("AWS_CONFIG_FILE", filepath.Join(t.TempDir(), "absent"))
79 t.Setenv("AWS_SHARED_CREDENTIALS_FILE", filepath.Join(t.TempDir(), "absent"))
80 t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)
81 keyB64, _, err := appconfig.GenerateKey()
82 if err != nil {
83 t.Fatal(err)
84 }
85 priv, err := appconfig.ParsePrivateKey(keyB64, "test key")
86 if err != nil {
87 t.Fatal(err)
88 }
89 sw := &sealWorld{offWorld: newOffWorld(t), fake: fake, keyB64: keyB64, priv: priv}
90 sw.ie = sw.sourceBox("ie-a")
91 sw.uk = sw.box("uk-box", "uk", map[string]string{"CARLOS_HIBERNATE_BUCKET": sealUK, "CARLOS_SECRETS_KEY": keyB64})
92 var last []byte
93 if err := edgesFileWrite(context.Background(), sw.s, sw.uk.reg, edgesFile{Box: "uk-box", Placement: "uk", DataBucket: sealUK, DataBucketRegion: sealUKRegion}, "", &last, false); err != nil {
94 t.Fatal(err)
95 }
96 return sw
97}
98
99// sourceBox is a home box armed as armMovesPass arms one: armOffline, then
100// the hibernate coordinates, the S3 mirror, the production checkpoint and
101// the box's exec backend for the launch environment.
102func (sw *sealWorld) sourceBox(name string) *offBox {
103 b := sw.box(name, "", map[string]string{"CARLOS_HIBERNATE_BUCKET": sealIE, "CARLOS_SECRETS_KEY": sw.keyB64})
104 sw.arm(b)
105 return b
106}
107
108func (sw *sealWorld) arm(b *offBox) {
109 c := &b.w.cfg
110 c.OwnBucket, c.OwnBucketRegion, c.OwnPrefix = sealIE, sealIERegion, sealPrefix
111 c.Mirror = func(ctx context.Context, dst string) (lease.MirrorStore, error) {
112 return lease.NewS3Mirror(ctx, sealIERegion, dst)
113 }
114 // armMovesPass's Checkpoint, as it is wired there.
115 c.Checkpoint = func(ctx context.Context, dbPath string) error {
116 host := strings.TrimSuffix(filepath.Base(dbPath), ".db")
117 rep, err := lease.ReplicateWithLease(ctx, dbPath, sealIE, activator.HibernatePrefix(sealPrefix, host), sealIERegion)
118 if err != nil {
119 return err
120 }
121 defer rep.Close(ctx)
122 return rep.SyncAndWait(ctx)
123 }
124 c.CarryRequest = func(ctx context.Context, host, bucket, prefix string) error {
125 sw.fake.Put(bucket, prefix+"/build-state.json", []byte(`{"request":"carried"}`))
126 return nil
127 }
128 c.LaunchLayers = (&activator.ExecBackend{DataDir: b.dd}).LaunchLayers
129 c.Page = func(_ context.Context, level, subject, body string) error {
130 sw.pages = append(sw.pages, level+": "+subject)
131 sw.bodies = append(sw.bodies, body)
132 return nil
133 }
134}
135
136// pass is a tick in the agent's order (reconcile, the edges file, the moves
137// pass) whose moves pass is movesPass's offline part: the keep view's
138// refresh from the record list, then offlineRound, returning the round's
139// error rather than logging it.
140func (sw *sealWorld) pass(b *offBox) error {
141 sw.t.Helper()
142 b.reconcile()
143 b.edges()
144 ctx := context.Background()
145 refreshOfflineView(ctx, sw.s, b.w.cfg.Offline)
146 return b.w.offlineRound(ctx, sw.s, sw.record())
147}
148
149func (sw *sealWorld) mustPass(b *offBox) {
150 sw.t.Helper()
151 if err := sw.pass(b); err != nil {
152 sw.t.Fatal(err)
153 }
154}
155
156func (sw *sealWorld) dbPath(b *offBox) string {
157 sw.t.Helper()
158 rt := b.row(offHost)
159 if rt == nil {
160 sw.t.Fatal("no row")
161 }
162 return rt.DBPath
163}
164
165// instance gives b a live instance of offHost, as a box that has served it
166// has: the row (minted ready by reconcile), its replication rung, next due
167// and timers, a database written through lease.ReplicateWithLease into
168// data-ie, the files directory, <host>.env with an instance key, the
169// rendered configuration and a live build.
170func (sw *sealWorld) instance(b *offBox) {
171 t := sw.t
172 t.Helper()
173 b.tick()
174 rt := b.row(offHost)
175 if rt == nil || rt.HoldOp != "" {
176 t.Fatalf("fixture: row %+v", rt)
177 }
178 if err := b.reg.SetReplication(offHost, registry.ReplicationBatched); err != nil {
179 t.Fatal(err)
180 }
181 if err := b.reg.SetNextDue(offHost, 1_800_000_000); err != nil {
182 t.Fatal(err)
183 }
184 for _, tm := range []registry.Timer{
185 {Host: offHost, Name: "nightly", Kind: "cron", Spec: "0 3 * * *", Path: "/jobs/nightly", NextAt: 1_800_000_100},
186 {Host: offHost, Name: "reminder", Kind: "once", Path: "/jobs/remind", NextAt: 1_800_000_200},
187 {Host: offHost, Name: "fired", Kind: "once", Path: "/jobs/fired", LastAt: 1_700_000_000, LastStatus: "200"},
188 } {
189 if err := b.reg.UpsertTimer(tm); err != nil {
190 t.Fatal(err)
191 }
192 }
193 writeFile(t, filepath.Join(b.dd, offHost, "instance.key"), "app key file")
194 writeFile(t, filepath.Join(b.dd, offHost, "uploads", "a.png"), "png bytes")
195 writeFile(t, filepath.Join(b.dd, offHost+".env"), "CARLOS_INSTANCE_KEY=the-instance-key\nCARLOS_ADMIN_TOKEN=token\n")
196 writeFile(t, filepath.Join(b.dd, offHost+".config.env"), "APP_COLOUR=green\n")
197 sw.writeRows(b, rt.DBPath, 0, 20, true)
198 versions := filepath.Join(b.opt, "versions", sealVersion, sw.acct, "app")
199 if err := os.MkdirAll(versions, 0o755); err != nil {
200 t.Fatal(err)
201 }
202 if err := os.MkdirAll(filepath.Dir(release.LiveLink(b.opt, offHost)), 0o755); err != nil {
203 t.Fatal(err)
204 }
205 if err := os.Symlink(versions, release.LiveLink(b.opt, offHost)); err != nil {
206 t.Fatal(err)
207 }
208}
209
210// writeRows inserts rows [from, to) into the database at dbPath, then, with
211// replicate, syncs it to data-ie under the lease as a wake's stop does.
212func (sw *sealWorld) writeRows(b *offBox, dbPath string, from, to int, replicate bool) {
213 t := sw.t
214 t.Helper()
215 db, err := sql.Open("sqlite", dbPath)
216 if err != nil {
217 t.Fatal(err)
218 }
219 for _, q := range []string{`PRAGMA journal_mode=WAL`, `CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, body TEXT)`} {
220 if _, err := db.Exec(q); err != nil {
221 t.Fatal(err)
222 }
223 }
224 for i := from; i < to; i++ {
225 if _, err := db.Exec(`INSERT INTO t (id, body) VALUES (?, ?)`, i, fmt.Sprintf("row-%d", i)); err != nil {
226 t.Fatal(err)
227 }
228 }
229 if err := db.Close(); err != nil {
230 t.Fatal(err)
231 }
232 if !replicate {
233 return
234 }
235 ctx := context.Background()
236 rep, err := lease.ReplicateWithLease(ctx, dbPath, sealIE, activator.HibernatePrefix(sealPrefix, offHost), sealIERegion)
237 if err != nil {
238 t.Fatal(err)
239 }
240 if err := rep.SyncAndWait(ctx); err != nil {
241 t.Fatal(err)
242 }
243 if err := rep.Close(ctx); err != nil {
244 t.Fatal(err)
245 }
246}
247
248func countRows(t *testing.T, dbPath string) int {
249 t.Helper()
250 db, err := sql.Open("sqlite", dbPath)
251 if err != nil {
252 t.Fatal(err)
253 }
254 defer db.Close()
255 var n int
256 if err := db.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil {
257 t.Fatal(err)
258 }
259 return n
260}
261
262// start moves offHost to the record's offline-stopping, ticks every source
263// box until it holds and reports, and names the sealer and the boxes as the
264// seal door does.
265func (sw *sealWorld) startAndSeal(sealer string, boxes ...*offBox) {
266 sw.t.Helper()
267 sw.start()
268 names := []string{}
269 for _, b := range boxes {
270 b.tick()
271 names = append(names, b.name)
272 }
273 sw.move(provision.OfflineSeal(offID, sealer, names))
274}
275
276func (sw *sealWorld) mirror() lease.MirrorStore {
277 sw.t.Helper()
278 m, err := lease.NewS3Mirror(context.Background(), sealIERegion, sealUKRegion)
279 if err != nil {
280 sw.t.Fatal(err)
281 }
282 return m
283}
284
285func (sw *sealWorld) list(bucket, prefix string) map[string]lease.MirrorObject {
286 sw.t.Helper()
287 objs, err := sw.mirror().List(context.Background(), bucket, prefix)
288 if err != nil {
289 sw.t.Fatal(err)
290 }
291 return objs
292}
293
294func (sw *sealWorld) lock() (owner string, expires time.Time, generation int64) {
295 sw.t.Helper()
296 body, ok := sw.fake.Current(sealIE, sealPrefix+"/"+offHost+"/lock.json")
297 if !ok {
298 return "", time.Time{}, 0
299 }
300 var l litestream.Lease
301 if err := json.Unmarshal(body, &l); err != nil {
302 sw.t.Fatal(err)
303 }
304 return l.Owner, l.ExpiresAt, l.Generation
305}
306
307func (sw *sealWorld) sealRecord() (moves.OfflineReport, []byte, bool) {
308 sw.t.Helper()
309 raw, _, err := sw.s.Get(context.Background(), moves.OfflineSealKey(offHost, offID))
310 if errors.Is(err, os.ErrNotExist) {
311 return moves.OfflineReport{}, nil, false
312 }
313 if err != nil {
314 sw.t.Fatal(err)
315 }
316 var r moves.OfflineReport
317 if err := json.Unmarshal(raw, &r); err != nil {
318 sw.t.Fatal(err)
319 }
320 return r, raw, true
321}
322
323// acquireAsAWake is what every wake does first (ExecBackend.Acquire).
324func acquireAsAWake(ctx context.Context) (*lease.Held, error) {
325 return lease.Acquire(ctx, sealIE, activator.HibernatePrefix(sealPrefix, offHost), sealIERegion)
326}
327
328func wantHeldByMove(t *testing.T, err error, what string) {
329 t.Helper()
330 var he *lease.HeldError
331 if !errors.As(err, &he) || he.Owner != moves.TombstoneOwner(offID) {
332 t.Fatalf("%s: %v, want a HeldError naming %s", what, err, moves.TombstoneOwner(offID))
333 }
334}
335
336// openArchive fetches the sealed archive and its manifest from data-uk,
337// verifies it and extracts it into a fresh flat files directory and env
338// path, returning them.
339func (sw *sealWorld) openArchive(seal moves.OfflineReport) (files, env string) {
340 t := sw.t
341 t.Helper()
342 prefix := moves.OfflineArchivePrefix(offHost, offID)
343 sealed, ok := sw.fake.Current(sealUK, prefix+offlineArchiveSealed)
344 if !ok {
345 t.Fatal("no sealed archive")
346 }
347 rawManifest, ok := sw.fake.Current(sealUK, prefix+offlineArchiveManifest)
348 if !ok {
349 t.Fatal("no archive manifest")
350 }
351 var m archive.Manifest
352 if err := json.Unmarshal(rawManifest, &m); err != nil {
353 t.Fatal(err)
354 }
355 bind := appconfig.Binding{Account: sw.acct, App: "app"}
356 if err := archive.Verify(bytes.NewReader(sealed), m, sw.priv, bind); err != nil {
357 t.Fatalf("Verify: %v", err)
358 }
359 sum := sha256.Sum256(sealed)
360 if m.Plain != seal.FilesPlain || m.Sealed != seal.FilesSealed || "sha256:"+hex.EncodeToString(sum[:]) != seal.FilesSealed || m.Bytes != seal.FilesBytes || len(m.Entries) != seal.FilesEntries {
361 t.Fatalf("manifest %+v against the seal %+v", m, seal)
362 }
363 var plain bytes.Buffer
364 if err := archive.Open(bytes.NewReader(sealed), m.SealedDataKey, sw.priv, bind, &plain); err != nil {
365 t.Fatalf("Open: %v", err)
366 }
367 parent := t.TempDir()
368 files, env = filepath.Join(parent, offHost), filepath.Join(parent, offHost+".env")
369 // A *bytes.Reader, not the buffer: zstd decodes a *bytes.Buffer under
370 // 128 KiB from its Bytes() without draining it, and Extract's
371 // end-of-stream check then finds the whole archive still unread.
372 if err := archive.Extract(bytes.NewReader(plain.Bytes()), files, env); err != nil {
373 t.Fatalf("Extract: %v", err)
374 }
375 if got, _, _, err := archive.PlainDigest(files, env); err != nil || got != seal.FilesPlain {
376 t.Fatalf("the extraction digests to %s (%v), the seal says %s", got, err, seal.FilesPlain)
377 }
378 return files, env
379}
380
381func readString(t *testing.T, p string) string {
382 t.Helper()
383 b, err := os.ReadFile(p)
384 if err != nil {
385 t.Fatal(err)
386 }
387 return string(b)
388}
389
390// (a) The happy seal, end to end, and the passes after it.
391func TestOfflineSealHappyPath(t *testing.T) {
392 sw := newSealWorld(t)
393 ie := sw.ie
394 sw.instance(ie)
395 // A tail the stop never checkpointed: rows in the local database only.
396 sw.writeRows(ie, sw.dbPath(ie), 20, 25, false)
397 sw.startAndSeal("ie-a", ie)
398 ctx := context.Background()
399
400 sw.mustPass(ie)
401 rec := sw.record()
402 if rec.Move.State != provision.OfflineSealed {
403 t.Fatalf("state %s, want sealed", rec.Move.State)
404 }
405 seal, raw, ok := sw.sealRecord()
406 if !ok {
407 t.Fatal("no seal record")
408 }
409 if err := moves.ValidateOfflineSeal(seal, offHost, offID); err != nil {
410 t.Fatal(err)
411 }
412 src := rec.Move.Source
413 if src == nil || src.Evidence != moves.OfflineSealEvidence(offHost, offID, raw) || src.FinalTXID != seal.FinalTXID ||
414 src.Files == nil || src.Files.Plain != seal.FilesPlain || src.Files.Sealed != seal.FilesSealed || src.TimersSHA256 != seal.TimersDigest ||
415 src.Release != sealVersion || src.EnvDigest != seal.EnvDigest || src.Replication != registry.ReplicationBatched || src.NextDue != 1_800_000_000 {
416 t.Fatalf("record source %+v, seal %+v", src, seal)
417 }
418
419 // The tombstone, and a wake meets it.
420 owner, expires, _ := sw.lock()
421 if owner != moves.TombstoneOwner(offID) || time.Until(expires) < moves.TombstoneTTL-time.Minute {
422 t.Fatalf("lock %s until %s", owner, expires)
423 }
424 if h, err := acquireAsAWake(ctx); err == nil {
425 h.Release(ctx)
426 t.Fatal("a wake took the lease after the seal")
427 } else {
428 wantHeldByMove(t, err, "a wake after the seal")
429 }
430
431 // The staged copy is the replica, object for object, and its listing is
432 // the seal's.
433 srcObjs := sw.list(sealIE, sealPrefix+"/"+offHost+"/")
434 staged := sw.list(sealUK, activator.MovesPrefix(offHost))
435 if len(staged) != len(seal.Staged) || len(staged) != len(srcObjs)-1 {
436 t.Fatalf("staged %d objects, seal %d, source %d (lock.json included)", len(staged), len(seal.Staged), len(srcObjs))
437 }
438 for k, o := range staged {
439 rel := strings.TrimPrefix(k, activator.MovesPrefix(offHost))
440 if s := seal.Staged[rel]; s.Size != o.Size || s.ETag != o.ETag {
441 t.Fatalf("%s staged %+v, sealed %+v", rel, o, s)
442 }
443 if so := srcObjs[sealPrefix+"/"+offHost+"/"+rel]; so.Size != o.Size {
444 t.Fatalf("%s staged %+v, source %+v", rel, o, so)
445 }
446 }
447 if _, ok := seal.Staged["build-state.json"]; !ok {
448 t.Fatal("the carried build state is not staged")
449 }
450 if replicaPosition(staged, activator.MovesPrefix(offHost)) != seal.FinalTXID {
451 t.Fatal("the staged newest TXID is not the seal's")
452 }
453 // The staged replica restores to every row, the tail included: the
454 // checkpoint ran before the tombstone.
455 restored := filepath.Join(t.TempDir(), "restored.db")
456 if ok, err := lease.Restore(ctx, restored, sealUK, strings.TrimSuffix(activator.MovesPrefix(offHost), "/"), sealUKRegion); err != nil || !ok {
457 t.Fatalf("restore the staged copy: %v %v", ok, err)
458 }
459 if n := countRows(t, restored); n != 25 {
460 t.Fatalf("the staged copy holds %d rows, want 25", n)
461 }
462
463 // The archive opens, verifies and extracts to the same plaintext, with
464 // the files and the env byte for byte.
465 files, env := sw.openArchive(seal)
466 for rel, want := range map[string]string{"instance.key": "app key file", "uploads/a.png": "png bytes"} {
467 if got := readString(t, filepath.Join(files, rel)); got != want {
468 t.Fatalf("%s = %q", rel, got)
469 }
470 }
471 envBody := readString(t, filepath.Join(ie.dd, offHost+".env"))
472 if readString(t, env) != envBody {
473 t.Fatal("the env file did not travel byte for byte")
474 }
475 sum := sha256.Sum256([]byte(envBody))
476 if seal.EnvSHA256 != hex.EncodeToString(sum[:]) || seal.KeySHA256 != instancekey.Digest("the-instance-key") {
477 t.Fatalf("env %s key %s", seal.EnvSHA256, seal.KeySHA256)
478 }
479
480 // The timers snapshot (active rows only) and the env report.
481 prefix := moves.OfflineArchivePrefix(offHost, offID)
482 timersBody, _ := sw.fake.Current(sealUK, prefix+offlineTimersObject)
483 snap, err := moves.DecodeTimers(timersBody)
484 if err != nil || len(snap.Rows) != 2 || moves.TimersDigest(snap.Rows) != seal.TimersDigest {
485 t.Fatalf("timers snapshot %+v %v", snap, err)
486 }
487 reportBody, _ := sw.fake.Current(sealUK, prefix+offlineEnvReportObject)
488 var envReport moves.EnvReport
489 if err := json.Unmarshal(reportBody, &envReport); err != nil || envReport.Digest != seal.EnvDigest || envReport.Vars["APP_COLOUR"] == "" || strings.Contains(string(reportBody), "green") {
490 t.Fatalf("env report %s %v", reportBody, err)
491 }
492 if seal.Release != sealVersion || seal.Replication != registry.ReplicationBatched || seal.NextDue != 1_800_000_000 {
493 t.Fatalf("seal %+v", seal)
494 }
495
496 // The living report: the tombstone and the pass's time.
497 living := sw.report(moves.OfflineRoleSealer, "ie-a")
498 if living.Phase != moves.OfflinePhaseSealed || living.TombstoneOwner != moves.TombstoneOwner(offID) || !living.TombstoneUntil.Equal(expires) || !living.At.Equal(sw.clock()) {
499 t.Fatalf("living report %+v", living)
500 }
501 if ok, bad := living.SourceItemsOK(); !ok {
502 t.Fatalf("living report items %v", bad)
503 }
504
505 // Later passes re-assert the expiry, keep the tombstone's generation,
506 // never touch the seal record, and rewrite the living report.
507 _, _, gen := sw.lock()
508 for _, state := range []func(){func() {}, sw.flip, sw.landed} {
509 state()
510 time.Sleep(5 * time.Millisecond)
511 sw.advance(time.Second)
512 if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
513 t.Fatal(err)
514 }
515 owner2, expires2, gen2 := sw.lock()
516 if owner2 != owner || !expires2.After(expires) || gen2 != gen {
517 t.Fatalf("after a later pass: %s until %s (generation %d), before %s until %s (generation %d)", owner2, expires2, gen2, owner, expires, gen)
518 }
519 expires = expires2
520 living = sw.report(moves.OfflineRoleSealer, "ie-a")
521 if !living.TombstoneUntil.Equal(expires2) || !living.At.Equal(sw.clock()) || living.Phase != moves.OfflinePhaseSealed {
522 t.Fatalf("living report after a later pass: %+v", living)
523 }
524 if _, again, _ := sw.sealRecord(); !bytes.Equal(again, raw) {
525 t.Fatal("the seal record was rewritten")
526 }
527 }
528 if len(sw.pages) != 0 {
529 t.Fatalf("paged: %v", sw.pages)
530 }
531}
532
533// (b) A live writer's lease: with a local database the checkpoint cannot
534// take it, without one the tombstone cannot; either way the state stays
535// sealing and nothing is staged or sealed. A wake that races the tombstone
536// meets it.
537func TestOfflineSealWaitsForALiveWriter(t *testing.T) {
538 ctx := context.Background()
539 for _, local := range []bool{true, false} {
540 t.Run(fmt.Sprintf("local database %v", local), func(t *testing.T) {
541 sw := newSealWorld(t)
542 ie := sw.ie
543 sw.instance(ie)
544 if !local {
545 db := sw.dbPath(ie)
546 for _, p := range []string{db, db + "-wal", db + "-shm", activatorMetaDir(db)} {
547 if err := os.RemoveAll(p); err != nil {
548 t.Fatal(err)
549 }
550 }
551 }
552 sw.startAndSeal("ie-a", ie)
553 wake, err := acquireAsAWake(ctx)
554 if err != nil {
555 t.Fatal(err)
556 }
557 err = sw.pass(ie)
558 var he *lease.HeldError
559 if !errors.As(err, &he) || he.Owner == moves.TombstoneOwner(offID) {
560 t.Fatalf("a live writer: %v", err)
561 }
562 if !local && !strings.Contains(err.Error(), "tombstone") {
563 t.Fatalf("not refused at the tombstone: %v", err)
564 }
565 if sw.record().Move.State != provision.OfflineSealing {
566 t.Fatal("the state moved")
567 }
568 if _, _, ok := sw.sealRecord(); ok {
569 t.Fatal("sealed under a live writer")
570 }
571 if n := len(sw.list(sealUK, "moves/")); n != 0 {
572 t.Fatalf("%d objects staged under a live writer", n)
573 }
574 if err := wake.Err(); err != nil {
575 t.Fatalf("the writer lost its lease: %v", err)
576 }
577 if err := wake.Release(ctx); err != nil {
578 t.Fatal(err)
579 }
580
581 // The race: a wake that comes for the lease the moment after
582 // the tombstone is written is refused by it.
583 raced := false
584 offlineSealFaults.Store(ie.w, func(point string) error {
585 if point == "tombstone" {
586 raced = true
587 h, err := acquireAsAWake(ctx)
588 if err == nil {
589 h.Release(ctx)
590 return errors.New("a wake took the lease after the tombstone")
591 }
592 wantHeldByMove(t, err, "a wake racing the tombstone")
593 }
594 return nil
595 })
596 t.Cleanup(func() { offlineSealFaults.Delete(ie.w) })
597 sw.mustPass(ie)
598 if !raced || sw.record().Move.State != provision.OfflineSealed {
599 t.Fatalf("raced %v, state %s", raced, sw.record().Move.State)
600 }
601 })
602 }
603}
604
605// (c) A source box whose local position is past the replica's final
606// position refuses the seal: the db-tail item fails, the report says
607// failed, nothing is sealed and the state stays sealing.
608func TestOfflineSealRefusesASourceAheadOfTheFinalPosition(t *testing.T) {
609 sw := newSealWorld(t)
610 ie := sw.ie
611 sw.instance(ie)
612 ieB := sw.sourceBox("ie-b")
613 ieB.tick()
614 final := replicaPosition(sw.list(sealIE, sealPrefix+"/"+offHost+"/"), sealPrefix+"/"+offHost+"/")
615 if final == 0 {
616 t.Fatal("fixture: no replica")
617 }
618 // ie-b woke the host once and kept a position past the replica's.
619 meta := filepath.Join(activatorMetaDir(provision.DefaultDBPath(ieB.dd, offHost)), "ltx", "0")
620 writeFile(t, filepath.Join(meta, ltxName(final+1, final+5)), "ltx")
621 sw.startAndSeal("ie-a", ie, ieB)
622 if r := sw.report(moves.OfflineRoleSource, "ie-b"); r.LocalTXID != final+5 {
623 t.Fatalf("fixture: ie-b reports %d", r.LocalTXID)
624 }
625 err := sw.pass(ie)
626 var ref *sealRefusal
627 if !errors.As(err, &ref) || ref.item != "db-tail" || !strings.Contains(ref.Reason, "ie-b") {
628 t.Fatalf("pass: %v", err)
629 }
630 r := sw.report(moves.OfflineRoleSealer, "ie-a")
631 if r.Phase != moves.OfflinePhaseFailed || r.Items["db-tail"].OK || !r.Items["sidecar"].OK {
632 t.Fatalf("report %+v", r)
633 }
634 if _, _, ok := sw.sealRecord(); ok || sw.record().Move.State != provision.OfflineSealing {
635 t.Fatal("sealed past a source ahead of the replica")
636 }
637}
638
639func ltxName(min, max uint64) string {
640 return ltx.FormatFilename(ltx.TXID(min), ltx.TXID(max))
641}
642
643// (d) After the seal the stop is kept, never taken again: a tombstone
644// stolen by another move, expired, lifted or deleted is a lost stop. The
645// pass reports failed with the lease item down, pages once for the move,
646// and leaves the lease exactly as it found it.
647func TestOfflineSealLostStopIsNeverRetaken(t *testing.T) {
648 sw := newSealWorld(t)
649 ie := sw.ie
650 sw.instance(ie)
651 sw.startAndSeal("ie-a", ie)
652 sw.mustPass(ie)
653 ctx := context.Background()
654 key := sealPrefix + "/" + offHost + "/lock.json"
655 ours, _ := sw.fake.Current(sealIE, key)
656 lockBody := func(owner string, expires time.Time) []byte {
657 b, err := json.Marshal(litestream.Lease{Generation: 1, Owner: owner, ExpiresAt: expires})
658 if err != nil {
659 t.Fatal(err)
660 }
661 return b
662 }
663 cases := []struct {
664 name string
665 damage func()
666 }{
667 {"stolen by another move", func() {
668 sw.fake.Put(sealIE, key, lockBody(moves.TombstoneOwner(offID2), time.Now().Add(time.Hour)))
669 }},
670 {"stolen by a wake", func() {
671 sw.fake.Put(sealIE, key, lockBody("box-x", time.Now().Add(time.Hour)))
672 }},
673 {"expired", func() {
674 sw.fake.Put(sealIE, key, lockBody(moves.TombstoneOwner(offID), time.Now().Add(-time.Second)))
675 }},
676 {"lifted", func() {
677 if _, err := lease.LiftTombstone(ctx, sealIE, sealPrefix+"/"+offHost, sealIERegion, func(o string) bool { return o == moves.TombstoneOwner(offID) }); err != nil {
678 t.Fatal(err)
679 }
680 }},
681 {"deleted", func() { sw.fake.BareDelete(sealIE, key) }},
682 }
683 for _, c := range cases {
684 // The stop holds: a pass is clean.
685 sw.fake.Put(sealIE, key, ours)
686 if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
687 t.Fatalf("%s: before: %v", c.name, err)
688 }
689 c.damage()
690 before := sw.fake.Snapshot()
691 for i := range 2 {
692 err := ie.w.offlineSealerStep(ctx, sw.s, sw.record())
693 if err == nil || !strings.Contains(err.Error(), "lost stop") {
694 t.Fatalf("%s, pass %d: %v", c.name, i, err)
695 }
696 if sw.fake.Snapshot() != before {
697 t.Fatalf("%s, pass %d: the lease was written after the stop was lost", c.name, i)
698 }
699 r := sw.report(moves.OfflineRoleSealer, "ie-a")
700 if r.Phase != moves.OfflinePhaseFailed || r.Items["lease"].OK || !strings.Contains(r.Items["lease"].Reason, "lost stop") || r.TombstoneOwner != "" || !r.Items["files"].OK {
701 t.Fatalf("%s: report %+v", c.name, r)
702 }
703 }
704 }
705 if len(sw.pages) != 1 || !strings.HasPrefix(sw.pages[0], "incident: ") {
706 t.Fatalf("pages %v, want one incident", sw.pages)
707 }
708 if sw.record().Move.State != provision.OfflineSealed {
709 t.Fatal("the record moved")
710 }
711}
712
713// (e) A crash after each step, then the next pass: one tombstone (one
714// generation, created once), staged objects copied once and identical, the
715// archive the seal names, and a sealed record.
716func TestOfflineSealReplaysAfterACrashAtEachStep(t *testing.T) {
717 sw := newSealWorld(t)
718 ie := sw.ie
719 sw.instance(ie)
720 sw.writeRows(ie, sw.dbPath(ie), 20, 23, false)
721 sw.startAndSeal("ie-a", ie)
722 ctx := context.Background()
723 points := []string{"checkpoint", "tombstone", "final", "mirror", "archive", "timers", "env", "seal", "cas"}
724 t.Cleanup(func() { offlineSealFaults.Delete(ie.w) })
725 var gen int64
726 var stagedAfter map[string]lease.MirrorObject
727 for i, crash := range points {
728 hit := false
729 offlineSealFaults.Store(ie.w, func(point string) error {
730 if point == crash {
731 hit = true
732 return errors.New("crash at " + point)
733 }
734 return nil
735 })
736 err := sw.pass(ie)
737 if !hit || err == nil || !strings.Contains(err.Error(), "crash at "+crash) {
738 t.Fatalf("crash at %s: hit %v, %v", crash, hit, err)
739 }
740 if i >= 1 { // the tombstone is down from the crash after it on
741 owner, _, g := sw.lock()
742 if owner != moves.TombstoneOwner(offID) || (gen != 0 && g != gen) {
743 t.Fatalf("after the crash at %s: lock %s generation %d (was %d)", crash, owner, g, gen)
744 }
745 gen = g
746 }
747 if crash == "mirror" {
748 stagedAfter = sw.list(sealUK, activator.MovesPrefix(offHost))
749 }
750 wantState := provision.OfflineSealing
751 if crash == "cas" {
752 wantState = provision.OfflineSealed
753 }
754 if s := sw.record().Move.State; s != wantState {
755 t.Fatalf("after the crash at %s: state %s", crash, s)
756 }
757 }
758 offlineSealFaults.Delete(ie.w)
759 sw.mustPass(ie)
760 if sw.record().Move.State != provision.OfflineSealed {
761 t.Fatal("not sealed after the replays")
762 }
763 if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseSealed || r.TombstoneOwner != moves.TombstoneOwner(offID) {
764 t.Fatalf("living report %+v", r)
765 }
766 // One tombstone: created once (one version with the move's owner and
767 // no expiry change before the seal), one generation throughout.
768 if _, _, g := sw.lock(); g != gen {
769 t.Fatalf("generation %d, was %d", g, gen)
770 }
771 versions, err := sw.fake.ListVersions(ctx, sealIE, sealPrefix+"/"+offHost+"/lock.json")
772 if err != nil {
773 t.Fatal(err)
774 }
775 created := 0
776 for _, v := range versions {
777 if v.DeleteMarker {
778 continue
779 }
780 body, err := sw.fake.GetVersion(ctx, sealIE, v.Key, v.VersionID)
781 if err != nil {
782 t.Fatal(err)
783 }
784 var l litestream.Lease
785 if err := json.Unmarshal(body, &l); err == nil && l.Owner == moves.TombstoneOwner(offID) && l.Generation != gen {
786 t.Fatalf("a tombstone of another generation %d", l.Generation)
787 }
788 if l.Owner == moves.TombstoneOwner(offID) {
789 created++
790 }
791 }
792 // The tombstone is written once before the seal (the replays after it
793 // find it this move's and write nothing); after the seal the re-asserts
794 // of the last two passes rewrite it, same generation.
795 if created != 3 {
796 t.Fatalf("%d versions of the tombstone", created)
797 }
798 // The staged objects: identical to those after the first mirror round,
799 // and each copied once.
800 final := sw.list(sealUK, activator.MovesPrefix(offHost))
801 if len(final) != len(stagedAfter) {
802 t.Fatalf("staged %d objects, %d after the first round", len(final), len(stagedAfter))
803 }
804 for k, o := range final {
805 if stagedAfter[k] != o {
806 t.Fatalf("%s changed: %+v, was %+v", k, o, stagedAfter[k])
807 }
808 vs, err := sw.fake.ListVersions(ctx, sealUK, k)
809 if err != nil || len(vs) != 1 {
810 t.Fatalf("%s has %d versions (%v)", k, len(vs), err)
811 }
812 }
813 seal, _, ok := sw.sealRecord()
814 if !ok {
815 t.Fatal("no seal record")
816 }
817 sw.openArchive(seal)
818 if n := len(seal.Staged); n != len(final) {
819 t.Fatalf("seal names %d staged objects, %d staged", n, len(final))
820 }
821}
822
823// (f) The revert: the sealer lifts this move's tombstone, deletes exactly
824// the staging lane and the archive prefix in the target's bucket, clears
825// the receipts and lifts the fence; a wake can take the lease again. A
826// lease that is not this move's is left alone. Every part is an ensure.
827func TestOfflineSealRevertLiftsAndUnfences(t *testing.T) {
828 sw := newSealWorld(t)
829 ie := sw.ie
830 sw.instance(ie)
831 sw.startAndSeal("ie-a", ie)
832 sw.mustPass(ie)
833 ctx := context.Background()
834 // Neighbours the revert must not touch.
835 sw.fake.Put(sealUK, "moves/other.example.com/0000/x.ltx", []byte("x"))
836 sw.fake.Put(sealUK, "moves/@offline/other.example.com/"+offID2+"/files.sealed", []byte("x"))
837 sw.fake.Put(sealUK, sealPrefix+"/"+offHost+"/0000/y.ltx", []byte("y"))
838 // Copy receipts, as a round over a multipart source leaves them.
839 sw.fake.Put(sealIE, activator.MoveReceiptsKey(offHost), []byte(`{"v":1}`))
840 sw.flip()
841 sw.abort()
842 if sw.record().Move.State != provision.OfflineReverting {
843 t.Fatal("fixture: not reverting")
844 }
845 if !ie.svc.Fenced(offHost) {
846 t.Fatal("fixture: not fenced")
847 }
848 // The sealer's own step lifts the fence too (the source step does it
849 // as well, earlier in the same round).
850 if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
851 t.Fatal(err)
852 }
853 if ie.svc.Fenced(offHost) {
854 t.Fatal("still fenced after the revert")
855 }
856 owner, expires, _ := sw.lock()
857 if !lease.IsLiftedTombstone(owner, expires) || owner != moves.TombstoneOwner(offID) {
858 t.Fatalf("lock %s %s, want this move's lifted tombstone", owner, expires)
859 }
860 for _, prefix := range []string{activator.MovesPrefix(offHost), moves.OfflineArchivePrefix(offHost, offID)} {
861 if n := len(sw.list(sealUK, prefix)); n != 0 {
862 t.Fatalf("%s holds %d objects", prefix, n)
863 }
864 }
865 for _, k := range []string{"moves/other.example.com/0000/x.ltx", "moves/@offline/other.example.com/" + offID2 + "/files.sealed", sealPrefix + "/" + offHost + "/0000/y.ltx"} {
866 if _, ok := sw.fake.Current(sealUK, k); !ok {
867 t.Fatalf("the revert deleted %s", k)
868 }
869 }
870 if _, ok := sw.fake.Current(sealIE, activator.MoveReceiptsKey(offHost)); ok {
871 t.Fatal("the receipts are still there")
872 }
873 if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseReverted {
874 t.Fatalf("report %+v", r)
875 }
876 // Again: nothing changes, and the lifted tombstone is not taken again.
877 before := sw.fake.Snapshot()
878 sw.mustPass(ie)
879 if sw.fake.Snapshot() != before {
880 t.Fatal("a second revert pass wrote to the buckets")
881 }
882 wake, err := acquireAsAWake(ctx)
883 if err != nil {
884 t.Fatalf("a wake after the revert: %v", err)
885 }
886 // A lease that is not this move's (a live wake now) is left alone.
887 if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
888 t.Fatal(err)
889 }
890 if err := wake.Err(); err != nil {
891 t.Fatalf("the revert disturbed a wake's lease: %v", err)
892 }
893 wake.Release(ctx)
894 // Another move's tombstone is never lifted by this one.
895 other, err := json.Marshal(litestream.Lease{Generation: 9, Owner: moves.TombstoneOwner(offID2), ExpiresAt: time.Now().Add(time.Hour)})
896 if err != nil {
897 t.Fatal(err)
898 }
899 sw.fake.Put(sealIE, sealPrefix+"/"+offHost+"/lock.json", other)
900 if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
901 t.Fatal(err)
902 }
903 if owner, expires, _ := sw.lock(); owner != moves.TombstoneOwner(offID2) || !time.Now().Before(expires) {
904 t.Fatalf("another move's tombstone was lifted: %s %s", owner, expires)
905 }
906}
907
908// The revert before anything was sealed (abort while sealing failed on a
909// refused item) lifts the tombstone the failed seal took, and a revert
910// with no tombstone at all is a clean pass.
911func TestOfflineSealRevertBeforeTheSeal(t *testing.T) {
912 sw := newSealWorld(t)
913 ie := sw.ie
914 sw.instance(ie)
915 writeFile(t, filepath.Join(ie.dd, offHost, "big"), "")
916 if err := os.Truncate(filepath.Join(ie.dd, offHost, "big"), archive.Cap+1); err != nil {
917 t.Fatal(err)
918 }
919 sw.startAndSeal("ie-a", ie)
920 if err := sw.pass(ie); err == nil {
921 t.Fatal("sealed over the cap")
922 }
923 if owner, _, _ := sw.lock(); owner != moves.TombstoneOwner(offID) {
924 t.Fatal("fixture: no tombstone")
925 }
926 sw.abort()
927 sw.mustPass(ie)
928 if owner, expires, _ := sw.lock(); !lease.IsLiftedTombstone(owner, expires) {
929 t.Fatalf("lock %s %s after the revert", owner, expires)
930 }
931 if n := len(sw.list(sealUK, "moves/")); n != 0 {
932 t.Fatalf("%d objects left in the staging prefixes", n)
933 }
934 sw.mustPass(ie)
935}
936
937// The files item: a file over the cap, a FIFO, and an entry of the host's
938// beside its files directory are failed items (the report says which),
939// never a partial archive and never a seal; once the file is gone the next
940// pass seals.
941func TestOfflineSealFailsAFileOverTheCapAndAFIFO(t *testing.T) {
942 sw := newSealWorld(t)
943 ie := sw.ie
944 sw.instance(ie)
945 sw.startAndSeal("ie-a", ie)
946 files := filepath.Join(ie.dd, offHost)
947 big := filepath.Join(files, "big.bin")
948 writeFile(t, big, "")
949 if err := os.Truncate(big, archive.Cap+1); err != nil {
950 t.Fatal(err)
951 }
952 fifo := filepath.Join(files, "pipe")
953 blobs := filepath.Join(ie.dd, offHost+"-blobs")
954 cases := []struct {
955 name string
956 setup func()
957 want string
958 }{
959 {"over the cap", func() {}, strings.TrimPrefix(archive.ErrOverCap.Error(), "archive: ")},
960 {"a FIFO", func() {
961 if err := os.Remove(big); err != nil {
962 t.Fatal(err)
963 }
964 if err := syscall.Mkfifo(fifo, 0o644); err != nil {
965 t.Fatal(err)
966 }
967 }, strings.TrimPrefix(archive.ErrSpecialFile.Error(), "archive: ") + ": pipe"},
968 // The flat layout's rules give <host>-blobs to the host, and the
969 // archive carries the files directory alone.
970 {"an entry beside the files directory", func() {
971 if err := os.Remove(fifo); err != nil {
972 t.Fatal(err)
973 }
974 writeFile(t, filepath.Join(blobs, "b"), "blob")
975 }, blobs},
976 }
977 for _, c := range cases {
978 c.setup()
979 err := sw.pass(ie)
980 var ref *sealRefusal
981 if !errors.As(err, &ref) || ref.item != "files" || !strings.Contains(ref.Reason, c.want) {
982 t.Fatalf("%s: %v", c.name, err)
983 }
984 r := sw.report(moves.OfflineRoleSealer, "ie-a")
985 if r.Phase != moves.OfflinePhaseFailed || r.Items["files"].OK || !r.Items["db"].OK {
986 t.Fatalf("%s: report %+v", c.name, r)
987 }
988 if _, _, ok := sw.sealRecord(); ok || sw.record().Move.State != provision.OfflineSealing {
989 t.Fatalf("%s: sealed", c.name)
990 }
991 if n := len(sw.list(sealUK, moves.OfflineArchivePrefix(offHost, offID))); n != 0 {
992 t.Fatalf("%s: %d archive objects written", c.name, n)
993 }
994 }
995 if err := os.RemoveAll(blobs); err != nil {
996 t.Fatal(err)
997 }
998 sw.mustPass(ie)
999 if sw.record().Move.State != provision.OfflineSealed {
1000 t.Fatal("not sealed once the files were fixed")
1001 }
1002 if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseSealed || !r.Items["files"].OK {
1003 t.Fatalf("report %+v", r)
1004 }
1005}
1006
1007// A nested instance, made nested by the real layout pass, which takes the
1008// flat files directory <data>/<host> into the instance directory as
1009// <instance dir>/<host>, where the app finds it beside its database. The
1010// archive is read through the raw instance directory and carries that
1011// files directory and the env file, never the database, its litestream
1012// state, the rendered configuration or any other platform entry; it
1013// extracts flat with the app's files at the flat files directory's root,
1014// where the flat app looks. An entry beside the files directory that the
1015// flat layout cannot carry refuses the files item until it is gone.
1016func TestOfflineSealArchivesANestedInstanceFlat(t *testing.T) {
1017 sw := newSealWorld(t)
1018 ie := sw.ie
1019 sw.instance(ie)
1020 raw := sw.nest(ie)
1021 // More platform entries, which stay behind: the generation and files
1022 // markers, the adoption record, the seed marker of a seeded database,
1023 // and the launch descriptors (fixed and per generation) as a real
1024 // nested host keeps them.
1025 for _, name := range []string{offHost + ".db.gen", ".files-gen", offHost + instancekey.AdoptionSuffix, offHost + ".db" + activator.SeedMarkerSuffix,
1026 "launch.run.json", "launch.seed.json", "launch.gen.7.json"} {
1027 writeFile(t, filepath.Join(raw, name), "platform")
1028 }
1029 // An entry of the app's beside its files directory, which no flat
1030 // files directory can hold.
1031 stray := filepath.Join(raw, offHost+"-blobs", "b")
1032 writeFile(t, stray, "blob")
1033 sw.startAndSeal("ie-a", ie)
1034 err := sw.pass(ie)
1035 var ref *sealRefusal
1036 if !errors.As(err, &ref) || ref.item != "files" || !strings.Contains(ref.Reason, offHost+"-blobs") {
1037 t.Fatalf("a stray entry: %v", err)
1038 }
1039 // The refusal says the move cannot carry it, and offers no fix that
1040 // moves or removes the app's data (review I3).
1041 if !strings.Contains(ref.Reason, "this move can carry only the files directory. see the offline move runbook") || strings.Contains(ref.Reason, "move it") || strings.Contains(ref.Reason, "remove") {
1042 t.Fatalf("the refusal: %q", ref.Reason)
1043 }
1044 for _, name := range []string{"launch.run.json", "launch.gen.7.json", ".db.seed"} {
1045 if strings.Contains(ref.Reason, name) {
1046 t.Fatalf("the refusal names the platform's %s: %q", name, ref.Reason)
1047 }
1048 }
1049 if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseFailed || r.Items["files"].OK {
1050 t.Fatalf("report %+v", r)
1051 }
1052 if err := os.RemoveAll(filepath.Dir(stray)); err != nil {
1053 t.Fatal(err)
1054 }
1055 sw.mustPass(ie)
1056 if sw.record().Move.State != provision.OfflineSealed {
1057 t.Fatal("not sealed")
1058 }
1059 seal, _, _ := sw.sealRecord()
1060 files, env := sw.openArchive(seal)
1061 got := map[string]bool{}
1062 if err := filepath.Walk(files, func(p string, info os.FileInfo, err error) error {
1063 if err != nil {
1064 return err
1065 }
1066 if !info.IsDir() {
1067 rel, _ := filepath.Rel(files, p)
1068 got[rel] = true
1069 }
1070 return nil
1071 }); err != nil {
1072 t.Fatal(err)
1073 }
1074 if len(got) != 2 || !got["instance.key"] || !got[filepath.Join("uploads", "a.png")] {
1075 t.Fatalf("the archive carried %v, want the app's two files", got)
1076 }
1077 if readString(t, env) != readString(t, filepath.Join(raw, offHost+".env")) {
1078 t.Fatal("the env file did not travel byte for byte")
1079 }
1080 if seal.KeySHA256 != instancekey.Digest("the-instance-key") {
1081 t.Fatal("the key digest")
1082 }
1083}
1084
1085// nest makes b's instance of offHost nested with the real layout pass,
1086// which takes the flat files directory <data>/<host> into the instance
1087// directory as <instance dir>/<host>, and returns the raw instance
1088// directory.
1089func (sw *sealWorld) nest(ie *offBox) string {
1090 t := sw.t
1091 t.Helper()
1092 ctx := context.Background()
1093 // The bind mount at <dd>/instances, as a link in the fixture.
1094 if err := os.MkdirAll(filepath.Join(ie.dd, ".instances"), 0o755); err != nil {
1095 t.Fatal(err)
1096 }
1097 if err := os.Symlink(".instances", filepath.Join(ie.dd, "instances")); err != nil {
1098 t.Fatal(err)
1099 }
1100 // The layout pass works on the row's own paths; the sockets live in a
1101 // short directory of the test's, as the pass's tests keep them.
1102 sd, err := os.MkdirTemp("", "ly")
1103 if err != nil {
1104 t.Fatal(err)
1105 }
1106 t.Cleanup(func() { os.RemoveAll(sd) })
1107 rt := ie.row(offHost)
1108 if err := ie.reg.SetInstancePaths(offHost, rt.DBPath, rt.Socket, rt.DBPath, filepath.Join(sd, offHost+".sock")); err != nil {
1109 t.Fatal(err)
1110 }
1111 pass := &layout.Pass{Registry: ie.reg, DataDir: ie.dd, SocketDir: sd, SocketsGID: -1, Now: time.Now, SweptAtStart: true, Logf: t.Logf}
1112 res := pass.Converge(ctx, layout.ModeNested)
1113 migrated := false
1114 for _, r := range res {
1115 if r.Host == offHost && r.Outcome == "migrated" {
1116 migrated = true
1117 }
1118 }
1119 if !migrated {
1120 t.Fatalf("the layout pass: %+v", res)
1121 }
1122 ie.nested = true
1123 raw := layout.RawInstanceDir(ie.dd, offHost)
1124 if rt := ie.row(offHost); !provision.NestedPaths(ie.dd, *rt) {
1125 t.Fatalf("fixture: row %+v is not nested", rt)
1126 }
1127 if lexists(filepath.Join(ie.dd, offHost)) {
1128 t.Fatal("fixture: the layout pass left the flat files directory")
1129 }
1130 for _, name := range []string{offHost + ".db", offHost + ".env", offHost + ".config.env", ".activator-" + offHost + ".db-litestream", filepath.Join(offHost, "instance.key")} {
1131 if !lexists(filepath.Join(raw, name)) {
1132 t.Fatalf("fixture: the layout pass did not move %s into the instance directory", name)
1133 }
1134 }
1135 return raw
1136}
1137
1138// Finishing lifts the tombstone only once every box the seal named reports
1139// gone: until then each pass keeps the stop. After the lift a wake takes
1140// the lease, the report says lifted, and a later pass changes nothing.
1141func TestOfflineSealFinishingLiftsOnlyAfterEverySourceIsGone(t *testing.T) {
1142 sw := newSealWorld(t)
1143 ie := sw.ie
1144 sw.instance(ie)
1145 ieB := sw.sourceBox("ie-b")
1146 sw.startAndSeal("ie-a", ie, ieB)
1147 sw.mustPass(ie)
1148 ieB.tick()
1149 if sw.record().Move.State != provision.OfflineSealed {
1150 t.Fatal("not sealed")
1151 }
1152 sw.flip()
1153 sw.landed()
1154 sw.open()
1155 sw.mustPass(ie)
1156 sw.mustPass(ieB)
1157 for _, b := range []string{"ie-a", "ie-b"} {
1158 if r := sw.report(moves.OfflineRoleSource, b); r.Phase != moves.OfflinePhaseParked {
1159 t.Fatalf("%s: %+v", b, r)
1160 }
1161 }
1162 sw.finishing()
1163 ctx := context.Background()
1164 // ie-a goes; ie-b has not run a pass since.
1165 for range 3 {
1166 sw.mustPass(ie)
1167 }
1168 if r := sw.report(moves.OfflineRoleSource, "ie-a"); r.Phase != moves.OfflinePhaseGone {
1169 t.Fatalf("ie-a: %+v", r)
1170 }
1171 owner, _, _ := sw.lock()
1172 if owner != moves.TombstoneOwner(offID) {
1173 t.Fatal("fixture: no tombstone")
1174 }
1175 if _, err := acquireAsAWake(ctx); err == nil {
1176 t.Fatal("the tombstone was lifted before ie-b was gone")
1177 }
1178 if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseSealed || r.TombstoneOwner != moves.TombstoneOwner(offID) {
1179 t.Fatalf("sealer report while waiting: %+v", r)
1180 }
1181 for range 3 {
1182 sw.mustPass(ieB)
1183 }
1184 if r := sw.report(moves.OfflineRoleSource, "ie-b"); r.Phase != moves.OfflinePhaseGone {
1185 t.Fatalf("ie-b: %+v", r)
1186 }
1187 sw.mustPass(ie)
1188 if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseLifted {
1189 t.Fatalf("sealer report: %+v", r)
1190 }
1191 owner, expires, _ := sw.lock()
1192 if !lease.IsLiftedTombstone(owner, expires) || owner != moves.TombstoneOwner(offID) {
1193 t.Fatalf("lock %s %s", owner, expires)
1194 }
1195 before := sw.fake.Snapshot()
1196 sw.mustPass(ie)
1197 if sw.fake.Snapshot() != before {
1198 t.Fatal("a pass after the lift wrote to the buckets")
1199 }
1200 wake, err := acquireAsAWake(ctx)
1201 if err != nil {
1202 t.Fatalf("a wake after the lift: %v", err)
1203 }
1204 wake.Release(ctx)
1205 if len(sw.pages) != 0 {
1206 t.Fatalf("paged: %v", sw.pages)
1207 }
1208}
1209
1210// A host that never kept a files directory (seen on real nested hosts: the
1211// instance directory holds the database, the env files and the launch
1212// descriptors, and no <host>) seals with an empty files item: the archive
1213// carries the env file alone, FilesPlain is archive.PlainDigest of no files
1214// directory and that env file (FilesEntries 1), and it extracts to an
1215// empty files directory beside the env file. Nested and flat alike.
1216func TestOfflineSealAHostWithNoFilesDirectory(t *testing.T) {
1217 for _, nested := range []bool{true, false} {
1218 t.Run(map[bool]string{true: "nested", false: "flat"}[nested], func(t *testing.T) {
1219 sw := newSealWorld(t)
1220 ie := sw.ie
1221 sw.instance(ie)
1222 files := filepath.Join(ie.dd, offHost)
1223 envPath := filepath.Join(ie.dd, offHost+".env")
1224 if nested {
1225 raw := sw.nest(ie)
1226 files, envPath = filepath.Join(raw, offHost), filepath.Join(raw, offHost+".env")
1227 writeFile(t, filepath.Join(raw, "launch.run.json"), "{}")
1228 }
1229 if err := os.RemoveAll(files); err != nil {
1230 t.Fatal(err)
1231 }
1232 sw.startAndSeal("ie-a", ie)
1233 sw.mustPass(ie)
1234 if sw.record().Move.State != provision.OfflineSealed {
1235 t.Fatal("not sealed")
1236 }
1237 seal, _, _ := sw.sealRecord()
1238 want, wantBytes, wantEntries, err := archive.PlainDigest(files, envPath)
1239 if err != nil {
1240 t.Fatal(err)
1241 }
1242 if seal.FilesPlain != want || seal.FilesBytes != wantBytes || seal.FilesEntries != wantEntries || wantEntries != 1 {
1243 t.Fatalf("seal files %s %d bytes %d entries, want %s %d bytes 1 entry (env only)", seal.FilesPlain, seal.FilesBytes, seal.FilesEntries, want, wantBytes)
1244 }
1245 if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseSealed || !r.Items["files"].OK {
1246 t.Fatalf("report %+v", r)
1247 }
1248 got, env := sw.openArchive(seal)
1249 ents, err := os.ReadDir(got)
1250 if err != nil || len(ents) != 0 {
1251 t.Fatalf("the landing's files directory: %v entries, %v", len(ents), err)
1252 }
1253 if readString(t, env) != readString(t, envPath) {
1254 t.Fatal("the env file did not travel byte for byte")
1255 }
1256 if lexists(files) {
1257 t.Fatal("the seal made a files directory on the source")
1258 }
1259 })
1260 }
1261}
1262
1263// The tombstone goes between the sealer's LockState and Reassert's own
1264// read: Reassert reports it lost (lease.ErrTombstoneLost) and writes
1265// nothing, so the sealer pages a lost stop once and never takes it again.
1266func TestOfflineSealLostBetweenTheReadAndTheReassert(t *testing.T) {
1267 sw := newSealWorld(t)
1268 ie := sw.ie
1269 sw.instance(ie)
1270 sw.startAndSeal("ie-a", ie)
1271 sw.mustPass(ie)
1272 ctx := context.Background()
1273 key := sealPrefix + "/" + offHost + "/lock.json"
1274 var after string
1275 t.Cleanup(func() { offlineSealFaults.Delete(ie.w) })
1276 offlineSealFaults.Store(ie.w, func(point string) error {
1277 if point == "reassert" {
1278 sw.fake.BareDelete(sealIE, key)
1279 after = sw.fake.Snapshot()
1280 }
1281 return nil
1282 })
1283 err := ie.w.offlineSealerStep(ctx, sw.s, sw.record())
1284 if err == nil || !strings.Contains(err.Error(), "lost stop") {
1285 t.Fatalf("pass: %v", err)
1286 }
1287 if after == "" || sw.fake.Snapshot() != after {
1288 t.Fatal("the bucket was written after the tombstone went")
1289 }
1290 if _, ok := sw.fake.Current(sealIE, key); ok {
1291 t.Fatal("the tombstone was taken again")
1292 }
1293 r := sw.report(moves.OfflineRoleSealer, "ie-a")
1294 if r.Phase != moves.OfflinePhaseFailed || r.Items["lease"].OK || r.Items["lease"].Reason != "lost stop: "+offlineLeaseCause(tombState{}) {
1295 t.Fatalf("report %+v", r)
1296 }
1297 // The next pass finds it gone at LockState: still nothing written,
1298 // still one page.
1299 offlineSealFaults.Delete(ie.w)
1300 if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err == nil || !strings.Contains(err.Error(), "lost stop") {
1301 t.Fatalf("second pass: %v", err)
1302 }
1303 if sw.fake.Snapshot() != after {
1304 t.Fatal("the second pass wrote to the bucket")
1305 }
1306 if len(sw.pages) != 1 {
1307 t.Fatalf("pages %v, want one", sw.pages)
1308 }
1309}
1310
1311// The lost stop's page says abort only while abort can still put the
1312// instance back. From open on, abort is refused, so it points to the
1313// status and the runbook instead. Each state in its own world: the page
1314// goes out once per move.
1315func TestOfflineSealLostStopPageByState(t *testing.T) {
1316 for _, state := range []string{provision.OfflineSealed, provision.OfflineOpen, provision.OfflineFinishing} {
1317 t.Run(state, func(t *testing.T) {
1318 sw := newSealWorld(t)
1319 ie := sw.ie
1320 sw.instance(ie)
1321 sw.startAndSeal("ie-a", ie)
1322 sw.mustPass(ie)
1323 ctx := context.Background()
1324 if state != provision.OfflineSealed {
1325 sw.flip()
1326 sw.landed()
1327 sw.open()
1328 }
1329 if state == provision.OfflineFinishing {
1330 sw.finishing()
1331 }
1332 if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
1333 t.Fatalf("the stop held: %v", err)
1334 }
1335 if got := sw.record().Move.State; got != state {
1336 t.Fatalf("fixture: state %s", got)
1337 }
1338 sw.fake.BareDelete(sealIE, sealPrefix+"/"+offHost+"/lock.json")
1339 if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err == nil || !strings.Contains(err.Error(), "lost stop") {
1340 t.Fatalf("lost: %v", err)
1341 }
1342 if len(sw.pages) != 1 || sw.pages[0] != "incident: offline move of "+offHost+" lost its stop" {
1343 t.Fatalf("pages %v", sw.pages)
1344 }
1345 body := sw.bodies[0]
1346 if !strings.HasPrefix(body, "The offline move "+offID+" of "+offHost+" is "+state+", and its stop no longer holds: the source's lease is absent, not this move's tombstone. ") {
1347 t.Fatalf("body: %s", body)
1348 }
1349 abort := strings.HasSuffix(body, "The move cannot go on. Read carlos move offline status --host "+offHost+" and abort the move.")
1350 runbook := strings.HasSuffix(body, "It cannot be aborted once it is open. Read carlos move offline status --host "+offHost+" and follow the offline move runbook.")
1351 if wantAbort := state == provision.OfflineSealed; abort != wantAbort || runbook == wantAbort {
1352 t.Fatalf("body in %s: %s", state, body)
1353 }
1354 })
1355 }
1356}
1357
1358// The backstop behind the holds and the fence (review M2): a source report
1359// written since the seal at a position past the sealed one is a lost stop
1360// in the db-tail item, even with the tombstone held. Nothing is written to
1361// the lease, and the page names the box.
1362func TestOfflineSealLostStopOnASourcePastTheSealedPosition(t *testing.T) {
1363 sw := newSealWorld(t)
1364 ie := sw.ie
1365 sw.instance(ie)
1366 ieB := sw.sourceBox("ie-b")
1367 sw.startAndSeal("ie-a", ie, ieB)
1368 sw.mustPass(ie)
1369 if sw.record().Move.State != provision.OfflineSealed {
1370 t.Fatal("not sealed")
1371 }
1372 seal, _, _ := sw.sealRecord()
1373 ctx := context.Background()
1374 // A pass with every report at or below the seal keeps the stop.
1375 ieB.tick()
1376 if err := ie.w.offlineSealerStep(ctx, sw.s, sw.record()); err != nil {
1377 t.Fatalf("at the seal: %v", err)
1378 }
1379 // ie-b writes past the sealed position, and reports it.
1380 meta := filepath.Join(activatorMetaDir(provision.DefaultDBPath(ieB.dd, offHost)), "ltx", "0")
1381 writeFile(t, filepath.Join(meta, ltxName(seal.FinalTXID+1, seal.FinalTXID+3)), "ltx")
1382 ieB.tick()
1383 if r := sw.report(moves.OfflineRoleSource, "ie-b"); r.LocalTXID != seal.FinalTXID+3 || r.At.Before(seal.At) {
1384 t.Fatalf("fixture: ie-b reports %d at %s", r.LocalTXID, r.At)
1385 }
1386 before := sw.fake.Snapshot()
1387 err := ie.w.offlineSealerStep(ctx, sw.s, sw.record())
1388 if err == nil || !strings.Contains(err.Error(), "lost stop") {
1389 t.Fatalf("past the seal: %v", err)
1390 }
1391 if sw.fake.Snapshot() != before {
1392 t.Fatal("the lease was written after the stop was lost")
1393 }
1394 cause := offlinePastSealCause("ie-b", seal.FinalTXID+3, seal.FinalTXID)
1395 r := sw.report(moves.OfflineRoleSealer, "ie-a")
1396 if r.Phase != moves.OfflinePhaseFailed || r.Items["db-tail"].OK || r.Items["db-tail"].Reason != "lost stop: "+cause || r.TombstoneOwner != "" || !r.Items["lease"].OK {
1397 t.Fatalf("report %+v", r)
1398 }
1399 if len(sw.bodies) != 1 || !strings.Contains(sw.bodies[0], "its stop no longer holds: "+cause+".") {
1400 t.Fatalf("pages %v %v", sw.pages, sw.bodies)
1401 }
1402 if owner, _, _ := sw.lock(); owner != moves.TombstoneOwner(offID) {
1403 t.Fatalf("the tombstone is %s", owner)
1404 }
1405}
1406
1407// Fix round 1, Astra's finding 2: the report refresh (offlineReassert) is
1408// report-only for the sealer. Repeated refreshes rewrite the sealer's report
1409// with the tombstone's expiry as read and leave the lock object byte for
1410// byte; an expired, missing or replaced tombstone makes the refresh report
1411// the lost stop, never a fresh report, and the lock is not written.
1412func TestOfflineSealerRefreshNeverRenewsTheTombstone(t *testing.T) {
1413 sw := newSealWorld(t)
1414 ie := sw.ie
1415 sw.instance(ie)
1416 sw.startAndSeal("ie-a", ie)
1417 sw.mustPass(ie) // seals, takes the tombstone, records sealed
1418 if st := sw.record().Move.State; st != provision.OfflineSealed {
1419 t.Fatalf("fixture: state %s", st)
1420 }
1421 sw.mustPass(ie) // the view learns sealed, as every pass refreshes it
1422 ctx := context.Background()
1423 key := sealPrefix + "/" + offHost + "/lock.json"
1424 ours, ok := sw.fake.Current(sealIE, key)
1425 if !ok {
1426 t.Fatal("fixture: no tombstone")
1427 }
1428 _, expires, _ := sw.lock()
1429 for i := range 3 {
1430 time.Sleep(20 * time.Millisecond) // a renewal would write a later expiry
1431 sw.advance(15 * time.Second)
1432 if n := ie.w.offlineReassert(ctx, sw.s); n != 1 {
1433 t.Fatalf("refresh %d acted on %d records", i, n)
1434 }
1435 if now, _ := sw.fake.Current(sealIE, key); !bytes.Equal(now, ours) {
1436 t.Fatalf("refresh %d rewrote the tombstone:\nbefore %s\nafter %s", i, ours, now)
1437 }
1438 r := sw.report(moves.OfflineRoleSealer, "ie-a")
1439 if r.Phase != moves.OfflinePhaseSealed || !r.At.Equal(sw.clock()) || !r.TombstoneUntil.Equal(expires.UTC()) || r.TombstoneOwner != moves.TombstoneOwner(offID) {
1440 t.Fatalf("refresh %d: report %+v, want sealed at %v with the expiry %v as read", i, r, sw.clock(), expires)
1441 }
1442 }
1443
1444 lockBody := func(owner string, exp time.Time) []byte {
1445 b, err := json.Marshal(litestream.Lease{Generation: 1, Owner: owner, ExpiresAt: exp})
1446 if err != nil {
1447 t.Fatal(err)
1448 }
1449 return b
1450 }
1451 for _, c := range []struct {
1452 name string
1453 damage func()
1454 }{
1455 {"expired", func() { sw.fake.Put(sealIE, key, lockBody(moves.TombstoneOwner(offID), time.Now().Add(-time.Second))) }},
1456 {"missing", func() { sw.fake.BareDelete(sealIE, key) }},
1457 {"replaced", func() { sw.fake.Put(sealIE, key, lockBody(moves.TombstoneOwner(offID2), time.Now().Add(time.Hour))) }},
1458 } {
1459 sw.fake.Put(sealIE, key, ours)
1460 sw.advance(15 * time.Second)
1461 ie.w.offlineReassert(ctx, sw.s)
1462 if r := sw.report(moves.OfflineRoleSealer, "ie-a"); r.Phase != moves.OfflinePhaseSealed {
1463 t.Fatalf("%s: before the damage the report is %s", c.name, r.Phase)
1464 }
1465 c.damage()
1466 before := sw.fake.Snapshot()
1467 sw.advance(15 * time.Second)
1468 ie.w.offlineReassert(ctx, sw.s)
1469 if sw.fake.Snapshot() != before {
1470 t.Fatalf("%s: the refresh wrote the lease", c.name)
1471 }
1472 r := sw.report(moves.OfflineRoleSealer, "ie-a")
1473 if r.Phase != moves.OfflinePhaseFailed || r.Items["lease"].OK || !strings.Contains(r.Items["lease"].Reason, "lost stop") || r.TombstoneOwner != "" {
1474 t.Fatalf("%s: report %+v, want the lost stop", c.name, r)
1475 }
1476 }
1477}
1478