amadan

carlos / platform Public

moves: the move log and member-stream events (§13, Plan 6 Task 3)

- internal/moves/log.go: LogEvent, LogKeyFor (replacing the counter-keyed
  LogKey), WriteLog (create-only; the first writer of a transition creates
  it, ErrCASConflict is success with nothing created, L27) and ReadLog
  (ordered by At).
- internal/host/movesevents.go: every box write of a move is wrapped in
  observedMove, which keeps the record the write found; logTransition names
  the transition from it and the record written: <state>-<phase>,
  probe_attempt-<n>, stuck-<epoch> and retry-<epoch> on claim Plan 10's
  RecoveryRetryEpoch (L11), and the terminal outcome. observeMove logs the
  console doors' request, manual cut, abort and retry, and a declared loss,
  the first time a box's pass sees them; logRevPhase a revert's claim
  steps. A worker hands each identity to LogEvent once.
- moveCfg.LogEvent is widened to (ctx, store, record, moves.LogEvent)
  (L10); production wires moveLogWriter with the agent's rollout sink as
  moveCfg.Events: the object create-only, and the member event only for
  the writer that created it (L9): M-C1, M-C5, M-C6, M-C7, M-C8, M-C9, and
  MV-N380 (pending) for stuck, with no operator command.
- Old event names map once (rollback -> rolling_back, target-released ->
  released, source-stuck/target-stuck -> stuck, source-opened ->
  rev_opened, ...); the rollback, recovery and revert harnesses follow, and
  the rollback harness retries through MoveRetryRecovery.
- B8 census: 17 writers' digests (each gains observedMove and its
  logTransition) and moves.WriteLog as a nonrecord Put (L21).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude ef267fb9097d6219c49d10b9d6f60e01a3f9c594 parent f2ebf4d
28 files changed, +967 −96
  • approved_moves_copy_test.go +5 −0
  • docs/superpowers/plans/2026-10-06-moves-phase1-copy.md +2 −1
  • internal/console/apiinstancemove_test.go +1 −1
  • internal/host/agent.go +1 −1
  • internal/host/movesabort.go +6 −1
  • internal/host/movescut.go +4 −1
  • internal/host/movesevents.go +342 −0
  • internal/host/movesevents_test.go +268 −0
  • internal/host/movesfinal.go +5 −1
  • internal/host/movesinstall.go +8 −2
  • internal/host/movesland.go +4 −1
  • internal/host/moveslandcommit.go +9 −2
  • internal/host/movesparks.go +4 −2
  • internal/host/movespass.go +14 −5
  • internal/host/movespass_test.go +2 −2
  • internal/host/movesphase1.go +11 −2
  • internal/host/movesphase1_test.go +1 −1
  • internal/host/movesrecover.go +16 −11
  • internal/host/movesrecover_test.go +5 −3
  • internal/host/movesrevert.go +7 −5
  • internal/host/movesrevert_test.go +3 −1
  • internal/host/movesrollback.go +14 −24
  • internal/host/movesrollback_test.go +13 −7
  • internal/moves/log.go +99 −0
  • internal/moves/log_test.go +97 −0
  • internal/moves/staging.go +5 −4
  • internal/moves/staging_test.go +1 −1
  • internal/provision/plan9_b8_census_test.go +20 −17
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index 960f9a71b..8b8716aa6 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -80,6 +80,7 @@ var movesCopyIDs = []string{
"MV-N357", "MV-N358", "MV-N359", "MV-N360", "MV-N361", "MV-N362", "MV-N363", "MV-N364",
"MV-N365", "MV-N366", "MV-N367", "MV-N368", "MV-N369", "MV-N370", "MV-N371",
"MV-N372", "MV-N373", "MV-N374", "MV-N375", "MV-N376", "MV-N377", "MV-N378", "MV-N379",
+ "MV-N380",
}
// copyRow is one row of the record. The File cell is the file the sentence is
@@ -587,6 +588,7 @@ var movesScanFiles = []string{
"internal/host/movesfloorgate.go",
"internal/host/movesfloorgate_linux.go",
"internal/host/movesfloorgate_other.go",
+ "internal/host/movesevents.go",
"internal/host/movesabort.go",
"internal/host/movesarchive.go",
"internal/host/movescut.go",
@@ -759,6 +761,9 @@ var movesReturnedSentences = map[string]map[string]bool{
"internal/host/movesreport.go": {"t1Below": true, "t3Differs": true, "t3Shadowed": true, "t4Failed": true,
"t6NoRoom": true, "t8Count": true, "t12NotQuiet": true, "replicaReason": true},
"internal/host/movesrollback.go": {"stuckPage": true},
+ // memberLine is what an app's members read for a move event (Plan 6
+ // Task 3, preflight L9).
+ "internal/host/movesevents.go": {"memberLine": true},
// moveDeleteRefusal is M-C19, which the delete door answers (Task 2b).
"internal/console/apiinstancemove.go": {"moveDeleteRefusal": true},
}
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 0f32c72ba..cd800c89c 100644
--- a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
+++ b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
@@ -284,7 +284,7 @@ Columns: id, file, text, status.
## Plan 6
-(completed by Plan 6 Task 7. Task 2 takes MV-N319 to MV-N324, and made the abort door speak M-C13 on a console without the host claim, so M-C13 is now approved verbatim. Task 2b made the delete door speak M-C19 through moveDeleteRefusal in internal/console/apiinstancemove.go, so its File cell moves there (preflight L6) and it is approved verbatim. Task 1a takes MV-N350 to MV-N371 and Task 1b MV-N372 to MV-N379 (Plan 5 owns MV-N325 to MV-N349) for `carlos move` (move.go, and its usage line in main.go), and speaks M-C1 to M-C12 and M-C22 verbatim from move.go, so those rows are approved. Task 1b speaks M-C18 verbatim from move.go and MV-N11 from the resolve door of a console without the host claim, so both are approved)
+(completed by Plan 6 Task 7. Task 2 takes MV-N319 to MV-N324, and made the abort door speak M-C13 on a console without the host claim, so M-C13 is now approved verbatim. Task 2b made the delete door speak M-C19 through moveDeleteRefusal in internal/console/apiinstancemove.go, so its File cell moves there (preflight L6) and it is approved verbatim. Task 1a takes MV-N350 to MV-N371 and Task 1b MV-N372 to MV-N379, Task 3 MV-N380 (the member line for stuck, preflight L9) (Plan 5 owns MV-N325 to MV-N349) for `carlos move` (move.go, and its usage line in main.go), and speaks M-C1 to M-C12 and M-C22 verbatim from move.go, so those rows are approved. Task 1b speaks M-C18 verbatim from move.go and MV-N11 from the resolve door of a console without the host claim, so both are approved)
| id | file | text | status |
|---|---|---|---|
@@ -319,6 +319,7 @@ Columns: id, file, text, status.
| MV-N377 | move.go | The console asks to confirm a loss carlos move does not know. Use carlos ops host resolve. | pending |
| MV-N378 | move.go | carlos move resolve takes --retry or --declare-lost, not both. | pending |
| MV-N379 | move.go | The move changed since it was read. Nothing was declared. Run the command again. | pending |
+| MV-N380 | internal/host/movesevents.go | <host> could not start in <to> or in <from>. It is offline. An operator has been paged. | pending |
## Review of 2026-10-09
diff --git a/internal/console/apiinstancemove_test.go b/internal/console/apiinstancemove_test.go
index 968918638..dec908463 100644
--- a/internal/console/apiinstancemove_test.go
+++ b/internal/console/apiinstancemove_test.go
@@ -1898,7 +1898,7 @@ func TestMoveDetailLeaksNothing(t *testing.T) {
putJSON(moves.TargetReportKey(prefix), dst)
putJSON(srcKey, map[string]any{"v": 1, "role": "source", "note": "evidence-body-canary-source", "env": env["DATABASE_URL"]})
putJSON(dstKey, map[string]any{"v": 1, "role": "target", "note": "evidence-body-canary-target"})
- logKeys := []string{moves.LogKey(prefix, 1, "requested"), moves.LogKey(prefix, 2, "rollback")}
+ logKeys := []string{moves.LogKeyFor(prefix, "requested"), moves.LogKeyFor(prefix, "rolling_back")}
for _, k := range logKeys {
putJSON(k, map[string]any{"event": "x", "note": "log-body-canary", "actor": "alice@keymail.dev"})
}
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 7457ed444..7fce00b5b 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -447,7 +447,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {
} else {
eb.Mirror = m
}
- armMovesPass(*cfg, dd, reg, svc, eb, claims.Writes)
+ armMovesPass(*cfg, dd, reg, svc, eb, claims.Writes, rolloutSink)
// The startup sweep (design §6.7): a leftover instance unit from a
// previous agent process, one this box's registry no longer holds
// a live wake for, is stopped before this process starts spawning
diff --git a/internal/host/movesabort.go b/internal/host/movesabort.go
index 7f1ecd3e1..31fed78b2 100644
--- a/internal/host/movesabort.go
+++ b/internal/host/movesabort.go
@@ -143,7 +143,12 @@ func (w *movesWorker) abortSource(ctx context.Context, s release.Store, rec prov
return nil
}
g := provision.MoveGuard{ID: id, Incarnation: rec.Move.Incarnation}
- if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveGuardedAbortComplete(g, cfg.Now())); err != nil {
+ var pre provision.Record
+ after, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, observedMove(&pre, provision.MoveGuardedAbortComplete(g, cfg.Now())))
+ if err == nil {
+ w.logTransition(ctx, s, pre, *after)
+ }
+ if err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: abort of %s: the record moved on (%v)", rec.Host, id, err)
return nil
diff --git a/internal/host/movescut.go b/internal/host/movescut.go
index 6605bb021..daeceb755 100644
--- a/internal/host/movescut.go
+++ b/internal/host/movescut.go
@@ -158,13 +158,16 @@ func (c *cutRun) step(ctx context.Context) (bool, error) {
// someone else's decision (an abort, a console write): logged, and the
// pass ends without error.
func (c *cutRun) cutTransition(ctx context.Context, what string, mut func(*provision.Record) error) (bool, error) {
- if _, err := provision.SetRecordMove(ctx, c.s, c.rec.Account, c.rec.App, c.rec.Host, mut); err != nil {
+ var pre provision.Record
+ after, err := provision.SetRecordMove(ctx, c.s, c.rec.Account, c.rec.App, c.rec.Host, observedMove(&pre, mut))
+ if err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: %s of %s refused (%v); the record moved on", c.rec.Host, what, c.g.ID, err)
return false, nil
}
return false, fmt.Errorf("%s: %w", what, err)
}
+ c.w.logTransition(ctx, c.s, pre, *after)
return true, nil
}
diff --git a/internal/host/movesevents.go b/internal/host/movesevents.go
new file mode 100644
index 000000000..0ef9d2166
--- /dev/null
+++ b/internal/host/movesevents.go
@@ -0,0 +1,342 @@
+package host
+
+import (
+ "context"
+ "fmt"
+ "log"
+ "math"
+ "strconv"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The move log and the member stream (spec §13, Plan 6 Task 3).
+//
+// Every record write a box makes for a move goes through observedMove, which
+// keeps the record as the write found it; logTransition names what the write
+// did, from that record and the one written: one event, one identity
+// (moves.LogEvent). Transitions another writer made (the console's request,
+// manual cut, abort and retry; a declared loss) are seen by the box's own
+// pass (observeMove) and named the same way. The identity is the log object's
+// key, written create-only behind moveCfg.LogEvent (moveLogWriter in
+// production), and only the writer that created the object tells the app's
+// members (preflight L9, L27). A worker also remembers what it logged, so a
+// pass that sees the same transition again does not even try.
+//
+// Recovery probe attempts are not logged: their counter is the recovery's own
+// attempt journal, not the record (Plan 6 Task 3 report).
+
+// moveLogEntry is what one transition is in the log.
+type moveLogEntry struct {
+ op, event, identity string
+ fields map[string]string
+}
+
+// movePhaseOf is the phase a record's state is at: the cut's, the landing's,
+// or (rolling back) the target's and the source's.
+func movePhaseOf(m *provision.MoveState) (phase, target, source string) {
+ switch m.State {
+ case provision.MoveCutting:
+ return m.CutPhase, "", ""
+ case provision.MoveLanding:
+ return m.LandingPhase, "", ""
+ case provision.MoveRollingBack:
+ if m.Rollback != nil {
+ return "", m.Rollback.TargetPhase, m.Rollback.SourcePhase
+ }
+ }
+ return "", "", ""
+}
+
+func moveRegions(m *provision.MoveState) map[string]string {
+ return map[string]string{"from": m.From, "to": m.To}
+}
+
+// moveTransition names the transition from before to after, one record write:
+// before as the write found it, after as written. ok is false when the write
+// changed nothing the log records (a pin, recovery inputs, a replay).
+func moveTransition(before, after provision.Record) (moveLogEntry, bool) {
+ bm, am := before.Move, after.Move
+ if am == nil {
+ // A close clears the move and leaves its outcome (D11).
+ if bm == nil || after.LastMove == nil || after.LastMove.ID != bm.ID || bm.ID == "" {
+ return moveLogEntry{}, false
+ }
+ f := moveRegions(bm)
+ f["offline_s"] = strconv.FormatFloat(after.LastMove.OfflineS, 'f', -1, 64)
+ return moveLogEntry{op: bm.ID, event: after.LastMove.Outcome, identity: after.LastMove.Outcome, fields: f}, true
+ }
+ if am.ID == "" || (bm != nil && bm.ID != am.ID) {
+ return moveLogEntry{}, false
+ }
+ e := moveLogEntry{op: am.ID, fields: moveRegions(am)}
+ if bm == nil {
+ bm = &provision.MoveState{}
+ }
+ switch {
+ case am.State == provision.MoveStuck:
+ if bm.State == provision.MoveStuck {
+ return moveLogEntry{}, false
+ }
+ e.event, e.identity = "stuck", "stuck-"+strconv.FormatUint(am.RecoveryRetryEpoch, 10)
+ return e, true
+ case am.State == provision.MoveLanding && am.Landing != nil && (bm.Landing == nil || am.Landing.Attempts > bm.Landing.Attempts) && am.Landing.Attempts > 0:
+ e.event, e.identity = "probe_attempt", "probe_attempt-"+strconv.Itoa(am.Landing.Attempts)
+ return e, true
+ case bm.State == provision.MoveCutting && am.State == provision.MoveLanding:
+ // The flip is the record's one write from cutting to landing.
+ e.event, e.identity = "flipped", "cutting-flipped"
+ return e, true
+ case am.State == provision.MoveCommitted && bm.State == provision.MoveCommitted:
+ switch {
+ case am.SourceParkedAt != nil && bm.SourceParkedAt == nil:
+ e.event, e.identity = "source_parked", "committed-source_parked"
+ case am.OpenedAt != nil && bm.OpenedAt == nil:
+ e.event, e.identity = "opened", "committed-opened"
+ if am.CutAt != nil {
+ e.fields["offline_s"] = strconv.FormatInt(int64(am.OpenedAt.Sub(*am.CutAt).Seconds()), 10)
+ }
+ default:
+ return moveLogEntry{}, false
+ }
+ return e, true
+ case am.State == provision.MoveReverting && bm.State == provision.MoveReverting:
+ if am.ReopenedAt == nil || bm.ReopenedAt != nil {
+ return moveLogEntry{}, false
+ }
+ e.event, e.identity = "rev_opened", "reverting-opened"
+ return e, true
+ }
+ phase, target, source := movePhaseOf(am)
+ bphase, btarget, bsource := movePhaseOf(bm)
+ if am.State == bm.State {
+ switch {
+ case am.State == provision.MoveRollingBack && target != btarget:
+ e.event, e.identity = target, "rolling_back-target-"+target
+ if target == provision.RbTargetLost {
+ e.event = "declared_lost"
+ }
+ case am.State == provision.MoveRollingBack && source != bsource:
+ e.event, e.identity = source, "rolling_back-source-"+source
+ if source == provision.RbSourceOpened {
+ e.fields["offline_s"] = moveOfflineSince(am)
+ }
+ case phase != bphase && phase != "":
+ e.event, e.identity = phase, am.State+"-"+phase
+ if am.State == provision.MoveCutting && phase == provision.CutRequested {
+ e.event = "cut_requested"
+ }
+ default:
+ return moveLogEntry{}, false
+ }
+ return e, true
+ }
+ // A new state.
+ switch am.State {
+ case provision.MovePreparing:
+ e.event, e.identity = "requested", "requested"
+ case provision.MoveCutting:
+ e.event, e.identity = phase, "cutting-"+phase
+ if phase == provision.CutRequested || phase == "" {
+ e.event, e.identity = "cut_requested", "cutting-requested"
+ }
+ case provision.MoveLanding:
+ e.event, e.identity = phase, "landing-"+phase
+ case provision.MoveRollingBack:
+ e.event, e.identity = "rolling_back", "rolling_back"
+ default:
+ // ready, committed, aborting, reverting, relocating: the state is the
+ // event.
+ e.event, e.identity = am.State, am.State
+ }
+ return e, true
+}
+
+// moveOfflineSince is the seconds from the cut to the source's reopen, for a
+// rollback's open, as the close counts it.
+func moveOfflineSince(m *provision.MoveState) string {
+ if m.CutAt == nil || m.Rollback == nil || m.Rollback.OpenedAt == nil {
+ return ""
+ }
+ return strconv.FormatInt(int64(m.Rollback.OpenedAt.Sub(*m.CutAt).Seconds()), 10)
+}
+
+// copyMove is a deep enough copy of a move for moveTransition: the fields it
+// compares, safe from a mutation that edits the journal in place.
+func copyMove(m *provision.MoveState) *provision.MoveState {
+ if m == nil {
+ return nil
+ }
+ c := *m
+ if m.Rollback != nil {
+ rb := *m.Rollback
+ c.Rollback = &rb
+ }
+ if m.Landing != nil {
+ l := *m.Landing
+ c.Landing = &l
+ }
+ return &c
+}
+
+// observedMove wraps a record mutation so *pre keeps the record as the write
+// found it: the transition's before, for the log.
+func observedMove(pre *provision.Record, mut func(*provision.Record) error) func(*provision.Record) error {
+ return func(rec *provision.Record) error {
+ *pre = *rec
+ pre.Move = copyMove(rec.Move)
+ return mut(rec)
+ }
+}
+
+// logTransition logs the transition one write made (moveTransition).
+func (w *movesWorker) logTransition(ctx context.Context, s release.Store, before, after provision.Record) {
+ if e, ok := moveTransition(before, after); ok {
+ w.record(ctx, s, after, e)
+ }
+}
+
+// observeMove logs the transitions another writer made that this box's pass
+// sees: the console's request, manual cut, abort and retry, and a declared
+// loss. Each is named as if it were this box's write from a record with no
+// move (or, for a retry, by its epoch), so it has the same identity whoever
+// logs it.
+func (w *movesWorker) observeMove(ctx context.Context, s release.Store, rec provision.Record) {
+ m := rec.Move
+ if m == nil || m.ID == "" {
+ return
+ }
+ switch {
+ case m.State == provision.MovePreparing, m.State == provision.MoveAborting,
+ m.State == provision.MoveCutting && m.CutPhase == provision.CutRequested:
+ w.logTransition(ctx, s, provision.Record{}, rec)
+ case m.State == provision.MoveRollingBack && m.Rollback != nil && m.Rollback.TargetPhase == provision.RbTargetLost:
+ before := provision.Record{Move: copyMove(m)}
+ before.Move.Rollback.TargetPhase = ""
+ w.logTransition(ctx, s, before, rec)
+ }
+ if m.RecoveryRetryEpoch > 0 && m.State != provision.MoveStuck {
+ w.record(ctx, s, rec, moveLogEntry{op: m.ID, event: "retry", identity: "retry-" + strconv.FormatUint(m.RecoveryRetryEpoch, 10), fields: moveRegions(m)})
+ }
+}
+
+// logRevPhase logs a revert's step, which is the claim's rev_phase, not the
+// record's: reverting-<phase>, event rev_<phase>.
+func (w *movesWorker) logRevPhase(ctx context.Context, s release.Store, rec provision.Record, before, after hostclaim.Observed) {
+ phase := func(o hostclaim.Observed) string {
+ if o.Claim == nil {
+ return ""
+ }
+ return o.Claim.Op.Journal[hostclaim.JournalRevPhase]
+ }
+ p := phase(after)
+ if p == "" || p == phase(before) || rec.Move == nil || rec.Move.ID == "" {
+ return
+ }
+ w.record(ctx, s, rec, moveLogEntry{op: rec.Move.ID, event: "rev_" + p, identity: "reverting-" + p, fields: moveRegions(rec.Move)})
+}
+
+// record hands one entry to the LogEvent hook, once per worker: a pass that
+// sees the same transition again does not write it again. With no hook (a
+// test's worker) the event is a log line.
+func (w *movesWorker) record(ctx context.Context, s release.Store, rec provision.Record, e moveLogEntry) {
+ key := rec.Host + "/" + e.op + "/" + e.identity
+ if _, seen := w.logged.LoadOrStore(key, true); seen {
+ return
+ }
+ ev := moves.LogEvent{V: 1, Op: e.op, Host: rec.Host, Event: e.event, Identity: e.identity, Actor: w.cfg.Box, Fields: e.fields}
+ if w.cfg.Now != nil {
+ ev.At = w.cfg.Now().UTC()
+ }
+ if f := w.cfg.LogEvent; f != nil {
+ f(ctx, s, rec, ev)
+ return
+ }
+ log.Printf("moves: %s: %s %s (%s)", rec.Host, e.op, e.event, e.identity)
+}
+
+// moveLogWriter is the production LogEvent: the object, create-only, under
+// the move's report prefix in the deployment bucket, and the member event
+// when this writer created it. A failed write is logged; the next pass that
+// sees the transition does not try again (the worker remembers it), so a lost
+// object is a gap in the log, never a second event.
+func moveLogWriter(sink activator.EventSink) func(context.Context, release.Store, provision.Record, moves.LogEvent) {
+ return func(ctx context.Context, s release.Store, rec provision.Record, ev moves.LogEvent) {
+ if s == nil {
+ return
+ }
+ created, err := moves.WriteLog(ctx, s, moves.ReportPrefix(rec.Account, rec.App, rec.Host, ev.Op), ev)
+ if err != nil {
+ log.Printf("moves: %s: the log of %s at %s: %v", rec.Host, ev.Op, ev.Identity, err)
+ return
+ }
+ if !created || sink == nil {
+ return
+ }
+ if line, ok := memberLine(ev, rec); ok {
+ sink.Event(rec.Account, rec.App, rec.Host, memberLevel(ev.Event), line)
+ }
+ }
+}
+
+// memberLine is what the app's members read for an event (preflight L9: M-C1,
+// M-C5, M-C6, M-C7, M-C8, M-C9, and for stuck a line with no operator command
+// in it, MV-N380). ok is false for an event members are not told of, and for
+// one whose regions the record names only as the home's "". Its level is
+// memberLevel's.
+func memberLine(ev moves.LogEvent, rec provision.Record) (line string, ok bool) {
+ host, from, to := rec.Host, ev.Fields["from"], ev.Fields["to"]
+ if from == "" || to == "" {
+ return "", false
+ }
+ switch ev.Event {
+ case "requested":
+ return fmt.Sprintf("Moving %s from %s to %s.", host, from, to), true
+ case "admitted":
+ if rec.Move == nil || rec.Move.CutBudgetS <= 0 {
+ return "", false
+ }
+ return fmt.Sprintf("%s is ready. Starting the cut. %s will be offline for about %s.", to, host, memberEstimate(rec.Move.CutBudgetS)), true
+ case "flipped":
+ return fmt.Sprintf("Stopped in %s. Starting in %s.", from, to), true
+ case "done":
+ offline, err := strconv.ParseFloat(ev.Fields["offline_s"], 64)
+ if err != nil {
+ return "", false
+ }
+ return fmt.Sprintf("Moved. %s is served from %s. It was offline for %d seconds.", host, to, int(math.Round(offline))), true
+ case "rolling_back":
+ if rec.Move == nil || rec.Move.Rollback == nil || rec.Move.Rollback.Reason == "" {
+ return "", false
+ }
+ return fmt.Sprintf("%s could not start %s: %s. Moving it back to %s.", to, host, rec.Move.Rollback.Reason, from), true
+ case "rolled_back":
+ return fmt.Sprintf("%s is back in %s and serving. Nothing was lost.", host, from), true
+ case "stuck":
+ return fmt.Sprintf("%s could not start in %s or in %s. It is offline. An operator has been paged.", host, to, from), true
+ }
+ return "", false
+}
+
+// memberLevel is the level of an event's member line.
+func memberLevel(event string) string {
+ switch event {
+ case "rolling_back", "rolled_back":
+ return "warn"
+ case "stuck":
+ return "error"
+ }
+ return "info"
+}
+
+// memberEstimate renders a cut budget for M-C5, as the CLI does.
+func memberEstimate(seconds int) string {
+ if seconds < 90 {
+ return fmt.Sprintf("%d seconds", seconds)
+ }
+ return fmt.Sprintf("%d minutes", (seconds+59)/60)
+}
diff --git a/internal/host/movesevents_test.go b/internal/host/movesevents_test.go
new file mode 100644
index 000000000..72540ec9f
--- /dev/null
+++ b/internal/host/movesevents_test.go
@@ -0,0 +1,268 @@
+package host
+
+import (
+ "context"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const evMoveID = "mv_abcdefghijklmnopqrstuvwxyz"
+
+// evRec is a record whose move is in the given state, edited by mut.
+func evRec(state string, mut func(*provision.MoveState)) provision.Record {
+ m := &provision.MoveState{ID: evMoveID, Incarnation: strings.Repeat("ab", 32), From: "ie", To: "uk", Cut: "auto", State: state, RequestedAt: time.Unix(1_800_000_000, 0).UTC()}
+ if mut != nil {
+ mut(m)
+ }
+ return provision.Record{Account: "bab", App: "titogo", Host: "acme.go.tito.io", Region: "ie", Move: m}
+}
+
+func evAt(sec int64) *time.Time {
+ t := time.Unix(1_800_000_000+sec, 0).UTC()
+ return &t
+}
+
+// The events a full move that ends done goes through, each record the one a
+// writer leaves, in order. The console's request is seen by the source's
+// pass (requested); every later record is a box writer's.
+func doneFlow() []provision.Record {
+ cut := func(phase string) provision.Record {
+ return evRec(provision.MoveCutting, func(m *provision.MoveState) {
+ m.CutPhase, m.CutAt, m.CutBudgetS = phase, evAt(10), 45
+ })
+ }
+ land := func(phase string, attempts int) provision.Record {
+ r := evRec(provision.MoveLanding, func(m *provision.MoveState) {
+ m.CutPhase, m.CutAt, m.CutBudgetS, m.FlippedAt, m.LandingPhase = provision.CutFlipped, evAt(10), 45, evAt(20), phase
+ if attempts > 0 {
+ m.Landing = &provision.MoveLandingJournal{Attempts: attempts}
+ }
+ })
+ r.Region = "uk"
+ return r
+ }
+ committed := func(mut func(*provision.MoveState)) provision.Record {
+ r := evRec(provision.MoveCommitted, func(m *provision.MoveState) {
+ m.CutPhase, m.CutAt, m.FlippedAt, m.CommittedAt = provision.CutFlipped, evAt(10), evAt(20), evAt(40)
+ m.Landing = &provision.MoveLandingJournal{Attempts: 1}
+ if mut != nil {
+ mut(m)
+ }
+ })
+ r.Region = "uk"
+ return r
+ }
+ closed := provision.Record{Account: "bab", App: "titogo", Host: "acme.go.tito.io", Region: "uk",
+ LastMove: &provision.MoveOutcome{ID: evMoveID, Outcome: "done", At: *evAt(60), OfflineS: 41.2}}
+ return []provision.Record{
+ evRec(provision.MovePreparing, nil),
+ evRec(provision.MoveReady, nil),
+ cut(provision.CutRequested),
+ cut(provision.CutAdmitted), cut(provision.CutStopping), cut(provision.CutStopped), cut(provision.CutStaged), cut(provision.CutArchived),
+ land(provision.LandChecking, 0),
+ land(provision.LandParking, 0), land(provision.LandBuilding, 0), land(provision.LandPublished, 0), land(provision.LandLocal, 0),
+ land(provision.LandConfigured, 0), land(provision.LandRestored, 0), land(provision.LandProbing, 0), land(provision.LandProbing, 1),
+ land(provision.LandProbed, 1),
+ committed(nil),
+ committed(func(m *provision.MoveState) { m.OpenedAt = evAt(51) }),
+ committed(func(m *provision.MoveState) { m.OpenedAt, m.SourceParkedAt = evAt(51), evAt(55) }),
+ closed,
+ }
+}
+
+var doneFlowEvents = []string{
+ "requested", "ready", "cut_requested", "admitted", "stopping", "stopped", "staged", "archived", "flipped",
+ "parking", "building", "published", "local", "configured", "restored", "probing", "probe_attempt", "probed",
+ "committed", "opened", "source_parked", "done",
+}
+
+// eventWorker is a worker whose LogEvent hook records every event it is
+// handed.
+func eventWorker(got *[]moves.LogEvent) *movesWorker {
+ w := &movesWorker{mirrors: map[string]lease.MirrorStore{}}
+ w.cfg.Box = "box-test"
+ w.cfg.Now = func() time.Time { return time.Unix(1_800_000_100, 0).UTC() }
+ w.cfg.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) {
+ *got = append(*got, ev)
+ }
+ return w
+}
+
+// Review Focus 3: across a full done flow, every transition gives one event,
+// in order, with its identity; the record seen again (a status poll, a pass
+// that changes nothing, a replay of the same write) gives none.
+func TestEveryTransitionEmitsOneEvent(t *testing.T) {
+ ctx := context.Background()
+ var got []moves.LogEvent
+ w := eventWorker(&got)
+ flow := doneFlow()
+ w.observeMove(ctx, nil, flow[0])
+ for i := 1; i < len(flow); i++ {
+ w.logTransition(ctx, nil, flow[i-1], flow[i])
+ // A status poll, a pass that changes nothing, a replayed write.
+ w.observeMove(ctx, nil, flow[i])
+ w.logTransition(ctx, nil, flow[i], flow[i])
+ w.logTransition(ctx, nil, flow[i-1], flow[i])
+ }
+ var events []string
+ for _, ev := range got {
+ events = append(events, ev.Event)
+ if ev.Op != evMoveID || ev.Host != "acme.go.tito.io" || ev.Actor != "box-test" || ev.V != 1 || ev.At.IsZero() {
+ t.Errorf("event %+v", ev)
+ }
+ }
+ if strings.Join(events, " ") != strings.Join(doneFlowEvents, " ") {
+ t.Fatalf("events\n got %v\nwant %v", events, doneFlowEvents)
+ }
+ ids := map[string]bool{}
+ for _, ev := range got {
+ if ids[ev.Identity] {
+ t.Fatalf("identity %q twice", ev.Identity)
+ }
+ ids[ev.Identity] = true
+ }
+ for _, want := range []string{"requested", "ready", "cutting-requested", "cutting-stopped", "cutting-flipped", "landing-probing", "probe_attempt-1", "committed", "committed-opened", "done"} {
+ if !ids[want] {
+ t.Errorf("no event with identity %q in %v", want, ids)
+ }
+ }
+ if f := got[len(got)-1].Fields; f["offline_s"] != "41.2" {
+ t.Errorf("done carries %v, want offline_s", f)
+ }
+ if f := got[19].Fields; got[19].Event != "opened" || f["offline_s"] != "41" {
+ t.Errorf("opened carries %v, want offline_s from cut_at to the open", f)
+ }
+}
+
+// A rollback, a stuck and a retry: the rollback's phases are named for the
+// half that moved; stuck and retry are keyed by the recovery retry epoch
+// (claim Plan 10's RecoveryRetryEpoch, preflight L11).
+func TestRollbackStuckAndRetryEvents(t *testing.T) {
+ ctx := context.Background()
+ var got []moves.LogEvent
+ w := eventWorker(&got)
+ rb := func(target, source string, mut func(*provision.MoveState)) provision.Record {
+ return evRec(provision.MoveRollingBack, func(m *provision.MoveState) {
+ m.CutAt = evAt(10)
+ m.Rollback = &provision.MoveRollback{ID: "rb_x", Reason: provision.RollbackLandingDeadline, TargetPhase: target, SourcePhase: source}
+ if mut != nil {
+ mut(m)
+ }
+ })
+ }
+ stuck := func(epoch uint64) provision.Record {
+ return evRec(provision.MoveStuck, func(m *provision.MoveState) {
+ m.CutAt = evAt(10)
+ m.Rollback = &provision.MoveRollback{ID: "rb_x", TargetPhase: provision.RbTargetCleaned, SourcePhase: provision.RbSourceVerified, StuckAt: provision.RbSourceVerified}
+ m.StuckReason, m.RecoveryRetryEpoch = provision.StuckBaselineMismatch, epoch
+ })
+ }
+ landing := evRec(provision.MoveLanding, func(m *provision.MoveState) { m.LandingPhase = provision.LandProbing })
+ seq := []provision.Record{
+ landing,
+ rb(provision.RbTargetReleasing, provision.RbSourceWaiting, nil),
+ rb(provision.RbTargetReleased, provision.RbSourceWaiting, nil),
+ rb(provision.RbTargetReleased, provision.RbSourceClaimed, nil),
+ rb(provision.RbTargetCleaned, provision.RbSourceVerified, nil),
+ stuck(0),
+ }
+ for i := 1; i < len(seq); i++ {
+ w.logTransition(ctx, nil, seq[i-1], seq[i])
+ }
+ // The operator's retry (the console's write, seen by the box) advances
+ // the epoch and leaves stuck; the second stuck is a new entry.
+ retried := rb(provision.RbTargetCleaned, provision.RbSourceVerified, func(m *provision.MoveState) { m.RecoveryRetryEpoch = 1 })
+ w.observeMove(ctx, nil, retried)
+ w.observeMove(ctx, nil, retried)
+ w.logTransition(ctx, nil, retried, stuck(1))
+ w.logTransition(ctx, nil, stuck(1), provision.Record{Account: "bab", App: "titogo", Host: "acme.go.tito.io", Region: "ie",
+ LastMove: &provision.MoveOutcome{ID: evMoveID, Outcome: "rolled_back", At: *evAt(90), OfflineS: 80}})
+ var events, ids []string
+ for _, ev := range got {
+ events = append(events, ev.Event)
+ ids = append(ids, ev.Identity)
+ }
+ want := "rolling_back released claimed cleaned stuck retry stuck rolled_back"
+ if strings.Join(events, " ") != want {
+ t.Fatalf("events %v, want %s", events, want)
+ }
+ wantIDs := "rolling_back rolling_back-target-released rolling_back-source-claimed rolling_back-target-cleaned stuck-0 retry-1 stuck-1 rolled_back"
+ if strings.Join(ids, " ") != wantIDs {
+ t.Fatalf("identities %v, want %s", ids, wantIDs)
+ }
+}
+
+type memberSink struct{ events []string }
+
+func (s *memberSink) App(acct, app, host, line string) {}
+func (s *memberSink) Event(acct, app, host, level, msg string) {
+ s.events = append(s.events, acct+"/"+app+"/"+host+" "+level+" "+msg)
+}
+
+// Preflight L9 and L27: the production LogEvent writes the object
+// create-only, and only the writer that created it tells members; a second
+// writer, or a replay, does neither. Steps members do not need (stopping)
+// are logged and not shown to them.
+func TestMoveLogWriterTellsMembersOnce(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ sink := &memberSink{}
+ write := moveLogWriter(sink)
+ flow := doneFlow()
+ ev := func(event, identity string, fields map[string]string) moves.LogEvent {
+ return moves.LogEvent{V: 1, Op: evMoveID, Host: "acme.go.tito.io", Event: event, Identity: identity, Actor: "box-uk", At: time.Now().UTC(), Fields: fields}
+ }
+ done := flow[len(flow)-1]
+ write(ctx, s, done, ev("done", "done", map[string]string{"offline_s": "41.2", "from": "ie", "to": "uk"}))
+ write(ctx, s, done, ev("done", "done", map[string]string{"offline_s": "41.2", "from": "ie", "to": "uk"}))
+ write(ctx, s, flow[4], ev("stopping", "cutting-stopping", map[string]string{"from": "ie", "to": "uk"}))
+ if len(sink.events) != 1 || sink.events[0] != "bab/titogo/acme.go.tito.io info Moved. acme.go.tito.io is served from uk. It was offline for 41 seconds." {
+ t.Fatalf("member events %q", sink.events)
+ }
+ prefix := moves.ReportPrefix("bab", "titogo", "acme.go.tito.io", evMoveID)
+ logged, err := moves.ReadLog(ctx, s, prefix)
+ if err != nil || len(logged) != 2 || logged[0].Actor != "box-uk" {
+ t.Fatalf("log %+v %v", logged, err)
+ }
+}
+
+// The member lines (preflight L9, Paul's choice): M-C1, M-C5, M-C6, M-C7,
+// M-C8, M-C9, and for stuck a line with no operator command in it.
+func TestMemberLines(t *testing.T) {
+ flow := doneFlow()
+ rolling := evRec(provision.MoveRollingBack, func(m *provision.MoveState) {
+ m.Rollback = &provision.MoveRollback{ID: "rb_x", Reason: provision.RollbackLandingDeadline}
+ })
+ stuck := evRec(provision.MoveStuck, nil)
+ rolledBack := provision.Record{Account: "bab", App: "titogo", Host: "acme.go.tito.io", Region: "ie",
+ LastMove: &provision.MoveOutcome{ID: evMoveID, Outcome: "rolled_back"}}
+ for _, c := range []struct {
+ event string
+ rec provision.Record
+ level string
+ line string
+ }{
+ {"requested", flow[0], "info", "Moving acme.go.tito.io from ie to uk."},
+ {"admitted", flow[3], "info", "uk is ready. Starting the cut. acme.go.tito.io will be offline for about 45 seconds."},
+ {"flipped", flow[8], "info", "Stopped in ie. Starting in uk."},
+ {"rolling_back", rolling, "warn", "uk could not start acme.go.tito.io: " + provision.RollbackLandingDeadline + ". Moving it back to ie."},
+ {"rolled_back", rolledBack, "warn", "acme.go.tito.io is back in ie and serving. Nothing was lost."},
+ {"stuck", stuck, "error", "acme.go.tito.io could not start in uk or in ie. It is offline. An operator has been paged."},
+ {"stopping", flow[4], "", ""},
+ } {
+ line, ok := memberLine(moves.LogEvent{Event: c.event, Fields: map[string]string{"from": "ie", "to": "uk"}}, c.rec)
+ level := ""
+ if ok {
+ level = memberLevel(c.event)
+ }
+ if ok != (c.line != "") || level != c.level || line != c.line {
+ t.Errorf("%s: %q %q %v, want %q %q", c.event, level, line, ok, c.level, c.line)
+ }
+ }
+}
diff --git a/internal/host/movesfinal.go b/internal/host/movesfinal.go
index 3c3ed9d32..bf83f6409 100644
--- a/internal/host/movesfinal.go
+++ b/internal/host/movesfinal.go
@@ -678,7 +678,11 @@ func (f *sourceFinal) configAck(ctx context.Context) error {
// close needs it; this commit without the stamp is an error.
func (f *sourceFinal) stampParked(ctx context.Context) error {
host := f.rec.Host
- _, err := provision.SetRecordMove(ctx, f.s, f.rec.Account, f.rec.App, host, provision.MoveSourceParked(f.g, f.now()))
+ var pre provision.Record
+ after, err := provision.SetRecordMove(ctx, f.s, f.rec.Account, f.rec.App, host, observedMove(&pre, provision.MoveSourceParked(f.g, f.now())))
+ if err == nil {
+ f.w.logTransition(ctx, f.s, pre, *after)
+ }
if errors.Is(err, provision.ErrMoveConflict) {
fresh, gerr := provision.GetRecord(ctx, f.s, f.rec.Account, f.rec.App, host)
if gerr != nil {
diff --git a/internal/host/movesinstall.go b/internal/host/movesinstall.go
index f90509e38..6b68e8fc5 100644
--- a/internal/host/movesinstall.go
+++ b/internal/host/movesinstall.go
@@ -264,13 +264,16 @@ func (ir *installRun) seal(c *hostclaim.Claim) (moves.Seal, error) {
// decision (a rollback, a stuck mark, a console write): logged, and the call
// ends without error.
func (ir *installRun) transition(ctx context.Context, what string, mut func(*provision.Record) error) (bool, error) {
- if _, err := provision.SetRecordMove(ctx, ir.s, ir.rec.Account, ir.rec.App, ir.rec.Host, mut); err != nil {
+ var pre provision.Record
+ after, err := provision.SetRecordMove(ctx, ir.s, ir.rec.Account, ir.rec.App, ir.rec.Host, observedMove(&pre, mut))
+ if err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: %s of %s refused (%v); the record moved on", ir.rec.Host, what, ir.g.ID, err)
return false, nil
}
return false, fmt.Errorf("%s: %w", what, err)
}
+ ir.w.logTransition(ctx, ir.s, pre, *after)
return true, nil
}
@@ -1082,13 +1085,16 @@ func recoveryInstallSource(w *movesWorker, s release.Store, rec provision.Record
default:
return false, "", fmt.Errorf("move %s: a return's install has no refusal %d", g.ID, why)
}
- if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, host, provision.MoveMarkStuck(g, reason)); err != nil {
+ var pre provision.Record
+ after, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, host, observedMove(&pre, provision.MoveMarkStuck(g, reason)))
+ if err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: the stuck mark of %s refused (%v); the record moved on", host, rb, err)
return false, "", nil
}
return false, "", fmt.Errorf("mark the return stuck: %w", err)
}
+ w.logTransition(ctx, s, pre, *after)
log.Printf("moves: %s: the return %s stops for an operator: %s", host, rb, reason)
nudgeSerial(ctx, s)
return true, reason, nil
diff --git a/internal/host/movesland.go b/internal/host/movesland.go
index 6367404db..f446b5b5f 100644
--- a/internal/host/movesland.go
+++ b/internal/host/movesland.go
@@ -285,13 +285,16 @@ func landingIndex(order []string, phase string) int {
// else's decision (a rollback, a console write): logged, and the pass ends
// without error.
func (l *landingRun) transition(ctx context.Context, what string, mut func(*provision.Record) error) (bool, error) {
- if _, err := provision.SetRecordMove(ctx, l.s, l.rec.Account, l.rec.App, l.rec.Host, mut); err != nil {
+ var pre provision.Record
+ after, err := provision.SetRecordMove(ctx, l.s, l.rec.Account, l.rec.App, l.rec.Host, observedMove(&pre, mut))
+ if err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: %s of %s refused (%v); the record moved on", l.rec.Host, what, l.g.ID, err)
return false, nil
}
return false, fmt.Errorf("%s: %w", what, err)
}
+ l.w.logTransition(ctx, l.s, pre, *after)
return true, nil
}
diff --git a/internal/host/moveslandcommit.go b/internal/host/moveslandcommit.go
index 2a564e71b..f0410e810 100644
--- a/internal/host/moveslandcommit.go
+++ b/internal/host/moveslandcommit.go
@@ -56,7 +56,11 @@ func (l *landingRun) probed(ctx context.Context) (bool, error) {
if err := l.fault("before-commit"); err != nil {
return false, err
}
- fresh, err := provision.SetRecordMove(ctx, l.s, l.rec.Account, l.rec.App, l.rec.Host, provision.MoveCommit(l.g, l.cfg().Now()))
+ var pre provision.Record
+ fresh, err := provision.SetRecordMove(ctx, l.s, l.rec.Account, l.rec.App, l.rec.Host, observedMove(&pre, provision.MoveCommit(l.g, l.cfg().Now())))
+ if err == nil {
+ l.w.logTransition(ctx, l.s, pre, *fresh)
+ }
switch {
case errors.Is(err, provision.ErrCutDeadline):
log.Printf("moves: %s: the commit of landing %s came after its deadline", l.rec.Host, l.g.ID)
@@ -243,10 +247,13 @@ func (l *landingRun) open(ctx context.Context) (stamped bool, err error) {
cfg.ConfirmProbeOpen(host, id)
}
if fresh.Move.OpenedAt == nil {
- if _, err = provision.SetRecordMove(ctx, l.s, l.rec.Account, l.rec.App, host, provision.MoveOpened(l.g, cfg.Now())); err != nil {
+ var pre provision.Record
+ var opened *provision.Record
+ if opened, err = provision.SetRecordMove(ctx, l.s, l.rec.Account, l.rec.App, host, observedMove(&pre, provision.MoveOpened(l.g, cfg.Now()))); err != nil {
err = fmt.Errorf("the record's opened_at: %w", err)
return
}
+ l.w.logTransition(ctx, l.s, pre, *opened)
stamped = true
}
})
diff --git a/internal/host/movesparks.go b/internal/host/movesparks.go
index 0e515bf47..8fb5df098 100644
--- a/internal/host/movesparks.go
+++ b/internal/host/movesparks.go
@@ -97,7 +97,6 @@ func (w *movesWorker) closeMove(ctx context.Context, s release.Store, rec provis
w.notice(host+"#close", "moves: %s: move %s waits for %d park(s) of the move to be acknowledged before it closes", host, id, owed)
return nil
}
- w.logEvent(ctx, *fresh, "done")
target := c.Op.Participants[1]
if _, err := hostclaim.Update(ctx, s, obs, hostclaim.ReleaseMove(id, target)); err != nil {
return fmt.Errorf("release the claim: %w", err)
@@ -110,13 +109,16 @@ func (w *movesWorker) closeMove(ctx context.Context, s release.Store, rec provis
}
}
g := provision.MoveGuard{ID: id, Incarnation: inc}
- if _, err := provision.SetRecordMove(ctx, s, fresh.Account, fresh.App, host, provision.MoveClose(g, w.now())); err != nil {
+ var pre provision.Record
+ closed, err := provision.SetRecordMove(ctx, s, fresh.Account, fresh.App, host, observedMove(&pre, provision.MoveClose(g, w.now())))
+ if err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: the close of move %s refused (%v); the record moved on", host, id, err)
return nil
}
return fmt.Errorf("close the move: %w", err)
}
+ w.logTransition(ctx, s, pre, *closed)
log.Printf("moves: %s: move %s is done; the instance is in %s", host, id, m.To)
nudgeSerial(ctx, s)
return nil
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index 4edda2f80..90d9dcb8a 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -90,7 +90,10 @@ type movesWorker struct {
archives map[string]verifiedArchive
measured moves.Measurements
notices sync.Map // key → time.Time of the last line
- strict atomic.Pointer[map[string]bool]
+ // logged is the move log identities this worker has handed to LogEvent
+ // (movesevents.go): host/op/identity → true.
+ logged sync.Map
+ strict atomic.Pointer[map[string]bool]
// holds is what reconcile and the schedule halves ask about this
// box's running moves (moveshold.go), rebuilt every pass. Nil with the
// switch off.
@@ -185,7 +188,7 @@ func (w *movesWorker) mirror(ctx context.Context, dstRegion string) (lease.Mirro
// Phase 1 launch environment and S2). Called once, before svc.Start: with
// the Phase 1 switch on it sets eb.Strict, which the spawn path reads; with
// it off eb.Strict stays nil, so every spawn is main's.
-func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, writes hostwrite.Config) {
+func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, writes hostwrite.Config, events activator.EventSink) {
readyDelta, ok := moveReadyDeltaFromEnv(os.Getenv)
if !ok {
log.Printf("agent: %s=%q ignored: want a whole number of bytes, 0 or more — keeping the built-in default of %d", moveReadyDeltaEnv, os.Getenv(moveReadyDeltaEnv), defaultMoveReadyDelta)
@@ -233,6 +236,9 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry,
Enabled: moves.Enabled(os.Getenv),
claimRecords: writes.Enabled,
Page: PageOperator,
+ // The move log and the member stream (Plan 6 Task 3, preflight L9).
+ Events: events,
+ LogEvent: moveLogWriter(events),
}
w := &movesWorker{mirrors: map[string]lease.MirrorStore{}}
if cfg.Enabled {
@@ -609,13 +615,16 @@ type moveCfg struct {
// The rollback on the target (§7.8, movesrollback.go). HandHold hands
// the row's hold from the move to its rollback (handHold, ruling J3).
- // LogEvent records a move event (moves.LogEvent, Plan 6): nil in
- // production, where the event is a log line (ruling J18). The target's
+ // LogEvent records a move event (moves.LogEvent, Plan 6 Task 3): in
+ // production moveLogWriter, the move log's create-only object and, when
+ // this writer created it, the member event through Events (the agent's
+ // rollout sink). Nil, the event is a log line. The target's
// half also uses the cut's stop hooks, the landing's park hooks and
// DeleteHeld above; ParkLeases nil leaves it unarmed in production
// (rollbackTargetArmed). rbFault is its test seam, nil in production.
HandHold func(host, status, toOp, fromOp string, validate func() error) (bool, error)
- LogEvent func(ctx context.Context, rec provision.Record, event string)
+ LogEvent func(ctx context.Context, s release.Store, rec provision.Record, ev moves.LogEvent)
+ Events activator.EventSink
rbFault func(point string) error
// Page pages an operator (moves spec §13, K11): PageOperator in
diff --git a/internal/host/movespass_test.go b/internal/host/movespass_test.go
index 7a4161b49..5fc8b47f5 100644
--- a/internal/host/movespass_test.go
+++ b/internal/host/movespass_test.go
@@ -926,7 +926,7 @@ func TestArmMovesPassReadsTheReadyDelta(t *testing.T) {
{"lots", defaultMoveReadyDelta},
} {
t.Setenv("CARLOS_MOVE_READY_DELTA_BYTES", tc.raw)
- armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{})
+ armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{}, nil)
if got := armedMovesWorker.Load().cfg.ReadyDelta; got != tc.want {
t.Errorf("CARLOS_MOVE_READY_DELTA_BYTES=%q armed a bar of %d, want %d", tc.raw, got, tc.want)
}
@@ -945,7 +945,7 @@ func TestArmMovesPassReadsTheSwitch(t *testing.T) {
{"", false}, {"on", true}, {"1", true}, {"true", true}, {"off", false}, {"yes", false},
} {
t.Setenv(moves.SwitchEnv, tc.raw)
- armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{})
+ armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{}, nil)
if got := armedMovesWorker.Load().cfg.Enabled; got != tc.want {
t.Errorf("%s=%q armed Enabled=%v, want %v", moves.SwitchEnv, tc.raw, got, tc.want)
}
diff --git a/internal/host/movesphase1.go b/internal/host/movesphase1.go
index 0e2eb9f8e..952c3a3c1 100644
--- a/internal/host/movesphase1.go
+++ b/internal/host/movesphase1.go
@@ -180,6 +180,9 @@ func writeMoveLedger(ctx context.Context, m lease.MirrorStore, bucket, key strin
// phase1MoveRound is the Phase 1 worker's step for one id'd record this box
// is the source of.
func phase1MoveRound(ctx context.Context, s release.Store, w *movesWorker, rec provision.Record, fleet moveFleet) error {
+ // The transitions another writer made (the console's doors, a declared
+ // loss) are logged by the box that sees them (movesevents.go).
+ w.observeMove(ctx, s, rec)
switch rec.Move.State {
case provision.MovePreparing, provision.MoveReady:
return w.preparingRound(ctx, s, rec, fleet)
@@ -899,13 +902,16 @@ func (w *movesWorker) decideReady(ctx context.Context, s release.Store, rec prov
if repin {
mut = provision.MoveGuardedRepin(g, in)
}
- if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, mut); err != nil {
+ var pre provision.Record
+ after, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, observedMove(&pre, mut))
+ if err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: the record no longer holds move %s as %s without a pin (%v); this round ends", rec.Host, g.ID, state, err)
return nil
}
return err
}
+ w.logTransition(ctx, s, pre, *after)
nudgeSerial(ctx, s)
if repin {
log.Printf("moves: %s: move %s pinned its inputs again", rec.Host, g.ID)
@@ -915,13 +921,16 @@ func (w *movesWorker) decideReady(ctx context.Context, s release.Store, rec prov
state = provision.MoveReady
}
if state == provision.MoveReady && rec.Move.Cut == "auto" && layoutErr == nil {
- if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveGuardedCut(g)); err != nil {
+ var pre provision.Record
+ after, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, observedMove(&pre, provision.MoveGuardedCut(g)))
+ if err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: the automatic cut of %s was refused (%v)", rec.Host, g.ID, err)
return nil
}
return err
}
+ w.logTransition(ctx, s, pre, *after)
nudgeSerial(ctx, s)
log.Printf("moves: %s: move %s cut requested automatically", rec.Host, g.ID)
}
diff --git a/internal/host/movesphase1_test.go b/internal/host/movesphase1_test.go
index a6efc9a4c..46106094b 100644
--- a/internal/host/movesphase1_test.go
+++ b/internal/host/movesphase1_test.go
@@ -144,7 +144,7 @@ func TestPhase1IsInertWithTheSwitchOff(t *testing.T) {
}
t.Setenv(moves.SwitchEnv, v)
eb := &activator.ExecBackend{}
- armMovesPass(activator.Config{}, t.TempDir(), nil, nil, eb, hostwrite.Config{})
+ armMovesPass(activator.Config{}, t.TempDir(), nil, nil, eb, hostwrite.Config{}, nil)
if (eb.Strict != nil) != on {
t.Fatalf("switch %v: eb.Strict set = %v", on, eb.Strict != nil)
}
diff --git a/internal/host/movesrecover.go b/internal/host/movesrecover.go
index db4e59706..d7d0f8a3c 100644
--- a/internal/host/movesrecover.go
+++ b/internal/host/movesrecover.go
@@ -273,11 +273,14 @@ func (r *recoverRun) step(ctx context.Context) (bool, error) {
return false, nil
}
-// phase writes one source phase step, then logs the event, if any, and
+// phase writes one source phase step, logs it (movesevents.go names it from
+// the record; the last argument, the step's old event name, is unused) and
// passes the step's fault point. A conflict is someone else's decision.
-func (r *recoverRun) phase(ctx context.Context, from, to string, stamp func(*provision.MoveRollback), event string) (bool, error) {
+func (r *recoverRun) phase(ctx context.Context, from, to string, stamp func(*provision.MoveRollback), _ string) (bool, error) {
host := r.rec.Host
- if _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveRollbackSourcePhase(r.g, from, to, stamp)); err != nil {
+ var pre provision.Record
+ after, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, observedMove(&pre, provision.MoveRollbackSourcePhase(r.g, from, to, stamp)))
+ if err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: the source's recovery %s to %s refused (%v); the record moved on", host, from, to, err)
return false, nil
@@ -285,9 +288,7 @@ func (r *recoverRun) phase(ctx context.Context, from, to string, stamp func(*pro
return false, fmt.Errorf("the source's recovery %s to %s: %w", from, to, err)
}
log.Printf("moves: %s: the source of rollback %s is %s", host, r.rb, to)
- if event != "" {
- r.w.logEvent(ctx, r.rec, event)
- }
+ r.w.logTransition(ctx, r.s, pre, *after)
if err := r.fault("after-record-" + to); err != nil {
return true, err
}
@@ -299,14 +300,16 @@ func (r *recoverRun) phase(ctx context.Context, from, to string, stamp func(*pro
func (r *recoverRun) stuck(ctx context.Context, reason string, why error) (bool, error) {
host := r.rec.Host
log.Printf("moves: %s: the recovery of %s stops for an operator (%v): %s", host, r.rb, why, reason)
- if _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveMarkStuck(r.g, reason)); err != nil {
+ var pre provision.Record
+ after, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, observedMove(&pre, provision.MoveMarkStuck(r.g, reason)))
+ if err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: the stuck mark of %s refused (%v); the record moved on", host, r.rb, err)
return false, nil
}
return false, fmt.Errorf("mark the recovery stuck: %w", err)
}
- r.w.logEvent(ctx, r.rec, "source-stuck")
+ r.w.logTransition(ctx, r.s, pre, *after)
r.w.pageStuck(ctx, r.rec, "source", reason)
nudgeSerial(ctx, r.s)
return true, nil
@@ -583,7 +586,7 @@ func (r *recoverRun) install(ctx context.Context) (bool, error) {
advanced, refused, err := installGeneration(ctx, r.s, r.w, r.rec, src)
if refused != "" {
log.Printf("moves: %s: the return of %s stopped for an operator: %s", host, r.rb, refused)
- r.w.logEvent(ctx, r.rec, "source-stuck")
+ // The stuck mark is installGeneration's write, logged there.
r.w.pageStuck(ctx, r.rec, "source", refused)
}
return advanced, err
@@ -802,7 +805,6 @@ func (r *recoverRun) finalize(ctx context.Context) (bool, error) {
r.w.notice(host+"#rc-final", "moves: %s: rollback %s waits for %d chore(s) of the move to be acknowledged before it closes", host, r.rb, owed)
return false, nil
}
- r.w.logEvent(ctx, r.rec, "rolled_back")
if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.ReleaseMove(id, cfg.Box)); err != nil {
return false, fmt.Errorf("release the claim: %w", err)
}
@@ -811,13 +813,16 @@ func (r *recoverRun) finalize(ctx context.Context) (bool, error) {
return false, err
}
}
- if _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveRolledBackClose(r.g, r.now())); err != nil {
+ var pre provision.Record
+ closed, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, observedMove(&pre, provision.MoveRolledBackClose(r.g, r.now())))
+ if err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: the close of rollback %s refused (%v); the record moved on", host, r.rb, err)
return false, nil
}
return false, fmt.Errorf("close the rolled-back move: %w", err)
}
+ r.w.logTransition(ctx, r.s, pre, *closed)
log.Printf("moves: %s: move %s is rolled back and closed", host, id)
nudgeSerial(ctx, r.s)
if err := r.fault("after-close"); err != nil {
diff --git a/internal/host/movesrecover_test.go b/internal/host/movesrecover_test.go
index 928368cb6..6f354333f 100644
--- a/internal/host/movesrecover_test.go
+++ b/internal/host/movesrecover_test.go
@@ -150,7 +150,9 @@ func (h *recoverHarness) sourceCfg(t *testing.T) moveCfg {
h.srcRoute.Status, h.srcRoute.HoldOp = "ready", ""
return true, nil
}
- c.LogEvent = func(_ context.Context, _ provision.Record, event string) { h.sEvents = append(h.sEvents, event) }
+ c.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) {
+ h.sEvents = append(h.sEvents, ev.Event)
+ }
c.Page = h.pages.hook()
c.recoveryProbe = func(ctx context.Context, _ release.Store, rec provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) {
h.probes = append(h.probes, req)
@@ -367,7 +369,7 @@ func TestCleanReturnVerifiesAndOpens(t *testing.T) {
t.Fatalf("a clean return queued %+v", o)
}
}
- if !strings.Contains(strings.Join(h.sEvents, " "), "source-recovered source-opened") {
+ if !strings.Contains(strings.Join(h.sEvents, " "), "recovered opened") {
t.Fatalf("events %v", h.sEvents)
}
// The record waits for the target's cleaning: a second pass changes
@@ -542,7 +544,7 @@ func TestVerifyMismatchIsStuck(t *testing.T) {
h.retry(t)
h.srcPass(t)
h.opened(t)
- if !strings.Contains(strings.Join(h.sEvents, " "), "source-stuck") {
+ if !strings.Contains(strings.Join(h.sEvents, " "), "stuck") {
t.Fatalf("events %v", h.sEvents)
}
})
diff --git a/internal/host/movesrevert.go b/internal/host/movesrevert.go
index 406b0294e..4e06954e4 100644
--- a/internal/host/movesrevert.go
+++ b/internal/host/movesrevert.go
@@ -277,22 +277,27 @@ func (r *revertRun) stillReverting(ctx context.Context) error {
// update writes one claim transition over obs.
func (r *revertRun) update(ctx context.Context, obs hostclaim.Observed, what string, step func(*hostclaim.Claim) error) error {
- if _, err := hostclaim.Update(ctx, r.s, obs, step); err != nil {
+ after, err := hostclaim.Update(ctx, r.s, obs, step)
+ if err != nil {
return fmt.Errorf("%s: %w", what, err)
}
+ r.w.logRevPhase(ctx, r.s, r.rec, obs, after)
return nil
}
// record writes one record transition. A conflict is someone else's
// decision: logged, and the pass ends without error.
func (r *revertRun) record(ctx context.Context, what string, mut func(*provision.Record) error) (bool, error) {
- if _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, r.rec.Host, mut); err != nil {
+ var pre provision.Record
+ after, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, r.rec.Host, observedMove(&pre, mut))
+ if err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: %s of %s refused (%v); the record moved on", r.rec.Host, what, r.g.ID, err)
return false, nil
}
return false, fmt.Errorf("%s: %w", what, err)
}
+ r.w.logTransition(ctx, r.s, pre, *after)
return true, nil
}
@@ -679,7 +684,6 @@ func (r *revertRun) probed(ctx context.Context, obs hostclaim.Observed) (bool, e
return false, err
}
log.Printf("moves: %s: the source serves again; the revert of %s waits for its lane's park", host, id)
- r.w.logEvent(ctx, r.rec, "source-opened")
nudgeSerial(ctx, r.s)
if err := r.fault("after-rev-opened"); err != nil {
return false, err
@@ -706,7 +710,6 @@ func (r *revertRun) opened(ctx context.Context, obs hostclaim.Observed) (bool, e
r.w.notice(host+"#rv-opened", "moves: %s: the revert of %s waits for %d chore(s) of the move, its lane's park, to be acknowledged", host, id, owed)
return false, nil
}
- r.w.logEvent(ctx, r.rec, "rolled_back")
if err := r.update(ctx, obs, "release the claim", hostclaim.ReleaseMove(id, cfg.Box)); err != nil {
return false, err
}
@@ -746,7 +749,6 @@ func (r *revertRun) stuck(ctx context.Context, reason string, why error) (bool,
if !ok {
return false, err
}
- r.w.logEvent(ctx, r.rec, "source-stuck")
r.w.pageStuck(ctx, r.rec, "source", reason)
nudgeSerial(ctx, r.s)
return true, nil
diff --git a/internal/host/movesrevert_test.go b/internal/host/movesrevert_test.go
index dd8dfd456..a291bcde2 100644
--- a/internal/host/movesrevert_test.go
+++ b/internal/host/movesrevert_test.go
@@ -132,7 +132,9 @@ func (h *revertHarness) arm(t *testing.T) {
h.route.Status, h.route.HoldOp = "ready", ""
return true, nil
}
- h.cfg.LogEvent = func(_ context.Context, _ provision.Record, event string) { h.events = append(h.events, event) }
+ h.cfg.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) {
+ h.events = append(h.events, ev.Event)
+ }
h.cfg.recoveryProbe = func(ctx context.Context, _ release.Store, _ provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) {
h.probes = append(h.probes, req)
if h.probe != nil {
diff --git a/internal/host/movesrollback.go b/internal/host/movesrollback.go
index d52afdedc..c3b60ff54 100644
--- a/internal/host/movesrollback.go
+++ b/internal/host/movesrollback.go
@@ -123,20 +123,6 @@ func (w *movesWorker) now() time.Time {
return time.Now()
}
-// logEvent records a move event: the LogEvent hook when wired (Plan 6's
-// moves.LogEvent), else a log line.
-func (w *movesWorker) logEvent(ctx context.Context, rec provision.Record, event string) {
- if f := w.cfg.LogEvent; f != nil {
- f(ctx, rec, event)
- return
- }
- id := ""
- if rec.Move != nil {
- id = rec.Move.ID
- }
- log.Printf("moves: %s: %s %s (the move log is Plan 6's)", rec.Host, id, event)
-}
-
// pageStuck pages an operator once a MoveMarkStuck has landed (moves spec
// §13: stuck is an incident; K11). role is this box's part in the move,
// "source" or "target", and reason the recorded stuck sentence. The page
@@ -193,7 +179,8 @@ func (w *movesWorker) rollBackLanding(ctx context.Context, s release.Store, rec
return true, err
}
g := provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}
- fresh, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveRollBack(g, rbID, reason, w.now()))
+ var pre provision.Record
+ fresh, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, observedMove(&pre, provision.MoveRollBack(g, rbID, reason, w.now())))
if errors.Is(err, provision.ErrMoveConflict) {
// A commit that won (the open is the commit's path), or another
// writer's rollback (the participant pass follows it).
@@ -204,7 +191,7 @@ func (w *movesWorker) rollBackLanding(ctx context.Context, s release.Store, rec
return true, fmt.Errorf("roll the landing back: %w", err)
}
log.Printf("moves: %s: landing %s rolls back as %s: %s", rec.Host, m.ID, rbID, fresh.Move.Rollback.Reason)
- w.logEvent(ctx, *fresh, "rollback")
+ w.logTransition(ctx, s, pre, *fresh)
if _, err := w.rollbackClaim(ctx, s, *fresh); err != nil {
log.Printf("moves: %s: the claim half of rollback %s waits for the next pass: %v", rec.Host, rbID, err)
}
@@ -332,11 +319,14 @@ func (r *rollbackRun) step(ctx context.Context) (bool, error) {
return false, nil // lost: the declared loss's (Task 10)
}
-// phase writes one target phase step, then logs the event, if any, and
+// phase writes one target phase step, logs it (movesevents.go names it from
+// the record; the last argument, the step's old event name, is unused) and
// passes the step's fault point. A conflict is someone else's decision.
-func (r *rollbackRun) phase(ctx context.Context, from, to string, stamp func(*provision.MoveRollback), event string) (bool, error) {
+func (r *rollbackRun) phase(ctx context.Context, from, to string, stamp func(*provision.MoveRollback), _ string) (bool, error) {
host := r.rec.Host
- if _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveRollbackTargetPhase(r.g, from, to, stamp)); err != nil {
+ var pre provision.Record
+ after, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, observedMove(&pre, provision.MoveRollbackTargetPhase(r.g, from, to, stamp)))
+ if err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: the target's rollback %s to %s refused (%v); the record moved on", host, from, to, err)
return false, nil
@@ -344,9 +334,7 @@ func (r *rollbackRun) phase(ctx context.Context, from, to string, stamp func(*pr
return false, fmt.Errorf("the target's rollback %s to %s: %w", from, to, err)
}
log.Printf("moves: %s: the target of rollback %s is %s", host, r.rb, to)
- if event != "" {
- r.w.logEvent(ctx, r.rec, event)
- }
+ r.w.logTransition(ctx, r.s, pre, *after)
if err := r.fault("after-record-" + to); err != nil {
return true, err
}
@@ -887,14 +875,16 @@ func databaseGone(err error) bool {
func (r *rollbackRun) stuck(ctx context.Context, refused *returnRefusal) (bool, error) {
host := r.rec.Host
log.Printf("moves: %s: the return of %s cannot be built (%v); it stops for an operator: %s", host, r.rb, refused.err, refused.reason)
- if _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveMarkStuck(r.g, refused.reason)); err != nil {
+ var pre provision.Record
+ after, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, observedMove(&pre, provision.MoveMarkStuck(r.g, refused.reason)))
+ if err != nil {
if errors.Is(err, provision.ErrMoveConflict) {
log.Printf("moves: %s: the stuck mark of %s refused (%v); the record moved on", host, r.rb, err)
return false, nil
}
return false, fmt.Errorf("mark the return stuck: %w", err)
}
- r.w.logEvent(ctx, r.rec, "target-stuck")
+ r.w.logTransition(ctx, r.s, pre, *after)
r.w.pageStuck(ctx, r.rec, "target", refused.reason)
nudgeSerial(ctx, r.s)
return true, nil
diff --git a/internal/host/movesrollback_test.go b/internal/host/movesrollback_test.go
index 43581e789..e38120f9d 100644
--- a/internal/host/movesrollback_test.go
+++ b/internal/host/movesrollback_test.go
@@ -22,6 +22,7 @@ import (
"github.com/carlosframework/platform/internal/park"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
"github.com/superfly/ltx"
)
@@ -135,7 +136,9 @@ func (h *rollbackHarness) rollbackCfg(t *testing.T) moveCfg {
h.rowGone = true
return true, nil
}
- c.LogEvent = func(_ context.Context, _ provision.Record, event string) { h.events = append(h.events, event) }
+ c.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) {
+ h.events = append(h.events, ev.Event)
+ }
c.Page = h.pages.hook()
c.rbFault = func(point string) error {
h.rbHits = append(h.rbHits, point)
@@ -289,7 +292,7 @@ func TestRollbackDecisionConvertsTheMark(t *testing.T) {
cl.Op.Exec != "uk-box" || cl.Op.Journal[hostclaim.JournalTargetStamp] != strconv.FormatUint(f, 10) {
t.Fatalf("claim: %+v", cl.Op)
}
- if len(h.events) != 1 || h.events[0] != "rollback" {
+ if len(h.events) != 1 || h.events[0] != "rolling_back" {
t.Fatalf("events %v", h.events)
}
})
@@ -431,7 +434,7 @@ func TestReleaseIsCleanWhenNothingChanged(t *testing.T) {
if got := h.rb(t); got.TargetPhase != provision.RbTargetReleased {
t.Fatalf("a clean release moved on before the source recovered: %+v", got)
}
- if strings.Join(h.events, " ") != "rollback target-released" {
+ if strings.Join(h.events, " ") != "rolling_back released" {
t.Fatalf("events %v", h.events)
}
}
@@ -525,7 +528,7 @@ func TestReleaseIsDirtyWithoutSrcCovered(t *testing.T) {
if h.lease.acquired == 0 || h.lease.acquired != h.lease.released {
t.Fatalf("the staging lease: acquired %d released %d", h.lease.acquired, h.lease.released)
}
- if strings.Join(h.events, " ") != "rollback target-released target-returned" {
+ if strings.Join(h.events, " ") != "rolling_back released returning returned" {
t.Fatalf("events %v", h.events)
}
}
@@ -869,7 +872,7 @@ func TestTargetCleaningLeavesNoLiveCopy(t *testing.T) {
if role := movesParticipantRecord(h.rollbackCfg(t), h.read(t, p1Host)); role != roleNone {
t.Fatalf("a cleaned target still routes as %s", role)
}
- if !strings.HasSuffix(strings.Join(h.events, " "), "target-cleaned") {
+ if !strings.HasSuffix(strings.Join(h.events, " "), "cleaning cleaned") {
t.Fatalf("events %v", h.events)
}
}
@@ -1027,7 +1030,10 @@ func (h *rollbackHarness) stuckAtReturning(t *testing.T, reason string) {
func (h *rollbackHarness) retry(t *testing.T) {
t.Helper()
g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
- if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRetry(g)); err != nil {
+ // The claim-backed retry (claim Plan 10): it advances the recovery retry
+ // epoch, so the move log keys the next stuck as a new entry (Plan 6
+ // Task 3, preflight L11).
+ if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRetryRecovery(g)); err != nil {
t.Fatal(err)
}
}
@@ -1053,7 +1059,7 @@ func TestDirtyReturnGoesStuck(t *testing.T) {
h.retry(t)
h.rollbackPass(t)
h.stuckAtReturning(t, provision.StuckReturnTargetDatabase)
- if got := strings.Join(h.events[before:], " "); got != "target-stuck" {
+ if got := strings.Join(h.events[before:], " "); got != "stuck" {
t.Fatalf("the retry did not run the return: events %q", got)
}
})
diff --git a/internal/moves/log.go b/internal/moves/log.go
new file mode 100644
index 000000000..955306272
--- /dev/null
+++ b/internal/moves/log.go
@@ -0,0 +1,99 @@
+package moves
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The move log (spec §13, Plan 6 Task 3): one object per transition under the
+// move's report prefix, log/<identity>.json, in the deployment bucket beside
+// the reports and the stop evidence.
+//
+// The key is the transition's identity, never a counter (round-2 finding
+// 23): <state>-<phase> for a phase step (cutting-stopped, landing-probing,
+// rolling_back-source-claimed), probe_attempt-<n> for the landing's n-th
+// probe, stuck-<epoch> and retry-<epoch> keyed by the record's recovery retry
+// epoch (claim Plan 10, preflight L11), and the terminal outcome itself
+// (done, rolled_back, aborted). An object is written create-only: the first
+// writer of a transition wins, and a second writer, or a replay, meets the
+// existing object, which is success with nothing created (preflight L27).
+// The log is read in At order (ReadLog), never by key.
+
+// LogEvent is one move log object.
+type LogEvent struct {
+ V int `json:"v"`
+ Op string `json:"op"`
+ Host string `json:"host"`
+ Event string `json:"event"`
+ Identity string `json:"identity"`
+ Actor string `json:"actor"`
+ At time.Time `json:"at"`
+ Fields map[string]string `json:"fields,omitempty"`
+}
+
+// ErrLogIdentity refuses an identity that is not one key segment.
+var ErrLogIdentity = errors.New("moves: a log identity is one key segment")
+
+func validLogIdentity(id string) bool {
+ return id != "" && id != "." && id != ".." && !strings.ContainsAny(id, "/\\\x00")
+}
+
+// WriteLog writes ev at LogKeyFor(prefix, ev.Identity), create-only. created
+// is true only for the writer that made the object; an existing object is
+// success with created false.
+func WriteLog(ctx context.Context, s release.Store, prefix string, ev LogEvent) (created bool, err error) {
+ if !validLogIdentity(ev.Identity) {
+ return false, fmt.Errorf("%w: %q", ErrLogIdentity, ev.Identity)
+ }
+ body, err := json.Marshal(ev)
+ if err != nil {
+ return false, err
+ }
+ if _, err := s.Put(ctx, LogKeyFor(prefix, ev.Identity), body, ""); err != nil {
+ if errors.Is(err, release.ErrCASConflict) {
+ return false, nil
+ }
+ return false, err
+ }
+ return true, nil
+}
+
+// ReadLog reads every object under prefix's log/, ordered by At (then by
+// identity, for two at the same instant). An object that went missing
+// between the list and the read is skipped.
+func ReadLog(ctx context.Context, s release.Store, prefix string) ([]LogEvent, error) {
+ keys, err := s.List(ctx, prefix+"log/")
+ if err != nil {
+ return nil, err
+ }
+ out := make([]LogEvent, 0, len(keys))
+ for _, k := range keys {
+ body, _, err := s.Get(ctx, k)
+ if errors.Is(err, os.ErrNotExist) {
+ continue
+ }
+ if err != nil {
+ return nil, err
+ }
+ var ev LogEvent
+ if err := json.Unmarshal(body, &ev); err != nil {
+ return nil, fmt.Errorf("moves: log object %s: %w", k, err)
+ }
+ out = append(out, ev)
+ }
+ sort.SliceStable(out, func(i, j int) bool {
+ if !out[i].At.Equal(out[j].At) {
+ return out[i].At.Before(out[j].At)
+ }
+ return out[i].Identity < out[j].Identity
+ })
+ return out, nil
+}
diff --git a/internal/moves/log_test.go b/internal/moves/log_test.go
new file mode 100644
index 000000000..c5c6832ef
--- /dev/null
+++ b/internal/moves/log_test.go
@@ -0,0 +1,97 @@
+package moves
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const tLogPrefix = "moves/bab/app/a.go.example.com/mv_abcdefghijklmnopqrstuvwxyz/"
+
+func logEvent(identity, event string, at time.Time, actor string) LogEvent {
+ return LogEvent{V: 1, Op: "mv_abcdefghijklmnopqrstuvwxyz", Host: "a.go.example.com", Event: event, Identity: identity, Actor: actor, At: at}
+}
+
+// Plan 6 Task 3 (preflight L10, L27): a log object's key is the transition's
+// identity, never a counter. A key is written create-only: the first writer
+// wins and creates it; a second writer of the same transition, or a replay,
+// meets the existing object, which is success with nothing created.
+func TestLogKeysAreTransitionIdentities(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ t0 := time.Date(2026, 10, 11, 9, 0, 0, 0, time.UTC)
+
+ // Two writers of landing-probing (two boxes, or a crash and a replay),
+ // with different bodies: one object, the first writer's.
+ created, err := WriteLog(ctx, s, tLogPrefix, logEvent("landing-probing", "probing", t0, "box-uk"))
+ if err != nil || !created {
+ t.Fatalf("first writer: created %v, %v", created, err)
+ }
+ created, err = WriteLog(ctx, s, tLogPrefix, logEvent("landing-probing", "probing", t0.Add(time.Second), "box-ie"))
+ if err != nil || created {
+ t.Fatalf("second writer: created %v, %v; want no object and no error", created, err)
+ }
+ // Three probe attempts: three objects.
+ for n, id := range []string{"probe_attempt-1", "probe_attempt-2", "probe_attempt-3"} {
+ if created, err := WriteLog(ctx, s, tLogPrefix, logEvent(id, "probe_attempt", t0.Add(time.Duration(n+2)*time.Second), "box-uk")); err != nil || !created {
+ t.Fatalf("%s: created %v, %v", id, created, err)
+ }
+ }
+ // A replayed cutting-stopped writes nothing new.
+ if created, err := WriteLog(ctx, s, tLogPrefix, logEvent("cutting-stopped", "stopped", t0.Add(-time.Minute), "box-ie")); err != nil || !created {
+ t.Fatalf("cutting-stopped: %v %v", created, err)
+ }
+ if created, err := WriteLog(ctx, s, tLogPrefix, logEvent("cutting-stopped", "stopped", t0.Add(-time.Minute), "box-ie")); err != nil || created {
+ t.Fatalf("replayed cutting-stopped: created %v, %v", created, err)
+ }
+
+ keys, err := s.List(ctx, tLogPrefix+"log/")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(keys) != 5 {
+ t.Fatalf("keys %v, want five objects", keys)
+ }
+ for _, id := range []string{"landing-probing", "probe_attempt-1", "probe_attempt-2", "probe_attempt-3", "cutting-stopped"} {
+ if _, _, err := s.Get(ctx, LogKeyFor(tLogPrefix, id)); err != nil {
+ t.Fatalf("%s: %v", id, err)
+ }
+ }
+
+ // ReadLog orders by At, never by key.
+ got, err := ReadLog(ctx, s, tLogPrefix)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var order []string
+ for _, ev := range got {
+ order = append(order, ev.Identity)
+ }
+ want := "cutting-stopped landing-probing probe_attempt-1 probe_attempt-2 probe_attempt-3"
+ if j := joinSpace(order); j != want {
+ t.Fatalf("ReadLog order %q, want %q", j, want)
+ }
+ if got[1].Actor != "box-uk" {
+ t.Fatalf("landing-probing is %+v, want the first writer's", got[1])
+ }
+
+ // An identity is one key segment.
+ for _, bad := range []string{"", "a/b", "..", "."} {
+ if _, err := WriteLog(ctx, s, tLogPrefix, logEvent(bad, "x", t0, "box")); err == nil {
+ t.Errorf("identity %q accepted", bad)
+ }
+ }
+}
+
+func joinSpace(xs []string) string {
+ out := ""
+ for i, x := range xs {
+ if i > 0 {
+ out += " "
+ }
+ out += x
+ }
+ return out
+}
diff --git a/internal/moves/staging.go b/internal/moves/staging.go
index 6adca63f3..3652a11ef 100644
--- a/internal/moves/staging.go
+++ b/internal/moves/staging.go
@@ -1,7 +1,5 @@
package moves
-import "fmt"
-
// The staging layout (§7.3): one prefix per operation in the target
// bucket, one subprefix per artifact class, so MirrorPrefix on replica/
// can never delete an archive, a snapshot or move.json.
@@ -44,8 +42,11 @@ func ReportPrefix(acct, app, host, opID string) string {
func SourceReportKey(prefix string) string { return prefix + "source.json" }
func TargetReportKey(prefix string) string { return prefix + "target.json" }
func StoppedKey(prefix, role string) string { return prefix + "stopped-" + role + ".json" }
-func LogKey(prefix string, seq int, event string) string {
- return fmt.Sprintf("%slog/%07d-%s.json", prefix, seq, event)
+
+// LogKeyFor is the move log object of one transition: its key is the
+// transition's identity (log.go), never a counter.
+func LogKeyFor(prefix, identity string) string {
+ return prefix + "log/" + identity + ".json"
}
// ProbeKey is where a target box measures regional S3 by putting and
diff --git a/internal/moves/staging_test.go b/internal/moves/staging_test.go
index 65fe4e17b..2035864f0 100644
--- a/internal/moves/staging_test.go
+++ b/internal/moves/staging_test.go
@@ -20,7 +20,7 @@ func TestStagingKeysAreScopedPerOperation(t *testing.T) {
t.Fatal("source-bucket keys")
}
p := ReportPrefix("bab", "app", host, op)
- if p != "moves/bab/app/a.go.example.com/mv_abcdefghijklmnopqrstuvwxyz/" || StoppedKey(p, "source") != p+"stopped-source.json" || TargetReportKey(p) != p+"target.json" || SourceReportKey(p) != p+"source.json" || LogKey(p, 7, "flipped") != p+"log/0000007-flipped.json" {
+ if p != "moves/bab/app/a.go.example.com/mv_abcdefghijklmnopqrstuvwxyz/" || StoppedKey(p, "source") != p+"stopped-source.json" || TargetReportKey(p) != p+"target.json" || SourceReportKey(p) != p+"source.json" || LogKeyFor(p, "cutting-flipped") != p+"log/cutting-flipped.json" {
t.Fatal("deployment-bucket keys")
}
// The park scope grammar has no trailing slash (claim rev 6.23 §2).
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go
index 318fa9e46..3fa6ea9b6 100644
--- a/internal/provision/plan9_b8_census_test.go
+++ b/internal/provision/plan9_b8_census_test.go
@@ -516,37 +516,37 @@ var plan9B8CensusInventory = []censusEntry{
// Copies config acknowledgements to moves/<account>/<app>/<host>/<op>/config-ack-source.json.
{"internal/host/movesfinal.go", "sourceFinal.configAck", "nonrecord", "sel:Put", "cca53109e6945a73b7f04705d8965b3b0997f2b7d2a82fa6d86cffa4b4174e93"},
// Stamps source_parked_at through SetRecordMove, updateRecord and WriteRecord.
- {"internal/host/movesfinal.go", "sourceFinal.stampParked", "record-writer", "key:SetRecordMove", "8af033d29c8e2275d996df9a4e53227da30a15e8fd4c198a4ab40865db7fabb9"},
+ {"internal/host/movesfinal.go", "sourceFinal.stampParked", "record-writer", "key:SetRecordMove", "5eeb2386021980ea668123d3acb2d2ccefce9a8561e32be8937504b115dd5ffb"},
// Applies install transitions only through SetRecordMove and its sequenced writer.
- {"internal/host/movesinstall.go", "installRun.transition", "record-writer", "key:SetRecordMove", "5679de7898730c831906e5350529a67259d85ad0a94132ac86175a1102e21753"},
+ {"internal/host/movesinstall.go", "installRun.transition", "record-writer", "key:SetRecordMove", "1c6d390393de6065acc6d11409a493ac2fc49299b1f18f08c329a8af68f88c20"},
// The refusal callback marks the return stuck through SetRecordMove; other callbacks only read or mutate memory.
- {"internal/host/movesinstall.go", "recoveryInstallSource", "record-writer", "key:SetRecordMove", "b4bb0798a032c1599b9063a165656ab76c00b6ea349cb17b5f6a95769432bd5f"},
+ {"internal/host/movesinstall.go", "recoveryInstallSource", "record-writer", "key:SetRecordMove", "2b5c592c20612eb5edb58856d05755244ea25bec6aa65632c6644c274a69474e"},
// Commits the landing through SetRecordMove (MoveCommit), its refusal the rollback mark.
- {"internal/host/moveslandcommit.go", "landingRun.probed", "record-writer", "key:SetRecordMove", "6f961790d8fe169af68e2f70790c400d37c2388a4f62c26921f22b8b47c859de"},
+ {"internal/host/moveslandcommit.go", "landingRun.probed", "record-writer", "key:SetRecordMove", "d914debf1ba5d011767fc06cb430c09e39a469dfb182f5996fa3bda806499952"},
// Stamps opened_at through SetRecordMove (MoveOpened) inside the hold lock, after the proved release.
- {"internal/host/moveslandcommit.go", "landingRun.open", "record-writer", "key:SetRecordMove", "65e6908056c3055d5d0df2c27bb88943e31c08691a2173a090a9edb94c8c0b49"},
+ {"internal/host/moveslandcommit.go", "landingRun.open", "record-writer", "key:SetRecordMove", "febac996d3bb9caf4021cbc4bcffe1c900bddba4287ff95f083bd5e0751e9bc2"},
// Closes a committed move through SetRecordMove (MoveClose, LastMove done) after the claim's ReleaseMove; never writes the claim after the record.
- {"internal/host/movesparks.go", "movesWorker.closeMove", "record-writer", "key:SetRecordMove", "849d114cf80a45004e9d01c66fcf11516381b046fe0f7c40294a628a3b922b90"},
+ {"internal/host/movesparks.go", "movesWorker.closeMove", "record-writer", "key:SetRecordMove", "da86a00f3ce1b8a9d885c9b533990beca7f4f9170d8ae3966702383342828984"},
// Closes the rollback record through SetRecordMove; claim release and local journal parking use separate storage.
- {"internal/host/movesrecover.go", "recoverRun.finalize", "record-writer", "key:SetRecordMove", "fff2e28c78db68bccb16da6ab70ca877b51243b0aa56b9cb2c513c6f5660f7f7"},
+ {"internal/host/movesrecover.go", "recoverRun.finalize", "record-writer", "key:SetRecordMove", "7c22716ee494d5f1c83ed2ae46fe8a08af37e12a10779208d87dc2c00471e347"},
// Advances the source recovery phase through SetRecordMove and its sequenced writer.
- {"internal/host/movesrecover.go", "recoverRun.phase", "record-writer", "key:SetRecordMove", "d67132b6e797a88a9677d962c5c7aa30fc7f85fdbdff06ce8318562397a9ee78"},
+ {"internal/host/movesrecover.go", "recoverRun.phase", "record-writer", "key:SetRecordMove", "be641eaba77b940ce471696720b637b9dfb9454de0893aca1fcc40750135719d"},
// Records probe inputs through SetRecordMove; phase and stuck helpers use the same writer.
{"internal/host/movesrecover.go", "recoverRun.probing", "record-writer", "key:SetRecordMove", "a0144975bf52b29bbc6c2354636d3665487365d24838d3ef4baef1c665ce52ab"},
// Marks recovery stuck through SetRecordMove; event and serial writes are not record objects.
- {"internal/host/movesrecover.go", "recoverRun.stuck", "record-writer", "key:SetRecordMove", "4ea620140bbc75b127a893926eefcf58685b40c8362f86ba94f22931a05b449a"},
+ {"internal/host/movesrecover.go", "recoverRun.stuck", "record-writer", "key:SetRecordMove", "d634bbd18794aedf081c51f6e3c16b62a9b6b82b21b24f636b949c840c8e5776"},
// Records revert inputs through SetRecordMove; stuck uses the reviewed record helper and journals are local.
{"internal/host/movesrevert.go", "revertRun.probing", "record-writer", "key:SetRecordMove", "55e249b195b2325917763cdc87cfcd44563d95095918d696f14fa5cec1859898"},
// Applies revert record mutations only through SetRecordMove and its sequenced writer.
- {"internal/host/movesrevert.go", "revertRun.record", "record-writer", "key:SetRecordMove", "843bea75b1a185cb4294e4e1afc65f44182767b489c6e604b96f829ae59909b0"},
+ {"internal/host/movesrevert.go", "revertRun.record", "record-writer", "key:SetRecordMove", "f42100f118eb62ceece23b49423bda2ef357f1aa065760b288bf20340e36a920"},
// Starts rollback through SetRecordMove; the companion claim update targets the separate claim namespace.
- {"internal/host/movesrollback.go", "movesWorker.rollBackLanding", "record-writer", "key:SetRecordMove", "40ff87900fbfacb5a4dff481db37ae7dbed41e5727f13185ba8623daf3e1681e"},
+ {"internal/host/movesrollback.go", "movesWorker.rollBackLanding", "record-writer", "key:SetRecordMove", "c834e9bb708ef05fbdb80d1677bc409ae02aaa8cae93fa5767fde71f446a571e"},
// Writes only moves/<host>/<rollback>/move.json in the guarded staging bucket.
{"internal/host/movesrollback.go", "rollbackRun.laneMoveJSON", "nonrecord", "sel:PutIfAbsentVersioned", "5fd541f4a1daf2600ff266583fc9627cc0cab35e276f1d9c8f097f128c2c4e53"},
// Advances the target rollback phase through SetRecordMove and its sequenced writer.
- {"internal/host/movesrollback.go", "rollbackRun.phase", "record-writer", "key:SetRecordMove", "76f1efb1076029e5da4c0cc0d750fbac5e1c94a6014c2b1e4af241797350827b"},
+ {"internal/host/movesrollback.go", "rollbackRun.phase", "record-writer", "key:SetRecordMove", "72369c7aafffb3df49f9b6e0f493e21967dd7262e32ddcb6eaff0dc9ff255d33"},
// Marks the return stuck through SetRecordMove; event and serial writes are not record objects.
- {"internal/host/movesrollback.go", "rollbackRun.stuck", "record-writer", "key:SetRecordMove", "84fcea399a55d77ceee17e35652ef269e5811453ffff62dff43208b68538b1ed"},
+ {"internal/host/movesrollback.go", "rollbackRun.stuck", "record-writer", "key:SetRecordMove", "ba42217c6ce7fee0afc051f2a34565e6e0af7a6d97ec7c1a983391414ef03b35"},
// Validates the pointer-digest marker key and Put/Delete method names without writing.
{"internal/hostops/mutation_classifier.go", "classifyPointerDigestMarker", "reference", "str:Put", "08111b5db5d98e788d1eef5861ad9c861cf926c5e1d5d54717fd650d2dcf9fa8"},
// Validates the promotion-record key (releases/<acct>/<app>/promoted/<version>, a release object, not a
@@ -660,16 +660,16 @@ var plan9B8CensusInventory = []censusEntry{
{"internal/host/fleetbind.go", "WriteFleetBinding", "nonrecord", "sel:Put", "44ab984c2c9b30abcc938132118f61a6922ccc0cd40ed2b6cb61c7654aa0a1c3"},
{"internal/host/logship.go", "logshipFlush", "nonrecord", "sel:Put", "a2fa594c1206028af170a2d0cbedef55ba00a2f1c698a6853e1214cc3b172dd9"},
{"internal/host/metrics_loop.go", "putAlertState", "nonrecord", "sel:Put", "d5b6e8b34dc5a731d6bbafeac0b7d0c5c20706d6f413dff38088b3321b35a26c"},
- {"internal/host/movesabort.go", "movesWorker.abortSource", "record-writer", "key:SetRecordMove", "70f2cd7ec7709045043081c078f1abe63ef822325eef7016ab9f595a16cabad5"},
+ {"internal/host/movesabort.go", "movesWorker.abortSource", "record-writer", "key:SetRecordMove", "07fe8eeb9224485c313929549234e429d5a7bf62f060e07560e0072cb3d5e667"},
{"internal/host/movesarchive.go", "cutRun.publishCutArchive", "nonrecord", "sel:CopyVersioned,sel:PutIfAbsent", "6c7fcfd2d4b2a0319ac15ecbfe003ed0292fe160457fd5704be508c2ab302a1b"},
- {"internal/host/movescut.go", "cutRun.cutTransition", "record-writer", "key:SetRecordMove", "d0d67399aa784036de43b3c143e88c3d85f8b3d9e6345f84908d3b58ff56191e"},
+ {"internal/host/movescut.go", "cutRun.cutTransition", "record-writer", "key:SetRecordMove", "3ec34659c98aa9ec12221b30363d0e79a3b2d77b99051e0484f2a923c884c0af"},
{"internal/host/movescut.go", "publishStoppedReport", "nonrecord", "sel:Put", "c710682f31d99127ad76524add246fda9f640c63934940e45c79ea25ba36eb3c"},
{"internal/host/movescut.go", "cutRun.snapshotTimers", "nonrecord", "sel:PutIfAbsent", "9137042d3c269f8cfc94bd9d04cde24423e8824986521906c73159cd0a552ca4"},
{"internal/host/movesguard.go", "guardedMirror.CopyVersioned", "versioned-transport", "def:CopyVersioned,sel:CopyVersionedAt", "069edf385ccc13d50452bd97eae222c0312a629a48d619c922f9985aa1c99a7b"},
{"internal/host/movesguard.go", "guardedMirror.Copy", "versioned-transport", "sel:CopyVersioned", "1689745fdd642c0dbe855b12b8d1ef220f3a9c3daed7ca49bdeddfd6ff9d5ade"},
{"internal/host/movesguard.go", "guardedMirror.PutIfAbsentVersioned", "versioned-transport", "def:PutIfAbsentVersioned,sel:PutIfAbsentVersionedAt", "68457755ed3c8b470064e9ccbefe42c819f877d596352669d7126d6032a7726f"},
{"internal/host/movesguard.go", "guardedMirror.Put", "versioned-transport", "def:Put,sel:Put", "a3168524604a41ea42e8fcf6e5a1335e4ea7be77fde089aca6ad74dc64b43649"},
- {"internal/host/movesland.go", "landingRun.transition", "record-writer", "key:SetRecordMove", "c943a81b6ddba9cba138710de340bc238689e0122d12953d1bb4c84e95735f8e"},
+ {"internal/host/movesland.go", "landingRun.transition", "record-writer", "key:SetRecordMove", "0b197834835dc9afad77f89358d2bcab9512128dcf43070a3765e430526dbfb1"},
{"internal/host/moveslandpark.go", "leasedStore.CopyVersion", "versioned-transport", "def:CopyVersion,sel:CopyVersion", "a8b33c9f61f59477b76fa50472250622c717ab2883f32837c601da5eeb497e9d"},
{"internal/host/moveslandpark.go", "leasedStore.PutIfAbsent", "versioned-transport", "def:PutIfAbsent,sel:PutIfAbsent", "33542589bf2a336b181f685bd0dc5faa0285449e5ca9e850e9f4e0e7e4164c6b"},
{"internal/host/movesmeasure.go", "measureS3", "nonrecord", "sel:Put", "3ae9fc093b453e19a5b19be0929533b30a0e2a3e998ba8dc766f41287db2caca"},
@@ -678,7 +678,7 @@ var plan9B8CensusInventory = []censusEntry{
{"internal/host/movesphase1.go", "movesWorker.sourceRound", "nonrecord", "sel:PutIfAbsentVersioned", "c5bc2e6cb57b819ae1da485d37093c9a9ed3cdc9e9e16e8a8e312fd5ad2e2fe3"},
{"internal/host/movesphase1.go", "movesWorker.refreshCrossRate", "nonrecord", "sel:CopyVersioned", "7e03e680bd61be26dddcf168cacd2427bb535571dfd326c4a3130ebe65ea1968"},
{"internal/host/movesphase1.go", "movesWorker.writeSourceReport", "nonrecord", "sel:PutReplace", "370407cca2c1b91f6e5e7f135238f04c0dce9d7986e6e1ee9bebbf88cad0b9e0"},
- {"internal/host/movesphase1.go", "movesWorker.decideReady", "record-writer", "key:SetRecordMove", "c228e6f38eca6ed10501538031d3260f016a3f26ce501d8544f1581ae7668d8f"},
+ {"internal/host/movesphase1.go", "movesWorker.decideReady", "record-writer", "key:SetRecordMove", "927e1d864c3232ef17dfa7e9329ee26e870db75da128a8101234610a762e6f9a"},
{"internal/host/movesreport.go", "movesWorker.targetReport", "nonrecord", "sel:PutReplace", "3f6aec125f1bd40271adf78833ac3268e536cbc5e8af3731b16f382669cd955b"},
{"internal/host/movesstore.go", "laneProbe", "nonrecord", "sel:PutIfAbsent", "f4a17a1143e7370b9039b2e66e736606015c46619bab92c553ce7b1ad6b7918b"},
{"internal/host/snapshot_loop.go", "registrySnapshotPass", "nonrecord", "sel:Put", "6b01b46b361c3a2d1d432659a0d234c01975849b9a8ac5571858fdd78d376849"},
@@ -803,6 +803,9 @@ var plan9B8CensusInventory = []censusEntry{
{"internal/moves/build.go", "putOrReuse", "nonrecord", "sel:PutIfAbsent", "932e5852c1d7c747db1013954352bbeb00364329d3515ae723a144b074567562"},
{"internal/moves/build.go", "putOrReadIntent", "nonrecord", "sel:PutIfAbsent", "d53112f8f9c0c48844f930d2b9ae5e58b75588ef31b0a3740b65a61c83fba613"},
{"internal/moves/build.go", "putOrReadManifest", "nonrecord", "sel:PutIfAbsent", "3187b04355f4ec75b86edf311a732c6cfecd722322b509f0731479f738b52a85"},
+ // Plan 6 Task 3: the move log, create-only objects under the report
+ // prefix; no record.
+ {"internal/moves/log.go", "WriteLog", "nonrecord", "sel:Put", "a3803e1ac4a43da86fb60d7fc5395ee2b0e9e8f86f33bf13bc251795dc3529cf"},
{"internal/moves/readiness.go", "PutReplace", "nonrecord", "def:PutReplace,sel:Put", "3ec4d72cb252b34a402175cb7866dee4ab4f5b2441958f9fc455f687ffb2eafd"},
{"internal/objstore/entitlement.go", "PutEntitlement", "nonrecord", "sel:Put", "35b3734ff8587089c913ba7151f17955e9b1f7456a2f5ae1b3fb0e6917104005"},
{"internal/objstore/memscanner.go", "MemScanner.Put", "nonrecord", "def:Put", "bef5afd48d145dafdcbda197b4ad221766833d1c8ba8223f81df98495c860a4b"},