| 1 | //go:build unix |
| 2 | |
| 3 | package host |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "strings" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/activator" |
| 16 | "github.com/carlosframework/platform/internal/hostbox" |
| 17 | "github.com/carlosframework/platform/internal/hostclaim" |
| 18 | "github.com/carlosframework/platform/internal/hostguard" |
| 19 | "github.com/carlosframework/platform/internal/registry" |
| 20 | "github.com/carlosframework/platform/internal/release" |
| 21 | ) |
| 22 | |
| 23 | // failingClaimPut is a control bucket whose next fail writes of one claim |
| 24 | // key fail: a lost or failed claim CAS. |
| 25 | type failingClaimPut struct { |
| 26 | release.Store |
| 27 | key string |
| 28 | fail int |
| 29 | } |
| 30 | |
| 31 | func (s *failingClaimPut) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| 32 | if key == s.key && s.fail > 0 { |
| 33 | s.fail-- |
| 34 | return "", errors.New("the claim CAS failed") |
| 35 | } |
| 36 | return s.Store.Put(ctx, key, body, ifMatch) |
| 37 | } |
| 38 | |
| 39 | // finishWorld is a host that serves on dst after relocation rel-1 |
| 40 | // completed there, undischarged: its row held for rel-1, the probe stop's |
| 41 | // owned fence on the activator, and the pass of dst with the production |
| 42 | // Finish (relocationReleaseHold). |
| 43 | type finishWorld struct { |
| 44 | t *testing.T |
| 45 | store *failingClaimPut |
| 46 | reg *registry.Registry |
| 47 | act *stopProbeActivator |
| 48 | order []string |
| 49 | discharged int |
| 50 | discharge error |
| 51 | runtime *RelocationRuntime |
| 52 | dataDir string |
| 53 | assembly *ClaimAssembly |
| 54 | } |
| 55 | |
| 56 | const finishHost = "shop.example" |
| 57 | |
| 58 | func newFinishWorld(t *testing.T) *finishWorld { |
| 59 | t.Helper() |
| 60 | ctx := t.Context() |
| 61 | w := &finishWorld{t: t, dataDir: t.TempDir()} |
| 62 | created := time.Date(2026, 10, 7, 0, 0, 0, 0, time.UTC) |
| 63 | inc, err := hostclaim.Incarnation("acct", "app", finishHost, created) |
| 64 | if err != nil { |
| 65 | t.Fatal(err) |
| 66 | } |
| 67 | done, _ := json.Marshal(hostclaim.CompletedRelocation{Op: "rel-1", Stop: hostclaim.ProbeStop{Box: "dst", Session: 12, Receipt: "fence-1", ReceiptSHA256: strings.Repeat("a", 64)}, Stamp: 9, Phase: hostclaim.PhaseRelocatingProbing, Generation: "P1"}) |
| 68 | c := &hostclaim.Claim{V: 2, Gen: 31, Host: finishHost, |
| 69 | Holder: &hostclaim.Holder{Family: "instance", Key: "apps/acct/app/instances/" + finishHost + ".json", Account: "acct", App: "app", Incarnation: inc, CreatedAt: created, |
| 70 | Location: "instances/" + finishHost, Stamp: 9, Session: 12, Generation: "P1", WriterBox: "dst", Admissions: []hostclaim.AdmissionRecord{{Session: 12, Box: "dst", Generation: "P1", Mode: "cold", Region: "ie"}}}, |
| 71 | Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "dst", Participants: []string{"dst"}, Journal: map[string]string{hostclaim.JournalCompleted: string(done)}}} |
| 72 | key, _ := hostclaim.Key(finishHost) |
| 73 | w.store = &failingClaimPut{Store: release.NewDirStore(t.TempDir()), key: key} |
| 74 | raw, _ := json.Marshal(c) |
| 75 | if _, err := w.store.Put(ctx, key, raw, ""); err != nil { |
| 76 | t.Fatal(err) |
| 77 | } |
| 78 | if w.reg, err = registry.Open(filepath.Join(t.TempDir(), "registry.db")); err != nil { |
| 79 | t.Fatal(err) |
| 80 | } |
| 81 | t.Cleanup(func() { w.reg.Close() }) |
| 82 | if err := w.reg.Upsert(registry.Route{Host: finishHost, Status: "relocating", HoldOp: "rel-1", Socket: "/run/shop.sock"}); err != nil { |
| 83 | t.Fatal(err) |
| 84 | } |
| 85 | marker, err := relocationProbeFencePath(w.dataDir, "rel-1") |
| 86 | if err != nil { |
| 87 | t.Fatal(err) |
| 88 | } |
| 89 | owner, err := relocationProbeFenceOwner(marker, finishHost, 12) |
| 90 | if err != nil { |
| 91 | t.Fatal(err) |
| 92 | } |
| 93 | w.act = newStopProbeActivator(&w.order) |
| 94 | fctx, cancel := context.WithTimeout(ctx, time.Second) |
| 95 | defer cancel() |
| 96 | if _, err := w.act.svc.FenceAndStopAs(fctx, finishHost, 12, owner); err != nil && !errors.Is(err, activator.ErrStopUnproven) { |
| 97 | t.Fatal(err) |
| 98 | } |
| 99 | a := ClaimAssembly{Claims: w.store, Guard: hostguard.Guard{DataDir: w.dataDir}, stopLocal: func(context.Context) (claimStopLocal, error) { |
| 100 | return claimStopLocal{Actor: hostbox.Actor{Box: "dst"}}, nil |
| 101 | }} |
| 102 | w.assembly = &a |
| 103 | cfg := RelocationAssemblyConfig{Registry: w.reg, Activator: w.act, DataDir: w.dataDir, Discharge: func(context.Context, hostclaim.Observed) error { |
| 104 | w.discharged++ |
| 105 | return w.discharge |
| 106 | }} |
| 107 | var steps []string |
| 108 | w.runtime = &RelocationRuntime{Worker: stepRecorder{name: "worker", order: &steps}, Participant: stepRecorder{name: "participant", order: &steps}, Box: "dst", |
| 109 | Finish: a.relocationReleaseHold(cfg), DataDir: w.dataDir} |
| 110 | return w |
| 111 | } |
| 112 | |
| 113 | func (w *finishWorld) pass() { w.runtime.once(w.t.Context(), w.store) } |
| 114 | |
| 115 | func (w *finishWorld) held() bool { |
| 116 | row, err := w.reg.Get(finishHost) |
| 117 | return err == nil && row != nil && row.Status == "relocating" && row.HoldOp == "rel-1" |
| 118 | } |
| 119 | |
| 120 | // marked reads the record, and whether a successor (a move from dst) may |
| 121 | // start: the completion barrier. |
| 122 | func (w *finishWorld) marked() (bool, error) { |
| 123 | obs, err := hostclaim.Read(w.t.Context(), w.store, finishHost) |
| 124 | if err != nil { |
| 125 | w.t.Fatal(err) |
| 126 | } |
| 127 | done, ok, err := hostclaim.ReadCompletedRelocation(obs.Claim) |
| 128 | if err != nil { |
| 129 | w.t.Fatal(err) |
| 130 | } |
| 131 | return ok && done.Discharged, hostclaim.StartGuard(obs.Claim, []string{"uk"}, []string{"spa"}, "") |
| 132 | } |
| 133 | |
| 134 | // done asserts the finished state: discharged as many times in all as |
| 135 | // the passes asked (Task 9 review M2: once the hold is released, a replay |
| 136 | // discharges again rather than inferring the discharge ran), the hold |
| 137 | // released, no fence, the completion marked so a successor may start, and |
| 138 | // the op's journals swept by the same pass. |
| 139 | func (w *finishWorld) done(discharges int) { |
| 140 | w.t.Helper() |
| 141 | ok, barrier := w.marked() |
| 142 | if w.discharged != discharges || w.held() || w.act.svc.Fenced(finishHost) || !ok || barrier != nil { |
| 143 | w.t.Fatalf("discharged %d (want %d) held=%v fenced=%v marked=%v barrier=%v", w.discharged, discharges, w.held(), w.act.svc.Fenced(finishHost), ok, barrier) |
| 144 | } |
| 145 | if _, err := os.Stat(filepath.Join(w.dataDir, "relocate", "ops", "rel-1")); !errors.Is(err, os.ErrNotExist) { |
| 146 | w.t.Fatalf("the finished op's journals survived: %v", err) |
| 147 | } |
| 148 | } |
| 149 | |
| 150 | // Astra round 7, finding 1 (ruling 80): a completion is finished from the |
| 151 | // claim's record on every pass, whatever the row says, and every step |
| 152 | // resumes after a failure or a crash at any point, through the pass. |
| 153 | func TestRelocatePassFinishesACompletionFromItsRecord(t *testing.T) { |
| 154 | t.Parallel() |
| 155 | t.Run("in one pass", func(t *testing.T) { |
| 156 | w := newFinishWorld(t) |
| 157 | w.pass() |
| 158 | w.done(1) |
| 159 | w.pass() |
| 160 | w.done(1) |
| 161 | }) |
| 162 | t.Run("a failed discharge keeps everything, the next pass finishes", func(t *testing.T) { |
| 163 | w := newFinishWorld(t) |
| 164 | w.discharge = errors.New("the participation slot is still open") |
| 165 | w.pass() |
| 166 | if ok, barrier := w.marked(); w.discharged != 1 || !w.held() || w.act.svc.FenceGeneration(finishHost) != 12 || ok || !errors.Is(barrier, hostclaim.ErrPendingChore) { |
| 167 | t.Fatalf("discharged %d held=%v fence=%d marked=%v barrier=%v", w.discharged, w.held(), w.act.svc.FenceGeneration(finishHost), ok, barrier) |
| 168 | } |
| 169 | w.discharge = nil |
| 170 | w.pass() |
| 171 | w.done(2) |
| 172 | }) |
| 173 | t.Run("the mark's CAS fails, the next pass marks", func(t *testing.T) { |
| 174 | w := newFinishWorld(t) |
| 175 | w.store.fail = 1 |
| 176 | w.pass() |
| 177 | if ok, barrier := w.marked(); w.discharged != 1 || w.held() || w.act.svc.Fenced(finishHost) || ok || !errors.Is(barrier, hostclaim.ErrPendingChore) { |
| 178 | t.Fatalf("discharged %d held=%v fenced=%v marked=%v barrier=%v", w.discharged, w.held(), w.act.svc.Fenced(finishHost), ok, barrier) |
| 179 | } |
| 180 | // The row is no longer held: the record alone brings it back, and |
| 181 | // the released-hold branch discharges again (Task 9 review M2). |
| 182 | w.pass() |
| 183 | w.done(2) |
| 184 | }) |
| 185 | t.Run("the lift never ran after the hold went, the next pass lifts", func(t *testing.T) { |
| 186 | w := newFinishWorld(t) |
| 187 | // An earlier call discharged and released the hold, then failed. |
| 188 | w.discharged = 1 |
| 189 | if _, err := w.reg.ReleaseHold(finishHost, "rel-1"); err != nil { |
| 190 | t.Fatal(err) |
| 191 | } |
| 192 | w.pass() |
| 193 | w.done(2) |
| 194 | }) |
| 195 | t.Run("a restart after the hold release", func(t *testing.T) { |
| 196 | w := newFinishWorld(t) |
| 197 | w.discharged = 1 |
| 198 | if _, err := w.reg.ReleaseHold(finishHost, "rel-1"); err != nil { |
| 199 | t.Fatal(err) |
| 200 | } |
| 201 | w.act.svc = &activator.Service{} // the restart: the fence went with the process |
| 202 | w.pass() |
| 203 | w.done(2) |
| 204 | }) |
| 205 | } |
| 206 | |
| 207 | // productionLocal is main's participation owner after the restart, bound |
| 208 | // to the world's assembly: DischargeCompleted is the production discharge |
| 209 | // (carry 6), its identity a verifying one for dst. |
| 210 | func (w *finishWorld) productionLocal() *productionClaimLocal { |
| 211 | identity := handoverRecoveryIdentity(w.t, hostbox.Actor{Box: "dst", Enrollment: "enr", Region: "ie", Revision: 1, Binding: "1"}) |
| 212 | return &productionClaimLocal{legacyParticipation: true, Resources: ClaimRuntimeResources{Mode: "agent", DataDir: w.dataDir}, Identity: identity, Store: w.store, |
| 213 | slots: map[string]*claimParticipation{}, assembly: w.assembly, lifetime: w.t.Context()} |
| 214 | } |
| 215 | |
| 216 | // Astra round 8 (ruling 82): A to B completes, the mark's CAS fails, and B |
| 217 | // restarts with CARLOS_REGION_HA off. Claim mode alone builds the |
| 218 | // completion-only runtime, which steps nothing, and its pass still lifts |
| 219 | // and marks, so the barrier opens. The restarted runtime's discharge is |
| 220 | // main's own (productionClaimLocal.DischargeCompleted, carry 6), which |
| 221 | // finds no slot and no .hostexec record. |
| 222 | func TestCompletionPassFinishesWithRegionHAOff(t *testing.T) { |
| 223 | w := newFinishWorld(t) |
| 224 | w.store.fail = 1 |
| 225 | w.pass() |
| 226 | if ok, barrier := w.marked(); ok || w.held() || !errors.Is(barrier, hostclaim.ErrPendingChore) { |
| 227 | t.Fatalf("the first pass marked or kept the hold: marked=%v held=%v barrier=%v", ok, w.held(), barrier) |
| 228 | } |
| 229 | w.act.svc = &activator.Service{} // the restart |
| 230 | claimOnly := func(k string) string { |
| 231 | if k == "CARLOS_HOST_CLAIM" { |
| 232 | return "1" |
| 233 | } |
| 234 | return "" |
| 235 | } |
| 236 | r, err := NewRelocationRuntime(t.Context(), claimOnly, w.store, w.productionLocal(), ClaimProviders{}, w.reg, w.act, relocationRuntimeInputs{}) |
| 237 | if err != nil || r == nil || r.Worker != nil || r.Participant != nil || r.Convert != nil || r.Finish == nil { |
| 238 | t.Fatalf("not a completion-only runtime: %+v %v", r, err) |
| 239 | } |
| 240 | w.runtime = r |
| 241 | w.pass() |
| 242 | w.done(1) |
| 243 | } |
| 244 | |
| 245 | // Carry 6: a probe-fence owner the hold release cannot find is ErrRefused, |
| 246 | // which the pass alerts and retries; it never skips the completion and |
| 247 | // never marks it. Nothing is released while the owner is missing; once it |
| 248 | // is back, the next pass finishes. |
| 249 | func TestRelocatePassRetriesAMissingProbeFenceOwner(t *testing.T) { |
| 250 | w := newFinishWorld(t) |
| 251 | marker, err := relocationProbeFencePath(w.dataDir, "rel-1") |
| 252 | if err != nil { |
| 253 | t.Fatal(err) |
| 254 | } |
| 255 | raw, err := os.ReadFile(marker) |
| 256 | if err != nil { |
| 257 | t.Fatal(err) |
| 258 | } |
| 259 | if err := os.Remove(marker); err != nil { |
| 260 | t.Fatal(err) |
| 261 | } |
| 262 | buf := captureRacyLog(t) |
| 263 | w.pass() |
| 264 | w.pass() |
| 265 | if ok, barrier := w.marked(); ok || !w.held() || w.discharged != 0 || w.act.svc.FenceGeneration(finishHost) != 12 || !errors.Is(barrier, hostclaim.ErrPendingChore) { |
| 266 | t.Fatalf("over a missing owner: marked=%v held=%v discharged=%d fence=%d barrier=%v", ok, w.held(), w.discharged, w.act.svc.FenceGeneration(finishHost), barrier) |
| 267 | } |
| 268 | if n := strings.Count(buf.String(), "has no probe-fence owner"); n != 1 { |
| 269 | t.Fatalf("alerted %d times, want once:\n%s", n, buf.String()) |
| 270 | } |
| 271 | // An operator puts the owner back. (The sweep has removed the op's |
| 272 | // directory meanwhile: with the marker gone it held no journal.) |
| 273 | if err := os.MkdirAll(filepath.Dir(marker), 0o700); err != nil { |
| 274 | t.Fatal(err) |
| 275 | } |
| 276 | if err := os.WriteFile(marker, raw, 0o600); err != nil { |
| 277 | t.Fatal(err) |
| 278 | } |
| 279 | w.pass() |
| 280 | w.done(1) |
| 281 | } |
| 282 | |
| 283 | // Review M1: a completion whose row is gone is still visited by the |
| 284 | // completion-only pass (its probe-fence journal names the host), and the |
| 285 | // hold release's refusal is said once; nothing is released or marked. |
| 286 | func TestCompletionPassSaysACompletionWhoseRowIsGone(t *testing.T) { |
| 287 | w := newFinishWorld(t) |
| 288 | if err := w.reg.Delete(finishHost); err != nil { |
| 289 | t.Fatal(err) |
| 290 | } |
| 291 | claimOnly := func(k string) string { |
| 292 | if k == "CARLOS_HOST_CLAIM" { |
| 293 | return "1" |
| 294 | } |
| 295 | return "" |
| 296 | } |
| 297 | r, err := NewRelocationRuntime(t.Context(), claimOnly, w.store, w.productionLocal(), ClaimProviders{}, w.reg, w.act, relocationRuntimeInputs{}) |
| 298 | if err != nil { |
| 299 | t.Fatal(err) |
| 300 | } |
| 301 | w.runtime = r |
| 302 | buf := captureRacyLog(t) |
| 303 | w.pass() |
| 304 | w.pass() |
| 305 | if n := strings.Count(buf.String(), "has no row on this box"); n != 1 { |
| 306 | t.Fatalf("said %d times, want once:\n%s", n, buf.String()) |
| 307 | } |
| 308 | if ok, _ := w.marked(); ok || !w.act.svc.Fenced(finishHost) { |
| 309 | t.Fatalf("finished over a missing row: marked=%v fenced=%v", ok, w.act.svc.Fenced(finishHost)) |
| 310 | } |
| 311 | } |
| 312 | |
amadan