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, ch.routes())
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 // Only the select is timed: the router logs the longest wait, which
351 // tells whether core holds up the backend loop.
352 start := time.Now()
353 select {
354 case ch.events <- ev:
355 case <-ctx.Done():
356 }
357 set.router.noteEventSend(time.Since(start))
358 }
359 // stopping holds outputs asked to stop: true when they restart with a
360 // new mode (core keeps the screen), false when they are gone.
361 stopping := map[string]bool{}
362 currentConfig := initial
363 var currentHeads ports.OutputHeads
364 lastHeads := map[*drmCard][]ports.OutputHead{}
365 progress := &applyProgress{completed: map[<-chan error]bool{}}
366 readySources := map[string]<-chan error{}
367 stopReady := map[string]chan struct{}{}
368 type readyResult struct {
369 name string
370 source <-chan error
371 err error
372 }
373 ready := make(chan readyResult)
374 // deadline is the pending apply's timer channel, nil while none is pending.
375 timer := b.clock.NewTimer(applyTimeout)
376 timer.Stop()
377 defer timer.Stop()
378 var deadline <-chan time.Time
379 // A small bounded retry budget avoids a busy loop on failed KMS revokes.
380 revokeTimer := b.clock.NewTimer(time.Second)
381 revokeTimer.Stop()
382 defer revokeTimer.Stop()
383 var revokeDeadline <-chan time.Time
384 var revokeAttempts int
385 var discardedOutputErr error // first unconfirmed lifetime; inventory retains every instance
386 barrier := func() {
387 if !set.state.Protected {
388 return
389 }
390 var errs []error
391 for _, c := range b.cards {
392 errs = append(errs, revokeSecurityLeases(c, func(id uint32) {
393 leases.send(ports.LeaseFinished{Card: c.Path(), LeaseID: id})
394 }))
395 leases.publish(c)
396 }
397 errs = append(errs, discardedOutputErr)
398 err := errors.Join(errs...)
399 // Registration runs synchronously in scan before any barrier. A gate
400 // transition racing native revocation belongs to a later round.
401 if securitySnapshot(ch.security).Generation > set.state.Generation {
402 return
403 }
404 set.securityEvent(ports.SecurityBackendBarrier{Generation: set.state.Generation, Err: err})
405 revokeAttempts++
406 if err != nil && revokeAttempts < 3 {
407 revokeTimer.Reset(time.Second)
408 revokeDeadline = revokeTimer.C()
409 }
410 }
411 var op uint64
412 complete := func(d applyDecision) {
413 if !d.reply {
414 return
415 }
416 timer.Stop()
417 deadline = nil
418 apply.finished(d.err)
419 }
420 publishInventory := func() {
421 var inventory ports.OutputHeads
422 for _, card := range b.cards {
423 found, err := card.ConnectedHeads()
424 if err != nil {
425 log.Warn().Err(err).Str("card", card.Path()).Msg("inventory outputs")
426 }
427 lastHeads[card] = retainedHeads(lastHeads[card], found, err)
428 inventory.Heads = append(inventory.Heads, lastHeads[card]...)
429 }
430 currentHeads = inventory
431 apply.heads(inventory)
432 }
433 scan := func() {
434 var scanErrors []error
435 for _, c := range b.cards {
436 w := want(currentConfig)
437 w.Device = c.Device()
438 c.SetWant(w)
439 added, removed, replaced, err := c.Scan()
440 if err != nil {
441 log.Warn().Err(err).Str("card", c.Path()).Msg("scan connectors")
442 scanErrors = append(scanErrors, fmt.Errorf("scan %s: %w", c.Path(), err))
443 continue
444 }
445 leases.publish(c)
446 stop := func(name string, restart bool) {
447 if r := set.outs[name]; r != nil && cards[name] == c {
448 if _, ok := stopping[name]; !ok {
449 stopping[name] = restart
450 progress.stopping(name)
451 r.stop()
452 }
453 }
454 }
455 for _, name := range removed {
456 log.Info().Str("connector", name).Msg("output unplugged")
457 stop(name, false)
458 }
459 for _, name := range replaced {
460 log.Info().Str("connector", name).Msg("output mode change")
461 stop(name, true)
462 }
463 for _, o := range added {
464 // Every physical lifetime gets the gate before any early exit:
465 // Close must not restore saved desktop, even on a name collision.
466 o.Security, o.Frames = ch.security, ch.frames
467 name := o.Info().Name
468 if other := cards[name]; other != nil && other != c {
469 // Core and clients know outputs by connector name.
470 log.Warn().Str("connector", name).Str("card", c.Path()).Msg("connector name already used by another card; ignored")
471 instance, registerErr := registerDiscardedOutput(set, name)
472 o.Instance = instance
473 o.Close() // synchronous final KMS disable; never starts a renderer
474 var shutdownErr error
475 if registerErr == nil {
476 shutdownErr = retireDiscardedOutput(set, instance, o.InactiveOnClose())
477 }
478 if err := errors.Join(registerErr, shutdownErr); err != nil {
479 if discardedOutputErr == nil {
480 discardedOutputErr = err
481 }
482 scanErrors = append(scanErrors, err)
483 log.Error().Err(err).Str("connector", name).Str("card", c.Path()).Msg("discarded output protection unconfirmed")
484 // An exhausted instance cannot represent an unsafe lifetime.
485 // Stop the backend rather than ever acknowledge this inventory.
486 if instance == 0 {
487 cancel()
488 }
489 }
490 c.Release(name)
491 continue
492 }
493 cards[name] = c
494 outputs[name] = o
495 readySources[name] = o.Ready()
496 if progress.active && !want(currentConfig).Disabled[name] {
497 progress.required[name] = true
498 }
499 source := readySources[name]
500 stopped := make(chan struct{})
501 stopReady[name] = stopped
502 go func() {
503 select {
504 case err := <-source:
505 select {
506 case ready <- readyResult{name, source, err}:
507 case <-ctx.Done():
508 }
509 case <-stopped:
510 case <-ctx.Done():
511 }
512 }()
513 o.NewCaptureRenderer = newRenderer
514 progress.started(name, source)
515 if cur := o.Cursor(); cur != nil {
516 curs.set(name, cur)
517 }
518 active := b.seat.Subscribe()
519 realtime := currentConfig.Performance.Realtime
520 started := set.start(ctx, name, func(octx context.Context, feed outputFeed) error {
521 defer b.seat.Unsubscribe(active)
522 o.Security, o.SecurityChanges, o.Instance, o.StartOff = ch.security, feed.Security, feed.Instance, feed.StartOff
523 return safe("output "+name, func() error {
524 defer o.Close()
525 b.addPreparer(o)
526 defer b.removePreparer(o) // before Close: it handles no request any more
527 // octx derives from the session: the probe ends with this
528 // output or the session.
529 b.addProbe(name, func() { o.Ping(octx) })
530 defer b.removeProbe(name)
531 if realtime {
532 runtime.LockOSThread()
533 // Keep the thread locked until this goroutine exits: an RT
534 // thread must not return to Go's general-purpose pool.
535 schedLog := logging.For(ctx, "sched")
536 if err := sched.Realtime(schedLog); err != nil {
537 schedLog.Warn().Err(err).Msg("output scheduling")
538 }
539 }
540 return o.Run(octx, newRenderer, loadCursor, active, feed.Scenes, feed.Contents, feed.Returns, feed.Cursor, ch.presented, feed.Captures, ch.captured)
541 })
542 }, &o.SecurityEvents)
543 if !started {
544 b.seat.Unsubscribe(active)
545 o.Close()
546 c.Release(name)
547 continue
548 }
549 send(ports.OutputAdded{Info: o.Info()})
550 }
551 }
552 publishInventory()
553 complete(progress.scanError(errors.Join(scanErrors...)))
554 barrier()
555 }
556 scan()
557 if len(set.outs) == 0 {
558 return errors.Join(errors.New("no connected display"), set.wait())
559 }
560 // configure starts cfg; complete reports its result to apply. apply
561 // hands out one configuration at a time, and a rollback or reload starts
562 // on a later iteration, never inside scan or another configure.
563 configure := func(cfg ports.Config) {
564 currentConfig = cfg
565 op++
566 timer.Stop()
567 deadline = nil
568 required := map[string]bool{}
569 // Desired enabled heads, not the inventory's current mode, require a first modeset.
570 w := want(currentConfig)
571 for _, h := range currentHeads.Heads {
572 if !w.Disabled[h.Info.Name] {
573 required[h.Info.Name] = true
574 }
575 }
576 for name := range set.outs {
577 if !w.Disabled[name] {
578 required[name] = true
579 }
580 }
581 progress.start(op, required, stopping)
582 scan()
583 running := map[string]<-chan error{}
584 for name := range set.outs {
585 running[name] = readySources[name]
586 }
587 d := progress.scanned(running)
588 if progress.active {
589 timer.Reset(applyTimeout)
590 deadline = timer.C()
591 }
592 complete(d)
593 }
594 for {
595 if cfg, ok := apply.next(); ok {
596 configure(cfg)
597 continue
598 }
599 configs, configNext := apply.configOut(ch.configured), apply.config
600 replies, replyNext := apply.replyOut(ch.replies)
601 heads, headsNext := apply.headsOut(ch.heads)
602 select {
603 case <-ctx.Done():
604 return set.wait()
605 case err := <-readerErr:
606 // No more flips on that card: its outputs would freeze.
607 return errors.Join(err, set.wait())
608 case state, ok := <-ch.securityChanges:
609 if !ok {
610 ch.securityChanges = nil
611 continue
612 }
613 if state.Generation < set.state.Generation {
614 continue
615 }
616 if set.setSecurity(state) {
617 revokeAttempts = 0
618 }
619 // start can refresh the gate while scanning; its queued transition
620 // must still receive a barrier even if forwarding already occurred.
621 revokeTimer.Stop()
622 revokeDeadline = nil
623 if set.state.Protected {
624 barrier()
625 } else {
626 for _, c := range b.cards {
627 leases.publish(c)
628 }
629 }
630 case <-revokeDeadline:
631 revokeDeadline = nil
632 barrier()
633 case <-hotplug:
634 log.Info().Msg("hotplug")
635 scan()
636 if progress.active {
637 running := map[string]<-chan error{}
638 for name := range set.outs {
639 running[name] = readySources[name]
640 }
641 complete(progress.scanned(running))
642 }
643 case enabled := <-active:
644 seatActive = enabled
645 if !enabled {
646 for _, c := range b.cards {
647 for _, id := range c.LeaseIDs() {
648 if err := c.Revoke(id); err != nil {
649 log.Warn().Err(err).Uint32("lease", id).Msg("VT lease revoke")
650 } else {
651 leases.send(ports.LeaseFinished{Card: c.Path(), LeaseID: id})
652 }
653 }
654 leases.publish(c)
655 }
656 } else {
657 for _, c := range b.cards {
658 leases.publish(c)
659 }
660 }
661 case event := <-ch.leaseRequests:
662 switch req := event.(type) {
663 case ports.LeaseRequest:
664 reply := ports.LeaseReply{ID: req.ID, Err: fmt.Errorf("unknown card %s", req.Card)}
665 if !seatActive || protected() {
666 reply.Err = errors.New("DRM seat inactive or session protected")
667 leases.send(reply)
668 break
669 }
670 for _, c := range b.cards {
671 if c.Path() == req.Card {
672 reply.FD, reply.LeaseID, reply.Err = guardedLease(c, ch.security, req.Connectors)
673 leases.publish(c)
674 break
675 }
676 }
677 leases.send(reply)
678 case ports.LeaseRevoke:
679 for _, c := range b.cards {
680 if c.Path() == req.Card {
681 if err := c.Revoke(req.LeaseID); err != nil {
682 log.Warn().Err(err).Uint32("lease", req.LeaseID).Msg("revoke lease")
683 } else {
684 leases.publish(c)
685 }
686 break
687 }
688 }
689 }
690 case ev := <-ch.reloads:
691 apply.reload(ev.Config)
692 case req := <-ch.requests:
693 apply.request(req)
694 case configs <- configNext:
695 apply.configSent()
696 case replies <- replyNext:
697 apply.replySent()
698 case heads <- headsNext:
699 apply.headsSent()
700 case result := <-ready:
701 complete(progress.readyEvent(result.name, result.source, readySources[result.name], result.err))
702 case <-deadline:
703 complete(progress.timeout(op))
704 case stopped := <-set.stopped:
705 if !set.stoppedCurrent(stopped) {
706 continue
707 }
708 name := stopped.name
709 output := outputs[name]
710 err := set.finish(name)
711 // finish observes done after the deferred Close. Only affirmative
712 // terminal KMS inactivity retires this exact lifetime; renderer
713 // errors and missing connector inventory are not physical proof.
714 if output != nil && output.InactiveOnClose() {
715 set.securityEvent(ports.SecurityOutputRemoved{Instance: stopped.instance})
716 }
717 delete(outputs, name)
718 if ctx.Err() != nil {
719 return errors.Join(err, set.wait())
720 }
721 restart := stopping[name]
722 delete(stopping, name)
723 curs.set(name, nil)
724 card := cards[name]
725 card.Release(name)
726 lastHeads[card] = releasedHead(lastHeads[card], name)
727 delete(cards, name)
728 source := readySources[name]
729 delete(readySources, name)
730 if stopped := stopReady[name]; stopped != nil {
731 close(stopped)
732 delete(stopReady, name)
733 }
734 stopDecision := progress.stopped(name, source, restart, err)
735 if err != nil {
736 // A broken output (e.g. its renderer) must not take the
737 // session down. It stays off until the next hotplug, so a
738 // lasting failure does not loop.
739 log.Error().Err(err).Str("connector", name).Msg("output stopped")
740 restart = false
741 }
742 // Release changes the enabled state in ConnectedHeads. Do not
743 // restart a failed or disabled output just to refresh inventory.
744 if restart {
745 scan()
746 } else {
747 publishInventory()
748 }
749 complete(stopDecision)
750 if restart {
751 // Same connector, new mode: core keeps its screen and
752 // workspaces; OutputAdded updates the size.
753 if set.outs[name] != nil {
754 running := map[string]<-chan error{}
755 for n := range set.outs {
756 running[n] = readySources[n]
757 }
758 complete(progress.scanned(running))
759 continue
760 }
761 }
762 send(ports.OutputRemoved{Name: name})
763 running := map[string]<-chan error{}
764 for n := range set.outs {
765 running[n] = readySources[n]
766 }
767 complete(progress.scanned(running))
768 }
769 }
770}