fix: Repair MREs from #823

snakemake/dag.py

@@ -617,8 +617,13 @@ def unneeded_files():
                 yield from filterfalse(partial(needed, job_), tempfiles & files)
 
             # temp output
-            if not job.dynamic_output and (
-                job not in self.targetjobs or job.rule.name == self.workflow.first_rule
+            if (
+                not job.dynamic_output
+                and not job.is_checkpoint
+                and (
+                    job not in self.targetjobs
+                    or job.rule.name == self.workflow.first_rule
+                )
             ):
                 tempfiles = (
                     f

snakemake/scheduler.py

@@ -42,6 +42,11 @@ def cumsum(iterable, zero=[0]):
     "Exiting because a job execution failed. " "Look above for error message"
 )
 
+_ERROR_MSG_ISSUE_823 = (
+    "BUG: Out of jobs ready to be started, but not all files built yet."
+    " Please check https://github.com/snakemake/snakemake/issues/823 for more information."
+)
+
 
 class DummyRateLimiter(ContextDecorator):
     def __enter__(self):
@@ -463,11 +468,23 @@ def schedule(self):
                         return False
                     continue
 
-                # normal shutdown because all jobs have been finished
+                # all runnable jobs have finished, normal shutdown
                 if not needrun and (not running or self.workflow.immediate_submit):
                     self._executor.shutdown()
                     if errors:
                         logger.error(_ERROR_MSG_FINAL)
+                    # we still have unfinished jobs. this is not good. direct
+                    # user to github issue
+                    if self.remaining_jobs:
+                        logger.error(_ERROR_MSG_ISSUE_823)
+                        logger.error(
+                            "Remaining jobs:\n"
+                            + "\n".join(
+                                " - " + str(job) + ": " + ", ".join(job.output)
+                                for job in self.remaining_jobs
+                            )
+                        )
+                        return False
                     return not errors
 
                 # continue if no new job needs to be executed

tests/test_issue823_1/Snakefile

@@ -0,0 +1,27 @@
+rule all:
+    input:
+        "b.txt"
+
+checkpoint a:
+    output:
+        # to reproduce the bug:
+        # >1 output required
+        # one or more temp() outputs required
+        # exception: will work if only the first output is temp()
+        temp('a1.txt'),
+        temp('a2.txt'),
+    shell:
+        "touch {output}"
+
+def _checkpoint_output(wildcards):
+    out = checkpoints.a.get(**wildcards).output
+    return out
+
+rule b:
+    input:
+        _checkpoint_output,
+    output:
+        'b.txt'
+    shell:
+        "touch {output}"
+

tests/test_issue823_2/Snakefile

@@ -0,0 +1,33 @@
+rule all:
+    input:
+        "c.txt",
+
+checkpoint a:
+    output:
+        "a.txt"
+    shell:
+        "touch {output}"
+
+rule b1:
+    output:
+        pipe("b.pipe")
+    shell:
+        ""
+
+rule b2:
+    input:
+        pipe("b.pipe")
+    output:
+        "b.txt"
+    shell:
+        "touch {output}"
+
+rule c:
+    input:
+        "a.txt",
+        "b.txt",
+    output:
+        "c.txt"
+    shell:
+        "touch {output}"
+

tests/test_issue823_3/Snakefile

@@ -0,0 +1,9 @@
+rule all:
+    input:
+        "a.txt"
+
+checkpoint a:
+    output:
+        "a.txt"
+    shell:
+        "touch {output}"

tests/tests.py

@@ -798,6 +798,18 @@ def test_issue805():
     run(dpath("test_issue805"), shouldfail=True)
 
 
+def test_issue823_1():
+    run(dpath("test_issue823_1"))
+
+
+def test_issue823_2():
+    run(dpath("test_issue823_2"))
+
+
+def test_issue823_3():
+    run(dpath("test_issue823_3s"))
+
+
 @skip_on_windows
 def test_pathlib():
     run(dpath("test_pathlib"))