diff --git a/pkg/media/ingest_subprocess_test.go b/pkg/media/ingest_subprocess_test.go index 0bdf83bb..26a3d1d2 100644 --- a/pkg/media/ingest_subprocess_test.go +++ b/pkg/media/ingest_subprocess_test.go @@ -153,9 +153,9 @@ func TestIngestWorkerSubprocess(t *testing.T) { t.Logf("worker subprocess emitted %d valid signed segments + clean End", segs) } -// TestMKVIngestIsolatedWedgeContained is the isolation guarantee: sample-stream.mkv -// carries four audio tracks, so the single-audio ingest pipeline leaves three -// matroskademux pads unlinked and wedges with no EOS — exactly the kind of native +// TestMKVIngestIsolatedWedgeContained is the isolation guarantee: an audio-only +// MKV starves the fMP4 muxer's video pad of both data and EOS, so the native +// pipeline wedges with no frames and no EOS — exactly the kind of native // wedge that would hang (or, with a runaway buffer, OOM-kill) an in-process // ingest and take the node with it. Run in a worker, it must be contained: the // watchdog kills the worker and MKVIngestIsolated returns an error, bounded in @@ -168,11 +168,10 @@ func TestMKVIngestIsolatedWedgeContained(t *testing.T) { mm, _ := getStaticTestMediaManager(t) ms := newBareSegmentSigner(t) - wedge, err := os.ReadFile(getFixture("sample-stream.mkv")) - require.NoError(t, err) + wedge := makeAudioOnlyAACMKV(t, context.Background(), 5) start := time.Now() - err = mm.MKVIngestIsolated(context.Background(), bytes.NewReader(wedge), ms) + err := mm.MKVIngestIsolated(context.Background(), bytes.NewReader(wedge), ms) elapsed := time.Since(start) require.Error(t, err, "a wedged worker must surface as an error, not a hang") diff --git a/pkg/media/ingest_worker_test.go b/pkg/media/ingest_worker_test.go index f4f16522..2615a631 100644 --- a/pkg/media/ingest_worker_test.go +++ b/pkg/media/ingest_worker_test.go @@ -74,6 +74,37 @@ func makeH264AACMKV(t *testing.T, ctx context.Context, srcMP4 string) []byte { return buf.Bytes() } +// makeAudioOnlyAACMKV synthesizes an AAC-audio-only streamable MKV — the +// canonical WEDGE input for watchdog/containment tests. The ingest pipeline +// hardwires a video and an audio branch; with no video track, matroskademux +// never creates a video pad, the fMP4 aggregator's video pad never sees data +// OR EOS, and the pipeline hangs forever with no frames and no EOS — a true +// native wedge that no queue sizing can fix. (The 4-audio sample-stream.mkv +// previously used for this stopped wedging once the ingest branches moved to +// Queue2Big: its wedge was really the 1s default-queue interleave deadlock.) +func makeAudioOnlyAACMKV(t *testing.T, ctx context.Context, seconds int) []byte { + t.Helper() + gstinit.InitGST() + desc := fmt.Sprintf("audiotestsrc num-buffers=%d samplesperbuffer=1024 ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! fdkaacenc ! aacparse ! matroskamux streamable=true ! appsink name=sink", seconds*47) + pipeline, err := gst.NewPipelineFromString(desc) + require.NoError(t, err) + + sinkEle, err := pipeline.GetElementByName("sink") + require.NoError(t, err) + var buf bytes.Buffer + app.SinkFromElement(sinkEle).SetCallbacks(&app.SinkCallbacks{ + NewSampleFunc: WriterNewSample(ctx, &buf), + }) + + busErr := make(chan error, 1) + go func() { busErr <- HandleBusMessages(ctx, pipeline) }() + require.NoError(t, pipeline.SetState(gst.StatePlaying)) + defer func() { _ = pipeline.SetState(gst.StateNull) }() + require.NoError(t, <-busErr, "synthesize audio-only MKV") + require.NotEmpty(t, buf.Bytes()) + return buf.Bytes() +} + // TestRunMKVIngestWorkerProducesValidSignedFrames drives the isolated ingest // worker's core directly (no subprocess): feed it an H264+AAC MKV, collect the // framed output, and verify every emitted segment is a valid signed canonical @@ -191,11 +222,10 @@ func TestRunMKVIngestWorkerRecords(t *testing.T) { // TestRunMKVIngestWorkerSelfWatchdog proves the worker's OWN watchdog contains a // wedge. This is the only wedge containment on the detached/WHIP paths, where // main can't kill a detached worker — so the worker has to notice it's stuck and -// exit itself. The 4-audio sample-stream.mkv leaves matroskademux pads unlinked, -// so it wedges with no EOS and emits no frames; the watchdog must tear the -// pipeline down and return rather than hang forever. (The fd-4 path's -// supervisor-side watchdog is covered separately by -// TestMKVIngestIsolatedWedgeContained.) +// exit itself. An audio-only MKV starves the muxer's video pad of both data and +// EOS, so the pipeline wedges with no frames; the watchdog must tear it down +// and return rather than hang forever. (The fd-4 path's supervisor-side +// watchdog is covered separately by TestMKVIngestIsolatedWedgeContained.) func TestRunMKVIngestWorkerSelfWatchdog(t *testing.T) { old := ingestWorkerWatchdog ingestWorkerWatchdog = 3 * time.Second @@ -215,8 +245,7 @@ func TestRunMKVIngestWorkerSelfWatchdog(t *testing.T) { BroadcasterHost: "test.example.com", } - wedge, err := os.ReadFile(getFixture("sample-stream.mkv")) - require.NoError(t, err) + wedge := makeAudioOnlyAACMKV(t, ctx, 5) start := time.Now() done := make(chan error, 1) diff --git a/pkg/media/key_revocation_test.go b/pkg/media/key_revocation_test.go index de25b85e..0416a5c2 100644 --- a/pkg/media/key_revocation_test.go +++ b/pkg/media/key_revocation_test.go @@ -4,7 +4,6 @@ import ( "bytes" "context" "encoding/json" - "os" "testing" "time" @@ -90,8 +89,10 @@ func TestWatchKeyRevocationStreamKick(t *testing.T) { // TestMKVIngestIsolatedBanContained proves the fix end to end: banning a streamer // mid-ingest tears their isolated worker down. The watchdog is set generously -// (60s) and the input is the wedging 4-audio MKV that never ends on its own — so -// a timely return can only be the ban kill, not the watchdog or a natural EOS. +// (60s) and the input is a wedging audio-only MKV that never ends on its own — +// so a timely return can only be the ban kill, not the watchdog or a natural +// EOS. (The 4-audio sample-stream.mkv previously used here now ingests to +// completion in a few seconds, which would race the ban.) func TestMKVIngestIsolatedBanContained(t *testing.T) { old := ingestWorkerWatchdog ingestWorkerWatchdog = 60 * time.Second @@ -99,8 +100,7 @@ func TestMKVIngestIsolatedBanContained(t *testing.T) { mm, _ := getStaticTestMediaManager(t) ms := newBareSegmentSigner(t) - wedge, err := os.ReadFile(getFixture("sample-stream.mkv")) - require.NoError(t, err) + wedge := makeAudioOnlyAACMKV(t, context.Background(), 5) // Ban the streamer once the worker is up and the watcher has subscribed. go func() { @@ -112,7 +112,7 @@ func TestMKVIngestIsolatedBanContained(t *testing.T) { }() start := time.Now() - err = mm.MKVIngestIsolated(context.Background(), bytes.NewReader(wedge), ms) + err := mm.MKVIngestIsolated(context.Background(), bytes.NewReader(wedge), ms) elapsed := time.Since(start) require.Error(t, err, "a banned stream is torn down, surfaced as an error") diff --git a/pkg/media/mist_mkv_ingest_test.go b/pkg/media/mist_mkv_ingest_test.go new file mode 100644 index 00000000..4ff83441 --- /dev/null +++ b/pkg/media/mist_mkv_ingest_test.go @@ -0,0 +1,206 @@ +package media + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "os" + "strings" + "testing" + "time" + + "github.com/go-gst/go-gst/gst" + "github.com/go-gst/go-gst/gst/app" + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/crypto/signers" + "stream.place/streamplace/pkg/gstinit" + "stream.place/streamplace/pkg/ingestframe" +) + +// runMKVThroughIngestWorker feeds an MKV byte stream through the isolated +// ingest worker and returns how many signed canonical segments it emitted. The +// worker watchdog is shortened so a wedged pipeline returns promptly instead of +// hanging the test (override via SP_TEST_WATCHDOG to e.g. park a wedge for a +// stack dump). With transcode=true the node keys are supplied so the worker +// completes to dual-codec, as production does. +func runMKVThroughIngestWorker(t *testing.T, mkv []byte, transcode bool) (int, error) { + t.Helper() + old := ingestWorkerWatchdog + ingestWorkerWatchdog = 10 * time.Second + if wd := os.Getenv("SP_TEST_WATCHDOG"); wd != "" { + d, perr := time.ParseDuration(wd) + require.NoError(t, perr) + ingestWorkerWatchdog = d + } + defer func() { ingestWorkerWatchdog = old }() + + ctx := context.Background() + ms := newBareSegmentSigner(t) + keyPEM, err := signers.MarshalES256KPrivateKeyPEM(ms.Signer) + require.NoError(t, err) + manifest, err := ms.buildManifest(ctx, time.Now().UnixMilli()) + require.NoError(t, err) + cfg := IngestWorkerConfig{ + StreamerDID: ms.Streamer(), + KeyPEM: keyPEM, + CertPEM: ms.Cert, + Manifest: manifest, + BroadcasterHost: "test.example.com", + } + if transcode { + cfg.NodeCertPEM = ms.Cert + cfg.NodeKeyPEM = keyPEM + } + + var buf bytes.Buffer + runErr := RunMKVIngestWorker(ctx, cfg, bytes.NewReader(mkv), ingestframe.NewWriter(&buf), func() []byte { return cfg.Manifest }) + + r := ingestframe.NewReader(&buf) + segs := 0 + for { + typ, _, rerr := r.ReadFrame() + if errors.Is(rerr, io.EOF) { + break + } + require.NoError(t, rerr) + if typ == ingestframe.Segment { + segs++ + } + } + return segs, runErr +} + +// makeSparseVideoAACMKV synthesizes the stream shape that wedged production +// ingest: video that degrades to keyframe-only at a low rate (here 0.5fps — +// MistServer drops all delta frames when a push falls behind, leaving ~1s-apart +// keyframes) alongside continuous 48kHz AAC audio, in a streamable MKV. The +// audio branch must buffer a full video-frame gap while matroskademux walks to +// the next video frame; gst's default 1s-capped queue can't, and the +// aggregator-based fMP4 muxer deadlocks (see buildMKVIngestPipeline). +func makeSparseVideoAACMKV(t *testing.T, ctx context.Context, seconds int) []byte { + t.Helper() + gstinit.InitGST() + desc := strings.Join([]string{ + fmt.Sprintf("videotestsrc num-buffers=%d ! video/x-raw,width=320,height=240,framerate=1/2 ! x264enc key-int-max=1 tune=zerolatency speed-preset=ultrafast ! h264parse ! matroskamux name=mux streamable=true ! appsink name=sink", (seconds+1)/2), + fmt.Sprintf("audiotestsrc num-buffers=%d samplesperbuffer=1024 ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! fdkaacenc ! aacparse ! mux.", seconds*47), + }, "\n") + pipeline, err := gst.NewPipelineFromString(desc) + require.NoError(t, err) + + sinkEle, err := pipeline.GetElementByName("sink") + require.NoError(t, err) + var buf bytes.Buffer + app.SinkFromElement(sinkEle).SetCallbacks(&app.SinkCallbacks{ + NewSampleFunc: WriterNewSample(ctx, &buf), + }) + + busErr := make(chan error, 1) + go func() { busErr <- HandleBusMessages(ctx, pipeline) }() + require.NoError(t, pipeline.SetState(gst.StatePlaying)) + defer func() { _ = pipeline.SetState(gst.StateNull) }() + require.NoError(t, <-busErr, "synthesize sparse-video MKV") + require.NotEmpty(t, buf.Bytes()) + return buf.Bytes() +} + +// TestMKVIngestSparseVideoNoWedge is the regression test for a production +// ingest wedge: a stream whose video goes sparse (keyframe-only, ≥1s between +// video frames) deadlocked the MKV ingest pipeline — audio backpressure +// through the default 1s-capped queue blocked the demux, the fMP4 aggregator +// starved on its video pad, and the stream hung with no EOS until the +// watchdog killed it. With Queue2Big on both ingest branches the same stream +// must segment to completion. +func TestMKVIngestSparseVideoNoWedge(t *testing.T) { + ctx := context.Background() + mkv := makeSparseVideoAACMKV(t, ctx, 12) + + segs, err := runMKVThroughIngestWorker(t, mkv, false) + require.NoError(t, err, "sparse-video stream ingests cleanly (wedge → watchdog → context canceled)") + // 12s of 2s-apart keyframes ≈ 6 GoPs; wedging yields 0-1 segments. + require.GreaterOrEqual(t, segs, 4, "sparse-video stream emits its segments") + t.Logf("sparse-video stream: %d segments", segs) +} + +// TestMKVIngestMistMetadataTrack: MistServer's MKV push declares a +// live-metadata track (CodecID M_JSON, TrackType 3) as track 1, ahead of the +// AAC audio and H264 video tracks. It was the initial suspect for the +// production wedge but proved benign — matroskademux ignores the unknown +// codec, and the same media segments identically with the 30-byte M_JSON +// TrackEntry stripped (the control). Kept as a canary for the MistServer +// track layout. (The real wedge: TestMKVIngestSparseVideoNoWedge.) +func TestMKVIngestMistMetadataTrack(t *testing.T) { + control, err := os.ReadFile("../../tmp-debug/nyc-head-nojson.mkv") + if err != nil { + t.Skipf("production sample not present: %v", err) + } + mist, err := os.ReadFile("../../tmp-debug/nyc-head.mkv") + require.NoError(t, err) + + segs, err := runMKVThroughIngestWorker(t, control, false) + require.NoError(t, err, "control (M_JSON TrackEntry stripped) ingests cleanly") + require.GreaterOrEqual(t, segs, 1, "control emits segments") + t.Logf("control: %d segments", segs) + + segs, err = runMKVThroughIngestWorker(t, mist, false) + require.NoError(t, err, "MistServer MKV (with M_JSON track) ingests cleanly") + require.GreaterOrEqual(t, segs, 1, "MistServer MKV emits segments") + t.Logf("with M_JSON track: %d segments", segs) +} + +// TestMKVIngestMistFullSample runs the entire 164s production capture through +// the worker with node transcode keys — the closest in-process approximation +// of the production ingest. The capture degrades to keyframe-only video at +// ~140s (MistServer behind-push frame-drop), which is what wedged production; +// with Queue2Big on the ingest branches the whole capture must segment. +func TestMKVIngestMistFullSample(t *testing.T) { + // KNOWN ISSUE (follow-up): with node keys, a full-speed feed of the whole + // capture progressively slows the transcode completion path (fine for 30 + // GoPs — see TestMKVIngestMistTail with transcode — but the ~170-GoP run + // degrades until the watchdog fires) and the worker then hangs in the + // post-cancel drain: transcoder.Feed blocks on its jobs channel under a + // context.WithoutCancel ctx, so RunMKVIngestWorker never returns. + if os.Getenv("SP_SLOW_TESTS") == "" { + t.Skip("slow/known-hanging with transcode; set SP_SLOW_TESTS=1 to run") + } + mkv, err := os.ReadFile("../../tmp-debug/nyc-full.mkv") + if err != nil { + t.Skipf("production sample not present: %v", err) + } + segs, err := runMKVThroughIngestWorker(t, mkv, true) + t.Logf("full sample: %d segments, err=%v", segs, err) + require.NoError(t, err, "full production sample ingests cleanly") + // 171 GoPs in the capture (~1s each); wedging yielded ~144. + require.GreaterOrEqual(t, segs, 160, "full sample emits the whole stream's segments") +} + +// TestMKVIngestMistTail is the fast sample-based wedge check: the tail sample +// starts at 135s, ~5s before the capture goes keyframe-only, so an unfixed +// pipeline wedges within seconds (4 segments) instead of minutes. +func TestMKVIngestMistTail(t *testing.T) { + mkv, err := os.ReadFile("../../tmp-debug/nyc-tail135.mkv") + if err != nil { + t.Skipf("production sample not present: %v", err) + } + segs, err := runMKVThroughIngestWorker(t, mkv, false) + t.Logf("tail sample: %d segments, err=%v", segs, err) + require.NoError(t, err, "tail of production sample ingests cleanly") + // 135s..164s at ~1s GoPs ≈ 29 segments; wedging yields ~5. + require.GreaterOrEqual(t, segs, 20, "tail sample emits segments past the keyframe-only transition") +} + +// TestMKVIngestMistFullSampleNoTranscode is TestMKVIngestMistFullSample +// without node keys (segment+sign only). During diagnosis this proved the +// wedge was in the core ingest pipeline, not the dual-codec transcode stage — +// both variants wedged at the same GoP. +func TestMKVIngestMistFullSampleNoTranscode(t *testing.T) { + mkv, err := os.ReadFile("../../tmp-debug/nyc-full.mkv") + if err != nil { + t.Skipf("production sample not present: %v", err) + } + segs, err := runMKVThroughIngestWorker(t, mkv, false) + t.Logf("full sample (no transcode): %d segments, err=%v", segs, err) + require.NoError(t, err, "full production sample ingests cleanly without transcode") + require.GreaterOrEqual(t, segs, 70, "full sample emits the whole stream's segments") +} diff --git a/pkg/media/mkv_ingest.go b/pkg/media/mkv_ingest.go index aac3a36c..03ae5d88 100644 --- a/pkg/media/mkv_ingest.go +++ b/pkg/media/mkv_ingest.go @@ -10,6 +10,7 @@ import ( "github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" "stream.place/streamplace/pkg/aqtime" + "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/log" ) @@ -70,10 +71,21 @@ func (mm *MediaManager) MKVIngest(ctx context.Context, input io.Reader, ms Media // where signerElem routes its segments (ValidateMP4 vs. a frame writer to the // main process). func buildMKVIngestPipeline(ctx context.Context, input io.Reader, signerElem *gst.Element) (*gst.Pipeline, error) { + // Queue sizing: matroskademux feeds both branches from one thread, and the + // fMP4 muxer downstream is an aggregator — it consumes NOTHING until every + // pad has data. If the video track goes sparse (e.g. MistServer drops all + // delta frames when a push falls behind, leaving ~1s-apart keyframes), the + // audio branch must buffer a full video-frame gap while the demux walks the + // byte stream to the next video frame. gst's default queue caps at + // max-size-time=1s, so a ≥1s video gap fills the audio queue, blocks the + // demux, starves the muxer's video pad, and deadlocks the whole graph with + // no EOS — a live stream wedges until the watchdog kills it. Use the shared + // Queue2Big preset (no time/buffer cap, generous byte cap) like the other + // demux-fed pipelines (transcode, rtmp_push, packetize, media_data_parser). pipelineSlice := []string{ "appsrc name=streamsrc ! matroskademux name=demux", - "demux. ! queue ! h264parse name=parse", - "demux. ! queue ! fdkaacdec ! audioresample ! opusenc name=audioenc", + "demux. ! " + constants.Queue2Big + " ! h264parse name=parse", + "demux. ! " + constants.Queue2Big + " ! fdkaacdec ! audioresample ! opusenc name=audioenc", } pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n")) if err != nil { diff --git a/pkg/media/muxl_segment.go b/pkg/media/muxl_segment.go index 1de14d82..c016539f 100644 --- a/pkg/media/muxl_segment.go +++ b/pkg/media/muxl_segment.go @@ -78,8 +78,15 @@ func muxlSignSegmentElem(ctx context.Context, cli *config.CLI, signStream SignSe return nil, nil, fmt.Errorf("failed to add audio ghost pad to bin") } + // sync=false: this sink feeds the signer, not a display — render as fast as + // upstream produces. The default (sync=true) made the appsink wait on the + // pipeline clock per buffer, pacing the whole ingest graph at realtime: + // harmless for a live source arriving at 1x, but it throttled tests/replays + // and kept the graph's queues near-full for no benefit. Every other appsink + // in the tree already sets this. appsink, err := gst.NewElementWithProperties("appsink", map[string]any{ "name": "muxl-appsink", + "sync": false, }) if err != nil { return nil, nil, fmt.Errorf("failed to create appsink element: %w", err) -- 2.51.2 From fe88e26166ee2d8073d4b5655a8fbad4824c8993 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Wed, 8 Jul 2026 07:12:32 -0700 Subject: [PATCH 2/4] media: set sync=false on the remaining non-pacing appsinks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit follow-up to the muxl-appsink sync fix: swept every appsink in the tree for clock sync. - thumbnailFromMP4 (thumbnail.go): the flat-MP4 thumbnail pipeline was the only consumer-path appsink still defaulting to sync=true. In practice videorate re-stamps the selected frame to PTS 0 so it usually rendered immediately, but there's no reason to leave it on the clock — and its twin Thumbnail() already sets sync=false. - SelfTest (gstreamer.go): untimestamped buffers so sync never actually waited; set sync=false for consistency so the tree greps clean. - sp whip (cmd/whip.go): the ONE appsink pair where sync=true is load-bearing — the client plays a file over WHIP as a live stream and the appsink clock wait is the only realtime pacing (pion WriteSample pushes immediately). Made sync=true explicit with a comment so a future sweep doesn't "fix" it. Already sync=false (no change): packetize, media_data_parser, segment_conv, transcode, vod_pipeline, segmenter (via sink-added), Thumbnail, muxl_segment. Co-Authored-By: Claude Fable 5 --- pkg/cmd/whip.go | 9 +++++++-- pkg/media/gstreamer.go | 1 + pkg/media/thumbnail.go | 2 +- 3 files changed, 9 insertions(+), 3 deletions(-) diff --git a/pkg/cmd/whip.go b/pkg/cmd/whip.go index 2f67e8b6..920f4702 100644 --- a/pkg/cmd/whip.go +++ b/pkg/cmd/whip.go @@ -74,8 +74,13 @@ func (w *WHIPClient) WHIP(ctx context.Context) error { "filesrc name=filesrc ! qtdemux name=demux", "demux.video_0 ! tee name=video_tee", "demux.audio_0 ! tee name=audio_tee", - "video_tee. ! queue ! h264parse config-interval=-1 ! video/x-h264,stream-format=byte-stream ! appsink name=videoappsink", - "audio_tee. ! queue ! opusparse ! appsink name=audioappsink", + // sync=true (the default) is load-bearing here, unlike every other + // appsink in the tree: these sinks are what pace the file at realtime + // so it plays as a live stream — WriteSample pushes to WebRTC + // immediately, so without clock sync the whole file would blast + // through in one burst. + "video_tee. ! queue ! h264parse config-interval=-1 ! video/x-h264,stream-format=byte-stream ! appsink sync=true name=videoappsink", + "audio_tee. ! queue ! opusparse ! appsink sync=true name=audioappsink", // "matroskamux name=mux ! fakesink name=fakesink sync=true", // "video_tee. ! mux.video_0", // "audio_tee. ! mux.audio_0", diff --git a/pkg/media/gstreamer.go b/pkg/media/gstreamer.go index 140f9460..f3c1f4d1 100644 --- a/pkg/media/gstreamer.go +++ b/pkg/media/gstreamer.go @@ -68,6 +68,7 @@ func SelfTest(ctx context.Context) error { sinkele, err := gst.NewElementWithProperties("appsink", map[string]interface{}{ "name": "self-test-sink", + "sync": false, }) if err != nil { return fmt.Errorf("failed to create appsink element: %w", err) diff --git a/pkg/media/thumbnail.go b/pkg/media/thumbnail.go index 1dd5a998..7b2d0a0c 100644 --- a/pkg/media/thumbnail.go +++ b/pkg/media/thumbnail.go @@ -71,7 +71,7 @@ func thumbnailFromMP4(ctx context.Context, flat []byte, w io.Writer, format stri pipeline, err := gst.NewPipelineFromString(strings.Join([]string{ "appsrc name=src ! decodebin ! videoconvert ! videoscale ! videorate ! capsfilter caps=video/x-raw,width=[1,1280],height=[1,720],pixel-aspect-ratio=1/1,framerate=1/999999 ! ", encoder, - " ! appsink name=appsink", + " ! appsink sync=false name=appsink", }, "\n")) if err != nil { return fmt.Errorf("create thumbnail pipeline: %w", err) -- 2.51.2 From 50be92bf251c8b1fe4011f13b249d4487c5c56d8 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Wed, 8 Jul 2026 15:14:27 -0700 Subject: [PATCH 3/4] media: run the signer drain on a non-cancellable ctx to fix flush deadlock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RunMKVIngestWorker uses ctx cancellation as its "flush the final GoP" signal after EOS. But muxl's event parser treats a cancelled ctx as "abandon the stream mid-read": it returns early, nobody reads or closes the wasm's stdout pipe, and the wasm deadlocks against its next write — the sign/unwrap call never returns, done never closes, and the worker hangs forever at `cancel(); <-done`. While ingest was clock-paced (appsink sync=true) every GoP had drained long before EOS+cancel arrived, so the race never lost. Once ingest ran at full speed, EOS+cancel landed while GoPs were still in flight, and audioCompletionTarget then STARTED a muxl unwrap with an already-dead ctx — a guaranteed deadlock. That's what wedged GitLab CI's TestWorkerServesFramesOverSocket into the go test 30m timeout (GitHub's slower runners kept winning the race). Fix, in two spots: - muxlSignSegmentElem: the signer + event drain run on context.WithoutCancel. Cancel still triggers the flush — it closes the input pipe, the signer sees EOF, signs the final GoP, and exits cleanly — it just no longer aborts the parser mid-stream. - workerSegmentSink: audioCompletionTarget's unwrap likewise runs WithoutCancel (same reasoning as the transcoder it feeds). Fallout, both good: - The old race was silently DROPPING the final GoP even when it didn't deadlock: the socket e2e test now serves 2 segments (was 1), the 135s-tail sample 31 (was 29). - The "known issue" from the previous commit — the full 164s production capture with node transcode keys progressively slowing until the watchdog fired, then hanging in the post-cancel drain — was this same bug. It now transcodes the whole capture in ~13s, so TestMKVIngestMistFullSample's SP_SLOW_TESTS skip-gate is removed. Upstream follow-up (streamplace/muxl): runWith should close its stdout pipe reader when parseEvents returns early, so a cancelled ctx can't permanently wedge a wasm instance; parseEvents' ctx.Done select can also randomly drop tail events. vod/livehls call sites remain exposed to that until it's fixed there. Co-Authored-By: Claude Fable 5 --- pkg/media/ingest_worker.go | 6 +++++- pkg/media/mist_mkv_ingest_test.go | 14 +++++--------- pkg/media/muxl_segment.go | 19 +++++++++++++++---- 3 files changed, 25 insertions(+), 14 deletions(-) diff --git a/pkg/media/ingest_worker.go b/pkg/media/ingest_worker.go index e8e64b84..8221738a 100644 --- a/pkg/media/ingest_worker.go +++ b/pkg/media/ingest_worker.go @@ -160,7 +160,11 @@ func (mm *MediaManager) workerSegmentSink(ctx context.Context, cfg IngestWorkerC return frames.Segment(segment) // no node signer → single-codec } if transcoder == nil { - target, need := mm.audioCompletionTarget(ctx, segment) + // WithoutCancel for the same reason as the transcoder below: this + // runs from the signer's post-cancel drain, and a muxl unwrap + // started with a cancelled ctx deadlocks its wasm mid-stream + // instead of returning. + target, need := mm.audioCompletionTarget(context.WithoutCancel(ctx), segment) if !need { return frames.Segment(segment) // already dual-codec / no audio track } diff --git a/pkg/media/mist_mkv_ingest_test.go b/pkg/media/mist_mkv_ingest_test.go index 4ff83441..7ecdd145 100644 --- a/pkg/media/mist_mkv_ingest_test.go +++ b/pkg/media/mist_mkv_ingest_test.go @@ -155,15 +155,11 @@ func TestMKVIngestMistMetadataTrack(t *testing.T) { // ~140s (MistServer behind-push frame-drop), which is what wedged production; // with Queue2Big on the ingest branches the whole capture must segment. func TestMKVIngestMistFullSample(t *testing.T) { - // KNOWN ISSUE (follow-up): with node keys, a full-speed feed of the whole - // capture progressively slows the transcode completion path (fine for 30 - // GoPs — see TestMKVIngestMistTail with transcode — but the ~170-GoP run - // degrades until the watchdog fires) and the worker then hangs in the - // post-cancel drain: transcoder.Feed blocks on its jobs channel under a - // context.WithoutCancel ctx, so RunMKVIngestWorker never returns. - if os.Getenv("SP_SLOW_TESTS") == "" { - t.Skip("slow/known-hanging with transcode; set SP_SLOW_TESTS=1 to run") - } + // This once "progressively slowed until the watchdog fired, then hung in + // the post-cancel drain" and was skip-gated as known-hanging — that was + // the muxl-event-drain-vs-cancel deadlock (see muxlSignSegmentElem's + // drainCtx); with the drain non-cancellable the full capture transcodes + // at full speed (~13s). mkv, err := os.ReadFile("../../tmp-debug/nyc-full.mkv") if err != nil { t.Skipf("production sample not present: %v", err) diff --git a/pkg/media/muxl_segment.go b/pkg/media/muxl_segment.go index c016539f..112c2cba 100644 --- a/pkg/media/muxl_segment.go +++ b/pkg/media/muxl_segment.go @@ -104,11 +104,22 @@ func muxlSignSegmentElem(ctx context.Context, cli *config.CLI, signStream SignSe r.Close() }() + // The signer and its event drain run on a non-cancellable ctx: cancelling + // ctx is the FLUSH signal, not an abort — it closes the input pipe above, + // the signer sees EOF, signs the final GoP, and exits cleanly. If the + // cancelled ctx reached muxl's event parser instead, the parser would + // abandon the stream mid-write and the signer wasm would deadlock against + // the unread stdout pipe — done would never close and the caller's drain + // (`cancel(); <-done`) would hang forever. That was rare while ingest was + // clock-paced (everything had drained by EOS); at full speed EOS+cancel + // land while GoPs are still in flight, and the abort path lost every time. + drainCtx := context.WithoutCancel(ctx) + // Stream the fMP4 through the per-segment signer; each event carries one // GoP's per-track signed canonical segments. eventCh := make(chan *muxl.MuxlEvent, 16) go func() { - err := signStream(ctx, r, eventCh) + err := signStream(drainCtx, r, eventCh) close(eventCh) if err != nil && ctx.Err() == nil { log.Error(ctx, "error running muxl sign-segment", "error", err) @@ -122,9 +133,9 @@ func muxlSignSegmentElem(ctx context.Context, cli *config.CLI, signStream SignSe continue } segment := concatTracksSorted(ev.Tracks) - cli.DumpDebugSegment(ctx, "muxl_signed_segment.m4s", bytes.NewReader(segment)) - if err := onSegment(ctx, segment); err != nil { - log.Error(ctx, "error handling signed segment", "error", err) + cli.DumpDebugSegment(drainCtx, "muxl_signed_segment.m4s", bytes.NewReader(segment)) + if err := onSegment(drainCtx, segment); err != nil { + log.Error(drainCtx, "error handling signed segment", "error", err) } } }() -- 2.51.2 From 6001b4b2d8f9ee929d11267af484cf049321f5b5 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Wed, 8 Jul 2026 18:39:41 -0700 Subject: [PATCH 4/4] media: move the nyc production capture to remote fixtures The four MistServer-capture ingest tests read tmp-debug/nyc-*.mkv and skipped everywhere but the machine that held the capture. The cuts are now uploaded to the fixture bucket, so the tests run anywhere (CI included) via remote.RemoteFixture, like the transcode-wedge fixtures: - nyc-head / nyc-head-nojson (10.5MB each): M_JSON metadata-track canary - nyc-tail135 (20MB): fast wedge check across the keyframe-only transition - nyc-full (124MB): the whole 164s capture, with and without transcode Fixture downloads are content-addressed and cached, so a warm runner pays nothing; a cold CI pod downloads ~165MB once per run. Co-Authored-By: Claude Fable 5 --- pkg/media/mist_mkv_ingest_test.go | 33 +++++++++++++++---------------- 1 file changed, 16 insertions(+), 17 deletions(-) diff --git a/pkg/media/mist_mkv_ingest_test.go b/pkg/media/mist_mkv_ingest_test.go index 7ecdd145..e0bc22cd 100644 --- a/pkg/media/mist_mkv_ingest_test.go +++ b/pkg/media/mist_mkv_ingest_test.go @@ -17,6 +17,7 @@ import ( "stream.place/streamplace/pkg/crypto/signers" "stream.place/streamplace/pkg/gstinit" "stream.place/streamplace/pkg/ingestframe" + "stream.place/streamplace/test/remote" ) // runMKVThroughIngestWorker feeds an MKV byte stream through the isolated @@ -123,6 +124,12 @@ func TestMKVIngestSparseVideoNoWedge(t *testing.T) { t.Logf("sparse-video stream: %d segments", segs) } +// The nyc-* fixtures are cuts of a real 164s production MistServer MKV push +// whose video degrades to keyframe-only at ~140s (behind-push frame-drop) — +// the capture that wedged production ingest. head = first ~10s; head-nojson = +// the same bytes with the 30-byte M_JSON TrackEntry stripped; tail135 = from +// 135s (~5s before the keyframe-only transition); full = the whole capture. + // TestMKVIngestMistMetadataTrack: MistServer's MKV push declares a // live-metadata track (CodecID M_JSON, TrackType 3) as track 1, ahead of the // AAC audio and H264 video tracks. It was the initial suspect for the @@ -131,11 +138,9 @@ func TestMKVIngestSparseVideoNoWedge(t *testing.T) { // TrackEntry stripped (the control). Kept as a canary for the MistServer // track layout. (The real wedge: TestMKVIngestSparseVideoNoWedge.) func TestMKVIngestMistMetadataTrack(t *testing.T) { - control, err := os.ReadFile("../../tmp-debug/nyc-head-nojson.mkv") - if err != nil { - t.Skipf("production sample not present: %v", err) - } - mist, err := os.ReadFile("../../tmp-debug/nyc-head.mkv") + control, err := os.ReadFile(remote.RemoteFixture("3284ef5658e7864bce326c296a909e985c4167d0b9a445b2ce944c2f0171c71e/nyc-head-nojson.mkv")) + require.NoError(t, err) + mist, err := os.ReadFile(remote.RemoteFixture("c0989e044f3350c55f1e129b76252bfb2859914058bb17d2431a605db9693467/nyc-head.mkv")) require.NoError(t, err) segs, err := runMKVThroughIngestWorker(t, control, false) @@ -160,10 +165,8 @@ func TestMKVIngestMistFullSample(t *testing.T) { // the muxl-event-drain-vs-cancel deadlock (see muxlSignSegmentElem's // drainCtx); with the drain non-cancellable the full capture transcodes // at full speed (~13s). - mkv, err := os.ReadFile("../../tmp-debug/nyc-full.mkv") - if err != nil { - t.Skipf("production sample not present: %v", err) - } + mkv, err := os.ReadFile(remote.RemoteFixture("3e4e5d9758e67053908e523379a3e2ef2cf60679d0657a940daf96590e866015/nyc-full.mkv")) + require.NoError(t, err) segs, err := runMKVThroughIngestWorker(t, mkv, true) t.Logf("full sample: %d segments, err=%v", segs, err) require.NoError(t, err, "full production sample ingests cleanly") @@ -175,10 +178,8 @@ func TestMKVIngestMistFullSample(t *testing.T) { // starts at 135s, ~5s before the capture goes keyframe-only, so an unfixed // pipeline wedges within seconds (4 segments) instead of minutes. func TestMKVIngestMistTail(t *testing.T) { - mkv, err := os.ReadFile("../../tmp-debug/nyc-tail135.mkv") - if err != nil { - t.Skipf("production sample not present: %v", err) - } + mkv, err := os.ReadFile(remote.RemoteFixture("03df698a342f1ab89dccc20ce0a0283e1270104e6382703575686c9f4a88881e/nyc-tail135.mkv")) + require.NoError(t, err) segs, err := runMKVThroughIngestWorker(t, mkv, false) t.Logf("tail sample: %d segments, err=%v", segs, err) require.NoError(t, err, "tail of production sample ingests cleanly") @@ -191,10 +192,8 @@ func TestMKVIngestMistTail(t *testing.T) { // wedge was in the core ingest pipeline, not the dual-codec transcode stage — // both variants wedged at the same GoP. func TestMKVIngestMistFullSampleNoTranscode(t *testing.T) { - mkv, err := os.ReadFile("../../tmp-debug/nyc-full.mkv") - if err != nil { - t.Skipf("production sample not present: %v", err) - } + mkv, err := os.ReadFile(remote.RemoteFixture("3e4e5d9758e67053908e523379a3e2ef2cf60679d0657a940daf96590e866015/nyc-full.mkv")) + require.NoError(t, err) segs, err := runMKVThroughIngestWorker(t, mkv, false) t.Logf("full sample (no transcode): %d segments, err=%v", segs, err) require.NoError(t, err, "full production sample ingests cleanly without transcode")