fix(client): close route ledger review gaps
- commit
- 121c2e15d9325e046a2fe513defcc1e9a144e1df
- author
- brice · · 2026-08-09 12:18 UTC
- parent
- 4dbec77
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
13 files · +424 -76
M .agent-reviews/redteam.md +10 -0
| @@ -36,3 +36,13 @@ objection with its resolution and acceptance check. | |||
| 36 | 36 | - **Snapshot leakage:** `RecentRouteEntry` contains only opaque key/generation and display fields. Dialers, requests, resume credentials, and exact targets remain in private `routeRecord` values and are copied out only through navigation lookup. | |
| 37 | 37 | - **Transactional navigation:** connector failure, stale key/generation, or committed-target mismatch returns before ledger commit; origin is restored from the selected private record rather than connector input. | |
| 38 | 38 | - **Bounds/strictness:** snapshot entry count, labels, refs, closed enums, bool encodings, truncation, trailing bytes, and protocol-version peeking are covered by tests; focused race tests pass. | |
| 39 | + | ||
| 40 | + | ## Follow-up fixes | |
| 41 | + | ||
| 42 | + | The implementation now commits a successful Welcome into the private ledger before | |
| 43 | + | raw-mode entry, publishes a complete snapshot frame, and accepts/validates that | |
| 44 | + | frame on the daemon attachment. The active route is metadata-only, the private | |
| 45 | + | history cap is 20, origin keys distinguish daemon endpoints, route labels reject | |
| 46 | + | invalid UTF-8/control text, navigation actions carry the snapshot generation, | |
| 47 | + | active selection is a no-op, and failed transitions invoke the prior-route | |
| 48 | + | restore hook before returning failure. | |
| @@ -36,3 +36,13 @@ objection with its resolution and acceptance check. | |||||
| 36 | - **Snapshot leakage:** `RecentRouteEntry` contains only opaque key/generation and display fields. Dialers, requests, resume credentials, and exact targets remain in private `routeRecord` values and are copied out only through navigation lookup. | 36 | - **Snapshot leakage:** `RecentRouteEntry` contains only opaque key/generation and display fields. Dialers, requests, resume credentials, and exact targets remain in private `routeRecord` values and are copied out only through navigation lookup. | ||
| 37 | - **Transactional navigation:** connector failure, stale key/generation, or committed-target mismatch returns before ledger commit; origin is restored from the selected private record rather than connector input. | 37 | - **Transactional navigation:** connector failure, stale key/generation, or committed-target mismatch returns before ledger commit; origin is restored from the selected private record rather than connector input. | ||
| 38 | - **Bounds/strictness:** snapshot entry count, labels, refs, closed enums, bool encodings, truncation, trailing bytes, and protocol-version peeking are covered by tests; focused race tests pass. | 38 | - **Bounds/strictness:** snapshot entry count, labels, refs, closed enums, bool encodings, truncation, trailing bytes, and protocol-version peeking are covered by tests; focused race tests pass. | ||
| 39 | + | ||||
| 40 | + | ## Follow-up fixes | |||
| 41 | + | ||||
| 42 | + | The implementation now commits a successful Welcome into the private ledger before | |||
| 43 | + | raw-mode entry, publishes a complete snapshot frame, and accepts/validates that | |||
| 44 | + | frame on the daemon attachment. The active route is metadata-only, the private | |||
| 45 | + | history cap is 20, origin keys distinguish daemon endpoints, route labels reject | |||
| 46 | + | invalid UTF-8/control text, navigation actions carry the snapshot generation, | |||
| 47 | + | active selection is a no-op, and failed transitions invoke the prior-route | |||
| 48 | + | restore hook before returning failure. | |||
M internal/app/run.go +8 -1
| @@ -867,6 +867,7 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac | |||
| 867 | 867 | var remoteSelection *domain.RemoteSessionTarget | |
| 868 | 868 | var remoteEnvironmentPolicy ports.EnvironmentPolicy | |
| 869 | 869 | routeOrigin := ports.RouteOriginLocal | |
| 870 | + | routeOriginKey := "local" | |
| 870 | 871 | seenHandoffs := make(map[attachHandoffKey]struct{}) | |
| 871 | 872 | handoffHops := 0 | |
| 872 | 873 | admitHandoff := func(next attachHandoffKey) error { | |
| @@ -882,6 +883,7 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac | |||
| 882 | 883 | } | |
| 883 | 884 | if remoteTarget != "" { | |
| 884 | 885 | routeOrigin = ports.RouteOriginRemote | |
| 886 | + | routeOriginKey = remoteTarget | |
| 885 | 887 | if modeErr != nil { | |
| 886 | 888 | return modeErr | |
| 887 | 889 | } | |
| @@ -929,6 +931,7 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac | |||
| 929 | 931 | SessionName: target.Session, | |
| 930 | 932 | Remote: true, | |
| 931 | 933 | Origin: ports.RouteOriginDiscovery, | |
| 934 | + | OriginKey: target.Endpoint, | |
| 932 | 935 | RemoteTarget: selection, | |
| 933 | 936 | EnvironmentPolicy: policy, | |
| 934 | 937 | }, nil | |
| @@ -955,11 +958,13 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac | |||
| 955 | 958 | AttachHandoff: handoff, | |
| 956 | 959 | Remote: true, | |
| 957 | 960 | Origin: ports.RouteOriginRemote, | |
| 961 | + | OriginKey: remoteTarget, | |
| 958 | 962 | }, client.AttachRequest{ | |
| 959 | 963 | Intent: intent, | |
| 960 | 964 | SessionName: name, | |
| 961 | 965 | Remote: true, | |
| 962 | 966 | Origin: routeOrigin, | |
| 967 | + | OriginKey: routeOriginKey, | |
| 963 | 968 | RemoteTarget: remoteSelection, | |
| 964 | 969 | EnvironmentPolicy: remoteEnvironmentPolicy, | |
| 965 | 970 | }) | |
| @@ -976,7 +981,8 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac | |||
| 976 | 981 | AttachHandoff: handoff, | |
| 977 | 982 | Remote: false, | |
| 978 | 983 | Origin: ports.RouteOriginLocal, | |
| 979 | - | }, client.AttachRequest{Intent: intent, SessionName: name, Origin: ports.RouteOriginLocal}) | |
| 984 | + | OriginKey: "local", | |
| 985 | + | }, client.AttachRequest{Intent: intent, SessionName: name, Origin: ports.RouteOriginLocal, OriginKey: "local"}) | |
| 980 | 986 | } | |
| 981 | 987 | ||
| 982 | 988 | var handoffErr *client.AttachTargetError | |
| @@ -1000,6 +1006,7 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac | |||
| 1000 | 1006 | } | |
| 1001 | 1007 | remoteTarget = handoffErr.Target.Endpoint | |
| 1002 | 1008 | routeOrigin = ports.RouteOriginDiscovery | |
| 1009 | + | routeOriginKey = remoteTarget | |
| 1003 | 1010 | name = handoffErr.Target.Session | |
| 1004 | 1011 | intent = handoffErr.Target.Intent | |
| 1005 | 1012 | remoteSelection = nil | |
| @@ -867,6 +867,7 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac | |||||
| 867 | var remoteSelection *domain.RemoteSessionTarget | 867 | var remoteSelection *domain.RemoteSessionTarget | ||
| 868 | var remoteEnvironmentPolicy ports.EnvironmentPolicy | 868 | var remoteEnvironmentPolicy ports.EnvironmentPolicy | ||
| 869 | routeOrigin := ports.RouteOriginLocal | 869 | routeOrigin := ports.RouteOriginLocal | ||
| 870 | + | routeOriginKey := "local" | |||
| 870 | seenHandoffs := make(map[attachHandoffKey]struct{}) | 871 | seenHandoffs := make(map[attachHandoffKey]struct{}) | ||
| 871 | handoffHops := 0 | 872 | handoffHops := 0 | ||
| 872 | admitHandoff := func(next attachHandoffKey) error { | 873 | admitHandoff := func(next attachHandoffKey) error { | ||
| @@ -882,6 +883,7 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac | |||||
| 882 | } | 883 | } | ||
| 883 | if remoteTarget != "" { | 884 | if remoteTarget != "" { | ||
| 884 | routeOrigin = ports.RouteOriginRemote | 885 | routeOrigin = ports.RouteOriginRemote | ||
| 886 | + | routeOriginKey = remoteTarget | |||
| 885 | if modeErr != nil { | 887 | if modeErr != nil { | ||
| 886 | return modeErr | 888 | return modeErr | ||
| 887 | } | 889 | } | ||
| @@ -929,6 +931,7 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac | |||||
| 929 | SessionName: target.Session, | 931 | SessionName: target.Session, | ||
| 930 | Remote: true, | 932 | Remote: true, | ||
| 931 | Origin: ports.RouteOriginDiscovery, | 933 | Origin: ports.RouteOriginDiscovery, | ||
| 934 | + | OriginKey: target.Endpoint, | |||
| 932 | RemoteTarget: selection, | 935 | RemoteTarget: selection, | ||
| 933 | EnvironmentPolicy: policy, | 936 | EnvironmentPolicy: policy, | ||
| 934 | }, nil | 937 | }, nil | ||
| @@ -955,11 +958,13 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac | |||||
| 955 | AttachHandoff: handoff, | 958 | AttachHandoff: handoff, | ||
| 956 | Remote: true, | 959 | Remote: true, | ||
| 957 | Origin: ports.RouteOriginRemote, | 960 | Origin: ports.RouteOriginRemote, | ||
| 961 | + | OriginKey: remoteTarget, | |||
| 958 | }, client.AttachRequest{ | 962 | }, client.AttachRequest{ | ||
| 959 | Intent: intent, | 963 | Intent: intent, | ||
| 960 | SessionName: name, | 964 | SessionName: name, | ||
| 961 | Remote: true, | 965 | Remote: true, | ||
| 962 | Origin: routeOrigin, | 966 | Origin: routeOrigin, | ||
| 967 | + | OriginKey: routeOriginKey, | |||
| 963 | RemoteTarget: remoteSelection, | 968 | RemoteTarget: remoteSelection, | ||
| 964 | EnvironmentPolicy: remoteEnvironmentPolicy, | 969 | EnvironmentPolicy: remoteEnvironmentPolicy, | ||
| 965 | }) | 970 | }) | ||
| @@ -976,7 +981,8 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac | |||||
| 976 | AttachHandoff: handoff, | 981 | AttachHandoff: handoff, | ||
| 977 | Remote: false, | 982 | Remote: false, | ||
| 978 | Origin: ports.RouteOriginLocal, | 983 | Origin: ports.RouteOriginLocal, | ||
| 979 | - | }, client.AttachRequest{Intent: intent, SessionName: name, Origin: ports.RouteOriginLocal}) | 984 | + | OriginKey: "local", |
| 985 | + | }, client.AttachRequest{Intent: intent, SessionName: name, Origin: ports.RouteOriginLocal, OriginKey: "local"}) | |||
| 980 | } | 986 | } | ||
| 981 | 987 | ||||
| 982 | var handoffErr *client.AttachTargetError | 988 | var handoffErr *client.AttachTargetError | ||
| @@ -1000,6 +1006,7 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac | |||||
| 1000 | } | 1006 | } | ||
| 1001 | remoteTarget = handoffErr.Target.Endpoint | 1007 | remoteTarget = handoffErr.Target.Endpoint | ||
| 1002 | routeOrigin = ports.RouteOriginDiscovery | 1008 | routeOrigin = ports.RouteOriginDiscovery | ||
| 1009 | + | routeOriginKey = remoteTarget | |||
| 1003 | name = handoffErr.Target.Session | 1010 | name = handoffErr.Target.Session | ||
| 1004 | intent = handoffErr.Target.Intent | 1011 | intent = handoffErr.Target.Intent | ||
| 1005 | remoteSelection = nil | 1012 | remoteSelection = nil | ||
M internal/app/run_test.go +2 -2
| @@ -959,11 +959,11 @@ func TestRunAttachWithDepsLocalPickerHandoffAttachesSelectedRemote(t *testing.T) | |||
| 959 | 959 | runClient: func(_ context.Context, deps client.Dependencies, request client.AttachRequest) error { | |
| 960 | 960 | if deps.Remote { | |
| 961 | 961 | remoteCalls++ | |
| 962 | - | require.Equal(t, client.AttachRequest{Intent: ports.IntentAttach, SessionName: "picked", Remote: true, Origin: ports.RouteOriginDiscovery, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned}, request) | |
| 962 | + | require.Equal(t, client.AttachRequest{Intent: ports.IntentAttach, SessionName: "picked", Remote: true, Origin: ports.RouteOriginDiscovery, OriginKey: "selected.example", EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned}, request) | |
| 963 | 963 | return nil | |
| 964 | 964 | } | |
| 965 | 965 | localCalls++ | |
| 966 | - | require.Equal(t, client.AttachRequest{Intent: ports.IntentAttach, SessionName: "work", Origin: ports.RouteOriginLocal}, request) | |
| 966 | + | require.Equal(t, client.AttachRequest{Intent: ports.IntentAttach, SessionName: "work", Origin: ports.RouteOriginLocal, OriginKey: "local"}, request) | |
| 967 | 967 | return &client.AttachTargetError{Target: ports.AttachTarget{Endpoint: "selected.example", Session: "picked", Intent: ports.IntentAttach}} | |
| 968 | 968 | }, | |
| 969 | 969 | }) | |
| @@ -959,11 +959,11 @@ func TestRunAttachWithDepsLocalPickerHandoffAttachesSelectedRemote(t *testing.T) | |||||
| 959 | runClient: func(_ context.Context, deps client.Dependencies, request client.AttachRequest) error { | 959 | runClient: func(_ context.Context, deps client.Dependencies, request client.AttachRequest) error { | ||
| 960 | if deps.Remote { | 960 | if deps.Remote { | ||
| 961 | remoteCalls++ | 961 | remoteCalls++ | ||
| 962 | - | require.Equal(t, client.AttachRequest{Intent: ports.IntentAttach, SessionName: "picked", Remote: true, Origin: ports.RouteOriginDiscovery, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned}, request) | 962 | + | require.Equal(t, client.AttachRequest{Intent: ports.IntentAttach, SessionName: "picked", Remote: true, Origin: ports.RouteOriginDiscovery, OriginKey: "selected.example", EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned}, request) |
| 963 | return nil | 963 | return nil | ||
| 964 | } | 964 | } | ||
| 965 | localCalls++ | 965 | localCalls++ | ||
| 966 | - | require.Equal(t, client.AttachRequest{Intent: ports.IntentAttach, SessionName: "work", Origin: ports.RouteOriginLocal}, request) | 966 | + | require.Equal(t, client.AttachRequest{Intent: ports.IntentAttach, SessionName: "work", Origin: ports.RouteOriginLocal, OriginKey: "local"}, request) |
| 967 | return &client.AttachTargetError{Target: ports.AttachTarget{Endpoint: "selected.example", Session: "picked", Intent: ports.IntentAttach}} | 967 | return &client.AttachTargetError{Target: ports.AttachTarget{Endpoint: "selected.example", Session: "picked", Intent: ports.IntentAttach}} | ||
| 968 | }, | 968 | }, | ||
| 969 | }) | 969 | }) | ||
M internal/ports/routes.go +30 -5
| @@ -3,6 +3,8 @@ package ports | |||
| 3 | 3 | import ( | |
| 4 | 4 | "errors" | |
| 5 | 5 | "fmt" | |
| 6 | + | "unicode" | |
| 7 | + | "unicode/utf8" | |
| 6 | 8 | ||
| 7 | 9 | "github.com/bnema/vev/internal/domain" | |
| 8 | 10 | ) | |
| @@ -83,6 +85,27 @@ func (r RouteReachability) Validate() error { | |||
| 83 | 85 | return nil | |
| 84 | 86 | } | |
| 85 | 87 | ||
| 88 | + | // ValidateRouteLabel rejects malformed or terminal-control text before it can | |
| 89 | + | // enter a daemon-rendered route snapshot. Empty labels are allowed only when | |
| 90 | + | // the caller explicitly marks them optional. | |
| 91 | + | func ValidateRouteLabel(value string, allowEmpty bool) error { | |
| 92 | + | if !allowEmpty && value == "" { | |
| 93 | + | return errors.New("route label is empty") | |
| 94 | + | } | |
| 95 | + | if len(value) > RouteLabelMaxBytes { | |
| 96 | + | return errors.New("route label is too long") | |
| 97 | + | } | |
| 98 | + | if !utf8.ValidString(value) { | |
| 99 | + | return errors.New("route label is not valid UTF-8") | |
| 100 | + | } | |
| 101 | + | for _, r := range value { | |
| 102 | + | if unicode.IsControl(r) { | |
| 103 | + | return errors.New("route label contains a control character") | |
| 104 | + | } | |
| 105 | + | } | |
| 106 | + | return nil | |
| 107 | + | } | |
| 108 | + | ||
| 86 | 109 | // ExactSessionTarget is the daemon-neutral identity required to attach to a | |
| 87 | 110 | // specific session lifecycle. The transport endpoint is deliberately absent: | |
| 88 | 111 | // the selected route's dialer owns that boundary. | |
| @@ -159,11 +182,12 @@ type RecentRouteSnapshot struct { | |||
| 159 | 182 | Entries []RecentRouteEntry | |
| 160 | 183 | } | |
| 161 | 184 | ||
| 162 | - | // RouteNavigationAction asks the client to resolve one captured route key and | |
| 163 | - | // generation. It carries no session name, endpoint, or credential. | |
| 185 | + | // RouteNavigationAction asks the client to resolve one entry from a specific | |
| 186 | + | // complete snapshot. It carries no session name, endpoint, or credential. | |
| 164 | 187 | type RouteNavigationAction struct { | |
| 165 | - | Key uint64 | |
| 166 | - | Generation uint64 | |
| 188 | + | SnapshotGeneration uint64 | |
| 189 | + | Key uint64 | |
| 190 | + | Generation uint64 | |
| 167 | 191 | } | |
| 168 | 192 | ||
| 169 | 193 | // RouteNavigationFailure is a bounded taxonomy for a rejected or stale route | |
| @@ -195,7 +219,8 @@ func (c RouteFailureCode) valid() bool { | |||
| 195 | 219 | ||
| 196 | 220 | const ( | |
| 197 | 221 | // RouteSnapshotMaxEntries bounds one immutable publication and the amount | |
| 198 | - | // of work a receiver performs before returning to its transport loop. | |
| 222 | + | // of work a receiver performs before returning to its transport loop. The | |
| 223 | + | // private client history may use a smaller product cap. | |
| 199 | 224 | RouteSnapshotMaxEntries = 32 | |
| 200 | 225 | RouteLabelMaxBytes = 128 | |
| 201 | 226 | ) | |
| @@ -3,6 +3,8 @@ package ports | |||||
| 3 | import ( | 3 | import ( | ||
| 4 | "errors" | 4 | "errors" | ||
| 5 | "fmt" | 5 | "fmt" | ||
| 6 | + | "unicode" | |||
| 7 | + | "unicode/utf8" | |||
| 6 | 8 | ||||
| 7 | "github.com/bnema/vev/internal/domain" | 9 | "github.com/bnema/vev/internal/domain" | ||
| 8 | ) | 10 | ) | ||
| @@ -83,6 +85,27 @@ func (r RouteReachability) Validate() error { | |||||
| 83 | return nil | 85 | return nil | ||
| 84 | } | 86 | } | ||
| 85 | 87 | ||||
| 88 | + | // ValidateRouteLabel rejects malformed or terminal-control text before it can | |||
| 89 | + | // enter a daemon-rendered route snapshot. Empty labels are allowed only when | |||
| 90 | + | // the caller explicitly marks them optional. | |||
| 91 | + | func ValidateRouteLabel(value string, allowEmpty bool) error { | |||
| 92 | + | if !allowEmpty && value == "" { | |||
| 93 | + | return errors.New("route label is empty") | |||
| 94 | + | } | |||
| 95 | + | if len(value) > RouteLabelMaxBytes { | |||
| 96 | + | return errors.New("route label is too long") | |||
| 97 | + | } | |||
| 98 | + | if !utf8.ValidString(value) { | |||
| 99 | + | return errors.New("route label is not valid UTF-8") | |||
| 100 | + | } | |||
| 101 | + | for _, r := range value { | |||
| 102 | + | if unicode.IsControl(r) { | |||
| 103 | + | return errors.New("route label contains a control character") | |||
| 104 | + | } | |||
| 105 | + | } | |||
| 106 | + | return nil | |||
| 107 | + | } | |||
| 108 | + | ||||
| 86 | // ExactSessionTarget is the daemon-neutral identity required to attach to a | 109 | // ExactSessionTarget is the daemon-neutral identity required to attach to a | ||
| 87 | // specific session lifecycle. The transport endpoint is deliberately absent: | 110 | // specific session lifecycle. The transport endpoint is deliberately absent: | ||
| 88 | // the selected route's dialer owns that boundary. | 111 | // the selected route's dialer owns that boundary. | ||
| @@ -159,11 +182,12 @@ type RecentRouteSnapshot struct { | |||||
| 159 | Entries []RecentRouteEntry | 182 | Entries []RecentRouteEntry | ||
| 160 | } | 183 | } | ||
| 161 | 184 | ||||
| 162 | - | // RouteNavigationAction asks the client to resolve one captured route key and | 185 | + | // RouteNavigationAction asks the client to resolve one entry from a specific |
| 163 | - | // generation. It carries no session name, endpoint, or credential. | 186 | + | // complete snapshot. It carries no session name, endpoint, or credential. |
| 164 | type RouteNavigationAction struct { | 187 | type RouteNavigationAction struct { | ||
| 165 | - | Key uint64 | 188 | + | SnapshotGeneration uint64 |
| 166 | - | Generation uint64 | 189 | + | Key uint64 |
| 190 | + | Generation uint64 | |||
| 167 | } | 191 | } | ||
| 168 | 192 | ||||
| 169 | // RouteNavigationFailure is a bounded taxonomy for a rejected or stale route | 193 | // RouteNavigationFailure is a bounded taxonomy for a rejected or stale route | ||
| @@ -195,7 +219,8 @@ func (c RouteFailureCode) valid() bool { | |||||
| 195 | 219 | ||||
| 196 | const ( | 220 | const ( | ||
| 197 | // RouteSnapshotMaxEntries bounds one immutable publication and the amount | 221 | // RouteSnapshotMaxEntries bounds one immutable publication and the amount | ||
| 198 | - | // of work a receiver performs before returning to its transport loop. | 222 | + | // of work a receiver performs before returning to its transport loop. The |
| 223 | + | // private client history may use a smaller product cap. | |||
| 199 | RouteSnapshotMaxEntries = 32 | 224 | RouteSnapshotMaxEntries = 32 | ||
| 200 | RouteLabelMaxBytes = 128 | 225 | RouteLabelMaxBytes = 128 | ||
| 201 | ) | 226 | ) | ||
M internal/ports/routes_test.go +18 -4
| @@ -89,7 +89,6 @@ func testRouteSnapshot() RecentRouteSnapshot { | |||
| 89 | 89 | Previous: RouteRef{Key: 12, Generation: 4}, | |
| 90 | 90 | Home: RouteRef{Key: 11, Generation: 3}, | |
| 91 | 91 | Entries: []RecentRouteEntry{ | |
| 92 | - | {Key: 11, Generation: 3, Name: "work", Kind: RouteKindLocal, Reachability: RouteReachabilityReachable}, | |
| 93 | 92 | {Key: 12, Generation: 4, Name: "logs", HostLabel: "edge", Kind: RouteKindRemote, Attention: true, Reachability: RouteReachabilityUnknown}, | |
| 94 | 93 | }, | |
| 95 | 94 | } | |
| @@ -110,7 +109,12 @@ func TestRecentRouteSnapshotCodecRoundTripAndBounds(t *testing.T) { | |||
| 110 | 109 | } | |
| 111 | 110 | ||
| 112 | 111 | invalid := want | |
| 113 | - | invalid.Active = RouteRef{Key: 99, Generation: 99} | |
| 112 | + | invalid.Previous = RouteRef{Key: 99, Generation: 99} | |
| 113 | + | _, err = MarshalRecentRouteSnapshot(invalid) | |
| 114 | + | require.ErrorIs(t, err, ErrInvalidRouteWire) | |
| 115 | + | ||
| 116 | + | invalid = want | |
| 117 | + | invalid.Entries = append(invalid.Entries, RecentRouteEntry{Key: 11, Generation: 3, Name: "work", Kind: RouteKindLocal}) | |
| 114 | 118 | _, err = MarshalRecentRouteSnapshot(invalid) | |
| 115 | 119 | require.ErrorIs(t, err, ErrInvalidRouteWire) | |
| 116 | 120 | ||
| @@ -128,8 +132,18 @@ func TestRecentRouteSnapshotCodecRoundTripAndBounds(t *testing.T) { | |||
| 128 | 132 | require.ErrorIs(t, err, ErrInvalidRouteWire) | |
| 129 | 133 | } | |
| 130 | 134 | ||
| 135 | + | func TestRouteLabelsRejectTerminalUnsafeText(t *testing.T) { | |
| 136 | + | for _, value := range []string{"line\nfeed", "\x1b[31mred", string([]byte{0xff, 0xfe})} { | |
| 137 | + | _, err := MarshalRecentRouteSnapshot(RecentRouteSnapshot{ | |
| 138 | + | Generation: 1, | |
| 139 | + | Entries: []RecentRouteEntry{{Key: 1, Generation: 1, Name: value, Kind: RouteKindLocal}}, | |
| 140 | + | }) | |
| 141 | + | require.Error(t, err) | |
| 142 | + | } | |
| 143 | + | } | |
| 144 | + | ||
| 131 | 145 | func TestRouteActionAndFailureCodecsAreBounded(t *testing.T) { | |
| 132 | - | action := RouteNavigationAction{Key: 7, Generation: 8} | |
| 146 | + | action := RouteNavigationAction{SnapshotGeneration: 9, Key: 7, Generation: 8} | |
| 133 | 147 | encoded, err := MarshalRouteNavigationAction(action) | |
| 134 | 148 | require.NoError(t, err) | |
| 135 | 149 | got, err := UnmarshalRouteNavigationAction(encoded) | |
| @@ -143,7 +157,7 @@ func TestRouteActionAndFailureCodecsAreBounded(t *testing.T) { | |||
| 143 | 157 | require.NoError(t, err) | |
| 144 | 158 | require.Equal(t, failure, gotFailure) | |
| 145 | 159 | ||
| 146 | - | for _, invalid := range []RouteNavigationAction{{}, {Key: 1}, {Generation: 1}} { | |
| 160 | + | for _, invalid := range []RouteNavigationAction{{}, {SnapshotGeneration: 1, Key: 1}, {SnapshotGeneration: 1, Generation: 1}, {Key: 1, Generation: 1}} { | |
| 147 | 161 | _, err := MarshalRouteNavigationAction(invalid) | |
| 148 | 162 | require.Error(t, err) | |
| 149 | 163 | } | |
| @@ -89,7 +89,6 @@ func testRouteSnapshot() RecentRouteSnapshot { | |||||
| 89 | Previous: RouteRef{Key: 12, Generation: 4}, | 89 | Previous: RouteRef{Key: 12, Generation: 4}, | ||
| 90 | Home: RouteRef{Key: 11, Generation: 3}, | 90 | Home: RouteRef{Key: 11, Generation: 3}, | ||
| 91 | Entries: []RecentRouteEntry{ | 91 | Entries: []RecentRouteEntry{ | ||
| 92 | - | {Key: 11, Generation: 3, Name: "work", Kind: RouteKindLocal, Reachability: RouteReachabilityReachable}, | |||
| 93 | {Key: 12, Generation: 4, Name: "logs", HostLabel: "edge", Kind: RouteKindRemote, Attention: true, Reachability: RouteReachabilityUnknown}, | 92 | {Key: 12, Generation: 4, Name: "logs", HostLabel: "edge", Kind: RouteKindRemote, Attention: true, Reachability: RouteReachabilityUnknown}, | ||
| 94 | }, | 93 | }, | ||
| 95 | } | 94 | } | ||
| @@ -110,7 +109,12 @@ func TestRecentRouteSnapshotCodecRoundTripAndBounds(t *testing.T) { | |||||
| 110 | } | 109 | } | ||
| 111 | 110 | ||||
| 112 | invalid := want | 111 | invalid := want | ||
| 113 | - | invalid.Active = RouteRef{Key: 99, Generation: 99} | 112 | + | invalid.Previous = RouteRef{Key: 99, Generation: 99} |
| 113 | + | _, err = MarshalRecentRouteSnapshot(invalid) | |||
| 114 | + | require.ErrorIs(t, err, ErrInvalidRouteWire) | |||
| 115 | + | ||||
| 116 | + | invalid = want | |||
| 117 | + | invalid.Entries = append(invalid.Entries, RecentRouteEntry{Key: 11, Generation: 3, Name: "work", Kind: RouteKindLocal}) | |||
| 114 | _, err = MarshalRecentRouteSnapshot(invalid) | 118 | _, err = MarshalRecentRouteSnapshot(invalid) | ||
| 115 | require.ErrorIs(t, err, ErrInvalidRouteWire) | 119 | require.ErrorIs(t, err, ErrInvalidRouteWire) | ||
| 116 | 120 | ||||
| @@ -128,8 +132,18 @@ func TestRecentRouteSnapshotCodecRoundTripAndBounds(t *testing.T) { | |||||
| 128 | require.ErrorIs(t, err, ErrInvalidRouteWire) | 132 | require.ErrorIs(t, err, ErrInvalidRouteWire) | ||
| 129 | } | 133 | } | ||
| 130 | 134 | ||||
| 135 | + | func TestRouteLabelsRejectTerminalUnsafeText(t *testing.T) { | |||
| 136 | + | for _, value := range []string{"line\nfeed", "\x1b[31mred", string([]byte{0xff, 0xfe})} { | |||
| 137 | + | _, err := MarshalRecentRouteSnapshot(RecentRouteSnapshot{ | |||
| 138 | + | Generation: 1, | |||
| 139 | + | Entries: []RecentRouteEntry{{Key: 1, Generation: 1, Name: value, Kind: RouteKindLocal}}, | |||
| 140 | + | }) | |||
| 141 | + | require.Error(t, err) | |||
| 142 | + | } | |||
| 143 | + | } | |||
| 144 | + | ||||
| 131 | func TestRouteActionAndFailureCodecsAreBounded(t *testing.T) { | 145 | func TestRouteActionAndFailureCodecsAreBounded(t *testing.T) { | ||
| 132 | - | action := RouteNavigationAction{Key: 7, Generation: 8} | 146 | + | action := RouteNavigationAction{SnapshotGeneration: 9, Key: 7, Generation: 8} |
| 133 | encoded, err := MarshalRouteNavigationAction(action) | 147 | encoded, err := MarshalRouteNavigationAction(action) | ||
| 134 | require.NoError(t, err) | 148 | require.NoError(t, err) | ||
| 135 | got, err := UnmarshalRouteNavigationAction(encoded) | 149 | got, err := UnmarshalRouteNavigationAction(encoded) | ||
| @@ -143,7 +157,7 @@ func TestRouteActionAndFailureCodecsAreBounded(t *testing.T) { | |||||
| 143 | require.NoError(t, err) | 157 | require.NoError(t, err) | ||
| 144 | require.Equal(t, failure, gotFailure) | 158 | require.Equal(t, failure, gotFailure) | ||
| 145 | 159 | ||||
| 146 | - | for _, invalid := range []RouteNavigationAction{{}, {Key: 1}, {Generation: 1}} { | 160 | + | for _, invalid := range []RouteNavigationAction{{}, {SnapshotGeneration: 1, Key: 1}, {SnapshotGeneration: 1, Generation: 1}, {Key: 1, Generation: 1}} { |
| 147 | _, err := MarshalRouteNavigationAction(invalid) | 161 | _, err := MarshalRouteNavigationAction(invalid) | ||
| 148 | require.Error(t, err) | 162 | require.Error(t, err) | ||
| 149 | } | 163 | } | ||
M internal/ports/routes_wire.go +19 -9
| @@ -8,11 +8,8 @@ import ( | |||
| 8 | 8 | var ErrInvalidRouteWire = errors.New("invalid route wire message") | |
| 9 | 9 | ||
| 10 | 10 | func validateRouteString(value, field string, allowEmpty bool) error { | |
| 11 | - | if !allowEmpty && value == "" { | |
| 12 | - | return fmt.Errorf("%s is empty", field) | |
| 13 | - | } | |
| 14 | - | if len(value) > RouteLabelMaxBytes { | |
| 15 | - | return fmt.Errorf("%s is too long", field) | |
| 11 | + | if err := ValidateRouteLabel(value, allowEmpty); err != nil { | |
| 12 | + | return fmt.Errorf("%s: %w", field, err) | |
| 16 | 13 | } | |
| 17 | 14 | return nil | |
| 18 | 15 | } | |
| @@ -237,6 +234,9 @@ func validateRecentRouteSnapshot(snapshot RecentRouteSnapshot) error { | |||
| 237 | 234 | } | |
| 238 | 235 | refs := make(map[RouteRef]struct{}, len(snapshot.Entries)) | |
| 239 | 236 | for _, entry := range snapshot.Entries { | |
| 237 | + | if snapshot.Active == (RouteRef{Key: entry.Key, Generation: entry.Generation}) { | |
| 238 | + | return fmt.Errorf("%w: active route must be metadata-only", ErrInvalidRouteWire) | |
| 239 | + | } | |
| 240 | 240 | if err := validateRecentRouteEntry(entry); err != nil { | |
| 241 | 241 | return err | |
| 242 | 242 | } | |
| @@ -246,12 +246,18 @@ func validateRecentRouteSnapshot(snapshot RecentRouteSnapshot) error { | |||
| 246 | 246 | } | |
| 247 | 247 | refs[ref] = struct{}{} | |
| 248 | 248 | } | |
| 249 | - | for _, ref := range []RouteRef{snapshot.Active, snapshot.Previous, snapshot.Home} { | |
| 250 | - | if ref.empty() { | |
| 249 | + | for name, ref := range map[string]RouteRef{ | |
| 250 | + | "active": snapshot.Active, | |
| 251 | + | "previous": snapshot.Previous, | |
| 252 | + | "home": snapshot.Home, | |
| 253 | + | } { | |
| 254 | + | if ref.empty() || ref == snapshot.Active { | |
| 255 | + | // Active is metadata-only and intentionally excluded from the | |
| 256 | + | // recent display entries. Home may point at active as well. | |
| 251 | 257 | continue | |
| 252 | 258 | } | |
| 253 | 259 | if _, exists := refs[ref]; !exists { | |
| 254 | - | return fmt.Errorf("%w: route reference is absent from snapshot", ErrInvalidRouteWire) | |
| 260 | + | return fmt.Errorf("%w: %s route reference is absent from snapshot", ErrInvalidRouteWire, name) | |
| 255 | 261 | } | |
| 256 | 262 | } | |
| 257 | 263 | return nil | |
| @@ -357,7 +363,7 @@ func UnmarshalRecentRouteSnapshot(b []byte) (RecentRouteSnapshot, error) { | |||
| 357 | 363 | } | |
| 358 | 364 | ||
| 359 | 365 | func validateRouteNavigationAction(action RouteNavigationAction) error { | |
| 360 | - | if action.Key == 0 || action.Generation == 0 { | |
| 366 | + | if action.SnapshotGeneration == 0 || action.Key == 0 || action.Generation == 0 { | |
| 361 | 367 | return fmt.Errorf("%w: navigation action identity is zero", ErrInvalidRouteWire) | |
| 362 | 368 | } | |
| 363 | 369 | return nil | |
| @@ -368,6 +374,7 @@ func MarshalRouteNavigationAction(action RouteNavigationAction) ([]byte, error) | |||
| 368 | 374 | return nil, err | |
| 369 | 375 | } | |
| 370 | 376 | w := payloadWriter{} | |
| 377 | + | w.putUint64(action.SnapshotGeneration) | |
| 371 | 378 | w.putUint64(action.Key) | |
| 372 | 379 | w.putUint64(action.Generation) | |
| 373 | 380 | return w.b, nil | |
| @@ -377,6 +384,9 @@ func UnmarshalRouteNavigationAction(b []byte) (RouteNavigationAction, error) { | |||
| 377 | 384 | r := payloadReader{b: b} | |
| 378 | 385 | var action RouteNavigationAction | |
| 379 | 386 | var err error | |
| 387 | + | if action.SnapshotGeneration, err = r.getUint64(); err != nil { | |
| 388 | + | return RouteNavigationAction{}, err | |
| 389 | + | } | |
| 380 | 390 | if action.Key, err = r.getUint64(); err != nil { | |
| 381 | 391 | return RouteNavigationAction{}, err | |
| 382 | 392 | } | |
| @@ -8,11 +8,8 @@ import ( | |||||
| 8 | var ErrInvalidRouteWire = errors.New("invalid route wire message") | 8 | var ErrInvalidRouteWire = errors.New("invalid route wire message") | ||
| 9 | 9 | ||||
| 10 | func validateRouteString(value, field string, allowEmpty bool) error { | 10 | func validateRouteString(value, field string, allowEmpty bool) error { | ||
| 11 | - | if !allowEmpty && value == "" { | 11 | + | if err := ValidateRouteLabel(value, allowEmpty); err != nil { |
| 12 | - | return fmt.Errorf("%s is empty", field) | 12 | + | return fmt.Errorf("%s: %w", field, err) |
| 13 | - | } | |||
| 14 | - | if len(value) > RouteLabelMaxBytes { | |||
| 15 | - | return fmt.Errorf("%s is too long", field) | |||
| 16 | } | 13 | } | ||
| 17 | return nil | 14 | return nil | ||
| 18 | } | 15 | } | ||
| @@ -237,6 +234,9 @@ func validateRecentRouteSnapshot(snapshot RecentRouteSnapshot) error { | |||||
| 237 | } | 234 | } | ||
| 238 | refs := make(map[RouteRef]struct{}, len(snapshot.Entries)) | 235 | refs := make(map[RouteRef]struct{}, len(snapshot.Entries)) | ||
| 239 | for _, entry := range snapshot.Entries { | 236 | for _, entry := range snapshot.Entries { | ||
| 237 | + | if snapshot.Active == (RouteRef{Key: entry.Key, Generation: entry.Generation}) { | |||
| 238 | + | return fmt.Errorf("%w: active route must be metadata-only", ErrInvalidRouteWire) | |||
| 239 | + | } | |||
| 240 | if err := validateRecentRouteEntry(entry); err != nil { | 240 | if err := validateRecentRouteEntry(entry); err != nil { | ||
| 241 | return err | 241 | return err | ||
| 242 | } | 242 | } | ||
| @@ -246,12 +246,18 @@ func validateRecentRouteSnapshot(snapshot RecentRouteSnapshot) error { | |||||
| 246 | } | 246 | } | ||
| 247 | refs[ref] = struct{}{} | 247 | refs[ref] = struct{}{} | ||
| 248 | } | 248 | } | ||
| 249 | - | for _, ref := range []RouteRef{snapshot.Active, snapshot.Previous, snapshot.Home} { | 249 | + | for name, ref := range map[string]RouteRef{ |
| 250 | - | if ref.empty() { | 250 | + | "active": snapshot.Active, |
| 251 | + | "previous": snapshot.Previous, | |||
| 252 | + | "home": snapshot.Home, | |||
| 253 | + | } { | |||
| 254 | + | if ref.empty() || ref == snapshot.Active { | |||
| 255 | + | // Active is metadata-only and intentionally excluded from the | |||
| 256 | + | // recent display entries. Home may point at active as well. | |||
| 251 | continue | 257 | continue | ||
| 252 | } | 258 | } | ||
| 253 | if _, exists := refs[ref]; !exists { | 259 | if _, exists := refs[ref]; !exists { | ||
| 254 | - | return fmt.Errorf("%w: route reference is absent from snapshot", ErrInvalidRouteWire) | 260 | + | return fmt.Errorf("%w: %s route reference is absent from snapshot", ErrInvalidRouteWire, name) |
| 255 | } | 261 | } | ||
| 256 | } | 262 | } | ||
| 257 | return nil | 263 | return nil | ||
| @@ -357,7 +363,7 @@ func UnmarshalRecentRouteSnapshot(b []byte) (RecentRouteSnapshot, error) { | |||||
| 357 | } | 363 | } | ||
| 358 | 364 | ||||
| 359 | func validateRouteNavigationAction(action RouteNavigationAction) error { | 365 | func validateRouteNavigationAction(action RouteNavigationAction) error { | ||
| 360 | - | if action.Key == 0 || action.Generation == 0 { | 366 | + | if action.SnapshotGeneration == 0 || action.Key == 0 || action.Generation == 0 { |
| 361 | return fmt.Errorf("%w: navigation action identity is zero", ErrInvalidRouteWire) | 367 | return fmt.Errorf("%w: navigation action identity is zero", ErrInvalidRouteWire) | ||
| 362 | } | 368 | } | ||
| 363 | return nil | 369 | return nil | ||
| @@ -368,6 +374,7 @@ func MarshalRouteNavigationAction(action RouteNavigationAction) ([]byte, error) | |||||
| 368 | return nil, err | 374 | return nil, err | ||
| 369 | } | 375 | } | ||
| 370 | w := payloadWriter{} | 376 | w := payloadWriter{} | ||
| 377 | + | w.putUint64(action.SnapshotGeneration) | |||
| 371 | w.putUint64(action.Key) | 378 | w.putUint64(action.Key) | ||
| 372 | w.putUint64(action.Generation) | 379 | w.putUint64(action.Generation) | ||
| 373 | return w.b, nil | 380 | return w.b, nil | ||
| @@ -377,6 +384,9 @@ func UnmarshalRouteNavigationAction(b []byte) (RouteNavigationAction, error) { | |||||
| 377 | r := payloadReader{b: b} | 384 | r := payloadReader{b: b} | ||
| 378 | var action RouteNavigationAction | 385 | var action RouteNavigationAction | ||
| 379 | var err error | 386 | var err error | ||
| 387 | + | if action.SnapshotGeneration, err = r.getUint64(); err != nil { | |||
| 388 | + | return RouteNavigationAction{}, err | |||
| 389 | + | } | |||
| 380 | if action.Key, err = r.getUint64(); err != nil { | 390 | if action.Key, err = r.getUint64(); err != nil { | ||
| 381 | return RouteNavigationAction{}, err | 391 | return RouteNavigationAction{}, err | ||
| 382 | } | 392 | } | ||
M internal/usecase/client/client.go +82 -13
| @@ -165,7 +165,8 @@ type Dependencies struct { | |||
| 165 | 165 | Remote bool | |
| 166 | 166 | // Origin is explicit composition metadata for the initial route. A zero | |
| 167 | 167 | // value is normalized to local for direct package callers and old tests. | |
| 168 | - | Origin ports.RouteOrigin | |
| 168 | + | Origin ports.RouteOrigin | |
| 169 | + | OriginKey string | |
| 169 | 170 | } | |
| 170 | 171 | ||
| 171 | 172 | // AttachRequest identifies the session for one client run. Remote is | |
| @@ -176,6 +177,7 @@ type AttachRequest struct { | |||
| 176 | 177 | SessionName string | |
| 177 | 178 | Remote bool | |
| 178 | 179 | Origin ports.RouteOrigin | |
| 180 | + | OriginKey string | |
| 179 | 181 | RemoteTarget *domain.RemoteSessionTarget | |
| 180 | 182 | ExactTarget *ports.ExactSessionTarget | |
| 181 | 183 | EnvironmentPolicy ports.EnvironmentPolicy | |
| @@ -250,6 +252,11 @@ func validateAttachRequest(request AttachRequest) error { | |||
| 250 | 252 | return fmt.Errorf("vev: invalid route origin: %w", err) | |
| 251 | 253 | } | |
| 252 | 254 | } | |
| 255 | + | if request.OriginKey != "" { | |
| 256 | + | if err := ports.ValidateRouteLabel(request.OriginKey, false); err != nil { | |
| 257 | + | return fmt.Errorf("vev: invalid route origin key: %w", err) | |
| 258 | + | } | |
| 259 | + | } | |
| 253 | 260 | if request.ExactTarget != nil { | |
| 254 | 261 | if err := request.ExactTarget.Validate(); err != nil { | |
| 255 | 262 | return fmt.Errorf("vev: invalid exact target: %w", err) | |
| @@ -484,6 +491,7 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||
| 484 | 491 | } | |
| 485 | 492 | result := (&attachAttempt{ | |
| 486 | 493 | runner: r, | |
| 494 | + | dialer: dialer, | |
| 487 | 495 | connection: connection, | |
| 488 | 496 | remote: remote, | |
| 489 | 497 | handshakeCtx: handshakeCtx, | |
| @@ -505,18 +513,6 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||
| 505 | 513 | }).run(ctx) | |
| 506 | 514 | stopHandshakeTransport() | |
| 507 | 515 | finishHandshake() | |
| 508 | - | if result.welcomed && result.err == nil && result.committedIdentity != nil && r.ledger != nil { | |
| 509 | - | if attemptRequest.ExactTarget != nil && *attemptRequest.ExactTarget != result.committedIdentity.Target { | |
| 510 | - | return errRouteTargetChanged | |
| 511 | - | } | |
| 512 | - | if attemptRequest.RemoteTarget != nil && (attemptRequest.RemoteTarget.LifecycleID != result.committedIdentity.Target.LifecycleID || attemptRequest.RemoteTarget.SessionName != result.committedIdentity.Target.SessionName) { | |
| 513 | - | return errRouteTargetChanged | |
| 514 | - | } | |
| 515 | - | home := r.ledger.homeRef().Key == 0 | |
| 516 | - | if _, err := r.ledger.commit(routeCandidateForAttach(attemptRequest, *result.committedIdentity, dialer, result.resumeToken, home)); err != nil { | |
| 517 | - | return fmt.Errorf("vev: committing route identity: %w", err) | |
| 518 | - | } | |
| 519 | - | } | |
| 520 | 516 | if result.welcomed { | |
| 521 | 517 | backoff = defaultReconnectBackoff.initial | |
| 522 | 518 | } | |
| @@ -784,6 +780,7 @@ func sleepReconnectWithResizeEvents(ctx context.Context, clk ports.Clock, d time | |||
| 784 | 780 | ||
| 785 | 781 | type attachAttempt struct { | |
| 786 | 782 | runner *Runner | |
| 783 | + | dialer ports.Dialer | |
| 787 | 784 | connection *SessionConnection | |
| 788 | 785 | remote bool | |
| 789 | 786 | transport ports.Transport | |
| @@ -998,6 +995,27 @@ func (a *attachAttempt) run(ctx context.Context) attachResult { | |||
| 998 | 995 | return handshakeFailure("validating welcome", err) | |
| 999 | 996 | } | |
| 1000 | 997 | welcomedResult := func(err error) attachResult { return result(true, err) } | |
| 998 | + | if committedIdentity != nil && a.runner.ledger != nil { | |
| 999 | + | if request.ExactTarget != nil && *request.ExactTarget != committedIdentity.Target { | |
| 1000 | + | return welcomedResult(errRouteTargetChanged) | |
| 1001 | + | } | |
| 1002 | + | if request.RemoteTarget != nil && (request.RemoteTarget.LifecycleID != committedIdentity.Target.LifecycleID || request.RemoteTarget.SessionName != committedIdentity.Target.SessionName) { | |
| 1003 | + | return welcomedResult(errRouteTargetChanged) | |
| 1004 | + | } | |
| 1005 | + | home := a.runner.ledger.homeRef().Key == 0 | |
| 1006 | + | if _, err := a.runner.ledger.commit(routeCandidateForAttach(request, *committedIdentity, a.dialer, resumeToken, home)); err != nil { | |
| 1007 | + | return welcomedResult(fmt.Errorf("vev: committing route identity: %w", err)) | |
| 1008 | + | } | |
| 1009 | + | payload, err := ports.MarshalRecentRouteSnapshot(a.runner.ledger.snapshot()) | |
| 1010 | + | if err != nil { | |
| 1011 | + | return welcomedResult(fmt.Errorf("vev: encoding route snapshot: %w", err)) | |
| 1012 | + | } | |
| 1013 | + | if err := sendHandshake(func() error { | |
| 1014 | + | return transport.Send(ports.Frame{Type: ports.MsgRecentRouteSnapshot, Payload: payload}) | |
| 1015 | + | }); err != nil { | |
| 1016 | + | return welcomedResult(fmt.Errorf("vev: publishing route snapshot: %w", err)) | |
| 1017 | + | } | |
| 1018 | + | } | |
| 1001 | 1019 | ||
| 1002 | 1020 | // 3. Enter raw mode after Welcome; Run owns restoration. | |
| 1003 | 1021 | if err := enterRaw(); err != nil { | |
| @@ -1209,6 +1227,22 @@ func (a *attachAttempt) run(ctx context.Context) attachResult { | |||
| 1209 | 1227 | return context.Canceled | |
| 1210 | 1228 | } | |
| 1211 | 1229 | } | |
| 1230 | + | sendRouteFailure := func(action ports.RouteNavigationAction, code ports.RouteFailureCode) error { | |
| 1231 | + | payload, err := ports.MarshalRouteNavigationFailure(ports.RouteNavigationFailure{ | |
| 1232 | + | Key: action.Key, | |
| 1233 | + | Generation: action.Generation, | |
| 1234 | + | Code: code, | |
| 1235 | + | }) | |
| 1236 | + | if err != nil { | |
| 1237 | + | return err | |
| 1238 | + | } | |
| 1239 | + | select { | |
| 1240 | + | case sendCh <- ports.Frame{Type: ports.MsgRouteNavigationFailure, Payload: payload}: | |
| 1241 | + | return nil | |
| 1242 | + | case <-loopCtx.Done(): | |
| 1243 | + | return context.Canceled | |
| 1244 | + | } | |
| 1245 | + | } | |
| 1212 | 1246 | dismissReconnect := func() error { | |
| 1213 | 1247 | if !reconnect.showing { | |
| 1214 | 1248 | return nil | |
| @@ -1372,6 +1406,41 @@ func (a *attachAttempt) run(ctx context.Context) attachResult { | |||
| 1372 | 1406 | navigation := welcomedResult(nil) | |
| 1373 | 1407 | navigation.action = action | |
| 1374 | 1408 | return navigation | |
| 1409 | + | case ports.MsgNavigateRecentRoute: | |
| 1410 | + | action, derr := ports.UnmarshalRouteNavigationAction(r.frame.Payload) | |
| 1411 | + | if derr != nil { | |
| 1412 | + | return welcomedResult(fmt.Errorf("vev: decoding route navigation action: %w", derr)) | |
| 1413 | + | } | |
| 1414 | + | if a.runner.ledger == nil { | |
| 1415 | + | if ferr := sendRouteFailure(action, ports.RouteFailureUnavailable); ferr != nil { | |
| 1416 | + | return welcomedResult(ferr) | |
| 1417 | + | } | |
| 1418 | + | continue | |
| 1419 | + | } | |
| 1420 | + | _, valid, noOp := a.runner.ledger.navigationRecord(action) | |
| 1421 | + | if !valid { | |
| 1422 | + | if ferr := sendRouteFailure(action, ports.RouteFailureStaleSelection); ferr != nil { | |
| 1423 | + | return welcomedResult(ferr) | |
| 1424 | + | } | |
| 1425 | + | continue | |
| 1426 | + | } | |
| 1427 | + | if noOp { | |
| 1428 | + | continue | |
| 1429 | + | } | |
| 1430 | + | // Stack 2 validates and rejects stale actions without changing | |
| 1431 | + | // command authority. The live handoff connector is installed by the | |
| 1432 | + | // atomic Stack 3 cutover. | |
| 1433 | + | if ferr := sendRouteFailure(action, ports.RouteFailureUnavailable); ferr != nil { | |
| 1434 | + | return welcomedResult(ferr) | |
| 1435 | + | } | |
| 1436 | + | case ports.MsgCommittedRouteIdentity: | |
| 1437 | + | if _, derr := ports.UnmarshalCommittedRouteIdentity(r.frame.Payload); derr != nil { | |
| 1438 | + | return welcomedResult(fmt.Errorf("vev: decoding committed route identity: %w", derr)) | |
| 1439 | + | } | |
| 1440 | + | case ports.MsgRouteNavigationFailure: | |
| 1441 | + | if _, derr := ports.UnmarshalRouteNavigationFailure(r.frame.Payload); derr != nil { | |
| 1442 | + | return welcomedResult(fmt.Errorf("vev: decoding route navigation failure: %w", derr)) | |
| 1443 | + | } | |
| 1375 | 1444 | case ports.MsgDetached: | |
| 1376 | 1445 | d, derr := ports.UnmarshalDetached(r.frame.Payload) | |
| 1377 | 1446 | if derr != nil { | |
| @@ -165,7 +165,8 @@ type Dependencies struct { | |||||
| 165 | Remote bool | 165 | Remote bool | ||
| 166 | // Origin is explicit composition metadata for the initial route. A zero | 166 | // Origin is explicit composition metadata for the initial route. A zero | ||
| 167 | // value is normalized to local for direct package callers and old tests. | 167 | // value is normalized to local for direct package callers and old tests. | ||
| 168 | - | Origin ports.RouteOrigin | 168 | + | Origin ports.RouteOrigin |
| 169 | + | OriginKey string | |||
| 169 | } | 170 | } | ||
| 170 | 171 | ||||
| 171 | // AttachRequest identifies the session for one client run. Remote is | 172 | // AttachRequest identifies the session for one client run. Remote is | ||
| @@ -176,6 +177,7 @@ type AttachRequest struct { | |||||
| 176 | SessionName string | 177 | SessionName string | ||
| 177 | Remote bool | 178 | Remote bool | ||
| 178 | Origin ports.RouteOrigin | 179 | Origin ports.RouteOrigin | ||
| 180 | + | OriginKey string | |||
| 179 | RemoteTarget *domain.RemoteSessionTarget | 181 | RemoteTarget *domain.RemoteSessionTarget | ||
| 180 | ExactTarget *ports.ExactSessionTarget | 182 | ExactTarget *ports.ExactSessionTarget | ||
| 181 | EnvironmentPolicy ports.EnvironmentPolicy | 183 | EnvironmentPolicy ports.EnvironmentPolicy | ||
| @@ -250,6 +252,11 @@ func validateAttachRequest(request AttachRequest) error { | |||||
| 250 | return fmt.Errorf("vev: invalid route origin: %w", err) | 252 | return fmt.Errorf("vev: invalid route origin: %w", err) | ||
| 251 | } | 253 | } | ||
| 252 | } | 254 | } | ||
| 255 | + | if request.OriginKey != "" { | |||
| 256 | + | if err := ports.ValidateRouteLabel(request.OriginKey, false); err != nil { | |||
| 257 | + | return fmt.Errorf("vev: invalid route origin key: %w", err) | |||
| 258 | + | } | |||
| 259 | + | } | |||
| 253 | if request.ExactTarget != nil { | 260 | if request.ExactTarget != nil { | ||
| 254 | if err := request.ExactTarget.Validate(); err != nil { | 261 | if err := request.ExactTarget.Validate(); err != nil { | ||
| 255 | return fmt.Errorf("vev: invalid exact target: %w", err) | 262 | return fmt.Errorf("vev: invalid exact target: %w", err) | ||
| @@ -484,6 +491,7 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||||
| 484 | } | 491 | } | ||
| 485 | result := (&attachAttempt{ | 492 | result := (&attachAttempt{ | ||
| 486 | runner: r, | 493 | runner: r, | ||
| 494 | + | dialer: dialer, | |||
| 487 | connection: connection, | 495 | connection: connection, | ||
| 488 | remote: remote, | 496 | remote: remote, | ||
| 489 | handshakeCtx: handshakeCtx, | 497 | handshakeCtx: handshakeCtx, | ||
| @@ -505,18 +513,6 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||||
| 505 | }).run(ctx) | 513 | }).run(ctx) | ||
| 506 | stopHandshakeTransport() | 514 | stopHandshakeTransport() | ||
| 507 | finishHandshake() | 515 | finishHandshake() | ||
| 508 | - | if result.welcomed && result.err == nil && result.committedIdentity != nil && r.ledger != nil { | |||
| 509 | - | if attemptRequest.ExactTarget != nil && *attemptRequest.ExactTarget != result.committedIdentity.Target { | |||
| 510 | - | return errRouteTargetChanged | |||
| 511 | - | } | |||
| 512 | - | if attemptRequest.RemoteTarget != nil && (attemptRequest.RemoteTarget.LifecycleID != result.committedIdentity.Target.LifecycleID || attemptRequest.RemoteTarget.SessionName != result.committedIdentity.Target.SessionName) { | |||
| 513 | - | return errRouteTargetChanged | |||
| 514 | - | } | |||
| 515 | - | home := r.ledger.homeRef().Key == 0 | |||
| 516 | - | if _, err := r.ledger.commit(routeCandidateForAttach(attemptRequest, *result.committedIdentity, dialer, result.resumeToken, home)); err != nil { | |||
| 517 | - | return fmt.Errorf("vev: committing route identity: %w", err) | |||
| 518 | - | } | |||
| 519 | - | } | |||
| 520 | if result.welcomed { | 516 | if result.welcomed { | ||
| 521 | backoff = defaultReconnectBackoff.initial | 517 | backoff = defaultReconnectBackoff.initial | ||
| 522 | } | 518 | } | ||
| @@ -784,6 +780,7 @@ func sleepReconnectWithResizeEvents(ctx context.Context, clk ports.Clock, d time | |||||
| 784 | 780 | ||||
| 785 | type attachAttempt struct { | 781 | type attachAttempt struct { | ||
| 786 | runner *Runner | 782 | runner *Runner | ||
| 783 | + | dialer ports.Dialer | |||
| 787 | connection *SessionConnection | 784 | connection *SessionConnection | ||
| 788 | remote bool | 785 | remote bool | ||
| 789 | transport ports.Transport | 786 | transport ports.Transport | ||
| @@ -998,6 +995,27 @@ func (a *attachAttempt) run(ctx context.Context) attachResult { | |||||
| 998 | return handshakeFailure("validating welcome", err) | 995 | return handshakeFailure("validating welcome", err) | ||
| 999 | } | 996 | } | ||
| 1000 | welcomedResult := func(err error) attachResult { return result(true, err) } | 997 | welcomedResult := func(err error) attachResult { return result(true, err) } | ||
| 998 | + | if committedIdentity != nil && a.runner.ledger != nil { | |||
| 999 | + | if request.ExactTarget != nil && *request.ExactTarget != committedIdentity.Target { | |||
| 1000 | + | return welcomedResult(errRouteTargetChanged) | |||
| 1001 | + | } | |||
| 1002 | + | if request.RemoteTarget != nil && (request.RemoteTarget.LifecycleID != committedIdentity.Target.LifecycleID || request.RemoteTarget.SessionName != committedIdentity.Target.SessionName) { | |||
| 1003 | + | return welcomedResult(errRouteTargetChanged) | |||
| 1004 | + | } | |||
| 1005 | + | home := a.runner.ledger.homeRef().Key == 0 | |||
| 1006 | + | if _, err := a.runner.ledger.commit(routeCandidateForAttach(request, *committedIdentity, a.dialer, resumeToken, home)); err != nil { | |||
| 1007 | + | return welcomedResult(fmt.Errorf("vev: committing route identity: %w", err)) | |||
| 1008 | + | } | |||
| 1009 | + | payload, err := ports.MarshalRecentRouteSnapshot(a.runner.ledger.snapshot()) | |||
| 1010 | + | if err != nil { | |||
| 1011 | + | return welcomedResult(fmt.Errorf("vev: encoding route snapshot: %w", err)) | |||
| 1012 | + | } | |||
| 1013 | + | if err := sendHandshake(func() error { | |||
| 1014 | + | return transport.Send(ports.Frame{Type: ports.MsgRecentRouteSnapshot, Payload: payload}) | |||
| 1015 | + | }); err != nil { | |||
| 1016 | + | return welcomedResult(fmt.Errorf("vev: publishing route snapshot: %w", err)) | |||
| 1017 | + | } | |||
| 1018 | + | } | |||
| 1001 | 1019 | ||||
| 1002 | // 3. Enter raw mode after Welcome; Run owns restoration. | 1020 | // 3. Enter raw mode after Welcome; Run owns restoration. | ||
| 1003 | if err := enterRaw(); err != nil { | 1021 | if err := enterRaw(); err != nil { | ||
| @@ -1209,6 +1227,22 @@ func (a *attachAttempt) run(ctx context.Context) attachResult { | |||||
| 1209 | return context.Canceled | 1227 | return context.Canceled | ||
| 1210 | } | 1228 | } | ||
| 1211 | } | 1229 | } | ||
| 1230 | + | sendRouteFailure := func(action ports.RouteNavigationAction, code ports.RouteFailureCode) error { | |||
| 1231 | + | payload, err := ports.MarshalRouteNavigationFailure(ports.RouteNavigationFailure{ | |||
| 1232 | + | Key: action.Key, | |||
| 1233 | + | Generation: action.Generation, | |||
| 1234 | + | Code: code, | |||
| 1235 | + | }) | |||
| 1236 | + | if err != nil { | |||
| 1237 | + | return err | |||
| 1238 | + | } | |||
| 1239 | + | select { | |||
| 1240 | + | case sendCh <- ports.Frame{Type: ports.MsgRouteNavigationFailure, Payload: payload}: | |||
| 1241 | + | return nil | |||
| 1242 | + | case <-loopCtx.Done(): | |||
| 1243 | + | return context.Canceled | |||
| 1244 | + | } | |||
| 1245 | + | } | |||
| 1212 | dismissReconnect := func() error { | 1246 | dismissReconnect := func() error { | ||
| 1213 | if !reconnect.showing { | 1247 | if !reconnect.showing { | ||
| 1214 | return nil | 1248 | return nil | ||
| @@ -1372,6 +1406,41 @@ func (a *attachAttempt) run(ctx context.Context) attachResult { | |||||
| 1372 | navigation := welcomedResult(nil) | 1406 | navigation := welcomedResult(nil) | ||
| 1373 | navigation.action = action | 1407 | navigation.action = action | ||
| 1374 | return navigation | 1408 | return navigation | ||
| 1409 | + | case ports.MsgNavigateRecentRoute: | |||
| 1410 | + | action, derr := ports.UnmarshalRouteNavigationAction(r.frame.Payload) | |||
| 1411 | + | if derr != nil { | |||
| 1412 | + | return welcomedResult(fmt.Errorf("vev: decoding route navigation action: %w", derr)) | |||
| 1413 | + | } | |||
| 1414 | + | if a.runner.ledger == nil { | |||
| 1415 | + | if ferr := sendRouteFailure(action, ports.RouteFailureUnavailable); ferr != nil { | |||
| 1416 | + | return welcomedResult(ferr) | |||
| 1417 | + | } | |||
| 1418 | + | continue | |||
| 1419 | + | } | |||
| 1420 | + | _, valid, noOp := a.runner.ledger.navigationRecord(action) | |||
| 1421 | + | if !valid { | |||
| 1422 | + | if ferr := sendRouteFailure(action, ports.RouteFailureStaleSelection); ferr != nil { | |||
| 1423 | + | return welcomedResult(ferr) | |||
| 1424 | + | } | |||
| 1425 | + | continue | |||
| 1426 | + | } | |||
| 1427 | + | if noOp { | |||
| 1428 | + | continue | |||
| 1429 | + | } | |||
| 1430 | + | // Stack 2 validates and rejects stale actions without changing | |||
| 1431 | + | // command authority. The live handoff connector is installed by the | |||
| 1432 | + | // atomic Stack 3 cutover. | |||
| 1433 | + | if ferr := sendRouteFailure(action, ports.RouteFailureUnavailable); ferr != nil { | |||
| 1434 | + | return welcomedResult(ferr) | |||
| 1435 | + | } | |||
| 1436 | + | case ports.MsgCommittedRouteIdentity: | |||
| 1437 | + | if _, derr := ports.UnmarshalCommittedRouteIdentity(r.frame.Payload); derr != nil { | |||
| 1438 | + | return welcomedResult(fmt.Errorf("vev: decoding committed route identity: %w", derr)) | |||
| 1439 | + | } | |||
| 1440 | + | case ports.MsgRouteNavigationFailure: | |||
| 1441 | + | if _, derr := ports.UnmarshalRouteNavigationFailure(r.frame.Payload); derr != nil { | |||
| 1442 | + | return welcomedResult(fmt.Errorf("vev: decoding route navigation failure: %w", derr)) | |||
| 1443 | + | } | |||
| 1375 | case ports.MsgDetached: | 1444 | case ports.MsgDetached: | ||
| 1376 | d, derr := ports.UnmarshalDetached(r.frame.Payload) | 1445 | d, derr := ports.UnmarshalDetached(r.frame.Payload) | ||
| 1377 | if derr != nil { | 1446 | if derr != nil { | ||
M internal/usecase/client/client_test.go +40 -0
| @@ -414,6 +414,46 @@ func TestAttachHelloIncludesTrueColor(t *testing.T) { | |||
| 414 | 414 | } | |
| 415 | 415 | } | |
| 416 | 416 | ||
| 417 | + | func TestAttachPublishesCommittedRouteSnapshot(t *testing.T) { | |
| 418 | + | term := newRunTerminal() | |
| 419 | + | defer term.in.unblock() | |
| 420 | + | lifecycle := domain.SessionLifecycleID{9, 8, 7} | |
| 421 | + | tr := &recordingTransport{recvs: []recvItem{ | |
| 422 | + | {f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{ | |
| 423 | + | SessionID: "s1", | |
| 424 | + | SessionName: "main", | |
| 425 | + | CommittedIdentity: &ports.CommittedRouteIdentity{ | |
| 426 | + | Target: ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: "main"}, | |
| 427 | + | }, | |
| 428 | + | }))}, | |
| 429 | + | {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))}, | |
| 430 | + | }} | |
| 431 | + | ||
| 432 | + | err := runTestClient(context.Background(), attachTestDependencies(tr, term, realClock{}), client.AttachRequest{ | |
| 433 | + | Intent: ports.IntentAttach, | |
| 434 | + | SessionName: "main", | |
| 435 | + | Origin: ports.RouteOriginLocal, | |
| 436 | + | OriginKey: "local", | |
| 437 | + | }) | |
| 438 | + | require.NoError(t, err) | |
| 439 | + | ||
| 440 | + | var snapshot ports.RecentRouteSnapshot | |
| 441 | + | found := false | |
| 442 | + | for _, frame := range tr.sends { | |
| 443 | + | if frame.Type != ports.MsgRecentRouteSnapshot { | |
| 444 | + | continue | |
| 445 | + | } | |
| 446 | + | var err error | |
| 447 | + | snapshot, err = ports.UnmarshalRecentRouteSnapshot(frame.Payload) | |
| 448 | + | require.NoError(t, err) | |
| 449 | + | found = true | |
| 450 | + | } | |
| 451 | + | require.True(t, found, "successful Welcome must publish a route snapshot") | |
| 452 | + | require.NotZero(t, snapshot.Generation) | |
| 453 | + | require.NotZero(t, snapshot.Active.Key) | |
| 454 | + | require.Empty(t, snapshot.Entries, "the active route is metadata-only") | |
| 455 | + | } | |
| 456 | + | ||
| 417 | 457 | func TestAttachHelloPreservesCompleteAttachRequest(t *testing.T) { | |
| 418 | 458 | term := newRunTerminal() | |
| 419 | 459 | defer term.in.unblock() | |
| @@ -414,6 +414,46 @@ func TestAttachHelloIncludesTrueColor(t *testing.T) { | |||||
| 414 | } | 414 | } | ||
| 415 | } | 415 | } | ||
| 416 | 416 | ||||
| 417 | + | func TestAttachPublishesCommittedRouteSnapshot(t *testing.T) { | |||
| 418 | + | term := newRunTerminal() | |||
| 419 | + | defer term.in.unblock() | |||
| 420 | + | lifecycle := domain.SessionLifecycleID{9, 8, 7} | |||
| 421 | + | tr := &recordingTransport{recvs: []recvItem{ | |||
| 422 | + | {f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{ | |||
| 423 | + | SessionID: "s1", | |||
| 424 | + | SessionName: "main", | |||
| 425 | + | CommittedIdentity: &ports.CommittedRouteIdentity{ | |||
| 426 | + | Target: ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: "main"}, | |||
| 427 | + | }, | |||
| 428 | + | }))}, | |||
| 429 | + | {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))}, | |||
| 430 | + | }} | |||
| 431 | + | ||||
| 432 | + | err := runTestClient(context.Background(), attachTestDependencies(tr, term, realClock{}), client.AttachRequest{ | |||
| 433 | + | Intent: ports.IntentAttach, | |||
| 434 | + | SessionName: "main", | |||
| 435 | + | Origin: ports.RouteOriginLocal, | |||
| 436 | + | OriginKey: "local", | |||
| 437 | + | }) | |||
| 438 | + | require.NoError(t, err) | |||
| 439 | + | ||||
| 440 | + | var snapshot ports.RecentRouteSnapshot | |||
| 441 | + | found := false | |||
| 442 | + | for _, frame := range tr.sends { | |||
| 443 | + | if frame.Type != ports.MsgRecentRouteSnapshot { | |||
| 444 | + | continue | |||
| 445 | + | } | |||
| 446 | + | var err error | |||
| 447 | + | snapshot, err = ports.UnmarshalRecentRouteSnapshot(frame.Payload) | |||
| 448 | + | require.NoError(t, err) | |||
| 449 | + | found = true | |||
| 450 | + | } | |||
| 451 | + | require.True(t, found, "successful Welcome must publish a route snapshot") | |||
| 452 | + | require.NotZero(t, snapshot.Generation) | |||
| 453 | + | require.NotZero(t, snapshot.Active.Key) | |||
| 454 | + | require.Empty(t, snapshot.Entries, "the active route is metadata-only") | |||
| 455 | + | } | |||
| 456 | + | ||||
| 417 | func TestAttachHelloPreservesCompleteAttachRequest(t *testing.T) { | 457 | func TestAttachHelloPreservesCompleteAttachRequest(t *testing.T) { | ||
| 418 | term := newRunTerminal() | 458 | term := newRunTerminal() | ||
| 419 | defer term.in.unblock() | 459 | defer term.in.unblock() | ||
M internal/usecase/client/routes.go +96 -30
| @@ -9,7 +9,10 @@ import ( | |||
| 9 | 9 | "github.com/bnema/vev/internal/ports" | |
| 10 | 10 | ) | |
| 11 | 11 | ||
| 12 | - | const maxRouteLedgerEntries = ports.RouteSnapshotMaxEntries | |
| 12 | + | // maxRouteLedgerEntries is the product history cap. It is deliberately | |
| 13 | + | // smaller than the wire parser bound so an active metadata entry can be | |
| 14 | + | // excluded without allowing an oversized client history. | |
| 15 | + | const maxRouteLedgerEntries = 20 | |
| 13 | 16 | ||
| 14 | 17 | var ( | |
| 15 | 18 | errRouteResumeUnavailable = errors.New("route resume unavailable") | |
| @@ -18,7 +21,9 @@ var ( | |||
| 18 | 21 | ) | |
| 19 | 22 | ||
| 20 | 23 | // routeKey is process-local identity. It is never derived from a daemon | |
| 21 | - | // lifecycle ID and is not exposed outside the client ledger. | |
| 24 | + | // lifecycle ID and is not exposed outside the client ledger. originKey keeps | |
| 25 | + | // distinct daemon endpoints separate even when their lifecycle/name pairs | |
| 26 | + | // happen to match; neither is serialized into the route snapshot. | |
| 22 | 27 | type routeKey uint64 | |
| 23 | 28 | ||
| 24 | 29 | type routeGeneration uint64 | |
| @@ -52,6 +57,7 @@ type routePresentation struct { | |||
| 52 | 57 | ||
| 53 | 58 | type routeCandidate struct { | |
| 54 | 59 | origin ports.RouteOrigin | |
| 60 | + | originKey string | |
| 55 | 61 | target ports.ExactSessionTarget | |
| 56 | 62 | presentation routePresentation | |
| 57 | 63 | dialer ports.Dialer | |
| @@ -63,6 +69,7 @@ type routeCandidate struct { | |||
| 63 | 69 | type routeRecord struct { | |
| 64 | 70 | identity routeIdentity | |
| 65 | 71 | origin ports.RouteOrigin | |
| 72 | + | originKey string | |
| 66 | 73 | target ports.ExactSessionTarget | |
| 67 | 74 | presentation routePresentation | |
| 68 | 75 | dialer ports.Dialer | |
| @@ -100,6 +107,7 @@ func cloneCommittedIdentity(identity *ports.CommittedRouteIdentity) *ports.Commi | |||
| 100 | 107 | ||
| 101 | 108 | func routeCandidateForAttach(request AttachRequest, identity ports.CommittedRouteIdentity, dialer ports.Dialer, resumeToken uint64, home bool) routeCandidate { | |
| 102 | 109 | origin := normalizeRouteOrigin(request.Origin, request.Remote) | |
| 110 | + | originKey := normalizeRouteOriginKey(request.OriginKey, origin) | |
| 103 | 111 | kind := ports.RouteKindLocal | |
| 104 | 112 | hostLabel := "" | |
| 105 | 113 | if request.Remote || request.RemoteTarget != nil || origin != ports.RouteOriginLocal { | |
| @@ -110,11 +118,13 @@ func routeCandidateForAttach(request AttachRequest, identity ports.CommittedRout | |||
| 110 | 118 | } | |
| 111 | 119 | request = cloneAttachRequest(request) | |
| 112 | 120 | request.Origin = origin | |
| 121 | + | request.OriginKey = originKey | |
| 113 | 122 | request.SessionName = identity.Target.SessionName | |
| 114 | 123 | request.ExactTarget = &identity.Target | |
| 115 | 124 | return routeCandidate{ | |
| 116 | - | origin: origin, | |
| 117 | - | target: identity.Target, | |
| 125 | + | origin: origin, | |
| 126 | + | originKey: originKey, | |
| 127 | + | target: identity.Target, | |
| 118 | 128 | presentation: routePresentation{ | |
| 119 | 129 | name: identity.Target.SessionName, | |
| 120 | 130 | hostLabel: hostLabel, | |
| @@ -139,12 +149,26 @@ func normalizeRouteOrigin(origin ports.RouteOrigin, remote bool) ports.RouteOrig | |||
| 139 | 149 | return ports.RouteOriginLocal | |
| 140 | 150 | } | |
| 141 | 151 | ||
| 152 | + | func normalizeRouteOriginKey(key string, origin ports.RouteOrigin) string { | |
| 153 | + | if key != "" { | |
| 154 | + | return key | |
| 155 | + | } | |
| 156 | + | switch origin { | |
| 157 | + | case ports.RouteOriginLocal: | |
| 158 | + | return "local" | |
| 159 | + | case ports.RouteOriginDiscovery: | |
| 160 | + | return "discovery" | |
| 161 | + | default: | |
| 162 | + | return "remote" | |
| 163 | + | } | |
| 164 | + | } | |
| 165 | + | ||
| 142 | 166 | func validateRoutePresentation(p routePresentation) error { | |
| 143 | - | if p.name == "" { | |
| 144 | - | return errors.New("route name is empty") | |
| 167 | + | if err := ports.ValidateRouteLabel(p.name, false); err != nil { | |
| 168 | + | return err | |
| 145 | 169 | } | |
| 146 | - | if len(p.name) > ports.RouteLabelMaxBytes || len(p.hostLabel) > ports.RouteLabelMaxBytes { | |
| 147 | - | return errors.New("route label is too long") | |
| 170 | + | if err := ports.ValidateRouteLabel(p.hostLabel, true); err != nil { | |
| 171 | + | return err | |
| 148 | 172 | } | |
| 149 | 173 | if err := p.kind.Validate(); err != nil { | |
| 150 | 174 | return err | |
| @@ -164,10 +188,10 @@ func (l *routeLedger) nextIdentityLocked() (routeIdentity, error) { | |||
| 164 | 188 | return routeIdentity{key: l.nextKey, generation: l.generation}, nil | |
| 165 | 189 | } | |
| 166 | 190 | ||
| 167 | - | func (l *routeLedger) findByTargetLocked(origin ports.RouteOrigin, target ports.ExactSessionTarget) int { | |
| 191 | + | func (l *routeLedger) findByTargetLocked(origin ports.RouteOrigin, originKey string, target ports.ExactSessionTarget) int { | |
| 168 | 192 | for i := range l.entries { | |
| 169 | 193 | entry := l.entries[i] | |
| 170 | - | if entry.origin == origin && entry.target == target { | |
| 194 | + | if entry.origin == origin && entry.originKey == originKey && entry.target == target { | |
| 171 | 195 | return i | |
| 172 | 196 | } | |
| 173 | 197 | } | |
| @@ -220,8 +244,10 @@ func (l *routeLedger) commit(candidate routeCandidate) (routeIdentity, error) { | |||
| 220 | 244 | if err := validateRoutePresentation(candidate.presentation); err != nil { | |
| 221 | 245 | return routeIdentity{}, err | |
| 222 | 246 | } | |
| 247 | + | candidate.originKey = normalizeRouteOriginKey(candidate.originKey, candidate.origin) | |
| 223 | 248 | candidate.request = cloneAttachRequest(candidate.request) | |
| 224 | 249 | candidate.request.Origin = candidate.origin | |
| 250 | + | candidate.request.OriginKey = candidate.originKey | |
| 225 | 251 | candidate.request.ExactTarget = &candidate.target | |
| 226 | 252 | ||
| 227 | 253 | l.mu.Lock() | |
| @@ -234,13 +260,14 @@ func (l *routeLedger) commitLocked(candidate routeCandidate) (routeIdentity, err | |||
| 234 | 260 | if err != nil { | |
| 235 | 261 | return routeIdentity{}, err | |
| 236 | 262 | } | |
| 237 | - | index := l.findByTargetLocked(candidate.origin, candidate.target) | |
| 263 | + | index := l.findByTargetLocked(candidate.origin, candidate.originKey, candidate.target) | |
| 238 | 264 | if index >= 0 { | |
| 239 | 265 | identity.key = l.entries[index].identity.key | |
| 240 | 266 | } | |
| 241 | 267 | entry := routeRecord{ | |
| 242 | 268 | identity: identity, | |
| 243 | 269 | origin: candidate.origin, | |
| 270 | + | originKey: candidate.originKey, | |
| 244 | 271 | target: candidate.target, | |
| 245 | 272 | presentation: candidate.presentation, | |
| 246 | 273 | dialer: candidate.dialer, | |
| @@ -316,6 +343,9 @@ func (l *routeLedger) snapshot() ports.RecentRouteSnapshot { | |||
| 316 | 343 | } | |
| 317 | 344 | snapshot.Entries = make([]ports.RecentRouteEntry, 0, len(l.entries)) | |
| 318 | 345 | for _, entry := range l.entries { | |
| 346 | + | if entry.identity == l.active { | |
| 347 | + | continue | |
| 348 | + | } | |
| 319 | 349 | snapshot.Entries = append(snapshot.Entries, ports.RecentRouteEntry{ | |
| 320 | 350 | Key: uint64(entry.identity.key), | |
| 321 | 351 | Generation: uint64(entry.identity.generation), | |
| @@ -352,10 +382,11 @@ func (l *routeLedger) homeRef() ports.RouteRef { | |||
| 352 | 382 | // preserves the original origin and exact target while leaving display labels | |
| 353 | 383 | // and transport capabilities separate. | |
| 354 | 384 | type routeTransitionPlan struct { | |
| 355 | - | identity routeIdentity | |
| 356 | - | record routeRecord | |
| 357 | - | origin ports.RouteOrigin | |
| 358 | - | target ports.ExactSessionTarget | |
| 385 | + | identity routeIdentity | |
| 386 | + | record routeRecord | |
| 387 | + | origin ports.RouteOrigin | |
| 388 | + | originKey string | |
| 389 | + | target ports.ExactSessionTarget | |
| 359 | 390 | } | |
| 360 | 391 | ||
| 361 | 392 | type routeTransitionCommit struct { | |
| @@ -368,15 +399,42 @@ type routeTransitionCommit struct { | |||
| 368 | 399 | type routeTransitionConnector interface { | |
| 369 | 400 | Resume(context.Context, routeRecord) (routeTransitionCommit, error) | |
| 370 | 401 | Attach(context.Context, routeRecord) (routeTransitionCommit, error) | |
| 402 | + | Restore(context.Context, routeRecord) error | |
| 403 | + | } | |
| 404 | + | ||
| 405 | + | // navigationRecord validates that an action belongs to the latest complete | |
| 406 | + | // snapshot. The active route is metadata-only and selecting it is a no-op. | |
| 407 | + | func (l *routeLedger) navigationRecord(action ports.RouteNavigationAction) (routeRecord, bool, bool) { | |
| 408 | + | if action.SnapshotGeneration == 0 || action.Key == 0 || action.Generation == 0 { | |
| 409 | + | return routeRecord{}, false, false | |
| 410 | + | } | |
| 411 | + | identity := identityFromWire(ports.RouteRef{Key: action.Key, Generation: action.Generation}) | |
| 412 | + | l.mu.RLock() | |
| 413 | + | defer l.mu.RUnlock() | |
| 414 | + | if uint64(l.generation) != action.SnapshotGeneration { | |
| 415 | + | return routeRecord{}, false, false | |
| 416 | + | } | |
| 417 | + | if identity == l.active { | |
| 418 | + | return routeRecord{}, true, true | |
| 419 | + | } | |
| 420 | + | index := l.indexByIdentityLocked(identity) | |
| 421 | + | if index < 0 { | |
| 422 | + | return routeRecord{}, false, false | |
| 423 | + | } | |
| 424 | + | return cloneRouteRecord(l.entries[index]), true, false | |
| 371 | 425 | } | |
| 372 | 426 | ||
| 373 | 427 | // navigate resolves a captured key/generation, attempts resume before exact | |
| 374 | 428 | // attach, and commits only after the connector returns a matching identity. | |
| 375 | - | // A failed transition leaves active/previous history unchanged. | |
| 429 | + | // A failed transition restores the prior live route and leaves history alone. | |
| 376 | 430 | func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigationAction, connector routeTransitionConnector) error { | |
| 377 | - | if action.Key == 0 || action.Generation == 0 { | |
| 431 | + | record, ok, noOp := l.navigationRecord(action) | |
| 432 | + | if !ok { | |
| 378 | 433 | return errRouteStaleSelection | |
| 379 | 434 | } | |
| 435 | + | if noOp { | |
| 436 | + | return nil | |
| 437 | + | } | |
| 380 | 438 | if connector == nil { | |
| 381 | 439 | return errors.New("route transition connector is nil") | |
| 382 | 440 | } | |
| @@ -384,15 +442,18 @@ func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigation | |||
| 384 | 442 | ctx = context.Background() | |
| 385 | 443 | } | |
| 386 | 444 | selected := ports.RouteRef{Key: action.Key, Generation: action.Generation} | |
| 387 | - | record, ok := l.lookup(selected) | |
| 388 | - | if !ok { | |
| 389 | - | return errRouteStaleSelection | |
| 390 | - | } | |
| 391 | 445 | plan := routeTransitionPlan{ | |
| 392 | - | identity: identityFromWire(selected), | |
| 393 | - | record: record, | |
| 394 | - | origin: record.origin, | |
| 395 | - | target: record.target, | |
| 446 | + | identity: identityFromWire(selected), | |
| 447 | + | record: record, | |
| 448 | + | origin: record.origin, | |
| 449 | + | originKey: record.originKey, | |
| 450 | + | target: record.target, | |
| 451 | + | } | |
| 452 | + | restoreOnFailure := func(err error) error { | |
| 453 | + | if restoreErr := connector.Restore(ctx, record); restoreErr != nil { | |
| 454 | + | return errors.Join(err, fmt.Errorf("restoring prior route: %w", restoreErr)) | |
| 455 | + | } | |
| 456 | + | return err | |
| 396 | 457 | } | |
| 397 | 458 | ||
| 398 | 459 | var commit routeTransitionCommit | |
| @@ -406,20 +467,21 @@ func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigation | |||
| 406 | 467 | commit, err = connector.Attach(ctx, record) | |
| 407 | 468 | } | |
| 408 | 469 | if err != nil { | |
| 409 | - | return err | |
| 470 | + | return restoreOnFailure(err) | |
| 410 | 471 | } | |
| 411 | 472 | if err := commit.identity.Validate(); err != nil { | |
| 412 | - | return fmt.Errorf("invalid committed route identity: %w", err) | |
| 473 | + | return restoreOnFailure(fmt.Errorf("invalid committed route identity: %w", err)) | |
| 413 | 474 | } | |
| 414 | 475 | if commit.identity.Target != plan.target { | |
| 415 | - | return errRouteTargetChanged | |
| 476 | + | return restoreOnFailure(errRouteTargetChanged) | |
| 416 | 477 | } | |
| 417 | 478 | if err := validateRoutePresentation(commit.presentation); err != nil { | |
| 418 | - | return err | |
| 479 | + | return restoreOnFailure(err) | |
| 419 | 480 | } | |
| 420 | 481 | ||
| 421 | 482 | candidate := routeCandidate{ | |
| 422 | 483 | origin: plan.origin, | |
| 484 | + | originKey: plan.originKey, | |
| 423 | 485 | target: plan.target, | |
| 424 | 486 | presentation: commit.presentation, | |
| 425 | 487 | dialer: commit.dialer, | |
| @@ -428,8 +490,12 @@ func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigation | |||
| 428 | 490 | home: plan.record.home, | |
| 429 | 491 | } | |
| 430 | 492 | candidate.request.Origin = plan.origin | |
| 493 | + | candidate.request.OriginKey = plan.originKey | |
| 431 | 494 | candidate.request.ExactTarget = &candidate.target | |
| 432 | - | return l.commitTransition(plan.identity, candidate) | |
| 495 | + | if err := l.commitTransition(plan.identity, candidate); err != nil { | |
| 496 | + | return restoreOnFailure(err) | |
| 497 | + | } | |
| 498 | + | return nil | |
| 433 | 499 | } | |
| 434 | 500 | ||
| 435 | 501 | func (l *routeLedger) commitTransition(selected routeIdentity, candidate routeCandidate) error { | |
| @@ -9,7 +9,10 @@ import ( | |||||
| 9 | "github.com/bnema/vev/internal/ports" | 9 | "github.com/bnema/vev/internal/ports" | ||
| 10 | ) | 10 | ) | ||
| 11 | 11 | ||||
| 12 | - | const maxRouteLedgerEntries = ports.RouteSnapshotMaxEntries | 12 | + | // maxRouteLedgerEntries is the product history cap. It is deliberately |
| 13 | + | // smaller than the wire parser bound so an active metadata entry can be | |||
| 14 | + | // excluded without allowing an oversized client history. | |||
| 15 | + | const maxRouteLedgerEntries = 20 | |||
| 13 | 16 | ||||
| 14 | var ( | 17 | var ( | ||
| 15 | errRouteResumeUnavailable = errors.New("route resume unavailable") | 18 | errRouteResumeUnavailable = errors.New("route resume unavailable") | ||
| @@ -18,7 +21,9 @@ var ( | |||||
| 18 | ) | 21 | ) | ||
| 19 | 22 | ||||
| 20 | // routeKey is process-local identity. It is never derived from a daemon | 23 | // routeKey is process-local identity. It is never derived from a daemon | ||
| 21 | - | // lifecycle ID and is not exposed outside the client ledger. | 24 | + | // lifecycle ID and is not exposed outside the client ledger. originKey keeps |
| 25 | + | // distinct daemon endpoints separate even when their lifecycle/name pairs | |||
| 26 | + | // happen to match; neither is serialized into the route snapshot. | |||
| 22 | type routeKey uint64 | 27 | type routeKey uint64 | ||
| 23 | 28 | ||||
| 24 | type routeGeneration uint64 | 29 | type routeGeneration uint64 | ||
| @@ -52,6 +57,7 @@ type routePresentation struct { | |||||
| 52 | 57 | ||||
| 53 | type routeCandidate struct { | 58 | type routeCandidate struct { | ||
| 54 | origin ports.RouteOrigin | 59 | origin ports.RouteOrigin | ||
| 60 | + | originKey string | |||
| 55 | target ports.ExactSessionTarget | 61 | target ports.ExactSessionTarget | ||
| 56 | presentation routePresentation | 62 | presentation routePresentation | ||
| 57 | dialer ports.Dialer | 63 | dialer ports.Dialer | ||
| @@ -63,6 +69,7 @@ type routeCandidate struct { | |||||
| 63 | type routeRecord struct { | 69 | type routeRecord struct { | ||
| 64 | identity routeIdentity | 70 | identity routeIdentity | ||
| 65 | origin ports.RouteOrigin | 71 | origin ports.RouteOrigin | ||
| 72 | + | originKey string | |||
| 66 | target ports.ExactSessionTarget | 73 | target ports.ExactSessionTarget | ||
| 67 | presentation routePresentation | 74 | presentation routePresentation | ||
| 68 | dialer ports.Dialer | 75 | dialer ports.Dialer | ||
| @@ -100,6 +107,7 @@ func cloneCommittedIdentity(identity *ports.CommittedRouteIdentity) *ports.Commi | |||||
| 100 | 107 | ||||
| 101 | func routeCandidateForAttach(request AttachRequest, identity ports.CommittedRouteIdentity, dialer ports.Dialer, resumeToken uint64, home bool) routeCandidate { | 108 | func routeCandidateForAttach(request AttachRequest, identity ports.CommittedRouteIdentity, dialer ports.Dialer, resumeToken uint64, home bool) routeCandidate { | ||
| 102 | origin := normalizeRouteOrigin(request.Origin, request.Remote) | 109 | origin := normalizeRouteOrigin(request.Origin, request.Remote) | ||
| 110 | + | originKey := normalizeRouteOriginKey(request.OriginKey, origin) | |||
| 103 | kind := ports.RouteKindLocal | 111 | kind := ports.RouteKindLocal | ||
| 104 | hostLabel := "" | 112 | hostLabel := "" | ||
| 105 | if request.Remote || request.RemoteTarget != nil || origin != ports.RouteOriginLocal { | 113 | if request.Remote || request.RemoteTarget != nil || origin != ports.RouteOriginLocal { | ||
| @@ -110,11 +118,13 @@ func routeCandidateForAttach(request AttachRequest, identity ports.CommittedRout | |||||
| 110 | } | 118 | } | ||
| 111 | request = cloneAttachRequest(request) | 119 | request = cloneAttachRequest(request) | ||
| 112 | request.Origin = origin | 120 | request.Origin = origin | ||
| 121 | + | request.OriginKey = originKey | |||
| 113 | request.SessionName = identity.Target.SessionName | 122 | request.SessionName = identity.Target.SessionName | ||
| 114 | request.ExactTarget = &identity.Target | 123 | request.ExactTarget = &identity.Target | ||
| 115 | return routeCandidate{ | 124 | return routeCandidate{ | ||
| 116 | - | origin: origin, | 125 | + | origin: origin, |
| 117 | - | target: identity.Target, | 126 | + | originKey: originKey, |
| 127 | + | target: identity.Target, | |||
| 118 | presentation: routePresentation{ | 128 | presentation: routePresentation{ | ||
| 119 | name: identity.Target.SessionName, | 129 | name: identity.Target.SessionName, | ||
| 120 | hostLabel: hostLabel, | 130 | hostLabel: hostLabel, | ||
| @@ -139,12 +149,26 @@ func normalizeRouteOrigin(origin ports.RouteOrigin, remote bool) ports.RouteOrig | |||||
| 139 | return ports.RouteOriginLocal | 149 | return ports.RouteOriginLocal | ||
| 140 | } | 150 | } | ||
| 141 | 151 | ||||
| 152 | + | func normalizeRouteOriginKey(key string, origin ports.RouteOrigin) string { | |||
| 153 | + | if key != "" { | |||
| 154 | + | return key | |||
| 155 | + | } | |||
| 156 | + | switch origin { | |||
| 157 | + | case ports.RouteOriginLocal: | |||
| 158 | + | return "local" | |||
| 159 | + | case ports.RouteOriginDiscovery: | |||
| 160 | + | return "discovery" | |||
| 161 | + | default: | |||
| 162 | + | return "remote" | |||
| 163 | + | } | |||
| 164 | + | } | |||
| 165 | + | ||||
| 142 | func validateRoutePresentation(p routePresentation) error { | 166 | func validateRoutePresentation(p routePresentation) error { | ||
| 143 | - | if p.name == "" { | 167 | + | if err := ports.ValidateRouteLabel(p.name, false); err != nil { |
| 144 | - | return errors.New("route name is empty") | 168 | + | return err |
| 145 | } | 169 | } | ||
| 146 | - | if len(p.name) > ports.RouteLabelMaxBytes || len(p.hostLabel) > ports.RouteLabelMaxBytes { | 170 | + | if err := ports.ValidateRouteLabel(p.hostLabel, true); err != nil { |
| 147 | - | return errors.New("route label is too long") | 171 | + | return err |
| 148 | } | 172 | } | ||
| 149 | if err := p.kind.Validate(); err != nil { | 173 | if err := p.kind.Validate(); err != nil { | ||
| 150 | return err | 174 | return err | ||
| @@ -164,10 +188,10 @@ func (l *routeLedger) nextIdentityLocked() (routeIdentity, error) { | |||||
| 164 | return routeIdentity{key: l.nextKey, generation: l.generation}, nil | 188 | return routeIdentity{key: l.nextKey, generation: l.generation}, nil | ||
| 165 | } | 189 | } | ||
| 166 | 190 | ||||
| 167 | - | func (l *routeLedger) findByTargetLocked(origin ports.RouteOrigin, target ports.ExactSessionTarget) int { | 191 | + | func (l *routeLedger) findByTargetLocked(origin ports.RouteOrigin, originKey string, target ports.ExactSessionTarget) int { |
| 168 | for i := range l.entries { | 192 | for i := range l.entries { | ||
| 169 | entry := l.entries[i] | 193 | entry := l.entries[i] | ||
| 170 | - | if entry.origin == origin && entry.target == target { | 194 | + | if entry.origin == origin && entry.originKey == originKey && entry.target == target { |
| 171 | return i | 195 | return i | ||
| 172 | } | 196 | } | ||
| 173 | } | 197 | } | ||
| @@ -220,8 +244,10 @@ func (l *routeLedger) commit(candidate routeCandidate) (routeIdentity, error) { | |||||
| 220 | if err := validateRoutePresentation(candidate.presentation); err != nil { | 244 | if err := validateRoutePresentation(candidate.presentation); err != nil { | ||
| 221 | return routeIdentity{}, err | 245 | return routeIdentity{}, err | ||
| 222 | } | 246 | } | ||
| 247 | + | candidate.originKey = normalizeRouteOriginKey(candidate.originKey, candidate.origin) | |||
| 223 | candidate.request = cloneAttachRequest(candidate.request) | 248 | candidate.request = cloneAttachRequest(candidate.request) | ||
| 224 | candidate.request.Origin = candidate.origin | 249 | candidate.request.Origin = candidate.origin | ||
| 250 | + | candidate.request.OriginKey = candidate.originKey | |||
| 225 | candidate.request.ExactTarget = &candidate.target | 251 | candidate.request.ExactTarget = &candidate.target | ||
| 226 | 252 | ||||
| 227 | l.mu.Lock() | 253 | l.mu.Lock() | ||
| @@ -234,13 +260,14 @@ func (l *routeLedger) commitLocked(candidate routeCandidate) (routeIdentity, err | |||||
| 234 | if err != nil { | 260 | if err != nil { | ||
| 235 | return routeIdentity{}, err | 261 | return routeIdentity{}, err | ||
| 236 | } | 262 | } | ||
| 237 | - | index := l.findByTargetLocked(candidate.origin, candidate.target) | 263 | + | index := l.findByTargetLocked(candidate.origin, candidate.originKey, candidate.target) |
| 238 | if index >= 0 { | 264 | if index >= 0 { | ||
| 239 | identity.key = l.entries[index].identity.key | 265 | identity.key = l.entries[index].identity.key | ||
| 240 | } | 266 | } | ||
| 241 | entry := routeRecord{ | 267 | entry := routeRecord{ | ||
| 242 | identity: identity, | 268 | identity: identity, | ||
| 243 | origin: candidate.origin, | 269 | origin: candidate.origin, | ||
| 270 | + | originKey: candidate.originKey, | |||
| 244 | target: candidate.target, | 271 | target: candidate.target, | ||
| 245 | presentation: candidate.presentation, | 272 | presentation: candidate.presentation, | ||
| 246 | dialer: candidate.dialer, | 273 | dialer: candidate.dialer, | ||
| @@ -316,6 +343,9 @@ func (l *routeLedger) snapshot() ports.RecentRouteSnapshot { | |||||
| 316 | } | 343 | } | ||
| 317 | snapshot.Entries = make([]ports.RecentRouteEntry, 0, len(l.entries)) | 344 | snapshot.Entries = make([]ports.RecentRouteEntry, 0, len(l.entries)) | ||
| 318 | for _, entry := range l.entries { | 345 | for _, entry := range l.entries { | ||
| 346 | + | if entry.identity == l.active { | |||
| 347 | + | continue | |||
| 348 | + | } | |||
| 319 | snapshot.Entries = append(snapshot.Entries, ports.RecentRouteEntry{ | 349 | snapshot.Entries = append(snapshot.Entries, ports.RecentRouteEntry{ | ||
| 320 | Key: uint64(entry.identity.key), | 350 | Key: uint64(entry.identity.key), | ||
| 321 | Generation: uint64(entry.identity.generation), | 351 | Generation: uint64(entry.identity.generation), | ||
| @@ -352,10 +382,11 @@ func (l *routeLedger) homeRef() ports.RouteRef { | |||||
| 352 | // preserves the original origin and exact target while leaving display labels | 382 | // preserves the original origin and exact target while leaving display labels | ||
| 353 | // and transport capabilities separate. | 383 | // and transport capabilities separate. | ||
| 354 | type routeTransitionPlan struct { | 384 | type routeTransitionPlan struct { | ||
| 355 | - | identity routeIdentity | 385 | + | identity routeIdentity |
| 356 | - | record routeRecord | 386 | + | record routeRecord |
| 357 | - | origin ports.RouteOrigin | 387 | + | origin ports.RouteOrigin |
| 358 | - | target ports.ExactSessionTarget | 388 | + | originKey string |
| 389 | + | target ports.ExactSessionTarget | |||
| 359 | } | 390 | } | ||
| 360 | 391 | ||||
| 361 | type routeTransitionCommit struct { | 392 | type routeTransitionCommit struct { | ||
| @@ -368,15 +399,42 @@ type routeTransitionCommit struct { | |||||
| 368 | type routeTransitionConnector interface { | 399 | type routeTransitionConnector interface { | ||
| 369 | Resume(context.Context, routeRecord) (routeTransitionCommit, error) | 400 | Resume(context.Context, routeRecord) (routeTransitionCommit, error) | ||
| 370 | Attach(context.Context, routeRecord) (routeTransitionCommit, error) | 401 | Attach(context.Context, routeRecord) (routeTransitionCommit, error) | ||
| 402 | + | Restore(context.Context, routeRecord) error | |||
| 403 | + | } | |||
| 404 | + | ||||
| 405 | + | // navigationRecord validates that an action belongs to the latest complete | |||
| 406 | + | // snapshot. The active route is metadata-only and selecting it is a no-op. | |||
| 407 | + | func (l *routeLedger) navigationRecord(action ports.RouteNavigationAction) (routeRecord, bool, bool) { | |||
| 408 | + | if action.SnapshotGeneration == 0 || action.Key == 0 || action.Generation == 0 { | |||
| 409 | + | return routeRecord{}, false, false | |||
| 410 | + | } | |||
| 411 | + | identity := identityFromWire(ports.RouteRef{Key: action.Key, Generation: action.Generation}) | |||
| 412 | + | l.mu.RLock() | |||
| 413 | + | defer l.mu.RUnlock() | |||
| 414 | + | if uint64(l.generation) != action.SnapshotGeneration { | |||
| 415 | + | return routeRecord{}, false, false | |||
| 416 | + | } | |||
| 417 | + | if identity == l.active { | |||
| 418 | + | return routeRecord{}, true, true | |||
| 419 | + | } | |||
| 420 | + | index := l.indexByIdentityLocked(identity) | |||
| 421 | + | if index < 0 { | |||
| 422 | + | return routeRecord{}, false, false | |||
| 423 | + | } | |||
| 424 | + | return cloneRouteRecord(l.entries[index]), true, false | |||
| 371 | } | 425 | } | ||
| 372 | 426 | ||||
| 373 | // navigate resolves a captured key/generation, attempts resume before exact | 427 | // navigate resolves a captured key/generation, attempts resume before exact | ||
| 374 | // attach, and commits only after the connector returns a matching identity. | 428 | // attach, and commits only after the connector returns a matching identity. | ||
| 375 | - | // A failed transition leaves active/previous history unchanged. | 429 | + | // A failed transition restores the prior live route and leaves history alone. |
| 376 | func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigationAction, connector routeTransitionConnector) error { | 430 | func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigationAction, connector routeTransitionConnector) error { | ||
| 377 | - | if action.Key == 0 || action.Generation == 0 { | 431 | + | record, ok, noOp := l.navigationRecord(action) |
| 432 | + | if !ok { | |||
| 378 | return errRouteStaleSelection | 433 | return errRouteStaleSelection | ||
| 379 | } | 434 | } | ||
| 435 | + | if noOp { | |||
| 436 | + | return nil | |||
| 437 | + | } | |||
| 380 | if connector == nil { | 438 | if connector == nil { | ||
| 381 | return errors.New("route transition connector is nil") | 439 | return errors.New("route transition connector is nil") | ||
| 382 | } | 440 | } | ||
| @@ -384,15 +442,18 @@ func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigation | |||||
| 384 | ctx = context.Background() | 442 | ctx = context.Background() | ||
| 385 | } | 443 | } | ||
| 386 | selected := ports.RouteRef{Key: action.Key, Generation: action.Generation} | 444 | selected := ports.RouteRef{Key: action.Key, Generation: action.Generation} | ||
| 387 | - | record, ok := l.lookup(selected) | |||
| 388 | - | if !ok { | |||
| 389 | - | return errRouteStaleSelection | |||
| 390 | - | } | |||
| 391 | plan := routeTransitionPlan{ | 445 | plan := routeTransitionPlan{ | ||
| 392 | - | identity: identityFromWire(selected), | 446 | + | identity: identityFromWire(selected), |
| 393 | - | record: record, | 447 | + | record: record, |
| 394 | - | origin: record.origin, | 448 | + | origin: record.origin, |
| 395 | - | target: record.target, | 449 | + | originKey: record.originKey, |
| 450 | + | target: record.target, | |||
| 451 | + | } | |||
| 452 | + | restoreOnFailure := func(err error) error { | |||
| 453 | + | if restoreErr := connector.Restore(ctx, record); restoreErr != nil { | |||
| 454 | + | return errors.Join(err, fmt.Errorf("restoring prior route: %w", restoreErr)) | |||
| 455 | + | } | |||
| 456 | + | return err | |||
| 396 | } | 457 | } | ||
| 397 | 458 | ||||
| 398 | var commit routeTransitionCommit | 459 | var commit routeTransitionCommit | ||
| @@ -406,20 +467,21 @@ func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigation | |||||
| 406 | commit, err = connector.Attach(ctx, record) | 467 | commit, err = connector.Attach(ctx, record) | ||
| 407 | } | 468 | } | ||
| 408 | if err != nil { | 469 | if err != nil { | ||
| 409 | - | return err | 470 | + | return restoreOnFailure(err) |
| 410 | } | 471 | } | ||
| 411 | if err := commit.identity.Validate(); err != nil { | 472 | if err := commit.identity.Validate(); err != nil { | ||
| 412 | - | return fmt.Errorf("invalid committed route identity: %w", err) | 473 | + | return restoreOnFailure(fmt.Errorf("invalid committed route identity: %w", err)) |
| 413 | } | 474 | } | ||
| 414 | if commit.identity.Target != plan.target { | 475 | if commit.identity.Target != plan.target { | ||
| 415 | - | return errRouteTargetChanged | 476 | + | return restoreOnFailure(errRouteTargetChanged) |
| 416 | } | 477 | } | ||
| 417 | if err := validateRoutePresentation(commit.presentation); err != nil { | 478 | if err := validateRoutePresentation(commit.presentation); err != nil { | ||
| 418 | - | return err | 479 | + | return restoreOnFailure(err) |
| 419 | } | 480 | } | ||
| 420 | 481 | ||||
| 421 | candidate := routeCandidate{ | 482 | candidate := routeCandidate{ | ||
| 422 | origin: plan.origin, | 483 | origin: plan.origin, | ||
| 484 | + | originKey: plan.originKey, | |||
| 423 | target: plan.target, | 485 | target: plan.target, | ||
| 424 | presentation: commit.presentation, | 486 | presentation: commit.presentation, | ||
| 425 | dialer: commit.dialer, | 487 | dialer: commit.dialer, | ||
| @@ -428,8 +490,12 @@ func (l *routeLedger) navigate(ctx context.Context, action ports.RouteNavigation | |||||
| 428 | home: plan.record.home, | 490 | home: plan.record.home, | ||
| 429 | } | 491 | } | ||
| 430 | candidate.request.Origin = plan.origin | 492 | candidate.request.Origin = plan.origin | ||
| 493 | + | candidate.request.OriginKey = plan.originKey | |||
| 431 | candidate.request.ExactTarget = &candidate.target | 494 | candidate.request.ExactTarget = &candidate.target | ||
| 432 | - | return l.commitTransition(plan.identity, candidate) | 495 | + | if err := l.commitTransition(plan.identity, candidate); err != nil { |
| 496 | + | return restoreOnFailure(err) | |||
| 497 | + | } | |||
| 498 | + | return nil | |||
| 433 | } | 499 | } | ||
| 434 | 500 | ||||
| 435 | func (l *routeLedger) commitTransition(selected routeIdentity, candidate routeCandidate) error { | 501 | func (l *routeLedger) commitTransition(selected routeIdentity, candidate routeCandidate) error { | ||
M internal/usecase/client/routes_test.go +68 -12
| @@ -45,8 +45,12 @@ func TestRouteLedgerBoundsAndImmutableSnapshot(t *testing.T) { | |||
| 45 | 45 | } | |
| 46 | 46 | ||
| 47 | 47 | snapshot := ledger.snapshot() | |
| 48 | - | require.Len(t, snapshot.Entries, maxRouteLedgerEntries) | |
| 48 | + | require.Len(t, ledger.entries, maxRouteLedgerEntries) | |
| 49 | + | require.Len(t, snapshot.Entries, maxRouteLedgerEntries-1) | |
| 49 | 50 | require.Equal(t, uint64(keys[len(keys)-1]), snapshot.Active.Key) | |
| 51 | + | for _, entry := range snapshot.Entries { | |
| 52 | + | require.NotEqual(t, snapshot.Active, ports.RouteRef{Key: entry.Key, Generation: entry.Generation}) | |
| 53 | + | } | |
| 50 | 54 | require.Equal(t, uint64(keys[len(keys)-2]), snapshot.Previous.Key) | |
| 51 | 55 | require.Equal(t, uint64(keys[0]), snapshot.Home.Key, "home is retained when older entries are evicted") | |
| 52 | 56 | ||
| @@ -91,10 +95,12 @@ func TestRouteLedgerDiscoveryCyclesAreIndependent(t *testing.T) { | |||
| 91 | 95 | } | |
| 92 | 96 | ||
| 93 | 97 | type routeTestConnector struct { | |
| 94 | - | resumeErr error | |
| 95 | - | attachErr error | |
| 96 | - | resumeCalls int | |
| 97 | - | attachCalls int | |
| 98 | + | resumeErr error | |
| 99 | + | attachErr error | |
| 100 | + | resumeCalls int | |
| 101 | + | attachCalls int | |
| 102 | + | restoreCalls int | |
| 103 | + | restoreErr error | |
| 98 | 104 | } | |
| 99 | 105 | ||
| 100 | 106 | func (c *routeTestConnector) Resume(context.Context, routeRecord) (routeTransitionCommit, error) { | |
| @@ -105,6 +111,11 @@ func (c *routeTestConnector) Resume(context.Context, routeRecord) (routeTransiti | |||
| 105 | 111 | return routeTransitionCommit{}, errors.New("unexpected resume success") | |
| 106 | 112 | } | |
| 107 | 113 | ||
| 114 | + | func (c *routeTestConnector) Restore(_ context.Context, _ routeRecord) error { | |
| 115 | + | c.restoreCalls++ | |
| 116 | + | return c.restoreErr | |
| 117 | + | } | |
| 118 | + | ||
| 108 | 119 | func (c *routeTestConnector) Attach(_ context.Context, record routeRecord) (routeTransitionCommit, error) { | |
| 109 | 120 | c.attachCalls++ | |
| 110 | 121 | if c.attachErr != nil { | |
| @@ -124,19 +135,25 @@ func (c *routeTestConnector) Attach(_ context.Context, record routeRecord) (rout | |||
| 124 | 135 | func TestRouteLedgerNavigateResumesThenFallsBackAndRestoresOrigin(t *testing.T) { | |
| 125 | 136 | ledger := newRouteLedger() | |
| 126 | 137 | candidate := routeTestCandidate(2, ports.RouteOriginDiscovery) | |
| 138 | + | candidate.originKey = "daemon-discovered" | |
| 127 | 139 | candidate.resumeToken = 77 | |
| 128 | 140 | identity, err := ledger.commit(candidate) | |
| 129 | 141 | require.NoError(t, err) | |
| 142 | + | _, err = ledger.commit(routeTestCandidate(4, ports.RouteOriginLocal)) | |
| 143 | + | require.NoError(t, err) | |
| 130 | 144 | ||
| 131 | 145 | connector := &routeTestConnector{resumeErr: errRouteResumeUnavailable} | |
| 132 | - | err = ledger.navigate(context.Background(), ports.RouteNavigationAction{Key: uint64(identity.key), Generation: uint64(identity.generation)}, connector) | |
| 146 | + | action := identity.wireNavigationAction(routeGeneration(ledger.snapshot().Generation)) | |
| 147 | + | err = ledger.navigate(context.Background(), action, connector) | |
| 133 | 148 | require.NoError(t, err) | |
| 134 | 149 | require.Equal(t, 1, connector.resumeCalls) | |
| 135 | 150 | require.Equal(t, 1, connector.attachCalls) | |
| 151 | + | require.Zero(t, connector.restoreCalls) | |
| 136 | 152 | ||
| 137 | 153 | active, ok := ledger.lookup(ledger.activeRef()) | |
| 138 | 154 | require.True(t, ok) | |
| 139 | 155 | require.Equal(t, ports.RouteOriginDiscovery, active.origin) | |
| 156 | + | require.Equal(t, "daemon-discovered", active.originKey) | |
| 140 | 157 | require.Equal(t, ports.RouteOriginDiscovery, active.request.Origin) | |
| 141 | 158 | require.Equal(t, candidate.target, active.target) | |
| 142 | 159 | require.Greater(t, active.identity.generation, identity.generation) | |
| @@ -146,20 +163,43 @@ func TestRouteLedgerNavigateIsTransactionalOnFailureOrTargetChange(t *testing.T) | |||
| 146 | 163 | ledger := newRouteLedger() | |
| 147 | 164 | identity, err := ledger.commit(routeTestCandidate(3, ports.RouteOriginLocal)) | |
| 148 | 165 | require.NoError(t, err) | |
| 166 | + | _, err = ledger.commit(routeTestCandidate(5, ports.RouteOriginLocal)) | |
| 167 | + | require.NoError(t, err) | |
| 149 | 168 | before := ledger.snapshot() | |
| 169 | + | action := identity.wireNavigationAction(routeGeneration(before.Generation)) | |
| 150 | 170 | ||
| 151 | 171 | connector := &routeTestConnector{attachErr: errors.New("offline")} | |
| 152 | - | err = ledger.navigate(context.Background(), identity.wireNavigationAction(), connector) | |
| 172 | + | err = ledger.navigate(context.Background(), action, connector) | |
| 153 | 173 | require.Error(t, err) | |
| 154 | 174 | require.Equal(t, before, ledger.snapshot()) | |
| 175 | + | require.Equal(t, 1, connector.restoreCalls) | |
| 155 | 176 | ||
| 156 | 177 | connector = &routeTestConnector{} | |
| 157 | - | connector.attachErr = nil | |
| 158 | 178 | // Force a mismatched committed target through a small wrapper. | |
| 159 | 179 | mismatch := mismatchingRouteConnector{routeTestConnector: connector} | |
| 160 | - | err = ledger.navigate(context.Background(), identity.wireNavigationAction(), mismatch) | |
| 180 | + | err = ledger.navigate(context.Background(), action, mismatch) | |
| 161 | 181 | require.ErrorIs(t, err, errRouteTargetChanged) | |
| 162 | 182 | require.Equal(t, before, ledger.snapshot()) | |
| 183 | + | require.Equal(t, 1, connector.restoreCalls) | |
| 184 | + | } | |
| 185 | + | ||
| 186 | + | func TestRouteLedgerNavigationRejectsStaleActionsAndActiveNoOp(t *testing.T) { | |
| 187 | + | ledger := newRouteLedger() | |
| 188 | + | active, err := ledger.commit(routeTestCandidate(6, ports.RouteOriginLocal)) | |
| 189 | + | require.NoError(t, err) | |
| 190 | + | firstSnapshot := ledger.snapshot() | |
| 191 | + | before := ledger.snapshot() | |
| 192 | + | ||
| 193 | + | require.NoError(t, ledger.navigate(context.Background(), active.wireNavigationAction(routeGeneration(firstSnapshot.Generation)), nil)) | |
| 194 | + | require.Equal(t, before, ledger.snapshot()) | |
| 195 | + | ||
| 196 | + | _, err = ledger.commit(routeTestCandidate(7, ports.RouteOriginLocal)) | |
| 197 | + | require.NoError(t, err) | |
| 198 | + | stale := active.wireNavigationAction(routeGeneration(firstSnapshot.Generation)) | |
| 199 | + | connector := &routeTestConnector{} | |
| 200 | + | require.ErrorIs(t, ledger.navigate(context.Background(), stale, connector), errRouteStaleSelection) | |
| 201 | + | require.Zero(t, connector.resumeCalls) | |
| 202 | + | require.Zero(t, connector.attachCalls) | |
| 163 | 203 | } | |
| 164 | 204 | ||
| 165 | 205 | type mismatchingRouteConnector struct{ *routeTestConnector } | |
| @@ -171,8 +211,8 @@ func (c mismatchingRouteConnector) Attach(_ context.Context, record routeRecord) | |||
| 171 | 211 | }, nil | |
| 172 | 212 | } | |
| 173 | 213 | ||
| 174 | - | func (i routeIdentity) wireNavigationAction() ports.RouteNavigationAction { | |
| 175 | - | return ports.RouteNavigationAction{Key: uint64(i.key), Generation: uint64(i.generation)} | |
| 214 | + | func (i routeIdentity) wireNavigationAction(snapshotGeneration routeGeneration) ports.RouteNavigationAction { | |
| 215 | + | return ports.RouteNavigationAction{SnapshotGeneration: uint64(snapshotGeneration), Key: uint64(i.key), Generation: uint64(i.generation)} | |
| 176 | 216 | } | |
| 177 | 217 | ||
| 178 | 218 | func TestRouteLedgerConcurrentSnapshotsAndCommits(t *testing.T) { | |
| @@ -190,5 +230,21 @@ func TestRouteLedgerConcurrentSnapshotsAndCommits(t *testing.T) { | |||
| 190 | 230 | }() | |
| 191 | 231 | } | |
| 192 | 232 | wg.Wait() | |
| 193 | - | require.LessOrEqual(t, len(ledger.snapshot().Entries), maxRouteLedgerEntries) | |
| 233 | + | require.LessOrEqual(t, len(ledger.snapshot().Entries), maxRouteLedgerEntries-1) | |
| 234 | + | } | |
| 235 | + | ||
| 236 | + | func TestRouteLedgerSeparatesRemoteDaemonOrigins(t *testing.T) { | |
| 237 | + | ledger := newRouteLedger() | |
| 238 | + | firstCandidate := routeTestCandidate(40, ports.RouteOriginRemote) | |
| 239 | + | firstCandidate.originKey = "daemon-a" | |
| 240 | + | secondCandidate := firstCandidate | |
| 241 | + | secondCandidate.originKey = "daemon-b" | |
| 242 | + | ||
| 243 | + | first, err := ledger.commit(firstCandidate) | |
| 244 | + | require.NoError(t, err) | |
| 245 | + | second, err := ledger.commit(secondCandidate) | |
| 246 | + | require.NoError(t, err) | |
| 247 | + | ||
| 248 | + | require.NotEqual(t, first.key, second.key) | |
| 249 | + | require.Len(t, ledger.snapshot().Entries, 1) | |
| 194 | 250 | } | |
| @@ -45,8 +45,12 @@ func TestRouteLedgerBoundsAndImmutableSnapshot(t *testing.T) { | |||||
| 45 | } | 45 | } | ||
| 46 | 46 | ||||
| 47 | snapshot := ledger.snapshot() | 47 | snapshot := ledger.snapshot() | ||
| 48 | - | require.Len(t, snapshot.Entries, maxRouteLedgerEntries) | 48 | + | require.Len(t, ledger.entries, maxRouteLedgerEntries) |
| 49 | + | require.Len(t, snapshot.Entries, maxRouteLedgerEntries-1) | |||
| 49 | require.Equal(t, uint64(keys[len(keys)-1]), snapshot.Active.Key) | 50 | require.Equal(t, uint64(keys[len(keys)-1]), snapshot.Active.Key) | ||
| 51 | + | for _, entry := range snapshot.Entries { | |||
| 52 | + | require.NotEqual(t, snapshot.Active, ports.RouteRef{Key: entry.Key, Generation: entry.Generation}) | |||
| 53 | + | } | |||
| 50 | require.Equal(t, uint64(keys[len(keys)-2]), snapshot.Previous.Key) | 54 | require.Equal(t, uint64(keys[len(keys)-2]), snapshot.Previous.Key) | ||
| 51 | require.Equal(t, uint64(keys[0]), snapshot.Home.Key, "home is retained when older entries are evicted") | 55 | require.Equal(t, uint64(keys[0]), snapshot.Home.Key, "home is retained when older entries are evicted") | ||
| 52 | 56 | ||||
| @@ -91,10 +95,12 @@ func TestRouteLedgerDiscoveryCyclesAreIndependent(t *testing.T) { | |||||
| 91 | } | 95 | } | ||
| 92 | 96 | ||||
| 93 | type routeTestConnector struct { | 97 | type routeTestConnector struct { | ||
| 94 | - | resumeErr error | 98 | + | resumeErr error |
| 95 | - | attachErr error | 99 | + | attachErr error |
| 96 | - | resumeCalls int | 100 | + | resumeCalls int |
| 97 | - | attachCalls int | 101 | + | attachCalls int |
| 102 | + | restoreCalls int | |||
| 103 | + | restoreErr error | |||
| 98 | } | 104 | } | ||
| 99 | 105 | ||||
| 100 | func (c *routeTestConnector) Resume(context.Context, routeRecord) (routeTransitionCommit, error) { | 106 | func (c *routeTestConnector) Resume(context.Context, routeRecord) (routeTransitionCommit, error) { | ||
| @@ -105,6 +111,11 @@ func (c *routeTestConnector) Resume(context.Context, routeRecord) (routeTransiti | |||||
| 105 | return routeTransitionCommit{}, errors.New("unexpected resume success") | 111 | return routeTransitionCommit{}, errors.New("unexpected resume success") | ||
| 106 | } | 112 | } | ||
| 107 | 113 | ||||
| 114 | + | func (c *routeTestConnector) Restore(_ context.Context, _ routeRecord) error { | |||
| 115 | + | c.restoreCalls++ | |||
| 116 | + | return c.restoreErr | |||
| 117 | + | } | |||
| 118 | + | ||||
| 108 | func (c *routeTestConnector) Attach(_ context.Context, record routeRecord) (routeTransitionCommit, error) { | 119 | func (c *routeTestConnector) Attach(_ context.Context, record routeRecord) (routeTransitionCommit, error) { | ||
| 109 | c.attachCalls++ | 120 | c.attachCalls++ | ||
| 110 | if c.attachErr != nil { | 121 | if c.attachErr != nil { | ||
| @@ -124,19 +135,25 @@ func (c *routeTestConnector) Attach(_ context.Context, record routeRecord) (rout | |||||
| 124 | func TestRouteLedgerNavigateResumesThenFallsBackAndRestoresOrigin(t *testing.T) { | 135 | func TestRouteLedgerNavigateResumesThenFallsBackAndRestoresOrigin(t *testing.T) { | ||
| 125 | ledger := newRouteLedger() | 136 | ledger := newRouteLedger() | ||
| 126 | candidate := routeTestCandidate(2, ports.RouteOriginDiscovery) | 137 | candidate := routeTestCandidate(2, ports.RouteOriginDiscovery) | ||
| 138 | + | candidate.originKey = "daemon-discovered" | |||
| 127 | candidate.resumeToken = 77 | 139 | candidate.resumeToken = 77 | ||
| 128 | identity, err := ledger.commit(candidate) | 140 | identity, err := ledger.commit(candidate) | ||
| 129 | require.NoError(t, err) | 141 | require.NoError(t, err) | ||
| 142 | + | _, err = ledger.commit(routeTestCandidate(4, ports.RouteOriginLocal)) | |||
| 143 | + | require.NoError(t, err) | |||
| 130 | 144 | ||||
| 131 | connector := &routeTestConnector{resumeErr: errRouteResumeUnavailable} | 145 | connector := &routeTestConnector{resumeErr: errRouteResumeUnavailable} | ||
| 132 | - | err = ledger.navigate(context.Background(), ports.RouteNavigationAction{Key: uint64(identity.key), Generation: uint64(identity.generation)}, connector) | 146 | + | action := identity.wireNavigationAction(routeGeneration(ledger.snapshot().Generation)) |
| 147 | + | err = ledger.navigate(context.Background(), action, connector) | |||
| 133 | require.NoError(t, err) | 148 | require.NoError(t, err) | ||
| 134 | require.Equal(t, 1, connector.resumeCalls) | 149 | require.Equal(t, 1, connector.resumeCalls) | ||
| 135 | require.Equal(t, 1, connector.attachCalls) | 150 | require.Equal(t, 1, connector.attachCalls) | ||
| 151 | + | require.Zero(t, connector.restoreCalls) | |||
| 136 | 152 | ||||
| 137 | active, ok := ledger.lookup(ledger.activeRef()) | 153 | active, ok := ledger.lookup(ledger.activeRef()) | ||
| 138 | require.True(t, ok) | 154 | require.True(t, ok) | ||
| 139 | require.Equal(t, ports.RouteOriginDiscovery, active.origin) | 155 | require.Equal(t, ports.RouteOriginDiscovery, active.origin) | ||
| 156 | + | require.Equal(t, "daemon-discovered", active.originKey) | |||
| 140 | require.Equal(t, ports.RouteOriginDiscovery, active.request.Origin) | 157 | require.Equal(t, ports.RouteOriginDiscovery, active.request.Origin) | ||
| 141 | require.Equal(t, candidate.target, active.target) | 158 | require.Equal(t, candidate.target, active.target) | ||
| 142 | require.Greater(t, active.identity.generation, identity.generation) | 159 | require.Greater(t, active.identity.generation, identity.generation) | ||
| @@ -146,20 +163,43 @@ func TestRouteLedgerNavigateIsTransactionalOnFailureOrTargetChange(t *testing.T) | |||||
| 146 | ledger := newRouteLedger() | 163 | ledger := newRouteLedger() | ||
| 147 | identity, err := ledger.commit(routeTestCandidate(3, ports.RouteOriginLocal)) | 164 | identity, err := ledger.commit(routeTestCandidate(3, ports.RouteOriginLocal)) | ||
| 148 | require.NoError(t, err) | 165 | require.NoError(t, err) | ||
| 166 | + | _, err = ledger.commit(routeTestCandidate(5, ports.RouteOriginLocal)) | |||
| 167 | + | require.NoError(t, err) | |||
| 149 | before := ledger.snapshot() | 168 | before := ledger.snapshot() | ||
| 169 | + | action := identity.wireNavigationAction(routeGeneration(before.Generation)) | |||
| 150 | 170 | ||||
| 151 | connector := &routeTestConnector{attachErr: errors.New("offline")} | 171 | connector := &routeTestConnector{attachErr: errors.New("offline")} | ||
| 152 | - | err = ledger.navigate(context.Background(), identity.wireNavigationAction(), connector) | 172 | + | err = ledger.navigate(context.Background(), action, connector) |
| 153 | require.Error(t, err) | 173 | require.Error(t, err) | ||
| 154 | require.Equal(t, before, ledger.snapshot()) | 174 | require.Equal(t, before, ledger.snapshot()) | ||
| 175 | + | require.Equal(t, 1, connector.restoreCalls) | |||
| 155 | 176 | ||||
| 156 | connector = &routeTestConnector{} | 177 | connector = &routeTestConnector{} | ||
| 157 | - | connector.attachErr = nil | |||
| 158 | // Force a mismatched committed target through a small wrapper. | 178 | // Force a mismatched committed target through a small wrapper. | ||
| 159 | mismatch := mismatchingRouteConnector{routeTestConnector: connector} | 179 | mismatch := mismatchingRouteConnector{routeTestConnector: connector} | ||
| 160 | - | err = ledger.navigate(context.Background(), identity.wireNavigationAction(), mismatch) | 180 | + | err = ledger.navigate(context.Background(), action, mismatch) |
| 161 | require.ErrorIs(t, err, errRouteTargetChanged) | 181 | require.ErrorIs(t, err, errRouteTargetChanged) | ||
| 162 | require.Equal(t, before, ledger.snapshot()) | 182 | require.Equal(t, before, ledger.snapshot()) | ||
| 183 | + | require.Equal(t, 1, connector.restoreCalls) | |||
| 184 | + | } | |||
| 185 | + | ||||
| 186 | + | func TestRouteLedgerNavigationRejectsStaleActionsAndActiveNoOp(t *testing.T) { | |||
| 187 | + | ledger := newRouteLedger() | |||
| 188 | + | active, err := ledger.commit(routeTestCandidate(6, ports.RouteOriginLocal)) | |||
| 189 | + | require.NoError(t, err) | |||
| 190 | + | firstSnapshot := ledger.snapshot() | |||
| 191 | + | before := ledger.snapshot() | |||
| 192 | + | ||||
| 193 | + | require.NoError(t, ledger.navigate(context.Background(), active.wireNavigationAction(routeGeneration(firstSnapshot.Generation)), nil)) | |||
| 194 | + | require.Equal(t, before, ledger.snapshot()) | |||
| 195 | + | ||||
| 196 | + | _, err = ledger.commit(routeTestCandidate(7, ports.RouteOriginLocal)) | |||
| 197 | + | require.NoError(t, err) | |||
| 198 | + | stale := active.wireNavigationAction(routeGeneration(firstSnapshot.Generation)) | |||
| 199 | + | connector := &routeTestConnector{} | |||
| 200 | + | require.ErrorIs(t, ledger.navigate(context.Background(), stale, connector), errRouteStaleSelection) | |||
| 201 | + | require.Zero(t, connector.resumeCalls) | |||
| 202 | + | require.Zero(t, connector.attachCalls) | |||
| 163 | } | 203 | } | ||
| 164 | 204 | ||||
| 165 | type mismatchingRouteConnector struct{ *routeTestConnector } | 205 | type mismatchingRouteConnector struct{ *routeTestConnector } | ||
| @@ -171,8 +211,8 @@ func (c mismatchingRouteConnector) Attach(_ context.Context, record routeRecord) | |||||
| 171 | }, nil | 211 | }, nil | ||
| 172 | } | 212 | } | ||
| 173 | 213 | ||||
| 174 | - | func (i routeIdentity) wireNavigationAction() ports.RouteNavigationAction { | 214 | + | func (i routeIdentity) wireNavigationAction(snapshotGeneration routeGeneration) ports.RouteNavigationAction { |
| 175 | - | return ports.RouteNavigationAction{Key: uint64(i.key), Generation: uint64(i.generation)} | 215 | + | return ports.RouteNavigationAction{SnapshotGeneration: uint64(snapshotGeneration), Key: uint64(i.key), Generation: uint64(i.generation)} |
| 176 | } | 216 | } | ||
| 177 | 217 | ||||
| 178 | func TestRouteLedgerConcurrentSnapshotsAndCommits(t *testing.T) { | 218 | func TestRouteLedgerConcurrentSnapshotsAndCommits(t *testing.T) { | ||
| @@ -190,5 +230,21 @@ func TestRouteLedgerConcurrentSnapshotsAndCommits(t *testing.T) { | |||||
| 190 | }() | 230 | }() | ||
| 191 | } | 231 | } | ||
| 192 | wg.Wait() | 232 | wg.Wait() | ||
| 193 | - | require.LessOrEqual(t, len(ledger.snapshot().Entries), maxRouteLedgerEntries) | 233 | + | require.LessOrEqual(t, len(ledger.snapshot().Entries), maxRouteLedgerEntries-1) |
| 234 | + | } | |||
| 235 | + | ||||
| 236 | + | func TestRouteLedgerSeparatesRemoteDaemonOrigins(t *testing.T) { | |||
| 237 | + | ledger := newRouteLedger() | |||
| 238 | + | firstCandidate := routeTestCandidate(40, ports.RouteOriginRemote) | |||
| 239 | + | firstCandidate.originKey = "daemon-a" | |||
| 240 | + | secondCandidate := firstCandidate | |||
| 241 | + | secondCandidate.originKey = "daemon-b" | |||
| 242 | + | ||||
| 243 | + | first, err := ledger.commit(firstCandidate) | |||
| 244 | + | require.NoError(t, err) | |||
| 245 | + | second, err := ledger.commit(secondCandidate) | |||
| 246 | + | require.NoError(t, err) | |||
| 247 | + | ||||
| 248 | + | require.NotEqual(t, first.key, second.key) | |||
| 249 | + | require.Len(t, ledger.snapshot().Entries, 1) | |||
| 194 | } | 250 | } | ||
M internal/usecase/daemon/attachment_effect_gate_test.go +24 -0
| @@ -667,6 +667,30 @@ func TestAttachmentEffectGateAdmittedActiveEffectsFinishBeforeReplacement(t *tes | |||
| 667 | 667 | } | |
| 668 | 668 | } | |
| 669 | 669 | ||
| 670 | + | func TestAttachedRouteSnapshotIsAcceptedAsAnAttachmentValue(t *testing.T) { | |
| 671 | + | d, source, ac, _ := newManualSessionWithPTYs(t, nil) | |
| 672 | + | transport := &closeTrackingTransport{} | |
| 673 | + | ac.replaceTransport(transport) | |
| 674 | + | rc := d.attachCoordinator(source, nil, ac, true) | |
| 675 | + | token := source.attachmentToken(ac, transport) | |
| 676 | + | token.lease = rc.attachmentLease(ac) | |
| 677 | + | ac.publishAttachmentCapability(token) | |
| 678 | + | ||
| 679 | + | snapshot := ports.RecentRouteSnapshot{ | |
| 680 | + | Generation: 4, | |
| 681 | + | Active: ports.RouteRef{Key: 8, Generation: 4}, | |
| 682 | + | Home: ports.RouteRef{Key: 8, Generation: 4}, | |
| 683 | + | Entries: []ports.RecentRouteEntry{{ | |
| 684 | + | Key: 7, Generation: 3, Name: "previous", Kind: ports.RouteKindLocal, | |
| 685 | + | }}, | |
| 686 | + | } | |
| 687 | + | payload, err := ports.MarshalRecentRouteSnapshot(snapshot) | |
| 688 | + | require.NoError(t, err) | |
| 689 | + | ||
| 690 | + | require.False(t, d.handleAttachmentClientFrame(token, ports.Frame{Type: ports.MsgRecentRouteSnapshot, Payload: payload})) | |
| 691 | + | require.Equal(t, snapshot, ac.routeSnapshotCopy()) | |
| 692 | + | } | |
| 693 | + | ||
| 670 | 694 | func TestAttachedNavigationCommandSendsResultAfterLocalTransition(t *testing.T) { | |
| 671 | 695 | d, source, ac, _ := newManualSessionWithPTYs(t, nil) | |
| 672 | 696 | target := &session{sessionCore: sessionCore{id: "target", name: "target"}, ctx: source.ctx, cancel: func() {}, tabs: []*tab{ | |
| @@ -667,6 +667,30 @@ func TestAttachmentEffectGateAdmittedActiveEffectsFinishBeforeReplacement(t *tes | |||||
| 667 | } | 667 | } | ||
| 668 | } | 668 | } | ||
| 669 | 669 | ||||
| 670 | + | func TestAttachedRouteSnapshotIsAcceptedAsAnAttachmentValue(t *testing.T) { | |||
| 671 | + | d, source, ac, _ := newManualSessionWithPTYs(t, nil) | |||
| 672 | + | transport := &closeTrackingTransport{} | |||
| 673 | + | ac.replaceTransport(transport) | |||
| 674 | + | rc := d.attachCoordinator(source, nil, ac, true) | |||
| 675 | + | token := source.attachmentToken(ac, transport) | |||
| 676 | + | token.lease = rc.attachmentLease(ac) | |||
| 677 | + | ac.publishAttachmentCapability(token) | |||
| 678 | + | ||||
| 679 | + | snapshot := ports.RecentRouteSnapshot{ | |||
| 680 | + | Generation: 4, | |||
| 681 | + | Active: ports.RouteRef{Key: 8, Generation: 4}, | |||
| 682 | + | Home: ports.RouteRef{Key: 8, Generation: 4}, | |||
| 683 | + | Entries: []ports.RecentRouteEntry{{ | |||
| 684 | + | Key: 7, Generation: 3, Name: "previous", Kind: ports.RouteKindLocal, | |||
| 685 | + | }}, | |||
| 686 | + | } | |||
| 687 | + | payload, err := ports.MarshalRecentRouteSnapshot(snapshot) | |||
| 688 | + | require.NoError(t, err) | |||
| 689 | + | ||||
| 690 | + | require.False(t, d.handleAttachmentClientFrame(token, ports.Frame{Type: ports.MsgRecentRouteSnapshot, Payload: payload})) | |||
| 691 | + | require.Equal(t, snapshot, ac.routeSnapshotCopy()) | |||
| 692 | + | } | |||
| 693 | + | ||||
| 670 | func TestAttachedNavigationCommandSendsResultAfterLocalTransition(t *testing.T) { | 694 | func TestAttachedNavigationCommandSendsResultAfterLocalTransition(t *testing.T) { | ||
| 671 | d, source, ac, _ := newManualSessionWithPTYs(t, nil) | 695 | d, source, ac, _ := newManualSessionWithPTYs(t, nil) | ||
| 672 | target := &session{sessionCore: sessionCore{id: "target", name: "target"}, ctx: source.ctx, cancel: func() {}, tabs: []*tab{ | 696 | target := &session{sessionCore: sessionCore{id: "target", name: "target"}, ctx: source.ctx, cancel: func() {}, tabs: []*tab{ | ||
M internal/usecase/daemon/client.go +16 -0
| @@ -99,6 +99,8 @@ type attachedClient struct { | |||
| 99 | 99 | previousSession Guarded[domain.SessionID] | |
| 100 | 100 | linkMu sync.Mutex | |
| 101 | 101 | sendMu sync.Mutex | |
| 102 | + | routeMu sync.RWMutex | |
| 103 | + | routeSnapshot ports.RecentRouteSnapshot | |
| 102 | 104 | // routeCreatedSession marks a session created by this attachment's route. | |
| 103 | 105 | // A handshake that never commits Welcome must tear down that exact empty | |
| 104 | 106 | // session, while an attachment routed to an existing session must not. | |
| @@ -355,6 +357,20 @@ func (ac *attachedClient) currentTransportIs(tr ports.Transport) bool { | |||
| 355 | 357 | return tr != nil && ac.transportIs(tr) | |
| 356 | 358 | } | |
| 357 | 359 | ||
| 360 | + | func (ac *attachedClient) setRouteSnapshot(snapshot ports.RecentRouteSnapshot) { | |
| 361 | + | ac.routeMu.Lock() | |
| 362 | + | ac.routeSnapshot = snapshot | |
| 363 | + | ac.routeMu.Unlock() | |
| 364 | + | } | |
| 365 | + | ||
| 366 | + | func (ac *attachedClient) routeSnapshotCopy() ports.RecentRouteSnapshot { | |
| 367 | + | ac.routeMu.RLock() | |
| 368 | + | defer ac.routeMu.RUnlock() | |
| 369 | + | snapshot := ac.routeSnapshot | |
| 370 | + | snapshot.Entries = append([]ports.RecentRouteEntry(nil), snapshot.Entries...) | |
| 371 | + | return snapshot | |
| 372 | + | } | |
| 373 | + | ||
| 358 | 374 | func (ac *attachedClient) ackOutputState(epoch, state uint64) { | |
| 359 | 375 | ac.sendMu.Lock() | |
| 360 | 376 | defer ac.sendMu.Unlock() | |
| @@ -99,6 +99,8 @@ type attachedClient struct { | |||||
| 99 | previousSession Guarded[domain.SessionID] | 99 | previousSession Guarded[domain.SessionID] | ||
| 100 | linkMu sync.Mutex | 100 | linkMu sync.Mutex | ||
| 101 | sendMu sync.Mutex | 101 | sendMu sync.Mutex | ||
| 102 | + | routeMu sync.RWMutex | |||
| 103 | + | routeSnapshot ports.RecentRouteSnapshot | |||
| 102 | // routeCreatedSession marks a session created by this attachment's route. | 104 | // routeCreatedSession marks a session created by this attachment's route. | ||
| 103 | // A handshake that never commits Welcome must tear down that exact empty | 105 | // A handshake that never commits Welcome must tear down that exact empty | ||
| 104 | // session, while an attachment routed to an existing session must not. | 106 | // session, while an attachment routed to an existing session must not. | ||
| @@ -355,6 +357,20 @@ func (ac *attachedClient) currentTransportIs(tr ports.Transport) bool { | |||||
| 355 | return tr != nil && ac.transportIs(tr) | 357 | return tr != nil && ac.transportIs(tr) | ||
| 356 | } | 358 | } | ||
| 357 | 359 | ||||
| 360 | + | func (ac *attachedClient) setRouteSnapshot(snapshot ports.RecentRouteSnapshot) { | |||
| 361 | + | ac.routeMu.Lock() | |||
| 362 | + | ac.routeSnapshot = snapshot | |||
| 363 | + | ac.routeMu.Unlock() | |||
| 364 | + | } | |||
| 365 | + | ||||
| 366 | + | func (ac *attachedClient) routeSnapshotCopy() ports.RecentRouteSnapshot { | |||
| 367 | + | ac.routeMu.RLock() | |||
| 368 | + | defer ac.routeMu.RUnlock() | |||
| 369 | + | snapshot := ac.routeSnapshot | |||
| 370 | + | snapshot.Entries = append([]ports.RecentRouteEntry(nil), snapshot.Entries...) | |||
| 371 | + | return snapshot | |||
| 372 | + | } | |||
| 373 | + | ||||
| 358 | func (ac *attachedClient) ackOutputState(epoch, state uint64) { | 374 | func (ac *attachedClient) ackOutputState(epoch, state uint64) { | ||
| 359 | ac.sendMu.Lock() | 375 | ac.sendMu.Lock() | ||
| 360 | defer ac.sendMu.Unlock() | 376 | defer ac.sendMu.Unlock() | ||
M internal/usecase/daemon/client_frame_routing.go +11 -0
| @@ -125,6 +125,17 @@ func (d *Daemon) handleAttachmentClientFrame(token attachmentConnectionToken, f | |||
| 125 | 125 | if _, derr := ports.UnmarshalOutputResetRequest(f.Payload); derr == nil { | |
| 126 | 126 | d.resetOutput(token) | |
| 127 | 127 | } | |
| 128 | + | case ports.MsgRecentRouteSnapshot: | |
| 129 | + | snapshot, derr := ports.UnmarshalRecentRouteSnapshot(f.Payload) | |
| 130 | + | if derr == nil { | |
| 131 | + | token.ac.setRouteSnapshot(snapshot) | |
| 132 | + | } else { | |
| 133 | + | d.log.Warn("malformed recent route snapshot", "err", derr) | |
| 134 | + | } | |
| 135 | + | case ports.MsgRouteNavigationFailure: | |
| 136 | + | if _, derr := ports.UnmarshalRouteNavigationFailure(f.Payload); derr != nil { | |
| 137 | + | d.log.Warn("malformed route navigation failure", "err", derr) | |
| 138 | + | } | |
| 128 | 139 | case ports.MsgCommand: | |
| 129 | 140 | request, derr := ports.UnmarshalCommandRequest(f.Payload) | |
| 130 | 141 | if derr != nil { | |
| @@ -125,6 +125,17 @@ func (d *Daemon) handleAttachmentClientFrame(token attachmentConnectionToken, f | |||||
| 125 | if _, derr := ports.UnmarshalOutputResetRequest(f.Payload); derr == nil { | 125 | if _, derr := ports.UnmarshalOutputResetRequest(f.Payload); derr == nil { | ||
| 126 | d.resetOutput(token) | 126 | d.resetOutput(token) | ||
| 127 | } | 127 | } | ||
| 128 | + | case ports.MsgRecentRouteSnapshot: | |||
| 129 | + | snapshot, derr := ports.UnmarshalRecentRouteSnapshot(f.Payload) | |||
| 130 | + | if derr == nil { | |||
| 131 | + | token.ac.setRouteSnapshot(snapshot) | |||
| 132 | + | } else { | |||
| 133 | + | d.log.Warn("malformed recent route snapshot", "err", derr) | |||
| 134 | + | } | |||
| 135 | + | case ports.MsgRouteNavigationFailure: | |||
| 136 | + | if _, derr := ports.UnmarshalRouteNavigationFailure(f.Payload); derr != nil { | |||
| 137 | + | d.log.Warn("malformed route navigation failure", "err", derr) | |||
| 138 | + | } | |||
| 128 | case ports.MsgCommand: | 139 | case ports.MsgCommand: | ||
| 129 | request, derr := ports.UnmarshalCommandRequest(f.Payload) | 140 | request, derr := ports.UnmarshalCommandRequest(f.Payload) | ||
| 130 | if derr != nil { | 141 | if derr != nil { | ||