fix: cron removed on module change (#3512)

fixes #3503

backend/controller/controller.go

@@ -1509,6 +1509,7 @@ func (s *Service) watchModuleChanges(ctx context.Context, sendChange func(respon
 	}
 	moduleState := map[string]moduleStateEntry{}
 	moduleByDeploymentKey := map[string]string{}
+	mostRecentDeploymentByModule := map[string]string{}
 
 	// Seed the notification channel with the current deployments.
 	seedDeployments, err := s.dal.GetActiveDeployments(ctx)
@@ -1549,13 +1550,18 @@ func (s *Service) watchModuleChanges(ctx context.Context, sendChange func(respon
 			// Deleted key
 			if deletion, ok := notification.Deleted.Get(); ok {
 				name := moduleByDeploymentKey[deletion.String()]
+				moduleRemoved := mostRecentDeploymentByModule[name] == deletion.String()
 				response = &ftlv1.PullSchemaResponse{
 					ModuleName:    name,
 					DeploymentKey: deletion.String(),
 					ChangeType:    ftlv1.DeploymentChangeType_DEPLOYMENT_REMOVED,
+					ModuleRemoved: moduleRemoved,
 				}
 				delete(moduleState, name)
 				delete(moduleByDeploymentKey, deletion.String())
+				if moduleRemoved {
+					delete(mostRecentDeploymentByModule, name)
+				}
 			} else if message, ok := notification.Message.Get(); ok {
 				moduleSchema := message.Schema.ToProto().(*schemapb.Module) //nolint:forcetypeassert
 				if moduleSchema.Runtime == nil {
@@ -1580,17 +1586,21 @@ func (s *Service) watchModuleChanges(ctx context.Context, sendChange func(respon
 					if !bytes.Equal(current.hash, newState.hash) || current.minReplicas != newState.minReplicas {
 						changeType := ftlv1.DeploymentChangeType_DEPLOYMENT_CHANGED
 						// A deployment is considered removed if its minReplicas is set to 0.
+						moduleRemoved := false
 						if current.minReplicas > 0 && message.MinReplicas == 0 {
 							changeType = ftlv1.DeploymentChangeType_DEPLOYMENT_REMOVED
+							moduleRemoved = mostRecentDeploymentByModule[message.Schema.Name] == message.Key.String()
 						}
 						response = &ftlv1.PullSchemaResponse{
 							ModuleName:    moduleSchema.Name,
 							DeploymentKey: message.Key.String(),
 							Schema:        moduleSchema,
 							ChangeType:    changeType,
+							ModuleRemoved: moduleRemoved,
 						}
 					}
 				} else {
+					mostRecentDeploymentByModule[message.Schema.Name] = message.Key.String()
 					response = &ftlv1.PullSchemaResponse{
 						ModuleName:    moduleSchema.Name,
 						DeploymentKey: message.Key.String(),

backend/cron/service.go

@@ -87,15 +87,17 @@ func run(ctx context.Context, verbClient CallClient, changes chan *ftlv1.PullSch
 		next, ok := scheduleNext(cronQueue)
 		var nextCh <-chan time.Time
 		if ok {
-			logger.Tracef("Next cron job scheduled in %s", next)
+			logger.Debugf("Next cron job scheduled in %s", next)
 			nextCh = time.After(next)
+		} else {
+			logger.Debugf("No cron jobs scheduled")
 		}
 		select {
 		case <-ctx.Done():
 			return fmt.Errorf("cron service stopped: %w", ctx.Err())
 
 		case resp := <-changes:
-			if err := updateCronJobs(cronJobs, resp); err != nil {
+			if err := updateCronJobs(ctx, cronJobs, resp); err != nil {
 				logger.Errorf(err, "Failed to update cron jobs")
 				continue
 			}
@@ -162,12 +164,21 @@ func scheduleNext(cronQueue []cronJob) (time.Duration, bool) {
 	return time.Until(cronQueue[0].next), true
 }
 
-func updateCronJobs(cronJobs map[string][]cronJob, resp *ftlv1.PullSchemaResponse) error {
+func updateCronJobs(ctx context.Context, cronJobs map[string][]cronJob, resp *ftlv1.PullSchemaResponse) error {
+	logger := log.FromContext(ctx).Scope("cron")
 	switch resp.ChangeType {
 	case ftlv1.DeploymentChangeType_DEPLOYMENT_REMOVED:
+		// We see the new state of the module before we see the removed deployment.
+		// We only want to actually remove if it was not replaced by a new deployment.
+		if !resp.ModuleRemoved {
+			logger.Debugf("Not removing cron jobs for %s as module is still present", resp.DeploymentKey)
+			return nil
+		}
+		logger.Debugf("Removing cron jobs for module %s", resp.ModuleName)
 		delete(cronJobs, resp.ModuleName)
 
 	case ftlv1.DeploymentChangeType_DEPLOYMENT_ADDED, ftlv1.DeploymentChangeType_DEPLOYMENT_CHANGED:
+		logger.Debugf("Updated cron jobs for module %s", resp.ModuleName)
 		moduleSchema, err := schema.ModuleFromProto(resp.Schema)
 		if err != nil {
 			return fmt.Errorf("failed to extract module schema: %w", err)
@@ -176,6 +187,7 @@ func updateCronJobs(cronJobs map[string][]cronJob, resp *ftlv1.PullSchemaRespons
 		if err != nil {
 			return fmt.Errorf("failed to extract cron jobs: %w", err)
 		}
+		logger.Debugf("Adding %d cron jobs for module %s", len(moduleJobs), resp.ModuleName)
 		cronJobs[resp.ModuleName] = moduleJobs
 	}
 	return nil

backend/ingress/service.go

@@ -100,13 +100,18 @@ func Start(ctx context.Context, config Config, pullSchemaClient PullSchemaClient
 					newState.protoSchema.Modules = append(newState.protoSchema.Modules, resp.Schema)
 				}
 			} else if existing != nil {
+				// We see the new state of the module before we see the removed deployment.
+				// We only want to actually remove if it was not replaced by a new deployment.
+				if !resp.ModuleRemoved {
+					logger.Debugf("Not removing ingress for %s as it is not the current deployment", resp.DeploymentKey)
+					return nil
+				}
 				for i := range existing.Modules {
 					if existing.Modules[i].Name != resp.ModuleName {
 						newState.protoSchema.Modules = append(newState.protoSchema.Modules, existing.Modules[i])
 					}
 				}
 			}
-
 			newState.httpRoutes = extractIngressRoutingEntries(newState.protoSchema)
 			sch, err := schema.FromProto(newState.protoSchema)
 			if err != nil {

backend/protos/xyz/block/ftl/v1/schemaservice.proto

@@ -29,6 +29,8 @@ message PullSchemaResponse {
   // If false this is the last schema change in the initial batch, but others may follow later.
   bool more = 3;
   DeploymentChangeType change_type = 5;
+  // If this is true then the module was removed as well as the deployment. This is only set for DEPLOYMENT_REMOVED.
+  bool module_removed = 6;
 }
 
 service SchemaService {