| 1 | //go:build linux |
| 2 | |
| 3 | package host |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "crypto/sha256" |
| 8 | "encoding/hex" |
| 9 | "encoding/json" |
| 10 | "errors" |
| 11 | "strconv" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/hostbox" |
| 16 | "github.com/carlosframework/platform/internal/hostclaim" |
| 17 | execlease "github.com/carlosframework/platform/internal/hostexec" |
| 18 | "github.com/carlosframework/platform/internal/hostguard" |
| 19 | ) |
| 20 | |
| 21 | // probeStopped moves the participation box's relocation to a passed and |
| 22 | // stopped probe: dst's session 12 admitted into P1 at the install stamp, |
| 23 | // the guard activated there, and root's writer-session stop of session 12 |
| 24 | // completed with its receipt in the guard journal (ruling 49). It returns |
| 25 | // the probing observation, the stop's fence target and root's receipt. |
| 26 | func (b *participationBox) probeStopped() (hostclaim.Observed, hostguard.FenceTarget, []byte) { |
| 27 | b.t.Helper() |
| 28 | ctx := b.t.Context() |
| 29 | probing := b.update(func(c *hostclaim.Claim) error { |
| 30 | stamp := c.Gen + 1 |
| 31 | c.Holder.Stamp, c.Holder.Session, c.Holder.Generation, c.Holder.WriterBox = stamp, 12, "P1", "dst" |
| 32 | c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Session: 12, Box: "dst", Generation: "P1", Predecessor: "g1", Mode: "cold", Region: "ie"}) |
| 33 | probe, _ := json.Marshal(hostclaim.Probe{Box: "dst", Session: 12, At: time.Date(2026, 10, 7, 3, 0, 0, 0, time.UTC)}) |
| 34 | c.Op.Phase = hostclaim.PhaseRelocatingProbing |
| 35 | c.Op.Journal = map[string]string{"install_stamp": strconv.FormatUint(stamp, 10), "prepared": "P1", "prepared_opened": "P1", "probe": string(probe)} |
| 36 | return nil |
| 37 | }) |
| 38 | if err := b.guard.Activate(ctx, participationHost, "dst", nil); err != nil { |
| 39 | b.t.Fatal(err) |
| 40 | } |
| 41 | h := probing.Claim.Holder |
| 42 | target := hostguard.FenceTarget{FenceID: "fence-1", Region: "ie", Host: participationHost, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: 12, Op: "rel-1", Phase: hostclaim.PhaseRelocatingProbing, |
| 43 | ObligationID: hostguard.WriterStopObligation(12), Box: "dst", Enrollment: b.actor.Enrollment, BootID: "boot", ControllerID: "ctl", ClaimETag: probing.ETag, DirectoryRevision: "4", RosterRevision: 4} |
| 44 | // The probe's launch is in the box's launch journal, as the wake |
| 45 | // journals it: main's ordinary cleanup would stop that session again. |
| 46 | if err := b.launches.Prelaunch(ctx, hostbox.LaunchInventory{ControllerPID: 4101, ProcessStart: "start-1", Target: target, |
| 47 | Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/" + claimContainmentName(target), Device: 1, Inode: 12}}, NoState: true}); err != nil { |
| 48 | b.t.Fatal(err) |
| 49 | } |
| 50 | if _, err := b.guard.BeginFence(ctx, target); err != nil { |
| 51 | b.t.Fatal(err) |
| 52 | } |
| 53 | at := time.Date(2026, 10, 7, 3, 1, 0, 0, time.UTC) |
| 54 | receipt := mustMarshalStopReceipt(StopReceipt{V: 1, ID: "r1", Kind: StopReceiptStopped, Region: "ie", ClaimETag: probing.ETag, ObligationID: target.ObligationID, FenceID: "fence-1", Host: participationHost, |
| 55 | Incarnation: h.Incarnation, InstallationStamp: h.Stamp, WriterSession: 12, OperationID: "rel-1", Phase: hostclaim.PhaseRelocatingProbing, Box: "dst", Enrollment: b.actor.Enrollment, |
| 56 | BootID: "boot", ControllerID: "ctl", DirectoryRevision: testReceiptNumeral(target), Binding: "4", EvidenceAt: at}) |
| 57 | if err := b.guard.CompleteFence(ctx, target, receipt, at, func(context.Context, hostguard.FenceTarget, []byte, time.Time) error { return nil }); err != nil { |
| 58 | b.t.Fatal(err) |
| 59 | } |
| 60 | return probing, target, receipt |
| 61 | } |
| 62 | |
| 63 | // serve is CompleteRelocation's serving CAS as the box sees it: kind |
| 64 | // serving on dst, the holder still the stopped probe session, and the |
| 65 | // serving op recording the stop. |
| 66 | func (b *participationBox) serve(receipt []byte) hostclaim.Observed { |
| 67 | b.t.Helper() |
| 68 | sum := sha256.Sum256(receipt) |
| 69 | return b.update(func(c *hostclaim.Claim) error { |
| 70 | done, _ := json.Marshal(hostclaim.CompletedRelocation{Op: "rel-1", Stop: hostclaim.ProbeStop{Box: "dst", Session: 12, Receipt: "fence-1", ReceiptSHA256: hex.EncodeToString(sum[:])}, |
| 71 | Stamp: c.Holder.Stamp, Phase: hostclaim.PhaseRelocatingProbing, Generation: c.Holder.Generation}) |
| 72 | c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "dst", Participants: []string{"dst"}, Journal: map[string]string{hostclaim.JournalCompleted: string(done)}} |
| 73 | return nil |
| 74 | }) |
| 75 | } |
| 76 | |
| 77 | func (b *participationBox) servingCheck(o hostclaim.Observed) error { |
| 78 | h := o.Claim.Holder |
| 79 | return b.p.Check(b.t.Context(), b.actor, o.Claim, hostbox.Selection{Host: participationHost, Box: "dst", Region: "ie", Captured: hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp}}) |
| 80 | } |
| 81 | |
| 82 | // Rev 3 finding 5, rulings 56 and 66, C10: completion without an agent |
| 83 | // restart. While the participation slot is open the serving Check denies |
| 84 | // (the wake cannot admit); DischargeCompleted closes the slot through main's |
| 85 | // Prepare, whose cleanup is C10's discharge by the slot's binding and the |
| 86 | // probe's receipt (no stop asked of root), ClearStale removes the record, |
| 87 | // the serving Check passes, and N6's handover finds the probe's receipt. |
| 88 | func TestRelocationCompletionDischargesParticipationWithoutRestart(t *testing.T) { |
| 89 | t.Parallel() |
| 90 | ctx := t.Context() |
| 91 | b := newParticipationBox(t) |
| 92 | if err := b.p.Prepare(ctx, b.read()); err != nil { // acknowledged at installing |
| 93 | t.Fatal(err) |
| 94 | } |
| 95 | _, target, receipt := b.probeStopped() |
| 96 | serving := b.serve(receipt) |
| 97 | if err := b.servingCheck(serving); err == nil { |
| 98 | t.Fatal("the serving wake was admissible with the participation open") |
| 99 | } |
| 100 | if err := b.p.DischargeCompleted(ctx, serving); err != nil { |
| 101 | t.Fatalf("discharge: %v", err) |
| 102 | } |
| 103 | if b.stops != 0 { |
| 104 | t.Fatalf("the discharge asked root for %d stops", b.stops) |
| 105 | } |
| 106 | if len(b.p.slots) != 0 || len(b.records()) != 0 { |
| 107 | t.Fatalf("slots %d records %v", len(b.p.slots), b.records()) |
| 108 | } |
| 109 | if err := b.servingCheck(b.read()); err != nil { |
| 110 | t.Fatalf("the serving Check after the discharge: %v", err) |
| 111 | } |
| 112 | h := relocationCompletedHandover{Guard: b.guard, Box: "dst"} |
| 113 | if err := h.Before(ctx, b.read()); err != nil { |
| 114 | t.Fatalf("N6 before admission: %v", err) |
| 115 | } |
| 116 | if r, err := h.receipt(ctx, b.read()); err != nil || r.Target != target { |
| 117 | t.Fatalf("N6's prior: %+v %v", r.Target, err) |
| 118 | } |
| 119 | } |
| 120 | |
| 121 | // Rev 4 finding 3, ruling 66: the agent restarts after the serving CAS, |
| 122 | // with the participation's .hostexec record left behind. The new process's |
| 123 | // DischargeCompleted reaches clearUntracked, C10 discharges by the |
| 124 | // record's binding (no stop asked, the probe's fence untouched), |
| 125 | // ClearStale removes the record, and the first serving wake's handover |
| 126 | // from the probe's receipt is accepted by the guard. |
| 127 | func TestRelocationRestartAfterServingCASRecoversParticipation(t *testing.T) { |
| 128 | t.Parallel() |
| 129 | ctx := t.Context() |
| 130 | b := newParticipationBox(t) |
| 131 | if err := b.p.Prepare(ctx, b.read()); err != nil { // acknowledged at installing |
| 132 | t.Fatal(err) |
| 133 | } |
| 134 | slot := b.p.slots[participationHost] |
| 135 | _, target, receipt := b.probeStopped() |
| 136 | serving := b.serve(receipt) |
| 137 | // The crash: the process ends and its controller with it; the record |
| 138 | // stays. The restarted agent has no slot. |
| 139 | _ = slot.Controller.Close(ctx) |
| 140 | b.p.mu.Lock() |
| 141 | b.p.slots = map[string]*claimParticipation{} |
| 142 | b.p.mu.Unlock() |
| 143 | if len(b.records()) != 1 { |
| 144 | t.Fatalf("records after the crash: %v", b.records()) |
| 145 | } |
| 146 | if err := b.p.DischargeCompleted(ctx, serving); err != nil { |
| 147 | t.Fatalf("discharge after the restart: %v", err) |
| 148 | } |
| 149 | if b.stops != 0 || len(b.records()) != 0 { |
| 150 | t.Fatalf("stops %d records %v", b.stops, b.records()) |
| 151 | } |
| 152 | h := serving.Claim.Holder |
| 153 | got, err := b.guard.CompletedWriterStops(ctx, participationHost, hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp}, "rel-1", "dst") |
| 154 | if err != nil || len(got) != 1 || got[0].Target != target { |
| 155 | t.Fatalf("the probe's receipt changed: %+v %v", got, err) |
| 156 | } |
| 157 | if err := b.servingCheck(b.read()); err != nil { |
| 158 | t.Fatalf("the serving Check: %v", err) |
| 159 | } |
| 160 | // The first serving wake: a cold admission at session and stamp 40; |
| 161 | // the guard's handover barrier over the probe's receipt is accepted. |
| 162 | admitted := b.update(func(c *hostclaim.Claim) error { |
| 163 | c.Holder.Stamp, c.Holder.Session, c.Holder.Generation = 40, 40, "g40" |
| 164 | c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Session: 40, Box: "dst", Generation: "g40", Predecessor: "P1", Mode: "cold", Region: "ie"}) |
| 165 | return nil |
| 166 | }) |
| 167 | barrier := target |
| 168 | barrier.FenceID, barrier.Stamp, barrier.Session, barrier.Op, barrier.Phase, barrier.ObligationID, barrier.ClaimETag = "handover-1", 40, 40, "", "", "", admitted.ETag |
| 169 | if err := b.guard.BeginColdHandover(ctx, target, barrier, func(*hostclaim.Claim) error { return nil }); err != nil { |
| 170 | t.Fatalf("the serving handover from the probe's receipt: %v", err) |
| 171 | } |
| 172 | } |
| 173 | |
| 174 | // Absence is never evidence: after the restart, a completion record that |
| 175 | // is present but cannot be read is not "no completion". C10 ends the |
| 176 | // cleanup with that error before anything else: main's cleanup, which |
| 177 | // would ask root to stop the probe's journaled launch again (replacing |
| 178 | // its fence), never runs, nothing is stopped, and the record stays for |
| 179 | // the next pass. The probe's receipt is as it was. |
| 180 | func TestRelocationRestartDischargeRefusesAnUnreadableCompletion(t *testing.T) { |
| 181 | t.Parallel() |
| 182 | ctx := t.Context() |
| 183 | b := newParticipationBox(t) |
| 184 | if err := b.p.Prepare(ctx, b.read()); err != nil { // acknowledged at installing |
| 185 | t.Fatal(err) |
| 186 | } |
| 187 | slot := b.p.slots[participationHost] |
| 188 | _, target, _ := b.probeStopped() |
| 189 | serving := b.update(func(c *hostclaim.Claim) error { |
| 190 | c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "dst", Participants: []string{"dst"}, Journal: map[string]string{hostclaim.JournalCompleted: "{not json"}} |
| 191 | return nil |
| 192 | }) |
| 193 | _ = slot.Controller.Close(ctx) |
| 194 | b.p.mu.Lock() |
| 195 | b.p.slots = map[string]*claimParticipation{} |
| 196 | b.p.mu.Unlock() |
| 197 | if err := b.p.DischargeCompleted(ctx, serving); !errors.Is(err, hostclaim.ErrRelocation) { |
| 198 | t.Fatalf("an unreadable completion was not refused: %v", err) |
| 199 | } |
| 200 | if b.stops != 0 || len(b.records()) != 1 { |
| 201 | t.Fatalf("stops %d records %v", b.stops, b.records()) |
| 202 | } |
| 203 | h := serving.Claim.Holder |
| 204 | got, err := b.guard.CompletedWriterStops(ctx, participationHost, hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp}, "rel-1", "dst") |
| 205 | if err != nil || len(got) != 1 || got[0].Target != target { |
| 206 | t.Fatalf("the probe's receipt changed: %+v %v", got, err) |
| 207 | } |
| 208 | } |
| 209 | |
| 210 | // C10's text: the discharge is by the persisted binding's exact op and |
| 211 | // incarnation. A binding for another op, or another incarnation, never |
| 212 | // discharges a completed relocation its participation was not part of: |
| 213 | // C10 falls through and main's cleanup answers for that binding (here it |
| 214 | // refuses: the box's fixture has no root to stop through), so the cleanup |
| 215 | // never succeeds on its account. The participation's own binding does. |
| 216 | func TestRelocationDischargeIsKeyedByTheBindingsOpAndIncarnation(t *testing.T) { |
| 217 | t.Parallel() |
| 218 | ctx := t.Context() |
| 219 | b := newParticipationBox(t) |
| 220 | if err := b.p.Prepare(ctx, b.read()); err != nil { // acknowledged at installing |
| 221 | t.Fatal(err) |
| 222 | } |
| 223 | slot := b.p.slots[participationHost] |
| 224 | _, _, receipt := b.probeStopped() |
| 225 | serving := b.serve(receipt) |
| 226 | _ = slot.Controller.Close(ctx) |
| 227 | own := slot.Config.Binding |
| 228 | for name, change := range map[string]func(*execlease.Binding){ |
| 229 | "another op": func(x *execlease.Binding) { x.Op = "rel-0" }, |
| 230 | "another incarnation": func(x *execlease.Binding) { x.Incarnation = "other" }, |
| 231 | } { |
| 232 | binding := own |
| 233 | change(&binding) |
| 234 | if err := b.p.cleanup(ctx, serving, binding); err == nil { |
| 235 | t.Fatalf("%s: the completed relocation was discharged by a binding that is not its participation's", name) |
| 236 | } |
| 237 | } |
| 238 | if err := b.p.cleanup(ctx, serving, own); err != nil { |
| 239 | t.Fatalf("the participation's own binding: %v", err) |
| 240 | } |
| 241 | if b.stops != 0 { |
| 242 | t.Fatalf("root was asked for %d stops", b.stops) |
| 243 | } |
| 244 | } |
| 245 | |
amadan