carlos / platform Public

host: the box resolves its instance id lazily, so a pinned participant takes its role once the metadata service answers

moveCfg.InstanceIDOf (production: cachedInstanceID through
throttledInstanceID, asked at most once a minute while it has no answer,
remembered once it has one) backs moveCfg.instance(), which the role
resolver, the replaced check, the keep view and every report's identity
read. A box with no id yet logs that it waits and takes no part; once the
id is known it takes its role and its reports carry the id.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

claude 2 days ago f43bd4f3e6c8b4906f212379894ba8f23473daba parent d59c4ff

6 files changed, +119 −11

internal/host/movesoffline.go +49 −5
internal/host/movesoffline_reports.go +1 −1
internal/host/movesoffline_reports_test.go +57 −0
internal/host/movesoffline_seal.go +3 −3
internal/host/movesoffline_stage.go +2 −2
internal/host/movespass.go +7 −0

diff --git a/internal/host/movesoffline.go b/internal/host/movesoffline.go

index fc01a42cd..b91258e94 100644

--- a/internal/host/movesoffline.go

+++ b/internal/host/movesoffline.go

@@ -118,12 +118,50 @@ func offlineHold(box moveCfg) func(rec provision.Record) (status, op string, ok

}

+// instance is this box's cloud instance id: InstanceIDOf's answer once it

+// has one, InstanceID until then.

+func (c moveCfg) instance() string {

+ if c.InstanceIDOf != nil {

+ if id := c.InstanceIDOf(); id != "" {

+ return id

+ }

+ return c.InstanceID

+// throttledInstanceID is read (cachedInstanceID in production) asked at most

+// once every `every` while it answers "", and remembered once it answers:

+// the instance id cannot change under a running process, and a metadata

+// service that does not answer costs up to two seconds a call, which the

+// keep view and every pass must not pay on each record.

+func throttledInstanceID(read func() string, every time.Duration, now func() time.Time) func() string {

+ var mu sync.Mutex

+ var id string

+ var last time.Time

+ return func() string {

+ mu.Lock()

+ if id != "" || (!last.IsZero() && now().Sub(last) < every) {

+ defer mu.Unlock()

+ return id

+ }

+ last = now()

+ mu.Unlock()

+ got := read()

+ mu.Lock()

+ defer mu.Unlock()

+ if got != "" && id == "" {

+ id = got

+ }

+ return id

+ }

// 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)

+ return moves.ParticipantRole(m, c.BoxID, c.instance(), c.Box, c.parkRegion(), c.ParkHomeRegion)

}

// releaseHeldSeq is the offline release in production: registry.

@@ -187,7 +225,7 @@ 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 {

- box := moveCfg{Box: cfg.Box, BoxID: cfg.BoxID, InstanceID: cfg.InstanceID, Region: cfg.Region, ParkHomeRegion: cfg.ParkHomeRegion}

+ box := moveCfg{Box: cfg.Box, BoxID: cfg.BoxID, InstanceID: cfg.InstanceID, InstanceIDOf: cfg.InstanceIDOf, Region: cfg.Region, ParkHomeRegion: cfg.ParkHomeRegion}

return &offlineView{home: cfg.ParkHomeRegion, region: cfg.parkRegion(), box: box, written: written, judge: cfg.Judge}

}

@@ -460,7 +498,13 @@ func (w *movesWorker) offlineRound(ctx context.Context, s release.Store, rec pro

if !m.Offline() {

return nil

}

- if moves.ParticipantReplaced(m, cfg.BoxID, cfg.InstanceID) {

+ if moves.ParticipantReplaced(m, cfg.BoxID, cfg.instance()) {

+ if cfg.instance() == "" {

+ // Not replaced: the metadata service has not answered yet, and

+ // the box takes its role once it does (moveCfg.instance).

+ w.notice(rec.Host+"#offline-no-instance", "moves: %s: box %s does not know its instance id yet; it takes no part in move %s until it does", rec.Host, cfg.Box, m.ID)

+ return nil

+ }

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)

}

@@ -634,7 +678,7 @@ func writeOfflineReport(ctx context.Context, s release.Store, r moves.OfflineRep

// 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

+ r.BoxID, r.InstanceID = w.cfg.BoxID, w.cfg.instance()

}

return writeOfflineReport(ctx, s, r)

}

@@ -877,7 +921,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,

+ Layout: w.offlineLayout(rt), BoxID: cfg.BoxID, InstanceID: cfg.instance(),

}, nil

}

diff --git a/internal/host/movesoffline_reports.go b/internal/host/movesoffline_reports.go

index 14b82e980..37370b1de 100644

--- a/internal/host/movesoffline_reports.go

+++ b/internal/host/movesoffline_reports.go

@@ -124,7 +124,7 @@ func (w *movesWorker) ownReport(ctx context.Context, s release.Store, rec provis

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}

+ BoxID: cfg.BoxID, InstanceID: cfg.instance()}

}

// offlineLaneStep is the stager's lane probe, once per move: a box that has

diff --git a/internal/host/movesoffline_reports_test.go b/internal/host/movesoffline_reports_test.go

index ebd8cdd60..07c6fcdbe 100644

--- a/internal/host/movesoffline_reports_test.go

+++ b/internal/host/movesoffline_reports_test.go

@@ -1125,3 +1125,60 @@ func TestVersionedArmedForOffline(t *testing.T) {

t.Fatal("armOffline replaced a lane probe or a box id it was given")

}

+// --- fix round 1 -------------------------------------------------------------

+// A box whose metadata service did not answer at arm time knows no instance

+// id; the console pins the one its status tick published later. The box

+// resolves its id lazily, so it takes its role once the id is known, and

+// its reports carry it.

+func TestInstanceIDResolvedAfterArm(t *testing.T) {

+ sw := newSSWorld(t)

+ ie := sw.ie

+ id := ""

+ ie.w.cfg.InstanceID = ""

+ ie.w.cfg.InstanceIDOf = func() string { return id }

+ ie.w.cfg.Offline = newOfflineView(ie.w.cfg, nil)

+ sw.instance(ie)

+ sw.queue()

+ sw.mustStagePass(ie)

+ if _, ok := sw.reportIf(moves.OfflineRoleLane, "ie-a"); ok {

+ t.Fatal("a box with no instance id took the pinned source's role")

+ }

+ if ie.w.cfg.Offline.keep(offHost) {

+ t.Fatal("a box with no instance id keeps the pinned source's row")

+ }

+ id = ssSourceInstance // the metadata service answers now

+ sw.mustStagePass(ie)

+ for _, role := range []string{moves.OfflineRoleLane, moves.OfflineRoleSource} {

+ r := sw.report(role, "ie-a")

+ if r.InstanceID != ssSourceInstance || r.BoxID != "ie-a" {

+ t.Fatalf("%s report names box %q instance %q", role, r.BoxID, r.InstanceID)

+ }

+ sw.validFor(r)

+ }

+ if !ie.w.cfg.Offline.keep(offHost) {

+ t.Fatal("the pinned source does not keep its row once its id is known")

+ }

+// The production resolver asks the metadata service at most once a minute

+// while it has no answer, and never again once it has one.

+func TestThrottledInstanceID(t *testing.T) {

+ now := time.Date(2026, 10, 23, 9, 0, 0, 0, time.UTC)

+ calls, answer := 0, ""

+ get := throttledInstanceID(func() string { calls++; return answer }, time.Minute, func() time.Time { return now })

+ if get() != "" || get() != "" || calls != 1 {

+ t.Fatalf("calls %d within the minute", calls)

+ }

+ now = now.Add(time.Minute)

+ answer = "i-1"

+ if get() != "i-1" || calls != 2 {

+ t.Fatalf("calls %d after the minute", calls)

+ }

+ answer = "i-other"

+ now = now.Add(time.Hour)

+ if get() != "i-1" || calls != 2 {

+ t.Fatal("a known id was asked again")

+ }

diff --git a/internal/host/movesoffline_seal.go b/internal/host/movesoffline_seal.go

index d269cd60d..90b327db8 100644

--- a/internal/host/movesoffline_seal.go

+++ b/internal/host/movesoffline_seal.go

@@ -617,7 +617,7 @@ func (w *movesWorker) offlineTakeSeal(ctx context.Context, s release.Store, rec

FilesPlain: a.plain, FilesSealed: a.sealed, FilesBytes: a.bytes, FilesEntries: a.entries,

EnvSHA256: envSHA, KeySHA256: keySHA, EnvDigest: envReport.Digest,

Release: version, TimersDigest: timersDigest, Replication: rt.Replication, NextDue: rt.NextDue,

- FinalCopiedBytes: round.BytesCopied, BoxID: cfg.BoxID, InstanceID: cfg.InstanceID,

+ FinalCopiedBytes: round.BytesCopied, BoxID: cfg.BoxID, InstanceID: cfg.instance(),

}

if err := moves.ValidateOfflineSeal(seal, host, m.ID); err != nil {

return seal, err

@@ -1228,7 +1228,7 @@ func (w *movesWorker) offlineSealRevert(ctx context.Context, s release.Store, re

liftedAt = cfg.now().UTC()

}

r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseReverting, At: cfg.now().UTC(),

- BoxID: cfg.BoxID, InstanceID: cfg.InstanceID, TombstoneLifted: true, TombstoneLiftedAt: liftedAt}

+ BoxID: cfg.BoxID, InstanceID: cfg.instance(), TombstoneLifted: true, TombstoneLiftedAt: liftedAt}

if err := w.putOfflineReport(ctx, s, r); err != nil {

return fmt.Errorf("offline move %s: report the lift of %s: %w", m.ID, host, err)

}

@@ -1259,6 +1259,6 @@ func (w *movesWorker) offlineSealRevert(ctx context.Context, s release.Store, re

clearOfflineReceipts(ctx, mirror, bucket, host)

w.offlineLiftFence(host, m)

r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC(),

- BoxID: cfg.BoxID, InstanceID: cfg.InstanceID, TombstoneLifted: !liftedAt.IsZero(), TombstoneLiftedAt: liftedAt}

+ BoxID: cfg.BoxID, InstanceID: cfg.instance(), TombstoneLifted: !liftedAt.IsZero(), TombstoneLiftedAt: liftedAt}

return w.putOfflineReport(ctx, s, r)

}

diff --git a/internal/host/movesoffline_stage.go b/internal/host/movesoffline_stage.go

index fdae13bcc..bc47de2cc 100644

--- a/internal/host/movesoffline_stage.go

+++ b/internal/host/movesoffline_stage.go

@@ -92,7 +92,7 @@ func (w *movesWorker) offlineStageStep(ctx context.Context, s release.Store, rec

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,

- BoxID: cfg.BoxID, InstanceID: cfg.InstanceID}

+ BoxID: cfg.BoxID, InstanceID: cfg.instance()}

book := w.stageBookFor(rec, prev)

book.Rounds++

p, round, err := w.offlineStageRound(ctx, s, rec, book.observe)

@@ -347,7 +347,7 @@ 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(), StagedRequest: m.StagedAt.UTC(), BoxID: w.cfg.BoxID, InstanceID: w.cfg.InstanceID}

+ At: w.cfg.now().UTC(), StageRequest: m.StagedAt.UTC(), StagedRequest: m.StagedAt.UTC(), BoxID: w.cfg.BoxID, InstanceID: w.cfg.instance()}

if prev != nil {

stageProgress{replicaObjects: prev.ReplicaObjects, replicaBytes: prev.ReplicaBytes, stagedObjects: prev.StagedObjects,

stagedBytes: prev.StagedBytes, roundObjects: prev.RoundObjects, roundBytes: prev.RoundBytes}.fill(&r)

diff --git a/internal/host/movespass.go b/internal/host/movespass.go

index c0d1bad50..010ca699e 100644

--- a/internal/host/movespass.go

+++ b/internal/host/movespass.go

@@ -342,6 +342,7 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry,

// writer's value, the label) and the instance its status report names.

cfg.BoxID = cfg.Box

cfg.InstanceID = cachedInstanceID()

+ cfg.InstanceIDOf = throttledInstanceID(cachedInstanceID, time.Minute, time.Now)

w := newMovesWorker(cfg)

if cfg.Enabled {

armPhase1(&cfg, w, hib, dataDir, reg, svc, eb, writes, os.Getenv)

@@ -1041,6 +1042,12 @@ type moveCfg struct {

// Box when it is empty.

BoxID string

InstanceID string

+ // InstanceIDOf resolves the instance id lazily (moveCfg.instance): in

+ // production the status tick's own cachedInstanceID, asked at most once

+ // a minute while it has no answer (throttledInstanceID), so a box whose

+ // metadata service did not answer at arm time takes its role once it

+ // does. Nil: InstanceID alone.

+ InstanceIDOf func() 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.