Race condition: concurrent pipeline load requests cause transient failures

[livepeer-tessa]

Summary

When multiple pipeline load requests arrive concurrently for the same pipeline, a race condition causes transient failures. The second request detects that another thread is already loading the pipeline but returns a failure instead of waiting for the first load to complete.

Error Logs

From Grafana fal.ai logs (2026-02-21 04:33 UTC):

04:33:27.277 - Loading 1 pipeline(s): ['streamdiffusionv2']
04:33:27.645 - Loading 1 pipeline(s): ['streamdiffusionv2']  # Second concurrent request
04:33:28.134 - Loading pipeline: streamdiffusionv2
04:33:28.135 - Pipeline streamdiffusionv2 already loading by another thread
04:33:28.138 - ERROR - Failed to load pipeline: streamdiffusionv2
04:33:28.139 - ERROR - Some pipelines failed to load

The pipeline eventually loaded successfully ~27 seconds later, but the intermediate failure triggers error logs and potentially user-facing errors.

Expected Behavior

When a pipeline load request detects that another thread is already loading the same pipeline, it should:

1. Wait for the first load to complete
2. Return success if the pipeline is now loaded (reusing the result from the first load)
3. Only fail if the first load also failed

Current Behavior

The second concurrent request immediately returns a failure when it detects another thread is loading.

Impact

• Unnecessary ERROR level logs in monitoring
• Potential user-facing errors during concurrent operations
• Self-recovers but creates confusion in logs

Component

scope/server/pipeline_manager.py

---

Filed automatically by Scope Error Monitor

[livepeer-tessa]

Fix Analysis

Looking at src/scope/server/pipeline_manager.py, the bug is at line 180-183:

# If already loading, wait
if self._pipeline_statuses.get(pipeline_id) == PipelineStatus.LOADING:
    logger.info(f"Pipeline {pipeline_id} already loading by another thread")
    return False  # BUG: Comment says 'wait' but code returns immediately

Proposed Fix

Add a threading.Condition per pipeline to enable waiting:

1. Add self._loading_conditions: dict[str, threading.Condition] = {} to __init__
2. When starting to load, create a condition: self._loading_conditions[pipeline_id] = threading.Condition(self._lock)
3. When another thread finds LOADING status, wait with timeout: condition.wait(timeout=120)
4. When loading completes (success or failure), notify all: condition.notify_all()
5. After waiting, check the final status and return accordingly

This is a moderate complexity fix due to:

• Threading safety requirements
• Need for proper timeout handling
• Edge cases (what if first load fails?)

---

Analysis by Scope Error Monitor