amadan

carlos / platform Public

moves-phase1-6 Task 3 fix round 4 (2): unresolved log rows take backoff and the hour warning

An unconfirmed row the drainer cannot settle yet (its claim still holds the
move, or is unreadable) is rescheduled like a failed delivery, with the same
one-hour warning, and the batch picks rows by their next attempt; a
confirmation makes a row due at once. TestUnresolvedMoveLogRowsDoNotStarveTheBatch.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 66e100de44753c6c8cfa79921d6e00e2da546189 parent 4bc5b0f
3 files changed, +78 −24
  • internal/console/movesrecovery.go +32 −20
  • internal/console/movesrecovery_test.go +41 −0
  • internal/console/store.go +5 −4
diff --git a/internal/console/movesrecovery.go b/internal/console/movesrecovery.go
index 4d5c605f9..7ef90d6c8 100644
--- a/internal/console/movesrecovery.go
+++ b/internal/console/movesrecovery.go
@@ -634,8 +634,14 @@ 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
+ if !r.Confirmed {
+ deliver, wait := s.settleUnconfirmed(ctx, r)
+ if wait != nil {
+ s.retryMoveLog(r, now, wait)
+ }
+ if !deliver {
+ continue
+ }
}
s.deliverMoveLog(ctx, lw, r, now)
}
@@ -644,29 +650,42 @@ func (s *Server) drainMoveLog(ctx context.Context, lw *hostops.MoveLogWriter) er
// 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 {
+// confirmed (deliver); still the move's, or the claim unreadable, it waits
+// (wait says why, and it takes a failed delivery's backoff and warning, so
+// 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) {
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
+ return false, fmt.Errorf("unconfirmed, and the claim cannot be read: %w", err)
}
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 false, fmt.Errorf("unconfirmed: %w", err)
}
- return true
+ return true, nil
case !o.Absent && o.Claim != nil && o.Claim.Op.ID == r.Op:
- return false
+ return false, errors.New("unconfirmed: the claim still holds the move")
}
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
+ 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) {
+ 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, why)
+ 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)
+ }
}
// deliverMoveLog writes one row, and settles it or schedules its retry.
@@ -693,14 +712,7 @@ func (s *Server) deliverMoveLog(ctx context.Context, lw *hostops.MoveLogWriter,
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)
- }
+ s.retryMoveLog(r, now, err)
}
}
diff --git a/internal/console/movesrecovery_test.go b/internal/console/movesrecovery_test.go
index d4367ac27..b910b484d 100644
--- a/internal/console/movesrecovery_test.go
+++ b/internal/console/movesrecovery_test.go
@@ -4,6 +4,7 @@ import (
"context"
"crypto/rand"
"encoding/json"
+ "fmt"
"os"
"path/filepath"
"strings"
@@ -957,3 +958,43 @@ func TestClaimReleasedIsPublishedOnlyAfterTheRelease(t *testing.T) {
}
})
}
+
+// Fix round 4, ruling 2: a row the drainer cannot settle yet (unconfirmed,
+// its claim still the move's) takes the same backoff and one-hour warning as
+// a failed delivery, and the batch picks rows by their next attempt, so a
+// full batch of them never starves a confirmed row.
+func TestUnresolvedMoveLogRowsDoNotStarveTheBatch(t *testing.T) {
+ journal := captureLog(t)
+ w := landingMovePass(t, "code-mvpass-starve")
+ body := func(name string) []byte {
+ b, err := json.Marshal(moves.LogEvent{V: 1, Op: w.id, Host: "acme.go.tito.io", Event: "x", Identity: name, Actor: "console", At: w.clock})
+ if err != nil {
+ t.Fatal(err)
+ }
+ return b
+ }
+ prefix := moves.ReportPrefix("bab", "titogo", "acme.go.tito.io", w.id)
+ for i := 0; i < moveLogDrainBatch; i++ {
+ name := fmt.Sprintf("held-%03d", i)
+ if err := w.fx.st.QueueMoveLog("acme.go.tito.io", w.id, name, prefix, body(name), false, w.clock); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := w.fx.st.QueueMoveLog("acme.go.tito.io", w.id, "late", prefix, body("late"), true, w.clock); err != nil {
+ t.Fatal(err)
+ }
+ for i := 0; i < 2 && moveLogOf(t, w)["late"].Identity == ""; i++ {
+ w.pass(t)
+ }
+ if moveLogOf(t, w)["late"].Identity != "late" {
+ t.Fatal("the confirmed row was starved by a full batch of unresolved rows")
+ }
+ // The unresolved rows are warned of once, an hour on.
+ for i := 0; i < 10; i++ {
+ w.advance(30 * time.Minute)
+ w.pass(t)
+ }
+ if n := strings.Count(journal.String(), "unwritten for over an hour"); n != moveLogDrainBatch {
+ t.Fatalf("%d hour-old warnings, want one per unresolved row (%d)", n, moveLogDrainBatch)
+ }
+}
diff --git a/internal/console/store.go b/internal/console/store.go
index 429483b26..6cfab03cf 100644
--- a/internal/console/store.go
+++ b/internal/console/store.go
@@ -2251,9 +2251,9 @@ ON CONFLICT(host, op, name) DO NOTHING`, host, op, name, prefix, body, at.Unix()
return err
}
-// ConfirmMoveLog marks a queued entry deliverable.
+// ConfirmMoveLog marks a queued entry deliverable, due at once.
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)
+ _, err := s.db.Exec(`UPDATE move_log_outbox SET confirmed = 1, next_at = 0 WHERE host = ? AND op = ? AND name = ?`, host, op, name)
return err
}
@@ -2284,9 +2284,10 @@ func (s *Store) MoveLogPending(host, op string) ([]MoveLogPendingEntry, error) {
}
// MoveLogDue is up to limit queued entries whose next attempt is due at now,
-// oldest first.
+// 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 rowid LIMIT ?`, now.Unix(), limit)
+ rows, err := s.db.Query(`SELECT `+moveLogColumns+` FROM move_log_outbox WHERE next_at <= ? ORDER BY next_at, rowid LIMIT ?`, now.Unix(), limit)
if err != nil {
return nil, err
}