diff --git a/README.md b/README.md index f924fe8..fbd2690 100644 --- a/README.md +++ b/README.md @@ -210,7 +210,7 @@ Edit `.env` before `docker compose up` when you need different ports or models: | `REVERB_HOST_PORT` | Host port for WebSockets | `8081` | | `WHISPER_HOST_PORT` | Host port for Whisper | `8090` | | `LOCAL_WHISPER_MODEL` | Whisper model id | `Systran/faster-whisper-base` | -| `TRANSCRIPTION_TIMEOUT` | Job/HTTP timeout (seconds) | `600` | +| `TRANSCRIPTION_TIMEOUT` | Job/HTTP timeout (seconds). Hung Whisper calls fail the job; UI can restart. | `600` | | `DB_QUEUE_RETRY_AFTER` | Must exceed `TRANSCRIPTION_TIMEOUT` | `660` | Inside Compose, Laravel talks to Whisper at `http://whisper:8000/v1` and publishes broadcasts to the `reverb` service. The browser connects to Reverb on `localhost:8081`. diff --git a/app/Jobs/TranscribeRecording.php b/app/Jobs/TranscribeRecording.php index bbe16c9..6b0c4bd 100644 --- a/app/Jobs/TranscribeRecording.php +++ b/app/Jobs/TranscribeRecording.php @@ -6,9 +6,11 @@ use App\Models\Recording; use App\Services\TranscriptionService; use Illuminate\Contracts\Queue\ShouldQueue; use Illuminate\Foundation\Queue\Queueable; +use Illuminate\Queue\Attributes\FailOnTimeout; use Illuminate\Support\Facades\Log; use Throwable; +#[FailOnTimeout] class TranscribeRecording implements ShouldQueue { use Queueable; @@ -20,6 +22,9 @@ class TranscribeRecording implements ShouldQueue /** * The number of seconds the job can run before timing out. + * + * Covers a hung Whisper HTTP call: the worker is killed, failed() runs, + * and the recording is marked failed so the UI can restart. */ public int $timeout; @@ -134,16 +139,7 @@ class TranscribeRecording implements ShouldQueue } if ($this->recording->ownsTranscriptionRun($this->runStartedAt)) { - $this->recording->forceFill([ - 'transcript' => $text, - 'transcription_status' => 'done', - 'transcription_progress' => 'Transcription complete', - 'transcription_percent' => 100, - 'transcription_error' => null, - 'transcribed_at' => now(), - ])->save(); - - $this->recording->broadcastTranscriptionUpdated(); + $this->recording->markTranscriptionComplete($text); return true; } @@ -152,22 +148,16 @@ class TranscribeRecording implements ShouldQueue if ( $this->recording->transcription_status === 'failed' && $this->recording->matchesTranscriptionRun($this->runStartedAt) - && str_contains((string) $this->recording->transcription_error, 'worker stopped') + && ( + str_contains((string) $this->recording->transcription_error, 'worker stopped') + || str_contains((string) $this->recording->transcription_error, 'timed out') + ) ) { Log::warning('Recovering transcript after false orphan failure', [ 'recording_id' => $this->recording->id, ]); - $this->recording->forceFill([ - 'transcript' => $text, - 'transcription_status' => 'done', - 'transcription_progress' => 'Transcription complete', - 'transcription_percent' => 100, - 'transcription_error' => null, - 'transcribed_at' => now(), - ])->save(); - - $this->recording->broadcastTranscriptionUpdated(); + $this->recording->markTranscriptionComplete($text); return true; } diff --git a/app/Models/Recording.php b/app/Models/Recording.php index a088274..6262cf2 100644 --- a/app/Models/Recording.php +++ b/app/Models/Recording.php @@ -41,6 +41,7 @@ class Recording extends Model 'transcription_driver', 'ollama_url', 'transcribed_at', + 'transcription_duration_seconds', ]; /** @@ -53,6 +54,7 @@ class Recording extends Model 'transcribed_at' => 'datetime', 'transcription_started_at' => 'datetime', 'duration_seconds' => 'integer', + 'transcription_duration_seconds' => 'integer', 'file_size_bytes' => 'integer', 'transcription_percent' => 'integer', 'user_id' => 'integer', @@ -138,6 +140,7 @@ class Recording extends Model 'transcription_percent' => null, 'transcription_started_at' => now(), 'transcription_error' => null, + 'transcription_duration_seconds' => null, // Keep the previous transcript until a new run succeeds. 'transcribed_at' => $this->transcribed_at, ]); @@ -228,33 +231,40 @@ class Recording extends Model return ($start > 0 ? '…' : '').$excerpt.(mb_strlen($transcript) > $start + mb_strlen($excerpt) ? '…' : ''); } + /** + * Seconds after which a reserved queue row is considered abandoned + * (worker died mid-Whisper without releasing the job). + */ + public function transcriptionJobStaleAfterSeconds(): int + { + return max(120, (int) config('ai.transcription_timeout', 600) + 90); + } + /** * Whether a TranscribeRecording job for this recording is still on the queue. + * + * Reserved jobs older than the transcription timeout (+ grace) are ignored so + * orphan recovery can unblock the UI when Whisper/the worker is wedged. */ public function hasActiveTranscriptionJob(): bool { + $staleBefore = now()->timestamp - $this->transcriptionJobStaleAfterSeconds(); + return DB::table('jobs') - ->pluck('payload') - ->contains(function (string $payload): bool { - if (! str_contains($payload, TranscribeRecording::class)) { + ->orderBy('id') + ->get(['id', 'payload', 'reserved_at']) + ->contains(function (object $job) use ($staleBefore): bool { + $payload = (string) $job->payload; + + if (! str_contains($payload, 'TranscribeRecording')) { return false; } - $data = json_decode($payload, true); - $command = $data['data']['command'] ?? null; - - if (! is_string($command)) { + if ($job->reserved_at !== null && (int) $job->reserved_at <= $staleBefore) { return false; } - try { - $job = unserialize($command); - } catch (Throwable) { - return $this->payloadMentionsRecording($payload); - } - - return $job instanceof TranscribeRecording - && (int) $job->recording->getKey() === (int) $this->id; + return $this->jobPayloadBelongsToRecording($payload); }); } @@ -263,8 +273,11 @@ class Recording extends Model */ private function payloadMentionsRecording(string $payload): bool { - return (bool) preg_match('/id";i:'.$this->id.';/', $payload) - || str_contains($payload, 'id";s:'.strlen((string) $this->id).':"'.$this->id.'"'); + // Jobs table stores JSON; the serialized command inside escapes quotes as \". + return (bool) preg_match('/id\\\\";i:'.$this->id.';/', $payload) + || (bool) preg_match('/id";i:'.$this->id.';/', $payload) + || str_contains($payload, 'id";s:'.strlen((string) $this->id).':"'.$this->id.'"') + || str_contains($payload, 'id\\\\";s:'.strlen((string) $this->id).':\\"'.$this->id.'\\"'); } /** @@ -332,34 +345,16 @@ class Recording extends Model ->each(function (object $job) use (&$deleted): void { $payload = (string) $job->payload; - if (! str_contains($payload, TranscribeRecording::class)) { + if (! str_contains($payload, 'TranscribeRecording')) { return; } - $data = json_decode($payload, true); - $command = $data['data']['command'] ?? null; - - if (! is_string($command)) { + if (! $this->jobPayloadBelongsToRecording($payload)) { return; } - try { - $queued = unserialize($command); - } catch (Throwable) { - if (! preg_match('/id";i:'.$this->id.';/', $payload)) { - return; - } - - DB::table('jobs')->where('id', $job->id)->delete(); - $deleted++; - - return; - } - - if ($queued instanceof TranscribeRecording && (int) $queued->recording->getKey() === (int) $this->id) { - DB::table('jobs')->where('id', $job->id)->delete(); - $deleted++; - } + DB::table('jobs')->where('id', $job->id)->delete(); + $deleted++; }); $this->releaseTranscriptionUniqueLock(); @@ -367,6 +362,36 @@ class Recording extends Model return $deleted; } + /** + * Whether a jobs.payload row targets this recording. + */ + private function jobPayloadBelongsToRecording(string $payload): bool + { + $data = json_decode($payload, true); + $command = $data['data']['command'] ?? null; + + if (is_string($command)) { + if ( + preg_match('/id";i:'.$this->id.';/', $command) + || str_contains($command, 'id";s:'.strlen((string) $this->id).':"'.$this->id.'"') + ) { + return true; + } + + try { + $queued = unserialize($command); + + if ($queued instanceof TranscribeRecording) { + return (int) $queued->recording->getKey() === (int) $this->id; + } + } catch (Throwable) { + // Fall through to escaped JSON heuristics. + } + } + + return $this->payloadMentionsRecording($payload); + } + /** * Stop transcription: drop queued jobs and mark the run cancelled. * @@ -400,6 +425,30 @@ class Recording extends Model )->forceRelease(); } + /** + * Persist a successful transcript and how long the run took. + */ + public function markTranscriptionComplete(string $text): void + { + $finishedAt = now(); + $startedAt = $this->transcription_started_at; + $durationSeconds = $startedAt === null + ? null + : max(0, (int) round($startedAt->diffInSeconds($finishedAt))); + + $this->forceFill([ + 'transcript' => $text, + 'transcription_status' => 'done', + 'transcription_progress' => 'Transcription complete', + 'transcription_percent' => 100, + 'transcription_error' => null, + 'transcribed_at' => $finishedAt, + 'transcription_duration_seconds' => $durationSeconds, + ])->save(); + + RecordingTranscriptionUpdated::dispatch($this->fresh()); + } + /** * Mark transcription as failed and unblock the UI. */ @@ -420,7 +469,7 @@ class Recording extends Model } /** - * Recover a stuck transcription if the queue job is gone. + * Recover a stuck transcription if the queue job is gone or a reservation is stale. */ public function recoverOrphanedTranscription(): bool { @@ -428,8 +477,11 @@ class Recording extends Model return false; } + // Drop abandoned reserved rows so a restart can enqueue cleanly. + $this->discardQueuedTranscriptionJobs(); + $this->markTranscriptionFailed( - 'Transcription worker stopped before finishing. Start transcription again.', + 'Transcription timed out or the worker stopped before finishing. Start transcription again.', ); return true; @@ -519,6 +571,10 @@ class Recording extends Model 'elapsed_seconds' => $elapsed, 'elapsed_human' => $elapsed === null ? null : $this->formatElapsed($elapsed), 'duration_seconds' => $this->duration_seconds, + 'transcription_duration_seconds' => $this->transcription_duration_seconds, + 'transcription_duration_human' => $this->transcription_duration_seconds === null + ? null + : $this->formatElapsed($this->transcription_duration_seconds), 'is_active' => $this->isTranscribing(), 'has_transcript' => filled($this->transcript), 'transcript' => $this->transcript, diff --git a/database/migrations/2026_08_12_220211_add_transcription_duration_seconds_to_recordings_table.php b/database/migrations/2026_08_12_220211_add_transcription_duration_seconds_to_recordings_table.php new file mode 100644 index 0000000..22cb481 --- /dev/null +++ b/database/migrations/2026_08_12_220211_add_transcription_duration_seconds_to_recordings_table.php @@ -0,0 +1,30 @@ +unsignedInteger('transcription_duration_seconds') + ->nullable() + ->after('transcribed_at'); + }); + } + + /** + * Reverse the migrations. + */ + public function down(): void + { + Schema::table('recordings', function (Blueprint $table) { + $table->dropColumn('transcription_duration_seconds'); + }); + } +}; diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index 2c4cf57..bb1901e 100755 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -36,7 +36,22 @@ if [ ! -f vendor/autoload.php ]; then composer install --prefer-dist --no-interaction fi -php artisan migrate --force --no-interaction -php artisan db:seed --force --no-interaction +# Only the web app should migrate/seed. Queue and Reverb share the DB and must +# not race on sqlite (locks) or re-seed on every restart/deploy. +should_bootstrap_db() { + case " $* " in + *" queue:work "*|*" queue:listen "*|*" reverb:start "*) + return 1 + ;; + *) + return 0 + ;; + esac +} + +if should_bootstrap_db "$@"; then + php artisan migrate --force --no-interaction + php artisan db:seed --force --no-interaction +fi exec "$@" diff --git a/resources/views/livewire/recordings/show.blade.php b/resources/views/livewire/recordings/show.blade.php index 173c686..01b0c34 100644 --- a/resources/views/livewire/recordings/show.blade.php +++ b/resources/views/livewire/recordings/show.blade.php @@ -216,7 +216,10 @@ diff --git a/tests/Feature/RecordingUploadTest.php b/tests/Feature/RecordingUploadTest.php index 8f189e0..a0b17ed 100644 --- a/tests/Feature/RecordingUploadTest.php +++ b/tests/Feature/RecordingUploadTest.php @@ -13,6 +13,7 @@ use App\Services\TranscriptionService; use Illuminate\Foundation\Testing\RefreshDatabase; use Illuminate\Http\UploadedFile; use Illuminate\Support\Facades\Bus; +use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Storage; use Laravel\Ai\Transcription; use Livewire\Livewire; @@ -209,6 +210,8 @@ class RecordingUploadTest extends TestCase Transcription::fake(['Hello from the recorder.']); + $this->freezeTime(); + $recording = Recording::query()->create([ 'user_id' => $this->user->id, 'title' => 'Sample', @@ -217,6 +220,7 @@ class RecordingUploadTest extends TestCase 'file_size_bytes' => 12, 'transcription_status' => 'pending', 'transcription_driver' => 'local', + 'transcription_started_at' => now()->subSeconds(42), ]); (new TranscribeRecording($recording))->handle(app(TranscriptionService::class)); @@ -227,6 +231,8 @@ class RecordingUploadTest extends TestCase $this->assertSame(100, $recording->transcription_percent); $this->assertSame('Transcription complete', $recording->transcription_progress); $this->assertNotNull($recording->transcribed_at); + $this->assertSame(42, $recording->transcription_duration_seconds); + $this->assertSame('42s', $recording->transcriptionStatusPayload()['transcription_duration_human']); } public function test_transcription_job_stores_error_on_failure(): void @@ -283,7 +289,70 @@ class RecordingUploadTest extends TestCase $recording->refresh(); $this->assertSame('failed', $recording->transcription_status); - $this->assertStringContainsString('worker stopped', $recording->transcription_error); + $this->assertStringContainsString('timed out', $recording->transcription_error); + } + + public function test_stale_reserved_job_is_treated_as_orphaned(): void + { + $recording = Recording::query()->create([ + 'user_id' => $this->user->id, + 'title' => 'Wedged whisper', + 'original_filename' => 'wedged.mp3', + 'file_path' => 'recordings/wedged.mp3', + 'file_size_bytes' => 100, + 'transcription_status' => 'processing', + 'transcription_progress' => 'Transcribing locally…', + 'transcription_percent' => 50, + 'transcription_driver' => 'local', + 'transcription_started_at' => now()->subMinutes(20), + 'updated_at' => now()->subMinutes(20), + ]); + + $job = new TranscribeRecording($recording); + $payload = json_encode([ + 'displayName' => TranscribeRecording::class, + 'data' => [ + 'command' => serialize($job), + ], + ], JSON_THROW_ON_ERROR); + + DB::table('jobs')->insert([ + 'queue' => 'default', + 'payload' => $payload, + 'attempts' => 1, + 'reserved_at' => now()->subMinutes(15)->timestamp, + 'available_at' => now()->subMinutes(20)->timestamp, + 'created_at' => now()->subMinutes(20)->timestamp, + ]); + + $this->assertFalse($recording->hasActiveTranscriptionJob()); + $this->assertTrue($recording->isOrphanedTranscription()); + + $deleted = $recording->discardQueuedTranscriptionJobs(); + $this->assertSame(1, $deleted, 'stale reserved job should be discarded'); + $this->assertDatabaseCount('jobs', 0); + + // Put the stale job back to exercise status-poll recovery. + DB::table('jobs')->insert([ + 'queue' => 'default', + 'payload' => $payload, + 'attempts' => 1, + 'reserved_at' => now()->subMinutes(15)->timestamp, + 'available_at' => now()->subMinutes(20)->timestamp, + 'created_at' => now()->subMinutes(20)->timestamp, + ]); + $recording->forceFill([ + 'transcription_status' => 'processing', + 'transcription_error' => null, + ])->save(); + + $this->getJson(route('recordings.transcription-status', $recording)) + ->assertOk() + ->assertJsonPath('status', 'failed'); + + $this->assertDatabaseCount('jobs', 0); + $recording->refresh(); + $this->assertSame('failed', $recording->transcription_status); } public function test_recent_processing_is_not_marked_orphaned(): void