| 1 | //go:build linux |
| 2 | |
| 3 | package host |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "errors" |
| 8 | "net" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "reflect" |
| 12 | "strings" |
| 13 | "testing" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/hostbox" |
| 17 | "github.com/carlosframework/platform/internal/hostclaim" |
| 18 | "golang.org/x/sys/unix" |
| 19 | ) |
| 20 | |
| 21 | func recoveryPage(t *testing.T, p *servingParkRoot) servingParkRecoveryPage { |
| 22 | t.Helper() |
| 23 | page, err := p.recoverServingParks(t.Context(), "", servingParkRecoveryLimit) |
| 24 | if err != nil || page.More || len(page.Results) != 1 { |
| 25 | t.Fatalf("recovery page: %+v %v", page, err) |
| 26 | } |
| 27 | return page |
| 28 | } |
| 29 | |
| 30 | func TestServingParkStartupResumesRecordedIntent(t *testing.T) { |
| 31 | for _, boundary := range []string{"first-rename", "last-rename", "parked", "retagged", "complete-slot-pending"} { |
| 32 | t.Run(boundary, func(t *testing.T) { |
| 33 | f := newServingRootFixture(t, true) |
| 34 | fs := f.p.ledger.fs.(*servingParkFS) |
| 35 | count := 0 |
| 36 | fs.rename = func(a int, b string, c int, d string, flags uint) error { |
| 37 | err := unix.Renameat2(a, b, c, d, flags) |
| 38 | count++ |
| 39 | if err == nil && (boundary == "first-rename" && count == 1 || boundary == "last-rename" && count == 4) { |
| 40 | return unix.EIO |
| 41 | } |
| 42 | return err |
| 43 | } |
| 44 | a, err := f.park() |
| 45 | if strings.HasSuffix(boundary, "rename") { |
| 46 | if !errors.Is(err, unix.EIO) { |
| 47 | t.Fatalf("lost rename reply: %v", err) |
| 48 | } |
| 49 | a.ID = servingRetirementID(f.prior, f.current, false) |
| 50 | } else if err != nil { |
| 51 | t.Fatal(err) |
| 52 | } |
| 53 | before, err := f.p.ledger.read(a.ID) |
| 54 | if err != nil { |
| 55 | t.Fatal(err) |
| 56 | } |
| 57 | if boundary == "retagged" || boundary == "complete-slot-pending" { |
| 58 | f.tags(f.current.Stamp) |
| 59 | } |
| 60 | if boundary == "complete-slot-pending" { |
| 61 | before.Complete = true |
| 62 | if err = f.p.ledger.save(before); err != nil { |
| 63 | t.Fatal(err) |
| 64 | } |
| 65 | } |
| 66 | claimBefore := f.f.read() |
| 67 | // New authority instance, same durable stores. Startup has no agent |
| 68 | // connection and must not invoke a transport peer callback. |
| 69 | restarted := *f.p |
| 70 | ledger := *f.p.ledger |
| 71 | restarted.ledger = &ledger |
| 72 | newfs := newServingParkFS(fs.data, fs.state) |
| 73 | newfs.owner, newfs.mount = fs.owner, fs.mount |
| 74 | ledger.fs = newfs |
| 75 | restarted.root.Stop.Authenticate = func(context.Context, net.Conn) (StopPeer, error) { panic("startup authenticated an agent") } |
| 76 | page := recoveryPage(t, &restarted) |
| 77 | wantComplete := boundary == "retagged" || boundary == "complete-slot-pending" |
| 78 | if wantComplete && page.Results[0].Err != nil || !wantComplete && !errors.Is(page.Results[0].Err, errServingParkRetagPending) { |
| 79 | t.Fatalf("unexpected recovery: %+v", page.Results) |
| 80 | } |
| 81 | after, err := ledger.read(a.ID) |
| 82 | if err != nil || !after.Parked || after.Complete != wantComplete || after.Authorization != before.Authorization || !sameServingRootStops(after.Receipts, before.Receipts) { |
| 83 | t.Fatalf("changed intent or incomplete rename: %+v %v", after, err) |
| 84 | } |
| 85 | if !before.ParkedAt.IsZero() && (!after.ParkedAt.Equal(before.ParkedAt) || !after.ReapAfter.Equal(before.ReapAfter)) { |
| 86 | t.Fatal("startup refreshed retention") |
| 87 | } |
| 88 | if !reflect.DeepEqual(f.f.read(), claimBefore) { |
| 89 | t.Fatal("startup changed claim or retry state") |
| 90 | } |
| 91 | for suffix, want := range map[string]string{"": "served bytes", ".gen": "G", "-wal": "WAL", "-shm": "SHM"} { |
| 92 | body, err := os.ReadFile(filepath.Join(f.f.guard.DataDir, "instances", ".serving-parked", a.ID, f.current.Host+".db"+suffix)) |
| 93 | if err != nil || string(body) != want { |
| 94 | t.Fatalf("retained bytes %s: %q %v", suffix, body, err) |
| 95 | } |
| 96 | } |
| 97 | if !wantComplete { |
| 98 | f.tags(f.current.Stamp) |
| 99 | if page = recoveryPage(t, &restarted); page.Results[0].Err != nil { |
| 100 | t.Fatal(page.Results[0].Err) |
| 101 | } |
| 102 | } |
| 103 | slot, err := ledger.slot(after.Authorization) |
| 104 | if err != nil || slot.Pending != "" || slot.Last != a.ID || slot.ID != a.ID { |
| 105 | t.Fatalf("completion replay references: %+v %v", slot, err) |
| 106 | } |
| 107 | }) |
| 108 | } |
| 109 | } |
| 110 | |
| 111 | func TestServingParkStartupRefusesChangedAuthority(t *testing.T) { |
| 112 | for _, change := range []string{"enrollment", "binding", "claim", "tags", "directory", "receipt", "launch-intent", "census", "cancel"} { |
| 113 | t.Run(change, func(t *testing.T) { |
| 114 | f := newServingRootFixture(t, true) |
| 115 | a, err := f.park() |
| 116 | if err != nil { |
| 117 | t.Fatal(err) |
| 118 | } |
| 119 | f.tags(f.current.Stamp) |
| 120 | before, _ := f.disk.Read("retirement-" + a.ID + ".json") |
| 121 | ctx := t.Context() |
| 122 | switch change { |
| 123 | case "enrollment": |
| 124 | f.actor.Enrollment = "new-enrollment" |
| 125 | case "binding": |
| 126 | f.actor.Binding = "8" |
| 127 | case "claim": |
| 128 | _, err = hostclaim.Update(ctx, f.f.claims, f.f.read(), func(c *hostclaim.Claim) error { c.Holder.WriterBox = "other-box"; return nil }) |
| 129 | case "tags": |
| 130 | f.tags(f.current.Stamp + 1) |
| 131 | case "directory": |
| 132 | err = os.Rename(filepath.Dir(f.db), filepath.Dir(f.db)+"-saved") |
| 133 | if err == nil { |
| 134 | err = os.Mkdir(filepath.Dir(f.db), 0700) |
| 135 | } |
| 136 | case "receipt": |
| 137 | f.mutateRoot(func(m map[string]any) { m["State"].(map[string]any)["fence_journal"] = []any{} }) |
| 138 | case "launch-intent": |
| 139 | target := f.receipt.Target |
| 140 | target.Stamp, target.Session = f.current.Stamp, f.current.Session |
| 141 | err = f.f.journal.Prelaunch(ctx, hostbox.LaunchInventory{Target: target, ControllerPID: 77, ProcessStart: "edge-start", NoState: true, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/parent", Device: 1, Inode: 2}}}) |
| 142 | if err == nil { |
| 143 | err = f.f.journal.IntendUnit(ctx, hostbox.LaunchReservation{ID: "new:main", Role: "main", InvocationID: "new", Inputs: "digest", Target: target, Group: hostbox.LaunchGroup{RelativePath: "system.slice/carlos-app@shop.service"}, Executable: hostbox.LaunchFile{Device: 1, Inode: 3}}) |
| 144 | } |
| 145 | case "census": |
| 146 | f.p.root.RequireNewWork = func(context.Context) error { return hostbox.ErrChanged } |
| 147 | case "cancel": |
| 148 | var cancel context.CancelFunc |
| 149 | ctx, cancel = context.WithCancel(ctx) |
| 150 | cancel() |
| 151 | } |
| 152 | if err != nil { |
| 153 | t.Fatal(err) |
| 154 | } |
| 155 | page, err := f.p.recoverServingParks(ctx, "", 128) |
| 156 | if err == nil && (len(page.Results) != 1 || page.Results[0].Err == nil) { |
| 157 | t.Fatalf("accepted changed %s: %+v", change, page) |
| 158 | } |
| 159 | after, _ := f.disk.Read("retirement-" + a.ID + ".json") |
| 160 | if string(after) != string(before) { |
| 161 | t.Fatal("refusal changed private retirement") |
| 162 | } |
| 163 | }) |
| 164 | } |
| 165 | } |
| 166 | |
| 167 | func TestServingParkStartupDeletionAndEmptyEvent(t *testing.T) { |
| 168 | for _, boundary := range []string{"unlink", "rmdir", "closed-old-event", "replacement", "unknown-child", "missing-no-intent"} { |
| 169 | t.Run(boundary, func(t *testing.T) { |
| 170 | f := newServingParkDiskFixture(t) |
| 171 | f.put("", "retain") |
| 172 | a, receipts := f.evidence(10, true, "missing") |
| 173 | _, err := f.park(a, receipts) |
| 174 | if err != nil { |
| 175 | t.Fatal(err) |
| 176 | } |
| 177 | f.complete(a) |
| 178 | f.now = f.now.Add(7 * 24 * time.Hour) |
| 179 | event := filepath.Dir(f.path(a, "")) |
| 180 | held, err := os.Open(event) |
| 181 | if err != nil { |
| 182 | t.Fatal(err) |
| 183 | } |
| 184 | defer held.Close() |
| 185 | f.fs.unlink = func(fd int, name string, flags int) error { |
| 186 | if boundary == "closed-old-event" && flags == unix.AT_REMOVEDIR { |
| 187 | return unix.EIO |
| 188 | } |
| 189 | err := unix.Unlinkat(fd, name, flags) |
| 190 | if err == nil && (boundary == "unlink" && flags == 0 || boundary != "unlink" && flags == unix.AT_REMOVEDIR) { |
| 191 | return unix.EIO |
| 192 | } |
| 193 | return err |
| 194 | } |
| 195 | err = f.ledger.withNamespace(t.Context(), func() error { return f.ledger.reapLocked(t.Context(), a.ID) }) |
| 196 | if !errors.Is(err, unix.EIO) { |
| 197 | t.Fatalf("missing fault: %v", err) |
| 198 | } |
| 199 | r, err := f.ledger.read(a.ID) |
| 200 | if err != nil { |
| 201 | t.Fatal(err) |
| 202 | } |
| 203 | if boundary == "closed-old-event" { |
| 204 | r.Closed, r.EventRemoveIntent = true, false |
| 205 | if err = f.ledger.save(r); err != nil { |
| 206 | t.Fatal(err) |
| 207 | } |
| 208 | } |
| 209 | if boundary == "replacement" || boundary == "unknown-child" { |
| 210 | if err = os.Mkdir(event, 0700); err != nil { |
| 211 | t.Fatal(err) |
| 212 | } |
| 213 | if boundary == "unknown-child" { |
| 214 | err = os.WriteFile(filepath.Join(event, "victim"), []byte("keep"), 0600) |
| 215 | } |
| 216 | if err != nil { |
| 217 | t.Fatal(err) |
| 218 | } |
| 219 | } |
| 220 | if boundary == "missing-no-intent" { |
| 221 | r.EventRemoveIntent = false |
| 222 | if err = f.ledger.save(r); err != nil { |
| 223 | t.Fatal(err) |
| 224 | } |
| 225 | } |
| 226 | f.fs.unlink = unix.Unlinkat |
| 227 | // No live claim/root host installation is needed for a deletion |
| 228 | // already authorized by a complete private retirement. |
| 229 | err = f.ledger.recoverCompleted(t.Context(), a.ID) |
| 230 | refuse := boundary == "replacement" || boundary == "unknown-child" || boundary == "missing-no-intent" |
| 231 | if (err != nil) != refuse { |
| 232 | t.Fatalf("recovery: %v, refuse=%v", err, refuse) |
| 233 | } |
| 234 | if refuse { |
| 235 | if boundary == "replacement" || boundary == "unknown-child" { |
| 236 | if _, err = os.Lstat(event); err != nil { |
| 237 | t.Fatal("refusal removed replacement", err) |
| 238 | } |
| 239 | } |
| 240 | if boundary == "unknown-child" { |
| 241 | body, err := os.ReadFile(filepath.Join(event, "victim")) |
| 242 | if err != nil || string(body) != "keep" { |
| 243 | t.Fatal("refusal removed unknown child", err) |
| 244 | } |
| 245 | } |
| 246 | return |
| 247 | } |
| 248 | r, err = f.ledger.read(a.ID) |
| 249 | if err != nil || !r.Closed || !r.EventRemoved { |
| 250 | t.Fatalf("unfinished cleanup: %+v %v", r, err) |
| 251 | } |
| 252 | if _, err = os.Lstat(event); !errors.Is(err, os.ErrNotExist) { |
| 253 | t.Fatal("empty event survived", err) |
| 254 | } |
| 255 | slot, err := f.ledger.slot(a) |
| 256 | if err != nil || slot.ID != a.ID || slot.Last != a.ID { |
| 257 | t.Fatal("lost replay reference", err) |
| 258 | } |
| 259 | if err = f.ledger.recoverCompleted(t.Context(), a.ID); err != nil { |
| 260 | t.Fatal("cleanup replay", err) |
| 261 | } |
| 262 | }) |
| 263 | } |
| 264 | } |
| 265 | |
| 266 | func TestServingParkStartupBoundedScanContinuesAfterCorruptRecord(t *testing.T) { |
| 267 | f := newServingRootFixture(t, true) |
| 268 | a, err := f.park() |
| 269 | if err != nil { |
| 270 | t.Fatal(err) |
| 271 | } |
| 272 | f.tags(f.current.Stamp) |
| 273 | if err = f.disk.Write("retirement-0000.json", []byte(`{}`)); err != nil { |
| 274 | t.Fatal(err) |
| 275 | } |
| 276 | page, err := f.p.recoverServingParks(t.Context(), "", 1) |
| 277 | if err != nil || !page.More || len(page.Results) != 1 || page.Results[0].Host != "unknown" || page.Results[0].Err == nil { |
| 278 | t.Fatalf("first page: %+v %v", page, err) |
| 279 | } |
| 280 | page, err = f.p.recoverServingParks(t.Context(), page.Next, 1) |
| 281 | if err != nil || page.More || len(page.Results) != 1 || page.Results[0].ID != a.ID || page.Results[0].Err != nil { |
| 282 | t.Fatalf("second page: %+v %v", page, err) |
| 283 | } |
| 284 | } |
| 285 | |
| 286 | func TestServingParkStartupLegacyKeepsConsumption(t *testing.T) { |
| 287 | for _, empty := range []bool{false, true} { |
| 288 | t.Run(map[bool]string{false: "legacy", true: "empty"}[empty], func(t *testing.T) { |
| 289 | f, _ := newServingLegacyRootFixture(t, empty) |
| 290 | a, err := f.p.parkLegacyServingDatabase(t.Context(), f.conn, f.current, servingLegacyPredecessor{Empty: empty}) |
| 291 | if err != nil { |
| 292 | t.Fatal(err) |
| 293 | } |
| 294 | key := servingLegacyKey("legacy", f.current) |
| 295 | consumedBefore, err := f.disk.Read(key) |
| 296 | if err != nil { |
| 297 | t.Fatal(err) |
| 298 | } |
| 299 | stopBefore, err := f.disk.Read(servingLegacyKey("legacy-stop", f.current)) |
| 300 | if err != nil { |
| 301 | t.Fatal(err) |
| 302 | } |
| 303 | before, err := f.p.ledger.read(a.ID) |
| 304 | if err != nil { |
| 305 | t.Fatal(err) |
| 306 | } |
| 307 | f.tags(f.current.Stamp) |
| 308 | page := recoveryPage(t, f.p) |
| 309 | if page.Results[0].Err != nil { |
| 310 | t.Fatal(page.Results[0].Err) |
| 311 | } |
| 312 | after, err := f.p.ledger.read(a.ID) |
| 313 | if err != nil || !after.Complete || after.Authorization != before.Authorization || !after.ReapAfter.Equal(before.ReapAfter) { |
| 314 | t.Fatalf("legacy recovery: %+v %v", after, err) |
| 315 | } |
| 316 | consumedAfter, _ := f.disk.Read(key) |
| 317 | stopAfter, _ := f.disk.Read(servingLegacyKey("legacy-stop", f.current)) |
| 318 | if string(consumedBefore) != string(consumedAfter) || string(stopBefore) != string(stopAfter) { |
| 319 | t.Fatal("startup changed legacy evidence") |
| 320 | } |
| 321 | }) |
| 322 | } |
| 323 | } |
| 324 | |
| 325 | func TestServingParkStartupRuntimeJoinsBeforeClose(t *testing.T) { |
| 326 | f := newServingRootFixture(t, true) |
| 327 | if _, err := f.park(); err != nil { |
| 328 | t.Fatal(err) |
| 329 | } |
| 330 | f.tags(f.current.Stamp) |
| 331 | entered, release := make(chan struct{}), make(chan struct{}) |
| 332 | f.p.root.RequireNewWork = func(ctx context.Context) error { |
| 333 | close(entered) |
| 334 | <-ctx.Done() |
| 335 | <-release |
| 336 | return ctx.Err() |
| 337 | } |
| 338 | closed := &servingCloseStore{Store: f.disk} |
| 339 | ctx, cancel := context.WithCancel(context.Background()) |
| 340 | defer cancel() |
| 341 | p := &servingParkRuntime{authority: f.p, state: closed, ctx: ctx, cancel: cancel, idle: make(chan struct{})} |
| 342 | done := make(chan error, 1) |
| 343 | go func() { done <- p.recoverPending(t.Context()) }() |
| 344 | select { |
| 345 | case <-entered: |
| 346 | case <-time.After(5 * time.Second): |
| 347 | t.Fatal("recovery did not enter authority") |
| 348 | } |
| 349 | short, stop := context.WithTimeout(t.Context(), 20*time.Millisecond) |
| 350 | defer stop() |
| 351 | if err := p.Close(short); !errors.Is(err, context.DeadlineExceeded) || closed.closed.Load() { |
| 352 | t.Fatal("closed underneath recovery", err) |
| 353 | } |
| 354 | close(release) |
| 355 | if err := <-done; err == nil { |
| 356 | t.Fatal("cancelled recovery succeeded") |
| 357 | } |
| 358 | if err := p.Close(t.Context()); err != nil || !closed.closed.Load() { |
| 359 | t.Fatal("recovery did not join", err) |
| 360 | } |
| 361 | if err := p.recoverPending(t.Context()); err == nil { |
| 362 | t.Fatal("closed runtime recovered") |
| 363 | } |
| 364 | } |
| 365 | |
| 366 | func TestServingParkStartupRunsAfterOtherRecoveryFailure(t *testing.T) { |
| 367 | f := newServingRootFixture(t, true) |
| 368 | a, err := f.park() |
| 369 | if err != nil { |
| 370 | t.Fatal(err) |
| 371 | } |
| 372 | f.tags(f.current.Stamp) |
| 373 | ctx, cancel := context.WithCancel(t.Context()) |
| 374 | defer cancel() |
| 375 | p := &servingParkRuntime{authority: f.p, state: f.disk, ctx: ctx, cancel: cancel, idle: make(chan struct{})} |
| 376 | defer p.Close(t.Context()) |
| 377 | h := f.p.root |
| 378 | h.servingParks = p |
| 379 | bad := filepath.Join(t.TempDir(), "not-a-directory") |
| 380 | if err = os.WriteFile(bad, nil, 0600); err != nil { |
| 381 | t.Fatal(err) |
| 382 | } |
| 383 | h.Reconcile.Retirement.JournalDir = bad |
| 384 | if err = h.RecoverPending(t.Context()); err == nil { |
| 385 | t.Fatal("other recovery error was swallowed") |
| 386 | } |
| 387 | r, err := f.p.ledger.read(a.ID) |
| 388 | if err != nil || !r.Complete { |
| 389 | t.Fatalf("serving recovery skipped: %+v %v", r, err) |
| 390 | } |
| 391 | } |
| 392 | |
amadan