amadan

carlos / platform Public

fix wave 1: a stopping instance keeps the lease until every sidecar it stopped or a restart returned has exited

The restart goroutine gives up its join only after its cleanup; an unproven returned or co-stopped sidecar (flag off) is kept on the entry and re-stopped by the pending-stop retry.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 70849825701b743987173c6bba16de771d96fe37 parent 6a81b13
4 files changed, +313 −24
  • internal/activator/activator.go +93 −18
  • internal/activator/activator_test.go +6 −1
  • internal/activator/opset.go +7 −5
  • internal/activator/stop_returned_sidecar_test.go +207 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 4240d6b1e..b7d07bc47 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -517,6 +517,13 @@ type actEntry struct {
// stop stays pending, with the lease, until it closes (Task 8, F6 plan
// 5), and only then is the host cgroup proven empty. Written under a.mu.
sidecarJoin chan struct{}
+ // unprovenSidecars are sidecars of this stopping entry, generations off,
+ // whose stop did not prove their exit: the co-stopped sidecar, or one a
+ // restart brought back after the stop began (fix wave 1, item 1). The
+ // lease is not released while one of them has not exited, whatever the
+ // layout: a nested row has no host-cgroup proof to reach them. The sweep
+ // re-stops them (reapCurrentGen). Written under a.mu.
+ unprovenSidecars []Proc
// launchLayout is the layout the wake launched this entry's child on
// (wokenInstance.launchLayout), set at install. A flag-off terminal stop
// decides from it whether the host cgroup must be proven empty
@@ -732,20 +739,63 @@ func stopDiscardedSidecar(sc Proc) {
}()
}
-// stopSidecarNow stops a sidecar a flag-off restart brought back for an
-// entry that is stopping (Task 8, F6 plan 5): synchronously, bounded by
-// sidecarStopTimeout, so the start's done channel closes only after it.
-// The stop's host proof follows it, and with a delegated cgroup that proof
-// also reaches whatever this stop left.
-func stopSidecarNow(host string, sc Proc) {
+// stopReturnedSidecar stops a sidecar a flag-off restart brought back for
+// an entry that is stopping (Task 8, F6 plan 5): synchronously, bounded by
+// sidecarStopTimeout, so the start's join closes only after it. A sidecar
+// whose stop did not prove its exit is not forgotten (fix wave 1, item 1):
+// it is kept on the entry, and the stop keeps its lease until it has exited
+// (sidecarsExited). The stop's host proof follows, and with a delegated
+// cgroup that proof also reaches whatever this stop left. Never called with
+// a.mu held.
+func (a *Service) stopReturnedSidecar(e *actEntry, sc Proc) {
if sc == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
- defer cancel()
- if err := sc.Stop(ctx); err != nil {
- log.Printf("activator: %s: stop the sidecar a restart brought back for a stopping entry: %v", host, err)
+ err := sc.Stop(ctx)
+ cancel()
+ if sc.Exited() {
+ return
}
+ log.Printf("activator: %s: the sidecar a restart brought back for a stopping entry has not exited (%v); keeping the lease until it has", e.host, err)
+ a.mu.Lock()
+ e.unprovenSidecars = append(e.unprovenSidecars, sc)
+ a.mu.Unlock()
+}
+
+// sidecarsExited reports whether every sidecar a flag-off stop kept
+// unproven (actEntry.unprovenSidecars) has exited: the sidecar's own stop
+// proof, which is the only one a nested row has. With restop it first
+// stops again, bounded, each one still running (the sweep's retry,
+// reapCurrentGen). Proven ones are forgotten. Never called with a.mu held.
+func (a *Service) sidecarsExited(e *actEntry, restop bool) bool {
+ a.mu.Lock()
+ kept := slices.Clone(e.unprovenSidecars)
+ a.mu.Unlock()
+ if len(kept) == 0 {
+ return true
+ }
+ var gone []Proc
+ for _, sc := range kept {
+ if restop && !sc.Exited() {
+ ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
+ if err := sc.Stop(ctx); err != nil && !errors.Is(err, ErrStillRunning) {
+ log.Printf("activator: %s: re-stop a sidecar not proven gone: %v", e.host, err)
+ }
+ cancel()
+ }
+ if sc.Exited() {
+ gone = append(gone, sc)
+ }
+ }
+ a.mu.Lock()
+ e.unprovenSidecars = slices.DeleteFunc(e.unprovenSidecars, func(p Proc) bool { return slices.Contains(gone, p) })
+ left := len(e.unprovenSidecars)
+ a.mu.Unlock()
+ if left > 0 {
+ log.Printf("activator: stop %s: %d sidecar(s) it stopped have not exited; keeping the lease until they have", e.host, left)
+ }
+ return left == 0
}
// discardSidecar is how a sidecar start that lost its entry is shed. With
@@ -3636,6 +3686,14 @@ func (a *Service) stop(e *actEntry) {
log.Printf("activator: stop sidecar %s: %v", e.host, err)
}
scCancel()
+ if !a.gens && !e.sidecar.Exited() {
+ // Flag off, its exit is kept to be proven before the lease goes
+ // (fix wave 1, item 1): a nested row has no host-cgroup proof to
+ // reach it.
+ a.mu.Lock()
+ e.unprovenSidecars = append(e.unprovenSidecars, e.sidecar)
+ a.mu.Unlock()
+ }
if a.gens {
// A sidecar is gone only when its leader exited AND its cgroup is
// empty (spec §7.4, §7.5). Unproven, it is tracked: the lease is
@@ -3760,6 +3818,15 @@ func (a *Service) stop(e *actEntry) {
a.mu.Unlock()
return
}
+ if !a.gens && !a.sidecarsExited(e, false) {
+ // A sidecar this stop stopped, or one a restart brought back, has
+ // not exited (fix wave 1, item 1): the stop stays pending with the
+ // lease, and the sweep re-stops it (reapCurrentGen).
+ a.mu.Lock()
+ a.keepPendingLocked(e, outcome)
+ a.mu.Unlock()
+ return
+ }
if !a.gens && !a.stopHostProven(e) {
// Flag off, the leader's exit says nothing of what it forked (Task
// 8, F6): the host cgroup is proven empty first. Unproven, the stop
@@ -4412,7 +4479,16 @@ func (a *Service) sweepOnce() {
var discard, stopNow Proc
defer func() { // after the unlock, before done closes
a.discardSidecar(route, discard)
- stopSidecarNow(host, stopNow)
+ a.stopReturnedSidecar(e0, stopNow)
+ // The join is given up only now, with the cleanup
+ // finished (fix wave 1, item 1): a stop that reads
+ // it while the sidecar this start brought back is
+ // still being stopped must wait for that stop.
+ a.mu.Lock()
+ if e0.sidecarStartDone == done {
+ e0.sidecarStarting, e0.sidecarStartDone = false, nil
+ }
+ a.mu.Unlock()
}()
a.mu.Lock()
defer a.mu.Unlock()
@@ -4428,14 +4504,13 @@ func (a *Service) sweepOnce() {
return
}
// Still e0: this attempt owns its sidecarStarting flag
- // and clears it (and the done channel already closing
- // via the defer above) unconditionally, regardless of
- // what state e0 is in now: a stop may have begun
- // (flipping e0 to actStopping) while the call was in
- // flight, and stop()'s own wait on sidecarStartDone
- // must not run the full sidecarStopTimeout for nothing.
- e0.sidecarStarting = false
- e0.sidecarStartDone = nil
+ // and join, and the deferred cleanup above clears them
+ // once it has stopped whatever must be stopped,
+ // regardless of what state e0 is in now: a stop may
+ // have begun (flipping e0 to actStopping) while the
+ // call was in flight. Cleared here, before the cleanup
+ // ran, a stop reading the join in between would miss
+ // the sidecar still being stopped (fix wave 1, item 1).
if e0.state != actAwake {
// Stopped mid-call, not yet forgotten: don't install
// onto a wake that's on its way out.
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index 9e53db5b1..ca59699ee 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -375,6 +375,11 @@ type fakeBackend struct {
// sidecarLabel, when set, labels the sidecars StartSidecar returns from
// then on ("sidecar" otherwise), so the order log can tell them apart.
sidecarLabel string
+ // sidecarStopGate and sidecarStuck, when set, are given to the sidecars
+ // StartSidecar returns from then on (fakeProc's stopGate and stuck): a
+ // returned sidecar whose stop blocks, or cannot prove its exit.
+ sidecarStopGate chan struct{}
+ sidecarStuck bool
// order interleaves backend calls and proc stops (start:instance,
// start:sidecar, nextDue, stop:sidecar, stop:instance) for the ordering
// assertions.
@@ -528,7 +533,7 @@ func (b *fakeBackend) StartSidecar(context.Context, registry.Route) (Proc, error
if b.sidecarLabel != "" {
label = b.sidecarLabel
}
- p := &fakeProc{label: label, rec: b}
+ p := &fakeProc{label: label, rec: b, stopGate: b.sidecarStopGate, stuck: b.sidecarStuck}
b.sidecars = append(b.sidecars, p)
return p, nil
}
diff --git a/internal/activator/opset.go b/internal/activator/opset.go
index 94e55bf39..d9ca3b9fe 100644
--- a/internal/activator/opset.go
+++ b/internal/activator/opset.go
@@ -682,12 +682,14 @@ func (a *Service) keepPendingLocked(e *actEntry, outcome *stopObservation) {
// reapCurrentGen finishes a pending stop whose main process has exited
// (finishPendingStops), once its termination proof holds: with generations
-// on, the retain-until-proven rule (releasable); with them off, the host
-// cgroup proven empty (stopHostProven, Task 8, F6). A failed proof leaves
-// the stop pending, and the next sweep tries again. The caller set
-// e.genReaping; this clears it. Never called with a.mu held.
+// on, the retain-until-proven rule (releasable); with them off, every
+// sidecar the stop kept unproven re-stopped and exited (sidecarsExited, fix
+// wave 1, item 1), then the host cgroup proven empty (stopHostProven, Task
+// 8, F6). A failed proof leaves the stop pending, and the next sweep tries
+// again. The caller set e.genReaping; this clears it. Never called with
+// a.mu held.
func (a *Service) reapCurrentGen(e *actEntry) {
- proven := a.stopHostProven
+ proven := func(e *actEntry) bool { return a.sidecarsExited(e, true) && a.stopHostProven(e) }
if a.gens {
proven = a.releasable
}
diff --git a/internal/activator/stop_returned_sidecar_test.go b/internal/activator/stop_returned_sidecar_test.go
new file mode 100644
index 000000000..904599c87
--- /dev/null
+++ b/internal/activator/stop_returned_sidecar_test.go
@@ -0,0 +1,207 @@
+package activator
+
+import (
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// stopEventGate is an EventSink that holds a terminal stop at its first
+// line ("instance stopping"), before it reads the sidecar join, until the
+// test lets it go: the stop's read of the join can then be placed after
+// anything the test arranges.
+type stopEventGate struct {
+ once sync.Once
+ entered chan struct{}
+ release chan struct{}
+}
+
+func newStopEventGate() *stopEventGate {
+ return &stopEventGate{entered: make(chan struct{}), release: make(chan struct{})}
+}
+
+func (g *stopEventGate) App(string, string, string, string) {}
+
+func (g *stopEventGate) Event(_, _, _, _, msg string) {
+ if msg != "instance stopping" {
+ return
+ }
+ first := false
+ g.once.Do(func() { first = true })
+ if !first {
+ return
+ }
+ close(g.entered)
+ <-g.release
+}
+
+// shrinkSidecarJoinWait shortens a flag-off stop's bounded wait on an
+// in-flight sidecar start for one test.
+func shrinkSidecarJoinWait(t *testing.T) {
+ t.Helper()
+ prev := sidecarStartJoinWait
+ sidecarStartJoinWait = 50 * time.Millisecond
+ t.Cleanup(func() { sidecarStartJoinWait = prev })
+}
+
+// Fix wave 1, item 1 (Astra final review 1, finding 1): the missed-join
+// window. A sidecar restart returns while its entry is stopping, and the
+// stop reads the join only after the restart's critical section has run
+// but while its cleanup is still stopping the sidecar it brought back. The
+// stop must still see the join, and the lease goes only once that sidecar
+// is stopped.
+func TestAStopThatReadsTheJoinDuringTheRestartsCleanupWaitsForIt(t *testing.T) {
+ shrinkSidecarJoinWait(t)
+ a, b, _, _ := testActivator(t, time.Hour)
+ h := &hostEmptyFake{fakeBackend: b}
+ a.backend = h
+ g := newStopEventGate()
+ a.Logs = g
+ b.sidecarStaged = true
+ host := "missed-join.example.com"
+
+ arrangeAwake(t, a, testRoute(host))
+ h.reset()
+ b.lastSidecar().exit() // the sweeper now wants a restart
+
+ startGate, entered := make(chan struct{}), make(chan struct{})
+ scStopGate := make(chan struct{})
+ b.mu.Lock()
+ b.sidecarGate, b.sidecarStartEntered = startGate, entered
+ b.sidecarStopGate = scStopGate // the returned sidecar's stop blocks
+ b.mu.Unlock()
+ a.sweepOnce()
+ <-entered
+
+ rep := b.lastRep()
+ if !a.Bounce(host) {
+ t.Fatal("Bounce did not find the awake entry")
+ }
+ <-g.entered // the stop is running, and has not read the join yet
+
+ close(startGate) // the restart returns onto a stopping entry
+ waitFor(t, "the restart to bring back its sidecar", func() bool { return b.sidecarStartCount() == 2 })
+ sc := b.lastSidecar()
+ rep.setOnClose(func() {
+ if !sc.isStopped() {
+ t.Error("the lease was released while the sidecar the restart brought back was still being stopped")
+ }
+ })
+ waitFor(t, "the restart's cleanup to begin stopping it", func() bool { return sc.stopEntered() >= 1 })
+ close(g.release) // now the stop reads the join
+
+ time.Sleep(4 * sidecarStartJoinWait)
+ for range 3 {
+ a.sweepOnce()
+ }
+ if rep.isClosed() {
+ t.Fatal("the lease went while the returned sidecar's stop was still in flight")
+ }
+
+ close(scStopGate)
+ waitFor(t, "the stop to finish once the returned sidecar is stopped", func() bool {
+ a.sweepOnce()
+ return rep.isClosed() && a.entryCount() == 0
+ })
+}
+
+// Fix wave 1, item 1: a nested instance's restart returns a sidecar whose
+// stop cannot prove its exit (ErrStillRunning). Nested rows skip the
+// host-cgroup proof, so the sidecar's own stop proof is the only one: the
+// stop stays pending with the lease, the sweep re-stops the sidecar, and
+// the lease goes only after it has exited.
+func TestANestedStopKeepsTheLeaseOverAReturnedSidecarItCannotProve(t *testing.T) {
+ shrinkSidecarJoinWait(t)
+ a, b, _, _ := testActivator(t, time.Hour)
+ h := &hostEmptyFake{fakeBackend: b}
+ a.backend = h
+ b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
+ b.sidecarStaged = true
+ host := "nested-stuck.example.com"
+
+ arrangeAwake(t, a, testRoute(host))
+ h.reset()
+ b.lastSidecar().exit()
+
+ startGate, entered := make(chan struct{}), make(chan struct{})
+ b.mu.Lock()
+ b.sidecarGate, b.sidecarStartEntered = startGate, entered
+ b.sidecarStuck = true // the returned sidecar's stop answers ErrStillRunning
+ b.mu.Unlock()
+ a.sweepOnce()
+ <-entered
+
+ rep, inst := b.lastRep(), b.lastProc()
+ if !a.Bounce(host) {
+ t.Fatal("Bounce did not find the awake entry")
+ }
+ waitFor(t, "the instance to stop", inst.isStopped)
+ close(startGate)
+ waitFor(t, "the restart to bring back its sidecar", func() bool { return b.sidecarStartCount() == 2 })
+ sc := b.lastSidecar()
+ rep.setOnClose(func() {
+ if !sc.Exited() {
+ t.Error("the lease was released while the returned sidecar had not exited")
+ }
+ })
+ waitFor(t, "the restart's stop of it", func() bool { return sc.stopCalls() >= 1 })
+ time.Sleep(4 * sidecarStartJoinWait)
+ for range 3 {
+ a.sweepOnce()
+ }
+ if rep.isClosed() || a.entryCount() != 1 {
+ t.Fatalf("closed=%v entries=%d: the lease went over a sidecar whose stop was unproven", rep.isClosed(), a.entryCount())
+ }
+ waitFor(t, "the sweep to re-stop the unproven sidecar", func() bool {
+ a.sweepOnce()
+ return sc.stopCalls() >= 2
+ })
+
+ sc.exit()
+ waitFor(t, "the stop to finish once the sidecar exited", func() bool {
+ a.sweepOnce()
+ return rep.isClosed() && a.entryCount() == 0
+ })
+ if n := h.count(); n != 0 {
+ t.Fatalf("host proofs = %d: a nested stop has no host-cgroup proof", n)
+ }
+}
+
+// Fix wave 1, item 1: the co-stop's own sidecar, flag off and nested, whose
+// stop cannot prove its exit, is held to the same rule: the lease is kept
+// until it has exited.
+func TestANestedStopKeepsTheLeaseOverItsCoStoppedSidecarUntilItExits(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ h := &hostEmptyFake{fakeBackend: b}
+ a.backend = h
+ b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
+ b.sidecarStaged = true
+ b.sidecarStuck = true
+ host := "nested-costop.example.com"
+
+ arrangeAwake(t, a, testRoute(host))
+ sc := b.lastSidecar()
+ rep, inst := b.lastRep(), b.lastProc()
+ rep.setOnClose(func() {
+ if !sc.Exited() {
+ t.Error("the lease was released while the co-stopped sidecar had not exited")
+ }
+ })
+ if !a.Bounce(host) {
+ t.Fatal("Bounce did not find the awake entry")
+ }
+ waitFor(t, "the instance to stop", inst.isStopped)
+ for range 3 {
+ a.sweepOnce()
+ }
+ if rep.isClosed() || a.entryCount() != 1 {
+ t.Fatalf("closed=%v entries=%d: the lease went over a sidecar whose stop was unproven", rep.isClosed(), a.entryCount())
+ }
+ sc.exit()
+ waitFor(t, "the stop to finish once the sidecar exited", func() bool {
+ a.sweepOnce()
+ return rep.isClosed() && a.entryCount() == 0
+ })
+}