From 47e11c37771e0666db40d169b1051cfeaa9cb525 Mon Sep 17 00:00:00 2001
From: Dongbin Nie <niedongbin@gmail.com>
Date: Fri, 24 Jul 2026 06:15:51 +0800
Subject: [PATCH] Fix xiaomi/cs2 stream restart on pop buffer overflow

When the media channel pop buffer fills up (e.g. the consumer pipeline
is temporarily slower than the camera: CPU throttling, slow downstream
client), dataChannel.Push returned a fatal error, which killed the
whole camera connection and forced a stream restart with re-probing.
Users saw repeated 'miss: read media: cs2: pop buffer is full' warnings
and periodic video interruptions.

Drop the oldest queued packet instead of failing the connection.
Consumers already tolerate packet loss (RTP sequence gaps) and decoders
resync on the next keyframe, which is far better than restarting the
whole stream.

Also make the pop buffer size configurable via the 'buffer' URL param
(default 256, previously hardcoded 100), so setups with slow consumers
can trade memory for fewer drops, e.g. xiaomi://...&buffer=1024.
---
 internal/xiaomi/README.md   |  7 +++++++
 pkg/xiaomi/miss/client.go   |  4 +++-
 pkg/xiaomi/miss/cs2/conn.go | 19 ++++++++++++++++---
 3 files changed, 26 insertions(+), 4 deletions(-)

diff --git a/internal/xiaomi/README.md b/internal/xiaomi/README.md
index 330183e8c..87eb1dca8 100644
--- a/internal/xiaomi/README.md
+++ b/internal/xiaomi/README.md
@@ -62,3 +62,10 @@ You can use a second channel for dual cameras: `channel=2`.
 streams:
   xiaomi1: xiaomi://***&channel=2
 ```
+
+You can change the media buffer size (in packets) if your consumers are slow: `buffer=256` (default).
+
+```yaml
+streams:
+  xiaomi1: xiaomi://***&buffer=1024
+```
diff --git a/pkg/xiaomi/miss/client.go b/pkg/xiaomi/miss/client.go
index 3113632a1..8c5519005 100644
--- a/pkg/xiaomi/miss/client.go
+++ b/pkg/xiaomi/miss/client.go
@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"net"
 	"net/url"
+	"strconv"
 	"time"
 
 	"github.com/AlexxIT/go2rtc/pkg/tutk"
@@ -55,7 +56,8 @@ func NewClient(rawURL string) (*Client, error) {
 	var conn Conn
 	switch s := query.Get("vendor"); s {
 	case "cs2":
-		conn, err = cs2.Dial(u.Host, query.Get("transport"))
+		popSize, _ := strconv.Atoi(query.Get("buffer"))
+		conn, err = cs2.Dial(u.Host, query.Get("transport"), popSize)
 	case "tutk":
 		conn, err = tutk.Dial(u.Host, query.Get("uid"), "Miss", "client")
 	default:
diff --git a/pkg/xiaomi/miss/cs2/conn.go b/pkg/xiaomi/miss/cs2/conn.go
index 2c1b395e9..b0577d1e3 100644
--- a/pkg/xiaomi/miss/cs2/conn.go
+++ b/pkg/xiaomi/miss/cs2/conn.go
@@ -12,7 +12,7 @@ import (
 	"time"
 )
 
-func Dial(host, transport string) (*Conn, error) {
+func Dial(host, transport string, popSize int) (*Conn, error) {
 	conn, err := handshake(host, transport)
 	if err != nil {
 		return nil, err
@@ -20,11 +20,15 @@ func Dial(host, transport string) (*Conn, error) {
 
 	_, isTCP := conn.(*tcpConn)
 
+	if popSize <= 0 {
+		popSize = 256
+	}
+
 	c := &Conn{
 		Conn:  conn,
 		isTCP: isTCP,
 		channels: [4]*dataChannel{
-			newDataChannel(0, 10), nil, newDataChannel(250, 100), nil,
+			newDataChannel(0, 10), nil, newDataChannel(250, popSize), nil,
 		},
 	}
 	go c.worker()
@@ -445,7 +449,16 @@ func (c *dataChannel) Push(b []byte) error {
 		select {
 		case c.popBuf <- c.waitData[:c.waitSize]:
 		default:
-			return fmt.Errorf("pop buffer is full")
+			// Slow consumer or retransmit burst. Drop the oldest packet
+			// instead of killing the whole connection (stream restart).
+			select {
+			case <-c.popBuf:
+			default:
+			}
+			select {
+			case c.popBuf <- c.waitData[:c.waitSize]:
+			default: // consumer raced us, drop the new packet
+			}
 		}
 
 		c.waitData = c.waitData[c.waitSize:]