Make new projects able to run ML jobs, and stop an unreachable service from failing project creation

.envs/.production/.django-example

@@ -141,8 +141,13 @@ DJANGO_ACCOUNT_ALLOW_REGISTRATION=True
 WEB_CONCURRENCY=4
 
 # Default processing service
-DEFAULT_PROCESSING_SERVICE_NAME="AMI Data Companion"
-DEFAULT_PROCESSING_SERVICE_ENDPOINT=https://ml.antenna.insectai.org/
+# Only needed when a single processing service is reached over HTTP at a fixed address.
+# Workers that poll for tasks register themselves and need no endpoint here; new projects are
+# connected to the ones marked "public" in the admin. Leave this unset unless you run a push-mode service,
+# and point it somewhere reachable if you do — an address that no longer answers means new
+# projects are created without its pipelines.
+# DEFAULT_PROCESSING_SERVICE_NAME="AMI Data Companion"
+# DEFAULT_PROCESSING_SERVICE_ENDPOINT=https://ml.example.org/
 DEFAULT_PIPELINES_ENABLED=global_moths_2024,quebec_vermont_moths_2023,panama_moths_2023,uk_denmark_moths_2023
 
 # NATS

ami/main/models.py

@@ -263,8 +263,15 @@ def create_related_defaults(self, project: "Project"):
         if not project.sourceimage_collections.exists():
             get_or_create_default_collection(project=project)
         if not project.processing_services.exists():
-            from ami.ml.models.processing_service import get_or_create_default_processing_service
+            from ami.ml.models.processing_service import (
+                attach_public_processing_services,
+                get_or_create_default_processing_service,
+            )
 
+            # The platform's own workers, marked public by an admin, are how a new project
+            # processes anything. A push-mode service is added on top only where one is
+            # configured, such as local development.
+            attach_public_processing_services(project)
             get_or_create_default_processing_service(project=project)
 
 

ami/main/tests.py

@@ -5,6 +5,7 @@
 from io import BytesIO
 from unittest import mock
 
+import requests
 from django.conf import settings
 from django.contrib.auth.models import AnonymousUser
 from django.core.files.uploadedfile import SimpleUploadedFile
@@ -135,6 +136,104 @@ def test_processing_service_if_not_configured(self):
             service, "Default processing service should not be created if environment variables are not set."
         )
 
+    def _create_pull_mode_service(self, name, pipeline_slugs=("pipeline_a", "pipeline_b"), is_public=False):
+        """Register a worker that polls for tasks, with the pipelines it reports it can run."""
+        service = ProcessingService.objects.create(name=name, endpoint_url=None, is_public=is_public)
+        for slug in pipeline_slugs:
+            pipeline, _ = Pipeline.objects.get_or_create(slug=slug, defaults={"name": slug, "version": 1})
+            service.pipelines.add(pipeline)
+        return service
+
+    @override_settings(
+        DEFAULT_PROCESSING_SERVICE_NAME=None,
+        DEFAULT_PROCESSING_SERVICE_ENDPOINT=None,
+        DEFAULT_PIPELINES_ENABLED=["pipeline_a"],
+    )
+    def test_new_project_is_connected_to_public_processing_services_only(self):
+        """
+        A new project is connected to the workers marked public, and to no other worker.
+
+        A job only reaches a worker through the project-to-service relation, so without the
+        public workers a new project cannot process anything. A worker that a project owner
+        registered for their own project must not be attached, or it would receive other
+        projects' images.
+        """
+        fleet = {self._create_pull_mode_service(f"Platform worker {i}", is_public=True) for i in (1, 2)}
+        self._create_pull_mode_service("A lab's own worker", pipeline_slugs=("pipeline_c",))
+
+        project = Project.objects.create(name="Project on the platform workers", create_defaults=True)
+
+        self.assertEqual(set(project.processing_services.all()), fleet)
+        configs = ProjectPipelineConfig.objects.filter(project=project)
+        self.assertEqual({config.pipeline.slug for config in configs}, {"pipeline_a", "pipeline_b"})
+        self.assertEqual(
+            {config.pipeline.slug for config in configs if config.enabled},
+            {"pipeline_a"},
+            "Only the pipelines named in DEFAULT_PIPELINES_ENABLED should start enabled.",
+        )
+
+    @override_settings(
+        DEFAULT_PROCESSING_SERVICE_NAME="Default Processing Service",
+        DEFAULT_PROCESSING_SERVICE_ENDPOINT="http://ml_backend:2000/",
+        DEFAULT_PIPELINES_ENABLED=None,
+    )
+    def test_push_mode_service_is_not_attached_unless_it_is_the_configured_default(self):
+        """
+        Only the configured default push-mode service is attached, not every push-mode service.
+
+        Push-mode services are called at a specific endpoint, so attaching a project to one
+        that was registered for somebody else would send that project's work to it.
+        """
+        # The manager health-checks a new service on creation; this one is never meant to be
+        # reached, and letting the check run costs the suite a DNS timeout.
+        with mock.patch.object(ProcessingService, "get_status"):
+            other = ProcessingService.objects.create(
+                name="Someone else's service", endpoint_url="http://elsewhere:2000/"
+            )
+
+        project = Project.objects.create(name="Project with an unrelated service", create_defaults=True)
+
+        self.assertNotIn(other, project.processing_services.all())
+
+    @override_settings(
+        DEFAULT_PROCESSING_SERVICE_NAME="Unreachable Service",
+        DEFAULT_PROCESSING_SERVICE_ENDPOINT="http://unreachable.invalid:2000/",
+        DEFAULT_PIPELINES_ENABLED=None,
+    )
+    def test_project_is_still_created_when_the_default_service_cannot_be_registered(self):
+        """
+        A default processing service that fails to register does not stop a project being created.
+
+        Registration fetches the service over the network, and a failed request must not turn
+        project creation into a server error. A database error during registration must be
+        contained too, or it aborts the transaction the project is written in. Creation also
+        must not wait on a health check of the service, which retries for minutes when the
+        host does not answer.
+        """
+
+        def fail_in_the_database(*args, **kwargs):
+            with connection.cursor() as cursor:
+                cursor.execute("SELECT 1/0")
+
+        failures = {
+            "network": requests.exceptions.SSLError("certificate verify failed"),
+            "database": fail_in_the_database,
+        }
+        for label, failure in failures.items():
+            with (
+                self.subTest(failure=label),
+                mock.patch.object(ProcessingService, "create_pipelines", side_effect=failure),
+                mock.patch.object(ProcessingService, "get_status") as get_status,
+            ):
+                project = Project.objects.create(name=f"Unregistered service ({label})", create_defaults=True)
+
+                get_status.assert_not_called()
+                self.assertTrue(Project.objects.filter(pk=project.pk).exists())
+                self.assertGreaterEqual(project.deployments.count(), 1)
+                self.assertGreaterEqual(project.sourceimage_collections.count(), 1)
+                self.assertGreaterEqual(project.processing_services.count(), 1)
+                self.assertEqual(ProjectPipelineConfig.objects.filter(project=project).count(), 0)
+
     @override_settings(
         DEFAULT_PROCESSING_SERVICE_NAME="Default Processing Service",
         DEFAULT_PROCESSING_SERVICE_ENDPOINT="http://ml_backend:2000/",

ami/ml/admin.py

@@ -70,8 +70,10 @@ class ProcessingServiceAdmin(AdminBase):
         "id",
         "name",
         "endpoint_url",
+        "is_public",
         "created_at",
     ]
+    list_filter = ["is_public"]
 
 
 @admin.register(AlgorithmCategoryMap)

ami/ml/migrations/0029_processing_service_is_public.py

@@ -0,0 +1,20 @@
+# Generated by Django 4.2.10 on 2026-09-10 20:02
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("ml", "0028_normalize_empty_endpoint_url_to_null"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="processingservice",
+            name="is_public",
+            field=models.BooleanField(
+                default=False,
+                help_text="Connect new projects to this service automatically. Mark only services run for the whole platform: a project's images are sent to every service it is connected to.",
+            ),
+        ),
+    ]

ami/ml/models/processing_service.py

@@ -6,7 +6,7 @@
 
 import requests
 from django.conf import settings
-from django.db import models
+from django.db import models, transaction
 
 from ami.base.models import BaseQuerySet
 from ami.main.models import BaseModel, Project
@@ -70,6 +70,13 @@ class ProcessingService(BaseModel):
     last_seen = models.DateTimeField(null=True)
     last_seen_live = models.BooleanField(null=True)
     last_seen_latency = models.FloatField(null=True)
+    is_public = models.BooleanField(
+        default=False,
+        help_text=(
+            "Connect new projects to this service automatically. Mark only services run for the "
+            "whole platform: a project's images are sent to every service it is connected to."
+        ),
+    )
 
     objects = ProcessingServiceManager()
 
@@ -122,22 +129,7 @@ def create_pipelines(
                 created = True
 
             for project in projects:
-                if enable_only is not None and pipeline.slug not in enable_only:
-                    enabled = False
-                else:
-                    enabled = True
-                project_pipeline_config, created = ProjectPipelineConfig.objects.get_or_create(
-                    pipeline=pipeline,
-                    project=project,
-                    defaults={"enabled": enabled, "config": {}},
-                )
-                if created:
-                    logger.debug(
-                        f"Created project pipeline config for {project.name} and {pipeline.name} (enabled: {enabled})."
-                    )
-                    project_pipeline_config.save()
-                else:
-                    logger.debug(f"Using existing project pipeline config for {project.name} and {pipeline.name}.")
+                configure_pipeline_for_project(project, pipeline, enable_only)
 
             self.pipelines.add(pipeline)
 
@@ -153,7 +145,7 @@ def create_pipelines(
                 if algorithm not in existing_algorithms:
                     logger.debug(f"Registered new algorithm {algorithm.name} to pipeline {pipeline.name}.")
                     pipeline.algorithms.add(algorithm)
-                    pipelines_created.append(algorithm.key)
+                    algorithms_created.append(algorithm.key)
                 else:
                     logger.debug(f"Using existing algorithm {algorithm.name}.")
 
@@ -301,15 +293,63 @@ def get_pipeline_configs(self, timeout=6):
         return info_data.pipelines
 
 
+def configure_pipeline_for_project(
+    project: "Project",
+    pipeline: "Pipeline",
+    enable_only: list[str] | None = None,
+) -> "ProjectPipelineConfig":
+    """
+    Give a project its configuration row for a pipeline, so the pipeline can be selected.
+
+    ``enable_only`` of None starts every pipeline enabled. A list of slugs starts only those
+    enabled, which keeps a new project from offering every pipeline the platform has.
+    """
+    enabled = True if enable_only is None else pipeline.slug in enable_only
+    config, created = ProjectPipelineConfig.objects.get_or_create(
+        pipeline=pipeline,
+        project=project,
+        defaults={"enabled": enabled, "config": {}},
+    )
+    if created:
+        logger.debug(f"Created project pipeline config for {project.name} and {pipeline.name} (enabled: {enabled}).")
+    else:
+        logger.debug(f"Using existing project pipeline config for {project.name} and {pipeline.name}.")
+    return config
+
+
+def attach_public_processing_services(project: "Project") -> list["ProcessingService"]:
+    """
+    Connect a project to the processing services marked public, so it can run jobs from the start.
+
+    A job only reaches a service through this project-to-service link, so a new project with
+    none attached cannot process anything. Only services an admin has marked ``is_public`` are
+    attached, never a worker that a project owner registered for their own project.
+
+    Nothing is requested over the network: the pipelines these services have already
+    registered are configured for the project.
+    """
+    services = list(ProcessingService.objects.filter(is_public=True))
+    if not services:
+        logger.info(f"No processing services are marked public; project {project} was connected to none")
+        return []
+    project.processing_services.add(*services)
+    # Workers in a fleet usually offer the same pipelines, so configure each pipeline once.
+    for pipeline in Pipeline.objects.filter(processing_services__in=services).distinct():
+        configure_pipeline_for_project(project, pipeline, settings.DEFAULT_PIPELINES_ENABLED)
+    logger.info(f"Connected project {project} to {len(services)} public processing services")
+    return services
+
+
 def get_or_create_default_processing_service(
     project: "Project",
     register_pipelines: bool = True,
 ) -> "ProcessingService | None":
     """
-    Create a default processing service for a project.
+    Create the push-mode default processing service for a project, if one is configured.
 
-    If configured, will use the global default processing service
-    for the current environment. Otherwise, it return None.
+    This covers deployments that talk to a single processing service over HTTP, such as a
+    local development stack. Returns None when no endpoint is configured, which is the
+    normal case for a deployment served by pull-mode workers.
 
     Set the "DEFAULT_PROCESSING_SERVICE_ENDPOINT" and "DEFAULT_PROCESSING_SERVICE_NAME"
     environment variables to configure & enable the default processing service.
@@ -318,9 +358,9 @@ def get_or_create_default_processing_service(
     name = settings.DEFAULT_PROCESSING_SERVICE_NAME or "Default Processing Service"
     endpoint_url = settings.DEFAULT_PROCESSING_SERVICE_ENDPOINT
     if not endpoint_url:
-        logger.warning(
-            "Default processing service is not configured. "
-            "Set the 'DEFAULT_PROCESSING_SERVICE_ENDPOINT' environment variable."
+        logger.info(
+            "No push-mode default processing service is configured. "
+            "Set the 'DEFAULT_PROCESSING_SERVICE_ENDPOINT' environment variable to add one."
         )
         return None
 
@@ -331,8 +371,18 @@ def get_or_create_default_processing_service(
     service.projects.add(project)
     logger.info(f"Created default processing service for project {project}")
     if register_pipelines:
-        service.create_pipelines(
-            enable_only=settings.DEFAULT_PIPELINES_ENABLED,
-            projects=Project.objects.filter(pk=project.pk),
-        )
+        # Registering pipelines fetches the service's /info endpoint, and an unreachable or
+        # misconfigured service must not take project creation down with it. The savepoint
+        # discards a partial registration without breaking the caller's transaction.
+        try:
+            with transaction.atomic():
+                service.create_pipelines(
+                    enable_only=settings.DEFAULT_PIPELINES_ENABLED,
+                    projects=Project.objects.filter(pk=project.pk),
+                )
+        except Exception:
+            logger.exception(
+                f"Could not register pipelines from the default processing service '{name}'. "
+                f"Project {project} was created without them."
+            )
     return service