fix(dgram): route acknowledgements through control path
- commit
- c3dcda7f49508efa5663e00254465f0fc1ae17ef
- author
- brice · · 2026-07-15 04:56 UTC
- parent
- ad8c50b
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +39 -35
M internal/adapters/dgram/control.go +1 -35
| @@ -216,39 +216,5 @@ func (t *Transport) sendControl(kind byte, id uint64) error { | |||
| 216 | 216 | } | |
| 217 | 217 | ||
| 218 | 218 | func (t *Transport) sendAck(seq uint64) { | |
| 219 | - | var b [9]byte | |
| 220 | - | b[0] = recAck | |
| 221 | - | binary.BigEndian.PutUint64(b[1:], seq) | |
| 222 | - | _ = t.sendPayload(b[:]) | |
| 223 | - | } | |
| 224 | - | func (t *Transport) sendPayload(p []byte) error { | |
| 225 | - | frags, err := pdgram.FragmentPayload(t.nextCounter(), p, t.mtu-pdgram.HeaderSize-t.codec.Overhead()) | |
| 226 | - | if err != nil { | |
| 227 | - | return err | |
| 228 | - | } | |
| 229 | - | for _, f := range frags { | |
| 230 | - | raw, err := pdgram.MarshalFragment(f) | |
| 231 | - | if err != nil { | |
| 232 | - | return err | |
| 233 | - | } | |
| 234 | - | pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil) | |
| 235 | - | t.mu.Lock() | |
| 236 | - | peer := t.peer | |
| 237 | - | closed := t.closed | |
| 238 | - | closeErr := t.closeErr | |
| 239 | - | t.mu.Unlock() | |
| 240 | - | if closed { | |
| 241 | - | if closeErr != nil { | |
| 242 | - | return closeErr | |
| 243 | - | } | |
| 244 | - | return errors.New("dgram: closed") | |
| 245 | - | } | |
| 246 | - | if peer == nil { | |
| 247 | - | return errors.New("dgram: no peer") | |
| 248 | - | } | |
| 249 | - | if err := t.writePacket(pkt, peer); err != nil { | |
| 250 | - | return err | |
| 251 | - | } | |
| 252 | - | } | |
| 253 | - | return nil | |
| 219 | + | _ = t.sendControl(recAck, seq) | |
| 254 | 220 | } | |
| @@ -216,39 +216,5 @@ func (t *Transport) sendControl(kind byte, id uint64) error { | |||||
| 216 | } | 216 | } | ||
| 217 | 217 | ||||
| 218 | func (t *Transport) sendAck(seq uint64) { | 218 | func (t *Transport) sendAck(seq uint64) { | ||
| 219 | - | var b [9]byte | 219 | + | _ = t.sendControl(recAck, seq) |
| 220 | - | b[0] = recAck | |||
| 221 | - | binary.BigEndian.PutUint64(b[1:], seq) | |||
| 222 | - | _ = t.sendPayload(b[:]) | |||
| 223 | - | } | |||
| 224 | - | func (t *Transport) sendPayload(p []byte) error { | |||
| 225 | - | frags, err := pdgram.FragmentPayload(t.nextCounter(), p, t.mtu-pdgram.HeaderSize-t.codec.Overhead()) | |||
| 226 | - | if err != nil { | |||
| 227 | - | return err | |||
| 228 | - | } | |||
| 229 | - | for _, f := range frags { | |||
| 230 | - | raw, err := pdgram.MarshalFragment(f) | |||
| 231 | - | if err != nil { | |||
| 232 | - | return err | |||
| 233 | - | } | |||
| 234 | - | pkt := t.codec.Seal(t.sendDir, t.nextCounter(), raw, nil) | |||
| 235 | - | t.mu.Lock() | |||
| 236 | - | peer := t.peer | |||
| 237 | - | closed := t.closed | |||
| 238 | - | closeErr := t.closeErr | |||
| 239 | - | t.mu.Unlock() | |||
| 240 | - | if closed { | |||
| 241 | - | if closeErr != nil { | |||
| 242 | - | return closeErr | |||
| 243 | - | } | |||
| 244 | - | return errors.New("dgram: closed") | |||
| 245 | - | } | |||
| 246 | - | if peer == nil { | |||
| 247 | - | return errors.New("dgram: no peer") | |||
| 248 | - | } | |||
| 249 | - | if err := t.writePacket(pkt, peer); err != nil { | |||
| 250 | - | return err | |||
| 251 | - | } | |||
| 252 | - | } | |||
| 253 | - | return nil | |||
| 254 | } | 220 | } | ||
M internal/adapters/dgram/control_plane_test.go +38 -0
| @@ -208,6 +208,44 @@ func TestACKDeadlineDispatchesWhileAdvisoryWriteIsBlocked(t *testing.T) { | |||
| 208 | 208 | awaitSignal(t, ackStarted, "ACK write while advisory probe remains blocked") | |
| 209 | 209 | } | |
| 210 | 210 | ||
| 211 | + | func TestACKBypassesDataWriterAndRemainsDecodable(t *testing.T) { | |
| 212 | + | aPC, bPC := newPair() | |
| 213 | + | a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{ | |
| 214 | + | ResendAfter: time.Hour, Heartbeat: time.Hour, | |
| 215 | + | }) | |
| 216 | + | if err != nil { | |
| 217 | + | t.Fatal(err) | |
| 218 | + | } | |
| 219 | + | defer func() { _ = a.Close() }() | |
| 220 | + | ||
| 221 | + | acks := make(chan uint64, 1) | |
| 222 | + | aPC.drop = func(pkt []byte, _ net.Addr) bool { | |
| 223 | + | _, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil) | |
| 224 | + | if err != nil { | |
| 225 | + | t.Errorf("open ACK: %v", err) | |
| 226 | + | return true | |
| 227 | + | } | |
| 228 | + | frag, err := pdgram.UnmarshalFragment(raw) | |
| 229 | + | if err != nil { | |
| 230 | + | t.Errorf("unmarshal ACK: %v", err) | |
| 231 | + | return true | |
| 232 | + | } | |
| 233 | + | if frag.Count != 1 || len(frag.Data) != 9 || frag.Data[0] != recAck { | |
| 234 | + | t.Errorf("control record=%x, want encoded ACK", frag.Data) | |
| 235 | + | return true | |
| 236 | + | } | |
| 237 | + | acks <- binary.BigEndian.Uint64(frag.Data[1:]) | |
| 238 | + | return true | |
| 239 | + | } | |
| 240 | + | ||
| 241 | + | a.writeMu.Lock() | |
| 242 | + | defer a.writeMu.Unlock() | |
| 243 | + | go a.sendAck(42) | |
| 244 | + | if seq := awaitResult(t, acks, "ACK while data writer is blocked"); seq != 42 { | |
| 245 | + | t.Fatalf("ACK sequence=%d, want 42", seq) | |
| 246 | + | } | |
| 247 | + | } | |
| 248 | + | ||
| 211 | 249 | func TestProbeDoesNotBlockOnDataWriter(t *testing.T) { | |
| 212 | 250 | aPC, bPC := newPair() | |
| 213 | 251 | a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{ | |
| @@ -208,6 +208,44 @@ func TestACKDeadlineDispatchesWhileAdvisoryWriteIsBlocked(t *testing.T) { | |||||
| 208 | awaitSignal(t, ackStarted, "ACK write while advisory probe remains blocked") | 208 | awaitSignal(t, ackStarted, "ACK write while advisory probe remains blocked") | ||
| 209 | } | 209 | } | ||
| 210 | 210 | ||||
| 211 | + | func TestACKBypassesDataWriterAndRemainsDecodable(t *testing.T) { | |||
| 212 | + | aPC, bPC := newPair() | |||
| 213 | + | a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{ | |||
| 214 | + | ResendAfter: time.Hour, Heartbeat: time.Hour, | |||
| 215 | + | }) | |||
| 216 | + | if err != nil { | |||
| 217 | + | t.Fatal(err) | |||
| 218 | + | } | |||
| 219 | + | defer func() { _ = a.Close() }() | |||
| 220 | + | ||||
| 221 | + | acks := make(chan uint64, 1) | |||
| 222 | + | aPC.drop = func(pkt []byte, _ net.Addr) bool { | |||
| 223 | + | _, raw, err := a.codec.Open(pkt, a.sendDir, nil, nil) | |||
| 224 | + | if err != nil { | |||
| 225 | + | t.Errorf("open ACK: %v", err) | |||
| 226 | + | return true | |||
| 227 | + | } | |||
| 228 | + | frag, err := pdgram.UnmarshalFragment(raw) | |||
| 229 | + | if err != nil { | |||
| 230 | + | t.Errorf("unmarshal ACK: %v", err) | |||
| 231 | + | return true | |||
| 232 | + | } | |||
| 233 | + | if frag.Count != 1 || len(frag.Data) != 9 || frag.Data[0] != recAck { | |||
| 234 | + | t.Errorf("control record=%x, want encoded ACK", frag.Data) | |||
| 235 | + | return true | |||
| 236 | + | } | |||
| 237 | + | acks <- binary.BigEndian.Uint64(frag.Data[1:]) | |||
| 238 | + | return true | |||
| 239 | + | } | |||
| 240 | + | ||||
| 241 | + | a.writeMu.Lock() | |||
| 242 | + | defer a.writeMu.Unlock() | |||
| 243 | + | go a.sendAck(42) | |||
| 244 | + | if seq := awaitResult(t, acks, "ACK while data writer is blocked"); seq != 42 { | |||
| 245 | + | t.Fatalf("ACK sequence=%d, want 42", seq) | |||
| 246 | + | } | |||
| 247 | + | } | |||
| 248 | + | ||||
| 211 | func TestProbeDoesNotBlockOnDataWriter(t *testing.T) { | 249 | func TestProbeDoesNotBlockOnDataWriter(t *testing.T) { | ||
| 212 | aPC, bPC := newPair() | 250 | aPC, bPC := newPair() | ||
| 213 | a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{ | 251 | a, err := NewTransportWithOptions(aPC, bPC.addr, key(), 1, 2, Options{ | ||