From 17e73b6c7bd645f1483d272e2af7e5697c69114e Mon Sep 17 00:00:00 2001 From: Sangyoon Park Date: Fri, 14 Aug 2026 11:20:56 -0400 Subject: [PATCH 1/2] Remove duplicated file queries in pipeline staging The pipeline counted the same files in more than one place: `waiting` re-scanned the input directory for files middleware already receives as `candidates`, and the finished-directory scan was repeated in the staging context, completion logging, and the inactivity check. - Drop `waiting` from `StagingContext`; the remaining counts describe files that are staged, completed, or failed - Add `_count_finished` and `_all_tracked_files_settled` as the single source for those counts --- docs/mkdocs/guides/pipeline.md | 5 ++- src/tigerflow/pipeline.py | 32 +++++++++---------- src/tigerflow/staging.py | 5 ++- .../integration/pipeline/test_file_staging.py | 29 ++++------------- tests/unit/test_staging.py | 3 -- 5 files changed, 27 insertions(+), 47 deletions(-) diff --git a/docs/mkdocs/guides/pipeline.md b/docs/mkdocs/guides/pipeline.md index 1489665..54195be 100644 --- a/docs/mkdocs/guides/pipeline.md +++ b/docs/mkdocs/guides/pipeline.md @@ -201,12 +201,11 @@ def my_filter(candidates: list[Path], context: StagingContext) -> list[Path]: ``` The `StagingContext` provides a read-only view of the current pipeline state. The -four counts partition input files by state, so each file is counted in exactly one -of them --- a file that failed in several tasks still counts once: +counts cover files the middleware cannot see --- use `candidates` for the files +awaiting a decision. A file that failed in several tasks still counts once: | Field | Type | Description | | ----- | ---- | ----------- | -| `waiting` | `int` | Files in the input directory not yet staged | | `staged` | `int` | Files staged and still live (failures excluded) | | `completed` | `int` | Files that have finished all tasks | | `failed` | `int` | Files that failed in at least one task | diff --git a/src/tigerflow/pipeline.py b/src/tigerflow/pipeline.py index 1c17118..57ecc94 100644 --- a/src/tigerflow/pipeline.py +++ b/src/tigerflow/pipeline.py @@ -252,26 +252,15 @@ def _start_tasks(self): def _build_staging_context(self) -> StagingContext: """Build the current context for staging middleware.""" failed_stems = self._failed_stems() - - n_finished = sum(1 for f in self._finished_dir.iterdir() if f.is_file()) n_staged = sum( 1 for f in self._symlinks_dir.iterdir() if f.is_file() and f.name.removesuffix(self._config.root_input_ext) not in failed_stems ) - n_waiting = sum( - 1 - for f in self._input_dir.iterdir() - if f.is_file() - and f.name.endswith(self._config.root_input_ext) - and f.name not in self._filenames - ) - return StagingContext( - waiting=n_waiting, staged=n_staged, - completed=n_finished, + completed=self._count_finished(), failed=len(failed_stems), input_dir=self._input_dir, output_dir=self._output_dir, @@ -406,6 +395,19 @@ def _report_failed_files(self): if n_files > 0: logger.error("[{}] {} failed files", task.name, n_files) + def _count_finished(self) -> int: + return sum(1 for file in self._finished_dir.iterdir() if file.is_file()) + + def _all_tracked_files_settled(self) -> bool: + """Whether every tracked file has either finished or failed. + + `_filenames` is never pruned, so it covers every file the pipeline has + ever tracked. + """ + return self._count_finished() + len(self._failed_stems()) >= len( + self._filenames + ) + def _failed_stems(self) -> set[str]: """Stems of input files that failed in at least one task. @@ -465,13 +467,11 @@ def _handle_processed_files(self): # Log progress if completed_file_ids: logger.info("Completed processing {} files", len(completed_file_ids)) - n_finished = sum(1 for f in self._finished_dir.iterdir() if f.is_file()) - if (n_finished + len(self._failed_stems())) >= len(self._filenames): + if self._all_tracked_files_settled(): logger.info("No more files to process, starting idle time count") def _check_inactivity(self): - n_finished = sum(1 for file in self._finished_dir.iterdir() if file.is_file()) - if (n_finished + len(self._failed_stems())) < len(self._filenames): + if not self._all_tracked_files_settled(): self._last_active = datetime.now() inactivity = datetime.now() - self._last_active diff --git a/src/tigerflow/staging.py b/src/tigerflow/staging.py index 2082605..51c408c 100644 --- a/src/tigerflow/staging.py +++ b/src/tigerflow/staging.py @@ -19,11 +19,10 @@ class StagingContext: """Read-only view of pipeline state for staging middleware. - The four counts partition input files by state, so each file is counted - in exactly one of them. + The counts describe files middleware cannot see from its `candidates` + argument. """ - waiting: int # Files in input_dir not yet staged staged: int # Files staged and still live (failures excluded) completed: int # Files in .finished directory failed: int # Files that failed in at least one task diff --git a/tests/integration/pipeline/test_file_staging.py b/tests/integration/pipeline/test_file_staging.py index 9916481..4309f08 100644 --- a/tests/integration/pipeline/test_file_staging.py +++ b/tests/integration/pipeline/test_file_staging.py @@ -135,31 +135,16 @@ def test_duplicate_candidates_do_not_end_run( class TestStagingContext: """Counts handed to middleware must reflect real pipeline state.""" - def test_counts_waiting_files( + def test_staged_count_reflects_staging( self, pipeline_factory: PipelineFactory, input_dir: Path ): - """Unstaged input files are reported as waiting.""" - pipeline = pipeline_factory() - for i in range(3): - (input_dir / f"f{i}.txt").write_text("x") - - context = pipeline._build_staging_context() - - assert context.waiting == 3 - assert context.staged == 0 - assert context.completed == 0 - - def test_counts_shift_after_staging( - self, pipeline_factory: PipelineFactory, input_dir: Path - ): - """Staging moves files from waiting to staged.""" + """Staging a file makes it count as staged.""" pipeline = pipeline_factory() (input_dir / "a.txt").write_text("x") pipeline._stage_new_files() context = pipeline._build_staging_context() - assert context.waiting == 0 assert context.staged == 1 def test_counts_completed_files( @@ -219,11 +204,11 @@ def test_max_staged_respects_capacity( def test_counts_stay_consistent_in_mixed_state( self, pipeline_factory: PipelineFactory, input_dir: Path ): - """All four counts hold together when staged, completed, and failed coexist. + """All counts hold together when staged, completed, and failed coexist. The tests above each set up one state at a time, so this is the only - place `waiting`, `staged`, `completed`, and `failed` are all pinned - against each other. Four files are staged, one completes, one fails: + place `staged`, `completed`, and `failed` are all pinned against each + other. Four files are staged, one completes, one fails: completion removes a symlink and failure does not, so `staged` counts the 3 remaining symlinks excluding the file that failed. The counts are deliberately unequal because one file per state lets several wrong @@ -242,8 +227,8 @@ def test_counts_stay_consistent_in_mixed_state( context = pipeline._build_staging_context() - counts = (context.waiting, context.staged, context.completed, context.failed) - assert counts == (0, 2, 1, 1) + counts = (context.staged, context.completed, context.failed) + assert counts == (2, 1, 1) def test_fan_out_failure_counts_one_file( self, pipeline_factory: PipelineFactory, input_dir: Path diff --git a/tests/unit/test_staging.py b/tests/unit/test_staging.py index 5742baa..69295fa 100644 --- a/tests/unit/test_staging.py +++ b/tests/unit/test_staging.py @@ -24,7 +24,6 @@ def mock_context(tmp_path: Path) -> StagingContext: """Create a mock staging context for testing.""" return StagingContext( - waiting=10, staged=5, completed=3, failed=1, @@ -200,7 +199,6 @@ def test_limits_based_on_staged_count(self, tmp_path: Path): for f in files: f.touch() context = StagingContext( - waiting=10, staged=8, completed=0, failed=0, @@ -216,7 +214,6 @@ def test_returns_empty_when_at_capacity(self, tmp_path: Path): for f in files: f.touch() context = StagingContext( - waiting=5, staged=10, completed=0, failed=0, From 3b871cc787d22bf9fc1f382b097cdd368de07d1b Mon Sep 17 00:00:00 2001 From: Sangyoon Park Date: Fri, 14 Aug 2026 15:22:35 -0400 Subject: [PATCH 2/2] Restore `waiting` count in StagingContext The previous commit dropped `waiting` because middleware already receives those files as `candidates`. That holds only for the first step in the chain: once a step filters the list, later steps can no longer see the full input backlog. Recovering the count would mean a second scan of the input directory, so merge the candidate scan into the context builder instead: one scan feeds both, and `waiting` comes from the list the chain receives. The builder is renamed `_prepare_staging_inputs` to match. --- docs/mkdocs/guides/pipeline.md | 5 ++- src/tigerflow/pipeline.py | 28 ++++++++----- src/tigerflow/staging.py | 5 ++- .../integration/pipeline/test_file_staging.py | 41 +++++++++++++------ tests/unit/test_staging.py | 3 ++ 5 files changed, 54 insertions(+), 28 deletions(-) diff --git a/docs/mkdocs/guides/pipeline.md b/docs/mkdocs/guides/pipeline.md index 54195be..1489665 100644 --- a/docs/mkdocs/guides/pipeline.md +++ b/docs/mkdocs/guides/pipeline.md @@ -201,11 +201,12 @@ def my_filter(candidates: list[Path], context: StagingContext) -> list[Path]: ``` The `StagingContext` provides a read-only view of the current pipeline state. The -counts cover files the middleware cannot see --- use `candidates` for the files -awaiting a decision. A file that failed in several tasks still counts once: +four counts partition input files by state, so each file is counted in exactly one +of them --- a file that failed in several tasks still counts once: | Field | Type | Description | | ----- | ---- | ----------- | +| `waiting` | `int` | Files in the input directory not yet staged | | `staged` | `int` | Files staged and still live (failures excluded) | | `completed` | `int` | Files that have finished all tasks | | `failed` | `int` | Files that failed in at least one task | diff --git a/src/tigerflow/pipeline.py b/src/tigerflow/pipeline.py index 57ecc94..5a6a554 100644 --- a/src/tigerflow/pipeline.py +++ b/src/tigerflow/pipeline.py @@ -249,8 +249,12 @@ def _start_tasks(self): else: raise ValueError(f"Unsupported task kind: {type(task)}") - def _build_staging_context(self) -> StagingContext: - """Build the current context for staging middleware.""" + def _prepare_staging_inputs(self) -> tuple[list[Path], StagingContext]: + """Collect the candidates and context for staging middleware. + + Both are built here so the input directory is scanned once: `waiting` + is derived from the same list the middleware chain receives. + """ failed_stems = self._failed_stems() n_staged = sum( 1 @@ -258,23 +262,25 @@ def _build_staging_context(self) -> StagingContext: if f.is_file() and f.name.removesuffix(self._config.root_input_ext) not in failed_stems ) - return StagingContext( + candidates = [ + f + for f in self._input_dir.iterdir() + if f.is_file() + and f.name.endswith(self._config.root_input_ext) + and f.name not in self._filenames + ] + context = StagingContext( + waiting=len(candidates), staged=n_staged, completed=self._count_finished(), failed=len(failed_stems), input_dir=self._input_dir, output_dir=self._output_dir, ) + return candidates, context def _stage_new_files(self): - context = self._build_staging_context() - candidates = [ - f - for f in self._input_dir.iterdir() - if f.is_file() - and f.name.endswith(self._config.root_input_ext) - and f.name not in self._filenames - ] + candidates, context = self._prepare_staging_inputs() to_stage = self._config.staging.process(candidates, context) for file in to_stage: self._symlinks_dir.joinpath(file.name).symlink_to(file) diff --git a/src/tigerflow/staging.py b/src/tigerflow/staging.py index 51c408c..2082605 100644 --- a/src/tigerflow/staging.py +++ b/src/tigerflow/staging.py @@ -19,10 +19,11 @@ class StagingContext: """Read-only view of pipeline state for staging middleware. - The counts describe files middleware cannot see from its `candidates` - argument. + The four counts partition input files by state, so each file is counted + in exactly one of them. """ + waiting: int # Files in input_dir not yet staged staged: int # Files staged and still live (failures excluded) completed: int # Files in .finished directory failed: int # Files that failed in at least one task diff --git a/tests/integration/pipeline/test_file_staging.py b/tests/integration/pipeline/test_file_staging.py index 4309f08..4ce1406 100644 --- a/tests/integration/pipeline/test_file_staging.py +++ b/tests/integration/pipeline/test_file_staging.py @@ -135,16 +135,31 @@ def test_duplicate_candidates_do_not_end_run( class TestStagingContext: """Counts handed to middleware must reflect real pipeline state.""" - def test_staged_count_reflects_staging( + def test_counts_waiting_files( self, pipeline_factory: PipelineFactory, input_dir: Path ): - """Staging a file makes it count as staged.""" + """Unstaged input files are reported as waiting.""" + pipeline = pipeline_factory() + for i in range(3): + (input_dir / f"f{i}.txt").write_text("x") + + _, context = pipeline._prepare_staging_inputs() + + assert context.waiting == 3 + assert context.staged == 0 + assert context.completed == 0 + + def test_counts_shift_after_staging( + self, pipeline_factory: PipelineFactory, input_dir: Path + ): + """Staging moves files from waiting to staged.""" pipeline = pipeline_factory() (input_dir / "a.txt").write_text("x") pipeline._stage_new_files() - context = pipeline._build_staging_context() + _, context = pipeline._prepare_staging_inputs() + assert context.waiting == 0 assert context.staged == 1 def test_counts_completed_files( @@ -159,7 +174,7 @@ def test_counts_completed_files( (task.output_dir / "a.txt").write_text("done") pipeline._handle_processed_files() - context = pipeline._build_staging_context() + _, context = pipeline._prepare_staging_inputs() assert context.completed == 1 assert context.staged == 0 @@ -176,7 +191,7 @@ def test_staged_count_excludes_failures( (task.output_dir / "a.err").write_text("boom") pipeline._report_failed_files() - context = pipeline._build_staging_context() + _, context = pipeline._prepare_staging_inputs() assert context.failed == 1 assert context.staged == 0 @@ -204,11 +219,11 @@ def test_max_staged_respects_capacity( def test_counts_stay_consistent_in_mixed_state( self, pipeline_factory: PipelineFactory, input_dir: Path ): - """All counts hold together when staged, completed, and failed coexist. + """All four counts hold together when staged, completed, and failed coexist. The tests above each set up one state at a time, so this is the only - place `staged`, `completed`, and `failed` are all pinned against each - other. Four files are staged, one completes, one fails: + place `waiting`, `staged`, `completed`, and `failed` are all pinned + against each other. Four files are staged, one completes, one fails: completion removes a symlink and failure does not, so `staged` counts the 3 remaining symlinks excluding the file that failed. The counts are deliberately unequal because one file per state lets several wrong @@ -225,10 +240,10 @@ def test_counts_stay_consistent_in_mixed_state( pipeline._report_failed_files() pipeline._handle_processed_files() - context = pipeline._build_staging_context() + _, context = pipeline._prepare_staging_inputs() - counts = (context.staged, context.completed, context.failed) - assert counts == (2, 1, 1) + counts = (context.waiting, context.staged, context.completed, context.failed) + assert counts == (0, 2, 1, 1) def test_fan_out_failure_counts_one_file( self, pipeline_factory: PipelineFactory, input_dir: Path @@ -271,7 +286,7 @@ def test_fan_out_failure_counts_one_file( "Failure must not remove the symlink the remaining task reads from" ) - context = pipeline._build_staging_context() + _, context = pipeline._prepare_staging_inputs() assert (context.staged, context.failed) == (1, 1) def test_failed_file_excluded_under_multi_part_extension( @@ -297,7 +312,7 @@ def test_failed_file_excluded_under_multi_part_extension( (task.output_dir / "sample.err").write_text("boom") pipeline._report_failed_files() - context = pipeline._build_staging_context() + _, context = pipeline._prepare_staging_inputs() assert (context.staged, context.failed) == (0, 1) diff --git a/tests/unit/test_staging.py b/tests/unit/test_staging.py index 69295fa..5742baa 100644 --- a/tests/unit/test_staging.py +++ b/tests/unit/test_staging.py @@ -24,6 +24,7 @@ def mock_context(tmp_path: Path) -> StagingContext: """Create a mock staging context for testing.""" return StagingContext( + waiting=10, staged=5, completed=3, failed=1, @@ -199,6 +200,7 @@ def test_limits_based_on_staged_count(self, tmp_path: Path): for f in files: f.touch() context = StagingContext( + waiting=10, staged=8, completed=0, failed=0, @@ -214,6 +216,7 @@ def test_returns_empty_when_at_capacity(self, tmp_path: Path): for f in files: f.touch() context = StagingContext( + waiting=5, staged=10, completed=0, failed=0,