diff --git a/go.mod b/go.mod index ad90ee77..be1d8449 100644 --- a/go.mod +++ b/go.mod @@ -38,6 +38,7 @@ require ( github.com/golangci/golangci-lint/v2 v2.1.6 github.com/google/uuid v1.6.0 github.com/gorilla/websocket v1.5.3 + github.com/hyphacoop/go-dasl v0.8.0 github.com/ipfs/go-cid v0.5.0 github.com/ipfs/go-ipld-cbor v0.2.0 github.com/ipld/go-car v0.6.1-0.20230509095817-92d28eb23ba4 @@ -312,7 +313,6 @@ require ( github.com/hexops/gotextdiff v1.0.3 // indirect github.com/holiman/uint256 v1.3.2 // indirect github.com/hyphacoop/cbor/v2 v2.0.0-20251007204234-2a4fa83e606e // indirect - github.com/hyphacoop/go-dasl v0.8.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/invopop/yaml v0.3.1 // indirect github.com/ipfs/bbloom v0.0.4 // indirect diff --git a/pkg/ingestframe/frame.go b/pkg/ingestframe/frame.go index 783293f6..22e52a42 100644 --- a/pkg/ingestframe/frame.go +++ b/pkg/ingestframe/frame.go @@ -1,26 +1,37 @@ // Package ingestframe defines the wire protocol a per-stream ingest worker uses -// to stream canonical MUXL fragments back to the main streamplace process. +// to stream canonical MUXL fragments (and status) back to the main streamplace +// process. // // Each incoming live stream is handled by an isolated worker subprocess that -// owns the socket, muxes + transcodes the media, and signs each GoP. It emits -// the resulting signed canonical .m4s segments to the main process as a sequence -// of typed, length-prefixed frames. +// owns the socket, muxes + transcodes + signs the media, and emits the resulting +// signed canonical .m4s segments to the main process as a sequence of typed +// messages. The same channel carries control messages (a clean end, a fatal +// error, a WHIP SDP answer, a status event). // -// The framing is deliberately transport-agnostic: today it rides the worker's -// stdout pipe, but a detached / reattachable worker (the zero-downtime-upgrade -// path, where workers keep buffering signed segments across a main restart) can -// carry the identical frames over a unix socket. Nothing above this package -// cares which. +// The wire format is a stream of concatenated DRISL CBOR items — the same codec +// muxl uses for its own stdio protocol. CBOR data items are self-delimiting (the +// length/count lives in each item's head), so no separate length prefix is +// needed, and a decoder reads exactly one item per call. Crucially this PRESERVES +// the crash-vs-clean-end signal the supervisor relies on: the decoder returns +// io.EOF at an item boundary (the stream ended cleanly between messages) and +// io.ErrUnexpectedEOF mid-item (a worker that died). A garbage/desynced stream +// fails to decode rather than being mis-parsed. +// +// The format is transport-agnostic: today it rides the worker's stdout pipe or a +// per-session unix socket (the zero-downtime detach/reattach path, where workers +// keep buffering signed segments across a main restart). Nothing above this +// package cares which. package ingestframe import ( - "encoding/binary" "fmt" "io" "sync" + + "github.com/hyphacoop/go-dasl/drisl" ) -// Type identifies a frame's payload. +// Type identifies a message's payload. type Type uint8 const ( @@ -28,13 +39,14 @@ const ( // ValidateMP4 ingests. Payload: the bare canonical segment bytes. Segment Type = 1 // End signals the worker finished the stream cleanly (graceful EOS). No - // payload. Its ABSENCE before EOF is how main tells a crash from a clean end. + // payload. It's the in-band "done" marker; its absence before EOF (together + // with the worker's exit code) is how main tells a crash from a clean end. End Type = 2 // Error carries a worker-side fatal error message (UTF-8). The worker emits // it just before exiting so main can log a cause, not a bare "worker exited". Error Type = 3 // Answer carries an SDP answer (UTF-8). The WHIP worker owns the - // PeerConnection, so it generates the answer and emits it as the FIRST frame + // PeerConnection, so it generates the answer and emits it as the FIRST message // on the socket; main reads it and returns it to the WHIP client before // consuming segments. Payload: the answer SDP. Answer Type = 4 @@ -62,15 +74,21 @@ func (t Type) String() string { } } -// magic prefixes every frame so a desynced/corrupt stream is caught immediately -// rather than mis-parsed as a length. -var magic = [4]byte{'S', 'P', 'F', '1'} - -// MaxPayload bounds a single frame so a corrupt or hostile length can't make the -// reader allocate unboundedly. Canonical GoP segments are well under this. -const MaxPayload = 64 << 20 // 64 MiB +// message is the on-wire DRISL CBOR item: one self-delimiting map per frame. +// Payload is a CBOR byte string (raw for Segment, UTF-8/JSON for the rest) and is +// omitted entirely for an empty body (e.g. End), so a bodyless frame is just +// {"type": N}. +type message struct { + Type Type `cbor:"type"` + Payload []byte `cbor:"payload,omitempty"` +} -const headerSize = 4 + 1 + 4 // magic + type + uint32 length +// frameDecoder is the streaming-decode surface we need (satisfied by drisl's +// *cbor.Decoder). Kept as an interface so this package needn't import the cbor +// module directly. +type frameDecoder interface { + Decode(v any) error +} // Writer serializes frames to an underlying stream. Safe for concurrent use: a // worker emits segments from more than one goroutine (the source signer and the @@ -80,33 +98,23 @@ type Writer struct { w io.Writer } -// NewWriter wraps w. w is typically the worker's os.Stdout. +// NewWriter wraps w. w is typically the worker's frame fd or a socket conn. func NewWriter(w io.Writer) *Writer { return &Writer{w: w} } // WriteFrame writes one whole frame atomically with respect to other WriteFrame -// calls on the same Writer. +// calls on the same Writer. The CBOR item is encoded up front, then written under +// the lock, so concurrent writers never interleave a frame's bytes. func (fw *Writer) WriteFrame(t Type, payload []byte) error { - if len(payload) > MaxPayload { - return fmt.Errorf("ingestframe: payload %d exceeds max %d", len(payload), MaxPayload) + b, err := drisl.Marshal(message{Type: t, Payload: payload}) + if err != nil { + return fmt.Errorf("ingestframe: encode %s: %w", t, err) } - var hdr [headerSize]byte - copy(hdr[0:4], magic[:]) - hdr[4] = byte(t) - binary.BigEndian.PutUint32(hdr[5:9], uint32(len(payload))) - fw.mu.Lock() defer fw.mu.Unlock() - if _, err := fw.w.Write(hdr[:]); err != nil { - return err - } - if len(payload) > 0 { - if _, err := fw.w.Write(payload); err != nil { - return err - } - } - return nil + _, err = fw.w.Write(b) + return err } // Segment frames a signed canonical .m4s segment. @@ -124,44 +132,27 @@ 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) } -// Reader decodes frames from an underlying stream. +// 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 +// read-ahead). type Reader struct { - r io.Reader + dec frameDecoder } -// NewReader wraps r, typically the worker's stdout pipe. +// NewReader wraps r, typically the worker's frame fd or a socket conn. func NewReader(r io.Reader) *Reader { - return &Reader{r: r} + return &Reader{dec: drisl.NewDecoder(r)} } -// ReadFrame decodes the next frame. It returns io.EOF only at a clean frame -// boundary (the stream ended between frames); a stream that dies mid-frame +// ReadFrame decodes the next frame. It returns io.EOF only at a clean item +// boundary (the stream ended between frames); a stream that dies mid-item // surfaces as io.ErrUnexpectedEOF, so an abrupt worker death is distinguishable -// from a clean close. +// from a clean close. A malformed/desynced item surfaces as a decode error. func (fr *Reader) ReadFrame() (Type, []byte, error) { - var hdr [headerSize]byte - if _, err := io.ReadFull(fr.r, hdr[:]); err != nil { - // io.EOF here = clean boundary. io.ReadFull maps a partial read to - // ErrUnexpectedEOF, which we keep: a torn header is an abrupt death. - return 0, nil, err - } - if [4]byte(hdr[0:4]) != magic { - return 0, nil, fmt.Errorf("ingestframe: bad magic %q (stream desynced)", hdr[0:4]) - } - t := Type(hdr[4]) - n := binary.BigEndian.Uint32(hdr[5:9]) - if n > MaxPayload { - return 0, nil, fmt.Errorf("ingestframe: frame length %d exceeds max %d", n, MaxPayload) - } - if n == 0 { - return t, nil, nil - } - payload := make([]byte, n) - if _, err := io.ReadFull(fr.r, payload); err != nil { - if err == io.EOF { - err = io.ErrUnexpectedEOF - } + var m message + if err := fr.dec.Decode(&m); err != nil { return 0, nil, err } - return t, payload, nil + return m.Type, m.Payload, nil } diff --git a/pkg/ingestframe/frame_test.go b/pkg/ingestframe/frame_test.go index a6ce8968..015028ed 100644 --- a/pkg/ingestframe/frame_test.go +++ b/pkg/ingestframe/frame_test.go @@ -2,7 +2,6 @@ package ingestframe import ( "bytes" - "encoding/binary" "errors" "fmt" "io" @@ -52,7 +51,8 @@ func TestRoundTrip(t *testing.T) { // TestTruncatedFrameIsUnexpectedEOF is the crash-vs-clean-end distinction the // supervisor relies on: a worker that dies mid-segment must NOT look like a -// graceful end. +// graceful end. CBOR's self-delimiting framing gives this for free — a byte +// string that declares more bytes than arrive surfaces as ErrUnexpectedEOF. func TestTruncatedFrameIsUnexpectedEOF(t *testing.T) { var buf bytes.Buffer require.NoError(t, NewWriter(&buf).Segment(bytes.Repeat([]byte{1}, 1000))) @@ -65,42 +65,26 @@ func TestTruncatedFrameIsUnexpectedEOF(t *testing.T) { require.ErrorIs(t, err, io.ErrUnexpectedEOF) } -// TestTornHeaderIsUnexpectedEOF: dying partway through the header is also an -// abrupt death, not a clean boundary. +// TestTornHeaderIsUnexpectedEOF: dying partway through the CBOR item head (here +// after the map header byte, before the first key) is also an abrupt death, not +// a clean boundary. func TestTornHeaderIsUnexpectedEOF(t *testing.T) { var buf bytes.Buffer require.NoError(t, NewWriter(&buf).End()) - torn := buf.Bytes()[:headerSize-2] + torn := buf.Bytes()[:1] // just the map header; the rest never arrives _, _, err := NewReader(bytes.NewReader(torn)).ReadFrame() require.ErrorIs(t, err, io.ErrUnexpectedEOF) } -// TestBadMagicRejected: a desynced/corrupt stream is caught, not mis-parsed. -func TestBadMagicRejected(t *testing.T) { - junk := append([]byte("XXXX"), make([]byte, headerSize)...) - _, _, err := NewReader(bytes.NewReader(junk)).ReadFrame() +// TestGarbageRejected: a desynced/corrupt stream is caught, not mis-parsed. A +// complete-but-wrong-shaped CBOR item (a bare integer, not a frame map) must +// surface as a decode error distinct from EOF / a torn frame. +func TestGarbageRejected(t *testing.T) { + _, _, err := NewReader(bytes.NewReader([]byte{0x01})).ReadFrame() require.Error(t, err) - require.Contains(t, err.Error(), "bad magic") -} - -// TestOversizeLengthRejected: a hostile length can't trigger an unbounded alloc. -func TestOversizeLengthRejected(t *testing.T) { - var hdr [headerSize]byte - copy(hdr[0:4], magic[:]) - hdr[4] = byte(Segment) - binary.BigEndian.PutUint32(hdr[5:9], uint32(MaxPayload+1)) - - _, _, err := NewReader(bytes.NewReader(hdr[:])).ReadFrame() - require.Error(t, err) - require.Contains(t, err.Error(), "exceeds max") -} - -// TestWriteOversizeRejected: the writer refuses to emit an over-cap frame. -func TestWriteOversizeRejected(t *testing.T) { - err := NewWriter(io.Discard).Segment(make([]byte, MaxPayload+1)) - require.Error(t, err) - require.Contains(t, err.Error(), "exceeds max") + require.NotErrorIs(t, err, io.EOF) + require.NotErrorIs(t, err, io.ErrUnexpectedEOF) } // TestConcurrentWritesDoNotInterleave: the worker emits segments from multiple diff --git a/pkg/media/ingest_daemon.go b/pkg/media/ingest_daemon.go index 9d9c068b..bf5a0edf 100644 --- a/pkg/media/ingest_daemon.go +++ b/pkg/media/ingest_daemon.go @@ -94,7 +94,9 @@ func (mm *MediaManager) ConsumeWorkerSocket(ctx context.Context, socketPath, str return fmt.Errorf("ingest worker socket gone before End: %w", err) } connectedOnce = true - sawEnd, _ := mm.consumeWorkerFrames(ctx, conn, streamer, onSegment, nil) + // 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) conn.Close() if sawEnd { return nil @@ -192,10 +194,11 @@ func dialWorkerSocket(ctx context.Context, socketPath string) (net.Conn, error) } } -// readWHIPAnswer reads frames on conn until the worker's Answer frame and returns -// its SDP. An Error/End/EOF before the answer is a setup failure. -func readWHIPAnswer(conn net.Conn) (string, error) { - fr := ingestframe.NewReader(conn) +// readWHIPAnswer reads frames until the worker's Answer frame and returns its +// SDP. An Error/End/EOF before the answer is a setup failure. It reads through +// the caller's Reader so the same decoder (and its buffered read-ahead) carries +// on to the segment stream. +func readWHIPAnswer(fr *ingestframe.Reader) (string, error) { for { typ, payload, err := fr.ReadFrame() if err != nil { @@ -248,10 +251,14 @@ func (mm *MediaManager) WHIPIngestDetached(ctx context.Context, offerSDP string, _ = proc.Kill() return "", fmt.Errorf("connect to whip worker: %w", err) } + // One Reader owns this connection for its whole lifetime: the streaming CBOR + // decoder reads ahead, so the Answer and the segments that follow must come + // through the SAME Reader (a second one would lose buffered read-ahead). + fr := ingestframe.NewReader(conn) if dl, ok := answerCtx.Deadline(); ok { _ = conn.SetReadDeadline(dl) } - answer, err := readWHIPAnswer(conn) + answer, err := readWHIPAnswer(fr) if err != nil { conn.Close() _ = proc.Kill() @@ -262,7 +269,7 @@ func (mm *MediaManager) WHIPIngestDetached(ctx context.Context, offerSDP string, // Consume the signed segments in the background; the HTTP handler returns the // answer now and the WebRTC media establishes directly to the worker. go func() { - sawEnd, _ := mm.consumeWorkerFrames(ctx, conn, ms.Streamer(), mm.validateSegment(ctx), nil) + sawEnd, _ := mm.consumeWorkerFrames(ctx, fr, ms.Streamer(), mm.validateSegment(ctx), nil) conn.Close() if !sawEnd && ctx.Err() == nil { // Connection dropped but the detached worker lives on — reconnect and diff --git a/pkg/media/ingest_supervisor.go b/pkg/media/ingest_supervisor.go index cdc7ae70..8ba69db1 100644 --- a/pkg/media/ingest_supervisor.go +++ b/pkg/media/ingest_supervisor.go @@ -144,7 +144,7 @@ func (mm *MediaManager) MKVIngestIsolated(ctx context.Context, input io.Reader, defer watchdog.Stop() // Read signed-segment frames and feed each into the normal chokepoint. - sawEnd, readErr := mm.consumeWorkerFrames(ctx, framesR, ms.Streamer(), mm.validateSegment(ctx), func() { + sawEnd, readErr := mm.consumeWorkerFrames(ctx, ingestframe.NewReader(framesR), ms.Streamer(), mm.validateSegment(ctx), func() { watchdog.Reset(ingestWorkerWatchdog) }) logsWG.Wait() @@ -167,8 +167,7 @@ func (mm *MediaManager) MKVIngestIsolated(ctx context.Context, input io.Reader, // over each. It returns whether a clean End frame was seen and the terminal read // error: nil on a clean close (End then EOF), or io.ErrUnexpectedEOF / a desync // error when the worker died mid-frame. -func (mm *MediaManager) consumeWorkerFrames(ctx context.Context, r io.Reader, streamer string, onSegment func([]byte) error, onProgress func()) (sawEnd bool, _ error) { - fr := ingestframe.NewReader(r) +func (mm *MediaManager) consumeWorkerFrames(ctx context.Context, fr *ingestframe.Reader, streamer string, onSegment func([]byte) error, onProgress func()) (sawEnd bool, _ error) { for { typ, payload, err := fr.ReadFrame() if err != nil { diff --git a/pkg/media/whip_worker_test.go b/pkg/media/whip_worker_test.go index b0ac69fb..717ebe96 100644 --- a/pkg/media/whip_worker_test.go +++ b/pkg/media/whip_worker_test.go @@ -86,7 +86,7 @@ func TestWHIPWorkerAnswersOffer(t *testing.T) { require.NoError(t, derr) defer conn.Close() - answerSDP, rerr := readWHIPAnswer(conn) + answerSDP, rerr := readWHIPAnswer(ingestframe.NewReader(conn)) require.NoError(t, rerr, "worker emits an SDP answer as its first frame") require.Contains(t, answerSDP, "v=0", "valid SDP answer") @@ -194,7 +194,11 @@ func TestWHIPWorkerLoopback(t *testing.T) { require.NoError(t, derr) defer conn.Close() - answerSDP, rerr := readWHIPAnswer(conn) + // One Reader for the whole connection: the streaming decoder reads ahead, so + // the Answer and the segments after it must come through the same Reader (the + // production WHIP path does the same). + fr := ingestframe.NewReader(conn) + answerSDP, rerr := readWHIPAnswer(fr) require.NoError(t, rerr) require.NoError(t, clientPC.SetRemoteDescription(webrtc.SessionDescription{Type: webrtc.SDPTypeAnswer, SDP: answerSDP})) @@ -208,10 +212,9 @@ func TestWHIPWorkerLoopback(t *testing.T) { // Read signed segments; require at least one valid dual-codec one. _ = conn.SetReadDeadline(time.Now().Add(45 * time.Second)) - r := ingestframe.NewReader(conn) var segs int for segs == 0 { - typ, payload, ferr := r.ReadFrame() + typ, payload, ferr := fr.ReadFrame() require.NoError(t, ferr, "reading worker frames") if typ != ingestframe.Segment { continue