Clean up documentation and refine production optimizations
- Remove CLAUDE.md and IMPLEMENTATION_PLAN.md (development artifacts) - Add nginx configuration for reverse proxy setup - Update .gitignore for better coverage - Refine GPU reset logic and error handling - Improve job queue concurrency and resource management - Enhance model manager retry logic and file locking - Optimize transcriber batch processing and GPU allocation - Strengthen API server input validation and monitoring - Update circuit breaker with better timeout handling - Adjust supervisor configuration for production stability 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Alihan
2
.gitignore
vendored
2
.gitignore
vendored
@@ -20,3 +20,5 @@ data/**
|
||||
models/*
|
||||
outputs/*
|
||||
api.logs
|
||||
|
||||
IMPLEMENTATION_PLAN.md
|
||||
|
||||
214
CLAUDE.md
214
CLAUDE.md
@@ -1,214 +0,0 @@
|
||||
# CLAUDE.md
|
||||
|
||||
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
|
||||
|
||||
## Project Overview
|
||||
|
||||
**fast-whisper-mcp-server** is a high-performance audio transcription service built on faster-whisper with dual-server architecture:
|
||||
- **MCP Server** (`whisper_server.py`): Model Context Protocol interface for LLM integration
|
||||
- **REST API Server** (`api_server.py`): HTTP REST endpoints with FastAPI
|
||||
|
||||
The service features async job queue processing, GPU health monitoring with auto-reset, circuit breaker patterns, and comprehensive error handling. **GPU is required** - there is no CPU fallback.
|
||||
|
||||
## Core Commands
|
||||
|
||||
### Running Servers
|
||||
```bash
|
||||
# MCP Server (for LLM integration via MCP)
|
||||
./run_mcp_server.sh
|
||||
|
||||
# REST API Server (for HTTP clients)
|
||||
./run_api_server.sh
|
||||
|
||||
# Both servers log to mcp.logs and api.logs respectively
|
||||
```
|
||||
|
||||
### Testing
|
||||
```bash
|
||||
# Run core component tests (GPU health, job queue, validation)
|
||||
python tests/test_core_components.py
|
||||
|
||||
# Run async API integration tests
|
||||
python tests/test_async_api_integration.py
|
||||
|
||||
# Run end-to-end integration tests
|
||||
python tests/test_e2e_integration.py
|
||||
```
|
||||
|
||||
### GPU Management
|
||||
```bash
|
||||
# Reset GPU drivers without rebooting (requires sudo)
|
||||
./reset_gpu.sh
|
||||
|
||||
# Check GPU status
|
||||
nvidia-smi
|
||||
|
||||
# Monitor GPU during transcription
|
||||
watch -n 1 nvidia-smi
|
||||
```
|
||||
|
||||
### Installation
|
||||
```bash
|
||||
# Create virtual environment
|
||||
python3.12 -m venv venv
|
||||
source venv/bin/activate
|
||||
|
||||
# Install dependencies (check requirements.txt for CUDA-specific instructions)
|
||||
pip install -r requirements.txt
|
||||
|
||||
# For CUDA 12.4:
|
||||
pip install torch==2.6.0 torchvision==0.21.0 torchaudio==2.6.0 --index-url https://download.pytorch.org/whl/cu124
|
||||
```
|
||||
|
||||
## Architecture
|
||||
|
||||
### Directory Structure
|
||||
```
|
||||
src/
|
||||
├── core/ # Core business logic
|
||||
│ ├── transcriber.py # Main transcription logic with env var defaults
|
||||
│ ├── model_manager.py # Whisper model loading/caching (GPU-only)
|
||||
│ ├── job_queue.py # Async FIFO job queue with worker thread
|
||||
│ ├── gpu_health.py # Real GPU health checks with circuit breaker
|
||||
│ └── gpu_reset.py # Automatic GPU driver reset logic
|
||||
├── servers/ # Server implementations
|
||||
│ ├── whisper_server.py # MCP server (stdio transport)
|
||||
│ └── api_server.py # FastAPI REST server
|
||||
└── utils/ # Utilities
|
||||
├── startup.py # Common startup sequence (GPU check, initialization)
|
||||
├── circuit_breaker.py # Circuit breaker pattern implementation
|
||||
├── audio_processor.py # Audio file validation
|
||||
├── formatters.py # Output format handlers (txt, vtt, srt, json)
|
||||
├── input_validation.py# Input validation utilities
|
||||
└── test_audio_generator.py # Generate test audio for health checks
|
||||
```
|
||||
|
||||
### Key Architectural Patterns
|
||||
|
||||
**Async Job Queue** (`job_queue.py`):
|
||||
- FIFO queue with background worker thread
|
||||
- Disk persistence of job metadata to `JOB_METADATA_DIR`
|
||||
- States: QUEUED → RUNNING → COMPLETED/FAILED
|
||||
- Jobs include full request params + results
|
||||
- Thread-safe operations with locks
|
||||
|
||||
**GPU Health Monitoring** (`gpu_health.py`):
|
||||
- Performs **real** GPU checks: loads tiny model + transcribes test audio
|
||||
- Circuit breaker prevents repeated failures (3 failures → open, 60s timeout)
|
||||
- Integration with GPU auto-reset on failures
|
||||
- Background monitoring thread in `HealthMonitor` class
|
||||
- Never falls back to CPU - raises RuntimeError if GPU unavailable
|
||||
|
||||
**GPU Auto-Reset** (`gpu_reset.py`):
|
||||
- Automatically resets GPU drivers via `reset_gpu.sh` when health checks fail
|
||||
- Cooldown mechanism (default 5 min via `GPU_RESET_COOLDOWN_MINUTES`)
|
||||
- Sudo required - script unloads/reloads nvidia kernel modules
|
||||
- Integrated with circuit breaker to avoid reset loops
|
||||
|
||||
**Startup Sequence** (`startup.py`):
|
||||
- Common startup logic for both servers
|
||||
- Phase 1: GPU health check with optional auto-reset
|
||||
- Phase 2: Initialize job queue
|
||||
- Phase 3: Initialize health monitor (background thread)
|
||||
- Exits on GPU failure unless configured otherwise
|
||||
|
||||
**Circuit Breaker** (`circuit_breaker.py`):
|
||||
- States: CLOSED → OPEN → HALF_OPEN → CLOSED
|
||||
- Configurable failure/success thresholds
|
||||
- Prevents cascading failures and resource exhaustion
|
||||
- Used for GPU health checks and model operations
|
||||
|
||||
### Environment Variables
|
||||
|
||||
Both server scripts set extensive environment variables. Key ones:
|
||||
|
||||
**GPU/CUDA**:
|
||||
- `CUDA_VISIBLE_DEVICES`: GPU index (default: 1)
|
||||
- `LD_LIBRARY_PATH`: CUDA library path
|
||||
- `TRANSCRIPTION_DEVICE`: "cuda" or "auto" (never "cpu")
|
||||
- `TRANSCRIPTION_COMPUTE_TYPE`: "float16", "int8", or "auto"
|
||||
|
||||
**Paths**:
|
||||
- `WHISPER_MODEL_DIR`: Where Whisper models are cached
|
||||
- `TRANSCRIPTION_OUTPUT_DIR`: Transcription output directory
|
||||
- `TRANSCRIPTION_BATCH_OUTPUT_DIR`: Batch output directory
|
||||
- `JOB_METADATA_DIR`: Job metadata persistence directory
|
||||
|
||||
**Transcription Defaults**:
|
||||
- `TRANSCRIPTION_MODEL`: Model name (default: "large-v3")
|
||||
- `TRANSCRIPTION_OUTPUT_FORMAT`: "txt", "vtt", "srt", or "json"
|
||||
- `TRANSCRIPTION_BEAM_SIZE`: Beam search size (default: 5 for API, 2 for MCP)
|
||||
- `TRANSCRIPTION_TEMPERATURE`: Sampling temperature (default: 0.0)
|
||||
|
||||
**Job Queue**:
|
||||
- `JOB_QUEUE_MAX_SIZE`: Max queued jobs (default: 100 for MCP, 5 for API)
|
||||
- `JOB_RETENTION_DAYS`: How long to keep job metadata (default: 7)
|
||||
|
||||
**Health Monitoring**:
|
||||
- `GPU_HEALTH_CHECK_ENABLED`: Enable background monitoring (default: true)
|
||||
- `GPU_HEALTH_CHECK_INTERVAL_MINUTES`: Check interval (default: 10)
|
||||
- `GPU_HEALTH_TEST_MODEL`: Model for health checks (default: "tiny")
|
||||
- `GPU_RESET_COOLDOWN_MINUTES`: Cooldown between reset attempts (default: 5)
|
||||
|
||||
### API Workflow (Async Jobs)
|
||||
|
||||
Both MCP and REST API use the same async workflow:
|
||||
|
||||
1. **Submit job**: `transcribe_async()` returns `job_id` immediately
|
||||
2. **Poll status**: `get_job_status(job_id)` returns status + queue_position
|
||||
3. **Get result**: When status="completed", `get_job_result(job_id)` returns transcription
|
||||
|
||||
The job queue processes one job at a time in a background worker thread.
|
||||
|
||||
### Model Loading Strategy
|
||||
|
||||
- Models are cached in `model_instances` dict (key: model_name + device + compute_type)
|
||||
- First load downloads model to `WHISPER_MODEL_DIR` (or default cache)
|
||||
- GPU health check on model load - may trigger auto-reset if GPU fails
|
||||
- No CPU fallback - raises `RuntimeError` if CUDA unavailable
|
||||
|
||||
## Important Implementation Details
|
||||
|
||||
**GPU-Only Architecture**:
|
||||
- All `device="auto"` resolution checks `torch.cuda.is_available()` and raises error if False
|
||||
- No silent fallback to CPU anywhere in the codebase
|
||||
- Health checks verify model actually ran on GPU (check `torch.cuda.memory_allocated`)
|
||||
|
||||
**Thread Safety**:
|
||||
- `JobQueue` uses `threading.Lock` for job dictionary access
|
||||
- Worker thread processes jobs from `queue.Queue` (thread-safe FIFO)
|
||||
- `HealthMonitor` runs in separate daemon thread
|
||||
|
||||
**Error Handling**:
|
||||
- Circuit breaker prevents retry storms on GPU failures
|
||||
- Input validation rejects invalid audio files, model names, languages
|
||||
- Job errors are captured and stored in job metadata with status=FAILED
|
||||
|
||||
**Shutdown Handling**:
|
||||
- `cleanup_on_shutdown()` waits for current job to complete
|
||||
- Stops health monitor thread
|
||||
- Saves final job states to disk
|
||||
|
||||
## Common Development Tasks
|
||||
|
||||
**Adding a new output format**:
|
||||
1. Add formatter function in `src/utils/formatters.py`
|
||||
2. Add case in `transcribe_audio()` in `src/core/transcriber.py`
|
||||
3. Update API docs and MCP tool descriptions
|
||||
|
||||
**Adjusting GPU health check behavior**:
|
||||
1. Modify circuit breaker params in `src/core/gpu_health.py`
|
||||
2. Adjust health check interval in environment variables
|
||||
3. Consider cooldown timing in `src/core/gpu_reset.py`
|
||||
|
||||
**Testing GPU reset logic**:
|
||||
1. Manually trigger GPU failure (e.g., occupy all GPU memory)
|
||||
2. Watch logs for circuit breaker state transitions
|
||||
3. Verify reset attempt with cooldown enforcement
|
||||
4. Check `nvidia-smi` before/after reset
|
||||
|
||||
**Debugging job queue issues**:
|
||||
1. Check job metadata files in `JOB_METADATA_DIR`
|
||||
2. Look for lock contention in logs
|
||||
3. Verify worker thread is running (check logs for "Job queue worker started")
|
||||
4. Test with `JOB_QUEUE_MAX_SIZE=1` to isolate serialization
|
||||
@@ -1,563 +0,0 @@
|
||||
# MCP Transcriptor - Performance & Security Implementation Plan
|
||||
|
||||
## Session Status: Phase 1-2 Complete ✅
|
||||
|
||||
**Last Updated:** 2025-10-12
|
||||
**Total Estimated Effort:** 17-24 hours
|
||||
**Completed:** ~9-11 hours (Phases 1-2)
|
||||
**Remaining:** ~8-13 hours (Phases 3-5)
|
||||
|
||||
---
|
||||
|
||||
## ✅ COMPLETED PHASES
|
||||
|
||||
### Phase 1: Critical Concurrency & I/O Fixes (COMPLETE)
|
||||
|
||||
#### 1.1 Fixed Async/Sync I/O in `api_server.py` ✅
|
||||
**Files Modified:** `src/servers/api_server.py`
|
||||
|
||||
**Changes:**
|
||||
- Added early queue capacity check before file upload (line 146-156)
|
||||
- Fixed temp file cleanup with `is not None` check (lines 199, 215)
|
||||
- Prevents wasted bandwidth when queue is full
|
||||
|
||||
**Impact:**
|
||||
- Backpressure mechanism prevents OOM from uploads
|
||||
- Proper error handling for temp file cleanup
|
||||
|
||||
---
|
||||
|
||||
#### 1.2 Resolved Job Queue Concurrency Issues ✅
|
||||
**Files Created:** `src/core/job_repository.py`
|
||||
**Files Modified:** `src/core/job_queue.py`
|
||||
|
||||
**Changes:**
|
||||
1. **Created JobRepository class** with write-behind caching:
|
||||
- Batched disk writes (1 second intervals or 50 dirty jobs)
|
||||
- TTL-based cleanup (24h default, configurable via `job_ttl_hours`)
|
||||
- Async I/O to avoid blocking main thread
|
||||
- Background flush thread
|
||||
|
||||
2. **Fine-grained locking in JobQueue**:
|
||||
- Replaced single `RLock` with two `Lock` objects:
|
||||
- `_jobs_lock`: Protects `_jobs` dict
|
||||
- `_queue_positions_lock`: Protects `_queued_job_ids` deque
|
||||
- Moved disk I/O outside lock boundaries
|
||||
|
||||
3. **Fixed TOCTOU race condition in `submit_job()`**:
|
||||
- Capture return data inside lock BEFORE persistence (lines 309-316)
|
||||
- Use `job.mark_for_persistence()` for async writes (line 319)
|
||||
|
||||
4. **Added TTL cleanup**:
|
||||
- `_maybe_cleanup_old_jobs()` runs every hour (lines 332-355)
|
||||
- Removes completed/failed jobs older than TTL
|
||||
- Prevents unbounded memory growth
|
||||
|
||||
**Impact:**
|
||||
- Zero race conditions under concurrent load
|
||||
- Disk I/O no longer blocks job submissions
|
||||
- Memory bounded by TTL (no more memory leaks)
|
||||
- ~90% reduction in disk I/O overhead
|
||||
|
||||
---
|
||||
|
||||
#### 1.3 Optimized Queue Position Tracking ✅
|
||||
**Files Modified:** `src/core/job_queue.py`
|
||||
|
||||
**Changes:**
|
||||
- Reduced recalculation frequency (lines 509-510, 558)
|
||||
- Only recalculate when jobs added/removed from queue
|
||||
- Removed unnecessary recalculation on job completion
|
||||
|
||||
**Impact:**
|
||||
- 3× reduction in position calculation overhead
|
||||
- Faster job status queries
|
||||
|
||||
---
|
||||
|
||||
### Phase 2: Performance Optimizations (COMPLETE)
|
||||
|
||||
#### 2.1 GPU Health Check Optimization ✅
|
||||
**Files Modified:** `src/core/job_queue.py`
|
||||
|
||||
**Changes:**
|
||||
- Added 30-second cache for GPU health results (lines 155-158)
|
||||
- Cache invalidation on failures (lines 270-272)
|
||||
- Cached result prevents redundant model loading tests (lines 247-267)
|
||||
|
||||
**Impact:**
|
||||
- GPU health check <100ms when cached (vs ~1000ms uncached)
|
||||
- Reduces GPU memory thrashing from repeated tiny model loads
|
||||
|
||||
---
|
||||
|
||||
#### 2.2 Reduced Lock Contention ✅
|
||||
**Achieved through Phase 1.2 fine-grained locking**
|
||||
|
||||
**Impact:**
|
||||
- Lock hold time reduced by ~80%
|
||||
- Parallel job status queries now possible
|
||||
- Worker thread no longer blocks API requests
|
||||
|
||||
---
|
||||
|
||||
## 🚧 REMAINING PHASES
|
||||
|
||||
### Phase 3: Input Validation & Security (Priority: HIGH)
|
||||
**Estimated Effort:** 2-3 hours
|
||||
|
||||
#### 3.1 Integrate Validation Module in API
|
||||
**Files to Modify:** `src/servers/api_server.py`
|
||||
|
||||
**Tasks:**
|
||||
- [ ] Add Pydantic validators to `SubmitJobRequest` model (line 68)
|
||||
```python
|
||||
from pydantic import validator
|
||||
from utils.input_validation import (
|
||||
validate_beam_size,
|
||||
validate_temperature,
|
||||
validate_model_name,
|
||||
validate_device,
|
||||
validate_compute_type,
|
||||
validate_output_format
|
||||
)
|
||||
|
||||
class SubmitJobRequest(BaseModel):
|
||||
# ... existing fields ...
|
||||
|
||||
@validator('beam_size')
|
||||
def check_beam_size(cls, v):
|
||||
return validate_beam_size(v)
|
||||
|
||||
@validator('temperature')
|
||||
def check_temperature(cls, v):
|
||||
return validate_temperature(v)
|
||||
|
||||
@validator('model_name')
|
||||
def check_model_name(cls, v):
|
||||
return validate_model_name(v)
|
||||
|
||||
# ... add validators for device, compute_type, output_format ...
|
||||
```
|
||||
|
||||
- [ ] Add validators to `/transcribe` endpoint form parameters (line 128)
|
||||
- [ ] Update exception handling to catch specific validation errors:
|
||||
- `ValidationError`
|
||||
- `PathTraversalError`
|
||||
- `InvalidFileTypeError`
|
||||
- `FileSizeError`
|
||||
|
||||
**Impact:**
|
||||
- Reject malformed inputs before GPU health check
|
||||
- Clear error messages for API users
|
||||
- Security against path traversal and injection
|
||||
|
||||
---
|
||||
|
||||
#### 3.2 Error Handling Improvements
|
||||
**Files to Modify:** `src/core/job_queue.py`, `src/core/transcriber.py`
|
||||
|
||||
**Tasks:**
|
||||
- [ ] Fix exception handling in `job_queue.submit_job()` (line 238)
|
||||
- Catch `ValidationError` base class instead of just `Exception`
|
||||
|
||||
- [ ] Improve empty segment handling in `transcriber.py` (line 224)
|
||||
```python
|
||||
if segment_count == 0:
|
||||
if info.duration < 1.0:
|
||||
return "Transcription failed: Audio too short (< 1 second)"
|
||||
else:
|
||||
return "Transcription failed: No speech detected (VAD filtered all segments)"
|
||||
```
|
||||
|
||||
- [ ] Add error codes to API responses
|
||||
```python
|
||||
{
|
||||
"error_code": "VALIDATION_ERROR",
|
||||
"error_type": "InvalidFileTypeError",
|
||||
"message": "...",
|
||||
"details": {...}
|
||||
}
|
||||
```
|
||||
|
||||
**Impact:**
|
||||
- Clearer error messages for debugging
|
||||
- Better distinction between user errors vs system errors
|
||||
|
||||
---
|
||||
|
||||
### Phase 4: Code Quality & Architecture (Priority: MEDIUM)
|
||||
**Estimated Effort:** 4-6 hours
|
||||
|
||||
#### 4.1 Refactor JobQueue God Class
|
||||
**Files to Create/Modify:** `src/core/job_queue.py`
|
||||
|
||||
**Tasks:**
|
||||
- [ ] **OPTIONAL**: Further extract `PositionTracker` class (if needed)
|
||||
- Most refactoring already done via `JobRepository`
|
||||
- Current separation (queue, repository, positions) is acceptable
|
||||
- Only extract if position logic becomes more complex
|
||||
|
||||
**Note:** Phase 1.2 already extracted major god class responsibilities via `JobRepository`. This task is now lower priority.
|
||||
|
||||
---
|
||||
|
||||
#### 4.2 Code Cleanup & Standards
|
||||
**Files to Modify:** Multiple
|
||||
|
||||
**Tasks:**
|
||||
- [ ] Extract magic numbers to constants
|
||||
```python
|
||||
# In job_queue.py
|
||||
DEFAULT_BATCH_INTERVAL_SECONDS = 1.0
|
||||
DEFAULT_JOB_TTL_HOURS = 24
|
||||
MAX_DIRTY_JOBS_BEFORE_FLUSH = 50
|
||||
GPU_HEALTH_CACHE_TTL_SECONDS = 30
|
||||
CLEANUP_INTERVAL_SECONDS = 3600
|
||||
|
||||
# In api_server.py
|
||||
UPLOAD_CHUNK_SIZE_BYTES = 8192
|
||||
QUEUE_CHECK_TIMEOUT_SECONDS = 1.0
|
||||
|
||||
# In gpu_health.py
|
||||
GPU_TEST_TIMEOUT_SECONDS = 2.0
|
||||
```
|
||||
|
||||
- [ ] Replace `RLock` with `Lock` in `circuit_breaker.py` (line 87)
|
||||
- No re-entrant calls detected in codebase
|
||||
- Standard `Lock` is 10-15% faster
|
||||
|
||||
- [ ] Use `time.monotonic()` consistently in `gpu_reset.py`
|
||||
- Replace `datetime.utcnow()` with `time.monotonic()` for cooldown (line 69)
|
||||
- Prevents NTP drift issues
|
||||
|
||||
- [ ] Fix timestamp format in `job_queue.py`
|
||||
```python
|
||||
# Option 1: Use timezone-aware datetimes
|
||||
from datetime import timezone
|
||||
created_at = datetime.now(timezone.utc)
|
||||
|
||||
# Option 2: Document as UTC convention
|
||||
# Current approach is acceptable if documented
|
||||
```
|
||||
|
||||
**Impact:**
|
||||
- Improved code maintainability
|
||||
- Easier configuration tuning
|
||||
- More accurate time-based logic
|
||||
|
||||
---
|
||||
|
||||
#### 4.3 GPU Health Check Consolidation
|
||||
**Files to Modify:** `src/core/gpu_health.py`
|
||||
|
||||
**Tasks:**
|
||||
- [ ] Merge three similar functions using strategy pattern
|
||||
```python
|
||||
class GPUHealthCheckStrategy(Enum):
|
||||
BASIC = "basic" # No auto-reset
|
||||
WITH_RESET = "with_reset" # Auto-reset on failure
|
||||
CIRCUIT_BREAKER = "circuit" # Circuit breaker only
|
||||
|
||||
def check_gpu_health(
|
||||
expected_device: str = "auto",
|
||||
strategy: GPUHealthCheckStrategy = GPUHealthCheckStrategy.CIRCUIT_BREAKER,
|
||||
auto_reset: bool = True
|
||||
) -> GPUHealthStatus:
|
||||
# Unified implementation
|
||||
...
|
||||
```
|
||||
|
||||
- [ ] Clarify circuit breaker vs reset behavior in docstrings
|
||||
|
||||
**Impact:**
|
||||
- Reduced code duplication
|
||||
- Clearer API for callers
|
||||
|
||||
---
|
||||
|
||||
### Phase 5: Reliability & Edge Cases (Priority: LOW)
|
||||
**Estimated Effort:** 2-3 hours
|
||||
|
||||
#### 5.1 Resource Management Improvements
|
||||
**Files to Modify:** `src/core/transcriber.py`, `src/servers/api_server.py`
|
||||
|
||||
**Tasks:**
|
||||
- [ ] Add explicit file handle cleanup in transcription streaming
|
||||
```python
|
||||
# In transcriber.py around line 176
|
||||
try:
|
||||
with open(output_path, "w", encoding="utf-8") as f:
|
||||
# ... existing streaming logic ...
|
||||
except Exception as write_error:
|
||||
logger.error(f"Failed to write transcription: {str(write_error)}")
|
||||
# File handle automatically closed by context manager
|
||||
if os.path.exists(output_path):
|
||||
try:
|
||||
os.remove(output_path)
|
||||
except Exception as cleanup_error:
|
||||
logger.warning(f"Failed to cleanup partial file: {cleanup_error}")
|
||||
raise
|
||||
```
|
||||
|
||||
- [ ] Add disk space checks before accepting uploads
|
||||
```python
|
||||
import shutil
|
||||
|
||||
def check_disk_space(path: str, required_bytes: int):
|
||||
stat = shutil.disk_usage(path)
|
||||
if stat.free < required_bytes * 1.1: # 10% buffer
|
||||
raise IOError(f"Insufficient disk space: {stat.free / 1e9:.1f}GB available")
|
||||
```
|
||||
|
||||
- [ ] Monitor file descriptor limits
|
||||
```python
|
||||
import resource
|
||||
|
||||
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
|
||||
logger.info(f"File descriptor limits: {soft}/{hard}")
|
||||
```
|
||||
|
||||
**Impact:**
|
||||
- No file descriptor leaks under sustained load
|
||||
- Graceful handling of disk full scenarios
|
||||
|
||||
---
|
||||
|
||||
#### 5.2 Observability Enhancements
|
||||
**Files to Modify:** `src/core/job_queue.py`, `src/servers/api_server.py`
|
||||
|
||||
**Tasks:**
|
||||
- [ ] Add detailed logging for edge cases
|
||||
```python
|
||||
# In transcriber.py
|
||||
if segment_count == 0:
|
||||
logger.warning(
|
||||
f"No transcription segments generated: "
|
||||
f"duration={info.duration:.2f}s, "
|
||||
f"language={info.language}, "
|
||||
f"vad_enabled=True"
|
||||
)
|
||||
```
|
||||
|
||||
- [ ] Add metrics endpoint (optional)
|
||||
```python
|
||||
@app.get("/metrics")
|
||||
async def get_metrics():
|
||||
return {
|
||||
"queue": {
|
||||
"size": job_queue._queue.qsize(),
|
||||
"max_size": job_queue._max_queue_size,
|
||||
"current_job_id": job_queue._current_job_id
|
||||
},
|
||||
"jobs": {
|
||||
"total": len(job_queue._jobs),
|
||||
"queued": len(job_queue._queued_job_ids),
|
||||
"running": 1 if job_queue._current_job_id else 0
|
||||
},
|
||||
"gpu_health_cache": {
|
||||
"cached": job_queue._gpu_health_cache is not None,
|
||||
"age_seconds": (
|
||||
(datetime.utcnow() - job_queue._gpu_health_cache_time).total_seconds()
|
||||
if job_queue._gpu_health_cache_time else None
|
||||
)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] Improve error messages with actionable guidance
|
||||
```python
|
||||
raise HTTPException(
|
||||
status_code=503,
|
||||
detail={
|
||||
"error": "Queue full",
|
||||
"message": "Job queue is full. Please try again later.",
|
||||
"retry_after_seconds": 60,
|
||||
"queue_size": job_queue._max_queue_size,
|
||||
"suggestion": "Consider increasing MAX_QUEUE_SIZE environment variable"
|
||||
}
|
||||
)
|
||||
```
|
||||
|
||||
**Impact:**
|
||||
- Easier debugging and monitoring
|
||||
- Better production observability
|
||||
- Actionable error messages for operators
|
||||
|
||||
---
|
||||
|
||||
## Testing Recommendations (Future Session)
|
||||
|
||||
When ready to test, prioritize these scenarios:
|
||||
|
||||
### Unit Tests
|
||||
- [ ] Queue position tracking with concurrent submissions
|
||||
- [ ] JobRepository write-behind caching and flush behavior
|
||||
- [ ] TTL cleanup edge cases
|
||||
- [ ] GPU health cache expiration
|
||||
|
||||
### Integration Tests
|
||||
- [ ] Large file upload with queue full scenario
|
||||
- [ ] Concurrent job submissions (10+ parallel)
|
||||
- [ ] Server restart with queued jobs on disk
|
||||
- [ ] GPU health check failures during job processing
|
||||
|
||||
### Load Tests
|
||||
- [ ] 100 concurrent job submissions
|
||||
- [ ] 1000+ jobs submitted over time (memory stability)
|
||||
- [ ] Queue status queries under load (lock contention)
|
||||
|
||||
### Chaos Tests
|
||||
- [ ] GPU reset during active transcription
|
||||
- [ ] Disk full during job persistence
|
||||
- [ ] Kill server during job processing (restart recovery)
|
||||
|
||||
---
|
||||
|
||||
## Configuration Changes
|
||||
|
||||
### New Environment Variables
|
||||
|
||||
Add to your `.env` or system environment:
|
||||
|
||||
```bash
|
||||
# Job queue configuration
|
||||
JOB_TTL_HOURS=24 # How long to keep completed jobs
|
||||
JOB_REPOSITORY_BATCH_INTERVAL=1.0 # Seconds between disk flushes
|
||||
GPU_HEALTH_CACHE_TTL_SECONDS=30 # GPU health check cache duration
|
||||
|
||||
# Existing variables (document for reference)
|
||||
TRANSCRIPTION_OUTPUT_DIR=/outputs
|
||||
MAX_QUEUE_SIZE=100
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Success Criteria
|
||||
|
||||
### Phase 1-2 (✅ ACHIEVED)
|
||||
- ✅ Zero race conditions under concurrent load
|
||||
- ✅ Non-blocking async I/O throughout FastAPI endpoints
|
||||
- ✅ Memory stable under 1000+ job submissions (TTL cleanup)
|
||||
- ✅ GPU health check <100ms when cached
|
||||
- ✅ Write-behind persistence reduces I/O by ~90%
|
||||
|
||||
### Phase 3-5 (TO VERIFY)
|
||||
- ⏳ All inputs validated with proper error messages
|
||||
- ⏳ <50ms p99 latency for job status queries
|
||||
- ⏳ Clean shutdown with no resource leaks
|
||||
- ⏳ Clear error codes for all failure modes
|
||||
|
||||
---
|
||||
|
||||
## Known Issues & Technical Debt
|
||||
|
||||
### Addressed in Phase 1-2
|
||||
- ~~CRITICAL: Race condition in job position capture~~
|
||||
- ~~CRITICAL: Blocking I/O in async FastAPI endpoint~~
|
||||
- ~~HIGH: Unbounded job memory growth~~
|
||||
- ~~HIGH: N+1 disk I/O in job persistence~~
|
||||
- ~~HIGH: Lock contention in worker thread~~
|
||||
- ~~MEDIUM: Duplicate GPU health checks~~
|
||||
|
||||
### Remaining (Phase 3-5)
|
||||
- MEDIUM: Unvalidated input parameters in API
|
||||
- LOW: Magic numbers scattered throughout code
|
||||
- LOW: RLock used instead of Lock in circuit breaker
|
||||
- LOW: Time drift vulnerability in GPU reset cooldown
|
||||
- LOW: Inconsistent timestamp format (Z suffix on naive datetime)
|
||||
|
||||
---
|
||||
|
||||
## File Change Summary
|
||||
|
||||
### New Files Created
|
||||
- `src/core/job_repository.py` (242 lines) - Write-behind persistence layer
|
||||
|
||||
### Modified Files
|
||||
- `src/core/job_queue.py` - Major refactor (fine-grained locking, caching, TTL cleanup)
|
||||
- `src/servers/api_server.py` - Backpressure + temp file fixes
|
||||
- All other files unchanged
|
||||
|
||||
### Lines of Code Changed
|
||||
- Added: ~450 lines (JobRepository + refactoring)
|
||||
- Modified: ~200 lines (JobQueue methods)
|
||||
- Deleted: ~50 lines (removed old save_to_disk, single RLock)
|
||||
- **Net:** +600 lines (~15% increase)
|
||||
|
||||
---
|
||||
|
||||
## Next Session Checklist
|
||||
|
||||
### Before Starting Phase 3
|
||||
|
||||
1. **Test Phase 1-2 changes:**
|
||||
```bash
|
||||
# Start server
|
||||
python src/servers/api_server.py
|
||||
|
||||
# Submit multiple concurrent jobs
|
||||
for i in {1..10}; do
|
||||
curl -X POST http://localhost:8000/jobs \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"audio_path": "/path/to/audio.mp3"}' &
|
||||
done
|
||||
|
||||
# Check job status queries are fast
|
||||
time curl http://localhost:8000/jobs
|
||||
|
||||
# Verify no memory leaks over time
|
||||
watch -n 5 'ps aux | grep api_server'
|
||||
```
|
||||
|
||||
2. **Verify TTL cleanup:**
|
||||
- Check `outputs/jobs/` directory for old job files
|
||||
- Confirm files older than 24h are deleted
|
||||
|
||||
3. **Test GPU health cache:**
|
||||
- Submit jobs back-to-back
|
||||
- Check logs for "Using cached GPU health check result"
|
||||
|
||||
4. **Code review:**
|
||||
- Review `job_repository.py` for any issues
|
||||
- Review locking changes in `job_queue.py`
|
||||
|
||||
### Starting Phase 3
|
||||
|
||||
1. Begin with `3.1 Integrate Validation Module in API`
|
||||
2. Test each validator incrementally
|
||||
3. Update API documentation with validation rules
|
||||
4. Add integration tests for validation errors
|
||||
|
||||
---
|
||||
|
||||
## Contact & Documentation
|
||||
|
||||
- **Analysis Report:** See analysis output from initial session
|
||||
- **Git Branch:** `alihan-specific` (current working branch)
|
||||
- **Main Branch:** `main` (for PRs)
|
||||
- **Recent Commits:**
|
||||
- `d47c284` - Fix GPU check at startup issue
|
||||
- `66b36e7` - Update documentation and configuration
|
||||
- `5fb742a` - Add circuit breaker, input validation, and refactor startup logic
|
||||
|
||||
---
|
||||
|
||||
## Rollback Plan
|
||||
|
||||
If Phase 1-2 changes cause issues:
|
||||
|
||||
```bash
|
||||
# Revert to before session
|
||||
git diff HEAD src/core/job_queue.py src/servers/api_server.py
|
||||
git checkout HEAD -- src/core/job_queue.py src/servers/api_server.py
|
||||
rm src/core/job_repository.py
|
||||
|
||||
# Or create a rollback branch
|
||||
git checkout -b rollback-phase-1-2
|
||||
git revert HEAD~5..HEAD # Adjust commit count as needed
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
**End of Implementation Plan**
|
||||
132
nginx/README.md
Normal file
132
nginx/README.md
Normal file
@@ -0,0 +1,132 @@
|
||||
# Nginx Configuration for Transcriptor API
|
||||
|
||||
This directory contains nginx reverse proxy configuration to fix 504 Gateway Timeout errors.
|
||||
|
||||
## Problem
|
||||
|
||||
The transcriptor API can take a long time (10+ minutes) to process large audio files with the `large-v3` model. Without proper timeout configuration, requests will fail with 504 Gateway Timeout.
|
||||
|
||||
## Solution
|
||||
|
||||
The provided `transcriptor.conf` file configures nginx with appropriate timeouts:
|
||||
|
||||
- **proxy_connect_timeout**: 600s (10 minutes)
|
||||
- **proxy_send_timeout**: 600s (10 minutes)
|
||||
- **proxy_read_timeout**: 3600s (1 hour)
|
||||
- **client_max_body_size**: 500M (for large audio files)
|
||||
|
||||
## Installation
|
||||
|
||||
### Option 1: Deploy nginx configuration (if using nginx)
|
||||
|
||||
```bash
|
||||
# Copy configuration to nginx
|
||||
sudo cp transcriptor.conf /etc/nginx/sites-available/
|
||||
|
||||
# Create symlink to enable it
|
||||
sudo ln -s /etc/nginx/sites-available/transcriptor.conf /etc/nginx/sites-enabled/
|
||||
|
||||
# Test configuration
|
||||
sudo nginx -t
|
||||
|
||||
# Reload nginx
|
||||
sudo systemctl reload nginx
|
||||
```
|
||||
|
||||
### Option 2: Run API server directly (current setup)
|
||||
|
||||
The API server at `src/servers/api_server.py` has been updated with:
|
||||
- `timeout_keep_alive=3600` (1 hour)
|
||||
- `timeout_graceful_shutdown=60`
|
||||
|
||||
No additional nginx configuration is needed if you're running the API directly.
|
||||
|
||||
## Restart Service
|
||||
|
||||
After making changes, restart the transcriptor service:
|
||||
|
||||
```bash
|
||||
# If using supervisor
|
||||
sudo supervisorctl restart transcriptor-api
|
||||
|
||||
# If using systemd
|
||||
sudo systemctl restart transcriptor-api
|
||||
|
||||
# If using docker
|
||||
docker restart <container-name>
|
||||
```
|
||||
|
||||
## Testing
|
||||
|
||||
Test the API is working:
|
||||
|
||||
```bash
|
||||
# Health check (should return 200)
|
||||
curl http://192.168.1.210:33767/health
|
||||
|
||||
# Check timeout configuration
|
||||
curl -X POST http://192.168.1.210:33767/transcribe \
|
||||
-F "file=@test_audio.mp3" \
|
||||
-F "model=large-v3" \
|
||||
-F "output_format=txt"
|
||||
```
|
||||
|
||||
## Monitoring
|
||||
|
||||
Check logs for timeout warnings:
|
||||
|
||||
```bash
|
||||
# Supervisor logs
|
||||
tail -f /home/uad/agents/tools/mcp-transcriptor/logs/transcriptor-api.log
|
||||
|
||||
# Look for messages like:
|
||||
# - "Job {job_id} is taking longer than expected: 610.5s elapsed (threshold: 600s)"
|
||||
# - "Job {job_id} exceeded maximum timeout: 3610.2s elapsed (max: 3600s)"
|
||||
```
|
||||
|
||||
## Configuration Environment Variables
|
||||
|
||||
You can also configure timeouts via environment variables in `supervisor/transcriptor-api.conf`:
|
||||
|
||||
```ini
|
||||
environment=
|
||||
...
|
||||
JOB_TIMEOUT_WARNING_SECONDS="600", # Warn after 10 minutes
|
||||
JOB_TIMEOUT_MAX_SECONDS="3600", # Fail after 1 hour
|
||||
```
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Still getting 504 errors?
|
||||
|
||||
1. **Check service is running**:
|
||||
```bash
|
||||
sudo supervisorctl status transcriptor-api
|
||||
```
|
||||
|
||||
2. **Check port is listening**:
|
||||
```bash
|
||||
sudo netstat -tlnp | grep 33767
|
||||
```
|
||||
|
||||
3. **Check logs for errors**:
|
||||
```bash
|
||||
tail -100 /home/uad/agents/tools/mcp-transcriptor/logs/transcriptor-api.log
|
||||
```
|
||||
|
||||
4. **Test direct connection** (bypass nginx):
|
||||
```bash
|
||||
curl http://localhost:33767/health
|
||||
```
|
||||
|
||||
5. **Verify GPU is working**:
|
||||
```bash
|
||||
curl http://192.168.1.210:33767/health/gpu
|
||||
```
|
||||
|
||||
### Job takes too long?
|
||||
|
||||
Consider:
|
||||
- Using a smaller model (e.g., `medium` instead of `large-v3`)
|
||||
- Splitting large audio files into smaller chunks
|
||||
- Increasing `JOB_TIMEOUT_MAX_SECONDS` for very long audio files
|
||||
85
nginx/transcriptor.conf
Normal file
85
nginx/transcriptor.conf
Normal file
@@ -0,0 +1,85 @@
|
||||
# Nginx reverse proxy configuration for Whisper Transcriptor API
|
||||
# Place this file in /etc/nginx/sites-available/ and symlink to sites-enabled/
|
||||
|
||||
upstream transcriptor_backend {
|
||||
# Backend transcriptor API server
|
||||
server 127.0.0.1:33767;
|
||||
|
||||
# Connection pooling
|
||||
keepalive 32;
|
||||
}
|
||||
|
||||
server {
|
||||
listen 80;
|
||||
server_name transcriptor.local; # Change to your domain
|
||||
|
||||
# Increase client body size for large audio uploads (up to 500MB)
|
||||
client_max_body_size 500M;
|
||||
|
||||
# Timeouts for long-running transcription jobs
|
||||
proxy_connect_timeout 600s; # 10 minutes to establish connection
|
||||
proxy_send_timeout 600s; # 10 minutes to send request
|
||||
proxy_read_timeout 3600s; # 1 hour to read response (transcription can be slow)
|
||||
|
||||
# Buffer settings for large responses
|
||||
proxy_buffering on;
|
||||
proxy_buffer_size 4k;
|
||||
proxy_buffers 8 4k;
|
||||
proxy_busy_buffers_size 8k;
|
||||
|
||||
# API endpoints
|
||||
location / {
|
||||
proxy_pass http://transcriptor_backend;
|
||||
|
||||
# Forward client info
|
||||
proxy_set_header Host $host;
|
||||
proxy_set_header X-Real-IP $remote_addr;
|
||||
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
|
||||
proxy_set_header X-Forwarded-Proto $scheme;
|
||||
|
||||
# HTTP/1.1 for keepalive
|
||||
proxy_http_version 1.1;
|
||||
proxy_set_header Connection "";
|
||||
|
||||
# Disable buffering for streaming endpoints
|
||||
proxy_request_buffering off;
|
||||
}
|
||||
|
||||
# Health check endpoint with shorter timeout
|
||||
location /health {
|
||||
proxy_pass http://transcriptor_backend;
|
||||
proxy_read_timeout 10s;
|
||||
|
||||
proxy_set_header Host $host;
|
||||
proxy_set_header X-Real-IP $remote_addr;
|
||||
}
|
||||
|
||||
# Access and error logs
|
||||
access_log /var/log/nginx/transcriptor_access.log;
|
||||
error_log /var/log/nginx/transcriptor_error.log warn;
|
||||
}
|
||||
|
||||
# HTTPS configuration (optional, recommended for production)
|
||||
# server {
|
||||
# listen 443 ssl http2;
|
||||
# server_name transcriptor.local;
|
||||
#
|
||||
# ssl_certificate /etc/ssl/certs/transcriptor.crt;
|
||||
# ssl_certificate_key /etc/ssl/private/transcriptor.key;
|
||||
#
|
||||
# # SSL settings
|
||||
# ssl_protocols TLSv1.2 TLSv1.3;
|
||||
# ssl_ciphers HIGH:!aNULL:!MD5;
|
||||
# ssl_prefer_server_ciphers on;
|
||||
#
|
||||
# # Same settings as HTTP above
|
||||
# client_max_body_size 500M;
|
||||
# proxy_connect_timeout 600s;
|
||||
# proxy_send_timeout 600s;
|
||||
# proxy_read_timeout 3600s;
|
||||
#
|
||||
# location / {
|
||||
# proxy_pass http://transcriptor_backend;
|
||||
# # ... (same proxy settings as above)
|
||||
# }
|
||||
# }
|
||||
@@ -12,6 +12,7 @@ mcp[cli]
|
||||
fastapi>=0.115.0
|
||||
uvicorn[standard]>=0.32.0
|
||||
python-multipart>=0.0.9
|
||||
aiofiles>=23.0.0 # Async file I/O
|
||||
|
||||
# Test audio generation dependencies
|
||||
gTTS>=2.3.0
|
||||
|
||||
@@ -15,12 +15,17 @@ from typing import Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Cooldown file location
|
||||
# Cooldown file location (stores monotonic timestamp for drift protection)
|
||||
RESET_TIMESTAMP_FILE = "/tmp/whisper-gpu-last-reset"
|
||||
|
||||
# Default cooldown period (minutes)
|
||||
DEFAULT_COOLDOWN_MINUTES = 5
|
||||
|
||||
# Cooldown period in seconds (for monotonic comparison)
|
||||
def get_cooldown_seconds() -> float:
|
||||
"""Get cooldown period in seconds."""
|
||||
return get_cooldown_minutes() * 60.0
|
||||
|
||||
|
||||
def get_cooldown_minutes() -> int:
|
||||
"""
|
||||
@@ -38,12 +43,12 @@ def get_cooldown_minutes() -> int:
|
||||
return DEFAULT_COOLDOWN_MINUTES
|
||||
|
||||
|
||||
def get_last_reset_time() -> Optional[datetime]:
|
||||
def get_last_reset_time() -> Optional[float]:
|
||||
"""
|
||||
Read timestamp of last GPU reset attempt.
|
||||
Read monotonic timestamp of last GPU reset attempt.
|
||||
|
||||
Returns:
|
||||
datetime object of last reset, or None if no previous reset
|
||||
Monotonic timestamp of last reset, or None if no previous reset
|
||||
"""
|
||||
try:
|
||||
if not os.path.exists(RESET_TIMESTAMP_FILE):
|
||||
@@ -52,7 +57,7 @@ def get_last_reset_time() -> Optional[datetime]:
|
||||
with open(RESET_TIMESTAMP_FILE, 'r') as f:
|
||||
timestamp_str = f.read().strip()
|
||||
|
||||
return datetime.fromisoformat(timestamp_str)
|
||||
return float(timestamp_str)
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to read last reset timestamp: {e}")
|
||||
@@ -63,15 +68,17 @@ def record_reset_attempt() -> None:
|
||||
"""
|
||||
Record current time as last GPU reset attempt.
|
||||
|
||||
Creates/updates timestamp file with current UTC time.
|
||||
Creates/updates timestamp file with monotonic time (drift-protected).
|
||||
"""
|
||||
try:
|
||||
timestamp = datetime.utcnow().isoformat()
|
||||
# Use monotonic time to prevent NTP drift issues
|
||||
timestamp_monotonic = time.monotonic()
|
||||
timestamp_iso = datetime.utcnow().isoformat() # For logging only
|
||||
|
||||
with open(RESET_TIMESTAMP_FILE, 'w') as f:
|
||||
f.write(timestamp)
|
||||
f.write(str(timestamp_monotonic))
|
||||
|
||||
logger.info(f"Recorded GPU reset timestamp: {timestamp}")
|
||||
logger.info(f"Recorded GPU reset timestamp: {timestamp_iso} (monotonic: {timestamp_monotonic:.2f})")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to record reset timestamp: {e}")
|
||||
@@ -81,35 +88,36 @@ def can_attempt_reset() -> bool:
|
||||
"""
|
||||
Check if GPU reset can be attempted based on cooldown period.
|
||||
|
||||
Uses monotonic time to prevent NTP drift issues.
|
||||
|
||||
Returns:
|
||||
True if reset is allowed (no recent reset or cooldown expired),
|
||||
False if cooldown is still active
|
||||
"""
|
||||
last_reset = get_last_reset_time()
|
||||
last_reset_monotonic = get_last_reset_time()
|
||||
|
||||
if last_reset is None:
|
||||
if last_reset_monotonic is None:
|
||||
# No previous reset recorded
|
||||
logger.debug("No previous GPU reset found, reset allowed")
|
||||
return True
|
||||
|
||||
cooldown_minutes = get_cooldown_minutes()
|
||||
cooldown_period = timedelta(minutes=cooldown_minutes)
|
||||
time_since_reset = datetime.utcnow() - last_reset
|
||||
# Use monotonic time for drift-safe comparison
|
||||
current_monotonic = time.monotonic()
|
||||
time_since_reset_seconds = current_monotonic - last_reset_monotonic
|
||||
cooldown_seconds = get_cooldown_seconds()
|
||||
|
||||
if time_since_reset < cooldown_period:
|
||||
remaining = cooldown_period - time_since_reset
|
||||
if time_since_reset_seconds < cooldown_seconds:
|
||||
remaining_seconds = cooldown_seconds - time_since_reset_seconds
|
||||
logger.warning(
|
||||
f"GPU reset cooldown active. "
|
||||
f"Last reset: {last_reset.isoformat()}, "
|
||||
f"Cooldown: {cooldown_minutes} min, "
|
||||
f"Remaining: {remaining.total_seconds():.0f}s"
|
||||
f"Cooldown: {get_cooldown_minutes()} min, "
|
||||
f"Remaining: {remaining_seconds:.0f}s"
|
||||
)
|
||||
return False
|
||||
|
||||
logger.info(
|
||||
f"GPU reset cooldown expired. "
|
||||
f"Last reset: {last_reset.isoformat()}, "
|
||||
f"Time since: {time_since_reset.total_seconds():.0f}s"
|
||||
f"Time since last reset: {time_since_reset_seconds:.0f}s"
|
||||
)
|
||||
return True
|
||||
|
||||
@@ -149,16 +157,29 @@ def reset_gpu_drivers() -> None:
|
||||
logger.error(error_msg)
|
||||
raise RuntimeError(error_msg)
|
||||
|
||||
logger.info(f"Executing GPU reset script: {script_path}")
|
||||
# Resolve to absolute path and validate it's the expected script
|
||||
# This prevents path injection if script_path was somehow manipulated
|
||||
resolved_path = script_path.resolve()
|
||||
|
||||
# Security check: Ensure resolved path is still in expected location
|
||||
expected_parent = Path(__file__).parent.parent.parent.resolve()
|
||||
if resolved_path.parent != expected_parent:
|
||||
error_msg = f"Security check failed: Script path outside expected directory"
|
||||
logger.error(error_msg)
|
||||
raise RuntimeError(error_msg)
|
||||
|
||||
logger.info(f"Executing GPU reset script: {resolved_path}")
|
||||
logger.warning("This will temporarily interrupt all GPU operations")
|
||||
|
||||
try:
|
||||
# Execute reset script with sudo
|
||||
# Using list form (not shell=True) prevents shell injection
|
||||
result = subprocess.run(
|
||||
['sudo', str(script_path)],
|
||||
['sudo', str(resolved_path)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30 # 30 second timeout
|
||||
timeout=30, # 30 second timeout
|
||||
shell=False # Explicitly disable shell to prevent injection
|
||||
)
|
||||
|
||||
# Log script output
|
||||
|
||||
@@ -24,6 +24,13 @@ from utils.audio_processor import validate_audio_file
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Constants
|
||||
DEFAULT_JOB_TTL_HOURS = 24 # How long to keep completed jobs in memory
|
||||
GPU_HEALTH_CACHE_TTL_SECONDS = 30 # Cache GPU health check results
|
||||
CLEANUP_INTERVAL_SECONDS = 3600 # Run TTL cleanup every hour (1 hour)
|
||||
JOB_TIMEOUT_WARNING_SECONDS = 600 # Warn if job takes > 10 minutes
|
||||
JOB_TIMEOUT_MAX_SECONDS = 3600 # Maximum 1 hour per job
|
||||
|
||||
|
||||
class JobStatus(Enum):
|
||||
"""Job status enumeration."""
|
||||
@@ -152,10 +159,10 @@ class JobQueue:
|
||||
# TTL cleanup tracking
|
||||
self._last_cleanup_time = datetime.utcnow()
|
||||
|
||||
# GPU health check caching (30 second cache)
|
||||
# GPU health check caching
|
||||
self._gpu_health_cache: Optional[any] = None
|
||||
self._gpu_health_cache_time: Optional[datetime] = None
|
||||
self._gpu_health_cache_ttl_seconds = 30
|
||||
self._gpu_health_cache_ttl_seconds = GPU_HEALTH_CACHE_TTL_SECONDS
|
||||
|
||||
def start(self):
|
||||
"""
|
||||
@@ -361,7 +368,7 @@ class JobQueue:
|
||||
"""Periodically cleanup old completed/failed jobs based on TTL."""
|
||||
# Only run cleanup every hour
|
||||
now = datetime.utcnow()
|
||||
if (now - self._last_cleanup_time).total_seconds() < 3600:
|
||||
if (now - self._last_cleanup_time).total_seconds() < CLEANUP_INTERVAL_SECONDS:
|
||||
return
|
||||
|
||||
self._last_cleanup_time = now
|
||||
@@ -542,24 +549,65 @@ class JobQueue:
|
||||
|
||||
logger.info(f"Job {job.job_id} started processing")
|
||||
|
||||
# Process job
|
||||
# Process job with timeout tracking
|
||||
start_time = time.time()
|
||||
try:
|
||||
result = transcribe_audio(
|
||||
audio_path=job.audio_path,
|
||||
model_name=job.model_name,
|
||||
device=job.device,
|
||||
compute_type=job.compute_type,
|
||||
language=job.language,
|
||||
output_format=job.output_format,
|
||||
beam_size=job.beam_size,
|
||||
temperature=job.temperature,
|
||||
initial_prompt=job.initial_prompt,
|
||||
output_directory=job.output_directory
|
||||
)
|
||||
# Start a monitoring thread for timeout warnings
|
||||
timeout_event = threading.Event()
|
||||
|
||||
def timeout_monitor():
|
||||
"""Monitor job execution time and emit warnings."""
|
||||
# Wait for warning threshold
|
||||
if timeout_event.wait(JOB_TIMEOUT_WARNING_SECONDS):
|
||||
return # Job completed before warning threshold
|
||||
|
||||
elapsed = time.time() - start_time
|
||||
logger.warning(
|
||||
f"Job {job.job_id} is taking longer than expected: "
|
||||
f"{elapsed:.1f}s elapsed (threshold: {JOB_TIMEOUT_WARNING_SECONDS}s)"
|
||||
)
|
||||
|
||||
# Wait for max timeout
|
||||
remaining = JOB_TIMEOUT_MAX_SECONDS - elapsed
|
||||
if remaining > 0:
|
||||
if timeout_event.wait(remaining):
|
||||
return # Job completed before max timeout
|
||||
|
||||
# Job exceeded max timeout
|
||||
elapsed = time.time() - start_time
|
||||
logger.error(
|
||||
f"Job {job.job_id} exceeded maximum timeout: "
|
||||
f"{elapsed:.1f}s elapsed (max: {JOB_TIMEOUT_MAX_SECONDS}s)"
|
||||
)
|
||||
|
||||
monitor_thread = threading.Thread(target=timeout_monitor, daemon=True)
|
||||
monitor_thread.start()
|
||||
|
||||
try:
|
||||
result = transcribe_audio(
|
||||
audio_path=job.audio_path,
|
||||
model_name=job.model_name,
|
||||
device=job.device,
|
||||
compute_type=job.compute_type,
|
||||
language=job.language,
|
||||
output_format=job.output_format,
|
||||
beam_size=job.beam_size,
|
||||
temperature=job.temperature,
|
||||
initial_prompt=job.initial_prompt,
|
||||
output_directory=job.output_directory
|
||||
)
|
||||
finally:
|
||||
# Signal timeout monitor to stop
|
||||
timeout_event.set()
|
||||
|
||||
# Check if job exceeded hard timeout
|
||||
elapsed = time.time() - start_time
|
||||
if elapsed > JOB_TIMEOUT_MAX_SECONDS:
|
||||
job.status = JobStatus.FAILED
|
||||
job.error = f"Job exceeded maximum timeout ({JOB_TIMEOUT_MAX_SECONDS}s): {elapsed:.1f}s elapsed"
|
||||
logger.error(f"Job {job.job_id} timed out: {job.error}")
|
||||
# Parse result
|
||||
if "saved to:" in result:
|
||||
elif "saved to:" in result:
|
||||
job.result_path = result.split("saved to:")[1].strip()
|
||||
job.status = JobStatus.COMPLETED
|
||||
logger.info(
|
||||
|
||||
@@ -7,7 +7,8 @@ Responsible for loading, caching, and managing Whisper models
|
||||
import os
|
||||
import time
|
||||
import logging
|
||||
from typing import Dict, Any
|
||||
from typing import Dict, Any, OrderedDict
|
||||
from collections import OrderedDict
|
||||
import torch
|
||||
from faster_whisper import WhisperModel, BatchedInferencePipeline
|
||||
|
||||
@@ -22,8 +23,10 @@ except ImportError:
|
||||
logger.warning("GPU health check with reset not available")
|
||||
GPU_HEALTH_CHECK_AVAILABLE = False
|
||||
|
||||
# Global model instance cache
|
||||
model_instances = {}
|
||||
# Global model instance cache with LRU eviction
|
||||
# Maximum number of models to keep in memory (prevents OOM)
|
||||
MAX_CACHED_MODELS = int(os.getenv("MAX_CACHED_MODELS", "3"))
|
||||
model_instances: OrderedDict[str, Dict[str, Any]] = OrderedDict()
|
||||
|
||||
def test_gpu_driver():
|
||||
"""Simple GPU driver test"""
|
||||
@@ -111,9 +114,11 @@ def get_whisper_model(model_name: str, device: str, compute_type: str) -> Dict[s
|
||||
# Generate model key
|
||||
model_key = f"{model_name}_{device}_{compute_type}"
|
||||
|
||||
# If model is already instantiated, return directly
|
||||
# If model is already instantiated, move to end (mark as recently used) and return
|
||||
if model_key in model_instances:
|
||||
logger.info(f"Using cached model instance: {model_key}")
|
||||
# Move to end for LRU
|
||||
model_instances.move_to_end(model_key)
|
||||
return model_instances[model_key]
|
||||
|
||||
# Test GPU driver before loading model and clean
|
||||
@@ -182,8 +187,33 @@ def get_whisper_model(model_name: str, device: str, compute_type: str) -> Dict[s
|
||||
'load_time': time.time()
|
||||
}
|
||||
|
||||
# Cache instance
|
||||
# Implement LRU eviction before adding new model
|
||||
if len(model_instances) >= MAX_CACHED_MODELS:
|
||||
# Remove oldest (least recently used) model
|
||||
evicted_key, evicted_model = model_instances.popitem(last=False)
|
||||
logger.info(
|
||||
f"Evicting cached model (LRU): {evicted_key} "
|
||||
f"(cache limit: {MAX_CACHED_MODELS})"
|
||||
)
|
||||
|
||||
# Clean up GPU memory if it was a CUDA model
|
||||
if evicted_model['device'] == 'cuda':
|
||||
try:
|
||||
# Delete model references
|
||||
del evicted_model['model']
|
||||
if evicted_model['batched_model'] is not None:
|
||||
del evicted_model['batched_model']
|
||||
torch.cuda.empty_cache()
|
||||
logger.info("GPU memory released for evicted model")
|
||||
except Exception as cleanup_error:
|
||||
logger.warning(f"Error cleaning up evicted model: {cleanup_error}")
|
||||
|
||||
# Cache instance (added to end of OrderedDict)
|
||||
model_instances[model_key] = result
|
||||
logger.info(
|
||||
f"Cached model: {model_key} "
|
||||
f"(cache size: {len(model_instances)}/{MAX_CACHED_MODELS})"
|
||||
)
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
|
||||
@@ -129,31 +129,7 @@ def transcribe_audio(
|
||||
logger.info("Using standard model for transcription...")
|
||||
segments, info = model_instance['model'].transcribe(audio_source, **options)
|
||||
|
||||
# Convert generator to list
|
||||
segment_list = list(segments)
|
||||
|
||||
if not segment_list:
|
||||
return "Transcription failed, no results obtained"
|
||||
|
||||
# Record transcription information
|
||||
elapsed_time = time.time() - start_time
|
||||
logger.info(f"Transcription completed, time used: {elapsed_time:.2f} seconds, detected language: {info.language}, audio length: {info.duration:.2f} seconds")
|
||||
|
||||
# Format transcription results based on output format
|
||||
output_format_lower = output_format.lower()
|
||||
|
||||
if output_format_lower == "vtt":
|
||||
transcription_result = format_vtt(segment_list)
|
||||
elif output_format_lower == "srt":
|
||||
transcription_result = format_srt(segment_list)
|
||||
elif output_format_lower == "txt":
|
||||
transcription_result = format_txt(segment_list)
|
||||
elif output_format_lower == "json":
|
||||
transcription_result = format_json(segment_list, info)
|
||||
else:
|
||||
raise ValueError(f"Unsupported output format: {output_format}. Supported formats: vtt, srt, txt, json")
|
||||
|
||||
# Determine output directory
|
||||
# Determine output directory and path early
|
||||
audio_dir = os.path.dirname(audio_path)
|
||||
audio_filename = os.path.splitext(os.path.basename(audio_path))[0]
|
||||
|
||||
@@ -170,14 +146,14 @@ def transcribe_audio(
|
||||
|
||||
# Generate filename with customizable format
|
||||
filename_parts = []
|
||||
|
||||
|
||||
# Add prefix if specified
|
||||
if FILENAME_PREFIX:
|
||||
filename_parts.append(FILENAME_PREFIX)
|
||||
|
||||
|
||||
# Add base filename
|
||||
filename_parts.append(audio_filename)
|
||||
|
||||
|
||||
# Add suffix if specified
|
||||
if FILENAME_SUFFIX:
|
||||
filename_parts.append(FILENAME_SUFFIX)
|
||||
@@ -188,19 +164,90 @@ def transcribe_audio(
|
||||
filename_parts.append(timestamp)
|
||||
|
||||
# Join parts and add extension
|
||||
output_format_lower = output_format.lower()
|
||||
base_name = "_".join(filename_parts)
|
||||
output_filename = f"{base_name}.{output_format_lower}"
|
||||
output_path = os.path.join(output_dir, output_filename)
|
||||
|
||||
# Write transcription results to file
|
||||
# Stream segments directly to file instead of loading all into memory
|
||||
# This prevents memory spikes with long audio files
|
||||
segment_count = 0
|
||||
try:
|
||||
with open(output_path, "w", encoding="utf-8") as f:
|
||||
f.write(transcription_result)
|
||||
logger.info(f"Transcription results saved to: {output_path}")
|
||||
return f"Transcription successful, results saved to: {output_path}"
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to save transcription results: {str(e)}")
|
||||
return f"Transcription successful, but failed to save results: {str(e)}"
|
||||
# Write format-specific header
|
||||
if output_format_lower == "vtt":
|
||||
f.write("WEBVTT\n\n")
|
||||
elif output_format_lower == "json":
|
||||
f.write('{"segments": [')
|
||||
|
||||
first_segment = True
|
||||
for segment in segments:
|
||||
segment_count += 1
|
||||
|
||||
# Format and write each segment immediately
|
||||
if output_format_lower == "vtt":
|
||||
start_time = format_time(segment.start)
|
||||
end_time = format_time(segment.end)
|
||||
f.write(f"{start_time} --> {end_time}\n{segment.text.strip()}\n\n")
|
||||
elif output_format_lower == "srt":
|
||||
start_time = format_time(segment.start).replace('.', ',')
|
||||
end_time = format_time(segment.end).replace('.', ',')
|
||||
f.write(f"{segment_count}\n{start_time} --> {end_time}\n{segment.text.strip()}\n\n")
|
||||
elif output_format_lower == "txt":
|
||||
f.write(segment.text.strip() + "\n")
|
||||
elif output_format_lower == "json":
|
||||
if not first_segment:
|
||||
f.write(',')
|
||||
import json as json_module
|
||||
segment_dict = {
|
||||
"start": segment.start,
|
||||
"end": segment.end,
|
||||
"text": segment.text.strip()
|
||||
}
|
||||
f.write(json_module.dumps(segment_dict))
|
||||
first_segment = False
|
||||
else:
|
||||
raise ValueError(f"Unsupported output format: {output_format}. Supported formats: vtt, srt, txt, json")
|
||||
|
||||
# Write format-specific footer
|
||||
if output_format_lower == "json":
|
||||
# Add metadata
|
||||
f.write(f'], "language": "{info.language}", "duration": {info.duration}}}')
|
||||
|
||||
except Exception as write_error:
|
||||
logger.error(f"Failed to write transcription during streaming: {str(write_error)}")
|
||||
# File handle automatically closed by context manager
|
||||
# Clean up partial file to prevent corrupted output
|
||||
if os.path.exists(output_path):
|
||||
try:
|
||||
os.remove(output_path)
|
||||
logger.info(f"Cleaned up partial file: {output_path}")
|
||||
except Exception as cleanup_error:
|
||||
logger.warning(f"Failed to cleanup partial file {output_path}: {cleanup_error}")
|
||||
raise
|
||||
|
||||
if segment_count == 0:
|
||||
if info.duration < 1.0:
|
||||
logger.warning(f"No segments: audio too short ({info.duration:.2f}s)")
|
||||
return "Transcription failed: Audio too short (< 1 second)"
|
||||
else:
|
||||
logger.warning(
|
||||
f"No segments generated: duration={info.duration:.2f}s, "
|
||||
f"language={info.language}, vad_enabled=True"
|
||||
)
|
||||
return "Transcription failed: No speech detected (VAD filtered all segments)"
|
||||
|
||||
# Record transcription information
|
||||
elapsed_time = time.time() - start_time
|
||||
logger.info(
|
||||
f"Transcription completed, time used: {elapsed_time:.2f} seconds, "
|
||||
f"detected language: {info.language}, audio length: {info.duration:.2f} seconds, "
|
||||
f"segments: {segment_count}"
|
||||
)
|
||||
|
||||
# File already written via streaming above
|
||||
logger.info(f"Transcription results saved to: {output_path}")
|
||||
return f"Transcription successful, results saved to: {output_path}"
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Transcription failed: {str(e)}")
|
||||
|
||||
@@ -9,12 +9,13 @@ import sys
|
||||
import logging
|
||||
import queue as queue_module
|
||||
import tempfile
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from contextlib import asynccontextmanager
|
||||
from typing import Optional, List
|
||||
from fastapi import FastAPI, HTTPException, UploadFile, File, Form
|
||||
from fastapi.responses import JSONResponse, FileResponse
|
||||
from pydantic import BaseModel, Field
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
import json
|
||||
import aiofiles # Async file I/O
|
||||
|
||||
@@ -22,16 +23,59 @@ from core.model_manager import get_model_info
|
||||
from core.job_queue import JobQueue, JobStatus
|
||||
from core.gpu_health import HealthMonitor, check_gpu_health, get_circuit_breaker_stats, reset_circuit_breaker
|
||||
from utils.startup import startup_sequence, cleanup_on_shutdown
|
||||
from utils.input_validation import (
|
||||
ValidationError,
|
||||
PathTraversalError,
|
||||
InvalidFileTypeError,
|
||||
FileSizeError,
|
||||
validate_beam_size,
|
||||
validate_temperature,
|
||||
validate_model_name,
|
||||
validate_device,
|
||||
validate_compute_type,
|
||||
validate_output_format
|
||||
)
|
||||
|
||||
# Logging configuration
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Constants
|
||||
UPLOAD_CHUNK_SIZE_BYTES = 8192 # 8KB chunks for streaming uploads
|
||||
GPU_TEST_SLOW_THRESHOLD_SECONDS = 2.0 # GPU health check performance threshold
|
||||
DISK_SPACE_BUFFER_PERCENT = 0.10 # Require 10% extra free space as buffer
|
||||
|
||||
# Global instances
|
||||
job_queue: Optional[JobQueue] = None
|
||||
health_monitor: Optional[HealthMonitor] = None
|
||||
|
||||
|
||||
def check_disk_space(path: str, required_bytes: int) -> None:
|
||||
"""
|
||||
Check if sufficient disk space is available.
|
||||
|
||||
Args:
|
||||
path: Path to check disk space for
|
||||
required_bytes: Required bytes
|
||||
|
||||
Raises:
|
||||
IOError: If insufficient disk space
|
||||
"""
|
||||
try:
|
||||
stat = shutil.disk_usage(path)
|
||||
required_with_buffer = required_bytes * (1.0 + DISK_SPACE_BUFFER_PERCENT)
|
||||
|
||||
if stat.free < required_with_buffer:
|
||||
raise IOError(
|
||||
f"Insufficient disk space: {stat.free / 1e9:.1f}GB available, "
|
||||
f"need {required_with_buffer / 1e9:.1f}GB (including {DISK_SPACE_BUFFER_PERCENT*100:.0f}% buffer)"
|
||||
)
|
||||
except IOError:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to check disk space: {e}")
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
"""FastAPI lifespan context manager for startup/shutdown"""
|
||||
@@ -77,6 +121,36 @@ class SubmitJobRequest(BaseModel):
|
||||
initial_prompt: Optional[str] = Field(None, description="Initial prompt text")
|
||||
output_directory: Optional[str] = Field(None, description="Output directory path")
|
||||
|
||||
@field_validator('beam_size')
|
||||
@classmethod
|
||||
def check_beam_size(cls, v):
|
||||
return validate_beam_size(v)
|
||||
|
||||
@field_validator('temperature')
|
||||
@classmethod
|
||||
def check_temperature(cls, v):
|
||||
return validate_temperature(v)
|
||||
|
||||
@field_validator('model_name')
|
||||
@classmethod
|
||||
def check_model_name(cls, v):
|
||||
return validate_model_name(v)
|
||||
|
||||
@field_validator('device')
|
||||
@classmethod
|
||||
def check_device(cls, v):
|
||||
return validate_device(v)
|
||||
|
||||
@field_validator('compute_type')
|
||||
@classmethod
|
||||
def check_compute_type(cls, v):
|
||||
return validate_compute_type(v)
|
||||
|
||||
@field_validator('output_format')
|
||||
@classmethod
|
||||
def check_output_format(cls, v):
|
||||
return validate_output_format(v)
|
||||
|
||||
|
||||
# API Endpoints
|
||||
|
||||
@@ -142,6 +216,21 @@ async def transcribe_upload(
|
||||
"""
|
||||
temp_file_path = None
|
||||
try:
|
||||
# Validate form parameters early
|
||||
try:
|
||||
model = validate_model_name(model)
|
||||
output_format = validate_output_format(output_format)
|
||||
beam_size = validate_beam_size(beam_size)
|
||||
temperature = validate_temperature(temperature)
|
||||
except ValidationError as ve:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error_code": "VALIDATION_ERROR",
|
||||
"error_type": type(ve).__name__,
|
||||
"message": str(ve)
|
||||
}
|
||||
)
|
||||
# Early queue capacity check (backpressure)
|
||||
if job_queue._queue.qsize() >= job_queue._max_queue_size:
|
||||
logger.warning("Job queue is full, rejecting upload before file transfer")
|
||||
@@ -159,6 +248,20 @@ async def transcribe_upload(
|
||||
upload_dir = Path(os.getenv("TRANSCRIPTION_OUTPUT_DIR", "/tmp")) / "uploads"
|
||||
upload_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Check disk space before accepting upload (estimate: file size * 2 for temp + output)
|
||||
if file.size:
|
||||
try:
|
||||
check_disk_space(str(upload_dir), file.size * 2)
|
||||
except IOError as disk_error:
|
||||
logger.error(f"Disk space check failed: {disk_error}")
|
||||
raise HTTPException(
|
||||
status_code=507, # Insufficient Storage
|
||||
detail={
|
||||
"error": "Insufficient disk space",
|
||||
"message": str(disk_error)
|
||||
}
|
||||
)
|
||||
|
||||
# Create temp file with original filename
|
||||
temp_file_path = upload_dir / file.filename
|
||||
|
||||
@@ -167,7 +270,7 @@ async def transcribe_upload(
|
||||
# Save uploaded file using async I/O to avoid blocking event loop
|
||||
async with aiofiles.open(temp_file_path, "wb") as f:
|
||||
# Read file in chunks to handle large files efficiently
|
||||
while chunk := await file.read(8192): # 8KB chunks
|
||||
while chunk := await file.read(UPLOAD_CHUNK_SIZE_BYTES):
|
||||
await f.write(chunk)
|
||||
|
||||
logger.info(f"Saved upload to: {temp_file_path}")
|
||||
@@ -254,6 +357,18 @@ async def submit_job(request: SubmitJobRequest):
|
||||
}
|
||||
)
|
||||
|
||||
except (ValidationError, PathTraversalError, InvalidFileTypeError, FileSizeError) as ve:
|
||||
# Input validation errors
|
||||
logger.error(f"Validation error: {ve}")
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error_code": "VALIDATION_ERROR",
|
||||
"error_type": type(ve).__name__,
|
||||
"message": str(ve)
|
||||
}
|
||||
)
|
||||
|
||||
except queue_module.Full:
|
||||
# Queue is full
|
||||
logger.warning("Job queue is full, rejecting request")
|
||||
@@ -475,7 +590,7 @@ async def gpu_health_check_endpoint():
|
||||
interpretation = "GPU not available on this system"
|
||||
elif not status.gpu_working:
|
||||
interpretation = f"GPU available but not working correctly: {status.error}"
|
||||
elif status.test_duration_seconds > 2.0:
|
||||
elif status.test_duration_seconds > GPU_TEST_SLOW_THRESHOLD_SECONDS:
|
||||
interpretation = f"GPU working but performance degraded (test took {status.test_duration_seconds:.2f}s, expected <1s)"
|
||||
|
||||
return JSONResponse(
|
||||
@@ -564,5 +679,9 @@ if __name__ == "__main__":
|
||||
app,
|
||||
host=host,
|
||||
port=port,
|
||||
log_level="info"
|
||||
log_level="info",
|
||||
timeout_keep_alive=3600, # 1 hour - for long transcription jobs
|
||||
timeout_graceful_shutdown=60,
|
||||
limit_concurrency=10, # Limit concurrent connections
|
||||
backlog=100 # Queue up to 100 pending connections
|
||||
)
|
||||
|
||||
@@ -80,9 +80,11 @@ class CircuitBreaker:
|
||||
self._state = CircuitState.CLOSED
|
||||
self._failure_count = 0
|
||||
self._success_count = 0
|
||||
self._last_failure_time: Optional[datetime] = None
|
||||
# Use monotonic clock for time drift protection
|
||||
self._last_failure_time_monotonic: Optional[float] = None
|
||||
self._last_failure_time_iso: Optional[str] = None # For logging only
|
||||
self._half_open_calls = 0
|
||||
self._lock = threading.RLock()
|
||||
self._lock = threading.RLock() # RLock needed: properties call self.state which acquires lock
|
||||
|
||||
logger.info(
|
||||
f"Circuit breaker '{name}' initialized: "
|
||||
@@ -113,15 +115,20 @@ class CircuitBreaker:
|
||||
return self.state == CircuitState.HALF_OPEN
|
||||
|
||||
def _update_state(self):
|
||||
"""Update state based on timeout and counters."""
|
||||
"""
|
||||
Update state based on timeout and counters.
|
||||
|
||||
Uses monotonic clock to prevent issues with system time changes
|
||||
(e.g., NTP adjustments, daylight saving time, manual clock changes).
|
||||
"""
|
||||
if self._state == CircuitState.OPEN:
|
||||
# Check if timeout has passed
|
||||
if self._last_failure_time:
|
||||
elapsed = datetime.utcnow() - self._last_failure_time
|
||||
if elapsed.total_seconds() >= self.config.timeout_seconds:
|
||||
# Check if timeout has passed using monotonic clock
|
||||
if self._last_failure_time_monotonic is not None:
|
||||
elapsed = time.monotonic() - self._last_failure_time_monotonic
|
||||
if elapsed >= self.config.timeout_seconds:
|
||||
logger.info(
|
||||
f"Circuit '{self.name}': Transitioning to HALF_OPEN "
|
||||
f"after {elapsed.total_seconds():.0f}s timeout"
|
||||
f"after {elapsed:.0f}s timeout"
|
||||
)
|
||||
self._state = CircuitState.HALF_OPEN
|
||||
self._half_open_calls = 0
|
||||
@@ -142,7 +149,8 @@ class CircuitBreaker:
|
||||
self._state = CircuitState.CLOSED
|
||||
self._failure_count = 0
|
||||
self._success_count = 0
|
||||
self._last_failure_time = None
|
||||
self._last_failure_time_monotonic = None
|
||||
self._last_failure_time_iso = None
|
||||
|
||||
elif self._state == CircuitState.CLOSED:
|
||||
# Reset failure count on success
|
||||
@@ -152,7 +160,9 @@ class CircuitBreaker:
|
||||
"""Handle failed call."""
|
||||
with self._lock:
|
||||
self._failure_count += 1
|
||||
self._last_failure_time = datetime.utcnow()
|
||||
# Record failure time using monotonic clock for accuracy
|
||||
self._last_failure_time_monotonic = time.monotonic()
|
||||
self._last_failure_time_iso = datetime.utcnow().isoformat()
|
||||
|
||||
if self._state == CircuitState.HALF_OPEN:
|
||||
logger.warning(
|
||||
@@ -200,11 +210,12 @@ class CircuitBreaker:
|
||||
raise CircuitBreakerOpen(
|
||||
f"Circuit '{self.name}' is OPEN. "
|
||||
f"Failing fast to prevent repeated failures. "
|
||||
f"Last failure: {self._last_failure_time.isoformat() if self._last_failure_time else 'unknown'}. "
|
||||
f"Last failure: {self._last_failure_time_iso or 'unknown'}. "
|
||||
f"Will retry in {self.config.timeout_seconds}s"
|
||||
)
|
||||
|
||||
# Check half-open call limit
|
||||
half_open_incremented = False
|
||||
if self._state == CircuitState.HALF_OPEN:
|
||||
if self._half_open_calls >= self.config.half_open_max_calls:
|
||||
raise CircuitBreakerOpen(
|
||||
@@ -212,6 +223,7 @@ class CircuitBreaker:
|
||||
f"Please wait for current test to complete."
|
||||
)
|
||||
self._half_open_calls += 1
|
||||
half_open_incremented = True
|
||||
|
||||
# Execute function
|
||||
try:
|
||||
@@ -224,9 +236,9 @@ class CircuitBreaker:
|
||||
raise
|
||||
|
||||
finally:
|
||||
# Decrement half-open counter
|
||||
with self._lock:
|
||||
if self._state == CircuitState.HALF_OPEN:
|
||||
# Decrement half-open counter only if we incremented it
|
||||
if half_open_incremented:
|
||||
with self._lock:
|
||||
self._half_open_calls -= 1
|
||||
|
||||
def decorator(self):
|
||||
@@ -260,7 +272,8 @@ class CircuitBreaker:
|
||||
self._state = CircuitState.CLOSED
|
||||
self._failure_count = 0
|
||||
self._success_count = 0
|
||||
self._last_failure_time = None
|
||||
self._last_failure_time_monotonic = None
|
||||
self._last_failure_time_iso = None
|
||||
self._half_open_calls = 0
|
||||
|
||||
def get_stats(self) -> dict:
|
||||
@@ -277,7 +290,7 @@ class CircuitBreaker:
|
||||
"state": self._state.value,
|
||||
"failure_count": self._failure_count,
|
||||
"success_count": self._success_count,
|
||||
"last_failure_time": self._last_failure_time.isoformat() if self._last_failure_time else None,
|
||||
"last_failure_time": self._last_failure_time_iso,
|
||||
"config": {
|
||||
"failure_threshold": self.config.failure_threshold,
|
||||
"success_threshold": self.config.success_threshold,
|
||||
|
||||
@@ -19,6 +19,8 @@ environment=
|
||||
TRANSCRIPTION_MODEL="large-v3",
|
||||
TRANSCRIPTION_DEVICE="auto",
|
||||
TRANSCRIPTION_COMPUTE_TYPE="auto",
|
||||
TRANSCRIPTION_OUTPUT_FORMAT="txt"
|
||||
TRANSCRIPTION_OUTPUT_FORMAT="txt",
|
||||
HTTP_PROXY=http://192.168.1.212:8080,
|
||||
HTTPS_PROXY=http://192.168.1.212:8080
|
||||
stopwaitsecs=10
|
||||
stopsignal=TERM
|
||||
|
||||
Reference in New Issue
Block a user