| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/ecdh" |
| 6 | "crypto/ed25519" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "log" |
| 11 | "os" |
| 12 | "path/filepath" |
| 13 | "slices" |
| 14 | "strconv" |
| 15 | "strings" |
| 16 | "sync" |
| 17 | "sync/atomic" |
| 18 | "time" |
| 19 | |
| 20 | "github.com/carlosframework/platform/internal/activator" |
| 21 | "github.com/carlosframework/platform/internal/appconfig" |
| 22 | "github.com/carlosframework/platform/internal/cgroup" |
| 23 | "github.com/carlosframework/platform/internal/controlsock" |
| 24 | "github.com/carlosframework/platform/internal/dbverify" |
| 25 | "github.com/carlosframework/platform/internal/hostbox" |
| 26 | "github.com/carlosframework/platform/internal/hostclaim" |
| 27 | "github.com/carlosframework/platform/internal/hostguard" |
| 28 | "github.com/carlosframework/platform/internal/hostwrite" |
| 29 | "github.com/carlosframework/platform/internal/layout" |
| 30 | "github.com/carlosframework/platform/internal/lease" |
| 31 | "github.com/carlosframework/platform/internal/moves" |
| 32 | "github.com/carlosframework/platform/internal/park" |
| 33 | "github.com/carlosframework/platform/internal/provision" |
| 34 | "github.com/carlosframework/platform/internal/registry" |
| 35 | "github.com/carlosframework/platform/internal/release" |
| 36 | "github.com/carlosframework/platform/internal/stopevidence" |
| 37 | ) |
| 38 | |
| 39 | // The residence-move worker (issue #331, instance-move spec): everything a |
| 40 | // move does to DATA happens here, on the SOURCE box — the box whose |
| 41 | // CARLOS_REGION matches the record's region for as long as the move is in |
| 42 | // flight. The console only writes requests; the target box only restores. |
| 43 | // |
| 44 | // One pass = one delta round per in-flight move. The caller runs it on the |
| 45 | // same trigger as the reconcile pass (serial bumps land in seconds via |
| 46 | // carlos-adopt.path), and prepare's convergence is the round-over-round |
| 47 | // shrinking of the delta, not a resident goroutine — restart-safe because |
| 48 | // every step derives from the record and the buckets. "Converged" is a |
| 49 | // round that moved (copied or deferred) at most moveCfg.ReadyDelta bytes, |
| 50 | // 16 MiB by default, not a round that copied nothing: an awake instance |
| 51 | // never produces the latter (#175). Ready therefore means "small gap", and |
| 52 | // the cut's final round below is what makes staging exact. "Unchanged" is |
| 53 | // exact: the copy receipts (lease.Receipts, kept in this box's own bucket |
| 54 | // at activator.MoveReceiptsKey) are what let a multipart source, whose |
| 55 | // copy never shares its ETag, count as unchanged at all. |
| 56 | // |
| 57 | // The cut ordering is the never-two-writers argument, mechanically: |
| 58 | // |
| 59 | // route Status "moving" (edge stops serving; RetireExcept stops the child) |
| 60 | // → wake fence (no tick or sweep can wake it again — G12) |
| 61 | // → child quiesced (Quiesced(host) — retries next pass until true) |
| 62 | // → final mirror round (own replica is byte-final; staging matches it) |
| 63 | // → local purge (the warm-cache trap: a stale local file must |
| 64 | // never resurrect a moved-away history) |
| 65 | // → CAS flip+clear (Region=To, Move=nil, ONE write — the last act) |
| 66 | // |
| 67 | // Every step is idempotent; a box that dies mid-cut resumes at the top and |
| 68 | // falls through the already-done steps. |
| 69 | |
| 70 | // movesWorker is the one home for the worker's cross-pass state: the wired |
| 71 | // config, the memoized per-region mirror clients (built once per target |
| 72 | // region, not per pass — a client build is a credential-chain resolution), |
| 73 | // the once-per-host status nudge, and the per-host trouble throttle. |
| 74 | // RunAgent writes the pointer once; both converge loops read it. |
| 75 | type movesWorker struct { |
| 76 | cfg moveCfg |
| 77 | // pass runs one movesPass at a time. The fast converge and the |
| 78 | // instances loop both call runMovesPass, and two concurrent passes |
| 79 | // could drive one cut twice, or lift a fence from a record list read |
| 80 | // before the cut that installed it. |
| 81 | pass sync.Mutex |
| 82 | mu sync.Mutex |
| 83 | mirrors map[string]lease.MirrorStore // dstRegion → client pair |
| 84 | nudged sync.Map // host → true: status→quiesce hop nudged |
| 85 | trouble sync.Map // host → time.Time of last logged error |
| 86 | fenced sync.Map // host → uint64: the cut's wake-fence generation |
| 87 | |
| 88 | // Phase 1 (CARLOS_MOVES_PHASE1). stagers memoizes the staging writer per |
| 89 | // target bucket region; archives keeps T5's verdict per move and exact |
| 90 | // archive version; measured is this box's last T11 measurements; |
| 91 | // notices rate-limits the worker's standing log lines; strict is the |
| 92 | // set of hosts with a Phase 1 move in flight here (ExecBackend.Strict). |
| 93 | stagers map[string]lease.StagingWriter |
| 94 | archives map[string]verifiedArchive |
| 95 | measured moves.Measurements |
| 96 | notices sync.Map // key → time.Time of the last line |
| 97 | // logged is the move log identities this worker has handed off |
| 98 | // (movesevents.go): host/op/identity, bounded. |
| 99 | logged moveLogMemo |
| 100 | // pageWarned is the host/op/rule pages this worker has warned are not |
| 101 | // sent because no pager is wired (movespaging.go). |
| 102 | pageWarned sync.Map |
| 103 | strict atomic.Pointer[map[string]bool] |
| 104 | // holds is what reconcile and the schedule halves ask about this |
| 105 | // box's running moves (moveshold.go), rebuilt every pass. Nil with the |
| 106 | // switch off. |
| 107 | holds *movesHoldView |
| 108 | // landLeases is the landing's own-location lease per host, held from |
| 109 | // building through published (movesland.go); guarded by mu. |
| 110 | landLeases map[string]park.HeldLease |
| 111 | // lost names the hosts whose claim lists this box in Lost, as this pass |
| 112 | // read them (lostHere); rebuilt every pass under pass. |
| 113 | lost map[string]bool |
| 114 | // walkedIdle says the last pass walked the bucket, found no record with |
| 115 | // a move (and none it could not decode), and left this worker holding |
| 116 | // no move state of its own (localMoveState). Only then may a pass skip |
| 117 | // its walk on the record view's word (movesIdle); guarded by pass. |
| 118 | walkedIdle bool |
| 119 | } |
| 120 | |
| 121 | // staging is the memoized staging writer for a target bucket region. |
| 122 | func (w *movesWorker) staging(ctx context.Context, dstRegion string) (lease.StagingWriter, error) { |
| 123 | if w.cfg.Staging == nil { |
| 124 | return nil, errors.New("no staging writer on this box") |
| 125 | } |
| 126 | w.mu.Lock() |
| 127 | m := w.stagers[dstRegion] |
| 128 | w.mu.Unlock() |
| 129 | if m != nil { |
| 130 | return m, nil |
| 131 | } |
| 132 | m, err := w.cfg.Staging(ctx, dstRegion) |
| 133 | if err != nil { |
| 134 | return nil, err |
| 135 | } |
| 136 | w.mu.Lock() |
| 137 | if w.stagers == nil { |
| 138 | w.stagers = map[string]lease.StagingWriter{} |
| 139 | } |
| 140 | w.stagers[dstRegion] = m |
| 141 | w.mu.Unlock() |
| 142 | return m, nil |
| 143 | } |
| 144 | |
| 145 | // notice logs a standing condition at most once a minute per key. |
| 146 | func (w *movesWorker) notice(key, format string, args ...any) { |
| 147 | now := time.Now() |
| 148 | if last, ok := w.notices.Load(key); ok && now.Sub(last.(time.Time)) < time.Minute { |
| 149 | return |
| 150 | } |
| 151 | w.notices.Store(key, now) |
| 152 | log.Printf(format, args...) |
| 153 | } |
| 154 | |
| 155 | // strictHost is ExecBackend.Strict with the switch on (ruling F9): a host |
| 156 | // is strict while its record has a Phase 1 move in flight that names this |
| 157 | // box's region, as the last pass read it. It is the one strict setter. |
| 158 | func (w *movesWorker) strictHost(host string) bool { |
| 159 | m := w.strict.Load() |
| 160 | return m != nil && (*m)[host] |
| 161 | } |
| 162 | |
| 163 | // setStrict names the hosts whose spawns on this box must be strict: every |
| 164 | // host with a Phase 1 move in flight of which this box is a participant. |
| 165 | // The record's region is the source until the flip and the target after |
| 166 | // it, so the source is also matched by Move.From: it stays strict after |
| 167 | // the flip for as long as it is a participant (spec §5.3), and the close |
| 168 | // CAS that clears the move ends it. |
| 169 | func (w *movesWorker) setStrict(recs []provision.Record) { |
| 170 | m := map[string]bool{} |
| 171 | for _, rec := range recs { |
| 172 | if movesNamesBox(w.cfg, rec) { |
| 173 | m[rec.Host] = true |
| 174 | } |
| 175 | } |
| 176 | w.strict.Store(&m) |
| 177 | } |
| 178 | |
| 179 | // movesNamesBox reports whether rec carries a Phase 1 move naming this |
| 180 | // box's region: its record's, its source's or its target's. Each record |
| 181 | // code is compared with this box's engine region by the shared placement |
| 182 | // rule (park.PlacementMatches under the verified home binding, |
| 183 | // ParkHomeRegion), so the home's roster name and its "" code are one. |
| 184 | func movesNamesBox(cfg moveCfg, rec provision.Record) bool { |
| 185 | if rec.Move == nil || rec.Move.Legacy() || rec.Move.Offline() { |
| 186 | return false |
| 187 | } |
| 188 | here := cfg.parkRegion() |
| 189 | for _, code := range []string{rec.Region, rec.Move.To, rec.Move.From} { |
| 190 | if park.PlacementMatches(cfg.ParkHomeRegion, code, here) { |
| 191 | return true |
| 192 | } |
| 193 | } |
| 194 | return false |
| 195 | } |
| 196 | |
| 197 | // lostHere is the lost-box gate (moves Q9, Plan 4 Task 10, delta M15) for |
| 198 | // a record's round: a record that names this box's region is gated by |
| 199 | // hostBlocked. A record that names neither this box's region nor a move is |
| 200 | // not read. |
| 201 | func (w *movesWorker) lostHere(ctx context.Context, s release.Store, rec provision.Record) bool { |
| 202 | if !movesNamesBox(w.cfg, rec) { |
| 203 | return false |
| 204 | } |
| 205 | return w.hostBlocked(ctx, s, rec.Host) |
| 206 | } |
| 207 | |
| 208 | // hostBlocked decides, once per pass, whether this box runs no move step |
| 209 | // for host: the host's claim lists this box in Lost (until the claim |
| 210 | // readmits it), or the claim does not read. The quarantine's local |
| 211 | // retirement is the claim's (readmission requires its retirement receipt); |
| 212 | // moves adds only this gate. The decision holds for the whole pass (w.lost): |
| 213 | // the record's round, the target's report pass and the op journal sweep |
| 214 | // all skip a blocked host, and one standing notice is logged. |
| 215 | func (w *movesWorker) hostBlocked(ctx context.Context, s release.Store, host string) bool { |
| 216 | if w.cfg.Box == "" { |
| 217 | return false |
| 218 | } |
| 219 | if w.lost == nil { |
| 220 | w.lost = map[string]bool{} |
| 221 | } |
| 222 | if blocked, decided := w.lost[host]; decided { |
| 223 | return blocked |
| 224 | } |
| 225 | blocked := false |
| 226 | obs, err := hostclaim.Read(ctx, s, host) |
| 227 | switch { |
| 228 | case err != nil: |
| 229 | blocked = true |
| 230 | w.notice(host+"#lost", "moves: %s: the host claim does not read: this box runs no move step for it this pass: %v", host, err) |
| 231 | case !obs.Absent && obs.Claim != nil: |
| 232 | for _, box := range obs.Claim.Lost { |
| 233 | if box == w.cfg.Box { |
| 234 | blocked = true |
| 235 | w.notice(host+"#lost", "moves: %s: this box (%s) is lost on this host's claim: it runs no move step until the claim readmits it", host, box) |
| 236 | break |
| 237 | } |
| 238 | } |
| 239 | } |
| 240 | w.lost[host] = blocked |
| 241 | return blocked |
| 242 | } |
| 243 | |
| 244 | var armedMovesWorker atomic.Pointer[movesWorker] |
| 245 | |
| 246 | func (w *movesWorker) mirror(ctx context.Context, dstRegion string) (lease.MirrorStore, error) { |
| 247 | w.mu.Lock() |
| 248 | m := w.mirrors[dstRegion] |
| 249 | w.mu.Unlock() |
| 250 | if m != nil { |
| 251 | return m, nil |
| 252 | } |
| 253 | m, err := w.cfg.Mirror(ctx, dstRegion) |
| 254 | if err != nil { |
| 255 | return nil, err |
| 256 | } |
| 257 | w.mu.Lock() |
| 258 | w.mirrors[dstRegion] = m |
| 259 | w.mu.Unlock() |
| 260 | return m, nil |
| 261 | } |
| 262 | |
| 263 | // armMovesPass wires the worker with what only RunAgent has in hand: the |
| 264 | // activator's hibernate coordinates (the own-bucket side of every mirror), |
| 265 | // the data dir the cut purges, the registry the cut quiesces through, the |
| 266 | // service that answers whether a child is down, and the exec backend (the |
| 267 | // Phase 1 launch environment and S2). Called once, before svc.Start: with |
| 268 | // the Phase 1 switch on it sets eb.Strict, which the spawn path reads; with |
| 269 | // it off eb.Strict stays nil, so every spawn is main's. providers is the |
| 270 | // claim wiring's (newClaimWiring builds it before this): with both switches |
| 271 | // on, the park arms take their engine from its MoveParks (armMovesParks, |
| 272 | // Plan 5 Task 9a), behind the wake-source gate (movesWakeSourceRegistered, |
| 273 | // Plan 5 Task 9b), which the moves wake source's registration opens, before |
| 274 | // this runs, in the same agent start. events is the agent's |
| 275 | // rollout sink, which the move log tells the app's members through (Plan 6 |
| 276 | // Task 3). |
| 277 | func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, writes hostwrite.Config, providers ClaimProviders, events activator.EventSink, carry func(ctx context.Context, host, bucket, prefix string) error) { |
| 278 | dataDir = movesDataDir(dataDir) |
| 279 | readyDelta, ok := moveReadyDeltaFromEnv(os.Getenv) |
| 280 | if !ok { |
| 281 | log.Printf("agent: %s=%q ignored: want a whole number of bytes, 0 or more — keeping the built-in default of %d", moveReadyDeltaEnv, os.Getenv(moveReadyDeltaEnv), defaultMoveReadyDelta) |
| 282 | } |
| 283 | cfg := moveCfg{ |
| 284 | registerLanding: providers.registerLanding, |
| 285 | Region: strings.TrimSpace(os.Getenv("CARLOS_REGION")), |
| 286 | OwnBucket: hib.Bucket, |
| 287 | OwnBucketRegion: hib.Region, |
| 288 | OwnPrefix: hib.Prefix, |
| 289 | DataDir: dataDir, |
| 290 | Mirror: func(ctx context.Context, dstRegion string) (lease.MirrorStore, error) { |
| 291 | return lease.NewS3Mirror(ctx, hib.Region, dstRegion) |
| 292 | }, |
| 293 | Quiesced: svc.Quiesced, |
| 294 | Fence: func(ctx context.Context, host string, generation uint64) error { |
| 295 | // A short deadline: FenceAndStop waits for an awake child to |
| 296 | // stop, and one slow stop must not hold up every other move in |
| 297 | // the pass. ErrStopUnproven still means the fence is installed, |
| 298 | // which is all the cut asks of it; Quiesced answers the rest. |
| 299 | fctx, cancel := context.WithTimeout(ctx, 2*time.Second) |
| 300 | defer cancel() |
| 301 | if _, err := svc.FenceAndStop(fctx, host, generation); err != nil && !errors.Is(err, activator.ErrStopUnproven) { |
| 302 | return err |
| 303 | } |
| 304 | return nil |
| 305 | }, |
| 306 | LiftFence: svc.LiftFence, |
| 307 | SetStatus: reg.SetStatus, |
| 308 | CarryRequest: carry, |
| 309 | OwnsRoute: func(host string) bool { |
| 310 | rt, err := reg.Get(host) |
| 311 | return err == nil && rt != nil && rt.Provisioned |
| 312 | }, |
| 313 | Route: reg.Get, |
| 314 | Checkpoint: func(ctx context.Context, dbPath string) error { |
| 315 | host := strings.TrimSuffix(filepath.Base(dbPath), ".db") |
| 316 | rep, err := lease.ReplicateWithLease(ctx, dbPath, hib.Bucket, activator.HibernatePrefix(hib.Prefix, host), hib.Region) |
| 317 | if err != nil { |
| 318 | return err |
| 319 | } |
| 320 | defer rep.Close(ctx) |
| 321 | return rep.SyncAndWait(ctx) |
| 322 | }, |
| 323 | Now: time.Now, |
| 324 | ReadyDelta: readyDelta, |
| 325 | Enabled: moves.Enabled(os.Getenv), |
| 326 | claimRecords: writes.Enabled, |
| 327 | Page: PageOperator, |
| 328 | // The move log and the member stream (Plan 6 Task 3, preflight L9). |
| 329 | Events: events, |
| 330 | LogEvent: moveLogWriter(events), |
| 331 | } |
| 332 | // The edges box label and the version store, whatever the switch says: |
| 333 | // the offline move reports under the label (movesoffline.go). armPhase1 |
| 334 | // sets the same two values. |
| 335 | cfg.Box = metricsBoxLabel(os.Getenv, readMachineID()) |
| 336 | cfg.OptDir = release.OptDir() |
| 337 | w := newMovesWorker(cfg) |
| 338 | if cfg.Enabled { |
| 339 | armPhase1(&cfg, w, hib, dataDir, reg, svc, eb, writes, os.Getenv) |
| 340 | if writes.Enabled && providers.guard.FenceView != nil { |
| 341 | cfg.Guard = providers.guard |
| 342 | cfg.Local.Guard = providers.guard |
| 343 | } |
| 344 | w.holds = newMovesHoldView(cfg.Box, cfg.Region, hostclaim.Read, holdStageRequester(dataDir)) |
| 345 | cfg.TimersDrained = w.holds.drainTimers |
| 346 | if cfg.claimRecords { |
| 347 | ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) |
| 348 | if err := armMovesParks(ctx, &cfg, providers); err != nil { |
| 349 | log.Printf("agent: moves: the park arms stay unarmed: %v", err) |
| 350 | } |
| 351 | cancel() |
| 352 | } |
| 353 | } |
| 354 | armOffline(&cfg, dataDir, reg, svc, eb, writes, os.Getenv) |
| 355 | w.cfg = cfg |
| 356 | armedMovesWorker.Store(w) |
| 357 | } |
| 358 | |
| 359 | // armOffline sets every moveCfg field the offline operator move reads |
| 360 | // (movesoffline.go and Tasks 4 and 5 of docs/superpowers/plans/2026-10-20- |
| 361 | // offline-operator-move.md), whatever the Phase 1 switch says, after |
| 362 | // armPhase1 so a hook it already set is kept. Nothing here acts: the fields |
| 363 | // are read only by offline steps, which run only on a capable box and only |
| 364 | // for a record with an offline move. |
| 365 | func armOffline(cfg *moveCfg, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, writes hostwrite.Config, getenv func(string) string) { |
| 366 | cfg.OfflineCapable = offlineCapable(getenv, hibernationConfigured(getenv)) |
| 367 | cfg.Offline = newOfflineView(*cfg, holdStageRequester(dataDir)) |
| 368 | if svc != nil { |
| 369 | cfg.FenceAs = fenceAsWith(svc) |
| 370 | cfg.LiftFenceAs = svc.LiftFenceAs |
| 371 | } |
| 372 | if reg != nil { |
| 373 | cfg.Registry = reg |
| 374 | // The caller holds provision.WithHostHoldLock; a release that |
| 375 | // landed touches the stage-requested marker, so root re-judges the |
| 376 | // now-ready row (registry.SetHold's contract, G10). |
| 377 | cfg.ReleaseHold = releaseHeld(reg, holdStageRequester(dataDir)) |
| 378 | // The row and the timers the move landed on it, in one |
| 379 | // transaction, only while op holds the row. |
| 380 | cfg.DeleteIfHold = reg.DeleteIfHoldWithTimers |
| 381 | } |
| 382 | if eb != nil { |
| 383 | cfg.LaunchLayers = eb.LaunchLayers |
| 384 | cgRoot := eb.CgroupRoot |
| 385 | if cgRoot == "" { |
| 386 | cgRoot = "/sys/fs/cgroup" |
| 387 | } |
| 388 | execRoot := eb.Cgroups.Root() |
| 389 | cfg.Procs = func(host string) (int, error) { |
| 390 | return cgroup.HostProcs(execRoot, host, hostSliceCgroup(cgRoot, host)) |
| 391 | } |
| 392 | } |
| 393 | if raw := strings.TrimSpace(getenv("CARLOS_SECRETS_KEY")); raw != "" { |
| 394 | if priv, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY"); err == nil { |
| 395 | cfg.ConfigPriv, cfg.ConfigPub = priv, priv.PublicKey() |
| 396 | if cfg.SecretsKey == "" { |
| 397 | cfg.SecretsKey = raw |
| 398 | } |
| 399 | } |
| 400 | } |
| 401 | if cfg.RenderHeld == nil { |
| 402 | cfg.RenderHeld = heldRenderer(writes, cfg.ConfigPriv, dataDir, cfg.Box) |
| 403 | } |
| 404 | optDir := cfg.OptDir |
| 405 | cfg.LiveVersion = func(host string) (string, error) { return release.LiveVersion(optDir, host) } |
| 406 | pub, perr := release.ParsePublicKey(getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY") |
| 407 | cfg.ChannelVersion = func(ctx context.Context, s release.Store, rt registry.Route) (string, error) { |
| 408 | if perr != nil { |
| 409 | return "", perr |
| 410 | } |
| 411 | return channelVersion(ctx, s, pub, rt) |
| 412 | } |
| 413 | if cfg.Now == nil { |
| 414 | cfg.Now = time.Now |
| 415 | } |
| 416 | } |
| 417 | |
| 418 | // channelVersion is the build the row's channel points at, as adoption |
| 419 | // resolves it: the account-qualified pointer of the row's account, app and |
| 420 | // channel ("stable" when the row names none), verified under the release |
| 421 | // key. |
| 422 | func channelVersion(ctx context.Context, s release.Store, pub ed25519.PublicKey, rt registry.Route) (string, error) { |
| 423 | channel := rt.Channel |
| 424 | if channel == "" { |
| 425 | channel = "stable" |
| 426 | } |
| 427 | ptr, err := release.Channel(ctx, s, pub, release.RouteAccount(rt), rt.App, channel) |
| 428 | if err != nil { |
| 429 | return "", err |
| 430 | } |
| 431 | return ptr.Version, nil |
| 432 | } |
| 433 | |
| 434 | // movesDataDir is the data directory the moves worker is built with, |
| 435 | // resolved once to an absolute, cleaned path (final re-review 1, I2): the |
| 436 | // outboxes under it sync every directory from an entry's own up to it, and |
| 437 | // that walk ends only at an absolute root. A CARLOS_DATA_DIR of "." or any |
| 438 | // relative path names the same directory as before. If the working |
| 439 | // directory cannot be read the path is kept as given and logged; the |
| 440 | // outboxes then refuse every write (outboxCreate), so nothing is |
| 441 | // acknowledged that is not durable. |
| 442 | func movesDataDir(dd string) string { |
| 443 | abs, err := filepath.Abs(dd) |
| 444 | if err != nil { |
| 445 | log.Printf("agent: moves: the data directory %q cannot be made absolute, so the moves outboxes keep nothing: %v", dd, err) |
| 446 | return dd |
| 447 | } |
| 448 | return abs |
| 449 | } |
| 450 | |
| 451 | // stagingLeaseKey is the staging lease of a lane in targetRegion, the |
| 452 | // lane's placement code (the record's To, or its From for a return lane): |
| 453 | // moves.StagingLeasePrefix keyed by the lane's engine region, so a lane in |
| 454 | // home is park.HomeRegion and never an empty segment |
| 455 | // (control/staging-leases//<host>; home-region parks, part 1). Named codes |
| 456 | // keep their keys. A code that is no placement refuses. |
| 457 | func stagingLeaseKey(targetRegion, host string) (string, error) { |
| 458 | r := park.PlacementRegion(targetRegion) |
| 459 | if r == "" { |
| 460 | return "", fmt.Errorf("%w: %q is not a placement code", park.ErrUnsafe, targetRegion) |
| 461 | } |
| 462 | return moves.StagingLeasePrefix(r, host), nil |
| 463 | } |
| 464 | |
| 465 | // errHeldRenderUnpinned is the held render under the claim refusing, before |
| 466 | // anything is selected or written, a merge whose inputs are not the |
| 467 | // landing's pins: the configured step turns it into the landing's |
| 468 | // "configuration changed" refusal (refuseConfigChanged), never the policy's |
| 469 | // refusal of the write (the probe's rule, fix round 1 finding 4). |
| 470 | var errHeldRenderUnpinned = errors.New("moves: the held render's inputs are not the pinned ones") |
| 471 | |
| 472 | // heldRenderer is the held row's render (§9.1; Task 7, Q2.2). With the claim |
| 473 | // off (writes.Enabled false) it is appconfig.RenderHeldBefore, unchanged. |
| 474 | // Under the claim it is appconfig.RenderHeldUnderBefore (platform-1e's |
| 475 | // answer to Q2.2): once the merged versions are known, a pin that is not |
| 476 | // them is errHeldRenderUnpinned, and otherwise the write is selected as |
| 477 | // held-config for the host (writes.Select) under the identity of exactly |
| 478 | // the merged configuration (HeldConfigIdentity), which MovesClaimPolicy |
| 479 | // grants at configured only when it is the pinned one (or, for a return, |
| 480 | // the recovery's). before still journals the rendered bytes' sha256 before |
| 481 | // the selection writes them (carry (c)). |
| 482 | func heldRenderer(writes hostwrite.Config, priv *ecdh.PrivateKey, dataDir, box string) func(ctx context.Context, s release.Store, rt registry.Route, op string, pin *provision.MoveInputs, before func(string) error) (appconfig.HeldRender, error) { |
| 483 | if !writes.Enabled { |
| 484 | return func(ctx context.Context, s release.Store, rt registry.Route, op string, _ *provision.MoveInputs, before func(string) error) (appconfig.HeldRender, error) { |
| 485 | return appconfig.RenderHeldBefore(ctx, s, priv, dataDir, rt, op, box, before) |
| 486 | } |
| 487 | } |
| 488 | return func(ctx context.Context, s release.Store, rt registry.Route, op string, pin *provision.MoveInputs, before func(string) error) (appconfig.HeldRender, error) { |
| 489 | return appconfig.RenderHeldUnderBefore(ctx, s, priv, dataDir, rt, op, box, func(pre appconfig.HeldRender) (hostwrite.Work, error) { |
| 490 | if pin != nil && (!sameBundle(pre.BundleETags, pin.BundleETags) || pre.OverlayETag != pin.OverlayETag) { |
| 491 | return nil, errHeldRenderUnpinned |
| 492 | } |
| 493 | w, err := writes.Select(ctx, rt.Host, movesActionHeldConfig) |
| 494 | if err != nil { |
| 495 | return nil, err |
| 496 | } |
| 497 | return heldIdentityWork{Work: w, ID: HeldConfigIdentity(pre.BundleETags, pre.OverlayETag)}, nil |
| 498 | }, before) |
| 499 | } |
| 500 | } |
| 501 | |
| 502 | // armPhase1 wires the Phase 1 hooks (moves design §7.3). Only with the |
| 503 | // switch on: off, none is set and eb.Strict stays nil. writes is the |
| 504 | // agent's host-write configuration (the claim wiring's): the landing's |
| 505 | // held timers import, held render and held release select their Work |
| 506 | // through it. |
| 507 | func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, writes hostwrite.Config, getenv func(string) string) { |
| 508 | cfg.Box = metricsBoxLabel(getenv, readMachineID()) |
| 509 | cfg.OptDir = release.OptDir() |
| 510 | cfg.DeployBucket = strings.TrimSpace(getenv("CARLOS_DEPLOYMENT_BUCKET")) |
| 511 | if cfg.DeployBucket != "" { |
| 512 | cfg.DeployBucketRegion = edgesFileRegion(getenv, readPlacementRegion()) |
| 513 | bucket, region := cfg.DeployBucket, cfg.DeployBucketRegion |
| 514 | cfg.StagingLease = func(ctx context.Context, targetRegion, host string) (moves.SafeUntilGuard, func(context.Context) error, error) { |
| 515 | key, err := stagingLeaseKey(targetRegion, host) |
| 516 | if err != nil { |
| 517 | return nil, nil, err |
| 518 | } |
| 519 | h, err := lease.AcquireTTL(ctx, bucket, key, region, moves.StagingLeaseTTL) |
| 520 | if err != nil { |
| 521 | return nil, nil, err |
| 522 | } |
| 523 | return h, h.Release, nil |
| 524 | } |
| 525 | } |
| 526 | cfg.Staging = func(ctx context.Context, dstRegion string) (lease.StagingWriter, error) { |
| 527 | return lease.NewStagingMirror(ctx, hib.Region, dstRegion) |
| 528 | } |
| 529 | cfg.Versioned = func(ctx context.Context, bucket, region string) (lease.VersionedStore, error) { |
| 530 | return lease.NewVersionedS3(ctx, map[string]string{bucket: region}) |
| 531 | } |
| 532 | if raw := strings.TrimSpace(getenv("CARLOS_SECRETS_KEY")); raw != "" { |
| 533 | if priv, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY"); err != nil { |
| 534 | log.Printf("agent: moves: CARLOS_SECRETS_KEY does not parse; Phase 1 rounds refuse: %v", err) |
| 535 | } else { |
| 536 | cfg.SecretsKey, cfg.SecretsPriv, cfg.SecretsPub = raw, priv, priv.PublicKey() |
| 537 | } |
| 538 | } |
| 539 | cfg.FilesDir = movesFilesDirFor(dataDir) |
| 540 | // The cut (movescut.go). |
| 541 | cfg.FinalCheckpoint = finalCheckpoint(hib.Bucket, hib.Prefix, hib.Region) |
| 542 | cfg.LocalPosition = localLTXPosition |
| 543 | cfg.InputVersions = func(ctx context.Context, s release.Store, rec provision.Record) (map[string]string, string, error) { |
| 544 | _, v, err := appconfig.LoadMergedHostVersions(ctx, s, rec.Account, rec.App, rec.Environment, rec.Host) |
| 545 | if err != nil { |
| 546 | return nil, "", err |
| 547 | } |
| 548 | return map[string]string{"default": v.Default, "environment": v.Environment}, v.Overlay, nil |
| 549 | } |
| 550 | if reg != nil { |
| 551 | cfg.HoldMoving = holdMoving(reg, holdStageRequester(dataDir)) |
| 552 | cfg.Timers = reg.TimersFor |
| 553 | } |
| 554 | // The landing (movesland.go). Everything that exists on main is wired |
| 555 | // here; ParkLeases, ParkView and ParkAuthorize come from the claim's |
| 556 | // MoveParks (armMovesParks, Plan 5 Task 9a, in armMovesPass), with |
| 557 | // park.Engine.RunHeld and the MoveParks local authorizer, and are set |
| 558 | // only once the wake-source gate opens (movesWakeSourceRegistered, Plan 5 |
| 559 | // Task 9b: the moves wake source registered with both switches on); until |
| 560 | // then landingArmed is false and the landing pass only logs. The held writes select their Work from writes (ruling |
| 561 | // G16; Task 7, Q2): the timers through schedule.SelectedTimers as |
| 562 | // held-timers (heldTimersReplacer), the render as held-config |
| 563 | // (heldRenderer) and the release as held-release (heldReleaseHooks). |
| 564 | // With CARLOS_HOST_CLAIM off each is exactly what it was; with it on, |
| 565 | // the claim's selection and hostbox.Policy.Moves (MovesClaimPolicy) |
| 566 | // decide. The landing still refuses at checking under the flag until |
| 567 | // the wake-source gate opens (movesland.go). |
| 568 | cfg.Guard = hostguard.Guard{DataDir: dataDir} |
| 569 | cfg.Local = park.Local{DataDir: dataDir, Box: cfg.Box} |
| 570 | cfg.OwnLease = hostbox.RenewingLeases{Region: hib.Region}.Acquire |
| 571 | cfg.RenderHeld = heldRenderer(writes, cfg.SecretsPriv, dataDir, cfg.Box) |
| 572 | cfg.RemoveVerifyEntry = dbverify.RemoveHost |
| 573 | cfg.WriteGen = hostbox.WriteGenerationMarker |
| 574 | if reg != nil { |
| 575 | cfg.ReplaceTimers = heldTimersReplacer(writes, reg) |
| 576 | cfg.SetReplication = reg.SetReplication |
| 577 | cfg.SetNextDue = reg.SetNextDue |
| 578 | // The source's finalization (movesfinal.go): a plain registry call |
| 579 | // under the host's hold lock, not a claim-selected write (ruling |
| 580 | // H5(d)). Unarmed with ParkLeases, as the landing is. |
| 581 | cfg.DeleteHeld = reg.DeleteIfHoldWithTimers |
| 582 | // The rollback's release hands the row's hold from the move to the |
| 583 | // rollback, touching the stage-requested marker (ruling J3). |
| 584 | cfg.HandHold = handHold(reg, holdStageRequester(dataDir)) |
| 585 | // The recovery's open ends the rollback's hold on the source's row, |
| 586 | // a plain registry call under the host's hold lock (J12). |
| 587 | cfg.ReleaseHeld = releaseHeld(reg, holdStageRequester(dataDir)) |
| 588 | // The revert's hold goes from moving to restoring under the move |
| 589 | // id, a conditional status change under the host's hold lock |
| 590 | // (movesrevert.go, Task 6). |
| 591 | cfg.RestatusHold = restatusHold(reg, holdStageRequester(dataDir)) |
| 592 | } |
| 593 | if eb != nil { |
| 594 | // A box with no delegated cgroup has no manager, and Reconstruct |
| 595 | // refuses it: such a box cannot produce stop evidence, so it never |
| 596 | // cuts (Task 3). The lease step reads the activator's own lock, the |
| 597 | // one its writer holds (ExecBackend.Acquire); the cut has refused |
| 598 | // any holder whose location is not that prefix before it gets |
| 599 | // here (ruling T5-lease), and a location that still differs is an |
| 600 | // error, never evidence. |
| 601 | cfg.Evidence = func(ctx context.Context, host, location string, dbPaths []string) (moves.StoppedReport, error) { |
| 602 | writerPrefix := activator.HibernatePrefix(hib.Prefix, host) |
| 603 | if strings.TrimSuffix(location, "/") != writerPrefix { |
| 604 | return moves.StoppedReport{}, fmt.Errorf("%w: %s is at %q, not the writer's lease %q", errHolderLocation, host, location, writerPrefix) |
| 605 | } |
| 606 | return stopevidence.Reconstruct(ctx, stopevidence.ReconstructInputs{Host: host, Cgroups: eb.Cgroups, DBPaths: dbPaths, |
| 607 | LeaseFree: func(ctx context.Context) (bool, error) { |
| 608 | held, _, _, err := lease.LockState(ctx, hib.Bucket, writerPrefix, hib.Region) |
| 609 | return !held, err |
| 610 | }}) |
| 611 | } |
| 612 | } |
| 613 | cfg.DiskFree = diskBytes |
| 614 | cfg.ProbeStore = probeStores |
| 615 | cfg.LaneProbe = laneProbe(cfg.Versioned) |
| 616 | cfg.MeasureDisk = measureDiskIn(dataDir, time.Now) |
| 617 | cfg.MeasureS3 = measureS3(func(ctx context.Context) (lease.MirrorStore, error) { return w.mirror(ctx, hib.Region) }, hib.Bucket, time.Now) |
| 618 | if reg != nil { |
| 619 | // D3-held: the held path's own gate, for the life of the worker, so |
| 620 | // its miss cache holds between passes; its bake directory is the |
| 621 | // healthy history adoption reads. |
| 622 | // |
| 623 | // Settings spec rev 11 §4.3a: the held stage installs only a build a |
| 624 | // promotion record or a pointer verified under the release key, and |
| 625 | // admitted by the agent's pointer digest ratchet, authorizes. A key |
| 626 | // that does not parse leaves pub nil, and every held stage is |
| 627 | // refused: fail closed. |
| 628 | pub, perr := release.ParsePublicKey(getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY") |
| 629 | if perr != nil { |
| 630 | log.Printf("moves: held stages are refused: %v", perr) |
| 631 | pub = nil |
| 632 | } |
| 633 | // |
| 634 | // Paul's T7 ruling (2026-10-12): under the claim T7 stages and |
| 635 | // verifies without publishing, and the configured step publishes |
| 636 | // under the held-release selection (heldReleaseHooks). With the |
| 637 | // claim off both are heldReleaseReady, as before. |
| 638 | // |
| 639 | // Task 7b: under the claim the agent cannot write /opt/carlos, so a |
| 640 | // blocked stage and every publication are asked of root over its |
| 641 | // control socket (heldReleaseHooksVia), which re-reads the claim, |
| 642 | // the record and the render journal and publishes only inside its |
| 643 | // own held-release selection. |
| 644 | via := heldReleaseVia{Region: cfg.Region} |
| 645 | if writes.Enabled { |
| 646 | via.Relay = newHeldReleaseRootClient(controlsock.Path) |
| 647 | } |
| 648 | cfg.ReleaseReady, cfg.ReleaseLive = heldReleaseHooksVia(via, writes, reg, cfg.OptDir, dataDir, cfg.Box, pub, pointerRatchet(false, cfg.OptDir, dataDir, getenv), |
| 649 | release.NewRootProofGate(release.ProofPolicyPath, dataDir), stageMarkerRequester(dataDir)) |
| 650 | if svc != nil { |
| 651 | // S1 asks the instance itself, woken through the activator. |
| 652 | // An uncommitted import's row (a gate set, or closed by its |
| 653 | // rollback) is never woken for S1: only its own verify requests |
| 654 | // reach it (seeded-instances spec §5.4), and it moves only once |
| 655 | // committed (§5.8). ServeScheduled does not read the gate, so |
| 656 | // this does, for every round that probes, preparing or ready. |
| 657 | cfg.ReadyProbe = func(ctx context.Context, host, path string) moves.Check { |
| 658 | rt, err := reg.Get(host) |
| 659 | if err != nil || rt == nil || rt.ImportGate != "" { |
| 660 | return moves.Check{Reason: s1NoAnswer(host)} |
| 661 | } |
| 662 | return probeInstance(ctx, svc.ServeScheduled, *rt, path) |
| 663 | } |
| 664 | } |
| 665 | } |
| 666 | if eb != nil { |
| 667 | cfg.StrictNext = eb.NextSpawnStrict |
| 668 | cfg.ComposeHeldEnv = composeHeldEnv(cfg.SecretsPriv, cfg.Region, eb.Cache != nil) |
| 669 | if reg != nil { |
| 670 | cfg.LaunchEnv = func(host string) (map[string]string, map[string]bool, error) { |
| 671 | rt, err := reg.Get(host) |
| 672 | if err != nil { |
| 673 | return nil, nil, err |
| 674 | } |
| 675 | if rt == nil { |
| 676 | return nil, nil, fmt.Errorf("%s has no registry row on this box", host) |
| 677 | } |
| 678 | env, instance, _, err := eb.LaunchLayers(*rt) |
| 679 | return env, instance, err |
| 680 | } |
| 681 | } |
| 682 | eb.Strict = w.strictHost |
| 683 | eb.Phase1 = true |
| 684 | } |
| 685 | // Recovery uses the same admitted Group D start path as other claim |
| 686 | // writers. Preserve the legacy nil hook with the claim switch off. |
| 687 | if cfg.claimRecords && svc != nil { |
| 688 | armRecoveryProbe(cfg, w, svc.RecoveryProbe, svc.RecoveryProbeReady) |
| 689 | } |
| 690 | } |
| 691 | |
| 692 | // runMovesPass is the loops' call: inert until armed (no activator = no |
| 693 | // exec instances = nothing this box could ever move). |
| 694 | func runMovesPass(ctx context.Context, s release.Store) { |
| 695 | w := armedMovesWorker.Load() |
| 696 | if w == nil || s == nil { |
| 697 | return |
| 698 | } |
| 699 | if err := movesPass(ctx, s, w); err != nil { |
| 700 | log.Printf("agent: moves pass: %v", err) |
| 701 | } |
| 702 | } |
| 703 | |
| 704 | // moveCfg carries what the worker needs, injectable for tests. |
| 705 | type moveCfg struct { |
| 706 | registerLanding func(context.Context, hostclaim.Observed) error |
| 707 | claimRecords bool // Captured CARLOS_HOST_CLAIM, independent of the moves switch. |
| 708 | Region string // this box's placement code, "" included (the home) |
| 709 | OwnBucket string // CARLOS_HIBERNATE_BUCKET |
| 710 | OwnBucketRegion string // CARLOS_HIBERNATE_REGION |
| 711 | OwnPrefix string // CARLOS_HIBERNATE_PREFIX ("instances" in production) |
| 712 | DataDir string // /data — where the local DB lives |
| 713 | Mirror func(ctx context.Context, dstRegion string) (lease.MirrorStore, error) |
| 714 | Quiesced func(host string) bool |
| 715 | // Fence installs the activator's wake fence for host (G12, the |
| 716 | // instance-moves design §6.1): status "moving" stops the edge, but the |
| 717 | // tick runner fires whatever row the registry returns, and a tick that |
| 718 | // woke the source after the purge would restore it from its own replica |
| 719 | // and write where the target never looks. A nil error means the fence |
| 720 | // is installed, whether or not the child has finished stopping. Nil |
| 721 | // skips (tests without an activator). |
| 722 | Fence func(ctx context.Context, host string, generation uint64) error |
| 723 | // LiftFence removes a fence Fence installed, at that generation only. |
| 724 | LiftFence func(host string, generation uint64) bool |
| 725 | SetStatus func(host, status string) error |
| 726 | // OwnsRoute answers whether THIS box's registry serves the host |
| 727 | // (review round 1): region equality alone would let a standby or |
| 728 | // second same-region box drive a cut whose child runs on its sibling — |
| 729 | // each box has its OWN registry, so the sibling's Quiesced is |
| 730 | // trivially, wrongly true. The box that owns the route owns the move. |
| 731 | OwnsRoute func(host string) bool |
| 732 | // Route reads the row itself, so a cutting move can checkpoint the |
| 733 | // row's OWN DBPath (isolated-exec design §9) instead of guessing the |
| 734 | // legacy <DataDir>/<host>.db. The guess is wrong for a nested row, |
| 735 | // whose DB lives under provision.InstanceDir. Nil (a test with no |
| 736 | // registry in hand) keeps the guess, unchanged. |
| 737 | Route func(host string) (*registry.Route, error) |
| 738 | // Checkpoint re-replicates the LOCAL DB to the own bucket and waits |
| 739 | // (review round 1): the activator's stop logs — and swallows — a failed |
| 740 | // final checkpoint, because pre-move the local warm cache made that |
| 741 | // safe. The cut deletes the warm cache, so it must first prove the own |
| 742 | // replica really carries the tail. Nil skips (tests without S3). |
| 743 | Checkpoint func(ctx context.Context, dbPath string) error |
| 744 | Now func() time.Time |
| 745 | // ReadyDelta is the readiness bar (#175): a preparing round whose |
| 746 | // lease.MirrorRound.DeltaBytes (bytes copied plus bytes deferred) is at |
| 747 | // most this promotes the move to ready. Zero means zero bytes moved: a |
| 748 | // round that only deleted still qualifies. armMovesPass sets it from |
| 749 | // moveReadyDeltaFromEnv. |
| 750 | ReadyDelta int64 |
| 751 | // Enabled is the Phase 1 moves switch (moves.Enabled, set in |
| 752 | // armMovesPass). Off, the worker runs the legacy protocol only. |
| 753 | Enabled bool |
| 754 | |
| 755 | // CarryRequest merges this box's consumed request record for host into |
| 756 | // the build state at bucket/prefix (release.CarryRequest: root's |
| 757 | // resolution and a failed completion; never a successful completion, |
| 758 | // never last_executed), under the cut's fence or lease, before a cut's |
| 759 | // final round copies the state for the seal: the legacy cut, the Phase 1 |
| 760 | // cut's final round and a dirty return's lane (ruling R-SPEC-43A-2; Astra |
| 761 | // round 1 finding 3). A request this box closed as failed then stays |
| 762 | // closed on the destination. Nil carries nothing (a box with no |
| 763 | // build-state store, tests). |
| 764 | CarryRequest func(ctx context.Context, host, bucket, prefix string) error |
| 765 | |
| 766 | // Phase 1 (moves design §7.3), wired by armMovesPass only with the |
| 767 | // switch on. A nil hook refuses its step, never falls back. |
| 768 | // |
| 769 | // Box is this box's label (metricsBoxLabel), the name the claim's |
| 770 | // participants and op.intent.target use. OptDir is the version store |
| 771 | // (the source report's Release is LiveVersion there). |
| 772 | Box string |
| 773 | OptDir string |
| 774 | // DeployBucket and DeployBucketRegion are where the staging-work lease |
| 775 | // lives (CARLOS_DEPLOYMENT_BUCKET, CARLOS_AWS_REGION or the IMDS |
| 776 | // region). StagingLease is nil without a deployment bucket, and then a |
| 777 | // Phase 1 round refuses (round-5 finding 11). |
| 778 | DeployBucket, DeployBucketRegion string |
| 779 | StagingLease func(ctx context.Context, targetRegion, host string) (moves.SafeUntilGuard, func(context.Context) error, error) |
| 780 | // Staging is the staging writer for a target bucket region (Task 6, |
| 781 | // F17): every staging write goes through it behind the guard. |
| 782 | Staging func(ctx context.Context, dstRegion string) (lease.StagingWriter, error) |
| 783 | // Versioned reads a bucket by exact version: the target's T5 in its own |
| 784 | // bucket and T10's lane in the source's. Never the source against the |
| 785 | // target's lane. |
| 786 | Versioned func(ctx context.Context, bucket, region string) (lease.VersionedStore, error) |
| 787 | // SecretsKey is CARLOS_SECRETS_KEY (the launch report's HMAC key is |
| 788 | // derived from it); SecretsPub seals the archive and SecretsPriv opens |
| 789 | // it on the target. |
| 790 | SecretsKey string |
| 791 | SecretsPub *ecdh.PublicKey |
| 792 | SecretsPriv *ecdh.PrivateKey |
| 793 | // LaunchEnv is the source's composed launch environment and the names |
| 794 | // the instance's own layers supplied (ExecBackend.LaunchLayers over the |
| 795 | // host's row). |
| 796 | LaunchEnv func(host string) (env map[string]string, instance map[string]bool, err error) |
| 797 | // FilesDir is the files directory and <host>.env the archive carries |
| 798 | // (flat: <DataDir>/<host>/ and <DataDir>/<host>.env). |
| 799 | FilesDir func(host string) (dir, envPath string) |
| 800 | // ReadyProbe is S1: GET path from the instance itself, woken through |
| 801 | // the activator (probeInstance). StrictNext is S2 |
| 802 | // (ExecBackend.NextSpawnStrict). |
| 803 | ReadyProbe func(ctx context.Context, host, path string) moves.Check |
| 804 | StrictNext func(host string) moves.Check |
| 805 | // The target's hooks: T3's composition of the held row, T4's store |
| 806 | // probe, T6's disk reading, T7's release, T10's lane probe and T11's |
| 807 | // regional S3 and disk measurements. |
| 808 | ComposeHeldEnv func(ctx context.Context, s release.Store, rec provision.Record, carriedEnv []byte) (env map[string]string, instance map[string]bool, shadowed []string, in provision.MoveInputs, err error) |
| 809 | ProbeStore func(ctx context.Context, env map[string]string, moveID string) (prefix string, err error) |
| 810 | DiskFree func(dir string) (total, avail uint64, err error) |
| 811 | ReleaseReady func(ctx context.Context, s release.Store, rec provision.Record, version string) moves.Check |
| 812 | LaneProbe func(ctx context.Context, bucket, region, moveID string) error |
| 813 | MeasureS3 func(ctx context.Context, key string) (moves.MeasuredRate, error) |
| 814 | MeasureDisk func(ctx context.Context) (moves.MeasuredRate, moves.FilesRate, error) |
| 815 | |
| 816 | // ReleaseLive is the configured step's release (§9.2 as amended |
| 817 | // 2026-10-12): the release published for the held row and proven. With |
| 818 | // the claim off it is ReleaseReady itself, which publishes; under the |
| 819 | // claim T7 only stages, and this publishes under the landing's |
| 820 | // held-release selection (heldReleaseHooks). Nil is ReleaseReady with |
| 821 | // the claim off; under the claim a nil ReleaseLive never falls back to |
| 822 | // T7's hook, and configured waits (moveCfg.releaseLive). |
| 823 | ReleaseLive func(ctx context.Context, s release.Store, rec provision.Record, version string) moves.Check |
| 824 | |
| 825 | // The cut's hooks (§5.3, §7.4, movescut.go). HoldMoving holds the row |
| 826 | // "moving" under the move id (holdMoving). TimersDrained waits until |
| 827 | // no timer delivery or convergence that could have missed the hold is |
| 828 | // still running (the hold view's drainTimers, Task 6). Evidence is |
| 829 | // §5.3 steps 1 to 4 (stopevidence.Reconstruct, its lease step read on |
| 830 | // the activator's own lease, which ruling T5-lease requires to be the |
| 831 | // claim holder's location). FinalCheckpoint is step 5's checkpoint on |
| 832 | // that same lease, bound to the claim's lineage, check being the |
| 833 | // authority read before each replica request (G3). LocalPosition is |
| 834 | // the stopped database's TXID (the brief's OwnReplicaPosition, G22). |
| 835 | // Timers reads the host's timer rows. InputVersions reads the current |
| 836 | // bundle etags (default, environment) and overlay etag, as T3 pinned |
| 837 | // them. |
| 838 | HoldMoving func(host, op string) error |
| 839 | TimersDrained func(ctx context.Context, host string) error |
| 840 | Evidence func(ctx context.Context, host, location string, dbPaths []string) (moves.StoppedReport, error) |
| 841 | FinalCheckpoint func(ctx context.Context, host, dbPath string, holder hostclaim.Holder, check func(context.Context) error) error |
| 842 | LocalPosition func(dbPath string) (uint64, error) |
| 843 | Timers func(host string) ([]registry.Timer, error) |
| 844 | InputVersions func(ctx context.Context, s release.Store, rec provision.Record) (bundle map[string]string, overlay string, err error) |
| 845 | // cutFault is a test seam: called after each side effect of the cut |
| 846 | // with the point's name, an error being a crash there. Nil in |
| 847 | // production. |
| 848 | cutFault func(point string) error |
| 849 | |
| 850 | // The landing's hooks (§7.6, movesland.go). Guard is the host guard |
| 851 | // over this box's data directory (its Claims are set per pass from the |
| 852 | // pass's store when nil); Local is park.Local over the same directory |
| 853 | // (its Guard, Claims and, when nil, Authorize are set per pass). |
| 854 | // ParkLeases, ParkView and ParkAuthorize are the claim's park engine's |
| 855 | // (MoveParks, armMovesParks): its leases, view and authorizer over the |
| 856 | // own bucket. Every arm requires all three (parkHooksArmed) and runs its |
| 857 | // own check before the authorizer (movesArmAuthorize); nothing stands |
| 858 | // in. In production they are set only once the wake-source gate opens |
| 859 | // (movesWakeSourceRegistered, Plan 5 Task 9b), which leaves every arm |
| 860 | // unarmed there. OwnLease |
| 861 | // takes the location's lease in the own bucket for the build |
| 862 | // (hostbox.RenewingLeases over lease.Acquire). RenderHeld is the held |
| 863 | // row's render (heldRenderer): appconfig.RenderHeldBefore with the claim |
| 864 | // off, appconfig.RenderHeldUnderBefore under a held-config selection with |
| 865 | // it on; before journals the render's ownership ahead of its write, |
| 866 | // carry (c), and pin, when not nil, is the landing's pinned inputs, which |
| 867 | // a render under the claim must merge before anything is selected or |
| 868 | // written (errHeldRenderUnpinned); ReplaceTimers is the |
| 869 | // registry's through schedule.SelectedTimers with a held-timers Work |
| 870 | // (heldTimersReplacer); SetReplication and SetNextDue are the |
| 871 | // registry's; RemoveVerifyEntry is |
| 872 | // dbverify.RemoveHost; WriteGen is hostbox.WriteGenerationMarker, the |
| 873 | // one <dbPath>.gen writer (ruling G2). |
| 874 | Guard hostguard.Guard |
| 875 | Local park.Local |
| 876 | ParkHomeRegion string |
| 877 | ParkLeases park.Leases |
| 878 | ParkView func(context.Context, hostclaim.Observed) (park.View, error) |
| 879 | ParkAuthorize func(context.Context, hostclaim.Observed, hostclaim.Obligation) error |
| 880 | // ParkRunHeld is park.Engine.RunHeld (movesRunHeld): it runs a chore |
| 881 | // under a lease the caller holds (M5: the landing's residue park under |
| 882 | // its own-location lease), borrowed through the claim's leases |
| 883 | // (AcquireHeld). ParkLocalAuthorize is the MoveParks local authorizer, |
| 884 | // the owner authorizer for this box's park.Local chores (M9). Nil in |
| 885 | // production until the wake-source gate opens; armMovesParks sets all |
| 886 | // five or none. |
| 887 | ParkRunHeld func(ctx context.Context, e park.Engine, w park.Work, held park.HeldLease) error |
| 888 | ParkLocalAuthorize func(ctx context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error |
| 889 | OwnLease func(ctx context.Context, bucket, location string) (park.HeldLease, error) |
| 890 | RenderHeld func(ctx context.Context, s release.Store, rt registry.Route, op string, pin *provision.MoveInputs, before func(sha256hex string) error) (appconfig.HeldRender, error) |
| 891 | ReplaceTimers func(ctx context.Context, host string, rows []registry.Timer) error |
| 892 | SetReplication func(host, class string) error |
| 893 | SetNextDue func(host string, due int64) error |
| 894 | RemoveVerifyEntry func(ctx context.Context, s release.Store, host string) error |
| 895 | WriteGen func(path, generation string) error |
| 896 | // Probe is the landing's probe wake (Plan 3 Task 6): Task 5's |
| 897 | // activator.Service.ProbeWake. Nil in production, with a standing |
| 898 | // notice, until claim Plan 11's launcher (R10) and Plan 5 Task 9's |
| 899 | // arming exist; the landing then waits at restored. |
| 900 | Probe func(ctx context.Context, cfg activator.ProbeConfig) (activator.ProbeResult, error) |
| 901 | // ProbeLive is the activator's LiveProbe: the awake probe entry for a |
| 902 | // host, its op and admitted session. Nil in production with Probe. |
| 903 | ProbeLive func(host string) (op string, res activator.ProbeResult, ok bool) |
| 904 | // MarkProbeOpen is the activator's OpenProbeEntry: the target's open |
| 905 | // marks the probe's live entry open before it releases the hold, so an |
| 906 | // edge reload in between keeps it (Plan 3 Task 7). Nil in production |
| 907 | // with Probe. |
| 908 | MarkProbeOpen func(host, op string) bool |
| 909 | // ConfirmProbeOpen is the activator's ConfirmProbeOpened: called only |
| 910 | // once the open's release of the hold is proved, ending the probe entry's |
| 911 | // idle protection (final review, ruling 3). Nil in production with Probe. |
| 912 | ConfirmProbeOpen func(host, op string) bool |
| 913 | // landFault and parkFault are test seams: the landing's crash points |
| 914 | // and the park engine's (park.Engine.After). Nil in production. |
| 915 | landFault func(point string) error |
| 916 | parkFault func(step string) error |
| 917 | |
| 918 | // The source's finalization (§7.9 step 3, movesfinal.go). DeleteHeld |
| 919 | // deletes this box's row and its timers rows in one transaction, only |
| 920 | // while op holds the row (registry.DeleteIfHoldWithTimers, ruling |
| 921 | // H5(d)). The finalization also uses Guard, Local, ParkLeases, |
| 922 | // ParkView, ParkAuthorize, Versioned and Route above; ParkLeases nil |
| 923 | // leaves it unarmed in production (sourceFinalArmed). finalFault is |
| 924 | // its test seam, nil in production. |
| 925 | DeleteHeld func(host, op string) (bool, error) |
| 926 | finalFault func(point string) error |
| 927 | |
| 928 | // The rollback on the target (§7.8, movesrollback.go). HandHold hands |
| 929 | // the row's hold from the move to its rollback (handHold, ruling J3). |
| 930 | // LogEvent records a move event (moves.LogEvent, Plan 6 Task 3): in |
| 931 | // production moveLogWriter, the move log's create-only object and, when |
| 932 | // this writer created it, the member event through Events (the agent's |
| 933 | // rollout sink). Nil, the event is a log line. The target's |
| 934 | // half also uses the cut's stop hooks, the landing's park hooks and |
| 935 | // DeleteHeld above; ParkLeases nil leaves it unarmed in production |
| 936 | // (rollbackTargetArmed). rbFault is its test seam, nil in production. |
| 937 | HandHold func(host, status, toOp, fromOp string, validate func() error) (bool, error) |
| 938 | LogEvent func(ctx context.Context, s release.Store, rec provision.Record, ev moves.LogEvent) error |
| 939 | Events activator.EventSink |
| 940 | rbFault func(point string) error |
| 941 | |
| 942 | // Page pages an operator (moves spec §13, K11): PageOperator in |
| 943 | // production (armMovesPass), nil in tests, where nothing pages. The box |
| 944 | // arms call it once per rule through the op journal (movespaging.go): |
| 945 | // where a MoveMarkStuck lands (pageStuck: the rollback's return on the |
| 946 | // target, and the recovery, its dirty return's install and the revert |
| 947 | // on the source), where this box closes a rollback or a revert, and |
| 948 | // where the cut's stop evidence finds a stray process. |
| 949 | Page func(ctx context.Context, level, subject, body string) error |
| 950 | |
| 951 | // The source's recovery after a rollback (§7.8, movesrecover.go). |
| 952 | // ReleaseHeld opens the row: registry.ReleaseHold under the op that |
| 953 | // holds it, touching the stage-requested marker when it landed |
| 954 | // (releaseHeld); the caller holds provision.WithHostHoldLock (J12). The |
| 955 | // recovery also uses HandHold, LogEvent, the landing's hooks (the dirty |
| 956 | // return's install) and Timers above; ParkLeases nil leaves it unarmed |
| 957 | // in production (recoverySourceArmed). recoveryProbe is the recovery's |
| 958 | // probe (Task 9, rulings J2 and J18; claim Plan 10 J2): in production |
| 959 | // armRecoveryProbe binds the activator's Group D entry when host claims |
| 960 | // are enabled. Missing runtime wiring waits before recording inputs. |
| 961 | // rcFault is the recovery's test seam, nil in production. |
| 962 | ReleaseHeld func(host, op string) (bool, error) |
| 963 | recoveryProbe func(ctx context.Context, s release.Store, rec provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) |
| 964 | rcFault func(point string) error |
| 965 | |
| 966 | // The revert on the source (§7.4 reverting, movesrevert.go). |
| 967 | // RestatusHold changes the status of a hold the op already has, moving |
| 968 | // to restoring under the move id: registry.SetHoldIf expecting the op |
| 969 | // itself, under the host's hold lock, after the caller's validation, |
| 970 | // touching the stage-requested marker when it landed (restatusHold). The |
| 971 | // revert also uses the cut's hooks, ReleaseHeld, Timers, FilesDir, |
| 972 | // LogEvent and recoveryProbe above; recoveryProbe nil leaves it unarmed in |
| 973 | // production (revertSourceArmed), and the target region's lane park needs |
| 974 | // ParkLeases (revertLaneArmed). rvFault is its test seam, nil in |
| 975 | // production. |
| 976 | RestatusHold func(host, status, op string, validate func() error) (bool, error) |
| 977 | rvFault func(point string) error |
| 978 | |
| 979 | // The abort (§7.11, movesabort.go): the source uses the staging lease, |
| 980 | // Versioned and its own bucket above; the target region's lane park is |
| 981 | // laneParkTarget's, armed as the revert's (revertLaneArmed). abFault is |
| 982 | // its test seam, nil in production. |
| 983 | abFault func(point string) error |
| 984 | |
| 985 | // The offline operator move (movesoffline.go; docs/superpowers/plans/ |
| 986 | // 2026-10-20-offline-operator-move.md, Tasks 3 to 5). armOffline sets |
| 987 | // every field below whatever the Phase 1 switch says, and so Box, |
| 988 | // OptDir, SecretsKey, RenderHeld and Now above; tests set them by hand. |
| 989 | // The sealer (Task 4) and the target (Task 5) read only these, plus the |
| 990 | // hibernate coordinates, DataDir, Region, Mirror, Quiesced, Route, |
| 991 | // Checkpoint and CarryRequest above. |
| 992 | // |
| 993 | // OfflineCapable is offlineCapable for this box: false, and every |
| 994 | // offline step logs one notice per host and does nothing. Offline is the |
| 995 | // keep view reconcile reads (offlineView.keep); the hold needs none. |
| 996 | OfflineCapable bool |
| 997 | Offline *offlineView |
| 998 | // FenceAs installs, or meets again, an owned activator fence |
| 999 | // (activator.Service.FenceAndStopAs under a 2 s deadline, |
| 1000 | // ErrStopUnproven counted as installed: fenceAsWith); LiftFenceAs lifts |
| 1001 | // exactly that fence. Nil (no activator): never fenced. |
| 1002 | FenceAs func(ctx context.Context, host string, generation uint64, owner string) error |
| 1003 | LiftFenceAs func(host string, generation uint64, owner string) bool |
| 1004 | // ReleaseHold opens a row held by op (registry.ReleaseHold, touching the |
| 1005 | // stage-requested marker when it landed); DeleteIfHold deletes a row |
| 1006 | // held by op, with its timers rows, in one transaction |
| 1007 | // (registry.DeleteIfHoldWithTimers). Both answer false when op does not |
| 1008 | // hold the row. The caller holds provision.WithHostHoldLock. |
| 1009 | ReleaseHold func(host, op string) (bool, error) |
| 1010 | DeleteIfHold func(host, op string) (bool, error) |
| 1011 | // Registry is this box's registry: TimersFor, ReplaceTimers, |
| 1012 | // SetNextDue, SetReplication and Get. |
| 1013 | Registry *registry.Registry |
| 1014 | // LaunchLayers is the exec backend's composed launch environment for a |
| 1015 | // row (activator.ExecBackend.LaunchLayers). |
| 1016 | LaunchLayers func(rt registry.Route) (env map[string]string, instance map[string]bool, shadowed []string, err error) |
| 1017 | // ConfigPriv and ConfigPub are the deployment's config keypair, parsed |
| 1018 | // from CARLOS_SECRETS_KEY: ConfigPub seals the files archive, ConfigPriv |
| 1019 | // opens it and renders the held config. Nil without the key. |
| 1020 | ConfigPriv *ecdh.PrivateKey |
| 1021 | ConfigPub *ecdh.PublicKey |
| 1022 | // LiveVersion is the build a row's live link names on this box |
| 1023 | // (release.LiveVersion over OptDir). ChannelVersion is the build the |
| 1024 | // row's channel points at, as adoption resolves it (channelVersion, |
| 1025 | // verified under CARLOS_RELEASE_PUBKEY; an error when that key does |
| 1026 | // not parse). |
| 1027 | LiveVersion func(host string) (string, error) |
| 1028 | ChannelVersion func(ctx context.Context, s release.Store, rt registry.Route) (string, error) |
| 1029 | // Procs counts the processes left for a host in its exec cgroup and its |
| 1030 | // nested slice (cgroup.HostProcs). Nil, or an error: unknown, which a |
| 1031 | // source report records as -1, not stopped. |
| 1032 | Procs func(host string) (int, error) |
| 1033 | } |
| 1034 | |
| 1035 | // defaultMoveReadyDelta is the readiness bar when |
| 1036 | // CARLOS_MOVE_READY_DELTA_BYTES is unset or rejected. A busy instance's |
| 1037 | // per-round churn of new L0 files sits well under 16 MiB, so it reaches |
| 1038 | // ready; a round in which compaction rewrote a large level file (138 MB |
| 1039 | // for the 293 MB database in #175) does not, and waits for a quieter one. |
| 1040 | // Ready keeps mirroring every pass until the cut, so the bar roughly |
| 1041 | // bounds what the cut's final round, run while the instance is down, has |
| 1042 | // left to copy. |
| 1043 | const defaultMoveReadyDelta int64 = 16 << 20 |
| 1044 | |
| 1045 | // moveReadyDeltaEnv overrides defaultMoveReadyDelta with a whole number of |
| 1046 | // bytes. Zero is accepted and means a round must move zero bytes. |
| 1047 | const moveReadyDeltaEnv = "CARLOS_MOVE_READY_DELTA_BYTES" |
| 1048 | |
| 1049 | // moveReadyDeltaFromEnv reads the readiness bar. An unset, unparsable or |
| 1050 | // negative value keeps the default, so an operator's typo never stops the |
| 1051 | // agent; ok is false only when a value was SET and rejected, so the caller |
| 1052 | // can say so instead of leaving the operator to believe it took. |
| 1053 | func moveReadyDeltaFromEnv(getenv func(string) string) (delta int64, ok bool) { |
| 1054 | v := strings.TrimSpace(getenv(moveReadyDeltaEnv)) |
| 1055 | if v == "" { |
| 1056 | return defaultMoveReadyDelta, true |
| 1057 | } |
| 1058 | n, err := strconv.ParseInt(v, 10, 64) |
| 1059 | if err != nil || n < 0 { |
| 1060 | return defaultMoveReadyDelta, false |
| 1061 | } |
| 1062 | return n, true |
| 1063 | } |
| 1064 | |
| 1065 | // movesPass runs one round over every in-flight move this box owns. |
| 1066 | func movesPass(ctx context.Context, s release.Store, w *movesWorker) error { |
| 1067 | w.pass.Lock() |
| 1068 | defer w.pass.Unlock() |
| 1069 | // With the change index armed, the record view (refreshed here inside |
| 1070 | // w.pass; lock order w.pass -> RecordView.mu, a leaf) decides only |
| 1071 | // whether this pass has any move work at all. Every move step has side |
| 1072 | // effects in both directions, and a snapshot can be older than a walk a |
| 1073 | // pass made while the view could not vouch (a bump that failed), so a |
| 1074 | // pass with work acts only on its own walk of the bucket, below, as |
| 1075 | // before the index (S3 change-index design §4.3, final review R1 |
| 1076 | // finding 1). A pass with none skips the walk. |
| 1077 | idle := w.movesIdle(ctx, s) |
| 1078 | cfg := w.cfg |
| 1079 | if cfg.claimRecords { |
| 1080 | ctx = provision.WithRecordSequence(ctx) |
| 1081 | } |
| 1082 | if cfg.Enabled { |
| 1083 | // The pages the outbox and the index keep undelivered are sent again |
| 1084 | // first, before any record or fleet read (movespaging.go, fix round |
| 1085 | // 3). The move log's kept entries are retried next, whatever their |
| 1086 | // moves now are (moveslogoutbox.go). With the switch off neither |
| 1087 | // outbox is touched: what they keep waits on disk until it is on |
| 1088 | // again (final review I1). |
| 1089 | w.flushPageOutbox(ctx) |
| 1090 | w.flushLogOutbox(ctx, s) |
| 1091 | } |
| 1092 | if idle { |
| 1093 | return nil |
| 1094 | } |
| 1095 | // Strict returns exactly ListAllRecords' records and errors, plus the |
| 1096 | // hosts whose record did not decode, which the offline keep view keeps. |
| 1097 | w.walkedIdle = false |
| 1098 | recs, unreadable, err := provision.ListAllRecordsStrict(ctx, s) |
| 1099 | if err != nil { |
| 1100 | return fmt.Errorf("moves: list records: %w", err) |
| 1101 | } |
| 1102 | var targets map[string]edgesFile // lazy: most passes carry no move |
| 1103 | var fleet *moveFleet // lazy, Phase 1 only |
| 1104 | loadFleet := func() bool { |
| 1105 | if fleet == nil { |
| 1106 | f, err := loadMoveFleet(ctx, s) |
| 1107 | if err != nil { |
| 1108 | log.Printf("moves: %v. every Phase 1 move waits for the next pass", err) |
| 1109 | return false |
| 1110 | } |
| 1111 | fleet = &f |
| 1112 | } |
| 1113 | return true |
| 1114 | } |
| 1115 | if cfg.Enabled { |
| 1116 | w.lost = map[string]bool{} |
| 1117 | w.setStrict(recs) |
| 1118 | if w.holds != nil { |
| 1119 | w.holds.refresh(ctx, s, recs) |
| 1120 | } |
| 1121 | } |
| 1122 | // The offline keep view, from the list this pass read (movesoffline.go). |
| 1123 | if cfg.Offline != nil { |
| 1124 | cfg.Offline.refresh(recs, unreadable) |
| 1125 | } |
| 1126 | for _, rec := range recs { |
| 1127 | // An offline operator move is routed first, ahead of the Phase 1 |
| 1128 | // skip and the lost-box gate below: its id makes it look like a |
| 1129 | // Phase 1 record, which with the switch off is skipped, and it |
| 1130 | // never reads a host claim (movesoffline.go). |
| 1131 | if rec.Move.Offline() { |
| 1132 | if err := w.offlineRound(ctx, s, rec); err != nil { |
| 1133 | w.logTrouble(rec.Host, err) |
| 1134 | continue |
| 1135 | } |
| 1136 | w.trouble.Delete(rec.Host) |
| 1137 | continue |
| 1138 | } |
| 1139 | // A move closed or replaced by another writer is forgotten by the |
| 1140 | // move log's memory (movesevents.go). With the switch off the pass |
| 1141 | // touches no outbox, index or memory of the move log at all (final |
| 1142 | // re-review 1, I1). |
| 1143 | if cfg.Enabled { |
| 1144 | w.retireMoves(rec) |
| 1145 | } |
| 1146 | // A seed moves only once placed, an import only once committed |
| 1147 | // (seeded-instances spec §5.8): a record that slipped past an old |
| 1148 | // console is refused here, before the first copy, in every round |
| 1149 | // (legacy, Phase 1 source, participant). targetReportPass holds |
| 1150 | // the same rule for the target. Only the move's source and target |
| 1151 | // say so: every box lists every record. |
| 1152 | if rec.Move != nil && rec.Move.State == provision.MovePreparing { |
| 1153 | if b := rec.Seed.MoveBlock(); b != "" { |
| 1154 | if rec.Region == cfg.Region || rec.Move.To == cfg.Region { |
| 1155 | w.notice(rec.Host+"#seed", "moves: %s: %s", rec.Host, provision.SeedMoveSentence(b)) |
| 1156 | } |
| 1157 | continue |
| 1158 | } |
| 1159 | } |
| 1160 | if rec.Move != nil && !rec.Move.Legacy() { |
| 1161 | // A Phase 1 record. With the switch off it is never touched |
| 1162 | // (none exists unless the switch was on somewhere). |
| 1163 | if !cfg.Enabled { |
| 1164 | if rec.Region == cfg.Region { |
| 1165 | w.notice(rec.Host+"#switch", "moves: %s: move %s waits: %s is off on this box", rec.Host, rec.Move.ID, moves.SwitchEnv) |
| 1166 | } |
| 1167 | continue |
| 1168 | } |
| 1169 | // The lost-box gate (Q9, Plan 4 Task 10): a box the host's claim |
| 1170 | // lists in Lost, or whose claim does not read, runs no move step |
| 1171 | // for the host this pass. |
| 1172 | if w.lostHere(ctx, s, rec) { |
| 1173 | continue |
| 1174 | } |
| 1175 | if rec.Move.State == provision.MoveRollingBack && movesParticipantRecord(cfg, rec) == roleSource { |
| 1176 | // The source's recovery after a rollback (movesrecover.go, |
| 1177 | // Plan 4 Task 5): the record's region is the source's again, |
| 1178 | // and this box is the move's source by the post-flip |
| 1179 | // participant rule (J10), whether or not it still has the row. |
| 1180 | // The transitions another writer made (an operator's retry, |
| 1181 | // a declared loss) are logged first: this branch never |
| 1182 | // reaches phase1MoveRound (final review I3). |
| 1183 | w.observeMove(ctx, s, rec) |
| 1184 | if err := w.recoverSource(ctx, s, rec); err != nil { |
| 1185 | w.logTrouble(rec.Host, err) |
| 1186 | continue |
| 1187 | } |
| 1188 | w.trouble.Delete(rec.Host) |
| 1189 | continue |
| 1190 | } |
| 1191 | if rec.Move.State == provision.MoveReverting && movesParticipantRecord(cfg, rec) == roleSource { |
| 1192 | // The revert on the source (movesrevert.go, Plan 4 Task 6): |
| 1193 | // the record never flipped, and this box is the move's |
| 1194 | // source by the same participant rule (J10). Its entry |
| 1195 | // gives up execution through the cut, which reads the |
| 1196 | // fleet; the rest does not need it. An operator's retry is |
| 1197 | // logged first, as for the recovery above (final review I3). |
| 1198 | w.observeMove(ctx, s, rec) |
| 1199 | var f *moveFleet |
| 1200 | if loadFleet() { |
| 1201 | f = fleet |
| 1202 | } |
| 1203 | if err := w.revertSource(ctx, s, rec, f); err != nil { |
| 1204 | w.logTrouble(rec.Host, err) |
| 1205 | continue |
| 1206 | } |
| 1207 | w.trouble.Delete(rec.Host) |
| 1208 | continue |
| 1209 | } |
| 1210 | if rec.Region != cfg.Region || (cfg.OwnsRoute != nil && !cfg.OwnsRoute(rec.Host)) { |
| 1211 | // After the flip the record's region is the target's, and |
| 1212 | // after the source's finalization deletes its row OwnsRoute |
| 1213 | // no longer holds: a participant that still has the row or |
| 1214 | // the op's journal is visited by the post-flip rule (ruling |
| 1215 | // H5(a), movesfinal.go), never by phase1MoveRound. |
| 1216 | if role := movesParticipantRecord(cfg, rec); role != roleNone { |
| 1217 | if err := w.participantRound(ctx, s, rec, role); err != nil { |
| 1218 | w.logTrouble(rec.Host, err) |
| 1219 | continue |
| 1220 | } |
| 1221 | w.trouble.Delete(rec.Host) |
| 1222 | } |
| 1223 | continue |
| 1224 | } |
| 1225 | if !loadFleet() { |
| 1226 | return nil |
| 1227 | } |
| 1228 | if err := phase1MoveRound(ctx, s, w, rec, *fleet); err != nil { |
| 1229 | w.logTrouble(rec.Host, err) |
| 1230 | continue |
| 1231 | } |
| 1232 | w.trouble.Delete(rec.Host) |
| 1233 | continue |
| 1234 | } |
| 1235 | // A legacy record with the switch on: only its abort completes, by |
| 1236 | // the legacy path (F7(a)); anything else is refused (MV-N7). |
| 1237 | if cfg.Enabled && rec.Move != nil && rec.Move.State != provision.MoveAborting { |
| 1238 | if rec.Region == cfg.Region { |
| 1239 | w.notice(rec.Host+"#legacy", "moves: %s: %s", rec.Host, legacyMoveSentence) |
| 1240 | } |
| 1241 | continue |
| 1242 | } |
| 1243 | if rec.Move == nil || rec.Region != cfg.Region { |
| 1244 | continue |
| 1245 | } |
| 1246 | // The rails the console's moveGate also holds (and this side must |
| 1247 | // re-hold — records have other writers): only an activator-managed |
| 1248 | // exec instance's state is this worker's to ship. A pooled record's |
| 1249 | // data lives on the pool's box, not this one (v1 refuses the whole |
| 1250 | // class), and a non-hibernating instance runs OUTSIDE the activator |
| 1251 | // — Quiesced would answer true over a live, writing process. |
| 1252 | if rec.Pool != "" || !rec.HibernateOrDefault() || (rec.Kind != "" && rec.Kind != "instance") { |
| 1253 | continue |
| 1254 | } |
| 1255 | if cfg.OwnsRoute != nil && !cfg.OwnsRoute(rec.Host) { |
| 1256 | continue // another box's child; its owner drives the move |
| 1257 | } |
| 1258 | if targets == nil { |
| 1259 | if targets, err = moveTargets(ctx, s); err != nil { |
| 1260 | log.Printf("moves: %v — every in-flight move waits for the next pass", err) |
| 1261 | return nil |
| 1262 | } |
| 1263 | } |
| 1264 | if err := legacyMoveRound(ctx, s, w, rec, targets); err != nil { |
| 1265 | // One move's trouble must not starve its neighbours — and a |
| 1266 | // PERSISTENT one (an unapplied staging grant's AccessDenied, |
| 1267 | // review round 1) must not flood the journal at the fast |
| 1268 | // converge loop's cadence: once a minute per host is plenty. |
| 1269 | w.logTrouble(rec.Host, err) |
| 1270 | continue |
| 1271 | } |
| 1272 | w.trouble.Delete(rec.Host) |
| 1273 | } |
| 1274 | switch { |
| 1275 | case !cfg.Enabled: |
| 1276 | case hasPhase1Into(recs, cfg.Region): |
| 1277 | if loadFleet() { |
| 1278 | w.targetReportPass(ctx, s, recs, *fleet) |
| 1279 | } |
| 1280 | default: |
| 1281 | w.evictArchives(nil) // no move into this box: no verdict to keep |
| 1282 | w.retireHeldRefusals(nil) |
| 1283 | } |
| 1284 | if cfg.Enabled { |
| 1285 | // The source's op journals that reached the stamp, whoever's record |
| 1286 | // no longer routes here (movesfinal.go, fix round 1). |
| 1287 | w.sweepFinalJournals(ctx, s, recs) |
| 1288 | } |
| 1289 | if cfg.Enabled { |
| 1290 | // Delivered pages of moves retired on this box are removed, now that |
| 1291 | // every record is listed (movespaging.go). With the switch off the |
| 1292 | // outbox is left as it is (final review I1). |
| 1293 | w.retirePageTombstones(recs) |
| 1294 | } |
| 1295 | w.liftFences(recs) |
| 1296 | w.walkedIdle = len(unreadable) == 0 && !slices.ContainsFunc(recs, func(r provision.Record) bool { return r.Move != nil }) && |
| 1297 | !w.localMoveState() |
| 1298 | return nil |
| 1299 | } |
| 1300 | |
| 1301 | // movesIdle reports whether this pass may skip its walk: the change index is |
| 1302 | // armed and the record view can vouch, the last pass walked and found |
| 1303 | // nothing (walkedIdle), the view's snapshot has no record with a move and |
| 1304 | // none it cannot decode, and this worker holds no move state of its own. Any |
| 1305 | // doubt is work: the pass walks the bucket. A move started by a write whose |
| 1306 | // bump failed waits for the backstop rebuild (or for any pass that walks). |
| 1307 | func (w *movesWorker) movesIdle(ctx context.Context, s release.Store) bool { |
| 1308 | vs, _, ok := viewStore(ctx, s) |
| 1309 | if !ok || !w.walkedIdle { |
| 1310 | return false |
| 1311 | } |
| 1312 | recs, unreadable, err := provision.ListAllRecordsStrict(ctx, vs) |
| 1313 | if err != nil || len(unreadable) > 0 || slices.ContainsFunc(recs, func(r provision.Record) bool { return r.Move != nil }) { |
| 1314 | return false |
| 1315 | } |
| 1316 | return !w.localMoveState() |
| 1317 | } |
| 1318 | |
| 1319 | // localMoveState reports whether this worker holds anything a moves pass |
| 1320 | // acts on or cleans up from its record walk, whatever the records say: a |
| 1321 | // cut's wake fence, a strict spawn set, the hold or offline keep view |
| 1322 | // unbuilt or naming a host, T5's archive verdicts, a landing's lease, or, |
| 1323 | // on disk, an op journal or a page outbox entry. Under pass. |
| 1324 | func (w *movesWorker) localMoveState() bool { |
| 1325 | fenced := false |
| 1326 | w.fenced.Range(func(any, any) bool { fenced = true; return false }) |
| 1327 | if fenced { |
| 1328 | return true |
| 1329 | } |
| 1330 | if m := w.strict.Load(); m != nil && len(*m) > 0 { |
| 1331 | return true |
| 1332 | } |
| 1333 | if w.holds != nil { |
| 1334 | if snap := w.holds.snap.Load(); snap == nil || len(snap.claims) > 0 || len(snap.unknown) > 0 { |
| 1335 | return true |
| 1336 | } |
| 1337 | } |
| 1338 | if v := w.cfg.Offline; v != nil { |
| 1339 | if snap := v.snap.Load(); snap == nil || snap.failed || len(snap.moves) > 0 || len(snap.unreadable) > 0 { |
| 1340 | return true |
| 1341 | } |
| 1342 | } |
| 1343 | w.mu.Lock() |
| 1344 | held := len(w.archives) > 0 || len(w.landLeases) > 0 |
| 1345 | w.mu.Unlock() |
| 1346 | if held { |
| 1347 | return true |
| 1348 | } |
| 1349 | return movesDirHolds(w.cfg.DataDir) |
| 1350 | } |
| 1351 | |
| 1352 | // movesDirHolds reports whether <dataDir>/.moves holds anything below its |
| 1353 | // top level: an op journal (.moves/<host>/<op>/, parked journals leave it) |
| 1354 | // or a page outbox entry (.moves/.page-outbox/...). An unreadable directory |
| 1355 | // holds something, as far as skipping a walk goes. |
| 1356 | func movesDirHolds(dataDir string) bool { |
| 1357 | if dataDir == "" { |
| 1358 | return false |
| 1359 | } |
| 1360 | root := filepath.Join(dataDir, ".moves") |
| 1361 | tops, err := os.ReadDir(root) |
| 1362 | if errors.Is(err, os.ErrNotExist) { |
| 1363 | return false |
| 1364 | } |
| 1365 | if err != nil { |
| 1366 | return true |
| 1367 | } |
| 1368 | for _, top := range tops { |
| 1369 | if !top.IsDir() { |
| 1370 | continue |
| 1371 | } |
| 1372 | inner, err := os.ReadDir(filepath.Join(root, top.Name())) |
| 1373 | if err != nil || len(inner) > 0 { |
| 1374 | return true |
| 1375 | } |
| 1376 | } |
| 1377 | return false |
| 1378 | } |
| 1379 | |
| 1380 | // legacyMoveSentence is MV-N7, logged (never shown to a user) when the |
| 1381 | // switch is on and a record carries a move with no id. |
| 1382 | const legacyMoveSentence = "this move was started by an older console and has no id. abort it and request it again" |
| 1383 | |
| 1384 | // hasPhase1Into reports whether any Phase 1 move targets region. |
| 1385 | func hasPhase1Into(recs []provision.Record, region string) bool { |
| 1386 | for _, rec := range recs { |
| 1387 | if rec.Move != nil && !rec.Move.Legacy() && !rec.Move.Offline() && rec.Move.To == region { |
| 1388 | return true |
| 1389 | } |
| 1390 | } |
| 1391 | return false |
| 1392 | } |
| 1393 | |
| 1394 | // moveFenceGeneration is the cut fence's generation: the move's request |
| 1395 | // time, so every pass of one cut installs the same fence (FenceAndStop is |
| 1396 | // idempotent at an equal generation) and a later move of the same host |
| 1397 | // fences above an earlier one. |
| 1398 | func moveFenceGeneration(m *provision.MoveState) uint64 { |
| 1399 | if n := m.RequestedAt.UnixNano(); n > 0 { |
| 1400 | return uint64(n) |
| 1401 | } |
| 1402 | return 1 |
| 1403 | } |
| 1404 | |
| 1405 | // liftFences lifts each cut fence once nothing on this box needs it: the |
| 1406 | // host's route is gone from this box's registry (the flip retired it), or |
| 1407 | // the record is at rest here again (a move back landed). Before either, |
| 1408 | // the fence holds, including after the flip while the "moving" row waits |
| 1409 | // for reconcile to retire it. The fences live in the activator's memory, so |
| 1410 | // an agent restart drops them all; ServeScheduled's status gate covers that |
| 1411 | // window, and a cut still in flight re-fences on its next pass. |
| 1412 | func (w *movesWorker) liftFences(recs []provision.Record) { |
| 1413 | if w.cfg.LiftFence == nil { |
| 1414 | return |
| 1415 | } |
| 1416 | atRest := map[string]bool{} |
| 1417 | for _, rec := range recs { |
| 1418 | if rec.Move == nil && rec.Region == w.cfg.Region { |
| 1419 | atRest[rec.Host] = true |
| 1420 | } |
| 1421 | } |
| 1422 | w.fenced.Range(func(k, v any) bool { |
| 1423 | host, gen := k.(string), v.(uint64) |
| 1424 | held := true |
| 1425 | if w.cfg.Route != nil { |
| 1426 | rt, err := w.cfg.Route(host) |
| 1427 | if err != nil { |
| 1428 | return true // an unreadable registry is not a retired route |
| 1429 | } |
| 1430 | held = rt != nil |
| 1431 | } |
| 1432 | if held && !atRest[host] { |
| 1433 | return true |
| 1434 | } |
| 1435 | w.cfg.LiftFence(host, gen) |
| 1436 | w.fenced.CompareAndDelete(host, gen) |
| 1437 | log.Printf("moves: %s: wake fence lifted", host) |
| 1438 | return true |
| 1439 | }) |
| 1440 | } |
| 1441 | |
| 1442 | // moveTargets folds the fleet's edges files into placement → coordinates. |
| 1443 | // A duplicate placement code keeps the lexically-first box (deterministic |
| 1444 | // everywhere, same rule as foldHomes) — and a target with no data bucket is |
| 1445 | // simply absent, which legacyMoveRound reports per-move. |
| 1446 | func moveTargets(ctx context.Context, s release.Store) (map[string]edgesFile, error) { |
| 1447 | bodies, err := loadEdgesFileBodies(ctx, s) |
| 1448 | if err != nil { |
| 1449 | return nil, err |
| 1450 | } |
| 1451 | out := map[string]edgesFile{} |
| 1452 | for _, body := range bodies { |
| 1453 | var t edgesFile // the writer's own shape — no parallel decl to drift |
| 1454 | if err := json.Unmarshal(body, &t); err != nil || t.V < 1 || t.Placement == "" || t.DataBucket == "" { |
| 1455 | continue |
| 1456 | } |
| 1457 | if prev, dup := out[t.Placement]; dup && prev.Box <= t.Box { |
| 1458 | continue |
| 1459 | } |
| 1460 | out[t.Placement] = t |
| 1461 | } |
| 1462 | return out, nil |
| 1463 | } |
| 1464 | |
| 1465 | // legacyMoveRound is the worker's round for a legacy record (no move id), |
| 1466 | // main's runMoveRound unchanged. It is the only round with the Phase 1 |
| 1467 | // switch off; with it on, it completes a legacy aborting record and nothing |
| 1468 | // else (F7(a)). |
| 1469 | func legacyMoveRound(ctx context.Context, s release.Store, w *movesWorker, rec provision.Record, targets map[string]edgesFile) error { |
| 1470 | cfg := w.cfg |
| 1471 | tgt, ok := targets[rec.Move.To] |
| 1472 | if !ok { |
| 1473 | return fmt.Errorf("target region %s publishes no data bucket yet — waiting", rec.Move.To) |
| 1474 | } |
| 1475 | // The mirror pair is built lazily and memoized per target region: the |
| 1476 | // cut's not-yet-quiesced early return must not pay a credential-chain |
| 1477 | // resolution it never uses. |
| 1478 | mirror := func() (lease.MirrorStore, error) { return w.mirror(ctx, tgt.DataBucketRegion) } |
| 1479 | srcPrefix := activator.HibernatePrefix(cfg.OwnPrefix, rec.Host) |
| 1480 | dstPrefix := activator.MovesPrefix(rec.Host) |
| 1481 | // The copy receipts (#175) live in this box's OWN bucket: the staging |
| 1482 | // grant on the target is push-only, so receipts there could be |
| 1483 | // written but never read back. |
| 1484 | receipts := lease.Receipts{Bucket: cfg.OwnBucket, Key: activator.MoveReceiptsKey(rec.Host)} |
| 1485 | |
| 1486 | switch rec.Move.State { |
| 1487 | case provision.MovePreparing, provision.MoveReady: |
| 1488 | m, err := mirror() |
| 1489 | if err != nil { |
| 1490 | return err |
| 1491 | } |
| 1492 | round, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix, receipts) |
| 1493 | if err != nil { |
| 1494 | return err |
| 1495 | } |
| 1496 | // Ready means the gap is small, not that it is closed (#175): an |
| 1497 | // awake instance writes new litestream files between any two |
| 1498 | // listings, so a round that copied NOTHING never comes. The cut |
| 1499 | // closes the gap itself (quiesce, fence, final checkpoint, a final |
| 1500 | // round), so this bar only bounds how much that last round, and |
| 1501 | // the downtime around it, has left to copy. |
| 1502 | if rec.Move.State == provision.MovePreparing && round.DeltaBytes() <= cfg.ReadyDelta { |
| 1503 | if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MovePromoteReady(cfg.Now())); err != nil && !errors.Is(err, provision.ErrMoveConflict) { |
| 1504 | return err |
| 1505 | } |
| 1506 | nudgeSerial(ctx, s) |
| 1507 | log.Printf("moves: %s: staged replica caught up in %s/%s (last round copied %d bytes, deferred %d, in %d changes; bar %d) — ready to cut", rec.Host, tgt.DataBucket, dstPrefix, round.BytesCopied, round.BytesDeferred, round.Changed, cfg.ReadyDelta) |
| 1508 | } |
| 1509 | return nil |
| 1510 | |
| 1511 | case provision.MoveAborting: |
| 1512 | // Abort's box half: delete the staged copy, THEN clear the move — |
| 1513 | // a lingering staging would win a later landing on the target |
| 1514 | // (staging-wins, activator.promoteStagedReplica). |
| 1515 | m, err := mirror() |
| 1516 | if err != nil { |
| 1517 | return err |
| 1518 | } |
| 1519 | if err := lease.DeletePrefix(ctx, m, tgt.DataBucket, dstPrefix); err != nil { |
| 1520 | return err |
| 1521 | } |
| 1522 | if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveAbortComplete()); err != nil && !errors.Is(err, provision.ErrMoveConflict) { |
| 1523 | return err |
| 1524 | } |
| 1525 | clearMoveReceipts(ctx, m, receipts, rec.Host) |
| 1526 | nudgeSerial(ctx, s) |
| 1527 | log.Printf("moves: %s: abort complete — staged copy deleted, instance stays in %q", rec.Host, rec.Region) |
| 1528 | return nil |
| 1529 | |
| 1530 | case provision.MoveCutting: |
| 1531 | // Step 1+2: quiesce. SetStatus is idempotent; a route already |
| 1532 | // "moving" (a resumed cut) just re-writes the word. |
| 1533 | if err := cfg.SetStatus(rec.Host, "moving"); err != nil { |
| 1534 | return fmt.Errorf("set route status: %w", err) |
| 1535 | } |
| 1536 | // Fence before the quiesce check, so the answer Quiesced gives stays |
| 1537 | // true: without it a tick could wake the host between this pass's |
| 1538 | // check and the purge (G12). Every pass re-installs the same fence, |
| 1539 | // which is also how a restarted agent gets it back. |
| 1540 | if cfg.Fence != nil { |
| 1541 | gen := moveFenceGeneration(rec.Move) |
| 1542 | if err := cfg.Fence(ctx, rec.Host, gen); err != nil { |
| 1543 | return fmt.Errorf("fence against wakes: %w", err) |
| 1544 | } |
| 1545 | w.fenced.Store(rec.Host, gen) |
| 1546 | } |
| 1547 | if !cfg.Quiesced(rec.Host) { |
| 1548 | if _, done := w.nudged.LoadOrStore(rec.Host, true); !done { |
| 1549 | nudgeSerial(ctx, s) |
| 1550 | } |
| 1551 | return nil // the retire is in flight; the nudged pass re-enters here |
| 1552 | } |
| 1553 | // A host whose layout migration is in flight (isolated-exec design |
| 1554 | // §8.2) is fenced here exactly as Classify fences it for a wake: its |
| 1555 | // DB may be mid-copy between the flat and nested paths, and neither |
| 1556 | // location is safe to checkpoint or purge until the journal clears. |
| 1557 | // Retried next pass, like every other not-yet-ready state above. |
| 1558 | // JournalPresent Lstats rather than Stats, so a dangling symlink at |
| 1559 | // the journal path still fences: the pass, the activator, SyncExec |
| 1560 | // and litestream all share this one predicate for "held", and a |
| 1561 | // journal this box cannot even read is not proof there isn't one. |
| 1562 | if layout.JournalPresent(cfg.DataDir, rec.Host) { |
| 1563 | log.Printf("moves: %s: layout migration in flight; leaving the move for the next pass", rec.Host) |
| 1564 | return nil |
| 1565 | } |
| 1566 | // Step 3: prove the own replica carries the tail. The activator's |
| 1567 | // stop SWALLOWS a failed final checkpoint (safe pre-move: the warm |
| 1568 | // cache stood behind it; fatal here: step 5 deletes that cache). |
| 1569 | // Re-checkpointing an already-synced DB is cheap and idempotent; a |
| 1570 | // held lease (the child not fully released) or an S3 blip returns |
| 1571 | // an error and the pass retries. |
| 1572 | // |
| 1573 | // dbPath defaults to the legacy flat guess and is overridden by the |
| 1574 | // row's own DBPath when Route is wired (isolated-exec design §9): a |
| 1575 | // nested row's DB never lived at the flat guess at all, so |
| 1576 | // checkpointing and purging (step 5, below) must follow the row. |
| 1577 | dbPath := filepath.Join(cfg.DataDir, rec.Host+".db") |
| 1578 | if cfg.Route != nil { |
| 1579 | if rt, err := cfg.Route(rec.Host); err == nil && rt != nil && rt.DBPath != "" { |
| 1580 | dbPath = rt.DBPath |
| 1581 | } |
| 1582 | } |
| 1583 | if cfg.Checkpoint != nil { |
| 1584 | if _, statErr := os.Stat(dbPath); statErr == nil { |
| 1585 | if err := cfg.Checkpoint(ctx, dbPath); err != nil { |
| 1586 | return fmt.Errorf("final checkpoint: %w", err) |
| 1587 | } |
| 1588 | } |
| 1589 | } |
| 1590 | // Step 3b: this box's request record into the build state, so the |
| 1591 | // final round carries a request it closed as failed (ruling |
| 1592 | // R-SPEC-43A-2). The instance is quiesced and fenced: no wake writes |
| 1593 | // the state now. |
| 1594 | if cfg.CarryRequest != nil { |
| 1595 | if err := cfg.CarryRequest(ctx, rec.Host, cfg.OwnBucket, srcPrefix); err != nil { |
| 1596 | return fmt.Errorf("carry the request record into the build state: %w", err) |
| 1597 | } |
| 1598 | } |
| 1599 | // Step 4: the byte-final mirror round. |
| 1600 | m, err := mirror() |
| 1601 | if err != nil { |
| 1602 | return err |
| 1603 | } |
| 1604 | round, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix, receipts) |
| 1605 | if err != nil { |
| 1606 | return err |
| 1607 | } |
| 1608 | // A deferred object is one the source changed under this round, so |
| 1609 | // staging is NOT byte-final. Nothing should write a quiesced, |
| 1610 | // fenced, checkpointed source, but the purge below is |
| 1611 | // irreversible, so it waits for a round that proves it. |
| 1612 | if round.Deferred > 0 { |
| 1613 | return fmt.Errorf("final mirror round: %d object(s) changed at the source during the copy; retrying the cut", round.Deferred) |
| 1614 | } |
| 1615 | // Step 5: the warm-cache purge. |
| 1616 | activator.PurgeLocalDBArtifacts(dbPath, true) |
| 1617 | // Step 6: the last act — Region flips, Move clears, one write. |
| 1618 | if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveComplete()); err != nil && !errors.Is(err, provision.ErrMoveConflict) { |
| 1619 | return err |
| 1620 | } |
| 1621 | clearMoveReceipts(ctx, m, receipts, rec.Host) |
| 1622 | w.nudged.Delete(rec.Host) |
| 1623 | // The flip's nudge is what turns the TARGET's mint from a |
| 1624 | // 5-minute wait into a seconds-cadence converge. |
| 1625 | nudgeSerial(ctx, s) |
| 1626 | log.Printf("moves: %s: cut complete — now homed in %s (route retires on the next reconcile pass)", rec.Host, rec.Move.To) |
| 1627 | return nil |
| 1628 | } |
| 1629 | return nil |
| 1630 | } |
| 1631 | |
| 1632 | // clearMoveReceipts deletes a finished or abandoned move's copy receipts. |
| 1633 | // Best-effort, after the record write that ended the move: receipts are |
| 1634 | // self-checking (a stale one can only fail to match, never vouch for the |
| 1635 | // wrong object), and some deployments' store policy denies DeleteObject on |
| 1636 | // the data bucket, so a failure costs one small object, never a move. |
| 1637 | func clearMoveReceipts(ctx context.Context, m lease.MirrorStore, rc lease.Receipts, host string) { |
| 1638 | if err := m.Delete(ctx, rc.Bucket, rc.Key); err != nil { |
| 1639 | log.Printf("moves: %s: could not delete copy receipts %s/%s: %v (harmless; they no longer match anything)", host, rc.Bucket, rc.Key, err) |
| 1640 | } |
| 1641 | } |
| 1642 | |
| 1643 | // nudgeSerial bumps control/serial after a box-side move write, so the |
| 1644 | // NEXT actor — this box's own cut continuation, the target's mint, the |
| 1645 | // tenant's poll — reacts on the fast-converge path instead of the 5-minute |
| 1646 | // floor. The live rehearsal measured that floor three times over in one |
| 1647 | // cut (status→quiesce→flip 331s, target mint+wake 700s); with the nudge, |
| 1648 | // each hop is a serial-triggered pass. Best-effort like the console's own |
| 1649 | // bumpSerial: a failed bump costs cadence, never correctness. |
| 1650 | func nudgeSerial(ctx context.Context, s release.Store) { |
| 1651 | if err := release.BumpSerial(ctx, s, time.Now()); err != nil { |
| 1652 | log.Printf("moves: serial nudge: %v (falling back to the timer floor)", err) |
| 1653 | } |
| 1654 | } |
| 1655 | |
| 1656 | // movingStatusNudged remembers which hosts this process already nudged for |
| 1657 | // the status→quiesce hop, so a child that refuses to die does not turn the |
| 1658 | // fast-converge loop into a 2s self-triggering cycle: ONE nudge per host |
| 1659 | // per process, then the 5-minute floor stands under a wedged stop. |
| 1660 | var movingStatusNudged sync.Map |
| 1661 | |
| 1662 | // logTrouble rate-limits per-host move errors to one line a minute; the |
| 1663 | // pass clears a host's entry on its next clean round. |
| 1664 | func (w *movesWorker) logTrouble(host string, err error) { |
| 1665 | now := time.Now() |
| 1666 | if last, ok := w.trouble.Load(host); ok && now.Sub(last.(time.Time)) < time.Minute { |
| 1667 | return |
| 1668 | } |
| 1669 | w.trouble.Store(host, now) |
| 1670 | log.Printf("moves: %s: %v", host, err) |
| 1671 | } |
| 1672 | |
amadan