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