Removed passing of EventBus to a child process worker to fix serialization issue (#4979)

warrkan · web-flow · commit 917cfe4561cf · 2025-11-04T16:44:24.000+01:00
diff --git a/application/backend/app/api/dependencies.py b/application/backend/app/api/dependencies.py
@@ -23,7 +23,7 @@
     ProjectService,
     ResourceNotFoundError,
     SinkService,
-    SourceService,
+    SourceUpdateService,
     SystemService,
 )
 from app.services.base_weights_service import BaseWeightsService
@@ -155,11 +155,11 @@ def get_sink_service(
     return SinkService(event_bus=event_bus, db_session=db)
 
 
-def get_source_service(
+def get_source_update_service(
     event_bus: Annotated[EventBus, Depends(get_event_bus)], db: Annotated[Session, Depends(get_db)]
-) -> SourceService:
-    """Provides a SourceService instance."""
-    return SourceService(event_bus=event_bus, db_session=db)
+) -> SourceUpdateService:
+    """Provides a SourceUpdateService instance."""
+    return SourceUpdateService(event_bus=event_bus, db_session=db)
 
 
 def get_pipeline_service(
diff --git a/application/backend/app/api/routers/sources.py b/application/backend/app/api/routers/sources.py
@@ -14,15 +14,15 @@
 from fastapi.responses import FileResponse, Response
 from pydantic import ValidationError
 
-from app.api.dependencies import get_source_id, get_source_service
+from app.api.dependencies import get_source_id, get_source_update_service
 from app.schemas import Source, SourceCreate
 from app.schemas.source import SourceCreateAdapter
 from app.services import (
     ResourceInUseError,
     ResourceNotFoundError,
     ResourceWithIdAlreadyExistsError,
     ResourceWithNameAlreadyExistsError,
-    SourceService,
+    SourceUpdateService,
 )
 
 logger = logging.getLogger(__name__)
@@ -109,11 +109,11 @@ def create_source(
     source_create: Annotated[
         SourceCreate, Body(description=CREATE_SOURCE_BODY_DESCRIPTION, openapi_examples=CREATE_SOURCE_BODY_EXAMPLES)
     ],
-    source_service: Annotated[SourceService, Depends(get_source_service)],
+    source_update_service: Annotated[SourceUpdateService, Depends(get_source_update_service)],
 ) -> Source:
     """Create and configure a new source"""
     try:
-        return source_service.create(source_create)
+        return source_update_service.create(source_create)
     except (ResourceWithNameAlreadyExistsError, ResourceWithIdAlreadyExistsError) as e:
         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e))
 
@@ -125,10 +125,10 @@ def create_source(
     },
 )
 def list_sources(
-    source_service: Annotated[SourceService, Depends(get_source_service)],
+    source_update_service: Annotated[SourceUpdateService, Depends(get_source_update_service)],
 ) -> list[Source]:
     """List the available sources"""
-    return source_service.list_all()
+    return source_update_service.list_all()
 
 
 @router.get(
@@ -141,11 +141,11 @@ def list_sources(
 )
 def get_source(
     source_id: Annotated[UUID, Depends(get_source_id)],
-    source_service: Annotated[SourceService, Depends(get_source_service)],
+    source_update_service: Annotated[SourceUpdateService, Depends(get_source_update_service)],
 ) -> Source:
     """Get info about a source"""
     try:
-        return source_service.get_by_id(source_id)
+        return source_update_service.get_by_id(source_id)
     except ResourceNotFoundError as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
 
@@ -168,14 +168,14 @@ def update_source(
             openapi_examples=UPDATE_SOURCE_BODY_EXAMPLES,
         ),
     ],
-    source_service: Annotated[SourceService, Depends(get_source_service)],
+    source_update_service: Annotated[SourceUpdateService, Depends(get_source_update_service)],
 ) -> Source:
     """Reconfigure an existing source"""
     if "source_type" in source_config:
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="The 'source_type' field cannot be changed")
     try:
-        source = source_service.get_by_id(source_id)
-        return source_service.update(source, source_config)
+        source = source_update_service.get_by_id(source_id)
+        return source_update_service.update(source, source_config)
     except ResourceNotFoundError as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
     except ResourceWithNameAlreadyExistsError as e:
@@ -198,10 +198,10 @@ def update_source(
 )
 def export_source(
     source_id: Annotated[UUID, Depends(get_source_id)],
-    source_service: Annotated[SourceService, Depends(get_source_service)],
+    source_update_service: Annotated[SourceUpdateService, Depends(get_source_update_service)],
 ) -> Response:
     """Export a source to file"""
-    source = source_service.get_by_id(source_id)
+    source = source_update_service.get_by_id(source_id)
     if not source:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Source with ID {source_id} not found")
 
@@ -226,14 +226,14 @@ def export_source(
 )
 def import_source(
     yaml_file: Annotated[UploadFile, File(description="YAML file containing the source configuration")],
-    source_service: Annotated[SourceService, Depends(get_source_service)],
+    source_update_service: Annotated[SourceUpdateService, Depends(get_source_update_service)],
 ) -> Source:
     """Import a source from file"""
     try:
         yaml_content = yaml_file.file.read()
         source_data = yaml.safe_load(yaml_content)
         source_create = SourceCreateAdapter.validate_python(source_data)
-        return source_service.create(source_create)
+        return source_update_service.create(source_create)
     except yaml.YAMLError as e:
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid YAML format: {str(e)}")
     except (ResourceWithNameAlreadyExistsError, ResourceWithIdAlreadyExistsError) as e:
@@ -256,11 +256,11 @@ def import_source(
 )
 def delete_source(
     source_id: Annotated[UUID, Depends(get_source_id)],
-    source_service: Annotated[SourceService, Depends(get_source_service)],
+    source_update_service: Annotated[SourceUpdateService, Depends(get_source_update_service)],
 ) -> None:
     """Remove a source"""
     try:
-        source_service.delete_by_id(source_id)
+        source_update_service.delete_by_id(source_id)
     except ResourceNotFoundError as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
     except ResourceInUseError as e:
diff --git a/application/backend/app/scheduler.py b/application/backend/app/scheduler.py
@@ -54,7 +54,7 @@ def start_workers(self) -> None:
 
         # Create and start processes
         stream_loader_proc = StreamLoader(
-            self._event_bus, self.frame_queue, self.mp_stop_event, self._event_bus.source_changed_condition
+            self.frame_queue, self.mp_stop_event, self._event_bus.source_changed_condition
         )
 
         inference_server_proc = InferenceWorker(
diff --git a/application/backend/app/services/__init__.py b/application/backend/app/services/__init__.py
@@ -18,7 +18,7 @@
 from .pipeline_service import PipelineService
 from .project_service import ProjectService
 from .sink_service import SinkService
-from .source_service import SourceService
+from .source_service import SourceService, SourceUpdateService
 from .system_service import SystemService
 from .video_stream_service import VideoStreamService
 
@@ -39,6 +39,7 @@
     "ResourceWithNameAlreadyExistsError",
     "SinkService",
     "SourceService",
+    "SourceUpdateService",
     "SystemService",
     "VideoStreamService",
 ]
diff --git a/application/backend/app/services/source_service.py b/application/backend/app/services/source_service.py
@@ -21,8 +21,7 @@
 
 
 class SourceService(GenericPersistenceService[Source, SourceRepository]):
-    def __init__(self, event_bus: EventBus, db_session: Session):
-        self._event_bus: EventBus = event_bus
+    def __init__(self, db_session: Session):
         super().__init__(ServiceConfig(SourceRepository, SourceMapper, ResourceType.SOURCE), db_session)
 
     def get_by_id(self, item_id: UUID) -> Source:
@@ -38,6 +37,21 @@ def create(self, item: Source) -> Source:
         except UniqueConstraintIntegrityError:
             raise ResourceWithNameAlreadyExistsError(ResourceType.SOURCE, item.name)
 
+    def get_active_source(self) -> Source | None:
+        with self._get_repo() as repo:
+            item_db = repo.get_active_source()
+            return self.config.mapper_class.to_schema(item_db) if item_db else None
+
+    @parent_process_only
+    def delete_by_id(self, item_id: UUID) -> None:
+        super().delete_by_id(item_id)
+
+
+class SourceUpdateService(SourceService):
+    def __init__(self, event_bus: EventBus, db_session: Session):
+        self._event_bus: EventBus = event_bus
+        super().__init__(db_session)
+
     @parent_process_only
     def update(self, source: Source, partial_config: dict) -> Source:
         try:
@@ -48,12 +62,3 @@ def update(self, source: Source, partial_config: dict) -> Source:
             return updated
         except UniqueConstraintIntegrityError:
             raise ResourceWithNameAlreadyExistsError(ResourceType.SOURCE, partial_config["name"])
-
-    def get_active_source(self) -> Source | None:
-        with self._get_repo() as repo:
-            item_db = repo.get_active_source()
-            return self.config.mapper_class.to_schema(item_db) if item_db else None
-
-    @parent_process_only
-    def delete_by_id(self, item_id: UUID) -> None:
-        super().delete_by_id(item_id)
diff --git a/application/backend/app/workers/stream_loading.py b/application/backend/app/workers/stream_loading.py
@@ -11,7 +11,6 @@
 from app.db import get_db_session
 from app.schemas import DisconnectedSourceConfig, Source, SourceType
 from app.services import SourceService, VideoStreamService
-from app.services.event.event_bus import EventBus
 from app.stream.stream_data import StreamData
 from app.stream.video_stream import VideoStream
 from app.workers.base import BaseProcessWorker
@@ -26,13 +25,11 @@ class StreamLoader(BaseProcessWorker):
 
     def __init__(
         self,
-        event_bus: EventBus,
         frame_queue: mp.Queue,
         stop_event: EventClass,
         source_changed_condition: Condition | None,
     ) -> None:
         super().__init__(stop_event=stop_event, queues_to_cancel=[frame_queue])
-        self._event_bus = event_bus
         self._frame_queue = frame_queue
         self._source_changed_condition = source_changed_condition
 
@@ -41,7 +38,7 @@ def __init__(
 
     def _load_source(self) -> None:
         with get_db_session() as db:
-            source = SourceService(event_bus=self._event_bus, db_session=db).get_active_source()
+            source = SourceService(db_session=db).get_active_source()
         self._source = source if source is not None else DisconnectedSourceConfig()
         logger.info(f"Active source set to {self._source}. Process: %s", mp.current_process().name)
         self._reset_stream()
diff --git a/application/backend/tests/integration/services/test_source_service.py b/application/backend/tests/integration/services/test_source_service.py
diff --git a/application/backend/tests/unit/routers/test_sources.py b/application/backend/tests/unit/routers/test_sources.py
diff --git a/application/backend/tests/unit/workers/test_stream_loading.py b/application/backend/tests/unit/workers/test_stream_loading.py

Original file line number	Diff line number	Diff line change
`@@ -54,7 +54,7 @@ def start_workers(self) -> None:`
`54`	`54`
`55`	`55`	`# Create and start processes`
`56`	`56`	`stream_loader_proc = StreamLoader(`
`57`		`- self._event_bus, self.frame_queue, self.mp_stop_event, self._event_bus.source_changed_condition`
	`57`	`+ self.frame_queue, self.mp_stop_event, self._event_bus.source_changed_condition`
`58`	`58`	`)`
`59`	`59`
`60`	`60`	`inference_server_proc = InferenceWorker(`