amadan

carlos / platform Public

zdd-5 task 15: relay drops overtaken heads, moves worker stays out of the relay

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 54a2a5bdf2169ce9107b8e55db7f67e89e663120 parent 3e2280a
4 files changed, +215 −37
  • internal/host/agent.go +17 −13
  • internal/host/movespass.go +1 −1
  • internal/host/stagerelay.go +82 −15
  • internal/host/stagerelay_test.go +115 −8
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 931569fa4..2f6217609 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -435,12 +435,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {
if logSink != nil { // a typed nil in the interface would not be nil
rolloutSink = logSink
}
- if armRollout(agentCtx, os.Getenv, svc, reg, dd, rolloutSink, openDeployStore) != nil {
- // Spec §7.7: a stage touch lost to a running root pass is asked
- // again. Only with the rollout armed, so with CARLOS_SWAP off
- // the stage path is byte-for-byte what it was.
- armStageRelay(agentCtx, reg, dd)
- }
+ armRolloutAndRelay(agentCtx, os.Getenv, svc, reg, dd, rolloutSink, openDeployStore)
// 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
@@ -1435,13 +1430,7 @@ const stageRequestEvery = 15 * time.Second
// cannot be written costs latency, never correctness, because
// carlos-adopt.timer remains the floor underneath it.
func stageRequester(dd string) func(host, ver string, err error) {
- ask := markerRequester(dd, func(host, ver string, err error) {
- if err != nil {
- log.Printf("agent: cannot request root stage for %s %s: %v", host, ver, err)
- return
- }
- log.Printf("agent: %s needs %s staged and this process cannot write the version store — asked for a root pass", host, ver)
- })
+ ask := stageMarkerRequester(dd)
return func(host, ver string, err error) {
ask(host, ver, err)
// The stage relay (stagerelay.go) learns the head here and not in
@@ -1453,6 +1442,21 @@ func stageRequester(dd string) func(host, ver string, err error) {
}
}
+// stageMarkerRequester is stageRequester without the relay: the same marker,
+// throttle and log lines, and nothing remembered. The moves worker uses it
+// (movespass.go): its version is a move's held version, not a channel head,
+// and a held stage neither moves the installed Version nor leaves a refusal,
+// so the relay would have no way to know the request was answered.
+func stageMarkerRequester(dd string) func(host, ver string, err error) {
+ return markerRequester(dd, func(host, ver string, err error) {
+ if err != nil {
+ log.Printf("agent: cannot request root stage for %s %s: %v", host, ver, err)
+ return
+ }
+ log.Printf("agent: %s needs %s staged and this process cannot write the version store — asked for a root pass", host, ver)
+ })
+}
+
// mintRequester is the mint asker's (proofask.go) request: the same marker
// and the same throttle shape as stageRequester's, with its own budget and
// its own words. The asker logs why it asks; this logs only a failed write.
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index 2bf3d96d5..6d17c8640 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -362,7 +362,7 @@ func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir strin
// its miss cache holds between passes; its bake directory is the
// healthy history adoption reads.
cfg.ReleaseReady = heldReleaseReady(reg, cfg.OptDir, dataDir, cfg.Box,
- release.NewProofGate(release.ProofPolicyPath, dataDir), stageRequester(dataDir))
+ release.NewProofGate(release.ProofPolicyPath, dataDir), stageMarkerRequester(dataDir))
if svc != nil {
// S1 asks the instance itself, woken through the activator.
cfg.ReadyProbe = func(ctx context.Context, host, path string) moves.Check {
diff --git a/internal/host/stagerelay.go b/internal/host/stagerelay.go
index 98706337c..64be9d048 100644
--- a/internal/host/stagerelay.go
+++ b/internal/host/stagerelay.go
@@ -6,8 +6,10 @@ import (
"sync/atomic"
"time"
+ "github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/rollout"
)
// stageRelay retransmits the stage request (spec §7.7). carlos-stage.path
@@ -17,7 +19,7 @@ import (
// installed, or its refusal record names it. Armed only with the flag on.
type stageRelay struct {
mu sync.Mutex
- pending map[string]string // host -> the channel head a pass could not stage
+ pending map[string]pendingHead // host -> the channel head a pass could not stage
get func(host string) (*registry.Route, error)
// refused reads root's refusal record for a host (release.ReadRefusal
// over the data dir): nil when there is none.
@@ -25,43 +27,95 @@ type stageRelay struct {
touch func()
}
+// pendingHead is a head the relay is asking root to stage, with the version
+// the host had installed when it was noted (the baseline). Root records an
+// outcome for a host by moving its installed Version, so a Version that is
+// neither the head nor the baseline means root did something else, and a
+// head still worth staging will be noted again by the next blocked pass.
+type pendingHead struct {
+ ver string
+ base string
+ baseSet bool // false while the row could not be read at note time
+}
+
+// installed is the host's installed Version as of now: ok is false when the
+// row cannot be read, and a vanished route reads as "".
+func (r *stageRelay) installed(host string) (ver string, ok bool) {
+ rt, err := r.get(host)
+ if err != nil {
+ return "", false
+ }
+ if rt == nil {
+ return "", true
+ }
+ return rt.Version, true
+}
+
// note records the channel head a pass could not stage. A newer head for the
-// same host replaces the older one: only the newest is worth asking for.
+// same host replaces the older one: only the newest is worth asking for. The
+// same head noted again only refreshes the baseline, and only when it has
+// changed and the row can be read.
func (r *stageRelay) note(host, ver string) {
+ base, ok := r.installed(host) // outside the lock: a registry read
r.mu.Lock()
- r.pending[host] = ver
- r.mu.Unlock()
+ defer r.mu.Unlock()
+ p, have := r.pending[host]
+ if have && p.ver == ver {
+ if ok && (!p.baseSet || p.base != base) {
+ p.base, p.baseSet = base, true
+ r.pending[host] = p
+ }
+ return
+ }
+ r.pending[host] = pendingHead{ver: ver, base: base, baseSet: ok}
}
-// tick touches the marker once if any head is still outstanding, and drops
-// the ones root has installed or refused, or whose route is gone. A row or a
-// record that cannot be read is no outcome, so it keeps the head pending.
+// tick touches the marker once if any head is still outstanding. A head is
+// dropped when root has installed it, when the host's installed Version has
+// moved off the baseline (root recorded some outcome, for instance by
+// installing a newer head straight from the channel), when root's refusal
+// record names it, or when the route is gone. Without the baseline rule a
+// head that was overtaken would be asked for every stageRequestEvery, each
+// touch starting a root pass, until the route went. A row or a record that
+// cannot be read is no outcome, so it keeps the head pending.
func (r *stageRelay) tick() bool {
r.mu.Lock()
- snap := make(map[string]string, len(r.pending))
- for h, v := range r.pending {
- snap[h] = v
+ snap := make(map[string]pendingHead, len(r.pending))
+ for h, p := range r.pending {
+ snap[h] = p
}
r.mu.Unlock()
outstanding := false
- for host, ver := range snap {
+ for host, p := range snap {
rt, err := r.get(host)
if err != nil {
outstanding = true
continue
}
- done := rt == nil || rt.Version == ver
+ done := rt == nil || rt.Version == p.ver
+ if !done && !p.baseSet {
+ // The row was unreadable at note time: this read is the
+ // earliest baseline there is.
+ r.mu.Lock()
+ if cur, ok := r.pending[host]; ok && cur == p {
+ cur.base, cur.baseSet = rt.Version, true
+ r.pending[host] = cur
+ }
+ r.mu.Unlock()
+ } else if !done {
+ done = rt.Version != p.base
+ }
if !done {
rec, rerr := r.refused(host)
if rerr != nil {
outstanding = true
continue
}
- done = rec != nil && rec.Version == ver
+ done = rec != nil && rec.Version == p.ver
}
if done {
r.mu.Lock()
- if r.pending[host] == ver { // a newer head noted meanwhile stays
+ if r.pending[host].ver == p.ver { // a newer head noted meanwhile stays
delete(r.pending, host)
}
r.mu.Unlock()
@@ -97,7 +151,7 @@ var armedStageRelay atomic.Pointer[stageRelay]
// newStageRelay is the relay over root's real records: the registry's stamp
// and the refusal files in the agent's data dir (its bake dir).
func newStageRelay(reg *registry.Registry, dd string) *stageRelay {
- return &stageRelay{pending: map[string]string{}, get: reg.Get,
+ return &stageRelay{pending: map[string]pendingHead{}, get: reg.Get,
refused: func(host string) (*release.Refusal, error) { return release.ReadRefusal(dd, host) },
touch: func() { touchStageMarker(dd) }}
}
@@ -108,3 +162,16 @@ func armStageRelay(ctx context.Context, reg *registry.Registry, dd string) {
armedStageRelay.Store(r)
go r.run(ctx, stageRequestEvery)
}
+
+// armRolloutAndRelay arms the scheduler and, only when it armed, the stage
+// relay (spec §7.7): a stage touch lost to a running root pass is asked
+// again. With CARLOS_SWAP off armedStageRelay stays nil and the stage path is
+// byte-for-byte what it was.
+func armRolloutAndRelay(ctx context.Context, getenv func(string) string, svc *activator.Service, reg *registry.Registry, dd string,
+ sink activator.EventSink, openStore func(context.Context) (release.Store, error)) *rollout.Manager {
+ m := armRollout(ctx, getenv, svc, reg, dd, sink, openStore)
+ if m != nil {
+ armStageRelay(ctx, reg, dd)
+ }
+ return m
+}
diff --git a/internal/host/stagerelay_test.go b/internal/host/stagerelay_test.go
index 884f083ff..f3e096282 100644
--- a/internal/host/stagerelay_test.go
+++ b/internal/host/stagerelay_test.go
@@ -18,7 +18,7 @@ func noRefusal(string) (*release.Refusal, error) { return nil, nil }
func TestTheRelayAsksAgainUntilTheHeadIsInstalled(t *testing.T) {
rows := map[string]*registry.Route{"a": {Host: "a", Version: "v1"}, "b": {Host: "b", Version: "v1"}}
touches := 0
- r := &stageRelay{pending: map[string]string{},
+ r := &stageRelay{pending: map[string]pendingHead{},
get: func(h string) (*registry.Route, error) { return rows[h], nil },
refused: noRefusal,
touch: func() { touches++ }}
@@ -48,7 +48,7 @@ func TestTheRelayStopsAtRootsRefusal(t *testing.T) {
rows := map[string]*registry.Route{"a": {Host: "a", Version: "v1"}, "b": {Host: "b", Version: "v1"}}
refusals := map[string]*release.Refusal{"a": {Host: "a", Version: "v2", Kind: release.RefusalWrongProgram}}
touches := 0
- r := &stageRelay{pending: map[string]string{},
+ r := &stageRelay{pending: map[string]pendingHead{},
get: func(h string) (*registry.Route, error) { return rows[h], nil },
refused: func(h string) (*release.Refusal, error) { return refusals[h], nil },
touch: func() { touches++ }}
@@ -72,15 +72,14 @@ func TestTheRelayStopsAtRootsRefusal(t *testing.T) {
func TestTheRelayFollowsTheNewestHead(t *testing.T) {
rows := map[string]*registry.Route{"a": {Host: "a", Version: "v1"}}
touches := 0
- r := &stageRelay{pending: map[string]string{},
+ r := &stageRelay{pending: map[string]pendingHead{},
get: func(h string) (*registry.Route, error) { return rows[h], nil },
refused: noRefusal,
touch: func() { touches++ }}
r.note("a", "v2")
r.note("a", "v3")
- rows["a"].Version = "v2" // root installed the superseded head
if !r.tick() || touches != 1 {
- t.Fatal("the superseded head's install must not end the ask for the newer one")
+ t.Fatal("the newest head is outstanding")
}
rows["a"].Version = "v3"
if r.tick() || touches != 1 {
@@ -88,11 +87,88 @@ func TestTheRelayFollowsTheNewestHead(t *testing.T) {
}
}
+// A head the channel has moved past is dropped. Root installed v3 straight
+// from the channel, so v2 will never match the row, and no blocked pass
+// notes it again; asking for it every 15 s would start a root pass each time
+// until the route went.
+func TestTheRelayDropsAHeadTheHostMovedPast(t *testing.T) {
+ rows := map[string]*registry.Route{"a": {Host: "a", Version: "v1"}}
+ touches := 0
+ r := &stageRelay{pending: map[string]pendingHead{},
+ get: func(h string) (*registry.Route, error) { return rows[h], nil },
+ refused: noRefusal,
+ touch: func() { touches++ }}
+ r.note("a", "v2")
+ if !r.tick() || touches != 1 {
+ t.Fatal("v2 is outstanding while the row is still at v1")
+ }
+ rows["a"].Version = "v3"
+ if r.tick() || touches != 1 {
+ t.Fatal("kept touching for v2 after root installed v3")
+ }
+ if len(r.pending) != 0 {
+ t.Fatalf("the overtaken head stayed pending: %v", r.pending)
+ }
+}
+
+// The baseline is the Version at note time. Noting the same head again
+// refreshes it when it changed, so a head noted after root moved the host is
+// judged against the new Version and stays pending.
+func TestTheRelayRenoteRefreshesTheBaseline(t *testing.T) {
+ rows := map[string]*registry.Route{"a": {Host: "a", Version: "v1"}}
+ touches := 0
+ r := &stageRelay{pending: map[string]pendingHead{},
+ get: func(h string) (*registry.Route, error) { return rows[h], nil },
+ refused: noRefusal,
+ touch: func() { touches++ }}
+ r.note("a", "v4")
+ rows["a"].Version = "v3" // root installed something else; a blocked pass notes v4 again
+ r.note("a", "v4")
+ if !r.tick() || touches != 1 {
+ t.Fatal("a re-noted head must be judged against the Version at the re-note")
+ }
+ r.get = func(string) (*registry.Route, error) { return nil, errors.New("database is locked") }
+ r.note("a", "v4") // an unreadable row must not clobber the baseline
+ r.get = func(h string) (*registry.Route, error) { return rows[h], nil }
+ if !r.tick() {
+ t.Fatal("the baseline was lost to an unreadable row")
+ }
+}
+
+// A row unreadable at note time gets its baseline from the first readable
+// tick, and the head stays pending until then.
+func TestTheRelayBaselinesAtTheFirstReadableTick(t *testing.T) {
+ rows := map[string]*registry.Route{"a": {Host: "a", Version: "v1"}}
+ readable := false
+ touches := 0
+ r := &stageRelay{pending: map[string]pendingHead{},
+ get: func(h string) (*registry.Route, error) {
+ if !readable {
+ return nil, errors.New("database is locked")
+ }
+ return rows[h], nil
+ },
+ refused: noRefusal,
+ touch: func() { touches++ }}
+ r.note("a", "v2")
+ if !r.tick() {
+ t.Fatal("an unreadable row is not an outcome")
+ }
+ readable = true
+ if !r.tick() {
+ t.Fatal("v2 is still outstanding at the first readable tick")
+ }
+ rows["a"].Version = "v3"
+ if r.tick() {
+ t.Fatal("kept asking after the host moved off its baseline")
+ }
+}
+
// The loop runs on its own goroutine, touches on its own interval and stops
// when the agent's context is cancelled.
func TestTheRelayLoopTicksAndStopsWithTheContext(t *testing.T) {
var touches atomic.Int32
- r := &stageRelay{pending: map[string]string{},
+ r := &stageRelay{pending: map[string]pendingHead{},
get: func(h string) (*registry.Route, error) { return &registry.Route{Host: h, Version: "v1"}, nil },
refused: noRefusal,
touch: func() { touches.Add(1) }}
@@ -129,7 +205,7 @@ func TestOnlyStageRequestsFeedTheRelay(t *testing.T) {
if _, err := os.Stat(filepath.Join(dd, stageRequestMarker)); err != nil {
t.Fatalf("marker not written with no relay armed: %v", err)
}
- r := &stageRelay{pending: map[string]string{}, get: func(string) (*registry.Route, error) { return nil, nil }, refused: noRefusal, touch: func() {}}
+ r := &stageRelay{pending: map[string]pendingHead{}, get: func(string) (*registry.Route, error) { return nil, nil }, refused: noRefusal, touch: func() {}}
armedStageRelay.Store(r)
t.Cleanup(func() { armedStageRelay.Store(nil) })
mintRequester(dd)("m", "v9", nil)
@@ -137,7 +213,7 @@ func TestOnlyStageRequestsFeedTheRelay(t *testing.T) {
t.Fatalf("a mint request reached the relay: %v", r.pending)
}
stageRequester(dd)("a", "v2", nil)
- if r.pending["a"] != "v2" {
+ if r.pending["a"].ver != "v2" {
t.Fatalf("a stage-blocked head did not reach the relay: %v", r.pending)
}
}
@@ -164,3 +240,34 @@ func TestArmStageRelayReadsRootsRecords(t *testing.T) {
t.Fatal("kept asking after root wrote a refusal for the head")
}
}
+
+// The moves worker's requester (movespass.go) asks root for a pass like
+// stageRequester but never feeds the relay: its version is a move's held
+// version, and nothing would ever tell the relay it was answered.
+func TestTheMovesRequesterDoesNotFeedTheRelay(t *testing.T) {
+ dd := t.TempDir()
+ r := &stageRelay{pending: map[string]pendingHead{}, get: func(string) (*registry.Route, error) { return nil, nil }, refused: noRefusal, touch: func() {}}
+ armedStageRelay.Store(r)
+ t.Cleanup(func() { armedStageRelay.Store(nil) })
+ stageMarkerRequester(dd)("a", "v2", nil)
+ if _, err := os.Stat(filepath.Join(dd, stageRequestMarker)); err != nil {
+ t.Fatalf("the moves requester did not touch the marker: %v", err)
+ }
+ if len(r.pending) != 0 {
+ t.Fatalf("a moves-path request entered the relay: %v", r.pending)
+ }
+}
+
+// The flag-off guarantee lives on armedStageRelay staying nil: runAgent arms
+// the relay through armRolloutAndRelay, and with CARLOS_SWAP unset it must
+// leave nothing armed.
+func TestTheFlagOffLeavesTheStageRelayUnarmed(t *testing.T) {
+ armedStageRelay.Store(nil)
+ svc, reg, dd := rolloutFixture(t, false)
+ if m := armRolloutAndRelay(context.Background(), noEnv, svc, reg, dd, nil, unusedStore); m != nil {
+ t.Fatal("CARLOS_SWAP unset armed the scheduler")
+ }
+ if armedStageRelay.Load() != nil {
+ t.Fatal("CARLOS_SWAP unset armed the stage relay")
+ }
+}