offline-move: landing fix round 1 (revert parks first, env-pin names, verified report before release, copy)
- I1: the target revert parks the files, env and config whenever the row is held for the move or the box has reported, before DeleteIfHold. - I2: a failed env pin names the variables CompareEnv finds against the source's env-report.json; never values. - M1: open writes the verified report before ReleaseHold; the already-ready path goes on only from a report with every item OK. - M2: a SetReplication write error waits, like SetNextDue. - M3: the test S3 wrapper is gone; tests use leasetest's server and the real sealer (T4's offlineSealerStep) for the seal. - M4: the newest TXID is replicaPosition, the sealer's rule. - A host with no files directory and a nested source's launch files land. - Copy: movesoffline_land.go is scanned; item reasons carry no error text and are recorded as pending rows MV-O110 to MV-O176.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
4 files changed,
+662
−714
approved_moves_copy_test.go+11 −2docs/superpowers/plans/2026-10-06-moves-phase1-copy.md+72 −0internal/host/movesoffline_land.go+283 −285internal/host/movesoffline_land_test.go+296 −427
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go| index ab3f90371..36666a1a1 100644 |
| --- a/approved_moves_copy_test.go |
| +++ b/approved_moves_copy_test.go |
| @@ -100,6 +100,11 @@ var movesCopyIDs = []string{ |
| "MV-O66", "MV-O67", "MV-O68", "MV-O69", "MV-O70", "MV-O71", "MV-O72", "MV-O73", "MV-O74", "MV-O75", "MV-O76", "MV-O77", "MV-O78", "MV-O79", "MV-O80", "MV-O81", |
| "MV-O82", "MV-O83", "MV-O84", "MV-O85", "MV-O86", "MV-O87", "MV-O88", "MV-O89", "MV-O90", "MV-O91", "MV-O92", "MV-O93", "MV-O94", "MV-O95", "MV-O96", "MV-O97", |
| "MV-O98", |
| + "MV-O110", "MV-O111", "MV-O112", "MV-O113", "MV-O114", "MV-O115", "MV-O116", "MV-O117", "MV-O118", "MV-O119", "MV-O120", "MV-O121", "MV-O122", "MV-O123", "MV-O124", "MV-O125", |
| + "MV-O126", "MV-O127", "MV-O128", "MV-O129", "MV-O130", "MV-O131", "MV-O132", "MV-O133", "MV-O134", "MV-O135", "MV-O136", "MV-O137", "MV-O138", "MV-O139", "MV-O140", "MV-O141", |
| + "MV-O142", "MV-O143", "MV-O144", "MV-O145", "MV-O146", "MV-O147", "MV-O148", "MV-O149", "MV-O150", "MV-O151", "MV-O152", "MV-O153", "MV-O154", "MV-O155", "MV-O156", "MV-O157", |
| + "MV-O158", "MV-O159", "MV-O160", "MV-O161", "MV-O162", "MV-O163", "MV-O164", "MV-O165", "MV-O166", "MV-O167", "MV-O168", "MV-O169", "MV-O170", "MV-O171", |
| + "MV-O172", "MV-O173", "MV-O174", "MV-O175", "MV-O176", |
| } |
| // copyRow is one row of the record. The File cell is the file the sentence is |
| @@ -1266,6 +1271,7 @@ var movesScanFiles = []string{ |
| "internal/host/movespass.go", |
| "internal/host/movesoffline.go", |
| "internal/host/movesoffline_seal.go", |
| + "internal/host/movesoffline_land.go", |
| "internal/host/movesphase1.go", |
| "internal/host/movespolicy.go", |
| "internal/host/movesprobe.go", |
| @@ -1410,8 +1416,11 @@ func movesScanBaseRef(t *testing.T) string { |
| // Passing return (bool, string), sourceRefusal (string, error)); its error |
| // results are not. |
| var movesLocalSinks = map[string]map[string]int{ |
| - "internal/moves/readiness.go": {"fail": 0, "reasonOr": 1}, |
| - "internal/host/movesreport.go": {"fail": 0}, |
| + "internal/moves/readiness.go": {"fail": 0, "reasonOr": 1}, |
| + "internal/host/movesreport.go": {"fail": 0}, |
| + // The landing's item reasons (offline move Task 5): every reason goes |
| + // through one of these, never an error's text (MV-O110 on). |
| + "internal/host/movesoffline_land.go": {"landWait": 1, "landWaitErr": 1, "landFail": 1, "landFailErr": 1, "landFailAll": 0, "landWaitAll": 0}, |
| "internal/console/apiinstancemove.go": {"moveTransitionSaying": 8}, |
| // Every operator sentence of `carlos move` goes through moveSay, L13's |
| // say (named so because the scan resolves helpers by name, and |
diff --git a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md| index f6db016aa..633163f8b 100644 |
| --- a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md |
| +++ b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md |
| @@ -609,3 +609,75 @@ The sealer's item reasons, which `carlos move offline status` shows as MV-O55's |
| | MV-O96 | internal/host/movesoffline_seal.go | live build <version> | pending | |
| | MV-O97 | internal/host/movesoffline_seal.go | replication "<rung>" | pending | |
| | MV-O98 | internal/host/movesoffline_seal.go | next due <unix time> | pending | |
| + |
| +The landing's item reasons (`internal/host/movesoffline_land.go`, Task 5 fix round 1), which `carlos move offline status` shows as MV-O55's `<detail>` for the target's items. The copy gate reads every Check reason, so each has a row. A reason never carries an error's text: the error behind a reason that waits or fails is the box's log line. `<names>` is the list of variables a failed env pin finds different, each as `<name> (<kind>)` from `moves.CompareEnv`, never a value. MV-O110 to MV-O176. |
| + |
| +| id | file | text | status | |
| +|---|---|---|---| |
| +| MV-O110 | internal/host/movesoffline_land.go | a stale tombstone in the target's bucket could not be lifted | pending | |
| +| MV-O111 | internal/host/movesoffline_land.go | a writer holds the target's lock | pending | |
| +| MV-O112 | internal/host/movesoffline_land.go | after extraction the files are not the seal's | pending | |
| +| MV-O113 | internal/host/movesoffline_land.go | next due could not be set | pending | |
| +| MV-O114 | internal/host/movesoffline_land.go | next due reads back differently from the seal's | pending | |
| +| MV-O115 | internal/host/movesoffline_land.go | no registry on this box | pending | |
| +| MV-O116 | internal/host/movesoffline_land.go | no row yet: reconcile mints it held | pending | |
| +| MV-O117 | internal/host/movesoffline_land.go | the archive could not be extracted | pending | |
| +| MV-O118 | internal/host/movesoffline_land.go | the archive could not be read | pending | |
| +| MV-O119 | internal/host/movesoffline_land.go | the archive does not verify | pending | |
| +| MV-O120 | internal/host/movesoffline_land.go | the archive is not the one the seal names | pending | |
| +| MV-O121 | internal/host/movesoffline_land.go | the channel could not be resolved | pending | |
| +| MV-O122 | internal/host/movesoffline_land.go | the channel points at another build than the sealed one | pending | |
| +| MV-O123 | internal/host/movesoffline_land.go | the config ack could not be read | pending | |
| +| MV-O124 | internal/host/movesoffline_land.go | the config ack does not name this move and the rendered file | pending | |
| +| MV-O125 | internal/host/movesoffline_land.go | the config could not be rendered | pending | |
| +| MV-O126 | internal/host/movesoffline_land.go | the config file could not be read | pending | |
| +| MV-O127 | internal/host/movesoffline_land.go | the config file does not hold the rendered bytes | pending | |
| +| MV-O128 | internal/host/movesoffline_land.go | the config was rendered at another path | pending | |
| +| MV-O129 | internal/host/movesoffline_land.go | the env file could not be read | pending | |
| +| MV-O130 | internal/host/movesoffline_land.go | the env file is missing | pending | |
| +| MV-O131 | internal/host/movesoffline_land.go | the env file is not the source's | pending | |
| +| MV-O132 | internal/host/movesoffline_land.go | the env file the target held could not be parked | pending | |
| +| MV-O133 | internal/host/movesoffline_land.go | the files directory the target held could not be parked | pending | |
| +| MV-O134 | internal/host/movesoffline_land.go | the instance key is not the source's | pending | |
| +| MV-O135 | internal/host/movesoffline_land.go | the launch environment could not be composed | pending | |
| +| MV-O136 | internal/host/movesoffline_land.go | the launch environment differs from the source's | pending | |
| +| MV-O137 | internal/host/movesoffline_land.go | the launch environment differs from the source's in <names> | pending | |
| +| MV-O138 | internal/host/movesoffline_land.go | the record and the seal disagree on this item | pending | |
| +| MV-O139 | internal/host/movesoffline_land.go | the record carries no seal evidence | pending | |
| +| MV-O140 | internal/host/movesoffline_land.go | the replication rung could not be set | pending | |
| +| MV-O141 | internal/host/movesoffline_land.go | the replication rung reads back differently from the seal's | pending | |
| +| MV-O142 | internal/host/movesoffline_land.go | the row is held by another operation | pending | |
| +| MV-O143 | internal/host/movesoffline_land.go | the row is not held landing by this move yet | pending | |
| +| MV-O144 | internal/host/movesoffline_land.go | the row's paths are nested, and an offline move lands flat rows only | pending | |
| +| MV-O145 | internal/host/movesoffline_land.go | the seal could not be read | pending | |
| +| MV-O146 | internal/host/movesoffline_land.go | the seal does not decode | pending | |
| +| MV-O147 | internal/host/movesoffline_land.go | the seal is missing | pending | |
| +| MV-O148 | internal/host/movesoffline_land.go | the seal is not a valid seal of this move | pending | |
| +| MV-O149 | internal/host/movesoffline_land.go | the seal is not the one the record names | pending | |
| +| MV-O150 | internal/host/movesoffline_land.go | the seal staged nothing | pending | |
| +| MV-O151 | internal/host/movesoffline_land.go | the staged replica could not be listed | pending | |
| +| MV-O152 | internal/host/movesoffline_land.go | the staged replica could not be verified this pass | pending | |
| +| MV-O153 | internal/host/movesoffline_land.go | the staged replica does not end at the sealed position | pending | |
| +| MV-O154 | internal/host/movesoffline_land.go | the staged replica does not restore clean | pending | |
| +| MV-O155 | internal/host/movesoffline_land.go | the staged replica is not the one the seal staged | pending | |
| +| MV-O156 | internal/host/movesoffline_land.go | the target's lock could not be read | pending | |
| +| MV-O157 | internal/host/movesoffline_land.go | the target's lock is still held after the lift | pending | |
| +| MV-O158 | internal/host/movesoffline_land.go | the target's lock is this move's own tombstone | pending | |
| +| MV-O159 | internal/host/movesoffline_land.go | the timers could not be read | pending | |
| +| MV-O160 | internal/host/movesoffline_land.go | the timers could not be replaced | pending | |
| +| MV-O161 | internal/host/movesoffline_land.go | the timers read back differently from the seal's | pending | |
| +| MV-O162 | internal/host/movesoffline_land.go | the timers snapshot could not be read | pending | |
| +| MV-O163 | internal/host/movesoffline_land.go | the timers snapshot is not the one the seal names | pending | |
| +| MV-O164 | internal/host/movesoffline_land.go | this box cannot resolve channels | pending | |
| +| MV-O165 | internal/host/movesoffline_land.go | this box has no exec backend to compose the launch environment | pending | |
| +| MV-O166 | internal/host/movesoffline_land.go | this box has no held render | pending | |
| +| MV-O167 | internal/host/movesoffline_land.go | this box holds no CARLOS_SECRETS_KEY to key the env report | pending | |
| +| MV-O168 | internal/host/movesoffline_land.go | this box holds no CARLOS_SECRETS_KEY to open the archive with | pending | |
| +| MV-O169 | internal/host/movesoffline_land.go | waits for the config item: the instance key is read from both files | pending | |
| +| MV-O170 | internal/host/movesoffline_land.go | waits for the files and config items | pending | |
| +| MV-O171 | internal/host/movesoffline_land.go | waits for the files item | pending | |
| +| MV-O172 | internal/host/movesoffline_land.go | a path the target held for the host could not be parked | pending | |
| +| MV-O173 | internal/host/movesoffline_land.go | a path the target held for the host is still there | pending | |
| +| MV-O174 | internal/host/movesoffline_land.go | not checked: the row is not held for this move | pending | |
| +| MV-O175 | internal/host/movesoffline_land.go | not checked: the seal could not be used | pending | |
| +| MV-O176 | internal/host/movesoffline_land.go | the row could not be read back | pending | |
diff --git a/internal/host/movesoffline_land.go b/internal/host/movesoffline_land.go| index ba6a11da2..487bad3fb 100644 |
| --- a/internal/host/movesoffline_land.go |
| +++ b/internal/host/movesoffline_land.go |
| @@ -11,14 +11,12 @@ import ( |
| "io" |
| "log" |
| "os" |
| - "path" |
| "path/filepath" |
| + "reflect" |
| "sort" |
| "strings" |
| "time" |
| - "github.com/superfly/ltx" |
| - |
| "github.com/carlosframework/platform/internal/activator" |
| "github.com/carlosframework/platform/internal/appconfig" |
| "github.com/carlosframework/platform/internal/instancekey" |
| @@ -50,6 +48,10 @@ import ( |
| // then releases the hold, and the build arrives the way every new instance's |
| // does, through ordinary adoption and root's existing stage pass. The first |
| // request promotes the staged replica (ExecBackend.restorePlain). |
| +// |
| +// An item's reason is what `carlos move offline status` shows beside it, so |
| +// each is a recorded sentence (copy rows MV-O110 on). An error behind a reason |
| +// is logged on the box, never put in the reason. |
| // offlineTargetStep is the target step of one offline state. Every step is an |
| // ensure: an error is logged by the pass, throttled per host, and the next |
| @@ -109,7 +111,7 @@ func (w *movesWorker) offlineLand(ctx context.Context, s release.Store, rec prov |
| // landCrash is the landing's test seam (moveCfg.landFault, the landing's |
| // crash points): an error stands for a crash at point, after the item of |
| // that name was ensured ("report" after the report is written, "landed" |
| -// after the CAS). Nil in production. |
| +// after the CAS, "released" after open's release). Nil in production. |
| func (w *movesWorker) landCrash(point string) error { |
| if w.cfg.landFault == nil { |
| return nil |
| @@ -119,13 +121,14 @@ func (w *movesWorker) landCrash(point string) error { |
| // offlineTargetOpen is the target step at offline-open. While the row is |
| // still held by the move, every landing item is ensured and verified against |
| -// the seal once more, and only when all are OK is the hold released (under |
| -// reconcileMu and the host's hold lock, as the source's revert releases its |
| -// own). The row then reads ready: ordinary adoption stages the channel's |
| -// build, and the first request promotes moves/<host>/. Then the move's |
| -// archive prefix is deleted from this box's bucket and the box reports |
| -// opened. A pass after the release finds the row ready and does not verify |
| -// again (the staging may already be promoted and gone). |
| +// the seal once more, and the verified report is written; only when every |
| +// item is OK is the hold released (under reconcileMu and the host's hold |
| +// lock, as the source's revert releases its own). The row then reads ready: |
| +// ordinary adoption stages the channel's build, and the first request |
| +// promotes moves/<host>/. Then the move's archive prefix is deleted from this |
| +// box's bucket and the box reports opened. A pass after the release finds the |
| +// row ready and does not verify again (the staging may already be promoted |
| +// and gone); it goes on only from this box's own report with every item OK. |
| func (w *movesWorker) offlineTargetOpen(ctx context.Context, s release.Store, rec provision.Record) error { |
| cfg := w.cfg |
| m, host := rec.Move, rec.Host |
| @@ -134,10 +137,7 @@ func (w *movesWorker) offlineTargetOpen(ctx context.Context, s release.Store, re |
| return err |
| } |
| prev := w.prevTargetReport(ctx, s, host, m.ID) |
| - items := map[string]moves.Check(nil) |
| - if prev != nil { |
| - items = prev.Items |
| - } |
| + var items map[string]moves.Check |
| switch { |
| case rt == nil: |
| return fmt.Errorf("offline move %s: %s: no row on the target at open", m.ID, host) |
| @@ -146,10 +146,12 @@ func (w *movesWorker) offlineTargetOpen(ctx context.Context, s release.Store, re |
| if err != nil { |
| return err |
| } |
| + // The verified report goes first, so a crash after the release |
| + // leaves the items this pass verified, not an older pass's. |
| + if err := writeOfflineReport(ctx, s, r); err != nil { |
| + return err |
| + } |
| if !ok { |
| - if err := writeOfflineReport(ctx, s, r); err != nil { |
| - return err |
| - } |
| _, bad := r.ItemsOK(moves.OfflineOwnerLanding) |
| return fmt.Errorf("offline move %s: %s: not opening: %s not OK on this box", m.ID, host, strings.Join(bad, ", ")) |
| } |
| @@ -173,13 +175,23 @@ func (w *movesWorker) offlineTargetOpen(ctx context.Context, s release.Store, re |
| return fmt.Errorf("offline move %s: %s: the hold was not released", m.ID, host) |
| } |
| log.Printf("moves: %s: offline move %s: opened; the row is ready and adoption stages its build", host, m.ID) |
| + if err := w.landCrash("released"); err != nil { |
| + return err |
| + } |
| case rt.HoldOp != "" || registry.IsHold(rt.Status): |
| return fmt.Errorf("offline move %s: %s is held %s by %q, not by this move", m.ID, host, rt.Status, rt.HoldOp) |
| - case prev == nil: |
| - // A ready row this box never landed (reconcile mints a missing |
| - // row ready once the record is open): this pass did not release |
| - // it and will not vouch for it. |
| - return fmt.Errorf("offline move %s: %s is ready on this box, which has no landing report for the move", m.ID, host) |
| + default: |
| + // The row is ready: this box released it on an earlier pass, which |
| + // wrote the verified report first. A ready row this box never |
| + // verified (reconcile mints a missing row ready once the record is |
| + // open) is not vouched for. |
| + if prev == nil { |
| + return fmt.Errorf("offline move %s: %s is ready on this box, which has no landing report for the move", m.ID, host) |
| + } |
| + if ok, bad := prev.ItemsOK(moves.OfflineOwnerLanding); !ok { |
| + return fmt.Errorf("offline move %s: %s is ready on this box, but its last report has %s not OK", m.ID, host, strings.Join(bad, ", ")) |
| + } |
| + items = prev.Items |
| } |
| if prev == nil || prev.Phase != moves.OfflinePhaseOpened { |
| st, err := w.landStore(ctx) |
| @@ -196,15 +208,19 @@ func (w *movesWorker) offlineTargetOpen(ctx context.Context, s release.Store, re |
| }) |
| } |
| -// offlineTargetRevert is the target step at offline-reverting: when this box |
| -// holds the row for the move, delete it (DeleteIfHold, which takes its |
| -// timers with it) under reconcileMu and the hold lock, so a reconcile pass |
| -// that listed the record at landing cannot mint it held again after the |
| -// delete; then rename what the landing extracted and rendered (the files |
| +// offlineTargetRevert is the target step at offline-reverting. When this box |
| +// holds the row for the move, or has reported for the move at all, it first |
| +// renames what the landing may have extracted and rendered (the files |
| // directory, <host>.env and <host>.config.env) into the move's target park |
| -// directory, and report reverted. The source is untouched, so the revert |
| -// loses nothing. A box that never landed anything (no row held by the move |
| -// and no report of its own) does nothing. |
| +// directory, and only then deletes the held row (DeleteIfHold, which takes |
| +// its timers with it) under reconcileMu and the hold lock, so a reconcile |
| +// pass that listed the record at landing cannot mint it held again after |
| +// the delete. Parking first means a crash anywhere leaves nothing of the |
| +// move's in the data directory without a row to say why; a first pass that |
| +// extracted and crashed before it could report is parked all the same, |
| +// since its row is still held. Then it reports reverted. The source is |
| +// untouched, so the revert loses nothing. A box that never had the row held |
| +// for the move and never reported does nothing. |
| func (w *movesWorker) offlineTargetRevert(ctx context.Context, s release.Store, rec provision.Record) error { |
| cfg := w.cfg |
| m, host := rec.Move, rec.Host |
| @@ -222,6 +238,15 @@ func (w *movesWorker) offlineTargetRevert(ctx context.Context, s release.Store, |
| if !held && prev == nil { |
| return nil |
| } |
| + parkDir := moves.OfflineParkDir(cfg.DataDir, host, m.ID, moves.OfflineRoleTarget) |
| + if parkDir == "" { |
| + return fmt.Errorf("offline move %s: no park directory for %s under %q", m.ID, host, cfg.DataDir) |
| + } |
| + for _, p := range []string{filepath.Join(cfg.DataDir, host), filepath.Join(cfg.DataDir, host+".env"), filepath.Join(cfg.DataDir, host+".config.env")} { |
| + if err := parkInto(p, parkDir); err != nil { |
| + return fmt.Errorf("offline move %s: park %s: %w", m.ID, p, err) |
| + } |
| + } |
| if held { |
| if cfg.DeleteIfHold == nil { |
| return errors.New("offline move: no registry to delete the held row through") |
| @@ -243,17 +268,6 @@ func (w *movesWorker) offlineTargetRevert(ctx context.Context, s release.Store, |
| } |
| log.Printf("moves: %s: offline move %s: the target's held row is deleted", host, m.ID) |
| } |
| - if prev != nil { |
| - parkDir := moves.OfflineParkDir(cfg.DataDir, host, m.ID, moves.OfflineRoleTarget) |
| - if parkDir == "" { |
| - return fmt.Errorf("offline move %s: no park directory for %s under %q", m.ID, host, cfg.DataDir) |
| - } |
| - for _, p := range []string{filepath.Join(cfg.DataDir, host), filepath.Join(cfg.DataDir, host+".env"), filepath.Join(cfg.DataDir, host+".config.env")} { |
| - if err := parkInto(p, parkDir); err != nil { |
| - return fmt.Errorf("offline move %s: park %s: %w", m.ID, p, err) |
| - } |
| - } |
| - } |
| return writeOfflineReport(ctx, s, moves.OfflineReport{ |
| Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleTarget, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC(), |
| RowPresent: rt != nil, Files: isDir(filepath.Join(cfg.DataDir, host)), Env: lexists(filepath.Join(cfg.DataDir, host+".env")), |
| @@ -310,7 +324,8 @@ func (w *movesWorker) prevTargetReport(ctx context.Context, s release.Store, hos |
| } |
| // landStore is this box's own data bucket (moveCfg.Mirror for its own |
| -// region): the staged replica, the archive and the timers snapshot. |
| +// region): the staged replica, the archive, the timers snapshot and the |
| +// source's env report. |
| func (w *movesWorker) landStore(ctx context.Context) (lease.MirrorStore, error) { |
| if w.cfg.Mirror == nil || w.cfg.OwnBucket == "" { |
| return nil, errors.New("offline move: no data bucket on this box") |
| @@ -331,6 +346,9 @@ type landRun struct { |
| items map[string]moves.Check |
| hard bool // an item failed in a way the next pass will not clear by itself |
| + // staged is the staged replica's listing as verified this pass (the |
| + // report's Staged): the db item restores once per listing. |
| + staged map[string]moves.StagedObject |
| st lease.MirrorStore |
| arch *offlineArchive |
| @@ -343,26 +361,35 @@ type landRun struct { |
| type offlineArchive struct { |
| manifest archive.Manifest |
| sealed []byte |
| - timers *moves.TimersSnapshot |
| } |
| func (r *landRun) cfg() moveCfg { return r.w.cfg } |
| func (r *landRun) ok(id string) { r.items[id] = moves.Check{OK: true} } |
| -func (r *landRun) okBecause(id, why string) { r.items[id] = moves.Check{OK: true, Reason: why} } |
| +// landWait records an item not yet OK that a later pass may make OK. reason is |
| +// a recorded sentence. |
| +func (r *landRun) landWait(id, reason string) { r.items[id] = moves.Check{Reason: reason} } |
| -// wait records an item not yet OK that a later pass may make OK. |
| -func (r *landRun) wait(id, format string, args ...any) { |
| - r.items[id] = moves.Check{Reason: fmt.Sprintf(format, args...)} |
| +// landWaitErr is landWait for an item an error stopped this pass: the error is the |
| +// box's log line, never the reason. |
| +func (r *landRun) landWaitErr(id, reason string, err error) { |
| + r.w.notice(r.host+"#offline-land-"+id, "moves: %s: offline move %s: %s: %s: %v", r.host, r.m.ID, id, reason, err) |
| + r.items[id] = moves.Check{Reason: reason} |
| } |
| -// fail records a mismatch: the report's phase becomes failed. |
| -func (r *landRun) fail(id, format string, args ...any) { |
| - r.items[id] = moves.Check{Reason: fmt.Sprintf(format, args...)} |
| +// landFail records a mismatch with the seal: the report's phase becomes failed. |
| +func (r *landRun) landFail(id, reason string) { |
| + r.items[id] = moves.Check{Reason: reason} |
| r.hard = true |
| } |
| +// landFailErr is landFail with the error behind it logged on the box. |
| +func (r *landRun) landFailErr(id, reason string, err error) { |
| + r.w.notice(r.host+"#offline-land-"+id, "moves: %s: offline move %s: %s: %s: %v", r.host, r.m.ID, id, reason, err) |
| + r.landFail(id, reason) |
| +} |
| + |
| func (r *landRun) isOK(id string) bool { return r.items[id].OK } |
| func (r *landRun) filesDir() string { return filepath.Join(r.cfg().DataDir, r.host) } |
| @@ -401,7 +428,7 @@ func (w *movesWorker) landReport(ctx context.Context, s release.Store, rec provi |
| {"files", func() { r.files(ctx) }}, |
| {"config", func() { r.config(ctx) }}, |
| {"env", r.env}, |
| - {"env-pin", r.envPin}, |
| + {"env-pin", func() { r.envPin(ctx) }}, |
| {"timers", func() { r.timers(ctx) }}, |
| {"next-due", r.nextDue}, |
| {"replication", r.replication}, |
| @@ -414,8 +441,13 @@ func (w *movesWorker) landReport(ctx context.Context, s release.Store, rec provi |
| } |
| } |
| for _, it := range moves.OfflineOwned(moves.OfflineOwnerLanding) { |
| - if _, done := r.items[it.ID]; !done { |
| - r.wait(it.ID, "not checked: %s", r.blocked()) |
| + if _, done := r.items[it.ID]; done { |
| + continue |
| + } |
| + if r.isOK("row") { |
| + r.landWait(it.ID, "not checked: the seal could not be used") |
| + } else { |
| + r.landWait(it.ID, "not checked: the row is not held for this move") |
| } |
| } |
| phase := moves.OfflinePhaseLanded |
| @@ -425,7 +457,7 @@ func (w *movesWorker) landReport(ctx context.Context, s release.Store, rec provi |
| out := moves.OfflineReport{ |
| Op: r.m.ID, Host: r.host, Box: cfg.Box, Role: moves.OfflineRoleTarget, Phase: phase, At: cfg.now().UTC(), Items: r.items, |
| Held: r.rt != nil && r.rt.HoldOp == r.m.ID && registry.IsHold(r.rt.Status), RowPresent: r.rt != nil, |
| - LocalDB: r.anyWarm(), Files: isDir(r.filesDir()), Env: lexists(r.envPath()), |
| + LocalDB: r.anyWarm(), Files: isDir(r.filesDir()), Env: lexists(r.envPath()), Staged: r.staged, |
| } |
| if r.rt != nil { |
| out.NextDue, out.Replication = r.rt.NextDue, r.rt.Replication |
| @@ -434,26 +466,18 @@ func (w *movesWorker) landReport(ctx context.Context, s release.Store, rec provi |
| return out, ok, nil |
| } |
| -// blocked says why the items after a failed row or seal were not checked. |
| -func (r *landRun) blocked() string { |
| - if !r.isOK("row") { |
| - return "the row is not held for this move" |
| - } |
| - return "the seal could not be read" |
| -} |
| - |
| // row: the reconciled row exists, held "landing" by this move, at flat paths. |
| func (r *landRun) row() { |
| rt := r.rt |
| switch { |
| case rt == nil: |
| - r.wait("row", "no row yet: reconcile mints it held") |
| + r.landWait("row", "no row yet: reconcile mints it held") |
| case rt.HoldOp != "" && rt.HoldOp != r.m.ID: |
| - r.fail("row", "held %s by %s, not by this move", rt.Status, rt.HoldOp) |
| + r.landFail("row", "the row is held by another operation") |
| case rt.HoldOp != r.m.ID || rt.Status != "landing": |
| - r.wait("row", "the row is %q, not held landing by this move yet", rt.Status) |
| + r.landWait("row", "the row is not held landing by this move yet") |
| case provision.NestedPaths(r.cfg().DataDir, *rt): |
| - r.fail("row", "the row's paths are nested; an offline move lands flat rows only") |
| + r.landFail("row", "the row's paths are nested, and an offline move lands flat rows only") |
| default: |
| r.ok("row") |
| } |
| @@ -464,49 +488,51 @@ func (r *landRun) row() { |
| func (r *landRun) readSeal(ctx context.Context) bool { |
| src := r.m.Source |
| if src == nil || src.Evidence == "" { |
| - r.failAll("the record carries no seal evidence") |
| + r.landFailAll("the record carries no seal evidence") |
| return false |
| } |
| key := moves.OfflineSealKey(r.host, r.m.ID) |
| raw, _, err := r.s.Get(ctx, key) |
| if errors.Is(err, os.ErrNotExist) { |
| - r.failAll("the seal %s is missing", key) |
| + r.landFailAll("the seal is missing") |
| return false |
| } |
| if err != nil { |
| - r.waitAll("the seal could not be read: %v", err) |
| + r.w.notice(r.host+"#offline-land-seal", "moves: %s: offline move %s: read %s: %v", r.host, r.m.ID, key, err) |
| + r.landWaitAll("the seal could not be read") |
| return false |
| } |
| - if got := moves.OfflineSealEvidence(r.host, r.m.ID, raw); got != src.Evidence { |
| - r.failAll("the seal's bytes are not the ones the record names (%s, recorded %s)", got, src.Evidence) |
| + if moves.OfflineSealEvidence(r.host, r.m.ID, raw) != src.Evidence { |
| + r.landFailAll("the seal is not the one the record names") |
| return false |
| } |
| var seal moves.OfflineReport |
| if err := json.Unmarshal(raw, &seal); err != nil { |
| - r.failAll("the seal does not decode: %v", err) |
| + r.landFailAll("the seal does not decode") |
| return false |
| } |
| if err := moves.ValidateOfflineSeal(seal, r.host, r.m.ID); err != nil { |
| - r.failAll("%v", err) |
| + r.w.notice(r.host+"#offline-land-seal", "moves: %s: offline move %s: the seal: %v", r.host, r.m.ID, err) |
| + r.landFailAll("the seal is not a valid seal of this move") |
| return false |
| } |
| r.seal = seal |
| return true |
| } |
| -// failAll fails every item but row for a seal that cannot be used. |
| -func (r *landRun) failAll(format string, args ...any) { |
| +// landFailAll fails every item but row for a seal that cannot be used. |
| +func (r *landRun) landFailAll(reason string) { |
| for _, it := range moves.OfflineOwned(moves.OfflineOwnerLanding) { |
| if it.ID != "row" { |
| - r.fail(it.ID, format, args...) |
| + r.landFail(it.ID, reason) |
| } |
| } |
| } |
| -func (r *landRun) waitAll(format string, args ...any) { |
| +func (r *landRun) landWaitAll(reason string) { |
| for _, it := range moves.OfflineOwned(moves.OfflineOwnerLanding) { |
| if it.ID != "row" { |
| - r.wait(it.ID, format, args...) |
| + r.landWait(it.ID, reason) |
| } |
| } |
| } |
| @@ -544,13 +570,13 @@ func (r *landRun) ensureAbsent(id string, paths []string) { |
| continue |
| } |
| if err := parkInto(p, dir); err != nil { |
| - r.wait(id, "%s could not be parked: %v", p, err) |
| + r.landWaitErr(id, "a path the target held for the host could not be parked", err) |
| return |
| } |
| } |
| for _, p := range paths { |
| if lexists(p) { |
| - r.wait(id, "%s is still present", p) |
| + r.landWait(id, "a path the target held for the host is still there") |
| return |
| } |
| } |
| @@ -570,13 +596,14 @@ func (r *landRun) keyAdoption() { |
| // lease: the target bucket's lock for the host is absent, a lifted |
| // tombstone, or an ordinary lease that has expired. A tombstone of another |
| -// move is lifted. Nothing here ever takes the lease. |
| +// move is lifted. Nothing here ever takes the lease. Expiry is read on the |
| +// wall clock, as the lease package reads it. |
| func (r *landRun) lease(ctx context.Context) { |
| cfg := r.cfg() |
| loc := activator.HibernatePrefix(cfg.OwnPrefix, r.host) |
| _, owner, expires, err := lease.LockState(ctx, cfg.OwnBucket, loc, cfg.OwnBucketRegion) |
| if err != nil { |
| - r.wait("lease", "the target's lock could not be read: %v", err) |
| + r.landWaitErr("lease", "the target's lock could not be read", err) |
| return |
| } |
| switch { |
| @@ -587,24 +614,25 @@ func (r *landRun) lease(ctx context.Context) { |
| r.ok("lease") |
| return |
| case owner == moves.TombstoneOwner(r.m.ID): |
| - r.fail("lease", "the target's lock is this move's own tombstone") |
| + r.landFail("lease", "the target's lock is this move's own tombstone") |
| return |
| case !strings.HasPrefix(owner, lease.TombstonePrefix): |
| if time.Now().After(expires) { |
| r.ok("lease") |
| return |
| } |
| - r.wait("lease", "a writer holds the target's lock: %s until %s", owner, expires.UTC().Format(time.RFC3339)) |
| + r.w.notice(r.host+"#offline-land-lease", "moves: %s: offline move %s: the target's lock is held by %s until %s", r.host, r.m.ID, owner, expires.UTC().Format(time.RFC3339)) |
| + r.landWait("lease", "a writer holds the target's lock") |
| return |
| } |
| stale := owner |
| if _, err := lease.LiftTombstone(ctx, cfg.OwnBucket, loc, cfg.OwnBucketRegion, func(o string) bool { return o == stale }); err != nil { |
| - r.wait("lease", "the stale tombstone %s could not be lifted: %v", stale, err) |
| + r.landWaitErr("lease", "a stale tombstone in the target's bucket could not be lifted", err) |
| return |
| } |
| _, owner, expires, err = lease.LockState(ctx, cfg.OwnBucket, loc, cfg.OwnBucketRegion) |
| if err != nil { |
| - r.wait("lease", "the target's lock could not be read after the lift: %v", err) |
| + r.landWaitErr("lease", "the target's lock could not be read", err) |
| return |
| } |
| if (owner == "" && expires.IsZero()) || lease.IsLiftedTombstone(owner, expires) { |
| @@ -612,7 +640,7 @@ func (r *landRun) lease(ctx context.Context) { |
| r.ok("lease") |
| return |
| } |
| - r.wait("lease", "the target's lock is %s after the lift", owner) |
| + r.landWait("lease", "the target's lock is still held after the lift") |
| } |
| func (r *landRun) store(ctx context.Context) (lease.MirrorStore, error) { |
| @@ -627,23 +655,7 @@ func (r *landRun) store(ctx context.Context) (lease.MirrorStore, error) { |
| return st, nil |
| } |
| -// stagedFingerprint names one listing of the staged replica under one seal: |
| -// the db item is verified once per fingerprint. |
| -func stagedFingerprint(evidence string, staged map[string]moves.StagedObject) string { |
| - keys := make([]string, 0, len(staged)) |
| - for k := range staged { |
| - keys = append(keys, k) |
| - } |
| - sort.Strings(keys) |
| - h := sha256.New() |
| - fmt.Fprintf(h, "%s\n", evidence) |
| - for _, k := range keys { |
| - fmt.Fprintf(h, "%s %d %s\n", k, staged[k].Size, staged[k].ETag) |
| - } |
| - return "sha256:" + hex.EncodeToString(h.Sum(nil)) |
| -} |
| - |
| -// stagedDiff describes how got differs from want, "" when equal. |
| +// stagedDiff describes how got differs from want, "" when equal: a log line. |
| func stagedDiff(want, got map[string]moves.StagedObject) string { |
| var missing, extra, changed []string |
| for k, w := range want { |
| @@ -660,95 +672,82 @@ func stagedDiff(want, got map[string]moves.StagedObject) string { |
| extra = append(extra, k) |
| } |
| } |
| - var parts []string |
| - for _, p := range []struct { |
| - what string |
| - keys []string |
| - }{{"missing", missing}, {"extra", extra}, {"changed", changed}} { |
| - if len(p.keys) == 0 { |
| - continue |
| - } |
| - sort.Strings(p.keys) |
| - more := "" |
| - if len(p.keys) > 3 { |
| - more = fmt.Sprintf(" and %d more", len(p.keys)-3) |
| - p.keys = p.keys[:3] |
| - } |
| - parts = append(parts, p.what+" "+strings.Join(p.keys, ", ")+more) |
| + sort.Strings(missing) |
| + sort.Strings(extra) |
| + sort.Strings(changed) |
| + if len(missing)+len(extra)+len(changed) == 0 { |
| + return "" |
| } |
| - return strings.Join(parts, "; ") |
| + return fmt.Sprintf("missing %v, extra %v, changed %v", missing, extra, changed) |
| } |
| -const dbVerifiedPrefix = "verified " |
| - |
| // db: the staged replica under moves/<host>/ is exactly what the seal |
| -// staged (nothing consumes it before open), its newest TXID is the seal's |
| -// FinalTXID, and it restores clean (lease.VerifyReplica into a scratch |
| -// directory under the move's verify park directory, removed after). The |
| -// restore runs once per seal and listing: the fingerprint is remembered in |
| -// the item's reason, and a pass whose listing is unchanged does not restore |
| -// again. |
| +// staged (nothing consumes it before open), its newest TXID |
| +// (replicaPosition, the sealer's own rule) is the seal's FinalTXID, and it |
| +// restores clean (lease.VerifyReplica into a scratch directory under the |
| +// move's verify park directory, removed after). The restore runs once per |
| +// listing: the verified listing is the report's Staged, and a pass whose |
| +// listing equals the last report's, with db OK there, does not restore |
| +// again. The report is this move's, so the seal it was verified against is |
| +// the same immutable one. |
| func (r *landRun) db(ctx context.Context) { |
| cfg := r.cfg() |
| st, err := r.store(ctx) |
| if err != nil { |
| - r.wait("db", "%v", err) |
| + r.landWaitErr("db", "the staged replica could not be listed", err) |
| return |
| } |
| prefix := activator.MovesPrefix(r.host) |
| objs, err := st.List(ctx, cfg.OwnBucket, prefix) |
| if err != nil { |
| - r.wait("db", "the staged replica could not be listed: %v", err) |
| + r.landWaitErr("db", "the staged replica could not be listed", err) |
| return |
| } |
| got := make(map[string]moves.StagedObject, len(objs)) |
| - var newest ltx.TXID |
| for k, o := range objs { |
| got[strings.TrimPrefix(k, prefix)] = moves.StagedObject{Size: o.Size, ETag: o.ETag} |
| - if _, maxTX, err := ltx.ParseFilename(path.Base(k)); err == nil && maxTX > newest { |
| - newest = maxTX |
| - } |
| } |
| if len(r.seal.Staged) == 0 { |
| - r.fail("db", "the seal staged nothing") |
| + r.landFail("db", "the seal staged nothing") |
| return |
| } |
| if diff := stagedDiff(r.seal.Staged, got); diff != "" { |
| - r.fail("db", "the staged replica is not the one the seal staged: %s", diff) |
| + r.w.notice(r.host+"#offline-land-db", "moves: %s: offline move %s: the staged replica differs from the seal's: %s", r.host, r.m.ID, diff) |
| + r.landFail("db", "the staged replica is not the one the seal staged") |
| return |
| } |
| - if uint64(newest) != r.seal.FinalTXID { |
| - r.fail("db", "the staged replica ends at TXID %d, the seal at %d", newest, r.seal.FinalTXID) |
| + if final := replicaPosition(objs, prefix); final != r.seal.FinalTXID { |
| + r.w.notice(r.host+"#offline-land-db", "moves: %s: offline move %s: the staged replica ends at TXID %d, the seal at %d", r.host, r.m.ID, final, r.seal.FinalTXID) |
| + r.landFail("db", "the staged replica does not end at the sealed position") |
| return |
| } |
| - verified := dbVerifiedPrefix + stagedFingerprint(r.m.Source.Evidence, got) |
| - if p := r.prev; p != nil { |
| - if c, ok := p.Items["db"]; ok && c.OK && c.Reason == verified { |
| - r.okBecause("db", verified) |
| - return |
| - } |
| + if p := r.prev; p != nil && p.Items["db"].OK && reflect.DeepEqual(p.Staged, got) { |
| + r.staged = got |
| + r.ok("db") |
| + return |
| } |
| dir := moves.OfflineParkDir(cfg.DataDir, r.host, r.m.ID, moves.OfflineParkVerify) |
| if dir == "" { |
| - r.fail("db", "no verify directory for %s", r.host) |
| + r.landFail("db", "the staged replica does not restore clean") |
| return |
| } |
| if err := os.MkdirAll(dir, 0o700); err != nil { |
| - r.wait("db", "the verify directory: %v", err) |
| + r.landWaitErr("db", "the staged replica could not be verified this pass", err) |
| return |
| } |
| defer os.RemoveAll(dir) |
| if _, err := lease.VerifyReplica(ctx, cfg.OwnBucket, strings.TrimSuffix(prefix, "/"), cfg.OwnBucketRegion, dir); err != nil { |
| - r.fail("db", "the staged replica does not restore clean: %v", err) |
| + r.landFailErr("db", "the staged replica does not restore clean", err) |
| return |
| } |
| - r.okBecause("db", verified) |
| + r.staged = got |
| + r.ok("db") |
| } |
| -// readArchive fetches the move's archive, its manifest and the timers |
| -// snapshot from this box's bucket, once per pass. The names are found by |
| -// listing the move's archive prefix: the one sealed object (".sealed"), the |
| -// manifest that names the seal's digests, and the snapshot of this host. |
| +// readArchive fetches the move's archive and its manifest from this box's |
| +// bucket (the sealer's offlineArchiveSealed and offlineArchiveManifest |
| +// under the move's archive prefix), once per pass and only when an item |
| +// needs it. hard is a manifest that is not the seal's. |
| func (r *landRun) readArchive(ctx context.Context) (*offlineArchive, bool, error) { |
| if r.arch != nil || r.archErr != nil { |
| return r.arch, r.archHard, r.archErr |
| @@ -758,83 +757,45 @@ func (r *landRun) readArchive(ctx context.Context) (*offlineArchive, bool, error |
| } |
| func (r *landRun) fetchArchive(ctx context.Context) (*offlineArchive, bool, error) { |
| - cfg := r.cfg() |
| - st, err := r.store(ctx) |
| + body, err := r.object(ctx, offlineArchiveManifest) |
| if err != nil { |
| return nil, false, err |
| } |
| - prefix := moves.OfflineArchivePrefix(r.host, r.m.ID) |
| - objs, err := st.List(ctx, cfg.OwnBucket, prefix) |
| - if err != nil { |
| - return nil, false, fmt.Errorf("list the archive: %w", err) |
| - } |
| - keys := make([]string, 0, len(objs)) |
| - for k := range objs { |
| - keys = append(keys, k) |
| - } |
| - sort.Strings(keys) |
| out := &offlineArchive{} |
| - var sealedKey string |
| - var manifests int |
| - for _, k := range keys { |
| - switch { |
| - case strings.HasSuffix(k, ".sealed"): |
| - if sealedKey != "" { |
| - return nil, true, fmt.Errorf("the archive prefix holds more than one sealed object (%s, %s)", sealedKey, k) |
| - } |
| - sealedKey = k |
| - case strings.HasSuffix(k, ".json"): |
| - body, err := st.Get(ctx, cfg.OwnBucket, k) |
| - if err != nil { |
| - return nil, false, fmt.Errorf("read %s: %w", k, err) |
| - } |
| - var am archive.Manifest |
| - if json.Unmarshal(body, &am) == nil && am.Plain != "" && am.SealedDataKey != "" { |
| - if am.Plain != r.seal.FilesPlain || (r.seal.FilesSealed != "" && am.Sealed != r.seal.FilesSealed) || am.Op != r.m.ID { |
| - return nil, true, fmt.Errorf("the archive manifest %s is not the sealed one (plain %s, sealed %s, op %s)", k, am.Plain, am.Sealed, am.Op) |
| - } |
| - out.manifest = am |
| - manifests++ |
| - continue |
| - } |
| - if snap, err := moves.DecodeTimers(body); err == nil && snap.Host == r.host { |
| - if out.timers != nil { |
| - return nil, true, fmt.Errorf("the archive prefix holds more than one timers snapshot") |
| - } |
| - out.timers = &snap |
| - } |
| - } |
| + if err := json.Unmarshal(body, &out.manifest); err != nil { |
| + return nil, true, fmt.Errorf("the archive manifest: %w", err) |
| } |
| - if sealedKey == "" || manifests != 1 { |
| - return nil, true, fmt.Errorf("the archive prefix %s holds %d sealed object(s) and %d manifest(s), want one of each", prefix, boolInt(sealedKey != ""), manifests) |
| + am := out.manifest |
| + if am.Op != r.m.ID || am.Plain != r.seal.FilesPlain || (r.seal.FilesSealed != "" && am.Sealed != r.seal.FilesSealed) { |
| + return nil, true, fmt.Errorf("the archive manifest names op %s, plain %s, sealed %s", am.Op, am.Plain, am.Sealed) |
| } |
| - if out.sealed, err = st.Get(ctx, cfg.OwnBucket, sealedKey); err != nil { |
| - return nil, false, fmt.Errorf("read %s: %w", sealedKey, err) |
| + if out.sealed, err = r.object(ctx, offlineArchiveSealed); err != nil { |
| + return nil, false, err |
| } |
| return out, false, nil |
| } |
| -func boolInt(b bool) int { |
| - if b { |
| - return 1 |
| +// object reads one of the sealer's objects under the move's archive prefix |
| +// in this box's bucket. |
| +func (r *landRun) object(ctx context.Context, name string) ([]byte, error) { |
| + st, err := r.store(ctx) |
| + if err != nil { |
| + return nil, err |
| } |
| - return 0 |
| + return st.Get(ctx, r.cfg().OwnBucket, moves.OfflineArchivePrefix(r.host, r.m.ID)+name) |
| } |
| // filesMatch reports whether the flat files directory and <host>.env are |
| -// the seal's, by the archive's own plaintext digest over the pair. |
| -func (r *landRun) filesMatch() (bool, string) { |
| +// the seal's, by the archive's own plaintext digest over the pair. A seal |
| +// of a host with no files directory carries no files/ entry at all and |
| +// requires no file; extraction then leaves an empty directory, whose digest |
| +// with the env file is the seal's. |
| +func (r *landRun) filesMatch() bool { |
| if !isDir(r.filesDir()) { |
| - return false, "the files directory is absent" |
| + return false |
| } |
| plain, _, _, err := archive.PlainDigest(r.filesDir(), r.envPath()) |
| - if err != nil { |
| - return false, err.Error() |
| - } |
| - if plain != r.seal.FilesPlain { |
| - return false, "digest " + plain |
| - } |
| - return true, "" |
| + return err == nil && plain == r.seal.FilesPlain |
| } |
| // files: <DataDir>/<host>/ and <host>.env carry the seal's plaintext digest. |
| @@ -842,48 +803,48 @@ func (r *landRun) filesMatch() (bool, string) { |
| // different: the files directory (when not empty) and an env file that is |
| // not the seal's are renamed into target-before first, then extracted. |
| func (r *landRun) files(ctx context.Context) { |
| - if ok, _ := r.filesMatch(); ok { |
| + if r.filesMatch() { |
| r.ok("files") |
| return |
| } |
| cfg := r.cfg() |
| if cfg.ConfigPriv == nil { |
| - r.wait("files", "this box has no CARLOS_SECRETS_KEY to open the archive with") |
| + r.landWait("files", "this box holds no CARLOS_SECRETS_KEY to open the archive with") |
| return |
| } |
| a, hard, err := r.readArchive(ctx) |
| - if err != nil { |
| - if hard { |
| - r.fail("files", "%v", err) |
| - } else { |
| - r.wait("files", "%v", err) |
| - } |
| + switch { |
| + case err != nil && hard: |
| + r.landFailErr("files", "the archive is not the one the seal names", err) |
| + return |
| + case err != nil: |
| + r.landWaitErr("files", "the archive could not be read", err) |
| return |
| } |
| bind := appconfig.Binding{Account: r.rec.Account, App: r.rec.App} |
| if err := archive.Verify(bytes.NewReader(a.sealed), a.manifest, cfg.ConfigPriv, bind); err != nil { |
| - r.fail("files", "the archive does not verify: %v", err) |
| + r.landFailErr("files", "the archive does not verify", err) |
| return |
| } |
| if nonEmptyOrNotDir(r.filesDir()) { |
| if err := parkInto(r.filesDir(), r.beforeDir()); err != nil { |
| - r.wait("files", "the files directory could not be parked: %v", err) |
| + r.landWaitErr("files", "the files directory the target held could not be parked", err) |
| return |
| } |
| } |
| if lexists(r.envPath()) && !sameSHA(r.seal.EnvSHA256, hexFileSHA256(r.envPath())) { |
| if err := parkInto(r.envPath(), r.beforeDir()); err != nil { |
| - r.wait("files", "the env file could not be parked: %v", err) |
| + r.landWaitErr("files", "the env file the target held could not be parked", err) |
| return |
| } |
| } |
| if err := extractSealed(a, cfg, bind, r.filesDir(), r.envPath()); err != nil { |
| - r.wait("files", "the archive could not be extracted: %v", err) |
| + r.landWaitErr("files", "the archive could not be extracted", err) |
| return |
| } |
| log.Printf("moves: %s: offline move %s: extracted the files and env", r.host, r.m.ID) |
| - if ok, why := r.filesMatch(); !ok { |
| - r.fail("files", "after extraction the files are not the seal's (%s, sealed %s)", why, r.seal.FilesPlain) |
| + if !r.filesMatch() { |
| + r.landFail("files", "after extraction the files are not the seal's") |
| return |
| } |
| r.ok("files") |
| @@ -951,30 +912,30 @@ func sameSHA(want, gotHex string) bool { |
| func (r *landRun) config(ctx context.Context) { |
| cfg := r.cfg() |
| if cfg.RenderHeld == nil { |
| - r.wait("config", "this box has no held render") |
| + r.landWait("config", "this box has no held render") |
| return |
| } |
| hr, err := cfg.RenderHeld(ctx, r.s, *r.rt, r.m.ID, nil, func(string) error { return nil }) |
| if err != nil { |
| - r.wait("config", "the config could not be rendered: %v", err) |
| + r.landWaitErr("config", "the config could not be rendered", err) |
| return |
| } |
| if hr.Path != r.configPath() { |
| - r.fail("config", "rendered at %s, not %s", hr.Path, r.configPath()) |
| + r.landFail("config", "the config was rendered at another path") |
| return |
| } |
| if hexFileSHA256(hr.Path) != hr.SHA256 { |
| - r.wait("config", "%s does not hold the rendered bytes", hr.Path) |
| + r.landWait("config", "the config file does not hold the rendered bytes") |
| return |
| } |
| body, _, err := r.s.Get(ctx, appconfig.ConfigAckKey(cfg.Box, r.host)) |
| if err != nil { |
| - r.wait("config", "the config ack could not be read: %v", err) |
| + r.landWaitErr("config", "the config ack could not be read", err) |
| return |
| } |
| var ack appconfig.ConfigAck |
| if err := json.Unmarshal(body, &ack); err != nil || ack.MoveOp != r.m.ID || ack.FileSHA256 != hr.SHA256 { |
| - r.wait("config", "the config ack does not name this move and the rendered file") |
| + r.landWait("config", "the config ack does not name this move and the rendered file") |
| return |
| } |
| r.ok("config") |
| @@ -987,28 +948,31 @@ func (r *landRun) config(ctx context.Context) { |
| // files item lands them; this is the check. |
| func (r *landRun) env() { |
| if !r.isOK("files") { |
| - r.wait("env", "waits for files") |
| + r.landWait("env", "waits for the files item") |
| return |
| } |
| own, err := os.ReadFile(r.envPath()) |
| switch { |
| case errors.Is(err, os.ErrNotExist) && r.seal.EnvSHA256 == "": |
| + case errors.Is(err, os.ErrNotExist): |
| + r.landFail("env", "the env file is missing") |
| + return |
| case err != nil: |
| - r.fail("env", "%s: %v", r.envPath(), err) |
| + r.landWaitErr("env", "the env file could not be read", err) |
| return |
| default: |
| if sum := sha256.Sum256(own); !sameSHA(r.seal.EnvSHA256, hex.EncodeToString(sum[:])) { |
| - r.fail("env", "%s is not the source's (sha256 %s, sealed %s)", r.envPath(), hex.EncodeToString(sum[:]), r.seal.EnvSHA256) |
| + r.landFail("env", "the env file is not the source's") |
| return |
| } |
| } |
| if !r.isOK("config") { |
| - r.wait("env", "waits for config: the instance key is read from both files") |
| + r.landWait("env", "waits for the config item: the instance key is read from both files") |
| return |
| } |
| cfgEnv, err := os.ReadFile(r.configPath()) |
| if err != nil && !errors.Is(err, os.ErrNotExist) { |
| - r.wait("env", "%s: %v", r.configPath(), err) |
| + r.landWaitErr("env", "the config file could not be read", err) |
| return |
| } |
| key, _ := instancekey.Effective(instancekey.ParseEnv(cfgEnv), instancekey.ParseEnv(own)) |
| @@ -1017,46 +981,81 @@ func (r *landRun) env() { |
| got = instancekey.Digest(key) |
| } |
| if !sameSHA(r.seal.KeySHA256, got) { |
| - r.fail("env", "the instance key is not the source's (sha256 %q, sealed %q)", got, r.seal.KeySHA256) |
| + r.landFail("env", "the instance key is not the source's") |
| return |
| } |
| r.ok("env") |
| } |
| +// pinNames is the variable names a failed env pin reports. It is not a |
| +// string, so a reason that prints it carries names, never values. |
| +type pinNames []string |
| + |
| +func (n pinNames) String() string { return strings.Join(n, ", ") } |
| + |
| // envPin: the composed launch environment of the held row |
| // (ExecBackend.LaunchLayers) has the source's portable digest. App settings |
| -// must be equal; box settings are left out (moves.Pinned). |
| -func (r *landRun) envPin() { |
| +// must be equal; box settings are left out (moves.Pinned). On a mismatch the |
| +// reason names the variables that differ (moves.CompareEnv against the |
| +// source's report the sealer wrote beside the archive), never their values. |
| +func (r *landRun) envPin(ctx context.Context) { |
| cfg := r.cfg() |
| if !r.isOK("files") || !r.isOK("config") { |
| - r.wait("env-pin", "waits for files and config") |
| + r.landWait("env-pin", "waits for the files and config items") |
| return |
| } |
| if r.m.Source.EnvDigest != "" && r.m.Source.EnvDigest != r.seal.EnvDigest { |
| - r.fail("env-pin", "the record's env digest and the seal's disagree") |
| + r.landFail("env-pin", "the record and the seal disagree on this item") |
| return |
| } |
| if cfg.LaunchLayers == nil { |
| - r.wait("env-pin", "this box has no exec backend to compose the launch environment") |
| + r.landWait("env-pin", "this box has no exec backend to compose the launch environment") |
| return |
| } |
| key, err := moves.ReportKeyFromSecrets(cfg.SecretsKey) |
| if err != nil { |
| - r.wait("env-pin", "no report key: %v", err) |
| + r.landWaitErr("env-pin", "this box holds no CARLOS_SECRETS_KEY to key the env report", err) |
| return |
| } |
| env, inst, _, err := cfg.LaunchLayers(*r.rt) |
| if err != nil { |
| - r.wait("env-pin", "the launch environment could not be composed: %v", err) |
| + r.landWaitErr("env-pin", "the launch environment could not be composed", err) |
| + return |
| + } |
| + target := moves.ReportLaunch(key, env, inst) |
| + if target.Digest == r.seal.EnvDigest { |
| + r.ok("env-pin") |
| return |
| } |
| - if got := moves.PortableDigest(key, env, inst); got != r.seal.EnvDigest { |
| - // The seal carries the source's digest only, not its per-variable |
| - // report, so CompareEnv cannot name the variables here. |
| - r.fail("env-pin", "the launch environment differs from the source's (%s, sealed %s)", got, r.seal.EnvDigest) |
| + names := r.envDiff(ctx, target) |
| + if len(names) == 0 { |
| + r.landFail("env-pin", "the launch environment differs from the source's") |
| return |
| } |
| - r.ok("env-pin") |
| + r.landFail("env-pin", fmt.Sprintf("the launch environment differs from the source's in %v", names)) |
| +} |
| + |
| +// envDiff is the names CompareEnv finds between the source's env report |
| +// (the sealer's offlineEnvReportObject, used only when its digest is the |
| +// seal's EnvDigest) and the target's, notes left out. None when the report |
| +// cannot be read or trusted. |
| +func (r *landRun) envDiff(ctx context.Context, target moves.EnvReport) pinNames { |
| + body, err := r.object(ctx, offlineEnvReportObject) |
| + if err != nil { |
| + r.w.notice(r.host+"#offline-land-envreport", "moves: %s: offline move %s: the source's env report: %v", r.host, r.m.ID, err) |
| + return nil |
| + } |
| + var src moves.EnvReport |
| + if json.Unmarshal(body, &src) != nil || !moves.CurrentEnvReport(src) || src.Digest != r.seal.EnvDigest { |
| + return nil |
| + } |
| + var out pinNames |
| + for _, d := range moves.CompareEnv(src, target, true, r.m.To == "") { |
| + if !d.Note() { |
| + out = append(out, d.Name+" ("+d.Kind+")") |
| + } |
| + } |
| + return out |
| } |
| // timers: the row's timer rows digest to the seal's. Otherwise they are |
| @@ -1065,48 +1064,42 @@ func (r *landRun) envPin() { |
| func (r *landRun) timers(ctx context.Context) { |
| reg := r.cfg().Registry |
| if reg == nil { |
| - r.wait("timers", "no registry on this box") |
| + r.landWait("timers", "no registry on this box") |
| return |
| } |
| if r.m.Source.TimersSHA256 != "" && r.m.Source.TimersSHA256 != r.seal.TimersDigest { |
| - r.fail("timers", "the record's timers digest and the seal's disagree") |
| + r.landFail("timers", "the record and the seal disagree on this item") |
| return |
| } |
| rows, err := reg.TimersFor(r.host) |
| if err != nil { |
| - r.wait("timers", "%v", err) |
| + r.landWaitErr("timers", "the timers could not be read", err) |
| return |
| } |
| if moves.TimersDigest(rows) == r.seal.TimersDigest { |
| r.ok("timers") |
| return |
| } |
| - a, hard, err := r.readArchive(ctx) |
| - switch { |
| - case err != nil && hard: |
| - r.fail("timers", "%v", err) |
| - return |
| - case err != nil: |
| - r.wait("timers", "%v", err) |
| - return |
| - case a.timers == nil: |
| - r.fail("timers", "the archive prefix holds no timers snapshot for %s", r.host) |
| + body, err := r.object(ctx, offlineTimersObject) |
| + if err != nil { |
| + r.landWaitErr("timers", "the timers snapshot could not be read", err) |
| return |
| } |
| - if got := moves.TimersDigest(a.timers.Rows); got != r.seal.TimersDigest { |
| - r.fail("timers", "the timers snapshot digests to %s, the seal %s", got, r.seal.TimersDigest) |
| + snap, err := moves.DecodeTimers(body) |
| + if err != nil || snap.Host != r.host || moves.TimersDigest(snap.Rows) != r.seal.TimersDigest { |
| + r.landFail("timers", "the timers snapshot is not the one the seal names") |
| return |
| } |
| - if err := reg.ReplaceTimers(r.host, a.timers.Rows); err != nil { |
| - r.wait("timers", "replace the timers: %v", err) |
| + if err := reg.ReplaceTimers(r.host, snap.Rows); err != nil { |
| + r.landWaitErr("timers", "the timers could not be replaced", err) |
| return |
| } |
| if rows, err = reg.TimersFor(r.host); err != nil { |
| - r.wait("timers", "%v", err) |
| + r.landWaitErr("timers", "the timers could not be read", err) |
| return |
| } |
| - if got := moves.TimersDigest(rows); got != r.seal.TimersDigest { |
| - r.fail("timers", "the timers read back as %s, the seal %s", got, r.seal.TimersDigest) |
| + if moves.TimersDigest(rows) != r.seal.TimersDigest { |
| + r.landFail("timers", "the timers read back differently from the seal's") |
| return |
| } |
| r.ok("timers") |
| @@ -1115,15 +1108,19 @@ func (r *landRun) timers(ctx context.Context) { |
| // reread refreshes the row after a column write. |
| func (r *landRun) reread(id string) bool { |
| rt, err := r.w.offlineRow(r.host) |
| - if err != nil || rt == nil { |
| - r.wait(id, "the row could not be read back: %v", err) |
| + if err == nil && rt == nil { |
| + err = errors.New("the row is gone") |
| + } |
| + if err != nil { |
| + r.landWaitErr(id, "the row could not be read back", err) |
| return false |
| } |
| r.rt = rt |
| return true |
| } |
| -// nextDue: the row's NextDue is the seal's; else set and read back. |
| +// nextDue: the row's NextDue is the seal's; else set and read back. A write |
| +// error waits; only a read-back that differs fails. |
| func (r *landRun) nextDue() { |
| if r.rt.NextDue == r.seal.NextDue { |
| r.ok("next-due") |
| @@ -1131,25 +1128,25 @@ func (r *landRun) nextDue() { |
| } |
| reg := r.cfg().Registry |
| if reg == nil { |
| - r.wait("next-due", "no registry on this box") |
| + r.landWait("next-due", "no registry on this box") |
| return |
| } |
| if err := reg.SetNextDue(r.host, r.seal.NextDue); err != nil { |
| - r.wait("next-due", "%v", err) |
| + r.landWaitErr("next-due", "next due could not be set", err) |
| return |
| } |
| if !r.reread("next-due") { |
| return |
| } |
| if r.rt.NextDue != r.seal.NextDue { |
| - r.fail("next-due", "reads back %d, the seal %d", r.rt.NextDue, r.seal.NextDue) |
| + r.landFail("next-due", "next due reads back differently from the seal's") |
| return |
| } |
| r.ok("next-due") |
| } |
| // replication: the row's replication rung is the seal's; else set and read |
| -// back. |
| +// back. A write error waits; only a read-back that differs fails. |
| func (r *landRun) replication() { |
| if r.rt.Replication == r.seal.Replication { |
| r.ok("replication") |
| @@ -1157,18 +1154,18 @@ func (r *landRun) replication() { |
| } |
| reg := r.cfg().Registry |
| if reg == nil { |
| - r.wait("replication", "no registry on this box") |
| + r.landWait("replication", "no registry on this box") |
| return |
| } |
| if err := reg.SetReplication(r.host, r.seal.Replication); err != nil { |
| - r.fail("replication", "%v", err) |
| + r.landWaitErr("replication", "the replication rung could not be set", err) |
| return |
| } |
| if !r.reread("replication") { |
| return |
| } |
| if r.rt.Replication != r.seal.Replication { |
| - r.fail("replication", "reads back %q, the seal %q", r.rt.Replication, r.seal.Replication) |
| + r.landFail("replication", "the replication rung reads back differently from the seal's") |
| return |
| } |
| r.ok("replication") |
| @@ -1180,20 +1177,21 @@ func (r *landRun) replication() { |
| func (r *landRun) release(ctx context.Context) { |
| cfg := r.cfg() |
| if r.m.Source.Release != "" && r.m.Source.Release != r.seal.Release { |
| - r.fail("release", "the record's release and the seal's disagree") |
| + r.landFail("release", "the record and the seal disagree on this item") |
| return |
| } |
| if cfg.ChannelVersion == nil { |
| - r.wait("release", "this box cannot resolve channels") |
| + r.landWait("release", "this box cannot resolve channels") |
| return |
| } |
| v, err := cfg.ChannelVersion(ctx, r.s, *r.rt) |
| if err != nil { |
| - r.wait("release", "the channel could not be resolved: %v", err) |
| + r.landWaitErr("release", "the channel could not be resolved", err) |
| return |
| } |
| if v != r.seal.Release { |
| - r.fail("release", "the channel points at %q, the seal recorded %q", v, r.seal.Release) |
| + r.w.notice(r.host+"#offline-land-release", "moves: %s: offline move %s: the channel points at %q, the seal recorded %q", r.host, r.m.ID, v, r.seal.Release) |
| + r.landFail("release", "the channel points at another build than the sealed one") |
| return |
| } |
| r.ok("release") |
diff --git a/internal/host/movesoffline_land_test.go b/internal/host/movesoffline_land_test.go| index 51593c609..4d2d99d42 100644 |
| --- a/internal/host/movesoffline_land_test.go |
| +++ b/internal/host/movesoffline_land_test.go |
| @@ -3,31 +3,20 @@ package host |
| import ( |
| "bytes" |
| "context" |
| - "crypto/ecdh" |
| "crypto/ed25519" |
| - "crypto/md5" |
| - "crypto/sha256" |
| - "database/sql" |
| - "encoding/hex" |
| "encoding/json" |
| - "encoding/xml" |
| "errors" |
| - "fmt" |
| - "io" |
| "io/fs" |
| - "net/http" |
| - "net/http/httptest" |
| "os" |
| - "path" |
| "path/filepath" |
| "reflect" |
| "sort" |
| - "strconv" |
| "strings" |
| "testing" |
| "time" |
| - "github.com/superfly/ltx" |
| + "crypto/sha256" |
| + "encoding/hex" |
| "github.com/carlosframework/platform/internal/activator" |
| "github.com/carlosframework/platform/internal/appconfig" |
| @@ -37,7 +26,6 @@ import ( |
| "github.com/carlosframework/platform/internal/lease" |
| "github.com/carlosframework/platform/internal/lease/leasetest" |
| "github.com/carlosframework/platform/internal/moves" |
| - "github.com/carlosframework/platform/internal/moves/archive" |
| "github.com/carlosframework/platform/internal/provision" |
| "github.com/carlosframework/platform/internal/registry" |
| "github.com/carlosframework/platform/internal/release" |
| @@ -47,153 +35,24 @@ import ( |
| // (Task 5 of docs/superpowers/plans/2026-10-20-offline-operator-move.md), |
| // tested through the functions the agent calls (newMovesWorker, armOffline, |
| // reconcileSerializedWith, the edges-file writer and movesPass), over one |
| -// DirStore for the deployment bucket, leasetest's S3 model for the data |
| -// buckets data-ie and data-uk, real registries, a real litestream replica |
| -// written by lease.ReplicateWithLease, the real archive, the real held |
| -// render and the exec backend's real launch composition. |
| +// DirStore for the deployment bucket, leasetest's S3 server for the data |
| +// buckets data-ie and data-uk, real registries, a real litestream replica, |
| +// the real archive, the real held render and the exec backend's real launch |
| +// composition. |
| // |
| -// The seal is this file's fixture (landWorld.sealFixture): Task 4's sealer |
| -// runs in parallel and is not on this branch. It writes what the sealer's |
| -// brief says the sealer writes, through the same functions: the staged |
| -// replica by lease.MirrorPrefix, the archive by archive.Build and Seal, the |
| -// timers snapshot, the env pin from LaunchLayers, and seal.json create-only |
| -// with the record's evidence. |
| - |
| -// landS3 is the data buckets of the land tests: leasetest's S3 model and its |
| -// wire server, with three calls the server does not speak yet answered from |
| -// the same model: ListObjectsV2 (litestream's replica client and |
| -// lease.NewS3Mirror list that way; Task 4 is adding a minimal one to |
| -// leasetest), DeleteObjects and HEAD. Every other request is |
| -// leasetest.NewS3Server's. |
| -type landS3 struct { |
| - f *leasetest.VersionedBucket |
| - srv *httptest.Server |
| -} |
| - |
| -func newLandS3(t *testing.T, buckets ...string) *landS3 { |
| - t.Helper() |
| - f := leasetest.NewVersionedBucket() |
| - for _, b := range buckets { |
| - f.AddBucket(b) |
| - } |
| - inner, _ := leasetest.NewS3Server(f) |
| - s := &landS3{f: f} |
| - s.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { // 127.0.0.1 only |
| - bucket, rawKey, _ := strings.Cut(strings.TrimPrefix(r.URL.EscapedPath(), "/"), "/") |
| - q := r.URL.Query() |
| - switch { |
| - case r.Method == http.MethodGet && rawKey == "" && q.Get("list-type") == "2": |
| - s.listV2(w, r.Context(), bucket, q.Get("prefix"), q.Get("delimiter")) |
| - case r.Method == http.MethodPost && rawKey == "" && q.Has("delete"): |
| - s.deleteObjects(w, r, bucket) |
| - case r.Method == http.MethodHead && rawKey != "": |
| - key := strings.ReplaceAll(rawKey, "%2F", "/") |
| - if body, ok := f.Current(bucket, key); ok { |
| - sum := md5.Sum(body) |
| - w.Header().Set("Content-Length", strconv.Itoa(len(body))) |
| - w.Header().Set("ETag", `"`+hex.EncodeToString(sum[:])+`"`) |
| - w.WriteHeader(http.StatusOK) |
| - return |
| - } |
| - w.WriteHeader(http.StatusNotFound) |
| - default: |
| - inner.ServeHTTP(w, r) |
| - } |
| - })) |
| - t.Cleanup(s.srv.Close) |
| - for k, v := range map[string]string{ |
| - "CARLOS_S3_ENDPOINT": s.srv.URL, "AWS_ACCESS_KEY_ID": "test-key", "AWS_SECRET_ACCESS_KEY": "test-secret", |
| - "AWS_SESSION_TOKEN": "", "AWS_PROFILE": "", "AWS_REGION": "", "AWS_EC2_METADATA_DISABLED": "true", |
| - "AWS_CONFIG_FILE": filepath.Join(t.TempDir(), "absent"), "AWS_SHARED_CREDENTIALS_FILE": filepath.Join(t.TempDir(), "absent"), |
| - } { |
| - t.Setenv(k, v) |
| - } |
| - return s |
| -} |
| - |
| -type v2Object struct { |
| - Key string `xml:"Key"` |
| - LastModified string `xml:"LastModified"` |
| - ETag string `xml:"ETag"` |
| - Size int64 `xml:"Size"` |
| - StorageClass string `xml:"StorageClass"` |
| -} |
| +// The seal is Task 4's: the source box ie-a runs the sealer step |
| +// (offlineSealerStep, through offlineRound) over T4's seal world, so the |
| +// staged replica, the archive, its manifest, the timers snapshot, the env |
| +// report and seal.json are exactly what the sealer writes. |
| -type v2Prefix struct { |
| - Prefix string `xml:"Prefix"` |
| -} |
| - |
| -type v2Result struct { |
| - XMLName xml.Name `xml:"ListBucketResult"` |
| - Name string `xml:"Name"` |
| - Prefix string `xml:"Prefix"` |
| - Delimiter string `xml:"Delimiter,omitempty"` |
| - KeyCount int `xml:"KeyCount"` |
| - MaxKeys int `xml:"MaxKeys"` |
| - IsTruncated bool `xml:"IsTruncated"` |
| - Contents []v2Object `xml:"Contents"` |
| - CommonPrefixes []v2Prefix `xml:"CommonPrefixes"` |
| -} |
| - |
| -// listV2 answers ListObjectsV2 with the current objects under prefix, in one |
| -// page, rolling keys up at the delimiter as S3 does. |
| -func (s *landS3) listV2(w http.ResponseWriter, ctx context.Context, bucket, prefix, delim string) { |
| - vers, err := s.f.ListVersions(ctx, bucket, prefix) |
| - if err != nil { |
| - http.Error(w, err.Error(), http.StatusNotFound) |
| - return |
| - } |
| - res := v2Result{Name: bucket, Prefix: prefix, Delimiter: delim, MaxKeys: 1000} |
| - seen := map[string]bool{} |
| - for _, v := range vers { |
| - if !v.IsLatest || v.DeleteMarker { |
| - continue |
| - } |
| - if delim != "" { |
| - if i := strings.Index(v.Key[len(prefix):], delim); i >= 0 { |
| - cp := v.Key[:len(prefix)+i+len(delim)] |
| - if !seen[cp] { |
| - seen[cp] = true |
| - res.CommonPrefixes = append(res.CommonPrefixes, v2Prefix{cp}) |
| - } |
| - continue |
| - } |
| - } |
| - res.Contents = append(res.Contents, v2Object{Key: v.Key, LastModified: v.LastModified.UTC().Format(time.RFC3339), ETag: `"` + v.ETag + `"`, Size: v.Size, StorageClass: "STANDARD"}) |
| - } |
| - sort.Slice(res.Contents, func(i, j int) bool { return res.Contents[i].Key < res.Contents[j].Key }) |
| - res.KeyCount = len(res.Contents) + len(res.CommonPrefixes) |
| - w.Header().Set("Content-Type", "application/xml") |
| - xml.NewEncoder(w).Encode(res) |
| -} |
| +const ( |
| + landKeyValue = "the-instance-key" // the key sealWorld.instance writes |
| + landStore = "shop-store-ie" |
| + landSecret = "store-secret-value" |
| +) |
| -func (s *landS3) deleteObjects(w http.ResponseWriter, r *http.Request, bucket string) { |
| - var req struct { |
| - Objects []struct { |
| - Key string `xml:"Key"` |
| - } `xml:"Object"` |
| - } |
| - body, _ := io.ReadAll(r.Body) |
| - if err := xml.Unmarshal(body, &req); err != nil { |
| - http.Error(w, err.Error(), http.StatusBadRequest) |
| - return |
| - } |
| - type deleted struct { |
| - Key string `xml:"Key"` |
| - } |
| - out := struct { |
| - XMLName xml.Name `xml:"DeleteResult"` |
| - Deleted []deleted `xml:"Deleted"` |
| - }{} |
| - for _, o := range req.Objects { |
| - if _, ok := s.f.Current(bucket, o.Key); ok { |
| - s.f.BareDelete(bucket, o.Key) |
| - } |
| - out.Deleted = append(out.Deleted, deleted{o.Key}) |
| - } |
| - w.Header().Set("Content-Type", "application/xml") |
| - xml.NewEncoder(w).Encode(out) |
| -} |
| +// landS3 reads the data buckets' model directly. |
| +type landS3 struct{ f *leasetest.VersionedBucket } |
| // current is the current objects under prefix in bucket. |
| func (s *landS3) current(t *testing.T, bucket, prefix string) map[string][]byte { |
| @@ -212,59 +71,48 @@ func (s *landS3) current(t *testing.T, bucket, prefix string) map[string][]byte |
| return out |
| } |
| -// --- the world --------------------------------------------------------------- |
| - |
| -const ( |
| - landKeyValue = "ik_0123456789abcdef0123456789abcdef" |
| - landStore = "shop-store-ie" |
| - landSecret = "store-secret-value" |
| -) |
| +// landOpts shapes the source: configKey sets the instance key in the app's |
| +// configuration (and none in <host>.env), noFiles gives the host no files |
| +// directory, nested makes it a nested row with the launch descriptors a real |
| +// nested host keeps. |
| +type landOpts struct { |
| + configKey string |
| + noFiles bool |
| + nested bool |
| +} |
| -// landWorld is offHost at home (ie-a, data-ie) moving to uk (uk-a, |
| +// landWorld is offHost at home (ie-a, data-ie) moving to uk (uk-box, |
| // data-uk), with a real database replica, a files directory, an env file |
| // carrying the instance key, an app config bundle carrying the store |
| -// binding, timers, a NextDue, a replication rung and a promoted release. |
| +// binding, timers, a NextDue, a replication rung and a promoted release, |
| +// sealed by the real sealer. |
| type landWorld struct { |
| - t *testing.T |
| - wd *offWorld |
| - s3 *landS3 |
| - ie, uk *offBox |
| - ieEB *activator.ExecBackend |
| - ukEB *activator.ExecBackend |
| + t *testing.T |
| + sw *sealWorld |
| + wd *offWorld |
| + s3 *landS3 |
| + ie *offBox |
| + uk *offBox |
| + ukEB *activator.ExecBackend |
| - secrets string |
| - priv *ecdh.PrivateKey |
| relPriv ed25519.PrivateKey |
| seal moves.OfflineReport |
| sealRaw []byte |
| src provision.MoveSource |
| - // configKey: the app's configuration sets CARLOS_INSTANCE_KEY to it and |
| - // <host>.env carries none, so the key the app is handed comes from the |
| - // rendered <host>.config.env (instancekey.LayerConfig). "" is the usual |
| - // shape: the key minted into <host>.env. |
| - configKey string |
| + opts landOpts |
| + // The source's files directory, env file and config file. |
| + srcFiles, srcEnv, srcConfig string |
| } |
| -func newLandWorld(t *testing.T) *landWorld { return newLandWorldWith(t, "") } |
| +func newLandWorld(t *testing.T) *landWorld { return newLandWorldWith(t, landOpts{}) } |
| -// newLandWorldWith is newLandWorld with the instance key set by the app's |
| -// configuration (configKey), when not "". |
| -func newLandWorldWith(t *testing.T, configKey string) *landWorld { |
| +func newLandWorldWith(t *testing.T, opts landOpts) *landWorld { |
| t.Helper() |
| ctx := context.Background() |
| - lw := &landWorld{t: t, wd: newOffWorld(t), configKey: configKey} |
| - lw.s3 = newLandS3(t, "data-ie", "data-uk") |
| - var err error |
| - privB64, _, err := appconfig.GenerateKey() |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - lw.secrets = privB64 |
| - if lw.priv, err = appconfig.ParsePrivateKey(privB64, "test"); err != nil { |
| - t.Fatal(err) |
| - } |
| + sw := newSealWorld(t) |
| + lw := &landWorld{t: t, sw: sw, wd: sw.offWorld, s3: &landS3{f: sw.fake}, ie: sw.ie, uk: sw.uk, opts: opts} |
| seed, pub, err := release.GenerateKey() |
| if err != nil { |
| t.Fatal(err) |
| @@ -272,75 +120,80 @@ func newLandWorldWith(t *testing.T, configKey string) *landWorld { |
| if lw.relPriv, err = release.ParseSigningKey(seed, "test"); err != nil { |
| t.Fatal(err) |
| } |
| - s := lw.wd.s |
| - acct := lw.wd.acct |
| - shipVersion(t, s, acct, "app", "v1") |
| - shipVersion(t, s, acct, "app", "v2") |
| - lw.promote("v1") |
| + shipVersion(t, sw.s, sw.acct, "app", sealVersion) |
| + shipVersion(t, sw.s, sw.acct, "app", "v2") |
| + lw.promote(sealVersion) |
| lw.bundle(landStore) |
| - |
| - env := func(bucket, region string) map[string]string { |
| - return map[string]string{"CARLOS_HIBERNATE_BUCKET": bucket, "CARLOS_HIBERNATE_REGION": region, "CARLOS_SECRETS_KEY": privB64, "CARLOS_RELEASE_PUBKEY": pub} |
| - } |
| - lw.ie = lw.wd.box("ie-a", "", env("data-ie", "eu-west-1")) |
| - lw.uk = lw.wd.box("uk-a", "uk", env("data-uk", "eu-west-2")) |
| - lw.ieEB = &activator.ExecBackend{DataDir: lw.ie.dd, OptDir: lw.ie.opt} |
| + lw.uk.env["CARLOS_HIBERNATE_REGION"] = sealUKRegion |
| + lw.uk.env["CARLOS_RELEASE_PUBKEY"] = pub |
| lw.ukEB = &activator.ExecBackend{DataDir: lw.uk.dd, OptDir: lw.uk.opt, Region: "uk"} |
| - lw.arm(lw.ie) |
| lw.arm(lw.uk) |
| - // The source at rest: its row, then everything the instance holds. |
| - lw.ie.tick() |
| - lw.uk.tick() |
| - rt := lw.ie.row(offHost) |
| - if rt == nil { |
| - t.Fatal("the source has no row") |
| - } |
| - dd := lw.ie.dd |
| - writeFile(t, filepath.Join(dd, offHost, "notes.txt"), "the app's own file\n") |
| - writeFile(t, filepath.Join(dd, offHost, "uploads", "a.bin"), "\x00\x01binary\x02") |
| - ownKey := "CARLOS_INSTANCE_KEY=" + landKeyValue + "\n" |
| - if configKey != "" { |
| - ownKey = "" |
| - } |
| - writeFile(t, filepath.Join(dd, offHost+".env"), ownKey+"CARLOS_ADMIN_TOKEN=admin-token\nAPP_OVERRIDE=1\n") |
| - if err := lw.ie.reg.ReplaceTimers(offHost, []registry.Timer{ |
| - {Host: offHost, Name: "digest", Kind: "cron", Spec: "0 * * * *", Path: "/cron/digest", NextAt: 1_800_000_000, CreatedAt: 1_700_000_000, DueAt: 1_800_000_000}, |
| - {Host: offHost, Name: "reminder", Kind: "once", Spec: "", Path: "/once/reminder", NextAt: 1_800_000_500, CreatedAt: 1_700_000_100, DueAt: 1_800_000_500}, |
| - }); err != nil { |
| - t.Fatal(err) |
| + // The source at rest, as a box that has served it. |
| + ie := lw.ie |
| + sw.instance(ie) |
| + lw.srcFiles, lw.srcEnv, lw.srcConfig = filepath.Join(ie.dd, offHost), filepath.Join(ie.dd, offHost+".env"), filepath.Join(ie.dd, offHost+".config.env") |
| + if opts.nested { |
| + raw := sw.nest(ie) |
| + lw.srcFiles, lw.srcEnv, lw.srcConfig = filepath.Join(raw, offHost), filepath.Join(raw, offHost+".env"), filepath.Join(raw, offHost+".config.env") |
| + for _, name := range []string{"launch.run.json", "launch.seed.json", "launch.gen.7.json"} { |
| + writeFile(t, filepath.Join(raw, name), "{}") |
| + } |
| + // The source's env pin is composed from the instance directory's |
| + // own env and config files, as the agent's exec backend reads them |
| + // through the instance mount. The fixture's mount is a symlink, |
| + // which the nested reader refuses (and the seal world's backend, |
| + // with no socket directory, classifies the row foreign and reads |
| + // the flat paths), so here the raw directory stands in for the |
| + // mount: the same two files, composed by the same code. |
| + ie.w.cfg.LaunchLayers = func(rt registry.Route) (map[string]string, map[string]bool, []string, error) { |
| + rt.DBPath, rt.Socket = provision.DefaultDBPath(raw, rt.Host), filepath.Join(provision.SocketDir, rt.Host+".sock") |
| + return (&activator.ExecBackend{DataDir: raw}).LaunchLayers(rt) |
| + } |
| } |
| - if err := lw.ie.reg.SetNextDue(offHost, 1_800_000_000); err != nil { |
| - t.Fatal(err) |
| + if opts.configKey != "" { |
| + writeFile(t, lw.srcEnv, "CARLOS_ADMIN_TOKEN=token\n") |
| } |
| - if err := lw.ie.reg.SetReplication(offHost, registry.ReplicationBatched); err != nil { |
| - t.Fatal(err) |
| + if opts.noFiles { |
| + if err := os.RemoveAll(lw.srcFiles); err != nil { |
| + t.Fatal(err) |
| + } |
| } |
| - lw.replicate(rt.DBPath, "one", "two", "three") |
| - // start, the source holds and reports, its config is rendered held |
| - // (as SyncExec would have rendered it at rest), seal, sealed. |
| - lw.wd.start() |
| - lw.ie.tick() |
| - if _, err := lw.ie.w.cfg.RenderHeld(ctx, s, *lw.ie.row(offHost), offID, nil, func(string) error { return nil }); err != nil { |
| + // start and the source's hold; its config rendered held from the bundle |
| + // (as SyncExec renders it at rest); seal; the sealer's pass. |
| + sw.startAndSeal("ie-a", ie) |
| + hr, err := ie.w.cfg.RenderHeld(ctx, sw.s, *ie.row(offHost), offID, nil, func(string) error { return nil }) |
| + if err != nil { |
| t.Fatal(err) |
| } |
| - lw.wd.seal() |
| - lw.sealFixture() |
| - lw.wd.move(provision.OfflineSealedBy(offID, "ie-a", lw.src)) |
| + if hr.Path != lw.srcConfig { |
| + // The nested fixture's sockets are not production's, so the render |
| + // took the flat path; production renders into the instance |
| + // directory, which is where the row's launch reads it. |
| + if err := os.Rename(hr.Path, lw.srcConfig); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + sw.mustPass(ie) |
| + rec := sw.record() |
| + if rec.Move.State != provision.OfflineSealed { |
| + t.Fatalf("not sealed: %+v (%+v)", rec.Move, sw.report(moves.OfflineRoleSealer, "ie-a").Items) |
| + } |
| + var ok bool |
| + if lw.seal, lw.sealRaw, ok = sw.sealRecord(); !ok { |
| + t.Fatal("no seal.json") |
| + } |
| + lw.src = *rec.Move.Source |
| return lw |
| } |
| -// arm wires a box as armMovesPass would, over its own data bucket. |
| +// arm wires the target box as armMovesPass would, over its own data bucket. |
| func (lw *landWorld) arm(b *offBox) { |
| bucket, region := b.getenv("CARLOS_HIBERNATE_BUCKET"), b.getenv("CARLOS_HIBERNATE_REGION") |
| - eb := lw.ieEB |
| - if b == lw.uk { |
| - eb = lw.ukEB |
| - } |
| b.svc = activator.New(activator.Config{Bucket: bucket, Idle: time.Hour}, &fakeWakeBackend{}, b.reg) |
| cfg := moveCfg{Region: b.region, Box: b.name, DataDir: b.dd, OptDir: b.opt, Quiesced: b.svc.Quiesced, Route: b.reg.Get, Now: b.wd.clock, |
| - OwnBucket: bucket, OwnBucketRegion: region, OwnPrefix: "instances", |
| + OwnBucket: bucket, OwnBucketRegion: region, OwnPrefix: sealPrefix, |
| Mirror: func(ctx context.Context, dst string) (lease.MirrorStore, error) { |
| return lease.NewS3Mirror(ctx, region, dst) |
| }} |
| @@ -348,7 +201,7 @@ func (lw *landWorld) arm(b *offBox) { |
| cfg.Offline = newOfflineView(cfg, nil) |
| cfg.ReleaseHold = releaseHeld(b.reg, nil) |
| cfg.Procs = func(host string) (int, error) { return cgroup.HostProcs(b.cg, host, "") } |
| - cfg.LaunchLayers = eb.LaunchLayers |
| + cfg.LaunchLayers = lw.ukEB.LaunchLayers |
| b.w = newMovesWorker(cfg) |
| } |
| @@ -364,10 +217,10 @@ func (lw *landWorld) promote(v string) { |
| } |
| // bundle writes the app's config: the store binding, its secret and a |
| -// setting. |
| +// setting, and the instance key when the world's configuration sets it. |
| func (lw *landWorld) bundle(store string) { |
| lw.t.Helper() |
| - env, err := appconfig.SealBound(lw.priv.PublicKey(), appconfig.Binding{Account: lw.wd.acct, App: "app"}, []byte(landSecret)) |
| + env, err := appconfig.SealBound(lw.sw.priv.PublicKey(), appconfig.Binding{Account: lw.wd.acct, App: "app"}, []byte(landSecret)) |
| if err != nil { |
| lw.t.Fatal(err) |
| } |
| @@ -377,144 +230,14 @@ func (lw *landWorld) bundle(store string) { |
| "STORE_SECRET_KEY": {Kind: "secret", Envelope: env}, |
| "FEATURE": {Kind: "env", Value: "on"}, |
| }} |
| - if lw.configKey != "" { |
| - m.Vars[instancekey.Var] = appconfig.Var{Kind: "env", Value: lw.configKey} |
| + if lw.opts.configKey != "" { |
| + m.Vars[instancekey.Var] = appconfig.Var{Kind: "env", Value: lw.opts.configKey} |
| } |
| if err := moves.PutReplace(context.Background(), lw.wd.s, appconfig.Key(lw.wd.acct, "app"), mustJSON(lw.t, m)); err != nil { |
| lw.t.Fatal(err) |
| } |
| } |
| -// replicate writes a real SQLite database at dbPath with rows, replicated |
| -// to data-ie under the host's own prefix by litestream, and closed. |
| -func (lw *landWorld) replicate(dbPath string, rows ...string) { |
| - t := lw.t |
| - t.Helper() |
| - ctx := context.Background() |
| - db, err := sql.Open("sqlite", dbPath) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - defer db.Close() |
| - if _, err := db.Exec(`CREATE TABLE notes(id INTEGER PRIMARY KEY, body TEXT)`); err != nil { |
| - t.Fatal(err) |
| - } |
| - rep, err := lease.ReplicateWithLease(ctx, dbPath, "data-ie", activator.HibernatePrefix("instances", offHost), "eu-west-1") |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - for _, r := range rows { |
| - if _, err := db.Exec(`INSERT INTO notes(body) VALUES (?)`, r); err != nil { |
| - t.Fatal(err) |
| - } |
| - if err := rep.SyncAndWait(ctx); err != nil { |
| - t.Fatal(err) |
| - } |
| - } |
| - if err := rep.Close(ctx); err != nil { |
| - t.Fatal(err) |
| - } |
| -} |
| - |
| -// sealFixture writes what Task 4's sealer writes at seal, from the source as |
| -// it stands: the staged replica in data-uk, the archive, its manifest and the |
| -// timers snapshot under the move's archive prefix, and seal.json; and sets |
| -// lw.src, the record's MoveSource with the seal's evidence. |
| -func (lw *landWorld) sealFixture() { |
| - t := lw.t |
| - t.Helper() |
| - ctx := context.Background() |
| - ie := lw.ie |
| - rt := ie.row(offHost) |
| - m, err := lease.NewS3Mirror(ctx, "eu-west-1", "eu-west-2") |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - round, err := lease.MirrorPrefix(ctx, m, "data-ie", activator.HibernatePrefix("instances", offHost)+"/", "data-uk", activator.MovesPrefix(offHost), lease.Receipts{}) |
| - if err != nil || round.Deferred != 0 { |
| - t.Fatalf("mirror: %+v %v", round, err) |
| - } |
| - staged := map[string]moves.StagedObject{} |
| - var final ltx.TXID |
| - objs, err := m.List(ctx, "data-uk", activator.MovesPrefix(offHost)) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - for k, o := range objs { |
| - staged[strings.TrimPrefix(k, activator.MovesPrefix(offHost))] = moves.StagedObject{Size: o.Size, ETag: o.ETag} |
| - if _, maxTX, err := ltx.ParseFilename(path.Base(k)); err == nil && maxTX > final { |
| - final = maxTX |
| - } |
| - } |
| - if final == 0 { |
| - t.Fatal("nothing staged") |
| - } |
| - |
| - filesDir, envPath := filepath.Join(ie.dd, offHost), filepath.Join(ie.dd, offHost+".env") |
| - var plainBuf, sealedBuf bytes.Buffer |
| - plain, entries, n, err := archive.Build(filesDir, envPath, &plainBuf) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - dk, sealedSHA, err := archive.Seal(&plainBuf, lw.priv.PublicKey(), appconfig.Binding{Account: lw.wd.acct, App: "app"}, &sealedBuf) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - man := archive.Manifest{V: archive.ManifestVersion, Op: offID, Plain: plain, Sealed: sealedSHA, Bytes: n, Entries: entries, SealedDataKey: dk, Chunk: archive.ChunkSize} |
| - prefix := moves.OfflineArchivePrefix(offHost, offID) |
| - lw.s3.f.Put("data-uk", prefix+"files.sealed", sealedBuf.Bytes()) |
| - lw.s3.f.Put("data-uk", prefix+"files.json", mustJSON(t, man)) |
| - timers, err := ie.reg.TimersFor(offHost) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - active := moves.ActiveTimers(timers) |
| - snap, err := moves.TimersSnapshot{Host: offHost, Rows: active, TakenAt: lw.wd.clock()}.Encode() |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - lw.s3.f.Put("data-uk", prefix+"timers.json", snap) |
| - |
| - envBytes, err := os.ReadFile(envPath) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - cfgBytes, _ := os.ReadFile(filepath.Join(ie.dd, offHost+".config.env")) |
| - key, _ := instancekey.Effective(instancekey.ParseEnv(cfgBytes), instancekey.ParseEnv(envBytes)) |
| - rk, err := moves.ReportKeyFromSecrets(lw.secrets) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - launch, inst, _, err := lw.ieEB.LaunchLayers(*rt) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - envSum := sha256.Sum256(envBytes) |
| - items := map[string]moves.Check{} |
| - for _, it := range moves.OfflineSourced() { |
| - items[it.ID] = moves.Check{OK: true} |
| - } |
| - lw.seal = moves.OfflineReport{ |
| - V: moves.OfflineReportVersion, Op: offID, Host: offHost, Box: "ie-a", Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseSealed, |
| - At: lw.wd.clock(), TombstoneOwner: moves.TombstoneOwner(offID), Items: items, |
| - FinalTXID: uint64(final), Staged: staged, FilesPlain: plain, FilesSealed: sealedSHA, FilesBytes: n, FilesEntries: len(entries), |
| - EnvSHA256: hex.EncodeToString(envSum[:]), KeySHA256: instancekey.Digest(key), EnvDigest: moves.PortableDigest(rk, launch, inst), |
| - Release: "v1", TimersDigest: moves.TimersDigest(active), Replication: rt.Replication, NextDue: rt.NextDue, |
| - } |
| - if err := moves.ValidateOfflineSeal(lw.seal, offHost, offID); err != nil { |
| - t.Fatal(err) |
| - } |
| - lw.sealRaw = mustJSON(t, lw.seal) |
| - if _, created, err := putOfflineCreateOnly(ctx, lw.wd.s, moves.OfflineSealKey(offHost, offID), lw.sealRaw); err != nil || !created { |
| - t.Fatalf("seal.json: created %v, %v", created, err) |
| - } |
| - lw.src = provision.MoveSource{ |
| - FinalTXID: uint64(final), Files: &provision.MoveFilesDigest{Plain: plain, Sealed: sealedSHA, Bytes: n, Entries: len(entries)}, |
| - TimersSHA256: lw.seal.TimersDigest, Release: "v1", EnvDigest: lw.seal.EnvDigest, Replication: rt.Replication, NextDue: rt.NextDue, |
| - Evidence: moves.OfflineSealEvidence(offHost, offID, lw.sealRaw), |
| - } |
| -} |
| - |
| // flipAndLand flips the record and runs the target box's ticks. |
| func (lw *landWorld) flipAndLand(ticks int) { |
| lw.wd.flip() |
| @@ -525,7 +248,7 @@ func (lw *landWorld) flipAndLand(ticks int) { |
| func (lw *landWorld) report() moves.OfflineReport { |
| lw.t.Helper() |
| - return lw.wd.report(moves.OfflineRoleTarget, "uk-a") |
| + return lw.wd.report(moves.OfflineRoleTarget, "uk-box") |
| } |
| func (lw *landWorld) state() string { |
| @@ -553,6 +276,9 @@ func (lw *landWorld) requireLanded() { |
| if len(r.Items) != len(moves.OfflineOwned(moves.OfflineOwnerLanding)) { |
| t.Fatalf("report carries %d items, want every landing item", len(r.Items)) |
| } |
| + if !reflect.DeepEqual(r.Staged, lw.seal.Staged) { |
| + t.Fatalf("the report's verified listing %v, the seal's %v", keysOf(r.Staged), keysOf(lw.seal.Staged)) |
| + } |
| rt := lw.uk.row(offHost) |
| if rt == nil || rt.Status != "landing" || rt.HoldOp != offID { |
| t.Fatalf("target row %+v", rt) |
| @@ -569,13 +295,13 @@ func (lw *landWorld) requireLanded() { |
| // host's lock is free. |
| func (lw *landWorld) requireNoTargetLease() { |
| lw.t.Helper() |
| - loc := activator.HibernatePrefix("instances", offHost) |
| - for k := range lw.s3.current(lw.t, "data-uk", "instances/") { |
| + loc := activator.HibernatePrefix(sealPrefix, offHost) |
| + for k := range lw.s3.current(lw.t, sealUK, "instances/") { |
| if k != loc+"/lock.json" { |
| lw.t.Fatalf("data-uk holds %s before the first wake", k) |
| } |
| } |
| - _, owner, expires, err := lease.LockState(context.Background(), "data-uk", loc, "eu-west-2") |
| + _, owner, expires, err := lease.LockState(context.Background(), sealUK, loc, sealUKRegion) |
| if err != nil || !(owner == "" || lease.IsLiftedTombstone(owner, expires)) { |
| lw.t.Fatalf("the target's lock is %q until %v (%v)", owner, expires, err) |
| } |
| @@ -629,17 +355,22 @@ func tree(t *testing.T, dir string) map[string]string { |
| return out |
| } |
| -// landed is what the target landed, by name: the files directory, the env |
| -// file and the config file, as the source has them. |
| +// requireSourceBytes: the target's files directory, env file and config |
| +// file are the source's byte for byte. A source with no files directory |
| +// lands an empty one. |
| func (lw *landWorld) requireSourceBytes() { |
| t := lw.t |
| t.Helper() |
| - src, dst := tree(t, filepath.Join(lw.ie.dd, offHost)), tree(t, filepath.Join(lw.uk.dd, offHost)) |
| - if !reflect.DeepEqual(src, dst) { |
| + dst := tree(t, filepath.Join(lw.uk.dd, offHost)) |
| + if lw.opts.noFiles { |
| + if len(dst) != 1 || dst["./"] == "" { |
| + t.Fatalf("a source with no files directory landed %v", dst) |
| + } |
| + } else if src := tree(t, lw.srcFiles); !reflect.DeepEqual(src, dst) { |
| t.Fatalf("files differ:\nsource %v\ntarget %v", src, dst) |
| } |
| - for _, name := range []string{offHost + ".env", offHost + ".config.env"} { |
| - a, err := os.ReadFile(filepath.Join(lw.ie.dd, name)) |
| + for src, name := range map[string]string{lw.srcEnv: offHost + ".env", lw.srcConfig: offHost + ".config.env"} { |
| + a, err := os.ReadFile(src) |
| if err != nil { |
| t.Fatal(err) |
| } |
| @@ -663,6 +394,19 @@ func (lw *landWorld) parked(role string) map[string]string { |
| return tree(lw.t, dir) |
| } |
| +// crashAt makes the target's landing crash once at point. |
| +func (lw *landWorld) crashAt(point string) *int { |
| + crashed := 0 |
| + lw.uk.w.cfg.landFault = func(p string) error { |
| + if p == point && crashed == 0 { |
| + crashed++ |
| + return errors.New("crash") |
| + } |
| + return nil |
| + } |
| + return &crashed |
| +} |
| + |
| // --- (a) and (f): the landing -------------------------------------------------- |
| // The full landing reaches offline-landed with every landing item OK, the |
| @@ -700,14 +444,14 @@ func TestOfflineLandHappyPath(t *testing.T) { |
| if err != nil { |
| t.Fatal(err) |
| } |
| - for k, want := range map[string]string{"STORE_BUCKET": landStore, "STORE_ENDPOINT": "https://store.example.com", "STORE_SECRET_KEY": landSecret, "CARLOS_INSTANCE_KEY": landKeyValue, "CARLOS_ADMIN_TOKEN": "admin-token", "APP_OVERRIDE": "1"} { |
| + for k, want := range map[string]string{"STORE_BUCKET": landStore, "STORE_ENDPOINT": "https://store.example.com", "STORE_SECRET_KEY": landSecret, "CARLOS_INSTANCE_KEY": landKeyValue, "CARLOS_ADMIN_TOKEN": "token"} { |
| if env[k] != want { |
| t.Fatalf("the target launches with %s=%q, want %q", k, env[k], want) |
| } |
| } |
| // The ack names the move; the staged replica is untouched. |
| ack := appconfig.ConfigAck{} |
| - body, _, err := lw.wd.s.Get(context.Background(), appconfig.ConfigAckKey("uk-a", offHost)) |
| + body, _, err := lw.wd.s.Get(context.Background(), appconfig.ConfigAckKey("uk-box", offHost)) |
| if err != nil || json.Unmarshal(body, &ack) != nil || ack.MoveOp != offID { |
| t.Fatalf("config ack %+v %v", ack, err) |
| } |
| @@ -756,10 +500,6 @@ func TestOfflineLandIsAnEnsure(t *testing.T) { |
| t.Fatalf("pass %d changed data-uk moves/", i) |
| } |
| } |
| - // The replica was restored once: the remembered fingerprint stands. |
| - if c := lw.report().Items["db"]; !strings.HasPrefix(c.Reason, dbVerifiedPrefix) { |
| - t.Fatalf("db %+v", c) |
| - } |
| } |
| // A crash at every point of the landing (after each item, after the report, |
| @@ -811,15 +551,15 @@ func TestOfflineLandMendsWhatIsMissing(t *testing.T) { |
| }{ |
| {"files directory removed", func(lw *landWorld) { os.RemoveAll(filepath.Join(lw.uk.dd, offHost)) }, nil}, |
| {"a file changed", func(lw *landWorld) { |
| - writeFile(lw.t, filepath.Join(lw.uk.dd, offHost, "notes.txt"), "tampered\n") |
| - }, []string{offHost + "/notes.txt"}}, |
| + writeFile(lw.t, filepath.Join(lw.uk.dd, offHost, "instance.key"), "tampered\n") |
| + }, []string{offHost + "/instance.key"}}, |
| {"a file added", func(lw *landWorld) { |
| writeFile(lw.t, filepath.Join(lw.uk.dd, offHost, "stray"), "x") |
| }, []string{offHost + "/stray"}}, |
| - {"env removed", func(lw *landWorld) { os.Remove(filepath.Join(lw.uk.dd, offHost+".env")) }, []string{offHost + "/notes.txt"}}, |
| + {"env removed", func(lw *landWorld) { os.Remove(filepath.Join(lw.uk.dd, offHost+".env")) }, []string{offHost + "/instance.key"}}, |
| {"env changed: a new instance key", func(lw *landWorld) { |
| writeFile(lw.t, filepath.Join(lw.uk.dd, offHost+".env"), "CARLOS_INSTANCE_KEY=minted-on-the-target\n") |
| - }, []string{offHost + ".env", offHost + "/notes.txt"}}, |
| + }, []string{offHost + ".env", offHost + "/instance.key"}}, |
| {"config removed", func(lw *landWorld) { os.Remove(filepath.Join(lw.uk.dd, offHost+".config.env")) }, nil}, |
| {"config changed", func(lw *landWorld) { |
| writeFile(lw.t, filepath.Join(lw.uk.dd, offHost+".config.env"), "STORE_BUCKET=elsewhere\n") |
| @@ -1019,7 +759,7 @@ func TestOfflineLandFailsEachMismatch(t *testing.T) { |
| if r := lw.report(); r.Phase != moves.OfflinePhaseFailed || r.Items["release"].OK { |
| t.Fatalf("%+v", r.Items["release"]) |
| } |
| - lw.promote("v1") |
| + lw.promote(sealVersion) |
| lw.uk.tick() |
| lw.requireLanded() |
| }) |
| @@ -1053,7 +793,7 @@ func TestOfflineLandOpenReleasesOnlyWhenEveryItemVerifies(t *testing.T) { |
| lw.requireNoTargetLease() |
| // An item the open's ensure mends (a timer lost) does not block it. |
| - lw.promote("v1") |
| + lw.promote(sealVersion) |
| if err := lw.uk.reg.ReplaceTimers(offHost, nil); err != nil { |
| t.Fatal(err) |
| } |
| @@ -1096,24 +836,8 @@ func TestOfflineLandOpenReleasesOnlyWhenEveryItemVerifies(t *testing.T) { |
| if err := lease.CheckDB(rt.DBPath); err != nil { |
| t.Fatal(err) |
| } |
| - db, err := sql.Open("sqlite", rt.DBPath) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - var bodies []string |
| - rows, err := db.Query(`SELECT body FROM notes ORDER BY id`) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - for rows.Next() { |
| - var b string |
| - rows.Scan(&b) |
| - bodies = append(bodies, b) |
| - } |
| - rows.Close() |
| - db.Close() |
| - if fmt.Sprint(bodies) != "[one two three]" { |
| - t.Fatalf("restored rows %v", bodies) |
| + if n := countRows(t, rt.DBPath); n != 20 { |
| + t.Fatalf("restored %d rows, want 20", n) |
| } |
| if got := lw.s3.current(t, "data-uk", activator.MovesPrefix(offHost)); len(got) != 0 { |
| t.Fatalf("staging left after the promotion: %v", keysOf(got)) |
| @@ -1205,7 +929,7 @@ func TestOfflineLandTargetRevertBeforeTheLanding(t *testing.T) { |
| if rt := lw.uk.row(offHost); rt == nil || rt.HoldOp != offID { |
| t.Fatalf("row %+v", rt) |
| } |
| - if _, ok := lw.wd.reportIf(moves.OfflineRoleTarget, "uk-a"); ok { |
| + if _, ok := lw.wd.reportIf(moves.OfflineRoleTarget, "uk-box"); ok { |
| t.Fatal("reported before the crash point") |
| } |
| lw.wd.abort() |
| @@ -1228,7 +952,7 @@ func TestOfflineLandTargetRevertBeforeTheLanding(t *testing.T) { |
| if rt := lw2.uk.row(offHost); rt != nil { |
| t.Fatalf("row %+v", rt) |
| } |
| - if _, ok := lw2.wd.reportIf(moves.OfflineRoleTarget, "uk-a"); ok { |
| + if _, ok := lw2.wd.reportIf(moves.OfflineRoleTarget, "uk-box"); ok { |
| t.Fatal("a target that never had a row reported") |
| } |
| } |
| @@ -1240,7 +964,7 @@ func TestOfflineLandTargetRevertBeforeTheLanding(t *testing.T) { |
| // env item by name (and the env pin), so the target never hands the app a key |
| // that is not the source's. |
| func TestOfflineLandConfiguredInstanceKey(t *testing.T) { |
| - lw := newLandWorldWith(t, landKeyValue) |
| + lw := newLandWorldWith(t, landOpts{configKey: landKeyValue}) |
| lw.flipAndLand(1) |
| lw.requireLanded() |
| lw.requireSourceBytes() |
| @@ -1249,8 +973,8 @@ func TestOfflineLandConfiguredInstanceKey(t *testing.T) { |
| t.Fatalf("the target hands over %q (%v)", env[instancekey.Var], err) |
| } |
| - lw2 := newLandWorldWith(t, landKeyValue) |
| - lw2.configKey = "ik_ffffffffffffffffffffffffffffffff" |
| + lw2 := newLandWorldWith(t, landOpts{configKey: landKeyValue}) |
| + lw2.opts.configKey = "ik_ffffffffffffffffffffffffffffffff" |
| lw2.bundle(landStore) |
| lw2.flipAndLand(2) |
| if st := lw2.state(); st != provision.OfflineLanding { |
| @@ -1264,3 +988,148 @@ func TestOfflineLandConfiguredInstanceKey(t *testing.T) { |
| t.Fatal("the env pin passed with another key") |
| } |
| } |
| + |
| +// Review I1: a first landing pass that extracted the files and env and |
| +// rendered the config, then crashed before it could report, followed by an |
| +// abort: the revert parks all three while the row is still held, before it |
| +// deletes the row, so nothing of the move's is left in the data directory. |
| +func TestOfflineLandRevertParksAfterACrashBeforeTheReport(t *testing.T) { |
| + lw := newLandWorld(t) |
| + lw.crashAt("offline-release") |
| + lw.flipAndLand(1) |
| + if _, ok := lw.wd.reportIf(moves.OfflineRoleTarget, "uk-box"); ok { |
| + t.Fatal("reported before the crash point") |
| + } |
| + names := []string{offHost, offHost + ".env", offHost + ".config.env"} |
| + for _, name := range names { |
| + if !lexists(filepath.Join(lw.uk.dd, name)) { |
| + t.Fatalf("fixture: %s was not landed before the crash", name) |
| + } |
| + } |
| + landed := tree(t, filepath.Join(lw.uk.dd, offHost)) |
| + lw.wd.abort() |
| + lw.restart() |
| + lw.uk.tick() |
| + parkDir := moves.OfflineParkDir(lw.uk.dd, offHost, offID, moves.OfflineRoleTarget) |
| + for _, name := range names { |
| + if lexists(filepath.Join(lw.uk.dd, name)) { |
| + t.Fatalf("%s left in the data directory", name) |
| + } |
| + if !lexists(filepath.Join(parkDir, name)) { |
| + t.Fatalf("%s not parked", name) |
| + } |
| + } |
| + if got := tree(t, filepath.Join(parkDir, offHost)); !reflect.DeepEqual(got, landed) { |
| + t.Fatalf("parked %v, landed %v", got, landed) |
| + } |
| + if rt := lw.uk.row(offHost); rt != nil { |
| + t.Fatalf("row %+v", rt) |
| + } |
| + if r := lw.report(); r.Phase != moves.OfflinePhaseReverted { |
| + t.Fatalf("report %+v", r) |
| + } |
| +} |
| + |
| +// Review I2: a failed env pin names the variables that differ, from the |
| +// source's env report beside the archive, and never their values. |
| +func TestOfflineLandEnvPinNamesTheVariables(t *testing.T) { |
| + lw := newLandWorld(t) |
| + lw.bundle("shop-store-somewhere-else") |
| + lw.flipAndLand(1) |
| + c := lw.report().Items["env-pin"] |
| + if c.OK || !strings.Contains(c.Reason, "STORE_BUCKET") || strings.Contains(c.Reason, "somewhere-else") || strings.Contains(c.Reason, landStore) { |
| + t.Fatalf("env-pin %+v", c) |
| + } |
| + for _, name := range []string{"STORE_ENDPOINT", "STORE_SECRET_KEY", "FEATURE", "CARLOS_REGION"} { |
| + if strings.Contains(c.Reason, name) { |
| + t.Fatalf("env-pin names %s, which did not change: %q", name, c.Reason) |
| + } |
| + } |
| + // Without the source's report the pin still fails, by digest alone. |
| + lw2 := newLandWorld(t) |
| + lw2.bundle("shop-store-somewhere-else") |
| + lw2.s3.f.BareDelete(sealUK, moves.OfflineArchivePrefix(offHost, offID)+offlineEnvReportObject) |
| + lw2.flipAndLand(1) |
| + if c := lw2.report().Items["env-pin"]; c.OK || c.Reason != "the launch environment differs from the source's" { |
| + t.Fatalf("env-pin without the source's report %+v", c) |
| + } |
| +} |
| + |
| +// Review M1: the verified report is written before the release, so a crash |
| +// after the release leaves the items that pass verified (here after an |
| +// earlier open pass failed), and the next pass reports opened with every |
| +// item OK. |
| +func TestOfflineLandOpenAfterACrashPastTheRelease(t *testing.T) { |
| + lw := newLandWorld(t) |
| + lw.flipAndLand(1) |
| + lw.requireLanded() |
| + lw.wd.open() |
| + lw.promote("v2") |
| + lw.uk.tick() |
| + if r := lw.report(); r.Items["release"].OK { |
| + t.Fatal("fixture: the release item passed") |
| + } |
| + lw.promote(sealVersion) |
| + crashed := lw.crashAt("offline-released") |
| + lw.uk.tick() |
| + if *crashed != 1 { |
| + t.Fatal("the crash point was never reached") |
| + } |
| + if rt := lw.uk.row(offHost); rt == nil || rt.Status != "ready" { |
| + t.Fatalf("row %+v", rt) |
| + } |
| + if r := lw.report(); r.Phase == moves.OfflinePhaseOpened { |
| + t.Fatal("fixture: reported opened before the crash") |
| + } else if ok, bad := r.ItemsOK(moves.OfflineOwnerLanding); !ok { |
| + t.Fatalf("the report before the release carries %v not OK", bad) |
| + } |
| + lw.restart() |
| + lw.uk.tick() |
| + r := lw.report() |
| + if ok, bad := r.ItemsOK(moves.OfflineOwnerLanding); r.Phase != moves.OfflinePhaseOpened || !ok { |
| + t.Fatalf("phase %s, not OK %v", r.Phase, bad) |
| + } |
| + if got := lw.s3.current(t, sealUK, moves.OfflineArchivePrefix(offHost, offID)); len(got) != 0 { |
| + t.Fatalf("the archive is still there: %v", keysOf(got)) |
| + } |
| +} |
| + |
| +// T4's contract for a host with no files directory: the archive carries no |
| +// files/ entry and requires no file. The landing accepts it: extraction |
| +// leaves an empty <DataDir>/<host> (0755), whose digest with the env file is |
| +// the seal's, and later passes leave it alone. |
| +func TestOfflineLandAHostWithNoFilesDirectory(t *testing.T) { |
| + lw := newLandWorldWith(t, landOpts{noFiles: true}) |
| + if lw.seal.FilesEntries != 1 { |
| + t.Fatalf("fixture: the seal carries %d entries, want the env alone", lw.seal.FilesEntries) |
| + } |
| + lw.flipAndLand(1) |
| + lw.requireLanded() |
| + lw.requireSourceBytes() |
| + st, err := os.Stat(filepath.Join(lw.uk.dd, offHost)) |
| + if err != nil || st.Mode().Perm() != 0o755 { |
| + t.Fatalf("the files directory: %v %v", st, err) |
| + } |
| + disk := tree(t, lw.uk.dd) |
| + lw.uk.tick() |
| + lw.requireLanded() |
| + if got := tree(t, lw.uk.dd); !reflect.DeepEqual(got, disk) { |
| + t.Fatal("a later pass changed the disk") |
| + } |
| +} |
| + |
| +// A nested source keeps launch descriptors (launch.<role>.json and |
| +// launch.gen.<n>.json) beside its files directory. They are the platform's, |
| +// not carried, and the landing neither expects nor makes them: it lands the |
| +// files directory alone, flat. |
| +func TestOfflineLandANestedSourceWithLaunchFiles(t *testing.T) { |
| + lw := newLandWorldWith(t, landOpts{nested: true}) |
| + lw.flipAndLand(1) |
| + lw.requireLanded() |
| + lw.requireSourceBytes() |
| + for rel := range tree(t, lw.uk.dd) { |
| + if strings.Contains(filepath.Base(rel), "launch.") { |
| + t.Fatalf("the target holds %s", rel) |
| + } |
| + } |
| +} |