detached 90639c0

refactor(app): route scenes, cursor and captures from a dedicated goroutine

The backend loop routed every output's scenes, cursor and capture requests
and also ran output lifecycle work inline (connector scans, modesets, output
start and stop, content attach waiting on the display goroutine), so a
hotplug or configuration on one monitor froze frames and cursor on the
other. An outputRouter goroutine now owns the per-output mailboxes; the
backend loop registers and unregisters lifetimes with it. Outputs receive an
outputFeed, whose StartOff comes from the router's latest scenes.
commit
90639c07818b1fb419e21d2be15f129a74813312
author
brice · · 2026-10-18 05:06 UTC
parent
9784d35
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
12 files · +462 -191
Unified
UnifiedSplit
M internal/app/capture_test.go +13 -18
@@ -11,12 +11,12 @@ import (
1111
1212 func TestCaptureUnknownOutputClosesFile(t *testing.T) {
1313 replies := make(chan ports.CaptureDone, 1)
14- set := newOutputSet(context.Background(), replies)
14+ router := newOutputRouter(context.Background(), replies, outputRoutes{})
1515 f, err := os.CreateTemp(t.TempDir(), "capture")
1616 if err != nil {
1717 t.Fatal(err)
1818 }
19- set.routeCapture(ports.CaptureRequest{ID: 9, Output: "missing", Dst: ports.SHMBuffer{File: f}})
19+ router.routeCapture(ports.CaptureRequest{ID: 9, Output: "missing", Dst: ports.SHMBuffer{File: f}})
2020 select {
2121 case result := <-replies:
2222 if result.ID != 9 || result.Err == nil {
@@ -34,17 +34,17 @@ func TestCaptureFullOutputQueueDoesNotBlockRouting(t *testing.T) {
3434 ctx, cancel := context.WithCancel(context.Background())
3535 defer cancel()
3636 replies := make(chan ports.CaptureDone, 1)
37- set := newOutputSet(ctx, replies)
37+ router := newOutputRouter(ctx, replies, outputRoutes{})
3838 queued := make(chan ports.CaptureRequest, 1)
3939 queued <- ports.CaptureRequest{ID: 1}
40- set.outs["test"] = &runningOutput{ctx: ctx, captures: queued, done: make(chan struct{}), stop: func() {}}
40+ router.outs["test"] = &mailbox{ctx: ctx, captures: queued, done: make(chan struct{}), stop: func() {}}
4141 f, err := os.CreateTemp(t.TempDir(), "capture")
4242 if err != nil {
4343 t.Fatal(err)
4444 }
4545 routed := make(chan struct{})
4646 go func() {
47- set.routeCapture(ports.CaptureRequest{ID: 2, Output: "test", Dst: ports.SHMBuffer{File: f}})
47+ router.routeCapture(ports.CaptureRequest{ID: 2, Output: "test", Dst: ports.SHMBuffer{File: f}})
4848 close(routed)
4949 }()
5050 select {
@@ -71,13 +71,13 @@ func TestStoppedOutputCaptureStillReplies(t *testing.T) {
7171 outputCtx, stop := context.WithCancel(ctx)
7272 stop()
7373 replies := make(chan ports.CaptureDone, 1)
74- set := newOutputSet(ctx, replies)
75- set.outs["test"] = &runningOutput{ctx: outputCtx, captures: make(chan ports.CaptureRequest)}
74+ router := newOutputRouter(ctx, replies, outputRoutes{})
75+ router.outs["test"] = &mailbox{ctx: outputCtx, captures: make(chan ports.CaptureRequest)}
7676 f, err := os.CreateTemp(t.TempDir(), "capture")
7777 if err != nil {
7878 t.Fatal(err)
7979 }
80- set.routeCapture(ports.CaptureRequest{ID: 7, Output: "test", Dst: ports.SHMBuffer{File: f}})
80+ router.routeCapture(ports.CaptureRequest{ID: 7, Output: "test", Dst: ports.SHMBuffer{File: f}})
8181 select {
8282 case result := <-replies:
8383 if result.ID != 7 || result.Err == nil {
@@ -93,17 +93,14 @@ func TestStoppedOutputCaptureStillReplies(t *testing.T) {
9393
9494 func TestFinishedOutputDrainsCapturesWithReplies(t *testing.T) {
9595 ctx := context.Background()
96- outputCtx, stop := context.WithCancel(ctx)
97- stop()
9896 replies := make(chan ports.CaptureDone, 1)
99- set := newOutputSet(ctx, replies)
10097 queued := make(chan ports.CaptureRequest, 1)
10198 f, err := os.CreateTemp(t.TempDir(), "capture")
10299 if err != nil {
103100 t.Fatal(err)
104101 }
105102 queued <- ports.CaptureRequest{ID: 8, Dst: ports.SHMBuffer{File: f}}
106- set.failQueued(&runningOutput{ctx: outputCtx, captures: queued})
103+ failQueued(ctx, &mailbox{captures: queued}, replies)
107104 select {
108105 case result := <-replies:
109106 if result.ID != 8 || result.Err == nil {
@@ -143,16 +140,14 @@ func TestOutputStopWithFullReplies(t *testing.T) {
143140 ctx, cancel := context.WithCancel(context.Background())
144141 replies := make(chan ports.CaptureDone, 1)
145142 replies <- ports.CaptureDone{}
146- set := newOutputSet(ctx, replies)
147- set.start(ctx, "test", func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, requests <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {
148- <-ctx.Done()
149- return nil
150- })
143+ captures := make(chan ports.CaptureRequest)
144+ set := newOutputSet(ctx, replies, outputRoutes{captures: captures})
145+ set.start(ctx, "test", idleRun)
151146 f, err := os.CreateTemp(t.TempDir(), "queued")
152147 if err != nil {
153148 t.Fatal(err)
154149 }
155- set.routeCapture(ports.CaptureRequest{Output: "test", Dst: ports.SHMBuffer{File: f}})
150+ captures <- ports.CaptureRequest{Output: "test", Dst: ports.SHMBuffer{File: f}}
156151 cancel()
157152 done := make(chan struct{})
158153 go func() { _ = set.wait(); close(done) }()
@@ -11,12 +11,12 @@ import (
11 11
12 func TestCaptureUnknownOutputClosesFile(t *testing.T) {12 func TestCaptureUnknownOutputClosesFile(t *testing.T) {
13 replies := make(chan ports.CaptureDone, 1)13 replies := make(chan ports.CaptureDone, 1)
14- set := newOutputSet(context.Background(), replies)14+ router := newOutputRouter(context.Background(), replies, outputRoutes{})
15 f, err := os.CreateTemp(t.TempDir(), "capture")15 f, err := os.CreateTemp(t.TempDir(), "capture")
16 if err != nil {16 if err != nil {
17 t.Fatal(err)17 t.Fatal(err)
18 }18 }
19- set.routeCapture(ports.CaptureRequest{ID: 9, Output: "missing", Dst: ports.SHMBuffer{File: f}})19+ router.routeCapture(ports.CaptureRequest{ID: 9, Output: "missing", Dst: ports.SHMBuffer{File: f}})
20 select {20 select {
21 case result := <-replies:21 case result := <-replies:
22 if result.ID != 9 || result.Err == nil {22 if result.ID != 9 || result.Err == nil {
@@ -34,17 +34,17 @@ func TestCaptureFullOutputQueueDoesNotBlockRouting(t *testing.T) {
34 ctx, cancel := context.WithCancel(context.Background())34 ctx, cancel := context.WithCancel(context.Background())
35 defer cancel()35 defer cancel()
36 replies := make(chan ports.CaptureDone, 1)36 replies := make(chan ports.CaptureDone, 1)
37- set := newOutputSet(ctx, replies)37+ router := newOutputRouter(ctx, replies, outputRoutes{})
38 queued := make(chan ports.CaptureRequest, 1)38 queued := make(chan ports.CaptureRequest, 1)
39 queued <- ports.CaptureRequest{ID: 1}39 queued <- ports.CaptureRequest{ID: 1}
40- set.outs["test"] = &runningOutput{ctx: ctx, captures: queued, done: make(chan struct{}), stop: func() {}}40+ router.outs["test"] = &mailbox{ctx: ctx, captures: queued, done: make(chan struct{}), stop: func() {}}
41 f, err := os.CreateTemp(t.TempDir(), "capture")41 f, err := os.CreateTemp(t.TempDir(), "capture")
42 if err != nil {42 if err != nil {
43 t.Fatal(err)43 t.Fatal(err)
44 }44 }
45 routed := make(chan struct{})45 routed := make(chan struct{})
46 go func() {46 go func() {
47- set.routeCapture(ports.CaptureRequest{ID: 2, Output: "test", Dst: ports.SHMBuffer{File: f}})47+ router.routeCapture(ports.CaptureRequest{ID: 2, Output: "test", Dst: ports.SHMBuffer{File: f}})
48 close(routed)48 close(routed)
49 }()49 }()
50 select {50 select {
@@ -71,13 +71,13 @@ func TestStoppedOutputCaptureStillReplies(t *testing.T) {
71 outputCtx, stop := context.WithCancel(ctx)71 outputCtx, stop := context.WithCancel(ctx)
72 stop()72 stop()
73 replies := make(chan ports.CaptureDone, 1)73 replies := make(chan ports.CaptureDone, 1)
74- set := newOutputSet(ctx, replies)74+ router := newOutputRouter(ctx, replies, outputRoutes{})
75- set.outs["test"] = &runningOutput{ctx: outputCtx, captures: make(chan ports.CaptureRequest)}75+ router.outs["test"] = &mailbox{ctx: outputCtx, captures: make(chan ports.CaptureRequest)}
76 f, err := os.CreateTemp(t.TempDir(), "capture")76 f, err := os.CreateTemp(t.TempDir(), "capture")
77 if err != nil {77 if err != nil {
78 t.Fatal(err)78 t.Fatal(err)
79 }79 }
80- set.routeCapture(ports.CaptureRequest{ID: 7, Output: "test", Dst: ports.SHMBuffer{File: f}})80+ router.routeCapture(ports.CaptureRequest{ID: 7, Output: "test", Dst: ports.SHMBuffer{File: f}})
81 select {81 select {
82 case result := <-replies:82 case result := <-replies:
83 if result.ID != 7 || result.Err == nil {83 if result.ID != 7 || result.Err == nil {
@@ -93,17 +93,14 @@ func TestStoppedOutputCaptureStillReplies(t *testing.T) {
93 93
94 func TestFinishedOutputDrainsCapturesWithReplies(t *testing.T) {94 func TestFinishedOutputDrainsCapturesWithReplies(t *testing.T) {
95 ctx := context.Background()95 ctx := context.Background()
96- outputCtx, stop := context.WithCancel(ctx)
97- stop()
98 replies := make(chan ports.CaptureDone, 1)96 replies := make(chan ports.CaptureDone, 1)
99- set := newOutputSet(ctx, replies)
100 queued := make(chan ports.CaptureRequest, 1)97 queued := make(chan ports.CaptureRequest, 1)
101 f, err := os.CreateTemp(t.TempDir(), "capture")98 f, err := os.CreateTemp(t.TempDir(), "capture")
102 if err != nil {99 if err != nil {
103 t.Fatal(err)100 t.Fatal(err)
104 }101 }
105 queued <- ports.CaptureRequest{ID: 8, Dst: ports.SHMBuffer{File: f}}102 queued <- ports.CaptureRequest{ID: 8, Dst: ports.SHMBuffer{File: f}}
106- set.failQueued(&runningOutput{ctx: outputCtx, captures: queued})103+ failQueued(ctx, &mailbox{captures: queued}, replies)
107 select {104 select {
108 case result := <-replies:105 case result := <-replies:
109 if result.ID != 8 || result.Err == nil {106 if result.ID != 8 || result.Err == nil {
@@ -143,16 +140,14 @@ func TestOutputStopWithFullReplies(t *testing.T) {
143 ctx, cancel := context.WithCancel(context.Background())140 ctx, cancel := context.WithCancel(context.Background())
144 replies := make(chan ports.CaptureDone, 1)141 replies := make(chan ports.CaptureDone, 1)
145 replies <- ports.CaptureDone{}142 replies <- ports.CaptureDone{}
146- set := newOutputSet(ctx, replies)143+ captures := make(chan ports.CaptureRequest)
147- set.start(ctx, "test", func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, requests <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {144+ set := newOutputSet(ctx, replies, outputRoutes{captures: captures})
148- <-ctx.Done()145+ set.start(ctx, "test", idleRun)
149- return nil
150- })
151 f, err := os.CreateTemp(t.TempDir(), "queued")146 f, err := os.CreateTemp(t.TempDir(), "queued")
152 if err != nil {147 if err != nil {
153 t.Fatal(err)148 t.Fatal(err)
154 }149 }
155- set.routeCapture(ports.CaptureRequest{Output: "test", Dst: ports.SHMBuffer{File: f}})150+ captures <- ports.CaptureRequest{Output: "test", Dst: ports.SHMBuffer{File: f}}
156 cancel()151 cancel()
157 done := make(chan struct{})152 done := make(chan struct{})
158 go func() { _ = set.wait(); close(done) }()153 go func() { _ = set.wait(); close(done) }()
M internal/app/contentattacher_test.go +5 -4
@@ -17,7 +17,7 @@ import (
1717 func TestOutputSetAttachesAndDetachesContents(t *testing.T) {
1818 ctx, cancel := context.WithCancel(context.Background())
1919 defer cancel()
20- set := newOutputSet(ctx, nil)
20+ set := newOutputSet(ctx, nil, outputRoutes{})
2121 events := make(chan ports.SecurityBackendEvent, 4)
2222 set.wireSecurity(outputChannels{securityEvents: events})
2323 attacher := newMockcontentAttacher(t)
@@ -31,7 +31,8 @@ func TestOutputSetAttachesAndDetachesContents(t *testing.T) {
3131 security ports.SecurityState
3232 }
3333 started := make(chan seen, 2)
34- run := func(ctx context.Context, _ <-chan ports.Scene, contents <-chan *ports.ContentBatch, returns ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, security <-chan ports.SecurityState, _ ports.OutputInstance) error {
34+ run := func(ctx context.Context, feed outputFeed) error {
35+ contents, returns, security := feed.Contents, feed.Returns, feed.Security
3536 started <- seen{contents, returns, <-security}
3637 <-ctx.Done()
3738 return nil
@@ -86,13 +87,13 @@ func TestOutputSetAttachesAndDetachesContents(t *testing.T) {
8687 func TestOutputSetWaitDetachesContents(t *testing.T) {
8788 ctx, cancel := context.WithCancel(context.Background())
8889 defer cancel()
89- set := newOutputSet(ctx, nil)
90+ set := newOutputSet(ctx, nil, outputRoutes{})
9091 attacher := newMockcontentAttacher(t)
9192 set.attach = attacher
9293 ret := portsmocks.NewMockContentReturn(t)
9394 attacher.EXPECT().AttachOutput("A").Return(nil, ret).Once()
9495 attacher.EXPECT().DetachOutput(ports.ContentReturn(ret)).Return().Once()
95- set.start(ctx, "A", func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {
96+ set.start(ctx, "A", func(ctx context.Context, feed outputFeed) error {
9697 <-ctx.Done()
9798 return nil
9899 })
@@ -17,7 +17,7 @@ import (
17 func TestOutputSetAttachesAndDetachesContents(t *testing.T) {17 func TestOutputSetAttachesAndDetachesContents(t *testing.T) {
18 ctx, cancel := context.WithCancel(context.Background())18 ctx, cancel := context.WithCancel(context.Background())
19 defer cancel()19 defer cancel()
20- set := newOutputSet(ctx, nil)20+ set := newOutputSet(ctx, nil, outputRoutes{})
21 events := make(chan ports.SecurityBackendEvent, 4)21 events := make(chan ports.SecurityBackendEvent, 4)
22 set.wireSecurity(outputChannels{securityEvents: events})22 set.wireSecurity(outputChannels{securityEvents: events})
23 attacher := newMockcontentAttacher(t)23 attacher := newMockcontentAttacher(t)
@@ -31,7 +31,8 @@ func TestOutputSetAttachesAndDetachesContents(t *testing.T) {
31 security ports.SecurityState31 security ports.SecurityState
32 }32 }
33 started := make(chan seen, 2)33 started := make(chan seen, 2)
34- run := func(ctx context.Context, _ <-chan ports.Scene, contents <-chan *ports.ContentBatch, returns ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, security <-chan ports.SecurityState, _ ports.OutputInstance) error {34+ run := func(ctx context.Context, feed outputFeed) error {
35+ contents, returns, security := feed.Contents, feed.Returns, feed.Security
35 started <- seen{contents, returns, <-security}36 started <- seen{contents, returns, <-security}
36 <-ctx.Done()37 <-ctx.Done()
37 return nil38 return nil
@@ -86,13 +87,13 @@ func TestOutputSetAttachesAndDetachesContents(t *testing.T) {
86 func TestOutputSetWaitDetachesContents(t *testing.T) {87 func TestOutputSetWaitDetachesContents(t *testing.T) {
87 ctx, cancel := context.WithCancel(context.Background())88 ctx, cancel := context.WithCancel(context.Background())
88 defer cancel()89 defer cancel()
89- set := newOutputSet(ctx, nil)90+ set := newOutputSet(ctx, nil, outputRoutes{})
90 attacher := newMockcontentAttacher(t)91 attacher := newMockcontentAttacher(t)
91 set.attach = attacher92 set.attach = attacher
92 ret := portsmocks.NewMockContentReturn(t)93 ret := portsmocks.NewMockContentReturn(t)
93 attacher.EXPECT().AttachOutput("A").Return(nil, ret).Once()94 attacher.EXPECT().AttachOutput("A").Return(nil, ret).Once()
94 attacher.EXPECT().DetachOutput(ports.ContentReturn(ret)).Return().Once()95 attacher.EXPECT().DetachOutput(ports.ContentReturn(ret)).Return().Once()
95- set.start(ctx, "A", func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {96+ set.start(ctx, "A", func(ctx context.Context, feed outputFeed) error {
96 <-ctx.Done()97 <-ctx.Done()
97 return nil98 return nil
98 })99 })
M internal/app/drm.go +4 -11
@@ -332,7 +332,7 @@ func (b *drmBackend) runOutputs(ctx context.Context, want func(ports.Config) drm
332332 active := b.seat.Subscribe()
333333 defer b.seat.Unsubscribe(active)
334334 seatActive := true // openDRM waits for the first seat enable
335- set := newOutputSet(ctx, ch.captured)
335+ set := newOutputSet(ctx, ch.captured, ch.routes())
336336 set.attach = ch.content
337337 set.wireSecurity(ch)
338338 protected := func() bool { return set.state.Protected || securitySnapshot(ch.security).Protected }
@@ -507,16 +507,15 @@ func (b *drmBackend) runOutputs(ctx context.Context, want func(ports.Config) drm
507507 }
508508 }()
509509 o.NewCaptureRenderer = newRenderer
510- o.StartOff = set.off[name]
511510 progress.started(name, source)
512511 if cur := o.Cursor(); cur != nil {
513512 curs.set(name, cur)
514513 }
515514 active := b.seat.Subscribe()
516515 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 {
516+ started := set.start(ctx, name, func(octx context.Context, feed outputFeed) error {
518517 defer b.seat.Unsubscribe(active)
519- o.Security, o.SecurityChanges, o.Instance = ch.security, secure, instance
518+ o.Security, o.SecurityChanges, o.Instance, o.StartOff = ch.security, feed.Security, feed.Instance, feed.StartOff
520519 return safe("output "+name, func() error {
521520 defer o.Close()
522521 b.addPreparer(o)
@@ -534,7 +533,7 @@ func (b *drmBackend) runOutputs(ctx context.Context, want func(ports.Config) drm
534533 schedLog.Warn().Err(err).Msg("output scheduling")
535534 }
536535 }
537- return o.Run(octx, newRenderer, loadCursor, active, sc, cc, rc, cu, ch.presented, cap, ch.captured)
536+ return o.Run(octx, newRenderer, loadCursor, active, feed.Scenes, feed.Contents, feed.Returns, feed.Cursor, ch.presented, feed.Captures, ch.captured)
538537 })
539538 }, &o.SecurityEvents)
540539 if !started {
@@ -698,12 +697,6 @@ func (b *drmBackend) runOutputs(ctx context.Context, want func(ports.Config) drm
698697 complete(progress.readyEvent(result.name, result.source, readySources[result.name], result.err))
699698 case <-deadline:
700699 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)
707700 case stopped := <-set.stopped:
708701 if !set.stoppedCurrent(stopped) {
709702 continue
@@ -332,7 +332,7 @@ func (b *drmBackend) runOutputs(ctx context.Context, want func(ports.Config) drm
332 active := b.seat.Subscribe()332 active := b.seat.Subscribe()
333 defer b.seat.Unsubscribe(active)333 defer b.seat.Unsubscribe(active)
334 seatActive := true // openDRM waits for the first seat enable334 seatActive := true // openDRM waits for the first seat enable
335- set := newOutputSet(ctx, ch.captured)335+ set := newOutputSet(ctx, ch.captured, ch.routes())
336 set.attach = ch.content336 set.attach = ch.content
337 set.wireSecurity(ch)337 set.wireSecurity(ch)
338 protected := func() bool { return set.state.Protected || securitySnapshot(ch.security).Protected }338 protected := func() bool { return set.state.Protected || securitySnapshot(ch.security).Protected }
@@ -507,16 +507,15 @@ func (b *drmBackend) runOutputs(ctx context.Context, want func(ports.Config) drm
507 }507 }
508 }()508 }()
509 o.NewCaptureRenderer = newRenderer509 o.NewCaptureRenderer = newRenderer
510- o.StartOff = set.off[name]
511 progress.started(name, source)510 progress.started(name, source)
512 if cur := o.Cursor(); cur != nil {511 if cur := o.Cursor(); cur != nil {
513 curs.set(name, cur)512 curs.set(name, cur)
514 }513 }
515 active := b.seat.Subscribe()514 active := b.seat.Subscribe()
516 realtime := currentConfig.Performance.Realtime515 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 {516+ started := set.start(ctx, name, func(octx context.Context, feed outputFeed) error {
518 defer b.seat.Unsubscribe(active)517 defer b.seat.Unsubscribe(active)
519- o.Security, o.SecurityChanges, o.Instance = ch.security, secure, instance518+ o.Security, o.SecurityChanges, o.Instance, o.StartOff = ch.security, feed.Security, feed.Instance, feed.StartOff
520 return safe("output "+name, func() error {519 return safe("output "+name, func() error {
521 defer o.Close()520 defer o.Close()
522 b.addPreparer(o)521 b.addPreparer(o)
@@ -534,7 +533,7 @@ func (b *drmBackend) runOutputs(ctx context.Context, want func(ports.Config) drm
534 schedLog.Warn().Err(err).Msg("output scheduling")533 schedLog.Warn().Err(err).Msg("output scheduling")
535 }534 }
536 }535 }
537- return o.Run(octx, newRenderer, loadCursor, active, sc, cc, rc, cu, ch.presented, cap, ch.captured)536+ return o.Run(octx, newRenderer, loadCursor, active, feed.Scenes, feed.Contents, feed.Returns, feed.Cursor, ch.presented, feed.Captures, ch.captured)
538 })537 })
539 }, &o.SecurityEvents)538 }, &o.SecurityEvents)
540 if !started {539 if !started {
@@ -698,12 +697,6 @@ func (b *drmBackend) runOutputs(ctx context.Context, want func(ports.Config) drm
698 complete(progress.readyEvent(result.name, result.source, readySources[result.name], result.err))697 complete(progress.readyEvent(result.name, result.source, readySources[result.name], result.err))
699 case <-deadline:698 case <-deadline:
700 complete(progress.timeout(op))699 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:700 case stopped := <-set.stopped:
708 if !set.stoppedCurrent(stopped) {701 if !set.stoppedCurrent(stopped) {
709 continue702 continue
M internal/app/headless.go +4 -10
@@ -26,7 +26,7 @@ type headlessOptions struct {
2626 //
2727 // newRenderer is configured for virtual outputs by newVulkanRenderer.
2828 func runHeadless(ctx context.Context, o headlessOptions, apply *outputApply, ch outputChannels, curs *cursors, newRenderer func(w, h int) (ports.Renderer, error), log zerowrap.Logger) error {
29- set := newOutputSet(ctx, ch.captured)
29+ set := newOutputSet(ctx, ch.captured, ch.routes())
3030 set.attach = ch.content
3131 set.wireSecurity(ch)
3232 inventory := ports.OutputHeads{}
@@ -45,9 +45,9 @@ func runHeadless(ctx context.Context, o headlessOptions, apply *outputApply, ch
4545 cur := &headless.Cursor{}
4646 curs.set(name, cur)
4747 opts := headless.Options{Cursor: cur, LoadCursor: loadCursor, Width: size[0], Height: size[1], ScreenshotDir: dir, HDR: o.hdr, RawHDR: o.raw, Formats: ch.formats, Log: log, NewRenderer: newRenderer, NewCaptureRenderer: newRenderer, Name: name, Presented: ch.presented, Frames: ch.frames, Captured: ch.captured}
48- 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 {
49- opts.Security, opts.SecurityChanges, opts.Instance = ch.security, secure, instance
50- return headless.Run(octx, opts, sc, cc, rc, cu, cap)
48+ started := set.start(ctx, name, func(octx context.Context, feed outputFeed) error {
49+ opts.Security, opts.SecurityChanges, opts.Instance = ch.security, feed.Security, feed.Instance
50+ return headless.Run(octx, opts, feed.Scenes, feed.Contents, feed.Returns, feed.Cursor, feed.Captures)
5151 }, &opts.SecurityEvents)
5252 if !started {
5353 return set.wait()
@@ -90,12 +90,6 @@ func runHeadless(ctx context.Context, o headlessOptions, apply *outputApply, ch
9090 apply.replySent()
9191 case heads <- headsNext:
9292 apply.headsSent()
93- case s := <-ch.scenes:
94- set.scenes(s)
95- case q := <-ch.captures:
96- set.routeCapture(q)
97- case c := <-ch.cursorChanges:
98- set.setCursor(c)
9993 case stopped := <-set.stopped:
10094 if !set.stoppedCurrent(stopped) {
10195 continue
@@ -26,7 +26,7 @@ type headlessOptions struct {
26 //26 //
27 // newRenderer is configured for virtual outputs by newVulkanRenderer.27 // newRenderer is configured for virtual outputs by newVulkanRenderer.
28 func runHeadless(ctx context.Context, o headlessOptions, apply *outputApply, ch outputChannels, curs *cursors, newRenderer func(w, h int) (ports.Renderer, error), log zerowrap.Logger) error {28 func runHeadless(ctx context.Context, o headlessOptions, apply *outputApply, ch outputChannels, curs *cursors, newRenderer func(w, h int) (ports.Renderer, error), log zerowrap.Logger) error {
29- set := newOutputSet(ctx, ch.captured)29+ set := newOutputSet(ctx, ch.captured, ch.routes())
30 set.attach = ch.content30 set.attach = ch.content
31 set.wireSecurity(ch)31 set.wireSecurity(ch)
32 inventory := ports.OutputHeads{}32 inventory := ports.OutputHeads{}
@@ -45,9 +45,9 @@ func runHeadless(ctx context.Context, o headlessOptions, apply *outputApply, ch
45 cur := &headless.Cursor{}45 cur := &headless.Cursor{}
46 curs.set(name, cur)46 curs.set(name, cur)
47 opts := headless.Options{Cursor: cur, LoadCursor: loadCursor, Width: size[0], Height: size[1], ScreenshotDir: dir, HDR: o.hdr, RawHDR: o.raw, Formats: ch.formats, Log: log, NewRenderer: newRenderer, NewCaptureRenderer: newRenderer, Name: name, Presented: ch.presented, Frames: ch.frames, Captured: ch.captured}47 opts := headless.Options{Cursor: cur, LoadCursor: loadCursor, Width: size[0], Height: size[1], ScreenshotDir: dir, HDR: o.hdr, RawHDR: o.raw, Formats: ch.formats, Log: log, NewRenderer: newRenderer, NewCaptureRenderer: newRenderer, Name: name, Presented: ch.presented, Frames: ch.frames, Captured: ch.captured}
48- 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 {48+ started := set.start(ctx, name, func(octx context.Context, feed outputFeed) error {
49- opts.Security, opts.SecurityChanges, opts.Instance = ch.security, secure, instance49+ opts.Security, opts.SecurityChanges, opts.Instance = ch.security, feed.Security, feed.Instance
50- return headless.Run(octx, opts, sc, cc, rc, cu, cap)50+ return headless.Run(octx, opts, feed.Scenes, feed.Contents, feed.Returns, feed.Cursor, feed.Captures)
51 }, &opts.SecurityEvents)51 }, &opts.SecurityEvents)
52 if !started {52 if !started {
53 return set.wait()53 return set.wait()
@@ -90,12 +90,6 @@ func runHeadless(ctx context.Context, o headlessOptions, apply *outputApply, ch
90 apply.replySent()90 apply.replySent()
91 case heads <- headsNext:91 case heads <- headsNext:
92 apply.headsSent()92 apply.headsSent()
93- case s := <-ch.scenes:
94- set.scenes(s)
95- case q := <-ch.captures:
96- set.routeCapture(q)
97- case c := <-ch.cursorChanges:
98- set.setCursor(c)
99 case stopped := <-set.stopped:93 case stopped := <-set.stopped:
100 if !set.stoppedCurrent(stopped) {94 if !set.stoppedCurrent(stopped) {
101 continue95 continue
M internal/app/outputrouting.go +5 -0
@@ -30,3 +30,8 @@ type outputChannels struct {
3030 requests <-chan ports.OutputApply
3131 replies chan<- ports.OutputApplied
3232 }
33+
34+// routes are the inputs the output router feeds the outputs from.
35+func (ch outputChannels) routes() outputRoutes {
36+ return outputRoutes{scenes: ch.scenes, cursor: ch.cursorChanges, captures: ch.captures}
37+}
@@ -30,3 +30,8 @@ type outputChannels struct {
30 requests <-chan ports.OutputApply30 requests <-chan ports.OutputApply
31 replies chan<- ports.OutputApplied31 replies chan<- ports.OutputApplied
32 }32 }
33+
34+// routes are the inputs the output router feeds the outputs from.
35+func (ch outputChannels) routes() outputRoutes {
36+ return outputRoutes{scenes: ch.scenes, cursor: ch.cursorChanges, captures: ch.captures}
37+}
M internal/app/outputrouting_test.go +159 -39
@@ -4,32 +4,57 @@ import (
44 "context"
55 "os"
66 "testing"
7+ "time"
78
9+ "github.com/stretchr/testify/mock"
10+
11+ portsmocks "github.com/bnema/neferwl/internal/mocks/ports"
812 "github.com/bnema/neferwl/internal/ports"
913 )
1014
15+// routeInputs are the router's input channels, as wired in production.
16+type routeInputs struct {
17+ scenes chan []ports.Scene
18+ cursor chan ports.CursorChange
19+ captures chan ports.CaptureRequest
20+}
21+
22+func newRouteInputs() routeInputs {
23+ return routeInputs{scenes: make(chan []ports.Scene), cursor: make(chan ports.CursorChange), captures: make(chan ports.CaptureRequest)}
24+}
25+
26+func (in routeInputs) routes() outputRoutes {
27+ return outputRoutes{scenes: in.scenes, cursor: in.cursor, captures: in.captures}
28+}
29+
30+// idleRun is an output that runs until stopped.
31+func idleRun(ctx context.Context, _ outputFeed) error {
32+ <-ctx.Done()
33+ return nil
34+}
35+
1136 func TestOutputRouting(t *testing.T) {
1237 ctx, cancel := context.WithCancel(context.Background())
1338 defer cancel()
1439 replies := make(chan ports.CaptureDone, 2)
15- set := newOutputSet(ctx, replies)
40+ in := newRouteInputs()
41+ set := newOutputSet(ctx, replies, in.routes())
1642 for _, name := range []string{"A", "B"} {
17- set.start(ctx, name, func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {
18- <-ctx.Done()
19- return nil
20- })
43+ set.start(ctx, name, idleRun)
2144 <-set.outs[name].cursor // initial cursor
2245 }
23- set.scenes([]ports.Scene{{Output: "A", Seq: 1}, {Output: "B", Seq: 2}})
24- set.scenes([]ports.Scene{{Output: "A", Seq: 3}})
46+ in.scenes <- []ports.Scene{{Output: "A", Seq: 1}, {Output: "B", Seq: 2}}
47+ in.scenes <- []ports.Scene{{Output: "A", Seq: 3}}
48+ in.cursor <- ports.CursorChange{} // the router took both scene sets
2549 if got := <-set.outs["A"].scenes; got.Seq != 3 {
2650 t.Fatalf("A scene: %+v", got)
2751 }
2852 if got := <-set.outs["B"].scenes; got.Seq != 2 {
2953 t.Fatalf("B scene: %+v", got)
3054 }
31- set.setCursor(ports.CursorChange{Shape: "text"})
32- set.setCursor(ports.CursorChange{Shape: "pointer"})
55+ in.cursor <- ports.CursorChange{Shape: "text"}
56+ in.cursor <- ports.CursorChange{Shape: "pointer"}
57+ in.scenes <- nil // the router took both cursors
3358 for _, name := range []string{"A", "B"} {
3459 if got := <-set.outs[name].cursor; got.Shape != "pointer" {
3560 t.Fatalf("%s cursor: %+v", name, got)
@@ -39,64 +64,159 @@ func TestOutputRouting(t *testing.T) {
3964 if err != nil {
4065 t.Fatal(err)
4166 }
42- set.routeCapture(ports.CaptureRequest{ID: 7, Output: "B", Dst: ports.SHMBuffer{File: f}})
67+ in.captures <- ports.CaptureRequest{ID: 7, Output: "B", Dst: ports.SHMBuffer{File: f}}
4368 if got := <-set.outs["B"].captures; got.ID != 7 {
4469 t.Fatalf("B capture: %+v", got)
4570 }
4671 if err := f.Close(); err != nil {
4772 t.Fatal(err)
4873 }
49- set.outs["A"].stop()
74+ a := set.outs["A"]
75+ a.stop()
5076 if err := set.finish("A"); err != nil {
5177 t.Fatal(err)
5278 }
53- set.scenes([]ports.Scene{{Output: "A", Seq: 4}})
54- if _, ok := set.outs["A"]; ok {
55- t.Fatal("removed output retained")
56- }
57- set.routeCapture(ports.CaptureRequest{ID: 8, Output: "A"})
79+ in.scenes <- []ports.Scene{{Output: "A", Seq: 4}}
80+ in.captures <- ports.CaptureRequest{ID: 8, Output: "A"}
5881 if got := <-replies; got.ID != 8 || got.Err == nil {
5982 t.Fatalf("removed output capture: %+v", got)
6083 }
84+ select {
85+ case sc := <-a.scenes:
86+ t.Fatalf("removed output fed a scene: %+v", sc)
87+ default:
88+ }
89+ if err := set.wait(); err != nil {
90+ t.Fatal(err)
91+ }
92+}
93+
94+// An output started after a scene turned its name off starts off; one
95+// turned back on starts on.
96+func TestOutputStartsOffAfterOffScene(t *testing.T) {
97+ ctx, cancel := context.WithCancel(context.Background())
98+ defer cancel()
99+ in := newRouteInputs()
100+ set := newOutputSet(ctx, nil, in.routes())
101+ startOff := make(chan bool, 1)
102+ run := func(ctx context.Context, feed outputFeed) error {
103+ startOff <- feed.StartOff
104+ <-ctx.Done()
105+ return nil
106+ }
107+ in.scenes <- []ports.Scene{{Output: "A", Off: true}, {Output: "B"}}
108+ for _, tc := range []struct {
109+ name string
110+ want bool
111+ }{{"A", true}, {"B", false}} {
112+ set.start(ctx, tc.name, run)
113+ if got := <-startOff; got != tc.want {
114+ t.Fatalf("%s start off %v, want %v", tc.name, got, tc.want)
115+ }
116+ }
117+ in.scenes <- []ports.Scene{{Output: "A"}}
118+ set.outs["A"].stop()
119+ if err := set.finish("A"); err != nil {
120+ t.Fatal(err)
121+ }
122+ set.start(ctx, "A", run)
123+ if <-startOff {
124+ t.Fatal("A starts off after a scene turned it on")
125+ }
61126 if err := set.wait(); err != nil {
62127 t.Fatal(err)
63128 }
64129 }
65130
66-// The set remembers which outputs scenes turned off, also once they are
67-// gone: drm starts an output that reconnects off (Output.StartOff).
68-func TestOutputSetRemembersOff(t *testing.T) {
69- set := newOutputSet(context.Background(), nil)
70- set.scenes([]ports.Scene{{Output: "A", Off: true}, {Output: "B"}})
71- if !set.off["A"] || set.off["B"] {
72- t.Fatal(set.off)
73- }
74- set.scenes([]ports.Scene{{Output: "A"}})
75- if len(set.off) != 0 {
76- t.Fatal(set.off)
131+// A late unregister of a previous lifetime never removes its successor.
132+func TestRouterUnregisterKeepsNewerLifetime(t *testing.T) {
133+ r := newOutputRouter(context.Background(), nil, outputRoutes{})
134+ newer := &mailbox{instance: 2, scenes: make(chan ports.Scene, 1), cursor: make(chan ports.CursorChange, 1)}
135+ r.register("A", newer)
136+ r.unregister("A", 1)
137+ if r.outs["A"] != newer {
138+ t.Fatal("stale unregister removed the newer lifetime")
139+ }
140+ r.unregister("A", 2)
141+ if _, ok := r.outs["A"]; ok {
142+ t.Fatal("unregister kept the lifetime")
143+ }
144+}
145+
146+// Lifecycle work on the backend loop never holds up routing: while a new
147+// output's content attach blocks (it waits for the display goroutine), the
148+// running output still gets its scenes and cursor.
149+func TestBlockedStartDoesNotDelayRouting(t *testing.T) {
150+ ctx, cancel := context.WithCancel(context.Background())
151+ defer cancel()
152+ in := newRouteInputs()
153+ set := newOutputSet(ctx, nil, in.routes())
154+ set.start(ctx, "A", idleRun)
155+ <-set.outs["A"].cursor
156+ attacher := newMockcontentAttacher(t)
157+ set.attach = attacher
158+ release := make(chan struct{})
159+ attaching := make(chan struct{})
160+ ret := portsmocks.NewMockContentReturn(t)
161+ attacher.EXPECT().AttachOutput("B").RunAndReturn(func(string) (<-chan *ports.ContentBatch, ports.ContentReturn) {
162+ close(attaching)
163+ <-release
164+ return nil, ret
165+ }).Once()
166+ attacher.EXPECT().DetachOutput(mock.Anything).Maybe()
167+ started := make(chan bool)
168+ go func() { started <- set.start(ctx, "B", idleRun) }()
169+ <-attaching
170+ a := set.outs["A"]
171+ deadline := time.After(2 * time.Second)
172+ for seq := uint64(1); seq <= 3; seq++ {
173+ select {
174+ case in.scenes <- []ports.Scene{{Output: "A", Seq: seq}}:
175+ case <-deadline:
176+ t.Fatal("router blocked behind a starting output")
177+ }
178+ select {
179+ case got := <-a.scenes:
180+ if got.Seq != seq {
181+ t.Fatalf("A scene %d, want %d", got.Seq, seq)
182+ }
183+ case <-deadline:
184+ t.Fatal("scene not routed while another output starts")
185+ }
186+ }
187+ select {
188+ case in.cursor <- ports.CursorChange{Shape: "text"}:
189+ case <-deadline:
190+ t.Fatal("cursor blocked behind a starting output")
191+ }
192+ if got := <-a.cursor; got.Shape != "text" {
193+ t.Fatalf("A cursor %+v", got)
194+ }
195+ close(release)
196+ if !<-started {
197+ t.Fatal("B did not start")
198+ }
199+ if err := set.wait(); err != nil {
200+ t.Fatal(err)
77201 }
78202 }
79203
80204 func TestOutputRoutingAllocations(t *testing.T) {
81- set := newOutputSet(context.Background(), nil)
205+ r := newOutputRouter(context.Background(), nil, outputRoutes{})
82206 ctx, cancel := context.WithCancel(context.Background())
83207 defer cancel()
84- set.start(ctx, "A", func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {
85- <-ctx.Done()
86- return nil
87- })
208+ m := newMailbox(1, ctx, cancel, make(chan struct{}))
209+ r.register("A", m)
210+ <-m.cursor
88211 scenes := []ports.Scene{{Output: "A"}}
89212 cursor := ports.CursorChange{Shape: "pointer"}
90213 capture := ports.CaptureRequest{Output: "A"}
91214 if got := testing.AllocsPerRun(100, func() {
92- set.scenes(scenes)
93- set.setCursor(cursor)
94- set.routeCapture(capture)
95- <-set.outs["A"].captures
215+ r.scenes(scenes)
216+ r.setCursor(cursor)
217+ r.routeCapture(capture)
218+ <-m.captures
96219 }); got != 0 {
97220 t.Fatalf("routing: %g allocations per run, want 0", got)
98221 }
99- if err := set.wait(); err != nil {
100- t.Fatal(err)
101- }
102222 }
@@ -4,32 +4,57 @@ import (
4 "context"4 "context"
5 "os"5 "os"
6 "testing"6 "testing"
7+ "time"
7 8
9+ "github.com/stretchr/testify/mock"
10+
11+ portsmocks "github.com/bnema/neferwl/internal/mocks/ports"
8 "github.com/bnema/neferwl/internal/ports"12 "github.com/bnema/neferwl/internal/ports"
9 )13 )
10 14
15+// routeInputs are the router's input channels, as wired in production.
16+type routeInputs struct {
17+ scenes chan []ports.Scene
18+ cursor chan ports.CursorChange
19+ captures chan ports.CaptureRequest
20+}
21+
22+func newRouteInputs() routeInputs {
23+ return routeInputs{scenes: make(chan []ports.Scene), cursor: make(chan ports.CursorChange), captures: make(chan ports.CaptureRequest)}
24+}
25+
26+func (in routeInputs) routes() outputRoutes {
27+ return outputRoutes{scenes: in.scenes, cursor: in.cursor, captures: in.captures}
28+}
29+
30+// idleRun is an output that runs until stopped.
31+func idleRun(ctx context.Context, _ outputFeed) error {
32+ <-ctx.Done()
33+ return nil
34+}
35+
11 func TestOutputRouting(t *testing.T) {36 func TestOutputRouting(t *testing.T) {
12 ctx, cancel := context.WithCancel(context.Background())37 ctx, cancel := context.WithCancel(context.Background())
13 defer cancel()38 defer cancel()
14 replies := make(chan ports.CaptureDone, 2)39 replies := make(chan ports.CaptureDone, 2)
15- set := newOutputSet(ctx, replies)40+ in := newRouteInputs()
41+ set := newOutputSet(ctx, replies, in.routes())
16 for _, name := range []string{"A", "B"} {42 for _, name := range []string{"A", "B"} {
17- set.start(ctx, name, func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {43+ set.start(ctx, name, idleRun)
18- <-ctx.Done()
19- return nil
20- })
21 <-set.outs[name].cursor // initial cursor44 <-set.outs[name].cursor // initial cursor
22 }45 }
23- set.scenes([]ports.Scene{{Output: "A", Seq: 1}, {Output: "B", Seq: 2}})46+ in.scenes <- []ports.Scene{{Output: "A", Seq: 1}, {Output: "B", Seq: 2}}
24- set.scenes([]ports.Scene{{Output: "A", Seq: 3}})47+ in.scenes <- []ports.Scene{{Output: "A", Seq: 3}}
48+ in.cursor <- ports.CursorChange{} // the router took both scene sets
25 if got := <-set.outs["A"].scenes; got.Seq != 3 {49 if got := <-set.outs["A"].scenes; got.Seq != 3 {
26 t.Fatalf("A scene: %+v", got)50 t.Fatalf("A scene: %+v", got)
27 }51 }
28 if got := <-set.outs["B"].scenes; got.Seq != 2 {52 if got := <-set.outs["B"].scenes; got.Seq != 2 {
29 t.Fatalf("B scene: %+v", got)53 t.Fatalf("B scene: %+v", got)
30 }54 }
31- set.setCursor(ports.CursorChange{Shape: "text"})55+ in.cursor <- ports.CursorChange{Shape: "text"}
32- set.setCursor(ports.CursorChange{Shape: "pointer"})56+ in.cursor <- ports.CursorChange{Shape: "pointer"}
57+ in.scenes <- nil // the router took both cursors
33 for _, name := range []string{"A", "B"} {58 for _, name := range []string{"A", "B"} {
34 if got := <-set.outs[name].cursor; got.Shape != "pointer" {59 if got := <-set.outs[name].cursor; got.Shape != "pointer" {
35 t.Fatalf("%s cursor: %+v", name, got)60 t.Fatalf("%s cursor: %+v", name, got)
@@ -39,64 +64,159 @@ func TestOutputRouting(t *testing.T) {
39 if err != nil {64 if err != nil {
40 t.Fatal(err)65 t.Fatal(err)
41 }66 }
42- set.routeCapture(ports.CaptureRequest{ID: 7, Output: "B", Dst: ports.SHMBuffer{File: f}})67+ in.captures <- ports.CaptureRequest{ID: 7, Output: "B", Dst: ports.SHMBuffer{File: f}}
43 if got := <-set.outs["B"].captures; got.ID != 7 {68 if got := <-set.outs["B"].captures; got.ID != 7 {
44 t.Fatalf("B capture: %+v", got)69 t.Fatalf("B capture: %+v", got)
45 }70 }
46 if err := f.Close(); err != nil {71 if err := f.Close(); err != nil {
47 t.Fatal(err)72 t.Fatal(err)
48 }73 }
49- set.outs["A"].stop()74+ a := set.outs["A"]
75+ a.stop()
50 if err := set.finish("A"); err != nil {76 if err := set.finish("A"); err != nil {
51 t.Fatal(err)77 t.Fatal(err)
52 }78 }
53- set.scenes([]ports.Scene{{Output: "A", Seq: 4}})79+ in.scenes <- []ports.Scene{{Output: "A", Seq: 4}}
54- if _, ok := set.outs["A"]; ok {80+ in.captures <- ports.CaptureRequest{ID: 8, Output: "A"}
55- t.Fatal("removed output retained")
56- }
57- set.routeCapture(ports.CaptureRequest{ID: 8, Output: "A"})
58 if got := <-replies; got.ID != 8 || got.Err == nil {81 if got := <-replies; got.ID != 8 || got.Err == nil {
59 t.Fatalf("removed output capture: %+v", got)82 t.Fatalf("removed output capture: %+v", got)
60 }83 }
84+ select {
85+ case sc := <-a.scenes:
86+ t.Fatalf("removed output fed a scene: %+v", sc)
87+ default:
88+ }
89+ if err := set.wait(); err != nil {
90+ t.Fatal(err)
91+ }
92+}
93+
94+// An output started after a scene turned its name off starts off; one
95+// turned back on starts on.
96+func TestOutputStartsOffAfterOffScene(t *testing.T) {
97+ ctx, cancel := context.WithCancel(context.Background())
98+ defer cancel()
99+ in := newRouteInputs()
100+ set := newOutputSet(ctx, nil, in.routes())
101+ startOff := make(chan bool, 1)
102+ run := func(ctx context.Context, feed outputFeed) error {
103+ startOff <- feed.StartOff
104+ <-ctx.Done()
105+ return nil
106+ }
107+ in.scenes <- []ports.Scene{{Output: "A", Off: true}, {Output: "B"}}
108+ for _, tc := range []struct {
109+ name string
110+ want bool
111+ }{{"A", true}, {"B", false}} {
112+ set.start(ctx, tc.name, run)
113+ if got := <-startOff; got != tc.want {
114+ t.Fatalf("%s start off %v, want %v", tc.name, got, tc.want)
115+ }
116+ }
117+ in.scenes <- []ports.Scene{{Output: "A"}}
118+ set.outs["A"].stop()
119+ if err := set.finish("A"); err != nil {
120+ t.Fatal(err)
121+ }
122+ set.start(ctx, "A", run)
123+ if <-startOff {
124+ t.Fatal("A starts off after a scene turned it on")
125+ }
61 if err := set.wait(); err != nil {126 if err := set.wait(); err != nil {
62 t.Fatal(err)127 t.Fatal(err)
63 }128 }
64 }129 }
65 130
66-// The set remembers which outputs scenes turned off, also once they are131+// A late unregister of a previous lifetime never removes its successor.
67-// gone: drm starts an output that reconnects off (Output.StartOff).132+func TestRouterUnregisterKeepsNewerLifetime(t *testing.T) {
68-func TestOutputSetRemembersOff(t *testing.T) {133+ r := newOutputRouter(context.Background(), nil, outputRoutes{})
69- set := newOutputSet(context.Background(), nil)134+ newer := &mailbox{instance: 2, scenes: make(chan ports.Scene, 1), cursor: make(chan ports.CursorChange, 1)}
70- set.scenes([]ports.Scene{{Output: "A", Off: true}, {Output: "B"}})135+ r.register("A", newer)
71- if !set.off["A"] || set.off["B"] {136+ r.unregister("A", 1)
72- t.Fatal(set.off)137+ if r.outs["A"] != newer {
73- }138+ t.Fatal("stale unregister removed the newer lifetime")
74- set.scenes([]ports.Scene{{Output: "A"}})139+ }
75- if len(set.off) != 0 {140+ r.unregister("A", 2)
76- t.Fatal(set.off)141+ if _, ok := r.outs["A"]; ok {
142+ t.Fatal("unregister kept the lifetime")
143+ }
144+}
145+
146+// Lifecycle work on the backend loop never holds up routing: while a new
147+// output's content attach blocks (it waits for the display goroutine), the
148+// running output still gets its scenes and cursor.
149+func TestBlockedStartDoesNotDelayRouting(t *testing.T) {
150+ ctx, cancel := context.WithCancel(context.Background())
151+ defer cancel()
152+ in := newRouteInputs()
153+ set := newOutputSet(ctx, nil, in.routes())
154+ set.start(ctx, "A", idleRun)
155+ <-set.outs["A"].cursor
156+ attacher := newMockcontentAttacher(t)
157+ set.attach = attacher
158+ release := make(chan struct{})
159+ attaching := make(chan struct{})
160+ ret := portsmocks.NewMockContentReturn(t)
161+ attacher.EXPECT().AttachOutput("B").RunAndReturn(func(string) (<-chan *ports.ContentBatch, ports.ContentReturn) {
162+ close(attaching)
163+ <-release
164+ return nil, ret
165+ }).Once()
166+ attacher.EXPECT().DetachOutput(mock.Anything).Maybe()
167+ started := make(chan bool)
168+ go func() { started <- set.start(ctx, "B", idleRun) }()
169+ <-attaching
170+ a := set.outs["A"]
171+ deadline := time.After(2 * time.Second)
172+ for seq := uint64(1); seq <= 3; seq++ {
173+ select {
174+ case in.scenes <- []ports.Scene{{Output: "A", Seq: seq}}:
175+ case <-deadline:
176+ t.Fatal("router blocked behind a starting output")
177+ }
178+ select {
179+ case got := <-a.scenes:
180+ if got.Seq != seq {
181+ t.Fatalf("A scene %d, want %d", got.Seq, seq)
182+ }
183+ case <-deadline:
184+ t.Fatal("scene not routed while another output starts")
185+ }
186+ }
187+ select {
188+ case in.cursor <- ports.CursorChange{Shape: "text"}:
189+ case <-deadline:
190+ t.Fatal("cursor blocked behind a starting output")
191+ }
192+ if got := <-a.cursor; got.Shape != "text" {
193+ t.Fatalf("A cursor %+v", got)
194+ }
195+ close(release)
196+ if !<-started {
197+ t.Fatal("B did not start")
198+ }
199+ if err := set.wait(); err != nil {
200+ t.Fatal(err)
77 }201 }
78 }202 }
79 203
80 func TestOutputRoutingAllocations(t *testing.T) {204 func TestOutputRoutingAllocations(t *testing.T) {
81- set := newOutputSet(context.Background(), nil)205+ r := newOutputRouter(context.Background(), nil, outputRoutes{})
82 ctx, cancel := context.WithCancel(context.Background())206 ctx, cancel := context.WithCancel(context.Background())
83 defer cancel()207 defer cancel()
84- set.start(ctx, "A", func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {208+ m := newMailbox(1, ctx, cancel, make(chan struct{}))
85- <-ctx.Done()209+ r.register("A", m)
86- return nil210+ <-m.cursor
87- })
88 scenes := []ports.Scene{{Output: "A"}}211 scenes := []ports.Scene{{Output: "A"}}
89 cursor := ports.CursorChange{Shape: "pointer"}212 cursor := ports.CursorChange{Shape: "pointer"}
90 capture := ports.CaptureRequest{Output: "A"}213 capture := ports.CaptureRequest{Output: "A"}
91 if got := testing.AllocsPerRun(100, func() {214 if got := testing.AllocsPerRun(100, func() {
92- set.scenes(scenes)215+ r.scenes(scenes)
93- set.setCursor(cursor)216+ r.setCursor(cursor)
94- set.routeCapture(capture)217+ r.routeCapture(capture)
95- <-set.outs["A"].captures218+ <-m.captures
96 }); got != 0 {219 }); got != 0 {
97 t.Fatalf("routing: %g allocations per run, want 0", got)220 t.Fatalf("routing: %g allocations per run, want 0", got)
98 }221 }
99- if err := set.wait(); err != nil {
100- t.Fatal(err)
101- }
102 }222 }
M internal/app/outputs.go +62 -86
@@ -13,7 +13,21 @@ import (
1313
1414 // outputRun drives one output until ctx ends: it renders the scenes and
1515 // surface contents it receives, with the cursor the client asks for.
16-type outputRun func(ctx context.Context, scenes <-chan ports.Scene, contents <-chan *ports.ContentBatch, returns ports.ContentReturn, cursor <-chan ports.CursorChange, captures <-chan ports.CaptureRequest, security <-chan ports.SecurityState, instance ports.OutputInstance) error
16+type outputRun func(ctx context.Context, feed outputFeed) error
17+
18+// outputFeed is what an output lifetime runs on. StartOff tells that the
19+// latest scene for its name turned it off: it starts off until core's first
20+// scene (a display reconnecting from deep sleep stays dark).
21+type outputFeed struct {
22+ Scenes <-chan ports.Scene
23+ Contents <-chan *ports.ContentBatch
24+ Returns ports.ContentReturn
25+ Cursor <-chan ports.CursorChange
26+ Captures <-chan ports.CaptureRequest
27+ Security <-chan ports.SecurityState
28+ Instance ports.OutputInstance
29+ StartOff bool
30+}
1731
1832 // contentAttacher connects an output to the surface contents: wayland feeds
1933 // every attached output batches of window contents, starting with the latest
@@ -29,19 +43,20 @@ type contentAttacher interface {
2943 DetachOutput(feed ports.ContentReturn)
3044 }
3145
32-// outputSet owns the running outputs and feeds them. Core sends one scene
33-// per output; each goes to its output, latest first. Surface contents reach
34-// every output (windows move between outputs) straight from wayland through
35-// attach, which also replays the current windows to a new output.
36-// Only the goroutine running loop touches the set.
46+// outputSet owns the output lifetimes: start and stop, security
47+// registration and content feeds. Only the backend loop touches it. Scenes,
48+// cursor and captures reach the outputs through the router, its own
49+// goroutine, so lifecycle work here never delays another output's frames.
50+// Surface contents reach every output (windows move between outputs)
51+// straight from wayland through attach, which also replays the current
52+// windows to a new output.
3753 type outputSet struct {
38- outs map[string]*runningOutput
54+ outs map[string]*runningOutput
55+ router *outputRouter
56+ // stopRouter ends the router (router.done closes once it returned).
57+ stopRouter context.CancelFunc
3958 // attach is nil when no wayland server feeds contents (tests).
40- attach contentAttacher
41- // off is the latest Scene.Off per output name, kept after the output
42- // goes: it reconnects off until core's first scene (Output.StartOff).
43- off map[string]bool
44- cursor ports.CursorChange
59+ attach contentAttacher
4560 stopped chan outputStopped
4661 captured chan<- ports.CaptureDone
4762 ctx context.Context
@@ -61,24 +76,25 @@ type outputStopped struct {
6176 instance ports.OutputInstance
6277 }
6378
79+// runningOutput is one output lifetime: the mailbox the router feeds, and
80+// what the backend loop owns.
6481 type runningOutput struct {
65- instance ports.OutputInstance
82+ *mailbox
6683 security chan ports.SecurityState
6784 securityEvents chan ports.SecurityBackendEvent
68- scenes chan ports.Scene
69- cursor chan ports.CursorChange
70- captures chan ports.CaptureRequest
71- ctx context.Context
72- stop context.CancelFunc
7385 err error
74- done chan struct{}
7586 // feed is the content feed AttachOutput returned (nil: none), the handle
7687 // to detach once the output stopped.
7788 feed ports.ContentReturn
7889 }
7990
80-func newOutputSet(ctx context.Context, captured chan<- ports.CaptureDone) *outputSet {
81- return &outputSet{captured: captured, ctx: ctx, outs: map[string]*runningOutput{}, off: map[string]bool{}, stopped: make(chan outputStopped), quit: make(chan struct{})}
91+// newOutputSet returns a set whose router feeds the outputs from in, until
92+// ctx ends or wait stops it.
93+func newOutputSet(ctx context.Context, captured chan<- ports.CaptureDone, in outputRoutes) *outputSet {
94+ rctx, stopRouter := context.WithCancel(ctx)
95+ s := &outputSet{captured: captured, ctx: ctx, outs: map[string]*runningOutput{}, router: newOutputRouter(ctx, captured, in), stopRouter: stopRouter, stopped: make(chan outputStopped), quit: make(chan struct{})}
96+ go s.router.run(rctx)
97+ return s
8298 }
8399
84100 // start runs an output in its own goroutine, fed the window contents.
@@ -98,22 +114,29 @@ func (s *outputSet) start(ctx context.Context, name string, run outputRun, wire
98114 s.setSecurity(s.security.Snapshot())
99115 }
100116 octx, stop := context.WithCancel(ctx)
101- r := &runningOutput{instance: instance, security: make(chan ports.SecurityState, 1), scenes: make(chan ports.Scene, 1), cursor: make(chan ports.CursorChange, 1), captures: make(chan ports.CaptureRequest, 16), ctx: octx, stop: stop, done: make(chan struct{})}
117+ r := &runningOutput{mailbox: newMailbox(instance, octx, stop, make(chan struct{})), security: make(chan ports.SecurityState, 1)}
102118 r.security <- s.state // available before any content replay
103119 r.securityEvents = s.forwardSecurity()
104120 for _, destination := range wire {
105121 *destination = r.securityEvents
106122 }
107123 // After the security state above: the replay of the windows starts here.
108- var contents <-chan *ports.ContentBatch
109- var returns ports.ContentReturn
124+ feed := outputFeed{Scenes: r.scenes, Cursor: r.cursor, Captures: r.captures, Security: r.security, Instance: instance}
110125 if s.attach != nil {
111- contents, returns = s.attach.AttachOutput(name)
126+ feed.Contents, feed.Returns = s.attach.AttachOutput(name)
127+ }
128+ r.feed = feed.Returns
129+ // The router feeds it from now on, current cursor first.
130+ off, ok := s.router.add(name, r.mailbox)
131+ if !ok {
132+ stop()
133+ s.detach(r)
134+ return false
112135 }
113- r.feed = returns
136+ feed.StartOff = off
114137 s.outs[name] = r
115138 go func() {
116- r.err = run(octx, r.scenes, contents, returns, r.cursor, r.captures, r.security, r.instance)
139+ r.err = run(octx, feed)
117140 if r.securityEvents != nil {
118141 close(r.securityEvents)
119142 }
@@ -124,7 +147,6 @@ func (s *outputSet) start(ctx context.Context, name string, run outputRun, wire
124147 case <-s.quit:
125148 }
126149 }()
127- r.cursor <- s.cursor
128150 return true
129151 }
130152
@@ -200,11 +222,7 @@ func (s *outputSet) setSecurity(state ports.SecurityState) bool {
200222 }
201223 s.state = state
202224 for _, r := range s.outs {
203- select {
204- case <-r.security:
205- default:
206- }
207- r.security <- state
225+ latest(r.security, state)
208226 }
209227 return true
210228 }
@@ -215,36 +233,6 @@ func (s *outputSet) stoppedCurrent(stopped outputStopped) bool {
215233 return r != nil && r.instance == stopped.instance
216234 }
217235
218-// setCursor gives every output the latest cursor, replacing an unread one.
219-func (s *outputSet) setCursor(c ports.CursorChange) {
220- s.cursor = c
221- for _, r := range s.outs {
222- select {
223- case <-r.cursor:
224- default:
225- }
226- r.cursor <- c
227- }
228-}
229-
230-// scenes routes each scene to its output, replacing an unread one.
231-func (s *outputSet) scenes(set []ports.Scene) {
232- for _, sc := range set {
233- if sc.Off {
234- s.off[sc.Output] = true
235- } else {
236- delete(s.off, sc.Output)
237- }
238- if r := s.outs[sc.Output]; r != nil {
239- select {
240- case <-r.scenes:
241- default:
242- }
243- r.scenes <- sc
244- }
245- }
246-}
247-
248236 // finish forgets a stopped output and returns its error.
249237 func (s *outputSet) finish(name string) error {
250238 r := s.outs[name]
@@ -253,7 +241,9 @@ func (s *outputSet) finish(name string) error {
253241 }
254242 <-r.done
255243 r.stop()
256- s.failQueued(r)
244+ // Once the router no longer feeds it, nothing else queues a capture.
245+ s.router.remove(name, r.instance)
246+ failQueued(s.ctx, r.mailbox, s.captured)
257247 s.detach(r)
258248 delete(s.outs, name)
259249 return r.err
@@ -276,10 +266,13 @@ func (s *outputSet) wait() error {
276266 }
277267 close(s.quit)
278268 s.forwarders.Wait()
269+ // No capture is routed past this point.
270+ s.stopRouter()
271+ <-s.router.done
279272 for name, r := range s.outs {
280273 r.stop()
281274 <-r.done
282- s.failQueued(r)
275+ failQueued(s.ctx, r.mailbox, s.captured)
283276 s.detach(r)
284277 if first == nil && r.err != nil {
285278 first = r.err
@@ -427,29 +420,12 @@ func (c *cursors) move(output string, x, y float64, motion bool) {
427420 }
428421
429422 // routeCapture transfers the request to its output or closes it on failure.
430-func (s *outputSet) routeCapture(req ports.CaptureRequest) {
431- r := s.outs[req.Output]
432- if r == nil {
433- capture.Fail(s.ctx, req, fmt.Errorf("output %q unavailable", req.Output), s.captured)
434- return
435- }
436- select {
437- case r.captures <- req:
438- case <-r.done:
439- r.stop()
440- capture.Fail(s.ctx, req, fmt.Errorf("output %q stopped", req.Output), s.captured)
441- case <-r.ctx.Done():
442- capture.Fail(s.ctx, req, fmt.Errorf("output %q stopped", req.Output), s.captured)
443- default:
444- capture.Fail(s.ctx, req, fmt.Errorf("output %q capture queue full", req.Output), s.captured)
445- }
446-}
447-
448-func (s *outputSet) failQueued(r *runningOutput) {
423+// failQueued fails the captures left in a stopped output's mailbox.
424+func failQueued(ctx context.Context, m *mailbox, captured chan<- ports.CaptureDone) {
449425 for {
450426 select {
451- case req := <-r.captures:
452- capture.Fail(s.ctx, req, capture.ErrOutputStopped, s.captured)
427+ case req := <-m.captures:
428+ capture.Fail(ctx, req, capture.ErrOutputStopped, captured)
453429 default:
454430 return
455431 }
@@ -13,7 +13,21 @@ import (
13 13
14 // outputRun drives one output until ctx ends: it renders the scenes and14 // outputRun drives one output until ctx ends: it renders the scenes and
15 // surface contents it receives, with the cursor the client asks for.15 // surface contents it receives, with the cursor the client asks for.
16-type outputRun func(ctx context.Context, scenes <-chan ports.Scene, contents <-chan *ports.ContentBatch, returns ports.ContentReturn, cursor <-chan ports.CursorChange, captures <-chan ports.CaptureRequest, security <-chan ports.SecurityState, instance ports.OutputInstance) error16+type outputRun func(ctx context.Context, feed outputFeed) error
17+
18+// outputFeed is what an output lifetime runs on. StartOff tells that the
19+// latest scene for its name turned it off: it starts off until core's first
20+// scene (a display reconnecting from deep sleep stays dark).
21+type outputFeed struct {
22+ Scenes <-chan ports.Scene
23+ Contents <-chan *ports.ContentBatch
24+ Returns ports.ContentReturn
25+ Cursor <-chan ports.CursorChange
26+ Captures <-chan ports.CaptureRequest
27+ Security <-chan ports.SecurityState
28+ Instance ports.OutputInstance
29+ StartOff bool
30+}
17 31
18 // contentAttacher connects an output to the surface contents: wayland feeds32 // contentAttacher connects an output to the surface contents: wayland feeds
19 // every attached output batches of window contents, starting with the latest33 // every attached output batches of window contents, starting with the latest
@@ -29,19 +43,20 @@ type contentAttacher interface {
29 DetachOutput(feed ports.ContentReturn)43 DetachOutput(feed ports.ContentReturn)
30 }44 }
31 45
32-// outputSet owns the running outputs and feeds them. Core sends one scene46+// outputSet owns the output lifetimes: start and stop, security
33-// per output; each goes to its output, latest first. Surface contents reach47+// registration and content feeds. Only the backend loop touches it. Scenes,
34-// every output (windows move between outputs) straight from wayland through48+// cursor and captures reach the outputs through the router, its own
35-// attach, which also replays the current windows to a new output.49+// goroutine, so lifecycle work here never delays another output's frames.
36-// Only the goroutine running loop touches the set.50+// Surface contents reach every output (windows move between outputs)
51+// straight from wayland through attach, which also replays the current
52+// windows to a new output.
37 type outputSet struct {53 type outputSet struct {
38- outs map[string]*runningOutput54+ outs map[string]*runningOutput
55+ router *outputRouter
56+ // stopRouter ends the router (router.done closes once it returned).
57+ stopRouter context.CancelFunc
39 // attach is nil when no wayland server feeds contents (tests).58 // attach is nil when no wayland server feeds contents (tests).
40- attach contentAttacher59+ attach contentAttacher
41- // off is the latest Scene.Off per output name, kept after the output
42- // goes: it reconnects off until core's first scene (Output.StartOff).
43- off map[string]bool
44- cursor ports.CursorChange
45 stopped chan outputStopped60 stopped chan outputStopped
46 captured chan<- ports.CaptureDone61 captured chan<- ports.CaptureDone
47 ctx context.Context62 ctx context.Context
@@ -61,24 +76,25 @@ type outputStopped struct {
61 instance ports.OutputInstance76 instance ports.OutputInstance
62 }77 }
63 78
79+// runningOutput is one output lifetime: the mailbox the router feeds, and
80+// what the backend loop owns.
64 type runningOutput struct {81 type runningOutput struct {
65- instance ports.OutputInstance82+ *mailbox
66 security chan ports.SecurityState83 security chan ports.SecurityState
67 securityEvents chan ports.SecurityBackendEvent84 securityEvents chan ports.SecurityBackendEvent
68- scenes chan ports.Scene
69- cursor chan ports.CursorChange
70- captures chan ports.CaptureRequest
71- ctx context.Context
72- stop context.CancelFunc
73 err error85 err error
74- done chan struct{}
75 // feed is the content feed AttachOutput returned (nil: none), the handle86 // feed is the content feed AttachOutput returned (nil: none), the handle
76 // to detach once the output stopped.87 // to detach once the output stopped.
77 feed ports.ContentReturn88 feed ports.ContentReturn
78 }89 }
79 90
80-func newOutputSet(ctx context.Context, captured chan<- ports.CaptureDone) *outputSet {91+// newOutputSet returns a set whose router feeds the outputs from in, until
81- return &outputSet{captured: captured, ctx: ctx, outs: map[string]*runningOutput{}, off: map[string]bool{}, stopped: make(chan outputStopped), quit: make(chan struct{})}92+// ctx ends or wait stops it.
93+func newOutputSet(ctx context.Context, captured chan<- ports.CaptureDone, in outputRoutes) *outputSet {
94+ rctx, stopRouter := context.WithCancel(ctx)
95+ s := &outputSet{captured: captured, ctx: ctx, outs: map[string]*runningOutput{}, router: newOutputRouter(ctx, captured, in), stopRouter: stopRouter, stopped: make(chan outputStopped), quit: make(chan struct{})}
96+ go s.router.run(rctx)
97+ return s
82 }98 }
83 99
84 // start runs an output in its own goroutine, fed the window contents.100 // start runs an output in its own goroutine, fed the window contents.
@@ -98,22 +114,29 @@ func (s *outputSet) start(ctx context.Context, name string, run outputRun, wire
98 s.setSecurity(s.security.Snapshot())114 s.setSecurity(s.security.Snapshot())
99 }115 }
100 octx, stop := context.WithCancel(ctx)116 octx, stop := context.WithCancel(ctx)
101- r := &runningOutput{instance: instance, security: make(chan ports.SecurityState, 1), scenes: make(chan ports.Scene, 1), cursor: make(chan ports.CursorChange, 1), captures: make(chan ports.CaptureRequest, 16), ctx: octx, stop: stop, done: make(chan struct{})}117+ r := &runningOutput{mailbox: newMailbox(instance, octx, stop, make(chan struct{})), security: make(chan ports.SecurityState, 1)}
102 r.security <- s.state // available before any content replay118 r.security <- s.state // available before any content replay
103 r.securityEvents = s.forwardSecurity()119 r.securityEvents = s.forwardSecurity()
104 for _, destination := range wire {120 for _, destination := range wire {
105 *destination = r.securityEvents121 *destination = r.securityEvents
106 }122 }
107 // After the security state above: the replay of the windows starts here.123 // After the security state above: the replay of the windows starts here.
108- var contents <-chan *ports.ContentBatch124+ feed := outputFeed{Scenes: r.scenes, Cursor: r.cursor, Captures: r.captures, Security: r.security, Instance: instance}
109- var returns ports.ContentReturn
110 if s.attach != nil {125 if s.attach != nil {
111- contents, returns = s.attach.AttachOutput(name)126+ feed.Contents, feed.Returns = s.attach.AttachOutput(name)
127+ }
128+ r.feed = feed.Returns
129+ // The router feeds it from now on, current cursor first.
130+ off, ok := s.router.add(name, r.mailbox)
131+ if !ok {
132+ stop()
133+ s.detach(r)
134+ return false
112 }135 }
113- r.feed = returns136+ feed.StartOff = off
114 s.outs[name] = r137 s.outs[name] = r
115 go func() {138 go func() {
116- r.err = run(octx, r.scenes, contents, returns, r.cursor, r.captures, r.security, r.instance)139+ r.err = run(octx, feed)
117 if r.securityEvents != nil {140 if r.securityEvents != nil {
118 close(r.securityEvents)141 close(r.securityEvents)
119 }142 }
@@ -124,7 +147,6 @@ func (s *outputSet) start(ctx context.Context, name string, run outputRun, wire
124 case <-s.quit:147 case <-s.quit:
125 }148 }
126 }()149 }()
127- r.cursor <- s.cursor
128 return true150 return true
129 }151 }
130 152
@@ -200,11 +222,7 @@ func (s *outputSet) setSecurity(state ports.SecurityState) bool {
200 }222 }
201 s.state = state223 s.state = state
202 for _, r := range s.outs {224 for _, r := range s.outs {
203- select {225+ latest(r.security, state)
204- case <-r.security:
205- default:
206- }
207- r.security <- state
208 }226 }
209 return true227 return true
210 }228 }
@@ -215,36 +233,6 @@ func (s *outputSet) stoppedCurrent(stopped outputStopped) bool {
215 return r != nil && r.instance == stopped.instance233 return r != nil && r.instance == stopped.instance
216 }234 }
217 235
218-// setCursor gives every output the latest cursor, replacing an unread one.
219-func (s *outputSet) setCursor(c ports.CursorChange) {
220- s.cursor = c
221- for _, r := range s.outs {
222- select {
223- case <-r.cursor:
224- default:
225- }
226- r.cursor <- c
227- }
228-}
229-
230-// scenes routes each scene to its output, replacing an unread one.
231-func (s *outputSet) scenes(set []ports.Scene) {
232- for _, sc := range set {
233- if sc.Off {
234- s.off[sc.Output] = true
235- } else {
236- delete(s.off, sc.Output)
237- }
238- if r := s.outs[sc.Output]; r != nil {
239- select {
240- case <-r.scenes:
241- default:
242- }
243- r.scenes <- sc
244- }
245- }
246-}
247-
248 // finish forgets a stopped output and returns its error.236 // finish forgets a stopped output and returns its error.
249 func (s *outputSet) finish(name string) error {237 func (s *outputSet) finish(name string) error {
250 r := s.outs[name]238 r := s.outs[name]
@@ -253,7 +241,9 @@ func (s *outputSet) finish(name string) error {
253 }241 }
254 <-r.done242 <-r.done
255 r.stop()243 r.stop()
256- s.failQueued(r)244+ // Once the router no longer feeds it, nothing else queues a capture.
245+ s.router.remove(name, r.instance)
246+ failQueued(s.ctx, r.mailbox, s.captured)
257 s.detach(r)247 s.detach(r)
258 delete(s.outs, name)248 delete(s.outs, name)
259 return r.err249 return r.err
@@ -276,10 +266,13 @@ func (s *outputSet) wait() error {
276 }266 }
277 close(s.quit)267 close(s.quit)
278 s.forwarders.Wait()268 s.forwarders.Wait()
269+ // No capture is routed past this point.
270+ s.stopRouter()
271+ <-s.router.done
279 for name, r := range s.outs {272 for name, r := range s.outs {
280 r.stop()273 r.stop()
281 <-r.done274 <-r.done
282- s.failQueued(r)275+ failQueued(s.ctx, r.mailbox, s.captured)
283 s.detach(r)276 s.detach(r)
284 if first == nil && r.err != nil {277 if first == nil && r.err != nil {
285 first = r.err278 first = r.err
@@ -427,29 +420,12 @@ func (c *cursors) move(output string, x, y float64, motion bool) {
427 }420 }
428 421
429 // routeCapture transfers the request to its output or closes it on failure.422 // routeCapture transfers the request to its output or closes it on failure.
430-func (s *outputSet) routeCapture(req ports.CaptureRequest) {423+// failQueued fails the captures left in a stopped output's mailbox.
431- r := s.outs[req.Output]424+func failQueued(ctx context.Context, m *mailbox, captured chan<- ports.CaptureDone) {
432- if r == nil {
433- capture.Fail(s.ctx, req, fmt.Errorf("output %q unavailable", req.Output), s.captured)
434- return
435- }
436- select {
437- case r.captures <- req:
438- case <-r.done:
439- r.stop()
440- capture.Fail(s.ctx, req, fmt.Errorf("output %q stopped", req.Output), s.captured)
441- case <-r.ctx.Done():
442- capture.Fail(s.ctx, req, fmt.Errorf("output %q stopped", req.Output), s.captured)
443- default:
444- capture.Fail(s.ctx, req, fmt.Errorf("output %q capture queue full", req.Output), s.captured)
445- }
446-}
447-
448-func (s *outputSet) failQueued(r *runningOutput) {
449 for {425 for {
450 select {426 select {
451- case req := <-r.captures:427+ case req := <-m.captures:
452- capture.Fail(s.ctx, req, capture.ErrOutputStopped, s.captured)428+ capture.Fail(ctx, req, capture.ErrOutputStopped, captured)
453 default:429 default:
454 return430 return
455 }431 }
M internal/app/outputs_test.go +6 -4
@@ -12,12 +12,14 @@ import (
1212 func TestOutputSetCursor(t *testing.T) {
1313 ctx, cancel := context.WithCancel(context.Background())
1414 defer cancel()
15- set := newOutputSet(context.Background(), nil)
15+ cursors := make(chan ports.CursorChange)
16+ set := newOutputSet(context.Background(), nil, outputRoutes{cursor: cursors})
1617 got := map[string]chan ports.CursorChange{}
1718 run := func(name string) outputRun {
1819 got[name] = make(chan ports.CursorChange, 8)
1920 out := got[name]
20- return func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, cursor <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {
21+ return func(ctx context.Context, feed outputFeed) error {
22+ cursor := feed.Cursor
2123 for {
2224 select {
2325 case <-ctx.Done():
@@ -41,11 +43,11 @@ func TestOutputSetCursor(t *testing.T) {
4143 }
4244 set.start(ctx, "A", run("A"))
4345 expect("A", ports.CursorChange{})
44- set.setCursor(ports.CursorChange{Shape: "text"})
46+ cursors <- ports.CursorChange{Shape: "text"}
4547 expect("A", ports.CursorChange{Shape: "text"})
4648 set.start(ctx, "B", run("B"))
4749 expect("B", ports.CursorChange{Shape: "text"})
48- set.setCursor(ports.CursorChange{Hidden: true})
50+ cursors <- ports.CursorChange{Hidden: true}
4951 expect("A", ports.CursorChange{Hidden: true})
5052 expect("B", ports.CursorChange{Hidden: true})
5153 if err := set.wait(); err != nil {
@@ -12,12 +12,14 @@ import (
12 func TestOutputSetCursor(t *testing.T) {12 func TestOutputSetCursor(t *testing.T) {
13 ctx, cancel := context.WithCancel(context.Background())13 ctx, cancel := context.WithCancel(context.Background())
14 defer cancel()14 defer cancel()
15- set := newOutputSet(context.Background(), nil)15+ cursors := make(chan ports.CursorChange)
16+ set := newOutputSet(context.Background(), nil, outputRoutes{cursor: cursors})
16 got := map[string]chan ports.CursorChange{}17 got := map[string]chan ports.CursorChange{}
17 run := func(name string) outputRun {18 run := func(name string) outputRun {
18 got[name] = make(chan ports.CursorChange, 8)19 got[name] = make(chan ports.CursorChange, 8)
19 out := got[name]20 out := got[name]
20- return func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, cursor <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {21+ return func(ctx context.Context, feed outputFeed) error {
22+ cursor := feed.Cursor
21 for {23 for {
22 select {24 select {
23 case <-ctx.Done():25 case <-ctx.Done():
@@ -41,11 +43,11 @@ func TestOutputSetCursor(t *testing.T) {
41 }43 }
42 set.start(ctx, "A", run("A"))44 set.start(ctx, "A", run("A"))
43 expect("A", ports.CursorChange{})45 expect("A", ports.CursorChange{})
44- set.setCursor(ports.CursorChange{Shape: "text"})46+ cursors <- ports.CursorChange{Shape: "text"}
45 expect("A", ports.CursorChange{Shape: "text"})47 expect("A", ports.CursorChange{Shape: "text"})
46 set.start(ctx, "B", run("B"))48 set.start(ctx, "B", run("B"))
47 expect("B", ports.CursorChange{Shape: "text"})49 expect("B", ports.CursorChange{Shape: "text"})
48- set.setCursor(ports.CursorChange{Hidden: true})50+ cursors <- ports.CursorChange{Hidden: true}
49 expect("A", ports.CursorChange{Hidden: true})51 expect("A", ports.CursorChange{Hidden: true})
50 expect("B", ports.CursorChange{Hidden: true})52 expect("B", ports.CursorChange{Hidden: true})
51 if err := set.wait(); err != nil {53 if err := set.wait(); err != nil {
A internal/app/router.go +180 -0
@@ -0,0 +1,180 @@
1+package app
2+
3+import (
4+ "context"
5+ "fmt"
6+
7+ "github.com/bnema/neferwl/internal/adapters/outputkit/capture"
8+ "github.com/bnema/neferwl/internal/ports"
9+)
10+
11+// outputRouter feeds the running outputs their scenes, cursor and capture
12+// requests. It runs in its own goroutine and never blocks: every mailbox is
13+// latest-wins (scenes, cursor) or fails fast when full (captures). Output
14+// lifecycle (scan, modeset, start and stop, security, leases) runs on the
15+// backend loop, so a hotplug or a configuration on one output never delays
16+// another output's frames or cursor.
17+//
18+// The backend loop registers an output before it runs and unregisters it
19+// once it stopped, over ops; registration is lossless and ordered, and an
20+// output's instance keys it so a late unregister never removes its
21+// successor.
22+type outputRouter struct {
23+ outs map[string]*mailbox
24+ // off is the latest Scene.Off per output name, kept after the output
25+ // goes: it reconnects off until core's first scene (drm StartOff).
26+ off map[string]bool
27+ cursor ports.CursorChange
28+ captured chan<- ports.CaptureDone
29+ ctx context.Context
30+
31+ in outputRoutes
32+ ops chan routerOp
33+ done chan struct{} // closed when run returns
34+}
35+
36+// outputRoutes are the router's inputs: core's scenes, wayland's cursor and
37+// capture requests.
38+type outputRoutes struct {
39+ scenes <-chan []ports.Scene
40+ cursor <-chan ports.CursorChange
41+ captures <-chan ports.CaptureRequest
42+}
43+
44+// mailbox is what the router feeds one output lifetime.
45+type mailbox struct {
46+ instance ports.OutputInstance
47+ scenes chan ports.Scene
48+ cursor chan ports.CursorChange
49+ captures chan ports.CaptureRequest
50+ // ctx ends and done closes with the output; stop cancels it.
51+ ctx context.Context
52+ stop context.CancelFunc
53+ done chan struct{}
54+}
55+
56+func newMailbox(instance ports.OutputInstance, ctx context.Context, stop context.CancelFunc, done chan struct{}) *mailbox {
57+ return &mailbox{instance: instance, scenes: make(chan ports.Scene, 1), cursor: make(chan ports.CursorChange, 1), captures: make(chan ports.CaptureRequest, 16), ctx: ctx, stop: stop, done: done}
58+}
59+
60+// routerOp registers (add set) or unregisters an output. reply carries the
61+// registration's start-off state; it is closed after an unregister.
62+type routerOp struct {
63+ name string
64+ add *mailbox
65+ gone ports.OutputInstance
66+ reply chan bool
67+}
68+
69+func newOutputRouter(ctx context.Context, captured chan<- ports.CaptureDone, in outputRoutes) *outputRouter {
70+ return &outputRouter{outs: map[string]*mailbox{}, off: map[string]bool{}, captured: captured, ctx: ctx, in: in, ops: make(chan routerOp), done: make(chan struct{})}
71+}
72+
73+// run routes until ctx ends.
74+func (r *outputRouter) run(ctx context.Context) {
75+ defer close(r.done)
76+ for {
77+ select {
78+ case <-ctx.Done():
79+ return
80+ case op := <-r.ops:
81+ r.apply(op)
82+ case set := <-r.in.scenes:
83+ r.scenes(set)
84+ case c := <-r.in.cursor:
85+ r.setCursor(c)
86+ case q := <-r.in.captures:
87+ r.routeCapture(q)
88+ }
89+ }
90+}
91+
92+func (r *outputRouter) apply(op routerOp) {
93+ if op.add != nil {
94+ op.reply <- r.register(op.name, op.add)
95+ return
96+ }
97+ r.unregister(op.name, op.gone)
98+ close(op.reply)
99+}
100+
101+// register feeds m from now on, starting with the current cursor, and
102+// reports whether the output's latest scene turned it off.
103+func (r *outputRouter) register(name string, m *mailbox) bool {
104+ r.outs[name] = m
105+ m.cursor <- r.cursor
106+ return r.off[name]
107+}
108+
109+// unregister stops feeding the output lifetime instance of name.
110+func (r *outputRouter) unregister(name string, instance ports.OutputInstance) {
111+ if m := r.outs[name]; m != nil && m.instance == instance {
112+ delete(r.outs, name)
113+ }
114+}
115+
116+// setCursor gives every output the latest cursor, replacing an unread one.
117+func (r *outputRouter) setCursor(c ports.CursorChange) {
118+ r.cursor = c
119+ for _, m := range r.outs {
120+ latest(m.cursor, c)
121+ }
122+}
123+
124+// scenes routes each scene to its output, replacing an unread one.
125+func (r *outputRouter) scenes(set []ports.Scene) {
126+ for _, sc := range set {
127+ if sc.Off {
128+ r.off[sc.Output] = true
129+ } else {
130+ delete(r.off, sc.Output)
131+ }
132+ if m := r.outs[sc.Output]; m != nil {
133+ latest(m.scenes, sc)
134+ }
135+ }
136+}
137+
138+// routeCapture hands the request to its output or fails it at once: a full
139+// or stopped output never holds up the router.
140+func (r *outputRouter) routeCapture(req ports.CaptureRequest) {
141+ m := r.outs[req.Output]
142+ if m == nil {
143+ capture.Fail(r.ctx, req, fmt.Errorf("output %q unavailable", req.Output), r.captured)
144+ return
145+ }
146+ select {
147+ case m.captures <- req:
148+ case <-m.done:
149+ m.stop()
150+ capture.Fail(r.ctx, req, fmt.Errorf("output %q stopped", req.Output), r.captured)
151+ case <-m.ctx.Done():
152+ capture.Fail(r.ctx, req, fmt.Errorf("output %q stopped", req.Output), r.captured)
153+ default:
154+ capture.Fail(r.ctx, req, fmt.Errorf("output %q capture queue full", req.Output), r.captured)
155+ }
156+}
157+
158+// add registers m for name and returns its start-off state. Backend loop
159+// only; false once the router has stopped.
160+func (r *outputRouter) add(name string, m *mailbox) (off, ok bool) {
161+ reply := make(chan bool, 1)
162+ select {
163+ case r.ops <- routerOp{name: name, add: m, reply: reply}:
164+ return <-reply, true
165+ case <-r.done:
166+ return false, false
167+ }
168+}
169+
170+// remove unregisters the lifetime instance of name and returns once the
171+// router no longer feeds it, so its queued captures can be failed. Backend
172+// loop only.
173+func (r *outputRouter) remove(name string, instance ports.OutputInstance) {
174+ reply := make(chan bool)
175+ select {
176+ case r.ops <- routerOp{name: name, gone: instance, reply: reply}:
177+ <-reply
178+ case <-r.done:
179+ }
180+}
@@ -0,0 +1,180 @@
1+package app
2+
3+import (
4+ "context"
5+ "fmt"
6+
7+ "github.com/bnema/neferwl/internal/adapters/outputkit/capture"
8+ "github.com/bnema/neferwl/internal/ports"
9+)
10+
11+// outputRouter feeds the running outputs their scenes, cursor and capture
12+// requests. It runs in its own goroutine and never blocks: every mailbox is
13+// latest-wins (scenes, cursor) or fails fast when full (captures). Output
14+// lifecycle (scan, modeset, start and stop, security, leases) runs on the
15+// backend loop, so a hotplug or a configuration on one output never delays
16+// another output's frames or cursor.
17+//
18+// The backend loop registers an output before it runs and unregisters it
19+// once it stopped, over ops; registration is lossless and ordered, and an
20+// output's instance keys it so a late unregister never removes its
21+// successor.
22+type outputRouter struct {
23+ outs map[string]*mailbox
24+ // off is the latest Scene.Off per output name, kept after the output
25+ // goes: it reconnects off until core's first scene (drm StartOff).
26+ off map[string]bool
27+ cursor ports.CursorChange
28+ captured chan<- ports.CaptureDone
29+ ctx context.Context
30+
31+ in outputRoutes
32+ ops chan routerOp
33+ done chan struct{} // closed when run returns
34+}
35+
36+// outputRoutes are the router's inputs: core's scenes, wayland's cursor and
37+// capture requests.
38+type outputRoutes struct {
39+ scenes <-chan []ports.Scene
40+ cursor <-chan ports.CursorChange
41+ captures <-chan ports.CaptureRequest
42+}
43+
44+// mailbox is what the router feeds one output lifetime.
45+type mailbox struct {
46+ instance ports.OutputInstance
47+ scenes chan ports.Scene
48+ cursor chan ports.CursorChange
49+ captures chan ports.CaptureRequest
50+ // ctx ends and done closes with the output; stop cancels it.
51+ ctx context.Context
52+ stop context.CancelFunc
53+ done chan struct{}
54+}
55+
56+func newMailbox(instance ports.OutputInstance, ctx context.Context, stop context.CancelFunc, done chan struct{}) *mailbox {
57+ return &mailbox{instance: instance, scenes: make(chan ports.Scene, 1), cursor: make(chan ports.CursorChange, 1), captures: make(chan ports.CaptureRequest, 16), ctx: ctx, stop: stop, done: done}
58+}
59+
60+// routerOp registers (add set) or unregisters an output. reply carries the
61+// registration's start-off state; it is closed after an unregister.
62+type routerOp struct {
63+ name string
64+ add *mailbox
65+ gone ports.OutputInstance
66+ reply chan bool
67+}
68+
69+func newOutputRouter(ctx context.Context, captured chan<- ports.CaptureDone, in outputRoutes) *outputRouter {
70+ return &outputRouter{outs: map[string]*mailbox{}, off: map[string]bool{}, captured: captured, ctx: ctx, in: in, ops: make(chan routerOp), done: make(chan struct{})}
71+}
72+
73+// run routes until ctx ends.
74+func (r *outputRouter) run(ctx context.Context) {
75+ defer close(r.done)
76+ for {
77+ select {
78+ case <-ctx.Done():
79+ return
80+ case op := <-r.ops:
81+ r.apply(op)
82+ case set := <-r.in.scenes:
83+ r.scenes(set)
84+ case c := <-r.in.cursor:
85+ r.setCursor(c)
86+ case q := <-r.in.captures:
87+ r.routeCapture(q)
88+ }
89+ }
90+}
91+
92+func (r *outputRouter) apply(op routerOp) {
93+ if op.add != nil {
94+ op.reply <- r.register(op.name, op.add)
95+ return
96+ }
97+ r.unregister(op.name, op.gone)
98+ close(op.reply)
99+}
100+
101+// register feeds m from now on, starting with the current cursor, and
102+// reports whether the output's latest scene turned it off.
103+func (r *outputRouter) register(name string, m *mailbox) bool {
104+ r.outs[name] = m
105+ m.cursor <- r.cursor
106+ return r.off[name]
107+}
108+
109+// unregister stops feeding the output lifetime instance of name.
110+func (r *outputRouter) unregister(name string, instance ports.OutputInstance) {
111+ if m := r.outs[name]; m != nil && m.instance == instance {
112+ delete(r.outs, name)
113+ }
114+}
115+
116+// setCursor gives every output the latest cursor, replacing an unread one.
117+func (r *outputRouter) setCursor(c ports.CursorChange) {
118+ r.cursor = c
119+ for _, m := range r.outs {
120+ latest(m.cursor, c)
121+ }
122+}
123+
124+// scenes routes each scene to its output, replacing an unread one.
125+func (r *outputRouter) scenes(set []ports.Scene) {
126+ for _, sc := range set {
127+ if sc.Off {
128+ r.off[sc.Output] = true
129+ } else {
130+ delete(r.off, sc.Output)
131+ }
132+ if m := r.outs[sc.Output]; m != nil {
133+ latest(m.scenes, sc)
134+ }
135+ }
136+}
137+
138+// routeCapture hands the request to its output or fails it at once: a full
139+// or stopped output never holds up the router.
140+func (r *outputRouter) routeCapture(req ports.CaptureRequest) {
141+ m := r.outs[req.Output]
142+ if m == nil {
143+ capture.Fail(r.ctx, req, fmt.Errorf("output %q unavailable", req.Output), r.captured)
144+ return
145+ }
146+ select {
147+ case m.captures <- req:
148+ case <-m.done:
149+ m.stop()
150+ capture.Fail(r.ctx, req, fmt.Errorf("output %q stopped", req.Output), r.captured)
151+ case <-m.ctx.Done():
152+ capture.Fail(r.ctx, req, fmt.Errorf("output %q stopped", req.Output), r.captured)
153+ default:
154+ capture.Fail(r.ctx, req, fmt.Errorf("output %q capture queue full", req.Output), r.captured)
155+ }
156+}
157+
158+// add registers m for name and returns its start-off state. Backend loop
159+// only; false once the router has stopped.
160+func (r *outputRouter) add(name string, m *mailbox) (off, ok bool) {
161+ reply := make(chan bool, 1)
162+ select {
163+ case r.ops <- routerOp{name: name, add: m, reply: reply}:
164+ return <-reply, true
165+ case <-r.done:
166+ return false, false
167+ }
168+}
169+
170+// remove unregisters the lifetime instance of name and returns once the
171+// router no longer feeds it, so its queued captures can be failed. Backend
172+// loop only.
173+func (r *outputRouter) remove(name string, instance ports.OutputInstance) {
174+ reply := make(chan bool)
175+ select {
176+ case r.ops <- routerOp{name: name, gone: instance, reply: reply}:
177+ <-reply
178+ case <-r.done:
179+ }
180+}
M internal/app/security_duplicate_test.go +3 -3
@@ -25,9 +25,9 @@ func TestSecurityDuplicatePhysicalLifetimeRequiresConfirmedInactive(t *testing.T
2525 t.Fatal(err)
2626 }
2727 events := make(chan ports.SecurityBackendEvent, 4)
28- set := newOutputSet(context.Background(), nil)
28+ set := newOutputSet(context.Background(), nil, outputRoutes{})
2929 set.wireSecurity(outputChannels{security: gate, securityEvents: events})
30- run := func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {
30+ run := func(ctx context.Context, feed outputFeed) error {
3131 <-ctx.Done()
3232 return nil
3333 }
@@ -93,7 +93,7 @@ func TestSecurityDuplicatePhysicalLifetimeRequiresConfirmedInactive(t *testing.T
9393
9494 func TestSecurityDiscardedLifetimeCounterExhaustion(t *testing.T) {
9595 events := make(chan ports.SecurityBackendEvent, 2)
96- set := newOutputSet(context.Background(), nil)
96+ set := newOutputSet(context.Background(), nil, outputRoutes{})
9797 set.wireSecurity(outputChannels{securityEvents: events})
9898 set.nextInstance = ^ports.OutputInstance(0) - 1
9999 instance, err := registerDiscardedOutput(set, "DP-1")
@@ -25,9 +25,9 @@ func TestSecurityDuplicatePhysicalLifetimeRequiresConfirmedInactive(t *testing.T
25 t.Fatal(err)25 t.Fatal(err)
26 }26 }
27 events := make(chan ports.SecurityBackendEvent, 4)27 events := make(chan ports.SecurityBackendEvent, 4)
28- set := newOutputSet(context.Background(), nil)28+ set := newOutputSet(context.Background(), nil, outputRoutes{})
29 set.wireSecurity(outputChannels{security: gate, securityEvents: events})29 set.wireSecurity(outputChannels{security: gate, securityEvents: events})
30- run := func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {30+ run := func(ctx context.Context, feed outputFeed) error {
31 <-ctx.Done()31 <-ctx.Done()
32 return nil32 return nil
33 }33 }
@@ -93,7 +93,7 @@ func TestSecurityDuplicatePhysicalLifetimeRequiresConfirmedInactive(t *testing.T
93 93
94 func TestSecurityDiscardedLifetimeCounterExhaustion(t *testing.T) {94 func TestSecurityDiscardedLifetimeCounterExhaustion(t *testing.T) {
95 events := make(chan ports.SecurityBackendEvent, 2)95 events := make(chan ports.SecurityBackendEvent, 2)
96- set := newOutputSet(context.Background(), nil)96+ set := newOutputSet(context.Background(), nil, outputRoutes{})
97 set.wireSecurity(outputChannels{securityEvents: events})97 set.wireSecurity(outputChannels{securityEvents: events})
98 set.nextInstance = ^ports.OutputInstance(0) - 198 set.nextInstance = ^ports.OutputInstance(0) - 1
99 instance, err := registerDiscardedOutput(set, "DP-1")99 instance, err := registerDiscardedOutput(set, "DP-1")
M internal/app/security_hotplug_test.go +12 -8
@@ -45,7 +45,7 @@ func TestSecurityHotplugLifetimeAndProofOrdering(t *testing.T) {
4545 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
4646 defer cancel()
4747 events := make(chan ports.SecurityBackendEvent, 8)
48- set := newOutputSet(ctx, nil)
48+ set := newOutputSet(ctx, nil, outputRoutes{})
4949 set.wireSecurity(outputChannels{securityEvents: events})
5050 defer func() {
5151 if err := set.wait(); err != nil {
@@ -66,7 +66,8 @@ func TestSecurityHotplugLifetimeAndProofOrdering(t *testing.T) {
6666 commands := make(chan []ports.SecurityBackendEvent, 1)
6767 closed := make(chan bool, 1)
6868 var ownerEvents chan<- ports.SecurityBackendEvent
69- if !set.start(ctx, "DP-1", func(octx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, security <-chan ports.SecurityState, instance ports.OutputInstance) error {
69+ if !set.start(ctx, "DP-1", func(octx context.Context, feed outputFeed) error {
70+ security, instance := feed.Security, feed.Instance
7071 defer func() { closed <- closeSuccess }()
7172 state := <-security
7273 proof := ports.SecurityOutputProof{Proof: ports.OutputProtection{Generation: state.Generation, Instance: instance, Kind: ports.ProtectionProtectedFrame}}
@@ -193,7 +194,7 @@ func securityOwnerFIFOSaturationDrainsOnNormalStop(t *testing.T) {
193194 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
194195 defer cancel()
195196 events := make(chan ports.SecurityBackendEvent) // block display after registration
196- set := newOutputSet(ctx, nil)
197+ set := newOutputSet(ctx, nil, outputRoutes{})
197198 set.wireSecurity(outputChannels{securityEvents: events})
198199 defer func() {
199200 if err := set.wait(); err != nil {
@@ -205,7 +206,8 @@ func securityOwnerFIFOSaturationDrainsOnNormalStop(t *testing.T) {
205206 startDone := make(chan bool, 1)
206207 var ownerEvents chan<- ports.SecurityBackendEvent
207208 go func() {
208- startDone <- set.start(ctx, "DP-1", func(octx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, instance ports.OutputInstance) error {
209+ startDone <- set.start(ctx, "DP-1", func(octx context.Context, feed outputFeed) error {
210+ instance := feed.Instance
209211 for i := range 40 {
210212 var event ports.SecurityBackendEvent = ports.SecurityOutputInvalidated{Generation: generation, Instance: instance}
211213 if i%2 == 0 {
@@ -285,7 +287,7 @@ func TestSecurityOwnerBufferedEvidenceDrainsAfterFinish(t *testing.T) {
285287 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
286288 defer cancel()
287289 events := make(chan ports.SecurityBackendEvent)
288- set := newOutputSet(ctx, nil)
290+ set := newOutputSet(ctx, nil, outputRoutes{})
289291 set.wireSecurity(outputChannels{securityEvents: events})
290292 defer func() {
291293 if err := set.wait(); err != nil {
@@ -295,7 +297,8 @@ func TestSecurityOwnerBufferedEvidenceDrainsAfterFinish(t *testing.T) {
295297 startDone := make(chan bool, 1)
296298 var ownerEvents chan<- ports.SecurityBackendEvent
297299 go func() {
298- startDone <- set.start(ctx, "DP-1", func(octx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, instance ports.OutputInstance) error {
300+ startDone <- set.start(ctx, "DP-1", func(octx context.Context, feed outputFeed) error {
301+ instance := feed.Instance
299302 for i := range 16 {
300303 event := ports.SecurityOutputInvalidated{Generation: ports.LockGeneration(i + 1), Instance: instance}
301304 select {
@@ -343,14 +346,15 @@ func securityOwnerFIFOSaturationCancelReleasesOwnerAndForwarders(t *testing.T) {
343346 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
344347 defer cancel()
345348 events := make(chan ports.SecurityBackendEvent)
346- set := newOutputSet(ctx, nil)
349+ set := newOutputSet(ctx, nil, outputRoutes{})
347350 set.wireSecurity(outputChannels{securityEvents: events})
348351 accepted := make(chan int, 40)
349352 ownerDone := make(chan struct{})
350353 startDone := make(chan bool, 1)
351354 var ownerEvents chan<- ports.SecurityBackendEvent
352355 go func() {
353- startDone <- set.start(ctx, "DP-1", func(octx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, instance ports.OutputInstance) error {
356+ startDone <- set.start(ctx, "DP-1", func(octx context.Context, feed outputFeed) error {
357+ instance := feed.Instance
354358 defer close(ownerDone)
355359 for i := range 40 {
356360 select {
@@ -45,7 +45,7 @@ func TestSecurityHotplugLifetimeAndProofOrdering(t *testing.T) {
45 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)45 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
46 defer cancel()46 defer cancel()
47 events := make(chan ports.SecurityBackendEvent, 8)47 events := make(chan ports.SecurityBackendEvent, 8)
48- set := newOutputSet(ctx, nil)48+ set := newOutputSet(ctx, nil, outputRoutes{})
49 set.wireSecurity(outputChannels{securityEvents: events})49 set.wireSecurity(outputChannels{securityEvents: events})
50 defer func() {50 defer func() {
51 if err := set.wait(); err != nil {51 if err := set.wait(); err != nil {
@@ -66,7 +66,8 @@ func TestSecurityHotplugLifetimeAndProofOrdering(t *testing.T) {
66 commands := make(chan []ports.SecurityBackendEvent, 1)66 commands := make(chan []ports.SecurityBackendEvent, 1)
67 closed := make(chan bool, 1)67 closed := make(chan bool, 1)
68 var ownerEvents chan<- ports.SecurityBackendEvent68 var ownerEvents chan<- ports.SecurityBackendEvent
69- if !set.start(ctx, "DP-1", func(octx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, security <-chan ports.SecurityState, instance ports.OutputInstance) error {69+ if !set.start(ctx, "DP-1", func(octx context.Context, feed outputFeed) error {
70+ security, instance := feed.Security, feed.Instance
70 defer func() { closed <- closeSuccess }()71 defer func() { closed <- closeSuccess }()
71 state := <-security72 state := <-security
72 proof := ports.SecurityOutputProof{Proof: ports.OutputProtection{Generation: state.Generation, Instance: instance, Kind: ports.ProtectionProtectedFrame}}73 proof := ports.SecurityOutputProof{Proof: ports.OutputProtection{Generation: state.Generation, Instance: instance, Kind: ports.ProtectionProtectedFrame}}
@@ -193,7 +194,7 @@ func securityOwnerFIFOSaturationDrainsOnNormalStop(t *testing.T) {
193 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)194 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
194 defer cancel()195 defer cancel()
195 events := make(chan ports.SecurityBackendEvent) // block display after registration196 events := make(chan ports.SecurityBackendEvent) // block display after registration
196- set := newOutputSet(ctx, nil)197+ set := newOutputSet(ctx, nil, outputRoutes{})
197 set.wireSecurity(outputChannels{securityEvents: events})198 set.wireSecurity(outputChannels{securityEvents: events})
198 defer func() {199 defer func() {
199 if err := set.wait(); err != nil {200 if err := set.wait(); err != nil {
@@ -205,7 +206,8 @@ func securityOwnerFIFOSaturationDrainsOnNormalStop(t *testing.T) {
205 startDone := make(chan bool, 1)206 startDone := make(chan bool, 1)
206 var ownerEvents chan<- ports.SecurityBackendEvent207 var ownerEvents chan<- ports.SecurityBackendEvent
207 go func() {208 go func() {
208- startDone <- set.start(ctx, "DP-1", func(octx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, instance ports.OutputInstance) error {209+ startDone <- set.start(ctx, "DP-1", func(octx context.Context, feed outputFeed) error {
210+ instance := feed.Instance
209 for i := range 40 {211 for i := range 40 {
210 var event ports.SecurityBackendEvent = ports.SecurityOutputInvalidated{Generation: generation, Instance: instance}212 var event ports.SecurityBackendEvent = ports.SecurityOutputInvalidated{Generation: generation, Instance: instance}
211 if i%2 == 0 {213 if i%2 == 0 {
@@ -285,7 +287,7 @@ func TestSecurityOwnerBufferedEvidenceDrainsAfterFinish(t *testing.T) {
285 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)287 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
286 defer cancel()288 defer cancel()
287 events := make(chan ports.SecurityBackendEvent)289 events := make(chan ports.SecurityBackendEvent)
288- set := newOutputSet(ctx, nil)290+ set := newOutputSet(ctx, nil, outputRoutes{})
289 set.wireSecurity(outputChannels{securityEvents: events})291 set.wireSecurity(outputChannels{securityEvents: events})
290 defer func() {292 defer func() {
291 if err := set.wait(); err != nil {293 if err := set.wait(); err != nil {
@@ -295,7 +297,8 @@ func TestSecurityOwnerBufferedEvidenceDrainsAfterFinish(t *testing.T) {
295 startDone := make(chan bool, 1)297 startDone := make(chan bool, 1)
296 var ownerEvents chan<- ports.SecurityBackendEvent298 var ownerEvents chan<- ports.SecurityBackendEvent
297 go func() {299 go func() {
298- startDone <- set.start(ctx, "DP-1", func(octx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, instance ports.OutputInstance) error {300+ startDone <- set.start(ctx, "DP-1", func(octx context.Context, feed outputFeed) error {
301+ instance := feed.Instance
299 for i := range 16 {302 for i := range 16 {
300 event := ports.SecurityOutputInvalidated{Generation: ports.LockGeneration(i + 1), Instance: instance}303 event := ports.SecurityOutputInvalidated{Generation: ports.LockGeneration(i + 1), Instance: instance}
301 select {304 select {
@@ -343,14 +346,15 @@ func securityOwnerFIFOSaturationCancelReleasesOwnerAndForwarders(t *testing.T) {
343 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)346 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
344 defer cancel()347 defer cancel()
345 events := make(chan ports.SecurityBackendEvent)348 events := make(chan ports.SecurityBackendEvent)
346- set := newOutputSet(ctx, nil)349+ set := newOutputSet(ctx, nil, outputRoutes{})
347 set.wireSecurity(outputChannels{securityEvents: events})350 set.wireSecurity(outputChannels{securityEvents: events})
348 accepted := make(chan int, 40)351 accepted := make(chan int, 40)
349 ownerDone := make(chan struct{})352 ownerDone := make(chan struct{})
350 startDone := make(chan bool, 1)353 startDone := make(chan bool, 1)
351 var ownerEvents chan<- ports.SecurityBackendEvent354 var ownerEvents chan<- ports.SecurityBackendEvent
352 go func() {355 go func() {
353- startDone <- set.start(ctx, "DP-1", func(octx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, instance ports.OutputInstance) error {356+ startDone <- set.start(ctx, "DP-1", func(octx context.Context, feed outputFeed) error {
357+ instance := feed.Instance
354 defer close(ownerDone)358 defer close(ownerDone)
355 for i := range 40 {359 for i := range 40 {
356 select {360 select {
M internal/app/security_test.go +9 -8
@@ -15,13 +15,14 @@ func TestSecurityRegistrationBeforeStart(t *testing.T) {
1515 ctx, cancel := context.WithCancel(context.Background())
1616 defer cancel()
1717 events := make(chan ports.SecurityBackendEvent)
18- set := newOutputSet(ctx, nil)
18+ set := newOutputSet(ctx, nil, outputRoutes{})
1919 set.wireSecurity(outputChannels{securityEvents: events})
2020 set.setSecurity(ports.SecurityState{Generation: 7, Protected: true})
2121 started := make(chan ports.OutputInstance, 1)
2222 result := make(chan bool, 1)
2323 go func() {
24- result <- set.start(ctx, "DP-1", func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, security <-chan ports.SecurityState, instance ports.OutputInstance) error {
24+ result <- set.start(ctx, "DP-1", func(ctx context.Context, feed outputFeed) error {
25+ security, instance := feed.Security, feed.Instance
2526 state := <-security
2627 if state != (ports.SecurityState{Generation: 7, Protected: true}) {
2728 return errors.New("missing startup protection")
@@ -65,9 +66,9 @@ func TestSecurityRegistrationBeforeStart(t *testing.T) {
6566 func TestSecurityReplacementAndStaleState(t *testing.T) {
6667 ctx := context.Background()
6768 events := make(chan ports.SecurityBackendEvent, 4)
68- set := newOutputSet(ctx, nil)
69+ set := newOutputSet(ctx, nil, outputRoutes{})
6970 set.wireSecurity(outputChannels{securityEvents: events})
70- run := func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {
71+ run := func(ctx context.Context, feed outputFeed) error {
7172 <-ctx.Done()
7273 return nil
7374 }
@@ -163,11 +164,11 @@ func TestSecurityRevokesAllLeasesAndRetainsFailures(t *testing.T) {
163164 }
164165
165166 func TestSecurityInstanceExhaustion(t *testing.T) {
166- set := newOutputSet(context.Background(), nil)
167+ set := newOutputSet(context.Background(), nil, outputRoutes{})
167168 events := make(chan ports.SecurityBackendEvent, 2)
168169 set.wireSecurity(outputChannels{securityEvents: events})
169170 set.nextInstance = ^ports.OutputInstance(0) - 1
170- run := func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {
171+ run := func(ctx context.Context, feed outputFeed) error {
171172 <-ctx.Done()
172173 return nil
173174 }
@@ -200,11 +201,11 @@ func TestSecurityInstanceExhaustion(t *testing.T) {
200201 func TestSecurityFullEventsAllowLeaseBarrier(t *testing.T) {
201202 ctx, cancel := context.WithCancel(context.Background())
202203 defer cancel()
203- set := newOutputSet(ctx, nil)
204+ set := newOutputSet(ctx, nil, outputRoutes{})
204205 events := make(chan ports.SecurityBackendEvent, 1)
205206 set.wireSecurity(outputChannels{securityEvents: events})
206207 var ownerEvents chan<- ports.SecurityBackendEvent
207- set.start(ctx, "DP-1", func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {
208+ set.start(ctx, "DP-1", func(ctx context.Context, feed outputFeed) error {
208209 <-ctx.Done()
209210 return nil
210211 }, &ownerEvents)
@@ -15,13 +15,14 @@ func TestSecurityRegistrationBeforeStart(t *testing.T) {
15 ctx, cancel := context.WithCancel(context.Background())15 ctx, cancel := context.WithCancel(context.Background())
16 defer cancel()16 defer cancel()
17 events := make(chan ports.SecurityBackendEvent)17 events := make(chan ports.SecurityBackendEvent)
18- set := newOutputSet(ctx, nil)18+ set := newOutputSet(ctx, nil, outputRoutes{})
19 set.wireSecurity(outputChannels{securityEvents: events})19 set.wireSecurity(outputChannels{securityEvents: events})
20 set.setSecurity(ports.SecurityState{Generation: 7, Protected: true})20 set.setSecurity(ports.SecurityState{Generation: 7, Protected: true})
21 started := make(chan ports.OutputInstance, 1)21 started := make(chan ports.OutputInstance, 1)
22 result := make(chan bool, 1)22 result := make(chan bool, 1)
23 go func() {23 go func() {
24- result <- set.start(ctx, "DP-1", func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, security <-chan ports.SecurityState, instance ports.OutputInstance) error {24+ result <- set.start(ctx, "DP-1", func(ctx context.Context, feed outputFeed) error {
25+ security, instance := feed.Security, feed.Instance
25 state := <-security26 state := <-security
26 if state != (ports.SecurityState{Generation: 7, Protected: true}) {27 if state != (ports.SecurityState{Generation: 7, Protected: true}) {
27 return errors.New("missing startup protection")28 return errors.New("missing startup protection")
@@ -65,9 +66,9 @@ func TestSecurityRegistrationBeforeStart(t *testing.T) {
65 func TestSecurityReplacementAndStaleState(t *testing.T) {66 func TestSecurityReplacementAndStaleState(t *testing.T) {
66 ctx := context.Background()67 ctx := context.Background()
67 events := make(chan ports.SecurityBackendEvent, 4)68 events := make(chan ports.SecurityBackendEvent, 4)
68- set := newOutputSet(ctx, nil)69+ set := newOutputSet(ctx, nil, outputRoutes{})
69 set.wireSecurity(outputChannels{securityEvents: events})70 set.wireSecurity(outputChannels{securityEvents: events})
70- run := func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {71+ run := func(ctx context.Context, feed outputFeed) error {
71 <-ctx.Done()72 <-ctx.Done()
72 return nil73 return nil
73 }74 }
@@ -163,11 +164,11 @@ func TestSecurityRevokesAllLeasesAndRetainsFailures(t *testing.T) {
163 }164 }
164 165
165 func TestSecurityInstanceExhaustion(t *testing.T) {166 func TestSecurityInstanceExhaustion(t *testing.T) {
166- set := newOutputSet(context.Background(), nil)167+ set := newOutputSet(context.Background(), nil, outputRoutes{})
167 events := make(chan ports.SecurityBackendEvent, 2)168 events := make(chan ports.SecurityBackendEvent, 2)
168 set.wireSecurity(outputChannels{securityEvents: events})169 set.wireSecurity(outputChannels{securityEvents: events})
169 set.nextInstance = ^ports.OutputInstance(0) - 1170 set.nextInstance = ^ports.OutputInstance(0) - 1
170- run := func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {171+ run := func(ctx context.Context, feed outputFeed) error {
171 <-ctx.Done()172 <-ctx.Done()
172 return nil173 return nil
173 }174 }
@@ -200,11 +201,11 @@ func TestSecurityInstanceExhaustion(t *testing.T) {
200 func TestSecurityFullEventsAllowLeaseBarrier(t *testing.T) {201 func TestSecurityFullEventsAllowLeaseBarrier(t *testing.T) {
201 ctx, cancel := context.WithCancel(context.Background())202 ctx, cancel := context.WithCancel(context.Background())
202 defer cancel()203 defer cancel()
203- set := newOutputSet(ctx, nil)204+ set := newOutputSet(ctx, nil, outputRoutes{})
204 events := make(chan ports.SecurityBackendEvent, 1)205 events := make(chan ports.SecurityBackendEvent, 1)
205 set.wireSecurity(outputChannels{securityEvents: events})206 set.wireSecurity(outputChannels{securityEvents: events})
206 var ownerEvents chan<- ports.SecurityBackendEvent207 var ownerEvents chan<- ports.SecurityBackendEvent
207- set.start(ctx, "DP-1", func(ctx context.Context, _ <-chan ports.Scene, _ <-chan *ports.ContentBatch, _ ports.ContentReturn, _ <-chan ports.CursorChange, _ <-chan ports.CaptureRequest, _ <-chan ports.SecurityState, _ ports.OutputInstance) error {208+ set.start(ctx, "DP-1", func(ctx context.Context, feed outputFeed) error {
208 <-ctx.Done()209 <-ctx.Done()
209 return nil210 return nil
210 }, &ownerEvents)211 }, &ownerEvents)

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 25

Tags 10

↑↓ moveenter openesc close