🐛 Fix file-picker downstream service notification issues (ITISFoundation#3058)

- file-picker propagated updates to downstream services due to a side effect (FIXED)
- file-picker properly sends notifications when the file source changes
- dynamic-sidecar properly purges directory when it downloads new data
- dynamic-sidecar inputs downloads are now queued and can no longer be happening in parallel

Author: GitHK

scripts/pydeps.bash

@@ -39,8 +39,8 @@ Run()
 # Examples:
 #  - SEE  https://pydeps.readthedocs.io/en/latest/#usage
 #
-# pydeps.bash services/web/server/src/simcore_service_webserver --cluster
-# pydeps.bash services/web/server/src/simcore_service_webserver --only "simcore_service_webserver.projects" --cluster
+# ./scripts/pydeps.bash services/web/server/src/simcore_service_webserver --cluster
+# ./scripts/pydeps.bash services/web/server/src/simcore_service_webserver --only "simcore_service_webserver.projects" --cluster
 #
 #
 

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/api/containers_extension.py

@@ -129,7 +129,7 @@ async def pull_input_ports(
         PortTypeName.INPUTS, mounted_volumes.disk_inputs_path, port_keys=port_keys
     )
     await send_message(rabbitmq, "Finished pulling inputs")
-    return transferred_bytes
+    return int(transferred_bytes)
 
 
 @containers_router.patch(
@@ -187,7 +187,7 @@ async def pull_output_ports(
         PortTypeName.OUTPUTS, mounted_volumes.disk_outputs_path, port_keys=port_keys
     )
     await send_message(rabbitmq, "Finished pulling output")
-    return transferred_bytes
+    return int(transferred_bytes)
 
 
 @containers_router.post(

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/nodeports.py

@@ -7,9 +7,10 @@
 import time
 from collections import deque
 from pathlib import Path
-from typing import Coroutine, Deque, Dict, List, Optional, Set, Tuple, cast
+from typing import Any, Coroutine, Optional, cast
 
 import magic
+from models_library.projects_nodes import OutputsDict
 from pydantic import ByteSize
 from servicelib.archiving_utils import PrunableFolder, archive_dir, unarchive_dir
 from servicelib.async_utils import run_sequentially_in_context
@@ -49,7 +50,7 @@ def _get_size_of_value(value: ItemConcreteValue) -> int:
 
 
 @run_sequentially_in_context()
-async def upload_outputs(outputs_path: Path, port_keys: List[str]) -> None:
+async def upload_outputs(outputs_path: Path, port_keys: list[str]) -> None:
     """calls to this function will get queued and invoked in sequence"""
     # pylint: disable=too-many-branches
     logger.info("uploading data to simcore...")
@@ -64,9 +65,9 @@ async def upload_outputs(outputs_path: Path, port_keys: List[str]) -> None:
     )
 
     # let's gather the tasks
-    temp_paths: Deque[Path] = deque()
-    ports_values: Dict[str, ItemConcreteValue] = {}
-    archiving_tasks: Deque[Coroutine[None, None, None]] = deque()
+    temp_paths: deque[Path] = deque()
+    ports_values: dict[str, ItemConcreteValue] = {}
+    archiving_tasks: deque[Coroutine[None, None, None]] = deque()
 
     for port in (await PORTS.outputs).values():
         logger.info("Checking port %s", port.key)
@@ -151,7 +152,7 @@ def _is_zip_file(file_path: Path) -> bool:
     return f"{mime_type}" == "application/zip"
 
 
-async def _get_data_from_port(port: Port) -> Tuple[Port, ItemConcreteValue]:
+async def _get_data_from_port(port: Port) -> tuple[Port, Optional[ItemConcreteValue]]:
     tag = f"[{port.key}] "
     logger.info("%s transfer started for %s", tag, port.key)
     start_time = time.perf_counter()
@@ -167,11 +168,83 @@ async def _get_data_from_port(port: Port) -> Tuple[Port, ItemConcreteValue]:
             size_mb,
             size_mb / elapsed_time,
         )
-    return (port, ret)
+    return port, ret
 
 
+async def _download_files(
+    target_path: Path, download_tasks: deque[Coroutine[Any, int, Any]]
+) -> tuple[OutputsDict, ByteSize]:
+    transferred_bytes = 0
+    data: OutputsDict = {}
+
+    if not download_tasks:
+        return data, ByteSize(transferred_bytes)
+
+    # TODO: limit concurrency to avoid saturating storage+db??
+    results: list[tuple[Port, ItemConcreteValue]] = cast(
+        list[tuple[Port, ItemConcreteValue]], await logged_gather(*download_tasks)
+    )
+    logger.info("completed download %s", results)
+    for port, value in results:
+
+        data[port.key] = {"key": port.key, "value": value}
+
+        if _FILE_TYPE_PREFIX in port.property_type:
+
+            # if there are files, move them to the final destination
+            downloaded_file: Optional[Path] = cast(Optional[Path], value)
+            dest_path: Path = target_path / port.key
+
+            if not downloaded_file or not downloaded_file.exists():
+                # the link may be empty
+                # remove files all files from disk when disconnecting port
+                logger.info("removing contents of dir %s", dest_path)
+                await remove_directory(
+                    dest_path, only_children=True, ignore_errors=True
+                )
+                continue
+
+            transferred_bytes = transferred_bytes + downloaded_file.stat().st_size
+
+            # in case of valid file, it is either uncompressed and/or moved to the final directory
+            logger.info("creating directory %s", dest_path)
+            dest_path.mkdir(exist_ok=True, parents=True)
+            data[port.key] = {"key": port.key, "value": str(dest_path)}
+
+            dest_folder = PrunableFolder(dest_path)
+
+            if _is_zip_file(downloaded_file):
+                # unzip updated data to dest_path
+                logger.info("unzipping %s", downloaded_file)
+                unarchived: set[Path] = await unarchive_dir(
+                    archive_to_extract=downloaded_file, destination_folder=dest_path
+                )
+
+                dest_folder.prune(exclude=unarchived)
+
+                logger.info("all unzipped in %s", dest_path)
+            else:
+                logger.info("moving %s", downloaded_file)
+                dest_path = dest_path / Path(downloaded_file).name
+                await async_on_threadpool(
+                    # pylint: disable=cell-var-from-loop
+                    lambda: shutil.move(str(downloaded_file), dest_path)
+                )
+
+                # NOTE: after the download the current value of the port
+                # makes sure previously downloaded files are removed
+                dest_folder.prune(exclude={dest_path})
+
+                logger.info("all moved to %s", dest_path)
+        else:
+            transferred_bytes = transferred_bytes + sys.getsizeof(value)
+
+    return data, ByteSize(transferred_bytes)
+
+
+@run_sequentially_in_context()
 async def download_target_ports(
-    port_type_name: PortTypeName, target_path: Path, port_keys: List[str]
+    port_type_name: PortTypeName, target_path: Path, port_keys: list[str]
 ) -> ByteSize:
     logger.info("retrieving data from simcore...")
     start_time = time.perf_counter()
@@ -183,10 +256,9 @@ async def download_target_ports(
         node_uuid=str(settings.DY_SIDECAR_NODE_ID),
         r_clone_settings=settings.DY_SIDECAR_R_CLONE_SETTINGS,
     )
-    data = {}
 
     # let's gather all the data
-    download_tasks = []
+    download_tasks: deque[Coroutine[Any, int, Any]] = deque()
     for port_value in (await getattr(PORTS, port_type_name.value)).values():
         # if port_keys contains some keys only download them
         logger.info("Checking node %s", port_value.key)
@@ -196,61 +268,7 @@ async def download_target_ports(
         download_tasks.append(_get_data_from_port(port_value))
     logger.info("retrieving %s data", len(download_tasks))
 
-    transfer_bytes = 0
-    if download_tasks:
-        # TODO: limit concurrency to avoid saturating storage+db??
-        results: List[Tuple[Port, ItemConcreteValue]] = cast(
-            List[Tuple[Port, ItemConcreteValue]], await logged_gather(*download_tasks)
-        )
-        logger.info("completed download %s", results)
-        for port, value in results:
-
-            data[port.key] = {"key": port.key, "value": value}
-
-            if _FILE_TYPE_PREFIX in port.property_type:
-
-                # if there are files, move them to the final destination
-                downloaded_file: Optional[Path] = cast(Optional[Path], value)
-                dest_path: Path = target_path / port.key
-
-                if not downloaded_file or not downloaded_file.exists():
-                    # the link may be empty
-                    # remove files all files from disk when disconnecting port
-                    await remove_directory(
-                        dest_path, only_children=True, ignore_errors=True
-                    )
-                    continue
-
-                transfer_bytes = transfer_bytes + downloaded_file.stat().st_size
-
-                # in case of valid file, it is either uncompressed and/or moved to the final directory
-                logger.info("creating directory %s", dest_path)
-                dest_path.mkdir(exist_ok=True, parents=True)
-                data[port.key] = {"key": port.key, "value": str(dest_path)}
-
-                if _is_zip_file(downloaded_file):
-
-                    dest_folder = PrunableFolder(dest_path)
-
-                    # unzip updated data to dest_path
-                    logger.info("unzipping %s", downloaded_file)
-                    unarchived: Set[Path] = await unarchive_dir(
-                        archive_to_extract=downloaded_file, destination_folder=dest_path
-                    )
-
-                    dest_folder.prune(exclude=unarchived)
-
-                    logger.info("all unzipped in %s", dest_path)
-                else:
-                    logger.info("moving %s", downloaded_file)
-                    dest_path = dest_path / Path(downloaded_file).name
-                    await async_on_threadpool(
-                        # pylint: disable=cell-var-from-loop
-                        lambda: shutil.move(str(downloaded_file), dest_path)
-                    )
-                    logger.info("all moved to %s", dest_path)
-            else:
-                transfer_bytes = transfer_bytes + sys.getsizeof(value)
+    data, transferred_bytes = await _download_files(target_path, download_tasks)
 
     # create/update the json file with the new values
     if data:
@@ -261,15 +279,13 @@ async def download_target_ports(
             data = {**current_data, **data}
         data_file.write_text(json.dumps(data))
 
-    transferred = ByteSize(transfer_bytes)
     elapsed_time = time.perf_counter() - start_time
     logger.info(
         "Downloaded %s in %s seconds",
-        transferred.human_readable(decimal=True),
+        transferred_bytes.human_readable(decimal=True),
         elapsed_time,
     )
-
-    return transferred
+    return transferred_bytes
 
 
 __all__ = ["dispatch_update_for_directory", "download_target_ports"]

services/web/server/src/simcore_service_webserver/catalog_client.py

@@ -7,13 +7,13 @@
 from typing import Any, Dict, List, Mapping, Optional
 
 from aiohttp import ClientSession, ClientTimeout, web
-from pydantic import parse_obj_as
 from aiohttp.client_exceptions import (
     ClientConnectionError,
     ClientResponseError,
     InvalidURL,
 )
 from models_library.services_resources import ServiceResourcesDict
+from pydantic import parse_obj_as
 from servicelib.aiohttp.client_session import get_client_session
 from servicelib.aiohttp.rest_responses import wrap_as_envelope
 from servicelib.json_serialization import json_dumps

services/web/server/src/simcore_service_webserver/computation_comp_tasks_listening_task.py

@@ -17,13 +17,12 @@
 from pydantic.types import PositiveInt
 from servicelib.aiohttp.application_keys import APP_DB_ENGINE_KEY
 from servicelib.logging_utils import log_decorator
-from servicelib.utils import logged_gather
 from simcore_postgres_database.webserver_models import DB_CHANNEL_NAME, projects
 from sqlalchemy.sql import select
 
 from .computation_utils import convert_state_from_db
 from .projects import projects_api, projects_exceptions
-from .projects.projects_utils import project_get_depending_nodes
+from .projects.projects_nodes_utils import update_node_outputs
 
 log = logging.getLogger(__name__)
 
@@ -54,43 +53,6 @@ async def _update_project_state(
     await projects_api.notify_project_state_update(app, project)
 
 
-@log_decorator(logger=log)
-async def _update_project_outputs(
-    app: web.Application,
-    user_id: PositiveInt,
-    project_uuid: str,
-    node_uuid: str,
-    outputs: Dict,
-    run_hash: Optional[str],
-    node_errors: Optional[List[ErrorDict]],
-) -> None:
-    # the new outputs might be {}, or {key_name: payload}
-    project, changed_keys = await projects_api.update_project_node_outputs(
-        app,
-        user_id,
-        project_uuid,
-        node_uuid,
-        new_outputs=outputs,
-        new_run_hash=run_hash,
-    )
-
-    await projects_api.notify_project_node_update(
-        app, project, f"{node_uuid}", errors=node_errors
-    )
-    # get depending node and notify for these ones as well
-    depending_node_uuids = await project_get_depending_nodes(project, f"{node_uuid}")
-    await logged_gather(
-        *[
-            projects_api.notify_project_node_update(app, project, nid, errors=None)
-            for nid in depending_node_uuids
-        ]
-    )
-    # notify
-    await projects_api.post_trigger_connected_service_retrieve(
-        app=app, project=project, updated_node_uuid=node_uuid, changed_keys=changed_keys
-    )
-
-
 async def listen(app: web.Application, db_engine: Engine):
     listen_query = f"LISTEN {DB_CHANNEL_NAME};"
     _LISTENING_TASK_BASE_SLEEPING_TIME_S = 1
@@ -138,14 +100,15 @@ async def listen(app: web.Application, db_engine: Engine):
                 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)
-                    await _update_project_outputs(
+                    await update_node_outputs(
                         app,
                         the_project_owner,
                         project_uuid,
                         node_uuid,
                         new_outputs,
                         new_run_hash,
                         node_errors=task_data.get("errors", None),
+                        ui_changed_keys=None,
                     )
 
                 if "state" in task_changes: