From 502e566ff0bb5077ada611159bdf799830d50931 Mon Sep 17 00:00:00 2001 From: Rolf Hoefer Date: Wed, 3 Jun 2026 21:22:01 +0800 Subject: [PATCH 1/3] fix(mediorum): stop terminal transcode retries --- pkg/mediorum/server/audio_analysis.go | 6 +- pkg/mediorum/server/audio_analysis_test.go | 37 +++++++- pkg/mediorum/server/transcode.go | 23 ++++- pkg/mediorum/server/transcode_test.go | 101 +++++++++++++++++++++ 4 files changed, 159 insertions(+), 8 deletions(-) diff --git a/pkg/mediorum/server/audio_analysis.go b/pkg/mediorum/server/audio_analysis.go index 710873ab..7eb1509b 100644 --- a/pkg/mediorum/server/audio_analysis.go +++ b/pkg/mediorum/server/audio_analysis.go @@ -117,8 +117,12 @@ func (ss *MediorumServer) findMissedAudioAnalysisCandidates(ctx context.Context, template = 'audio' AND audio_analysis_status IS DISTINCT FROM 'done' AND COALESCE(audio_analysis_error_count, 0) < ? + AND NOT ( + COALESCE(error_count, 0) > ? + AND COALESCE(COALESCE(transcode_results::jsonb, '{}'::jsonb) ->> '320', '') = '' + ) AND (audio_analyzed_at IS NULL OR audio_analyzed_at <= ?) - `, MAX_TRIES, cutoff). + `, MAX_TRIES, missedTranscodeMaxErrorCount, cutoff). Order("COALESCE(audio_analysis_error_count, 0) ASC"). Order("audio_analyzed_at ASC NULLS FIRST"). Order("id ASC"). diff --git a/pkg/mediorum/server/audio_analysis_test.go b/pkg/mediorum/server/audio_analysis_test.go index a7bf6b63..7c6c412e 100644 --- a/pkg/mediorum/server/audio_analysis_test.go +++ b/pkg/mediorum/server/audio_analysis_test.go @@ -72,6 +72,36 @@ func TestFindMissedAudioAnalysisCandidatesBoundsRetriesAndBackoff(t *testing.T) AudioAnalysisStatus: "", AudioAnalyzedAt: time.Time{}, }, + { + ID: prefix + "transcode-terminal-no-result", + Template: JobTemplateAudio, + Status: JobStatusError, + ErrorCount: missedTranscodeMaxErrorCount + 1, + TranscodeResults: map[string]string{}, + AudioAnalysisStatus: "", + AudioAnalysisErrorCount: 0, + AudioAnalyzedAt: time.Time{}, + }, + { + ID: prefix + "transcode-terminal-empty-result", + Template: JobTemplateAudio, + Status: JobStatusError, + ErrorCount: missedTranscodeMaxErrorCount + 1, + TranscodeResults: map[string]string{"320": ""}, + AudioAnalysisStatus: "", + AudioAnalysisErrorCount: 0, + AudioAnalyzedAt: time.Time{}, + }, + { + ID: prefix + "transcode-terminal-with-result", + Template: JobTemplateAudio, + Status: JobStatusError, + ErrorCount: missedTranscodeMaxErrorCount + 1, + TranscodeResults: map[string]string{"320": "cid-320"}, + AudioAnalysisStatus: "", + AudioAnalysisErrorCount: 0, + AudioAnalyzedAt: time.Time{}, + }, } for i := range uploads { require.NoError(t, ss.crud.DB.Create(&uploads[i]).Error) @@ -88,8 +118,9 @@ func TestFindMissedAudioAnalysisCandidatesBoundsRetriesAndBackoff(t *testing.T) } require.Equal(t, map[string]bool{ - prefix + "never-tried": true, - prefix + "old-error": true, - prefix + "max-minus-one": true, + prefix + "never-tried": true, + prefix + "old-error": true, + prefix + "max-minus-one": true, + prefix + "transcode-terminal-with-result": true, }, got) } diff --git a/pkg/mediorum/server/transcode.go b/pkg/mediorum/server/transcode.go index 44cd2ecc..d68d874f 100644 --- a/pkg/mediorum/server/transcode.go +++ b/pkg/mediorum/server/transcode.go @@ -5,6 +5,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "image" "image/gif" @@ -30,9 +31,11 @@ import ( ) var ( - audioPreviewDuration = "30" // seconds - missedTranscodeBatchSize = 100 - missedTranscodeRetryBackoff = time.Minute + audioPreviewDuration = "30" // seconds + missedTranscodeBatchSize = 100 + missedTranscodeMaxErrorCount = 5 + missedTranscodeRetryBackoff = time.Minute + errTranscodeRetryLimitExceeded = errors.New("transcode retry limit exceeded") ) func (ss *MediorumServer) startTranscoder(ctx context.Context) error { @@ -153,7 +156,7 @@ func (ss *MediorumServer) findMissedJobCandidates(ctx context.Context, now time. AND orig_file_cid != '' AND COALESCE(error_count, 0) <= ? AND (transcoded_at IS NULL OR transcoded_at <= ?) - `, JobTemplateAudio, []string{JobStatusNew, JobStatusError}, 5, cutoff). + `, JobTemplateAudio, []string{JobStatusNew, JobStatusError}, missedTranscodeMaxErrorCount, cutoff). Order("transcoded_at ASC NULLS FIRST"). Order("id ASC"). Limit(limit). @@ -390,6 +393,9 @@ func (ss *MediorumServer) transcode(ctx context.Context, upload *Upload) error { if err := ss.crud.DB.Where("id = ?", upload.ID).First(&dbUpload).Error; err != nil { return fmt.Errorf("failed to get upload from DB: %w", err) } + if transcodeRetryLimitExceeded(dbUpload) { + return fmt.Errorf("%w: upload=%s error_count=%d", errTranscodeRetryLimitExceeded, dbUpload.ID, dbUpload.ErrorCount) + } dbUpload.TranscodedBy = ss.Config.Self.Host dbUpload.TranscodedAt = time.Now().UTC() dbUpload.Status = JobStatusBusy @@ -490,6 +496,15 @@ func (ss *MediorumServer) transcode(ctx context.Context, upload *Upload) error { return nil } +func transcodeRetryLimitExceeded(upload Upload) bool { + return upload.ErrorCount > missedTranscodeMaxErrorCount && !hasTranscodeResult(upload) +} + +func hasTranscodeResult(upload Upload) bool { + cid, ok := upload.TranscodeResults["320"] + return ok && cid != "" +} + type FFProbeResult struct { Format struct { Filename string `json:"filename"` diff --git a/pkg/mediorum/server/transcode_test.go b/pkg/mediorum/server/transcode_test.go index 80a0d24a..a9296629 100644 --- a/pkg/mediorum/server/transcode_test.go +++ b/pkg/mediorum/server/transcode_test.go @@ -30,6 +30,69 @@ func TestFilterErrorLines(t *testing.T) { assert.Equal(t, expectedFilteredError, filteredError) } +func TestTranscodeRetryLimitExceededRequiresMissingTranscodeResult(t *testing.T) { + tests := []struct { + name string + upload Upload + want bool + }{ + { + name: "below limit", + upload: Upload{ + Status: JobStatusError, + ErrorCount: missedTranscodeMaxErrorCount, + }, + }, + { + name: "error over limit without result", + upload: Upload{ + Status: JobStatusError, + ErrorCount: missedTranscodeMaxErrorCount + 1, + }, + want: true, + }, + { + name: "busy over limit without result", + upload: Upload{ + Status: JobStatusBusy, + ErrorCount: missedTranscodeMaxErrorCount + 1, + }, + want: true, + }, + { + name: "new over limit without result", + upload: Upload{ + Status: JobStatusNew, + ErrorCount: missedTranscodeMaxErrorCount + 1, + }, + want: true, + }, + { + name: "empty result over limit", + upload: Upload{ + Status: JobStatusError, + ErrorCount: missedTranscodeMaxErrorCount + 1, + TranscodeResults: map[string]string{"320": ""}, + }, + want: true, + }, + { + name: "result present over limit", + upload: Upload{ + Status: JobStatusError, + ErrorCount: missedTranscodeMaxErrorCount + 1, + TranscodeResults: map[string]string{"320": "cid-320"}, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.Equal(t, tt.want, transcodeRetryLimitExceeded(tt.upload)) + }) + } +} + func TestFindMissedJobCandidatesBoundsRetriesAndBackoff(t *testing.T) { ctx := context.Background() ss := testNetwork[0] @@ -154,6 +217,44 @@ func TestFindMissedJobsMarksUploadsBusyBeforeEnqueue(t *testing.T) { } } +func TestTranscodeRetryLimitReturnsBeforeBusyUpdate(t *testing.T) { + ctx := context.Background() + ss := testNetwork[0] + now := time.Now().UTC().Truncate(time.Second) + id := fmt.Sprintf("transcode-limit-%d", now.UnixNano()) + + cleanup := func() { + require.NoError(t, ss.crud.DB.Where("id = ?", id).Delete(&Upload{}).Error) + } + cleanup() + t.Cleanup(cleanup) + + require.NoError(t, ss.crud.DB.Create(&Upload{ + ID: id, + Template: JobTemplateAudio, + OrigFileCID: "cid-terminal", + Status: JobStatusBusy, + ErrorCount: missedTranscodeMaxErrorCount + 1, + TranscodedAt: now.Add(-48 * time.Hour), + }).Error) + + var opsBefore int64 + require.NoError(t, ss.crud.DB.Table("ops").Count(&opsBefore).Error) + + err := ss.transcode(ctx, &Upload{ID: id}) + require.ErrorIs(t, err, errTranscodeRetryLimitExceeded) + + var opsAfter int64 + require.NoError(t, ss.crud.DB.Table("ops").Count(&opsAfter).Error) + require.Equal(t, opsBefore, opsAfter) + + var upload Upload + require.NoError(t, ss.crud.DB.First(&upload, "id = ?", id).Error) + require.Equal(t, JobStatusBusy, upload.Status) + require.Equal(t, missedTranscodeMaxErrorCount+1, upload.ErrorCount) + require.True(t, upload.TranscodedAt.Equal(now.Add(-48*time.Hour))) +} + func uploadIDs(uploads []*Upload) []string { ids := []string{} for _, upload := range uploads { From be76e304f4f20b44adab613ebec65067bf75b880 Mon Sep 17 00:00:00 2001 From: Rolf Hoefer Date: Thu, 4 Jun 2026 00:57:12 +0800 Subject: [PATCH 2/3] fix(mediorum): enforce transcode retry cap boundary --- pkg/mediorum/server/transcode.go | 4 ++-- pkg/mediorum/server/transcode_test.go | 22 +++++++++++----------- 2 files changed, 13 insertions(+), 13 deletions(-) diff --git a/pkg/mediorum/server/transcode.go b/pkg/mediorum/server/transcode.go index d68d874f..1ccf8385 100644 --- a/pkg/mediorum/server/transcode.go +++ b/pkg/mediorum/server/transcode.go @@ -154,7 +154,7 @@ func (ss *MediorumServer) findMissedJobCandidates(ctx context.Context, now time. template = ? AND status in ? AND orig_file_cid != '' - AND COALESCE(error_count, 0) <= ? + AND COALESCE(error_count, 0) < ? AND (transcoded_at IS NULL OR transcoded_at <= ?) `, JobTemplateAudio, []string{JobStatusNew, JobStatusError}, missedTranscodeMaxErrorCount, cutoff). Order("transcoded_at ASC NULLS FIRST"). @@ -497,7 +497,7 @@ func (ss *MediorumServer) transcode(ctx context.Context, upload *Upload) error { } func transcodeRetryLimitExceeded(upload Upload) bool { - return upload.ErrorCount > missedTranscodeMaxErrorCount && !hasTranscodeResult(upload) + return upload.ErrorCount >= missedTranscodeMaxErrorCount && !hasTranscodeResult(upload) } func hasTranscodeResult(upload Upload) bool { diff --git a/pkg/mediorum/server/transcode_test.go b/pkg/mediorum/server/transcode_test.go index a9296629..891ac6df 100644 --- a/pkg/mediorum/server/transcode_test.go +++ b/pkg/mediorum/server/transcode_test.go @@ -37,19 +37,19 @@ func TestTranscodeRetryLimitExceededRequiresMissingTranscodeResult(t *testing.T) want bool }{ { - name: "below limit", + name: "at limit without result", upload: Upload{ Status: JobStatusError, ErrorCount: missedTranscodeMaxErrorCount, }, + want: true, }, { - name: "error over limit without result", + name: "below limit", upload: Upload{ Status: JobStatusError, - ErrorCount: missedTranscodeMaxErrorCount + 1, + ErrorCount: missedTranscodeMaxErrorCount - 1, }, - want: true, }, { name: "busy over limit without result", @@ -118,7 +118,7 @@ func TestFindMissedJobCandidatesBoundsRetriesAndBackoff(t *testing.T) { Template: JobTemplateAudio, OrigFileCID: "cid-too-many-errors", Status: JobStatusError, - ErrorCount: 6, + ErrorCount: missedTranscodeMaxErrorCount, TranscodedAt: now.Add(-48 * time.Hour), }, { @@ -152,11 +152,11 @@ func TestFindMissedJobCandidatesBoundsRetriesAndBackoff(t *testing.T) { TranscodedAt: time.Time{}, }, { - ID: prefix + "007-old-error", + ID: prefix + "007-at-limit", Template: JobTemplateAudio, - OrigFileCID: "cid-old-error", + OrigFileCID: "cid-at-limit", Status: JobStatusError, - ErrorCount: 5, + ErrorCount: missedTranscodeMaxErrorCount, TranscodedAt: now.Add(-48 * time.Hour), }, { @@ -176,7 +176,7 @@ func TestFindMissedJobCandidatesBoundsRetriesAndBackoff(t *testing.T) { require.NoError(t, err) require.Equal(t, []string{ prefix + "006-new", - prefix + "007-old-error", + prefix + "008-old-error", }, uploadIDs(candidates)) } @@ -234,7 +234,7 @@ func TestTranscodeRetryLimitReturnsBeforeBusyUpdate(t *testing.T) { Template: JobTemplateAudio, OrigFileCID: "cid-terminal", Status: JobStatusBusy, - ErrorCount: missedTranscodeMaxErrorCount + 1, + ErrorCount: missedTranscodeMaxErrorCount, TranscodedAt: now.Add(-48 * time.Hour), }).Error) @@ -251,7 +251,7 @@ func TestTranscodeRetryLimitReturnsBeforeBusyUpdate(t *testing.T) { var upload Upload require.NoError(t, ss.crud.DB.First(&upload, "id = ?", id).Error) require.Equal(t, JobStatusBusy, upload.Status) - require.Equal(t, missedTranscodeMaxErrorCount+1, upload.ErrorCount) + require.Equal(t, missedTranscodeMaxErrorCount, upload.ErrorCount) require.True(t, upload.TranscodedAt.Equal(now.Add(-48*time.Hour))) } From 2dab238391f0710648c2d22e3407a48971fb2806 Mon Sep 17 00:00:00 2001 From: Rolf Hoefer Date: Thu, 4 Jun 2026 20:35:10 +0800 Subject: [PATCH 3/3] test(mediorum): align audio analysis transcode boundary --- pkg/mediorum/server/audio_analysis.go | 2 +- pkg/mediorum/server/audio_analysis_test.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/mediorum/server/audio_analysis.go b/pkg/mediorum/server/audio_analysis.go index 7eb1509b..b458f9cd 100644 --- a/pkg/mediorum/server/audio_analysis.go +++ b/pkg/mediorum/server/audio_analysis.go @@ -118,7 +118,7 @@ func (ss *MediorumServer) findMissedAudioAnalysisCandidates(ctx context.Context, AND audio_analysis_status IS DISTINCT FROM 'done' AND COALESCE(audio_analysis_error_count, 0) < ? AND NOT ( - COALESCE(error_count, 0) > ? + COALESCE(error_count, 0) >= ? AND COALESCE(COALESCE(transcode_results::jsonb, '{}'::jsonb) ->> '320', '') = '' ) AND (audio_analyzed_at IS NULL OR audio_analyzed_at <= ?) diff --git a/pkg/mediorum/server/audio_analysis_test.go b/pkg/mediorum/server/audio_analysis_test.go index 7c6c412e..54ac38c6 100644 --- a/pkg/mediorum/server/audio_analysis_test.go +++ b/pkg/mediorum/server/audio_analysis_test.go @@ -76,7 +76,7 @@ func TestFindMissedAudioAnalysisCandidatesBoundsRetriesAndBackoff(t *testing.T) ID: prefix + "transcode-terminal-no-result", Template: JobTemplateAudio, Status: JobStatusError, - ErrorCount: missedTranscodeMaxErrorCount + 1, + ErrorCount: missedTranscodeMaxErrorCount, TranscodeResults: map[string]string{}, AudioAnalysisStatus: "", AudioAnalysisErrorCount: 0,