d8baccc fix(remote): harden proxy handoff routing · vev

M internal/usecase/daemon/client_frame_routing.go +39 -28


@@ -7,6 +7,27 @@ import (
7	7		"github.com/bnema/vev/internal/usecase/command"
8	8		)
9	9
	10	+	// currentConnectionRole retries the lock-free session-to-role snapshot when a
	11	+	// handoff lands between those reads. The role gate still performs the final
	12	+	// mutation admission after this function returns.
	13	+	func (d Daemon) currentConnectionRole(ac attachedClient, tr ports.Transport) (attachmentSession, attachmentRoleToken, bool) {
	14	+	for ac.currentTransportIs(tr) {
	15	+	sess := ac.currentAttachmentSession()
	16	+	if sess == nil {
	17	+	return nil, attachmentRoleToken{}, false
	18	+	}
	19	+	if d.afterConnectionSessionSnapshot != nil {
	20	+	d.afterConnectionSessionSnapshot(sess)
	21	+	}
	22	+	token := attachmentToken(sess, ac, tr)
	23	+	if token.role == attachmentDetached \|\| ac.currentAttachmentSession() != sess {
	24	+	continue
	25	+	}
	26	+	return sess, token, true
	27	+	}
	28	+	return nil, attachmentRoleToken{}, false
	29	+	}
	30	+
10	31		// runConnLoop is the per-connection input router: it pumps client messages
11	32		// until detach, EOF, or a transport error. Role is resolved after every Recv so
12	33		// a transport displaced while blocked in Recv immediately adopts its restricted

@@ -17,40 +38,32 @@ func (d Daemon) runConnLoop(ac attachedClient) {
17	38		return
18	39		}
19	40		for {
20		-	if !ac.currentTransportIs(tr) {
21		-	return
22		-	}
23		-	sess := ac.currentAttachmentSession()
24		-	if sess == nil {
	41	+	if !ac.currentTransportIs(tr) \|\| ac.currentAttachmentSession() == nil {
25	42		return
26	43		}
27	44		f, err := tr.Recv()
28	45		if err != nil {
29		-	if !ac.currentTransportIs(tr) {
30		-	return
31		-	}
32		-	sess = ac.currentAttachmentSession()
33		-	if sess == nil {
34		-	return
35		-	}
36		-	token := attachmentToken(sess, ac, tr)
37		-	if token.role == attachmentSnatched {
38		-	d.parkOrDropSnatchedAttachment(token)
39		-	} else if proxy, ok := sess.(*proxySession); ok {
40		-	d.detachProxyOnSendError(proxy, ac, tr)
41		-	} else if local, ok := localSession(sess); ok {
42		-	d.clientGone(local, ac, tr, false)
	46	+	for {
	47	+	sess, token, ok := d.currentConnectionRole(ac, tr)
	48	+	if !ok {
	49	+	return
	50	+	}
	51	+	if token.role == attachmentSnatched {
	52	+	d.parkOrDropSnatchedAttachment(token)
	53	+	} else if proxy, ok := sess.(*proxySession); ok {
	54	+	d.detachProxyOnSendError(proxy, ac, tr)
	55	+	} else if local, ok := localSession(sess); ok {
	56	+	d.clientGone(local, ac, tr, false)
	57	+	}
	58	+	if !ac.currentTransportIs(tr) \|\| ac.currentAttachmentSession() == nil {
	59	+	return
	60	+	}
43	61		}
44		-	return
45		-	}
46		-	if !ac.currentTransportIs(tr) {
47		-	return
48	62		}
49		-	sess = ac.currentAttachmentSession()
50		-	if sess == nil {
	63	+	_, token, ok := d.currentConnectionRole(ac, tr)
	64	+	if !ok {
51	65		return
52	66		}
53		-	token := attachmentToken(sess, ac, tr)
54	67		switch token.role {
55	68		case attachmentActive:
56	69		if d.handleActiveClientFrame(token, f) {

@@ -60,8 +73,6 @@ func (d Daemon) runConnLoop(ac attachedClient) {
60	73		if d.handleSnatchedClientFrame(token, f) {
61	74		return
62	75		}
63		-	default:
64		-	return
65	76		}
66	77		}
67	78		}

 	"github.com/bnema/vev/internal/usecase/command"
 )
+// 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) {
+		sess := ac.currentAttachmentSession()
+		if sess == nil {
+			return nil, attachmentRoleToken{}, false
+		}
+		if d.afterConnectionSessionSnapshot != nil {
+			d.afterConnectionSessionSnapshot(sess)
+		}
+		token := attachmentToken(sess, ac, tr)
+		if token.role == attachmentDetached || ac.currentAttachmentSession() != sess {
+			continue
+		}
+		return sess, token, true
+	}
+	return nil, attachmentRoleToken{}, false
+}
 // runConnLoop is the per-connection input router: it pumps client messages
 // until detach, EOF, or a transport error. Role is resolved after every Recv so
 // a transport displaced while blocked in Recv immediately adopts its restricted
 		return
 	}
 	for {
-		if !ac.currentTransportIs(tr) {
+		if !ac.currentTransportIs(tr) || ac.currentAttachmentSession() == nil {
-			return
-		}
-		sess := ac.currentAttachmentSession()
-		if sess == nil {
 			return
 		}
 		f, err := tr.Recv()
 		if err != nil {
-			if !ac.currentTransportIs(tr) {
+			for {
-				return
+				sess, token, ok := d.currentConnectionRole(ac, tr)
-			}
+				if !ok {
-			sess = ac.currentAttachmentSession()
+					return
-			if sess == nil {
+				}
-				return
+				if token.role == attachmentSnatched {
-			}
+					d.parkOrDropSnatchedAttachment(token)
-			token := attachmentToken(sess, ac, tr)
+				} else if proxy, ok := sess.(*proxySession); ok {
-			if token.role == attachmentSnatched {
+					d.detachProxyOnSendError(proxy, ac, tr)
-				d.parkOrDropSnatchedAttachment(token)
+				} else if local, ok := localSession(sess); ok {
-			} else if proxy, ok := sess.(*proxySession); ok {
+					d.clientGone(local, ac, tr, false)
-				d.detachProxyOnSendError(proxy, ac, tr)
+				}
-			} else if local, ok := localSession(sess); ok {
+				if !ac.currentTransportIs(tr) || ac.currentAttachmentSession() == nil {
-				d.clientGone(local, ac, tr, false)
+					return
+				}
 			}
-			return
-		}
-		if !ac.currentTransportIs(tr) {
-			return
 		}
-		sess = ac.currentAttachmentSession()
+		_, token, ok := d.currentConnectionRole(ac, tr)
-		if sess == nil {
+		if !ok {
 			return
 		}
-		token := attachmentToken(sess, ac, tr)
 		switch token.role {
 		case attachmentActive:
 			if d.handleActiveClientFrame(token, f) {
 			if d.handleSnatchedClientFrame(token, f) {
 				return
 			}
-		default:
-			return
 		}
 	}
 }

M internal/usecase/daemon/daemon.go +3 -0


@@ -153,6 +153,9 @@ type Daemon struct {
153	153		// afterSnatchedEscapeAttempt reports whether that callback acquired its exact
154	154		// role capability after revalidation.
155	155		afterSnatchedEscapeAttempt func(bool)
	156	+	// afterConnectionSessionSnapshot is a deterministic test seam after the
	157	+	// connection loop reads its current session and before it captures that role.
	158	+	afterConnectionSessionSnapshot func(attachmentSession)
156	159		// afterActiveFrameDispatch is a deterministic test seam after the connection
157	160		// loop's first role check and before a decoded active frame takes effect.
158	161		afterActiveFrameDispatch func(attachmentRoleToken)


@@ -153,6 +153,9 @@ type Daemon struct {
153	// afterSnatchedEscapeAttempt reports whether that callback acquired its exact	153		// afterSnatchedEscapeAttempt reports whether that callback acquired its exact
154	// role capability after revalidation.	154		// role capability after revalidation.
155	afterSnatchedEscapeAttempt func(bool)	155		afterSnatchedEscapeAttempt func(bool)
		156	+	// afterConnectionSessionSnapshot is a deterministic test seam after the
		157	+	// connection loop reads its current session and before it captures that role.
		158	+	afterConnectionSessionSnapshot func(attachmentSession)
156	// afterActiveFrameDispatch is a deterministic test seam after the connection	159		// afterActiveFrameDispatch is a deterministic test seam after the connection
157	// loop's first role check and before a decoded active frame takes effect.	160		// loop's first role check and before a decoded active frame takes effect.
158	afterActiveFrameDispatch func(attachmentRoleToken)	161		afterActiveFrameDispatch func(attachmentRoleToken)

M internal/usecase/daemon/picker.go +6 -0


@@ -587,6 +587,12 @@ func (d *Daemon) switchActiveTargetForRoleGuarded(token attachmentRoleToken, tar
587	587		}
588	588		d.mu.Lock()
589	589		targetSess, ok := localSession(d.sessions[target.Session])
	590	+	if ok {
	591	+	targetSess.mu.Lock()
	592	+	ok = targetMatchesLifecycle(target, targetSess.name, targetSess.createdAt) &&
	593	+	(target.Incarnation == (domain.IncarnationID{}) \|\| target.Incarnation == targetSess.incarnation)
	594	+	targetSess.mu.Unlock()
	595	+	}
590	596		d.mu.Unlock()
591	597		if !ok \|\| targetSess == token.sess {
592	598		return nil


@@ -587,6 +587,12 @@ func (d *Daemon) switchActiveTargetForRoleGuarded(token attachmentRoleToken, tar
587	}	587		}
588	d.mu.Lock()	588		d.mu.Lock()
589	targetSess, ok := localSession(d.sessions[target.Session])	589		targetSess, ok := localSession(d.sessions[target.Session])
		590	+	if ok {
		591	+	targetSess.mu.Lock()
		592	+	ok = targetMatchesLifecycle(target, targetSess.name, targetSess.createdAt) &&
		593	+	(target.Incarnation == (domain.IncarnationID{}) \|\| target.Incarnation == targetSess.incarnation)
		594	+	targetSess.mu.Unlock()
		595	+	}
590	d.mu.Unlock()	596		d.mu.Unlock()
591	if !ok \|\| targetSess == token.sess {	597		if !ok \|\| targetSess == token.sess {
592	return nil	598		return nil

M internal/usecase/daemon/proxy_input_test.go +62 -0


@@ -703,6 +703,32 @@ func TestRemotePickerSwitchesFromProxyBackToLocalSession(t *testing.T) {
703	703		require.NoError(t, d.switchToTargetForRole(handler.roleToken, target, sessionHandoffGuard{closePicker: true}, "picker-select"))
704	704		require.Same(t, local, ac.currentSession())
705	705		require.False(t, ac.overlays.pickerActive(), "successful handoff must close the picker")
	706	+
	707	+	d.remoteDialerFactory = portsmocks.NewMockRemoteDialerFactory(t)
	708	+	localToken := roleEffectForTest(t, local.attachmentToken(ac, ac.transport()))
	709	+	require.NoError(t, d.backSessionForRole(localToken))
	710	+	require.Same(t, proxy, ac.currentAttachmentSession(), "back-session must return to the previous warm proxy")
	711	+	}
	712	+
	713	+	func TestRemotePickerRejectsReplacedLocalLifecycle(t *testing.T) {
	714	+	d, proxy, ac, _, handler := newProxyInputHarness(t)
	715	+	tb := newTab(newQuietPTY(), domain.Size{Cols: 80, Rows: 22})
	716	+	local := &session{sessionCore: sessionCore{id: "local", name: "local", createdAt: 1}, tabs: []*tab{tb}}
	717	+	publishTiledPaneOwners(local, tb)
	718	+	d.mu.Lock()
	719	+	require.True(t, d.registerSessionLocked(local))
	720	+	d.mu.Unlock()
	721	+
	722	+	model := d.newPickerModel(proxy, pickerNavigate, moveSourceLocator{}, picker.SourceFilter{Session: local.id, TabID: domain.TabStableID(tb.stableID)})
	723	+	target, selectable := model.Selected()
	724	+	require.True(t, selectable)
	725	+	local.mu.Lock()
	726	+	local.name = "replacement"
	727	+	local.createdAt++
	728	+	local.mu.Unlock()
	729	+
	730	+	require.NoError(t, d.switchToTargetForRole(handler.roleToken, target, sessionHandoffGuard{}, "picker-select"))
	731	+	require.Same(t, proxy, ac.currentAttachmentSession(), "stale picker lifecycle must not redirect the attachment")
706	732		}
707	733
708	734		func TestRemotePickerCanReselectWarmProxyAfterReturningLocal(t *testing.T) {

@@ -736,6 +762,42 @@ func TestRemotePickerCanReselectWarmProxyAfterReturningLocal(t *testing.T) {
736	762		require.False(t, ac.overlays.pickerActive())
737	763		}
738	764
	765	+	func TestConnectionLoopRetriesRoleResolutionAcrossHandoff(t *testing.T) {
	766	+	d, local, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
	767	+	client := newProxyTestTransport()
	768	+	ac.replaceTransport(client)
	769	+	d.attachCoordinator(local, nil, ac, true)
	770	+
	771	+	key := domain.RemoteSessionKey{Host: "arch", Name: "work"}
	772	+	remote := newProxyTestTransport()
	773	+	remote.recv <- proxyRecv{frame: proxyWelcome(key.Name, 1, ports.CapabilityProxied)}
	774	+	remote.recv <- proxyRecv{frame: proxyMeta(key.Name)}
	775	+	d.remoteDialerFactory = newProxyConstructionFactory(remote)
	776	+	d.remoteTransportMode = ports.RemoteTransportUDP
	777	+	token := roleEffectForTest(t, local.attachmentToken(ac, client))
	778	+	transitionErr := make(chan error, 1)
	779	+	d.afterConnectionSessionSnapshot = func(snapshot attachmentSession) {
	780	+	d.afterConnectionSessionSnapshot = nil
	781	+	require.Same(t, local, snapshot)
	782	+	transitionErr <- d.switchToTargetForRole(token, picker.Target{Session: key.ID(), RemoteKey: &key, TabIndex: -1}, sessionHandoffGuard{}, "picker-select")
	783	+	}
	784	+
	785	+	loopDone := make(chan struct{})
	786	+	go func() {
	787	+	d.runConnLoop(ac)
	788	+	close(loopDone)
	789	+	}()
	790	+	client.recv <- proxyRecv{frame: ports.Frame{Type: ports.MsgPing, Payload: ports.MarshalPing(ports.Ping{})}}
	791	+	require.NoError(t, awaitTestValue(t, transitionErr, "handoff did not run during role resolution"))
	792	+	proxy, ok := ac.currentAttachmentSession().(*proxySession)
	793	+	require.True(t, ok)
	794	+	t.Cleanup(func() { stopProxy(t, proxy) })
	795	+	_ = awaitFrame(t, client.sent, ports.MsgPong)
	796	+
	797	+	client.recv <- proxyRecv{frame: ports.Frame{Type: ports.MsgDetach, Payload: ports.MarshalDetach(ports.Detach{})}}
	798	+	awaitTestCompletion(t, loopDone, "connection loop did not continue after retrying role resolution")
	799	+	}
	800	+
739	801		func TestConnectionLoopCleansCurrentProxyAfterHandoffReceiveError(t *testing.T) {
740	802		d, local, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
741	803		client := newProxyTestTransport()


@@ -703,6 +703,32 @@ func TestRemotePickerSwitchesFromProxyBackToLocalSession(t *testing.T) {
703	require.NoError(t, d.switchToTargetForRole(handler.roleToken, target, sessionHandoffGuard{closePicker: true}, "picker-select"))	703		require.NoError(t, d.switchToTargetForRole(handler.roleToken, target, sessionHandoffGuard{closePicker: true}, "picker-select"))
704	require.Same(t, local, ac.currentSession())	704		require.Same(t, local, ac.currentSession())
705	require.False(t, ac.overlays.pickerActive(), "successful handoff must close the picker")	705		require.False(t, ac.overlays.pickerActive(), "successful handoff must close the picker")
		706	+
		707	+	d.remoteDialerFactory = portsmocks.NewMockRemoteDialerFactory(t)
		708	+	localToken := roleEffectForTest(t, local.attachmentToken(ac, ac.transport()))
		709	+	require.NoError(t, d.backSessionForRole(localToken))
		710	+	require.Same(t, proxy, ac.currentAttachmentSession(), "back-session must return to the previous warm proxy")
		711	+	}
		712	+
		713	+	func TestRemotePickerRejectsReplacedLocalLifecycle(t *testing.T) {
		714	+	d, proxy, ac, _, handler := newProxyInputHarness(t)
		715	+	tb := newTab(newQuietPTY(), domain.Size{Cols: 80, Rows: 22})
		716	+	local := &session{sessionCore: sessionCore{id: "local", name: "local", createdAt: 1}, tabs: []*tab{tb}}
		717	+	publishTiledPaneOwners(local, tb)
		718	+	d.mu.Lock()
		719	+	require.True(t, d.registerSessionLocked(local))
		720	+	d.mu.Unlock()
		721	+
		722	+	model := d.newPickerModel(proxy, pickerNavigate, moveSourceLocator{}, picker.SourceFilter{Session: local.id, TabID: domain.TabStableID(tb.stableID)})
		723	+	target, selectable := model.Selected()
		724	+	require.True(t, selectable)
		725	+	local.mu.Lock()
		726	+	local.name = "replacement"
		727	+	local.createdAt++
		728	+	local.mu.Unlock()
		729	+
		730	+	require.NoError(t, d.switchToTargetForRole(handler.roleToken, target, sessionHandoffGuard{}, "picker-select"))
		731	+	require.Same(t, proxy, ac.currentAttachmentSession(), "stale picker lifecycle must not redirect the attachment")
706	}	732		}
707		733
708	func TestRemotePickerCanReselectWarmProxyAfterReturningLocal(t *testing.T) {	734		func TestRemotePickerCanReselectWarmProxyAfterReturningLocal(t *testing.T) {

@@ -736,6 +762,42 @@ func TestRemotePickerCanReselectWarmProxyAfterReturningLocal(t *testing.T) {
736	require.False(t, ac.overlays.pickerActive())	762		require.False(t, ac.overlays.pickerActive())
737	}	763		}
738		764
		765	+	func TestConnectionLoopRetriesRoleResolutionAcrossHandoff(t *testing.T) {
		766	+	d, local, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
		767	+	client := newProxyTestTransport()
		768	+	ac.replaceTransport(client)
		769	+	d.attachCoordinator(local, nil, ac, true)
		770	+
		771	+	key := domain.RemoteSessionKey{Host: "arch", Name: "work"}
		772	+	remote := newProxyTestTransport()
		773	+	remote.recv <- proxyRecv{frame: proxyWelcome(key.Name, 1, ports.CapabilityProxied)}
		774	+	remote.recv <- proxyRecv{frame: proxyMeta(key.Name)}
		775	+	d.remoteDialerFactory = newProxyConstructionFactory(remote)
		776	+	d.remoteTransportMode = ports.RemoteTransportUDP
		777	+	token := roleEffectForTest(t, local.attachmentToken(ac, client))
		778	+	transitionErr := make(chan error, 1)
		779	+	d.afterConnectionSessionSnapshot = func(snapshot attachmentSession) {
		780	+	d.afterConnectionSessionSnapshot = nil
		781	+	require.Same(t, local, snapshot)
		782	+	transitionErr <- d.switchToTargetForRole(token, picker.Target{Session: key.ID(), RemoteKey: &key, TabIndex: -1}, sessionHandoffGuard{}, "picker-select")
		783	+	}
		784	+
		785	+	loopDone := make(chan struct{})
		786	+	go func() {
		787	+	d.runConnLoop(ac)
		788	+	close(loopDone)
		789	+	}()
		790	+	client.recv <- proxyRecv{frame: ports.Frame{Type: ports.MsgPing, Payload: ports.MarshalPing(ports.Ping{})}}
		791	+	require.NoError(t, awaitTestValue(t, transitionErr, "handoff did not run during role resolution"))
		792	+	proxy, ok := ac.currentAttachmentSession().(*proxySession)
		793	+	require.True(t, ok)
		794	+	t.Cleanup(func() { stopProxy(t, proxy) })
		795	+	_ = awaitFrame(t, client.sent, ports.MsgPong)
		796	+
		797	+	client.recv <- proxyRecv{frame: ports.Frame{Type: ports.MsgDetach, Payload: ports.MarshalDetach(ports.Detach{})}}
		798	+	awaitTestCompletion(t, loopDone, "connection loop did not continue after retrying role resolution")
		799	+	}
		800	+
739	func TestConnectionLoopCleansCurrentProxyAfterHandoffReceiveError(t *testing.T) {	801		func TestConnectionLoopCleansCurrentProxyAfterHandoffReceiveError(t *testing.T) {
740	d, local, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())	802		d, local, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
741	client := newProxyTestTransport()	803		client := newProxyTestTransport()

M internal/usecase/daemon/session_back.go +6 -1


@@ -13,7 +13,12 @@ func (d *Daemon) backSessionForRole(token attachmentRoleToken) error {
13	13		token.ac.clearPreviousSessionIf(target)
14	14		return nil
15	15		}
16		-	return d.switchToTargetForRole(token, picker.Target{Session: target.core().id, TabIndex: -1}, sessionHandoffGuard{}, "back-session")
	16	+	pickerTarget := picker.Target{Session: target.core().id, TabIndex: -1}
	17	+	if proxy, ok := target.(*proxySession); ok {
	18	+	key := proxy.key
	19	+	pickerTarget.RemoteKey = &key
	20	+	}
	21	+	return d.switchToTargetForRole(token, pickerTarget, sessionHandoffGuard{}, "back-session")
17	22		}
18	23
19	24		func (d Daemon) backSession(current session, ac *attachedClient) {

 		token.ac.clearPreviousSessionIf(target)
 		return nil
 	}
-	return d.switchToTargetForRole(token, picker.Target{Session: target.core().id, TabIndex: -1}, sessionHandoffGuard{}, "back-session")
+	pickerTarget := picker.Target{Session: target.core().id, TabIndex: -1}
+	if proxy, ok := target.(*proxySession); ok {
+		key := proxy.key
+		pickerTarget.RemoteKey = &key
+	}
+	return d.switchToTargetForRole(token, pickerTarget, sessionHandoffGuard{}, "back-session")
 }
 func (d *Daemon) backSession(current *session, ac *attachedClient) {