fix: read heartbeats from the leader for activation and exit when the level rises above the build

spinifex/foundation/release/featurelevel/admit.go

@@ -0,0 +1,77 @@
+package featurelevel
+
+import (
+	"context"
+	"sync"
+)
+
+// Gate admits N-only work. It counts a caller in flight before it reads the
+// record, so a rollback that sees zero in flight and an acknowledged revision
+// cannot be overtaken by a request that passed the check and then paused.
+type Gate struct {
+	store *Store
+
+	mu       sync.Mutex
+	inFlight map[int]int
+}
+
+// NewGate returns a gate over store.
+func NewGate(store *Store) *Gate {
+	return &Gate{store: store, inFlight: map[int]int{}}
+}
+
+// Admit reports whether level n is active: the recorded level is at least n and
+// no rollback below n is in progress. The caller holds done until its obligation
+// is durably recorded or abandoned, and calls it exactly once on every path,
+// including when err is non-nil. A read failure returns an error and false; each
+// feature decides its safe answer. It is one uncached read.
+func (g *Gate) Admit(ctx context.Context, n int) (done func(), ok bool, err error) {
+	g.mu.Lock()
+	g.inFlight[n]++
+	g.mu.Unlock()
+	var once sync.Once
+	done = func() {
+		once.Do(func() {
+			g.mu.Lock()
+			g.inFlight[n]--
+			g.mu.Unlock()
+		})
+	}
+
+	rec, _, err := g.store.Read(ctx)
+	if err != nil {
+		return done, false, err
+	}
+	return done, active(rec, n), nil
+}
+
+func active(rec Record, n int) bool {
+	if rec.Level < n {
+		return false
+	}
+	return rec.Rollback == nil || rec.Rollback.Target >= n
+}
+
+// InFlight returns the callers currently admitted per level.
+func (g *Gate) InFlight() map[int]int {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	out := map[int]int{}
+	for n, c := range g.inFlight {
+		if c > 0 {
+			out[n] = c
+		}
+	}
+	return out
+}
+
+// InFlightAbove counts the callers admitted for levels above target.
+func (g *Gate) InFlightAbove(target int) int {
+	total := 0
+	for n, c := range g.InFlight() {
+		if n > target {
+			total += c
+		}
+	}
+	return total
+}

spinifex/foundation/release/featurelevel/featurelevel.go

@@ -0,0 +1,106 @@
+// Package featurelevel is the cluster feature level: a recorded integer that
+// advances once every member runs a build supporting it and only a rollback
+// lowers. It imports no domain; a feature supplies its own drain check.
+package featurelevel
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+	"sync"
+)
+
+// Build is the highest level this binary supports. Any change that adds an
+// N-only semantic raises it and declares the level below.
+const Build = 1
+
+// Level names one N-only semantic.
+type Level struct {
+	N    int
+	Name string
+}
+
+// Baseline is level 1: this build's first release, with no N-only semantic.
+const Baseline = 1
+
+var declared = []Level{{N: Baseline, Name: "baseline"}}
+
+// Levels returns every declared level in ascending order.
+func Levels() []Level { return slices.Clone(declared) }
+
+// DrainCheck reports how many obligations of a level are still outstanding, so
+// a rollback can wait for them instead of discarding or reinterpreting them.
+type DrainCheck func(ctx context.Context) (outstanding int, err error)
+
+// NothingToDrain is the check for a level that creates no obligations.
+func NothingToDrain(context.Context) (int, error) { return 0, nil }
+
+var (
+	drainMu sync.Mutex
+	drains  = map[int]DrainCheck{Baseline: NothingToDrain}
+)
+
+// RegisterDrain declares level n's one drain check. A level that is not
+// declared, or already has a check, is a programming error.
+func RegisterDrain(n int, check DrainCheck) error {
+	drainMu.Lock()
+	defer drainMu.Unlock()
+	if !slices.ContainsFunc(declared, func(l Level) bool { return l.N == n }) {
+		return fmt.Errorf("featurelevel: level %d is not declared", n)
+	}
+	if check == nil {
+		return fmt.Errorf("featurelevel: level %d drain check is nil", n)
+	}
+	if _, dup := drains[n]; dup {
+		return fmt.Errorf("featurelevel: level %d already has a drain check", n)
+	}
+	drains[n] = check
+	return nil
+}
+
+// DrainChecks returns the registered checks for every level above target.
+func DrainChecks(target int) map[int]DrainCheck {
+	drainMu.Lock()
+	defer drainMu.Unlock()
+	out := map[int]DrainCheck{}
+	for n, c := range drains {
+		if n > target {
+			out[n] = c
+		}
+	}
+	return out
+}
+
+// UncoveredLevels lists the declared levels with no drain check. It must be
+// empty at the composition root.
+func UncoveredLevels() []int {
+	drainMu.Lock()
+	defer drainMu.Unlock()
+	return uncovered(declared, drains)
+}
+
+func uncovered(levels []Level, checks map[int]DrainCheck) []int {
+	var missing []int
+	for _, l := range levels {
+		if checks[l.N] == nil {
+			missing = append(missing, l.N)
+		}
+	}
+	return missing
+}
+
+var (
+	// ErrCLITooOld is returned before anything is written when the recorded
+	// level is above the caller's Build: it cannot know the newer levels' drains.
+	ErrCLITooOld = errors.New("featurelevel: recorded level is above this build")
+	// ErrNoRollback is returned by a rollback step with no rollback recorded.
+	ErrNoRollback = errors.New("featurelevel: no rollback in progress")
+	// ErrRollbackInProgress is returned when a different rollback is already recorded.
+	ErrRollbackInProgress = errors.New("featurelevel: a different rollback is in progress")
+	// ErrRollbackChanged is returned when the rollback was cancelled or restarted
+	// after its checks ran, so they no longer describe the record.
+	ErrRollbackChanged = errors.New("featurelevel: the rollback changed since it was verified")
+	// ErrBadTarget is returned for a rollback target that is not below the level.
+	ErrBadTarget = errors.New("featurelevel: rollback target must be below the recorded level and not negative")
+)

spinifex/foundation/release/featurelevel/featurelevel_test.go

@@ -0,0 +1,348 @@
+package featurelevel
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func newStore(t *testing.T) (*Store, jetstream.KeyValue) {
+	t.Helper()
+	_, _, js := testutil.StartTestJetStream(t)
+	kv, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: "cluster-config-test", History: 1})
+	require.NoError(t, err)
+	return NewStore(kv), kv
+}
+
+func members(level int, current ...bool) []Member {
+	var out []Member
+	for i, c := range current {
+		out = append(out, Member{Node: string(rune('a' + i)), Level: level, Current: c})
+	}
+	return out
+}
+
+func TestAbsentRecordReadsAsLevelZeroAndGarbageIsAnError(t *testing.T) {
+	t.Parallel()
+	s, kv := newStore(t)
+	rec, rev, err := s.Read(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, Record{}, rec)
+	assert.Zero(t, rev)
+
+	_, err = kv.Put(t.Context(), Key, []byte("{not json"))
+	require.NoError(t, err)
+	_, _, err = s.Read(t.Context())
+	require.Error(t, err)
+}
+
+func TestAdvance_OnlyWhenEveryMemberIsCurrentAndNeverLowers(t *testing.T) {
+	t.Parallel()
+	s, _ := newStore(t)
+
+	rec, advanced, err := s.Advance(t.Context(), members(1, true, false, true))
+	require.NoError(t, err)
+	assert.False(t, advanced, "a stale member blocks advance")
+	assert.Zero(t, rec.Level)
+
+	_, advanced, err = s.Advance(t.Context(), []Member{{Node: "a", Level: 1, Current: true}, {Node: "b", Level: 0, Current: true}})
+	require.NoError(t, err)
+	assert.False(t, advanced, "a member reporting an absent field (0) blocks advance")
+
+	_, advanced, err = s.Advance(t.Context(), nil)
+	require.NoError(t, err)
+	assert.False(t, advanced, "an unknown member set never advances")
+
+	rec, advanced, err = s.Advance(t.Context(), members(1, true, true, true))
+	require.NoError(t, err)
+	assert.True(t, advanced)
+	assert.Equal(t, 1, rec.Level)
+
+	rec, advanced, err = s.Advance(t.Context(), members(1, true, false, true))
+	require.NoError(t, err)
+	assert.False(t, advanced)
+	assert.Equal(t, 1, rec.Level, "a member going stale afterwards does not lower the level")
+
+	rec, _, err = s.Advance(t.Context(), members(0, true, true))
+	require.NoError(t, err)
+	assert.Equal(t, 1, rec.Level, "members reporting lower never lower it")
+}
+
+func TestAdvance_IsCappedAtThisBuild(t *testing.T) {
+	t.Parallel()
+	s, _ := newStore(t)
+	rec, _, err := s.Advance(t.Context(), members(Build+5, true, true))
+	require.NoError(t, err)
+	assert.Equal(t, Build, rec.Level, "a daemon cannot advance past a level it does not understand")
+}
+
+func TestAdvance_ConcurrentDaemonsLeaveOneConsistentRecord(t *testing.T) {
+	t.Parallel()
+	s, _ := newStore(t)
+	var wg sync.WaitGroup
+	var advanced atomic.Int32
+	for range 8 {
+		wg.Go(func() {
+			_, ok, err := s.Advance(t.Context(), members(1, true, true, true))
+			assert.NoError(t, err)
+			if ok {
+				advanced.Add(1)
+			}
+		})
+	}
+	wg.Wait()
+	assert.EqualValues(t, 1, advanced.Load(), "the compare-and-set makes exactly one writer win")
+	rec, _, err := s.Read(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, Record{Level: 1}, rec)
+}
+
+func TestHold_CapsAdvance(t *testing.T) {
+	t.Parallel()
+	s, kv := newStore(t)
+	_, err := kv.Create(t.Context(), Key, []byte(`{"level":0,"hold":{"held":true,"level":0}}`))
+	require.NoError(t, err)
+	rec, advanced, err := s.Advance(t.Context(), members(1, true, true))
+	require.NoError(t, err)
+	assert.False(t, advanced)
+	assert.Zero(t, rec.Level, "a hold at 0 is a hold, not no hold")
+
+	_, err = s.ReleaseHold(t.Context())
+	require.NoError(t, err)
+	rec, advanced, err = s.Advance(t.Context(), members(1, true, true))
+	require.NoError(t, err)
+	assert.True(t, advanced)
+	assert.Equal(t, 1, rec.Level)
+}
+
+func TestRollbackToZeroHoldsAgainstRepeatedTicks(t *testing.T) {
+	t.Parallel()
+	s, _ := newStore(t)
+	_, _, err := s.Advance(t.Context(), members(1, true, true))
+	require.NoError(t, err)
+
+	rec, rev, err := s.StartRollback(t.Context(), 0, Build)
+	require.NoError(t, err)
+	assert.Equal(t, &Rollback{Target: 0}, rec.Rollback)
+	assert.Equal(t, Hold{Held: true, Level: 0}, rec.Hold)
+	assert.NotZero(t, rev)
+
+	_, advanced, err := s.Advance(t.Context(), members(1, true, true))
+	require.NoError(t, err)
+	assert.False(t, advanced, "advance stops while the rollback is in progress")
+
+	rec, err = s.FinishRollback(t.Context(), rev, 0)
+	require.NoError(t, err)
+	assert.Zero(t, rec.Level)
+	assert.Nil(t, rec.Rollback)
+	assert.Equal(t, Hold{Held: true, Level: 0}, rec.Hold, "the hold outlives the rollback")
+
+	for range 5 {
+		rec, advanced, err = s.Advance(t.Context(), members(1, true, true, true))
+		require.NoError(t, err)
+		assert.False(t, advanced)
+		assert.Zero(t, rec.Level, "running newer builds do not re-advance a level rolled back to 0")
+	}
+
+	_, err = s.ReleaseHold(t.Context())
+	require.NoError(t, err)
+	rec, advanced, err = s.Advance(t.Context(), members(1, true, true))
+	require.NoError(t, err)
+	assert.True(t, advanced)
+	assert.Equal(t, 1, rec.Level)
+}
+
+func TestRollback_StateTransitions(t *testing.T) {
+	t.Parallel()
+	s, _ := newStore(t)
+	_, _, err := s.StartRollback(t.Context(), 0, Build)
+	require.ErrorIs(t, err, ErrBadTarget, "nothing above level 0 to roll back")
+	_, err = s.FinishRollback(t.Context(), 1, 0)
+	require.ErrorIs(t, err, ErrNoRollback)
+	_, err = s.CancelRollback(t.Context())
+	require.ErrorIs(t, err, ErrNoRollback)
+
+	_, _, err = s.Advance(t.Context(), members(1, true, true))
+	require.NoError(t, err)
+	_, _, err = s.StartRollback(t.Context(), 1, Build)
+	require.ErrorIs(t, err, ErrBadTarget)
+	_, _, err = s.StartRollback(t.Context(), -1, Build)
+	require.ErrorIs(t, err, ErrBadTarget)
+
+	first, rev1, err := s.StartRollback(t.Context(), 0, Build)
+	require.NoError(t, err)
+	again, rev2, err := s.StartRollback(t.Context(), 0, Build)
+	require.NoError(t, err)
+	assert.Equal(t, first, again, "repeating the same rollback resumes it")
+	assert.Equal(t, rev1, rev2, "and writes nothing")
+
+	_, err = s.ReleaseHold(t.Context())
+	require.ErrorIs(t, err, ErrRollbackInProgress)
+
+	rec, err := s.CancelRollback(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, Record{Level: 1}, rec, "cancel clears the rollback and the hold it set")
+	_, advanced, err := s.Advance(t.Context(), members(1, true, true))
+	require.NoError(t, err)
+	assert.False(t, advanced, "already at level 1")
+}
+
+func TestStartRollback_CLIBelowRecordedLevelRefusesWithoutWriting(t *testing.T) {
+	t.Parallel()
+	s, kv := newStore(t)
+	_, err := kv.Create(t.Context(), Key, []byte(`{"level":2,"hold":{"held":false,"level":0}}`))
+	require.NoError(t, err)
+	_, before, err := s.Read(t.Context())
+	require.NoError(t, err)
+
+	_, _, err = s.StartRollback(t.Context(), 0, 1)
+	require.ErrorIs(t, err, ErrCLITooOld)
+	rec, after, err := s.Read(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, before, after, "no write")
+	assert.Nil(t, rec.Rollback)
+}
+
+func TestAdmit_FalseAboveARollbackTargetFromTheMomentItIsWritten(t *testing.T) {
+	t.Parallel()
+	s, _ := newStore(t)
+	g := NewGate(s)
+
+	done, ok, err := g.Admit(t.Context(), 1)
+	require.NoError(t, err)
+	done()
+	assert.False(t, ok, "level 0 does not admit level 1")
+
+	_, _, err = s.Advance(t.Context(), members(1, true, true))
+	require.NoError(t, err)
+	done, ok, err = g.Admit(t.Context(), 1)
+	require.NoError(t, err)
+	done()
+	assert.True(t, ok)
+
+	_, _, err = s.StartRollback(t.Context(), 0, Build)
+	require.NoError(t, err)
+	done, ok, err = g.Admit(t.Context(), 1)
+	require.NoError(t, err)
+	done()
+	assert.False(t, ok, "false for N above the target the moment the rollback is written")
+	done, ok, err = g.Admit(t.Context(), 0)
+	require.NoError(t, err)
+	done()
+	assert.True(t, ok, "levels at or below the target stay active")
+}
+
+// gatedKV blocks Get until released, to pause a request between counting it in
+// flight and reading the record.
+type gatedKV struct {
+	KV
+
+	entered chan struct{}
+	release chan struct{}
+	fail    bool
+}
+
+func (k *gatedKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) {
+	if k.entered != nil {
+		close(k.entered)
+		<-k.release
+	}
+	if k.fail {
+		return nil, context.DeadlineExceeded
+	}
+	return k.KV.Get(ctx, key)
+}
+
+func TestAdmit_CountsInFlightBeforeReadingAndUntilDone(t *testing.T) {
+	t.Parallel()
+	s, kv := newStore(t)
+	_, _, err := s.Advance(t.Context(), members(1, true, true))
+	require.NoError(t, err)
+	paused := &gatedKV{KV: kv, entered: make(chan struct{}), release: make(chan struct{})}
+	g := NewGate(NewStore(paused))
+
+	type result struct {
+		done func()
+		ok   bool
+	}
+	res := make(chan result, 1)
+	go func() {
+		done, ok, err := g.Admit(t.Context(), 1)
+		assert.NoError(t, err)
+		res <- result{done, ok}
+	}()
+	<-paused.entered
+	assert.Equal(t, 1, g.InFlightAbove(0), "counted before the record is read")
+
+	close(paused.release)
+	r := <-res
+	assert.True(t, r.ok)
+	assert.Equal(t, 1, g.InFlightAbove(0), "still counted after admission until the caller finishes recording")
+	assert.Zero(t, g.InFlightAbove(1), "only levels above the target count")
+	r.done()
+	r.done()
+	assert.Zero(t, g.InFlightAbove(0), "done is idempotent and releases the count")
+	assert.Empty(t, g.InFlight())
+}
+
+func TestAdmit_ReadFailureIsAnErrorAndStillReleasesTheCount(t *testing.T) {
+	t.Parallel()
+	s, kv := newStore(t)
+	_ = s
+	g := NewGate(NewStore(&gatedKV{KV: kv, fail: true}))
+	done, ok, err := g.Admit(t.Context(), 1)
+	require.Error(t, err)
+	assert.False(t, ok)
+	assert.Equal(t, 1, g.InFlightAbove(0))
+	done()
+	assert.Zero(t, g.InFlightAbove(0))
+}
+
+func TestEveryDeclaredLevelHasExactlyOneDrainCheck(t *testing.T) {
+	t.Parallel()
+	assert.Empty(t, UncoveredLevels(), "every declared level must register a drain check")
+	require.NotEmpty(t, Levels())
+	assert.Equal(t, Build, Levels()[len(Levels())-1].N, "Build is the highest declared level")
+
+	assert.Equal(t, []int{2}, uncovered([]Level{{N: 1}, {N: 2}}, map[int]DrainCheck{1: NothingToDrain}), "a level with no check is reported")
+	require.Error(t, RegisterDrain(Baseline, NothingToDrain), "a second check for a level is refused")
+	require.Error(t, RegisterDrain(Build+10, NothingToDrain), "an undeclared level is refused")
+	require.Error(t, RegisterDrain(Baseline, nil))
+	assert.Contains(t, DrainChecks(0), Baseline)
+	assert.NotContains(t, DrainChecks(Baseline), Baseline, "only levels above the target drain")
+}
+
+func TestCancelRollback_RestoresThePriorHold(t *testing.T) {
+	t.Parallel()
+	s, kv := newStore(t)
+	// Held at 1 after an earlier rollback that was never released.
+	_, err := kv.Create(t.Context(), Key, []byte(`{"level":1,"hold":{"held":true,"level":1}}`))
+	require.NoError(t, err)
+
+	rec, _, err := s.StartRollback(t.Context(), 0, Build)
+	require.NoError(t, err)
+	assert.Equal(t, Hold{Held: true, Level: 0}, rec.Hold)
+	assert.Equal(t, Hold{Held: true, Level: 1}, rec.Rollback.PriorHold)
+
+	rec, err = s.CancelRollback(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, Hold{Held: true, Level: 1}, rec.Hold, "the hold that existed before the rollback is back, not cleared")
+	assert.Nil(t, rec.Rollback)
+	assert.Equal(t, 1, rec.Level)
+
+	// A cluster that had no hold goes back to no hold.
+	s2, _ := newStore(t)
+	_, _, err = s2.Advance(t.Context(), members(1, true, true))
+	require.NoError(t, err)
+	_, _, err = s2.StartRollback(t.Context(), 0, Build)
+	require.NoError(t, err)
+	rec, err = s2.CancelRollback(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, Hold{}, rec.Hold)
+}

spinifex/foundation/release/featurelevel/leaderkv.go

@@ -0,0 +1,55 @@
+package featurelevel
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// leaderKV reads through the stream leader so a lagging replica never returns
+// the record from before a level was raised or a rollback began.
+type leaderKV struct {
+	jetstream.KeyValue
+
+	js jetstream.JetStream
+}
+
+var _ KV = (*leaderKV)(nil)
+
+// LeaderKV returns kv with Get answered by the stream leader. Writes are
+// unchanged: the leader already serialises them.
+func LeaderKV(js jetstream.JetStream, kv jetstream.KeyValue) KV {
+	return &leaderKV{KeyValue: kv, js: js}
+}
+
+func (k *leaderKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) {
+	bucket := k.Bucket()
+	msg, err := kvstore.LeaderMsgGet(ctx, k.js, "KV_"+bucket, kvstore.MsgGetRequest{LastFor: "$KV." + bucket + "." + key})
+	if errors.Is(err, jetstream.ErrMsgNotFound) {
+		return nil, jetstream.ErrKeyNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+	if op := msg.Header.Get("KV-Operation"); op == "DEL" || op == "PURGE" {
+		return nil, jetstream.ErrKeyNotFound
+	}
+	return &leaderEntry{bucket: bucket, key: key, msg: msg}, nil
+}
+
+// leaderEntry is the value and revision of one leader read.
+type leaderEntry struct {
+	bucket, key string
+	msg         *jetstream.RawStreamMsg
+}
+
+func (e *leaderEntry) Bucket() string                  { return e.bucket }
+func (e *leaderEntry) Key() string                     { return e.key }
+func (e *leaderEntry) Value() []byte                   { return e.msg.Data }
+func (e *leaderEntry) Revision() uint64                { return e.msg.Sequence }
+func (e *leaderEntry) Created() time.Time              { return e.msg.Time }
+func (e *leaderEntry) Delta() uint64                   { return 0 }
+func (e *leaderEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut }

spinifex/foundation/release/featurelevel/leaderkv_test.go

@@ -0,0 +1,73 @@
+package featurelevel
+
+import (
+	"strings"
+	"sync"
+	"testing"
+
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Reads of the record go to the stream leader (a message get), never a direct
+// get that a lagging follower may answer.
+func TestLeaderKV_ReadsThroughTheLeaderNotADirectGet(t *testing.T) {
+	t.Parallel()
+	_, nc, js := testutil.StartTestJetStream(t)
+	kv, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: "cluster-config-leader", History: 1})
+	require.NoError(t, err)
+	s := NewStore(LeaderKV(js, kv))
+
+	var mu sync.Mutex
+	var subjects []string
+	sub, err := nc.Subscribe("$JS.API.>", func(m *nats.Msg) {
+		mu.Lock()
+		subjects = append(subjects, m.Subject)
+		mu.Unlock()
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+
+	rec, rev, err := s.Read(t.Context())
+	require.NoError(t, err)
+	assert.Zero(t, rev)
+	assert.Zero(t, rec.Level, "an absent key reads as level 0")
+
+	_, adv, err := s.Advance(t.Context(), members(1, true))
+	require.NoError(t, err)
+	require.True(t, adv)
+	rec, rev, err = s.Read(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, 1, rec.Level)
+	assert.NotZero(t, rev)
+
+	require.NoError(t, nc.Flush())
+	mu.Lock()
+	defer mu.Unlock()
+	gets := 0
+	for _, subj := range subjects {
+		assert.False(t, strings.HasPrefix(subj, "$JS.API.DIRECT.GET."), "feature level read used a direct get: %s", subj)
+		if subj == "$JS.API.STREAM.MSG.GET.KV_cluster-config-leader" {
+			gets++
+		}
+	}
+	assert.GreaterOrEqual(t, gets, 3, "every Read is a leader message get")
+}
+
+func TestLeaderKV_ADeletedKeyReadsAsAbsent(t *testing.T) {
+	t.Parallel()
+	_, _, js := testutil.StartTestJetStream(t)
+	kv, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: "cluster-config-del", History: 1})
+	require.NoError(t, err)
+	s := NewStore(LeaderKV(js, kv))
+	_, _, err = s.Advance(t.Context(), members(1, true))
+	require.NoError(t, err)
+	require.NoError(t, kv.Delete(t.Context(), Key))
+	rec, rev, err := s.Read(t.Context())
+	require.NoError(t, err)
+	assert.Zero(t, rec.Level)
+	assert.Zero(t, rev)
+}

spinifex/foundation/release/featurelevel/rollback.go

@@ -0,0 +1,153 @@
+package featurelevel
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+	"strings"
+	"time"
+)
+
+const (
+	// DefaultAckTimeout bounds how long a rollback waits for every member to
+	// acknowledge it.
+	DefaultAckTimeout = 60 * time.Second
+	ackPollInterval   = 2 * time.Second
+)
+
+// InFlightAbove counts the callers the replying node has admitted for levels
+// above target.
+func (r *StatusReply) InFlightAbove(target int) int {
+	total := 0
+	for n, c := range r.InFlight {
+		if n > target {
+			total += c
+		}
+	}
+	return total
+}
+
+// RollbackDeps are the collaborators a rollback runs against.
+type RollbackDeps struct {
+	Store *Store
+	// Probe asks every configured member for its status.
+	Probe func(ctx context.Context) []NodeStatus
+	// Drains returns the registered drain checks above a target.
+	Drains func(target int) map[int]DrainCheck
+	// Build is the running CLI's own build.
+	Build int
+	// AckTimeout and PollInterval default to DefaultAckTimeout and 2s.
+	AckTimeout   time.Duration
+	PollInterval time.Duration
+}
+
+// UnackedError names the members that did not acknowledge the rollback.
+type UnackedError struct {
+	Target   int
+	Revision uint64
+	Nodes    []string
+	Reasons  map[string]string
+}
+
+func (e *UnackedError) Error() string {
+	parts := make([]string, 0, len(e.Nodes))
+	for _, n := range e.Nodes {
+		parts = append(parts, n+" ("+e.Reasons[n]+")")
+	}
+	return fmt.Sprintf("rollback to %d is recorded but not acknowledged by: %s; it stays in place, rerun the same command to resume or use --cancel",
+		e.Target, strings.Join(parts, ", "))
+}
+
+// DrainOutstandingError names a level whose drain check found work left.
+type DrainOutstandingError struct {
+	Level       int
+	Outstanding int
+	Err         error
+}
+
+func (e *DrainOutstandingError) Error() string {
+	if e.Err != nil {
+		return fmt.Sprintf("rollback blocked: the level %d drain check failed: %v; the rollback stays in place", e.Level, e.Err)
+	}
+	return fmt.Sprintf("rollback blocked: level %d has %d outstanding item(s) to drain; the rollback stays in place, rerun the same command once they are gone", e.Level, e.Outstanding)
+}
+
+// ackedBy returns the nodes that have not acknowledged: each must have answered
+// with a record at or after rev and nothing in flight above target.
+func ackedBy(statuses []NodeStatus, rev uint64, target int) (nodes []string, reasons map[string]string) {
+	reasons = map[string]string{}
+	for _, s := range statuses {
+		switch {
+		case s.Reply == nil:
+			reasons[s.Node] = "no usable answer: " + s.Reason
+		case s.Reply.Revision < rev:
+			reasons[s.Node] = fmt.Sprintf("has read revision %d, need %d", s.Reply.Revision, rev)
+		case s.Reply.InFlightAbove(target) > 0:
+			reasons[s.Node] = fmt.Sprintf("%d in flight above level %d", s.Reply.InFlightAbove(target), target)
+		default:
+			continue
+		}
+		nodes = append(nodes, s.Node)
+	}
+	slices.Sort(nodes)
+	return nodes, reasons
+}
+
+// RunRollback lowers the cluster level to target. It records the rollback, waits
+// until every member has read it and has nothing admitted above target, runs
+// every drain check for the levels being given up, and only then lowers the
+// level. Any failure leaves the rollback recorded so the same command resumes.
+func RunRollback(ctx context.Context, target int, d RollbackDeps) (Record, error) {
+	rec, rev, err := d.Store.StartRollback(ctx, target, d.Build)
+	if err != nil {
+		return rec, err
+	}
+	recorded := rec.Level
+
+	timeout, interval := d.AckTimeout, d.PollInterval
+	if timeout <= 0 {
+		timeout = DefaultAckTimeout
+	}
+	if interval <= 0 {
+		interval = ackPollInterval
+	}
+	deadline := time.Now().Add(timeout)
+	for {
+		pctx, cancel := context.WithDeadline(ctx, deadline)
+		statuses := d.Probe(pctx)
+		cancel()
+		nodes, reasons := ackedBy(statuses, rev, target)
+		if len(statuses) > 0 && len(nodes) == 0 {
+			break
+		}
+		if ctx.Err() != nil || !time.Now().Add(interval).Before(deadline) {
+			if len(statuses) == 0 {
+				return rec, errors.New("rollback has no configured members to acknowledge it")
+			}
+			return rec, &UnackedError{Target: target, Revision: rev, Nodes: nodes, Reasons: reasons}
+		}
+		select {
+		case <-ctx.Done():
+		case <-time.After(interval):
+		}
+	}
+
+	checks := d.Drains(target)
+	for n := target + 1; n <= recorded; n++ {
+		check := checks[n]
+		if check == nil {
+			return rec, &DrainOutstandingError{Level: n, Err: errors.New("no drain check is registered")}
+		}
+		outstanding, cerr := check(ctx)
+		if cerr != nil {
+			return rec, &DrainOutstandingError{Level: n, Err: cerr}
+		}
+		if outstanding > 0 {
+			return rec, &DrainOutstandingError{Level: n, Outstanding: outstanding}
+		}
+	}
+	return d.Store.FinishRollback(ctx, rev, target)
+}
+
+func (e *DrainOutstandingError) Unwrap() error { return e.Err }

spinifex/foundation/release/featurelevel/rollback_integration_test.go

@@ -0,0 +1,102 @@
+package featurelevel
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// serveGate answers node's status route from the node's real gate, the way a
+// daemon's responder does.
+func serveGate(t *testing.T, nc *nats.Conn, node string, g *Gate) {
+	t.Helper()
+	handle := g.ServeStatus(node, Build)
+	sub, err := nc.Subscribe(StatusSubject(node), func(m *nats.Msg) {
+		var req StatusRequest
+		if json.Unmarshal(m.Data, &req) != nil {
+			return
+		}
+		reply, err := handle(t.Context(), &req, "")
+		if err != nil {
+			_ = m.Respond(nil)
+			return
+		}
+		out := nats.NewMsg("")
+		out.Data, _ = json.Marshal(reply)
+		out.Header.Set(natsmsg.NodeIDHeader, node)
+		_ = m.RespondMsg(out)
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+}
+
+// A request admitted before the rollback was recorded keeps the rollback from
+// finishing until it is done, through the real gate, responder, probe and runner.
+func TestRollback_AnAdmittedRequestBlocksTheFinishUntilItIsReleased(t *testing.T) {
+	t.Parallel()
+	_, nc, js := testutil.StartTestJetStream(t)
+	kv, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: "cluster-config-it", History: 1})
+	require.NoError(t, err)
+	store := NewStore(LeaderKV(js, kv))
+	_, adv, err := store.Advance(t.Context(), members(1, true, true))
+	require.NoError(t, err)
+	require.True(t, adv)
+
+	gate := NewGate(store)
+	serveGate(t, nc, "n1", gate)
+	serveGate(t, nc, "n2", NewGate(store))
+
+	done, ok, err := gate.Admit(t.Context(), 1)
+	require.NoError(t, err)
+	require.True(t, ok)
+
+	probe := NewProbe(nc, []string{"n1", "n2"})
+	probe.wait, probe.grace = 400*time.Millisecond, 20*time.Millisecond
+	d := RollbackDeps{
+		Store: store, Probe: probe.Probe, Drains: DrainChecks, Build: Build,
+		AckTimeout: 20 * time.Second, PollInterval: 50 * time.Millisecond,
+	}
+
+	result := make(chan error, 1)
+	go func() {
+		_, err := RunRollback(t.Context(), 0, d)
+		result <- err
+	}()
+
+	// While the admitted request is held the rollback is recorded but unfinished.
+	require.Eventually(t, func() bool {
+		rec, _, rerr := store.Read(t.Context())
+		return rerr == nil && rec.Rollback != nil
+	}, 5*time.Second, 20*time.Millisecond)
+	select {
+	case err := <-result:
+		t.Fatalf("rollback finished with an admitted request still held: %v", err)
+	case <-time.After(600 * time.Millisecond):
+	}
+	rec, _, err := store.Read(t.Context())
+	require.NoError(t, err)
+	assert.Equal(t, 1, rec.Level)
+	refusedDone, ok, err := gate.Admit(t.Context(), 1)
+	refusedDone()
+	require.NoError(t, err)
+	assert.False(t, ok, "new admissions above the target are refused while the rollback is recorded")
+
+	done()
+	select {
+	case err := <-result:
+		require.NoError(t, err)
+	case <-time.After(15 * time.Second):
+		t.Fatal("rollback did not finish after the admitted request was released")
+	}
+	rec, _, err = store.Read(t.Context())
+	require.NoError(t, err)
+	assert.Zero(t, rec.Level)
+	assert.Nil(t, rec.Rollback)
+}

spinifex/foundation/release/featurelevel/rollback_test.go

@@ -0,0 +1,232 @@
+package featurelevel
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func atLevelOne(t *testing.T) *Store {
+	t.Helper()
+	s, _ := newStore(t)
+	_, adv, err := s.Advance(t.Context(), members(1, true, true))
+	require.NoError(t, err)
+	require.True(t, adv)
+	return s
+}
+
+// ackFor answers the way a healthy node does once it has read the rollback.
+func ackFor(s *Store, node string, inFlight map[int]int) NodeStatus {
+	_, rev, _ := s.Read(context.Background())
+	total := 0
+	for _, c := range inFlight {
+		total += c
+	}
+	return NodeStatus{Node: node, Reply: &StatusReply{Node: node, Build: Build, Revision: rev, InFlight: inFlight, InFlightTotal: total}}
+}
+
+func deps(s *Store, probe func(context.Context) []NodeStatus, drains func(int) map[int]DrainCheck) RollbackDeps {
+	if drains == nil {
+		drains = func(int) map[int]DrainCheck { return map[int]DrainCheck{1: NothingToDrain} }
+	}
+	return RollbackDeps{Store: s, Probe: probe, Drains: drains, Build: Build, AckTimeout: 150 * time.Millisecond, PollInterval: 10 * time.Millisecond}
+}
+
+func TestRollback_FullSuccessLowersTheLevelAndKeepsTheHold(t *testing.T) {
+	t.Parallel()
+	s := atLevelOne(t)
+	probe := func(context.Context) []NodeStatus { return []NodeStatus{ackFor(s, "a", nil), ackFor(s, "b", nil)} }
+	rec, err := RunRollback(t.Context(), 0, deps(s, probe, nil))
+	require.NoError(t, err)
+	assert.Zero(t, rec.Level)
+	assert.Nil(t, rec.Rollback)
+	assert.Equal(t, Hold{Held: true, Level: 0}, rec.Hold)
+	got, _, _ := s.Read(t.Context())
+	assert.Equal(t, rec, got)
+}
+
+func TestRollback_SilentNodeFailsAndLeavesTheRollbackInPlace(t *testing.T) {
+	t.Parallel()
+	s := atLevelOne(t)
+	probe := func(context.Context) []NodeStatus {
+		return []NodeStatus{ackFor(s, "a", nil), {Node: "b", Reason: "no-reply"}}
+	}
+	_, err := RunRollback(t.Context(), 0, deps(s, probe, nil))
+	var unacked *UnackedError
+	require.ErrorAs(t, err, &unacked)
+	assert.Equal(t, []string{"b"}, unacked.Nodes)
+	assert.Contains(t, err.Error(), "b (no usable answer: no-reply)")
+	got, _, _ := s.Read(t.Context())
+	assert.Equal(t, 1, got.Level, "the level is not lowered")
+	require.NotNil(t, got.Rollback)
+	assert.Equal(t, 0, got.Rollback.Target)
+}
+
+func TestRollback_AStaleRevisionIsNotAnAck(t *testing.T) {
+	t.Parallel()
+	s := atLevelOne(t)
+	probe := func(context.Context) []NodeStatus {
+		st := ackFor(s, "a", nil)
+		st.Reply.Revision--
+		return []NodeStatus{st}
+	}
+	_, err := RunRollback(t.Context(), 0, deps(s, probe, nil))
+	var unacked *UnackedError
+	require.ErrorAs(t, err, &unacked)
+	assert.Contains(t, unacked.Reasons["a"], "has read revision")
+}
+
+func TestRollback_InFlightAboveTargetBlocksUntilItIsDone(t *testing.T) {
+	t.Parallel()
+	s := atLevelOne(t)
+	var calls atomic.Int32
+	probe := func(context.Context) []NodeStatus {
+		if calls.Add(1) < 3 {
+			return []NodeStatus{ackFor(s, "a", map[int]int{1: 2})}
+		}
+		return []NodeStatus{ackFor(s, "a", nil)}
+	}
+	d := deps(s, probe, nil)
+	d.AckTimeout = 2 * time.Second
+	rec, err := RunRollback(t.Context(), 0, d)
+	require.NoError(t, err)
+	assert.Zero(t, rec.Level)
+	assert.GreaterOrEqual(t, calls.Load(), int32(3), "it polled until the in-flight work was gone")
+
+	s2 := atLevelOne(t)
+	stuck := func(context.Context) []NodeStatus { return []NodeStatus{ackFor(s2, "a", map[int]int{1: 1})} }
+	_, err = RunRollback(t.Context(), 0, deps(s2, stuck, nil))
+	var unacked *UnackedError
+	require.ErrorAs(t, err, &unacked)
+	assert.Contains(t, unacked.Reasons["a"], "1 in flight above level 0")
+}
+
+func TestRollback_InFlightAtOrBelowTheTargetDoesNotBlock(t *testing.T) {
+	t.Parallel()
+	s := atLevelOne(t)
+	probe := func(context.Context) []NodeStatus { return []NodeStatus{ackFor(s, "a", map[int]int{0: 4})} }
+	_, err := RunRollback(t.Context(), 0, deps(s, probe, nil))
+	require.NoError(t, err)
+}
+
+func TestRollback_OutstandingDrainBlocksAndLeavesTheRollback(t *testing.T) {
+	t.Parallel()
+	s := atLevelOne(t)
+	probe := func(context.Context) []NodeStatus { return []NodeStatus{ackFor(s, "a", nil)} }
+	left := 3
+	drains := func(int) map[int]DrainCheck {
+		return map[int]DrainCheck{1: func(context.Context) (int, error) { return left, nil }}
+	}
+	_, err := RunRollback(t.Context(), 0, deps(s, probe, drains))
+	var out *DrainOutstandingError
+	require.ErrorAs(t, err, &out)
+	assert.Equal(t, 1, out.Level)
+	assert.Equal(t, 3, out.Outstanding)
+	got, _, _ := s.Read(t.Context())
+	assert.Equal(t, 1, got.Level)
+	assert.NotNil(t, got.Rollback)
+
+	left = 0
+	rec, err := RunRollback(t.Context(), 0, deps(s, probe, drains))
+	require.NoError(t, err, "the same command resumes once the work is gone")
+	assert.Zero(t, rec.Level)
+}
+
+func TestRollback_ADrainErrorAndAMissingCheckBothBlock(t *testing.T) {
+	t.Parallel()
+	s := atLevelOne(t)
+	probe := func(context.Context) []NodeStatus { return []NodeStatus{ackFor(s, "a", nil)} }
+	boom := errors.New("kv down")
+	_, err := RunRollback(t.Context(), 0, deps(s, probe, func(int) map[int]DrainCheck {
+		return map[int]DrainCheck{1: func(context.Context) (int, error) { return 0, boom }}
+	}))
+	require.ErrorIs(t, err, boom)
+
+	_, err = RunRollback(t.Context(), 0, deps(s, probe, func(int) map[int]DrainCheck { return nil }))
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "no drain check is registered")
+	got, _, _ := s.Read(t.Context())
+	assert.Equal(t, 1, got.Level)
+}
+
+func TestRollback_ResumesWithTheSameTargetAfterAFailure(t *testing.T) {
+	t.Parallel()
+	s := atLevelOne(t)
+	silent := func(context.Context) []NodeStatus { return []NodeStatus{{Node: "a", Reason: "no-reply"}} }
+	_, err := RunRollback(t.Context(), 0, deps(s, silent, nil))
+	require.Error(t, err)
+	healthy := func(context.Context) []NodeStatus { return []NodeStatus{ackFor(s, "a", nil)} }
+	rec, err := RunRollback(t.Context(), 0, deps(s, healthy, nil))
+	require.NoError(t, err)
+	assert.Zero(t, rec.Level)
+}
+
+func TestRollback_OldCLIRefusesBeforeWriting(t *testing.T) {
+	t.Parallel()
+	s := atLevelOne(t)
+	_, before, _ := s.Read(t.Context())
+	d := deps(s, func(context.Context) []NodeStatus { return nil }, nil)
+	d.Build = 0
+	_, err := RunRollback(t.Context(), 0, d)
+	require.ErrorIs(t, err, ErrCLITooOld)
+	got, after, _ := s.Read(t.Context())
+	assert.Equal(t, before, after)
+	assert.Nil(t, got.Rollback)
+	assert.False(t, got.Hold.Held)
+}
+
+func TestRollback_NoMembersCannotAcknowledge(t *testing.T) {
+	t.Parallel()
+	s := atLevelOne(t)
+	_, err := RunRollback(t.Context(), 0, deps(s, func(context.Context) []NodeStatus { return nil }, nil))
+	require.Error(t, err)
+	got, _, _ := s.Read(t.Context())
+	assert.NotNil(t, got.Rollback)
+}
+
+func TestServeStatus_NilGateAnswersWithAnErrorNeverZeroInFlight(t *testing.T) {
+	t.Parallel()
+	var g *Gate
+	_, err := g.ServeStatus("n1", Build)(t.Context(), &StatusRequest{Node: "n1"}, "")
+	require.Error(t, err)
+}
+
+func TestRollback_ACancelAndRestartAfterTheChecksFailsTheFinish(t *testing.T) {
+	t.Parallel()
+	s := atLevelOne(t)
+	probe := func(context.Context) []NodeStatus { return []NodeStatus{ackFor(s, "a", nil)} }
+	// The interruption lands after the drain check and before the finish.
+	drains := func(int) map[int]DrainCheck {
+		return map[int]DrainCheck{1: func(ctx context.Context) (int, error) {
+			if _, err := s.CancelRollback(ctx); err != nil {
+				return 0, err
+			}
+			_, _, err := s.StartRollback(ctx, 0, Build)
+			return 0, err
+		}}
+	}
+	_, err := RunRollback(t.Context(), 0, deps(s, probe, drains))
+	require.ErrorIs(t, err, ErrRollbackChanged)
+	got, _, _ := s.Read(t.Context())
+	assert.Equal(t, 1, got.Level, "the stale checks did not lower the level")
+	assert.NotNil(t, got.Rollback)
+}
+
+func TestFinishRollback_RefusesAWrongRevisionOrTarget(t *testing.T) {
+	t.Parallel()
+	s := atLevelOne(t)
+	_, rev, err := s.StartRollback(t.Context(), 0, Build)
+	require.NoError(t, err)
+	_, err = s.FinishRollback(t.Context(), rev-1, 0)
+	require.ErrorIs(t, err, ErrRollbackChanged)
+	_, err = s.FinishRollback(t.Context(), rev, 1)
+	require.ErrorIs(t, err, ErrRollbackChanged)
+	rec, err := s.FinishRollback(t.Context(), rev, 0)
+	require.NoError(t, err)
+	assert.Zero(t, rec.Level)
+}

spinifex/foundation/release/featurelevel/status.go

@@ -0,0 +1,148 @@
+package featurelevel
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"slices"
+	"strings"
+	"sync"
+	"time"
+
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+)
+
+const (
+	probeWait  = 3 * time.Second
+	probeGrace = 150 * time.Millisecond
+)
+
+// StatusSubject is one daemon's read-only status route, addressed per node so a
+// probe reaches exactly the node it asks about.
+func StatusSubject(node string) string {
+	return "cluster.v1.FeatureLevelStatus." + node
+}
+
+// ValidNode reports whether node is exactly one NATS subject token. Responder
+// and probe both check it, so neither can address a wildcard or another node.
+func ValidNode(node string) bool {
+	return node != "" && !strings.ContainsAny(node, ".*>\t\n\r ")
+}
+
+// StatusRequest asks node for its feature-level status.
+type StatusRequest struct {
+	Node string `json:"node"`
+}
+
+// StatusReply is one node's status. Revision is the record revision the node
+// read to answer, not one it holds from earlier. InFlight is the count of
+// admitted callers per level and InFlightTotal their sum.
+type StatusReply struct {
+	Node          string      `json:"node"`
+	Build         int         `json:"build"`
+	Revision      uint64      `json:"revision"`
+	InFlight      map[int]int `json:"in_flight"`
+	InFlightTotal int         `json:"in_flight_total"`
+}
+
+// ServeStatus answers the status route for node. It re-reads the record on every
+// request and refuses a request that names another node.
+func (g *Gate) ServeStatus(node string, build int) func(context.Context, *StatusRequest, string) (*StatusReply, error) {
+	return func(ctx context.Context, req *StatusRequest, _ string) (*StatusReply, error) {
+		if g == nil || g.store == nil {
+			return nil, fmt.Errorf("feature level gate on node %q is not ready", node)
+		}
+		if req.Node != node {
+			return nil, fmt.Errorf("feature level status for node %q answered by node %q", req.Node, node)
+		}
+		_, rev, err := g.store.Read(ctx)
+		if err != nil {
+			return nil, err
+		}
+		in := g.InFlight()
+		total := 0
+		for _, c := range in {
+			total += c
+		}
+		return &StatusReply{Node: node, Build: build, Revision: rev, InFlight: in, InFlightTotal: total}, nil
+	}
+}
+
+// NodeStatus is one configured node's probe result. Reply is nil, with Reason
+// set, when the node gave no usable answer.
+type NodeStatus struct {
+	Node   string
+	Reply  *StatusReply
+	Reason string
+}
+
+// Probe asks every configured node for its status, in parallel under one shared
+// wait, so N nodes cost one wait. It never caches.
+type Probe struct {
+	nc      *nats.Conn
+	nodeIDs []string
+	wait    time.Duration
+	grace   time.Duration
+}
+
+// NewProbe probes nodeIDs, the configured nodes.
+func NewProbe(nc *nats.Conn, nodeIDs []string) *Probe {
+	return &Probe{nc: nc, nodeIDs: slices.Sorted(slices.Values(nodeIDs)), wait: probeWait, grace: probeGrace}
+}
+
+// Probe returns one entry per configured node, sorted by node.
+func (p *Probe) Probe(ctx context.Context) []NodeStatus {
+	wait := p.wait
+	if deadline, ok := ctx.Deadline(); ok {
+		wait = min(wait, time.Until(deadline))
+	}
+	results := make([]NodeStatus, len(p.nodeIDs))
+	var wg sync.WaitGroup
+	for i, node := range p.nodeIDs {
+		wg.Go(func() {
+			if wait <= 0 || ctx.Err() != nil {
+				results[i] = NodeStatus{Node: node, Reason: "no-reply"}
+				return
+			}
+			results[i] = p.probeNode(ctx, node, wait)
+		})
+	}
+	wg.Wait()
+	return results
+}
+
+func (p *Probe) probeNode(ctx context.Context, node string, wait time.Duration) NodeStatus {
+	if !ValidNode(node) {
+		return NodeStatus{Node: node, Reason: "invalid-node-id"}
+	}
+	data, err := json.Marshal(StatusRequest{Node: node})
+	if err != nil {
+		return NodeStatus{Node: node, Reason: "malformed"}
+	}
+	frames, sum, err := natsmsg.Gather(ctx, p.nc, StatusSubject(node), data, natsmsg.GatherOpts{
+		Timeout: wait, ExpectedResponders: 1, Mode: natsmsg.CollectUntilDeadline, ResponderGrace: p.grace,
+	})
+	switch {
+	case err != nil:
+		return NodeStatus{Node: node, Reason: "transport"}
+	case sum.CapHit, len(sum.ConflictNodes) > 0, sum.DuplicateFrames > 0:
+		return NodeStatus{Node: node, Reason: "duplicate"}
+	case sum.Unidentified > 0:
+		return NodeStatus{Node: node, Reason: "wrong-identity"}
+	case len(sum.ErrorResponders) > 0:
+		return NodeStatus{Node: node, Reason: "error-reply"}
+	case sum.Received == 0, len(frames) == 0:
+		return NodeStatus{Node: node, Reason: "no-reply"}
+	case frames[0].NodeID != node:
+		return NodeStatus{Node: node, Reason: "wrong-identity"}
+	}
+	var reply StatusReply
+	if json.Unmarshal(frames[0].Data, &reply) != nil {
+		return NodeStatus{Node: node, Reason: "malformed"}
+	}
+	if reply.Node != node {
+		return NodeStatus{Node: node, Reason: "wrong-identity"}
+	}
+	return NodeStatus{Node: node, Reply: &reply}
+}

spinifex/foundation/release/featurelevel/status_test.go

@@ -0,0 +1,115 @@
+package featurelevel
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
+	"github.com/nats-io/nats.go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// serve answers node's status route with reply, naming headerNode in the reply
+// header the way a daemon names itself.
+func serve(t *testing.T, nc *nats.Conn, node, headerNode string, reply any) {
+	t.Helper()
+	sub, err := nc.Subscribe(StatusSubject(node), func(m *nats.Msg) {
+		out := nats.NewMsg("")
+		out.Data, _ = json.Marshal(reply)
+		out.Header.Set(natsmsg.NodeIDHeader, headerNode)
+		_ = m.RespondMsg(out)
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+}
+
+func shortProbe(nc *nats.Conn, nodes ...string) *Probe {
+	p := NewProbe(nc, nodes)
+	p.wait, p.grace = 400*time.Millisecond, 20*time.Millisecond
+	return p
+}
+
+func TestServeStatus_RefusesARequestNamingAnotherNode(t *testing.T) {
+	t.Parallel()
+	s, _ := newStore(t)
+	h := NewGate(s).ServeStatus("n1", Build)
+	_, err := h(t.Context(), &StatusRequest{Node: "n2"}, "")
+	require.Error(t, err)
+}
+
+func TestServeStatus_RereadsTheRecordAndReportsInFlightPerLevel(t *testing.T) {
+	t.Parallel()
+	s, _ := newStore(t)
+	g := NewGate(s)
+	h := g.ServeStatus("n1", Build)
+
+	first, err := h(t.Context(), &StatusRequest{Node: "n1"}, "")
+	require.NoError(t, err)
+	assert.Equal(t, "n1", first.Node)
+	assert.Equal(t, Build, first.Build)
+	assert.Zero(t, first.Revision, "no record yet")
+
+	_, _, err = s.Advance(t.Context(), members(1, true, true))
+	require.NoError(t, err)
+	done, _, err := g.Admit(t.Context(), 1)
+	require.NoError(t, err)
+	second, err := h(t.Context(), &StatusRequest{Node: "n1"}, "")
+	require.NoError(t, err)
+	assert.Greater(t, second.Revision, first.Revision, "the reply reads the record afresh")
+	assert.Equal(t, map[int]int{1: 1}, second.InFlight)
+	assert.Equal(t, 1, second.InFlightTotal)
+	done()
+	third, err := h(t.Context(), &StatusRequest{Node: "n1"}, "")
+	require.NoError(t, err)
+	assert.Zero(t, third.InFlightTotal)
+}
+
+func TestProbe_ReportsEachNodeAndTheReasonForASilentOne(t *testing.T) {
+	t.Parallel()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	serve(t, nc, "n1", "n1", StatusReply{Node: "n1", Build: Build, Revision: 7})
+
+	got := shortProbe(nc, "n1", "silent").Probe(t.Context())
+	require.Len(t, got, 2)
+	assert.Equal(t, "n1", got[0].Node)
+	require.NotNil(t, got[0].Reply)
+	assert.EqualValues(t, 7, got[0].Reply.Revision)
+	assert.Equal(t, "silent", got[1].Node)
+	assert.Nil(t, got[1].Reply)
+	assert.Equal(t, "no-reply", got[1].Reason)
+}
+
+func TestProbe_RejectsAReplyNamingAnotherNode(t *testing.T) {
+	t.Parallel()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	serve(t, nc, "body", "body", StatusReply{Node: "other", Build: Build})
+	serve(t, nc, "header", "other", StatusReply{Node: "header", Build: Build})
+
+	got := shortProbe(nc, "body", "header").Probe(t.Context())
+	assert.Nil(t, got[0].Reply, "a body naming another node is rejected")
+	assert.Equal(t, "wrong-identity", got[0].Reason)
+	assert.Nil(t, got[1].Reply, "a header naming another node is rejected")
+	assert.Equal(t, "wrong-identity", got[1].Reason)
+}
+
+func TestProbe_TwoRespondersForOneNodeIsADuplicate(t *testing.T) {
+	t.Parallel()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	serve(t, nc, "n1", "n1", StatusReply{Node: "n1", Build: Build})
+	serve(t, nc, "n1", "n1", StatusReply{Node: "n1", Build: Build})
+	got := shortProbe(nc, "n1").Probe(t.Context())
+	assert.Nil(t, got[0].Reply)
+	assert.Equal(t, "duplicate", got[0].Reason)
+}
+
+func TestProbe_MalformedReplyAndInvalidNodeID(t *testing.T) {
+	t.Parallel()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	serve(t, nc, "n1", "n1", "not an object")
+	got := shortProbe(nc, "n1", "a.b").Probe(t.Context())
+	assert.Equal(t, "invalid-node-id", got[0].Reason, "sorted: a.b first")
+	assert.Equal(t, "malformed", got[1].Reason)
+}

spinifex/foundation/release/featurelevel/store.go

@@ -0,0 +1,244 @@
+package featurelevel
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"math/rand/v2"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Key is where the cluster record lives in the cluster-config bucket, which has
+// no TTL.
+const Key = "feature-level"
+
+const casAttempts = 16
+
+// Hold caps advance. Held distinguishes a hold at level 0 from no hold.
+type Hold struct {
+	Held  bool `json:"held"`
+	Level int  `json:"level"`
+}
+
+// Rollback is a rollback to Target that has not lowered the level yet.
+type Rollback struct {
+	Target int `json:"target"`
+	// PriorHold is the hold in force before the rollback, which a cancel restores.
+	PriorHold Hold `json:"prior_hold"`
+}
+
+// Record is the one cluster record. A missing key reads as the zero Record.
+type Record struct {
+	Level    int       `json:"level"`
+	Hold     Hold      `json:"hold"`
+	Rollback *Rollback `json:"rollback,omitempty"`
+}
+
+// Member is one configured node's reported level. Current is false for a node
+// with no heartbeat or a stale one, which blocks advance.
+type Member struct {
+	Node    string
+	Level   int
+	Current bool
+}
+
+// KV is the part of a JetStream key-value bucket the store uses.
+type KV interface {
+	Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error)
+	Create(ctx context.Context, key string, value []byte, opts ...jetstream.KVCreateOpt) (uint64, error)
+	Update(ctx context.Context, key string, value []byte, last uint64) (uint64, error)
+}
+
+// Store reads and compare-and-sets the cluster record.
+type Store struct{ kv KV }
+
+// NewStore returns a store over kv.
+func NewStore(kv KV) *Store { return &Store{kv: kv} }
+
+// Read returns the record and the revision it was read at (0 when absent).
+func (s *Store) Read(ctx context.Context) (Record, uint64, error) {
+	entry, err := s.kv.Get(ctx, Key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return Record{}, 0, nil
+	}
+	if err != nil {
+		return Record{}, 0, fmt.Errorf("read feature level: %w", err)
+	}
+	var rec Record
+	if err := json.Unmarshal(entry.Value(), &rec); err != nil {
+		return Record{}, 0, fmt.Errorf("decode feature level: %w", err)
+	}
+	return rec, entry.Revision(), nil
+}
+
+// mutate applies f to the record under a compare-and-set loop. f returns false
+// to leave the record as read. It returns the record and the revision it holds.
+func (s *Store) mutate(ctx context.Context, f func(*Record) (bool, error)) (Record, uint64, error) {
+	var lastErr error
+	for i := range casAttempts {
+		rec, rev, err := s.Read(ctx)
+		if err != nil {
+			return Record{}, 0, err
+		}
+		changed, err := f(&rec)
+		if err != nil {
+			return Record{}, 0, err
+		}
+		if !changed {
+			return rec, rev, nil
+		}
+		data, err := json.Marshal(rec)
+		if err != nil {
+			return Record{}, 0, fmt.Errorf("encode feature level: %w", err)
+		}
+		if rev == 0 {
+			rev, err = s.kv.Create(ctx, Key, data)
+		} else {
+			rev, err = s.kv.Update(ctx, Key, data, rev)
+		}
+		if err == nil {
+			return rec, rev, nil
+		}
+		if !kvstore.IsRevisionConflict(err) {
+			return Record{}, 0, fmt.Errorf("write feature level: %w", err)
+		}
+		lastErr = err
+		backoff := time.Duration(i+1) * 2 * time.Millisecond
+		select {
+		case <-time.After(backoff/2 + rand.N(backoff)): //nolint:gosec // jitter
+		case <-ctx.Done():
+			return Record{}, 0, ctx.Err()
+		}
+	}
+	return Record{}, 0, fmt.Errorf("feature level contended, CAS exhausted: %w", lastErr)
+}
+
+// candidate is the highest level every member supports, capped at Build, or -1
+// when any member is not current or none are known.
+func candidate(members []Member) int {
+	if len(members) == 0 {
+		return -1
+	}
+	level := Build
+	for _, m := range members {
+		if !m.Current {
+			return -1
+		}
+		level = min(level, m.Level)
+	}
+	return level
+}
+
+// Advance raises the level to the highest one every member reports, respecting a
+// hold and never lowering it. It writes only when that is above the recorded
+// level, so any number of daemons may call it. A stale member, an unknown member
+// set or a rollback in progress leaves the record alone.
+func (s *Store) Advance(ctx context.Context, members []Member) (Record, bool, error) {
+	want := candidate(members)
+	advanced := false
+	rec, _, err := s.mutate(ctx, func(r *Record) (bool, error) {
+		advanced = false
+		if r.Rollback != nil {
+			return false, nil
+		}
+		to := want
+		if r.Hold.Held {
+			to = min(to, r.Hold.Level)
+		}
+		if to <= r.Level {
+			return false, nil
+		}
+		r.Level = to
+		advanced = true
+		return true, nil
+	})
+	return rec, advanced, err
+}
+
+// StartRollback records the rollback to target and a hold at target in one
+// write, and returns the revision to wait for. cliBuild is the caller's own
+// Build: a recorded level above it refuses before anything is written. Repeating
+// the same rollback resumes it.
+func (s *Store) StartRollback(ctx context.Context, target, cliBuild int) (Record, uint64, error) {
+	return s.mutate(ctx, func(r *Record) (bool, error) {
+		if r.Level > cliBuild {
+			return false, fmt.Errorf("%w: recorded %d, this build %d", ErrCLITooOld, r.Level, cliBuild)
+		}
+		if r.Rollback != nil {
+			if r.Rollback.Target != target {
+				return false, fmt.Errorf("%w: target %d", ErrRollbackInProgress, r.Rollback.Target)
+			}
+			return false, nil
+		}
+		if target < 0 || target >= r.Level {
+			return false, fmt.Errorf("%w: target %d, level %d", ErrBadTarget, target, r.Level)
+		}
+		r.Rollback = &Rollback{Target: target, PriorHold: r.Hold}
+		r.Hold = Hold{Held: true, Level: target}
+		return true, nil
+	})
+}
+
+// FinishRollback lowers the level to the target and clears the rollback, keeping
+// the hold so running newer builds cannot advance it again. It succeeds only when
+// the record still holds the rollback write at rev for target, so a rollback that
+// was cancelled or restarted since its checks ran is not completed on their evidence.
+func (s *Store) FinishRollback(ctx context.Context, rev uint64, target int) (Record, error) {
+	rec, cur, err := s.Read(ctx)
+	if err != nil {
+		return Record{}, err
+	}
+	if rec.Rollback == nil {
+		return Record{}, ErrNoRollback
+	}
+	if cur != rev || rec.Rollback.Target != target {
+		return Record{}, fmt.Errorf("%w: recorded revision %d target %d, verified revision %d target %d", ErrRollbackChanged, cur, rec.Rollback.Target, rev, target)
+	}
+	rec.Level = target
+	rec.Rollback = nil
+	data, err := json.Marshal(rec)
+	if err != nil {
+		return Record{}, fmt.Errorf("encode feature level: %w", err)
+	}
+	if _, err := s.kv.Update(ctx, Key, data, rev); err != nil {
+		if kvstore.IsRevisionConflict(err) {
+			return Record{}, fmt.Errorf("%w: the record changed while finishing", ErrRollbackChanged)
+		}
+		return Record{}, fmt.Errorf("write feature level: %w", err)
+	}
+	return rec, nil
+}
+
+// CancelRollback clears a rollback that has not lowered the level and restores
+// the hold that existed before it started.
+func (s *Store) CancelRollback(ctx context.Context) (Record, error) {
+	rec, _, err := s.mutate(ctx, func(r *Record) (bool, error) {
+		if r.Rollback == nil {
+			return false, ErrNoRollback
+		}
+		r.Hold = r.Rollback.PriorHold
+		r.Rollback = nil
+		return true, nil
+	})
+	return rec, err
+}
+
+// ReleaseHold clears the hold once the operator has finished or abandoned the
+// downgrade. It refuses while a rollback is in progress.
+func (s *Store) ReleaseHold(ctx context.Context) (Record, error) {
+	rec, _, err := s.mutate(ctx, func(r *Record) (bool, error) {
+		if r.Rollback != nil {
+			return false, ErrRollbackInProgress
+		}
+		if !r.Hold.Held {
+			return false, nil
+		}
+		r.Hold = Hold{}
+		return true, nil
+	})
+	return rec, err
+}

spinifex/operator/cli/admin_cluster_level.go

@@ -0,0 +1,125 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"maps"
+	"os"
+	"slices"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/release/featurelevel"
+	"github.com/mulgadc/spinifex/spinifex/runtime/daemon"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/spf13/cobra"
+)
+
+var clusterStatusCmd = &cobra.Command{
+	Use:   "status",
+	Short: "Show the cluster feature level and each member's state",
+	Long: `status prints the recorded cluster feature level, any hold and any rollback in
+progress, then each configured member's reported level, heartbeat age, and its
+answer to the status request (or the reason it gave none). It only reads.`,
+	Run: runClusterStatus,
+}
+
+func init() {
+	clusterCmd.AddCommand(clusterStatusCmd)
+}
+
+// memberLevelReport is what status shows for one configured member.
+type memberLevelReport struct {
+	Node          string
+	HasHeartbeat  bool
+	HeartbeatNote string // why there is no usable heartbeat
+	BeatLevel     int
+	BeatAge       time.Duration
+	Status        featurelevel.NodeStatus
+}
+
+func runClusterStatus(_ *cobra.Command, _ []string) {
+	cfg, nc, err := loadConfigAndConnect()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+		os.Exit(1)
+	}
+	defer nc.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	jsm, err := daemon.NewJetStreamManager(nc)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+		os.Exit(1)
+	}
+	store, err := jsm.AttachFeatureLevel(ctx)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+		os.Exit(1)
+	}
+	rec, rev, err := store.Read(ctx)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+		os.Exit(1)
+	}
+
+	nodes := slices.Sorted(maps.Keys(cfg.Nodes))
+	statuses := featurelevel.NewProbe(nc, nodes).Probe(ctx)
+	now := time.Now()
+	members := make([]memberLevelReport, len(nodes))
+	for i, node := range nodes {
+		members[i] = memberLevelReport{Node: node, Status: statuses[i]}
+		h, herr := jsm.ReadHeartbeat(node)
+		switch {
+		case errors.Is(herr, jetstream.ErrKeyNotFound):
+			members[i].HeartbeatNote = "no heartbeat"
+		case herr != nil:
+			members[i].HeartbeatNote = "heartbeat unreadable: " + herr.Error()
+		default:
+			written, perr := time.Parse(time.RFC3339, h.Timestamp)
+			if perr != nil {
+				members[i].HeartbeatNote = "heartbeat timestamp unreadable"
+				break
+			}
+			members[i].HasHeartbeat = true
+			members[i].BeatLevel = h.FeatureLevel
+			members[i].BeatAge = now.Sub(written)
+		}
+	}
+	renderLevelStatus(os.Stdout, rec, rev, featurelevel.Build, members)
+}
+
+// renderLevelStatus prints the record, then one line per member.
+func renderLevelStatus(w io.Writer, rec featurelevel.Record, rev uint64, build int, members []memberLevelReport) {
+	fmt.Fprintf(w, "Feature level: %d (this build supports %d, record revision %d)\n", rec.Level, build, rev)
+	if rec.Hold.Held {
+		fmt.Fprintf(w, "Hold:          held at %d (release with: spx admin cluster release-hold)\n", rec.Hold.Level)
+	} else {
+		fmt.Fprintln(w, "Hold:          none")
+	}
+	if rec.Rollback != nil {
+		fmt.Fprintf(w, "Rollback:      to %d in progress\n", rec.Rollback.Target)
+	} else {
+		fmt.Fprintln(w, "Rollback:      none")
+	}
+	fmt.Fprintln(w, "Members:")
+	for _, m := range members {
+		fmt.Fprintf(w, "  %s: %s; %s\n", m.Node, heartbeatText(m), statusText(m.Status))
+	}
+}
+
+func heartbeatText(m memberLevelReport) string {
+	if !m.HasHeartbeat {
+		return m.HeartbeatNote
+	}
+	return fmt.Sprintf("heartbeat level %d, age %s", m.BeatLevel, m.BeatAge.Round(time.Second))
+}
+
+func statusText(s featurelevel.NodeStatus) string {
+	if s.Reply == nil {
+		return "status: none (" + s.Reason + ")"
+	}
+	return fmt.Sprintf("status: build %d, revision %d, in flight %d", s.Reply.Build, s.Reply.Revision, s.Reply.InFlightTotal)
+}

spinifex/operator/cli/admin_cluster_level_test.go

@@ -0,0 +1,55 @@
+package cli
+
+import (
+	"bytes"
+	"testing"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/release/featurelevel"
+	"github.com/stretchr/testify/assert"
+)
+
+func answered(node string, build int, rev uint64, inFlight int) featurelevel.NodeStatus {
+	return featurelevel.NodeStatus{Node: node, Reply: &featurelevel.StatusReply{Node: node, Build: build, Revision: rev, InFlightTotal: inFlight}}
+}
+
+func TestRenderLevelStatus_HeldAndRollbackStatesAndASilentMember(t *testing.T) {
+	t.Parallel()
+	members := []memberLevelReport{
+		{Node: "n1", HasHeartbeat: true, BeatLevel: 1, BeatAge: 4 * time.Second, Status: answered("n1", 1, 9, 2)},
+		{Node: "n2", HeartbeatNote: "no heartbeat", Status: featurelevel.NodeStatus{Node: "n2", Reason: "no-reply"}},
+	}
+
+	var plain bytes.Buffer
+	renderLevelStatus(&plain, featurelevel.Record{Level: 1}, 9, 1, members)
+	out := plain.String()
+	assert.Contains(t, out, "Feature level: 1 (this build supports 1, record revision 9)")
+	assert.Contains(t, out, "Hold:          none")
+	assert.Contains(t, out, "Rollback:      none")
+	assert.Contains(t, out, "n1: heartbeat level 1, age 4s; status: build 1, revision 9, in flight 2")
+	assert.Contains(t, out, "n2: no heartbeat; status: none (no-reply)")
+
+	var held bytes.Buffer
+	renderLevelStatus(&held, featurelevel.Record{Level: 0, Hold: featurelevel.Hold{Held: true, Level: 0}}, 10, 1, members)
+	assert.Contains(t, held.String(), "Hold:          held at 0", "a hold at level 0 is shown as a hold")
+	assert.Contains(t, held.String(), "spx admin cluster release-hold")
+
+	var rolling bytes.Buffer
+	renderLevelStatus(&rolling, featurelevel.Record{
+		Level: 2, Hold: featurelevel.Hold{Held: true, Level: 1}, Rollback: &featurelevel.Rollback{Target: 1},
+	}, 11, 2, members)
+	assert.Contains(t, rolling.String(), "Rollback:      to 1 in progress")
+	assert.Contains(t, rolling.String(), "Hold:          held at 1")
+}
+
+func TestDrainChecksCoverEveryDeclaredLevel(t *testing.T) {
+	t.Parallel()
+	assert.NoError(t, registerDrainChecks(), "a level declared without a registered drain check")
+}
+
+func TestCancelError_ExplainsALoweredLevel(t *testing.T) {
+	t.Parallel()
+	assert.ErrorContains(t, cancelError(featurelevel.ErrNoRollback), "already finished")
+	other := assert.AnError
+	assert.Equal(t, other, cancelError(other))
+}

spinifex/operator/cli/admin_cluster_rollback.go

@@ -0,0 +1,138 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"maps"
+	"os"
+	"slices"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/release/featurelevel"
+	"github.com/mulgadc/spinifex/spinifex/runtime/daemon"
+	"github.com/spf13/cobra"
+)
+
+var (
+	rollbackTo      int
+	rollbackCancel  bool
+	rollbackTimeout time.Duration
+)
+
+var clusterRollbackCmd = &cobra.Command{
+	Use:   "rollback",
+	Short: "Lower the cluster feature level so a previous build can be installed",
+	Long: `rollback --to N records the rollback, waits until every configured member has
+read it and has no work admitted above level N, runs the drain check of every
+level being given up, then lowers the level. The hold stays until release-hold.
+A failure leaves the rollback recorded: rerun the same command to resume, or
+use --cancel to abandon it (refused once the level has been lowered).`,
+	Run: runClusterRollback,
+}
+
+var clusterReleaseHoldCmd = &cobra.Command{
+	Use:   "release-hold",
+	Short: "Allow the cluster feature level to advance again after a rollback",
+	Run:   runClusterReleaseHold,
+}
+
+func init() {
+	clusterCmd.AddCommand(clusterRollbackCmd, clusterReleaseHoldCmd)
+	clusterRollbackCmd.Flags().IntVar(&rollbackTo, "to", -1, "feature level to roll back to")
+	clusterRollbackCmd.Flags().BoolVar(&rollbackCancel, "cancel", false, "abandon a rollback that has not yet lowered the level")
+	clusterRollbackCmd.Flags().DurationVar(&rollbackTimeout, "timeout", featurelevel.DefaultAckTimeout, "how long to wait for every member to acknowledge")
+}
+
+// registerDrainChecks is the composition root for drain checks: every declared
+// level above the baseline registers its check here, and a level left without
+// one stops a rollback from running.
+func registerDrainChecks() error {
+	if missing := featurelevel.UncoveredLevels(); len(missing) > 0 {
+		return fmt.Errorf("feature levels %v have no drain check registered; this build cannot roll back safely", missing)
+	}
+	return nil
+}
+
+func fail(err error) {
+	fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+	os.Exit(1)
+}
+
+func runClusterRollback(cmd *cobra.Command, _ []string) {
+	if !rollbackCancel && rollbackTo < 0 {
+		fail(errors.New("--to is required (or --cancel)"))
+	}
+	if err := registerDrainChecks(); err != nil {
+		fail(err)
+	}
+	cfg, nc, err := loadConfigAndConnect()
+	if err != nil {
+		fail(err)
+	}
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), rollbackTimeout+5*time.Minute)
+	defer cancel()
+	jsm, err := daemon.NewJetStreamManager(nc)
+	if err != nil {
+		fail(err)
+	}
+	store, err := jsm.AttachFeatureLevel(ctx)
+	if err != nil {
+		fail(err)
+	}
+
+	if rollbackCancel {
+		rec, cerr := store.CancelRollback(ctx)
+		if cerr != nil {
+			fail(cancelError(cerr))
+		}
+		fmt.Printf("Rollback cancelled; feature level stays %d.\n", rec.Level)
+		return
+	}
+
+	nodes := slices.Sorted(maps.Keys(cfg.Nodes))
+	probe := featurelevel.NewProbe(nc, nodes)
+	rec, err := featurelevel.RunRollback(ctx, rollbackTo, featurelevel.RollbackDeps{
+		Store:      store,
+		Probe:      probe.Probe,
+		Drains:     featurelevel.DrainChecks,
+		Build:      featurelevel.Build,
+		AckTimeout: rollbackTimeout,
+	})
+	if err != nil {
+		fail(err)
+	}
+	fmt.Printf("Feature level lowered to %d. The previous build may now be installed.\n", rec.Level)
+	fmt.Printf("The hold at %d stays until you run: spx admin cluster release-hold\n", rec.Hold.Level)
+}
+
+func cancelError(err error) error {
+	if errors.Is(err, featurelevel.ErrNoRollback) {
+		return errors.New("no rollback is in progress; if it already finished the level has been lowered and cannot be cancelled")
+	}
+	return err
+}
+
+func runClusterReleaseHold(_ *cobra.Command, _ []string) {
+	_, nc, err := loadConfigAndConnect()
+	if err != nil {
+		fail(err)
+	}
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	jsm, err := daemon.NewJetStreamManager(nc)
+	if err != nil {
+		fail(err)
+	}
+	store, err := jsm.AttachFeatureLevel(ctx)
+	if err != nil {
+		fail(err)
+	}
+	rec, err := store.ReleaseHold(ctx)
+	if err != nil {
+		fail(err)
+	}
+	fmt.Printf("Hold released; the feature level (%d) may advance once every member is current.\n", rec.Level)
+}

spinifex/runtime/daemon/daemon.go

@@ -83,6 +83,7 @@ import (
 	route53domain "github.com/mulgadc/spinifex/spinifex/domains/route53"
 	"github.com/mulgadc/spinifex/spinifex/domains/sqs/workerlaunch"
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
+	"github.com/mulgadc/spinifex/spinifex/foundation/release/featurelevel"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/clustersize"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
@@ -181,6 +182,7 @@ type Daemon struct {
 	insight               *nodeInsight
 	eipService            ec2eip.EIPService
 	elbv2Service          *elbv2domain.ELBv2ServiceImpl
+	featureGate           *featurelevel.Gate // admission for N-only work; set by checkFeatureLevel
 	eksService            *ekscluster.EKSServiceImpl
 	ecsService            *ecsdomain.Service
 	ecsScheduler          *ecsdomain.Scheduler
@@ -1391,10 +1393,22 @@ func (d *Daemon) subscribeAll() error {
 	} else {
 		slog.Error("Not serving account-settings seal capability: node ID is not a single NATS subject token", "node", d.node)
 	}
+	subs = append(subs, d.featureLevelStatusSubs()...)
 
 	return d.registerNatsSubs(subs)
 }
 
+// featureLevelStatusSubs serves the status route only once the gate exists, so a
+// node that has not read the record never answers as having nothing in flight.
+// Per-node, no queue group, like the seal responder.
+func (d *Daemon) featureLevelStatusSubs() []natsSub {
+	if d.featureGate == nil || !featurelevel.ValidNode(d.node) {
+		slog.Error("Not serving feature level status", "node", d.node, "gate_ready", d.featureGate != nil)
+		return nil
+	}
+	return []natsSub{{featurelevel.StatusSubject(d.node), handleNATSRequest(d.node, d.featureGate.ServeStatus(d.node, featurelevel.Build)), ""}}
+}
+
 // registerNatsSubs turns each entry in subs into a live NATS subscription and
 // records it in d.natsSubscriptions. Shared by subscribeAll's boot-time table
 // and any subject registered later, off the main boot goroutine (e.g. Ochre's
@@ -2014,6 +2028,17 @@ func (d *Daemon) startCluster() error {
 		return fmt.Errorf("initialize JetStream: %w", err)
 	}
 
+	// Before any cluster state is claimed: a build below the recorded feature
+	// level would write what the newer builds no longer read. A refusal exits so
+	// the unit shows failed rather than a healthy idle daemon.
+	if err := d.checkFeatureLevel(); err != nil {
+		if errors.Is(err, errStartupPermanent) {
+			slog.Error("Refusing to start: cluster feature level", "err", err)
+			d.exitFunc(1)
+		}
+		return fmt.Errorf("feature level: %w", err)
+	}
+
 	// Before any service is built: a node on a different provider from the
 	// cluster's would address existing volumes through the wrong implementation.
 	// A refusal exits so the unit shows failed rather than a healthy idle daemon.

spinifex/runtime/daemon/feature_level.go

@@ -0,0 +1,222 @@
+package daemon
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"maps"
+	"slices"
+	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/release/featurelevel"
+	"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
+	instancecache "github.com/mulgadc/spinifex/spinifex/runtime/compute/cache"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// FeatureLevelStore returns the cluster feature-level record over the
+// cluster-config bucket, which has no TTL. Reads go to the stream leader.
+func (m *JetStreamManager) FeatureLevelStore(ctx context.Context) (*featurelevel.Store, error) {
+	if m.clusterConfig == nil {
+		return nil, errors.New("cluster config KV not initialized")
+	}
+	kv, err := m.clusterConfig.KV(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return featurelevel.NewStore(featurelevel.LeaderKV(m.js, kv)), nil
+}
+
+// AttachFeatureLevel opens the buckets an operator command reads that a daemon
+// already created, without creating, repairing or migrating either. The
+// cluster-state bucket is optional: a missing one is reported as unavailable
+// heartbeats rather than created.
+func (m *JetStreamManager) AttachFeatureLevel(ctx context.Context) (*featurelevel.Store, error) {
+	kv, err := m.js.KeyValue(ctx, ClusterConfigBucket)
+	if err != nil {
+		return nil, fmt.Errorf("cluster config bucket %s is unavailable: %w", ClusterConfigBucket, err)
+	}
+	if state, serr := m.js.KeyValue(ctx, ClusterStateBucket); serr == nil {
+		m.clusterKV = state
+	}
+	return featurelevel.NewStore(featurelevel.LeaderKV(m.js, kv)), nil
+}
+
+// featureLevelMembers reads every configured node's heartbeat. A member is
+// current when its heartbeat is within NodeStaleAfter. A node with no heartbeat
+// (never written, or expired from the bucket) or an unreadable timestamp is not
+// current; any other read failure is returned so the tick advances nothing.
+func featureLevelMembers(ctx context.Context, read func(ctx context.Context, node string) (*Heartbeat, error), nodes []string, now time.Time) ([]featurelevel.Member, error) {
+	members := make([]featurelevel.Member, 0, len(nodes))
+	for _, node := range nodes {
+		if err := ctx.Err(); err != nil {
+			return nil, fmt.Errorf("read heartbeat of %s: %w", node, err)
+		}
+		h, err := read(ctx, node)
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			members = append(members, featurelevel.Member{Node: node})
+			continue
+		}
+		if err != nil {
+			return nil, fmt.Errorf("read heartbeat of %s: %w", node, err)
+		}
+		written, perr := time.Parse(time.RFC3339, h.Timestamp)
+		current := perr == nil && now.UTC().Sub(written.UTC()) <= instancecache.NodeStaleAfter
+		members = append(members, featurelevel.Member{Node: node, Level: h.FeatureLevel, Current: current})
+	}
+	return members, nil
+}
+
+// advanceFeatureLevel raises the recorded level when every configured node has
+// a current heartbeat reporting it. Any number of daemons may call it.
+func advanceFeatureLevel(ctx context.Context, store *featurelevel.Store, read func(context.Context, string) (*Heartbeat, error), nodes []string, now time.Time) (featurelevel.Record, bool, error) {
+	members, err := featureLevelMembers(ctx, read, nodes, now)
+	if err != nil {
+		return featurelevel.Record{}, false, err
+	}
+	return store.Advance(ctx, members)
+}
+
+// tickFeatureLevel runs on each heartbeat tick. A failure is logged and never
+// fatal: the level stays where it is until a later tick succeeds.
+func (d *Daemon) tickFeatureLevel() {
+	if d.jsManager == nil || d.clusterConfig == nil {
+		return
+	}
+	// Half the interval, so a slow tick never delays the next heartbeat.
+	ctx, cancel := context.WithTimeout(d.ctx, heartbeatInterval/2)
+	defer cancel()
+	store, err := d.jsManager.FeatureLevelStore(ctx)
+	if err != nil {
+		slog.Warn("Feature level: store unavailable", "error", err)
+		return
+	}
+	// A level raised above this build while it runs (a downgrade without a
+	// rollback) ends the daemon within one tick.
+	if err := refuseAboveBuild(ctx, store); err != nil {
+		if errors.Is(err, errStartupPermanent) {
+			slog.Error("Exiting: cluster feature level is above this build", "err", err)
+			d.exitFunc(1)
+			return
+		}
+		slog.Warn("Feature level: fence read failed", "error", err)
+		return
+	}
+	nodes := slices.Sorted(maps.Keys(d.clusterConfig.Nodes))
+	rec, advanced, err := advanceFeatureLevel(ctx, store, d.jsManager.ReadHeartbeatLeader, nodes, time.Now())
+	if err != nil {
+		slog.Warn("Feature level: advance failed", "error", err)
+		return
+	}
+	if advanced {
+		slog.Info("Cluster feature level advanced", "level", rec.Level, "build", featurelevel.Build)
+	}
+}
+
+// FeatureLevelRefusedError is returned when the cluster records a feature level
+// above what this build supports, so an older build cannot join and write what
+// the newer one no longer reads.
+type FeatureLevelRefusedError struct {
+	Build    int
+	Recorded int
+}
+
+func (e *FeatureLevelRefusedError) Error() string {
+	return fmt.Sprintf("cluster feature level is %d but this build supports %d; "+
+		"lower the cluster first with `spx admin cluster rollback --to %d`, then install this build",
+		e.Recorded, e.Build, e.Build)
+}
+
+// Unwrap makes the refusal permanent: no retry clears it.
+func (e *FeatureLevelRefusedError) Unwrap() error { return errStartupPermanent }
+
+// featureLevelReadTimeout bounds one attempt at reading the record.
+const featureLevelReadTimeout = 10 * time.Second
+
+// featureLevelSleep is the backoff seam; tests replace it.
+var featureLevelSleep = func(ctx context.Context, d time.Duration) error {
+	t := time.NewTimer(d)
+	defer t.Stop()
+	select {
+	case <-t.C:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+// retryFeatureLevel runs fn with the startup backoff (500ms to 10s) until it
+// succeeds, fails permanently, or ctx is cancelled. A cluster outage of any
+// length leaves the daemon retrying rather than abandoning bootstrap.
+func retryFeatureLevel(ctx context.Context, fn func() error) error {
+	delay := 500 * time.Millisecond
+	start := time.Now()
+	for attempt := 1; ; attempt++ {
+		err := fn()
+		if err == nil {
+			if attempt > 1 {
+				slog.Info("feature level initialized successfully", "attempts", attempt, "elapsed_ms", otelsetup.Millis(time.Since(start)))
+			}
+			return nil
+		}
+		if errors.Is(err, errStartupPermanent) {
+			return err
+		}
+		if ctx.Err() != nil {
+			return fmt.Errorf("feature level check cancelled after %d attempts: %w", attempt, err)
+		}
+		slog.Warn("Failed to init feature level", "error", err, "attempt", attempt, "elapsed_ms", otelsetup.Millis(time.Since(start)))
+		if serr := featureLevelSleep(ctx, delay); serr != nil {
+			return fmt.Errorf("feature level check cancelled after %d attempts: %w", attempt, err)
+		}
+		delay = min(delay*2, 10*time.Second)
+	}
+}
+
+// refuseAboveBuild reads the record from the leader and refuses when it is above
+// this build.
+func refuseAboveBuild(ctx context.Context, store *featurelevel.Store) error {
+	rctx, cancel := context.WithTimeout(ctx, featureLevelReadTimeout)
+	defer cancel()
+	rec, _, err := store.Read(rctx)
+	if err != nil {
+		return err
+	}
+	if rec.Level > featurelevel.Build {
+		return &FeatureLevelRefusedError{Build: featurelevel.Build, Recorded: rec.Level}
+	}
+	return nil
+}
+
+// enforceFeatureLevel refuses when the recorded level is above this build. An
+// absent record is level 0. A read failure retries until ctx is cancelled and
+// neither starts nor refuses.
+//
+// Once the level passes, publish overwrites any predecessor's heartbeat with
+// this build's own, so its capability cannot count for the build now running.
+// The level is read again afterwards: one that rose in between refuses.
+func enforceFeatureLevel(ctx context.Context, store *featurelevel.Store, publish func() error) error {
+	return retryFeatureLevel(ctx, func() error {
+		if err := refuseAboveBuild(ctx, store); err != nil {
+			return err
+		}
+		if err := publish(); err != nil {
+			return fmt.Errorf("publish heartbeat: %w", err)
+		}
+		return refuseAboveBuild(ctx, store)
+	})
+}
+
+// checkFeatureLevel runs the startup refusal against the cluster record.
+func (d *Daemon) checkFeatureLevel() error {
+	store, err := d.jsManager.FeatureLevelStore(d.ctx)
+	if err != nil {
+		return err
+	}
+	if err := enforceFeatureLevel(d.ctx, store, func() error { return d.jsManager.WriteHeartbeat(d.buildHeartbeat()) }); err != nil {
+		return err
+	}
+	d.featureGate = featurelevel.NewGate(store)
+	return nil
+}

spinifex/runtime/daemon/feature_level_test.go

@@ -0,0 +1,477 @@
+package daemon
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1"
+	testutil "github.com/mulgadc/spinifex/internal/testkit"
+	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
+	"github.com/mulgadc/spinifex/spinifex/foundation/release/featurelevel"
+	instancecache "github.com/mulgadc/spinifex/spinifex/runtime/compute/cache"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// levelManager opens both buckets on a server of its own, so the cluster-config
+// record is not shared with parallel tests.
+func levelManager(t *testing.T) (*JetStreamManager, *featurelevel.Store) {
+	t.Helper()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	m, err := NewJetStreamManager(nc)
+	require.NoError(t, err)
+	require.NoError(t, m.InitClusterConfigBucket())
+	require.NoError(t, m.InitClusterStateBucket())
+	store, err := m.FeatureLevelStore(t.Context())
+	require.NoError(t, err)
+	return m, store
+}
+
+func levelBeat(t *testing.T, m *JetStreamManager, node string, level int, now time.Time, age time.Duration) {
+	t.Helper()
+	require.NoError(t, m.WriteHeartbeat(&Heartbeat{
+		Node: node, Timestamp: now.Add(-age).UTC().Format(time.RFC3339), FeatureLevel: level,
+	}))
+}
+
+func tick(t *testing.T, m *JetStreamManager, store *featurelevel.Store, nodes []string, now time.Time) (featurelevel.Record, bool) {
+	t.Helper()
+	rec, advanced, err := advanceFeatureLevel(t.Context(), store, m.ReadHeartbeatLeader, nodes, now)
+	require.NoError(t, err)
+	return rec, advanced
+}
+
+func TestHeartbeatWithoutFeatureLevelReadsAsZeroAndBlocksAdvance(t *testing.T) {
+	t.Parallel()
+	m, store := levelManager(t)
+	now := time.Now()
+	// The shape an older build writes: no feature_level key at all.
+	old := `{"node":"old","epoch":1,"timestamp":"` + now.UTC().Format(time.RFC3339) + `","services":["daemon"],"vm_count":0}`
+	_, err := m.clusterKV.Put(t.Context(), clusterv1.HeartbeatKey("old"), []byte(old))
+	require.NoError(t, err)
+	h, err := m.ReadHeartbeat("old")
+	require.NoError(t, err)
+	assert.Zero(t, h.FeatureLevel)
+
+	levelBeat(t, m, "new", featurelevel.Build, now, 0)
+	rec, advanced := tick(t, m, store, []string{"new", "old"}, now)
+	assert.False(t, advanced, "a member on an older build holds the cluster at 0")
+	assert.Zero(t, rec.Level)
+}
+
+func TestFeatureLevelAdvancesOnceEveryMemberIsCurrent(t *testing.T) {
+	t.Parallel()
+	m, store := levelManager(t)
+	now := time.Now()
+	nodes := []string{"a", "b", "c"}
+	levelBeat(t, m, "a", featurelevel.Build, now, 0)
+	levelBeat(t, m, "b", featurelevel.Build, now, 0)
+
+	rec, advanced := tick(t, m, store, nodes, now)
+	assert.False(t, advanced, "member c has no heartbeat key (never written or expired from the 1h bucket)")
+	assert.Zero(t, rec.Level)
+
+	levelBeat(t, m, "c", featurelevel.Build, now, 0)
+	rec, advanced = tick(t, m, store, nodes, now)
+	assert.True(t, advanced)
+	assert.Equal(t, featurelevel.Build, rec.Level)
+}
+
+func TestFeatureLevelStaleMemberBlocksAndLaterStalenessNeverLowers(t *testing.T) {
+	t.Parallel()
+	m, store := levelManager(t)
+	now := time.Now()
+	nodes := []string{"a", "b"}
+	levelBeat(t, m, "a", featurelevel.Build, now, 0)
+	levelBeat(t, m, "b", featurelevel.Build, now, instancecache.NodeStaleAfter+5*time.Second)
+
+	rec, advanced := tick(t, m, store, nodes, now)
+	assert.False(t, advanced, "a heartbeat older than NodeStaleAfter is not current")
+	assert.Zero(t, rec.Level)
+
+	levelBeat(t, m, "b", featurelevel.Build, now, instancecache.NodeStaleAfter-5*time.Second)
+	rec, advanced = tick(t, m, store, nodes, now)
+	assert.True(t, advanced)
+	assert.Equal(t, featurelevel.Build, rec.Level)
+
+	later := now.Add(10 * time.Minute)
+	rec, advanced = tick(t, m, store, nodes, later)
+	assert.False(t, advanced)
+	assert.Equal(t, featurelevel.Build, rec.Level, "every member going stale afterwards does not lower the level")
+}
+
+func TestFeatureLevelReadErrorAdvancesNothing(t *testing.T) {
+	t.Parallel()
+	_, store := levelManager(t)
+	boom := errors.New("kv unavailable")
+	read := func(context.Context, string) (*Heartbeat, error) { return nil, boom }
+	rec, advanced, err := advanceFeatureLevel(t.Context(), store, read, []string{"a"}, time.Now())
+	require.ErrorIs(t, err, boom)
+	assert.False(t, advanced)
+	assert.Zero(t, rec.Level)
+	got, _, rerr := store.Read(t.Context())
+	require.NoError(t, rerr)
+	assert.Zero(t, got.Level)
+}
+
+func TestFeatureLevelUnparseableTimestampIsNotCurrent(t *testing.T) {
+	t.Parallel()
+	m, store := levelManager(t)
+	require.NoError(t, m.WriteHeartbeat(&Heartbeat{Node: "a", Timestamp: "yesterday", FeatureLevel: featurelevel.Build}))
+	_, advanced := tick(t, m, store, []string{"a"}, time.Now())
+	assert.False(t, advanced)
+}
+
+func TestFeatureLevelStoreNeedsTheClusterConfigBucket(t *testing.T) {
+	t.Parallel()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	m, err := NewJetStreamManager(nc)
+	require.NoError(t, err)
+	_, err = m.FeatureLevelStore(context.Background())
+	require.Error(t, err)
+}
+
+// flakyKV fails its first n Gets, then reads through.
+type flakyKV struct {
+	featurelevel.KV
+
+	failures int
+}
+
+func (k *flakyKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) {
+	if k.failures > 0 {
+		k.failures--
+		return nil, errors.New("kv unavailable")
+	}
+	return k.KV.Get(ctx, key)
+}
+
+func recordLevel(t *testing.T, m *JetStreamManager, body string) {
+	t.Helper()
+	kv, err := m.clusterConfig.KV(t.Context())
+	require.NoError(t, err)
+	_, err = kv.Put(t.Context(), featurelevel.Key, []byte(body))
+	require.NoError(t, err)
+}
+
+func TestEnforceFeatureLevel_RefusesOnlyWhenTheBuildIsBelowTheRecord(t *testing.T) {
+	t.Parallel()
+	m, store := levelManager(t)
+
+	require.NoError(t, enforceFeatureLevel(t.Context(), store, noPublish), "an absent record is level 0")
+
+	recordLevel(t, m, `{"level":1}`)
+	require.NoError(t, enforceFeatureLevel(t.Context(), store, noPublish), "equal starts")
+
+	recordLevel(t, m, `{"level":2}`)
+	err := enforceFeatureLevel(t.Context(), store, noPublish)
+	var refused *FeatureLevelRefusedError
+	require.ErrorAs(t, err, &refused)
+	require.ErrorIs(t, err, errStartupPermanent)
+	assert.Equal(t, featurelevel.Build, refused.Build)
+	assert.Equal(t, 2, refused.Recorded)
+	assert.Contains(t, err.Error(), "spx admin cluster rollback --to 1")
+	assert.Contains(t, err.Error(), "2")
+}
+
+func TestEnforceFeatureLevel_ReadFailureRetriesThenStartsWithoutRefusing(t *testing.T) {
+	noSleep(t)
+
+	m, store := levelManager(t)
+	_ = store
+	recordLevel(t, m, `{"level":1}`)
+	kv, err := m.clusterConfig.KV(t.Context())
+	require.NoError(t, err)
+	flaky := &flakyKV{KV: kv, failures: 3}
+	require.NoError(t, enforceFeatureLevel(t.Context(), featurelevel.NewStore(flaky), noPublish))
+	assert.Zero(t, flaky.failures, "it retried through every failure and then read the record")
+}
+
+func TestEnforceFeatureLevel_ReadFailureThenARefusingRecordStillRefuses(t *testing.T) {
+	noSleep(t)
+
+	m, _ := levelManager(t)
+	recordLevel(t, m, `{"level":3}`)
+	kv, err := m.clusterConfig.KV(t.Context())
+	require.NoError(t, err)
+	err = enforceFeatureLevel(t.Context(), featurelevel.NewStore(&flakyKV{KV: kv, failures: 2}), noPublish)
+	require.ErrorIs(t, err, errStartupPermanent)
+}
+
+// A build below the recorded level exits 1 before the EBS provider is claimed
+// or any cluster service is built.
+func TestStartCluster_BuildBelowFeatureLevelExitsBeforeAnyClaim(t *testing.T) {
+	t.Setenv("SPINIFEX_REQUIRE_NATS", "")
+	ns, _, js := testutil.StartTestJetStream(t)
+	seed := newLockManager(t, ns.ClientURL())
+	recordLevel(t, seed, `{"level":2}`)
+
+	tmpDir := t.TempDir()
+	cfg := &config.ClusterConfig{
+		Node: "node-1",
+		Nodes: map[string]config.Config{
+			"node-1": {BaseDir: tmpDir, DataDir: tmpDir, NATS: config.NATSConfig{Host: ns.ClientURL()}},
+		},
+	}
+	d, err := NewDaemon(cfg)
+	require.NoError(t, err)
+	t.Cleanup(d.cancel)
+	exitCh := make(chan int, 1)
+	d.exitFunc = func(code int) { exitCh <- code }
+
+	err = d.startCluster()
+	var refused *FeatureLevelRefusedError
+	require.ErrorAs(t, err, &refused)
+	select {
+	case code := <-exitCh:
+		assert.Equal(t, 1, code)
+	default:
+		t.Fatal("a build below the recorded level must request exit(1)")
+	}
+
+	kv, kerr := js.KeyValue(t.Context(), ClusterConfigBucket)
+	require.NoError(t, kerr)
+	_, gerr := kv.Get(t.Context(), EBSProviderKey)
+	require.ErrorIs(t, gerr, jetstream.ErrKeyNotFound, "claimEBSProvider must not have been reached")
+	assert.Nil(t, d.instanceService)
+	assert.Zero(t, d.vmMgr.Count())
+}
+
+// A node that has not set its gate must not answer the status route at all: a
+// reply would read as nothing in flight to the rollback barrier.
+func TestFeatureLevelStatusSubs_NilGateServesNothing(t *testing.T) {
+	t.Parallel()
+	d := &Daemon{node: "node-1"}
+	assert.Empty(t, d.featureLevelStatusSubs())
+
+	_, store := levelManager(t)
+	d.featureGate = featurelevel.NewGate(store)
+	subs := d.featureLevelStatusSubs()
+	require.Len(t, subs, 1)
+	assert.Equal(t, featurelevel.StatusSubject("node-1"), subs[0].topic)
+	assert.Empty(t, subs[0].queueGroup)
+
+	d.node = "a.b"
+	assert.Empty(t, d.featureLevelStatusSubs(), "an invalid node id serves nothing")
+}
+
+// A read that never returns on its own is cut off by the tick's context, so the
+// reconciliation cannot hold the heartbeat loop past its budget.
+func TestFeatureLevelAdvance_SlowReadsStopAtTheTickDeadline(t *testing.T) {
+	t.Parallel()
+	_, store := levelManager(t)
+	reads := 0
+	slow := func(ctx context.Context, _ string) (*Heartbeat, error) {
+		reads++
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}
+	ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
+	defer cancel()
+	start := time.Now()
+	_, advanced, err := advanceFeatureLevel(ctx, store, slow, []string{"a", "b", "c", "d"}, time.Now())
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+	assert.False(t, advanced)
+	assert.Equal(t, 1, reads, "scanning stops once the deadline has passed")
+	assert.Less(t, time.Since(start), 2*time.Second)
+}
+
+// The store a daemon and the operator commands use reads through the stream
+// leader, so a lagging replica cannot return the level from before a raise.
+func TestFeatureLevelStore_ReadsThroughTheLeader(t *testing.T) {
+	t.Parallel()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	m, err := NewJetStreamManager(nc)
+	require.NoError(t, err)
+	require.NoError(t, m.InitClusterConfigBucket())
+	store, err := m.FeatureLevelStore(t.Context())
+	require.NoError(t, err)
+
+	var mu sync.Mutex
+	var subjects []string
+	sub, err := nc.Subscribe("$JS.API.>", func(msg *nats.Msg) {
+		mu.Lock()
+		subjects = append(subjects, msg.Subject)
+		mu.Unlock()
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+
+	require.NoError(t, enforceFeatureLevel(t.Context(), store, noPublish))
+	require.NoError(t, nc.Flush())
+	mu.Lock()
+	defer mu.Unlock()
+	leaderGets := 0
+	for _, s := range subjects {
+		assert.False(t, strings.HasPrefix(s, "$JS.API.DIRECT.GET."), "startup check used a direct get: %s", s)
+		if s == "$JS.API.STREAM.MSG.GET.KV_"+ClusterConfigBucket {
+			leaderGets++
+		}
+	}
+	assert.Equal(t, 2, leaderGets, "the check and the re-read after the heartbeat write")
+}
+
+// Attaching never creates a bucket: a cluster whose daemons have not made them
+// reports them unavailable instead of changing the cluster.
+func TestAttachFeatureLevel_NeverCreatesABucket(t *testing.T) {
+	t.Parallel()
+	_, nc, js := testutil.StartTestJetStream(t)
+	m, err := NewJetStreamManager(nc)
+	require.NoError(t, err)
+	_, err = m.AttachFeatureLevel(t.Context())
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), ClusterConfigBucket)
+	_, kerr := js.KeyValue(t.Context(), ClusterConfigBucket)
+	require.ErrorIs(t, kerr, jetstream.ErrBucketNotFound, "attach must not create the bucket")
+	_, kerr = js.KeyValue(t.Context(), ClusterStateBucket)
+	require.ErrorIs(t, kerr, jetstream.ErrBucketNotFound)
+
+	other, onc, _ := testutil.StartTestJetStream(t)
+	_ = other
+	om, err := NewJetStreamManager(onc)
+	require.NoError(t, err)
+	require.NoError(t, om.InitClusterConfigBucket())
+	store, err := om.AttachFeatureLevel(t.Context())
+	require.NoError(t, err)
+	rec, _, err := store.Read(t.Context())
+	require.NoError(t, err)
+	assert.Zero(t, rec.Level)
+	assert.Nil(t, om.clusterKV, "a missing cluster-state bucket stays unattached and uncreated")
+}
+
+func noPublish() error { return nil }
+
+func noSleep(t *testing.T) *time.Duration {
+	t.Helper()
+	var slept time.Duration
+	old := featureLevelSleep
+	featureLevelSleep = func(_ context.Context, d time.Duration) error { slept += d; return nil }
+	t.Cleanup(func() { featureLevelSleep = old })
+	return &slept
+}
+
+// The startup read keeps retrying past the old five-minute budget; only the
+// daemon context ends it.
+func TestEnforceFeatureLevel_RetriesPastTheOldBudgetUntilCancelled(t *testing.T) {
+	slept := noSleep(t)
+	m, _ := levelManager(t)
+	recordLevel(t, m, `{"level":1}`)
+	kv, err := m.clusterConfig.KV(t.Context())
+	require.NoError(t, err)
+	flaky := &flakyKV{KV: kv, failures: 100}
+	require.NoError(t, enforceFeatureLevel(t.Context(), featurelevel.NewStore(flaky), noPublish))
+	assert.Greater(t, *slept, 5*time.Minute, "backoff time spent exceeds the old budget")
+
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	err = enforceFeatureLevel(ctx, featurelevel.NewStore(&flakyKV{KV: kv, failures: 1000}), noPublish)
+	require.Error(t, err)
+	assert.NotErrorIs(t, err, errStartupPermanent)
+}
+
+// A predecessor's heartbeat is replaced by this build's own before the daemon
+// starts, so its capability cannot count towards an advance.
+func TestEnforceFeatureLevel_OverwritesThePredecessorHeartbeat(t *testing.T) {
+	t.Parallel()
+	m, store := levelManager(t)
+	now := time.Now()
+	levelBeat(t, m, "n1", featurelevel.Build+1, now, 0)
+
+	publish := func() error {
+		return m.WriteHeartbeat(&Heartbeat{Node: "n1", Timestamp: now.UTC().Format(time.RFC3339), FeatureLevel: featurelevel.Build})
+	}
+	require.NoError(t, enforceFeatureLevel(t.Context(), store, publish))
+	h, err := m.ReadHeartbeat("n1")
+	require.NoError(t, err)
+	assert.Equal(t, featurelevel.Build, h.FeatureLevel)
+	rec, _ := tick(t, m, store, []string{"n1"}, now)
+	assert.LessOrEqual(t, rec.Level, featurelevel.Build)
+}
+
+// A raise that lands between the check and the heartbeat write is caught by the
+// re-read.
+func TestEnforceFeatureLevel_RaiseBetweenCheckAndWriteRefuses(t *testing.T) {
+	t.Parallel()
+	m, store := levelManager(t)
+	publish := func() error {
+		recordLevel(t, m, `{"level":2}`)
+		return nil
+	}
+	err := enforceFeatureLevel(t.Context(), store, publish)
+	var refused *FeatureLevelRefusedError
+	require.ErrorAs(t, err, &refused)
+	require.ErrorIs(t, err, errStartupPermanent)
+	assert.Equal(t, 2, refused.Recorded)
+}
+
+// Activation reads heartbeats from the stream leader, never a follower.
+func TestReadHeartbeatLeader_UsesTheLeaderPath(t *testing.T) {
+	t.Parallel()
+	_, nc, _ := testutil.StartTestJetStream(t)
+	m, err := NewJetStreamManager(nc)
+	require.NoError(t, err)
+	require.NoError(t, m.InitClusterStateBucket())
+	levelBeat(t, m, "n1", 1, time.Now(), 0)
+
+	var mu sync.Mutex
+	var subjects []string
+	sub, err := nc.Subscribe("$JS.API.>", func(msg *nats.Msg) {
+		mu.Lock()
+		subjects = append(subjects, msg.Subject)
+		mu.Unlock()
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+
+	h, err := m.ReadHeartbeatLeader(t.Context(), "n1")
+	require.NoError(t, err)
+	assert.Equal(t, 1, h.FeatureLevel)
+	_, err = m.ReadHeartbeatLeader(t.Context(), "absent")
+	require.ErrorIs(t, err, jetstream.ErrKeyNotFound)
+	require.NoError(t, nc.Flush())
+	mu.Lock()
+	defer mu.Unlock()
+	gets := 0
+	for _, s := range subjects {
+		assert.False(t, strings.HasPrefix(s, "$JS.API.DIRECT.GET."), "heartbeat read used a direct get: %s", s)
+		if s == "$JS.API.STREAM.MSG.GET.KV_"+ClusterStateBucket {
+			gets++
+		}
+	}
+	assert.Equal(t, 2, gets)
+}
+
+// A level raised above the running build ends the daemon on the next tick.
+func TestTickFeatureLevel_ExitsWhenTheLevelRisesAboveTheBuild(t *testing.T) {
+	t.Parallel()
+	m, _ := levelManager(t)
+	exitCh := make(chan int, 1)
+	d := &Daemon{
+		node:          "n1",
+		ctx:           t.Context(),
+		jsManager:     m,
+		clusterConfig: &config.ClusterConfig{Nodes: map[string]config.Config{"n1": {}}},
+		exitFunc:      func(code int) { exitCh <- code },
+	}
+	d.tickFeatureLevel()
+	select {
+	case <-exitCh:
+		t.Fatal("exited while the level was within the build")
+	default:
+	}
+
+	recordLevel(t, m, `{"level":2}`)
+	d.tickFeatureLevel()
+	select {
+	case code := <-exitCh:
+		assert.Equal(t, 1, code)
+	default:
+		t.Fatal("a level above the build must request exit(1) on the next tick")
+	}
+}

spinifex/runtime/daemon/heartbeat.go

@@ -1,9 +1,11 @@
 package daemon
 
 import (
-	"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
 	"log/slog"
 	"time"
+
+	"github.com/mulgadc/spinifex/spinifex/foundation/release/featurelevel"
+	"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
 )
 
 const heartbeatInterval = 10 * time.Second
@@ -20,6 +22,7 @@ func (d *Daemon) startHeartbeat() {
 	go func() {
 		// Fire immediately on startup
 		d.publishHeartbeat()
+		d.tickFeatureLevel()
 
 		ticker := time.NewTicker(heartbeatInterval)
 		defer ticker.Stop()
@@ -31,6 +34,7 @@ func (d *Daemon) startHeartbeat() {
 				return
 			case <-ticker.C:
 				d.publishHeartbeat()
+				d.tickFeatureLevel()
 			}
 		}
 	}()
@@ -69,5 +73,6 @@ func (d *Daemon) buildHeartbeat() *Heartbeat {
 		AvailableMem:  totalMem - allocMem,
 		ReservedVCPU:  reservedVCPU,
 		ReservedMem:   reservedMem,
+		FeatureLevel:  featurelevel.Build,
 	}
 }

spinifex/runtime/daemon/heartbeat_test.go

@@ -8,6 +8,7 @@ import (
 
 	clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1"
 	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
+	"github.com/mulgadc/spinifex/spinifex/foundation/release/featurelevel"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/cache"
 	"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
 	"github.com/nats-io/nats.go"
@@ -35,6 +36,7 @@ func TestBuildHeartbeat(t *testing.T) {
 	}
 
 	h := d.buildHeartbeat()
+	assert.Equal(t, featurelevel.Build, h.FeatureLevel, "heartbeat reports this build's feature level")
 
 	assert.Equal(t, "test-node", h.Node)
 	assert.Equal(t, uint64(5), h.Epoch)

spinifex/runtime/daemon/jetstream.go

@@ -12,6 +12,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1"
 	"github.com/mulgadc/spinifex/spinifex/domains/ec2/instance/uefistore"
+	"github.com/mulgadc/spinifex/spinifex/foundation/release/featurelevel"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
 	"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
 	"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
@@ -275,6 +276,10 @@ type Heartbeat struct {
 	AvailableMem  float64  `json:"available_mem_gb"`
 	ReservedVCPU  int      `json:"reserved_vcpu"`
 	ReservedMem   float64  `json:"reserved_mem_gb"`
+
+	// FeatureLevel is the highest cluster feature level this build supports.
+	// Heartbeats written before the field existed read as 0.
+	FeatureLevel int `json:"feature_level,omitempty"`
 }
 
 // WriteHeartbeat writes a heartbeat entry for the given node to the cluster-state KV.
@@ -292,10 +297,30 @@ func (m *JetStreamManager) WriteHeartbeat(h *Heartbeat) error {
 
 // ReadHeartbeat reads the heartbeat entry for the given node from the cluster-state KV.
 func (m *JetStreamManager) ReadHeartbeat(nodeID string) (*Heartbeat, error) {
+	return m.ReadHeartbeatContext(context.Background(), nodeID)
+}
+
+// ReadHeartbeatContext is ReadHeartbeat bounded by ctx.
+func (m *JetStreamManager) ReadHeartbeatContext(ctx context.Context, nodeID string) (*Heartbeat, error) {
 	if m.clusterKV == nil {
 		return nil, errors.New("cluster state KV not initialized")
 	}
-	entry, err := m.clusterKV.Get(context.Background(), clusterv1.HeartbeatKey(nodeID))
+	return readHeartbeat(ctx, m.clusterKV, nodeID)
+}
+
+// ReadHeartbeatLeader reads the heartbeat from the stream leader, so a lagging
+// replica cannot return one a restart has since replaced.
+func (m *JetStreamManager) ReadHeartbeatLeader(ctx context.Context, nodeID string) (*Heartbeat, error) {
+	if m.clusterKV == nil {
+		return nil, errors.New("cluster state KV not initialized")
+	}
+	return readHeartbeat(ctx, featurelevel.LeaderKV(m.js, m.clusterKV), nodeID)
+}
+
+func readHeartbeat(ctx context.Context, kv interface {
+	Get(context.Context, string) (jetstream.KeyValueEntry, error)
+}, nodeID string) (*Heartbeat, error) {
+	entry, err := kv.Get(ctx, clusterv1.HeartbeatKey(nodeID))
 	if err != nil {
 		return nil, err
 	}