amadan
host: participant roles in every box path, the queued and staging reports, and a move-bound stage book
offlineRound, offlineReassert, the hold and the keep view resolve a box's role through moves.ParticipantRole (box id and instance id with Participants, today's region and label rules without); a box the move pins on another instance logs once and does nothing. In offline-queued the stager probes the lane once per move (the existing laneProbe into the staging bucket), every source participant writes its precheck (a dry run of archive.Build over the seal's selection, the layout, the strays, the build, the stage path) and the target its ready report (route, timers, live link, programs file, data paths, park directory, disk, the staging bucket's two prefixes and lock, capabilities, key fingerprint, stage path), each for the record's precheck_at; in offline-staging both report again for a newer stamp. A self-serve start removes empty stray directories once the stop holds. The stager keeps a book of every decision its rounds made (lease.MirrorPrefixKeepObserved), seeded from its last valid report and reconciled on each complete round, and reports it with StagedKnown, the record's destination and StagedRequest. The sealer finds a pinned move's target by box id. armOffline arms BoxID, the lane probe over NewVersionedS3, DiskFree and StagePathArmed; armMovesPass sets BoxID and InstanceID. Every report goes through putOfflineReport, which stamps the box's identity. The status tick's per-host facts (hostFacts, measureFiles, one replica listing per tick) are here, wired next.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
8 files changed,
+1124
−49
internal/host/movesoffline.go+127 −22internal/host/movesoffline_land.go+242 −4internal/host/movesoffline_reports.go+549 −0internal/host/movesoffline_stage.go+135 −20internal/host/movesoffline_stage_test.go+2 −2internal/host/movespass.go+48 −0internal/host/movesreport.go+10 −1internal/host/probes.go+11 −0
diff --git a/internal/host/movesoffline.go b/internal/host/movesoffline.go| index 3052233d5..fc01a42cd 100644 |
| --- a/internal/host/movesoffline.go |
| +++ b/internal/host/movesoffline.go |
| @@ -89,20 +89,28 @@ func offlineFenceOwner(id string) string { return "offline:" + id } |
| // 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. |
| +// |
| +// A move with pinned participants (a self-serve move) holds by role, not |
| +// by region: only the source participants hold "moving" and only the |
| +// target participant holds "landing" (moves.ParticipantRole), so another |
| +// box of either region, or a replaced box, holds nothing. |
| 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 == "" { |
| + if !m.Offline() || m.ID == "" { |
| + return "", "", false |
| + } |
| + source, target, _, _ := box.offlineRoles(m) |
| + if m.Participants == nil && box.parkRegion() == "" { |
| return "", "", false |
| } |
| switch m.State { |
| case provision.OfflineStopping, provision.OfflineSealing, provision.OfflineSealed: |
| - if park.PlacementMatches(home, m.From, here) { |
| + if source { |
| return "moving", m.ID, true |
| } |
| case provision.OfflineLanding, provision.OfflineLanded: |
| - if rec.Region == m.To && park.PlacementMatches(home, m.To, here) { |
| + if rec.Region == m.To && target { |
| return "landing", m.ID, true |
| } |
| } |
| @@ -110,6 +118,14 @@ func offlineHold(box moveCfg) func(rec provision.Record) (status, op string, ok |
| } |
| } |
| +// offlineRoles is this box's role in the offline move m, the one resolver |
| +// every box path asks (moves.ParticipantRole): by box id and instance id |
| +// for a move with pinned participants, by placement and label for any |
| +// other. |
| +func (c moveCfg) offlineRoles(m *provision.MoveState) (source, target, sealer, stager bool) { |
| + return moves.ParticipantRole(m, c.BoxID, c.InstanceID, c.Box, c.parkRegion(), c.ParkHomeRegion) |
| +} |
| + |
| // releaseHeldSeq is the offline release in production: registry. |
| // ReleaseHoldSeq, then written(host) only when the release changed a row, as |
| // releaseHeld does, so root re-judges the now-ready row (registry.SetHold's |
| @@ -143,6 +159,9 @@ func releaseHeldSeq(reg *registry.Registry, written func(host string)) func(host |
| // always have. |
| type offlineView struct { |
| home, region string |
| + // box is the box the view keeps rows for, as offlineRoles reads it: a |
| + // move with pinned participants is kept by role, never by region. |
| + box moveCfg |
| // judge is the box's lifetime judge (moveCfg.Judge): a listing read |
| // outside a moves pass (refreshOfflineView) is judged by it, so a record |
| // its slot no longer vouches for keeps no row. Nil: unjudged. |
| @@ -168,7 +187,8 @@ type offlineSnapshot struct { |
| // 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, judge: cfg.Judge} |
| + box := moveCfg{Box: cfg.Box, BoxID: cfg.BoxID, InstanceID: cfg.InstanceID, Region: cfg.Region, ParkHomeRegion: cfg.ParkHomeRegion} |
| + return &offlineView{home: cfg.ParkHomeRegion, region: cfg.parkRegion(), box: box, written: written, judge: cfg.Judge} |
| } |
| // refresh rebuilds the view from a record list that was read, with the |
| @@ -219,7 +239,19 @@ func (v *offlineView) keep(host string) bool { |
| return true |
| } |
| m, ok := snap.moves[host] |
| - if !ok || v.region == "" { |
| + if !ok { |
| + return false |
| + } |
| + if m.Participants != nil { |
| + // Pinned participants: kept by role (offlineRoles), whatever the |
| + // region; a box the move does not name keeps nothing for it. |
| + source, target, _, _ := v.box.offlineRoles(&m) |
| + if target { |
| + return true |
| + } |
| + return source && m.State != provision.OfflineFinishing |
| + } |
| + if v.region == "" { |
| return false |
| } |
| if park.PlacementMatches(v.home, m.To, v.region) { |
| @@ -413,29 +445,53 @@ func hostSliceCgroup(cgroupRoot, host string) string { |
| // 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. |
| +// |
| +// A move with pinned participants (a self-serve move) is resolved by box id |
| +// and instance id (moveCfg.offlineRoles): a box that is not a participant |
| +// does nothing for it, whatever its region, and a box whose box id the move |
| +// pinned on another instance logs once per move that it is not the pinned |
| +// instance and does nothing. In offline-queued a participant only reports |
| +// (offlineQueuedStep): the stager's lane probe, the source's precheck and |
| +// the target's ready report; in offline-staging the stager also stages and |
| +// a participant reports again for a newer precheck stamp. |
| 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 moves.ParticipantReplaced(m, cfg.BoxID, cfg.InstanceID) { |
| + if _, seen := w.replaced.LoadOrStore(m.ID, true); !seen { |
| + log.Printf("moves: %s: box %s is not the pinned instance for move %s", rec.Host, cfg.Box, m.ID) |
| + } |
| + return nil |
| + } |
| + source, target, sealer, stager := cfg.offlineRoles(m) |
| 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 { |
| + switch m.State { |
| + case provision.OfflineQueued: |
| + // Nothing is copied, held or fenced: the participants report. |
| + if m.Participants == nil { |
| + return nil |
| + } |
| + return w.offlineQueuedStep(ctx, s, rec, source, target, stager) |
| + case 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) |
| + // the instance, and only the stager stages. Without pinned |
| + // participants the stage step itself checks the stager's label. |
| + var errs []error |
| + if (m.Participants == nil && source) || stager { |
| + errs = append(errs, w.offlineStageStep(ctx, s, rec)) |
| } |
| - return nil |
| + if m.Participants != nil { |
| + errs = append(errs, w.offlineReReport(ctx, s, rec, source, target)) |
| + } |
| + return errors.Join(errs...) |
| } |
| var errs []error |
| if source { |
| @@ -496,10 +552,9 @@ func (w *movesWorker) offlineReassert(ctx context.Context, s release.Store) int |
| 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) |
| + source, _, sealer, _ := cfg.offlineRoles(m) |
| + return offlineReassertRoles(m, source, sealer) |
| } |
| var hosts []string |
| for host, m := range snap.moves { |
| @@ -573,6 +628,17 @@ func writeOfflineReport(ctx context.Context, s release.Store, r moves.OfflineRep |
| return moves.PutReplace(ctx, s, key, body) |
| } |
| +// putOfflineReport is writeOfflineReport for this box: a report that names |
| +// no identity is stamped with this box's (moveCfg.BoxID, InstanceID), so |
| +// every report a self-serve move reads carries the identity its pinned |
| +// participants are checked against (moves.ValidateOfflineReportFor). |
| +func (w *movesWorker) putOfflineReport(ctx context.Context, s release.Store, r moves.OfflineReport) error { |
| + if r.BoxID == "" && r.InstanceID == "" { |
| + r.BoxID, r.InstanceID = w.cfg.BoxID, w.cfg.InstanceID |
| + } |
| + return writeOfflineReport(ctx, s, r) |
| +} |
| + |
| // putOfflineCreateOnly writes body at key only if nothing is there yet: |
| // the immutable seal record (control/offline-moves/<host>/<id>/seal.json, |
| // moves.OfflineSealKey, written once by the sealer and never rewritten). |
| @@ -811,6 +877,7 @@ func (w *movesWorker) sourceReport(host string, m *provision.MoveState, rt *regi |
| 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), |
| + Layout: w.offlineLayout(rt), BoxID: cfg.BoxID, InstanceID: cfg.InstanceID, |
| }, nil |
| } |
| @@ -819,6 +886,12 @@ func (w *movesWorker) sourceReport(host string, m *provision.MoveState, rt *regi |
| // 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). |
| +// |
| +// A self-serve move's start also removes the empty directories the |
| +// instance keeps beside its files directory (removeEmptyStrays), in |
| +// offline-stopping and only once the stop holds (held, fenced, quiesced, no |
| +// process): the stopped report carries what was removed, kept from every |
| +// earlier pass of the move, and what was left. |
| func (w *movesWorker) offlineSourceStop(ctx context.Context, s release.Store, rec provision.Record) error { |
| m := rec.Move |
| fenced := w.offlineFence(ctx, rec.Host, m) |
| @@ -830,7 +903,39 @@ func (w *movesWorker) offlineSourceStop(ctx context.Context, s release.Store, re |
| if err != nil { |
| return err |
| } |
| - return writeOfflineReport(ctx, s, r) |
| + if m.SelfServe != nil && m.State == provision.OfflineStopping { |
| + p, err := w.startStrays(ctx, s, rec, rt, r) |
| + if err != nil { |
| + return err |
| + } |
| + r.Precheck = p |
| + } |
| + return w.putOfflineReport(ctx, s, r) |
| +} |
| + |
| +// startStrays is the start's removal of empty stray directories for the |
| +// stopped report r: only once r says the stop holds. What earlier passes |
| +// of this move removed is kept from this box's last report. |
| +func (w *movesWorker) startStrays(ctx context.Context, s release.Store, rec provision.Record, rt *registry.Route, r moves.OfflineReport) (*moves.SourcePrecheck, error) { |
| + prev, err := w.ownReport(ctx, s, rec, moves.OfflineRoleSource) |
| + if err != nil { |
| + return nil, fmt.Errorf("offline move %s: read the source report: %w", rec.Move.ID, err) |
| + } |
| + p := &moves.SourcePrecheck{Layout: r.Layout} |
| + if prev != nil && prev.Phase == moves.OfflinePhaseStopped && prev.Precheck != nil { |
| + p.StraysRemoved = append(p.StraysRemoved, prev.Precheck.StraysRemoved...) |
| + } |
| + if r.Held && r.Fenced && r.Quiesced && r.Procs == 0 && !r.LayoutJournal { |
| + removed, left, err := w.removeEmptyStrays(rec.Host, rt) |
| + if err != nil { |
| + return nil, fmt.Errorf("offline move %s: read what %s keeps: %w", rec.Move.ID, rec.Host, err) |
| + } |
| + p.StraysRemoved, p.StraysLeft = append(p.StraysRemoved, removed...), left |
| + } |
| + if len(p.StraysRemoved) == 0 && len(p.StraysLeft) == 0 { |
| + return nil, nil |
| + } |
| + return p, nil |
| } |
| // retirePath is one path of the manifest's source-retire row on this box: |
| @@ -1124,7 +1229,7 @@ func (w *movesWorker) offlineSourcePark(ctx context.Context, s release.Store, re |
| return err |
| } |
| r.Retired = retired |
| - if err := writeOfflineReport(ctx, s, r); err != nil { |
| + if err := w.putOfflineReport(ctx, s, r); err != nil { |
| return err |
| } |
| if retry { |
| @@ -1152,7 +1257,7 @@ func (w *movesWorker) offlineSourceGone(ctx context.Context, s release.Store, re |
| if err != nil { |
| return err |
| } |
| - return writeOfflineReport(ctx, s, r) |
| + return w.putOfflineReport(ctx, s, r) |
| } |
| // offlineSourceRevert is every source box's part of offline-reverting: the |
| @@ -1201,5 +1306,5 @@ func (w *movesWorker) offlineSourceRevert(ctx context.Context, s release.Store, |
| return err |
| } |
| r.Held = false |
| - return writeOfflineReport(ctx, s, r) |
| + return w.putOfflineReport(ctx, s, r) |
| } |
diff --git a/internal/host/movesoffline_land.go b/internal/host/movesoffline_land.go| index 3dc24e6e7..39aa6c120 100644 |
| --- a/internal/host/movesoffline_land.go |
| +++ b/internal/host/movesoffline_land.go |
| @@ -20,6 +20,7 @@ import ( |
| "github.com/carlosframework/platform/internal/activator" |
| "github.com/carlosframework/platform/internal/appconfig" |
| "github.com/carlosframework/platform/internal/instancekey" |
| + "github.com/carlosframework/platform/internal/layout" |
| "github.com/carlosframework/platform/internal/lease" |
| "github.com/carlosframework/platform/internal/moves" |
| "github.com/carlosframework/platform/internal/moves/archive" |
| @@ -83,7 +84,7 @@ func (w *movesWorker) offlineLand(ctx context.Context, s release.Store, rec prov |
| if err != nil { |
| return err |
| } |
| - if err := writeOfflineReport(ctx, s, r); err != nil { |
| + if err := w.putOfflineReport(ctx, s, r); err != nil { |
| return err |
| } |
| if err := w.landCrash("report"); err != nil { |
| @@ -150,7 +151,7 @@ func (w *movesWorker) offlineTargetOpen(ctx context.Context, s release.Store, re |
| } |
| // The verified report goes first, so a crash after the release |
| // leaves the items this pass verified, not an older pass's. |
| - if err := writeOfflineReport(ctx, s, r); err != nil { |
| + if err := w.putOfflineReport(ctx, s, r); err != nil { |
| return err |
| } |
| if !ok { |
| @@ -209,7 +210,7 @@ func (w *movesWorker) offlineTargetOpen(ctx context.Context, s release.Store, re |
| return fmt.Errorf("offline move %s: delete the archive: %w", m.ID, err) |
| } |
| } |
| - return writeOfflineReport(ctx, s, moves.OfflineReport{ |
| + return w.putOfflineReport(ctx, s, moves.OfflineReport{ |
| Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleTarget, Phase: moves.OfflinePhaseOpened, At: cfg.now().UTC(), |
| Items: landingItemsOnly(items), RowPresent: rt != nil, Files: isDir(filepath.Join(cfg.DataDir, host)), Env: lexists(filepath.Join(cfg.DataDir, host+".env")), |
| }) |
| @@ -275,7 +276,7 @@ func (w *movesWorker) offlineTargetRevert(ctx context.Context, s release.Store, |
| } |
| log.Printf("moves: %s: offline move %s: the target's held row is deleted", host, m.ID) |
| } |
| - return writeOfflineReport(ctx, s, moves.OfflineReport{ |
| + return w.putOfflineReport(ctx, s, moves.OfflineReport{ |
| Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleTarget, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC(), |
| RowPresent: rt != nil, Files: isDir(filepath.Join(cfg.DataDir, host)), Env: lexists(filepath.Join(cfg.DataDir, host+".env")), |
| }) |
| @@ -1253,3 +1254,240 @@ func (r *landRun) liveNotSealed() error { |
| } |
| return nil |
| } |
| + |
| +// ---- the target's ready report (self-serve moves plan 2 Task 6) ---- |
| + |
| +// offlineReadyDiskFloor is the free space below which the target's own |
| +// disk fact is not OK, whatever the move needs: the console compares |
| +// DiskFreeBytes with twice the replica's size, and this box keeps T6's |
| +// margin (checkDisk) for itself. |
| +const offlineReadyDiskFloor = 2 << 30 |
| + |
| +// offlineReadyReport writes the target's ready report for stamp. |
| +func (w *movesWorker) offlineReadyReport(ctx context.Context, s release.Store, rec provision.Record, stamp time.Time) error { |
| + ready, err := w.offlineTargetReady(ctx, s, rec) |
| + if err != nil { |
| + return fmt.Errorf("offline move %s: ready report for %s: %w", rec.Move.ID, rec.Host, err) |
| + } |
| + r := w.selfServeReport(rec, moves.OfflineRoleTarget, moves.OfflinePhaseReady) |
| + r.Request, r.Ready = stamp, &ready |
| + return w.putOfflineReport(ctx, s, r) |
| +} |
| + |
| +// offlineTargetReady is the target box's facts for a self-serve move before |
| +// any copy and again before the stop (moves.TargetReadyReport). Each fact |
| +// says whether the box could tell (Known) and whether what it saw is clean |
| +// (OK); the console decides. It reads and never changes anything: the |
| +// registry row and timers, the live link and the programs file under the |
| +// version store and the data directory, the instance's data paths, the |
| +// park directory, the disk, the staging bucket's moves/<host>/ and |
| +// instances/<host>/ prefixes and their lock.json, and this box's own edges |
| +// file for its capabilities and secrets key fingerprint. It errs only on a |
| +// record without a move. |
| +func (w *movesWorker) offlineTargetReady(ctx context.Context, s release.Store, rec provision.Record) (moves.TargetReadyReport, error) { |
| + cfg := w.cfg |
| + m, host := rec.Move, rec.Host |
| + if m == nil { |
| + return moves.TargetReadyReport{}, errors.New("the record carries no move") |
| + } |
| + var out moves.TargetReadyReport |
| + |
| + // The row and the timers. |
| + if rt, err := w.offlineRow(host); err != nil { |
| + out.RouteAbsent = moves.Fact{Detail: err.Error()} |
| + } else if rt != nil { |
| + out.RouteAbsent = moves.Fact{Known: true, Detail: fmt.Sprintf("a %q row held by %q", rt.Status, rt.HoldOp)} |
| + } else { |
| + out.RouteAbsent = moves.Fact{Known: true, OK: true} |
| + } |
| + if cfg.Registry == nil { |
| + out.TimersAbsent = moves.Fact{Detail: "no registry on this box"} |
| + } else if rows, err := cfg.Registry.TimersFor(host); err != nil { |
| + out.TimersAbsent = moves.Fact{Detail: err.Error()} |
| + } else { |
| + out.TimersAbsent = moves.Fact{Known: true, OK: len(rows) == 0} |
| + if len(rows) > 0 { |
| + out.TimersAbsent.Detail = fmt.Sprintf("%d timer rows", len(rows)) |
| + } |
| + } |
| + |
| + // The build the record's channel resolves to, and what the live link |
| + // and the programs file name. |
| + route := registry.Route{Host: host, Account: rec.Account, App: rec.App, Channel: rec.Channel} |
| + var channelErr error |
| + if cfg.ChannelVersion == nil { |
| + channelErr = errors.New("this box cannot resolve channels") |
| + } else if v, err := cfg.ChannelVersion(ctx, s, route); err != nil { |
| + channelErr = err |
| + } else { |
| + out.ChannelBuild = v |
| + } |
| + optDir := cfg.OptDir |
| + if optDir == "" { |
| + optDir = release.OptDir() |
| + } |
| + out.LiveLink = liveLinkFact(optDir, route, out.ChannelBuild, channelErr) |
| + programs := filepath.Join(cfg.DataDir, "programs", host+".json") |
| + switch _, err := os.Lstat(programs); { |
| + case errors.Is(err, os.ErrNotExist): |
| + out.ProgramsFile = moves.Fact{Known: true, OK: true} |
| + case err != nil: |
| + out.ProgramsFile = moves.Fact{Detail: err.Error()} |
| + default: |
| + // The file names programs (Go main package paths), not builds: a |
| + // present one is reported with what it lists. |
| + out.ProgramsFile = moves.Fact{Known: true, Detail: strings.Join(release.ReadPrograms(cfg.DataDir, host), ", ")} |
| + } |
| + |
| + // The instance's paths on this box, every one absent. |
| + out.DataPaths = map[string]moves.Fact{} |
| + for _, p := range []string{provision.InstanceDir(cfg.DataDir, host), layout.RawInstanceDir(cfg.DataDir, host), |
| + filepath.Join(cfg.DataDir, host), filepath.Join(cfg.DataDir, host+".env"), filepath.Join(cfg.DataDir, host+".config.env"), |
| + filepath.Join(cfg.DataDir, host+".db")} { |
| + out.DataPaths[p] = absentFact(p) |
| + } |
| + |
| + // The park directory: absent, or holding only earlier moves' |
| + // directories. |
| + out.ParkDir = parkDirFact(filepath.Join(cfg.DataDir, ".offline-moves", host), m.ID) |
| + |
| + // The disk. |
| + if cfg.DiskFree == nil { |
| + out.DiskFree = moves.Fact{Detail: "this box cannot read its free disk"} |
| + } else if total, avail, err := cfg.DiskFree(cfg.DataDir); err != nil || total == 0 { |
| + out.DiskFree = moves.Fact{Detail: fmt.Sprint("the free disk could not be read: ", err)} |
| + } else { |
| + out.DiskFreeBytes = int64(avail) |
| + used := total - avail |
| + ok := avail >= offlineReadyDiskFloor && float64(used)/float64(total) < 0.85 |
| + out.DiskFree = moves.Fact{Known: true, OK: ok, Detail: fmt.Sprintf("%d of %d bytes free", avail, total)} |
| + } |
| + |
| + // The staging bucket's two prefixes and the lock. |
| + bucket, region := m.StageBucket, m.StageRegion |
| + if bucket == "" { |
| + bucket, region = cfg.OwnBucket, cfg.OwnBucketRegion |
| + } |
| + var locks []string |
| + movesPrefix := activator.MovesPrefix(host) |
| + instancesPrefix := activator.HibernatePrefix(cfg.OwnPrefix, host) + "/" |
| + lockKnown := true |
| + if bucket == "" || cfg.Versioned == nil { |
| + lockKnown = false |
| + } else if vs, err := cfg.Versioned(ctx, bucket, region); err != nil { |
| + w.notice(host+"#offline-ready-bucket", "moves: %s: offline move %s: the staging bucket %s cannot be read: %v", host, m.ID, bucket, err) |
| + lockKnown = false |
| + } else { |
| + var lock bool |
| + out.MovesPrefix, lock = prefixFacts(ctx, vs, bucket, movesPrefix) |
| + if lock { |
| + locks = append(locks, movesPrefix+"lock.json") |
| + } |
| + lockKnown = lockKnown && out.MovesPrefix.Known |
| + out.InstancesPrefix, lock = prefixFacts(ctx, vs, bucket, instancesPrefix) |
| + if lock { |
| + locks = append(locks, instancesPrefix+"lock.json") |
| + } |
| + lockKnown = lockKnown && out.InstancesPrefix.Known |
| + } |
| + out.Lock = moves.Fact{Known: lockKnown, OK: lockKnown && len(locks) == 0, Detail: strings.Join(locks, ", ")} |
| + |
| + // This box's own published capabilities and key fingerprint. |
| + if raw, _, err := s.Get(ctx, edgesFileKey(cfg.Box)); err == nil { |
| + var f edgesFile |
| + if json.Unmarshal(raw, &f) == nil { |
| + out.Capabilities, out.SecretsKeyFP = f.Capabilities, f.SecretsKeyFP |
| + } |
| + } |
| + if out.Capabilities == nil { |
| + out.Capabilities = []string{} |
| + } |
| + out.StagePathArmed = armedFact(cfg.StagePathArmed) |
| + return out, nil |
| +} |
| + |
| +// liveLinkFact is the live link for route: absent is clean; a symlink that |
| +// names the channel's build where PointLive puts it is clean; anything else |
| +// is not, with what it names. With no channel build to compare, a present |
| +// link is not known. |
| +func liveLinkFact(optDir string, route registry.Route, build string, channelErr error) moves.Fact { |
| + link := release.LiveLink(optDir, route.Host) |
| + fi, err := os.Lstat(link) |
| + switch { |
| + case errors.Is(err, os.ErrNotExist): |
| + return moves.Fact{Known: true, OK: true} |
| + case err != nil: |
| + return moves.Fact{Detail: err.Error()} |
| + case fi.Mode()&os.ModeSymlink == 0: |
| + return moves.Fact{Known: true, Detail: fmt.Sprintf("%s is not a symlink (%s)", link, fi.Mode().Type())} |
| + } |
| + target, err := os.Readlink(link) |
| + if err != nil { |
| + return moves.Fact{Detail: err.Error()} |
| + } |
| + if channelErr != nil || build == "" { |
| + return moves.Fact{Detail: fmt.Sprintf("%s names %s; the channel could not be resolved: %v", link, target, channelErr)} |
| + } |
| + want := release.VersionPath(optDir, release.RouteAccount(route), build, route.App) |
| + return moves.Fact{Known: true, OK: target == want, Detail: target} |
| +} |
| + |
| +// absentFact is a path that must not exist, read by lstat. |
| +func absentFact(p string) moves.Fact { |
| + fi, err := os.Lstat(p) |
| + switch { |
| + case errors.Is(err, os.ErrNotExist): |
| + return moves.Fact{Known: true, OK: true} |
| + case err != nil: |
| + return moves.Fact{Detail: err.Error()} |
| + } |
| + return moves.Fact{Known: true, Detail: "present (" + fi.Mode().Type().String() + ")"} |
| +} |
| + |
| +// parkDirFact is the host's park directory: absent, or holding only |
| +// directories of other moves. |
| +func parkDirFact(dir, id string) moves.Fact { |
| + ents, err := os.ReadDir(dir) |
| + switch { |
| + case errors.Is(err, os.ErrNotExist): |
| + return moves.Fact{Known: true, OK: true} |
| + case err != nil: |
| + return moves.Fact{Detail: err.Error()} |
| + } |
| + var names []string |
| + ok := true |
| + for _, e := range ents { |
| + names = append(names, e.Name()) |
| + if !e.IsDir() || e.Name() == id { |
| + ok = false |
| + } |
| + } |
| + return moves.Fact{Known: true, OK: ok, Detail: strings.Join(names, ", ")} |
| +} |
| + |
| +// prefixFacts lists bucket/prefix by version: the current objects, keyed |
| +// relative to the prefix, delete markers and lock.json left out; lock says |
| +// whether a current lock.json is there. |
| +func prefixFacts(ctx context.Context, vs lease.VersionedStore, bucket, prefix string) (moves.PrefixFacts, bool) { |
| + vers, err := vs.ListVersions(ctx, bucket, prefix) |
| + if err != nil { |
| + return moves.PrefixFacts{}, false |
| + } |
| + out := moves.PrefixFacts{Known: true} |
| + lock := false |
| + for _, v := range vers { |
| + if !v.IsLatest || v.DeleteMarker || !strings.HasPrefix(v.Key, prefix) { |
| + continue |
| + } |
| + rel := strings.TrimPrefix(v.Key, prefix) |
| + if rel == "lock.json" { |
| + lock = true |
| + continue |
| + } |
| + if out.Objects == nil { |
| + out.Objects = map[string]moves.PrefixObject{} |
| + } |
| + out.Objects[rel] = moves.PrefixObject{ETag: v.ETag, Size: v.Size, VersionID: v.VersionID, LastModified: v.LastModified.UTC()} |
| + } |
| + return out, lock |
| +} |
diff --git a/internal/host/movesoffline_reports.go b/internal/host/movesoffline_reports.go| new file mode 100644 |
| index 000000000..14b82e980 |
| --- /dev/null |
| +++ b/internal/host/movesoffline_reports.go |
| @@ -0,0 +1,549 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "io/fs" |
| + "log" |
| + "os" |
| + "path/filepath" |
| + "sort" |
| + "strings" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/moves/archive" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/status" |
| +) |
| + |
| +// The boxes' reports of a self-serve move (self-serve moves plan 2 Task 6): |
| +// what a participant writes before anything is copied and again before the |
| +// stop. Every report is evidence for the console, never a decision: the |
| +// box reports facts and the console words its refusals. |
| +// |
| +// - The stager (the first source participant) probes the staging lane |
| +// once per move while the move is offline-queued and writes the lane |
| +// report (role lane). |
| +// - Every source participant runs the precheck for the record's |
| +// precheck_at and writes it (role source, phase precheck, Request). |
| +// - The target participant writes its ready report (role target, phase |
| +// ready, Request) for the same stamp (offlineTargetReady, |
| +// movesoffline_land.go). |
| +// |
| +// In offline-staging the source and the target report again whenever the |
| +// record's precheck_at is newer than the Request of their last report, so a |
| +// precheck the console asks for before the cut is answered fresh. |
| +// |
| +// The status tick publishes each instance's facts in the box's private |
| +// status report (hostFacts): the row's lifetime, the replica's size and the |
| +// files' size as the archive would count them. |
| + |
| +// offlineQueuedStep is a participant's part of an offline-queued move: the |
| +// lane probe (the stager), the precheck (a source) and the ready report (the |
| +// target). Nothing is copied, held or fenced. |
| +func (w *movesWorker) offlineQueuedStep(ctx context.Context, s release.Store, rec provision.Record, source, target, stager bool) error { |
| + if w.cfg.Box == "" { |
| + return errors.New("offline move: this box has no label to report under") |
| + } |
| + var errs []error |
| + if stager { |
| + errs = append(errs, w.offlineLaneStep(ctx, s, rec)) |
| + } |
| + errs = append(errs, w.offlineReReport(ctx, s, rec, source, target)) |
| + return errors.Join(errs...) |
| +} |
| + |
| +// offlineReReport writes the source's precheck and the target's ready |
| +// report for the record's precheck_at, each only when this box's last |
| +// report of its role does not already answer it (its Request is not |
| +// older than the stamp). |
| +func (w *movesWorker) offlineReReport(ctx context.Context, s release.Store, rec provision.Record, source, target bool) error { |
| + m := rec.Move |
| + if m.PrecheckAt == nil || (!source && !target) { |
| + return nil |
| + } |
| + stamp := m.PrecheckAt.UTC() |
| + var errs []error |
| + if source { |
| + if answered, err := w.answers(ctx, s, rec, moves.OfflineRoleSource, moves.OfflinePhasePrecheck, stamp); err != nil { |
| + errs = append(errs, err) |
| + } else if !answered { |
| + errs = append(errs, w.offlinePrecheckReport(ctx, s, rec, stamp)) |
| + } |
| + } |
| + if target { |
| + if answered, err := w.answers(ctx, s, rec, moves.OfflineRoleTarget, moves.OfflinePhaseReady, stamp); err != nil { |
| + errs = append(errs, err) |
| + } else if !answered { |
| + errs = append(errs, w.offlineReadyReport(ctx, s, rec, stamp)) |
| + } |
| + } |
| + return errors.Join(errs...) |
| +} |
| + |
| +// answers reports whether this box's own report of role for the move is a |
| +// phase report whose Request is not older than stamp. |
| +func (w *movesWorker) answers(ctx context.Context, s release.Store, rec provision.Record, role, phase string, stamp time.Time) (bool, error) { |
| + own, err := w.ownReport(ctx, s, rec, role) |
| + if err != nil || own == nil { |
| + return false, err |
| + } |
| + return own.Phase == phase && !own.Request.Before(stamp), nil |
| +} |
| + |
| +// ownReport is this box's report of role for the move, nil when it has |
| +// written none. One that does not parse is an error. |
| +func (w *movesWorker) ownReport(ctx context.Context, s release.Store, rec provision.Record, role string) (*moves.OfflineReport, error) { |
| + key := moves.OfflineReportKey(rec.Host, rec.Move.ID, role, w.cfg.Box) |
| + if key == "" { |
| + return nil, fmt.Errorf("offline move %s: %s has no %s report key", rec.Move.ID, rec.Host, role) |
| + } |
| + raw, _, err := s.Get(ctx, key) |
| + if errors.Is(err, os.ErrNotExist) { |
| + return nil, nil |
| + } |
| + if err != nil { |
| + return nil, err |
| + } |
| + var r moves.OfflineReport |
| + if err := json.Unmarshal(raw, &r); err != nil { |
| + return nil, fmt.Errorf("%s: %w", key, err) |
| + } |
| + return &r, nil |
| +} |
| + |
| +// selfServeReport is a report of this box for the move in role and phase, |
| +// stamped now and with this box's identity. |
| +func (w *movesWorker) selfServeReport(rec provision.Record, role, phase string) moves.OfflineReport { |
| + cfg := w.cfg |
| + return moves.OfflineReport{Op: rec.Move.ID, Host: rec.Host, Box: cfg.Box, Role: role, Phase: phase, At: cfg.now().UTC(), |
| + BoxID: cfg.BoxID, InstanceID: cfg.InstanceID} |
| +} |
| + |
| +// offlineLaneStep is the stager's lane probe, once per move: a box that has |
| +// written its lane report for the move writes nothing more. The probe is |
| +// the existing Phase 1 one (laneProbe: put-if-absent, list and |
| +// delete-version of moves.LaneKey(id)), run against the staging bucket the |
| +// record fixed. |
| +func (w *movesWorker) offlineLaneStep(ctx context.Context, s release.Store, rec provision.Record) error { |
| + m := rec.Move |
| + own, err := w.ownReport(ctx, s, rec, moves.OfflineRoleLane) |
| + if err != nil { |
| + return fmt.Errorf("offline move %s: read the lane report: %w", m.ID, err) |
| + } |
| + if own != nil { |
| + return nil |
| + } |
| + lane := moves.LaneReport{Bucket: m.StageBucket, Region: m.StageRegion, Key: moves.LaneKey(m.ID)} |
| + switch { |
| + case m.StageBucket == "": |
| + lane.Error = "the record names no staging bucket" |
| + case w.cfg.LaneProbe == nil: |
| + lane.Error = "this box has no lane probe" |
| + default: |
| + if err := w.cfg.LaneProbe(ctx, m.StageBucket, m.StageRegion, m.ID); err != nil { |
| + lane.Error = err.Error() |
| + } else { |
| + lane.OK = true |
| + } |
| + } |
| + r := w.selfServeReport(rec, moves.OfflineRoleLane, moves.OfflinePhaseLane) |
| + r.Lane = &lane |
| + log.Printf("moves: %s: offline move %s: lane probe of %s/%s: ok %v %s", rec.Host, m.ID, lane.Bucket, lane.Key, lane.OK, lane.Error) |
| + return w.putOfflineReport(ctx, s, r) |
| +} |
| + |
| +// offlinePrecheckReport writes the source's precheck for stamp. |
| +func (w *movesWorker) offlinePrecheckReport(ctx context.Context, s release.Store, rec provision.Record, stamp time.Time) error { |
| + p, err := w.offlineSourcePrecheck(ctx, rec) |
| + if err != nil { |
| + return fmt.Errorf("offline move %s: precheck %s: %w", rec.Move.ID, rec.Host, err) |
| + } |
| + r := w.selfServeReport(rec, moves.OfflineRoleSource, moves.OfflinePhasePrecheck) |
| + r.Request, r.Precheck, r.Layout = stamp, &p, p.Layout |
| + return w.putOfflineReport(ctx, s, r) |
| +} |
| + |
| +// offlineLayout is the row's layout as a report names it. |
| +func (w *movesWorker) offlineLayout(rt *registry.Route) string { |
| + if rt != nil && provision.NestedPaths(w.cfg.DataDir, *rt) { |
| + return moves.LayoutNested |
| + } |
| + return moves.LayoutFlat |
| +} |
| + |
| +// precheckFilesBudget is how long the precheck's and the status tick's |
| +// file measurement may walk one instance's files. |
| +const precheckFilesBudget = 60 * time.Second |
| + |
| +// offlineSourcePrecheck is the source's facts for the host (moves. |
| +// SourcePrecheck): a dry run of the archive (archive.Build into |
| +// io.Discard over the seal's own selection, offlineFilesOf), the layout, |
| +// the stray entries beside the files directory, the build the row runs and |
| +// whether root's stage path is armed. Only a row that cannot be read is an |
| +// error; every other failure is a fact that is not known. |
| +func (w *movesWorker) offlineSourcePrecheck(ctx context.Context, rec provision.Record) (moves.SourcePrecheck, error) { |
| + cfg := w.cfg |
| + host := rec.Host |
| + rt, err := w.offlineRow(host) |
| + if err != nil { |
| + return moves.SourcePrecheck{}, err |
| + } |
| + p := moves.SourcePrecheck{Layout: w.offlineLayout(rt), StagePathArmed: armedFact(cfg.StagePathArmed)} |
| + files, envPath, _, stray, err := w.offlineFilesOf(host, rt) |
| + if err != nil { |
| + p.Strays = moves.StraysFact{Detail: err.Error()} |
| + p.ArchiveEligible = moves.Fact{Detail: err.Error()} |
| + } else { |
| + p.Strays = moves.StraysFact{Known: true, Entries: strayEntries(stray)} |
| + // The instance is serving: a file that changes while it is read is |
| + // read again, a few times, before the archive's refusal stands. |
| + _, entries, bytes, berr := archive.Build(files, envPath, io.Discard) |
| + for try := 0; try < 2 && errors.Is(berr, archive.ErrChanged); try++ { |
| + _, entries, bytes, berr = archive.Build(files, envPath, io.Discard) |
| + } |
| + if berr == nil { |
| + p.ArchiveEligible = moves.Fact{Known: true, OK: true} |
| + p.FilesBytes, p.FilesEntries = bytes, len(entries) |
| + } else { |
| + // The archive's own reason; the sizes are the bounded count. |
| + p.ArchiveEligible = moves.Fact{Known: true, Detail: berr.Error()} |
| + p.FilesBytes, p.FilesEntries, _, _, _ = measureFiles(ctx, files, envPath, precheckFilesBudget, archive.Cap) |
| + } |
| + } |
| + if cfg.LiveVersion != nil { |
| + v, err := cfg.LiveVersion(host) |
| + if err != nil { |
| + w.notice(host+"#offline-precheck-build", "moves: %s: offline move %s: the live build cannot be read: %v", host, rec.Move.ID, err) |
| + } |
| + p.Build = v |
| + } |
| + return p, nil |
| +} |
| + |
| +// armedFact is a probe that answers nil for unknown, as a Fact. |
| +func armedFact(probe func() *bool) moves.Fact { |
| + if probe == nil { |
| + return moves.Fact{} |
| + } |
| + v := probe() |
| + if v == nil { |
| + return moves.Fact{} |
| + } |
| + return moves.Fact{Known: true, OK: *v} |
| +} |
| + |
| +// strayEntries describes each stray path, sorted: its kind, and for a |
| +// directory its bytes and entries, read by lstat and directory listings |
| +// only, never by following a link. |
| +func strayEntries(paths []string) []moves.StrayEntry { |
| + sorted := append([]string(nil), paths...) |
| + sort.Strings(sorted) |
| + out := make([]moves.StrayEntry, 0, len(sorted)) |
| + for _, p := range sorted { |
| + out = append(out, strayEntry(p)) |
| + } |
| + return out |
| +} |
| + |
| +func strayEntry(p string) moves.StrayEntry { |
| + e := moves.StrayEntry{Path: p} |
| + st, err := os.Lstat(p) |
| + if err != nil { |
| + e.Kind, e.Detail = moves.StrayOther, err.Error() |
| + return e |
| + } |
| + switch { |
| + case st.Mode().IsRegular(): |
| + e.Kind, e.Bytes, e.Entries, e.Known = moves.StrayFile, st.Size(), 1, true |
| + case st.Mode()&os.ModeSymlink != 0: |
| + e.Kind, e.Known = moves.StraySymlink, true |
| + case st.IsDir(): |
| + e.Kind = moves.StrayDir |
| + err := filepath.WalkDir(p, func(q string, d fs.DirEntry, err error) error { |
| + if err != nil { |
| + return err |
| + } |
| + if q == p { |
| + return nil |
| + } |
| + e.Entries++ |
| + if d.Type().IsRegular() { |
| + if info, err := d.Info(); err == nil { |
| + e.Bytes += info.Size() |
| + } |
| + } |
| + return nil |
| + }) |
| + if err != nil { |
| + e.Detail = err.Error() |
| + } else { |
| + e.Known = true |
| + } |
| + default: |
| + e.Kind, e.Known = moves.StrayOther, true |
| + } |
| + return e |
| +} |
| + |
| +// removeEmptyStrays is the start's part of the precheck (a self-serve move, |
| +// in offline-stopping, once the stop holds): every stray entry that is an |
| +// empty directory is removed, as the operators did by hand (the app makes |
| +// it again on the target); anything else is left where it is. A directory |
| +// that is not empty, or whose listing or removal fails, is in left with |
| +// why. Nothing outside the stray paths is touched. |
| +func (w *movesWorker) removeEmptyStrays(host string, rt *registry.Route) (removed []string, left []moves.StrayEntry, err error) { |
| + _, _, _, stray, err := w.offlineFilesOf(host, rt) |
| + if err != nil { |
| + return nil, nil, err |
| + } |
| + sort.Strings(stray) |
| + for _, p := range stray { |
| + st, err := os.Lstat(p) |
| + if err != nil || !st.IsDir() { |
| + continue // only an empty directory is the start's to remove |
| + } |
| + ents, err := os.ReadDir(p) |
| + switch { |
| + case err != nil: |
| + left = append(left, moves.StrayEntry{Path: p, Kind: moves.StrayDir, Detail: err.Error()}) |
| + case len(ents) > 0: |
| + left = append(left, moves.StrayEntry{Path: p, Kind: moves.StrayDir, Entries: len(ents), Known: true, Detail: "not empty"}) |
| + default: |
| + if err := os.Remove(p); err != nil { |
| + left = append(left, moves.StrayEntry{Path: p, Kind: moves.StrayDir, Known: true, Detail: err.Error()}) |
| + continue |
| + } |
| + log.Printf("moves: %s: removed the empty directory %s beside the files directory", host, p) |
| + removed = append(removed, p) |
| + } |
| + } |
| + return removed, left, nil |
| +} |
| + |
| +// measureFiles counts what the offline move's archive would carry of files |
| +// and envPath by the archive's own rules (regular files, the env file |
| +// included, the files marker left out, nothing followed through a link), |
| +// without stopping at the cap: past capBytes it records overCap and goes |
| +// on, until budget has passed or the count reaches twice capBytes, where it |
| +// stops and records truncated. A missing files directory is an empty one. |
| +// An error is a walk that could not read part of the tree. |
| +func measureFiles(ctx context.Context, files, envPath string, budget time.Duration, capBytes int64) (bytes int64, entries int, overCap, truncated bool, err error) { |
| + deadline := time.Now().Add(budget) |
| + stop := errors.New("stop") |
| + add := func(n int64) bool { |
| + bytes += n |
| + entries++ |
| + if bytes > capBytes { |
| + overCap = true |
| + } |
| + if bytes >= 2*capBytes { |
| + truncated = true |
| + return false |
| + } |
| + return true |
| + } |
| + if envPath != "" { |
| + if st, err := os.Lstat(envPath); err == nil && st.Mode().IsRegular() { |
| + if !add(st.Size()) { |
| + return bytes, entries, overCap, truncated, nil |
| + } |
| + } else if err != nil && !errors.Is(err, os.ErrNotExist) { |
| + return bytes, entries, overCap, truncated, err |
| + } |
| + } |
| + st, lerr := os.Lstat(files) |
| + switch { |
| + case errors.Is(lerr, os.ErrNotExist): |
| + return bytes, entries, overCap, truncated, nil |
| + case lerr != nil: |
| + return bytes, entries, overCap, truncated, lerr |
| + case !st.IsDir(): |
| + return bytes, entries, overCap, truncated, fmt.Errorf("%s is not a directory", files) |
| + } |
| + werr := filepath.WalkDir(files, func(p string, d fs.DirEntry, err error) error { |
| + if err != nil { |
| + return err |
| + } |
| + if ctx.Err() != nil || time.Now().After(deadline) { |
| + truncated = true |
| + return stop |
| + } |
| + if p == files { |
| + return nil |
| + } |
| + if rel, _ := filepath.Rel(files, p); filepath.ToSlash(rel) == archive.MarkerName { |
| + if d.IsDir() { |
| + return fs.SkipDir |
| + } |
| + return nil |
| + } |
| + if !d.Type().IsRegular() { |
| + return nil |
| + } |
| + info, err := d.Info() |
| + if err != nil { |
| + return err |
| + } |
| + if !add(info.Size()) { |
| + return stop |
| + } |
| + return nil |
| + }) |
| + if werr != nil && !errors.Is(werr, stop) { |
| + return bytes, entries, overCap, truncated, werr |
| + } |
| + return bytes, entries, overCap, truncated, nil |
| +} |
| + |
| +// ---- the status tick's per-host facts ---- |
| + |
| +// replicaListing is one listing of this box's replica prefix, taken once per |
| +// status tick and kept, with when it was taken, for a tick whose listing |
| +// fails. |
| +type replicaListing struct { |
| + at time.Time |
| + sizes map[string][2]int64 // host → bytes, objects |
| +} |
| + |
| +// hostFactsBudget bounds the whole collection of one tick's host facts, |
| +// beside each host's own file budget. |
| +const hostFactsBudget = 2 * time.Minute |
| + |
| +// hostFacts is the status tick's per-host block (status.HostFacts) for |
| +// every instance row this box holds: the row's lifetime, the replica's |
| +// bytes and objects from one listing of the box's replica prefix per tick, |
| +// and the files' bytes and entries as the archive would count them, with |
| +// the strays beside the files directory. Known only when every part was |
| +// collected this tick. nil when the box cannot take part in an offline move |
| +// or holds no instance row. |
| +func (w *movesWorker) hostFacts(ctx context.Context, routes []registry.Route, now time.Time) map[string]status.HostFacts { |
| + if !w.cfg.OfflineCapable { |
| + return nil |
| + } |
| + var rows []registry.Route |
| + for _, rt := range routes { |
| + if rt.Platform || rt.Kind == "static" || rt.IsSidecar() || !provision.ValidHostShape(rt.Host) { |
| + continue |
| + } |
| + rows = append(rows, rt) |
| + } |
| + if len(rows) == 0 { |
| + return nil |
| + } |
| + ctx, cancel := context.WithTimeout(ctx, hostFactsBudget) |
| + defer cancel() |
| + listing, listErr := w.listReplicas(ctx, now) |
| + out := make(map[string]status.HostFacts, len(rows)) |
| + for _, rt := range rows { |
| + f := status.HostFacts{CollectedAt: now.Unix(), Incarnation: rt.Incarnation} |
| + var errs []string |
| + if listing != nil { |
| + sz := listing.sizes[rt.Host] |
| + f.ReplicaBytes, f.ReplicaObjects, f.ReplicaCollectedAt = sz[0], int(sz[1]), listing.at.Unix() |
| + } |
| + if listErr != nil { |
| + errs = append(errs, "replica: "+listErr.Error()) |
| + } |
| + files, envPath, _, stray, err := w.offlineFilesOf(rt.Host, &rt) |
| + if err != nil { |
| + errs = append(errs, "files: "+err.Error()) |
| + } else { |
| + budget := precheckFilesBudget |
| + if dl, ok := ctx.Deadline(); ok && time.Until(dl) < budget { |
| + budget = time.Until(dl) |
| + } |
| + var merr error |
| + f.FilesBytes, f.FilesEntries, f.OverCap, f.Truncated, merr = measureFiles(ctx, files, envPath, budget, archive.Cap) |
| + if merr != nil { |
| + errs = append(errs, "files: "+merr.Error()) |
| + } |
| + sort.Strings(stray) |
| + f.Strays = stray |
| + } |
| + f.Known = len(errs) == 0 |
| + f.Err = strings.Join(errs, "; ") |
| + out[rt.Host] = f |
| + } |
| + return out |
| +} |
| + |
| +// listReplicas lists this box's replica prefix once and sums it per host |
| +// (lock.json aside). A listing that fails returns the last good one, with |
| +// the time it was taken, and the error. |
| +func (w *movesWorker) listReplicas(ctx context.Context, now time.Time) (*replicaListing, error) { |
| + cfg := w.cfg |
| + w.mu.Lock() |
| + last := w.replicas |
| + w.mu.Unlock() |
| + if cfg.OwnBucket == "" { |
| + return last, errors.New("this box has no hibernate bucket") |
| + } |
| + mirror, err := w.offlineMirror(ctx, cfg.OwnBucketRegion) |
| + if err != nil { |
| + return last, err |
| + } |
| + prefix := strings.TrimSuffix(cfg.OwnPrefix, "/") + "/" |
| + if cfg.OwnPrefix == "" { |
| + prefix = "" |
| + } |
| + objs, err := mirror.List(ctx, cfg.OwnBucket, prefix) |
| + if err != nil { |
| + return last, fmt.Errorf("list %s/%s: %w", cfg.OwnBucket, prefix, err) |
| + } |
| + l := &replicaListing{at: now, sizes: map[string][2]int64{}} |
| + for k, o := range objs { |
| + rest := strings.TrimPrefix(k, prefix) |
| + host, rel, ok := strings.Cut(rest, "/") |
| + if !ok || rel == "lock.json" || activator.HibernatePrefix(cfg.OwnPrefix, host)+"/"+rel != k { |
| + continue |
| + } |
| + sz := l.sizes[host] |
| + sz[0] += o.Size |
| + sz[1]++ |
| + l.sizes[host] = sz |
| + } |
| + w.mu.Lock() |
| + w.replicas = l |
| + w.mu.Unlock() |
| + return l, nil |
| +} |
| + |
| +// moveLanesEnv names the staging buckets this box has a lane to, comma or |
| +// space separated (BoxMetrics.Lanes). |
| +const moveLanesEnv = "CARLOS_MOVE_LANES" |
| + |
| +// moveLanes reads moveLanesEnv: trimmed, empty names dropped, sorted and |
| +// without repeats; nil when unset. |
| +func moveLanes(getenv func(string) string) []string { |
| + fields := strings.FieldsFunc(getenv(moveLanesEnv), func(r rune) bool { return r == ',' || r == ' ' || r == '\t' || r == '\n' }) |
| + seen := map[string]bool{} |
| + var out []string |
| + for _, f := range fields { |
| + if f = strings.TrimSpace(f); f != "" && !seen[f] { |
| + seen[f] = true |
| + out = append(out, f) |
| + } |
| + } |
| + sort.Strings(out) |
| + return out |
| +} |
| + |
| +// selfServeStatusProbes fills the status tick's self-serve block: the box id |
| +// the edges writer publishes, the lanes, and, from the armed moves worker, |
| +// the per-host facts. A box with no armed worker publishes the id and the |
| +// lanes alone. |
| +func selfServeStatusProbes(ctx context.Context, p *status.Probes, w *movesWorker, boxID string, getenv func(string) string, routes []registry.Route, now time.Time) { |
| + p.BoxID = boxID |
| + p.Lanes = moveLanes(getenv) |
| + if w != nil { |
| + p.Hosts = w.hostFacts(ctx, routes, now) |
| + } |
| +} |
diff --git a/internal/host/movesoffline_stage.go b/internal/host/movesoffline_stage.go| index c3cb2767c..fdae13bcc 100644 |
| --- a/internal/host/movesoffline_stage.go |
| +++ b/internal/host/movesoffline_stage.go |
| @@ -91,24 +91,130 @@ func (w *movesWorker) offlineStageStep(ctx context.Context, s release.Store, rec |
| if prev != nil && prev.StageRequest.Equal(request) && (prev.Phase == moves.OfflinePhaseStaged || prev.Phase == moves.OfflinePhaseFailed) { |
| return nil // this request has had its round |
| } |
| - r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleStager, Phase: moves.OfflinePhaseStaged, StageRequest: request} |
| - p, err := w.offlineStageRound(ctx, s, rec) |
| + r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleStager, Phase: moves.OfflinePhaseStaged, StageRequest: request, |
| + BoxID: cfg.BoxID, InstanceID: cfg.InstanceID} |
| + book := w.stageBookFor(rec, prev) |
| + book.Rounds++ |
| + p, round, err := w.offlineStageRound(ctx, s, rec, book.observe) |
| + if err == nil && cfg.stageFault != nil { |
| + err = cfg.stageFault("round") |
| + if err != nil { |
| + return err // a crash after the round, before its report |
| + } |
| + } |
| r.At = cfg.now().UTC() |
| + book.fill(&r, m, request, round) |
| if err != nil { |
| // The progress stays what the last good round found; the error is a |
| - // box diagnostic the status view shows as it is. |
| + // box diagnostic the status view shows as it is. The book is as the |
| + // round's events left it: the next complete round reconciles it. |
| if prev != nil { |
| p = stageProgress{replicaObjects: prev.ReplicaObjects, replicaBytes: prev.ReplicaBytes, stagedObjects: prev.StagedObjects, stagedBytes: prev.StagedBytes} |
| } |
| p.fill(&r) |
| r.Phase, r.StageError = moves.OfflinePhaseFailed, err.Error() |
| - return errors.Join(fmt.Errorf("offline move %s: staging %s: %w", m.ID, host, err), writeOfflineReport(ctx, s, r)) |
| + return errors.Join(fmt.Errorf("offline move %s: staging %s: %w", m.ID, host, err), w.putOfflineReport(ctx, s, r)) |
| } |
| p.fill(&r) |
| if p.roundObjects > 0 { |
| log.Printf("moves: %s: offline move %s: staged round: %d object(s) changed, %d bytes copied; %d of %d objects staged", host, m.ID, p.roundObjects, p.roundBytes, p.stagedObjects, p.replicaObjects) |
| } |
| - return writeOfflineReport(ctx, s, r) |
| + return w.putOfflineReport(ctx, s, r) |
| +} |
| + |
| +// stageBook is the stager's book of one move (self-serve moves plan 2 Task |
| +// 6): every object the move's rounds staged, keyed relative to the staging |
| +// prefix, as the mirror reported each decision (lease.CopyEvent), so the |
| +// staged report says what this move put there and not what a listing finds. |
| +// It lives across rounds in the worker (movesWorker.books, under the pass |
| +// lock), seeded at a move's first round on this worker from its own last |
| +// stager report when that report validates for the move, and empty |
| +// otherwise. |
| +type stageBook struct { |
| + MoveID string |
| + Objects map[string]moves.StagedObject |
| + Rounds int |
| +} |
| + |
| +// stageBookFor is the book of rec's move, seeded from prev, this box's last |
| +// stager report, when it is this move's book of the record's destination: |
| +// Op the move, Box this box, role stager, StagedKnown, and StagedBucket and |
| +// StagedRegion the record's. Anything else seeds an empty book, logged. |
| +func (w *movesWorker) stageBookFor(rec provision.Record, prev *moves.OfflineReport) *stageBook { |
| + m := rec.Move |
| + w.mu.Lock() |
| + defer w.mu.Unlock() |
| + if b := w.books[rec.Host]; b != nil && b.MoveID == m.ID { |
| + return b |
| + } |
| + b := &stageBook{MoveID: m.ID, Objects: map[string]moves.StagedObject{}} |
| + switch { |
| + case prev == nil: |
| + case prev.Op == m.ID && prev.Box == w.cfg.Box && prev.Role == moves.OfflineRoleStager && prev.StagedKnown && |
| + prev.StagedBucket == m.StageBucket && prev.StagedRegion == m.StageRegion: |
| + for k, o := range prev.Staged { |
| + b.Objects[k] = o |
| + } |
| + b.Rounds = prev.Rounds |
| + default: |
| + log.Printf("moves: %s: offline move %s: the last stager report is not this move's book of %s (%s); the book starts empty", rec.Host, m.ID, m.StageBucket, m.StageRegion) |
| + } |
| + if w.books == nil { |
| + w.books = map[string]*stageBook{} |
| + } |
| + w.books[rec.Host] = b |
| + return b |
| +} |
| + |
| +// observe applies one of the round's decisions. A copy or a skip records |
| +// the object as the mirror saw it (the copy's own ETag and version, or the |
| +// receipt's); a deferral marks the entry deferred without changing it, or |
| +// records a deferred entry with nothing copied; a deletion forgets it. The |
| +// round's one Complete event reconciles the book with the round's key sets: |
| +// an entry whose key is in neither set (a deletion a crashed round made but |
| +// never reported) is dropped, and so is a deferred entry never copied whose |
| +// key left the source. The round calls observe one event at a time. |
| +func (b *stageBook) observe(e lease.CopyEvent) { |
| + switch { |
| + case e.Complete: |
| + src, dst := map[string]bool{}, map[string]bool{} |
| + for _, k := range e.SourceKeys { |
| + src[k] = true |
| + } |
| + for _, k := range e.DestKeys { |
| + dst[k] = true |
| + } |
| + for k, o := range b.Objects { |
| + if (!src[k] && !dst[k]) || (o.Deferred && o.ETag == "" && !src[k]) { |
| + delete(b.Objects, k) |
| + } |
| + } |
| + case e.Deleted: |
| + delete(b.Objects, e.Rel) |
| + case e.Deferred: |
| + if o, ok := b.Objects[e.Rel]; ok { |
| + o.Deferred = true |
| + b.Objects[e.Rel] = o |
| + } else { |
| + b.Objects[e.Rel] = moves.StagedObject{Deferred: true} |
| + } |
| + default: |
| + b.Objects[e.Rel] = moves.StagedObject{Size: e.SrcSize, ETag: e.DstETag, VersionID: e.DstVersionID} |
| + } |
| +} |
| + |
| +// fill writes the book into a stager report of m for request: the whole |
| +// book, known, at the record's destination, with the round's copied and |
| +// deferred amounts and the rounds run. |
| +func (b *stageBook) fill(r *moves.OfflineReport, m *provision.MoveState, request time.Time, round lease.MirrorRound) { |
| + r.Staged = make(map[string]moves.StagedObject, len(b.Objects)) |
| + for k, o := range b.Objects { |
| + r.Staged[k] = o |
| + } |
| + r.StagedKnown, r.StagedBucket, r.StagedRegion = true, m.StageBucket, m.StageRegion |
| + r.StageRequest, r.StagedRequest = request, request |
| + r.RoundCopiedBytes, r.RoundDeferredBytes, r.RoundDeferredObjects = round.BytesCopied, round.BytesDeferred, round.Deferred |
| + r.Rounds = b.Rounds |
| } |
| // stageProgress is what one stage round found. |
| @@ -129,22 +235,26 @@ func (p stageProgress) fill(r *moves.OfflineReport) { |
| // exactly that bucket and region; a target that publishes anything else is |
| // refused, never followed. The round's delete phase reads the record again |
| // first and stops if the move has left staging (finding 1). |
| -func (w *movesWorker) offlineStageRound(ctx context.Context, s release.Store, rec provision.Record) (stageProgress, error) { |
| +// |
| +// observe is told every decision of the round (lease. |
| +// MirrorPrefixKeepObserved): the stager's book. |
| +func (w *movesWorker) offlineStageRound(ctx context.Context, s release.Store, rec provision.Record, observe func(lease.CopyEvent)) (stageProgress, lease.MirrorRound, error) { |
| cfg := w.cfg |
| host, m := rec.Host, rec.Move |
| + none := lease.MirrorRound{} |
| if m.StageBucket == "" { |
| - return stageProgress{}, errors.New("the record names no staging bucket") |
| + return stageProgress{}, none, errors.New("the record names no staging bucket") |
| } |
| tgt, err := w.offlineTarget(ctx, s, m) |
| if err != nil { |
| - return stageProgress{}, err |
| + return stageProgress{}, none, err |
| } |
| if tgt.DataBucket != m.StageBucket || tgt.DataBucketRegion != m.StageRegion { |
| - return stageProgress{}, fmt.Errorf("the target now publishes %s (%s), not the staging bucket %s (%s) the stage fixed", tgt.DataBucket, tgt.DataBucketRegion, m.StageBucket, m.StageRegion) |
| + return stageProgress{}, none, fmt.Errorf("the target now publishes %s (%s), not the staging bucket %s (%s) the stage fixed", tgt.DataBucket, tgt.DataBucketRegion, m.StageBucket, m.StageRegion) |
| } |
| bucket, loc, _, err := cfg.tombstoneAt(host) |
| if err != nil { |
| - return stageProgress{}, err |
| + return stageProgress{}, none, err |
| } |
| // The activator promotes a host's moves/<host>/ in its OWN bucket over |
| // its replica on every restore (promoteStagedReplica: staging wins). A |
| @@ -152,11 +262,11 @@ func (w *movesWorker) offlineStageRound(ctx context.Context, s release.Store, re |
| // instance restore from a copy behind its replica, so the round refuses |
| // (the stage door refuses it first). |
| if m.StageBucket == bucket { |
| - return stageProgress{}, fmt.Errorf("the staging bucket is this box's own bucket %s; a staged copy there would replace the serving replica on its next wake", bucket) |
| + return stageProgress{}, none, fmt.Errorf("the staging bucket is this box's own bucket %s; a staged copy there would replace the serving replica on its next wake", bucket) |
| } |
| mirror, err := w.offlineMirror(ctx, m.StageRegion) |
| if err != nil { |
| - return stageProgress{}, err |
| + return stageProgress{}, none, err |
| } |
| gated := &stageDeleteGate{MirrorStore: mirror, check: func(ctx context.Context) error { |
| cur, err := judgedRecord(ctx, s, w.cfg.Judge, rec.Account, rec.App, rec.Host) |
| @@ -171,17 +281,17 @@ func (w *movesWorker) offlineStageRound(ctx context.Context, s release.Store, re |
| srcPrefix := loc + "/" |
| dstPrefix := activator.MovesPrefix(host) |
| receipts := lease.Receipts{Bucket: bucket, Key: activator.MoveReceiptsKey(host)} |
| - round, err := lease.MirrorPrefix(ctx, gated, bucket, srcPrefix, m.StageBucket, dstPrefix, receipts) |
| + round, err := lease.MirrorPrefixKeepObserved(ctx, gated, bucket, srcPrefix, m.StageBucket, dstPrefix, receipts, nil, observe) |
| if err != nil { |
| - return stageProgress{}, err |
| + return stageProgress{}, round, err |
| } |
| source, err := mirror.List(ctx, bucket, srcPrefix) |
| if err != nil { |
| - return stageProgress{}, fmt.Errorf("list the replica: %w", err) |
| + return stageProgress{}, round, fmt.Errorf("list the replica: %w", err) |
| } |
| staged, err := mirror.List(ctx, m.StageBucket, dstPrefix) |
| if err != nil { |
| - return stageProgress{}, fmt.Errorf("list the staged copy: %w", err) |
| + return stageProgress{}, round, fmt.Errorf("list the staged copy: %w", err) |
| } |
| p := stageProgress{roundObjects: round.Changed, roundBytes: round.BytesCopied} |
| for k, o := range source { |
| @@ -196,7 +306,7 @@ func (w *movesWorker) offlineStageRound(ctx context.Context, s release.Store, re |
| p.stagedBytes += o.Size |
| } |
| } |
| - return p, nil |
| + return p, round, nil |
| } |
| // stageDeleteGate is a stage round's mirror with one check before the |
| @@ -237,12 +347,17 @@ func (w *movesWorker) offlineStagerStopped(ctx context.Context, s release.Store, |
| return fmt.Errorf("offline move %s: read the stager report: %w", m.ID, err) |
| } |
| r := moves.OfflineReport{Op: m.ID, Host: rec.Host, Box: w.cfg.Box, Role: moves.OfflineRoleStager, Phase: moves.OfflinePhaseStopped, |
| - At: w.cfg.now().UTC(), StageRequest: m.StagedAt.UTC()} |
| + At: w.cfg.now().UTC(), StageRequest: m.StagedAt.UTC(), StagedRequest: m.StagedAt.UTC(), BoxID: w.cfg.BoxID, InstanceID: w.cfg.InstanceID} |
| if prev != nil { |
| stageProgress{replicaObjects: prev.ReplicaObjects, replicaBytes: prev.ReplicaBytes, stagedObjects: prev.StagedObjects, |
| stagedBytes: prev.StagedBytes, roundObjects: prev.RoundObjects, roundBytes: prev.RoundBytes}.fill(&r) |
| + // The book of the last round, kept as it was. |
| + if prev.Op == m.ID && prev.StagedKnown && prev.StagedBucket == m.StageBucket && prev.StagedRegion == m.StageRegion { |
| + r.Staged, r.StagedKnown, r.StagedBucket, r.StagedRegion = prev.Staged, true, prev.StagedBucket, prev.StagedRegion |
| + r.RoundCopiedBytes, r.RoundDeferredBytes, r.RoundDeferredObjects, r.Rounds = prev.RoundCopiedBytes, prev.RoundDeferredBytes, prev.RoundDeferredObjects, prev.Rounds |
| + } |
| } |
| - return writeOfflineReport(ctx, s, r) |
| + return w.putOfflineReport(ctx, s, r) |
| } |
| // ownStagerReport is this box's stager report of the move, nil when it has |
| @@ -317,5 +432,5 @@ func (w *movesWorker) offlineUnstage(ctx context.Context, s release.Store, rec p |
| log.Printf("moves: %s: offline move %s: the staged copy in %s/%s is deleted", host, m.ID, m.StageBucket, prefix) |
| clearOfflineReceipts(ctx, mirror, cfg.OwnBucket, host) |
| r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleStager, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC()} |
| - return writeOfflineReport(ctx, s, r) |
| + return w.putOfflineReport(ctx, s, r) |
| } |
diff --git a/internal/host/movesoffline_stage_test.go b/internal/host/movesoffline_stage_test.go| index e685ad4a5..f13b0961e 100644 |
| --- a/internal/host/movesoffline_stage_test.go |
| +++ b/internal/host/movesoffline_stage_test.go |
| @@ -372,7 +372,7 @@ func TestOfflineAbortCleansARoundThatNeverReported(t *testing.T) { |
| sw.stage() |
| sw.stagerEdges(ie) |
| // The round's copies land; the box dies before its report. |
| - if _, err := ie.w.offlineStageRound(context.Background(), sw.s, sw.record()); err != nil { |
| + if _, _, err := ie.w.offlineStageRound(context.Background(), sw.s, sw.record(), nil); err != nil { |
| t.Fatal(err) |
| } |
| if len(sw.stagedRel()) == 0 { |
| @@ -419,7 +419,7 @@ func TestOfflineAbortWaitsForThePinnedStager(t *testing.T) { |
| } |
| // ie-b's round, which began before the abort, lands its copies now. |
| sw.fake.Put(sealIE, sealPrefix+"/"+offHost+"/scratch/late.bin", []byte("late")) |
| - if _, err := ieB.w.offlineStageRound(context.Background(), sw.s, sw.record()); err != nil { |
| + if _, _, err := ieB.w.offlineStageRound(context.Background(), sw.s, sw.record(), nil); err != nil { |
| t.Fatal(err) |
| } |
| if len(sw.stagedRel()) == 0 { |
diff --git a/internal/host/movespass.go b/internal/host/movespass.go| index e4e76452a..c0d1bad50 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -110,6 +110,15 @@ type movesWorker struct { |
| // lost names the hosts whose claim lists this box in Lost, as this pass |
| // read them (lostHere); rebuilt every pass under pass. |
| lost map[string]bool |
| + |
| + // The self-serve move's box state (self-serve moves plan 2 Task 6). |
| + // books is the stager's book per host (movesoffline_stage.go), kept |
| + // across rounds under pass; replaced logs a replaced box's line once per |
| + // move (offlineRound); replicas is the last replica listing the status |
| + // tick took (hostFacts), guarded by mu. |
| + books map[string]*stageBook |
| + replaced sync.Map // move id → true |
| + replicas *replicaListing |
| } |
| // staging is the memoized staging writer for a target bucket region. |
| @@ -329,6 +338,10 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, |
| // sets the same two values. |
| cfg.Box = metricsBoxLabel(os.Getenv, readMachineID()) |
| cfg.OptDir = release.OptDir() |
| + // The self-serve move pins a participant by its box id (the edges |
| + // writer's value, the label) and the instance its status report names. |
| + cfg.BoxID = cfg.Box |
| + cfg.InstanceID = cachedInstanceID() |
| w := newMovesWorker(cfg) |
| if cfg.Enabled { |
| armPhase1(&cfg, w, hib, dataDir, reg, svc, eb, writes, os.Getenv) |
| @@ -359,6 +372,26 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, |
| // for a record with an offline move. |
| func armOffline(cfg *moveCfg, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, writes hostwrite.Config, getenv func(string) string) { |
| cfg.OfflineCapable = offlineCapable(getenv, hibernationConfigured(getenv)) |
| + if cfg.BoxID == "" { |
| + cfg.BoxID = cfg.Box |
| + } |
| + // The self-serve move's lane probe reads the staging bucket by exact |
| + // version, whatever the Phase 1 switch says (armPhase1 sets the same |
| + // two when it runs, and they are kept). |
| + if cfg.Versioned == nil { |
| + cfg.Versioned = func(ctx context.Context, bucket, region string) (lease.VersionedStore, error) { |
| + return lease.NewVersionedS3(ctx, map[string]string{bucket: region}) |
| + } |
| + } |
| + if cfg.LaneProbe == nil { |
| + cfg.LaneProbe = laneProbe(cfg.Versioned) |
| + } |
| + if cfg.DiskFree == nil { |
| + cfg.DiskFree = diskBytes |
| + } |
| + if cfg.StagePathArmed == nil { |
| + cfg.StagePathArmed = stagePathActive |
| + } |
| cfg.Offline = newOfflineView(*cfg, holdStageRequester(dataDir)) |
| if svc != nil { |
| cfg.FenceAs = fenceAsWith(svc) |
| @@ -1001,6 +1034,21 @@ type moveCfg struct { |
| // keep view reconcile reads (offlineView.keep); the hold needs none. |
| OfflineCapable bool |
| Offline *offlineView |
| + // BoxID and InstanceID are this box's identity as a self-serve move |
| + // pins it (moves.ParticipantRole): the edges file's box id (the label, |
| + // as edgesFileLoop publishes it) and the cloud instance it runs on |
| + // (status.InstanceID, set by armMovesPass). armOffline sets BoxID to |
| + // Box when it is empty. |
| + BoxID string |
| + InstanceID string |
| + // StagePathArmed answers whether root's stage path (carlos-stage.path) |
| + // is active on this box, nil when it cannot tell: the source's precheck |
| + // and the target's ready report carry it. Nil hook: unknown. |
| + StagePathArmed func() *bool |
| + // stageFault is the stage round's test seam: called after a round's |
| + // copies with "round", before its report, an error being a crash there. |
| + // Nil in production. |
| + stageFault func(point string) error |
| // FenceAs installs, or meets again, an owned activator fence |
| // (activator.Service.FenceAndStopAs under a 2 s deadline, |
| // ErrStopUnproven counted as installed: fenceAsWith); LiftFenceAs lifts |
diff --git a/internal/host/movesreport.go b/internal/host/movesreport.go| index 1d35140c4..7e8f73ee5 100644 |
| --- a/internal/host/movesreport.go |
| +++ b/internal/host/movesreport.go |
| @@ -101,6 +101,10 @@ func t12NotQuiet(region string, until time.Time) string { |
| type moveFleet struct { |
| targets map[string]edgesFile |
| boxes map[string][]string |
| + // byBox is every edges file that publishes a data bucket, by its box |
| + // id (its label when it publishes none): a self-serve move's target is |
| + // its pinned participant's box (offlineTarget). |
| + byBox map[string]edgesFile |
| } |
| func loadMoveFleet(ctx context.Context, s release.Store) (moveFleet, error) { |
| @@ -108,7 +112,7 @@ func loadMoveFleet(ctx context.Context, s release.Store) (moveFleet, error) { |
| if err != nil { |
| return moveFleet{}, err |
| } |
| - f := moveFleet{targets: map[string]edgesFile{}, boxes: map[string][]string{}} |
| + f := moveFleet{targets: map[string]edgesFile{}, boxes: map[string][]string{}, byBox: map[string]edgesFile{}} |
| for _, body := range bodies { |
| var t edgesFile |
| if err := json.Unmarshal(body, &t); err != nil || t.V < 1 { |
| @@ -120,6 +124,11 @@ func loadMoveFleet(ctx context.Context, s release.Store) (moveFleet, error) { |
| if t.DataBucket == "" { |
| continue |
| } |
| + id := t.BoxID |
| + if id == "" { |
| + id = t.Box |
| + } |
| + f.byBox[id] = t |
| if prev, dup := f.targets[t.Placement]; dup && prev.Box <= t.Box { |
| continue |
| } |
diff --git a/internal/host/probes.go b/internal/host/probes.go| index dc42fd9ac..d4f4f4e43 100644 |
| --- a/internal/host/probes.go |
| +++ b/internal/host/probes.go |
| @@ -83,6 +83,17 @@ func restartPathActive() *bool { |
| return parseIsActive(string(out)) |
| } |
| +// stagePathUnit is root's stage trigger (carlos-stage.path): the self-serve |
| +// move's source precheck and target ready report say whether it is active, |
| +// because a build reaches the target through it after open. |
| +const stagePathUnit = "carlos-stage.path" |
| + |
| +// stagePathActive asks configPathActive's question about stagePathUnit. |
| +func stagePathActive() *bool { |
| + out, _ := exec.Command("systemctl", "is-active", stagePathUnit).Output() |
| + return parseIsActive(string(out)) |
| +} |
| + |
| // systemUpdatePathUnit is the unit whose liveness IS "an agent's fetched |
| // platform candidate reaches a root apply" (spec §4): the path unit watching |
| // /data/system-update-requested. Installed and armed by bootstrap alongside |