amadan
moves log: the console that declares a lost target logs declared_lost (final review I4)
FinishLoss moves the target from lost on to released within the declaration's own request, so a box pass usually never sees it lost and the event was missed. Both console doors that declare a move's lost target (the move resolve door and the host operator door) now log it through the console's move_log_outbox in claim_released's two phases: queued unconfirmed before the declaration, confirmed after it succeeds (queued again confirmed, in case a pass dropped the row in between). The drainer settles an unconfirmed row from the claim's loss_refs, which the declaration's claim CAS writes as it commits. A matching signed loss record confirms the row. A claim or loss record that cannot be read waits. Anything else is dropped with a journal line. moves.DeclaredLost names the entry for both producers, so the box's pass, which still logs a lost target it sees, writes the same identity, and create-only storage keeps one.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
6 files changed,
+235
−6
internal/console/apihostclaim.go+3 −1internal/console/apiinstancemove.go+3 −1internal/console/movesrecovery.go+98 −1internal/console/movesrecovery_test.go+98 −0internal/host/movesevents.go+5 −3internal/moves/transition.go+28 −0
diff --git a/internal/console/apihostclaim.go b/internal/console/apihostclaim.go| index 3886970ae..e3f230e1e 100644 |
| --- a/internal/console/apihostclaim.go |
| +++ b/internal/console/apihostclaim.go |
| @@ -120,7 +120,9 @@ func (s *Server) handleAPIHostResolve(w http.ResponseWriter, r *http.Request) { |
| // is the signed challenge's. |
| if in.Action == "declare-lost" { |
| if account, app, ok := s.claimMoveLossHolder(r, r.PathValue("host")); ok { |
| - if err := s.cfg.ClaimAuthority.DeclareMoveLoss(r.Context(), actor, account, app, r.PathValue("host"), s.claimChallengeBox(in.Submission), in.Submission); err != nil { |
| + if err := s.declareMoveLossLogged(r.Context(), account, app, r.PathValue("host"), func() error { |
| + return s.cfg.ClaimAuthority.DeclareMoveLoss(r.Context(), actor, account, app, r.PathValue("host"), s.claimChallengeBox(in.Submission), in.Submission) |
| + }); err != nil { |
| writeClaimResponseError(w, err) |
| return |
| } |
diff --git a/internal/console/apiinstancemove.go b/internal/console/apiinstancemove.go| index 90cf4ec3a..8a9e4e48b 100644 |
| --- a/internal/console/apiinstancemove.go |
| +++ b/internal/console/apiinstancemove.go |
| @@ -1072,7 +1072,9 @@ func (s *Server) claimMoveLoss(w http.ResponseWriter, r *http.Request, acct *Acc |
| writeJSON(w, http.StatusOK, in) |
| return |
| } |
| - if err := a.DeclareMoveLoss(ctx, actor, acct.Sqid, app, host, box, *sub); err != nil { |
| + if err := s.declareMoveLossLogged(ctx, acct.Sqid, app, host, func() error { |
| + return a.DeclareMoveLoss(ctx, actor, acct.Sqid, app, host, box, *sub) |
| + }); err != nil { |
| claimMoveLossFailed(w, acct, app, host, err) |
| return |
| } |
diff --git a/internal/console/movesrecovery.go b/internal/console/movesrecovery.go| index 074e04689..c75bc7816 100644 |
| --- a/internal/console/movesrecovery.go |
| +++ b/internal/console/movesrecovery.go |
| @@ -2,6 +2,7 @@ package console |
| import ( |
| "context" |
| + "crypto/ed25519" |
| "crypto/sha256" |
| "encoding/hex" |
| "encoding/json" |
| @@ -549,7 +550,11 @@ func moveAbortOverduePage(host, id, to, box string) (subject, body string) { |
| // 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. |
| +// release, or a crash before the confirm) from the claim. A declared loss of |
| +// a move's target (declared_lost) is the console's too, from either door |
| +// that performs the declaration, in the same two phases (final review I4, |
| +// declareMoveLossLogged); the drainer settles its unconfirmed row from the |
| +// loss the claim records (settleDeclaredLost). |
| // |
| // 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 |
| @@ -606,6 +611,50 @@ func (s *Server) queueMoveLogAs(o hostops.ConsoleMoveObservation, t moves.Transi |
| } |
| } |
| +// confirmMoveLog confirms t's entry once the write it names succeeded. The |
| +// row is queued again, confirmed, in case a pass dropped the unconfirmed one |
| +// in between (settleDeclaredLost drops a row whose loss is not committed |
| +// yet), and then marked confirmed if it was there already. |
| +func (s *Server) confirmMoveLog(o hostops.ConsoleMoveObservation, t moves.Transition, now time.Time) { |
| + s.queueMoveLogAs(o, t, now, true) |
| + if err := s.store.ConfirmMoveLog(o.Host, o.OpID, t.Identity); err != nil { |
| + log.Printf("console: moves: %s: the log of move %s at %s is not confirmed in its log outbox (the drainer settles it): %v", o.Host, o.OpID, t.Identity, err) |
| + } |
| +} |
| + |
| +// declareMoveLossLogged runs one declaration of a move's lost target |
| +// (declare, the configured authority's DeclareMoveLoss) with its move log |
| +// entry, declared_lost (final review I4): queued unconfirmed before the |
| +// declaration, from the record as the door reads it, and confirmed once the |
| +// declaration succeeds. The declaration moves the target on from lost to |
| +// released within the request, so the box's pass may never see it lost; the |
| +// console that performed it is the entry's producer. A record that cannot be |
| +// read leaves the declaration unlogged here, with a journal line. |
| +func (s *Server) declareMoveLossLogged(ctx context.Context, account, app, host string, declare func() error) error { |
| + var ( |
| + o hostops.ConsoleMoveObservation |
| + t moves.Transition |
| + logged bool |
| + ) |
| + rec, err := provision.GetRecord(ctx, s.cfg.Releases, account, app, host) |
| + switch { |
| + case err != nil: |
| + log.Printf("console: moves: %s: the declared loss is not queued in the move log: the record cannot be read: %v", host, err) |
| + case rec.Move != nil: |
| + if t, logged = moves.DeclaredLost(rec.Move); logged { |
| + o = hostops.ConsoleMoveObservation{Host: host, OpID: rec.Move.ID, Incarnation: rec.Move.Incarnation} |
| + s.queueMoveLogAs(o, t, s.movesNow(), false) |
| + } |
| + } |
| + if err := declare(); err != nil { |
| + return err |
| + } |
| + if logged { |
| + s.confirmMoveLog(o, t, s.movesNow()) |
| + } |
| + return nil |
| +} |
| + |
| // drainMoveLog is the Recovery.MoveLogs hook: every due row of the outbox, |
| // delivered through the pass's writer. |
| func (s *Server) drainMoveLog(ctx context.Context, lw *hostops.MoveLogWriter) error { |
| @@ -639,6 +688,9 @@ func (s *Server) drainMoveLog(ctx context.Context, lw *hostops.MoveLogWriter) er |
| // it rotates out of the batch); anything else, it is dropped with a journal |
| // line. |
| func (s *Server) settleUnconfirmed(ctx context.Context, r MoveLogPendingEntry) (deliver bool, wait error) { |
| + if r.Name == moves.DeclaredLostIdentity { |
| + return s.settleDeclaredLost(ctx, r) |
| + } |
| o, err := hostclaim.Read(ctx, s.cfg.Releases, r.Host) |
| if err != nil { |
| return false, fmt.Errorf("unconfirmed, and the claim cannot be read: %w", err) |
| @@ -659,6 +711,51 @@ func (s *Server) settleUnconfirmed(ctx context.Context, r MoveLogPendingEntry) ( |
| return false, nil |
| } |
| +// settleDeclaredLost settles an unconfirmed declared_lost row (final review |
| +// I4) from the committed loss evidence the claim keeps: its loss_refs, which |
| +// the declaration's claim CAS writes as it commits and nothing removes. A |
| +// loss there whose signed record names the row's host, move and incarnation |
| +// confirms the row (deliver). A claim or loss record that cannot be read now |
| +// waits. Anything else is dropped with a journal line: no declaration of the |
| +// move has committed, and a later one, through either door, queues its entry |
| +// again and confirms it (confirmMoveLog). |
| +func (s *Server) settleDeclaredLost(ctx context.Context, r MoveLogPendingEntry) (deliver bool, wait error) { |
| + o, err := hostclaim.Read(ctx, s.cfg.Releases, r.Host) |
| + if err != nil { |
| + return false, fmt.Errorf("unconfirmed, and the claim cannot be read: %w", err) |
| + } |
| + if !o.Absent && o.Claim != nil && len(o.Claim.LossRefs) > 0 { |
| + var pub ed25519.PublicKey |
| + if a := s.cfg.ClaimAuthority; a != nil { |
| + pub = a.Signer.RootPublicKey() |
| + } |
| + if len(pub) != ed25519.PublicKeySize { |
| + return false, errors.New("unconfirmed, and there is no key to read the claim's loss records") |
| + } |
| + for _, id := range o.Claim.LossRefs { |
| + l, err := hostclaim.ReadLossRecord(ctx, s.cfg.Releases, pub, id) |
| + if errors.Is(err, hostclaim.ErrLossRecordAbsent) || errors.Is(err, hostclaim.ErrLossRecord) { |
| + continue |
| + } |
| + if err != nil { |
| + return false, fmt.Errorf("unconfirmed, and a loss record cannot be read: %w", err) |
| + } |
| + if l.Record.Host != r.Host || l.Record.OpID != r.Op || l.Record.HolderIncarnation != r.Incarnation { |
| + continue |
| + } |
| + if err := s.store.ConfirmMoveLog(r.Host, r.Op, r.Name); err != nil { |
| + return false, fmt.Errorf("unconfirmed: %w", err) |
| + } |
| + return true, nil |
| + } |
| + } |
| + log.Printf("console: moves: %s: the log of move %s at %s was never confirmed and the claim records no committed loss of 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, nil |
| +} |
| + |
| // retryMoveLog moves a row's next attempt on after one that did not settle |
| // it (why), and warns of it once in the journal when it is an hour old. |
| func (s *Server) retryMoveLog(r MoveLogPendingEntry, now time.Time, why error) { |
diff --git a/internal/console/movesrecovery_test.go b/internal/console/movesrecovery_test.go| index 2a08ec5db..abcf99f65 100644 |
| --- a/internal/console/movesrecovery_test.go |
| +++ b/internal/console/movesrecovery_test.go |
| @@ -6,6 +6,7 @@ import ( |
| "encoding/json" |
| "errors" |
| "fmt" |
| + "net/http" |
| "os" |
| "path/filepath" |
| "strings" |
| @@ -15,6 +16,7 @@ import ( |
| "github.com/carlosframework/platform/internal/alerting" |
| "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/hostfence" |
| "github.com/carlosframework/platform/internal/hostops" |
| "github.com/carlosframework/platform/internal/moves" |
| "github.com/carlosframework/platform/internal/provision" |
| @@ -1236,3 +1238,99 @@ func TestARefusedConfirmedRowIsHeldNotDeleted(t *testing.T) { |
| t.Fatalf("the incarnation's purge left %+v", got) |
| } |
| } |
| + |
| +// declaredLostEntries counts the move's log objects named |
| +// rolling_back-target-lost. |
| +func declaredLostEntries(t *testing.T, w *movePassWorld) int { |
| + t.Helper() |
| + got, err := moves.ReadLog(context.Background(), w.fx.rel, moves.ReportPrefix("bab", "titogo", "acme.go.tito.io", w.id)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + n := 0 |
| + for _, ev := range got { |
| + if ev.Identity == "rolling_back-target-lost" { |
| + n++ |
| + } |
| + } |
| + return n |
| +} |
| + |
| +// Final review I4: declared_lost is logged by the console that performs the |
| +// declaration, as claim_released is: queued unconfirmed before the |
| +// declaration, confirmed once it succeeds, and a row left unconfirmed is |
| +// settled by a later pass from the loss the claim records. The declaration |
| +// moves the target on to released within the request, so no box pass ever |
| +// sees it lost; the event is still logged, once. |
| +func TestDeclaredLostIsLoggedByTheConsole(t *testing.T) { |
| + declare := func(t *testing.T, code string) (*movePassWorld, string, string, hostfence.Submission) { |
| + t.Helper() |
| + fx, token, moveURL := j4LandingMove(t, code) |
| + w := newMovePassWorld(t, fx, j17MoveState(t, fx).ID) |
| + c, _, out := rawReq(t, http.MethodPost, moveURL+"/resolve", token, `{"declare_lost":"uk-box"}`) |
| + if c != http.StatusOK { |
| + t.Fatalf("inspect = %d %s", c, out) |
| + } |
| + return w, token, moveURL, j4Submission(t, fx, out) |
| + } |
| + t.Run("the declaration completes before any box pass", func(t *testing.T) { |
| + w, token, moveURL, sub := declare(t, "code-mvlog-declared") |
| + body := j4Body(t, map[string]any{"declare_lost": "uk-box", "submission": sub}) |
| + if c, _, out := rawReq(t, http.MethodPost, moveURL+"/resolve", token, body); c != http.StatusAccepted { |
| + t.Fatalf("declare = %d %s", c, out) |
| + } |
| + if rb := w.record(t).Move.Rollback; rb == nil || rb.TargetPhase != provision.RbTargetReleased { |
| + t.Fatalf("the declaration did not move the target on: %+v", rb) |
| + } |
| + w.pass(t) |
| + ev := moveLogOf(t, w)["rolling_back-target-lost"] |
| + if ev.Event != "declared_lost" || ev.Actor != "console" || ev.Op != w.id || ev.Fields["from"] != "ie" || ev.Fields["to"] != "uk" { |
| + t.Fatalf("declared_lost %+v in %+v", ev, moveLogOf(t, w)) |
| + } |
| + // A replay of the same declaration, and more passes, log nothing more. |
| + if c, _, out := rawReq(t, http.MethodPost, moveURL+"/resolve", token, body); c != http.StatusAccepted { |
| + t.Fatalf("replay = %d %s", c, out) |
| + } |
| + w.pass(t) |
| + w.pass(t) |
| + if n := declaredLostEntries(t, w); n != 1 { |
| + t.Fatalf("%d declared_lost entries, want one", n) |
| + } |
| + noPendingLog(t, w) |
| + }) |
| + t.Run("a crash between the declaration and its confirmation", func(t *testing.T) { |
| + w, token, moveURL, sub := declare(t, "code-mvlog-declared-crash") |
| + if c, _, out := rawReq(t, http.MethodPost, moveURL+"/resolve", token, j4Body(t, map[string]any{"declare_lost": "uk-box", "submission": sub})); c != http.StatusAccepted { |
| + t.Fatalf("declare = %d %s", c, out) |
| + } |
| + // As if the console stopped after the declaration and before the |
| + // confirm. |
| + res, err := w.fx.st.db.Exec(`UPDATE move_log_outbox SET confirmed = 0 WHERE op = ? AND name = 'rolling_back-target-lost'`, w.id) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if n, _ := res.RowsAffected(); n != 1 { |
| + t.Fatalf("the declaration queued %d rows, want one", n) |
| + } |
| + w.pass(t) |
| + if ev := moveLogOf(t, w)["rolling_back-target-lost"]; ev.Event != "declared_lost" { |
| + t.Fatalf("the committed loss's entry was not settled and published: %+v", moveLogOf(t, w)) |
| + } |
| + noPendingLog(t, w) |
| + }) |
| + t.Run("a declaration that is refused publishes nothing", func(t *testing.T) { |
| + w, token, moveURL, sub := declare(t, "code-mvlog-declared-stale") |
| + j17SetRecord(t, w.fx, func(r *provision.Record) { r.Move.LandingPhase = provision.LandProbed }) |
| + if c, _, out := rawReq(t, http.MethodPost, moveURL+"/resolve", token, j4Body(t, map[string]any{"declare_lost": "uk-box", "submission": sub})); c != http.StatusConflict { |
| + t.Fatalf("stale = %d %s", c, out) |
| + } |
| + if p := pendingLog(t, w); strings.Join(p, " ") != "rolling_back-target-lost(unconfirmed)" { |
| + t.Fatalf("pending %v, want the declaration's row unconfirmed", p) |
| + } |
| + w.pass(t) |
| + if n := declaredLostEntries(t, w); n != 0 { |
| + t.Fatalf("a refused declaration published %d entries", n) |
| + } |
| + noPendingLog(t, w) |
| + }) |
| +} |
diff --git a/internal/host/movesevents.go b/internal/host/movesevents.go| index 9898423b3..4e3d0862b 100644 |
| --- a/internal/host/movesevents.go |
| +++ b/internal/host/movesevents.go |
| @@ -99,9 +99,11 @@ func (w *movesWorker) observeMove(ctx context.Context, s release.Store, rec prov |
| m.State == provision.MoveCutting && m.CutPhase == provision.CutRequested: |
| w.logTransition(ctx, s, provision.Record{}, rec) |
| case m.State == provision.MoveRollingBack && m.Rollback != nil && m.Rollback.TargetPhase == provision.RbTargetLost: |
| - before := provision.Record{Move: copyMove(m)} |
| - before.Move.Rollback.TargetPhase = "" |
| - w.logTransition(ctx, s, before, rec) |
| + // The console that performed the declaration logs it first (final |
| + // review I4); this is the second producer of the same identity. |
| + if e, ok := moves.DeclaredLost(m); ok { |
| + w.record(ctx, s, rec, e) |
| + } |
| } |
| if m.RecoveryRetryEpoch > 0 && m.State != provision.MoveStuck { |
| w.record(ctx, s, rec, moves.Transition{Op: m.ID, Event: "retry", Identity: "retry-" + strconv.FormatUint(m.RecoveryRetryEpoch, 10), |
diff --git a/internal/moves/transition.go b/internal/moves/transition.go| index a682af150..a71fc31fb 100644 |
| --- a/internal/moves/transition.go |
| +++ b/internal/moves/transition.go |
| @@ -75,6 +75,34 @@ func seconds(a, b interface{ Unix() int64 }) string { |
| return strconv.FormatInt(b.Unix()-a.Unix(), 10) |
| } |
| +// DeclaredLostIdentity is the identity of a declared loss's entry: the |
| +// rollback's target phase going to lost (TransitionOf), whoever logs it. |
| +const DeclaredLostIdentity = "rolling_back-target-" + provision.RbTargetLost |
| + |
| +// DeclaredLost is the entry of a declared loss of m's target (§7.10), event |
| +// declared_lost, named as TransitionOf names the rollback's target going to |
| +// lost, from whatever state the record is in when the writer names it: the |
| +// console that performs the declaration names it from the record it read |
| +// before (a landing, or a rollback whose target is releasing or returning), |
| +// and a box whose pass sees the target lost names it from that record. The |
| +// identity, DeclaredLostIdentity, is the same for both, so create-only |
| +// storage keeps one (final review I4). ok is false for a move with no id. |
| +func DeclaredLost(m *provision.MoveState) (Transition, bool) { |
| + if m == nil || m.ID == "" { |
| + return Transition{}, false |
| + } |
| + var rb provision.MoveRollback |
| + if m.Rollback != nil { |
| + rb = *m.Rollback |
| + } |
| + before, after := *m, *m |
| + before.State, after.State = provision.MoveRollingBack, provision.MoveRollingBack |
| + brb, arb := rb, rb |
| + brb.TargetPhase, arb.TargetPhase = "", provision.RbTargetLost |
| + before.Rollback, after.Rollback = &brb, &arb |
| + return TransitionOf(provision.Record{Move: &before}, provision.Record{Move: &after}) |
| +} |
| + |
| // TransitionOf names the transition from before to after, one record write: |
| // before as the write found it, after as written. ok is false when the |
| // write changed nothing the log records (a pin, recovery inputs, a replay). |