Add more workers and update structure #46

New issue

Open

opened 2026-06-25 00:18:32 +00:00 by theis · 0 comments

theis commented 2026-06-25 00:18:32 +00:00

Owner

Copy link

name	overview	todos	isProject
Scale TTS Workers	Split TTS job processing into a dedicated worker service (separate binary and Docker image) that scales independently from the HTTP API. Postgres is the durable job queue with LISTEN/NOTIFY wake-up. Each TTS node runs one model, processes one synthesis at a time via an internal FIFO queue, and scales horizontally across GPUs.	id content status worker-package Create backend/internal/worker/ package: config, pool, lifecycle; keep TTS domain logic in internal/tts/ pending id content status worker-binary Create backend/cmd/worker/main.go with graceful shutdown; remove worker startup from cmd/web/main.go; API-only runs migrations pending id content status worker-docker Add backend/Dockerfile.worker and backend-worker compose service with separate webspeaker-worker image pending id content status listen-notify Add Postgres LISTEN/NOTIFY on job enqueue to wake workers without polling loop pending id content status busy-retry Add ReleaseTTSJob for transient TTS failures; requeue instead of marking failed pending id content status stale-sweeper Add periodic stale-job sweeper with segment cleanup on stale requeue pending id content status tts-queue Replace TTS _synthesis_lock busy-reject with internal FIFO queue (one model, serialized synthesis) pending id content status tests Add concurrency tests for ClaimNextQueuedTTSJob, NOTIFY wake-up, and integration test with mock TTS pending	false

Scale TTS Processing Across Multiple Workers and TTS Nodes

Current state

flowchart LR
  Client -->|POST job| API
  API -->|INSERT queued| DB[(Postgres)]
  Worker -->|claim SKIP LOCKED| DB
  Worker -->|POST SSE| TTS
  TTS -->|segments| Worker
  Worker -->|INSERT segments| DB
  Client -->|poll status| API

Ready for scaling:

• Job queue in backend/db/migrations/009_create_tts_jobs.up.sql with index (status, created_at)
• ClaimNextQueuedTTSJob uses FOR UPDATE SKIP LOCKED
• Segment upsert on (job_id, segment_index) is idempotent

Bottlenecks today:

• One worker goroutine in backend/cmd/web/main.go
• Hardcoded TTS URL in backend/internal/tts/worker.go
• TTS rejects concurrent requests via _synthesis_lock in tts/main.py
• Worker polls DB every 2s when idle

---

Architecture

Separate the API (enqueue + poll) from workers (claim + synthesize + store). Scale each layer independently.

flowchart TB
  subgraph clients [Clients]
    FE[Frontend]
  end

  subgraph api_layer [API - webspeaker-back]
    API[backend x N]
  end

  subgraph worker_layer [Workers - webspeaker-worker]
    W[backend-worker x M]
  end

  subgraph tts_layer [TTS nodes]
    LB[TTS_SERVICE_URL]
    TTS1[tts node 1]
    TTS2[tts node 2]
    TTSN[tts node N]
  end

  DB[(Postgres)]

  FE --> API
  API -->|INSERT + NOTIFY| DB
  W -->|LISTEN + claim| DB
  W --> LB
  LB --> TTS1
  LB --> TTS2
  LB --> TTSN

Throughput:

• 1 TTS node (1 GPU): ~1 synthesis at a time; internal queue absorbs burst
• N TTS nodes (N GPUs): ~N concurrent syntheses
• Worker concurrency: match TTS node count + small buffer

---

Design principles

Job queue: Postgres + LISTEN/NOTIFY

Postgres remains the durable job queue and source of truth. Workers claim jobs via ClaimNextQueuedTTSJob (SKIP LOCKED). No gRPC dispatch from the API — workers stay independent of API uptime.

Workers wake via Postgres LISTEN/NOTIFY instead of polling:

sequenceDiagram
  participant API
  participant DB as Postgres
  participant Worker

  API->>DB: INSERT job + NOTIFY tts_job_queued
  Worker->>DB: LISTEN tts_job_queued
  DB-->>Worker: notification
  Worker->>DB: ClaimNextQueuedTTSJob
  Worker->>Worker: process job

• CreateTTSJob / requeue paths emit NOTIFY tts_job_queued on commit
• Workers block on the channel; fallback timeout re-checks queue if NOTIFY is missed

Migrations: API only

• backend/cmd/web/main.go runs migrate.Up() on startup
• backend/cmd/worker/main.go connects to DB but does not migrate
• Worker Docker image does not ship migration files
• Deploy API before workers on schema changes

Code layout

backend/
  cmd/
    web/main.go          # API entrypoint, runs migrations
    worker/main.go       # Worker entrypoint
  internal/
    worker/
      config.go          # env-based WorkerConfig
      pool.go            # claim loops, LISTEN/NOTIFY, lifecycle
    tts/
      processor.go       # TTS HTTP/SSE call (refactor from worker.go)
      sse.go
    storage/postgres/
      tts_job_store.go   # claim, notify, release, stale sweep

Package	Responsibility
internal/worker	Config, pool lifecycle, LISTEN/NOTIFY, graceful shutdown, stale sweeper
internal/tts	TTS HTTP/SSE communication, segment decode/store
cmd/worker	Wiring only: config → DB pool → start pool → signal handling

Docker: separate images

Image	Dockerfile	Entrypoint	Notes
webspeaker-back:latest	backend/Dockerfile	cmd/web	Ships migrations, port 8080
webspeaker-worker:latest	backend/Dockerfile.worker	cmd/worker	No migrations, optional health port

Both build from the same Go module. Share builder-stage pattern to avoid drift.

TTS nodes: one model per container, one synthesis at a time

Each TTS container/node loads one model instance and processes one synthesis at a time. Spare VRAM (e.g. on a 16GB GPU with a ~5–8GB 1.7B bf16 model) is activation headroom, not used for duplicate model copies.

• Within a node: internal FIFO queue accepts waiting requests; synthesize serially on the single model
• Across nodes: one TTS container per GPU, pinned via CUDA_VISIBLE_DEVICES; load-balance via TTS_SERVICE_URL
• Do not: load multiple model copies on the same GPU — shared compute and memory bandwidth yield poor throughput and OOM risk

flowchart TB
  subgraph tts_node [TTS node - 1 GPU]
    Q[Internal FIFO queue]
    M[One model]
    Q --> M
  end

  subgraph cluster [TTS cluster]
    LB[Load balancer]
    N1[Node 1 - 1 model]
    N2[Node 2 - 1 model]
    LB --> N1
    LB --> N2
  end

Future optimization (optional): profile GPU utilization; if consistently low, investigate segment batching within a single model before any architectural changes.

---

Phase 1 — Worker package and binary split

Config (internal/worker/config.go)

Env var	Default	Purpose
TTS_SERVICE_URL	http://tts:8000/tts/stream	TTS load balancer URL
TTS_WORKER_CONCURRENCY	1	Claim loops per worker process
TTS_NOTIFY_CHANNEL	tts_job_queued	Postgres NOTIFY channel
TTS_NOTIFY_FALLBACK	30s	Max wait before re-checking queue
TTS_STALE_JOB_AFTER	30m	Requeue stuck running jobs
TTS_BUSY_RETRY_MAX	5	Retries on transient TTS failure
TTS_BUSY_RETRY_BACKOFF	2s	Backoff between retries

cmd/worker/main.go

• Load config + DB pool
• Start worker pool
• Graceful shutdown on SIGTERM

cmd/web/main.go

• Remove worker goroutine (line 62)
• Keep migrations only here

Compose

• Add backend-worker service using webspeaker-worker:latest

---

Phase 2 — Worker pool with LISTEN/NOTIFY

In internal/worker/pool.go:

1. LISTEN tts_job_queued on startup
2. On NOTIFY or fallback timeout: ClaimNextQueuedTTSJob
3. If claimed: internal/tts.Processor.ProcessJob(ctx, job)
4. If nil: wait for next NOTIFY

In tts_job_store.go, emit on enqueue/requeue:

NOTIFY tts_job_queued, '<job_id>';

---

Phase 3 — TTS internal queue

In tts/main.py:

• Replace _synthesis_lock busy-reject with an internal FIFO queue
• /tts/stream enqueues the request; one consumer calls provider.synthesize() at a time
• Expose queue depth on /health for observability

Multi-GPU: deploy one TTS replica per GPU, each with its own queue and model, behind TTS_SERVICE_URL.

---

Phase 4 — Resilience

• ReleaseTTSJob: running → queued on transient TTS connection failure so another worker/node can retry
• Stale-job sweeper: periodic ticker; requeue jobs stuck in running past TTS_STALE_JOB_AFTER; delete partial segments
• Context propagation: pass worker ctx into TTS HTTP call for graceful shutdown

---

Phase 5 — Observability

• Structured logs: job_id, worker_id, queue_wait, synthesis_duration
• Metrics: tts_jobs_queued, tts_jobs_running, tts_queue_depth
• Worker health: DB ping + TTS /health

---

Rollout order

1. internal/worker + cmd/worker + Dockerfile.worker (concurrency=1, poll-based initially)
2. Remove worker from API; verify end-to-end
3. Add LISTEN/NOTIFY
4. TTS internal queue
5. Increase worker concurrency; add TTS nodes per GPU
6. Stale sweeper + observability

---

Unchanged

• Frontend polling (frontend/src/pages/AudioTest/index.tsx)
• REST handlers (backend/internal/handlers/tts_jobs.go)
• Claim SQL (FOR UPDATE SKIP LOCKED)
• SSE protocol (backend/internal/tts/sse.go)

---

Testing

• Concurrent ClaimNextQueuedTTSJob from N goroutines → distinct jobs
• NOTIFY wakes worker without polling
• Enqueue N jobs → worker pool + mock TTS → all complete
• Multiple simultaneous /tts/stream connections → queued, not rejected

---

Risks

Risk	Mitigation
Worker concurrency >> TTS capacity	Internal TTS queue; size workers to ~TTS node count
NOTIFY missed while worker down	Fallback timeout re-checks queue
Stale running after crash	Periodic stale sweeper + segment cleanup
Schema change before worker deploy	Deploy API (migrations) first
Single GPU throughput ceiling	Add TTS nodes (one per GPU)

--- name: Scale TTS Workers overview: Split TTS job processing into a dedicated worker service (separate binary and Docker image) that scales independently from the HTTP API. Postgres is the durable job queue with LISTEN/NOTIFY wake-up. Each TTS node runs one model, processes one synthesis at a time via an internal FIFO queue, and scales horizontally across GPUs. todos: - id: worker-package content: "Create backend/internal/worker/ package: config, pool, lifecycle; keep TTS domain logic in internal/tts/" status: pending - id: worker-binary content: Create backend/cmd/worker/main.go with graceful shutdown; remove worker startup from cmd/web/main.go; API-only runs migrations status: pending - id: worker-docker content: Add backend/Dockerfile.worker and backend-worker compose service with separate webspeaker-worker image status: pending - id: listen-notify content: Add Postgres LISTEN/NOTIFY on job enqueue to wake workers without polling loop status: pending - id: busy-retry content: "Add ReleaseTTSJob for transient TTS failures; requeue instead of marking failed" status: pending - id: stale-sweeper content: Add periodic stale-job sweeper with segment cleanup on stale requeue status: pending - id: tts-queue content: Replace TTS _synthesis_lock busy-reject with internal FIFO queue (one model, serialized synthesis) status: pending - id: tests content: Add concurrency tests for ClaimNextQueuedTTSJob, NOTIFY wake-up, and integration test with mock TTS status: pending isProject: false --- # Scale TTS Processing Across Multiple Workers and TTS Nodes ## Current state ```mermaid flowchart LR Client -->|POST job| API API -->|INSERT queued| DB[(Postgres)] Worker -->|claim SKIP LOCKED| DB Worker -->|POST SSE| TTS TTS -->|segments| Worker Worker -->|INSERT segments| DB Client -->|poll status| API ``` **Ready for scaling:** - Job queue in [`backend/db/migrations/009_create_tts_jobs.up.sql`](backend/db/migrations/009_create_tts_jobs.up.sql) with index `(status, created_at)` - [`ClaimNextQueuedTTSJob`](backend/internal/storage/postgres/tts_job_store.go) uses `FOR UPDATE SKIP LOCKED` - Segment upsert on `(job_id, segment_index)` is idempotent **Bottlenecks today:** - One worker goroutine in [`backend/cmd/web/main.go`](backend/cmd/web/main.go) - Hardcoded TTS URL in [`backend/internal/tts/worker.go`](backend/internal/tts/worker.go) - TTS rejects concurrent requests via `_synthesis_lock` in [`tts/main.py`](tts/main.py) - Worker polls DB every 2s when idle --- ## Architecture Separate the API (enqueue + poll) from workers (claim + synthesize + store). Scale each layer independently. ```mermaid flowchart TB subgraph clients [Clients] FE[Frontend] end subgraph api_layer [API - webspeaker-back] API[backend x N] end subgraph worker_layer [Workers - webspeaker-worker] W[backend-worker x M] end subgraph tts_layer [TTS nodes] LB[TTS_SERVICE_URL] TTS1[tts node 1] TTS2[tts node 2] TTSN[tts node N] end DB[(Postgres)] FE --> API API -->|INSERT + NOTIFY| DB W -->|LISTEN + claim| DB W --> LB LB --> TTS1 LB --> TTS2 LB --> TTSN ``` **Throughput:** - 1 TTS node (1 GPU): ~1 synthesis at a time; internal queue absorbs burst - N TTS nodes (N GPUs): ~N concurrent syntheses - Worker concurrency: match TTS node count + small buffer --- ## Design principles ### Job queue: Postgres + LISTEN/NOTIFY Postgres remains the durable job queue and source of truth. Workers claim jobs via `ClaimNextQueuedTTSJob` (`SKIP LOCKED`). No gRPC dispatch from the API — workers stay independent of API uptime. Workers wake via Postgres `LISTEN/NOTIFY` instead of polling: ```mermaid sequenceDiagram participant API participant DB as Postgres participant Worker API->>DB: INSERT job + NOTIFY tts_job_queued Worker->>DB: LISTEN tts_job_queued DB-->>Worker: notification Worker->>DB: ClaimNextQueuedTTSJob Worker->>Worker: process job ``` - `CreateTTSJob` / requeue paths emit `NOTIFY tts_job_queued` on commit - Workers block on the channel; fallback timeout re-checks queue if NOTIFY is missed ### Migrations: API only - [`backend/cmd/web/main.go`](backend/cmd/web/main.go) runs `migrate.Up()` on startup - [`backend/cmd/worker/main.go`](backend/cmd/worker/main.go) connects to DB but does not migrate - Worker Docker image does not ship migration files - Deploy API before workers on schema changes ### Code layout ``` backend/ cmd/ web/main.go # API entrypoint, runs migrations worker/main.go # Worker entrypoint internal/ worker/ config.go # env-based WorkerConfig pool.go # claim loops, LISTEN/NOTIFY, lifecycle tts/ processor.go # TTS HTTP/SSE call (refactor from worker.go) sse.go storage/postgres/ tts_job_store.go # claim, notify, release, stale sweep ``` | Package | Responsibility | |---------|----------------| | `internal/worker` | Config, pool lifecycle, LISTEN/NOTIFY, graceful shutdown, stale sweeper | | `internal/tts` | TTS HTTP/SSE communication, segment decode/store | | `cmd/worker` | Wiring only: config → DB pool → start pool → signal handling | ### Docker: separate images | Image | Dockerfile | Entrypoint | Notes | |-------|------------|------------|-------| | `webspeaker-back:latest` | [`backend/Dockerfile`](backend/Dockerfile) | `cmd/web` | Ships migrations, port 8080 | | `webspeaker-worker:latest` | `backend/Dockerfile.worker` | `cmd/worker` | No migrations, optional health port | Both build from the same Go module. Share builder-stage pattern to avoid drift. ### TTS nodes: one model per container, one synthesis at a time Each TTS container/node loads **one model instance** and processes **one synthesis at a time**. Spare VRAM (e.g. on a 16GB GPU with a ~5–8GB 1.7B bf16 model) is activation headroom, not used for duplicate model copies. - **Within a node:** internal FIFO queue accepts waiting requests; synthesize serially on the single model - **Across nodes:** one TTS container per GPU, pinned via `CUDA_VISIBLE_DEVICES`; load-balance via `TTS_SERVICE_URL` - **Do not:** load multiple model copies on the same GPU — shared compute and memory bandwidth yield poor throughput and OOM risk ```mermaid flowchart TB subgraph tts_node [TTS node - 1 GPU] Q[Internal FIFO queue] M[One model] Q --> M end subgraph cluster [TTS cluster] LB[Load balancer] N1[Node 1 - 1 model] N2[Node 2 - 1 model] LB --> N1 LB --> N2 end ``` Future optimization (optional): profile GPU utilization; if consistently low, investigate segment batching within a single model before any architectural changes. --- ## Phase 1 — Worker package and binary split ### Config (`internal/worker/config.go`) | Env var | Default | Purpose | |---------|---------|---------| | `TTS_SERVICE_URL` | `http://tts:8000/tts/stream` | TTS load balancer URL | | `TTS_WORKER_CONCURRENCY` | `1` | Claim loops per worker process | | `TTS_NOTIFY_CHANNEL` | `tts_job_queued` | Postgres NOTIFY channel | | `TTS_NOTIFY_FALLBACK` | `30s` | Max wait before re-checking queue | | `TTS_STALE_JOB_AFTER` | `30m` | Requeue stuck `running` jobs | | `TTS_BUSY_RETRY_MAX` | `5` | Retries on transient TTS failure | | `TTS_BUSY_RETRY_BACKOFF` | `2s` | Backoff between retries | ### `cmd/worker/main.go` - Load config + DB pool - Start worker pool - Graceful shutdown on `SIGTERM` ### `cmd/web/main.go` - Remove worker goroutine (line 62) - Keep migrations only here ### Compose - Add `backend-worker` service using `webspeaker-worker:latest` --- ## Phase 2 — Worker pool with LISTEN/NOTIFY In `internal/worker/pool.go`: 1. `LISTEN tts_job_queued` on startup 2. On NOTIFY or fallback timeout: `ClaimNextQueuedTTSJob` 3. If claimed: `internal/tts.Processor.ProcessJob(ctx, job)` 4. If nil: wait for next NOTIFY In [`tts_job_store.go`](backend/internal/storage/postgres/tts_job_store.go), emit on enqueue/requeue: ```sql NOTIFY tts_job_queued, '<job_id>'; ``` --- ## Phase 3 — TTS internal queue In [`tts/main.py`](tts/main.py): - Replace `_synthesis_lock` busy-reject with an internal FIFO queue - `/tts/stream` enqueues the request; one consumer calls `provider.synthesize()` at a time - Expose queue depth on `/health` for observability Multi-GPU: deploy one TTS replica per GPU, each with its own queue and model, behind `TTS_SERVICE_URL`. --- ## Phase 4 — Resilience - **`ReleaseTTSJob`:** `running → queued` on transient TTS connection failure so another worker/node can retry - **Stale-job sweeper:** periodic ticker; requeue jobs stuck in `running` past `TTS_STALE_JOB_AFTER`; delete partial segments - **Context propagation:** pass worker ctx into TTS HTTP call for graceful shutdown --- ## Phase 5 — Observability - Structured logs: `job_id`, `worker_id`, `queue_wait`, `synthesis_duration` - Metrics: `tts_jobs_queued`, `tts_jobs_running`, `tts_queue_depth` - Worker health: DB ping + TTS `/health` --- ## Rollout order 1. `internal/worker` + `cmd/worker` + `Dockerfile.worker` (concurrency=1, poll-based initially) 2. Remove worker from API; verify end-to-end 3. Add LISTEN/NOTIFY 4. TTS internal queue 5. Increase worker concurrency; add TTS nodes per GPU 6. Stale sweeper + observability --- ## Unchanged - Frontend polling ([`frontend/src/pages/AudioTest/index.tsx`](frontend/src/pages/AudioTest/index.tsx)) - REST handlers ([`backend/internal/handlers/tts_jobs.go`](backend/internal/handlers/tts_jobs.go)) - Claim SQL (`FOR UPDATE SKIP LOCKED`) - SSE protocol ([`backend/internal/tts/sse.go`](backend/internal/tts/sse.go)) --- ## Testing - Concurrent `ClaimNextQueuedTTSJob` from N goroutines → distinct jobs - NOTIFY wakes worker without polling - Enqueue N jobs → worker pool + mock TTS → all complete - Multiple simultaneous `/tts/stream` connections → queued, not rejected --- ## Risks | Risk | Mitigation | |------|------------| | Worker concurrency >> TTS capacity | Internal TTS queue; size workers to ~TTS node count | | NOTIFY missed while worker down | Fallback timeout re-checks queue | | Stale `running` after crash | Periodic stale sweeper + segment cleanup | | Schema change before worker deploy | Deploy API (migrations) first | | Single GPU throughput ceiling | Add TTS nodes (one per GPU) |

theis added the

feature

label 2026-06-25 00:18:50 +00:00

Sign in to join this conversation.

qwen-tts

Branches Tags

Clear current reference

main

preview_player_ui

voice_management_settings

single_sign_on_integration

Clear current reference

0.5.1

0.5.0

0.4.0

0.3.0

0.2.92

0.2.91

0.2.9

0.2.8

0.2.7

0.2.51

0.2.5

0.2.4

0.2.3

0.2.2

0.2.1

0.2.0

Passkey-Base

0.1

Labels

Clear labels   

bug

Something that is broken and need fixing or misbehaving

  

Clean up

  

feature

  

improvment

No labels

bug

Clean up

feature

improvment

Milestone

Clear milestone

No items

No milestone

Projects

Clear projects

No items

No project

Assignees

Clear assignees

No assignees

1 participant

Notifications

Subscribe

Due date

The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set

Reference

theis/webspeaker#46

No description provided.