| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "os/exec" |
| 7 | "path/filepath" |
| 8 | "strings" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/activator" |
| 13 | "github.com/carlosframework/platform/internal/hostbox" |
| 14 | "github.com/carlosframework/platform/internal/hostclaim" |
| 15 | "github.com/carlosframework/platform/internal/hostwrite" |
| 16 | "github.com/carlosframework/platform/internal/park" |
| 17 | "github.com/carlosframework/platform/internal/provision" |
| 18 | "github.com/carlosframework/platform/internal/registry" |
| 19 | ) |
| 20 | |
| 21 | // plan10J2Rollback is the rollback id a restoring op carries in its intent. |
| 22 | const plan10J2Rollback = "rb_abcdefghijklmnopqrstuvwxyz" |
| 23 | |
| 24 | // plan10J2Op is the source's recovery op as box-1 (the source) executes it. |
| 25 | func plan10J2Op(kind, phase string) hostclaim.Op { |
| 26 | op := hostclaim.Op{Kind: kind, ID: "m1", Phase: phase, Exec: "box-1", Participants: []string{"box-1", "box-2"}, Intent: map[string]string{"to": "uk", "source": "box-1", "target": "box-2"}} |
| 27 | switch kind { |
| 28 | case hostclaim.KindRestoring: |
| 29 | op.Intent["rollback"] = plan10J2Rollback |
| 30 | case hostclaim.KindReverting: |
| 31 | op.Journal = map[string]string{hostclaim.JournalRevPhase: phase} |
| 32 | } |
| 33 | return op |
| 34 | } |
| 35 | |
| 36 | func plan10J2Observed(kind, phase, exec string) hostclaim.Observed { |
| 37 | op := plan10J2Op(kind, phase) |
| 38 | op.Exec = exec |
| 39 | return hostclaim.Observed{Host: recoveryHost, Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{WriterBox: "box-1"}, Op: op}} |
| 40 | } |
| 41 | |
| 42 | // plan10J2Recovery turns the recovery fixture's created host into the |
| 43 | // source's recovery op at phase, with box-1 executing, and admits one writer |
| 44 | // there through the real runtime. The moves C1 owners are bound. |
| 45 | func plan10J2Recovery(t *testing.T, kind, phase string, p activator.StartPolicy) (*recoveryFixture, *recoveryController, *plan9D2Attempts, *int, hostclaim.Observed) { |
| 46 | t.Helper() |
| 47 | f := newRecoveryFixture(t) |
| 48 | ctx := t.Context() |
| 49 | if _, err := hostclaim.Update(ctx, f.claims, f.read(), func(c *hostclaim.Claim) error { |
| 50 | c.Op = plan10J2Op(kind, phase) |
| 51 | return nil |
| 52 | }); err != nil { |
| 53 | t.Fatal(err) |
| 54 | } |
| 55 | c := f.controller("boot-1", "controller-1", 4101, "start-1") |
| 56 | c.wake.Inputs = plan9D2Inputs{recoveryInputs{f}} |
| 57 | ops, probes := &plan9D2Attempts{}, new(int) |
| 58 | c.wake.Continue = hostclaim.SessionContinues |
| 59 | c.wake.Operations = ops |
| 60 | c.wake.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { |
| 61 | *probes++ |
| 62 | return nil |
| 63 | } |
| 64 | live, cancel := context.WithCancel(context.Background()) |
| 65 | t.Cleanup(cancel) |
| 66 | done := make(chan struct{}) |
| 67 | go func() { <-live.Done(); close(done) }() |
| 68 | o := f.read() |
| 69 | c.wake.mu.Lock() |
| 70 | c.wake.pending[recoveryHost] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: live, cancel: cancel, done: done} |
| 71 | c.wake.mu.Unlock() |
| 72 | admitted, err := c.runtime.Admit(ctx, registry.Route{Host: recoveryHost}, o, p) |
| 73 | if err != nil { |
| 74 | t.Fatal(err) |
| 75 | } |
| 76 | if admitted.Claim.Op.Kind != kind || admitted.Claim.Op.Phase != phase || admitted.Claim.Holder.WriterBox != "box-1" || admitted.Claim.Holder.WakeRetry != nil { |
| 77 | t.Fatalf("not a %s admission at %s: %+v", kind, phase, admitted.Claim) |
| 78 | } |
| 79 | return f, c, ops, probes, admitted |
| 80 | } |
| 81 | |
| 82 | // The recovery probe is admitted at probing and nowhere else; an ordinary |
| 83 | // wake of a recovery op is not admitted here at any phase, although |
| 84 | // hostbox.WakePhase's own table grants probed and opened to a wake, because |
| 85 | // the assembly admits exactly the probe shape for these kinds. |
| 86 | func TestPlan10J2RecoveryAdmissionPath(t *testing.T) { |
| 87 | probe, wake := activator.StartPolicy{Probe: true}, activator.StartPolicy{} |
| 88 | bound := &ClaimWakeAssembly{Operations: &plan9D2Attempts{}, ProbeReady: func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }} |
| 89 | unbound := &ClaimWakeAssembly{} |
| 90 | noProbe := &ClaimWakeAssembly{Operations: &plan9D2Attempts{}} |
| 91 | var refusal *ClaimDependencyError |
| 92 | for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} { |
| 93 | t.Run(kind, func(t *testing.T) { |
| 94 | if err := bound.admitsOperation(plan10J2Observed(kind, "probing", "box-1"), probe, "box-1"); err != nil { |
| 95 | t.Fatalf("recovery probe at probing: %v", err) |
| 96 | } |
| 97 | for name, a := range map[string]*ClaimWakeAssembly{"unbound": unbound, "no probe readiness": noProbe} { |
| 98 | if err := a.admitsOperation(plan10J2Observed(kind, "probing", "box-1"), probe, "box-1"); !errors.As(err, &refusal) { |
| 99 | t.Fatalf("%s: %v", name, err) |
| 100 | } |
| 101 | } |
| 102 | for _, phase := range []string{"probed", "opened", "claimed", "released", "recovered", ""} { |
| 103 | if err := bound.admitsOperation(plan10J2Observed(kind, phase, "box-1"), probe, "box-1"); !errors.Is(err, hostbox.ErrChanged) { |
| 104 | t.Fatalf("probe at %q: %v", phase, err) |
| 105 | } |
| 106 | } |
| 107 | for _, phase := range []string{"probing", "probed", "opened"} { |
| 108 | if err := bound.admitsOperation(plan10J2Observed(kind, phase, "box-1"), wake, "box-1"); !errors.As(err, &refusal) { |
| 109 | t.Fatalf("ordinary wake at %q was not refused: %v", phase, err) |
| 110 | } |
| 111 | } |
| 112 | if err := bound.admitsOperation(plan10J2Observed(kind, "probing", "box-2"), probe, "box-1"); !errors.Is(err, hostbox.ErrChanged) { |
| 113 | t.Fatalf("another executor: %v", err) |
| 114 | } |
| 115 | lost := plan10J2Observed(kind, "probing", "box-1") |
| 116 | lost.Claim.Lost = []string{"box-1"} |
| 117 | if err := bound.admitsOperation(lost, probe, "box-1"); !errors.Is(err, hostbox.ErrChanged) { |
| 118 | t.Fatalf("a box declared lost: %v", err) |
| 119 | } |
| 120 | }) |
| 121 | } |
| 122 | for _, kind := range []string{hostclaim.KindMoving, hostclaim.KindPurging, hostclaim.KindRelocating} { |
| 123 | if err := bound.admitsOperation(plan10J2Observed(kind, "probing", "box-1"), probe, "box-1"); !errors.As(err, &refusal) && !errors.Is(err, hostwrite.ErrUnavailable) { |
| 124 | t.Fatalf("%s probe: %v", kind, err) |
| 125 | } |
| 126 | } |
| 127 | if err := bound.admitsOperation(plan10J2Observed(hostclaim.KindLanding, "committed", "box-1"), wake, "box-1"); err != nil { |
| 128 | t.Fatalf("landing wake at committed: %v", err) |
| 129 | } |
| 130 | } |
| 131 | |
| 132 | func TestPlan10J2PrepareBindsTheAttemptsAndProbeReadinessToARecoveryProbe(t *testing.T) { |
| 133 | probe := activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true} |
| 134 | for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} { |
| 135 | t.Run(kind, func(t *testing.T) { |
| 136 | _, c, ops, probes, admitted := plan10J2Recovery(t, kind, hostclaim.PhaseProbing, probe) |
| 137 | ctx := t.Context() |
| 138 | s, err := c.runtime.Prepare(ctx, registry.Route{Host: recoveryHost}, admitted, probe) |
| 139 | if err != nil { |
| 140 | t.Fatal(err) |
| 141 | } |
| 142 | if s.OpenProbe == nil || s.CurrentAttempt == nil || s.Reserve == nil || s.Failed == nil || s.Healthy == nil || s.Ready == nil { |
| 143 | t.Fatal("the recovery probe start is missing its operation bindings") |
| 144 | } |
| 145 | opened := time.Date(2026, 10, 8, 9, 0, 0, 0, time.UTC) |
| 146 | if got, err := s.OpenProbe(ctx, opened); err != nil || !got.Equal(opened) { |
| 147 | t.Fatalf("open: %v %v", got, err) |
| 148 | } |
| 149 | if err := s.Reserve(ctx, 0); err != nil { |
| 150 | t.Fatal(err) |
| 151 | } |
| 152 | if a := s.CurrentAttempt(); a.Session != admitted.Claim.Holder.Session { |
| 153 | t.Fatalf("attempt: %+v", a) |
| 154 | } |
| 155 | // The second readiness request is the probe check, which needs |
| 156 | // the started role's cgroup: before any launch it refuses and |
| 157 | // the app is never asked. |
| 158 | if err := s.Healthy(ctx, 1, exec.Command("/bin/true")); !errors.Is(err, hostwrite.ErrUnavailable) || *probes != 0 { |
| 159 | t.Fatalf("healthy before launch: %v probes=%d", err, *probes) |
| 160 | } |
| 161 | want := recoveryTuple(admitted, c.local) |
| 162 | for i, start := range ops.starts { |
| 163 | if start.Admitted.ETag != admitted.ETag || start.Owner != want || !start.Policy.Probe || start.Admitted.Claim.Op.Kind != kind || start.Admitted.Claim.Op.Phase != hostclaim.PhaseProbing { |
| 164 | t.Fatalf("call %s was bound to %+v", ops.calls[i], start) |
| 165 | } |
| 166 | } |
| 167 | if got := strings.Join(ops.calls, ","); got != "open,reserve-0,current,healthy-1" { |
| 168 | t.Fatalf("calls: %s", got) |
| 169 | } |
| 170 | // A probe of this kind without the probe readiness is refused. |
| 171 | c.wake.ProbeReady = nil |
| 172 | if _, err := c.wake.readiness(admitted, probe, registry.Route{Host: recoveryHost}, ClaimLaunchInputs{}, c.wake.get(recoveryHost), nil); err == nil { |
| 173 | t.Fatal("a recovery probe got readiness without its launch") |
| 174 | } |
| 175 | }) |
| 176 | } |
| 177 | } |
| 178 | |
| 179 | // A recovery writer admitted at probing keeps working one adjacent forward |
| 180 | // step at a time (claimSameWriter under Continue): probing to probed to |
| 181 | // opened. A skipped or backward phase, or any change of op, kind, session, |
| 182 | // stamp, generation or executor ends the attempt's session. |
| 183 | func TestPlan10J2AttemptAdvancesOnlyAlongTheRecoveryRow(t *testing.T) { |
| 184 | for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} { |
| 185 | other := hostclaim.KindReverting |
| 186 | if kind == hostclaim.KindReverting { |
| 187 | other = hostclaim.KindRestoring |
| 188 | } |
| 189 | at := func(phase string, mutate func(*hostclaim.Claim)) hostclaim.Observed { |
| 190 | o := plan10J2Observed(kind, phase, "box-1") |
| 191 | o.Claim.Holder = &hostclaim.Holder{Incarnation: "inc", Stamp: 10, Session: 11, Generation: "g1", WriterBox: "box-1"} |
| 192 | if mutate != nil { |
| 193 | mutate(o.Claim) |
| 194 | } |
| 195 | return o |
| 196 | } |
| 197 | start := func(phase string) *claimWakeAttempt { |
| 198 | return &claimWakeAttempt{observed: at(phase, nil)} |
| 199 | } |
| 200 | t.Run(kind+"/forward", func(t *testing.T) { |
| 201 | a := start("probing") |
| 202 | for _, phase := range []string{"probing", "probed", "opened"} { |
| 203 | if !a.advance(at(phase, nil), hostclaim.SessionContinues) { |
| 204 | t.Fatalf("did not advance to %s", phase) |
| 205 | } |
| 206 | } |
| 207 | }) |
| 208 | for name, tc := range map[string]struct { |
| 209 | from, to string |
| 210 | mutate func(*hostclaim.Claim) |
| 211 | }{ |
| 212 | "skipped phase": {"probing", "opened", nil}, |
| 213 | "backward phase": {"probed", "probing", nil}, |
| 214 | "before the row": {"probing", "claimed", nil}, |
| 215 | "changed session": {"probing", "probed", func(c *hostclaim.Claim) { c.Holder.Session++ }}, |
| 216 | "changed stamp": {"probing", "probed", func(c *hostclaim.Claim) { c.Holder.Stamp++ }}, |
| 217 | "changed gen": {"probing", "probed", func(c *hostclaim.Claim) { c.Holder.Generation = "fresh" }}, |
| 218 | "changed inc": {"probing", "probed", func(c *hostclaim.Claim) { c.Holder.Incarnation = "other" }}, |
| 219 | "changed op": {"probing", "probed", func(c *hostclaim.Claim) { c.Op.ID = "m2" }}, |
| 220 | "changed exec": {"probing", "probed", func(c *hostclaim.Claim) { c.Op.Exec = "box-2" }}, |
| 221 | "changed kind": {"probing", "probed", func(c *hostclaim.Claim) { c.Op.Kind = other }}, |
| 222 | "landing kind": {"probing", "probed", func(c *hostclaim.Claim) { c.Op.Kind = hostclaim.KindLanding }}, |
| 223 | } { |
| 224 | t.Run(kind+"/"+name, func(t *testing.T) { |
| 225 | a := start(tc.from) |
| 226 | before := a.observed |
| 227 | if a.advance(at(tc.to, tc.mutate), hostclaim.SessionContinues) { |
| 228 | t.Fatalf("%s advanced", name) |
| 229 | } |
| 230 | if a.observed.Claim.Op.Phase != before.Claim.Op.Phase { |
| 231 | t.Fatalf("a refused advance moved the checkpoint to %s", a.observed.Claim.Op.Phase) |
| 232 | } |
| 233 | }) |
| 234 | } |
| 235 | } |
| 236 | } |
| 237 | |
| 238 | // Teardown of an admitted recovery writer follows the same row: a stop |
| 239 | // names the session at any phase it went on to reach, never an earlier one. |
| 240 | func TestPlan10J2TeardownReachesForwardOnlyAlongTheRecoveryRow(t *testing.T) { |
| 241 | for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} { |
| 242 | at := func(phase string) hostclaim.Observed { |
| 243 | o := plan10J2Observed(kind, phase, "box-1") |
| 244 | o.Claim.Holder = &hostclaim.Holder{Incarnation: "inc", Stamp: 10, Session: 11, Generation: "g1", WriterBox: "box-1", Location: "loc"} |
| 245 | return o |
| 246 | } |
| 247 | reaches := hostclaim.SessionReaches |
| 248 | if !sameStopWriter(at("probing"), at("opened"), reaches) || !sameStopWriter(at("probed"), at("opened"), reaches) { |
| 249 | t.Fatalf("%s: a stop did not reach the later phase", kind) |
| 250 | } |
| 251 | if sameStopWriter(at("opened"), at("probing"), reaches) || sameStopWriter(at("probed"), at("probing"), reaches) { |
| 252 | t.Fatalf("%s: a stop named an earlier phase", kind) |
| 253 | } |
| 254 | if sameStopWriter(at("claimed"), at("probing"), reaches) { |
| 255 | t.Fatalf("%s: a stop crossed from before the row", kind) |
| 256 | } |
| 257 | } |
| 258 | } |
| 259 | |
| 260 | // An admission error for a recovery probe releases its pending lease, |
| 261 | // exactly as a landing's does (before any prelaunch record), and a stop of an |
| 262 | // attempt with no prelaunch record never runs the local root's stop. |
| 263 | func TestPlan10J2AdmissionFailureReleasesTheRecoveryLease(t *testing.T) { |
| 264 | t.Parallel() |
| 265 | probe := activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true} |
| 266 | for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} { |
| 267 | t.Run(kind, func(t *testing.T) { |
| 268 | f := newRecoveryFixture(t) |
| 269 | ctx := t.Context() |
| 270 | if _, err := hostclaim.Update(ctx, f.claims, f.read(), func(c *hostclaim.Claim) error { |
| 271 | c.Op = plan10J2Op(kind, hostclaim.PhaseProbing) |
| 272 | return nil |
| 273 | }); err != nil { |
| 274 | t.Fatal(err) |
| 275 | } |
| 276 | c := f.controller("boot-1", "controller-1", 4101, "start-1") |
| 277 | c.wake.Inputs = plan9D2Inputs{recoveryInputs{f}} |
| 278 | c.wake.Continue = hostclaim.SessionContinues |
| 279 | c.wake.Operations = &plan9D2Attempts{} |
| 280 | c.wake.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil } |
| 281 | stops := 0 |
| 282 | c.wake.Stop = func(context.Context, hostclaim.Observed) error { stops++; return nil } |
| 283 | dead, kill := context.WithCancel(context.Background()) |
| 284 | kill() |
| 285 | done := make(chan struct{}) |
| 286 | close(done) |
| 287 | o := f.read() |
| 288 | c.wake.mu.Lock() |
| 289 | c.wake.pending[recoveryHost] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: dead, cancel: kill, done: done} |
| 290 | c.wake.mu.Unlock() |
| 291 | if _, err := c.runtime.Admit(ctx, registry.Route{Host: recoveryHost}, o, probe); err == nil { |
| 292 | t.Fatal("admission under a dead lease succeeded") |
| 293 | } |
| 294 | if c.wake.get(recoveryHost) != nil { |
| 295 | t.Fatal("the failed recovery admission left its pending lease") |
| 296 | } |
| 297 | if stops != 0 { |
| 298 | t.Fatalf("the local stop ran %d times for an attempt with no launch", stops) |
| 299 | } |
| 300 | }) |
| 301 | } |
| 302 | } |
| 303 | |
| 304 | // The recovery's local artifact retirement stays fail-closed: the landing's |
| 305 | // retire-artifacts is bound to the landing writer (hostguard.LandingWriter), |
| 306 | // and a recovery op is not one. Recover has nothing to re-drive for it. |
| 307 | func TestPlan10J2ArtifactRetirementRefusesUnadmittedRecovery(t *testing.T) { |
| 308 | f := newRecoveryFixture(t) |
| 309 | dataDir := f.guard.DataDir |
| 310 | r := claimArtifactRetirement{Local: park.Local{Claims: f.claims, Box: "box-1", DataDir: dataDir}, Continue: hostclaim.SessionContinues} |
| 311 | in := ClaimLaunchInputs{DBPath: filepath.Join(dataDir, "state", "data.db")} |
| 312 | for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} { |
| 313 | o := plan10J2Observed(kind, "probing", "box-1") |
| 314 | o.Claim.Holder = &hostclaim.Holder{Incarnation: "inc", Stamp: 10, Session: 11, Generation: "g1", WriterBox: "box-1"} |
| 315 | if err := r.Retire(t.Context(), o, in, "m1"); !errors.Is(err, hostbox.ErrChanged) { |
| 316 | t.Fatalf("%s retire: %v", kind, err) |
| 317 | } |
| 318 | if err := r.Recover(t.Context(), o, in); !errors.Is(err, hostbox.ErrChanged) { |
| 319 | t.Fatalf("%s recover: %v", kind, err) |
| 320 | } |
| 321 | } |
| 322 | } |
| 323 | |
| 324 | // A clean return records its current inputs at probing before the held |
| 325 | // render and launch. No configured phase is needed. The adapter must call |
| 326 | // recoveryProbeRequest.Record first; movesrecoveryprobe.go supplies the adapter. |
| 327 | // This test exercises the existing recording transition and the policy checks |
| 328 | // for the recorded configuration, release and launch. |
| 329 | func TestPlan10J2CleanReturnRecordsRecoveryInputsBeforeRenderAndLaunch(t *testing.T) { |
| 330 | g := provision.MoveGuard{ID: mpOp, Incarnation: mpInc} |
| 331 | rec := rpRestoringRecord(provision.RbSourceProbing) |
| 332 | rec.Move.Rollback.Dirty, rec.Move.Rollback.RecoveryInputs = false, nil |
| 333 | c := rpRestoringClaim(hostclaim.PhaseProbing) |
| 334 | read := (&mpReader{rec: rec}).read |
| 335 | check := func(action string, id hostwrite.Identity) error { |
| 336 | p := hostbox.Policy{Moves: MovesClaimPolicy(read), Held: func(context.Context, string) (string, error) { return rpRB, nil }} |
| 337 | return p.Check(t.Context(), c, rpSelection(c, action), id) |
| 338 | } |
| 339 | recovered := rpRecovery() |
| 340 | rpRefused(t, check("spawn", rpLaunch(recovered)), "no recorded recovery inputs") |
| 341 | if err := provision.MoveRecordRecoveryInputs(g, rpRB, *recovered)(rec); err != nil { |
| 342 | t.Fatalf("recording at probing: %v", err) |
| 343 | } |
| 344 | if rec.Move.Rollback.RecoveryInputs == nil || rec.Move.Rollback.Dirty { |
| 345 | t.Fatalf("the record is %+v", rec.Move.Rollback) |
| 346 | } |
| 347 | if err := check("held-config", rpIdentity("held-config")); err != nil { |
| 348 | t.Fatalf("render of the recorded inputs: %v", err) |
| 349 | } |
| 350 | if err := check("held-release", rpIdentity("held-release")); err != nil { |
| 351 | t.Fatalf("release of the recorded inputs: %v", err) |
| 352 | } |
| 353 | if err := check("spawn", rpLaunch(recovered)); err != nil { |
| 354 | t.Fatalf("the launch of the recorded inputs: %v", err) |
| 355 | } |
| 356 | other := *recovered |
| 357 | other.Release = "v9.9.9" |
| 358 | rpRefused(t, check("spawn", rpLaunch(&other)), "the launch is not the recovery's") |
| 359 | // Recording is not legal at any other source phase. |
| 360 | for _, phase := range []string{provision.RbSourceRestored, provision.RbSourceProbed, provision.RbSourceOpened} { |
| 361 | later := rpRestoringRecord(phase) |
| 362 | later.Move.Rollback.Dirty, later.Move.Rollback.RecoveryInputs = false, nil |
| 363 | if err := provision.MoveRecordRecoveryInputs(g, rpRB, *recovered)(later); err == nil { |
| 364 | t.Fatalf("recording was legal at %s", phase) |
| 365 | } |
| 366 | } |
| 367 | } |
| 368 | |
amadan