From ae76c5397fec3b9a873261d47e983f880513855b Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Wed, 21 Oct 2026 13:15:48 +0000 Subject: [PATCH 1/3] Wake placement as soon as capacity or preparation is ready --- .../internal/execution/runtime_compute.go | 4 ++ .../execution/runtime_compute_wake.go | 3 ++ .../execution/runtime_initialization.go | 3 ++ .../internal/execution/runtime_lifecycle.go | 4 ++ .../internal/execution/runtime_manager.go | 16 +++++++- .../internal/execution/runtime_pending.go | 8 ++++ .../runtime_placement_demand_test.go | 39 +++++++++++++++++++ 7 files changed, 76 insertions(+), 1 deletion(-) diff --git a/services/core/internal/execution/runtime_compute.go b/services/core/internal/execution/runtime_compute.go index 8845718ed..c6ca50a16 100644 --- a/services/core/internal/execution/runtime_compute.go +++ b/services/core/internal/execution/runtime_compute.go @@ -40,6 +40,9 @@ func (r *runtimeLifecycle) saveCompute(ctx context.Context, owner deployment.All next, err := r.deployment.SetCompute(ctx, owner, phase, raw, until) if err == nil { log.Ctx(ctx).Info("Runtime compute phase saved", "allocation_id", owner.ID, "node_id", owner.NodeID, "from", owner.ComputePhase, "to", phase) + if phase == "suspended" { + r.freedSlot() + } } return next, err } @@ -233,6 +236,7 @@ func (r *runtimeLifecycle) releaseLostCompute(ctx context.Context, p sandbox.San return err } r.clearRuntimeState(released) + r.freedSlot() return nil } diff --git a/services/core/internal/execution/runtime_compute_wake.go b/services/core/internal/execution/runtime_compute_wake.go index 5e3ecb063..a3578a837 100644 --- a/services/core/internal/execution/runtime_compute_wake.go +++ b/services/core/internal/execution/runtime_compute_wake.go @@ -130,6 +130,9 @@ func (r *runtimeLifecycle) cleanupCompute(ctx context.Context, p sandbox.Sandbox } } _, err := r.deployment.ReleaseAllocation(ctx, owner) + if err == nil { + r.freedSlot() + } return err } diff --git a/services/core/internal/execution/runtime_initialization.go b/services/core/internal/execution/runtime_initialization.go index 14cf1e203..3b9762901 100644 --- a/services/core/internal/execution/runtime_initialization.go +++ b/services/core/internal/execution/runtime_initialization.go @@ -98,6 +98,9 @@ func (w *Worker) initializeEnvironment(ctx context.Context, owner sessions.Envir err := w.prepareEnvironment(operation, owner, &failure) if err == nil { err = w.dispatcher.sessionExecution.CompleteEnvironmentInitialization(operation, owner) + if err == nil { + w.wakeScheduler() + } } if err != nil { log.Warn(ctx, "Environment preparation failed", "environment_id", owner.EnvironmentID, "session_id", owner.SessionID) diff --git a/services/core/internal/execution/runtime_lifecycle.go b/services/core/internal/execution/runtime_lifecycle.go index b9d6bb456..e24655d99 100644 --- a/services/core/internal/execution/runtime_lifecycle.go +++ b/services/core/internal/execution/runtime_lifecycle.go @@ -68,6 +68,7 @@ type runtimeLifecycle struct { wakeHints chan struct{} pendingWake chan struct{} hintDestination func(string) + wakePlacement func() } func newRuntimeManager(owner Owner, deployments *deployment.Service, deploymentReader deployment.Reader, sessionReader sessions.Reader, registry *runtimegateway.Registry, config *RuntimeProvider) (*runtimeManager, error) { @@ -440,6 +441,9 @@ func (r *runtimeLifecycle) observe(ctx context.Context, owner deployment.Allocat return nil // Keep scanning unknown creation; absence is not a final receipt. } _, err = r.deployment.ReleaseAllocation(ctx, owner) + if err == nil { + r.freedSlot() + } return err } // A stopped or missing container does not authorize destroying retained diff --git a/services/core/internal/execution/runtime_manager.go b/services/core/internal/execution/runtime_manager.go index 339bea38b..49ae68855 100644 --- a/services/core/internal/execution/runtime_manager.go +++ b/services/core/internal/execution/runtime_manager.go @@ -109,7 +109,7 @@ func (m *runtimeManager) node(id string) (*runtimeNode, error) { deployment: m.deployment, deployments: m.deploymentService, reader: m.deploymentReader, lease: m.lease, registry: m.registry, config: m.config, nodeID: id, gate: make(chan struct{}, 1), ctx: ctx, stop: stop, - connections: make(map[string]*runtimeConnection), connectionWake: make(chan *runtimegateway.Session, 1), wakeHints: make(chan struct{}, 1), pendingWake: make(chan struct{}, 1), hintDestination: m.hintDestination, + connections: make(map[string]*runtimeConnection), connectionWake: make(chan *runtimegateway.Session, 1), wakeHints: make(chan struct{}, 1), pendingWake: make(chan struct{}, 1), hintDestination: m.hintDestination, wakePlacement: m.wakePlacement, }} m.nodes[id] = n } @@ -142,6 +142,20 @@ func (m *runtimeManager) hintDestination(id string) { } } +// A freed slot may admit waiting demand; the placement ticker recovers a missed wake. +func (m *runtimeManager) wakePlacement() { + select { + case m.placementWake <- struct{}{}: + default: + } +} + +func (r *runtimeLifecycle) freedSlot() { + if r.wakePlacement != nil { + r.wakePlacement() + } +} + func (m *runtimeManager) hints(id string) chan<- struct{} { m.mu.Lock() defer m.mu.Unlock() diff --git a/services/core/internal/execution/runtime_pending.go b/services/core/internal/execution/runtime_pending.go index a5a8e8445..035299188 100644 --- a/services/core/internal/execution/runtime_pending.go +++ b/services/core/internal/execution/runtime_pending.go @@ -78,6 +78,7 @@ func (m *runtimeManager) reservePlacements(ctx context.Context) error { m.placementCursor = deployment.PlacementDemandCursor{} return nil } + incomplete := false for _, environment := range rows { if operation.Err() != nil { if ownership := m.lease.CheckOwnership(ctx); ownership != nil { @@ -100,6 +101,7 @@ func (m *runtimeManager) reservePlacements(ctx context.Context) error { return nil } log.Ctx(ctx).Warn("managed Runtime placement incomplete", "environment_id", environment.ID) + incomplete = true continue } node, err := m.node(reserved.NodeID) @@ -121,6 +123,12 @@ func (m *runtimeManager) reservePlacements(ctx context.Context) error { if next.EnvironmentID == "" { m.placementCursor = deployment.PlacementDemandCursor{} } + if incomplete { + // Capacity-blocked demand should start suspension or expiry now, not on each lane's next tick. + for id := range m.snapshotNodes() { + m.hintDestination(id) + } + } return nil } diff --git a/services/core/internal/execution/runtime_placement_demand_test.go b/services/core/internal/execution/runtime_placement_demand_test.go index f4d5faa85..40255bb9c 100644 --- a/services/core/internal/execution/runtime_placement_demand_test.go +++ b/services/core/internal/execution/runtime_placement_demand_test.go @@ -144,3 +144,42 @@ func TestPlacementBudgetEndsBeforeAttemptPreservesCursor(t *testing.T) { t.Fatal("unattempted demand advanced cursor", m.placementCursor) } } + +func TestCapacityFailedPlacementHintsNodeLanes(t *testing.T) { + m := testRuntimeManager(t) + m.config.InstallationID = uuid.NewString() + node, err := m.node("a") + if err != nil { + t.Fatal(err) + } + m.deploymentReader = &strictDeploymentReader{t: t, placementDemand: func(context.Context, deployment.PlacementDemandCursor) ([]deployment.PlacementDemand, deployment.PlacementDemandCursor, error) { + return []deployment.PlacementDemand{{UnallocatedEnvironment: deployment.UnallocatedEnvironment{ID: uuid.NewString(), TenantID: uuid.NewString()}, At: time.Now()}}, deployment.PlacementDemandCursor{}, nil + }} + _, m.deployment = deploymentOperations(t, &strictDeploymentStorage{t: t}, &strictDeploymentReader{t: t}, &strictExecutionStorage{t: t, withReservation: func(context.Context, deployment.AllocationKey, func(sessions.LockedSession, deployment.ReservationTx) error) error { + return placement.ErrNodeUnavailable + }}) + if err := m.reservePlacements(t.Context()); err != nil { + t.Fatal(err) + } + select { + case <-node.lifecycle.wakeHints: + default: + t.Fatal("capacity-blocked demand did not hint the node lane") + } +} + +func TestFreedSlotWakesPlacement(t *testing.T) { + m := testRuntimeManager(t) + m.placementWake = make(chan struct{}, 1) + node, err := m.node("a") + if err != nil { + t.Fatal(err) + } + node.lifecycle.freedSlot() + node.lifecycle.freedSlot() // coalesced, never blocks + select { + case <-m.placementWake: + default: + t.Fatal("freed slot did not wake placement") + } +} From 788da38a970a921da06a45857384f39941a89b08 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Wed, 21 Oct 2026 13:17:47 +0000 Subject: [PATCH 2/3] Hint only existing node lanes on capacity-blocked placement --- services/core/internal/execution/runtime_pending.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/services/core/internal/execution/runtime_pending.go b/services/core/internal/execution/runtime_pending.go index 035299188..124f2637b 100644 --- a/services/core/internal/execution/runtime_pending.go +++ b/services/core/internal/execution/runtime_pending.go @@ -125,8 +125,11 @@ func (m *runtimeManager) reservePlacements(ctx context.Context) error { } if incomplete { // Capacity-blocked demand should start suspension or expiry now, not on each lane's next tick. - for id := range m.snapshotNodes() { - m.hintDestination(id) + for _, n := range m.snapshotNodes() { + select { + case n.lifecycle.wakeHints <- struct{}{}: + default: + } } } return nil From f613d0b851e02b30999cebdee3dd76979df45c57 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Wed, 21 Oct 2026 13:38:29 +0000 Subject: [PATCH 3/3] Hint node lanes only for full nodes and skip direct lanes Placement now hints the lanes only when it failed for capacity, so a generation rollout or reset does not double every lane's scans. A direct lane frees no node slot, and the freed-slot wake is a plain channel send. --- .../internal/execution/runtime_lifecycle.go | 2 +- .../core/internal/execution/runtime_manager.go | 18 +++++++----------- .../core/internal/execution/runtime_pending.go | 5 +++-- 3 files changed, 11 insertions(+), 14 deletions(-) diff --git a/services/core/internal/execution/runtime_lifecycle.go b/services/core/internal/execution/runtime_lifecycle.go index e24655d99..4eb08d34c 100644 --- a/services/core/internal/execution/runtime_lifecycle.go +++ b/services/core/internal/execution/runtime_lifecycle.go @@ -68,7 +68,7 @@ type runtimeLifecycle struct { wakeHints chan struct{} pendingWake chan struct{} hintDestination func(string) - wakePlacement func() + placementWake chan<- struct{} } func newRuntimeManager(owner Owner, deployments *deployment.Service, deploymentReader deployment.Reader, sessionReader sessions.Reader, registry *runtimegateway.Registry, config *RuntimeProvider) (*runtimeManager, error) { diff --git a/services/core/internal/execution/runtime_manager.go b/services/core/internal/execution/runtime_manager.go index 49ae68855..8cc3f274f 100644 --- a/services/core/internal/execution/runtime_manager.go +++ b/services/core/internal/execution/runtime_manager.go @@ -109,7 +109,7 @@ func (m *runtimeManager) node(id string) (*runtimeNode, error) { deployment: m.deployment, deployments: m.deploymentService, reader: m.deploymentReader, lease: m.lease, registry: m.registry, config: m.config, nodeID: id, gate: make(chan struct{}, 1), ctx: ctx, stop: stop, - connections: make(map[string]*runtimeConnection), connectionWake: make(chan *runtimegateway.Session, 1), wakeHints: make(chan struct{}, 1), pendingWake: make(chan struct{}, 1), hintDestination: m.hintDestination, wakePlacement: m.wakePlacement, + connections: make(map[string]*runtimeConnection), connectionWake: make(chan *runtimegateway.Session, 1), wakeHints: make(chan struct{}, 1), pendingWake: make(chan struct{}, 1), hintDestination: m.hintDestination, placementWake: m.placementWake, }} m.nodes[id] = n } @@ -142,17 +142,13 @@ func (m *runtimeManager) hintDestination(id string) { } } -// A freed slot may admit waiting demand; the placement ticker recovers a missed wake. -func (m *runtimeManager) wakePlacement() { - select { - case m.placementWake <- struct{}{}: - default: - } -} - +// A freed node slot may admit waiting demand; the placement ticker recovers a missed wake. func (r *runtimeLifecycle) freedSlot() { - if r.wakePlacement != nil { - r.wakePlacement() + if r.nodeID != "" { + select { + case r.placementWake <- struct{}{}: + default: + } } } diff --git a/services/core/internal/execution/runtime_pending.go b/services/core/internal/execution/runtime_pending.go index 124f2637b..fc2f3f6a7 100644 --- a/services/core/internal/execution/runtime_pending.go +++ b/services/core/internal/execution/runtime_pending.go @@ -7,6 +7,7 @@ import ( "github.com/MiniMax-AI/OpenAgentCore/internal/obs/log" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment/placement" "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox" ) @@ -101,7 +102,7 @@ func (m *runtimeManager) reservePlacements(ctx context.Context) error { return nil } log.Ctx(ctx).Warn("managed Runtime placement incomplete", "environment_id", environment.ID) - incomplete = true + incomplete = incomplete || errors.Is(err, placement.ErrNodeUnavailable) continue } node, err := m.node(reserved.NodeID) @@ -124,7 +125,7 @@ func (m *runtimeManager) reservePlacements(ctx context.Context) error { m.placementCursor = deployment.PlacementDemandCursor{} } if incomplete { - // Capacity-blocked demand should start suspension or expiry now, not on each lane's next tick. + // Demand blocked by full nodes should start suspension or expiry now, not on each lane's next tick. for _, n := range m.snapshotNodes() { select { case n.lifecycle.wakeHints <- struct{}{}: