fix: the remaining Unbounded workflow histories sub-items

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,

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

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

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

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
 		}

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