Files
narratio/internal/app/transcribe_checkpoint_recovery_test.go

473 lines
18 KiB
Go

package app
import (
"context"
"errors"
"os"
"path/filepath"
"slices"
"strings"
"testing"
"time"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
"gitea.maximumdirect.net/eric/narratio/internal/stage"
)
func TestExecuteStagesDelegatesEmptyAudioAcceptanceToWhisperX(t *testing.T) {
for _, rejected := range []bool{false, true} {
name := "accepted"
if rejected {
name = "rejected"
}
t.Run(name, func(t *testing.T) {
cfg, store := setupTranscribeRecoveryFixture(t, []string{"alice"})
paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
if err := os.WriteFile(filepath.Join(paths.AudioDir, "alice.flac"), nil, 0o644); err != nil {
t.Fatal(err)
}
fake := &whisperx.FakeClient{}
if rejected {
fake.Err = errors.New("service rejected audio")
}
transcribe := mustSelectStage(t, "transcribe")
_, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: store, WhisperX: fake},
})
if !errors.Is(err, fake.Err) {
t.Fatalf("transcribe error = %v, want %v", err, fake.Err)
}
if len(fake.RequestsSnapshot()) != 1 {
t.Fatal("empty audio was not sent to service")
}
m := loadTranscribeRecoveryManifest(t, store, cfg)
if rejected {
if len(m.Stages["transcribe"].TranscribeSpeakers) != 0 {
t.Fatal("rejected audio was checkpointed")
}
return
}
record, ok := m.Stages["transcribe"].TranscribeSpeakers["alice"]
if !ok || record.AudioSize != 0 {
t.Fatalf("missing zero-size checkpoint: %#v", record)
}
retry := &whisperx.FakeClient{}
summary, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: store, WhisperX: retry},
})
if err != nil {
t.Fatal(err)
}
if len(summary.Skipped) != 1 || len(retry.RequestsSnapshot()) != 0 {
t.Fatal("successful empty-audio result was not reused")
}
})
}
}
func TestExecuteStagesRetriesOnlyUncommittedTranscribeSpeakers(t *testing.T) {
cfg, baseStore := setupTranscribeRecoveryFixture(t, []string{"a", "b", "c", "d"})
concurrency := 1
cfg.Pipeline.WhisperX.Concurrency = &concurrency
cause := errors.New("speaker c failed")
failing := &whisperx.FakeClient{TranscribeFn: func(
_ context.Context,
req whisperx.TranscribeRequest,
) (whisperx.TranscribeResult, error) {
if req.SpeakerID == "c" {
return whisperx.TranscribeResult{}, cause
}
if req.SpeakerID == "d" {
t.Fatalf("speaker d was dispatched after c failed")
}
return writeTranscribeRecoveryResult(req)
}}
transcribe := mustSelectStage(t, "transcribe")
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: baseStore, WhisperX: failing},
}); !errors.Is(err, cause) {
t.Fatalf("failed executeStages() error = %v, want %v", err, cause)
}
failed := loadTranscribeRecoveryManifest(t, baseStore, cfg)
if failed.Stages["transcribe"].Status != manifest.StatusFailed {
t.Fatalf("transcribe status = %q, want failed", failed.Stages["transcribe"].Status)
}
if got := transcribeSpeakerKeys(failed); !slices.Equal(got, []string{"a", "b"}) {
t.Fatalf("checkpointed speakers = %#v, want a and b", got)
}
if failed.Stages["prepare"].Status != manifest.StatusSucceeded {
t.Fatalf("prepare status = %q, want unchanged success", failed.Stages["prepare"].Status)
}
if failed.Stages["merge"].Status != manifest.StatusStale {
t.Fatalf("merge status = %q, want stale after transcribe failure", failed.Stages["merge"].Status)
}
retry := &whisperx.FakeClient{}
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: baseStore, WhisperX: retry},
}); err != nil {
t.Fatalf("retry executeStages() error = %v", err)
}
if got := requestedTranscribeSpeakers(retry); !slices.Equal(got, []string{"c", "d"}) {
t.Fatalf("retry speakers = %#v, want c and d", got)
}
completed := loadTranscribeRecoveryManifest(t, baseStore, cfg)
if completed.Stages["transcribe"].Status != manifest.StatusSucceeded {
t.Fatalf("transcribe status = %q, want succeeded", completed.Stages["transcribe"].Status)
}
if got := transcribeSpeakerKeys(completed); !slices.Equal(got, []string{"a", "b", "c", "d"}) {
t.Fatalf("completed speakers = %#v", got)
}
}
func TestExecuteStagesRebuildsTranscribeAggregateAfterFinalSaveFailure(t *testing.T) {
cfg, baseStore := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"})
store := &transcribeAggregateSaveFailingStore{
Store: baseStore,
Err: errors.New("aggregate save failed"),
}
fake := &whisperx.FakeClient{}
transcribe := mustSelectStage(t, "transcribe")
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: store, WhisperX: fake},
}); !errors.Is(err, store.Err) {
t.Fatalf("executeStages() error = %v, want aggregate save failure", err)
}
failed := loadTranscribeRecoveryManifest(t, baseStore, cfg)
if failed.Stages["transcribe"].Status != manifest.StatusFailed {
t.Fatalf("transcribe status = %q, want failed", failed.Stages["transcribe"].Status)
}
if got := transcribeSpeakerKeys(failed); !slices.Equal(got, []string{"alice", "bob"}) {
t.Fatalf("checkpointed speakers = %#v, want both speakers", got)
}
retry := &whisperx.FakeClient{}
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: baseStore, WhisperX: retry},
}); err != nil {
t.Fatalf("retry executeStages() error = %v", err)
}
if got := len(retry.RequestsSnapshot()); got != 0 {
t.Fatalf("retry adapter requests = %d, want aggregate-only rebuild", got)
}
completed := loadTranscribeRecoveryManifest(t, baseStore, cfg)
if completed.Stages["transcribe"].Status != manifest.StatusSucceeded || len(completed.Stages["transcribe"].Outputs) != 2 {
t.Fatalf("rebuilt aggregate = %#v", completed.Stages["transcribe"])
}
}
func TestExecuteStagesRepairsSucceededTranscribeAggregateWithoutRequests(t *testing.T) {
for _, test := range []struct {
name string
mutate func(*manifest.Manifest)
removed bool
}{
{name: "removed speaker", removed: true, mutate: func(m *manifest.Manifest) {
m.Inputs = slices.DeleteFunc(m.Inputs, func(in manifest.InputRecord) bool { return filepath.Base(in.Path) == "bob.flac" })
}},
{name: "missing output", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs = m.Stages["transcribe"].Outputs[:1] }},
{name: "duplicate output", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs[1] = m.Stages["transcribe"].Outputs[0] }},
{name: "wrong path", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs[0].LocalPath = "transcripts/raw/other.json" }},
{name: "wrong checksum", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs[0].Checksum = "obsolete" }},
{name: "wrong producer", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs[0].ProducerRunID = "other-run" }},
} {
t.Run(test.name, func(t *testing.T) {
cfg, store := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"})
transcribe := mustSelectStage(t, "transcribe")
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: store, WhisperX: &whisperx.FakeClient{}},
}); err != nil {
t.Fatal(err)
}
m := loadTranscribeRecoveryManifest(t, store, cfg)
test.mutate(m)
m.MarkStageSucceeded("merge", time.Now().UTC(), nil)
if err := store.Save(context.Background(), manifestPathFor(cfg), m); err != nil {
t.Fatal(err)
}
retry := &whisperx.FakeClient{}
summary, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: store, WhisperX: retry},
})
if err != nil {
t.Fatal(err)
}
if len(summary.Skipped) != 0 || len(retry.RequestsSnapshot()) != 0 {
t.Fatalf("want aggregate repair without HTTP; summary=%#v requests=%#v", summary, retry.RequestsSnapshot())
}
completed := loadTranscribeRecoveryManifest(t, store, cfg)
want := []string{"alice", "bob"}
if test.removed {
want = []string{"alice"}
}
record := completed.Stages["transcribe"]
if !slices.Equal(transcribeSpeakerKeys(completed), want) || len(record.Outputs) != len(want) {
t.Fatalf("unexpected repaired state: %#v", record)
}
for _, output := range record.Outputs {
speaker := record.TranscribeSpeakers[strings.TrimSuffix(filepath.Base(output.LocalPath), ".json")]
if output.Checksum != speaker.Output.Checksum || output.ProducerRunID != speaker.ProducerRunID {
t.Fatalf("aggregate evidence does not match checkpoint: %#v", output)
}
}
if completed.Stages["merge"].Status != manifest.StatusStale {
t.Fatal("aggregate repair did not stale merge")
}
if test.removed {
paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
for _, name := range []string{"speakers.yml", "autocorrect.yml"} {
if err := os.WriteFile(filepath.Join(paths.InputsDir, name), []byte("{}"), 0o644); err != nil {
t.Fatal(err)
}
}
merge := &seriatim.FakeRunner{}
if _, err := executeStages(context.Background(), cfg, []stage.Stage{mustSelectStage(t, "merge")}, RunOptions{
Env: &Env{ManifestStore: store, Seriatim: merge},
}); err != nil {
t.Fatal(err)
}
if len(merge.NormalizeRequests) != 1 || filepath.Base(merge.NormalizeRequests[0].InputTranscriptPath) != "alice.json" {
t.Fatalf("removed speaker reached merge: %#v", merge.NormalizeRequests)
}
if _, err := os.Stat(filepath.Join(paths.TranscriptsRawDir, "bob.json")); err != nil {
t.Fatalf("incidental transcript was removed: %v", err)
}
}
})
}
}
func TestExecuteStagesResumesInterruptedRunningTranscribeState(t *testing.T) {
cfg, baseStore := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"})
concurrency := 1
cfg.Pipeline.WhisperX.Concurrency = &concurrency
stageErr := errors.New("speaker b failed")
terminalErr := errors.New("terminal save interrupted")
store := &interruptAfterTranscribeCheckpointStore{Store: baseStore, Err: terminalErr}
fake := &whisperx.FakeClient{TranscribeFn: func(
_ context.Context,
req whisperx.TranscribeRequest,
) (whisperx.TranscribeResult, error) {
if req.SpeakerID == "bob" {
return whisperx.TranscribeResult{}, stageErr
}
return writeTranscribeRecoveryResult(req)
}}
transcribe := mustSelectStage(t, "transcribe")
_, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: store, WhisperX: fake},
})
if !errors.Is(err, stageErr) || !errors.Is(err, terminalErr) {
t.Fatalf("executeStages() error = %v, want stage and terminal failures", err)
}
interrupted := loadTranscribeRecoveryManifest(t, baseStore, cfg)
if interrupted.Stages["transcribe"].Status != manifest.StatusRunning {
t.Fatalf("transcribe status = %q, want durable running state", interrupted.Stages["transcribe"].Status)
}
if got := transcribeSpeakerKeys(interrupted); !slices.Equal(got, []string{"alice"}) {
t.Fatalf("checkpointed speakers = %#v, want alice", got)
}
retry := &whisperx.FakeClient{}
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: baseStore, WhisperX: retry},
}); err != nil {
t.Fatalf("retry executeStages() error = %v", err)
}
if got := requestedTranscribeSpeakers(retry); !slices.Equal(got, []string{"bob"}) {
t.Fatalf("retry speakers = %#v, want bob", got)
}
}
func TestExecuteStagesPersistsForcedTranscribeResetAcrossRetry(t *testing.T) {
cfg, store := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"})
concurrency := 1
cfg.Pipeline.WhisperX.Concurrency = &concurrency
transcribe := mustSelectStage(t, "transcribe")
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: store, WhisperX: &whisperx.FakeClient{}},
}); err != nil {
t.Fatalf("baseline executeStages() error = %v", err)
}
cause := errors.New("forced bob failed")
forced := &whisperx.FakeClient{TranscribeFn: func(
_ context.Context,
req whisperx.TranscribeRequest,
) (whisperx.TranscribeResult, error) {
if req.SpeakerID == "bob" {
return whisperx.TranscribeResult{}, cause
}
return writeTranscribeRecoveryResult(req)
}}
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Force: true,
Env: &Env{ManifestStore: store, WhisperX: forced},
}); !errors.Is(err, cause) {
t.Fatalf("forced executeStages() error = %v, want %v", err, cause)
}
failed := loadTranscribeRecoveryManifest(t, store, cfg)
if got := transcribeSpeakerKeys(failed); !slices.Equal(got, []string{"alice"}) {
t.Fatalf("speakers after forced failure = %#v, want only replacement alice", got)
}
retry := &whisperx.FakeClient{}
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: store, WhisperX: retry},
}); err != nil {
t.Fatalf("retry executeStages() error = %v", err)
}
if got := requestedTranscribeSpeakers(retry); !slices.Equal(got, []string{"bob"}) {
t.Fatalf("retry speakers = %#v, want bob", got)
}
}
func TestExecuteStagesUpgradesLegacyTranscribeSuccessOnce(t *testing.T) {
cfg, store := setupTranscribeRecoveryFixture(t, []string{"alice"})
m := loadTranscribeRecoveryManifest(t, store, cfg)
m.MarkStageSucceeded("transcribe", time.Now().UTC(), []manifest.ArtifactRecord{{
Kind: "transcript_raw", LocalPath: "transcripts/raw/alice.json",
}})
if err := store.Save(context.Background(), manifestPathFor(cfg), m); err != nil {
t.Fatalf("save legacy transcribe success: %v", err)
}
transcribe := mustSelectStage(t, "transcribe")
upgrade := &whisperx.FakeClient{}
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: store, WhisperX: upgrade},
}); err != nil {
t.Fatalf("upgrade executeStages() error = %v", err)
}
if got := requestedTranscribeSpeakers(upgrade); !slices.Equal(got, []string{"alice"}) {
t.Fatalf("upgrade speakers = %#v, want alice", got)
}
upgraded := loadTranscribeRecoveryManifest(t, store, cfg)
if upgraded.Stages["transcribe"].TranscribeStateVersion != manifest.TranscribeStateContractVersion {
t.Fatalf("transcribe state version = %d, want current", upgraded.Stages["transcribe"].TranscribeStateVersion)
}
reused := &whisperx.FakeClient{}
summary, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: store, WhisperX: reused},
})
if err != nil {
t.Fatalf("reuse executeStages() error = %v", err)
}
if len(summary.Skipped) != 1 || len(reused.RequestsSnapshot()) != 0 {
t.Fatalf("reuse summary = %#v, requests = %#v, want skipped without HTTP", summary, reused.RequestsSnapshot())
}
}
type transcribeAggregateSaveFailingStore struct {
manifest.Store
Err error
Failed bool
}
func (store *transcribeAggregateSaveFailingStore) Save(
ctx context.Context,
path string,
m *manifest.Manifest,
) error {
record := m.Stages["transcribe"]
if !store.Failed && record != nil && record.Status == manifest.StatusSucceeded &&
record.TranscribeStateVersion == manifest.TranscribeStateContractVersion {
store.Failed = true
return store.Err
}
return store.Store.Save(ctx, path, m)
}
type interruptAfterTranscribeCheckpointStore struct {
manifest.Store
Err error
Interrupted bool
}
func (store *interruptAfterTranscribeCheckpointStore) Save(
ctx context.Context,
path string,
m *manifest.Manifest,
) error {
if store.Interrupted {
return store.Err
}
record := m.Stages["transcribe"]
err := store.Store.Save(ctx, path, m)
if err == nil && record != nil && record.Status == manifest.StatusRunning && len(record.TranscribeSpeakers) == 1 {
store.Interrupted = true
}
return err
}
func setupTranscribeRecoveryFixture(t *testing.T, speakers []string) (*config.Config, *manifest.LocalStore) {
t.Helper()
cfg := testConfig(t)
store := &manifest.LocalStore{}
artifactStore := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root)
paths, err := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID)
if err != nil {
t.Fatalf("prepare session layout: %v", err)
}
m := manifest.New(cfg.Session.SessionID, time.Now().UTC())
m.Campaign = cfg.Session.Campaign
m.Inputs = make([]manifest.InputRecord, 0, len(speakers))
for _, speaker := range speakers {
relative := filepath.ToSlash(filepath.Join("audio", speaker+".flac"))
if err := os.WriteFile(filepath.Join(paths.AudioDir, speaker+".flac"), []byte("audio-"+speaker), 0o644); err != nil {
t.Fatalf("write audio fixture: %v", err)
}
m.Inputs = append(m.Inputs, manifest.InputRecord{Kind: "audio", Path: relative})
}
m.MarkStageSucceeded("prepare", time.Now().UTC(), nil)
m.MarkStageSucceeded("merge", time.Now().UTC(), nil)
if err := store.Save(context.Background(), paths.ManifestPath, m); err != nil {
t.Fatalf("save recovery fixture: %v", err)
}
return cfg, store
}
func writeTranscribeRecoveryResult(req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) {
if err := os.MkdirAll(filepath.Dir(req.OutputRawTranscriptPath), 0o755); err != nil {
return whisperx.TranscribeResult{}, err
}
if err := os.WriteFile(req.OutputRawTranscriptPath, []byte(`{"segments":[]}`), 0o644); err != nil {
return whisperx.TranscribeResult{}, err
}
return whisperx.TranscribeResult{OutputRawTranscriptPath: req.OutputRawTranscriptPath}, nil
}
func loadTranscribeRecoveryManifest(t *testing.T, store manifest.Store, cfg *config.Config) *manifest.Manifest {
t.Helper()
m, err := store.Load(context.Background(), manifestPathFor(cfg))
if err != nil {
t.Fatalf("load transcribe recovery manifest: %v", err)
}
return m
}
func transcribeSpeakerKeys(m *manifest.Manifest) []string {
keys := make([]string, 0, len(m.Stages["transcribe"].TranscribeSpeakers))
for key := range m.Stages["transcribe"].TranscribeSpeakers {
keys = append(keys, key)
}
slices.Sort(keys)
return keys
}
func requestedTranscribeSpeakers(fake *whisperx.FakeClient) []string {
requests := fake.RequestsSnapshot()
speakers := make([]string, 0, len(requests))
for _, request := range requests {
speakers = append(speakers, request.SpeakerID)
}
return speakers
}