| 1 | //go:build linux |
| 2 | |
| 3 | package hostexec |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "golang.org/x/sys/unix" |
| 10 | "os" |
| 11 | "os/exec" |
| 12 | "sync" |
| 13 | "sync/atomic" |
| 14 | "testing" |
| 15 | "time" |
| 16 | |
| 17 | "github.com/carlosframework/platform/internal/hostclaim" |
| 18 | ) |
| 19 | |
| 20 | func TestLostControllerCannotPublishStaleActiveState(t *testing.T) { |
| 21 | cfg, _, clock, _, _ := fixture(t) |
| 22 | c, err := Open(t.Context(), cfg) |
| 23 | if err != nil { |
| 24 | t.Fatal(err) |
| 25 | } |
| 26 | defer closeController(t, c) |
| 27 | active := ready(t, c) |
| 28 | c.mu.Lock() |
| 29 | c.loseLocked(ErrLost) |
| 30 | c.mu.Unlock() |
| 31 | <-c.cleanupDone |
| 32 | if err = c.publish(active); !errors.Is(err, ErrLost) { |
| 33 | t.Fatalf("stale publication = %v", err) |
| 34 | } |
| 35 | if err = CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); err == nil { |
| 36 | t.Fatal("stale active state revived cross-process admission") |
| 37 | } |
| 38 | } |
| 39 | |
| 40 | func TestWaitingPublicationDoesNotBlockWatchdogOrCleanup(t *testing.T) { |
| 41 | cfg, _, clock, _, cleaned := fixture(t) |
| 42 | c, err := Open(t.Context(), cfg) |
| 43 | if err != nil { |
| 44 | t.Fatal(err) |
| 45 | } |
| 46 | defer closeController(t, c) |
| 47 | active := ready(t, c) |
| 48 | held, release := make(chan struct{}), make(chan struct{}) |
| 49 | guardDone := make(chan error, 1) |
| 50 | go func() { |
| 51 | guardDone <- cfg.Guard.WithLocal(context.Background(), active.Host, func() error { |
| 52 | close(held) |
| 53 | <-release |
| 54 | return nil |
| 55 | }) |
| 56 | }() |
| 57 | <-held |
| 58 | published := make(chan error, 1) |
| 59 | go func() { published <- c.publish(active) }() |
| 60 | clock.advance(TTL) |
| 61 | // Ready samples the suspend-aware clock even if a watchdog wake races. |
| 62 | if _, err = c.Ready(t.Context()); err == nil { |
| 63 | t.Fatal("expired controller remained ready") |
| 64 | } |
| 65 | select { |
| 66 | case <-c.Lost(): |
| 67 | case <-time.After(2 * time.Second): |
| 68 | t.Fatal("host guard blocked watchdog") |
| 69 | } |
| 70 | until(t, func() bool { return cleaned.Load() >= 2 }) |
| 71 | close(release) |
| 72 | if err = <-guardDone; err != nil { |
| 73 | t.Fatal(err) |
| 74 | } |
| 75 | if err = <-published; !errors.Is(err, ErrLost) { |
| 76 | t.Fatalf("publication after expiration = %v", err) |
| 77 | } |
| 78 | <-c.cleanupDone |
| 79 | if err = CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); err == nil { |
| 80 | t.Fatal("expired publication admitted work") |
| 81 | } |
| 82 | } |
| 83 | |
| 84 | func TestOrphanActiveStateCannotAuthorizeAfterControllerExit(t *testing.T) { |
| 85 | cfg, _, clock, _, _ := fixture(t) |
| 86 | c, err := Open(t.Context(), cfg) |
| 87 | if err != nil { |
| 88 | t.Fatal(err) |
| 89 | } |
| 90 | active := ready(t, c) |
| 91 | closeController(t, c) |
| 92 | // Simulate an active rename surviving a crash before its fsync outcome was |
| 93 | // known. The persisted deadline is still future, but no controller lives. |
| 94 | d, err := openLocal(cfg.Guard.DataDir) |
| 95 | if err != nil { |
| 96 | t.Fatal(err) |
| 97 | } |
| 98 | defer d.close() |
| 99 | if err = d.write(active); err != nil { |
| 100 | t.Fatal(err) |
| 101 | } |
| 102 | if err = CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) { |
| 103 | t.Fatalf("orphan active state = %v", err) |
| 104 | } |
| 105 | } |
| 106 | |
| 107 | type blockedClaimRead struct { |
| 108 | base *claims |
| 109 | entered chan struct{} |
| 110 | release chan struct{} |
| 111 | } |
| 112 | |
| 113 | func (r blockedClaimRead) Read(ctx context.Context, host string) (hostclaim.Observed, error) { |
| 114 | close(r.entered) |
| 115 | select { |
| 116 | case <-ctx.Done(): |
| 117 | return hostclaim.Observed{}, ctx.Err() |
| 118 | case <-r.release: |
| 119 | return r.base.Read(ctx, host) |
| 120 | } |
| 121 | } |
| 122 | |
| 123 | func TestCC9LockDoesNotReviveOrphanDuringClaimRead(t *testing.T) { |
| 124 | cfg, r, clock, _, _ := fixture(t) |
| 125 | c, err := Open(t.Context(), cfg) |
| 126 | if err != nil { |
| 127 | t.Fatal(err) |
| 128 | } |
| 129 | active := ready(t, c) |
| 130 | closeController(t, c) |
| 131 | d, err := openLocal(cfg.Guard.DataDir) |
| 132 | if err != nil { |
| 133 | t.Fatal(err) |
| 134 | } |
| 135 | defer d.close() |
| 136 | if err = d.write(active); err != nil { |
| 137 | t.Fatal(err) |
| 138 | } |
| 139 | blocked := blockedClaimRead{base: r, entered: make(chan struct{}), release: make(chan struct{})} |
| 140 | cfg.Guard.Claims = blocked |
| 141 | done := make(chan error, 1) |
| 142 | go func() { done <- ClearStale(t.Context(), cfg) }() |
| 143 | <-blocked.entered |
| 144 | checkErr := CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir) |
| 145 | close(blocked.release) |
| 146 | if checkErr == nil { |
| 147 | t.Fatal("CC9 lifetime lock revived orphan active bytes") |
| 148 | } |
| 149 | if err = <-done; !errors.Is(err, ErrInactive) { |
| 150 | t.Fatalf("CC9 with participation retained = %v", err) |
| 151 | } |
| 152 | } |
| 153 | |
| 154 | func TestCC9SuccessorExclusionClearsOnlyExpectedOperation(t *testing.T) { |
| 155 | cfg, r, _, _, _ := fixture(t) |
| 156 | c, err := Open(t.Context(), cfg) |
| 157 | if err != nil { |
| 158 | t.Fatal(err) |
| 159 | } |
| 160 | ready(t, c) |
| 161 | closeController(t, c) |
| 162 | r.mu.Lock() |
| 163 | r.q.Op.ID = "successor-op" |
| 164 | r.q.Op.Exec = "other-box" |
| 165 | r.q.Op.Participants = []string{"other-box"} |
| 166 | r.mu.Unlock() |
| 167 | if err = ClearStale(t.Context(), cfg); err != nil { |
| 168 | t.Fatalf("successor excludes box: %v", err) |
| 169 | } |
| 170 | d, err := openLocal(cfg.Guard.DataDir) |
| 171 | if err != nil { |
| 172 | t.Fatal(err) |
| 173 | } |
| 174 | defer d.close() |
| 175 | if _, err = d.read(cfg.Binding.Host); !errors.Is(err, os.ErrNotExist) { |
| 176 | t.Fatalf("expected stale record removed: %v", err) |
| 177 | } |
| 178 | unrelated := State{V: 1, Binding: cfg.Binding, State: "lost", Expires: time.Now().Add(TTL), SafeUntil: 99} |
| 179 | unrelated.ControllerID = "other-controller" |
| 180 | unrelated.Op = "unrelated-op" |
| 181 | if err = d.write(unrelated); err != nil { |
| 182 | t.Fatal(err) |
| 183 | } |
| 184 | if err = ClearStale(t.Context(), cfg); !errors.Is(err, ErrState) { |
| 185 | t.Fatalf("unrelated local fence: %v", err) |
| 186 | } |
| 187 | still, err := d.read(cfg.Binding.Host) |
| 188 | if err != nil || still.Op != unrelated.Op { |
| 189 | t.Fatalf("unrelated fence changed: %v, %v", still, err) |
| 190 | } |
| 191 | } |
| 192 | |
| 193 | func TestAmbiguousRenewalENOSPCBeforeRenameDeniesImmediately(t *testing.T) { |
| 194 | cfg, _, clock, store, cleaned := fixture(t) |
| 195 | store.failAt = 2 |
| 196 | var noSpace atomic.Bool |
| 197 | noSpace.Store(true) |
| 198 | var failures atomic.Int32 |
| 199 | cfg.writeLocal = func(d *localDir, s State) error { |
| 200 | if s.State != "lost" || !noSpace.Load() { |
| 201 | return d.write(s) |
| 202 | } |
| 203 | original := d.sync |
| 204 | d.sync = func(int) error { return unix.ENOSPC } |
| 205 | defer func() { d.sync = original }() |
| 206 | err := d.write(s) // file fsync fails before rename; old active bytes remain |
| 207 | if errors.Is(err, unix.ENOSPC) { |
| 208 | failures.Add(1) |
| 209 | } |
| 210 | return err |
| 211 | } |
| 212 | c, err := Open(t.Context(), cfg) |
| 213 | if err != nil { |
| 214 | t.Fatal(err) |
| 215 | } |
| 216 | defer func() { noSpace.Store(false); closeController(t, c) }() |
| 217 | active := ready(t, c) |
| 218 | until(t, func() bool { |
| 219 | clock.mu.Lock() |
| 220 | defer clock.mu.Unlock() |
| 221 | for _, a := range clock.alarms { |
| 222 | if a.at == clock.boot+int64(Renewal) { |
| 223 | return true |
| 224 | } |
| 225 | } |
| 226 | return false |
| 227 | }) |
| 228 | clock.advance(Renewal) |
| 229 | select { |
| 230 | case <-c.Lost(): |
| 231 | case <-time.After(2 * time.Second): |
| 232 | t.Fatal("ambiguous renewal did not latch loss") |
| 233 | } |
| 234 | // Check from the loss signal onward, before waiting for any persistence. |
| 235 | if err := CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) { |
| 236 | t.Fatalf("admitted immediately after ambiguity: %v", err) |
| 237 | } |
| 238 | until(t, func() bool { return failures.Load() >= 2 && cleaned.Load() >= 2 }) |
| 239 | if _, err := c.Ready(t.Context()); err == nil { |
| 240 | t.Fatal("in-memory admission survived loss") |
| 241 | } |
| 242 | d, err := openLocal(cfg.Guard.DataDir) |
| 243 | if err != nil { |
| 244 | t.Fatal(err) |
| 245 | } |
| 246 | defer d.close() |
| 247 | old, err := d.read(active.Host) |
| 248 | if err != nil || old != active { |
| 249 | t.Fatalf("test did not leave the old active file: %+v %v", old, err) |
| 250 | } |
| 251 | _, boot, _ := clock.Now() |
| 252 | if boot >= active.SafeUntil { |
| 253 | t.Fatal("test reached expiry instead of exercising immediate denial") |
| 254 | } |
| 255 | binding, _ := json.Marshal(active.Binding) |
| 256 | cmd := exec.CommandContext(t.Context(), os.Args[0], "-test.run=^TestCheckLocalRefusesLostLiveLock$") |
| 257 | cmd.Env = append(os.Environ(), "HOSTEXEC_PROBE_BINDING="+string(binding), "HOSTEXEC_PROBE_DATA="+cfg.Guard.DataDir, "HOSTEXEC_PROBE_LOCKS="+cfg.LockDir) |
| 258 | if output, err := cmd.CombinedOutput(); err != nil { |
| 259 | t.Fatalf("independent process admitted old active state: %v\n%s", err, output) |
| 260 | } |
| 261 | ctx, cancel := context.WithTimeout(t.Context(), 30*time.Millisecond) |
| 262 | defer cancel() |
| 263 | entered := false |
| 264 | err = cfg.Guard.WithLocal(ctx, active.Host, func() error { entered = true; return nil }) |
| 265 | if !errors.Is(err, context.DeadlineExceeded) || entered { |
| 266 | t.Fatalf("guard released while denial write failed: %v entered=%v", err, entered) |
| 267 | } |
| 268 | if unlock, err := tryAcquire(cfg.LockDir, active.Host); !errors.Is(err, ErrControllerBusy) { |
| 269 | if unlock != nil { |
| 270 | unlock() |
| 271 | } |
| 272 | t.Fatalf("controller exclusion released: %v", err) |
| 273 | } |
| 274 | ctxClose, cancelClose := context.WithTimeout(t.Context(), 30*time.Millisecond) |
| 275 | defer cancelClose() |
| 276 | if err := c.Close(ctxClose); !errors.Is(err, context.DeadlineExceeded) { |
| 277 | t.Fatalf("Close did not retain the unfinished controller: %v", err) |
| 278 | } |
| 279 | noSpace.Store(false) |
| 280 | select { |
| 281 | case <-c.cleanupDone: |
| 282 | case <-time.After(2 * time.Second): |
| 283 | t.Fatal("lost state was not retried after disk recovery") |
| 284 | } |
| 285 | lost, err := d.read(active.Host) |
| 286 | if err != nil || lost.State != "lost" { |
| 287 | t.Fatalf("denial not durable: %+v %v", lost, err) |
| 288 | } |
| 289 | if err := cfg.Guard.WithLocal(t.Context(), active.Host, func() error { |
| 290 | if err := CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) { |
| 291 | t.Fatalf("disk recovery revived admission: %v", err) |
| 292 | } |
| 293 | return nil |
| 294 | }); err != nil { |
| 295 | t.Fatal(err) |
| 296 | } |
| 297 | } |
| 298 | |
| 299 | func TestCheckLocalRefusesLostLiveLock(t *testing.T) { |
| 300 | raw := os.Getenv("HOSTEXEC_PROBE_BINDING") |
| 301 | if raw == "" { |
| 302 | t.Skip("subprocess witness") |
| 303 | } |
| 304 | var binding Binding |
| 305 | if err := json.Unmarshal([]byte(raw), &binding); err != nil { |
| 306 | t.Fatal(err) |
| 307 | } |
| 308 | clock := &testClock{boot: int64(time.Hour + Renewal)} |
| 309 | if err := CheckLocal(os.Getenv("HOSTEXEC_PROBE_DATA"), binding, clock, os.Getenv("HOSTEXEC_PROBE_LOCKS")); !errors.Is(err, ErrLost) { |
| 310 | t.Fatalf("old active record authorized an independent process: %v", err) |
| 311 | } |
| 312 | } |
| 313 | |
| 314 | func TestOpenOrphanControllerPendingWriteFailureCleansBeforeReleasingExclusion(t *testing.T) { |
| 315 | testOpenOrphanControllerPersistence(t, false) |
| 316 | } |
| 317 | |
| 318 | func TestOpenBlockedPendingPersistenceStillCleansBeforeReleasingExclusion(t *testing.T) { |
| 319 | testOpenOrphanControllerPersistence(t, true) |
| 320 | } |
| 321 | |
| 322 | func testOpenOrphanControllerPersistence(t *testing.T, blocked bool) { |
| 323 | cfg, _, clock, store, _ := fixture(t) |
| 324 | wall, boot, _ := clock.Now() |
| 325 | orphan := State{V: 1, Binding: cfg.Binding, State: "active", LeaseETag: "orphan-etag", Expires: wall.Add(TTL), SafeUntil: boot + int64(TTL-Margin)} |
| 326 | orphan.ControllerID = "dead-controller" |
| 327 | d, err := openLocal(cfg.Guard.DataDir) |
| 328 | if err != nil { |
| 329 | t.Fatal(err) |
| 330 | } |
| 331 | defer d.close() |
| 332 | if err := d.write(orphan); err != nil { |
| 333 | t.Fatal(err) |
| 334 | } |
| 335 | // The controller's locks have vanished, but its active bytes and a host |
| 336 | // process survived. Admission refusal alone cannot stop that process. |
| 337 | process := exec.Command("sleep", "60") |
| 338 | if err := process.Start(); err != nil { |
| 339 | t.Fatal(err) |
| 340 | } |
| 341 | defer process.Process.Kill() |
| 342 | hostExited := make(chan struct{}) |
| 343 | go func() { _ = process.Wait(); close(hostExited) }() |
| 344 | allowJoin := make(chan struct{}) |
| 345 | var joinOnce sync.Once |
| 346 | defer joinOnce.Do(func() { close(allowJoin) }) |
| 347 | cleaned := make(chan struct{}) |
| 348 | cleanupResult := make(chan error, 1) |
| 349 | cfg.Cleanup = func(ctx context.Context) error { |
| 350 | cleanupResult <- ctx.Err() |
| 351 | if err := process.Process.Kill(); err != nil { |
| 352 | return err |
| 353 | } |
| 354 | <-hostExited |
| 355 | close(cleaned) |
| 356 | <-allowJoin |
| 357 | return nil |
| 358 | } |
| 359 | ctx, cancel := context.WithCancel(t.Context()) |
| 360 | defer cancel() |
| 361 | writeEntered := make(chan struct{}) |
| 362 | allowWrite := make(chan struct{}) |
| 363 | var writeOnce sync.Once |
| 364 | defer writeOnce.Do(func() { close(allowWrite) }) |
| 365 | var noSpace atomic.Bool |
| 366 | var failures atomic.Int32 |
| 367 | noSpace.Store(true) |
| 368 | defer noSpace.Store(false) |
| 369 | cfg.writeLocal = func(d *localDir, s State) error { |
| 370 | if noSpace.Load() { |
| 371 | if s.State == "pending" { |
| 372 | cancel() // cleanup must not inherit the failed opener's lifetime |
| 373 | } |
| 374 | original := d.sync |
| 375 | d.sync = func(int) error { |
| 376 | if blocked && s.State == "pending" { |
| 377 | close(writeEntered) |
| 378 | <-allowWrite // model an fsync that cannot return until released |
| 379 | } |
| 380 | return unix.ENOSPC |
| 381 | } |
| 382 | defer func() { d.sync = original }() |
| 383 | err := d.write(s) |
| 384 | if errors.Is(err, unix.ENOSPC) { |
| 385 | failures.Add(1) |
| 386 | } |
| 387 | return err |
| 388 | } |
| 389 | return d.write(s) |
| 390 | } |
| 391 | done := make(chan error, 1) |
| 392 | go func() { _, err := Open(ctx, cfg); done <- err }() |
| 393 | select { |
| 394 | case <-cleaned: |
| 395 | case <-time.After(2 * time.Second): |
| 396 | t.Fatal("failed pending write skipped orphan process cleanup") |
| 397 | } |
| 398 | if err := <-cleanupResult; err != nil { |
| 399 | t.Fatalf("cleanup inherited caller cancellation: %v", err) |
| 400 | } |
| 401 | |
| 402 | if old, err := d.read(orphan.Host); err != nil || old != orphan { |
| 403 | t.Fatalf("failed persistence did not retain orphan active bytes: %+v %v", old, err) |
| 404 | } |
| 405 | assertExcluded := func() { |
| 406 | t.Helper() |
| 407 | if unlock, err := tryAcquire(cfg.LockDir, orphan.Host); !errors.Is(err, ErrControllerBusy) { |
| 408 | if unlock != nil { |
| 409 | unlock() |
| 410 | } |
| 411 | t.Fatalf("exclusion released before cleanup joined: %v", err) |
| 412 | } |
| 413 | select { |
| 414 | case err := <-done: |
| 415 | t.Fatalf("Open returned before cleanup joined: %v", err) |
| 416 | default: |
| 417 | } |
| 418 | } |
| 419 | if blocked { |
| 420 | select { |
| 421 | case <-writeEntered: |
| 422 | case <-time.After(2 * time.Second): |
| 423 | t.Fatal("pending fsync was not reached") |
| 424 | } |
| 425 | if failures.Load() != 0 { |
| 426 | t.Fatal("pending fsync returned before cleanup was observed") |
| 427 | } |
| 428 | assertExcluded() |
| 429 | writeOnce.Do(func() { close(allowWrite) }) |
| 430 | } |
| 431 | until(t, func() bool { return failures.Load() >= 1 }) |
| 432 | assertExcluded() |
| 433 | if err := CheckLocal(cfg.Guard.DataDir, orphan.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) { |
| 434 | t.Fatalf("orphan bytes admitted work: %v", err) |
| 435 | } |
| 436 | // Persistence is still unavailable when cleanup joins. No live lock was |
| 437 | // acquired, so orphan bytes cannot authorize work after exclusion ends. |
| 438 | joinOnce.Do(func() { close(allowJoin) }) |
| 439 | select { |
| 440 | case err := <-done: |
| 441 | if !errors.Is(err, unix.ENOSPC) { |
| 442 | t.Fatalf("Open lost pending-write failure: %v", err) |
| 443 | } |
| 444 | case <-time.After(2 * time.Second): |
| 445 | t.Fatal("Open did not join cleanup while persistence remained unavailable") |
| 446 | } |
| 447 | unlock, err := tryAcquire(cfg.LockDir, orphan.Host) |
| 448 | if err != nil { |
| 449 | t.Fatalf("exclusion retained after cleanup joined: %v", err) |
| 450 | } |
| 451 | unlock() |
| 452 | if err := CheckLocal(cfg.Guard.DataDir, orphan.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) { |
| 453 | t.Fatalf("released exclusion revived orphan bytes: %v", err) |
| 454 | } |
| 455 | store.mu.Lock() |
| 456 | defer store.mu.Unlock() |
| 457 | if store.calls != 0 { |
| 458 | t.Fatalf("failed startup attempted %d lease writes", store.calls) |
| 459 | } |
| 460 | } |
| 461 | |
amadan