Use errgroup for bounded transcription work

docs/internal/stage-transcribe.md

@@ -16,7 +16,7 @@ Generate raw per-speaker transcripts from prepared audio using WhisperX.
 
 - discovers prepared audio from manifest inputs or canonical audio directory.
 - derives the transcript identity from the prepared `.flac` filename.
-- dispatches WhisperX requests through a bounded worker pool.
+- dispatches one WhisperX request per prepared file through a bounded errgroup.
 - validates each output as JSON.
 - writes run-local outputs then materializes canonical transcript outputs only
   after every planned request succeeds.

docs/roadmap/implementation.md

@@ -372,7 +372,7 @@ own those files.
 
 ## Stage 5 — Replace The Transcription Worker Pool With Errgroup
 
-**Status: Pending**
+**Status: Completed**
 
 ### Goal
 

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/credentials v1.19.16
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.101.0
 	github.com/aws/smithy-go v1.25.1
+	golang.org/x/sync v0.22.0
 	golang.org/x/sys v0.47.0
 	gopkg.in/yaml.v3 v3.0.1
 )

go.sum

@@ -34,6 +34,8 @@ github.com/aws/aws-sdk-go-v2/service/sts v1.42.1 h1:F/M5Y9I3nwr2IEpshZgh1GeHpOIt
 github.com/aws/aws-sdk-go-v2/service/sts v1.42.1/go.mod h1:mTNxImtovCOEEuD65mKW7DCsL+2gjEH+RPEAexAzAio=
 github.com/aws/smithy-go v1.25.1 h1:J8ERsGSU7d+aCmdQur5Txg6bVoYelvQJgtZehD12GkI=
 github.com/aws/smithy-go v1.25.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc=
+golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
+golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
 golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=

internal/stage/transcribe.go

@@ -10,6 +10,8 @@ import (
 	"strings"
 	"sync"
 
+	"golang.org/x/sync/errgroup"
+
 	"gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx"
 	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
 	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
@@ -93,106 +95,61 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest)
 		})
 	}
 
-	stageCtx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
 	var (
-		wg        sync.WaitGroup
-		jobCh     = make(chan job)
 		mu        sync.Mutex
-		firstErr  error
 		completed int
 		perFile   = map[string]map[string]any{}
-		outputRef = map[string]artifacts.Ref{}
 	)
-
+	group, groupCtx := errgroup.WithContext(ctx)
-	worker := func() {
+	group.SetLimit(concurrency)
-		defer wg.Done()
+	for _, job := range jobs {
-		for j := range jobCh {
+		job := job
-			if stageCtx.Err() != nil {
+		group.Go(func() error {
-				return
+			if err := groupCtx.Err(); err != nil {
+				return err
 			}
-
+			res, runErr := env.WhisperX.Transcribe(groupCtx, whisperx.TranscribeRequest{
-			res, runErr := env.WhisperX.Transcribe(stageCtx, whisperx.TranscribeRequest{
+				SpeakerID:               job.speakerID,
-				SpeakerID:               j.speakerID,
+				AudioPath:               job.audioPath,
-				AudioPath:               j.audioPath,
+				OutputRawTranscriptPath: job.outPath,
-				OutputRawTranscriptPath: j.outPath,
 			})
 			if runErr != nil {
-				mu.Lock()
+				return fmt.Errorf("speaker %q (%s): %w", job.speakerID, filepath.Base(job.audioPath), runErr)
-				if firstErr == nil {
-					firstErr = fmt.Errorf("speaker %q (%s): %w", j.speakerID, filepath.Base(j.audioPath), runErr)
-					cancel()
-				}
-				mu.Unlock()
-				continue
 			}
 
-			if _, err := authoritativeOutputPath(j.outPath, res.OutputRawTranscriptPath); err != nil {
+			if _, err := authoritativeOutputPath(job.outPath, res.OutputRawTranscriptPath); err != nil {
-				mu.Lock()
+				return fmt.Errorf("speaker %q: %w", job.speakerID, err)
-				if firstErr == nil {
-					firstErr = fmt.Errorf("speaker %q: %w", j.speakerID, err)
-					cancel()
-				}
-				mu.Unlock()
-				continue
 			}
-			if err := validateTranscriptJSONFile(j.outPath); err != nil {
+			if err := validateTranscriptJSONFile(job.outPath); err != nil {
-				mu.Lock()
+				return fmt.Errorf("speaker %q output %q invalid: %w", job.speakerID, job.outPath, err)
-				if firstErr == nil {
-					firstErr = fmt.Errorf("speaker %q output %q invalid: %w", j.speakerID, j.outPath, err)
-					cancel()
-				}
-				mu.Unlock()
-				continue
 			}
 
 			meta := map[string]any{
 				"attempts":    res.Attempts,
 				"http_status": res.HTTPStatus,
 				"duration_ms": res.Duration.Milliseconds(),
-				"output_path": j.outPath,
+				"output_path": job.outPath,
 			}
 
 			mu.Lock()
-			perFile[j.speakerID] = meta
+			perFile[job.speakerID] = meta
-			outputRef[j.speakerID] = artifacts.Ref{
-				Kind:         "transcript_raw",
-				Category:     "transcripts",
-				SessionID:    sessionID,
-				AbsolutePath: j.outPath,
-			}
 			completed++
 			mu.Unlock()
-		}
+			return nil
-	}
+		})
-
-	wg.Add(concurrency)
-	for i := 0; i < concurrency; i++ {
-		go worker()
-	}
-
-	dispatched := 0
-dispatch:
-	for _, j := range jobs {
-		select {
-		case <-stageCtx.Done():
-			break dispatch
-		case jobCh <- j:
-			dispatched++
-		}
-	}
-	close(jobCh)
-	wg.Wait()
-
-	if firstErr != nil {
-		return nil, firstErr
 	}
+	groupErr := group.Wait()
+	mu.Lock()
+	completedCount := completed
+	mu.Unlock()
 	if err := ctx.Err(); err != nil {
-		return nil, fmt.Errorf("transcribe: canceled after dispatching %d of %d audio jobs and completing %d: %w", dispatched, len(jobs), completed, err)
+		return nil, fmt.Errorf("transcribe: canceled after completing %d of %d audio jobs: %w", completedCount, len(jobs), err)
 	}
-	if completed != len(jobs) {
+	if groupErr != nil {
-		return nil, fmt.Errorf("transcribe: completed %d of %d planned audio jobs after dispatching %d", completed, len(jobs), dispatched)
+		return nil, groupErr
+	}
+	if completedCount != len(jobs) {
+		return nil, fmt.Errorf("transcribe: completed %d of %d planned audio jobs", completedCount, len(jobs))
 	}
 
 	speakers := make([]string, 0, len(perFile))
@@ -206,10 +163,17 @@ dispatch:
 	outputPaths := make([]string, 0, len(speakers))
 	orderedPerFile := make(map[string]any, len(speakers))
 	for _, speaker := range speakers {
-		ref := outputRef[speaker]
+		var runOut string
-		runOutputPaths = append(runOutputPaths, ref.AbsolutePath)
+		for _, job := range jobs {
+			if job.speakerID == speaker {
+				runOut = job.outPath
+				break
+			}
+		}
+		runOutputPaths = append(runOutputPaths, runOut)
 		canonicalOut := filepath.Join(paths.TranscriptsRawDir, speaker+".json")
-		materialized, err := materializeRunLocalOutput(env.ArtifactStore, ref.AbsolutePath, canonicalOut, ref)
+		ref := artifacts.Ref{Kind: "transcript_raw", Category: "transcripts", SessionID: sessionID, AbsolutePath: runOut}
+		materialized, err := materializeRunLocalOutput(env.ArtifactStore, runOut, canonicalOut, ref)
 		if err != nil {
 			return nil, fmt.Errorf("transcribe: materialize %q output: %w", speaker, err)
 		}