Schedule Turns as soon as a resumed compute runs

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()
 		}
 	}

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) {

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 != "" {

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")
+	}
+}

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()