| 1 | //go:build linux |
| 2 | |
| 3 | package hostexec |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "sync/atomic" |
| 14 | "testing" |
| 15 | "time" |
| 16 | |
| 17 | "github.com/carlosframework/platform/internal/hostclaim" |
| 18 | "github.com/carlosframework/platform/internal/hostguard" |
| 19 | "github.com/carlosframework/platform/internal/hostquarantine" |
| 20 | "github.com/carlosframework/platform/internal/release" |
| 21 | ) |
| 22 | |
| 23 | type alarm struct { |
| 24 | at int64 |
| 25 | ch chan time.Time |
| 26 | } |
| 27 | type testClock struct { |
| 28 | mu sync.Mutex |
| 29 | wall time.Time |
| 30 | boot int64 |
| 31 | alarms []alarm |
| 32 | } |
| 33 | |
| 34 | func (k *testClock) Now() (time.Time, int64, error) { |
| 35 | k.mu.Lock() |
| 36 | defer k.mu.Unlock() |
| 37 | return k.wall, k.boot, nil |
| 38 | } |
| 39 | func (k *testClock) After(d time.Duration) <-chan time.Time { |
| 40 | k.mu.Lock() |
| 41 | defer k.mu.Unlock() |
| 42 | ch := make(chan time.Time, 1) |
| 43 | if d <= 0 { |
| 44 | ch <- k.wall |
| 45 | } else { |
| 46 | k.alarms = append(k.alarms, alarm{k.boot + int64(d), ch}) |
| 47 | } |
| 48 | return ch |
| 49 | } |
| 50 | func (k *testClock) advance(d time.Duration) { |
| 51 | k.mu.Lock() |
| 52 | defer k.mu.Unlock() |
| 53 | k.boot += int64(d) |
| 54 | k.wall = k.wall.Add(d) |
| 55 | rest := k.alarms[:0] |
| 56 | for _, a := range k.alarms { |
| 57 | if a.at <= k.boot { |
| 58 | a.ch <- k.wall |
| 59 | } else { |
| 60 | rest = append(rest, a) |
| 61 | } |
| 62 | } |
| 63 | k.alarms = rest |
| 64 | } |
| 65 | |
| 66 | type claims struct { |
| 67 | mu sync.Mutex |
| 68 | q hostclaim.Claim |
| 69 | err error |
| 70 | } |
| 71 | |
| 72 | func (r *claims) Read(context.Context, string) (hostclaim.Observed, error) { |
| 73 | r.mu.Lock() |
| 74 | defer r.mu.Unlock() |
| 75 | raw, _ := json.Marshal(r.q) |
| 76 | var q hostclaim.Claim |
| 77 | _ = json.Unmarshal(raw, &q) |
| 78 | return hostclaim.Observed{Host: q.Host, Claim: &q, ETag: "claim-etag"}, r.err |
| 79 | } |
| 80 | |
| 81 | type directory struct { |
| 82 | state string |
| 83 | err error |
| 84 | } |
| 85 | |
| 86 | func (d directory) Lookup(context.Context, string) (hostquarantine.Actor, error) { |
| 87 | return hostquarantine.Actor{Enrollment: "enroll", State: d.state}, d.err |
| 88 | } |
| 89 | |
| 90 | type delayedStore struct { |
| 91 | release.Store |
| 92 | mu sync.Mutex |
| 93 | calls int |
| 94 | onPut func(int) |
| 95 | failAt int |
| 96 | called chan int |
| 97 | } |
| 98 | |
| 99 | func (s *delayedStore) Put(ctx context.Context, key string, b []byte, etag string) (string, error) { |
| 100 | s.mu.Lock() |
| 101 | s.calls++ |
| 102 | n := s.calls |
| 103 | s.mu.Unlock() |
| 104 | if s.called != nil { |
| 105 | s.called <- n |
| 106 | } |
| 107 | if s.onPut != nil { |
| 108 | s.onPut(n) |
| 109 | } |
| 110 | if n == s.failAt { |
| 111 | return "", errors.New("ambiguous network failure") |
| 112 | } |
| 113 | // Deliberately permit a late response even after cancellation. |
| 114 | return s.Store.Put(context.Background(), key, b, etag) |
| 115 | } |
| 116 | func (s *delayedStore) DeleteIfMatch(ctx context.Context, key, etag string) error { |
| 117 | return release.DeleteIfMatch(ctx, s.Store, key, etag) |
| 118 | } |
| 119 | |
| 120 | func fixture(t *testing.T) (Config, *claims, *testClock, *delayedStore, *atomic.Int32) { |
| 121 | t.Helper() |
| 122 | dir := t.TempDir() |
| 123 | for _, name := range []string{"hostguard", ".hostexec", "locks", "exec-locks"} { |
| 124 | if err := os.Mkdir(filepath.Join(dir, name), 0770); err != nil { |
| 125 | t.Fatal(err) |
| 126 | } |
| 127 | } |
| 128 | host := "shop.example.com" |
| 129 | inc, _ := hostclaim.Incarnation("bab", "jam", host, time.Time{}) |
| 130 | q := hostclaim.Claim{V: 2, Host: host, Gen: 7, Holder: &hostclaim.Holder{Family: "instance", Key: "apps/bab/jam/instances/" + host + ".json", Account: "bab", App: "jam", Incarnation: inc, Location: "instances/" + host, Stamp: 3}, Op: hostclaim.Op{Kind: hostclaim.KindDeleting, ID: "op", Phase: "stop", Exec: "box", Participants: []string{"box"}}} |
| 131 | if err := q.Validate(); err != nil { |
| 132 | t.Fatal(err) |
| 133 | } |
| 134 | r := &claims{q: q} |
| 135 | clock := &testClock{wall: time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC), boot: int64(time.Hour)} |
| 136 | store := &delayedStore{Store: release.NewDirStore(t.TempDir()), called: make(chan int, 20)} |
| 137 | cleaned := &atomic.Int32{} |
| 138 | cfg := Config{Binding: Binding{Box: "box", Enrollment: "enroll", Host: host, Incarnation: inc, Op: "op", BootID: "boot"}, Store: store, Guard: hostguard.Guard{DataDir: dir, LockDir: filepath.Join(dir, "locks"), Claims: r}, Directory: directory{state: "active"}, Clock: clock, LockDir: filepath.Join(dir, "exec-locks"), CloseAdmission: func() {}, CancelWork: func() {}, Cleanup: func(context.Context) error { cleaned.Add(1); return nil }} |
| 139 | return cfg, r, clock, store, cleaned |
| 140 | } |
| 141 | func ready(t *testing.T, c *Controller) State { |
| 142 | t.Helper() |
| 143 | ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) |
| 144 | defer cancel() |
| 145 | s, err := c.Ready(ctx) |
| 146 | if err != nil { |
| 147 | t.Fatal(err) |
| 148 | } |
| 149 | return s |
| 150 | } |
| 151 | func closeController(t *testing.T, c *Controller) { |
| 152 | t.Helper() |
| 153 | ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) |
| 154 | defer cancel() |
| 155 | if err := c.Close(ctx); err != nil { |
| 156 | t.Error(err) |
| 157 | } |
| 158 | } |
| 159 | func until(t *testing.T, fn func() bool) { |
| 160 | t.Helper() |
| 161 | deadline := time.Now().Add(2 * time.Second) |
| 162 | for !fn() { |
| 163 | if time.Now().After(deadline) { |
| 164 | t.Fatal("condition did not become true") |
| 165 | } |
| 166 | time.Sleep(time.Millisecond) |
| 167 | } |
| 168 | } |
| 169 | |
| 170 | func TestRenewalRequestStartAndIndependentExpiry(t *testing.T) { |
| 171 | cfg, _, clock, store, cleaned := fixture(t) |
| 172 | blocked := make(chan struct{}) |
| 173 | store.onPut = func(n int) { |
| 174 | if n == 2 { |
| 175 | <-blocked |
| 176 | } |
| 177 | } |
| 178 | c, err := Open(t.Context(), cfg) |
| 179 | if err != nil { |
| 180 | t.Fatal(err) |
| 181 | } |
| 182 | defer closeController(t, c) |
| 183 | first := ready(t, c) |
| 184 | key, _ := Key(first.Host, first.Box) |
| 185 | b, _, err := store.Get(t.Context(), key) |
| 186 | if err != nil { |
| 187 | t.Fatal(err) |
| 188 | } |
| 189 | var wire map[string]any |
| 190 | _ = json.Unmarshal(b, &wire) |
| 191 | if len(wire) != 2 || wire["op"] != "op" { |
| 192 | t.Fatalf("wire %s", b) |
| 193 | } |
| 194 | until(t, func() bool { clock.mu.Lock(); defer clock.mu.Unlock(); return len(clock.alarms) >= 2 }) |
| 195 | clock.advance(Renewal) |
| 196 | for { |
| 197 | select { |
| 198 | case n := <-store.called: |
| 199 | if n == 2 { |
| 200 | goto renewing |
| 201 | } |
| 202 | case <-time.After(2 * time.Second): |
| 203 | t.Fatal("no renewal") |
| 204 | } |
| 205 | } |
| 206 | renewing: |
| 207 | clock.advance(65 * time.Second) |
| 208 | select { |
| 209 | case <-c.Lost(): |
| 210 | case <-time.After(2 * time.Second): |
| 211 | t.Fatal("blocked PUT blocked watchdog") |
| 212 | } |
| 213 | if err := CheckLocal(cfg.Guard.DataDir, first.Binding, clock, cfg.LockDir); err == nil { |
| 214 | t.Fatal("writer admitted at deadline") |
| 215 | } |
| 216 | close(blocked) |
| 217 | <-c.done |
| 218 | if _, err := c.Ready(t.Context()); err == nil { |
| 219 | t.Fatal("late success revived controller") |
| 220 | } |
| 221 | until(t, func() bool { return cleaned.Load() >= 2 }) |
| 222 | } |
| 223 | |
| 224 | func TestRenewalUsesStartAndWallRollbackCannotExtend(t *testing.T) { |
| 225 | cfg, _, clock, store, _ := fixture(t) |
| 226 | store.onPut = func(n int) { |
| 227 | if n == 1 { |
| 228 | clock.advance(9 * time.Second) |
| 229 | } |
| 230 | } |
| 231 | c, err := Open(t.Context(), cfg) |
| 232 | if err != nil { |
| 233 | t.Fatal(err) |
| 234 | } |
| 235 | defer closeController(t, c) |
| 236 | s := ready(t, c) |
| 237 | if s.SafeUntil != int64(time.Hour+85*time.Second) { |
| 238 | t.Fatalf("response-relative deadline %d", s.SafeUntil) |
| 239 | } |
| 240 | clock.mu.Lock() |
| 241 | clock.wall = clock.wall.Add(-time.Hour) |
| 242 | clock.mu.Unlock() |
| 243 | clock.advance(76 * time.Second) |
| 244 | if err := CheckLocal(cfg.Guard.DataDir, s.Binding, clock, cfg.LockDir); err == nil { |
| 245 | t.Fatal("wall rollback extended deadline") |
| 246 | } |
| 247 | } |
| 248 | |
| 249 | func TestAmbiguousPUTLatchesAndConditionalReleasePreservesSuccessor(t *testing.T) { |
| 250 | cfg, _, _, store, _ := fixture(t) |
| 251 | store.failAt = 1 |
| 252 | c, err := Open(t.Context(), cfg) |
| 253 | if err != nil { |
| 254 | t.Fatal(err) |
| 255 | } |
| 256 | if _, err = c.Ready(t.Context()); err == nil { |
| 257 | t.Fatal("ambiguous PUT admitted") |
| 258 | } |
| 259 | closeController(t, c) |
| 260 | cfg, _, _, store, _ = fixture(t) |
| 261 | c, err = Open(t.Context(), cfg) |
| 262 | if err != nil { |
| 263 | t.Fatal(err) |
| 264 | } |
| 265 | s := ready(t, c) |
| 266 | key, _ := Key(s.Host, s.Box) |
| 267 | successor := []byte(`{"op":"successor","expires":"2030-01-01T00:00:00Z"}`) |
| 268 | if _, err = store.Store.Put(t.Context(), key, successor, s.LeaseETag); err != nil { |
| 269 | t.Fatal(err) |
| 270 | } |
| 271 | if err = c.Close(t.Context()); !errors.Is(err, release.ErrCASConflict) { |
| 272 | t.Fatalf("release = %v", err) |
| 273 | } |
| 274 | got, _, err := store.Get(t.Context(), key) |
| 275 | if err != nil || string(got) != string(successor) { |
| 276 | t.Fatal("deleted successor") |
| 277 | } |
| 278 | } |
| 279 | |
| 280 | func TestControllerExclusionBootAndCC9(t *testing.T) { |
| 281 | cfg, r, clock, _, _ := fixture(t) |
| 282 | c, err := Open(t.Context(), cfg) |
| 283 | if err != nil { |
| 284 | t.Fatal(err) |
| 285 | } |
| 286 | s := ready(t, c) |
| 287 | ctx, cancel := context.WithTimeout(t.Context(), 30*time.Millisecond) |
| 288 | defer cancel() |
| 289 | if _, err = Open(ctx, cfg); !errors.Is(err, context.DeadlineExceeded) { |
| 290 | t.Fatalf("second controller = %v", err) |
| 291 | } |
| 292 | wrong := s.Binding |
| 293 | wrong.BootID = "next-boot" |
| 294 | if err = CheckLocal(cfg.Guard.DataDir, wrong, clock, cfg.LockDir); err == nil { |
| 295 | t.Fatal("inherited prior boot") |
| 296 | } |
| 297 | closeController(t, c) |
| 298 | if err = ClearStale(t.Context(), cfg); !errors.Is(err, ErrInactive) { |
| 299 | t.Fatalf("cleared active op: %v", err) |
| 300 | } |
| 301 | r.mu.Lock() |
| 302 | r.q.Op = hostclaim.Op{Kind: hostclaim.KindServing, Participants: []string{"box"}} |
| 303 | r.mu.Unlock() |
| 304 | if err = ClearStale(t.Context(), cfg); err != nil { |
| 305 | t.Fatal(err) |
| 306 | } |
| 307 | if _, err = os.Stat(filepath.Join(cfg.Guard.DataDir, ".hostexec", cfg.Binding.Host+".json")); !errors.Is(err, os.ErrNotExist) { |
| 308 | t.Fatal("stale state remains", err) |
| 309 | } |
| 310 | // No lease traffic for retained serving participants, even when enabled. |
| 311 | if _, err = Open(t.Context(), cfg); !errors.Is(err, ErrInactive) { |
| 312 | t.Fatalf("serving controller = %v", err) |
| 313 | } |
| 314 | } |
| 315 | |
| 316 | func TestMalformedPendingMissingAndQuarantineRefuse(t *testing.T) { |
| 317 | cfg, _, clock, _, _ := fixture(t) |
| 318 | b := cfg.Binding |
| 319 | b.ControllerID = "controller" |
| 320 | if err := CheckLocal(cfg.Guard.DataDir, b, clock, cfg.LockDir); err == nil { |
| 321 | t.Fatal("missing state admitted") |
| 322 | } |
| 323 | p := filepath.Join(cfg.Guard.DataDir, ".hostexec", b.Host+".json") |
| 324 | if err := os.WriteFile(p, []byte(`{"v":1}`), 0660); err != nil { |
| 325 | t.Fatal(err) |
| 326 | } |
| 327 | if _, err := Open(t.Context(), cfg); err == nil { |
| 328 | t.Fatal("malformed state replaced") |
| 329 | } |
| 330 | if err := os.Remove(p); err != nil { |
| 331 | t.Fatal(err) |
| 332 | } |
| 333 | cfg.Directory = directory{state: "quarantined"} |
| 334 | if _, err := Open(t.Context(), cfg); err == nil { |
| 335 | t.Fatal("quarantined box admitted") |
| 336 | } |
| 337 | } |
| 338 | |
| 339 | func TestLocalPathsPinnedAndSymlinkRefused(t *testing.T) { |
| 340 | cfg, _, _, _, _ := fixture(t) |
| 341 | outside := filepath.Join(t.TempDir(), "outside") |
| 342 | if err := os.WriteFile(outside, []byte("untouched"), 0600); err != nil { |
| 343 | t.Fatal(err) |
| 344 | } |
| 345 | record := filepath.Join(cfg.Guard.DataDir, ".hostexec", cfg.Binding.Host+".json") |
| 346 | if err := os.Symlink(outside, record); err != nil { |
| 347 | t.Fatal(err) |
| 348 | } |
| 349 | if _, err := Open(t.Context(), cfg); err == nil { |
| 350 | t.Fatal("symlink admitted") |
| 351 | } |
| 352 | b, _ := os.ReadFile(outside) |
| 353 | st, _ := os.Stat(outside) |
| 354 | if string(b) != "untouched" || st.Mode().Perm() != 0600 { |
| 355 | t.Fatal("changed symlink target") |
| 356 | } |
| 357 | if err := os.Remove(record); err != nil { |
| 358 | t.Fatal(err) |
| 359 | } |
| 360 | d, err := openLocal(cfg.Guard.DataDir) |
| 361 | if err != nil { |
| 362 | t.Fatal(err) |
| 363 | } |
| 364 | defer d.close() |
| 365 | old := filepath.Join(cfg.Guard.DataDir, ".hostexec") |
| 366 | if err = os.Rename(old, old+".pinned"); err != nil { |
| 367 | t.Fatal(err) |
| 368 | } |
| 369 | if err = os.Mkdir(old, 0770); err != nil { |
| 370 | t.Fatal(err) |
| 371 | } |
| 372 | s := State{V: 1, Binding: cfg.Binding, State: "pending", Expires: time.Now().Add(TTL), SafeUntil: 99} |
| 373 | s.ControllerID = "controller" |
| 374 | if err = d.write(s); err != nil { |
| 375 | t.Fatal(err) |
| 376 | } |
| 377 | if _, err = os.Stat(filepath.Join(old+".pinned", s.Host+".json")); err != nil { |
| 378 | t.Fatal(err) |
| 379 | } |
| 380 | if _, err = os.Stat(filepath.Join(old, s.Host+".json")); !errors.Is(err, os.ErrNotExist) { |
| 381 | t.Fatal("followed directory replacement") |
| 382 | } |
| 383 | d.sync = func(int) error { return errors.New("fsync failure") } |
| 384 | if err = d.write(s); err == nil { |
| 385 | t.Fatal("ignored persistence failure") |
| 386 | } |
| 387 | entries, _ := os.ReadDir(old + ".pinned") |
| 388 | for _, e := range entries { |
| 389 | if strings.HasPrefix(e.Name(), ".tmp-") { |
| 390 | t.Fatal("leaked temp file") |
| 391 | } |
| 392 | } |
| 393 | } |
| 394 | |
| 395 | func TestSuccessfulRenewalDurablyExtendsFromItsRequestStart(t *testing.T) { |
| 396 | cfg, _, clock, store, _ := fixture(t) |
| 397 | c, err := Open(t.Context(), cfg) |
| 398 | if err != nil { |
| 399 | t.Fatal(err) |
| 400 | } |
| 401 | defer closeController(t, c) |
| 402 | first := ready(t, c) |
| 403 | until(t, func() bool { |
| 404 | clock.mu.Lock() |
| 405 | defer clock.mu.Unlock() |
| 406 | for _, a := range clock.alarms { |
| 407 | if a.at == clock.boot+int64(Renewal) { |
| 408 | return true |
| 409 | } |
| 410 | } |
| 411 | return false |
| 412 | }) |
| 413 | clock.advance(Renewal) |
| 414 | until(t, func() bool { |
| 415 | c.mu.Lock() |
| 416 | defer c.mu.Unlock() |
| 417 | return c.state.SafeUntil == first.SafeUntil+int64(Renewal) |
| 418 | }) |
| 419 | renewed := ready(t, c) |
| 420 | if renewed.LeaseETag == first.LeaseETag || !renewed.Expires.Equal(first.Expires.Add(Renewal)) { |
| 421 | t.Fatalf("renewal did not advance coherent tuple: %+v", renewed) |
| 422 | } |
| 423 | d, err := openLocal(cfg.Guard.DataDir) |
| 424 | if err != nil { |
| 425 | t.Fatal(err) |
| 426 | } |
| 427 | defer d.close() |
| 428 | disk, err := d.read(renewed.Host) |
| 429 | if err != nil || disk != renewed { |
| 430 | t.Fatalf("renewal not durable: %+v %v", disk, err) |
| 431 | } |
| 432 | key, _ := Key(renewed.Host, renewed.Box) |
| 433 | raw, etag, err := store.Get(t.Context(), key) |
| 434 | if err != nil || etag != renewed.LeaseETag { |
| 435 | t.Fatalf("renewed remote ownership: %q %v", etag, err) |
| 436 | } |
| 437 | var wire Record |
| 438 | if err = json.Unmarshal(raw, &wire); err != nil || wire.Op != renewed.Op || !wire.Expires.Equal(renewed.Expires) { |
| 439 | t.Fatalf("renewed wire: %+v %v", wire, err) |
| 440 | } |
| 441 | } |
| 442 | |
amadan