detached 4a05b71

fix(screencast): close streams on shutdown, scope screenshot jobs to the attempt, pause without ending

- Shutdown (context end or bus loss) pumps every stream, including those
  without a source yet: CastClose goes out, a frame in flight is awaited,
  then the source is destroyed and its buffers released. On timeout the
  leaked source keeps its handler state and the session table is replaced,
  so reconnects do not exhaust the cast and source tables.
- Screenshot jobs get a per-attempt context, carry a job ID so a stale
  result completes nothing, and are cancelled and joined before the state
  is reset.
- A paused PipeWire node (consumer left) gets no frames and keeps its
  session, as in Mutter; the first frame is forced on STREAMING.
- A clean nefercap exit is a success only with a regular, non-empty file
  (Lstat, no symlink follow); otherwise the reply is (false, "").
commit
4a05b71aa477724bc454405be9c9c85ce6780f1c
author
brice · · 2026-10-14 21:30 UTC
committer
brice · · 2026-10-14 21:39 UTC
parent
fe541fa
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
9 files · +625 -78
Unified
UnifiedSplit
M docs/capture.md +1 -1
@@ -67,7 +67,7 @@ NeferWL serves the GNOME ScreenCast, DisplayConfig, Introspect and Screenshot D-
6767
6868 - **Screen sharing** in browsers and apps: install `xdg-desktop-portal-gnome` and run PipeWire. The portal dialog lists monitors and windows with their titles; the stream is a PipeWire video node. Without PipeWire, the ScreenCast interface is not offered.
6969 - **Screenshots** through the portal, interactive or not: install [nefercap](https://github.com/bnema/nefercap), which writes the PNG.
70-- A cast keeps the indicator on until it ends. It ends when the session locks, the output or window goes away, or the client disconnects.
70+- A cast keeps the indicator on until it ends. It ends when the session locks, the output or window goes away, the client stops it or disconnects, or NeferWL loses the D-Bus connection. When the receiving program stops reading, the stream pauses (no frames are rendered) and the cast stays open for the next reader, as in GNOME.
7171 - Unsupported: area and virtual casts, and the "Select area" and "Window" choices of the screenshot dialog.
7272
7373 The packaged `neferwl-portals.conf` routes `ScreenCast` and `Screenshot` to the gnome portal.
@@ -67,7 +67,7 @@ NeferWL serves the GNOME ScreenCast, DisplayConfig, Introspect and Screenshot D-
67 67
68 - **Screen sharing** in browsers and apps: install `xdg-desktop-portal-gnome` and run PipeWire. The portal dialog lists monitors and windows with their titles; the stream is a PipeWire video node. Without PipeWire, the ScreenCast interface is not offered.68 - **Screen sharing** in browsers and apps: install `xdg-desktop-portal-gnome` and run PipeWire. The portal dialog lists monitors and windows with their titles; the stream is a PipeWire video node. Without PipeWire, the ScreenCast interface is not offered.
69 - **Screenshots** through the portal, interactive or not: install [nefercap](https://github.com/bnema/nefercap), which writes the PNG.69 - **Screenshots** through the portal, interactive or not: install [nefercap](https://github.com/bnema/nefercap), which writes the PNG.
70-- A cast keeps the indicator on until it ends. It ends when the session locks, the output or window goes away, or the client disconnects.70+- A cast keeps the indicator on until it ends. It ends when the session locks, the output or window goes away, the client stops it or disconnects, or NeferWL loses the D-Bus connection. When the receiving program stops reading, the stream pauses (no frames are rendered) and the cast stays open for the next reader, as in GNOME.
71 - Unsupported: area and virtual casts, and the "Select area" and "Window" choices of the screenshot dialog.71 - Unsupported: area and virtual casts, and the "Select area" and "Window" choices of the screenshot dialog.
72 72
73 The packaged `neferwl-portals.conf` routes `ScreenCast` and `Screenshot` to the gnome portal.73 The packaged `neferwl-portals.conf` routes `ScreenCast` and `Screenshot` to the gnome portal.
M internal/adapters/screencast/harness_test.go +54 -11
@@ -19,10 +19,11 @@ import (
1919 "github.com/stretchr/testify/mock"
2020 )
2121
22-// privateBus starts a dbus-daemon of its own, so tests never touch the
23-// desktop's session bus. CI must run these tests: there, a missing daemon
24-// fails.
25-func privateBus(t *testing.T) string {
22+// startBus starts a dbus-daemon of its own, so tests never touch the
23+// desktop's session bus, and returns its address and a function killing it,
24+// which is how a test loses the bus. CI must run these tests: there, a
25+// missing daemon fails.
26+func startBus(t *testing.T) (string, func()) {
2627 t.Helper()
2728 bin, err := exec.LookPath("dbus-daemon")
2829 if err != nil {
@@ -39,12 +40,13 @@ func privateBus(t *testing.T) string {
3940 if err := cmd.Start(); err != nil {
4041 t.Fatal(err)
4142 }
42- t.Cleanup(func() { _ = cmd.Process.Kill(); _ = cmd.Wait() })
43+ kill := func() { _ = cmd.Process.Kill() }
44+ t.Cleanup(func() { kill(); _ = cmd.Wait() })
4345 line, err := bufio.NewReader(out).ReadString('\n')
4446 if err != nil {
4547 t.Fatal(err)
4648 }
47- return strings.TrimSpace(line)
49+ return strings.TrimSpace(line), kill
4850 }
4951
5052 // logBuf is a locked bytes.Buffer, a real io.Writer.
@@ -78,6 +80,23 @@ type peers struct {
7880 pid atomic.Int64
7981 openErr error
8082 exePid func(open int) int // pid Exe reports; the opened one when nil
83+
84+ // exeFile, when set, tells the executable of one pidfd: the pidfds are
85+ // numbered in the order Open handed them out (0 is the first peer).
86+ exeFile func(n int) string
87+ mu sync.Mutex
88+ opened []*os.File
89+}
90+
91+func (p *peers) number(f *os.File) int {
92+ p.mu.Lock()
93+ defer p.mu.Unlock()
94+ for i, o := range p.opened {
95+ if o == f {
96+ return i
97+ }
98+ }
99+ return -1
81100 }
82101
83102 // svcHarness runs a Service on a private bus with mocked executables and
@@ -92,17 +111,23 @@ type svcHarness struct {
92111 run *mockrunner
93112 log *logBuf
94113 done chan struct{}
114+
115+ killBus func() // loses the bus: the daemon dies
116+ closed chan struct{} // a PipeWire connection was closed
95117 }
96118
97119 type harnessOpts struct {
98120 allow *captureallow.Policy
99121 noPW bool
100122 holdOp bool
123+ // iterate runs inside every PipeWire loop iteration, on the owner.
124+ iterate func()
101125 }
102126
103127 func newSvcHarness(t *testing.T, o harnessOpts) *svcHarness {
104128 t.Helper()
105- h := &svcHarness{t: t, addr: privateBus(t), cmds: make(chan ports.CastCommand, 64), peers: &peers{}, log: &logBuf{}, done: make(chan struct{})}
129+ h := &svcHarness{t: t, cmds: make(chan ports.CastCommand, 64), peers: &peers{}, log: &logBuf{}, done: make(chan struct{}), closed: make(chan struct{}, 8)}
130+ h.addr, h.killBus = startBus(t)
106131 h.peers.exe.Store("/usr/bin/grim")
107132 if o.holdOp {
108133 h.peers.openGo = make(chan struct{})
@@ -123,14 +148,23 @@ func newSvcHarness(t *testing.T, o harnessOpts) *svcHarness {
123148 return nil, h.peers.openErr
124149 }
125150 h.peers.pid.Store(int64(pid))
126- return os.Open(os.DevNull)
151+ f, err := os.Open(os.DevNull)
152+ if err == nil {
153+ h.peers.mu.Lock()
154+ h.peers.opened = append(h.peers.opened, f)
155+ h.peers.mu.Unlock()
156+ }
157+ return f, err
127158 }).Maybe()
128- res.EXPECT().Exe(mock.Anything).RunAndReturn(func(*os.File) (string, int, error) {
159+ res.EXPECT().Exe(mock.Anything).RunAndReturn(func(f *os.File) (string, int, error) {
129160 open := int(h.peers.pid.Load())
130161 pid := open
131162 if h.peers.exePid != nil {
132163 pid = h.peers.exePid(open)
133164 }
165+ if h.peers.exeFile != nil {
166+ return h.peers.exeFile(h.peers.number(f)), pid, nil
167+ }
134168 if h.peers.exeNow != nil {
135169 return h.peers.exeNow(), pid, nil
136170 }
@@ -151,8 +185,17 @@ func newSvcHarness(t *testing.T, o harnessOpts) *svcHarness {
151185 t.Cleanup(func() { r.Close(); w.Close() })
152186 h.core = newMockvideoCore(t)
153187 h.core.EXPECT().Fd().Return(int(r.Fd()))
154- h.core.EXPECT().Iterate().Maybe()
155- h.core.EXPECT().Close().Return().Maybe()
188+ h.core.EXPECT().Iterate().Run(func() {
189+ if o.iterate != nil {
190+ o.iterate()
191+ }
192+ }).Maybe()
193+ h.core.EXPECT().Close().Run(func() {
194+ select {
195+ case h.closed <- struct{}{}:
196+ default:
197+ }
198+ }).Maybe()
156199 pwOpen.EXPECT().Open().Return(h.core, nil)
157200 }
158201 h.svc.pw = pwOpen
@@ -19,10 +19,11 @@ import (
19 "github.com/stretchr/testify/mock"19 "github.com/stretchr/testify/mock"
20 )20 )
21 21
22-// privateBus starts a dbus-daemon of its own, so tests never touch the22+// startBus starts a dbus-daemon of its own, so tests never touch the
23-// desktop's session bus. CI must run these tests: there, a missing daemon23+// desktop's session bus, and returns its address and a function killing it,
24-// fails.24+// which is how a test loses the bus. CI must run these tests: there, a
25-func privateBus(t *testing.T) string {25+// missing daemon fails.
26+func startBus(t *testing.T) (string, func()) {
26 t.Helper()27 t.Helper()
27 bin, err := exec.LookPath("dbus-daemon")28 bin, err := exec.LookPath("dbus-daemon")
28 if err != nil {29 if err != nil {
@@ -39,12 +40,13 @@ func privateBus(t *testing.T) string {
39 if err := cmd.Start(); err != nil {40 if err := cmd.Start(); err != nil {
40 t.Fatal(err)41 t.Fatal(err)
41 }42 }
42- t.Cleanup(func() { _ = cmd.Process.Kill(); _ = cmd.Wait() })43+ kill := func() { _ = cmd.Process.Kill() }
44+ t.Cleanup(func() { kill(); _ = cmd.Wait() })
43 line, err := bufio.NewReader(out).ReadString('\n')45 line, err := bufio.NewReader(out).ReadString('\n')
44 if err != nil {46 if err != nil {
45 t.Fatal(err)47 t.Fatal(err)
46 }48 }
47- return strings.TrimSpace(line)49+ return strings.TrimSpace(line), kill
48 }50 }
49 51
50 // logBuf is a locked bytes.Buffer, a real io.Writer.52 // logBuf is a locked bytes.Buffer, a real io.Writer.
@@ -78,6 +80,23 @@ type peers struct {
78 pid atomic.Int6480 pid atomic.Int64
79 openErr error81 openErr error
80 exePid func(open int) int // pid Exe reports; the opened one when nil82 exePid func(open int) int // pid Exe reports; the opened one when nil
83+
84+ // exeFile, when set, tells the executable of one pidfd: the pidfds are
85+ // numbered in the order Open handed them out (0 is the first peer).
86+ exeFile func(n int) string
87+ mu sync.Mutex
88+ opened []*os.File
89+}
90+
91+func (p *peers) number(f *os.File) int {
92+ p.mu.Lock()
93+ defer p.mu.Unlock()
94+ for i, o := range p.opened {
95+ if o == f {
96+ return i
97+ }
98+ }
99+ return -1
81 }100 }
82 101
83 // svcHarness runs a Service on a private bus with mocked executables and102 // svcHarness runs a Service on a private bus with mocked executables and
@@ -92,17 +111,23 @@ type svcHarness struct {
92 run *mockrunner111 run *mockrunner
93 log *logBuf112 log *logBuf
94 done chan struct{}113 done chan struct{}
114+
115+ killBus func() // loses the bus: the daemon dies
116+ closed chan struct{} // a PipeWire connection was closed
95 }117 }
96 118
97 type harnessOpts struct {119 type harnessOpts struct {
98 allow *captureallow.Policy120 allow *captureallow.Policy
99 noPW bool121 noPW bool
100 holdOp bool122 holdOp bool
123+ // iterate runs inside every PipeWire loop iteration, on the owner.
124+ iterate func()
101 }125 }
102 126
103 func newSvcHarness(t *testing.T, o harnessOpts) *svcHarness {127 func newSvcHarness(t *testing.T, o harnessOpts) *svcHarness {
104 t.Helper()128 t.Helper()
105- h := &svcHarness{t: t, addr: privateBus(t), cmds: make(chan ports.CastCommand, 64), peers: &peers{}, log: &logBuf{}, done: make(chan struct{})}129+ h := &svcHarness{t: t, cmds: make(chan ports.CastCommand, 64), peers: &peers{}, log: &logBuf{}, done: make(chan struct{}), closed: make(chan struct{}, 8)}
130+ h.addr, h.killBus = startBus(t)
106 h.peers.exe.Store("/usr/bin/grim")131 h.peers.exe.Store("/usr/bin/grim")
107 if o.holdOp {132 if o.holdOp {
108 h.peers.openGo = make(chan struct{})133 h.peers.openGo = make(chan struct{})
@@ -123,14 +148,23 @@ func newSvcHarness(t *testing.T, o harnessOpts) *svcHarness {
123 return nil, h.peers.openErr148 return nil, h.peers.openErr
124 }149 }
125 h.peers.pid.Store(int64(pid))150 h.peers.pid.Store(int64(pid))
126- return os.Open(os.DevNull)151+ f, err := os.Open(os.DevNull)
152+ if err == nil {
153+ h.peers.mu.Lock()
154+ h.peers.opened = append(h.peers.opened, f)
155+ h.peers.mu.Unlock()
156+ }
157+ return f, err
127 }).Maybe()158 }).Maybe()
128- res.EXPECT().Exe(mock.Anything).RunAndReturn(func(*os.File) (string, int, error) {159+ res.EXPECT().Exe(mock.Anything).RunAndReturn(func(f *os.File) (string, int, error) {
129 open := int(h.peers.pid.Load())160 open := int(h.peers.pid.Load())
130 pid := open161 pid := open
131 if h.peers.exePid != nil {162 if h.peers.exePid != nil {
132 pid = h.peers.exePid(open)163 pid = h.peers.exePid(open)
133 }164 }
165+ if h.peers.exeFile != nil {
166+ return h.peers.exeFile(h.peers.number(f)), pid, nil
167+ }
134 if h.peers.exeNow != nil {168 if h.peers.exeNow != nil {
135 return h.peers.exeNow(), pid, nil169 return h.peers.exeNow(), pid, nil
136 }170 }
@@ -151,8 +185,17 @@ func newSvcHarness(t *testing.T, o harnessOpts) *svcHarness {
151 t.Cleanup(func() { r.Close(); w.Close() })185 t.Cleanup(func() { r.Close(); w.Close() })
152 h.core = newMockvideoCore(t)186 h.core = newMockvideoCore(t)
153 h.core.EXPECT().Fd().Return(int(r.Fd()))187 h.core.EXPECT().Fd().Return(int(r.Fd()))
154- h.core.EXPECT().Iterate().Maybe()188+ h.core.EXPECT().Iterate().Run(func() {
155- h.core.EXPECT().Close().Return().Maybe()189+ if o.iterate != nil {
190+ o.iterate()
191+ }
192+ }).Maybe()
193+ h.core.EXPECT().Close().Run(func() {
194+ select {
195+ case h.closed <- struct{}{}:
196+ default:
197+ }
198+ }).Maybe()
156 pwOpen.EXPECT().Open().Return(h.core, nil)199 pwOpen.EXPECT().Open().Return(h.core, nil)
157 }200 }
158 h.svc.pw = pwOpen201 h.svc.pw = pwOpen
M internal/adapters/screencast/producer.go +39 -14
@@ -19,6 +19,15 @@ import (
1919 // The buffer's mapping is lent to the wayland server until the CastFrameDone
2020 // of that buffer, so the source is never resized or destroyed while a frame
2121 // is in flight.
22+//
23+// A cast ends (the session signals Closed) on Stop, when the owner leaves
24+// the bus (NameOwnerChanged), when wayland stops the cast (output or window
25+// gone, lock), when the PipeWire stream goes to ERROR or back to
26+// UNCONNECTED, and when the service shuts down. It does not end when the
27+// consumer leaves: as in Mutter, the node then goes to PAUSED, no frame is
28+// requested while it is not STREAMING (a frame already in flight is
29+// released instead of queued), and the first frame after it is STREAMING
30+// again is forced. The D-Bus session stays up for the next consumer.
2231
2332 // wl_shm formats of the supported video formats.
2433 const (
@@ -104,10 +113,19 @@ func (s *Service) giveBack(st *stream) {
104113 // frameDone settles the frame in flight: a complete image is queued and
105114 // triggered, anything else gives the buffer back.
106115 func (s *Service) frameDone(st *stream, ev *ports.CastEvent) {
107- if b := s.settleFrame(st, ev); b != nil {
108- st.src.Queue(b, st.seq, st.pts, false)
109- st.src.Trigger()
116+ b := s.settleFrame(st, ev)
117+ if b == nil {
118+ return
119+ }
120+ if st.pwState != pipewire.StreamStateStreaming {
121+ // The consumer left while the frame was drawn: nobody wants it.
122+ // The picture is kept for the consumer that comes next.
123+ st.src.Release(b)
124+ st.dirty = true
125+ return
110126 }
127+ st.src.Queue(b, st.seq, st.pts, false)
128+ st.src.Trigger()
111129 }
112130
113131 // settleFrame updates the stream for the CastFrameDone of its frame and
@@ -299,14 +317,16 @@ func (s *Service) issueFrame(st *stream, b *pipewire.VideoBuffer, now time.Time)
299317 }
300318
301319 // pumpClosing finishes a stream of an ended session: CastClose, then the
302-// PipeWire node once no frame is in flight.
320+// PipeWire node once no frame is in flight. Destroying the node does not
321+// wait for CastClose to be sent (only a frame in flight lends its buffer),
322+// but the slot is kept until it is.
303323 func (s *Service) pumpClosing(st *stream) {
304324 if st.needClose {
305- if !s.sendCmd(ports.CastCommand{Kind: ports.CastClose, ID: st.cast}) {
325+ if s.sendCmd(ports.CastCommand{Kind: ports.CastClose, ID: st.cast}) {
326+ st.needClose = false
327+ } else {
306328 s.wait(retryOpen)
307- return
308329 }
309- st.needClose = false
310330 }
311331 if st.inflight {
312332 return
@@ -316,7 +336,9 @@ func (s *Service) pumpClosing(st *stream) {
316336 st.src.Destroy()
317337 st.src = nil
318338 }
319- *st = stream{}
339+ if !st.needClose {
340+ *st = stream{}
341+ }
320342 }
321343
322344 // reap frees the dying sessions once every stream released its resources.
@@ -336,10 +358,13 @@ func (s *Service) reap() {
336358 }
337359 }
338360
339-// streamsLive reports whether a stream still holds PipeWire resources or a
340-// lent buffer.
341-func (s *Service) streamsLive() bool {
342- live := false
343- s.forEachStream(func(st *stream) { live = live || st.src != nil || st.inflight })
344- return live
361+// streamCount is the number of streams that still hold something: a
362+// PipeWire source, a lent buffer or an unsent CastClose.
363+func (s *Service) streamCount() int {
364+ n := 0
365+ s.forEachStream(func(*stream) { n++ })
366+ return n
345367 }
368+
369+// streamsLive reports whether a stream still holds something to release.
370+func (s *Service) streamsLive() bool { return s.streamCount() > 0 }
@@ -19,6 +19,15 @@ import (
19 // The buffer's mapping is lent to the wayland server until the CastFrameDone19 // The buffer's mapping is lent to the wayland server until the CastFrameDone
20 // of that buffer, so the source is never resized or destroyed while a frame20 // of that buffer, so the source is never resized or destroyed while a frame
21 // is in flight.21 // is in flight.
22+//
23+// A cast ends (the session signals Closed) on Stop, when the owner leaves
24+// the bus (NameOwnerChanged), when wayland stops the cast (output or window
25+// gone, lock), when the PipeWire stream goes to ERROR or back to
26+// UNCONNECTED, and when the service shuts down. It does not end when the
27+// consumer leaves: as in Mutter, the node then goes to PAUSED, no frame is
28+// requested while it is not STREAMING (a frame already in flight is
29+// released instead of queued), and the first frame after it is STREAMING
30+// again is forced. The D-Bus session stays up for the next consumer.
22 31
23 // wl_shm formats of the supported video formats.32 // wl_shm formats of the supported video formats.
24 const (33 const (
@@ -104,10 +113,19 @@ func (s *Service) giveBack(st *stream) {
104 // frameDone settles the frame in flight: a complete image is queued and113 // frameDone settles the frame in flight: a complete image is queued and
105 // triggered, anything else gives the buffer back.114 // triggered, anything else gives the buffer back.
106 func (s *Service) frameDone(st *stream, ev *ports.CastEvent) {115 func (s *Service) frameDone(st *stream, ev *ports.CastEvent) {
107- if b := s.settleFrame(st, ev); b != nil {116+ b := s.settleFrame(st, ev)
108- st.src.Queue(b, st.seq, st.pts, false)117+ if b == nil {
109- st.src.Trigger()118+ return
119+ }
120+ if st.pwState != pipewire.StreamStateStreaming {
121+ // The consumer left while the frame was drawn: nobody wants it.
122+ // The picture is kept for the consumer that comes next.
123+ st.src.Release(b)
124+ st.dirty = true
125+ return
110 }126 }
127+ st.src.Queue(b, st.seq, st.pts, false)
128+ st.src.Trigger()
111 }129 }
112 130
113 // settleFrame updates the stream for the CastFrameDone of its frame and131 // settleFrame updates the stream for the CastFrameDone of its frame and
@@ -299,14 +317,16 @@ func (s *Service) issueFrame(st *stream, b *pipewire.VideoBuffer, now time.Time)
299 }317 }
300 318
301 // pumpClosing finishes a stream of an ended session: CastClose, then the319 // pumpClosing finishes a stream of an ended session: CastClose, then the
302-// PipeWire node once no frame is in flight.320+// PipeWire node once no frame is in flight. Destroying the node does not
321+// wait for CastClose to be sent (only a frame in flight lends its buffer),
322+// but the slot is kept until it is.
303 func (s *Service) pumpClosing(st *stream) {323 func (s *Service) pumpClosing(st *stream) {
304 if st.needClose {324 if st.needClose {
305- if !s.sendCmd(ports.CastCommand{Kind: ports.CastClose, ID: st.cast}) {325+ if s.sendCmd(ports.CastCommand{Kind: ports.CastClose, ID: st.cast}) {
326+ st.needClose = false
327+ } else {
306 s.wait(retryOpen)328 s.wait(retryOpen)
307- return
308 }329 }
309- st.needClose = false
310 }330 }
311 if st.inflight {331 if st.inflight {
312 return332 return
@@ -316,7 +336,9 @@ func (s *Service) pumpClosing(st *stream) {
316 st.src.Destroy()336 st.src.Destroy()
317 st.src = nil337 st.src = nil
318 }338 }
319- *st = stream{}339+ if !st.needClose {
340+ *st = stream{}
341+ }
320 }342 }
321 343
322 // reap frees the dying sessions once every stream released its resources.344 // reap frees the dying sessions once every stream released its resources.
@@ -336,10 +358,13 @@ func (s *Service) reap() {
336 }358 }
337 }359 }
338 360
339-// streamsLive reports whether a stream still holds PipeWire resources or a361+// streamCount is the number of streams that still hold something: a
340-// lent buffer.362+// PipeWire source, a lent buffer or an unsent CastClose.
341-func (s *Service) streamsLive() bool {363+func (s *Service) streamCount() int {
342- live := false364+ n := 0
343- s.forEachStream(func(st *stream) { live = live || st.src != nil || st.inflight })365+ s.forEachStream(func(*stream) { n++ })
344- return live366+ return n
345 }367 }
368+
369+// streamsLive reports whether a stream still holds something to release.
370+func (s *Service) streamsLive() bool { return s.streamCount() > 0 }
M internal/adapters/screencast/producer_test.go +140 -0
@@ -322,3 +322,143 @@ func TestCastProducerAllocations(t *testing.T) {
322322 t.Fatalf("%v allocations per frame, want 0", n)
323323 }
324324 }
325+
326+// The consumer leaves: the node goes to PAUSED and no frame is requested,
327+// whatever damage and time bring, while the session stays open. A frame
328+// already in flight is released, not queued. When the node streams again the
329+// first frame is forced.
330+func TestPausedNodeGetsNoFramesAndSessionStays(t *testing.T) {
331+ p := newProducer(t)
332+ p.streaming()
333+ p.src.EXPECT().Dequeue().Return(p.buf).Twice()
334+ p.s.pump(p.st, p.now)
335+ if c := p.cmd(); c.Kind != ports.CastFrame {
336+ t.Fatalf("%+v", c)
337+ }
338+ // Paused with the frame in flight: it is given back, not queued (no
339+ // Queue or Trigger expectation).
340+ p.st.h2.StateChanged(pipewire.StreamStatePaused, 42, "")
341+ p.src.EXPECT().Release(p.buf).Return().Once()
342+ p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3, Time: time.Unix(5, 0)})
343+ for range 5 {
344+ p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastDamage, ID: 1})
345+ p.now = p.now.Add(time.Second)
346+ p.s.pumpAll(p.now) // Dequeue is not expected a second time while paused
347+ p.none()
348+ }
349+ if p.s.sessions[0].dying || p.st.closing {
350+ t.Fatal("the session ended with the consumer")
351+ }
352+ if got := p.s.timeout(); got != -1 {
353+ t.Fatalf("paused stream asks for a timeout of %d ms", got)
354+ }
355+ // A consumer is back: the picture goes out at once, damage or not.
356+ p.st.h2.StateChanged(pipewire.StreamStateStreaming, 42, "")
357+ p.s.pumpAll(p.now)
358+ if c := p.cmd(); c.Kind != ports.CastFrame {
359+ t.Fatalf("no frame after the node streams again: %+v", c)
360+ }
361+}
362+
363+// Shutdown pumps every stream: CastClose goes out, the node is destroyed and
364+// PipeWire closed, with no frame in flight.
365+func TestShutdownClosesStreams(t *testing.T) {
366+ p := newProducer(t)
367+ p.streaming()
368+ var order []string
369+ p.src.EXPECT().Destroy().Run(func() { order = append(order, "destroy") }).Once()
370+ p.core.EXPECT().Close().Run(func() { order = append(order, "close") }).Once()
371+ p.s.shutdown()
372+ if c := p.cmd(); c.Kind != ports.CastClose || c.ID != 1 {
373+ t.Fatalf("%+v", c)
374+ }
375+ if len(order) != 2 || order[0] != "destroy" || order[1] != "close" || p.s.pwc != nil {
376+ t.Fatalf("order %v", order)
377+ }
378+ if p.st.used || p.s.sessions[0].used {
379+ t.Fatal("slots not freed")
380+ }
381+}
382+
383+// A stream opened on the wayland side that has no node yet is closed too.
384+func TestShutdownClosesStreamWithoutSource(t *testing.T) {
385+ p := newProducer(t)
386+ p.s.pump(p.st, p.now)
387+ if c := p.cmd(); c.Kind != ports.CastOpen {
388+ t.Fatalf("%+v", c)
389+ }
390+ p.core.EXPECT().Close().Return().Once()
391+ p.s.shutdown()
392+ if c := p.cmd(); c.Kind != ports.CastClose || c.ID != 1 {
393+ t.Fatalf("%+v", c)
394+ }
395+ if p.st.used {
396+ t.Fatal("stream not freed")
397+ }
398+}
399+
400+// A frame in flight at shutdown is waited for: nothing is destroyed before
401+// its CastFrameDone, then the buffer is released and the node destroyed.
402+func TestShutdownWaitsForFrameInFlight(t *testing.T) {
403+ p := newProducer(t)
404+ p.streaming()
405+ p.src.EXPECT().Dequeue().Return(p.buf).Once()
406+ p.s.pump(p.st, p.now)
407+ p.cmd()
408+ var order []string
409+ p.src.EXPECT().Release(p.buf).Run(func(*pipewire.VideoBuffer) { order = append(order, "release") }).Once()
410+ p.src.EXPECT().Destroy().Run(func() { order = append(order, "destroy") }).Once()
411+ p.core.EXPECT().Close().Run(func() { order = append(order, "close") }).Once()
412+ go func() {
413+ time.Sleep(30 * time.Millisecond)
414+ p.s.Push(ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3})
415+ }()
416+ start := time.Now()
417+ p.s.shutdown()
418+ if time.Since(start) < 30*time.Millisecond {
419+ t.Fatal("shutdown did not wait for the frame")
420+ }
421+ if c := p.cmd(); c.Kind != ports.CastClose {
422+ t.Fatalf("%+v", c)
423+ }
424+ if len(order) != 3 || order[0] != "release" || order[1] != "destroy" || order[2] != "close" {
425+ t.Fatalf("order %v", order)
426+ }
427+}
428+
429+// When the frame never comes back, nothing is destroyed or closed (the
430+// source's handler and its buffer are still in use), the storage the
431+// handler points into is left untouched and abandoned, and the service has
432+// a fresh table: repeated reconnects do not exhaust it.
433+func TestShutdownTimeoutKeepsHandlerStateAndFreesTable(t *testing.T) {
434+ p := newProducer(t)
435+ p.s.drain = 20 * time.Millisecond
436+ p.streaming()
437+ p.src.EXPECT().Dequeue().Return(p.buf).Once()
438+ p.s.pump(p.st, p.now)
439+ p.cmd()
440+ old := p.s.sessions
441+ p.s.shutdown() // no Destroy, Release or Close is expected
442+ if c := p.cmd(); c.Kind != ports.CastClose {
443+ t.Fatalf("%+v", c)
444+ }
445+ if p.s.sessions == old {
446+ t.Fatal("the table the leaked handler points into is still in use")
447+ }
448+ if !p.st.used || p.st.src == nil || !p.st.inflight || p.st.h2.st != p.st {
449+ t.Fatal("the leaked stream's state was reset")
450+ }
451+ if p.s.streamsLive() || p.s.pwc != nil {
452+ t.Fatal("new table not empty or PipeWire kept")
453+ }
454+ // The handler of the leaked source still works on its own storage.
455+ p.st.h2.StateChanged(pipewire.StreamStatePaused, 42, "")
456+ if p.st.pwState != pipewire.StreamStatePaused {
457+ t.Fatal("handler lost its stream")
458+ }
459+ // A new attempt starts clean and can fill its table again.
460+ p.s.reset(p.s.ctx, p.bus)
461+ for i := range maxSessions {
462+ p.s.sessions[i] = session{used: true, id: uint32(i + 1)}
463+ }
464+}
@@ -322,3 +322,143 @@ func TestCastProducerAllocations(t *testing.T) {
322 t.Fatalf("%v allocations per frame, want 0", n)322 t.Fatalf("%v allocations per frame, want 0", n)
323 }323 }
324 }324 }
325+
326+// The consumer leaves: the node goes to PAUSED and no frame is requested,
327+// whatever damage and time bring, while the session stays open. A frame
328+// already in flight is released, not queued. When the node streams again the
329+// first frame is forced.
330+func TestPausedNodeGetsNoFramesAndSessionStays(t *testing.T) {
331+ p := newProducer(t)
332+ p.streaming()
333+ p.src.EXPECT().Dequeue().Return(p.buf).Twice()
334+ p.s.pump(p.st, p.now)
335+ if c := p.cmd(); c.Kind != ports.CastFrame {
336+ t.Fatalf("%+v", c)
337+ }
338+ // Paused with the frame in flight: it is given back, not queued (no
339+ // Queue or Trigger expectation).
340+ p.st.h2.StateChanged(pipewire.StreamStatePaused, 42, "")
341+ p.src.EXPECT().Release(p.buf).Return().Once()
342+ p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3, Time: time.Unix(5, 0)})
343+ for range 5 {
344+ p.s.onCastEvent(&ports.CastEvent{Kind: ports.CastDamage, ID: 1})
345+ p.now = p.now.Add(time.Second)
346+ p.s.pumpAll(p.now) // Dequeue is not expected a second time while paused
347+ p.none()
348+ }
349+ if p.s.sessions[0].dying || p.st.closing {
350+ t.Fatal("the session ended with the consumer")
351+ }
352+ if got := p.s.timeout(); got != -1 {
353+ t.Fatalf("paused stream asks for a timeout of %d ms", got)
354+ }
355+ // A consumer is back: the picture goes out at once, damage or not.
356+ p.st.h2.StateChanged(pipewire.StreamStateStreaming, 42, "")
357+ p.s.pumpAll(p.now)
358+ if c := p.cmd(); c.Kind != ports.CastFrame {
359+ t.Fatalf("no frame after the node streams again: %+v", c)
360+ }
361+}
362+
363+// Shutdown pumps every stream: CastClose goes out, the node is destroyed and
364+// PipeWire closed, with no frame in flight.
365+func TestShutdownClosesStreams(t *testing.T) {
366+ p := newProducer(t)
367+ p.streaming()
368+ var order []string
369+ p.src.EXPECT().Destroy().Run(func() { order = append(order, "destroy") }).Once()
370+ p.core.EXPECT().Close().Run(func() { order = append(order, "close") }).Once()
371+ p.s.shutdown()
372+ if c := p.cmd(); c.Kind != ports.CastClose || c.ID != 1 {
373+ t.Fatalf("%+v", c)
374+ }
375+ if len(order) != 2 || order[0] != "destroy" || order[1] != "close" || p.s.pwc != nil {
376+ t.Fatalf("order %v", order)
377+ }
378+ if p.st.used || p.s.sessions[0].used {
379+ t.Fatal("slots not freed")
380+ }
381+}
382+
383+// A stream opened on the wayland side that has no node yet is closed too.
384+func TestShutdownClosesStreamWithoutSource(t *testing.T) {
385+ p := newProducer(t)
386+ p.s.pump(p.st, p.now)
387+ if c := p.cmd(); c.Kind != ports.CastOpen {
388+ t.Fatalf("%+v", c)
389+ }
390+ p.core.EXPECT().Close().Return().Once()
391+ p.s.shutdown()
392+ if c := p.cmd(); c.Kind != ports.CastClose || c.ID != 1 {
393+ t.Fatalf("%+v", c)
394+ }
395+ if p.st.used {
396+ t.Fatal("stream not freed")
397+ }
398+}
399+
400+// A frame in flight at shutdown is waited for: nothing is destroyed before
401+// its CastFrameDone, then the buffer is released and the node destroyed.
402+func TestShutdownWaitsForFrameInFlight(t *testing.T) {
403+ p := newProducer(t)
404+ p.streaming()
405+ p.src.EXPECT().Dequeue().Return(p.buf).Once()
406+ p.s.pump(p.st, p.now)
407+ p.cmd()
408+ var order []string
409+ p.src.EXPECT().Release(p.buf).Run(func(*pipewire.VideoBuffer) { order = append(order, "release") }).Once()
410+ p.src.EXPECT().Destroy().Run(func() { order = append(order, "destroy") }).Once()
411+ p.core.EXPECT().Close().Run(func() { order = append(order, "close") }).Once()
412+ go func() {
413+ time.Sleep(30 * time.Millisecond)
414+ p.s.Push(ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 3})
415+ }()
416+ start := time.Now()
417+ p.s.shutdown()
418+ if time.Since(start) < 30*time.Millisecond {
419+ t.Fatal("shutdown did not wait for the frame")
420+ }
421+ if c := p.cmd(); c.Kind != ports.CastClose {
422+ t.Fatalf("%+v", c)
423+ }
424+ if len(order) != 3 || order[0] != "release" || order[1] != "destroy" || order[2] != "close" {
425+ t.Fatalf("order %v", order)
426+ }
427+}
428+
429+// When the frame never comes back, nothing is destroyed or closed (the
430+// source's handler and its buffer are still in use), the storage the
431+// handler points into is left untouched and abandoned, and the service has
432+// a fresh table: repeated reconnects do not exhaust it.
433+func TestShutdownTimeoutKeepsHandlerStateAndFreesTable(t *testing.T) {
434+ p := newProducer(t)
435+ p.s.drain = 20 * time.Millisecond
436+ p.streaming()
437+ p.src.EXPECT().Dequeue().Return(p.buf).Once()
438+ p.s.pump(p.st, p.now)
439+ p.cmd()
440+ old := p.s.sessions
441+ p.s.shutdown() // no Destroy, Release or Close is expected
442+ if c := p.cmd(); c.Kind != ports.CastClose {
443+ t.Fatalf("%+v", c)
444+ }
445+ if p.s.sessions == old {
446+ t.Fatal("the table the leaked handler points into is still in use")
447+ }
448+ if !p.st.used || p.st.src == nil || !p.st.inflight || p.st.h2.st != p.st {
449+ t.Fatal("the leaked stream's state was reset")
450+ }
451+ if p.s.streamsLive() || p.s.pwc != nil {
452+ t.Fatal("new table not empty or PipeWire kept")
453+ }
454+ // The handler of the leaked source still works on its own storage.
455+ p.st.h2.StateChanged(pipewire.StreamStatePaused, 42, "")
456+ if p.st.pwState != pipewire.StreamStatePaused {
457+ t.Fatal("handler lost its stream")
458+ }
459+ // A new attempt starts clean and can fill its table again.
460+ p.s.reset(p.s.ctx, p.bus)
461+ for i := range maxSessions {
462+ p.s.sessions[i] = session{used: true, id: uint32(i + 1)}
463+ }
464+}
M internal/adapters/screencast/screencast.go +49 -20
@@ -54,6 +54,10 @@ const (
5454
5555 // shutdownWait bounds the wait for frames still in flight at shutdown.
5656 shutdownWait = 2 * time.Second
57+
58+ // shutdownPoll is how often shutdown looks again for the events that
59+ // settle a frame in flight.
60+ shutdownPoll = 5 * time.Millisecond
5761 )
5862
5963 var _ ports.CastSink = (*Service)(nil)
@@ -87,22 +91,27 @@ type Service struct {
8791 shotRes shotResult
8892
8993 // Owner state.
90- ctx context.Context
91- conn bus
92- pwc videoCore
93- owns [4]bool // nameScreenCast, nameDisplayConfig, nameIntrospect, nameScreenshot
94- inv ports.CastInventory
95- haveInv bool
96- callers [maxCallers]caller
97- pending [maxPending]pending
98- sessions [maxSessions]session
94+ ctx context.Context // the attempt's: it ends with the attempt
95+ conn bus
96+ pwc videoCore
97+ owns [4]bool // nameScreenCast, nameDisplayConfig, nameIntrospect, nameScreenshot
98+ inv ports.CastInventory
99+ haveInv bool
100+ callers [maxCallers]caller
101+ pending [maxPending]pending
102+ // sessions is a pointer so that a shutdown that had to leave a PipeWire
103+ // source behind can abandon the table: the source's handler still
104+ // points into it, so it is never reset or reused.
105+ sessions *[maxSessions]session
99106 nextSess uint32
100107 nextStrm uint32
101108 nextCast uint64
102109 waitFor time.Duration // time to the next paced frame, 0 when none
103110 shot shotState
111+ shotSeq uint64 // ID of the last screenshot job; never reset
104112 fatal error
105113 jobs sync.WaitGroup
114+ drain time.Duration // how long shutdown waits for frames in flight
106115 }
107116
108117 // New creates the service. policy returns the capture allowlist in force;
@@ -118,6 +127,9 @@ func New(policy func() *captureallow.Policy, cmds chan<- ports.CastCommand, log
118127 peers: pidfdPeers{},
119128 queue: make([]ports.CastEvent, 0, sinkCap),
120129 spare: make([]ports.CastEvent, 0, sinkCap),
130+
131+ sessions: new([maxSessions]session),
132+ drain: shutdownWait,
121133 }
122134 s.poller, s.initErr = fdwait.New()
123135 return s
@@ -204,6 +216,9 @@ func (s *Service) attempt(ctx context.Context) (err error) {
204216 return err
205217 }
206218 defer func() { _ = conn.Close() }()
219+ // The screenshot job of this attempt ends with it.
220+ ctx, cancel := context.WithCancel(ctx)
221+ defer cancel()
207222 s.reset(ctx, conn)
208223 if s.pwc, err = s.pw.Open(); err != nil {
209224 s.pwc = nil
@@ -223,7 +238,7 @@ func (s *Service) reset(ctx context.Context, conn bus) {
223238 s.haveInv = false
224239 s.callers = [maxCallers]caller{}
225240 s.pending = [maxPending]pending{}
226- s.sessions = [maxSessions]session{}
241+ *s.sessions = [maxSessions]session{}
227242 s.shot = shotState{}
228243 }
229244
@@ -326,24 +341,38 @@ func (s *Service) service() {
326341 s.reap()
327342 }
328343
329-// shutdown ends every session and releases PipeWire. A frame still in
330-// flight is waited for: its buffer is lent to the wayland server.
344+// shutdown ends the attempt: the screenshot job is stopped and joined, every
345+// session is closed, and every stream is pumped until it released its
346+// resources: CastClose goes to the wayland server, a frame still in flight
347+// is waited for (its buffer is lent to the wayland server), then the source
348+// is destroyed and its buffers released. Only then is PipeWire closed.
349+//
350+// When a frame does not come back in time, the source cannot be destroyed
351+// safely. It is left as it is, with its loop: the sessions table that its
352+// handler points into is abandoned (never reset or reused) and a new one
353+// takes its place, so later attempts neither touch that storage nor run out
354+// of slots.
331355 func (s *Service) shutdown() {
356+ s.stopShot()
332357 s.forEachSession(func(sess *session) { s.closeSession(sess, true) })
333- deadline := time.Now().Add(shutdownWait)
334- for s.streamsLive() && time.Now().Before(deadline) {
358+ deadline := time.Now().Add(s.drain)
359+ for {
335360 evs, _, _, _ := s.take()
336361 for i := range evs {
337362 s.onCastEvent(&evs[i])
338363 }
364+ s.pumpAll(s.clock.Now())
339365 s.reap()
340- if s.streamsLive() {
341- time.Sleep(5 * time.Millisecond)
366+ if !s.streamsLive() {
367+ break
342368 }
343- }
344- if s.streamsLive() {
345- s.log.Warn().Msg("frames still in flight at shutdown; PipeWire resources are left to the process exit")
346- s.pwc = nil
369+ if !time.Now().Before(deadline) {
370+ s.log.Warn().Int("streams", s.streamCount()).Msg("frames still in flight at shutdown; PipeWire resources are left to the process exit")
371+ s.sessions = new([maxSessions]session)
372+ s.pwc = nil
373+ return
374+ }
375+ time.Sleep(shutdownPoll)
347376 }
348377 if s.pwc != nil {
349378 s.pwc.Close()
@@ -54,6 +54,10 @@ const (
54 54
55 // shutdownWait bounds the wait for frames still in flight at shutdown.55 // shutdownWait bounds the wait for frames still in flight at shutdown.
56 shutdownWait = 2 * time.Second56 shutdownWait = 2 * time.Second
57+
58+ // shutdownPoll is how often shutdown looks again for the events that
59+ // settle a frame in flight.
60+ shutdownPoll = 5 * time.Millisecond
57 )61 )
58 62
59 var _ ports.CastSink = (*Service)(nil)63 var _ ports.CastSink = (*Service)(nil)
@@ -87,22 +91,27 @@ type Service struct {
87 shotRes shotResult91 shotRes shotResult
88 92
89 // Owner state.93 // Owner state.
90- ctx context.Context94+ ctx context.Context // the attempt's: it ends with the attempt
91- conn bus95+ conn bus
92- pwc videoCore96+ pwc videoCore
93- owns [4]bool // nameScreenCast, nameDisplayConfig, nameIntrospect, nameScreenshot97+ owns [4]bool // nameScreenCast, nameDisplayConfig, nameIntrospect, nameScreenshot
94- inv ports.CastInventory98+ inv ports.CastInventory
95- haveInv bool99+ haveInv bool
96- callers [maxCallers]caller100+ callers [maxCallers]caller
97- pending [maxPending]pending101+ pending [maxPending]pending
98- sessions [maxSessions]session102+ // sessions is a pointer so that a shutdown that had to leave a PipeWire
103+ // source behind can abandon the table: the source's handler still
104+ // points into it, so it is never reset or reused.
105+ sessions *[maxSessions]session
99 nextSess uint32106 nextSess uint32
100 nextStrm uint32107 nextStrm uint32
101 nextCast uint64108 nextCast uint64
102 waitFor time.Duration // time to the next paced frame, 0 when none109 waitFor time.Duration // time to the next paced frame, 0 when none
103 shot shotState110 shot shotState
111+ shotSeq uint64 // ID of the last screenshot job; never reset
104 fatal error112 fatal error
105 jobs sync.WaitGroup113 jobs sync.WaitGroup
114+ drain time.Duration // how long shutdown waits for frames in flight
106 }115 }
107 116
108 // New creates the service. policy returns the capture allowlist in force;117 // New creates the service. policy returns the capture allowlist in force;
@@ -118,6 +127,9 @@ func New(policy func() *captureallow.Policy, cmds chan<- ports.CastCommand, log
118 peers: pidfdPeers{},127 peers: pidfdPeers{},
119 queue: make([]ports.CastEvent, 0, sinkCap),128 queue: make([]ports.CastEvent, 0, sinkCap),
120 spare: make([]ports.CastEvent, 0, sinkCap),129 spare: make([]ports.CastEvent, 0, sinkCap),
130+
131+ sessions: new([maxSessions]session),
132+ drain: shutdownWait,
121 }133 }
122 s.poller, s.initErr = fdwait.New()134 s.poller, s.initErr = fdwait.New()
123 return s135 return s
@@ -204,6 +216,9 @@ func (s *Service) attempt(ctx context.Context) (err error) {
204 return err216 return err
205 }217 }
206 defer func() { _ = conn.Close() }()218 defer func() { _ = conn.Close() }()
219+ // The screenshot job of this attempt ends with it.
220+ ctx, cancel := context.WithCancel(ctx)
221+ defer cancel()
207 s.reset(ctx, conn)222 s.reset(ctx, conn)
208 if s.pwc, err = s.pw.Open(); err != nil {223 if s.pwc, err = s.pw.Open(); err != nil {
209 s.pwc = nil224 s.pwc = nil
@@ -223,7 +238,7 @@ func (s *Service) reset(ctx context.Context, conn bus) {
223 s.haveInv = false238 s.haveInv = false
224 s.callers = [maxCallers]caller{}239 s.callers = [maxCallers]caller{}
225 s.pending = [maxPending]pending{}240 s.pending = [maxPending]pending{}
226- s.sessions = [maxSessions]session{}241+ *s.sessions = [maxSessions]session{}
227 s.shot = shotState{}242 s.shot = shotState{}
228 }243 }
229 244
@@ -326,24 +341,38 @@ func (s *Service) service() {
326 s.reap()341 s.reap()
327 }342 }
328 343
329-// shutdown ends every session and releases PipeWire. A frame still in344+// shutdown ends the attempt: the screenshot job is stopped and joined, every
330-// flight is waited for: its buffer is lent to the wayland server.345+// session is closed, and every stream is pumped until it released its
346+// resources: CastClose goes to the wayland server, a frame still in flight
347+// is waited for (its buffer is lent to the wayland server), then the source
348+// is destroyed and its buffers released. Only then is PipeWire closed.
349+//
350+// When a frame does not come back in time, the source cannot be destroyed
351+// safely. It is left as it is, with its loop: the sessions table that its
352+// handler points into is abandoned (never reset or reused) and a new one
353+// takes its place, so later attempts neither touch that storage nor run out
354+// of slots.
331 func (s *Service) shutdown() {355 func (s *Service) shutdown() {
356+ s.stopShot()
332 s.forEachSession(func(sess *session) { s.closeSession(sess, true) })357 s.forEachSession(func(sess *session) { s.closeSession(sess, true) })
333- deadline := time.Now().Add(shutdownWait)358+ deadline := time.Now().Add(s.drain)
334- for s.streamsLive() && time.Now().Before(deadline) {359+ for {
335 evs, _, _, _ := s.take()360 evs, _, _, _ := s.take()
336 for i := range evs {361 for i := range evs {
337 s.onCastEvent(&evs[i])362 s.onCastEvent(&evs[i])
338 }363 }
364+ s.pumpAll(s.clock.Now())
339 s.reap()365 s.reap()
340- if s.streamsLive() {366+ if !s.streamsLive() {
341- time.Sleep(5 * time.Millisecond)367+ break
342 }368 }
343- }369+ if !time.Now().Before(deadline) {
344- if s.streamsLive() {370+ s.log.Warn().Int("streams", s.streamCount()).Msg("frames still in flight at shutdown; PipeWire resources are left to the process exit")
345- s.log.Warn().Msg("frames still in flight at shutdown; PipeWire resources are left to the process exit")371+ s.sessions = new([maxSessions]session)
346- s.pwc = nil372+ s.pwc = nil
373+ return
374+ }
375+ time.Sleep(shutdownPoll)
347 }376 }
348 if s.pwc != nil {377 if s.pwc != nil {
349 s.pwc.Close()378 s.pwc.Close()
M internal/adapters/screencast/screenshot.go +87 -31
@@ -15,24 +15,34 @@ import (
1515 // InteractiveScreenshot nefercap shot -file F
1616 //
1717 // F is $XDG_RUNTIME_DIR/neferwl/screenshot-<unixnano>.png. One run at most
18-// is in flight. The run is a goroutine; its result goes back through the
19-// service's transport and the owner sends the reply, so the bus is still
20-// used by the owner only. SelectArea and ScreenshotWindow (the "Select
18+// is in flight. The run is a goroutine; its result, tagged with the ID of
19+// the job, goes back through the service's transport and the owner sends
20+// the reply, so the bus is still used by the owner only. A result whose ID
21+// is not the job in flight is dropped. A job belongs to one bus attempt:
22+// when the attempt ends, the job is cancelled and joined before the state is
23+// reset (stopShot). A run that exits cleanly is a success only if it left a
24+// non-empty regular file at F; otherwise the reply is (false, ""), which is
25+// also how a cancelled interactive screenshot is answered. SelectArea and ScreenshotWindow (the "Select
2126 // area" and "Window" choices of the GNOME dialog) are not supported;
2227 // include_cursor and flash are ignored.
2328
2429 // shotState is the screenshot in flight; one at most.
2530 type shotState struct {
26- busy bool
27- req request
28- uri bool // reply with a file:// URI
29- path string
31+ busy bool
32+ id uint64 // names the job; results of other jobs are stale
33+ req request
34+ uri bool // reply with a file:// URI
35+ path string
36+ cancel context.CancelFunc // stops the run
37+ done chan struct{} // closed when the job goroutine is past its result
3038 }
3139
3240 // shotResult is what a finished nefercap run leaves for the owner.
3341 type shotResult struct {
3442 ready bool
35- err error
43+ id uint64 // the job it belongs to
44+ err error // nefercap failed
45+ made bool // the output file exists: a regular, non-empty file
3646 }
3747
3848 // screenshot answers Screenshot, ScreenshotArea and InteractiveScreenshot.
@@ -74,8 +84,9 @@ func (s *Service) screenshot(req *request) {
7484 s.replyErr(req, errFailed, "screenshot directory not created")
7585 return
7686 }
77- s.shot = shotState{busy: true, req: *req, uri: req.method == mShotInteractive, path: path}
78- s.runShot(s.ctx, args)
87+ s.shotSeq++
88+ s.shot = shotState{busy: true, id: s.shotSeq, req: *req, uri: req.method == mShotInteractive, path: path}
89+ s.runShot(args)
7990 }
8091
8192 // shotDir is $XDG_RUNTIME_DIR/neferwl.
@@ -105,44 +116,89 @@ func (s *Service) areaTarget(x, y, w, h int32) (string, string, bool) {
105116 return "", "", false
106117 }
107118
108-// runShot runs nefercap in a goroutine and leaves the result for the owner.
109-func (s *Service) runShot(parent context.Context, args []string) {
119+// runShot runs nefercap in a goroutine for the job in s.shot and leaves the
120+// tagged result for the owner. The run ends with the attempt's context.
121+func (s *Service) runShot(args []string) {
122+ ctx, cancel := context.WithTimeout(s.ctx, shotTimeout)
123+ done := make(chan struct{})
124+ s.shot.cancel, s.shot.done = cancel, done
125+ id, path := s.shot.id, s.shot.path
126+ run := s.runner
110127 s.jobs.Add(1)
111128 go func() {
112129 defer s.jobs.Done()
113- ctx, cancel := context.WithTimeout(parent, shotTimeout)
130+ defer close(done)
114131 defer cancel()
115- err := s.runner.Run(ctx, args)
132+ err := run.Run(ctx, args)
133+ res := shotResult{ready: true, id: id, err: err}
134+ if err == nil {
135+ res.made = isOutputFile(path)
136+ }
116137 s.mu.Lock()
117- s.shotRes = shotResult{ready: true, err: err}
138+ s.shotRes = res
118139 s.woken = true
119140 s.mu.Unlock()
120141 s.wake()
121142 }()
122143 }
123144
124-// onShotDone sends the reply of the screenshot that finished.
125-func (s *Service) onShotDone(r *shotResult) {
145+// isOutputFile reports whether path is a regular, non-empty file. A symlink
146+// is not followed: it is not a regular file.
147+func isOutputFile(path string) bool {
148+ fi, err := os.Lstat(path)
149+ return err == nil && fi.Mode().IsRegular() && fi.Size() > 0
150+}
151+
152+// stopShot cancels the screenshot in flight, waits for its goroutine and
153+// forgets its result, so no job outlives the attempt that started it. The
154+// requester is told it failed when the bus still carries the answer.
155+func (s *Service) stopShot() {
126156 st := s.shot
127- s.shot = shotState{}
128157 if !st.busy {
129158 return
130159 }
131- if r.err != nil {
160+ st.cancel()
161+ <-st.done
162+ s.mu.Lock()
163+ s.shotRes = shotResult{}
164+ s.mu.Unlock()
165+ s.shot = shotState{}
166+ _ = os.Remove(st.path)
167+ s.replyErr(&st.req, errFailed, "screenshot cancelled")
168+}
169+
170+// onShotDone sends the reply of the screenshot that finished. A result of
171+// another job than the one in flight is stale and completes nothing.
172+func (s *Service) onShotDone(r *shotResult) {
173+ st := s.shot
174+ if !st.busy || r.id != st.id {
175+ return
176+ }
177+ s.shot = shotState{}
178+ switch {
179+ case r.err != nil:
132180 s.log.Warn().Err(r.err).Msg("screenshot failed")
133181 _ = os.Remove(st.path)
134182 s.replyErr(&st.req, errFailed, "screenshot failed")
135- return
136- }
137- out := st.path
138- if st.uri {
139- out = "file://" + out
140- }
141- e := s.replyTo(&st.req, "bs")
142- if e == nil {
143- return
183+ case !r.made:
184+ // nefercap exited cleanly and left nothing usable (an interactive
185+ // screenshot that was cancelled, for one).
186+ s.log.Info().Msg("screenshot produced no file")
187+ _ = os.Remove(st.path)
188+ if e := s.replyTo(&st.req, "bs"); e != nil {
189+ e.Bool(false)
190+ e.Str("")
191+ s.send()
192+ }
193+ default:
194+ out := st.path
195+ if st.uri {
196+ out = "file://" + out
197+ }
198+ if e := s.replyTo(&st.req, "bs"); e != nil {
199+ e.Bool(true)
200+ e.Str(out)
201+ s.send()
202+ }
144203 }
145- e.Bool(true)
146- e.Str(out)
147- s.send()
148204 }
@@ -15,24 +15,34 @@ import (
15 // InteractiveScreenshot nefercap shot -file F15 // InteractiveScreenshot nefercap shot -file F
16 //16 //
17 // F is $XDG_RUNTIME_DIR/neferwl/screenshot-<unixnano>.png. One run at most17 // F is $XDG_RUNTIME_DIR/neferwl/screenshot-<unixnano>.png. One run at most
18-// is in flight. The run is a goroutine; its result goes back through the18+// is in flight. The run is a goroutine; its result, tagged with the ID of
19-// service's transport and the owner sends the reply, so the bus is still19+// the job, goes back through the service's transport and the owner sends
20-// used by the owner only. SelectArea and ScreenshotWindow (the "Select20+// the reply, so the bus is still used by the owner only. A result whose ID
21+// is not the job in flight is dropped. A job belongs to one bus attempt:
22+// when the attempt ends, the job is cancelled and joined before the state is
23+// reset (stopShot). A run that exits cleanly is a success only if it left a
24+// non-empty regular file at F; otherwise the reply is (false, ""), which is
25+// also how a cancelled interactive screenshot is answered. SelectArea and ScreenshotWindow (the "Select
21 // area" and "Window" choices of the GNOME dialog) are not supported;26 // area" and "Window" choices of the GNOME dialog) are not supported;
22 // include_cursor and flash are ignored.27 // include_cursor and flash are ignored.
23 28
24 // shotState is the screenshot in flight; one at most.29 // shotState is the screenshot in flight; one at most.
25 type shotState struct {30 type shotState struct {
26- busy bool31+ busy bool
27- req request32+ id uint64 // names the job; results of other jobs are stale
28- uri bool // reply with a file:// URI33+ req request
29- path string34+ uri bool // reply with a file:// URI
35+ path string
36+ cancel context.CancelFunc // stops the run
37+ done chan struct{} // closed when the job goroutine is past its result
30 }38 }
31 39
32 // shotResult is what a finished nefercap run leaves for the owner.40 // shotResult is what a finished nefercap run leaves for the owner.
33 type shotResult struct {41 type shotResult struct {
34 ready bool42 ready bool
35- err error43+ id uint64 // the job it belongs to
44+ err error // nefercap failed
45+ made bool // the output file exists: a regular, non-empty file
36 }46 }
37 47
38 // screenshot answers Screenshot, ScreenshotArea and InteractiveScreenshot.48 // screenshot answers Screenshot, ScreenshotArea and InteractiveScreenshot.
@@ -74,8 +84,9 @@ func (s *Service) screenshot(req *request) {
74 s.replyErr(req, errFailed, "screenshot directory not created")84 s.replyErr(req, errFailed, "screenshot directory not created")
75 return85 return
76 }86 }
77- s.shot = shotState{busy: true, req: *req, uri: req.method == mShotInteractive, path: path}87+ s.shotSeq++
78- s.runShot(s.ctx, args)88+ s.shot = shotState{busy: true, id: s.shotSeq, req: *req, uri: req.method == mShotInteractive, path: path}
89+ s.runShot(args)
79 }90 }
80 91
81 // shotDir is $XDG_RUNTIME_DIR/neferwl.92 // shotDir is $XDG_RUNTIME_DIR/neferwl.
@@ -105,44 +116,89 @@ func (s *Service) areaTarget(x, y, w, h int32) (string, string, bool) {
105 return "", "", false116 return "", "", false
106 }117 }
107 118
108-// runShot runs nefercap in a goroutine and leaves the result for the owner.119+// runShot runs nefercap in a goroutine for the job in s.shot and leaves the
109-func (s *Service) runShot(parent context.Context, args []string) {120+// tagged result for the owner. The run ends with the attempt's context.
121+func (s *Service) runShot(args []string) {
122+ ctx, cancel := context.WithTimeout(s.ctx, shotTimeout)
123+ done := make(chan struct{})
124+ s.shot.cancel, s.shot.done = cancel, done
125+ id, path := s.shot.id, s.shot.path
126+ run := s.runner
110 s.jobs.Add(1)127 s.jobs.Add(1)
111 go func() {128 go func() {
112 defer s.jobs.Done()129 defer s.jobs.Done()
113- ctx, cancel := context.WithTimeout(parent, shotTimeout)130+ defer close(done)
114 defer cancel()131 defer cancel()
115- err := s.runner.Run(ctx, args)132+ err := run.Run(ctx, args)
133+ res := shotResult{ready: true, id: id, err: err}
134+ if err == nil {
135+ res.made = isOutputFile(path)
136+ }
116 s.mu.Lock()137 s.mu.Lock()
117- s.shotRes = shotResult{ready: true, err: err}138+ s.shotRes = res
118 s.woken = true139 s.woken = true
119 s.mu.Unlock()140 s.mu.Unlock()
120 s.wake()141 s.wake()
121 }()142 }()
122 }143 }
123 144
124-// onShotDone sends the reply of the screenshot that finished.145+// isOutputFile reports whether path is a regular, non-empty file. A symlink
125-func (s *Service) onShotDone(r *shotResult) {146+// is not followed: it is not a regular file.
147+func isOutputFile(path string) bool {
148+ fi, err := os.Lstat(path)
149+ return err == nil && fi.Mode().IsRegular() && fi.Size() > 0
150+}
151+
152+// stopShot cancels the screenshot in flight, waits for its goroutine and
153+// forgets its result, so no job outlives the attempt that started it. The
154+// requester is told it failed when the bus still carries the answer.
155+func (s *Service) stopShot() {
126 st := s.shot156 st := s.shot
127- s.shot = shotState{}
128 if !st.busy {157 if !st.busy {
129 return158 return
130 }159 }
131- if r.err != nil {160+ st.cancel()
161+ <-st.done
162+ s.mu.Lock()
163+ s.shotRes = shotResult{}
164+ s.mu.Unlock()
165+ s.shot = shotState{}
166+ _ = os.Remove(st.path)
167+ s.replyErr(&st.req, errFailed, "screenshot cancelled")
168+}
169+
170+// onShotDone sends the reply of the screenshot that finished. A result of
171+// another job than the one in flight is stale and completes nothing.
172+func (s *Service) onShotDone(r *shotResult) {
173+ st := s.shot
174+ if !st.busy || r.id != st.id {
175+ return
176+ }
177+ s.shot = shotState{}
178+ switch {
179+ case r.err != nil:
132 s.log.Warn().Err(r.err).Msg("screenshot failed")180 s.log.Warn().Err(r.err).Msg("screenshot failed")
133 _ = os.Remove(st.path)181 _ = os.Remove(st.path)
134 s.replyErr(&st.req, errFailed, "screenshot failed")182 s.replyErr(&st.req, errFailed, "screenshot failed")
135- return183+ case !r.made:
136- }184+ // nefercap exited cleanly and left nothing usable (an interactive
137- out := st.path185+ // screenshot that was cancelled, for one).
138- if st.uri {186+ s.log.Info().Msg("screenshot produced no file")
139- out = "file://" + out187+ _ = os.Remove(st.path)
140- }188+ if e := s.replyTo(&st.req, "bs"); e != nil {
141- e := s.replyTo(&st.req, "bs")189+ e.Bool(false)
142- if e == nil {190+ e.Str("")
143- return191+ s.send()
192+ }
193+ default:
194+ out := st.path
195+ if st.uri {
196+ out = "file://" + out
197+ }
198+ if e := s.replyTo(&st.req, "bs"); e != nil {
199+ e.Bool(true)
200+ e.Str(out)
201+ s.send()
202+ }
144 }203 }
145- e.Bool(true)
146- e.Str(out)
147- s.send()
148 }204 }
M internal/adapters/screencast/screenshot_test.go +114 -1
@@ -12,6 +12,19 @@ import (
1212 "github.com/stretchr/testify/mock"
1313 )
1414
15+// writeShot is what nefercap does: it writes the PNG named by -file.
16+func writeShot(t *testing.T, args []string) {
17+ t.Helper()
18+ i := slices.Index(args, "-file")
19+ if i < 0 || i+1 >= len(args) {
20+ t.Errorf("no -file in %v", args)
21+ return
22+ }
23+ if err := os.WriteFile(args[i+1], []byte("\x89PNG"), 0o600); err != nil {
24+ t.Error(err)
25+ }
26+}
27+
1528 // Screenshot: nefercap on the focused output; the reply carries the file.
1629 func TestScreenshotRunsNefercapOnFocusedOutput(t *testing.T) {
1730 rt := t.TempDir()
@@ -20,6 +33,7 @@ func TestScreenshotRunsNefercapOnFocusedOutput(t *testing.T) {
2033 got := make(chan []string, 1)
2134 h.run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, args []string) error {
2235 got <- slices.Clone(args)
36+ writeShot(t, args)
2337 return nil
2438 }).Once()
2539 name, m := callErr(c, nameScreenshot, pathScreenshot, ifaceScreenshot, "Screenshot", "bbs", func(e *zerobus.Encoder) {
@@ -53,6 +67,7 @@ func TestScreenshotAreaMapsToRegion(t *testing.T) {
5367 got := make(chan []string, 1)
5468 h.run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, args []string) error {
5569 got <- slices.Clone(args)
70+ writeShot(t, args)
5671 return nil
5772 }).Once()
5873 area := func(x, y, w, hh int32) func(*zerobus.Encoder) {
@@ -91,7 +106,11 @@ func TestInteractiveScreenshotBusyAndFailure(t *testing.T) {
91106 got := make(chan []string, 2)
92107 h.run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, args []string) error {
93108 got <- slices.Clone(args)
94- return <-release
109+ err := <-release
110+ if err == nil {
111+ writeShot(t, args)
112+ }
113+ return err
95114 }).Twice()
96115
97116 c.NewCall(nameScreenshot, pathScreenshot, ifaceScreenshot, "InteractiveScreenshot", "")
@@ -140,3 +159,97 @@ func TestScreenshotUnsupportedMethods(t *testing.T) {
140159 }
141160 }
142161 }
162+
163+// nefercap that exits cleanly without leaving a regular, non-empty file is
164+// not a success: the file is missing, empty, or a symlink.
165+func TestScreenshotWithoutUsableFileIsNotSuccess(t *testing.T) {
166+ rt := t.TempDir()
167+ t.Setenv("XDG_RUNTIME_DIR", rt)
168+ target := filepath.Join(rt, "real.png")
169+ if err := os.WriteFile(target, []byte("png"), 0o600); err != nil {
170+ t.Fatal(err)
171+ }
172+ for name, mk := range map[string]func(path string){
173+ "missing": func(string) {},
174+ "empty": func(path string) { _ = os.WriteFile(path, nil, 0o600) },
175+ "symlink": func(path string) { _ = os.Symlink(target, path) },
176+ "dir": func(path string) { _ = os.Mkdir(path, 0o700) },
177+ } {
178+ t.Run(name, func(t *testing.T) {
179+ h, c := ready(t, harnessOpts{})
180+ h.run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, args []string) error {
181+ mk(args[slices.Index(args, "-file")+1])
182+ return nil
183+ }).Once()
184+ for _, member := range []string{"Screenshot", "InteractiveScreenshot"} {
185+ if member == "InteractiveScreenshot" {
186+ h.run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, args []string) error {
187+ mk(args[slices.Index(args, "-file")+1])
188+ return nil
189+ }).Once()
190+ }
191+ sig := ""
192+ body := func(*zerobus.Encoder) {}
193+ if member == "Screenshot" {
194+ sig = "bbs"
195+ body = func(e *zerobus.Encoder) { e.Bool(false); e.Bool(false); e.Str("x") }
196+ }
197+ errName, m := callErr(c, nameScreenshot, pathScreenshot, ifaceScreenshot, member, sig, body)
198+ if errName != "" {
199+ t.Fatalf("%s: %s", member, errName)
200+ }
201+ r := m.Body()
202+ if ok, file := r.Bool(), r.Str(); ok || file != "" {
203+ t.Fatalf("%s: success %v %q without a file", member, ok, file)
204+ }
205+ }
206+ })
207+ }
208+}
209+
210+// A result that belongs to another job completes nothing, and a job that
211+// the attempt's end cancels is joined before its state is reset.
212+func TestStaleScreenshotResultIsDropped(t *testing.T) {
213+ p := newProducer(t)
214+ run := newMockrunner(t)
215+ p.s.runner = run
216+ ctx, cancel := context.WithCancel(context.Background())
217+ defer cancel()
218+ p.s.ctx = ctx
219+ started := make(chan struct{})
220+ var seen error
221+ run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(c context.Context, _ []string) error {
222+ close(started)
223+ <-c.Done()
224+ seen = c.Err()
225+ return c.Err()
226+ }).Once()
227+ p.bus.EXPECT().NewError(mock.Anything, errFailed, "s").RunAndReturn(
228+ func(*zerobus.Message, string, string) *zerobus.Encoder { return &zerobus.Encoder{} }).Once()
229+
230+ p.s.shotSeq++
231+ p.s.shot = shotState{busy: true, id: p.s.shotSeq, path: filepath.Join(t.TempDir(), "x.png")}
232+ p.s.runShot([]string{"shot"})
233+ <-started
234+ // A result of some earlier job is in the transport: it is not this job's.
235+ p.s.onShotDone(&shotResult{ready: true, id: p.s.shotSeq - 1, made: true})
236+ if !p.s.shot.busy {
237+ t.Fatal("a stale result completed the job in flight")
238+ }
239+ // The attempt ends: the job is cancelled and joined before the reset.
240+ p.s.stopShot()
241+ if seen == nil {
242+ t.Fatal("stopShot returned before the job saw its cancellation")
243+ }
244+ if p.s.shot.busy || p.s.shotRes.ready {
245+ t.Fatal("job state survives the attempt")
246+ }
247+ // Its own late result (not possible after the join, but harmless) and a
248+ // new job with a new ID: the first job's ID completes nothing.
249+ p.s.shotSeq++
250+ p.s.shot = shotState{busy: true, id: p.s.shotSeq}
251+ p.s.onShotDone(&shotResult{ready: true, id: p.s.shotSeq - 1, made: true})
252+ if !p.s.shot.busy {
253+ t.Fatal("a result of the cancelled job completed the next request")
254+ }
255+}
@@ -12,6 +12,19 @@ import (
12 "github.com/stretchr/testify/mock"12 "github.com/stretchr/testify/mock"
13 )13 )
14 14
15+// writeShot is what nefercap does: it writes the PNG named by -file.
16+func writeShot(t *testing.T, args []string) {
17+ t.Helper()
18+ i := slices.Index(args, "-file")
19+ if i < 0 || i+1 >= len(args) {
20+ t.Errorf("no -file in %v", args)
21+ return
22+ }
23+ if err := os.WriteFile(args[i+1], []byte("\x89PNG"), 0o600); err != nil {
24+ t.Error(err)
25+ }
26+}
27+
15 // Screenshot: nefercap on the focused output; the reply carries the file.28 // Screenshot: nefercap on the focused output; the reply carries the file.
16 func TestScreenshotRunsNefercapOnFocusedOutput(t *testing.T) {29 func TestScreenshotRunsNefercapOnFocusedOutput(t *testing.T) {
17 rt := t.TempDir()30 rt := t.TempDir()
@@ -20,6 +33,7 @@ func TestScreenshotRunsNefercapOnFocusedOutput(t *testing.T) {
20 got := make(chan []string, 1)33 got := make(chan []string, 1)
21 h.run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, args []string) error {34 h.run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, args []string) error {
22 got <- slices.Clone(args)35 got <- slices.Clone(args)
36+ writeShot(t, args)
23 return nil37 return nil
24 }).Once()38 }).Once()
25 name, m := callErr(c, nameScreenshot, pathScreenshot, ifaceScreenshot, "Screenshot", "bbs", func(e *zerobus.Encoder) {39 name, m := callErr(c, nameScreenshot, pathScreenshot, ifaceScreenshot, "Screenshot", "bbs", func(e *zerobus.Encoder) {
@@ -53,6 +67,7 @@ func TestScreenshotAreaMapsToRegion(t *testing.T) {
53 got := make(chan []string, 1)67 got := make(chan []string, 1)
54 h.run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, args []string) error {68 h.run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, args []string) error {
55 got <- slices.Clone(args)69 got <- slices.Clone(args)
70+ writeShot(t, args)
56 return nil71 return nil
57 }).Once()72 }).Once()
58 area := func(x, y, w, hh int32) func(*zerobus.Encoder) {73 area := func(x, y, w, hh int32) func(*zerobus.Encoder) {
@@ -91,7 +106,11 @@ func TestInteractiveScreenshotBusyAndFailure(t *testing.T) {
91 got := make(chan []string, 2)106 got := make(chan []string, 2)
92 h.run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, args []string) error {107 h.run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, args []string) error {
93 got <- slices.Clone(args)108 got <- slices.Clone(args)
94- return <-release109+ err := <-release
110+ if err == nil {
111+ writeShot(t, args)
112+ }
113+ return err
95 }).Twice()114 }).Twice()
96 115
97 c.NewCall(nameScreenshot, pathScreenshot, ifaceScreenshot, "InteractiveScreenshot", "")116 c.NewCall(nameScreenshot, pathScreenshot, ifaceScreenshot, "InteractiveScreenshot", "")
@@ -140,3 +159,97 @@ func TestScreenshotUnsupportedMethods(t *testing.T) {
140 }159 }
141 }160 }
142 }161 }
162+
163+// nefercap that exits cleanly without leaving a regular, non-empty file is
164+// not a success: the file is missing, empty, or a symlink.
165+func TestScreenshotWithoutUsableFileIsNotSuccess(t *testing.T) {
166+ rt := t.TempDir()
167+ t.Setenv("XDG_RUNTIME_DIR", rt)
168+ target := filepath.Join(rt, "real.png")
169+ if err := os.WriteFile(target, []byte("png"), 0o600); err != nil {
170+ t.Fatal(err)
171+ }
172+ for name, mk := range map[string]func(path string){
173+ "missing": func(string) {},
174+ "empty": func(path string) { _ = os.WriteFile(path, nil, 0o600) },
175+ "symlink": func(path string) { _ = os.Symlink(target, path) },
176+ "dir": func(path string) { _ = os.Mkdir(path, 0o700) },
177+ } {
178+ t.Run(name, func(t *testing.T) {
179+ h, c := ready(t, harnessOpts{})
180+ h.run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, args []string) error {
181+ mk(args[slices.Index(args, "-file")+1])
182+ return nil
183+ }).Once()
184+ for _, member := range []string{"Screenshot", "InteractiveScreenshot"} {
185+ if member == "InteractiveScreenshot" {
186+ h.run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, args []string) error {
187+ mk(args[slices.Index(args, "-file")+1])
188+ return nil
189+ }).Once()
190+ }
191+ sig := ""
192+ body := func(*zerobus.Encoder) {}
193+ if member == "Screenshot" {
194+ sig = "bbs"
195+ body = func(e *zerobus.Encoder) { e.Bool(false); e.Bool(false); e.Str("x") }
196+ }
197+ errName, m := callErr(c, nameScreenshot, pathScreenshot, ifaceScreenshot, member, sig, body)
198+ if errName != "" {
199+ t.Fatalf("%s: %s", member, errName)
200+ }
201+ r := m.Body()
202+ if ok, file := r.Bool(), r.Str(); ok || file != "" {
203+ t.Fatalf("%s: success %v %q without a file", member, ok, file)
204+ }
205+ }
206+ })
207+ }
208+}
209+
210+// A result that belongs to another job completes nothing, and a job that
211+// the attempt's end cancels is joined before its state is reset.
212+func TestStaleScreenshotResultIsDropped(t *testing.T) {
213+ p := newProducer(t)
214+ run := newMockrunner(t)
215+ p.s.runner = run
216+ ctx, cancel := context.WithCancel(context.Background())
217+ defer cancel()
218+ p.s.ctx = ctx
219+ started := make(chan struct{})
220+ var seen error
221+ run.EXPECT().Run(mock.Anything, mock.Anything).RunAndReturn(func(c context.Context, _ []string) error {
222+ close(started)
223+ <-c.Done()
224+ seen = c.Err()
225+ return c.Err()
226+ }).Once()
227+ p.bus.EXPECT().NewError(mock.Anything, errFailed, "s").RunAndReturn(
228+ func(*zerobus.Message, string, string) *zerobus.Encoder { return &zerobus.Encoder{} }).Once()
229+
230+ p.s.shotSeq++
231+ p.s.shot = shotState{busy: true, id: p.s.shotSeq, path: filepath.Join(t.TempDir(), "x.png")}
232+ p.s.runShot([]string{"shot"})
233+ <-started
234+ // A result of some earlier job is in the transport: it is not this job's.
235+ p.s.onShotDone(&shotResult{ready: true, id: p.s.shotSeq - 1, made: true})
236+ if !p.s.shot.busy {
237+ t.Fatal("a stale result completed the job in flight")
238+ }
239+ // The attempt ends: the job is cancelled and joined before the reset.
240+ p.s.stopShot()
241+ if seen == nil {
242+ t.Fatal("stopShot returned before the job saw its cancellation")
243+ }
244+ if p.s.shot.busy || p.s.shotRes.ready {
245+ t.Fatal("job state survives the attempt")
246+ }
247+ // Its own late result (not possible after the join, but harmless) and a
248+ // new job with a new ID: the first job's ID completes nothing.
249+ p.s.shotSeq++
250+ p.s.shot = shotState{busy: true, id: p.s.shotSeq}
251+ p.s.onShotDone(&shotResult{ready: true, id: p.s.shotSeq - 1, made: true})
252+ if !p.s.shot.busy {
253+ t.Fatal("a result of the cancelled job completed the next request")
254+ }
255+}
M internal/adapters/screencast/seams.go +3 -0
@@ -116,6 +116,9 @@ func (r execRunner) Run(ctx context.Context, args []string) error {
116116 return err
117117 }
118118 cmd := exec.CommandContext(ctx, bin, args...)
119+ // A cancelled run is joined by the owner: never wait long for pipes
120+ // that a grandchild keeps open.
121+ cmd.WaitDelay = 2 * time.Second
119122 if len(r.env) > 0 {
120123 cmd.Env = r.env
121124 }
@@ -116,6 +116,9 @@ func (r execRunner) Run(ctx context.Context, args []string) error {
116 return err116 return err
117 }117 }
118 cmd := exec.CommandContext(ctx, bin, args...)118 cmd := exec.CommandContext(ctx, bin, args...)
119+ // A cancelled run is joined by the owner: never wait long for pipes
120+ // that a grandchild keeps open.
121+ cmd.WaitDelay = 2 * time.Second
119 if len(r.env) > 0 {122 if len(r.env) > 0 {
120 cmd.Env = r.env123 cmd.Env = r.env
121 }124 }
A internal/adapters/screencast/shutdown_test.go +138 -0
@@ -0,0 +1,138 @@
1+package screencast
2+
3+import (
4+ "testing"
5+ "time"
6+
7+ "github.com/bnema/neferwl/internal/ports"
8+ "github.com/bnema/purego-pipewire/pipewire"
9+ "github.com/stretchr/testify/mock"
10+)
11+
12+// liveCast is a cast on a private bus, brought to STREAMING with a mocked
13+// PipeWire source, so a test can lose the bus under it.
14+type liveCast struct {
15+ h *svcHarness
16+ src *mockvideoSource
17+ buf *pipewire.VideoBuffer
18+ // onLoop runs functions inside the PipeWire loop iteration, where the
19+ // real handler callbacks run.
20+ onLoop chan func()
21+ hdl chan pipewire.VideoHandler
22+}
23+
24+func (l *liveCast) loop(f func()) {
25+ l.onLoop <- f
26+ l.h.svc.Push(ports.CastEvent{Kind: ports.CastDamage, ID: 1 << 40}) // wakes the owner
27+}
28+
29+func startLiveCast(t *testing.T, withFrame bool) *liveCast {
30+ t.Helper()
31+ l := &liveCast{onLoop: make(chan func(), 8), hdl: make(chan pipewire.VideoHandler, 1)}
32+ l.h, _ = ready(t, harnessOpts{iterate: func() {
33+ select {
34+ case f := <-l.onLoop:
35+ f()
36+ default:
37+ }
38+ }})
39+ h := l.h
40+ c := h.client()
41+ l.src = newMockvideoSource(t)
42+ l.buf = &pipewire.VideoBuffer{Index: 2, Data: make([]byte, 64*32*4), Stride: 256, Width: 64, Height: 32, Format: pipewire.VideoFormatBGRx}
43+ h.core.EXPECT().NewVideoSource(mock.Anything, mock.Anything).RunAndReturn(
44+ func(_ pipewire.VideoConfig, hd pipewire.VideoHandler) (videoSource, error) {
45+ l.hdl <- hd
46+ return l.src, nil
47+ }).Once()
48+ if withFrame {
49+ l.src.EXPECT().Dequeue().Return(l.buf).Once()
50+ } else {
51+ l.src.EXPECT().Dequeue().Return(nil).Maybe()
52+ }
53+
54+ sess := createSession(t, c)
55+ if name, _ := recordMonitor(c, sess, "HEADLESS-1"); name != "" {
56+ t.Fatal(name)
57+ }
58+ if name, _ := callErr(c, nameScreenCast, sess, ifaceSession, "Start", "", nil); name != "" {
59+ t.Fatal(name)
60+ }
61+ if cmd := recvCmd(t, h); cmd.Kind != ports.CastOpen {
62+ t.Fatalf("%+v", cmd)
63+ }
64+ h.svc.Push(ports.CastEvent{Kind: ports.CastSize, ID: 1, Width: 64, Height: 32})
65+ var hd pipewire.VideoHandler
66+ select {
67+ case hd = <-l.hdl:
68+ case <-time.After(5 * time.Second):
69+ t.Fatal("no PipeWire source")
70+ }
71+ l.loop(func() {
72+ hd.StateChanged(pipewire.StreamStatePaused, 42, "")
73+ hd.FormatChanged(pipewire.VideoFormatBGRx, 64, 32, 256)
74+ hd.StateChanged(pipewire.StreamStateStreaming, 42, "")
75+ })
76+ if withFrame {
77+ if cmd := recvCmd(t, h); cmd.Kind != ports.CastFrame || cmd.Buf != 2 {
78+ t.Fatalf("%+v", cmd)
79+ }
80+ }
81+ return l
82+}
83+
84+func recvCmd(t *testing.T, h *svcHarness) ports.CastCommand {
85+ t.Helper()
86+ select {
87+ case c := <-h.cmds:
88+ return c
89+ case <-time.After(5 * time.Second):
90+ t.Fatal("no command\n" + h.log.String())
91+ return ports.CastCommand{}
92+ }
93+}
94+
95+func waitClosed(t *testing.T, h *svcHarness) {
96+ t.Helper()
97+ select {
98+ case <-h.closed:
99+ case <-time.After(5 * time.Second):
100+ t.Fatal("PipeWire not closed\n" + h.log.String())
101+ }
102+}
103+
104+// Losing the bus with a live stream and no frame in flight: the cast is
105+// closed on the wayland side, the source destroyed, PipeWire closed.
106+func TestBusLossTearsDownStream(t *testing.T) {
107+ l := startLiveCast(t, false)
108+ l.src.EXPECT().Destroy().Return().Once()
109+ l.h.killBus()
110+ if cmd := recvCmd(t, l.h); cmd.Kind != ports.CastClose || cmd.ID != 1 {
111+ t.Fatalf("%+v", cmd)
112+ }
113+ waitClosed(t, l.h)
114+}
115+
116+// With a frame in flight: CastClose goes out at once, but the source is
117+// destroyed only after the frame's CastFrameDone, and its buffer released.
118+func TestBusLossWaitsForFrameInFlight(t *testing.T) {
119+ l := startLiveCast(t, true)
120+ if cmd := func() ports.CastCommand { l.h.killBus(); return recvCmd(t, l.h) }(); cmd.Kind != ports.CastClose || cmd.ID != 1 {
121+ t.Fatalf("%+v", cmd)
122+ }
123+ select {
124+ case <-l.h.closed:
125+ t.Fatal("PipeWire closed with a frame in flight")
126+ case <-time.After(200 * time.Millisecond):
127+ }
128+ released := make(chan struct{})
129+ l.src.EXPECT().Release(l.buf).Run(func(*pipewire.VideoBuffer) { close(released) }).Once()
130+ l.src.EXPECT().Destroy().Return().Once()
131+ l.h.svc.Push(ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 2})
132+ waitClosed(t, l.h)
133+ select {
134+ case <-released:
135+ default:
136+ t.Fatal("buffer not released")
137+ }
138+}
@@ -0,0 +1,138 @@
1+package screencast
2+
3+import (
4+ "testing"
5+ "time"
6+
7+ "github.com/bnema/neferwl/internal/ports"
8+ "github.com/bnema/purego-pipewire/pipewire"
9+ "github.com/stretchr/testify/mock"
10+)
11+
12+// liveCast is a cast on a private bus, brought to STREAMING with a mocked
13+// PipeWire source, so a test can lose the bus under it.
14+type liveCast struct {
15+ h *svcHarness
16+ src *mockvideoSource
17+ buf *pipewire.VideoBuffer
18+ // onLoop runs functions inside the PipeWire loop iteration, where the
19+ // real handler callbacks run.
20+ onLoop chan func()
21+ hdl chan pipewire.VideoHandler
22+}
23+
24+func (l *liveCast) loop(f func()) {
25+ l.onLoop <- f
26+ l.h.svc.Push(ports.CastEvent{Kind: ports.CastDamage, ID: 1 << 40}) // wakes the owner
27+}
28+
29+func startLiveCast(t *testing.T, withFrame bool) *liveCast {
30+ t.Helper()
31+ l := &liveCast{onLoop: make(chan func(), 8), hdl: make(chan pipewire.VideoHandler, 1)}
32+ l.h, _ = ready(t, harnessOpts{iterate: func() {
33+ select {
34+ case f := <-l.onLoop:
35+ f()
36+ default:
37+ }
38+ }})
39+ h := l.h
40+ c := h.client()
41+ l.src = newMockvideoSource(t)
42+ l.buf = &pipewire.VideoBuffer{Index: 2, Data: make([]byte, 64*32*4), Stride: 256, Width: 64, Height: 32, Format: pipewire.VideoFormatBGRx}
43+ h.core.EXPECT().NewVideoSource(mock.Anything, mock.Anything).RunAndReturn(
44+ func(_ pipewire.VideoConfig, hd pipewire.VideoHandler) (videoSource, error) {
45+ l.hdl <- hd
46+ return l.src, nil
47+ }).Once()
48+ if withFrame {
49+ l.src.EXPECT().Dequeue().Return(l.buf).Once()
50+ } else {
51+ l.src.EXPECT().Dequeue().Return(nil).Maybe()
52+ }
53+
54+ sess := createSession(t, c)
55+ if name, _ := recordMonitor(c, sess, "HEADLESS-1"); name != "" {
56+ t.Fatal(name)
57+ }
58+ if name, _ := callErr(c, nameScreenCast, sess, ifaceSession, "Start", "", nil); name != "" {
59+ t.Fatal(name)
60+ }
61+ if cmd := recvCmd(t, h); cmd.Kind != ports.CastOpen {
62+ t.Fatalf("%+v", cmd)
63+ }
64+ h.svc.Push(ports.CastEvent{Kind: ports.CastSize, ID: 1, Width: 64, Height: 32})
65+ var hd pipewire.VideoHandler
66+ select {
67+ case hd = <-l.hdl:
68+ case <-time.After(5 * time.Second):
69+ t.Fatal("no PipeWire source")
70+ }
71+ l.loop(func() {
72+ hd.StateChanged(pipewire.StreamStatePaused, 42, "")
73+ hd.FormatChanged(pipewire.VideoFormatBGRx, 64, 32, 256)
74+ hd.StateChanged(pipewire.StreamStateStreaming, 42, "")
75+ })
76+ if withFrame {
77+ if cmd := recvCmd(t, h); cmd.Kind != ports.CastFrame || cmd.Buf != 2 {
78+ t.Fatalf("%+v", cmd)
79+ }
80+ }
81+ return l
82+}
83+
84+func recvCmd(t *testing.T, h *svcHarness) ports.CastCommand {
85+ t.Helper()
86+ select {
87+ case c := <-h.cmds:
88+ return c
89+ case <-time.After(5 * time.Second):
90+ t.Fatal("no command\n" + h.log.String())
91+ return ports.CastCommand{}
92+ }
93+}
94+
95+func waitClosed(t *testing.T, h *svcHarness) {
96+ t.Helper()
97+ select {
98+ case <-h.closed:
99+ case <-time.After(5 * time.Second):
100+ t.Fatal("PipeWire not closed\n" + h.log.String())
101+ }
102+}
103+
104+// Losing the bus with a live stream and no frame in flight: the cast is
105+// closed on the wayland side, the source destroyed, PipeWire closed.
106+func TestBusLossTearsDownStream(t *testing.T) {
107+ l := startLiveCast(t, false)
108+ l.src.EXPECT().Destroy().Return().Once()
109+ l.h.killBus()
110+ if cmd := recvCmd(t, l.h); cmd.Kind != ports.CastClose || cmd.ID != 1 {
111+ t.Fatalf("%+v", cmd)
112+ }
113+ waitClosed(t, l.h)
114+}
115+
116+// With a frame in flight: CastClose goes out at once, but the source is
117+// destroyed only after the frame's CastFrameDone, and its buffer released.
118+func TestBusLossWaitsForFrameInFlight(t *testing.T) {
119+ l := startLiveCast(t, true)
120+ if cmd := func() ports.CastCommand { l.h.killBus(); return recvCmd(t, l.h) }(); cmd.Kind != ports.CastClose || cmd.ID != 1 {
121+ t.Fatalf("%+v", cmd)
122+ }
123+ select {
124+ case <-l.h.closed:
125+ t.Fatal("PipeWire closed with a frame in flight")
126+ case <-time.After(200 * time.Millisecond):
127+ }
128+ released := make(chan struct{})
129+ l.src.EXPECT().Release(l.buf).Run(func(*pipewire.VideoBuffer) { close(released) }).Once()
130+ l.src.EXPECT().Destroy().Return().Once()
131+ l.h.svc.Push(ports.CastEvent{Kind: ports.CastFrameDone, ID: 1, Buf: 2})
132+ waitClosed(t, l.h)
133+ select {
134+ case <-released:
135+ default:
136+ t.Fatal("buffer not released")
137+ }
138+}

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