| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "testing" |
| 7 | "time" |
| 8 | |
| 9 | "github.com/carlosframework/platform/internal/hostbox" |
| 10 | "github.com/carlosframework/platform/internal/hostclaim" |
| 11 | "github.com/carlosframework/platform/internal/registry" |
| 12 | ) |
| 13 | |
| 14 | // These tests enter sweepOnce with the cut fence still installed. Calling |
| 15 | // reviewRecoveryHold directly would miss the fenced-host early continue. |
| 16 | func sweptProbe(t *testing.T) (*Service, *ClaimRuntime, *actEntry, *fakeProc, *fakeRep, hostclaim.Observed) { |
| 17 | t.Helper() |
| 18 | obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box")) |
| 19 | admitted := probeAdmitted(obs) |
| 20 | a, r, _ := probeService(t, obs) |
| 21 | a.fences = map[string]uint64{probeHost: 9} |
| 22 | e, proc, rep, err := heldProbe(t, a, 9, admitted, probeInputs()) |
| 23 | if err != nil { |
| 24 | t.Fatal(err) |
| 25 | } |
| 26 | e.recoveryPolicy.RecoveryFence = 9 |
| 27 | r.Read = func(context.Context, string) (hostclaim.Observed, error) { return admitted, nil } |
| 28 | return a, r, e, proc, rep, admitted |
| 29 | } |
| 30 | |
| 31 | func waitProbe(t *testing.T, a *Service, predicate func() bool) { |
| 32 | t.Helper() |
| 33 | deadline := time.Now().Add(time.Second) |
| 34 | for time.Now().Before(deadline) { |
| 35 | a.mu.Lock() |
| 36 | ok := predicate() |
| 37 | a.mu.Unlock() |
| 38 | if ok { |
| 39 | return |
| 40 | } |
| 41 | time.Sleep(time.Millisecond) |
| 42 | } |
| 43 | t.Fatal("probe supervision did not finish") |
| 44 | } |
| 45 | |
| 46 | func TestRecoveryProbeSweepRetiresFailedBehindCutFence(t *testing.T) { |
| 47 | t.Parallel() |
| 48 | for _, mode := range []string{"exit", "lease", "join-exit", "join-lease"} { |
| 49 | t.Run(mode, func(t *testing.T) { |
| 50 | a, r, e, proc, rep, admitted := sweptProbe(t) |
| 51 | if mode == "exit" || mode == "join-exit" { |
| 52 | proc.exit() |
| 53 | } else { |
| 54 | rep.setErr(errors.New("lease lost")) |
| 55 | } |
| 56 | if mode == "join-exit" || mode == "join-lease" { |
| 57 | if err := a.joinRecoveryProbe(t.Context(), e, *e.recoveryPolicy); !errors.Is(err, hostbox.ErrChanged) { |
| 58 | t.Fatalf("join: %v", err) |
| 59 | } |
| 60 | } else { |
| 61 | a.sweepOnce() |
| 62 | } |
| 63 | waitProbe(t, a, func() bool { return a.entries[probeHost] == nil }) |
| 64 | if !proc.isStopped() || !rep.isClosed() || a.FenceGeneration(probeHost) != 9 { |
| 65 | t.Fatal("failed probe was not retired with its cut fence retained") |
| 66 | } |
| 67 | // A retry reaches the admission/start path again, rather than joining |
| 68 | // the dead entry forever. The fixture launcher deliberately refuses. |
| 69 | next := probeAdmitted(admitted) |
| 70 | next.Claim.Holder.Generation = "three" |
| 71 | reads := 0 |
| 72 | r.Read = func(context.Context, string) (hostclaim.Observed, error) { |
| 73 | reads++ |
| 74 | if reads == 1 { |
| 75 | return admitted, nil |
| 76 | } |
| 77 | return next, nil |
| 78 | } |
| 79 | r.Admit = func(context.Context, registry.Route, hostclaim.Observed, StartPolicy) (hostclaim.Observed, error) { |
| 80 | return next, nil |
| 81 | } |
| 82 | if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, *e.recoveryPolicy); !errors.Is(err, ErrClaimIsolationUnavailable) { |
| 83 | t.Fatalf("retry did not reach the launcher: %v", err) |
| 84 | } |
| 85 | }) |
| 86 | } |
| 87 | } |
| 88 | |
| 89 | func TestRecoveryProbeSweepOpensOnlyExactSessionAndFence(t *testing.T) { |
| 90 | t.Parallel() |
| 91 | for _, phase := range []string{hostclaim.PhaseProbing, hostclaim.PhaseProbed, hostclaim.PhaseOpened, "serving", "new-session", "foreign-executor", "unreadable"} { |
| 92 | t.Run(phase, func(t *testing.T) { |
| 93 | a, r, e, proc, _, admitted := sweptProbe(t) |
| 94 | // The cut fence carries an owner (FenceAndStopAs at the same |
| 95 | // generation): the opening's lift clears it with the fence. |
| 96 | a.mu.Lock() |
| 97 | a.fenceOwners = map[string]string{probeHost: "owner-1"} |
| 98 | a.mu.Unlock() |
| 99 | fresh := admitted |
| 100 | c, h := *admitted.Claim, *admitted.Claim.Holder |
| 101 | c.Holder = &h |
| 102 | fresh.Claim = &c |
| 103 | c.Op.Phase = phase |
| 104 | switch phase { |
| 105 | case "serving": |
| 106 | c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"} |
| 107 | case "new-session": |
| 108 | c.Op.Phase = hostclaim.PhaseOpened |
| 109 | h.Session++ |
| 110 | case "foreign-executor": |
| 111 | c.Op.Phase = hostclaim.PhaseOpened |
| 112 | c.Op.Exec = "other" |
| 113 | } |
| 114 | read := make(chan struct{}, 1) |
| 115 | r.Read = func(context.Context, string) (hostclaim.Observed, error) { |
| 116 | read <- struct{}{} |
| 117 | if phase == "unreadable" { |
| 118 | return hostclaim.Observed{}, errors.New("unreadable") |
| 119 | } |
| 120 | return fresh, nil |
| 121 | } |
| 122 | a.sweepOnce() |
| 123 | select { |
| 124 | case <-read: |
| 125 | case <-time.After(time.Second): |
| 126 | t.Fatal("sweep skipped held probe behind its cut fence") |
| 127 | } |
| 128 | waitProbe(t, a, func() bool { return !e.probeReviewing }) |
| 129 | opens := phase == hostclaim.PhaseOpened || phase == "serving" |
| 130 | stops := phase == "new-session" || phase == "foreign-executor" |
| 131 | if stops { |
| 132 | waitProbe(t, a, func() bool { return a.entries[probeHost] == nil }) |
| 133 | } |
| 134 | if opens { |
| 135 | if a.Fenced(probeHost) || proc.isStopped() { |
| 136 | t.Fatal("exact opening did not release the cut fence and retain its process") |
| 137 | } |
| 138 | a.mu.Lock() |
| 139 | _, owned := a.fenceOwners[probeHost] |
| 140 | a.mu.Unlock() |
| 141 | if owned { |
| 142 | t.Fatal("the opening lifted the cut fence but left its owner") |
| 143 | } |
| 144 | release, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost}) |
| 145 | if err != nil { |
| 146 | t.Fatalf("opened session cannot serve: %v", err) |
| 147 | } |
| 148 | release() |
| 149 | } else { |
| 150 | if a.FenceGeneration(probeHost) != 9 { |
| 151 | t.Fatal("non-opening changed fence") |
| 152 | } |
| 153 | if _, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost}); !errors.Is(err, ErrFenced) { |
| 154 | t.Fatalf("request: %v", err) |
| 155 | } |
| 156 | if !stops && proc.isStopped() { |
| 157 | t.Fatal("a held or unreadable session was stopped") |
| 158 | } |
| 159 | } |
| 160 | }) |
| 161 | } |
| 162 | } |
| 163 | |
| 164 | func TestRecoveryProbeSweepPreservesReplacementFence(t *testing.T) { |
| 165 | t.Parallel() |
| 166 | for _, afterRead := range []bool{false, true} { |
| 167 | t.Run(map[bool]string{false: "before-read", true: "during-read"}[afterRead], func(t *testing.T) { |
| 168 | a, r, e, proc, rep, admitted := sweptProbe(t) |
| 169 | read, resume := make(chan struct{}), make(chan struct{}) |
| 170 | r.Read = func(context.Context, string) (hostclaim.Observed, error) { |
| 171 | close(read) |
| 172 | <-resume |
| 173 | c := *admitted.Claim |
| 174 | c.Op.Phase = hostclaim.PhaseOpened |
| 175 | admitted.Claim = &c |
| 176 | return admitted, nil |
| 177 | } |
| 178 | replace := func() { a.mu.Lock(); a.fences[probeHost] = 12; a.mu.Unlock() } |
| 179 | if !afterRead { |
| 180 | replace() |
| 181 | } |
| 182 | a.sweepOnce() |
| 183 | if afterRead { |
| 184 | select { |
| 185 | case <-read: |
| 186 | case <-time.After(time.Second): |
| 187 | close(resume) |
| 188 | t.Fatal("sweep skipped cut fence") |
| 189 | } |
| 190 | replace() |
| 191 | } |
| 192 | close(resume) |
| 193 | waitProbe(t, a, func() bool { return a.entries[probeHost] == nil && !e.probeReviewing }) |
| 194 | if a.FenceGeneration(probeHost) != 12 || !proc.isStopped() || !rep.isClosed() { |
| 195 | t.Fatal("replacement fence or retirement lost") |
| 196 | } |
| 197 | }) |
| 198 | } |
| 199 | } |
| 200 | |
| 201 | func TestRecoveryProbeRefusesAnotherOperationsFence(t *testing.T) { |
| 202 | obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box")) |
| 203 | a, _, run := probeService(t, obs) |
| 204 | a.fences = map[string]uint64{probeHost: 12} |
| 205 | p := probePolicy(probeInputs()) |
| 206 | p.RecoveryFence = 9 |
| 207 | if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, p); !errors.Is(err, ErrFenced) { |
| 208 | t.Fatalf("foreign fence: %v", err) |
| 209 | } |
| 210 | if len(run.seen()) != 0 || a.Owns(probeHost) || a.FenceGeneration(probeHost) != 12 { |
| 211 | t.Fatal("foreign fence allowed recovery admission") |
| 212 | } |
| 213 | } |
| 214 | |
| 215 | // The moves worker may finish the rollback and lift its own fence before |
| 216 | // the slower activator sweep observes opened/serving. The exact session |
| 217 | // still owns its process; a missing fence alone is not an opening proof. |
| 218 | func TestRecoveryProbeSweepAcceptsOwnerLiftOnlyWhenOpened(t *testing.T) { |
| 219 | t.Parallel() |
| 220 | for _, phase := range []string{hostclaim.PhaseOpened, "serving", hostclaim.PhaseProbing} { |
| 221 | for _, duringRead := range []bool{false, true} { |
| 222 | t.Run(phase+map[bool]string{false: "/before", true: "/during"}[duringRead], func(t *testing.T) { |
| 223 | a, r, e, proc, _, admitted := sweptProbe(t) |
| 224 | read, resume := make(chan struct{}), make(chan struct{}) |
| 225 | r.Read = func(context.Context, string) (hostclaim.Observed, error) { |
| 226 | close(read) |
| 227 | <-resume |
| 228 | c := *admitted.Claim |
| 229 | c.Op.Phase = phase |
| 230 | if phase == "serving" { |
| 231 | c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"} |
| 232 | } |
| 233 | admitted.Claim = &c |
| 234 | return admitted, nil |
| 235 | } |
| 236 | if !duringRead && !a.LiftFence(probeHost, 9) { |
| 237 | t.Fatal("owner lift refused") |
| 238 | } |
| 239 | a.sweepOnce() |
| 240 | select { |
| 241 | case <-read: |
| 242 | case <-time.After(time.Second): |
| 243 | close(resume) |
| 244 | t.Fatal("owner lift retired probe before reading claim") |
| 245 | } |
| 246 | if duringRead && !a.LiftFence(probeHost, 9) { |
| 247 | close(resume) |
| 248 | t.Fatal("owner lift refused") |
| 249 | } |
| 250 | close(resume) |
| 251 | waitProbe(t, a, func() bool { return !e.probeReviewing }) |
| 252 | if phase == hostclaim.PhaseProbing { |
| 253 | a.mu.Lock() |
| 254 | held := a.entries[probeHost] == e && e.state == actAwake && e.probeHeld |
| 255 | a.mu.Unlock() |
| 256 | if !held || proc.isStopped() { |
| 257 | t.Fatal("an owner lift with a probing read retired the session") |
| 258 | } |
| 259 | if _, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost}); !errors.Is(err, ErrFenced) { |
| 260 | t.Fatalf("a missing fence opened a still-held probe: %v", err) |
| 261 | } |
| 262 | // The earlier read may precede opening. A later sweep |
| 263 | // must see the same session opened before it can serve. |
| 264 | r.Read = func(context.Context, string) (hostclaim.Observed, error) { |
| 265 | c := *admitted.Claim |
| 266 | c.Op.Phase = hostclaim.PhaseOpened |
| 267 | return hostclaim.Observed{Host: admitted.Host, Claim: &c, ETag: admitted.ETag}, nil |
| 268 | } |
| 269 | a.sweepOnce() |
| 270 | waitProbe(t, a, func() bool { return !e.probeHeld }) |
| 271 | } |
| 272 | if proc.isStopped() || a.Fenced(probeHost) { |
| 273 | t.Fatal("opened session lost its process") |
| 274 | } |
| 275 | release, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost}) |
| 276 | if err != nil { |
| 277 | t.Fatal(err) |
| 278 | } |
| 279 | release() |
| 280 | }) |
| 281 | } |
| 282 | } |
| 283 | } |
| 284 | |
| 285 | func TestRecoveryProbeHealthyFinishAfterOwnerLift(t *testing.T) { |
| 286 | obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box")) |
| 287 | a, r, _ := probeService(t, obs) |
| 288 | admitted := probeAdmitted(obs) |
| 289 | e, proc, _, err := heldProbe(t, a, 9, admitted, probeInputs()) |
| 290 | if err != nil { |
| 291 | t.Fatalf("owner lift prematurely retired healthy finish: %v", err) |
| 292 | } |
| 293 | c := *admitted.Claim |
| 294 | c.Op.Phase = hostclaim.PhaseOpened |
| 295 | admitted.Claim = &c |
| 296 | r.Read = func(context.Context, string) (hostclaim.Observed, error) { return admitted, nil } |
| 297 | a.sweepOnce() |
| 298 | waitProbe(t, a, func() bool { return !e.probeHeld }) |
| 299 | if proc.isStopped() { |
| 300 | t.Fatal("opened process stopped") |
| 301 | } |
| 302 | } |
| 303 | |
amadan