amadan

carlos / platform Public

zdd plan 3 task 3: swap admission through hostwrite, claims off and on

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude f51b07666e336aaecf7d5c89ae2101ef49b05528 parent aeff21c
2 files changed, +304 −0
  • internal/activator/swap_admit_test.go +188 −0
  • internal/activator/swapadmit.go +116 −0
diff --git a/internal/activator/swap_admit_test.go b/internal/activator/swap_admit_test.go
new file mode 100644
index 000000000..de3dc3e51
--- /dev/null
+++ b/internal/activator/swap_admit_test.go
@@ -0,0 +1,188 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostwrite"
+)
+
+var errClaimChanged = errors.New("test: claim changed")
+
+// swapWork is claim-backed work. denyAt makes its n-th With refuse before
+// running fn, which is how hostguard refuses a changed claim or stamp.
+type swapWork struct {
+ mu sync.Mutex
+ calls int
+ denyAt int
+}
+
+func (w *swapWork) With(_ context.Context, id hostwrite.Identity, fn func() error) error {
+ w.mu.Lock()
+ w.calls++
+ n := w.calls
+ w.mu.Unlock()
+ if id != (hostwrite.Identity{}) {
+ return errors.New("test: a swap must not claim config or release inputs")
+ }
+ if n == w.denyAt {
+ return errClaimChanged
+ }
+ return fn()
+}
+
+func (w *swapWork) withCalls() int {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ return w.calls
+}
+
+type swapSelector struct {
+ work *swapWork
+ err error
+}
+
+func (s *swapSelector) Select(_ context.Context, _, action string) (hostwrite.Work, error) {
+ if action != "spawn" {
+ return nil, errors.New("test: unexpected action " + action)
+ }
+ if s.work == nil {
+ return nil, s.err
+ }
+ return s.work, s.err
+}
+
+// The four §7.4 checks, one at a time.
+func TestSwapChecks(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ host := "checks.example.com"
+ arrangeAwake(t, a, testRoute(host))
+ at := Attempt{Host: host, SchedRev: 4}
+ four := func(string) uint64 { return 4 }
+ check := func(newest func(string) uint64) error {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ return a.swapChecksLocked(a.entries[host], at, newest)
+ }
+ if err := check(four); err != nil {
+ t.Fatalf("an awake, leased, newest, unfenced entry was refused: %v", err)
+ }
+ if err := check(func(string) uint64 { return 5 }); !errors.Is(err, errSwapSuperseded) {
+ t.Fatalf("superseded: %v", err)
+ }
+ a.mu.Lock()
+ a.fences = map[string]uint64{host: 3}
+ a.mu.Unlock()
+ if err := check(four); !errors.Is(err, errSwapFenced) {
+ t.Fatalf("fenced: %v", err)
+ }
+ a.mu.Lock()
+ delete(a.fences, host)
+ a.shuttingDown = true
+ a.mu.Unlock()
+ if err := check(four); !errors.Is(err, errSwapShutdown) {
+ t.Fatalf("shutting down: %v", err)
+ }
+ a.mu.Lock()
+ a.shuttingDown = false
+ a.mu.Unlock()
+ b.lastRep().setErr(errors.New("renewal failed"))
+ if err := check(four); !errors.Is(err, errSwapLeaseLost) {
+ t.Fatalf("lease lost: %v", err)
+ }
+ b.lastRep().setErr(nil)
+ a.mu.Lock()
+ a.entries[host].state = actStopping
+ a.mu.Unlock()
+ if err := check(four); !errors.Is(err, errSwapNotAwake) {
+ t.Fatalf("stopping: %v", err)
+ }
+ a.mu.Lock()
+ a.entries[host].state = actAwake
+ err := a.swapChecksLocked(nil, at, four)
+ a.mu.Unlock()
+ if !errors.Is(err, errSwapNotAwake) {
+ t.Fatalf("asleep: %v", err)
+ }
+}
+
+// Spec §7.4: ownership is checked apart from supersession, so an exec that
+// is not an attempt's (a crash restart, a fallback, a restore, a sidecar)
+// can ask it. A superseded attempt still owns the host; a lost lease does
+// not, whatever the revision says.
+func TestOwnershipIsCheckedApartFromSupersession(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ host := "owned.example.com"
+ arrangeAwake(t, a, testRoute(host))
+ owned := func(waking bool) error {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ return a.ownedLocked(a.entries[host], waking)
+ }
+ if err := owned(false); err != nil {
+ t.Fatalf("an awake, leased, unfenced entry: %v", err)
+ }
+ b.lastRep().setErr(errors.New("renewal failed"))
+ if err := owned(false); !errors.Is(err, errSwapLeaseLost) {
+ t.Fatalf("lease lost: %v", err)
+ }
+ if err := owned(true); err != nil {
+ t.Fatalf("a waking entry's lease is the wake's own: %v", err)
+ }
+ b.lastRep().setErr(nil)
+ a.mu.Lock()
+ a.fences = map[string]uint64{host: 1}
+ a.mu.Unlock()
+ if err := owned(true); !errors.Is(err, errSwapFenced) {
+ t.Fatalf("fenced while waking: %v", err)
+ }
+}
+
+// Claims off: hostwrite's legacy work. The check runs alone, under the
+// entry lock that fn takes itself.
+func TestAdmitClaimsOffRunsTheCheckAlone(t *testing.T) {
+ a := New(Config{}, &fakeBackend{}, nil)
+ work, err := a.selectWork(context.Background(), "h.example.com")
+ if err != nil || !hostwrite.Legacy(work) {
+ t.Fatalf("claims off selected %T, %v; want the legacy work", work, err)
+ }
+ ran := 0
+ if err := a.admit(context.Background(), work, func() error { ran++; return nil }); err != nil || ran != 1 {
+ t.Fatalf("admit = %v, ran %d", err, ran)
+ }
+}
+
+// Claims on: the claim and stamp are validated first, and a refusal is
+// final. A selector that cannot answer refuses. Nothing is ever read as
+// claims off.
+func TestAdmitClaimsOnValidatesFirstAndNeverFallsBack(t *testing.T) {
+ ctx := context.Background()
+ w := &swapWork{}
+ a := New(Config{HostWrites: hostwrite.Config{Enabled: true, Selector: &swapSelector{work: w}}}, &fakeBackend{}, nil)
+ work, err := a.selectWork(ctx, "h.example.com")
+ if err != nil || hostwrite.Legacy(work) {
+ t.Fatalf("claims on selected %T, %v", work, err)
+ }
+ ran := 0
+ fn := func() error { ran++; return nil }
+ if err := a.admit(ctx, work, fn); err != nil || ran != 1 || w.withCalls() != 1 {
+ t.Fatalf("admit = %v, ran %d, With %d", err, ran, w.withCalls())
+ }
+ w.denyAt = 2
+ if err := a.admit(ctx, work, fn); !errors.Is(err, errClaimChanged) || ran != 1 {
+ t.Fatalf("refused claim: err %v, ran %d (fn must not run)", err, ran)
+ }
+ for name, sel := range map[string]hostwrite.Selector{
+ "selector error": &swapSelector{err: hostwrite.ErrUnavailable},
+ "no selector": nil,
+ "nil work": &swapSelector{},
+ } {
+ a.cfg.HostWrites.Selector = sel
+ if _, err := a.selectWork(ctx, "h.example.com"); !errors.Is(err, hostwrite.ErrUnavailable) {
+ t.Errorf("%s: %v, want hostwrite.ErrUnavailable", name, err)
+ }
+ }
+}
diff --git a/internal/activator/swapadmit.go b/internal/activator/swapadmit.go
new file mode 100644
index 000000000..d11532d6e
--- /dev/null
+++ b/internal/activator/swapadmit.go
@@ -0,0 +1,116 @@
+package activator
+
+import (
+ "context"
+ "errors"
+
+ "github.com/carlosframework/platform/internal/hostwrite"
+)
+
+var (
+ errSwapNotAwake = errors.New("swap: instance not awake")
+ errSwapLeaseLost = errors.New("swap: lease lost")
+ errSwapFenced = errors.New("swap: a stop-evidence operation holds the host")
+ errSwapShutdown = errors.New("swap: agent shutting down")
+ errSwapSuperseded = errors.New("swap: superseded by a newer attempt")
+ errSwapStopped = errors.New("swap: an intentional stop cancelled the attempt")
+)
+
+// selectWork picks the attempt's lifecycle work through hostwrite, never
+// hostguard directly (spec §7.4). Claims off gives the legacy work. Claims
+// on gives the claim-backed work, and a selector that cannot answer is a
+// refusal, never the legacy work.
+func (a *Service) selectWork(ctx context.Context, host string) (hostwrite.Work, error) {
+ work, err := a.cfg.HostWrites.Select(ctx, host, "spawn")
+ if err != nil {
+ return nil, err
+ }
+ if work == nil {
+ return nil, hostwrite.ErrUnavailable
+ }
+ return work, nil
+}
+
+// admit runs fn as a §7.4 re-check. Claims off (hostwrite.Legacy): fn
+// alone. fn takes a.mu itself, so the check is the entry lock against the
+// barrier, the lease and the local fences that maintenance and moves set.
+// Claims on: fn runs inside the selected work's With, so the claim and the
+// installation stamp are validated first. A refusal there is returned as
+// is and is never retried down the claims-off path.
+func (a *Service) admit(ctx context.Context, work hostwrite.Work, fn func() error) error {
+ if work == nil {
+ return hostwrite.ErrUnavailable
+ }
+ if hostwrite.Legacy(work) {
+ return fn()
+ }
+ return work.With(ctx, hostwrite.Identity{}, fn)
+}
+
+// swapChecksLocked is the four §7.4 checks on the entry the attempt targets:
+// the entry is still awake, the lease is still held, no stop-evidence
+// operation (the local fence) is in progress, and the attempt is still the
+// newest. The agent's own shutdown counts as a stop in progress. newest is
+// called with a.mu held. The caller holds a.mu.
+func (a *Service) swapChecksLocked(e *actEntry, at Attempt, newest func(string) uint64) error {
+ return a.attemptCurrentLocked(e, at, newest, false)
+}
+
+// attemptCurrentLocked is THE attempt gate (spec §6.3, §7.4): every exec
+// point and every completion path of every attempt asks it, through
+// swapChecksLocked, memberCurrentLocked, a hold's pinCheck, a claims-on
+// spawn's before hook and a socketless respawn's attemptGate. waking is a
+// hold's pinned wake, whose entry is still waking and holds no lease yet:
+// the entry must be the host's, and the agent, the fences and the revision
+// are checked. Otherwise the entry must be awake with its lease. The caller
+// holds a.mu.
+func (a *Service) attemptCurrentLocked(e *actEntry, at Attempt, newest func(string) uint64, waking bool) error {
+ if e == nil || a.entries[at.Host] != e {
+ return errSwapNotAwake
+ }
+ if err := a.ownedLocked(e, waking); err != nil {
+ return err
+ }
+ if newest(at.Host) != at.SchedRev {
+ return errSwapSuperseded
+ }
+ return nil
+}
+
+// ownedLocked is the gate's OWNERSHIP check, apart from supersession (spec
+// §7.4). Every exec asks it, an attempt's or not: a wake's and its
+// fallback's, a worker respawn's (a crash restart, an attempt, its fallback,
+// its restore) and a legacy sidecar's. Its refusals (errSwapNotAwake,
+// errSwapLeaseLost, errSwapFenced, errSwapShutdown) are all gateNoOwnership:
+// nothing spawns, a restore included. waking is an entry still waking,
+// whose lease the wake holds itself (doWake's lease, not e.rep): the entry
+// must be the host's, with no fence and no shutdown. The caller holds a.mu.
+func (a *Service) ownedLocked(e *actEntry, waking bool) error {
+ switch {
+ case e == nil || a.entries[e.host] != e:
+ return errSwapNotAwake
+ case !waking && e.state != actAwake: // swapsim:M5
+ return errSwapNotAwake
+ case !waking && (e.rep == nil || e.rep.Err() != nil):
+ return errSwapLeaseLost
+ case a.fences[e.host] != 0:
+ return errSwapFenced
+ case a.shuttingDown:
+ return errSwapShutdown
+ }
+ return nil
+}
+
+// memberCurrentLocked is the gate for an attempt with a member in the
+// operation set (an overlap's successor): the attempt is current, and its
+// member is still in the set and not cancelled by the barrier. The caller
+// holds a.mu.
+func (a *Service) memberCurrentLocked(e *actEntry, m *genMember, mctx context.Context, at Attempt, newest func(string) uint64) error {
+ if err := a.attemptCurrentLocked(e, at, newest, false); err != nil {
+ return err
+ }
+ if e.ops.members[m.gen.N] != m || mctx.Err() != nil {
+ return errSwapStopped
+ }
+ return nil
+}