Fixed recovery of loops

streamflow/core/recovery.py

@@ -113,14 +113,22 @@ async def recover(self, failed_job: Job, failed_step: Step) -> None: ...
 
 
 class RetryRequest:
-    __slots__ = ("job_token", "lock", "output_tokens", "version", "workflow")
+    __slots__ = (
+        "job_token",
+        "lock",
+        "output_tokens",
+        "version",
+        "workflow",
+        "workflow_ready",
+    )
 
     def __init__(self) -> None:
         self.job_token: JobToken | None = None
         self.lock: asyncio.Lock = asyncio.Lock()
         self.output_tokens: MutableMapping[str, Token] = {}
         self.version: int = 1
         self.workflow: Workflow | None = None
+        self.workflow_ready: asyncio.Event = asyncio.Event()
 
 
 class TokenAvailability(IntEnum):

streamflow/core/workflow.py

@@ -18,6 +18,7 @@
     PersistableEntity,
 )
 from streamflow.core.utils import get_class_from_name, get_class_fullname
+from streamflow.log_handler import logger
 
 if TYPE_CHECKING:
     from typing import Any
@@ -329,6 +330,7 @@ class Status(IntEnum):
     CANCELLED = 6
     ROLLBACK = 7
     RECOVERY = 8
+    RECOVERED = 9
 
 
 class Step(PersistableEntity, ABC):
@@ -340,6 +342,7 @@ def __init__(self, name: str, workflow: Workflow):
         self.status: Status = Status.WAITING
         self.terminated: bool = False
         self.workflow: Workflow = workflow
+        logger.info(f"STARTING Step {self.name}")
 
     def _add_port(self, name: str, port: Port, type_: DependencyType) -> None:
         if port.name not in self.workflow.ports:

streamflow/cwl/main.py

@@ -85,6 +85,21 @@ async def main(
     if getattr(args, "validate", False):
         return
     await workflow.save(context)
+
+    # import datetime
+    # import posixpath
+    #
+    # from streamflow.token_printer import dag_workflow
+    #
+    # dag_workflow(
+    #     workflow,
+    #     title=posixpath.join(
+    #         os.getcwd(),
+    #         "dev",
+    #         str(datetime.datetime.now()).replace(" ", "_").replace(":", "."),
+    #         "wf",
+    #     ),
+    # )
     if logger.isEnabledFor(logging.INFO):
         logger.info("COMPLETED building of workflow execution plan")
     executor = StreamFlowExecutor(workflow)

streamflow/cwl/transformer.py

@@ -335,6 +335,10 @@ async def transform(
                 self._only_default = True
                 token = None
             token = await self._get_next_token(token, inputs)
+        if not token.recoverable:
+            raise WorkflowDefinitionException(
+                "DEBUG: default token must be recoverable"
+            )
         return {self.get_output_name(): token}
 
 

streamflow/deployment/connector/kubernetes.py

@@ -224,12 +224,16 @@ async def copy_local_to_remote(
         locations: MutableSequence[ExecutionLocation],
         read_only: bool = False,
     ) -> None:
-        await super().copy_local_to_remote(
-            src=src,
-            dst=dst,
-            locations=await self._get_effective_locations(locations, dst),
-            read_only=read_only,
-        )
+        try:
+            await super().copy_local_to_remote(
+                src=src,
+                dst=dst,
+                locations=await self._get_effective_locations(locations, dst),
+                read_only=read_only,
+            )
+        except Exception as e:
+            # ClientConnectionResetError
+            raise WorkflowExecutionException(e)
 
     async def copy_remote_to_remote(
         self,
@@ -240,37 +244,42 @@ async def copy_remote_to_remote(
         source_connector: Connector | None = None,
         read_only: bool = False,
     ) -> None:
-        source_connector = source_connector or self
-        if locations := await copy_same_connector(
-            connector=self,
-            locations=await self._get_effective_locations(locations, dst),
-            source_location=source_location,
-            src=src,
-            dst=dst,
-            read_only=read_only,
-        ):
-            await copy_remote_to_remote(
+        try:
+            source_connector = source_connector or self
+            if locations := await copy_same_connector(
                 connector=self,
-                locations=locations,
+                locations=await self._get_effective_locations(locations, dst),
+                source_location=source_location,
                 src=src,
                 dst=dst,
-                source_connector=source_connector,
-                source_location=source_location,
-                writer_command=[
-                    "sh",
-                    "-c",
-                    " ".join(
-                        await utils.get_remote_to_remote_write_command(
-                            src_connector=source_connector,
-                            src_location=source_location,
-                            src=src,
-                            dst_connector=self,
-                            dst_locations=locations,
-                            dst=dst,
-                        )
-                    ),
-                ],
-            )
+                read_only=read_only,
+            ):
+                await copy_remote_to_remote(
+                    connector=self,
+                    locations=locations,
+                    src=src,
+                    dst=dst,
+                    source_connector=source_connector,
+                    source_location=source_location,
+                    writer_command=[
+                        "sh",
+                        "-c",
+                        " ".join(
+                            await utils.get_remote_to_remote_write_command(
+                                src_connector=source_connector,
+                                src_location=source_location,
+                                src=src,
+                                dst_connector=self,
+                                dst_locations=locations,
+                                dst=dst,
+                            )
+                        ),
+                    ],
+                )
+        except Exception as e:
+            # ClientConnectorError(ConnectionKey(host='130.192.100.107', port=6443, is_ssl=True, ssl=True, proxy=None, proxy_auth=None, proxy_headers_hash=None),
+            # ConnectionRefusedError(111, "Connect call failed ('130.192.100.107', 6443)")))
+            raise WorkflowExecutionException(e)
 
     async def _get_container(
         self, location: ExecutionLocation
@@ -454,51 +463,59 @@ async def run(
             + [utils.encode_command(command)]
         )
         pod, container = location.name.split(":")
-        # noinspection PyUnresolvedReferences
-        result = await asyncio.wait_for(
-            cast(
-                Awaitable,
-                self.client_ws.connect_get_namespaced_pod_exec(
-                    name=pod,
-                    namespace=self.namespace or "default",
-                    container=container,
-                    command=command,
-                    stderr=True,
-                    stdin=False,
-                    stdout=True,
-                    tty=False,
-                    _preload_content=not capture_output,
+        try:
+            # noinspection PyUnresolvedReferences
+            result = await asyncio.wait_for(
+                cast(
+                    Awaitable,
+                    self.client_ws.connect_get_namespaced_pod_exec(
+                        name=pod,
+                        namespace=self.namespace or "default",
+                        container=container,
+                        command=command,
+                        stderr=True,
+                        stdin=False,
+                        stdout=True,
+                        tty=False,
+                        _preload_content=not capture_output,
+                    ),
                 ),
-            ),
-            timeout=timeout,
-        )
-        if capture_output:
-            with io.StringIO() as out_buffer, io.StringIO() as err_buffer:
-                async with result as response:
-                    while not response.closed:
-                        async for msg in response:
-                            data = msg.data.decode("utf-8", "replace")
-                            channel = ord(data[0])
-                            data = data[1:]
-                            if data and channel in [
-                                ws_client.STDOUT_CHANNEL,
-                                ws_client.STDERR_CHANNEL,
-                            ]:
-                                out_buffer.write(data)
-                            elif data and channel == ws_client.ERROR_CHANNEL:
-                                err_buffer.write(data)
-                err = yaml.safe_load(err_buffer.getvalue())
-                if err["status"] == "Success":
-                    return out_buffer.getvalue(), 0
-                else:
-                    if "code" in err:
-                        return err["message"], int(err["code"])
+                timeout=timeout,
+            )
+            if capture_output:
+                with io.StringIO() as out_buffer, io.StringIO() as err_buffer:
+                    async with result as response:
+                        while not response.closed:
+                            async for msg in response:
+                                data = msg.data.decode("utf-8", "replace")
+                                channel = ord(data[0])
+                                data = data[1:]
+                                if data and channel in [
+                                    ws_client.STDOUT_CHANNEL,
+                                    ws_client.STDERR_CHANNEL,
+                                ]:
+                                    out_buffer.write(data)
+                                elif data and channel == ws_client.ERROR_CHANNEL:
+                                    err_buffer.write(data)
+                    err = yaml.safe_load(err_buffer.getvalue())
+                    if err["status"] == "Success":
+                        return out_buffer.getvalue(), 0
                     else:
-                        return err["message"], int(
-                            err["details"]["causes"][0]["message"]
-                        )
-        else:
-            return None
+                        if "code" in err:
+                            return err["message"], int(err["code"])
+                        else:
+                            return err["message"], int(
+                                err["details"]["causes"][0]["message"]
+                            )
+            else:
+                return None
+        except TimeoutError as e:
+            raise WorkflowExecutionException(e)
+        except Exception as e:
+            # AttributeError("'ServerTimeoutError' object has no attribute 'decode'")
+            # ClientConnectorError(ConnectionKey(host='130.192.100.107', port=6443, is_ssl=True, ssl=True, proxy=None, proxy_auth=None, proxy_headers_hash=None),
+            # ConnectionRefusedError(111, "Connect call failed ('130.192.100.107', 6443)"))
+            raise WorkflowExecutionException(e)
 
     async def undeploy(self, external: bool):
         if self.client is not None:

streamflow/log_handler.py

@@ -126,5 +126,5 @@ def highlight(self, msg: str | Any) -> str:
 )
 defaultStreamHandler.setFormatter(formatter)
 logger.addHandler(defaultStreamHandler)
-logger.setLevel(logging.INFO)
+logger.setLevel(logging.DEBUG)
 logger.propagate = False