amadan
agent: arm the moves pass on both converge loops + the staging-fallback mirror (#331)
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PEypt7de2X1G84aQ9rNkCf
2 files changed,
+56
−0
internal/host/agent.go+14 −0internal/host/movespass.go+42 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go| index 13bb1580c..c6ead7c4f 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -307,6 +307,18 @@ func RunAgent(args []string) { |
| if logSink != nil { |
| svc.Logs = logSink |
| } |
| + // The residence-move machinery (issue #331), armed with the |
| + // activator: the staging-fallback restore needs a same-region |
| + // mirror client, and the move worker needs the service's quiesce |
| + // answer. Both are best-effort — a mirror client that cannot |
| + // build logs and leaves the fallback dark, exactly like every |
| + // other optional duty here. |
| + if m, err := lease.NewS3Mirror(context.Background(), cfg.Region, cfg.Region); err != nil { |
| + log.Printf("agent: staging-fallback mirror client: %v (restore fallback disarmed)", err) |
| + } else { |
| + eb.Mirror = m |
| + } |
| + armMovesPass(*cfg, dd, reg, svc) |
| svc.Start() |
| act = svc |
| log.Printf("agent: hibernation activator armed (bucket %s, idle %s)", cfg.Bucket, cfg.Idle) |
| @@ -1451,6 +1463,7 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu |
| if err := reconcileSerialized(ctx, store, reg, provisionTickConfig(os.Getenv)); err != nil { |
| log.Printf("agent: fast converge reconcile: %v", err) |
| } |
| + runMovesPass(ctx, store) |
| // Folded once per pass and handed in, never re-read inside the |
| // loop: this leg runs on every serial bump (a seconds cadence), |
| // and the fold is a LoadBoxes plus a List plus one Get per edges |
| @@ -1916,6 +1929,7 @@ func instancesLoop(reg *registry.Registry) { |
| if err := reconcileSerialized(ctx, store, reg, cfg); err != nil { |
| log.Printf("agent: instances pass: %v", err) |
| } |
| + runMovesPass(ctx, store) |
| deploymentRoutes := edgesfile.DeploymentRoutes(ctx, store, time.Now()) |
| if err := domains.ReconcileAliases(ctx, store, reg, time.Now, deploymentRoutes); err != nil { |
| log.Printf("agent: instances alias pass: %v", err) |
diff --git a/internal/host/movespass.go b/internal/host/movespass.go| index 4c2c07786..37df170a4 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -5,13 +5,16 @@ import ( |
| "encoding/json" |
| "fmt" |
| "log" |
| + "os" |
| "path/filepath" |
| "strings" |
| + "sync/atomic" |
| "time" |
| "github.com/carlosframework/platform/internal/activator" |
| "github.com/carlosframework/platform/internal/lease" |
| "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/registry" |
| "github.com/carlosframework/platform/internal/release" |
| ) |
| @@ -38,6 +41,45 @@ import ( |
| // Every step is idempotent; a box that dies mid-cut resumes at the top and |
| // falls through the already-done steps. |
| +// armedMoveCfg is RunAgent's one write; runMovesPass' reads are its only |
| +// other touch. An atomic pointer rather than plain assignment because the |
| +// arm happens on RunAgent's goroutine and the passes run on the converge |
| +// and instances loops. |
| +var armedMoveCfg atomic.Pointer[moveCfg] |
| + |
| +// armMovesPass wires the worker with what only RunAgent has in hand: the |
| +// activator's hibernate coordinates (the own-bucket side of every mirror), |
| +// the data dir the cut purges, the registry the cut quiesces through, and |
| +// the service that answers whether a child is down. Called once, before |
| +// svc.Start. |
| +func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service) { |
| + cfg := moveCfg{ |
| + Region: strings.TrimSpace(os.Getenv("CARLOS_REGION")), |
| + OwnBucket: hib.Bucket, |
| + OwnPrefix: hib.Prefix, |
| + DataDir: dataDir, |
| + Mirror: func(ctx context.Context, dstRegion string) (lease.MirrorStore, error) { |
| + return lease.NewS3Mirror(ctx, hib.Region, dstRegion) |
| + }, |
| + Quiesced: svc.Quiesced, |
| + SetStatus: reg.SetStatus, |
| + Now: time.Now, |
| + } |
| + armedMoveCfg.Store(&cfg) |
| +} |
| + |
| +// runMovesPass is the loops' call: inert until armed (no activator = no |
| +// exec instances = nothing this box could ever move). |
| +func runMovesPass(ctx context.Context, s release.Store) { |
| + cfg := armedMoveCfg.Load() |
| + if cfg == nil || s == nil { |
| + return |
| + } |
| + if err := movesPass(ctx, s, *cfg); err != nil { |
| + log.Printf("agent: moves pass: %v", err) |
| + } |
| +} |
| + |
| // moveCfg carries what the worker needs, injectable for tests. |
| type moveCfg struct { |
| Region string // this box's placement code, "" included (the home) |