ben/AndyTranscribe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden stuck Whisper recovery and record transcription duration.

Browse Source 
Fail jobs on timeout, treat stale reserved queue rows as orphans, skip migrate/seed on queue/reverb boot, and store how long each successful run took.
This commit is contained in:
ben
2026-08-13 00:12:06 +02:00
parent 2b126ee4e6
commit f42d124593
7 changed files with 230 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/Models/Recording.php
+97 -41
View File
@@ -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,