| 1 | package host |
| 2 | |
| 3 | // The box agent's reads through the S3 change index (S3 change-index design |
| 4 | // §4.3-§4.5, discussion #336). Armed only by CARLOS_CHANGE_INDEX=read; unset, |
| 5 | // every pointer below stays nil and every loop reads the bucket exactly as |
| 6 | // before. |
| 7 | |
| 8 | import ( |
| 9 | "context" |
| 10 | "crypto/ecdh" |
| 11 | "crypto/ed25519" |
| 12 | "fmt" |
| 13 | "log" |
| 14 | "math/rand/v2" |
| 15 | "os" |
| 16 | "sort" |
| 17 | "strings" |
| 18 | "sync" |
| 19 | "sync/atomic" |
| 20 | "time" |
| 21 | |
| 22 | "github.com/carlosframework/platform/internal/activator" |
| 23 | "github.com/carlosframework/platform/internal/domains" |
| 24 | "github.com/carlosframework/platform/internal/edgesfile" |
| 25 | "github.com/carlosframework/platform/internal/hostwrite" |
| 26 | "github.com/carlosframework/platform/internal/provision" |
| 27 | "github.com/carlosframework/platform/internal/registry" |
| 28 | "github.com/carlosframework/platform/internal/release" |
| 29 | ) |
| 30 | |
| 31 | // changeIndexArmed reports whether this process reads through the change |
| 32 | // index. Only the exact value "read" arms it (spec §7). |
| 33 | func changeIndexArmed(getenv func(string) string) bool { |
| 34 | return getenv("CARLOS_CHANGE_INDEX") == "read" |
| 35 | } |
| 36 | |
| 37 | // armedChangeWatch is the process's one watch on the change index, and |
| 38 | // armedRecordView the record view that follows it. RunAgent sets both once |
| 39 | // at start-up when armed; nil means today's behaviour everywhere. |
| 40 | var ( |
| 41 | armedChangeWatch atomic.Pointer[release.ChangeWatch] |
| 42 | armedRecordView atomic.Pointer[provision.RecordView] |
| 43 | ) |
| 44 | |
| 45 | // backstopFullDue is the backstop's request for the unrestricted full pass: |
| 46 | // set after each Rebuild, taken by fastTick on the fast converge loop's own |
| 47 | // goroutine (fastDeps' maps are single-goroutine by construction). |
| 48 | var backstopFullDue atomic.Bool |
| 49 | |
| 50 | // armChangeIndex arms the change index for this process: a watch on the |
| 51 | // deployment store, and a record view over the same store. A fleet box |
| 52 | // stays disarmed with the variable set: its store door has no conditional |
| 53 | // read, so its watch would be unusable and every Refresh would walk anyway |
| 54 | // (spec §3.4). The view's store must take no agent lock on any read path |
| 55 | // (RecordView.mu is a leaf, spec §4.3): open hands back the plain deployment |
| 56 | // store the other loops use. |
| 57 | func armChangeIndex(ctx context.Context, getenv func(string) string, open func(context.Context) (release.Store, error)) *provision.RecordView { |
| 58 | if !changeIndexArmed(getenv) { |
| 59 | return nil |
| 60 | } |
| 61 | if shouldSkipPublisherElection(getenv) { |
| 62 | log.Printf("agent: change index: CARLOS_CHANGE_INDEX=read is ignored on a fleet box; record reads stay on the walk") |
| 63 | return nil |
| 64 | } |
| 65 | store, err := open(ctx) |
| 66 | if err != nil { |
| 67 | log.Printf("agent: change index: not armed: %v", err) |
| 68 | return nil |
| 69 | } |
| 70 | watch := release.NewChangeWatch(store) |
| 71 | v := provision.NewRecordView(store, watch) |
| 72 | armedChangeWatch.Store(watch) |
| 73 | armedRecordView.Store(v) |
| 74 | log.Printf("agent: change index armed: record walks read the record view (backstop %s)", release.ChangeIndexBackstop) |
| 75 | return v |
| 76 | } |
| 77 | |
| 78 | // newChangeGate is a read gate over the armed watch, built by a loop when it |
| 79 | // starts: RunAgent arms the index before it starts any loop, so the watch a |
| 80 | // gate is built over is the one the process keeps. Disarmed, the watch is nil |
| 81 | // and every read is due, as before the index (spec §5). |
| 82 | func newChangeGate() *release.ReadGate { return newChangeGateOver(armedChangeWatch.Load()) } |
| 83 | |
| 84 | // newChangeGateOver is the one place the gated reads' maximum age is set: a |
| 85 | // successful read stands in for the next one for at most the backstop. |
| 86 | func newChangeGateOver(w *release.ChangeWatch) *release.ReadGate { |
| 87 | return release.NewReadGate(w, release.ChangeIndexBackstop) |
| 88 | } |
| 89 | |
| 90 | // viewUse remembers whether the last viewStore call used the view, so the |
| 91 | // switch is logged once per change of state, never per tick. |
| 92 | var viewUse struct { |
| 93 | mu sync.Mutex |
| 94 | last int // 0 none yet, 1 using the view, 2 walking |
| 95 | } |
| 96 | |
| 97 | func noteViewUse(ok bool, why string) { |
| 98 | state := 2 |
| 99 | if ok { |
| 100 | state = 1 |
| 101 | } |
| 102 | viewUse.mu.Lock() |
| 103 | changed := viewUse.last != state |
| 104 | viewUse.last = state |
| 105 | viewUse.mu.Unlock() |
| 106 | if !changed { |
| 107 | return |
| 108 | } |
| 109 | if ok { |
| 110 | log.Printf("agent: record view: record walks read the view") |
| 111 | return |
| 112 | } |
| 113 | log.Printf("agent: record view: %s; record walks read the bucket until it can vouch again", why) |
| 114 | } |
| 115 | |
| 116 | // viewStore is the store a record-family walker reads: the armed view's |
| 117 | // current snapshot over live, refreshed first, when the refresh returned nil |
| 118 | // and the snapshot is Complete. Otherwise ok is false and s is live, so the |
| 119 | // caller walks the bucket as before the index (spec §3.4). |
| 120 | // |
| 121 | // Only record-family walkers take it (Reconcile, ReconcileSidecars, |
| 122 | // steering), and the moves pass, which asks it only whether there is any |
| 123 | // move work and acts on its own walk of the bucket (movesIdle). Its ListDirs of |
| 124 | // apps/<a>/ names only apps with record-family state, so app enumerators |
| 125 | // (hostops.Apps, purge and activation sweeps, mail, economics) must never |
| 126 | // get it. |
| 127 | // |
| 128 | // live must take no agent lock on any read path: RecordView.mu is a leaf |
| 129 | // (lock order w.pass -> reconcileMu -> RecordView.mu). |
| 130 | func viewStore(ctx context.Context, live release.Store) (s release.Store, epoch uint64, ok bool) { |
| 131 | v := armedRecordView.Load() |
| 132 | if v == nil || live == nil { |
| 133 | return live, 0, false |
| 134 | } |
| 135 | err := v.Refresh(ctx) |
| 136 | sn := v.Snapshot() |
| 137 | switch { |
| 138 | case err != nil: |
| 139 | noteViewUse(false, fmt.Sprintf("refresh failed (%v)", err)) |
| 140 | return live, 0, false |
| 141 | case sn == nil || !sn.Complete: |
| 142 | noteViewUse(false, "the snapshot is incomplete") |
| 143 | return live, 0, false |
| 144 | } |
| 145 | noteViewUse(true, "") |
| 146 | return sn.Store(live), sn.Epoch, true |
| 147 | } |
| 148 | |
| 149 | // recordViewBackstopLoop is the backstop (spec §4.4): Rebuild at start and |
| 150 | // every release.ChangeIndexBackstop, jittered by +-10%, each followed by the |
| 151 | // unrestricted full pass, so every leg is re-run over everything at least |
| 152 | // that often. The full pass runs on the fast converge loop's goroutine |
| 153 | // (backstopFullDue); on a box where that loop is not armed only the |
| 154 | // rebuild runs. |
| 155 | func recordViewBackstopLoop(ctx context.Context, v *provision.RecordView) { |
| 156 | for { |
| 157 | recordViewBackstop(ctx, v, &backstopFullDue) |
| 158 | select { |
| 159 | case <-ctx.Done(): |
| 160 | return |
| 161 | case <-time.After(backstopEvery()): |
| 162 | } |
| 163 | } |
| 164 | } |
| 165 | |
| 166 | // recordViewBackstop is one backstop round: the rebuild, then the request |
| 167 | // for a full pass, whether or not the rebuild succeeded (a failed one |
| 168 | // leaves the view unable to vouch, so that pass walks). |
| 169 | func recordViewBackstop(ctx context.Context, v *provision.RecordView, due *atomic.Bool) { |
| 170 | if err := v.Rebuild(ctx); err != nil { |
| 171 | log.Printf("agent: record view backstop: %v", err) |
| 172 | } |
| 173 | due.Store(true) |
| 174 | } |
| 175 | |
| 176 | // backstopEvery is release.ChangeIndexBackstop jittered by +-10%. |
| 177 | func backstopEvery() time.Duration { |
| 178 | b := release.ChangeIndexBackstop |
| 179 | return b - b/10 + rand.N(b/5) |
| 180 | } |
| 181 | |
| 182 | // indexScopes is the scopes of obs whose generation differs from acted's, |
| 183 | // sorted, kept to the scopes a pass acts on (apps/<a>/<b> and accounts/<a>). |
| 184 | // The control/ scopes belong to other readers. |
| 185 | func indexScopes(acted, obs map[string]uint64) []string { |
| 186 | var moved []string |
| 187 | for scope, gen := range obs { |
| 188 | if have, ok := acted[scope]; ok && have == gen { |
| 189 | continue |
| 190 | } |
| 191 | if _, _, ok := provision.ScopeApp(scope); ok { |
| 192 | moved = append(moved, scope) |
| 193 | continue |
| 194 | } |
| 195 | if a, ok := strings.CutPrefix(scope, "accounts/"); ok && a != "" && !strings.Contains(a, "/") { |
| 196 | moved = append(moved, scope) |
| 197 | } |
| 198 | } |
| 199 | sort.Strings(moved) |
| 200 | return moved |
| 201 | } |
| 202 | |
| 203 | // movedRoutes keeps the routes whose (account, app) has an app scope in |
| 204 | // moved. Account scopes keep nothing here: their effect on a box (an |
| 205 | // account soft-deleted) is Reconcile's, which runs whole. |
| 206 | func movedRoutes(moved []string) func(registry.Route) bool { |
| 207 | apps := map[[2]string]bool{} |
| 208 | for _, scope := range moved { |
| 209 | if a, b, ok := provision.ScopeApp(scope); ok { |
| 210 | apps[[2]string{a, b}] = true |
| 211 | } |
| 212 | } |
| 213 | return func(rt registry.Route) bool { return apps[[2]string{release.RouteAccount(rt), rt.App}] } |
| 214 | } |
| 215 | |
| 216 | // passLegs is the serial-triggered pass set's inputs (fastConvergeLoop), |
| 217 | // shared by the full pass and the scoped one so the two run the same legs |
| 218 | // in the same order and cannot drift. |
| 219 | type passLegs struct { |
| 220 | store release.Store |
| 221 | reg *registry.Registry |
| 222 | pub ed25519.PublicKey |
| 223 | optDir string |
| 224 | restart func(string) error |
| 225 | opts []release.PassOption |
| 226 | priv *ecdh.PrivateKey // nil: no config leg (no CARLOS_SECRETS_KEY) |
| 227 | dd string |
| 228 | svc *activator.Service |
| 229 | writes hostwrite.Config |
| 230 | bs *buildStates |
| 231 | cfg func() provision.ReconcileConfig // read per pass, as provisionTickConfig always was |
| 232 | } |
| 233 | |
| 234 | // passAfterReconcileHook, when set, runs in every pass right after its |
| 235 | // reconcile leg. Tests only. |
| 236 | var passAfterReconcileHook func() |
| 237 | |
| 238 | // full is the serial-triggered pass set, in spec order: mint routes |
| 239 | // (reconcile), then reconcile aliases, then stage them (adopt), then deliver |
| 240 | // their config, over every hosted route. |
| 241 | func (l *passLegs) full(ctx context.Context) { l.run(ctx, "serial pass stages", nil) } |
| 242 | |
| 243 | // scoped is the pass an index-only change runs (spec §4.5, codex 10): the |
| 244 | // same legs, with build states, adopt and config sync restricted to the |
| 245 | // hosted routes whose app scope moved, and without the box-wide alias and |
| 246 | // plans legs, so changing one app re-reads that app only. Reconcile and the moves pass |
| 247 | // run whole, over the record view. Telemetry resolves every hosted app (its |
| 248 | // sink takes a whole map), but its reads are gated on their own scopes |
| 249 | // (telemetryReads), so only the moved apps' files are read. |
| 250 | func (l *passLegs) scoped(ctx context.Context, moved []string) { |
| 251 | l.run(ctx, "scoped pass stages", movedRoutes(moved)) |
| 252 | } |
| 253 | |
| 254 | // run is one pass set; keep nil is the unrestricted full pass. Each error is |
| 255 | // one log line, never fatal: the hourly floors stand underneath every leg. |
| 256 | func (l *passLegs) run(ctx context.Context, what string, keep func(registry.Route) bool) { |
| 257 | // Each stage's time, logged together when the set is slow, so the |
| 258 | // journal says WHICH stage held a promote back. |
| 259 | var took []string |
| 260 | t := time.Now() |
| 261 | mark := func(name string) { |
| 262 | took = append(took, fmt.Sprintf("%s %s", name, time.Since(t).Round(100*time.Millisecond))) |
| 263 | t = time.Now() |
| 264 | } |
| 265 | defer func(start time.Time) { logSlow(what, start, strings.Join(took, ", ")) }(time.Now()) |
| 266 | |
| 267 | if err := reconcileSerialized(ctx, l.store, l.reg, l.cfg()); err != nil { |
| 268 | log.Printf("agent: fast converge reconcile: %v", err) |
| 269 | } |
| 270 | mark("reconcile") |
| 271 | if hook := passAfterReconcileHook; hook != nil { |
| 272 | hook() |
| 273 | } |
| 274 | // The registry's build-state copies, refreshed at every reconcile |
| 275 | // (settings spec §4.2), before adopt reads them; refreshAll keeps |
| 276 | // to its own budget, and the stage is timed apart from reconcile. |
| 277 | // A scoped pass refreshes only the moved apps' hosts. |
| 278 | if l.bs != nil { |
| 279 | l.bs.refreshKept(ctx, keep) |
| 280 | mark("build states") |
| 281 | } |
| 282 | telemetryPass(ctx, l.store, l.reg) |
| 283 | mark("telemetry") |
| 284 | // The plans pass is box-wide (the marker, then every served account's |
| 285 | // limits record, published as one whole snapshot), so a scoped pass |
| 286 | // leaves it out: the full pass, the instances floor and the backstop's |
| 287 | // full pass run it. |
| 288 | if keep == nil { |
| 289 | plansPass(ctx, l.store, l.reg, os.Getenv) |
| 290 | mark("plans") |
| 291 | } |
| 292 | // The moves pass takes its own snapshot inside w.pass (spec §4.3): it |
| 293 | // is handed the plain store, never this pass's reconcile snapshot. |
| 294 | runMovesPass(ctx, l.store) |
| 295 | mark("moves") |
| 296 | runRunsPass(ctx) |
| 297 | mark("runs") |
| 298 | // A serial move may be a forwarded db read: list the queue now. The |
| 299 | // scoped pass nudges too: a request written through an indexed store |
| 300 | // may move only the index. The nudge is a non-blocking channel send. |
| 301 | nudgeDBReads() |
| 302 | // Aliases are box-wide (every domain claim, every edges file), so the |
| 303 | // scoped pass leaves them out: the serial-triggered full pass, the |
| 304 | // instances floor and the backstop's full pass run them. |
| 305 | if keep == nil { |
| 306 | // Folded once per pass and handed in, never re-read inside the |
| 307 | // loop: this leg runs on every serial bump (a seconds cadence), |
| 308 | // and the fold is a LoadBoxes plus a List plus one Get per edges |
| 309 | // file. On a fleet of under a dozen boxes that is a handful of |
| 310 | // Gets per bump, which is accepted here rather than cached — a |
| 311 | // cache would need its own staleness rule, and a STALE route set |
| 312 | // is the one thing a security guard must not have. Aliases read |
| 313 | // the domain claims, not the record families, so they read the |
| 314 | // bucket. |
| 315 | deploymentRoutes := edgesfile.DeploymentRoutes(ctx, l.store, time.Now()) |
| 316 | if err := domains.ReconcileAliases(ctx, l.store, l.reg, time.Now, deploymentRoutes, &aliasClaimCache); err != nil { |
| 317 | log.Printf("agent: fast converge alias pass: %v", err) |
| 318 | } |
| 319 | mark("aliases") |
| 320 | } |
| 321 | if err := release.PassFilter(ctx, l.reg, l.store, l.pub, l.optDir, l.restart, keep, l.opts...); err != nil { |
| 322 | log.Printf("agent: fast converge adopt pass: %v", err) |
| 323 | } |
| 324 | mark("adopt") |
| 325 | if l.priv != nil { |
| 326 | if err := syncExecSerializedFiltered(ctx, keep, l.reg, l.store, l.priv, l.dd, l.svc, l.writes); err != nil { |
| 327 | log.Printf("agent: fast converge config sync: %v", err) |
| 328 | } |
| 329 | mark("config") |
| 330 | } |
| 331 | } |
| 332 | |
amadan