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..4eb08d34c 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) + placementWake chan<- struct{} } 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..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, + 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,6 +142,16 @@ func (m *runtimeManager) hintDestination(id string) { } } +// A freed node slot may admit waiting demand; the placement ticker recovers a missed wake. +func (r *runtimeLifecycle) freedSlot() { + if r.nodeID != "" { + select { + case r.placementWake <- struct{}{}: + default: + } + } +} + 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..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" ) @@ -78,6 +79,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 +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 = incomplete || errors.Is(err, placement.ErrNodeUnavailable) continue } node, err := m.node(reserved.NodeID) @@ -121,6 +124,15 @@ func (m *runtimeManager) reservePlacements(ctx context.Context) error { if next.EnvironmentID == "" { m.placementCursor = deployment.PlacementDemandCursor{} } + if incomplete { + // 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{}{}: + default: + } + } + } 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") + } +}