diff --git a/docs/internal/stage-transcribe.md b/docs/internal/stage-transcribe.md index e8a8acb..b26f8be 100644 --- a/docs/internal/stage-transcribe.md +++ b/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. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 420c071..63668fd 100644 --- a/docs/roadmap/implementation.md +++ b/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 diff --git a/go.mod b/go.mod index 2a376a7..3749684 100644 --- a/go.mod +++ b/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 ) diff --git a/go.sum b/go.sum index 89e7212..0308ba4 100644 --- a/go.sum +++ b/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= diff --git a/internal/stage/transcribe.go b/internal/stage/transcribe.go index 36991dc..0415556 100644 --- a/internal/stage/transcribe.go +++ b/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{} ) - - worker := func() { - defer wg.Done() - for j := range jobCh { - if stageCtx.Err() != nil { - return + group, groupCtx := errgroup.WithContext(ctx) + group.SetLimit(concurrency) + for _, job := range jobs { + job := job + group.Go(func() error { + if err := groupCtx.Err(); err != nil { + return err } - - res, runErr := env.WhisperX.Transcribe(stageCtx, whisperx.TranscribeRequest{ - SpeakerID: j.speakerID, - AudioPath: j.audioPath, - OutputRawTranscriptPath: j.outPath, + res, runErr := env.WhisperX.Transcribe(groupCtx, whisperx.TranscribeRequest{ + SpeakerID: job.speakerID, + AudioPath: job.audioPath, + OutputRawTranscriptPath: job.outPath, }) if runErr != nil { - mu.Lock() - if firstErr == nil { - firstErr = fmt.Errorf("speaker %q (%s): %w", j.speakerID, filepath.Base(j.audioPath), runErr) - cancel() - } - mu.Unlock() - continue + return fmt.Errorf("speaker %q (%s): %w", job.speakerID, filepath.Base(job.audioPath), runErr) } - if _, err := authoritativeOutputPath(j.outPath, res.OutputRawTranscriptPath); err != nil { - mu.Lock() - if firstErr == nil { - firstErr = fmt.Errorf("speaker %q: %w", j.speakerID, err) - cancel() - } - mu.Unlock() - continue + if _, err := authoritativeOutputPath(job.outPath, res.OutputRawTranscriptPath); err != nil { + return fmt.Errorf("speaker %q: %w", job.speakerID, err) } - if err := validateTranscriptJSONFile(j.outPath); err != nil { - mu.Lock() - if firstErr == nil { - firstErr = fmt.Errorf("speaker %q output %q invalid: %w", j.speakerID, j.outPath, err) - cancel() - } - mu.Unlock() - continue + if err := validateTranscriptJSONFile(job.outPath); err != nil { + return fmt.Errorf("speaker %q output %q invalid: %w", job.speakerID, job.outPath, err) } 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 - outputRef[j.speakerID] = artifacts.Ref{ - Kind: "transcript_raw", - Category: "transcripts", - SessionID: sessionID, - AbsolutePath: j.outPath, - } + perFile[job.speakerID] = meta completed++ mu.Unlock() - } - } - - 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 + return nil + }) } + 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) { - return nil, fmt.Errorf("transcribe: completed %d of %d planned audio jobs after dispatching %d", completed, len(jobs), dispatched) + if groupErr != nil { + 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] - runOutputPaths = append(runOutputPaths, ref.AbsolutePath) + var runOut string + 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) }