diff --git a/pkg/api/duplicate_mist.go b/pkg/api/duplicate_mist.go index df0349a0e..ce71a02d9 100644 --- a/pkg/api/duplicate_mist.go +++ b/pkg/api/duplicate_mist.go @@ -7,7 +7,7 @@ import ( "fmt" "io" "net" - "sync/atomic" + "sync" "time" "github.com/bluenviron/gortmplib" @@ -24,22 +24,68 @@ func (replayRTMP) Write(p []byte) (int, error) { return len(p), nil } var errShadowIdleTimeout = errors.New("shadow RTMP input idle timeout") +// Only a pending input Read is idle. Parsing buffered bytes, initializing native +// media, and draining after EOF must not consume the publisher's idle allowance. type shadowIdleReader struct { io.Reader - timer *time.Timer + mu sync.Mutex + timer *time.Timer + deadline time.Time + timedOut bool + cancel context.CancelFunc } -func (r shadowIdleReader) Read(p []byte) (int, error) { +func newShadowIdleReader(input io.Reader, cancel context.CancelFunc) *shadowIdleReader { + r := &shadowIdleReader{Reader: input, cancel: cancel} + r.timer = time.AfterFunc(RTMPTimeout, r.expire) + r.timer.Stop() + return r +} + +func (r *shadowIdleReader) expire() { + r.mu.Lock() + defer r.mu.Unlock() + if r.deadline.IsZero() { + return + } + if remaining := time.Until(r.deadline); remaining > 0 { + // Stop cannot retract a callback already queued by an earlier Read. + // That callback must honor the current Read's full idle allowance. + r.timer.Reset(remaining) + return + } + r.timedOut = true + r.cancel() +} + +func (r *shadowIdleReader) Read(p []byte) (int, error) { + r.mu.Lock() + if r.timedOut { + r.mu.Unlock() + return 0, errShadowIdleTimeout + } + r.deadline = time.Now().Add(RTMPTimeout) + r.timer.Reset(RTMPTimeout) + r.mu.Unlock() n, err := r.Reader.Read(p) - if err != nil { - // Publisher EOF starts downstream drain, not another input deadline. - r.timer.Stop() - } else if n > 0 { - r.timer.Reset(RTMPTimeout) + r.mu.Lock() + r.deadline = time.Time{} + r.timer.Stop() + if r.timedOut { + err = errShadowIdleTimeout } + r.mu.Unlock() return n, err } +func (r *shadowIdleReader) stop() bool { + r.mu.Lock() + defer r.mu.Unlock() + r.deadline = time.Time{} + r.timer.Stop() + return r.timedOut +} + // RunDuplicateMistWorker consumes the decrypted, client-only RTMP conversation. // gortmplib's non-strict plain handshake consumes C2 without comparing it to // the shadow's S1, so the publisher's response to Mist's handshake is valid. @@ -57,19 +103,14 @@ func RunDuplicateMistWorker(parent context.Context, input io.Reader, progress io } }) defer stopInput() - var timedOut atomic.Bool - idle := time.AfterFunc(RTMPTimeout, func() { - timedOut.Store(true) - cancel() - }) + inputReader := newShadowIdleReader(input, cancel) defer func() { - idle.Stop() - if timedOut.Load() { + if inputReader.stop() { retErr = errShadowIdleTimeout } }() - br := bufio.NewReader(shadowIdleReader{Reader: input, timer: idle}) + br := bufio.NewReader(inputReader) version, err := br.Peek(1) if err != nil { return fmt.Errorf("shadow RTMP handshake: %w", err) diff --git a/pkg/api/duplicate_mist_test.go b/pkg/api/duplicate_mist_test.go index df228d27b..18c45b102 100644 --- a/pkg/api/duplicate_mist_test.go +++ b/pkg/api/duplicate_mist_test.go @@ -2,14 +2,66 @@ package api import ( "bytes" + "context" "fmt" "io" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/require" ) +func TestShadowIdleReaderAllowsConsumerWork(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + reader := newShadowIdleReader(bytes.NewReader([]byte{3}), cancel) + defer reader.stop() + var buf [1]byte + n, err := reader.Read(buf[:]) + require.NoError(t, err) + require.Equal(t, 1, n) + + // A caller processing bytes is not waiting on idle publisher input. + time.Sleep(2 * RTMPTimeout) + require.NoError(t, ctx.Err()) + n, err = reader.Read(buf[:]) + require.ErrorIs(t, err, io.EOF) + require.Zero(t, n) + time.Sleep(2 * RTMPTimeout) + require.NoError(t, ctx.Err(), "publisher EOF must allow downstream drain") + }) +} + +func TestShadowIdleReaderStaleCallbackHonorsPendingRead(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + input, publisher := io.Pipe() + defer input.Close() + defer publisher.Close() + stopInput := context.AfterFunc(ctx, func() { _ = input.Close() }) + defer stopInput() + reader := newShadowIdleReader(input, cancel) + defer reader.stop() + result := make(chan error, 1) + go func() { + var buf [1]byte + _, err := reader.Read(buf[:]) + result <- err + }() + synctest.Wait() + time.Sleep(RTMPTimeout / 2) + // A previous timer callback can already be queued when a new Read starts. + reader.expire() + require.NoError(t, ctx.Err(), "an old callback must not shorten this Read") + time.Sleep(RTMPTimeout / 2) + require.ErrorIs(t, <-result, errShadowIdleTimeout) + require.ErrorIs(t, ctx.Err(), context.Canceled) + }) +} + func TestDuplicateMistIdleHandshakeClosesOnlyShadowInput(t *testing.T) { input, publisher := io.Pipe() defer input.Close() diff --git a/pkg/cmd/duplicate_mist_test.go b/pkg/cmd/duplicate_mist_test.go index f45657477..31e90548e 100644 --- a/pkg/cmd/duplicate_mist_test.go +++ b/pkg/cmd/duplicate_mist_test.go @@ -19,6 +19,8 @@ import ( "time" "github.com/bluenviron/gortmplib" + "github.com/bluenviron/gortmplib/pkg/bytecounter" + "github.com/bluenviron/gortmplib/pkg/message" "github.com/bluenviron/gortsplib/v5/pkg/format" "github.com/go-gst/go-gst/gst" "github.com/prometheus/client_golang/prometheus/promhttp" @@ -251,16 +253,36 @@ func TestDuplicateMistEndToEnd(t *testing.T) { stop := context.AfterFunc(ctx, func() { client.Close(); server.Close() }) defer stop() replies := make(chan struct{}) - go func() { _, _ = io.Copy(client, server); client.Close(); close(replies) }() + go func() { + _, _ = io.Copy(client, server) + client.Close() + // The encoder can finish before the TLS publisher's EOF gate opens. + _, _ = io.Copy(io.Discard, server) + close(replies) + }() forwarded, err := io.Copy(server, client) if err == nil { - // Keep the actual TLS publisher connected until the parent has - // observed live verification, even if encoding already reached EOS. - select { - case <-publisherEOF: - case <-ctx.Done(): - err = ctx.Err() - } + // Keep the actual TLS publisher active until live verification, + // even when encoding reaches EOS before a slow worker catches up. + counter := bytecounter.NewWriter(server) + keepalive := message.NewWriter(counter, counter, false) + err = func() error { + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + for { + select { + case <-publisherEOF: + return nil + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + if err := keepalive.Write(&message.UserControlPingRequest{}); err != nil { + return err + } + } + } + }() + forwarded += int64(counter.Count()) } if corrupt && err == nil { // Invalid RTMP after verified media must produce a failure. @@ -296,6 +318,13 @@ func TestDuplicateMistEndToEnd(t *testing.T) { busDone := make(chan error, 1) go func() { busDone <- media.HandleBusMessages(ctx, pipeline) }() require.NoError(t, pipeline.SetState(gst.StatePlaying)) + select { + case err := <-busDone: + require.NoError(t, err) + case <-ctx.Done(): + t.Fatal("publisher timed out") + } + require.NoError(t, pipeline.SetState(gst.StateNull)) if paced { // The TLS publisher has not sent EOF; progress cannot be deferred // until worker completion. Encoding speed is irrelevant. @@ -305,13 +334,6 @@ func TestDuplicateMistEndToEnd(t *testing.T) { } close(publisherEOF) select { - case err := <-busDone: - require.NoError(t, err) - case <-ctx.Done(): - t.Fatal("publisher timed out") - } - require.NoError(t, pipeline.SetState(gst.StateNull)) - select { case err := <-primaryResult: require.NoError(t, err) case <-ctx.Done(): diff --git a/pkg/rtmps/shadow.go b/pkg/rtmps/shadow.go index 05e73570b..7121185e0 100644 --- a/pkg/rtmps/shadow.go +++ b/pkg/rtmps/shadow.go @@ -69,10 +69,12 @@ publisher identifiers or stream keys; use the failure logs for details. The shadow has an 8 MiB per-connection buffer and a cap of 16 workers per addon listener, including draining workers. At capacity, new connections -forward only to Mist and log a shadow failure. Shadow input idle for 10 -seconds is aborted without closing the primary connection. Buffer overflow -also kills only the shadow; Mist continues unchanged. On disconnect it -drains the native pipeline through EOS and verifies the final segment. +forward only to Mist and log a shadow failure. A pending shadow input read +idle for 10 seconds is aborted without closing the primary connection. +Parsing buffered input, native initialization, and signing do not consume +this idle allowance. Buffer overflow also kills only the shadow; Mist +continues unchanged. On disconnect it drains the native pipeline through +EOS and verifies the final segment. The playback handshake deadline is cleared for this drain; the 30-second worker-exit deadline still applies. Node shutdown kills/reaps workers. This is opt-in diagnostic work with additional CPU and memory cost, up to @@ -84,9 +86,11 @@ The backend end-to-end regression is `TestDuplicateMistEndToEnd` in through verified segments beyond the former few-second failure interval, including a publisher outside `/live/` and all final segments after paced and unpaced publisher EOF. -The fixture holds TLS publisher EOF until live metrics show verification, -so slow worker startup is not confused with a metrics failure. Corrupt RTMP -must export a failure while subsequent primary traffic still arrives. +The fixture stops the encoder at EOS but holds TLS publisher EOF until live +metrics show verification, sending RTMP keepalives while it waits. Slow +worker startup or signing is not confused with idle publisher input. +Corrupt RTMP must export a failure while subsequent primary traffic still +arrives. No client UI or platform-specific iOS/Android/Web behavior changes. */