lautaromvillalba/ruistper repository overview

src/main.rs
  #[tokio::main]; tracing_subscriber init; app::run()

src/app.rs
  async fn run() → Result<(), AppError>
  Startup sequence: Config → tmp_dir → Metrics → shutdown pair
    → RabbitMQ pool (workers+2) → WhisperModel (blocking, once)
    → WhisperEngine → RabbitProducer → RabbitConsumer → WorkerPool

src/config.rs
  struct Config { rabbitmq_url, workers_count, whisper_model,
                  whisper_device, models_dir, max_file_size_mb,
                  max_audio_duration_sec, audio_sample_rate,
                  tmp_dir, api_host, api_port }
  Config::load() → reads env vars, validates
  Config::model_path() → PathBuf  ({models_dir}/ggml-{whisper_model}.bin)
  Config::max_file_size_bytes() → u64
  Config::log_summary()

src/retry.rs
  const MAX_RETRIES: i32 = 2   // 3 total attempts
  enum RetryDecision { Retry, GiveUp }
  RetryPolicy::decide(retry_count) → RetryDecision

src/metrics.rs
  struct Metrics {
    received:  AtomicU64,
    succeeded: AtomicU64,
    failed:    AtomicU64,
    retried:   AtomicU64,
    in_flight: AtomicI64,
  }
  Arc<Metrics> shared across workers

src/shutdown.rs
  fn new_pair() → (ShutdownHandle, ShutdownSignal)
  fn wait_for_os_signal()   // SIGINT + SIGTERM

src/model/request.rs
  struct TranscriptionRequest {
    attachment_id:   i64,
    audio_file_path: String,
    language:        Option<String>,
    import_batch_id: Option<i64>,
    retry_count:     i32,
  }

src/model/result.rs
  struct TranscriptionResult {
    attachment_id:   i64,
    texto:           String,
    duration:        f64,
    model:           String,
    success:         bool,
    import_batch_id: Option<i64>,
    error_message:   Option<String>,  // skip_serializing_if = None
  }
  TranscriptionResult::success(attachment_id, texto, duration, model, import_batch_id) → Self
  TranscriptionResult::failure(attachment_id, model, import_batch_id, error_message)   → Self

src/audio/decoder.rs
  fn decode(path: &Path) → Result<DecodedAudio, DecodeError>
  fn is_ogg_opus(path) → bool   // reads first 128 bytes for "OpusHead" magic
  OGG/Opus path → decode_ogg_opus() via libopus
  All other formats → Symphonia probe/decode loop

src/audio/resampler.rs
  fn to_mono_16k(decoded: &DecodedAudio) → Vec<f32>   // rubato SincFixedIn

src/audio/pipeline.rs
  struct AudioLimits { max_file_size_bytes: u64, max_duration_secs: f64 }
  struct ProcessedAudio { samples: Vec<f32>, duration_secs: f64 }
  fn process(path: &Path, limits: AudioLimits) → Result<ProcessedAudio, AudioError>
  Validation order: exists+size → extension → decode → duration → resample
  enum AudioError {
    FileNotFound(String),            // Deterministic
    UnsupportedFormat(String),       // Deterministic
    FileTooLarge { size_bytes, limit_bytes },    // Deterministic
    DurationExceeded { duration_secs, limit_secs }, // Deterministic
    Decode(DecodeError),             // Transient
    Resample(ResampleError),         // Transient
  }

src/whisper/model.rs
  struct WhisperModel { context: WhisperContext, name: String }
  unsafe impl Send + Sync   // required; context is not Send by default
  WhisperModel::load(path: &Path, name: String) → Result<Self, ModelError>

src/whisper/engine.rs
  #[derive(Clone)]
  struct WhisperEngine { model: Arc<WhisperModel> }
  WhisperEngine::new(model: Arc<WhisperModel>) → Self
  WhisperEngine::model_name() → &str
  WhisperEngine::transcribe(&self, samples: &[f32], language: Option<&str>)
    → Result<TranscriptionOutput, EngineError>
  Params: BeamSearch { beam_size: 5 }, no_speech_thold = 0.6
  WhisperState: allocated fresh per call (not reused across calls)
  struct TranscriptionOutput { text: String }
  enum EngineError { StateCreation(String), Inference(String), SegmentRead(String) }

src/messaging/rabbit.rs
  pub type Pool = deadpool_lapin::Pool
  fn build_pool(url: &str, max_size: usize) → Result<Pool, RabbitError>
  // Exchange / queue / routing-key string constants:
  WHISPER_EXCHANGE       = "whisper_exchange"
  WHISPER_QUEUE          = "whisper_transcriptions"
  WHISPER_ROUTING_KEY    = "transcription.request"
  RESULTS_EXCHANGE       = "whisper_results_exchange"
  RESULTS_QUEUE          = "whisper_results"
  RESULTS_ROUTING_KEY    = "transcription.result"
  RETRY_EXCHANGE         = "whisper_retry_exchange"
  RETRY_QUEUE            = "whisper_retry_queue"
  RETRY_ROUTING_KEY      = "transcription.retry"
  RETRY_TTL_MS           = 5000

src/messaging/consumer.rs
  struct Job { request: TranscriptionRequest, delivery: Delivery }
  struct RabbitConsumer
  RabbitConsumer::new(pool: &Pool, prefetch: u16) → Result<Self, ConsumerError>
  RabbitConsumer::into_receiver(self) → Result<mpsc::Receiver<Job>, ConsumerError>
  // Invalid JSON → NACK(requeue=false); valid → forwarded to channel

src/messaging/producer.rs
  #[derive(Clone)]
  struct RabbitProducer { pool: Pool, model_name: String }
  RabbitProducer::new(pool: &Pool, model_name: String) → Result<Self, ProducerError>
  RabbitProducer::publish_success(result: TranscriptionResult) → Result<(), ProducerError>
  RabbitProducer::publish_error(result: TranscriptionResult)   → Result<(), ProducerError>
  RabbitProducer::publish_retry(request: TranscriptionRequest) → Result<(), ProducerError>
  // publish_retry: increments retry_count; sets x-retry-count AMQP header

src/worker/task.rs
  async fn process(worker_id: usize, job: Job, engine: WhisperEngine,
                   producer: RabbitProducer, limits: AudioLimits,
                   metrics: Arc<Metrics>)
  enum TaskError { Deterministic(String), Transient(String) }
  // Deterministic → publish_error + ACK (no retry)
  // Transient     → handle_failure (retry via publish_retry, or final publish_error + ACK)
  // JoinError (panic in spawn_blocking) → treated as Transient
  // Any publish failure → NACK(requeue=true)
  // Blocking work dispatched via tokio::task::spawn_blocking:
  //   audio::pipeline::process → WhisperEngine::transcribe

src/worker/pool.rs
  struct WorkerPool { workers_count: usize, engine: WhisperEngine,
                      producer: RabbitProducer, limits: AudioLimits,
                      metrics: Arc<Metrics> }
  WorkerPool::run(self, jobs_rx: mpsc::Receiver<Job>, shutdown_signal: ShutdownSignal)
  // N worker tasks share Arc<Mutex<Receiver>>
  // Channel capacity = workers_count × 2

INPUT
  exchange     "whisper_exchange"           type=direct, durable
  queue        "whisper_transcriptions"     durable
  routing_key  "transcription.request"

OUTPUT
  exchange     "whisper_results_exchange"   type=direct, durable
  queue        "whisper_results"            durable
  routing_key  "transcription.result"

RETRY (Dead-Letter Exchange pattern)
  exchange     "whisper_retry_exchange"     type=direct, durable
  queue        "whisper_retry_queue"        durable, x-message-ttl=5000ms
  routing_key  "transcription.retry"
  DLX on TTL expiry → "whisper_exchange"   (message re-enters main queue)

  MAX_RETRIES = 2  →  3 total processing attempts

{
  "attachment_id":   123,
  "audio_file_path": "/mnt/audio/voice_001.ogg",
  "language":        "es",
  "import_batch_id": 42,
  "retry_count":     0
}

{
  "attachment_id":   123,
  "texto":           "Hola, ¿cómo estás?",
  "duration":        4.32,
  "model":           "base-q5_1",
  "success":         true,
  "import_batch_id": 42
}

{
  "attachment_id":   123,
  "texto":           "",
  "duration":        0.0,
  "model":           "base-q5_1",
  "success":         false,
  "import_batch_id": 42,
  "error_message":   "file not found: /mnt/audio/voice_001.ogg"
}

Builder stage:  rust:1-bookworm  +  libopus-dev  +  libclang-dev  +  cmake
Runtime stage:  debian:bookworm-slim  +  libopus0  +  libgomp1  +  libstdc++6

Dep-cache trick:
  1. COPY Cargo.toml + stub main.rs → cargo build --release  (caches all deps)
  2. COPY src/ → cargo build --release  (only recompiles ruistper itself)

After modifying src/:
  docker-compose build --no-cache ruistper
  (required to bypass the COPY src/ layer cache)

┌──────────────────────────────────────────────────────────────┐
│                          Ruistper                            │
│                                                              │
│  RabbitMQ ──► Consumer ──► WorkerPool ──► Producer ──► RabbitMQ
│               │             │    │                          │
│               │          N workers                         │
│               │          ┌──────────────────────────────┐  │
│               │          │  AudioPipeline               │  │
│               │          │  ├─ validate (size/duration) │  │
│               │          │  ├─ decode (Symphonia/Opus)  │  │
│               │          │  └─ resample → mono 16 kHz   │  │
│               │          │                              │  │
│               │          │  WhisperEngine               │  │
│               │          │  └─ whisper.cpp (GGML)       │  │
│               └──────────└──────────────────────────────┘  │
└──────────────────────────────────────────────────────────────┘

{
  "attachment_id":   123,
  "audio_file_path": "/mnt/audio/voice_001.ogg",
  "language":        "es",
  "import_batch_id": 42,
  "retry_count":     0
}

{
  "attachment_id":   123,
  "texto":           "Hola, ¿cómo estás?",
  "duration":        4.32,
  "model":           "base-q5_1",
  "success":         true,
  "import_batch_id": 42
}

{
  "attachment_id":   123,
  "texto":           "",
  "duration":        0.0,
  "model":           "base-q5_1",
  "success":         false,
  "import_batch_id": 42,
  "error_message":   "file not found: /mnt/audio/voice_001.ogg"
}

Main queue ──► [transient failure] ──► whisper_retry_exchange
                                             │
                                        whisper_retry_queue  (TTL = 5 s)
                                             │  x-dead-letter-exchange
                                             └──► whisper_exchange  (re-queued)

# First run — builds the image, downloads the model, starts services
docker-compose up --build

# Rebuild after source changes (force-bypass layer cache)
docker-compose build --no-cache ruistper
docker-compose up

https://huggingface.co/ggerganov/whisper.cpp/resolve/main/ggml-{WHISPER_MODEL}.bin

mkdir -p models
MODEL=base-q5_1
curl -L "https://huggingface.co/ggerganov/whisper.cpp/resolve/main/ggml-${MODEL}.bin" \
     -o "models/ggml-${MODEL}.bin"

services:

  rabbitmq:
    image: rabbitmq:3-management
    container_name: rabbitmq
    ports:
      - "5672:5672"
      - "15672:15672"   # management UI → http://localhost:15672
    environment:
      RABBITMQ_DEFAULT_USER: admin
      RABBITMQ_DEFAULT_PASS: admin
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "ping"]
      interval: 10s
      timeout: 5s
      retries: 10

  ruistper:
    build:
      context: ./Ruistper
      dockerfile: Dockerfile
    container_name: ruistper
    depends_on:
      rabbitmq:
        condition: service_healthy
    volumes:
      - ./models:/app/models   # persists downloaded GGML models
    environment:
      RABBITMQ_URL: amqp://admin:admin@rabbitmq:5672/%2F
      RABBITMQ_QUEUE_NAME: whisper_transcriptions
      WORKERS_COUNT: "4"
      WHISPER_MODEL: base-q5_1
      WHISPER_DEVICE: cpu
      MODELS_DIR: /app/models
      MAX_FILE_SIZE_MB: "100"
      MAX_AUDIO_DURATION_SEC: "3600"
      TMP_DIR: /tmp/whisper
      RUST_LOG: ruistper=info
    restart: unless-stopped

cd Ruistper
cargo build --release
# binary: ./target/release/ruistper