diff --git a/pkg/media/ingest_worker.go b/pkg/media/ingest_worker.go index 694b8720c..6ca6f5e96 100644 --- a/pkg/media/ingest_worker.go +++ b/pkg/media/ingest_worker.go @@ -248,7 +248,7 @@ func RunMP4IngestWorker(ctx context.Context, cfg IngestWorkerConfig, stdin io.Re if cfg.Record { log.Log(ctx, "recording ingest media to file", "streamer", cfg.StreamerDID) var finalize func() - media, finalize = mm.recordTee(ctx, stdin, cfg.StreamerDID, ".rtmp.mp4") + media, finalize = mm.recordTee(ctx, stdin, cfg.StreamerDID) // Registered before the pipeline's SetState(Null) defer, so it runs after // the pipeline stops reading — and before this worker process exits, which // would otherwise strand an uncommitted S3 upload. diff --git a/pkg/media/ingest_worker_test.go b/pkg/media/ingest_worker_test.go index fe49c8a65..9af6fae17 100644 --- a/pkg/media/ingest_worker_test.go +++ b/pkg/media/ingest_worker_test.go @@ -230,7 +230,7 @@ func TestRunMP4IngestWorkerFMP4DirectKeepsTracks(t *testing.T) { // TestRunMP4IngestWorkerRecords proves debug recording works INSIDE the worker: // with cfg.Record set and a DataDir handed over, the worker tees its ingest -// media to debug-recordings//.rtmp.mp4. This is what keeps debug +// media to debug-recordings//.mp4. This is what keeps debug // recording working on the isolated paths where main is out of the data path // (it can't tee the bytes itself), so main decides and the worker records. func TestRunMP4IngestWorkerRecords(t *testing.T) { @@ -258,10 +258,10 @@ func TestRunMP4IngestWorkerRecords(t *testing.T) { // Frames are irrelevant here (recording is on the input side); discard them. require.NoError(t, RunMP4IngestWorker(ctx, cfg, bytes.NewReader(mp4), ingestframe.NewWriter(io.Discard), func() []byte { return cfg.Manifest })) - // The recording lands at debug-recordings//.rtmp.mp4 and + // The recording lands at debug-recordings//.mp4 and // must contain exactly the media the worker ingested. The dump goroutine // flushes asynchronously, so allow it a moment to finish the last write. - glob := filepath.Join(dataDir, "debug-recordings", "*", "*.rtmp.mp4") + glob := filepath.Join(dataDir, "debug-recordings", "*", "*.mp4") require.Eventually(t, func() bool { matches, _ := filepath.Glob(glob) if len(matches) != 1 { @@ -420,11 +420,11 @@ func TestRunMP4IngestWorkerRecordsToS3(t *testing.T) { require.NoError(t, RunMP4IngestWorker(ctx, cfg, bytes.NewReader(mp4), ingestframe.NewWriter(io.Discard), func() []byte { return cfg.Manifest })) // The upload commits asynchronously (the dump goroutine's Close); the object - // must appear at debug-bucket/debug-recordings//.rtmp.mp4 holding + // must appear at debug-bucket/debug-recordings//.mp4 holding // exactly the ingested media. wantPrefix := "debug-bucket/debug-recordings/" + ms.Streamer() + "/" require.Eventually(t, func() bool { - got, ok := fake.objectWithPrefix(wantPrefix, ".rtmp.mp4") + got, ok := fake.objectWithPrefix(wantPrefix, ".mp4") return ok && bytes.Equal(got, mp4) }, 10*time.Second, 25*time.Millisecond, "worker streams the recording to the S3 bucket verbatim") diff --git a/pkg/media/mp4_ingest.go b/pkg/media/mp4_ingest.go index 19b40d7ad..4da54c5c2 100644 --- a/pkg/media/mp4_ingest.go +++ b/pkg/media/mp4_ingest.go @@ -28,7 +28,7 @@ func (mm *MediaManager) MP4Ingest(ctx context.Context, input io.Reader, ms Media if shouldRecord { log.Log(ctx, "recording ingest stream to file", "streamer", ms.Streamer()) var finalize func() - input, finalize = mm.recordTee(ctx, input, ms.Streamer(), ".rtmp.mp4") + input, finalize = mm.recordTee(ctx, input, ms.Streamer()) defer finalize() } else { log.Log(ctx, "not recording ingest stream to file", "streamer", ms.Streamer()) @@ -200,7 +200,7 @@ const debugRecordingFlushTimeout = 5 * time.Minute // cleanup, which for S3 waits on every outstanding request) and // recordingTeeWriter drops every later write. The failure is logged; losing a // recording is acceptable, stalling a broadcast is not. -func (mm *MediaManager) recordTee(ctx context.Context, r io.Reader, user string, filesuffix string) (io.Reader, func()) { +func (mm *MediaManager) recordTee(ctx context.Context, r io.Reader, user string) (io.Reader, func()) { pr, pw := io.Pipe() done := make(chan struct{}) // stopIngest is how the dump releases the live stream: it closes the tee's @@ -208,7 +208,7 @@ func (mm *MediaManager) recordTee(ctx context.Context, r io.Reader, user string, stopIngest := func(err error) { _ = pw.CloseWithError(err) } go func() { defer close(done) - if err := mm.dumpToFile(ctx, pr, user, filesuffix, stopIngest); err != nil { + if err := mm.dumpToFile(ctx, pr, user, stopIngest); err != nil { log.Error(ctx, "debug recording failed; continuing without it", "error", err, "streamer", user) } }() @@ -240,9 +240,8 @@ func (t recordingTeeWriter) Write(p []byte) (int, error) { // S3 Close commits the upload (flushing a final part and waiting on every // outstanding request, bounded only by the s3 package's per-operation timeouts), // and the live ingest stream must not sit behind that wait. -func (mm *MediaManager) dumpToFile(ctx context.Context, r io.Reader, user string, filesuffix string, stopIngest func(error)) error { - now := aqtime.FromTime(time.Now()) - filename := fmt.Sprintf("%s%s", now.FileSafeString(), filesuffix) +func (mm *MediaManager) dumpToFile(ctx context.Context, r io.Reader, user string, stopIngest func(error)) error { + filename := aqtime.FromTime(time.Now()).FileSafeString() + ".mp4" // Streams to S3 when configured (production), else a local file under DataDir // (dev). Close finalizes either target — for S3 it commits the upload. f, err := mm.cli.DebugRecordingCreate(ctx, []string{"debug-recordings", user, filename}, "video/mp4", false) diff --git a/pkg/media/record_tee_test.go b/pkg/media/record_tee_test.go index 649ea8e17..9c98fd395 100644 --- a/pkg/media/record_tee_test.go +++ b/pkg/media/record_tee_test.go @@ -75,7 +75,7 @@ func TestRecordTeeSurvivesRecorderCreateFailure(t *testing.T) { mm := recordingTeeTestManager(t, srv.URL) payload := recordingTeePayload(1 << 20) - input, finalize := mm.recordTee(context.Background(), bytes.NewReader(payload), "did:plc:test", ".rtmp.mp4") + input, finalize := mm.recordTee(context.Background(), bytes.NewReader(payload), "did:plc:test") defer finalize() require.Equal(t, payload, readTeeWithin(t, input, 15*time.Second), "the ingest stream reaches its consumer verbatim") @@ -146,7 +146,7 @@ func TestRecordTeeSurvivesRecorderMidStreamFailure(t *testing.T) { // 18 MB puts the first part on the wire with 6 MB still to come: exactly the // in-flight stream a dead recorder used to strand. payload := recordingTeePayload(24 << 20) - input, finalize := mm.recordTee(context.Background(), midStreamInput(payload, fake), "did:plc:test", ".rtmp.mp4") + input, finalize := mm.recordTee(context.Background(), midStreamInput(payload, fake), "did:plc:test") defer finalize() require.Equal(t, payload, readTeeWithin(t, input, 30*time.Second), "the ingest stream reaches its consumer verbatim") @@ -171,7 +171,7 @@ func TestRecordTeeSurvivesRecorderCleanupStall(t *testing.T) { mm := recordingTeeTestManager(t, srv.URL) payload := recordingTeePayload(24 << 20) - input, finalize := mm.recordTee(context.Background(), midStreamInput(payload, fake), "did:plc:test", ".rtmp.mp4") + input, finalize := mm.recordTee(context.Background(), midStreamInput(payload, fake), "did:plc:test") // Unblock the stalled request before the dump finishes and before the server // closes (httptest's Close waits on outstanding handlers). defer finalize()