17d2014 fix(routes): guard navigation commits and welcome framing · vev

M internal/usecase/client/client.go +58 -44


@@ -344,6 +344,7 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
344	344		returnNavigationPending := false
345	345		routeNavigationPending := false
346	346		routeNavigationResumeFallback := false
	347	+	var routeNavigationSelection *routeNavigationSelection
347	348		var routeNavigationFallback *attachRoute
348	349		var routeNavigationAction *ports.RouteNavigationAction
349	350		backoff := defaultReconnectBackoff.initial

@@ -496,23 +497,24 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
496	497		linkEvents = reporter.LinkEvents()
497	498		}
498	499		result := (&attachAttempt{
499		-	runner: r,
500		-	dialer: dialer,
501		-	connection: connection,
502		-	remote: remote,
503		-	handshakeCtx: handshakeCtx,
504		-	handshakeTimedOut: timedOut,
505		-	finishHandshake: finishHandshake,
506		-	stopHandshakeTransport: stopHandshakeTransport,
507		-	request: attemptRequest,
508		-	resumeToken: resumeToken,
509		-	clientID: processClientID,
510		-	milestones: &ms,
511		-	themeState: themeState,
512		-	enterRaw: enterRaw,
513		-	reconnect: reconnect,
514		-	linkEvents: linkEvents,
515		-	rememberRemoteHost: rememberRemoteHost,
	500	+	runner: r,
	501	+	dialer: dialer,
	502	+	connection: connection,
	503	+	remote: remote,
	504	+	handshakeCtx: handshakeCtx,
	505	+	handshakeTimedOut: timedOut,
	506	+	finishHandshake: finishHandshake,
	507	+	stopHandshakeTransport: stopHandshakeTransport,
	508	+	request: attemptRequest,
	509	+	resumeToken: resumeToken,
	510	+	routeNavigationSelection: routeNavigationSelection,
	511	+	clientID: processClientID,
	512	+	milestones: &ms,
	513	+	themeState: themeState,
	514	+	enterRaw: enterRaw,
	515	+	reconnect: reconnect,
	516	+	linkEvents: linkEvents,
	517	+	rememberRemoteHost: rememberRemoteHost,
516	518		terminalInput: func() *terminalInputPump {
517	519		return input
518	520		},

@@ -541,6 +543,13 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
541	543		}
542	544		attemptRequest.ExactTarget = &identity.Target
543	545		}
	546	+	if result.err == nil && result.welcomed && routeNavigationPending {
	547	+	routeNavigationPending = false
	548	+	routeNavigationSelection = nil
	549	+	routeNavigationResumeFallback = false
	550	+	routeNavigationFallback = nil
	551	+	routeNavigationAction = nil
	552	+	}
544	553		if result.routeAction != nil {
545	554		if r.ledger == nil {
546	555		return errors.New("vev: route ledger unavailable")

@@ -572,6 +581,7 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
572	581		}
573	582		resumeToken = selection.selected.resumeToken
574	583		routeNavigationPending = true
	584	+	routeNavigationSelection = &selection
575	585		routeNavigationResumeFallback = selection.selected.resumeToken != 0
576	586		remote = syncReconnectRemote(reconnect, attemptRequest.Remote \|\| r.remote)
577	587		backoff = defaultReconnectBackoff.initial

@@ -647,12 +657,6 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
647	657		continue
648	658		}
649	659		if result.err == nil {
650		-	if result.welcomed && routeNavigationPending {
651		-	routeNavigationPending = false
652		-	routeNavigationResumeFallback = false
653		-	routeNavigationFallback = nil
654		-	routeNavigationAction = nil
655		-	}
656	660		if result.welcomed && returnNavigationPending {
657	661		returnNavigationPending = false
658	662		returnResumeFallback = false

@@ -691,6 +695,7 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
691	695		route := *routeNavigationFallback
692	696		routeNavigationFallback = nil
693	697		routeNavigationPending = false
	698	+	routeNavigationSelection = nil
694	699		routeNavigationResumeFallback = false
695	700		returnNavigationPending = true
696	701		restoreReturnRoute(route)

@@ -860,25 +865,26 @@ func sleepReconnectWithResizeEvents(ctx context.Context, clk ports.Clock, d time
860	865		}
861	866
862	867		type attachAttempt struct {
863		-	runner *Runner
864		-	dialer ports.Dialer
865		-	connection *SessionConnection
866		-	remote bool
867		-	transport ports.Transport
868		-	handshakeCtx context.Context
869		-	handshakeTimedOut <-chan struct{}
870		-	finishHandshake func()
871		-	stopHandshakeTransport func()
872		-	request AttachRequest
873		-	resumeToken uint64
874		-	clientID [16]byte
875		-	milestones *milestones
876		-	themeState *terminalThemeState
877		-	enterRaw func() error
878		-	reconnect *reconnectUI
879		-	linkEvents <-chan ports.LinkEvent
880		-	rememberRemoteHost func()
881		-	terminalInput func() *terminalInputPump
	868	+	runner *Runner
	869	+	dialer ports.Dialer
	870	+	connection *SessionConnection
	871	+	remote bool
	872	+	transport ports.Transport
	873	+	handshakeCtx context.Context
	874	+	handshakeTimedOut <-chan struct{}
	875	+	finishHandshake func()
	876	+	stopHandshakeTransport func()
	877	+	request AttachRequest
	878	+	resumeToken uint64
	879	+	routeNavigationSelection *routeNavigationSelection
	880	+	clientID [16]byte
	881	+	milestones *milestones
	882	+	themeState *terminalThemeState
	883	+	enterRaw func() error
	884	+	reconnect *reconnectUI
	885	+	linkEvents <-chan ports.LinkEvent
	886	+	rememberRemoteHost func()
	887	+	terminalInput func() *terminalInputPump
882	888		}
883	889
884	890		type attachResult struct {

@@ -1084,8 +1090,16 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
1084	1090		if request.RemoteTarget != nil && (request.RemoteTarget.LifecycleID != committedIdentity.Target.LifecycleID \|\| request.RemoteTarget.SessionName != committedIdentity.Target.SessionName) {
1085	1091		return welcomedResult(errRouteTargetChanged)
1086	1092		}
1087		-	if _, err := a.runner.ledger.commitAttach(routeCandidateForAttach(request, *committedIdentity, a.dialer, resumeToken)); err != nil {
1088		-	return welcomedResult(fmt.Errorf("vev: committing route identity: %w", err))
	1093	+	candidate := routeCandidateForAttach(request, *committedIdentity, a.dialer, resumeToken)
	1094	+	var commitErr error
	1095	+	if a.routeNavigationSelection != nil {
	1096	+	selection := *a.routeNavigationSelection
	1097	+	commitErr = a.runner.ledger.commitTransition(selection.selected.identity, selection.snapshotGeneration, selection.active, candidate)
	1098	+	} else {
	1099	+	_, commitErr = a.runner.ledger.commitAttach(candidate)
	1100	+	}
	1101	+	if commitErr != nil {
	1102	+	return welcomedResult(fmt.Errorf("vev: committing route identity: %w", commitErr))
1089	1103		}
1090	1104		payload, err := ports.MarshalRecentRouteSnapshot(a.runner.ledger.snapshot())
1091	1105		if err != nil {

 	returnNavigationPending := false
 	routeNavigationPending := false
 	routeNavigationResumeFallback := false
+	var routeNavigationSelection *routeNavigationSelection
 	var routeNavigationFallback *attachRoute
 	var routeNavigationAction *ports.RouteNavigationAction
 	backoff := defaultReconnectBackoff.initial
 			linkEvents = reporter.LinkEvents()
 		}
 		result := (&attachAttempt{
-			runner:                 r,
+			runner:                   r,
-			dialer:                 dialer,
+			dialer:                   dialer,
-			connection:             connection,
+			connection:               connection,
-			remote:                 remote,
+			remote:                   remote,
-			handshakeCtx:           handshakeCtx,
+			handshakeCtx:             handshakeCtx,
-			handshakeTimedOut:      timedOut,
+			handshakeTimedOut:        timedOut,
-			finishHandshake:        finishHandshake,
+			finishHandshake:          finishHandshake,
-			stopHandshakeTransport: stopHandshakeTransport,
+			stopHandshakeTransport:   stopHandshakeTransport,
-			request:                attemptRequest,
+			request:                  attemptRequest,
-			resumeToken:            resumeToken,
+			resumeToken:              resumeToken,
-			clientID:               processClientID,
+			routeNavigationSelection: routeNavigationSelection,
-			milestones:             &ms,
+			clientID:                 processClientID,
-			themeState:             themeState,
+			milestones:               &ms,
-			enterRaw:               enterRaw,
+			themeState:               themeState,
-			reconnect:              reconnect,
+			enterRaw:                 enterRaw,
-			linkEvents:             linkEvents,
+			reconnect:                reconnect,
-			rememberRemoteHost:     rememberRemoteHost,
+			linkEvents:               linkEvents,
+			rememberRemoteHost:       rememberRemoteHost,
 			terminalInput: func() *terminalInputPump {
 				return input
 			},
 			}
 			attemptRequest.ExactTarget = &identity.Target
 		}
+		if result.err == nil && result.welcomed && routeNavigationPending {
+			routeNavigationPending = false
+			routeNavigationSelection = nil
+			routeNavigationResumeFallback = false
+			routeNavigationFallback = nil
+			routeNavigationAction = nil
+		}
 		if result.routeAction != nil {
 			if r.ledger == nil {
 				return errors.New("vev: route ledger unavailable")
 			}
 			resumeToken = selection.selected.resumeToken
 			routeNavigationPending = true
+			routeNavigationSelection = &selection
 			routeNavigationResumeFallback = selection.selected.resumeToken != 0
 			remote = syncReconnectRemote(reconnect, attemptRequest.Remote || r.remote)
 			backoff = defaultReconnectBackoff.initial
 			continue
 		}
 		if result.err == nil {
-			if result.welcomed && routeNavigationPending {
-				routeNavigationPending = false
-				routeNavigationResumeFallback = false
-				routeNavigationFallback = nil
-				routeNavigationAction = nil
-			}
 			if result.welcomed && returnNavigationPending {
 				returnNavigationPending = false
 				returnResumeFallback = false
 			route := *routeNavigationFallback
 			routeNavigationFallback = nil
 			routeNavigationPending = false
+			routeNavigationSelection = nil
 			routeNavigationResumeFallback = false
 			returnNavigationPending = true
 			restoreReturnRoute(route)
 }
 type attachAttempt struct {
-	runner                 *Runner
+	runner                   *Runner
-	dialer                 ports.Dialer
+	dialer                   ports.Dialer
-	connection             *SessionConnection
+	connection               *SessionConnection
-	remote                 bool
+	remote                   bool
-	transport              ports.Transport
+	transport                ports.Transport
-	handshakeCtx           context.Context
+	handshakeCtx             context.Context
-	handshakeTimedOut      <-chan struct{}
+	handshakeTimedOut        <-chan struct{}
-	finishHandshake        func()
+	finishHandshake          func()
-	stopHandshakeTransport func()
+	stopHandshakeTransport   func()
-	request                AttachRequest
+	request                  AttachRequest
-	resumeToken            uint64
+	resumeToken              uint64
-	clientID               [16]byte
+	routeNavigationSelection *routeNavigationSelection
-	milestones             *milestones
+	clientID                 [16]byte
-	themeState             *terminalThemeState
+	milestones               *milestones
-	enterRaw               func() error
+	themeState               *terminalThemeState
-	reconnect              *reconnectUI
+	enterRaw                 func() error
-	linkEvents             <-chan ports.LinkEvent
+	reconnect                *reconnectUI
-	rememberRemoteHost     func()
+	linkEvents               <-chan ports.LinkEvent
-	terminalInput          func() *terminalInputPump
+	rememberRemoteHost       func()
+	terminalInput            func() *terminalInputPump
 }
 type attachResult struct {
 		if request.RemoteTarget != nil && (request.RemoteTarget.LifecycleID != committedIdentity.Target.LifecycleID || request.RemoteTarget.SessionName != committedIdentity.Target.SessionName) {
 			return welcomedResult(errRouteTargetChanged)
 		}
-		if _, err := a.runner.ledger.commitAttach(routeCandidateForAttach(request, *committedIdentity, a.dialer, resumeToken)); err != nil {
+		candidate := routeCandidateForAttach(request, *committedIdentity, a.dialer, resumeToken)
-			return welcomedResult(fmt.Errorf("vev: committing route identity: %w", err))
+		var commitErr error
+		if a.routeNavigationSelection != nil {
+			selection := *a.routeNavigationSelection
+			commitErr = a.runner.ledger.commitTransition(selection.selected.identity, selection.snapshotGeneration, selection.active, candidate)
+		} else {
+			_, commitErr = a.runner.ledger.commitAttach(candidate)
+		}
+		if commitErr != nil {
+			return welcomedResult(fmt.Errorf("vev: committing route identity: %w", commitErr))
 		}
 		payload, err := ports.MarshalRecentRouteSnapshot(a.runner.ledger.snapshot())
 		if err != nil {

M internal/usecase/daemon/daemon.go +5 -1


@@ -1172,7 +1172,11 @@ func (d *Daemon) handleHelloWithContext(handshakeCtx context.Context, timedOut <
1172	1172		return
1173	1173		}
1174	1174		welcomeDone, welcomeErr := boundedHandshakeOperationTracked(handshakeCtx, tr, func() error {
1175		-	return ac.sendExpectedTransportForAttachment(expected, frameWelcome(sess, d.resumeTokenSnapshot(ac)), welcomeTicket)
	1175	+	frame, err := frameWelcome(sess, d.resumeTokenSnapshot(ac))
	1176	+	if err != nil {
	1177	+	return err
	1178	+	}
	1179	+	return ac.sendExpectedTransportForAttachment(expected, frame, welcomeTicket)
1176	1180		})
1177	1181		if welcomeErr != nil {
1178	1182		<-welcomeDone

 		return
 	}
 	welcomeDone, welcomeErr := boundedHandshakeOperationTracked(handshakeCtx, tr, func() error {
-		return ac.sendExpectedTransportForAttachment(expected, frameWelcome(sess, d.resumeTokenSnapshot(ac)), welcomeTicket)
+		frame, err := frameWelcome(sess, d.resumeTokenSnapshot(ac))
+		if err != nil {
+			return err
+		}
+		return ac.sendExpectedTransportForAttachment(expected, frame, welcomeTicket)
 	})
 	if welcomeErr != nil {
 		<-welcomeDone

M internal/usecase/daemon/wire.go +6 -2


@@ -6,7 +6,7 @@ import (
6	6		"github.com/bnema/vev/internal/ports"
7	7		)
8	8
9		-	func frameWelcome(s *session, resumeToken uint64) ports.Frame {
	9	+	func frameWelcome(s *session, resumeToken uint64) (ports.Frame, error) {
10	10		s.mu.Lock()
11	11		var identity *ports.CommittedRouteIdentity
12	12		if s.incarnation != (domain.SessionLifecycleID{}) {

@@ -24,7 +24,11 @@ func frameWelcome(s *session, resumeToken uint64) ports.Frame {
24	24		}
25	25		s.mu.Unlock()
26	26		w.ResumeToken = resumeToken
27		-	return ports.Frame{Type: ports.MsgWelcome, Payload: ports.MarshalWelcome(w)}
	27	+	payload := ports.MarshalWelcome(w)
	28	+	if payload == nil {
	29	+	return ports.Frame{}, ports.ErrInvalidRouteWire
	30	+	}
	31	+	return ports.Frame{Type: ports.MsgWelcome, Payload: payload}, nil
28	32		}
29	33
30	34		func frameError(code uint16, text string) ports.Frame {

 	"github.com/bnema/vev/internal/ports"
 )
-func frameWelcome(s *session, resumeToken uint64) ports.Frame {
+func frameWelcome(s *session, resumeToken uint64) (ports.Frame, error) {
 	s.mu.Lock()
 	var identity *ports.CommittedRouteIdentity
 	if s.incarnation != (domain.SessionLifecycleID{}) {
 	}
 	s.mu.Unlock()
 	w.ResumeToken = resumeToken
-	return ports.Frame{Type: ports.MsgWelcome, Payload: ports.MarshalWelcome(w)}
+	payload := ports.MarshalWelcome(w)
+	if payload == nil {
+		return ports.Frame{}, ports.ErrInvalidRouteWire
+	}
+	return ports.Frame{Type: ports.MsgWelcome, Payload: payload}, nil
 }
 func frameError(code uint16, text string) ports.Frame {

M internal/usecase/daemon/wire_metadata_test.go +4 -1


@@ -23,7 +23,10 @@ func TestFrameWelcomeSnapshotsSessionMetadataUnderLock(t *testing.T) {
23	23		})
24	24
25	25		for range 1000 {
26		-	frame := frameWelcome(sess, ac.resumeToken)
	26	+	frame, frameErr := frameWelcome(sess, ac.resumeToken)
	27	+	if frameErr != nil {
	28	+	t.Fatalf("frameWelcome() error = %v", frameErr)
	29	+	}
27	30		welcome, err := ports.UnmarshalWelcome(frame.Payload)
28	31		if err != nil {
29	32		t.Fatalf("UnmarshalWelcome() error = %v", err)

 	})
 	for range 1000 {
-		frame := frameWelcome(sess, ac.resumeToken)
+		frame, frameErr := frameWelcome(sess, ac.resumeToken)
+		if frameErr != nil {
+			t.Fatalf("frameWelcome() error = %v", frameErr)
+		}
 		welcome, err := ports.UnmarshalWelcome(frame.Payload)
 		if err != nil {
 			t.Fatalf("UnmarshalWelcome() error = %v", err)