amadan

carlos / platform Public

moves-phase1-6 Task 3 fix round 3 (1): a console log drainer, decoupled from grants and the release

- hostops: Recovery.MoveLogs runs once a pass with a MoveLogWriter minted
  inside RunRecovery (runMoveLogDrain). WriteMoveLog derives the key from
  the entry's account, app, host and move, takes a single-segment name equal
  to the identity, caps the entry at 4 KiB, requires the host's claim to
  still name the app, and writes create-only: same bytes a no-op, other
  bytes ErrMoveLogTaken.
- console: every entry the move pass makes is queued in move_log_outbox
  (with its report prefix) and delivered by drainMoveLog, whatever the move
  has done since. moveLogReleaseWait and the grant flush are gone: no release
  waits on logging. Failed rows retry with backoff indefinitely and get one
  journal warning at an hour old.
- Census rows: the authority census (runMoveLogDrain, WriteMoveLog; the
  grant Log call sites are gone) and B8 (MoveLogWriter.WriteMoveLog).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude b00577b7fd9c088ce54dcc11c217a181e72add37 parent fa41acd
12 files changed, +448 −188
  • internal/console/claim_authority_census_test.go +6 −1
  • internal/console/claim_runtime.go +7 −0
  • internal/console/movesrecovery.go +91 −76
  • internal/console/movesrecovery_test.go +51 −82
  • internal/console/store.go +60 −17
  • internal/console/store_gate.go +1 −1
  • internal/console/testdata/claim-authority-sites.txt +2 −2
  • internal/hostops/move_log_writer.go +115 −0
  • internal/hostops/move_log_writer_test.go +100 −0
  • internal/hostops/move_recovery_grant.go +0 −7
  • internal/hostops/recovery.go +11 −2
  • internal/provision/plan9_b8_census_test.go +4 −0
diff --git a/internal/console/claim_authority_census_test.go b/internal/console/claim_authority_census_test.go
index 6a88a348c..6c5a51572 100644
--- a/internal/console/claim_authority_census_test.go
+++ b/internal/console/claim_authority_census_test.go
@@ -31,7 +31,12 @@ func claimAuthoritySites(t *testing.T) []string {
"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,
+ "Log": true,
+ // The recovery pass's move log writer (Plan 6 Task 3, fix round 3
+ // ruling 1): minted inside RunRecovery for the console's outbox
+ // drainer; one create-only entry per call, under the entry's own
+ // move's report prefix, for a host whose claim still names the app.
+ "runMoveLogDrain": true, "WriteMoveLog": true,
"BeginPurge": true, "SavePurgeJournal": true, "RemovePurgeJournal": true,
"AcquireMailGate": true, "ReleaseMailGate": true, "EnrollNewApp": true,
"EnrollBackfillApp": true, "OpenGated": true, "WithAppMutation": true,
diff --git a/internal/console/claim_runtime.go b/internal/console/claim_runtime.go
index c1bc23eb5..f70b793ee 100644
--- a/internal/console/claim_runtime.go
+++ b/internal/console/claim_runtime.go
@@ -124,8 +124,15 @@ func (s *Server) ConfigureClaimRuntime(ctx context.Context) error {
return hostops.StampLostMoveTargetParked(ctx, records, host, now)
}
}
+ var moveLogs func(context.Context, *hostops.MoveLogWriter) error
+ if s.cfg.MovesEnabled {
+ // The delivery of the move pass's log entries (Plan 6 Task 3), once
+ // a pass and independent of any move's state or grant.
+ moveLogs = s.drainMoveLog
+ }
s.cfg.HostClaimRecovery = &hostops.Recovery{
Moves: s.cfg.ClaimMoveRecovery,
+ MoveLogs: moveLogs,
MoveFenceAuthority: hostfence.ClaimFenceAuthority{Directory: a.Directory, Now: c.Now},
Controller: c,
SweepRecords: func(ctx context.Context) error { return c.SweepAllRecords(ctx) },
diff --git a/internal/console/movesrecovery.go b/internal/console/movesrecovery.go
index c35b83150..4c5e57641 100644
--- a/internal/console/movesrecovery.go
+++ b/internal/console/movesrecovery.go
@@ -10,6 +10,7 @@ import (
"log"
"os"
"slices"
+ "strings"
"time"
"github.com/carlosframework/platform/internal/alerting"
@@ -283,9 +284,6 @@ 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 {
@@ -336,20 +334,13 @@ func (s *Server) movesRecovery(ctx context.Context, g *hostops.MoveRecoveryGrant
}
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
}
- if d.Claim == hostops.RecoveryMoveRollback {
+ switch d.Claim {
+ case hostops.RecoveryMoveRelease:
+ s.queueMoveLog(o, claimTransition(o, rec, "claim_released", "claim-released"), now)
+ case hostops.RecoveryMoveRollback:
s.queueMoveLog(o, claimTransition(o, rec, "claim_rollback", "claim-rollback"), now)
}
return nil
@@ -547,59 +538,36 @@ 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 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 move log of the console's own move writes (Plan 6 Task 3; fix round 3,
+// ruling 1): 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 outbox of move %s: %v", o.Host, o.OpID, err)
- return 0, time.Time{}
- }
- 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
+// 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:
+// 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:
+// the record never named that move.
+
+// moveLogWarnAfter is the age at which an undelivered entry is warned of.
+const moveLogWarnAfter = time.Hour
+
+// moveLogDrainBatch bounds one pass's deliveries.
+const moveLogDrainBatch = 256
+
+// moveLogBackoff is the wait after a row's attempts-th failure: 30 seconds,
+// doubling, at most an hour.
+func moveLogBackoff(attempts int) time.Duration {
+ d := 30 * time.Second
+ for i := 1; i < attempts && d < time.Hour; i++ {
+ d *= 2
+ }
+ return min(d, time.Hour)
}
// moveLogEntry is the console's entry for t.
@@ -607,28 +575,75 @@ func moveLogEntry(o hostops.ConsoleMoveObservation, t moves.Transition, now time
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.
+// queueMoveLog queues the entry for t, for the drainer.
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)
+ err = s.store.QueueMoveLog(o.Host, o.OpID, t.Identity, moves.ReportPrefix(o.Account, o.App, o.Host, o.OpID), body, true, 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)
+// moveLogPrefixParts is the account and app of a row's report prefix, which
+// must be exactly moves.ReportPrefix of them, the row's host and its move.
+func moveLogPrefixParts(prefix, host, op string) (account, app string, ok bool) {
+ p := strings.Split(prefix, "/")
+ if len(p) != 6 || p[0] != "moves" || p[5] != "" || moves.ReportPrefix(p[1], p[2], host, op) != prefix {
+ return "", "", false
+ }
+ return p[1], p[2], true
+}
+
+// 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 {
+ now := s.movesNow()
+ rows, err := s.store.MoveLogDue(now, moveLogDrainBatch)
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 fmt.Errorf("moves: the log outbox: %w", err)
+ }
+ for _, r := range rows {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ s.deliverMoveLog(ctx, lw, r, now)
+ }
+ return nil
+}
+
+// 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() {
+ 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)
+ }
+ }
+ account, app, ok := moveLogPrefixParts(r.Prefix, r.Host, r.Op)
+ if !ok {
+ log.Printf("console: moves: %s: the log of move %s at %s names no move's report prefix (%q); dropped", r.Host, r.Op, r.Name, r.Prefix)
+ settle()
return
}
- 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)
+ _, err := lw.WriteMoveLog(ctx, hostops.MoveLogEntry{Account: account, App: app, Host: r.Host, Op: r.Op, Name: r.Name, Body: r.Body})
+ switch {
+ case err == nil:
+ settle()
+ case errors.Is(err, hostops.ErrMoveLogTaken):
+ 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()
+ default:
+ attempts, warned := r.Attempts+1, r.Warned
+ if !warned && now.Sub(r.QueuedAt) >= moveLogWarnAfter {
+ log.Printf("console: moves: warning: %s: the log of move %s at %s has been unwritten for over an hour: %v", r.Host, r.Op, r.Name, err)
+ warned = true
+ }
+ if err := s.store.RetryMoveLog(r.Host, r.Op, r.Name, attempts, now.Add(moveLogBackoff(attempts)), warned); err != nil {
+ log.Printf("console: moves: %s: the log outbox of move %s: %v", r.Host, r.Op, err)
}
}
}
diff --git a/internal/console/movesrecovery_test.go b/internal/console/movesrecovery_test.go
index 99a2dd29b..c4e3e1241 100644
--- a/internal/console/movesrecovery_test.go
+++ b/internal/console/movesrecovery_test.go
@@ -229,7 +229,7 @@ func (w *movePassWorld) pass(t *testing.T) {
a := w.fx.srv.cfg.ClaimAuthority
nop := func(context.Context) error { return nil }
nopObs := func(context.Context, hostclaim.Observed) error { return nil }
- r := hostops.Recovery{Moves: w.fx.srv.movesRecovery, MoveFenceAuthority: w.auth,
+ r := hostops.Recovery{Moves: w.fx.srv.movesRecovery, MoveLogs: w.fx.srv.drainMoveLog, MoveFenceAuthority: w.auth,
Controller: hostops.Controller{Store: w.fx.srv.cfg.Releases, Control: a.Control, Authority: a, Now: time.Now, NewID: claimOperationID},
Purge: nopObs, Placement: nopObs, SweepRecords: nop, SweepPurges: nop, SweepAdmissions: nop}
if err := a.Control.RecoveryPass(context.Background(), r); err != nil {
@@ -736,14 +736,11 @@ func moveLogOf(t *testing.T, w *movePassWorld) map[string]moves.LogEvent {
return out
}
-// 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.
+// Plan 6 Task 3 fix round 1, ruling 3, and fix round 3: 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, queued in the
+// console's outbox and delivered by the pass's drainer (drainMoveLog).
func TestRecoveryPassWritesTheMoveLog(t *testing.T) {
t.Run("close and release", func(t *testing.T) {
w := landingMovePass(t, "code-mvpass-log-close")
@@ -751,7 +748,7 @@ 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 close's entry, then the claim's release
+ w.pass(t) // 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" {
@@ -767,7 +764,6 @@ func TestRecoveryPassWritesTheMoveLog(t *testing.T) {
past := time.Now().UTC().Add(-time.Minute)
j17SetRecord(t, w.fx, func(r *provision.Record) { r.Move.LandingDeadline = &past })
w.pass(t)
- 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))
}
@@ -779,7 +775,6 @@ 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))
}
@@ -813,7 +808,11 @@ func pendingLog(t *testing.T, w *movePassWorld) []string {
}
var names []string
for _, e := range got {
- names = append(names, e.Name)
+ n := e.Name
+ if !e.Confirmed {
+ n += "(unconfirmed)"
+ }
+ names = append(names, n)
}
return names
}
@@ -825,76 +824,46 @@ func noPendingLog(t *testing.T, w *movePassWorld) {
}
}
-// 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.
+// Fix round 3, ruling 1: the console's log delivery is the pass's drainer,
+// independent of any move's state or grant, and no release waits for it. A
+// row whose write fails is retried with backoff, indefinitely, and warned of
+// once in the journal when it is an hour old; it is still delivered after
+// the claim's release.
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
+ journal := captureLog(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 queued and its delivery fails
+ if rec := w.record(t); rec.Move != nil {
+ t.Fatalf("not closed: %+v", rec.Move)
+ }
+ w.pass(t) // the release does not wait for the log
+ if c := w.claim(t); c.Op.Kind != hostclaim.KindServing {
+ t.Fatalf("the release waited for the log: %+v", c.Op)
+ }
+ if p := pendingLog(t, w); strings.Join(p, " ") != "done claim-released" {
+ t.Fatalf("pending %v, want the close's and the release's entries", p)
+ }
+ for i := 0; i < 8; i++ { // over an hour of failing passes
+ w.advance(10 * time.Minute)
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)
+ }
+ for i := 0; i < 6; i++ { // and more failed attempts after the warning
+ w.advance(30 * 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")
- }
- })
+ }
+ if n := strings.Count(journal.String(), "unwritten for over an hour"); n != 2 {
+ t.Fatalf("%d hour-old warnings, want one per row:\n%s", n, journal.String())
+ }
+ unblock()
+ w.advance(2 * time.Hour) // past any backoff
+ w.pass(t)
+ log := moveLogOf(t, w)
+ if log["done"].Actor != "console" || log["claim-released"].Event != "claim_released" {
+ t.Fatalf("log %+v", log)
+ }
+ noPendingLog(t, w)
}
diff --git a/internal/console/store.go b/internal/console/store.go
index 7e848aada..429483b26 100644
--- a/internal/console/store.go
+++ b/internal/console/store.go
@@ -361,13 +361,22 @@ func (s *Store) migrate() error {
);`,
// 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.
+ // bytes and the report prefix they go under, until the recovery
+ // pass's drainer writes them or finds them written. An unconfirmed
+ // row waits for the write it records to be confirmed; next_at is
+ // the earliest next attempt, and warned marks the one journal
+ // warning an hour-old row gets.
`CREATE TABLE IF NOT EXISTS move_log_outbox (
host TEXT NOT NULL,
op TEXT NOT NULL,
name TEXT NOT NULL,
+ prefix 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,
PRIMARY KEY (host, op, name)
);`,
} {
@@ -2226,31 +2235,35 @@ ON CONFLICT(rule) DO UPDATE SET attempted_at = excluded.attempted_at`, rule, at.
// MoveLogPendingEntry is one move log entry the move pass owes.
type MoveLogPendingEntry struct {
- Name string
- Body []byte
- QueuedAt time.Time
+ Host, Op, Name, Prefix string
+ Body []byte
+ QueuedAt time.Time
+ Confirmed, Warned bool
+ Attempts int
+}
+
+// QueueMoveLog keeps an entry for the drainer. An entry already queued under
+// the name is kept as it was. An unconfirmed entry is not delivered until
+// ConfirmMoveLog.
+func (s *Store) QueueMoveLog(host, op, name, prefix string, body []byte, confirmed bool, at time.Time) error {
+ _, err := s.db.Exec(`INSERT INTO move_log_outbox (host, op, name, prefix, body, queued_at, confirmed) VALUES (?, ?, ?, ?, ?, ?, ?)
+ON CONFLICT(host, op, name) DO NOTHING`, host, op, name, prefix, body, at.Unix(), confirmed)
+ return err
}
-// 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())
+// ConfirmMoveLog marks a queued entry deliverable.
+func (s *Store) ConfirmMoveLog(host, op, name string) error {
+ _, err := s.db.Exec(`UPDATE move_log_outbox SET confirmed = 1 WHERE host = ? AND op = ? AND name = ?`, host, op, name)
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
- }
+func scanMoveLogRows(rows *sql.Rows) ([]MoveLogPendingEntry, error) {
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 {
+ if err := rows.Scan(&e.Host, &e.Op, &e.Name, &e.Prefix, &e.Body, &at, &e.Confirmed, &e.Attempts, &e.Warned); err != nil {
return nil, err
}
e.QueuedAt = time.Unix(at, 0).UTC()
@@ -2259,7 +2272,37 @@ func (s *Store) MoveLogPending(host, op string) ([]MoveLogPendingEntry, error) {
return out, rows.Err()
}
-// SettleMoveLog drops an entry once it is written, or found written.
+const moveLogColumns = `host, op, name, prefix, body, queued_at, confirmed, attempts, warned`
+
+// MoveLogPending is one move's queued entries, oldest first.
+func (s *Store) MoveLogPending(host, op string) ([]MoveLogPendingEntry, error) {
+ rows, err := s.db.Query(`SELECT `+moveLogColumns+` FROM move_log_outbox WHERE host = ? AND op = ? ORDER BY rowid`, host, op)
+ if err != nil {
+ return nil, err
+ }
+ return scanMoveLogRows(rows)
+}
+
+// MoveLogDue is up to limit queued entries whose next attempt is due at now,
+// oldest first.
+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 rowid LIMIT ?`, now.Unix(), limit)
+ if err != nil {
+ return nil, err
+ }
+ return scanMoveLogRows(rows)
+}
+
+// RetryMoveLog records a failed attempt: the attempt count, the next
+// attempt's time and whether the row's warning has been given.
+func (s *Store) RetryMoveLog(host, op, name string, attempts int, next time.Time, warned bool) error {
+ _, err := s.db.Exec(`UPDATE move_log_outbox SET attempts = ?, next_at = ?, warned = ? WHERE host = ? AND op = ? AND name = ?`,
+ attempts, next.Unix(), warned, host, op, name)
+ return err
+}
+
+// SettleMoveLog drops an entry once it is written, found written, or can
+// never be.
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
diff --git a/internal/console/store_gate.go b/internal/console/store_gate.go
index f6160901b..804ff3bbc 100644
--- a/internal/console/store_gate.go
+++ b/internal/console/store_gate.go
@@ -210,7 +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"},
+ "move_log_outbox": {NonApp: "the move pass's undelivered move log entries (platform observability, never app state), keyed by host, move and entry, each with the report prefix it goes under; the writer re-derives the key and checks the host's claim still names that app; 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 e57d978f5..15ad0defa 100644
--- a/internal/console/testdata/claim-authority-sites.txt
+++ b/internal/console/testdata/claim-authority-sites.txt
@@ -35,6 +35,7 @@
../hostops/purge_journal.go|DeletePurgeSweep|w.SavePurgeJournal(ctx, fresh, tag)
../hostops/purge_journal.go|PreparePurgeSweep|w.SavePurgeJournal(ctx, j, tag)
../hostops/purge_journal.go|VerifyPurgeSweep|w.SavePurgeJournal(ctx, j, tag)
+../hostops/recovery.go|Pass|runMoveLogDrain(ctx, r.MoveLogs)
../hostops/recovery.go|Pass|runMoveRecovery(ctx, obs, r.Moves, r.MoveFenceAuthority)
../hostops/sidecarwake.go|TouchSidecarWake|a.Control.RegisterStart(ctx, account, app, appID, id, reg, a.Now())
../mail/aws.go|NewAWSProvisioner|iam.NewFromConfig(cfg)
@@ -106,8 +107,7 @@ 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|deliverMoveLog|lw.WriteMoveLog(ctx, hostops.MoveLogEntry{Account: account, App: app, Host: r.Host, Op: r.Op, Name: r.Name, Body: r.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/hostops/move_log_writer.go b/internal/hostops/move_log_writer.go
new file mode 100644
index 000000000..7a31c0ce9
--- /dev/null
+++ b/internal/hostops/move_log_writer.go
@@ -0,0 +1,115 @@
+package hostops
+
+import (
+ "bytes"
+ "context"
+ "errors"
+
+ "github.com/carlosframework/platform/internal/controlname"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The recovery pass's move log writer (Plan 6 Task 3, fix round 3, ruling 1).
+// The console queues the move log entries its move pass makes in its outbox
+// (move_log_outbox) and delivers them through this writer, once per recovery
+// pass (Recovery.MoveLogs), whatever the moves have done since: a grant is
+// spent by the time its transition's entry exists, and the claim's release
+// never waits for the log. The writer exposes no store: one create-only entry
+// per call, under the entry's own move's report prefix.
+
+// MoveLogEntryMax bounds one entry the writer writes, in bytes.
+const MoveLogEntryMax = 4096
+
+// ErrMoveLogTaken is an entry whose 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")
+
+// MoveLogEntry is one move log entry to deliver: the move's account, app,
+// host and id, the entry's name (its identity) and its exact bytes.
+type MoveLogEntry struct {
+ Account, App, Host, Op, Name string
+ Body []byte
+}
+
+// MoveLogWriter is one recovery pass's move log writer, valid only while its
+// hook runs.
+type MoveLogWriter struct {
+ w *ControlWriter
+ state *moveRecoveryGrantState
+}
+
+// runMoveLogDrain mints the pass's writer, inside RunRecovery only, and hands
+// it to hook. The hook receives the writer, never the minting context.
+func runMoveLogDrain(ctx context.Context, hook func(context.Context, *MoveLogWriter) error) error {
+ w := recoveryWriter(ctx)
+ if w == nil || hook == nil {
+ return ErrContract
+ }
+ lw := &MoveLogWriter{w: w, state: &moveRecoveryGrantState{active: true}}
+ state := lw.state
+ defer func() { state.mu.Lock(); state.active = false; state.mu.Unlock() }()
+ ctx = context.WithValue(ctx, lifecycleGrantKey{}, []lifecycleGrant(nil))
+ ctx = context.WithValue(ctx, moveRecoveryPassKey{}, (*moveRecoveryPassScope)(nil))
+ return hook(ctx, lw)
+}
+
+// WriteMoveLog writes one entry, with the checks MoveRecoveryGrant.Log made:
+//
+// - The key is derived, never given: e.Name under the report prefix of
+// e's account, app, host and move. Each is a valid name; the move id and
+// the entry name are one segment of letters, digits, '_' and '-'.
+// - Body is at most MoveLogEntryMax bytes and decodes, strictly, as a
+// moves.LogEvent of that move and host, named e.Name.
+// - The host's claim still names the app (so a purged or re-created
+// instance's log is never written).
+// - The write is create-only. A replay of the same bytes is a no-op
+// (created false); other bytes under the name are ErrMoveLogTaken.
+//
+// ErrIdentity is an entry that can never be written. It answers ErrPending
+// once its hook has returned.
+func (lw *MoveLogWriter) WriteMoveLog(ctx context.Context, e MoveLogEntry) (created bool, err error) {
+ if lw == nil || lw.state == nil || lw.w == nil {
+ return false, ErrPending
+ }
+ lw.state.mu.Lock()
+ defer lw.state.mu.Unlock()
+ if !lw.state.active {
+ return false, ErrPending
+ }
+ if !controlname.ValidAccount(e.Account) || !controlname.ValidApp(e.App) || !moveLogName(e.Op) || !moveLogName(e.Name) || len(e.Body) > MoveLogEntryMax {
+ return false, ErrIdentity
+ }
+ if _, err := hostclaim.Key(e.Host); err != nil {
+ return false, ErrIdentity
+ }
+ var ev moves.LogEvent
+ if err := strictJSON(e.Body, &ev); err != nil {
+ return false, ErrIdentity
+ }
+ if ev.V != 1 || ev.Op != e.Op || ev.Host != e.Host || ev.Identity != e.Name {
+ return false, ErrIdentity
+ }
+ o, err := hostclaim.Read(ctx, lw.w.raw, e.Host)
+ if err != nil {
+ return false, err
+ }
+ if o.Absent || o.Claim == nil || o.Claim.Holder == nil || o.Claim.Holder.Account != e.Account || o.Claim.Holder.App != e.App {
+ return false, ErrIdentity
+ }
+ key := moves.LogKeyFor(moves.ReportPrefix(e.Account, e.App, e.Host, e.Op), e.Name)
+ if _, err := lw.w.raw.Put(ctx, key, e.Body, ""); err == nil {
+ return true, nil
+ } else if !errors.Is(err, release.ErrCASConflict) {
+ return false, err
+ }
+ have, _, err := lw.w.raw.Get(ctx, key)
+ if err != nil {
+ return false, err
+ }
+ if !bytes.Equal(have, e.Body) {
+ return false, ErrMoveLogTaken
+ }
+ return false, nil
+}
diff --git a/internal/hostops/move_log_writer_test.go b/internal/hostops/move_log_writer_test.go
new file mode 100644
index 000000000..26d1e49c0
--- /dev/null
+++ b/internal/hostops/move_log_writer_test.go
@@ -0,0 +1,100 @@
+package hostops
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// Plan 6 Task 3 fix round 3, ruling 1: the recovery pass's move log writer
+// (Recovery.MoveLogs) delivers the console's queued entries whatever their
+// moves now are, with Grant.Log's key derivation and checks: the key is
+// derived from the entry's account, app, host and move under the report
+// prefix; the name is one segment and the entry's identity; the entry is at
+// most MoveLogEntryMax bytes and this move's and host's; the host's claim
+// still names the app; the write is create-only, a replay of the same bytes
+// a no-op and other bytes ErrMoveLogTaken. It answers only during its hook,
+// inside a recovery pass.
+func TestMoveLogWriterWritesOnlyVerifiedEntries(t *testing.T) {
+ w := newC7World(t, func(c *hostclaim.Claim, r *provision.Record, _ *AppAdmission) {})
+ ctx := context.Background()
+ entry := func(op, host, id, actor string) []byte {
+ b, err := json.Marshal(moves.LogEvent{V: 1, Op: op, Host: host, Event: "x", Identity: id, Actor: actor, At: w.now})
+ if err != nil {
+ t.Fatal(err)
+ }
+ return b
+ }
+ good := MoveLogEntry{Account: "bab", App: "app", Host: w.host, Op: "move", Name: "done", Body: entry("move", w.host, "done", "console")}
+ with := func(f func(*MoveLogEntry)) MoveLogEntry { e := good; f(&e); return e }
+ big := moves.LogEvent{V: 1, Op: "move", Host: w.host, Event: "x", Identity: "done", Fields: map[string]string{"pad": strings.Repeat("a", MoveLogEntryMax)}}
+ bigBody, _ := json.Marshal(big)
+ var saved *MoveLogWriter
+ var created, again bool
+ got := map[string]error{}
+ err := w.control.RunRecovery(ctx, func(ctx context.Context) error {
+ return runMoveLogDrain(ctx, func(ctx context.Context, lw *MoveLogWriter) error {
+ saved = lw
+ var err error
+ if created, err = lw.WriteMoveLog(ctx, good); err != nil {
+ return err
+ }
+ if again, err = lw.WriteMoveLog(ctx, good); err != nil {
+ return err
+ }
+ for what, e := range map[string]MoveLogEntry{
+ "other bytes": with(func(e *MoveLogEntry) { e.Body = entry("move", w.host, "done", "target") }),
+ "another app": with(func(e *MoveLogEntry) { e.App = "other" }),
+ "another account": with(func(e *MoveLogEntry) { e.Account = "cab" }),
+ "another move's entry": with(func(e *MoveLogEntry) { e.Body = entry("other", w.host, "done", "console") }),
+ "another host's entry": with(func(e *MoveLogEntry) { e.Body = entry("move", "elsewhere.example", "done", "console") }),
+ "a name that is not the identity": with(func(e *MoveLogEntry) { e.Name = "opened" }),
+ "a path": with(func(e *MoveLogEntry) { e.Name = "../source" }),
+ "a move id that is a path": with(func(e *MoveLogEntry) { e.Op = "../move" }),
+ "a host with no claim": with(func(e *MoveLogEntry) {
+ e.Host = "gone.example"
+ e.Body = entry("move", "gone.example", "done", "console")
+ }),
+ "too big": with(func(e *MoveLogEntry) { e.Body = bigBody }),
+ } {
+ _, got[what] = lw.WriteMoveLog(ctx, e)
+ }
+ return nil
+ })
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !created || again {
+ t.Fatalf("created=%v, a replay of the same bytes created=%v", created, again)
+ }
+ for what, err := range got {
+ want := ErrIdentity
+ if what == "other bytes" {
+ want = ErrMoveLogTaken
+ }
+ if !errors.Is(err, want) {
+ t.Errorf("%s: %v, want %v", what, err, want)
+ }
+ }
+ prefix := moves.ReportPrefix("bab", "app", w.host, "move")
+ raw, _, err := w.raw.Get(ctx, moves.LogKeyFor(prefix, "done"))
+ if err != nil || string(raw) != string(good.Body) {
+ t.Fatalf("the object is not the entry's bytes: %s %v", raw, err)
+ }
+ if logged, err := moves.ReadLog(ctx, w.raw, prefix); err != nil || len(logged) != 1 {
+ t.Fatalf("log %+v %v", logged, err)
+ }
+ if _, err := saved.WriteMoveLog(ctx, with(func(e *MoveLogEntry) { e.Name = "late"; e.Body = entry("move", w.host, "late", "console") })); !errors.Is(err, ErrPending) {
+ t.Fatalf("a writer kept past its hook wrote: %v", err)
+ }
+ if err := runMoveLogDrain(ctx, func(context.Context, *MoveLogWriter) error { return nil }); !errors.Is(err, ErrContract) {
+ t.Fatalf("a writer minted outside a recovery pass: %v", err)
+ }
+}
diff --git a/internal/hostops/move_recovery_grant.go b/internal/hostops/move_recovery_grant.go
index b65e5033b..f1e925ba1 100644
--- a/internal/hostops/move_recovery_grant.go
+++ b/internal/hostops/move_recovery_grant.go
@@ -126,13 +126,6 @@ func (g *MoveRecoveryGrant) current(ctx context.Context) (hostclaim.Observed, Re
return o, r, rec, err
}
-// 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):
//
diff --git a/internal/hostops/recovery.go b/internal/hostops/recovery.go
index 8b0d0f9d4..f8e8e0f4e 100644
--- a/internal/hostops/recovery.go
+++ b/internal/hostops/recovery.go
@@ -38,8 +38,12 @@ func ListClaims(ctx context.Context, s release.Store) ([]string, error) {
type Recovery struct {
MoveFenceAuthority hostclaim.FenceAuthority
Moves func(context.Context, *MoveRecoveryGrant) error
- SweepAdmissions func(context.Context) error
- Controller Controller
+ // MoveLogs, when set, runs once a pass, after every claim: the console's
+ // delivery of the move log entries its move pass queued, through the
+ // pass's MoveLogWriter (move_log_writer.go).
+ MoveLogs func(context.Context, *MoveLogWriter) error
+ SweepAdmissions func(context.Context) error
+ Controller Controller
// These are required integrations, not no-op defaults.
Purge func(context.Context, hostclaim.Observed) error
Placement func(context.Context, hostclaim.Observed) error
@@ -98,6 +102,11 @@ func (r Recovery) Pass(ctx context.Context) error {
all = append(all, err)
}
}
+ if r.MoveLogs != nil {
+ if err := runMoveLogDrain(ctx, r.MoveLogs); err != nil && !errors.Is(err, ErrPending) {
+ all = append(all, err)
+ }
+ }
// Records have their own uncapped scan: late orphans outlive the one-hour
// cancellation hint; records without claims need the shared adoption path.
if err := r.SweepRecords(ctx); err != nil {
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go
index b65ef0067..307ca32c2 100644
--- a/internal/provision/plan9_b8_census_test.go
+++ b/internal/provision/plan9_b8_census_test.go
@@ -473,6 +473,10 @@ 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"},
+ // The recovery pass's move log writer: one create-only entry of a move's
+ // log under its report prefix, for a host whose claim still names the
+ // app; never a record (Plan 6 Task 3 fix round 3, ruling 1).
+ {"internal/hostops/move_log_writer.go", "MoveLogWriter.WriteMoveLog", "nonrecord", "sel:Put", "9dffdb6f7f15f7dfffc104e0f09df4b6c6d7868a31ee1c85cfb811a04ce5c90c"},
// 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).