amadan
moves-phase1-6 Task 3 fix round 3 (2): claim_released only from a successful release
The release's entry is queued unconfirmed before g.Claim and confirmed after it succeeds; the drainer delivers confirmed rows only and settles an unconfirmed one from the claim: released for that move, confirmed and delivered; still the move's, left for the release; otherwise dropped with a journal line. moveReleaseFault is a test seam for a release that fails after its entry is queued.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3 files changed,
+155
−5
internal/console/movesrecovery.go+58 −2internal/console/movesrecovery_test.go+90 −0internal/console/server.go+7 −3
diff --git a/internal/console/movesrecovery.go b/internal/console/movesrecovery.go| index 4c5e57641..4d5c605f9 100644 |
| --- a/internal/console/movesrecovery.go |
| +++ b/internal/console/movesrecovery.go |
| @@ -334,12 +334,26 @@ func (s *Server) movesRecovery(ctx context.Context, g *hostops.MoveRecoveryGrant |
| } |
| return nil |
| case moveActClaim: |
| + if d.Claim == hostops.RecoveryMoveRelease { |
| + // claim_released is published only from a release that |
| + // happened: queued unconfirmed now, confirmed once the release |
| + // succeeds (a crash between the two is settled by the drainer, |
| + // from the claim). |
| + s.queueMoveLogAs(o, claimTransition(o, rec, "claim_released", "claim-released"), now, false) |
| + if f := s.moveReleaseFault; f != nil { |
| + if err := f(); err != nil { |
| + return err |
| + } |
| + } |
| + } |
| if err := g.Claim(ctx, d.Claim); err != nil { |
| return err |
| } |
| switch d.Claim { |
| case hostops.RecoveryMoveRelease: |
| - s.queueMoveLog(o, claimTransition(o, rec, "claim_released", "claim-released"), now) |
| + if err := s.store.ConfirmMoveLog(o.Host, o.OpID, "claim-released"); err != nil { |
| + log.Printf("console: moves: %s: the release of move %s is not confirmed in its log outbox (the drainer settles it from the claim): %v", o.Host, o.OpID, err) |
| + } |
| case hostops.RecoveryMoveRollback: |
| s.queueMoveLog(o, claimTransition(o, rec, "claim_rollback", "claim-rollback"), now) |
| } |
| @@ -545,6 +559,11 @@ func moveAbortOverduePage(host, id, to, box string) (subject, body string) { |
| // 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 release's own entry (claim_released) is published only from a release |
| +// that happened (ruling 2): queued unconfirmed before g.Claim and confirmed |
| +// after it succeeds; the drainer settles a row left unconfirmed (a failed |
| +// release, or a crash before the confirm) from the claim. |
| +// |
| // An entry is queued in the console's outbox (move_log_outbox) with its |
| // exact bytes and its report prefix, and the recovery pass's drainer |
| // (drainMoveLog) delivers it, once a pass, whatever its move has done since: |
| @@ -577,9 +596,16 @@ func moveLogEntry(o hostops.ConsoleMoveObservation, t moves.Transition, now time |
| // queueMoveLog queues the entry for t, for the drainer. |
| func (s *Server) queueMoveLog(o hostops.ConsoleMoveObservation, t moves.Transition, now time.Time) { |
| + s.queueMoveLogAs(o, t, now, true) |
| +} |
| + |
| +// queueMoveLogAs queues the entry for t, confirmed or not: the drainer |
| +// delivers a confirmed entry, and settles an unconfirmed one from the claim |
| +// (settleUnconfirmed). |
| +func (s *Server) queueMoveLogAs(o hostops.ConsoleMoveObservation, t moves.Transition, now time.Time, confirmed bool) { |
| body, err := moveLogEntry(o, t, now) |
| if err == nil { |
| - err = s.store.QueueMoveLog(o.Host, o.OpID, t.Identity, moves.ReportPrefix(o.Account, o.App, o.Host, o.OpID), body, true, now) |
| + err = s.store.QueueMoveLog(o.Host, o.OpID, t.Identity, moves.ReportPrefix(o.Account, o.App, o.Host, o.OpID), body, confirmed, 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) |
| @@ -608,11 +634,41 @@ func (s *Server) drainMoveLog(ctx context.Context, lw *hostops.MoveLogWriter) er |
| if err := ctx.Err(); err != nil { |
| return err |
| } |
| + if !r.Confirmed && !s.settleUnconfirmed(ctx, r) { |
| + continue |
| + } |
| s.deliverMoveLog(ctx, lw, r, now) |
| } |
| return nil |
| } |
| +// settleUnconfirmed settles a row queued for a release that was not yet |
| +// confirmed, from the host's claim: released for the row's move, it is |
| +// confirmed (true: deliver it); still the move's, it is left for the |
| +// release; anything else, it is dropped with a journal line. |
| +func (s *Server) settleUnconfirmed(ctx context.Context, r MoveLogPendingEntry) bool { |
| + o, err := hostclaim.Read(ctx, s.cfg.Releases, r.Host) |
| + if err != nil { |
| + log.Printf("console: moves: %s: the unconfirmed log of move %s at %s waits: the claim: %v", r.Host, r.Op, r.Name, err) |
| + return false |
| + } |
| + switch { |
| + case !o.Absent && o.Claim != nil && hostclaim.RequireReleasedMove(r.Op)(o.Claim) == nil: |
| + if err := s.store.ConfirmMoveLog(r.Host, r.Op, r.Name); err != nil { |
| + log.Printf("console: moves: %s: the log outbox of move %s: %v", r.Host, r.Op, err) |
| + return false |
| + } |
| + return true |
| + case !o.Absent && o.Claim != nil && o.Claim.Op.ID == r.Op: |
| + return false |
| + } |
| + log.Printf("console: moves: %s: the log of move %s at %s was never confirmed and the claim no longer holds the move; dropped", r.Host, r.Op, r.Name) |
| + if err := s.store.SettleMoveLog(r.Host, r.Op, r.Name); err != nil { |
| + log.Printf("console: moves: %s: the log outbox of move %s: %v", r.Host, r.Op, err) |
| + } |
| + return false |
| +} |
| + |
| // deliverMoveLog writes one row, and settles it or schedules its retry. |
| func (s *Server) deliverMoveLog(ctx context.Context, lw *hostops.MoveLogWriter, r MoveLogPendingEntry, now time.Time) { |
| settle := func() { |
diff --git a/internal/console/movesrecovery_test.go b/internal/console/movesrecovery_test.go| index c4e3e1241..d4367ac27 100644 |
| --- a/internal/console/movesrecovery_test.go |
| +++ b/internal/console/movesrecovery_test.go |
| @@ -800,6 +800,22 @@ func blockMoveLog(t *testing.T, w *movePassWorld) func() { |
| } |
| } |
| +// blockMoveLogEntry makes the write of one entry fail (a directory where its |
| +// object goes) until the returned func is called. |
| +func blockMoveLogEntry(t *testing.T, w *movePassWorld, name string) func() { |
| + t.Helper() |
| + root := w.fx.rel.(*deletableStore).root |
| + p := filepath.Join(root, filepath.FromSlash(moves.LogKeyFor(moves.ReportPrefix("bab", "titogo", "acme.go.tito.io", w.id), name))) |
| + if err := os.MkdirAll(p, 0o755); 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) |
| @@ -867,3 +883,77 @@ func TestRecoveryPassLogOutboxRetries(t *testing.T) { |
| } |
| noPendingLog(t, w) |
| } |
| + |
| +// Fix round 3, ruling 2: claim_released is published only from a successful |
| +// release. The pass queues it unconfirmed before the release and confirms it |
| +// after; the drainer delivers confirmed rows only, and settles an unconfirmed |
| +// one by the claim. |
| +func TestClaimReleasedIsPublishedOnlyAfterTheRelease(t *testing.T) { |
| + closed := func(t *testing.T, code string) *movePassWorld { |
| + w := landingMovePass(t, code) |
| + committedAndOpened(t, w) |
| + w.pass(t) |
| + w.advance(moveCloseGrace + time.Minute) |
| + w.pass(t) // the close |
| + return w |
| + } |
| + t.Run("a release that fails publishes nothing", func(t *testing.T) { |
| + w := closed(t, "code-mvpass-release-fails") |
| + w.fx.srv.moveReleaseFault = func() error { return hostops.ErrPending } |
| + w.pass(t) |
| + if c := w.claim(t); c.Op.Kind == hostclaim.KindServing { |
| + t.Fatal("released despite the fault") |
| + } |
| + if _, ok := moveLogOf(t, w)["claim-released"]; ok { |
| + t.Fatal("claim_released published for a release that failed") |
| + } |
| + if p := pendingLog(t, w); strings.Join(p, " ") != "claim-released(unconfirmed)" { |
| + t.Fatalf("pending %v", p) |
| + } |
| + w.pass(t) // the claim is still held: the row is left for the release |
| + if _, ok := moveLogOf(t, w)["claim-released"]; ok { |
| + t.Fatal("claim_released published while the claim is held") |
| + } |
| + w.fx.srv.moveReleaseFault = nil |
| + w.pass(t) |
| + if ev := moveLogOf(t, w)["claim-released"]; ev.Event != "claim_released" { |
| + t.Fatalf("not published after the release: %+v", moveLogOf(t, w)) |
| + } |
| + noPendingLog(t, w) |
| + }) |
| + t.Run("a crash between the release and its confirmation", func(t *testing.T) { |
| + w := closed(t, "code-mvpass-release-crash") |
| + unblock := blockMoveLogEntry(t, w, "claim-released") |
| + w.pass(t) // the release; its entry is confirmed, its delivery fails |
| + if c := w.claim(t); c.Op.Kind != hostclaim.KindServing { |
| + t.Fatalf("not released: %+v", c.Op) |
| + } |
| + // As if the console stopped after the release and before the confirm. |
| + if _, err := w.fx.st.db.Exec(`UPDATE move_log_outbox SET confirmed = 0 WHERE op = ? AND name = 'claim-released'`, w.id); err != nil { |
| + t.Fatal(err) |
| + } |
| + unblock() |
| + w.advance(2 * time.Hour) |
| + w.pass(t) |
| + if ev := moveLogOf(t, w)["claim-released"]; ev.Event != "claim_released" { |
| + t.Fatalf("the released claim's entry was not confirmed and published: %+v", moveLogOf(t, w)) |
| + } |
| + noPendingLog(t, w) |
| + }) |
| + t.Run("an unconfirmed row whose claim another move holds", func(t *testing.T) { |
| + journal := captureLog(t) |
| + w := closed(t, "code-mvpass-release-other") |
| + body := []byte(`{"v":1}`) |
| + if err := w.fx.st.QueueMoveLog("acme.go.tito.io", "mv_another", "claim-released", moves.ReportPrefix("bab", "titogo", "acme.go.tito.io", "mv_another"), body, false, w.clock); err != nil { |
| + t.Fatal(err) |
| + } |
| + w.pass(t) |
| + got, err := w.fx.st.MoveLogPending("acme.go.tito.io", "mv_another") |
| + if err != nil || len(got) != 0 { |
| + t.Fatalf("pending %+v %v", got, err) |
| + } |
| + if !strings.Contains(journal.String(), "mv_another") { |
| + t.Fatalf("no journal line for the dropped row:\n%s", journal.String()) |
| + } |
| + }) |
| +} |
diff --git a/internal/console/server.go b/internal/console/server.go| index 1e5c08d1d..efe082076 100644 |
| --- a/internal/console/server.go |
| +++ b/internal/console/server.go |
| @@ -837,9 +837,13 @@ type Server struct { |
| // conditional record store, bound by ConfigureClaimRuntime with the |
| // moves switch on; nil otherwise. |
| stampLostMoveTarget func(ctx context.Context, host string, now time.Time) error |
| - dbHandleKey []byte |
| - cfg Config |
| - store *Store |
| + // moveReleaseFault, when set, fails the pass's claim release just before |
| + // it is made (a TEST SEAM: a release that fails after its log entry is |
| + // queued, movesrecovery.go). |
| + moveReleaseFault func() error |
| + dbHandleKey []byte |
| + cfg Config |
| + store *Store |
| // appRegionOf and beforeClaimInsert are TEST SEAMS (nil in |
| // production): a store failure on the app-region read (C20), and a |
| // competing insert landing after the region was resolved. |