amadan

carlos / platform Public

moves: an anomaly page is marked sent only once delivered (Task 9a fix round 1, I4)

movesRefusals.observe set its "paged" mark before calling the pager, so a
failed page suppressed every later one until the agent restarted. Each
anomaly (host and reason) now keeps a page state:

- delivered only after the pager succeeds;
- an in-flight guard: a concurrent observation while a page is sending
  sends nothing;
- a failure is retried by a later observation after a doubling backoff
  (1 minute, capped at 30), at most movesPageAttempts (5) times in all,
  each failure logged and the last one saying no more are sent.

Tests: TestMovesAnomalyPageRetriesUntilDelivered (the first page fails, a
later observation past the backoff pages, success then suppresses; a pager
that always fails gets exactly five tries), TestMovesAnomalyPageIsNotDoubled
(run with -race).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude d66523f4de66d4cbd7e83966f7fa588a2ce9ff1d parent 44681a3
2 files changed, +146 −12
  • internal/host/movesrefusals.go +64 −12
  • internal/host/movesrefusals_test.go +82 −0
diff --git a/internal/host/movesrefusals.go b/internal/host/movesrefusals.go
index d8d74eb9a..ba327236d 100644
--- a/internal/host/movesrefusals.go
+++ b/internal/host/movesrefusals.go
@@ -13,8 +13,10 @@ package host
// logged every time, as before;
// - evidence wraps the provider: a refusal marked Anomaly (the history
// contradicts itself, or something wrote where no move could) pages once
-// per host and reason through the box pager, as a warning. The worker
-// gets every error unchanged.
+// per host and reason through the box pager, as a warning: marked sent
+// only once delivered, never twice at once, a failure retried with
+// backoff a bounded number of times. The worker gets every error
+// unchanged.
//
// Both are wired only with CARLOS_MOVES_PHASE1 and CARLOS_HOST_CLAIM on;
// otherwise the worker's report is the plain log line it always was.
@@ -37,6 +39,9 @@ import (
// movesRefusalLogEvery is how often one expected refusal is logged.
const movesRefusalLogEvery = time.Hour
+// movesPageAttempts bounds the tries one anomaly page gets.
+const movesPageAttempts = 5
+
// movesRefusals is the agent's refusal log and anomaly pager for the moves
// evidence. Its memory is the process's: a restarted agent logs and pages
// each standing condition once more.
@@ -48,11 +53,28 @@ type movesRefusals struct {
mu sync.Mutex
logged map[string]time.Time
- paged map[string]bool
+ pages map[string]*movesPageState
+}
+
+// movesPageState is one anomaly's page: delivered once a send succeeds;
+// while a send is in flight no other starts; a failed send is retried by a
+// later observation after a doubling backoff, at most movesPageAttempts
+// times in all.
+type movesPageState struct {
+ delivered, inflight bool
+ failures int
+ next time.Time
}
+// movesPageBackoff is the wait after the first failed page; it doubles with
+// each further failure, up to movesPageBackoffMax.
+const (
+ movesPageBackoff = time.Minute
+ movesPageBackoffMax = 30 * time.Minute
+)
+
func newMovesRefusals(box string, page func(context.Context, string, string, string) error, logf func(string, ...any), now func() time.Time) *movesRefusals {
- return &movesRefusals{box: box, page: page, logf: logf, now: now, logged: map[string]time.Time{}, paged: map[string]bool{}}
+ return &movesRefusals{box: box, page: page, logf: logf, now: now, logged: map[string]time.Time{}, pages: map[string]*movesPageState{}}
}
// movesQuietEvidence is Task 5's provider with its anomalies paged. It is
@@ -87,23 +109,53 @@ func (e movesQuietEvidence) VerifyMovesSweep(ctx context.Context, o hostclaim.Ob
return proof, err
}
-// observe pages an anomaly once per host and reason.
+// observe pages an anomaly once per host and reason (fix round 1, I4): it
+// is marked delivered only after the pager succeeds; a concurrent
+// observation while a page is in flight sends nothing; a failed page is
+// retried by a later observation after its backoff, at most
+// movesPageAttempts times.
func (r *movesRefusals) observe(ctx context.Context, err error) {
var refusal *MovesSweepRefusal
- if r == nil || !errors.As(err, &refusal) || !refusal.Anomaly {
+ if r == nil || r.page == nil || !errors.As(err, &refusal) || !refusal.Anomaly {
return
}
key := refusal.Host + "\x00" + refusal.Reason
+ now := r.now()
r.mu.Lock()
- seen := r.paged[key]
- r.paged[key] = true
- r.mu.Unlock()
- if seen || r.page == nil {
+ st := r.pages[key]
+ if st == nil {
+ st = &movesPageState{}
+ r.pages[key] = st
+ }
+ if st.delivered || st.inflight || st.failures >= movesPageAttempts || now.Before(st.next) {
+ r.mu.Unlock()
return
}
+ st.inflight = true
+ r.mu.Unlock()
subject, body := movesAnomalyPage(refusal.Host, r.box)
- if perr := r.page(ctx, alerting.LevelWarning, subject, body); perr != nil && r.logf != nil {
- r.logf("moves: %s: the page for the refused cleanup: %v", refusal.Host, perr)
+ perr := r.page(ctx, alerting.LevelWarning, subject, body)
+ r.mu.Lock()
+ st.inflight = false
+ if perr == nil {
+ st.delivered = true
+ r.mu.Unlock()
+ return
+ }
+ st.failures++
+ wait := movesPageBackoff << (st.failures - 1)
+ if wait > movesPageBackoffMax || wait <= 0 {
+ wait = movesPageBackoffMax
+ }
+ st.next = now.Add(wait)
+ failures := st.failures
+ r.mu.Unlock()
+ if r.logf != nil {
+ if failures >= movesPageAttempts {
+ r.logf("moves: %s: the page for the refused cleanup failed %d times; no more are sent while this agent runs: %v", refusal.Host, failures, perr)
+ } else {
+ r.logf("moves: %s: the page for the refused cleanup failed (try %d of %d, again after %s): %v", refusal.Host, failures, movesPageAttempts, wait, perr)
+ }
}
}
diff --git a/internal/host/movesrefusals_test.go b/internal/host/movesrefusals_test.go
index ab24f46c0..14ff1b521 100644
--- a/internal/host/movesrefusals_test.go
+++ b/internal/host/movesrefusals_test.go
@@ -1,6 +1,7 @@
package host
import (
+ "context"
"errors"
"fmt"
"strings"
@@ -142,3 +143,84 @@ func TestRegionalWorkerRefusalsAreQuiet(t *testing.T) {
t.Fatalf("pages %+v, log lines %q over three passes", pages, lines)
}
}
+
+// anomalyErr is one anomalous refusal of evHost.
+var anomalyErr = &MovesSweepRefusal{Host: evHost, Reason: "the record is not the holder's incarnation", Anomaly: true}
+
+// TestMovesAnomalyPageRetriesUntilDelivered (fix round 1, I4): a page that
+// fails is not marked delivered; a later observation, past the backoff,
+// pages again; once one succeeds nothing more is sent. A pager that always
+// fails is tried a bounded number of times.
+func TestMovesAnomalyPageRetriesUntilDelivered(t *testing.T) {
+ l := &refusalLog{now: evNow}
+ calls, fail := 0, true
+ r := newMovesRefusals("ie-box", func(context.Context, string, string, string) error {
+ calls++
+ if fail {
+ return errors.New("the webhook answered 503")
+ }
+ return nil
+ }, l.logf, l.clock)
+ r.observe(t.Context(), anomalyErr)
+ if calls != 1 {
+ t.Fatalf("first observation: %d calls", calls)
+ }
+ l.now = evNow.Add(30 * time.Second)
+ r.observe(t.Context(), anomalyErr)
+ if calls != 1 {
+ t.Fatalf("inside the backoff: %d calls", calls)
+ }
+ fail = false
+ l.now = evNow.Add(2 * time.Minute)
+ r.observe(t.Context(), anomalyErr)
+ if calls != 2 {
+ t.Fatalf("after the backoff: %d calls", calls)
+ }
+ for i := 0; i < 5; i++ {
+ l.now = l.now.Add(time.Hour)
+ r.observe(t.Context(), anomalyErr)
+ }
+ if calls != 2 {
+ t.Fatalf("after a delivered page: %d calls", calls)
+ }
+
+ l2 := &refusalLog{now: evNow}
+ always := 0
+ r2 := newMovesRefusals("ie-box", func(context.Context, string, string, string) error { always++; return errors.New("down") }, l2.logf, l2.clock)
+ for i := 0; i < 50; i++ {
+ r2.observe(t.Context(), anomalyErr)
+ l2.now = l2.now.Add(2 * time.Hour)
+ }
+ if always != movesPageAttempts {
+ t.Fatalf("a pager that always fails was called %d times, want %d", always, movesPageAttempts)
+ }
+}
+
+// TestMovesAnomalyPageIsNotDoubled (fix round 1, I4): while one page is in
+// flight, a concurrent observation of the same anomaly does not page again.
+func TestMovesAnomalyPageIsNotDoubled(t *testing.T) {
+ l := &refusalLog{now: evNow}
+ var mu sync.Mutex
+ calls := 0
+ entered, release := make(chan struct{}), make(chan struct{})
+ r := newMovesRefusals("ie-box", func(context.Context, string, string, string) error {
+ mu.Lock()
+ calls++
+ mu.Unlock()
+ close(entered)
+ <-release
+ return nil
+ }, l.logf, l.clock)
+ done := make(chan struct{})
+ go func() { r.observe(context.Background(), anomalyErr); close(done) }()
+ <-entered
+ r.observe(t.Context(), anomalyErr) // in flight: skipped, does not block
+ close(release)
+ <-done
+ r.observe(t.Context(), anomalyErr)
+ mu.Lock()
+ defer mu.Unlock()
+ if calls != 1 {
+ t.Fatalf("%d pages, want one", calls)
+ }
+}