| 1 | package hostexec |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/rand" |
| 6 | "encoding/hex" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "github.com/carlosframework/platform/internal/version" |
| 10 | "os" |
| 11 | "runtime" |
| 12 | "slices" |
| 13 | "sync" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/hostclaim" |
| 17 | "github.com/carlosframework/platform/internal/hostfloor" |
| 18 | "github.com/carlosframework/platform/internal/hostguard" |
| 19 | "github.com/carlosframework/platform/internal/hostquarantine" |
| 20 | "github.com/carlosframework/platform/internal/release" |
| 21 | "path/filepath" |
| 22 | ) |
| 23 | |
| 24 | type Config struct { |
| 25 | Binding Binding |
| 26 | Store release.Store |
| 27 | Guard hostguard.Guard |
| 28 | Directory hostquarantine.ActorDirectory |
| 29 | Clock Clock |
| 30 | UpdateGuardDir string |
| 31 | UpdateJournal string |
| 32 | verifyRunning func(context.Context) error |
| 33 | LockDir string |
| 34 | // These close/cancel hooks must be nonblocking and safe to call repeatedly. |
| 35 | CloseAdmission func() |
| 36 | CancelWork func() |
| 37 | // Cleanup terminates only this host's processes and closes its replicators. |
| 38 | // It must not wait for the host guard: uncertain disk writes retain that |
| 39 | // guard to deny other processes, while termination must proceed independently. |
| 40 | // Its failure remains a separate denial owned by the caller, even after CC9. |
| 41 | Cleanup func(context.Context) error |
| 42 | writeLocal func(*localDir, State) error // test seam; nil uses durable descriptor-relative I/O |
| 43 | } |
| 44 | |
| 45 | type Controller struct { |
| 46 | cfg Config |
| 47 | mu sync.Mutex |
| 48 | state State |
| 49 | err error |
| 50 | cleanupErr error |
| 51 | releaseETag string |
| 52 | cleanupOnce sync.Once |
| 53 | cleanupDone chan struct{} |
| 54 | ready chan struct{} |
| 55 | readyOnce sync.Once |
| 56 | lost chan struct{} |
| 57 | ctx context.Context |
| 58 | cancel context.CancelFunc |
| 59 | done chan struct{} |
| 60 | watchDone chan struct{} |
| 61 | unlock func() |
| 62 | unlive func() |
| 63 | closeMu sync.Mutex |
| 64 | closed bool |
| 65 | } |
| 66 | |
| 67 | func Open(ctx context.Context, cfg Config) (*Controller, error) { |
| 68 | if cfg.Clock == nil { |
| 69 | cfg.Clock = SystemClock{} |
| 70 | } |
| 71 | if cfg.LockDir == "" { |
| 72 | if cfg.UpdateGuardDir == "" { |
| 73 | cfg.UpdateGuardDir = "/opt/carlos/system/guard" |
| 74 | cfg.UpdateJournal = filepath.Join("/opt/carlos/system/router", "journal.json") |
| 75 | } |
| 76 | cfg.LockDir = "/run/carlos/hostexec" |
| 77 | } |
| 78 | var nonce [16]byte |
| 79 | if _, err := rand.Read(nonce[:]); err != nil { |
| 80 | return nil, err |
| 81 | } |
| 82 | cfg.Binding.ControllerID = hex.EncodeToString(nonce[:]) |
| 83 | if !cfg.Binding.valid() || cfg.Store == nil || cfg.Guard.Claims == nil || cfg.Directory == nil || |
| 84 | cfg.CloseAdmission == nil || cfg.CancelWork == nil || cfg.Cleanup == nil { |
| 85 | return nil, ErrState |
| 86 | } |
| 87 | wall, boot, err := cfg.Clock.Now() |
| 88 | if err != nil { |
| 89 | return nil, err |
| 90 | } |
| 91 | process, err := currentControllerProcess() |
| 92 | if err != nil { |
| 93 | return nil, err |
| 94 | } |
| 95 | cctx, cancel := context.WithCancel(ctx) |
| 96 | c := &Controller{cfg: cfg, state: State{Process: process, V: 1, Binding: cfg.Binding, State: "pending", Expires: wall.Add(TTL), SafeUntil: boot + int64(TTL-Margin)}, ctx: cctx, cancel: cancel, |
| 97 | ready: make(chan struct{}), lost: make(chan struct{}), done: make(chan struct{}), watchDone: make(chan struct{}), cleanupDone: make(chan struct{})} |
| 98 | // Hold the host guard across lifetime-lock acquisition and the pending |
| 99 | // write. Acquire the live lock only after pending state is durable, so |
| 100 | // orphaned active bytes cannot regain admission. Never wait for the |
| 101 | // lifetime lock while holding the guard. |
| 102 | var cleaned <-chan error |
| 103 | pendingWritten := false |
| 104 | for { |
| 105 | err = cfg.Guard.WithLocal(ctx, cfg.Binding.Host, func() error { |
| 106 | var lockErr error |
| 107 | c.unlock, lockErr = tryAcquire(cfg.LockDir, cfg.Binding.Host) |
| 108 | if lockErr != nil { |
| 109 | return lockErr |
| 110 | } |
| 111 | // Start cleanup before any persistence can block. Exclusion stays |
| 112 | // held until this independent cleanup has joined on every exit. |
| 113 | result := make(chan error, 1) |
| 114 | cleaned = result |
| 115 | go func() { result <- cfg.Cleanup(context.Background()) }() |
| 116 | cfg.CloseAdmission() |
| 117 | cfg.CancelWork() |
| 118 | d, openErr := openLocal(cfg.Guard.DataDir) |
| 119 | if openErr != nil { |
| 120 | return openErr |
| 121 | } |
| 122 | defer d.close() |
| 123 | if writeErr := c.write(d, c.state); writeErr != nil { |
| 124 | return writeErr |
| 125 | } |
| 126 | pendingWritten = true |
| 127 | if cfg.UpdateGuardDir != "" { |
| 128 | if err := cfg.checkUpdateIdentity(ctx); err != nil { |
| 129 | return err |
| 130 | } |
| 131 | } |
| 132 | c.unlive, lockErr = tryAcquireLive(cfg.LockDir, cfg.Binding.Host) |
| 133 | return lockErr |
| 134 | }) |
| 135 | if c.unlock != nil || !errors.Is(err, ErrControllerBusy) { |
| 136 | break |
| 137 | } |
| 138 | select { |
| 139 | case <-ctx.Done(): |
| 140 | err = ctx.Err() |
| 141 | case <-time.After(10 * time.Millisecond): |
| 142 | continue |
| 143 | } |
| 144 | break |
| 145 | } |
| 146 | if err == nil { |
| 147 | err = <-cleaned |
| 148 | cleaned = nil |
| 149 | } |
| 150 | if err == nil { |
| 151 | err = c.authority(ctx) |
| 152 | } |
| 153 | if err != nil { |
| 154 | pending := c.state |
| 155 | c.state.State = "lost" |
| 156 | hadLive := c.unlive != nil |
| 157 | if c.unlive != nil { |
| 158 | c.unlive() |
| 159 | c.unlive = nil |
| 160 | } |
| 161 | if c.unlock != nil { |
| 162 | // Before live-lock acquisition, no local state can authorize work. |
| 163 | // A failed pending write (including an unsafe path) must not make |
| 164 | // cleanup wait for another write to the same failed destination. |
| 165 | if hadLive { |
| 166 | err = errors.Join(err, c.save(c.state)) |
| 167 | } |
| 168 | cleanupErr := error(nil) |
| 169 | if cleaned != nil { |
| 170 | cleanupErr = <-cleaned |
| 171 | err = errors.Join(err, cleanupErr) |
| 172 | } |
| 173 | // No live lock or remote authority was issued. After positive joined |
| 174 | // cleanup, abort only this Open's exact pending tuple while retaining |
| 175 | // lifetime exclusion. Failed/uncertain cleanup retains its denial. |
| 176 | if !hadLive && pendingWritten && cleanupErr == nil { |
| 177 | err = errors.Join(err, cfg.Guard.WithLocal(context.Background(), pending.Host, func() error { |
| 178 | d, e := openLocal(cfg.Guard.DataDir) |
| 179 | if e != nil { |
| 180 | return e |
| 181 | } |
| 182 | defer d.close() |
| 183 | current, e := d.read(pending.Host) |
| 184 | if e != nil { |
| 185 | return e |
| 186 | } |
| 187 | // JSON retains the instant, not time.Time's monotonic/location |
| 188 | // metadata. Compare that field semantically, all others exactly. |
| 189 | if !current.Expires.Equal(pending.Expires) { |
| 190 | return ErrState |
| 191 | } |
| 192 | current.Expires = pending.Expires |
| 193 | if current != pending { |
| 194 | return ErrState |
| 195 | } |
| 196 | return d.remove(pending.Host) |
| 197 | })) |
| 198 | } |
| 199 | c.unlock() |
| 200 | } |
| 201 | cancel() |
| 202 | return nil, err |
| 203 | } |
| 204 | go c.watch() |
| 205 | go c.run() |
| 206 | return c, nil |
| 207 | } |
| 208 | |
| 209 | func (c *Controller) write(d *localDir, s State) error { |
| 210 | if c.cfg.writeLocal != nil { |
| 211 | return c.cfg.writeLocal(d, s) |
| 212 | } |
| 213 | return d.write(s) |
| 214 | } |
| 215 | |
| 216 | // save retains the guard until lost state is durable, including failures before |
| 217 | // rename. The revoked live lock denies CheckLocal while the old file survives. |
| 218 | func (c *Controller) save(s State) error { |
| 219 | return c.cfg.Guard.WithLocal(context.Background(), s.Host, func() error { |
| 220 | for { |
| 221 | d, err := openLocal(c.cfg.Guard.DataDir) |
| 222 | if err == nil { |
| 223 | err = c.write(d, s) |
| 224 | d.close() |
| 225 | } |
| 226 | if err == nil || s.State != "lost" { |
| 227 | return err |
| 228 | } |
| 229 | time.Sleep(100 * time.Millisecond) |
| 230 | } |
| 231 | }) |
| 232 | } |
| 233 | func (c *Controller) authority(ctx context.Context) error { |
| 234 | if err := c.cfg.checkUpdateIdentity(ctx); err != nil { |
| 235 | return err |
| 236 | } |
| 237 | if err := c.cfg.checkRemoteUpdateIdentity(ctx); err != nil { |
| 238 | return err |
| 239 | } |
| 240 | b := c.cfg.Binding |
| 241 | if err := hostquarantine.Check(ctx, c.cfg.Directory, b.Box, b.Enrollment); err != nil { |
| 242 | return err |
| 243 | } |
| 244 | obs, err := c.cfg.Guard.Claims.Read(ctx, b.Host) |
| 245 | if err != nil { |
| 246 | return err |
| 247 | } |
| 248 | if obs.Absent || obs.Claim == nil || obs.Host != b.Host || obs.Claim.Host != b.Host { |
| 249 | return ErrInactive |
| 250 | } |
| 251 | q := obs.Claim |
| 252 | if q.Validate() != nil || q.Holder == nil || q.Holder.Incarnation != b.Incarnation || q.Op.ID != b.Op || |
| 253 | q.Op.Kind == hostclaim.KindServing || q.Free() || !slices.Contains(q.Op.Participants, b.Box) || slices.Contains(q.Lost, b.Box) { |
| 254 | return ErrInactive |
| 255 | } |
| 256 | return nil |
| 257 | } |
| 258 | |
| 259 | // loseLocked latches denial before attempting disk I/O or process cleanup. |
| 260 | func (c *Controller) loseLocked(err error) { |
| 261 | if c.err != nil { |
| 262 | return |
| 263 | } |
| 264 | c.err = err |
| 265 | c.state.State = "lost" |
| 266 | // Revoke cross-process admission before publishing loss or starting I/O. |
| 267 | // This close needs no disk space. The lifetime lock remains held. |
| 268 | if c.unlive != nil { |
| 269 | c.unlive() |
| 270 | c.unlive = nil |
| 271 | } |
| 272 | close(c.lost) |
| 273 | c.cancel() |
| 274 | c.cfg.CloseAdmission() |
| 275 | c.cfg.CancelWork() |
| 276 | c.readyOnce.Do(func() { close(c.ready) }) |
| 277 | s := c.state |
| 278 | c.cleanupOnce.Do(func() { |
| 279 | go func() { |
| 280 | // Neither a blocked disk nor blocked cleanup can prevent the other. |
| 281 | cleaned := make(chan error, 1) |
| 282 | go func() { cleaned <- c.cfg.Cleanup(context.Background()) }() |
| 283 | persistErr := c.save(s) |
| 284 | cleanupErr := <-cleaned |
| 285 | c.mu.Lock() |
| 286 | c.cleanupErr = errors.Join(persistErr, cleanupErr) |
| 287 | c.mu.Unlock() |
| 288 | close(c.cleanupDone) |
| 289 | }() |
| 290 | }) |
| 291 | } |
| 292 | func (c *Controller) checkLocked() error { |
| 293 | if c.err != nil { |
| 294 | return c.err |
| 295 | } |
| 296 | _, boot, err := c.cfg.Clock.Now() |
| 297 | if err != nil { |
| 298 | c.loseLocked(err) |
| 299 | } else if boot >= c.state.SafeUntil { |
| 300 | c.loseLocked(ErrLost) |
| 301 | } |
| 302 | return c.err |
| 303 | } |
| 304 | func (c *Controller) Lost() <-chan struct{} { return c.lost } |
| 305 | func (c *Controller) Ready(ctx context.Context) (State, error) { |
| 306 | select { |
| 307 | case <-ctx.Done(): |
| 308 | return State{}, ctx.Err() |
| 309 | case <-c.ready: |
| 310 | } |
| 311 | c.mu.Lock() |
| 312 | defer c.mu.Unlock() |
| 313 | if err := c.checkLocked(); err != nil { |
| 314 | return State{}, err |
| 315 | } |
| 316 | if c.state.State != "active" { |
| 317 | return State{}, ErrState |
| 318 | } |
| 319 | return c.state, nil |
| 320 | } |
| 321 | func (c *Controller) CleanupError() error { c.mu.Lock(); defer c.mu.Unlock(); return c.cleanupErr } |
| 322 | |
| 323 | func (c *Controller) watch() { |
| 324 | defer close(c.watchDone) |
| 325 | for { |
| 326 | c.mu.Lock() |
| 327 | err := c.checkLocked() |
| 328 | c.mu.Unlock() |
| 329 | if err != nil { |
| 330 | return |
| 331 | } |
| 332 | // CLOCK_BOOTTIME is sampled on every wake and every admission. A normal Go |
| 333 | // timer does not itself prove that a suspended machine has remained stopped. |
| 334 | select { |
| 335 | case <-c.ctx.Done(): |
| 336 | c.mu.Lock() |
| 337 | c.loseLocked(c.ctx.Err()) |
| 338 | c.mu.Unlock() |
| 339 | return |
| 340 | case <-c.cfg.Clock.After(100 * time.Millisecond): |
| 341 | } |
| 342 | } |
| 343 | } |
| 344 | func (c *Controller) run() { |
| 345 | defer close(c.done) |
| 346 | key, _ := Key(c.cfg.Binding.Host, c.cfg.Binding.Box) |
| 347 | body, etag, err := c.cfg.Store.Get(c.ctx, key) |
| 348 | if errors.Is(err, os.ErrNotExist) { |
| 349 | err = nil |
| 350 | etag = "" |
| 351 | } else if err == nil { |
| 352 | var r Record |
| 353 | if decode(body, &r) != nil || r.Op == "" || r.Expires.IsZero() || etag == "" { |
| 354 | err = ErrState |
| 355 | } |
| 356 | } |
| 357 | if err != nil { |
| 358 | c.mu.Lock() |
| 359 | c.loseLocked(err) |
| 360 | c.mu.Unlock() |
| 361 | return |
| 362 | } |
| 363 | for { |
| 364 | if err = c.authority(c.ctx); err != nil { |
| 365 | c.mu.Lock() |
| 366 | c.loseLocked(err) |
| 367 | c.mu.Unlock() |
| 368 | return |
| 369 | } |
| 370 | wall, start, clockErr := c.cfg.Clock.Now() |
| 371 | if clockErr != nil { |
| 372 | c.mu.Lock() |
| 373 | c.loseLocked(clockErr) |
| 374 | c.mu.Unlock() |
| 375 | return |
| 376 | } |
| 377 | c.mu.Lock() |
| 378 | err = c.checkLocked() |
| 379 | captured := c.state.Binding |
| 380 | c.mu.Unlock() |
| 381 | if err != nil { |
| 382 | return |
| 383 | } |
| 384 | proposed := Record{Op: captured.Op, Expires: wall.Add(TTL)} |
| 385 | bytes, _ := json.Marshal(proposed) |
| 386 | next, putErr := c.cfg.Store.Put(c.ctx, key, bytes, etag) |
| 387 | c.mu.Lock() |
| 388 | // A late acknowledgement can only contribute its exact ETag to release. |
| 389 | if captured != c.state.Binding { |
| 390 | c.mu.Unlock() |
| 391 | return |
| 392 | } |
| 393 | if putErr == nil && next != "" { |
| 394 | c.releaseETag = next |
| 395 | etag = next |
| 396 | } |
| 397 | if err = c.checkLocked(); err == nil && (putErr != nil || next == "") { |
| 398 | if putErr == nil { |
| 399 | putErr = ErrState |
| 400 | } |
| 401 | c.loseLocked(putErr) |
| 402 | err = putErr |
| 403 | } |
| 404 | c.mu.Unlock() |
| 405 | if err != nil { |
| 406 | return |
| 407 | } |
| 408 | // Network reads never hold either the controller mutex or the host guard. |
| 409 | if err = c.authority(c.ctx); err != nil { |
| 410 | c.mu.Lock() |
| 411 | c.loseLocked(err) |
| 412 | c.mu.Unlock() |
| 413 | return |
| 414 | } |
| 415 | c.mu.Lock() |
| 416 | if err = c.checkLocked(); err != nil { |
| 417 | c.mu.Unlock() |
| 418 | return |
| 419 | } |
| 420 | nextState := c.state |
| 421 | nextState.State = "active" |
| 422 | nextState.LeaseETag = next |
| 423 | nextState.Expires = proposed.Expires |
| 424 | nextState.SafeUntil = start + int64(TTL-Margin) |
| 425 | // Never use a response-relative deadline, including when the wall clock moved. |
| 426 | _, now, clockErr := c.cfg.Clock.Now() |
| 427 | if clockErr != nil || now >= nextState.SafeUntil { |
| 428 | c.loseLocked(ErrLost) |
| 429 | c.mu.Unlock() |
| 430 | return |
| 431 | } |
| 432 | c.mu.Unlock() |
| 433 | // Publication and any intervening loss are serialized under the same host |
| 434 | // guard, so another process never observes a stale active overwrite. |
| 435 | if err = c.publish(nextState); err != nil { |
| 436 | c.mu.Lock() |
| 437 | c.loseLocked(err) |
| 438 | c.mu.Unlock() |
| 439 | return |
| 440 | } |
| 441 | _, now, clockErr = c.cfg.Clock.Now() |
| 442 | if clockErr != nil { |
| 443 | c.mu.Lock() |
| 444 | c.loseLocked(clockErr) |
| 445 | c.mu.Unlock() |
| 446 | return |
| 447 | } |
| 448 | delay := time.Duration(start + int64(Renewal) - now) |
| 449 | if delay < 0 { |
| 450 | delay = 0 |
| 451 | } |
| 452 | select { |
| 453 | case <-c.ctx.Done(): |
| 454 | return |
| 455 | case <-c.cfg.Clock.After(delay): |
| 456 | } |
| 457 | } |
| 458 | } |
| 459 | |
| 460 | // publish leaves mu available to the watchdog during filesystem calls. If a |
| 461 | // write has an uncertain outcome, it retains the host guard until denial is |
| 462 | // durable; releasing it over possibly active bytes would admit another process. |
| 463 | func (c *Controller) publish(next State) error { |
| 464 | return c.cfg.Guard.WithLocal(context.Background(), next.Host, func() error { |
| 465 | d, err := openLocal(c.cfg.Guard.DataDir) |
| 466 | if err != nil { |
| 467 | return err |
| 468 | } |
| 469 | defer d.close() |
| 470 | c.mu.Lock() |
| 471 | err = c.checkLocked() |
| 472 | if err == nil && next.Binding != c.state.Binding { |
| 473 | c.loseLocked(ErrState) |
| 474 | err = ErrState |
| 475 | } |
| 476 | c.mu.Unlock() |
| 477 | if err != nil { |
| 478 | return err |
| 479 | } |
| 480 | writeErr := c.write(d, next) |
| 481 | c.mu.Lock() |
| 482 | if writeErr != nil { |
| 483 | c.loseLocked(writeErr) |
| 484 | } |
| 485 | err = c.checkLocked() |
| 486 | if err == nil { |
| 487 | _, boot, clockErr := c.cfg.Clock.Now() |
| 488 | if clockErr != nil || boot >= next.SafeUntil { |
| 489 | c.loseLocked(ErrLost) |
| 490 | err = c.err |
| 491 | } |
| 492 | } |
| 493 | if err == nil { |
| 494 | c.state = next |
| 495 | c.readyOnce.Do(func() { close(c.ready) }) |
| 496 | c.mu.Unlock() |
| 497 | return nil |
| 498 | } |
| 499 | denied := c.state |
| 500 | denied.State = "lost" |
| 501 | c.mu.Unlock() |
| 502 | // Cancellation does not release the guard over uncertain active bytes. |
| 503 | // Close may time out; it keeps the lifetime lock until this join completes. |
| 504 | for { |
| 505 | if persistErr := c.write(d, denied); persistErr == nil { |
| 506 | return err |
| 507 | } |
| 508 | time.Sleep(100 * time.Millisecond) |
| 509 | } |
| 510 | }) |
| 511 | } |
| 512 | |
| 513 | // Close cancels and joins all lease requests before conditional release. It |
| 514 | // retains the lifetime lock on a timed-out join; call Close again to finish. |
| 515 | func (c *Controller) Close(ctx context.Context) error { |
| 516 | c.closeMu.Lock() |
| 517 | defer c.closeMu.Unlock() |
| 518 | if c.closed { |
| 519 | return nil |
| 520 | } |
| 521 | c.mu.Lock() |
| 522 | c.loseLocked(ErrLost) |
| 523 | c.mu.Unlock() |
| 524 | for _, done := range []<-chan struct{}{c.done, c.watchDone, c.cleanupDone} { |
| 525 | select { |
| 526 | case <-ctx.Done(): |
| 527 | return ctx.Err() |
| 528 | case <-done: |
| 529 | } |
| 530 | } |
| 531 | c.mu.Lock() |
| 532 | etag := c.releaseETag |
| 533 | c.mu.Unlock() |
| 534 | var err error |
| 535 | if etag != "" { |
| 536 | key, _ := Key(c.cfg.Binding.Host, c.cfg.Binding.Box) |
| 537 | err = release.DeleteIfMatch(ctx, c.cfg.Store, key, etag) |
| 538 | } |
| 539 | c.closed = true |
| 540 | c.unlock() |
| 541 | return errors.Join(err, c.CleanupError()) |
| 542 | } |
| 543 | |
| 544 | // CheckLocal must run inside Guard.With, before each bounded host write. It |
| 545 | // does not grant authority and never clears quarantine or cleanup denial. |
| 546 | func CheckLocal(dataDir string, expected Binding, clock Clock, lockDir string) error { |
| 547 | if !expected.valid() || clock == nil { |
| 548 | return ErrState |
| 549 | } |
| 550 | if lockDir == "" { |
| 551 | lockDir = "/run/carlos/hostexec" |
| 552 | } |
| 553 | // Guard.With serializes these probes with other writers and Open. A free |
| 554 | // live lock denies even if ENOSPC left a future-dated active file behind. |
| 555 | for _, probe := range []func(string, string) (func(), error){probeLifetime, probeLive} { |
| 556 | unlock, err := probe(lockDir, expected.Host) |
| 557 | if err == nil { |
| 558 | unlock() |
| 559 | return ErrLost |
| 560 | } |
| 561 | if errors.Is(err, os.ErrNotExist) { |
| 562 | return ErrLost |
| 563 | } |
| 564 | if !errors.Is(err, ErrControllerBusy) { |
| 565 | return err |
| 566 | } |
| 567 | } |
| 568 | d, err := openLocal(dataDir) |
| 569 | if err != nil { |
| 570 | return err |
| 571 | } |
| 572 | defer d.close() |
| 573 | s, err := d.read(expected.Host) |
| 574 | if err != nil { |
| 575 | return err |
| 576 | } |
| 577 | if s.Binding != expected || s.State != "active" { |
| 578 | return ErrLost |
| 579 | } |
| 580 | _, boot, err := clock.Now() |
| 581 | if err != nil { |
| 582 | return err |
| 583 | } |
| 584 | if boot >= s.SafeUntil { |
| 585 | return ErrLost |
| 586 | } |
| 587 | return nil |
| 588 | } |
| 589 | |
| 590 | // ClearStale performs CC9 with the lifetime lock held, after any old controller |
| 591 | // has joined. A failed or mismatched claim read leaves local denial intact. |
| 592 | func ClearStale(ctx context.Context, cfg Config) error { |
| 593 | lockDir := cfg.LockDir |
| 594 | if lockDir == "" { |
| 595 | lockDir = "/run/carlos/hostexec" |
| 596 | } |
| 597 | if cfg.Guard.Claims == nil || !validID(cfg.Binding.Box) || !validID(cfg.Binding.Enrollment) { |
| 598 | return ErrState |
| 599 | } |
| 600 | var unlock func() |
| 601 | var err error |
| 602 | for { |
| 603 | err = cfg.Guard.WithLocal(ctx, cfg.Binding.Host, func() error { |
| 604 | var lockErr error |
| 605 | unlock, lockErr = tryAcquire(lockDir, cfg.Binding.Host) |
| 606 | if lockErr != nil { |
| 607 | return lockErr |
| 608 | } |
| 609 | denied := false |
| 610 | defer func() { |
| 611 | if !denied { |
| 612 | unlock() |
| 613 | unlock = nil |
| 614 | } |
| 615 | }() |
| 616 | d, openErr := openLocal(cfg.Guard.DataDir) |
| 617 | if openErr != nil { |
| 618 | return openErr |
| 619 | } |
| 620 | defer d.close() |
| 621 | old, readErr := d.read(cfg.Binding.Host) |
| 622 | if errors.Is(readErr, os.ErrNotExist) { |
| 623 | denied = true |
| 624 | return nil |
| 625 | } |
| 626 | if readErr != nil { |
| 627 | return readErr |
| 628 | } |
| 629 | if old.Box != cfg.Binding.Box || old.Enrollment != cfg.Binding.Enrollment || |
| 630 | old.Op != cfg.Binding.Op || old.Incarnation != cfg.Binding.Incarnation { |
| 631 | return ErrState |
| 632 | } |
| 633 | // Holding the lifetime lock must never make an orphaned active record |
| 634 | // appear to belong to a running controller during the claim read. |
| 635 | old.State = "lost" |
| 636 | writeErr := d.write(old) |
| 637 | denied = writeErr == nil |
| 638 | return writeErr |
| 639 | }) |
| 640 | if !errors.Is(err, ErrControllerBusy) { |
| 641 | break |
| 642 | } |
| 643 | select { |
| 644 | case <-ctx.Done(): |
| 645 | return ctx.Err() |
| 646 | case <-time.After(10 * time.Millisecond): |
| 647 | } |
| 648 | } |
| 649 | if err != nil { |
| 650 | return err |
| 651 | } |
| 652 | defer unlock() |
| 653 | obs, err := cfg.Guard.Claims.Read(ctx, cfg.Binding.Host) |
| 654 | if err != nil { |
| 655 | return err |
| 656 | } |
| 657 | if obs.Absent || obs.Claim == nil || obs.Host != cfg.Binding.Host || obs.Claim.Host != cfg.Binding.Host || obs.Claim.Validate() != nil { |
| 658 | return ErrState |
| 659 | } |
| 660 | q := obs.Claim |
| 661 | return cfg.Guard.WithLocal(ctx, cfg.Binding.Host, func() error { |
| 662 | d, err := openLocal(cfg.Guard.DataDir) |
| 663 | if err != nil { |
| 664 | return err |
| 665 | } |
| 666 | defer d.close() |
| 667 | old, err := d.read(cfg.Binding.Host) |
| 668 | if errors.Is(err, os.ErrNotExist) { |
| 669 | return nil |
| 670 | } |
| 671 | if err != nil { |
| 672 | return err |
| 673 | } |
| 674 | if old.Box != cfg.Binding.Box || old.Enrollment != cfg.Binding.Enrollment || |
| 675 | old.Op != cfg.Binding.Op || old.Incarnation != cfg.Binding.Incarnation { |
| 676 | return ErrState |
| 677 | } |
| 678 | // Only the caller's expected operation is cleared. A fresh successor |
| 679 | // operation may exclude this box; that is also CC9 discharge evidence. |
| 680 | if !q.Free() && q.Op.Kind != hostclaim.KindServing && |
| 681 | slices.Contains(q.Op.Participants, old.Box) { |
| 682 | return ErrInactive |
| 683 | } |
| 684 | return d.remove(old.Host) |
| 685 | }) |
| 686 | } |
| 687 | |
| 688 | // checkUpdateIdentity reads the executing inode, never the replaceable current |
| 689 | // path. Run it after pending is durable and before each acknowledgement. |
| 690 | func (cfg Config) checkUpdateIdentity(ctx context.Context) error { |
| 691 | if cfg.UpdateGuardDir == "" { |
| 692 | return nil |
| 693 | } |
| 694 | if err := hostfloor.UpdateIdle(cfg.UpdateGuardDir, cfg.UpdateJournal); err != nil { |
| 695 | return err |
| 696 | } |
| 697 | if cfg.verifyRunning != nil { |
| 698 | return cfg.verifyRunning(ctx) |
| 699 | } |
| 700 | pub, err := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY") |
| 701 | if err != nil { |
| 702 | return err |
| 703 | } |
| 704 | d := hostfloor.Disk{Dir: cfg.UpdateGuardDir, Bucket: os.Getenv("CARLOS_DEPLOYMENT_BUCKET"), PublicKey: pub} |
| 705 | running, err := d.Running(version.Build, runtime.GOOS+"-"+runtime.GOARCH) |
| 706 | if err != nil || !running.GuardReady { |
| 707 | if err != nil { |
| 708 | return err |
| 709 | } |
| 710 | return hostfloor.ErrFloor |
| 711 | } |
| 712 | _, floor, err := d.Load() |
| 713 | if err != nil || running.FloorSHA256 != floor.Digest { |
| 714 | return hostfloor.ErrFloor |
| 715 | } |
| 716 | return hostfloor.UpdateIdle(cfg.UpdateGuardDir, cfg.UpdateJournal) |
| 717 | } |
| 718 | |
| 719 | // authority calls this outside the controller mutex and host guard. Recheck |
| 720 | // the executing identity against the exact local digest compared remotely. |
| 721 | func (cfg Config) checkRemoteUpdateIdentity(ctx context.Context) error { |
| 722 | if cfg.UpdateGuardDir == "" || cfg.verifyRunning != nil { |
| 723 | return nil |
| 724 | } |
| 725 | pub, err := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY") |
| 726 | if err != nil { |
| 727 | return err |
| 728 | } |
| 729 | d := hostfloor.Disk{Dir: cfg.UpdateGuardDir, Bucket: os.Getenv("CARLOS_DEPLOYMENT_BUCKET"), PublicKey: pub} |
| 730 | running, err := d.Running(version.Build, runtime.GOOS+"-"+runtime.GOARCH) |
| 731 | if err != nil { |
| 732 | return err |
| 733 | } |
| 734 | _, floor, err := d.Load() |
| 735 | if err != nil || !running.GuardReady || running.FloorSHA256 != floor.Digest { |
| 736 | return hostfloor.ErrFloor |
| 737 | } |
| 738 | return hostfloor.CheckRemote(ctx, cfg.Store, pub, d.Bucket, floor) |
| 739 | } |
| 740 | |
amadan