From 48d4b99105a5dc4f2c8cc50aed87a32e4ea24ad0 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Fri, 9 Oct 2026 18:21:55 -0700 Subject: [PATCH] Ingest the OBS plugin's fragmented MP4 push, signed as-is by muxl POST /api/ingest/client-configuration answers the plugin's GoLiveApi-shaped configuration request: a bad stream key comes back as a status error the plugin shows, a good one gets the FMP4 ingest endpoint. No encoder configurations yet, so the plugin keeps OBS's own encoders. POST /api/ingest/fmp4/:key takes the push through the same path as the /live route (ban check, then a detached worker or in-process ingest), but with a new fmp4-direct transport that hands the body straight to muxl's per-segment signer instead of the GStreamer demux/remux graph. That keeps every track the client sends, which multitrack audio and video will need. The signer's event drain is shared between both paths now, and segment tracks are concatenated in numeric track-id order (the string sort put track 10 before track 2). `streamplace e2e --ingest none` starts the harness without its WHIP loop and prints STREAM_KEY, for pushing from OBS. --- pkg/api/api.go | 2 + pkg/api/api_internal.go | 49 +-------- pkg/api/fmp4_ingest.go | 164 +++++++++++++++++++++++++++++ pkg/cmd/e2e.go | 21 +++- pkg/media/dynamic_manifest_test.go | 2 +- pkg/media/ingest_daemon.go | 6 +- pkg/media/ingest_worker.go | 25 ++++- pkg/media/ingest_worker_test.go | 52 +++++++++ pkg/media/mist_pull.go | 4 +- pkg/media/mp4_ingest.go | 19 +++- pkg/media/muxl_segment.go | 98 +++++++++-------- pkg/media/segmenter.go | 9 +- pkg/media/transcode_stream_test.go | 2 +- pkg/media/validate_bare_test.go | 2 +- 14 files changed, 342 insertions(+), 113 deletions(-) create mode 100644 pkg/api/fmp4_ingest.go diff --git a/pkg/api/api.go b/pkg/api/api.go index 2fb84a486..f7f979d11 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -226,6 +226,8 @@ func (a *StreamplaceAPI) Handler(ctx context.Context) (http.Handler, error) { addHandle(apiRouter, "POST", "/api/playback/:user/webrtc", a.HandleWebRTCPlayback(ctx)) addHandle(apiRouter, "POST", "/api/ingest/webrtc", a.HandleWebRTCIngest(ctx)) addHandle(apiRouter, "POST", "/api/ingest/webrtc/:key", a.HandleWebRTCIngest(ctx)) + addFunc(apiRouter, "POST", "/api/ingest/client-configuration", a.HandleClientConfiguration(ctx)) + addHandle(apiRouter, "POST", "/api/ingest/fmp4/:key", a.HandleFMP4Ingest(ctx)) addHandle(apiRouter, "POST", "/api/player-event", a.HandlePlayerEvent(ctx)) addHandle(apiRouter, "GET", "/api/chat/:repoDID", a.HandleChat(ctx)) addHandle(apiRouter, "GET", "/api/websocket/:repoDID", a.HandleWebsocket(ctx)) diff --git a/pkg/api/api_internal.go b/pkg/api/api_internal.go index 0a6e3400b..483df7053 100644 --- a/pkg/api/api_internal.go +++ b/pkg/api/api_internal.go @@ -201,54 +201,7 @@ func (a *StreamplaceAPI) InternalHandler(ctx context.Context) (http.Handler, err } } - reqCtx = log.WithLogValues(reqCtx, "streamer", mediaSigner.Streamer()) - - err = a.checkBanned(reqCtx, mediaSigner.Streamer()) - if err != nil { - errors.WriteHTTPUnauthorized(w, err.Error(), err) - return - } - - if a.CLI.IsolatedIngest { - // Zero-downtime path: hijack the authed push connection and hand it to a - // DETACHED worker that owns the connection (so it survives a main - // restart) and serves signed segments back over its socket. - if hj, ok := w.(http.Hijacker); ok { - conn, bufrw, herr := hj.Hijack() - if herr != nil { - log.Error(reqCtx, "ingest hijack failed", "error", herr) - return - } - var prebuf []byte - if n := bufrw.Reader.Buffered(); n > 0 { - prebuf = make([]byte, n) - _, _ = io.ReadFull(bufrw.Reader, prebuf) - } - chunked := len(httpReq.TransferEncoding) > 0 && httpReq.TransferEncoding[0] == "chunked" - if derr := a.MediaManager.MP4IngestDetached(reqCtx, conn, prebuf, chunked, mediaSigner); derr != nil { - log.Log(reqCtx, "isolated stream ended", "error", derr) - } - return // connection hijacked; the HTTP response is ours now - } - // The isolated path needs a hijackable HTTP/1.1 connection (which the - // real /live clients — the `streamplace live` CLI and tests pushing - // over localhost — always are; the Mist ingest itself now arrives via - // MistPullIngest, not this route). We don't support a non-hijack - // fallback: it couldn't receive mid-stream manifest updates and would - // stay stuck pre-live, so refuse. Such a client can use WHIP instead. - log.Error(reqCtx, "isolated ingest requires a hijackable HTTP/1.1 connection; refusing push") - errors.WriteHTTPInternalServerError(w, "isolated ingest requires a hijackable HTTP/1.1 connection; use WHIP", fmt.Errorf("connection is not hijackable")) - return - } else { - err = a.MediaManager.MP4Ingest(reqCtx, r, mediaSigner) - } - - if err != nil { - log.Log(reqCtx, "stream error", "error", err) - errors.WriteHTTPInternalServerError(w, "stream error", err) - return - } - log.Log(reqCtx, "stream success", "url", httpReq.URL.String()) + a.ingestFMP4Push(w, httpReq, r, mediaSigner, "") } // route to accept an incoming fragmented-MP4 stream (the `streamplace live` diff --git a/pkg/api/fmp4_ingest.go b/pkg/api/fmp4_ingest.go new file mode 100644 index 000000000..a9bd6def1 --- /dev/null +++ b/pkg/api/fmp4_ingest.go @@ -0,0 +1,164 @@ +package api + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + + "github.com/google/uuid" + "github.com/julienschmidt/httprouter" + "stream.place/streamplace/pkg/errors" + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/media" +) + +// The Streamplace OBS plugin asks for its stream configuration with the same +// request and response shapes as OBS's Enhanced RTMP multitrack video +// configuration (GoLiveApi, see obs-studio's +// frontend/utility/models/multitrack-video.hpp), so one endpoint can configure +// both. Only the fields the node reads or sets are modeled here. +type clientConfigurationRequest struct { + Authentication string `json:"authentication"` +} + +type clientConfiguration struct { + Meta clientConfigurationMeta `json:"meta"` + Status clientConfigurationStatus `json:"status"` + IngestEndpoints []ingestEndpoint `json:"ingest_endpoints,omitempty"` +} + +type clientConfigurationMeta struct { + Service string `json:"service"` + SchemaVersion string `json:"schema_version"` + ConfigID string `json:"config_id"` +} + +type clientConfigurationStatus struct { + Result string `json:"result"` + HTMLEnUS string `json:"html_en_us,omitempty"` +} + +type ingestEndpoint struct { + Protocol string `json:"protocol"` + URLTemplate string `json:"url_template"` +} + +// HandleClientConfiguration answers the plugin's configuration request. A bad +// stream key is reported in status, as GoLiveApi clients expect, so OBS shows +// the message. The response names no encoders, so the plugin keeps the +// encoders from OBS's output settings: one video and one audio track, which is +// what the node's segment pipeline handles today. +func (a *StreamplaceAPI) HandleClientConfiguration(ctx context.Context) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req clientConfigurationRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + errors.WriteHTTPBadRequest(w, "invalid configuration request", err) + return + } + config := clientConfiguration{ + Meta: clientConfigurationMeta{ + Service: "Streamplace", + SchemaVersion: "2025-01-25", + ConfigID: uuid.NewString(), + }, + Status: clientConfigurationStatus{Result: "success"}, + } + if problem := a.streamKeyProblem(r.Context(), req.Authentication); problem != "" { + config.Status = clientConfigurationStatus{Result: "error", HTMLEnUS: problem} + } else { + scheme := "http" + if r.TLS != nil { + scheme = "https" + } + config.IngestEndpoints = []ingestEndpoint{{ + Protocol: "FMP4", + URLTemplate: fmt.Sprintf("%s://%s/api/ingest/fmp4/{stream_key}", scheme, r.Host), + }} + } + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(config); err != nil { + log.Error(ctx, "error writing client configuration", "error", err) + } + } +} + +// streamKeyProblem explains to the streamer why key may not stream, or returns +// "" if it may. +func (a *StreamplaceAPI) streamKeyProblem(ctx context.Context, key string) string { + ms, err := a.MakeMediaSigner(ctx, key) + if err != nil { + return "This stream key isn't valid. Copy a fresh one from your Streamplace dashboard." + } + if err := a.checkBanned(ctx, ms.Streamer()); err != nil { + return "This account can't stream right now." + } + return "" +} + +// HandleFMP4Ingest accepts the OBS plugin's push: a chunked POST of fragmented +// MP4 that is signed as-is (media.IngestTransportFMP4Direct). +func (a *StreamplaceAPI) HandleFMP4Ingest(ctx context.Context) httprouter.Handle { + return func(w http.ResponseWriter, r *http.Request, p httprouter.Params) { + mediaSigner, err := a.MakeMediaSigner(r.Context(), p.ByName("key")) + if err != nil { + errors.WriteHTTPUnauthorized(w, "invalid authorization key", err) + return + } + a.ingestFMP4Push(w, r, r.Body, mediaSigner, media.IngestTransportFMP4Direct) + } +} + +// ingestFMP4Push runs an authenticated fragmented-MP4 push to completion: in a +// detached worker that takes over the connection (--isolated-ingest), or in +// process, reading body. +func (a *StreamplaceAPI) ingestFMP4Push(w http.ResponseWriter, httpReq *http.Request, body io.Reader, mediaSigner media.MediaSigner, transport string) { + reqCtx := log.WithLogValues(httpReq.Context(), "streamer", mediaSigner.Streamer()) + + if err := a.checkBanned(reqCtx, mediaSigner.Streamer()); err != nil { + errors.WriteHTTPUnauthorized(w, err.Error(), err) + return + } + + if !a.CLI.IsolatedIngest { + if err := a.MediaManager.MP4Ingest(reqCtx, body, mediaSigner, transport); err != nil { + log.Log(reqCtx, "stream error", "error", err) + errors.WriteHTTPInternalServerError(w, "stream error", err) + return + } + log.Log(reqCtx, "stream success") + return + } + + // Zero-downtime path: hijack the authed push connection and hand it to a + // DETACHED worker that owns the connection (so it survives a main + // restart) and serves signed segments back over its socket. + hj, ok := w.(http.Hijacker) + if !ok { + // The isolated path needs a hijackable HTTP/1.1 connection (which the + // real push clients — the `streamplace live` CLI, the OBS plugin, and + // tests pushing over localhost — always are; the Mist ingest itself + // arrives via MistPullIngest). We don't support a non-hijack fallback: + // it couldn't receive mid-stream manifest updates and would stay stuck + // pre-live, so refuse. Such a client can use WHIP instead. + log.Error(reqCtx, "isolated ingest requires a hijackable HTTP/1.1 connection; refusing push") + errors.WriteHTTPInternalServerError(w, "isolated ingest requires a hijackable HTTP/1.1 connection; use WHIP", fmt.Errorf("connection is not hijackable")) + return + } + conn, bufrw, err := hj.Hijack() + if err != nil { + log.Error(reqCtx, "ingest hijack failed", "error", err) + return + } + var prebuf []byte + if n := bufrw.Reader.Buffered(); n > 0 { + prebuf = make([]byte, n) + _, _ = io.ReadFull(bufrw.Reader, prebuf) + } + chunked := len(httpReq.TransferEncoding) > 0 && httpReq.TransferEncoding[0] == "chunked" + if err := a.MediaManager.MP4IngestDetached(reqCtx, conn, prebuf, chunked, mediaSigner, transport); err != nil { + log.Log(reqCtx, "isolated stream ended", "error", err) + } + // The connection is hijacked; the HTTP response is the worker's now. +} diff --git a/pkg/cmd/e2e.go b/pkg/cmd/e2e.go index 0bbbb3102..335c5b7c1 100644 --- a/pkg/cmd/e2e.go +++ b/pkg/cmd/e2e.go @@ -112,6 +112,12 @@ func makeE2eCommand(build *config.BuildFlags) *urfavecli.Command { Usage: "bundle id of the mobile app under test, which the node hands OAuth logins back to (its --app-bundle-id)", Sources: urfavecli.EnvVars("SP_E2E_APP_BUNDLE_ID"), }, + &urfavecli.StringFlag{ + Name: "ingest", + Usage: "how the harness streams for the test account: whip (loop a fixture over WHIP) or none (stream nothing; push with STREAM_KEY yourself, e.g. from OBS)", + Value: "whip", + Sources: urfavecli.EnvVars("SP_E2E_INGEST"), + }, }, Action: func(ctx context.Context, cmd *urfavecli.Command) error { // Canonical form, so they compare equal to the SNI names the @@ -124,7 +130,11 @@ func makeE2eCommand(build *config.BuildFlags) *urfavecli.Command { if cmd.Bool("upstream-station") && pdsHost == "" { return errors.New("--upstream-station requires the HTTPS PDS and station hostnames") } - return runE2E(ctx, cmd.String("dev-env"), pdsHost, stationHost, int(cmd.Int("https-port")), cmd.String("app-bundle-id"), cmd.Bool("upstream-station"), cmd.String("atmoq-bin")) + ingest := cmd.String("ingest") + if ingest != "whip" && ingest != "none" { + return fmt.Errorf("--ingest must be whip or none, not %q", ingest) + } + return runE2E(ctx, cmd.String("dev-env"), pdsHost, stationHost, int(cmd.Int("https-port")), cmd.String("app-bundle-id"), cmd.Bool("upstream-station"), cmd.String("atmoq-bin"), ingest) }, } } @@ -182,7 +192,7 @@ func startNode(ctx context.Context, self string, env e2eDevEnv, dataDir string, return nodeCmd, nil } -func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost string, httpsPort int, appBundleID string, upstream bool, atmoqBin string) error { +func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost string, httpsPort int, appBundleID string, upstream bool, atmoqBin, ingest string) error { // Ctrl-C / SIGTERM must unwind through the normal path, or none of the // teardown below runs: Go's default handling exits immediately, which left // the dev-env node, the forked node and the temp data dir behind. @@ -566,6 +576,9 @@ func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost stri } g, streamCtx := errgroup.WithContext(ctx) g.Go(func() error { + if ingest == "none" { + return nil + } for { select { case <-streamCtx.Done(): @@ -598,8 +611,8 @@ func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost stri if local != nil { serverAddr = local.addr } - vars := fmt.Sprintf("SERVER_URL=http://%s\nSERVER2_URL=http://%s\nSERVER3_URL=http://%s\nACCOUNT_HANDLE=%s\nACCOUNT_DID=%s\nACCOUNT_PASSWORD=%s\nVIDEO_URI=%s\nE2E_FIXTURE_MP4=%s\n", - serverAddr, peerAddr, pushAddr, out.Handle, out.Did, password, videoURI, fixture) + vars := fmt.Sprintf("SERVER_URL=http://%s\nSERVER2_URL=http://%s\nSERVER3_URL=http://%s\nACCOUNT_HANDLE=%s\nACCOUNT_DID=%s\nACCOUNT_PASSWORD=%s\nVIDEO_URI=%s\nE2E_FIXTURE_MP4=%s\nSTREAM_KEY=%s\n", + serverAddr, peerAddr, pushAddr, out.Handle, out.Did, password, videoURI, fixture, priv) if tlsEnv != nil { vars += fmt.Sprintf("PDS_HTTPS_URL=%s\nE2E_PROXY_URL=%s\nE2E_TLS_SPKI=%s\nE2E_TLS_CA=%s\n", tlsEnv.PDSURL(), tlsEnv.ProxyURL(), tlsEnv.spki, tlsEnv.caPath) diff --git a/pkg/media/dynamic_manifest_test.go b/pkg/media/dynamic_manifest_test.go index 3f387adaa..291abf2e0 100644 --- a/pkg/media/dynamic_manifest_test.go +++ b/pkg/media/dynamic_manifest_test.go @@ -89,7 +89,7 @@ func TestSignSegmentStreamRefreshesManifestPerSegment(t *testing.T) { if ev.Type != "signed-segment" { continue } - gops = append(gops, concatTracksSorted(ev.Tracks)) + gops = append(gops, concatTracksByID(ev.Tracks)) } require.NoError(t, <-errCh) require.GreaterOrEqual(t, len(gops), 2, diff --git a/pkg/media/ingest_daemon.go b/pkg/media/ingest_daemon.go index cab8825ba..04776026c 100644 --- a/pkg/media/ingest_daemon.go +++ b/pkg/media/ingest_daemon.go @@ -250,13 +250,14 @@ func (mm *MediaManager) ingestWorkerSocketDir() (string, error) { // worker (own session, survives a main restart) which ingests the media directly // and serves signed segments over a per-session unix socket, and then consumes // those frames into ValidateMP4 with reconnect. prebuf is any body bytes main -// already read past the headers; chunked says the push body is chunked. +// already read past the headers; chunked says the push body is chunked; +// transport is the worker's cfg.Transport ("" for the demux pipeline). // // Because the worker owns the connection and is detached, a main restart neither // breaks the ingest nor loses output: the worker keeps signing into its buffer, // and the restarted main rediscovers the socket (DiscoverWorkerSockets) and // drains it. -func (mm *MediaManager) MP4IngestDetached(ctx context.Context, conn net.Conn, prebuf []byte, chunked bool, ms MediaSigner) error { +func (mm *MediaManager) MP4IngestDetached(ctx context.Context, conn net.Conn, prebuf []byte, chunked bool, ms MediaSigner, transport string) error { cfg, err := mm.buildWorkerConfig(ctx, ms) if err != nil { return err @@ -269,6 +270,7 @@ func (mm *MediaManager) MP4IngestDetached(ctx context.Context, conn net.Conn, pr cfg.InputFD = 4 cfg.Prebuf = prebuf cfg.Chunked = chunked + cfg.Transport = transport tcp, ok := conn.(*net.TCPConn) if !ok { diff --git a/pkg/media/ingest_worker.go b/pkg/media/ingest_worker.go index f2f54d4b5..1d1d85bad 100644 --- a/pkg/media/ingest_worker.go +++ b/pkg/media/ingest_worker.go @@ -105,8 +105,10 @@ type IngestWorkerConfig struct { S3 *s3.Config `json:"s3,omitempty"` // Transport selects the worker's ingest source: "" / "mp4" reads fragmented - // MP4 media (stdin or InputFD); "whip" makes the worker own the WebRTC - // PeerConnection, built from OfferSDP — no media fd to pass. + // MP4 media (stdin or InputFD) through the demux pipeline; + // IngestTransportFMP4Direct reads the same way but signs the fMP4 as-is; + // "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) @@ -117,6 +119,12 @@ type IngestWorkerConfig struct { // IngestTransportWHIP is the cfg.Transport value selecting the WHIP worker. const IngestTransportWHIP = "whip" +// IngestTransportFMP4Direct is the cfg.Transport value (and MP4Ingest +// transport) for a client whose fMP4 needs no remuxing, such as the +// Streamplace OBS plugin: it is signed as-is, keeping every track (see +// signFMP4Direct). +const IngestTransportFMP4Direct = "fmp4-direct" + // workerCLI assembles the minimal config.CLI a worker runs with: the // broadcaster identity plus the debug-recording destination (S3 when main // handed its config over the handshake, else local disk under DataDir). Shared @@ -206,9 +214,10 @@ func (mm *MediaManager) workerSegmentSink(ctx context.Context, cfg IngestWorkerC // RunMP4IngestWorker is the body of the `ingest-worker` subcommand. It reads a // fragmented-MP4 stream from stdin, runs the same demux + Opus re-encode + muxl-sign -// pipeline as the in-process MP4Ingest, 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. +// pipeline as the in-process MP4Ingest (or, for IngestTransportFMP4Direct, signs +// the stream as-is), 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 @@ -246,6 +255,12 @@ func RunMP4IngestWorker(ctx context.Context, cfg IngestWorkerConfig, stdin io.Re defer finalize() } + if cfg.Transport == IngestTransportFMP4Direct { + err := signFMP4Direct(ctx, mm.cli, workerSignStream(cfg, getManifest), media, onSegment) + flush() + return err + } + signerElem, done, err := muxlSignSegmentElem(ctx, mm.cli, workerSignStream(cfg, getManifest), onSegment) if err != nil { return fmt.Errorf("build signer element: %w", err) diff --git a/pkg/media/ingest_worker_test.go b/pkg/media/ingest_worker_test.go index d5a298e22..82e0b258e 100644 --- a/pkg/media/ingest_worker_test.go +++ b/pkg/media/ingest_worker_test.go @@ -174,6 +174,58 @@ func TestRunMP4IngestWorkerProducesValidSignedFrames(t *testing.T) { t.Logf("worker emitted %d valid dual-codec segments", segs) } +// TestRunMP4IngestWorkerFMP4DirectKeepsTracks covers the OBS plugin's +// transport: the worker signs the pushed fMP4 as-is, so a stream with an AAC +// and an Opus track comes out with both, unconverted. The demux pipeline keeps +// one audio track and re-encodes it to Opus, which is exactly what multitrack +// audio must not do. +func TestRunMP4IngestWorkerFMP4DirectKeepsTracks(t *testing.T) { + 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, + Transport: IngestTransportFMP4Direct, + } + + mp4 := runSynthPipeline(t, ctx, strings.Join([]string{ + "videotestsrc num-buffers=90 ! video/x-raw,width=320,height=240,framerate=30/1 ! x264enc key-int-max=30 tune=zerolatency speed-preset=ultrafast ! h264parse ! mp4mux name=mux fragment-duration=500 ! appsink name=sink", + "audiotestsrc num-buffers=141 samplesperbuffer=1024 ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! fdkaacenc ! aacparse ! mux.", + "audiotestsrc num-buffers=150 samplesperbuffer=960 ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! opusenc ! opusparse ! mux.", + }, "\n")) + + var buf bytes.Buffer + require.NoError(t, RunMP4IngestWorker(ctx, cfg, bytes.NewReader(mp4), ingestframe.NewWriter(&buf), func() []byte { return cfg.Manifest })) + + r := ingestframe.NewReader(&buf) + var segs int + for { + typ, payload, err := r.ReadFrame() + if errors.Is(err, io.EOF) { + break + } + require.NoError(t, err) + require.Equal(t, ingestframe.Segment, typ) + + out, err := muxl.RunMuxlVerify(ctx, bytes.NewReader(payload)) + require.NoError(t, err, "segment %d verify", segs) + require.NotContains(t, out, `"validation_state":"Invalid"`, "segment %d must validate", segs) + + codecs := audioCodecsOf(t, ctx, payload) + require.Len(t, codecs, 2, "segment %d keeps both audio tracks (got %v)", segs, codecs) + require.True(t, isAACCodec(codecs[0]), "segment %d keeps the AAC track as AAC (got %v)", segs, codecs) + require.True(t, isOpusCodec(codecs[1]), "segment %d keeps the Opus track (got %v)", segs, codecs) + segs++ + } + require.GreaterOrEqual(t, segs, 2, "a 3s stream with 1s GoPs signs into several segments") +} + // TestRunMP4IngestWorkerRecords proves debug recording works INSIDE the worker: // with cfg.Record set and a DataDir handed over, the worker tees its ingest // media to debug-recordings//.rtmp.mp4. This is what keeps debug diff --git a/pkg/media/mist_pull.go b/pkg/media/mist_pull.go index 66d473be9..aa951c367 100644 --- a/pkg/media/mist_pull.go +++ b/pkg/media/mist_pull.go @@ -54,11 +54,11 @@ func (mm *MediaManager) MistPullIngest(ctx context.Context, mistStreamName strin // it survives a main restart) and serves signed segments back over its // socket — the same machinery as the old hijacked inbound push, with // the connection pointing the other way. - return mm.MP4IngestDetached(ctx, conn, prebuf, chunked, ms) + return mm.MP4IngestDetached(ctx, conn, prebuf, chunked, ms, "") } defer conn.Close() body := WorkerInput(IngestWorkerConfig{Prebuf: prebuf, Chunked: chunked}, conn) - return mm.MP4Ingest(ctx, body, ms) + return mm.MP4Ingest(ctx, body, ms, "") } // mistPullConnect dials Mist's HTTP output and issues the GET by hand — not diff --git a/pkg/media/mp4_ingest.go b/pkg/media/mp4_ingest.go index d0dbf407e..f710ff24a 100644 --- a/pkg/media/mp4_ingest.go +++ b/pkg/media/mp4_ingest.go @@ -17,9 +17,10 @@ import ( "stream.place/streamplace/pkg/log" ) -// ingest a H264+AAC fragmented-MP4 stream (the MistServer live .mp4 output, or -// an fMP4 push to /live) -func (mm *MediaManager) MP4Ingest(ctx context.Context, input io.Reader, ms MediaSigner) error { +// ingest a fragmented-MP4 stream: H264+AAC through the demux pipeline (the +// MistServer live .mp4 output, or an fMP4 push to /live), or, with transport +// IngestTransportFMP4Direct, signed as-is (the OBS plugin's push). +func (mm *MediaManager) MP4Ingest(ctx context.Context, input io.Reader, ms MediaSigner, transport string) error { shouldRecord, err := mm.shouldRecord(ctx, ms.Streamer()) if err != nil { return err @@ -32,8 +33,16 @@ func (mm *MediaManager) MP4Ingest(ctx context.Context, input io.Reader, ms Media } else { log.Log(ctx, "not recording ingest stream to file", "streamer", ms.Streamer()) } - ctx, cancel := context.WithCancel(ctx) - defer cancel() + ctx, cancel := context.WithCancelCause(ctx) + defer cancel(nil) + + if transport == IngestTransportFMP4Direct { + go mm.watchKeyRevocation(ctx, ms.Streamer(), ms.DID(), func(reason string) { + cancel(fmt.Errorf("ending stream: %s", reason)) + }) + ctx, onSegment := mm.validatingSegmentSink(ctx, ms) + return signFMP4Direct(ctx, mm.cli, ms.SignSegmentStream, input, onSegment) + } signer, err := mm.SegmentAndSignElem(ctx, ms) if err != nil { diff --git a/pkg/media/muxl_segment.go b/pkg/media/muxl_segment.go index 112c2cbad..f8afaf32a 100644 --- a/pkg/media/muxl_segment.go +++ b/pkg/media/muxl_segment.go @@ -5,7 +5,6 @@ import ( "context" "fmt" "io" - "sort" "github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" @@ -104,39 +103,13 @@ 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(drainCtx, r, eventCh) - close(eventCh) - if err != nil && ctx.Err() == nil { - log.Error(ctx, "error running muxl sign-segment", "error", err) - } - }() done := make(chan struct{}) go func() { defer close(done) - for ev := range eventCh { - if ev.Type != "signed-segment" { - continue - } - segment := concatTracksSorted(ev.Tracks) - 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) - } + // Non-cancellable: cancelling ctx is the FLUSH signal, not an abort — see + // signSegments. + if err := signSegments(context.WithoutCancel(ctx), cli, signStream, r, onSegment); err != nil && ctx.Err() == nil { + log.Error(ctx, "error running muxl sign-segment", "error", err) } }() @@ -148,18 +121,57 @@ func muxlSignSegmentElem(ctx context.Context, cli *config.CLI, signStream SignSe return bin.Element, done, nil } -// concatTracksSorted joins the per-track canonical segment bytes for one GoP -// in ascending track-id order — the canonical interleave a multi-track .m4s -// uses, which muxl's unwrap/verify/wrap all expect. -func concatTracksSorted(tracks map[string][]byte) []byte { - keys := make([]string, 0, len(tracks)) - for k := range tracks { - keys = append(keys, k) +// signSegments streams an fMP4 input through the per-segment signer and hands +// each GoP's bare canonical .m4s to onSegment, returning once all of them have +// been handed over. +// +// ctx MUST NOT be cancellable; callers stop the signer by closing input +// instead. Closing input is the FLUSH signal, not an abort: the signer sees +// EOF, signs the final GoP, and exits cleanly. If a 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, so this +// would never return. 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. +func signSegments(ctx context.Context, cli *config.CLI, signStream SignSegmentStreamFunc, input io.Reader, onSegment func(ctx context.Context, segment []byte) error) error { + eventCh := make(chan *muxl.MuxlEvent, 16) + errCh := make(chan error, 1) + go func() { + err := signStream(ctx, input, eventCh) + close(eventCh) + errCh <- err + }() + for ev := range eventCh { + if ev.Type != "signed-segment" { + continue + } + segment := concatTracksByID(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) + } } - sort.Strings(keys) - var out []byte - for _, k := range keys { - out = append(out, tracks[k]...) + return <-errCh +} + +// signFMP4Direct signs an fMP4 stream that needs no remuxing straight through +// the per-segment signer, with no GStreamer graph in front of it. That fits a +// client that already muxes every track the way a segment carries it — the +// Streamplace OBS plugin, whose encoders flag only IDR frames as keyframes — +// and it keeps every track the client sends, where the demux pipeline keeps +// one video and one audio track. Cancelling ctx flushes the final GoP; it +// returns once every segment has been handed to onSegment. +func signFMP4Direct(ctx context.Context, cli *config.CLI, signStream SignSegmentStreamFunc, input io.Reader, onSegment func(ctx context.Context, segment []byte) error) error { + r, w := io.Pipe() + go func() { + _, err := io.Copy(w, input) + w.CloseWithError(err) + }() + stop := context.AfterFunc(ctx, func() { r.Close() }) + defer stop() + err := signSegments(context.WithoutCancel(ctx), cli, signStream, r, onSegment) + if ctx.Err() != nil { + return context.Cause(ctx) } - return out + return err } diff --git a/pkg/media/segmenter.go b/pkg/media/segmenter.go index 9269efb2b..40de60568 100644 --- a/pkg/media/segmenter.go +++ b/pkg/media/segmenter.go @@ -177,6 +177,13 @@ func SegmentElem(ctx context.Context, cli *config.CLI, streamer string, doH264Pa } func (mm *MediaManager) SegmentAndSignElem(ctx context.Context, ms MediaSigner) (*gst.Element, error) { + ctx, onSegment := mm.validatingSegmentSink(ctx, ms) + return MuxlSignSegmentElem(ctx, mm.cli, ms, onSegment) +} + +// validatingSegmentSink returns the onSegment handler for one ingest session, +// along with the context its signer must run under. +func (mm *MediaManager) validatingSegmentSink(ctx context.Context, ms MediaSigner) (context.Context, func(ctx context.Context, segment []byte) error) { tracer := otel.Tracer("signer") streamer := ms.Streamer() @@ -213,7 +220,7 @@ func (mm *MediaManager) SegmentAndSignElem(ctx context.Context, ms MediaSigner) } return nil } - return MuxlSignSegmentElem(ctx, mm.cli, ms, onSegment) + return ctx, onSegment } func SegmentFileUnsigned(ctx context.Context, cli *config.CLI, streamer string, input string, ch chan *SplitSegment) error { diff --git a/pkg/media/transcode_stream_test.go b/pkg/media/transcode_stream_test.go index 9a458c064..647ee3343 100644 --- a/pkg/media/transcode_stream_test.go +++ b/pkg/media/transcode_stream_test.go @@ -134,7 +134,7 @@ func allSignedBareSegments(t *testing.T, ctx context.Context, ms *MediaSignerLoc var out [][]byte for ev := range eventCh { if ev.Type == "signed-segment" { - out = append(out, concatTracksSorted(ev.Tracks)) + out = append(out, concatTracksByID(ev.Tracks)) } } require.NoError(t, <-errCh) diff --git a/pkg/media/validate_bare_test.go b/pkg/media/validate_bare_test.go index 8f136468f..5ae094d3a 100644 --- a/pkg/media/validate_bare_test.go +++ b/pkg/media/validate_bare_test.go @@ -51,7 +51,7 @@ func TestValidateMP4MediaBareSegment(t *testing.T) { var m4s []byte for ev := range eventCh { if ev.Type == "signed-segment" && m4s == nil { - m4s = concatTracksSorted(ev.Tracks) + m4s = concatTracksByID(ev.Tracks) } } require.NoError(t, <-errCh) -- 2.54.0