amadan
moves-phase1-6 Task 3 fix round 2: durable log outboxes, assembled recovery attempts, bounded memo, grant Log conditions
- Box: a failed move log write is kept, with its bytes, in <DataDir>/.moves/.log-outbox; every pass retries it whatever the record says, deleting it only once created or found. Member lines are built from the entry alone (cut_budget_s joins the entry's fields). - Console: entries go to a move_log_outbox table and are written by the move's next pass through a fresh grant; the release waits for them (at most an hour) and writes its own entry just before it. Entries come from the write's own evidence (MoveRecoveryGrant.Transition). - Recovery probe starts log from the record their binding read, through movesWorker.probeAttempts, which wires RecoveryAttempt. - The dedupe memo is bounded (4096, oldest first) and retires a host's moves when a pass sees its record with no move or another one. - MoveRecoveryGrant.Log(name, body): key from the grant's own move, single-segment name equal to the identity, 4 KiB bound, create-only with same-bytes replay a no-op and ErrMoveLogTaken otherwise, validity through current(); authority and B8 census rows.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
16 files changed,
+1061
−247
approved_moves_copy_test.go+1 −0internal/console/claim_authority_census_test.go+3 −0internal/console/movesrecovery.go+121 −40internal/console/movesrecovery_test.go+131 −6internal/console/store.go+52 −0internal/console/store_gate.go+1 −0internal/console/testdata/claim-authority-sites.txt+2 −0internal/host/movesevents.go+140 −37internal/host/movesevents_test.go+222 −85internal/host/moveslogoutbox.go+169 −0internal/host/movespass.go+9 −3internal/host/movesprobeattempts.go+46 −27internal/hostops/move_recovery_grant.go+99 −21internal/hostops/move_recovery_grant_test.go+52 −23internal/moves/transition.go+8 −4internal/provision/plan9_b8_census_test.go+5 −1
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go| index cf515305c..4bae046d7 100644 |
| --- a/approved_moves_copy_test.go |
| +++ b/approved_moves_copy_test.go |
| @@ -593,6 +593,7 @@ var movesScanFiles = []string{ |
| "internal/host/movesfloorgate_linux.go", |
| "internal/host/movesfloorgate_other.go", |
| "internal/host/movesevents.go", |
| + "internal/host/moveslogoutbox.go", |
| "internal/host/movesabort.go", |
| "internal/host/movesarchive.go", |
| "internal/host/movescut.go", |
diff --git a/internal/console/claim_authority_census_test.go b/internal/console/claim_authority_census_test.go| index 455e34e1f..6a88a348c 100644 |
| --- a/internal/console/claim_authority_census_test.go |
| +++ b/internal/console/claim_authority_census_test.go |
| @@ -29,6 +29,9 @@ func claimAuthoritySites(t *testing.T) []string { |
| // The console's move pass (moves Plan 5 Task 7): every write of the |
| // move-recovery grant, and the lost target's stamp. |
| "ResolveStart": true, "Convert": true, "Claim": true, "Record": true, "StampLostMoveTargetParked": true, "stampLostMoveTarget": true, |
| + // The move log the pass writes through the grant (Plan 6 Task 3, fix |
| + // round 2 ruling 4): an entry of the grant's own move, create-only. |
| + "Log": true, |
| "BeginPurge": true, "SavePurgeJournal": true, "RemovePurgeJournal": true, |
| "AcquireMailGate": true, "ReleaseMailGate": true, "EnrollNewApp": true, |
| "EnrollBackfillApp": true, "OpenGated": true, "WithAppMutation": true, |
diff --git a/internal/console/movesrecovery.go b/internal/console/movesrecovery.go| index 080fa5570..c35b83150 100644 |
| --- a/internal/console/movesrecovery.go |
| +++ b/internal/console/movesrecovery.go |
| @@ -4,6 +4,7 @@ import ( |
| "context" |
| "crypto/sha256" |
| "encoding/hex" |
| + "encoding/json" |
| "errors" |
| "fmt" |
| "log" |
| @@ -282,6 +283,9 @@ func (s *Server) movesRecovery(ctx context.Context, g *hostops.MoveRecoveryGrant |
| return err |
| } |
| now := s.movesNow() |
| + // What earlier passes owe the move's log is written first, while the |
| + // grant is unspent (MoveRecoveryGrant.Log). |
| + pending, oldest := s.flushMoveLog(ctx, g, o) |
| in := moveRecoveryInput{Now: now, Claim: c, Rec: rec} |
| ours := rec.Move != nil && rec.Move.ID == o.OpID |
| if c.Op.Kind == hostclaim.KindMoving && c.Op.Phase == hostclaim.PhaseMoving && !ours { |
| @@ -327,19 +331,35 @@ func (s *Server) movesRecovery(ctx context.Context, g *hostops.MoveRecoveryGrant |
| return err |
| } |
| log.Printf("console: moves: %s: move %s is still preparing after %s; aborted", o.Host, o.OpID, moves.AbortAfter) |
| - s.logMoveWrite(ctx, g, o, rec) |
| + if t, ok := abortTransition(rec); ok { |
| + s.queueMoveLog(o, t, now) |
| + } |
| return nil |
| case moveActClaim: |
| + if d.Claim == hostops.RecoveryMoveRelease { |
| + // The release ends what the console can log for the move: its |
| + // owed entries go first, for at most moveLogReleaseWait, and the |
| + // release's own entry is written just before it. |
| + if pending > 0 && now.Sub(oldest) < moveLogReleaseWait { |
| + log.Printf("console: moves: %s: the release of move %s waits for %d unwritten log entries", o.Host, o.OpID, pending) |
| + return nil |
| + } |
| + s.logMoveNow(ctx, g, o, claimTransition(o, rec, "claim_released", "claim-released"), now) |
| + } |
| if err := g.Claim(ctx, d.Claim); err != nil { |
| return err |
| } |
| - s.logMoveClaim(ctx, g, o, rec, d.Claim) |
| + if d.Claim == hostops.RecoveryMoveRollback { |
| + s.queueMoveLog(o, claimTransition(o, rec, "claim_rollback", "claim-rollback"), now) |
| + } |
| return nil |
| case moveActRecord: |
| if err := g.Record(ctx, d.Record); err != nil { |
| return err |
| } |
| - s.logMoveWrite(ctx, g, o, rec) |
| + if t, ok := g.Transition(); ok { |
| + s.queueMoveLog(o, t, now) |
| + } |
| return nil |
| case moveActStampLost: |
| if s.stampLostMoveTarget == nil { |
| @@ -354,7 +374,9 @@ func (s *Server) movesRecovery(ctx context.Context, g *hostops.MoveRecoveryGrant |
| return err |
| } |
| log.Printf("console: moves: %s: move %s closed by the console (%s); the release follows", o.Host, o.OpID, d.Record) |
| - s.logMoveWrite(ctx, g, o, rec) |
| + if t, ok := g.Transition(); ok { |
| + s.queueMoveLog(o, t, now) |
| + } |
| s.pageMove(ctx, movePageCloseOverdue, o, c, rec, now) |
| return nil |
| case moveActAbortOverdue: |
| @@ -525,51 +547,110 @@ func moveAbortOverduePage(host, id, to, box string) (subject, body string) { |
| return subject, body |
| } |
| -// The move log of the console's own move writes (Plan 6 Task 3, fix round 1, |
| -// ruling 3): the six-hour abort, the landing deadline's rollback, the closes, |
| -// and the claim's release and rollback. The console has no member sink; the |
| -// durable log under the move's report prefix is what counts, written through |
| -// the grant (MoveRecoveryGrant.Log, moves.WriteLog), which names no store. |
| -// Each entry is named as a box's write of the same transition would be |
| -// (moves.TransitionOf) and written create-only, so a box that also logs it |
| -// changes nothing. A failed write is logged and not retried: the pass that |
| -// made the write does not see it again. |
| - |
| -// logMoveWrite logs the record write this pass just made: rec as the pass |
| -// read it, against the record now. |
| -func (s *Server) logMoveWrite(ctx context.Context, g *hostops.MoveRecoveryGrant, o hostops.ConsoleMoveObservation, rec provision.Record) { |
| - after, err := provision.GetRecord(ctx, s.cfg.Releases, o.Account, o.App, o.Host) |
| +// The move log of the console's own move writes (Plan 6 Task 3; fix round 2, |
| +// rulings 1 and 4): the closes, the landing deadline's rollback, the six-hour |
| +// abort, and the claim's rollback and release. Each entry is named as a box's |
| +// write of the same transition would be (moves.TransitionOf), from the |
| +// write's own evidence (the grant's Transition, or the record the pass |
| +// acted on), and carries the console as its actor; the console has no member |
| +// sink. |
| +// |
| +// The grant writes only while unspent, so an entry a pass's write makes is |
| +// queued in the console's outbox (move_log_outbox, with its exact bytes) and |
| +// written by the move's next pass before it acts. An entry stays queued until |
| +// it is written or found written, whatever the move does meanwhile. The |
| +// release is the last pass with a grant for the move, so it waits for the |
| +// move's queued entries, for at most moveLogReleaseWait (a log that cannot be |
| +// written must not hold a claim forever), and writes its own entry just |
| +// before it releases. Entries still queued after the release stay in the |
| +// outbox, logged; no later grant can write them. A release of a start never |
| +// applied is not logged: the record never named that move. |
| + |
| +// moveLogReleaseWait bounds how long the claim's release waits for the |
| +// move's unwritten log entries. |
| +const moveLogReleaseWait = time.Hour |
| + |
| +// flushMoveLog writes the move's queued entries through the grant: the count |
| +// still unwritten, and when the oldest of them was queued. |
| +func (s *Server) flushMoveLog(ctx context.Context, g *hostops.MoveRecoveryGrant, o hostops.ConsoleMoveObservation) (int, time.Time) { |
| + queued, err := s.store.MoveLogPending(o.Host, o.OpID) |
| if err != nil { |
| - log.Printf("console: moves: %s: the log of move %s: the record: %v", o.Host, o.OpID, err) |
| - return |
| + log.Printf("console: moves: %s: the log outbox of move %s: %v", o.Host, o.OpID, err) |
| + return 0, time.Time{} |
| } |
| - if e, ok := moves.TransitionOf(rec, *after); ok { |
| - s.writeMoveLog(ctx, g, o, e) |
| + n, oldest := 0, time.Time{} |
| + for _, e := range queued { |
| + _, err := g.Log(ctx, e.Name, e.Body) |
| + switch { |
| + case err == nil: |
| + case errors.Is(err, hostops.ErrMoveLogTaken): |
| + log.Printf("console: moves: %s: the log of move %s at %s was written by another writer first", o.Host, o.OpID, e.Name) |
| + case errors.Is(err, hostops.ErrIdentity): |
| + log.Printf("console: moves: %s: the log of move %s at %s is refused, and dropped: %v", o.Host, o.OpID, e.Name, err) |
| + default: |
| + log.Printf("console: moves: %s: the log of move %s at %s waits for the next pass: %v", o.Host, o.OpID, e.Name, err) |
| + if n == 0 || e.QueuedAt.Before(oldest) { |
| + oldest = e.QueuedAt |
| + } |
| + n++ |
| + continue |
| + } |
| + if err := s.store.SettleMoveLog(o.Host, o.OpID, e.Name); err != nil { |
| + log.Printf("console: moves: %s: the log outbox of move %s: %v", o.Host, o.OpID, err) |
| + } |
| } |
| + return n, oldest |
| } |
| -// logMoveClaim logs a claim write that ends or turns the move: the release |
| -// after a close (or of a start never applied) and the claim's rollback. |
| -func (s *Server) logMoveClaim(ctx context.Context, g *hostops.MoveRecoveryGrant, o hostops.ConsoleMoveObservation, rec provision.Record, action hostops.MoveRecoveryClaimAction) { |
| - var e moves.Transition |
| - switch action { |
| - case hostops.RecoveryMoveRelease, hostops.RecoveryMoveReleaseNotApplied: |
| - e = moves.Transition{Op: o.OpID, Event: "claim_released", Identity: "claim-released"} |
| - case hostops.RecoveryMoveRollback: |
| - e = moves.Transition{Op: o.OpID, Event: "claim_rollback", Identity: "claim-rollback"} |
| - default: |
| +// moveLogEntry is the console's entry for t. |
| +func moveLogEntry(o hostops.ConsoleMoveObservation, t moves.Transition, now time.Time) ([]byte, error) { |
| + return json.Marshal(moves.LogEvent{V: 1, Op: t.Op, Host: o.Host, Event: t.Event, Identity: t.Identity, Actor: "console", At: now.UTC(), Fields: t.Fields}) |
| +} |
| + |
| +// queueMoveLog queues the entry for t, which the move's next pass writes. |
| +func (s *Server) queueMoveLog(o hostops.ConsoleMoveObservation, t moves.Transition, now time.Time) { |
| + body, err := moveLogEntry(o, t, now) |
| + if err == nil { |
| + err = s.store.QueueMoveLog(o.Host, o.OpID, t.Identity, body, now) |
| + } |
| + if err != nil { |
| + log.Printf("console: moves: %s: the log of move %s at %s is lost: %v", o.Host, o.OpID, t.Identity, err) |
| + } |
| +} |
| + |
| +// logMoveNow writes the entry for t through the grant, or queues it. |
| +func (s *Server) logMoveNow(ctx context.Context, g *hostops.MoveRecoveryGrant, o hostops.ConsoleMoveObservation, t moves.Transition, now time.Time) { |
| + body, err := moveLogEntry(o, t, now) |
| + if err != nil { |
| + log.Printf("console: moves: %s: the log of move %s at %s is lost: %v", o.Host, o.OpID, t.Identity, err) |
| return |
| } |
| - if m := rec.Move; m != nil { |
| - e.Fields = map[string]string{"from": m.From, "to": m.To} |
| + if _, err := g.Log(ctx, t.Identity, body); err != nil && !errors.Is(err, hostops.ErrMoveLogTaken) { |
| + log.Printf("console: moves: %s: the log of move %s at %s is queued: %v", o.Host, o.OpID, t.Identity, err) |
| + if err := s.store.QueueMoveLog(o.Host, o.OpID, t.Identity, body, now); err != nil { |
| + log.Printf("console: moves: %s: the log of move %s at %s is lost: %v", o.Host, o.OpID, t.Identity, err) |
| + } |
| } |
| - s.writeMoveLog(ctx, g, o, e) |
| } |
| -// writeMoveLog writes one console entry. |
| -func (s *Server) writeMoveLog(ctx context.Context, g *hostops.MoveRecoveryGrant, o hostops.ConsoleMoveObservation, e moves.Transition) { |
| - ev := moves.LogEvent{V: 1, Op: e.Op, Host: o.Host, Event: e.Event, Identity: e.Identity, Actor: "console", At: s.movesNow().UTC(), Fields: e.Fields} |
| - if _, err := g.Log(ctx, ev); err != nil { |
| - log.Printf("console: moves: %s: the log of move %s at %s: %v", o.Host, o.OpID, e.Identity, err) |
| +// abortTransition is the six-hour abort's entry: the record the pass acted |
| +// on, as the abort leaves it (aborting). |
| +func abortTransition(rec provision.Record) (moves.Transition, bool) { |
| + if rec.Move == nil { |
| + return moves.Transition{}, false |
| + } |
| + after, m := rec, *rec.Move |
| + m.State = provision.MoveAborting |
| + after.Move = &m |
| + return moves.TransitionOf(rec, after) |
| +} |
| + |
| +// claimTransition is the entry for a claim write of the console's: the move's |
| +// regions from the record the pass acted on. |
| +func claimTransition(o hostops.ConsoleMoveObservation, rec provision.Record, event, identity string) moves.Transition { |
| + t := moves.Transition{Op: o.OpID, Event: event, Identity: identity} |
| + if m := rec.Move; m != nil { |
| + t.Fields = map[string]string{"from": m.From, "to": m.To} |
| } |
| + return t |
| } |
diff --git a/internal/console/movesrecovery_test.go b/internal/console/movesrecovery_test.go| index 3148590e4..99a2dd29b 100644 |
| --- a/internal/console/movesrecovery_test.go |
| +++ b/internal/console/movesrecovery_test.go |
| @@ -4,6 +4,8 @@ import ( |
| "context" |
| "crypto/rand" |
| "encoding/json" |
| + "os" |
| + "path/filepath" |
| "strings" |
| "sync" |
| "testing" |
| @@ -734,10 +736,14 @@ func moveLogOf(t *testing.T, w *movePassWorld) map[string]moves.LogEvent { |
| return out |
| } |
| -// Plan 6 Task 3 fix round 1, ruling 3: the console's move pass writes the |
| -// durable move log for the transitions it makes: the close and the claim's |
| -// release after it, the landing deadline's rollback, and the six-hour abort. |
| -// Each entry is the console's, under the identity the box would give it. |
| +// Plan 6 Task 3 fix round 1, ruling 3, and fix round 2, ruling 1: the |
| +// console's move pass logs the transitions it makes: the close and the |
| +// claim's release after it, the landing deadline's rollback, and the six-hour |
| +// abort. Each entry is the console's, under the identity the box would give |
| +// it. The grant logs only while it is unspent, so an entry made by a pass's |
| +// write is kept in the console's outbox and written by the move's next pass, |
| +// before that pass acts; the release's own entry is written just before the |
| +// release. |
| func TestRecoveryPassWritesTheMoveLog(t *testing.T) { |
| t.Run("close and release", func(t *testing.T) { |
| w := landingMovePass(t, "code-mvpass-log-close") |
| @@ -745,19 +751,24 @@ func TestRecoveryPassWritesTheMoveLog(t *testing.T) { |
| w.pass(t) // the close's grace runs from the pass's first sighting |
| w.advance(moveCloseGrace + time.Minute) |
| w.pass(t) // the close |
| - w.pass(t) // the claim's release |
| + w.pass(t) // the close's entry, then the claim's release |
| log := moveLogOf(t, w) |
| done, released := log["done"], log["claim-released"] |
| if done.Event != "done" || done.Actor != "console" || done.Op != w.id || released.Event != "claim_released" || released.Actor != "console" { |
| t.Fatalf("log %+v", log) |
| } |
| + if done.Fields["offline_s"] == "" || done.Fields["from"] != "ie" { |
| + t.Fatalf("the close's entry lacks its evidence: %+v", done.Fields) |
| + } |
| + noPendingLog(t, w) |
| }) |
| t.Run("deadline rollback", func(t *testing.T) { |
| w := landingMovePass(t, "code-mvpass-log-deadline") |
| past := time.Now().UTC().Add(-time.Minute) |
| j17SetRecord(t, w.fx, func(r *provision.Record) { r.Move.LandingDeadline = &past }) |
| w.pass(t) |
| - if ev := moveLogOf(t, w)["rolling_back"]; ev.Event != "rolling_back" || ev.Actor != "console" { |
| + w.pass(t) |
| + if ev := moveLogOf(t, w)["rolling_back"]; ev.Event != "rolling_back" || ev.Actor != "console" || ev.Fields["reason"] == "" { |
| t.Fatalf("log %+v", moveLogOf(t, w)) |
| } |
| }) |
| @@ -768,8 +779,122 @@ func TestRecoveryPassWritesTheMoveLog(t *testing.T) { |
| w.clock = requested.Add(moves.AbortAfter + time.Minute) |
| w.mu.Unlock() |
| w.pass(t) |
| + w.pass(t) |
| if ev := moveLogOf(t, w)["aborting"]; ev.Event != "aborting" || ev.Actor != "console" { |
| t.Fatalf("log %+v", moveLogOf(t, w)) |
| } |
| }) |
| } |
| + |
| +// blockMoveLog makes every write under the move's log fail (a file where |
| +// its directory goes) until the returned func is called. |
| +func blockMoveLog(t *testing.T, w *movePassWorld) func() { |
| + t.Helper() |
| + root := w.fx.rel.(*deletableStore).root |
| + p := filepath.Join(root, filepath.FromSlash(moves.ReportPrefix("bab", "titogo", "acme.go.tito.io", w.id)+"log")) |
| + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(p, nil, 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + return func() { |
| + if err := os.Remove(p); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| +} |
| + |
| +func pendingLog(t *testing.T, w *movePassWorld) []string { |
| + t.Helper() |
| + got, err := w.fx.st.MoveLogPending("acme.go.tito.io", w.id) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var names []string |
| + for _, e := range got { |
| + names = append(names, e.Name) |
| + } |
| + return names |
| +} |
| + |
| +func noPendingLog(t *testing.T, w *movePassWorld) { |
| + t.Helper() |
| + if p := pendingLog(t, w); len(p) != 0 { |
| + t.Fatalf("entries still pending: %v", p) |
| + } |
| +} |
| + |
| +// Fix round 2, ruling 1: a console entry whose write fails is kept, with its |
| +// bytes, in the console's outbox, and written by a later pass after the move |
| +// has moved on. The claim's release waits for the move's pending entries |
| +// (at most moveLogReleaseWait), so the close's entry is not stranded by it. |
| +func TestRecoveryPassLogOutboxRetries(t *testing.T) { |
| + t.Run("the move closes, then the retry", func(t *testing.T) { |
| + w := landingMovePass(t, "code-mvpass-outbox-close") |
| + committedAndOpened(t, w) |
| + w.pass(t) |
| + w.advance(moveCloseGrace + time.Minute) |
| + unblock := blockMoveLog(t, w) |
| + w.pass(t) // the close; its entry is pending |
| + if rec := w.record(t); rec.Move != nil { |
| + t.Fatalf("not closed: %+v", rec.Move) |
| + } |
| + w.pass(t) // the write fails again: the release waits |
| + if p := pendingLog(t, w); strings.Join(p, " ") != "done" { |
| + t.Fatalf("pending %v, want the close's entry", p) |
| + } |
| + if c := w.claim(t); c.Op.Kind != hostclaim.KindLanding { |
| + t.Fatalf("released with the close's entry unwritten: %+v", c.Op) |
| + } |
| + unblock() |
| + w.pass(t) |
| + log := moveLogOf(t, w) |
| + if log["done"].Actor != "console" || log["claim-released"].Actor != "console" { |
| + t.Fatalf("log %+v", log) |
| + } |
| + if c := w.claim(t); c.Op.Kind != hostclaim.KindServing { |
| + t.Fatalf("not released once the entry was written: %+v", c.Op) |
| + } |
| + noPendingLog(t, w) |
| + }) |
| + t.Run("the phase advances, then the retry", func(t *testing.T) { |
| + w := landingMovePass(t, "code-mvpass-outbox-deadline") |
| + past := time.Now().UTC().Add(-time.Minute) |
| + j17SetRecord(t, w.fx, func(r *provision.Record) { r.Move.LandingDeadline = &past }) |
| + unblock := blockMoveLog(t, w) |
| + w.pass(t) // the record's rollback; its entry is pending |
| + w.pass(t) // the claim's rollback; the write fails again |
| + if p := pendingLog(t, w); len(p) != 2 || p[0] != "rolling_back" { |
| + t.Fatalf("pending %v, want the rollback's entries", p) |
| + } |
| + j17SetRecord(t, w.fx, func(r *provision.Record) { r.Move.Rollback.TargetPhase = provision.RbTargetReleasing }) |
| + unblock() |
| + w.pass(t) |
| + log := moveLogOf(t, w) |
| + if log["rolling_back"].Actor != "console" || log["claim-rollback"].Actor != "console" { |
| + t.Fatalf("log %+v", log) |
| + } |
| + noPendingLog(t, w) |
| + }) |
| + t.Run("the release does not wait past its bound", func(t *testing.T) { |
| + w := landingMovePass(t, "code-mvpass-outbox-bound") |
| + committedAndOpened(t, w) |
| + w.pass(t) |
| + w.advance(moveCloseGrace + time.Minute) |
| + blockMoveLog(t, w) |
| + w.pass(t) // the close |
| + w.pass(t) |
| + if c := w.claim(t); c.Op.Kind != hostclaim.KindLanding { |
| + t.Fatalf("released early: %+v", c.Op) |
| + } |
| + w.advance(moveLogReleaseWait + time.Minute) |
| + w.pass(t) |
| + if c := w.claim(t); c.Op.Kind != hostclaim.KindServing { |
| + t.Fatalf("the release waited past its bound: %+v", c.Op) |
| + } |
| + if p := pendingLog(t, w); len(p) == 0 { |
| + t.Fatal("the unwritten entries were dropped") |
| + } |
| + }) |
| +} |
diff --git a/internal/console/store.go b/internal/console/store.go| index 51d0619d6..7e848aada 100644 |
| --- a/internal/console/store.go |
| +++ b/internal/console/store.go |
| @@ -358,6 +358,17 @@ func (s *Store) migrate() error { |
| `CREATE TABLE IF NOT EXISTS move_pages ( |
| rule TEXT PRIMARY KEY, |
| attempted_at INTEGER NOT NULL |
| +);`, |
| + // The move pass's outbox (movesrecovery.go, Plan 6 Task 3): move log |
| + // entries the pass owes and has not yet written, with their exact |
| + // bytes, until a pass's grant writes them or finds them written. |
| + `CREATE TABLE IF NOT EXISTS move_log_outbox ( |
| + host TEXT NOT NULL, |
| + op TEXT NOT NULL, |
| + name TEXT NOT NULL, |
| + body BLOB NOT NULL, |
| + queued_at INTEGER NOT NULL, |
| + PRIMARY KEY (host, op, name) |
| );`, |
| } { |
| if _, err := s.db.Exec(stmt); err != nil { |
| @@ -2213,6 +2224,47 @@ ON CONFLICT(rule) DO UPDATE SET attempted_at = excluded.attempted_at`, rule, at. |
| return err |
| } |
| +// MoveLogPendingEntry is one move log entry the move pass owes. |
| +type MoveLogPendingEntry struct { |
| + Name string |
| + Body []byte |
| + QueuedAt time.Time |
| +} |
| + |
| +// QueueMoveLog keeps an entry the move pass could not write yet. An entry |
| +// already queued under the name is kept as it was. |
| +func (s *Store) QueueMoveLog(host, op, name string, body []byte, at time.Time) error { |
| + _, err := s.db.Exec(`INSERT INTO move_log_outbox (host, op, name, body, queued_at) VALUES (?, ?, ?, ?, ?) |
| +ON CONFLICT(host, op, name) DO NOTHING`, host, op, name, body, at.Unix()) |
| + return err |
| +} |
| + |
| +// MoveLogPending is the move's queued entries, oldest first. |
| +func (s *Store) MoveLogPending(host, op string) ([]MoveLogPendingEntry, error) { |
| + rows, err := s.db.Query(`SELECT name, body, queued_at FROM move_log_outbox WHERE host = ? AND op = ? ORDER BY rowid`, host, op) |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer rows.Close() |
| + var out []MoveLogPendingEntry |
| + for rows.Next() { |
| + var e MoveLogPendingEntry |
| + var at int64 |
| + if err := rows.Scan(&e.Name, &e.Body, &at); err != nil { |
| + return nil, err |
| + } |
| + e.QueuedAt = time.Unix(at, 0).UTC() |
| + out = append(out, e) |
| + } |
| + return out, rows.Err() |
| +} |
| + |
| +// SettleMoveLog drops an entry once it is written, or found written. |
| +func (s *Store) SettleMoveLog(host, op, name string) error { |
| + _, err := s.db.Exec(`DELETE FROM move_log_outbox WHERE host = ? AND op = ? AND name = ?`, host, op, name) |
| + return err |
| +} |
| + |
| // ---- grants (the console's Keymail access tokens) --------------------------- |
| // SetGrantToken stores the Keymail access token the user's OAuth approval |
diff --git a/internal/console/store_gate.go b/internal/console/store_gate.go| index c2d72d223..f6160901b 100644 |
| --- a/internal/console/store_gate.go |
| +++ b/internal/console/store_gate.go |
| @@ -210,6 +210,7 @@ var claimSQLTables = map[string]claimSQLTable{ |
| "invites": {NonApp: "account invites, keyed by account and address; names no app"}, |
| "alert_seen": {NonApp: "operator alert memory, keyed by box, rule and address"}, |
| "move_pages": {NonApp: "the move pass's last page attempt per finding (rule names the host and the move); no app column"}, |
| + "move_log_outbox": {NonApp: "the move pass's unwritten move log entries, keyed by host, move and entry; the grant that writes one derives its key; no app column"}, |
| "edge_proof_seen": {NonApp: "single-row edge proof marker"}, |
| "cli_auth_requests": {NonApp: "device-code logins in flight, keyed by code hash"}, |
| "step_ups": {NonApp: "passkey step-up marks, keyed by session hash"}, |
diff --git a/internal/console/testdata/claim-authority-sites.txt b/internal/console/testdata/claim-authority-sites.txt| index 8e7d07046..e57d978f5 100644 |
| --- a/internal/console/testdata/claim-authority-sites.txt |
| +++ b/internal/console/testdata/claim-authority-sites.txt |
| @@ -106,6 +106,8 @@ geoippublish.go|Commit|os.Open(st.path) |
| hostclaimbackfill.go|RunClaimRecoveryPass|a.Control.RunRecovery(ctx, s.runClaimRecoveryPassBody) |
| hostwakereset.go|registerWakeRetryReset|a.Control.RegisterStart(ctx, h.Account, h.App, reg.AppID, id, reg, registeredAt) |
| hostwakereset.go|registerWakeRetryReset|start.FinishWakeRetryReset(cleanup) |
| +movesrecovery.go|flushMoveLog|g.Log(ctx, e.Name, e.Body) |
| +movesrecovery.go|logMoveNow|g.Log(ctx, t.Identity, body) |
| movesrecovery.go|movesRecovery|a.Control.ConsoleMove(ctx, co, hostops.ConsoleMoveAbort) |
| movesrecovery.go|movesRecovery|g.Claim(ctx, d.Claim) |
| movesrecovery.go|movesRecovery|g.Convert(ctx, *in.Fence) |
diff --git a/internal/host/movesevents.go b/internal/host/movesevents.go| index 619b61736..9898423b3 100644 |
| --- a/internal/host/movesevents.go |
| +++ b/internal/host/movesevents.go |
| @@ -1,12 +1,13 @@ |
| package host |
| import ( |
| + "container/list" |
| "context" |
| "fmt" |
| "log" |
| "math" |
| "strconv" |
| - "strings" |
| + "sync" |
| "github.com/carlosframework/platform/internal/activator" |
| "github.com/carlosframework/platform/internal/hostclaim" |
| @@ -29,11 +30,18 @@ import ( |
| // only the writer that created the object tells the app's members (preflight |
| // L9, L27). |
| // |
| -// A worker remembers the identities whose objects exist (created, or found |
| -// there), so a pass that sees the same transition again does not even try; |
| -// a write that failed is not remembered, so the next sighting tries again. |
| -// A move's identities are forgotten once its terminal outcome is logged: |
| -// create-only storage still refuses a replay. |
| +// A write that fails is kept, with the entry's exact bytes, in the worker's |
| +// outbox on disk (moveslogoutbox.go), and every pass retries the outbox |
| +// whatever the record now says, until the entry is created or found there; |
| +// so neither a phase that moves on nor a close loses it (fix round 2, |
| +// ruling 1). |
| +// |
| +// A worker remembers the identities it has handed off (written, found, or |
| +// kept in the outbox), so a pass that sees the same transition again does |
| +// not even try. The memory is bounded (moveLogMemoCap, oldest dropped), and |
| +// a move is forgotten once its terminal outcome is logged or a pass sees its |
| +// record with no move or another move (retireMoves): create-only storage |
| +// still refuses a replay. |
| // observedMove wraps a record mutation so *pre keeps the record as the write |
| // found it: the transition's before, for the log. |
| @@ -81,6 +89,7 @@ func (w *movesWorker) logTransition(ctx context.Context, s release.Store, before |
| // 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) { |
| + w.retireMoves(rec) |
| m := rec.Move |
| if m == nil || m.ID == "" { |
| return |
| @@ -118,9 +127,10 @@ func (w *movesWorker) logRevPhase(ctx context.Context, s release.Store, rec prov |
| } |
| // recoveryAttemptLogger is the attempt journal's RecoveryAttempt hook for |
| -// this worker: each recovery probe start the journal counts is |
| -// probe_attempt-<epoch>-<n> (preflight L11), the starts of one wake |
| -// included. s is the store the log is written to. |
| +// this worker (set by probeAttempts): each recovery probe start the journal |
| +// counts is probe_attempt-<epoch>-<n> (preflight L11), the starts of one wake |
| +// included. s is the store the log is written to. A write that fails goes to |
| +// the outbox like any other. |
| func (w *movesWorker) recoveryAttemptLogger(s release.Store) func(ctx context.Context, rec provision.Record, epoch uint64, n int) { |
| return func(ctx context.Context, rec provision.Record, epoch uint64, n int) { |
| if rec.Move == nil || rec.Move.ID == "" { |
| @@ -135,13 +145,13 @@ func (w *movesWorker) recoveryAttemptLogger(s release.Store) func(ctx context.Co |
| // terminalEvents are the events that end a move's log. |
| var terminalEvents = map[string]bool{"done": true, "rolled_back": true, "aborted": true} |
| -// record hands one entry to the LogEvent hook unless this worker already |
| -// knows its object exists. Only a hook that succeeded (the object created, |
| -// or found there) is remembered. With no hook (a test's worker) the event is |
| -// a log line. |
| +// record hands one entry to the LogEvent hook unless this worker has |
| +// already handed it off. An entry whose write fails is kept in the outbox, |
| +// for every later pass to retry; only an entry written, found there or kept |
| +// is remembered. 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 moves.Transition) { |
| key := rec.Host + "/" + e.Op + "/" + e.Identity |
| - if _, seen := w.logged.Load(key); seen { |
| + if w.logged.seen(key) { |
| 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} |
| @@ -150,28 +160,123 @@ func (w *movesWorker) record(ctx context.Context, s release.Store, rec provision |
| } |
| if f := w.cfg.LogEvent; f != nil { |
| if err := f(ctx, s, rec, ev); err != nil { |
| - log.Printf("moves: %s: the log of %s at %s waits for the next sighting: %v", rec.Host, e.Op, e.Identity, err) |
| - return |
| + if qerr := w.queueLog(rec, ev); qerr != nil { |
| + log.Printf("moves: %s: the log of %s at %s waits for the next sighting: %v; the outbox: %v", rec.Host, e.Op, e.Identity, err, qerr) |
| + return |
| + } |
| + log.Printf("moves: %s: the log of %s at %s is kept for the next pass: %v", rec.Host, e.Op, e.Identity, err) |
| } |
| } else { |
| log.Printf("moves: %s: %s %s (%s)", rec.Host, e.Op, e.Event, e.Identity) |
| } |
| if terminalEvents[e.Event] { |
| - w.forgetMove(rec.Host, e.Op) |
| + w.logged.forget(rec.Host, e.Op) |
| return |
| } |
| - w.logged.Store(key, true) |
| + w.logged.add(rec.Host, e.Op, key) |
| +} |
| + |
| +// retireMoves forgets the moves of rec's host other than the one its record |
| +// names now: a move closed, or replaced, by another writer. |
| +func (w *movesWorker) retireMoves(rec provision.Record) { |
| + keep := "" |
| + if rec.Move != nil { |
| + keep = rec.Move.ID |
| + } |
| + w.logged.retire(rec.Host, keep) |
| } |
| -// forgetMove drops a closed move's remembered identities. |
| -func (w *movesWorker) forgetMove(host, op string) { |
| - prefix := host + "/" + op + "/" |
| - w.logged.Range(func(k, _ any) bool { |
| - if strings.HasPrefix(k.(string), prefix) { |
| - w.logged.Delete(k) |
| +// moveLogMemoCap bounds a worker's memory of handed-off identities. |
| +const moveLogMemoCap = 4096 |
| + |
| +// moveLogMemo is the identities a worker has handed off: at most |
| +// moveLogMemoCap, the oldest dropped first, indexed by host and move so a |
| +// move is forgotten whole. The zero value is ready. |
| +type moveLogMemo struct { |
| + mu sync.Mutex |
| + order *list.List // memoEntry, oldest first |
| + at map[string]*list.Element // key to its element |
| + moves map[string]map[string][]*list.Element // host to move to its elements |
| +} |
| + |
| +type memoEntry struct{ host, op, key string } |
| + |
| +func (m *moveLogMemo) seen(key string) bool { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + _, ok := m.at[key] |
| + return ok |
| +} |
| + |
| +func (m *moveLogMemo) len() int { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + return len(m.at) |
| +} |
| + |
| +func (m *moveLogMemo) add(host, op, key string) { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + if m.order == nil { |
| + m.order, m.at, m.moves = list.New(), map[string]*list.Element{}, map[string]map[string][]*list.Element{} |
| + } |
| + if _, ok := m.at[key]; ok { |
| + return |
| + } |
| + for len(m.at) >= moveLogMemoCap { |
| + m.drop(m.order.Front()) |
| + } |
| + el := m.order.PushBack(memoEntry{host, op, key}) |
| + m.at[key] = el |
| + if m.moves[host] == nil { |
| + m.moves[host] = map[string][]*list.Element{} |
| + } |
| + m.moves[host][op] = append(m.moves[host][op], el) |
| +} |
| + |
| +// drop removes one element from every index. |
| +func (m *moveLogMemo) drop(el *list.Element) { |
| + e := el.Value.(memoEntry) |
| + m.order.Remove(el) |
| + delete(m.at, e.key) |
| + els := m.moves[e.host][e.op] |
| + for i, x := range els { |
| + if x == el { |
| + els = append(els[:i], els[i+1:]...) |
| + break |
| + } |
| + } |
| + if len(els) > 0 { |
| + m.moves[e.host][e.op] = els |
| + return |
| + } |
| + delete(m.moves[e.host], e.op) |
| + if len(m.moves[e.host]) == 0 { |
| + delete(m.moves, e.host) |
| + } |
| +} |
| + |
| +// forget drops one move's identities. |
| +func (m *moveLogMemo) forget(host, op string) { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + for _, el := range append([]*list.Element(nil), m.moves[host][op]...) { |
| + m.drop(el) |
| + } |
| +} |
| + |
| +// retire drops the identities of host's moves other than keep. |
| +func (m *moveLogMemo) retire(host, keep string) { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + for op, els := range m.moves[host] { |
| + if op == keep { |
| + continue |
| + } |
| + for _, el := range append([]*list.Element(nil), els...) { |
| + m.drop(el) |
| } |
| - return true |
| - }) |
| + } |
| } |
| // moveLogWriter is the production LogEvent: the object, create-only, under |
| @@ -190,7 +295,7 @@ func moveLogWriter(sink activator.EventSink) func(context.Context, release.Store |
| if !created || sink == nil { |
| return nil |
| } |
| - if line, ok := memberLine(ev, rec); ok { |
| + if line, ok := memberLine(ev, rec.Host); ok { |
| sink.Event(rec.Account, rec.App, rec.Host, memberLevel(ev.Event), line) |
| } |
| return nil |
| @@ -200,20 +305,21 @@ func moveLogWriter(sink activator.EventSink) func(context.Context, release.Store |
| // 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 MV-N380, a line with no |
| // operator command in it), from the shared templates (moves/operatorcopy.go, |
| -// L12). A region the record names as the home's "" is moves.HomeRegionText. |
| -// ok is false for an event members are not told of. Its level is |
| -// memberLevel's. |
| -func memberLine(ev moves.LogEvent, rec provision.Record) (line string, ok bool) { |
| - host := rec.Host |
| +// L12), built from the entry alone, so an entry the outbox writes later says |
| +// the same. A region the record names as the home's "" is |
| +// moves.HomeRegionText. ok is false for an event members are not told of. |
| +// Its level is memberLevel's. |
| +func memberLine(ev moves.LogEvent, host string) (line string, ok bool) { |
| from, to := moves.RegionText(ev.Fields["from"]), moves.RegionText(ev.Fields["to"]) |
| switch ev.Event { |
| case "requested": |
| return moves.MoveStartingLine(host, from, to), true |
| case "admitted": |
| - if rec.Move == nil || rec.Move.CutBudgetS <= 0 { |
| + budget, err := strconv.Atoi(ev.Fields["cut_budget_s"]) |
| + if err != nil || budget <= 0 { |
| return "", false |
| } |
| - return moves.CutStartingLine(to, host, moves.CutBudgetText(rec.Move.CutBudgetS)), true |
| + return moves.CutStartingLine(to, host, moves.CutBudgetText(budget)), true |
| case "flipped": |
| return moves.FlippedLine(from, to), true |
| case "done": |
| @@ -224,9 +330,6 @@ func memberLine(ev moves.LogEvent, rec provision.Record) (line string, ok bool) |
| return moves.MovedLine(host, to, int(math.Round(offline))), true |
| case "rolling_back": |
| reason := ev.Fields["reason"] |
| - if reason == "" && rec.Move != nil && rec.Move.Rollback != nil { |
| - reason = rec.Move.Rollback.Reason |
| - } |
| if reason == "" { |
| return "", false |
| } |
diff --git a/internal/host/movesevents_test.go b/internal/host/movesevents_test.go| index 631d30b10..549cb2663 100644 |
| --- a/internal/host/movesevents_test.go |
| +++ b/internal/host/movesevents_test.go |
| @@ -2,8 +2,13 @@ package host |
| import ( |
| "context" |
| + "encoding/json" |
| "errors" |
| - "fmt" |
| + "io/fs" |
| + "os" |
| + "path/filepath" |
| + "sort" |
| + "strconv" |
| "strings" |
| "testing" |
| "time" |
| @@ -266,7 +271,20 @@ func TestMemberLines(t *testing.T) { |
| {"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) |
| + // The entry's own fields, as the writer names the transition (the |
| + // admitted cut's budget, the rollback's reason): the line needs |
| + // nothing else, so an entry the outbox writes later says the same. |
| + ev := moves.LogEvent{Event: c.event, Fields: map[string]string{"from": "ie", "to": "uk"}} |
| + if before := (provision.Record{Host: c.rec.Host}); c.rec.Move != nil { |
| + before.Move = &provision.MoveState{ID: c.rec.Move.ID, From: "ie", To: "uk", State: provision.MoveLanding} |
| + if c.event == "admitted" { |
| + before = flow[2] |
| + } |
| + if tr, ok := moves.TransitionOf(before, c.rec); ok && tr.Event == c.event { |
| + ev.Fields = tr.Fields |
| + } |
| + } |
| + line, ok := memberLine(ev, c.rec.Host) |
| level := "" |
| if ok { |
| level = memberLevel(c.event) |
| @@ -279,47 +297,6 @@ func TestMemberLines(t *testing.T) { |
| // ---- Task 3 fix round 1 ---- |
| -// Ruling 1: an identity is done only once its object exists. A write that |
| -// fails stays pending, and the next sighting tries again. |
| -func TestFailedLogWriteStaysPending(t *testing.T) { |
| - ctx := context.Background() |
| - calls := 0 |
| - w := &movesWorker{mirrors: map[string]lease.MirrorStore{}} |
| - w.cfg.Now = time.Now |
| - w.cfg.LogEvent = func(context.Context, release.Store, provision.Record, moves.LogEvent) error { |
| - calls++ |
| - if calls == 1 { |
| - return errors.New("the bucket is down") |
| - } |
| - return nil |
| - } |
| - rec := mvEvRec(provision.MovePreparing, nil) |
| - w.observeMove(ctx, nil, rec) |
| - w.observeMove(ctx, nil, rec) |
| - w.observeMove(ctx, nil, rec) |
| - if calls != 2 { |
| - t.Fatalf("the hook was called %d times, want a failure, a retry that succeeds, and nothing after", calls) |
| - } |
| -} |
| - |
| -// Ruling 7: a move's identities are forgotten once its terminal outcome is |
| -// logged (create-only storage still refuses a replay). |
| -func TestLogMemoForgetsAClosedMove(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]) |
| - } |
| - n := 0 |
| - w.logged.Range(func(k, _ any) bool { n++; return true }) |
| - if n != 0 { |
| - t.Fatalf("%d identities remembered after the close", n) |
| - } |
| -} |
| - |
| // Ruling 4: the box knows the home region only as "", so the member line |
| // names it with the reviewed phrase (MV-N381), never dropping the line. |
| func TestHomeRegionMemberLines(t *testing.T) { |
| @@ -335,7 +312,7 @@ func TestHomeRegionMemberLines(t *testing.T) { |
| {toHome, "flipped", "Stopped in ie. Starting in the home region."}, |
| } { |
| ev := moves.LogEvent{Event: c.event, Fields: map[string]string{"from": c.rec.Move.From, "to": c.rec.Move.To}} |
| - if line, ok := memberLine(ev, c.rec); !ok || line != c.want { |
| + if line, ok := memberLine(ev, c.rec.Host); !ok || line != c.want { |
| t.Errorf("%s: %q %v, want %q", c.event, line, ok, c.want) |
| } |
| } |
| @@ -404,47 +381,6 @@ func TestEventsCarryTheirEvidence(t *testing.T) { |
| } |
| } |
| -// Ruling 2: each recovery probe start the attempt journal counts is a log |
| -// event, probe_attempt-<epoch>-<n>, the starts of one wake included; a |
| -// reservation that only validates writes none. |
| -func TestRecoveryProbeAttemptsAreLogged(t *testing.T) { |
| - f := recoveryPAFixture(t, hostclaim.KindRestoring) |
| - var got []string |
| - hook := func(_ context.Context, rec provision.Record, epoch uint64, n int) { |
| - got = append(got, fmt.Sprintf("%s %d-%d", rec.Host, epoch, n)) |
| - } |
| - f.ops = f.attempts() |
| - f.ops.RecoveryAttempt = hook |
| - for n := 0; n < 3; n++ { |
| - if _, err := f.ops.OpenProbe(t.Context(), f.start, time.Now()); err != nil { |
| - t.Fatal(err) |
| - } |
| - if err := f.ops.Reserve(t.Context(), f.start, n); err != nil { |
| - t.Fatalf("start %d: %v", n, err) |
| - } |
| - if n == 0 { |
| - if err := f.ops.Reserve(t.Context(), f.start, 0); err != nil { |
| - t.Fatal(err) |
| - } |
| - } |
| - if err := f.ops.Failed(t.Context(), f.start, n); err != nil { |
| - t.Fatal(err) |
| - } |
| - } |
| - want := []string{f.rec.Host + " 0-1", f.rec.Host + " 0-2", f.rec.Host + " 0-3"} |
| - if strings.Join(got, ",") != strings.Join(want, ",") { |
| - t.Fatalf("attempts %v, want %v", got, want) |
| - } |
| - |
| - // The worker's hook names them probe_attempt-<epoch>-<n>. |
| - var evs []moves.LogEvent |
| - w := eventWorker(&evs) |
| - w.recoveryAttemptLogger(nil)(context.Background(), *f.rec, 2, 4) |
| - if len(evs) != 1 || evs[0].Event != "probe_attempt" || evs[0].Identity != "probe_attempt-2-4" { |
| - t.Fatalf("logged %+v", evs) |
| - } |
| -} |
| - |
| // Ruling 7: the real writers log, through the move harness. The landing |
| // harness drives the source's readiness, the console's cut, the cut and the |
| // flip, then the landing to installed; the close harness the probe, the |
| @@ -511,3 +447,204 @@ func wantSubsequence(t *testing.T, got []moves.LogEvent, want ...string) { |
| t.Fatalf("events %v lack %v in order (matched %d)", events, want, i) |
| } |
| } |
| + |
| +// flakyLogStore fails every write under a move's log while fail is set. |
| +type flakyLogStore struct { |
| + release.Store |
| + fail bool |
| +} |
| + |
| +func (f *flakyLogStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + if f.fail && strings.Contains(key, "/log/") { |
| + return "", errors.New("the bucket is down") |
| + } |
| + return f.Store.Put(ctx, key, body, ifMatch) |
| +} |
| + |
| +// outboxWorker is a worker with the production log writer over s and an |
| +// outbox under dataDir. |
| +func outboxWorker(dataDir string) *movesWorker { |
| + w := &movesWorker{mirrors: map[string]lease.MirrorStore{}} |
| + w.cfg.Box = "box-test" |
| + w.cfg.DataDir = dataDir |
| + w.cfg.Now = func() time.Time { return time.Unix(1_800_000_100, 0).UTC() } |
| + w.cfg.LogEvent = moveLogWriter(nil) |
| + return w |
| +} |
| + |
| +func outboxFiles(t *testing.T, dataDir string) []string { |
| + t.Helper() |
| + var out []string |
| + root := moveLogOutboxDir(dataDir) |
| + err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error { |
| + if errors.Is(err, fs.ErrNotExist) { |
| + return fs.SkipAll |
| + } |
| + if err != nil { |
| + return err |
| + } |
| + if !d.IsDir() { |
| + rel, _ := filepath.Rel(root, p) |
| + out = append(out, filepath.ToSlash(rel)) |
| + } |
| + return nil |
| + }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return out |
| +} |
| + |
| +// Fix round 2, ruling 1: a log write that fails is kept, with the entry's |
| +// bytes, in the worker's outbox on disk, and every pass retries the outbox, |
| +// whatever the record now says. Here the write fails at the request and |
| +// again at the close; the move is gone from the record by the time the |
| +// bucket answers, and a fresh worker (an agent restart) writes both. |
| +func TestLogOutboxOutlivesTheMove(t *testing.T) { |
| + ctx := context.Background() |
| + dataDir := t.TempDir() |
| + s := &flakyLogStore{Store: release.NewDirStore(t.TempDir()), fail: true} |
| + w := outboxWorker(dataDir) |
| + flow := doneFlow() |
| + w.observeMove(ctx, s, flow[0]) |
| + last := flow[len(flow)-1] |
| + w.logTransition(ctx, s, flow[len(flow)-2], last) |
| + if last.Move != nil { |
| + t.Fatal("the fixture's last record still has its move") |
| + } |
| + files := outboxFiles(t, dataDir) |
| + if strings.Join(files, " ") != "acme.go.tito.io/"+mvEvMoveID+"/done.json acme.go.tito.io/"+mvEvMoveID+"/requested.json" { |
| + t.Fatalf("outbox %v", files) |
| + } |
| + w.flushLogOutbox(ctx, s) // still failing: kept |
| + if len(outboxFiles(t, dataDir)) != 2 { |
| + t.Fatal("a failed retry dropped an entry") |
| + } |
| + pending, err := os.ReadFile(filepath.Join(moveLogOutboxDir(dataDir), "acme.go.tito.io", mvEvMoveID, "requested.json")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var kept moveLogPending |
| + if err := json.Unmarshal(pending, &kept); err != nil { |
| + t.Fatal(err) |
| + } |
| + s.fail = false |
| + outboxWorker(dataDir).flushLogOutbox(ctx, s) |
| + if f := outboxFiles(t, dataDir); len(f) != 0 { |
| + t.Fatalf("written entries left in the outbox: %v", f) |
| + } |
| + prefix := moves.ReportPrefix("bab", "titogo", "acme.go.tito.io", mvEvMoveID) |
| + got, err := moves.ReadLog(ctx, s, prefix) |
| + if err != nil || len(got) != 2 { |
| + t.Fatalf("log %+v %v", got, err) |
| + } |
| + body, _, err := s.Get(ctx, moves.LogKeyFor(prefix, "requested")) |
| + if err != nil || string(body) != string(kept.Entry) { |
| + t.Fatalf("the object is not the kept bytes:\n%s\n%s", body, kept.Entry) |
| + } |
| + // Another sighting of the closed move writes nothing more. |
| + w.observeMove(ctx, s, last) |
| + if f := outboxFiles(t, dataDir); len(f) != 0 { |
| + t.Fatalf("outbox %v", f) |
| + } |
| +} |
| + |
| +// Fix round 2, ruling 2: recovery probe starts are logged through the |
| +// journal as a box builds it (movesWorker.probeAttempts), and a start whose |
| +// log write fails is kept in the outbox. |
| +func TestRecoveryProbeAttemptsAreLogged(t *testing.T) { |
| + ctx := context.Background() |
| + f := recoveryPAFixture(t, hostclaim.KindRestoring) |
| + s := &flakyLogStore{Store: f.claims, fail: true} |
| + w := outboxWorker(f.dataDir) |
| + ids := 0 |
| + ops := w.probeAttempts(s, func(context.Context, string, string, string) (*provision.Record, error) { return f.rec, nil }, |
| + func() string { ids++; return "id-" + strings.Repeat("x", ids) }) |
| + for n := 0; n < 3; n++ { |
| + if _, err := ops.OpenProbe(ctx, f.start, time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := ops.Reserve(ctx, f.start, n); err != nil { |
| + t.Fatalf("start %d: %v", n, err) |
| + } |
| + if err := ops.Failed(ctx, f.start, n); err != nil { |
| + t.Fatal(err) |
| + } |
| + s.fail = false // only the first start's write fails |
| + } |
| + if files := outboxFiles(t, f.dataDir); len(files) != 1 || !strings.HasSuffix(files[0], "/probe_attempt-0-1.json") { |
| + t.Fatalf("outbox %v, want the first start's entry", files) |
| + } |
| + w.flushLogOutbox(ctx, s) |
| + got, err := moves.ReadLog(ctx, s, moves.ReportPrefix(f.rec.Account, f.rec.App, f.rec.Host, f.rec.Move.ID)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var names []string |
| + for _, ev := range got { |
| + if ev.Event != "probe_attempt" { |
| + t.Fatalf("event %+v", ev) |
| + } |
| + names = append(names, ev.Identity) |
| + } |
| + sort.Strings(names) |
| + if strings.Join(names, " ") != "probe_attempt-0-1 probe_attempt-0-2 probe_attempt-0-3" { |
| + t.Fatalf("logged %v", names) |
| + } |
| +} |
| + |
| +// Fix round 2, ruling 3: the memo forgets a move as soon as a pass sees its |
| +// record with no move or another move: here a source box whose move was |
| +// closed on the target, then the same box when the next move replaced it. |
| +// The memo is also bounded, dropping its oldest identities. |
| +func TestLogMemoRetiresAMoveClosedElsewhere(t *testing.T) { |
| + ctx := context.Background() |
| + var got []moves.LogEvent |
| + w := eventWorker(&got) |
| + flow := doneFlow() |
| + w.observeMove(ctx, nil, flow[0]) |
| + w.logTransition(ctx, nil, flow[0], flow[1]) |
| + if w.logged.len() == 0 { |
| + t.Fatal("nothing remembered") |
| + } |
| + s := release.NewDirStore(t.TempDir()) |
| + closed := flow[len(flow)-1] |
| + if err := provision.CreateRecord(ctx, s, closed); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := movesPass(ctx, s, w); err != nil { |
| + t.Fatal(err) |
| + } |
| + if n := w.logged.len(); n != 0 { |
| + t.Fatalf("%d identities remembered after the pass saw the move closed elsewhere", n) |
| + } |
| + w.observeMove(ctx, nil, flow[0]) |
| + next := mvEvRec(provision.MovePreparing, func(m *provision.MoveState) { m.ID = "mv_next" }) |
| + s = release.NewDirStore(t.TempDir()) |
| + if err := provision.CreateRecord(ctx, s, next); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := movesPass(ctx, s, w); err != nil { |
| + t.Fatal(err) |
| + } |
| + if w.logged.seen(next.Host + "/" + mvEvMoveID + "/requested") { |
| + t.Fatal("the replaced move is still remembered") |
| + } |
| +} |
| + |
| +func TestLogMemoIsBounded(t *testing.T) { |
| + var m moveLogMemo |
| + for i := 0; i < moveLogMemoCap+10; i++ { |
| + m.add("h", "op", "h/op/"+strconv.Itoa(i)) |
| + } |
| + if m.len() != moveLogMemoCap { |
| + t.Fatalf("%d entries, want %d", m.len(), moveLogMemoCap) |
| + } |
| + if m.seen("h/op/0") || !m.seen("h/op/"+strconv.Itoa(moveLogMemoCap+9)) { |
| + t.Fatal("not the oldest dropped") |
| + } |
| + m.forget("h", "op") |
| + if m.len() != 0 { |
| + t.Fatalf("%d entries after forget", m.len()) |
| + } |
| +} |
diff --git a/internal/host/moveslogoutbox.go b/internal/host/moveslogoutbox.go| new file mode 100644 |
| index 000000000..ef969b67a |
| --- /dev/null |
| +++ b/internal/host/moveslogoutbox.go |
| @@ -0,0 +1,169 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "io/fs" |
| + "log" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// The move log's outbox (Plan 6 Task 3, fix round 2, ruling 1). An entry |
| +// whose write fails is kept on the box's disk, under |
| +// <DataDir>/.moves/.log-outbox/<host>/<op>/<identity>.json, with the entry's |
| +// exact bytes and the app it belongs to. Every pass retries every kept entry |
| +// (flushLogOutbox), whatever its move has done since, and deletes the file |
| +// only once the entry is created or found there. The directory's leading dot |
| +// keeps it apart from the moves journal's host directories. |
| + |
| +// moveLogOutboxDir is the outbox's root. |
| +func moveLogOutboxDir(dataDir string) string { |
| + return filepath.Join(dataDir, ".moves", ".log-outbox") |
| +} |
| + |
| +// moveLogPending is one kept entry: Entry is the bytes the log object holds |
| +// (json.Marshal of the moves.LogEvent, as moves.WriteLog writes it). |
| +type moveLogPending struct { |
| + V int `json:"v"` |
| + Account string `json:"account"` |
| + App string `json:"app"` |
| + Host string `json:"host"` |
| + Entry json.RawMessage `json:"entry"` |
| +} |
| + |
| +// outboxSegment refuses a path segment that is empty or could leave its |
| +// directory. |
| +func outboxSegment(s string) bool { |
| + return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00") |
| +} |
| + |
| +// queueLog keeps ev, for rec's app and host, in the outbox: written to a |
| +// temporary file, synced, and renamed into place. |
| +func (w *movesWorker) queueLog(rec provision.Record, ev moves.LogEvent) error { |
| + if w.cfg.DataDir == "" { |
| + return errors.New("no data directory for the move log's outbox") |
| + } |
| + if !outboxSegment(rec.Host) || !outboxSegment(ev.Op) || !outboxSegment(ev.Identity) { |
| + return fmt.Errorf("an unusable move log entry name: %s/%s/%s", rec.Host, ev.Op, ev.Identity) |
| + } |
| + entry, err := json.Marshal(ev) |
| + if err != nil { |
| + return err |
| + } |
| + body, err := json.Marshal(moveLogPending{V: 1, Account: rec.Account, App: rec.App, Host: rec.Host, Entry: entry}) |
| + if err != nil { |
| + return err |
| + } |
| + dir := filepath.Join(moveLogOutboxDir(w.cfg.DataDir), rec.Host, ev.Op) |
| + if err := os.MkdirAll(dir, 0o700); err != nil { |
| + return err |
| + } |
| + tmp, err := os.CreateTemp(dir, ".pending-*") |
| + if err != nil { |
| + return err |
| + } |
| + defer os.Remove(tmp.Name()) |
| + if _, err := tmp.Write(body); err != nil { |
| + tmp.Close() |
| + return err |
| + } |
| + if err := tmp.Sync(); err != nil { |
| + tmp.Close() |
| + return err |
| + } |
| + if err := tmp.Close(); err != nil { |
| + return err |
| + } |
| + if err := os.Rename(tmp.Name(), filepath.Join(dir, ev.Identity+".json")); err != nil { |
| + return err |
| + } |
| + if d, err := os.Open(dir); err == nil { |
| + _ = d.Sync() |
| + d.Close() |
| + } |
| + return nil |
| +} |
| + |
| +// flushLogOutbox retries every kept entry through the LogEvent hook, into s. |
| +// A written entry's file is deleted; one that fails again stays. A file that |
| +// cannot be read as an entry is logged and deleted: it can never be written. |
| +func (w *movesWorker) flushLogOutbox(ctx context.Context, s release.Store) { |
| + f := w.cfg.LogEvent |
| + if w.cfg.DataDir == "" || f == nil || s == nil { |
| + return |
| + } |
| + root := moveLogOutboxDir(w.cfg.DataDir) |
| + err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error { |
| + if errors.Is(err, fs.ErrNotExist) && p == root { |
| + return fs.SkipAll |
| + } |
| + if err != nil { |
| + return err |
| + } |
| + if d.IsDir() || !strings.HasSuffix(d.Name(), ".json") || strings.HasPrefix(d.Name(), ".") { |
| + return nil |
| + } |
| + if err := ctx.Err(); err != nil { |
| + return err |
| + } |
| + w.flushLogFile(ctx, s, f, p) |
| + return nil |
| + }) |
| + if err != nil { |
| + log.Printf("moves: the move log's outbox: %v", err) |
| + } |
| + pruneEmptyDirs(root) |
| +} |
| + |
| +// flushLogFile retries one kept entry. |
| +func (w *movesWorker) flushLogFile(ctx context.Context, s release.Store, f func(context.Context, release.Store, provision.Record, moves.LogEvent) error, p string) { |
| + body, err := os.ReadFile(p) |
| + if err != nil { |
| + log.Printf("moves: the move log's outbox: %v", err) |
| + return |
| + } |
| + var kept moveLogPending |
| + var ev moves.LogEvent |
| + if err := json.Unmarshal(body, &kept); err != nil || kept.V != 1 || json.Unmarshal(kept.Entry, &ev) != nil || |
| + ev.Identity+".json" != filepath.Base(p) || ev.Op != filepath.Base(filepath.Dir(p)) || kept.Host != filepath.Base(filepath.Dir(filepath.Dir(p))) { |
| + log.Printf("moves: the move log's outbox: %s is not a kept entry; deleted", p) |
| + _ = os.Remove(p) |
| + return |
| + } |
| + if err := f(ctx, s, provision.Record{Account: kept.Account, App: kept.App, Host: kept.Host}, ev); err != nil { |
| + log.Printf("moves: %s: the log of %s at %s is kept for the next pass: %v", kept.Host, ev.Op, ev.Identity, err) |
| + return |
| + } |
| + if err := os.Remove(p); err != nil { |
| + log.Printf("moves: the move log's outbox: %v", err) |
| + } |
| +} |
| + |
| +// pruneEmptyDirs removes the outbox's empty host and move directories. |
| +func pruneEmptyDirs(root string) { |
| + hosts, err := os.ReadDir(root) |
| + if err != nil { |
| + return |
| + } |
| + for _, h := range hosts { |
| + if !h.IsDir() { |
| + continue |
| + } |
| + hd := filepath.Join(root, h.Name()) |
| + ops, _ := os.ReadDir(hd) |
| + for _, o := range ops { |
| + if o.IsDir() { |
| + _ = os.Remove(filepath.Join(hd, o.Name())) // only if empty |
| + } |
| + } |
| + _ = os.Remove(hd) |
| + } |
| +} |
diff --git a/internal/host/movespass.go b/internal/host/movespass.go| index adb7e8555..1fd81240d 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -90,9 +90,9 @@ type movesWorker struct { |
| archives map[string]verifiedArchive |
| measured moves.Measurements |
| notices sync.Map // key → time.Time of the last line |
| - // logged is the move log identities this worker has handed to LogEvent |
| - // (movesevents.go): host/op/identity → true. |
| - logged sync.Map |
| + // logged is the move log identities this worker has handed off |
| + // (movesevents.go): host/op/identity, bounded. |
| + logged moveLogMemo |
| 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 |
| @@ -707,6 +707,9 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error { |
| if cfg.claimRecords { |
| ctx = provision.WithRecordSequence(ctx) |
| } |
| + // The move log's kept entries are retried first, whatever their moves |
| + // now are (moveslogoutbox.go). |
| + w.flushLogOutbox(ctx, s) |
| recs, err := provision.ListAllRecords(ctx, s) |
| if err != nil { |
| return fmt.Errorf("moves: list records: %w", err) |
| @@ -731,6 +734,9 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error { |
| } |
| } |
| for _, rec := range recs { |
| + // A move closed or replaced by another writer is forgotten by the |
| + // move log's memory (movesevents.go). |
| + w.retireMoves(rec) |
| if rec.Move != nil && !rec.Move.Legacy() { |
| // A Phase 1 record. With the switch off it is never touched |
| // (none exists unless the switch was on somewhere). |
diff --git a/internal/host/movesprobeattempts.go b/internal/host/movesprobeattempts.go| index 826dda6bb..6a1c71225 100644 |
| --- a/internal/host/movesprobeattempts.go |
| +++ b/internal/host/movesprobeattempts.go |
| @@ -4,7 +4,6 @@ import ( |
| "context" |
| "encoding/json" |
| "fmt" |
| - "log" |
| "os/exec" |
| "strconv" |
| "sync" |
| @@ -71,7 +70,7 @@ type movesProbeAttempts struct { |
| // RecoveryAttempt, when set, is told of each recovery probe start this |
| // journal counts: the record, its recovery retry epoch and the start's |
| // number (the move log's probe_attempt-<epoch>-<n>, Plan 6 Task 3; |
| - // movesWorker.recoveryAttemptLogger). Whoever builds the journal sets it. |
| + // movesWorker.recoveryAttemptLogger, set by movesWorker.probeAttempts). |
| RecoveryAttempt func(ctx context.Context, rec provision.Record, epoch uint64, n int) |
| mu sync.Mutex |
| @@ -115,14 +114,28 @@ func (p *movesProbeAttempts) valid() bool { |
| // or -1 for a recovery probe. Every case validates the admitted writer. |
| // Recovery also returns the budget key from the same record read. |
| func (p *movesProbeAttempts) bound(ctx context.Context, s OperationStart) (int, string, error) { |
| + k, key, _, err := p.boundRecord(ctx, s) |
| + return k, key, err |
| +} |
| + |
| +// boundRecord is bound, with the record a recovery probe was bound against |
| +// (nil otherwise): Reserve logs a recovery start from it. |
| +func (p *movesProbeAttempts) boundRecord(ctx context.Context, s OperationStart) (int, string, *provision.Record, error) { |
| if !p.valid() { |
| - return 0, "", hostwrite.ErrUnavailable |
| + return 0, "", nil, hostwrite.ErrUnavailable |
| } |
| a := s.Admitted |
| if a.Claim != nil && claimRecoveryKind(a.Claim.Op.Kind) { |
| - key, err := p.boundRecovery(ctx, s) |
| - return -1, key, err |
| + key, rec, err := p.boundRecovery(ctx, s) |
| + return -1, key, rec, err |
| } |
| + k, key, err := p.boundLanding(ctx, s) |
| + return k, key, nil, err |
| +} |
| + |
| +// boundLanding is bound for a landing's (or a committed wake's) probe. |
| +func (p *movesProbeAttempts) boundLanding(ctx context.Context, s OperationStart) (int, string, error) { |
| + a := s.Admitted |
| if a.Claim == nil || a.Claim.Holder == nil || a.Claim.Op.Kind != hostclaim.KindLanding { |
| return 0, "", hostbox.ErrUnsupported |
| } |
| @@ -303,7 +316,7 @@ func sameStartOwner(a hostclaim.WakeRetryAttempt, s OperationStart, digest strin |
| // reserved by the first Reserve(0) of the admission and validated by every |
| // later one; start n > 0 only after start n-1 was settled failed. |
| func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n int) (retErr error) { |
| - k, recoveryKey, err := p.bound(ctx, s) |
| + k, recoveryKey, bindRec, err := p.boundRecord(ctx, s) |
| if err != nil { |
| return err |
| } |
| @@ -364,7 +377,7 @@ func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n in |
| } |
| defer func() { |
| if counted > 0 && retErr == nil { |
| - p.recoveryAttempted(ctx, s, counted) |
| + p.recoveryAttempted(ctx, bindRec, counted) |
| } |
| }() |
| o := s.Owner |
| @@ -516,38 +529,38 @@ func probeAlive(cmd *exec.Cmd) bool { |
| // RecoveryProbeAttemptsReady is checked before the adapter records inputs. |
| func (p *movesProbeAttempts) RecoveryProbeAttemptsReady() bool { return p.valid() } |
| -func (p *movesProbeAttempts) boundRecovery(ctx context.Context, s OperationStart) (string, error) { |
| +func (p *movesProbeAttempts) boundRecovery(ctx context.Context, s OperationStart) (string, *provision.Record, error) { |
| a := s.Admitted |
| if a.Claim.Holder == nil || !s.Policy.Probe { |
| - return "", hostbox.ErrChanged |
| + return "", nil, hostbox.ErrChanged |
| } |
| h := a.Claim.Holder |
| if s.Owner.Host != a.Host || s.Owner.Incarnation != h.Incarnation || s.Owner.Stamp != h.Stamp || s.Owner.Session != h.Session || s.Owner.Box != a.Claim.Op.Exec { |
| - return "", hostbox.ErrChanged |
| + return "", nil, hostbox.ErrChanged |
| } |
| fresh, err := hostclaim.Read(ctx, p.Claims, a.Host) |
| if err != nil { |
| - return "", err |
| + return "", nil, err |
| } |
| if !claimSameWriter(a, fresh, hostclaim.SessionContinues) || fresh.Claim.Op.Phase != hostclaim.PhaseProbing || fresh.Claim.Op.Exec != s.Owner.Box { |
| - return "", hostbox.ErrChanged |
| + return "", nil, hostbox.ErrChanged |
| } |
| rec, err := p.Records(ctx, h.Account, h.App, a.Host) |
| if err != nil { |
| - return "", err |
| + return "", nil, err |
| } |
| if rec == nil || rec.Host != a.Host || rec.Move == nil || rec.Move.ID != a.Claim.Op.ID || rec.Move.Incarnation != h.Incarnation { |
| - return "", hostbox.ErrChanged |
| + return "", nil, hostbox.ErrChanged |
| } |
| m := rec.Move |
| if a.Claim.Op.Kind == hostclaim.KindReverting { |
| if m.State != provision.MoveReverting || m.ReopenedAt != nil || fresh.Claim.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing { |
| - return "", hostbox.ErrChanged |
| + return "", nil, hostbox.ErrChanged |
| } |
| } else if m.State != provision.MoveRollingBack || m.Rollback == nil || m.Rollback.ID == "" || m.Rollback.ID != a.Claim.Op.Intent["rollback"] || m.Rollback.ID != fresh.Claim.Op.Intent["rollback"] || m.Rollback.SourcePhase != provision.RbSourceProbing { |
| - return "", hostbox.ErrChanged |
| + return "", nil, hostbox.ErrChanged |
| } |
| - return recoveryAttemptKey(s, m.RecoveryRetryEpoch), nil |
| + return recoveryAttemptKey(s, m.RecoveryRetryEpoch), rec, nil |
| } |
| // The move has its own journal directory. JSON preserves the rollback and |
| @@ -580,17 +593,23 @@ func recoveryAttemptCount(j *moves.Journal, key string) (int, error) { |
| } |
| // recoveryAttempted tells RecoveryAttempt of start n of a recovery probe, |
| -// read against the record its admission names (for the retry epoch). A |
| -// record that cannot be read is logged, and the start is not reported. |
| -func (p *movesProbeAttempts) recoveryAttempted(ctx context.Context, s OperationStart, n int) { |
| - if p.RecoveryAttempt == nil || p.Records == nil || s.Admitted.Claim == nil || s.Admitted.Claim.Holder == nil { |
| - return |
| - } |
| - h := s.Admitted.Claim.Holder |
| - rec, err := p.Records(ctx, h.Account, h.App, s.Admitted.Host) |
| - if err != nil || rec == nil || rec.Move == nil { |
| - log.Printf("moves: %s: the log of recovery probe start %d: the record: %v", s.Admitted.Host, n, err) |
| +// with the record the start was bound against (its retry epoch): the read |
| +// the binding already made, so a counted start is never lost to a second |
| +// read. The hook's own failed write goes to the outbox. |
| +func (p *movesProbeAttempts) recoveryAttempted(ctx context.Context, rec *provision.Record, n int) { |
| + if p.RecoveryAttempt == nil || rec == nil || rec.Move == nil { |
| return |
| } |
| p.RecoveryAttempt(ctx, *rec, rec.Move.RecoveryRetryEpoch, n) |
| } |
| + |
| +// probeAttempts is the attempt journal a box builds for its probe wakes |
| +// (Plan 3 Task 5), over the deployment store s, its records and newID, with |
| +// this worker's data directory and clock. Its RecoveryAttempt is the |
| +// worker's move log hook (recoveryAttemptLogger), so every recovery probe |
| +// start it counts is logged (fix round 2, ruling 2). Production builds the |
| +// journal here once Plan 9b registers the wake source. |
| +func (w *movesWorker) probeAttempts(s release.Store, records MovesRecordReader, newID func() string) *movesProbeAttempts { |
| + return &movesProbeAttempts{Claims: s, Records: records, DataDir: w.cfg.DataDir, Admitted: MovesRecordAdmittedAttempts, |
| + NewID: newID, Now: w.cfg.Now, RecoveryAttempt: w.recoveryAttemptLogger(s)} |
| +} |
diff --git a/internal/hostops/move_recovery_grant.go b/internal/hostops/move_recovery_grant.go| index b0dd28552..b65e5033b 100644 |
| --- a/internal/hostops/move_recovery_grant.go |
| +++ b/internal/hostops/move_recovery_grant.go |
| @@ -1,6 +1,7 @@ |
| package hostops |
| import ( |
| + "bytes" |
| "context" |
| "crypto/rand" |
| "encoding/json" |
| @@ -27,14 +28,14 @@ type MoveRecoveryGrant struct { |
| o ConsoleMoveObservation |
| authority hostclaim.FenceAuthority |
| scope *moveRecoveryPassScope |
| + // wrote is what the grant's record write did (Transition), once it has |
| + // made one. |
| + wrote *moves.Transition |
| } |
| type moveRecoveryGrantState struct { |
| mu sync.Mutex |
| active bool |
| - // running is true while the hook that holds the grant runs: the life of |
| - // Log, which outlives the one mutation active allows. |
| - running bool |
| } |
| func moveRecoveryKind(kind string) bool { |
| @@ -90,9 +91,9 @@ func runMoveRecovery(ctx context.Context, obs hostclaim.Observed, hook func(cont |
| if len(authorities) == 1 { |
| authority = authorities[0] |
| } |
| - g := &MoveRecoveryGrant{state: &moveRecoveryGrantState{active: true, running: true}, w: w, scope: scope, authority: authority, o: ConsoleMoveObservation{Account: h.Account, App: h.App, AppID: c.Op.Intent["app_id"], Host: obs.Host, Incarnation: h.Incarnation, OpID: c.Op.ID, ClaimETag: obs.ETag, RecordETag: r.ETag, RecordSeq: seq}} |
| + g := &MoveRecoveryGrant{state: &moveRecoveryGrantState{active: true}, w: w, scope: scope, authority: authority, o: ConsoleMoveObservation{Account: h.Account, App: h.App, AppID: c.Op.Intent["app_id"], Host: obs.Host, Incarnation: h.Incarnation, OpID: c.Op.ID, ClaimETag: obs.ETag, RecordETag: r.ETag, RecordSeq: seq}} |
| state := g.state |
| - defer func() { state.mu.Lock(); state.active, state.running = false, false; state.mu.Unlock() }() |
| + defer func() { state.mu.Lock(); state.active = false; state.mu.Unlock() }() |
| // The owner receives only this capability, never the minting context. |
| ctx = context.WithValue(ctx, lifecycleGrantKey{}, []lifecycleGrant(nil)) |
| ctx = context.WithValue(ctx, moveRecoveryPassKey{}, (*moveRecoveryPassScope)(nil)) |
| @@ -125,26 +126,96 @@ func (g *MoveRecoveryGrant) current(ctx context.Context) (hostclaim.Observed, Re |
| return o, r, rec, err |
| } |
| -// Log writes one entry of this move's log (Plan 6 Task 3): create-only |
| -// (moves.WriteLog), under this grant's move's own report prefix, while the |
| -// hook that holds the grant runs, after its mutation as before it. It is the |
| -// pass's one write beside the record and the claim; an entry is a record of |
| -// a transition, never an input to one, so it needs no fresh observation. |
| -// created is false when the entry was already there. |
| -func (g *MoveRecoveryGrant) Log(ctx context.Context, ev moves.LogEvent) (created bool, err error) { |
| - if g == nil || g.state == nil || g.w == nil { |
| +// MoveLogEntryMax bounds one entry the grant's Log writes, in bytes. |
| +const MoveLogEntryMax = 4096 |
| + |
| +// ErrMoveLogTaken is a Log whose entry name already holds other bytes: the |
| +// same transition, logged first by another writer. Nothing is written. |
| +var ErrMoveLogTaken = errors.New("hostops: the move log entry was written by another writer") |
| + |
| +// Log writes one entry of this grant's move's log (Plan 6 Task 3; the |
| +// conditions are platform-1e's, fix round 2 ruling 4): |
| +// |
| +// - The key is the grant's alone: name under the report prefix of the |
| +// grant's move (its account, app, host and move id), whose record still |
| +// names that move at the grant's incarnation, live or closed as its |
| +// last. The caller gives only name, one segment, which must be the |
| +// entry's identity. |
| +// - body is at most MoveLogEntryMax bytes and decodes, strictly, as a |
| +// moves.LogEvent of this move and host. |
| +// - The write is create-only. A replay of the same bytes is a no-op |
| +// (created false); other bytes under the name are ErrMoveLogTaken. |
| +// - It answers only while the grant is valid, by the check every other |
| +// method makes (current): never after the grant's one mutation, after |
| +// its hook, or across a changed claim or record. A pass logs what it |
| +// owes before it acts; an entry its own write makes waits in the |
| +// caller's outbox for the next pass's grant. |
| +// |
| +// An entry records a transition and is never an input to one. |
| +func (g *MoveRecoveryGrant) Log(ctx context.Context, name string, body []byte) (created bool, err error) { |
| + if g == nil || g.state == nil { |
| return false, ErrPending |
| } |
| g.state.mu.Lock() |
| - running := g.state.running |
| - g.state.mu.Unlock() |
| - if !running { |
| - return false, ErrPending |
| + defer g.state.mu.Unlock() |
| + _, _, rec, err := g.current(ctx) |
| + if err != nil { |
| + return false, err |
| + } |
| + if !moveLogName(name) || len(body) > MoveLogEntryMax { |
| + return false, ErrIdentity |
| + } |
| + var ev moves.LogEvent |
| + if err := strictJSON(body, &ev); err != nil { |
| + return false, ErrIdentity |
| } |
| - if ev.Op != g.o.OpID || ev.Host != g.o.Host { |
| + if ev.V != 1 || ev.Op != g.o.OpID || ev.Host != g.o.Host || ev.Identity != name { |
| return false, ErrIdentity |
| } |
| - return moves.WriteLog(ctx, g.w.raw, moves.ReportPrefix(g.o.Account, g.o.App, g.o.Host, g.o.OpID), ev) |
| + live := rec.Move != nil && rec.Move.ID == g.o.OpID && rec.Move.Incarnation == g.o.Incarnation |
| + closed := rec.Move == nil && rec.LastMove != nil && rec.LastMove.ID == g.o.OpID |
| + if !live && !closed { |
| + return false, ErrIdentity |
| + } |
| + key := moves.LogKeyFor(moves.ReportPrefix(g.o.Account, g.o.App, g.o.Host, g.o.OpID), name) |
| + if _, err := g.w.raw.Put(ctx, key, body, ""); err == nil { |
| + return true, nil |
| + } else if !errors.Is(err, release.ErrCASConflict) { |
| + return false, err |
| + } |
| + have, _, err := g.w.raw.Get(ctx, key) |
| + if err != nil { |
| + return false, err |
| + } |
| + if !bytes.Equal(have, body) { |
| + return false, ErrMoveLogTaken |
| + } |
| + return false, nil |
| +} |
| + |
| +// moveLogName is one path segment of letters, digits, '_' and '-', at most |
| +// 128 bytes: every identity the move log names (moves.TransitionOf) fits. |
| +func moveLogName(name string) bool { |
| + if name == "" || len(name) > 128 { |
| + return false |
| + } |
| + for _, r := range name { |
| + if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '_' || r == '-') { |
| + return false |
| + } |
| + } |
| + return true |
| +} |
| + |
| +// Transition is what the grant's record write did, named as the move log |
| +// names it (moves.TransitionOf over the record before and after the write), |
| +// once Record has written. It is a value, never authority: the caller's log |
| +// entry is built from the write's own evidence, not from a later read. |
| +func (g *MoveRecoveryGrant) Transition() (moves.Transition, bool) { |
| + if g == nil || g.wrote == nil { |
| + return moves.Transition{}, false |
| + } |
| + return *g.wrote, true |
| } |
| // Observation contains values, never authority. |
| @@ -231,8 +302,15 @@ func (g *MoveRecoveryGrant) Record(ctx context.Context, action MoveRecoveryRecor |
| if err != nil { |
| return err |
| } |
| - _, _, err = provision.WriteRecord(provision.WithRecordSequence(ctx), g.w.raw, r.Holder.Key, r.Body, body, r.ETag) |
| - return err |
| + if _, _, err = provision.WriteRecord(provision.WithRecordSequence(ctx), g.w.raw, r.Holder.Key, r.Body, body, r.ETag); err != nil { |
| + return err |
| + } |
| + if before, err := provision.DecodeRecord(r.Body); err == nil { |
| + if t, ok := moves.TransitionOf(before, rec); ok { |
| + g.wrote = &t |
| + } |
| + } |
| + return nil |
| } |
| func (g *MoveRecoveryGrant) prepareRecord(ctx context.Context, rec provision.Record) ([]byte, error) { |
diff --git a/internal/hostops/move_recovery_grant_test.go b/internal/hostops/move_recovery_grant_test.go| index ab7898158..827032803 100644 |
| --- a/internal/hostops/move_recovery_grant_test.go |
| +++ b/internal/hostops/move_recovery_grant_test.go |
| @@ -2,8 +2,10 @@ package hostops |
| import ( |
| "context" |
| + "encoding/json" |
| "errors" |
| "fmt" |
| + "strings" |
| "testing" |
| "time" |
| @@ -183,9 +185,15 @@ func TestMoveRecoveryRecordCloseThenRelease(t *testing.T) { |
| } |
| } |
| -// Plan 6 Task 3 fix round 1, ruling 3: the grant's Log writes this move's |
| -// log, create-only, under its own report prefix, after the grant's one |
| -// mutation as before it, and only while its hook runs. |
| +// Plan 6 Task 3 fix round 2, ruling 4 (platform-1e's conditions): the |
| +// grant's Log writes one entry of its own move's log. The key is derived |
| +// from the grant alone (its move's report prefix); the caller names only a |
| +// single-segment entry. The entry is at most MoveLogEntryMax bytes, is this |
| +// move's and this host's, and is named by its identity. The write is |
| +// create-only: a replay of the same bytes is a no-op, other bytes under the |
| +// name are ErrMoveLogTaken. Log answers only while the grant is valid, by |
| +// the check every other grant method uses: not after its one mutation, and |
| +// not after its hook. |
| func TestMoveRecoveryGrantLogsItsOwnMove(t *testing.T) { |
| w := newC7World(t, func(c *hostclaim.Claim, r *provision.Record, _ *AppAdmission) { |
| c.Op.Kind, c.Op.Phase, c.Op.Exec = hostclaim.KindLanding, hostclaim.PhaseCommitted, "target" |
| @@ -198,29 +206,41 @@ func TestMoveRecoveryGrantLogsItsOwnMove(t *testing.T) { |
| r.Move.SourceParkedAt = &now |
| }) |
| ctx := context.Background() |
| - ev := moves.LogEvent{V: 1, Op: "move", Host: w.host, Event: "done", Identity: "done", Actor: "console", At: w.now} |
| + entry := func(op, host, id string) []byte { |
| + b, err := json.Marshal(moves.LogEvent{V: 1, Op: op, Host: host, Event: "x", Identity: id, Actor: "console", At: w.now}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return b |
| + } |
| + body := entry("move", w.host, "opened") |
| var saved *MoveRecoveryGrant |
| + got := map[string]error{} |
| var created, again bool |
| - var other, wrongHost error |
| err := w.control.RunRecovery(ctx, func(ctx context.Context) error { |
| return runMoveRecovery(ctx, w.claim(t), func(ctx context.Context, g *MoveRecoveryGrant) error { |
| saved = g |
| - if err := g.Record(ctx, RecoveryMoveClose); err != nil { |
| + var err error |
| + if created, err = g.Log(ctx, "opened", body); err != nil { |
| return err |
| } |
| - var err error |
| - if created, err = g.Log(ctx, ev); err != nil { |
| + if again, err = g.Log(ctx, "opened", body); err != nil { |
| return err |
| } |
| - if again, err = g.Log(ctx, ev); err != nil { |
| + other, _ := json.Marshal(moves.LogEvent{V: 1, Op: "move", Host: w.host, Event: "x", Identity: "opened", Actor: "target", At: w.now}) |
| + _, got["other bytes"] = g.Log(ctx, "opened", other) |
| + _, got["another move"] = g.Log(ctx, "other", entry("other", w.host, "other")) |
| + _, got["another host"] = g.Log(ctx, "elsewhere", entry("move", "elsewhere.example", "elsewhere")) |
| + _, got["a name that is not the identity"] = g.Log(ctx, "done", entry("move", w.host, "opened")) |
| + _, got["a path"] = g.Log(ctx, "../source", entry("move", w.host, "../source")) |
| + _, got["an empty name"] = g.Log(ctx, "", entry("move", w.host, "")) |
| + big := moves.LogEvent{V: 1, Op: "move", Host: w.host, Event: "x", Identity: "big", Fields: map[string]string{"pad": strings.Repeat("a", MoveLogEntryMax)}} |
| + bb, _ := json.Marshal(big) |
| + _, got["too big"] = g.Log(ctx, "big", bb) |
| + if err := g.Record(ctx, RecoveryMoveClose); err != nil { |
| return err |
| } |
| - o := ev |
| - o.Op = "other" |
| - _, other = g.Log(ctx, o) |
| - h := ev |
| - h.Host = "elsewhere.example" |
| - _, wrongHost = g.Log(ctx, h) |
| + _, got["after the mutation"] = g.Log(ctx, "done", entry("move", w.host, "done")) |
| return nil |
| }) |
| }) |
| @@ -228,17 +248,26 @@ func TestMoveRecoveryGrantLogsItsOwnMove(t *testing.T) { |
| t.Fatal(err) |
| } |
| if !created || again { |
| - t.Fatalf("created=%v, a replay created=%v", created, again) |
| + t.Fatalf("created=%v, a replay of the same bytes created=%v", created, again) |
| + } |
| + for what, want := range map[string]error{ |
| + "other bytes": ErrMoveLogTaken, "another move": ErrIdentity, "another host": ErrIdentity, |
| + "a name that is not the identity": ErrIdentity, "a path": ErrIdentity, "an empty name": ErrIdentity, |
| + "too big": ErrIdentity, "after the mutation": ErrPending, |
| + } { |
| + if !errors.Is(got[what], want) { |
| + t.Errorf("%s: %v, want %v", what, got[what], want) |
| + } |
| } |
| - if !errors.Is(other, ErrIdentity) || !errors.Is(wrongHost, ErrIdentity) { |
| - t.Fatalf("another move's log: %v, another host's: %v", other, wrongHost) |
| + logged, err := moves.ReadLog(ctx, w.raw, moves.ReportPrefix("bab", "app", w.host, "move")) |
| + if err != nil || len(logged) != 1 || logged[0].Identity != "opened" { |
| + t.Fatalf("log %+v %v", logged, err) |
| } |
| - got, err := moves.ReadLog(ctx, w.raw, moves.ReportPrefix("bab", "app", w.host, "move")) |
| - if err != nil || len(got) != 1 || got[0].Identity != "done" || got[0].Actor != "console" { |
| - t.Fatalf("log %+v %v", got, err) |
| + raw, _, err := w.raw.Get(ctx, moves.LogKeyFor(moves.ReportPrefix("bab", "app", w.host, "move"), "opened")) |
| + if err != nil || string(raw) != string(body) { |
| + t.Fatalf("the object is not the caller's bytes: %s %v", raw, err) |
| } |
| - ev.Identity, ev.Event = "claim-released", "claim_released" |
| - if _, err := saved.Log(ctx, ev); !errors.Is(err, ErrPending) { |
| + if _, err := saved.Log(ctx, "late", entry("move", w.host, "late")); !errors.Is(err, ErrPending) { |
| t.Fatalf("a grant kept past its hook logged: %v", err) |
| } |
| } |
diff --git a/internal/moves/transition.go b/internal/moves/transition.go| index db7e27c9c..934ff6e49 100644 |
| --- a/internal/moves/transition.go |
| +++ b/internal/moves/transition.go |
| @@ -32,10 +32,10 @@ func phaseOf(m *provision.MoveState) (phase, target, source string) { |
| return "", "", "" |
| } |
| -// transitionFields are a move's regions and the digests its evidence has |
| -// recorded so far: the source's stop (final TXID, database, files, timers), |
| -// the landing's installed database, the pinned inputs (release, env |
| -// report). |
| +// transitionFields are a move's regions, the digests its evidence has |
| +// recorded so far (the source's stop: final TXID, database, files, timers; |
| +// the landing's installed database; the pinned inputs: release, env report) |
| +// and its admitted cut budget. |
| func transitionFields(m *provision.MoveState) map[string]string { |
| f := map[string]string{"from": m.From, "to": m.To} |
| set := func(k, v string) { |
| @@ -60,6 +60,10 @@ func transitionFields(m *provision.MoveState) map[string]string { |
| set("release", in.Release) |
| set("env_report", in.EnvReport) |
| } |
| + if m.CutBudgetS > 0 { |
| + // The admitted cut's budget, which the member line for it states. |
| + f["cut_budget_s"] = strconv.Itoa(m.CutBudgetS) |
| + } |
| return f |
| } |
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go| index e1b4e0592..b65ef0067 100644 |
| --- a/internal/provision/plan9_b8_census_test.go |
| +++ b/internal/provision/plan9_b8_census_test.go |
| @@ -473,7 +473,11 @@ func Reviewed(ctx context.Context, s Store, k string) error { |
| // BEGIN REVIEWED INVENTORY |
| var plan9B8CensusInventory = []censusEntry{ |
| {"internal/hostops/move_recovery_grant.go", "MoveRecoveryGrant.Convert", "record-writer", "key:WriteRecord", "794963a24a0a50d980505f83ac912973a10267b21f74bd41c56584ab9b0b25a1"}, |
| - {"internal/hostops/move_recovery_grant.go", "MoveRecoveryGrant.Record", "record-writer", "key:WriteRecord", "6b8bd4131842d8f22a6ab55aaae736329df40e34abb348d3d68cbd0a713503ad"}, |
| + // Writes one entry of the grant's own move's log, create-only, under its |
| + // report prefix (moves/<acct>/<app>/<host>/<op>/log/); never a record |
| + // (Plan 6 Task 3 fix round 2, ruling 4). |
| + {"internal/hostops/move_recovery_grant.go", "MoveRecoveryGrant.Log", "nonrecord", "sel:Put", "9f90ff153382fc6278ed0c17dffa5e0bfda01287716af1fa3149fef9772213c2"}, |
| + {"internal/hostops/move_recovery_grant.go", "MoveRecoveryGrant.Record", "record-writer", "key:WriteRecord", "f7f04a9e2c579eaa7d254ce6310e2726d95855307d61f3f442ee9c1fa2a5d825"}, |
| // Plan 10: reviewed typed console move and loss writers. |
| {"internal/hostops/console_move_grant.go", "ControlWriter.ConsoleMove", "record-writer", "key:WriteRecord", "21e7a6ddc4dbe5b3bf06e65b39a228e2eb588af1edc43beb09d91dd72570dd44"}, |
| {"internal/hostops/console_move_grant.go", "ControlWriter.consoleMoveClaim", "key-site", "key:RecordKey", "265291bd356de061663d89e5ebe33b74ea32c086f8238736d6f971a5afcab722"}, |