fix(remote): bound connection handoff retries

commit: 55169b26322e3cae78ed30eace5414b71560e099
author: brice · 2mo ago · 2026-07-31 21:43 UTC
parent: 01e78fb
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +73 -4

M internal/usecase/daemon/client_frame_routing.go +12 -3


@@ -8,11 +8,16 @@ import (
8	8		"github.com/bnema/vev/internal/usecase/command"
9	9		)
10	10
	11	+	const connectionRoleAttempts = 8
	12	+
11	13		// currentConnectionRole retries the lock-free session-to-role snapshot when a
12	14		// handoff lands between those reads. The role gate still performs the final
13	15		// mutation admission after this function returns.
14	16		func (d Daemon) currentConnectionRole(ac attachedClient, tr ports.Transport) (attachmentSession, attachmentRoleToken, bool) {
15		-	for ac.currentTransportIs(tr) {
	17	+	for range connectionRoleAttempts {
	18	+	if !ac.currentTransportIs(tr) {
	19	+	break
	20	+	}
16	21		sess := ac.currentAttachmentSession()
17	22		if sess == nil {
18	23		return nil, attachmentRoleToken{}, false

@@ -45,7 +50,7 @@ func (d Daemon) runConnLoop(ac attachedClient) {
45	50		}
46	51		f, err := tr.Recv()
47	52		if err != nil {
48		-	for {
	53	+	for range connectionRoleAttempts {
49	54		sess, token, ok := d.currentConnectionRole(ac, tr)
50	55		if !ok {
51	56		return

@@ -56,11 +61,15 @@ func (d Daemon) runConnLoop(ac attachedClient) {
56	61		d.detachProxyOnSendError(proxy, ac, tr)
57	62		} else if local, ok := localSession(sess); ok {
58	63		d.clientGone(local, ac, tr, false)
	64	+	} else {
	65	+	return
59	66		}
60		-	if !ac.currentTransportIs(tr) \|\| ac.currentAttachmentSession() == nil {
	67	+	current := ac.currentAttachmentSession()
	68	+	if current == sess \|\| !ac.currentTransportIs(tr) \|\| current == nil {
61	69		return
62	70		}
63	71		}
	72	+	return
64	73		}
65	74		_, token, ok := d.currentConnectionRole(ac, tr)
66	75		if !ok {

 	"github.com/bnema/vev/internal/usecase/command"
 )
+const connectionRoleAttempts = 8
 // currentConnectionRole retries the lock-free session-to-role snapshot when a
 // handoff lands between those reads. The role gate still performs the final
 // mutation admission after this function returns.
 func (d *Daemon) currentConnectionRole(ac *attachedClient, tr ports.Transport) (attachmentSession, attachmentRoleToken, bool) {
-	for ac.currentTransportIs(tr) {
+	for range connectionRoleAttempts {
+		if !ac.currentTransportIs(tr) {
+			break
+		}
 		sess := ac.currentAttachmentSession()
 		if sess == nil {
 			return nil, attachmentRoleToken{}, false
 		}
 		f, err := tr.Recv()
 		if err != nil {
-			for {
+			for range connectionRoleAttempts {
 				sess, token, ok := d.currentConnectionRole(ac, tr)
 				if !ok {
 					return
 					d.detachProxyOnSendError(proxy, ac, tr)
 				} else if local, ok := localSession(sess); ok {
 					d.clientGone(local, ac, tr, false)
+				} else {
+					return
 				}
-				if !ac.currentTransportIs(tr) || ac.currentAttachmentSession() == nil {
+				current := ac.currentAttachmentSession()
+				if current == sess || !ac.currentTransportIs(tr) || current == nil {
 					return
 				}
 			}
+			return
 		}
 		_, token, ok := d.currentConnectionRole(ac, tr)
 		if !ok {

M internal/usecase/daemon/proxy_input_test.go +61 -1


@@ -830,6 +830,64 @@ func TestRemotePickerCanReselectWarmProxyAfterReturningLocal(t *testing.T) {
830	830		require.False(t, ac.overlays.pickerActive())
831	831		}
832	832
	833	+	func TestConnectionRoleResolutionStopsAfterUnstableSnapshots(t *testing.T) {
	834	+	d := newTestDaemon(t, nil, stubClock{})
	835	+	client := newProxyTestTransport()
	836	+	ac := &attachedClient{}
	837	+	ac.replaceTransport(client)
	838	+	first := &session{sessionCore: sessionCore{client: ac}}
	839	+	second := &session{sessionCore: sessionCore{client: ac}}
	840	+	ac.setSession(first)
	841	+
	842	+	const unstableSnapshots = 100
	843	+	attempts := 0
	844	+	d.afterConnectionSessionSnapshot = func(snapshot attachmentSession) {
	845	+	attempts++
	846	+	if attempts >= unstableSnapshots {
	847	+	return
	848	+	}
	849	+	if snapshot == first {
	850	+	ac.setSession(second)
	851	+	} else {
	852	+	ac.setSession(first)
	853	+	}
	854	+	}
	855	+
	856	+	_, _, ok := d.currentConnectionRole(ac, client)
	857	+	require.False(t, ok)
	858	+	require.Less(t, attempts, unstableSnapshots)
	859	+	}
	860	+
	861	+	func TestConnectionLoopStopsWhenReceiveErrorCleanupMakesNoProgress(t *testing.T) {
	862	+	d := newTestDaemon(t, nil, stubClock{})
	863	+	client := newProxyTestTransport()
	864	+	ac := &attachedClient{}
	865	+	ac.replaceTransport(client)
	866	+	local := &session{sessionCore: sessionCore{client: ac}}
	867	+	ac.setSession(local)
	868	+
	869	+	cleanupAttempts := 0
	870	+	d.beforeClientGoneDetach = func() {
	871	+	cleanupAttempts++
	872	+	local.mu.Lock()
	873	+	local.client = &attachedClient{}
	874	+	local.mu.Unlock()
	875	+	}
	876	+	snapshots := 0
	877	+	d.afterConnectionSessionSnapshot = func(attachmentSession) {
	878	+	snapshots++
	879	+	if snapshots == 100 {
	880	+	ac.setSession(nil)
	881	+	}
	882	+	}
	883	+	client.recv <- proxyRecv{err: io.EOF}
	884	+
	885	+	d.runConnLoop(ac)
	886	+
	887	+	require.Equal(t, 1, cleanupAttempts)
	888	+	require.Equal(t, 1, snapshots)
	889	+	}
	890	+
833	891		func TestConnectionLoopRetriesRoleResolutionAcrossHandoff(t *testing.T) {
834	892		d, local, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
835	893		client := newProxyTestTransport()

@@ -843,10 +901,11 @@ func TestConnectionLoopRetriesRoleResolutionAcrossHandoff(t *testing.T) {
843	901		d.remoteDialerFactory = newProxyConstructionFactory(remote)
844	902		d.remoteTransportMode = ports.RemoteTransportUDP
845	903		token := roleEffectForTest(t, local.attachmentToken(ac, client))
	904	+	snapshots := make(chan attachmentSession, 1)
846	905		transitionErr := make(chan error, 1)
847	906		d.afterConnectionSessionSnapshot = func(snapshot attachmentSession) {
848	907		d.afterConnectionSessionSnapshot = nil
849		-	require.Same(t, local, snapshot)
	908	+	snapshots <- snapshot
850	909		transitionErr <- d.switchToTargetForRole(token, picker.Target{Session: key.ID(), RemoteKey: &key, TabIndex: -1}, sessionHandoffGuard{}, "picker-select")
851	910		}
852	911

@@ -856,6 +915,7 @@ func TestConnectionLoopRetriesRoleResolutionAcrossHandoff(t *testing.T) {
856	915		close(loopDone)
857	916		}()
858	917		client.recv <- proxyRecv{frame: ports.Frame{Type: ports.MsgPing, Payload: ports.MarshalPing(ports.Ping{})}}
	918	+	require.Same(t, local, awaitTestValue(t, snapshots, "connection role was not snapshotted"))
859	919		require.NoError(t, awaitTestValue(t, transitionErr, "handoff did not run during role resolution"))
860	920		proxy, ok := ac.currentAttachmentSession().(*proxySession)
861	921		require.True(t, ok)


@@ -830,6 +830,64 @@ func TestRemotePickerCanReselectWarmProxyAfterReturningLocal(t *testing.T) {
830		require.False(t, ac.overlays.pickerActive())	830		require.False(t, ac.overlays.pickerActive())
831		}	831		}
832			832
			833	+	func TestConnectionRoleResolutionStopsAfterUnstableSnapshots(t *testing.T) {
			834	+	d := newTestDaemon(t, nil, stubClock{})
			835	+	client := newProxyTestTransport()
			836	+	ac := &attachedClient{}
			837	+	ac.replaceTransport(client)
			838	+	first := &session{sessionCore: sessionCore{client: ac}}
			839	+	second := &session{sessionCore: sessionCore{client: ac}}
			840	+	ac.setSession(first)
			841	+
			842	+	const unstableSnapshots = 100
			843	+	attempts := 0
			844	+	d.afterConnectionSessionSnapshot = func(snapshot attachmentSession) {
			845	+	attempts++
			846	+	if attempts >= unstableSnapshots {
			847	+	return
			848	+	}
			849	+	if snapshot == first {
			850	+	ac.setSession(second)
			851	+	} else {
			852	+	ac.setSession(first)
			853	+	}
			854	+	}
			855	+
			856	+	_, _, ok := d.currentConnectionRole(ac, client)
			857	+	require.False(t, ok)
			858	+	require.Less(t, attempts, unstableSnapshots)
			859	+	}
			860	+
			861	+	func TestConnectionLoopStopsWhenReceiveErrorCleanupMakesNoProgress(t *testing.T) {
			862	+	d := newTestDaemon(t, nil, stubClock{})
			863	+	client := newProxyTestTransport()
			864	+	ac := &attachedClient{}
			865	+	ac.replaceTransport(client)
			866	+	local := &session{sessionCore: sessionCore{client: ac}}
			867	+	ac.setSession(local)
			868	+
			869	+	cleanupAttempts := 0
			870	+	d.beforeClientGoneDetach = func() {
			871	+	cleanupAttempts++
			872	+	local.mu.Lock()
			873	+	local.client = &attachedClient{}
			874	+	local.mu.Unlock()
			875	+	}
			876	+	snapshots := 0
			877	+	d.afterConnectionSessionSnapshot = func(attachmentSession) {
			878	+	snapshots++
			879	+	if snapshots == 100 {
			880	+	ac.setSession(nil)
			881	+	}
			882	+	}
			883	+	client.recv <- proxyRecv{err: io.EOF}
			884	+
			885	+	d.runConnLoop(ac)
			886	+
			887	+	require.Equal(t, 1, cleanupAttempts)
			888	+	require.Equal(t, 1, snapshots)
			889	+	}
			890	+
833		func TestConnectionLoopRetriesRoleResolutionAcrossHandoff(t *testing.T) {	891		func TestConnectionLoopRetriesRoleResolutionAcrossHandoff(t *testing.T) {
834		d, local, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())	892		d, local, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
835		client := newProxyTestTransport()	893		client := newProxyTestTransport()

@@ -843,10 +901,11 @@ func TestConnectionLoopRetriesRoleResolutionAcrossHandoff(t *testing.T) {
843		d.remoteDialerFactory = newProxyConstructionFactory(remote)	901		d.remoteDialerFactory = newProxyConstructionFactory(remote)
844		d.remoteTransportMode = ports.RemoteTransportUDP	902		d.remoteTransportMode = ports.RemoteTransportUDP
845		token := roleEffectForTest(t, local.attachmentToken(ac, client))	903		token := roleEffectForTest(t, local.attachmentToken(ac, client))
			904	+	snapshots := make(chan attachmentSession, 1)
846		transitionErr := make(chan error, 1)	905		transitionErr := make(chan error, 1)
847		d.afterConnectionSessionSnapshot = func(snapshot attachmentSession) {	906		d.afterConnectionSessionSnapshot = func(snapshot attachmentSession) {
848		d.afterConnectionSessionSnapshot = nil	907		d.afterConnectionSessionSnapshot = nil
849	-	require.Same(t, local, snapshot)	908	+	snapshots <- snapshot
850		transitionErr <- d.switchToTargetForRole(token, picker.Target{Session: key.ID(), RemoteKey: &key, TabIndex: -1}, sessionHandoffGuard{}, "picker-select")	909		transitionErr <- d.switchToTargetForRole(token, picker.Target{Session: key.ID(), RemoteKey: &key, TabIndex: -1}, sessionHandoffGuard{}, "picker-select")
851		}	910		}
852			911

@@ -856,6 +915,7 @@ func TestConnectionLoopRetriesRoleResolutionAcrossHandoff(t *testing.T) {
856		close(loopDone)	915		close(loopDone)
857		}()	916		}()
858		client.recv <- proxyRecv{frame: ports.Frame{Type: ports.MsgPing, Payload: ports.MarshalPing(ports.Ping{})}}	917		client.recv <- proxyRecv{frame: ports.Frame{Type: ports.MsgPing, Payload: ports.MarshalPing(ports.Ping{})}}
			918	+	require.Same(t, local, awaitTestValue(t, snapshots, "connection role was not snapshotted"))
859		require.NoError(t, awaitTestValue(t, transitionErr, "handoff did not run during role resolution"))	919		require.NoError(t, awaitTestValue(t, transitionErr, "handoff did not run during role resolution"))
860		proxy, ok := ac.currentAttachmentSession().(*proxySession)	920		proxy, ok := ac.currentAttachmentSession().(*proxySession)
861		require.True(t, ok)	921		require.True(t, ok)