diff --git a/pkg/cmd/fmp4push.go b/pkg/cmd/fmp4push.go index 9a60df3e3..9169a44d3 100644 --- a/pkg/cmd/fmp4push.go +++ b/pkg/cmd/fmp4push.go @@ -67,6 +67,10 @@ func pushFMP4(ctx context.Context, endpoint, streamKey, file string) error { req.Header.Set("Content-Type", "video/mp4") var resp *http.Response if resp, err = http.DefaultClient.Do(req); err == nil { + if resp.StatusCode != http.StatusOK { + msg, _ := io.ReadAll(resp.Body) + err = fmt.Errorf("node answered %s: %s", resp.Status, strings.TrimSpace(string(msg))) + } resp.Body.Close() } } diff --git a/pkg/media/muxl_segment.go b/pkg/media/muxl_segment.go index 6cf832c57..d39b85ee7 100644 --- a/pkg/media/muxl_segment.go +++ b/pkg/media/muxl_segment.go @@ -162,6 +162,9 @@ func signSegments(ctx context.Context, cli *config.CLI, signStream SignSegmentSt // 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() + // Closing r on return also releases the copy below if the signer stopped + // reading early. + defer r.Close() go func() { _, err := io.Copy(w, input) w.CloseWithError(err) diff --git a/pkg/media/muxl_segment_test.go b/pkg/media/muxl_segment_test.go new file mode 100644 index 000000000..a847cfd1b --- /dev/null +++ b/pkg/media/muxl_segment_test.go @@ -0,0 +1,50 @@ +package media + +import ( + "context" + "errors" + "io" + "testing" + "time" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/muxl" +) + +// endlessInput is an ingest body that never ends. io.Copy hands it the +// destination through WriteTo, so it sees the moment its writes start failing. +type endlessInput struct{ released chan struct{} } + +func (endlessInput) Read(p []byte) (int, error) { return len(p), nil } + +func (e endlessInput) WriteTo(w io.Writer) (int64, error) { + buf := make([]byte, 1024) + var total int64 + for { + n, err := w.Write(buf) + total += int64(n) + if err != nil { + close(e.released) + return total, err + } + } +} + +// When the signer stops reading before the stream ends (a muxl error), the +// goroutine feeding it the ingest body must not stay blocked forever: every +// failed push would otherwise leak one. +func TestSignFMP4DirectReleasesInputWhenSignerStops(t *testing.T) { + signErr := errors.New("signer failed") + stopEarly := func(context.Context, io.Reader, chan *muxl.MuxlEvent) error { return signErr } + input := endlessInput{released: make(chan struct{})} + + err := signFMP4Direct(context.Background(), &config.CLI{}, stopEarly, input, nil) + require.ErrorIs(t, err, signErr) + + select { + case <-input.released: + case <-time.After(5 * time.Second): + t.Fatal("the input copy is still blocked after signFMP4Direct returned") + } +}