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