Bump go-astits and fix async KLV processing logic

go.mod

@@ -4,7 +4,7 @@ go 1.24.0
 
 require (
 	github.com/abema/go-mp4 v1.4.1
-	github.com/asticode/go-astits v1.14.0
+	github.com/asticode/go-astits v1.15.0
 	github.com/stretchr/testify v1.11.1
 )
 

go.sum

@@ -2,8 +2,8 @@ github.com/abema/go-mp4 v1.4.1 h1:YoS4VRqd+pAmddRPLFf8vMk74kuGl6ULSjzhsIqwr6M=
 github.com/abema/go-mp4 v1.4.1/go.mod h1:vPl9t5ZK7K0x68jh12/+ECWBCXoWuIDtNgPtU2f04ws=
 github.com/asticode/go-astikit v0.30.0 h1:DkBkRQRIxYcknlaU7W7ksNfn4gMFsB0tqMJflxkRsZA=
 github.com/asticode/go-astikit v0.30.0/go.mod h1:h4ly7idim1tNhaVkdVBeXQZEE3L0xblP7fCWbgwipF0=
-github.com/asticode/go-astits v1.14.0 h1:zkgnZzipx2XX5mWycqsSBeEyDH58+i4HtyF4j2ROb00=
-github.com/asticode/go-astits v1.14.0/go.mod h1:QSHmknZ51pf6KJdHKZHJTLlMegIrhega3LPWz3ND/iI=
+github.com/asticode/go-astits v1.15.0 h1:yRyCiUc8Jj4F7clt2GDxHghMpWuFL5rkaLuGUd2/0J4=
+github.com/asticode/go-astits v1.15.0/go.mod h1:QSHmknZ51pf6KJdHKZHJTLlMegIrhega3LPWz3ND/iI=
 github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=

pkg/formats/mpegts/reader.go

@@ -185,6 +185,10 @@ type Reader struct {
 	onData          map[uint16]func(int64, int64, []byte) error
 	lastPTSReceived bool
 	lastPTS         int64
+
+	pendingAsync         map[uint16][][]byte // PID -> queued PES payloads waiting for first PTS
+	pendingAsyncBytes    int
+	pendingAsyncMaxBytes int
 }
 
 // Initialize initializes a Reader.
@@ -230,6 +234,10 @@ func (r *Reader) Initialize() error {
 	r.onDecodeError = func(_ error) {}
 	r.onData = make(map[uint16]func(int64, int64, []byte) error)
 
+	r.pendingAsync = make(map[uint16][][]byte)
+	r.pendingAsyncMaxBytes = 4 * 1024 * 1024
+	r.pendingAsyncBytes = 0
+
 	return nil
 }
 
@@ -510,6 +518,10 @@ func (r *Reader) Read() error {
 
 	if klvCodec, ok2 := track.Codec.(*codecs.KLV); ok2 && !klvCodec.Synchronous {
 		if !r.lastPTSReceived {
+			if _, hasOnData := r.onData[data.PID]; hasOnData {
+				r.pushPendingAsync(data.PID, data.PES.Data)
+			}
+
 			return nil
 		}
 
@@ -532,6 +544,21 @@ func (r *Reader) Read() error {
 
 		r.lastPTS = pts
 		r.lastPTSReceived = true
+
+		// Flush any pending async PES now that we have a timeline.
+		if len(r.pendingAsync) != 0 {
+			for pid, bufs := range r.pendingAsync {
+				if onData, hasOnData := r.onData[pid]; hasOnData {
+					for _, b := range bufs {
+						_ = onData(r.lastPTS, r.lastPTS, b)
+					}
+				}
+
+				delete(r.pendingAsync, pid)
+			}
+
+			r.pendingAsyncBytes = 0
+		}
 	}
 
 	onData, ok := r.onData[data.PID]
@@ -541,3 +568,31 @@ func (r *Reader) Read() error {
 
 	return onData(pts, dts, data.PES.Data)
 }
+
+func (r *Reader) pushPendingAsync(pid uint16, payload []byte) {
+	r.pendingAsync[pid] = append(r.pendingAsync[pid], payload)
+	r.pendingAsyncBytes += len(payload)
+
+	for r.pendingAsyncBytes > r.pendingAsyncMaxBytes {
+		for p, q := range r.pendingAsync {
+			if len(q) == 0 {
+				continue
+			}
+
+			dropped := r.pendingAsync[p][0]
+			r.pendingAsync[p] = r.pendingAsync[p][1:]
+			r.pendingAsyncBytes -= len(dropped)
+
+			if len(r.pendingAsync[p]) == 0 {
+				delete(r.pendingAsync, p)
+			}
+
+			r.onDecodeError(fmt.Errorf(
+				"pending async buffer overflow: dropped %d bytes (pid %d), limit %d bytes",
+				len(dropped), p, r.pendingAsyncMaxBytes,
+			))
+
+			break
+		}
+	}
+}