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)