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)