From 909ef4dd359775ff7cf8a84989fe39ff7b0f9869 Mon Sep 17 00:00:00 2001 From: Matjaz Debelak Date: Thu, 13 Aug 2026 08:55:36 +0200 Subject: [PATCH 1/5] fix(misc): continue MoveFilesWorkerFlow as new before its history grows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The finding calls this an eternal fixed-ID signal workflow whose history is never reset. It is not eternal: ReceiveWithTimeout gives up after ten seconds and the run completes, to be started again by the next SignalWithStartWorkflow. The exposure is narrower and real. Requests arriving less than ten seconds apart keep one run alive indefinitely, and every move adds a GetShapes call plus one MoveFileWait per shape to the same history — so a bulk move never lets the run end and never resets the history. It now continues as new when the server suggests it, which is the signal that takes the actual event count and size into account rather than a threshold guessed here. The check runs only while the signal channel is empty, so no queued request is dropped; a request that arrives after the check but before the server applies the continue-as-new is handled by the server, which turns an unhandled signal at completion into a new workflow task. Co-Authored-By: Claude Opus 5 (1M context) --- workflows/misc/slow_move_files.go | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/workflows/misc/slow_move_files.go b/workflows/misc/slow_move_files.go index ab8de35e..e1d789b1 100644 --- a/workflows/misc/slow_move_files.go +++ b/workflows/misc/slow_move_files.go @@ -70,6 +70,11 @@ func StartFilesWorkerFlow(ctx context.Context, params MoveMBFileParams) error { const MoveMBFileSignalName = "move_mb_file" +// moveFilesIdleTimeout is how long the worker flow waits for another request +// before completing. The next request starts it again through +// SignalWithStartWorkflow. +const moveFilesIdleTimeout = 10 * time.Second + type MBStorage struct { VXID string BasePath string @@ -147,13 +152,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 idleTimeout. +// +// A steady stream of requests keeps one run alive indefinitely, and every move +// adds activity events to the same history, so the run continues as new once +// the server says the history is getting long. Continuing only while the +// channel is empty means no queued request is dropped on the way over. func MoveFilesWorkerFlow(ctx workflow.Context) error { ch := workflow.GetSignalChannel(ctx, MoveMBFileSignalName) msg := &MoveMBFileParams{} for { - ok, _ := ch.ReceiveWithTimeout(ctx, 10*time.Second, msg) + 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 } From 049131a3a29527c28af24abcbea1516fa570f2f2 Mon Sep 17 00:00:00 2001 From: Matjaz Debelak Date: Fri, 14 Aug 2026 11:36:36 +0200 Subject: [PATCH 2/5] fix(export): stop persisting the merge inputs as a SideEffect marker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MergeExportData computed its merge inputs inside a workflow.SideEffect, which writes the returned value into the history as a marker. For a long export that value is every clip path, every language and every offset — recorded once as the marker and again in the activity inputs derived from it. exportDataToMergeInputs takes an ExportData and two paths and returns a value; nothing in it reads a clock, a file or a global, so the SideEffect bought nothing. It does range over the per-clip audio and subtitle maps, which is why it now sorts those keys: each language accumulates into its own MergeInput so the order was already immaterial, but running in workflow code it has to be visibly so — to workflowcheck and to the next reader. Unguarded: this changes what the workflow writes to its history, so an export started before the deploy fails to replay it. None has to survive the deploy. Four tests, including one that calls the function fifty times and compares, so a future edit that lets map order leak into the result is caught rather than being caught by a customer. Co-Authored-By: Claude Opus 5 (1M context) --- workflows/export/merge_export_data.go | 17 ++-- workflows/export/merge_export_data_test.go | 96 ++++++++++++++++++++++ 2 files changed, 104 insertions(+), 9 deletions(-) create mode 100644 workflows/export/merge_export_data_test.go diff --git a/workflows/export/merge_export_data.go b/workflows/export/merge_export_data.go index 9910dc73..27cede6a 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,11 @@ func exportDataToMergeInputs(data *vidispine.ExportData, tempDir, subtitlesDir p }) } - for lan, af := range clip.AudioFiles { + // Sorted rather than ranged: this runs in workflow code now, and although + // each language accumulates into its own MergeInput, the checker cannot + // see that and neither can the next reader. + 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 +222,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..06b012b2 --- /dev/null +++ b/workflows/export/merge_export_data_test.go @@ -0,0 +1,96 @@ +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", + }, + }, + }, + } +} + +// The result is used directly in workflow code now rather than being frozen by +// a SideEffect marker, so it has to be the same on every call — including +// 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; the map iteration must not + // be able to reorder it. + 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) +} From 36d239c4017b8ad6221f3cf54293c4e9f5278d40 Mon Sep 17 00:00:00 2001 From: Matjaz Debelak Date: Thu, 13 Aug 2026 09:00:06 +0200 Subject: [PATCH 3/5] fix(scheduled): return cleanup counts instead of every deleted path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CleanupTemp sweeps around sixty folders and returned the full list of deleted paths as its result. A workflow result is written into the completion event, so a fortnight of temp files landed in the history and in every caller that fetches the result — to say something a number says. DeletedFiles is replaced by DeletedCountPerRoot, which is more useful for the thing the result is actually read for: seeing that a folder is not being cleaned. The paths are still in the activity results and the worker logs. The per-folder log line moves below the error check, where it can report the folder it just finished rather than the running total — as written it logged 0 for the first folder no matter how much it had deleted. Co-Authored-By: Claude Opus 5 (1M context) --- workflows/scheduled/files_cleanup.go | 29 +++++++++++++++++++-------- workflows/scheduled/scheduled_test.go | 8 ++++++++ 2 files changed, 29 insertions(+), 8 deletions(-) diff --git a/workflows/scheduled/files_cleanup.go b/workflows/scheduled/files_cleanup.go index e5287b7c..23bd763f 100644 --- a/workflows/scheduled/files_cleanup.go +++ b/workflows/scheduled/files_cleanup.go @@ -9,9 +9,17 @@ import ( "go.temporal.io/sdk/workflow" ) +// CleanupResult reports what a cleanup run removed. +// +// Counts rather than paths: the workflow sweeps around sixty folders and the +// result is written into the workflow's completion event, so returning every +// deleted path put a fortnight of temp files into the history — and into every +// caller that fetches the result. The paths themselves are in the activity +// results and the worker logs, which is where anyone chasing a specific file +// looks anyway. type CleanupResult struct { - DeletedFiles []string - DeletedCount int + DeletedCount int + DeletedCountPerRoot map[string]int } func CleanupTemp(ctx workflow.Context) (*CleanupResult, error) { @@ -100,7 +108,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 +124,18 @@ 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...) + // Counted after the error check: the old log line ran before it and + // reported the running total rather than this folder's, so it printed 0 + // for the first folder however much it had deleted. + 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 +148,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..94334433 100644 --- a/workflows/scheduled/scheduled_test.go +++ b/workflows/scheduled/scheduled_test.go @@ -82,6 +82,14 @@ 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: the result goes + // into the workflow's completion event, and this sweeps around sixty folders. + 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) { From ea0e06d8d31156c2b9a66f3ada30eb0c962692e3 Mon Sep 17 00:00:00 2001 From: Matjaz Debelak Date: Thu, 13 Aug 2026 09:02:14 +0200 Subject: [PATCH 4/5] feat(ingest): let ImportSubtitles take a path instead of the transcription MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The transcription arrives as a workflow argument, and a workflow argument is stored in the WorkflowExecutionStarted event. A word-level transcription of a long programme runs to megabytes, so every import writes that into the history — and a big enough one does not merely bloat it, it exceeds Temporal's payload limit and the workflow cannot be started at all. SubtitlesFile is an alternative input: point it at the same JSON on shared storage and an activity reads it, keeping the payload out of the history. When it is set, Subtitles is ignored. Subtitles is kept rather than replaced. Nothing in this repository starts ImportSubtitles — it is registered and triggered from outside — so the field cannot be removed until whatever produces the transcription has moved over. The comment on it says why to prefer the path. Co-Authored-By: Claude Opus 5 (1M context) --- workflows/ingest/import_subtitles.go | 40 +++++++++++-- workflows/ingest/import_subtitles_test.go | 70 +++++++++++++++++++++++ 2 files changed, 106 insertions(+), 4 deletions(-) diff --git a/workflows/ingest/import_subtitles.go b/workflows/ingest/import_subtitles.go index 70d4796c..cccb4ea5 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,20 @@ type Word struct { } type ImportSubtitlesInput struct { - VXID string `json:"vxid"` + VXID string `json:"vxid"` + + // Subtitles carries the transcription inline. A workflow argument is stored + // in the WorkflowExecutionStarted event, and a word-level transcription of a + // long programme is megabytes, so this both bloats the history and can push + // the start over 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, keeping the payload out of + // the history entirely. + 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 +107,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 +120,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 +194,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 returns the transcription to import, reading it from shared +// storage when the caller passed a path rather than 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..d720b059 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,75 @@ func (s *ImportSubtitlesTestSuite) Test_ImportSubtitlesWorkflow() { s.NoError(err) } +// A word-level transcription of a long programme is megabytes, and a workflow +// argument lives in the WorkflowExecutionStarted event. Passing a path instead +// keeps it out of the history; 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)) } From c7516d24ddb029dca0972f2d79ba223cfbb0559d Mon Sep 17 00:00:00 2001 From: Matjaz Debelak Date: Fri, 14 Aug 2026 11:35:57 +0200 Subject: [PATCH 5/5] docs: trim the comments in the payload changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Roughly a quarter of the lines added by this branch were comment, and most of the excess was the commit message written into the source a second time — why the change was made rather than what a reader of the code needs. What stays is the part that is not recoverable from the code: why the channel has to be empty before continuing as new, why the merge inputs are sorted, and why the cleanup result carries counts. What goes is the retelling around it. One comment went entirely rather than shrinking: the note explaining that the per-folder log line had moved below the error check described the change, not the code, and the line reads correctly on its own. Co-Authored-By: Claude Opus 5 (1M context) --- workflows/export/merge_export_data.go | 4 +--- workflows/export/merge_export_data_test.go | 8 +++----- workflows/ingest/import_subtitles.go | 16 +++++++--------- workflows/ingest/import_subtitles_test.go | 5 ++--- workflows/misc/slow_move_files.go | 17 ++++++++--------- workflows/scheduled/files_cleanup.go | 12 +++--------- workflows/scheduled/scheduled_test.go | 3 +-- 7 files changed, 25 insertions(+), 40 deletions(-) diff --git a/workflows/export/merge_export_data.go b/workflows/export/merge_export_data.go index 27cede6a..8fd6422d 100644 --- a/workflows/export/merge_export_data.go +++ b/workflows/export/merge_export_data.go @@ -200,9 +200,7 @@ func exportDataToMergeInputs(data *vidispine.ExportData, tempDir, subtitlesDir p }) } - // Sorted rather than ranged: this runs in workflow code now, and although - // each language accumulates into its own MergeInput, the checker cannot - // see that and neither can the next reader. + // 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 { diff --git a/workflows/export/merge_export_data_test.go b/workflows/export/merge_export_data_test.go index 06b012b2..17c051ff 100644 --- a/workflows/export/merge_export_data_test.go +++ b/workflows/export/merge_export_data_test.go @@ -46,9 +46,8 @@ func twoClipExportData() *vidispine.ExportData { } } -// The result is used directly in workflow code now rather than being frozen by -// a SideEffect marker, so it has to be the same on every call — including -// across the randomized map iteration inside. +// 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") @@ -69,8 +68,7 @@ func TestExportDataToMergeInputsBuildsOneInputPerLanguage(t *testing.T) { require.Len(t, got.AudioMergeInputs, 3) require.Len(t, got.SubtitleMergeInputs, 2) - // Clip order drives item order within a language; the map iteration must not - // be able to reorder it. + // 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) diff --git a/workflows/ingest/import_subtitles.go b/workflows/ingest/import_subtitles.go index cccb4ea5..d05635ec 100644 --- a/workflows/ingest/import_subtitles.go +++ b/workflows/ingest/import_subtitles.go @@ -41,15 +41,13 @@ type Word struct { type ImportSubtitlesInput struct { VXID string `json:"vxid"` - // Subtitles carries the transcription inline. A workflow argument is stored - // in the WorkflowExecutionStarted event, and a word-level transcription of a - // long programme is megabytes, so this both bloats the history and can push - // the start over Temporal's payload limit. Prefer SubtitlesFile. + // 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"` - // SubtitlesFile points at the same JSON on shared storage. When set it is - // read by an activity and Subtitles is ignored, keeping the payload out of - // the history entirely. + // 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"` @@ -195,8 +193,8 @@ func ImportSubtitles(ctx workflow.Context, input ImportSubtitlesInput) error { return nil } -// resolveSubtitles returns the transcription to import, reading it from shared -// storage when the caller passed a path rather than the whole thing. +// 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 diff --git a/workflows/ingest/import_subtitles_test.go b/workflows/ingest/import_subtitles_test.go index d720b059..16086de5 100644 --- a/workflows/ingest/import_subtitles_test.go +++ b/workflows/ingest/import_subtitles_test.go @@ -149,9 +149,8 @@ func (s *ImportSubtitlesTestSuite) Test_ImportSubtitlesWorkflow() { s.NoError(err) } -// A word-level transcription of a long programme is megabytes, and a workflow -// argument lives in the WorkflowExecutionStarted event. Passing a path instead -// keeps it out of the history; the workflow reads it and behaves identically. +// Given a path instead of the transcription, the workflow reads it and behaves +// identically. func (s *ImportSubtitlesTestSuite) Test_ImportSubtitlesFromFile() { vxid := "VX-123" segments := []Segment{ diff --git a/workflows/misc/slow_move_files.go b/workflows/misc/slow_move_files.go index e1d789b1..31373e6b 100644 --- a/workflows/misc/slow_move_files.go +++ b/workflows/misc/slow_move_files.go @@ -70,9 +70,8 @@ func StartFilesWorkerFlow(ctx context.Context, params MoveMBFileParams) error { const MoveMBFileSignalName = "move_mb_file" -// moveFilesIdleTimeout is how long the worker flow waits for another request -// before completing. The next request starts it again through -// SignalWithStartWorkflow. +// 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 { @@ -152,19 +151,19 @@ 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 idleTimeout. +// 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 keeps one run alive indefinitely, and every move -// adds activity events to the same history, so the run continues as new once -// the server says the history is getting long. Continuing only while the -// channel is empty means no queued request is dropped on the way over. +// 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 { + // 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()) diff --git a/workflows/scheduled/files_cleanup.go b/workflows/scheduled/files_cleanup.go index 23bd763f..5c237e8c 100644 --- a/workflows/scheduled/files_cleanup.go +++ b/workflows/scheduled/files_cleanup.go @@ -11,12 +11,9 @@ import ( // CleanupResult reports what a cleanup run removed. // -// Counts rather than paths: the workflow sweeps around sixty folders and the -// result is written into the workflow's completion event, so returning every -// deleted path put a fortnight of temp files into the history — and into every -// caller that fetches the result. The paths themselves are in the activity -// results and the worker logs, which is where anyone chasing a specific file -// looks anyway. +// 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 { DeletedCount int DeletedCountPerRoot map[string]int @@ -129,9 +126,6 @@ func CleanupTemp(ctx workflow.Context) (*CleanupResult, error) { return nil, err } - // Counted after the error check: the old log line ran before it and - // reported the running total rather than this folder's, so it printed 0 - // for the first folder however much it had deleted. logger.Info("Deleted files", "root", folder, "count", len(deletedFilesLoop)) deletedPerRoot[folder] = len(deletedFilesLoop) diff --git a/workflows/scheduled/scheduled_test.go b/workflows/scheduled/scheduled_test.go index 94334433..e0785861 100644 --- a/workflows/scheduled/scheduled_test.go +++ b/workflows/scheduled/scheduled_test.go @@ -83,8 +83,7 @@ func (s *ScheduledTestSuite) Test_CleanupTemp() { s.env.GetWorkflowResult(&result) s.Greater(result.DeletedCount, 0) - // Two files per folder, counted per root rather than listed: the result goes - // into the workflow's completion event, and this sweeps around sixty folders. + // 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 {