Run id file should be written prior to execution when resuming (#1051)

* Run id file should be written prior to execution when resuming

* separate run id test so we can exclude from batch

Author: jackie-ob

metaflow/cli.py

@@ -729,11 +729,10 @@ def resume(
         max_num_splits=max_num_splits,
         max_log_size=max_log_size * 1024 * 1024,
     )
+    write_run_id(run_id_file, runtime.run_id)
     runtime.persist_constants()
     runtime.execute()
 
-    write_run_id(run_id_file, runtime.run_id)
-
 
 @parameters.add_custom_parameters(deploy_mode=True)
 @cli.command(help="Run the workflow locally.")

test/core/contexts.json

@@ -92,7 +92,8 @@
                 "NestedUnboundedForeachTest",
                 "DetectSegFaultTest",
                 "TimeoutDecoratorTest",
-                "CardExtensionsImportTest"
+                "CardExtensionsImportTest",
+                "RunIdFileTest"
             ]
         },
         {
@@ -128,7 +129,8 @@
                 "NestedUnboundedForeachTest",
                 "DetectSegFaultTest",
                 "TimeoutDecoratorTest",
-                "CardExtensionsImportTest"
+                "CardExtensionsImportTest",
+                "RunIdFileTest"
             ]
         }
     ],

test/core/tests/run_id_file.py

@@ -0,0 +1,34 @@
+from metaflow_test import MetaflowTest, ExpectationFailed, steps
+
+
+class RunIdFileTest(MetaflowTest):
+    """
+    Resuming and initial running of a flow should write run id file early (prior to execution)
+    """
+
+    RESUME = True
+    PRIORITY = 3
+
+    @steps(0, ["singleton-start"], required=True)
+    def step_start(self):
+        import os
+        from metaflow import current
+
+        # Whether we are in "run" or "resume" mode, --run-id-file must be written prior to execution
+        assert os.path.isfile(
+            "run-id"
+        ), "run id file should exist before resume execution"
+        with open("run-id", "r") as f:
+            run_id_from_file = f.read()
+        assert run_id_from_file == current.run_id
+
+        # Test both regular run and resume paths
+        if not is_resumed():
+            raise ResumeFromHere()
+
+    @steps(2, ["all"])
+    def step_all(self):
+        pass
+
+    def check_results(self, flow, checker):
+        pass