fix(daemon): keep same-peer back switches in band (#244)
- commit
- e5918209f502170e91c7ac4714d3ea968f52ef04
- author
- bnema · · 2026-08-31 17:40 UTC
- committer
- GitHub · · 2026-08-31 17:40 UTC
- parent
- b09c5d7
- signature
- GPG Unverified
B5690EEEBB952194The key is unknown
6 files · +165 -29
M docs/remote-resilience.md +4 -3
| @@ -38,9 +38,10 @@ across local, direct remote, and picker-discovered attachments. Catalog targets | |||
| 38 | 38 | retain exact endpoints, while route snapshots keep endpoints private; rows from | |
| 39 | 39 | those sources therefore remain separate when their authority cannot be compared | |
| 40 | 40 | exactly. The daemon renders the latest attachment snapshot; active routes are | |
| 41 | - | metadata-only, while `JRS` and `BCK` send typed | |
| 42 | - | key/generation actions back to the client. Route history is process-local and | |
| 43 | - | is not persisted or shared between clients. | |
| 41 | + | metadata-only. `JRS` and cross-daemon `BCK` transitions send typed | |
| 42 | + | key/generation actions back to the client, while `BCK` to a live session on the | |
| 43 | + | current authenticated daemon uses an in-band switch. Route history is | |
| 44 | + | process-local and is not persisted or shared between clients. | |
| 44 | 45 | ||
| 45 | 46 | A picker-selected remote target is a direct handoff with one input pump. Exact | |
| 46 | 47 | lifecycle/name identities are carried separately from display labels, and a | |
| @@ -38,9 +38,10 @@ across local, direct remote, and picker-discovered attachments. Catalog targets | |||||
| 38 | retain exact endpoints, while route snapshots keep endpoints private; rows from | 38 | retain exact endpoints, while route snapshots keep endpoints private; rows from | ||
| 39 | those sources therefore remain separate when their authority cannot be compared | 39 | those sources therefore remain separate when their authority cannot be compared | ||
| 40 | exactly. The daemon renders the latest attachment snapshot; active routes are | 40 | exactly. The daemon renders the latest attachment snapshot; active routes are | ||
| 41 | - | metadata-only, while `JRS` and `BCK` send typed | 41 | + | metadata-only. `JRS` and cross-daemon `BCK` transitions send typed |
| 42 | - | key/generation actions back to the client. Route history is process-local and | 42 | + | key/generation actions back to the client, while `BCK` to a live session on the |
| 43 | - | is not persisted or shared between clients. | 43 | + | current authenticated daemon uses an in-band switch. Route history is |
| 44 | + | process-local and is not persisted or shared between clients. | |||
| 44 | 45 | ||||
| 45 | A picker-selected remote target is a direct handoff with one input pump. Exact | 46 | A picker-selected remote target is a direct handoff with one input pump. Exact | ||
| 46 | lifecycle/name identities are carried separately from display labels, and a | 47 | lifecycle/name identities are carried separately from display labels, and a | ||
M internal/usecase/client/client_test.go +55 -0
| @@ -823,6 +823,61 @@ func TestHybridPickerSameHostSwitchReusesRemoteTransport(t *testing.T) { | |||
| 823 | 823 | require.Equal(t, int32(1), remoteDialer.calls.Load(), "same-host picker switch must retain the authenticated remote transport") | |
| 824 | 824 | } | |
| 825 | 825 | ||
| 826 | + | func TestHybridBackSessionSamePeerOfferReusesRemoteTransport(t *testing.T) { | |
| 827 | + | term := newRunTerminal() | |
| 828 | + | defer term.in.unblock() | |
| 829 | + | ||
| 830 | + | localLifecycle := domain.SessionLifecycleID{1} | |
| 831 | + | sourceLifecycle := domain.SessionLifecycleID{2} | |
| 832 | + | targetLifecycle := domain.SessionLifecycleID{3} | |
| 833 | + | sourceExact := protocol.ExactSessionTarget{LifecycleID: sourceLifecycle, SessionName: "source"} | |
| 834 | + | targetExact := protocol.ExactSessionTarget{LifecycleID: targetLifecycle, SessionName: "target"} | |
| 835 | + | sourceTarget := domain.RemoteSessionTarget{ | |
| 836 | + | Endpoint: "remote", DisplayOrigin: "remote", LifecycleID: sourceLifecycle, | |
| 837 | + | SessionName: "source", LiveTabID: "source-tab", | |
| 838 | + | } | |
| 839 | + | localInitial := hybridLocalBootstrap(localLifecycle, sourceTarget) | |
| 840 | + | firstSwitchSent := make(chan struct{}) | |
| 841 | + | backSwitchSent := make(chan struct{}) | |
| 842 | + | remote := &recordingTransport{recvs: []recvItem{ | |
| 843 | + | {f: hybridWelcomeFrame("source", sourceLifecycle)}, | |
| 844 | + | {f: frameOf(wire.MsgAttachTarget, wire.MarshalAttachTarget(protocol.AttachTarget{ | |
| 845 | + | Session: "target", Intent: protocol.IntentAttach, ExactTarget: &targetExact, | |
| 846 | + | EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, SamePeer: true, | |
| 847 | + | }))}, | |
| 848 | + | {f: frameOf(wire.MsgCommittedRouteIdentity, mustMarshalCommittedIdentity(protocol.CommittedRouteIdentity{Target: targetExact})), wait: firstSwitchSent}, | |
| 849 | + | {f: frameOf(wire.MsgAttachTarget, wire.MarshalAttachTarget(protocol.AttachTarget{ | |
| 850 | + | Session: "source", Intent: protocol.IntentAttach, ExactTarget: &sourceExact, | |
| 851 | + | EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, SamePeer: true, | |
| 852 | + | }))}, | |
| 853 | + | {f: frameOf(wire.MsgCommittedRouteIdentity, mustMarshalCommittedIdentity(protocol.CommittedRouteIdentity{Target: sourceExact})), wait: backSwitchSent}, | |
| 854 | + | {f: frameOf(wire.MsgDetached, wire.MarshalDetached(protocol.Detached{Reason: protocol.ReasonDetach}))}, | |
| 855 | + | }} | |
| 856 | + | remote.onSend = func(frame wire.Frame) { | |
| 857 | + | if frame.Type != wire.MsgSamePeerSwitchRequest { | |
| 858 | + | return | |
| 859 | + | } | |
| 860 | + | request, err := wire.UnmarshalSamePeerSwitchRequest(frame.Payload) | |
| 861 | + | require.NoError(t, err) | |
| 862 | + | switch request.Target { | |
| 863 | + | case targetExact: | |
| 864 | + | close(firstSwitchSent) | |
| 865 | + | case sourceExact: | |
| 866 | + | close(backSwitchSent) | |
| 867 | + | } | |
| 868 | + | } | |
| 869 | + | localDialer := &sequenceDialer{trs: []wire.Transport{localInitial}} | |
| 870 | + | remoteDialer := &sequenceDialer{trs: []wire.Transport{markedDatagramTransport{Transport: remote}}} | |
| 871 | + | deps := hybridPickerDependencies(localDialer, term, realClock{}, map[string]ports.ClientDialer{"remote": remoteDialer}) | |
| 872 | + | ctx, cancel := context.WithTimeout(context.Background(), time.Second) | |
| 873 | + | defer cancel() | |
| 874 | + | ||
| 875 | + | require.NoError(t, runTestClient(ctx, deps, client.AttachRequest{ | |
| 876 | + | Intent: protocol.IntentAttach, SessionName: "local", Origin: protocol.RouteOriginLocal, OriginKey: "local", | |
| 877 | + | })) | |
| 878 | + | require.Equal(t, int32(1), remoteDialer.calls.Load(), "BCK must retain the authenticated remote transport") | |
| 879 | + | } | |
| 880 | + | ||
| 826 | 881 | func TestHybridPickerLocalSelectionCommitsLocalRoute(t *testing.T) { | |
| 827 | 882 | term := newRunTerminal() | |
| 828 | 883 | defer term.in.unblock() | |
| @@ -823,6 +823,61 @@ func TestHybridPickerSameHostSwitchReusesRemoteTransport(t *testing.T) { | |||||
| 823 | require.Equal(t, int32(1), remoteDialer.calls.Load(), "same-host picker switch must retain the authenticated remote transport") | 823 | require.Equal(t, int32(1), remoteDialer.calls.Load(), "same-host picker switch must retain the authenticated remote transport") | ||
| 824 | } | 824 | } | ||
| 825 | 825 | ||||
| 826 | + | func TestHybridBackSessionSamePeerOfferReusesRemoteTransport(t *testing.T) { | |||
| 827 | + | term := newRunTerminal() | |||
| 828 | + | defer term.in.unblock() | |||
| 829 | + | ||||
| 830 | + | localLifecycle := domain.SessionLifecycleID{1} | |||
| 831 | + | sourceLifecycle := domain.SessionLifecycleID{2} | |||
| 832 | + | targetLifecycle := domain.SessionLifecycleID{3} | |||
| 833 | + | sourceExact := protocol.ExactSessionTarget{LifecycleID: sourceLifecycle, SessionName: "source"} | |||
| 834 | + | targetExact := protocol.ExactSessionTarget{LifecycleID: targetLifecycle, SessionName: "target"} | |||
| 835 | + | sourceTarget := domain.RemoteSessionTarget{ | |||
| 836 | + | Endpoint: "remote", DisplayOrigin: "remote", LifecycleID: sourceLifecycle, | |||
| 837 | + | SessionName: "source", LiveTabID: "source-tab", | |||
| 838 | + | } | |||
| 839 | + | localInitial := hybridLocalBootstrap(localLifecycle, sourceTarget) | |||
| 840 | + | firstSwitchSent := make(chan struct{}) | |||
| 841 | + | backSwitchSent := make(chan struct{}) | |||
| 842 | + | remote := &recordingTransport{recvs: []recvItem{ | |||
| 843 | + | {f: hybridWelcomeFrame("source", sourceLifecycle)}, | |||
| 844 | + | {f: frameOf(wire.MsgAttachTarget, wire.MarshalAttachTarget(protocol.AttachTarget{ | |||
| 845 | + | Session: "target", Intent: protocol.IntentAttach, ExactTarget: &targetExact, | |||
| 846 | + | EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, SamePeer: true, | |||
| 847 | + | }))}, | |||
| 848 | + | {f: frameOf(wire.MsgCommittedRouteIdentity, mustMarshalCommittedIdentity(protocol.CommittedRouteIdentity{Target: targetExact})), wait: firstSwitchSent}, | |||
| 849 | + | {f: frameOf(wire.MsgAttachTarget, wire.MarshalAttachTarget(protocol.AttachTarget{ | |||
| 850 | + | Session: "source", Intent: protocol.IntentAttach, ExactTarget: &sourceExact, | |||
| 851 | + | EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, SamePeer: true, | |||
| 852 | + | }))}, | |||
| 853 | + | {f: frameOf(wire.MsgCommittedRouteIdentity, mustMarshalCommittedIdentity(protocol.CommittedRouteIdentity{Target: sourceExact})), wait: backSwitchSent}, | |||
| 854 | + | {f: frameOf(wire.MsgDetached, wire.MarshalDetached(protocol.Detached{Reason: protocol.ReasonDetach}))}, | |||
| 855 | + | }} | |||
| 856 | + | remote.onSend = func(frame wire.Frame) { | |||
| 857 | + | if frame.Type != wire.MsgSamePeerSwitchRequest { | |||
| 858 | + | return | |||
| 859 | + | } | |||
| 860 | + | request, err := wire.UnmarshalSamePeerSwitchRequest(frame.Payload) | |||
| 861 | + | require.NoError(t, err) | |||
| 862 | + | switch request.Target { | |||
| 863 | + | case targetExact: | |||
| 864 | + | close(firstSwitchSent) | |||
| 865 | + | case sourceExact: | |||
| 866 | + | close(backSwitchSent) | |||
| 867 | + | } | |||
| 868 | + | } | |||
| 869 | + | localDialer := &sequenceDialer{trs: []wire.Transport{localInitial}} | |||
| 870 | + | remoteDialer := &sequenceDialer{trs: []wire.Transport{markedDatagramTransport{Transport: remote}}} | |||
| 871 | + | deps := hybridPickerDependencies(localDialer, term, realClock{}, map[string]ports.ClientDialer{"remote": remoteDialer}) | |||
| 872 | + | ctx, cancel := context.WithTimeout(context.Background(), time.Second) | |||
| 873 | + | defer cancel() | |||
| 874 | + | ||||
| 875 | + | require.NoError(t, runTestClient(ctx, deps, client.AttachRequest{ | |||
| 876 | + | Intent: protocol.IntentAttach, SessionName: "local", Origin: protocol.RouteOriginLocal, OriginKey: "local", | |||
| 877 | + | })) | |||
| 878 | + | require.Equal(t, int32(1), remoteDialer.calls.Load(), "BCK must retain the authenticated remote transport") | |||
| 879 | + | } | |||
| 880 | + | ||||
| 826 | func TestHybridPickerLocalSelectionCommitsLocalRoute(t *testing.T) { | 881 | func TestHybridPickerLocalSelectionCommitsLocalRoute(t *testing.T) { | ||
| 827 | term := newRunTerminal() | 882 | term := newRunTerminal() | ||
| 828 | defer term.in.unblock() | 883 | defer term.in.unblock() | ||
M internal/usecase/daemon/picker.go +19 -23
| @@ -1031,43 +1031,39 @@ func (d *Daemon) sendLocalAttachTargetForAttachment(effect *attachmentEffect, ta | |||
| 1031 | 1031 | if exactTarget == nil { | |
| 1032 | 1032 | return errAttachmentTransition | |
| 1033 | 1033 | } | |
| 1034 | + | // A close-and-dial handoff leaves this daemon's Kitty namespace. A | |
| 1035 | + | // same-peer transition keeps the attachment and its namespace; the target | |
| 1036 | + | // scene diff deletes and replaces the source session's placements. | |
| 1037 | + | if samePeerEligible { | |
| 1038 | + | if err := offerSamePeerAttachTarget(effect, *exactTarget); err != nil { | |
| 1039 | + | return domain.UserErr(domain.NoticeSessionUnavailable, "couldn't offer local session switch", err) | |
| 1040 | + | } | |
| 1041 | + | return nil | |
| 1042 | + | } | |
| 1034 | 1043 | handoff := protocol.AttachTarget{ | |
| 1035 | 1044 | Session: sessionName, | |
| 1036 | 1045 | Intent: protocol.IntentAttach, | |
| 1037 | 1046 | ExactTarget: exactTarget, | |
| 1038 | 1047 | EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, | |
| 1039 | - | SamePeer: samePeerEligible, | |
| 1040 | 1048 | } | |
| 1041 | 1049 | if protocol.ValidateAttachTarget(handoff) != nil { | |
| 1042 | 1050 | return errAttachmentTransition | |
| 1043 | 1051 | } | |
| 1044 | - | // A close-and-dial handoff leaves this daemon's Kitty namespace. A | |
| 1045 | - | // same-peer transition keeps the attachment and its namespace; the target | |
| 1046 | - | // scene diff deletes and replaces the source session's placements. | |
| 1047 | - | if !samePeerEligible { | |
| 1048 | - | if err := d.cleanupAttachmentOutput(effect.ac); err != nil { | |
| 1049 | - | return domain.UserErr(domain.NoticeSessionUnavailable, "couldn't clean up attachment output before local handoff", err) | |
| 1050 | - | } | |
| 1051 | - | } else { | |
| 1052 | - | effect.ac.offerSamePeerTarget(*exactTarget) | |
| 1052 | + | if err := d.cleanupAttachmentOutput(effect.ac); err != nil { | |
| 1053 | + | return domain.UserErr(domain.NoticeSessionUnavailable, "couldn't clean up attachment output before local handoff", err) | |
| 1053 | 1054 | } | |
| 1054 | 1055 | if err := effect.sendControl(handoff); err != nil { | |
| 1055 | - | if samePeerEligible { | |
| 1056 | - | effect.ac.clearSamePeerOffer() | |
| 1057 | - | } | |
| 1058 | 1056 | return domain.UserErr(domain.NoticeSessionUnavailable, "couldn't offer local session switch", err) | |
| 1059 | 1057 | } | |
| 1060 | - | if !samePeerEligible { | |
| 1061 | - | // Stopped and explicit non-default index targets retain the close-and-dial | |
| 1062 | - | // handoff instead of losing their target-specific transition semantics. | |
| 1063 | - | // Keep the source link open until the client receives the ordered cleanup | |
| 1064 | - | // and handoff frames, then let the client's close drive ordinary parking. | |
| 1065 | - | if guard.closePicker { | |
| 1066 | - | d.closePickerForGuard(effect.ac, guard) | |
| 1067 | - | } | |
| 1068 | - | effect.bindActionEnd(d, "detach") | |
| 1069 | - | effect.End() | |
| 1058 | + | // Stopped and explicit non-default index targets retain the close-and-dial | |
| 1059 | + | // handoff instead of losing their target-specific transition semantics. | |
| 1060 | + | // Keep the source link open until the client receives the ordered cleanup | |
| 1061 | + | // and handoff frames, then let the client's close drive ordinary parking. | |
| 1062 | + | if guard.closePicker { | |
| 1063 | + | d.closePickerForGuard(effect.ac, guard) | |
| 1070 | 1064 | } | |
| 1065 | + | effect.bindActionEnd(d, "detach") | |
| 1066 | + | effect.End() | |
| 1071 | 1067 | return nil | |
| 1072 | 1068 | } | |
| 1073 | 1069 | ||
| @@ -1031,43 +1031,39 @@ func (d *Daemon) sendLocalAttachTargetForAttachment(effect *attachmentEffect, ta | |||||
| 1031 | if exactTarget == nil { | 1031 | if exactTarget == nil { | ||
| 1032 | return errAttachmentTransition | 1032 | return errAttachmentTransition | ||
| 1033 | } | 1033 | } | ||
| 1034 | + | // A close-and-dial handoff leaves this daemon's Kitty namespace. A | |||
| 1035 | + | // same-peer transition keeps the attachment and its namespace; the target | |||
| 1036 | + | // scene diff deletes and replaces the source session's placements. | |||
| 1037 | + | if samePeerEligible { | |||
| 1038 | + | if err := offerSamePeerAttachTarget(effect, *exactTarget); err != nil { | |||
| 1039 | + | return domain.UserErr(domain.NoticeSessionUnavailable, "couldn't offer local session switch", err) | |||
| 1040 | + | } | |||
| 1041 | + | return nil | |||
| 1042 | + | } | |||
| 1034 | handoff := protocol.AttachTarget{ | 1043 | handoff := protocol.AttachTarget{ | ||
| 1035 | Session: sessionName, | 1044 | Session: sessionName, | ||
| 1036 | Intent: protocol.IntentAttach, | 1045 | Intent: protocol.IntentAttach, | ||
| 1037 | ExactTarget: exactTarget, | 1046 | ExactTarget: exactTarget, | ||
| 1038 | EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, | 1047 | EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, | ||
| 1039 | - | SamePeer: samePeerEligible, | |||
| 1040 | } | 1048 | } | ||
| 1041 | if protocol.ValidateAttachTarget(handoff) != nil { | 1049 | if protocol.ValidateAttachTarget(handoff) != nil { | ||
| 1042 | return errAttachmentTransition | 1050 | return errAttachmentTransition | ||
| 1043 | } | 1051 | } | ||
| 1044 | - | // A close-and-dial handoff leaves this daemon's Kitty namespace. A | 1052 | + | if err := d.cleanupAttachmentOutput(effect.ac); err != nil { |
| 1045 | - | // same-peer transition keeps the attachment and its namespace; the target | 1053 | + | return domain.UserErr(domain.NoticeSessionUnavailable, "couldn't clean up attachment output before local handoff", err) |
| 1046 | - | // scene diff deletes and replaces the source session's placements. | |||
| 1047 | - | if !samePeerEligible { | |||
| 1048 | - | if err := d.cleanupAttachmentOutput(effect.ac); err != nil { | |||
| 1049 | - | return domain.UserErr(domain.NoticeSessionUnavailable, "couldn't clean up attachment output before local handoff", err) | |||
| 1050 | - | } | |||
| 1051 | - | } else { | |||
| 1052 | - | effect.ac.offerSamePeerTarget(*exactTarget) | |||
| 1053 | } | 1054 | } | ||
| 1054 | if err := effect.sendControl(handoff); err != nil { | 1055 | if err := effect.sendControl(handoff); err != nil { | ||
| 1055 | - | if samePeerEligible { | |||
| 1056 | - | effect.ac.clearSamePeerOffer() | |||
| 1057 | - | } | |||
| 1058 | return domain.UserErr(domain.NoticeSessionUnavailable, "couldn't offer local session switch", err) | 1056 | return domain.UserErr(domain.NoticeSessionUnavailable, "couldn't offer local session switch", err) | ||
| 1059 | } | 1057 | } | ||
| 1060 | - | if !samePeerEligible { | 1058 | + | // Stopped and explicit non-default index targets retain the close-and-dial |
| 1061 | - | // Stopped and explicit non-default index targets retain the close-and-dial | 1059 | + | // handoff instead of losing their target-specific transition semantics. |
| 1062 | - | // handoff instead of losing their target-specific transition semantics. | 1060 | + | // Keep the source link open until the client receives the ordered cleanup |
| 1063 | - | // Keep the source link open until the client receives the ordered cleanup | 1061 | + | // and handoff frames, then let the client's close drive ordinary parking. |
| 1064 | - | // and handoff frames, then let the client's close drive ordinary parking. | 1062 | + | if guard.closePicker { |
| 1065 | - | if guard.closePicker { | 1063 | + | d.closePickerForGuard(effect.ac, guard) |
| 1066 | - | d.closePickerForGuard(effect.ac, guard) | |||
| 1067 | - | } | |||
| 1068 | - | effect.bindActionEnd(d, "detach") | |||
| 1069 | - | effect.End() | |||
| 1070 | } | 1064 | } | ||
| 1065 | + | effect.bindActionEnd(d, "detach") | |||
| 1066 | + | effect.End() | |||
| 1071 | return nil | 1067 | return nil | ||
| 1072 | } | 1068 | } | ||
| 1073 | 1069 | ||||
M internal/usecase/daemon/same_peer_switch.go +24 -0
| @@ -11,6 +11,30 @@ func (ac *attachedClient) offerSamePeerTarget(target protocol.ExactSessionTarget | |||
| 11 | 11 | ac.samePeerOfferMu.Unlock() | |
| 12 | 12 | } | |
| 13 | 13 | ||
| 14 | + | // offerSamePeerAttachTarget arms the exact target before publishing its | |
| 15 | + | // endpoint-empty handoff, then rolls the offer back if publication fails. | |
| 16 | + | func offerSamePeerAttachTarget(effect *attachmentEffect, target protocol.ExactSessionTarget) error { | |
| 17 | + | if effect == nil || effect.ac == nil || !effect.current() { | |
| 18 | + | return errAttachmentTransition | |
| 19 | + | } | |
| 20 | + | handoff := protocol.AttachTarget{ | |
| 21 | + | Session: target.SessionName, | |
| 22 | + | Intent: protocol.IntentAttach, | |
| 23 | + | ExactTarget: &target, | |
| 24 | + | EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, | |
| 25 | + | SamePeer: true, | |
| 26 | + | } | |
| 27 | + | if protocol.ValidateAttachTarget(handoff) != nil { | |
| 28 | + | return errAttachmentTransition | |
| 29 | + | } | |
| 30 | + | effect.ac.offerSamePeerTarget(target) | |
| 31 | + | if err := effect.sendControl(handoff); err != nil { | |
| 32 | + | effect.ac.clearSamePeerOffer() | |
| 33 | + | return err | |
| 34 | + | } | |
| 35 | + | return nil | |
| 36 | + | } | |
| 37 | + | ||
| 14 | 38 | func (ac *attachedClient) clearSamePeerOffer() { | |
| 15 | 39 | ac.samePeerOfferMu.Lock() | |
| 16 | 40 | ac.samePeerOffer = nil | |
| @@ -11,6 +11,30 @@ func (ac *attachedClient) offerSamePeerTarget(target protocol.ExactSessionTarget | |||||
| 11 | ac.samePeerOfferMu.Unlock() | 11 | ac.samePeerOfferMu.Unlock() | ||
| 12 | } | 12 | } | ||
| 13 | 13 | ||||
| 14 | + | // offerSamePeerAttachTarget arms the exact target before publishing its | |||
| 15 | + | // endpoint-empty handoff, then rolls the offer back if publication fails. | |||
| 16 | + | func offerSamePeerAttachTarget(effect *attachmentEffect, target protocol.ExactSessionTarget) error { | |||
| 17 | + | if effect == nil || effect.ac == nil || !effect.current() { | |||
| 18 | + | return errAttachmentTransition | |||
| 19 | + | } | |||
| 20 | + | handoff := protocol.AttachTarget{ | |||
| 21 | + | Session: target.SessionName, | |||
| 22 | + | Intent: protocol.IntentAttach, | |||
| 23 | + | ExactTarget: &target, | |||
| 24 | + | EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, | |||
| 25 | + | SamePeer: true, | |||
| 26 | + | } | |||
| 27 | + | if protocol.ValidateAttachTarget(handoff) != nil { | |||
| 28 | + | return errAttachmentTransition | |||
| 29 | + | } | |||
| 30 | + | effect.ac.offerSamePeerTarget(target) | |||
| 31 | + | if err := effect.sendControl(handoff); err != nil { | |||
| 32 | + | effect.ac.clearSamePeerOffer() | |||
| 33 | + | return err | |||
| 34 | + | } | |||
| 35 | + | return nil | |||
| 36 | + | } | |||
| 37 | + | ||||
| 14 | func (ac *attachedClient) clearSamePeerOffer() { | 38 | func (ac *attachedClient) clearSamePeerOffer() { | ||
| 15 | ac.samePeerOfferMu.Lock() | 39 | ac.samePeerOfferMu.Lock() | ||
| 16 | ac.samePeerOffer = nil | 40 | ac.samePeerOffer = nil | ||
M internal/usecase/daemon/session_back.go +9 -2
| @@ -6,8 +6,9 @@ import ( | |||
| 6 | 6 | ) | |
| 7 | 7 | ||
| 8 | 8 | // backSessionForAttachment delegates previous-route ownership to the client | |
| 9 | - | // once it has published a complete route snapshot. There is no daemon-local | |
| 10 | - | // previous-session fallback after the global-history cutover. | |
| 9 | + | // once it has published a complete route snapshot. A live route on the current | |
| 10 | + | // daemon uses the authenticated same-peer path; all other routes return to the | |
| 11 | + | // client for close-and-dial navigation. | |
| 11 | 12 | func (d *Daemon) backSessionForAttachment(effect *attachmentEffect) error { | |
| 12 | 13 | if d == nil || !effect.current() || effect.sess == nil || effect.ac == nil { | |
| 13 | 14 | return nil | |
| @@ -22,6 +23,12 @@ func (d *Daemon) backSessionForAttachment(effect *attachmentEffect) error { | |||
| 22 | 23 | if snapshot.Previous.Key == 0 || snapshot.Previous.Generation == 0 { | |
| 23 | 24 | return nil | |
| 24 | 25 | } | |
| 26 | + | if target, ok := effect.ac.routeAttentionTarget(snapshot.Previous); ok { | |
| 27 | + | targetSession, _, live := d.samePeerTarget(protocol.SamePeerSwitchRequest{Target: target}) | |
| 28 | + | if live && targetSession != effect.sess { | |
| 29 | + | return offerSamePeerAttachTarget(effect, target) | |
| 30 | + | } | |
| 31 | + | } | |
| 25 | 32 | return d.sendRecentRouteNavigationActionForAttachment(effect, protocol.RouteNavigationAction{ | |
| 26 | 33 | SnapshotGeneration: snapshot.Generation, | |
| 27 | 34 | Key: snapshot.Previous.Key, | |
| @@ -6,8 +6,9 @@ import ( | |||||
| 6 | ) | 6 | ) | ||
| 7 | 7 | ||||
| 8 | // backSessionForAttachment delegates previous-route ownership to the client | 8 | // backSessionForAttachment delegates previous-route ownership to the client | ||
| 9 | - | // once it has published a complete route snapshot. There is no daemon-local | 9 | + | // once it has published a complete route snapshot. A live route on the current |
| 10 | - | // previous-session fallback after the global-history cutover. | 10 | + | // daemon uses the authenticated same-peer path; all other routes return to the |
| 11 | + | // client for close-and-dial navigation. | |||
| 11 | func (d *Daemon) backSessionForAttachment(effect *attachmentEffect) error { | 12 | func (d *Daemon) backSessionForAttachment(effect *attachmentEffect) error { | ||
| 12 | if d == nil || !effect.current() || effect.sess == nil || effect.ac == nil { | 13 | if d == nil || !effect.current() || effect.sess == nil || effect.ac == nil { | ||
| 13 | return nil | 14 | return nil | ||
| @@ -22,6 +23,12 @@ func (d *Daemon) backSessionForAttachment(effect *attachmentEffect) error { | |||||
| 22 | if snapshot.Previous.Key == 0 || snapshot.Previous.Generation == 0 { | 23 | if snapshot.Previous.Key == 0 || snapshot.Previous.Generation == 0 { | ||
| 23 | return nil | 24 | return nil | ||
| 24 | } | 25 | } | ||
| 26 | + | if target, ok := effect.ac.routeAttentionTarget(snapshot.Previous); ok { | |||
| 27 | + | targetSession, _, live := d.samePeerTarget(protocol.SamePeerSwitchRequest{Target: target}) | |||
| 28 | + | if live && targetSession != effect.sess { | |||
| 29 | + | return offerSamePeerAttachTarget(effect, target) | |||
| 30 | + | } | |||
| 31 | + | } | |||
| 25 | return d.sendRecentRouteNavigationActionForAttachment(effect, protocol.RouteNavigationAction{ | 32 | return d.sendRecentRouteNavigationActionForAttachment(effect, protocol.RouteNavigationAction{ | ||
| 26 | SnapshotGeneration: snapshot.Generation, | 33 | SnapshotGeneration: snapshot.Generation, | ||
| 27 | Key: snapshot.Previous.Key, | 34 | Key: snapshot.Previous.Key, | ||
M internal/usecase/daemon/session_back_route_test.go +54 -1
| @@ -3,6 +3,7 @@ package daemon | |||
| 3 | 3 | import ( | |
| 4 | 4 | "testing" | |
| 5 | 5 | ||
| 6 | + | "github.com/bnema/vev/internal/domain" | |
| 6 | 7 | "github.com/bnema/vev/internal/protocol" | |
| 7 | 8 | "github.com/bnema/vev/internal/protocol/wire" | |
| 8 | 9 | "github.com/stretchr/testify/require" | |
| @@ -19,12 +20,16 @@ func TestBackSessionUsesClientPreviousRouteAfterSnapshotPublication(t *testing.T | |||
| 19 | 20 | effect, admitted := ac.beginAttachmentEffect(token) | |
| 20 | 21 | require.True(t, admitted) | |
| 21 | 22 | defer effect.End() | |
| 23 | + | previous := protocol.RouteRef{Key: 7, Generation: 3} | |
| 22 | 24 | ac.setRouteSnapshot(protocol.RecentRouteSnapshot{ | |
| 23 | 25 | Generation: 5, | |
| 24 | 26 | Active: protocol.RouteRef{Key: 9, Generation: 5}, | |
| 25 | - | Previous: protocol.RouteRef{Key: 7, Generation: 3}, | |
| 27 | + | Previous: previous, | |
| 26 | 28 | Entries: []protocol.RecentRouteEntry{testRouteEntry(7, 3, "previous", 7, protocol.RouteKindLocal)}, | |
| 27 | 29 | }) | |
| 30 | + | ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{ | |
| 31 | + | Targets: []protocol.RouteAttentionTarget{{Ref: previous, Target: testRouteTarget("previous", 7)}}, | |
| 32 | + | }) | |
| 28 | 33 | ||
| 29 | 34 | require.NoError(t, d.backSessionForAttachment(effect)) | |
| 30 | 35 | frames := transport.Sends() | |
| @@ -43,3 +48,51 @@ func TestBackSessionUsesClientPreviousRouteAfterSnapshotPublication(t *testing.T | |||
| 43 | 48 | require.True(t, found) | |
| 44 | 49 | require.Equal(t, protocol.RouteNavigationAction{SnapshotGeneration: 5, Key: 7, Generation: 3}, action) | |
| 45 | 50 | } | |
| 51 | + | ||
| 52 | + | func TestBackSessionOffersPreviousRouteOnCurrentDaemon(t *testing.T) { | |
| 53 | + | d, source, ac, _ := newManualSessionWithPTYs(t, nil) | |
| 54 | + | target := &session{ | |
| 55 | + | sessionCore: sessionCore{ | |
| 56 | + | id: "target", name: "target", incarnation: testRouteTarget("target", 7).LifecycleID, | |
| 57 | + | attachments: make(map[*attachedClient]struct{}), | |
| 58 | + | }, | |
| 59 | + | ctx: source.ctx, cancel: func() {}, | |
| 60 | + | tabs: []*tab{newTab(nil, domain.Size{Cols: 80, Rows: 23})}, | |
| 61 | + | } | |
| 62 | + | publishTiledPaneOwners(target, target.tabs[0]) | |
| 63 | + | d.mu.Lock() | |
| 64 | + | d.sessions[target.id] = target | |
| 65 | + | d.mu.Unlock() | |
| 66 | + | ||
| 67 | + | transport := &closeTrackingTransport{} | |
| 68 | + | ac.replaceTransport(transport) | |
| 69 | + | rc := d.attachCoordinator(source, nil, ac, true) | |
| 70 | + | token := source.captureAttachmentCapability(ac, transport) | |
| 71 | + | token.lease = rc.attachmentLease(ac) | |
| 72 | + | ac.installTestAttachmentCapability(token) | |
| 73 | + | effect, admitted := ac.beginAttachmentEffect(token) | |
| 74 | + | require.True(t, admitted) | |
| 75 | + | defer effect.End() | |
| 76 | + | previous := protocol.RouteRef{Key: 7, Generation: 3} | |
| 77 | + | exact := protocol.ExactSessionTarget{LifecycleID: target.incarnation, SessionName: target.name} | |
| 78 | + | ac.setRouteSnapshot(protocol.RecentRouteSnapshot{ | |
| 79 | + | Generation: 5, | |
| 80 | + | Active: protocol.RouteRef{Key: 9, Generation: 5}, | |
| 81 | + | Previous: previous, | |
| 82 | + | Entries: []protocol.RecentRouteEntry{testRouteEntry(7, 3, target.name, 7, protocol.RouteKindRemote)}, | |
| 83 | + | }) | |
| 84 | + | ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{ | |
| 85 | + | Targets: []protocol.RouteAttentionTarget{{Ref: previous, Target: exact}}, | |
| 86 | + | }) | |
| 87 | + | ||
| 88 | + | require.NoError(t, d.backSessionForAttachment(effect)) | |
| 89 | + | frames := transport.Sends() | |
| 90 | + | require.Len(t, frames, 1) | |
| 91 | + | require.Equal(t, wire.MsgAttachTarget, frames[0].Type) | |
| 92 | + | handoff, err := wire.UnmarshalAttachTarget(frames[0].Payload) | |
| 93 | + | require.NoError(t, err) | |
| 94 | + | require.Equal(t, protocol.AttachTarget{ | |
| 95 | + | Session: target.name, Intent: protocol.IntentAttach, ExactTarget: &exact, | |
| 96 | + | EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, SamePeer: true, | |
| 97 | + | }, handoff) | |
| 98 | + | } | |
| @@ -3,6 +3,7 @@ package daemon | |||||
| 3 | import ( | 3 | import ( | ||
| 4 | "testing" | 4 | "testing" | ||
| 5 | 5 | ||||
| 6 | + | "github.com/bnema/vev/internal/domain" | |||
| 6 | "github.com/bnema/vev/internal/protocol" | 7 | "github.com/bnema/vev/internal/protocol" | ||
| 7 | "github.com/bnema/vev/internal/protocol/wire" | 8 | "github.com/bnema/vev/internal/protocol/wire" | ||
| 8 | "github.com/stretchr/testify/require" | 9 | "github.com/stretchr/testify/require" | ||
| @@ -19,12 +20,16 @@ func TestBackSessionUsesClientPreviousRouteAfterSnapshotPublication(t *testing.T | |||||
| 19 | effect, admitted := ac.beginAttachmentEffect(token) | 20 | effect, admitted := ac.beginAttachmentEffect(token) | ||
| 20 | require.True(t, admitted) | 21 | require.True(t, admitted) | ||
| 21 | defer effect.End() | 22 | defer effect.End() | ||
| 23 | + | previous := protocol.RouteRef{Key: 7, Generation: 3} | |||
| 22 | ac.setRouteSnapshot(protocol.RecentRouteSnapshot{ | 24 | ac.setRouteSnapshot(protocol.RecentRouteSnapshot{ | ||
| 23 | Generation: 5, | 25 | Generation: 5, | ||
| 24 | Active: protocol.RouteRef{Key: 9, Generation: 5}, | 26 | Active: protocol.RouteRef{Key: 9, Generation: 5}, | ||
| 25 | - | Previous: protocol.RouteRef{Key: 7, Generation: 3}, | 27 | + | Previous: previous, |
| 26 | Entries: []protocol.RecentRouteEntry{testRouteEntry(7, 3, "previous", 7, protocol.RouteKindLocal)}, | 28 | Entries: []protocol.RecentRouteEntry{testRouteEntry(7, 3, "previous", 7, protocol.RouteKindLocal)}, | ||
| 27 | }) | 29 | }) | ||
| 30 | + | ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{ | |||
| 31 | + | Targets: []protocol.RouteAttentionTarget{{Ref: previous, Target: testRouteTarget("previous", 7)}}, | |||
| 32 | + | }) | |||
| 28 | 33 | ||||
| 29 | require.NoError(t, d.backSessionForAttachment(effect)) | 34 | require.NoError(t, d.backSessionForAttachment(effect)) | ||
| 30 | frames := transport.Sends() | 35 | frames := transport.Sends() | ||
| @@ -43,3 +48,51 @@ func TestBackSessionUsesClientPreviousRouteAfterSnapshotPublication(t *testing.T | |||||
| 43 | require.True(t, found) | 48 | require.True(t, found) | ||
| 44 | require.Equal(t, protocol.RouteNavigationAction{SnapshotGeneration: 5, Key: 7, Generation: 3}, action) | 49 | require.Equal(t, protocol.RouteNavigationAction{SnapshotGeneration: 5, Key: 7, Generation: 3}, action) | ||
| 45 | } | 50 | } | ||
| 51 | + | ||||
| 52 | + | func TestBackSessionOffersPreviousRouteOnCurrentDaemon(t *testing.T) { | |||
| 53 | + | d, source, ac, _ := newManualSessionWithPTYs(t, nil) | |||
| 54 | + | target := &session{ | |||
| 55 | + | sessionCore: sessionCore{ | |||
| 56 | + | id: "target", name: "target", incarnation: testRouteTarget("target", 7).LifecycleID, | |||
| 57 | + | attachments: make(map[*attachedClient]struct{}), | |||
| 58 | + | }, | |||
| 59 | + | ctx: source.ctx, cancel: func() {}, | |||
| 60 | + | tabs: []*tab{newTab(nil, domain.Size{Cols: 80, Rows: 23})}, | |||
| 61 | + | } | |||
| 62 | + | publishTiledPaneOwners(target, target.tabs[0]) | |||
| 63 | + | d.mu.Lock() | |||
| 64 | + | d.sessions[target.id] = target | |||
| 65 | + | d.mu.Unlock() | |||
| 66 | + | ||||
| 67 | + | transport := &closeTrackingTransport{} | |||
| 68 | + | ac.replaceTransport(transport) | |||
| 69 | + | rc := d.attachCoordinator(source, nil, ac, true) | |||
| 70 | + | token := source.captureAttachmentCapability(ac, transport) | |||
| 71 | + | token.lease = rc.attachmentLease(ac) | |||
| 72 | + | ac.installTestAttachmentCapability(token) | |||
| 73 | + | effect, admitted := ac.beginAttachmentEffect(token) | |||
| 74 | + | require.True(t, admitted) | |||
| 75 | + | defer effect.End() | |||
| 76 | + | previous := protocol.RouteRef{Key: 7, Generation: 3} | |||
| 77 | + | exact := protocol.ExactSessionTarget{LifecycleID: target.incarnation, SessionName: target.name} | |||
| 78 | + | ac.setRouteSnapshot(protocol.RecentRouteSnapshot{ | |||
| 79 | + | Generation: 5, | |||
| 80 | + | Active: protocol.RouteRef{Key: 9, Generation: 5}, | |||
| 81 | + | Previous: previous, | |||
| 82 | + | Entries: []protocol.RecentRouteEntry{testRouteEntry(7, 3, target.name, 7, protocol.RouteKindRemote)}, | |||
| 83 | + | }) | |||
| 84 | + | ac.setRouteAttentionSubscription(protocol.RouteAttentionSubscription{ | |||
| 85 | + | Targets: []protocol.RouteAttentionTarget{{Ref: previous, Target: exact}}, | |||
| 86 | + | }) | |||
| 87 | + | ||||
| 88 | + | require.NoError(t, d.backSessionForAttachment(effect)) | |||
| 89 | + | frames := transport.Sends() | |||
| 90 | + | require.Len(t, frames, 1) | |||
| 91 | + | require.Equal(t, wire.MsgAttachTarget, frames[0].Type) | |||
| 92 | + | handoff, err := wire.UnmarshalAttachTarget(frames[0].Payload) | |||
| 93 | + | require.NoError(t, err) | |||
| 94 | + | require.Equal(t, protocol.AttachTarget{ | |||
| 95 | + | Session: target.name, Intent: protocol.IntentAttach, ExactTarget: &exact, | |||
| 96 | + | EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned, SamePeer: true, | |||
| 97 | + | }, handoff) | |||
| 98 | + | } | |||