| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "log" |
| 9 | "os" |
| 10 | "sort" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/activator" |
| 16 | "github.com/carlosframework/platform/internal/hostclaim" |
| 17 | "github.com/carlosframework/platform/internal/lease" |
| 18 | "github.com/carlosframework/platform/internal/moves" |
| 19 | "github.com/carlosframework/platform/internal/registry" |
| 20 | "github.com/carlosframework/platform/internal/release" |
| 21 | ) |
| 22 | |
| 23 | // buildStates is the agent's side of settings spec §4.2: the build-state |
| 24 | // object beside each instance's replica, under its lease, and its registry |
| 25 | // copy root reads. |
| 26 | // |
| 27 | // The object sits beside the lease's lock.json (ruling R-CLAIMPREFIX). With |
| 28 | // host claims off that is the hibernate bucket at HibernatePrefix; with them |
| 29 | // on, the claim actor's bucket at the claim holder's Location (never the |
| 30 | // session's g-<gen>/ prefix). A wake that holds the lease says which on its |
| 31 | // context (activator.WithBuildStateLocation), and that wins. |
| 32 | type buildStates struct { |
| 33 | hibBucket string // CARLOS_HIBERNATE_BUCKET |
| 34 | prefix func(host string) string // HibernatePrefix, host claims off |
| 35 | // claims reads a host's claim; nil means host claims are off. With them |
| 36 | // on, every exec instance wakes through its claim (doWake -> |
| 37 | // doClaimWake), so every host is located through it. |
| 38 | claims func(context.Context, string) (hostclaim.Observed, error) |
| 39 | claimBucket string // the claim actor's bucket, where claimed wakes lease |
| 40 | // staged is the legacy move round's staging prefix for host in the |
| 41 | // hibernate bucket (activator.MovesPrefix). nil unless host claims and |
| 42 | // Phase 1 are both off (ruling R-STAGEDREAD). |
| 43 | staged func(host string) string |
| 44 | storeFor func(ctx context.Context, bucket string) (release.Store, error) |
| 45 | reg *registry.Registry |
| 46 | optDir string |
| 47 | // bakeDir is the data dir every agent and root pass hands release.BakeDir: |
| 48 | // root's refusal records (release.ReadRefusal) live under it. |
| 49 | bakeDir string |
| 50 | now func() time.Time |
| 51 | // budget bounds one refreshAll; 0 means refreshBudget. |
| 52 | budget time.Duration |
| 53 | // beforePut, when set (tests only), runs in update between its read and |
| 54 | // its conditional Put. |
| 55 | beforePut func(ctx context.Context, store release.Store, key string) |
| 56 | |
| 57 | mu sync.Mutex |
| 58 | hosts map[string]*sync.Mutex // per-host: a read or update and its column write are one step |
| 59 | // resume is the first host the last refreshAll did not reach; the next |
| 60 | // one starts there (ruling R-REFRESH-ROTATE). Guarded by mu. |
| 61 | resume string |
| 62 | // written is each object write last put, by bucket and key, for the |
| 63 | // next write's conditional first try (writtenCopy). Guarded by mu. |
| 64 | written map[string]writtenCopy |
| 65 | } |
| 66 | |
| 67 | // newBuildStates is nil, nil on a box with no replica bucket: it runs no |
| 68 | // hibernating instances, so it has no build state to keep. cfg is the |
| 69 | // activator's, after configureWakeClaims: cfg.HostWrites.Enabled is the |
| 70 | // CARLOS_HOST_CLAIM switch and cfg.Claims its wake runtime. |
| 71 | func newBuildStates(cfg *activator.Config, reg *registry.Registry, optDir string) (*buildStates, error) { |
| 72 | if cfg == nil || cfg.Bucket == "" { |
| 73 | return nil, nil |
| 74 | } |
| 75 | regions, err := buildStateRegions(cfg) |
| 76 | if err != nil { |
| 77 | return nil, err |
| 78 | } |
| 79 | b := &buildStates{hibBucket: cfg.Bucket, prefix: func(h string) string { return activator.HibernatePrefix(cfg.Prefix, h) }, |
| 80 | reg: reg, optDir: optDir, bakeDir: dataDir(), now: time.Now} |
| 81 | if cfg.HostWrites.Enabled { |
| 82 | b.claims, b.claimBucket = cfg.Claims.Read, cfg.Claims.Bucket |
| 83 | } else if !moves.Enabled(os.Getenv) { |
| 84 | // The legacy move round's staging is promoted at the first wake's |
| 85 | // restore only with Phase 1 off (ExecBackend.Phase1, set from the |
| 86 | // same switch); with host claims on the state is at the claim |
| 87 | // holder's location, which a Phase 1 landing fills. |
| 88 | b.staged = activator.MovesPrefix |
| 89 | } |
| 90 | b.storeFor = cachedStores(regions, lease.ExecutionStore) |
| 91 | return b, nil |
| 92 | } |
| 93 | |
| 94 | // buildStateRegions is the one bucket the box leases in, with the region its |
| 95 | // leases sign for. With host claims on every wake is a claimed wake |
| 96 | // (doWake -> doClaimWake) and leases in the claim actor's bucket, so only |
| 97 | // that bucket is registered: the hibernate bucket is never located there, |
| 98 | // even when it is the same bucket under another region setting. |
| 99 | func buildStateRegions(cfg *activator.Config) (map[string]string, error) { |
| 100 | if !cfg.HostWrites.Enabled { |
| 101 | return map[string]string{cfg.Bucket: cfg.Region}, nil |
| 102 | } |
| 103 | if cfg.Claims == nil || cfg.Claims.Read == nil || cfg.Claims.Bucket == "" { |
| 104 | return nil, errors.New("host claims are on but the claim reader or the claim bucket is missing") |
| 105 | } |
| 106 | return map[string]string{cfg.Claims.Bucket: cfg.Claims.LeaseRegion}, nil |
| 107 | } |
| 108 | |
| 109 | // cachedStores opens one store per bucket the box leases in, with the |
| 110 | // lease's own S3 configuration (lease.ExecutionStore: the region the lock |
| 111 | // was signed for), and keeps it. Any other bucket is refused. |
| 112 | func cachedStores(regions map[string]string, open func(ctx context.Context, bucket, region string) (release.Store, error)) func(context.Context, string) (release.Store, error) { |
| 113 | var mu sync.Mutex |
| 114 | stores := map[string]release.Store{} |
| 115 | return func(ctx context.Context, bucket string) (release.Store, error) { |
| 116 | region, ok := regions[bucket] |
| 117 | if !ok { |
| 118 | return nil, fmt.Errorf("bucket %q is not one this box leases in", bucket) |
| 119 | } |
| 120 | mu.Lock() |
| 121 | defer mu.Unlock() |
| 122 | if s := stores[bucket]; s != nil { |
| 123 | return s, nil |
| 124 | } |
| 125 | s, err := open(ctx, bucket, region) |
| 126 | if err != nil { |
| 127 | return nil, err |
| 128 | } |
| 129 | stores[bucket] = s |
| 130 | return s, nil |
| 131 | } |
| 132 | } |
| 133 | |
| 134 | // locate is the store and prefix of host's build state: the context's |
| 135 | // location when a wake set one; otherwise, with host claims on, the claim |
| 136 | // holder's Location in the claim bucket (an unreadable claim is an error, |
| 137 | // never the hibernate prefix); otherwise the hibernate bucket and prefix. |
| 138 | func (b *buildStates) locate(ctx context.Context, host string) (release.Store, string, error) { |
| 139 | _, s, prefix, err := b.locateIn(ctx, host) |
| 140 | return s, prefix, err |
| 141 | } |
| 142 | |
| 143 | // locateIn is locate with the bucket it chose, which keys write's copy of |
| 144 | // the object it last wrote. |
| 145 | func (b *buildStates) locateIn(ctx context.Context, host string) (string, release.Store, string, error) { |
| 146 | bucket, prefix := b.hibBucket, "" |
| 147 | if l, ok := activator.BuildStateLocationFrom(ctx); ok { |
| 148 | bucket, prefix = l.Bucket, l.Prefix |
| 149 | } else if b.claims != nil { |
| 150 | obs, err := b.claims(ctx, host) |
| 151 | if err != nil { |
| 152 | return "", nil, "", fmt.Errorf("read claim: %w", err) |
| 153 | } |
| 154 | if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil || obs.Claim.Holder.Location == "" { |
| 155 | return "", nil, "", fmt.Errorf("claim of %s has no holder location", host) |
| 156 | } |
| 157 | bucket, prefix = b.claimBucket, obs.Claim.Holder.Location |
| 158 | } else { |
| 159 | prefix = b.prefix(host) |
| 160 | } |
| 161 | prefix = strings.TrimSuffix(prefix, "/") |
| 162 | s, err := b.storeFor(ctx, bucket) |
| 163 | if err != nil { |
| 164 | return "", nil, "", err |
| 165 | } |
| 166 | return bucket, s, prefix, nil |
| 167 | } |
| 168 | |
| 169 | func (b *buildStates) hostLock(host string) *sync.Mutex { |
| 170 | b.mu.Lock() |
| 171 | defer b.mu.Unlock() |
| 172 | if b.hosts == nil { |
| 173 | b.hosts = map[string]*sync.Mutex{} |
| 174 | } |
| 175 | m := b.hosts[host] |
| 176 | if m == nil { |
| 177 | m = &sync.Mutex{} |
| 178 | b.hosts[host] = m |
| 179 | } |
| 180 | return m |
| 181 | } |
| 182 | |
| 183 | // getState reads the object at key: nil and "" when there is none, an error |
| 184 | // when it does not parse or names another host. That error names the key: |
| 185 | // update never overwrites such an object (ruling R-BS-CORRUPT), so every |
| 186 | // spawn on the host is refused until an operator repairs or deletes it. |
| 187 | func getState(ctx context.Context, store release.Store, key, host string) (*release.BuildState, string, error) { |
| 188 | raw, etag, err := store.Get(ctx, key) |
| 189 | switch { |
| 190 | case errors.Is(err, os.ErrNotExist): |
| 191 | return nil, "", nil |
| 192 | case err != nil: |
| 193 | return nil, "", err |
| 194 | } |
| 195 | var s release.BuildState |
| 196 | if err := json.Unmarshal(raw, &s); err != nil || s.Host != host { |
| 197 | return nil, "", fmt.Errorf("build state %s for %s does not parse or names another host; repair or delete that object", key, host) |
| 198 | } |
| 199 | return &s, etag, nil |
| 200 | } |
| 201 | |
| 202 | // hasReplica is whether prefix holds any object besides the lease's |
| 203 | // lock.json and build-state.json. Three keys decide it, so a store with a |
| 204 | // bounded listing (release.PageLister) is asked for three; any other lists |
| 205 | // the prefix whole. |
| 206 | func hasReplica(ctx context.Context, store release.Store, prefix string) (bool, error) { |
| 207 | var keys []string |
| 208 | var err error |
| 209 | if pl, ok := store.(release.PageLister); ok { |
| 210 | keys, err = pl.ListPage(ctx, prefix+"/", "", 3) |
| 211 | } else { |
| 212 | keys, err = store.List(ctx, prefix+"/") |
| 213 | } |
| 214 | if err != nil { |
| 215 | return false, err |
| 216 | } |
| 217 | for _, k := range keys { |
| 218 | rel := strings.TrimPrefix(k, prefix+"/") |
| 219 | if rel != "lock.json" && rel != "build-state.json" { |
| 220 | return true, nil |
| 221 | } |
| 222 | } |
| 223 | return false, nil |
| 224 | } |
| 225 | |
| 226 | // read returns host's build state (nil when there is none) and whether its |
| 227 | // replica has any object besides the lease's lock.json and build-state.json. |
| 228 | // |
| 229 | // A legacy move's staging, when there is one, is read in place of the own |
| 230 | // prefix (settings spec rev 11 §4.3a): between the cut and the first wake |
| 231 | // the replica and the build state are only there, and the wake's restore |
| 232 | // promotes the staging over the own prefix whenever it holds anything |
| 233 | // (promoteStagedReplica: staging wins, and its mirror deletes what the |
| 234 | // staging lacks). So the staging is what the instance's state will be. A |
| 235 | // held instance must not read as new in that window, or root installs the |
| 236 | // head. update still writes only the own prefix: the wake promotes first. |
| 237 | // |
| 238 | // When both hold a state, an own state with a strictly later At wins (a |
| 239 | // tie is the staging's, as at restore): a wake that promoted the |
| 240 | // staging and then failed to drop it (dropStagedReplica is best-effort) |
| 241 | // leaves a pre-move staged state behind its own newer record, and that |
| 242 | // record is the instance's state. |
| 243 | func (b *buildStates) read(ctx context.Context, host string) (*release.BuildState, bool, error) { |
| 244 | staged, stagedReplica, ok, err := b.readStaged(ctx, host) |
| 245 | if err != nil { |
| 246 | return nil, false, err |
| 247 | } |
| 248 | if ok && staged == nil { |
| 249 | return nil, stagedReplica, nil |
| 250 | } |
| 251 | st, replica, err := b.readOwn(ctx, host) |
| 252 | if err != nil { |
| 253 | return nil, false, err |
| 254 | } |
| 255 | if ok && (st == nil || !st.At.After(staged.At)) { |
| 256 | return staged, stagedReplica || replica, nil |
| 257 | } |
| 258 | return st, replica || (ok && stagedReplica), nil |
| 259 | } |
| 260 | |
| 261 | // readOwn reads host's build state and replica at its own location. |
| 262 | func (b *buildStates) readOwn(ctx context.Context, host string) (*release.BuildState, bool, error) { |
| 263 | store, prefix, err := b.locate(ctx, host) |
| 264 | if err != nil { |
| 265 | return nil, false, err |
| 266 | } |
| 267 | st, _, err := getState(ctx, store, release.BuildStateKey(prefix), host) |
| 268 | if err != nil { |
| 269 | return nil, false, err |
| 270 | } |
| 271 | replica, err := hasReplica(ctx, store, prefix) |
| 272 | if err != nil { |
| 273 | return nil, false, err |
| 274 | } |
| 275 | return st, replica, nil |
| 276 | } |
| 277 | |
| 278 | // readStaged reads a legacy move's staging for host (ruling R-STAGEDREAD): |
| 279 | // ok is true when moves/<host>/ in the hibernate bucket holds any key of |
| 280 | // the legacy round's, the same "anything staged" promoteStagedReplica |
| 281 | // promotes on. Phase 1 op lanes (moves/<host>/<mv_ or rb_ id>/...) are not |
| 282 | // the legacy round's and are skipped. The staged build-state.json is the |
| 283 | // state, and any other staged object but lock.json is a replica. Off (ok |
| 284 | // false) unless staged is set: host claims and Phase 1 both off. |
| 285 | func (b *buildStates) readStaged(ctx context.Context, host string) (*release.BuildState, bool, bool, error) { |
| 286 | if b.staged == nil { |
| 287 | return nil, false, false, nil |
| 288 | } |
| 289 | prefix := strings.TrimSuffix(b.staged(host), "/") |
| 290 | store, err := b.storeFor(ctx, b.hibBucket) |
| 291 | if err != nil { |
| 292 | return nil, false, false, err |
| 293 | } |
| 294 | keys, err := store.List(ctx, prefix+"/") |
| 295 | if err != nil { |
| 296 | return nil, false, false, fmt.Errorf("list the staged move %s: %w", prefix, err) |
| 297 | } |
| 298 | var staged, state, replica bool |
| 299 | for _, k := range keys { |
| 300 | rel := strings.TrimPrefix(k, prefix+"/") |
| 301 | if first, _, ok := strings.Cut(rel, "/"); ok && moves.ValidOpID(first) { |
| 302 | continue |
| 303 | } |
| 304 | staged = true |
| 305 | switch rel { |
| 306 | case "build-state.json": |
| 307 | state = true |
| 308 | case "lock.json": |
| 309 | default: |
| 310 | replica = true |
| 311 | } |
| 312 | } |
| 313 | if !staged { |
| 314 | return nil, false, false, nil |
| 315 | } |
| 316 | var st *release.BuildState |
| 317 | if state { |
| 318 | if st, _, err = getState(ctx, store, release.BuildStateKey(prefix), host); err != nil { |
| 319 | return nil, false, false, err |
| 320 | } |
| 321 | } |
| 322 | return st, replica, true, nil |
| 323 | } |
| 324 | |
| 325 | // refresh stores what read found. A failed read keeps the last known state |
| 326 | // for root (a known copy is not wiped by a blip; an unread one stays |
| 327 | // unknown) and records the failure in ReadErr, so the box's report says this |
| 328 | // host's build state is incomplete and the housekeeper prunes nothing for its |
| 329 | // app (Astra I1 #6). A good read clears ReadErr. |
| 330 | func (b *buildStates) refresh(ctx context.Context, host string) error { |
| 331 | m := b.hostLock(host) |
| 332 | m.Lock() |
| 333 | defer m.Unlock() |
| 334 | st, replica, err := b.read(ctx, host) |
| 335 | if err != nil { |
| 336 | rt, gerr := b.reg.Get(host) |
| 337 | prev := release.BuildStateCopy{} |
| 338 | if gerr == nil && rt != nil { |
| 339 | prev, _ = release.ParseBuildStateCopy(rt.BuildState) |
| 340 | } |
| 341 | prev.ReadErr = err.Error() |
| 342 | if serr := b.reg.SetBuildState(host, prev.Column()); serr != nil { |
| 343 | return errors.Join(err, serr) |
| 344 | } |
| 345 | return err |
| 346 | } |
| 347 | c := release.BuildStateCopy{Known: true, Replica: replica, State: st, ReadAt: b.now().UTC().Truncate(time.Second)} |
| 348 | return b.reg.SetBuildState(host, c.Column()) |
| 349 | } |
| 350 | |
| 351 | // keepsBuildState is a row whose spawns run under a lease in the replica |
| 352 | // bucket, so the store keeps a build state beside it (ruling R-WORKER-ROWS): |
| 353 | // an activator-managed exec instance (activator.ActivatorManaged: Kind |
| 354 | // instance, Hibernate, a socket, no Backing, no Tunnel), or a standalone |
| 355 | // worker (Kind "sidecar"; every worker wake takes the lease at |
| 356 | // HibernatePrefix(its host), activator.go doWake step 1, whatever its |
| 357 | // Hibernate flag), exec-backed and not tunnel-served. Never a named unit or |
| 358 | // the platform's own row. Settings apply only to Kind instance; that is |
| 359 | // root's concern, not this store's. |
| 360 | // |
| 361 | // With host claims on (claims) a worker never leases: the claim runtime |
| 362 | // refuses sidecar rows before any lease (claimwakeassembly.go |
| 363 | // Prerequisites), so it has no build state and no claim to locate one by. |
| 364 | func keepsBuildState(rt registry.Route, claims bool) bool { |
| 365 | if rt.Unit != "" || rt.Platform { |
| 366 | return false |
| 367 | } |
| 368 | if rt.IsSidecar() { |
| 369 | return !claims && rt.Backing == "" && rt.Tunnel == "" |
| 370 | } |
| 371 | return activator.ActivatorManaged(rt) |
| 372 | } |
| 373 | |
| 374 | // refreshBudget bounds one refreshAll: it runs inline in the reconcile legs, |
| 375 | // before the adopt pass. A host it does not reach keeps its last copy; a |
| 376 | // host whose read the deadline cuts records the failure like any other. |
| 377 | const refreshBudget = 20 * time.Second |
| 378 | |
| 379 | // refreshAll runs after every reconcile, for every row keepsBuildState |
| 380 | // names, within the budget. Each host is located on its own: a location a |
| 381 | // caller's context carries names one wake's lease, so it is masked here. It |
| 382 | // returns how many hosts it did not reach. |
| 383 | // |
| 384 | // The hosts are taken in host order starting at the first one the last |
| 385 | // call did not reach, wrapping round (ruling R-REFRESH-ROTATE): a box whose |
| 386 | // budget runs out every pass reaches every host in turn, rather than |
| 387 | // leaving the same tail unread (and its app unprunable) forever. |
| 388 | func (b *buildStates) refreshAll(ctx context.Context) int { return b.refreshKept(ctx, nil) } |
| 389 | |
| 390 | // refreshKept is refreshAll over the rows keep accepts (nil: every row): the |
| 391 | // scoped pass refreshes only the hosts of the apps that moved (S3 |
| 392 | // change-index design §4.5). A filtered call neither reads nor moves the |
| 393 | // rotation point, which stays the full refresh's. |
| 394 | func (b *buildStates) refreshKept(ctx context.Context, keep func(registry.Route) bool) int { |
| 395 | budget := b.budget |
| 396 | if budget <= 0 { |
| 397 | budget = refreshBudget |
| 398 | } |
| 399 | ctx, cancel := context.WithTimeout(activator.WithBuildStateLocation(ctx, activator.BuildStateLocation{}), budget) |
| 400 | defer cancel() |
| 401 | routes, err := b.reg.List() |
| 402 | if err != nil { |
| 403 | log.Printf("agent: build state refresh: %v", err) |
| 404 | return 0 |
| 405 | } |
| 406 | var hosts []string |
| 407 | for _, rt := range routes { |
| 408 | if keepsBuildState(rt, b.claims != nil) && (keep == nil || keep(rt)) { |
| 409 | hosts = append(hosts, rt.Host) |
| 410 | } |
| 411 | } |
| 412 | sort.Strings(hosts) |
| 413 | start := 0 |
| 414 | if keep == nil { |
| 415 | b.mu.Lock() |
| 416 | start = sort.SearchStrings(hosts, b.resume) |
| 417 | b.mu.Unlock() |
| 418 | } |
| 419 | if start >= len(hosts) { |
| 420 | start = 0 |
| 421 | } |
| 422 | hosts = append(hosts[start:len(hosts):len(hosts)], hosts[:start]...) |
| 423 | for i, host := range hosts { |
| 424 | if ctx.Err() != nil { |
| 425 | if keep == nil { |
| 426 | b.mu.Lock() |
| 427 | b.resume = host |
| 428 | b.mu.Unlock() |
| 429 | } |
| 430 | skipped := len(hosts) - i |
| 431 | log.Printf("agent: build state refresh: the %s budget ran out; %d of %d hosts not reached keep their last copy", budget, skipped, len(hosts)) |
| 432 | return skipped |
| 433 | } |
| 434 | if err := b.refresh(ctx, host); err != nil { |
| 435 | log.Printf("agent: build state of %s: %v", host, err) |
| 436 | } |
| 437 | } |
| 438 | return 0 |
| 439 | } |
| 440 | |
| 441 | // updateTries bounds update's compare-and-swap loop. |
| 442 | const updateTries = 3 |
| 443 | |
| 444 | // update is the only writer of the object (ruling R-BS-UPDATE). Callers hold |
| 445 | // the host's lease. Under the host's lock it reads the object (absent is a |
| 446 | // nil cur and a create-only Put), merges, and Puts against the etag it read, |
| 447 | // so a write that landed in between is never overwritten: a CAS conflict |
| 448 | // re-reads and merges again, up to updateTries times. The registry copy is |
| 449 | // set under the same lock, so a concurrent refresh cannot put older state |
| 450 | // over it. A failed update writes nothing the caller may run on. |
| 451 | func (b *buildStates) update(ctx context.Context, host string, merge func(cur *release.BuildState) release.BuildState) (release.BuildState, error) { |
| 452 | return b.write(ctx, host, func(cur *release.BuildState) (release.BuildState, bool, error) { return merge(cur), true, nil }, nil) |
| 453 | } |
| 454 | |
| 455 | // write is update with a merge that may refuse (an error: nothing is |
| 456 | // written) or find nothing to write (false: the object and the registry copy |
| 457 | // are left as they are, and the state as read is returned). replica says |
| 458 | // whether the instance has a replica, for the registry copy; nil is yes (a |
| 459 | // process is about to run against the database). |
| 460 | func (b *buildStates) write(ctx context.Context, host string, merge func(cur *release.BuildState) (release.BuildState, bool, error), |
| 461 | replica func(ctx context.Context, store release.Store, prefix string) (bool, error)) (release.BuildState, error) { |
| 462 | m := b.hostLock(host) |
| 463 | m.Lock() |
| 464 | defer m.Unlock() |
| 465 | bucket, store, prefix, err := b.locateIn(ctx, host) |
| 466 | if err != nil { |
| 467 | return release.BuildState{}, err |
| 468 | } |
| 469 | key := release.BuildStateKey(prefix) |
| 470 | // The first try goes on the copy this agent last wrote, when it has |
| 471 | // one: a wake's record is then one conditional write, not a read and a |
| 472 | // write. The condition is that copy's etag, so the write lands only on |
| 473 | // the very object the merge saw, and the merge decides exactly what it |
| 474 | // would on a fresh read. Anything else (the object moved on, the merge |
| 475 | // refused or had nothing to write on a copy that may be stale) is |
| 476 | // decided again on a fresh read, as before. |
| 477 | cached, hasCopy := b.lastWritten(bucket, key) |
| 478 | tries := updateTries |
| 479 | if hasCopy { |
| 480 | tries++ // the copy's try is extra: the fresh reads keep all theirs |
| 481 | } |
| 482 | for try := 0; try < tries; try++ { |
| 483 | var ( |
| 484 | cur *release.BuildState |
| 485 | etag string |
| 486 | ) |
| 487 | fromCopy := hasCopy && try == 0 |
| 488 | if fromCopy { |
| 489 | cur, etag = cached.state(), cached.etag |
| 490 | } else if cur, etag, err = getState(ctx, store, key, host); err != nil { |
| 491 | return release.BuildState{}, err |
| 492 | } |
| 493 | next, put, err := merge(cur) |
| 494 | if fromCopy && (err != nil || !put) { |
| 495 | continue |
| 496 | } |
| 497 | if err != nil { |
| 498 | return release.BuildState{}, err |
| 499 | } |
| 500 | if !put { |
| 501 | if cur == nil { |
| 502 | return release.BuildState{}, nil |
| 503 | } |
| 504 | return *cur, nil |
| 505 | } |
| 506 | next.Schema, next.Host = 1, host |
| 507 | next.At = b.now().UTC().Truncate(time.Second) |
| 508 | raw, err := json.MarshalIndent(next, "", " ") |
| 509 | if err != nil { |
| 510 | return release.BuildState{}, err |
| 511 | } |
| 512 | if b.beforePut != nil { |
| 513 | b.beforePut(ctx, store, key) |
| 514 | } |
| 515 | newEtag, err := store.Put(ctx, key, raw, etag) |
| 516 | if errors.Is(err, release.ErrCASConflict) { |
| 517 | b.forget(bucket, key) |
| 518 | continue |
| 519 | } else if err != nil { |
| 520 | b.forget(bucket, key) |
| 521 | return release.BuildState{}, err |
| 522 | } |
| 523 | b.remember(bucket, key, raw, newEtag) |
| 524 | // A process about to run against the database has (or is about to |
| 525 | // have) a replica; any other writer says. |
| 526 | hasRep := true |
| 527 | if replica != nil { |
| 528 | if hasRep, err = replica(ctx, store, prefix); err != nil { |
| 529 | return next, fmt.Errorf("build state of %s written, its replica not listed: %w", host, err) |
| 530 | } |
| 531 | } |
| 532 | c := release.BuildStateCopy{Known: true, Replica: hasRep, State: &next, ReadAt: b.now().UTC().Truncate(time.Second)} |
| 533 | if err := b.reg.SetBuildState(host, c.Column()); err != nil { |
| 534 | return next, fmt.Errorf("build state of %s written, registry copy not: %w", host, err) |
| 535 | } |
| 536 | return next, nil |
| 537 | } |
| 538 | return release.BuildState{}, fmt.Errorf("build state of %s: %d tries lost the race: %w", host, updateTries, release.ErrCASConflict) |
| 539 | } |
| 540 | |
| 541 | // writtenCopy is the object write last put at a key, and the etag that |
| 542 | // write returned: the state the next write's conditional first try goes on. |
| 543 | type writtenCopy struct { |
| 544 | raw []byte |
| 545 | etag string |
| 546 | } |
| 547 | |
| 548 | // state decodes a fresh copy each time, so a merge can never change the |
| 549 | // cached bytes. |
| 550 | func (c writtenCopy) state() *release.BuildState { |
| 551 | var s release.BuildState |
| 552 | if json.Unmarshal(c.raw, &s) != nil { |
| 553 | return nil |
| 554 | } |
| 555 | return &s |
| 556 | } |
| 557 | |
| 558 | func (b *buildStates) lastWritten(bucket, key string) (writtenCopy, bool) { |
| 559 | b.mu.Lock() |
| 560 | defer b.mu.Unlock() |
| 561 | c, ok := b.written[bucket+"\x00"+key] |
| 562 | return c, ok && c.etag != "" && c.state() != nil |
| 563 | } |
| 564 | |
| 565 | func (b *buildStates) remember(bucket, key string, raw []byte, etag string) { |
| 566 | b.mu.Lock() |
| 567 | defer b.mu.Unlock() |
| 568 | if etag == "" { |
| 569 | delete(b.written, bucket+"\x00"+key) |
| 570 | return |
| 571 | } |
| 572 | if b.written == nil { |
| 573 | b.written = map[string]writtenCopy{} |
| 574 | } |
| 575 | b.written[bucket+"\x00"+key] = writtenCopy{raw: append([]byte(nil), raw...), etag: etag} |
| 576 | } |
| 577 | |
| 578 | func (b *buildStates) forget(bucket, key string) { |
| 579 | b.mu.Lock() |
| 580 | defer b.mu.Unlock() |
| 581 | delete(b.written, bucket+"\x00"+key) |
| 582 | } |
| 583 | |
| 584 | // beforeExec is Service.BeforeExec (settings spec §4.2): before build runs |
| 585 | // against host's database, under its lease, the object names build as |
| 586 | // last_executed, carries root's resolution from the consumed file, and |
| 587 | // completes the request if build is its target (release.NextBuildState). |
| 588 | // That write is what completes a request: at first run, not at install. |
| 589 | // |
| 590 | // It fails closed. A consumed file the agent cannot read refuses the exec: |
| 591 | // a write without it could miss a resolution and leave R uncompleted after |
| 592 | // its T ran. update sets the registry copy itself, and refuses an object it |
| 593 | // cannot parse rather than overwrite it (ruling R-BS-CORRUPT). For a row |
| 594 | // settings apply to, a state that now names another held target than build |
| 595 | // (release.HeldExecStale) refuses it too: the wake decided on a stale copy. |
| 596 | // The one exception is the zero-downtime planner's checked selection of |
| 597 | // build in place of that target, which the activator carries on ctx for this |
| 598 | // exec only (activator.PlannedSelection; ruling R-PLANNED-RECOVERY): it is |
| 599 | // admitted, and still recorded. A consumed file another tenant of the |
| 600 | // hostname left (Consumed.ForeignTo) is read as none (ruling R26). |
| 601 | func (b *buildStates) beforeExec(ctx context.Context, host, build string) error { |
| 602 | if build == "" { |
| 603 | return fmt.Errorf("no build to record for %s", host) |
| 604 | } |
| 605 | consumed, err := release.ReadConsumed(b.optDir, host) |
| 606 | if err != nil { |
| 607 | return fmt.Errorf("read root's consumed file for %s: %w", host, err) |
| 608 | } |
| 609 | // The held target is revalidated against the state read under the lease |
| 610 | // (Astra round 1 finding 1): a wake that decided on a stale copy is |
| 611 | // refused here, before anything is written or run. A route that does |
| 612 | // not read refuses the exec before any write (Astra r4 finding 3): the |
| 613 | // tenant check below needs it, and an unread tenant must not merge a |
| 614 | // hostname-keyed file into the state. A route that is absent (nil, no |
| 615 | // error) is read as before. |
| 616 | rt, err := b.reg.Get(host) |
| 617 | if err != nil { |
| 618 | return fmt.Errorf("read the route of %s: %w", host, err) |
| 619 | } |
| 620 | // A file root wrote for another tenant of this hostname carries nothing |
| 621 | // into this tenant's state (ruling R26). An untagged one is read as the |
| 622 | // route's own, as before. |
| 623 | if rt != nil && consumed.ForeignTo(*rt) { |
| 624 | consumed = nil |
| 625 | } |
| 626 | held := rt != nil && settingsApply(*rt) |
| 627 | inPlaceOf := "" |
| 628 | if sel, ok := activator.PlannedSelectionFrom(ctx); ok && sel.Build == build { |
| 629 | inPlaceOf = sel.InPlaceOf |
| 630 | } |
| 631 | _, err = b.write(ctx, host, func(cur *release.BuildState) (release.BuildState, bool, error) { |
| 632 | if held { |
| 633 | if err := release.HeldExecStale(release.DecideHeldWake(b.heldWakeInput(*rt, cur)), cur, build, inPlaceOf); err != nil { |
| 634 | return release.BuildState{}, false, err |
| 635 | } |
| 636 | } |
| 637 | return release.NextBuildState(cur, consumed, build), true, nil |
| 638 | }, nil) |
| 639 | return err |
| 640 | } |
| 641 | |
| 642 | // carryRequest is moveCfg.CarryRequest (ruling R-SPEC-43A-2; Astra round 1 |
| 643 | // finding 3): before a cut seals host's build state at bucket/prefix (the |
| 644 | // lease's location: the hibernate prefix for a legacy cut, the claim |
| 645 | // holder's location for a Phase 1 cut or a return), root's consumed record |
| 646 | // on this box for the route's tenant (not one Consumed.ForeignTo it, ruling |
| 647 | // R26) is merged into it with release.CarryRequest: a resolution and |
| 648 | // a failed completion, never a successful completion, never last_executed. |
| 649 | // The object is written only when that changes its request state, under the |
| 650 | // host's lock and against the etag it read (write), so it never overwrites a |
| 651 | // newer record. A consumed file that does not read refuses the carry, and so |
| 652 | // the cut: a request this box closed would reopen on the destination. |
| 653 | // |
| 654 | // The carry runs inside the carry fence (release.FenceCarry; ruling |
| 655 | // R-CARRY-FENCE): root's publication for host and this carry serialize on |
| 656 | // one lock, and root publishes only for a row no move holds. So the row's |
| 657 | // move hold must already be effective here (every lane sets it before its |
| 658 | // carry: the legacy cut's "moving", the Phase 1 cut's held "moving", the |
| 659 | // return lane's "releasing"); a row in no hold refuses the carry, as root |
| 660 | // could still publish after it. A row gone from the registry gets no |
| 661 | // publication either. Then the fence, held across the read and the write: a |
| 662 | // root record in flight lands first and is carried. |
| 663 | func (b *buildStates) carryRequest(ctx context.Context, host, bucket, prefix string) error { |
| 664 | rt, err := b.reg.Get(host) |
| 665 | if err != nil { |
| 666 | return fmt.Errorf("read the route of %s: %w", host, err) |
| 667 | } |
| 668 | if rt != nil && !registry.IsHold(rt.Status) { |
| 669 | return fmt.Errorf("%s is in no move hold (status %q), so root could still publish a request record after the carry", host, rt.Status) |
| 670 | } |
| 671 | unlock, err := release.FenceCarry(ctx, b.optDir, host) |
| 672 | if err != nil { |
| 673 | return fmt.Errorf("the carry fence for %s: %w", host, err) |
| 674 | } |
| 675 | defer unlock() |
| 676 | consumed, err := release.ReadConsumed(b.optDir, host) |
| 677 | if err != nil { |
| 678 | return fmt.Errorf("read root's consumed file for %s: %w", host, err) |
| 679 | } |
| 680 | // Another tenant's file is no request record of this route's (ruling |
| 681 | // R26): nothing to carry. An untagged one is carried as before. |
| 682 | if consumed == nil || (rt != nil && consumed.ForeignTo(*rt)) { |
| 683 | return nil |
| 684 | } |
| 685 | ctx = activator.WithBuildStateLocation(ctx, activator.BuildStateLocation{Bucket: bucket, Prefix: prefix}) |
| 686 | _, err = b.write(ctx, host, func(cur *release.BuildState) (release.BuildState, bool, error) { |
| 687 | next := release.CarryRequest(cur, consumed) |
| 688 | return next, !release.SameRequestState(cur, &next), nil |
| 689 | }, hasReplica) |
| 690 | return err |
| 691 | } |
| 692 | |
| 693 | // moveCarry is moveCfg.CarryRequest for this box: the build-state store's |
| 694 | // carry, a refusal when the store did not start (a cut must not seal a |
| 695 | // state it could not bring up to date), and nil on a box with no replica |
| 696 | // bucket (no build state to carry). |
| 697 | func moveCarry(bs *buildStates, initErr error) func(ctx context.Context, host, bucket, prefix string) error { |
| 698 | if initErr != nil { |
| 699 | return func(context.Context, string, string, string) error { |
| 700 | return fmt.Errorf("the build state store did not start on this box (%v)", initErr) |
| 701 | } |
| 702 | } |
| 703 | if bs == nil { |
| 704 | return nil |
| 705 | } |
| 706 | return bs.carryRequest |
| 707 | } |
| 708 | |
| 709 | // buildStateHook is Service.BeforeExec for this box (settings spec §4.2, |
| 710 | // Astra I3 #1). Only a box with no replica bucket configured (bs nil, no |
| 711 | // error) keeps the nil hook: it runs no hibernating instance, so it has no |
| 712 | // build state. A box whose bucket is configured but whose build-state store |
| 713 | // failed to start refuses every exec: the write before execution is |
| 714 | // mandatory, and "could not record" never means "run unrecorded". |
| 715 | func buildStateHook(bs *buildStates, initErr error) func(ctx context.Context, host, build string) error { |
| 716 | if initErr != nil { |
| 717 | return func(context.Context, string, string) error { |
| 718 | return fmt.Errorf("the build state store did not start on this box (%v); restart the agent once it is fixed", initErr) |
| 719 | } |
| 720 | } |
| 721 | if bs == nil { |
| 722 | return nil |
| 723 | } |
| 724 | return bs.beforeExec |
| 725 | } |
| 726 | |
| 727 | // heldTarget is the held wake check's answer (activator.Service.HeldTarget; |
| 728 | // settings spec §4.3 rule 3): release.DecideHeldWake (ruling R-HELD-PURE) |
| 729 | // on what root reads and what root wrote: the route's settings column, its |
| 730 | // build-state copy, root's consumed file and refusal record, and the build |
| 731 | // the live link names. The decision is root's rule order (Target): a pin, |
| 732 | // then a pending request's T, then rule 3's target (last_executed, else the |
| 733 | // installed build, each failure-checked). Root's consumed reason counts only |
| 734 | // for the settings column it was decided for: with none, a wake that could |
| 735 | // be held holds and asks root, never "not held" (Astra round 1 finding 4). |
| 736 | // A copy whose last refresh failed is refused as unread (finding 1). |
| 737 | // |
| 738 | // A held, decided target that root does not report failed is offered with |
| 739 | // this box's verified retained copy of it, if any (Staged). |
| 740 | // |
| 741 | // Settings hold only the rows root applies them to (settingsApply; ruling |
| 742 | // R-HELD-SCOPE): every other row, a named unit or an exec worker under an |
| 743 | // app-policy setting, is not held. |
| 744 | func (b *buildStates) heldTarget(host string) activator.HeldWake { |
| 745 | rt, err := b.reg.Get(host) |
| 746 | if err != nil || rt == nil || !settingsApply(*rt) { |
| 747 | return activator.HeldWake{} |
| 748 | } |
| 749 | d := release.DecideHeldWake(b.heldWakeInput(*rt, nil)) |
| 750 | h := activator.HeldWake{Target: d.Target, Held: d.Held, Unknown: d.Unknown, Failed: d.Failed, AskRoot: d.AskRoot, Unread: d.Unread} |
| 751 | if h.Held && h.Target != "" && !h.Failed && !h.AskRoot && !h.Unknown && !h.Unread { |
| 752 | h.Staged = b.retained(*rt, h.Target) |
| 753 | } |
| 754 | return h |
| 755 | } |
| 756 | |
| 757 | // heldWakeInput reads DecideHeldWake's input for rt. state, when set, is |
| 758 | // the build state just read under the lease (beforeExec's revalidation), in |
| 759 | // place of the registry copy. A consumed file or refusal record that does |
| 760 | // not read is no word from root (nil), and so is a consumed file tagged for |
| 761 | // another tenant (Consumed.ForeignTo); a copy that does not parse is unread. |
| 762 | func (b *buildStates) heldWakeInput(rt registry.Route, state *release.BuildState) release.HeldWakeInput { |
| 763 | in := release.HeldWakeInput{Settings: rt.UpdateSettings} |
| 764 | // Another tenant's file is no word from root for this route (ruling |
| 765 | // R26), so a wake that could be held holds and asks root. An untagged |
| 766 | // file, from before ruling R22, is read as before, as every other read |
| 767 | // of it is (ruling R28): ignoring its failed decision would let a wake |
| 768 | // whose target is already live run a build root reports failed, as |
| 769 | // awaitRootDecision does not wait when the live link names the target. |
| 770 | if c, _ := release.ReadConsumed(b.optDir, rt.Host); !c.ForeignTo(rt) { |
| 771 | in.Consumed = c |
| 772 | } |
| 773 | in.Refusal, _ = release.ReadRefusal(b.bakeDir, rt.Host) |
| 774 | in.Installed, _ = release.LiveVersion(b.optDir, rt.Host) |
| 775 | if state != nil { |
| 776 | in.Build = release.BuildStateCopy{Known: true, Replica: true, State: state} |
| 777 | } else { |
| 778 | in.Build, _ = release.ParseBuildStateCopy(rt.BuildState) |
| 779 | } |
| 780 | return in |
| 781 | } |
| 782 | |
| 783 | // settingsApply reports a row root applies update settings to: Kind |
| 784 | // instance with no Unit, release's adopt condition (ruling R-HELD-SCOPE). |
| 785 | // Root and the wake must agree on it, or a wake waits for an install root |
| 786 | // never makes. |
| 787 | func settingsApply(rt registry.Route) bool { return rt.Kind == "instance" && rt.Unit == "" } |
| 788 | |
| 789 | // retained is v's staged executable on this box when its bytes match its |
| 790 | // staged manifest's sha256 for this platform (release.RetainedStaged, the |
| 791 | // check stageAuthorized's recovery uses), else nil (Astra I3 #5). |
| 792 | func (b *buildStates) retained(rt registry.Route, v string) *release.StagedIdentity { |
| 793 | id, _ := release.RetainedStaged(b.optDir, release.RouteAccount(rt), rt.App, v) |
| 794 | return id |
| 795 | } |
| 796 | |
| 797 | // wakeHeldBudget bounds wakeTarget's refresh: the wake waits on it. |
| 798 | const wakeHeldBudget = 5 * time.Second |
| 799 | |
| 800 | // wakeTarget is heldTarget with the copy refreshed first (§4.2: "refreshed |
| 801 | // at every reconcile and before every wake"), for a route whose settings |
| 802 | // could hold it. A route with default settings, or one root applies no |
| 803 | // settings to (settingsApply), is today's wake: nothing is read for it. A |
| 804 | // refresh that fails refuses the wake as unread (Astra round 1 finding 1): |
| 805 | // the last copy may name a build the data has since moved past. |
| 806 | func (b *buildStates) wakeTarget(ctx context.Context, host string) activator.HeldWake { |
| 807 | rt, err := b.reg.Get(host) |
| 808 | if err != nil || rt == nil || !settingsApply(*rt) { |
| 809 | return activator.HeldWake{} |
| 810 | } |
| 811 | if u, err := release.ParseUpdateSettingsLenient(rt.UpdateSettings); err == nil && u.Default() { |
| 812 | return activator.HeldWake{} |
| 813 | } |
| 814 | if keepsBuildState(*rt, b.claims != nil) { |
| 815 | ctx, cancel := context.WithTimeout(ctx, wakeHeldBudget) |
| 816 | err := b.refresh(ctx, host) |
| 817 | cancel() |
| 818 | if err != nil { |
| 819 | // Refused, never decided from the last copy (Astra round 1 |
| 820 | // finding 1): the durable state may name another build by now. |
| 821 | // refresh recorded the failure in the copy, which DecideHeldWake |
| 822 | // refuses too, unless the registry write itself failed. |
| 823 | log.Printf("agent: %s not woken: its build state could not be read before the wake: %v", host, err) |
| 824 | return activator.HeldWake{Held: true, Unread: true} |
| 825 | } |
| 826 | } |
| 827 | return b.heldTarget(host) |
| 828 | } |
| 829 | |
amadan