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.wireSecurity(ch)
337 protected := func() bool { return set.state.Protected || securitySnapshot(ch.security).Protected }
338 leases := newLeasePublisher(ctx, ch.leaseEvents, func() bool { return seatActive }, protected, log)
339 defer leases.close()
340 hotplug := make(chan struct{}, 1)
341 go func() {
342 if err := safe("hotplug", func() error { return drm.WatchHotplug(ctx, hotplug) }); err != nil {
343 log.Warn().Err(err).Msg("hotplug watch disabled")
344 }
345 }()
346 cards := map[string]*drmCard{}
347 outputs := map[string]*drm.Output{} // owner result, read only after done
348 send := func(ev ports.OutputEvent) {
349 select {
350 case ch.events <- ev:
351 case <-ctx.Done():
352 }
353 }
354 // stopping holds outputs asked to stop: true when they restart with a
355 // new mode (core keeps the screen), false when they are gone.
356 stopping := map[string]bool{}
357 currentConfig := initial
358 var currentHeads ports.OutputHeads
359 lastHeads := map[*drmCard][]ports.OutputHead{}
360 progress := &applyProgress{completed: map[<-chan error]bool{}}
361 readySources := map[string]<-chan error{}
362 stopReady := map[string]chan struct{}{}
363 type readyResult struct {
364 name string
365 source <-chan error
366 err error
367 }
368 ready := make(chan readyResult)
369 // deadline is the pending apply's timer channel, nil while none is pending.
370 timer := b.clock.NewTimer(applyTimeout)
371 timer.Stop()
372 defer timer.Stop()
373 var deadline <-chan time.Time
374 // A small bounded retry budget avoids a busy loop on failed KMS revokes.
375 revokeTimer := b.clock.NewTimer(time.Second)
376 revokeTimer.Stop()
377 defer revokeTimer.Stop()
378 var revokeDeadline <-chan time.Time
379 var revokeAttempts int
380 var discardedOutputErr error // first unconfirmed lifetime; inventory retains every instance
381 barrier := func() {
382 if !set.state.Protected {
383 return
384 }
385 var errs []error
386 for _, c := range b.cards {
387 errs = append(errs, revokeSecurityLeases(c, func(id uint32) {
388 leases.send(ports.LeaseFinished{Card: c.Path(), LeaseID: id})
389 }))
390 leases.publish(c)
391 }
392 errs = append(errs, discardedOutputErr)
393 err := errors.Join(errs...)
394 // Registration runs synchronously in scan before any barrier. A gate
395 // transition racing native revocation belongs to a later round.
396 if securitySnapshot(ch.security).Generation > set.state.Generation {
397 return
398 }
399 set.securityEvent(ports.SecurityBackendBarrier{Generation: set.state.Generation, Err: err})
400 revokeAttempts++
401 if err != nil && revokeAttempts < 3 {
402 revokeTimer.Reset(time.Second)
403 revokeDeadline = revokeTimer.C()
404 }
405 }
406 var op uint64
407 complete := func(d applyDecision) {
408 if !d.reply {
409 return
410 }
411 timer.Stop()
412 deadline = nil
413 apply.finished(d.err)
414 }
415 publishInventory := func() {
416 var inventory ports.OutputHeads
417 for _, card := range b.cards {
418 found, err := card.ConnectedHeads()
419 if err != nil {
420 log.Warn().Err(err).Str("card", card.Path()).Msg("inventory outputs")
421 }
422 lastHeads[card] = retainedHeads(lastHeads[card], found, err)
423 inventory.Heads = append(inventory.Heads, lastHeads[card]...)
424 }
425 currentHeads = inventory
426 apply.heads(inventory)
427 }
428 scan := func() {
429 var scanErrors []error
430 for _, c := range b.cards {
431 w := want(currentConfig)
432 w.Device = c.Device()
433 c.SetWant(w)
434 added, removed, replaced, err := c.Scan()
435 if err != nil {
436 log.Warn().Err(err).Str("card", c.Path()).Msg("scan connectors")
437 scanErrors = append(scanErrors, fmt.Errorf("scan %s: %w", c.Path(), err))
438 continue
439 }
440 leases.publish(c)
441 stop := func(name string, restart bool) {
442 if r := set.outs[name]; r != nil && cards[name] == c {
443 if _, ok := stopping[name]; !ok {
444 stopping[name] = restart
445 progress.stopping(name)
446 r.stop()
447 }
448 }
449 }
450 for _, name := range removed {
451 log.Info().Str("connector", name).Msg("output unplugged")
452 stop(name, false)
453 }
454 for _, name := range replaced {
455 log.Info().Str("connector", name).Msg("output mode change")
456 stop(name, true)
457 }
458 for _, o := range added {
459 // Every physical lifetime gets the gate before any early exit:
460 // Close must not restore saved desktop, even on a name collision.
461 o.Security = ch.security
462 name := o.Info().Name
463 if other := cards[name]; other != nil && other != c {
464 // Core and clients know outputs by connector name.
465 log.Warn().Str("connector", name).Str("card", c.Path()).Msg("connector name already used by another card; ignored")
466 instance, registerErr := registerDiscardedOutput(set, name)
467 o.Instance = instance
468 o.Close() // synchronous final KMS disable; never starts a renderer
469 var shutdownErr error
470 if registerErr == nil {
471 shutdownErr = retireDiscardedOutput(set, instance, o.InactiveOnClose())
472 }
473 if err := errors.Join(registerErr, shutdownErr); err != nil {
474 if discardedOutputErr == nil {
475 discardedOutputErr = err
476 }
477 scanErrors = append(scanErrors, err)
478 log.Error().Err(err).Str("connector", name).Str("card", c.Path()).Msg("discarded output protection unconfirmed")
479 // An exhausted instance cannot represent an unsafe lifetime.
480 // Stop the backend rather than ever acknowledge this inventory.
481 if instance == 0 {
482 cancel()
483 }
484 }
485 c.Release(name)
486 continue
487 }
488 cards[name] = c
489 outputs[name] = o
490 readySources[name] = o.Ready()
491 if progress.active && !want(currentConfig).Disabled[name] {
492 progress.required[name] = true
493 }
494 source := readySources[name]
495 stopped := make(chan struct{})
496 stopReady[name] = stopped
497 go func() {
498 select {
499 case err := <-source:
500 select {
501 case ready <- readyResult{name, source, err}:
502 case <-ctx.Done():
503 }
504 case <-stopped:
505 case <-ctx.Done():
506 }
507 }()
508 o.NewCaptureRenderer = newRenderer
509 o.StartOff = set.off[name]
510 progress.started(name, source)
511 if cur := o.Cursor(); cur != nil {
512 curs.set(name, cur)
513 }
514 active := b.seat.Subscribe()
515 realtime := currentConfig.Performance.Realtime
516 started := set.start(ctx, name, func(octx context.Context, sc <-chan ports.Scene, cc <-chan ports.SurfaceContent, cu <-chan ports.CursorChange, cap <-chan ports.CaptureRequest, secure <-chan ports.SecurityState, instance ports.OutputInstance) error {
517 defer b.seat.Unsubscribe(active)
518 o.Security, o.SecurityChanges, o.Instance = ch.security, secure, instance
519 return safe("output "+name, func() error {
520 defer o.Close()
521 b.addPreparer(o)
522 defer b.removePreparer(o) // before Close: it handles no request any more
523 // octx derives from the session: the probe ends with this
524 // output or the session.
525 b.addProbe(name, func() { o.Ping(octx) })
526 defer b.removeProbe(name)
527 if realtime {
528 runtime.LockOSThread()
529 // Keep the thread locked until this goroutine exits: an RT
530 // thread must not return to Go's general-purpose pool.
531 schedLog := logging.For(ctx, "sched")
532 if err := sched.Realtime(schedLog); err != nil {
533 schedLog.Warn().Err(err).Msg("output scheduling")
534 }
535 }
536 return o.Run(octx, newRenderer, loadCursor, active, sc, cc, cu, ch.presented, cap, ch.captured)
537 })
538 }, &o.SecurityEvents)
539 if !started {
540 b.seat.Unsubscribe(active)
541 o.Close()
542 c.Release(name)
543 continue
544 }
545 send(ports.OutputAdded{Info: o.Info()})
546 }
547 }
548 publishInventory()
549 complete(progress.scanError(errors.Join(scanErrors...)))
550 barrier()
551 }
552 scan()
553 if len(set.outs) == 0 {
554 return errors.Join(errors.New("no connected display"), set.wait())
555 }
556 // configure starts cfg; complete reports its result to apply. apply
557 // hands out one configuration at a time, and a rollback or reload starts
558 // on a later iteration, never inside scan or another configure.
559 configure := func(cfg ports.Config) {
560 currentConfig = cfg
561 op++
562 timer.Stop()
563 deadline = nil
564 required := map[string]bool{}
565 // Desired enabled heads, not the inventory's current mode, require a first modeset.
566 w := want(currentConfig)
567 for _, h := range currentHeads.Heads {
568 if !w.Disabled[h.Info.Name] {
569 required[h.Info.Name] = true
570 }
571 }
572 for name := range set.outs {
573 if !w.Disabled[name] {
574 required[name] = true
575 }
576 }
577 progress.start(op, required, stopping)
578 scan()
579 running := map[string]<-chan error{}
580 for name := range set.outs {
581 running[name] = readySources[name]
582 }
583 d := progress.scanned(running)
584 if progress.active {
585 timer.Reset(applyTimeout)
586 deadline = timer.C()
587 }
588 complete(d)
589 }
590 for {
591 if cfg, ok := apply.next(); ok {
592 configure(cfg)
593 continue
594 }
595 configs, configNext := apply.configOut(ch.configured), apply.config
596 replies, replyNext := apply.replyOut(ch.replies)
597 heads, headsNext := apply.headsOut(ch.heads)
598 contents, contentOut, contentNext, contentOwner := set.contentOut(ch.contents)
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 c := <-contents:
704 set.content(c)
705 case contentOut <- contentNext:
706 contentOwner.contentSent()
707 case q := <-ch.captures:
708 set.routeCapture(q)
709 case c := <-ch.cursorChanges:
710 set.setCursor(c)
711 case stopped := <-set.stopped:
712 if !set.stoppedCurrent(stopped) {
713 continue
714 }
715 name := stopped.name
716 output := outputs[name]
717 err := set.finish(name)
718 // finish observes done after the deferred Close. Only affirmative
719 // terminal KMS inactivity retires this exact lifetime; renderer
720 // errors and missing connector inventory are not physical proof.
721 if output != nil && output.InactiveOnClose() {
722 set.securityEvent(ports.SecurityOutputRemoved{Instance: stopped.instance})
723 }
724 delete(outputs, name)
725 if ctx.Err() != nil {
726 return errors.Join(err, set.wait())
727 }
728 restart := stopping[name]
729 delete(stopping, name)
730 curs.set(name, nil)
731 card := cards[name]
732 card.Release(name)
733 lastHeads[card] = releasedHead(lastHeads[card], name)
734 delete(cards, name)
735 source := readySources[name]
736 delete(readySources, name)
737 if stopped := stopReady[name]; stopped != nil {
738 close(stopped)
739 delete(stopReady, name)
740 }
741 stopDecision := progress.stopped(name, source, restart, err)
742 if err != nil {
743 // A broken output (e.g. its renderer) must not take the
744 // session down. It stays off until the next hotplug, so a
745 // lasting failure does not loop.
746 log.Error().Err(err).Str("connector", name).Msg("output stopped")
747 restart = false
748 }
749 // Release changes the enabled state in ConnectedHeads. Do not
750 // restart a failed or disabled output just to refresh inventory.
751 if restart {
752 scan()
753 } else {
754 publishInventory()
755 }
756 complete(stopDecision)
757 if restart {
758 // Same connector, new mode: core keeps its screen and
759 // workspaces; OutputAdded updates the size.
760 if set.outs[name] != nil {
761 running := map[string]<-chan error{}
762 for n := range set.outs {
763 running[n] = readySources[n]
764 }
765 complete(progress.scanned(running))
766 continue
767 }
768 }
769 send(ports.OutputRemoved{Name: name})
770 running := map[string]<-chan error{}
771 for n := range set.outs {
772 running[n] = readySources[n]
773 }
774 complete(progress.scanned(running))
775 }
776 }
777}