| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "strings" |
| 8 | "sync" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/alerting" |
| 13 | "github.com/carlosframework/platform/internal/park" |
| 14 | ) |
| 15 | |
| 16 | // Quiet refusals (Plan 5 Task 9a): the Task 5 provider's expected refusals |
| 17 | // are rate-limited in the regional worker's log, and an anomaly pages once |
| 18 | // through the box pager. |
| 19 | |
| 20 | type refusalLog struct { |
| 21 | mu sync.Mutex |
| 22 | lines []string |
| 23 | now time.Time |
| 24 | } |
| 25 | |
| 26 | func (l *refusalLog) logf(format string, args ...any) { |
| 27 | l.mu.Lock() |
| 28 | defer l.mu.Unlock() |
| 29 | l.lines = append(l.lines, fmt.Sprintf(format, args...)) |
| 30 | } |
| 31 | |
| 32 | func (l *refusalLog) clock() time.Time { |
| 33 | l.mu.Lock() |
| 34 | defer l.mu.Unlock() |
| 35 | return l.now |
| 36 | } |
| 37 | |
| 38 | func (l *refusalLog) take() []string { |
| 39 | l.mu.Lock() |
| 40 | defer l.mu.Unlock() |
| 41 | out := l.lines |
| 42 | l.lines = nil |
| 43 | return out |
| 44 | } |
| 45 | |
| 46 | // TestExpectedMovesRefusalsAreRateLimited: a live move, residue the close |
| 47 | // does not vouch for and a named dependency (the wake-source gate's) are |
| 48 | // each logged once, then not again within the hour; an unexpected error is |
| 49 | // logged every time. |
| 50 | func TestExpectedMovesRefusalsAreRateLimited(t *testing.T) { |
| 51 | l := &refusalLog{now: evNow} |
| 52 | var pages pagesSeen |
| 53 | r := newMovesRefusals("ie-box", pages.hook(), l.logf, l.clock) |
| 54 | boom := errors.New("listing failed") |
| 55 | pass := errors.Join( |
| 56 | &MovesSweepRefusal{Host: "a.example", Reason: "the record carries the live move mv_x"}, |
| 57 | &NotMovesEvidence{Host: "b.example", Generation: "g", Reason: "written after the close"}, |
| 58 | &ClaimDependencyError{Contracts: []string{movesWakeSourceContract}}, |
| 59 | boom, |
| 60 | ) |
| 61 | r.report(pass) |
| 62 | if got := l.take(); len(got) != 4 { |
| 63 | t.Fatalf("the first pass logged %q", got) |
| 64 | } |
| 65 | l.now = evNow.Add(30 * time.Minute) |
| 66 | r.report(pass) |
| 67 | if got := l.take(); len(got) != 1 || !strings.Contains(got[0], boom.Error()) { |
| 68 | t.Fatalf("within the hour the pass logged %q; want only the unexpected error", got) |
| 69 | } |
| 70 | l.now = evNow.Add(61 * time.Minute) |
| 71 | r.report(pass) |
| 72 | if got := l.take(); len(got) != 4 { |
| 73 | t.Fatalf("after the hour the pass logged %q", got) |
| 74 | } |
| 75 | if len(pages) != 0 { |
| 76 | t.Fatalf("an expected refusal paged: %+v", pages) |
| 77 | } |
| 78 | } |
| 79 | |
| 80 | // TestMovesAnomalyPagesOnce: an anomaly from either method pages once per |
| 81 | // host and reason, as a warning naming the host and the box and carrying no |
| 82 | // path; an expected refusal pages nothing, and the error reaches the worker |
| 83 | // unchanged. |
| 84 | func TestMovesAnomalyPagesOnce(t *testing.T) { |
| 85 | l := &refusalLog{now: evNow} |
| 86 | var pages pagesSeen |
| 87 | r := newMovesRefusals("ie-box", pages.hook(), l.logf, l.clock) |
| 88 | f := newEvFixture(t) |
| 89 | f.rec.CreatedAt = evCreated.Add(time.Hour) // a record of another incarnation: an anomaly |
| 90 | e := r.evidence(f.provider()) |
| 91 | obs, census := f.observe(t), f.census(t) |
| 92 | for i := 0; i < 3; i++ { |
| 93 | _, _, err := e.VerifyRetirement(t.Context(), obs, census) |
| 94 | var refusal *MovesSweepRefusal |
| 95 | if !errors.As(err, &refusal) || !refusal.Anomaly { |
| 96 | t.Fatalf("the wrapper changed the refusal: %v", err) |
| 97 | } |
| 98 | if _, err := e.VerifyMovesSweep(t.Context(), obs, census); !errors.As(err, &refusal) { |
| 99 | t.Fatalf("the wrapper changed the sweep refusal: %v", err) |
| 100 | } |
| 101 | } |
| 102 | if len(pages) != 1 { |
| 103 | t.Fatalf("pages %+v, want one", pages) |
| 104 | } |
| 105 | subject, body := movesAnomalyPage(evHost, "ie-box") |
| 106 | if p := pages[0]; p.level != alerting.LevelWarning || p.subject != subject || p.body != body || strings.Contains(p.subject+p.body, "/") { |
| 107 | t.Fatalf("page %+v", p) |
| 108 | } |
| 109 | // An expected refusal of another host pages nothing. |
| 110 | g := newEvFixture(t) |
| 111 | g.rec.LastMove = nil |
| 112 | if _, _, err := r.evidence(g.provider()).VerifyRetirement(t.Context(), g.observe(t), g.census(t)); err == nil { |
| 113 | t.Fatal("no refusal") |
| 114 | } |
| 115 | if len(pages) != 1 { |
| 116 | t.Fatalf("an expected refusal paged: %+v", pages) |
| 117 | } |
| 118 | } |
| 119 | |
| 120 | // TestRegionalWorkerRefusalsAreQuiet: through the claim's real regional |
| 121 | // worker, an anomalous host refuses the sweep every pass, pages once and is |
| 122 | // logged once within the hour. |
| 123 | func TestRegionalWorkerRefusalsAreQuiet(t *testing.T) { |
| 124 | l := &refusalLog{now: evNow} |
| 125 | var pages pagesSeen |
| 126 | r := newMovesRefusals("ie-box", pages.hook(), l.logf, l.clock) |
| 127 | f := newEvFixture(t) |
| 128 | f.rec.CreatedAt = evCreated.Add(time.Hour) |
| 129 | f.lane(evMoveID, "move.json") // the census finds the host by its objects |
| 130 | w := f.worker(t, evNow) |
| 131 | e := r.evidence(f.provider()) |
| 132 | w.Evidence = e |
| 133 | var verifier park.MoveSweepVerifier = e |
| 134 | w.Regional.Engine.VerifyMovesSweep = verifier.VerifyMovesSweep |
| 135 | for i := 0; i < 3; i++ { |
| 136 | err := w.Run(t.Context()) |
| 137 | if err == nil { |
| 138 | t.Fatal("the sweep ran on an anomaly") |
| 139 | } |
| 140 | r.report(err) |
| 141 | } |
| 142 | if lines := l.take(); len(pages) != 1 || len(lines) != 1 { |
| 143 | t.Fatalf("pages %+v, log lines %q over three passes", pages, lines) |
| 144 | } |
| 145 | } |
| 146 | |
| 147 | // anomalyErr is one anomalous refusal of evHost. |
| 148 | var anomalyErr = &MovesSweepRefusal{Host: evHost, Reason: "the record is not the holder's incarnation", Anomaly: true} |
| 149 | |
| 150 | // TestMovesAnomalyPageRetriesUntilDelivered (fix round 1, I4): a page that |
| 151 | // fails is not marked delivered; a later observation, past the backoff, |
| 152 | // pages again; once one succeeds nothing more is sent. A pager that always |
| 153 | // fails is tried a bounded number of times. |
| 154 | func TestMovesAnomalyPageRetriesUntilDelivered(t *testing.T) { |
| 155 | l := &refusalLog{now: evNow} |
| 156 | calls, fail := 0, true |
| 157 | r := newMovesRefusals("ie-box", func(context.Context, string, string, string) error { |
| 158 | calls++ |
| 159 | if fail { |
| 160 | return errors.New("the webhook answered 503") |
| 161 | } |
| 162 | return nil |
| 163 | }, l.logf, l.clock) |
| 164 | r.observe(t.Context(), anomalyErr) |
| 165 | if calls != 1 { |
| 166 | t.Fatalf("first observation: %d calls", calls) |
| 167 | } |
| 168 | l.now = evNow.Add(30 * time.Second) |
| 169 | r.observe(t.Context(), anomalyErr) |
| 170 | if calls != 1 { |
| 171 | t.Fatalf("inside the backoff: %d calls", calls) |
| 172 | } |
| 173 | fail = false |
| 174 | l.now = evNow.Add(2 * time.Minute) |
| 175 | r.observe(t.Context(), anomalyErr) |
| 176 | if calls != 2 { |
| 177 | t.Fatalf("after the backoff: %d calls", calls) |
| 178 | } |
| 179 | for i := 0; i < 5; i++ { |
| 180 | l.now = l.now.Add(time.Hour) |
| 181 | r.observe(t.Context(), anomalyErr) |
| 182 | } |
| 183 | if calls != 2 { |
| 184 | t.Fatalf("after a delivered page: %d calls", calls) |
| 185 | } |
| 186 | |
| 187 | l2 := &refusalLog{now: evNow} |
| 188 | always := 0 |
| 189 | r2 := newMovesRefusals("ie-box", func(context.Context, string, string, string) error { always++; return errors.New("down") }, l2.logf, l2.clock) |
| 190 | for i := 0; i < 50; i++ { |
| 191 | r2.observe(t.Context(), anomalyErr) |
| 192 | l2.now = l2.now.Add(2 * time.Hour) |
| 193 | } |
| 194 | if always != movesPageAttempts { |
| 195 | t.Fatalf("a pager that always fails was called %d times, want %d", always, movesPageAttempts) |
| 196 | } |
| 197 | } |
| 198 | |
| 199 | // TestMovesAnomalyPageIsNotDoubled (fix round 1, I4): while one page is in |
| 200 | // flight, a concurrent observation of the same anomaly does not page again. |
| 201 | func TestMovesAnomalyPageIsNotDoubled(t *testing.T) { |
| 202 | l := &refusalLog{now: evNow} |
| 203 | var mu sync.Mutex |
| 204 | calls := 0 |
| 205 | entered, release := make(chan struct{}), make(chan struct{}) |
| 206 | r := newMovesRefusals("ie-box", func(context.Context, string, string, string) error { |
| 207 | mu.Lock() |
| 208 | calls++ |
| 209 | mu.Unlock() |
| 210 | close(entered) |
| 211 | <-release |
| 212 | return nil |
| 213 | }, l.logf, l.clock) |
| 214 | done := make(chan struct{}) |
| 215 | go func() { r.observe(context.Background(), anomalyErr); close(done) }() |
| 216 | <-entered |
| 217 | r.observe(t.Context(), anomalyErr) // in flight: skipped, does not block |
| 218 | close(release) |
| 219 | <-done |
| 220 | r.observe(t.Context(), anomalyErr) |
| 221 | mu.Lock() |
| 222 | defer mu.Unlock() |
| 223 | if calls != 1 { |
| 224 | t.Fatalf("%d pages, want one", calls) |
| 225 | } |
| 226 | } |
| 227 | |
amadan