amadan

carlos / platform Public

zdd plan 3 task 5: canary settle; an intentional stop cancels it and is not a failure

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude e0daa4e8bf01328e4614fd2a18288b6494b2ceca parent 06de1aa
3 files changed, +209 −0
  • internal/activator/settle.go +58 −0
  • internal/activator/settle_test.go +137 −0
  • internal/activator/swap.go +14 −0
diff --git a/internal/activator/settle.go b/internal/activator/settle.go
new file mode 100644
index 000000000..03d559330
--- /dev/null
+++ b/internal/activator/settle.go
@@ -0,0 +1,58 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "time"
+)
+
+// swapSettle is the canary's settle (spec §6.2, §7.5 step 4). A successor
+// that passed its health check must then stay up this long before the
+// switch. A var so tests can shrink it, like spawnHealthTimeout.
+var swapSettle = 10 * time.Second
+
+var errSettleExit = errors.New("swap: successor exited during its settle")
+
+type settleResult int
+
+const (
+ settleOK settleResult = iota // stayed up for the whole window
+ settleExited // died on its own: a failure of the build
+ settleCancelled // an intentional stop ended the observation
+)
+
+// observeSettle watches p for window after it became healthy (spec §6.2,
+// §6.4). Cancellation wins over an exit seen at the same moment. The
+// barrier cancels an attempt BEFORE anything is stopped, so a process that
+// dies once a stop has begun died of the stop, not of its build.
+func observeSettle(ctx context.Context, p Proc, window time.Duration) settleResult {
+ verdict := func() (settleResult, bool) {
+ if ctx.Err() != nil {
+ return settleCancelled, true
+ }
+ if p.Exited() {
+ return settleExited, true
+ }
+ return settleOK, false
+ }
+ if r, over := verdict(); over || window <= 0 {
+ return r
+ }
+ deadline := time.NewTimer(window)
+ defer deadline.Stop()
+ tick := time.NewTicker(settlePoll(window))
+ defer tick.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return settleCancelled
+ case <-tick.C:
+ if r, over := verdict(); over {
+ return r
+ }
+ case <-deadline.C:
+ r, _ := verdict()
+ return r
+ }
+ }
+}
diff --git a/internal/activator/settle_test.go b/internal/activator/settle_test.go
new file mode 100644
index 000000000..b70505e95
--- /dev/null
+++ b/internal/activator/settle_test.go
@@ -0,0 +1,137 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestObserveSettle(t *testing.T) {
+ if r := observeSettle(context.Background(), &fakeProc{}, 20*time.Millisecond); r != settleOK {
+ t.Fatalf("a process that stayed up: %v", r)
+ }
+ p := &fakeProc{}
+ go func() { time.Sleep(5 * time.Millisecond); p.exit() }()
+ if r := observeSettle(context.Background(), p, time.Second); r != settleExited {
+ t.Fatalf("a process that died: %v", r)
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ go func() { time.Sleep(5 * time.Millisecond); cancel() }()
+ if r := observeSettle(ctx, &fakeProc{}, time.Second); r != settleCancelled {
+ t.Fatalf("a cancelled observation: %v", r)
+ }
+ // Cancellation wins over an exit seen at the same moment (spec §6.4).
+ dead := &fakeProc{}
+ dead.exit()
+ gone, stop := context.WithCancel(context.Background())
+ stop()
+ if r := observeSettle(gone, dead, time.Second); r != settleCancelled {
+ t.Fatalf("cancelled and exited: %v, want cancelled", r)
+ }
+}
+
+func TestSwapCanarySettle(t *testing.T) {
+ old := swapSettle
+ t.Cleanup(func() { swapSettle = old })
+
+ t.Run("exit during the settle fails the build", func(t *testing.T) {
+ swapSettle = 2 * time.Second
+ f := newSwapFixture(t)
+ f.a.genHealth = func(context.Context, Gen) error {
+ go func() { time.Sleep(20 * time.Millisecond); _, p := f.b.lastGen(); p.exit() }()
+ return nil
+ }
+ at := f.attempt(ModeOverlap)
+ at.Canary = true
+ res := f.a.Swap(context.Background(), at, f.newest)
+ if res.Outcome != SwapFailed || !strings.Contains(res.Reason, errSettleExit.Error()) {
+ t.Fatalf("Swap = %+v", res)
+ }
+ if got := f.appSocket(t); got != f.incumbent.Socket {
+ t.Fatalf("AppSocket = %q, want the incumbent", got)
+ }
+ })
+ t.Run("a canary waits out the settle", func(t *testing.T) {
+ swapSettle = 100 * time.Millisecond
+ f := newSwapFixture(t)
+ at := f.attempt(ModeOverlap)
+ at.Canary = true
+ start := time.Now()
+ if res := f.a.Swap(context.Background(), at, f.newest); res.Outcome != SwapDone {
+ t.Fatalf("Swap = %+v", res)
+ }
+ if took := time.Since(start); took < swapSettle {
+ t.Fatalf("switched after %s, inside the %s settle", took, swapSettle)
+ }
+ })
+ t.Run("a swap after the canary does not settle", func(t *testing.T) {
+ swapSettle = time.Hour
+ f := newSwapFixture(t)
+ done := make(chan SwapResult, 1)
+ go func() { done <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }()
+ select {
+ case res := <-done:
+ if res.Outcome != SwapDone {
+ t.Fatalf("Swap = %+v", res)
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("a non-canary swap waited for a settle")
+ }
+ })
+}
+
+// An intentional stop of the entry during the settle cancels the
+// observation and is not a failure (spec §6.4). The successor is stopped
+// before the incumbent, through the barrier.
+func TestSwapIntentionalStopDuringSettleIsNotAFailure(t *testing.T) {
+ old := swapSettle
+ swapSettle = 5 * time.Second
+ t.Cleanup(func() { swapSettle = old })
+ for _, tc := range []struct {
+ name string
+ stop func(f *swapFixture)
+ }{
+ {"restart", func(f *swapFixture) { f.a.Bounce(f.host) }},
+ {"hibernate", func(f *swapFixture) { f.clk.advance(2 * time.Hour); f.a.sweepOnce() }},
+ {"lease lost", func(f *swapFixture) { f.b.lastRep().setErr(errors.New("renewal failed")); f.a.sweepOnce() }},
+ {"maintenance fence", func(f *swapFixture) {
+ go func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ f.a.FenceAndStop(ctx, f.host, 1) //nolint:errcheck
+ }()
+ }},
+ {"move or park retire", func(f *swapFixture) { f.a.RetireExcept(map[string]bool{}) }},
+ {"shutdown", func(f *swapFixture) { go f.a.Shutdown(context.Background()) }},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ f := newSwapFixture(t)
+ healthy := make(chan struct{})
+ f.a.genHealth = func(context.Context, Gen) error { close(healthy); return nil }
+ rep := f.b.lastRep()
+ at := f.attempt(ModeOverlap)
+ at.Canary = true
+ results := make(chan SwapResult, 1)
+ go func() { results <- f.a.Swap(context.Background(), at, f.newest) }()
+ <-healthy
+ time.Sleep(30 * time.Millisecond)
+ tc.stop(f)
+ var res SwapResult
+ select {
+ case res = <-results:
+ case <-time.After(10 * time.Second):
+ t.Fatal("Swap did not return after the stop")
+ }
+ if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapStopped.Error()) {
+ t.Fatalf("Swap = %+v, want abandoned by the stop", res)
+ }
+ waitFor(t, "the lease release", rep.isClosed)
+ ev := f.b.events()
+ if i, j := indexPrefix(ev, "stop:gen:"), indexPrefix(ev, "stop:instance"); i < 0 || j < 0 || i > j {
+ t.Fatalf("order %v: the successor must stop before the incumbent", ev)
+ }
+ })
+ }
+}
diff --git a/internal/activator/swap.go b/internal/activator/swap.go
index 75536c291..e5ddde3b7 100644
--- a/internal/activator/swap.go
+++ b/internal/activator/swap.go
@@ -428,6 +428,20 @@ func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, n
return failSwap(fmt.Errorf("health: %w", err))
}
+ // Step 4, the canary's half: the 10 s settle (spec §6.2). An
+ // intentional stop of the entry cancels the observation through the
+ // barrier, and that is not a failure (spec §6.4).
+ if at.Canary {
+ switch observeSettle(mctx, m.proc, swapSettle) {
+ case settleCancelled:
+ a.dropMember(e, m)
+ return abandon(errSwapStopped)
+ case settleExited:
+ a.dropMember(e, m)
+ return failSwap(errSettleExit)
+ }
+ }
+
// Step 5: the switch. The checks run again, then, under the entry lock,
// AppSocket, the serving build and the cache generation all move with
// e.gen (contract C2).