1package daemon
2
3import (
4 "errors"
5 "runtime"
6
7 "github.com/bnema/vev/internal/domain"
8 "github.com/bnema/vev/internal/ports"
9 "github.com/bnema/vev/internal/usecase/command"
10)
11
12const connectionSnapshotAttempts = 8
13
14// currentAttachmentConnection retries the lock-free session-to-role snapshot when a
15// handoff lands between those reads. The attachment gate still performs the final
16// mutation admission after this function returns.
17func (d *Daemon) currentAttachmentConnection(ac *attachedClient, tr ports.Transport) (*session, attachmentConnectionToken, bool) {
18 for range connectionSnapshotAttempts {
19 if !ac.currentTransportIs(tr) {
20 break
21 }
22 sess := ac.currentAttachmentSession()
23 if sess == nil {
24 return nil, attachmentConnectionToken{}, false
25 }
26 if d.afterConnectionSessionSnapshot != nil {
27 d.afterConnectionSessionSnapshot(sess)
28 }
29 token := attachmentToken(sess, ac, tr)
30 if token.ac == nil || ac.currentAttachmentSession() != sess {
31 runtime.Gosched()
32 continue
33 }
34 return sess, token, true
35 }
36 return nil, attachmentConnectionToken{}, false
37}
38
39// runConnLoop is the per-connection input router: it pumps client messages
40// until detach, EOF, or a transport error. The attachment generation is
41// resolved after every Recv so a stale transport cannot execute a new frame.
42func (d *Daemon) runConnLoop(ac *attachedClient) {
43 tr := ac.transport()
44 if tr == nil {
45 return
46 }
47 for {
48 if !ac.currentTransportIs(tr) || ac.currentAttachmentSession() == nil {
49 return
50 }
51 f, err := tr.Recv()
52 if err != nil {
53 for range connectionSnapshotAttempts {
54 sess, _, ok := d.currentAttachmentConnection(ac, tr)
55 if !ok {
56 return
57 }
58 d.clientGone(sess, ac, tr, false)
59 current := ac.currentAttachmentSession()
60 if current == sess || !ac.currentTransportIs(tr) || current == nil {
61 return
62 }
63 }
64 return
65 }
66 _, token, ok := d.currentAttachmentConnection(ac, tr)
67 if !ok {
68 return
69 }
70 if d.handleAttachmentClientFrame(token, f) {
71 return
72 }
73 }
74}
75
76// handleAttachmentClientFrame owns the attached-client protocol.
77func (d *Daemon) handleAttachmentClientFrame(token attachmentConnectionToken, f ports.Frame) bool {
78 if !token.attachmentEffectCurrent() {
79 return false
80 }
81 if d.afterAttachmentFrameDispatch != nil {
82 d.afterAttachmentFrameDispatch(token)
83 }
84 ticket, admitted := token.ac.beginAttachmentEffect(token)
85 if !admitted {
86 return false
87 }
88 token.effect = ticket
89 defer token.endAttachmentEffect()
90 if d.afterAttachmentEffectAdmitted != nil {
91 d.afterAttachmentEffectAdmitted(token)
92 }
93 switch f.Type {
94 case ports.MsgInput:
95 if in, derr := ports.UnmarshalInput(f.Payload); derr == nil {
96 d.handleSequencedInputForAttachment(token, in.InputSeq, in.Data)
97 }
98 case ports.MsgResize:
99 if rz, derr := ports.UnmarshalResize(f.Payload); derr == nil && token.attachmentEffectCurrent() {
100 d.resizeAttachmentForLease(token, rz.Size)
101 }
102 case ports.MsgTheme:
103 if th, derr := ports.UnmarshalTheme(f.Payload); derr == nil {
104 d.applyThemeForAttachment(token, th)
105 }
106 case ports.MsgImagePush:
107 if ip, derr := ports.UnmarshalImagePush(f.Payload); derr == nil {
108 d.handleSequencedImagePushForAttachment(token, ip.InputSeq, ip)
109 }
110 case ports.MsgClientNotice:
111 if notice, derr := ports.UnmarshalClientNotice(f.Payload); derr == nil {
112 d.handleClientNoticeForAttachment(token, notice)
113 } else {
114 d.log.Warn("malformed client notice", "err", derr)
115 }
116 case ports.MsgDetach:
117 if token.attachmentEffectCurrent() {
118 d.clientGoneForAttachment(token, true)
119 return true
120 }
121 case ports.MsgAck:
122 if ack, derr := ports.UnmarshalAck(f.Payload); derr == nil {
123 d.ackOutput(token, ack.Epoch, ack.State)
124 }
125 case ports.MsgOutputResetRequest:
126 if _, derr := ports.UnmarshalOutputResetRequest(f.Payload); derr == nil {
127 d.resetOutput(token)
128 }
129 case ports.MsgRecentRouteSnapshot:
130 snapshot, derr := ports.UnmarshalRecentRouteSnapshot(f.Payload)
131 if derr != nil {
132 d.log.Warn("malformed recent route snapshot", "err", derr)
133 break
134 }
135 replayIdentity := token.ac.setRouteSnapshot(snapshot)
136 d.invalidateRender(token.sess, token.ac, false, "client_frame_routing.go:route-snapshot")
137 if !replayIdentity {
138 break
139 }
140 if err := d.sendCommittedRouteIdentityForAttachment(token); err != nil {
141 d.detachOnAttachmentSendError(token, token.transport.transport)
142 }
143 case ports.MsgRouteAttentionSubscription:
144 subscription, derr := ports.UnmarshalRouteAttentionSubscription(f.Payload)
145 if derr != nil {
146 d.log.Warn("malformed route attention subscription", "err", derr)
147 break
148 }
149 token.ac.setRouteAttentionSubscription(subscription)
150 d.invalidateRender(token.sess, token.ac, false, "client_frame_routing.go:route-attention")
151 case ports.MsgRouteNavigationFailure:
152 failure, derr := ports.UnmarshalRouteNavigationFailure(f.Payload)
153 if derr != nil {
154 d.log.Warn("malformed route navigation failure", "err", derr)
155 break
156 }
157 message := "route navigation failed"
158 switch failure.Code {
159 case ports.RouteFailureStaleSelection:
160 message = "that recent route is no longer available"
161 case ports.RouteFailureTargetChanged:
162 message = "that recent route changed before attach"
163 case ports.RouteFailureOriginUnavailable:
164 message = "the original route is no longer available"
165 case ports.RouteFailureUnavailable:
166 message = "the selected route is unavailable"
167 case ports.RouteFailureNoSuchRoute:
168 message = "that recent route no longer exists"
169 }
170 d.notify(token.sess, domain.NoticeWarn, domain.NoticeSessionUnavailable, message, nil)
171 case ports.MsgCommand:
172 request, derr := ports.UnmarshalCommandRequest(f.Payload)
173 if derr != nil {
174 return false
175 }
176 result := d.executeAttachedCommand(token, request)
177 resultToken, ok := d.commandResultToken(token)
178 if !ok {
179 // A remote handoff or explicit detach deliberately retired the old
180 // attachment; it must not be treated as a stale-result send error.
181 return true
182 }
183 if err := resultToken.sendControl(frameCommandResult(result)); err != nil {
184 resultToken.endAttachmentEffect()
185 d.detachOnAttachmentSendError(resultToken, resultToken.transport.transport)
186 return true
187 }
188 resultToken.endAttachmentEffect()
189 case ports.MsgPing:
190 if err := token.sendControl(framePong()); err != nil {
191 token.endAttachmentEffect()
192 d.detachOnAttachmentSendError(token, token.transport.transport)
193 return true
194 }
195 default:
196 // Unknown/out-of-band client messages are ignored so a newer
197 // client can add message types without breaking an older daemon.
198 }
199 return false
200}
201
202// commandResultToken renews the attachment effect after a local navigation
203// transition ends the command's initiating effect. A retired transport/session
204// means the command deliberately handed off or detached and has no result to
205// send on the old carriage.
206func (d *Daemon) commandResultToken(token attachmentConnectionToken) (attachmentConnectionToken, bool) {
207 if token.ac == nil || !token.ac.currentTransportIs(token.transport.transport) {
208 return attachmentConnectionToken{}, false
209 }
210 sess := token.ac.currentAttachmentSession()
211 if sess == nil {
212 return attachmentConnectionToken{}, false
213 }
214 resultToken := attachmentToken(sess, token.ac, token.transport.transport)
215 if resultToken.ac == nil {
216 return attachmentConnectionToken{}, false
217 }
218 if token.effect != nil && !token.effect.ended.Load() && token.current() {
219 return token, true
220 }
221 ticket, admitted := token.ac.beginAttachmentEffect(resultToken)
222 if !admitted {
223 return attachmentConnectionToken{}, false
224 }
225 resultToken.effect = ticket
226 return resultToken, true
227}
228
229func (d *Daemon) executeAttachedCommand(token attachmentConnectionToken, request ports.CommandRequest) ports.CommandResult {
230 result := ports.CommandResult{RequestID: request.RequestID}
231 if request.Version != ports.ProtocolVersion {
232 result.Code = ports.ErrInvalidCommandArgs
233 result.Text = "unsupported command protocol version"
234 return result
235 }
236 if !request.Attached {
237 result.Code = ports.ErrInvalidCommandArgs
238 result.Text = "attached command flag is required"
239 return result
240 }
241 if request.RequestID == 0 {
242 result.Code = ports.ErrInvalidCommandArgs
243 result.Text = "command request id is required"
244 return result
245 }
246 if request.Self {
247 result.Code = ports.ErrInvalidCommandArgs
248 result.Text = "attached commands cannot target themselves"
249 return result
250 }
251 if request.TargetSession != "" || request.TargetTab != "" || request.TargetPane != "" {
252 result.Code = ports.ErrInvalidCommandArgs
253 result.Text = "attached commands cannot override their active session target"
254 return result
255 }
256 sess := token.sess
257 if sess == nil || !token.attachmentEffectCurrent() {
258 result.Code = ports.ErrNoSuchTarget
259 result.Text = "attached session is no longer active"
260 return result
261 }
262 cmd, ok := command.BySlug(request.Slug)
263 if !ok {
264 result.Code = ports.ErrUnknownCommand
265 result.Text = "unknown command: " + request.Slug
266 return result
267 }
268 if !cmd.PaletteVisible {
269 result.Code = ports.ErrNotScriptable
270 result.Text = request.Slug + " is not available in the palette"
271 return result
272 }
273 if cmd.Run == nil {
274 result.Code = ports.ErrNotScriptable
275 result.Text = request.Slug + " is not scriptable"
276 return result
277 }
278 // The frame's attachment token is the target capability. No registry/name lookup is
279 // performed, so the command cannot escape to another remote session.
280 err := sess.runMutation(func() error {
281 return cmd.Run(paletteExec{d: d, sess: sess, ac: token.ac, effect: token.effect}, request.Args)
282 })
283 if err == nil {
284 result.OK = true
285 return result
286 }
287 if errors.Is(err, command.ErrInvalidArguments) {
288 result.Code = ports.ErrInvalidCommandArgs
289 result.Text = "usage: " + cmd.Usage
290 return result
291 }
292 result.Code = ports.ErrInternal
293 result.Text = err.Error()
294 return result
295}
296
297// resetOutput rebases only the exact attachment and transport admitted
298// by the frame router. sendMu serializes the rebase with every output send;
299// transport revalidation under that lock rejects a link replaced while waiting.
300func (d *Daemon) resetOutput(token attachmentConnectionToken) bool {
301 ac := token.ac
302 if ac == nil || token.effect == nil || token.effect.ended.Load() {
303 return false
304 }
305 ac.sendMu.Lock()
306 if token.effect.ended.Load() || !token.attachmentCurrent() || ac.output == nil {
307 ac.sendMu.Unlock()
308 return false
309 }
310 ac.rebaseOutput()
311 ac.pipelineCache = composeCacheInput{}
312 ac.pipelineScratch = composeCacheInput{}
313 ac.sendMu.Unlock()
314 go d.paint(token.sess, ac, true, token.lease)
315 return true
316}
317
318func (d *Daemon) ackOutput(token attachmentConnectionToken, epoch, state uint64) bool {
319 ac := token.ac
320 if token.effect == nil || token.effect.ended.Load() {
321 return false
322 }
323 ac.sendMu.Lock()
324 if ac.output == nil {
325 ac.sendMu.Unlock()
326 return false
327 }
328 acknowledged := ac.output.ack(epoch, state)
329 ac.sendMu.Unlock()
330 if acknowledged {
331 if rc := token.sess.renderCoordinator(); rc != nil {
332 rc.notifyAckForLease(token.lease)
333 }
334 }
335 return acknowledged
336}