amadan

carlos / platform Public

hostexec, hostguard, lease: the durable execution controller and independent writer checks (claim Plan 8 Task 3)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude aa8b7023dcf60c2159b9ffababa38f396f1ea876 parent b422192
13 files changed, +2006 −0
  • internal/hostexec/clock_linux.go +32 −0
  • internal/hostexec/clock_other.go +11 −0
  • internal/hostexec/clock_other_test.go +17 −0
  • internal/hostexec/controller.go +626 −0
  • internal/hostexec/controller_test.go +441 −0
  • internal/hostexec/local_other.go +17 −0
  • internal/hostexec/local_unix.go +197 −0
  • internal/hostexec/record.go +94 −0
  • internal/hostexec/review_test.go +460 −0
  • internal/hostguard/local.go +13 −0
  • internal/hostguard/local_test.go +45 −0
  • internal/lease/execution_store.go +20 −0
  • internal/lease/execution_store_test.go +33 −0
diff --git a/internal/hostexec/clock_linux.go b/internal/hostexec/clock_linux.go
new file mode 100644
index 000000000..edb44e56a
--- /dev/null
+++ b/internal/hostexec/clock_linux.go
@@ -0,0 +1,32 @@
+//go:build linux
+
+package hostexec
+
+import (
+ "golang.org/x/sys/unix"
+ "os"
+ "strings"
+ "time"
+)
+
+type SystemClock struct{}
+
+func (SystemClock) Now() (time.Time, int64, error) {
+ var ts unix.Timespec
+ if err := unix.ClockGettime(unix.CLOCK_BOOTTIME, &ts); err != nil {
+ return time.Time{}, 0, err
+ }
+ return time.Now(), ts.Nano(), nil
+}
+func (SystemClock) After(d time.Duration) <-chan time.Time { return time.After(d) }
+func BootID() (string, error) {
+ b, err := os.ReadFile("/proc/sys/kernel/random/boot_id")
+ if err != nil {
+ return "", err
+ }
+ id := strings.TrimSpace(string(b))
+ if !validID(id) {
+ return "", ErrState
+ }
+ return id, nil
+}
diff --git a/internal/hostexec/clock_other.go b/internal/hostexec/clock_other.go
new file mode 100644
index 000000000..7809dcee3
--- /dev/null
+++ b/internal/hostexec/clock_other.go
@@ -0,0 +1,11 @@
+//go:build !linux
+
+package hostexec
+
+import "time"
+
+type SystemClock struct{}
+
+func (SystemClock) Now() (time.Time, int64, error) { return time.Time{}, 0, ErrUnsupported }
+func (SystemClock) After(d time.Duration) <-chan time.Time { return time.After(d) }
+func BootID() (string, error) { return "", ErrUnsupported }
diff --git a/internal/hostexec/clock_other_test.go b/internal/hostexec/clock_other_test.go
new file mode 100644
index 000000000..381ffd8d4
--- /dev/null
+++ b/internal/hostexec/clock_other_test.go
@@ -0,0 +1,17 @@
+//go:build !linux
+
+package hostexec
+
+import (
+ "errors"
+ "testing"
+)
+
+func TestSystemClockRefusesUnsupportedPlatform(t *testing.T) {
+ if _, _, err := (SystemClock{}).Now(); !errors.Is(err, ErrUnsupported) {
+ t.Fatal(err)
+ }
+ if _, err := BootID(); !errors.Is(err, ErrUnsupported) {
+ t.Fatal(err)
+ }
+}
diff --git a/internal/hostexec/controller.go b/internal/hostexec/controller.go
new file mode 100644
index 000000000..6f1bf820d
--- /dev/null
+++ b/internal/hostexec/controller.go
@@ -0,0 +1,626 @@
+package hostexec
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "os"
+ "slices"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/hostquarantine"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type Config struct {
+ Binding Binding
+ Store release.Store
+ Guard hostguard.Guard
+ Directory hostquarantine.ActorDirectory
+ Clock Clock
+ LockDir string
+ // These close/cancel hooks must be nonblocking and safe to call repeatedly.
+ CloseAdmission func()
+ CancelWork func()
+ // Cleanup terminates only this host's processes and closes its replicators.
+ // It must not wait for the host guard: uncertain disk writes retain that
+ // guard to deny other processes, while termination must proceed independently.
+ // Its failure remains a separate denial owned by the caller, even after CC9.
+ Cleanup func(context.Context) error
+ writeLocal func(*localDir, State) error // test seam; nil uses durable descriptor-relative I/O
+}
+
+type Controller struct {
+ cfg Config
+ mu sync.Mutex
+ state State
+ err error
+ cleanupErr error
+ releaseETag string
+ cleanupOnce sync.Once
+ cleanupDone chan struct{}
+ ready chan struct{}
+ readyOnce sync.Once
+ lost chan struct{}
+ ctx context.Context
+ cancel context.CancelFunc
+ done chan struct{}
+ watchDone chan struct{}
+ unlock func()
+ unlive func()
+ closeMu sync.Mutex
+ closed bool
+}
+
+func Open(ctx context.Context, cfg Config) (*Controller, error) {
+ if cfg.Clock == nil {
+ cfg.Clock = SystemClock{}
+ }
+ if cfg.LockDir == "" {
+ cfg.LockDir = "/run/carlos/hostexec"
+ }
+ var nonce [16]byte
+ if _, err := rand.Read(nonce[:]); err != nil {
+ return nil, err
+ }
+ cfg.Binding.ControllerID = hex.EncodeToString(nonce[:])
+ if !cfg.Binding.valid() || cfg.Store == nil || cfg.Guard.Claims == nil || cfg.Directory == nil ||
+ cfg.CloseAdmission == nil || cfg.CancelWork == nil || cfg.Cleanup == nil {
+ return nil, ErrState
+ }
+ wall, boot, err := cfg.Clock.Now()
+ if err != nil {
+ return nil, err
+ }
+ cctx, cancel := context.WithCancel(ctx)
+ c := &Controller{cfg: cfg, state: State{V: 1, Binding: cfg.Binding, State: "pending", Expires: wall.Add(TTL), SafeUntil: boot + int64(TTL-Margin)}, ctx: cctx, cancel: cancel,
+ ready: make(chan struct{}), lost: make(chan struct{}), done: make(chan struct{}), watchDone: make(chan struct{}), cleanupDone: make(chan struct{})}
+ // Hold the host guard across lifetime-lock acquisition and the pending
+ // write. Acquire the live lock only after pending state is durable, so
+ // orphaned active bytes cannot regain admission. Never wait for the
+ // lifetime lock while holding the guard.
+ var cleaned <-chan error
+ for {
+ err = cfg.Guard.WithLocal(ctx, cfg.Binding.Host, func() error {
+ var lockErr error
+ c.unlock, lockErr = tryAcquire(cfg.LockDir, cfg.Binding.Host)
+ if lockErr != nil {
+ return lockErr
+ }
+ // Start cleanup before any persistence can block. Exclusion stays
+ // held until this independent cleanup has joined on every exit.
+ result := make(chan error, 1)
+ cleaned = result
+ go func() { result <- cfg.Cleanup(context.Background()) }()
+ cfg.CloseAdmission()
+ cfg.CancelWork()
+ d, openErr := openLocal(cfg.Guard.DataDir)
+ if openErr != nil {
+ return openErr
+ }
+ defer d.close()
+ if writeErr := c.write(d, c.state); writeErr != nil {
+ return writeErr
+ }
+ c.unlive, lockErr = tryAcquireLive(cfg.LockDir, cfg.Binding.Host)
+ return lockErr
+ })
+ if c.unlock != nil || !errors.Is(err, ErrControllerBusy) {
+ break
+ }
+ select {
+ case <-ctx.Done():
+ err = ctx.Err()
+ case <-time.After(10 * time.Millisecond):
+ continue
+ }
+ break
+ }
+ if err == nil {
+ err = <-cleaned
+ cleaned = nil
+ }
+ if err == nil {
+ err = c.authority(ctx)
+ }
+ if err != nil {
+ c.state.State = "lost"
+ hadLive := c.unlive != nil
+ if c.unlive != nil {
+ c.unlive()
+ c.unlive = nil
+ }
+ if c.unlock != nil {
+ // Before live-lock acquisition, no local state can authorize work.
+ // A failed pending write (including an unsafe path) must not make
+ // cleanup wait for another write to the same failed destination.
+ if hadLive {
+ err = errors.Join(err, c.save(c.state))
+ }
+ if cleaned != nil {
+ err = errors.Join(err, <-cleaned)
+ }
+ c.unlock()
+ }
+ cancel()
+ return nil, err
+ }
+ go c.watch()
+ go c.run()
+ return c, nil
+}
+
+func (c *Controller) write(d *localDir, s State) error {
+ if c.cfg.writeLocal != nil {
+ return c.cfg.writeLocal(d, s)
+ }
+ return d.write(s)
+}
+
+// save retains the guard until lost state is durable, including failures before
+// rename. The revoked live lock denies CheckLocal while the old file survives.
+func (c *Controller) save(s State) error {
+ return c.cfg.Guard.WithLocal(context.Background(), s.Host, func() error {
+ for {
+ d, err := openLocal(c.cfg.Guard.DataDir)
+ if err == nil {
+ err = c.write(d, s)
+ d.close()
+ }
+ if err == nil || s.State != "lost" {
+ return err
+ }
+ time.Sleep(100 * time.Millisecond)
+ }
+ })
+}
+func (c *Controller) authority(ctx context.Context) error {
+ b := c.cfg.Binding
+ if err := hostquarantine.Check(ctx, c.cfg.Directory, b.Box, b.Enrollment); err != nil {
+ return err
+ }
+ obs, err := c.cfg.Guard.Claims.Read(ctx, b.Host)
+ if err != nil {
+ return err
+ }
+ if obs.Absent || obs.Claim == nil || obs.Host != b.Host || obs.Claim.Host != b.Host {
+ return ErrInactive
+ }
+ q := obs.Claim
+ if q.Validate() != nil || q.Holder == nil || q.Holder.Incarnation != b.Incarnation || q.Op.ID != b.Op ||
+ q.Op.Kind == hostclaim.KindServing || q.Free() || !slices.Contains(q.Op.Participants, b.Box) || slices.Contains(q.Lost, b.Box) {
+ return ErrInactive
+ }
+ return nil
+}
+
+// loseLocked latches denial before attempting disk I/O or process cleanup.
+func (c *Controller) loseLocked(err error) {
+ if c.err != nil {
+ return
+ }
+ c.err = err
+ c.state.State = "lost"
+ // Revoke cross-process admission before publishing loss or starting I/O.
+ // This close needs no disk space. The lifetime lock remains held.
+ if c.unlive != nil {
+ c.unlive()
+ c.unlive = nil
+ }
+ close(c.lost)
+ c.cancel()
+ c.cfg.CloseAdmission()
+ c.cfg.CancelWork()
+ c.readyOnce.Do(func() { close(c.ready) })
+ s := c.state
+ c.cleanupOnce.Do(func() {
+ go func() {
+ // Neither a blocked disk nor blocked cleanup can prevent the other.
+ cleaned := make(chan error, 1)
+ go func() { cleaned <- c.cfg.Cleanup(context.Background()) }()
+ persistErr := c.save(s)
+ cleanupErr := <-cleaned
+ c.mu.Lock()
+ c.cleanupErr = errors.Join(persistErr, cleanupErr)
+ c.mu.Unlock()
+ close(c.cleanupDone)
+ }()
+ })
+}
+func (c *Controller) checkLocked() error {
+ if c.err != nil {
+ return c.err
+ }
+ _, boot, err := c.cfg.Clock.Now()
+ if err != nil {
+ c.loseLocked(err)
+ } else if boot >= c.state.SafeUntil {
+ c.loseLocked(ErrLost)
+ }
+ return c.err
+}
+func (c *Controller) Lost() <-chan struct{} { return c.lost }
+func (c *Controller) Ready(ctx context.Context) (State, error) {
+ select {
+ case <-ctx.Done():
+ return State{}, ctx.Err()
+ case <-c.ready:
+ }
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ if err := c.checkLocked(); err != nil {
+ return State{}, err
+ }
+ if c.state.State != "active" {
+ return State{}, ErrState
+ }
+ return c.state, nil
+}
+func (c *Controller) CleanupError() error { c.mu.Lock(); defer c.mu.Unlock(); return c.cleanupErr }
+
+func (c *Controller) watch() {
+ defer close(c.watchDone)
+ for {
+ c.mu.Lock()
+ err := c.checkLocked()
+ c.mu.Unlock()
+ if err != nil {
+ return
+ }
+ // CLOCK_BOOTTIME is sampled on every wake and every admission. A normal Go
+ // timer does not itself prove that a suspended machine has remained stopped.
+ select {
+ case <-c.ctx.Done():
+ c.mu.Lock()
+ c.loseLocked(c.ctx.Err())
+ c.mu.Unlock()
+ return
+ case <-c.cfg.Clock.After(100 * time.Millisecond):
+ }
+ }
+}
+func (c *Controller) run() {
+ defer close(c.done)
+ key, _ := Key(c.cfg.Binding.Host, c.cfg.Binding.Box)
+ body, etag, err := c.cfg.Store.Get(c.ctx, key)
+ if errors.Is(err, os.ErrNotExist) {
+ err = nil
+ etag = ""
+ } else if err == nil {
+ var r Record
+ if decode(body, &r) != nil || r.Op == "" || r.Expires.IsZero() || etag == "" {
+ err = ErrState
+ }
+ }
+ if err != nil {
+ c.mu.Lock()
+ c.loseLocked(err)
+ c.mu.Unlock()
+ return
+ }
+ for {
+ if err = c.authority(c.ctx); err != nil {
+ c.mu.Lock()
+ c.loseLocked(err)
+ c.mu.Unlock()
+ return
+ }
+ wall, start, clockErr := c.cfg.Clock.Now()
+ if clockErr != nil {
+ c.mu.Lock()
+ c.loseLocked(clockErr)
+ c.mu.Unlock()
+ return
+ }
+ c.mu.Lock()
+ err = c.checkLocked()
+ captured := c.state.Binding
+ c.mu.Unlock()
+ if err != nil {
+ return
+ }
+ proposed := Record{Op: captured.Op, Expires: wall.Add(TTL)}
+ bytes, _ := json.Marshal(proposed)
+ next, putErr := c.cfg.Store.Put(c.ctx, key, bytes, etag)
+ c.mu.Lock()
+ // A late acknowledgement can only contribute its exact ETag to release.
+ if captured != c.state.Binding {
+ c.mu.Unlock()
+ return
+ }
+ if putErr == nil && next != "" {
+ c.releaseETag = next
+ etag = next
+ }
+ if err = c.checkLocked(); err == nil && (putErr != nil || next == "") {
+ if putErr == nil {
+ putErr = ErrState
+ }
+ c.loseLocked(putErr)
+ err = putErr
+ }
+ c.mu.Unlock()
+ if err != nil {
+ return
+ }
+ // Network reads never hold either the controller mutex or the host guard.
+ if err = c.authority(c.ctx); err != nil {
+ c.mu.Lock()
+ c.loseLocked(err)
+ c.mu.Unlock()
+ return
+ }
+ c.mu.Lock()
+ if err = c.checkLocked(); err != nil {
+ c.mu.Unlock()
+ return
+ }
+ nextState := c.state
+ nextState.State = "active"
+ nextState.LeaseETag = next
+ nextState.Expires = proposed.Expires
+ nextState.SafeUntil = start + int64(TTL-Margin)
+ // Never use a response-relative deadline, including when the wall clock moved.
+ _, now, clockErr := c.cfg.Clock.Now()
+ if clockErr != nil || now >= nextState.SafeUntil {
+ c.loseLocked(ErrLost)
+ c.mu.Unlock()
+ return
+ }
+ c.mu.Unlock()
+ // Publication and any intervening loss are serialized under the same host
+ // guard, so another process never observes a stale active overwrite.
+ if err = c.publish(nextState); err != nil {
+ c.mu.Lock()
+ c.loseLocked(err)
+ c.mu.Unlock()
+ return
+ }
+ _, now, clockErr = c.cfg.Clock.Now()
+ if clockErr != nil {
+ c.mu.Lock()
+ c.loseLocked(clockErr)
+ c.mu.Unlock()
+ return
+ }
+ delay := time.Duration(start + int64(Renewal) - now)
+ if delay < 0 {
+ delay = 0
+ }
+ select {
+ case <-c.ctx.Done():
+ return
+ case <-c.cfg.Clock.After(delay):
+ }
+ }
+}
+
+// publish leaves mu available to the watchdog during filesystem calls. If a
+// write has an uncertain outcome, it retains the host guard until denial is
+// durable; releasing it over possibly active bytes would admit another process.
+func (c *Controller) publish(next State) error {
+ return c.cfg.Guard.WithLocal(context.Background(), next.Host, func() error {
+ d, err := openLocal(c.cfg.Guard.DataDir)
+ if err != nil {
+ return err
+ }
+ defer d.close()
+ c.mu.Lock()
+ err = c.checkLocked()
+ if err == nil && next.Binding != c.state.Binding {
+ c.loseLocked(ErrState)
+ err = ErrState
+ }
+ c.mu.Unlock()
+ if err != nil {
+ return err
+ }
+ writeErr := c.write(d, next)
+ c.mu.Lock()
+ if writeErr != nil {
+ c.loseLocked(writeErr)
+ }
+ err = c.checkLocked()
+ if err == nil {
+ _, boot, clockErr := c.cfg.Clock.Now()
+ if clockErr != nil || boot >= next.SafeUntil {
+ c.loseLocked(ErrLost)
+ err = c.err
+ }
+ }
+ if err == nil {
+ c.state = next
+ c.readyOnce.Do(func() { close(c.ready) })
+ c.mu.Unlock()
+ return nil
+ }
+ denied := c.state
+ denied.State = "lost"
+ c.mu.Unlock()
+ // Cancellation does not release the guard over uncertain active bytes.
+ // Close may time out; it keeps the lifetime lock until this join completes.
+ for {
+ if persistErr := c.write(d, denied); persistErr == nil {
+ return err
+ }
+ time.Sleep(100 * time.Millisecond)
+ }
+ })
+}
+
+// Close cancels and joins all lease requests before conditional release. It
+// retains the lifetime lock on a timed-out join; call Close again to finish.
+func (c *Controller) Close(ctx context.Context) error {
+ c.closeMu.Lock()
+ defer c.closeMu.Unlock()
+ if c.closed {
+ return nil
+ }
+ c.mu.Lock()
+ c.loseLocked(ErrLost)
+ c.mu.Unlock()
+ for _, done := range []<-chan struct{}{c.done, c.watchDone, c.cleanupDone} {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-done:
+ }
+ }
+ c.mu.Lock()
+ etag := c.releaseETag
+ c.mu.Unlock()
+ var err error
+ if etag != "" {
+ key, _ := Key(c.cfg.Binding.Host, c.cfg.Binding.Box)
+ err = release.DeleteIfMatch(ctx, c.cfg.Store, key, etag)
+ }
+ c.closed = true
+ c.unlock()
+ return errors.Join(err, c.CleanupError())
+}
+
+// CheckLocal must run inside Guard.With, before each bounded host write. It
+// does not grant authority and never clears quarantine or cleanup denial.
+func CheckLocal(dataDir string, expected Binding, clock Clock, lockDir string) error {
+ if !expected.valid() || clock == nil {
+ return ErrState
+ }
+ if lockDir == "" {
+ lockDir = "/run/carlos/hostexec"
+ }
+ // Guard.With serializes these probes with other writers and Open. A free
+ // live lock denies even if ENOSPC left a future-dated active file behind.
+ for _, probe := range []func(string, string) (func(), error){tryAcquire, tryAcquireLive} {
+ unlock, err := probe(lockDir, expected.Host)
+ if err == nil {
+ unlock()
+ return ErrLost
+ }
+ if !errors.Is(err, ErrControllerBusy) {
+ return err
+ }
+ }
+ d, err := openLocal(dataDir)
+ if err != nil {
+ return err
+ }
+ defer d.close()
+ s, err := d.read(expected.Host)
+ if err != nil {
+ return err
+ }
+ if s.Binding != expected || s.State != "active" {
+ return ErrLost
+ }
+ _, boot, err := clock.Now()
+ if err != nil {
+ return err
+ }
+ if boot >= s.SafeUntil {
+ return ErrLost
+ }
+ return nil
+}
+
+// ClearStale performs CC9 with the lifetime lock held, after any old controller
+// has joined. A failed or mismatched claim read leaves local denial intact.
+func ClearStale(ctx context.Context, cfg Config) error {
+ lockDir := cfg.LockDir
+ if lockDir == "" {
+ lockDir = "/run/carlos/hostexec"
+ }
+ if cfg.Guard.Claims == nil || !validID(cfg.Binding.Box) || !validID(cfg.Binding.Enrollment) {
+ return ErrState
+ }
+ var unlock func()
+ var err error
+ for {
+ err = cfg.Guard.WithLocal(ctx, cfg.Binding.Host, func() error {
+ var lockErr error
+ unlock, lockErr = tryAcquire(lockDir, cfg.Binding.Host)
+ if lockErr != nil {
+ return lockErr
+ }
+ denied := false
+ defer func() {
+ if !denied {
+ unlock()
+ unlock = nil
+ }
+ }()
+ d, openErr := openLocal(cfg.Guard.DataDir)
+ if openErr != nil {
+ return openErr
+ }
+ defer d.close()
+ old, readErr := d.read(cfg.Binding.Host)
+ if errors.Is(readErr, os.ErrNotExist) {
+ denied = true
+ return nil
+ }
+ if readErr != nil {
+ return readErr
+ }
+ if old.Box != cfg.Binding.Box || old.Enrollment != cfg.Binding.Enrollment ||
+ old.Op != cfg.Binding.Op || old.Incarnation != cfg.Binding.Incarnation {
+ return ErrState
+ }
+ // Holding the lifetime lock must never make an orphaned active record
+ // appear to belong to a running controller during the claim read.
+ old.State = "lost"
+ writeErr := d.write(old)
+ denied = writeErr == nil
+ return writeErr
+ })
+ if !errors.Is(err, ErrControllerBusy) {
+ break
+ }
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-time.After(10 * time.Millisecond):
+ }
+ }
+ if err != nil {
+ return err
+ }
+ defer unlock()
+ obs, err := cfg.Guard.Claims.Read(ctx, cfg.Binding.Host)
+ if err != nil {
+ return err
+ }
+ if obs.Absent || obs.Claim == nil || obs.Host != cfg.Binding.Host || obs.Claim.Host != cfg.Binding.Host || obs.Claim.Validate() != nil {
+ return ErrState
+ }
+ q := obs.Claim
+ return cfg.Guard.WithLocal(ctx, cfg.Binding.Host, func() error {
+ d, err := openLocal(cfg.Guard.DataDir)
+ if err != nil {
+ return err
+ }
+ defer d.close()
+ old, err := d.read(cfg.Binding.Host)
+ if errors.Is(err, os.ErrNotExist) {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+ if old.Box != cfg.Binding.Box || old.Enrollment != cfg.Binding.Enrollment ||
+ old.Op != cfg.Binding.Op || old.Incarnation != cfg.Binding.Incarnation {
+ return ErrState
+ }
+ // Only the caller's expected operation is cleared. A fresh successor
+ // operation may exclude this box; that is also CC9 discharge evidence.
+ if !q.Free() && q.Op.Kind != hostclaim.KindServing &&
+ slices.Contains(q.Op.Participants, old.Box) {
+ return ErrInactive
+ }
+ return d.remove(old.Host)
+ })
+}
diff --git a/internal/hostexec/controller_test.go b/internal/hostexec/controller_test.go
new file mode 100644
index 000000000..c4cac8b18
--- /dev/null
+++ b/internal/hostexec/controller_test.go
@@ -0,0 +1,441 @@
+//go:build linux
+
+package hostexec
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/hostquarantine"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type alarm struct {
+ at int64
+ ch chan time.Time
+}
+type testClock struct {
+ mu sync.Mutex
+ wall time.Time
+ boot int64
+ alarms []alarm
+}
+
+func (k *testClock) Now() (time.Time, int64, error) {
+ k.mu.Lock()
+ defer k.mu.Unlock()
+ return k.wall, k.boot, nil
+}
+func (k *testClock) After(d time.Duration) <-chan time.Time {
+ k.mu.Lock()
+ defer k.mu.Unlock()
+ ch := make(chan time.Time, 1)
+ if d <= 0 {
+ ch <- k.wall
+ } else {
+ k.alarms = append(k.alarms, alarm{k.boot + int64(d), ch})
+ }
+ return ch
+}
+func (k *testClock) advance(d time.Duration) {
+ k.mu.Lock()
+ defer k.mu.Unlock()
+ k.boot += int64(d)
+ k.wall = k.wall.Add(d)
+ rest := k.alarms[:0]
+ for _, a := range k.alarms {
+ if a.at <= k.boot {
+ a.ch <- k.wall
+ } else {
+ rest = append(rest, a)
+ }
+ }
+ k.alarms = rest
+}
+
+type claims struct {
+ mu sync.Mutex
+ q hostclaim.Claim
+ err error
+}
+
+func (r *claims) Read(context.Context, string) (hostclaim.Observed, error) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ raw, _ := json.Marshal(r.q)
+ var q hostclaim.Claim
+ _ = json.Unmarshal(raw, &q)
+ return hostclaim.Observed{Host: q.Host, Claim: &q, ETag: "claim-etag"}, r.err
+}
+
+type directory struct {
+ state string
+ err error
+}
+
+func (d directory) Lookup(context.Context, string) (hostquarantine.Actor, error) {
+ return hostquarantine.Actor{Enrollment: "enroll", State: d.state}, d.err
+}
+
+type delayedStore struct {
+ release.Store
+ mu sync.Mutex
+ calls int
+ onPut func(int)
+ failAt int
+ called chan int
+}
+
+func (s *delayedStore) Put(ctx context.Context, key string, b []byte, etag string) (string, error) {
+ s.mu.Lock()
+ s.calls++
+ n := s.calls
+ s.mu.Unlock()
+ if s.called != nil {
+ s.called <- n
+ }
+ if s.onPut != nil {
+ s.onPut(n)
+ }
+ if n == s.failAt {
+ return "", errors.New("ambiguous network failure")
+ }
+ // Deliberately permit a late response even after cancellation.
+ return s.Store.Put(context.Background(), key, b, etag)
+}
+func (s *delayedStore) DeleteIfMatch(ctx context.Context, key, etag string) error {
+ return release.DeleteIfMatch(ctx, s.Store, key, etag)
+}
+
+func fixture(t *testing.T) (Config, *claims, *testClock, *delayedStore, *atomic.Int32) {
+ t.Helper()
+ dir := t.TempDir()
+ for _, name := range []string{"hostguard", ".hostexec", "locks", "exec-locks"} {
+ if err := os.Mkdir(filepath.Join(dir, name), 0770); err != nil {
+ t.Fatal(err)
+ }
+ }
+ host := "shop.example.com"
+ inc, _ := hostclaim.Incarnation("bab", "jam", host, time.Time{})
+ 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"}}}
+ if err := q.Validate(); err != nil {
+ t.Fatal(err)
+ }
+ r := &claims{q: q}
+ clock := &testClock{wall: time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC), boot: int64(time.Hour)}
+ store := &delayedStore{Store: release.NewDirStore(t.TempDir()), called: make(chan int, 20)}
+ cleaned := &atomic.Int32{}
+ 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 }}
+ return cfg, r, clock, store, cleaned
+}
+func ready(t *testing.T, c *Controller) State {
+ t.Helper()
+ ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
+ defer cancel()
+ s, err := c.Ready(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return s
+}
+func closeController(t *testing.T, c *Controller) {
+ t.Helper()
+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+ defer cancel()
+ if err := c.Close(ctx); err != nil {
+ t.Error(err)
+ }
+}
+func until(t *testing.T, fn func() bool) {
+ t.Helper()
+ deadline := time.Now().Add(2 * time.Second)
+ for !fn() {
+ if time.Now().After(deadline) {
+ t.Fatal("condition did not become true")
+ }
+ time.Sleep(time.Millisecond)
+ }
+}
+
+func TestRenewalRequestStartAndIndependentExpiry(t *testing.T) {
+ cfg, _, clock, store, cleaned := fixture(t)
+ blocked := make(chan struct{})
+ store.onPut = func(n int) {
+ if n == 2 {
+ <-blocked
+ }
+ }
+ c, err := Open(t.Context(), cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer closeController(t, c)
+ first := ready(t, c)
+ key, _ := Key(first.Host, first.Box)
+ b, _, err := store.Get(t.Context(), key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var wire map[string]any
+ _ = json.Unmarshal(b, &wire)
+ if len(wire) != 2 || wire["op"] != "op" {
+ t.Fatalf("wire %s", b)
+ }
+ until(t, func() bool { clock.mu.Lock(); defer clock.mu.Unlock(); return len(clock.alarms) >= 2 })
+ clock.advance(Renewal)
+ for {
+ select {
+ case n := <-store.called:
+ if n == 2 {
+ goto renewing
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("no renewal")
+ }
+ }
+renewing:
+ clock.advance(65 * time.Second)
+ select {
+ case <-c.Lost():
+ case <-time.After(2 * time.Second):
+ t.Fatal("blocked PUT blocked watchdog")
+ }
+ if err := CheckLocal(cfg.Guard.DataDir, first.Binding, clock, cfg.LockDir); err == nil {
+ t.Fatal("writer admitted at deadline")
+ }
+ close(blocked)
+ <-c.done
+ if _, err := c.Ready(t.Context()); err == nil {
+ t.Fatal("late success revived controller")
+ }
+ until(t, func() bool { return cleaned.Load() >= 2 })
+}
+
+func TestRenewalUsesStartAndWallRollbackCannotExtend(t *testing.T) {
+ cfg, _, clock, store, _ := fixture(t)
+ store.onPut = func(n int) {
+ if n == 1 {
+ clock.advance(9 * time.Second)
+ }
+ }
+ c, err := Open(t.Context(), cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer closeController(t, c)
+ s := ready(t, c)
+ if s.SafeUntil != int64(time.Hour+85*time.Second) {
+ t.Fatalf("response-relative deadline %d", s.SafeUntil)
+ }
+ clock.mu.Lock()
+ clock.wall = clock.wall.Add(-time.Hour)
+ clock.mu.Unlock()
+ clock.advance(76 * time.Second)
+ if err := CheckLocal(cfg.Guard.DataDir, s.Binding, clock, cfg.LockDir); err == nil {
+ t.Fatal("wall rollback extended deadline")
+ }
+}
+
+func TestAmbiguousPUTLatchesAndConditionalReleasePreservesSuccessor(t *testing.T) {
+ cfg, _, _, store, _ := fixture(t)
+ store.failAt = 1
+ c, err := Open(t.Context(), cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err = c.Ready(t.Context()); err == nil {
+ t.Fatal("ambiguous PUT admitted")
+ }
+ closeController(t, c)
+ cfg, _, _, store, _ = fixture(t)
+ c, err = Open(t.Context(), cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ s := ready(t, c)
+ key, _ := Key(s.Host, s.Box)
+ successor := []byte(`{"op":"successor","expires":"2030-01-01T00:00:00Z"}`)
+ if _, err = store.Store.Put(t.Context(), key, successor, s.LeaseETag); err != nil {
+ t.Fatal(err)
+ }
+ if err = c.Close(t.Context()); !errors.Is(err, release.ErrCASConflict) {
+ t.Fatalf("release = %v", err)
+ }
+ got, _, err := store.Get(t.Context(), key)
+ if err != nil || string(got) != string(successor) {
+ t.Fatal("deleted successor")
+ }
+}
+
+func TestControllerExclusionBootAndCC9(t *testing.T) {
+ cfg, r, clock, _, _ := fixture(t)
+ c, err := Open(t.Context(), cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ s := ready(t, c)
+ ctx, cancel := context.WithTimeout(t.Context(), 30*time.Millisecond)
+ defer cancel()
+ if _, err = Open(ctx, cfg); !errors.Is(err, context.DeadlineExceeded) {
+ t.Fatalf("second controller = %v", err)
+ }
+ wrong := s.Binding
+ wrong.BootID = "next-boot"
+ if err = CheckLocal(cfg.Guard.DataDir, wrong, clock, cfg.LockDir); err == nil {
+ t.Fatal("inherited prior boot")
+ }
+ closeController(t, c)
+ if err = ClearStale(t.Context(), cfg); !errors.Is(err, ErrInactive) {
+ t.Fatalf("cleared active op: %v", err)
+ }
+ r.mu.Lock()
+ r.q.Op = hostclaim.Op{Kind: hostclaim.KindServing, Participants: []string{"box"}}
+ r.mu.Unlock()
+ if err = ClearStale(t.Context(), cfg); err != nil {
+ t.Fatal(err)
+ }
+ if _, err = os.Stat(filepath.Join(cfg.Guard.DataDir, ".hostexec", cfg.Binding.Host+".json")); !errors.Is(err, os.ErrNotExist) {
+ t.Fatal("stale state remains", err)
+ }
+ // No lease traffic for retained serving participants, even when enabled.
+ if _, err = Open(t.Context(), cfg); !errors.Is(err, ErrInactive) {
+ t.Fatalf("serving controller = %v", err)
+ }
+}
+
+func TestMalformedPendingMissingAndQuarantineRefuse(t *testing.T) {
+ cfg, _, clock, _, _ := fixture(t)
+ b := cfg.Binding
+ b.ControllerID = "controller"
+ if err := CheckLocal(cfg.Guard.DataDir, b, clock, cfg.LockDir); err == nil {
+ t.Fatal("missing state admitted")
+ }
+ p := filepath.Join(cfg.Guard.DataDir, ".hostexec", b.Host+".json")
+ if err := os.WriteFile(p, []byte(`{"v":1}`), 0660); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := Open(t.Context(), cfg); err == nil {
+ t.Fatal("malformed state replaced")
+ }
+ if err := os.Remove(p); err != nil {
+ t.Fatal(err)
+ }
+ cfg.Directory = directory{state: "quarantined"}
+ if _, err := Open(t.Context(), cfg); err == nil {
+ t.Fatal("quarantined box admitted")
+ }
+}
+
+func TestLocalPathsPinnedAndSymlinkRefused(t *testing.T) {
+ cfg, _, _, _, _ := fixture(t)
+ outside := filepath.Join(t.TempDir(), "outside")
+ if err := os.WriteFile(outside, []byte("untouched"), 0600); err != nil {
+ t.Fatal(err)
+ }
+ record := filepath.Join(cfg.Guard.DataDir, ".hostexec", cfg.Binding.Host+".json")
+ if err := os.Symlink(outside, record); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := Open(t.Context(), cfg); err == nil {
+ t.Fatal("symlink admitted")
+ }
+ b, _ := os.ReadFile(outside)
+ st, _ := os.Stat(outside)
+ if string(b) != "untouched" || st.Mode().Perm() != 0600 {
+ t.Fatal("changed symlink target")
+ }
+ if err := os.Remove(record); err != nil {
+ t.Fatal(err)
+ }
+ d, err := openLocal(cfg.Guard.DataDir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer d.close()
+ old := filepath.Join(cfg.Guard.DataDir, ".hostexec")
+ if err = os.Rename(old, old+".pinned"); err != nil {
+ t.Fatal(err)
+ }
+ if err = os.Mkdir(old, 0770); err != nil {
+ t.Fatal(err)
+ }
+ s := State{V: 1, Binding: cfg.Binding, State: "pending", Expires: time.Now().Add(TTL), SafeUntil: 99}
+ s.ControllerID = "controller"
+ if err = d.write(s); err != nil {
+ t.Fatal(err)
+ }
+ if _, err = os.Stat(filepath.Join(old+".pinned", s.Host+".json")); err != nil {
+ t.Fatal(err)
+ }
+ if _, err = os.Stat(filepath.Join(old, s.Host+".json")); !errors.Is(err, os.ErrNotExist) {
+ t.Fatal("followed directory replacement")
+ }
+ d.sync = func(int) error { return errors.New("fsync failure") }
+ if err = d.write(s); err == nil {
+ t.Fatal("ignored persistence failure")
+ }
+ entries, _ := os.ReadDir(old + ".pinned")
+ for _, e := range entries {
+ if strings.HasPrefix(e.Name(), ".tmp-") {
+ t.Fatal("leaked temp file")
+ }
+ }
+}
+
+func TestSuccessfulRenewalDurablyExtendsFromItsRequestStart(t *testing.T) {
+ cfg, _, clock, store, _ := fixture(t)
+ c, err := Open(t.Context(), cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer closeController(t, c)
+ first := ready(t, c)
+ until(t, func() bool {
+ clock.mu.Lock()
+ defer clock.mu.Unlock()
+ for _, a := range clock.alarms {
+ if a.at == clock.boot+int64(Renewal) {
+ return true
+ }
+ }
+ return false
+ })
+ clock.advance(Renewal)
+ until(t, func() bool {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return c.state.SafeUntil == first.SafeUntil+int64(Renewal)
+ })
+ renewed := ready(t, c)
+ if renewed.LeaseETag == first.LeaseETag || !renewed.Expires.Equal(first.Expires.Add(Renewal)) {
+ t.Fatalf("renewal did not advance coherent tuple: %+v", renewed)
+ }
+ d, err := openLocal(cfg.Guard.DataDir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer d.close()
+ disk, err := d.read(renewed.Host)
+ if err != nil || disk != renewed {
+ t.Fatalf("renewal not durable: %+v %v", disk, err)
+ }
+ key, _ := Key(renewed.Host, renewed.Box)
+ raw, etag, err := store.Get(t.Context(), key)
+ if err != nil || etag != renewed.LeaseETag {
+ t.Fatalf("renewed remote ownership: %q %v", etag, err)
+ }
+ var wire Record
+ if err = json.Unmarshal(raw, &wire); err != nil || wire.Op != renewed.Op || !wire.Expires.Equal(renewed.Expires) {
+ t.Fatalf("renewed wire: %+v %v", wire, err)
+ }
+}
diff --git a/internal/hostexec/local_other.go b/internal/hostexec/local_other.go
new file mode 100644
index 000000000..1d2f2734b
--- /dev/null
+++ b/internal/hostexec/local_other.go
@@ -0,0 +1,17 @@
+//go:build !unix
+
+package hostexec
+
+import "context"
+
+type localDir struct{}
+
+func openLocal(string) (*localDir, error) { return nil, ErrUnsupported }
+func (*localDir) close() {}
+func (*localDir) read(string) (State, error) { return State{}, ErrUnsupported }
+func (*localDir) write(State) error { return ErrUnsupported }
+func (*localDir) remove(string) error { return ErrUnsupported }
+func acquire(context.Context, string, string) (func(), error) { return nil, ErrUnsupported }
+
+func tryAcquire(string, string) (func(), error) { return nil, ErrUnsupported }
+func tryAcquireLive(string, string) (func(), error) { return nil, ErrUnsupported }
diff --git a/internal/hostexec/local_unix.go b/internal/hostexec/local_unix.go
new file mode 100644
index 000000000..932d86330
--- /dev/null
+++ b/internal/hostexec/local_unix.go
@@ -0,0 +1,197 @@
+//go:build unix
+
+package hostexec
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "io"
+ "os"
+ "path/filepath"
+ "time"
+
+ "golang.org/x/sys/unix"
+)
+
+type localDir struct {
+ fd int
+ sync func(int) error
+}
+
+func openDir(path string) (int, error) {
+ fd, err := unix.Open(path, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_NOFOLLOW|unix.O_CLOEXEC, 0)
+ if err != nil {
+ return -1, err
+ }
+ var st unix.Stat_t
+ if err = unix.Fstat(fd, &st); err != nil {
+ unix.Close(fd)
+ return -1, err
+ }
+ if st.Mode&unix.S_IFMT != unix.S_IFDIR {
+ unix.Close(fd)
+ return -1, ErrState
+ }
+ return fd, nil
+}
+func openLocal(dataDir string) (*localDir, error) {
+ fd, err := openDir(filepath.Join(dataDir, ".hostexec"))
+ if err != nil {
+ return nil, err
+ }
+ return &localDir{fd: fd, sync: unix.Fsync}, nil
+}
+func (d *localDir) close() { _ = unix.Close(d.fd) }
+func (d *localDir) read(host string) (State, error) {
+ var s State
+ if _, err := Key(host, "check"); err != nil {
+ return s, err
+ }
+ fd, err := unix.Openat(d.fd, host+".json", unix.O_RDONLY|unix.O_NOFOLLOW|unix.O_CLOEXEC|unix.O_NONBLOCK, 0)
+ if err != nil {
+ return s, err
+ }
+ f := os.NewFile(uintptr(fd), host)
+ defer f.Close()
+ var st unix.Stat_t
+ if err = unix.Fstat(fd, &st); err != nil {
+ return s, err
+ }
+ if st.Mode&unix.S_IFMT != unix.S_IFREG {
+ return s, ErrState
+ }
+ b, err := io.ReadAll(io.LimitReader(f, 65537))
+ if err != nil {
+ return s, err
+ }
+ if len(b) > 65536 {
+ return s, ErrState
+ }
+ if err = decode(b, &s); err != nil {
+ return s, err
+ }
+ if err = s.validate(); err != nil {
+ return s, err
+ }
+ if s.Host != host {
+ return s, ErrState
+ }
+ return s, nil
+}
+func (d *localDir) write(s State) error {
+ if err := s.validate(); err != nil {
+ return err
+ }
+ if _, err := d.read(s.Host); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ var nonce [16]byte
+ if _, err := rand.Read(nonce[:]); err != nil {
+ return err
+ }
+ tmp := ".tmp-" + hex.EncodeToString(nonce[:])
+ fd, err := unix.Openat(d.fd, tmp, unix.O_WRONLY|unix.O_CREAT|unix.O_EXCL|unix.O_NOFOLLOW|unix.O_CLOEXEC, 0660)
+ if err != nil {
+ return err
+ }
+ defer unix.Unlinkat(d.fd, tmp, 0)
+ f := os.NewFile(uintptr(fd), tmp)
+ defer f.Close()
+ if err = unix.Fchmod(fd, 0660); err != nil {
+ return err
+ }
+ b, err := json.Marshal(s)
+ if err != nil {
+ return err
+ }
+ if _, err = f.Write(append(b, '\n')); err != nil {
+ return err
+ }
+ if err = d.sync(fd); err != nil {
+ return err
+ }
+ if err = f.Close(); err != nil {
+ return err
+ }
+ if err = unix.Renameat(d.fd, tmp, d.fd, s.Host+".json"); err != nil {
+ return err
+ }
+ return d.sync(d.fd)
+}
+func (d *localDir) remove(host string) error {
+ if _, err := d.read(host); err != nil {
+ return err
+ }
+ if err := unix.Unlinkat(d.fd, host+".json", 0); err != nil {
+ return err
+ }
+ return d.sync(d.fd)
+}
+func tryAcquire(dir, host string) (func(), error) {
+ return tryAcquireNamed(dir, host, ".lock")
+}
+
+func tryAcquireLive(dir, host string) (func(), error) {
+ return tryAcquireNamed(dir, host, ".live")
+}
+
+func tryAcquireNamed(dir, host, suffix string) (func(), error) {
+ if _, err := Key(host, "check"); err != nil {
+ return nil, err
+ }
+ fd, err := openDir(dir)
+ if err != nil {
+ return nil, err
+ }
+ defer unix.Close(fd)
+ flags := unix.O_RDWR | unix.O_CREAT | unix.O_NOFOLLOW | unix.O_CLOEXEC | unix.O_NONBLOCK
+ lock, err := unix.Openat(fd, host+suffix, flags|unix.O_EXCL, 0660)
+ created := err == nil
+ if errors.Is(err, unix.EEXIST) {
+ lock, err = unix.Openat(fd, host+suffix, flags, 0660)
+ }
+ if err != nil {
+ return nil, err
+ }
+ fail := func(e error) (func(), error) { unix.Close(lock); return nil, e }
+ var st unix.Stat_t
+ if err = unix.Fstat(lock, &st); err != nil {
+ return fail(err)
+ }
+ if st.Mode&unix.S_IFMT != unix.S_IFREG {
+ return fail(ErrState)
+ }
+ if created || st.Mode&07777 != 0660 {
+ if err = unix.Fchmod(lock, 0660); err != nil {
+ return fail(err)
+ }
+ }
+ err = unix.Flock(lock, unix.LOCK_EX|unix.LOCK_NB)
+ if err == nil {
+ return func() { _ = unix.Flock(lock, unix.LOCK_UN); _ = unix.Close(lock) }, nil
+ }
+ if errors.Is(err, unix.EWOULDBLOCK) || errors.Is(err, unix.EAGAIN) {
+ return fail(ErrControllerBusy)
+ }
+ return fail(err)
+}
+
+func acquire(ctx context.Context, dir, host string) (func(), error) {
+ for {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ unlock, err := tryAcquire(dir, host)
+ if !errors.Is(err, ErrControllerBusy) {
+ return unlock, err
+ }
+ select {
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ case <-time.After(10 * time.Millisecond):
+ }
+ }
+}
diff --git a/internal/hostexec/record.go b/internal/hostexec/record.go
new file mode 100644
index 000000000..4a5927099
--- /dev/null
+++ b/internal/hostexec/record.go
@@ -0,0 +1,94 @@
+package hostexec
+
+import (
+ "bytes"
+ "encoding/json"
+ "errors"
+ "io"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/controlname"
+)
+
+const TTL = 90 * time.Second
+const Renewal = 20 * time.Second
+const Margin = 5 * time.Second
+
+var ErrControllerBusy = errors.New("hostexec: controller already holds lifetime lock")
+var ErrLost = errors.New("hostexec: lease lost")
+var ErrInactive = errors.New("hostexec: no active participation")
+var ErrState = errors.New("hostexec: invalid local state")
+var ErrUnsupported = errors.New("hostexec: suspend-aware clock unavailable")
+
+// Record is the complete remote wire object. It grants no claim authority.
+type Record struct {
+ Op string `json:"op"`
+ Expires time.Time `json:"expires"`
+}
+
+type Binding struct {
+ Box string `json:"box"`
+ Enrollment string `json:"enrollment"`
+ Host string `json:"host"`
+ Incarnation string `json:"incarnation"`
+ Op string `json:"op"`
+ BootID string `json:"boot_id"`
+ ControllerID string `json:"controller_id"`
+}
+
+type State struct {
+ V int `json:"v"`
+ Binding
+ LeaseETag string `json:"lease_etag"`
+ Expires time.Time `json:"expires"`
+ SafeUntil int64 `json:"safe_until_boottime_ns"`
+ State string `json:"state"`
+}
+
+func validID(s string) bool {
+ return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00\r\n\t ")
+}
+func (b Binding) valid() bool {
+ return controlname.ValidHostShape(b.Host) && validID(b.Box) && validID(b.Enrollment) &&
+ validID(b.Incarnation) && validID(b.Op) && validID(b.BootID) && validID(b.ControllerID)
+}
+func (s State) validate() error {
+ if s.V != 1 || !s.Binding.valid() || s.SafeUntil <= 0 || s.Expires.IsZero() {
+ return ErrState
+ }
+ switch s.State {
+ case "pending", "lost":
+ case "active":
+ if s.LeaseETag == "" {
+ return ErrState
+ }
+ default:
+ return ErrState
+ }
+ return nil
+}
+func decode(b []byte, dst any) error {
+ d := json.NewDecoder(bytes.NewReader(b))
+ d.DisallowUnknownFields()
+ if err := d.Decode(dst); err != nil {
+ return err
+ }
+ var extra any
+ if err := d.Decode(&extra); err != io.EOF {
+ return ErrState
+ }
+ return nil
+}
+func Key(host, box string) (string, error) {
+ if !controlname.ValidHostShape(host) || !validID(box) {
+ return "", ErrState
+ }
+ return "control/hosts/" + host + "/exec/" + box + ".json", nil
+}
+
+// BootNanos includes time spent suspended. Wall time is used only on the wire.
+type Clock interface {
+ Now() (wall time.Time, bootNanos int64, err error)
+ After(time.Duration) <-chan time.Time
+}
diff --git a/internal/hostexec/review_test.go b/internal/hostexec/review_test.go
new file mode 100644
index 000000000..cf1c78e15
--- /dev/null
+++ b/internal/hostexec/review_test.go
@@ -0,0 +1,460 @@
+//go:build linux
+
+package hostexec
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "golang.org/x/sys/unix"
+ "os"
+ "os/exec"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+)
+
+func TestLostControllerCannotPublishStaleActiveState(t *testing.T) {
+ cfg, _, clock, _, _ := fixture(t)
+ c, err := Open(t.Context(), cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer closeController(t, c)
+ active := ready(t, c)
+ c.mu.Lock()
+ c.loseLocked(ErrLost)
+ c.mu.Unlock()
+ <-c.cleanupDone
+ if err = c.publish(active); !errors.Is(err, ErrLost) {
+ t.Fatalf("stale publication = %v", err)
+ }
+ if err = CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); err == nil {
+ t.Fatal("stale active state revived cross-process admission")
+ }
+}
+
+func TestWaitingPublicationDoesNotBlockWatchdogOrCleanup(t *testing.T) {
+ cfg, _, clock, _, cleaned := fixture(t)
+ c, err := Open(t.Context(), cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer closeController(t, c)
+ active := ready(t, c)
+ held, release := make(chan struct{}), make(chan struct{})
+ guardDone := make(chan error, 1)
+ go func() {
+ guardDone <- cfg.Guard.WithLocal(context.Background(), active.Host, func() error {
+ close(held)
+ <-release
+ return nil
+ })
+ }()
+ <-held
+ published := make(chan error, 1)
+ go func() { published <- c.publish(active) }()
+ clock.advance(TTL)
+ // Ready samples the suspend-aware clock even if a watchdog wake races.
+ if _, err = c.Ready(t.Context()); err == nil {
+ t.Fatal("expired controller remained ready")
+ }
+ select {
+ case <-c.Lost():
+ case <-time.After(2 * time.Second):
+ t.Fatal("host guard blocked watchdog")
+ }
+ until(t, func() bool { return cleaned.Load() >= 2 })
+ close(release)
+ if err = <-guardDone; err != nil {
+ t.Fatal(err)
+ }
+ if err = <-published; !errors.Is(err, ErrLost) {
+ t.Fatalf("publication after expiration = %v", err)
+ }
+ <-c.cleanupDone
+ if err = CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); err == nil {
+ t.Fatal("expired publication admitted work")
+ }
+}
+
+func TestOrphanActiveStateCannotAuthorizeAfterControllerExit(t *testing.T) {
+ cfg, _, clock, _, _ := fixture(t)
+ c, err := Open(t.Context(), cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ active := ready(t, c)
+ closeController(t, c)
+ // Simulate an active rename surviving a crash before its fsync outcome was
+ // known. The persisted deadline is still future, but no controller lives.
+ d, err := openLocal(cfg.Guard.DataDir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer d.close()
+ if err = d.write(active); err != nil {
+ t.Fatal(err)
+ }
+ if err = CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) {
+ t.Fatalf("orphan active state = %v", err)
+ }
+}
+
+type blockedClaimRead struct {
+ base *claims
+ entered chan struct{}
+ release chan struct{}
+}
+
+func (r blockedClaimRead) Read(ctx context.Context, host string) (hostclaim.Observed, error) {
+ close(r.entered)
+ select {
+ case <-ctx.Done():
+ return hostclaim.Observed{}, ctx.Err()
+ case <-r.release:
+ return r.base.Read(ctx, host)
+ }
+}
+
+func TestCC9LockDoesNotReviveOrphanDuringClaimRead(t *testing.T) {
+ cfg, r, clock, _, _ := fixture(t)
+ c, err := Open(t.Context(), cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ active := ready(t, c)
+ closeController(t, c)
+ d, err := openLocal(cfg.Guard.DataDir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer d.close()
+ if err = d.write(active); err != nil {
+ t.Fatal(err)
+ }
+ blocked := blockedClaimRead{base: r, entered: make(chan struct{}), release: make(chan struct{})}
+ cfg.Guard.Claims = blocked
+ done := make(chan error, 1)
+ go func() { done <- ClearStale(t.Context(), cfg) }()
+ <-blocked.entered
+ checkErr := CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir)
+ close(blocked.release)
+ if checkErr == nil {
+ t.Fatal("CC9 lifetime lock revived orphan active bytes")
+ }
+ if err = <-done; !errors.Is(err, ErrInactive) {
+ t.Fatalf("CC9 with participation retained = %v", err)
+ }
+}
+
+func TestCC9SuccessorExclusionClearsOnlyExpectedOperation(t *testing.T) {
+ cfg, r, _, _, _ := fixture(t)
+ c, err := Open(t.Context(), cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ready(t, c)
+ closeController(t, c)
+ r.mu.Lock()
+ r.q.Op.ID = "successor-op"
+ r.q.Op.Exec = "other-box"
+ r.q.Op.Participants = []string{"other-box"}
+ r.mu.Unlock()
+ if err = ClearStale(t.Context(), cfg); err != nil {
+ t.Fatalf("successor excludes box: %v", err)
+ }
+ d, err := openLocal(cfg.Guard.DataDir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer d.close()
+ if _, err = d.read(cfg.Binding.Host); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("expected stale record removed: %v", err)
+ }
+ unrelated := State{V: 1, Binding: cfg.Binding, State: "lost", Expires: time.Now().Add(TTL), SafeUntil: 99}
+ unrelated.ControllerID = "other-controller"
+ unrelated.Op = "unrelated-op"
+ if err = d.write(unrelated); err != nil {
+ t.Fatal(err)
+ }
+ if err = ClearStale(t.Context(), cfg); !errors.Is(err, ErrState) {
+ t.Fatalf("unrelated local fence: %v", err)
+ }
+ still, err := d.read(cfg.Binding.Host)
+ if err != nil || still.Op != unrelated.Op {
+ t.Fatalf("unrelated fence changed: %v, %v", still, err)
+ }
+}
+
+func TestAmbiguousRenewalENOSPCBeforeRenameDeniesImmediately(t *testing.T) {
+ cfg, _, clock, store, cleaned := fixture(t)
+ store.failAt = 2
+ var noSpace atomic.Bool
+ noSpace.Store(true)
+ var failures atomic.Int32
+ cfg.writeLocal = func(d *localDir, s State) error {
+ if s.State != "lost" || !noSpace.Load() {
+ return d.write(s)
+ }
+ original := d.sync
+ d.sync = func(int) error { return unix.ENOSPC }
+ defer func() { d.sync = original }()
+ err := d.write(s) // file fsync fails before rename; old active bytes remain
+ if errors.Is(err, unix.ENOSPC) {
+ failures.Add(1)
+ }
+ return err
+ }
+ c, err := Open(t.Context(), cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer func() { noSpace.Store(false); closeController(t, c) }()
+ active := ready(t, c)
+ until(t, func() bool {
+ clock.mu.Lock()
+ defer clock.mu.Unlock()
+ for _, a := range clock.alarms {
+ if a.at == clock.boot+int64(Renewal) {
+ return true
+ }
+ }
+ return false
+ })
+ clock.advance(Renewal)
+ select {
+ case <-c.Lost():
+ case <-time.After(2 * time.Second):
+ t.Fatal("ambiguous renewal did not latch loss")
+ }
+ // Check from the loss signal onward, before waiting for any persistence.
+ if err := CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) {
+ t.Fatalf("admitted immediately after ambiguity: %v", err)
+ }
+ until(t, func() bool { return failures.Load() >= 2 && cleaned.Load() >= 2 })
+ if _, err := c.Ready(t.Context()); err == nil {
+ t.Fatal("in-memory admission survived loss")
+ }
+ d, err := openLocal(cfg.Guard.DataDir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer d.close()
+ old, err := d.read(active.Host)
+ if err != nil || old != active {
+ t.Fatalf("test did not leave the old active file: %+v %v", old, err)
+ }
+ _, boot, _ := clock.Now()
+ if boot >= active.SafeUntil {
+ t.Fatal("test reached expiry instead of exercising immediate denial")
+ }
+ binding, _ := json.Marshal(active.Binding)
+ cmd := exec.CommandContext(t.Context(), os.Args[0], "-test.run=^TestCheckLocalRefusesLostLiveLock$")
+ cmd.Env = append(os.Environ(), "HOSTEXEC_PROBE_BINDING="+string(binding), "HOSTEXEC_PROBE_DATA="+cfg.Guard.DataDir, "HOSTEXEC_PROBE_LOCKS="+cfg.LockDir)
+ if output, err := cmd.CombinedOutput(); err != nil {
+ t.Fatalf("independent process admitted old active state: %v\n%s", err, output)
+ }
+ ctx, cancel := context.WithTimeout(t.Context(), 30*time.Millisecond)
+ defer cancel()
+ entered := false
+ err = cfg.Guard.WithLocal(ctx, active.Host, func() error { entered = true; return nil })
+ if !errors.Is(err, context.DeadlineExceeded) || entered {
+ t.Fatalf("guard released while denial write failed: %v entered=%v", err, entered)
+ }
+ if unlock, err := tryAcquire(cfg.LockDir, active.Host); !errors.Is(err, ErrControllerBusy) {
+ if unlock != nil {
+ unlock()
+ }
+ t.Fatalf("controller exclusion released: %v", err)
+ }
+ ctxClose, cancelClose := context.WithTimeout(t.Context(), 30*time.Millisecond)
+ defer cancelClose()
+ if err := c.Close(ctxClose); !errors.Is(err, context.DeadlineExceeded) {
+ t.Fatalf("Close did not retain the unfinished controller: %v", err)
+ }
+ noSpace.Store(false)
+ select {
+ case <-c.cleanupDone:
+ case <-time.After(2 * time.Second):
+ t.Fatal("lost state was not retried after disk recovery")
+ }
+ lost, err := d.read(active.Host)
+ if err != nil || lost.State != "lost" {
+ t.Fatalf("denial not durable: %+v %v", lost, err)
+ }
+ if err := cfg.Guard.WithLocal(t.Context(), active.Host, func() error {
+ if err := CheckLocal(cfg.Guard.DataDir, active.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) {
+ t.Fatalf("disk recovery revived admission: %v", err)
+ }
+ return nil
+ }); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestCheckLocalRefusesLostLiveLock(t *testing.T) {
+ raw := os.Getenv("HOSTEXEC_PROBE_BINDING")
+ if raw == "" {
+ t.Skip("subprocess witness")
+ }
+ var binding Binding
+ if err := json.Unmarshal([]byte(raw), &binding); err != nil {
+ t.Fatal(err)
+ }
+ clock := &testClock{boot: int64(time.Hour + Renewal)}
+ if err := CheckLocal(os.Getenv("HOSTEXEC_PROBE_DATA"), binding, clock, os.Getenv("HOSTEXEC_PROBE_LOCKS")); !errors.Is(err, ErrLost) {
+ t.Fatalf("old active record authorized an independent process: %v", err)
+ }
+}
+
+func TestOpenOrphanControllerPendingWriteFailureCleansBeforeReleasingExclusion(t *testing.T) {
+ testOpenOrphanControllerPersistence(t, false)
+}
+
+func TestOpenBlockedPendingPersistenceStillCleansBeforeReleasingExclusion(t *testing.T) {
+ testOpenOrphanControllerPersistence(t, true)
+}
+
+func testOpenOrphanControllerPersistence(t *testing.T, blocked bool) {
+ cfg, _, clock, store, _ := fixture(t)
+ wall, boot, _ := clock.Now()
+ orphan := State{V: 1, Binding: cfg.Binding, State: "active", LeaseETag: "orphan-etag", Expires: wall.Add(TTL), SafeUntil: boot + int64(TTL-Margin)}
+ orphan.ControllerID = "dead-controller"
+ d, err := openLocal(cfg.Guard.DataDir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer d.close()
+ if err := d.write(orphan); err != nil {
+ t.Fatal(err)
+ }
+ // The controller's locks have vanished, but its active bytes and a host
+ // process survived. Admission refusal alone cannot stop that process.
+ process := exec.Command("sleep", "60")
+ if err := process.Start(); err != nil {
+ t.Fatal(err)
+ }
+ defer process.Process.Kill()
+ hostExited := make(chan struct{})
+ go func() { _ = process.Wait(); close(hostExited) }()
+ allowJoin := make(chan struct{})
+ var joinOnce sync.Once
+ defer joinOnce.Do(func() { close(allowJoin) })
+ cleaned := make(chan struct{})
+ cleanupResult := make(chan error, 1)
+ cfg.Cleanup = func(ctx context.Context) error {
+ cleanupResult <- ctx.Err()
+ if err := process.Process.Kill(); err != nil {
+ return err
+ }
+ <-hostExited
+ close(cleaned)
+ <-allowJoin
+ return nil
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ writeEntered := make(chan struct{})
+ allowWrite := make(chan struct{})
+ var writeOnce sync.Once
+ defer writeOnce.Do(func() { close(allowWrite) })
+ var noSpace atomic.Bool
+ var failures atomic.Int32
+ noSpace.Store(true)
+ defer noSpace.Store(false)
+ cfg.writeLocal = func(d *localDir, s State) error {
+ if noSpace.Load() {
+ if s.State == "pending" {
+ cancel() // cleanup must not inherit the failed opener's lifetime
+ }
+ original := d.sync
+ d.sync = func(int) error {
+ if blocked && s.State == "pending" {
+ close(writeEntered)
+ <-allowWrite // model an fsync that cannot return until released
+ }
+ return unix.ENOSPC
+ }
+ defer func() { d.sync = original }()
+ err := d.write(s)
+ if errors.Is(err, unix.ENOSPC) {
+ failures.Add(1)
+ }
+ return err
+ }
+ return d.write(s)
+ }
+ done := make(chan error, 1)
+ go func() { _, err := Open(ctx, cfg); done <- err }()
+ select {
+ case <-cleaned:
+ case <-time.After(2 * time.Second):
+ t.Fatal("failed pending write skipped orphan process cleanup")
+ }
+ if err := <-cleanupResult; err != nil {
+ t.Fatalf("cleanup inherited caller cancellation: %v", err)
+ }
+
+ if old, err := d.read(orphan.Host); err != nil || old != orphan {
+ t.Fatalf("failed persistence did not retain orphan active bytes: %+v %v", old, err)
+ }
+ assertExcluded := func() {
+ t.Helper()
+ if unlock, err := tryAcquire(cfg.LockDir, orphan.Host); !errors.Is(err, ErrControllerBusy) {
+ if unlock != nil {
+ unlock()
+ }
+ t.Fatalf("exclusion released before cleanup joined: %v", err)
+ }
+ select {
+ case err := <-done:
+ t.Fatalf("Open returned before cleanup joined: %v", err)
+ default:
+ }
+ }
+ if blocked {
+ select {
+ case <-writeEntered:
+ case <-time.After(2 * time.Second):
+ t.Fatal("pending fsync was not reached")
+ }
+ if failures.Load() != 0 {
+ t.Fatal("pending fsync returned before cleanup was observed")
+ }
+ assertExcluded()
+ writeOnce.Do(func() { close(allowWrite) })
+ }
+ until(t, func() bool { return failures.Load() >= 1 })
+ assertExcluded()
+ if err := CheckLocal(cfg.Guard.DataDir, orphan.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) {
+ t.Fatalf("orphan bytes admitted work: %v", err)
+ }
+ // Persistence is still unavailable when cleanup joins. No live lock was
+ // acquired, so orphan bytes cannot authorize work after exclusion ends.
+ joinOnce.Do(func() { close(allowJoin) })
+ select {
+ case err := <-done:
+ if !errors.Is(err, unix.ENOSPC) {
+ t.Fatalf("Open lost pending-write failure: %v", err)
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("Open did not join cleanup while persistence remained unavailable")
+ }
+ unlock, err := tryAcquire(cfg.LockDir, orphan.Host)
+ if err != nil {
+ t.Fatalf("exclusion retained after cleanup joined: %v", err)
+ }
+ unlock()
+ if err := CheckLocal(cfg.Guard.DataDir, orphan.Binding, clock, cfg.LockDir); !errors.Is(err, ErrLost) {
+ t.Fatalf("released exclusion revived orphan bytes: %v", err)
+ }
+ store.mu.Lock()
+ defer store.mu.Unlock()
+ if store.calls != 0 {
+ t.Fatalf("failed startup attempted %d lease writes", store.calls)
+ }
+}
diff --git a/internal/hostguard/local.go b/internal/hostguard/local.go
new file mode 100644
index 000000000..2b4553096
--- /dev/null
+++ b/internal/hostguard/local.go
@@ -0,0 +1,13 @@
+package hostguard
+
+import "context"
+
+// WithLocal holds the existing host lock for local bookkeeping or cleanup.
+// It grants no claim, installation, spawn or write authority. Ordinary host
+// work must still use With. The callback must not wait on a network request.
+func (g Guard) WithLocal(ctx context.Context, host string, fn func() error) error {
+ if fn == nil {
+ return ErrInvalidInput
+ }
+ return g.locked(ctx, host, func(_ *recordDir) error { return fn() })
+}
diff --git a/internal/hostguard/local_test.go b/internal/hostguard/local_test.go
new file mode 100644
index 000000000..a9b5f5149
--- /dev/null
+++ b/internal/hostguard/local_test.go
@@ -0,0 +1,45 @@
+//go:build unix
+
+package hostguard
+
+import (
+ "context"
+ "errors"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+)
+
+func TestWithLocalSerializesWithoutClaimAuthority(t *testing.T) {
+ dir := t.TempDir()
+ locks := t.TempDir()
+ if err := os.Mkdir(filepath.Join(dir, "hostguard"), 0770); err != nil {
+ t.Fatal(err)
+ }
+ g := Guard{DataDir: dir, LockDir: locks}
+ entered := make(chan struct{})
+ release := make(chan struct{})
+ done := make(chan error, 1)
+ go func() {
+ done <- g.WithLocal(t.Context(), "shop.example.com", func() error { close(entered); <-release; return nil })
+ }()
+ <-entered
+ ctx, cancel := context.WithTimeout(t.Context(), 25*time.Millisecond)
+ defer cancel()
+ called := false
+ err := g.WithLocal(ctx, "shop.example.com", func() error { called = true; return nil })
+ close(release)
+ if first := <-done; first != nil {
+ t.Fatal(first)
+ }
+ if !errors.Is(err, context.DeadlineExceeded) || called {
+ t.Fatalf("concurrent local write: %v %v", err, called)
+ }
+ // No Claims reader is configured: WithLocal may clean up, but With cannot
+ // turn the same lock into installation authority.
+ if err := g.With(t.Context(), "shop.example.com", Captured{}, func(_ *hostclaim.Claim) error { return nil }, func() error { return nil }); !errors.Is(err, ErrNoClaims) {
+ t.Fatal(err)
+ }
+}
diff --git a/internal/lease/execution_store.go b/internal/lease/execution_store.go
new file mode 100644
index 000000000..0fe927281
--- /dev/null
+++ b/internal/lease/execution_store.go
@@ -0,0 +1,20 @@
+package lease
+
+import (
+ "context"
+ "errors"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// ExecutionStore reuses the replica leaser's S3 configuration and the release
+// store's conditional primitives. The execution controller owns its wire schema.
+func ExecutionStore(ctx context.Context, bucket, region string) (release.Store, error) {
+ if bucket == "" {
+ return nil, errors.New("lease: control bucket required")
+ }
+ client, _, err := newLeaserClient(ctx, region)
+ if err != nil {
+ return nil, err
+ }
+ return &release.S3Store{Client: client, Bucket: bucket}, nil
+}
diff --git a/internal/lease/execution_store_test.go b/internal/lease/execution_store_test.go
new file mode 100644
index 000000000..19d68e72f
--- /dev/null
+++ b/internal/lease/execution_store_test.go
@@ -0,0 +1,33 @@
+package lease
+
+import (
+ "github.com/carlosframework/platform/internal/release"
+ "testing"
+)
+
+func TestExecutionStoreUsesExistingConditionalTransport(t *testing.T) {
+ t.Setenv("AWS_ACCESS_KEY_ID", "test")
+ t.Setenv("AWS_SECRET_ACCESS_KEY", "test")
+ t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
+ t.Setenv("AWS_CONFIG_FILE", t.TempDir()+"/absent")
+ t.Setenv("AWS_SHARED_CREDENTIALS_FILE", t.TempDir()+"/absent")
+ t.Setenv("CARLOS_S3_ENDPOINT", "https://objects.invalid")
+ if _, err := ExecutionStore(t.Context(), "", "eu-west-1"); err == nil {
+ t.Fatal("empty bucket accepted")
+ }
+ store, err := ExecutionStore(t.Context(), "control", "eu-west-1")
+ if err != nil {
+ t.Fatal(err)
+ }
+ s, ok := store.(*release.S3Store)
+ if !ok || s.Bucket != "control" {
+ t.Fatal("wrong store")
+ }
+ opts := s.Client.Options()
+ if opts.Region != "eu-west-1" || opts.BaseEndpoint == nil || *opts.BaseEndpoint != "https://objects.invalid" || !opts.UsePathStyle {
+ t.Fatalf("lost leaser configuration: %+v", opts)
+ }
+ if _, ok := store.(release.ConditionalDeleter); !ok {
+ t.Fatal("no conditional release")
+ }
+}