From 5da7a7ce9b9e5703d7e7f719d2cf609de088e7b2 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Fri, 2 Oct 2026 14:41:33 -0700 Subject: [PATCH] media: release ingest before the recording sink's cleanup wait MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-up. recordTee closed the pipe only once dumpToFile returned, but the copy-error path in dumpToFile runs the sink's Close first — for S3 that commits the upload, flushing a final part and waiting on every outstanding request (bounded only by the s3 package's per-op timeouts). One rejected part plus one hung request therefore still held the live stream for minutes. dumpToFile now reports "stopped reading" the moment io.Copy ends, so the tee unblocks ingest before that cleanup; TestRecordTeeSurvivesRecorderCleanupStall (a rejected part plus a part the fake never answers) hangs >30s without it. Also wait past the node's 30s live-window retention in the web spec's liveness check, so a stream that stopped producing segments can't pass on stale window entries. --- .../flows/best-effort-recording.spec.ts | 8 +-- pkg/media/ingest_worker_test.go | 28 ++++++++-- pkg/media/mp4_ingest.go | 27 +++++++--- pkg/media/record_tee_test.go | 53 +++++++++++++++---- 4 files changed, 93 insertions(+), 23 deletions(-) diff --git a/js/e2e-web/flows/best-effort-recording.spec.ts b/js/e2e-web/flows/best-effort-recording.spec.ts index f6210716b..04d09d8e9 100644 --- a/js/e2e-web/flows/best-effort-recording.spec.ts +++ b/js/e2e-web/flows/best-effort-recording.spec.ts @@ -31,9 +31,11 @@ test("best-effort-recording: a failing recording sink does not block playback", // And the stream stays up. The harness restarts its WHIP stream every few // seconds, so checking well past that window means recording failed again and - // again without the broadcast going dark. The playlist query is served from the - // node's in-memory live window, so it only succeeds while segments are flowing. - await page.waitForTimeout(15_000); + // again without the broadcast going dark. The wait also clears the node's + // 30s live-window retention: the playlist query is served from that window, so + // only a stream still ingesting fresh segments answers 200 after it (a stream + // that stopped would keep serving stale segments for up to 30s). + await page.waitForTimeout(35_000); await expect .poll( async () => { diff --git a/pkg/media/ingest_worker_test.go b/pkg/media/ingest_worker_test.go index 2cbc43412..d5a298e22 100644 --- a/pkg/media/ingest_worker_test.go +++ b/pkg/media/ingest_worker_test.go @@ -227,16 +227,21 @@ type fakeS3Server struct { objects map[string][]byte // completed "/" → body // failInitiate rejects the multipart-upload create, failPart rejects part // uploads — the two shapes of a bucket that refuses writes (bad credentials, - // unpaid billing). The counters let a test prove the failure it asked for - // actually fired. + // unpaid billing). stallPartNum blocks the given part's request until + // releaseParts is closed, standing in for a hung connection the client is + // still waiting on after a sibling part failed. The counters let a test prove + // the failure it asked for actually fired. failInitiate bool failPart bool + stallPartNum string + releaseParts chan struct{} initiateFailures int partFailures int + partsStalled int } func newFakeS3Server() *fakeS3Server { - return &fakeS3Server{parts: map[string][]byte{}, objects: map[string][]byte{}} + return &fakeS3Server{parts: map[string][]byte{}, objects: map[string][]byte{}, releaseParts: make(chan struct{})} } // failureCounts returns how many create/part requests the fake rejected. @@ -246,7 +251,24 @@ func (f *fakeS3Server) failureCounts() (initiate, part int) { return f.initiateFailures, f.partFailures } +// stalledCount returns how many part requests have reached the stall and are +// still waiting on releaseParts. +func (f *fakeS3Server) stalledCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.partsStalled +} + func (f *fakeS3Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { + // Stall before taking the lock: a blocked request must not hold up the rest + // of the fake. + if f.stallPartNum != "" && r.Method == "PUT" && r.URL.Query().Get("partNumber") == f.stallPartNum { + f.mu.Lock() + f.partsStalled++ + release := f.releaseParts + f.mu.Unlock() + <-release + } f.mu.Lock() defer f.mu.Unlock() path := strings.TrimPrefix(r.URL.Path, "/") diff --git a/pkg/media/mp4_ingest.go b/pkg/media/mp4_ingest.go index 275376814..d0dbf407e 100644 --- a/pkg/media/mp4_ingest.go +++ b/pkg/media/mp4_ingest.go @@ -183,17 +183,20 @@ const debugRecordingFlushTimeout = 5 * time.Minute // mid-stream (billing, credentials), an unwritable disk — would otherwise leave // the pipe writer with no reader, blocking the next read of the ingest stream // forever and taking the whole broadcast down with it. Instead the dump -// goroutine closes the pipe with its error on the way out (unblocking any -// in-flight write) and recordingTeeWriter drops every later write. The failure -// is logged; losing a recording is acceptable, stalling a broadcast is not. +// disconnects the pipe the moment it stops reading it (before the sink's own +// 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()) { pr, pw := io.Pipe() done := make(chan struct{}) + // stopIngest is how the dump releases the live stream: it closes the tee's + // pipe (unblocking any in-flight write) and makes every later write a no-op. + stopIngest := func(err error) { _ = pw.CloseWithError(err) } go func() { defer close(done) - if err := mm.dumpToFile(ctx, pr, user, filesuffix); err != nil { + if err := mm.dumpToFile(ctx, pr, user, filesuffix, stopIngest); err != nil { log.Error(ctx, "debug recording failed; continuing without it", "error", err, "streamer", user) - _ = pw.CloseWithError(err) } }() finalize := func() { @@ -219,18 +222,26 @@ func (t recordingTeeWriter) Write(p []byte) (int, error) { return len(p), nil } -func (mm *MediaManager) dumpToFile(ctx context.Context, r io.Reader, user string, filesuffix string) error { +// dumpToFile copies the recording stream into the sink and finalizes it. +// stopIngest is called the moment copying stops, BEFORE the sink is closed: an +// 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) // 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) if err != nil { + stopIngest(err) return fmt.Errorf("failed to create debug recording: %w", err) } - if _, err = io.Copy(f, r); err != nil { + _, cerr := io.Copy(f, r) + stopIngest(cerr) + if cerr != nil { f.Close() - return fmt.Errorf("failed to copy to debug recording: %w", err) + return fmt.Errorf("failed to copy to debug recording: %w", cerr) } if err = f.Close(); err != nil { return fmt.Errorf("failed to finalize debug recording: %w", err) diff --git a/pkg/media/record_tee_test.go b/pkg/media/record_tee_test.go index 899958e4e..649ea8e17 100644 --- a/pkg/media/record_tee_test.go +++ b/pkg/media/record_tee_test.go @@ -117,6 +117,21 @@ func (g *gatedReader) Read(b []byte) (int, error) { return n, err } +// midStreamInput feeds payload through a gate that holds the tail until the fake +// has rejected a part upload, so the tee is provably mid-stream when the +// recorder dies rather than racing the upload's HTTP round trip. +func midStreamInput(payload []byte, fake *fakeS3Server) *gatedReader { + return &gatedReader{ + r: bytes.NewReader(payload), + gateAt: 18 << 20, + gate: func() bool { + _, parts := fake.failureCounts() + return parts > 0 + }, + delay: 250 * time.Microsecond, + } +} + // TestRecordTeeSurvivesRecorderMidStreamFailure covers the mid-stream shape: the // upload starts, a part is rejected (a long recording dies after its first // 16 MB part), and ingest must keep flowing from wherever the recorder died — @@ -131,18 +146,38 @@ 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(), &gatedReader{ - r: bytes.NewReader(payload), - gateAt: 18 << 20, - gate: func() bool { - _, parts := fake.failureCounts() - return parts > 0 - }, - delay: 250 * time.Microsecond, - }, "did:plc:test", ".rtmp.mp4") + input, finalize := mm.recordTee(context.Background(), midStreamInput(payload, fake), "did:plc:test", ".rtmp.mp4") defer finalize() require.Equal(t, payload, readTeeWithin(t, input, 30*time.Second), "the ingest stream reaches its consumer verbatim") _, parts := fake.failureCounts() require.Positive(t, parts, "the recorder must have attempted (and failed) a part upload mid-stream") } + +// TestRecordTeeSurvivesRecorderCleanupStall covers the wait AFTER the copy dies: +// one rejected part plus a request the bucket never answers. S3's Close commits +// the upload and waits on every outstanding request (bounded only by the s3 +// package's per-operation timeouts), so a tee that released ingest only once the +// dump returned would still hold the live stream hostage behind that cleanup. +// The tee must release ingest the moment copying stops. +func TestRecordTeeSurvivesRecorderCleanupStall(t *testing.T) { + fake := newFakeS3Server() + fake.failPart = true + // The tail becomes a second part at Close; hang it, so the completion wait is + // provably still in flight when ingest has to be released. + fake.stallPartNum = "2" + srv := httptest.NewServer(fake) + defer srv.Close() + + mm := recordingTeeTestManager(t, srv.URL) + payload := recordingTeePayload(24 << 20) + input, finalize := mm.recordTee(context.Background(), midStreamInput(payload, fake), "did:plc:test", ".rtmp.mp4") + // Unblock the stalled request before the dump finishes and before the server + // closes (httptest's Close waits on outstanding handlers). + defer finalize() + defer close(fake.releaseParts) + + require.Equal(t, payload, readTeeWithin(t, input, 30*time.Second), "ingest is released while the recording is still cleaning up") + require.Eventually(t, func() bool { return fake.stalledCount() > 0 }, 5*time.Second, 10*time.Millisecond, + "the stalled S3 request must still be outstanding when ingest was released") +} -- 2.51.2