package host import ( "context" "encoding/json" "errors" "fmt" "log" "os" "path" "path/filepath" "sort" "strings" "sync" "sync/atomic" "syscall" "time" "github.com/carlosframework/platform/internal/activator" "github.com/carlosframework/platform/internal/hostwrite" "github.com/carlosframework/platform/internal/instancectl" "github.com/carlosframework/platform/internal/layout" "github.com/carlosframework/platform/internal/lease" "github.com/carlosframework/platform/internal/moves" "github.com/carlosframework/platform/internal/park" "github.com/carlosframework/platform/internal/provision" "github.com/carlosframework/platform/internal/registry" "github.com/carlosframework/platform/internal/release" ) // The offline operator move on a box (docs/superpowers/plans/2026-10-20- // offline-operator-move.md, Task 3): the hold reconcile asks, the keep view, // the capability, the moves pass's dispatch, and the source boxes' steps. // The sealer's step is movesoffline_seal.go (Task 4) and the target's is // movesoffline_land.go (Task 5). // // The stop has three layers. This file holds the first two on every // source-region box: reconcile holds the row from the record alone // (offlineHold), and each pass fences the activator and writes the box's // source report. The third, the tombstone lease, is the sealer's. // // Dormant, with no offline record anywhere, nothing here acts: the hold // answers false for every record, keep answers false once the startup // listing is read, the dispatch matches no record, and the edges file // carries one more capability string. // capOfflineMove1 is the edges-file capability a box publishes when it can // take part in an offline move (offlineCapable). The console's start door // refuses a region holding a box that does not publish it. const capOfflineMove1 = "offline-move-1" // offlineCapable is the box's own witness that it can take part in an // offline move: hibernation is configured and neither CARLOS_HOST_CLAIM nor // CARLOS_MOVES_PHASE1 is set to anything at all. The guard mode is not read // (platform-1e's ruling of 2026-10-20, the plan's Global Constraints): the // target's build stages through ordinary adoption, which root's own floor // check already covers on a guarded box. A box that is not capable publishes // no capability and runs no offline step. func offlineCapable(getenv func(string) string, hibernation bool) bool { if !hibernation || hostwrite.Enabled(getenv) { return false } return strings.TrimSpace(getenv(moves.SwitchEnv)) == "" && !moves.Enabled(getenv) } // hibernationConfigured is the edges file's HibernationConfigured, read the // same way, so the capability and the field beside it never disagree. func hibernationConfigured(getenv func(string) string) bool { return strings.TrimSpace(getenv("CARLOS_HIBERNATE_BUCKET")) != "" } // offlineFenceOwner is the activator fence owner of one offline move. func offlineFenceOwner(id string) string { return "offline:" + id } // offlineHold is reconcile's Hold for offline moves: a pure function of the // record it is handed and this box's placement, never of a snapshot (Fable's // B1). A target box that restarts after the flip reconciles before its first // moves pass, and must mint the landing row held even then. // // - A source-region box (placement equals move.from) holds the host's row // "moving" under the move id from offline-stopping through // offline-sealed. After the flip the record names the target, the row // is out of this box's scope and offlineView.keep keeps it, held as it // is, until offline-finishing. // - A target-region box (placement equals move.to) holds it "landing" // while the record names the target and the move is offline-landing or // offline-landed. offline-open is the target's to release (Task 5). // // Placements are compared as movesNamesBox compares them, by // park.PlacementMatches under the verified home binding, so the home's "" // and its roster name are one. func offlineHold(box moveCfg) func(rec provision.Record) (status, op string, ok bool) { home, here := box.ParkHomeRegion, box.parkRegion() return func(rec provision.Record) (string, string, bool) { m := rec.Move if !m.Offline() || m.ID == "" || here == "" { return "", "", false } switch m.State { case provision.OfflineStopping, provision.OfflineSealing, provision.OfflineSealed: if park.PlacementMatches(home, m.From, here) { return "moving", m.ID, true } case provision.OfflineLanding, provision.OfflineLanded: if rec.Region == m.To && park.PlacementMatches(home, m.To, here) { return "landing", m.ID, true } } return "", "", false } } // offlineView is what reconcile's KeepParticipant asks about offline moves: // the offline moves in flight, as the last listing read them. Only keep // reads it; the hold never does (offlineHold). // // A source box's view has seen the move before the record flips: the seal // door needs this box's report for the move, and the pass that writes it // refreshes the view first; an agent restart reads the listing again at // start. From then until finishing every listing carries the move. // // It is refreshed from every moves pass's record list and once at agent // start, before the reconcile loops (refreshOfflineView). keep fails closed, // keeping every row, only while no listing has been read (reconcileSerialized- // With reads one first when the view is unbuilt or its last listing failed, // so a failed startup listing heals on the next reconcile), and for a host // whose record does not decode. Once a listing is read, a host with no // offline move is not kept, so removals happen on the first pass as they // always have. type offlineView struct { home, region string // written is reconcile's HoldWritten for the offline hooks: the // stage-requested marker after a hold write (registry.SetHold's // contract, G10). Nil writes nothing. written func(host string) snap atomic.Pointer[offlineSnapshot] logged sync.Map // host -> time.Time: the last unreadable-record line } type offlineSnapshot struct { moves map[string]provision.MoveState // host -> its offline move owners map[string][2]string // host -> the account and app its record lives under failed bool // the listing failed: keep everything // unreadable are the hosts whose record read but did not decode, or // disagreed with its path (provision.ListAllRecordsStrict): their move // cannot be known, so their rows are kept. unreadable map[string]bool } // newOfflineView builds the view for the box cfg describes. Unbuilt until // its first refresh. func newOfflineView(cfg moveCfg, written func(host string)) *offlineView { return &offlineView{home: cfg.ParkHomeRegion, region: cfg.parkRegion(), written: written} } // refresh rebuilds the view from a record list that was read, with the // hosts provision.ListAllRecordsStrict could not decode: those are kept // (fail closed for keep only; the hold never reads the view) and logged at // most once a minute each. func (v *offlineView) refresh(recs []provision.Record, unreadable map[string]bool) { next := &offlineSnapshot{moves: map[string]provision.MoveState{}, owners: map[string][2]string{}, unreadable: map[string]bool{}} for _, rec := range recs { if rec.Move.Offline() { next.moves[rec.Host] = *rec.Move next.owners[rec.Host] = [2]string{rec.Account, rec.App} } } now := time.Now() for host := range unreadable { next.unreadable[host] = true if last, ok := v.logged.Load(host); !ok || now.Sub(last.(time.Time)) >= time.Minute { v.logged.Store(host, now) log.Printf("moves: offline view: %s: the record does not decode; its row is kept until it does", host) } } v.snap.Store(next) } // fail records a listing that could not be read, before any was: keep // answers true for every host until a listing is read. func (v *offlineView) fail() { v.snap.Store(&offlineSnapshot{failed: true}) } // built reports whether any listing, read or failed, has been recorded. func (v *offlineView) built() bool { return v.snap.Load() != nil } // failed reports whether the last listing recorded failed: keep answers true // for every host until one reads. func (v *offlineView) failed() bool { snap := v.snap.Load() return snap != nil && snap.failed } // keep is reconcile's KeepParticipant: the row of a host whose offline move // names this box's region stays, whatever the record's region says. The // source keeps it until offline-finishing (when it is removed, as the plan's // finishing step relies on); the target keeps it until the move is cleared // (the target's revert deletes it itself). An unknown state keeps. func (v *offlineView) keep(host string) bool { snap := v.snap.Load() if snap == nil || snap.failed || snap.unreadable[host] { return true } m, ok := snap.moves[host] if !ok || v.region == "" { return false } if park.PlacementMatches(v.home, m.To, v.region) { return true } if park.PlacementMatches(v.home, m.From, v.region) { return m.State != provision.OfflineFinishing } return false } // elsewhere reports whether host's offline move has its record name a region // other than this box's: the target's from landing to finishing, the source's // while reverting. Such a host is not this box's to touch whatever its row // says (scheduleSkip). A view with no listing yet, and a host with no // offline move, answer false: the row's own hold answers for those. func (v *offlineView) elsewhere(host string) bool { snap := v.snap.Load() if snap == nil || snap.failed || v.region == "" { return false } m, ok := snap.moves[host] if !ok { return false } region := m.From switch m.State { case provision.OfflineLanding, provision.OfflineLanded, provision.OfflineOpen, provision.OfflineFinishing: region = m.To } return !park.PlacementMatches(v.home, region, v.region) } // refreshOfflineView reads the record list once, outside a moves pass: at // agent start, before the reconcile loops, next to refreshMovesHolds. A // failed listing leaves keep failing closed until a listing is read. func refreshOfflineView(ctx context.Context, s release.Store, v *offlineView) { if v == nil || s == nil { return } recs, unreadable, err := provision.ListAllRecordsStrict(ctx, s) if err != nil { log.Printf("moves: offline view: list records: %v; every row is kept until a listing reads", err) v.fail() return } v.refresh(recs, unreadable) } // armedOfflineView is the armed worker's view, nil when no worker is armed // or the box cannot take part in an offline move. func armedOfflineView() *offlineView { w := armedMovesWorker.Load() if w == nil || !w.cfg.OfflineCapable { return nil } return w.cfg.Offline } // armedOfflineCapable is whether this box publishes offline-move-1: the // armed worker is capable and has its keep view, exactly the condition // under which reconcileSerializedWith installs the offline hooks. func armedOfflineCapable() bool { return armedOfflineView() != nil } // withOfflineHolds gives a reconcile pass the offline hooks, only when the // caller and Phase 1's view set none (withMovesHolds runs first): the two // are never armed together, since a box with Phase 1 on is not capable. func withOfflineHolds(cfg provision.ReconcileConfig, hold func(provision.Record) (string, string, bool), v *offlineView) provision.ReconcileConfig { if hold == nil || v == nil || cfg.Hold != nil || cfg.KeepParticipant != nil { return cfg } cfg.Hold, cfg.KeepParticipant = hold, v.keep if cfg.HoldWritten == nil && v.written != nil { cfg.HoldWritten = v.written } return cfg } // newMovesWorker is the worker for cfg: armMovesPass's, and each test box's. func newMovesWorker(cfg moveCfg) *movesWorker { return &movesWorker{cfg: cfg, mirrors: map[string]lease.MirrorStore{}} } // fenceAsWith is moveCfg.FenceAs over an activator: FenceAndStopAs under a // short deadline, so one slow stop does not hold up the pass. // ErrStopUnproven still means the fence is installed, which is all the step // asks of it; Quiesced answers the rest. func fenceAsWith(svc *activator.Service) func(ctx context.Context, host string, generation uint64, owner string) error { return func(ctx context.Context, host string, generation uint64, owner string) error { fctx, cancel := context.WithTimeout(ctx, 2*time.Second) defer cancel() if _, err := svc.FenceAndStopAs(fctx, host, generation, owner); err != nil && !errors.Is(err, activator.ErrStopUnproven) { return err } return nil } } // hostSliceCgroup is the absolute cgroup directory of a nested host's // systemd slice under cgroupRoot (the cgroup v2 mount): every role and // generation unit of the host runs in it. func hostSliceCgroup(cgroupRoot, host string) string { return filepath.Join(cgroupRoot, filepath.FromSlash(path.Dir(instancectl.ClaimUnitControlGroup(host)))) } // offlineRound is the moves pass's round for a record with an offline move, // reached before every other branch of the pass (the Phase 1 skip and the // lost-box gate included). A box that is not capable logs one notice per // host and does nothing. Otherwise it runs the source step when its // placement is move.from, the sealer's when it is move.sealer, and the // target's when its placement is move.to, in that order, so the sealer // reads the report this box has just written. func (w *movesWorker) offlineRound(ctx context.Context, s release.Store, rec provision.Record) error { cfg := w.cfg m := rec.Move if !m.Offline() { return nil } here := cfg.parkRegion() source := here != "" && park.PlacementMatches(cfg.ParkHomeRegion, m.From, here) target := here != "" && park.PlacementMatches(cfg.ParkHomeRegion, m.To, here) sealer := cfg.Box != "" && cfg.Box == m.Sealer if !cfg.OfflineCapable { if source || target || sealer { w.notice(rec.Host+"#offline", "moves: %s: offline move %s waits: this box cannot take part in an offline move (it needs hibernation configured, and %s and %s unset)", rec.Host, m.ID, "CARLOS_HOST_CLAIM", moves.SwitchEnv) } return nil } if m.State == provision.OfflineStaging { // The pre-copy (movesoffline_stage.go): nothing holds or fences // the instance, and only the stager acts. if source { return w.offlineStageStep(ctx, s, rec) } return nil } var errs []error if source { errs = append(errs, w.offlineSourceStep(ctx, s, rec)) } if sealer { errs = append(errs, w.offlineSealerStep(ctx, s, rec)) } if target { errs = append(errs, w.offlineTargetStep(ctx, s, rec)) } return errors.Join(errs...) } // offlineReassertEvery is how often a box that takes part in an offline move // rewrites the reports a door judges by age (harness Run 9, finding 7). The // doors believe a report for two minutes (offlineReportFresh, internal/ // console), and the moves pass runs on a serial move or every five to six // minutes. So a report written on the pass one door's bump started was stale // by the time an operator who waited over two minutes ran the next door, and // flip and open refused until the timed pass: most of Run 9's 366 s offline. // At this cadence a participant's report is never much more than 20 s old. const offlineReassertEvery = 15 * time.Second // offlineReassertRoles says which of this box's steps keep its report current // while a move is in m's state: a source box's stop from stopping to landed, // and the sealer's keep from sealed to landed. Those are the reports seal, // flip and open judge by age. The seal itself (sealing) and the target's // landing are not repeated here: they are long, and they run on the pass a // door's serial bump starts. From open on no door reads a report by age. // The sealer only re-reads and reports (offlineSealerRefresh): its ordinary // pass re-asserts the tombstone, and the refresh never does. func offlineReassertRoles(m *provision.MoveState, source, sealer bool) (asSource, asSealer bool) { switch m.State { case provision.OfflineStopping, provision.OfflineSealing: return source, false case provision.OfflineSealed, provision.OfflineLanding, provision.OfflineLanded: return source, sealer } return false, false } // offlineReassert is one tick of offlineReassertLoop: for each offline move // the view holds in which this box's report is judged by age, it reads the // record again and runs this box's source or sealer step for it, in // offlineRound's order, under the pass lock. Each step is an ensure that ends // by rewriting the box's report (offlineSourceStop, and offlineSealerRefresh, // which checks the tombstone but never re-asserts it). With no // such move in flight it reads nothing at all. It returns how many records it // acted on. func (w *movesWorker) offlineReassert(ctx context.Context, s release.Store) int { cfg := w.cfg v := cfg.Offline if !cfg.OfflineCapable || v == nil || s == nil || cfg.Box == "" { return 0 } snap := v.snap.Load() if snap == nil || snap.failed || len(snap.moves) == 0 { return 0 } here := cfg.parkRegion() roles := func(m *provision.MoveState) (bool, bool) { source := here != "" && park.PlacementMatches(cfg.ParkHomeRegion, m.From, here) return offlineReassertRoles(m, source, m.Sealer == cfg.Box) } var hosts []string for host, m := range snap.moves { if asSource, asSealer := roles(&m); asSource || asSealer { hosts = append(hosts, host) } } if len(hosts) == 0 { return 0 } sort.Strings(hosts) w.pass.Lock() defer w.pass.Unlock() acted := 0 for _, host := range hosts { owner := snap.owners[host] rec, err := provision.GetRecord(ctx, s, owner[0], owner[1], host) if err != nil { w.logTrouble(host, fmt.Errorf("offline move: read the record to report again: %w", err)) continue } if !rec.Move.Offline() { continue } asSource, asSealer := roles(rec.Move) if !asSource && !asSealer { continue } acted++ var errs []error if asSource { errs = append(errs, w.offlineSourceStep(ctx, s, *rec)) } if asSealer { errs = append(errs, w.offlineSealerRefresh(ctx, s, *rec)) } if err := errors.Join(errs...); err != nil { w.logTrouble(host, err) } } return acted } // now is cfg.Now, or the wall clock when a test left it unset. func (c moveCfg) now() time.Time { if c.Now != nil { return c.Now() } return time.Now() } // writeOfflineReport writes r at its key (moves.OfflineReportKey), stamped // with the report version, after the shape check every reader makes. The // source, sealer and target steps all write through it. func writeOfflineReport(ctx context.Context, s release.Store, r moves.OfflineReport) error { r.V = moves.OfflineReportVersion key := moves.OfflineReportKey(r.Host, r.Op, r.Role, r.Box) if key == "" { return fmt.Errorf("offline report: host %q, op %q, role %q or box %q is not usable", r.Host, r.Op, r.Role, r.Box) } if err := moves.ValidateOfflineReport(r); err != nil { return err } body, err := json.Marshal(r) if err != nil { return err } // A report is rewritten every pass: read its etag and replace it // (release.Store.Put with no etag is create-only). return moves.PutReplace(ctx, s, key, body) } // putOfflineCreateOnly writes body at key only if nothing is there yet: // the immutable seal record (control/offline-moves///seal.json, // moves.OfflineSealKey, written once by the sealer and never rewritten). // It is an ensure: when the key already exists it writes nothing and // returns what is stored, with created false, so a sealer re-running after // a crash compares rather than overwrites, and a reader (flip, the landing) // always sees the first write. key must be non-empty. func putOfflineCreateOnly(ctx context.Context, s release.Store, key string, body []byte) (stored []byte, created bool, err error) { if key == "" { return nil, false, errors.New("offline move: no key to write") } _, err = s.Put(ctx, key, body, "") if err == nil { return body, true, nil } if !errors.Is(err, release.ErrCASConflict) { return nil, false, fmt.Errorf("write %s: %w", key, err) } stored, _, err = s.Get(ctx, key) if err != nil { return nil, false, fmt.Errorf("read %s after a create-only write found it: %w", key, err) } return stored, false, nil } // readOfflineReports reads every report of one role for one move, keyed by // box. A report that does not validate is an error, never skipped. func readOfflineReports(ctx context.Context, s release.Store, host, op, role string) (map[string]moves.OfflineReport, error) { prefix := moves.OfflineReportPrefix(host, op) if prefix == "" { return nil, fmt.Errorf("offline report: host %q or op %q is not usable", host, op) } prefix += role + "/" keys, err := s.List(ctx, prefix) if err != nil { return nil, err } out := map[string]moves.OfflineReport{} for _, key := range keys { if !strings.HasSuffix(key, ".json") { continue } body, _, err := s.Get(ctx, key) if err != nil { return nil, err } var r moves.OfflineReport if err := json.Unmarshal(body, &r); err != nil { return nil, fmt.Errorf("%s: %w", key, err) } if err := moves.ValidateOfflineReport(r); err != nil { return nil, fmt.Errorf("%s: %w", key, err) } if moves.OfflineReportKey(r.Host, r.Op, r.Role, r.Box) != key { return nil, fmt.Errorf("%s: the report names %s/%s/%s/%s", key, r.Host, r.Op, r.Role, r.Box) } out[r.Box] = r } return out, nil } // offlineSourceStep is a source-region box's part of an offline move. Every // state's step is an ensure: it reads what is on disk, in the registry and // in the activator, does only what is missing, and reports what it found. func (w *movesWorker) offlineSourceStep(ctx context.Context, s release.Store, rec provision.Record) error { if w.cfg.Box == "" { return errors.New("offline move: this box has no label to report under") } switch rec.Move.State { case provision.OfflineStopping, provision.OfflineSealing, provision.OfflineSealed, provision.OfflineLanding, provision.OfflineLanded: // The pinned stager's stop acknowledgement first: the seal door // needs it beside this box's stop (movesoffline_stage.go). return errors.Join(w.offlineStagerStopped(ctx, s, rec), w.offlineSourceStop(ctx, s, rec)) case provision.OfflineOpen: return w.offlineSourcePark(ctx, s, rec) case provision.OfflineFinishing: return w.offlineSourceGone(ctx, s, rec) case provision.OfflineReverting: return w.offlineSourceRevert(ctx, s, rec) } return nil } // offlineFence installs, or meets again, this move's activator fence for // host. Fences live in the activator's memory, so every pass installs it // again and an agent restart regains it on its next pass. True when the // fence is this move's now. func (w *movesWorker) offlineFence(ctx context.Context, host string, m *provision.MoveState) bool { if w.cfg.FenceAs == nil { return false } if err := w.cfg.FenceAs(ctx, host, moveFenceGeneration(m), offlineFenceOwner(m.ID)); err != nil { w.notice(host+"#offline-fence", "moves: %s: offline move %s: the wake fence is not installed: %v", host, m.ID, err) return false } return true } // offlineLiftFence lifts this move's fence, and only it. func (w *movesWorker) offlineLiftFence(host string, m *provision.MoveState) { if w.cfg.LiftFenceAs != nil && w.cfg.LiftFenceAs(host, moveFenceGeneration(m), offlineFenceOwner(m.ID)) { log.Printf("moves: %s: offline move %s: wake fence lifted", host, m.ID) } } // offlineRow reads the host's row on this box; nil when it has none. func (w *movesWorker) offlineRow(host string) (*registry.Route, error) { if w.cfg.Route == nil { return nil, errors.New("offline move: no registry on this box") } return w.cfg.Route(host) } // heldBy reports whether rt is held by op. An absent row counts: nothing // can serve or wake a host this box has no row for. func heldBy(rt *registry.Route, op string) bool { return rt == nil || (rt.HoldOp == op && registry.IsHold(rt.Status)) } // offlineLocal is what a box keeps on disk for a host, judged by lstat and // directory listings only, never by following a link. A row's own layout is // read at the row's paths; a box with no row, or a leftover of the other // layout, is read at both. type offlineLocal struct { LocalDB, Files, Env, Sidecar bool LocalTXID uint64 } // flatDBPath is where a flat row's database is: the row's own path when it // has one and is flat, else the default. func (c moveCfg) flatDBPath(host string, rt *registry.Route) string { if rt != nil && rt.DBPath != "" && !provision.NestedPaths(c.DataDir, *rt) { return rt.DBPath } return provision.DefaultDBPath(c.DataDir, host) } // activatorMetaDir is the activator's litestream meta directory beside a // database (localLTXPosition reads it). func activatorMetaDir(dbPath string) string { return filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream") } func lexists(p string) bool { _, err := os.Lstat(p) return err == nil } func isDir(p string) bool { st, err := os.Lstat(p) return err == nil && st.IsDir() } // offlinePlatformName reports a nested instance directory's platform entry // (layout.IsNestedPlatformName: the database and its files, the env files, // the generation, seed and files markers, the adoption record, and the // launch descriptors, launch.gen..json included): everything else in it // is the app's files. func offlinePlatformName(host, name string) bool { return layout.IsNestedPlatformName(host, name) } // localState reads what this box keeps for host: the DB file or either // litestream meta dir (S8: a meta dir alone is local state, since // localLTXPosition reads it), the files directory, the env file and a // sidecar env file, under the row's layout and any leftover of the other. func (w *movesWorker) localState(host string, rt *registry.Route) (offlineLocal, error) { cfg := w.cfg var out offlineLocal nestedRow := rt != nil && provision.NestedPaths(cfg.DataDir, *rt) // Flat: a flat row's own paths, or a nested row's flat leftovers. { db := cfg.flatDBPath(host, rt) out.LocalDB = lexists(db) || lexists(db+"-litestream") || lexists(activatorMetaDir(db)) out.Files = isDir(filepath.Join(cfg.DataDir, host)) out.Env = lexists(filepath.Join(cfg.DataDir, host+".env")) out.Sidecar = lexists(filepath.Join(cfg.DataDir, host+"-sidecar.env")) pos, err := localLTXPosition(db) if err != nil { return out, err } out.LocalTXID = pos } // Nested, read through the raw directory (layout.RawInstanceDir). raw := layout.RawInstanceDir(cfg.DataDir, host) if nestedRow || isDir(raw) { db := filepath.Join(raw, host+".db") out.LocalDB = out.LocalDB || lexists(db) || lexists(db+"-litestream") || lexists(activatorMetaDir(db)) out.Env = out.Env || lexists(filepath.Join(raw, host+".env")) out.Sidecar = out.Sidecar || lexists(filepath.Join(raw, host+"-sidecar.env")) ents, err := os.ReadDir(raw) if err != nil && !errors.Is(err, os.ErrNotExist) { return out, err } for _, e := range ents { if !offlinePlatformName(host, e.Name()) { out.Files = true break } } dbPath := db if nestedRow && rt.DBPath != "" { dbPath = rt.DBPath } pos, err := localLTXPosition(dbPath) if err != nil { return out, err } out.LocalTXID = max(out.LocalTXID, pos) } return out, nil } // procs is the process count for the report: -1 when it cannot be read, // which the flip door reads as not stopped. func (w *movesWorker) procs(host string) int { if w.cfg.Procs == nil { return -1 } n, err := w.cfg.Procs(host) if err != nil { w.notice(host+"#offline-procs", "moves: %s: the host's processes cannot be counted on this box: %v", host, err) return -1 } return n } // sourceReport is the source report's evidence as it stands now. func (w *movesWorker) sourceReport(host string, m *provision.MoveState, rt *registry.Route, fenced bool, phase string) (moves.OfflineReport, error) { cfg := w.cfg local, err := w.localState(host, rt) if err != nil { return moves.OfflineReport{}, err } quiesced := cfg.Quiesced != nil && cfg.Quiesced(host) return moves.OfflineReport{ Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSource, Phase: phase, At: cfg.now().UTC(), Held: heldBy(rt, m.ID), Fenced: fenced, Quiesced: quiesced, RowPresent: rt != nil, LocalDB: local.LocalDB, Files: local.Files, Env: local.Env, Sidecar: local.Sidecar, LocalTXID: local.LocalTXID, Procs: w.procs(host), LayoutJournal: layout.JournalPresent(cfg.DataDir, host), }, nil } // offlineSourceStop is the source step from offline-stopping through // offline-landed: install the fence, read the row and the local state, and // rewrite this box's source report with At on every pass (the open door // needs it current). The row's hold is reconcile's (offlineHold) and, // after the flip, its keep (offlineView.keep). func (w *movesWorker) offlineSourceStop(ctx context.Context, s release.Store, rec provision.Record) error { m := rec.Move fenced := w.offlineFence(ctx, rec.Host, m) rt, err := w.offlineRow(rec.Host) if err != nil { return err } r, err := w.sourceReport(rec.Host, m, rt, fenced, moves.OfflinePhaseStopped) if err != nil { return err } return writeOfflineReport(ctx, s, r) } // retirePath is one path of the manifest's source-retire row on this box: // where it is, its name in the park directory, and, for a path the agent // never renames, why. type retirePath struct { path, name, left string } // Why a path is left where it is. const ( leftRootOwned = "root-owned: listed for the census, not renamed" leftView = "the mount view of the instance directory: it stays, empty" leftInstance = "the instance directory stays, empty" ) // nestedParkPrefix is where a nested instance directory's entries go inside // the park directory, apart from the flat names. const nestedParkPrefix = "instance" // retirePaths is the manifest's source-retire row as paths on this box, in // a fixed order: the row's layout in full, the other layout's paths only // where they exist, the app's own entries of a nested instance directory as // found there and in the park directory, then what is never renamed. func (w *movesWorker) retirePaths(host, parkDir string, rt *registry.Route) ([]retirePath, error) { cfg := w.cfg nestedRow := rt != nil && provision.NestedPaths(cfg.DataDir, *rt) raw := layout.RawInstanceDir(cfg.DataDir, host) var out []retirePath add := func(p, name string, always bool) { if always || lexists(p) { out = append(out, retirePath{path: p, name: name}) } } flatDB := cfg.flatDBPath(host, rt) flat := !nestedRow add(flatDB, filepath.Base(flatDB), flat) add(flatDB+"-wal", filepath.Base(flatDB)+"-wal", flat) add(flatDB+"-shm", filepath.Base(flatDB)+"-shm", flat) add(flatDB+"-litestream", filepath.Base(flatDB)+"-litestream", flat) add(activatorMetaDir(flatDB), filepath.Base(activatorMetaDir(flatDB)), flat) add(filepath.Join(cfg.DataDir, host), host, flat) add(filepath.Join(cfg.DataDir, host+".env"), host+".env", flat) add(filepath.Join(cfg.DataDir, host+".config.env"), host+".config.env", flat) nested := nestedRow || isDir(raw) if nested { named := map[string]bool{} for _, n := range []string{host + ".db", host + ".db-wal", host + ".db-shm", host + ".db-litestream", ".activator-" + host + ".db-litestream", host + ".env", host + ".config.env"} { named[n] = true add(filepath.Join(raw, n), nestedParkPrefix+"/"+n, true) } // The app's files, and every other entry the directory holds: what // is still there, and what an earlier pass already parked. // A numbered copy in the park directory (parkName) belongs to the // entry it is a copy of, not to the app. others := map[string]bool{} listed := map[string][]string{} for _, dir := range []string{raw, filepath.Join(parkDir, nestedParkPrefix)} { ents, err := os.ReadDir(dir) if err != nil && !errors.Is(err, os.ErrNotExist) { return nil, err } for _, e := range ents { listed[dir] = append(listed[dir], e.Name()) } } known := map[string]bool{} for n := range named { known[n] = true } for _, n := range listed[raw] { known[n] = true } for _, n := range listed[filepath.Join(parkDir, nestedParkPrefix)] { known[n] = true } for _, n := range listed[raw] { if !named[n] { others[n] = true } } for _, n := range listed[filepath.Join(parkDir, nestedParkPrefix)] { if !named[n] && !isNumberedCopy(n, known) { others[n] = true } } names := make([]string, 0, len(others)) for n := range others { names = append(names, n) } sort.Strings(names) for _, n := range names { add(filepath.Join(raw, n), nestedParkPrefix+"/"+n, true) } } add(filepath.Join(cfg.DataDir, "programs", host+".json"), host+".json", true) if nested { out = append(out, retirePath{path: provision.InstanceDir(cfg.DataDir, host), left: leftView}) out = append(out, retirePath{path: raw, left: leftInstance}) } optDir := cfg.OptDir if optDir == "" { optDir = release.OptDir() } out = append(out, retirePath{path: filepath.Join(optDir, "live", host), left: leftRootOwned}) return out, nil } // parkCopySep separates a name from its copy number in the park directory // (parkName). Chosen to be unlikely in an app's own file names. const parkCopySep = ".parked-" // isNumberedCopy reports whether n is a known entry's numbered park copy // (parkName's ".parked-"), which belongs to that entry. func isNumberedCopy(n string, known map[string]bool) bool { i := strings.LastIndex(n, parkCopySep) if i <= 0 || !known[n[:i]] || i+len(parkCopySep) == len(n) { return false } for _, c := range n[i+len(parkCopySep):] { if c < '0' || c > '9' { return false } } return true } // parkName is where name goes in dir: name itself, or name.parked- for // the first free n when something of that name was parked before (the // source path came back after an earlier pass parked it). The park // directory is the move's own, so the first park of every path lands at its // plain name and a later pass reports it from there. func parkName(dir, name string) string { dst := filepath.Join(dir, name) if !lexists(dst) { return dst } for n := 1; ; n++ { if c := dst + parkCopySep + fmt.Sprint(n); !lexists(c) { return c } } } // Why a rename the park attempted did not happen. A path whose rename fails // with an error a later pass would meet again (a mount, EXDEV, EBUSY; a path // the agent may not rename, EPERM or EACCES; a read-only filesystem) is // left for good: reported once, with leftCannot and the error, and never // tried again for this move, so a pass does not re-fence, fail and lift // over it. Any other error is leftNotYet, and the next pass tries again. const ( leftCannot = "cannot be renamed: " leftNotYet = "not renamed yet: " ) // renameTerminal reports whether a rename error is one a later pass would // meet again. func renameTerminal(err error) bool { return errors.Is(err, os.ErrPermission) || errors.Is(err, syscall.EXDEV) || errors.Is(err, syscall.EBUSY) || errors.Is(err, syscall.EROFS) } // retire reports one path's action from what is on disk, renaming it into // the park directory first when park is true and it is there to rename. // cannot is the reason an earlier pass recorded for a path it found it can // never rename ("" when none): such a path is reported left again, as it // is, and not tried. Nothing is ever deleted. func retire(p retirePath, parkDir string, park bool, cannot string) moves.RetiredPath { present := lexists(p.path) switch { case p.left != "" && present: return moves.RetiredPath{Path: p.path, Action: moves.RetiredLeft, Reason: p.left} case p.left != "": return moves.RetiredPath{Path: p.path, Action: moves.RetiredAbsent} case present && cannot != "": return moves.RetiredPath{Path: p.path, Action: moves.RetiredLeft, Reason: cannot} case present && park: dst := parkName(parkDir, p.name) err := os.MkdirAll(filepath.Dir(dst), 0o700) if err == nil { err = os.Rename(p.path, dst) } switch { case err == nil: return moves.RetiredPath{Path: p.path, Action: moves.RetiredParked} case errors.Is(err, os.ErrNotExist) && !lexists(p.path): return moves.RetiredPath{Path: p.path, Action: moves.RetiredAbsent} case renameTerminal(err): return moves.RetiredPath{Path: p.path, Action: moves.RetiredLeft, Reason: leftCannot + err.Error()} } return moves.RetiredPath{Path: p.path, Action: moves.RetiredLeft, Reason: leftNotYet + err.Error()} case present: return moves.RetiredPath{Path: p.path, Action: moves.RetiredLeft, Reason: leftNotYet + "the stop could not be confirmed this pass"} case lexists(filepath.Join(parkDir, p.name)): return moves.RetiredPath{Path: p.path, Action: moves.RetiredParked} } return moves.RetiredPath{Path: p.path, Action: moves.RetiredAbsent} } // cannotRename reads this box's own last report for the move and returns // the paths it recorded as never renameable (leftCannot), with their // reasons. Reading them from the report, not from memory, keeps the answer // across an agent restart. func (w *movesWorker) cannotRename(ctx context.Context, s release.Store, host, op string) (map[string]string, error) { out := map[string]string{} body, _, err := s.Get(ctx, moves.OfflineReportKey(host, op, moves.OfflineRoleSource, w.cfg.Box)) if errors.Is(err, os.ErrNotExist) { return out, nil } if err != nil { return nil, err } var r moves.OfflineReport if err := json.Unmarshal(body, &r); err != nil { return nil, err } for _, rp := range r.Retired { if rp.Action == moves.RetiredLeft && strings.HasPrefix(rp.Reason, leftCannot) { out[rp.Path] = rp.Reason } } return out, nil } // offlineSourcePark is the source step at offline-open: with the fence held, // the row held (or absent), the host quiesced, no process left and no layout // journal, rename every path of the source-retire row that is still here // into the park directory; then lift the fence and report parked with the // inventory. A pass that finds nothing left to rename does not fence again: // it reports from what is parked. A path that can never be renamed // (leftCannot) is reported left with its reason and does not count as left // to rename, so it neither blocks the parked report nor brings the fence // back on every pass. A rename that may yet succeed (leftNotYet) reports // failed, and the next pass tries again. Nothing is deleted. func (w *movesWorker) offlineSourcePark(ctx context.Context, s release.Store, rec provision.Record) error { cfg := w.cfg m, host := rec.Move, rec.Host parkDir := moves.OfflineParkDir(cfg.DataDir, host, m.ID, moves.OfflineRoleSource) if parkDir == "" { return fmt.Errorf("offline move %s: no park directory for %s under %q", m.ID, host, cfg.DataDir) } rt, err := w.offlineRow(host) if err != nil { return err } paths, err := w.retirePaths(host, parkDir, rt) if err != nil { return err } cannot, err := w.cannotRename(ctx, s, host, m.ID) if err != nil { return err } pending := false for _, p := range paths { if p.left == "" && cannot[p.path] == "" && lexists(p.path) { pending = true break } } if pending { fenced := w.offlineFence(ctx, host, m) quiesced := cfg.Quiesced != nil && cfg.Quiesced(host) procs := w.procs(host) journal := layout.JournalPresent(cfg.DataDir, host) if !fenced || !heldBy(rt, m.ID) || !quiesced || procs != 0 || journal { return fmt.Errorf("offline move %s: not parking %s yet: fenced %v, held %v, quiesced %v, processes %d, layout journal %v", m.ID, host, fenced, heldBy(rt, m.ID), quiesced, procs, journal) } if err := os.MkdirAll(parkDir, 0o700); err != nil { return err } } retired := make([]moves.RetiredPath, 0, len(paths)) retry := false for _, p := range paths { rp := retire(p, parkDir, pending, cannot[p.path]) if rp.Action == moves.RetiredLeft && strings.HasPrefix(rp.Reason, leftNotYet) { retry = true } retired = append(retired, rp) } w.offlineLiftFence(host, m) phase := moves.OfflinePhaseParked if retry { phase = moves.OfflinePhaseFailed } r, err := w.sourceReport(host, m, rt, false, phase) if err != nil { return err } r.Retired = retired if err := writeOfflineReport(ctx, s, r); err != nil { return err } if retry { return fmt.Errorf("offline move %s: %s: some paths were not renamed this pass; the next pass tries again", m.ID, host) } return nil } // offlineSourceGone is the source step at offline-finishing: keep no longer // holds the row, so reconcile removes it; once it is gone, report gone. The // fence was lifted at open and is lifted again, should a restart have // re-fenced in between. func (w *movesWorker) offlineSourceGone(ctx context.Context, s release.Store, rec provision.Record) error { m, host := rec.Move, rec.Host rt, err := w.offlineRow(host) if err != nil { return err } w.offlineLiftFence(host, m) if rt != nil { w.notice(host+"#offline-gone", "moves: %s: offline move %s: waiting for reconcile to remove the row", host, m.ID) return nil } r, err := w.sourceReport(host, m, nil, false, moves.OfflinePhaseGone) if err != nil { return err } return writeOfflineReport(ctx, s, r) } // offlineSourceRevert is every source box's part of offline-reverting: the // row leaves this move's hold (ReleaseHold under the host's hold lock), the // fence is lifted, and the box reports reverted. The sealer's extra part is // its own step's (Task 4). // // The release also holds reconcileMu: a reconcile pass that listed the // records before the abort would otherwise mint a missing row held again // from its stale listing after the release, and a row held by a move that // has finished is never released. A pass that starts after the release // lists the record at offline-reverting, which holds nothing. func (w *movesWorker) offlineSourceRevert(ctx context.Context, s release.Store, rec provision.Record) error { cfg := w.cfg m, host := rec.Move, rec.Host if cfg.ReleaseHold == nil { return errors.New("offline move: no registry to release the hold through") } var relErr error reconcileMu.Lock() provision.WithHostHoldLock(host, func() { _, relErr = cfg.ReleaseHold(host, m.ID) }) reconcileMu.Unlock() if relErr != nil { return relErr } w.offlineLiftFence(host, m) rt, err := w.offlineRow(host) if err != nil { return err } if rt != nil && rt.HoldOp == m.ID { return fmt.Errorf("offline move %s: %s is still held by the move", m.ID, host) } // A staged move's pre-copy is removed before this box says reverted. if err := w.offlineUnstage(ctx, s, rec); err != nil { return err } r, err := w.sourceReport(host, m, rt, false, moves.OfflinePhaseReverted) if err != nil { return err } r.Held = false return writeOfflineReport(ctx, s, r) }