diff --git a/docs/guides/client-pool.md b/docs/guides/client-pool.md index f9b36a431..bf05e93da 100644 --- a/docs/guides/client-pool.md +++ b/docs/guides/client-pool.md @@ -298,41 +298,71 @@ best-effort kill attempt. The Go method is intentionally outside the ### Retiring an old pool namespace -The retirement procedure differs across SDKs because Go does not currently ship a -`SandboxPoolManager` or the destroy / tombstone primitives that Python and Kotlin -have. - -**Python / Kotlin** — use `SandboxPoolManager.destroy(poolName, options)`. The manager -applies a full `DESTROYING → DESTROYED` protocol: write a `DESTROYING` fence into the -state store (so any still-running peer instance sees it), best-effort drain and kill -every idle sandbox up to `drain_timeout`, clear the persistent per-pool state, then -write a `DESTROYED` tombstone with `tombstone_ttl` (default 7 days) so future callers -cannot silently rebind to the same `pool_name`. - -**Go** — the Go SDK has no equivalent API and no state-store primitives for -tombstones or fences. The closest safe approximation is an operator-driven, out-of-band -sequence: - -1. Stop every process that instantiates a pool against the old `pool_name`. Call - `pool.Shutdown(ctx, true)` on each. This releases each node's primary lock but - leaves idle entries in the store. -2. From one still-alive pool instance (or a throwaway one bound to the same - `PoolName` + `StateStore`), call `pool.ReleaseAllIdle(ctx)` to drain and - best-effort kill every idle sandbox. -3. Set `store.SetMaxIdle(ctx, poolName, 0)` so any peer that races back in cannot - warm up new sandboxes. -4. Move all future callers to a new `PoolName` (for example - `orders-v2` → `orders-v3`). This is the Go substitute for the `DESTROYED` - tombstone: without a shared marker, name rotation is the only way to guarantee no - accidental reuse. -5. If you are using the Redis store and want to reclaim keys, delete them directly - with `DEL` / `SCAN` against your Redis instance — the Go SDK does not expose a - destroy helper for this. - -Without a fence, steps 2 and 3 race with any surviving peer that has not yet been -stopped. If you cannot guarantee "all writers stopped" before step 2, the only correct -option is to rotate `PoolName` first (step 4) and let the old namespace's idle entries -expire naturally via `idle_timeout`. +Every SDK exposes a `SandboxPoolManager` with a `destroy` operation that applies the +same `DESTROYING → DESTROYED` protocol: + +1. Write a `DESTROYING` fence into the state store, so any still-running peer instance + sees it and stops replenishing instead of racing the retirement. +2. Best-effort drain and kill every idle sandbox, bounded by the drain timeout. +3. Clear the persistent per-pool state. +4. Write a `DESTROYED` tombstone with the tombstone TTL (default 7 days) so future + callers cannot silently rebind to the same `pool_name`. + +Destroy is idempotent: calling it on an already-tombstoned namespace reports +`DESTROYED` without draining or killing anything. If the drain or the cleanup cannot +finish, the namespace stays `DESTROYING` and the call reports the destroy as +incomplete; retrying is safe and picks up where it left off. + +**Python / Kotlin** — `SandboxPoolManager.destroy(poolName, options)`, configured +through `PoolDestroyOptions` (`strategy`, `drain_timeout`, `tombstone_ttl`). + +**Go** — `(*SandboxPoolManager).Destroy(ctx, poolName, options)`: + +```go +manager, err := opensandbox.NewSandboxPoolManagerBuilder(). + StateStore(store). + ConnectionConfig(connCfg). + Build() +if err != nil { + return err +} + +result, err := manager.Destroy(ctx, "orders-v2", opensandbox.PoolDestroyOptions{}) +if err != nil { + return err +} +log.Printf("retired %s: drained=%d killed=%d", + result.PoolName, result.DrainedIdleCount, result.KilledIdleCount) +``` + +`PoolDestroyOptions` mirrors the other SDKs. `Strategy` selects the algorithm and only +`PoolDestroyForce` is implemented. `DrainTimeout` and `TombstoneTTL` are `*time.Duration`: +leave them nil for the defaults (30s and 7 days), or set an explicit zero to drain +without a deadline and to write a tombstone that never expires. + +The fence is what makes retirement safe without stopping every writer first, and it +is enforced on two levels. The state store refuses `PutIdle`, `SetMaxIdle` and +`SetIdleEntryTTL` with a `*PoolDestroyedError` and hands out no primary lock, which +stops replenishment. The pool itself also checks the fence when it starts, before every +acquire, again once an acquire holds a live sandbox, and on each reconcile tick: a +surviving peer stops outright on its next tick, an in-flight acquire fails rather +than minting a fresh sandbox into the retired namespace through the direct-create +fallthrough, and a sandbox obtained just before the fence landed is killed instead +of handed out. The post-acquire check matters because the idle take is deliberately +left unfenced so `destroy` can drain: once an ID has been taken, `destroy` can no +longer reach it, so the acquire has to dispose of it itself. +Starting a fresh pool against a tombstoned `PoolName` fails for the same reason, so +rebinding the name requires either waiting out the tombstone TTL or rotating to a new +`PoolName`. + +One deliberate exception: if the state store itself is unreachable, the destroy state +is unknowable, so policies that already fall through to direct create on a store +outage (`DIRECT_CREATE`, `RETRY_NEXT_IDLE_THEN_CREATE`) assume `ACTIVE` and proceed, +matching the existing `try_take_idle` outage behavior in the OSEP-0005 error-code +matrix. `FAIL_FAST` and `RETRY_NEXT_IDLE` surface the outage instead. That relaxation +stops at a sandbox already taken from the idle buffer: there the check is fail-closed +and an unreachable store means the sandbox is killed, because nothing else is tracking +it any more. ## Further reading diff --git a/sdks/sandbox/go/pool.go b/sdks/sandbox/go/pool.go index 7114e59f1..1048ea3ca 100644 --- a/sdks/sandbox/go/pool.go +++ b/sdks/sandbox/go/pool.go @@ -16,6 +16,7 @@ package opensandbox import ( "context" + "errors" "fmt" "sync" "sync/atomic" @@ -91,6 +92,20 @@ func (p *DefaultSandboxPool) Start(ctx context.Context) error { startMaxIdle := p.config.MaxIdle p.mu.Unlock() + // Refuse to bind a retired namespace. Only a definite fence blocks startup; + // a store outage is left to the writes below to surface. + if err := p.ensureNamespaceActive(ctx); err != nil { + var destroyed *PoolDestroyedError + if errors.As(err, &destroyed) { + p.mu.Lock() + if p.lifecycleState == PoolLifecycleStarting { + p.lifecycleState = PoolLifecycleNotStarted + } + p.mu.Unlock() + return err + } + } + // Initialize state store with pool configuration. if err := p.config.StateStore.SetMaxIdle(ctx, p.config.PoolName, startMaxIdle); err != nil { p.mu.Lock() @@ -187,6 +202,17 @@ func (p *DefaultSandboxPool) syncHealthState() { func (p *DefaultSandboxPool) runReconcileTick(ctx context.Context) { p.reconMu.Lock() defer p.reconMu.Unlock() + + // A destroy fences the namespace for every peer. Stop rather than keep + // replenishing a pool that is being retired. + if err := p.ensureNamespaceActive(ctx); err != nil { + var destroyed *PoolDestroyedError + if errors.As(err, &destroyed) { + p.stopAfterNamespaceDestroyed(destroyed.State) + return + } + } + createFn := func(ctx context.Context, reason PooledSandboxCreateReason) (string, error) { return p.createOneSandbox(ctx, reason) } @@ -215,6 +241,12 @@ func (p *DefaultSandboxPool) Acquire(ctx context.Context, opts AcquireOptions) ( policy = *opts.Policy } + // A fenced namespace must not mint new sandboxes, so this has to run before the + // direct-create fallthrough below and not only on the store write paths. + if err := p.ensureNamespaceActiveForAcquire(ctx, policy); err != nil { + return nil, err + } + // Resolve minTTL. minTTL := p.config.AcquireMinRemainingTTL if opts.MinRemainingTTL > 0 { @@ -297,6 +329,12 @@ func (p *DefaultSandboxPool) Acquire(ctx context.Context, opts AcquireOptions) ( go p.killDiscardedAliveSandboxes(pendingKill) return nil, &PoolNotRunningError{PoolName: p.config.PoolName, State: currentState} } + // A destroy may have landed since the preflight check. Stop retrying rather + // than pop further idle IDs out from under the drain. + if err := p.ensureNamespaceActiveForAcquire(ctx, policy); err != nil { + go p.killDiscardedAliveSandboxes(pendingKill) + return nil, err + } continue } // Connect + readiness succeeded. From here on the sandbox is a healthy, borrowable idle: @@ -320,6 +358,15 @@ func (p *DefaultSandboxPool) Acquire(ctx context.Context, opts AcquireOptions) ( return nil, fmt.Errorf("opensandbox: pool acquire: renew after connect failed: %w", renewErr) } } + // TryTakeIdle is unfenced so the destroy manager can drain, so this ID is + // already out of the store and a destroy can no longer reach it. Re-check + // before handing it over, fail-closed: if the store cannot confirm the + // namespace is ACTIVE, kill the sandbox rather than leak it into a + // namespace that may be retired. + if err := p.ensureNamespaceActiveAfterCreate(ctx, sb, nil); err != nil { + go p.killDiscardedAliveSandboxes(pendingKill) + return nil, err + } go p.killDiscardedAliveSandboxes(pendingKill) p.config.Logger.Debug("acquire: from idle", "pool_name", p.config.PoolName, @@ -348,7 +395,7 @@ func (p *DefaultSandboxPool) Acquire(ctx context.Context, opts AcquireOptions) ( "attempted_any", attemptedAny, "loop_exhausted", loopExhausted, "last_sandbox_id", lastSandboxID) - return p.directCreate(ctx, opts) + return p.directCreate(ctx, opts, policy) } // tryTakeIdle wraps the store's take primitives, returning a nil result on a legitimate empty @@ -404,7 +451,7 @@ func (p *DefaultSandboxPool) connectIdle(ctx context.Context, sandboxID string, }) } -func (p *DefaultSandboxPool) directCreate(ctx context.Context, opts AcquireOptions) (*Sandbox, error) { +func (p *DefaultSandboxPool) directCreate(ctx context.Context, opts AcquireOptions, policy AcquirePolicy) (*Sandbox, error) { var sb *Sandbox var err error @@ -428,7 +475,15 @@ func (p *DefaultSandboxPool) directCreate(ctx context.Context, opts AcquireOptio if err != nil { return nil, err } - return p.postCreateChecks(ctx, sb, opts) + sb, err = p.postCreateChecks(ctx, sb, opts) + if err != nil { + return nil, err + } + // Re-check: a destroy may have landed while this sandbox was being created. + if err := p.ensureNamespaceActiveAfterCreate(ctx, sb, &policy); err != nil { + return nil, err + } + return sb, nil } // postCreateChecks applies renew to a freshly created sandbox. @@ -810,6 +865,109 @@ done: return nil } +// ensureNamespaceActive returns *PoolDestroyedError when a destroy has fenced or +// tombstoned this pool's namespace, and *PoolStateStoreUnavailableError when the +// state store cannot answer. +func (p *DefaultSandboxPool) ensureNamespaceActive(ctx context.Context) error { + state, err := p.config.StateStore.GetDestroyState(ctx, p.config.PoolName) + if err != nil { + var unavailable *PoolStateStoreUnavailableError + if errors.As(err, &unavailable) { + return err + } + return &PoolStateStoreUnavailableError{Operation: "GetDestroyState", Cause: err} + } + if state != PoolDestroyStateActive { + return &PoolDestroyedError{PoolName: p.config.PoolName, State: state} + } + return nil +} + +// ensureNamespaceActiveForAcquire is ensureNamespaceActive with the same +// store-outage degradation the take path already applies: policies that fall +// through to direct create treat an unreachable store as "state unknown" and +// proceed, so a full store outage does not make them less available than +// documented (OSEP-0005 error-code matrix). Fail-closed policies surface it. +func (p *DefaultSandboxPool) ensureNamespaceActiveForAcquire(ctx context.Context, policy AcquirePolicy) error { + err := p.ensureNamespaceActive(ctx) + if err == nil { + return nil + } + var unavailable *PoolStateStoreUnavailableError + if errors.As(err, &unavailable) && policyFallsThroughToDirectCreate(policy) { + p.config.Logger.Warn("acquire: state store unavailable during namespace check, "+ + "assuming ACTIVE and degrading to direct create", + "pool_name", p.config.PoolName, + "policy", policy, + "error", err) + return nil + } + return err +} + +// ensureNamespaceActiveAfterCreate re-checks the fence once the acquire path holds +// a live sandbox, so a destroy that landed mid-acquire does not leak one into a +// retired namespace. On a fence the sandbox is killed and closed. +// +// policy is non-nil only for the direct-create path, where a store outage degrades +// the same way the rest of that path does. The idle path passes nil and stays +// fail-closed: that sandbox is already out of the store, so an unconfirmed +// namespace has to be treated as retired. +func (p *DefaultSandboxPool) ensureNamespaceActiveAfterCreate(ctx context.Context, sb *Sandbox, policy *AcquirePolicy) error { + err := p.ensureNamespaceActive(ctx) + if err == nil { + return nil + } + var unavailable *PoolStateStoreUnavailableError + if errors.As(err, &unavailable) && policy != nil && policyFallsThroughToDirectCreate(*policy) { + p.config.Logger.Warn("acquire: state store unavailable during post-create namespace check, "+ + "keeping sandbox and degrading per policy", + "pool_name", p.config.PoolName, + "sandbox_id", sb.ID(), + "policy", *policy, + "error", err) + return nil + } + go p.killSandboxBestEffort(sb.ID()) + _ = sb.Close() + return err +} + +// stopAfterNamespaceDestroyed stops the pool once its namespace has been retired. +// It runs on the reconcile goroutine, so unlike Shutdown it must not wait on p.wg. +func (p *DefaultSandboxPool) stopAfterNamespaceDestroyed(state PoolDestroyState) { + p.mu.Lock() + if p.lifecycleState == PoolLifecycleStopped || p.lifecycleState == PoolLifecycleDraining { + p.mu.Unlock() + return + } + p.lifecycleState = PoolLifecycleStopped + if p.ticker != nil { + p.ticker.Stop() + } + if p.done != nil && !p.doneClosed { + close(p.done) + p.doneClosed = true + } + cancelFn := p.reconCancel + sdCh := p.shutdownDone + p.mu.Unlock() + + if cancelFn != nil { + cancelFn() + } + if sdCh != nil { + select { + case <-sdCh: + default: + close(sdCh) + } + } + p.config.Logger.Info("pool stopped: namespace destroyed", + "pool_name", p.config.PoolName, + "destroy_state", state) +} + const ( killSandboxTimeout = 30 * time.Second ) diff --git a/sdks/sandbox/go/pool_errors.go b/sdks/sandbox/go/pool_errors.go index b1251b222..fcb7d6900 100644 --- a/sdks/sandbox/go/pool_errors.go +++ b/sdks/sandbox/go/pool_errors.go @@ -63,3 +63,31 @@ func (e *PoolStateStoreUnavailableError) Error() string { } func (e *PoolStateStoreUnavailableError) Unwrap() error { return e.Cause } + +// PoolDestroyedError is returned when a write targets a pool namespace that a +// destroy has fenced, i.e. one that is DESTROYING or DESTROYED. +type PoolDestroyedError struct { + PoolName string + State PoolDestroyState +} + +func (e *PoolDestroyedError) Error() string { + return fmt.Sprintf("opensandbox: pool %q is %s", e.PoolName, e.State) +} + +// PoolDestroyIncompleteError is returned when a destroy could not run to +// completion. The namespace stays DESTROYING and the caller should retry. +type PoolDestroyIncompleteError struct { + PoolName string + Reason string + Cause error +} + +func (e *PoolDestroyIncompleteError) Error() string { + if e.Cause == nil { + return fmt.Sprintf("opensandbox: pool %q destroy incomplete: %s", e.PoolName, e.Reason) + } + return fmt.Sprintf("opensandbox: pool %q destroy incomplete: %s: %v", e.PoolName, e.Reason, e.Cause) +} + +func (e *PoolDestroyIncompleteError) Unwrap() error { return e.Cause } diff --git a/sdks/sandbox/go/pool_manager.go b/sdks/sandbox/go/pool_manager.go new file mode 100644 index 000000000..871a9b281 --- /dev/null +++ b/sdks/sandbox/go/pool_manager.go @@ -0,0 +1,248 @@ +// Copyright 2026 Alibaba Group Holding Ltd. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package opensandbox + +import ( + "context" + crypto_rand "crypto/rand" + "errors" + "fmt" + "os" + "strings" + "time" +) + +// SandboxPoolManager performs namespace-level maintenance on a shared sandbox +// pool. It does not acquire sandboxes: it exists so an operator can retire a +// pool namespace without reconstructing the SandboxPool object that originally +// owned it. +type SandboxPoolManager struct { + stateStore PoolStateStore + manager *SandboxManager + ownerID string + logger PoolLogger +} + +// Destroy retires a pool namespace. +// +// FORCE destroy writes a shared DESTROYING fence first, which every peer sharing +// the state store observes: fenced pools cannot hold the primary lock or publish +// idle sandboxes, so replenishment stops instead of racing the destroy. The +// manager then drains the visible idle IDs and kills them best-effort, clears the +// persistent coordination state, and writes a DESTROYED tombstone so later +// callers cannot silently rebind the namespace. +// +// If the drain or the cleanup cannot finish, the namespace stays DESTROYING and +// the error is a *PoolDestroyIncompleteError; retrying Destroy is safe. Calling +// Destroy on an already-tombstoned namespace is a no-op that reports DESTROYED. +func (m *SandboxPoolManager) Destroy(ctx context.Context, poolName string, options PoolDestroyOptions) (*PoolDestroyResult, error) { + if strings.TrimSpace(poolName) == "" { + return nil, fmt.Errorf("opensandbox: pool manager: poolName must not be blank") + } + if options.Strategy != PoolDestroyForce { + return nil, fmt.Errorf("opensandbox: pool manager: only FORCE destroy strategy is supported, got %s", options.Strategy) + } + + drainTimeout := DefaultPoolDrainTimeout + if options.DrainTimeout != nil { + drainTimeout = *options.DrainTimeout + if drainTimeout < 0 { + return nil, fmt.Errorf("opensandbox: pool manager: DrainTimeout must not be negative, got %v", drainTimeout) + } + } + tombstoneTTL := DefaultPoolTombstoneTTL + if options.TombstoneTTL != nil { + tombstoneTTL = *options.TombstoneTTL + if tombstoneTTL < 0 { + return nil, fmt.Errorf("opensandbox: pool manager: TombstoneTTL must not be negative, got %v", tombstoneTTL) + } + } + + state, err := m.stateStore.GetDestroyState(ctx, poolName) + if err != nil { + return nil, err + } + if state == PoolDestroyStateDestroyed { + return alreadyDestroyedResult(poolName), nil + } + + if err := m.stateStore.BeginDestroy(ctx, poolName, m.ownerID); err != nil { + // Lost the race to a concurrent destroy that already tombstoned the + // namespace; the outcome the caller asked for already holds. + var destroyed *PoolDestroyedError + if errors.As(err, &destroyed) { + return alreadyDestroyedResult(poolName), nil + } + return nil, err + } + + drained := 0 + killed := 0 + deadline := time.Now().Add(drainTimeout) + for { + sandboxID, err := m.stateStore.TryTakeIdle(ctx, poolName) + if err != nil { + return nil, &PoolDestroyIncompleteError{ + PoolName: poolName, + Reason: fmt.Sprintf("failed to drain idle sandboxes after %d drained", drained), + Cause: err, + } + } + if sandboxID == "" { + break + } + drained++ + if err := m.manager.KillSandbox(ctx, sandboxID); err != nil { + m.logger.Warn("pool destroy failed to kill idle sandbox (best-effort)", + "pool_name", poolName, + "sandbox_id", sandboxID, + "error", err) + } else { + killed++ + } + if drainTimeout > 0 && time.Now().After(deadline) { + return nil, &PoolDestroyIncompleteError{ + PoolName: poolName, + Reason: fmt.Sprintf("drain timed out after %v with %d idle sandboxes drained", drainTimeout, drained), + } + } + } + + if err := m.stateStore.ClearPoolState(ctx, poolName); err != nil { + return nil, &PoolDestroyIncompleteError{ + PoolName: poolName, + Reason: "failed to clear persistent state", + Cause: err, + } + } + if err := m.stateStore.MarkDestroyed(ctx, poolName, m.ownerID, tombstoneTTL); err != nil { + return nil, &PoolDestroyIncompleteError{ + PoolName: poolName, + Reason: "failed to write destroyed tombstone", + Cause: err, + } + } + + m.logger.Info("pool namespace destroyed", + "pool_name", poolName, + "drained_idle_count", drained, + "killed_idle_count", killed) + + return &PoolDestroyResult{ + PoolName: poolName, + State: PoolDestroyStateDestroyed, + DrainedIdleCount: drained, + KilledIdleCount: killed, + PersistentStateCleared: true, + }, nil +} + +func alreadyDestroyedResult(poolName string) *PoolDestroyResult { + return &PoolDestroyResult{ + PoolName: poolName, + State: PoolDestroyStateDestroyed, + DrainedIdleCount: 0, + KilledIdleCount: 0, + PersistentStateCleared: false, + } +} + +// SandboxPoolManagerBuilder configures and creates a SandboxPoolManager. +type SandboxPoolManagerBuilder struct { + stateStore PoolStateStore + connectionConfig ConnectionConfig + connectionConfigSet bool + ownerID string + logger PoolLogger +} + +// NewSandboxPoolManagerBuilder creates a new builder. +func NewSandboxPoolManagerBuilder() *SandboxPoolManagerBuilder { + return &SandboxPoolManagerBuilder{} +} + +// StateStore sets the pool state store to operate on (required). It must be the +// same store the pool being retired coordinates through. +func (b *SandboxPoolManagerBuilder) StateStore(s PoolStateStore) *SandboxPoolManagerBuilder { + b.stateStore = s + return b +} + +// ConnectionConfig sets the connection configuration used to kill drained +// sandboxes (required). +func (b *SandboxPoolManagerBuilder) ConnectionConfig(c ConnectionConfig) *SandboxPoolManagerBuilder { + b.connectionConfig = c + b.connectionConfigSet = true + return b +} + +// OwnerID sets the identifier recorded alongside the fence and the tombstone. +// Defaults to a generated per-process value. +func (b *SandboxPoolManagerBuilder) OwnerID(id string) *SandboxPoolManagerBuilder { + b.ownerID = id + return b +} + +// PoolLogger sets a custom structured logger. Defaults to a no-op logger. +func (b *SandboxPoolManagerBuilder) PoolLogger(l PoolLogger) *SandboxPoolManagerBuilder { + b.logger = l + return b +} + +// Build validates configuration and creates a SandboxPoolManager. +func (b *SandboxPoolManagerBuilder) Build() (*SandboxPoolManager, error) { + if b.stateStore == nil { + return nil, fmt.Errorf("opensandbox: pool manager builder: StateStore is required") + } + if !b.connectionConfigSet { + return nil, fmt.Errorf("opensandbox: pool manager builder: ConnectionConfig is required") + } + + ownerID := b.ownerID + if ownerID == "" { + generated, err := generatePoolManagerOwnerID() + if err != nil { + return nil, err + } + ownerID = generated + } + if strings.TrimSpace(ownerID) == "" { + return nil, fmt.Errorf("opensandbox: pool manager builder: OwnerID must not be blank") + } + + logger := b.logger + if logger == nil { + logger = noopPoolLogger{} + } + + return &SandboxPoolManager{ + stateStore: b.stateStore, + manager: NewSandboxManager(b.connectionConfig), + ownerID: ownerID, + logger: logger, + }, nil +} + +func generatePoolManagerOwnerID() (string, error) { + hostname, err := os.Hostname() + if err != nil || hostname == "" { + hostname = "unknown" + } + var randBytes [4]byte + if _, randErr := crypto_rand.Read(randBytes[:]); randErr != nil { + return "", fmt.Errorf("opensandbox: pool manager builder: failed to generate random owner ID: %w", randErr) + } + return fmt.Sprintf("pool-manager-%s-%d-%d-%x", hostname, os.Getpid(), time.Now().UnixNano(), randBytes), nil +} diff --git a/sdks/sandbox/go/pool_manager_test.go b/sdks/sandbox/go/pool_manager_test.go new file mode 100644 index 000000000..da9335cf0 --- /dev/null +++ b/sdks/sandbox/go/pool_manager_test.go @@ -0,0 +1,1020 @@ +// Copyright 2026 Alibaba Group Holding Ltd. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package opensandbox + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// killRecorder is a mock lifecycle server that records DELETE calls and can be +// told to fail them or to stall before answering. +type killRecorder struct { + srv *httptest.Server + deleted atomic.Int32 + fail atomic.Bool + delay atomic.Int64 // nanoseconds +} + +func newKillRecorder(t *testing.T) *killRecorder { + t.Helper() + rec := &killRecorder{} + rec.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodDelete { + w.WriteHeader(http.StatusNotFound) + return + } + if d := time.Duration(rec.delay.Load()); d > 0 { + time.Sleep(d) + } + rec.deleted.Add(1) + if rec.fail.Load() { + w.WriteHeader(http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusNoContent) + })) + t.Cleanup(rec.srv.Close) + return rec +} + +func newTestPoolManager(t *testing.T, store PoolStateStore, serverURL string) *SandboxPoolManager { + t.Helper() + manager, err := NewSandboxPoolManagerBuilder(). + StateStore(store). + ConnectionConfig(ConnectionConfig{Domain: serverURL, Protocol: "http"}). + OwnerID("test-pool-manager"). + Build() + if err != nil { + t.Fatalf("newTestPoolManager: Build failed: %v", err) + } + return manager +} + +func seedIdle(t *testing.T, store PoolStateStore, poolName string, n int) { + t.Helper() + ctx := context.Background() + for i := 0; i < n; i++ { + if err := store.PutIdle(ctx, poolName, fmt.Sprintf("sbx-idle-%d", i)); err != nil { + t.Fatalf("seedIdle: PutIdle failed: %v", err) + } + } +} + +func durationPtr(d time.Duration) *time.Duration { return &d } + +// ---------- Destroy Protocol Tests ---------- + +func TestSandboxPoolManager_Destroy(t *testing.T) { + tests := []struct { + name string + idleCount int + killsFail bool + options PoolDestroyOptions + wantDrained int + wantKilled int + wantDeletions int32 + }{ + { + name: "empty pool", + options: PoolDestroyOptions{}, + }, + { + name: "drains and kills every idle sandbox", + idleCount: 3, + wantDrained: 3, + wantKilled: 3, + wantDeletions: 3, + }, + { + name: "kill failures are best-effort", + idleCount: 2, + killsFail: true, + wantDrained: 2, + wantKilled: 0, + wantDeletions: 2, + }, + { + name: "zero drain timeout disables the deadline", + idleCount: 2, + options: PoolDestroyOptions{DrainTimeout: durationPtr(0)}, + wantDrained: 2, + wantKilled: 2, + wantDeletions: 2, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + store := NewInMemoryPoolStateStore() + rec := newKillRecorder(t) + rec.fail.Store(tt.killsFail) + seedIdle(t, store, "test-pool", tt.idleCount) + + manager := newTestPoolManager(t, store, rec.srv.URL) + result, err := manager.Destroy(ctx, "test-pool", tt.options) + if err != nil { + t.Fatalf("Destroy failed: %v", err) + } + + if result.State != PoolDestroyStateDestroyed { + t.Errorf("state = %s, want DESTROYED", result.State) + } + if result.PoolName != "test-pool" { + t.Errorf("poolName = %q, want %q", result.PoolName, "test-pool") + } + if !result.PersistentStateCleared { + t.Error("PersistentStateCleared = false, want true") + } + if result.DrainedIdleCount != tt.wantDrained { + t.Errorf("DrainedIdleCount = %d, want %d", result.DrainedIdleCount, tt.wantDrained) + } + if result.KilledIdleCount != tt.wantKilled { + t.Errorf("KilledIdleCount = %d, want %d", result.KilledIdleCount, tt.wantKilled) + } + if got := rec.deleted.Load(); got != tt.wantDeletions { + t.Errorf("DELETE requests = %d, want %d", got, tt.wantDeletions) + } + + state, err := store.GetDestroyState(ctx, "test-pool") + if err != nil { + t.Fatalf("GetDestroyState failed: %v", err) + } + if state != PoolDestroyStateDestroyed { + t.Errorf("store destroy state = %s, want DESTROYED", state) + } + counters, err := store.SnapshotCounters(ctx, "test-pool") + if err != nil { + t.Fatalf("SnapshotCounters failed: %v", err) + } + if counters.IdleCount != 0 { + t.Errorf("idle count after destroy = %d, want 0", counters.IdleCount) + } + }) + } +} + +// recordingPoolLogger captures warnings so tests can assert on best-effort paths. +type recordingPoolLogger struct { + mu sync.Mutex + warns []string +} + +func (l *recordingPoolLogger) Info(_ string, _ ...interface{}) {} +func (l *recordingPoolLogger) Debug(_ string, _ ...interface{}) {} + +func (l *recordingPoolLogger) Warn(msg string, _ ...interface{}) { + l.mu.Lock() + defer l.mu.Unlock() + l.warns = append(l.warns, msg) +} + +func (l *recordingPoolLogger) warnCount() int { + l.mu.Lock() + defer l.mu.Unlock() + return len(l.warns) +} + +func TestSandboxPoolManager_Destroy_LogsBestEffortKillFailures(t *testing.T) { + ctx := context.Background() + store := NewInMemoryPoolStateStore() + rec := newKillRecorder(t) + rec.fail.Store(true) + seedIdle(t, store, "test-pool", 2) + + logger := &recordingPoolLogger{} + manager, err := NewSandboxPoolManagerBuilder(). + StateStore(store). + ConnectionConfig(ConnectionConfig{Domain: rec.srv.URL, Protocol: "http"}). + PoolLogger(logger). + Build() + if err != nil { + t.Fatalf("Build failed: %v", err) + } + + result, err := manager.Destroy(ctx, "test-pool", PoolDestroyOptions{}) + if err != nil { + t.Fatalf("Destroy failed: %v", err) + } + if result.State != PoolDestroyStateDestroyed { + t.Errorf("state = %s, want DESTROYED (kill failures must not abort the destroy)", result.State) + } + if got := logger.warnCount(); got != 2 { + t.Errorf("warn count = %d, want 2 (one per failed kill)", got) + } +} + +func TestSandboxPoolManager_Destroy_IsIdempotent(t *testing.T) { + ctx := context.Background() + store := NewInMemoryPoolStateStore() + rec := newKillRecorder(t) + seedIdle(t, store, "test-pool", 2) + + manager := newTestPoolManager(t, store, rec.srv.URL) + if _, err := manager.Destroy(ctx, "test-pool", PoolDestroyOptions{}); err != nil { + t.Fatalf("first Destroy failed: %v", err) + } + + result, err := manager.Destroy(ctx, "test-pool", PoolDestroyOptions{}) + if err != nil { + t.Fatalf("second Destroy failed: %v", err) + } + if result.State != PoolDestroyStateDestroyed { + t.Errorf("state = %s, want DESTROYED", result.State) + } + if result.PersistentStateCleared { + t.Error("PersistentStateCleared = true on a repeat destroy, want false") + } + if result.DrainedIdleCount != 0 || result.KilledIdleCount != 0 { + t.Errorf("repeat destroy drained/killed = %d/%d, want 0/0", result.DrainedIdleCount, result.KilledIdleCount) + } + if got := rec.deleted.Load(); got != 2 { + t.Errorf("DELETE requests = %d, want 2 (the repeat destroy must not kill again)", got) + } +} + +func TestSandboxPoolManager_Destroy_DrainTimeoutLeavesFence(t *testing.T) { + ctx := context.Background() + store := NewInMemoryPoolStateStore() + rec := newKillRecorder(t) + rec.delay.Store(int64(30 * time.Millisecond)) + seedIdle(t, store, "test-pool", 5) + + manager := newTestPoolManager(t, store, rec.srv.URL) + _, err := manager.Destroy(ctx, "test-pool", PoolDestroyOptions{ + DrainTimeout: durationPtr(10 * time.Millisecond), + }) + + var incomplete *PoolDestroyIncompleteError + if !errors.As(err, &incomplete) { + t.Fatalf("Destroy error = %v, want *PoolDestroyIncompleteError", err) + } + if incomplete.PoolName != "test-pool" { + t.Errorf("PoolName = %q, want %q", incomplete.PoolName, "test-pool") + } + + // The namespace stays fenced so a retry can finish the job. + state, err := store.GetDestroyState(ctx, "test-pool") + if err != nil { + t.Fatalf("GetDestroyState failed: %v", err) + } + if state != PoolDestroyStateDestroying { + t.Errorf("state after timeout = %s, want DESTROYING", state) + } + + rec.delay.Store(0) + result, err := manager.Destroy(ctx, "test-pool", PoolDestroyOptions{}) + if err != nil { + t.Fatalf("retry Destroy failed: %v", err) + } + if result.State != PoolDestroyStateDestroyed { + t.Errorf("state after retry = %s, want DESTROYED", result.State) + } +} + +func TestSandboxPoolManager_Destroy_TombstoneTTLExpires(t *testing.T) { + ctx := context.Background() + store := NewInMemoryPoolStateStore() + rec := newKillRecorder(t) + + manager := newTestPoolManager(t, store, rec.srv.URL) + if _, err := manager.Destroy(ctx, "test-pool", PoolDestroyOptions{ + TombstoneTTL: durationPtr(20 * time.Millisecond), + }); err != nil { + t.Fatalf("Destroy failed: %v", err) + } + + if err := store.PutIdle(ctx, "test-pool", "sbx-blocked"); err == nil { + t.Fatal("PutIdle succeeded while the tombstone was live, want *PoolDestroyedError") + } + + time.Sleep(40 * time.Millisecond) + + state, err := store.GetDestroyState(ctx, "test-pool") + if err != nil { + t.Fatalf("GetDestroyState failed: %v", err) + } + if state != PoolDestroyStateActive { + t.Errorf("state after tombstone TTL = %s, want ACTIVE", state) + } + if err := store.PutIdle(ctx, "test-pool", "sbx-rebound"); err != nil { + t.Errorf("PutIdle after tombstone expiry failed: %v", err) + } +} + +func TestSandboxPoolManager_Destroy_ZeroTombstoneTTLNeverExpires(t *testing.T) { + ctx := context.Background() + store := NewInMemoryPoolStateStore() + rec := newKillRecorder(t) + + manager := newTestPoolManager(t, store, rec.srv.URL) + if _, err := manager.Destroy(ctx, "test-pool", PoolDestroyOptions{ + TombstoneTTL: durationPtr(0), + }); err != nil { + t.Fatalf("Destroy failed: %v", err) + } + + time.Sleep(20 * time.Millisecond) + + state, err := store.GetDestroyState(ctx, "test-pool") + if err != nil { + t.Fatalf("GetDestroyState failed: %v", err) + } + if state != PoolDestroyStateDestroyed { + t.Errorf("state = %s, want DESTROYED (a zero TTL must never expire)", state) + } +} + +func TestSandboxPoolManager_Destroy_InvalidOptions(t *testing.T) { + tests := []struct { + name string + poolName string + options PoolDestroyOptions + }{ + {name: "blank pool name", poolName: " ", options: PoolDestroyOptions{}}, + {name: "unsupported strategy", poolName: "test-pool", options: PoolDestroyOptions{Strategy: PoolDestroyStrategy(99)}}, + {name: "negative drain timeout", poolName: "test-pool", options: PoolDestroyOptions{DrainTimeout: durationPtr(-time.Second)}}, + {name: "negative tombstone TTL", poolName: "test-pool", options: PoolDestroyOptions{TombstoneTTL: durationPtr(-time.Second)}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store := NewInMemoryPoolStateStore() + rec := newKillRecorder(t) + manager := newTestPoolManager(t, store, rec.srv.URL) + + if _, err := manager.Destroy(context.Background(), tt.poolName, tt.options); err == nil { + t.Fatal("Destroy succeeded, want validation error") + } + + // A rejected destroy must not have fenced anything. + state, err := store.GetDestroyState(context.Background(), "test-pool") + if err != nil { + t.Fatalf("GetDestroyState failed: %v", err) + } + if state != PoolDestroyStateActive { + t.Errorf("state = %s, want ACTIVE", state) + } + }) + } +} + +// ---------- Fence Observation Tests ---------- + +func TestSandboxPoolManager_Destroy_FenceStopsLivePool(t *testing.T) { + ctx := context.Background() + execdSrv := newMockExecdServer(t) + lifecycleSrv := newMockLifecycleServer(t, execdSrv.URL) + store := NewInMemoryPoolStateStore() + + pool := newTestPool(t, lifecycleSrv.URL, func(b *SandboxPoolBuilder) { + b.StateStore(store).MaxIdle(2).ReconcileInterval(10 * time.Millisecond) + }) + if err := pool.Start(ctx); err != nil { + t.Fatalf("Start failed: %v", err) + } + t.Cleanup(func() { _ = pool.Shutdown(context.Background(), false) }) + + waitForIdleCount(t, store, "test-pool", 2) + + manager := newTestPoolManager(t, store, lifecycleSrv.URL) + if _, err := manager.Destroy(ctx, "test-pool", PoolDestroyOptions{}); err != nil { + t.Fatalf("Destroy failed: %v", err) + } + + // The still-running pool must not replenish the destroyed namespace. + for i := 0; i < 5; i++ { + time.Sleep(20 * time.Millisecond) + counters, err := store.SnapshotCounters(ctx, "test-pool") + if err != nil { + t.Fatalf("SnapshotCounters failed: %v", err) + } + if counters.IdleCount != 0 { + t.Fatalf("idle count = %d after destroy, want 0 (fenced pool must not warm up)", counters.IdleCount) + } + } + + // The reconcile tick observes the fence and stops the pool outright. + deadline := time.Now().Add(5 * time.Second) + for { + snapshot, err := pool.Snapshot(ctx) + if err != nil { + t.Fatalf("Snapshot failed: %v", err) + } + if snapshot.LifecycleState == PoolLifecycleStopped { + break + } + if time.Now().After(deadline) { + t.Fatalf("pool state = %s after destroy, want STOPPED", snapshot.LifecycleState) + } + time.Sleep(10 * time.Millisecond) + } + + // A peer starting fresh against the tombstoned namespace must refuse to run. + peer := newTestPool(t, lifecycleSrv.URL, func(b *SandboxPoolBuilder) { + b.StateStore(store).MaxIdle(2) + }) + err := peer.Start(ctx) + var destroyed *PoolDestroyedError + if !errors.As(err, &destroyed) { + t.Fatalf("peer Start error = %v, want *PoolDestroyedError", err) + } +} + +// countingLifecycleServer is a mock lifecycle API that records how many +// sandboxes were created and killed. +type countingLifecycleServer struct { + srv *httptest.Server + created atomic.Int32 + deleted atomic.Int32 +} + +func newCountingLifecycleServer(t *testing.T, execdURL string) *countingLifecycleServer { + t.Helper() + c := &countingLifecycleServer{} + c.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + path := r.URL.Path + switch { + case r.Method == http.MethodPost && path == "/v1/sandboxes": + c.created.Add(1) + jsonResponse(w, http.StatusCreated, SandboxInfo{ + ID: fmt.Sprintf("sbx-created-%d", c.created.Load()), + Status: SandboxStatus{State: StateRunning}, + Entrypoint: []string{"tail", "-f", "/dev/null"}, + CreatedAt: time.Now().UTC(), + }) + case r.Method == http.MethodGet && strings.Contains(path, "/endpoints/"): + jsonResponse(w, http.StatusOK, Endpoint{ + Endpoint: execdURL, + Headers: map[string]string{"X-EXECD-ACCESS-TOKEN": "test-token"}, + }) + case r.Method == http.MethodGet && strings.HasPrefix(path, "/v1/sandboxes/"): + parts := strings.Split(path, "/") + jsonResponse(w, http.StatusOK, SandboxInfo{ + ID: parts[len(parts)-1], + Status: SandboxStatus{State: StateRunning}, + Entrypoint: []string{"tail", "-f", "/dev/null"}, + CreatedAt: time.Now().UTC(), + }) + case r.Method == http.MethodDelete && strings.HasPrefix(path, "/v1/sandboxes/"): + c.deleted.Add(1) + w.WriteHeader(http.StatusNoContent) + case r.Method == http.MethodPost && strings.HasSuffix(path, "/renew-expiration"): + jsonResponse(w, http.StatusOK, RenewExpirationResponse{ExpiresAt: time.Now().Add(time.Hour).UTC()}) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + t.Cleanup(c.srv.Close) + return c +} + +// scriptedDestroyStateStore overrides GetDestroyState so a test can drive the +// fence independently of the rest of the store. +type scriptedDestroyStateStore struct { + *InMemoryPoolStateStore + + // err, when set, is returned from every GetDestroyState call. + err error + // activeCalls is how many leading calls report ACTIVE before the namespace + // starts reporting DESTROYED. Ignored when err is set. + activeCalls int32 + + calls atomic.Int32 +} + +func (s *scriptedDestroyStateStore) GetDestroyState(_ context.Context, _ string) (PoolDestroyState, error) { + n := s.calls.Add(1) + if s.err != nil { + return PoolDestroyStateActive, s.err + } + if n <= s.activeCalls { + return PoolDestroyStateActive, nil + } + return PoolDestroyStateDestroyed, nil +} + +// TestSandboxPoolManager_Destroy_BlocksDirectCreateOnLivePool covers the case a +// store-level fence alone cannot: a peer that is still RUNNING when the fence +// lands would otherwise find an empty idle buffer and mint a fresh sandbox into +// the retired namespace via the direct-create fallthrough. +func TestSandboxPoolManager_Destroy_BlocksDirectCreateOnLivePool(t *testing.T) { + ctx := context.Background() + execdSrv := newMockExecdServer(t) + lifecycle := newCountingLifecycleServer(t, execdSrv.URL) + store := NewInMemoryPoolStateStore() + + // MaxIdle 0 and a long interval keep the pool RUNNING: no reconcile tick + // fires to observe the fence and stop it. + pool := newTestPool(t, lifecycle.srv.URL, func(b *SandboxPoolBuilder) { + b.StateStore(store).MaxIdle(0).ReconcileInterval(time.Hour). + EmptyBehavior(AcquirePolicyDirectCreate) + }) + if err := pool.Start(ctx); err != nil { + t.Fatalf("Start failed: %v", err) + } + t.Cleanup(func() { _ = pool.Shutdown(context.Background(), false) }) + + // Sanity check: before the destroy, direct create is the expected behavior. + sb, err := pool.Acquire(ctx, AcquireOptions{}) + if err != nil { + t.Fatalf("Acquire before destroy failed: %v", err) + } + _ = sb.Close() + if got := lifecycle.created.Load(); got != 1 { + t.Fatalf("created = %d before destroy, want 1", got) + } + + manager := newTestPoolManager(t, store, lifecycle.srv.URL) + if _, err := manager.Destroy(ctx, "test-pool", PoolDestroyOptions{}); err != nil { + t.Fatalf("Destroy failed: %v", err) + } + + snapshot, err := pool.Snapshot(ctx) + if err != nil { + t.Fatalf("Snapshot failed: %v", err) + } + if snapshot.LifecycleState != PoolLifecycleRunning { + t.Fatalf("pool state = %s, want RUNNING (the test needs a live peer)", snapshot.LifecycleState) + } + + _, err = pool.Acquire(ctx, AcquireOptions{}) + var destroyed *PoolDestroyedError + if !errors.As(err, &destroyed) { + t.Fatalf("Acquire after destroy = %v, want *PoolDestroyedError", err) + } + if got := lifecycle.created.Load(); got != 1 { + t.Errorf("created = %d after destroy, want 1 (no sandbox may be minted into a retired namespace)", got) + } +} + +// TestPool_Acquire_KillsSandboxFencedMidCreate covers a destroy that lands while +// a direct create is already in flight. +func TestPool_Acquire_KillsSandboxFencedMidCreate(t *testing.T) { + ctx := context.Background() + execdSrv := newMockExecdServer(t) + lifecycle := newCountingLifecycleServer(t, execdSrv.URL) + + // The namespace is checked at Start and again before the acquire; both must + // see ACTIVE. The third check is the post-create one, and that is the one + // this test wants fenced. + store := &scriptedDestroyStateStore{ + InMemoryPoolStateStore: NewInMemoryPoolStateStore(), + activeCalls: 2, + } + + pool := newTestPool(t, lifecycle.srv.URL, func(b *SandboxPoolBuilder) { + b.StateStore(store).MaxIdle(0).ReconcileInterval(time.Hour) + }) + if err := pool.Start(ctx); err != nil { + t.Fatalf("Start failed: %v", err) + } + t.Cleanup(func() { _ = pool.Shutdown(context.Background(), false) }) + + _, err := pool.Acquire(ctx, AcquireOptions{}) + var destroyed *PoolDestroyedError + if !errors.As(err, &destroyed) { + t.Fatalf("Acquire = %v, want *PoolDestroyedError", err) + } + if got := lifecycle.created.Load(); got != 1 { + t.Fatalf("created = %d, want 1", got) + } + + // The orphaned sandbox is killed asynchronously. + deadline := time.Now().Add(5 * time.Second) + for lifecycle.deleted.Load() == 0 { + if time.Now().After(deadline) { + t.Fatal("sandbox created before the fence was never killed") + } + time.Sleep(10 * time.Millisecond) + } +} + +// TestPool_Acquire_KillsIdleSandboxFencedMidAcquire covers the fence landing +// between the preflight check and the idle take. TryTakeIdle is unfenced so the +// destroy manager can drain, which means the ID is already out of the store by +// then and a concurrent Destroy can no longer reach it: the acquire has to kill +// it rather than hand back a sandbox from a retired namespace. +func TestPool_Acquire_KillsIdleSandboxFencedMidAcquire(t *testing.T) { + ctx := context.Background() + execdSrv := newMockExecdServer(t) + lifecycle := newCountingLifecycleServer(t, execdSrv.URL) + + // Start and the acquire preflight both see ACTIVE; the post-connect check + // is the third call and sees the fence. + store := &scriptedDestroyStateStore{ + InMemoryPoolStateStore: NewInMemoryPoolStateStore(), + activeCalls: 2, + } + + pool := newTestPool(t, lifecycle.srv.URL, func(b *SandboxPoolBuilder) { + b.StateStore(store).MaxIdle(0).ReconcileInterval(time.Hour) + }) + if err := pool.Start(ctx); err != nil { + t.Fatalf("Start failed: %v", err) + } + t.Cleanup(func() { _ = pool.Shutdown(context.Background(), false) }) + + if err := store.PutIdle(ctx, "test-pool", "sbx-idle-fenced"); err != nil { + t.Fatalf("PutIdle failed: %v", err) + } + + sb, err := pool.Acquire(ctx, AcquireOptions{}) + var destroyed *PoolDestroyedError + if !errors.As(err, &destroyed) { + if sb != nil { + _ = sb.Close() + } + t.Fatalf("Acquire = %v, want *PoolDestroyedError", err) + } + if got := lifecycle.created.Load(); got != 0 { + t.Errorf("created = %d, want 0 (the idle candidate must not be replaced)", got) + } + + // The idle sandbox is no longer tracked anywhere, so the acquire must kill it. + deadline := time.Now().Add(5 * time.Second) + for lifecycle.deleted.Load() == 0 { + if time.Now().After(deadline) { + t.Fatal("idle sandbox taken before the fence was never killed") + } + time.Sleep(10 * time.Millisecond) + } +} + +// TestPool_Acquire_IdlePathFenceCheckIsFailClosed pins the deliberate asymmetry +// with the direct-create path: an idle sandbox is already out of the store, so an +// unreachable store cannot be assumed ACTIVE the way direct create may. +func TestPool_Acquire_IdlePathFenceCheckIsFailClosed(t *testing.T) { + ctx := context.Background() + execdSrv := newMockExecdServer(t) + lifecycle := newCountingLifecycleServer(t, execdSrv.URL) + + inner := NewInMemoryPoolStateStore() + if err := inner.PutIdle(ctx, "test-pool", "sbx-idle-outage"); err != nil { + t.Fatalf("PutIdle failed: %v", err) + } + + // Report ACTIVE for Start and the preflight, then fail. DIRECT_CREATE would + // degrade and keep going; the idle path must not. + store := &outageAfterNCallsStore{ + InMemoryPoolStateStore: inner, + okCalls: 2, + } + + pool := newTestPool(t, lifecycle.srv.URL, func(b *SandboxPoolBuilder) { + b.StateStore(store).MaxIdle(0).ReconcileInterval(time.Hour). + EmptyBehavior(AcquirePolicyDirectCreate) + }) + if err := pool.Start(ctx); err != nil { + t.Fatalf("Start failed: %v", err) + } + t.Cleanup(func() { _ = pool.Shutdown(context.Background(), false) }) + + sb, err := pool.Acquire(ctx, AcquireOptions{}) + if err == nil { + _ = sb.Close() + t.Fatal("Acquire succeeded with an unconfirmable namespace, want an error") + } + var unavailable *PoolStateStoreUnavailableError + if !errors.As(err, &unavailable) { + t.Fatalf("Acquire = %v, want *PoolStateStoreUnavailableError", err) + } + + deadline := time.Now().Add(5 * time.Second) + for lifecycle.deleted.Load() == 0 { + if time.Now().After(deadline) { + t.Fatal("idle sandbox was never killed after the fail-closed check") + } + time.Sleep(10 * time.Millisecond) + } +} + +// outageAfterNCallsStore answers GetDestroyState normally for the first okCalls +// calls and then reports the store as unreachable. +type outageAfterNCallsStore struct { + *InMemoryPoolStateStore + + okCalls int32 + calls atomic.Int32 +} + +func (s *outageAfterNCallsStore) GetDestroyState(ctx context.Context, poolName string) (PoolDestroyState, error) { + if s.calls.Add(1) <= s.okCalls { + return s.InMemoryPoolStateStore.GetDestroyState(ctx, poolName) + } + return PoolDestroyStateActive, &PoolStateStoreUnavailableError{ + Operation: "GetDestroyState", + Cause: errors.New("redis is down"), + } +} + +// TestPool_Acquire_NamespaceCheckDegradesOnStoreOutage keeps a store outage from +// making direct-create policies less available than the OSEP-0005 matrix +// documents, while fail-closed policies still surface it. +func TestPool_Acquire_NamespaceCheckDegradesOnStoreOutage(t *testing.T) { + tests := []struct { + name string + policy AcquirePolicy + wantCreated int32 + wantErr bool + }{ + {name: "direct create degrades", policy: AcquirePolicyDirectCreate, wantCreated: 1}, + {name: "retry then create degrades", policy: AcquirePolicyRetryNextIdleThenCreate, wantCreated: 1}, + {name: "fail fast surfaces the outage", policy: AcquirePolicyFailFast, wantErr: true}, + {name: "retry next idle surfaces the outage", policy: AcquirePolicyRetryNextIdle, wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + execdSrv := newMockExecdServer(t) + lifecycle := newCountingLifecycleServer(t, execdSrv.URL) + store := &scriptedDestroyStateStore{ + InMemoryPoolStateStore: NewInMemoryPoolStateStore(), + err: errors.New("redis is down"), + } + + pool := newTestPool(t, lifecycle.srv.URL, func(b *SandboxPoolBuilder) { + b.StateStore(store).MaxIdle(0).ReconcileInterval(time.Hour). + EmptyBehavior(tt.policy) + }) + if err := pool.Start(ctx); err != nil { + t.Fatalf("Start failed: %v", err) + } + t.Cleanup(func() { _ = pool.Shutdown(context.Background(), false) }) + + sb, err := pool.Acquire(ctx, AcquireOptions{}) + if tt.wantErr { + var unavailable *PoolStateStoreUnavailableError + if !errors.As(err, &unavailable) { + t.Fatalf("Acquire = %v, want *PoolStateStoreUnavailableError", err) + } + } else { + if err != nil { + t.Fatalf("Acquire failed: %v", err) + } + _ = sb.Close() + } + if got := lifecycle.created.Load(); got != tt.wantCreated { + t.Errorf("created = %d, want %d", got, tt.wantCreated) + } + }) + } +} + +func TestPool_Start_RefusesDestroyedNamespace(t *testing.T) { + ctx := context.Background() + execdSrv := newMockExecdServer(t) + lifecycleSrv := newMockLifecycleServer(t, execdSrv.URL) + store := &scriptedDestroyStateStore{ + InMemoryPoolStateStore: NewInMemoryPoolStateStore(), + } + + pool := newTestPool(t, lifecycleSrv.URL, func(b *SandboxPoolBuilder) { + b.StateStore(store).MaxIdle(0).ReconcileInterval(time.Hour) + }) + + err := pool.Start(ctx) + var destroyed *PoolDestroyedError + if !errors.As(err, &destroyed) { + t.Fatalf("Start = %v, want *PoolDestroyedError", err) + } + + snapshot, err := pool.Snapshot(ctx) + if err != nil { + t.Fatalf("Snapshot failed: %v", err) + } + if snapshot.LifecycleState != PoolLifecycleNotStarted { + t.Errorf("state after refused Start = %s, want NOT_STARTED", snapshot.LifecycleState) + } +} + +func TestInMemoryPoolStateStore_FenceRejectsWrites(t *testing.T) { + ctx := context.Background() + + for _, state := range []PoolDestroyState{PoolDestroyStateDestroying, PoolDestroyStateDestroyed} { + t.Run(state.String(), func(t *testing.T) { + store := NewInMemoryPoolStateStore() + if err := store.BeginDestroy(ctx, "test-pool", "owner-1"); err != nil { + t.Fatalf("BeginDestroy failed: %v", err) + } + if state == PoolDestroyStateDestroyed { + if err := store.MarkDestroyed(ctx, "test-pool", "owner-1", time.Hour); err != nil { + t.Fatalf("MarkDestroyed failed: %v", err) + } + } + + writes := map[string]func() error{ + "PutIdle": func() error { return store.PutIdle(ctx, "test-pool", "sbx-1") }, + "SetMaxIdle": func() error { return store.SetMaxIdle(ctx, "test-pool", 5) }, + "SetIdleEntryTTL": func() error { return store.SetIdleEntryTTL(ctx, "test-pool", time.Minute) }, + } + for name, write := range writes { + var destroyed *PoolDestroyedError + if err := write(); !errors.As(err, &destroyed) { + t.Errorf("%s error = %v, want *PoolDestroyedError", name, err) + } else if destroyed.State != state { + t.Errorf("%s error state = %s, want %s", name, destroyed.State, state) + } + } + + acquired, err := store.TryAcquirePrimaryLock(ctx, "test-pool", "owner-2", time.Minute) + if err != nil { + t.Fatalf("TryAcquirePrimaryLock failed: %v", err) + } + if acquired { + t.Error("TryAcquirePrimaryLock succeeded on a fenced namespace, want false") + } + + renewed, err := store.RenewPrimaryLock(ctx, "test-pool", "owner-2", time.Minute) + if err != nil { + t.Fatalf("RenewPrimaryLock failed: %v", err) + } + if renewed { + t.Error("RenewPrimaryLock succeeded on a fenced namespace, want false") + } + }) + } +} + +func TestInMemoryPoolStateStore_BeginDestroyRejectsTombstoned(t *testing.T) { + ctx := context.Background() + store := NewInMemoryPoolStateStore() + + if err := store.BeginDestroy(ctx, "test-pool", "owner-1"); err != nil { + t.Fatalf("BeginDestroy failed: %v", err) + } + // Re-entrant while DESTROYING, so a retrying owner can make progress. + if err := store.BeginDestroy(ctx, "test-pool", "owner-1"); err != nil { + t.Fatalf("second BeginDestroy on a DESTROYING namespace failed: %v", err) + } + + if err := store.MarkDestroyed(ctx, "test-pool", "owner-1", time.Hour); err != nil { + t.Fatalf("MarkDestroyed failed: %v", err) + } + + var destroyed *PoolDestroyedError + if err := store.BeginDestroy(ctx, "test-pool", "owner-2"); !errors.As(err, &destroyed) { + t.Fatalf("BeginDestroy on a tombstoned namespace = %v, want *PoolDestroyedError", err) + } +} + +func TestInMemoryPoolStateStore_ClearPoolStateKeepsFence(t *testing.T) { + ctx := context.Background() + store := NewInMemoryPoolStateStore() + + if err := store.SetMaxIdle(ctx, "test-pool", 7); err != nil { + t.Fatalf("SetMaxIdle failed: %v", err) + } + if err := store.PutIdle(ctx, "test-pool", "sbx-1"); err != nil { + t.Fatalf("PutIdle failed: %v", err) + } + if err := store.BeginDestroy(ctx, "test-pool", "owner-1"); err != nil { + t.Fatalf("BeginDestroy failed: %v", err) + } + if err := store.ClearPoolState(ctx, "test-pool"); err != nil { + t.Fatalf("ClearPoolState failed: %v", err) + } + + counters, err := store.SnapshotCounters(ctx, "test-pool") + if err != nil { + t.Fatalf("SnapshotCounters failed: %v", err) + } + if counters.IdleCount != 0 { + t.Errorf("idle count = %d, want 0", counters.IdleCount) + } + maxIdle, err := store.GetMaxIdle(ctx, "test-pool") + if err != nil { + t.Fatalf("GetMaxIdle failed: %v", err) + } + if maxIdle != 0 { + t.Errorf("maxIdle = %d, want 0", maxIdle) + } + state, err := store.GetDestroyState(ctx, "test-pool") + if err != nil { + t.Fatalf("GetDestroyState failed: %v", err) + } + if state != PoolDestroyStateDestroying { + t.Errorf("state = %s, want DESTROYING (ClearPoolState must not lift the fence)", state) + } +} + +func TestInMemoryPoolStateStore_MarkDestroyedRejectsBlankOwnerAndNegativeTTL(t *testing.T) { + ctx := context.Background() + store := NewInMemoryPoolStateStore() + + if err := store.MarkDestroyed(ctx, "test-pool", "", time.Hour); err == nil { + t.Error("MarkDestroyed with a blank owner succeeded, want error") + } + if err := store.MarkDestroyed(ctx, "test-pool", "owner-1", -time.Second); err == nil { + t.Error("MarkDestroyed with a negative TTL succeeded, want error") + } + if err := store.BeginDestroy(ctx, "test-pool", ""); err == nil { + t.Error("BeginDestroy with a blank owner succeeded, want error") + } +} + +// ---------- Builder Tests ---------- + +func TestSandboxPoolManagerBuilder_Validation(t *testing.T) { + tests := []struct { + name string + build func() (*SandboxPoolManager, error) + wantErr bool + }{ + { + name: "missing state store", + build: func() (*SandboxPoolManager, error) { + return NewSandboxPoolManagerBuilder(). + ConnectionConfig(ConnectionConfig{Domain: "localhost:8080"}). + Build() + }, + wantErr: true, + }, + { + name: "missing connection config", + build: func() (*SandboxPoolManager, error) { + return NewSandboxPoolManagerBuilder(). + StateStore(NewInMemoryPoolStateStore()). + Build() + }, + wantErr: true, + }, + { + name: "blank owner ID", + build: func() (*SandboxPoolManager, error) { + return NewSandboxPoolManagerBuilder(). + StateStore(NewInMemoryPoolStateStore()). + ConnectionConfig(ConnectionConfig{Domain: "localhost:8080"}). + OwnerID(" "). + Build() + }, + wantErr: true, + }, + { + name: "defaults the owner ID", + build: func() (*SandboxPoolManager, error) { + return NewSandboxPoolManagerBuilder(). + StateStore(NewInMemoryPoolStateStore()). + ConnectionConfig(ConnectionConfig{Domain: "localhost:8080"}). + Build() + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + manager, err := tt.build() + if tt.wantErr { + if err == nil { + t.Fatal("Build succeeded, want error") + } + return + } + if err != nil { + t.Fatalf("Build failed: %v", err) + } + if manager.ownerID == "" { + t.Error("ownerID is empty, want a generated value") + } + }) + } +} + +// waitForIdleCount blocks until the store reports want idle entries. +func waitForIdleCount(t *testing.T, store PoolStateStore, poolName string, want int) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + counters, err := store.SnapshotCounters(context.Background(), poolName) + if err != nil { + t.Fatalf("SnapshotCounters failed: %v", err) + } + if counters.IdleCount >= want { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("timed out waiting for %d idle entries in pool %q", want, poolName) +} diff --git a/sdks/sandbox/go/pool_store.go b/sdks/sandbox/go/pool_store.go index e45f994ca..6a7ae44bf 100644 --- a/sdks/sandbox/go/pool_store.go +++ b/sdks/sandbox/go/pool_store.go @@ -71,4 +71,23 @@ type PoolStateStore interface { // SetIdleEntryTTL persists the idle entry TTL for the pool. SetIdleEntryTTL(ctx context.Context, poolName string, ttl time.Duration) error + + // GetDestroyState returns the destroy state of the pool namespace. + // An expired tombstone reads back as ACTIVE. + GetDestroyState(ctx context.Context, poolName string) (PoolDestroyState, error) + + // BeginDestroy writes the DESTROYING fence, making the namespace + // unwritable for every peer sharing this store. Returns *PoolDestroyedError + // if the namespace is already tombstoned. Re-entrant while DESTROYING. + BeginDestroy(ctx context.Context, poolName string, ownerID string) error + + // ClearPoolState wipes the pool's coordination state: idle entries, the + // primary lock, maxIdle, and the idle entry TTL. The destroy state itself + // is left in place. + ClearPoolState(ctx context.Context, poolName string) error + + // MarkDestroyed replaces the fence with a DESTROYED tombstone so later + // callers cannot silently rebind the namespace. A zero tombstoneTTL writes + // a tombstone that never expires; it must not be negative. + MarkDestroyed(ctx context.Context, poolName string, ownerID string, tombstoneTTL time.Duration) error } diff --git a/sdks/sandbox/go/pool_store_memory.go b/sdks/sandbox/go/pool_store_memory.go index 4557374a7..86661449b 100644 --- a/sdks/sandbox/go/pool_store_memory.go +++ b/sdks/sandbox/go/pool_store_memory.go @@ -42,6 +42,12 @@ type poolState struct { // configurable per-pool settings. idleTTL time.Duration maxIdle int + + // destroy fence / tombstone state. + destroyState PoolDestroyState + destroyOwnerID string + // destroyExpiresAt is zero when the current destroy state never expires. + destroyExpiresAt time.Time } // InMemoryPoolStateStore is a pure in-memory implementation of PoolStateStore. @@ -165,6 +171,9 @@ func (s *InMemoryPoolStateStore) PutIdle(_ context.Context, poolName string, san defer ps.mu.Unlock() now := time.Now() + if err := ps.rejectIfFencedLocked(poolName, now); err != nil { + return err + } if existing, exists := ps.idleMap[sandboxID]; exists { if existing.ExpiresAt.IsZero() || now.Before(existing.ExpiresAt) { return nil // still alive, idempotent no-op @@ -204,6 +213,9 @@ func (s *InMemoryPoolStateStore) TryAcquirePrimaryLock(_ context.Context, poolNa defer ps.mu.Unlock() now := time.Now() + if ps.destroyStateLocked(now) != PoolDestroyStateActive { + return false, nil + } if ps.lock.ownerID != "" && now.Before(ps.lock.expiresAt) { // Lock is held and not expired. if ps.lock.ownerID == ownerID { @@ -228,6 +240,9 @@ func (s *InMemoryPoolStateStore) RenewPrimaryLock(_ context.Context, poolName st defer ps.mu.Unlock() now := time.Now() + if ps.destroyStateLocked(now) != PoolDestroyStateActive { + return false, nil + } if ps.lock.ownerID != ownerID { return false, nil } @@ -352,6 +367,9 @@ func (s *InMemoryPoolStateStore) SetMaxIdle(_ context.Context, poolName string, ps := s.getOrCreatePool(poolName) ps.mu.Lock() defer ps.mu.Unlock() + if err := ps.rejectIfFencedLocked(poolName, time.Now()); err != nil { + return err + } ps.maxIdle = maxIdle return nil } @@ -364,10 +382,100 @@ func (s *InMemoryPoolStateStore) SetIdleEntryTTL(_ context.Context, poolName str ps := s.getOrCreatePool(poolName) ps.mu.Lock() defer ps.mu.Unlock() + if err := ps.rejectIfFencedLocked(poolName, time.Now()); err != nil { + return err + } ps.idleTTL = ttl return nil } +// GetDestroyState returns the destroy state of the pool namespace. +func (s *InMemoryPoolStateStore) GetDestroyState(_ context.Context, poolName string) (PoolDestroyState, error) { + ps := s.getOrCreatePool(poolName) + ps.mu.Lock() + defer ps.mu.Unlock() + return ps.destroyStateLocked(time.Now()), nil +} + +// BeginDestroy writes the DESTROYING fence. Returns *PoolDestroyedError if the +// namespace already carries a live tombstone. +func (s *InMemoryPoolStateStore) BeginDestroy(_ context.Context, poolName string, ownerID string) error { + if ownerID == "" { + return fmt.Errorf("opensandbox: ownerID must not be blank") + } + ps := s.getOrCreatePool(poolName) + ps.mu.Lock() + defer ps.mu.Unlock() + + if ps.destroyStateLocked(time.Now()) == PoolDestroyStateDestroyed { + return &PoolDestroyedError{PoolName: poolName, State: PoolDestroyStateDestroyed} + } + ps.destroyState = PoolDestroyStateDestroying + ps.destroyOwnerID = ownerID + ps.destroyExpiresAt = time.Time{} + return nil +} + +// ClearPoolState wipes the pool's coordination state, leaving the destroy state +// in place so the fence survives the cleanup. +func (s *InMemoryPoolStateStore) ClearPoolState(_ context.Context, poolName string) error { + ps := s.getOrCreatePool(poolName) + ps.mu.Lock() + defer ps.mu.Unlock() + + ps.idleMap = make(map[string]*IdleEntry) + ps.idleQueue = nil + ps.lock = poolLock{} + ps.idleTTL = DefaultIdleTimeout + ps.maxIdle = 0 + return nil +} + +// MarkDestroyed writes the DESTROYED tombstone. A zero tombstoneTTL never expires. +func (s *InMemoryPoolStateStore) MarkDestroyed(_ context.Context, poolName string, ownerID string, tombstoneTTL time.Duration) error { + if ownerID == "" { + return fmt.Errorf("opensandbox: ownerID must not be blank") + } + if tombstoneTTL < 0 { + return fmt.Errorf("opensandbox: tombstoneTTL must not be negative, got %v", tombstoneTTL) + } + ps := s.getOrCreatePool(poolName) + ps.mu.Lock() + defer ps.mu.Unlock() + + ps.destroyState = PoolDestroyStateDestroyed + ps.destroyOwnerID = ownerID + if tombstoneTTL == 0 { + ps.destroyExpiresAt = time.Time{} + } else { + ps.destroyExpiresAt = time.Now().Add(tombstoneTTL) + } + return nil +} + +// destroyStateLocked returns the current destroy state, clearing an expired +// tombstone on the way. Must be called with ps.mu held. +func (ps *poolState) destroyStateLocked(now time.Time) PoolDestroyState { + if ps.destroyState == PoolDestroyStateActive { + return PoolDestroyStateActive + } + if !ps.destroyExpiresAt.IsZero() && !now.Before(ps.destroyExpiresAt) { + ps.destroyState = PoolDestroyStateActive + ps.destroyOwnerID = "" + ps.destroyExpiresAt = time.Time{} + } + return ps.destroyState +} + +// rejectIfFencedLocked returns *PoolDestroyedError when the namespace is fenced. +// Must be called with ps.mu held. +func (ps *poolState) rejectIfFencedLocked(poolName string, now time.Time) error { + if state := ps.destroyStateLocked(now); state != PoolDestroyStateActive { + return &PoolDestroyedError{PoolName: poolName, State: state} + } + return nil +} + // compactQueueIfNeeded copies the queue to a right-sized slice when the // underlying array has grown much larger than needed. Must be called with // ps.mu held. diff --git a/sdks/sandbox/go/pool_types.go b/sdks/sandbox/go/pool_types.go index 7065be58a..ba6b57a88 100644 --- a/sdks/sandbox/go/pool_types.go +++ b/sdks/sandbox/go/pool_types.go @@ -258,3 +258,89 @@ type AcquireOptions struct { // DefaultIdleTimeout is the default TTL for idle pool entries (24 hours, per OSEP-0005). const DefaultIdleTimeout = 24 * time.Hour + +// PoolDestroyState represents the destroy lifecycle of a pool namespace as seen +// by every process sharing the same state store. +// +// - ACTIVE: the namespace is usable. +// - DESTROYING: a destroy fence is in place. Peer pools must stop replenishing +// and must not fall back to direct create. +// - DESTROYED: a tombstone is in place. Callers must not rebind the namespace +// until the tombstone expires. +type PoolDestroyState int + +const ( + PoolDestroyStateActive PoolDestroyState = iota + PoolDestroyStateDestroying + PoolDestroyStateDestroyed +) + +func (s PoolDestroyState) String() string { + switch s { + case PoolDestroyStateActive: + return "ACTIVE" + case PoolDestroyStateDestroying: + return "DESTROYING" + case PoolDestroyStateDestroyed: + return "DESTROYED" + default: + return "UNKNOWN" + } +} + +// PoolDestroyStrategy selects how a namespace is retired. Only FORCE is +// implemented; the iota order MUST stay append-only. +type PoolDestroyStrategy int + +const ( + PoolDestroyForce PoolDestroyStrategy = iota +) + +func (s PoolDestroyStrategy) String() string { + switch s { + case PoolDestroyForce: + return "FORCE" + default: + return "UNKNOWN" + } +} + +// DefaultPoolDrainTimeout bounds the idle-drain phase of a pool destroy. +const DefaultPoolDrainTimeout = 30 * time.Second + +// DefaultPoolTombstoneTTL is how long a DESTROYED tombstone survives before the +// namespace may be rebound. +const DefaultPoolTombstoneTTL = 7 * 24 * time.Hour + +// PoolDestroyOptions configures a single SandboxPoolManager.Destroy call. +// The zero value is valid and selects FORCE with all defaults. +type PoolDestroyOptions struct { + // Strategy selects the destroy algorithm. Only PoolDestroyForce is supported. + Strategy PoolDestroyStrategy + + // DrainTimeout bounds the idle-drain loop. Nil selects DefaultPoolDrainTimeout; + // an explicit zero drains without a deadline. Must not be negative. + DrainTimeout *time.Duration + + // TombstoneTTL is how long the DESTROYED tombstone survives. Nil selects + // DefaultPoolTombstoneTTL; an explicit zero writes a tombstone that never + // expires. Must not be negative. + TombstoneTTL *time.Duration +} + +// PoolDestroyResult reports what a destroy actually did. +type PoolDestroyResult struct { + PoolName string + State PoolDestroyState + + // DrainedIdleCount is how many idle entries were taken from the store. + DrainedIdleCount int + + // KilledIdleCount is how many of those sandboxes were successfully killed. + // Killing is best-effort, so this may be lower than DrainedIdleCount. + KilledIdleCount int + + // PersistentStateCleared reports whether this call cleared the coordination + // state. It is false when the namespace was already tombstoned. + PersistentStateCleared bool +} diff --git a/sdks/sandbox/go/poolredis/store.go b/sdks/sandbox/go/poolredis/store.go index bb29b21da..57a9ad647 100644 --- a/sdks/sandbox/go/poolredis/store.go +++ b/sdks/sandbox/go/poolredis/store.go @@ -56,6 +56,10 @@ end `) putIdleScript = redis.NewScript(` +local destroy_state = redis.call('GET', KEYS[3]) +if destroy_state then + return -1 +end local redis_time = redis.call('TIME') local now_ms = tonumber(redis_time[1]) * 1000 + math.floor(tonumber(redis_time[2]) / 1000) local expires_at = now_ms + tonumber(ARGV[2]) @@ -94,6 +98,10 @@ return discarded_alive `) acquireLockScript = redis.NewScript(` +local destroy_state = redis.call('GET', KEYS[2]) +if destroy_state then + return 0 +end local current = redis.call('GET', KEYS[1]) if not current then redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2]) @@ -106,6 +114,10 @@ return 0 `) renewLockScript = redis.NewScript(` +local destroy_state = redis.call('GET', KEYS[2]) +if destroy_state then + return 0 +end if redis.call('GET', KEYS[1]) == ARGV[1] then redis.call('PEXPIRE', KEYS[1], ARGV[2]) return 1 @@ -138,6 +150,39 @@ for i = 1, #entries, 2 do end end return count +`) + + // setFencedValueScript writes a single pool setting, refusing the write when + // the namespace carries a destroy fence or tombstone. + setFencedValueScript = redis.NewScript(` +local destroy_state = redis.call('GET', KEYS[2]) +if destroy_state then + return -1 +end +redis.call('SET', KEYS[1], ARGV[1]) +return 1 +`) + + beginDestroyScript = redis.NewScript(` +local destroy_state = redis.call('GET', KEYS[1]) +if destroy_state == ARGV[2] then + return -1 +end +redis.call('SET', KEYS[1], ARGV[1]) +redis.call('SET', KEYS[2], ARGV[3]) +return 1 +`) + + markDestroyedScript = redis.NewScript(` +local ttl_ms = tonumber(ARGV[3]) +if ttl_ms and ttl_ms > 0 then + redis.call('SET', KEYS[1], ARGV[1], 'PX', ttl_ms) + redis.call('SET', KEYS[2], ARGV[2], 'PX', ttl_ms) +else + redis.call('SET', KEYS[1], ARGV[1]) + redis.call('SET', KEYS[2], ARGV[2]) +end +return 1 `) ) @@ -257,14 +302,14 @@ func (s *RedisPoolStateStore) PutIdle(ctx context.Context, poolName string, sand return err } - keys := []string{s.idleListKey(poolName), s.idleExpiresKey(poolName)} + keys := []string{s.idleListKey(poolName), s.idleExpiresKey(poolName), s.destroyStateKey(poolName)} argv := []interface{}{sandboxID, strconv.FormatInt(idleTTLMs, 10)} - _, err = putIdleScript.Run(ctx, s.client, keys, argv...).Result() + result, err := putIdleScript.Run(ctx, s.client, keys, argv...).Int64() if err != nil && err != redis.Nil { return &opensandbox.PoolStateStoreUnavailableError{Operation: "PutIdle", Cause: err} } - return nil + return s.destroyedErrorIfFenced(ctx, poolName, result) } // RemoveIdle atomically removes a sandbox from the idle pool. Idempotent. @@ -289,7 +334,7 @@ func (s *RedisPoolStateStore) TryAcquirePrimaryLock(ctx context.Context, poolNam ttlMs = 1 } result, err := acquireLockScript.Run(ctx, s.client, - []string{s.PrimaryLockKey(poolName)}, + []string{s.PrimaryLockKey(poolName), s.destroyStateKey(poolName)}, ownerID, strconv.FormatInt(ttlMs, 10)).Int64() if err != nil && err != redis.Nil { return false, &opensandbox.PoolStateStoreUnavailableError{Operation: "TryAcquirePrimaryLock", Cause: err} @@ -304,7 +349,7 @@ func (s *RedisPoolStateStore) RenewPrimaryLock(ctx context.Context, poolName str ttlMs = 1 } - keys := []string{s.PrimaryLockKey(poolName)} + keys := []string{s.PrimaryLockKey(poolName), s.destroyStateKey(poolName)} argv := []interface{}{ownerID, strconv.FormatInt(ttlMs, 10)} result, err := renewLockScript.Run(ctx, s.client, keys, argv...).Int64() @@ -433,11 +478,7 @@ func (s *RedisPoolStateStore) GetMaxIdle(ctx context.Context, poolName string) ( // SetMaxIdle persists the maxIdle value for the pool. func (s *RedisPoolStateStore) SetMaxIdle(ctx context.Context, poolName string, maxIdle int) error { - err := s.client.Set(ctx, s.maxIdleKey(poolName), strconv.Itoa(maxIdle), 0).Err() - if err != nil { - return &opensandbox.PoolStateStoreUnavailableError{Operation: "SetMaxIdle", Cause: err} - } - return nil + return s.setFencedValue(ctx, poolName, "SetMaxIdle", s.maxIdleKey(poolName), strconv.Itoa(maxIdle)) } // SetIdleEntryTTL persists the idle entry TTL for the pool. @@ -446,13 +487,123 @@ func (s *RedisPoolStateStore) SetIdleEntryTTL(ctx context.Context, poolName stri if ms < 1 { ms = 1 } - err := s.client.Set(ctx, s.idleTTLKey(poolName), strconv.FormatInt(ms, 10), 0).Err() + return s.setFencedValue(ctx, poolName, "SetIdleEntryTTL", s.idleTTLKey(poolName), strconv.FormatInt(ms, 10)) +} + +func (s *RedisPoolStateStore) setFencedValue(ctx context.Context, poolName string, operation string, key string, value string) error { + keys := []string{key, s.destroyStateKey(poolName)} + + result, err := setFencedValueScript.Run(ctx, s.client, keys, value).Int64() + if err != nil && err != redis.Nil { + return &opensandbox.PoolStateStoreUnavailableError{Operation: operation, Cause: err} + } + return s.destroyedErrorIfFenced(ctx, poolName, result) +} + +// GetDestroyState returns the destroy state of the pool namespace. An expired +// tombstone has already been dropped by Redis and reads back as ACTIVE. +func (s *RedisPoolStateStore) GetDestroyState(ctx context.Context, poolName string) (opensandbox.PoolDestroyState, error) { + val, err := s.client.Get(ctx, s.destroyStateKey(poolName)).Result() + if err == redis.Nil { + return opensandbox.PoolDestroyStateActive, nil + } if err != nil { - return &opensandbox.PoolStateStoreUnavailableError{Operation: "SetIdleEntryTTL", Cause: err} + return opensandbox.PoolDestroyStateActive, &opensandbox.PoolStateStoreUnavailableError{Operation: "GetDestroyState", Cause: err} + } + switch val { + case opensandbox.PoolDestroyStateDestroying.String(): + return opensandbox.PoolDestroyStateDestroying, nil + case opensandbox.PoolDestroyStateDestroyed.String(): + return opensandbox.PoolDestroyStateDestroyed, nil + default: + return opensandbox.PoolDestroyStateActive, nil + } +} + +// BeginDestroy writes the DESTROYING fence. Returns *opensandbox.PoolDestroyedError +// if the namespace already carries a live tombstone. +func (s *RedisPoolStateStore) BeginDestroy(ctx context.Context, poolName string, ownerID string) error { + if ownerID == "" { + return fmt.Errorf("opensandbox: ownerID must not be blank") + } + keys := []string{s.destroyStateKey(poolName), s.destroyOwnerKey(poolName)} + argv := []interface{}{ + opensandbox.PoolDestroyStateDestroying.String(), + opensandbox.PoolDestroyStateDestroyed.String(), + ownerID, + } + + result, err := beginDestroyScript.Run(ctx, s.client, keys, argv...).Int64() + if err != nil && err != redis.Nil { + return &opensandbox.PoolStateStoreUnavailableError{Operation: "BeginDestroy", Cause: err} + } + if result == -1 { + return &opensandbox.PoolDestroyedError{PoolName: poolName, State: opensandbox.PoolDestroyStateDestroyed} + } + return nil +} + +// ClearPoolState deletes the pool's coordination keys, leaving the destroy keys +// in place so the fence survives the cleanup. +func (s *RedisPoolStateStore) ClearPoolState(ctx context.Context, poolName string) error { + err := s.client.Del(ctx, + s.idleListKey(poolName), + s.idleExpiresKey(poolName), + s.PrimaryLockKey(poolName), + s.maxIdleKey(poolName), + s.idleTTLKey(poolName), + ).Err() + if err != nil && err != redis.Nil { + return &opensandbox.PoolStateStoreUnavailableError{Operation: "ClearPoolState", Cause: err} + } + return nil +} + +// MarkDestroyed writes the DESTROYED tombstone. A zero tombstoneTTL writes a +// tombstone that never expires. +func (s *RedisPoolStateStore) MarkDestroyed(ctx context.Context, poolName string, ownerID string, tombstoneTTL time.Duration) error { + if ownerID == "" { + return fmt.Errorf("opensandbox: ownerID must not be blank") + } + if tombstoneTTL < 0 { + return fmt.Errorf("opensandbox: tombstoneTTL must not be negative, got %v", tombstoneTTL) + } + + // A sub-millisecond TTL would round to zero and be read as "never expires", + // so clamp it to the smallest expiry Redis can represent. + ttlMs := tombstoneTTL.Milliseconds() + if tombstoneTTL > 0 && ttlMs < 1 { + ttlMs = 1 + } + + keys := []string{s.destroyStateKey(poolName), s.destroyOwnerKey(poolName)} + argv := []interface{}{ + opensandbox.PoolDestroyStateDestroyed.String(), + ownerID, + strconv.FormatInt(ttlMs, 10), + } + + _, err := markDestroyedScript.Run(ctx, s.client, keys, argv...).Result() + if err != nil && err != redis.Nil { + return &opensandbox.PoolStateStoreUnavailableError{Operation: "MarkDestroyed", Cause: err} } return nil } +// destroyedErrorIfFenced converts the -1 sentinel returned by the fenced-write +// scripts into a *opensandbox.PoolDestroyedError carrying the observed state. +func (s *RedisPoolStateStore) destroyedErrorIfFenced(ctx context.Context, poolName string, scriptResult int64) error { + if scriptResult != -1 { + return nil + } + state, err := s.GetDestroyState(ctx, poolName) + if err != nil { + // The write was refused; report that rather than the follow-up read failure. + state = opensandbox.PoolDestroyStateDestroying + } + return &opensandbox.PoolDestroyedError{PoolName: poolName, State: state} +} + // resolveIdleTTL reads the configured idle TTL from Redis (in ms). // Falls back to DefaultIdleTimeout if not set. func (s *RedisPoolStateStore) resolveIdleTTL(ctx context.Context, poolName string) (int64, error) { @@ -501,5 +652,13 @@ func (s *RedisPoolStateStore) idleTTLKey(poolName string) string { return s.poolKey(poolName, "idleTtlMillis") } +func (s *RedisPoolStateStore) destroyStateKey(poolName string) string { + return s.poolKey(poolName, "destroy:state") +} + +func (s *RedisPoolStateStore) destroyOwnerKey(poolName string) string { + return s.poolKey(poolName, "destroy:owner") +} + // Compile-time interface check. var _ opensandbox.PoolStateStore = (*RedisPoolStateStore)(nil) diff --git a/sdks/sandbox/go/poolredis/store_test.go b/sdks/sandbox/go/poolredis/store_test.go index 2ee0e16dc..94b5e900b 100644 --- a/sdks/sandbox/go/poolredis/store_test.go +++ b/sdks/sandbox/go/poolredis/store_test.go @@ -57,6 +57,8 @@ func cleanupPool(t *testing.T, store *RedisPoolStateStore, poolName string) { store.PrimaryLockKey(poolName), store.maxIdleKey(poolName), store.idleTTLKey(poolName), + store.destroyStateKey(poolName), + store.destroyOwnerKey(poolName), } store.client.Del(ctx, keys...) } @@ -498,3 +500,220 @@ func TestRedisStore_WrapsClientFailures(t *testing.T) { t.Errorf("Operation = %q, want %q", storeErr.Operation, "GetMaxIdle") } } + +// ---------- Destroy Fence And Tombstone Tests ---------- + +func TestRedisStore_GetDestroyState_DefaultsToActive(t *testing.T) { + store := newRedisTestStore(t) + ctx := context.Background() + poolName := "destroy-" + t.Name() + t.Cleanup(func() { cleanupPool(t, store, poolName) }) + + state, err := store.GetDestroyState(ctx, poolName) + if err != nil { + t.Fatalf("GetDestroyState error: %v", err) + } + if state != opensandbox.PoolDestroyStateActive { + t.Errorf("state = %s, want ACTIVE", state) + } +} + +func TestRedisStore_BeginDestroy_FencesWrites(t *testing.T) { + store := newRedisTestStore(t) + ctx := context.Background() + poolName := "destroy-" + t.Name() + t.Cleanup(func() { cleanupPool(t, store, poolName) }) + + if err := store.BeginDestroy(ctx, poolName, "owner-1"); err != nil { + t.Fatalf("BeginDestroy error: %v", err) + } + + state, err := store.GetDestroyState(ctx, poolName) + if err != nil { + t.Fatalf("GetDestroyState error: %v", err) + } + if state != opensandbox.PoolDestroyStateDestroying { + t.Fatalf("state = %s, want DESTROYING", state) + } + + writes := map[string]func() error{ + "PutIdle": func() error { return store.PutIdle(ctx, poolName, "sb-fenced") }, + "SetMaxIdle": func() error { return store.SetMaxIdle(ctx, poolName, 3) }, + "SetIdleEntryTTL": func() error { return store.SetIdleEntryTTL(ctx, poolName, time.Hour) }, + } + for name, write := range writes { + var destroyed *opensandbox.PoolDestroyedError + if err := write(); !errors.As(err, &destroyed) { + t.Errorf("%s error = %v, want *PoolDestroyedError", name, err) + } else if destroyed.State != opensandbox.PoolDestroyStateDestroying { + t.Errorf("%s error state = %s, want DESTROYING", name, destroyed.State) + } + } + + acquired, err := store.TryAcquirePrimaryLock(ctx, poolName, "owner-2", time.Minute) + if err != nil { + t.Fatalf("TryAcquirePrimaryLock error: %v", err) + } + if acquired { + t.Error("TryAcquirePrimaryLock succeeded on a fenced namespace, want false") + } + + renewed, err := store.RenewPrimaryLock(ctx, poolName, "owner-2", time.Minute) + if err != nil { + t.Fatalf("RenewPrimaryLock error: %v", err) + } + if renewed { + t.Error("RenewPrimaryLock succeeded on a fenced namespace, want false") + } +} + +func TestRedisStore_BeginDestroy_RejectsTombstoned(t *testing.T) { + store := newRedisTestStore(t) + ctx := context.Background() + poolName := "destroy-" + t.Name() + t.Cleanup(func() { cleanupPool(t, store, poolName) }) + + if err := store.BeginDestroy(ctx, poolName, "owner-1"); err != nil { + t.Fatalf("BeginDestroy error: %v", err) + } + // Re-entrant while DESTROYING so a retrying owner can make progress. + if err := store.BeginDestroy(ctx, poolName, "owner-1"); err != nil { + t.Fatalf("second BeginDestroy error: %v", err) + } + if err := store.MarkDestroyed(ctx, poolName, "owner-1", time.Hour); err != nil { + t.Fatalf("MarkDestroyed error: %v", err) + } + + var destroyed *opensandbox.PoolDestroyedError + if err := store.BeginDestroy(ctx, poolName, "owner-2"); !errors.As(err, &destroyed) { + t.Fatalf("BeginDestroy on a tombstoned namespace = %v, want *PoolDestroyedError", err) + } +} + +func TestRedisStore_ClearPoolState_KeepsFence(t *testing.T) { + store := newRedisTestStore(t) + ctx := context.Background() + poolName := "destroy-" + t.Name() + t.Cleanup(func() { cleanupPool(t, store, poolName) }) + + if err := store.SetMaxIdle(ctx, poolName, 4); err != nil { + t.Fatalf("SetMaxIdle error: %v", err) + } + if err := store.PutIdle(ctx, poolName, "sb-1"); err != nil { + t.Fatalf("PutIdle error: %v", err) + } + if err := store.BeginDestroy(ctx, poolName, "owner-1"); err != nil { + t.Fatalf("BeginDestroy error: %v", err) + } + if err := store.ClearPoolState(ctx, poolName); err != nil { + t.Fatalf("ClearPoolState error: %v", err) + } + + counters, err := store.SnapshotCounters(ctx, poolName) + if err != nil { + t.Fatalf("SnapshotCounters error: %v", err) + } + if counters.IdleCount != 0 { + t.Errorf("idle count = %d, want 0", counters.IdleCount) + } + maxIdle, err := store.GetMaxIdle(ctx, poolName) + if err != nil { + t.Fatalf("GetMaxIdle error: %v", err) + } + if maxIdle != 0 { + t.Errorf("maxIdle = %d, want 0", maxIdle) + } + state, err := store.GetDestroyState(ctx, poolName) + if err != nil { + t.Fatalf("GetDestroyState error: %v", err) + } + if state != opensandbox.PoolDestroyStateDestroying { + t.Errorf("state = %s, want DESTROYING (ClearPoolState must not lift the fence)", state) + } +} + +func TestRedisStore_MarkDestroyed_TombstoneTTLExpires(t *testing.T) { + store := newRedisTestStore(t) + ctx := context.Background() + poolName := "destroy-" + t.Name() + t.Cleanup(func() { cleanupPool(t, store, poolName) }) + + if err := store.BeginDestroy(ctx, poolName, "owner-1"); err != nil { + t.Fatalf("BeginDestroy error: %v", err) + } + if err := store.MarkDestroyed(ctx, poolName, "owner-1", 200*time.Millisecond); err != nil { + t.Fatalf("MarkDestroyed error: %v", err) + } + + state, err := store.GetDestroyState(ctx, poolName) + if err != nil { + t.Fatalf("GetDestroyState error: %v", err) + } + if state != opensandbox.PoolDestroyStateDestroyed { + t.Fatalf("state = %s, want DESTROYED", state) + } + + deadline := time.Now().Add(5 * time.Second) + for { + state, err = store.GetDestroyState(ctx, poolName) + if err != nil { + t.Fatalf("GetDestroyState error: %v", err) + } + if state == opensandbox.PoolDestroyStateActive { + break + } + if time.Now().After(deadline) { + t.Fatalf("state = %s after the tombstone TTL, want ACTIVE", state) + } + time.Sleep(20 * time.Millisecond) + } + + if err := store.PutIdle(ctx, poolName, "sb-rebound"); err != nil { + t.Errorf("PutIdle after tombstone expiry error: %v", err) + } +} + +func TestRedisStore_MarkDestroyed_ZeroTTLNeverExpires(t *testing.T) { + store := newRedisTestStore(t) + ctx := context.Background() + poolName := "destroy-" + t.Name() + t.Cleanup(func() { cleanupPool(t, store, poolName) }) + + if err := store.MarkDestroyed(ctx, poolName, "owner-1", 0); err != nil { + t.Fatalf("MarkDestroyed error: %v", err) + } + + ttl, err := store.client.PTTL(ctx, store.destroyStateKey(poolName)).Result() + if err != nil { + t.Fatalf("PTTL error: %v", err) + } + // -1 is Redis' answer for a key that exists with no expiry. + if ttl != -1*time.Nanosecond && ttl >= 0 { + t.Errorf("tombstone PTTL = %v, want no expiry", ttl) + } + + state, err := store.GetDestroyState(ctx, poolName) + if err != nil { + t.Fatalf("GetDestroyState error: %v", err) + } + if state != opensandbox.PoolDestroyStateDestroyed { + t.Errorf("state = %s, want DESTROYED", state) + } +} + +func TestRedisStore_MarkDestroyed_RejectsInvalidInput(t *testing.T) { + store := newRedisTestStore(t) + ctx := context.Background() + poolName := "destroy-" + t.Name() + t.Cleanup(func() { cleanupPool(t, store, poolName) }) + + if err := store.MarkDestroyed(ctx, poolName, "", time.Hour); err == nil { + t.Error("MarkDestroyed with a blank owner succeeded, want error") + } + if err := store.MarkDestroyed(ctx, poolName, "owner-1", -time.Second); err == nil { + t.Error("MarkDestroyed with a negative TTL succeeded, want error") + } + if err := store.BeginDestroy(ctx, poolName, ""); err == nil { + t.Error("BeginDestroy with a blank owner succeeded, want error") + } +}