| index ae4533382..603fbfc35 100644 |
| --- a/internal/activator/maintenance_test.go |
| +++ b/internal/activator/maintenance_test.go |
| @@ -3,6 +3,7 @@ package activator |
| import ( |
| "context" |
| "errors" |
| + "fmt" |
| "github.com/carlosframework/platform/internal/edge" |
| "github.com/carlosframework/platform/internal/registry" |
| "net/http" |
| @@ -40,12 +41,26 @@ func (p *maintenanceProc) Stop(ctx context.Context) error { |
| func (p *maintenanceProc) Exited() bool { return p.exited.Load() } |
| |
| type maintenanceRep struct { |
| - calls atomic.Int32 |
| - err error |
| + gate <-chan struct{} |
| + entered chan 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 (r *maintenanceRep) Err() error { return nil } |
| +func (r *maintenanceRep) Close(ctx context.Context) error { |
| + if r.calls.Add(1) == 1 && r.entered != nil { |
| + close(r.entered) |
| + } |
| + if r.gate != nil { |
| + select { |
| + case <-r.gate: |
| + case <-ctx.Done(): |
| + return ctx.Err() |
| + } |
| + } |
| + return r.err |
| +} |
| func maintenanceContext(t *testing.T, duration time.Duration) context.Context { |
| t.Helper() |
| ctx, cancel := context.WithTimeout(t.Context(), duration) |
| @@ -235,12 +250,15 @@ type maintenanceBackend struct { |
| side *maintenanceProc |
| rep *maintenanceRep |
| acquireGate <-chan struct{} |
| + acquireFailure error |
| acquireEntered chan struct{} |
| replicateGate <-chan struct{} |
| replicateEntered chan struct{} |
| sideGate <-chan struct{} |
| sideEntered chan struct{} |
| startFailure error |
| + sideFailure error |
| + replicateFailure error |
| } |
| |
| func (b *maintenanceBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, error) { |
| @@ -254,14 +272,17 @@ func (b *maintenanceBackend) Acquire(ctx context.Context, rt registry.Route) (Le |
| if err != nil { |
| return nil, err |
| } |
| - return &maintenanceLease{Lease: lease, b: b}, nil |
| + return &maintenanceLease{Lease: lease, b: b}, b.acquireFailure |
| } |
| 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 |
| + if b.main == nil { |
| + return nil, "", b.startFailure |
| + } |
| + return b.main, "", b.startFailure |
| } |
| return b.main, "", nil |
| } |
| @@ -275,7 +296,7 @@ func (b *maintenanceBackend) StartSidecar(context.Context, registry.Route) (Proc |
| if b.sideGate != nil { |
| <-b.sideGate |
| } |
| - return b.side, nil |
| + return b.side, b.sideFailure |
| } |
| |
| type maintenanceLease struct { |
| @@ -290,7 +311,10 @@ func (l *maintenanceLease) Replicate(context.Context, string, time.Duration) (Re |
| if l.b.replicateGate != nil { |
| <-l.b.replicateGate |
| } |
| - return l.b.rep, nil |
| + if l.b.rep == nil { |
| + return nil, l.b.replicateFailure |
| + } |
| + return l.b.rep, l.b.replicateFailure |
| } |
| func TestMaintenanceFenceDuringWakeOwnsItsCleanup(t *testing.T) { |
| for _, phase := range []string{"acquire", "replicate"} { |
| @@ -433,3 +457,159 @@ func TestMaintenanceAwakeFenceSuppressesSidecarAdoption(t *testing.T) { |
| t.Fatal(err) |
| } |
| } |
| + |
| +func TestMaintenanceFailedStartCleansReturnedHandleBeforeRelease(t *testing.T) { |
| + a, base, _, reg := testActivator(t, time.Hour) |
| + defer reg.Close() |
| + gate, entered := make(chan struct{}), make(chan struct{}) |
| + proc := &maintenanceProc{gate: gate, entered: entered} |
| + a.backend = &maintenanceBackend{fakeBackend: base, main: proc, startFailure: errors.New("started then failed")} |
| + rt := testRoute("partial.example") |
| + done := make(chan error, 1) |
| + go func() { _, err := a.beginRequest(t.Context(), rt); done <- err }() |
| + select { |
| + case <-entered: |
| + case <-time.After(3 * time.Second): |
| + t.Fatal("returned process cleanup was never attempted") |
| + } |
| + base.mu.Lock() |
| + released := base.releases |
| + base.mu.Unlock() |
| + if released != 0 { |
| + t.Fatal("lease released before returned process cleanup") |
| + } |
| + close(gate) |
| + if err := <-done; err == nil { |
| + t.Fatal("partial failed start admitted") |
| + } |
| + if proc.calls.Load() != 1 || !proc.Exited() { |
| + t.Fatal("returned process handle was abandoned") |
| + } |
| + base.mu.Lock() |
| + released = base.releases |
| + base.mu.Unlock() |
| + if released != 1 { |
| + t.Fatal("lease release missing or repeated", released) |
| + } |
| + if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), rt.Host, 1); err != ErrStopUnproven || proof != nil { |
| + t.Fatal("partial failure erased uncertainty", err) |
| + } |
| +} |
| +func TestMaintenanceFailedSidecarHandlesRemainOwned(t *testing.T) { |
| + for _, phase := range []string{"wake", "sweep"} { |
| + t.Run(phase, func(t *testing.T) { |
| + a, base, _, reg := testActivator(t, time.Hour) |
| + defer reg.Close() |
| + base.sidecarStaged = true |
| + main, partial, rep := &maintenanceProc{}, &maintenanceProc{}, &maintenanceRep{} |
| + b := &maintenanceBackend{fakeBackend: base, main: main, side: partial, rep: rep, sideFailure: errors.New("sidecar started then failed")} |
| + a.backend = b |
| + rt := testRoute("partial-sidecar.example") |
| + if phase == "wake" { |
| + release, err := a.beginRequest(t.Context(), rt) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + release() |
| + } else { |
| + old := &maintenanceProc{} |
| + old.exited.Store(true) |
| + a.entries[rt.Host] = &actEntry{host: rt.Host, route: rt, state: actAwake, proc: main, rep: rep, sidecar: old, sidecarWanted: true, lastActive: a.now()} |
| + a.sweepOnce() |
| + } |
| + a.mu.Lock() |
| + entry := a.entries[rt.Host] |
| + owned := entry.sidecar == partial && entry.sidecarUncertain |
| + a.mu.Unlock() |
| + if !owned { |
| + t.Fatal("failed sidecar handle was discarded") |
| + } |
| + before := base.sidecarStartCount() |
| + a.sweepOnce() |
| + if base.sidecarStartCount() != before { |
| + t.Fatal("live partial sidecar was replaced") |
| + } |
| + if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), rt.Host, 1); err != ErrStopUnproven || proof != nil { |
| + t.Fatal("failed sidecar yielded trusted evidence", err) |
| + } |
| + if partial.calls.Load() != 1 || !partial.Exited() { |
| + t.Fatal("checked stop did not own partial sidecar") |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestMaintenanceFailedReplicationClosesReturnedReplicaWithoutDoubleRelease(t *testing.T) { |
| + for _, closeFails := range []bool{false, true} { |
| + t.Run(fmt.Sprint(closeFails), func(t *testing.T) { |
| + a, base, _, reg := testActivator(t, time.Hour) |
| + defer reg.Close() |
| + mainGate, mainEntered := make(chan struct{}), make(chan struct{}) |
| + repGate, repEntered := make(chan struct{}), make(chan struct{}) |
| + main := &maintenanceProc{gate: mainGate, entered: mainEntered} |
| + rep := &maintenanceRep{gate: repGate, entered: repEntered} |
| + if closeFails { |
| + rep.err = errors.New("replica close failed") |
| + } |
| + a.backend = &maintenanceBackend{fakeBackend: base, main: main, rep: rep, replicateFailure: errors.New("replica opened then failed")} |
| + rt := testRoute("partial-replica.example") |
| + done := make(chan error, 1) |
| + go func() { _, err := a.beginRequest(t.Context(), rt); done <- err }() |
| + select { |
| + case <-mainEntered: |
| + case <-time.After(3 * time.Second): |
| + t.Fatal("returned main cleanup missing") |
| + } |
| + if rep.calls.Load() != 0 { |
| + t.Fatal("replica closed before main stopped") |
| + } |
| + close(mainGate) |
| + select { |
| + case <-repEntered: |
| + case <-time.After(3 * time.Second): |
| + t.Fatal("returned replica cleanup missing") |
| + } |
| + if !main.Exited() { |
| + t.Fatal("replica closed before main exited") |
| + } |
| + base.mu.Lock() |
| + released := base.releases |
| + base.mu.Unlock() |
| + if released != 0 { |
| + t.Fatal("lease released while replica close pending") |
| + } |
| + close(repGate) |
| + if err := <-done; err == nil { |
| + t.Fatal("partial replica failure admitted") |
| + } |
| + base.mu.Lock() |
| + released = base.releases |
| + base.mu.Unlock() |
| + if released != 0 || rep.calls.Load() != 1 { |
| + t.Fatal("returned replica lease released twice", released, rep.calls.Load()) |
| + } |
| + if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), rt.Host, 1); err != ErrStopUnproven || proof != nil { |
| + t.Fatal("partial replica uncertainty erased", err) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestMaintenanceFailedAcquireReleasesReturnedLease(t *testing.T) { |
| + a, base, _, reg := testActivator(t, time.Hour) |
| + defer reg.Close() |
| + a.backend = &maintenanceBackend{fakeBackend: base, acquireFailure: errors.New("lease returned with acquisition failure")} |
| + rt := testRoute("partial-lease.example") |
| + if _, err := a.beginRequest(t.Context(), rt); err == nil { |
| + t.Fatal("failed acquisition admitted") |
| + } |
| + base.mu.Lock() |
| + released, starts := base.releases, base.starts |
| + base.mu.Unlock() |
| + if released != 1 || starts != 0 { |
| + t.Fatal("returned lease abandoned or wake continued", released, starts) |
| + } |
| + if proof, err := a.FenceAndStop(maintenanceContext(t, time.Second), rt.Host, 1); err != ErrStopUnproven || proof != nil { |
| + t.Fatal("failed acquisition became stop proof", err) |
| + } |
| +} |
amadan