c01d864 Merge pull request #60 from bnema/fix/udp-output-supersede-msgoutput · vev

M internal/adapters/dgram/transport.go +116 -37


@@ -172,18 +172,20 @@ type Transport struct {
172	172
173	173		writeMu sync.Mutex
174	174
175		-	deliverMu sync.Mutex
176		-	deliverCond *sync.Cond
177		-	nextRecvSeq uint64
178		-	recvBuf map[uint64]ports.Frame
	175	+	deliverMu sync.Mutex
	176	+	deliverCond *sync.Cond
	177	+	nextRecvSeq uint64
	178	+	recvBuf map[uint64]ports.Frame
	179	+	recvSkipThrough uint64
179	180		}
180	181
181	182		type pending struct {
182		-	frame ports.Frame
183		-	first time.Time
184		-	last time.Time
185		-	attempts int
186		-	retransmitted bool
	183	+	frame ports.Frame
	184	+	first time.Time
	185	+	last time.Time
	186	+	attempts int
	187	+	retransmitted bool
	188	+	outputSkipThrough uint64
187	189		}
188	190
189	191		func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32) (*Transport, error) {

@@ -245,13 +247,12 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
245	247		func (t *Transport) DatagramTransport() {}
246	248
247	249		func (t *Transport) Send(f ports.Frame) error {
248		-	// MsgOutput supersession is intentionally deferred: this reliable stream uses
249		-	// strict sequence delivery, so deleting an older pending output without a
250		-	// receive-side skip/ack signal would stall later frames behind the missing
251		-	// sequence. Add an explicit droppable/skip protocol before replacing pending
252		-	// outputs.
253	250		reliable := true
254	251		t.mu.Lock()
	252	+	outputSkipThrough := uint64(0)
	253	+	if f.Type == ports.MsgOutput {
	254	+	outputSkipThrough = t.supersedePendingOutputLocked()
	255	+	}
255	256		var pendingWaitStarted time.Time
256	257		for reliable && len(t.pending) >= t.maxPending && !t.closed {
257	258		if t.linkState != ports.LinkStateConnected {

@@ -289,10 +290,10 @@ func (t *Transport) Send(f ports.Frame) error {
289	290		seq := t.seq
290	291		if reliable {
291	292		now := t.clock.Now()
292		-	t.pending[seq] = &pending{frame: f, first: now, last: now}
	293	+	t.pending[seq] = &pending{frame: f, first: now, last: now, outputSkipThrough: outputSkipThrough}
293	294		}
294	295		t.mu.Unlock()
295		-	if err := t.sendData(seq, reliable, f); err != nil {
	296	+	if err := t.sendData(seq, reliable, f, outputSkipThrough); err != nil {
296	297		if reliable {
297	298		t.mu.Lock()
298	299		if _, ok := t.pending[seq]; ok {

@@ -306,6 +307,34 @@ func (t *Transport) Send(f ports.Frame) error {
306	307		return nil
307	308		}
308	309
	310	+	func (t *Transport) supersedePendingOutputLocked() uint64 {
	311	+	var barrier uint64
	312	+	for seq, p := range t.pending {
	313	+	if p != nil && p.frame.Type != ports.MsgOutput && seq > barrier {
	314	+	barrier = seq
	315	+	}
	316	+	}
	317	+	var newest uint64
	318	+	for seq, p := range t.pending {
	319	+	if p == nil \|\| p.frame.Type != ports.MsgOutput \|\| seq <= barrier {
	320	+	continue
	321	+	}
	322	+	if seq > newest {
	323	+	newest = seq
	324	+	}
	325	+	}
	326	+	if newest == 0 {
	327	+	return 0
	328	+	}
	329	+	for seq, p := range t.pending {
	330	+	if p != nil && p.frame.Type == ports.MsgOutput && seq > barrier && seq <= newest {
	331	+	delete(t.pending, seq)
	332	+	}
	333	+	}
	334	+	t.notifySendWaitersLocked()
	335	+	return newest
	336	+	}
	337	+
309	338		func (t *Transport) Recv() (ports.Frame, error) {
310	339		select {
311	340		case f := <-t.in:

@@ -387,8 +416,8 @@ func (t *Transport) unregisterProbe(id uint64) {
387	416		t.mu.Unlock()
388	417		}
389	418
390		-	func (t *Transport) sendData(seq uint64, reliable bool, f ports.Frame) error {
391		-	p := encodeData(seq, reliable, f)
	419	+	func (t *Transport) sendData(seq uint64, reliable bool, f ports.Frame, outputSkipThrough uint64) error {
	420	+	p := encodeData(seq, reliable, f, outputSkipThrough)
392	421		return t.sendPayload(p)
393	422		}
394	423		func (t *Transport) sendAck(seq uint64) {

@@ -547,12 +576,12 @@ func (t *Transport) handleRecord(p []byte) {
547	576		}
548	577		t.mu.Unlock()
549	578		case recData:
550		-	seq, reliable, f, ok := decodeData(p)
	579	+	seq, reliable, f, outputSkipThrough, ok := decodeData(p)
551	580		if !ok {
552	581		return
553	582		}
554	583		if reliable {
555		-	ackSeq, ack, queued := t.enqueueReliable(seq, f)
	584	+	ackSeq, ack, queued := t.enqueueReliable(seq, f, outputSkipThrough)
556	585		if ack {
557	586		t.sendAck(ackSeq)
558	587		}

@@ -565,9 +594,18 @@ func (t *Transport) handleRecord(p []byte) {
565	594		}
566	595		}
567	596
568		-	func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) (ackSeq uint64, ack bool, queued bool) {
	597	+	func (t *Transport) enqueueReliable(seq uint64, f ports.Frame, outputSkipThrough uint64) (ackSeq uint64, ack bool, queued bool) {
569	598		t.deliverMu.Lock()
570	599		defer t.deliverMu.Unlock()
	600	+	if f.Type == ports.MsgOutput && outputSkipThrough > t.recvSkipThrough {
	601	+	t.recvSkipThrough = outputSkipThrough
	602	+	for bufferedSeq, buffered := range t.recvBuf {
	603	+	if bufferedSeq <= outputSkipThrough && buffered.Type == ports.MsgOutput {
	604	+	delete(t.recvBuf, bufferedSeq)
	605	+	}
	606	+	}
	607	+	t.deliverCond.Signal()
	608	+	}
571	609		if seq < t.nextRecvSeq {
572	610		return t.highestContiguousRecvLocked(), true, false
573	611		}

@@ -584,10 +622,15 @@ func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) (ackSeq uint64, a
584	622		func (t *Transport) highestContiguousRecvLocked() uint64 {
585	623		seq := t.nextRecvSeq
586	624		for {
587		-	if _, ok := t.recvBuf[seq]; !ok {
588		-	break
	625	+	if _, ok := t.recvBuf[seq]; ok {
	626	+	seq++
	627	+	continue
	628	+	}
	629	+	if seq <= t.recvSkipThrough {
	630	+	seq++
	631	+	continue
589	632		}
590		-	seq++
	633	+	break
591	634		}
592	635		return seq - 1
593	636		}

@@ -602,6 +645,10 @@ func (t *Transport) deliveryLoop() {
602	645		for {
603	646		t.deliverMu.Lock()
604	647		f, ok := t.recvBuf[t.nextRecvSeq]
	648	+	for !ok && t.nextRecvSeq <= t.recvSkipThrough {
	649	+	t.nextRecvSeq++
	650	+	f, ok = t.recvBuf[t.nextRecvSeq]
	651	+	}
605	652		for !ok {
606	653		if t.isClosed() {
607	654		t.deliverMu.Unlock()

@@ -609,6 +656,16 @@ func (t *Transport) deliveryLoop() {
609	656		}
610	657		t.deliverCond.Wait()
611	658		f, ok = t.recvBuf[t.nextRecvSeq]
	659	+	for !ok && t.nextRecvSeq <= t.recvSkipThrough {
	660	+	t.nextRecvSeq++
	661	+	f, ok = t.recvBuf[t.nextRecvSeq]
	662	+	}
	663	+	}
	664	+	if t.nextRecvSeq <= t.recvSkipThrough && f.Type == ports.MsgOutput {
	665	+	delete(t.recvBuf, t.nextRecvSeq)
	666	+	t.nextRecvSeq++
	667	+	t.deliverMu.Unlock()
	668	+	continue
612	669		}
613	670		delete(t.recvBuf, t.nextRecvSeq)
614	671		t.nextRecvSeq++

@@ -675,8 +732,9 @@ func (t *Transport) resendLoop() {
675	732		func (t *Transport) resendPending() {
676	733		now := t.clock.Now()
677	734		var resend []struct {
678		-	seq uint64
679		-	f ports.Frame
	735	+	seq uint64
	736	+	f ports.Frame
	737	+	outputSkipThrough uint64
680	738		}
681	739		t.mu.Lock()
682	740		seqs := make([]uint64, 0, len(t.pending))

@@ -697,14 +755,15 @@ func (t *Transport) resendPending() {
697	755		p.attempts++
698	756		p.retransmitted = true
699	757		resend = append(resend, struct {
700		-	seq uint64
701		-	f ports.Frame
702		-	}{seq, p.frame})
	758	+	seq uint64
	759	+	f ports.Frame
	760	+	outputSkipThrough uint64
	761	+	}{seq, p.frame, p.outputSkipThrough})
703	762		limit--
704	763		}
705	764		t.mu.Unlock()
706	765		for _, r := range resend {
707		-	_ = t.sendData(r.seq, true, r.f)
	766	+	_ = t.sendData(r.seq, true, r.f, r.outputSkipThrough)
708	767		}
709	768		}
710	769

@@ -887,21 +946,41 @@ func (t *Transport) closeWithError(err error) {
887	946		t.deliverMu.Unlock()
888	947		}
889	948
890		-	func encodeData(seq uint64, reliable bool, f ports.Frame) []byte {
891		-	b := make([]byte, 12+len(f.Payload))
	949	+	func encodeData(seq uint64, reliable bool, f ports.Frame, outputSkipThrough uint64) []byte {
	950	+	headerLen := 12
	951	+	if outputSkipThrough > 0 {
	952	+	headerLen = 20
	953	+	}
	954	+	b := make([]byte, headerLen+len(f.Payload))
892	955		b[0] = recData
893	956		binary.BigEndian.PutUint64(b[1:9], seq)
894	957		if reliable {
895	958		b[9] = 1
896	959		}
897	960		b[10] = byte(f.Type)
898		-	b[11] = 0
899		-	copy(b[12:], f.Payload)
	961	+	if outputSkipThrough > 0 {
	962	+	b[11] = 1
	963	+	binary.BigEndian.PutUint64(b[12:20], outputSkipThrough)
	964	+	}
	965	+	copy(b[headerLen:], f.Payload)
900	966		return b
901	967		}
902		-	func decodeData(b []byte) (uint64, bool, ports.Frame, bool) {
	968	+	func decodeData(b []byte) (uint64, bool, ports.Frame, uint64, bool) {
903	969		if len(b) < 12 \|\| b[0] != recData {
904		-	return 0, false, ports.Frame{}, false
	970	+	return 0, false, ports.Frame{}, 0, false
	971	+	}
	972	+	flags := b[11]
	973	+	if flags&^byte(1) != 0 {
	974	+	return 0, false, ports.Frame{}, 0, false
	975	+	}
	976	+	headerLen := 12
	977	+	outputSkipThrough := uint64(0)
	978	+	if flags&1 != 0 {
	979	+	if len(b) < 20 {
	980	+	return 0, false, ports.Frame{}, 0, false
	981	+	}
	982	+	headerLen = 20
	983	+	outputSkipThrough = binary.BigEndian.Uint64(b[12:20])
905	984		}
906		-	return binary.BigEndian.Uint64(b[1:9]), b[9] == 1, ports.Frame{Type: ports.MsgType(b[10]), Payload: append([]byte(nil), b[12:]...)}, true
	985	+	return binary.BigEndian.Uint64(b[1:9]), b[9] == 1, ports.Frame{Type: ports.MsgType(b[10]), Payload: append([]byte(nil), b[headerLen:]...)}, outputSkipThrough, true
907	986		}

 	writeMu sync.Mutex
-	deliverMu   sync.Mutex
+	deliverMu       sync.Mutex
-	deliverCond *sync.Cond
+	deliverCond     *sync.Cond
-	nextRecvSeq uint64
+	nextRecvSeq     uint64
-	recvBuf     map[uint64]ports.Frame
+	recvBuf         map[uint64]ports.Frame
+	recvSkipThrough uint64
 }
 type pending struct {
-	frame         ports.Frame
+	frame             ports.Frame
-	first         time.Time
+	first             time.Time
-	last          time.Time
+	last              time.Time
-	attempts      int
+	attempts          int
-	retransmitted bool
+	retransmitted     bool
+	outputSkipThrough uint64
 }
 func NewTransport(pc net.PacketConn, peer net.Addr, key []byte, sendDir, recvDir uint32) (*Transport, error) {
 func (t *Transport) DatagramTransport() {}
 func (t *Transport) Send(f ports.Frame) error {
-	// MsgOutput supersession is intentionally deferred: this reliable stream uses
-	// strict sequence delivery, so deleting an older pending output without a
-	// receive-side skip/ack signal would stall later frames behind the missing
-	// sequence. Add an explicit droppable/skip protocol before replacing pending
-	// outputs.
 	reliable := true
 	t.mu.Lock()
+	outputSkipThrough := uint64(0)
+	if f.Type == ports.MsgOutput {
+		outputSkipThrough = t.supersedePendingOutputLocked()
+	}
 	var pendingWaitStarted time.Time
 	for reliable && len(t.pending) >= t.maxPending && !t.closed {
 		if t.linkState != ports.LinkStateConnected {
 	seq := t.seq
 	if reliable {
 		now := t.clock.Now()
-		t.pending[seq] = &pending{frame: f, first: now, last: now}
+		t.pending[seq] = &pending{frame: f, first: now, last: now, outputSkipThrough: outputSkipThrough}
 	}
 	t.mu.Unlock()
-	if err := t.sendData(seq, reliable, f); err != nil {
+	if err := t.sendData(seq, reliable, f, outputSkipThrough); err != nil {
 		if reliable {
 			t.mu.Lock()
 			if _, ok := t.pending[seq]; ok {
 	return nil
 }
+func (t *Transport) supersedePendingOutputLocked() uint64 {
+	var barrier uint64
+	for seq, p := range t.pending {
+		if p != nil && p.frame.Type != ports.MsgOutput && seq > barrier {
+			barrier = seq
+		}
+	}
+	var newest uint64
+	for seq, p := range t.pending {
+		if p == nil || p.frame.Type != ports.MsgOutput || seq <= barrier {
+			continue
+		}
+		if seq > newest {
+			newest = seq
+		}
+	}
+	if newest == 0 {
+		return 0
+	}
+	for seq, p := range t.pending {
+		if p != nil && p.frame.Type == ports.MsgOutput && seq > barrier && seq <= newest {
+			delete(t.pending, seq)
+		}
+	}
+	t.notifySendWaitersLocked()
+	return newest
+}
 func (t *Transport) Recv() (ports.Frame, error) {
 	select {
 	case f := <-t.in:
 	t.mu.Unlock()
 }
-func (t *Transport) sendData(seq uint64, reliable bool, f ports.Frame) error {
+func (t *Transport) sendData(seq uint64, reliable bool, f ports.Frame, outputSkipThrough uint64) error {
-	p := encodeData(seq, reliable, f)
+	p := encodeData(seq, reliable, f, outputSkipThrough)
 	return t.sendPayload(p)
 }
 func (t *Transport) sendAck(seq uint64) {
 		}
 		t.mu.Unlock()
 	case recData:
-		seq, reliable, f, ok := decodeData(p)
+		seq, reliable, f, outputSkipThrough, ok := decodeData(p)
 		if !ok {
 			return
 		}
 		if reliable {
-			ackSeq, ack, queued := t.enqueueReliable(seq, f)
+			ackSeq, ack, queued := t.enqueueReliable(seq, f, outputSkipThrough)
 			if ack {
 				t.sendAck(ackSeq)
 			}
 	}
 }
-func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) (ackSeq uint64, ack bool, queued bool) {
+func (t *Transport) enqueueReliable(seq uint64, f ports.Frame, outputSkipThrough uint64) (ackSeq uint64, ack bool, queued bool) {
 	t.deliverMu.Lock()
 	defer t.deliverMu.Unlock()
+	if f.Type == ports.MsgOutput && outputSkipThrough > t.recvSkipThrough {
+		t.recvSkipThrough = outputSkipThrough
+		for bufferedSeq, buffered := range t.recvBuf {
+			if bufferedSeq <= outputSkipThrough && buffered.Type == ports.MsgOutput {
+				delete(t.recvBuf, bufferedSeq)
+			}
+		}
+		t.deliverCond.Signal()
+	}
 	if seq < t.nextRecvSeq {
 		return t.highestContiguousRecvLocked(), true, false
 	}
 func (t *Transport) highestContiguousRecvLocked() uint64 {
 	seq := t.nextRecvSeq
 	for {
-		if _, ok := t.recvBuf[seq]; !ok {
+		if _, ok := t.recvBuf[seq]; ok {
-			break
+			seq++
+			continue
+		}
+		if seq <= t.recvSkipThrough {
+			seq++
+			continue
 		}
-		seq++
+		break
 	}
 	return seq - 1
 }
 	for {
 		t.deliverMu.Lock()
 		f, ok := t.recvBuf[t.nextRecvSeq]
+		for !ok && t.nextRecvSeq <= t.recvSkipThrough {
+			t.nextRecvSeq++
+			f, ok = t.recvBuf[t.nextRecvSeq]
+		}
 		for !ok {
 			if t.isClosed() {
 				t.deliverMu.Unlock()
 			}
 			t.deliverCond.Wait()
 			f, ok = t.recvBuf[t.nextRecvSeq]
+			for !ok && t.nextRecvSeq <= t.recvSkipThrough {
+				t.nextRecvSeq++
+				f, ok = t.recvBuf[t.nextRecvSeq]
+			}
+		}
+		if t.nextRecvSeq <= t.recvSkipThrough && f.Type == ports.MsgOutput {
+			delete(t.recvBuf, t.nextRecvSeq)
+			t.nextRecvSeq++
+			t.deliverMu.Unlock()
+			continue
 		}
 		delete(t.recvBuf, t.nextRecvSeq)
 		t.nextRecvSeq++
 func (t *Transport) resendPending() {
 	now := t.clock.Now()
 	var resend []struct {
-		seq uint64
+		seq               uint64
-		f   ports.Frame
+		f                 ports.Frame
+		outputSkipThrough uint64
 	}
 	t.mu.Lock()
 	seqs := make([]uint64, 0, len(t.pending))
 		p.attempts++
 		p.retransmitted = true
 		resend = append(resend, struct {
-			seq uint64
+			seq               uint64
-			f   ports.Frame
+			f                 ports.Frame
-		}{seq, p.frame})
+			outputSkipThrough uint64
+		}{seq, p.frame, p.outputSkipThrough})
 		limit--
 	}
 	t.mu.Unlock()
 	for _, r := range resend {
-		_ = t.sendData(r.seq, true, r.f)
+		_ = t.sendData(r.seq, true, r.f, r.outputSkipThrough)
 	}
 }
 	t.deliverMu.Unlock()
 }
-func encodeData(seq uint64, reliable bool, f ports.Frame) []byte {
+func encodeData(seq uint64, reliable bool, f ports.Frame, outputSkipThrough uint64) []byte {
-	b := make([]byte, 12+len(f.Payload))
+	headerLen := 12
+	if outputSkipThrough > 0 {
+		headerLen = 20
+	}
+	b := make([]byte, headerLen+len(f.Payload))
 	b[0] = recData
 	binary.BigEndian.PutUint64(b[1:9], seq)
 	if reliable {
 		b[9] = 1
 	}
 	b[10] = byte(f.Type)
-	b[11] = 0
+	if outputSkipThrough > 0 {
-	copy(b[12:], f.Payload)
+		b[11] = 1
+		binary.BigEndian.PutUint64(b[12:20], outputSkipThrough)
+	}
+	copy(b[headerLen:], f.Payload)
 	return b
 }
-func decodeData(b []byte) (uint64, bool, ports.Frame, bool) {
+func decodeData(b []byte) (uint64, bool, ports.Frame, uint64, bool) {
 	if len(b) < 12 || b[0] != recData {
-		return 0, false, ports.Frame{}, false
+		return 0, false, ports.Frame{}, 0, false
+	}
+	flags := b[11]
+	if flags&^byte(1) != 0 {
+		return 0, false, ports.Frame{}, 0, false
+	}
+	headerLen := 12
+	outputSkipThrough := uint64(0)
+	if flags&1 != 0 {
+		if len(b) < 20 {
+			return 0, false, ports.Frame{}, 0, false
+		}
+		headerLen = 20
+		outputSkipThrough = binary.BigEndian.Uint64(b[12:20])
 	}
-	return binary.BigEndian.Uint64(b[1:9]), b[9] == 1, ports.Frame{Type: ports.MsgType(b[10]), Payload: append([]byte(nil), b[12:]...)}, true
+	return binary.BigEndian.Uint64(b[1:9]), b[9] == 1, ports.Frame{Type: ports.MsgType(b[10]), Payload: append([]byte(nil), b[headerLen:]...)}, outputSkipThrough, true
 }

M internal/adapters/dgram/transport_test.go +96 -13


@@ -389,7 +389,7 @@ func TestAuthenticatedPeerChangeUpdatesPeerAndEmitsConnected(t *testing.T) {
389	389		t.Fatalf("unauthenticated packet changed peer to %v", a.Peer())
390	390		}
391	391		c, _ := pdgram.NewCodec(key())
392		-	rec := encodeData(7, true, ports.Frame{Type: ports.MsgPing})
	392	+	rec := encodeData(7, true, ports.Frame{Type: ports.MsgPing}, 0)
393	393		frags, _ := pdgram.FragmentPayload(9, rec, pdgram.DefaultMTU)
394	394		raw, _ := pdgram.MarshalFragment(frags[0])
395	395		aPC.in <- packet{c.Seal(2, 99, raw, nil), testAddr("evil")}

@@ -511,11 +511,11 @@ func TestServerWithoutRebindDoesNotHopPorts(t *testing.T) {
511	511
512	512		func TestReliableReceiveBufferDoesNotDropAckedFutureFrame(t *testing.T) {
513	513		tr := &Transport{maxRecvBuffer: 1, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}
514		-	_, ack, queued := tr.enqueueReliable(2, ports.Frame{Type: ports.MsgOutput, Payload: []byte("future")})
	514	+	_, ack, queued := tr.enqueueReliable(2, ports.Frame{Type: ports.MsgOutput, Payload: []byte("future")}, 0)
515	515		if !ack \|\| !queued {
516	516		t.Fatalf("future enqueue ack=%v queued=%v, want true/true", ack, queued)
517	517		}
518		-	_, ack, queued = tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")})
	518	+	_, ack, queued = tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}, 0)
519	519		if !ack \|\| !queued {
520	520		t.Fatalf("next enqueue ack=%v queued=%v, want true/true", ack, queued)
521	521		}

@@ -524,6 +524,32 @@ func TestReliableReceiveBufferDoesNotDropAckedFutureFrame(t *testing.T) {
524	524		}
525	525		}
526	526
	527	+	func TestDataRecordWireEncodingOutputSkip(t *testing.T) {
	528	+	got := encodeData(5, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("x")}, 3)
	529	+	want := []byte{recData, 0, 0, 0, 0, 0, 0, 0, 5, 1, byte(ports.MsgOutput), 1, 0, 0, 0, 0, 0, 0, 0, 3, 'x'}
	530	+	if !bytes.Equal(got, want) {
	531	+	t.Fatalf("encoded data=%v, want %v", got, want)
	532	+	}
	533	+	seq, reliable, frame, skip, ok := decodeData(got)
	534	+	if !ok \|\| seq != 5 \|\| !reliable \|\| frame.Type != ports.MsgOutput \|\| string(frame.Payload) != "x" \|\| skip != 3 {
	535	+	t.Fatalf("decoded seq=%d reliable=%v frame=%+v skip=%d ok=%v", seq, reliable, frame, skip, ok)
	536	+	}
	537	+	}
	538	+
	539	+	func TestDataRecordWireDecodeRejectsTruncatedOutputSkip(t *testing.T) {
	540	+	truncated := []byte{recData, 0, 0, 0, 0, 0, 0, 0, 5, 1, byte(ports.MsgOutput), 1, 0}
	541	+	if _, _, _, _, ok := decodeData(truncated); ok {
	542	+	t.Fatal("decodeData accepted truncated output skip header")
	543	+	}
	544	+	}
	545	+
	546	+	func TestDataRecordWireDecodeRejectsUnknownFlags(t *testing.T) {
	547	+	withTrailingGarbage := []byte{recData, 0, 0, 0, 0, 0, 0, 0, 5, 1, byte(ports.MsgOutput), 0x80, 'x'}
	548	+	if _, _, _, _, ok := decodeData(withTrailingGarbage); ok {
	549	+	t.Fatal("decodeData accepted unknown data-record flags with trailing garbage")
	550	+	}
	551	+	}
	552	+
527	553		func TestOutputRetransmitsUntilAck(t *testing.T) {
528	554		aPC, bPC := newPair()
529	555		var dropped atomic.Bool

@@ -551,7 +577,7 @@ func TestOutputRetransmitsUntilAck(t *testing.T) {
551	577		}
552	578		}
553	579
554		-	func TestReliableDeliveryPreservesSequenceOrder(t *testing.T) {
	580	+	func TestReliableDeliveryPreservesInputSequenceOrder(t *testing.T) {
555	581		aPC, bPC := newPair()
556	582		a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)
557	583		b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)

@@ -559,18 +585,51 @@ func TestReliableDeliveryPreservesSequenceOrder(t *testing.T) {
559	585		defer func() { _ = b.Close() }()
560	586
561	587		for i := range 100 {
562		-	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
	588	+	if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte{byte(i)}}); err != nil {
563	589		t.Fatal(err)
564	590		}
565	591		}
566	592		for i := range 100 {
567	593		got := recvWithin(t, b, time.Second)
568		-	if got.Type != ports.MsgOutput \|\| len(got.Payload) != 1 \|\| got.Payload[0] != byte(i) {
	594	+	if got.Type != ports.MsgInput \|\| len(got.Payload) != 1 \|\| got.Payload[0] != byte(i) {
569	595		t.Fatalf("frame %d delivered out of order: %+v", i, got)
570	596		}
571	597		}
572	598		}
573	599
	600	+	func TestOutputSupersessionSkipsDroppedOlderOutput(t *testing.T) {
	601	+	aPC, bPC := newPair()
	602	+	var sendAttempts atomic.Int32
	603	+	aPC.drop = func(_ []byte, addr net.Addr) bool {
	604	+	return addr.String() == "b" && sendAttempts.Add(1) <= 2
	605	+	}
	606	+	a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)
	607	+	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
	608	+	defer func() { _ = a.Close() }()
	609	+	defer func() { _ = b.Close() }()
	610	+
	611	+	for _, payload := range []string{"state1", "state2", "state3"} {
	612	+	if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte(payload)}); err != nil {
	613	+	t.Fatal(err)
	614	+	}
	615	+	}
	616	+
	617	+	a.mu.Lock()
	618	+	pending := len(a.pending)
	619	+	a.mu.Unlock()
	620	+	if pending > 1 {
	621	+	t.Fatalf("pending outputs=%d, want collapsed to newest", pending)
	622	+	}
	623	+
	624	+	got := recvWithin(t, b, time.Second)
	625	+	if got.Type != ports.MsgOutput \|\| string(got.Payload) != "state3" {
	626	+	t.Fatalf("got=%+v, want newest output after skipped predecessors", got)
	627	+	}
	628	+	if sendAttempts.Load() < 3 {
	629	+	t.Fatalf("send attempts=%d, want first two output sends dropped before newest arrived", sendAttempts.Load())
	630	+	}
	631	+	}
	632	+
574	633		func TestAckProcessingContinuesWhenConsumerBackpressured(t *testing.T) {
575	634		aPC, bPC := newPair()
576	635		a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)

@@ -719,7 +778,7 @@ func TestCumulativeAckReleasesAllEarlierPendingFrames(t *testing.T) {
719	778		func TestReliableReceiverAcksOnlyHighestContiguousSequence(t *testing.T) {
720	779		tr := &Transport{maxRecvBuffer: maxRecvBuffer, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}
721	780
722		-	ackSeq, ack, queued := tr.enqueueReliable(3, ports.Frame{Type: ports.MsgOutput, Payload: []byte("third")})
	781	+	ackSeq, ack, queued := tr.enqueueReliable(3, ports.Frame{Type: ports.MsgOutput, Payload: []byte("third")}, 0)
723	782		if !ack \|\| !queued {
724	783		t.Fatalf("out-of-order enqueue ack=%v queued=%v, want true/true", ack, queued)
725	784		}

@@ -727,7 +786,7 @@ func TestReliableReceiverAcksOnlyHighestContiguousSequence(t *testing.T) {
727	786		t.Fatalf("out-of-order ack=%d, want highest contiguous %d", got, want)
728	787		}
729	788
730		-	ackSeq, ack, queued = tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("first")})
	789	+	ackSeq, ack, queued = tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("first")}, 0)
731	790		if !ack \|\| !queued {
732	791		t.Fatalf("first enqueue ack=%v queued=%v, want true/true", ack, queued)
733	792		}

@@ -735,7 +794,7 @@ func TestReliableReceiverAcksOnlyHighestContiguousSequence(t *testing.T) {
735	794		t.Fatalf("first contiguous ack=%d, want %d", got, want)
736	795		}
737	796
738		-	ackSeq, ack, queued = tr.enqueueReliable(2, ports.Frame{Type: ports.MsgOutput, Payload: []byte("second")})
	797	+	ackSeq, ack, queued = tr.enqueueReliable(2, ports.Frame{Type: ports.MsgOutput, Payload: []byte("second")}, 0)
739	798		if !ack \|\| !queued {
740	799		t.Fatalf("gap fill enqueue ack=%v queued=%v, want true/true", ack, queued)
741	800		}

@@ -966,12 +1025,12 @@ func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) {
966	1025		tr := &Transport{nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), maxRecvBuffer: maxRecvBuffer, done: make(chan struct{})}
967	1026		tr.deliverCond = sync.NewCond(&tr.deliverMu)
968	1027		for i := range maxRecvBuffer + 100 {
969		-	tr.enqueueReliable(uint64(1000+i), ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}})
	1028	+	tr.enqueueReliable(uint64(1000+i), ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}, 0)
970	1029		}
971	1030		if got := len(tr.recvBuf); got != maxRecvBuffer {
972	1031		t.Fatalf("recvBuf len=%d, want %d", got, maxRecvBuffer)
973	1032		}
974		-	tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")})
	1033	+	tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}, 0)
975	1034		if _, ok := tr.recvBuf[1]; !ok {
976	1035		t.Fatalf("next expected sequence was dropped when far-future buffer was full")
977	1036		}

@@ -980,6 +1039,30 @@ func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) {
980	1039		}
981	1040		}
982	1041
	1042	+	func TestOutputSupersessionPrunesFullReceiveBuffer(t *testing.T) {
	1043	+	tr := &Transport{
	1044	+	nextRecvSeq: 1,
	1045	+	recvBuf: map[uint64]ports.Frame{2: {Type: ports.MsgOutput, Payload: []byte("stale")}},
	1046	+	maxRecvBuffer: 1,
	1047	+	done: make(chan struct{}),
	1048	+	}
	1049	+	tr.deliverCond = sync.NewCond(&tr.deliverMu)
	1050	+
	1051	+	ackSeq, ack, queued := tr.enqueueReliable(3, ports.Frame{Type: ports.MsgOutput, Payload: []byte("replacement")}, 2)
	1052	+	if !ack \|\| !queued {
	1053	+	t.Fatalf("replacement ack=%v queued=%v, want true/true", ack, queued)
	1054	+	}
	1055	+	if ackSeq != 3 {
	1056	+	t.Fatalf("ack seq=%d, want 3 after skipped output", ackSeq)
	1057	+	}
	1058	+	if _, ok := tr.recvBuf[2]; ok {
	1059	+	t.Fatal("superseded output remained in full receive buffer")
	1060	+	}
	1061	+	if got := string(tr.recvBuf[3].Payload); got != "replacement" {
	1062	+	t.Fatalf("queued payload=%q, want replacement", got)
	1063	+	}
	1064	+	}
	1065	+
983	1066		func TestReliableFullRecvBufferDoesNotAckDroppedFarFutureFrame(t *testing.T) {
984	1067		aPC, bPC := newPair()
985	1068		codec, err := pdgram.NewCodec(key())

@@ -994,14 +1077,14 @@ func TestReliableFullRecvBufferDoesNotAckDroppedFarFutureFrame(t *testing.T) {
994	1077		for i := range maxRecvBuffer {
995	1078		tr.recvBuf[uint64(1000+i)] = ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}
996	1079		}
997		-	tr.handleRecord(encodeData(9999, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("dropped")}))
	1080	+	tr.handleRecord(encodeData(9999, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("dropped")}, 0))
998	1081		select {
999	1082		case pkt := <-bPC.in:
1000	1083		t.Fatalf("unexpected ACK packet for dropped frame: %x", pkt.b)
1001	1084		case <-time.After(25 * time.Millisecond):
1002	1085		}
1003	1086
1004		-	tr.handleRecord(encodeData(1, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}))
	1087	+	tr.handleRecord(encodeData(1, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}, 0))
1005	1088		select {
1006	1089		case <-bPC.in:
1007	1090		case <-time.After(time.Second):

 		t.Fatalf("unauthenticated packet changed peer to %v", a.Peer())
 	}
 	c, _ := pdgram.NewCodec(key())
-	rec := encodeData(7, true, ports.Frame{Type: ports.MsgPing})
+	rec := encodeData(7, true, ports.Frame{Type: ports.MsgPing}, 0)
 	frags, _ := pdgram.FragmentPayload(9, rec, pdgram.DefaultMTU)
 	raw, _ := pdgram.MarshalFragment(frags[0])
 	aPC.in <- packet{c.Seal(2, 99, raw, nil), testAddr("evil")}
 func TestReliableReceiveBufferDoesNotDropAckedFutureFrame(t *testing.T) {
 	tr := &Transport{maxRecvBuffer: 1, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}
-	_, ack, queued := tr.enqueueReliable(2, ports.Frame{Type: ports.MsgOutput, Payload: []byte("future")})
+	_, ack, queued := tr.enqueueReliable(2, ports.Frame{Type: ports.MsgOutput, Payload: []byte("future")}, 0)
 	if !ack || !queued {
 		t.Fatalf("future enqueue ack=%v queued=%v, want true/true", ack, queued)
 	}
-	_, ack, queued = tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")})
+	_, ack, queued = tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}, 0)
 	if !ack || !queued {
 		t.Fatalf("next enqueue ack=%v queued=%v, want true/true", ack, queued)
 	}
 	}
 }
+func TestDataRecordWireEncodingOutputSkip(t *testing.T) {
+	got := encodeData(5, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("x")}, 3)
+	want := []byte{recData, 0, 0, 0, 0, 0, 0, 0, 5, 1, byte(ports.MsgOutput), 1, 0, 0, 0, 0, 0, 0, 0, 3, 'x'}
+	if !bytes.Equal(got, want) {
+		t.Fatalf("encoded data=%v, want %v", got, want)
+	}
+	seq, reliable, frame, skip, ok := decodeData(got)
+	if !ok || seq != 5 || !reliable || frame.Type != ports.MsgOutput || string(frame.Payload) != "x" || skip != 3 {
+		t.Fatalf("decoded seq=%d reliable=%v frame=%+v skip=%d ok=%v", seq, reliable, frame, skip, ok)
+	}
+}
+func TestDataRecordWireDecodeRejectsTruncatedOutputSkip(t *testing.T) {
+	truncated := []byte{recData, 0, 0, 0, 0, 0, 0, 0, 5, 1, byte(ports.MsgOutput), 1, 0}
+	if _, _, _, _, ok := decodeData(truncated); ok {
+		t.Fatal("decodeData accepted truncated output skip header")
+	}
+}
+func TestDataRecordWireDecodeRejectsUnknownFlags(t *testing.T) {
+	withTrailingGarbage := []byte{recData, 0, 0, 0, 0, 0, 0, 0, 5, 1, byte(ports.MsgOutput), 0x80, 'x'}
+	if _, _, _, _, ok := decodeData(withTrailingGarbage); ok {
+		t.Fatal("decodeData accepted unknown data-record flags with trailing garbage")
+	}
+}
 func TestOutputRetransmitsUntilAck(t *testing.T) {
 	aPC, bPC := newPair()
 	var dropped atomic.Bool
 	}
 }
-func TestReliableDeliveryPreservesSequenceOrder(t *testing.T) {
+func TestReliableDeliveryPreservesInputSequenceOrder(t *testing.T) {
 	aPC, bPC := newPair()
 	a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)
 	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
 	defer func() { _ = b.Close() }()
 	for i := range 100 {
-		if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}); err != nil {
+		if err := a.Send(ports.Frame{Type: ports.MsgInput, Payload: []byte{byte(i)}}); err != nil {
 			t.Fatal(err)
 		}
 	}
 	for i := range 100 {
 		got := recvWithin(t, b, time.Second)
-		if got.Type != ports.MsgOutput || len(got.Payload) != 1 || got.Payload[0] != byte(i) {
+		if got.Type != ports.MsgInput || len(got.Payload) != 1 || got.Payload[0] != byte(i) {
 			t.Fatalf("frame %d delivered out of order: %+v", i, got)
 		}
 	}
 }
+func TestOutputSupersessionSkipsDroppedOlderOutput(t *testing.T) {
+	aPC, bPC := newPair()
+	var sendAttempts atomic.Int32
+	aPC.drop = func(_ []byte, addr net.Addr) bool {
+		return addr.String() == "b" && sendAttempts.Add(1) <= 2
+	}
+	a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)
+	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
+	defer func() { _ = a.Close() }()
+	defer func() { _ = b.Close() }()
+	for _, payload := range []string{"state1", "state2", "state3"} {
+		if err := a.Send(ports.Frame{Type: ports.MsgOutput, Payload: []byte(payload)}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	a.mu.Lock()
+	pending := len(a.pending)
+	a.mu.Unlock()
+	if pending > 1 {
+		t.Fatalf("pending outputs=%d, want collapsed to newest", pending)
+	}
+	got := recvWithin(t, b, time.Second)
+	if got.Type != ports.MsgOutput || string(got.Payload) != "state3" {
+		t.Fatalf("got=%+v, want newest output after skipped predecessors", got)
+	}
+	if sendAttempts.Load() < 3 {
+		t.Fatalf("send attempts=%d, want first two output sends dropped before newest arrived", sendAttempts.Load())
+	}
+}
 func TestAckProcessingContinuesWhenConsumerBackpressured(t *testing.T) {
 	aPC, bPC := newPair()
 	a, _ := NewTransport(aPC, testAddr("b"), key(), 1, 2)
 func TestReliableReceiverAcksOnlyHighestContiguousSequence(t *testing.T) {
 	tr := &Transport{maxRecvBuffer: maxRecvBuffer, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame)}
-	ackSeq, ack, queued := tr.enqueueReliable(3, ports.Frame{Type: ports.MsgOutput, Payload: []byte("third")})
+	ackSeq, ack, queued := tr.enqueueReliable(3, ports.Frame{Type: ports.MsgOutput, Payload: []byte("third")}, 0)
 	if !ack || !queued {
 		t.Fatalf("out-of-order enqueue ack=%v queued=%v, want true/true", ack, queued)
 	}
 		t.Fatalf("out-of-order ack=%d, want highest contiguous %d", got, want)
 	}
-	ackSeq, ack, queued = tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("first")})
+	ackSeq, ack, queued = tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("first")}, 0)
 	if !ack || !queued {
 		t.Fatalf("first enqueue ack=%v queued=%v, want true/true", ack, queued)
 	}
 		t.Fatalf("first contiguous ack=%d, want %d", got, want)
 	}
-	ackSeq, ack, queued = tr.enqueueReliable(2, ports.Frame{Type: ports.MsgOutput, Payload: []byte("second")})
+	ackSeq, ack, queued = tr.enqueueReliable(2, ports.Frame{Type: ports.MsgOutput, Payload: []byte("second")}, 0)
 	if !ack || !queued {
 		t.Fatalf("gap fill enqueue ack=%v queued=%v, want true/true", ack, queued)
 	}
 	tr := &Transport{nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), maxRecvBuffer: maxRecvBuffer, done: make(chan struct{})}
 	tr.deliverCond = sync.NewCond(&tr.deliverMu)
 	for i := range maxRecvBuffer + 100 {
-		tr.enqueueReliable(uint64(1000+i), ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}})
+		tr.enqueueReliable(uint64(1000+i), ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}, 0)
 	}
 	if got := len(tr.recvBuf); got != maxRecvBuffer {
 		t.Fatalf("recvBuf len=%d, want %d", got, maxRecvBuffer)
 	}
-	tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")})
+	tr.enqueueReliable(1, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}, 0)
 	if _, ok := tr.recvBuf[1]; !ok {
 		t.Fatalf("next expected sequence was dropped when far-future buffer was full")
 	}
 	}
 }
+func TestOutputSupersessionPrunesFullReceiveBuffer(t *testing.T) {
+	tr := &Transport{
+		nextRecvSeq:   1,
+		recvBuf:       map[uint64]ports.Frame{2: {Type: ports.MsgOutput, Payload: []byte("stale")}},
+		maxRecvBuffer: 1,
+		done:          make(chan struct{}),
+	}
+	tr.deliverCond = sync.NewCond(&tr.deliverMu)
+	ackSeq, ack, queued := tr.enqueueReliable(3, ports.Frame{Type: ports.MsgOutput, Payload: []byte("replacement")}, 2)
+	if !ack || !queued {
+		t.Fatalf("replacement ack=%v queued=%v, want true/true", ack, queued)
+	}
+	if ackSeq != 3 {
+		t.Fatalf("ack seq=%d, want 3 after skipped output", ackSeq)
+	}
+	if _, ok := tr.recvBuf[2]; ok {
+		t.Fatal("superseded output remained in full receive buffer")
+	}
+	if got := string(tr.recvBuf[3].Payload); got != "replacement" {
+		t.Fatalf("queued payload=%q, want replacement", got)
+	}
+}
 func TestReliableFullRecvBufferDoesNotAckDroppedFarFutureFrame(t *testing.T) {
 	aPC, bPC := newPair()
 	codec, err := pdgram.NewCodec(key())
 	for i := range maxRecvBuffer {
 		tr.recvBuf[uint64(1000+i)] = ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}
 	}
-	tr.handleRecord(encodeData(9999, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("dropped")}))
+	tr.handleRecord(encodeData(9999, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("dropped")}, 0))
 	select {
 	case pkt := <-bPC.in:
 		t.Fatalf("unexpected ACK packet for dropped frame: %x", pkt.b)
 	case <-time.After(25 * time.Millisecond):
 	}
-	tr.handleRecord(encodeData(1, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}))
+	tr.handleRecord(encodeData(1, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}, 0))
 	select {
 	case <-bPC.in:
 	case <-time.After(time.Second):

M internal/ports/frame.go +1 -1

@@ -1,7 +1,7 @@
1	1		package ports
2	2
3	3		// ProtocolVersion is the current vev IPC wire protocol version.
4		-	const ProtocolVersion uint16 = 11
	4	+	const ProtocolVersion uint16 = 12
5	5
6	6		// MsgType identifies the kind of payload carried by a Frame.
7	7		type MsgType uint8

@@ -1,7 +1,7 @@
1		package ports	1		package ports
2			2
3		// ProtocolVersion is the current vev IPC wire protocol version.	3		// ProtocolVersion is the current vev IPC wire protocol version.
4	-	const ProtocolVersion uint16 = 11	4	+	const ProtocolVersion uint16 = 12
5			5
6		// MsgType identifies the kind of payload carried by a Frame.	6		// MsgType identifies the kind of payload carried by a Frame.
7		type MsgType uint8	7		type MsgType uint8

M internal/ports/frame_test.go +2 -2


@@ -3,8 +3,8 @@ package ports
3	3		import "testing"
4	4
5	5		func TestProtocolVersion(t *testing.T) {
6		-	if ProtocolVersion != 11 {
7		-	t.Fatalf("ProtocolVersion = %d, want 11", ProtocolVersion)
	6	+	if ProtocolVersion != 12 {
	7	+	t.Fatalf("ProtocolVersion = %d, want 12", ProtocolVersion)
8	8		}
9	9		}
10	10


@@ -3,8 +3,8 @@ package ports
3		import "testing"	3		import "testing"
4			4
5		func TestProtocolVersion(t *testing.T) {	5		func TestProtocolVersion(t *testing.T) {
6	-	if ProtocolVersion != 11 {	6	+	if ProtocolVersion != 12 {
7	-	t.Fatalf("ProtocolVersion = %d, want 11", ProtocolVersion)	7	+	t.Fatalf("ProtocolVersion = %d, want 12", ProtocolVersion)
8		}	8		}
9		}	9		}
10			10