fix(packet_pool): don't drop a payload that is flushed by the same packet that completes a single-packet PES

[jhcho-hanvision]

Problem

When a TS packet both starts a new payload (PUSI) and completes a bounded PES on its own, packetAccumulator.add assigns to ps twice, so the payload accumulated so far is silently dropped — it is neither returned nor kept in the queue, and no error is reported.

if p.Header.PayloadUnitStartIndicator {
    ps = mps                    // previous payload
    mps = make([]*Packet, 0, cap(mps))
}
mps = append(mps, p)
...
} else if isPESPayload(mps[0].Payload) && isPESComplete(mps) {
    ps = mps                    // overwrites the previous payload
    mps = nil
}

The immediate flush of a complete single-packet PES was added in #72 to reduce latency for sparse streams (e.g. DVB teletext).

How we hit it

GStreamer's mpegtsmux writes PES_packet_length = 0 for video only when the PES exceeds 65535 bytes, and the real length otherwise. So a large IDR access unit (length 0, spanning many TS packets) followed by a tiny P frame that fits into a single TS packet (length set) makes the IDR disappear.

Downstream this shows up as video corruption every few seconds with no demuxer error and no transport-level loss — we chased it for a while in an SRT ingest (mediamtx, which uses go-astits). FFmpeg always writes 0 for video PES, which is why streams produced by FFmpeg never trigger it. Both writers are spec compliant: ISO/IEC 13818-1 allows 0 only for video PES, but does not require it.

Fix

If this call is already returning the previous payload, keep the newly started payload queued; it is flushed on the next payload start, as before #72 for this particular sequence. The latency improvement of #72 is therefore unaffected for sparse streams, where a payload start does not coincide with a flush.

A complete fix would let add return several payloads, but that changes the internal API and all of its call sites, so this keeps the change minimal.

Test

TestPacketPoolUnboundedPESFollowedByOnePacketPES reproduces the sequence with synthetic packets. It fails on master (only the one-packet PES is returned, len=1 instead of len=2) and passes with this change. The rest of the test suite is unaffected.

[asticode]

Thanks for the PR ❤️

A few minor changes needed 👍

[asticode]

Can you move those tests in packet_pool_test.go instead?

[asticode]

since both pesPacketPayload and tsPacket are only used in TestPacketPoolUnboundedPESFollowedByOnePacketPES, could you declare them inside TestPacketPoolUnboundedPESFollowedByOnePacketPES instead ?

func TestPacketPoolUnboundedPESFollowedByOnePacketPES(t *testing.T) {
  // pesPacketPayload builds a PES payload: start code, stream id, packet length
  // (0 = unbounded), minimal optional header, then n data bytes.
  pesPacketPayload := func(packetLength uint16, n int) []byte {
	p := []byte{0, 0, 1, 0xe0, byte(packetLength >> 8), byte(packetLength), 0x80, 0x00, 0x00}
	return append(p, make([]byte, n)...)
  }
...

...
}

[asticode]

Could you rename into packet?

[jhcho-hanvision]

Thanks for the review! Done in 3256154:

• moved the test into packet_pool_test.go
• declared pesPacketPayload and the packet helper inside the test
• renamed tsPacket to packet

The test still fails without the fix in packet_pool.go and the whole suite passes.

[asticode]

Thanks for the PR ❤️

Let me know whether you need a tag 👍

[jhcho-hanvision]

No tag needed on our side for now — we only use it as an indirect dependency. Thanks for merging! 🙏