amadan
moves log: keep a declared_lost row while the move can still be declared (final re-review 1, I4)
The drainer could drop an unconfirmed declared_lost row from a claim read taken before the declaration committed and confirmed it, deleting the confirmed row. A declaration also went ahead when its row could not be queued. Now: - settleDeclaredLost confirms the row when the claim records a committed loss of the move (loss_refs, then a verified loss record). - It keeps the row, with the usual backoff and one warning, while the claim's op is still the move and not released to serving. - It drops the row only when the claim shows the move closed or released with no such loss, or the instance's incarnation gone. The drop is DropUnconfirmedMoveLog, which deletes only a row still unconfirmed. - declareMoveLossLogged refuses the declaration (fails closed) when the record cannot be read or the row cannot be queued. Regressions: a declaration that commits and confirms while the drainer settles from its earlier read (through a test seam after the claim read) keeps its row and publishes once. A refused declaration's row is kept while the move can still be declared lost and is dropped once the claim is gone. A queue failure refuses the declaration on both doors and nothing is written.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
4 files changed,
+161
−27
internal/console/movesrecovery.go+51 −24internal/console/movesrecovery_test.go+95 −0internal/console/server.go+7 −3internal/console/store.go+8 −0
diff --git a/internal/console/movesrecovery.go b/internal/console/movesrecovery.go| index c75bc7816..d777e43fa 100644 |
| --- a/internal/console/movesrecovery.go |
| +++ b/internal/console/movesrecovery.go |
| @@ -598,17 +598,22 @@ func (s *Server) queueMoveLog(o hostops.ConsoleMoveObservation, t moves.Transiti |
| // (settleUnconfirmed). The row is bound to the incarnation t's evidence |
| // carried, which must be the grant's move and instance. |
| func (s *Server) queueMoveLogAs(o hostops.ConsoleMoveObservation, t moves.Transition, now time.Time, confirmed bool) { |
| + if err := s.queueMoveLogErr(o, t, now, confirmed); err != nil { |
| + log.Printf("console: moves: %s: the log of move %s at %s is lost: %v", o.Host, o.OpID, t.Identity, err) |
| + } |
| +} |
| + |
| +// queueMoveLogErr is queueMoveLogAs, answering whether the row is queued: |
| +// for a caller that must not act unless it is (declareMoveLossLogged). |
| +func (s *Server) queueMoveLogErr(o hostops.ConsoleMoveObservation, t moves.Transition, now time.Time, confirmed bool) error { |
| if t.Op != o.OpID || t.Incarnation != o.Incarnation { |
| - log.Printf("console: moves: %s: the log of move %s at %s names move %s of incarnation %q, not the grant's; not queued", o.Host, o.OpID, t.Identity, t.Op, t.Incarnation) |
| - return |
| + return fmt.Errorf("the entry names move %s of incarnation %q, not move %s of %q", t.Op, t.Incarnation, o.OpID, o.Incarnation) |
| } |
| body, err := moveLogEntry(o, t, now) |
| - if err == nil { |
| - err = s.store.QueueMoveLog(o.Host, o.OpID, t.Identity, t.Incarnation, 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) |
| + return err |
| } |
| + return s.store.QueueMoveLog(o.Host, o.OpID, t.Identity, t.Incarnation, body, confirmed, now) |
| } |
| // confirmMoveLog confirms t's entry once the write it names succeeded. The |
| @@ -629,7 +634,9 @@ func (s *Server) confirmMoveLog(o hostops.ConsoleMoveObservation, t moves.Transi |
| // 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. |
| +// read, or a row that cannot be queued, refuses the declaration: an |
| +// operator's action fails closed rather than leave no retry for its entry |
| +// (final re-review 1, I4). |
| func (s *Server) declareMoveLossLogged(ctx context.Context, account, app, host string, declare func() error) error { |
| var ( |
| o hostops.ConsoleMoveObservation |
| @@ -637,13 +644,17 @@ func (s *Server) declareMoveLossLogged(ctx context.Context, account, app, host s |
| 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 err != nil { |
| + // Fail closed (final re-review 1, I4): a declaration whose entry |
| + // cannot be queued is not made. |
| + return fmt.Errorf("moves: the declared loss cannot be queued in the move log: the record cannot be read: %w", err) |
| + } |
| + if 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 := s.queueMoveLogErr(o, t, s.movesNow(), false); err != nil { |
| + return fmt.Errorf("moves: the declared loss cannot be queued in the move log: %w", err) |
| + } |
| } |
| } |
| if err := declare(); err != nil { |
| @@ -712,19 +723,36 @@ func (s *Server) settleUnconfirmed(ctx context.Context, r MoveLogPendingEntry) ( |
| } |
| // 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). |
| +// I4; final re-review 1) from the claim. A committed loss of the move's |
| +// target, in the claim's loss_refs (which the declaration's claim CAS writes |
| +// as it commits, and nothing removes), confirms the row (deliver). While the |
| +// move can still be declared lost (the claim's op is still the move, not |
| +// released to serving) the row is kept, however long, with the failed |
| +// delivery's backoff and its one warning. It is dropped only once the claim |
| +// shows the move closed or released with no such loss, or the instance's |
| +// incarnation gone (no claim, or another holder); the incarnation's purge |
| +// deletes it too (PurgeMoveLogIncarnation). The drop deletes only a row |
| +// still unconfirmed, so a declaration that confirmed its row after the claim |
| +// was read here keeps it. A claim or loss record that cannot be read waits. |
| 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 { |
| + if f := s.moveLogSettleHook; f != nil { |
| + f() |
| + } |
| + drop := func(why string) (bool, error) { |
| + log.Printf("console: moves: %s: the log of move %s at %s was never confirmed and %s; dropped", r.Host, r.Op, r.Name, why) |
| + if err := s.store.DropUnconfirmedMoveLog(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 |
| + } |
| + if o.Absent || o.Claim == nil || o.Claim.Holder == nil || o.Claim.Holder.Incarnation != r.Incarnation { |
| + return drop("the claim no longer holds the instance it was queued for") |
| + } |
| + if len(o.Claim.LossRefs) > 0 { |
| var pub ed25519.PublicKey |
| if a := s.cfg.ClaimAuthority; a != nil { |
| pub = a.Signer.RootPublicKey() |
| @@ -749,11 +777,10 @@ func (s *Server) settleDeclaredLost(ctx context.Context, r MoveLogPendingEntry) |
| 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) |
| + if o.Claim.Op.ID == r.Op && o.Claim.Op.Kind != hostclaim.KindServing { |
| + return false, errors.New("unconfirmed: no loss of the move's target has committed, and the move can still be declared lost") |
| } |
| - return false, nil |
| + return drop("the claim shows the move closed or released with no loss of its target") |
| } |
| // retryMoveLog moves a row's next attempt on after one that did not settle |
diff --git a/internal/console/movesrecovery_test.go b/internal/console/movesrecovery_test.go| index abcf99f65..bc588c5c1 100644 |
| --- a/internal/console/movesrecovery_test.go |
| +++ b/internal/console/movesrecovery_test.go |
| @@ -17,10 +17,12 @@ import ( |
| "github.com/carlosframework/platform/internal/alerting" |
| "github.com/carlosframework/platform/internal/hostclaim" |
| "github.com/carlosframework/platform/internal/hostfence" |
| + "github.com/carlosframework/platform/internal/hostfloor" |
| "github.com/carlosframework/platform/internal/hostops" |
| "github.com/carlosframework/platform/internal/moves" |
| "github.com/carlosframework/platform/internal/provision" |
| "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/svcrole" |
| ) |
| // The console's move pass (moves Plan 5 Task 7; claim Plan 10's |
| @@ -1327,10 +1329,103 @@ func TestDeclaredLostIsLoggedByTheConsole(t *testing.T) { |
| if p := pendingLog(t, w); strings.Join(p, " ") != "rolling_back-target-lost(unconfirmed)" { |
| t.Fatalf("pending %v, want the declaration's row unconfirmed", p) |
| } |
| + // Final re-review 1, I4: the move can still be declared lost, so the |
| + // row is kept, unpublished, however many passes run. |
| + w.pass(t) |
| + w.advance(2 * time.Hour) |
| w.pass(t) |
| if n := declaredLostEntries(t, w); n != 0 { |
| t.Fatalf("a refused declaration published %d entries", n) |
| } |
| + if p := pendingLog(t, w); strings.Join(p, " ") != "rolling_back-target-lost(unconfirmed)" { |
| + t.Fatalf("pending %v, want the row kept while the move can still be declared lost", p) |
| + } |
| + // The claim gone (the incarnation retired): the row is dropped. |
| + key, err := hostclaim.Key("acme.go.tito.io") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := w.fx.rel.Delete(context.Background(), key); err != nil { |
| + t.Fatal(err) |
| + } |
| + w.advance(2 * time.Hour) |
| + w.pass(t) |
| + noPendingLog(t, w) |
| + }) |
| + // Final re-review 1, I4: the drainer reads the claim with no loss in it; |
| + // the declaration then commits and confirms its row; the drainer, going on |
| + // from its earlier read, must not drop the row. It survives and is |
| + // published once. |
| + t.Run("a declaration that commits while the drainer settles its row", func(t *testing.T) { |
| + w, token, moveURL, sub := declare(t, "code-mvlog-declared-race") |
| + rec := w.record(t) |
| + e, ok := moves.DeclaredLost(rec.Move) |
| + if !ok { |
| + t.Fatal("no entry for the fixture's move") |
| + } |
| + // A declaration in flight: its row queued, its claim CAS not yet. |
| + o := hostops.ConsoleMoveObservation{Host: rec.Host, OpID: rec.Move.ID, Incarnation: rec.Move.Incarnation} |
| + w.fx.srv.queueMoveLogAs(o, e, w.fx.srv.movesNow(), false) |
| + declared := false |
| + w.fx.srv.moveLogSettleHook = func() { |
| + if declared { |
| + return |
| + } |
| + declared = true |
| + 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.Errorf("declare = %d %s", c, out) |
| + } |
| + } |
| + w.pass(t) |
| + if !declared { |
| + t.Fatal("the drainer never settled the row") |
| + } |
| + if p := pendingLog(t, w); strings.Join(p, " ") != "rolling_back-target-lost" { |
| + t.Fatalf("pending %v: the row the declaration confirmed was dropped", p) |
| + } |
| + w.advance(2 * time.Hour) |
| + 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) |
| }) |
| + // Final re-review 1, I4: a declaration whose row cannot be queued is |
| + // refused, through either door, and nothing is declared. |
| + t.Run("a row that cannot be queued refuses the declaration", func(t *testing.T) { |
| + for _, door := range []string{"move", "host"} { |
| + t.Run(door, func(t *testing.T) { |
| + w, token, moveURL, sub := declare(t, "code-mvlog-declared-noqueue-"+door) |
| + url, body := moveURL+"/resolve", j4Body(t, map[string]any{"declare_lost": "uk-box", "submission": sub}) |
| + if door == "host" { |
| + url, body = j4HostDoor(t, w.fx), j4Body(t, map[string]any{"action": "declare-lost", "submission": sub}) |
| + } |
| + before := j4World(t, w.fx) |
| + if _, err := w.fx.st.db.Exec(`DROP TABLE move_log_outbox`); err != nil { |
| + t.Fatal(err) |
| + } |
| + if c, _, out := rawReq(t, http.MethodPost, url, token, body); c == http.StatusAccepted || c == http.StatusOK { |
| + t.Fatalf("declare with no outbox = %d %s, want a refusal", c, out) |
| + } |
| + if j4World(t, w.fx) != before { |
| + t.Fatal("a declaration whose log row could not be queued wrote") |
| + } |
| + }) |
| + } |
| + }) |
| +} |
| + |
| +// j4HostDoor registers the host operator door on fx (NewServer registers it |
| +// only with the host claim on at build) and returns its resolve URL for the |
| +// j4 host. |
| +func j4HostDoor(t *testing.T, fx *consoleFixture) string { |
| + t.Helper() |
| + fx.srv.cfg.Bucket = moveTestBucket |
| + fx.srv.cliRoute("POST /api/cli/ops/hosts/{host}/resolve/inspect", svcrole.Set{}, fx.srv.handleAPIHostResolveInspect) |
| + fx.srv.cliRoute("POST /api/cli/ops/hosts/{host}/resolve", svcrole.Set{}, fx.srv.handleAPIHostResolve) |
| + if err := hostfloor.Publish(context.Background(), fx.rel, release.RootSigner(fx.priv), moveTestBucket, hostfloor.Floor{Min: 1, MinRelease: "v1.0.0"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + return fx.ts.URL + "/api/cli/ops/hosts/" + j4Host + "/resolve" |
| } |
diff --git a/internal/console/server.go b/internal/console/server.go| index 2d6bd3e7f..a3fa0cfd6 100644 |
| --- a/internal/console/server.go |
| +++ b/internal/console/server.go |
| @@ -841,9 +841,13 @@ type Server struct { |
| // 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 |
| + // moveLogSettleHook, when set, runs as the drainer settles an |
| + // unconfirmed declared_lost row, after it read the claim (a TEST SEAM: |
| + // a declaration that commits in between, movesrecovery.go). |
| + moveLogSettleHook func() |
| + 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. |
diff --git a/internal/console/store.go b/internal/console/store.go| index 8bc3af1c9..e301a3ebc 100644 |
| --- a/internal/console/store.go |
| +++ b/internal/console/store.go |
| @@ -2327,6 +2327,14 @@ func (s *Store) SettleMoveLog(host, op, name string) error { |
| return err |
| } |
| +// DropUnconfirmedMoveLog deletes a queued entry only while it is still |
| +// unconfirmed: a confirmation that landed since the caller read the row |
| +// keeps it. |
| +func (s *Store) DropUnconfirmedMoveLog(host, op, name string) error { |
| + _, err := s.db.Exec(`DELETE FROM move_log_outbox WHERE host = ? AND op = ? AND name = ? AND confirmed = 0`, host, op, name) |
| + return err |
| +} |
| + |
| // ---- grants (the console's Keymail access tokens) --------------------------- |
| // SetGrantToken stores the Keymail access token the user's OAuth approval |