diff --git a/pkg/media/ingest_worker.go b/pkg/media/ingest_worker.go index e8e64b841..8221738a0 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 4ff83441c..7ecdd1457 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 c016539f4..112c2cbad 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) } } }()