fix(mediorum): stop terminal transcode retries

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").

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,
+			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)
 }

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 {
@@ -151,9 +154,9 @@ 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}, 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"`

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: "at limit without result",
+			upload: Upload{
+				Status:     JobStatusError,
+				ErrorCount: missedTranscodeMaxErrorCount,
+			},
+			want: true,
+		},
+		{
+			name: "below limit",
+			upload: Upload{
+				Status:     JobStatusError,
+				ErrorCount: missedTranscodeMaxErrorCount - 1,
+			},
+		},
+		{
+			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]
@@ -55,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),
 		},
 		{
@@ -89,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),
 		},
 		{
@@ -113,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))
 }
 
@@ -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,
+		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, upload.ErrorCount)
+	require.True(t, upload.TranscodedAt.Equal(now.Add(-48*time.Hour)))
+}
+
 func uploadIDs(uploads []*Upload) []string {
 	ids := []string{}
 	for _, upload := range uploads {