| 1 | package host |
| 2 | |
| 3 | // Quiet refusals (moves Plan 5 Task 9a; the Task 5 rulings). The claim's |
| 4 | // regional worker asks Task 5's provider about every host at rest, every |
| 5 | // minute, and an ordinary state refuses: a live move, residue the move's |
| 6 | // close does not vouch for, a joining claim with no holder to name a record |
| 7 | // (a free host's lanes are the claim's own free-host proof, |
| 8 | // claimFreeMoveProof). The worker reports each refusal, so the agent's log would repeat them every |
| 9 | // minute. movesRefusals keeps them quiet: |
| 10 | // |
| 11 | // - report is the worker's report: a provider refusal (MovesSweepRefusal, |
| 12 | // NotMovesEvidence) or a named dependency (ClaimDependencyError) is |
| 13 | // logged at most once an hour per host and reason; anything else is |
| 14 | // logged every time, as before; |
| 15 | // - evidence wraps the provider: a refusal marked Anomaly (the history |
| 16 | // contradicts itself, or something wrote where no move could) pages once |
| 17 | // per host and reason through the box pager, as a warning: marked sent |
| 18 | // only once delivered, never twice at once, a failure retried with |
| 19 | // backoff a bounded number of times. The worker gets every error |
| 20 | // unchanged. |
| 21 | // |
| 22 | // Both are wired only with CARLOS_MOVES_PHASE1 and CARLOS_HOST_CLAIM on; |
| 23 | // otherwise the worker's report is the plain log line it always was. |
| 24 | |
| 25 | import ( |
| 26 | "context" |
| 27 | "errors" |
| 28 | "fmt" |
| 29 | "strings" |
| 30 | "sync" |
| 31 | "time" |
| 32 | |
| 33 | "github.com/carlosframework/platform/internal/alerting" |
| 34 | "github.com/carlosframework/platform/internal/hostbox" |
| 35 | "github.com/carlosframework/platform/internal/hostclaim" |
| 36 | "github.com/carlosframework/platform/internal/lease" |
| 37 | "github.com/carlosframework/platform/internal/park" |
| 38 | ) |
| 39 | |
| 40 | // movesRefusalLogEvery is how often one expected refusal is logged. |
| 41 | const movesRefusalLogEvery = time.Hour |
| 42 | |
| 43 | // movesPageAttempts bounds the tries one anomaly page gets. |
| 44 | const movesPageAttempts = 5 |
| 45 | |
| 46 | // movesRefusals is the agent's refusal log and anomaly pager for the moves |
| 47 | // evidence. Its memory is the process's: a restarted agent logs and pages |
| 48 | // each standing condition once more. |
| 49 | type movesRefusals struct { |
| 50 | box string |
| 51 | page func(ctx context.Context, level, subject, body string) error |
| 52 | logf func(format string, args ...any) |
| 53 | now func() time.Time |
| 54 | |
| 55 | mu sync.Mutex |
| 56 | logged map[string]time.Time |
| 57 | pages map[string]*movesPageState |
| 58 | } |
| 59 | |
| 60 | // movesPageState is one anomaly's page: delivered once a send succeeds; |
| 61 | // while a send is in flight no other starts; a failed send is retried by a |
| 62 | // later observation after a doubling backoff, at most movesPageAttempts |
| 63 | // times in all. |
| 64 | type movesPageState struct { |
| 65 | delivered, inflight bool |
| 66 | failures int |
| 67 | next time.Time |
| 68 | } |
| 69 | |
| 70 | // movesPageBackoff is the wait after the first failed page; it doubles with |
| 71 | // each further failure, up to movesPageBackoffMax. |
| 72 | const ( |
| 73 | movesPageBackoff = time.Minute |
| 74 | movesPageBackoffMax = 30 * time.Minute |
| 75 | ) |
| 76 | |
| 77 | func newMovesRefusals(box string, page func(context.Context, string, string, string) error, logf func(string, ...any), now func() time.Time) *movesRefusals { |
| 78 | return &movesRefusals{box: box, page: page, logf: logf, now: now, logged: map[string]time.Time{}, pages: map[string]*movesPageState{}} |
| 79 | } |
| 80 | |
| 81 | // movesQuietEvidence is Task 5's provider with its anomalies paged. It is |
| 82 | // the regional evidence and the sweep verifier the claim registers. |
| 83 | type movesQuietEvidence struct { |
| 84 | movesRetirementEvidence |
| 85 | r *movesRefusals |
| 86 | } |
| 87 | |
| 88 | var ( |
| 89 | _ hostbox.RegionalEvidence = movesQuietEvidence{} |
| 90 | _ park.MoveSweepVerifier = movesQuietEvidence{} |
| 91 | ) |
| 92 | |
| 93 | func (r *movesRefusals) evidence(e movesRetirementEvidence) movesQuietEvidence { |
| 94 | return movesQuietEvidence{movesRetirementEvidence: e, r: r} |
| 95 | } |
| 96 | |
| 97 | func (e movesQuietEvidence) VerifyRetirement(ctx context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (string, time.Time, error) { |
| 98 | inc, at, err := e.movesRetirementEvidence.VerifyRetirement(ctx, o, versions) |
| 99 | if err != nil { |
| 100 | e.r.observe(ctx, err) |
| 101 | } |
| 102 | return inc, at, err |
| 103 | } |
| 104 | |
| 105 | func (e movesQuietEvidence) VerifyMovesSweep(ctx context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (park.MoveSweepProof, error) { |
| 106 | proof, err := e.movesRetirementEvidence.VerifyMovesSweep(ctx, o, versions) |
| 107 | if err != nil { |
| 108 | e.r.observe(ctx, err) |
| 109 | } |
| 110 | return proof, err |
| 111 | } |
| 112 | |
| 113 | // observe pages an anomaly once per host and reason (fix round 1, I4): it |
| 114 | // is marked delivered only after the pager succeeds; a concurrent |
| 115 | // observation while a page is in flight sends nothing; a failed page is |
| 116 | // retried by a later observation after its backoff, at most |
| 117 | // movesPageAttempts times. |
| 118 | func (r *movesRefusals) observe(ctx context.Context, err error) { |
| 119 | var refusal *MovesSweepRefusal |
| 120 | if r == nil || r.page == nil || !errors.As(err, &refusal) || !refusal.Anomaly { |
| 121 | return |
| 122 | } |
| 123 | key := refusal.Host + "\x00" + refusal.Reason |
| 124 | now := r.now() |
| 125 | r.mu.Lock() |
| 126 | st := r.pages[key] |
| 127 | if st == nil { |
| 128 | st = &movesPageState{} |
| 129 | r.pages[key] = st |
| 130 | } |
| 131 | if st.delivered || st.inflight || st.failures >= movesPageAttempts || now.Before(st.next) { |
| 132 | r.mu.Unlock() |
| 133 | return |
| 134 | } |
| 135 | st.inflight = true |
| 136 | r.mu.Unlock() |
| 137 | subject, body := movesAnomalyPage(refusal.Host, r.box) |
| 138 | perr := r.page(ctx, alerting.LevelWarning, subject, body) |
| 139 | r.mu.Lock() |
| 140 | st.inflight = false |
| 141 | if perr == nil { |
| 142 | st.delivered = true |
| 143 | r.mu.Unlock() |
| 144 | return |
| 145 | } |
| 146 | st.failures++ |
| 147 | wait := movesPageBackoff << (st.failures - 1) |
| 148 | if wait > movesPageBackoffMax || wait <= 0 { |
| 149 | wait = movesPageBackoffMax |
| 150 | } |
| 151 | st.next = now.Add(wait) |
| 152 | failures := st.failures |
| 153 | r.mu.Unlock() |
| 154 | if r.logf != nil { |
| 155 | if failures >= movesPageAttempts { |
| 156 | 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) |
| 157 | } else { |
| 158 | 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) |
| 159 | } |
| 160 | } |
| 161 | } |
| 162 | |
| 163 | // movesAnomalyPage is the anomaly's page: the subject (MV-N336) and the |
| 164 | // body (MV-N337). The reason stays in the agent's log, never in the page. |
| 165 | func movesAnomalyPage(host, box string) (subject, body string) { |
| 166 | subject = fmt.Sprintf("carlos move cleanup refused: %s", host) |
| 167 | body = fmt.Sprintf("The cleanup of %s on %s stopped: its move history does not add up, or something wrote where no move could. Nothing of %s is cleaned up in this region until an operator checks the agent's log on %s.", |
| 168 | host, box, host, box) |
| 169 | return subject, body |
| 170 | } |
| 171 | |
| 172 | // report is the regional worker's report: each error of the pass, the |
| 173 | // expected refusals at most once an hour per host and reason. |
| 174 | func (r *movesRefusals) report(err error) { |
| 175 | for _, leaf := range joinedErrors(err) { |
| 176 | key, expected := movesExpectedRefusal(leaf) |
| 177 | if !expected { |
| 178 | r.logf("agent: regional claims: %v", leaf) |
| 179 | continue |
| 180 | } |
| 181 | now := r.now() |
| 182 | r.mu.Lock() |
| 183 | last, seen := r.logged[key] |
| 184 | if !seen || !now.Before(last.Add(movesRefusalLogEvery)) { |
| 185 | r.logged[key] = now |
| 186 | seen = false |
| 187 | } |
| 188 | r.mu.Unlock() |
| 189 | if !seen { |
| 190 | r.logf("agent: regional claims: %v (logged at most once an hour while it stands)", leaf) |
| 191 | } |
| 192 | } |
| 193 | } |
| 194 | |
| 195 | // joinedErrors flattens errors.Join's tree into its leaves. |
| 196 | func joinedErrors(err error) []error { |
| 197 | if err == nil { |
| 198 | return nil |
| 199 | } |
| 200 | j, ok := err.(interface{ Unwrap() []error }) |
| 201 | if !ok { |
| 202 | return []error{err} |
| 203 | } |
| 204 | var out []error |
| 205 | for _, e := range j.Unwrap() { |
| 206 | out = append(out, joinedErrors(e)...) |
| 207 | } |
| 208 | return out |
| 209 | } |
| 210 | |
| 211 | // movesExpectedRefusal names a refusal that stands until its condition |
| 212 | // changes: the provider's typed refusals and a named dependency. |
| 213 | func movesExpectedRefusal(err error) (string, bool) { |
| 214 | var refusal *MovesSweepRefusal |
| 215 | var notMoves *NotMovesEvidence |
| 216 | var dep *ClaimDependencyError |
| 217 | switch { |
| 218 | case errors.As(err, &refusal): |
| 219 | return "sweep\x00" + refusal.Host + "\x00" + refusal.Reason, true |
| 220 | case errors.As(err, ¬Moves): |
| 221 | return "not-moves\x00" + notMoves.Host + "\x00" + notMoves.Generation + "\x00" + notMoves.Reason, true |
| 222 | case errors.As(err, &dep): |
| 223 | return "dependency\x00" + strings.Join(dep.Contracts, "\x00"), true |
| 224 | } |
| 225 | return "", false |
| 226 | } |
| 227 | |
amadan