detached eca6382

fix(routes): retire deleted sessions from recent history (#264)

* fix(routes): retire deleted sessions from recent history

* fix(routes): fence retirement by observation and connection
commit
eca638226096b7becf0dd448f39181125569906d
author
bnema · · 2026-09-09 06:09 UTC
committer
GitHub · · 2026-09-09 06:09 UTC
parent
680dbb1
signature
GPG UnverifiedB5690EEEBB952194 The key is unknown
35 files · +628 -49
M docs/architecture.md +1 -1
@@ -56,4 +56,4 @@ Use cases exchange `protocol.ClientMessage` and `protocol.ServerMessage` values.
5656 - Add remote catalogue schema fields and validation to `internal/protocol/catalogue`.
5757 - Add message IDs, binary layouts, strict decoding, compression, or raw carriage contracts to `internal/protocol/wire`.
5858 - Implement I/O, queues, workers, environment integration, or technology selection in an adapter or `internal/app`.
59-- Bump `internal/protocol.Version` for negotiated wire layout changes (currently `42`, including navigation inventory, UI view context and processing-fence messages).
59+- Bump `internal/protocol.Version` for negotiated wire layout changes (currently `43`, including source-scoped route subscriptions and lifecycle retirement).
@@ -56,4 +56,4 @@ Use cases exchange `protocol.ClientMessage` and `protocol.ServerMessage` values.
56 - Add remote catalogue schema fields and validation to `internal/protocol/catalogue`.56 - Add remote catalogue schema fields and validation to `internal/protocol/catalogue`.
57 - Add message IDs, binary layouts, strict decoding, compression, or raw carriage contracts to `internal/protocol/wire`.57 - Add message IDs, binary layouts, strict decoding, compression, or raw carriage contracts to `internal/protocol/wire`.
58 - Implement I/O, queues, workers, environment integration, or technology selection in an adapter or `internal/app`.58 - Implement I/O, queues, workers, environment integration, or technology selection in an adapter or `internal/app`.
59-- Bump `internal/protocol.Version` for negotiated wire layout changes (currently `42`, including navigation inventory, UI view context and processing-fence messages).59+- Bump `internal/protocol.Version` for negotiated wire layout changes (currently `43`, including source-scoped route subscriptions and lifecycle retirement).
M internal/adapters/sessionwire/client.go +2 -0
@@ -107,6 +107,8 @@ func decodeServer(frame wire.Frame) (protocol.ServerMessage, error) {
107107 return wire.UnmarshalRouteNavigationFailure(frame.Payload)
108108 case wire.MsgRoutePosition:
109109 return wire.UnmarshalRoutePosition(frame.Payload)
110+ case wire.MsgRouteRetired:
111+ return wire.UnmarshalRouteRetired(frame.Payload)
110112 case wire.MsgSamePeerSwitchFailure:
111113 return wire.UnmarshalSamePeerSwitchFailure(frame.Payload)
112114 case wire.MsgParkedRouteResponse:
@@ -107,6 +107,8 @@ func decodeServer(frame wire.Frame) (protocol.ServerMessage, error) {
107 return wire.UnmarshalRouteNavigationFailure(frame.Payload)107 return wire.UnmarshalRouteNavigationFailure(frame.Payload)
108 case wire.MsgRoutePosition:108 case wire.MsgRoutePosition:
109 return wire.UnmarshalRoutePosition(frame.Payload)109 return wire.UnmarshalRoutePosition(frame.Payload)
110+ case wire.MsgRouteRetired:
111+ return wire.UnmarshalRouteRetired(frame.Payload)
110 case wire.MsgSamePeerSwitchFailure:112 case wire.MsgSamePeerSwitchFailure:
111 return wire.UnmarshalSamePeerSwitchFailure(frame.Payload)113 return wire.UnmarshalSamePeerSwitchFailure(frame.Payload)
112 case wire.MsgParkedRouteResponse:114 case wire.MsgParkedRouteResponse:
A internal/adapters/sessionwire/route_retired_test.go +24 -0
@@ -0,0 +1,24 @@
1+package sessionwire
2+
3+import (
4+ "testing"
5+
6+ "github.com/bnema/vev/internal/domain"
7+ "github.com/bnema/vev/internal/protocol"
8+ "github.com/bnema/vev/internal/protocol/wire"
9+ "github.com/stretchr/testify/require"
10+)
11+
12+func TestRouteRetiredServerMessage(t *testing.T) {
13+ message := protocol.RouteRetired{Ref: protocol.RouteRef{Key: 1, Generation: 2}, Target: protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{3}, SessionName: "work"}}
14+ for _, value := range []protocol.ServerMessage{message, &message} {
15+ frame, err := encodeServer(value)
16+ require.NoError(t, err)
17+ require.Equal(t, wire.MsgRouteRetired, frame.Type)
18+ decoded, err := decodeServer(frame)
19+ require.NoError(t, err)
20+ require.Equal(t, message, decoded)
21+ _, err = decodeClient(frame)
22+ require.ErrorIs(t, err, ErrWrongDirection)
23+ }
24+}
@@ -0,0 +1,24 @@
1+package sessionwire
2+
3+import (
4+ "testing"
5+
6+ "github.com/bnema/vev/internal/domain"
7+ "github.com/bnema/vev/internal/protocol"
8+ "github.com/bnema/vev/internal/protocol/wire"
9+ "github.com/stretchr/testify/require"
10+)
11+
12+func TestRouteRetiredServerMessage(t *testing.T) {
13+ message := protocol.RouteRetired{Ref: protocol.RouteRef{Key: 1, Generation: 2}, Target: protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{3}, SessionName: "work"}}
14+ for _, value := range []protocol.ServerMessage{message, &message} {
15+ frame, err := encodeServer(value)
16+ require.NoError(t, err)
17+ require.Equal(t, wire.MsgRouteRetired, frame.Type)
18+ decoded, err := decodeServer(frame)
19+ require.NoError(t, err)
20+ require.Equal(t, message, decoded)
21+ _, err = decodeClient(frame)
22+ require.ErrorIs(t, err, ErrWrongDirection)
23+ }
24+}
M internal/adapters/sessionwire/server.go +8 -1
@@ -252,7 +252,7 @@ func decodeClient(frame wire.Frame) (protocol.ClientMessage, error) {
252252 wire.MsgCommittedRouteIdentity, wire.MsgNavigateRecentRoute, wire.MsgRouteCreateSession,
253253 wire.MsgRoutePosition, wire.MsgSamePeerSwitchFailure, wire.MsgParkedRouteResponse,
254254 wire.MsgUIReceipt, wire.MsgUIViewUpdate, wire.MsgNavigationInventoryResponse,
255- wire.MsgNavigationInventoryDemand, wire.MsgNavigationInventorySelection:
255+ wire.MsgNavigationInventoryDemand, wire.MsgNavigationInventorySelection, wire.MsgRouteRetired:
256256 return nil, ErrWrongDirection
257257 default:
258258 return nil, ErrUnknownMessageType
@@ -319,6 +319,9 @@ func encodeServer(message protocol.ServerMessage) (wire.Frame, error) {
319319 case protocol.RouteNavigationFailure:
320320 payload, err := wire.MarshalRouteNavigationFailure(m)
321321 return wire.Frame{Type: wire.MsgRouteNavigationFailure, Payload: payload}, err
322+ case protocol.RouteRetired:
323+ payload, err := wire.MarshalRouteRetired(m)
324+ return wire.Frame{Type: wire.MsgRouteRetired, Payload: payload}, err
322325 case protocol.RoutePosition:
323326 payload, err := wire.MarshalRoutePosition(m)
324327 return wire.Frame{Type: wire.MsgRoutePosition, Payload: payload}, err
@@ -413,6 +416,10 @@ func encodeServer(message protocol.ServerMessage) (wire.Frame, error) {
413416 if m != nil {
414417 return encodeServer(*m)
415418 }
419+ case *protocol.RouteRetired:
420+ if m != nil {
421+ return encodeServer(*m)
422+ }
416423 case *protocol.RoutePosition:
417424 if m != nil {
418425 return encodeServer(*m)
@@ -252,7 +252,7 @@ func decodeClient(frame wire.Frame) (protocol.ClientMessage, error) {
252 wire.MsgCommittedRouteIdentity, wire.MsgNavigateRecentRoute, wire.MsgRouteCreateSession,252 wire.MsgCommittedRouteIdentity, wire.MsgNavigateRecentRoute, wire.MsgRouteCreateSession,
253 wire.MsgRoutePosition, wire.MsgSamePeerSwitchFailure, wire.MsgParkedRouteResponse,253 wire.MsgRoutePosition, wire.MsgSamePeerSwitchFailure, wire.MsgParkedRouteResponse,
254 wire.MsgUIReceipt, wire.MsgUIViewUpdate, wire.MsgNavigationInventoryResponse,254 wire.MsgUIReceipt, wire.MsgUIViewUpdate, wire.MsgNavigationInventoryResponse,
255- wire.MsgNavigationInventoryDemand, wire.MsgNavigationInventorySelection:255+ wire.MsgNavigationInventoryDemand, wire.MsgNavigationInventorySelection, wire.MsgRouteRetired:
256 return nil, ErrWrongDirection256 return nil, ErrWrongDirection
257 default:257 default:
258 return nil, ErrUnknownMessageType258 return nil, ErrUnknownMessageType
@@ -319,6 +319,9 @@ func encodeServer(message protocol.ServerMessage) (wire.Frame, error) {
319 case protocol.RouteNavigationFailure:319 case protocol.RouteNavigationFailure:
320 payload, err := wire.MarshalRouteNavigationFailure(m)320 payload, err := wire.MarshalRouteNavigationFailure(m)
321 return wire.Frame{Type: wire.MsgRouteNavigationFailure, Payload: payload}, err321 return wire.Frame{Type: wire.MsgRouteNavigationFailure, Payload: payload}, err
322+ case protocol.RouteRetired:
323+ payload, err := wire.MarshalRouteRetired(m)
324+ return wire.Frame{Type: wire.MsgRouteRetired, Payload: payload}, err
322 case protocol.RoutePosition:325 case protocol.RoutePosition:
323 payload, err := wire.MarshalRoutePosition(m)326 payload, err := wire.MarshalRoutePosition(m)
324 return wire.Frame{Type: wire.MsgRoutePosition, Payload: payload}, err327 return wire.Frame{Type: wire.MsgRoutePosition, Payload: payload}, err
@@ -413,6 +416,10 @@ func encodeServer(message protocol.ServerMessage) (wire.Frame, error) {
413 if m != nil {416 if m != nil {
414 return encodeServer(*m)417 return encodeServer(*m)
415 }418 }
419+ case *protocol.RouteRetired:
420+ if m != nil {
421+ return encodeServer(*m)
422+ }
416 case *protocol.RoutePosition:423 case *protocol.RoutePosition:
417 if m != nil {424 if m != nil {
418 return encodeServer(*m)425 return encodeServer(*m)
M internal/protocol/messages.go +1 -0
@@ -46,6 +46,7 @@ func (RouteNavigationAction) serverMessage() {}
4646 func (RouteCreateSessionAction) serverMessage() {}
4747 func (RouteNavigationFailure) serverMessage() {}
4848 func (RoutePosition) serverMessage() {}
49+func (RouteRetired) serverMessage() {}
4950 func (SamePeerSwitchFailure) serverMessage() {}
5051 func (ParkedRouteResponse) serverMessage() {}
5152 func (UIReceipt) serverMessage() {}
@@ -46,6 +46,7 @@ func (RouteNavigationAction) serverMessage() {}
46 func (RouteCreateSessionAction) serverMessage() {}46 func (RouteCreateSessionAction) serverMessage() {}
47 func (RouteNavigationFailure) serverMessage() {}47 func (RouteNavigationFailure) serverMessage() {}
48 func (RoutePosition) serverMessage() {}48 func (RoutePosition) serverMessage() {}
49+func (RouteRetired) serverMessage() {}
49 func (SamePeerSwitchFailure) serverMessage() {}50 func (SamePeerSwitchFailure) serverMessage() {}
50 func (ParkedRouteResponse) serverMessage() {}51 func (ParkedRouteResponse) serverMessage() {}
51 func (UIReceipt) serverMessage() {}52 func (UIReceipt) serverMessage() {}
M internal/protocol/routes.go +21 -3
@@ -1,6 +1,8 @@
11 package protocol
22
33 import (
4+ "crypto/sha256"
5+ "encoding/hex"
46 "errors"
57 "fmt"
68 "unicode"
@@ -172,15 +174,31 @@ type RouteRef struct {
172174 }
173175
174176 // RouteAttentionTarget binds one displayed route reference to an exact
175-// lifecycle on the daemon currently serving the client. It carries no dialer,
176-// credential, or endpoint.
177+// lifecycle on the serving daemon or a monitored inventory source. It carries
178+// no dialer, credential, or endpoint.
177179 type RouteAttentionTarget struct {
178180 Ref RouteRef
179181 Target ExactSessionTarget
182+ // SourceKey is empty for the serving daemon, or an opaque monitored
183+ // inventory source. It never carries an endpoint or credential.
184+ SourceKey string
185+}
186+
187+// RouteRetired invalidates one subscribed exact lifecycle, not its name.
188+// The reference fences delayed notifications against replacement routes.
189+type RouteRetired struct {
190+ Ref RouteRef
191+ Target ExactSessionTarget
192+}
193+
194+// RemoteInventorySourceKey identifies a configured endpoint without disclosing it.
195+func RemoteInventorySourceKey(endpoint string) string {
196+ sum := sha256.Sum256([]byte("vev-inventory-remote\x00" + endpoint))
197+ return NavigationInventoryRemoteSourcePrefix + hex.EncodeToString(sum[:12])
180198 }
181199
182200 // RouteAttentionSubscription is the bounded, client-owned mapping a daemon
183-// uses solely to resolve live attention while rendering its route snapshot.
201+// uses to resolve live attention and retire authoritatively absent lifecycles.
184202 type RouteAttentionSubscription struct {
185203 Targets []RouteAttentionTarget
186204 }
@@ -1,6 +1,8 @@
1 package protocol1 package protocol
2 2
3 import (3 import (
4+ "crypto/sha256"
5+ "encoding/hex"
4 "errors"6 "errors"
5 "fmt"7 "fmt"
6 "unicode"8 "unicode"
@@ -172,15 +174,31 @@ type RouteRef struct {
172 }174 }
173 175
174 // RouteAttentionTarget binds one displayed route reference to an exact176 // RouteAttentionTarget binds one displayed route reference to an exact
175-// lifecycle on the daemon currently serving the client. It carries no dialer,177+// lifecycle on the serving daemon or a monitored inventory source. It carries
176-// credential, or endpoint.178+// no dialer, credential, or endpoint.
177 type RouteAttentionTarget struct {179 type RouteAttentionTarget struct {
178 Ref RouteRef180 Ref RouteRef
179 Target ExactSessionTarget181 Target ExactSessionTarget
182+ // SourceKey is empty for the serving daemon, or an opaque monitored
183+ // inventory source. It never carries an endpoint or credential.
184+ SourceKey string
185+}
186+
187+// RouteRetired invalidates one subscribed exact lifecycle, not its name.
188+// The reference fences delayed notifications against replacement routes.
189+type RouteRetired struct {
190+ Ref RouteRef
191+ Target ExactSessionTarget
192+}
193+
194+// RemoteInventorySourceKey identifies a configured endpoint without disclosing it.
195+func RemoteInventorySourceKey(endpoint string) string {
196+ sum := sha256.Sum256([]byte("vev-inventory-remote\x00" + endpoint))
197+ return NavigationInventoryRemoteSourcePrefix + hex.EncodeToString(sum[:12])
180 }198 }
181 199
182 // RouteAttentionSubscription is the bounded, client-owned mapping a daemon200 // RouteAttentionSubscription is the bounded, client-owned mapping a daemon
183-// uses solely to resolve live attention while rendering its route snapshot.201+// uses to resolve live attention and retire authoritatively absent lifecycles.
184 type RouteAttentionSubscription struct {202 type RouteAttentionSubscription struct {
185 Targets []RouteAttentionTarget203 Targets []RouteAttentionTarget
186 }204 }
M internal/protocol/session.go +1 -1
@@ -13,7 +13,7 @@ import (
1313 )
1414
1515 // Version is the negotiated vev session protocol version.
16-const Version uint16 = 42
16+const Version uint16 = 43
1717
1818 // HandshakeTimeout bounds every transport handshake from connect through the
1919 // first committed publication.
@@ -13,7 +13,7 @@ import (
13 )13 )
14 14
15 // Version is the negotiated vev session protocol version.15 // Version is the negotiated vev session protocol version.
16-const Version uint16 = 4216+const Version uint16 = 43
17 17
18 // HandshakeTimeout bounds every transport handshake from connect through the18 // HandshakeTimeout bounds every transport handshake from connect through the
19 // first committed publication.19 // first committed publication.
M internal/protocol/session_test.go +1 -1
@@ -10,7 +10,7 @@ import (
1010 )
1111
1212 func TestVersionRemainsWireCompatible(t *testing.T) {
13- require.Equal(t, uint16(42), Version)
13+ require.Equal(t, uint16(43), Version)
1414 require.Equal(t, (16<<20)-407, MaxOutputDataLen)
1515 }
1616
@@ -10,7 +10,7 @@ import (
10 )10 )
11 11
12 func TestVersionRemainsWireCompatible(t *testing.T) {12 func TestVersionRemainsWireCompatible(t *testing.T) {
13- require.Equal(t, uint16(42), Version)13+ require.Equal(t, uint16(43), Version)
14 require.Equal(t, (16<<20)-407, MaxOutputDataLen)14 require.Equal(t, (16<<20)-407, MaxOutputDataLen)
15 }15 }
16 16
M internal/protocol/wire/frame.go +2 -1
@@ -10,7 +10,7 @@ type MsgType uint8
1010 // Frame message types use non-contiguous allocations: the legacy client range
1111 // is 1–13 and 15, and newer client controls use 32–33, 35, 37, 39, 42, 45,
1212 // and 47. Server-originated messages occupy 16–23, 25–31, 34, 36, 38, 40–41,
13-// and 43–44 and 46. Values 14 and 24 remain reserved for future extensions.
13+// and 43–44, 46, and 48. Values 14 and 24 remain reserved for future extensions.
1414 const (
1515 MsgHello MsgType = 1
1616 MsgInput MsgType = 2
@@ -58,6 +58,7 @@ const (
5858 MsgNavigationInventoryResponse MsgType = 43
5959 MsgNavigationInventoryDemand MsgType = 44
6060 MsgNavigationInventorySelection MsgType = 46
61+ MsgRouteRetired MsgType = 48
6162 )
6263
6364 // Frame is the unit of exchange over a Transport: a typed, length-delimited
@@ -10,7 +10,7 @@ type MsgType uint8
10 // Frame message types use non-contiguous allocations: the legacy client range10 // Frame message types use non-contiguous allocations: the legacy client range
11 // is 1–13 and 15, and newer client controls use 32–33, 35, 37, 39, 42, 45,11 // is 1–13 and 15, and newer client controls use 32–33, 35, 37, 39, 42, 45,
12 // and 47. Server-originated messages occupy 16–23, 25–31, 34, 36, 38, 40–41,12 // and 47. Server-originated messages occupy 16–23, 25–31, 34, 36, 38, 40–41,
13-// and 43–44 and 46. Values 14 and 24 remain reserved for future extensions.13+// and 43–44, 46, and 48. Values 14 and 24 remain reserved for future extensions.
14 const (14 const (
15 MsgHello MsgType = 115 MsgHello MsgType = 1
16 MsgInput MsgType = 216 MsgInput MsgType = 2
@@ -58,6 +58,7 @@ const (
58 MsgNavigationInventoryResponse MsgType = 4358 MsgNavigationInventoryResponse MsgType = 43
59 MsgNavigationInventoryDemand MsgType = 4459 MsgNavigationInventoryDemand MsgType = 44
60 MsgNavigationInventorySelection MsgType = 4660 MsgNavigationInventorySelection MsgType = 46
61+ MsgRouteRetired MsgType = 48
61 )62 )
62 63
63 // Frame is the unit of exchange over a Transport: a typed, length-delimited64 // Frame is the unit of exchange over a Transport: a typed, length-delimited
M internal/protocol/wire/frame_test.go +2 -2
@@ -7,8 +7,8 @@ import (
77 )
88
99 func TestProtocolVersion(t *testing.T) {
10- if protocol.Version != 42 {
11- t.Fatalf("ProtocolVersion = %d, want 42", protocol.Version)
10+ if protocol.Version != 43 {
11+ t.Fatalf("ProtocolVersion = %d, want 43", protocol.Version)
1212 }
1313 }
1414
@@ -7,8 +7,8 @@ import (
7 )7 )
8 8
9 func TestProtocolVersion(t *testing.T) {9 func TestProtocolVersion(t *testing.T) {
10- if protocol.Version != 42 {10+ if protocol.Version != 43 {
11- t.Fatalf("ProtocolVersion = %d, want 42", protocol.Version)11+ t.Fatalf("ProtocolVersion = %d, want 43", protocol.Version)
12 }12 }
13 }13 }
14 14
M internal/protocol/wire/navigation_test.go +1 -1
@@ -153,7 +153,7 @@ func TestHelloNavigationValidationTable(t *testing.T) {
153153 {name: "client-owned home picker rejected", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityHomePicker}, valid: false},
154154 {name: "daemon-owned back picker rejected", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityBack, StartupOverlay: protocol.StartupOverlaySessionPicker, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned}, valid: false},
155155 {name: "remote-target back picker rejected", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Name: "work", Size: domain.Size{Cols: 80, Rows: 24}, RemoteTarget: remoteTarget, NavigationCapabilities: protocol.NavigationCapabilityBack, StartupOverlay: protocol.StartupOverlaySessionPicker, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned}, valid: false},
156- {name: "back with startup picker", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityBack, StartupOverlay: protocol.StartupOverlaySessionPicker}, valid: true, capability: protocol.NavigationCapabilityBack, overlay: protocol.StartupOverlaySessionPicker, wantPayload: "002a02" + "00000000000000000000000000000000" + "0000000000000000" + "0000" + "00500018" + "00000000" + "0000" + "0000" + "00" + "00" + "00000000" + "00" + "00" + "00" + "0000" + "02010000"},
156+ {name: "back with startup picker", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityBack, StartupOverlay: protocol.StartupOverlaySessionPicker}, valid: true, capability: protocol.NavigationCapabilityBack, overlay: protocol.StartupOverlaySessionPicker, wantPayload: "002b02" + "00000000000000000000000000000000" + "0000000000000000" + "0000" + "00500018" + "00000000" + "0000" + "0000" + "00" + "00" + "00000000" + "00" + "00" + "00" + "0000" + "02010000"},
157157 {name: "new accepts retained home picker", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentNew, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityHomePicker}, valid: true, capability: protocol.NavigationCapabilityHomePicker},
158158 {name: "new rejects back navigation", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentNew, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityBack, StartupOverlay: protocol.StartupOverlaySessionPicker}, valid: false},
159159 {name: "unknown capability", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: 8}, valid: false},
@@ -153,7 +153,7 @@ func TestHelloNavigationValidationTable(t *testing.T) {
153 {name: "client-owned home picker rejected", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityHomePicker}, valid: false},153 {name: "client-owned home picker rejected", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityHomePicker}, valid: false},
154 {name: "daemon-owned back picker rejected", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityBack, StartupOverlay: protocol.StartupOverlaySessionPicker, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned}, valid: false},154 {name: "daemon-owned back picker rejected", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityBack, StartupOverlay: protocol.StartupOverlaySessionPicker, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned}, valid: false},
155 {name: "remote-target back picker rejected", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Name: "work", Size: domain.Size{Cols: 80, Rows: 24}, RemoteTarget: remoteTarget, NavigationCapabilities: protocol.NavigationCapabilityBack, StartupOverlay: protocol.StartupOverlaySessionPicker, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned}, valid: false},155 {name: "remote-target back picker rejected", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Name: "work", Size: domain.Size{Cols: 80, Rows: 24}, RemoteTarget: remoteTarget, NavigationCapabilities: protocol.NavigationCapabilityBack, StartupOverlay: protocol.StartupOverlaySessionPicker, EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned}, valid: false},
156- {name: "back with startup picker", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityBack, StartupOverlay: protocol.StartupOverlaySessionPicker}, valid: true, capability: protocol.NavigationCapabilityBack, overlay: protocol.StartupOverlaySessionPicker, wantPayload: "002a02" + "00000000000000000000000000000000" + "0000000000000000" + "0000" + "00500018" + "00000000" + "0000" + "0000" + "00" + "00" + "00000000" + "00" + "00" + "00" + "0000" + "02010000"},156+ {name: "back with startup picker", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityBack, StartupOverlay: protocol.StartupOverlaySessionPicker}, valid: true, capability: protocol.NavigationCapabilityBack, overlay: protocol.StartupOverlaySessionPicker, wantPayload: "002b02" + "00000000000000000000000000000000" + "0000000000000000" + "0000" + "00500018" + "00000000" + "0000" + "0000" + "00" + "00" + "00000000" + "00" + "00" + "00" + "0000" + "02010000"},
157 {name: "new accepts retained home picker", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentNew, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityHomePicker}, valid: true, capability: protocol.NavigationCapabilityHomePicker},157 {name: "new accepts retained home picker", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentNew, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityHomePicker}, valid: true, capability: protocol.NavigationCapabilityHomePicker},
158 {name: "new rejects back navigation", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentNew, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityBack, StartupOverlay: protocol.StartupOverlaySessionPicker}, valid: false},158 {name: "new rejects back navigation", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentNew, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: protocol.NavigationCapabilityBack, StartupOverlay: protocol.StartupOverlaySessionPicker}, valid: false},
159 {name: "unknown capability", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: 8}, valid: false},159 {name: "unknown capability", hello: protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}, NavigationCapabilities: 8}, valid: false},
A internal/protocol/wire/route_retired.go +29 -0
@@ -0,0 +1,29 @@
1+package wire
2+
3+import "github.com/bnema/vev/internal/protocol"
4+
5+func MarshalRouteRetired(message protocol.RouteRetired) ([]byte, error) {
6+ if message.Ref.IsZero() || message.Ref.Validate() != nil || message.Target.Validate() != nil {
7+ return nil, protocol.ErrInvalidRouteWire
8+ }
9+ w := payloadWriter{}
10+ marshalRouteRef(&w, message.Ref)
11+ marshalExactSessionTarget(&w, message.Target)
12+ return w.b, nil
13+}
14+
15+func UnmarshalRouteRetired(b []byte) (protocol.RouteRetired, error) {
16+ r := payloadReader{b: b}
17+ ref, err := unmarshalRouteRef(&r)
18+ if err != nil || ref.IsZero() {
19+ return protocol.RouteRetired{}, protocol.ErrInvalidRouteWire
20+ }
21+ target, err := unmarshalExactSessionTarget(&r)
22+ if err != nil {
23+ return protocol.RouteRetired{}, err
24+ }
25+ if err := r.done(); err != nil {
26+ return protocol.RouteRetired{}, err
27+ }
28+ return protocol.RouteRetired{Ref: ref, Target: target}, nil
29+}
@@ -0,0 +1,29 @@
1+package wire
2+
3+import "github.com/bnema/vev/internal/protocol"
4+
5+func MarshalRouteRetired(message protocol.RouteRetired) ([]byte, error) {
6+ if message.Ref.IsZero() || message.Ref.Validate() != nil || message.Target.Validate() != nil {
7+ return nil, protocol.ErrInvalidRouteWire
8+ }
9+ w := payloadWriter{}
10+ marshalRouteRef(&w, message.Ref)
11+ marshalExactSessionTarget(&w, message.Target)
12+ return w.b, nil
13+}
14+
15+func UnmarshalRouteRetired(b []byte) (protocol.RouteRetired, error) {
16+ r := payloadReader{b: b}
17+ ref, err := unmarshalRouteRef(&r)
18+ if err != nil || ref.IsZero() {
19+ return protocol.RouteRetired{}, protocol.ErrInvalidRouteWire
20+ }
21+ target, err := unmarshalExactSessionTarget(&r)
22+ if err != nil {
23+ return protocol.RouteRetired{}, err
24+ }
25+ if err := r.done(); err != nil {
26+ return protocol.RouteRetired{}, err
27+ }
28+ return protocol.RouteRetired{Ref: ref, Target: target}, nil
29+}
A internal/protocol/wire/route_retired_test.go +26 -0
@@ -0,0 +1,26 @@
1+package wire
2+
3+import (
4+ "encoding/hex"
5+ "testing"
6+
7+ "github.com/bnema/vev/internal/protocol"
8+ "github.com/stretchr/testify/require"
9+)
10+
11+func TestRouteRetiredCodec(t *testing.T) {
12+ want := protocol.RouteRetired{Ref: protocol.RouteRef{Key: 7, Generation: 3}, Target: testExactTarget()}
13+ encoded, err := MarshalRouteRetired(want)
14+ require.NoError(t, err)
15+ require.Equal(t, "00000000000000070000000000000003010203000000000000000000000000000004776f726b", hex.EncodeToString(encoded))
16+ got, err := UnmarshalRouteRetired(encoded)
17+ require.NoError(t, err)
18+ require.Equal(t, want, got)
19+ assertAllPrefixesFail(t, encoded, UnmarshalRouteRetired)
20+ _, err = UnmarshalRouteRetired(append(append([]byte(nil), encoded...), 0))
21+ require.Error(t, err)
22+ for _, message := range []protocol.RouteRetired{{}, {Ref: want.Ref}, {Target: want.Target}, {Ref: protocol.RouteRef{Key: 1}, Target: want.Target}} {
23+ _, err := MarshalRouteRetired(message)
24+ require.Error(t, err)
25+ }
26+}
@@ -0,0 +1,26 @@
1+package wire
2+
3+import (
4+ "encoding/hex"
5+ "testing"
6+
7+ "github.com/bnema/vev/internal/protocol"
8+ "github.com/stretchr/testify/require"
9+)
10+
11+func TestRouteRetiredCodec(t *testing.T) {
12+ want := protocol.RouteRetired{Ref: protocol.RouteRef{Key: 7, Generation: 3}, Target: testExactTarget()}
13+ encoded, err := MarshalRouteRetired(want)
14+ require.NoError(t, err)
15+ require.Equal(t, "00000000000000070000000000000003010203000000000000000000000000000004776f726b", hex.EncodeToString(encoded))
16+ got, err := UnmarshalRouteRetired(encoded)
17+ require.NoError(t, err)
18+ require.Equal(t, want, got)
19+ assertAllPrefixesFail(t, encoded, UnmarshalRouteRetired)
20+ _, err = UnmarshalRouteRetired(append(append([]byte(nil), encoded...), 0))
21+ require.Error(t, err)
22+ for _, message := range []protocol.RouteRetired{{}, {Ref: want.Ref}, {Target: want.Target}, {Ref: protocol.RouteRef{Key: 1}, Target: want.Target}} {
23+ _, err := MarshalRouteRetired(message)
24+ require.Error(t, err)
25+ }
26+}
M internal/protocol/wire/routes_test.go +1 -1
@@ -182,7 +182,7 @@ func TestRouteAttentionSubscriptionCodecIsStrict(t *testing.T) {
182182 }}}
183183 encoded, err := MarshalRouteAttentionSubscription(want)
184184 require.NoError(t, err)
185- require.Equal(t, "0100000000000000070000000000000003010203000000000000000000000000000004776f726b", hex.EncodeToString(encoded))
185+ require.Equal(t, "0100000000000000070000000000000003010203000000000000000000000000000004776f726b0000", hex.EncodeToString(encoded))
186186 assertAllPrefixesFail(t, encoded, UnmarshalRouteAttentionSubscription)
187187 _, err = UnmarshalRouteAttentionSubscription(append(append([]byte(nil), encoded...), 0))
188188 require.Error(t, err)
@@ -182,7 +182,7 @@ func TestRouteAttentionSubscriptionCodecIsStrict(t *testing.T) {
182 }}}182 }}}
183 encoded, err := MarshalRouteAttentionSubscription(want)183 encoded, err := MarshalRouteAttentionSubscription(want)
184 require.NoError(t, err)184 require.NoError(t, err)
185- require.Equal(t, "0100000000000000070000000000000003010203000000000000000000000000000004776f726b", hex.EncodeToString(encoded))185+ require.Equal(t, "0100000000000000070000000000000003010203000000000000000000000000000004776f726b0000", hex.EncodeToString(encoded))
186 assertAllPrefixesFail(t, encoded, UnmarshalRouteAttentionSubscription)186 assertAllPrefixesFail(t, encoded, UnmarshalRouteAttentionSubscription)
187 _, err = UnmarshalRouteAttentionSubscription(append(append([]byte(nil), encoded...), 0))187 _, err = UnmarshalRouteAttentionSubscription(append(append([]byte(nil), encoded...), 0))
188 require.Error(t, err)188 require.Error(t, err)
M internal/protocol/wire/routes_wire.go +9 -1
@@ -478,6 +478,9 @@ func validateRouteAttentionSubscription(subscription protocol.RouteAttentionSubs
478478 }
479479 refs := make(map[protocol.RouteRef]struct{}, len(subscription.Targets))
480480 for _, target := range subscription.Targets {
481+ if err := protocol.ValidateRouteLabel(target.SourceKey, true); err != nil {
482+ return fmt.Errorf("%w: invalid route source", protocol.ErrInvalidRouteWire)
483+ }
481484 if err := validateRouteRef(target.Ref); err != nil || target.Ref.IsZero() {
482485 return fmt.Errorf("%w: invalid attention route reference", protocol.ErrInvalidRouteWire)
483486 }
@@ -503,6 +506,7 @@ func MarshalRouteAttentionSubscription(subscription protocol.RouteAttentionSubsc
503506 for _, target := range subscription.Targets {
504507 marshalRouteRef(&w, target.Ref)
505508 marshalExactSessionTarget(&w, target.Target)
509+ w.putString(target.SourceKey)
506510 }
507511 return w.b, nil
508512 }
@@ -528,7 +532,11 @@ func UnmarshalRouteAttentionSubscription(b []byte) (protocol.RouteAttentionSubsc
528532 if err != nil {
529533 return protocol.RouteAttentionSubscription{}, err
530534 }
531- subscription.Targets = append(subscription.Targets, protocol.RouteAttentionTarget{Ref: ref, Target: target})
535+ source, err := r.getString()
536+ if err != nil {
537+ return protocol.RouteAttentionSubscription{}, err
538+ }
539+ subscription.Targets = append(subscription.Targets, protocol.RouteAttentionTarget{Ref: ref, Target: target, SourceKey: source})
532540 }
533541 if err := r.done(); err != nil {
534542 return protocol.RouteAttentionSubscription{}, err
@@ -478,6 +478,9 @@ func validateRouteAttentionSubscription(subscription protocol.RouteAttentionSubs
478 }478 }
479 refs := make(map[protocol.RouteRef]struct{}, len(subscription.Targets))479 refs := make(map[protocol.RouteRef]struct{}, len(subscription.Targets))
480 for _, target := range subscription.Targets {480 for _, target := range subscription.Targets {
481+ if err := protocol.ValidateRouteLabel(target.SourceKey, true); err != nil {
482+ return fmt.Errorf("%w: invalid route source", protocol.ErrInvalidRouteWire)
483+ }
481 if err := validateRouteRef(target.Ref); err != nil || target.Ref.IsZero() {484 if err := validateRouteRef(target.Ref); err != nil || target.Ref.IsZero() {
482 return fmt.Errorf("%w: invalid attention route reference", protocol.ErrInvalidRouteWire)485 return fmt.Errorf("%w: invalid attention route reference", protocol.ErrInvalidRouteWire)
483 }486 }
@@ -503,6 +506,7 @@ func MarshalRouteAttentionSubscription(subscription protocol.RouteAttentionSubsc
503 for _, target := range subscription.Targets {506 for _, target := range subscription.Targets {
504 marshalRouteRef(&w, target.Ref)507 marshalRouteRef(&w, target.Ref)
505 marshalExactSessionTarget(&w, target.Target)508 marshalExactSessionTarget(&w, target.Target)
509+ w.putString(target.SourceKey)
506 }510 }
507 return w.b, nil511 return w.b, nil
508 }512 }
@@ -528,7 +532,11 @@ func UnmarshalRouteAttentionSubscription(b []byte) (protocol.RouteAttentionSubsc
528 if err != nil {532 if err != nil {
529 return protocol.RouteAttentionSubscription{}, err533 return protocol.RouteAttentionSubscription{}, err
530 }534 }
531- subscription.Targets = append(subscription.Targets, protocol.RouteAttentionTarget{Ref: ref, Target: target})535+ source, err := r.getString()
536+ if err != nil {
537+ return protocol.RouteAttentionSubscription{}, err
538+ }
539+ subscription.Targets = append(subscription.Targets, protocol.RouteAttentionTarget{Ref: ref, Target: target, SourceKey: source})
532 }540 }
533 if err := r.done(); err != nil {541 if err := r.done(); err != nil {
534 return protocol.RouteAttentionSubscription{}, err542 return protocol.RouteAttentionSubscription{}, err
M internal/protocol/wire/wire_final_test.go +3 -3
@@ -12,8 +12,8 @@ import (
1212 )
1313
1414 func TestFinalProtocolVersionAndNoV21Hello(t *testing.T) {
15- if protocol.Version != 42 {
16- t.Fatalf("ProtocolVersion = %d, want 42", protocol.Version)
15+ if protocol.Version != 43 {
16+ t.Fatalf("ProtocolVersion = %d, want 43", protocol.Version)
1717 }
1818 payload := MarshalHello(protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}})
1919 if len(payload) < 2 {
@@ -27,7 +27,7 @@ func TestFinalProtocolVersionAndNoV21Hello(t *testing.T) {
2727
2828 func TestFinalHelloGoldenStrict(t *testing.T) {
2929 msg := protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}}
30- want := append([]byte{0, 42, 2}, make([]byte, 16+8)...)
30+ want := append([]byte{0, 43, 2}, make([]byte, 16+8)...)
3131 want = append(want, 0, 0, 0, 80, 0, 24, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
3232 want = append(want, 0, 0)
3333 got := MarshalHello(msg)
@@ -12,8 +12,8 @@ import (
12 )12 )
13 13
14 func TestFinalProtocolVersionAndNoV21Hello(t *testing.T) {14 func TestFinalProtocolVersionAndNoV21Hello(t *testing.T) {
15- if protocol.Version != 42 {15+ if protocol.Version != 43 {
16- t.Fatalf("ProtocolVersion = %d, want 42", protocol.Version)16+ t.Fatalf("ProtocolVersion = %d, want 43", protocol.Version)
17 }17 }
18 payload := MarshalHello(protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}})18 payload := MarshalHello(protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}})
19 if len(payload) < 2 {19 if len(payload) < 2 {
@@ -27,7 +27,7 @@ func TestFinalProtocolVersionAndNoV21Hello(t *testing.T) {
27 27
28 func TestFinalHelloGoldenStrict(t *testing.T) {28 func TestFinalHelloGoldenStrict(t *testing.T) {
29 msg := protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}}29 msg := protocol.Hello{Version: protocol.Version, Intent: protocol.IntentAttach, Size: domain.Size{Cols: 80, Rows: 24}}
30- want := append([]byte{0, 42, 2}, make([]byte, 16+8)...)30+ want := append([]byte{0, 43, 2}, make([]byte, 16+8)...)
31 want = append(want, 0, 0, 0, 80, 0, 24, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)31 want = append(want, 0, 0, 0, 80, 0, 24, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
32 want = append(want, 0, 0)32 want = append(want, 0, 0)
33 got := MarshalHello(msg)33 got := MarshalHello(msg)
M internal/protocol/wire/wire_test.go +6 -6
@@ -136,7 +136,7 @@ func TestHelloEnvironmentCodec(t *testing.T) {
136136 }
137137 got := MarshalHello(msg)
138138 want := []byte{
139- 0x00, 0x2a, 0x00, // version, intent, first client ID byte
139+ 0x00, 0x2b, 0x00, // version, intent, first client ID byte
140140 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
141141 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // client ID
142142 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // resume token
@@ -367,7 +367,7 @@ func TestThemeGenerationClearedWireGoldenPreservesProtocolVersion(t *testing.T)
367367 }
368368 want := append([]byte{0x0f, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x00, 0x00}, make([]byte, 48)...)
369369 require.Equal(t, want, MarshalTheme(cleared))
370- require.Equal(t, uint16(42), protocol.Version)
370+ require.Equal(t, uint16(43), protocol.Version)
371371 }
372372
373373 func TestResizeGoldenAndRoundTrip(t *testing.T) {
@@ -528,7 +528,7 @@ func TestCommandRequestGoldenAndRoundTrip(t *testing.T) {
528528 name: "minimal",
529529 msg: protocol.CommandRequest{Version: protocol.Version, Slug: "split-right"},
530530 want: []byte{
531- 0x00, 0x2a,
531+ 0x00, 0x2b,
532532 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // request ID
533533 0x00, // attached
534534 0x00, // self
@@ -551,7 +551,7 @@ func TestCommandRequestGoldenAndRoundTrip(t *testing.T) {
551551 JSON: true,
552552 },
553553 want: []byte{
554- 0x00, 0x2a,
554+ 0x00, 0x2b,
555555 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // request ID
556556 0x00, // attached
557557 0x01, // self
@@ -999,7 +999,7 @@ func TestCommandCorrelationGoldenAndStrict(t *testing.T) {
999999 name: "attached",
10001000 msg: protocol.CommandRequest{Version: protocol.Version, RequestID: 0x0102030405060708, Attached: true, Self: true, Slug: "split-right", Args: []string{"--vertical"}},
10011001 want: []byte{
1002- 0x00, 0x2a, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x01, 0x01,
1002+ 0x00, 0x2b, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x01, 0x01,
10031003 0x00, 0x0b, 's', 'p', 'l', 'i', 't', '-', 'r', 'i', 'g', 'h', 't',
10041004 0x00, 0x01, 0x00, 0x00, 0x00, 0x0a, '-', '-', 'v', 'e', 'r', 't', 'i', 'c', 'a', 'l',
10051005 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
@@ -1009,7 +1009,7 @@ func TestCommandCorrelationGoldenAndStrict(t *testing.T) {
10091009 name: "control",
10101010 msg: protocol.CommandRequest{Version: protocol.Version, Slug: "ls"},
10111011 want: []byte{
1012- 0x00, 0x2a, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
1012+ 0x00, 0x2b, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
10131013 0x00, 0x02, 'l', 's', 0x00, 0x00,
10141014 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
10151015 },
@@ -136,7 +136,7 @@ func TestHelloEnvironmentCodec(t *testing.T) {
136 }136 }
137 got := MarshalHello(msg)137 got := MarshalHello(msg)
138 want := []byte{138 want := []byte{
139- 0x00, 0x2a, 0x00, // version, intent, first client ID byte139+ 0x00, 0x2b, 0x00, // version, intent, first client ID byte
140 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,140 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
141 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // client ID141 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // client ID
142 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // resume token142 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // resume token
@@ -367,7 +367,7 @@ func TestThemeGenerationClearedWireGoldenPreservesProtocolVersion(t *testing.T)
367 }367 }
368 want := append([]byte{0x0f, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x00, 0x00}, make([]byte, 48)...)368 want := append([]byte{0x0f, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x00, 0x00}, make([]byte, 48)...)
369 require.Equal(t, want, MarshalTheme(cleared))369 require.Equal(t, want, MarshalTheme(cleared))
370- require.Equal(t, uint16(42), protocol.Version)370+ require.Equal(t, uint16(43), protocol.Version)
371 }371 }
372 372
373 func TestResizeGoldenAndRoundTrip(t *testing.T) {373 func TestResizeGoldenAndRoundTrip(t *testing.T) {
@@ -528,7 +528,7 @@ func TestCommandRequestGoldenAndRoundTrip(t *testing.T) {
528 name: "minimal",528 name: "minimal",
529 msg: protocol.CommandRequest{Version: protocol.Version, Slug: "split-right"},529 msg: protocol.CommandRequest{Version: protocol.Version, Slug: "split-right"},
530 want: []byte{530 want: []byte{
531- 0x00, 0x2a,531+ 0x00, 0x2b,
532 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // request ID532 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // request ID
533 0x00, // attached533 0x00, // attached
534 0x00, // self534 0x00, // self
@@ -551,7 +551,7 @@ func TestCommandRequestGoldenAndRoundTrip(t *testing.T) {
551 JSON: true,551 JSON: true,
552 },552 },
553 want: []byte{553 want: []byte{
554- 0x00, 0x2a,554+ 0x00, 0x2b,
555 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // request ID555 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // request ID
556 0x00, // attached556 0x00, // attached
557 0x01, // self557 0x01, // self
@@ -999,7 +999,7 @@ func TestCommandCorrelationGoldenAndStrict(t *testing.T) {
999 name: "attached",999 name: "attached",
1000 msg: protocol.CommandRequest{Version: protocol.Version, RequestID: 0x0102030405060708, Attached: true, Self: true, Slug: "split-right", Args: []string{"--vertical"}},1000 msg: protocol.CommandRequest{Version: protocol.Version, RequestID: 0x0102030405060708, Attached: true, Self: true, Slug: "split-right", Args: []string{"--vertical"}},
1001 want: []byte{1001 want: []byte{
1002- 0x00, 0x2a, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x01, 0x01,1002+ 0x00, 0x2b, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x01, 0x01,
1003 0x00, 0x0b, 's', 'p', 'l', 'i', 't', '-', 'r', 'i', 'g', 'h', 't',1003 0x00, 0x0b, 's', 'p', 'l', 'i', 't', '-', 'r', 'i', 'g', 'h', 't',
1004 0x00, 0x01, 0x00, 0x00, 0x00, 0x0a, '-', '-', 'v', 'e', 'r', 't', 'i', 'c', 'a', 'l',1004 0x00, 0x01, 0x00, 0x00, 0x00, 0x0a, '-', '-', 'v', 'e', 'r', 't', 'i', 'c', 'a', 'l',
1005 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,1005 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
@@ -1009,7 +1009,7 @@ func TestCommandCorrelationGoldenAndStrict(t *testing.T) {
1009 name: "control",1009 name: "control",
1010 msg: protocol.CommandRequest{Version: protocol.Version, Slug: "ls"},1010 msg: protocol.CommandRequest{Version: protocol.Version, Slug: "ls"},
1011 want: []byte{1011 want: []byte{
1012- 0x00, 0x2a, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,1012+ 0x00, 0x2b, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
1013 0x00, 0x02, 'l', 's', 0x00, 0x00,1013 0x00, 0x02, 'l', 's', 0x00, 0x00,
1014 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,1014 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
1015 },1015 },
M internal/usecase/client/client.go +11 -2
@@ -1668,7 +1668,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
16681668 }
16691669 }
16701670 snapshot := a.runner.ledger.snapshot()
1671- attention := a.runner.ledger.attentionSubscription()
1671+ attention := a.runner.ledger.attentionSubscriptionFor(request)
16721672 if err := sendHandshake(func() error {
16731673 if err := transport.SendClient(snapshot); err != nil {
16741674 return err
@@ -2174,7 +2174,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
21742174 }
21752175 messages := []protocol.ClientMessage{
21762176 a.runner.ledger.snapshot(),
2177- a.runner.ledger.attentionSubscription(),
2177+ a.runner.ledger.attentionSubscriptionFor(request),
21782178 }
21792179 for _, message := range messages {
21802180 select {
@@ -2614,6 +2614,9 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
26142614 if response.Status != protocol.ParkedRouteResumed {
26152615 return welcomedResult(errors.New("vev: parked route could not resume"))
26162616 }
2617+ if err := publishRouteSnapshot(); err != nil {
2618+ return welcomedResult(err)
2619+ }
26172620 awaitParkedFull()
26182621 case protocol.ParkedRouteSwitch:
26192622 switch response.Status {
@@ -2818,6 +2821,12 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
28182821 return welcomedResult(fmt.Errorf("vev: remembering route position: %w", derr))
28192822 }
28202823 routePosition = cloneRoutePosition(&position)
2824+ case protocol.RouteRetired:
2825+ if a.runner.ledger != nil && a.runner.ledger.retireRoute(message, a.runner.ledger.attentionSubscriptionFor(request)) {
2826+ if err := publishRouteSnapshot(); err != nil {
2827+ return welcomedResult(err)
2828+ }
2829+ }
28212830 case protocol.RouteNavigationFailure:
28222831 failure := message
28232832 log.Warn("route navigation rejected", "key", failure.Key, "generation", failure.Generation, "code", failure.Code)
@@ -1668,7 +1668,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
1668 }1668 }
1669 }1669 }
1670 snapshot := a.runner.ledger.snapshot()1670 snapshot := a.runner.ledger.snapshot()
1671- attention := a.runner.ledger.attentionSubscription()1671+ attention := a.runner.ledger.attentionSubscriptionFor(request)
1672 if err := sendHandshake(func() error {1672 if err := sendHandshake(func() error {
1673 if err := transport.SendClient(snapshot); err != nil {1673 if err := transport.SendClient(snapshot); err != nil {
1674 return err1674 return err
@@ -2174,7 +2174,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
2174 }2174 }
2175 messages := []protocol.ClientMessage{2175 messages := []protocol.ClientMessage{
2176 a.runner.ledger.snapshot(),2176 a.runner.ledger.snapshot(),
2177- a.runner.ledger.attentionSubscription(),2177+ a.runner.ledger.attentionSubscriptionFor(request),
2178 }2178 }
2179 for _, message := range messages {2179 for _, message := range messages {
2180 select {2180 select {
@@ -2614,6 +2614,9 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
2614 if response.Status != protocol.ParkedRouteResumed {2614 if response.Status != protocol.ParkedRouteResumed {
2615 return welcomedResult(errors.New("vev: parked route could not resume"))2615 return welcomedResult(errors.New("vev: parked route could not resume"))
2616 }2616 }
2617+ if err := publishRouteSnapshot(); err != nil {
2618+ return welcomedResult(err)
2619+ }
2617 awaitParkedFull()2620 awaitParkedFull()
2618 case protocol.ParkedRouteSwitch:2621 case protocol.ParkedRouteSwitch:
2619 switch response.Status {2622 switch response.Status {
@@ -2818,6 +2821,12 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
2818 return welcomedResult(fmt.Errorf("vev: remembering route position: %w", derr))2821 return welcomedResult(fmt.Errorf("vev: remembering route position: %w", derr))
2819 }2822 }
2820 routePosition = cloneRoutePosition(&position)2823 routePosition = cloneRoutePosition(&position)
2824+ case protocol.RouteRetired:
2825+ if a.runner.ledger != nil && a.runner.ledger.retireRoute(message, a.runner.ledger.attentionSubscriptionFor(request)) {
2826+ if err := publishRouteSnapshot(); err != nil {
2827+ return welcomedResult(err)
2828+ }
2829+ }
2821 case protocol.RouteNavigationFailure:2830 case protocol.RouteNavigationFailure:
2822 failure := message2831 failure := message
2823 log.Warn("route navigation rejected", "key", failure.Key, "generation", failure.Generation, "code", failure.Code)2832 log.Warn("route navigation rejected", "key", failure.Key, "generation", failure.Generation, "code", failure.Code)
M internal/usecase/client/home_picker_context_test.go +67 -3
@@ -18,9 +18,12 @@ func TestHomePickerPreservesActiveRouteSnapshot(t *testing.T) {
1818 udp bool
1919 rename bool
2020 selectLocal bool
21+ retireLocal bool
2122 }{
2223 {name: "UDP", udp: true},
2324 {name: "stdio"},
25+ {name: "UDP local deletion", udp: true, retireLocal: true},
26+ {name: "stdio local deletion", retireLocal: true},
2427 {name: "UDP backing session renamed", udp: true, rename: true},
2528 {name: "stdio backing session renamed", rename: true},
2629 {name: "stdio select new local destination", selectLocal: true},
@@ -47,14 +50,34 @@ func TestHomePickerPreservesActiveRouteSnapshot(t *testing.T) {
4750 remoteDialer := &sequenceDialer{trs: []wire.Transport{remote, remoteReturn}}
4851 if tc.udp {
4952 prepareSent, resumeSent := make(chan struct{}), make(chan struct{})
53+ var retiredSnapshot chan struct{}
54+ if tc.retireLocal {
55+ retiredSnapshot = make(chan struct{})
56+ }
5057 remote.onSend = hybridParkedRequestHandler(nil, nil, map[protocol.ParkedRouteAction]chan struct{}{
5158 protocol.ParkedRoutePrepare: prepareSent,
5259 protocol.ParkedRouteResume: resumeSent,
5360 })
61+ if tc.retireLocal {
62+ parkedHandler := remote.onSend
63+ remote.onSend = func(frame wire.Frame) {
64+ parkedHandler(frame)
65+ if frame.Type == wire.MsgRecentRouteSnapshot {
66+ snapshot, err := wire.UnmarshalRecentRouteSnapshot(frame.Payload)
67+ if err == nil && snapshot.Generation > 2 && len(snapshot.Entries) == 0 {
68+ select {
69+ case <-retiredSnapshot:
70+ default:
71+ close(retiredSnapshot)
72+ }
73+ }
74+ }
75+ }
76+ }
5477 remote.recvs = append(remote.recvs,
5578 recvItem{f: frameOf(wire.MsgParkedRouteResponse, wire.MarshalParkedRouteResponse(protocol.ParkedRouteResponse{RequestID: 1, Status: protocol.ParkedRouteReady})), wait: prepareSent},
5679 recvItem{f: frameOf(wire.MsgParkedRouteResponse, wire.MarshalParkedRouteResponse(protocol.ParkedRouteResponse{RequestID: 2, Status: protocol.ParkedRouteResumed})), wait: resumeSent},
57- recvItem{f: frameOf(wire.MsgDetached, wire.MarshalDetached(protocol.Detached{Reason: protocol.ReasonDetach}))},
80+ recvItem{f: frameOf(wire.MsgDetached, wire.MarshalDetached(protocol.Detached{Reason: protocol.ReasonDetach})), wait: retiredSnapshot},
5881 )
5982 remoteDialer.trs = []wire.Transport{markedDatagramTransport{Transport: remote}}
6083 }
@@ -64,6 +87,25 @@ func TestHomePickerPreservesActiveRouteSnapshot(t *testing.T) {
6487 Target: protocol.ExactSessionTarget{LifecycleID: localLifecycle, SessionName: "renamed-local"},
6588 }))})
6689 }
90+ var localRetiredSnapshot chan struct{}
91+ if tc.retireLocal {
92+ localRetiredSnapshot = make(chan struct{})
93+ localPicker.onSend = func(frame wire.Frame) {
94+ if frame.Type == wire.MsgRecentRouteSnapshot {
95+ snapshot, err := wire.UnmarshalRecentRouteSnapshot(frame.Payload)
96+ if err == nil && snapshot.Generation > 2 && len(snapshot.Entries) == 0 {
97+ select {
98+ case <-localRetiredSnapshot:
99+ default:
100+ close(localRetiredSnapshot)
101+ }
102+ }
103+ }
104+ }
105+ payload, err := wire.MarshalRouteRetired(protocol.RouteRetired{Ref: protocol.RouteRef{Key: 1, Generation: 1}, Target: protocol.ExactSessionTarget{LifecycleID: localLifecycle, SessionName: "local"}})
106+ require.NoError(t, err)
107+ localPicker.recvs = append(localPicker.recvs, recvItem{f: frameOf(wire.MsgRouteRetired, payload)})
108+ }
67109 localDialer := &sequenceDialer{trs: []wire.Transport{initial, localPicker}}
68110 var localDestination *recordingTransport
69111 if tc.selectLocal {
@@ -78,7 +120,7 @@ func TestHomePickerPreservesActiveRouteSnapshot(t *testing.T) {
78120 }}
79121 localDialer.trs = append(localDialer.trs, localDestination)
80122 } else {
81- localPicker.recvs = append(localPicker.recvs, recvItem{f: navigationDirectiveFrame(protocol.NavigationBack)})
123+ localPicker.recvs = append(localPicker.recvs, recvItem{f: navigationDirectiveFrame(protocol.NavigationBack), wait: localRetiredSnapshot})
82124 }
83125 clock := &reconnectTestClock{}
84126 deps := hybridPickerDependencies(localDialer, term, clock, map[string]ports.ClientDialer{"igor": remoteDialer})
@@ -104,9 +146,31 @@ func TestHomePickerPreservesActiveRouteSnapshot(t *testing.T) {
104146 publications++
105147 snapshot, err := wire.UnmarshalRecentRouteSnapshot(sent.Payload)
106148 require.NoError(t, err)
107- require.Equal(t, source, snapshot, "rendering the home picker must not commit its backing session as the active route")
149+ if tc.retireLocal && publications > 1 {
150+ require.Empty(t, snapshot.Entries)
151+ require.Equal(t, source.Active, snapshot.Active)
152+ } else {
153+ require.Equal(t, source, snapshot, "rendering the home picker must not commit its backing session as the active route")
154+ }
108155 }
109156 require.Positive(t, publications)
157+ if tc.retireLocal {
158+ require.Greater(t, publications, 1)
159+ returned := remoteReturn
160+ if tc.udp {
161+ returned = remote
162+ }
163+ var latest protocol.RecentRouteSnapshot
164+ for _, sent := range returned.Sends() {
165+ if sent.Type == wire.MsgRecentRouteSnapshot {
166+ var err error
167+ latest, err = wire.UnmarshalRecentRouteSnapshot(sent.Payload)
168+ require.NoError(t, err)
169+ }
170+ }
171+ require.Equal(t, source.ActiveEntry.Target, latest.ActiveEntry.Target)
172+ require.Empty(t, latest.Entries, "returning to remote must publish the pruned ledger")
173+ }
110174 wantDials := int32(2)
111175 if tc.udp || tc.selectLocal {
112176 wantDials = 1
@@ -18,9 +18,12 @@ func TestHomePickerPreservesActiveRouteSnapshot(t *testing.T) {
18 udp bool18 udp bool
19 rename bool19 rename bool
20 selectLocal bool20 selectLocal bool
21+ retireLocal bool
21 }{22 }{
22 {name: "UDP", udp: true},23 {name: "UDP", udp: true},
23 {name: "stdio"},24 {name: "stdio"},
25+ {name: "UDP local deletion", udp: true, retireLocal: true},
26+ {name: "stdio local deletion", retireLocal: true},
24 {name: "UDP backing session renamed", udp: true, rename: true},27 {name: "UDP backing session renamed", udp: true, rename: true},
25 {name: "stdio backing session renamed", rename: true},28 {name: "stdio backing session renamed", rename: true},
26 {name: "stdio select new local destination", selectLocal: true},29 {name: "stdio select new local destination", selectLocal: true},
@@ -47,14 +50,34 @@ func TestHomePickerPreservesActiveRouteSnapshot(t *testing.T) {
47 remoteDialer := &sequenceDialer{trs: []wire.Transport{remote, remoteReturn}}50 remoteDialer := &sequenceDialer{trs: []wire.Transport{remote, remoteReturn}}
48 if tc.udp {51 if tc.udp {
49 prepareSent, resumeSent := make(chan struct{}), make(chan struct{})52 prepareSent, resumeSent := make(chan struct{}), make(chan struct{})
53+ var retiredSnapshot chan struct{}
54+ if tc.retireLocal {
55+ retiredSnapshot = make(chan struct{})
56+ }
50 remote.onSend = hybridParkedRequestHandler(nil, nil, map[protocol.ParkedRouteAction]chan struct{}{57 remote.onSend = hybridParkedRequestHandler(nil, nil, map[protocol.ParkedRouteAction]chan struct{}{
51 protocol.ParkedRoutePrepare: prepareSent,58 protocol.ParkedRoutePrepare: prepareSent,
52 protocol.ParkedRouteResume: resumeSent,59 protocol.ParkedRouteResume: resumeSent,
53 })60 })
61+ if tc.retireLocal {
62+ parkedHandler := remote.onSend
63+ remote.onSend = func(frame wire.Frame) {
64+ parkedHandler(frame)
65+ if frame.Type == wire.MsgRecentRouteSnapshot {
66+ snapshot, err := wire.UnmarshalRecentRouteSnapshot(frame.Payload)
67+ if err == nil && snapshot.Generation > 2 && len(snapshot.Entries) == 0 {
68+ select {
69+ case <-retiredSnapshot:
70+ default:
71+ close(retiredSnapshot)
72+ }
73+ }
74+ }
75+ }
76+ }
54 remote.recvs = append(remote.recvs,77 remote.recvs = append(remote.recvs,
55 recvItem{f: frameOf(wire.MsgParkedRouteResponse, wire.MarshalParkedRouteResponse(protocol.ParkedRouteResponse{RequestID: 1, Status: protocol.ParkedRouteReady})), wait: prepareSent},78 recvItem{f: frameOf(wire.MsgParkedRouteResponse, wire.MarshalParkedRouteResponse(protocol.ParkedRouteResponse{RequestID: 1, Status: protocol.ParkedRouteReady})), wait: prepareSent},
56 recvItem{f: frameOf(wire.MsgParkedRouteResponse, wire.MarshalParkedRouteResponse(protocol.ParkedRouteResponse{RequestID: 2, Status: protocol.ParkedRouteResumed})), wait: resumeSent},79 recvItem{f: frameOf(wire.MsgParkedRouteResponse, wire.MarshalParkedRouteResponse(protocol.ParkedRouteResponse{RequestID: 2, Status: protocol.ParkedRouteResumed})), wait: resumeSent},
57- recvItem{f: frameOf(wire.MsgDetached, wire.MarshalDetached(protocol.Detached{Reason: protocol.ReasonDetach}))},80+ recvItem{f: frameOf(wire.MsgDetached, wire.MarshalDetached(protocol.Detached{Reason: protocol.ReasonDetach})), wait: retiredSnapshot},
58 )81 )
59 remoteDialer.trs = []wire.Transport{markedDatagramTransport{Transport: remote}}82 remoteDialer.trs = []wire.Transport{markedDatagramTransport{Transport: remote}}
60 }83 }
@@ -64,6 +87,25 @@ func TestHomePickerPreservesActiveRouteSnapshot(t *testing.T) {
64 Target: protocol.ExactSessionTarget{LifecycleID: localLifecycle, SessionName: "renamed-local"},87 Target: protocol.ExactSessionTarget{LifecycleID: localLifecycle, SessionName: "renamed-local"},
65 }))})88 }))})
66 }89 }
90+ var localRetiredSnapshot chan struct{}
91+ if tc.retireLocal {
92+ localRetiredSnapshot = make(chan struct{})
93+ localPicker.onSend = func(frame wire.Frame) {
94+ if frame.Type == wire.MsgRecentRouteSnapshot {
95+ snapshot, err := wire.UnmarshalRecentRouteSnapshot(frame.Payload)
96+ if err == nil && snapshot.Generation > 2 && len(snapshot.Entries) == 0 {
97+ select {
98+ case <-localRetiredSnapshot:
99+ default:
100+ close(localRetiredSnapshot)
101+ }
102+ }
103+ }
104+ }
105+ payload, err := wire.MarshalRouteRetired(protocol.RouteRetired{Ref: protocol.RouteRef{Key: 1, Generation: 1}, Target: protocol.ExactSessionTarget{LifecycleID: localLifecycle, SessionName: "local"}})
106+ require.NoError(t, err)
107+ localPicker.recvs = append(localPicker.recvs, recvItem{f: frameOf(wire.MsgRouteRetired, payload)})
108+ }
67 localDialer := &sequenceDialer{trs: []wire.Transport{initial, localPicker}}109 localDialer := &sequenceDialer{trs: []wire.Transport{initial, localPicker}}
68 var localDestination *recordingTransport110 var localDestination *recordingTransport
69 if tc.selectLocal {111 if tc.selectLocal {
@@ -78,7 +120,7 @@ func TestHomePickerPreservesActiveRouteSnapshot(t *testing.T) {
78 }}120 }}
79 localDialer.trs = append(localDialer.trs, localDestination)121 localDialer.trs = append(localDialer.trs, localDestination)
80 } else {122 } else {
81- localPicker.recvs = append(localPicker.recvs, recvItem{f: navigationDirectiveFrame(protocol.NavigationBack)})123+ localPicker.recvs = append(localPicker.recvs, recvItem{f: navigationDirectiveFrame(protocol.NavigationBack), wait: localRetiredSnapshot})
82 }124 }
83 clock := &reconnectTestClock{}125 clock := &reconnectTestClock{}
84 deps := hybridPickerDependencies(localDialer, term, clock, map[string]ports.ClientDialer{"igor": remoteDialer})126 deps := hybridPickerDependencies(localDialer, term, clock, map[string]ports.ClientDialer{"igor": remoteDialer})
@@ -104,9 +146,31 @@ func TestHomePickerPreservesActiveRouteSnapshot(t *testing.T) {
104 publications++146 publications++
105 snapshot, err := wire.UnmarshalRecentRouteSnapshot(sent.Payload)147 snapshot, err := wire.UnmarshalRecentRouteSnapshot(sent.Payload)
106 require.NoError(t, err)148 require.NoError(t, err)
107- require.Equal(t, source, snapshot, "rendering the home picker must not commit its backing session as the active route")149+ if tc.retireLocal && publications > 1 {
150+ require.Empty(t, snapshot.Entries)
151+ require.Equal(t, source.Active, snapshot.Active)
152+ } else {
153+ require.Equal(t, source, snapshot, "rendering the home picker must not commit its backing session as the active route")
154+ }
108 }155 }
109 require.Positive(t, publications)156 require.Positive(t, publications)
157+ if tc.retireLocal {
158+ require.Greater(t, publications, 1)
159+ returned := remoteReturn
160+ if tc.udp {
161+ returned = remote
162+ }
163+ var latest protocol.RecentRouteSnapshot
164+ for _, sent := range returned.Sends() {
165+ if sent.Type == wire.MsgRecentRouteSnapshot {
166+ var err error
167+ latest, err = wire.UnmarshalRecentRouteSnapshot(sent.Payload)
168+ require.NoError(t, err)
169+ }
170+ }
171+ require.Equal(t, source.ActiveEntry.Target, latest.ActiveEntry.Target)
172+ require.Empty(t, latest.Entries, "returning to remote must publish the pruned ledger")
173+ }
110 wantDials := int32(2)174 wantDials := int32(2)
111 if tc.udp || tc.selectLocal {175 if tc.udp || tc.selectLocal {
112 wantDials = 1176 wantDials = 1
A internal/usecase/client/route_deletion_test.go +61 -0
@@ -0,0 +1,61 @@
1+package client
2+
3+import (
4+ "testing"
5+
6+ "github.com/bnema/vev/internal/protocol"
7+ "github.com/stretchr/testify/require"
8+)
9+
10+func TestRouteRetirementThenRecreationDoesNotDuplicateHistory(t *testing.T) {
11+ for _, origin := range []protocol.RouteOrigin{protocol.RouteOriginLocal, protocol.RouteOriginRemote} {
12+ t.Run(map[protocol.RouteOrigin]string{protocol.RouteOriginLocal: "local", protocol.RouteOriginRemote: "remote"}[origin], func(t *testing.T) {
13+ ledger := newRouteLedger()
14+ victim := routeTestCandidate(0, origin)
15+ victim.originKey = "source"
16+ id, err := ledger.commit(victim)
17+ require.NoError(t, err)
18+ active := routeTestCandidate(1, origin)
19+ active.originKey = "source"
20+ _, err = ledger.commit(active)
21+ require.NoError(t, err)
22+ sub := ledger.attentionSubscriptionFor(AttachRequest{Origin: origin, OriginKey: "source"})
23+ retired := protocol.RouteRetired{Ref: id.wire(), Target: victim.target}
24+ wrong := retired
25+ wrong.Target = active.target
26+ require.False(t, ledger.retireRoute(wrong, sub))
27+ require.False(t, ledger.retireRoute(retired, protocol.RouteAttentionSubscription{}))
28+ require.True(t, ledger.retireRoute(retired, sub))
29+ require.Empty(t, ledger.snapshot().Entries)
30+ replacement := victim
31+ replacement.target.LifecycleID[0] = 9
32+ _, err = ledger.commit(replacement)
33+ require.NoError(t, err)
34+ _, err = ledger.commit(active)
35+ require.NoError(t, err)
36+ require.False(t, ledger.retireRoute(retired, sub), "late deletion must not retire recreation")
37+ snapshot := ledger.snapshot()
38+ require.Len(t, snapshot.Entries, 1)
39+ require.Equal(t, replacement.target, snapshot.Entries[0].Target)
40+ })
41+ }
42+}
43+
44+func TestHomePickerSubscribesToServingDaemonHistory(t *testing.T) {
45+ ledger := newRouteLedger()
46+ local := routeTestCandidate(0, protocol.RouteOriginLocal)
47+ local.originKey = "local"
48+ localID, err := ledger.commit(local)
49+ require.NoError(t, err)
50+ remote := routeTestCandidate(1, protocol.RouteOriginRemote)
51+ remote.originKey = "remote"
52+ _, err = ledger.commit(remote)
53+ require.NoError(t, err)
54+
55+ // A transient home picker preserves the remote active route, but its
56+ // lifecycle observations must cover the local daemon it actually serves.
57+ subscription := ledger.attentionSubscriptionFor(AttachRequest{Origin: protocol.RouteOriginLocal, OriginKey: "local"})
58+ require.Contains(t, subscription.Targets, protocol.RouteAttentionTarget{
59+ Ref: localID.wire(), Target: local.target,
60+ })
61+}
@@ -0,0 +1,61 @@
1+package client
2+
3+import (
4+ "testing"
5+
6+ "github.com/bnema/vev/internal/protocol"
7+ "github.com/stretchr/testify/require"
8+)
9+
10+func TestRouteRetirementThenRecreationDoesNotDuplicateHistory(t *testing.T) {
11+ for _, origin := range []protocol.RouteOrigin{protocol.RouteOriginLocal, protocol.RouteOriginRemote} {
12+ t.Run(map[protocol.RouteOrigin]string{protocol.RouteOriginLocal: "local", protocol.RouteOriginRemote: "remote"}[origin], func(t *testing.T) {
13+ ledger := newRouteLedger()
14+ victim := routeTestCandidate(0, origin)
15+ victim.originKey = "source"
16+ id, err := ledger.commit(victim)
17+ require.NoError(t, err)
18+ active := routeTestCandidate(1, origin)
19+ active.originKey = "source"
20+ _, err = ledger.commit(active)
21+ require.NoError(t, err)
22+ sub := ledger.attentionSubscriptionFor(AttachRequest{Origin: origin, OriginKey: "source"})
23+ retired := protocol.RouteRetired{Ref: id.wire(), Target: victim.target}
24+ wrong := retired
25+ wrong.Target = active.target
26+ require.False(t, ledger.retireRoute(wrong, sub))
27+ require.False(t, ledger.retireRoute(retired, protocol.RouteAttentionSubscription{}))
28+ require.True(t, ledger.retireRoute(retired, sub))
29+ require.Empty(t, ledger.snapshot().Entries)
30+ replacement := victim
31+ replacement.target.LifecycleID[0] = 9
32+ _, err = ledger.commit(replacement)
33+ require.NoError(t, err)
34+ _, err = ledger.commit(active)
35+ require.NoError(t, err)
36+ require.False(t, ledger.retireRoute(retired, sub), "late deletion must not retire recreation")
37+ snapshot := ledger.snapshot()
38+ require.Len(t, snapshot.Entries, 1)
39+ require.Equal(t, replacement.target, snapshot.Entries[0].Target)
40+ })
41+ }
42+}
43+
44+func TestHomePickerSubscribesToServingDaemonHistory(t *testing.T) {
45+ ledger := newRouteLedger()
46+ local := routeTestCandidate(0, protocol.RouteOriginLocal)
47+ local.originKey = "local"
48+ localID, err := ledger.commit(local)
49+ require.NoError(t, err)
50+ remote := routeTestCandidate(1, protocol.RouteOriginRemote)
51+ remote.originKey = "remote"
52+ _, err = ledger.commit(remote)
53+ require.NoError(t, err)
54+
55+ // A transient home picker preserves the remote active route, but its
56+ // lifecycle observations must cover the local daemon it actually serves.
57+ subscription := ledger.attentionSubscriptionFor(AttachRequest{Origin: protocol.RouteOriginLocal, OriginKey: "local"})
58+ require.Contains(t, subscription.Targets, protocol.RouteAttentionTarget{
59+ Ref: localID.wire(), Target: local.target,
60+ })
61+}
M internal/usecase/client/routes.go +43 -11
@@ -499,30 +499,62 @@ func cloneRouteRecord(entry routeRecord) routeRecord {
499499 return entry
500500 }
501501
502-// attentionSubscription publishes only routes served by the active route's
503-// daemon. Other route origins remain private to their owning daemon.
504-func (l *routeLedger) attentionSubscription() protocol.RouteAttentionSubscription {
502+// attentionSubscriptionFor binds observation to the serving connection, which
503+// can differ from the active ledger route while a home picker is open.
504+func (l *routeLedger) attentionSubscriptionFor(request AttachRequest) protocol.RouteAttentionSubscription {
505505 l.mu.RLock()
506506 defer l.mu.RUnlock()
507+ origin := normalizeRouteOrigin(request.Origin, request.Remote)
508+ return l.attentionSubscriptionLocked(origin, normalizeRouteOriginKey(request.OriginKey, origin))
509+}
507510
508- activeIndex := l.indexByIdentityLocked(l.active)
509- if activeIndex < 0 {
510- return protocol.RouteAttentionSubscription{}
511- }
512- active := l.entries[activeIndex]
511+func (l *routeLedger) attentionSubscriptionLocked(origin protocol.RouteOrigin, originKey string) protocol.RouteAttentionSubscription {
513512 subscription := protocol.RouteAttentionSubscription{Targets: make([]protocol.RouteAttentionTarget, 0, len(l.entries))}
514513 for _, entry := range l.entries {
515- if entry.identity == active.identity || entry.origin != active.origin || entry.originKey != active.originKey {
514+ if entry.identity == l.active {
516515 continue
517516 }
517+ sourceKey := ""
518+ if entry.originKey != originKey || (entry.origin == protocol.RouteOriginLocal) != (origin == protocol.RouteOriginLocal) {
519+ if origin != protocol.RouteOriginLocal || entry.origin == protocol.RouteOriginLocal {
520+ continue
521+ }
522+ sourceKey = protocol.RemoteInventorySourceKey(entry.originKey)
523+ }
518524 subscription.Targets = append(subscription.Targets, protocol.RouteAttentionTarget{
519- Ref: entry.identity.wire(),
520- Target: entry.target,
525+ Ref: entry.identity.wire(), Target: entry.target, SourceKey: sourceKey,
521526 })
522527 }
523528 return subscription
524529 }
525530
531+func (l *routeLedger) retireRoute(retired protocol.RouteRetired, subscription protocol.RouteAttentionSubscription) bool {
532+ l.mu.Lock()
533+ defer l.mu.Unlock()
534+ identity := identityFromWire(retired.Ref)
535+ index := l.indexByIdentityLocked(identity)
536+ if index < 0 || identity == l.active || l.entries[index].target != retired.Target {
537+ return false
538+ }
539+ for _, target := range subscription.Targets {
540+ if target.Ref != retired.Ref || target.Target != retired.Target {
541+ continue
542+ }
543+ l.removeAtLocked(index)
544+ if l.previous.empty() {
545+ for _, entry := range l.entries {
546+ if entry.identity != l.active {
547+ l.previous = entry.identity
548+ break
549+ }
550+ }
551+ }
552+ l.generation++
553+ return true
554+ }
555+ return false
556+}
557+
526558 func (r routeRecord) snapshotEntry() protocol.RecentRouteEntry {
527559 return protocol.RecentRouteEntry{
528560 Key: uint64(r.identity.key),
@@ -499,30 +499,62 @@ func cloneRouteRecord(entry routeRecord) routeRecord {
499 return entry499 return entry
500 }500 }
501 501
502-// attentionSubscription publishes only routes served by the active route's502+// attentionSubscriptionFor binds observation to the serving connection, which
503-// daemon. Other route origins remain private to their owning daemon.503+// can differ from the active ledger route while a home picker is open.
504-func (l *routeLedger) attentionSubscription() protocol.RouteAttentionSubscription {504+func (l *routeLedger) attentionSubscriptionFor(request AttachRequest) protocol.RouteAttentionSubscription {
505 l.mu.RLock()505 l.mu.RLock()
506 defer l.mu.RUnlock()506 defer l.mu.RUnlock()
507+ origin := normalizeRouteOrigin(request.Origin, request.Remote)
508+ return l.attentionSubscriptionLocked(origin, normalizeRouteOriginKey(request.OriginKey, origin))
509+}
507 510
508- activeIndex := l.indexByIdentityLocked(l.active)511+func (l *routeLedger) attentionSubscriptionLocked(origin protocol.RouteOrigin, originKey string) protocol.RouteAttentionSubscription {
509- if activeIndex < 0 {
510- return protocol.RouteAttentionSubscription{}
511- }
512- active := l.entries[activeIndex]
513 subscription := protocol.RouteAttentionSubscription{Targets: make([]protocol.RouteAttentionTarget, 0, len(l.entries))}512 subscription := protocol.RouteAttentionSubscription{Targets: make([]protocol.RouteAttentionTarget, 0, len(l.entries))}
514 for _, entry := range l.entries {513 for _, entry := range l.entries {
515- if entry.identity == active.identity || entry.origin != active.origin || entry.originKey != active.originKey {514+ if entry.identity == l.active {
516 continue515 continue
517 }516 }
517+ sourceKey := ""
518+ if entry.originKey != originKey || (entry.origin == protocol.RouteOriginLocal) != (origin == protocol.RouteOriginLocal) {
519+ if origin != protocol.RouteOriginLocal || entry.origin == protocol.RouteOriginLocal {
520+ continue
521+ }
522+ sourceKey = protocol.RemoteInventorySourceKey(entry.originKey)
523+ }
518 subscription.Targets = append(subscription.Targets, protocol.RouteAttentionTarget{524 subscription.Targets = append(subscription.Targets, protocol.RouteAttentionTarget{
519- Ref: entry.identity.wire(),525+ Ref: entry.identity.wire(), Target: entry.target, SourceKey: sourceKey,
520- Target: entry.target,
521 })526 })
522 }527 }
523 return subscription528 return subscription
524 }529 }
525 530
531+func (l *routeLedger) retireRoute(retired protocol.RouteRetired, subscription protocol.RouteAttentionSubscription) bool {
532+ l.mu.Lock()
533+ defer l.mu.Unlock()
534+ identity := identityFromWire(retired.Ref)
535+ index := l.indexByIdentityLocked(identity)
536+ if index < 0 || identity == l.active || l.entries[index].target != retired.Target {
537+ return false
538+ }
539+ for _, target := range subscription.Targets {
540+ if target.Ref != retired.Ref || target.Target != retired.Target {
541+ continue
542+ }
543+ l.removeAtLocked(index)
544+ if l.previous.empty() {
545+ for _, entry := range l.entries {
546+ if entry.identity != l.active {
547+ l.previous = entry.identity
548+ break
549+ }
550+ }
551+ }
552+ l.generation++
553+ return true
554+ }
555+ return false
556+}
557+
526 func (r routeRecord) snapshotEntry() protocol.RecentRouteEntry {558 func (r routeRecord) snapshotEntry() protocol.RecentRouteEntry {
527 return protocol.RecentRouteEntry{559 return protocol.RecentRouteEntry{
528 Key: uint64(r.identity.key),560 Key: uint64(r.identity.key),
M internal/usecase/client/routes_test.go +1 -1
@@ -305,7 +305,7 @@ func TestRouteAttentionSubscriptionIncludesOnlyActiveOriginRoutes(t *testing.T)
305305 _, err = ledger.commit(active)
306306 require.NoError(t, err)
307307
308- subscription := ledger.attentionSubscription()
308+ subscription := ledger.attentionSubscriptionFor(AttachRequest{Origin: protocol.RouteOriginRemote, OriginKey: "host-a"})
309309
310310 require.Equal(t, []protocol.RouteAttentionTarget{{
311311 Ref: firstIdentity.wire(),
@@ -305,7 +305,7 @@ func TestRouteAttentionSubscriptionIncludesOnlyActiveOriginRoutes(t *testing.T)
305 _, err = ledger.commit(active)305 _, err = ledger.commit(active)
306 require.NoError(t, err)306 require.NoError(t, err)
307 307
308- subscription := ledger.attentionSubscription()308+ subscription := ledger.attentionSubscriptionFor(AttachRequest{Origin: protocol.RouteOriginRemote, OriginKey: "host-a"})
309 309
310 require.Equal(t, []protocol.RouteAttentionTarget{{310 require.Equal(t, []protocol.RouteAttentionTarget{{
311 Ref: firstIdentity.wire(),311 Ref: firstIdentity.wire(),
M internal/usecase/client/typed_connection_external_test.go +2 -0
@@ -178,6 +178,8 @@ func externalServerMessage(f wire.Frame) (protocol.ServerMessage, error) {
178178 return wire.UnmarshalCommittedRouteIdentity(f.Payload)
179179 case wire.MsgSamePeerSwitchFailure:
180180 return wire.UnmarshalSamePeerSwitchFailure(f.Payload)
181+ case wire.MsgRouteRetired:
182+ return wire.UnmarshalRouteRetired(f.Payload)
181183 case wire.MsgRoutePosition:
182184 return wire.UnmarshalRoutePosition(f.Payload)
183185 case wire.MsgRouteNavigationFailure:
@@ -178,6 +178,8 @@ func externalServerMessage(f wire.Frame) (protocol.ServerMessage, error) {
178 return wire.UnmarshalCommittedRouteIdentity(f.Payload)178 return wire.UnmarshalCommittedRouteIdentity(f.Payload)
179 case wire.MsgSamePeerSwitchFailure:179 case wire.MsgSamePeerSwitchFailure:
180 return wire.UnmarshalSamePeerSwitchFailure(f.Payload)180 return wire.UnmarshalSamePeerSwitchFailure(f.Payload)
181+ case wire.MsgRouteRetired:
182+ return wire.UnmarshalRouteRetired(f.Payload)
181 case wire.MsgRoutePosition:183 case wire.MsgRoutePosition:
182 return wire.UnmarshalRoutePosition(f.Payload)184 return wire.UnmarshalRoutePosition(f.Payload)
183 case wire.MsgRouteNavigationFailure:185 case wire.MsgRouteNavigationFailure:
M internal/usecase/client/typed_connection_test.go +2 -0
@@ -187,6 +187,8 @@ func testServerMessage(frame wire.Frame) (protocol.ServerMessage, error) {
187187 return wire.UnmarshalRouteNavigationAction(frame.Payload)
188188 case wire.MsgRouteNavigationFailure:
189189 return wire.UnmarshalRouteNavigationFailure(frame.Payload)
190+ case wire.MsgRouteRetired:
191+ return wire.UnmarshalRouteRetired(frame.Payload)
190192 case wire.MsgRoutePosition:
191193 return wire.UnmarshalRoutePosition(frame.Payload)
192194 case wire.MsgSamePeerSwitchFailure:
@@ -187,6 +187,8 @@ func testServerMessage(frame wire.Frame) (protocol.ServerMessage, error) {
187 return wire.UnmarshalRouteNavigationAction(frame.Payload)187 return wire.UnmarshalRouteNavigationAction(frame.Payload)
188 case wire.MsgRouteNavigationFailure:188 case wire.MsgRouteNavigationFailure:
189 return wire.UnmarshalRouteNavigationFailure(frame.Payload)189 return wire.UnmarshalRouteNavigationFailure(frame.Payload)
190+ case wire.MsgRouteRetired:
191+ return wire.UnmarshalRouteRetired(frame.Payload)
190 case wire.MsgRoutePosition:192 case wire.MsgRoutePosition:
191 return wire.UnmarshalRoutePosition(frame.Payload)193 return wire.UnmarshalRoutePosition(frame.Payload)
192 case wire.MsgSamePeerSwitchFailure:194 case wire.MsgSamePeerSwitchFailure:
M internal/usecase/daemon/client.go +17 -3
@@ -97,6 +97,8 @@ type attachedClient struct {
9797 parkedRouteOutput atomic.Bool
9898 parkedRouteFullPending atomic.Bool
9999 routeAttentionSubscription protocol.RouteAttentionSubscription
100+ routeSubscriptionTransport transportSnapshot
101+ routeObservationAfter map[protocol.RouteAttentionTarget]time.Time
100102 // routeCreatedSession marks a session created by this attachment's route.
101103 // A handshake that never commits Welcome must tear down that exact empty
102104 // session, while an attachment routed to an existing session must not.
@@ -371,18 +373,30 @@ func (ac *attachedClient) routeSnapshotCopy() protocol.RecentRouteSnapshot {
371373 return snapshot
372374 }
373375
374-func (ac *attachedClient) setRouteAttentionSubscription(subscription protocol.RouteAttentionSubscription) {
376+func (ac *attachedClient) setRouteAttentionSubscription(subscription protocol.RouteAttentionSubscription, expected transportSnapshot, now time.Time) {
375377 subscription.Targets = append([]protocol.RouteAttentionTarget(nil), subscription.Targets...)
376378 ac.routeMu.Lock()
379+ defer ac.routeMu.Unlock()
380+ after := make(map[protocol.RouteAttentionTarget]time.Time, len(subscription.Targets))
381+ for _, target := range subscription.Targets {
382+ admitted := now
383+ // Preserve the causal observation fence across unchanged republishes;
384+ // a new connection must establish a fresh post-Welcome subscription.
385+ if prior, ok := ac.routeObservationAfter[target]; ok && ac.routeSubscriptionTransport == expected {
386+ admitted = prior
387+ }
388+ after[target] = admitted
389+ }
390+ ac.routeObservationAfter = after
391+ ac.routeSubscriptionTransport = expected
377392 ac.routeAttentionSubscription = subscription
378- ac.routeMu.Unlock()
379393 }
380394
381395 func (ac *attachedClient) routeAttentionTarget(ref protocol.RouteRef) (protocol.ExactSessionTarget, bool) {
382396 ac.routeMu.RLock()
383397 defer ac.routeMu.RUnlock()
384398 for _, target := range ac.routeAttentionSubscription.Targets {
385- if target.Ref == ref {
399+ if target.Ref == ref && target.SourceKey == "" {
386400 return target.Target, true
387401 }
388402 }
@@ -97,6 +97,8 @@ type attachedClient struct {
97 parkedRouteOutput atomic.Bool97 parkedRouteOutput atomic.Bool
98 parkedRouteFullPending atomic.Bool98 parkedRouteFullPending atomic.Bool
99 routeAttentionSubscription protocol.RouteAttentionSubscription99 routeAttentionSubscription protocol.RouteAttentionSubscription
100+ routeSubscriptionTransport transportSnapshot
101+ routeObservationAfter map[protocol.RouteAttentionTarget]time.Time
100 // routeCreatedSession marks a session created by this attachment's route.102 // routeCreatedSession marks a session created by this attachment's route.
101 // A handshake that never commits Welcome must tear down that exact empty103 // A handshake that never commits Welcome must tear down that exact empty
102 // session, while an attachment routed to an existing session must not.104 // session, while an attachment routed to an existing session must not.
@@ -371,18 +373,30 @@ func (ac *attachedClient) routeSnapshotCopy() protocol.RecentRouteSnapshot {
371 return snapshot373 return snapshot
372 }374 }
373 375
374-func (ac *attachedClient) setRouteAttentionSubscription(subscription protocol.RouteAttentionSubscription) {376+func (ac *attachedClient) setRouteAttentionSubscription(subscription protocol.RouteAttentionSubscription, expected transportSnapshot, now time.Time) {
375 subscription.Targets = append([]protocol.RouteAttentionTarget(nil), subscription.Targets...)377 subscription.Targets = append([]protocol.RouteAttentionTarget(nil), subscription.Targets...)
376 ac.routeMu.Lock()378 ac.routeMu.Lock()
379+ defer ac.routeMu.Unlock()
380+ after := make(map[protocol.RouteAttentionTarget]time.Time, len(subscription.Targets))
381+ for _, target := range subscription.Targets {
382+ admitted := now
383+ // Preserve the causal observation fence across unchanged republishes;
384+ // a new connection must establish a fresh post-Welcome subscription.
385+ if prior, ok := ac.routeObservationAfter[target]; ok && ac.routeSubscriptionTransport == expected {
386+ admitted = prior
387+ }
388+ after[target] = admitted
389+ }
390+ ac.routeObservationAfter = after
391+ ac.routeSubscriptionTransport = expected
377 ac.routeAttentionSubscription = subscription392 ac.routeAttentionSubscription = subscription
378- ac.routeMu.Unlock()
379 }393 }
380 394
381 func (ac *attachedClient) routeAttentionTarget(ref protocol.RouteRef) (protocol.ExactSessionTarget, bool) {395 func (ac *attachedClient) routeAttentionTarget(ref protocol.RouteRef) (protocol.ExactSessionTarget, bool) {
382 ac.routeMu.RLock()396 ac.routeMu.RLock()
383 defer ac.routeMu.RUnlock()397 defer ac.routeMu.RUnlock()
384 for _, target := range ac.routeAttentionSubscription.Targets {398 for _, target := range ac.routeAttentionSubscription.Targets {
385- if target.Ref == ref {399+ if target.Ref == ref && target.SourceKey == "" {
386 return target.Target, true400 return target.Target, true
387 }401 }
388 }402 }
M internal/usecase/daemon/client_frame_routing.go +2 -1
@@ -149,7 +149,8 @@ func (d *Daemon) handleAttachmentClientMessage(capability attachmentCapability,
149149 d.detachOnAttachmentSendError(effect.capability(), effect.transport.transport)
150150 }
151151 case protocol.RouteAttentionSubscription:
152- effect.ac.setRouteAttentionSubscription(message)
152+ effect.ac.setRouteAttentionSubscription(message, effect.transport, d.clock.Now())
153+ d.reconcileRouteHistory(effect.ac, effect)
153154 d.invalidateRender(effect.sess, effect.ac, false, "client_frame_routing.go:route-attention")
154155 case protocol.SessionCreationFailure:
155156 notice := "couldn't create session"
@@ -149,7 +149,8 @@ func (d *Daemon) handleAttachmentClientMessage(capability attachmentCapability,
149 d.detachOnAttachmentSendError(effect.capability(), effect.transport.transport)149 d.detachOnAttachmentSendError(effect.capability(), effect.transport.transport)
150 }150 }
151 case protocol.RouteAttentionSubscription:151 case protocol.RouteAttentionSubscription:
152- effect.ac.setRouteAttentionSubscription(message)152+ effect.ac.setRouteAttentionSubscription(message, effect.transport, d.clock.Now())
153+ d.reconcileRouteHistory(effect.ac, effect)
153 d.invalidateRender(effect.sess, effect.ac, false, "client_frame_routing.go:route-attention")154 d.invalidateRender(effect.sess, effect.ac, false, "client_frame_routing.go:route-attention")
154 case protocol.SessionCreationFailure:155 case protocol.SessionCreationFailure:
155 notice := "couldn't create session"156 notice := "couldn't create session"
M internal/usecase/daemon/navigation_inventory.go +1 -3
@@ -1,7 +1,6 @@
11 package daemon
22
33 import (
4- "crypto/sha256"
54 "encoding/hex"
65
76 "github.com/bnema/vev/internal/domain"
@@ -149,8 +148,7 @@ func remoteInventoryGroup(host ports.RemoteHostSnapshot) protocol.NavigationInve
149148 // configured remote endpoint. The hash domain-separates inventory keys so
150149 // relayed identifiers disclose nothing about local SSH configuration.
151150 func remoteInventorySourceKey(endpoint string) string {
152- sum := sha256.Sum256([]byte("vev-inventory-remote\x00" + endpoint))
153- return protocol.NavigationInventoryRemoteSourcePrefix + hex.EncodeToString(sum[:12])
151+ return protocol.RemoteInventorySourceKey(endpoint)
154152 }
155153
156154 // navigationInventoryEncodedBudget mirrors the wire export budget without
@@ -1,7 +1,6 @@
1 package daemon1 package daemon
2 2
3 import (3 import (
4- "crypto/sha256"
5 "encoding/hex"4 "encoding/hex"
6 5
7 "github.com/bnema/vev/internal/domain"6 "github.com/bnema/vev/internal/domain"
@@ -149,8 +148,7 @@ func remoteInventoryGroup(host ports.RemoteHostSnapshot) protocol.NavigationInve
149 // configured remote endpoint. The hash domain-separates inventory keys so148 // configured remote endpoint. The hash domain-separates inventory keys so
150 // relayed identifiers disclose nothing about local SSH configuration.149 // relayed identifiers disclose nothing about local SSH configuration.
151 func remoteInventorySourceKey(endpoint string) string {150 func remoteInventorySourceKey(endpoint string) string {
152- sum := sha256.Sum256([]byte("vev-inventory-remote\x00" + endpoint))151+ return protocol.RemoteInventorySourceKey(endpoint)
153- return protocol.NavigationInventoryRemoteSourcePrefix + hex.EncodeToString(sum[:12])
154 }152 }
155 153
156 // navigationInventoryEncodedBudget mirrors the wire export budget without154 // navigationInventoryEncodedBudget mirrors the wire export budget without
M internal/usecase/daemon/recent_route_snapshot_test.go +1 -1
@@ -53,7 +53,7 @@ func TestAttachmentStatusResolvesSubscribedRouteAttention(t *testing.T) {
5353 LifecycleID: tt.targetID,
5454 SessionName: sess.name,
5555 },
56- }}})
56+ }}}, ac.transportSnapshot(), d.clock.Now())
5757
5858 state := d.barStateForAttachmentPaletteHintsFor(sess, ac, "", nil, protocol.RecentRouteSnapshot{})
5959
@@ -53,7 +53,7 @@ func TestAttachmentStatusResolvesSubscribedRouteAttention(t *testing.T) {
53 LifecycleID: tt.targetID,53 LifecycleID: tt.targetID,
54 SessionName: sess.name,54 SessionName: sess.name,
55 },55 },
56- }}})56+ }}}, ac.transportSnapshot(), d.clock.Now())
57 57
58 state := d.barStateForAttachmentPaletteHintsFor(sess, ac, "", nil, protocol.RecentRouteSnapshot{})58 state := d.barStateForAttachmentPaletteHintsFor(sess, ac, "", nil, protocol.RecentRouteSnapshot{})
59 59
M internal/usecase/daemon/remote_directory_view.go +1 -0
@@ -66,6 +66,7 @@ func (d *Daemon) refreshRemoteDirectoryViews() {
6666 continue
6767 }
6868 for _, ac := range sess.snapshotAttachments() {
69+ d.reconcileRouteHistory(ac)
6970 d.refreshRemoteDirectoryViewsFor(ac)
7071 }
7172 }
@@ -66,6 +66,7 @@ func (d *Daemon) refreshRemoteDirectoryViews() {
66 continue66 continue
67 }67 }
68 for _, ac := range sess.snapshotAttachments() {68 for _, ac := range sess.snapshotAttachments() {
69+ d.reconcileRouteHistory(ac)
69 d.refreshRemoteDirectoryViewsFor(ac)70 d.refreshRemoteDirectoryViewsFor(ac)
70 }71 }
71 }72 }
A internal/usecase/daemon/route_retirement.go +105 -0
@@ -0,0 +1,105 @@
1+package daemon
2+
3+import (
4+ "errors"
5+ "maps"
6+ "time"
7+
8+ "github.com/bnema/vev/internal/domain"
9+ "github.com/bnema/vev/internal/protocol"
10+)
11+
12+// retiredRoutes uses only authoritative observations. An unavailable remote
13+// source retains its history; a successful empty catalogue proves absence.
14+func retiredRoutes(subscription protocol.RouteAttentionSubscription, after map[protocol.RouteAttentionTarget]time.Time, inv sessionInventory) []protocol.RouteRetired {
15+ var retired []protocol.RouteRetired
16+ for _, target := range subscription.Targets {
17+ known, present := target.SourceKey == "", false
18+ if known {
19+ for _, item := range inv.live {
20+ present = present || item.view.incarnation == target.Target.LifecycleID
21+ }
22+ for _, stopped := range inv.stopped {
23+ present = present || stopped.incarnation == target.Target.LifecycleID
24+ }
25+ } else {
26+ for _, host := range inv.hosts {
27+ if protocol.RemoteInventorySourceKey(host.Endpoint) != target.SourceKey {
28+ continue
29+ }
30+ // A cached success (or a fetch already running at admission)
31+ // may predate creation of this lifecycle. Only a later-started,
32+ // completed successful observation can establish absence.
33+ admitted, fenced := after[target]
34+ known = fenced && host.InventoryKnown && host.Availability == domain.RemoteAvailabilityReachable &&
35+ !host.Checking && host.LastAttempt.After(admitted) && !host.LastSuccess.Before(host.LastAttempt)
36+ for _, session := range host.Sessions {
37+ present = present || session.LifecycleID == target.Target.LifecycleID
38+ }
39+ break
40+ }
41+ }
42+ if known && !present {
43+ retired = append(retired, protocol.RouteRetired{Ref: target.Ref, Target: target.Target})
44+ }
45+ }
46+ return retired
47+}
48+
49+func (d *Daemon) reconcileRouteHistory(ac *attachedClient, admitted ...*attachmentEffect) {
50+ if ac == nil {
51+ return
52+ }
53+ ac.routeMu.RLock()
54+ expected := ac.routeSubscriptionTransport
55+ subscription := ac.routeAttentionSubscription
56+ subscription.Targets = append([]protocol.RouteAttentionTarget(nil), subscription.Targets...)
57+ after := maps.Clone(ac.routeObservationAfter)
58+ ac.routeMu.RUnlock()
59+ if expected.transport == nil || len(subscription.Targets) == 0 || !ac.transportSnapshotCurrent(expected) {
60+ return
61+ }
62+ var effect *attachmentEffect
63+ if len(admitted) != 0 {
64+ effect = admitted[0]
65+ } else {
66+ _, token, ok := d.currentAttachmentConnection(ac, expected.transport)
67+ if !ok {
68+ return
69+ }
70+ var okEffect bool
71+ effect, okEffect = ac.beginAttachmentEffect(token)
72+ if !okEffect {
73+ return
74+ }
75+ defer effect.End()
76+ }
77+ if effect == nil || effect.transport != expected || !effect.current() {
78+ return
79+ }
80+ retired := retiredRoutes(subscription, after, d.captureSessionInventory(viewOptions{}, false))
81+ for _, message := range retired {
82+ // Bind to the connection that supplied this subscription; a resumed
83+ // attachment must not receive observations from its old transport.
84+ _, err := d.boundedSendWith(expected.transport, func() error {
85+ return effect.sendControl(message)
86+ })
87+ if err != nil {
88+ if errors.Is(err, errSendTimedOut) {
89+ _ = ac.closeCapturedTransport(expected.transport)
90+ }
91+ return
92+ }
93+ }
94+}
95+
96+func (d *Daemon) reconcileAllRouteHistories() {
97+ d.mu.Lock()
98+ sessions := d.sessionsSnapshotLocked()
99+ d.mu.Unlock()
100+ for _, sess := range sessions {
101+ for _, ac := range sess.snapshotAttachments() {
102+ d.reconcileRouteHistory(ac)
103+ }
104+ }
105+}
@@ -0,0 +1,105 @@
1+package daemon
2+
3+import (
4+ "errors"
5+ "maps"
6+ "time"
7+
8+ "github.com/bnema/vev/internal/domain"
9+ "github.com/bnema/vev/internal/protocol"
10+)
11+
12+// retiredRoutes uses only authoritative observations. An unavailable remote
13+// source retains its history; a successful empty catalogue proves absence.
14+func retiredRoutes(subscription protocol.RouteAttentionSubscription, after map[protocol.RouteAttentionTarget]time.Time, inv sessionInventory) []protocol.RouteRetired {
15+ var retired []protocol.RouteRetired
16+ for _, target := range subscription.Targets {
17+ known, present := target.SourceKey == "", false
18+ if known {
19+ for _, item := range inv.live {
20+ present = present || item.view.incarnation == target.Target.LifecycleID
21+ }
22+ for _, stopped := range inv.stopped {
23+ present = present || stopped.incarnation == target.Target.LifecycleID
24+ }
25+ } else {
26+ for _, host := range inv.hosts {
27+ if protocol.RemoteInventorySourceKey(host.Endpoint) != target.SourceKey {
28+ continue
29+ }
30+ // A cached success (or a fetch already running at admission)
31+ // may predate creation of this lifecycle. Only a later-started,
32+ // completed successful observation can establish absence.
33+ admitted, fenced := after[target]
34+ known = fenced && host.InventoryKnown && host.Availability == domain.RemoteAvailabilityReachable &&
35+ !host.Checking && host.LastAttempt.After(admitted) && !host.LastSuccess.Before(host.LastAttempt)
36+ for _, session := range host.Sessions {
37+ present = present || session.LifecycleID == target.Target.LifecycleID
38+ }
39+ break
40+ }
41+ }
42+ if known && !present {
43+ retired = append(retired, protocol.RouteRetired{Ref: target.Ref, Target: target.Target})
44+ }
45+ }
46+ return retired
47+}
48+
49+func (d *Daemon) reconcileRouteHistory(ac *attachedClient, admitted ...*attachmentEffect) {
50+ if ac == nil {
51+ return
52+ }
53+ ac.routeMu.RLock()
54+ expected := ac.routeSubscriptionTransport
55+ subscription := ac.routeAttentionSubscription
56+ subscription.Targets = append([]protocol.RouteAttentionTarget(nil), subscription.Targets...)
57+ after := maps.Clone(ac.routeObservationAfter)
58+ ac.routeMu.RUnlock()
59+ if expected.transport == nil || len(subscription.Targets) == 0 || !ac.transportSnapshotCurrent(expected) {
60+ return
61+ }
62+ var effect *attachmentEffect
63+ if len(admitted) != 0 {
64+ effect = admitted[0]
65+ } else {
66+ _, token, ok := d.currentAttachmentConnection(ac, expected.transport)
67+ if !ok {
68+ return
69+ }
70+ var okEffect bool
71+ effect, okEffect = ac.beginAttachmentEffect(token)
72+ if !okEffect {
73+ return
74+ }
75+ defer effect.End()
76+ }
77+ if effect == nil || effect.transport != expected || !effect.current() {
78+ return
79+ }
80+ retired := retiredRoutes(subscription, after, d.captureSessionInventory(viewOptions{}, false))
81+ for _, message := range retired {
82+ // Bind to the connection that supplied this subscription; a resumed
83+ // attachment must not receive observations from its old transport.
84+ _, err := d.boundedSendWith(expected.transport, func() error {
85+ return effect.sendControl(message)
86+ })
87+ if err != nil {
88+ if errors.Is(err, errSendTimedOut) {
89+ _ = ac.closeCapturedTransport(expected.transport)
90+ }
91+ return
92+ }
93+ }
94+}
95+
96+func (d *Daemon) reconcileAllRouteHistories() {
97+ d.mu.Lock()
98+ sessions := d.sessionsSnapshotLocked()
99+ d.mu.Unlock()
100+ for _, sess := range sessions {
101+ for _, ac := range sess.snapshotAttachments() {
102+ d.reconcileRouteHistory(ac)
103+ }
104+ }
105+}
A internal/usecase/daemon/route_retirement_fence_test.go +68 -0
@@ -0,0 +1,68 @@
1+package daemon
2+
3+import (
4+ "testing"
5+ "time"
6+
7+ "github.com/bnema/vev/internal/domain"
8+ "github.com/bnema/vev/internal/ports"
9+ "github.com/bnema/vev/internal/protocol"
10+ "github.com/stretchr/testify/require"
11+)
12+
13+func TestRouteRetirementRequiresFetchStartedAfterSubscription(t *testing.T) {
14+ admitted := time.Unix(10, 0)
15+ target := protocol.RouteAttentionTarget{Ref: protocol.RouteRef{Key: 1, Generation: 1}, Target: protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "new"}, SourceKey: protocol.RemoteInventorySourceKey("remote")}
16+ sub := protocol.RouteAttentionSubscription{Targets: []protocol.RouteAttentionTarget{target}}
17+ for _, tc := range []struct {
18+ name string
19+ start, finish int64
20+ checking, want bool
21+ }{
22+ {name: "old successful catalogue", start: 8, finish: 9},
23+ {name: "old fetch completes after admission", start: 9, finish: 11},
24+ {name: "new fetch still running", start: 11, finish: 9, checking: true},
25+ {name: "new successful absence", start: 11, finish: 12, want: true},
26+ } {
27+ t.Run(tc.name, func(t *testing.T) {
28+ inv := sessionInventory{hosts: []ports.RemoteHostSnapshot{{Endpoint: "remote", InventoryKnown: true, Availability: domain.RemoteAvailabilityReachable, LastAttempt: time.Unix(tc.start, 0), LastSuccess: time.Unix(tc.finish, 0), Checking: tc.checking}}}
29+ after := map[protocol.RouteAttentionTarget]time.Time{target: admitted}
30+ require.Equal(t, tc.want, len(retiredRoutes(sub, after, inv)) == 1)
31+ })
32+ }
33+ _, _, ac, _ := newManualSessionWithPTYs(t, nil)
34+ ac.setRouteAttentionSubscription(sub, ac.transportSnapshot(), admitted)
35+ ac.setRouteAttentionSubscription(sub, ac.transportSnapshot(), admitted.Add(time.Second))
36+ require.Equal(t, admitted, ac.routeObservationAfter[target], "unchanged publications must preserve the freshness fence")
37+}
38+
39+func TestRouteRetirementDoesNotUseSubscriptionFromPreviousTransport(t *testing.T) {
40+ d, sess, ac, _ := newManualSessionWithPTYs(t, nil)
41+ ac.installTestAttachmentCapability(sess.captureAttachmentCapability(ac, ac.transport()))
42+ sub := protocol.RouteAttentionSubscription{Targets: []protocol.RouteAttentionTarget{{Ref: protocol.RouteRef{Key: 1, Generation: 1}, Target: protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{9}, SessionName: "deleted"}}}}
43+ ac.setRouteAttentionSubscription(sub, ac.transportSnapshot(), d.clock.Now())
44+ // Resume installs a transport while retaining the old subscription, before
45+ // Welcome. Reconciliation must emit nothing on the replacement connection.
46+ replacement := &closeTrackingTransport{}
47+ ac.replaceTransport(replacement)
48+ ac.installTestAttachmentCapability(sess.captureAttachmentCapability(ac, ac.transport()))
49+ d.reconcileRouteHistory(ac)
50+ require.Empty(t, replacement.Sends())
51+ ac.setRouteAttentionSubscription(sub, ac.transportSnapshot(), d.clock.Now())
52+ d.reconcileRouteHistory(ac)
53+ require.Len(t, replacement.Sends(), 1, "fresh post-Welcome subscription enables retirement")
54+}
55+
56+func TestRouteRetirementDoesNotSendThroughFrozenAttachment(t *testing.T) {
57+ d, sess, ac, sends := newManualSessionWithPTYs(t, nil)
58+ ac.installTestAttachmentCapability(sess.captureAttachmentCapability(ac, ac.transport()))
59+ ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{Targets: []protocol.RouteAttentionTarget{{Ref: protocol.RouteRef{Key: 1, Generation: 1}, Target: protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{9}, SessionName: "deleted"}}}}, ac.transportSnapshot(), d.clock.Now())
60+ frozen := freezeAttachmentEffectGates(ac)
61+ defer frozen.unfreeze()
62+ d.reconcileRouteHistory(ac)
63+ select {
64+ case message := <-sends:
65+ t.Fatalf("unexpected retirement during transition: %v", message.Type)
66+ default:
67+ }
68+}
@@ -0,0 +1,68 @@
1+package daemon
2+
3+import (
4+ "testing"
5+ "time"
6+
7+ "github.com/bnema/vev/internal/domain"
8+ "github.com/bnema/vev/internal/ports"
9+ "github.com/bnema/vev/internal/protocol"
10+ "github.com/stretchr/testify/require"
11+)
12+
13+func TestRouteRetirementRequiresFetchStartedAfterSubscription(t *testing.T) {
14+ admitted := time.Unix(10, 0)
15+ target := protocol.RouteAttentionTarget{Ref: protocol.RouteRef{Key: 1, Generation: 1}, Target: protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "new"}, SourceKey: protocol.RemoteInventorySourceKey("remote")}
16+ sub := protocol.RouteAttentionSubscription{Targets: []protocol.RouteAttentionTarget{target}}
17+ for _, tc := range []struct {
18+ name string
19+ start, finish int64
20+ checking, want bool
21+ }{
22+ {name: "old successful catalogue", start: 8, finish: 9},
23+ {name: "old fetch completes after admission", start: 9, finish: 11},
24+ {name: "new fetch still running", start: 11, finish: 9, checking: true},
25+ {name: "new successful absence", start: 11, finish: 12, want: true},
26+ } {
27+ t.Run(tc.name, func(t *testing.T) {
28+ inv := sessionInventory{hosts: []ports.RemoteHostSnapshot{{Endpoint: "remote", InventoryKnown: true, Availability: domain.RemoteAvailabilityReachable, LastAttempt: time.Unix(tc.start, 0), LastSuccess: time.Unix(tc.finish, 0), Checking: tc.checking}}}
29+ after := map[protocol.RouteAttentionTarget]time.Time{target: admitted}
30+ require.Equal(t, tc.want, len(retiredRoutes(sub, after, inv)) == 1)
31+ })
32+ }
33+ _, _, ac, _ := newManualSessionWithPTYs(t, nil)
34+ ac.setRouteAttentionSubscription(sub, ac.transportSnapshot(), admitted)
35+ ac.setRouteAttentionSubscription(sub, ac.transportSnapshot(), admitted.Add(time.Second))
36+ require.Equal(t, admitted, ac.routeObservationAfter[target], "unchanged publications must preserve the freshness fence")
37+}
38+
39+func TestRouteRetirementDoesNotUseSubscriptionFromPreviousTransport(t *testing.T) {
40+ d, sess, ac, _ := newManualSessionWithPTYs(t, nil)
41+ ac.installTestAttachmentCapability(sess.captureAttachmentCapability(ac, ac.transport()))
42+ sub := protocol.RouteAttentionSubscription{Targets: []protocol.RouteAttentionTarget{{Ref: protocol.RouteRef{Key: 1, Generation: 1}, Target: protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{9}, SessionName: "deleted"}}}}
43+ ac.setRouteAttentionSubscription(sub, ac.transportSnapshot(), d.clock.Now())
44+ // Resume installs a transport while retaining the old subscription, before
45+ // Welcome. Reconciliation must emit nothing on the replacement connection.
46+ replacement := &closeTrackingTransport{}
47+ ac.replaceTransport(replacement)
48+ ac.installTestAttachmentCapability(sess.captureAttachmentCapability(ac, ac.transport()))
49+ d.reconcileRouteHistory(ac)
50+ require.Empty(t, replacement.Sends())
51+ ac.setRouteAttentionSubscription(sub, ac.transportSnapshot(), d.clock.Now())
52+ d.reconcileRouteHistory(ac)
53+ require.Len(t, replacement.Sends(), 1, "fresh post-Welcome subscription enables retirement")
54+}
55+
56+func TestRouteRetirementDoesNotSendThroughFrozenAttachment(t *testing.T) {
57+ d, sess, ac, sends := newManualSessionWithPTYs(t, nil)
58+ ac.installTestAttachmentCapability(sess.captureAttachmentCapability(ac, ac.transport()))
59+ ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{Targets: []protocol.RouteAttentionTarget{{Ref: protocol.RouteRef{Key: 1, Generation: 1}, Target: protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{9}, SessionName: "deleted"}}}}, ac.transportSnapshot(), d.clock.Now())
60+ frozen := freezeAttachmentEffectGates(ac)
61+ defer frozen.unfreeze()
62+ d.reconcileRouteHistory(ac)
63+ select {
64+ case message := <-sends:
65+ t.Fatalf("unexpected retirement during transition: %v", message.Type)
66+ default:
67+ }
68+}
A internal/usecase/daemon/route_retirement_test.go +99 -0
@@ -0,0 +1,99 @@
1+package daemon
2+
3+import (
4+ "testing"
5+ "time"
6+
7+ "github.com/bnema/vev/internal/domain"
8+ "github.com/bnema/vev/internal/ports"
9+ "github.com/bnema/vev/internal/protocol"
10+ "github.com/bnema/vev/internal/protocol/catalogue"
11+ "github.com/bnema/vev/internal/protocol/wire"
12+ "github.com/bnema/vev/internal/usecase/picker"
13+ "github.com/stretchr/testify/require"
14+)
15+
16+func TestPickerDeleteRetiresSubscribedHistory(t *testing.T) {
17+ for _, stopped := range []bool{false, true} {
18+ t.Run(map[bool]string{false: "live", true: "stopped"}[stopped], func(t *testing.T) {
19+ d, source, ac, sends := newManualSessionWithPTYs(t, nil)
20+ victim := &session{sessionCore: sessionCore{id: "victim", name: "victim", incarnation: domain.SessionLifecycleID{2}}, ctx: source.ctx, cancel: func() {}}
21+ d.mu.Lock()
22+ d.sessions[victim.id] = victim
23+ d.mu.Unlock()
24+ target := protocol.ExactSessionTarget{LifecycleID: victim.incarnation, SessionName: victim.name}
25+ ref := protocol.RouteRef{Key: 2, Generation: 1}
26+ ac.installTestAttachmentCapability(source.captureAttachmentCapability(ac, ac.transport()))
27+ ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{Targets: []protocol.RouteAttentionTarget{{Ref: ref, Target: target}}}, ac.transportSnapshot(), d.clock.Now())
28+ if stopped {
29+ require.NoError(t, d.killSession(victim, protocol.ReasonSessionKilled, false))
30+ }
31+ require.NoError(t, d.killPickerTarget(picker.Target{Session: victim.id, Name: victim.name, Incarnation: victim.incarnation, Stopped: stopped}))
32+ frame := awaitFrame(t, sends, wire.MsgRouteRetired)
33+ message, err := wire.UnmarshalRouteRetired(frame.Payload)
34+ require.NoError(t, err)
35+ require.Equal(t, protocol.RouteRetired{Ref: ref, Target: target}, message)
36+ require.Equal(t, source, ac.currentAttachmentSession())
37+ })
38+ }
39+}
40+
41+func TestRemoteDirectoryDeletionRetiresHistoryWithPickerClosed(t *testing.T) {
42+ d, source, ac, sends := newManualSessionWithPTYs(t, nil)
43+ ac.installTestAttachmentCapability(source.captureAttachmentCapability(ac, ac.transport()))
44+ target := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{9}, SessionName: "remote-work"}
45+ ref := protocol.RouteRef{Key: 3, Generation: 2}
46+ ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{Targets: []protocol.RouteAttentionTarget{{Ref: ref, Target: target, SourceKey: protocol.RemoteInventorySourceKey("remote")}}}, ac.transportSnapshot(), d.clock.Now())
47+ seedRemoteDirectory(t, d, ports.RemoteHostSnapshot{Endpoint: "remote", InventoryKnown: true, Availability: domain.RemoteAvailabilityReachable, LastAttempt: d.clock.Now().Add(time.Second), LastSuccess: d.clock.Now().Add(2 * time.Second)})
48+ require.False(t, ac.overlays.pickerActive())
49+ d.refreshRemoteDirectoryViews()
50+ frame := awaitFrame(t, sends, wire.MsgRouteRetired)
51+ message, err := wire.UnmarshalRouteRetired(frame.Payload)
52+ require.NoError(t, err)
53+ require.Equal(t, protocol.RouteRetired{Ref: ref, Target: target}, message)
54+}
55+
56+func TestRetiredRoutesRequiresAuthoritativeAbsence(t *testing.T) {
57+ target := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "work"}
58+ for _, tc := range []struct {
59+ name string
60+ remote bool
61+ known bool
62+ unreachable bool
63+ present bool
64+ replacement bool
65+ want bool
66+ }{
67+ {name: "local deleted", want: true},
68+ {name: "local present", present: true},
69+ {name: "local recreated", replacement: true, want: true},
70+ {name: "remote deleted", remote: true, known: true, want: true},
71+ {name: "remote present", remote: true, known: true, present: true},
72+ {name: "remote recreated", remote: true, known: true, replacement: true, want: true},
73+ {name: "remote unknown", remote: true},
74+ {name: "remote unreachable", remote: true, known: true, unreachable: true},
75+ } {
76+ t.Run(tc.name, func(t *testing.T) {
77+ sub := protocol.RouteAttentionSubscription{Targets: []protocol.RouteAttentionTarget{{Ref: protocol.RouteRef{Key: 1, Generation: 1}, Target: target}}}
78+ inv := sessionInventory{}
79+ lifecycle := target.LifecycleID
80+ if tc.replacement {
81+ lifecycle = domain.SessionLifecycleID{2}
82+ }
83+ if tc.remote {
84+ sub.Targets[0].SourceKey = protocol.RemoteInventorySourceKey("remote")
85+ host := ports.RemoteHostSnapshot{Endpoint: "remote", InventoryKnown: tc.known, Availability: domain.RemoteAvailabilityReachable, LastAttempt: time.Unix(2, 0), LastSuccess: time.Unix(3, 0)}
86+ if tc.unreachable {
87+ host.Availability = domain.RemoteAvailabilityUnreachable
88+ }
89+ if tc.present || tc.replacement {
90+ host.Sessions = []catalogue.RemoteCatalogSession{{Name: "work", LifecycleID: lifecycle}}
91+ }
92+ inv.hosts = []ports.RemoteHostSnapshot{host}
93+ } else if tc.present || tc.replacement {
94+ inv.live = []inventoriedLiveSession{{view: sessionView{name: "work", incarnation: lifecycle}}}
95+ }
96+ require.Equal(t, tc.want, len(retiredRoutes(sub, map[protocol.RouteAttentionTarget]time.Time{sub.Targets[0]: time.Unix(1, 0)}, inv)) == 1)
97+ })
98+ }
99+}
@@ -0,0 +1,99 @@
1+package daemon
2+
3+import (
4+ "testing"
5+ "time"
6+
7+ "github.com/bnema/vev/internal/domain"
8+ "github.com/bnema/vev/internal/ports"
9+ "github.com/bnema/vev/internal/protocol"
10+ "github.com/bnema/vev/internal/protocol/catalogue"
11+ "github.com/bnema/vev/internal/protocol/wire"
12+ "github.com/bnema/vev/internal/usecase/picker"
13+ "github.com/stretchr/testify/require"
14+)
15+
16+func TestPickerDeleteRetiresSubscribedHistory(t *testing.T) {
17+ for _, stopped := range []bool{false, true} {
18+ t.Run(map[bool]string{false: "live", true: "stopped"}[stopped], func(t *testing.T) {
19+ d, source, ac, sends := newManualSessionWithPTYs(t, nil)
20+ victim := &session{sessionCore: sessionCore{id: "victim", name: "victim", incarnation: domain.SessionLifecycleID{2}}, ctx: source.ctx, cancel: func() {}}
21+ d.mu.Lock()
22+ d.sessions[victim.id] = victim
23+ d.mu.Unlock()
24+ target := protocol.ExactSessionTarget{LifecycleID: victim.incarnation, SessionName: victim.name}
25+ ref := protocol.RouteRef{Key: 2, Generation: 1}
26+ ac.installTestAttachmentCapability(source.captureAttachmentCapability(ac, ac.transport()))
27+ ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{Targets: []protocol.RouteAttentionTarget{{Ref: ref, Target: target}}}, ac.transportSnapshot(), d.clock.Now())
28+ if stopped {
29+ require.NoError(t, d.killSession(victim, protocol.ReasonSessionKilled, false))
30+ }
31+ require.NoError(t, d.killPickerTarget(picker.Target{Session: victim.id, Name: victim.name, Incarnation: victim.incarnation, Stopped: stopped}))
32+ frame := awaitFrame(t, sends, wire.MsgRouteRetired)
33+ message, err := wire.UnmarshalRouteRetired(frame.Payload)
34+ require.NoError(t, err)
35+ require.Equal(t, protocol.RouteRetired{Ref: ref, Target: target}, message)
36+ require.Equal(t, source, ac.currentAttachmentSession())
37+ })
38+ }
39+}
40+
41+func TestRemoteDirectoryDeletionRetiresHistoryWithPickerClosed(t *testing.T) {
42+ d, source, ac, sends := newManualSessionWithPTYs(t, nil)
43+ ac.installTestAttachmentCapability(source.captureAttachmentCapability(ac, ac.transport()))
44+ target := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{9}, SessionName: "remote-work"}
45+ ref := protocol.RouteRef{Key: 3, Generation: 2}
46+ ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{Targets: []protocol.RouteAttentionTarget{{Ref: ref, Target: target, SourceKey: protocol.RemoteInventorySourceKey("remote")}}}, ac.transportSnapshot(), d.clock.Now())
47+ seedRemoteDirectory(t, d, ports.RemoteHostSnapshot{Endpoint: "remote", InventoryKnown: true, Availability: domain.RemoteAvailabilityReachable, LastAttempt: d.clock.Now().Add(time.Second), LastSuccess: d.clock.Now().Add(2 * time.Second)})
48+ require.False(t, ac.overlays.pickerActive())
49+ d.refreshRemoteDirectoryViews()
50+ frame := awaitFrame(t, sends, wire.MsgRouteRetired)
51+ message, err := wire.UnmarshalRouteRetired(frame.Payload)
52+ require.NoError(t, err)
53+ require.Equal(t, protocol.RouteRetired{Ref: ref, Target: target}, message)
54+}
55+
56+func TestRetiredRoutesRequiresAuthoritativeAbsence(t *testing.T) {
57+ target := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "work"}
58+ for _, tc := range []struct {
59+ name string
60+ remote bool
61+ known bool
62+ unreachable bool
63+ present bool
64+ replacement bool
65+ want bool
66+ }{
67+ {name: "local deleted", want: true},
68+ {name: "local present", present: true},
69+ {name: "local recreated", replacement: true, want: true},
70+ {name: "remote deleted", remote: true, known: true, want: true},
71+ {name: "remote present", remote: true, known: true, present: true},
72+ {name: "remote recreated", remote: true, known: true, replacement: true, want: true},
73+ {name: "remote unknown", remote: true},
74+ {name: "remote unreachable", remote: true, known: true, unreachable: true},
75+ } {
76+ t.Run(tc.name, func(t *testing.T) {
77+ sub := protocol.RouteAttentionSubscription{Targets: []protocol.RouteAttentionTarget{{Ref: protocol.RouteRef{Key: 1, Generation: 1}, Target: target}}}
78+ inv := sessionInventory{}
79+ lifecycle := target.LifecycleID
80+ if tc.replacement {
81+ lifecycle = domain.SessionLifecycleID{2}
82+ }
83+ if tc.remote {
84+ sub.Targets[0].SourceKey = protocol.RemoteInventorySourceKey("remote")
85+ host := ports.RemoteHostSnapshot{Endpoint: "remote", InventoryKnown: tc.known, Availability: domain.RemoteAvailabilityReachable, LastAttempt: time.Unix(2, 0), LastSuccess: time.Unix(3, 0)}
86+ if tc.unreachable {
87+ host.Availability = domain.RemoteAvailabilityUnreachable
88+ }
89+ if tc.present || tc.replacement {
90+ host.Sessions = []catalogue.RemoteCatalogSession{{Name: "work", LifecycleID: lifecycle}}
91+ }
92+ inv.hosts = []ports.RemoteHostSnapshot{host}
93+ } else if tc.present || tc.replacement {
94+ inv.live = []inventoriedLiveSession{{view: sessionView{name: "work", incarnation: lifecycle}}}
95+ }
96+ require.Equal(t, tc.want, len(retiredRoutes(sub, map[protocol.RouteAttentionTarget]time.Time{sub.Targets[0]: time.Unix(1, 0)}, inv)) == 1)
97+ })
98+ }
99+}
M internal/usecase/daemon/session.go +4 -0
@@ -1301,6 +1301,7 @@ func (d *Daemon) retryStoppedPurgeContextExact(ctx context.Context, name string,
13011301 delete(d.inactive, name)
13021302 }
13031303 d.mu.Unlock()
1304+ d.reconcileAllRouteHistories()
13041305 return nil
13051306 }
13061307
@@ -1854,6 +1855,9 @@ func (d *Daemon) killSessionWithSnapshotDeadlineAndCondition(sess *session, reas
18541855 d.mu.Unlock()
18551856 }
18561857 }
1858+ if purge && purgeErr == nil {
1859+ d.reconcileAllRouteHistories()
1860+ }
18571861 if empty {
18581862 d.doneOnce.Do(func() { close(d.done) })
18591863 } else if ringing {
@@ -1301,6 +1301,7 @@ func (d *Daemon) retryStoppedPurgeContextExact(ctx context.Context, name string,
1301 delete(d.inactive, name)1301 delete(d.inactive, name)
1302 }1302 }
1303 d.mu.Unlock()1303 d.mu.Unlock()
1304+ d.reconcileAllRouteHistories()
1304 return nil1305 return nil
1305 }1306 }
1306 1307
@@ -1854,6 +1855,9 @@ func (d *Daemon) killSessionWithSnapshotDeadlineAndCondition(sess *session, reas
1854 d.mu.Unlock()1855 d.mu.Unlock()
1855 }1856 }
1856 }1857 }
1858+ if purge && purgeErr == nil {
1859+ d.reconcileAllRouteHistories()
1860+ }
1857 if empty {1861 if empty {
1858 d.doneOnce.Do(func() { close(d.done) })1862 d.doneOnce.Do(func() { close(d.done) })
1859 } else if ringing {1863 } else if ringing {
M internal/usecase/daemon/session_back_route_test.go +2 -2
@@ -29,7 +29,7 @@ func TestBackSessionUsesClientPreviousRouteAfterSnapshotPublication(t *testing.T
2929 })
3030 ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{
3131 Targets: []protocol.RouteAttentionTarget{{Ref: previous, Target: testRouteTarget("previous", 7)}},
32- })
32+ }, ac.transportSnapshot(), d.clock.Now())
3333
3434 require.NoError(t, d.backSessionForAttachment(effect))
3535 frames := transport.Sends()
@@ -83,7 +83,7 @@ func TestBackSessionOffersPreviousRouteOnCurrentDaemon(t *testing.T) {
8383 })
8484 ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{
8585 Targets: []protocol.RouteAttentionTarget{{Ref: previous, Target: exact}},
86- })
86+ }, ac.transportSnapshot(), d.clock.Now())
8787
8888 require.NoError(t, d.backSessionForAttachment(effect))
8989 frames := transport.Sends()
@@ -29,7 +29,7 @@ func TestBackSessionUsesClientPreviousRouteAfterSnapshotPublication(t *testing.T
29 })29 })
30 ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{30 ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{
31 Targets: []protocol.RouteAttentionTarget{{Ref: previous, Target: testRouteTarget("previous", 7)}},31 Targets: []protocol.RouteAttentionTarget{{Ref: previous, Target: testRouteTarget("previous", 7)}},
32- })32+ }, ac.transportSnapshot(), d.clock.Now())
33 33
34 require.NoError(t, d.backSessionForAttachment(effect))34 require.NoError(t, d.backSessionForAttachment(effect))
35 frames := transport.Sends()35 frames := transport.Sends()
@@ -83,7 +83,7 @@ func TestBackSessionOffersPreviousRouteOnCurrentDaemon(t *testing.T) {
83 })83 })
84 ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{84 ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{
85 Targets: []protocol.RouteAttentionTarget{{Ref: previous, Target: exact}},85 Targets: []protocol.RouteAttentionTarget{{Ref: previous, Target: exact}},
86- })86+ }, ac.transportSnapshot(), d.clock.Now())
87 87
88 require.NoError(t, d.backSessionForAttachment(effect))88 require.NoError(t, d.backSessionForAttachment(effect))
89 frames := transport.Sends()89 frames := transport.Sends()
M internal/usecase/daemon/typed_connection_test.go +3 -0
@@ -314,6 +314,9 @@ func testServerFrame(message protocol.ServerMessage) (wire.Frame, error) {
314314 case protocol.RouteNavigationFailure:
315315 payload, err := wire.MarshalRouteNavigationFailure(m)
316316 return wire.Frame{Type: wire.MsgRouteNavigationFailure, Payload: payload}, err
317+ case protocol.RouteRetired:
318+ payload, err := wire.MarshalRouteRetired(m)
319+ return wire.Frame{Type: wire.MsgRouteRetired, Payload: payload}, err
317320 case protocol.RoutePosition:
318321 payload, err := wire.MarshalRoutePosition(m)
319322 return wire.Frame{Type: wire.MsgRoutePosition, Payload: payload}, err
@@ -314,6 +314,9 @@ func testServerFrame(message protocol.ServerMessage) (wire.Frame, error) {
314 case protocol.RouteNavigationFailure:314 case protocol.RouteNavigationFailure:
315 payload, err := wire.MarshalRouteNavigationFailure(m)315 payload, err := wire.MarshalRouteNavigationFailure(m)
316 return wire.Frame{Type: wire.MsgRouteNavigationFailure, Payload: payload}, err316 return wire.Frame{Type: wire.MsgRouteNavigationFailure, Payload: payload}, err
317+ case protocol.RouteRetired:
318+ payload, err := wire.MarshalRouteRetired(m)
319+ return wire.Frame{Type: wire.MsgRouteRetired, Payload: payload}, err
317 case protocol.RoutePosition:320 case protocol.RoutePosition:
318 payload, err := wire.MarshalRoutePosition(m)321 payload, err := wire.MarshalRoutePosition(m)
319 return wire.Frame{Type: wire.MsgRoutePosition, Payload: payload}, err322 return wire.Frame{Type: wire.MsgRoutePosition, Payload: payload}, err

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close