fix(dgram): wake delivery after output supersession
- commit
- 3389194e084ad6d5c1059198a41cfc25ca896701
- author
- brice · · 2026-07-09 18:34 UTC
- parent
- 386c1d8
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +33 -3
M internal/adapters/dgram/transport.go +6 -0
| @@ -599,6 +599,12 @@ func (t *Transport) enqueueReliable(seq uint64, f ports.Frame, outputSkipThrough | |||
| 599 | 599 | defer t.deliverMu.Unlock() | |
| 600 | 600 | if f.Type == ports.MsgOutput && outputSkipThrough > t.recvSkipThrough { | |
| 601 | 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() | |
| 602 | 608 | } | |
| 603 | 609 | if seq < t.nextRecvSeq { | |
| 604 | 610 | return t.highestContiguousRecvLocked(), true, false | |
| @@ -599,6 +599,12 @@ func (t *Transport) enqueueReliable(seq uint64, f ports.Frame, outputSkipThrough | |||||
| 599 | defer t.deliverMu.Unlock() | 599 | defer t.deliverMu.Unlock() | ||
| 600 | if f.Type == ports.MsgOutput && outputSkipThrough > t.recvSkipThrough { | 600 | if f.Type == ports.MsgOutput && outputSkipThrough > t.recvSkipThrough { | ||
| 601 | t.recvSkipThrough = outputSkipThrough | 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() | |||
| 602 | } | 608 | } | ||
| 603 | if seq < t.nextRecvSeq { | 609 | if seq < t.nextRecvSeq { | ||
| 604 | return t.highestContiguousRecvLocked(), true, false | 610 | return t.highestContiguousRecvLocked(), true, false | ||
M internal/adapters/dgram/transport_test.go +24 -0
| @@ -1039,6 +1039,30 @@ func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) { | |||
| 1039 | 1039 | } | |
| 1040 | 1040 | } | |
| 1041 | 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 | + | ||
| 1042 | 1066 | func TestReliableFullRecvBufferDoesNotAckDroppedFarFutureFrame(t *testing.T) { | |
| 1043 | 1067 | aPC, bPC := newPair() | |
| 1044 | 1068 | codec, err := pdgram.NewCodec(key()) | |
| @@ -1039,6 +1039,30 @@ func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) { | |||||
| 1039 | } | 1039 | } | ||
| 1040 | } | 1040 | } | ||
| 1041 | 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 | + | ||||
| 1042 | func TestReliableFullRecvBufferDoesNotAckDroppedFarFutureFrame(t *testing.T) { | 1066 | func TestReliableFullRecvBufferDoesNotAckDroppedFarFutureFrame(t *testing.T) { | ||
| 1043 | aPC, bPC := newPair() | 1067 | aPC, bPC := newPair() | ||
| 1044 | codec, err := pdgram.NewCodec(key()) | 1068 | codec, err := pdgram.NewCodec(key()) | ||
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 | ||||