fix: GetActiveDeployments fix

backend/controller/controller.go

@@ -575,15 +575,15 @@ func (s *Service) RegisterRunner(ctx context.Context, stream *connect.ClientStre
 			initialised = true
 		}
 
-		maybeDeployment, err := model.ParseDeploymentKey(msg.Deployment)
+		deploymentKey, err := model.ParseDeploymentKey(msg.Deployment)
 		if err != nil {
 			return nil, connect.NewError(connect.CodeInvalidArgument, err)
 		}
 		err = s.dal.UpsertRunner(ctx, dal.Runner{
 			Key:        runnerKey,
 			Endpoint:   msg.Endpoint,
 			State:      dal.RunnerStateFromProto(msg.State),
-			Deployment: maybeDeployment,
+			Deployment: deploymentKey,
 			Labels:     msg.Labels.AsMap(),
 		})
 		if errors.Is(err, libdal.ErrConflict) {
@@ -601,7 +601,6 @@ func (s *Service) RegisterRunner(ctx context.Context, stream *connect.ClientStre
 			}()
 			deferredDeregistration = true
 		}
-
 		routes, err := s.dal.GetRoutingTable(ctx, nil)
 		if errors.Is(err, libdal.ErrNotFound) {
 			routes = map[string][]dal.Route{}
@@ -687,21 +686,13 @@ func (s *Service) Ping(ctx context.Context, req *connect.Request[ftlv1.PingReque
 		return connect.NewResponse(&ftlv1.PingResponse{}), nil
 	}
 
-	// Check if all required deployments are active.
-	modules, err := s.dal.GetActiveDeployments(ctx)
-	if err != nil {
-		return nil, err
-	}
+	//It's not actually ready until it is in the routes table
+	routes := s.routes.Load()
 	var missing []string
-nextModule:
 	for _, module := range s.config.WaitFor {
-		for _, m := range modules {
-			replicas, ok := m.Replicas.Get()
-			if ok && replicas > 0 && m.Module == module {
-				continue nextModule
-			}
+		if _, ok := routes[module]; !ok {
+			missing = append(missing, module)
 		}
-		missing = append(missing, module)
 	}
 	if len(missing) == 0 {
 		return connect.NewResponse(&ftlv1.PingResponse{}), nil

backend/controller/dal/dal.go

@@ -114,7 +114,6 @@ type RunnerState string
 // Runner states.
 const (
 	RunnerStateNew      = RunnerState(dalsql.RunnerStateNew)
-	RunnerStateReserved = RunnerState(dalsql.RunnerStateReserved)
 	RunnerStateAssigned = RunnerState(dalsql.RunnerStateAssigned)
 	RunnerStateDead     = RunnerState(dalsql.RunnerStateDead)
 )

backend/controller/dal/dal_test.go

@@ -130,7 +130,6 @@ func TestDAL(t *testing.T) {
 		Key:        runnerID,
 		Labels:     labels,
 		Endpoint:   "http://localhost:8080",
-		State:      RunnerStateReserved,
 		Deployment: deploymentKey,
 	}
 

backend/controller/dal/internal/sql/queries.sql

@@ -123,8 +123,8 @@ ORDER BY r.key;
 SELECT sqlc.embed(d), m.name AS module_name, m.language, COUNT(r.id) AS replicas
 FROM deployments d
   JOIN modules m ON d.module_id = m.id
-  LEFT JOIN runners r ON d.id = r.deployment_id
-WHERE min_replicas > 0 AND r.state = 'assigned'
+  LEFT JOIN runners r ON d.id = r.deployment_id AND r.state = 'assigned'
+WHERE min_replicas > 0
 GROUP BY d.id, m.name, m.language;
 
 -- name: GetDeploymentsWithMinReplicas :many
@@ -631,12 +631,20 @@ VALUES (
 -- Results may not be ready to be scheduled yet due to event consumption delay
 -- Sorting ensures that brand new events (that may not be ready for consumption)
 -- don't prevent older events from being consumed
+-- We also make sure that the subscription belongs to a deployment that has at least one runner
+WITH runner_count AS (
+  SELECT count(r.deployment_id) as runner_count,
+         r.deployment_id as deployment
+        FROM runners r WHERE r.state = 'assigned'
+        GROUP BY deployment HAVING count(r.deployment_id) > 0
+)
 SELECT
   subs.key::subscription_key as key,
   curser.key as cursor,
   topics.key::topic_key as topic,
   subs.name
 FROM topic_subscriptions subs
+JOIN runner_count on subs.deployment_id = runner_count.deployment
 LEFT JOIN topics ON subs.topic_id = topics.id
 LEFT JOIN topic_events curser ON subs.cursor = curser.id
 WHERE subs.cursor IS DISTINCT FROM topics.head