Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
16 kB · 413 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414package media
import ( "bytes" "context" "errors" "fmt" "io" "net/http" "net/http/httptest" "os" "path/filepath" "strings" "sync" "testing" "time"
"github.com/go-gst/go-gst/gst" "github.com/go-gst/go-gst/gst/app" "github.com/stretchr/testify/require" "stream.place/streamplace/pkg/crypto/signers" "stream.place/streamplace/pkg/gstinit" "stream.place/streamplace/pkg/ingestframe" "stream.place/streamplace/pkg/muxl" "stream.place/streamplace/pkg/s3")
// TestWorkerInputDeframes checks the body-deframing the worker applies to the// fd-passed push connection: prebuf (bytes main read past the headers) is// prepended, and a chunked transfer-encoding is decoded back to the raw media.func TestWorkerInputDeframes(t *testing.T) { payload := []byte("the-actual-media-bytes-pretend-this-is-mp4-data") // A textbook chunked body: one chunk then the zero terminator. body := []byte(fmt.Sprintf("%x\r\n%s\r\n0\r\n\r\n", len(payload), payload))
// prebuf = the slice main already read; the rest is still on the fd. cfg := IngestWorkerConfig{Chunked: true, Prebuf: append([]byte(nil), body[:5]...)} got, err := io.ReadAll(WorkerInput(cfg, bytes.NewReader(body[5:]))) require.NoError(t, err) require.Equal(t, payload, got, "prebuf + chunked fd de-frames to the original media")
// Raw (no prebuf, not chunked) passes through unchanged. got, err = io.ReadAll(WorkerInput(IngestWorkerConfig{}, bytes.NewReader(payload))) require.NoError(t, err) require.Equal(t, payload, got)}
// makeH264AACFMP4 builds a clean, 1-video-1-audio fragmented H264+AAC MP4 from// an H264+Opus MP4 fixture (video passed through, audio transcoded Opus→AAC) —// exactly the shape MistServer's live .mp4 output delivers for an RTMP push.func makeH264AACFMP4(t *testing.T, ctx context.Context, srcMP4 string) []byte { t.Helper() gstinit.InitGST() desc := strings.Join([]string{ "filesrc location=" + srcMP4 + " ! qtdemux name=d", "d. ! queue ! h264parse ! mp4mux name=mux fragment-duration=500 ! appsink name=sink", "d. ! queue ! opusdec ! audioconvert ! audioresample ! fdkaacenc ! aacparse ! mux.", }, "\n") pipeline, err := gst.NewPipelineFromString(desc) require.NoError(t, err)
sinkEle, err := pipeline.GetElementByName("sink") require.NoError(t, err) var buf bytes.Buffer app.SinkFromElement(sinkEle).SetCallbacks(&app.SinkCallbacks{ NewSampleFunc: WriterNewSample(ctx, &buf), })
busErr := make(chan error, 1) go func() { busErr <- HandleBusMessages(ctx, pipeline) }() require.NoError(t, pipeline.SetState(gst.StatePlaying)) defer func() { _ = pipeline.SetState(gst.StateNull) }() require.NoError(t, <-busErr, "remux to fragmented H264+AAC MP4") require.NotEmpty(t, buf.Bytes(), "remux produced an fMP4") return buf.Bytes()}
// makeAudioOnlyAACFMP4 synthesizes an AAC-audio-only fragmented MP4 — the// canonical WEDGE input for watchdog/containment tests. The ingest pipeline// hardwires a video and an audio branch; with no video track, qtdemux never// creates a video pad, the fMP4 aggregator's video pad never sees data OR// EOS, and the pipeline hangs forever with no frames and no EOS — a true// native wedge that no queue sizing can fix.func makeAudioOnlyAACFMP4(t *testing.T, ctx context.Context, seconds int) []byte { t.Helper() gstinit.InitGST() desc := fmt.Sprintf("audiotestsrc num-buffers=%d samplesperbuffer=1024 ! audio/x-raw,rate=48000,channels=2 ! audioconvert ! fdkaacenc ! aacparse ! mp4mux fragment-duration=500 ! appsink name=sink", seconds*47) pipeline, err := gst.NewPipelineFromString(desc) require.NoError(t, err)
sinkEle, err := pipeline.GetElementByName("sink") require.NoError(t, err) var buf bytes.Buffer app.SinkFromElement(sinkEle).SetCallbacks(&app.SinkCallbacks{ NewSampleFunc: WriterNewSample(ctx, &buf), })
busErr := make(chan error, 1) go func() { busErr <- HandleBusMessages(ctx, pipeline) }() require.NoError(t, pipeline.SetState(gst.StatePlaying)) defer func() { _ = pipeline.SetState(gst.StateNull) }() require.NoError(t, <-busErr, "synthesize audio-only fMP4") require.NotEmpty(t, buf.Bytes()) return buf.Bytes()}
// TestRunMP4IngestWorkerProducesValidSignedFrames drives the isolated ingest// worker's core directly (no subprocess): feed it an H264+AAC fMP4, collect the// framed output, and verify every emitted segment is a valid signed canonical// .m4s. This is the contract the supervisor relies on — frames it can hand// straight to ValidateMP4. (The real subprocess spawn + fault injection is// Stage 3.)func TestRunMP4IngestWorkerProducesValidSignedFrames(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)
// Provide the node transcode signer (the same test key serves both roles // here, as in the transcoder tests) so the worker completes to dual-codec. cfg := IngestWorkerConfig{ StreamerDID: ms.Streamer(), KeyPEM: keyPEM, CertPEM: ms.Cert, Manifest: manifest, NodeCertPEM: ms.Cert, NodeKeyPEM: keyPEM, BroadcasterHost: "test.example.com", }
mp4 := makeH264AACFMP4(t, ctx, getFixture("5sec.mp4"))
// All frame writes complete before RunMP4IngestWorker returns (it waits on // the signer drain), so reading the buffer single-threaded afterwards is safe. var buf bytes.Buffer frames := ingestframe.NewWriter(&buf) require.NoError(t, RunMP4IngestWorker(ctx, cfg, bytes.NewReader(mp4), frames, 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, "worker emits only Segment frames; End is the subcommand's job") require.NotEmpty(t, payload)
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)
// With a node key the worker completes to dual-codec: every segment must // carry both the source Opus and a worker-transcoded AAC track. codecs := audioCodecsOf(t, ctx, payload) hasOpus, hasAAC := false, false for _, c := range codecs { if isOpusCodec(c) { hasOpus = true } if isAACCodec(c) { hasAAC = true } } require.True(t, hasOpus, "segment %d keeps source Opus (got %v)", segs, codecs) require.True(t, hasAAC, "segment %d gains worker-transcoded AAC (got %v)", segs, codecs) segs++ } require.GreaterOrEqual(t, segs, 1, "worker emitted at least one signed dual-codec segment") t.Logf("worker emitted %d valid dual-codec segments", segs)}
// 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/<did>/<ts>.rtmp.mp4. This is what keeps debug// recording working on the isolated paths where main is out of the data path// (it can't tee the bytes itself), so main decides and the worker records.func TestRunMP4IngestWorkerRecords(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)
dataDir := t.TempDir() cfg := IngestWorkerConfig{ StreamerDID: ms.Streamer(), KeyPEM: keyPEM, CertPEM: ms.Cert, Manifest: manifest, BroadcasterHost: "test.example.com", Record: true, DataDir: dataDir, }
mp4 := makeH264AACFMP4(t, ctx, getFixture("5sec.mp4"))
// Frames are irrelevant here (recording is on the input side); discard them. require.NoError(t, RunMP4IngestWorker(ctx, cfg, bytes.NewReader(mp4), ingestframe.NewWriter(io.Discard), func() []byte { return cfg.Manifest }))
// The recording lands at debug-recordings/<sanitized-did>/<ts>.rtmp.mp4 and // must contain exactly the media the worker ingested. The dump goroutine // flushes asynchronously, so allow it a moment to finish the last write. glob := filepath.Join(dataDir, "debug-recordings", "*", "*.rtmp.mp4") require.Eventually(t, func() bool { matches, _ := filepath.Glob(glob) if len(matches) != 1 { return false } got, rerr := os.ReadFile(matches[0]) return rerr == nil && bytes.Equal(got, mp4) }, 10*time.Second, 25*time.Millisecond, "worker records the ingest media verbatim")}
// fakeS3Server is a minimal path-style S3 endpoint speaking just enough of the// multipart-upload protocol for UploadWriter: initiate → upload parts →// complete. Completed objects land in objects keyed by "<bucket>/<key>".type fakeS3Server struct { mu sync.Mutex parts map[string][]byte // "<path>#<partNumber>" → body objects map[string][]byte // completed "<bucket>/<key>" → body}
func newFakeS3Server() *fakeS3Server { return &fakeS3Server{parts: map[string][]byte{}, objects: map[string][]byte{}}}
func (f *fakeS3Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { f.mu.Lock() defer f.mu.Unlock() path := strings.TrimPrefix(r.URL.Path, "/") q := r.URL.Query() switch { case r.Method == "POST" && q.Has("uploads"): fmt.Fprintf(w, `<InitiateMultipartUploadResult><UploadId>test-upload</UploadId></InitiateMultipartUploadResult>`) case r.Method == "PUT" && q.Has("partNumber"): body, _ := io.ReadAll(r.Body) f.parts[path+"#"+q.Get("partNumber")] = body w.Header().Set("ETag", `"part-`+q.Get("partNumber")+`"`) case r.Method == "POST" && q.Has("uploadId"): var buf []byte for i := 1; ; i++ { part, ok := f.parts[fmt.Sprintf("%s#%d", path, i)] if !ok { break } buf = append(buf, part...) } f.objects[path] = buf fmt.Fprintf(w, `<CompleteMultipartUploadResult><Key>%s</Key></CompleteMultipartUploadResult>`, path) default: w.WriteHeader(http.StatusBadRequest) }}
// objectWithPrefix finds a completed object whose key starts with prefix and// ends with suffix (the recording's timestamped filename isn't predictable).func (f *fakeS3Server) objectWithPrefix(prefix, suffix string) ([]byte, bool) { f.mu.Lock() defer f.mu.Unlock() for path, b := range f.objects { if strings.HasPrefix(path, prefix) && strings.HasSuffix(path, suffix) { return b, true } } return nil, false}
// TestRunMP4IngestWorkerRecordsToS3 proves the debug recording streams to S3// when main hands its S3 config over the handshake (cfg.S3) — the production// shape. Without that plumbing the worker's minimal CLI has no S3 fields and// DebugRecordingCreate silently falls back to local disk, which is exactly the// regression this guards against: recordings must land in the bucket, not under// DataDir.func TestRunMP4IngestWorkerRecordsToS3(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)
fake := newFakeS3Server() srv := httptest.NewServer(fake) defer srv.Close()
dataDir := t.TempDir() cfg := IngestWorkerConfig{ StreamerDID: ms.Streamer(), KeyPEM: keyPEM, CertPEM: ms.Cert, Manifest: manifest, BroadcasterHost: "test.example.com", Record: true, DataDir: dataDir, S3: &s3.Config{ Endpoint: srv.URL, Bucket: "debug-bucket", AccessKeyID: "test-access", SecretAccessKey: "test-secret", Region: "auto", }, }
mp4 := makeH264AACFMP4(t, ctx, getFixture("5sec.mp4"))
require.NoError(t, RunMP4IngestWorker(ctx, cfg, bytes.NewReader(mp4), ingestframe.NewWriter(io.Discard), func() []byte { return cfg.Manifest }))
// The upload commits asynchronously (the dump goroutine's Close); the object // must appear at debug-bucket/debug-recordings/<did>/<ts>.rtmp.mp4 holding // exactly the ingested media. wantPrefix := "debug-bucket/debug-recordings/" + ms.Streamer() + "/" require.Eventually(t, func() bool { got, ok := fake.objectWithPrefix(wantPrefix, ".rtmp.mp4") return ok && bytes.Equal(got, mp4) }, 10*time.Second, 25*time.Millisecond, "worker streams the recording to the S3 bucket verbatim")
// And nothing fell back to local disk. matches, _ := filepath.Glob(filepath.Join(dataDir, "debug-recordings", "*", "*")) require.Empty(t, matches, "recording must go to S3, not DataDir")}
// TestRunMP4IngestWorkerSelfWatchdog proves the worker's OWN watchdog contains a// wedge. This is the only wedge containment on the detached/WHIP paths, where// main can't kill a detached worker — so the worker has to notice it's stuck and// exit itself. An audio-only fMP4 starves the muxer's video pad of both data and// EOS, so the pipeline wedges with no frames; the watchdog must tear it down// and return rather than hang forever. (The fd-4 path's supervisor-side// watchdog is covered separately by TestMP4IngestIsolatedWedgeContained.)func TestRunMP4IngestWorkerSelfWatchdog(t *testing.T) { old := ingestWorkerWatchdog ingestWorkerWatchdog = 3 * time.Second defer func() { ingestWorkerWatchdog = old }()
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, BroadcasterHost: "test.example.com", }
wedge := makeAudioOnlyAACFMP4(t, ctx, 5)
start := time.Now() done := make(chan error, 1) go func() { done <- RunMP4IngestWorker(ctx, cfg, bytes.NewReader(wedge), ingestframe.NewWriter(io.Discard), func() []byte { return cfg.Manifest }) }() select { case <-done: elapsed := time.Since(start) require.Less(t, elapsed, 25*time.Second, "watchdog bounded the wedge") t.Logf("worker self-terminated on wedge in %s", elapsed.Round(time.Second)) case <-time.After(30 * time.Second): t.Fatal("worker-side watchdog did not contain the wedge (RunMP4IngestWorker hung)") }}
// TestRunMP4IngestWorkerSignsWithSuppliedManifest 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 TestRunMP4IngestWorkerSignsWithSuppliedManifest(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")
mp4 := makeH264AACFMP4(t, ctx, getFixture("5sec.mp4"))
firstSegmentPublished := func(manifest []byte) bool { var buf bytes.Buffer require.NoError(t, RunMP4IngestWorker(ctx, cfg, bytes.NewReader(mp4), 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)")}