fix(vp8channel): serialize concurrent track writes (issue #95)

pion TrackLocalStaticSample.WriteSample packetizes under its lock but
emits RTP after releasing it, so concurrent callers interleave sequence
numbers on the wire. The server runs a per-peer writer pump for bulk data
alongside writerLoop for control/keepalive, both hitting the shared track;
the remote VP8 reassembler enforces strict sequence contiguity and drops
the interleaved frames, stalling server->client traffic. Funnel every
WriteSample through one mutex so each frame's packetize+emit is atomic.

internal/transport/vp8channel/concurrency_test.go

@@ -0,0 +1,65 @@
+package vp8channel
+
+import (
+	"sync"
+	"testing"
+	"time"
+)
+
+// TestWriteSampleLockedSerializesConcurrentWriters is the regression guard for
+// issue #95. The server writes bulk data through per-peer pumps while
+// writerLoop writes control/keepalive frames, both calling writeSampleLocked
+// concurrently. pion's TrackLocalStaticSample.WriteSample is not safe for
+// concurrent use: a frame's packetize step and its RTP emit are not atomic, so
+// unsynchronized callers interleave packets on the wire and the remote VP8
+// reassembler drops the corrupted frames. This test installs a sampleWriter
+// that fails if two calls are ever in flight at once, then hammers it from many
+// goroutines.
+func TestWriteSampleLockedSerializesConcurrentWriters(t *testing.T) {
+	var (
+		inFlight int32
+		mu       sync.Mutex
+		overlaps int
+	)
+
+	p := &streamTransport{frameInterval: time.Second / 30}
+	p.sampleWriter = func([]byte) bool {
+		mu.Lock()
+		inFlight++
+		if inFlight > 1 {
+			overlaps++
+		}
+		mu.Unlock()
+
+		// Hold the "wire" briefly so any missing serialization manifests as
+		// an observed overlap rather than a lucky interleave.
+		time.Sleep(50 * time.Microsecond)
+
+		mu.Lock()
+		inFlight--
+		mu.Unlock()
+		return true
+	}
+
+	const (
+		writers     = 16
+		perWriter   = 64
+	)
+	var wg sync.WaitGroup
+	wg.Add(writers)
+	for range writers {
+		go func() {
+			defer wg.Done()
+			payload := []byte{1, 2, 3, 4}
+			for range perWriter {
+				p.writeSampleLocked(payload)
+			}
+		}()
+	}
+	wg.Wait()
+
+	if overlaps != 0 {
+		t.Fatalf("writeSampleLocked allowed %d concurrent track writes; "+
+			"WriteSample must be fully serialized (issue #95 regression)", overlaps)
+	}
+}

internal/transport/vp8channel/transport.go

@@ -111,6 +111,16 @@ type videoSession interface {
 type streamTransport struct {
 	stream        videoSession
 	track         *webrtc.TrackLocalStaticSample
+	// writeMu serializes all track.WriteSample calls. pion's WriteSample is
+	// not safe for concurrent use (see writeSampleLocked); the server writes
+	// bulk data from per-peer pumps while writerLoop writes control frames
+	// and keepalives, so both paths must funnel through this lock.
+	writeMu       sync.Mutex
+	// sampleWriter, when set, replaces the real track.WriteSample call.
+	// Tests inject a writer here to observe the exact byte stream that
+	// reaches the track and to assert that writeSampleLocked serializes
+	// concurrent callers. Always invoked under writeMu.
+	sampleWriter  func([]byte) bool
 	onData        func([]byte)
 	onPeerData    func(peerID string, data []byte)
 	// onControlData is called with every reassembled message from the
@@ -647,9 +657,30 @@ type writerState struct {
 }
 
 func (w *writerState) writeSample(data []byte) bool {
-	return w.p.track.WriteSample(media.Sample{
+	return w.p.writeSampleLocked(data)
+}
+
+// writeSampleLocked serializes every WriteSample call on the shared video
+// track behind a single mutex. pion's TrackLocalStaticSample.WriteSample is
+// NOT safe for concurrent use: it packetizes under its own lock but then
+// releases that lock before pushing the resulting RTP packets onto the wire.
+// Two concurrent callers therefore each reserve a contiguous block of RTP
+// sequence numbers and then race to emit their packets, interleaving them on
+// the wire. The receiver's VP8 reassembler enforces strict sequence
+// contiguity, so any interleaved frame is discarded - which is exactly the
+// server->client bulk-data stall in issue #95 (the server runs a per-peer
+// peerWriterPump for data plus writerLoop for control/keepalive, both hitting
+// this track at once). Funneling all writes through this mutex makes each
+// sample's packetize+send atomic and keeps sequence numbers monotonic.
+func (p *streamTransport) writeSampleLocked(data []byte) bool {
+	p.writeMu.Lock()
+	defer p.writeMu.Unlock()
+	if p.sampleWriter != nil {
+		return p.sampleWriter(data)
+	}
+	return p.track.WriteSample(media.Sample{
 		Data:     data,
-		Duration: w.p.frameInterval,
+		Duration: p.frameInterval,
 	}) == nil
 }
 
@@ -1066,10 +1097,7 @@ func (p *streamTransport) peerWriterPump(_ uint32, out chan []byte) {
 			if !ok {
 				return
 			}
-			_ = p.track.WriteSample(media.Sample{
-				Data:     frame,
-				Duration: p.frameInterval,
-			})
+			_ = p.writeSampleLocked(frame)
 		}
 	}
 }