feat(daemon): resume snatched clients without reclaiming
- commit
- 3de5569eaf16534c0f99ed10fa98159742cf4e94
- author
- brice · · 2026-07-28 00:30 UTC
- parent
- 09a2f38
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
8 files · +713 -48
M internal/usecase/daemon/client.go +19 -13
| @@ -62,15 +62,21 @@ type attachedClient struct { | |||
| 62 | 62 | // captureFrames is keyed by pane ownership, not the tab-local PaneID, so | |
| 63 | 63 | // snapshots cannot leak when an attachment switches tabs or sessions. | |
| 64 | 64 | captureFrames map[*pane]capturedPaneRenderState // only touched while sendMu is held | |
| 65 | - | size domain.Size | |
| 66 | - | keys *keys.Router | |
| 67 | - | sess Guarded[*session] | |
| 68 | - | mouseScan mouse.Scanner | |
| 69 | - | themeMu sync.Mutex | |
| 70 | - | clientTheme themeui.Theme | |
| 71 | - | appliedTheme appliedTheme | |
| 72 | - | lastCursor cursorOut | |
| 73 | - | renderStages renderStageHooks // optional render and handoff observability hooks | |
| 65 | + | // initialSnatchedMu elects exactly one reset-panel sender per role generation. | |
| 66 | + | // A Welcome handshake and displaced-client cleanup may both discover the | |
| 67 | + | // same post-transition snatched role; the loser waits for the elected send. | |
| 68 | + | initialSnatchedMu sync.Mutex | |
| 69 | + | initialSnatchedGeneration uint64 | |
| 70 | + | initialSnatchedAttempt *initialSnatchedPanelAttempt | |
| 71 | + | size domain.Size | |
| 72 | + | keys *keys.Router | |
| 73 | + | sess Guarded[*session] | |
| 74 | + | mouseScan mouse.Scanner | |
| 75 | + | themeMu sync.Mutex | |
| 76 | + | clientTheme themeui.Theme | |
| 77 | + | appliedTheme appliedTheme | |
| 78 | + | lastCursor cursorOut | |
| 79 | + | renderStages renderStageHooks // optional render and handoff observability hooks | |
| 74 | 80 | // previousSession is guarded independently. It is retained through temporary | |
| 75 | 81 | // setSession(nil) hand-offs and cleared only on terminal teardown. | |
| 76 | 82 | previousSession Guarded[*session] | |
| @@ -705,7 +711,7 @@ func (d *Daemon) runConnLoop(ac *attachedClient) { | |||
| 705 | 711 | if err != nil { | |
| 706 | 712 | token := sess.attachmentToken(ac, tr) | |
| 707 | 713 | if token.role == attachmentSnatched { | |
| 708 | - | d.dropSnatchedAttachment(token) | |
| 714 | + | d.parkOrDropSnatchedAttachment(token) | |
| 709 | 715 | } else { | |
| 710 | 716 | d.clientGone(sess, ac, tr, false) | |
| 711 | 717 | } | |
| @@ -843,7 +849,7 @@ func (d *Daemon) handleSnatchedClientFrame(token attachmentRoleToken, f ports.Fr | |||
| 843 | 849 | ac.sendMu.Unlock() | |
| 844 | 850 | if err := d.sendSnatchedPanel(ac, token.transport, token.generation, "", token.effect); err != nil { | |
| 845 | 851 | token.endRoleEffect() | |
| 846 | - | d.dropSnatchedAttachment(token) | |
| 852 | + | d.parkOrDropSnatchedAttachment(token) | |
| 847 | 853 | return true | |
| 848 | 854 | } | |
| 849 | 855 | } | |
| @@ -861,7 +867,7 @@ func (d *Daemon) handleSnatchedClientFrame(token attachmentRoleToken, f ports.Fr | |||
| 861 | 867 | ac.sendMu.Unlock() | |
| 862 | 868 | if err := d.sendSnatchedPanel(ac, token.transport, token.generation, "", token.effect); err != nil { | |
| 863 | 869 | token.endRoleEffect() | |
| 864 | - | d.dropSnatchedAttachment(token) | |
| 870 | + | d.parkOrDropSnatchedAttachment(token) | |
| 865 | 871 | return true | |
| 866 | 872 | } | |
| 867 | 873 | } | |
| @@ -872,7 +878,7 @@ func (d *Daemon) handleSnatchedClientFrame(token attachmentRoleToken, f ports.Fr | |||
| 872 | 878 | case ports.MsgPing: | |
| 873 | 879 | if err := d.sendSnatchedControl(token, framePong()); err != nil { | |
| 874 | 880 | token.endRoleEffect() | |
| 875 | - | d.dropSnatchedAttachment(token) | |
| 881 | + | d.parkOrDropSnatchedAttachment(token) | |
| 876 | 882 | return true | |
| 877 | 883 | } | |
| 878 | 884 | case ports.MsgDetach: | |
| @@ -62,15 +62,21 @@ type attachedClient struct { | |||||
| 62 | // captureFrames is keyed by pane ownership, not the tab-local PaneID, so | 62 | // captureFrames is keyed by pane ownership, not the tab-local PaneID, so | ||
| 63 | // snapshots cannot leak when an attachment switches tabs or sessions. | 63 | // snapshots cannot leak when an attachment switches tabs or sessions. | ||
| 64 | captureFrames map[*pane]capturedPaneRenderState // only touched while sendMu is held | 64 | captureFrames map[*pane]capturedPaneRenderState // only touched while sendMu is held | ||
| 65 | - | size domain.Size | 65 | + | // initialSnatchedMu elects exactly one reset-panel sender per role generation. |
| 66 | - | keys *keys.Router | 66 | + | // A Welcome handshake and displaced-client cleanup may both discover the |
| 67 | - | sess Guarded[*session] | 67 | + | // same post-transition snatched role; the loser waits for the elected send. |
| 68 | - | mouseScan mouse.Scanner | 68 | + | initialSnatchedMu sync.Mutex |
| 69 | - | themeMu sync.Mutex | 69 | + | initialSnatchedGeneration uint64 |
| 70 | - | clientTheme themeui.Theme | 70 | + | initialSnatchedAttempt *initialSnatchedPanelAttempt |
| 71 | - | appliedTheme appliedTheme | 71 | + | size domain.Size |
| 72 | - | lastCursor cursorOut | 72 | + | keys *keys.Router |
| 73 | - | renderStages renderStageHooks // optional render and handoff observability hooks | 73 | + | sess Guarded[*session] |
| 74 | + | mouseScan mouse.Scanner | |||
| 75 | + | themeMu sync.Mutex | |||
| 76 | + | clientTheme themeui.Theme | |||
| 77 | + | appliedTheme appliedTheme | |||
| 78 | + | lastCursor cursorOut | |||
| 79 | + | renderStages renderStageHooks // optional render and handoff observability hooks | |||
| 74 | // previousSession is guarded independently. It is retained through temporary | 80 | // previousSession is guarded independently. It is retained through temporary | ||
| 75 | // setSession(nil) hand-offs and cleared only on terminal teardown. | 81 | // setSession(nil) hand-offs and cleared only on terminal teardown. | ||
| 76 | previousSession Guarded[*session] | 82 | previousSession Guarded[*session] | ||
| @@ -705,7 +711,7 @@ func (d *Daemon) runConnLoop(ac *attachedClient) { | |||||
| 705 | if err != nil { | 711 | if err != nil { | ||
| 706 | token := sess.attachmentToken(ac, tr) | 712 | token := sess.attachmentToken(ac, tr) | ||
| 707 | if token.role == attachmentSnatched { | 713 | if token.role == attachmentSnatched { | ||
| 708 | - | d.dropSnatchedAttachment(token) | 714 | + | d.parkOrDropSnatchedAttachment(token) |
| 709 | } else { | 715 | } else { | ||
| 710 | d.clientGone(sess, ac, tr, false) | 716 | d.clientGone(sess, ac, tr, false) | ||
| 711 | } | 717 | } | ||
| @@ -843,7 +849,7 @@ func (d *Daemon) handleSnatchedClientFrame(token attachmentRoleToken, f ports.Fr | |||||
| 843 | ac.sendMu.Unlock() | 849 | ac.sendMu.Unlock() | ||
| 844 | if err := d.sendSnatchedPanel(ac, token.transport, token.generation, "", token.effect); err != nil { | 850 | if err := d.sendSnatchedPanel(ac, token.transport, token.generation, "", token.effect); err != nil { | ||
| 845 | token.endRoleEffect() | 851 | token.endRoleEffect() | ||
| 846 | - | d.dropSnatchedAttachment(token) | 852 | + | d.parkOrDropSnatchedAttachment(token) |
| 847 | return true | 853 | return true | ||
| 848 | } | 854 | } | ||
| 849 | } | 855 | } | ||
| @@ -861,7 +867,7 @@ func (d *Daemon) handleSnatchedClientFrame(token attachmentRoleToken, f ports.Fr | |||||
| 861 | ac.sendMu.Unlock() | 867 | ac.sendMu.Unlock() | ||
| 862 | if err := d.sendSnatchedPanel(ac, token.transport, token.generation, "", token.effect); err != nil { | 868 | if err := d.sendSnatchedPanel(ac, token.transport, token.generation, "", token.effect); err != nil { | ||
| 863 | token.endRoleEffect() | 869 | token.endRoleEffect() | ||
| 864 | - | d.dropSnatchedAttachment(token) | 870 | + | d.parkOrDropSnatchedAttachment(token) |
| 865 | return true | 871 | return true | ||
| 866 | } | 872 | } | ||
| 867 | } | 873 | } | ||
| @@ -872,7 +878,7 @@ func (d *Daemon) handleSnatchedClientFrame(token attachmentRoleToken, f ports.Fr | |||||
| 872 | case ports.MsgPing: | 878 | case ports.MsgPing: | ||
| 873 | if err := d.sendSnatchedControl(token, framePong()); err != nil { | 879 | if err := d.sendSnatchedControl(token, framePong()); err != nil { | ||
| 874 | token.endRoleEffect() | 880 | token.endRoleEffect() | ||
| 875 | - | d.dropSnatchedAttachment(token) | 881 | + | d.parkOrDropSnatchedAttachment(token) |
| 876 | return true | 882 | return true | ||
| 877 | } | 883 | } | ||
| 878 | case ports.MsgDetach: | 884 | case ports.MsgDetach: | ||
M internal/usecase/daemon/daemon.go +44 -11
| @@ -256,6 +256,7 @@ type Daemon struct { | |||
| 256 | 256 | type parkedAttachment struct { | |
| 257 | 257 | sess *session | |
| 258 | 258 | ac *attachedClient | |
| 259 | + | role attachmentRole | |
| 259 | 260 | timer ports.Timer | |
| 260 | 261 | done chan struct{} | |
| 261 | 262 | doneOnce sync.Once | |
| @@ -965,25 +966,57 @@ func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) { | |||
| 965 | 966 | welcomeToken := sess.attachmentToken(ac, tr) | |
| 966 | 967 | welcomeToken.lease = lease | |
| 967 | 968 | welcomeTicket, admitted := ac.beginRoleEffect(welcomeToken) | |
| 968 | - | if lease == nil || expected.transport != tr || !admitted { | |
| 969 | - | d.clientGone(sess, ac, tr, false) | |
| 969 | + | validRole := welcomeToken.role == attachmentSnatched || welcomeToken.role == attachmentActive && lease != nil | |
| 970 | + | if expected.transport != tr || !admitted || !validRole { | |
| 971 | + | if admitted { | |
| 972 | + | welcomeTicket.End() | |
| 973 | + | } | |
| 974 | + | if welcomeToken.role == attachmentSnatched { | |
| 975 | + | d.parkOrDropSnatchedAttachment(welcomeToken) | |
| 976 | + | } else { | |
| 977 | + | d.clientGone(sess, ac, tr, false) | |
| 978 | + | } | |
| 970 | 979 | return | |
| 971 | 980 | } | |
| 972 | 981 | if err := ac.sendExpectedTransportForRole(expected, frameWelcome(sess, ac), welcomeTicket); err != nil { | |
| 973 | 982 | welcomeTicket.End() | |
| 983 | + | if welcomeToken.role == attachmentSnatched { | |
| 984 | + | d.parkOrDropSnatchedAttachment(welcomeToken) | |
| 985 | + | } else { | |
| 986 | + | d.clientGone(sess, ac, tr, false) | |
| 987 | + | } | |
| 988 | + | return | |
| 989 | + | } | |
| 990 | + | // Welcome is the only effect allowed to retain the route-time role. Release | |
| 991 | + | // it before discovering post-handshake authority so a replacement blocked | |
| 992 | + | // behind the send can publish its generation and lease first. | |
| 993 | + | welcomeTicket.End() | |
| 994 | + | postWelcomeToken, postWelcomeTicket, admitted := ac.beginCurrentRoleEffect(sess, tr) | |
| 995 | + | if !admitted { | |
| 974 | 996 | d.clientGone(sess, ac, tr, false) | |
| 975 | 997 | return | |
| 976 | 998 | } | |
| 977 | - | if !rc.markAttachmentReady(lease) { | |
| 978 | - | welcomeTicket.End() | |
| 999 | + | if postWelcomeToken.role == attachmentSnatched { | |
| 1000 | + | if err := d.sendInitialSnatchedPanel(postWelcomeToken, postWelcomeTicket); err != nil { | |
| 1001 | + | postWelcomeTicket.End() | |
| 1002 | + | d.parkOrDropSnatchedAttachment(postWelcomeToken) | |
| 1003 | + | return | |
| 1004 | + | } | |
| 1005 | + | postWelcomeTicket.End() | |
| 1006 | + | d.runConnLoop(ac) | |
| 1007 | + | _ = tr.Close() | |
| 1008 | + | return | |
| 1009 | + | } | |
| 1010 | + | postWelcomeLease := postWelcomeToken.lease | |
| 1011 | + | if postWelcomeToken.role != attachmentActive || rc == nil || postWelcomeLease == nil || !rc.markAttachmentReady(postWelcomeLease) { | |
| 1012 | + | postWelcomeTicket.End() | |
| 979 | 1013 | // The attachment was displaced or detached while Welcome was in flight; | |
| 980 | 1014 | // never let this stale handshake emit an Output frame. | |
| 981 | 1015 | d.clientGone(sess, ac, tr, false) | |
| 982 | 1016 | return | |
| 983 | 1017 | } | |
| 984 | - | welcomeTicket.End() | |
| 1018 | + | postWelcomeTicket.End() | |
| 985 | 1019 | paintToken := sess.attachmentToken(ac, tr) | |
| 986 | - | paintToken.lease = lease | |
| 987 | 1020 | if !d.firstPaintForTransition(paintToken) { | |
| 988 | 1021 | d.clientGone(sess, ac, tr, false) | |
| 989 | 1022 | return | |
| @@ -1092,10 +1125,14 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||
| 1092 | 1125 | } | |
| 1093 | 1126 | term := terminalEnv{TrueColor: h.TrueColor} | |
| 1094 | 1127 | ||
| 1095 | - | if h.Intent == ports.IntentResume { | |
| 1128 | + | // A non-zero token is an authoritative resume credential. If it is unknown, | |
| 1129 | + | // expired, or raced with lifecycle teardown, fail closed instead of routing | |
| 1130 | + | // the Hello as an ordinary attach that could create or replace ownership. | |
| 1131 | + | if h.ResumeToken != 0 { | |
| 1096 | 1132 | if sess, ac, ok, err := d.resumeParked(h, tr, sz); ok || err != nil { | |
| 1097 | 1133 | return sess, ac, err | |
| 1098 | 1134 | } | |
| 1135 | + | return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |
| 1099 | 1136 | } | |
| 1100 | 1137 | if h.Intent == ports.IntentResume || h.Intent == ports.IntentAttach { | |
| 1101 | 1138 | ctx := d.serveCtx | |
| @@ -1134,7 +1171,6 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||
| 1134 | 1171 | return nil, nil, err | |
| 1135 | 1172 | } | |
| 1136 | 1173 | } | |
| 1137 | - | d.purgeParkedForSessionLocked(sess) | |
| 1138 | 1174 | ac, err := d.finishAttach(sess, tr, sz, term, h) | |
| 1139 | 1175 | return sess, ac, err | |
| 1140 | 1176 | ||
| @@ -1145,7 +1181,6 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||
| 1145 | 1181 | d.mu.Unlock() | |
| 1146 | 1182 | return nil, nil, err | |
| 1147 | 1183 | } | |
| 1148 | - | d.purgeParkedForSessionLocked(sess) | |
| 1149 | 1184 | ac, err := d.finishAttach(sess, tr, sz, term, h) | |
| 1150 | 1185 | return sess, ac, err | |
| 1151 | 1186 | ||
| @@ -1167,7 +1202,6 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||
| 1167 | 1202 | d.mu.Unlock() | |
| 1168 | 1203 | return nil, nil, err | |
| 1169 | 1204 | } | |
| 1170 | - | d.purgeParkedForSessionLocked(sess) | |
| 1171 | 1205 | ac, err := d.finishAttach(sess, tr, sz, term, h) | |
| 1172 | 1206 | return sess, ac, err | |
| 1173 | 1207 | ||
| @@ -1187,7 +1221,6 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||
| 1187 | 1221 | return nil, nil, err | |
| 1188 | 1222 | } | |
| 1189 | 1223 | } | |
| 1190 | - | d.purgeParkedForSessionLocked(sess) | |
| 1191 | 1224 | ac, err := d.finishAttach(sess, tr, sz, term, h) | |
| 1192 | 1225 | return sess, ac, err | |
| 1193 | 1226 | ||
| @@ -256,6 +256,7 @@ type Daemon struct { | |||||
| 256 | type parkedAttachment struct { | 256 | type parkedAttachment struct { | ||
| 257 | sess *session | 257 | sess *session | ||
| 258 | ac *attachedClient | 258 | ac *attachedClient | ||
| 259 | + | role attachmentRole | |||
| 259 | timer ports.Timer | 260 | timer ports.Timer | ||
| 260 | done chan struct{} | 261 | done chan struct{} | ||
| 261 | doneOnce sync.Once | 262 | doneOnce sync.Once | ||
| @@ -965,25 +966,57 @@ func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) { | |||||
| 965 | welcomeToken := sess.attachmentToken(ac, tr) | 966 | welcomeToken := sess.attachmentToken(ac, tr) | ||
| 966 | welcomeToken.lease = lease | 967 | welcomeToken.lease = lease | ||
| 967 | welcomeTicket, admitted := ac.beginRoleEffect(welcomeToken) | 968 | welcomeTicket, admitted := ac.beginRoleEffect(welcomeToken) | ||
| 968 | - | if lease == nil || expected.transport != tr || !admitted { | 969 | + | validRole := welcomeToken.role == attachmentSnatched || welcomeToken.role == attachmentActive && lease != nil |
| 969 | - | d.clientGone(sess, ac, tr, false) | 970 | + | if expected.transport != tr || !admitted || !validRole { |
| 971 | + | if admitted { | |||
| 972 | + | welcomeTicket.End() | |||
| 973 | + | } | |||
| 974 | + | if welcomeToken.role == attachmentSnatched { | |||
| 975 | + | d.parkOrDropSnatchedAttachment(welcomeToken) | |||
| 976 | + | } else { | |||
| 977 | + | d.clientGone(sess, ac, tr, false) | |||
| 978 | + | } | |||
| 970 | return | 979 | return | ||
| 971 | } | 980 | } | ||
| 972 | if err := ac.sendExpectedTransportForRole(expected, frameWelcome(sess, ac), welcomeTicket); err != nil { | 981 | if err := ac.sendExpectedTransportForRole(expected, frameWelcome(sess, ac), welcomeTicket); err != nil { | ||
| 973 | welcomeTicket.End() | 982 | welcomeTicket.End() | ||
| 983 | + | if welcomeToken.role == attachmentSnatched { | |||
| 984 | + | d.parkOrDropSnatchedAttachment(welcomeToken) | |||
| 985 | + | } else { | |||
| 986 | + | d.clientGone(sess, ac, tr, false) | |||
| 987 | + | } | |||
| 988 | + | return | |||
| 989 | + | } | |||
| 990 | + | // Welcome is the only effect allowed to retain the route-time role. Release | |||
| 991 | + | // it before discovering post-handshake authority so a replacement blocked | |||
| 992 | + | // behind the send can publish its generation and lease first. | |||
| 993 | + | welcomeTicket.End() | |||
| 994 | + | postWelcomeToken, postWelcomeTicket, admitted := ac.beginCurrentRoleEffect(sess, tr) | |||
| 995 | + | if !admitted { | |||
| 974 | d.clientGone(sess, ac, tr, false) | 996 | d.clientGone(sess, ac, tr, false) | ||
| 975 | return | 997 | return | ||
| 976 | } | 998 | } | ||
| 977 | - | if !rc.markAttachmentReady(lease) { | 999 | + | if postWelcomeToken.role == attachmentSnatched { |
| 978 | - | welcomeTicket.End() | 1000 | + | if err := d.sendInitialSnatchedPanel(postWelcomeToken, postWelcomeTicket); err != nil { |
| 1001 | + | postWelcomeTicket.End() | |||
| 1002 | + | d.parkOrDropSnatchedAttachment(postWelcomeToken) | |||
| 1003 | + | return | |||
| 1004 | + | } | |||
| 1005 | + | postWelcomeTicket.End() | |||
| 1006 | + | d.runConnLoop(ac) | |||
| 1007 | + | _ = tr.Close() | |||
| 1008 | + | return | |||
| 1009 | + | } | |||
| 1010 | + | postWelcomeLease := postWelcomeToken.lease | |||
| 1011 | + | if postWelcomeToken.role != attachmentActive || rc == nil || postWelcomeLease == nil || !rc.markAttachmentReady(postWelcomeLease) { | |||
| 1012 | + | postWelcomeTicket.End() | |||
| 979 | // The attachment was displaced or detached while Welcome was in flight; | 1013 | // The attachment was displaced or detached while Welcome was in flight; | ||
| 980 | // never let this stale handshake emit an Output frame. | 1014 | // never let this stale handshake emit an Output frame. | ||
| 981 | d.clientGone(sess, ac, tr, false) | 1015 | d.clientGone(sess, ac, tr, false) | ||
| 982 | return | 1016 | return | ||
| 983 | } | 1017 | } | ||
| 984 | - | welcomeTicket.End() | 1018 | + | postWelcomeTicket.End() |
| 985 | paintToken := sess.attachmentToken(ac, tr) | 1019 | paintToken := sess.attachmentToken(ac, tr) | ||
| 986 | - | paintToken.lease = lease | |||
| 987 | if !d.firstPaintForTransition(paintToken) { | 1020 | if !d.firstPaintForTransition(paintToken) { | ||
| 988 | d.clientGone(sess, ac, tr, false) | 1021 | d.clientGone(sess, ac, tr, false) | ||
| 989 | return | 1022 | return | ||
| @@ -1092,10 +1125,14 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||||
| 1092 | } | 1125 | } | ||
| 1093 | term := terminalEnv{TrueColor: h.TrueColor} | 1126 | term := terminalEnv{TrueColor: h.TrueColor} | ||
| 1094 | 1127 | ||||
| 1095 | - | if h.Intent == ports.IntentResume { | 1128 | + | // A non-zero token is an authoritative resume credential. If it is unknown, |
| 1129 | + | // expired, or raced with lifecycle teardown, fail closed instead of routing | |||
| 1130 | + | // the Hello as an ordinary attach that could create or replace ownership. | |||
| 1131 | + | if h.ResumeToken != 0 { | |||
| 1096 | if sess, ac, ok, err := d.resumeParked(h, tr, sz); ok || err != nil { | 1132 | if sess, ac, ok, err := d.resumeParked(h, tr, sz); ok || err != nil { | ||
| 1097 | return sess, ac, err | 1133 | return sess, ac, err | ||
| 1098 | } | 1134 | } | ||
| 1135 | + | return nil, nil, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |||
| 1099 | } | 1136 | } | ||
| 1100 | if h.Intent == ports.IntentResume || h.Intent == ports.IntentAttach { | 1137 | if h.Intent == ports.IntentResume || h.Intent == ports.IntentAttach { | ||
| 1101 | ctx := d.serveCtx | 1138 | ctx := d.serveCtx | ||
| @@ -1134,7 +1171,6 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||||
| 1134 | return nil, nil, err | 1171 | return nil, nil, err | ||
| 1135 | } | 1172 | } | ||
| 1136 | } | 1173 | } | ||
| 1137 | - | d.purgeParkedForSessionLocked(sess) | |||
| 1138 | ac, err := d.finishAttach(sess, tr, sz, term, h) | 1174 | ac, err := d.finishAttach(sess, tr, sz, term, h) | ||
| 1139 | return sess, ac, err | 1175 | return sess, ac, err | ||
| 1140 | 1176 | ||||
| @@ -1145,7 +1181,6 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||||
| 1145 | d.mu.Unlock() | 1181 | d.mu.Unlock() | ||
| 1146 | return nil, nil, err | 1182 | return nil, nil, err | ||
| 1147 | } | 1183 | } | ||
| 1148 | - | d.purgeParkedForSessionLocked(sess) | |||
| 1149 | ac, err := d.finishAttach(sess, tr, sz, term, h) | 1184 | ac, err := d.finishAttach(sess, tr, sz, term, h) | ||
| 1150 | return sess, ac, err | 1185 | return sess, ac, err | ||
| 1151 | 1186 | ||||
| @@ -1167,7 +1202,6 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||||
| 1167 | d.mu.Unlock() | 1202 | d.mu.Unlock() | ||
| 1168 | return nil, nil, err | 1203 | return nil, nil, err | ||
| 1169 | } | 1204 | } | ||
| 1170 | - | d.purgeParkedForSessionLocked(sess) | |||
| 1171 | ac, err := d.finishAttach(sess, tr, sz, term, h) | 1205 | ac, err := d.finishAttach(sess, tr, sz, term, h) | ||
| 1172 | return sess, ac, err | 1206 | return sess, ac, err | ||
| 1173 | 1207 | ||||
| @@ -1187,7 +1221,6 @@ func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedCl | |||||
| 1187 | return nil, nil, err | 1221 | return nil, nil, err | ||
| 1188 | } | 1222 | } | ||
| 1189 | } | 1223 | } | ||
| 1190 | - | d.purgeParkedForSessionLocked(sess) | |||
| 1191 | ac, err := d.finishAttach(sess, tr, sz, term, h) | 1224 | ac, err := d.finishAttach(sess, tr, sz, term, h) | ||
| 1192 | return sess, ac, err | 1225 | return sess, ac, err | ||
| 1193 | 1226 | ||||
M internal/usecase/daemon/resume.go +32 -9
| @@ -45,6 +45,10 @@ func (d *Daemon) prepareParkAttachment(sess *session, ac *attachedClient) bool { | |||
| 45 | 45 | } | |
| 46 | 46 | ||
| 47 | 47 | func (d *Daemon) parkAttachment(sess *session, ac *attachedClient) bool { | |
| 48 | + | return d.parkAttachmentAs(sess, ac, attachmentActive) | |
| 49 | + | } | |
| 50 | + | ||
| 51 | + | func (d *Daemon) parkAttachmentAs(sess *session, ac *attachedClient, role attachmentRole) bool { | |
| 48 | 52 | if !d.prepareParkAttachment(sess, ac) { | |
| 49 | 53 | return false | |
| 50 | 54 | } | |
| @@ -54,10 +58,17 @@ func (d *Daemon) parkAttachment(sess *session, ac *attachedClient) bool { | |||
| 54 | 58 | // remain outside d.mu: sendMu is ordered before the daemon lock. | |
| 55 | 59 | ac.clearCaptureFrames() | |
| 56 | 60 | d.mu.Lock() | |
| 57 | - | if d.closing { | |
| 61 | + | if d.closing || d.sessions[sess.id] != sess { | |
| 58 | 62 | d.mu.Unlock() | |
| 59 | 63 | return false | |
| 60 | 64 | } | |
| 65 | + | if role == attachmentActive { | |
| 66 | + | sess.mu.Lock() | |
| 67 | + | if sess.client != nil { | |
| 68 | + | role = attachmentSnatched | |
| 69 | + | } | |
| 70 | + | sess.mu.Unlock() | |
| 71 | + | } | |
| 61 | 72 | token := ac.resumeToken | |
| 62 | 73 | if old := d.parked[token]; old != nil { | |
| 63 | 74 | old.timer.Stop() | |
| @@ -65,7 +76,7 @@ func (d *Daemon) parkAttachment(sess *session, ac *attachedClient) bool { | |||
| 65 | 76 | } | |
| 66 | 77 | grace := d.resumeParkGrace | |
| 67 | 78 | timer := d.clock.NewTimer(grace) | |
| 68 | - | parked := &parkedAttachment{sess: sess, ac: ac, timer: timer, done: make(chan struct{})} | |
| 79 | + | parked := &parkedAttachment{sess: sess, ac: ac, role: role, timer: timer, done: make(chan struct{})} | |
| 69 | 80 | ac.parked = true | |
| 70 | 81 | d.parked[token] = parked | |
| 71 | 82 | d.mu.Unlock() | |
| @@ -112,9 +123,18 @@ func (p *parkedAttachment) closeDone() { | |||
| 112 | 123 | p.doneOnce.Do(func() { close(p.done) }) | |
| 113 | 124 | } | |
| 114 | 125 | ||
| 126 | + | // demoteParkedActiveForSessionLocked preserves resumable predecessors when a | |
| 127 | + | // new active owner is published. Caller holds d.mu. | |
| 128 | + | func (d *Daemon) demoteParkedActiveForSessionLocked(sess *session) { | |
| 129 | + | for _, parked := range d.parked { | |
| 130 | + | if parked.sess == sess && parked.role == attachmentActive { | |
| 131 | + | parked.role = attachmentSnatched | |
| 132 | + | } | |
| 133 | + | } | |
| 134 | + | } | |
| 135 | + | ||
| 115 | 136 | // purgeParkedForSessionLocked invalidates every parked token for sess. Caller | |
| 116 | - | // holds d.mu. Use before killing or normally attaching to a session so stale | |
| 117 | - | // resume tokens cannot resurrect or replace an active attachment. | |
| 137 | + | // holds d.mu. It is reserved for terminal session kill and daemon shutdown. | |
| 118 | 138 | func (d *Daemon) purgeParkedForSessionLocked(sess *session) { | |
| 119 | 139 | for token, parked := range d.parked { | |
| 120 | 140 | if parked.sess == sess { | |
| @@ -158,13 +178,16 @@ func (d *Daemon) resumeParkedLocked(h ports.Hello, tr ports.Transport, sz domain | |||
| 158 | 178 | } | |
| 159 | 179 | sess := parked.sess | |
| 160 | 180 | registered := d.sessions[sess.id] == sess | |
| 181 | + | if !registered { | |
| 182 | + | d.removeParkedLocked(h.ResumeToken, parked) | |
| 183 | + | d.log.Warn("resume rejected", "session", sess.name, "registered", false) | |
| 184 | + | return nil, nil, false, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |
| 185 | + | } | |
| 161 | 186 | sess.mu.Lock() | |
| 162 | 187 | active := sess.client != nil | |
| 163 | 188 | sess.mu.Unlock() | |
| 164 | - | if !registered || active { | |
| 165 | - | d.removeParkedLocked(h.ResumeToken, parked) | |
| 166 | - | d.log.Warn("resume rejected", "session", sess.name, "registered", registered, "active", active) | |
| 167 | - | return nil, nil, false, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |
| 189 | + | if active && parked.role == attachmentActive { | |
| 190 | + | parked.role = attachmentSnatched | |
| 168 | 191 | } | |
| 169 | 192 | ac := parked.ac | |
| 170 | 193 | if ac.clientID != h.ClientID { | |
| @@ -198,7 +221,7 @@ func (d *Daemon) resumeParkedLocked(h ports.Hello, tr ports.Transport, sz domain | |||
| 198 | 221 | target: sess, | |
| 199 | 222 | next: ac, | |
| 200 | 223 | expectedRole: attachmentDetached, | |
| 201 | - | targetRole: attachmentActive, | |
| 224 | + | targetRole: parked.role, | |
| 202 | 225 | expectedTransport: ac.transportSnapshot(), | |
| 203 | 226 | ready: false, | |
| 204 | 227 | }) | |
| @@ -45,6 +45,10 @@ func (d *Daemon) prepareParkAttachment(sess *session, ac *attachedClient) bool { | |||||
| 45 | } | 45 | } | ||
| 46 | 46 | ||||
| 47 | func (d *Daemon) parkAttachment(sess *session, ac *attachedClient) bool { | 47 | func (d *Daemon) parkAttachment(sess *session, ac *attachedClient) bool { | ||
| 48 | + | return d.parkAttachmentAs(sess, ac, attachmentActive) | |||
| 49 | + | } | |||
| 50 | + | ||||
| 51 | + | func (d *Daemon) parkAttachmentAs(sess *session, ac *attachedClient, role attachmentRole) bool { | |||
| 48 | if !d.prepareParkAttachment(sess, ac) { | 52 | if !d.prepareParkAttachment(sess, ac) { | ||
| 49 | return false | 53 | return false | ||
| 50 | } | 54 | } | ||
| @@ -54,10 +58,17 @@ func (d *Daemon) parkAttachment(sess *session, ac *attachedClient) bool { | |||||
| 54 | // remain outside d.mu: sendMu is ordered before the daemon lock. | 58 | // remain outside d.mu: sendMu is ordered before the daemon lock. | ||
| 55 | ac.clearCaptureFrames() | 59 | ac.clearCaptureFrames() | ||
| 56 | d.mu.Lock() | 60 | d.mu.Lock() | ||
| 57 | - | if d.closing { | 61 | + | if d.closing || d.sessions[sess.id] != sess { |
| 58 | d.mu.Unlock() | 62 | d.mu.Unlock() | ||
| 59 | return false | 63 | return false | ||
| 60 | } | 64 | } | ||
| 65 | + | if role == attachmentActive { | |||
| 66 | + | sess.mu.Lock() | |||
| 67 | + | if sess.client != nil { | |||
| 68 | + | role = attachmentSnatched | |||
| 69 | + | } | |||
| 70 | + | sess.mu.Unlock() | |||
| 71 | + | } | |||
| 61 | token := ac.resumeToken | 72 | token := ac.resumeToken | ||
| 62 | if old := d.parked[token]; old != nil { | 73 | if old := d.parked[token]; old != nil { | ||
| 63 | old.timer.Stop() | 74 | old.timer.Stop() | ||
| @@ -65,7 +76,7 @@ func (d *Daemon) parkAttachment(sess *session, ac *attachedClient) bool { | |||||
| 65 | } | 76 | } | ||
| 66 | grace := d.resumeParkGrace | 77 | grace := d.resumeParkGrace | ||
| 67 | timer := d.clock.NewTimer(grace) | 78 | timer := d.clock.NewTimer(grace) | ||
| 68 | - | parked := &parkedAttachment{sess: sess, ac: ac, timer: timer, done: make(chan struct{})} | 79 | + | parked := &parkedAttachment{sess: sess, ac: ac, role: role, timer: timer, done: make(chan struct{})} |
| 69 | ac.parked = true | 80 | ac.parked = true | ||
| 70 | d.parked[token] = parked | 81 | d.parked[token] = parked | ||
| 71 | d.mu.Unlock() | 82 | d.mu.Unlock() | ||
| @@ -112,9 +123,18 @@ func (p *parkedAttachment) closeDone() { | |||||
| 112 | p.doneOnce.Do(func() { close(p.done) }) | 123 | p.doneOnce.Do(func() { close(p.done) }) | ||
| 113 | } | 124 | } | ||
| 114 | 125 | ||||
| 126 | + | // demoteParkedActiveForSessionLocked preserves resumable predecessors when a | |||
| 127 | + | // new active owner is published. Caller holds d.mu. | |||
| 128 | + | func (d *Daemon) demoteParkedActiveForSessionLocked(sess *session) { | |||
| 129 | + | for _, parked := range d.parked { | |||
| 130 | + | if parked.sess == sess && parked.role == attachmentActive { | |||
| 131 | + | parked.role = attachmentSnatched | |||
| 132 | + | } | |||
| 133 | + | } | |||
| 134 | + | } | |||
| 135 | + | ||||
| 115 | // purgeParkedForSessionLocked invalidates every parked token for sess. Caller | 136 | // purgeParkedForSessionLocked invalidates every parked token for sess. Caller | ||
| 116 | - | // holds d.mu. Use before killing or normally attaching to a session so stale | 137 | + | // holds d.mu. It is reserved for terminal session kill and daemon shutdown. |
| 117 | - | // resume tokens cannot resurrect or replace an active attachment. | |||
| 118 | func (d *Daemon) purgeParkedForSessionLocked(sess *session) { | 138 | func (d *Daemon) purgeParkedForSessionLocked(sess *session) { | ||
| 119 | for token, parked := range d.parked { | 139 | for token, parked := range d.parked { | ||
| 120 | if parked.sess == sess { | 140 | if parked.sess == sess { | ||
| @@ -158,13 +178,16 @@ func (d *Daemon) resumeParkedLocked(h ports.Hello, tr ports.Transport, sz domain | |||||
| 158 | } | 178 | } | ||
| 159 | sess := parked.sess | 179 | sess := parked.sess | ||
| 160 | registered := d.sessions[sess.id] == sess | 180 | registered := d.sessions[sess.id] == sess | ||
| 181 | + | if !registered { | |||
| 182 | + | d.removeParkedLocked(h.ResumeToken, parked) | |||
| 183 | + | d.log.Warn("resume rejected", "session", sess.name, "registered", false) | |||
| 184 | + | return nil, nil, false, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |||
| 185 | + | } | |||
| 161 | sess.mu.Lock() | 186 | sess.mu.Lock() | ||
| 162 | active := sess.client != nil | 187 | active := sess.client != nil | ||
| 163 | sess.mu.Unlock() | 188 | sess.mu.Unlock() | ||
| 164 | - | if !registered || active { | 189 | + | if active && parked.role == attachmentActive { |
| 165 | - | d.removeParkedLocked(h.ResumeToken, parked) | 190 | + | parked.role = attachmentSnatched |
| 166 | - | d.log.Warn("resume rejected", "session", sess.name, "registered", registered, "active", active) | |||
| 167 | - | return nil, nil, false, &protoErr{ports.ErrNoSuchSession, "resume token is no longer valid"} | |||
| 168 | } | 191 | } | ||
| 169 | ac := parked.ac | 192 | ac := parked.ac | ||
| 170 | if ac.clientID != h.ClientID { | 193 | if ac.clientID != h.ClientID { | ||
| @@ -198,7 +221,7 @@ func (d *Daemon) resumeParkedLocked(h ports.Hello, tr ports.Transport, sz domain | |||||
| 198 | target: sess, | 221 | target: sess, | ||
| 199 | next: ac, | 222 | next: ac, | ||
| 200 | expectedRole: attachmentDetached, | 223 | expectedRole: attachmentDetached, | ||
| 201 | - | targetRole: attachmentActive, | 224 | + | targetRole: parked.role, |
| 202 | expectedTransport: ac.transportSnapshot(), | 225 | expectedTransport: ac.transportSnapshot(), | ||
| 203 | ready: false, | 226 | ready: false, | ||
| 204 | }) | 227 | }) | ||
M internal/usecase/daemon/resume_test.go +7 -5
| @@ -606,7 +606,7 @@ func TestKilledSessionPurgesParkedResumeToken(t *testing.T) { | |||
| 606 | 606 | require.False(t, ok, "killed session cannot be resumed") | |
| 607 | 607 | } | |
| 608 | 608 | ||
| 609 | - | func TestStaleParkedTokenCannotStealActiveAttachment(t *testing.T) { | |
| 609 | + | func TestParkedPredecessorResumesSnatchedWithoutStealingActiveAttachment(t *testing.T) { | |
| 610 | 610 | pty, release := newBlockingPTY(t) | |
| 611 | 611 | defer release() | |
| 612 | 612 | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| @@ -622,13 +622,15 @@ func TestStaleParkedTokenCannotStealActiveAttachment(t *testing.T) { | |||
| 622 | 622 | require.NotSame(t, oldAC, activeAC) | |
| 623 | 623 | ||
| 624 | 624 | d.mu.Lock() | |
| 625 | - | _, parked := d.parked[token] | |
| 625 | + | parked := d.parked[token] | |
| 626 | 626 | d.mu.Unlock() | |
| 627 | + | require.NotNil(t, parked, "normal attach preserves the parked predecessor") | |
| 628 | + | require.Equal(t, attachmentSnatched, parked.role) | |
| 627 | 629 | _, resumedAC, ok, err := d.resumeParked(helloResumeCapable(ports.IntentResume, "work", token), &closeTrackingTransport{}, domain.Size{Cols: 80, Rows: 24}) | |
| 628 | - | require.False(t, parked, "normal attach invalidates stale parked token") | |
| 629 | 630 | require.NoError(t, err) | |
| 630 | - | require.False(t, ok) | |
| 631 | - | require.Nil(t, resumedAC) | |
| 631 | + | require.True(t, ok) | |
| 632 | + | require.Same(t, oldAC, resumedAC) | |
| 633 | + | require.Equal(t, attachmentSnatched, sess.attachmentRole(resumedAC)) | |
| 632 | 634 | require.Same(t, activeAC, sess.client) | |
| 633 | 635 | } | |
| 634 | 636 | ||
| @@ -606,7 +606,7 @@ func TestKilledSessionPurgesParkedResumeToken(t *testing.T) { | |||||
| 606 | require.False(t, ok, "killed session cannot be resumed") | 606 | require.False(t, ok, "killed session cannot be resumed") | ||
| 607 | } | 607 | } | ||
| 608 | 608 | ||||
| 609 | - | func TestStaleParkedTokenCannotStealActiveAttachment(t *testing.T) { | 609 | + | func TestParkedPredecessorResumesSnatchedWithoutStealingActiveAttachment(t *testing.T) { |
| 610 | pty, release := newBlockingPTY(t) | 610 | pty, release := newBlockingPTY(t) | ||
| 611 | defer release() | 611 | defer release() | ||
| 612 | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | 612 | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | ||
| @@ -622,13 +622,15 @@ func TestStaleParkedTokenCannotStealActiveAttachment(t *testing.T) { | |||||
| 622 | require.NotSame(t, oldAC, activeAC) | 622 | require.NotSame(t, oldAC, activeAC) | ||
| 623 | 623 | ||||
| 624 | d.mu.Lock() | 624 | d.mu.Lock() | ||
| 625 | - | _, parked := d.parked[token] | 625 | + | parked := d.parked[token] |
| 626 | d.mu.Unlock() | 626 | d.mu.Unlock() | ||
| 627 | + | require.NotNil(t, parked, "normal attach preserves the parked predecessor") | |||
| 628 | + | require.Equal(t, attachmentSnatched, parked.role) | |||
| 627 | _, resumedAC, ok, err := d.resumeParked(helloResumeCapable(ports.IntentResume, "work", token), &closeTrackingTransport{}, domain.Size{Cols: 80, Rows: 24}) | 629 | _, resumedAC, ok, err := d.resumeParked(helloResumeCapable(ports.IntentResume, "work", token), &closeTrackingTransport{}, domain.Size{Cols: 80, Rows: 24}) | ||
| 628 | - | require.False(t, parked, "normal attach invalidates stale parked token") | |||
| 629 | require.NoError(t, err) | 630 | require.NoError(t, err) | ||
| 630 | - | require.False(t, ok) | 631 | + | require.True(t, ok) |
| 631 | - | require.Nil(t, resumedAC) | 632 | + | require.Same(t, oldAC, resumedAC) |
| 633 | + | require.Equal(t, attachmentSnatched, sess.attachmentRole(resumedAC)) | |||
| 632 | require.Same(t, activeAC, sess.client) | 634 | require.Same(t, activeAC, sess.client) | ||
| 633 | } | 635 | } | ||
| 634 | 636 | ||||
A internal/usecase/daemon/role_aware_parking_test.go +482 -0
| @@ -0,0 +1,482 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "errors" | |
| 5 | + | "sync" | |
| 6 | + | "testing" | |
| 7 | + | ||
| 8 | + | "github.com/stretchr/testify/require" | |
| 9 | + | ||
| 10 | + | "github.com/bnema/vev/internal/domain" | |
| 11 | + | "github.com/bnema/vev/internal/ports" | |
| 12 | + | themeui "github.com/bnema/vev/internal/usecase/theme" | |
| 13 | + | "github.com/bnema/vev/pkg/renderer" | |
| 14 | + | ) | |
| 15 | + | ||
| 16 | + | func TestActiveLinkLossParksActiveAndDetachesCoordinator(t *testing.T) { | |
| 17 | + | pty, release := newBlockingPTY(t) | |
| 18 | + | defer release() | |
| 19 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 20 | + | ||
| 21 | + | transport := &closeTrackingTransport{} | |
| 22 | + | sess, active, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), transport) | |
| 23 | + | require.NoError(t, err) | |
| 24 | + | token := active.resumeToken | |
| 25 | + | rc := sess.renderCoordinator() | |
| 26 | + | require.NotNil(t, rc.attachmentLease(active)) | |
| 27 | + | ||
| 28 | + | d.clientGone(sess, active, transport, false) | |
| 29 | + | ||
| 30 | + | d.mu.Lock() | |
| 31 | + | parked := d.parked[token] | |
| 32 | + | d.mu.Unlock() | |
| 33 | + | require.NotNil(t, parked) | |
| 34 | + | require.Equal(t, attachmentActive, parked.role) | |
| 35 | + | require.Nil(t, sess.client) | |
| 36 | + | require.Nil(t, rc.attachmentLease(active), "active loss should detach coordinator ownership") | |
| 37 | + | } | |
| 38 | + | ||
| 39 | + | func TestSnatchedLinkLossParksWithoutDisturbingActiveOwner(t *testing.T) { | |
| 40 | + | pty, release := newBlockingPTY(t) | |
| 41 | + | defer release() | |
| 42 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 43 | + | ||
| 44 | + | oldTransport := &closeTrackingTransport{} | |
| 45 | + | sess, old, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport) | |
| 46 | + | require.NoError(t, err) | |
| 47 | + | oldToken := old.resumeToken | |
| 48 | + | ||
| 49 | + | activeTransport := &closeTrackingTransport{} | |
| 50 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), activeTransport) | |
| 51 | + | require.NoError(t, err) | |
| 52 | + | d.attachmentCleanupWg.Wait() | |
| 53 | + | lease := sess.renderCoordinator().attachmentLease(active) | |
| 54 | + | require.NotNil(t, lease) | |
| 55 | + | ||
| 56 | + | d.runConnLoop(old) | |
| 57 | + | ||
| 58 | + | d.mu.Lock() | |
| 59 | + | parked := d.parked[oldToken] | |
| 60 | + | d.mu.Unlock() | |
| 61 | + | require.NotNil(t, parked, "resumable snatched attachment should survive link loss") | |
| 62 | + | require.Equal(t, attachmentSnatched, parked.role) | |
| 63 | + | require.Same(t, active, sess.client) | |
| 64 | + | require.Same(t, lease, sess.renderCoordinator().attachmentLease(active)) | |
| 65 | + | require.False(t, activeTransport.Closed()) | |
| 66 | + | } | |
| 67 | + | ||
| 68 | + | type sendErrorTransport struct { | |
| 69 | + | closeTrackingTransport | |
| 70 | + | } | |
| 71 | + | ||
| 72 | + | func (*sendErrorTransport) Send(ports.Frame) error { return errors.New("send failed") } | |
| 73 | + | ||
| 74 | + | func TestSnatchedPanelSendFailureParksResumableAttachment(t *testing.T) { | |
| 75 | + | pty, release := newBlockingPTY(t) | |
| 76 | + | defer release() | |
| 77 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 78 | + | ||
| 79 | + | failed := &sendErrorTransport{} | |
| 80 | + | sess, waiting, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), failed) | |
| 81 | + | require.NoError(t, err) | |
| 82 | + | token := waiting.resumeToken | |
| 83 | + | ||
| 84 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), &closeTrackingTransport{}) | |
| 85 | + | require.NoError(t, err) | |
| 86 | + | d.attachmentCleanupWg.Wait() | |
| 87 | + | ||
| 88 | + | d.mu.Lock() | |
| 89 | + | parked := d.parked[token] | |
| 90 | + | d.mu.Unlock() | |
| 91 | + | require.NotNil(t, parked, "failed snatched panel should park a resumable attachment") | |
| 92 | + | require.Equal(t, attachmentSnatched, parked.role) | |
| 93 | + | require.Nil(t, waiting.transport(), "parking should revoke the exact failed transport") | |
| 94 | + | require.True(t, failed.Closed()) | |
| 95 | + | require.Same(t, active, sess.client) | |
| 96 | + | } | |
| 97 | + | ||
| 98 | + | func TestStaleSnatchedTransportCannotParkFreshIncarnation(t *testing.T) { | |
| 99 | + | pty, release := newBlockingPTY(t) | |
| 100 | + | defer release() | |
| 101 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 102 | + | ||
| 103 | + | oldTransport := &closeTrackingTransport{} | |
| 104 | + | sess, waiting, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport) | |
| 105 | + | require.NoError(t, err) | |
| 106 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), &closeTrackingTransport{}) | |
| 107 | + | require.NoError(t, err) | |
| 108 | + | d.attachmentCleanupWg.Wait() | |
| 109 | + | stale := sess.attachmentToken(waiting, oldTransport) | |
| 110 | + | fresh := &closeTrackingTransport{} | |
| 111 | + | waiting.replaceTransport(fresh) | |
| 112 | + | ||
| 113 | + | require.False(t, d.parkOrDropSnatchedAttachment(stale)) | |
| 114 | + | d.mu.Lock() | |
| 115 | + | _, parked := d.parked[waiting.resumeToken] | |
| 116 | + | d.mu.Unlock() | |
| 117 | + | require.False(t, parked) | |
| 118 | + | require.Equal(t, attachmentSnatched, sess.attachmentRole(waiting)) | |
| 119 | + | require.Same(t, fresh, waiting.transport()) | |
| 120 | + | require.False(t, fresh.Closed()) | |
| 121 | + | require.Same(t, active, sess.client) | |
| 122 | + | } | |
| 123 | + | ||
| 124 | + | func TestSnatchedParkExpiryDoesNotAffectActiveOwner(t *testing.T) { | |
| 125 | + | pty, release := newBlockingPTY(t) | |
| 126 | + | defer release() | |
| 127 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 128 | + | ||
| 129 | + | waitingTransport := &closeTrackingTransport{} | |
| 130 | + | sess, waiting, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), waitingTransport) | |
| 131 | + | require.NoError(t, err) | |
| 132 | + | token := waiting.resumeToken | |
| 133 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), &closeTrackingTransport{}) | |
| 134 | + | require.NoError(t, err) | |
| 135 | + | d.attachmentCleanupWg.Wait() | |
| 136 | + | lease := sess.renderCoordinator().attachmentLease(active) | |
| 137 | + | ||
| 138 | + | d.runConnLoop(waiting) | |
| 139 | + | d.mu.Lock() | |
| 140 | + | parked := d.parked[token] | |
| 141 | + | d.mu.Unlock() | |
| 142 | + | require.NotNil(t, parked) | |
| 143 | + | ||
| 144 | + | d.expireParked(token, parked) | |
| 145 | + | ||
| 146 | + | d.mu.Lock() | |
| 147 | + | _, retained := d.parked[token] | |
| 148 | + | d.mu.Unlock() | |
| 149 | + | require.False(t, retained) | |
| 150 | + | require.Same(t, active, sess.client) | |
| 151 | + | require.Same(t, lease, sess.renderCoordinator().attachmentLease(active)) | |
| 152 | + | } | |
| 153 | + | ||
| 154 | + | func TestNonzeroResumeTokenOverridesOrdinaryRoutingFallbacks(t *testing.T) { | |
| 155 | + | pty, release := newBlockingPTY(t) | |
| 156 | + | defer release() | |
| 157 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 158 | + | ||
| 159 | + | work, active, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), &closeTrackingTransport{}) | |
| 160 | + | require.NoError(t, err) | |
| 161 | + | const unknownToken = uint64(0xdecafbad) | |
| 162 | + | for _, tc := range []struct { | |
| 163 | + | name string | |
| 164 | + | intent uint8 | |
| 165 | + | target string | |
| 166 | + | }{ | |
| 167 | + | {name: "attach existing", intent: ports.IntentAttach, target: "work"}, | |
| 168 | + | {name: "create unique", intent: ports.IntentNew, target: "fallback"}, | |
| 169 | + | {name: "create ephemeral", intent: ports.IntentEphemeral}, | |
| 170 | + | } { | |
| 171 | + | t.Run(tc.name, func(t *testing.T) { | |
| 172 | + | h := helloResumeCapable(tc.intent, tc.target, unknownToken) | |
| 173 | + | routedSession, routedClient, routeErr := d.route(h, &closeTrackingTransport{}) | |
| 174 | + | require.Error(t, routeErr) | |
| 175 | + | require.Nil(t, routedSession) | |
| 176 | + | require.Nil(t, routedClient) | |
| 177 | + | var protocolErr *protoErr | |
| 178 | + | require.ErrorAs(t, routeErr, &protocolErr) | |
| 179 | + | require.Equal(t, ports.ErrNoSuchSession, protocolErr.code) | |
| 180 | + | require.Same(t, active, work.client) | |
| 181 | + | d.mu.Lock() | |
| 182 | + | require.Len(t, d.sessions, 1) | |
| 183 | + | require.Nil(t, d.findByNameLocked("fallback")) | |
| 184 | + | d.mu.Unlock() | |
| 185 | + | }) | |
| 186 | + | } | |
| 187 | + | ||
| 188 | + | _, replacement, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), &closeTrackingTransport{}) | |
| 189 | + | require.NoError(t, err, "a zero token keeps ordinary attach semantics") | |
| 190 | + | require.NotSame(t, active, replacement) | |
| 191 | + | require.Same(t, replacement, work.client) | |
| 192 | + | require.Equal(t, attachmentSnatched, work.attachmentRole(active)) | |
| 193 | + | } | |
| 194 | + | ||
| 195 | + | func TestHandleHelloExpiredSnatchedResumeFailsClosedAfterSessionEnds(t *testing.T) { | |
| 196 | + | workPTY, releaseWork := newBlockingPTY(t) | |
| 197 | + | keeperPTY, releaseKeeper := newBlockingPTY(t) | |
| 198 | + | fallbackPTY, releaseFallback := newBlockingPTY(t) | |
| 199 | + | defer releaseWork() | |
| 200 | + | defer releaseKeeper() | |
| 201 | + | defer releaseFallback() | |
| 202 | + | d := newTestDaemon(t, newFactorySeq(t, workPTY, keeperPTY, fallbackPTY), stubClock{}) | |
| 203 | + | ||
| 204 | + | oldTransport := &closeTrackingTransport{} | |
| 205 | + | work, old, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport) | |
| 206 | + | require.NoError(t, err) | |
| 207 | + | oldToken := old.resumeToken | |
| 208 | + | ||
| 209 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), &closeTrackingTransport{}) | |
| 210 | + | require.NoError(t, err) | |
| 211 | + | d.attachmentCleanupWg.Wait() | |
| 212 | + | require.Equal(t, attachmentSnatched, work.attachmentRole(old)) | |
| 213 | + | require.Same(t, active, work.client) | |
| 214 | + | ||
| 215 | + | d.runConnLoop(old) | |
| 216 | + | d.mu.Lock() | |
| 217 | + | parked := d.parked[oldToken] | |
| 218 | + | d.mu.Unlock() | |
| 219 | + | require.NotNil(t, parked) | |
| 220 | + | require.Equal(t, attachmentSnatched, parked.role) | |
| 221 | + | ||
| 222 | + | keeper, _, err := d.route(helloResumeCapable(ports.IntentNew, "keeper", 0), &closeTrackingTransport{}) | |
| 223 | + | require.NoError(t, err) | |
| 224 | + | require.NoError(t, d.killSession(work, ports.ReasonSessionKilled, false)) | |
| 225 | + | ||
| 226 | + | d.mu.Lock() | |
| 227 | + | _, tokenRetained := d.parked[oldToken] | |
| 228 | + | stoppedWork, lifecycleRetained := d.stopped["work"] | |
| 229 | + | d.mu.Unlock() | |
| 230 | + | require.False(t, tokenRetained, "terminal lifecycle teardown must purge the parked token") | |
| 231 | + | require.True(t, lifecycleRetained, "the stopped lifecycle makes attach fallback observable") | |
| 232 | + | ||
| 233 | + | reconnect := &closeTrackingTransport{} | |
| 234 | + | d.handleHello(reconnect, ports.Frame{ | |
| 235 | + | Type: ports.MsgHello, | |
| 236 | + | Payload: ports.MarshalHello(helloResumeCapable(ports.IntentResume, "work", oldToken)), | |
| 237 | + | }) | |
| 238 | + | ||
| 239 | + | frames := reconnect.Sends() | |
| 240 | + | require.Len(t, frames, 1) | |
| 241 | + | require.Equal(t, ports.MsgError, frames[0].Type) | |
| 242 | + | wireErr, err := ports.UnmarshalErrorMsg(frames[0].Payload) | |
| 243 | + | require.NoError(t, err) | |
| 244 | + | require.Equal(t, ports.ErrNoSuchSession, wireErr.Code) | |
| 245 | + | require.True(t, reconnect.Closed()) | |
| 246 | + | d.mu.Lock() | |
| 247 | + | require.Same(t, keeper, d.findByNameLocked("keeper")) | |
| 248 | + | require.Nil(t, d.findByNameLocked("work"), "expired resume must not recreate the stopped session") | |
| 249 | + | require.Equal(t, stoppedWork, d.stopped["work"]) | |
| 250 | + | require.Len(t, d.sessions, 1) | |
| 251 | + | d.mu.Unlock() | |
| 252 | + | require.Nil(t, work.client, "expired resume must not reclaim the ended lifecycle") | |
| 253 | + | require.Nil(t, active.currentSession()) | |
| 254 | + | } | |
| 255 | + | ||
| 256 | + | func TestSnatchedResumeHandshakeSendsWelcomeAndPanelWithoutReplacingOwner(t *testing.T) { | |
| 257 | + | pty, release := newBlockingPTY(t) | |
| 258 | + | defer release() | |
| 259 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 260 | + | ||
| 261 | + | oldTransport := &closeTrackingTransport{} | |
| 262 | + | sess, old, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport) | |
| 263 | + | require.NoError(t, err) | |
| 264 | + | token := old.resumeToken | |
| 265 | + | d.clientGone(sess, old, oldTransport, false) | |
| 266 | + | ||
| 267 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), &closeTrackingTransport{}) | |
| 268 | + | require.NoError(t, err) | |
| 269 | + | ||
| 270 | + | resumedTransport := &closeTrackingTransport{} | |
| 271 | + | hello := helloResumeCapable(ports.IntentResume, "work", token) | |
| 272 | + | d.handleHello(resumedTransport, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}) | |
| 273 | + | ||
| 274 | + | frames := resumedTransport.Sends() | |
| 275 | + | require.GreaterOrEqual(t, len(frames), 2) | |
| 276 | + | require.Equal(t, ports.MsgWelcome, frames[0].Type) | |
| 277 | + | welcome, err := ports.UnmarshalWelcome(frames[0].Payload) | |
| 278 | + | require.NoError(t, err) | |
| 279 | + | newToken := welcome.ResumeToken | |
| 280 | + | require.NotZero(t, newToken) | |
| 281 | + | require.NotEqual(t, token, newToken, "successful snatched resume rotates the credential") | |
| 282 | + | require.Equal(t, newToken, old.resumeToken, "the Welcome publishes the one rotated credential") | |
| 283 | + | require.Equal(t, ports.MsgOutput, frames[1].Type) | |
| 284 | + | output, err := ports.UnmarshalOutput(frames[1].Payload) | |
| 285 | + | require.NoError(t, err) | |
| 286 | + | require.Contains(t, string(output.Data), "Session snatched") | |
| 287 | + | d.mu.Lock() | |
| 288 | + | _, oldTokenParked := d.parked[token] | |
| 289 | + | parked := d.parked[newToken] | |
| 290 | + | d.mu.Unlock() | |
| 291 | + | require.False(t, oldTokenParked, "the consumed credential cannot be reused") | |
| 292 | + | require.NotNil(t, parked) | |
| 293 | + | require.Equal(t, attachmentSnatched, parked.role, "the rotated credential remains bound to the snatched role") | |
| 294 | + | require.Same(t, active, sess.client) | |
| 295 | + | ||
| 296 | + | replay := &closeTrackingTransport{} | |
| 297 | + | d.handleHello(replay, ports.Frame{ | |
| 298 | + | Type: ports.MsgHello, | |
| 299 | + | Payload: ports.MarshalHello(helloResumeCapable(ports.IntentResume, "work", token)), | |
| 300 | + | }) | |
| 301 | + | replayFrames := replay.Sends() | |
| 302 | + | require.Len(t, replayFrames, 1) | |
| 303 | + | require.Equal(t, ports.MsgError, replayFrames[0].Type) | |
| 304 | + | replayErr, err := ports.UnmarshalErrorMsg(replayFrames[0].Payload) | |
| 305 | + | require.NoError(t, err) | |
| 306 | + | require.Equal(t, ports.ErrNoSuchSession, replayErr.Code) | |
| 307 | + | require.True(t, replay.Closed()) | |
| 308 | + | require.Same(t, active, sess.client, "replaying the old credential must not replace active ownership") | |
| 309 | + | d.mu.Lock() | |
| 310 | + | require.Same(t, parked, d.parked[newToken]) | |
| 311 | + | d.mu.Unlock() | |
| 312 | + | } | |
| 313 | + | ||
| 314 | + | func TestResumeWelcomeRechecksRoleAfterReplacement(t *testing.T) { | |
| 315 | + | pty, release := newBlockingPTY(t) | |
| 316 | + | defer release() | |
| 317 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 318 | + | ||
| 319 | + | oldTransport := &closeTrackingTransport{} | |
| 320 | + | sess, resumed, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport) | |
| 321 | + | require.NoError(t, err) | |
| 322 | + | token := resumed.resumeToken | |
| 323 | + | d.clientGone(sess, resumed, oldTransport, false) | |
| 324 | + | ||
| 325 | + | blocked := newWelcomeBlockingTransport(t) | |
| 326 | + | handshakeDone := make(chan struct{}) | |
| 327 | + | go func() { | |
| 328 | + | hello := helloResumeCapable(ports.IntentResume, "work", token) | |
| 329 | + | d.handleHello(blocked.tr, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}) | |
| 330 | + | close(handshakeDone) | |
| 331 | + | }() | |
| 332 | + | awaitTestCompletion(t, blocked.welcomeEntered, "resume did not start Welcome") | |
| 333 | + | welcome := awaitFrame(t, blocked.sends, ports.MsgWelcome) | |
| 334 | + | require.Equal(t, ports.MsgWelcome, welcome.Type) | |
| 335 | + | requireNoCoordinatorOutputFrame(t, blocked.sends) | |
| 336 | + | resumedLease := sess.renderCoordinator().attachmentLease(resumed) | |
| 337 | + | require.NotNil(t, resumedLease) | |
| 338 | + | require.False(t, resumedLease.ready, "blocked Welcome must keep the resumed lease unready") | |
| 339 | + | ||
| 340 | + | resumeGateFrozen := make(chan struct{}) | |
| 341 | + | var frozenOnce sync.Once | |
| 342 | + | d.afterRoleEffectGateFrozen = func(_ string, ac *attachedClient) { | |
| 343 | + | if ac == resumed { | |
| 344 | + | frozenOnce.Do(func() { close(resumeGateFrozen) }) | |
| 345 | + | } | |
| 346 | + | } | |
| 347 | + | type attachResult struct { | |
| 348 | + | ac *attachedClient | |
| 349 | + | err error | |
| 350 | + | } | |
| 351 | + | replacementResult := make(chan attachResult, 1) | |
| 352 | + | go func() { | |
| 353 | + | _, replacement, routeErr := d.route( | |
| 354 | + | helloResumeCapable(ports.IntentAttach, "work", 0), | |
| 355 | + | &closeTrackingTransport{}, | |
| 356 | + | ) | |
| 357 | + | replacementResult <- attachResult{ac: replacement, err: routeErr} | |
| 358 | + | }() | |
| 359 | + | awaitTestCompletion(t, resumeGateFrozen, "replacement did not freeze the resumed Welcome role") | |
| 360 | + | ||
| 361 | + | blocked.release() | |
| 362 | + | replacement := awaitTestValue(t, replacementResult, "replacement did not finish after Welcome") | |
| 363 | + | require.NoError(t, replacement.err) | |
| 364 | + | d.attachmentCleanupWg.Wait() | |
| 365 | + | panelFrame := awaitFrame(t, blocked.sends, ports.MsgOutput) | |
| 366 | + | panel, err := ports.UnmarshalOutput(panelFrame.Payload) | |
| 367 | + | require.NoError(t, err) | |
| 368 | + | require.Zero(t, panel.BaseStateNum, "post-Welcome snatched panel must reset output state") | |
| 369 | + | require.Contains(t, string(panel.Data), "Session snatched") | |
| 370 | + | requireNoCoordinatorOutputFrame(t, blocked.sends) | |
| 371 | + | ||
| 372 | + | require.Same(t, replacement.ac, sess.client) | |
| 373 | + | require.Equal(t, attachmentSnatched, sess.attachmentRole(resumed)) | |
| 374 | + | rc := sess.renderCoordinator() | |
| 375 | + | rc.mu.Lock() | |
| 376 | + | require.False(t, resumedLease.ready, "replacement during Welcome must not ready the stale active lease") | |
| 377 | + | rc.mu.Unlock() | |
| 378 | + | require.Nil(t, rc.attachmentLease(resumed)) | |
| 379 | + | require.NotNil(t, rc.attachmentLease(replacement.ac)) | |
| 380 | + | ||
| 381 | + | blocked.finish() | |
| 382 | + | awaitTestCompletion(t, handshakeDone, "resumed snatched connection did not continue after replacement") | |
| 383 | + | } | |
| 384 | + | ||
| 385 | + | func TestSnatchedResumePreservesOwnerCapabilityAndCanReclaim(t *testing.T) { | |
| 386 | + | pty, release := newBlockingPTY(t) | |
| 387 | + | defer release() | |
| 388 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 389 | + | ||
| 390 | + | oldTransport := &closeTrackingTransport{} | |
| 391 | + | sess, waiting, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport) | |
| 392 | + | require.NoError(t, err) | |
| 393 | + | token := waiting.resumeToken | |
| 394 | + | waiting.setClientTheme(themeui.Theme{ | |
| 395 | + | Foreground: renderer.RGB{R: 90, G: 10, B: 10}, | |
| 396 | + | Background: renderer.RGB{R: 20, G: 1, B: 1}, | |
| 397 | + | HasFG: true, HasBG: true, Known: true, | |
| 398 | + | }) | |
| 399 | + | d.clientGone(sess, waiting, oldTransport, false) | |
| 400 | + | ||
| 401 | + | activeTransport := &closeTrackingTransport{} | |
| 402 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), activeTransport) | |
| 403 | + | require.NoError(t, err) | |
| 404 | + | rc := sess.renderCoordinator() | |
| 405 | + | activeGeneration := active.roleGeneration.Load() | |
| 406 | + | activeLease := rc.attachmentLease(active) | |
| 407 | + | require.NotNil(t, activeLease) | |
| 408 | + | activeTheme := themeui.Theme{ | |
| 409 | + | Foreground: renderer.RGB{R: 10, G: 80, B: 20}, | |
| 410 | + | Background: renderer.RGB{R: 1, G: 15, B: 3}, | |
| 411 | + | HasFG: true, HasBG: true, Known: true, | |
| 412 | + | } | |
| 413 | + | active.setClientTheme(activeTheme) | |
| 414 | + | require.True(t, d.applyHostTheme(sess, active, activeTheme, false)) | |
| 415 | + | assertSessionDefaultColors(t, sess, activeTheme.Foreground, activeTheme.Background) | |
| 416 | + | ||
| 417 | + | resumedTransport := &closeTrackingTransport{} | |
| 418 | + | resumedSess, resumed, ok, err := d.resumeParked( | |
| 419 | + | helloResumeCapable(ports.IntentResume, "work", token), | |
| 420 | + | resumedTransport, | |
| 421 | + | domain.Size{Cols: 80, Rows: 24}, | |
| 422 | + | ) | |
| 423 | + | require.NoError(t, err) | |
| 424 | + | require.True(t, ok) | |
| 425 | + | require.Same(t, sess, resumedSess) | |
| 426 | + | require.Same(t, waiting, resumed) | |
| 427 | + | require.Same(t, active, sess.client) | |
| 428 | + | require.Equal(t, activeGeneration, active.roleGeneration.Load()) | |
| 429 | + | require.Same(t, activeLease, rc.attachmentLease(active)) | |
| 430 | + | require.Nil(t, rc.attachmentLease(waiting), "snatched resume must not acquire a render lease") | |
| 431 | + | require.Equal(t, attachmentSnatched, sess.attachmentRole(waiting)) | |
| 432 | + | assertSessionDefaultColors(t, sess, activeTheme.Foreground, activeTheme.Background) | |
| 433 | + | ||
| 434 | + | resumedGeneration := waiting.roleGeneration.Load() | |
| 435 | + | require.False(t, d.handleSnatchedClientFrame(sess.attachmentToken(waiting, resumedTransport), ports.Frame{ | |
| 436 | + | Type: ports.MsgInput, | |
| 437 | + | Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}), | |
| 438 | + | })) | |
| 439 | + | d.attachmentCleanupWg.Wait() | |
| 440 | + | ||
| 441 | + | require.Same(t, waiting, sess.client) | |
| 442 | + | require.Equal(t, attachmentActive, sess.attachmentRole(waiting)) | |
| 443 | + | require.Greater(t, waiting.roleGeneration.Load(), resumedGeneration) | |
| 444 | + | require.NotNil(t, rc.attachmentLease(waiting)) | |
| 445 | + | require.Equal(t, attachmentSnatched, sess.attachmentRole(active)) | |
| 446 | + | } | |
| 447 | + | ||
| 448 | + | func TestOrdinaryAttachDemotesParkedActiveAndAllowsSnatchedResume(t *testing.T) { | |
| 449 | + | pty, release := newBlockingPTY(t) | |
| 450 | + | defer release() | |
| 451 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 452 | + | ||
| 453 | + | oldTransport := &closeTrackingTransport{} | |
| 454 | + | sess, old, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport) | |
| 455 | + | require.NoError(t, err) | |
| 456 | + | token := old.resumeToken | |
| 457 | + | d.clientGone(sess, old, oldTransport, false) | |
| 458 | + | ||
| 459 | + | activeTransport := &closeTrackingTransport{} | |
| 460 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), activeTransport) | |
| 461 | + | require.NoError(t, err) | |
| 462 | + | ||
| 463 | + | d.mu.Lock() | |
| 464 | + | parked := d.parked[token] | |
| 465 | + | d.mu.Unlock() | |
| 466 | + | require.NotNil(t, parked, "ordinary attach must preserve the predecessor's resume token") | |
| 467 | + | require.Equal(t, attachmentSnatched, parked.role) | |
| 468 | + | require.Equal(t, token, old.resumeToken) | |
| 469 | + | ||
| 470 | + | resumedTransport := &closeTrackingTransport{} | |
| 471 | + | resumedSess, resumed, ok, err := d.resumeParked( | |
| 472 | + | helloResumeCapable(ports.IntentResume, "work", token), | |
| 473 | + | resumedTransport, | |
| 474 | + | domain.Size{Cols: 80, Rows: 24}, | |
| 475 | + | ) | |
| 476 | + | require.NoError(t, err) | |
| 477 | + | require.True(t, ok) | |
| 478 | + | require.Same(t, sess, resumedSess) | |
| 479 | + | require.Same(t, old, resumed) | |
| 480 | + | require.Equal(t, attachmentSnatched, sess.attachmentRole(old)) | |
| 481 | + | require.Same(t, active, sess.client) | |
| 482 | + | } | |
| @@ -0,0 +1,482 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "errors" | |||
| 5 | + | "sync" | |||
| 6 | + | "testing" | |||
| 7 | + | ||||
| 8 | + | "github.com/stretchr/testify/require" | |||
| 9 | + | ||||
| 10 | + | "github.com/bnema/vev/internal/domain" | |||
| 11 | + | "github.com/bnema/vev/internal/ports" | |||
| 12 | + | themeui "github.com/bnema/vev/internal/usecase/theme" | |||
| 13 | + | "github.com/bnema/vev/pkg/renderer" | |||
| 14 | + | ) | |||
| 15 | + | ||||
| 16 | + | func TestActiveLinkLossParksActiveAndDetachesCoordinator(t *testing.T) { | |||
| 17 | + | pty, release := newBlockingPTY(t) | |||
| 18 | + | defer release() | |||
| 19 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 20 | + | ||||
| 21 | + | transport := &closeTrackingTransport{} | |||
| 22 | + | sess, active, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), transport) | |||
| 23 | + | require.NoError(t, err) | |||
| 24 | + | token := active.resumeToken | |||
| 25 | + | rc := sess.renderCoordinator() | |||
| 26 | + | require.NotNil(t, rc.attachmentLease(active)) | |||
| 27 | + | ||||
| 28 | + | d.clientGone(sess, active, transport, false) | |||
| 29 | + | ||||
| 30 | + | d.mu.Lock() | |||
| 31 | + | parked := d.parked[token] | |||
| 32 | + | d.mu.Unlock() | |||
| 33 | + | require.NotNil(t, parked) | |||
| 34 | + | require.Equal(t, attachmentActive, parked.role) | |||
| 35 | + | require.Nil(t, sess.client) | |||
| 36 | + | require.Nil(t, rc.attachmentLease(active), "active loss should detach coordinator ownership") | |||
| 37 | + | } | |||
| 38 | + | ||||
| 39 | + | func TestSnatchedLinkLossParksWithoutDisturbingActiveOwner(t *testing.T) { | |||
| 40 | + | pty, release := newBlockingPTY(t) | |||
| 41 | + | defer release() | |||
| 42 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 43 | + | ||||
| 44 | + | oldTransport := &closeTrackingTransport{} | |||
| 45 | + | sess, old, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport) | |||
| 46 | + | require.NoError(t, err) | |||
| 47 | + | oldToken := old.resumeToken | |||
| 48 | + | ||||
| 49 | + | activeTransport := &closeTrackingTransport{} | |||
| 50 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), activeTransport) | |||
| 51 | + | require.NoError(t, err) | |||
| 52 | + | d.attachmentCleanupWg.Wait() | |||
| 53 | + | lease := sess.renderCoordinator().attachmentLease(active) | |||
| 54 | + | require.NotNil(t, lease) | |||
| 55 | + | ||||
| 56 | + | d.runConnLoop(old) | |||
| 57 | + | ||||
| 58 | + | d.mu.Lock() | |||
| 59 | + | parked := d.parked[oldToken] | |||
| 60 | + | d.mu.Unlock() | |||
| 61 | + | require.NotNil(t, parked, "resumable snatched attachment should survive link loss") | |||
| 62 | + | require.Equal(t, attachmentSnatched, parked.role) | |||
| 63 | + | require.Same(t, active, sess.client) | |||
| 64 | + | require.Same(t, lease, sess.renderCoordinator().attachmentLease(active)) | |||
| 65 | + | require.False(t, activeTransport.Closed()) | |||
| 66 | + | } | |||
| 67 | + | ||||
| 68 | + | type sendErrorTransport struct { | |||
| 69 | + | closeTrackingTransport | |||
| 70 | + | } | |||
| 71 | + | ||||
| 72 | + | func (*sendErrorTransport) Send(ports.Frame) error { return errors.New("send failed") } | |||
| 73 | + | ||||
| 74 | + | func TestSnatchedPanelSendFailureParksResumableAttachment(t *testing.T) { | |||
| 75 | + | pty, release := newBlockingPTY(t) | |||
| 76 | + | defer release() | |||
| 77 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 78 | + | ||||
| 79 | + | failed := &sendErrorTransport{} | |||
| 80 | + | sess, waiting, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), failed) | |||
| 81 | + | require.NoError(t, err) | |||
| 82 | + | token := waiting.resumeToken | |||
| 83 | + | ||||
| 84 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), &closeTrackingTransport{}) | |||
| 85 | + | require.NoError(t, err) | |||
| 86 | + | d.attachmentCleanupWg.Wait() | |||
| 87 | + | ||||
| 88 | + | d.mu.Lock() | |||
| 89 | + | parked := d.parked[token] | |||
| 90 | + | d.mu.Unlock() | |||
| 91 | + | require.NotNil(t, parked, "failed snatched panel should park a resumable attachment") | |||
| 92 | + | require.Equal(t, attachmentSnatched, parked.role) | |||
| 93 | + | require.Nil(t, waiting.transport(), "parking should revoke the exact failed transport") | |||
| 94 | + | require.True(t, failed.Closed()) | |||
| 95 | + | require.Same(t, active, sess.client) | |||
| 96 | + | } | |||
| 97 | + | ||||
| 98 | + | func TestStaleSnatchedTransportCannotParkFreshIncarnation(t *testing.T) { | |||
| 99 | + | pty, release := newBlockingPTY(t) | |||
| 100 | + | defer release() | |||
| 101 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 102 | + | ||||
| 103 | + | oldTransport := &closeTrackingTransport{} | |||
| 104 | + | sess, waiting, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport) | |||
| 105 | + | require.NoError(t, err) | |||
| 106 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), &closeTrackingTransport{}) | |||
| 107 | + | require.NoError(t, err) | |||
| 108 | + | d.attachmentCleanupWg.Wait() | |||
| 109 | + | stale := sess.attachmentToken(waiting, oldTransport) | |||
| 110 | + | fresh := &closeTrackingTransport{} | |||
| 111 | + | waiting.replaceTransport(fresh) | |||
| 112 | + | ||||
| 113 | + | require.False(t, d.parkOrDropSnatchedAttachment(stale)) | |||
| 114 | + | d.mu.Lock() | |||
| 115 | + | _, parked := d.parked[waiting.resumeToken] | |||
| 116 | + | d.mu.Unlock() | |||
| 117 | + | require.False(t, parked) | |||
| 118 | + | require.Equal(t, attachmentSnatched, sess.attachmentRole(waiting)) | |||
| 119 | + | require.Same(t, fresh, waiting.transport()) | |||
| 120 | + | require.False(t, fresh.Closed()) | |||
| 121 | + | require.Same(t, active, sess.client) | |||
| 122 | + | } | |||
| 123 | + | ||||
| 124 | + | func TestSnatchedParkExpiryDoesNotAffectActiveOwner(t *testing.T) { | |||
| 125 | + | pty, release := newBlockingPTY(t) | |||
| 126 | + | defer release() | |||
| 127 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 128 | + | ||||
| 129 | + | waitingTransport := &closeTrackingTransport{} | |||
| 130 | + | sess, waiting, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), waitingTransport) | |||
| 131 | + | require.NoError(t, err) | |||
| 132 | + | token := waiting.resumeToken | |||
| 133 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), &closeTrackingTransport{}) | |||
| 134 | + | require.NoError(t, err) | |||
| 135 | + | d.attachmentCleanupWg.Wait() | |||
| 136 | + | lease := sess.renderCoordinator().attachmentLease(active) | |||
| 137 | + | ||||
| 138 | + | d.runConnLoop(waiting) | |||
| 139 | + | d.mu.Lock() | |||
| 140 | + | parked := d.parked[token] | |||
| 141 | + | d.mu.Unlock() | |||
| 142 | + | require.NotNil(t, parked) | |||
| 143 | + | ||||
| 144 | + | d.expireParked(token, parked) | |||
| 145 | + | ||||
| 146 | + | d.mu.Lock() | |||
| 147 | + | _, retained := d.parked[token] | |||
| 148 | + | d.mu.Unlock() | |||
| 149 | + | require.False(t, retained) | |||
| 150 | + | require.Same(t, active, sess.client) | |||
| 151 | + | require.Same(t, lease, sess.renderCoordinator().attachmentLease(active)) | |||
| 152 | + | } | |||
| 153 | + | ||||
| 154 | + | func TestNonzeroResumeTokenOverridesOrdinaryRoutingFallbacks(t *testing.T) { | |||
| 155 | + | pty, release := newBlockingPTY(t) | |||
| 156 | + | defer release() | |||
| 157 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 158 | + | ||||
| 159 | + | work, active, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), &closeTrackingTransport{}) | |||
| 160 | + | require.NoError(t, err) | |||
| 161 | + | const unknownToken = uint64(0xdecafbad) | |||
| 162 | + | for _, tc := range []struct { | |||
| 163 | + | name string | |||
| 164 | + | intent uint8 | |||
| 165 | + | target string | |||
| 166 | + | }{ | |||
| 167 | + | {name: "attach existing", intent: ports.IntentAttach, target: "work"}, | |||
| 168 | + | {name: "create unique", intent: ports.IntentNew, target: "fallback"}, | |||
| 169 | + | {name: "create ephemeral", intent: ports.IntentEphemeral}, | |||
| 170 | + | } { | |||
| 171 | + | t.Run(tc.name, func(t *testing.T) { | |||
| 172 | + | h := helloResumeCapable(tc.intent, tc.target, unknownToken) | |||
| 173 | + | routedSession, routedClient, routeErr := d.route(h, &closeTrackingTransport{}) | |||
| 174 | + | require.Error(t, routeErr) | |||
| 175 | + | require.Nil(t, routedSession) | |||
| 176 | + | require.Nil(t, routedClient) | |||
| 177 | + | var protocolErr *protoErr | |||
| 178 | + | require.ErrorAs(t, routeErr, &protocolErr) | |||
| 179 | + | require.Equal(t, ports.ErrNoSuchSession, protocolErr.code) | |||
| 180 | + | require.Same(t, active, work.client) | |||
| 181 | + | d.mu.Lock() | |||
| 182 | + | require.Len(t, d.sessions, 1) | |||
| 183 | + | require.Nil(t, d.findByNameLocked("fallback")) | |||
| 184 | + | d.mu.Unlock() | |||
| 185 | + | }) | |||
| 186 | + | } | |||
| 187 | + | ||||
| 188 | + | _, replacement, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), &closeTrackingTransport{}) | |||
| 189 | + | require.NoError(t, err, "a zero token keeps ordinary attach semantics") | |||
| 190 | + | require.NotSame(t, active, replacement) | |||
| 191 | + | require.Same(t, replacement, work.client) | |||
| 192 | + | require.Equal(t, attachmentSnatched, work.attachmentRole(active)) | |||
| 193 | + | } | |||
| 194 | + | ||||
| 195 | + | func TestHandleHelloExpiredSnatchedResumeFailsClosedAfterSessionEnds(t *testing.T) { | |||
| 196 | + | workPTY, releaseWork := newBlockingPTY(t) | |||
| 197 | + | keeperPTY, releaseKeeper := newBlockingPTY(t) | |||
| 198 | + | fallbackPTY, releaseFallback := newBlockingPTY(t) | |||
| 199 | + | defer releaseWork() | |||
| 200 | + | defer releaseKeeper() | |||
| 201 | + | defer releaseFallback() | |||
| 202 | + | d := newTestDaemon(t, newFactorySeq(t, workPTY, keeperPTY, fallbackPTY), stubClock{}) | |||
| 203 | + | ||||
| 204 | + | oldTransport := &closeTrackingTransport{} | |||
| 205 | + | work, old, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport) | |||
| 206 | + | require.NoError(t, err) | |||
| 207 | + | oldToken := old.resumeToken | |||
| 208 | + | ||||
| 209 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), &closeTrackingTransport{}) | |||
| 210 | + | require.NoError(t, err) | |||
| 211 | + | d.attachmentCleanupWg.Wait() | |||
| 212 | + | require.Equal(t, attachmentSnatched, work.attachmentRole(old)) | |||
| 213 | + | require.Same(t, active, work.client) | |||
| 214 | + | ||||
| 215 | + | d.runConnLoop(old) | |||
| 216 | + | d.mu.Lock() | |||
| 217 | + | parked := d.parked[oldToken] | |||
| 218 | + | d.mu.Unlock() | |||
| 219 | + | require.NotNil(t, parked) | |||
| 220 | + | require.Equal(t, attachmentSnatched, parked.role) | |||
| 221 | + | ||||
| 222 | + | keeper, _, err := d.route(helloResumeCapable(ports.IntentNew, "keeper", 0), &closeTrackingTransport{}) | |||
| 223 | + | require.NoError(t, err) | |||
| 224 | + | require.NoError(t, d.killSession(work, ports.ReasonSessionKilled, false)) | |||
| 225 | + | ||||
| 226 | + | d.mu.Lock() | |||
| 227 | + | _, tokenRetained := d.parked[oldToken] | |||
| 228 | + | stoppedWork, lifecycleRetained := d.stopped["work"] | |||
| 229 | + | d.mu.Unlock() | |||
| 230 | + | require.False(t, tokenRetained, "terminal lifecycle teardown must purge the parked token") | |||
| 231 | + | require.True(t, lifecycleRetained, "the stopped lifecycle makes attach fallback observable") | |||
| 232 | + | ||||
| 233 | + | reconnect := &closeTrackingTransport{} | |||
| 234 | + | d.handleHello(reconnect, ports.Frame{ | |||
| 235 | + | Type: ports.MsgHello, | |||
| 236 | + | Payload: ports.MarshalHello(helloResumeCapable(ports.IntentResume, "work", oldToken)), | |||
| 237 | + | }) | |||
| 238 | + | ||||
| 239 | + | frames := reconnect.Sends() | |||
| 240 | + | require.Len(t, frames, 1) | |||
| 241 | + | require.Equal(t, ports.MsgError, frames[0].Type) | |||
| 242 | + | wireErr, err := ports.UnmarshalErrorMsg(frames[0].Payload) | |||
| 243 | + | require.NoError(t, err) | |||
| 244 | + | require.Equal(t, ports.ErrNoSuchSession, wireErr.Code) | |||
| 245 | + | require.True(t, reconnect.Closed()) | |||
| 246 | + | d.mu.Lock() | |||
| 247 | + | require.Same(t, keeper, d.findByNameLocked("keeper")) | |||
| 248 | + | require.Nil(t, d.findByNameLocked("work"), "expired resume must not recreate the stopped session") | |||
| 249 | + | require.Equal(t, stoppedWork, d.stopped["work"]) | |||
| 250 | + | require.Len(t, d.sessions, 1) | |||
| 251 | + | d.mu.Unlock() | |||
| 252 | + | require.Nil(t, work.client, "expired resume must not reclaim the ended lifecycle") | |||
| 253 | + | require.Nil(t, active.currentSession()) | |||
| 254 | + | } | |||
| 255 | + | ||||
| 256 | + | func TestSnatchedResumeHandshakeSendsWelcomeAndPanelWithoutReplacingOwner(t *testing.T) { | |||
| 257 | + | pty, release := newBlockingPTY(t) | |||
| 258 | + | defer release() | |||
| 259 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 260 | + | ||||
| 261 | + | oldTransport := &closeTrackingTransport{} | |||
| 262 | + | sess, old, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport) | |||
| 263 | + | require.NoError(t, err) | |||
| 264 | + | token := old.resumeToken | |||
| 265 | + | d.clientGone(sess, old, oldTransport, false) | |||
| 266 | + | ||||
| 267 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), &closeTrackingTransport{}) | |||
| 268 | + | require.NoError(t, err) | |||
| 269 | + | ||||
| 270 | + | resumedTransport := &closeTrackingTransport{} | |||
| 271 | + | hello := helloResumeCapable(ports.IntentResume, "work", token) | |||
| 272 | + | d.handleHello(resumedTransport, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}) | |||
| 273 | + | ||||
| 274 | + | frames := resumedTransport.Sends() | |||
| 275 | + | require.GreaterOrEqual(t, len(frames), 2) | |||
| 276 | + | require.Equal(t, ports.MsgWelcome, frames[0].Type) | |||
| 277 | + | welcome, err := ports.UnmarshalWelcome(frames[0].Payload) | |||
| 278 | + | require.NoError(t, err) | |||
| 279 | + | newToken := welcome.ResumeToken | |||
| 280 | + | require.NotZero(t, newToken) | |||
| 281 | + | require.NotEqual(t, token, newToken, "successful snatched resume rotates the credential") | |||
| 282 | + | require.Equal(t, newToken, old.resumeToken, "the Welcome publishes the one rotated credential") | |||
| 283 | + | require.Equal(t, ports.MsgOutput, frames[1].Type) | |||
| 284 | + | output, err := ports.UnmarshalOutput(frames[1].Payload) | |||
| 285 | + | require.NoError(t, err) | |||
| 286 | + | require.Contains(t, string(output.Data), "Session snatched") | |||
| 287 | + | d.mu.Lock() | |||
| 288 | + | _, oldTokenParked := d.parked[token] | |||
| 289 | + | parked := d.parked[newToken] | |||
| 290 | + | d.mu.Unlock() | |||
| 291 | + | require.False(t, oldTokenParked, "the consumed credential cannot be reused") | |||
| 292 | + | require.NotNil(t, parked) | |||
| 293 | + | require.Equal(t, attachmentSnatched, parked.role, "the rotated credential remains bound to the snatched role") | |||
| 294 | + | require.Same(t, active, sess.client) | |||
| 295 | + | ||||
| 296 | + | replay := &closeTrackingTransport{} | |||
| 297 | + | d.handleHello(replay, ports.Frame{ | |||
| 298 | + | Type: ports.MsgHello, | |||
| 299 | + | Payload: ports.MarshalHello(helloResumeCapable(ports.IntentResume, "work", token)), | |||
| 300 | + | }) | |||
| 301 | + | replayFrames := replay.Sends() | |||
| 302 | + | require.Len(t, replayFrames, 1) | |||
| 303 | + | require.Equal(t, ports.MsgError, replayFrames[0].Type) | |||
| 304 | + | replayErr, err := ports.UnmarshalErrorMsg(replayFrames[0].Payload) | |||
| 305 | + | require.NoError(t, err) | |||
| 306 | + | require.Equal(t, ports.ErrNoSuchSession, replayErr.Code) | |||
| 307 | + | require.True(t, replay.Closed()) | |||
| 308 | + | require.Same(t, active, sess.client, "replaying the old credential must not replace active ownership") | |||
| 309 | + | d.mu.Lock() | |||
| 310 | + | require.Same(t, parked, d.parked[newToken]) | |||
| 311 | + | d.mu.Unlock() | |||
| 312 | + | } | |||
| 313 | + | ||||
| 314 | + | func TestResumeWelcomeRechecksRoleAfterReplacement(t *testing.T) { | |||
| 315 | + | pty, release := newBlockingPTY(t) | |||
| 316 | + | defer release() | |||
| 317 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 318 | + | ||||
| 319 | + | oldTransport := &closeTrackingTransport{} | |||
| 320 | + | sess, resumed, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport) | |||
| 321 | + | require.NoError(t, err) | |||
| 322 | + | token := resumed.resumeToken | |||
| 323 | + | d.clientGone(sess, resumed, oldTransport, false) | |||
| 324 | + | ||||
| 325 | + | blocked := newWelcomeBlockingTransport(t) | |||
| 326 | + | handshakeDone := make(chan struct{}) | |||
| 327 | + | go func() { | |||
| 328 | + | hello := helloResumeCapable(ports.IntentResume, "work", token) | |||
| 329 | + | d.handleHello(blocked.tr, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}) | |||
| 330 | + | close(handshakeDone) | |||
| 331 | + | }() | |||
| 332 | + | awaitTestCompletion(t, blocked.welcomeEntered, "resume did not start Welcome") | |||
| 333 | + | welcome := awaitFrame(t, blocked.sends, ports.MsgWelcome) | |||
| 334 | + | require.Equal(t, ports.MsgWelcome, welcome.Type) | |||
| 335 | + | requireNoCoordinatorOutputFrame(t, blocked.sends) | |||
| 336 | + | resumedLease := sess.renderCoordinator().attachmentLease(resumed) | |||
| 337 | + | require.NotNil(t, resumedLease) | |||
| 338 | + | require.False(t, resumedLease.ready, "blocked Welcome must keep the resumed lease unready") | |||
| 339 | + | ||||
| 340 | + | resumeGateFrozen := make(chan struct{}) | |||
| 341 | + | var frozenOnce sync.Once | |||
| 342 | + | d.afterRoleEffectGateFrozen = func(_ string, ac *attachedClient) { | |||
| 343 | + | if ac == resumed { | |||
| 344 | + | frozenOnce.Do(func() { close(resumeGateFrozen) }) | |||
| 345 | + | } | |||
| 346 | + | } | |||
| 347 | + | type attachResult struct { | |||
| 348 | + | ac *attachedClient | |||
| 349 | + | err error | |||
| 350 | + | } | |||
| 351 | + | replacementResult := make(chan attachResult, 1) | |||
| 352 | + | go func() { | |||
| 353 | + | _, replacement, routeErr := d.route( | |||
| 354 | + | helloResumeCapable(ports.IntentAttach, "work", 0), | |||
| 355 | + | &closeTrackingTransport{}, | |||
| 356 | + | ) | |||
| 357 | + | replacementResult <- attachResult{ac: replacement, err: routeErr} | |||
| 358 | + | }() | |||
| 359 | + | awaitTestCompletion(t, resumeGateFrozen, "replacement did not freeze the resumed Welcome role") | |||
| 360 | + | ||||
| 361 | + | blocked.release() | |||
| 362 | + | replacement := awaitTestValue(t, replacementResult, "replacement did not finish after Welcome") | |||
| 363 | + | require.NoError(t, replacement.err) | |||
| 364 | + | d.attachmentCleanupWg.Wait() | |||
| 365 | + | panelFrame := awaitFrame(t, blocked.sends, ports.MsgOutput) | |||
| 366 | + | panel, err := ports.UnmarshalOutput(panelFrame.Payload) | |||
| 367 | + | require.NoError(t, err) | |||
| 368 | + | require.Zero(t, panel.BaseStateNum, "post-Welcome snatched panel must reset output state") | |||
| 369 | + | require.Contains(t, string(panel.Data), "Session snatched") | |||
| 370 | + | requireNoCoordinatorOutputFrame(t, blocked.sends) | |||
| 371 | + | ||||
| 372 | + | require.Same(t, replacement.ac, sess.client) | |||
| 373 | + | require.Equal(t, attachmentSnatched, sess.attachmentRole(resumed)) | |||
| 374 | + | rc := sess.renderCoordinator() | |||
| 375 | + | rc.mu.Lock() | |||
| 376 | + | require.False(t, resumedLease.ready, "replacement during Welcome must not ready the stale active lease") | |||
| 377 | + | rc.mu.Unlock() | |||
| 378 | + | require.Nil(t, rc.attachmentLease(resumed)) | |||
| 379 | + | require.NotNil(t, rc.attachmentLease(replacement.ac)) | |||
| 380 | + | ||||
| 381 | + | blocked.finish() | |||
| 382 | + | awaitTestCompletion(t, handshakeDone, "resumed snatched connection did not continue after replacement") | |||
| 383 | + | } | |||
| 384 | + | ||||
| 385 | + | func TestSnatchedResumePreservesOwnerCapabilityAndCanReclaim(t *testing.T) { | |||
| 386 | + | pty, release := newBlockingPTY(t) | |||
| 387 | + | defer release() | |||
| 388 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 389 | + | ||||
| 390 | + | oldTransport := &closeTrackingTransport{} | |||
| 391 | + | sess, waiting, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport) | |||
| 392 | + | require.NoError(t, err) | |||
| 393 | + | token := waiting.resumeToken | |||
| 394 | + | waiting.setClientTheme(themeui.Theme{ | |||
| 395 | + | Foreground: renderer.RGB{R: 90, G: 10, B: 10}, | |||
| 396 | + | Background: renderer.RGB{R: 20, G: 1, B: 1}, | |||
| 397 | + | HasFG: true, HasBG: true, Known: true, | |||
| 398 | + | }) | |||
| 399 | + | d.clientGone(sess, waiting, oldTransport, false) | |||
| 400 | + | ||||
| 401 | + | activeTransport := &closeTrackingTransport{} | |||
| 402 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), activeTransport) | |||
| 403 | + | require.NoError(t, err) | |||
| 404 | + | rc := sess.renderCoordinator() | |||
| 405 | + | activeGeneration := active.roleGeneration.Load() | |||
| 406 | + | activeLease := rc.attachmentLease(active) | |||
| 407 | + | require.NotNil(t, activeLease) | |||
| 408 | + | activeTheme := themeui.Theme{ | |||
| 409 | + | Foreground: renderer.RGB{R: 10, G: 80, B: 20}, | |||
| 410 | + | Background: renderer.RGB{R: 1, G: 15, B: 3}, | |||
| 411 | + | HasFG: true, HasBG: true, Known: true, | |||
| 412 | + | } | |||
| 413 | + | active.setClientTheme(activeTheme) | |||
| 414 | + | require.True(t, d.applyHostTheme(sess, active, activeTheme, false)) | |||
| 415 | + | assertSessionDefaultColors(t, sess, activeTheme.Foreground, activeTheme.Background) | |||
| 416 | + | ||||
| 417 | + | resumedTransport := &closeTrackingTransport{} | |||
| 418 | + | resumedSess, resumed, ok, err := d.resumeParked( | |||
| 419 | + | helloResumeCapable(ports.IntentResume, "work", token), | |||
| 420 | + | resumedTransport, | |||
| 421 | + | domain.Size{Cols: 80, Rows: 24}, | |||
| 422 | + | ) | |||
| 423 | + | require.NoError(t, err) | |||
| 424 | + | require.True(t, ok) | |||
| 425 | + | require.Same(t, sess, resumedSess) | |||
| 426 | + | require.Same(t, waiting, resumed) | |||
| 427 | + | require.Same(t, active, sess.client) | |||
| 428 | + | require.Equal(t, activeGeneration, active.roleGeneration.Load()) | |||
| 429 | + | require.Same(t, activeLease, rc.attachmentLease(active)) | |||
| 430 | + | require.Nil(t, rc.attachmentLease(waiting), "snatched resume must not acquire a render lease") | |||
| 431 | + | require.Equal(t, attachmentSnatched, sess.attachmentRole(waiting)) | |||
| 432 | + | assertSessionDefaultColors(t, sess, activeTheme.Foreground, activeTheme.Background) | |||
| 433 | + | ||||
| 434 | + | resumedGeneration := waiting.roleGeneration.Load() | |||
| 435 | + | require.False(t, d.handleSnatchedClientFrame(sess.attachmentToken(waiting, resumedTransport), ports.Frame{ | |||
| 436 | + | Type: ports.MsgInput, | |||
| 437 | + | Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}), | |||
| 438 | + | })) | |||
| 439 | + | d.attachmentCleanupWg.Wait() | |||
| 440 | + | ||||
| 441 | + | require.Same(t, waiting, sess.client) | |||
| 442 | + | require.Equal(t, attachmentActive, sess.attachmentRole(waiting)) | |||
| 443 | + | require.Greater(t, waiting.roleGeneration.Load(), resumedGeneration) | |||
| 444 | + | require.NotNil(t, rc.attachmentLease(waiting)) | |||
| 445 | + | require.Equal(t, attachmentSnatched, sess.attachmentRole(active)) | |||
| 446 | + | } | |||
| 447 | + | ||||
| 448 | + | func TestOrdinaryAttachDemotesParkedActiveAndAllowsSnatchedResume(t *testing.T) { | |||
| 449 | + | pty, release := newBlockingPTY(t) | |||
| 450 | + | defer release() | |||
| 451 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 452 | + | ||||
| 453 | + | oldTransport := &closeTrackingTransport{} | |||
| 454 | + | sess, old, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport) | |||
| 455 | + | require.NoError(t, err) | |||
| 456 | + | token := old.resumeToken | |||
| 457 | + | d.clientGone(sess, old, oldTransport, false) | |||
| 458 | + | ||||
| 459 | + | activeTransport := &closeTrackingTransport{} | |||
| 460 | + | _, active, err := d.route(helloResumeCapable(ports.IntentAttach, "work", 0), activeTransport) | |||
| 461 | + | require.NoError(t, err) | |||
| 462 | + | ||||
| 463 | + | d.mu.Lock() | |||
| 464 | + | parked := d.parked[token] | |||
| 465 | + | d.mu.Unlock() | |||
| 466 | + | require.NotNil(t, parked, "ordinary attach must preserve the predecessor's resume token") | |||
| 467 | + | require.Equal(t, attachmentSnatched, parked.role) | |||
| 468 | + | require.Equal(t, token, old.resumeToken) | |||
| 469 | + | ||||
| 470 | + | resumedTransport := &closeTrackingTransport{} | |||
| 471 | + | resumedSess, resumed, ok, err := d.resumeParked( | |||
| 472 | + | helloResumeCapable(ports.IntentResume, "work", token), | |||
| 473 | + | resumedTransport, | |||
| 474 | + | domain.Size{Cols: 80, Rows: 24}, | |||
| 475 | + | ) | |||
| 476 | + | require.NoError(t, err) | |||
| 477 | + | require.True(t, ok) | |||
| 478 | + | require.Same(t, sess, resumedSess) | |||
| 479 | + | require.Same(t, old, resumed) | |||
| 480 | + | require.Equal(t, attachmentSnatched, sess.attachmentRole(old)) | |||
| 481 | + | require.Same(t, active, sess.client) | |||
| 482 | + | } | |||
M internal/usecase/daemon/role_effect_gate.go +33 -0
| @@ -260,6 +260,39 @@ func (ac *attachedClient) beginRoleEffect(token attachmentRoleToken) (*roleEffec | |||
| 260 | 260 | return &roleEffectTicket{gate: g, token: captured}, true | |
| 261 | 261 | } | |
| 262 | 262 | ||
| 263 | + | // beginCurrentRoleEffect waits out a transition that already froze ac, then | |
| 264 | + | // admits the role capability derived from the session registries. Handshakes | |
| 265 | + | // use it after Welcome has completed so a replacement blocked behind that send | |
| 266 | + | // can publish before readiness, first paint, or the snatched reset panel. | |
| 267 | + | func (ac *attachedClient) beginCurrentRoleEffect(sess *session, tr ports.Transport) (attachmentRoleToken, *roleEffectTicket, bool) { | |
| 268 | + | if ac == nil || sess == nil || tr == nil { | |
| 269 | + | return attachmentRoleToken{}, nil, false | |
| 270 | + | } | |
| 271 | + | for { | |
| 272 | + | token := sess.attachmentToken(ac, tr) | |
| 273 | + | if token.role == attachmentDetached { | |
| 274 | + | return token, nil, false | |
| 275 | + | } | |
| 276 | + | if ticket, admitted := ac.beginRoleEffect(token); admitted { | |
| 277 | + | return token, ticket, true | |
| 278 | + | } | |
| 279 | + | ||
| 280 | + | g := &ac.roleEffects | |
| 281 | + | g.mu.Lock() | |
| 282 | + | g.initLocked() | |
| 283 | + | if g.phase == roleEffectsFrozen { | |
| 284 | + | changed := g.changed | |
| 285 | + | g.mu.Unlock() | |
| 286 | + | <-changed | |
| 287 | + | continue | |
| 288 | + | } | |
| 289 | + | g.mu.Unlock() | |
| 290 | + | // Publication may have completed between token capture and admission. | |
| 291 | + | // Re-read the registry and exact generation instead of using stale role | |
| 292 | + | // or lease authority. | |
| 293 | + | } | |
| 294 | + | } | |
| 295 | + | ||
| 263 | 296 | // publishRoleCapability is used by direct/headless setup and by transition | |
| 264 | 297 | // publication while the gate is frozen. Production role changes publish only | |
| 265 | 298 | // through publishFrozenRoleCapability. | |
| @@ -260,6 +260,39 @@ func (ac *attachedClient) beginRoleEffect(token attachmentRoleToken) (*roleEffec | |||||
| 260 | return &roleEffectTicket{gate: g, token: captured}, true | 260 | return &roleEffectTicket{gate: g, token: captured}, true | ||
| 261 | } | 261 | } | ||
| 262 | 262 | ||||
| 263 | + | // beginCurrentRoleEffect waits out a transition that already froze ac, then | |||
| 264 | + | // admits the role capability derived from the session registries. Handshakes | |||
| 265 | + | // use it after Welcome has completed so a replacement blocked behind that send | |||
| 266 | + | // can publish before readiness, first paint, or the snatched reset panel. | |||
| 267 | + | func (ac *attachedClient) beginCurrentRoleEffect(sess *session, tr ports.Transport) (attachmentRoleToken, *roleEffectTicket, bool) { | |||
| 268 | + | if ac == nil || sess == nil || tr == nil { | |||
| 269 | + | return attachmentRoleToken{}, nil, false | |||
| 270 | + | } | |||
| 271 | + | for { | |||
| 272 | + | token := sess.attachmentToken(ac, tr) | |||
| 273 | + | if token.role == attachmentDetached { | |||
| 274 | + | return token, nil, false | |||
| 275 | + | } | |||
| 276 | + | if ticket, admitted := ac.beginRoleEffect(token); admitted { | |||
| 277 | + | return token, ticket, true | |||
| 278 | + | } | |||
| 279 | + | ||||
| 280 | + | g := &ac.roleEffects | |||
| 281 | + | g.mu.Lock() | |||
| 282 | + | g.initLocked() | |||
| 283 | + | if g.phase == roleEffectsFrozen { | |||
| 284 | + | changed := g.changed | |||
| 285 | + | g.mu.Unlock() | |||
| 286 | + | <-changed | |||
| 287 | + | continue | |||
| 288 | + | } | |||
| 289 | + | g.mu.Unlock() | |||
| 290 | + | // Publication may have completed between token capture and admission. | |||
| 291 | + | // Re-read the registry and exact generation instead of using stale role | |||
| 292 | + | // or lease authority. | |||
| 293 | + | } | |||
| 294 | + | } | |||
| 295 | + | ||||
| 263 | // publishRoleCapability is used by direct/headless setup and by transition | 296 | // publishRoleCapability is used by direct/headless setup and by transition | ||
| 264 | // publication while the gate is frozen. Production role changes publish only | 297 | // publication while the gate is frozen. Production role changes publish only | ||
| 265 | // through publishFrozenRoleCapability. | 298 | // through publishFrozenRoleCapability. | ||
M internal/usecase/daemon/snatched_client.go +94 -8
| @@ -14,6 +14,58 @@ const ( | |||
| 14 | 14 | attachmentSnatched | |
| 15 | 15 | ) | |
| 16 | 16 | ||
| 17 | + | type initialSnatchedPanelAttempt struct { | |
| 18 | + | done chan struct{} | |
| 19 | + | err error | |
| 20 | + | } | |
| 21 | + | ||
| 22 | + | func (ac *attachedClient) claimInitialSnatchedPanel(generation uint64) (*initialSnatchedPanelAttempt, bool) { | |
| 23 | + | ac.initialSnatchedMu.Lock() | |
| 24 | + | defer ac.initialSnatchedMu.Unlock() | |
| 25 | + | if generation < ac.initialSnatchedGeneration { | |
| 26 | + | return nil, false | |
| 27 | + | } | |
| 28 | + | if generation == ac.initialSnatchedGeneration && ac.initialSnatchedAttempt != nil { | |
| 29 | + | return ac.initialSnatchedAttempt, false | |
| 30 | + | } | |
| 31 | + | attempt := &initialSnatchedPanelAttempt{done: make(chan struct{})} | |
| 32 | + | ac.initialSnatchedGeneration = generation | |
| 33 | + | ac.initialSnatchedAttempt = attempt | |
| 34 | + | return attempt, true | |
| 35 | + | } | |
| 36 | + | ||
| 37 | + | func (ac *attachedClient) initialSnatchedPanelClaimed(generation uint64) bool { | |
| 38 | + | ac.initialSnatchedMu.Lock() | |
| 39 | + | defer ac.initialSnatchedMu.Unlock() | |
| 40 | + | return ac.initialSnatchedGeneration == generation && ac.initialSnatchedAttempt != nil | |
| 41 | + | } | |
| 42 | + | ||
| 43 | + | func (ac *attachedClient) completeInitialSnatchedPanel(attempt *initialSnatchedPanelAttempt, err error) { | |
| 44 | + | ac.initialSnatchedMu.Lock() | |
| 45 | + | attempt.err = err | |
| 46 | + | close(attempt.done) | |
| 47 | + | ac.initialSnatchedMu.Unlock() | |
| 48 | + | } | |
| 49 | + | ||
| 50 | + | func (d *Daemon) sendInitialSnatchedPanel(token attachmentRoleToken, ticket *roleEffectTicket) error { | |
| 51 | + | attempt, owner := token.ac.claimInitialSnatchedPanel(token.generation) | |
| 52 | + | if attempt == nil { | |
| 53 | + | return errSnatchedOutputStale | |
| 54 | + | } | |
| 55 | + | if !owner { | |
| 56 | + | <-attempt.done | |
| 57 | + | return attempt.err | |
| 58 | + | } | |
| 59 | + | ||
| 60 | + | if !token.current() || !d.clearForSnatch(token) || !d.clearCaptureFramesForSnatch(token) { | |
| 61 | + | token.ac.completeInitialSnatchedPanel(attempt, errSnatchedOutputStale) | |
| 62 | + | return errSnatchedOutputStale | |
| 63 | + | } | |
| 64 | + | err := d.sendSnatchedPanel(token.ac, token.transport, token.generation, "", ticket) | |
| 65 | + | token.ac.completeInitialSnatchedPanel(attempt, err) | |
| 66 | + | return err | |
| 67 | + | } | |
| 68 | + | ||
| 17 | 69 | // attachmentRole derives the attachment's role from the session-owned | |
| 18 | 70 | // registries. There is deliberately no second role field that could drift. | |
| 19 | 71 | func (s *session) attachmentRole(ac *attachedClient) attachmentRole { | |
| @@ -483,6 +535,9 @@ func (d *Daemon) transitionAttachmentRoutedLocked(req attachmentTransitionReques | |||
| 483 | 535 | return attachmentTransitionResult{}, errAttachmentTransition | |
| 484 | 536 | } | |
| 485 | 537 | ||
| 538 | + | if req.targetRole == attachmentActive { | |
| 539 | + | d.demoteParkedActiveForSessionLocked(req.target) | |
| 540 | + | } | |
| 486 | 541 | if req.copySourceEnvironment { | |
| 487 | 542 | req.target.terminal = source.terminal | |
| 488 | 543 | req.target.env = copyEnvironment(source.env) | |
| @@ -618,6 +673,10 @@ func (d *Daemon) boundedSendSnatchedControl(token attachmentRoleToken, frame por | |||
| 618 | 673 | // transport, allowing callers that already interrupted a blocked send to avoid | |
| 619 | 674 | // closing the same link twice. | |
| 620 | 675 | func (d *Daemon) removeSnatchedAttachment(token attachmentRoleToken) bool { | |
| 676 | + | return d.unrouteSnatchedAttachment(token, true) | |
| 677 | + | } | |
| 678 | + | ||
| 679 | + | func (d *Daemon) unrouteSnatchedAttachment(token attachmentRoleToken, terminal bool) bool { | |
| 621 | 680 | if token.sess == nil || token.ac == nil || token.transport.transport == nil { | |
| 622 | 681 | return false | |
| 623 | 682 | } | |
| @@ -641,12 +700,36 @@ func (d *Daemon) removeSnatchedAttachment(token attachmentRoleToken) bool { | |||
| 641 | 700 | ||
| 642 | 701 | token.ac.clearSnatchedInput() | |
| 643 | 702 | d.unregisterPreview(token.ac) | |
| 644 | - | token.ac.clearPreviousSession() | |
| 703 | + | if terminal { | |
| 704 | + | token.ac.clearPreviousSession() | |
| 705 | + | } | |
| 645 | 706 | token.ac.setSession(nil) | |
| 646 | 707 | token.ac.clearCaptureFrames() | |
| 647 | 708 | return true | |
| 648 | 709 | } | |
| 649 | 710 | ||
| 711 | + | // parkSnatchedAttachment removes only the exact lost snatched incarnation and | |
| 712 | + | // retains its resume identity for a role-preserving reconnect. | |
| 713 | + | func (d *Daemon) parkSnatchedAttachment(token attachmentRoleToken) bool { | |
| 714 | + | if token.ac == nil || !token.ac.resumeCapable || !d.unrouteSnatchedAttachment(token, false) { | |
| 715 | + | return false | |
| 716 | + | } | |
| 717 | + | if d.parkAttachmentAs(token.sess, token.ac, attachmentSnatched) { | |
| 718 | + | _ = token.ac.closeCapturedTransport(token.ac.revokeTransport(token.transport.transport)) | |
| 719 | + | return true | |
| 720 | + | } | |
| 721 | + | token.ac.clearPreviousSession() | |
| 722 | + | _ = token.ac.closeCapturedTransport(token.ac.revokeTransport(token.transport.transport)) | |
| 723 | + | return false | |
| 724 | + | } | |
| 725 | + | ||
| 726 | + | func (d *Daemon) parkOrDropSnatchedAttachment(token attachmentRoleToken) bool { | |
| 727 | + | if d.parkSnatchedAttachment(token) { | |
| 728 | + | return true | |
| 729 | + | } | |
| 730 | + | return d.dropSnatchedAttachment(token) | |
| 731 | + | } | |
| 732 | + | ||
| 650 | 733 | // cleanupInterruptedSnatchedAttachment owns terminal cleanup for a displaced | |
| 651 | 734 | // transport that was closed to drain an admitted send. It waits on the gate's | |
| 652 | 735 | // condition rather than ordinary effect admission, then removes only the exact | |
| @@ -690,11 +773,17 @@ func (d *Daemon) cleanupInterruptedSnatchedAttachment(token attachmentRoleToken) | |||
| 690 | 773 | return false | |
| 691 | 774 | } | |
| 692 | 775 | ||
| 776 | + | token.ac.setSession(nil) | |
| 777 | + | if token.ac.resumeCapable && d.parkAttachmentAs(token.sess, token.ac, attachmentSnatched) { | |
| 778 | + | _ = token.ac.closeCapturedTransport(token.ac.revokeTransport(token.transport.transport)) | |
| 779 | + | token.ac.clearSnatchedInput() | |
| 780 | + | d.unregisterPreview(token.ac) | |
| 781 | + | return true | |
| 782 | + | } | |
| 693 | 783 | _ = token.ac.closeCapturedTransport(token.ac.revokeTransport(token.transport.transport)) | |
| 694 | 784 | token.ac.clearSnatchedInput() | |
| 695 | 785 | d.unregisterPreview(token.ac) | |
| 696 | 786 | token.ac.clearPreviousSession() | |
| 697 | - | token.ac.setSession(nil) | |
| 698 | 787 | token.ac.clearCaptureFrames() | |
| 699 | 788 | return true | |
| 700 | 789 | } | |
| @@ -776,7 +865,7 @@ func lockAttachmentCoordinators(a *session, aCoordinator *renderCoordinator, b * | |||
| 776 | 865 | func (d *Daemon) deferAttachmentTransitionCleanups(result attachmentTransitionResult) { | |
| 777 | 866 | if token := result.displaced; token.ac != nil && token.transport.transport != nil { | |
| 778 | 867 | blockedRender := result.displacedInterrupted | |
| 779 | - | if !blockedRender { | |
| 868 | + | if !blockedRender && !token.ac.initialSnatchedPanelClaimed(token.generation) { | |
| 780 | 869 | blockedRender = !token.ac.sendMu.TryLock() | |
| 781 | 870 | if !blockedRender { | |
| 782 | 871 | token.ac.sendMu.Unlock() | |
| @@ -798,12 +887,9 @@ func (d *Daemon) deferAttachmentTransitionCleanups(result attachmentTransitionRe | |||
| 798 | 887 | return | |
| 799 | 888 | } | |
| 800 | 889 | defer ticket.End() | |
| 801 | - | if !token.current() || !d.clearForSnatch(token) || !d.clearCaptureFramesForSnatch(token) { | |
| 802 | - | return | |
| 803 | - | } | |
| 804 | - | if err := d.sendSnatchedPanel(token.ac, token.transport, token.generation, "", ticket); err != nil { | |
| 890 | + | if err := d.sendInitialSnatchedPanel(token, ticket); err != nil { | |
| 805 | 891 | ticket.End() | |
| 806 | - | d.dropSnatchedAttachment(token) | |
| 892 | + | d.parkOrDropSnatchedAttachment(token) | |
| 807 | 893 | } | |
| 808 | 894 | }) | |
| 809 | 895 | } | |
| @@ -14,6 +14,58 @@ const ( | |||||
| 14 | attachmentSnatched | 14 | attachmentSnatched | ||
| 15 | ) | 15 | ) | ||
| 16 | 16 | ||||
| 17 | + | type initialSnatchedPanelAttempt struct { | |||
| 18 | + | done chan struct{} | |||
| 19 | + | err error | |||
| 20 | + | } | |||
| 21 | + | ||||
| 22 | + | func (ac *attachedClient) claimInitialSnatchedPanel(generation uint64) (*initialSnatchedPanelAttempt, bool) { | |||
| 23 | + | ac.initialSnatchedMu.Lock() | |||
| 24 | + | defer ac.initialSnatchedMu.Unlock() | |||
| 25 | + | if generation < ac.initialSnatchedGeneration { | |||
| 26 | + | return nil, false | |||
| 27 | + | } | |||
| 28 | + | if generation == ac.initialSnatchedGeneration && ac.initialSnatchedAttempt != nil { | |||
| 29 | + | return ac.initialSnatchedAttempt, false | |||
| 30 | + | } | |||
| 31 | + | attempt := &initialSnatchedPanelAttempt{done: make(chan struct{})} | |||
| 32 | + | ac.initialSnatchedGeneration = generation | |||
| 33 | + | ac.initialSnatchedAttempt = attempt | |||
| 34 | + | return attempt, true | |||
| 35 | + | } | |||
| 36 | + | ||||
| 37 | + | func (ac *attachedClient) initialSnatchedPanelClaimed(generation uint64) bool { | |||
| 38 | + | ac.initialSnatchedMu.Lock() | |||
| 39 | + | defer ac.initialSnatchedMu.Unlock() | |||
| 40 | + | return ac.initialSnatchedGeneration == generation && ac.initialSnatchedAttempt != nil | |||
| 41 | + | } | |||
| 42 | + | ||||
| 43 | + | func (ac *attachedClient) completeInitialSnatchedPanel(attempt *initialSnatchedPanelAttempt, err error) { | |||
| 44 | + | ac.initialSnatchedMu.Lock() | |||
| 45 | + | attempt.err = err | |||
| 46 | + | close(attempt.done) | |||
| 47 | + | ac.initialSnatchedMu.Unlock() | |||
| 48 | + | } | |||
| 49 | + | ||||
| 50 | + | func (d *Daemon) sendInitialSnatchedPanel(token attachmentRoleToken, ticket *roleEffectTicket) error { | |||
| 51 | + | attempt, owner := token.ac.claimInitialSnatchedPanel(token.generation) | |||
| 52 | + | if attempt == nil { | |||
| 53 | + | return errSnatchedOutputStale | |||
| 54 | + | } | |||
| 55 | + | if !owner { | |||
| 56 | + | <-attempt.done | |||
| 57 | + | return attempt.err | |||
| 58 | + | } | |||
| 59 | + | ||||
| 60 | + | if !token.current() || !d.clearForSnatch(token) || !d.clearCaptureFramesForSnatch(token) { | |||
| 61 | + | token.ac.completeInitialSnatchedPanel(attempt, errSnatchedOutputStale) | |||
| 62 | + | return errSnatchedOutputStale | |||
| 63 | + | } | |||
| 64 | + | err := d.sendSnatchedPanel(token.ac, token.transport, token.generation, "", ticket) | |||
| 65 | + | token.ac.completeInitialSnatchedPanel(attempt, err) | |||
| 66 | + | return err | |||
| 67 | + | } | |||
| 68 | + | ||||
| 17 | // attachmentRole derives the attachment's role from the session-owned | 69 | // attachmentRole derives the attachment's role from the session-owned | ||
| 18 | // registries. There is deliberately no second role field that could drift. | 70 | // registries. There is deliberately no second role field that could drift. | ||
| 19 | func (s *session) attachmentRole(ac *attachedClient) attachmentRole { | 71 | func (s *session) attachmentRole(ac *attachedClient) attachmentRole { | ||
| @@ -483,6 +535,9 @@ func (d *Daemon) transitionAttachmentRoutedLocked(req attachmentTransitionReques | |||||
| 483 | return attachmentTransitionResult{}, errAttachmentTransition | 535 | return attachmentTransitionResult{}, errAttachmentTransition | ||
| 484 | } | 536 | } | ||
| 485 | 537 | ||||
| 538 | + | if req.targetRole == attachmentActive { | |||
| 539 | + | d.demoteParkedActiveForSessionLocked(req.target) | |||
| 540 | + | } | |||
| 486 | if req.copySourceEnvironment { | 541 | if req.copySourceEnvironment { | ||
| 487 | req.target.terminal = source.terminal | 542 | req.target.terminal = source.terminal | ||
| 488 | req.target.env = copyEnvironment(source.env) | 543 | req.target.env = copyEnvironment(source.env) | ||
| @@ -618,6 +673,10 @@ func (d *Daemon) boundedSendSnatchedControl(token attachmentRoleToken, frame por | |||||
| 618 | // transport, allowing callers that already interrupted a blocked send to avoid | 673 | // transport, allowing callers that already interrupted a blocked send to avoid | ||
| 619 | // closing the same link twice. | 674 | // closing the same link twice. | ||
| 620 | func (d *Daemon) removeSnatchedAttachment(token attachmentRoleToken) bool { | 675 | func (d *Daemon) removeSnatchedAttachment(token attachmentRoleToken) bool { | ||
| 676 | + | return d.unrouteSnatchedAttachment(token, true) | |||
| 677 | + | } | |||
| 678 | + | ||||
| 679 | + | func (d *Daemon) unrouteSnatchedAttachment(token attachmentRoleToken, terminal bool) bool { | |||
| 621 | if token.sess == nil || token.ac == nil || token.transport.transport == nil { | 680 | if token.sess == nil || token.ac == nil || token.transport.transport == nil { | ||
| 622 | return false | 681 | return false | ||
| 623 | } | 682 | } | ||
| @@ -641,12 +700,36 @@ func (d *Daemon) removeSnatchedAttachment(token attachmentRoleToken) bool { | |||||
| 641 | 700 | ||||
| 642 | token.ac.clearSnatchedInput() | 701 | token.ac.clearSnatchedInput() | ||
| 643 | d.unregisterPreview(token.ac) | 702 | d.unregisterPreview(token.ac) | ||
| 644 | - | token.ac.clearPreviousSession() | 703 | + | if terminal { |
| 704 | + | token.ac.clearPreviousSession() | |||
| 705 | + | } | |||
| 645 | token.ac.setSession(nil) | 706 | token.ac.setSession(nil) | ||
| 646 | token.ac.clearCaptureFrames() | 707 | token.ac.clearCaptureFrames() | ||
| 647 | return true | 708 | return true | ||
| 648 | } | 709 | } | ||
| 649 | 710 | ||||
| 711 | + | // parkSnatchedAttachment removes only the exact lost snatched incarnation and | |||
| 712 | + | // retains its resume identity for a role-preserving reconnect. | |||
| 713 | + | func (d *Daemon) parkSnatchedAttachment(token attachmentRoleToken) bool { | |||
| 714 | + | if token.ac == nil || !token.ac.resumeCapable || !d.unrouteSnatchedAttachment(token, false) { | |||
| 715 | + | return false | |||
| 716 | + | } | |||
| 717 | + | if d.parkAttachmentAs(token.sess, token.ac, attachmentSnatched) { | |||
| 718 | + | _ = token.ac.closeCapturedTransport(token.ac.revokeTransport(token.transport.transport)) | |||
| 719 | + | return true | |||
| 720 | + | } | |||
| 721 | + | token.ac.clearPreviousSession() | |||
| 722 | + | _ = token.ac.closeCapturedTransport(token.ac.revokeTransport(token.transport.transport)) | |||
| 723 | + | return false | |||
| 724 | + | } | |||
| 725 | + | ||||
| 726 | + | func (d *Daemon) parkOrDropSnatchedAttachment(token attachmentRoleToken) bool { | |||
| 727 | + | if d.parkSnatchedAttachment(token) { | |||
| 728 | + | return true | |||
| 729 | + | } | |||
| 730 | + | return d.dropSnatchedAttachment(token) | |||
| 731 | + | } | |||
| 732 | + | ||||
| 650 | // cleanupInterruptedSnatchedAttachment owns terminal cleanup for a displaced | 733 | // cleanupInterruptedSnatchedAttachment owns terminal cleanup for a displaced | ||
| 651 | // transport that was closed to drain an admitted send. It waits on the gate's | 734 | // transport that was closed to drain an admitted send. It waits on the gate's | ||
| 652 | // condition rather than ordinary effect admission, then removes only the exact | 735 | // condition rather than ordinary effect admission, then removes only the exact | ||
| @@ -690,11 +773,17 @@ func (d *Daemon) cleanupInterruptedSnatchedAttachment(token attachmentRoleToken) | |||||
| 690 | return false | 773 | return false | ||
| 691 | } | 774 | } | ||
| 692 | 775 | ||||
| 776 | + | token.ac.setSession(nil) | |||
| 777 | + | if token.ac.resumeCapable && d.parkAttachmentAs(token.sess, token.ac, attachmentSnatched) { | |||
| 778 | + | _ = token.ac.closeCapturedTransport(token.ac.revokeTransport(token.transport.transport)) | |||
| 779 | + | token.ac.clearSnatchedInput() | |||
| 780 | + | d.unregisterPreview(token.ac) | |||
| 781 | + | return true | |||
| 782 | + | } | |||
| 693 | _ = token.ac.closeCapturedTransport(token.ac.revokeTransport(token.transport.transport)) | 783 | _ = token.ac.closeCapturedTransport(token.ac.revokeTransport(token.transport.transport)) | ||
| 694 | token.ac.clearSnatchedInput() | 784 | token.ac.clearSnatchedInput() | ||
| 695 | d.unregisterPreview(token.ac) | 785 | d.unregisterPreview(token.ac) | ||
| 696 | token.ac.clearPreviousSession() | 786 | token.ac.clearPreviousSession() | ||
| 697 | - | token.ac.setSession(nil) | |||
| 698 | token.ac.clearCaptureFrames() | 787 | token.ac.clearCaptureFrames() | ||
| 699 | return true | 788 | return true | ||
| 700 | } | 789 | } | ||
| @@ -776,7 +865,7 @@ func lockAttachmentCoordinators(a *session, aCoordinator *renderCoordinator, b * | |||||
| 776 | func (d *Daemon) deferAttachmentTransitionCleanups(result attachmentTransitionResult) { | 865 | func (d *Daemon) deferAttachmentTransitionCleanups(result attachmentTransitionResult) { | ||
| 777 | if token := result.displaced; token.ac != nil && token.transport.transport != nil { | 866 | if token := result.displaced; token.ac != nil && token.transport.transport != nil { | ||
| 778 | blockedRender := result.displacedInterrupted | 867 | blockedRender := result.displacedInterrupted | ||
| 779 | - | if !blockedRender { | 868 | + | if !blockedRender && !token.ac.initialSnatchedPanelClaimed(token.generation) { |
| 780 | blockedRender = !token.ac.sendMu.TryLock() | 869 | blockedRender = !token.ac.sendMu.TryLock() | ||
| 781 | if !blockedRender { | 870 | if !blockedRender { | ||
| 782 | token.ac.sendMu.Unlock() | 871 | token.ac.sendMu.Unlock() | ||
| @@ -798,12 +887,9 @@ func (d *Daemon) deferAttachmentTransitionCleanups(result attachmentTransitionRe | |||||
| 798 | return | 887 | return | ||
| 799 | } | 888 | } | ||
| 800 | defer ticket.End() | 889 | defer ticket.End() | ||
| 801 | - | if !token.current() || !d.clearForSnatch(token) || !d.clearCaptureFramesForSnatch(token) { | 890 | + | if err := d.sendInitialSnatchedPanel(token, ticket); err != nil { |
| 802 | - | return | |||
| 803 | - | } | |||
| 804 | - | if err := d.sendSnatchedPanel(token.ac, token.transport, token.generation, "", ticket); err != nil { | |||
| 805 | ticket.End() | 891 | ticket.End() | ||
| 806 | - | d.dropSnatchedAttachment(token) | 892 | + | d.parkOrDropSnatchedAttachment(token) |
| 807 | } | 893 | } | ||
| 808 | }) | 894 | }) | ||
| 809 | } | 895 | } | ||
M internal/usecase/daemon/snatched_input.go +2 -2
| @@ -213,7 +213,7 @@ func (d *Daemon) reclaimSnatchedAttachment(token attachmentRoleToken) bool { | |||
| 213 | 213 | }) | |
| 214 | 214 | if err != nil { | |
| 215 | 215 | if errors.Is(err, errSendTimedOut) { | |
| 216 | - | d.attachmentCleanupWg.Go(func() { d.dropSnatchedAttachment(token) }) | |
| 216 | + | d.attachmentCleanupWg.Go(func() { d.parkOrDropSnatchedAttachment(token) }) | |
| 217 | 217 | } else { | |
| 218 | 218 | d.showSnatchedUnavailable(token) | |
| 219 | 219 | } | |
| @@ -244,7 +244,7 @@ func (d *Daemon) showSnatchedUnavailable(token attachmentRoleToken) { | |||
| 244 | 244 | defer fresh.endRoleEffect() | |
| 245 | 245 | if err := d.sendSnatchedPanel(token.ac, token.transport, token.generation, "Session is no longer available.", ticket); err != nil { | |
| 246 | 246 | fresh.endRoleEffect() | |
| 247 | - | d.dropSnatchedAttachment(token) | |
| 247 | + | d.parkOrDropSnatchedAttachment(token) | |
| 248 | 248 | } | |
| 249 | 249 | } | |
| 250 | 250 | ||
| @@ -213,7 +213,7 @@ func (d *Daemon) reclaimSnatchedAttachment(token attachmentRoleToken) bool { | |||||
| 213 | }) | 213 | }) | ||
| 214 | if err != nil { | 214 | if err != nil { | ||
| 215 | if errors.Is(err, errSendTimedOut) { | 215 | if errors.Is(err, errSendTimedOut) { | ||
| 216 | - | d.attachmentCleanupWg.Go(func() { d.dropSnatchedAttachment(token) }) | 216 | + | d.attachmentCleanupWg.Go(func() { d.parkOrDropSnatchedAttachment(token) }) |
| 217 | } else { | 217 | } else { | ||
| 218 | d.showSnatchedUnavailable(token) | 218 | d.showSnatchedUnavailable(token) | ||
| 219 | } | 219 | } | ||
| @@ -244,7 +244,7 @@ func (d *Daemon) showSnatchedUnavailable(token attachmentRoleToken) { | |||||
| 244 | defer fresh.endRoleEffect() | 244 | defer fresh.endRoleEffect() | ||
| 245 | if err := d.sendSnatchedPanel(token.ac, token.transport, token.generation, "Session is no longer available.", ticket); err != nil { | 245 | if err := d.sendSnatchedPanel(token.ac, token.transport, token.generation, "Session is no longer available.", ticket); err != nil { | ||
| 246 | fresh.endRoleEffect() | 246 | fresh.endRoleEffect() | ||
| 247 | - | d.dropSnatchedAttachment(token) | 247 | + | d.parkOrDropSnatchedAttachment(token) |
| 248 | } | 248 | } | ||
| 249 | } | 249 | } | ||
| 250 | 250 | ||||