1package app
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "os"
8 "path/filepath"
9 "runtime"
10 "runtime/debug"
11 "sort"
12 "sync"
13 "sync/atomic"
14 "time"
15
16 "github.com/bnema/neferwl/internal/adapters/clock"
17 "github.com/bnema/neferwl/internal/adapters/config"
18 "github.com/bnema/neferwl/internal/adapters/drm"
19 "github.com/bnema/neferwl/internal/adapters/logging"
20 "github.com/bnema/neferwl/internal/adapters/sched"
21 "github.com/bnema/neferwl/internal/adapters/seat"
22 "github.com/bnema/neferwl/internal/adapters/watchdog"
23 "github.com/bnema/neferwl/internal/ports"
24 "github.com/bnema/zerowrap"
25)
26
27var errTimeout = errors.New("timeout")
28
29// securityLeaseCard keeps the native operation and its tracked lease inventory
30// together. A failed revoke must remain in LeaseIDs for the next barrier attempt.
31type securityLeaseCard interface {
32 Path() string
33 Lease([]string) (*os.File, uint32, error)
34 LeaseIDs() []uint32
35 Revoke(uint32) error
36}
37
38func securitySnapshot(gate ports.SessionSecurity) ports.SecurityState {
39 if gate == nil {
40 return ports.SecurityState{}
41 }
42 return gate.Snapshot()
43}
44
45// guardedLease denies requests pending at acquisition and closes any FD created
46// across a transition, even when the native call succeeds. Failed revocations
47// stay tracked by the card and prevent the later backend barrier.
48func guardedLease(c securityLeaseCard, gate ports.SessionSecurity, connectors []string) (*os.File, uint32, error) {
49 before := securitySnapshot(gate)
50 if before.Protected {
51 return nil, 0, errors.New("session protected")
52 }
53 fd, id, err := c.Lease(connectors)
54 if err != nil {
55 return fd, id, err
56 }
57 after := securitySnapshot(gate)
58 if after.Protected || after.Generation != before.Generation {
59 if fd != nil {
60 fd.Close()
61 }
62 return nil, 0, errors.Join(errors.New("session transition during lease"), c.Revoke(id))
63 }
64 return fd, id, nil
65}
66
67func revokeSecurityLeases(c securityLeaseCard, finished func(uint32)) error {
68 var errs []error
69 for _, id := range c.LeaseIDs() {
70 if err := c.Revoke(id); err != nil {
71 errs = append(errs, fmt.Errorf("%s lease %d: %w", c.Path(), id, err))
72 } else {
73 finished(id)
74 }
75 }
76 if ids := c.LeaseIDs(); len(ids) != 0 {
77 errs = append(errs, fmt.Errorf("%s leases still active: %v", c.Path(), ids))
78 }
79 return errors.Join(errs...)
80}
81
82// registerDiscardedOutput accounts for physical output lifetimes that cannot
83// enter connector-name routing. They still require protection just like owners.
84// The same backend-owned counter is used by outputSet.start; no instance reuse.
85func registerDiscardedOutput(set *outputSet, name string) (ports.OutputInstance, error) {
86 if set.nextInstance == ^ports.OutputInstance(0) {
87 set.startErr = errors.New("output instance exhausted")
88 return 0, set.startErr
89 }
90 set.nextInstance++
91 instance := set.nextInstance
92 if !set.securityEvent(ports.SecurityOutputAdded{Instance: instance, Output: name}) {
93 return instance, set.ctx.Err()
94 }
95 return instance, nil
96}
97
98// retireDiscardedOutput consumes only affirmative terminal evidence. A failed
99// close remains in authoritative inventory, even though it has no render owner.
100func retireDiscardedOutput(set *outputSet, instance ports.OutputInstance, inactive bool) error {
101 if !inactive {
102 return fmt.Errorf("discarded output %d shutdown inactivity unconfirmed", instance)
103 }
104 if !set.securityEvent(ports.SecurityOutputRemoved{Instance: instance}) {
105 return set.ctx.Err()
106 }
107 return nil
108}
109
110// applyTimeout bounds how long outputs take to apply a configuration.
111const applyTimeout = 5 * time.Second
112
113// drmBackend is the seat and the cards with at least one usable output.
114type drmBackend struct {
115 seat *seat.Seat
116 cards []*drmCard
117 clock ports.Clock
118 log zerowrap.Logger
119 // preparers are the running outputs, guarded by mu: the seat's goroutine
120 // reads them while the output loop adds and removes them.
121 mu sync.Mutex
122 preparers map[seatDisablePreparer]struct{}
123 // probes are the running output loops the watchdog pings, by
124 // connector, guarded by mu.
125 probes map[string]func()
126}
127
128func (b *drmBackend) addProbe(name string, ping func()) {
129 b.mu.Lock()
130 defer b.mu.Unlock()
131 if b.probes == nil {
132 b.probes = map[string]func(){}
133 }
134 b.probes[name] = ping
135}
136
137func (b *drmBackend) removeProbe(name string) {
138 b.mu.Lock()
139 defer b.mu.Unlock()
140 delete(b.probes, name)
141}
142
143// watchdogProbes lists the running output loops for the watchdog.
144func (b *drmBackend) watchdogProbes() []watchdog.Probe {
145 b.mu.Lock()
146 defer b.mu.Unlock()
147 out := make([]watchdog.Probe, 0, len(b.probes))
148 for name, ping := range b.probes {
149 out = append(out, watchdog.Probe{Name: "output-" + name, Ping: ping})
150 }
151 return out
152}
153
154// seatDisablePreparer is an output that can reset its planes before the
155// seat is disabled (drm.Output).
156type seatDisablePreparer interface {
157 PrepareSeatDisable(timeout time.Duration) bool
158}
159
160// seatDisableTimeout bounds how long a seat disable waits for the outputs:
161// the VT switch must not hang on a stuck one.
162const seatDisableTimeout = 100 * time.Millisecond
163
164func (b *drmBackend) addPreparer(p seatDisablePreparer) {
165 b.mu.Lock()
166 defer b.mu.Unlock()
167 if b.preparers == nil {
168 b.preparers = map[seatDisablePreparer]struct{}{}
169 }
170 b.preparers[p] = struct{}{}
171}
172
173func (b *drmBackend) removePreparer(p seatDisablePreparer) {
174 b.mu.Lock()
175 defer b.mu.Unlock()
176 delete(b.preparers, p)
177}
178
179// prepareSeatDisable runs on the seat's goroutine before the seat is
180// disabled and its devices revoked: every running output commits what it
181// must leave behind (planes on Bypass), all in parallel under one bound. A
182// timeout is logged and the disable goes on.
183func (b *drmBackend) prepareSeatDisable() {
184 b.mu.Lock()
185 outs := make([]seatDisablePreparer, 0, len(b.preparers))
186 for p := range b.preparers {
187 outs = append(outs, p)
188 }
189 b.mu.Unlock()
190 var wg sync.WaitGroup
191 var late atomic.Int32
192 for _, p := range outs {
193 wg.Go(func() {
194 if !p.PrepareSeatDisable(seatDisableTimeout) {
195 late.Add(1)
196 }
197 })
198 }
199 wg.Wait()
200 if n := late.Load(); n > 0 {
201 b.log.Warn().Int("outputs", int(n)).Dur("timeout", seatDisableTimeout).Msg("outputs did not prepare for the seat disable in time; disabling anyway")
202 }
203}
204
205type drmCard struct {
206 *drm.Card
207 fd int
208}
209
210// A failed card read must not erase its heads from a multi-card inventory.
211func retainedHeads(previous, found []ports.OutputHead, err error) []ports.OutputHead {
212 if err != nil {
213 return previous
214 }
215 return found
216}
217
218// A stopped instance is no longer enabled, even if the inventory read fails.
219func releasedHead(heads []ports.OutputHead, name string) []ports.OutputHead {
220 heads = append([]ports.OutputHead(nil), heads...)
221 for i := range heads {
222 if heads[i].Info.Name == name {
223 heads[i].Enabled = false
224 heads[i].Current = nil
225 }
226 }
227 return heads
228}
229
230// safe turns a panic in a hardware goroutine into an error, so the TTY is still restored.
231func safe(name string, fn func() error) (err error) {
232 defer func() {
233 if p := recover(); p != nil {
234 err = fmt.Errorf("%s panic: %v\n%s", name, p, debug.Stack())
235 }
236 }()
237 return fn()
238}
239
240// openDRM opens the seat and every card with a connected display. Outputs
241// are found by the first Scan of each card.
242func openDRM(ctx context.Context, cfg ports.Config) (*drmBackend, error) {
243 log := logging.For(ctx, "drm")
244 s, err := seat.Open(ctx, logging.For(ctx, "seat"))
245 if err != nil {
246 log.Error().Err(err).Msg("open seat")
247 return nil, fmt.Errorf("open seat: %w", err)
248 }
249 want := wantFromConfig(cfg)
250 want.TraceFlips = logging.Enabled(ctx, "drm-flip")
251 paths, _ := filepath.Glob("/dev/dri/card[0-9]*")
252 sort.Strings(paths)
253 b := &drmBackend{seat: s, clock: clock.System{}, log: log}
254 s.SetBeforeDisable(b.prepareSeatDisable)
255 var errs []error
256 for _, path := range paths {
257 fd, err := s.OpenDevice(path)
258 if err != nil {
259 errs = append(errs, err)
260 continue
261 }
262 card, err := drm.OpenCard(fd, path, want, log, logging.For(ctx, "render"))
263 if err != nil {
264 log.Warn().Err(err).Str("card", path).Msg("card unusable")
265 errs = append(errs, err)
266 s.CloseDevice(fd)
267 continue
268 }
269 b.cards = append(b.cards, &drmCard{Card: card, fd: fd})
270 }
271 if len(b.cards) == 0 {
272 s.Close()
273 err = fmt.Errorf("no usable DRM card in %v: %w", paths, errors.Join(errs...))
274 log.Error().Err(err).Msg("open drm")
275 return nil, err
276 }
277 return b, nil
278}
279
280// wantFromConfig turns output.<name> and render.* entries into connector choices.
281func wantFromConfig(cfg ports.Config) drm.Want {
282 r := cfg.Render
283 want := drm.Want{Disabled: map[string]bool{}, Modes: map[string][3]float64{}, HDR: map[string]drm.HDRSettings{}, NoScanout: !r.DirectScanout, NoTearing: !r.Tearing, NoVRR: !r.VRR, VRRFlipGap: r.VRRFlipGap}
284 for _, o := range cfg.Outputs {
285 nits := o.SDRBrightness
286 if nits == 0 {
287 nits = ports.DefaultSDRBrightness
288 }
289 want.HDR[o.Name] = drm.HDRSettings{Enabled: o.HDR, SDRBrightness: nits}
290 if o.Off {
291 want.Disabled[o.Name] = true
292 continue
293 }
294 if o.Mode != "" {
295 // Validated at config load.
296 w, h, hz, _ := config.ParseMode(o.Mode)
297 want.Modes[o.Name] = [3]float64{float64(w), float64(h), hz}
298 }
299 }
300 return want
301}
302
303// close releases the cards and the seat so the TTY comes back. Outputs are
304// closed by their goroutines first.
305func (b *drmBackend) close() {
306 for _, c := range b.cards {
307 b.seat.CloseDevice(c.fd)
308 }
309 b.seat.Close()
310}
311
312// runOutputs drives every output of every card: one goroutine per output,
313// one flip reader per card, and a udev watcher that rescans connectors on
314// hotplug. Core learns about outputs through events. The loop also owns
315// output configuration: reloads and protocol requests go through apply, and
316// each configuration is finished when its outputs are ready, fail or time
317// out. It returns when ctx ends, after every output is closed.
318func (b *drmBackend) runOutputs(ctx context.Context, want func(ports.Config) drm.Want, initial ports.Config, apply *outputApply, ch outputChannels, curs *cursors, newRenderer func(w, h int) (ports.Renderer, error), log zerowrap.Logger) error {
319 ctx, cancel := context.WithCancel(ctx)
320 defer cancel()
321 var readers sync.WaitGroup
322 readerErr := make(chan error, len(b.cards))
323 for _, c := range b.cards {
324 c.SetFormats(ch.formats)
325 readers.Go(func() {
326 if err := safe("drm events", func() error { return c.ReadEvents(ctx) }); err != nil {
327 readerErr <- fmt.Errorf("%s: %w", c.Path(), err)
328 }
329 })
330 }
331 defer readers.Wait()
332 active := b.seat.Subscribe()
333 defer b.seat.Unsubscribe(active)
334 seatActive := true // openDRM waits for the first seat enable
335 set := newOutputSet(ctx, ch.captured)
336 set.attach = ch.content
337 set.wireSecurity(ch)
338 protected := func() bool { return set.state.Protected || securitySnapshot(ch.security).Protected }
339 leases := newLeasePublisher(ctx, ch.leaseEvents, func() bool { return seatActive }, protected, log)
340 defer leases.close()
341 hotplug := make(chan struct{}, 1)
342 go func() {
343 if err := safe("hotplug", func() error { return drm.WatchHotplug(ctx, hotplug) }); err != nil {
344 log.Warn().Err(err).Msg("hotplug watch disabled")
345 }
346 }()
347 cards := map[string]*drmCard{}
348 outputs := map[string]*drm.Output{} // owner result, read only after done
349 send := func(ev ports.OutputEvent) {
350 select {
351 case ch.events <- ev:
352 case <-ctx.Done():
353 }
354 }
355 // stopping holds outputs asked to stop: true when they restart with a
356 // new mode (core keeps the screen), false when they are gone.
357 stopping := map[string]bool{}
358 currentConfig := initial
359 var currentHeads ports.OutputHeads
360 lastHeads := map[*drmCard][]ports.OutputHead{}
361 progress := &applyProgress{completed: map[<-chan error]bool{}}
362 readySources := map[string]<-chan error{}
363 stopReady := map[string]chan struct{}{}
364 type readyResult struct {
365 name string
366 source <-chan error
367 err error
368 }
369 ready := make(chan readyResult)
370 // deadline is the pending apply's timer channel, nil while none is pending.
371 timer := b.clock.NewTimer(applyTimeout)
372 timer.Stop()
373 defer timer.Stop()
374 var deadline <-chan time.Time
375 // A small bounded retry budget avoids a busy loop on failed KMS revokes.
376 revokeTimer := b.clock.NewTimer(time.Second)
377 revokeTimer.Stop()
378 defer revokeTimer.Stop()
379 var revokeDeadline <-chan time.Time
380 var revokeAttempts int
381 var discardedOutputErr error // first unconfirmed lifetime; inventory retains every instance
382 barrier := func() {
383 if !set.state.Protected {
384 return
385 }
386 var errs []error
387 for _, c := range b.cards {
388 errs = append(errs, revokeSecurityLeases(c, func(id uint32) {
389 leases.send(ports.LeaseFinished{Card: c.Path(), LeaseID: id})
390 }))
391 leases.publish(c)
392 }
393 errs = append(errs, discardedOutputErr)
394 err := errors.Join(errs...)
395 // Registration runs synchronously in scan before any barrier. A gate
396 // transition racing native revocation belongs to a later round.
397 if securitySnapshot(ch.security).Generation > set.state.Generation {
398 return
399 }
400 set.securityEvent(ports.SecurityBackendBarrier{Generation: set.state.Generation, Err: err})
401 revokeAttempts++
402 if err != nil && revokeAttempts < 3 {
403 revokeTimer.Reset(time.Second)
404 revokeDeadline = revokeTimer.C()
405 }
406 }
407 var op uint64
408 complete := func(d applyDecision) {
409 if !d.reply {
410 return
411 }
412 timer.Stop()
413 deadline = nil
414 apply.finished(d.err)
415 }
416 publishInventory := func() {
417 var inventory ports.OutputHeads
418 for _, card := range b.cards {
419 found, err := card.ConnectedHeads()
420 if err != nil {
421 log.Warn().Err(err).Str("card", card.Path()).Msg("inventory outputs")
422 }
423 lastHeads[card] = retainedHeads(lastHeads[card], found, err)
424 inventory.Heads = append(inventory.Heads, lastHeads[card]...)
425 }
426 currentHeads = inventory
427 apply.heads(inventory)
428 }
429 scan := func() {
430 var scanErrors []error
431 for _, c := range b.cards {
432 w := want(currentConfig)
433 w.Device = c.Device()
434 c.SetWant(w)
435 added, removed, replaced, err := c.Scan()
436 if err != nil {
437 log.Warn().Err(err).Str("card", c.Path()).Msg("scan connectors")
438 scanErrors = append(scanErrors, fmt.Errorf("scan %s: %w", c.Path(), err))
439 continue
440 }
441 leases.publish(c)
442 stop := func(name string, restart bool) {
443 if r := set.outs[name]; r != nil && cards[name] == c {
444 if _, ok := stopping[name]; !ok {
445 stopping[name] = restart
446 progress.stopping(name)
447 r.stop()
448 }
449 }
450 }
451 for _, name := range removed {
452 log.Info().Str("connector", name).Msg("output unplugged")
453 stop(name, false)
454 }
455 for _, name := range replaced {
456 log.Info().Str("connector", name).Msg("output mode change")
457 stop(name, true)
458 }
459 for _, o := range added {
460 // Every physical lifetime gets the gate before any early exit:
461 // Close must not restore saved desktop, even on a name collision.
462 o.Security, o.Frames = ch.security, ch.frames
463 name := o.Info().Name
464 if other := cards[name]; other != nil && other != c {
465 // Core and clients know outputs by connector name.
466 log.Warn().Str("connector", name).Str("card", c.Path()).Msg("connector name already used by another card; ignored")
467 instance, registerErr := registerDiscardedOutput(set, name)
468 o.Instance = instance
469 o.Close() // synchronous final KMS disable; never starts a renderer
470 var shutdownErr error
471 if registerErr == nil {
472 shutdownErr = retireDiscardedOutput(set, instance, o.InactiveOnClose())
473 }
474 if err := errors.Join(registerErr, shutdownErr); err != nil {
475 if discardedOutputErr == nil {
476 discardedOutputErr = err
477 }
478 scanErrors = append(scanErrors, err)
479 log.Error().Err(err).Str("connector", name).Str("card", c.Path()).Msg("discarded output protection unconfirmed")
480 // An exhausted instance cannot represent an unsafe lifetime.
481 // Stop the backend rather than ever acknowledge this inventory.
482 if instance == 0 {
483 cancel()
484 }
485 }
486 c.Release(name)
487 continue
488 }
489 cards[name] = c
490 outputs[name] = o
491 readySources[name] = o.Ready()
492 if progress.active && !want(currentConfig).Disabled[name] {
493 progress.required[name] = true
494 }
495 source := readySources[name]
496 stopped := make(chan struct{})
497 stopReady[name] = stopped
498 go func() {
499 select {
500 case err := <-source:
501 select {
502 case ready <- readyResult{name, source, err}:
503 case <-ctx.Done():
504 }
505 case <-stopped:
506 case <-ctx.Done():
507 }
508 }()
509 o.NewCaptureRenderer = newRenderer
510 o.StartOff = set.off[name]
511 progress.started(name, source)
512 if cur := o.Cursor(); cur != nil {
513 curs.set(name, cur)
514 }
515 active := b.seat.Subscribe()
516 realtime := currentConfig.Performance.Realtime
517 started := set.start(ctx, name, func(octx context.Context, sc <-chan ports.Scene, cc <-chan *ports.ContentBatch, rc ports.ContentReturn, cu <-chan ports.CursorChange, cap <-chan ports.CaptureRequest, secure <-chan ports.SecurityState, instance ports.OutputInstance) error {
518 defer b.seat.Unsubscribe(active)
519 o.Security, o.SecurityChanges, o.Instance = ch.security, secure, instance
520 return safe("output "+name, func() error {
521 defer o.Close()
522 b.addPreparer(o)
523 defer b.removePreparer(o) // before Close: it handles no request any more
524 // octx derives from the session: the probe ends with this
525 // output or the session.
526 b.addProbe(name, func() { o.Ping(octx) })
527 defer b.removeProbe(name)
528 if realtime {
529 runtime.LockOSThread()
530 // Keep the thread locked until this goroutine exits: an RT
531 // thread must not return to Go's general-purpose pool.
532 schedLog := logging.For(ctx, "sched")
533 if err := sched.Realtime(schedLog); err != nil {
534 schedLog.Warn().Err(err).Msg("output scheduling")
535 }
536 }
537 return o.Run(octx, newRenderer, loadCursor, active, sc, cc, rc, cu, ch.presented, cap, ch.captured)
538 })
539 }, &o.SecurityEvents)
540 if !started {
541 b.seat.Unsubscribe(active)
542 o.Close()
543 c.Release(name)
544 continue
545 }
546 send(ports.OutputAdded{Info: o.Info()})
547 }
548 }
549 publishInventory()
550 complete(progress.scanError(errors.Join(scanErrors...)))
551 barrier()
552 }
553 scan()
554 if len(set.outs) == 0 {
555 return errors.Join(errors.New("no connected display"), set.wait())
556 }
557 // configure starts cfg; complete reports its result to apply. apply
558 // hands out one configuration at a time, and a rollback or reload starts
559 // on a later iteration, never inside scan or another configure.
560 configure := func(cfg ports.Config) {
561 currentConfig = cfg
562 op++
563 timer.Stop()
564 deadline = nil
565 required := map[string]bool{}
566 // Desired enabled heads, not the inventory's current mode, require a first modeset.
567 w := want(currentConfig)
568 for _, h := range currentHeads.Heads {
569 if !w.Disabled[h.Info.Name] {
570 required[h.Info.Name] = true
571 }
572 }
573 for name := range set.outs {
574 if !w.Disabled[name] {
575 required[name] = true
576 }
577 }
578 progress.start(op, required, stopping)
579 scan()
580 running := map[string]<-chan error{}
581 for name := range set.outs {
582 running[name] = readySources[name]
583 }
584 d := progress.scanned(running)
585 if progress.active {
586 timer.Reset(applyTimeout)
587 deadline = timer.C()
588 }
589 complete(d)
590 }
591 for {
592 if cfg, ok := apply.next(); ok {
593 configure(cfg)
594 continue
595 }
596 configs, configNext := apply.configOut(ch.configured), apply.config
597 replies, replyNext := apply.replyOut(ch.replies)
598 heads, headsNext := apply.headsOut(ch.heads)
599 select {
600 case <-ctx.Done():
601 return set.wait()
602 case err := <-readerErr:
603 // No more flips on that card: its outputs would freeze.
604 return errors.Join(err, set.wait())
605 case state, ok := <-ch.securityChanges:
606 if !ok {
607 ch.securityChanges = nil
608 continue
609 }
610 if state.Generation < set.state.Generation {
611 continue
612 }
613 if set.setSecurity(state) {
614 revokeAttempts = 0
615 }
616 // start can refresh the gate while scanning; its queued transition
617 // must still receive a barrier even if forwarding already occurred.
618 revokeTimer.Stop()
619 revokeDeadline = nil
620 if set.state.Protected {
621 barrier()
622 } else {
623 for _, c := range b.cards {
624 leases.publish(c)
625 }
626 }
627 case <-revokeDeadline:
628 revokeDeadline = nil
629 barrier()
630 case <-hotplug:
631 log.Info().Msg("hotplug")
632 scan()
633 if progress.active {
634 running := map[string]<-chan error{}
635 for name := range set.outs {
636 running[name] = readySources[name]
637 }
638 complete(progress.scanned(running))
639 }
640 case enabled := <-active:
641 seatActive = enabled
642 if !enabled {
643 for _, c := range b.cards {
644 for _, id := range c.LeaseIDs() {
645 if err := c.Revoke(id); err != nil {
646 log.Warn().Err(err).Uint32("lease", id).Msg("VT lease revoke")
647 } else {
648 leases.send(ports.LeaseFinished{Card: c.Path(), LeaseID: id})
649 }
650 }
651 leases.publish(c)
652 }
653 } else {
654 for _, c := range b.cards {
655 leases.publish(c)
656 }
657 }
658 case event := <-ch.leaseRequests:
659 switch req := event.(type) {
660 case ports.LeaseRequest:
661 reply := ports.LeaseReply{ID: req.ID, Err: fmt.Errorf("unknown card %s", req.Card)}
662 if !seatActive || protected() {
663 reply.Err = errors.New("DRM seat inactive or session protected")
664 leases.send(reply)
665 break
666 }
667 for _, c := range b.cards {
668 if c.Path() == req.Card {
669 reply.FD, reply.LeaseID, reply.Err = guardedLease(c, ch.security, req.Connectors)
670 leases.publish(c)
671 break
672 }
673 }
674 leases.send(reply)
675 case ports.LeaseRevoke:
676 for _, c := range b.cards {
677 if c.Path() == req.Card {
678 if err := c.Revoke(req.LeaseID); err != nil {
679 log.Warn().Err(err).Uint32("lease", req.LeaseID).Msg("revoke lease")
680 } else {
681 leases.publish(c)
682 }
683 break
684 }
685 }
686 }
687 case ev := <-ch.reloads:
688 apply.reload(ev.Config)
689 case req := <-ch.requests:
690 apply.request(req)
691 case configs <- configNext:
692 apply.configSent()
693 case replies <- replyNext:
694 apply.replySent()
695 case heads <- headsNext:
696 apply.headsSent()
697 case result := <-ready:
698 complete(progress.readyEvent(result.name, result.source, readySources[result.name], result.err))
699 case <-deadline:
700 complete(progress.timeout(op))
701 case s := <-ch.scenes:
702 set.scenes(s)
703 case q := <-ch.captures:
704 set.routeCapture(q)
705 case c := <-ch.cursorChanges:
706 set.setCursor(c)
707 case stopped := <-set.stopped:
708 if !set.stoppedCurrent(stopped) {
709 continue
710 }
711 name := stopped.name
712 output := outputs[name]
713 err := set.finish(name)
714 // finish observes done after the deferred Close. Only affirmative
715 // terminal KMS inactivity retires this exact lifetime; renderer
716 // errors and missing connector inventory are not physical proof.
717 if output != nil && output.InactiveOnClose() {
718 set.securityEvent(ports.SecurityOutputRemoved{Instance: stopped.instance})
719 }
720 delete(outputs, name)
721 if ctx.Err() != nil {
722 return errors.Join(err, set.wait())
723 }
724 restart := stopping[name]
725 delete(stopping, name)
726 curs.set(name, nil)
727 card := cards[name]
728 card.Release(name)
729 lastHeads[card] = releasedHead(lastHeads[card], name)
730 delete(cards, name)
731 source := readySources[name]
732 delete(readySources, name)
733 if stopped := stopReady[name]; stopped != nil {
734 close(stopped)
735 delete(stopReady, name)
736 }
737 stopDecision := progress.stopped(name, source, restart, err)
738 if err != nil {
739 // A broken output (e.g. its renderer) must not take the
740 // session down. It stays off until the next hotplug, so a
741 // lasting failure does not loop.
742 log.Error().Err(err).Str("connector", name).Msg("output stopped")
743 restart = false
744 }
745 // Release changes the enabled state in ConnectedHeads. Do not
746 // restart a failed or disabled output just to refresh inventory.
747 if restart {
748 scan()
749 } else {
750 publishInventory()
751 }
752 complete(stopDecision)
753 if restart {
754 // Same connector, new mode: core keeps its screen and
755 // workspaces; OutputAdded updates the size.
756 if set.outs[name] != nil {
757 running := map[string]<-chan error{}
758 for n := range set.outs {
759 running[n] = readySources[n]
760 }
761 complete(progress.scanned(running))
762 continue
763 }
764 }
765 send(ports.OutputRemoved{Name: name})
766 running := map[string]<-chan error{}
767 for n := range set.outs {
768 running[n] = readySources[n]
769 }
770 complete(progress.scanned(running))
771 }
772 }
773}