diff --git a/workflows/export/merge_export_data.go b/workflows/export/merge_export_data.go index 9910dc73..8fd6422d 100644 --- a/workflows/export/merge_export_data.go +++ b/workflows/export/merge_export_data.go @@ -36,13 +36,7 @@ func MergeExportData(ctx workflow.Context, params MergeExportDataParams) (*Merge logger.Info("Starting MergeExportData") data := params.ExportData - var dataMergeInputs MergeInput - err := workflow.SideEffect(ctx, func(ctx workflow.Context) interface{} { - return exportDataToMergeInputs(data, params.TempDir, params.SubtitlesDir) - }).Get(&dataMergeInputs) - if err != nil { - return nil, err - } + dataMergeInputs := exportDataToMergeInputs(data, params.TempDir, params.SubtitlesDir) mergeInput := dataMergeInputs.MergeInput audioMergeInputs := dataMergeInputs.AudioMergeInputs @@ -206,7 +200,9 @@ func exportDataToMergeInputs(data *vidispine.ExportData, tempDir, subtitlesDir p }) } - for lan, af := range clip.AudioFiles { + // Sorted: this runs in workflow code, where map order must not leak. + for _, lan := range wfutils.SortedKeys(clip.AudioFiles) { + af := clip.AudioFiles[lan] if _, ok := audioMergeInputs[lan]; !ok { audioMergeInputs[lan] = &common.MergeInput{ Title: data.SafeTitle + "-" + lan, @@ -224,7 +220,8 @@ func exportDataToMergeInputs(data *vidispine.ExportData, tempDir, subtitlesDir p }) } - for lan, sf := range clip.SubtitleFiles { + for _, lan := range wfutils.SortedKeys(clip.SubtitleFiles) { + sf := clip.SubtitleFiles[lan] if _, ok := subtitleMergeInputs[lan]; !ok { subtitleMergeInputs[lan] = &common.MergeInput{ Title: data.SafeTitle + "-" + lan, diff --git a/workflows/export/merge_export_data_test.go b/workflows/export/merge_export_data_test.go new file mode 100644 index 00000000..17c051ff --- /dev/null +++ b/workflows/export/merge_export_data_test.go @@ -0,0 +1,94 @@ +package export + +import ( + "testing" + + "github.com/bcc-code/bcc-media-flows/paths" + "github.com/bcc-code/bcc-media-flows/services/vidispine" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func twoClipExportData() *vidispine.ExportData { + return &vidispine.ExportData{ + SafeTitle: "Some_Title", + Clips: []*vidispine.Clip{ + { + VideoFile: "/mnt/isilon/a.mxf", + InSeconds: 0, + OutSeconds: 10, + AudioFiles: map[string]*vidispine.AudioFile{ + "nor": {File: "/mnt/isilon/a-nor.wav"}, + "eng": {File: "/mnt/isilon/a-eng.wav"}, + "deu": {File: "/mnt/isilon/a-deu.wav"}, + }, + SubtitleFiles: map[string]string{ + "nor": "/mnt/isilon/a-nor.srt", + "eng": "/mnt/isilon/a-eng.srt", + }, + JSONTranscriptFile: "/mnt/isilon/a.json", + }, + { + VideoFile: "/mnt/isilon/b.mxf", + InSeconds: 5, + OutSeconds: 20, + AudioFiles: map[string]*vidispine.AudioFile{ + "nor": {File: "/mnt/isilon/b-nor.wav"}, + "eng": {File: "/mnt/isilon/b-eng.wav"}, + "deu": {File: "/mnt/isilon/b-deu.wav"}, + }, + SubtitleFiles: map[string]string{ + "nor": "/mnt/isilon/b-nor.srt", + "eng": "/mnt/isilon/b-eng.srt", + }, + }, + }, + } +} + +// Workflow code calls this directly, so it has to give the same answer every +// time, across the randomized map iteration inside. +func TestExportDataToMergeInputsIsStableAcrossCalls(t *testing.T) { + tempDir := paths.MustParse("/mnt/isilon/temp") + subsDir := paths.MustParse("/mnt/isilon/subs") + + want := exportDataToMergeInputs(twoClipExportData(), tempDir, subsDir) + + for i := 0; i < 50; i++ { + assert.Equal(t, want, exportDataToMergeInputs(twoClipExportData(), tempDir, subsDir)) + } +} + +func TestExportDataToMergeInputsBuildsOneInputPerLanguage(t *testing.T) { + tempDir := paths.MustParse("/mnt/isilon/temp") + subsDir := paths.MustParse("/mnt/isilon/subs") + + got := exportDataToMergeInputs(twoClipExportData(), tempDir, subsDir) + + require.Len(t, got.AudioMergeInputs, 3) + require.Len(t, got.SubtitleMergeInputs, 2) + + // Clip order drives item order within a language. + nor := got.AudioMergeInputs["nor"] + require.Len(t, nor.Items, 2) + assert.Equal(t, paths.MustParse("/mnt/isilon/a-nor.wav"), nor.Items[0].Path) + assert.Equal(t, paths.MustParse("/mnt/isilon/b-nor.wav"), nor.Items[1].Path) + assert.Equal(t, "Some_Title-nor", nor.Title) + + assert.Equal(t, subsDir, got.SubtitleMergeInputs["eng"].OutputDir) + assert.Equal(t, tempDir, got.SubtitleMergeInputs["eng"].WorkDir) + + // 10 + 15 seconds of video, and only the first clip has a transcript. + assert.Equal(t, float64(25), got.MergeInput.Duration) + require.NotNil(t, got.JSONTranscriptInput) + assert.Equal(t, float64(10), got.JSONTranscriptInput.Duration) +} + +func TestExportDataToMergeInputsHasNoTranscriptWhenNoClipHasOne(t *testing.T) { + data := twoClipExportData() + data.Clips[0].JSONTranscriptFile = "" + + got := exportDataToMergeInputs(data, paths.MustParse("/mnt/isilon/temp"), paths.MustParse("/mnt/isilon/subs")) + + assert.Nil(t, got.JSONTranscriptInput) +} diff --git a/workflows/ingest/import_subtitles.go b/workflows/ingest/import_subtitles.go index 70d4796c..d05635ec 100644 --- a/workflows/ingest/import_subtitles.go +++ b/workflows/ingest/import_subtitles.go @@ -6,6 +6,7 @@ import ( "strings" vsactivity "github.com/bcc-code/bcc-media-flows/activities/vidispine" + "github.com/bcc-code/bcc-media-flows/paths" wfutils "github.com/bcc-code/bcc-media-flows/utils/workflows" "go.temporal.io/sdk/workflow" ) @@ -38,9 +39,18 @@ type Word struct { } type ImportSubtitlesInput struct { - VXID string `json:"vxid"` + VXID string `json:"vxid"` + + // Subtitles carries the transcription inline, which puts it in the + // WorkflowExecutionStarted event. A word-level transcription of a long + // programme can exceed Temporal's payload limit. Prefer SubtitlesFile. Subtitles Transcription `json:"subtitles"` - Language string `json:"language"` + + // SubtitlesFile points at the same JSON on shared storage. When set, it is + // read by an activity and Subtitles is ignored. + SubtitlesFile *paths.Path `json:"subtitlesFile,omitempty"` + + Language string `json:"language"` } // convertSecondsToSRTTimestamp converts a float64 number of seconds to SRT timestamp format: HH:MM:SS,mmm @@ -95,6 +105,11 @@ func ImportSubtitles(ctx workflow.Context, input ImportSubtitlesInput) error { return fmt.Errorf("missing language") } + subtitles, err := resolveSubtitles(ctx, input) + if err != nil { + return err + } + outputPath, err := wfutils.GetWorkflowAuxOutputFolder(ctx) if err != nil { return fmt.Errorf("failed to get aux output folder: %w", err) @@ -103,14 +118,14 @@ func ImportSubtitles(ctx workflow.Context, input ImportSubtitlesInput) error { srtFilePath := outputPath.Append(input.VXID + "_subtitles.srt") jsonFilePath := outputPath.Append(input.VXID + "_subtitles.json") - srtData := ToSRT(input.Subtitles.Segments, false) + srtData := ToSRT(subtitles.Segments, false) err = wfutils.WriteFile(ctx, srtFilePath, []byte(srtData)) if err != nil { return fmt.Errorf("failed to write SRT file: %w", err) } - jsonData, err := json.MarshalIndent(input.Subtitles, "", " ") + jsonData, err := json.MarshalIndent(subtitles, "", " ") if err != nil { return fmt.Errorf("failed to marshal subtitles to JSON: %w", err) } @@ -177,3 +192,18 @@ func ImportSubtitles(ctx workflow.Context, input ImportSubtitlesInput) error { logger.Info("Subtitle SRT and JSON imported as shapes; SRT as sidecar (async)", "vxid", input.VXID) return nil } + +// resolveSubtitles reads the transcription from storage when the caller passed +// a path instead of the whole thing. +func resolveSubtitles(ctx workflow.Context, input ImportSubtitlesInput) (Transcription, error) { + if input.SubtitlesFile == nil { + return input.Subtitles, nil + } + + subtitles, err := wfutils.UnmarshalJSONFile[Transcription](ctx, *input.SubtitlesFile) + if err != nil { + return Transcription{}, fmt.Errorf("failed to read subtitles from %s: %w", input.SubtitlesFile.Local(), err) + } + + return *subtitles, nil +} diff --git a/workflows/ingest/import_subtitles_test.go b/workflows/ingest/import_subtitles_test.go index 10921d11..16086de5 100644 --- a/workflows/ingest/import_subtitles_test.go +++ b/workflows/ingest/import_subtitles_test.go @@ -4,6 +4,7 @@ import ( "github.com/bcc-code/bcc-media-flows/activities" vsactivity "github.com/bcc-code/bcc-media-flows/activities/vidispine" "github.com/bcc-code/bcc-media-flows/paths" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" //vsactivity "github.com/bcc-code/bcc-media-flows/activities/vidispine" @@ -148,6 +149,74 @@ func (s *ImportSubtitlesTestSuite) Test_ImportSubtitlesWorkflow() { s.NoError(err) } +// Given a path instead of the transcription, the workflow reads it and behaves +// identically. +func (s *ImportSubtitlesTestSuite) Test_ImportSubtitlesFromFile() { + vxid := "VX-123" + segments := []Segment{ + {Start: 0.0, End: 1.0, Text: "Hello"}, + {Start: 1.5, End: 2.5, Text: "World"}, + } + stored := Transcription{Segments: segments} + subtitlesFile := paths.Path{Drive: paths.Drive{Value: "isilon"}, Path: "Production/aux/incoming.json"} + + input := ImportSubtitlesInput{ + VXID: vxid, + Language: "en", + SubtitlesFile: &subtitlesFile, + } + + storedJSON, err := json.Marshal(stored) + s.NoError(err) + + s.env.OnActivity(activities.Util.ReadFile, mock.Anything, activities.FileInput{Path: subtitlesFile}). + Return(storedJSON, nil).Once() + + s.env.OnActivity(activities.Vidispine.ImportFileAsShapeActivity, mock.Anything, mock.Anything).Return(&vsactivity.ImportFileResult{JobID: "job"}, nil) + s.env.OnActivity(activities.Util.CreateFolder, mock.Anything, mock.Anything).Return("", nil) + + datePath := time.Now().Format("2006/01/02") + testPath := "Production/aux/" + datePath + "/VX-123_subtitles" + + jsonString, err := json.MarshalIndent(stored, "", " ") + s.NoError(err) + + s.env.OnActivity(activities.Util.WriteFile, mock.Anything, activities.WriteFileInput{ + Path: paths.Path{Drive: paths.Drive{Value: "isilon"}, Path: testPath + ".json"}, + Data: []byte(jsonString), + }).Return("", nil).Once() + + s.env.OnActivity(activities.Util.WriteFile, mock.Anything, activities.WriteFileInput{ + Path: paths.Path{Drive: paths.Drive{Value: "isilon"}, Path: testPath + ".srt"}, + Data: []byte("1\n00:00:00,000 --> 00:00:01,000\nHello\n\n2\n00:00:01,500 --> 00:00:02,500\nWorld\n\n"), + }).Return("", nil).Once() + + s.env.OnActivity(activities.Vidispine.ImportFileAsSidecarActivity, mock.Anything, mock.Anything).Return(nil, nil) + s.env.OnActivity(activities.Vidispine.JobCompleteOrErr, mock.Anything, mock.Anything).Return(true, nil) + + s.env.ExecuteWorkflow(ImportSubtitles, input) + s.True(s.env.IsWorkflowCompleted()) + s.NoError(s.env.GetWorkflowError()) +} + +func (s *ImportSubtitlesTestSuite) Test_ImportSubtitlesFromMissingFile() { + subtitlesFile := paths.Path{Drive: paths.Drive{Value: "isilon"}, Path: "Production/aux/gone.json"} + + s.env.OnActivity(activities.Util.ReadFile, mock.Anything, activities.FileInput{Path: subtitlesFile}). + Return(nil, assert.AnError) + + s.env.ExecuteWorkflow(ImportSubtitles, ImportSubtitlesInput{ + VXID: "VX-123", + Language: "en", + SubtitlesFile: &subtitlesFile, + }) + s.True(s.env.IsWorkflowCompleted()) + + err := s.env.GetWorkflowError() + s.Error(err) + s.Contains(err.Error(), "failed to read subtitles") +} + func TestImportSubtitlesTestSuite(t *testing.T) { suite.Run(t, new(ImportSubtitlesTestSuite)) } diff --git a/workflows/misc/slow_move_files.go b/workflows/misc/slow_move_files.go index ab8de35e..31373e6b 100644 --- a/workflows/misc/slow_move_files.go +++ b/workflows/misc/slow_move_files.go @@ -70,6 +70,10 @@ func StartFilesWorkerFlow(ctx context.Context, params MoveMBFileParams) error { const MoveMBFileSignalName = "move_mb_file" +// moveFilesIdleTimeout is how long to wait for another request before +// completing. SignalWithStartWorkflow starts the flow again for the next one. +const moveFilesIdleTimeout = 10 * time.Second + type MBStorage struct { VXID string BasePath string @@ -147,13 +151,26 @@ func FindStorageForVXID(vxid string) *MBStorage { return nil } +// MoveFilesWorkerFlow drains move requests signalled to the fixed +// "move_mb_file" execution, and exits once none has arrived for +// moveFilesIdleTimeout. +// +// A steady stream of requests would otherwise keep one run, and one history, +// alive indefinitely, so it continues as new when the server suggests it. func MoveFilesWorkerFlow(ctx workflow.Context) error { ch := workflow.GetSignalChannel(ctx, MoveMBFileSignalName) msg := &MoveMBFileParams{} for { - ok, _ := ch.ReceiveWithTimeout(ctx, 10*time.Second, msg) + // Only while the channel is empty, so no queued request is dropped. + if workflow.GetInfo(ctx).GetContinueAsNewSuggested() && ch.Len() == 0 { + workflow.GetLogger(ctx).Info("History is long, continuing as new", + "events", workflow.GetInfo(ctx).GetCurrentHistoryLength()) + return workflow.NewContinueAsNewError(ctx, MoveFilesWorkerFlow) + } + + ok, _ := ch.ReceiveWithTimeout(ctx, moveFilesIdleTimeout, msg) if !ok { break } diff --git a/workflows/scheduled/files_cleanup.go b/workflows/scheduled/files_cleanup.go index e5287b7c..5c237e8c 100644 --- a/workflows/scheduled/files_cleanup.go +++ b/workflows/scheduled/files_cleanup.go @@ -9,9 +9,14 @@ import ( "go.temporal.io/sdk/workflow" ) +// CleanupResult reports what a cleanup run removed. +// +// Counts rather than paths: this is the workflow's completion event, and the +// paths of a fortnight of temp files across sixty folders do not belong in the +// history. They are in the activity results and the worker logs. type CleanupResult struct { - DeletedFiles []string - DeletedCount int + DeletedCount int + DeletedCountPerRoot map[string]int } func CleanupTemp(ctx workflow.Context) (*CleanupResult, error) { @@ -100,7 +105,8 @@ func CleanupTemp(ctx workflow.Context) (*CleanupResult, error) { "/mnt/isilon/Export": workflow.Now(ctx).Add(-14 * 24 * time.Hour), } - deletedFiles := []string{} + deletedTotal := 0 + deletedPerRoot := map[string]int{} folders, err := wfutils.GetMapKeysSafely(ctx, foldersToCleanup) if err != nil { @@ -115,14 +121,15 @@ func CleanupTemp(ctx workflow.Context) (*CleanupResult, error) { OlderThan: olderThan, }).Result(ctx) - logger.Info("Deleted files", "count", len(deletedFiles)) - if err != nil { logger.Error("Error during temp files cleanup", "error", err) return nil, err } - deletedFiles = append(deletedFiles, deletedFilesLoop...) + logger.Info("Deleted files", "root", folder, "count", len(deletedFilesLoop)) + + deletedPerRoot[folder] = len(deletedFilesLoop) + deletedTotal += len(deletedFilesLoop) err = wfutils.ExecuteWithLowPrioQueue(ctx, activities.Util.DeleteEmptyDirectories, activities.CleanupInput{ Root: paths.MustParse(folder), @@ -135,8 +142,8 @@ func CleanupTemp(ctx workflow.Context) (*CleanupResult, error) { } res := &CleanupResult{ - DeletedFiles: deletedFiles, - DeletedCount: len(deletedFiles), + DeletedCount: deletedTotal, + DeletedCountPerRoot: deletedPerRoot, } return res, nil diff --git a/workflows/scheduled/scheduled_test.go b/workflows/scheduled/scheduled_test.go index d995b330..e0785861 100644 --- a/workflows/scheduled/scheduled_test.go +++ b/workflows/scheduled/scheduled_test.go @@ -82,6 +82,13 @@ func (s *ScheduledTestSuite) Test_CleanupTemp() { var result CleanupResult s.env.GetWorkflowResult(&result) s.Greater(result.DeletedCount, 0) + + // Two files per folder, counted per root rather than listed. + s.NotEmpty(result.DeletedCountPerRoot) + s.Equal(2*len(result.DeletedCountPerRoot), result.DeletedCount) + for root, count := range result.DeletedCountPerRoot { + s.Equal(2, count, "unexpected count for %s", root) + } } func TestScheduledTestSuite(t *testing.T) {