1package daemon
2
3import (
4 "runtime"
5 "sync"
6 "time"
7
8 "github.com/bnema/vev/internal/ports"
9)
10
11type renderCoordinator struct {
12 mu sync.Mutex
13 opts renderCoordinatorOptions
14
15 // Coalesced pending state: latest wins, reset stays sticky, and at most
16 // one deadline wake may be armed at a time (cap-1).
17 pending bool
18 pendingReset bool
19 pendingUrgent bool
20 // queueMarked spans the coordinator's actual pending interval, from first
21 // enqueue until fire consumes the coalesced work.
22 queueMarked bool
23 queueCorrelation ports.RuntimeCorrelation
24 // pendingPreview is accounted separately from the attachment wake:
25 // viewers may consume a coalesced target snapshot while output is
26 // still ACK-blocked.
27 pendingPreview bool
28 // ackDeferred records that at least one attachment's render wake was
29 // blocked by output-window capacity. ACK notifications use ackDeferredFor
30 // to flush only the acknowledging attachment's work.
31 ackDeferred bool
32 ackDeferredFor map[*attachedClient]bool
33 // pendingGeneration identifies the latest coalesced session mutation.
34 // delivered records which attachment incarnations have consumed it, so a
35 // blocked peer can catch up without replaying the wake to healthy peers.
36 pendingGeneration uint64
37 delivered map[*attachedClient]uint64
38 // pendingShared is set by session-owned producers. Attachment-owned
39 // invalidations retain only their exact target lease here.
40 pendingShared bool
41 pendingTargets map[*attachedClient]*attachmentLease
42 // ackBlocked retains the first rejected ACK-capacity probe until the
43 // matching consume or lifecycle closure releases that exact interval.
44 ackBlocked *ackBlockedSpan
45 // deadlineDue closes the handoff between a timer worker receiving its tick
46 // and fire observing unavailable ACK capacity.
47 deadlineDue bool
48 coalesced int
49 // outputPressure carries recent bulk-burst pressure between completed
50 // batches. It selects a bounded 8–16ms deadline without letting a busy
51 // batch perpetually reset its own timer.
52 outputPressure int
53
54 // previewWakes tracks picker subscriptions by viewer, so one inactive session
55 // cannot replace another viewer's live preview.
56 previewWakes map[*attachedClient]previewSubscription
57
58 // attachments and leases are coordinator-owned lifecycle snapshots. Shared
59 // render work is not tied to any one attachment; leases only fence callbacks
60 // for the attachment that captured them.
61 attachments map[*attachedClient]struct{}
62 leases map[*attachedClient]*attachmentLease
63 torndown bool
64
65 resize resizeRequestMetadata
66 // Each lane owns one complete immutable worker token at a time.
67 resizeLane timerLane
68 retryLane timerLane
69 metrics renderCoordinatorBurstMetrics
70 // normalLane owns render invalidation deadlines. The supervisor is only
71 // waited by the session owner after terminal teardown has rejected future
72 // registrations.
73 normalLane timerLane
74 armed bool
75 supervisor timerSupervisor
76 // syncBatches is keyed by the stable pane owner. Each pane owns an
77 // independent watchdog; composition remains gated until every batch ends.
78 syncBatches map[*pane]*syncBatch
79 // syncRegistryVersion changes for every batch identity mutation. fire
80 // snapshots it before evaluating visibility and retries if a producer
81 // publishes or completes a batch during that unlocked evaluation.
82 syncRegistryVersion uint64
83}
84
85// newRenderCoordinator constructs a coordinator bound to opts. It starts no
86// goroutine; deadline scheduling is armed lazily by invalidate.
87func newRenderCoordinator(opts renderCoordinatorOptions) *renderCoordinator {
88 return &renderCoordinator{opts: opts}
89}
90
91// detachNormalTimerLocked removes the complete current token. The caller may
92// stop it only after dropping c.mu and must never wait for its worker.
93func (c *renderCoordinator) detachNormalTimerLocked() *timerToken {
94 return c.normalLane.detachLocked()
95}
96
97type syncBatch struct {
98 generation uint64
99 // renderable is evaluated outside c.mu: tab/session visibility locks may be
100 // acquired by this predicate.
101 renderable func() bool
102 force func()
103 lane timerLane
104}
105
106func detachSyncBatchLocked(batch *syncBatch) *timerToken { return batch.lane.detachLocked() }
107
108func (c *renderCoordinator) detachSyncTimersLocked() []*timerToken {
109 workers := make([]*timerToken, 0, len(c.syncBatches))
110 for pane, batch := range c.syncBatches {
111 workers = append(workers, detachSyncBatchLocked(batch))
112 delete(c.syncBatches, pane)
113 c.syncRegistryVersion++
114 }
115 return workers
116}
117
118// invalidate publishes a coordinator-owned producer transition. Callers that
119// carry an attachment identity must use invalidateForAttachment instead.
120func (c *renderCoordinator) invalidate(inv renderInvalidation) bool {
121 return c.invalidateForAttachment(nil, inv)
122}
123
124// invalidateForAttachment publishes only while source is still registered.
125// This check shares c.mu with attachment lifecycle changes, making a stale
126// timer callback unable to enqueue work for a newer attachment incarnation.
127// A nil source represents an internal pane/session invalidation.
128func (c *renderCoordinator) invalidateForAttachment(source *attachedClient, inv renderInvalidation) bool {
129 return c.invalidateForAttachmentAtResizeEpoch(source, 0, inv)
130}
131
132// invalidateForLease publishes an attachment-owned transition only for the
133// exact lifecycle incarnation that captured it.
134func (c *renderCoordinator) invalidateForLease(source *attachedClient, lease *attachmentLease, inv renderInvalidation) bool {
135 return c.invalidateForLeaseAtResizeEpoch(source, lease, 0, inv)
136}
137
138// invalidateForAttachmentAtResizeEpoch additionally requires epoch to remain
139// current when nonzero, so a completed resize cannot publish after a newer
140// request supersedes it.
141func (c *renderCoordinator) invalidateForAttachmentAtResizeEpoch(source *attachedClient, epoch uint64, inv renderInvalidation) bool {
142 return c.invalidateForLeaseAtResizeEpoch(source, nil, epoch, inv)
143}
144
145// renderInvalidationReservation is the in-memory half of an invalidation.
146// finish performs observer, timer, and callback work after ownership fences are
147// released.
148type renderInvalidationReservation struct {
149 coordinator *renderCoordinator
150 invalidation renderInvalidation
151 onInvalidate func(renderInvalidation)
152 queueStart bool
153 queueCorrelation ports.RuntimeCorrelation
154 observer ports.RuntimeObserver
155 old *timerToken
156 arm bool
157 generation uint64
158 delay time.Duration
159 clock ports.Clock
160}
161
162// reserveInvalidationForLeaseAtResizeEpoch publishes only coordinator memory.
163// It is safe inside a short pane-owner fence and performs no timer or observer
164// calls.
165func (c *renderCoordinator) reserveInvalidationForLeaseAtResizeEpoch(source *attachedClient, lease *attachmentLease, epoch uint64, inv renderInvalidation) (*renderInvalidationReservation, bool) {
166 c.mu.Lock()
167 defer c.mu.Unlock()
168 if c.torndown || (lease != nil && (!c.leaseCurrentLocked(lease, true) || lease.attachment != source)) ||
169 (source != nil && !c.attachmentRegisteredLocked(source)) ||
170 (epoch != 0 && (c.resize.epoch != epoch || c.resize.source != source || (lease != nil && c.resize.lease != lease))) {
171 return nil, false
172 }
173 reservation := &renderInvalidationReservation{coordinator: c, invalidation: inv, onInvalidate: c.opts.onInvalidate}
174 c.metrics.invalidations.Add(1)
175 c.pendingGeneration++
176 if c.pendingGeneration == 0 {
177 c.pendingGeneration = 1
178 }
179 wasPending, wasUrgent, wasPreviewPending := c.pending, c.pendingUrgent, c.pendingPreview
180 if !wasPending {
181 c.ackDeferred = false
182 c.ackDeferredFor = nil
183 c.deadlineDue = false
184 if c.opts.observer != nil {
185 c.queueMarked = true
186 c.queueCorrelation = ports.NewRuntimeCorrelation()
187 reservation.queueCorrelation = c.queueCorrelation
188 reservation.queueStart = true
189 }
190 }
191 c.pending = true
192 if source == nil {
193 c.pendingShared = true
194 c.pendingTargets = nil
195 } else if !c.pendingShared {
196 if c.pendingTargets == nil {
197 c.pendingTargets = make(map[*attachedClient]*attachmentLease)
198 }
199 targetLease := lease
200 if targetLease == nil {
201 targetLease = c.leases[source]
202 }
203 if targetLease != nil {
204 c.pendingTargets[source] = targetLease
205 }
206 }
207 c.pendingReset = c.pendingReset || inv.reset
208 c.pendingUrgent = c.pendingUrgent || inv.class == invalidateUrgent
209 c.pendingPreview = c.pendingPreview || len(c.previewWakes) != 0
210 c.coalesced++
211 reservation.arm = !wasPending || !c.armed || (!wasUrgent && c.pendingUrgent) || (!wasPreviewPending && c.pendingPreview) || c.deadlineDue
212 if reservation.arm {
213 _, reservation.old = c.normalLane.replaceLocked()
214 }
215 reservation.generation = c.normalLane.generation
216 reservation.delay = minOutputRenderDeadline + time.Duration(c.outputPressure)*time.Millisecond
217 if c.pendingUrgent {
218 reservation.delay = urgentRenderDeadline
219 }
220 reservation.clock = c.opts.clock
221 reservation.observer = c.opts.observer
222 c.armed = c.armed || reservation.arm
223 return reservation, true
224}
225
226func (r *renderInvalidationReservation) finish() {
227 if r == nil || r.coordinator == nil {
228 return
229 }
230 c := r.coordinator
231 if r.queueStart && r.observer != nil {
232 r.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", r.queueCorrelation, ports.RuntimeQueueEnqueued, 0, true))
233 }
234 stopDetachedTimer(r.old)
235 if !r.arm || r.clock == nil {
236 if r.onInvalidate != nil {
237 r.onInvalidate(r.invalidation)
238 }
239 return
240 }
241 timer := r.clock.NewTimer(r.delay)
242 timerC := timer.C()
243 c.mu.Lock()
244 valid := !c.torndown && c.pending && c.normalLane.generation == r.generation
245 var token *timerToken
246 if valid && timerC != nil {
247 token = c.normalLane.publishLocked(r.generation, timer)
248 valid = token != nil
249 if valid {
250 c.supervisor.startLocked(token, timerC, func() {
251 c.markDeadlineDue(token)
252 c.fireFromTimer(token, false, true)
253 })
254 }
255 }
256 c.mu.Unlock()
257 if !valid {
258 timer.Stop()
259 } else if timerC == nil {
260 timer.Stop()
261 c.fire(r.generation, false, true)
262 }
263 if r.onInvalidate != nil {
264 r.onInvalidate(r.invalidation)
265 }
266 if valid && timerC != nil {
267 runtime.Gosched()
268 }
269}
270
271// invalidateForLeaseAtResizeEpoch is the attachment-callback variant: it
272// rejects a callback whose source pointer still matches but whose lifecycle
273// lease has been revoked.
274func (c *renderCoordinator) invalidateForLeaseAtResizeEpoch(source *attachedClient, lease *attachmentLease, epoch uint64, inv renderInvalidation) bool {
275 reservation, ok := c.reserveInvalidationForLeaseAtResizeEpoch(source, lease, epoch, inv)
276 if !ok {
277 return false
278 }
279 reservation.finish()
280 return true
281}
282
283// notifyAck reports that the client acknowledged an output state, releasing
284// at most one attachment's blocked wake. It never bypasses an unexpired normal
285// or urgent deadline.
286func (c *renderCoordinator) notifyAck() { c.notifyAckForLease(nil) }
287
288// notifyAckForLease prevents an acknowledgement captured by an obsolete active
289// frame from flushing another attachment's render queue. A nil lease is
290// reserved for coordinator-owned/direct test callers.
291func (c *renderCoordinator) notifyAckForLease(lease *attachmentLease) {
292 c.mu.Lock()
293 if lease != nil {
294 if !c.leaseCurrentLocked(lease, true) || !c.ackDeferredFor[lease.attachment] {
295 c.mu.Unlock()
296 return
297 }
298 } else if !c.ackDeferred && !c.deadlineDue {
299 c.mu.Unlock()
300 return
301 }
302 if !c.pending {
303 c.mu.Unlock()
304 return
305 }
306 gen := c.normalLane.generation
307 blocked := c.ackBlocked
308 c.ackBlocked = nil
309 c.mu.Unlock()
310 if blocked != nil {
311 blocked.finish(true)
312 }
313 c.fire(gen, false, false)
314}
315
316// retryCapacity restores only the attachment whose wake failed the final
317// capacity check. Readiness is an atomic observation, so paint may discover
318// backpressure after fire has marked the wake delivered. Requeueing the exact
319// wake generation keeps that stale result from clearing shared work and avoids
320// making a healthy peer wait for the blocked attachment.
321func (c *renderCoordinator) retryCapacity(w renderWake, lease *attachmentLease) {
322 if lease == nil || w.generation == 0 {
323 return
324 }
325 var queueStart bool
326 c.mu.Lock()
327 if !c.leaseCurrentLocked(lease, true) || c.pendingGeneration < w.generation {
328 c.mu.Unlock()
329 return
330 }
331 // A newer wake already attempted this lease; its callback owns any
332 // capacity retry. An older callback may still widen a newer targeted wake
333 // to include this lease when that newer wake never reached it.
334 if c.pendingGeneration > w.generation && lease.deliveredGeneration == c.pendingGeneration {
335 c.mu.Unlock()
336 return
337 }
338 if !c.pending {
339 c.pending = true
340 c.pendingReset = w.reset
341 c.pendingUrgent = w.urgent
342 c.pendingShared = false
343 c.pendingTargets = map[*attachedClient]*attachmentLease{lease.attachment: lease}
344 c.coalesced = max(1, w.coalesced)
345 c.deadlineDue = false
346 c.armed = false
347 if c.opts.observer != nil && !c.queueMarked {
348 c.queueMarked = true
349 c.queueCorrelation = ports.NewRuntimeCorrelation()
350 queueStart = true
351 }
352 } else {
353 c.pendingReset = c.pendingReset || w.reset
354 c.pendingUrgent = c.pendingUrgent || w.urgent
355 if !c.pendingShared {
356 if c.pendingTargets == nil {
357 c.pendingTargets = make(map[*attachedClient]*attachmentLease)
358 }
359 c.pendingTargets[lease.attachment] = lease
360 }
361 }
362 if c.delivered != nil {
363 delete(c.delivered, lease.attachment)
364 }
365 if c.ackDeferredFor == nil {
366 c.ackDeferredFor = make(map[*attachedClient]bool)
367 }
368 c.ackDeferredFor[lease.attachment] = true
369 c.ackDeferred = true
370 generation := c.pendingGeneration
371 queueCorrelation := c.queueCorrelation
372 observer := c.opts.observer
373 c.supervisor.startTaskLocked(func() { c.retryPendingGeneration(generation) })
374 c.mu.Unlock()
375 if queueStart && observer != nil {
376 observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueEnqueued, 0, true))
377 }
378}
379
380func (c *renderCoordinator) retryPendingGeneration(generation uint64) {
381 c.mu.Lock()
382 if c.torndown || !c.pending || c.pendingGeneration != generation {
383 c.mu.Unlock()
384 return
385 }
386 gen, token := c.normalLane.generation, c.normalLane.token
387 c.mu.Unlock()
388 c.fireWithTimerToken(token, gen, false, false)
389}
390
391// noteSyncBegin records a synchronized-output batch for its stable pane and
392// arms that pane's watchdog. Overlapping pane batches are independent.
393// markDeadlineDue records a received timer tick before the worker probes ACK
394// readiness, so a concurrent ACK cannot miss an already-expired deadline.
395func (c *renderCoordinator) markDeadlineDue(token *timerToken) {
396 c.mu.Lock()
397 if c.fireValidLocked(token, token.generation, false) {
398 c.deadlineDue = true
399 }
400 c.mu.Unlock()
401}
402
403func (c *renderCoordinator) noteSyncBegin(p *pane, gen uint64, force ...func()) {
404 var renderable func() bool
405 if c.opts.syncRenderable != nil {
406 renderable = func() bool { return c.opts.syncRenderable(p) }
407 }
408 c.noteSyncBeginWithRenderability(p, gen, renderable, force...)
409}
410
411// syncTimerCleanup carries detached timer work across an enclosing pane or
412// coordinator critical section. finish never waits for a worker.
413type syncTimerCleanup struct{ tokens []*timerToken }
414
415func (cleanup *syncTimerCleanup) add(token *timerToken) {
416 if token != nil {
417 cleanup.tokens = append(cleanup.tokens, token)
418 }
419}
420
421func (cleanup *syncTimerCleanup) append(other syncTimerCleanup) {
422 cleanup.tokens = append(cleanup.tokens, other.tokens...)
423}
424
425func (cleanup syncTimerCleanup) finish() {
426 for _, token := range cleanup.tokens {
427 stopDetachedTimer(token)
428 }
429}
430
431// noteSyncBeginWithRenderability records lifecycle unconditionally while the
432// supplied predicate decides dynamically whether this batch gates composition.
433func (c *renderCoordinator) noteSyncBeginWithRenderability(p *pane, gen uint64, renderable func() bool, force ...func()) {
434 c.beginSyncBatchWithRenderability(p, gen, renderable, force...).finish()
435}
436
437// beginSyncBatchWithRenderability publishes a batch and returns every timer
438// detached while doing so. Callers that already own pane.mu defer finish until
439// after releasing the parsing fence.
440func (c *renderCoordinator) beginSyncBatchWithRenderability(p *pane, gen uint64, renderable func() bool, force ...func()) syncTimerCleanup {
441 var cleanup syncTimerCleanup
442 c.mu.Lock()
443 if c.torndown {
444 c.mu.Unlock()
445 return cleanup
446 }
447 if c.syncBatches == nil {
448 c.syncBatches = make(map[*pane]*syncBatch)
449 }
450 if previous := c.syncBatches[p]; previous != nil {
451 cleanup.add(detachSyncBatchLocked(previous))
452 }
453 batch := &syncBatch{generation: gen, renderable: renderable}
454 if len(force) != 0 {
455 batch.force = force[0]
456 }
457 c.syncBatches[p] = batch
458 c.syncRegistryVersion++
459 clock := c.opts.clock
460 c.mu.Unlock()
461 if clock == nil {
462 return cleanup
463 }
464
465 // Both timer operations are external. A reentrant callback may replace or
466 // remove batch; identity validation below rejects this stale timer.
467 timer := clock.NewTimer(maxSyncUpdateDuration)
468 timerC := timer.C()
469 c.mu.Lock()
470 valid := !c.torndown && c.syncBatches[p] == batch && batch.generation == gen && batch.lane.generation == 0
471 var token *timerToken
472 if valid && timerC != nil {
473 batch.lane.generation = gen
474 token = batch.lane.publishLocked(gen, timer)
475 valid = token != nil
476 if valid {
477 c.supervisor.startLocked(token, timerC, func() { c.runSyncWatchdog(p, batch, token) })
478 }
479 }
480 c.mu.Unlock()
481 if !valid || timerC == nil {
482 cleanup.add(&timerToken{timer: timer})
483 }
484 return cleanup
485}
486
487// detachSyncBatchGeneration removes only the exact pane generation. It returns
488// timer cleanup to the caller so no Timer method runs under an enclosing pane
489// or coordinator lock.
490func (c *renderCoordinator) detachSyncBatchGeneration(p *pane, gen uint64) syncTimerCleanup {
491 var cleanup syncTimerCleanup
492 c.mu.Lock()
493 batch := c.syncBatches[p]
494 if batch != nil && batch.generation == gen {
495 delete(c.syncBatches, p)
496 c.syncRegistryVersion++
497 cleanup.add(detachSyncBatchLocked(batch))
498 }
499 c.mu.Unlock()
500 return cleanup
501}
502
503func (c *renderCoordinator) runSyncWatchdog(p *pane, batch *syncBatch, token *timerToken) {
504 // Keep the batch registered while force runs. A concurrent render must
505 // continue to observe the synchronized-output gate until the VT has
506 // authoritatively closed the batch.
507 c.mu.Lock()
508 current := c.syncBatches[p]
509 valid := !c.torndown && current == batch && current.generation == token.generation && current.lane.token == token
510 c.mu.Unlock()
511 if !valid {
512 return
513 }
514 if batch.force != nil {
515 batch.force()
516 }
517
518 // force may synchronously end this batch, replace it with a newer
519 // generation, or tear down the coordinator. Remove only the exact
520 // snapshot that expired, then publish one urgent completion wake.
521 c.mu.Lock()
522 current = c.syncBatches[p]
523 valid = !c.torndown && current == batch && current.generation == token.generation && current.lane.token == token
524 var stopped *timerToken
525 if valid {
526 delete(c.syncBatches, p)
527 c.syncRegistryVersion++
528 stopped = detachSyncBatchLocked(batch)
529 if len(c.syncBatches) == 0 && c.pending {
530 c.pendingUrgent = true
531 }
532 }
533 c.mu.Unlock()
534 if valid {
535 stopDetachedTimer(stopped)
536 // Reevaluate the aggregate gate only after forcing this pane.
537 c.fireCurrent(true)
538 }
539}
540
541// noteSyncEnd records completion for exactly one pane batch. It only flushes
542// after the aggregate gate opens.
543func (c *renderCoordinator) noteSyncEnd(p *pane, gen uint64) {
544 // Visibility may acquire session/tab locks, so evaluate it outside c.mu.
545 c.mu.Lock()
546 batch := c.syncBatches[p]
547 if batch == nil || batch.generation != gen {
548 c.mu.Unlock()
549 return
550 }
551 renderable := batch.renderable
552 c.mu.Unlock()
553 wasRenderable := renderable == nil || renderable()
554 if c.removeSyncEnd(p, gen, wasRenderable) {
555 c.fireCurrent(false)
556 }
557}
558
559// removeSyncEnd removes a batch without invoking a predicate or callback. PTY
560// processing calls it while pane.mu is still held, so a deadline can never
561// compose the post-Write partial screen between mutation and publication.
562// The caller must invoke fireCurrent only after releasing pane.mu.
563func (c *renderCoordinator) removeSyncEnd(p *pane, gen uint64, renderable bool) bool {
564 c.mu.Lock()
565 batch := c.syncBatches[p]
566 if batch == nil || batch.generation != gen {
567 c.mu.Unlock()
568 return false
569 }
570 delete(c.syncBatches, p)
571 c.syncRegistryVersion++
572 worker := detachSyncBatchLocked(batch)
573 if renderable && len(c.syncBatches) == 0 && c.pending {
574 c.pendingUrgent = true
575 }
576 c.mu.Unlock()
577 stopDetachedTimer(worker)
578 return renderable
579}
580
581// noteSyncPaneRemoved releases a pane watchdog when pane lifecycle ends.
582func (c *renderCoordinator) noteSyncPaneRemoved(p *pane) {
583 c.mu.Lock()
584 var worker *timerToken
585 if batch := c.syncBatches[p]; batch != nil {
586 delete(c.syncBatches, p)
587 c.syncRegistryVersion++
588 worker = detachSyncBatchLocked(batch)
589 }
590 c.mu.Unlock()
591 stopDetachedTimer(worker)
592}
593
594type previewSubscription struct {
595 generation uint64
596 fn func(renderWake)
597}
598
599// subscribePreviewFor installs the dynamic picker observer owned by viewer.
600// A newer picker generation wins, so a delayed subscription cannot replace it.
601func (c *renderCoordinator) subscribePreviewFor(viewer *attachedClient, generation uint64, fn func(renderWake)) bool {
602 if viewer == nil || fn == nil {
603 return false
604 }
605 c.mu.Lock()
606 defer c.mu.Unlock()
607 if c.torndown {
608 return false
609 }
610 if current, ok := c.previewWakes[viewer]; ok && current.generation > generation {
611 return false
612 }
613 if c.previewWakes == nil {
614 c.previewWakes = make(map[*attachedClient]previewSubscription)
615 }
616 c.previewWakes[viewer] = previewSubscription{generation: generation, fn: fn}
617 return true
618}
619
620// teardownPreviewFor removes viewer's observer only when it still belongs to
621// generation. A delayed teardown cannot clear a newer subscription.
622func (c *renderCoordinator) teardownPreviewFor(viewer *attachedClient, generation uint64) {
623 c.mu.Lock()
624 if current, ok := c.previewWakes[viewer]; ok && current.generation == generation {
625 delete(c.previewWakes, viewer)
626 }
627 c.mu.Unlock()
628}
629
630func (c *renderCoordinator) attachmentRegisteredLocked(ac *attachedClient) bool {
631 if ac == nil {
632 return false
633 }
634 _, ok := c.attachments[ac]
635 return ok
636}
637
638func (c *renderCoordinator) installLeaseLocked(ac *attachedClient, ready bool) *attachmentLease {
639 if ac == nil {
640 return nil
641 }
642 if c.attachments == nil {
643 c.attachments = make(map[*attachedClient]struct{})
644 }
645 if c.leases == nil {
646 c.leases = make(map[*attachedClient]*attachmentLease)
647 }
648 c.attachments[ac] = struct{}{}
649 if lease := c.leases[ac]; lease != nil && lease.active {
650 lease.ready = ready
651 return lease
652 }
653 lease := &attachmentLease{attachment: ac, ready: ready, active: true}
654 c.leases[ac] = lease
655 if c.delivered != nil {
656 delete(c.delivered, ac)
657 }
658 if c.ackDeferredFor != nil {
659 delete(c.ackDeferredFor, ac)
660 }
661 return lease
662}
663
664// attachmentLease returns the exact currently published lease for ac. Welcome
665// continuations capture this before their transport write and revalidate it at
666// every side-effect boundary.
667func (c *renderCoordinator) attachmentLease(ac *attachedClient) *attachmentLease {
668 c.mu.Lock()
669 defer c.mu.Unlock()
670 if c.torndown {
671 return nil
672 }
673 lease := c.leases[ac]
674 if lease == nil || !lease.active {
675 return nil
676 }
677 return lease
678}
679
680func (c *renderCoordinator) leaseCurrent(lease *attachmentLease, ready bool) bool {
681 c.mu.Lock()
682 defer c.mu.Unlock()
683 return c.leaseCurrentLocked(lease, ready)
684}
685
686// leaseCurrentLocked is a short, non-reentrant effect admission check. It
687// must not be used to retain c.mu across input routing, transport operations,
688// or other arbitrary handler work.
689func (c *renderCoordinator) leaseCurrentLocked(lease *attachmentLease, ready bool) bool {
690 return !c.torndown && lease != nil && lease.active &&
691 c.leases[lease.attachment] == lease && (!ready || lease.ready)
692}
693
694// attach binds an already-handshaken internal attachment. Route/resume use
695// attachWithReadiness(..., false) until their Welcome frame is accepted.
696func (c *renderCoordinator) attach(ac *attachedClient) { c.attachWithReadiness(ac, true) }
697
698func (c *renderCoordinator) attachWithReadiness(ac *attachedClient, ready bool) *attachmentLease {
699 c.mu.Lock()
700 defer c.mu.Unlock()
701 return c.attachWithReadinessLocked(ac, ready)
702}
703
704// attachWithReadinessLocked publishes one attachment while the coordinator
705// lock is already held by an ordered transition publication.
706func (c *renderCoordinator) attachWithReadinessLocked(ac *attachedClient, ready bool) *attachmentLease {
707 if c.torndown {
708 return nil
709 }
710 return c.installLeaseLocked(ac, ready)
711}
712
713// rebindAttachmentWithReadinessLocked retires the previous lease before
714// publishing a new one for the same attachment object. A queued wake carries
715// the old lease pointer, so reusing it would let that wake paint a rebound
716// transport.
717func (c *renderCoordinator) rebindAttachmentWithReadinessLocked(ac *attachedClient, ready bool) *attachmentLease {
718 if c.torndown {
719 return nil
720 }
721 if lease := c.leases[ac]; lease != nil {
722 lease.active = false
723 delete(c.leases, ac)
724 }
725 return c.installLeaseLocked(ac, ready)
726}
727
728// markAttachmentReady completes only the captured attachment incarnation.
729func (c *renderCoordinator) markAttachmentReady(lease *attachmentLease) bool {
730 c.mu.Lock()
731 defer c.mu.Unlock()
732 if !c.leaseCurrentLocked(lease, false) {
733 return false
734 }
735 lease.ready = true
736 return true
737}
738
739// wakeCurrent validates an attachment-specific wake. Shared wakes have no
740// lease and remain valid while the coordinator is live; their paint callback
741// rechecks each attachment's connection fence.
742func (c *renderCoordinator) wakeCurrent(w renderWake) bool {
743 c.mu.Lock()
744 defer c.mu.Unlock()
745 return !c.torndown && (w.lease == nil || c.leaseCurrentLocked(w.lease, true))
746}
747
748type renderLifecycleCleanup struct {
749 tokens []*timerToken
750 observer ports.RuntimeObserver
751 queueCorrelation ports.RuntimeCorrelation
752 queueMarked bool
753 ackBlocked *ackBlockedSpan
754}
755
756func (cleanup renderLifecycleCleanup) finish() {
757 for _, token := range cleanup.tokens {
758 stopDetachedTimer(token)
759 }
760 if cleanup.queueMarked && cleanup.observer != nil {
761 cleanup.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", cleanup.queueCorrelation, ports.RuntimeQueueDequeued, 0, false))
762 }
763 if cleanup.ackBlocked != nil {
764 cleanup.ackBlocked.finish(false)
765 }
766}
767
768// beginDetach invalidates pending work. It never waits for detached workers.
769func (c *renderCoordinator) beginDetach(ac *attachedClient) renderLifecycleCleanup {
770 c.mu.Lock()
771 cleanup := c.beginDetachLocked(ac)
772 c.mu.Unlock()
773 return cleanup
774}
775
776// beginDetachLocked is the prelocked lifecycle seam used when registry,
777// current-session, and lease publication must share one commit boundary.
778func (c *renderCoordinator) beginDetachLocked(ac *attachedClient) renderLifecycleCleanup {
779 if ac == nil || !c.attachmentRegisteredLocked(ac) {
780 return renderLifecycleCleanup{}
781 }
782 delete(c.attachments, ac)
783 if lease := c.leases[ac]; lease != nil {
784 lease.active = false
785 delete(c.leases, ac)
786 }
787 delete(c.delivered, ac)
788 delete(c.ackDeferredFor, ac)
789 // Shared pending output belongs to the session, not the detached
790 // attachment. Keep it alive while another attachment remains registered.
791 if len(c.attachments) != 0 {
792 var cleanup renderLifecycleCleanup
793 if c.resize.source == ac {
794 _, resizeTimer := c.resizeLane.replaceLocked()
795 _, retryTimer := c.retryLane.replaceLocked()
796 cleanup.tokens = []*timerToken{resizeTimer, retryTimer}
797 }
798 return cleanup
799 }
800 return c.resetAttachmentLifecycleLocked()
801}
802
803func (c *renderCoordinator) resetAttachmentLifecycleLocked() renderLifecycleCleanup {
804 c.pending = false
805 c.pendingReset = false
806 c.pendingUrgent = false
807 c.ackDeferred = false
808 c.ackDeferredFor = nil
809 c.delivered = nil
810 c.pendingPreview = false
811 c.pendingShared = false
812 c.pendingTargets = nil
813 c.coalesced = 0
814 _, timer := c.normalLane.replaceLocked()
815 c.armed = false
816 _, resizeTimer := c.resizeLane.replaceLocked()
817 _, retryTimer := c.retryLane.replaceLocked()
818 cleanup := renderLifecycleCleanup{
819 tokens: []*timerToken{timer, resizeTimer, retryTimer},
820 observer: c.opts.observer,
821 queueCorrelation: c.queueCorrelation,
822 queueMarked: c.queueMarked,
823 ackBlocked: c.ackBlocked,
824 }
825 c.queueMarked, c.ackBlocked = false, nil
826 return cleanup
827}
828
829// noteDetach is for callers that hold no outer lifecycle locks.
830func (c *renderCoordinator) noteDetach(ac *attachedClient) {
831 c.beginDetach(ac).finish()
832}
833
834// beginSessionTeardown is terminal phase one. It rejects future worker
835// registration and atomically detaches every complete lane token.
836func (c *renderCoordinator) beginSessionTeardown() renderLifecycleCleanup {
837 c.mu.Lock()
838 if c.torndown {
839 c.mu.Unlock()
840 return renderLifecycleCleanup{}
841 }
842 c.torndown = true
843 for ac, lease := range c.leases {
844 lease.active = false
845 delete(c.attachments, ac)
846 }
847 c.leases = nil
848 c.attachments = nil
849 c.previewWakes = nil
850 c.pending = false
851 _, resizeTimer := c.resizeLane.replaceLocked()
852 _, retryTimer := c.retryLane.replaceLocked()
853 c.ackDeferred = false
854 c.ackDeferredFor = nil
855 c.delivered = nil
856 c.pendingPreview = false
857 c.pendingShared = false
858 c.pendingTargets = nil
859 _, timer := c.normalLane.replaceLocked()
860 c.armed = false
861 workers := c.detachSyncTimersLocked()
862 observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
863 ackBlocked := c.ackBlocked
864 c.queueMarked, c.ackBlocked = false, nil
865 c.mu.Unlock()
866 return renderLifecycleCleanup{
867 tokens: append([]*timerToken{timer, resizeTimer, retryTimer}, workers...),
868 observer: observer,
869 queueCorrelation: queueCorrelation,
870 queueMarked: queueMarked,
871 ackBlocked: ackBlocked,
872 }
873}
874
875// waitForTimerWorkers is terminal phase two. Only the session owner invokes it,
876// after beginSessionTeardown has made future supervisor registration impossible.
877func (c *renderCoordinator) waitForTimerWorkers() { c.supervisor.wait() }
878
879// invalidateRender is the sole producer fan-in. In tests and transitional
880// headless paths without an attached coordinator it retains the old private
881// compositor; attached sessions always schedule through their coordinator.
882func (d *Daemon) invalidateRender(entry *session, ac *attachedClient, reset bool, producer string) {
883 if rc := attachmentRenderCoordinator(entry); rc != nil {
884 invalidation := renderInvalidation{class: invalidateUrgent, reset: reset, producer: producer}
885 if rc.invalidateForAttachment(ac, invalidation) {
886 return
887 }
888 // Additional attachments do not borrow another attachment's lease.
889 // Their shared mutation still schedules the session wake, whose fanout
890 // paints every live attachment with its own output state.
891 if ac != nil && attachmentRegistered(entry, ac) && ac.currentAttachmentSession() == entry {
892 if rc.invalidate(invalidation) {
893 return
894 }
895 d.paint(entry, ac, reset, nil)
896 }
897 return
898 }
899 if ac != nil {
900 d.paint(entry, ac, reset, nil)
901 }
902}
903
904// invalidateRenderNow publishes through the coordinator but immediately
905// flushes the wake when the client can accept state. Attach uses this path so
906// the required first full frame never depends on a debounce timer.
907func (d *Daemon) invalidateRenderNow(entry *session, ac *attachedClient, reset bool, producer string) {
908 if rc := attachmentRenderCoordinator(entry); rc != nil {
909 if rc.invalidateForAttachment(ac, renderInvalidation{class: invalidateUrgent, reset: reset, producer: producer}) {
910 rc.fireCurrent(false)
911 return
912 }
913 if ac != nil && attachmentRegistered(entry, ac) && ac.currentAttachmentSession() == entry {
914 d.paint(entry, ac, reset, nil)
915 }
916 return
917 }
918 if ac != nil {
919 d.paint(entry, ac, reset, nil)
920 }
921}
922
923func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) {
924 c.mu.Lock()
925 if !watchdog && c.normalLane.generation != gen {
926 c.mu.Unlock()
927 return
928 }
929 token := c.normalLane.token
930 c.mu.Unlock()
931 c.fireWithTimerToken(token, gen, watchdog, deadline)
932}
933
934// fireCurrentForLease makes the selected attachment's wake synchronous. The
935// handshake must not finish before that attachment's first output is emitted,
936// while other attachments remain independently asynchronous.
937func (c *renderCoordinator) fireCurrentForLease(lease *attachmentLease) {
938 c.mu.Lock()
939 gen := c.normalLane.generation
940 token := c.normalLane.token
941 c.mu.Unlock()
942 c.fireWithTimerTokenAndLease(token, gen, false, false, lease)
943}
944
945// fireFromTimer consumes a normal deadline using its exact immutable token.
946func (c *renderCoordinator) fireFromTimer(token *timerToken, watchdog, deadline bool) {
947 c.fireWithTimerToken(token, token.generation, watchdog, deadline)
948}
949
950func (c *renderCoordinator) fireWithTimerToken(token *timerToken, gen uint64, watchdog, deadline bool) {
951 c.fireWithTimerTokenAndLease(token, gen, watchdog, deadline, nil)
952}
953
954func (c *renderCoordinator) attachmentAckReady(ac *attachedClient) bool {
955 if c.opts.ackReadyFor != nil {
956 return c.opts.ackReadyFor(ac)
957 }
958 if c.opts.ackReady != nil {
959 return c.opts.ackReady()
960 }
961 return true
962}
963
964func (c *renderCoordinator) fireWithTimerTokenAndLease(token *timerToken, gen uint64, watchdog, deadline bool, syncLease *attachmentLease) {
965 // Readiness probes enter each attachment's send path, which may in turn
966 // read coordinator metadata. Never hold c.mu across those callbacks.
967 c.mu.Lock()
968 if !c.fireValidLocked(token, gen, watchdog) {
969 c.mu.Unlock()
970 return
971 }
972 if c.pendingGeneration == 0 {
973 c.pendingGeneration = 1
974 }
975 leases := make(map[*attachedClient]*attachmentLease, len(c.leases))
976 for ac, lease := range c.leases {
977 if lease != nil && lease.active {
978 leases[ac] = lease
979 }
980 }
981 previewOnly := len(leases) == 0 && len(c.previewWakes) != 0
982 legacyAckReady := c.opts.ackReady
983 c.mu.Unlock()
984
985 ready := make(map[*attachmentLease]bool, len(leases))
986 if len(leases) != 0 {
987 for ac, lease := range leases {
988 // Welcome holds attachment.sendMu while its transport Send is in
989 // flight. Do not probe an incarnation before it is ready.
990 if lease.ready {
991 ready[lease] = c.attachmentAckReady(ac)
992 }
993 }
994 } else if !previewOnly && legacyAckReady != nil {
995 // Keep coordinator-only/headless callers compatible; production always
996 // has attachment leases and uses the per-attachment callback above.
997 ready[nil] = legacyAckReady()
998 } else {
999 ready[nil] = true
1000 }
1001
1002 // The readiness callback can detach, replace, or publish a sync batch.
1003 // syncGateOpenLocked retries predicate evaluation on every registry change
1004 // and returns holding c.mu, so pending consumption has no intervening gap.
1005 c.mu.Lock()
1006 if !c.fireValidLocked(token, gen, watchdog) {
1007 c.mu.Unlock()
1008 return
1009 }
1010 if !c.syncGateOpenLocked() {
1011 c.mu.Unlock()
1012 return
1013 }
1014 if !c.fireValidLocked(token, gen, watchdog) {
1015 c.mu.Unlock()
1016 return
1017 }
1018
1019 // Re-snapshot after every external readiness probe. A replacement may have
1020 // retired one of the leases we probed while the callback was unlocked.
1021 leases = make(map[*attachedClient]*attachmentLease, len(c.leases))
1022 for ac, lease := range c.leases {
1023 if lease != nil && lease.active {
1024 leases[ac] = lease
1025 }
1026 }
1027 var attachmentLeases map[*attachedClient]*attachmentLease
1028 if len(leases) != 0 {
1029 attachmentLeases = make(map[*attachedClient]*attachmentLease, len(leases))
1030 }
1031 blockedByAck := false
1032 allDelivered := true
1033 if len(leases) != 0 {
1034 for ac, lease := range leases {
1035 if !c.pendingShared && len(c.pendingTargets) != 0 {
1036 targetLease, targeted := c.pendingTargets[ac]
1037 if !targeted || targetLease != lease {
1038 continue
1039 }
1040 }
1041 if c.delivered[ac] == c.pendingGeneration {
1042 continue
1043 }
1044 allDelivered = false
1045 if !lease.ready {
1046 continue
1047 }
1048 canReady, probed := ready[lease]
1049 if !probed {
1050 continue
1051 }
1052 if canReady {
1053 attachmentLeases[ac] = lease
1054 if c.ackDeferredFor != nil {
1055 delete(c.ackDeferredFor, ac)
1056 }
1057 continue
1058 }
1059 blockedByAck = true
1060 if c.ackDeferredFor == nil {
1061 c.ackDeferredFor = make(map[*attachedClient]bool)
1062 }
1063 c.ackDeferredFor[ac] = true
1064 }
1065 } else if !ready[nil] {
1066 blockedByAck = true
1067 allDelivered = false
1068 }
1069 legacyOutput := len(leases) == 0 && ready[nil] && (c.pendingShared || len(c.pendingTargets) == 0)
1070 outputReady := len(attachmentLeases) != 0 || legacyOutput
1071 if len(leases) != 0 {
1072 for ac, lease := range leases {
1073 if !c.pendingShared && len(c.pendingTargets) != 0 {
1074 targetLease, targeted := c.pendingTargets[ac]
1075 if !targeted || targetLease != lease {
1076 continue
1077 }
1078 }
1079 if c.delivered[ac] != c.pendingGeneration {
1080 allDelivered = false
1081 break
1082 }
1083 }
1084 }
1085 if len(leases) == 0 && legacyOutput {
1086 allDelivered = true
1087 }
1088 wakeGeneration := uint64(0)
1089 if len(leases) != 0 {
1090 wakeGeneration = c.pendingGeneration
1091 }
1092 w := renderWake{reset: c.pendingReset, urgent: c.pendingUrgent, coalesced: c.coalesced, generation: wakeGeneration, attachmentLeases: attachmentLeases, watchdog: watchdog}
1093 if syncLease != nil {
1094 if selected, ok := attachmentLeases[syncLease.attachment]; ok && selected == syncLease {
1095 w.lease = syncLease
1096 }
1097 }
1098 previews := c.takePendingPreviewsLocked()
1099 worker := c.detachNormalTimerLocked()
1100 c.armed = false
1101
1102 var ackStart *ackBlockedSpan
1103 if blockedByAck && deadline && c.ackBlocked == nil && c.opts.observer != nil {
1104 // Install the span while coordinator state is owned. Its Start
1105 // publication runs unlocked, and its own handoff defers End until that
1106 // call has returned.
1107 ackStart = newACKBlockedSpan(c.opts.observer)
1108 c.ackBlocked = ackStart
1109 }
1110 c.ackDeferred = len(c.ackDeferredFor) != 0 || (len(leases) == 0 && blockedByAck && deadline)
1111 if !outputReady && !allDelivered {
1112 c.mu.Unlock()
1113 stopDetachedTimer(worker)
1114 if ackStart != nil {
1115 ackStart.publishStart()
1116 }
1117 c.notifyPreviews(w, previews)
1118 return
1119 }
1120
1121 if outputReady {
1122 for ac, lease := range attachmentLeases {
1123 if c.delivered == nil {
1124 c.delivered = make(map[*attachedClient]uint64)
1125 }
1126 c.delivered[ac] = c.pendingGeneration
1127 lease.deliveredGeneration = c.pendingGeneration
1128 }
1129 }
1130 if len(leases) != 0 {
1131 allDelivered = true
1132 for ac, lease := range leases {
1133 if !c.pendingShared && len(c.pendingTargets) != 0 {
1134 targetLease, targeted := c.pendingTargets[ac]
1135 if !targeted || targetLease != lease {
1136 continue
1137 }
1138 }
1139 if c.delivered[ac] != c.pendingGeneration {
1140 allDelivered = false
1141 break
1142 }
1143 }
1144 }
1145 complete := allDelivered || (len(leases) == 0 && outputReady)
1146 if complete {
1147 c.pending, c.pendingReset, c.pendingUrgent, c.ackDeferred, c.deadlineDue, c.coalesced, c.armed = false, false, false, false, false, 0, false
1148 c.ackDeferredFor = nil
1149 c.delivered = nil
1150 c.pendingShared = false
1151 c.pendingTargets = nil
1152 }
1153 if outputReady {
1154 c.metrics.wakes.Add(1)
1155 c.metrics.coalesced.Add(uint64(w.coalesced))
1156 if !w.urgent {
1157 if w.coalesced > 1 {
1158 c.outputPressure += w.coalesced - 1
1159 if c.outputPressure > int((maxOutputRenderDeadline-minOutputRenderDeadline)/time.Millisecond) {
1160 c.outputPressure = int((maxOutputRenderDeadline - minOutputRenderDeadline) / time.Millisecond)
1161 }
1162 } else if c.outputPressure > 0 {
1163 c.outputPressure--
1164 }
1165 }
1166 }
1167 observer, queueCorrelation, queueMarked := c.opts.observer, c.queueCorrelation, c.queueMarked
1168 ackBlocked := c.ackBlocked
1169 if complete {
1170 c.queueMarked, c.ackBlocked = false, nil
1171 }
1172 wake := c.opts.wake
1173 c.mu.Unlock()
1174 stopDetachedTimer(worker)
1175 if ackStart != nil {
1176 ackStart.publishStart()
1177 }
1178 if complete && queueMarked && observer != nil {
1179 observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", queueCorrelation, ports.RuntimeQueueDequeued, 0, true))
1180 }
1181 if complete && ackBlocked != nil {
1182 ackBlocked.finish(true)
1183 }
1184 if outputReady && wake != nil {
1185 wake(w)
1186 }
1187 c.notifyPreviews(w, previews)
1188}
1189
1190func (c *renderCoordinator) fireValidLocked(token *timerToken, gen uint64, watchdog bool) bool {
1191 if c.torndown || !c.pending || watchdog {
1192 return !c.torndown && c.pending
1193 }
1194 return c.normalLane.generation == gen && (token == nil || c.normalLane.token == token)
1195}
1196
1197// syncGateOpen evaluates pane visibility without c.mu, then reacquires and
1198// validates the complete batch identity set before accepting that result.
1199// syncGateOpen evaluates visibility outside c.mu and returns with c.mu held
1200// only when the registry version still matches the snapshot. The caller must
1201// consume pending state before unlocking, eliminating the gate-to-consume gap.
1202func (c *renderCoordinator) syncGateOpenLocked() bool {
1203 for {
1204 version := c.syncRegistryVersion
1205 batches := make([]*syncBatch, 0, len(c.syncBatches))
1206 for _, batch := range c.syncBatches {
1207 batches = append(batches, batch)
1208 }
1209 c.mu.Unlock()
1210
1211 gated := false
1212 for _, batch := range batches {
1213 if batch.renderable == nil || batch.renderable() {
1214 gated = true
1215 break
1216 }
1217 }
1218 if hook := c.opts.afterSyncGateEvaluated; hook != nil {
1219 hook()
1220 }
1221
1222 c.mu.Lock()
1223 if version == c.syncRegistryVersion {
1224 return !gated
1225 }
1226 }
1227}
1228
1229// takePendingPreviewsLocked snapshots and consumes picker preview delivery
1230// for this target generation. Preview callbacks are never ACK-gated.
1231func (c *renderCoordinator) takePendingPreviewsLocked() []func(renderWake) {
1232 if !c.pendingPreview {
1233 return nil
1234 }
1235 c.pendingPreview = false
1236 previews := make([]func(renderWake), 0, len(c.previewWakes))
1237 for _, subscription := range c.previewWakes {
1238 previews = append(previews, subscription.fn)
1239 }
1240 return previews
1241}
1242
1243func (c *renderCoordinator) notifyPreviews(w renderWake, previews []func(renderWake)) {
1244 for _, fn := range previews {
1245 fn(w)
1246 }
1247}