Drain live MPEG-TS input and SRT output cleanly at event end

deps/ffmpeg/9/0019-avformat-libsrt-live-close-drain.patch

@@ -0,0 +1,107 @@
+From eecb04adc1ed5ff7c2a48844e21626da5864e8f8 Mon Sep 17 00:00:00 2001
+From: avplumber patches <patches@local>
+Date: Fri, 9 Oct 2026 14:19:16 +0200
+Subject: [PATCH] avformat/libsrt: drain live publishers before explicit linger
+ close
+
+---
+ doc/protocols.texi   |  8 +++++++
+ libavformat/libsrt.c | 54 +++++++++++++++++++++++++++++++++++++++++++-
+ 2 files changed, 61 insertions(+), 1 deletion(-)
+
+diff --git a/doc/protocols.texi b/doc/protocols.texi
+index dbb1f166c7..26139fe3bd 100644
+--- a/doc/protocols.texi
++++ b/doc/protocols.texi
+@@ -2098,6 +2098,14 @@ Default is -1. -1 means auto (off with 0 seconds in live mode, on with 180
+ seconds in file mode). The range for this option is integers in the
+ 0 - @code{INT_MAX}.
+
++For live outputs, a positive value also bounds orderly completion: wait for
++the sender buffer to empty, then keep the connection open for the negotiated
++peer delivery delay plus two measured round-trip times. This lets the peer
++deliver its final buffered packets before shutdown. The entire close drain
++must fit within @var{seconds}; expiration is reported as a timeout. Zero or
++an unset value preserves immediate live close. This changes only stream
++shutdown, not packet delivery latency during transmission.
++
+ @item tsbpd=@var{1|0}
+ When true, use Timestamp-based Packet Delivery mode. The default behavior
+ depends on the transmission type: enabled in live mode, disabled in file
+diff --git a/libavformat/libsrt.c b/libavformat/libsrt.c
+index c1d93b2550..5e72d520b2 100644
+--- a/libavformat/libsrt.c
++++ b/libavformat/libsrt.c
+@@ -608,16 +608,68 @@ static int libsrt_write(URLContext *h, const uint8_t *buf, int size)
+     return ret;
+ }
+
++/* Positive linger opts a live publisher into a bounded, application-visible
++ * drain. ACKs clear the sender buffer before the peer's TSBPD delivery time;
++ * keep the connection open for its negotiated delay plus two RTTs afterwards.
++ * srt_close() alone is insufficient: SNDSYN is disabled and final cleanup may
++ * destroy asynchronous linger state. See Haivision/srt issue #2760. */
++static int libsrt_drain_live_output(URLContext *h)
++{
++    SRTContext *s = h->priv_data;
++    const int64_t start = av_gettime_relative();
++    const int64_t deadline = start + (int64_t)s->linger * 1000000;
++    int64_t quiet_until = -1;
++    SRT_TRACEBSTATS stats = { 0 };
++    size_t blocks = 0, bytes = 0;
++
++    if (!(h->flags & AVIO_FLAG_WRITE) || s->linger <= 0 ||
++        s->transtype == SRTT_FILE)
++        return 0;
++
++    while (av_gettime_relative() < deadline) {
++        if (ff_check_interrupt(&h->interrupt_callback))
++            return AVERROR_EXIT;
++        if (srt_getsndbuffer(s->fd, &blocks, &bytes) < 0)
++            return libsrt_neterrno(h);
++        if (!blocks) {
++            if (quiet_until < 0) {
++                double quiet_ms;
++                if (srt_bstats(s->fd, &stats, 0) < 0)
++                    return libsrt_neterrno(h);
++                if (!stats.byteSentTotal)
++                    return 0;
++                quiet_ms = FFMAX(stats.msSndTsbPdDelay, 0) +
++                           2.0 * FFMAX(stats.msRTT, 0.0);
++                /* Round up so sub-millisecond RTTs are never truncated. */
++                quiet_until = av_gettime_relative() + (int64_t)(quiet_ms * 1000) + 1000;
++            }
++            if (av_gettime_relative() >= quiet_until) {
++                av_log(h, AV_LOG_VERBOSE, "SRT output drained in %"PRId64" us "
++                       "(peer delay %d ms, RTT %.3f ms)\n",
++                       av_gettime_relative() - start, stats.msSndTsbPdDelay, stats.msRTT);
++                return 0;
++            }
++        } else {
++            quiet_until = -1;
++        }
++        av_usleep(1000);
++    }
++    av_log(h, AV_LOG_ERROR, "SRT output drain exceeded linger=%d seconds "
++           "(%zu blocks, %zu bytes pending)\n", s->linger, blocks, bytes);
++    return AVERROR(ETIMEDOUT);
++}
++
+ static int libsrt_close(URLContext *h)
+ {
+     SRTContext *s = h->priv_data;
++    int ret = libsrt_drain_live_output(h);
+
+     srt_epoll_release(s->eid);
+     srt_close(s->fd);
+
+     srt_cleanup();
+
+-    return 0;
++    return ret;
+ }
+
+ static const AVClass libsrt_class = {
+-- 
+2.55.0
+

deps/ffmpeg/9/bases.env

@@ -1,6 +1,6 @@
 upstream_ref=master-98e92563
 upstream_commit=98e92563a3b60dbf6d370fd3491d7f896398e4c1
-upstream_tree=ab815f9f1ab4c1644e14192dd713b4f632a30ad9
-patch_count=18
+upstream_tree=51bce4909d067f70630fe970fa05e920a6302222
+patch_count=19
 nv_codec_headers_tag=n13.1.15.0
 nv_codec_headers_commit=0a6fba9a2820628b8103464f4c8753ee05838baa

deps/ffmpeg/README.md

@@ -22,6 +22,13 @@ to fetch both pinned revisions; the default remains `n8.1`.
 
 ## Development-series differences
 
+Live SRT outputs with explicit positive `linger` drain the sender buffer and
+keep the connection open for the negotiated peer latency plus two measured
+RTTs, within that total linger deadline. This protects the receiver's final
+TSBPD-buffered packets; asynchronous socket linger alone does not. Timeout is
+reported as an error. Unset/zero linger, SRT inputs and file-mode sockets retain
+their previous close behavior. See the upstream [live shutdown discussion](https://github.com/Haivision/srt/issues/2760#issuecomment-1625033028).
+
 All seven custom CUDA filters remain: `convert_cuda`, `crop_cuda`,
 `overlay_many_cuda`, `pad_cuda`, `transition_cuda` (including 10-bit and dip),
 `tonemap_cuda` and `band_blur_cuda`. The RFC 4175, V4L2, optional NDI registration

doc/NODES.md

@@ -33,6 +33,15 @@ It has the same parameters as the `input` node and some additional
 -   `preseek` (float, seconds, default 0) - how many seconds to preseek back to increase chance of finding a keyframe, when seek to timestamp (`seek` command) is requested
 -   `team` (string, name of instance-shared object) - if specified, seeks on the `input` nodes may be synchronized inside specified team 
 
+Runtime control: `node.object.set <name> request-eof true` interrupts further
+input reads while keeping the producer alive to drain buffered demux/parser
+packets. It then sends one common packet EOF and finishes (respecting
+`stop_delay`), even if looping or ordinary EOF notification was disabled.
+This is irreversible and intended for a running forward input whose live
+publisher has finished but whose relay keeps the connection open. It preserves
+normal downstream backpressure; the caller must keep consumers running and
+enforce a shutdown deadline. `node.stop` remains an immediate abort.
+
 ### `realtime`
 
 Rate limit output packets/frames to wallclock. This way, DTS (in

src/avplumber_pybind.cpp

@@ -867,6 +867,8 @@ PYBIND11_MODULE(_avplumber, m) {
 
     py::class_<av::Packet, std::shared_ptr<av::Packet>>(m, "Packet")
         .def(py::init<>())
+        .def_static("eof", []() { return createEofPacket(); },
+                    "Create the native packet EOF marker (an empty Packet is not EOF).")
         .def_property_readonly("pts", [](const av::Packet &p) { return p.pts(); })
         .def_property_readonly("dts", [](const av::Packet &p) { return p.dts(); })
         .def_property_readonly("duration", [](const av::Packet &p) { return p.duration(); })

src/nodes/input_rec.cpp

@@ -36,6 +36,7 @@ class RecordingInput: public NodeSingleOutput<av::Packet>, public IStreamsInput,
 protected:
     av::FormatContext ictx_;
     std::atomic_bool should_end_ {false};
+    std::atomic_bool eof_requested_ {false};
     bool input_url_set_ = false;
     AVTS wait_start_;
     AVTS wait_max_ = AV_NOPTS_VALUE;
@@ -92,6 +93,31 @@ class RecordingInput: public NodeSingleOutput<av::Packet>, public IStreamsInput,
 
     av::Timestamp shift_ = NOTS;
     Parameters streams_object_, programs_object_;
+    av::Packet readPacket() {
+        std::error_code error;
+        av::Packet packet = ictx_.readPacket(error);
+        if (!error) {
+            return packet;
+        }
+        if (!eof_requested_ || error != std::error_code(AVERROR_EXIT, av::ffmpeg_category())) {
+            throw std::system_error(error);
+        }
+        if (packet.isNull()) {
+            // av_read_frame has exhausted its PES and codec-parser buffers.
+            return packet;
+        }
+        // avcpp reports pb->error even when av_read_frame successfully returns
+        // a buffered packet, before setting its time base/completeness. During
+        // explicit EOF drain, retain that packet and finish avcpp's bookkeeping.
+        // This context does not enable avcpp's subtract-start-time option.
+        if (ictx_.raw()->pb && ictx_.raw()->pb->error == AVERROR_EXIT &&
+            packet.streamIndex() >= 0 && size_t(packet.streamIndex()) < ictx_.streamsCount()) {
+            packet.setTimeBase(ictx_.stream(packet.streamIndex()).timeBase());
+            packet.setComplete(true);
+            return packet;
+        }
+        throw std::system_error(error);
+    }
     void closeInput(bool warn = true) {
         try {
             ictx_.close();
@@ -400,6 +426,11 @@ class RecordingInput: public NodeSingleOutput<av::Packet>, public IStreamsInput,
 public:
     RecordingInput(std::unique_ptr<Sink<av::Packet>> &&sink): NodeSingleOutput<av::Packet>(std::move(sink)) {
         ictx_.setInterruptCallback([this]() -> int {
+            if (eof_requested_) {
+                // Interrupt further reads, not the processing loop: demuxers
+                // can still return their final PES/parser packets after this.
+                return 1;
+            }
             if ( (wait_max_!=AV_NOPTS_VALUE) && ((wallclock.pts() - wait_start_) > wait_max_) ) {
                 logstream << "Timeout " << wait_max_ << " exceeded";
                 this->finished_ = true;
@@ -669,7 +700,7 @@ class RecordingInput: public NodeSingleOutput<av::Packet>, public IStreamsInput,
     virtual void process() {
         if (notify_eof_) {
             this->sink_->put(createEofPacket(video_stream_));
-            if (stop_on_eof_) {
+            if (stop_on_eof_ || eof_requested_) {
                 doStop();
                 eof_sent_ = false;
             }
@@ -768,7 +799,7 @@ class RecordingInput: public NodeSingleOutput<av::Packet>, public IStreamsInput,
             }
         }
         wait_start_ = wallclock.pts();
-        av::Packet pkt = ictx_.readPacket();
+        av::Packet pkt = readPacket();
 
         if (play_direction_ == IPlaybackControl::EPlaybackDirection::pd_Backward) {
             // read only video frames, discard all other frames
@@ -777,19 +808,19 @@ class RecordingInput: public NodeSingleOutput<av::Packet>, public IStreamsInput,
             while (!pkt.isNull() && rep--) {
                 if (pkt.streamIndex() == video_stream_)
                     break;
-                pkt = ictx_.readPacket();
+                pkt = readPacket();
             }
         }
 
-        if (pkt.isNull() && loop_ && (play_direction_ == EPlaybackDirection::pd_Forward)) {
+        if (pkt.isNull() && loop_ && !eof_requested_ && (play_direction_ == EPlaybackDirection::pd_Forward)) {
             loopForward();
             return;
         }
         if (pkt.isNull()) {
             if (!eof_sent_) {
                 // we are at the end os recording
                 logstream << "end of video reached";
-                notify_eof_ = send_eof_;
+                notify_eof_ = send_eof_ || eof_requested_;
                 eof_sent_ = true;
             }
             std::this_thread::sleep_for(5ms);
@@ -1234,7 +1265,12 @@ class RecordingInput: public NodeSingleOutput<av::Packet>, public IStreamsInput,
         }
     }
     virtual void setObject(const std::string name, const Parameters& p) {
-        if (name == "stream-limits") {
+        if (name == "request-eof") {
+            if (!p.is_boolean() || !p.get<bool>()) {
+                throw Error("request-eof expects true; EOF cannot be cancelled");
+            }
+            eof_requested_ = true;
+        } else if (name == "stream-limits") {
             if (p.count("start") > 0) {
                 if (p["start"].is_null()) {
                     start_ts_ = StreamTarget();

tests/python/smoke_input_eof_drain.py

@@ -0,0 +1,116 @@
+"""Compare MPEG-TS file EOF with explicit EOF on a still-open TCP publisher.
+
+Run remotely with the native module under an external timeout. Supply a short
+H.264/AAC TS fixture: this test does no encoding and needs no GPU. It compares
+every elementary packet, including those held in the final PES/parser buffers.
+"""
+
+import argparse
+import hashlib
+import socket
+import threading
+import time
+from collections import defaultdict
+from pathlib import Path
+
+from pyplumber import AVPlumber
+from pyplumber.node import InputRec
+
+
+def capture(url, sent=None):
+    avp = AVPlumber()
+    errors = []
+    avp.on_exception = lambda *args: errors.append(args)
+    packets = defaultdict(list)
+    avp.executeCommandsFromString("queue.plan_capacity packets 2\n")
+    avp.addNode(InputRec({"name": "input", "group": "test", "dst": "packets",
+                         "url": url, "format": "mpegts", "timeout": 10,
+                         "options": {"analyzeduration": "1000000", "probesize": "4194304"}}))
+    eof_count = 0
+    requested = False
+    before_drain = None
+    try:
+        avp.group("test").startNodes()
+        edge = avp.getEdge("packets", "Packet")
+        deadline = time.monotonic() + 10
+        while not avp.node("input").isWorking:
+            assert not errors, errors
+            assert time.monotonic() < deadline, "Input did not start"
+            time.sleep(0.01)
+        # Exercise normal bounded-queue backpressure before releasing consumers.
+        time.sleep(0.05)
+        deadline = time.monotonic() + 15
+        last_packet = time.monotonic()
+        while avp.node("input").isWorking or edge.occupied:
+            assert time.monotonic() < deadline, "Input did not finish bounded drain"
+            assert not errors, errors
+            packet = edge.tryGet(20)
+            if packet is not None:
+                last_packet = time.monotonic()
+                if packet.size == 1 and packet.data == b"\xff":
+                    eof_count += 1
+                else:
+                    assert eof_count == 0, "Media appeared after EOF"
+                    packets[packet.stream_index].append((
+                        packet.pts.timestamp, packet.dts.timestamp,
+                        packet.size, hashlib.sha256(packet.data).hexdigest()))
+            if (sent is not None and sent.is_set() and not requested
+                    and time.monotonic() - last_packet > 0.2):
+                # The publisher is still connected. The input has consumed all
+                # available bytes, but cannot flush its final PES/parser yet.
+                before_drain = sum(map(len, packets.values()))
+                avp.executeCommandsFromString("node.object.set input request-eof true\n")
+                # The command is idempotent while drain remains in progress.
+                avp.executeCommandsFromString("node.object.set input request-eof true\n")
+                requested = True
+        assert not errors, errors
+        assert eof_count == 1, f"Expected exactly one EOF, got {eof_count}"
+        if sent is not None:
+            assert requested
+            assert before_drain < sum(map(len, packets.values())), "Fixture did not exercise parser tail"
+        return dict(packets)
+    finally:
+        avp.shutdown()
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("input", type=Path)
+    args = parser.parse_args()
+    expected = capture(str(args.input.resolve()))
+    assert len(expected) >= 2, "Fixture must contain audio and video"
+    sent, release = threading.Event(), threading.Event()
+    failures = []
+    with socket.socket() as listener:
+        listener.bind(("127.0.0.1", 0))
+        listener.listen(1)
+
+        def publish():
+            try:
+                with listener.accept()[0] as client:
+                    with args.input.open("rb") as source:
+                        while block := source.read(64 * 1024):
+                            client.sendall(block)
+                    sent.set()
+                    assert release.wait(30), "Test did not release publisher"
+            except Exception as exc:
+                failures.append(exc)
+
+        worker = threading.Thread(target=publish, daemon=True)
+        worker.start()
+        try:
+            actual = capture(f"tcp://127.0.0.1:{listener.getsockname()[1]}", sent)
+            assert actual == expected, {
+                "file_packets": {key: len(value) for key, value in expected.items()},
+                "live_packets": {key: len(value) for key, value in actual.items()},
+            }
+        finally:
+            release.set()
+            worker.join(5)
+        assert not failures, failures
+    print("Live input parser drain matches file EOF:",
+          {key: len(value) for key, value in expected.items()}, flush=True)
+
+
+if __name__ == "__main__":
+    main()