From 75af90a47fc7907364201876e91ebb5895c28a6a Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Sun, 7 Jun 2026 09:58:59 -0700 Subject: [PATCH] =?UTF-8?q?media:=20refresh=20isolated=20workers'=20manife?= =?UTF-8?q?st=20so=20pre-live=20=E2=86=92=20live=20takes=20effect?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes a regression on --isolated-ingest: a streamer goes live but stays stuck "pre-live" (live HLS never populates, viewers get "segment is not published"). Root cause: "published" is baked into each segment's signed C2PA manifest (a c2pa.published action, added once the streamer has a live livestream record). The in-process signer rebuilds the manifest FRESH per GoP from the model, so a pre-live → live transition shows up mid-stream. The isolated worker has no model, so buildWorkerConfig froze the manifest at spawn (pre-live) and workerSignStream reused it forever — re-introducing exactly the "sealed at connection start" bug the in-process path fixed. Fix: main pushes manifest refreshes to the worker over the existing socket and the worker signs each GoP with the latest. - ingestframe.Manifest: a main→worker control frame (reverse of the segment stream). - Worker holds the manifest in a manifestHolder (init = the spawn manifest); workerSignStream reads it per GoP; serveFrameSocket reads control frames from main and swaps the holder on a Manifest frame. - Main rebuilds the manifest from NewManifestBuilder(model, cli).BuildManifest for the streamer DID — needs only the model + cli, no signing key — and pushes on change (pushManifestUpdates, fixed start so only real changes diff). Wired through ConsumeWorkerSocket for MKVIngestDetached, WHIPIngestDetached, and ResumeDetachedWorkers (DID from the sidecar — so a worker that outlived a main restart gets refreshed too, no signer reconstruction). Also drops the fd-4 non-hijack fallback: it has no back-channel for manifest updates and would stay stuck pre-live, so a non-hijackable push now errors (use WHIP). The only real MKV/RTMP client is a co-located MistServer pushing HTTP/1.1 over localhost, which always hijacks. MKVIngestIsolated stays for its tests but is no longer wired into the API. Tests: pushManifestUpdates emits on change only; serveFrameSocket applies a pushed Manifest; and TestRunMKVIngestWorkerSignsWithSuppliedManifest proves end to end that the worker signs unpublished vs published segments purely from the manifest the getter returns. Co-Authored-By: Claude Opus 4.8 --- pkg/api/api_internal.go | 11 ++-- pkg/cmd/streamplace.go | 5 +- pkg/ingestframe/frame.go | 12 +++++ pkg/media/frame_server.go | 29 ++++++---- pkg/media/frame_server_test.go | 2 +- pkg/media/ingest_daemon.go | 83 +++++++++++++++++++++++++++-- pkg/media/ingest_daemon_test.go | 2 +- pkg/media/ingest_subprocess_test.go | 2 +- pkg/media/ingest_worker.go | 42 ++++++++++++--- pkg/media/ingest_worker_test.go | 48 +++++++++++++++-- pkg/media/manifest_refresh_test.go | 81 ++++++++++++++++++++++++++++ pkg/media/whip_worker.go | 5 +- 12 files changed, 290 insertions(+), 32 deletions(-) create mode 100644 pkg/media/manifest_refresh_test.go diff --git a/pkg/api/api_internal.go b/pkg/api/api_internal.go index 67b54ae38..144bf0a65 100644 --- a/pkg/api/api_internal.go +++ b/pkg/api/api_internal.go @@ -283,9 +283,14 @@ func (a *StreamplaceAPI) InternalHandler(ctx context.Context) (http.Handler, err } return // connection hijacked; the HTTP response is ours now } - // No hijack support (HTTP/2, some proxies) → fd-4-pipe isolation: still - // fault-isolated, just no restart-survival. - err = a.MediaManager.MKVIngestIsolated(reqCtx, r, mediaSigner) + // The isolated path needs a hijackable HTTP/1.1 connection (which the + // only real MKV/RTMP-push client — a co-located MistServer pushing over + // localhost — always is). 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.MKVIngest(reqCtx, r, mediaSigner) } diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index 2c6a9ca3b..31f9625f3 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -915,7 +915,10 @@ func makeIngestWorkerCommand(build *config.BuildFlags) *urfavecli.Command { defer framesFile.Close() frames := ingestframe.NewWriter(framesFile) - if err := media.RunMKVIngestWorker(ctx, cfg, os.Stdin, frames); err != nil { + // This fd-4 path has no back-channel for manifest updates, so the + // manifest stays whatever main built at spawn. It's not used in prod + // (api requires a hijackable connection); kept for the worker self-test. + if err := media.RunMKVIngestWorker(ctx, cfg, os.Stdin, frames, func() []byte { return cfg.Manifest }); err != nil { _ = frames.Error(err.Error()) return err } diff --git a/pkg/ingestframe/frame.go b/pkg/ingestframe/frame.go index 22e52a425..c335185c5 100644 --- a/pkg/ingestframe/frame.go +++ b/pkg/ingestframe/frame.go @@ -55,6 +55,13 @@ const ( // with bytes acked) back to main, which writes it to the DB — the worker has // no DB access of its own. Payload: JSON {status, message}. Event Type = 5 + // Manifest carries an updated C2PA manifest (UTF-8 JSON) from MAIN TO the + // worker — the reverse of the segment stream, over the same socket. The worker + // signs each GoP with the latest manifest it holds, so main can flip a stream + // pre-live → live (and apply other mid-stream manifest changes) without the + // worker reconnecting — it has no model of its own to notice the change. + // Payload: the manifest JSON. + Manifest Type = 6 ) func (t Type) String() string { @@ -69,6 +76,8 @@ func (t Type) String() string { return "answer" case Event: return "event" + case Manifest: + return "manifest" default: return fmt.Sprintf("unknown(%d)", uint8(t)) } @@ -132,6 +141,9 @@ func (fw *Writer) Answer(sdp string) error { return fw.WriteFrame(Answer, []byte // Event frames a worker status update (JSON payload). func (fw *Writer) Event(payload []byte) error { return fw.WriteFrame(Event, payload) } +// Manifest frames an updated C2PA manifest (main → worker). +func (fw *Writer) Manifest(payload []byte) error { return fw.WriteFrame(Manifest, payload) } + // Reader decodes frames from an underlying stream. The decoder buffers/reads // ahead, so a Reader OWNS its stream for the stream's lifetime — don't create a // second Reader on the same connection (it would lose the first's buffered diff --git a/pkg/media/frame_server.go b/pkg/media/frame_server.go index 6b2dad416..6447e8232 100644 --- a/pkg/media/frame_server.go +++ b/pkg/media/frame_server.go @@ -189,9 +189,10 @@ func ServeMKVIngestWorkerSocket(ctx context.Context, cfg IngestWorkerConfig, std }() srv := newFrameServer(workerFrameBuffer) - go serveFrameSocket(ctx, ln, srv) + manifest := newManifestHolder(cfg.Manifest) + go serveFrameSocket(ctx, ln, srv, manifest) - runErr := RunMKVIngestWorker(ctx, cfg, stdin, srv) + runErr := RunMKVIngestWorker(ctx, cfg, stdin, srv, manifest.get) if runErr != nil { _ = srv.Error(runErr.Error()) } else { @@ -207,10 +208,12 @@ func ServeMKVIngestWorkerSocket(ctx context.Context, cfg IngestWorkerConfig, std } // serveFrameSocket accepts client connections on ln and attaches each to the -// server, replacing any prior client (main reconnecting after a restart). Each -// connection is watched for close so the server reverts to buffering. Returns -// when ctx is cancelled or the listener is closed. -func serveFrameSocket(ctx context.Context, ln net.Listener, s *frameServer) { +// server, replacing any prior client (main reconnecting after a restart). The +// reverse direction carries control frames from main: it reads them and applies +// a Manifest update to the holder; any other frame is ignored. The read also +// unblocks when main disconnects, at which point the server reverts to +// buffering. Returns when ctx is cancelled or the listener is closed. +func serveFrameSocket(ctx context.Context, ln net.Listener, s *frameServer, manifest *manifestHolder) { go func() { <-ctx.Done() ln.Close() @@ -223,9 +226,17 @@ func serveFrameSocket(ctx context.Context, ln net.Listener, s *frameServer) { log.Log(ctx, "ingest worker: main attached to frame socket") s.attach(conn) go func(c net.Conn) { - // Main only reads frames; this drains anything it sends (nothing today) - // and unblocks when it disconnects, at which point we revert to buffering. - _, _ = io.Copy(io.Discard, c) + fr := ingestframe.NewReader(c) + for { + typ, payload, rerr := fr.ReadFrame() + if rerr != nil { + break // main disconnected (or sent garbage); revert to buffering + } + if typ == ingestframe.Manifest { + manifest.set(payload) + log.Log(ctx, "ingest worker: manifest updated by main", "bytes", len(payload)) + } + } s.detachConn(c) log.Log(ctx, "ingest worker: main detached from frame socket") }(conn) diff --git a/pkg/media/frame_server_test.go b/pkg/media/frame_server_test.go index 68b4007fa..22810e2de 100644 --- a/pkg/media/frame_server_test.go +++ b/pkg/media/frame_server_test.go @@ -126,7 +126,7 @@ func TestServeFrameSocketAttachAndFlush(t *testing.T) { for i := 0; i < 3; i++ { // buffered before anyone connects require.NoError(t, srv.Segment(seg(i))) } - go serveFrameSocket(ctx, ln, srv) + go serveFrameSocket(ctx, ln, srv, newManifestHolder(nil)) client, err := net.Dial("unix", sock) require.NoError(t, err) diff --git a/pkg/media/ingest_daemon.go b/pkg/media/ingest_daemon.go index 4a60fba04..3cee3517b 100644 --- a/pkg/media/ingest_daemon.go +++ b/pkg/media/ingest_daemon.go @@ -1,6 +1,7 @@ package media import ( + "bytes" "context" "encoding/json" "errors" @@ -129,7 +130,11 @@ func removeWorkerFiles(socketPath string) { // outlived a main restart keeps buffering and replays on reconnect; ValidateMP4's // dedup makes any replayed overlap idempotent. Returns nil on a clean End, or an // error if the socket vanishes without one (worker crashed — contained). -func (mm *MediaManager) ConsumeWorkerSocket(ctx context.Context, socketPath, streamer string, onSegment func([]byte) error) error { +// +// manifestSource, when non-nil, is polled to refresh the worker's C2PA manifest +// over the same socket (pushManifestUpdates) — so a pre-live → live transition +// reaches a worker that has no model of its own. It's re-armed per connection. +func (mm *MediaManager) ConsumeWorkerSocket(ctx context.Context, socketPath, streamer string, onSegment func([]byte) error, manifestSource func() ([]byte, error)) error { connectedOnce := false giveUp := time.Now().Add(workerConnectGrace) for { @@ -157,9 +162,16 @@ func (mm *MediaManager) ConsumeWorkerSocket(ctx context.Context, socketPath, str return fmt.Errorf("ingest worker socket gone before End: %w", err) } connectedOnce = true + // Push manifest refreshes to the worker over this connection (re-armed per + // connect); stop the pusher when the connection ends. + connCtx, connCancel := context.WithCancel(ctx) + if manifestSource != nil { + go pushManifestUpdates(connCtx, conn, manifestSource) + } // Fresh Reader per connection: a reconnect is a new stream where the worker // replays its buffer from the start. sawEnd, _ := mm.consumeWorkerFrames(ctx, ingestframe.NewReader(conn), streamer, onSegment, nil) + connCancel() conn.Close() if sawEnd { return nil @@ -177,6 +189,52 @@ func (mm *MediaManager) ConsumeWorkerSocket(ctx context.Context, socketPath, str } } +// manifestRefreshInterval bounds how stale an isolated worker's manifest can be — +// roughly how long after a pre-live → live transition before the worker starts +// signing published segments. Var (not const) so tests can shorten it. +var manifestRefreshInterval = 2 * time.Second + +// pushManifestUpdates periodically rebuilds the streamer's manifest and sends it +// to the worker over conn whenever it changes, so a pre-live → live transition +// (or a mid-stream title/metadata change) reaches the isolated worker — which +// has no model to notice it — without reconnecting. manifestSource builds with a +// fixed start, so the bytes change only on real content changes (muxl stamps the +// per-segment timestamp). Returns on ctx cancel or a write error (the connection +// dropped; the consume loop reconnects and re-arms the pusher). +func pushManifestUpdates(ctx context.Context, conn net.Conn, manifestSource func() ([]byte, error)) { + w := ingestframe.NewWriter(conn) + var last []byte + tick := time.NewTicker(manifestRefreshInterval) + defer tick.Stop() + for { + manifest, err := manifestSource() + if err != nil { + log.Error(ctx, "ingest worker: build manifest for refresh", "error", err) + } else if !bytes.Equal(manifest, last) { + if werr := w.Manifest(manifest); werr != nil { + return + } + last = manifest + } + select { + case <-ctx.Done(): + return + case <-tick.C: + } + } +} + +// streamerManifest builds the current C2PA manifest for streamerDID from live +// model state (title, pre-live → live c2pa.published) — what an isolated worker +// needs but can't compute itself (no model). It needs only the model + cli, not +// the streamer's signing key, so it works the same for a live worker (DID from +// the signer) and a resumed one (DID from the sidecar). start seeds a placeholder +// dc:date muxl overwrites per segment, so a fixed value keeps bytes stable for +// change detection. +func (mm *MediaManager) streamerManifest(ctx context.Context, streamerDID string, start int64) ([]byte, error) { + return NewManifestBuilder(mm.model, mm.cli).BuildManifest(ctx, streamerDID, start) +} + // ingestWorkerSocketDir returns (creating it) the directory of per-session // worker frame sockets — the set a restarting main scans to resume. func (mm *MediaManager) ingestWorkerSocketDir() (string, error) { @@ -238,7 +296,11 @@ func (mm *MediaManager) MKVIngestDetached(ctx context.Context, conn net.Conn, pr _ = proc.Kill() }) - err = mm.ConsumeWorkerSocket(ctx, cfg.SocketPath, ms.Streamer(), mm.validateSegment(ctx)) + // Refresh the worker's manifest from live model state (pre-live → live), since + // it signs with a frozen one otherwise. Fixed start for stable change detection. + start := time.Now().UnixMilli() + manifestSource := func() ([]byte, error) { return mm.streamerManifest(ctx, ms.Streamer(), start) } + err = mm.ConsumeWorkerSocket(ctx, cfg.SocketPath, ms.Streamer(), mm.validateSegment(ctx), manifestSource) recordWorkerExit("mkv", err, ctx.Err()) // Reap the worker unless we're deliberately leaving it running across a main // restart (ctx cancel). On a clean end OR a crash the worker has exited, so @@ -355,13 +417,19 @@ func (mm *MediaManager) WHIPIngestDetached(ctx context.Context, offerSDP string, _ = proc.Kill() }) + // Refresh the worker's manifest (pre-live → live) over this connection, + // scoped to the consume; the reconnect fallback re-arms its own. + start := time.Now().UnixMilli() + manifestSource := func() ([]byte, error) { return mm.streamerManifest(ctx, ms.Streamer(), start) } + go pushManifestUpdates(wctx, conn, manifestSource) + sawEnd, _ := mm.consumeWorkerFrames(ctx, fr, ms.Streamer(), mm.validateSegment(ctx), nil) conn.Close() var exitErr error if !sawEnd && ctx.Err() == nil { // Connection dropped but the detached worker lives on — reconnect and // drain its buffer. Its terminal result is the worker's true outcome. - exitErr = mm.ConsumeWorkerSocket(ctx, cfg.SocketPath, ms.Streamer(), mm.validateSegment(ctx)) + exitErr = mm.ConsumeWorkerSocket(ctx, cfg.SocketPath, ms.Streamer(), mm.validateSegment(ctx), manifestSource) } recordWorkerExit("whip", exitErr, ctx.Err()) go func() { _, _ = proc.Wait() }() @@ -397,15 +465,20 @@ func (mm *MediaManager) ResumeDetachedWorkers(ctx context.Context) { // Re-arm ban enforcement for a worker we didn't spawn. We have no process // handle, so kill by the PID in the sidecar (guarded against PID reuse). // Without metadata we can still drain the worker, just not enforce bans. + var manifestSource func() ([]byte, error) if merr != nil || meta.StreamerDID == "" { - log.Warn(ctx, "resumed ingest worker: no resume metadata; cannot enforce bans on it", "socket", sock, "error", merr) + log.Warn(ctx, "resumed ingest worker: no resume metadata; cannot enforce bans or refresh manifest on it", "socket", sock, "error", merr) } else { go mm.watchKeyRevocation(wctx, meta.StreamerDID, meta.StreamerDID, func(reason string) { log.Warn(wctx, "resumed ingest worker: ending stream", "reason", reason, "streamer", meta.StreamerDID) killWorkerPID(wctx, meta.PID, meta.StreamerDID) }) + // Keep refreshing the resumed worker's manifest too (no signer needed — + // streamerManifest only uses the model + cli + the sidecar DID). + start := time.Now().UnixMilli() + manifestSource = func() ([]byte, error) { return mm.streamerManifest(ctx, meta.StreamerDID, start) } } - cerr := mm.ConsumeWorkerSocket(wctx, sock, streamer, mm.validateSegment(ctx)) + cerr := mm.ConsumeWorkerSocket(wctx, sock, streamer, mm.validateSegment(ctx), manifestSource) if cerr != nil { log.Error(ctx, "resumed ingest worker ended", "socket", sock, "error", cerr) } diff --git a/pkg/media/ingest_daemon_test.go b/pkg/media/ingest_daemon_test.go index 99502fd8e..18e36c4d3 100644 --- a/pkg/media/ingest_daemon_test.go +++ b/pkg/media/ingest_daemon_test.go @@ -94,7 +94,7 @@ func TestDetachedWorkerZeroDowntime(t *testing.T) { segs++ return nil } - require.NoError(t, (&MediaManager{}).ConsumeWorkerSocket(ctx, socks[0], ms.Streamer(), onSegment)) + require.NoError(t, (&MediaManager{}).ConsumeWorkerSocket(ctx, socks[0], ms.Streamer(), onSegment, nil)) require.GreaterOrEqual(t, segs, 1, "detached worker served signed segments") _, _ = proc.Wait() // reap the detached worker diff --git a/pkg/media/ingest_subprocess_test.go b/pkg/media/ingest_subprocess_test.go index f9f46d9b3..0bdf83bb1 100644 --- a/pkg/media/ingest_subprocess_test.go +++ b/pkg/media/ingest_subprocess_test.go @@ -69,7 +69,7 @@ func runIngestWorkerHelper() int { } defer framesFile.Close() frames := ingestframe.NewWriter(framesFile) - if err := RunMKVIngestWorker(context.Background(), cfg, os.Stdin, frames); err != nil { + if err := RunMKVIngestWorker(context.Background(), cfg, os.Stdin, frames, func() []byte { return cfg.Manifest }); err != nil { _ = frames.Error(err.Error()) return 1 } diff --git a/pkg/media/ingest_worker.go b/pkg/media/ingest_worker.go index 09d47263b..e8e64b841 100644 --- a/pkg/media/ingest_worker.go +++ b/pkg/media/ingest_worker.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "net/http/httputil" + "sync" "github.com/go-gst/go-gst/gst" "stream.place/streamplace/pkg/config" @@ -14,6 +15,32 @@ import ( "stream.place/streamplace/pkg/muxl" ) +// manifestHolder holds the worker's current C2PA manifest. It starts as the +// manifest main built at spawn (cfg.Manifest) and is swapped when main pushes an +// updated one over the socket (an ingestframe.Manifest control frame). The +// signer reads the latest per GoP, so a pre-live → live transition reaches the +// worker mid-stream — mirroring the in-process signer's fresh-per-GoP manifest, +// which the worker otherwise can't do (it has no model). Concurrent-safe: the +// signer reads while the socket goroutine writes. +type manifestHolder struct { + mu sync.RWMutex + b []byte +} + +func newManifestHolder(initial []byte) *manifestHolder { return &manifestHolder{b: initial} } + +func (h *manifestHolder) get() []byte { + h.mu.RLock() + defer h.mu.RUnlock() + return h.b +} + +func (h *manifestHolder) set(b []byte) { + h.mu.Lock() + defer h.mu.Unlock() + h.b = b +} + // IngestWorkerConfig is the startup handshake the main process hands an ingest // worker over a dedicated pipe fd — kept off argv/env so key material never // lands in a process listing. @@ -101,11 +128,14 @@ func WorkerInput(cfg IngestWorkerConfig, raw io.Reader) io.Reader { } // 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 { +// the streamer key PEM + cert straight to muxl-sign, no MediaSigner / model / DB +// needed. The manifest is read FRESH per GoP from the holder, so a manifest main +// pushes mid-stream (e.g. pre-live → live) takes effect on the next GoP — the +// same fresh-per-GoP shape as the in-process signer. Shared by the MKV and WHIP +// workers. +func workerSignStream(cfg IngestWorkerConfig, getManifest func() []byte) SignSegmentStreamFunc { return func(ctx context.Context, input io.Reader, eventCh chan *muxl.MuxlEvent) error { - fetchManifest := func() ([]byte, error) { return cfg.Manifest, nil } + fetchManifest := func() ([]byte, error) { return getManifest(), nil } return muxl.RunMuxlSignSegment(ctx, input, muxl.SignerInput{ CertPEM: cfg.CertPEM, KeyPEM: cfg.KeyPEM, @@ -163,7 +193,7 @@ func (mm *MediaManager) workerSegmentSink(ctx context.Context, cfg IngestWorkerC // 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 { +func RunMKVIngestWorker(ctx context.Context, cfg IngestWorkerConfig, stdin io.Reader, frames FrameWriter, getManifest func() []byte) error { gstinit.InitGST() ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -196,7 +226,7 @@ func RunMKVIngestWorker(ctx context.Context, cfg IngestWorkerConfig, stdin io.Re }() } - signerElem, done, err := muxlSignSegmentElem(ctx, mm.cli, workerSignStream(cfg), onSegment) + 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 618d170a3..f4f16522d 100644 --- a/pkg/media/ingest_worker_test.go +++ b/pkg/media/ingest_worker_test.go @@ -107,7 +107,7 @@ func TestRunMKVIngestWorkerProducesValidSignedFrames(t *testing.T) { // the signer drain), so reading the buffer single-threaded afterwards is safe. var buf bytes.Buffer frames := ingestframe.NewWriter(&buf) - require.NoError(t, RunMKVIngestWorker(ctx, cfg, bytes.NewReader(mkv), frames)) + require.NoError(t, RunMKVIngestWorker(ctx, cfg, bytes.NewReader(mkv), frames, func() []byte { return cfg.Manifest })) r := ingestframe.NewReader(&buf) var segs int @@ -172,7 +172,7 @@ func TestRunMKVIngestWorkerRecords(t *testing.T) { mkv := makeH264AACMKV(t, ctx, getFixture("5sec.mp4")) // Frames are irrelevant here (recording is on the input side); discard them. - require.NoError(t, RunMKVIngestWorker(ctx, cfg, bytes.NewReader(mkv), ingestframe.NewWriter(io.Discard))) + require.NoError(t, RunMKVIngestWorker(ctx, cfg, bytes.NewReader(mkv), ingestframe.NewWriter(io.Discard), func() []byte { return cfg.Manifest })) // The recording lands at debug-recordings//.rtmp.mkv and // must contain exactly the media the worker ingested. The dump goroutine @@ -221,7 +221,7 @@ func TestRunMKVIngestWorkerSelfWatchdog(t *testing.T) { start := time.Now() done := make(chan error, 1) go func() { - done <- RunMKVIngestWorker(ctx, cfg, bytes.NewReader(wedge), ingestframe.NewWriter(io.Discard)) + done <- RunMKVIngestWorker(ctx, cfg, bytes.NewReader(wedge), ingestframe.NewWriter(io.Discard), func() []byte { return cfg.Manifest }) }() select { case <-done: @@ -232,3 +232,45 @@ func TestRunMKVIngestWorkerSelfWatchdog(t *testing.T) { t.Fatal("worker-side watchdog did not contain the wedge (RunMKVIngestWorker hung)") } } + +// TestRunMKVIngestWorkerSignsWithSuppliedManifest is the core of the pre-live → +// live fix: the worker signs each GoP with whatever the manifest getter returns, +// NOT a frozen one. Two runs of the same media differ only in the getter — a +// pre-live manifest yields unpublished segments; a live manifest (the same plus +// a c2pa.published action, as main would push on go-live) yields published ones. +func TestRunMKVIngestWorkerSignsWithSuppliedManifest(t *testing.T) { + ctx := context.Background() + ms := newBareSegmentSigner(t) + keyPEM, err := signers.MarshalES256KPrivateKeyPEM(ms.Signer) + require.NoError(t, err) + cfg := IngestWorkerConfig{ + StreamerDID: ms.Streamer(), + KeyPEM: keyPEM, + CertPEM: ms.Cert, + BroadcasterHost: "test.example.com", + } + + prelive := ms.PrebuiltManifest // c2pa.created only → unpublished + live := bytes.Replace(prelive, + []byte(`{"action":"c2pa.created"}`), + []byte(`{"action":"c2pa.created"},{"action":"c2pa.published"}`), 1) + require.NotEqual(t, string(prelive), string(live), "the live manifest must add c2pa.published") + + mkv := makeH264AACMKV(t, ctx, getFixture("5sec.mp4")) + + firstSegmentPublished := func(manifest []byte) bool { + var buf bytes.Buffer + require.NoError(t, RunMKVIngestWorker(ctx, cfg, bytes.NewReader(mkv), + ingestframe.NewWriter(&buf), func() []byte { return manifest })) + r := ingestframe.NewReader(&buf) + typ, payload, rerr := r.ReadFrame() + require.NoError(t, rerr) + require.Equal(t, ingestframe.Segment, typ) + res, verr := ValidateMP4Media(ctx, payload) + require.NoError(t, verr) + return res.Meta.Published + } + + require.False(t, firstSegmentPublished(prelive), "pre-live manifest → unpublished segments") + require.True(t, firstSegmentPublished(live), "live manifest → published segments (the fix)") +} diff --git a/pkg/media/manifest_refresh_test.go b/pkg/media/manifest_refresh_test.go new file mode 100644 index 000000000..27ae898d7 --- /dev/null +++ b/pkg/media/manifest_refresh_test.go @@ -0,0 +1,81 @@ +package media + +import ( + "context" + "net" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/ingestframe" +) + +// TestPushManifestUpdatesOnChange checks the main side: pushManifestUpdates sends +// a Manifest frame for the initial manifest and again whenever it changes (e.g. +// a pre-live → live transition), but not while it's unchanged. +func TestPushManifestUpdatesOnChange(t *testing.T) { + old := manifestRefreshInterval + manifestRefreshInterval = 20 * time.Millisecond + defer func() { manifestRefreshInterval = old }() + + mainConn, workerConn := net.Pipe() + defer mainConn.Close() + defer workerConn.Close() + + var mu sync.Mutex + cur := []byte("manifest-prelive") + source := func() ([]byte, error) { + mu.Lock() + defer mu.Unlock() + return append([]byte(nil), cur...), nil + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go pushManifestUpdates(ctx, mainConn, source) + + fr := ingestframe.NewReader(workerConn) + readManifest := func() string { + t.Helper() + typ, payload, err := fr.ReadFrame() + require.NoError(t, err) + require.Equal(t, ingestframe.Manifest, typ) + return string(payload) + } + + require.Equal(t, "manifest-prelive", readManifest(), "initial manifest is pushed") + + mu.Lock() + cur = []byte("manifest-live-published") + mu.Unlock() + // The next frame is the changed manifest — unchanged polls in between don't + // emit a frame, so this read can only return the new content. + require.Equal(t, "manifest-live-published", readManifest(), "a changed manifest is pushed") +} + +// TestServeFrameSocketAppliesManifestUpdate checks the worker side: a Manifest +// control frame from main swaps the worker's manifest holder, so the signer +// picks it up on the next GoP. +func TestServeFrameSocketAppliesManifestUpdate(t *testing.T) { + sock := filepath.Join(t.TempDir(), "w.sock") + ln, err := net.Listen("unix", sock) + require.NoError(t, err) + defer ln.Close() + + holder := newManifestHolder([]byte("prelive")) + srv := newFrameServer(workerFrameBuffer) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go serveFrameSocket(ctx, ln, srv, holder) + + client, err := net.Dial("unix", sock) + require.NoError(t, err) + defer client.Close() + require.NoError(t, ingestframe.NewWriter(client).Manifest([]byte("live-published"))) + + require.Eventually(t, func() bool { + return string(holder.get()) == "live-published" + }, 3*time.Second, 20*time.Millisecond, "worker swaps to the manifest main pushed") +} diff --git a/pkg/media/whip_worker.go b/pkg/media/whip_worker.go index f8270b8ac..de598bcbe 100644 --- a/pkg/media/whip_worker.go +++ b/pkg/media/whip_worker.go @@ -43,7 +43,8 @@ func ServeWHIPIngestWorkerSocket(ctx context.Context, cfg IngestWorkerConfig) er }() srv := newFrameServer(workerFrameBuffer) - go serveFrameSocket(ctx, ln, srv) + manifest := newManifestHolder(cfg.Manifest) + go serveFrameSocket(ctx, ln, srv, manifest) // finish flushes the trailing End/Error, waits for main to drain the buffer // (incl. the Answer), then closes the connection for a clean EOF. @@ -84,7 +85,7 @@ func ServeWHIPIngestWorkerSocket(ctx context.Context, cfg IngestWorkerConfig) er defer wd.stop() onSegment, flush := mm.workerSegmentSink(ctx, cfg, wd.wrap(srv)) - signerElem, signerDone, err := muxlSignSegmentElem(ctx, mm.cli, workerSignStream(cfg), onSegment) + signerElem, signerDone, err := muxlSignSegmentElem(ctx, mm.cli, workerSignStream(cfg, manifest.get), onSegment) if err != nil { return finish(fmt.Errorf("build signer element: %w", err)) } -- 2.51.2