| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "net" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "strings" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/hostbox" |
| 15 | "github.com/carlosframework/platform/internal/hostclaim" |
| 16 | "github.com/carlosframework/platform/internal/hostguard" |
| 17 | "github.com/carlosframework/platform/internal/instancectl" |
| 18 | "github.com/carlosframework/platform/internal/registry" |
| 19 | ) |
| 20 | |
| 21 | func retirementRecoveryRow(t *testing.T, f *recoveryFixture) registry.Route { |
| 22 | t.Helper() |
| 23 | o := f.read() |
| 24 | var err error |
| 25 | o, err = hostclaim.Update(t.Context(), f.claims, o, hostclaim.BeginDelete("delete-1", "console", `"record-etag"`, f.now, nil, nil, "")) |
| 26 | if err != nil { |
| 27 | t.Fatal(err) |
| 28 | } |
| 29 | o, err = hostclaim.Update(t.Context(), f.claims, o, hostclaim.ToParking("delete-1", nil, nil, f.newID)) |
| 30 | if err != nil { |
| 31 | t.Fatal(err) |
| 32 | } |
| 33 | |
| 34 | h := o.Claim.Holder |
| 35 | return registry.Route{Host: o.Host, ClaimOwnership: registry.ClaimOwnership{ClaimIncarnation: h.Incarnation, ClaimStamp: h.Stamp, ClaimOwner: "installation", ClaimRowID: "route-1"}} |
| 36 | } |
| 37 | |
| 38 | func recordRetirementSuccessor(t *testing.T, f *recoveryFixture, admitted hostclaim.Observed, actor hostbox.Actor) (hostguard.FenceTarget, hostbox.LaunchReservation) { |
| 39 | t.Helper() |
| 40 | successor := recoveryTuple(admitted, claimStopLocal{Actor: actor, BootID: "boot-1", ControllerID: "controller-2"}) |
| 41 | successor.Session++ |
| 42 | successor.Stamp++ |
| 43 | successor.DirectoryRevision = "8" |
| 44 | if err := f.journal.Prelaunch(t.Context(), hostbox.LaunchInventory{ControllerPID: 4202, ProcessStart: "start-2", Target: successor, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/retirement-successor", Device: 1, Inode: 99}}, NoState: true}); err != nil { |
| 45 | t.Fatal(err) |
| 46 | } |
| 47 | reservation := hostbox.LaunchReservation{ID: "successor:main", Role: "main", InvocationID: "successor-invocation", Inputs: "inputs", Target: successor, Group: hostbox.LaunchGroup{RelativePath: "carlos/retirement-successor/role", Device: 1, Inode: 100, InvocationID: "successor-invocation"}, Executable: hostbox.LaunchFile{Device: 1, Inode: 101}} |
| 48 | if err := f.journal.Reserve(t.Context(), reservation); err != nil { |
| 49 | t.Fatal(err) |
| 50 | } |
| 51 | return successor, reservation |
| 52 | } |
| 53 | |
| 54 | func retirementRecoveryReceipts(t *testing.T, f *recoveryFixture) []StopReceipt { |
| 55 | t.Helper() |
| 56 | keys, err := f.evidence.List(t.Context(), "control/hosts/"+recoveryHost+"/evidence/delete-1/box-1/") |
| 57 | if err != nil { |
| 58 | t.Fatal(err) |
| 59 | } |
| 60 | out := make([]StopReceipt, 0, len(keys)) |
| 61 | for _, key := range keys { |
| 62 | if !strings.HasSuffix(key, ".json") { |
| 63 | continue |
| 64 | } |
| 65 | body, _, err := f.evidence.Get(t.Context(), key) |
| 66 | if err != nil { |
| 67 | t.Fatal(err) |
| 68 | } |
| 69 | var receipt StopReceipt |
| 70 | if err := json.Unmarshal(body, &receipt); err != nil { |
| 71 | t.Fatal(err) |
| 72 | } |
| 73 | out = append(out, receipt) |
| 74 | } |
| 75 | return out |
| 76 | } |
| 77 | |
| 78 | func startRetirementRecoveryTransport(t *testing.T, handler ClaimRootHTTP) string { |
| 79 | t.Helper() |
| 80 | directory, err := os.MkdirTemp("", "retire-") |
| 81 | if err != nil { |
| 82 | t.Fatal(err) |
| 83 | } |
| 84 | t.Cleanup(func() { os.RemoveAll(directory) }) |
| 85 | socket := filepath.Join(directory, "s") |
| 86 | listener, err := net.Listen("unix", socket) |
| 87 | if err != nil { |
| 88 | t.Fatal(err) |
| 89 | } |
| 90 | ctx, cancel := context.WithCancel(t.Context()) |
| 91 | done := make(chan error, 1) |
| 92 | server := instancectl.Server{ClaimStop: handler, PeerOK: func(net.Conn) error { return nil }} |
| 93 | go func() { done <- server.Serve(ctx, listener) }() |
| 94 | t.Cleanup(func() { |
| 95 | cancel() |
| 96 | _ = listener.Close() |
| 97 | if err := <-done; err != nil { |
| 98 | t.Errorf("retirement transport: %v", err) |
| 99 | } |
| 100 | }) |
| 101 | return socket |
| 102 | } |
| 103 | |
| 104 | func TestClaimRetirementRecoversPersistedLaunchOwner(t *testing.T) { |
| 105 | t.Parallel() |
| 106 | f := newRecoveryFixture(t) |
| 107 | former := f.controller("boot-1", "controller-1", 4101, "start-1") |
| 108 | admitted := former.admit(t, f) |
| 109 | successor, reservation := recordRetirementSuccessor(t, f, admitted, former.local.Actor) |
| 110 | row := retirementRecoveryRow(t, f) |
| 111 | |
| 112 | f.revision = 8 |
| 113 | restarted := f.controller("boot-1", "controller-2", 4202, "start-2") |
| 114 | f.kernel.alive = true |
| 115 | if _, err := restarted.assembly.collectRetirementStop(t.Context(), row); !errors.Is(err, ErrClaimStopControllerAlive) { |
| 116 | t.Fatalf("live launch owner retired = %v", err) |
| 117 | } |
| 118 | if len(f.requests) != 1 { |
| 119 | t.Fatalf("stop requests before death proof = %+v", f.requests) |
| 120 | } |
| 121 | request := f.requests[0] |
| 122 | if !request.Recover || request.BootID != "boot-1" || request.ControllerID != "controller-1" || request.ExpectedDirectoryRevision != "7" { |
| 123 | t.Fatalf("retirement substituted the restarted controller tuple: %+v", request) |
| 124 | } |
| 125 | if got := retirementRecoveryReceipts(t, f); len(got) != 0 { |
| 126 | t.Fatalf("receipt before root death proof: %+v", got) |
| 127 | } |
| 128 | |
| 129 | f.kernel.alive = false |
| 130 | target, err := restarted.assembly.collectRetirementStop(t.Context(), row) |
| 131 | if err != nil { |
| 132 | t.Fatal(err) |
| 133 | } |
| 134 | if target.BootID != "boot-1" || target.ControllerID != "controller-1" || target.DirectoryRevision != "7" || target.Session != admitted.Claim.Holder.Session { |
| 135 | t.Fatalf("retirement target = %+v", target) |
| 136 | } |
| 137 | if len(f.kernel.verified) != 0 || len(f.kernel.recovered) != 2 || f.kernel.recovered[0].FenceID != f.kernel.recovered[1].FenceID { |
| 138 | t.Fatalf("verified=%v recovered=%v", f.kernel.verified, f.kernel.recovered) |
| 139 | } |
| 140 | other, err := f.journal.Read(t.Context(), successor) |
| 141 | expected := reservation |
| 142 | expected.State = "reserved" |
| 143 | if err != nil || len(other.Reservations) != 1 || other.Reservations[0] != expected { |
| 144 | t.Fatalf("retirement changed successor: %+v %v", other, err) |
| 145 | } |
| 146 | if got := retirementRecoveryReceipts(t, f); len(got) != 1 || got[0].ControllerID != "controller-1" || got[0].DirectoryRevision != "7" || got[0].FenceID != target.FenceID { |
| 147 | t.Fatalf("receipts = %+v", got) |
| 148 | } |
| 149 | |
| 150 | // A bookkeeping CAS changes the claim ETag but not the stop authority. |
| 151 | // Recollection must replay the immutable receipt and retain its old tuple. |
| 152 | current := f.read() |
| 153 | if _, err = hostclaim.Update(t.Context(), f.claims, current, func(*hostclaim.Claim) error { return nil }); err != nil { |
| 154 | t.Fatal(err) |
| 155 | } |
| 156 | recoveredBefore := len(f.kernel.recovered) |
| 157 | replayed, err := restarted.assembly.collectRetirementStop(t.Context(), row) |
| 158 | if err != nil { |
| 159 | t.Fatalf("receipt replay after bookkeeping CAS: %v", err) |
| 160 | } |
| 161 | if replayed != target || len(f.kernel.recovered) != recoveredBefore { |
| 162 | t.Fatalf("replay target=%+v recovered=%d, want %+v and %d", replayed, len(f.kernel.recovered), target, recoveredBefore) |
| 163 | } |
| 164 | |
| 165 | // Retirement acknowledgement reads the claim from the evidence store in |
| 166 | // production. Mirror the fixture's claim there before exercising root HTTP. |
| 167 | claimKey, err := hostclaim.Key(recoveryHost) |
| 168 | if err != nil { |
| 169 | t.Fatal(err) |
| 170 | } |
| 171 | claimBody, _, err := f.claims.Get(t.Context(), claimKey) |
| 172 | if err != nil { |
| 173 | t.Fatal(err) |
| 174 | } |
| 175 | if _, err = f.evidence.Put(t.Context(), claimKey, claimBody, ""); err != nil { |
| 176 | t.Fatal(err) |
| 177 | } |
| 178 | |
| 179 | base := t.TempDir() |
| 180 | inventory, err := openLegacyInventory(t, filepath.Join(base, "inventory.db")) |
| 181 | if err != nil { |
| 182 | t.Fatal(err) |
| 183 | } |
| 184 | t.Cleanup(func() { inventory.Close() }) |
| 185 | reg, err := registry.Open(filepath.Join(base, "registry.db")) |
| 186 | if err != nil { |
| 187 | t.Fatal(err) |
| 188 | } |
| 189 | t.Cleanup(func() { reg.Close() }) |
| 190 | owner := row.ClaimOwnership |
| 191 | in := hostbox.StopInventory{Complete: true, Host: row.Host, Incarnation: owner.ClaimIncarnation, InstallationStamp: owner.ClaimStamp, Box: "box-1", Enrollment: "enrollment-1", DirectoryRevision: "7", Entries: []hostbox.StopInventoryEntry{{Kind: "route", ArtifactID: owner.ClaimRowID, ManagedRoot: "registry", RelativePath: row.Host, RowIdentity: owner.ClaimRowID, Ownership: owner.ClaimOwner, Materialization: "materialized"}}} |
| 192 | if err = inventory.WriteStopInventory(t.Context(), in); err != nil { |
| 193 | t.Fatal(err) |
| 194 | } |
| 195 | successorRow := row |
| 196 | successorRow.ClaimStamp++ |
| 197 | successorRow.ClaimOwner = "successor" |
| 198 | successorRow.ClaimRowID = "route-2" |
| 199 | if err = reg.Upsert(row); err != nil { |
| 200 | t.Fatal(err) |
| 201 | } |
| 202 | coordinator := f.root(recoveryStopDrain{}) |
| 203 | root := ClaimRootHTTP{ |
| 204 | Stop: ClaimStopHTTP{Coordinator: coordinator, Authenticate: func(context.Context, net.Conn) (StopPeer, error) { return f.peer, nil }}, |
| 205 | Roots: map[string]string{"data": base}, |
| 206 | Reconcile: hostbox.ReconcileAdapter{ |
| 207 | Box: "box-1", Enrollment: "enrollment-1", Inventory: inventory, Registry: reg, |
| 208 | Retirement: hostbox.TypedRetirement{JournalDir: filepath.Join(base, "retire")}, |
| 209 | }, |
| 210 | } |
| 211 | socket := startRetirementRecoveryTransport(t, root) |
| 212 | unknown := target |
| 213 | unknown.ControllerID = "unknown-controller" |
| 214 | if err = RetireClaimOverSocket(t.Context(), socket, unknown); !errors.Is(err, ErrClaimStopReceipt) { |
| 215 | t.Fatalf("unknown old target = %v", err) |
| 216 | } |
| 217 | if got, e := reg.Get(row.Host); e != nil || got == nil || got.ClaimRowID != row.ClaimRowID { |
| 218 | t.Fatalf("unknown target changed row: %+v %v", got, e) |
| 219 | } |
| 220 | if err = RetireClaimOverSocket(t.Context(), socket, target); err != nil { |
| 221 | t.Fatalf("retire recovered target over HTTP: %v", err) |
| 222 | } |
| 223 | if got, e := reg.Get(row.Host); e != nil || got != nil { |
| 224 | t.Fatalf("retirement left original route: %+v %v", got, e) |
| 225 | } |
| 226 | if err = reg.Upsert(successorRow); err != nil { |
| 227 | t.Fatal(err) |
| 228 | } |
| 229 | if err = RetireClaimOverSocket(t.Context(), socket, target); err != nil { |
| 230 | t.Fatalf("replay recovered retirement over HTTP: %v", err) |
| 231 | } |
| 232 | if got, e := reg.Get(row.Host); e != nil || got == nil || got.ClaimRowID != successorRow.ClaimRowID { |
| 233 | t.Fatalf("recovered retirement changed successor: %+v %v", got, e) |
| 234 | } |
| 235 | } |
| 236 | |
| 237 | func TestClaimRetirementRefusesMissingOrAmbiguousLaunchJournal(t *testing.T) { |
| 238 | t.Run("missing", func(t *testing.T) { |
| 239 | f := newRecoveryFixture(t) |
| 240 | f.recordPrelaunch = false |
| 241 | former := f.controller("boot-1", "controller-1", 4101, "start-1") |
| 242 | former.admit(t, f) |
| 243 | row := retirementRecoveryRow(t, f) |
| 244 | f.revision = 8 |
| 245 | restarted := f.controller("boot-1", "controller-2", 4202, "start-2") |
| 246 | if _, err := restarted.assembly.collectRetirementStop(t.Context(), row); !errors.Is(err, hostbox.ErrChanged) { |
| 247 | t.Fatalf("missing launch journal = %v", err) |
| 248 | } |
| 249 | if len(f.requests) != 0 || len(f.kernel.verified) != 0 || len(f.kernel.recovered) != 0 { |
| 250 | t.Fatalf("missing journal caused stop effects: requests=%+v verified=%v recovered=%v", f.requests, f.kernel.verified, f.kernel.recovered) |
| 251 | } |
| 252 | if got := retirementRecoveryReceipts(t, f); len(got) != 0 { |
| 253 | t.Fatalf("missing journal produced evidence: %+v", got) |
| 254 | } |
| 255 | }) |
| 256 | |
| 257 | t.Run("ambiguous", func(t *testing.T) { |
| 258 | f := newRecoveryFixture(t) |
| 259 | former := f.controller("boot-1", "controller-1", 4101, "start-1") |
| 260 | admitted := former.admit(t, f) |
| 261 | second := recoveryTuple(admitted, claimStopLocal{Actor: former.local.Actor, BootID: "boot-1", ControllerID: "controller-other"}) |
| 262 | if err := f.journal.Prelaunch(t.Context(), hostbox.LaunchInventory{ControllerPID: 4199, ProcessStart: "start-other", Target: second, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/ambiguous", Device: 1, Inode: 98}}, NoState: true}); err != nil { |
| 263 | t.Fatal(err) |
| 264 | } |
| 265 | row := retirementRecoveryRow(t, f) |
| 266 | f.revision = 8 |
| 267 | restarted := f.controller("boot-1", "controller-2", 4202, "start-2") |
| 268 | if _, err := restarted.assembly.collectRetirementStop(t.Context(), row); !errors.Is(err, hostbox.ErrChanged) { |
| 269 | t.Fatalf("ambiguous launch journal = %v", err) |
| 270 | } |
| 271 | if len(f.requests) != 0 || len(f.kernel.verified) != 0 || len(f.kernel.recovered) != 0 { |
| 272 | t.Fatalf("ambiguous journal caused partial stop: requests=%+v verified=%v recovered=%v", f.requests, f.kernel.verified, f.kernel.recovered) |
| 273 | } |
| 274 | if got := retirementRecoveryReceipts(t, f); len(got) != 0 { |
| 275 | t.Fatalf("ambiguous journal produced evidence: %+v", got) |
| 276 | } |
| 277 | }) |
| 278 | } |
| 279 | |
| 280 | func TestClaimRetirementAfterServingStopSameController(t *testing.T) { |
| 281 | f := newRecoveryFixture(t) |
| 282 | controller := f.controller("boot-1", "controller-1", 4101, "start-1") |
| 283 | admitted := controller.admit(t, f) |
| 284 | if err := controller.wake.Stop(t.Context(), admitted); err != nil { |
| 285 | t.Fatal("serving stop", err) |
| 286 | } |
| 287 | row := retirementRecoveryRow(t, f) |
| 288 | target, err := controller.assembly.collectRetirementStop(t.Context(), row) |
| 289 | if err != nil { |
| 290 | t.Fatal("retirement after serving stop", err) |
| 291 | } |
| 292 | if target.Op != "delete-1" || target.Phase != "park" || target.ObligationID == "" || target.Session != admitted.Claim.Holder.Session { |
| 293 | t.Fatalf("retirement tuple: %+v", target) |
| 294 | } |
| 295 | } |
| 296 | |
| 297 | func TestClaimRetirementCannotReplaceUnfinishedDrain(t *testing.T) { |
| 298 | target := hostguard.FenceTarget{Host: "app.example", Incarnation: "inc", Stamp: 1, Session: 1, Box: "box", Enrollment: "en", BootID: "boot", ControllerID: "controller"} |
| 299 | p, err := NewClaimProducers(target, "start", func(context.Context, hostguard.FenceTarget, string) error { return nil }) |
| 300 | if err != nil { |
| 301 | t.Fatal(err) |
| 302 | } |
| 303 | finish := make(chan struct{}) |
| 304 | defer func() { |
| 305 | select { |
| 306 | case <-finish: |
| 307 | default: |
| 308 | close(finish) |
| 309 | } |
| 310 | }() |
| 311 | if err = p.Launch(t.Context(), target, "writer", func(ctx context.Context) error { <-ctx.Done(); <-finish; return ctx.Err() }); err != nil { |
| 312 | t.Fatal(err) |
| 313 | } |
| 314 | req := StopRequest{Host: target.Host, Incarnation: target.Incarnation, InstallationStamp: 1, WriterSession: 1, Box: target.Box, Enrollment: target.Enrollment, BootID: target.BootID, ControllerID: target.ControllerID} |
| 315 | short, cancel := context.WithTimeout(t.Context(), 10*time.Millisecond) |
| 316 | _, err = p.DrainClaim(short, req, "serving-fence") |
| 317 | cancel() |
| 318 | if !errors.Is(err, context.DeadlineExceeded) { |
| 319 | t.Fatal("unfinished drain", err) |
| 320 | } |
| 321 | short, cancel = context.WithTimeout(t.Context(), 20*time.Millisecond) |
| 322 | _, err = p.DrainClaim(short, req, "retirement-fence") |
| 323 | cancel() |
| 324 | if !errors.Is(err, ErrClaimStopIdentity) { |
| 325 | t.Fatal("replaced unfinished drain", err) |
| 326 | } |
| 327 | close(finish) |
| 328 | if _, err = p.DrainClaim(t.Context(), req, "serving-fence"); err != nil { |
| 329 | t.Fatal(err) |
| 330 | } |
| 331 | ack, err := p.DrainClaim(t.Context(), req, "retirement-fence") |
| 332 | if err != nil || ack.FenceID != "retirement-fence" { |
| 333 | t.Fatal("completed drain retirement", ack, err) |
| 334 | } |
| 335 | req.WriterSession++ |
| 336 | if _, err = p.DrainClaim(t.Context(), req, "foreign-fence"); !errors.Is(err, ErrClaimStopIdentity) { |
| 337 | t.Fatal("foreign session rebound", err) |
| 338 | } |
| 339 | if err = p.Launch(t.Context(), target, "new-writer", func(context.Context) error { return nil }); err == nil { |
| 340 | t.Fatal("retirement reopened producer admission") |
| 341 | } |
| 342 | } |
| 343 | |
amadan