carlos / platform Public

Fence activator admission and retain checked stop evidence

Paul Campbell pushed by paul@keymail.dev 1 month ago 897fbc2845736e48f3f0d4c784e1f3e710fcc737 parent b3a7ff5

3 files changed, +709 −27

internal/activator/activator.go +119 −27
internal/activator/maintenance.go +157 −0
internal/activator/maintenance_test.go +433 −0

diff --git a/internal/activator/activator.go b/internal/activator/activator.go

index dd5fbbf24..a22fd3e9a 100644

--- a/internal/activator/activator.go

+++ b/internal/activator/activator.go

@@ -260,9 +260,12 @@ type actEntry struct {

route registry.Route

state actState

- ready chan struct{} // waking: closed when the wake resolves; then check err

- err error // waking: the wake's outcome

- stopped chan struct{} // stopping: closed when the stop completes

+ ready chan struct{} // waking: closed when the wake resolves; then check err

+ err error // waking: the wake's outcome

+ stopped chan struct{} // stopping: closed when the stop completes

+ wakeCancel context.CancelFunc

+ stopOutcome *stopObservation

+ sidecarUncertain bool

// Activity tracking: an instance is idle only when nothing is in flight

// AND the idle window has passed since the last request finished. In-flight

@@ -396,8 +399,10 @@ type Service struct {

// codepath was touched at all. Single-box v1 drops the field; it re-enters

// with the mesh port.

- mu sync.Mutex

- entries map[string]*actEntry

+ mu sync.Mutex

+ entries map[string]*actEntry

+ fences map[string]uint64

+ stopHistory map[string]*stopObservation

sweepStop chan struct{}

sweepDone chan struct{}

@@ -454,6 +459,10 @@ func (a *Service) begin(ctx context.Context, rt registry.Route, stamp bool) (rel

defer deadline.Stop()

for {

a.mu.Lock()

+ if a.fences[rt.Host] != 0 {

+ a.mu.Unlock()

+ return nil, ErrFenced

+ }

e := a.entries[rt.Host]

if e == nil {

// Asleep → kick off ONE wake; this request and any latecomers all

@@ -549,6 +558,10 @@ func (a *Service) BeginSidecarRequest(ctx context.Context, rt registry.Route) er

defer deadline.Stop()

for {

a.mu.Lock()

+ if a.fences[rt.Host] != 0 {

+ a.mu.Unlock()

+ return ErrFenced

+ }

e := a.entries[rt.Host]

if e == nil {

a.mu.Unlock()

@@ -628,6 +641,12 @@ func (a *Service) releaseFunc(e *actEntry, stamp bool) func() {

func (a *Service) wake(e *actEntry) {

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)

defer cancel()

+ a.mu.Lock()

+ e.wakeCancel = cancel

+ if a.fences[e.host] != 0 {

+ cancel()

+ }

+ a.mu.Unlock()

wakeStart := a.now()

w, err := a.doWake(ctx, e.route)

if w != nil {

@@ -646,11 +665,17 @@ func (a *Service) wake(e *actEntry) {

a.mu.Lock()

if err != nil {

e.err = err

+ outcome := &stopObservation{uncertain: true}

+ if w != nil && w.stopOutcome != nil {

+ outcome = w.stopOutcome

+ }

+ a.rememberStopLocked(e.host, outcome)

delete(a.entries, e.host) // next request retries from scratch

} else {

e.state = actAwake

e.rep, e.proc = w.rep, w.proc

e.sidecar, e.sidecarWanted = w.sidecar, w.sidecarWanted

+ e.sidecarUncertain = w.sidecarUncertain

e.sidecarStarted = a.now()

e.sidecarBackoff = 0

e.sidecarRetryAt = time.Time{} // a wake-time spawn failure retries on the next sweep

@@ -662,7 +687,10 @@ func (a *Service) wake(e *actEntry) {

// cycle. Fresh wake, so it can't be stale yet.

e.spawnVersion = w.spawnVersion

e.staleSince = time.Time{}

- if a.OnWake != nil {

+ if a.fences[e.host] != 0 {

+ a.beginStopLocked(e)

+ }

+ if a.OnWake != nil && a.fences[e.host] == 0 {

// Fire-and-forget, off this goroutine: e.host is read once here,

// before this goroutine can be reused for anything else — the

// hook must never share e.host's own goroutine's fate (a wedged

@@ -719,11 +747,13 @@ func wakeEventMessage(w *wokenInstance) string {

// members > bots, so a host whose sidecar won't start still serves (the

// sweeper retries via sidecarWanted).

type wokenInstance struct {

- rep Rep

- proc Proc

- sidecar Proc

- sidecarWanted bool

- spawnVersion string // live/<host> target read at exec — for build convergence

+ stopOutcome *stopObservation

+ sidecarUncertain bool

+ rep Rep

+ proc Proc

+ sidecar Proc

+ sidecarWanted bool

+ spawnVersion string // live/<host> target read at exec — for build convergence

// spawnTook is how long Backend.Start took: exec to the socket accepting,

// which is the instance's OWN boot and nothing of the platform's. wakeTook

// is the whole wake around it, filled in by wake() rather than here

@@ -740,6 +770,10 @@ type wokenInstance struct {

func (a *Service) doWake(ctx context.Context, rt registry.Route) (*wokenInstance, error) {

host := rt.Host

+ w := &wokenInstance{stopOutcome: &stopObservation{}}

+ if ctx.Err() != nil || !a.wakeAllowed(host) {

+ return w, ErrFenced

+ }

// 1. The lease first: everything after this line is done as the prefix's

// single writer. Contention is not retried here — on one box it means a

@@ -752,9 +786,13 @@ func (a *Service) doWake(ctx context.Context, rt registry.Route) (*wokenInstance

log.Printf("activator: LEASE HELD for %s by %q — a previous incarnation hasn't released; refusing to wake (retry after %s)",

host, held.holder, held.retryAfter)

}

- return nil, err

+ return w, err

}

+ if ctx.Err() != nil || !a.wakeAllowed(host) {

+ w.stopOutcome.leaseReleased = lease.Release(ctx) == nil

+ return w, ErrFenced

+ }

// 2. Restore the DB from S3 when the local copy is gone — the restore must

// happen *under* the lock (restoring without it could later shadow newer

// S3 state written by the true holder). When it's still on disk it is the

@@ -762,8 +800,8 @@ func (a *Service) doWake(ctx context.Context, rt registry.Route) (*wokenInstance

// what we ourselves checkpointed at the last sleep.

fresh, err := a.backend.Restore(ctx, rt)

if err != nil {

- _ = lease.Release(ctx)

- return nil, fmt.Errorf("restore %s: %w", host, err)

+ w.stopOutcome.leaseReleased = lease.Release(ctx) == nil

+ return w, fmt.Errorf("restore %s: %w", host, err)

}

// 3. First-boot seed, strictly between the restore and the spawn — so it

@@ -774,11 +812,17 @@ func (a *Service) doWake(ctx context.Context, rt registry.Route) (*wokenInstance

// can never run over recovered data. A seed that fails fails the WHOLE

// wake (the Start-failure precedent: release the lease, surface the

// error) — a half-seeded instance must not serve.

+ if ctx.Err() != nil || !a.wakeAllowed(host) {

+ w.stopOutcome.leaseReleased = lease.Release(ctx) == nil

+ return w, ErrFenced

+ }

if fresh && rt.Seed != "" {

+ w.stopOutcome.uncertain = true

if err := a.backend.Seed(ctx, rt); err != nil {

- _ = lease.Release(ctx)

- return nil, fmt.Errorf("seed %s: %w", host, err)

+ w.stopOutcome.leaseReleased = lease.Release(ctx) == nil

+ return w, fmt.Errorf("seed %s: %w", host, err)

}

+ w.stopOutcome.uncertain = false

}

// 4. Spawn before replication starts: a fresh instance creates its own DB

@@ -791,14 +835,24 @@ func (a *Service) doWake(ctx context.Context, rt registry.Route) (*wokenInstance

// parsing templates — and none of the platform's lease, restore or

// replication work. An app measuring its own TTFB starts the clock after

// all of it (discussion #6).

+ if ctx.Err() != nil || !a.wakeAllowed(host) {

+ w.stopOutcome.leaseReleased = lease.Release(ctx) == nil

+ return w, ErrFenced

+ }

+ w.stopOutcome.uncertain = true

spawnStart := a.now()

proc, spawnVer, err := a.backend.Start(ctx, rt)

spawnTook := a.now().Sub(spawnStart)

+ w.proc = proc

if err != nil {

- _ = lease.Release(ctx)

- return nil, fmt.Errorf("start %s: %w", host, err)

+ w.stopOutcome.leaseReleased = lease.Release(ctx) == nil

+ return w, fmt.Errorf("start %s: %w", host, err)

}

+ if ctx.Err() != nil || !a.wakeAllowed(host) {

+ a.cleanupFailedWake(ctx, w, lease)

+ return w, ErrFenced

+ }

// 5. Trade the held lease for running replication, at the route's

// replication-class sync interval (the per-plan durability ladder).

// The DB path rides the route; when empty the backend defaults it to

@@ -806,11 +860,8 @@ func (a *Service) doWake(ctx context.Context, rt registry.Route) (*wokenInstance

// global — base dirs are backend fields now).

rep, err := lease.Replicate(ctx, rt.DBPath, replicationSyncInterval(rt.Replication))

if err != nil {

- stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)

- defer cancel()

- _ = proc.Stop(stopCtx)

- _ = lease.Release(ctx)

- return nil, fmt.Errorf("replicate %s: %w", host, err)

+ a.cleanupFailedWake(ctx, w, lease)

+ return w, fmt.Errorf("replicate %s: %w", host, err)

}

// 6. Co-spawn the sidecar, last — only once the instance is healthy and

@@ -818,10 +869,15 @@ func (a *Service) doWake(ctx context.Context, rt registry.Route) (*wokenInstance

// marked requests pass through the edge while we're awake). A spawn

// failure is loud but never fails the wake: members come before bots, and

// the sweeper retries with backoff.

- w := &wokenInstance{rep: rep, proc: proc, spawnVersion: spawnVer, spawnTook: spawnTook}

+ w.rep, w.spawnVersion, w.spawnTook = rep, spawnVer, spawnTook

+ w.stopOutcome = nil

+ if ctx.Err() != nil || !a.wakeAllowed(host) {

+ return w, nil

+ }

if a.backend.HasSidecar(rt) {

w.sidecarWanted = true

if sc, err := a.backend.StartSidecar(ctx, rt); err != nil {

+ w.sidecarUncertain = true

log.Printf("activator: SIDECAR for %s failed to start: %v — serving without it (the sweep retries)", host, err)

} else {

w.sidecar = sc

@@ -889,6 +945,7 @@ func (a *Service) Bounce(host string) bool {

// the entry. The local DB stays in place — it's the warm cache for the next

// wake; S3 is the durable truth under the lease.

func (a *Service) stop(e *actEntry) {

+ outcome := &stopObservation{sidecarStop: e.sidecar == nil, sidecarExited: e.sidecar == nil, uncertain: e.sidecarUncertain}

// The deliberate-stop marker for the member's timeline: hibernation,

// convergence cycling, bounce, shutdown all pass through here, so an

// "exited" line WITHOUT this line before it reads as what it is — a

@@ -903,7 +960,9 @@ func (a *Service) stop(e *actEntry) {

// first, the schedule the next-due probe reads can't shift under it.

if e.sidecar != nil {

scCtx, scCancel := context.WithTimeout(context.Background(), sidecarStopTimeout)

- if err := e.sidecar.Stop(scCtx); err != nil {

+ err := e.sidecar.Stop(scCtx)

+ outcome.sidecarStop, outcome.sidecarExited = err == nil, e.sidecar.Exited()

+ if err != nil {

log.Printf("activator: stop sidecar %s: %v", e.host, err)

}

scCancel()

@@ -939,7 +998,12 @@ func (a *Service) stop(e *actEntry) {

// Separate budgets: a SIGTERM drain that runs long must not eat the final

// checkpoint's time — the S3 sync is the part that makes sleep durable.

procCtx, cancelProc := context.WithTimeout(context.Background(), actStopTimeout)

- err = e.proc.Stop(procCtx)

+ if e.proc != nil {

+ err = e.proc.Stop(procCtx)

+ outcome.mainStop, outcome.mainExited = err == nil, e.proc.Exited()

+ } else {

+ err = ErrStopUnproven

+ }

cancelProc()

if err != nil {

log.Printf("activator: stop %s: %v", e.host, err)

@@ -949,10 +1013,18 @@ func (a *Service) stop(e *actEntry) {

// a local checkpoint.

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)

defer cancel()

- if err := e.rep.Close(ctx); err != nil {

+ repErr := ErrStopUnproven

+ if e.rep != nil {

+ repErr = e.rep.Close(ctx)

+ }

+ outcome.replicaClosed = repErr == nil

+ if err := repErr; err != nil {

log.Printf("activator: final checkpoint %s: %v", e.host, err)

}

a.mu.Lock()

+ outcome.uncertain = !outcome.complete()

+ e.stopOutcome = outcome

+ a.rememberStopLocked(e.host, outcome)

delete(a.entries, e.host)

a.mu.Unlock()

close(e.stopped)

@@ -1018,6 +1090,9 @@ func (a *Service) sweepOnce() {

a.mu.Lock()

cycled := 0 // build-convergence cycles started this tick (rate-limited)

for _, e := range a.entries {

+ if a.fences[e.host] != 0 {

+ continue

+ }

if e.state != actAwake {

continue

}

@@ -1105,6 +1180,7 @@ func (a *Service) sweepOnce() {

e.sidecarRetryAt = now.Add(e.sidecarBackoff)

sc, err := a.backend.StartSidecar(context.Background(), e.route)

if err != nil {

+ e.sidecarUncertain = true

log.Printf("activator: SIDECAR restart for %s failed: %v (next attempt in %s)", e.host, err, e.sidecarBackoff)

e.sidecar = nil

continue

@@ -1134,6 +1210,9 @@ func (a *Service) sweepOnce() {

if !rt.Hibernate || rt.Status != "ready" {

continue

}

+ if a.fences[rt.Host] != 0 {

+ continue

+ }

if _, live := a.entries[rt.Host]; live {

continue

}

@@ -1271,6 +1350,13 @@ func (a *Service) ServeScheduled(w http.ResponseWriter, r *http.Request, rt regi

}

func (a *Service) serve(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler, stamp bool) bool {

+ a.mu.Lock()

+ fenced := a.fences[rt.Host] != 0

+ a.mu.Unlock()

+ if fenced {

+ maintenanceResponse(w)

+ return true

+ }

// Per-host gating: only a hibernate-flagged instance is activator-managed;

// unflagged instances and sibling services are the edge's to proxy

// straight — the deploy-safety property that lets the bucket env ship

@@ -1296,6 +1382,10 @@ func (a *Service) serve(w http.ResponseWriter, r *http.Request, rt registry.Rout

// could spoof.

r.Header.Del(SidecarHeader)

if err := a.BeginSidecarRequest(r.Context(), rt); err != nil {

+ if errors.Is(err, ErrFenced) {

+ maintenanceResponse(w)

+ return true

+ }

if errors.Is(err, context.Canceled) {

return true // the sidecar hung up while queued on a wake

}

@@ -1320,6 +1410,8 @@ func (a *Service) serve(w http.ResponseWriter, r *http.Request, rt registry.Rout

if err != nil {

var held *errLeaseHeld

switch {

+ case errors.Is(err, ErrFenced):

+ maintenanceResponse(w)

case errors.As(err, &held):

retry := int(held.retryAfter/time.Second) + 1

w.Header().Set("Retry-After", strconv.Itoa(retry))

diff --git a/internal/activator/maintenance.go b/internal/activator/maintenance.go

new file mode 100644

index 000000000..a643d3eb5

--- /dev/null

+++ b/internal/activator/maintenance.go

@@ -0,0 +1,157 @@

+package activator

+import (

+ "context"

+ "errors"

+ "net/http"

+ "time"

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

+var (

+ ErrFenced = errors.New("activator: maintenance fence active")

+ ErrFenceGeneration = errors.New("activator: maintenance generation refused")

+ ErrStopUnproven = errors.New("activator: stop evidence unavailable")

+// StopEvidence records local process and replica closure. It grants no lease,

+// cgroup, database or resume authority.

+type StopEvidence struct {

+ host string

+ generation uint64

+ observation *stopObservation

+func (e *StopEvidence) Matches(host string, generation uint64) bool {

+ return e != nil && e.host == host && e.generation == generation && e.observation != nil && e.observation.complete()

+type stopObservation struct {

+ leaseReleased bool

+ mainStop, mainExited bool

+ sidecarStop, sidecarExited bool

+ replicaClosed bool

+ uncertain bool

+func (o *stopObservation) complete() bool {

+ return o != nil && !o.uncertain && o.mainStop && o.mainExited && o.sidecarStop && o.sidecarExited && o.replicaClosed

+// FenceAndStop keeps admission closed after cancellation or failure. A missing

+// entry is not evidence that an earlier process or replica stopped successfully.

+func (a *Service) FenceAndStop(ctx context.Context, host string, generation uint64) (*StopEvidence, error) {

+ if a == nil || !provision.ValidHostShape(host) || generation == 0 {

+ return nil, ErrFenceGeneration

+ }

+ if ctx == nil {

+ return nil, ErrStopUnproven

+ }

+ deadline, ok := ctx.Deadline()

+ if !ok || ctx.Err() != nil || !time.Now().Before(deadline) {

+ return nil, ErrStopUnproven

+ }

+ for {

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

+ return nil, err

+ }

+ if a.fences == nil {

+ a.fences = make(map[string]uint64)

+ }

+ if prior := a.fences[host]; prior > generation {

+ a.mu.Unlock()

+ return nil, ErrFenceGeneration

+ }

+ a.fences[host] = generation

+ e := a.entries[host]

+ if e == nil {

+ outcome := a.stopHistory[host]

+ a.mu.Unlock()

+ if ctx.Err() != nil || !time.Now().Before(deadline) {

+ return nil, ErrStopUnproven

+ }

+ if !outcome.complete() {

+ return nil, ErrStopUnproven

+ }

+ return &StopEvidence{host: host, generation: generation, observation: outcome}, nil

+ }

+ var wait <-chan struct{}

+ switch e.state {

+ case actWaking:

+ if e.wakeCancel != nil {

+ e.wakeCancel()

+ }

+ wait = e.ready

+ case actAwake:

+ a.beginStopLocked(e)

+ wait = e.stopped

+ case actStopping:

+ wait = e.stopped

+ default:

+ a.mu.Unlock()

+ return nil, ErrStopUnproven

+ }

+ a.mu.Unlock()

+ select {

+ case <-ctx.Done():

+ return nil, ErrStopUnproven

+ case <-wait:

+ }

+func (a *Service) lockFence(ctx context.Context) error {

+ timer := time.NewTicker(time.Millisecond)

+ defer timer.Stop()

+ for {

+ if ctx.Err() != nil {

+ return ErrStopUnproven

+ }

+ if a.mu.TryLock() {

+ if ctx.Err() != nil {

+ a.mu.Unlock()

+ return ErrStopUnproven

+ }

+ return nil

+ }

+ select {

+ case <-ctx.Done():

+ return ErrStopUnproven

+ case <-timer.C:

+ }

+func (a *Service) wakeAllowed(host string) bool {

+ a.mu.Lock()

+ defer a.mu.Unlock()

+ return a.fences[host] == 0

+func (a *Service) rememberStopLocked(host string, observation *stopObservation) {

+ if a.stopHistory == nil {

+ a.stopHistory = make(map[string]*stopObservation)

+ }

+ if prior := a.stopHistory[host]; prior != nil && prior.uncertain {

+ retained := *observation

+ retained.uncertain = true

+ observation = &retained

+ }

+ a.stopHistory[host] = observation

+func maintenanceResponse(w http.ResponseWriter) {

+ w.Header().Set("Cache-Control", "no-store")

+ w.Header().Set("Retry-After", "5")

+ http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)

+func (a *Service) cleanupFailedWake(ctx context.Context, w *wokenInstance, lease Lease) {

+ stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)

+ defer cancel()

+ if w.proc != nil {

+ err := w.proc.Stop(stopCtx)

+ w.stopOutcome.mainStop, w.stopOutcome.mainExited = err == nil, w.proc.Exited()

+ }

+ // Replicate failure does not prove that no partial replica was opened.

+ // Preserve that uncertainty even when both cleanup calls return nil.

+ w.stopOutcome.leaseReleased = lease.Release(ctx) == nil

diff --git a/internal/activator/maintenance_test.go b/internal/activator/maintenance_test.go

new file mode 100644

index 000000000..82cc6cf0c

--- /dev/null

+++ b/internal/activator/maintenance_test.go

@@ -0,0 +1,433 @@

+package activator

+import (

+ "context"

+ "errors"

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

+ "net/http"

+ "net/http/httptest"

+ "sync/atomic"

+ "testing"

+ "time"

+type maintenanceProc struct {

+ calls atomic.Int32

+ exited atomic.Bool

+ gate <-chan struct{}

+ entered chan struct{}

+ err error

+ remains bool

+func (p *maintenanceProc) Stop(ctx context.Context) error {

+ if p.calls.Add(1) == 1 && p.entered != nil {

+ close(p.entered)

+ }

+ if p.gate != nil {

+ select {

+ case <-p.gate:

+ case <-ctx.Done():

+ return ctx.Err()

+ }

+ if !p.remains {

+ p.exited.Store(true)

+ }

+ return p.err

+func (p *maintenanceProc) Exited() bool { return p.exited.Load() }

+type maintenanceRep struct {

+ calls atomic.Int32

+ err error

+func (r *maintenanceRep) Err() error { return nil }

+func (r *maintenanceRep) Close(context.Context) error { r.calls.Add(1); return r.err }

+func maintenanceContext(t *testing.T, duration time.Duration) context.Context {

+ t.Helper()

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

+ t.Cleanup(cancel)

+ return ctx

+func maintenanceAwake(t *testing.T) (*Service, *actEntry, *maintenanceProc, *maintenanceProc, *maintenanceRep) {

+ t.Helper()

+ a, _, _, reg := testActivator(t, time.Hour)

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

+ main, side, rep := &maintenanceProc{}, &maintenanceProc{}, &maintenanceRep{}

+ e := &actEntry{host: "fence.example", route: testRoute("fence.example"), state: actAwake, proc: main, sidecar: side, rep: rep, lastActive: a.now()}

+ a.entries[e.host] = e

+ return a, e, main, side, rep

+func TestMaintenanceFenceChecksEveryStopOutcome(t *testing.T) {

+ for _, scenario := range []string{"success", "main-error", "sidecar-error", "replica-error", "main-alive", "sidecar-alive", "missing-main", "missing-replica", "uncertain-sidecar"} {

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

+ a, e, main, side, rep := maintenanceAwake(t)

+ fail := errors.New("private backend detail")

+ switch scenario {

+ case "main-error":

+ main.err = fail

+ case "sidecar-error":

+ side.err = fail

+ case "replica-error":

+ rep.err = fail

+ case "main-alive":

+ main.remains = true

+ case "sidecar-alive":

+ side.remains = true

+ case "missing-main":

+ e.proc = nil

+ case "missing-replica":

+ e.rep = nil

+ case "uncertain-sidecar":

+ e.sidecarUncertain = true

+ }

+ proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), e.host, 1)

+ if scenario == "success" {

+ if err != nil || !proof.Matches(e.host, 1) || proof.Matches(e.host, 2) {

+ t.Fatal("checked stop missing", err)

+ }

+ } else if err != ErrStopUnproven || proof != nil {

+ t.Fatal("failed stop became evidence", err)

+ }

+ if a.fences[e.host] != 1 {

+ t.Fatal("failure removed fence")

+ }

+ if !a.Quiesced(e.host) {

+ t.Fatal("ordinary entry removal changed")

+ }

+ retry, retryErr := a.FenceAndStop(maintenanceContext(t, time.Second), e.host, 1)

+ if scenario != "success" && (retryErr != ErrStopUnproven || retry != nil) {

+ t.Fatal("deleted entry erased failure")

+ }

+ if main.calls.Load() > 1 || side.calls.Load() > 1 || rep.calls.Load() > 1 {

+ t.Fatal("replayed spent stop")

+ }

+ })

+ }

+func TestMaintenanceFenceBlocksEveryAdmissionIncludingStaleRoute(t *testing.T) {

+ a, _, _, reg := testActivator(t, time.Hour)

+ defer reg.Close()

+ host := "absent.example"

+ if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), host, 3); err != ErrStopUnproven || proof != nil {

+ t.Fatal("absence became stop proof", err)

+ }

+ rt := testRoute(host)

+ if _, err := a.beginRequest(t.Context(), rt); err != ErrFenced {

+ t.Fatal("member wake passed", err)

+ }

+ if err := a.BeginSidecarRequest(t.Context(), rt); err != ErrFenced {

+ t.Fatal("sidecar passed", err)

+ }

+ for _, scheduled := range []bool{false, true} {

+ for _, stale := range []bool{false, true} {

+ for _, sidecar := range []bool{false, true} {

+ route := rt

+ if stale {

+ route.Hibernate = false

+ route.Kind = "service"

+ }

+ req := httptest.NewRequest("GET", "https://"+host+"/", nil)

+ if sidecar {

+ req.Header.Set(SidecarHeader, "1")

+ }

+ out := httptest.NewRecorder()

+ proxied := false

+ proxy := http.HandlerFunc(func(http.ResponseWriter, *http.Request) { proxied = true })

+ var handled bool

+ if scheduled {

+ handled = a.ServeScheduled(out, req, route, proxy)

+ } else {

+ handled = a.ServeHTTP(out, req, route, proxy)

+ }

+ if !handled || proxied || out.Code != 503 || out.Header().Get("Cache-Control") != "no-store" {

+ t.Fatal("fence bypassed by stale or scheduled route")

+ }

+ if _, err := a.FenceAndStop(maintenanceContext(t, time.Second), host, 2); err != ErrFenceGeneration {

+ t.Fatal("older generation accepted")

+ }

+ // A new process has no closure evidence from the old activator's lifetime.

+ fresh := New(a.cfg, a.backend, reg)

+ if proof, err := fresh.FenceAndStop(maintenanceContext(t, time.Second), host, 3); err != ErrStopUnproven || proof != nil {

+ t.Fatal("restart invented old evidence")

+ }

+func TestMaintenanceFenceContextAndGenerationDoNotRestoreAdmission(t *testing.T) {

+ a, e, main, _, _ := maintenanceAwake(t)

+ gate := make(chan struct{})

+ main.gate = gate

+ main.entered = make(chan struct{})

+ first := make(chan error, 1)

+ go func() { _, err := a.FenceAndStop(maintenanceContext(t, time.Second), e.host, 1); first <- err }()

+ <-main.entered

+ if proof, err := a.FenceAndStop(maintenanceContext(t, 15*time.Millisecond), e.host, 2); err != ErrStopUnproven || proof != nil {

+ t.Fatal("timed-out stop claimed proof")

+ }

+ if _, err := a.beginRequest(t.Context(), e.route); err != ErrFenced {

+ t.Fatal("timeout reopened admission")

+ }

+ close(gate)

+ if err := <-first; err != ErrFenceGeneration {

+ t.Fatal("old waiter survived newer fence", err)

+ }

+ proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), e.host, 2)

+ if err != nil || !proof.Matches(e.host, 2) {

+ t.Fatal("checked outcome lost after timeout", err)

+ }

+ for _, ctx := range []context.Context{nil, context.Background(), maintenanceContext(t, -time.Second)} {

+ if p, err := a.FenceAndStop(ctx, e.host, 2); err == nil || p != nil {

+ t.Fatal("unbounded or expired stop accepted")

+ }

+ a.mu.Lock()

+ started := time.Now()

+ if _, err := a.FenceAndStop(maintenanceContext(t, 15*time.Millisecond), e.host, 2); err != ErrStopUnproven {

+ t.Fatal(err)

+ }

+ a.mu.Unlock()

+ if time.Since(started) > 250*time.Millisecond {

+ t.Fatal("fence mutex escaped deadline")

+ }

+func TestMaintenanceOrdinaryStopFailureRemainsObservable(t *testing.T) {

+ a, e, main, _, _ := maintenanceAwake(t)

+ main.err = errors.New("stop refused")

+ a.mu.Lock()

+ a.beginStopLocked(e)

+ done := e.stopped

+ a.mu.Unlock()

+ <-done

+ if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), e.host, 4); err != ErrStopUnproven || proof != nil {

+ t.Fatal("ordinary stop erased failure", err)

+ }

+func TestMaintenanceLaterOrdinaryStopCannotEraseUnresolvedCleanup(t *testing.T) {

+ a, e, main, _, _ := maintenanceAwake(t)

+ main.err = errors.New("earlier process may still run")

+ a.mu.Lock()

+ a.beginStopLocked(e)

+ done := e.stopped

+ a.mu.Unlock()

+ <-done

+ replacement := &actEntry{host: e.host, route: e.route, state: actAwake, proc: &maintenanceProc{}, sidecar: &maintenanceProc{}, rep: &maintenanceRep{}}

+ a.mu.Lock()

+ a.entries[e.host] = replacement

+ a.beginStopLocked(replacement)

+ done = replacement.stopped

+ a.mu.Unlock()

+ <-done

+ if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), e.host, 1); err != ErrStopUnproven || proof != nil {

+ t.Fatal("new entry erased unresolved predecessor", err)

+ }

+type maintenanceBackend struct {

+ *fakeBackend

+ main *maintenanceProc

+ side *maintenanceProc

+ rep *maintenanceRep

+ acquireGate <-chan struct{}

+ acquireEntered chan struct{}

+ replicateGate <-chan struct{}

+ replicateEntered chan struct{}

+ sideGate <-chan struct{}

+ sideEntered chan struct{}

+ startFailure error

+func (b *maintenanceBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, error) {

+ if b.acquireEntered != nil {

+ close(b.acquireEntered)

+ }

+ if b.acquireGate != nil {

+ <-b.acquireGate

+ }

+ lease, err := b.fakeBackend.Acquire(ctx, rt)

+ if err != nil {

+ return nil, err

+ }

+ return &maintenanceLease{Lease: lease, b: b}, nil

+func (b *maintenanceBackend) Start(context.Context, registry.Route) (Proc, string, error) {

+ b.mu.Lock()

+ b.starts++

+ b.mu.Unlock()

+ if b.startFailure != nil {

+ return nil, "", b.startFailure

+ }

+ return b.main, "", nil

+func (b *maintenanceBackend) StartSidecar(context.Context, registry.Route) (Proc, error) {

+ b.mu.Lock()

+ b.sidecarStarts++

+ b.mu.Unlock()

+ if b.sideEntered != nil {

+ close(b.sideEntered)

+ }

+ if b.sideGate != nil {

+ <-b.sideGate

+ }

+ return b.side, nil

+type maintenanceLease struct {

+ Lease

+ b *maintenanceBackend

+func (l *maintenanceLease) Replicate(context.Context, string, time.Duration) (Rep, error) {

+ if l.b.replicateEntered != nil {

+ close(l.b.replicateEntered)

+ }

+ if l.b.replicateGate != nil {

+ <-l.b.replicateGate

+ }

+ return l.b.rep, nil

+func TestMaintenanceFenceDuringWakeOwnsItsCleanup(t *testing.T) {

+ for _, phase := range []string{"acquire", "replicate"} {

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

+ a, base, _, reg := testActivator(t, time.Hour)

+ defer reg.Close()

+ gate, entered := make(chan struct{}), make(chan struct{})

+ b := &maintenanceBackend{fakeBackend: base, main: &maintenanceProc{}, rep: &maintenanceRep{}}

+ if phase == "acquire" {

+ b.acquireGate = gate

+ b.acquireEntered = entered

+ } else {

+ b.replicateGate = gate

+ b.replicateEntered = entered

+ }

+ a.backend = b

+ rt := testRoute("waking.example")

+ request := make(chan error, 1)

+ go func() {

+ release, err := a.beginRequest(t.Context(), rt)

+ if release != nil {

+ release()

+ }

+ request <- err

+ }()

+ <-entered

+ if proof, err := a.FenceAndStop(maintenanceContext(t, 15*time.Millisecond), rt.Host, 1); err != ErrStopUnproven || proof != nil {

+ t.Fatal("in-flight wake yielded proof", err)

+ }

+ close(gate)

+ if err := <-request; err != ErrFenced {

+ t.Fatal("waking request admitted after fence", err)

+ }

+ proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), rt.Host, 1)

+ if phase == "acquire" {

+ if err != ErrStopUnproven || proof != nil {

+ t.Fatal("canceled pre-spawn wake invented closure", err)

+ }

+ if b.starts != 0 || b.restores != 0 {

+ t.Fatal("fenced acquisition continued startup")

+ }

+ } else {

+ if err != nil || !proof.Matches(rt.Host, 1) {

+ t.Fatal("started wake did not yield checked cleanup", err)

+ }

+ if b.main.calls.Load() != 1 || b.rep.calls.Load() != 1 {

+ t.Fatal("partial wake resources not closed")

+ }

+ })

+ }

+func TestMaintenanceFailedBackendStartRemainsUnproven(t *testing.T) {

+ a, base, _, reg := testActivator(t, time.Hour)

+ defer reg.Close()

+ a.backend = &maintenanceBackend{fakeBackend: base, startFailure: errors.New("hidden backend cleanup failed")}

+ rt := testRoute("hidden.example")

+ if _, err := a.beginRequest(t.Context(), rt); err == nil {

+ t.Fatal("failed backend started")

+ }

+ if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), rt.Host, 1); err != ErrStopUnproven || proof != nil {

+ t.Fatal("hidden cleanup failure became proof", err)

+ }

+func TestMaintenanceFenceSerializesSidecarSweepReplacement(t *testing.T) {

+ a, e, _, side, rep := maintenanceAwake(t)

+ gate, entered := make(chan struct{}), make(chan struct{})

+ base := a.backend.(*fakeBackend)

+ base.sidecarStaged = true

+ next := &maintenanceProc{}

+ a.backend = &maintenanceBackend{fakeBackend: base, side: next, sideGate: gate, sideEntered: entered, rep: rep}

+ side.exited.Store(true)

+ e.sidecarWanted = true

+ sweep := make(chan struct{})

+ go func() { a.sweepOnce(); close(sweep) }()

+ <-entered

+ if proof, err := a.FenceAndStop(maintenanceContext(t, 15*time.Millisecond), e.host, 1); err != ErrStopUnproven || proof != nil {

+ t.Fatal("blocked replacement yielded proof", err)

+ }

+ close(gate)

+ <-sweep

+ if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), e.host, 1); err != nil || !proof.Matches(e.host, 1) {

+ t.Fatal("replacement escaped checked stop", err)

+ }

+ if next.calls.Load() != 1 {

+ t.Fatal("replacement sidecar was not stopped")

+ }

+ before := base.sidecarStartCount()

+ a.sweepOnce()

+ if base.sidecarStartCount() != before {

+ t.Fatal("fenced sweep restarted sidecar")

+ }

+type maintenanceDue struct {

+ DueStore

+ route registry.Route

+ entered chan struct{}

+ gate <-chan struct{}

+func (d *maintenanceDue) DueInstances(int64) ([]registry.Route, error) {

+ close(d.entered)

+ <-d.gate

+ return []registry.Route{d.route}, nil

+func TestMaintenanceFenceStopsStaleScheduledWake(t *testing.T) {

+ a, b, _, reg := testActivator(t, time.Hour)

+ defer reg.Close()

+ rt := testRoute("scheduled.example")

+ rt.Status = "ready"

+ gate, entered := make(chan struct{}), make(chan struct{})

+ a.due = &maintenanceDue{DueStore: reg, route: rt, entered: entered, gate: gate}

+ swept := make(chan struct{})

+ go func() { a.sweepOnce(); close(swept) }()

+ <-entered

+ if _, err := a.FenceAndStop(maintenanceContext(t, time.Second), rt.Host, 1); err != ErrStopUnproven {

+ t.Fatal(err)

+ }

+ close(gate)

+ <-swept

+ b.mu.Lock()

+ starts, acquires := b.starts, b.acquires

+ b.mu.Unlock()

+ if starts != 0 || acquires != 0 || a.entryCount() != 0 {

+ t.Fatal("stale due query crossed installed fence")

+ }

+func TestMaintenanceAwakeFenceSuppressesSidecarAdoption(t *testing.T) {

+ a, e, _, side, _ := maintenanceAwake(t)

+ base := a.backend.(*fakeBackend)

+ base.sidecarStaged = true

+ side.exited.Store(true)

+ a.fences = map[string]uint64{e.host: 1}

+ a.sweepOnce()

+ if base.sidecarStartCount() != 0 {

+ t.Fatal("awake fence allowed sidecar adoption")

+ }

+ if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), e.host, 1); err != nil || proof == nil {

+ t.Fatal(err)

+ }