1package daemon
2
3import (
4 "sync"
5
6 "github.com/bnema/vev/internal/ports"
7 "github.com/bnema/vev/internal/protocol"
8)
9
10// Suspension retains membership for transport ownership and session teardown,
11// but removes the render lease and geometry claim. The lifecycle gate, rather
12// than membership alone, admits all ordinary attachment effects.
13func (d *Daemon) suspendAttachment(token attachmentCapability, request protocol.SuspendAttachment) error {
14 if request.Validate() != nil || token.ac == nil {
15 return errAttachmentTransition
16 }
17 ac, sess := token.ac, token.sess
18 deadline := newAttachmentEffectDrainDeadline(d.clock)
19 defer deadline.stop()
20 refreshObservations := false
21 frozen := freezeAttachmentEffectGatesWith(attachmentEffectFreezeOptions{done: deadline.Done, afterFrozen: func(ac *attachedClient) {
22 if d.afterAttachmentEffectGateFrozen != nil {
23 d.afterAttachmentEffectGateFrozen("suspend", ac)
24 }
25 }}, ac)
26 defer func() {
27 frozen.unfreeze()
28 if refreshObservations {
29 d.reconcileAllRouteHistories()
30 }
31 }()
32 if !frozen.acquired || !frozen.drained || !d.sourceAttachmentCapabilityCurrentFrozen(token) || ac.attachmentActivity() != attachmentActive {
33 return errAttachmentTransition
34 }
35 d.mu.Lock()
36 sess.mu.Lock()
37 if d.closing || d.sessions[sess.id] != sess || !token.currentInSessionLocked(sess) {
38 sess.mu.Unlock()
39 d.mu.Unlock()
40 return errAttachmentTransition
41 }
42 target := protocol.ExactSessionTarget{LifecycleID: sess.incarnation, SessionName: sess.name}
43 var cleanup renderLifecycleCleanup
44 if rc := sess.coordinator.Load(); rc != nil {
45 cleanup = rc.beginDetach(ac)
46 }
47 sess.geometry.removeAttachmentLocked(sess.core(), ac)
48 token.lease = nil
49 ac.publishFrozenAttachmentCapability(token)
50 ac.lifecycle.mu.Lock()
51 ac.lifecycle.activity = attachmentSuspended
52 ac.lifecycle.retained = target
53 ac.lifecycle.mu.Unlock()
54 // A suspended attachment drops modifier reports; its activation starts
55 // from a released Ctrl and the client reports a hold afresh.
56 ac.ctrlHeld.Store(false)
57 sess.mu.Unlock()
58 d.mu.Unlock()
59 cleanup.finish()
60 ac.clearSamePeerOffer()
61 d.retirePicker(ac)
62 d.closePalette(ac)
63 sess.geometry.reconcileAndInvalidate(d, sess, nil, "attachment_suspension.go")
64 // Ordered SendServer is a transport publication barrier after drained renders.
65 if err := d.boundedSendErr(ac, protocol.AttachmentSuspended{RequestID: request.RequestID, Target: target}); err != nil {
66 return err
67 }
68 // Arm the daemon-owned safety expiry only after the ACK is on the wire and
69 // while the transition is still frozen, so a racing activation must either
70 // observe the record or already own this gate.
71 d.armSuspendedExpiry(ac, sess)
72 refreshObservations = true
73 return nil
74}
75
76func (d *Daemon) activateAttachment(ac *attachedClient, expected transportSnapshot, request protocol.ActivateAttachment) error {
77 if ac == nil || request.Validate() != nil {
78 return errAttachmentTransition
79 }
80 deadline := newAttachmentEffectDrainDeadline(d.clock)
81 defer deadline.stop()
82 refreshObservations := false
83 frozen := freezeAttachmentEffectGatesWith(attachmentEffectFreezeOptions{done: deadline.Done}, ac)
84 defer func() {
85 frozen.unfreeze()
86 if refreshObservations {
87 d.reconcileAllRouteHistories()
88 }
89 }()
90 if !frozen.acquired || !frozen.drained {
91 return errAttachmentTransition
92 }
93 sess := ac.currentAttachmentSession()
94 if sess == nil {
95 return errAttachmentTransition
96 }
97 d.mu.Lock()
98 sess.mu.Lock()
99 ac.lifecycle.mu.Lock()
100 valid := ac.lifecycle.activity == attachmentSuspended && ac.lifecycle.retained == request.Target
101 ac.lifecycle.mu.Unlock()
102 if !valid || d.closing || d.sessions[sess.id] != sess || sess.incarnation != request.Target.LifecycleID || sess.name != request.Target.SessionName || !attachmentRegisteredLocked(sess, ac) || !ac.transportSnapshotCurrent(expected) {
103 sess.mu.Unlock()
104 d.mu.Unlock()
105 return errAttachmentTransition
106 }
107 geometry := request.Geometry()
108 ac.setGeometry(geometry)
109 rc := d.ensureRenderCoordinatorPrelocked(sess)
110 rc.mu.Lock()
111 lease := rc.attachWithReadinessLocked(ac, true)
112 rc.mu.Unlock()
113 sess.geometry.claimLocked(sess.core(), ac, geometry)
114 // A committed activation ends the suspended lifetime: cancel the daemon-owned
115 // safety expiry before the new capability generation is published.
116 d.clearSuspendedExpiryLocked(ac)
117 token := ac.publishFrozenAttachmentCapability(attachmentCapability{sess: sess, ac: ac, transport: expected, lease: lease})
118 // Reserve the sole activation effect under the still-frozen gate. No ordinary
119 // effect can enter until the full output and correlated success both commit.
120 ac.lifecycle.mu.Lock()
121 ac.lifecycle.activity = attachmentActivating
122 ac.lifecycle.inFlight++
123 ac.lifecycle.mu.Unlock()
124 effect := &attachmentEffect{attachmentCapability: token, lifecycle: &ac.lifecycle}
125 defer effect.End()
126 sess.mu.Unlock()
127 d.mu.Unlock()
128 sess.geometry.reconcileAndInvalidate(d, sess, nil, "attachment_suspension.go")
129 ac.sendMu.Lock()
130 ac.publishWindowLocked(request.Size.Rows, true)
131 ac.sendMu.Unlock()
132 paintDone := make(chan struct{})
133 paintTimer := d.clock.NewTimer(detachNotifyTimeout)
134 go func() {
135 select {
136 case <-paintTimer.C():
137 select {
138 case <-paintDone:
139 return
140 default:
141 }
142 d.log.Warn("activation publication timed out; force closing client transport")
143 _ = ac.closeCapturedTransport(expected.transport)
144 case <-paintDone:
145 paintTimer.Stop()
146 }
147 }()
148 paintResult := d.paintWithActivationEffect(sess, ac, true, lease, effect)
149 close(paintDone)
150 if paintResult != paintEmitted || !effect.current() {
151 return errAttachmentTransition
152 }
153 ac.sendMu.Lock()
154 ac.output.lockView()
155 output := ac.output
156 ack := protocol.AttachmentActivated{RequestID: request.RequestID, Identity: output.lastViewContext.Route, Epoch: output.epoch, State: output.next, ViewPublication: output.lastViewContext.Publication}
157 ac.output.unlockView()
158 ac.sendMu.Unlock()
159 if ack.Validate() != nil {
160 return errAttachmentTransition
161 }
162 if err := d.boundedSendErr(ac, ack); err != nil {
163 return err
164 }
165 ac.lifecycle.mu.Lock()
166 ac.lifecycle.activity = attachmentActive
167 ac.lifecycle.retained = protocol.ExactSessionTarget{}
168 ac.lifecycle.mu.Unlock()
169 refreshObservations = true
170 return nil
171}
172
173// suspendedAttachmentRetention is the daemon-owned safety expiry for exactly one
174// suspended attachment incarnation. generation and transport pin the suspension
175// that armed it, so a stale record can never retire a later activation, a
176// rebound transport, or a different session lifetime.
177type suspendedAttachmentRetention struct {
178 ac *attachedClient
179 sess *session
180 generation uint64
181 transport transportSnapshot
182 timer ports.Timer
183 done chan struct{}
184 doneOnce sync.Once
185}
186
187// stop ends the record exactly once: the timer is cancelled and any watcher is
188// released. It is safe to call from the watcher itself.
189func (r *suspendedAttachmentRetention) stop() {
190 if r == nil {
191 return
192 }
193 if r.timer != nil {
194 r.timer.Stop()
195 }
196 r.closeDone()
197}
198
199func (r *suspendedAttachmentRetention) closeDone() {
200 if r == nil || r.done == nil {
201 return
202 }
203 r.doneOnce.Do(func() { close(r.done) })
204}
205
206// armSuspendedExpiry publishes the daemon-owned safety expiry for a suspension
207// that has already committed and acknowledged. The caller must still hold ac's
208// effect-gate freeze, so a racing activation is serialized behind the same gate
209// and cannot miss the record. The watcher is admitted from the connection
210// handler, so daemon shutdown joins it only after connWg drains: no Add can race
211// the terminal Wait.
212func (d *Daemon) armSuspendedExpiry(ac *attachedClient, sess *session) {
213 if d == nil || ac == nil || sess == nil || d.suspendedSafetyExpiry <= 0 {
214 return
215 }
216 transport := ac.transportSnapshot()
217 if transport.transport == nil {
218 return
219 }
220 record := &suspendedAttachmentRetention{
221 ac: ac,
222 sess: sess,
223 generation: ac.lifecycle.generationValue(),
224 transport: transport,
225 timer: d.clock.NewTimer(d.suspendedSafetyExpiry),
226 done: make(chan struct{}),
227 }
228 d.mu.Lock()
229 if d.closing {
230 d.mu.Unlock()
231 record.stop()
232 return
233 }
234 // Re-suspension replaces any prior record for this attachment; the stale
235 // watcher must not outlive the suspension that armed it.
236 if previous := d.suspended[ac]; previous != nil {
237 previous.stop()
238 }
239 d.suspended[ac] = record
240 d.attachmentCleanupWg.Add(1)
241 d.mu.Unlock()
242 go func() {
243 defer d.attachmentCleanupWg.Done()
244 select {
245 case <-record.timer.C():
246 d.expireSuspendedAttachment(record)
247 case <-record.done:
248 }
249 }()
250}
251
252// clearSuspendedExpiryLocked cancels ac's safety expiry. Caller holds d.mu.
253func (d *Daemon) clearSuspendedExpiryLocked(ac *attachedClient) {
254 if d == nil || ac == nil {
255 return
256 }
257 record := d.suspended[ac]
258 if record == nil {
259 return
260 }
261 delete(d.suspended, ac)
262 record.stop()
263}
264
265// clearSuspendedExpiry cancels ac's safety expiry. Callers must not hold daemon,
266// session, or routing locks.
267func (d *Daemon) clearSuspendedExpiry(ac *attachedClient) {
268 if d == nil || ac == nil {
269 return
270 }
271 d.mu.Lock()
272 d.clearSuspendedExpiryLocked(ac)
273 d.mu.Unlock()
274}
275
276// purgeSuspendedForSessionLocked cancels every safety expiry owned by sess.
277// Caller holds d.mu.
278func (d *Daemon) purgeSuspendedForSessionLocked(sess *session) {
279 if d == nil || sess == nil {
280 return
281 }
282 for ac, record := range d.suspended {
283 if record.sess != sess {
284 continue
285 }
286 delete(d.suspended, ac)
287 record.stop()
288 }
289}
290
291// purgeAllSuspendedLocked cancels every safety expiry. Caller holds d.mu.
292func (d *Daemon) purgeAllSuspendedLocked() {
293 if d == nil {
294 return
295 }
296 for ac, record := range d.suspended {
297 delete(d.suspended, ac)
298 record.stop()
299 }
300}
301
302// expireSuspendedAttachment force-retires one suspended attachment whose
303// daemon-owned safety timer fired. It freezes and drains the exact gate before
304// taking any architecture lock, then revalidates the record, lifecycle
305// generation, transport incarnation, suspended activity, and session
306// registration under that same freeze. The retained session name is
307// deliberately not compared: a rename keeps the incarnation and must not
308// postpone terminal eviction. Expiry is a no-notice terminal eviction with no
309// resume credential, and it leaves the session to existing headless lifetime
310// policy.
311func (d *Daemon) expireSuspendedAttachment(record *suspendedAttachmentRetention) {
312 if d == nil || record == nil || record.ac == nil {
313 return
314 }
315 ac := record.ac
316 if d.beforeSuspendedExpiryFreeze != nil {
317 d.beforeSuspendedExpiryFreeze(ac)
318 }
319 // Freeze and drain without a bound: terminal eviction acts on a suspended
320 // attachment whose gate has no admitted effects, and a gate owner always
321 // releases, so waiting here can never strand the expiry the way an expired
322 // acquisition budget would.
323 frozen := freezeAttachmentEffectGatesWith(attachmentEffectFreezeOptions{afterFrozen: func(ac *attachedClient) {
324 if d.afterAttachmentEffectGateFrozen != nil {
325 d.afterAttachmentEffectGateFrozen("expire", ac)
326 }
327 }}, ac)
328 if !frozen.acquired || !frozen.drained {
329 frozen.unfreeze()
330 return
331 }
332 d.mu.Lock()
333 sess := record.sess
334 detached := false
335 if d.suspended[ac] == record && !d.closing && sess != nil && d.sessions[sess.id] == sess && ac.currentAttachmentSession() == sess {
336 sess.mu.Lock()
337 ac.lifecycle.mu.Lock()
338 current := ac.lifecycle.activity == attachmentSuspended &&
339 ac.lifecycle.generationValue() == record.generation &&
340 ac.transportSnapshotCurrent(record.transport) &&
341 attachmentRegisteredLocked(sess, ac)
342 ac.lifecycle.mu.Unlock()
343 if current {
344 detached = detachFrozenAttachmentLocked(sess, ac)
345 }
346 sess.mu.Unlock()
347 if detached {
348 delete(d.suspended, ac)
349 }
350 }
351 d.mu.Unlock()
352 frozen.unfreeze()
353 if !detached {
354 return
355 }
356 // The expiry owns the record now; release a watcher that may still be
357 // selecting on the fired timer channel.
358 record.stop()
359 d.log.Warn("suspended attachment expired", "session", sess.nameSnapshot(), "expiry", d.suspendedSafetyExpiry)
360 // Shared terminal eviction: geometry reconciliation, coordinator detach,
361 // picker/palette/output cleanup, and the physical transport close all run
362 // outside every architecture lock. Expiry never parks a resume credential.
363 d.finishClientGone(sess, ac, record.transport.transport, true, false)
364 d.reapAbandonedEphemeral(sess)
365}