amadan

carlos / platform Public

moves-phase1-6 Task 3 final fix (1): hold, never delete, a confirmed row the writer refuses

A confirmed move_log_outbox row whose write the writer refuses (its
incarnation or record no longer vouches for it: a later move's close, a
recreation, a retirement outside the console's door) is held: its bytes kept,
no longer retried, one journal warning, until its incarnation's purge removes
it. The writer still refuses the write. TestARefusedConfirmedRowIsHeldNotDeleted;
TestMoveLogRowsAreBoundToTheirIncarnation and the retire test now expect held
rows, and the retire purges a held row.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude eb8448080ff697486406fed899356182409bb7af parent 979b6a7
4 files changed, +125 −21
  • internal/console/movelog_retire_test.go +11 −0
  • internal/console/movesrecovery.go +12 −3
  • internal/console/movesrecovery_test.go +82 −7
  • internal/console/store.go +20 −11
diff --git a/internal/console/movelog_retire_test.go b/internal/console/movelog_retire_test.go
index f169d3528..3ebe550c1 100644
--- a/internal/console/movelog_retire_test.go
+++ b/internal/console/movelog_retire_test.go
@@ -55,6 +55,17 @@ func TestRetiringAnInstancePurgesItsMoveLogRows(t *testing.T) {
inc := holderIncarnation(t, cx, "acme.go.tito.io")
queueTestMoveLog(t, cx.st, "acme.go.tito.io", "mv_retired", "done", inc)
queueTestMoveLog(t, cx.st, "acme.go.tito.io", "mv_earlier", "done", "an-earlier-incarnation")
+ // A row the writer refuses (the record never named its move) is held,
+ // not deleted (final fix, ruling 1): the drainer runs once, as the moves
+ // switch would wire it, and the row stays, held, for the retire's purge.
+ r := *cx.srv.cfg.HostClaimRecovery
+ r.MoveLogs = cx.srv.drainMoveLog
+ if err := cx.srv.cfg.ClaimAuthority.Control.RecoveryPass(context.Background(), r); err != nil {
+ t.Logf("the recovery pass: %v", err)
+ }
+ if got, err := cx.st.MoveLogPending("acme.go.tito.io", "mv_retired"); err != nil || len(got) != 1 || !got[0].Held {
+ t.Fatalf("the refused row is not held: %+v %v", got, err)
+ }
if status := claimRetireDoor("/api/cli/apps/bab/titogo/instances/acme.go.tito.io", "acme.go.tito.io")(t, cx); status != http.StatusAccepted {
t.Fatalf("delete answered %d", status)
}
diff --git a/internal/console/movesrecovery.go b/internal/console/movesrecovery.go
index a231b261f..0e6583b91 100644
--- a/internal/console/movesrecovery.go
+++ b/internal/console/movesrecovery.go
@@ -569,7 +569,9 @@ func moveAbortOverduePage(host, id, to, box string) (subject, body string) {
// independent of any grant (spent by the time the entry exists) and of the
// claim's release, which never waits for the log. A row whose write fails
// is retried with backoff, indefinitely, and warned of once in the journal
-// when it is an hour old. A release of a start never applied is not logged:
+// when it is an hour old. A confirmed row the writer refuses outright is
+// held, never deleted: kept, not retried and warned of once, until its
+// incarnation is retired or purged (PurgeMoveLogIncarnation). A release of a start never applied is not logged:
// the record never named that move.
// moveLogWarnAfter is the age at which an undelivered entry is warned of.
@@ -697,8 +699,15 @@ func (s *Server) deliverMoveLog(ctx context.Context, lw *hostops.MoveLogWriter,
log.Printf("console: moves: %s: the log of move %s at %s was written by another writer first", r.Host, r.Op, r.Name)
settle()
case errors.Is(err, hostops.ErrIdentity):
- log.Printf("console: moves: %s: the log of move %s at %s can never be written, and is dropped: %v", r.Host, r.Op, r.Name, err)
- settle()
+ // The writer refuses it (its claim's incarnation or record no
+ // longer vouches for it: a later move's close, a recreation, a
+ // retirement outside the console's door). The row is never deleted
+ // for that: it is held, its bytes kept and no longer retried, until
+ // its incarnation's purge (final fix, ruling 1).
+ log.Printf("console: moves: warning: %s: the log of move %s at %s is held: the writer refuses it (%v); its entry is kept until its incarnation is purged", r.Host, r.Op, r.Name, err)
+ if herr := s.store.HoldMoveLog(r.Host, r.Op, r.Name); herr != nil {
+ log.Printf("console: moves: %s: the log outbox of move %s: %v", r.Host, r.Op, herr)
+ }
default:
s.retryMoveLog(r, now, err)
}
diff --git a/internal/console/movesrecovery_test.go b/internal/console/movesrecovery_test.go
index 560c4e03a..38468fc0a 100644
--- a/internal/console/movesrecovery_test.go
+++ b/internal/console/movesrecovery_test.go
@@ -842,6 +842,9 @@ func pendingLog(t *testing.T, w *movePassWorld) []string {
if !e.Confirmed {
n += "(unconfirmed)"
}
+ if e.Held {
+ n += "(held)"
+ }
names = append(names, n)
}
return names
@@ -1107,14 +1110,34 @@ func TestMoveLogRowsAreBoundToTheirIncarnation(t *testing.T) {
if forged, err := moves.ReadLog(context.Background(), w.fx.rel, moves.ReportPrefix("bab", "titogo", "acme.go.tito.io", "mv_forged")); err != nil || len(forged) != 0 {
t.Fatalf("a move the record never named was logged: %+v %v", forged, err)
}
- for _, op := range []string{w.id, "mv_forged"} {
- for _, h := range []string{"acme.go.tito.io", "kass.carlosapp.dev"} {
- if got, err := w.fx.st.MoveLogPending(h, op); err != nil || len(got) != 0 {
- t.Fatalf("%s %s still pending: %+v %v", h, op, got, err)
+ // Final fix, ruling 1: a confirmed row the writer refuses is never
+ // deleted; it is held, with its bytes, until its incarnation's purge.
+ held := func(host, op string) []string {
+ t.Helper()
+ got, err := w.fx.st.MoveLogPending(host, op)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var names []string
+ for _, e := range got {
+ if !e.Held {
+ t.Fatalf("%s %s %s refused but not held: %+v", host, op, e.Name, e)
}
+ names = append(names, e.Name)
}
+ return names
}
- // Retirement: once the instance's claim is gone, its rows are dropped.
+ if h := held("acme.go.tito.io", w.id); strings.Join(h, " ") != "stale" {
+ t.Fatalf("held %v, want the stale incarnation's row", h)
+ }
+ if h := held("acme.go.tito.io", "mv_forged"); strings.Join(h, " ") != "forged" {
+ t.Fatalf("held %v, want the forged move's row", h)
+ }
+ if h := held("kass.carlosapp.dev", w.id); strings.Join(h, " ") != "elsewhere" {
+ t.Fatalf("held %v, want the other app's host's row", h)
+ }
+ // A claim gone (retired outside the console's door): the instance's own
+ // row is held too, never written.
queue("acme.go.tito.io", w.id, "retired", inc)
claimKey, err := hostclaim.Key("acme.go.tito.io")
if err != nil {
@@ -1126,7 +1149,59 @@ func TestMoveLogRowsAreBoundToTheirIncarnation(t *testing.T) {
if err := w.passErr(); err != nil {
t.Fatal(err)
}
- if got, err := w.fx.st.MoveLogPending("acme.go.tito.io", w.id); err != nil || len(got) != 0 {
- t.Fatalf("a retired instance's rows still pending: %+v %v", got, err)
+ if h := held("acme.go.tito.io", w.id); strings.Join(h, " ") != "stale retired" {
+ t.Fatalf("held %v", h)
+ }
+ if moveLogOf(t, w)["retired"].Identity != "" {
+ t.Fatal("a row whose claim is gone was written")
+ }
+}
+
+// Final fix, ruling 1: a confirmed row whose delivery failed, and whose move
+// a later move then replaced as the record's last, is refused by the writer
+// (the record names neither) and held: never written, never retried, warned
+// of once, its bytes kept until its incarnation's purge.
+func TestARefusedConfirmedRowIsHeldNotDeleted(t *testing.T) {
+ journal := captureLog(t)
+ w := landingMovePass(t, "code-mvpass-held")
+ committedAndOpened(t, w)
+ w.pass(t)
+ w.advance(moveCloseGrace + time.Minute)
+ unblock := blockMoveLog(t, w)
+ w.pass(t) // the close; its entry's delivery fails
+ w.pass(t) // the release
+ inc := w.claim(t).Holder.Incarnation
+ j17SetRecord(t, w.fx, func(r *provision.Record) {
+ r.LastMove = &provision.MoveOutcome{ID: "mv_later", Outcome: "done", At: time.Now().UTC()}
+ })
+ unblock()
+ for i := 0; i < 3; i++ {
+ w.advance(2 * time.Hour)
+ w.pass(t)
+ }
+ if _, ok := moveLogOf(t, w)["done"]; ok {
+ t.Fatal("a row whose move the record no longer names was written")
+ }
+ got, err := w.fx.st.MoveLogPending("acme.go.tito.io", w.id)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var done *MoveLogPendingEntry
+ for i := range got {
+ if got[i].Name == "done" {
+ done = &got[i]
+ }
+ }
+ if done == nil || !done.Held || !done.Confirmed || len(done.Body) == 0 {
+ t.Fatalf("the refused row was not kept and held: %+v", got)
+ }
+ if n := strings.Count(journal.String(), "the log of move "+w.id+" at done is held"); n != 1 {
+ t.Fatalf("%d held warnings, want one:\n%s", n, journal.String())
+ }
+ if err := w.fx.st.PurgeMoveLogIncarnation("acme.go.tito.io", inc); err != nil {
+ t.Fatal(err)
+ }
+ if got, _ := w.fx.st.MoveLogPending("acme.go.tito.io", w.id); len(got) != 0 {
+ t.Fatalf("the incarnation's purge left %+v", got)
}
}
diff --git a/internal/console/store.go b/internal/console/store.go
index 5d955744c..8bc3af1c9 100644
--- a/internal/console/store.go
+++ b/internal/console/store.go
@@ -369,16 +369,17 @@ func (s *Store) migrate() error {
// gets. This is the table's only definition: it exists only on this
// unmerged branch, so no console database has an older shape.
`CREATE TABLE IF NOT EXISTS move_log_outbox (
- host TEXT NOT NULL,
- op TEXT NOT NULL,
+ host TEXT NOT NULL,
+ op TEXT NOT NULL,
name TEXT NOT NULL,
incarnation TEXT NOT NULL,
body BLOB NOT NULL,
- queued_at INTEGER NOT NULL,
- confirmed INTEGER NOT NULL,
- attempts INTEGER NOT NULL DEFAULT 0,
- next_at INTEGER NOT NULL DEFAULT 0,
- warned INTEGER NOT NULL DEFAULT 0,
+ queued_at INTEGER NOT NULL,
+ confirmed INTEGER NOT NULL,
+ attempts INTEGER NOT NULL DEFAULT 0,
+ next_at INTEGER NOT NULL DEFAULT 0,
+ warned INTEGER NOT NULL DEFAULT 0,
+ held INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (host, op, name)
);`,
} {
@@ -2240,7 +2241,7 @@ type MoveLogPendingEntry struct {
Host, Op, Name, Incarnation string
Body []byte
QueuedAt time.Time
- Confirmed, Warned bool
+ Confirmed, Warned, Held bool
Attempts int
}
@@ -2265,7 +2266,7 @@ func scanMoveLogRows(rows *sql.Rows) ([]MoveLogPendingEntry, error) {
for rows.Next() {
var e MoveLogPendingEntry
var at int64
- if err := rows.Scan(&e.Host, &e.Op, &e.Name, &e.Incarnation, &e.Body, &at, &e.Confirmed, &e.Attempts, &e.Warned); err != nil {
+ if err := rows.Scan(&e.Host, &e.Op, &e.Name, &e.Incarnation, &e.Body, &at, &e.Confirmed, &e.Attempts, &e.Warned, &e.Held); err != nil {
return nil, err
}
e.QueuedAt = time.Unix(at, 0).UTC()
@@ -2274,7 +2275,7 @@ func scanMoveLogRows(rows *sql.Rows) ([]MoveLogPendingEntry, error) {
return out, rows.Err()
}
-const moveLogColumns = `host, op, name, incarnation, body, queued_at, confirmed, attempts, warned`
+const moveLogColumns = `host, op, name, incarnation, body, queued_at, confirmed, attempts, warned, held`
// MoveLogPending is one move's queued entries, oldest first.
func (s *Store) MoveLogPending(host, op string) ([]MoveLogPendingEntry, error) {
@@ -2289,7 +2290,7 @@ func (s *Store) MoveLogPending(host, op string) ([]MoveLogPendingEntry, error) {
// the longest due first: every attempt that does not settle a row moves its
// next attempt on, so due rows rotate through the batch.
func (s *Store) MoveLogDue(now time.Time, limit int) ([]MoveLogPendingEntry, error) {
- rows, err := s.db.Query(`SELECT `+moveLogColumns+` FROM move_log_outbox WHERE next_at <= ? ORDER BY next_at, rowid LIMIT ?`, now.Unix(), limit)
+ rows, err := s.db.Query(`SELECT `+moveLogColumns+` FROM move_log_outbox WHERE held = 0 AND next_at <= ? ORDER BY next_at, rowid LIMIT ?`, now.Unix(), limit)
if err != nil {
return nil, err
}
@@ -2304,6 +2305,14 @@ func (s *Store) RetryMoveLog(host, op, name string, attempts int, next time.Time
return err
}
+// HoldMoveLog stops a row's delivery for good, keeping its bytes: the writer
+// refused it (it can never be written as it stands), and only its
+// incarnation's purge removes it. Its warning has been given.
+func (s *Store) HoldMoveLog(host, op, name string) error {
+ _, err := s.db.Exec(`UPDATE move_log_outbox SET held = 1, warned = 1 WHERE host = ? AND op = ? AND name = ?`, host, op, name)
+ return err
+}
+
// PurgeMoveLogIncarnation drops every row of one instance incarnation: the
// incarnation is retired or purged, so none of them can ever be written.
func (s *Store) PurgeMoveLogIncarnation(host, incarnation string) error {