diff --git a/docs/sandbox-provider.md b/docs/sandbox-provider.md index ba352d096..2290bbd56 100644 --- a/docs/sandbox-provider.md +++ b/docs/sandbox-provider.md @@ -242,7 +242,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. A node with no free active slot but a free retained slot also keeps one active slot free: after the free-slot inactivity in [Timeouts and limits](#timeouts-and-limits), Core can suspend an idle allocation without waiting demand, so the next Session does not wait for a suspension. It does not while a Session placed on the node still waits for its sandbox, which the node would otherwise create only after that suspension. 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. +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. A node with no free active slot but a free retained slot also keeps one active slot free: after the free-slot inactivity in [Timeouts and limits](#timeouts-and-limits), Core can suspend an idle allocation without waiting demand, so the next Session does not wait for a suspension. It does not while a Session placed on the node still waits for its sandbox or its initialization, because the node runs that Session's create and preparation only after the suspension. 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 f74e846cb..09ff9bbde 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: 33567fa0e722db83edabdd553bbed29224f02a3f94959224f97daca836c6ca31 +source_hash: 2f9163d7e65eb780ab057b590d6ceca0bb24b5cd2a2376a13f440a53e9dd64e2 --- 沙箱提供商协议连接 Core 与为 Core 管理的（`openai_hosted`）Environment 提供计算资源的 adapter，并由该 adapter 运行启动 Runtime daemon 的有界引导流程。本页面向 adapter 作者：说明提供商必须实现什么，以及可以依赖 Core 做什么。Docker、microsandbox 和 E2B 实现了该协议。 @@ -243,7 +243,7 @@ sequenceDiagram ### 暂停 {#suspension} -Core 在初始化完成、没有 root 或 Subagent Turn 处于排队、进行中或等待状态、没有待处理的 input 或文件操作，并且真实活动已空闲达到策略的空闲超时之后暂停 allocation。Session 不必运行过 Turn。存在兼容的等待需求时，容量压力可以在[超时与限制](#timeouts-and-limits)所列的无活动宽限期之后，提前暂停已经空闲的 node allocation。没有空闲 active slot、但有空闲 retained slot 的 node 还会保留一个空闲 active slot：达到[超时与限制](#timeouts-and-limits)所列的空闲 slot 无活动时间后，即使没有等待需求，Core 也可以暂停空闲的 allocation，让下一个 Session 不必等待暂停。如果已放置到该 node 的 Session 还在等待创建沙箱，则不会这样做，否则 node 要等这次暂停完成后才会创建那个沙箱。它从不打断活动 Turn，也不会在常规 quiesce、capture 和 source-stop 证明完成之前释放容量。如果暂停无法让容量可用，等待中的请求保留原有期限。retained slot 已满时，容量压力可以[释放空闲检查点](#cold-replacement-after-checkpoint-retention)，并且只暂停其检查点随后可被释放的 allocation。 +Core 在初始化完成、没有 root 或 Subagent Turn 处于排队、进行中或等待状态、没有待处理的 input 或文件操作，并且真实活动已空闲达到策略的空闲超时之后暂停 allocation。Session 不必运行过 Turn。存在兼容的等待需求时，容量压力可以在[超时与限制](#timeouts-and-limits)所列的无活动宽限期之后，提前暂停已经空闲的 node allocation。没有空闲 active slot、但有空闲 retained slot 的 node 还会保留一个空闲 active slot：达到[超时与限制](#timeouts-and-limits)所列的空闲 slot 无活动时间后，即使没有等待需求，Core 也可以暂停空闲的 allocation，让下一个 Session 不必等待暂停。如果已放置到该 node 的 Session 还在等待创建沙箱或完成初始化，则不会这样做，否则 node 要等这次暂停完成后才会为那个 Session 创建沙箱和执行准备。它从不打断活动 Turn，也不会在常规 quiesce、capture 和 source-stop 证明完成之前释放容量。如果暂停无法让容量可用，等待中的请求保留原有期限。retained slot 已满时，容量压力可以[释放空闲检查点](#cold-replacement-after-checkpoint-retention)，并且只暂停其检查点随后可被释放的 allocation。 提供商依赖以下规则： diff --git a/services/core/internal/db/queries/runtime_suspension_pressure.sql b/services/core/internal/db/queries/runtime_suspension_pressure.sql index 3ae0acf61..a2bc87869 100644 --- a/services/core/internal/db/queries/runtime_suspension_pressure.sql +++ b/services/core/internal/db/queries/runtime_suspension_pressure.sql @@ -5,6 +5,14 @@ WHERE a.node_id IS NOT NULL AND a.state <> 'released' AND (a.compute_phase IN ('quiescing', 'suspending') OR (a.compute_retained_until <= clock_timestamp() AND a.compute_phase <> 'disabled')); +-- name: ListRuntimeStartingNodes :many +-- A node is starting a Session while that Session's sandbox is not created or +-- its initialization has not finished. +SELECT DISTINCT p.node_id +FROM runtime_placements p JOIN environments e ON e.id=p.environment_id +LEFT JOIN runtime_allocations a ON a.environment_id=p.environment_id AND a.state <> 'released' +WHERE p.released_at IS NULL AND (a.id IS NULL OR e.initialization IN ('pending', 'running')); + -- name: ListWaitingCheckpointRestores :many SELECT DISTINCT a.node_id, a.deployment_generation, (a.compute_state->'snapshot'->'Compatibility')::jsonb AS checkpoint FROM runtime_allocations a JOIN environments e ON e.id=a.environment_id JOIN sessions s ON s.id=e.session_id diff --git a/services/core/internal/db/sqlc/runtime_suspension_pressure.sql.go b/services/core/internal/db/sqlc/runtime_suspension_pressure.sql.go index b55c9ba9d..689f160a7 100644 --- a/services/core/internal/db/sqlc/runtime_suspension_pressure.sql.go +++ b/services/core/internal/db/sqlc/runtime_suspension_pressure.sql.go @@ -79,6 +79,35 @@ func (q *Queries) ListIdleRuntimeCheckpoints(ctx context.Context, nodeID pgtype. return items, nil } +const listRuntimeStartingNodes = `-- name: ListRuntimeStartingNodes :many +SELECT DISTINCT p.node_id +FROM runtime_placements p JOIN environments e ON e.id=p.environment_id +LEFT JOIN runtime_allocations a ON a.environment_id=p.environment_id AND a.state <> 'released' +WHERE p.released_at IS NULL AND (a.id IS NULL OR e.initialization IN ('pending', 'running')) +` + +// A node is starting a Session while that Session's sandbox is not created or +// its initialization has not finished. +func (q *Queries) ListRuntimeStartingNodes(ctx context.Context) ([]pgtype.UUID, error) { + rows, err := q.db.Query(ctx, listRuntimeStartingNodes) + if err != nil { + return nil, err + } + defer rows.Close() + items := []pgtype.UUID{} + for rows.Next() { + var node_id pgtype.UUID + if err := rows.Scan(&node_id); err != nil { + return nil, err + } + items = append(items, node_id) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const listRuntimeSuspensionInFlightNodes = `-- name: ListRuntimeSuspensionInFlightNodes :many SELECT DISTINCT a.node_id FROM runtime_allocations a diff --git a/services/core/internal/deployment/placement/placement.go b/services/core/internal/deployment/placement/placement.go index 0dd80bc09..151deea6a 100644 --- a/services/core/internal/deployment/placement/placement.go +++ b/services/core/internal/deployment/placement/placement.go @@ -86,12 +86,11 @@ type Node struct { ID string Online, ServingReady bool // ReadyGeneration is nil until the node serves a generation. - ReadyGeneration *uint64 - TargetState string - // Reserved counts placements still waiting for their sandbox; Active includes them. - Active, Retained, Reserved int64 - MaxActive, MaxRetained int - CoreURL string + ReadyGeneration *uint64 + TargetState string + Active, Retained int64 + MaxActive, MaxRetained int + CoreURL string } // Placement is the node and generation a new Session's Environment reserves. diff --git a/services/core/internal/deployment/suspension_pressure.go b/services/core/internal/deployment/suspension_pressure.go index 018e69d79..05c217e43 100644 --- a/services/core/internal/deployment/suspension_pressure.go +++ b/services/core/internal/deployment/suspension_pressure.go @@ -20,6 +20,7 @@ type SuspensionDemand struct { Deployment placement.Deployment Nodes []placement.Node InFlightNodes []string + StartingNodes []string CheckpointRestores []CheckpointRestoreDemand } @@ -48,11 +49,11 @@ func fullNode(pressure SuspensionDemand, owner Allocation) *placement.Node { // canKeepFreeSlot allows suspending an allocation that stayed idle for // freeSlotIdle on a full node with no reclamation in flight. The node must // also have a free retained slot, or the freed active slot admits no Session, -// and no placed Session waiting for its sandbox, which the node's lane would -// otherwise create only after this suspension. +// and no Session still starting on it: the node runs one operation at a time, +// so that Session's create and preparation would wait for this suspension. func canKeepFreeSlot(pressure SuspensionDemand, owner Allocation) bool { n := fullNode(pressure, owner) - return n != nil && n.Retained < int64(n.MaxRetained) && n.Reserved == 0 && !slices.Contains(pressure.InFlightNodes, owner.NodeID) + return n != nil && n.Retained < int64(n.MaxRetained) && !slices.Contains(pressure.InFlightNodes, owner.NodeID) && !slices.Contains(pressure.StartingNodes, owner.NodeID) } // canSuspendForDemand reports whether compatible waiting demand would place on diff --git a/services/core/internal/persistence/postgres/deploymentpg/suspension_pressure.go b/services/core/internal/persistence/postgres/deploymentpg/suspension_pressure.go index ee3c8b59f..c47d30825 100644 --- a/services/core/internal/persistence/postgres/deploymentpg/suspension_pressure.go +++ b/services/core/internal/persistence/postgres/deploymentpg/suspension_pressure.go @@ -31,6 +31,13 @@ func (t *allocationTx) LoadSuspensionDemand(current deployment.Allocation) (depl for _, node := range inFlight { result.InFlightNodes = append(result.InFlightNodes, uuidString(node)) } + starting, err := t.q.ListRuntimeStartingNodes(t.ctx) + if err != nil { + return result, err + } + for _, node := range starting { + result.StartingNodes = append(result.StartingNodes, uuidString(node)) + } result.Nodes, err = placementpg.LoadNodes(t.ctx, t.q) if err != nil { return result, err diff --git a/services/core/internal/persistence/postgres/deploymentpg/suspension_pressure_test.go b/services/core/internal/persistence/postgres/deploymentpg/suspension_pressure_test.go index 6b90cb4a2..ab811e76c 100644 --- a/services/core/internal/persistence/postgres/deploymentpg/suspension_pressure_test.go +++ b/services/core/internal/persistence/postgres/deploymentpg/suspension_pressure_test.go @@ -88,20 +88,36 @@ func TestPressureSuspensionRequiresUsefulCapacity(t *testing.T) { } } -func TestFreeSlotWaitsForPlacedSession(t *testing.T) { - f := newFixture(t) - changes, _ := f.execution(t) - installation, view := f.initialize(t, changes, sandbox.Selection{Provider: "microsandbox", DeploymentSpec: testSpecification("microsandbox")}) - node := f.enroll(t, view, deployment.Capacity{MaxActive: 2, MaxRetained: 3}) - f.connect(t, node.NodeID) - owner := runningPressureOwner(t, f, changes, installation, node.NodeID, view.Generation, "3 minutes") - placed := hostedEnvironment(t, f.pool) - if _, err := f.pool.Exec(t.Context(), `INSERT INTO runtime_placements(environment_id,node_id,deployment_generation) VALUES($1,$2,$3)`, placed.EnvironmentID, node.NodeID, view.Generation); err != nil { - t.Fatal(err) - } - until := time.Now().Add(time.Hour) - if _, err := changes.SetCompute(t.Context(), owner, "quiescing", json.RawMessage(`{}`), &until); !errors.Is(err, deployment.ErrNotIdle) { - t.Fatalf("quiesce: %v, want %v", err, deployment.ErrNotIdle) +func TestFreeSlotWaitsForStartingSession(t *testing.T) { + for _, created := range []bool{false, true} { + t.Run(map[bool]string{false: "waiting for its sandbox", true: "initializing"}[created], func(t *testing.T) { + f := newFixture(t) + changes, _ := f.execution(t) + installation, view := f.initialize(t, changes, sandbox.Selection{Provider: "microsandbox", DeploymentSpec: testSpecification("microsandbox")}) + node := f.enroll(t, view, deployment.Capacity{MaxActive: 2, MaxRetained: 3}) + f.connect(t, node.NodeID) + owner := runningPressureOwner(t, f, changes, installation, node.NodeID, view.Generation, "3 minutes") + placed := hostedEnvironment(t, f.pool) + if _, err := f.pool.Exec(t.Context(), `INSERT INTO runtime_placements(environment_id,node_id,deployment_generation) VALUES($1,$2,$3)`, placed.EnvironmentID, node.NodeID, view.Generation); err != nil { + t.Fatal(err) + } + if created { + starting, err := changes.ReserveAllocation(t.Context(), placed, installation, credentialHash()) + if err != nil { + t.Fatal(err) + } + if _, err = f.pool.Exec(t.Context(), `UPDATE runtime_allocations SET state='running',create_settled=true WHERE id=$1`, starting.ID); err != nil { + t.Fatal(err) + } + if _, err = f.pool.Exec(t.Context(), `UPDATE environments SET initialization='running' WHERE id=$1`, placed.EnvironmentID); err != nil { + t.Fatal(err) + } + } + until := time.Now().Add(time.Hour) + if _, err := changes.SetCompute(t.Context(), owner, "quiescing", json.RawMessage(`{}`), &until); !errors.Is(err, deployment.ErrNotIdle) { + t.Fatalf("quiesce: %v, want %v", err, deployment.ErrNotIdle) + } + }) } } diff --git a/services/core/internal/persistence/postgres/placementpg/placementpg.go b/services/core/internal/persistence/postgres/placementpg/placementpg.go index 3279274ef..ad5abde8a 100644 --- a/services/core/internal/persistence/postgres/placementpg/placementpg.go +++ b/services/core/internal/persistence/postgres/placementpg/placementpg.go @@ -106,7 +106,7 @@ func ComputeBlocksAdmission(ctx context.Context, q *sqlc.Queries, session pgtype func node(row sqlc.ListRuntimeNodesRow) placement.Node { n := placement.Node{ ID: uuidString(row.ID), Online: row.Online, ServingReady: row.ServingReady, TargetState: row.TargetState, - Active: row.Active, Retained: row.Retained, Reserved: row.Reserved, MaxActive: int(row.MaxActive), MaxRetained: int(row.MaxRetained), CoreURL: row.CoreUrl, + Active: row.Active, Retained: row.Retained, MaxActive: int(row.MaxActive), MaxRetained: int(row.MaxRetained), CoreURL: row.CoreUrl, } if row.ReadyGeneration.Valid { ready := uint64(row.ReadyGeneration.Int64)