fix(episode): coordinate frame cache extraction with filelock and safe staging

[sivasurya05]

Summary

Fixes race conditions and cache clobbering in Episode.frames() and Episode.frames_at_indices() under concurrent multi-threaded and multi-process runtimes (e.g., PyTorch DataLoader(num_workers > 1) and prefetch).

Uses the standard library's fcntl.flock and isolated staging directories to ensure safe cache extraction and publication, with zero new direct dependencies.

Problems Addressed (#634)

1. Deterministic .tmp path collisions: Both methods previously staged extracted frames using hardcoded paths (frames_{label}.tmp and <output_dir>.tmp). Concurrent workers extracting the same cache key could overwrite or delete each other's in-progress files.

2. Cache deletion race in frames_at_indices(): Uncoordinated cache checks could allow a lagging worker to execute shutil.rmtree(output_directory), deleting another worker's freshly completed cache.

3. Unnecessary third-party locking dependency: Uses the built-in fcntl.flock, following the existing _mirror_entry_lock pattern in src/hflow/storage.py. HFlow targets POSIX runtimes, with Windows supported via WSL2.

4. Read-only workdir support: Cache hits return immediately through a fast path, avoiding directory creation and lock file writes on read-only mounts.

5. Safe orphan cleanup: Abrupt worker termination (e.g., SIGKILL or OOM) can leave partial staging directories behind. Retries safely clean up these directories under the lock using an internal .hflow_staging marker, while preserving caller-owned backup files and directories matching <cache_key>.*.tmp.

Changes Made

src/hflow/episode.py

• Added the _frame_cache_staging(output_directory, is_cached) helper to coordinate cache extraction and publication.
• Fast-path cache hits: Returns immediately when is_cached() is satisfied, avoiding unnecessary locking and filesystem writes.
• Advisory locking: Uses fcntl.flock on <output_dir>.lock to coordinate concurrent threads and processes.
• Double-checked validation: Rechecks the cache state after acquiring the lock to prevent duplicate extraction.
• Safe orphan cleanup: Removes stranded staging directories authenticated by the .hflow_staging marker while holding the lock.
• Isolated staging: Uses tempfile.TemporaryDirectory(prefix=f"{name}.", suffix=".tmp", dir=parent) to create unique staging directories.
• Atomic publication: Publishes completed caches using .replace(output_directory).
• Refactored Episode.frames() and Episode.frames_at_indices() to use _frame_cache_staging.

Dependencies

• No changes to pyproject.toml or uv.lock.
• Zero new direct dependencies.

Test Coverage

Added regression tests in tests/test_frame_cache_concurrency.py:

• test_concurrent_identical_requests_frames — Thread concurrency for frames().
• test_concurrent_identical_requests_frames_at_indices — Thread concurrency for frames_at_indices().
• test_separate_processes_coordination_frames — Multi-process coordination using ProcessPoolExecutor with the spawn context.
• test_separate_processes_coordination_frames_at_indices — Multi-process coordination for indexed frame extraction.
• test_failed_extraction_leaves_no_artifacts_and_retry_succeeds_frames — Failure cleanup and retry for frames().
• test_failed_extraction_leaves_no_artifacts_and_retry_succeeds_frames_at_indices — Failure recovery for indexed frame extraction.
• test_incomplete_indexed_cache_is_rebuilt_safely — Safe rebuilding of incomplete caches under the lock.
• test_cache_hits_succeed_in_read_only_workdir — Ensures cache hits work in read-only workdirs without lock acquisition.
• test_stranded_staging_directories_cleaned_up_on_retry — Verifies cleanup of orphaned staging directories.
• test_unrelated_backup_files_and_directories_are_preserved_on_cache_miss — Ensures unrelated caller-owned backup files and directories remain untouched.

Related Issue

Closes #634

[github-actions]

👋 Hi @sivasurya05 — thank you so much for your first contribution to HFlow!

A maintainer will review your pull request as soon as possible. In the meantime:

• ⭐ Star the repo: https://github.com/Hebbian-Robotics/hflow
• 💬 Join our Discord: https://discord.gg/vacepQvjmg
• 📖 Check CONTRIBUTING.md for the validation commands CI will run
• 📣 Share the project with your network

💡 Tip: one open pull request per contributor at a time. Before starting an issue, check its sidebar for an assignee or a linked pull request. Either one means somebody is already on it; everything else is fair game.

We are excited to have you here and appreciate your help making the project better! 🙌

[greptile-apps]

Confidence Score: 5/5

[Medium risk] Adds file locking to frame cache extraction.

The PR appears safe to merge; no new blocking issue was established.

Summary

The PR coordinates frame-cache extraction with per-cache filesystem locks and isolated staging directories, and adds concurrency and failure-cleanup tests.

• The latest changes replace FileLock with the repository’s existing POSIX flock pattern and remove the direct dependency.
• No new actionable issue was established.

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart TD
  A[Frame request] --> B{Cache complete?}
  B -- Yes --> C[Return cached frames]
  B -- No --> D[Acquire per-cache flock]
  D --> E{Cache complete now?}
  E -- Yes --> C
  E -- No --> F[Remove incomplete cache and marked stranded staging]
  F --> G[Extract into isolated staging directory]
  G --> H{Extraction valid?}
  H -- No --> I[Clean up staging and raise]
  H -- Yes --> J[Publish staging as cache]
  J --> K[Release lock and return frames]

Loading

_{Reviews (5) · Last reviewed commit: "fix(episode): coordinate frame cache ext..."}