| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "log" |
| 6 | "math" |
| 7 | "os" |
| 8 | "sort" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/hostclaim" |
| 12 | "github.com/carlosframework/platform/internal/registry" |
| 13 | ) |
| 14 | |
| 15 | // Region HA spec 1 §5.4: with CARLOS_REGION_HA on, a relocatable host is |
| 16 | // published as home only by the box its claim's op.exec names, carrying |
| 17 | // the claim's holder.stamp so the edge's home map believes the newest |
| 18 | // installation (HA4). The edge never reads claims; this publisher does, on |
| 19 | // a slow cache, because the safety gate is the activator's fresh read at |
| 20 | // every spawn (claim C4), not this file. |
| 21 | |
| 22 | const ( |
| 23 | // edgesClaimTTL is how long one claim read stands: reconcile's sweep |
| 24 | // (H9). A held row's entry can be stale this long, which can only send |
| 25 | // a request to a box that then refuses to spawn. |
| 26 | edgesClaimTTL = 10 * time.Minute |
| 27 | // edgesClaimReadsPerPass bounds the GETs one pass makes, so a box with |
| 28 | // thousands of hosts spreads its reads over several passes instead of |
| 29 | // stalling one. A host not read yet publishes as it did before this |
| 30 | // file existed. |
| 31 | edgesClaimReadsPerPass = 256 |
| 32 | // edgesClaimPassEvery is how often the loop runs an extra pass while |
| 33 | // claim reads are due. The registry timer alone (5 to 6 minutes) would |
| 34 | // take 40 minutes to refresh 2,000 hosts at 256 a pass; with these |
| 35 | // passes every host is re-read within about edgesClaimTTL plus four |
| 36 | // minutes. |
| 37 | edgesClaimPassEvery = 30 * time.Second |
| 38 | // edgesClaimPassDeadline bounds one pass's claim reads, so a store that |
| 39 | // stops answering cannot stall the edges file. A read the deadline cuts |
| 40 | // short is a failed read: the host keeps its last decision. |
| 41 | edgesClaimPassDeadline = 20 * time.Second |
| 42 | // edgesClaimLogEvery: a run of passes with failed claim reads is logged |
| 43 | // on its first pass and every this many after. |
| 44 | edgesClaimLogEvery = 60 |
| 45 | ) |
| 46 | |
| 47 | // edgesDecision is what one claim says about this box and one host. |
| 48 | type edgesDecision struct { |
| 49 | withhold bool // the claim names another executor, or none |
| 50 | stamp int64 // holder.stamp when this box is the executor |
| 51 | } |
| 52 | |
| 53 | // edgesClaimEntry is one host's last decision and last read attempt. A |
| 54 | // failed read moves attempted and keeps the decision, so a claim that |
| 55 | // cannot be read goes to the back of the queue instead of holding its |
| 56 | // place at the front (claim C6: an error is never absence). |
| 57 | type edgesClaimEntry struct { |
| 58 | decision edgesDecision |
| 59 | decided bool // some read of this claim has succeeded |
| 60 | attempted time.Time |
| 61 | } |
| 62 | |
| 63 | // edgesClaims caches one decision per host. It is used only by the edges |
| 64 | // file loop's goroutine. |
| 65 | type edgesClaims struct { |
| 66 | self string // this box's reference: the edges file's box_id |
| 67 | read func(ctx context.Context, host string) (hostclaim.Observed, error) |
| 68 | now func() time.Time |
| 69 | entries map[string]edgesClaimEntry |
| 70 | hosts []string // the relocatable hosts of the last pass |
| 71 | // passDeadline bounds one pass's claim reads (edgesClaimPassDeadline). |
| 72 | passDeadline time.Duration |
| 73 | logf func(format string, a ...any) |
| 74 | failingRun int // consecutive passes with a failed claim read |
| 75 | // addrs is the box's own addresses, read before every publication |
| 76 | // (region HA spec 2 §6). It rides here because this cache exists |
| 77 | // exactly when region HA is on. |
| 78 | addrs AddressSource |
| 79 | } |
| 80 | |
| 81 | func newEdgesClaims(self string, read func(context.Context, string) (hostclaim.Observed, error)) *edgesClaims { |
| 82 | return &edgesClaims{self: self, read: read, now: time.Now, entries: map[string]edgesClaimEntry{}, |
| 83 | passDeadline: edgesClaimPassDeadline, logf: log.Printf, addrs: boxAddressSource(os.Getenv)} |
| 84 | } |
| 85 | |
| 86 | // relocatableRow is spec 1 §2's relocatable host as a registry row shows |
| 87 | // it: a provisioned instance whose process is the activator's own exec |
| 88 | // child. Unit-backed, addr-backed, static and service routes keep their |
| 89 | // box and are published as before; sidecars are never published at all. |
| 90 | func relocatableRow(r registry.Route) bool { |
| 91 | return r.Provisioned && r.Kind == "instance" && r.Backing == "" && r.Unit == "" && |
| 92 | r.Addr == "" && r.Addrs == "" && r.Socket != "" |
| 93 | } |
| 94 | |
| 95 | // decideClaim turns one observation into a decision. An absent claim |
| 96 | // leaves the host as it was before claims existed: a region can only be |
| 97 | // given a second box once every host has one (spec §8). |
| 98 | func (c *edgesClaims) decideClaim(obs hostclaim.Observed) edgesDecision { |
| 99 | if obs.Absent || obs.Claim == nil { |
| 100 | return edgesDecision{} |
| 101 | } |
| 102 | cl := obs.Claim |
| 103 | if cl.Holder == nil || cl.Op.Exec != c.self { |
| 104 | return edgesDecision{withhold: true} |
| 105 | } |
| 106 | stamp := cl.Holder.Stamp |
| 107 | if stamp > math.MaxInt64 { |
| 108 | stamp = 0 |
| 109 | } |
| 110 | return edgesDecision{stamp: int64(stamp)} |
| 111 | } |
| 112 | |
| 113 | func (c *edgesClaims) dueAt(h string, now time.Time) bool { |
| 114 | e, ok := c.entries[h] |
| 115 | return !ok || now.Sub(e.attempted) >= edgesClaimTTL |
| 116 | } |
| 117 | |
| 118 | // backlog reports whether any host of the last pass has a read due. |
| 119 | func (c *edgesClaims) backlog(now time.Time) bool { |
| 120 | for _, h := range c.hosts { |
| 121 | if c.dueAt(h, now) { |
| 122 | return true |
| 123 | } |
| 124 | } |
| 125 | return false |
| 126 | } |
| 127 | |
| 128 | // decide returns a decision for each host that has one. Hosts whose last |
| 129 | // attempt is older than edgesClaimTTL are read, never-attempted first and |
| 130 | // then oldest attempt first, up to edgesClaimReadsPerPass. A read that |
| 131 | // fails keeps the previous decision; a host with no successful read has |
| 132 | // none, and publishes as before. Once ctx ends no further read is made: |
| 133 | // the hosts not reached stay due, in their place, for a later pass. |
| 134 | func (c *edgesClaims) decide(ctx context.Context, hosts []string) map[string]edgesDecision { |
| 135 | now := c.now() |
| 136 | c.hosts = append(c.hosts[:0], hosts...) |
| 137 | want := make(map[string]bool, len(hosts)) |
| 138 | var due []string |
| 139 | for _, h := range hosts { |
| 140 | want[h] = true |
| 141 | if c.dueAt(h, now) { |
| 142 | due = append(due, h) |
| 143 | } |
| 144 | } |
| 145 | for h := range c.entries { |
| 146 | if !want[h] { |
| 147 | delete(c.entries, h) |
| 148 | } |
| 149 | } |
| 150 | sort.Slice(due, func(i, j int) bool { |
| 151 | ei, iok := c.entries[due[i]] |
| 152 | ej, jok := c.entries[due[j]] |
| 153 | if iok != jok { |
| 154 | return !iok // never attempted first |
| 155 | } |
| 156 | if !ei.attempted.Equal(ej.attempted) { |
| 157 | return ei.attempted.Before(ej.attempted) |
| 158 | } |
| 159 | return due[i] < due[j] |
| 160 | }) |
| 161 | if len(due) > edgesClaimReadsPerPass { |
| 162 | due = due[:edgesClaimReadsPerPass] |
| 163 | } |
| 164 | var failed, made int |
| 165 | var firstErr error |
| 166 | var firstHost string |
| 167 | for _, h := range due { |
| 168 | if ctx.Err() != nil { |
| 169 | break |
| 170 | } |
| 171 | e := c.entries[h] |
| 172 | e.attempted = now |
| 173 | made++ |
| 174 | if obs, err := c.read(ctx, h); err == nil { |
| 175 | e.decision, e.decided = c.decideClaim(obs), true |
| 176 | } else { |
| 177 | if failed == 0 { |
| 178 | firstErr, firstHost = err, h |
| 179 | } |
| 180 | failed++ |
| 181 | } |
| 182 | c.entries[h] = e |
| 183 | } |
| 184 | c.logReads(made, failed, firstHost, firstErr) |
| 185 | out := make(map[string]edgesDecision, len(c.entries)) |
| 186 | for h, e := range c.entries { |
| 187 | if e.decided { |
| 188 | out[h] = e.decision |
| 189 | } |
| 190 | } |
| 191 | return out |
| 192 | } |
| 193 | |
| 194 | // logReads logs a pass's failed claim reads, throttled: the first pass of |
| 195 | // a failing run and every edgesClaimLogEvery-th after it, then once when a |
| 196 | // pass's reads all succeed again. |
| 197 | func (c *edgesClaims) logReads(made, failed int, host string, err error) { |
| 198 | if failed > 0 { |
| 199 | c.failingRun++ |
| 200 | if c.failingRun == 1 || c.failingRun%edgesClaimLogEvery == 0 { |
| 201 | c.logf("agent: edges file: %d of %d claim reads failed (failing pass #%d), first %s: %v; those hosts keep their last decision", failed, made, c.failingRun, host, err) |
| 202 | } |
| 203 | return |
| 204 | } |
| 205 | if made > 0 && c.failingRun > 0 { |
| 206 | c.logf("agent: edges file: claim reads recovered after %d failing passes", c.failingRun) |
| 207 | c.failingRun = 0 |
| 208 | } |
| 209 | } |
| 210 | |
| 211 | // applyEdgesClaims withholds or stamps every relocatable row in hosts |
| 212 | // (edgesFileWrite's served map, before aliases join it, so an alias of a |
| 213 | // withheld host is withheld with it and an alias of a stamped host takes |
| 214 | // its stamp). claims nil leaves hosts untouched. A host hostclaim.Key |
| 215 | // rejects has no claim to read and publishes as before. The pass's reads |
| 216 | // share one deadline, claims.passDeadline. |
| 217 | func applyEdgesClaims(ctx context.Context, claims *edgesClaims, routes []registry.Route, hosts map[string]edgeHost) { |
| 218 | if claims == nil { |
| 219 | return |
| 220 | } |
| 221 | var reloc []string |
| 222 | for _, r := range routes { |
| 223 | if _, served := hosts[r.Host]; served && relocatableRow(r) { |
| 224 | if _, err := hostclaim.Key(r.Host); err != nil { |
| 225 | continue |
| 226 | } |
| 227 | reloc = append(reloc, r.Host) |
| 228 | } |
| 229 | } |
| 230 | ctx, cancel := context.WithTimeout(ctx, claims.passDeadline) |
| 231 | defer cancel() |
| 232 | for host, d := range claims.decide(ctx, reloc) { |
| 233 | h, ok := hosts[host] |
| 234 | if !ok { |
| 235 | continue |
| 236 | } |
| 237 | if d.withhold { |
| 238 | delete(hosts, host) |
| 239 | continue |
| 240 | } |
| 241 | h.Stamp = d.stamp |
| 242 | hosts[host] = h |
| 243 | } |
| 244 | } |
| 245 | |
| 246 | // edgesLoopState is edgesFileLoop's per-tick decision: the registry and |
| 247 | // timer schedule, plus, with region HA on, an extra pass every |
| 248 | // edgesClaimPassEvery while claim reads are due. A claim pass is a |
| 249 | // publication like any other: it waits out the schedule's retry backoff |
| 250 | // after a failed write, and its own outcome moves that backoff, so a bucket |
| 251 | // that refuses every write is not asked every 30 seconds (Astra plan R2-1). |
| 252 | type edgesLoopState struct { |
| 253 | sched *edgesFileSchedule |
| 254 | claims *edgesClaims |
| 255 | claimsNext time.Time |
| 256 | schedDue bool |
| 257 | claimDue bool |
| 258 | } |
| 259 | |
| 260 | // due reports whether to run a pass on this tick. |
| 261 | func (s *edgesLoopState) due(now time.Time, fp string, fpOK bool) bool { |
| 262 | s.schedDue = s.sched.due(now, fp, fpOK) |
| 263 | s.claimDue = s.claims != nil && !now.Before(s.claimsNext) && !now.Before(s.sched.retryAt) && s.claims.backlog(now) |
| 264 | return s.schedDue || s.claimDue |
| 265 | } |
| 266 | |
| 267 | // timed reports whether the pass due just allowed is the timer's. |
| 268 | func (s *edgesLoopState) timed() bool { return s.schedDue && s.sched.timed } |
| 269 | |
| 270 | // done records the pass's outcome. Every pass, claim passes included, goes |
| 271 | // through the schedule: a failure advances its backoff, and a success |
| 272 | // clears it and records the fingerprint the pass published. A claim pass |
| 273 | // is never the timer's (due cleared timed), so the timer does not move. |
| 274 | func (s *edgesLoopState) done(now time.Time, ok bool) { |
| 275 | if s.schedDue || s.claimDue { |
| 276 | s.sched.done(now, ok) |
| 277 | } |
| 278 | if s.claims != nil { |
| 279 | s.claimsNext = now.Add(edgesClaimPassEvery) |
| 280 | } |
| 281 | } |
| 282 | |
amadan