diff --git a/pkg/api/playback.go b/pkg/api/playback.go index 1445d37e4..260060d59 100644 --- a/pkg/api/playback.go +++ b/pkg/api/playback.go @@ -105,15 +105,30 @@ func (a *StreamplaceAPI) HandleWebRTCIngest(ctx context.Context) httprouter.Hand return } offer := webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: string(body)} - pc, err := a.MediaManager.NewPeerConnection(ctx, mediaSigner.Streamer()) - if err != nil { - errors.WriteHTTPInternalServerError(w, "unable to create peer connection", err) - return - } - answer, err := a.MediaManager.WebRTCIngest(ctx, &offer, mediaSigner, pc, make(chan error, 1)) - if err != nil { - errors.WriteHTTPInternalServerError(w, fmt.Sprintf("error ingesting: %s", err.Error()), err) - return + + // Isolated WHIP: a detached worker owns the PeerConnection (and survives a + // main restart), returning the SDP answer over its frame socket. + // --isolated-ingest is forced off where unsupported (see runMain), so the + // flag alone gates this. + var answerSDP string + if a.CLI.IsolatedIngest { + answerSDP, err = a.MediaManager.WHIPIngestDetached(ctx, offer.SDP, mediaSigner) + if err != nil { + errors.WriteHTTPInternalServerError(w, fmt.Sprintf("error ingesting: %s", err.Error()), err) + return + } + } else { + pc, pcErr := a.MediaManager.NewPeerConnection(ctx, mediaSigner.Streamer()) + if pcErr != nil { + errors.WriteHTTPInternalServerError(w, "unable to create peer connection", pcErr) + return + } + answer, ingestErr := a.MediaManager.WebRTCIngest(ctx, &offer, mediaSigner, pc, make(chan error, 1)) + if ingestErr != nil { + errors.WriteHTTPInternalServerError(w, fmt.Sprintf("error ingesting: %s", ingestErr.Error()), ingestErr) + return + } + answerSDP = answer.SDP } host := r.Host if host == "" { @@ -127,7 +142,7 @@ func (a *StreamplaceAPI) HandleWebRTCIngest(ctx context.Context) httprouter.Hand log.Log(ctx, "location", "location", location) w.Header().Set("Location", location) w.WriteHeader(201) - if _, err := w.Write([]byte(answer.SDP)); err != nil { + if _, err := w.Write([]byte(answerSDP)); err != nil { log.Error(ctx, "error writing response", "error", err) } } diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index 36ce386f3..7f166c249 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -875,6 +875,12 @@ func makeIngestWorkerCommand(build *config.BuildFlags) *urfavecli.Command { return fmt.Errorf("ingest-worker: parse config: %w", err) } + // WHIP transport: the worker owns the PeerConnection (built from the + // offer in the config) and serves frames over the socket — no media fd. + if cfg.Transport == media.IngestTransportWHIP { + return media.ServeWHIPIngestWorkerSocket(ctx, cfg) + } + // Detach/reattach transport: serve frames over a unix socket with // buffered reconnect (survives a main restart) instead of the fd-4 pipe. // Media comes from the fd-passed ingest connection (InputFD) when main diff --git a/pkg/ingestframe/frame.go b/pkg/ingestframe/frame.go index 31f16d390..c8a746a36 100644 --- a/pkg/ingestframe/frame.go +++ b/pkg/ingestframe/frame.go @@ -33,6 +33,11 @@ const ( // Error carries a worker-side fatal error message (UTF-8). The worker emits // it just before exiting so main can log a cause, not a bare "worker exited". Error Type = 3 + // Answer carries an SDP answer (UTF-8). The WHIP worker owns the + // PeerConnection, so it generates the answer and emits it as the FIRST frame + // on the socket; main reads it and returns it to the WHIP client before + // consuming segments. Payload: the answer SDP. + Answer Type = 4 ) func (t Type) String() string { @@ -43,6 +48,8 @@ func (t Type) String() string { return "end" case Error: return "error" + case Answer: + return "answer" default: return fmt.Sprintf("unknown(%d)", uint8(t)) } @@ -104,6 +111,9 @@ func (fw *Writer) End() error { return fw.WriteFrame(End, nil) } // Error frames a fatal worker-side error message. func (fw *Writer) Error(msg string) error { return fw.WriteFrame(Error, []byte(msg)) } +// Answer frames the WHIP SDP answer (emitted first, before any segments). +func (fw *Writer) Answer(sdp string) error { return fw.WriteFrame(Answer, []byte(sdp)) } + // Reader decodes frames from an underlying stream. type Reader struct { r io.Reader diff --git a/pkg/ingestframe/frame_test.go b/pkg/ingestframe/frame_test.go index 70881fcf6..aa309f930 100644 --- a/pkg/ingestframe/frame_test.go +++ b/pkg/ingestframe/frame_test.go @@ -20,6 +20,7 @@ func TestRoundTrip(t *testing.T) { w := NewWriter(&buf) big := bytes.Repeat([]byte{0xAB}, 500_000) + require.NoError(t, w.Answer("v=0\r\no=- 1 1 IN IP4 0.0.0.0\r\n")) require.NoError(t, w.Segment([]byte("seg-one"))) require.NoError(t, w.Segment(nil)) // zero-length segment is legal require.NoError(t, w.Segment(big)) @@ -35,6 +36,7 @@ func TestRoundTrip(t *testing.T) { require.Equal(t, wantT, gotT) require.Equal(t, wantPayload, got) } + assertFrame(Answer, []byte("v=0\r\no=- 1 1 IN IP4 0.0.0.0\r\n")) assertFrame(Segment, []byte("seg-one")) assertFrame(Segment, nil) assertFrame(Segment, big) diff --git a/pkg/media/frame_server.go b/pkg/media/frame_server.go index 47585c77c..20b4965b6 100644 --- a/pkg/media/frame_server.go +++ b/pkg/media/frame_server.go @@ -87,6 +87,7 @@ func (s *frameServer) push(typ ingestframe.Type, payload []byte) { func (s *frameServer) Segment(seg []byte) error { s.push(ingestframe.Segment, seg); return nil } func (s *frameServer) End() error { s.push(ingestframe.End, nil); return nil } func (s *frameServer) Error(msg string) error { s.push(ingestframe.Error, []byte(msg)); return nil } +func (s *frameServer) Answer(sdp string) error { s.push(ingestframe.Answer, []byte(sdp)); return nil } // dropped reports how many buffered frames were discarded because the buffer // overflowed (main was disconnected longer than the buffer window). diff --git a/pkg/media/ingest_daemon.go b/pkg/media/ingest_daemon.go index c027f4bbc..455253814 100644 --- a/pkg/media/ingest_daemon.go +++ b/pkg/media/ingest_daemon.go @@ -13,6 +13,7 @@ import ( "time" "github.com/google/uuid" + "stream.place/streamplace/pkg/ingestframe" "stream.place/streamplace/pkg/log" ) @@ -50,8 +51,13 @@ func SpawnIngestWorkerDetached(cfg IngestWorkerConfig, media *os.File) (*os.Proc defer cfgW.Close() cmd := exec.Command(exe, "ingest-worker") - setDetached(cmd) // own session, survives a main restart (Linux) - cmd.ExtraFiles = []*os.File{cfgR, media} // → child fd 3 (config), fd 4 (media) + setDetached(cmd) // own session, survives a main restart (Linux) + // fd 3 = config; fd 4 = the fd-passed media connection (MKV/RTMP). WHIP owns + // its own PeerConnection, so it passes no media fd. + cmd.ExtraFiles = []*os.File{cfgR} + if media != nil { + cmd.ExtraFiles = append(cmd.ExtraFiles, media) + } cmd.Stderr = os.Stderr if err := cmd.Start(); err != nil { return nil, fmt.Errorf("spawn ingest worker: %w", err) @@ -164,6 +170,108 @@ func (mm *MediaManager) MKVIngestDetached(ctx context.Context, conn net.Conn, pr return err } +// whipAnswerTimeout bounds how long main waits for the worker to produce the SDP +// answer (worker startup + ICE gathering) before giving up on the WHIP request. +const whipAnswerTimeout = 20 * time.Second + +// dialWorkerSocket connects to a worker's frame socket, retrying until it's up or +// ctx is done (a freshly-spawned worker takes a moment to start listening). +func dialWorkerSocket(ctx context.Context, socketPath string) (net.Conn, error) { + for { + conn, err := net.Dial("unix", socketPath) + if err == nil { + return conn, nil + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(ingestReconnectBackoff): + } + } +} + +// readWHIPAnswer reads frames on conn until the worker's Answer frame and returns +// its SDP. An Error/End/EOF before the answer is a setup failure. +func readWHIPAnswer(conn net.Conn) (string, error) { + fr := ingestframe.NewReader(conn) + for { + typ, payload, err := fr.ReadFrame() + if err != nil { + return "", fmt.Errorf("read whip answer: %w", err) + } + switch typ { + case ingestframe.Answer: + return string(payload), nil + case ingestframe.Error: + return "", fmt.Errorf("whip worker error: %s", payload) + case ingestframe.End: + return "", fmt.Errorf("whip worker ended before sending an answer") + } + // A Segment before the Answer shouldn't happen; ignore it defensively. + } +} + +// WHIPIngestDetached is the WHIP zero-downtime entry. Main has authed the WHIP +// request; this spawns a DETACHED worker that owns the PeerConnection (built from +// offerSDP, binding its own UDP sockets) and serves signed segments over a +// per-session socket. It reads the worker's SDP answer (the first frame) to +// return to the client, then consumes segments into ValidateMP4 in the +// background with reconnect. Because the worker owns the WebRTC session and is +// detached, both the session and its buffered output survive a main restart (the +// restarted main reconnects via discovery). +func (mm *MediaManager) WHIPIngestDetached(ctx context.Context, offerSDP string, ms MediaSigner) (string, error) { + cfg, err := mm.buildWorkerConfig(ctx, ms) + if err != nil { + return "", err + } + dir, err := mm.ingestWorkerSocketDir() + if err != nil { + return "", err + } + cfg.SocketPath = filepath.Join(dir, uuid.NewString()+".sock") + cfg.Transport = IngestTransportWHIP + cfg.OfferSDP = offerSDP + + proc, err := SpawnIngestWorkerDetached(cfg, nil) // worker owns the PeerConnection + if err != nil { + return "", fmt.Errorf("spawn detached whip worker: %w", err) + } + + // Connect + read the SDP answer (the worker's first frame), bounded so a + // wedged setup can't hang the WHIP client. + answerCtx, answerCancel := context.WithTimeout(ctx, whipAnswerTimeout) + defer answerCancel() + conn, err := dialWorkerSocket(answerCtx, cfg.SocketPath) + if err != nil { + _ = proc.Kill() + return "", fmt.Errorf("connect to whip worker: %w", err) + } + if dl, ok := answerCtx.Deadline(); ok { + _ = conn.SetReadDeadline(dl) + } + answer, err := readWHIPAnswer(conn) + if err != nil { + conn.Close() + _ = proc.Kill() + return "", err + } + _ = conn.SetReadDeadline(time.Time{}) // clear; streaming has no deadline + + // Consume the signed segments in the background; the HTTP handler returns the + // answer now and the WebRTC media establishes directly to the worker. + go func() { + sawEnd, _ := mm.consumeWorkerFrames(ctx, conn, ms.Streamer(), mm.validateSegment(ctx), nil) + conn.Close() + if !sawEnd && ctx.Err() == nil { + // Connection dropped but the detached worker lives on — reconnect and + // drain its buffer. + _ = mm.ConsumeWorkerSocket(ctx, cfg.SocketPath, ms.Streamer(), mm.validateSegment(ctx)) + } + go func() { _, _ = proc.Wait() }() + }() + return answer, nil +} + // ResumeDetachedWorkers reconnects to any ingest workers still running from // before a main restart and resumes consuming their frames (draining whatever // they buffered while main was down). Intended to run once at main startup. diff --git a/pkg/media/ingest_worker.go b/pkg/media/ingest_worker.go index 9a9321bc9..5941b691a 100644 --- a/pkg/media/ingest_worker.go +++ b/pkg/media/ingest_worker.go @@ -61,8 +61,20 @@ type IngestWorkerConfig struct { // stdin / raw fd input. Prebuf []byte `json:"prebuf,omitempty"` Chunked bool `json:"chunked,omitempty"` + + // Transport selects the worker's ingest source: "" / "mkv" reads MKV media + // (stdin or InputFD); "whip" makes the worker own the WebRTC PeerConnection, + // built from OfferSDP — no media fd to pass. + Transport string `json:"transport,omitempty"` + // OfferSDP is the WHIP client's SDP offer (transport "whip"). The worker + // generates the answer and emits it as the first frame (ingestframe.Answer) + // so main can return it to the client before consuming segments. + OfferSDP string `json:"offer_sdp,omitempty"` } +// IngestTransportWHIP is the cfg.Transport value selecting the WHIP worker. +const IngestTransportWHIP = "whip" + // WorkerInput reconstructs the raw media stream the gst pipeline reads from the // fd-passed push connection: prepend any bytes main already read past the headers // (Prebuf), then de-chunk if the push used chunked transfer-encoding. For stdin @@ -78,28 +90,11 @@ func WorkerInput(cfg IngestWorkerConfig, raw io.Reader) io.Reader { return r } -// RunMKVIngestWorker is the body of the `ingest-worker` subcommand. It reads an -// MKV stream from stdin, runs the same demux + Opus re-encode + muxl-sign -// pipeline as the in-process MKVIngest, and emits each signed canonical .m4s -// segment to frames; the main process reads those frames and runs ValidateMP4 -// over each, exactly as if onSegment had called it directly. -// -// It returns when the stream ends cleanly (EOS) or the pipeline errors. The -// caller frames End or Error accordingly. All segment frames are guaranteed -// flushed before it returns, so a trailing End can never race ahead of the last -// Segment. -func RunMKVIngestWorker(ctx context.Context, cfg IngestWorkerConfig, stdin io.Reader, frames FrameWriter) error { - gstinit.InitGST() - ctx, cancel := context.WithCancel(ctx) - defer cancel() - - // Minimal manager: just the broadcaster identity the transcode completion - // (finishTranscodedSegment) stamps into the node-signed AAC track. - mm := &MediaManager{cli: &config.CLI{BroadcasterHost: cfg.BroadcasterHost}} - - // The worker signs everything itself: forward the streamer key PEM + cert + - // prebuilt manifest straight to muxl-sign. No MediaSigner / model / DB needed. - signStream := func(ctx context.Context, input io.Reader, eventCh chan *muxl.MuxlEvent) error { +// workerSignStream returns the streaming muxl signer a worker uses: it forwards +// the streamer key PEM + cert + prebuilt manifest straight to muxl-sign, no +// MediaSigner / model / DB needed. Shared by the MKV and WHIP workers. +func workerSignStream(cfg IngestWorkerConfig) SignSegmentStreamFunc { + return func(ctx context.Context, input io.Reader, eventCh chan *muxl.MuxlEvent) error { fetchManifest := func() ([]byte, error) { return cfg.Manifest, nil } return muxl.RunMuxlSignSegment(ctx, input, muxl.SignerInput{ CertPEM: cfg.CertPEM, @@ -108,16 +103,19 @@ func RunMKVIngestWorker(ctx context.Context, cfg IngestWorkerConfig, stdin io.Re WrapperManifestFn: fetchManifest, }, nil, nil, eventCh) } +} - // With a node transcode key, the worker completes each single-codec source - // segment to dual-codec itself: feed the signed source segment into a - // per-stream transcoder running in THIS process; its completion callback - // frames the finished dual-codec segment. The transcoder runs on a - // non-cancellable context so draining the signer (cancel, below) can't kill it - // before its ~1-GoP tail is flushed by Close. One process == one session, so - // the per-DID transcoder-reuse hazard simply can't arise here. +// workerSegmentSink returns the onSegment handler a worker hands to +// muxlSignSegmentElem, plus a flush to call once the signer has drained. With a +// node transcode key it completes each single-codec source segment to dual-codec +// via an in-process transcoder (its completion callback frames the finished +// segment); flush Closes that transcoder so its ~1-GoP tail is framed before the +// worker exits. The transcoder runs on a non-cancellable context so draining the +// signer can't kill it early. One process == one session, so the per-DID +// transcoder-reuse hazard can't arise. Shared by the MKV and WHIP workers. +func (mm *MediaManager) workerSegmentSink(ctx context.Context, cfg IngestWorkerConfig, frames FrameWriter) (onSegment func(context.Context, []byte) error, flush func()) { var transcoder *streamTranscoder - onSegment := func(_ context.Context, segment []byte) error { + onSegment = func(_ context.Context, segment []byte) error { if len(cfg.NodeKeyPEM) == 0 { return frames.Segment(segment) // no node signer → single-codec } @@ -135,8 +133,37 @@ func RunMKVIngestWorker(ctx context.Context, cfg IngestWorkerConfig, stdin io.Re } return transcoder.Feed(segment, nil) } + flush = func() { + if transcoder != nil { + if cerr := transcoder.Close(); cerr != nil { + log.Error(ctx, "ingest worker: transcoder close", "error", cerr) + } + } + } + return onSegment, flush +} - signerElem, done, err := muxlSignSegmentElem(ctx, mm.cli, signStream, onSegment) +// RunMKVIngestWorker is the body of the `ingest-worker` subcommand. It reads an +// MKV stream from stdin, runs the same demux + Opus re-encode + muxl-sign +// pipeline as the in-process MKVIngest, and emits each signed canonical .m4s +// segment to frames; the main process reads those frames and runs ValidateMP4 +// over each, exactly as if onSegment had called it directly. +// +// It returns when the stream ends cleanly (EOS) or the pipeline errors. The +// caller frames End or Error accordingly. All segment frames are guaranteed +// flushed before it returns, so a trailing End can never race ahead of the last +// Segment. +func RunMKVIngestWorker(ctx context.Context, cfg IngestWorkerConfig, stdin io.Reader, frames FrameWriter) error { + gstinit.InitGST() + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + // Minimal manager: just the broadcaster identity the transcode completion + // (finishTranscodedSegment) stamps into the node-signed AAC track. + mm := &MediaManager{cli: &config.CLI{BroadcasterHost: cfg.BroadcasterHost}} + onSegment, flush := mm.workerSegmentSink(ctx, cfg, frames) + + signerElem, done, err := muxlSignSegmentElem(ctx, mm.cli, workerSignStream(cfg), onSegment) if err != nil { return fmt.Errorf("build signer element: %w", err) } @@ -159,18 +186,12 @@ func RunMKVIngestWorker(ctx context.Context, cfg IngestWorkerConfig, stdin io.Re } }() - // Wait for the pipeline to finish (EOS or error), then drain the signer: - // cancelling unblocks the signer's input pipe so it flushes the final GoP, and - // <-done guarantees every source segment has been fed. Then flush the - // transcoder's tail so the last dual-codec completions are framed before we - // return (the caller's End can't race ahead of them). + // Pipeline done (EOS/error) → drain the signer (cancel flushes the final GoP; + // <-done means every source segment has been fed) → flush the transcoder tail + // so the last dual-codec completions are framed before we return. pipeErr := <-busErr cancel() <-done - if transcoder != nil { - if cerr := transcoder.Close(); cerr != nil { - log.Error(ctx, "ingest worker: transcoder close", "error", cerr) - } - } + flush() return pipeErr } diff --git a/pkg/media/media.go b/pkg/media/media.go index 258f4baed..453598171 100644 --- a/pkg/media/media.go +++ b/pkg/media/media.go @@ -12,8 +12,6 @@ import ( "sync/atomic" "github.com/google/uuid" - "github.com/pion/interceptor" - "github.com/pion/interceptor/pkg/intervalpli" "github.com/pion/webrtc/v4" "go.opentelemetry.io/otel" "stream.place/streamplace/pkg/aqtime" @@ -111,52 +109,10 @@ func MakeMediaManager(ctx context.Context, cli *config.CLI, signer crypto.Signer return nil, fmt.Errorf("error in gstreamer self-test: %w", err) } - m := &webrtc.MediaEngine{} - // Create a InterceptorRegistry. This is the user configurable RTP/RTCP Pipeline. - // This provides NACKs, RTCP Reports and other features. If you use `webrtc.NewPeerConnection` - // this is enabled by default. If you are manually managing You MUST create a InterceptorRegistry - // for each PeerConnection. - i := &interceptor.Registry{} - - // Register a intervalpli factory - // This interceptor sends a PLI every 3 seconds. A PLI causes a video keyframe to be generated by the sender. - // This makes our video seekable and more error resilent, but at a cost of lower picture quality and higher bitrates - // A real world application should process incoming RTCP packets from viewers and forward them to senders - intervalPliFactory, err := intervalpli.NewReceiverInterceptor() + api, config, err := newWebRTCAPI() if err != nil { - return nil, fmt.Errorf("failed to create intervalpli factory: %w", err) - } - i.Add(intervalPliFactory) - - if err := m.RegisterCodec(webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, Channels: 0, SDPFmtpLine: "", RTCPFeedback: nil}, - PayloadType: 102, - }, webrtc.RTPCodecTypeVideo); err != nil { - return nil, err - } - if err := m.RegisterCodec(webrtc.RTPCodecParameters{ - RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus, ClockRate: 48000, Channels: 0, SDPFmtpLine: "", RTCPFeedback: nil}, - PayloadType: 111, - }, webrtc.RTPCodecTypeAudio); err != nil { return nil, err } - - // Use the default set of Interceptors - if err = webrtc.RegisterDefaultInterceptors(m, i); err != nil { - return nil, fmt.Errorf("failed to register default interceptors: %w", err) - } - - // Create the API object with the MediaEngine - api := webrtc.NewAPI(webrtc.WithMediaEngine(m), webrtc.WithInterceptorRegistry(i)) - - // Prepare the configuration - config := webrtc.Configuration{ - ICEServers: []webrtc.ICEServer{ - { - URLs: []string{"stun:stun.l.google.com:19302"}, - }, - }, - } return &MediaManager{ cli: cli, liveWindows: map[string]*livehls.Writer{}, diff --git a/pkg/media/webrtc_api.go b/pkg/media/webrtc_api.go new file mode 100644 index 000000000..e8bc019ed --- /dev/null +++ b/pkg/media/webrtc_api.go @@ -0,0 +1,49 @@ +package media + +import ( + "fmt" + + "github.com/pion/interceptor" + "github.com/pion/interceptor/pkg/intervalpli" + "github.com/pion/webrtc/v4" +) + +// newWebRTCAPI builds the pion API + configuration Streamplace uses for WebRTC +// ingest (H264 video + Opus audio, default interceptors plus an interval PLI so +// the publisher keeps sending keyframes). Shared by MakeMediaManager and the +// isolated WHIP worker — the worker builds its own API since it owns the +// PeerConnection in its own process. +func newWebRTCAPI() (*webrtc.API, webrtc.Configuration, error) { + m := &webrtc.MediaEngine{} + i := &interceptor.Registry{} + + intervalPliFactory, err := intervalpli.NewReceiverInterceptor() + if err != nil { + return nil, webrtc.Configuration{}, fmt.Errorf("failed to create intervalpli factory: %w", err) + } + i.Add(intervalPliFactory) + + if err := m.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000}, + PayloadType: 102, + }, webrtc.RTPCodecTypeVideo); err != nil { + return nil, webrtc.Configuration{}, err + } + if err := m.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus, ClockRate: 48000}, + PayloadType: 111, + }, webrtc.RTPCodecTypeAudio); err != nil { + return nil, webrtc.Configuration{}, err + } + if err := webrtc.RegisterDefaultInterceptors(m, i); err != nil { + return nil, webrtc.Configuration{}, fmt.Errorf("failed to register default interceptors: %w", err) + } + + api := webrtc.NewAPI(webrtc.WithMediaEngine(m), webrtc.WithInterceptorRegistry(i)) + config := webrtc.Configuration{ + ICEServers: []webrtc.ICEServer{ + {URLs: []string{"stun:stun.l.google.com:19302"}}, + }, + } + return api, config, nil +} diff --git a/pkg/media/webrtc_ingest.go b/pkg/media/webrtc_ingest.go index e4fd50bc6..70b6b2bfd 100644 --- a/pkg/media/webrtc_ingest.go +++ b/pkg/media/webrtc_ingest.go @@ -15,19 +15,36 @@ import ( "stream.place/streamplace/pkg/rtcrec" ) -// This function remains in scope for the duration of a single users' playback +// WebRTCIngest is the in-process WHIP entry: it builds the signing element via +// SegmentAndSignElem (→ ValidateMP4) and runs the shared ingest pipeline. Stays +// in scope for the duration of a single stream. func (mm *MediaManager) WebRTCIngest(ctx context.Context, offer *webrtc.SessionDescription, signer MediaSigner, peerConnection rtcrec.PeerConnection, done chan error) (*webrtc.SessionDescription, error) { uu, err := uuid.NewV7() if err != nil { return nil, err } - ctx = log.WithLogValues(ctx, "webrtcID", uu.String(), "mediafunc", "WebRTCIngest", "streamer", signer.Streamer()) + ctx, cancel := context.WithCancel(ctx) + signerElem, err := mm.SegmentAndSignElem(ctx, signer) + if err != nil { + cancel() + return nil, fmt.Errorf("failed create signer element: %w", err) + } + return mm.webRTCIngestPipeline(ctx, cancel, offer, peerConnection, signerElem, signer, done) +} +// webRTCIngestPipeline runs WebRTC ingest over a pre-built signer element: +// depay/parse the incoming RTP into the muxl signing bin, answer the offer, and +// stream in the background. The in-process path passes a SegmentAndSignElem (→ +// ValidateMP4) and the streamer's signer (for key revocation); the isolated WHIP +// worker passes a muxlSignSegmentElem wired to its frame socket and a nil +// keyRevSigner. The cancellable ctx and signerElem are built by the caller (the +// signer element's goroutines are tied to ctx). +func (mm *MediaManager) webRTCIngestPipeline(ctx context.Context, cancel context.CancelFunc, offer *webrtc.SessionDescription, peerConnection rtcrec.PeerConnection, signerElem *gst.Element, keyRevSigner MediaSigner, done chan error) (*webrtc.SessionDescription, error) { // Allow us to receive 1 audio track, and 1 video track - if _, err = peerConnection.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio); err != nil { + if _, err := peerConnection.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio); err != nil { return nil, fmt.Errorf("failed to add audio transceiver: %w", err) - } else if _, err = peerConnection.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo); err != nil { + } else if _, err := peerConnection.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo); err != nil { return nil, fmt.Errorf("failed to add video transceiver: %w", err) } @@ -92,12 +109,7 @@ func (mm *MediaManager) WebRTCIngest(ctx context.Context, offer *webrtc.SessionD // Create channel that is blocked until ICE Gathering is complete gatherComplete := rtcrec.GatheringCompletePromise(peerConnection) - ctx, cancel := context.WithCancel(ctx) - signerElem, err := mm.SegmentAndSignElem(ctx, signer) - if err != nil { - cancel() - return nil, fmt.Errorf("failed create signer element: %w", err) - } + // cancel + signerElem are provided by the caller. err = pipeline.Add(signerElem) if err != nil { cancel() @@ -153,8 +165,11 @@ func (mm *MediaManager) WebRTCIngest(ctx context.Context, offer *webrtc.SessionD } }() - // subscription to bus messages for key revocation - go mm.HandleKeyRevocation(ctx, signer, pipeline) + // subscription to bus messages for key revocation (in-process only; the + // isolated worker has no model-backed signer to revoke against) + if keyRevSigner != nil { + go mm.HandleKeyRevocation(ctx, keyRevSigner, pipeline) + } go func() { <-ctx.Done() diff --git a/pkg/media/whip_worker.go b/pkg/media/whip_worker.go new file mode 100644 index 000000000..ffb88592f --- /dev/null +++ b/pkg/media/whip_worker.go @@ -0,0 +1,105 @@ +package media + +import ( + "context" + "fmt" + "net" + "os" + + "github.com/pion/webrtc/v4" + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/gstinit" + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/rtcrec" +) + +// ServeWHIPIngestWorkerSocket is the WHIP counterpart of +// ServeMKVIngestWorkerSocket. Unlike MKV there's no socket/fd to pass in: the +// worker OWNS the PeerConnection, so it creates it from cfg.OfferSDP (binding its +// own UDP sockets), generates the SDP answer, and emits it as the FIRST frame on +// the unix socket — main reads that Answer frame and returns it to the WHIP +// client, then keeps reading the signed dual-codec segments. The worker is +// detached, so the WebRTC session (and the buffered segment stream) survive a +// main restart. +func ServeWHIPIngestWorkerSocket(ctx context.Context, cfg IngestWorkerConfig) error { + if cfg.SocketPath == "" { + return fmt.Errorf("ServeWHIPIngestWorkerSocket: empty socket path") + } + if cfg.OfferSDP == "" { + return fmt.Errorf("ServeWHIPIngestWorkerSocket: empty offer") + } + gstinit.InitGST() + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + _ = os.Remove(cfg.SocketPath) // clear any stale socket from a prior worker + ln, err := net.Listen("unix", cfg.SocketPath) + if err != nil { + return fmt.Errorf("listen %s: %w", cfg.SocketPath, err) + } + defer func() { + ln.Close() + _ = os.Remove(cfg.SocketPath) + }() + + srv := newFrameServer(workerFrameBuffer) + go serveFrameSocket(ctx, ln, srv) + + // finish flushes the trailing End/Error, waits for main to drain the buffer + // (incl. the Answer), then closes the connection for a clean EOF. + finish := func(runErr error) error { + if runErr != nil { + _ = srv.Error(runErr.Error()) + } else { + _ = srv.End() + } + srv.waitDrained(ctx, workerDrainGrace) + srv.closeConn() + return runErr + } + + mm := &MediaManager{cli: &config.CLI{BroadcasterHost: cfg.BroadcasterHost}} + + // The worker owns the PeerConnection (its own UDP sockets), built with the + // same codec/interceptor setup as the in-process server. No recording here — + // the worker has no model-backed settings. + api, webrtcConfig, err := newWebRTCAPI() + if err != nil { + return finish(fmt.Errorf("webrtc api: %w", err)) + } + pionpc, err := api.NewPeerConnection(webrtcConfig) + if err != nil { + return finish(fmt.Errorf("peer connection: %w", err)) + } + pc, err := rtcrec.NewRecordingPeerConnection(ctx, *mm.cli, cfg.StreamerDID, pionpc, false) + if err != nil { + return finish(fmt.Errorf("peer connection wrapper: %w", err)) + } + + onSegment, flush := mm.workerSegmentSink(ctx, cfg, srv) + signerElem, signerDone, err := muxlSignSegmentElem(ctx, mm.cli, workerSignStream(cfg), onSegment) + if err != nil { + return finish(fmt.Errorf("build signer element: %w", err)) + } + + offer := &webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: cfg.OfferSDP} + streamDone := make(chan error, 1) + answer, err := mm.webRTCIngestPipeline(ctx, cancel, offer, pc, signerElem, nil, streamDone) + if err != nil { + return finish(fmt.Errorf("webrtc ingest: %w", err)) + } + + // Hand the answer back to main FIRST; the segment frames stream behind it. + if aerr := srv.Answer(answer.SDP); aerr != nil { + log.Error(ctx, "whip worker: frame answer", "error", aerr) + } + + // Streaming runs until the peer disconnects / errors; webRTCIngestPipeline + // cancels ctx then, which drains the signer. Wait for that, flush the + // transcoder tail, then finish. + streamErr := <-streamDone + cancel() + <-signerDone + flush() + return finish(streamErr) +} diff --git a/pkg/media/whip_worker_test.go b/pkg/media/whip_worker_test.go new file mode 100644 index 000000000..b0ac69fb5 --- /dev/null +++ b/pkg/media/whip_worker_test.go @@ -0,0 +1,237 @@ +package media + +import ( + "bytes" + "context" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/go-gst/go-gst/gst" + "github.com/go-gst/go-gst/gst/app" + pionmedia "github.com/pion/webrtc/v4/pkg/media" + + "github.com/pion/webrtc/v4" + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/crypto/signers" + "stream.place/streamplace/pkg/ingestframe" + "stream.place/streamplace/pkg/muxl" +) + +// whipClientOffer builds a WHIP-style SDP offer (H264 video + Opus audio tracks), +// the way a real WHIP client does, returning the client PC, its tracks, and the +// offer. +func whipClientOffer(t *testing.T) (*webrtc.PeerConnection, *webrtc.TrackLocalStaticSample, *webrtc.TrackLocalStaticSample, webrtc.SessionDescription) { + t.Helper() + pc, err := webrtc.NewPeerConnection(webrtc.Configuration{}) + require.NoError(t, err) + videoTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264}, "video", "pion") + require.NoError(t, err) + if _, err = pc.AddTrack(videoTrack); err != nil { + t.Fatal(err) + } + audioTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus}, "audio", "pion") + require.NoError(t, err) + if _, err = pc.AddTrack(audioTrack); err != nil { + t.Fatal(err) + } + offer, err := pc.CreateOffer(nil) + require.NoError(t, err) + require.NoError(t, pc.SetLocalDescription(offer)) + return pc, videoTrack, audioTrack, offer +} + +// TestWHIPWorkerAnswersOffer verifies the WHIP worker's answer back-channel: from +// an offer it builds the PeerConnection, generates the SDP answer, and emits it +// as the FIRST frame on its socket — the synchronous reply main returns to the +// WHIP client. (Media flow → signed segments rides the same webRTCIngestPipeline +// the in-process path uses, plus the transcoder/frame machinery the MKV tests +// already cover.) +func TestWHIPWorkerAnswersOffer(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + 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) + + clientPC, _, _, offer := whipClientOffer(t) + defer clientPC.Close() + + sock := filepath.Join(t.TempDir(), "whip.sock") + cfg := IngestWorkerConfig{ + StreamerDID: ms.Streamer(), + KeyPEM: keyPEM, + CertPEM: ms.Cert, + Manifest: manifest, + NodeCertPEM: ms.Cert, + NodeKeyPEM: keyPEM, + BroadcasterHost: "test.example.com", + SocketPath: sock, + Transport: IngestTransportWHIP, + OfferSDP: offer.SDP, + } + + serveDone := make(chan error, 1) + go func() { serveDone <- ServeWHIPIngestWorkerSocket(ctx, cfg) }() + + // Connect to the worker socket (retry until up) and read the Answer frame. + dctx, dcancel := context.WithTimeout(ctx, 20*time.Second) + defer dcancel() + conn, derr := dialWorkerSocket(dctx, sock) + require.NoError(t, derr) + defer conn.Close() + + answerSDP, rerr := readWHIPAnswer(conn) + require.NoError(t, rerr, "worker emits an SDP answer as its first frame") + require.Contains(t, answerSDP, "v=0", "valid SDP answer") + + // The answer must apply cleanly as the client's remote description — i.e. it's + // a real, negotiated answer to the offer. + require.NoError(t, clientPC.SetRemoteDescription(webrtc.SessionDescription{ + Type: webrtc.SDPTypeAnswer, SDP: answerSDP, + }), "answer applies as the client's remote description") + + t.Logf("whip worker produced a %d-byte SDP answer", len(answerSDP)) + + cancel() + select { + case <-serveDone: + case <-time.After(25 * time.Second): + t.Fatal("worker did not exit after cancel") + } +} + +// produceWHIPMedia streams synthetic H264 + Opus into the WHIP client's tracks +// via a gst encode pipeline until ctx is cancelled — i.e. a real WHIP publisher. +func produceWHIPMedia(t *testing.T, ctx context.Context, video, audio *webrtc.TrackLocalStaticSample) { + t.Helper() + desc := strings.Join([]string{ + "videotestsrc is-live=true ! video/x-raw,width=320,height=240,framerate=30/1 ! x264enc key-int-max=15 tune=zerolatency speed-preset=ultrafast ! h264parse ! video/x-h264,stream-format=byte-stream,alignment=au ! appsink name=vsink", + "audiotestsrc is-live=true ! audioconvert ! audioresample ! opusenc ! opusparse ! appsink name=asink", + }, "\n") + pipeline, err := gst.NewPipelineFromString(desc) + require.NoError(t, err) + + pump := func(name string, track *webrtc.TrackLocalStaticSample, dur time.Duration) { + ele, gerr := pipeline.GetElementByName(name) + require.NoError(t, gerr) + app.SinkFromElement(ele).SetCallbacks(&app.SinkCallbacks{ + NewSampleFunc: func(sink *app.Sink) gst.FlowReturn { + sample := sink.PullSample() + if sample == nil { + return gst.FlowEOS + } + buf := sample.GetBuffer() + data := buf.Map(gst.MapRead).Bytes() + buf.Unmap() + if werr := track.WriteSample(pionmedia.Sample{Data: data, Duration: dur}); werr != nil { + return gst.FlowError + } + return gst.FlowOK + }, + }) + } + pump("vsink", video, time.Second/30) + pump("asink", audio, 20*time.Millisecond) + + go func() { + <-ctx.Done() + _ = pipeline.SetState(gst.StateNull) + }() + require.NoError(t, pipeline.SetState(gst.StatePlaying)) +} + +// TestWHIPWorkerLoopback is the full WHIP media path: a pion client offers, +// connects to the worker (which owns the PeerConnection), and streams real +// H264+Opus RTP; the worker must mux+sign+transcode it and serve a valid signed +// dual-codec segment over its socket — the WHIP parity of the MKV worker e2e +// test. +func TestWHIPWorkerLoopback(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + 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) + + clientPC, videoTrack, audioTrack, offer := whipClientOffer(t) + defer clientPC.Close() + + connected := make(chan struct{}) + var once sync.Once + clientPC.OnConnectionStateChange(func(s webrtc.PeerConnectionState) { + if s == webrtc.PeerConnectionStateConnected { + once.Do(func() { close(connected) }) + } + }) + + sock := filepath.Join(t.TempDir(), "whip.sock") + cfg := IngestWorkerConfig{ + StreamerDID: ms.Streamer(), + KeyPEM: keyPEM, + CertPEM: ms.Cert, + Manifest: manifest, + NodeCertPEM: ms.Cert, + NodeKeyPEM: keyPEM, + BroadcasterHost: "test.example.com", + SocketPath: sock, + Transport: IngestTransportWHIP, + OfferSDP: offer.SDP, + } + serveDone := make(chan error, 1) + go func() { serveDone <- ServeWHIPIngestWorkerSocket(ctx, cfg) }() + + dctx, dcancel := context.WithTimeout(ctx, 20*time.Second) + defer dcancel() + conn, derr := dialWorkerSocket(dctx, sock) + require.NoError(t, derr) + defer conn.Close() + + answerSDP, rerr := readWHIPAnswer(conn) + require.NoError(t, rerr) + require.NoError(t, clientPC.SetRemoteDescription(webrtc.SessionDescription{Type: webrtc.SDPTypeAnswer, SDP: answerSDP})) + + select { + case <-connected: + case <-time.After(20 * time.Second): + t.Fatal("client PC did not connect to the worker") + } + + produceWHIPMedia(t, ctx, videoTrack, audioTrack) + + // Read signed segments; require at least one valid dual-codec one. + _ = conn.SetReadDeadline(time.Now().Add(45 * time.Second)) + r := ingestframe.NewReader(conn) + var segs int + for segs == 0 { + typ, payload, ferr := r.ReadFrame() + require.NoError(t, ferr, "reading worker frames") + if typ != ingestframe.Segment { + continue + } + out, verr := muxl.RunMuxlVerify(ctx, bytes.NewReader(payload)) + require.NoError(t, verr) + require.NotContains(t, out, `"validation_state":"Invalid"`, "segment must validate") + segs++ + } + _ = conn.SetReadDeadline(time.Time{}) + t.Logf("whip worker produced %d signed segment(s) from real RTP media", segs) + + // Close the connection before tearing down: in production main's connection + // breaks on shutdown, which detaches the worker's frame server so it drains + // into its buffer instead of blocking on an unread socket. + conn.Close() + cancel() + select { + case <-serveDone: + case <-time.After(25 * time.Second): + t.Fatal("worker did not exit after cancel") + } +}