1package wayland
2
3import (
4 "context"
5 "errors"
6 "image"
7 "os"
8 "testing"
9 "time"
10
11 "github.com/bnema/neferwl/internal/adapters/capture"
12 "github.com/bnema/neferwl/internal/adapters/sessionsecurity"
13 "github.com/bnema/neferwl/internal/logging"
14 portsmocks "github.com/bnema/neferwl/internal/mocks/ports"
15 "github.com/bnema/neferwl/internal/ports"
16 "github.com/stretchr/testify/mock"
17 "golang.org/x/sys/unix"
18)
19
20// Real server fixture: channels are wired before Run, so the completion loop
21// cannot latch a nil channel. No backend or protocol handler is replaced.
22type barrierServer struct {
23 s *Server
24 gate *sessionsecurity.Gate
25 changes chan ports.SecurityState
26 captures chan ports.CaptureRequest
27 captured chan ports.CaptureDone
28 cancel context.CancelFunc
29 stopped chan struct{}
30 dir string
31}
32
33func newBarrierServer(t *testing.T) *barrierServer {
34 t.Helper()
35 h := &barrierServer{gate: &sessionsecurity.Gate{}, changes: make(chan ports.SecurityState, 1), captures: make(chan ports.CaptureRequest, 4), captured: make(chan ports.CaptureDone, 4), stopped: make(chan struct{}), dir: t.TempDir()}
36 layout := ports.Layout{{Info: ports.OutputInfo{Name: "HEADLESS-1", Width: 2, Height: 2}, Width: 2, Height: 2, Scale: 1}}
37 var err error
38 h.s, err = New(Options{RuntimeDir: h.dir, Outputs: layout, Security: h.gate, CaptureAllow: allowStore(t, "*\n")}, Channels{SecurityChanges: h.changes, SecurityEvents: make(chan ports.SecurityBackendEvent, 16), Captures: h.captures, Captured: h.captured}, logging.For(context.Background(), "wayland"))
39 if err != nil {
40 t.Fatal(err)
41 }
42 ctx, cancel := context.WithCancel(context.Background())
43 h.cancel = cancel
44 var runErr error
45 go func() { runErr = h.s.Run(ctx); close(h.stopped) }()
46 t.Cleanup(func() {
47 cancel()
48 select {
49 case <-h.stopped:
50 if runErr != nil {
51 t.Error(runErr)
52 }
53 case <-time.After(3 * time.Second):
54 t.Error("barrier server shutdown blocked")
55 }
56 })
57 return h
58}
59
60func TestSessionLockBarrierWaitsForAdmittedCopyAfterFrameDeath(t *testing.T) {
61 h := newBarrierServer(t)
62 // Reuse the real raw-wire admission client, retaining its SHM memfd so a
63 // write performed after lock admission can be observed independently.
64 client := protocolClient(t, h.s, h.dir)
65 output := bindVersion(t, client, "wl_output", 4)
66 registerProtocol(t, client, output)
67 _, buffer, fd := captureSmallBuffer(t, client)
68 defer unix.Close(fd)
69 a := &admissionClient{c: client, wlr: bindVersion(t, client, "zwlr_screencopy_manager_v1", 3), output: output, buf: buffer}
70 frame := a.captureWlr(t)
71 var request ports.CaptureRequest
72 select {
73 case request = <-h.captures:
74 case <-time.After(2 * time.Second):
75 t.Fatal("actual capture not admitted")
76 }
77 // Pipeline owns the dup'd destination from this point, including closure.
78 renderer := portsmocks.NewMockRenderer(t)
79 pixels := portsmocks.NewMockCaptureFrame(t)
80 entered, release := make(chan struct{}), make(chan struct{})
81 pixels.EXPECT().Done().Return(nil)
82 pixels.EXPECT().Read(image.Rect(0, 0, 2, 2), mock.Anything, 8).RunAndReturn(func(_ image.Rectangle, dst []byte, _ int) error {
83 close(entered)
84 <-release
85 copy(dst, []byte{0x42, 0x31, 0x20, 0xff})
86 return nil
87 }).Once()
88 renderer.EXPECT().BeginCapture().Return(pixels, nil).Once()
89 renderer.EXPECT().EndCapture(pixels).Return().Once()
90 pipelineReplies := make(chan ports.CaptureDone, 1)
91 pipeline := capture.NewPipeline(context.Background(), pipelineReplies)
92 pipeline.Security = h.gate
93 released := false
94 t.Cleanup(func() {
95 if !released {
96 close(release)
97 }
98 pipeline.Close(renderer)
99 // testify records the mmap-backed Read argument. Forget its call
100 // history after joining the worker: expectation counters still verify
101 // Once, but cleanup must not format bytes that are now unmapped.
102 pixels.Calls = nil // worker joined; follows capture/pipeline_test.go
103 })
104 pipeline.SubmitScoped(h.gate.Snapshot(), renderer, []ports.CaptureRequest{request})
105 select {
106 case <-entered:
107 case <-time.After(2 * time.Second):
108 t.Fatal("capture worker did not enter Read")
109 }
110 a.destroy(t, frame)
111 locker := protocolClient(t, h.s, h.dir)
112 _, locked := lifecycleAcquire(t, locker)
113 state := lifecycleTransition(t, h.changes)
114 h.s.display.Do(func() {
115 h.s.applySecurityEvent(ports.SecurityOutputAdded{Instance: 1, Output: "HEADLESS-1"})
116 h.s.applySecurityEvent(ports.SecurityBackendBarrier{Generation: state.Generation})
117 lifecycleProof(h.s, state, 1)
118 if len(h.s.captureInflight) != 1 || h.s.sessionLock.locked {
119 t.Error("destroyed resource bypassed held copy barrier")
120 }
121 })
122 roundtrip(t, locker)
123 if *locked != 0 {
124 t.Fatal("locked while Read can still write normal pixels")
125 }
126 close(release)
127 released = true
128 var completion ports.CaptureDone
129 select {
130 case completion = <-pipelineReplies:
131 case <-time.After(2 * time.Second):
132 t.Fatal("capture pipeline did not complete")
133 }
134 if completion.Err != nil || completion.ID != request.ID {
135 t.Fatalf("capture completion %+v", completion)
136 }
137 if _, err := request.Dst.File.Stat(); !errors.Is(err, os.ErrClosed) {
138 t.Fatalf("completion precedes destination close: %v", err)
139 }
140 got := make([]byte, 4)
141 if _, err := unix.Pread(fd, got, 0); err != nil {
142 t.Fatal(err)
143 }
144 if got[0] != 0x42 || got[3] != 0xff {
145 t.Fatalf("actual copy not visible: %x", got)
146 }
147 h.s.display.Do(func() {
148 if len(h.s.captureInflight) != 1 || h.s.sessionLock.locked {
149 t.Error("worker completion not forwarded yet must retain credit")
150 }
151 })
152 // Reply channel is the sole production completion funnel. Closing a frame
153 // never synthesizes this reply or cancels its outstanding memory write.
154 processed := make(chan struct{})
155 h.s.display.Do(func() {
156 original := h.s.captureReplies[request.ID]
157 h.s.captureReplies[request.ID] = func(done ports.CaptureDone) {
158 if original != nil {
159 original(done)
160 } // frame destruction removes the wire reply, not credit
161 close(processed)
162 }
163 })
164 h.captured <- completion
165 select {
166 case <-processed:
167 case <-time.After(2 * time.Second):
168 t.Fatal("server did not process actual capture completion")
169 }
170 h.s.display.Do(func() {
171 if len(h.s.captureInflight) != 0 || !h.s.sessionLock.locked {
172 t.Error("completed copy did not clear lock barrier")
173 }
174 })
175 // The opaque completed token returns the real capture lease on its owner.
176 select {
177 case token := <-pipeline.Completed():
178 pipeline.Recycle(token, renderer)
179 case <-time.After(2 * time.Second):
180 t.Fatal("capture lease not returned")
181 }
182 roundtrip(t, locker)
183 if *locked != 1 {
184 t.Fatalf("locked events=%d", *locked)
185 }
186}
187
188func TestSessionLockBarrierTakeoverSaturationRefusesWithoutGenerationChange(t *testing.T) {
189 h := newBarrierServer(t)
190 sentinel := ports.SecurityState{Generation: 999}
191 h.changes <- sentinel // authoritative backend stalled for the whole churn
192 for i := 0; i < maxSecurityPending; i++ {
193 client := protocolClient(t, h.s, h.dir)
194 _, locked := lifecycleAcquire(t, client)
195 if *locked != 0 {
196 t.Fatal("stalled backend confirmed lock")
197 }
198 orphaned := make(chan struct{})
199 h.s.display.Do(func() { r := h.s.sessionLock.res; old := r.OnDestroy; r.OnDestroy = func() { old(); close(orphaned) } })
200 if err := client.Close(); err != nil {
201 t.Fatal(err)
202 }
203 select {
204 case <-orphaned:
205 case <-time.After(2 * time.Second):
206 t.Fatal("owner death not processed")
207 }
208 }
209 before := h.gate.Snapshot()
210 client := protocolClient(t, h.s, h.dir)
211 refused, err := bindLockManager(t, client).Lock()
212 if err != nil {
213 t.Fatal(err)
214 }
215 finished := 0
216 refused.OnFinished(func() { finished++ })
217 roundtrip(t, client)
218 h.s.display.Do(func() {
219 if len(h.s.securityPending) != maxSecurityPending || h.s.sessionLock != nil {
220 t.Error("stalled outbox exceeded bound or admitted owner")
221 }
222 })
223 if finished != 1 || h.gate.Snapshot() != before || !before.Protected {
224 t.Fatalf("refusal finished=%d before=%+v after=%+v", finished, before, h.gate.Snapshot())
225 }
226}
227
228func TestSessionLockBarrierReservedReleaseAndBlockedSenderShutdown(t *testing.T) {
229 h := newBarrierServer(t)
230 sentinel := ports.SecurityState{Generation: 999}
231 h.changes <- sentinel
232 client := protocolClient(t, h.s, h.dir)
233 lock, locked := lifecycleAcquire(t, client)
234 state := h.gate.Snapshot()
235 var queued []ports.SecurityState
236 h.s.display.Do(func() {
237 // Immutable channel messages exercise the reserved release slot without
238 // fabricating another accepted owner while this one is live.
239 for len(h.s.securityPending) < maxSecurityPending {
240 h.s.requestBackendSecurity(ports.SecurityState{Generation: ports.LockGeneration(100 + len(h.s.securityPending)), Protected: true})
241 }
242 queued = append(queued, h.s.securityPending...)
243 h.s.applySecurityEvent(ports.SecurityBackendBarrier{Generation: state.Generation})
244 })
245 roundtrip(t, client)
246 if *locked != 1 {
247 t.Fatal("owner not confirmed")
248 }
249 if err := lock.UnlockAndDestroy(); err != nil {
250 t.Fatal(err)
251 }
252 roundtrip(t, client)
253 released := h.gate.Snapshot()
254 h.s.display.Do(func() {
255 if len(h.s.securityPending) != maxSecurityPending+1 || h.s.securityPending[maxSecurityPending] != released {
256 t.Error("reserved release slot lost")
257 }
258 })
259 if got := lifecycleTransition(t, h.changes); got != sentinel {
260 t.Fatal("backend sentinel changed")
261 }
262 for _, want := range append(queued, released) {
263 if got := lifecycleTransition(t, h.changes); got != want {
264 t.Fatalf("outbox got %+v want %+v", got, want)
265 }
266 }
267 // Block another transition and cancel the real server. Run joins the sender;
268 // successful return proves it did not leak in a blocked channel send.
269 h.changes <- sentinel
270 h.s.display.Do(func() { h.s.requestBackendSecurity(ports.SecurityState{Generation: 777, Protected: true}) })
271 h.cancel()
272 select {
273 case <-h.stopped:
274 case <-time.After(3 * time.Second):
275 t.Fatal("blocked backend sender prevented Run shutdown")
276 }
277}