| 1 | //go:build linux |
| 2 | |
| 3 | package host |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "database/sql" |
| 8 | "encoding/json" |
| 9 | "errors" |
| 10 | "net" |
| 11 | "path/filepath" |
| 12 | "strconv" |
| 13 | "strings" |
| 14 | "testing" |
| 15 | |
| 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/provision" |
| 20 | "github.com/carlosframework/platform/internal/registry" |
| 21 | ) |
| 22 | |
| 23 | // rootPeer is the fixture's live requester: the agent's process tuple. |
| 24 | // ServeHTTP authenticates it with SO_PEERCRED and the controller registry in |
| 25 | // production, which this test host does not have. |
| 26 | var rootPeer = StopPeer{UID: 1001, PID: 42, ProcessStart: "start-42", BootID: "boot-1", ControllerID: "controller-1"} |
| 27 | |
| 28 | // serve puts root's current handler on a Unix socket, as instancectl does, |
| 29 | // with peer authentication fixed to rootPeer. A restart needs a new socket. |
| 30 | func (w *assemblyWorld) serve() string { |
| 31 | w.t.Helper() |
| 32 | root := w.root |
| 33 | root.Stop.Authenticate = func(context.Context, net.Conn) (StopPeer, error) { return rootPeer, nil } |
| 34 | return startRetirementRecoveryTransport(w.t, root) |
| 35 | } |
| 36 | |
| 37 | // recoveryAgent is the agent's cross-box handover as claimconstruction.go |
| 38 | // builds it in agent mode: root's recovery rebind, inventory and fence view |
| 39 | // all through root's socket, the production identity, the box's registry. |
| 40 | func (w *assemblyWorld) recoveryAgent(socket string, sent *recoveryRebindRequest) claimCrossBoxHandover { |
| 41 | w.t.Helper() |
| 42 | _, inventory := hostbox.RemoteJournals(rootJournalClient(socket)) |
| 43 | guard := hostguard.Guard{DataDir: w.r.DataDir, LockDir: w.locks, Claims: hostguard.StoreReader{Store: w.store}, FenceView: rootFenceViewClient(rootJournalClient(socket))} |
| 44 | client := recoveryRebindClient(socket) |
| 45 | return claimCrossBoxHandover{RecoveryRebind: func(ctx context.Context, q recoveryRebindRequest) error { |
| 46 | if sent != nil { |
| 47 | *sent = q |
| 48 | } |
| 49 | return client(ctx, q) |
| 50 | }, Guard: guard, Evidence: w.store, Box: "box-1", Inventory: inventory, Identity: w.owners.base.Identity, Registry: w.reg} |
| 51 | } |
| 52 | |
| 53 | // recoveryAdmission moves host's claim into a recovery this box executes |
| 54 | // (a restoring op at probing whose source stamp is the installed one) by a |
| 55 | // TEST-ONLY claim transition, then admits a cold writer through the real |
| 56 | // hostclaim.AdmitWriter. It returns the claim before and after admission. |
| 57 | func (w *assemblyWorld) recoveryAdmission(host string) (prior, admitted hostclaim.Observed) { |
| 58 | w.t.Helper() |
| 59 | ctx := w.t.Context() |
| 60 | o := w.claim(host) |
| 61 | stamp := o.Claim.Holder.Stamp |
| 62 | prior, err := hostclaim.Update(ctx, w.store, o, func(c *hostclaim.Claim) error { |
| 63 | c.Op = plan10J2Op(hostclaim.KindRestoring, hostclaim.PhaseProbing) |
| 64 | c.Op.Journal = map[string]string{hostclaim.JournalSourceStamp: strconv.FormatUint(stamp, 10)} |
| 65 | return nil |
| 66 | }) |
| 67 | if err != nil { |
| 68 | w.t.Fatal(err) |
| 69 | } |
| 70 | h := prior.Claim.Holder |
| 71 | admitted, err = hostclaim.Update(ctx, w.store, prior, hostclaim.AdmitWriter(hostclaim.Admission{ObservedOpID: prior.Claim.Op.ID, ObservedPhase: prior.Claim.Op.Phase, ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: "box-1", Mode: "cold", NewGeneration: "recovered", Region: "ie", PredecessorRegion: "ie"})) |
| 72 | if err != nil { |
| 73 | w.t.Fatal(err) |
| 74 | } |
| 75 | if admitted.Claim.Holder.Stamp <= stamp { |
| 76 | w.t.Fatal("admission did not move the stamp") |
| 77 | } |
| 78 | return prior, admitted |
| 79 | } |
| 80 | |
| 81 | // restampInventory rewrites ONLY the stamp of host's installed inventory in |
| 82 | // root's journal database, leaving its revision and every other field: the |
| 83 | // one fact the predecessor check reads. |
| 84 | func (w *assemblyWorld) restampInventory(host string, captured hostguard.Captured, stamp string) { |
| 85 | w.t.Helper() |
| 86 | db, err := sql.Open("sqlite", w.journal) |
| 87 | if err != nil { |
| 88 | w.t.Fatal(err) |
| 89 | } |
| 90 | defer db.Close() |
| 91 | var body []byte |
| 92 | if err = db.QueryRow("SELECT body FROM stop_inventory WHERE host=? AND incarnation=? AND stamp=?", host, captured.Incarnation, captured.Stamp).Scan(&body); err != nil { |
| 93 | w.t.Fatal(err) |
| 94 | } |
| 95 | var in hostbox.StopInventory |
| 96 | if err = json.Unmarshal(body, &in); err != nil { |
| 97 | w.t.Fatal(err) |
| 98 | } |
| 99 | in.DirectoryRevision = stamp |
| 100 | if body, err = json.Marshal(in); err != nil { |
| 101 | w.t.Fatal(err) |
| 102 | } |
| 103 | if _, err = db.Exec("UPDATE stop_inventory SET body=? WHERE host=? AND incarnation=? AND stamp=?", body, host, captured.Incarnation, captured.Stamp); err != nil { |
| 104 | w.t.Fatal(err) |
| 105 | } |
| 106 | } |
| 107 | |
| 108 | func (w *assemblyWorld) view(host string, c hostguard.Captured) hostguard.FenceView { |
| 109 | w.t.Helper() |
| 110 | v, err := w.assembly.Guard.Root.View(w.t.Context(), host, c) |
| 111 | if err != nil { |
| 112 | w.t.Fatal(err) |
| 113 | } |
| 114 | return v |
| 115 | } |
| 116 | |
| 117 | // successorInventory is root's inventory of admitted's installation, or |
| 118 | // revision 0 when there is none. |
| 119 | func (w *assemblyWorld) successorInventory(admitted hostclaim.Observed) (hostbox.StopInventory, uint64) { |
| 120 | w.t.Helper() |
| 121 | h := admitted.Claim.Holder |
| 122 | in, revision, err := w.assembly.Journals.Inventory.Snapshot(w.t.Context(), admitted.Host, hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp}) |
| 123 | if err != nil { |
| 124 | w.t.Fatal(err) |
| 125 | } |
| 126 | return in, revision |
| 127 | } |
| 128 | |
| 129 | // N4 (roster binding task 6, R3): a recovery rebind through the real |
| 130 | // producer, installRecoveryRemote, and root's serveRecoveryRebind over the |
| 131 | // Unix socket, with real hold churn between the predecessor's install and |
| 132 | // the recovery and a hold held through it. The rebound inventory carries the |
| 133 | // binding. Then the same, interrupted after root's intent is durable and |
| 134 | // finished by a restarted root's RecoverPending (the startup replay of the |
| 135 | // pending rebind intent), then the agent's retry. Then only the predecessor |
| 136 | // inventory's stamp is changed: the rebind is refused at the predecessor |
| 137 | // check, after the successor check has passed. |
| 138 | func TestRemoteRecoveryRebindsSameBinding(t *testing.T) { |
| 139 | ctx := t.Context() |
| 140 | w := newAssemblyWorld(t) |
| 141 | w.mustStart() |
| 142 | b0 := w.fresh() |
| 143 | agent := w.agent(b0) |
| 144 | recA, routeA := w.newHost("a") |
| 145 | recB, routeB := w.newHost("b") |
| 146 | recC, routeC := w.newHost("c") |
| 147 | for _, h := range []struct { |
| 148 | rec provision.Record |
| 149 | route registry.Route |
| 150 | }{{recA, routeA}, {recB, routeB}, {recC, routeC}} { |
| 151 | if err := agent.install(h.rec, h.route); err != nil { |
| 152 | t.Fatal(err) |
| 153 | } |
| 154 | } |
| 155 | w.roster.holdChurn() |
| 156 | w.roster.holdChurn() |
| 157 | held := w.roster.reserveHold() |
| 158 | if actor := w.fresh(); actor.Revision == b0.Revision || actor.Binding != b0.Binding { |
| 159 | t.Fatalf("hold churn: revision %d -> %d, binding moved %v", b0.Revision, actor.Revision, actor.Binding != b0.Binding) |
| 160 | } |
| 161 | |
| 162 | // Success through the producer and root's handler. |
| 163 | socket := w.serve() |
| 164 | prior, admitted := w.recoveryAdmission(recA.Host) |
| 165 | var sent recoveryRebindRequest |
| 166 | if err := w.recoveryAgent(socket, &sent).installRecoveryRemote(ctx, prior, admitted); err != nil { |
| 167 | t.Fatal("recovery rebind after hold churn:", err) |
| 168 | } |
| 169 | if sent.Next.DirectoryRevision != b0.Binding { |
| 170 | t.Fatalf("producer sent successor stamp %q, want the binding", sent.Next.DirectoryRevision) |
| 171 | } |
| 172 | rebound, revision := w.successorInventory(admitted) |
| 173 | if revision != 1 || rebound.DirectoryRevision != b0.Binding || rebound.InstallationStamp != admitted.Claim.Holder.Stamp { |
| 174 | t.Fatalf("rebound inventory: revision %d stamp %q installation %d", revision, rebound.DirectoryRevision, rebound.InstallationStamp) |
| 175 | } |
| 176 | sh := admitted.Claim.Holder |
| 177 | if v := w.view(recA.Host, hostguard.Captured{Incarnation: sh.Incarnation, Stamp: sh.Stamp}); v.Installation.Stamp != sh.Stamp { |
| 178 | t.Fatalf("root installation not rebound: %+v", v.Installation) |
| 179 | } |
| 180 | if row, err := w.reg.Get(recA.Host); err != nil || row == nil || row.ClaimStamp != sh.Stamp { |
| 181 | t.Fatalf("rows not rebound: %+v %v", row, err) |
| 182 | } |
| 183 | w.roster.releaseHold(held) |
| 184 | |
| 185 | // Startup replay: root dies after its rebind intent is durable (the |
| 186 | // intent write succeeds, the commit write fails), restarts, and its |
| 187 | // RecoverPending finishes the intent under the unchanged binding. |
| 188 | prior, admitted = w.recoveryAdmission(recB.Host) |
| 189 | *w.interrupt = 1 |
| 190 | err := w.recoveryAgent(socket, nil).installRecoveryRemote(ctx, prior, admitted) |
| 191 | *w.interrupt = -1 |
| 192 | if err == nil || w.pending() != 1 { |
| 193 | t.Fatalf("rebind was not interrupted after its intent: %v, pending %d", err, w.pending()) |
| 194 | } |
| 195 | w.roster.holdChurn() |
| 196 | if err := w.restart(); err != nil { |
| 197 | t.Fatal(err) |
| 198 | } |
| 199 | if err := w.root.RecoverPending(ctx); err != nil { |
| 200 | t.Fatal("restart with a pending same-binding rebind:", err) |
| 201 | } |
| 202 | if w.pending() != 0 { |
| 203 | t.Fatal("startup did not finish the pending rebind") |
| 204 | } |
| 205 | if in, revision := w.successorInventory(admitted); revision != 1 || in.DirectoryRevision != b0.Binding { |
| 206 | t.Fatalf("replayed rebind: revision %d stamp %q", revision, in.DirectoryRevision) |
| 207 | } |
| 208 | socket = w.serve() |
| 209 | if err := w.recoveryAgent(socket, nil).installRecoveryRemote(ctx, prior, admitted); err != nil { |
| 210 | t.Fatal("agent retry after the startup replay:", err) |
| 211 | } |
| 212 | if row, err := w.reg.Get(recB.Host); err != nil || row == nil || row.ClaimStamp != admitted.Claim.Holder.Stamp { |
| 213 | t.Fatalf("rows not rebound after the replay: %+v %v", row, err) |
| 214 | } |
| 215 | |
| 216 | // Only the predecessor's stamp changes: refused at the predecessor check. |
| 217 | ch := w.claim(recC.Host).Claim.Holder |
| 218 | source := hostguard.Captured{Incarnation: ch.Incarnation, Stamp: ch.Stamp} |
| 219 | w.restampInventory(recC.Host, source, strings.Repeat("d", 64)) |
| 220 | before := w.view(recC.Host, source) |
| 221 | prior, admitted = w.recoveryAdmission(recC.Host) |
| 222 | if err := w.recoveryAgent(socket, &sent).installRecoveryRemote(ctx, prior, admitted); err == nil { |
| 223 | t.Fatal("a predecessor stamped under another binding was rebound") |
| 224 | } |
| 225 | if sent.Next.DirectoryRevision != w.fresh().Binding { |
| 226 | t.Fatal("the successor stamp was not today's binding") |
| 227 | } |
| 228 | if _, revision := w.successorInventory(admitted); revision != 0 || w.pending() != 0 { |
| 229 | t.Fatalf("refusal left successor state: revision %d pending %d", revision, w.pending()) |
| 230 | } |
| 231 | if after := w.view(recC.Host, source); after.Installation != before.Installation { |
| 232 | t.Fatalf("refusal changed root: %+v", after.Installation) |
| 233 | } |
| 234 | // The refusal is the predecessor check's (ErrTypedInventory), not the |
| 235 | // successor check's (ErrChanged). |
| 236 | if err := w.root.rebindRecoveryInventory(ctx, sent); !errors.Is(err, hostbox.ErrTypedInventory) { |
| 237 | t.Fatalf("refused at %v, want the predecessor check", err) |
| 238 | } |
| 239 | } |
| 240 | |
| 241 | // L6 (roster binding R3): recovery never adopts a predecessor stamped under |
| 242 | // another binding. Rebind (remote) and handover recovery (local) refuse a |
| 243 | // predecessor installed at B0 once readmission makes the binding B1, and a |
| 244 | // numeral-stamped predecessor; the local path still adopts a predecessor at |
| 245 | // today's binding after hold churn. A numeral predecessor's receipt (no |
| 246 | // binding field) fails verifyRetirementReceipt, so the operation cold |
| 247 | // handover cannot park it. (The cold handover that parks a B0 predecessor |
| 248 | // with a matching receipt and starts a successor at B1 is N5.) |
| 249 | func TestClaimBindingRecoveryRefusesPredecessor(t *testing.T) { |
| 250 | t.Run("rebind", func(t *testing.T) { |
| 251 | ctx := t.Context() |
| 252 | w := newAssemblyWorld(t) |
| 253 | w.mustStart() |
| 254 | b0 := w.fresh() |
| 255 | agent := w.agent(b0) |
| 256 | recA, routeA := w.newHost("a") |
| 257 | recN, routeN := w.newHost("n") |
| 258 | for _, h := range []struct { |
| 259 | rec provision.Record |
| 260 | route registry.Route |
| 261 | }{{recA, routeA}, {recN, routeN}} { |
| 262 | if err := agent.install(h.rec, h.route); err != nil { |
| 263 | t.Fatal(err) |
| 264 | } |
| 265 | } |
| 266 | nh := w.claim(recN.Host).Claim.Holder |
| 267 | w.restampInventory(recN.Host, hostguard.Captured{Incarnation: nh.Incarnation, Stamp: nh.Stamp}, strconv.FormatUint(b0.Revision, 10)) |
| 268 | w.roster.quarantine("loss_1") |
| 269 | w.roster.readmit() |
| 270 | b1 := w.fresh() |
| 271 | socket := w.serve() |
| 272 | for _, host := range []string{recA.Host, recN.Host} { |
| 273 | var sent recoveryRebindRequest |
| 274 | prior, admitted := w.recoveryAdmission(host) |
| 275 | if err := w.recoveryAgent(socket, &sent).installRecoveryRemote(ctx, prior, admitted); err == nil { |
| 276 | t.Fatalf("%s: a stale predecessor was rebound", host) |
| 277 | } |
| 278 | if sent.Next.DirectoryRevision != b1.Binding { |
| 279 | t.Fatalf("%s: successor stamp %q, want B1", host, sent.Next.DirectoryRevision) |
| 280 | } |
| 281 | if err := w.root.rebindRecoveryInventory(ctx, sent); !errors.Is(err, hostbox.ErrTypedInventory) { |
| 282 | t.Fatalf("%s: refused at %v, want the predecessor check", host, err) |
| 283 | } |
| 284 | if _, revision := w.successorInventory(admitted); revision != 0 { |
| 285 | t.Fatalf("%s: refusal wrote a successor inventory", host) |
| 286 | } |
| 287 | } |
| 288 | }) |
| 289 | |
| 290 | t.Run("handover recovery", func(t *testing.T) { |
| 291 | for _, stale := range []string{"none", "readmission", "numeral"} { |
| 292 | t.Run(stale, func(t *testing.T) { |
| 293 | ctx := t.Context() |
| 294 | roster := newBindingRosterFixture(t, "box-1", "ie") |
| 295 | identity := roster.identity() |
| 296 | b0, err := identity.Verify(ctx) |
| 297 | if err != nil { |
| 298 | t.Fatal(err) |
| 299 | } |
| 300 | data := t.TempDir() |
| 301 | inventory, err := openLegacyInventory(t, filepath.Join(t.TempDir(), "inventory.db")) |
| 302 | if err != nil { |
| 303 | t.Fatal(err) |
| 304 | } |
| 305 | t.Cleanup(func() { inventory.Close() }) |
| 306 | reg := restartTestRegistry(t) |
| 307 | host := "shop.bab.oncarlos.com" |
| 308 | prior := hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Incarnation: "inc-1", Stamp: 4}, Op: hostclaim.Op{Journal: map[string]string{hostclaim.JournalSourceStamp: "4"}}}} |
| 309 | admitted := hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Incarnation: "inc-1", Stamp: 9}}} |
| 310 | route := registry.Route{Host: host, Kind: "instance", DBPath: provision.NestedDBPath(data, host), ClaimOwnership: registry.ClaimOwnership{ClaimIncarnation: "inc-1", ClaimStamp: 4, ClaimOwner: "old-owner", ClaimRowID: "old-row"}} |
| 311 | if err := reg.Upsert(route); err != nil { |
| 312 | t.Fatal(err) |
| 313 | } |
| 314 | stamp := b0.Binding |
| 315 | if stale == "numeral" { |
| 316 | stamp = strconv.FormatUint(b0.Revision, 10) |
| 317 | } |
| 318 | if err := inventory.WriteStopInventory(ctx, hostbox.StopInventory{Complete: true, Host: host, Incarnation: "inc-1", InstallationStamp: 4, Box: "box-1", Enrollment: b0.Enrollment, DirectoryRevision: stamp, Entries: []hostbox.StopInventoryEntry{{Kind: "route", ArtifactID: "route", ManagedRoot: "registry", RelativePath: host, RowIdentity: "old-row", Ownership: "old-owner", Materialization: "materialized"}}}); err != nil { |
| 319 | t.Fatal(err) |
| 320 | } |
| 321 | switch stale { |
| 322 | case "none": |
| 323 | roster.holdChurn() |
| 324 | case "readmission": |
| 325 | roster.quarantine("loss_1") |
| 326 | roster.readmit() |
| 327 | } |
| 328 | h := claimCrossBoxHandover{Guard: hostguard.Guard{DataDir: data}, Box: "box-1", Inventory: inventory, Identity: identity, Registry: reg} |
| 329 | err = h.installRecoveryInventory(ctx, prior, admitted) |
| 330 | _, readErr := inventory.ReadInstallation(ctx, host, "inc-1", 9) |
| 331 | if stale == "none" { |
| 332 | if err != nil { |
| 333 | t.Fatal("handover recovery at today's binding after hold churn:", err) |
| 334 | } |
| 335 | successor, e := inventory.ReadInstallation(ctx, host, "inc-1", 9) |
| 336 | if e != nil || successor.DirectoryRevision != b0.Binding { |
| 337 | t.Fatalf("successor inventory %q %v", successor.DirectoryRevision, e) |
| 338 | } |
| 339 | return |
| 340 | } |
| 341 | if !errors.Is(err, hostbox.ErrTypedInventory) || !errors.Is(readErr, sql.ErrNoRows) { |
| 342 | t.Fatalf("stale predecessor: %v, successor read %v", err, readErr) |
| 343 | } |
| 344 | if row, e := reg.Get(host); e != nil || row.ClaimOwnership != route.ClaimOwnership { |
| 345 | t.Fatalf("refusal rebound rows: %+v %v", row, e) |
| 346 | } |
| 347 | }) |
| 348 | } |
| 349 | }) |
| 350 | |
| 351 | t.Run("numeral receipt", func(t *testing.T) { |
| 352 | target := hostguard.FenceTarget{FenceID: "fence-1", Region: "ie", Host: "shop.bab.oncarlos.com", Incarnation: "inc-1", Stamp: 4, Session: 5, Op: "purge-1", Phase: "stopping", Box: "box-1", Enrollment: strings.Repeat("a", 32), BootID: "boot-1", ControllerID: "controller-1", ClaimETag: "etag", DirectoryRevision: "7", RosterRevision: 7} |
| 353 | pre := StopReceipt{Kind: "stopped", ID: "receipt-1", FenceID: target.FenceID, Region: target.Region, Host: target.Host, Incarnation: target.Incarnation, InstallationStamp: target.Stamp, WriterSession: target.Session, OperationID: target.Op, Phase: target.Phase, Box: target.Box, Enrollment: target.Enrollment, BootID: target.BootID, ControllerID: target.ControllerID, ClaimETag: target.ClaimETag, DirectoryRevision: "7"} |
| 354 | if err := verifyRetirementReceipt(t.Context(), hostguard.FenceReceipt{Target: target, Bytes: mustMarshalStopReceipt(pre), EvidenceAt: pre.EvidenceAt}); !errors.Is(err, ErrClaimStopReceipt) { |
| 355 | t.Fatalf("a receipt without a binding proved a numeral predecessor: %v", err) |
| 356 | } |
| 357 | // Control: the binding field is the only difference. |
| 358 | carried := pre |
| 359 | carried.Binding = target.DirectoryRevision |
| 360 | if err := verifyRetirementReceipt(t.Context(), hostguard.FenceReceipt{Target: target, Bytes: mustMarshalStopReceipt(carried), EvidenceAt: carried.EvidenceAt}); err != nil { |
| 361 | t.Fatalf("control receipt refused: %v", err) |
| 362 | } |
| 363 | }) |
| 364 | } |
| 365 | |
amadan