Fill freed capacity in the same tracking cycle

src/tigerflow/pipeline.py

@@ -60,6 +60,7 @@ def __init__(
 
         self._idle_timeout = timedelta(minutes=idle_timeout)
         self._last_active = datetime.now()
+        self._idle_announced = False
 
         self._delete_input = delete_input
 
@@ -216,14 +217,17 @@ def _run_tracking_cycle(self):
         """Run one iteration of the pipeline tracking loop.
 
         `_handle_task_timeout` reads the task status that `_check_task_status`
-        refreshes, so it must not run before it; a stale status resubmits a
-        Slurm job that has already been replaced.
+        refreshes, so a stale status would resubmit an already replaced Slurm job.
+        `_stage_new_files` follows `_report_failed_files` and `_handle_processed_files`
+        so that slots those free up can be filled in the same cycle. `_check_inactivity`
+        runs last because staging adds to `_filenames`, and a cycle that just started
+        work must not be counted as idle.
         """
         self._check_task_status()
         self._handle_task_timeout()
-        self._stage_new_files()
         self._report_failed_files()
         self._handle_processed_files()
+        self._stage_new_files()
         self._check_inactivity()
 
     def _start_tasks(self):
@@ -473,12 +477,14 @@ def _handle_processed_files(self):
         # Log progress
         if completed_file_ids:
             logger.info("Completed processing {} files", len(completed_file_ids))
-            if self._all_tracked_files_settled():
-                logger.info("No more files to process, starting idle time count")
 
     def _check_inactivity(self):
-        if not self._all_tracked_files_settled():
+        settled = self._all_tracked_files_settled()
+        if not settled:
             self._last_active = datetime.now()
+        elif not self._idle_announced:
+            logger.info("No more files to process, starting idle time count")
+        self._idle_announced = settled
 
         inactivity = datetime.now() - self._last_active
         if inactivity > self._idle_timeout:

tests/integration/pipeline/conftest.py

@@ -137,6 +137,26 @@ def error_logs() -> Iterator[list[str]]:
         logger.remove(sink_id)
 
 
+@pytest.fixture
+def idle_start_logs() -> Iterator[list[str]]:
+    """Yield a list collecting the message of every idle-clock-start record logged.
+
+    Filtered by message content as well as level: the tracking cycle logs other
+    INFO records that a level-only sink would mix in and throw off exact counts.
+    """
+    records: list[str] = []
+
+    sink_id = logger.add(
+        lambda message: records.append(message.record["message"]),
+        level="INFO",
+        filter=lambda record: "starting idle time count" in record["message"],
+    )
+    try:
+        yield records
+    finally:
+        logger.remove(sink_id)
+
+
 @pytest.fixture
 def pending_worker_logs() -> Iterator[list[str]]:
     """Yield a list collecting the message of every "Workers pending" record logged.

tests/integration/pipeline/test_file_staging.py

@@ -13,7 +13,7 @@
 from tigerflow.pipeline import Pipeline
 from tigerflow.staging import StagingContext
 
-from .helpers import PipelineFactory, task_spec
+from .helpers import PipelineFactory, start_fake_tasks, task_spec, write_output
 
 DUPLICATING_STEP = {
     "kind": "callable",
@@ -216,6 +216,37 @@ def test_max_staged_respects_capacity(
             "Capacity is already full, so no further file should be staged"
         )
 
+    def test_completion_frees_capacity_within_the_same_cycle(
+        self, pipeline_factory: PipelineFactory, input_dir: Path
+    ):
+        """A file completing lets a waiting file take its slot in that same cycle.
+
+        Which two files stage first is left to directory order, so the assertions
+        pin the count and the swap rather than specific names.
+        """
+        pipeline = pipeline_factory(
+            staging={"steps": [{"kind": "max_staged", "count": 2}]}
+        )
+        start_fake_tasks(pipeline)
+        for i in range(3):
+            (input_dir / f"f{i}.txt").write_text("x")
+
+        pipeline._run_tracking_cycle()
+        first_batch = {f.name for f in pipeline._symlinks_dir.iterdir()}
+        assert len(first_batch) == 2
+
+        completed = sorted(first_batch)[0]
+        write_output(pipeline, Path(completed).stem)
+        pipeline._run_tracking_cycle()
+
+        staged = {f.name for f in pipeline._symlinks_dir.iterdir()}
+        assert len(staged) == 2, (
+            "The completed file frees a slot that the waiting file should take "
+            "in the same cycle"
+        )
+        assert completed not in staged
+        assert staged - first_batch, "The waiting file should now be staged"
+
     def test_counts_stay_consistent_in_mixed_state(
         self, pipeline_factory: PipelineFactory, input_dir: Path
     ):

tests/integration/pipeline/test_lifecycle.py

@@ -51,6 +51,34 @@ def test_stages_then_completes_across_cycles(
         assert (pipeline._finished_dir / "a.txt").exists()
         assert not (pipeline._symlinks_dir / "a.txt").exists()
 
+    def test_idle_start_logged_once_per_settling(
+        self,
+        pipeline_factory: PipelineFactory,
+        input_dir: Path,
+        idle_start_logs: list[str],
+    ):
+        """The idle clock is announced when work runs out, then stays quiet."""
+        pipeline = pipeline_factory()
+        start_fake_tasks(pipeline)
+
+        (input_dir / "a.txt").write_text("payload")
+        pipeline._run_tracking_cycle()
+        assert idle_start_logs == []
+
+        write_output(pipeline, "a")
+        pipeline._run_tracking_cycle()
+        assert len(idle_start_logs) == 1
+
+        pipeline._run_tracking_cycle()
+        assert len(idle_start_logs) == 1
+
+        # A new file makes the pipeline active again, so settling announces afresh
+        (input_dir / "b.txt").write_text("payload")
+        pipeline._run_tracking_cycle()
+        write_output(pipeline, "b")
+        pipeline._run_tracking_cycle()
+        assert len(idle_start_logs) == 2
+
     def test_file_staged_only_once(
         self, pipeline_factory: PipelineFactory, input_dir: Path
     ):