amadan

carlos / platform Public

Merge branch 'fix-maintenance-admission-ordering' into main

claude 12e9c80928d4014e9ab0550806cf02b6efa27ff9 parent caf60d1parent 4b8e76a
6 files changed, +145 −24
  • internal/activator/activator.go +5 −4
  • internal/activator/admission_ordering_test.go +70 −0
  • internal/activator/maintenance.go +30 −8
  • internal/activator/maintenance_test.go +5 −2
  • internal/activator/scheduled_token.go +28 −9
  • internal/activator/scheduled_token_test.go +7 −1

CI passed — run details

diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index f8c892862..fac275fa8 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -399,10 +399,11 @@ 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
- fences map[string]uint64
- stopHistory map[string]*stopObservation
+ mu sync.Mutex
+ entries map[string]*actEntry
+ fences map[string]uint64
+ tokenPreparations map[string]chan struct{}
+ stopHistory map[string]*stopObservation
sweepStop chan struct{}
sweepDone chan struct{}
diff --git a/internal/activator/admission_ordering_test.go b/internal/activator/admission_ordering_test.go
new file mode 100644
index 000000000..5a4cf7da5
--- /dev/null
+++ b/internal/activator/admission_ordering_test.go
@@ -0,0 +1,70 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "testing"
+ "time"
+)
+
+func TestFenceInitialAdmissionTimeoutHasNoFence(t *testing.T) {
+ a := &Service{}
+ a.mu.Lock()
+ ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+ proof, err := a.FenceAndStop(ctx, "timeout.example.com", 1)
+ cancel()
+ if proof != nil || !errors.Is(err, ErrFenceNotInstalled) {
+ a.mu.Unlock()
+ t.Fatalf("uninstalled fence reported as installed: proof=%v err=%v", proof != nil, err)
+ }
+ if a.fences["timeout.example.com"] != 0 {
+ a.mu.Unlock()
+ t.Fatal("fence installed while mutex held")
+ }
+ a.mu.Unlock()
+ if !a.wakeAllowed("timeout.example.com") {
+ t.Fatal("unexpected admission state")
+ }
+}
+
+func TestScheduledTokenSlowHostDoesNotBlockSiblingFence(t *testing.T) {
+ a := &Service{}
+ entered, release := make(chan struct{}), make(chan struct{})
+ done := make(chan error, 1)
+ go func() {
+ _, e := a.PrepareScheduledToken("slow.example.com", func() (string, error) { close(entered); <-release; return "token", nil })
+ done <- e
+ }()
+ <-entered
+ defer func() { close(release); <-done }()
+ ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+ defer cancel()
+ proof, err := a.FenceAndStop(ctx, "sibling.example.com", 1)
+ if proof != nil || !errors.Is(err, ErrStopUnproven) {
+ t.Fatalf("sibling fence blocked by unrelated IO: %v", err)
+ }
+ callback := make(chan error, 1)
+ go func() {
+ _, e := a.PrepareScheduledToken("fast.example.com", func() (string, error) { return "fast", nil })
+ callback <- e
+ }()
+ select {
+ case e := <-callback:
+ if e != nil {
+ t.Fatal(e)
+ }
+ case <-time.After(time.Second):
+ t.Fatal("sibling token callback blocked")
+ }
+ // A separate goroutine detects the former global mutex hold without hanging the test.
+ result := make(chan bool, 1)
+ go func() { result <- a.wakeAllowed("third.example.com") }()
+ select {
+ case allowed := <-result:
+ if !allowed {
+ t.Fatal("unrelated admission refused")
+ }
+ case <-time.After(time.Second):
+ t.Fatal("unrelated admission blocked")
+ }
+}
diff --git a/internal/activator/maintenance.go b/internal/activator/maintenance.go
index 37c804c81..7717dd27e 100644
--- a/internal/activator/maintenance.go
+++ b/internal/activator/maintenance.go
@@ -11,9 +11,10 @@ import (
)
var (
- ErrFenced = errors.New("activator: maintenance fence active")
- ErrFenceGeneration = errors.New("activator: maintenance generation refused")
- ErrStopUnproven = errors.New("activator: stop evidence unavailable")
+ ErrFenceNotInstalled = errors.New("activator: maintenance fence not installed")
+ 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,
@@ -40,22 +41,42 @@ 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.
+// FenceAndStop installs a local fence before collecting stop evidence. Once this
+// call installs its fence, cancellation or failure leaves admission closed.
+// ErrFenceNotInstalled means this call could not install or confirm its fence;
+// it makes no claim about admission, including a fence from an earlier call.
+// ErrStopUnproven guarantees this call installed a fence, but proves no stop.
+// A missing entry is not evidence that a prior process or replica stopped.
func (a *Service) FenceAndStop(ctx context.Context, host string, generation uint64) (*StopEvidence, error) {
if a == nil || !controlname.ValidHostShape(host) || generation == 0 {
return nil, ErrFenceGeneration
}
if ctx == nil {
- return nil, ErrStopUnproven
+ return nil, ErrFenceNotInstalled
}
deadline, ok := ctx.Deadline()
if !ok || ctx.Err() != nil || !time.Now().Before(deadline) {
- return nil, ErrStopUnproven
+ return nil, ErrFenceNotInstalled
}
+ installed := false
for {
if err := a.lockFence(ctx); err != nil {
- return nil, err
+ if installed {
+ return nil, ErrStopUnproven
+ }
+ return nil, ErrFenceNotInstalled
+ }
+ if preparation := a.tokenPreparations[host]; preparation != nil {
+ a.mu.Unlock()
+ select {
+ case <-ctx.Done():
+ if installed {
+ return nil, ErrStopUnproven
+ }
+ return nil, ErrFenceNotInstalled
+ case <-preparation:
+ continue
+ }
}
if a.fences == nil {
a.fences = make(map[string]uint64)
@@ -65,6 +86,7 @@ func (a *Service) FenceAndStop(ctx context.Context, host string, generation uint
return nil, ErrFenceGeneration
}
a.fences[host] = generation
+ installed = true
e := a.entries[host]
if e == nil {
outcome := a.stopHistory[host]
diff --git a/internal/activator/maintenance_test.go b/internal/activator/maintenance_test.go
index 603fbfc35..e90d7628f 100644
--- a/internal/activator/maintenance_test.go
+++ b/internal/activator/maintenance_test.go
@@ -203,7 +203,7 @@ func TestMaintenanceFenceContextAndGenerationDoNotRestoreAdmission(t *testing.T)
}
a.mu.Lock()
started := time.Now()
- if _, err := a.FenceAndStop(maintenanceContext(t, 15*time.Millisecond), e.host, 2); err != ErrStopUnproven {
+ if _, err := a.FenceAndStop(maintenanceContext(t, 15*time.Millisecond), e.host, 2); err != ErrFenceNotInstalled {
t.Fatal(err)
}
a.mu.Unlock()
@@ -391,11 +391,14 @@ func TestMaintenanceFenceSerializesSidecarSweepReplacement(t *testing.T) {
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 {
+ if proof, err := a.FenceAndStop(maintenanceContext(t, 15*time.Millisecond), e.host, 1); err != ErrFenceNotInstalled || proof != nil {
t.Fatal("blocked replacement yielded proof", err)
}
close(gate)
<-sweep
+ if !a.wakeAllowed(e.host) {
+ t.Fatal("timed-out admission attempt installed a fence")
+ }
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)
}
diff --git a/internal/activator/scheduled_token.go b/internal/activator/scheduled_token.go
index 0074b6320..7f1b133d4 100644
--- a/internal/activator/scheduled_token.go
+++ b/internal/activator/scheduled_token.go
@@ -1,17 +1,36 @@
package activator
-// PrepareScheduledToken serializes token preparation with the local maintenance
-// fence. A preparation already in progress finishes before the fence is installed;
-// a fenced host never enters prepare. The callback must not reenter Service.
-// This does not authorize a wake or establish durable maintenance authority.
+// PrepareScheduledToken orders token preparation with this host's local fence.
+// An admitted callback finishes before its fence is installed. Other hosts do
+// not wait for its IO. The callback must not reenter preparation or fencing for
+// the same host. This establishes neither wake nor durable maintenance authority.
func (a *Service) PrepareScheduledToken(host string, prepare func() (string, error)) (string, error) {
if a == nil || prepare == nil {
return "", ErrFenced
}
- a.mu.Lock()
- defer a.mu.Unlock()
- if a.fences[host] != 0 {
- return "", ErrFenced
+ for {
+ a.mu.Lock()
+ if a.fences[host] != 0 {
+ a.mu.Unlock()
+ return "", ErrFenced
+ }
+ if pending := a.tokenPreparations[host]; pending != nil {
+ a.mu.Unlock()
+ <-pending
+ continue
+ }
+ if a.tokenPreparations == nil {
+ a.tokenPreparations = make(map[string]chan struct{})
+ }
+ done := make(chan struct{})
+ a.tokenPreparations[host] = done
+ a.mu.Unlock()
+ defer func() {
+ a.mu.Lock()
+ delete(a.tokenPreparations, host)
+ close(done)
+ a.mu.Unlock()
+ }()
+ return prepare()
}
- return prepare()
}
diff --git a/internal/activator/scheduled_token_test.go b/internal/activator/scheduled_token_test.go
index ee2841e43..0acea3a8b 100644
--- a/internal/activator/scheduled_token_test.go
+++ b/internal/activator/scheduled_token_test.go
@@ -54,8 +54,14 @@ func TestScheduledTokenPreparationOrdersFence(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
proof, err := a.FenceAndStop(ctx, host, 1)
cancel()
+ a.mu.Lock()
+ installed := a.fences[host]
+ a.mu.Unlock()
close(release)
- if proof != nil || !errors.Is(err, ErrStopUnproven) {
+ if installed != 0 {
+ t.Fatal("fence installed before admitted IO finished")
+ }
+ if proof != nil || !errors.Is(err, ErrFenceNotInstalled) {
t.Fatalf("fence crossed active preparation: proof=%v err=%v", proof != nil, err)
}
select {