56875c7 fix(remote): preserve transport identity on sends · vev

M internal/adapters/dgram/transport.go +8 -6


@@ -288,26 +288,27 @@ func (t *Transport) handleRecord(p []byte) {
288	288		return
289	289		}
290	290		if reliable {
291		-	t.sendAck(seq)
292		-	t.enqueueReliable(seq, f)
	291	+	if t.enqueueReliable(seq, f) {
	292	+	t.sendAck(seq)
	293	+	}
293	294		return
294	295		}
295	296		t.deliver(f)
296	297		}
297	298		}
298	299
299		-	func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) {
	300	+	func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) bool {
300	301		t.deliverMu.Lock()
301	302		defer t.deliverMu.Unlock()
302	303		if seq < t.nextRecvSeq {
303		-	return
	304	+	return true
304	305		}
305	306		if _, exists := t.recvBuf[seq]; exists {
306		-	return
	307	+	return true
307	308		}
308	309		if len(t.recvBuf) >= maxRecvBuffer {
309	310		if seq != t.nextRecvSeq {
310		-	return
	311	+	return false
311	312		}
312	313		for bufferedSeq := range t.recvBuf {
313	314		delete(t.recvBuf, bufferedSeq)

@@ -316,6 +317,7 @@ func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) {
316	317		}
317	318		t.recvBuf[seq] = f
318	319		t.deliverCond.Signal()
	320	+	return true
319	321		}
320	322
321	323		func (t *Transport) deliveryLoop() {

 			return
 		}
 		if reliable {
-			t.sendAck(seq)
+			if t.enqueueReliable(seq, f) {
-			t.enqueueReliable(seq, f)
+				t.sendAck(seq)
+			}
 			return
 		}
 		t.deliver(f)
 	}
 }
-func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) {
+func (t *Transport) enqueueReliable(seq uint64, f ports.Frame) bool {
 	t.deliverMu.Lock()
 	defer t.deliverMu.Unlock()
 	if seq < t.nextRecvSeq {
-		return
+		return true
 	}
 	if _, exists := t.recvBuf[seq]; exists {
-		return
+		return true
 	}
 	if len(t.recvBuf) >= maxRecvBuffer {
 		if seq != t.nextRecvSeq {
-			return
+			return false
 		}
 		for bufferedSeq := range t.recvBuf {
 			delete(t.recvBuf, bufferedSeq)
 	}
 	t.recvBuf[seq] = f
 	t.deliverCond.Signal()
+	return true
 }
 func (t *Transport) deliveryLoop() {

M internal/adapters/dgram/transport_test.go +29 -0


@@ -337,6 +337,35 @@ func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) {
337	337		}
338	338		}
339	339
	340	+	func TestReliableFullRecvBufferDoesNotAckDroppedFarFutureFrame(t *testing.T) {
	341	+	aPC, bPC := newPair()
	342	+	codec, err := pdgram.NewCodec(key())
	343	+	if err != nil {
	344	+	t.Fatal(err)
	345	+	}
	346	+	tr := &Transport{pc: aPC, peer: testAddr("b"), codec: codec, sendDir: 1, mtu: pdgram.DefaultMTU, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), done: make(chan struct{})}
	347	+	tr.deliverCond = sync.NewCond(&tr.deliverMu)
	348	+	defer func() { _ = aPC.Close() }()
	349	+	defer func() { _ = bPC.Close() }()
	350	+
	351	+	for i := 0; i < maxRecvBuffer; i++ {
	352	+	tr.recvBuf[uint64(1000+i)] = ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}
	353	+	}
	354	+	tr.handleRecord(encodeData(9999, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("dropped")}))
	355	+	select {
	356	+	case pkt := <-bPC.in:
	357	+	t.Fatalf("unexpected ACK packet for dropped frame: %x", pkt.b)
	358	+	case <-time.After(25 * time.Millisecond):
	359	+	}
	360	+
	361	+	tr.handleRecord(encodeData(1, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}))
	362	+	select {
	363	+	case <-bPC.in:
	364	+	case <-time.After(time.Second):
	365	+	t.Fatal("expected ACK for buffered contiguous frame")
	366	+	}
	367	+	}
	368	+
340	369		func recvMaybe(tr *Transport, d time.Duration) (ports.Frame, bool) {
341	370		ch := make(chan ports.Frame, 1)
342	371		go func() { f, _ := tr.Recv(); ch <- f }()


@@ -337,6 +337,35 @@ func TestReliableRecvBufferBoundedForFarFutureSequences(t *testing.T) {
337	}	337		}
338	}	338		}
339		339
		340	+	func TestReliableFullRecvBufferDoesNotAckDroppedFarFutureFrame(t *testing.T) {
		341	+	aPC, bPC := newPair()
		342	+	codec, err := pdgram.NewCodec(key())
		343	+	if err != nil {
		344	+	t.Fatal(err)
		345	+	}
		346	+	tr := &Transport{pc: aPC, peer: testAddr("b"), codec: codec, sendDir: 1, mtu: pdgram.DefaultMTU, nextRecvSeq: 1, recvBuf: make(map[uint64]ports.Frame), done: make(chan struct{})}
		347	+	tr.deliverCond = sync.NewCond(&tr.deliverMu)
		348	+	defer func() { _ = aPC.Close() }()
		349	+	defer func() { _ = bPC.Close() }()
		350	+
		351	+	for i := 0; i < maxRecvBuffer; i++ {
		352	+	tr.recvBuf[uint64(1000+i)] = ports.Frame{Type: ports.MsgOutput, Payload: []byte{byte(i)}}
		353	+	}
		354	+	tr.handleRecord(encodeData(9999, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("dropped")}))
		355	+	select {
		356	+	case pkt := <-bPC.in:
		357	+	t.Fatalf("unexpected ACK packet for dropped frame: %x", pkt.b)
		358	+	case <-time.After(25 * time.Millisecond):
		359	+	}
		360	+
		361	+	tr.handleRecord(encodeData(1, true, ports.Frame{Type: ports.MsgOutput, Payload: []byte("next")}))
		362	+	select {
		363	+	case <-bPC.in:
		364	+	case <-time.After(time.Second):
		365	+	t.Fatal("expected ACK for buffered contiguous frame")
		366	+	}
		367	+	}
		368	+
340	func recvMaybe(tr *Transport, d time.Duration) (ports.Frame, bool) {	369		func recvMaybe(tr *Transport, d time.Duration) (ports.Frame, bool) {
341	ch := make(chan ports.Frame, 1)	370		ch := make(chan ports.Frame, 1)
342	go func() { f, _ := tr.Recv(); ch <- f }()	371		go func() { f, _ := tr.Recv(); ch <- f }()

M internal/usecase/daemon/client.go +39 -20


@@ -208,13 +208,18 @@ func (ac *attachedClient) nextOutputFrameLocked(b []byte) ports.Frame {
208	208
209	209		// send serialises a frame onto the client's transport.
210	210		func (ac *attachedClient) send(f ports.Frame) error {
	211	+	_, err := ac.sendTransport(f)
	212	+	return err
	213	+	}
	214	+
	215	+	func (ac *attachedClient) sendTransport(f ports.Frame) (ports.Transport, error) {
211	216		ac.sendMu.Lock()
212	217		defer ac.sendMu.Unlock()
213	218		tr := ac.transport()
214	219		if tr == nil {
215		-	return errors.New("client transport is nil")
	220	+	return nil, errors.New("client transport is nil")
216	221		}
217		-	return tr.Send(f)
	222	+	return tr, tr.Send(f)
218	223		}
219	224
220	225		// boundedSend sends f to ac with a deadline watchdog: if the send (including

@@ -227,7 +232,15 @@ func (d Daemon) boundedSend(ac attachedClient, f ports.Frame) {
227	232		}
228	233
229	234		func (d Daemon) boundedSendErr(ac attachedClient, f ports.Frame) error {
230		-	return d.boundedSendWith(ac, func() error { return ac.send(f) })
	235	+	tr, err := d.boundedSendWith(ac, func(capture func(ports.Transport)) error {
	236	+	tr, err := ac.sendTransport(f)
	237	+	capture(tr)
	238	+	return err
	239	+	})
	240	+	if err != nil && errors.Is(err, errSendTimedOut) {
	241	+	_ = ac.closeCapturedTransport(tr)
	242	+	}
	243	+	return err
231	244		}
232	245
233	246		func (d Daemon) boundedSendOutputErr(ac attachedClient, b []byte) error {

@@ -236,45 +249,51 @@ func (d Daemon) boundedSendOutputErr(ac attachedClient, b []byte) error {
236	249		}
237	250
238	251		func (d Daemon) boundedSendOutputErrTransport(ac attachedClient, b []byte) (ports.Transport, error) {
239		-	var (
240		-	usedMu sync.Mutex
241		-	used ports.Transport
242		-	)
243		-	err := d.boundedSendWith(ac, func() error {
	252	+	return d.boundedSendWith(ac, func(capture func(ports.Transport)) error {
244	253		ac.sendMu.Lock()
245	254		defer ac.sendMu.Unlock()
246	255		tr := ac.transport()
247		-	usedMu.Lock()
248		-	used = tr
249		-	usedMu.Unlock()
	256	+	capture(tr)
250	257		if tr == nil {
251	258		return errors.New("client transport is nil")
252	259		}
253	260		return tr.Send(ac.nextOutputFrameLocked(b))
254	261		})
255		-	usedMu.Lock()
256		-	defer usedMu.Unlock()
257		-	return used, err
258	262		}
259	263
260		-	func (d Daemon) boundedSendWith(ac attachedClient, send func() error) error {
	264	+	var errSendTimedOut = errors.New("send timed out")
	265	+
	266	+	func (d Daemon) boundedSendWith(ac attachedClient, send func(capture func(ports.Transport)) error) (ports.Transport, error) {
261	267		timer := d.clock.NewTimer(detachNotifyTimeout)
262	268		result := make(chan error, 1)
	269	+	var (
	270	+	capturedMu sync.Mutex
	271	+	captured ports.Transport
	272	+	)
	273	+	capture := func(tr ports.Transport) {
	274	+	capturedMu.Lock()
	275	+	captured = tr
	276	+	capturedMu.Unlock()
	277	+	}
	278	+	capturedTransport := func() ports.Transport {
	279	+	capturedMu.Lock()
	280	+	defer capturedMu.Unlock()
	281	+	return captured
	282	+	}
263	283		go func() {
264		-	result <- send()
	284	+	result <- send(capture)
265	285		}()
266	286		select {
267	287		case err := <-result:
268	288		timer.Stop()
269		-	return err
	289	+	return capturedTransport(), err
270	290		case <-timer.C():
271	291		select {
272	292		case err := <-result:
273		-	return err
	293	+	return capturedTransport(), err
274	294		default:
275	295		}
276		-	_ = ac.closeTransport()
277		-	return errors.New("send timed out")
	296	+	return capturedTransport(), errSendTimedOut
278	297		}
279	298		}
280	299

 // send serialises a frame onto the client's transport.
 func (ac *attachedClient) send(f ports.Frame) error {
+	_, err := ac.sendTransport(f)
+	return err
+}
+func (ac *attachedClient) sendTransport(f ports.Frame) (ports.Transport, error) {
 	ac.sendMu.Lock()
 	defer ac.sendMu.Unlock()
 	tr := ac.transport()
 	if tr == nil {
-		return errors.New("client transport is nil")
+		return nil, errors.New("client transport is nil")
 	}
-	return tr.Send(f)
+	return tr, tr.Send(f)
 }
 // boundedSend sends f to ac with a deadline watchdog: if the send (including
 }
 func (d *Daemon) boundedSendErr(ac *attachedClient, f ports.Frame) error {
-	return d.boundedSendWith(ac, func() error { return ac.send(f) })
+	tr, err := d.boundedSendWith(ac, func(capture func(ports.Transport)) error {
+		tr, err := ac.sendTransport(f)
+		capture(tr)
+		return err
+	})
+	if err != nil && errors.Is(err, errSendTimedOut) {
+		_ = ac.closeCapturedTransport(tr)
+	}
+	return err
 }
 func (d *Daemon) boundedSendOutputErr(ac *attachedClient, b []byte) error {
 }
 func (d *Daemon) boundedSendOutputErrTransport(ac *attachedClient, b []byte) (ports.Transport, error) {
-	var (
+	return d.boundedSendWith(ac, func(capture func(ports.Transport)) error {
-		usedMu sync.Mutex
-		used   ports.Transport
-	)
-	err := d.boundedSendWith(ac, func() error {
 		ac.sendMu.Lock()
 		defer ac.sendMu.Unlock()
 		tr := ac.transport()
-		usedMu.Lock()
+		capture(tr)
-		used = tr
-		usedMu.Unlock()
 		if tr == nil {
 			return errors.New("client transport is nil")
 		}
 		return tr.Send(ac.nextOutputFrameLocked(b))
 	})
-	usedMu.Lock()
-	defer usedMu.Unlock()
-	return used, err
 }
-func (d *Daemon) boundedSendWith(ac *attachedClient, send func() error) error {
+var errSendTimedOut = errors.New("send timed out")
+func (d *Daemon) boundedSendWith(ac *attachedClient, send func(capture func(ports.Transport)) error) (ports.Transport, error) {
 	timer := d.clock.NewTimer(detachNotifyTimeout)
 	result := make(chan error, 1)
+	var (
+		capturedMu sync.Mutex
+		captured   ports.Transport
+	)
+	capture := func(tr ports.Transport) {
+		capturedMu.Lock()
+		captured = tr
+		capturedMu.Unlock()
+	}
+	capturedTransport := func() ports.Transport {
+		capturedMu.Lock()
+		defer capturedMu.Unlock()
+		return captured
+	}
 	go func() {
-		result <- send()
+		result <- send(capture)
 	}()
 	select {
 	case err := <-result:
 		timer.Stop()
-		return err
+		return capturedTransport(), err
 	case <-timer.C():
 		select {
 		case err := <-result:
-			return err
+			return capturedTransport(), err
 		default:
 		}
-		_ = ac.closeTransport()
+		return capturedTransport(), errSendTimedOut
-		return errors.New("send timed out")
 	}
 }