From 6d0d2c8d1e109a1090b74453e2191b9c564fcc32 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Tue, 20 Oct 2026 06:28:53 +0000 Subject: [PATCH] Let capacity pressure free a node whose retained slots are full With active and retained slots both full, pressure could neither suspend an idle allocation (retained full) nor release a checkpoint (active full), so a waiting Session waited for the idle timeout. Pressure now suspends an idle allocation on a full node when its checkpoint can then be released, and the release frees the retained slot on the next pass. --- docs/sandbox-provider.md | 2 +- docs/zh/sandbox-provider.md | 4 +- .../deployment/suspension_pressure.go | 10 +++- .../deploymentpg/retention_pressure_test.go | 54 ++++++++++++------- 4 files changed, 47 insertions(+), 23 deletions(-) diff --git a/docs/sandbox-provider.md b/docs/sandbox-provider.md index 07459f16e..7120e8af4 100644 --- a/docs/sandbox-provider.md +++ b/docs/sandbox-provider.md @@ -240,7 +240,7 @@ sequenceDiagram ### Suspension -Core suspends an allocation after initialization completes, when no root or Subagent Turn is queued, in progress or waiting, no input or file operation is pending, and real activity has been idle for the policy's idle timeout. A Session need not have run a Turn. Capacity pressure can suspend an already idle node allocation earlier, after the inactivity grace in [Timeouts and limits](#timeouts-and-limits), when compatible waiting demand exists. It never interrupts an active Turn or releases capacity before the ordinary quiesce, capture and source-stop proofs complete. If suspension cannot make capacity usable, the waiting request keeps its original deadline. When retained slots are full, pressure can instead [release an idle checkpoint](#cold-replacement-after-checkpoint-retention). +Core suspends an allocation after initialization completes, when no root or Subagent Turn is queued, in progress or waiting, no input or file operation is pending, and real activity has been idle for the policy's idle timeout. A Session need not have run a Turn. Capacity pressure can suspend an already idle node allocation earlier, after the inactivity grace in [Timeouts and limits](#timeouts-and-limits), when compatible waiting demand exists. It never interrupts an active Turn or releases capacity before the ordinary quiesce, capture and source-stop proofs complete. If suspension cannot make capacity usable, the waiting request keeps its original deadline. When retained slots are full, pressure can [release an idle checkpoint](#cold-replacement-after-checkpoint-retention), and it suspends an allocation only if that allocation's checkpoint could then be released. A Provider relies on these rules: diff --git a/docs/zh/sandbox-provider.md b/docs/zh/sandbox-provider.md index 290a40764..e2a4fa51a 100644 --- a/docs/zh/sandbox-provider.md +++ b/docs/zh/sandbox-provider.md @@ -1,7 +1,7 @@ --- title: "沙箱提供商协议" source: docs/sandbox-provider.md -source_hash: e8fe32e265b90a6afb8898eed636c7deabf6b90cff8d07bb37ad7198c298f9c4 +source_hash: 90b575c1ba7ab3e5ff9525c81643f93472e6f4f1cee214f96d2c174437258068 --- 沙箱提供商协议连接 Core 与为 Core 管理的（`openai_hosted`）Environment 提供计算资源的 adapter，并由该 adapter 运行启动 Runtime daemon 的有界引导流程。本页面向 adapter 作者：说明提供商必须实现什么，以及可以依赖 Core 做什么。Docker、microsandbox 和 E2B 实现了该协议。 @@ -241,7 +241,7 @@ sequenceDiagram ### 暂停 {#suspension} -Core 在初始化完成、没有 root 或 Subagent Turn 处于排队、进行中或等待状态、没有待处理的 input 或文件操作，并且真实活动已空闲达到策略的空闲超时之后暂停 allocation。Session 不必运行过 Turn。存在兼容的等待需求时，容量压力可以在[超时与限制](#timeouts-and-limits)所列的无活动宽限期之后，提前暂停已经空闲的 node allocation。它从不打断活动 Turn，也不会在常规 quiesce、capture 和 source-stop 证明完成之前释放容量。如果暂停无法让容量可用，等待中的请求保留原有期限。retained slot 已满时，容量压力可以改为[释放空闲检查点](#cold-replacement-after-checkpoint-retention)。 +Core 在初始化完成、没有 root 或 Subagent Turn 处于排队、进行中或等待状态、没有待处理的 input 或文件操作，并且真实活动已空闲达到策略的空闲超时之后暂停 allocation。Session 不必运行过 Turn。存在兼容的等待需求时，容量压力可以在[超时与限制](#timeouts-and-limits)所列的无活动宽限期之后，提前暂停已经空闲的 node allocation。它从不打断活动 Turn，也不会在常规 quiesce、capture 和 source-stop 证明完成之前释放容量。如果暂停无法让容量可用，等待中的请求保留原有期限。retained slot 已满时，容量压力可以[释放空闲检查点](#cold-replacement-after-checkpoint-retention)，并且只暂停其检查点随后可被释放的 allocation。 提供商依赖以下规则： diff --git a/services/core/internal/deployment/suspension_pressure.go b/services/core/internal/deployment/suspension_pressure.go index aa6b6110e..789139b08 100644 --- a/services/core/internal/deployment/suspension_pressure.go +++ b/services/core/internal/deployment/suspension_pressure.go @@ -74,10 +74,16 @@ func (e *ExecutionOperations) canSuspendForDemand(tx AllocationTx, owner Allocat return false, err } } + release := func(n *placement.Node) { n.Active-- } if source.Retained >= int64(source.MaxRetained) { - return false, nil + // A full node frees a retained slot only by releasing a checkpoint + // after this suspension, so the owner must itself be releasable. + if retain, err := tx.CanRetainEnvironment(owner); err != nil || !retain { + return false, err + } + release = func(n *placement.Node) { n.Active--; n.Retained-- } } - return e.placementNeedsCapacity(tx, pressure, owner.NodeID, func(n *placement.Node) { n.Active-- }) + return e.placementNeedsCapacity(tx, pressure, owner.NodeID, release) } // ExpireCheckpointForDemand ends the retention of an idle suspended diff --git a/services/core/internal/persistence/postgres/deploymentpg/retention_pressure_test.go b/services/core/internal/persistence/postgres/deploymentpg/retention_pressure_test.go index 025dfbf90..266f36f01 100644 --- a/services/core/internal/persistence/postgres/deploymentpg/retention_pressure_test.go +++ b/services/core/internal/persistence/postgres/deploymentpg/retention_pressure_test.go @@ -121,23 +121,41 @@ func TestRetentionPressureKeepsIneligibleCheckpoints(t *testing.T) { } } -func TestPressurePreservesRetainableOwnerWhenRetainedSlotsAreFull(t *testing.T) { - f := newFixture(t) - changes, _ := f.execution(t) - installation, view := f.initialize(t, changes, sandbox.Selection{Provider: "microsandbox", DeploymentSpec: retainedSpecification()}) - node := f.enroll(t, view, deployment.Capacity{MaxActive: 1, MaxRetained: 1}) - f.connect(t, node.NodeID) - owner := qualifyPressureRetention(t, f, runningPressureOwner(t, f, changes, installation, node.NodeID, view.Generation)) - if _, err := f.pool.Exec(t.Context(), `UPDATE runtime_allocations SET compute_phase='running',compute_retained_until=NULL WHERE id=$1`, owner.ID); err != nil { - t.Fatal(err) - } - owner, err := f.adapter.EnvironmentAllocation(t.Context(), owner.Key()) - if err != nil { - t.Fatal(err) - } - pendingPressureDemand(t, f) - until := time.Now().Add(24 * time.Hour) - if _, err = changes.SetCompute(t.Context(), owner, "quiescing", json.RawMessage(`{}`), &until); !errors.Is(err, deployment.ErrNotIdle) { - t.Fatal("suspended without usable capacity", err) +// A full node suspends only an owner whose checkpoint can then be released. +func TestPressureSuspendsReleasableOwnerWhenRetainedSlotsAreFull(t *testing.T) { + for _, test := range []struct { + name string + damage string + suspend bool + }{ + {"releasable", "", true}, + {"no retained native history", `UPDATE devices SET supported_agent_kinds='[]' WHERE id=(SELECT device_id FROM runtime_allocations WHERE id=$1)`, false}, + } { + t.Run(test.name, func(t *testing.T) { + f := newFixture(t) + changes, _ := f.execution(t) + installation, view := f.initialize(t, changes, sandbox.Selection{Provider: "microsandbox", DeploymentSpec: retainedSpecification()}) + node := f.enroll(t, view, deployment.Capacity{MaxActive: 1, MaxRetained: 1}) + f.connect(t, node.NodeID) + owner := qualifyPressureRetention(t, f, runningPressureOwner(t, f, changes, installation, node.NodeID, view.Generation)) + if _, err := f.pool.Exec(t.Context(), `UPDATE runtime_allocations SET compute_phase='running',compute_retained_until=NULL WHERE id=$1`, owner.ID); err != nil { + t.Fatal(err) + } + if test.damage != "" { + if _, err := f.pool.Exec(t.Context(), test.damage, owner.ID); err != nil { + t.Fatal(err) + } + } + owner, err := f.adapter.EnvironmentAllocation(t.Context(), owner.Key()) + if err != nil { + t.Fatal(err) + } + pendingPressureDemand(t, f) + until := time.Now().Add(24 * time.Hour) + _, err = changes.SetCompute(t.Context(), owner, "quiescing", json.RawMessage(`{}`), &until) + if test.suspend != (err == nil) || (err != nil && !errors.Is(err, deployment.ErrNotIdle)) { + t.Fatal("unexpected suspension decision", err) + } + }) } }