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 token.ac.setRouteSnapshot(snapshot)
133 } else {
134 d.log.Warn("malformed recent route snapshot", "err", derr)
135 }
136 case ports.MsgRouteNavigationFailure:
137 failure, derr := ports.UnmarshalRouteNavigationFailure(f.Payload)
138 if derr != nil {
139 d.log.Warn("malformed route navigation failure", "err", derr)
140 break
141 }
142 message := "route navigation failed"
143 switch failure.Code {
144 case ports.RouteFailureStaleSelection:
145 message = "that recent route is no longer available"
146 case ports.RouteFailureTargetChanged:
147 message = "that recent route changed before attach"
148 case ports.RouteFailureOriginUnavailable:
149 message = "the original route is no longer available"
150 case ports.RouteFailureUnavailable:
151 message = "the selected route is unavailable"
152 case ports.RouteFailureNoSuchRoute:
153 message = "that recent route no longer exists"
154 }
155 d.notify(token.sess, domain.NoticeWarn, domain.NoticeSessionUnavailable, message, nil)
156 case ports.MsgCommand:
157 request, derr := ports.UnmarshalCommandRequest(f.Payload)
158 if derr != nil {
159 return false
160 }
161 result := d.executeAttachedCommand(token, request)
162 resultToken, ok := d.commandResultToken(token)
163 if !ok {
164 // A remote handoff or explicit detach deliberately retired the old
165 // attachment; it must not be treated as a stale-result send error.
166 return true
167 }
168 if err := resultToken.sendControl(frameCommandResult(result)); err != nil {
169 resultToken.endAttachmentEffect()
170 d.detachOnAttachmentSendError(resultToken, resultToken.transport.transport)
171 return true
172 }
173 resultToken.endAttachmentEffect()
174 case ports.MsgPing:
175 if err := token.sendControl(framePong()); err != nil {
176 token.endAttachmentEffect()
177 d.detachOnAttachmentSendError(token, token.transport.transport)
178 return true
179 }
180 default:
181 // Unknown/out-of-band client messages are ignored so a newer
182 // client can add message types without breaking an older daemon.
183 }
184 return false
185}
186
187// commandResultToken renews the attachment effect after a local navigation
188// transition ends the command's initiating effect. A retired transport/session
189// means the command deliberately handed off or detached and has no result to
190// send on the old carriage.
191func (d *Daemon) commandResultToken(token attachmentConnectionToken) (attachmentConnectionToken, bool) {
192 if token.ac == nil || !token.ac.currentTransportIs(token.transport.transport) {
193 return attachmentConnectionToken{}, false
194 }
195 sess := token.ac.currentAttachmentSession()
196 if sess == nil {
197 return attachmentConnectionToken{}, false
198 }
199 resultToken := attachmentToken(sess, token.ac, token.transport.transport)
200 if resultToken.ac == nil {
201 return attachmentConnectionToken{}, false
202 }
203 if token.effect != nil && !token.effect.ended.Load() && token.current() {
204 return token, true
205 }
206 ticket, admitted := token.ac.beginAttachmentEffect(resultToken)
207 if !admitted {
208 return attachmentConnectionToken{}, false
209 }
210 resultToken.effect = ticket
211 return resultToken, true
212}
213
214func (d *Daemon) executeAttachedCommand(token attachmentConnectionToken, request ports.CommandRequest) ports.CommandResult {
215 result := ports.CommandResult{RequestID: request.RequestID}
216 if request.Version != ports.ProtocolVersion {
217 result.Code = ports.ErrInvalidCommandArgs
218 result.Text = "unsupported command protocol version"
219 return result
220 }
221 if !request.Attached {
222 result.Code = ports.ErrInvalidCommandArgs
223 result.Text = "attached command flag is required"
224 return result
225 }
226 if request.RequestID == 0 {
227 result.Code = ports.ErrInvalidCommandArgs
228 result.Text = "command request id is required"
229 return result
230 }
231 if request.Self {
232 result.Code = ports.ErrInvalidCommandArgs
233 result.Text = "attached commands cannot target themselves"
234 return result
235 }
236 if request.TargetSession != "" || request.TargetTab != "" || request.TargetPane != "" {
237 result.Code = ports.ErrInvalidCommandArgs
238 result.Text = "attached commands cannot override their active session target"
239 return result
240 }
241 sess := token.sess
242 if sess == nil || !token.attachmentEffectCurrent() {
243 result.Code = ports.ErrNoSuchTarget
244 result.Text = "attached session is no longer active"
245 return result
246 }
247 cmd, ok := command.BySlug(request.Slug)
248 if !ok {
249 result.Code = ports.ErrUnknownCommand
250 result.Text = "unknown command: " + request.Slug
251 return result
252 }
253 if !cmd.PaletteVisible {
254 result.Code = ports.ErrNotScriptable
255 result.Text = request.Slug + " is not available in the palette"
256 return result
257 }
258 if cmd.Run == nil {
259 result.Code = ports.ErrNotScriptable
260 result.Text = request.Slug + " is not scriptable"
261 return result
262 }
263 // The frame's attachment token is the target capability. No registry/name lookup is
264 // performed, so the command cannot escape to another remote session.
265 err := sess.runMutation(func() error {
266 return cmd.Run(paletteExec{d: d, sess: sess, ac: token.ac, effect: token.effect}, request.Args)
267 })
268 if err == nil {
269 result.OK = true
270 return result
271 }
272 if errors.Is(err, command.ErrInvalidArguments) {
273 result.Code = ports.ErrInvalidCommandArgs
274 result.Text = "usage: " + cmd.Usage
275 return result
276 }
277 result.Code = ports.ErrInternal
278 result.Text = err.Error()
279 return result
280}
281
282// resetOutput rebases only the exact attachment and transport admitted
283// by the frame router. sendMu serializes the rebase with every output send;
284// transport revalidation under that lock rejects a link replaced while waiting.
285func (d *Daemon) resetOutput(token attachmentConnectionToken) bool {
286 ac := token.ac
287 if ac == nil || token.effect == nil || token.effect.ended.Load() {
288 return false
289 }
290 ac.sendMu.Lock()
291 if token.effect.ended.Load() || !token.attachmentCurrent() || ac.output == nil {
292 ac.sendMu.Unlock()
293 return false
294 }
295 ac.rebaseOutput()
296 ac.pipelineCache = composeCacheInput{}
297 ac.pipelineScratch = composeCacheInput{}
298 ac.sendMu.Unlock()
299 go d.paint(token.sess, ac, true, token.lease)
300 return true
301}
302
303func (d *Daemon) ackOutput(token attachmentConnectionToken, epoch, state uint64) bool {
304 ac := token.ac
305 if token.effect == nil || token.effect.ended.Load() {
306 return false
307 }
308 ac.sendMu.Lock()
309 if ac.output == nil {
310 ac.sendMu.Unlock()
311 return false
312 }
313 acknowledged := ac.output.ack(epoch, state)
314 ac.sendMu.Unlock()
315 if acknowledged {
316 if rc := token.sess.renderCoordinator(); rc != nil {
317 rc.notifyAckForLease(token.lease)
318 }
319 }
320 return acknowledged
321}