Problem
Pipeline asks the same few questions about files in several places, and every place answers them with its own inline code. None of them is defined once, so nothing keeps the copies in step, and one of them is already causing a bug.
- How many files have finished? The same
iterdir count over _finished_dir is spelled out in _build_staging_context, _handle_processed_files, and _check_inactivity.
- Which input files are still waiting? The same three-part filter (is a file, has the root input extension, not already in
_filenames) is spelled out in _build_staging_context and _stage_new_files.
- Has every tracked file finished or failed?
_handle_processed_files and _check_inactivity both compare finished-plus-failed against len(self._filenames), one with >= and one with <. Same question twice, with the comparator flipped because one caller acts when the answer is yes and the other when it is no.
The second one is the bug. _stage_new_files calls _build_staging_context first, then does its own scan, so each cycle walks the input directory twice. A file that arrives between the two walks gets counted by one and missed by the other, and the staging middleware ends up with a context.waiting smaller than the candidate list it receives in that same call.
Notes
_filenames isn't the set of currently staged files. It's seeded at construction from the symlink and finished directories and never pruned, so it really means every file the pipeline has ever tracked. Whatever the settled-state check ends up being called, it should say tracked rather than staged.
Problem
Pipelineasks the same few questions about files in several places, and every place answers them with its own inline code. None of them is defined once, so nothing keeps the copies in step, and one of them is already causing a bug.iterdircount over_finished_diris spelled out in_build_staging_context,_handle_processed_files, and_check_inactivity._filenames) is spelled out in_build_staging_contextand_stage_new_files._handle_processed_filesand_check_inactivityboth compare finished-plus-failed againstlen(self._filenames), one with>=and one with<. Same question twice, with the comparator flipped because one caller acts when the answer is yes and the other when it is no.The second one is the bug.
_stage_new_filescalls_build_staging_contextfirst, then does its own scan, so each cycle walks the input directory twice. A file that arrives between the two walks gets counted by one and missed by the other, and the staging middleware ends up with acontext.waitingsmaller than the candidate list it receives in that same call.Notes
_filenamesisn't the set of currently staged files. It's seeded at construction from the symlink and finished directories and never pruned, so it really means every file the pipeline has ever tracked. Whatever the settled-state check ends up being called, it should say tracked rather than staged.