# How VoiceStudio's Batch Queue Handles Progress, Folder Watching, and Pinned-Voice Routing with Incompatible Engines

> Discover how VoiceStudio's batch queue manages progress, folder watching, and pinned-voice routing with incompatible engines using async detection and fallback mechanisms.

- Repository: [Palash Debnath/VoiceStudio](https://github.com/debpalash/VoiceStudio)
- Tags: internals
- Published: 2026-09-06

---

**VoiceStudio's batch queue uses an async folder watcher to detect new jobs, creates prioritized Task objects, and automatically falls back to compatible engines or per-segment generation when a pinned voice engine lacks native batch support.**

The batch queue in the [debpalash/VoiceStudio](https://github.com/debpalash/VoiceStudio) repository is the core orchestration layer for large-scale TTS and dubbing workflows. It continuously monitors a `progress/` directory for incoming job folders, transforms each discovery into a trackable `Task`, and intelligently routes work to voice engines—gracefully handling cases where a user-pinned engine cannot process batch operations.

---

## Folder Watcher Architecture

The queue relies on a filesystem watcher to trigger job processing without explicit API calls. This design enables integration with external systems that simply drop job manifests into a monitored directory.

### Async Watcher Implementation

In [`api/routers/batch.py`](https://github.com/debpalash/VoiceStudio/blob/main/api/routers/batch.py), the watcher runs as a background task during FastAPI startup:

```python

# api/routers/batch.py (simplified)

async def _watch_progress_folder():
    async for event in awatch(PROGRESS_ROOT):
        if event.is_dir_created:
            await _enqueue_job(event.path)

```

The watcher uses an asyncio-compatible observer (such as `watchdog` with async support) to detect directory creation events in real time.

### Job Enqueueing

Once a folder appears, `_enqueue_job` loads the manifest and creates a prioritized task:

```python
async def _enqueue_job(job_path: Path):
    job = await _load_job_manifest(job_path)
    task = Task(
        job_id=job.id,
        priority=PriorityClass.BATCH,
        payload=job
    )
    task_store.create(task, now=time.time())

```

By default, batch tasks receive `PriorityClass.BATCH`, which ranks below `PriorityClass.INTERACTIVE`. This ensures that on-demand requests are prioritized over background workloads.

---

## Task Priority and Scheduling

The `TaskStore` manages queue ordering based on priority classes. The implementation in [`api/routers/batch.py`](https://github.com/debpalash/VoiceStudio/blob/main/api/routers/batch.py) → `TaskStore.create` enforces strict precedence rules.

Verified behavior from [`tests/test_worker_scheduler.py`](https://github.com/debpalash/VoiceStudio/blob/main/tests/test_worker_scheduler.py):

- Batch tasks enter the queue with status `QUEUED`
- Interactive tasks always outrank batch tasks when workers poll for work
- Priority inversion is prevented through timestamp-based tiebreaking

This guarantees responsive user-facing generation even under heavy batch load.

---

## Pinned-Voice Routing and Engine Resolution

Users can pin a job to a specific voice engine via the `engine_id` field in the job manifest. The resolution logic in `_resolve_engine` balances user preference against technical capability.

### Engine Registry Lookup

The global `_REGISTRY` in [`api/engines/_registry.py`](https://github.com/debpalash/VoiceStudio/blob/main/api/engines/_registry.py) contains all installed voice backends. Resolution follows this order:

1. Check if `engine_id` is specified and registered
2. Verify `engine.supports_batch` property
3. If incompatible, fall back through the chain

```python
async def _resolve_engine(task: Task) -> Engine:
    pinned = task.payload.get("engine_id")
    if pinned and pinned in _REGISTRY:
        engine = _REGISTRY[pinned]
        if engine.supports_batch:
            return engine
        # Incompatible → proceed to fallback

    return _REGISTRY.get(DEFAULT_BATCH_ENGINE) or _fallback_engine()

```

---

## Incompatible-Engine Fallback Strategies

When a pinned engine cannot handle batch operations, the queue implements a two-tier fallback in `_run_batch_pipeline` (located in [`api/routers/batch.py`](https://github.com/debpalash/VoiceStudio/blob/main/api/routers/batch.py)):

### Tier 1: Native Batch Engine

If `DEFAULT_BATCH_ENGINE` exists in `_REGISTRY` and reports `supports_batch=True`, the job is transparently redirected:

```python
engine = _REGISTRY.get(pinned_id)
if engine and not engine.supports_batch:
    logger.warning("Pinned engine %s lacks batch support – using fallback.", pinned_id)
    engine = _REGISTRY.get(DEFAULT_BATCH_ENGINE)

```

This fallback is exercised in [`tests/test_text_normalization_routes.py`](https://github.com/debpalash/VoiceStudio/blob/main/tests/test_text_normalization_routes.py), which verifies that batch width is respected (e.g., 8 segments then 2 segments for a 10-segment job).

### Tier 2: Per-Segment Generation

If no native batch engine is available, the pipeline decomposes the job into individual generation calls:

```python
async def _run_batch_pipeline(job_id: str, job: Job):
    engine = await _resolve_engine(task)
    if engine.supports_batch:
        await engine.generate_batch(job.segments, **job.opts)
    else:
        # Per-segment fallback

        for seg in job.segments:
            await engine.generate(seg, **job.opts)

```

**Performance implication:** Per-segment generation incurs higher overhead but guarantees completion. The [`tests/test_perf_operation_budgets.py`](https://github.com/debpalash/VoiceStudio/blob/main/tests/test_perf_operation_budgets.py) suite validates that native batch engines never trigger this slower path unexpectedly.

---

## Post-Processing and Watermarking

After audio generation completes, the queue handles:

- **File placement:** Results written to `job_path / "dubbed_<lang>.wav"`
- **Watermark injection:** If the `marking` flag is set, [`api/routers/watermark.py`](https://github.com/debpalash/VoiceStudio/blob/main/api/routers/watermark.py) processes the audio

The watermark integration is verified in [`tests/test_synthetic_audio_watermark_1169.py`](https://github.com/debpalash/VoiceStudio/blob/main/tests/test_synthetic_audio_watermark_1169.py), confirming that batch-dubbed tracks carry correct synthetic audio markers.

---

## Complete Workflow Example

### Client: Creating a Batch Job

```python
import httpx, pathlib, json

payload = {
    "text": ["Hello", "world", "..."],
    "engine_id": "my-pinned-engine",  # optional; may be incompatible

    "options": {"duration": [1.0, 1.0, 1.0]}
}

job_dir = pathlib.Path("/tmp/progress") / "job123"
job_dir.mkdir(parents=True)
(job_dir / "manifest.json").write_text(json.dumps(payload))

# Server's folder watcher detects job123 automatically

```

### Server: Fallback Handling

```python

# Excerpt from api/routers/batch.py

engine = _REGISTRY.get(pinned_id)
if engine and not engine.supports_batch:
    logger.warning("Pinned engine %s lacks batch support – using fallback.", pinned_id)
    engine = _REGISTRY.get(DEFAULT_BATCH_ENGINE)

```

---

## Key Source Files

| File | Purpose |
|------|---------|
| [`api/routers/batch.py`](https://github.com/debpalash/VoiceStudio/blob/main/api/routers/batch.py) | Main batch-queue implementation, folder watcher, task enqueuing, engine resolution, fallback logic |
| [`api/engines/_registry.py`](https://github.com/debpalash/VoiceStudio/blob/main/api/engines/_registry.py) | Global voice-engine registry (`_REGISTRY`) |
| [`tests/test_worker_scheduler.py`](https://github.com/debpalash/VoiceStudio/blob/main/tests/test_worker_scheduler.py) | Priority queue behavior and task state verification |
| [`tests/test_text_normalization_routes.py`](https://github.com/debpalash/VoiceStudio/blob/main/tests/test_text_normalization_routes.py) | Pinned-engine selection and native-batch fallback testing |
| [`tests/test_perf_operation_budgets.py`](https://github.com/debpalash/VoiceStudio/blob/main/tests/test_perf_operation_budgets.py) | Validates native-batch operation budgets |
| [`tests/test_synthetic_audio_watermark_1169.py`](https://github.com/debpalash/VoiceStudio/blob/main/tests/test_synthetic_audio_watermark_1169.py) | Confirms watermark application on batch output |

---

## Summary

- **Folder watching** uses an async filesystem observer in [`api/routers/batch.py`](https://github.com/debpalash/VoiceStudio/blob/main/api/routers/batch.py) to trigger job ingestion without explicit API calls
- **Task priority** separates interactive and batch workloads, with `INTERACTIVE` always outranking `BATCH`
- **Pinned-voice routing** respects user preference through `engine_id` metadata, with capability checks against `_REGISTRY`
- **Incompatible-engine fallback** first attempts `DEFAULT_BATCH_ENGINE`, then degrades to per-segment generation if necessary
- **Graceful degradation** ensures job completion regardless of engine compatibility, verified across multiple test suites

---

## Frequently Asked Questions

### What triggers the batch queue to start processing a job?

The folder watcher in [`api/routers/batch.py`](https://github.com/debpalash/VoiceStudio/blob/main/api/routers/batch.py) detects new directory creation events in the `progress/` folder. When a job folder appears, `_enqueue_job` is invoked automatically—no HTTP request required. This enables drop-in integration with external job submission systems.

### How does the queue handle a pinned engine that cannot process batches?

The `_resolve_engine` function checks `engine.supports_batch`. If false, it logs a warning and falls back to `DEFAULT_BATCH_ENGINE`. If no batch-capable engine exists, `_run_batch_pipeline` switches to per-segment generation. Both paths guarantee job completion.

### Why are there two priority classes for tasks?

`INTERACTIVE` and `BATCH` priority classes ensure responsive user-facing generation. Tests in [`tests/test_worker_scheduler.py`](https://github.com/debpalash/VoiceStudio/blob/main/tests/test_worker_scheduler.py) confirm that interactive tasks are always serviced before batch tasks, preventing latency spikes for on-demand requests during heavy background workloads.

### Where is the fallback behavior tested?

Three test files cover fallback scenarios: [`tests/test_text_normalization_routes.py`](https://github.com/debpalash/VoiceStudio/blob/main/tests/test_text_normalization_routes.py) verifies engine selection and native-batch width, [`tests/test_perf_operation_budgets.py`](https://github.com/debpalash/VoiceStudio/blob/main/tests/test_perf_operation_budgets.py) ensures budgeted operation counts, and [`tests/test_worker_scheduler.py`](https://github.com/debpalash/VoiceStudio/blob/main/tests/test_worker_scheduler.py) validates priority ordering throughout the fallback chain.