diff --git a/services/core/internal/execution/runtime_compute.go b/services/core/internal/execution/runtime_compute.go index 3ec7b8fa5..f3bc4f863 100644 --- a/services/core/internal/execution/runtime_compute.go +++ b/services/core/internal/execution/runtime_compute.go @@ -40,7 +40,10 @@ 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" { + switch phase { + case "running": + r.computeRunning() + case "suspended": r.freedSlot() } } diff --git a/services/core/internal/execution/runtime_lifecycle.go b/services/core/internal/execution/runtime_lifecycle.go index 4eb08d34c..ff8de1e0e 100644 --- a/services/core/internal/execution/runtime_lifecycle.go +++ b/services/core/internal/execution/runtime_lifecycle.go @@ -69,6 +69,7 @@ type runtimeLifecycle struct { pendingWake chan struct{} hintDestination func(string) placementWake chan<- struct{} + scheduleWake 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 8cc3f274f..8223d2fe4 100644 --- a/services/core/internal/execution/runtime_manager.go +++ b/services/core/internal/execution/runtime_manager.go @@ -34,6 +34,7 @@ type runtimeManager struct { workspaceCursor string placementCursor deployment.PlacementDemandCursor placementWake chan struct{} + scheduleWake chan<- struct{} workspaceGate chan struct{} sessions sessions.Reader sessionExecution *sessions.ExecutionOperations @@ -109,7 +110,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, placementWake: m.placementWake, + 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, scheduleWake: m.scheduleWake, }} m.nodes[id] = n } @@ -142,6 +143,14 @@ func (m *runtimeManager) hintDestination(id string) { } } +// Work waiting for this compute may start now; the scheduler ticker recovers a missed wake. +func (r *runtimeLifecycle) computeRunning() { + select { + case r.scheduleWake <- struct{}{}: + default: + } +} + // A freed node slot may admit waiting demand; the placement ticker recovers a missed wake. func (r *runtimeLifecycle) freedSlot() { if r.nodeID != "" { diff --git a/services/core/internal/execution/runtime_observation_test.go b/services/core/internal/execution/runtime_observation_test.go index cc13c2f0c..502f33fd2 100644 --- a/services/core/internal/execution/runtime_observation_test.go +++ b/services/core/internal/execution/runtime_observation_test.go @@ -91,3 +91,34 @@ func TestComputeFailureObservationUsesCommittedReceipt(t *testing.T) { }) } } + +func TestRunningComputeWakesScheduler(t *testing.T) { + fixture := workspaceSettlementFixtureWithSetup(t, &observationCaptureProvider{}, workspaceNodeSetup(t, true)) + r, session := fixture.lifecycle, fixture.session + wake := make(chan struct{}, 1) + r.scheduleWake = wake + initial, err := r.provision(t.Context(), session.TenantID, session.Environment.ID, r.config.InstallationID) + if err != nil { + t.Fatal(err) + } + if _, err = fixture.pool.Exec(t.Context(), `UPDATE runtime_allocations SET compute_phase='waking',compute_revision=2,compute_retained_until=clock_timestamp()+interval '1 hour' WHERE id=$1`, initial.ID); err != nil { + t.Fatal(err) + } + waking, err := r.reader.EnvironmentAllocation(t.Context(), initial.Key()) + if err != nil { + t.Fatal(err) + } + select { + case <-wake: + default: + } + // A Turn admitted while the compute was suspended is bound only once it runs. + if _, err = r.saveCompute(t.Context(), waking, "running", runtimeCompute{Current: sandbox.Compute{ID: "source", Name: "source"}}, nil); err != nil { + t.Fatal(err) + } + select { + case <-wake: + default: + t.Fatal("running compute did not wake the scheduler") + } +} diff --git a/services/core/internal/execution/worker.go b/services/core/internal/execution/worker.go index 3b15d0a70..b0068f5ad 100644 --- a/services/core/internal/execution/worker.go +++ b/services/core/internal/execution/worker.go @@ -81,6 +81,7 @@ func StartWorker(ctx context.Context, dispatcher *Dispatcher, owner Owner) (_ *W if err != nil { return nil, err } + worker.runtimes.scheduleWake = worker.scheduleWake defer func() { if err != nil { worker.runtimes.stop()