carlos / platform Public

host: the moves write policy for the claim (hostbox.Policy.Moves) and the held timers import through SelectedTimers (Task 10b)

MovesClaimPolicy is the value for ClaimOwnerBindings.MovesPolicy, which
becomes hostbox.Policy.Moves: it grants held-config, held-release,
held-timers and the landing's restore, spawn, seed, writer and install
rights only to the target box after the flip, under a landing op whose
phase permits the action (held-timers at local, held-config and
held-release at configured, restore and install at restored), with the
selection's capture, kind, phase, op and host equal to the claim's and an
identity that fits the action; everything else is a typed
MovesPolicyRefusal (ErrMovesRefused). Nothing registers it in production
yet: RegisterProductionClaimOwners needs the Plan 3 and Plan 5 owners
too, so G16's refusal at checking under CARLOS_HOST_CLAIM stands, with
the comment saying what lifts it.

The landing's timers import goes through schedule.SelectedTimers with a
held-timers Work selected from the agent's hostwrite.Config
(heldTimersReplacer); with the flag off that is the registry's own
ReplaceTimers, as before. The hook takes the pass's context; armMovesPass
and armPhase1 take the claim wiring's Writes.

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

claude 2 weeks ago af6d89909ea193e33ca4668e867702a84507e55c parent d50652c

9 files changed, +647 −22

approved_moves_copy_test.go +1 −0
internal/host/agent.go +1 −1
internal/host/movesland.go +13 −3
internal/host/movesland_test.go +3 −2
internal/host/movespass.go +21 −13
internal/host/movespass_test.go +3 −2
internal/host/movesphase1_test.go +2 −1
internal/host/movespolicy.go +197 −0
internal/host/movespolicy_test.go +406 −0

diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go

index 8bf6ace78..cbc522e43 100644

--- a/approved_moves_copy_test.go

+++ b/approved_moves_copy_test.go

@@ -565,6 +565,7 @@ var movesScanFiles = []string{

"internal/host/movesmeasure_other.go",

"internal/host/movespass.go",

"internal/host/movesphase1.go",

+ "internal/host/movespolicy.go",

"internal/host/movesprobe.go",

"internal/host/movesreport.go",

"internal/host/movesstage.go",

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

index 35eaa126f..632c641d3 100644

--- a/internal/host/agent.go

+++ b/internal/host/agent.go

@@ -407,7 +407,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {

} else {

eb.Mirror = m

}

- armMovesPass(*cfg, dd, reg, svc, eb)

+ armMovesPass(*cfg, dd, reg, svc, eb, claims.Writes)

// The startup sweep (design §6.7): a leftover instance unit from a

// previous agent process, one this box's registry no longer holds

// a live wake for, is stopped before this process starts spawning

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

index 4ec659e2f..62ef09889 100644

--- a/internal/host/movesland.go

+++ b/internal/host/movesland.go

@@ -69,8 +69,13 @@ import (

// production and the pass logs a standing notice instead of landing; the

// tests arm everything with fakes and the stand-ins below. Under

// CARLOS_HOST_CLAIM the landing refuses at checking with a recorded reason

-// (ruling G16): the held writes it makes have no claim selection until Task

-// 10b supplies the adapters.

+// (ruling G16). The policy its held writes need, hostbox.Policy.Moves, is

+// MovesClaimPolicy (movespolicy.go) and the timers import already selects

+// its Work as held-timers; the refusal stands because the production claim

+// registration (RegisterProductionClaimOwners) requires every owner and the

+// Plan 3 (Inputs, PublicationEvidence, AdmissionEvidence) and Plan 5 (View,

+// Authorize, Evidence, LeaseSet) owners do not exist yet. It lifts when

+// those owners are registered.

// landingStepsPerPass bounds the phases one pass runs.

const landingStepsPerPass = 8

@@ -432,6 +437,11 @@ func (l *landingRun) ensureLocalDirs() error {

func (l *landingRun) checking(ctx context.Context) (bool, error) {

cfg := l.cfg()

host, id := l.rec.Host, l.g.ID

+ // Ruling G16: with host claims on, this box's held writes would need

+ // the claim's selection, and the production registration does not

+ // carry the Plan 3 and Plan 5 owners yet, so no selector exists here.

+ // MovesClaimPolicy (movespolicy.go) is the Policy.Moves half, ready;

+ // this refusal lifts when RegisterProductionClaimOwners has every owner.

if hostwrite.Enabled(os.Getenv) {

return l.markRollback(ctx, provision.RollbackClaimWired)

}

@@ -895,7 +905,7 @@ func (l *landingRun) local_(ctx context.Context) (bool, error) {

if err != nil {

return false, err

}

- if err := cfg.ReplaceTimers(host, rows); err != nil {

+ if err := cfg.ReplaceTimers(ctx, host, rows); err != nil {

return false, fmt.Errorf("replace the timers: %w", err)

}

if err := mark(landStepTimers, strconv.Itoa(len(rows))); err != nil {

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

index 60b729d10..726fda813 100644

--- a/internal/host/movesland_test.go

+++ b/internal/host/movesland_test.go

@@ -24,6 +24,7 @@ import (

"github.com/carlosframework/platform/internal/hostbox"

"github.com/carlosframework/platform/internal/hostclaim"

"github.com/carlosframework/platform/internal/hostguard"

+ "github.com/carlosframework/platform/internal/hostwrite"

"github.com/carlosframework/platform/internal/lease"

"github.com/carlosframework/platform/internal/moves"

"github.com/carlosframework/platform/internal/moves/archive"

@@ -337,7 +338,7 @@ func (h *landingHarness) landingCfg(t *testing.T) moveCfg {

return h.render()

}

c.ReleaseReady = func(context.Context, release.Store, provision.Record, string) moves.Check { return h.release }

- c.ReplaceTimers = func(host string, rows []registry.Timer) error {

+ c.ReplaceTimers = func(_ context.Context, host string, rows []registry.Timer) error {

h.replaced = append(h.replaced, append([]registry.Timer(nil), rows...))

return nil

}

@@ -1213,7 +1214,7 @@ func TestLandingWaitsUnarmedInProduction(t *testing.T) {

// Production's own wiring: everything but the park engine's hooks.

var prod moveCfg

hib := activator.Config{Bucket: "data-uk", Region: "eu-west-2", Prefix: "instances"}

- armPhase1(&prod, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, h.tdir, nil, nil, nil, func(string) string { return "" })

+ armPhase1(&prod, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, h.tdir, nil, nil, nil, hostwrite.Config{}, func(string) string { return "" })

if prod.Guard.DataDir != h.tdir || prod.Local.DataDir != h.tdir || prod.OwnLease == nil || prod.RenderHeld == nil || prod.RemoveVerifyEntry == nil || prod.WriteGen == nil {

t.Fatalf("production wiring: %+v", prod)

}

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

index b7d21ee68..b570dd677 100644

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

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

@@ -21,6 +21,7 @@ import (

"github.com/carlosframework/platform/internal/hostbox"

"github.com/carlosframework/platform/internal/hostclaim"

"github.com/carlosframework/platform/internal/hostguard"

+ "github.com/carlosframework/platform/internal/hostwrite"

"github.com/carlosframework/platform/internal/layout"

"github.com/carlosframework/platform/internal/lease"

"github.com/carlosframework/platform/internal/moves"

@@ -184,7 +185,7 @@ func (w *movesWorker) mirror(ctx context.Context, dstRegion string) (lease.Mirro

// Phase 1 launch environment and S2). Called once, before svc.Start: with

// the Phase 1 switch on it sets eb.Strict, which the spawn path reads; with

// it off eb.Strict stays nil, so every spawn is main's.

-func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend) {

+func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, writes hostwrite.Config) {

readyDelta, ok := moveReadyDeltaFromEnv(os.Getenv)

if !ok {

log.Printf("agent: %s=%q ignored: want a whole number of bytes, 0 or more — keeping the built-in default of %d", moveReadyDeltaEnv, os.Getenv(moveReadyDeltaEnv), defaultMoveReadyDelta)

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

}

w := &movesWorker{mirrors: map[string]lease.MirrorStore{}}

if cfg.Enabled {

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

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

w.holds = newMovesHoldView(cfg.Box, cfg.Region, hostclaim.Read, holdStageRequester(dataDir))

cfg.TimersDrained = w.holds.drainTimers

}

@@ -242,8 +243,10 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry,

}

// armPhase1 wires the Phase 1 hooks (moves design §7.3). Only with the

-// switch on: off, none is set and eb.Strict stays nil.

-func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, getenv func(string) string) {

+// switch on: off, none is set and eb.Strict stays nil. writes is the

+// agent's host-write configuration (the claim wiring's): the landing's

+// held timers import selects its Work through it.

+func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service, eb *activator.ExecBackend, writes hostwrite.Config, getenv func(string) string) {

cfg.Box = metricsBoxLabel(getenv, readMachineID())

cfg.OptDir = release.OptDir()

cfg.DeployBucket = strings.TrimSpace(getenv("CARLOS_DEPLOYMENT_BUCKET"))

@@ -292,11 +295,14 @@ func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir strin

// ParkLeases, ParkView and ParkAuthorize are not (ruling G17: the

// residue parks need claim Plan 6b's regional runner for their

// acknowledgement and Plan 5's view and authorizer), so landingArmed is

- // false in production and the landing pass only logs. Under

- // CARLOS_HOST_CLAIM the held timers write would go through

- // schedule.SelectedTimers with a held-timers Work (ruling G16); until

- // Task 10b supplies hostbox.Policy.Moves the landing refuses at checking,

- // so the registry's own ReplaceTimers is the pass-through here.

+ // false in production and the landing pass only logs. The held timers

+ // write goes through schedule.SelectedTimers with a held-timers Work

+ // selected from writes (ruling G16, heldTimersReplacer): with

+ // CARLOS_HOST_CLAIM off that is the registry's own ReplaceTimers, as

+ // before; with it on, the claim's selection and hostbox.Policy.Moves

+ // (MovesClaimPolicy) decide, once the production registration carries

+ // the Plan 3 and Plan 5 owners. Until then the landing refuses at

+ // checking under the flag (movesland.go).

cfg.Guard = hostguard.Guard{DataDir: dataDir}

cfg.Local = park.Local{DataDir: dataDir, Box: cfg.Box}

cfg.OwnLease = hostbox.RenewingLeases{Region: hib.Region}.Acquire

@@ -306,7 +312,7 @@ func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir strin

cfg.RemoveVerifyEntry = dbverify.RemoveHost

cfg.WriteGen = hostbox.WriteGenerationMarker

if reg != nil {

- cfg.ReplaceTimers = reg.ReplaceTimers

+ cfg.ReplaceTimers = heldTimersReplacer(writes, reg)

cfg.SetReplication = reg.SetReplication

cfg.SetNextDue = reg.SetNextDue

}

@@ -518,8 +524,10 @@ type moveCfg struct {

// Authorize with ParkLeases set gets the landing's stand-ins. OwnLease

// takes the location's lease in the own bucket for the build

// (hostbox.RenewingLeases over lease.Acquire). RenderHeld is

- // appconfig.RenderHeld for the held row; ReplaceTimers, SetReplication

- // and SetNextDue are the registry's; RemoveVerifyEntry is

+ // appconfig.RenderHeld for the held row; ReplaceTimers is the

+ // registry's through schedule.SelectedTimers with a held-timers Work

+ // (heldTimersReplacer); SetReplication and SetNextDue are the

+ // registry's; RemoveVerifyEntry is

// dbverify.RemoveHost; WriteGen is hostbox.WriteGenerationMarker, the

// one <dbPath>.gen writer (ruling G2).

Guard hostguard.Guard

@@ -529,7 +537,7 @@ type moveCfg struct {

ParkAuthorize func(context.Context, hostclaim.Observed, hostclaim.Obligation) error

OwnLease func(ctx context.Context, bucket, location string) (park.HeldLease, error)

RenderHeld func(ctx context.Context, s release.Store, rt registry.Route, op string) (appconfig.HeldRender, error)

- ReplaceTimers func(host string, rows []registry.Timer) error

+ ReplaceTimers func(ctx context.Context, host string, rows []registry.Timer) error

SetReplication func(host, class string) error

SetNextDue func(host string, due int64) error

RemoveVerifyEntry func(ctx context.Context, s release.Store, host string) error

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

index 93f74b5dd..7a4161b49 100644

--- a/internal/host/movespass_test.go

+++ b/internal/host/movespass_test.go

@@ -12,6 +12,7 @@ import (

"time"

"github.com/carlosframework/platform/internal/activator"

+ "github.com/carlosframework/platform/internal/hostwrite"

"github.com/carlosframework/platform/internal/lease"

"github.com/carlosframework/platform/internal/moves"

"github.com/carlosframework/platform/internal/provision"

@@ -925,7 +926,7 @@ func TestArmMovesPassReadsTheReadyDelta(t *testing.T) {

{"lots", defaultMoveReadyDelta},

} {

t.Setenv("CARLOS_MOVE_READY_DELTA_BYTES", tc.raw)

- armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil)

+ armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{})

if got := armedMovesWorker.Load().cfg.ReadyDelta; got != tc.want {

t.Errorf("CARLOS_MOVE_READY_DELTA_BYTES=%q armed a bar of %d, want %d", tc.raw, got, tc.want)

}

@@ -944,7 +945,7 @@ func TestArmMovesPassReadsTheSwitch(t *testing.T) {

{"", false}, {"on", true}, {"1", true}, {"true", true}, {"off", false}, {"yes", false},

} {

t.Setenv(moves.SwitchEnv, tc.raw)

- armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil)

+ armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{})

if got := armedMovesWorker.Load().cfg.Enabled; got != tc.want {

t.Errorf("%s=%q armed Enabled=%v, want %v", moves.SwitchEnv, tc.raw, got, tc.want)

}

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

index 5d3c95b2a..4c1fb6e5a 100644

--- a/internal/host/movesphase1_test.go

+++ b/internal/host/movesphase1_test.go

@@ -14,6 +14,7 @@ import (

"github.com/carlosframework/platform/internal/activator"

"github.com/carlosframework/platform/internal/hostclaim"

+ "github.com/carlosframework/platform/internal/hostwrite"

"github.com/carlosframework/platform/internal/lease"

"github.com/carlosframework/platform/internal/moves"

"github.com/carlosframework/platform/internal/moves/archive"

@@ -143,7 +144,7 @@ func TestPhase1IsInertWithTheSwitchOff(t *testing.T) {

}

t.Setenv(moves.SwitchEnv, v)

eb := &activator.ExecBackend{}

- armMovesPass(activator.Config{}, t.TempDir(), nil, nil, eb)

+ armMovesPass(activator.Config{}, t.TempDir(), nil, nil, eb, hostwrite.Config{})

if (eb.Strict != nil) != on {

t.Fatalf("switch %v: eb.Strict set = %v", on, eb.Strict != nil)

}

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

new file mode 100644

index 000000000..a83754cb3

--- /dev/null

+++ b/internal/host/movespolicy.go

@@ -0,0 +1,197 @@

+package host

+import (

+ "context"

+ "errors"

+ "fmt"

+ "github.com/carlosframework/platform/internal/hostbox"

+ "github.com/carlosframework/platform/internal/hostclaim"

+ "github.com/carlosframework/platform/internal/hostwrite"

+ "github.com/carlosframework/platform/internal/provision"

+ "github.com/carlosframework/platform/internal/registry"

+ "github.com/carlosframework/platform/internal/schedule"

+// The moves write policy (moves Phase 1 Plan 2 Task 10b; ruling T10b-scope).

+//

+// hostbox.Policy.Check separates ordinary host writes from the moves-owned

+// ones: under a move kind it checks the registry hold and the selection's

+// op and box, and then asks Policy.Moves. MovesClaimPolicy is that answer,

+// the value for ClaimOwnerBindings.MovesPolicy. It grants a write only when

+// every one of these holds, and refuses everything else with a

+// *MovesPolicyRefusal (fail closed):

+//

+// - the claim's op is the landing of a move (hostclaim.KindLanding). The

+// other move kinds grant nothing here: the source writes nothing held

+// while moving, and the recoveries (reverting, restoring) are Plan 4's,

+// which extends this table when it builds them;

+// - the selection is the claim's: the same host, op, kind, phase and

+// capture (holder incarnation and stamp);

+// - this box is the target after the flip: the executor, the second

+// participant, intent.target, and not intent.source; in the move's

+// target region (intent.to); and not in lost;

+// - the landing's phase permits the action (movesPhaseRights): held-timers

+// from local, held-config and held-release from configured, restore and

+// install from restored. These are the phases Plan 2 built (checking to

+// restored). The probe's rights at probing (spawn, seed, writer, and the

+// held render before every probe, §7.7) and the ordinary wake rights at

+// committed (§19.1) are Plan 3's rows, added when the probe is built;

+// - the identity fits the action: a held-config write names the rendered

+// config and nothing else, a held-release write names the release and

+// nothing else, and every other action carries no identity.

+//

+// Nothing here reads the record: the landing pins the record's inputs

+// itself at configured (the etag comparison against Move.Inputs), and the

+// claim says everything the policy needs about who may write what.

+//

+// Production registration. RegisterProductionClaimOwners requires every

+// owner, and the Inputs, PublicationEvidence and AdmissionEvidence owners

+// (Plan 3) and View, Authorize, Evidence and LeaseSet (moves Plan 5) do not

+// exist yet, so nothing registers this in production: it is tested and

+// ready to plug in. Until that registration, the landing keeps refusing at

+// checking while CARLOS_HOST_CLAIM is on (ruling G16, movesland.go).

+// ErrMovesRefused is what every MovesPolicyRefusal unwraps to.

+var ErrMovesRefused = errors.New("host: the moves policy refuses the write")

+// MovesPolicyRefusal is MovesClaimPolicy's typed refusal: which action was

+// asked under which op, kind and phase, by which box, and why.

+type MovesPolicyRefusal struct {

+ Host, Op, Action, Kind, Phase, Box, Reason string

+func (e *MovesPolicyRefusal) Error() string {

+ return fmt.Sprintf("host: the moves policy refuses %s for %s under %s/%q on %s: %s", e.Action, e.Host, e.Kind, e.Phase, e.Box, e.Reason)

+func (e *MovesPolicyRefusal) Unwrap() error { return ErrMovesRefused }

+// The held actions and the landing's rights hostbox.Policy.Check asks

+// Policy.Moves about (wakepolicy.go).

+const (

+ movesActionHeldConfig = "held-config"

+ movesActionHeldRelease = "held-release"

+ movesActionHeldTimers = "held-timers"

+ movesActionRestore = "restore"

+ movesActionSpawn = "spawn"

+ movesActionSeed = "seed"

+ movesActionWriter = "writer"

+ movesActionInstall = "install"

+// movesPhaseRights is the landing's table: the actions each phase permits

+// (§7.6). A phase or action not listed is refused.

+var movesPhaseRights = map[string]map[string]bool{

+ provision.LandLocal: {movesActionHeldTimers: true},

+ provision.LandConfigured: {movesActionHeldConfig: true, movesActionHeldRelease: true},

+ provision.LandRestored: {movesActionRestore: true, movesActionInstall: true},

+// movesPolicyActions is every action the policy knows. Anything else is

+// refused before the table is consulted.

+var movesPolicyActions = map[string]bool{

+ movesActionHeldConfig: true, movesActionHeldRelease: true, movesActionHeldTimers: true,

+ movesActionRestore: true, movesActionSpawn: true, movesActionSeed: true, movesActionWriter: true, movesActionInstall: true,

+// MovesClaimPolicy is hostbox.Policy.Moves for the moves plans: see the

+// file comment for what it grants.

+func MovesClaimPolicy(ctx context.Context, c *hostclaim.Claim, s hostbox.Selection, id hostwrite.Identity) error {

+ refuse := func(reason string) error {

+ r := &MovesPolicyRefusal{Host: s.Host, Op: s.Op, Action: s.Action, Box: s.Box, Reason: reason}

+ if c != nil {

+ r.Kind, r.Phase = c.Op.Kind, c.Op.Phase

+ }

+ return r

+ }

+ if ctx == nil {

+ return refuse("no context")

+ }

+ if err := ctx.Err(); err != nil {

+ return err

+ }

+ if c == nil || c.Holder == nil {

+ return refuse("the claim has no holder")

+ }

+ if !movesPolicyActions[s.Action] {

+ return refuse("not a moves action")

+ }

+ if s.Host == "" || s.Box == "" || s.Region == "" || s.Op == "" {

+ return refuse("the selection names no host, box, region or op")

+ }

+ if c.Host != s.Host {

+ return refuse("the claim is another host's")

+ }

+ if c.Op.ID != s.Op {

+ return refuse("the claim is under another op")

+ }

+ if c.Op.Kind != s.Kind || c.Op.Phase != s.Phase {

+ return refuse("the claim's kind or phase is not the selection's")

+ }

+ if s.Captured.Stamp == 0 || c.Holder.Stamp != s.Captured.Stamp || c.Holder.Incarnation != s.Captured.Incarnation {

+ return refuse("the capture is not the holder's")

+ }

+ switch c.Op.Kind {

+ case hostclaim.KindLanding:

+ case hostclaim.KindMoving, hostclaim.KindReverting, hostclaim.KindRestoring:

+ return refuse("only the landing writes held state")

+ default:

+ return refuse("not a move")

+ }

+ if c.Op.Exec != s.Box {

+ return refuse("this box is not the executor")

+ }

+ if len(c.Op.Participants) != 2 || c.Op.Participants[1] != s.Box || c.Op.Intent["target"] != s.Box || c.Op.Intent["source"] == s.Box {

+ return refuse("this box is not the move's target")

+ }

+ if c.Op.Intent["to"] != s.Region {

+ return refuse("this region is not the move's target region")

+ }

+ for _, lost := range c.Lost {

+ if lost == s.Box {

+ return refuse("this box is declared lost")

+ }

+ if !movesPhaseRights[c.Op.Phase][s.Action] {

+ return refuse("the phase does not permit the action")

+ }

+ if !movesIdentityFits(s.Action, id) {

+ return refuse("the identity does not fit the action")

+ }

+ return nil

+// movesIdentityFits: a held-config write names the rendered config and no

+// release, a held-release write the release and no config, and every other

+// action carries no identity at all.

+func movesIdentityFits(action string, id hostwrite.Identity) bool {

+ switch action {

+ case movesActionHeldConfig:

+ return id.Config != "" && id.Release == ""

+ case movesActionHeldRelease:

+ return id.Release != "" && id.Config == ""

+ default:

+ return id == hostwrite.Identity{}

+ }

+// heldTimersReplacer is the landing's timers import (ruling G16):

+// schedule.SelectedTimers over the registry, its Work selected as

+// held-timers for the host, so under CARLOS_HOST_CLAIM the write runs

+// inside the claim's selection and Policy.Moves is asked first. With the

+// flag off hostwrite.Config.Select answers the legacy pass-through and the

+// registry's own ReplaceTimers runs exactly as before. With the flag on and

+// no selector it refuses (hostwrite.ErrUnavailable) and writes nothing.

+func heldTimersReplacer(writes hostwrite.Config, reg *registry.Registry) func(ctx context.Context, host string, rows []registry.Timer) error {

+ return func(ctx context.Context, host string, rows []registry.Timer) error {

+ if ctx == nil || reg == nil {

+ return hostwrite.ErrUnavailable

+ }

+ work, err := writes.Select(ctx, host, movesActionHeldTimers)

+ if err != nil {

+ return err

+ }

+ return schedule.SelectedTimers{Store: reg, Work: work, Context: ctx, Host: host}.ReplaceTimers(host, rows)

+ }

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

new file mode 100644

index 000000000..ac35bd185

--- /dev/null

+++ b/internal/host/movespolicy_test.go

@@ -0,0 +1,406 @@

+package host

+import (

+ "context"

+ "errors"

+ "path/filepath"

+ "reflect"

+ "testing"

+ "github.com/carlosframework/platform/internal/activator"

+ "github.com/carlosframework/platform/internal/hostbox"

+ "github.com/carlosframework/platform/internal/hostclaim"

+ "github.com/carlosframework/platform/internal/hostguard"

+ "github.com/carlosframework/platform/internal/hostwrite"

+ "github.com/carlosframework/platform/internal/lease"

+ "github.com/carlosframework/platform/internal/provision"

+ "github.com/carlosframework/platform/internal/registry"

+const (

+ mpHost = "shop.example"

+ mpOp = "mv_policy"

+ mpInc = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"

+ mpTarget = "uk-box"

+ mpSource = "ie-box"

+// mpClaim is the claim after the flip: landing for mpOp at phase, the

+// target executing, the holder at stamp 7.

+func mpClaim(kind, phase string) *hostclaim.Claim {

+ return &hostclaim.Claim{V: 1, Host: mpHost, Gen: 9,

+ Holder: &hostclaim.Holder{Family: "instance", Incarnation: mpInc, Stamp: 7, Generation: "g-src", Session: 3, Location: "instances/" + mpHost + "/" + mpInc},

+ Op: hostclaim.Op{Kind: kind, ID: mpOp, Phase: phase, Exec: mpTarget, Participants: []string{mpSource, mpTarget},

+ Intent: map[string]string{"to": "uk", "source": mpSource, "target": mpTarget, "source_enrollment": "enr-ie"}}}

+// mpSelection is the selection hostbox.WorkSource would capture for that

+// claim on the target.

+func mpSelection(c *hostclaim.Claim, action string) hostbox.Selection {

+ return hostbox.Selection{Host: mpHost, Action: action, Box: mpTarget, Region: "uk",

+ Captured: hostguard.Captured{Incarnation: c.Holder.Incarnation, Stamp: c.Holder.Stamp},

+ Kind: c.Op.Kind, Op: c.Op.ID, Phase: c.Op.Phase, Generation: c.Holder.Generation, Session: c.Holder.Session}

+// mpIdentity is the identity that fits the action.

+func mpIdentity(action string) hostwrite.Identity {

+ switch action {

+ case "held-config":

+ return hostwrite.Identity{Config: "sha256:abc;default=true;environment=false;instance=false"}

+ case "held-release":

+ return hostwrite.Identity{Release: "v1.2.3"}

+ }

+ return hostwrite.Identity{}

+var mpActions = []string{"held-config", "held-release", "held-timers", "restore", "spawn", "seed", "writer", "install"}

+// Every action against every landing phase, including the probe's and the

+// commit's: granted exactly where the table says, refused with the typed

+// error everywhere else. The same cells through hostbox.Policy.Check with

+// the registry hold the landing has.

+func TestMovesClaimPolicyGrantsByPhase(t *testing.T) {

+ granted := map[string]map[string]bool{

+ provision.LandLocal: {"held-timers": true},

+ provision.LandConfigured: {"held-config": true, "held-release": true},

+ provision.LandRestored: {"restore": true, "install": true},

+ }

+ phases := append(provision.LandingOrder(), "committed")

+ policy := hostbox.Policy{Moves: MovesClaimPolicy, Held: func(context.Context, string) (string, error) { return mpOp, nil }}

+ grants := 0

+ for _, phase := range phases {

+ for _, action := range mpActions {

+ c := mpClaim(hostclaim.KindLanding, phase)

+ s := mpSelection(c, action)

+ err := MovesClaimPolicy(t.Context(), c, s, mpIdentity(action))

+ want := granted[phase][action]

+ if (err == nil) != want {

+ t.Errorf("%s/%s: err %v, want granted %v", phase, action, err, want)

+ continue

+ }

+ if err != nil {

+ var r *MovesPolicyRefusal

+ if !errors.As(err, &r) || !errors.Is(err, ErrMovesRefused) || r.Action != action || r.Phase != phase || r.Kind != hostclaim.KindLanding || r.Reason == "" {

+ t.Errorf("%s/%s: not the typed refusal: %#v", phase, action, err)

+ }

+ if chk := policy.Check(t.Context(), c, s, mpIdentity(action)); (chk == nil) != want {

+ t.Errorf("%s/%s through Policy.Check: %v, want granted %v", phase, action, chk, want)

+ }

+ if want {

+ grants++

+ }

+ if grants != 5 {

+ t.Fatalf("%d cells granted, want 5", grants)

+ }

+// The claim's role and the selection's shape: a non-move op, the other move

+// kinds, a mismatched op, box or region, the source after the flip, a lost

+// box, a stale capture, a mismatched kind or phase and a wrong identity are

+// all refused, and nothing but the exact landing cell passes.

+func TestMovesClaimPolicyRefusesOtherRoles(t *testing.T) {

+ base := func() (*hostclaim.Claim, hostbox.Selection) {

+ c := mpClaim(hostclaim.KindLanding, provision.LandLocal)

+ return c, mpSelection(c, "held-timers")

+ }

+ cases := map[string]func(c *hostclaim.Claim, s *hostbox.Selection, id *hostwrite.Identity){

+ "serving op": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {

+ c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: mpTarget}

+ s.Kind, s.Op, s.Phase = c.Op.Kind, "", ""

+ },

+ "creating op": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {

+ c.Op.Kind = hostclaim.KindCreating

+ s.Kind = c.Op.Kind

+ },

+ "moving, the source executing": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {

+ c.Op.Kind, c.Op.Phase, c.Op.Exec = hostclaim.KindMoving, hostclaim.PhaseSealed, mpSource

+ s.Kind, s.Phase, s.Box, s.Region = c.Op.Kind, c.Op.Phase, mpSource, "ie"

+ },

+ "moving, the target asking": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {

+ c.Op.Kind, c.Op.Phase = hostclaim.KindMoving, provision.LandLocal

+ s.Kind = c.Op.Kind

+ },

+ "reverting": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {

+ c.Op.Kind = hostclaim.KindReverting

+ s.Kind = c.Op.Kind

+ },

+ "restoring": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {

+ c.Op.Kind = hostclaim.KindRestoring

+ s.Kind = c.Op.Kind

+ },

+ "another op selected": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) { s.Op = "mv_other" },

+ "the claim under another op": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) {

+ c.Op.ID = "mv_other"

+ },

+ "another host's claim": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) { c.Host = "other.example" },

+ "the source box after the flip": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {

+ s.Box, s.Region = mpSource, "ie"

+ },

+ "the source named executor": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {

+ c.Op.Exec = mpSource

+ s.Box, s.Region = mpSource, "ie"

+ },

+ "exec none": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) { c.Op.Exec = "" },

+ "a third box executing": func(c *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {

+ c.Op.Exec = "de-box"

+ s.Box = "de-box"

+ },

+ "the target not the second participant": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) {

+ c.Op.Participants = []string{mpTarget, mpSource}

+ },

+ "the intent names another target": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) {

+ c.Op.Intent["target"] = "de-box"

+ },

+ "the intent names this box as source too": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) {

+ c.Op.Intent["source"] = mpTarget

+ },

+ "another region": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) { s.Region = "ie" },

+ "no region": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) { s.Region = "" },

+ "the box is lost": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) {

+ c.Lost = []string{mpTarget}

+ },

+ "a stale capture": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) { s.Captured.Stamp = 6 },

+ "an empty capture": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {

+ s.Captured = hostguard.Captured{}

+ },

+ "another incarnation captured": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {

+ s.Captured.Incarnation = "fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210"

+ },

+ "the claim moved on since the selection": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) {

+ c.Op.Phase = provision.LandConfigured

+ },

+ "the selection's kind is not the claim's": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {

+ s.Kind = hostclaim.KindServing

+ },

+ "no holder": func(c *hostclaim.Claim, _ *hostbox.Selection, _ *hostwrite.Identity) { c.Holder = nil },

+ "no action": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) { s.Action = "" },

+ "an ordinary action": func(_ *hostclaim.Claim, s *hostbox.Selection, _ *hostwrite.Identity) {

+ s.Action = "timer-delivery"

+ },

+ "a config identity on the timers": func(_ *hostclaim.Claim, _ *hostbox.Selection, id *hostwrite.Identity) {

+ id.Config = "sha256:abc"

+ },

+ "a release identity on the timers": func(_ *hostclaim.Claim, _ *hostbox.Selection, id *hostwrite.Identity) {

+ id.Release = "v1"

+ },

+ }

+ for name, mutate := range cases {

+ t.Run(name, func(t *testing.T) {

+ c, s := base()

+ id := hostwrite.Identity{}

+ mutate(c, &s, &id)

+ err := MovesClaimPolicy(t.Context(), c, s, id)

+ if err == nil {

+ t.Fatal("granted")

+ }

+ if !errors.Is(err, ErrMovesRefused) {

+ t.Fatalf("not the typed refusal: %v", err)

+ }

+ })

+ }

+ t.Run("the base cell is granted", func(t *testing.T) {

+ c, s := base()

+ if err := MovesClaimPolicy(t.Context(), c, s, hostwrite.Identity{}); err != nil {

+ t.Fatal(err)

+ }

+ })

+ t.Run("held-config needs the rendered identity and no release", func(t *testing.T) {

+ c := mpClaim(hostclaim.KindLanding, provision.LandConfigured)

+ s := mpSelection(c, "held-config")

+ for _, id := range []hostwrite.Identity{{}, {Release: "v1"}, {Config: "sha256:abc", Release: "v1"}} {

+ if err := MovesClaimPolicy(t.Context(), c, s, id); !errors.Is(err, ErrMovesRefused) {

+ t.Fatalf("%+v: %v", id, err)

+ }

+ if err := MovesClaimPolicy(t.Context(), c, s, hostwrite.Identity{Config: "sha256:abc"}); err != nil {

+ t.Fatal(err)

+ }

+ })

+ t.Run("held-release needs the release and no config", func(t *testing.T) {

+ c := mpClaim(hostclaim.KindLanding, provision.LandConfigured)

+ s := mpSelection(c, "held-release")

+ for _, id := range []hostwrite.Identity{{}, {Config: "sha256:abc"}, {Config: "sha256:abc", Release: "v1"}} {

+ if err := MovesClaimPolicy(t.Context(), c, s, id); !errors.Is(err, ErrMovesRefused) {

+ t.Fatalf("%+v: %v", id, err)

+ }

+ if err := MovesClaimPolicy(t.Context(), c, s, hostwrite.Identity{Release: "v1"}); err != nil {

+ t.Fatal(err)

+ }

+ })

+ t.Run("a nil claim or a dead context", func(t *testing.T) {

+ c, s := base()

+ if err := MovesClaimPolicy(t.Context(), nil, s, hostwrite.Identity{}); !errors.Is(err, ErrMovesRefused) {

+ t.Fatal(err)

+ }

+ ctx, cancel := context.WithCancel(t.Context())

+ cancel()

+ if err := MovesClaimPolicy(ctx, c, s, hostwrite.Identity{}); !errors.Is(err, context.Canceled) {

+ t.Fatal(err)

+ }

+ //lint:ignore SA1012 the nil context is the case under test

+ if err := MovesClaimPolicy(nil, c, s, hostwrite.Identity{}); !errors.Is(err, ErrMovesRefused) { //nolint:staticcheck

+ t.Fatal(err)

+ }

+ })

+ t.Run("the refusal says what and why", func(t *testing.T) {

+ c, s := base()

+ s.Region = "ie"

+ err := MovesClaimPolicy(t.Context(), c, s, hostwrite.Identity{})

+ var r *MovesPolicyRefusal

+ if !errors.As(err, &r) || r.Host != mpHost || r.Op != mpOp || r.Action != "held-timers" || r.Box != mpTarget || r.Kind != hostclaim.KindLanding || r.Phase != provision.LandLocal || r.Reason == "" || err.Error() == "" {

+ t.Fatalf("%#v", err)

+ }

+ })

+// The held-timers import with CARLOS_HOST_CLAIM off is the registry's own

+// ReplaceTimers: the rows land exactly as before, bookkeeping included,

+// another host's rows untouched. With the flag on, the write runs inside the

+// selected Work, or not at all.

+func TestHeldTimersReplacerPassesThroughWithTheFlagOff(t *testing.T) {

+ open := func(t *testing.T) *registry.Registry {

+ t.Helper()

+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))

+ if err != nil {

+ t.Fatal(err)

+ }

+ t.Cleanup(func() { reg.Close() })

+ for _, tm := range []registry.Timer{

+ {Host: mpHost, Name: "old", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 100, CreatedAt: 1},

+ {Host: "other.example", Name: "other", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 100, CreatedAt: 1},

+ } {

+ if err := reg.UpsertTimer(tm); err != nil {

+ t.Fatal(err)

+ }

+ return reg

+ }

+ rows := []registry.Timer{

+ {Host: mpHost, Name: "nightly", Kind: "schedule", Spec: "0 3 * * *", Path: "/n", NextAt: 5000, LastAt: 4000, LastStatus: "ok", LastMs: 12, Attempts: 2, CreatedAt: 10, DueAt: 4900, Registration: "reg-nightly"},

+ {Host: mpHost, Name: "once-1", Kind: "once", Spec: "2026-10-06T00:00:00Z", Path: "/o", NextAt: 6000, CreatedAt: 11, DueAt: 6000, Registration: "reg-once"},

+ }

+ unchanged := func(t *testing.T, reg *registry.Registry) {

+ t.Helper()

+ got, _ := reg.TimersFor(mpHost)

+ if len(got) != 1 || got[0].Name != "old" {

+ t.Fatalf("the rows changed: %+v", got)

+ }

+ t.Run("flag off: the registry's replacement, exactly", func(t *testing.T) {

+ reg, want := open(t), open(t)

+ if err := heldTimersReplacer(hostwrite.Config{}, reg)(t.Context(), mpHost, rows); err != nil {

+ t.Fatal(err)

+ }

+ if err := want.ReplaceTimers(mpHost, rows); err != nil {

+ t.Fatal(err)

+ }

+ got, _ := reg.TimersFor(mpHost)

+ ref, _ := want.TimersFor(mpHost)

+ if len(got) != 2 || !reflect.DeepEqual(got, ref) {

+ t.Fatalf("through the replacer:\n%+v\nthrough the registry:\n%+v", got, ref)

+ }

+ if other, _ := reg.TimersFor("other.example"); len(other) != 1 {

+ t.Fatalf("another host's rows: %+v", other)

+ }

+ })

+ t.Run("flag off: a row for another host writes nothing", func(t *testing.T) {

+ reg := open(t)

+ bad := append([]registry.Timer{{Host: "other.example", Name: "x", Kind: "schedule", Spec: "@every 1m", Path: "/t", NextAt: 1, CreatedAt: 1}}, rows...)

+ if err := heldTimersReplacer(hostwrite.Config{}, reg)(t.Context(), mpHost, bad); err == nil {

+ t.Fatal("accepted")

+ }

+ unchanged(t, reg)

+ })

+ t.Run("flag on, no selector: refused, nothing written", func(t *testing.T) {

+ reg := open(t)

+ if err := heldTimersReplacer(hostwrite.Config{Enabled: true}, reg)(t.Context(), mpHost, rows); !errors.Is(err, hostwrite.ErrUnavailable) {

+ t.Fatal(err)

+ }

+ unchanged(t, reg)

+ })

+ t.Run("flag on: the write runs inside the selected held-timers Work", func(t *testing.T) {

+ reg := open(t)

+ sel := &mpSelector{}

+ if err := heldTimersReplacer(hostwrite.Config{Enabled: true, Selector: sel}, reg)(t.Context(), mpHost, rows); err != nil {

+ t.Fatal(err)

+ }

+ if sel.host != mpHost || sel.action != "held-timers" || sel.with != 1 || sel.identity != (hostwrite.Identity{}) {

+ t.Fatalf("selection: %+v", sel)

+ }

+ if got, _ := reg.TimersFor(mpHost); len(got) != 2 {

+ t.Fatalf("rows: %+v", got)

+ }

+ })

+ t.Run("flag on: a refused selection writes nothing", func(t *testing.T) {

+ reg := open(t)

+ sel := &mpSelector{deny: errors.New("refused")}

+ if err := heldTimersReplacer(hostwrite.Config{Enabled: true, Selector: sel}, reg)(t.Context(), mpHost, rows); err == nil || !errors.Is(err, sel.deny) {

+ t.Fatal(err)

+ }

+ unchanged(t, reg)

+ })

+ t.Run("no registry: refused", func(t *testing.T) {

+ if err := heldTimersReplacer(hostwrite.Config{}, nil)(t.Context(), mpHost, rows); !errors.Is(err, hostwrite.ErrUnavailable) {

+ t.Fatal(err)

+ }

+ })

+// mpSelector records the selection the replacer asks for and runs, or

+// refuses, the write.

+type mpSelector struct {

+ host, action string

+ with int

+ identity hostwrite.Identity

+ deny error

+func (s *mpSelector) Select(_ context.Context, host, action string) (hostwrite.Work, error) {

+ s.host, s.action = host, action

+ return s, nil

+func (s *mpSelector) With(_ context.Context, id hostwrite.Identity, fn func() error) error {

+ s.with++

+ s.identity = id

+ if s.deny != nil {

+ return s.deny

+ }

+ return fn()

+// armPhase1 wires the landing's timers import through the replacer: with

+// the flag off it is the registry's replacement, as before.

+func TestArmPhase1WiresTheHeldTimersImport(t *testing.T) {

+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))

+ if err != nil {

+ t.Fatal(err)

+ }

+ defer reg.Close()

+ var cfg moveCfg

+ hib := activator.Config{Bucket: "data-uk", Region: "eu-west-2", Prefix: "instances"}

+ armPhase1(&cfg, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, t.TempDir(), reg, nil, nil, hostwrite.Config{}, func(string) string { return "" })

+ if cfg.ReplaceTimers == nil {

+ t.Fatal("no timers import")

+ }

+ rows := []registry.Timer{{Host: mpHost, Name: "nightly", Kind: "schedule", Spec: "0 3 * * *", Path: "/n", NextAt: 5000, CreatedAt: 10, DueAt: 4900, Registration: "reg-nightly"}}

+ if err := cfg.ReplaceTimers(t.Context(), mpHost, rows); err != nil {

+ t.Fatal(err)

+ }

+ if got, _ := reg.TimersFor(mpHost); len(got) != 1 || got[0].Name != "nightly" || got[0].Registration != "reg-nightly" {

+ t.Fatalf("rows: %+v", got)

+ }

+ // The flag on without a selector: the import refuses rather than

+ // writing outside the claim.

+ var on moveCfg

+ armPhase1(&on, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, t.TempDir(), reg, nil, nil, hostwrite.Config{Enabled: true}, func(string) string { return "" })

+ if err := on.ReplaceTimers(t.Context(), mpHost, nil); !errors.Is(err, hostwrite.ErrUnavailable) {

+ t.Fatal(err)

+ }

+ if got, _ := reg.TimersFor(mpHost); len(got) != 1 {

+ t.Fatalf("a refused import wrote: %+v", got)

+ }