ongoing

Author: sanderegg

services/web/server/src/simcore_service_webserver/db_listener/_db_comp_tasks_listening_task.py

@@ -19,6 +19,7 @@
 from models_library.projects_nodes_io import NodeID
 from models_library.projects_state import RunningState
 from pydantic.types import PositiveInt
+from simcore_postgres_database.models.comp_tasks import comp_tasks
 from simcore_postgres_database.webserver_models import DB_CHANNEL_NAME, projects
 from sqlalchemy.sql import select
 
@@ -61,59 +62,75 @@ async def _update_project_state(
 @dataclass(frozen=True)
 class _CompTaskNotificationPayload:
     action: str
-    data: dict
     changes: dict
     table: str
+    task_id: str | None = None
+    project_id: str | None = None
+    node_id: str | None = None
 
 
 async def _handle_db_notification(
     app: web.Application, payload: _CompTaskNotificationPayload, conn: SAConnection
 ) -> None:
-    task_data = payload.data
+    project_uuid = payload.project_id
+    node_uuid = payload.node_id
     task_changes = payload.changes
 
-    project_uuid = task_data.get("project_id", None)
-    node_uuid = task_data.get("node_id", None)
     if any(x is None for x in [project_uuid, node_uuid]):
         _logger.warning(
             "comp_tasks row is corrupted. TIP: please check DB entry containing '%s'",
-            f"{task_data=}",
+            f"{payload=}",
         )
         return
 
     assert project_uuid  # nosec
     assert node_uuid  # nosec
 
     try:
-        # NOTE: we need someone with the rights to modify that project. the owner is one.
-        # find the user(s) linked to that project
         the_project_owner = await _get_project_owner(conn, project_uuid)
 
-        if any(f in task_changes for f in ["outputs", "run_hash"]):
-            new_outputs = task_data.get("outputs", {})
-            new_run_hash = task_data.get("run_hash", None)
+        # Fetch the latest comp_tasks row for this node/project
+        result = await conn.execute(
+            select(comp_tasks).where(
+                (comp_tasks.c.project_id == project_uuid)
+                & (comp_tasks.c.node_id == node_uuid)
+            )
+        )
+        row = await result.first()
+        if not row:
+            _logger.warning(
+                "No comp_tasks row found for project_id=%s node_id=%s",
+                project_uuid,
+                node_uuid,
+            )
+            return
 
+        if any(f in task_changes for f in ["outputs", "run_hash"]):
+            new_outputs = row.outputs if hasattr(row, "outputs") else {}
+            new_run_hash = row.run_hash if hasattr(row, "run_hash") else None
+            node_errors = row.errors if hasattr(row, "errors") else None
             await update_node_outputs(
                 app,
                 the_project_owner,
                 ProjectID(project_uuid),
                 NodeID(node_uuid),
                 new_outputs,
                 new_run_hash,
-                node_errors=task_data.get("errors", None),
+                node_errors=node_errors,
                 ui_changed_keys=None,
             )
 
         if "state" in task_changes:
-            new_state = convert_state_from_db(task_data["state"])
-            await _update_project_state(
-                app,
-                the_project_owner,
-                ProjectID(project_uuid),
-                NodeID(node_uuid),
-                new_state,
-                node_errors=task_data.get("errors", None),
-            )
+            new_state = row.state if hasattr(row, "state") else None
+            if new_state is not None:
+                await _update_project_state(
+                    app,
+                    the_project_owner,
+                    ProjectID(project_uuid),
+                    NodeID(node_uuid),
+                    convert_state_from_db(new_state),
+                    node_errors=row.errors if hasattr(row, "errors") else None,
+                )
 
     except exceptions.ProjectNotFoundError as exc:
         _logger.warning(
@@ -151,7 +168,6 @@ async def _listen(app: web.Application, db_engine: Engine) -> NoReturn:
                 await asyncio.sleep(_LISTENING_TASK_BASE_SLEEPING_TIME_S)
                 continue
             notification = conn.connection.notifies.get_nowait()
-            # get the data and the info on what changed
             payload = _CompTaskNotificationPayload(**json_loads(notification.payload))
             _logger.debug("received update from database: %s", f"{payload=}")
             await _handle_db_notification(app, payload, conn)
@@ -161,12 +177,10 @@ async def _comp_tasks_listening_task(app: web.Application) -> None:
     _logger.info("starting comp_task db listening task...")
     while True:
         try:
-            # create a special connection here
             db_engine = get_database_engine(app)
             _logger.info("listening to comp_task events...")
             await _listen(app, db_engine)
-        except asyncio.CancelledError:  # noqa: PERF203
-            # we are closing the app..
+        except asyncio.CancelledError:
             _logger.info("cancelled comp_tasks events")
             raise
         except Exception:  # pylint: disable=broad-except