fix: give each connection ownership of its received descriptors

Ancillary file descriptors were queued in a package-global ring shared by
every Display, capped at four per read, consumed in an order unrelated to
the connection that received them, and never closed.

Descriptors now belong to the Display that received them: readChunk parses
every SCM_RIGHTS control message into a per-connection FIFO, Event.Fd pops
from that connection in arrival order, a read that would need more than 64
descriptors is reported as a protocol failure instead of dropping them
silently, and queued descriptors are closed on protocol failure and on
Close. GetNextFD and the lock-free global queue are removed, and the unused
conn.File duplicate is gone.

Tests cover single and multiple descriptors per control message, an event
without descriptors after one with descriptors, two concurrently connected
displays, truncated control data, closure of unconsumed descriptors on
shutdown and after a malformed frame, and a 100-connection descriptor
leak loop.

commit: ca248e67adf46bd1a715e64f8570fa9385b40826
author: brice · 39d ago · 2026-09-13 06:18 UTC
parent: d09a309
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +607 -119

M client.go +30 -9


@@ -66,12 +66,15 @@ type Display struct {
66	66
67	67		// Connection-local receive state. rbuf holds bytes read from the socket
68	68		// that do not yet form a complete frame; recvBuf is the scratch buffer
69		-	// used for a single read.
70		-	rbuf []byte
71		-	recvBuf []byte
72		-	readErr error
73		-	maxMsgLen uint32
74		-	closed atomic.Bool
	69	+	// used for a single read. pendingFDs holds ancillary descriptors that
	70	+	// arrived with those bytes and have not been consumed yet.
	71	+	rbuf []byte
	72	+	recvBuf []byte
	73	+	readErr error
	74	+	maxMsgLen uint32
	75	+	closed atomic.Bool
	76	+	fdMu sync.Mutex
	77	+	pendingFDs []int
75	78		}
76	79
77	80		// newDisplay builds a Display over an established connection.

@@ -176,6 +179,7 @@ func (d *Display) Close() error {
176	179		if !d.closed.CompareAndSwap(false, true) {
177	180		return nil
178	181		}
	182	+	d.closePendingFDs()
179	183		return d.conn.Close()
180	184		}
181	185

@@ -326,9 +330,23 @@ func (d *Display) Dispatch() error {
326	330
327	331		frame, err := d.nextFrame()
328	332		if err != nil {
329		-	return err
	333	+	return d.fail(err)
	334	+	}
	335	+	return d.fail(d.dispatchFrame(frame))
	336	+	}
	337	+
	338	+	// fail releases resources that cannot be used once the connection is
	339	+	// unusable. A protocol violation invalidates the whole connection, so queued
	340	+	// descriptors are closed; transport errors may be transient and keep them.
	341	+	func (d *Display) fail(err error) error {
	342	+	if err == nil {
	343	+	return nil
	344	+	}
	345	+	var perr *ProtocolError
	346	+	if errors.As(err, &perr) {
	347	+	d.closePendingFDs()
330	348		}
331		-	return d.dispatchFrame(frame)
	349	+	return err
332	350		}
333	351
334	352		// maxMessageSize reports the largest accepted message size.

@@ -388,7 +406,10 @@ func (d *Display) fill() error {
388	406		err = io.ErrUnexpectedEOF
389	407		}
390	408		d.readErr = err
391		-	if n == 0 {
	409	+	// A control-message failure corrupts the descriptor stream, so it is
	410	+	// reported immediately instead of after the buffered bytes are used.
	411	+	var perr *ProtocolError
	412	+	if n == 0 \|\| errors.As(err, &perr) {
392	413		return err
393	414		}
394	415		}

 	// Connection-local receive state. rbuf holds bytes read from the socket
 	// that do not yet form a complete frame; recvBuf is the scratch buffer
-	// used for a single read.
+	// used for a single read. pendingFDs holds ancillary descriptors that
-	rbuf      []byte
+	// arrived with those bytes and have not been consumed yet.
-	recvBuf   []byte
+	rbuf       []byte
-	readErr   error
+	recvBuf    []byte
-	maxMsgLen uint32
+	readErr    error
-	closed    atomic.Bool
+	maxMsgLen  uint32
+	closed     atomic.Bool
+	fdMu       sync.Mutex
+	pendingFDs []int
 }
 // newDisplay builds a Display over an established connection.
 	if !d.closed.CompareAndSwap(false, true) {
 		return nil
 	}
+	d.closePendingFDs()
 	return d.conn.Close()
 }
 	frame, err := d.nextFrame()
 	if err != nil {
-		return err
+		return d.fail(err)
+	}
+	return d.fail(d.dispatchFrame(frame))
+}
+// fail releases resources that cannot be used once the connection is
+// unusable. A protocol violation invalidates the whole connection, so queued
+// descriptors are closed; transport errors may be transient and keep them.
+func (d *Display) fail(err error) error {
+	if err == nil {
+		return nil
+	}
+	var perr *ProtocolError
+	if errors.As(err, &perr) {
+		d.closePendingFDs()
 	}
-	return d.dispatchFrame(frame)
+	return err
 }
 // maxMessageSize reports the largest accepted message size.
 			err = io.ErrUnexpectedEOF
 		}
 		d.readErr = err
-		if n == 0 {
+		// A control-message failure corrupts the descriptor stream, so it is
+		// reported immediately instead of after the buffered bytes are used.
+		var perr *ProtocolError
+		if n == 0 || errors.As(err, &perr) {
 			return err
 		}
 	}

M core.go +9 -6


@@ -221,14 +221,17 @@ func (e *Event) Array() []byte {
221	221		return arr
222	222		}
223	223
224		-	// Fd reads a file descriptor from the event
	224	+	// Fd reads a file descriptor that was delivered out-of-band for this event.
	225	+	// Descriptors are queued on the connection in arrival order, as the Wayland
	226	+	// wire protocol requires. The returned descriptor is owned by the caller,
	227	+	// which must close it. It returns 0 when the event carried no descriptor.
225	228		func (e *Event) Fd() uintptr {
226		-	// File descriptors are passed out-of-band via SCM_RIGHTS
227		-	// We read them from our lock-free queue
228		-	if fd, ok := GetNextFD(); ok {
229		-	return uintptr(fd)
	229	+	if e.display != nil {
	230	+	if fd, ok := e.display.nextFD(); ok {
	231	+	return uintptr(fd)
	232	+	}
230	233		}
231		-	// Fallback: read placeholder value from message
	234	+	// No descriptor was delivered: consume the placeholder argument.
232	235		_ = e.Uint32()
233	236		return 0
234	237		}

 	return arr
 }
-// Fd reads a file descriptor from the event
+// Fd reads a file descriptor that was delivered out-of-band for this event.
+// Descriptors are queued on the connection in arrival order, as the Wayland
+// wire protocol requires. The returned descriptor is owned by the caller,
+// which must close it. It returns 0 when the event carried no descriptor.
 func (e *Event) Fd() uintptr {
-	// File descriptors are passed out-of-band via SCM_RIGHTS
+	if e.display != nil {
-	// We read them from our lock-free queue
+		if fd, ok := e.display.nextFD(); ok {
-	if fd, ok := GetNextFD(); ok {
+			return uintptr(fd)
-		return uintptr(fd)
+		}
 	}
-	// Fallback: read placeholder value from message
+	// No descriptor was delivered: consume the placeholder argument.
 	_ = e.Uint32()
 	return 0
 }

A fd_test.go +483 -0

@@ -0,0 +1,483 @@
	1	+	//go:build linux
	2	+	// +build linux
	3	+
	4	+	package wlturbo
	5	+
	6	+	import (
	7	+	"errors"
	8	+	"io"
	9	+	"net"
	10	+	"os"
	11	+	"testing"
	12	+	"time"
	13	+
	14	+	"golang.org/x/sys/unix"
	15	+	)
	16	+
	17	+	// fdRecordingProxy consumes the file descriptor of every event it receives.
	18	+	// Events without a descriptor are left untouched.
	19	+	type fdRecordingProxy struct {
	20	+	BaseProxy
	21	+	fds *[]int
	22	+	}
	23	+
	24	+	func (p fdRecordingProxy) Dispatch(event Event) {
	25	+	if fd := event.Fd(); fd != 0 {
	26	+	p.fds = append(p.fds, int(fd))
	27	+	return
	28	+	}
	29	+	// No descriptor arrived. Fd already consumed the placeholder argument.
	30	+	}
	31	+
	32	+	// fdIgnoringProxy never reads the descriptor of an event.
	33	+	type fdIgnoringProxy struct {
	34	+	BaseProxy
	35	+	}
	36	+
	37	+	func (p fdIgnoringProxy) Dispatch(Event) {}
	38	+
	39	+	// unixSocketPair returns a connected socket pair as net.UnixConns.
	40	+	func unixSocketPair(t testing.T) (client, peer net.UnixConn) {
	41	+	t.Helper()
	42	+
	43	+	fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM\|unix.SOCK_CLOEXEC, 0)
	44	+	if err != nil {
	45	+	t.Fatalf("socketpair: %v", err)
	46	+	}
	47	+
	48	+	clientFile := os.NewFile(uintptr(fds[0]), "client")
	49	+	peerFile := os.NewFile(uintptr(fds[1]), "peer")
	50	+
	51	+	clientConn, err := net.FileConn(clientFile)
	52	+	clientFile.Close()
	53	+	if err != nil {
	54	+	peerFile.Close()
	55	+	t.Fatalf("net.FileConn(client): %v", err)
	56	+	}
	57	+
	58	+	peerConn, err := net.FileConn(peerFile)
	59	+	peerFile.Close()
	60	+	if err != nil {
	61	+	clientConn.Close()
	62	+	t.Fatalf("net.FileConn(peer): %v", err)
	63	+	}
	64	+
	65	+	t.Cleanup(func() {
	66	+	clientConn.Close()
	67	+	peerConn.Close()
	68	+	})
	69	+
	70	+	return clientConn.(net.UnixConn), peerConn.(net.UnixConn)
	71	+	}
	72	+
	73	+	// sendMessageWithFDs writes one framed message and attaches fds to it.
	74	+	func sendMessageWithFDs(t testing.T, peer net.UnixConn, msg []byte, fds []int) {
	75	+	t.Helper()
	76	+
	77	+	var oob []byte
	78	+	if len(fds) > 0 {
	79	+	oob = unix.UnixRights(fds...)
	80	+	}
	81	+	if _, _, err := peer.WriteMsgUnix(msg, oob, nil); err != nil {
	82	+	t.Fatalf("WriteMsgUnix: %v", err)
	83	+	}
	84	+	}
	85	+
	86	+	// newFDDisplay builds a display over the client end with a proxy registered at
	87	+	// object ID 7 and returns the descriptors that proxy consumes.
	88	+	func newFDDisplay(t testing.T, client net.UnixConn) (Display, []int) {
	89	+	t.Helper()
	90	+
	91	+	d := newDisplay(client)
	92	+	fds := []int{}
	93	+	d.objects.Store(uint32(7), &fdRecordingProxy{
	94	+	BaseProxy: BaseProxy{id: 7, context: d.context},
	95	+	fds: &fds,
	96	+	})
	97	+	return d, &fds
	98	+	}
	99	+
	100	+	// writeSentinel writes a unique byte string through fd and closes the caller's
	101	+	// copy of the descriptor. Reading the same sentinel from the paired read end
	102	+	// proves the descriptor really is the expected one.
	103	+	func writeSentinel(t *testing.T, fd uintptr, sentinel string) {
	104	+	t.Helper()
	105	+	if _, err := unix.Write(int(fd), []byte(sentinel)); err != nil {
	106	+	t.Fatalf("write sentinel through received fd: %v", err)
	107	+	}
	108	+	if err := unix.Close(int(fd)); err != nil {
	109	+	t.Fatalf("close received fd: %v", err)
	110	+	}
	111	+	}
	112	+
	113	+	// readSentinel reads exactly len(sentinel) bytes and compares them.
	114	+	func readSentinel(t testing.T, r os.File, sentinel string) {
	115	+	t.Helper()
	116	+	buf := make([]byte, len(sentinel))
	117	+	if _, err := io.ReadFull(r, buf); err != nil {
	118	+	t.Fatalf("read sentinel: %v", err)
	119	+	}
	120	+	if string(buf) != sentinel {
	121	+	t.Fatalf("sentinel = %q, want %q", buf, sentinel)
	122	+	}
	123	+	}
	124	+
	125	+	func TestDisplayFD_Single(t *testing.T) {
	126	+	client, peer := unixSocketPair(t)
	127	+	d, received := newFDDisplay(t, client)
	128	+
	129	+	r, w, err := os.Pipe()
	130	+	if err != nil {
	131	+	t.Fatalf("pipe: %v", err)
	132	+	}
	133	+	defer r.Close()
	134	+
	135	+	sendMessageWithFDs(t, peer, message(7, 3, []byte{0, 0, 0, 0}), []int{int(w.Fd())})
	136	+	w.Close()
	137	+
	138	+	if err := d.Dispatch(); err != nil {
	139	+	t.Fatalf("Dispatch: %v", err)
	140	+	}
	141	+	if len(*received) != 1 {
	142	+	t.Fatalf("received %d descriptors, want 1", len(*received))
	143	+	}
	144	+
	145	+	writeSentinel(t, uintptr((*received)[0]), "single-fd-sentinel")
	146	+	readSentinel(t, r, "single-fd-sentinel")
	147	+	}
	148	+
	149	+	func TestDisplayFD_MultipleInOneControlMessage(t *testing.T) {
	150	+	client, peer := unixSocketPair(t)
	151	+	d, received := newFDDisplay(t, client)
	152	+
	153	+	firstR, firstW, err := os.Pipe()
	154	+	if err != nil {
	155	+	t.Fatalf("pipe: %v", err)
	156	+	}
	157	+	defer firstR.Close()
	158	+	secondR, secondW, err := os.Pipe()
	159	+	if err != nil {
	160	+	t.Fatalf("pipe: %v", err)
	161	+	}
	162	+	defer secondR.Close()
	163	+
	164	+	sendMessageWithFDs(t, peer, message(7, 3, []byte{0, 0, 0, 0}),
	165	+	[]int{int(firstW.Fd()), int(secondW.Fd())})
	166	+	firstW.Close()
	167	+	secondW.Close()
	168	+
	169	+	if err := d.Dispatch(); err != nil {
	170	+	t.Fatalf("Dispatch: %v", err)
	171	+	}
	172	+
	173	+	// A handler consumes one descriptor per fd argument, so the test drains
	174	+	// whatever the connection still holds, in arrival order.
	175	+	for {
	176	+	fd, ok := d.nextFD()
	177	+	if !ok {
	178	+	break
	179	+	}
	180	+	received = append(received, fd)
	181	+	}
	182	+
	183	+	if len(*received) != 2 {
	184	+	t.Fatalf("received %d descriptors, want 2", len(*received))
	185	+	}
	186	+
	187	+	writeSentinel(t, uintptr((*received)[0]), "first-pipe")
	188	+	writeSentinel(t, uintptr((*received)[1]), "second-pipe")
	189	+
	190	+	readSentinel(t, firstR, "first-pipe")
	191	+	readSentinel(t, secondR, "second-pipe")
	192	+	}
	193	+
	194	+	func TestDisplayFD_EventWithoutFDsAfterOneWithFDs(t *testing.T) {
	195	+	client, peer := unixSocketPair(t)
	196	+	d, received := newFDDisplay(t, client)
	197	+
	198	+	r, w, err := os.Pipe()
	199	+	if err != nil {
	200	+	t.Fatalf("pipe: %v", err)
	201	+	}
	202	+	defer r.Close()
	203	+
	204	+	// First event carries a descriptor.
	205	+	sendMessageWithFDs(t, peer, message(7, 3, []byte{0, 0, 0, 0}), []int{int(w.Fd())})
	206	+	w.Close()
	207	+	if err := d.Dispatch(); err != nil {
	208	+	t.Fatalf("first Dispatch: %v", err)
	209	+	}
	210	+
	211	+	// Second event carries none and must not steal the first one.
	212	+	sendMessageWithFDs(t, peer, message(7, 4, []byte{0, 0, 0, 0}), nil)
	213	+	if err := d.Dispatch(); err != nil {
	214	+	t.Fatalf("second Dispatch: %v", err)
	215	+	}
	216	+
	217	+	if len(*received) != 1 {
	218	+	t.Fatalf("received %d descriptors, want exactly 1", len(*received))
	219	+	}
	220	+
	221	+	writeSentinel(t, uintptr((*received)[0]), "only-fd")
	222	+	readSentinel(t, r, "only-fd")
	223	+	}
	224	+
	225	+	func TestDisplayFD_TwoDisplaysAreIsolated(t *testing.T) {
	226	+	firstClient, firstPeer := unixSocketPair(t)
	227	+	secondClient, secondPeer := unixSocketPair(t)
	228	+
	229	+	firstDisplay, firstReceived := newFDDisplay(t, firstClient)
	230	+	secondDisplay, secondReceived := newFDDisplay(t, secondClient)
	231	+
	232	+	firstR, firstW, err := os.Pipe()
	233	+	if err != nil {
	234	+	t.Fatalf("pipe: %v", err)
	235	+	}
	236	+	defer firstR.Close()
	237	+	secondR, secondW, err := os.Pipe()
	238	+	if err != nil {
	239	+	t.Fatalf("pipe: %v", err)
	240	+	}
	241	+	defer secondR.Close()
	242	+
	243	+	sendMessageWithFDs(t, firstPeer, message(7, 3, []byte{0, 0, 0, 0}), []int{int(firstW.Fd())})
	244	+	firstW.Close()
	245	+	sendMessageWithFDs(t, secondPeer, message(7, 3, []byte{0, 0, 0, 0}), []int{int(secondW.Fd())})
	246	+	secondW.Close()
	247	+
	248	+	if err := secondDisplay.Dispatch(); err != nil {
	249	+	t.Fatalf("second Dispatch: %v", err)
	250	+	}
	251	+	if err := firstDisplay.Dispatch(); err != nil {
	252	+	t.Fatalf("first Dispatch: %v", err)
	253	+	}
	254	+
	255	+	// Each connection must only see its own descriptor.
	256	+	if len(*firstReceived) != 1 {
	257	+	t.Fatalf("first display received %d descriptors, want 1", len(*firstReceived))
	258	+	}
	259	+	if len(*secondReceived) != 1 {
	260	+	t.Fatalf("second display received %d descriptors, want 1", len(*secondReceived))
	261	+	}
	262	+
	263	+	writeSentinel(t, uintptr((*secondReceived)[0]), "second-display")
	264	+	writeSentinel(t, uintptr((*firstReceived)[0]), "first-display")
	265	+
	266	+	readSentinel(t, secondR, "second-display")
	267	+	readSentinel(t, firstR, "first-display")
	268	+	}
	269	+
	270	+	func TestDisplayFD_TruncatedControlData(t *testing.T) {
	271	+	client, peer := unixSocketPair(t)
	272	+	d, _ := newFDDisplay(t, client)
	273	+
	274	+	// One more descriptor than a single read can carry.
	275	+	devNull, err := os.Open(os.DevNull)
	276	+	if err != nil {
	277	+	t.Fatalf("open %s: %v", os.DevNull, err)
	278	+	}
	279	+	defer devNull.Close()
	280	+
	281	+	fds := make([]int, 0, maxRecvFDs+1)
	282	+	for i := 0; i < maxRecvFDs+1; i++ {
	283	+	dup, err := unix.Dup(int(devNull.Fd()))
	284	+	if err != nil {
	285	+	t.Fatalf("dup: %v", err)
	286	+	}
	287	+	t.Cleanup(func() { _ = unix.Close(dup) })
	288	+	fds = append(fds, dup)
	289	+	}
	290	+
	291	+	sendMessageWithFDs(t, peer, message(7, 3, []byte{0, 0, 0, 0}), fds)
	292	+
	293	+	err = d.Dispatch()
	294	+	if !errors.Is(err, ErrMalformedFrame) {
	295	+	t.Fatalf("Dispatch error = %v, want %v", err, ErrMalformedFrame)
	296	+	}
	297	+
	298	+	var perr *ProtocolError
	299	+	if !errors.As(err, &perr) {
	300	+	t.Fatalf("Dispatch error type = %T, want *ProtocolError", err)
	301	+	}
	302	+	if perr.Kind != "control_truncated" {
	303	+	t.Fatalf("error kind = %q, want %q", perr.Kind, "control_truncated")
	304	+	}
	305	+
	306	+	// A truncated control message must not leave descriptors queued.
	307	+	d.fdMu.Lock()
	308	+	pending := len(d.pendingFDs)
	309	+	d.fdMu.Unlock()
	310	+	if pending != 0 {
	311	+	t.Fatalf("%d descriptors still queued after a truncated control message", pending)
	312	+	}
	313	+	}
	314	+
	315	+	func TestDisplayFD_UnconsumedDescriptorClosedOnShutdown(t *testing.T) {
	316	+	client, peer := unixSocketPair(t)
	317	+	d := newDisplay(client)
	318	+	d.objects.Store(uint32(7), &fdIgnoringProxy{
	319	+	BaseProxy: BaseProxy{id: 7, context: d.context},
	320	+	})
	321	+
	322	+	r, w, err := os.Pipe()
	323	+	if err != nil {
	324	+	t.Fatalf("pipe: %v", err)
	325	+	}
	326	+	defer r.Close()
	327	+
	328	+	sendMessageWithFDs(t, peer, message(7, 3, []byte{0, 0, 0, 0}), []int{int(w.Fd())})
	329	+	// Drop our own copy: the display now holds the only writer.
	330	+	w.Close()
	331	+
	332	+	if err := d.Dispatch(); err != nil {
	333	+	t.Fatalf("Dispatch: %v", err)
	334	+	}
	335	+
	336	+	// While the display still holds the descriptor the pipe stays open.
	337	+	if err := r.SetReadDeadline(time.Now().Add(200 * time.Millisecond)); err != nil {
	338	+	t.Fatalf("SetReadDeadline: %v", err)
	339	+	}
	340	+	if _, err := r.Read(make([]byte, 1)); !errors.Is(err, os.ErrDeadlineExceeded) {
	341	+	t.Fatalf("read with the display holding the descriptor = %v, want %v", err, os.ErrDeadlineExceeded)
	342	+	}
	343	+
	344	+	// Closing the display must release it, so the read end sees EOF.
	345	+	if err := d.Close(); err != nil {
	346	+	t.Fatalf("Close: %v", err)
	347	+	}
	348	+	if err := r.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil {
	349	+	t.Fatalf("SetReadDeadline: %v", err)
	350	+	}
	351	+	if _, err := r.Read(make([]byte, 1)); !errors.Is(err, io.EOF) {
	352	+	t.Fatalf("read after Close = %v, want %v", err, io.EOF)
	353	+	}
	354	+	}
	355	+
	356	+	// rawSocketPair returns a connected socket pair without registering cleanup,
	357	+	// so loop tests can close descriptors deterministically.
	358	+	func rawSocketPair(t testing.T) (client, peer net.UnixConn) {
	359	+	t.Helper()
	360	+
	361	+	fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM\|unix.SOCK_CLOEXEC, 0)
	362	+	if err != nil {
	363	+	t.Fatalf("socketpair: %v", err)
	364	+	}
	365	+
	366	+	clientFile := os.NewFile(uintptr(fds[0]), "client")
	367	+	clientConn, err := net.FileConn(clientFile)
	368	+	clientFile.Close()
	369	+	if err != nil {
	370	+	t.Fatalf("net.FileConn(client): %v", err)
	371	+	}
	372	+
	373	+	peerFile := os.NewFile(uintptr(fds[1]), "peer")
	374	+	peerConn, err := net.FileConn(peerFile)
	375	+	peerFile.Close()
	376	+	if err != nil {
	377	+	t.Fatalf("net.FileConn(peer): %v", err)
	378	+	}
	379	+
	380	+	return clientConn.(net.UnixConn), peerConn.(net.UnixConn)
	381	+	}
	382	+
	383	+	// openFDs reports how many descriptors the process holds.
	384	+	func openFDs(t *testing.T) int {
	385	+	t.Helper()
	386	+
	387	+	entries, err := os.ReadDir("/proc/self/fd")
	388	+	if err != nil {
	389	+	t.Skipf("cannot count open descriptors: %v", err)
	390	+	}
	391	+	return len(entries)
	392	+	}
	393	+
	394	+	func TestDisplayFD_NoLeakAcrossConnections(t *testing.T) {
	395	+	const iterations = 100
	396	+
	397	+	devNull, err := os.Open(os.DevNull)
	398	+	if err != nil {
	399	+	t.Fatalf("open %s: %v", os.DevNull, err)
	400	+	}
	401	+	defer devNull.Close()
	402	+
	403	+	// Warm up so caches and pools are populated before measuring.
	404	+	runFDLeakIteration(t, devNull)
	405	+
	406	+	before := openFDs(t)
	407	+	for i := 0; i < iterations; i++ {
	408	+	runFDLeakIteration(t, devNull)
	409	+	}
	410	+	after := openFDs(t)
	411	+
	412	+	if after > before {
	413	+	t.Fatalf("descriptor count grew from %d to %d over %d connections", before, after, iterations)
	414	+	}
	415	+	}
	416	+
	417	+	// runFDLeakIteration opens a connection, delivers a descriptor nobody reads and
	418	+	// tears everything down.
	419	+	func runFDLeakIteration(t testing.T, devNull os.File) {
	420	+	t.Helper()
	421	+
	422	+	client, peer := rawSocketPair(t)
	423	+	d := newDisplay(client)
	424	+	d.objects.Store(uint32(7), &fdIgnoringProxy{
	425	+	BaseProxy: BaseProxy{id: 7, context: d.context},
	426	+	})
	427	+
	428	+	sent, err := unix.Dup(int(devNull.Fd()))
	429	+	if err != nil {
	430	+	t.Fatalf("dup: %v", err)
	431	+	}
	432	+	sendMessageWithFDs(t, peer, message(7, 3, []byte{0, 0, 0, 0}), []int{sent})
	433	+	if err := unix.Close(sent); err != nil {
	434	+	t.Fatalf("close sent descriptor: %v", err)
	435	+	}
	436	+
	437	+	if err := d.Dispatch(); err != nil {
	438	+	t.Fatalf("Dispatch: %v", err)
	439	+	}
	440	+	if err := d.Close(); err != nil {
	441	+	t.Fatalf("Close: %v", err)
	442	+	}
	443	+	if err := peer.Close(); err != nil {
	444	+	t.Fatalf("close peer: %v", err)
	445	+	}
	446	+	}
	447	+
	448	+	func TestDisplayFD_ParseFailureClosesQueuedDescriptor(t *testing.T) {
	449	+	client, peer := unixSocketPair(t)
	450	+	d := newDisplay(client)
	451	+	d.objects.Store(uint32(7), &fdIgnoringProxy{
	452	+	BaseProxy: BaseProxy{id: 7, context: d.context},
	453	+	})
	454	+
	455	+	r, w, err := os.Pipe()
	456	+	if err != nil {
	457	+	t.Fatalf("pipe: %v", err)
	458	+	}
	459	+	defer r.Close()
	460	+
	461	+	// A descriptor arrives together with a malformed frame.
	462	+	sendMessageWithFDs(t, peer, header(7, 3, 0), []int{int(w.Fd())})
	463	+	w.Close()
	464	+
	465	+	err = d.Dispatch()
	466	+	if !errors.Is(err, ErrMalformedFrame) {
	467	+	t.Fatalf("Dispatch error = %v, want %v", err, ErrMalformedFrame)
	468	+	}
	469	+
	470	+	d.fdMu.Lock()
	471	+	pending := len(d.pendingFDs)
	472	+	d.fdMu.Unlock()
	473	+	if pending != 0 {
	474	+	t.Fatalf("%d descriptors still queued after a malformed frame", pending)
	475	+	}
	476	+
	477	+	if err := r.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil {
	478	+	t.Fatalf("SetReadDeadline: %v", err)
	479	+	}
	480	+	if _, err := r.Read(make([]byte, 1)); !errors.Is(err, io.EOF) {
	481	+	t.Fatalf("read after malformed frame = %v, want %v", err, io.EOF)
	482	+	}
	483	+	}

@@ -0,0 +1,483 @@
	1	+	//go:build linux
	2	+	// +build linux
	3	+
	4	+	package wlturbo
	5	+
	6	+	import (
	7	+	"errors"
	8	+	"io"
	9	+	"net"
	10	+	"os"
	11	+	"testing"
	12	+	"time"
	13	+
	14	+	"golang.org/x/sys/unix"
	15	+	)
	16	+
	17	+	// fdRecordingProxy consumes the file descriptor of every event it receives.
	18	+	// Events without a descriptor are left untouched.
	19	+	type fdRecordingProxy struct {
	20	+	BaseProxy
	21	+	fds *[]int
	22	+	}
	23	+
	24	+	func (p fdRecordingProxy) Dispatch(event Event) {
	25	+	if fd := event.Fd(); fd != 0 {
	26	+	p.fds = append(p.fds, int(fd))
	27	+	return
	28	+	}
	29	+	// No descriptor arrived. Fd already consumed the placeholder argument.
	30	+	}
	31	+
	32	+	// fdIgnoringProxy never reads the descriptor of an event.
	33	+	type fdIgnoringProxy struct {
	34	+	BaseProxy
	35	+	}
	36	+
	37	+	func (p fdIgnoringProxy) Dispatch(Event) {}
	38	+
	39	+	// unixSocketPair returns a connected socket pair as net.UnixConns.
	40	+	func unixSocketPair(t testing.T) (client, peer net.UnixConn) {
	41	+	t.Helper()
	42	+
	43	+	fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM\|unix.SOCK_CLOEXEC, 0)
	44	+	if err != nil {
	45	+	t.Fatalf("socketpair: %v", err)
	46	+	}
	47	+
	48	+	clientFile := os.NewFile(uintptr(fds[0]), "client")
	49	+	peerFile := os.NewFile(uintptr(fds[1]), "peer")
	50	+
	51	+	clientConn, err := net.FileConn(clientFile)
	52	+	clientFile.Close()
	53	+	if err != nil {
	54	+	peerFile.Close()
	55	+	t.Fatalf("net.FileConn(client): %v", err)
	56	+	}
	57	+
	58	+	peerConn, err := net.FileConn(peerFile)
	59	+	peerFile.Close()
	60	+	if err != nil {
	61	+	clientConn.Close()
	62	+	t.Fatalf("net.FileConn(peer): %v", err)
	63	+	}
	64	+
	65	+	t.Cleanup(func() {
	66	+	clientConn.Close()
	67	+	peerConn.Close()
	68	+	})
	69	+
	70	+	return clientConn.(net.UnixConn), peerConn.(net.UnixConn)
	71	+	}
	72	+
	73	+	// sendMessageWithFDs writes one framed message and attaches fds to it.
	74	+	func sendMessageWithFDs(t testing.T, peer net.UnixConn, msg []byte, fds []int) {
	75	+	t.Helper()
	76	+
	77	+	var oob []byte
	78	+	if len(fds) > 0 {
	79	+	oob = unix.UnixRights(fds...)
	80	+	}
	81	+	if _, _, err := peer.WriteMsgUnix(msg, oob, nil); err != nil {
	82	+	t.Fatalf("WriteMsgUnix: %v", err)
	83	+	}
	84	+	}
	85	+
	86	+	// newFDDisplay builds a display over the client end with a proxy registered at
	87	+	// object ID 7 and returns the descriptors that proxy consumes.
	88	+	func newFDDisplay(t testing.T, client net.UnixConn) (Display, []int) {
	89	+	t.Helper()
	90	+
	91	+	d := newDisplay(client)
	92	+	fds := []int{}
	93	+	d.objects.Store(uint32(7), &fdRecordingProxy{
	94	+	BaseProxy: BaseProxy{id: 7, context: d.context},
	95	+	fds: &fds,
	96	+	})
	97	+	return d, &fds
	98	+	}
	99	+
	100	+	// writeSentinel writes a unique byte string through fd and closes the caller's
	101	+	// copy of the descriptor. Reading the same sentinel from the paired read end
	102	+	// proves the descriptor really is the expected one.
	103	+	func writeSentinel(t *testing.T, fd uintptr, sentinel string) {
	104	+	t.Helper()
	105	+	if _, err := unix.Write(int(fd), []byte(sentinel)); err != nil {
	106	+	t.Fatalf("write sentinel through received fd: %v", err)
	107	+	}
	108	+	if err := unix.Close(int(fd)); err != nil {
	109	+	t.Fatalf("close received fd: %v", err)
	110	+	}
	111	+	}
	112	+
	113	+	// readSentinel reads exactly len(sentinel) bytes and compares them.
	114	+	func readSentinel(t testing.T, r os.File, sentinel string) {
	115	+	t.Helper()
	116	+	buf := make([]byte, len(sentinel))
	117	+	if _, err := io.ReadFull(r, buf); err != nil {
	118	+	t.Fatalf("read sentinel: %v", err)
	119	+	}
	120	+	if string(buf) != sentinel {
	121	+	t.Fatalf("sentinel = %q, want %q", buf, sentinel)
	122	+	}
	123	+	}
	124	+
	125	+	func TestDisplayFD_Single(t *testing.T) {
	126	+	client, peer := unixSocketPair(t)
	127	+	d, received := newFDDisplay(t, client)
	128	+
	129	+	r, w, err := os.Pipe()
	130	+	if err != nil {
	131	+	t.Fatalf("pipe: %v", err)
	132	+	}
	133	+	defer r.Close()
	134	+
	135	+	sendMessageWithFDs(t, peer, message(7, 3, []byte{0, 0, 0, 0}), []int{int(w.Fd())})
	136	+	w.Close()
	137	+
	138	+	if err := d.Dispatch(); err != nil {
	139	+	t.Fatalf("Dispatch: %v", err)
	140	+	}
	141	+	if len(*received) != 1 {
	142	+	t.Fatalf("received %d descriptors, want 1", len(*received))
	143	+	}
	144	+
	145	+	writeSentinel(t, uintptr((*received)[0]), "single-fd-sentinel")
	146	+	readSentinel(t, r, "single-fd-sentinel")
	147	+	}
	148	+
	149	+	func TestDisplayFD_MultipleInOneControlMessage(t *testing.T) {
	150	+	client, peer := unixSocketPair(t)
	151	+	d, received := newFDDisplay(t, client)
	152	+
	153	+	firstR, firstW, err := os.Pipe()
	154	+	if err != nil {
	155	+	t.Fatalf("pipe: %v", err)
	156	+	}
	157	+	defer firstR.Close()
	158	+	secondR, secondW, err := os.Pipe()
	159	+	if err != nil {
	160	+	t.Fatalf("pipe: %v", err)
	161	+	}
	162	+	defer secondR.Close()
	163	+
	164	+	sendMessageWithFDs(t, peer, message(7, 3, []byte{0, 0, 0, 0}),
	165	+	[]int{int(firstW.Fd()), int(secondW.Fd())})
	166	+	firstW.Close()
	167	+	secondW.Close()
	168	+
	169	+	if err := d.Dispatch(); err != nil {
	170	+	t.Fatalf("Dispatch: %v", err)
	171	+	}
	172	+
	173	+	// A handler consumes one descriptor per fd argument, so the test drains
	174	+	// whatever the connection still holds, in arrival order.
	175	+	for {
	176	+	fd, ok := d.nextFD()
	177	+	if !ok {
	178	+	break
	179	+	}
	180	+	received = append(received, fd)
	181	+	}
	182	+
	183	+	if len(*received) != 2 {
	184	+	t.Fatalf("received %d descriptors, want 2", len(*received))
	185	+	}
	186	+
	187	+	writeSentinel(t, uintptr((*received)[0]), "first-pipe")
	188	+	writeSentinel(t, uintptr((*received)[1]), "second-pipe")
	189	+
	190	+	readSentinel(t, firstR, "first-pipe")
	191	+	readSentinel(t, secondR, "second-pipe")
	192	+	}
	193	+
	194	+	func TestDisplayFD_EventWithoutFDsAfterOneWithFDs(t *testing.T) {
	195	+	client, peer := unixSocketPair(t)
	196	+	d, received := newFDDisplay(t, client)
	197	+
	198	+	r, w, err := os.Pipe()
	199	+	if err != nil {
	200	+	t.Fatalf("pipe: %v", err)
	201	+	}
	202	+	defer r.Close()
	203	+
	204	+	// First event carries a descriptor.
	205	+	sendMessageWithFDs(t, peer, message(7, 3, []byte{0, 0, 0, 0}), []int{int(w.Fd())})
	206	+	w.Close()
	207	+	if err := d.Dispatch(); err != nil {
	208	+	t.Fatalf("first Dispatch: %v", err)
	209	+	}
	210	+
	211	+	// Second event carries none and must not steal the first one.
	212	+	sendMessageWithFDs(t, peer, message(7, 4, []byte{0, 0, 0, 0}), nil)
	213	+	if err := d.Dispatch(); err != nil {
	214	+	t.Fatalf("second Dispatch: %v", err)
	215	+	}
	216	+
	217	+	if len(*received) != 1 {
	218	+	t.Fatalf("received %d descriptors, want exactly 1", len(*received))
	219	+	}
	220	+
	221	+	writeSentinel(t, uintptr((*received)[0]), "only-fd")
	222	+	readSentinel(t, r, "only-fd")
	223	+	}
	224	+
	225	+	func TestDisplayFD_TwoDisplaysAreIsolated(t *testing.T) {
	226	+	firstClient, firstPeer := unixSocketPair(t)
	227	+	secondClient, secondPeer := unixSocketPair(t)
	228	+
	229	+	firstDisplay, firstReceived := newFDDisplay(t, firstClient)
	230	+	secondDisplay, secondReceived := newFDDisplay(t, secondClient)
	231	+
	232	+	firstR, firstW, err := os.Pipe()
	233	+	if err != nil {
	234	+	t.Fatalf("pipe: %v", err)
	235	+	}
	236	+	defer firstR.Close()
	237	+	secondR, secondW, err := os.Pipe()
	238	+	if err != nil {
	239	+	t.Fatalf("pipe: %v", err)
	240	+	}
	241	+	defer secondR.Close()
	242	+
	243	+	sendMessageWithFDs(t, firstPeer, message(7, 3, []byte{0, 0, 0, 0}), []int{int(firstW.Fd())})
	244	+	firstW.Close()
	245	+	sendMessageWithFDs(t, secondPeer, message(7, 3, []byte{0, 0, 0, 0}), []int{int(secondW.Fd())})
	246	+	secondW.Close()
	247	+
	248	+	if err := secondDisplay.Dispatch(); err != nil {
	249	+	t.Fatalf("second Dispatch: %v", err)
	250	+	}
	251	+	if err := firstDisplay.Dispatch(); err != nil {
	252	+	t.Fatalf("first Dispatch: %v", err)
	253	+	}
	254	+
	255	+	// Each connection must only see its own descriptor.
	256	+	if len(*firstReceived) != 1 {
	257	+	t.Fatalf("first display received %d descriptors, want 1", len(*firstReceived))
	258	+	}
	259	+	if len(*secondReceived) != 1 {
	260	+	t.Fatalf("second display received %d descriptors, want 1", len(*secondReceived))
	261	+	}
	262	+
	263	+	writeSentinel(t, uintptr((*secondReceived)[0]), "second-display")
	264	+	writeSentinel(t, uintptr((*firstReceived)[0]), "first-display")
	265	+
	266	+	readSentinel(t, secondR, "second-display")
	267	+	readSentinel(t, firstR, "first-display")
	268	+	}
	269	+
	270	+	func TestDisplayFD_TruncatedControlData(t *testing.T) {
	271	+	client, peer := unixSocketPair(t)
	272	+	d, _ := newFDDisplay(t, client)
	273	+
	274	+	// One more descriptor than a single read can carry.
	275	+	devNull, err := os.Open(os.DevNull)
	276	+	if err != nil {
	277	+	t.Fatalf("open %s: %v", os.DevNull, err)
	278	+	}
	279	+	defer devNull.Close()
	280	+
	281	+	fds := make([]int, 0, maxRecvFDs+1)
	282	+	for i := 0; i < maxRecvFDs+1; i++ {
	283	+	dup, err := unix.Dup(int(devNull.Fd()))
	284	+	if err != nil {
	285	+	t.Fatalf("dup: %v", err)
	286	+	}
	287	+	t.Cleanup(func() { _ = unix.Close(dup) })
	288	+	fds = append(fds, dup)
	289	+	}
	290	+
	291	+	sendMessageWithFDs(t, peer, message(7, 3, []byte{0, 0, 0, 0}), fds)
	292	+
	293	+	err = d.Dispatch()
	294	+	if !errors.Is(err, ErrMalformedFrame) {
	295	+	t.Fatalf("Dispatch error = %v, want %v", err, ErrMalformedFrame)
	296	+	}
	297	+
	298	+	var perr *ProtocolError
	299	+	if !errors.As(err, &perr) {
	300	+	t.Fatalf("Dispatch error type = %T, want *ProtocolError", err)
	301	+	}
	302	+	if perr.Kind != "control_truncated" {
	303	+	t.Fatalf("error kind = %q, want %q", perr.Kind, "control_truncated")
	304	+	}
	305	+
	306	+	// A truncated control message must not leave descriptors queued.
	307	+	d.fdMu.Lock()
	308	+	pending := len(d.pendingFDs)
	309	+	d.fdMu.Unlock()
	310	+	if pending != 0 {
	311	+	t.Fatalf("%d descriptors still queued after a truncated control message", pending)
	312	+	}
	313	+	}
	314	+
	315	+	func TestDisplayFD_UnconsumedDescriptorClosedOnShutdown(t *testing.T) {
	316	+	client, peer := unixSocketPair(t)
	317	+	d := newDisplay(client)
	318	+	d.objects.Store(uint32(7), &fdIgnoringProxy{
	319	+	BaseProxy: BaseProxy{id: 7, context: d.context},
	320	+	})
	321	+
	322	+	r, w, err := os.Pipe()
	323	+	if err != nil {
	324	+	t.Fatalf("pipe: %v", err)
	325	+	}
	326	+	defer r.Close()
	327	+
	328	+	sendMessageWithFDs(t, peer, message(7, 3, []byte{0, 0, 0, 0}), []int{int(w.Fd())})
	329	+	// Drop our own copy: the display now holds the only writer.
	330	+	w.Close()
	331	+
	332	+	if err := d.Dispatch(); err != nil {
	333	+	t.Fatalf("Dispatch: %v", err)
	334	+	}
	335	+
	336	+	// While the display still holds the descriptor the pipe stays open.
	337	+	if err := r.SetReadDeadline(time.Now().Add(200 * time.Millisecond)); err != nil {
	338	+	t.Fatalf("SetReadDeadline: %v", err)
	339	+	}
	340	+	if _, err := r.Read(make([]byte, 1)); !errors.Is(err, os.ErrDeadlineExceeded) {
	341	+	t.Fatalf("read with the display holding the descriptor = %v, want %v", err, os.ErrDeadlineExceeded)
	342	+	}
	343	+
	344	+	// Closing the display must release it, so the read end sees EOF.
	345	+	if err := d.Close(); err != nil {
	346	+	t.Fatalf("Close: %v", err)
	347	+	}
	348	+	if err := r.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil {
	349	+	t.Fatalf("SetReadDeadline: %v", err)
	350	+	}
	351	+	if _, err := r.Read(make([]byte, 1)); !errors.Is(err, io.EOF) {
	352	+	t.Fatalf("read after Close = %v, want %v", err, io.EOF)
	353	+	}
	354	+	}
	355	+
	356	+	// rawSocketPair returns a connected socket pair without registering cleanup,
	357	+	// so loop tests can close descriptors deterministically.
	358	+	func rawSocketPair(t testing.T) (client, peer net.UnixConn) {
	359	+	t.Helper()
	360	+
	361	+	fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM\|unix.SOCK_CLOEXEC, 0)
	362	+	if err != nil {
	363	+	t.Fatalf("socketpair: %v", err)
	364	+	}
	365	+
	366	+	clientFile := os.NewFile(uintptr(fds[0]), "client")
	367	+	clientConn, err := net.FileConn(clientFile)
	368	+	clientFile.Close()
	369	+	if err != nil {
	370	+	t.Fatalf("net.FileConn(client): %v", err)
	371	+	}
	372	+
	373	+	peerFile := os.NewFile(uintptr(fds[1]), "peer")
	374	+	peerConn, err := net.FileConn(peerFile)
	375	+	peerFile.Close()
	376	+	if err != nil {
	377	+	t.Fatalf("net.FileConn(peer): %v", err)
	378	+	}
	379	+
	380	+	return clientConn.(net.UnixConn), peerConn.(net.UnixConn)
	381	+	}
	382	+
	383	+	// openFDs reports how many descriptors the process holds.
	384	+	func openFDs(t *testing.T) int {
	385	+	t.Helper()
	386	+
	387	+	entries, err := os.ReadDir("/proc/self/fd")
	388	+	if err != nil {
	389	+	t.Skipf("cannot count open descriptors: %v", err)
	390	+	}
	391	+	return len(entries)
	392	+	}
	393	+
	394	+	func TestDisplayFD_NoLeakAcrossConnections(t *testing.T) {
	395	+	const iterations = 100
	396	+
	397	+	devNull, err := os.Open(os.DevNull)
	398	+	if err != nil {
	399	+	t.Fatalf("open %s: %v", os.DevNull, err)
	400	+	}
	401	+	defer devNull.Close()
	402	+
	403	+	// Warm up so caches and pools are populated before measuring.
	404	+	runFDLeakIteration(t, devNull)
	405	+
	406	+	before := openFDs(t)
	407	+	for i := 0; i < iterations; i++ {
	408	+	runFDLeakIteration(t, devNull)
	409	+	}
	410	+	after := openFDs(t)
	411	+
	412	+	if after > before {
	413	+	t.Fatalf("descriptor count grew from %d to %d over %d connections", before, after, iterations)
	414	+	}
	415	+	}
	416	+
	417	+	// runFDLeakIteration opens a connection, delivers a descriptor nobody reads and
	418	+	// tears everything down.
	419	+	func runFDLeakIteration(t testing.T, devNull os.File) {
	420	+	t.Helper()
	421	+
	422	+	client, peer := rawSocketPair(t)
	423	+	d := newDisplay(client)
	424	+	d.objects.Store(uint32(7), &fdIgnoringProxy{
	425	+	BaseProxy: BaseProxy{id: 7, context: d.context},
	426	+	})
	427	+
	428	+	sent, err := unix.Dup(int(devNull.Fd()))
	429	+	if err != nil {
	430	+	t.Fatalf("dup: %v", err)
	431	+	}
	432	+	sendMessageWithFDs(t, peer, message(7, 3, []byte{0, 0, 0, 0}), []int{sent})
	433	+	if err := unix.Close(sent); err != nil {
	434	+	t.Fatalf("close sent descriptor: %v", err)
	435	+	}
	436	+
	437	+	if err := d.Dispatch(); err != nil {
	438	+	t.Fatalf("Dispatch: %v", err)
	439	+	}
	440	+	if err := d.Close(); err != nil {
	441	+	t.Fatalf("Close: %v", err)
	442	+	}
	443	+	if err := peer.Close(); err != nil {
	444	+	t.Fatalf("close peer: %v", err)
	445	+	}
	446	+	}
	447	+
	448	+	func TestDisplayFD_ParseFailureClosesQueuedDescriptor(t *testing.T) {
	449	+	client, peer := unixSocketPair(t)
	450	+	d := newDisplay(client)
	451	+	d.objects.Store(uint32(7), &fdIgnoringProxy{
	452	+	BaseProxy: BaseProxy{id: 7, context: d.context},
	453	+	})
	454	+
	455	+	r, w, err := os.Pipe()
	456	+	if err != nil {
	457	+	t.Fatalf("pipe: %v", err)
	458	+	}
	459	+	defer r.Close()
	460	+
	461	+	// A descriptor arrives together with a malformed frame.
	462	+	sendMessageWithFDs(t, peer, header(7, 3, 0), []int{int(w.Fd())})
	463	+	w.Close()
	464	+
	465	+	err = d.Dispatch()
	466	+	if !errors.Is(err, ErrMalformedFrame) {
	467	+	t.Fatalf("Dispatch error = %v, want %v", err, ErrMalformedFrame)
	468	+	}
	469	+
	470	+	d.fdMu.Lock()
	471	+	pending := len(d.pendingFDs)
	472	+	d.fdMu.Unlock()
	473	+	if pending != 0 {
	474	+	t.Fatalf("%d descriptors still queued after a malformed frame", pending)
	475	+	}
	476	+
	477	+	if err := r.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil {
	478	+	t.Fatalf("SetReadDeadline: %v", err)
	479	+	}
	480	+	if _, err := r.Read(make([]byte, 1)); !errors.Is(err, io.EOF) {
	481	+	t.Fatalf("read after malformed frame = %v, want %v", err, io.EOF)
	482	+	}
	483	+	}

M scm_linux.go +83 -101


@@ -4,92 +4,33 @@
4	4		package wlturbo
5	5
6	6		import (
	7	+	"errors"
7	8		"fmt"
8	9		"golang.org/x/sys/unix"
9		-	"net"
10	10		"sync"
11		-	"sync/atomic"
12	11		"syscall"
13	12		)
14	13
15		-	// fdQueue is a lock-free multi-producer single-consumer queue for file descriptors
16		-	type fdQueue struct {
17		-	items [256]atomic.Pointer[fdItem]
18		-	head atomic.Uint64
19		-	tail atomic.Uint64
20		-	}
	14	+	// maxRecvFDs bounds how many ancillary file descriptors a single read accepts.
	15	+	// A socket read that carries more than this is reported as a protocol failure
	16	+	// instead of silently dropping descriptors.
	17	+	const maxRecvFDs = 64
21	18
22		-	type fdItem struct {
23		-	fd int
24		-	next atomic.Pointer[fdItem]
	19	+	// controlBufferPool provides control-message buffers sized for maxRecvFDs.
	20	+	var controlBufferPool = sync.Pool{
	21	+	New: func() interface{} {
	22	+	return make([]byte, unix.CmsgSpace(maxRecvFDs*4))
	23	+	},
25	24		}
26	25
27		-	var (
28		-	// Global FD queue for zero-allocation FD passing
29		-	globalFDQueue = &fdQueue{}
30		-
31		-	// Pre-allocated FD items pool
32		-	fdItemPool = sync.Pool{
33		-	New: func() interface{} {
34		-	return &fdItem{}
35		-	},
36		-	}
37		-
38		-	// Pre-allocated buffers for control messages
39		-	controlBufferPool = sync.Pool{
40		-	New: func() interface{} {
41		-	return make([]byte, unix.CmsgSpace(4*4)) // Space for 4 FDs
42		-	},
43		-	}
44		-	)
45		-
46		-	// enqueueFD adds a file descriptor to the lock-free queue
47		-	func (q *fdQueue) enqueueFD(fd int) {
48		-	item := fdItemPool.Get().(*fdItem)
49		-	item.fd = fd
50		-	item.next.Store(nil)
51		-
52		-	for {
53		-	tail := q.tail.Load()
54		-	next := tail & 255
55		-
56		-	if q.items[next].CompareAndSwap(nil, item) {
57		-	q.tail.Add(1)
58		-	return
59		-	}
60		-	// Retry if CAS failed
61		-	}
62		-	}
63		-
64		-	// dequeueFD removes and returns a file descriptor from the queue
65		-	func (q *fdQueue) dequeueFD() (int, bool) {
66		-	for {
67		-	head := q.head.Load()
68		-	tail := q.tail.Load()
69		-
70		-	if head >= tail {
71		-	return -1, false
72		-	}
73		-
74		-	next := head & 255
75		-	item := q.items[next].Load()
76		-
77		-	if item == nil {
78		-	continue
79		-	}
80		-
81		-	if q.head.CompareAndSwap(head, head+1) {
82		-	fd := item.fd
83		-	q.items[next].Store(nil)
84		-	fdItemPool.Put(item)
85		-	return fd, true
86		-	}
87		-	}
88		-	}
89		-
90		-	// readChunk reads one chunk of bytes from the display connection, collecting
91		-	// any ancillary file descriptors that arrive with it. A single read never
92		-	// defines a message boundary; callers frame the returned bytes themselves.
	26	+	// readChunk reads one chunk of bytes from the display connection and queues any
	27	+	// ancillary file descriptors that arrive with it. A single read never defines a
	28	+	// message boundary; callers frame the returned bytes themselves.
	29	+	//
	30	+	// Descriptors are owned by the connection and consumed in arrival order by
	31	+	// Event.Fd, which is what the Wayland wire protocol requires: an fd belongs to
	32	+	// the argument position in the byte stream, not to whichever message happens to
	33	+	// end first.
93	34		func (d *Display) readChunk() (int, error) {
94	35		if d.unix == nil {
95	36		n, err := d.conn.Read(d.recvBuf)

@@ -102,63 +43,104 @@ func (d *Display) readChunk() (int, error) {
102	43		oob := controlBufferPool.Get().([]byte)
103	44		defer controlBufferPool.Put(oob)
104	45
105		-	n, oobn, _, _, err := d.unix.ReadMsgUnix(d.recvBuf, oob)
	46	+	n, oobn, flags, _, err := d.unix.ReadMsgUnix(d.recvBuf, oob)
106	47		if n < 0 {
107	48		n = 0
108	49		}
	50	+
109	51		if oobn > 0 {
110		-	if perr := globalFDQueue.parseControl(oob[:oobn]); perr != nil {
111		-	return n, perr
	52	+	fds, perr := parseFDs(oob[:oobn])
	53	+	d.queueFDs(fds)
	54	+	if perr != nil && err == nil {
	55	+	err = perr
112	56		}
113	57		}
	58	+
	59	+	if flags&unix.MSG_CTRUNC != 0 && err == nil {
	60	+	// Descriptors were dropped by the kernel, so the connection can no
	61	+	// longer be demarshalled correctly: treat it as a protocol failure so
	62	+	// the caller closes the descriptors that did arrive.
	63	+	err = &ProtocolError{Kind: "control_truncated", Err: ErrMalformedFrame}
	64	+	}
114	65		return n, err
115	66		}
116	67
117		-	// parseControl queues the file descriptors carried by socket control messages.
118		-	func (q *fdQueue) parseControl(oob []byte) error {
	68	+	// parseFDs extracts the file descriptors carried by socket control messages.
	69	+	// Descriptors parsed before an error are returned so the caller can close them.
	70	+	func parseFDs(oob []byte) ([]int, error) {
119	71		scms, err := syscall.ParseSocketControlMessage(oob)
120	72		if err != nil {
121		-	return fmt.Errorf("parse control message: %w", err)
	73	+	return nil, fmt.Errorf("parse control message: %w", err)
122	74		}
123		-	for _, scm := range scms {
	75	+
	76	+	var fds []int
	77	+	for i := range scms {
	78	+	scm := &scms[i]
124	79		if scm.Header.Type != syscall.SCM_RIGHTS {
125	80		continue
126	81		}
127		-	parsedFDs, err := syscall.ParseUnixRights(&scm)
	82	+	parsed, err := syscall.ParseUnixRights(scm)
128	83		if err != nil {
129		-	return fmt.Errorf("parse unix rights: %w", err)
	84	+	return fds, fmt.Errorf("parse unix rights: %w", err)
130	85		}
131		-	for _, fd := range parsedFDs {
132		-	q.enqueueFD(fd)
	86	+	fds = append(fds, parsed...)
	87	+	}
	88	+	return fds, nil
	89	+	}
	90	+
	91	+	// queueFDs appends descriptors to this connection's pending list.
	92	+	func (d *Display) queueFDs(fds []int) {
	93	+	if len(fds) == 0 {
	94	+	return
	95	+	}
	96	+	d.fdMu.Lock()
	97	+	d.pendingFDs = append(d.pendingFDs, fds...)
	98	+	d.fdMu.Unlock()
	99	+	}
	100	+
	101	+	// nextFD removes and returns the oldest pending descriptor.
	102	+	func (d *Display) nextFD() (int, bool) {
	103	+	d.fdMu.Lock()
	104	+	defer d.fdMu.Unlock()
	105	+	if len(d.pendingFDs) == 0 {
	106	+	return -1, false
	107	+	}
	108	+	fd := d.pendingFDs[0]
	109	+	d.pendingFDs = d.pendingFDs[1:]
	110	+	return fd, true
	111	+	}
	112	+
	113	+	// closePendingFDs closes every descriptor this connection still holds.
	114	+	func (d *Display) closePendingFDs() {
	115	+	d.fdMu.Lock()
	116	+	fds := d.pendingFDs
	117	+	d.pendingFDs = nil
	118	+	d.fdMu.Unlock()
	119	+
	120	+	for _, fd := range fds {
	121	+	if fd >= 0 {
	122	+	_ = unix.Close(fd)
133	123		}
134	124		}
135		-	return nil
136	125		}
137	126
138		-	// sendmsgWithFDs sends a message potentially containing file descriptors
	127	+	// sendmsgWithFDs sends a message, attaching file descriptors when present.
139	128		func (d *Display) sendmsgWithFDs(buf []byte, fds []int) error {
140	129		d.sendMu.Lock()
141	130		defer d.sendMu.Unlock()
142	131
143	132		if len(fds) == 0 {
144		-	// Fast path: no FDs to send
145	133		_, err := d.conn.Write(buf)
146	134		return err
147	135		}
	136	+	if d.unix == nil {
	137	+	return errors.New("wlturbo: cannot send file descriptors over a non-Unix connection")
	138	+	}
148	139
149		-	// Create control message with file descriptors
150		-	oob := syscall.UnixRights(fds...)
151		-
152		-	// Send message with control data
153		-	_, _, err := d.conn.(*net.UnixConn).WriteMsgUnix(buf, oob, nil)
	140	+	_, _, err := d.unix.WriteMsgUnix(buf, unix.UnixRights(fds...), nil)
154	141		return err
155	142		}
156	143
157		-	// GetNextFD retrieves the next file descriptor from the queue
158		-	func GetNextFD() (int, bool) {
159		-	return globalFDQueue.dequeueFD()
160		-	}
161		-
162	144		// Memory mapping helpers for shared memory buffers
163	145
164	146		// CreateAnonymousFile creates an anonymous file for shared memory


@@ -4,92 +4,33 @@
4		package wlturbo	4		package wlturbo
5			5
6		import (	6		import (
			7	+	"errors"
7		"fmt"	8		"fmt"
8		"golang.org/x/sys/unix"	9		"golang.org/x/sys/unix"
9	-	"net"
10		"sync"	10		"sync"
11	-	"sync/atomic"
12		"syscall"	11		"syscall"
13		)	12		)
14			13
15	-	// fdQueue is a lock-free multi-producer single-consumer queue for file descriptors	14	+	// maxRecvFDs bounds how many ancillary file descriptors a single read accepts.
16	-	type fdQueue struct {	15	+	// A socket read that carries more than this is reported as a protocol failure
17	-	items [256]atomic.Pointer[fdItem]	16	+	// instead of silently dropping descriptors.
18	-	head atomic.Uint64	17	+	const maxRecvFDs = 64
19	-	tail atomic.Uint64
20	-	}
21			18
22	-	type fdItem struct {	19	+	// controlBufferPool provides control-message buffers sized for maxRecvFDs.
23	-	fd int	20	+	var controlBufferPool = sync.Pool{
24	-	next atomic.Pointer[fdItem]	21	+	New: func() interface{} {
			22	+	return make([]byte, unix.CmsgSpace(maxRecvFDs*4))
			23	+	},
25		}	24		}
26			25
27	-	var (	26	+	// readChunk reads one chunk of bytes from the display connection and queues any
28	-	// Global FD queue for zero-allocation FD passing	27	+	// ancillary file descriptors that arrive with it. A single read never defines a
29	-	globalFDQueue = &fdQueue{}	28	+	// message boundary; callers frame the returned bytes themselves.
30	-		29	+	//
31	-	// Pre-allocated FD items pool	30	+	// Descriptors are owned by the connection and consumed in arrival order by
32	-	fdItemPool = sync.Pool{	31	+	// Event.Fd, which is what the Wayland wire protocol requires: an fd belongs to
33	-	New: func() interface{} {	32	+	// the argument position in the byte stream, not to whichever message happens to
34	-	return &fdItem{}	33	+	// end first.
35	-	},
36	-	}
37	-
38	-	// Pre-allocated buffers for control messages
39	-	controlBufferPool = sync.Pool{
40	-	New: func() interface{} {
41	-	return make([]byte, unix.CmsgSpace(4*4)) // Space for 4 FDs
42	-	},
43	-	}
44	-	)
45	-
46	-	// enqueueFD adds a file descriptor to the lock-free queue
47	-	func (q *fdQueue) enqueueFD(fd int) {
48	-	item := fdItemPool.Get().(*fdItem)
49	-	item.fd = fd
50	-	item.next.Store(nil)
51	-
52	-	for {
53	-	tail := q.tail.Load()
54	-	next := tail & 255
55	-
56	-	if q.items[next].CompareAndSwap(nil, item) {
57	-	q.tail.Add(1)
58	-	return
59	-	}
60	-	// Retry if CAS failed
61	-	}
62	-	}
63	-
64	-	// dequeueFD removes and returns a file descriptor from the queue
65	-	func (q *fdQueue) dequeueFD() (int, bool) {
66	-	for {
67	-	head := q.head.Load()
68	-	tail := q.tail.Load()
69	-
70	-	if head >= tail {
71	-	return -1, false
72	-	}
73	-
74	-	next := head & 255
75	-	item := q.items[next].Load()
76	-
77	-	if item == nil {
78	-	continue
79	-	}
80	-
81	-	if q.head.CompareAndSwap(head, head+1) {
82	-	fd := item.fd
83	-	q.items[next].Store(nil)
84	-	fdItemPool.Put(item)
85	-	return fd, true
86	-	}
87	-	}
88	-	}
89	-
90	-	// readChunk reads one chunk of bytes from the display connection, collecting
91	-	// any ancillary file descriptors that arrive with it. A single read never
92	-	// defines a message boundary; callers frame the returned bytes themselves.
93		func (d *Display) readChunk() (int, error) {	34		func (d *Display) readChunk() (int, error) {
94		if d.unix == nil {	35		if d.unix == nil {
95		n, err := d.conn.Read(d.recvBuf)	36		n, err := d.conn.Read(d.recvBuf)

@@ -102,63 +43,104 @@ func (d *Display) readChunk() (int, error) {
102		oob := controlBufferPool.Get().([]byte)	43		oob := controlBufferPool.Get().([]byte)
103		defer controlBufferPool.Put(oob)	44		defer controlBufferPool.Put(oob)
104			45
105	-	n, oobn, _, _, err := d.unix.ReadMsgUnix(d.recvBuf, oob)	46	+	n, oobn, flags, _, err := d.unix.ReadMsgUnix(d.recvBuf, oob)
106		if n < 0 {	47		if n < 0 {
107		n = 0	48		n = 0
108		}	49		}
			50	+
109		if oobn > 0 {	51		if oobn > 0 {
110	-	if perr := globalFDQueue.parseControl(oob[:oobn]); perr != nil {	52	+	fds, perr := parseFDs(oob[:oobn])
111	-	return n, perr	53	+	d.queueFDs(fds)
			54	+	if perr != nil && err == nil {
			55	+	err = perr
112		}	56		}
113		}	57		}
			58	+
			59	+	if flags&unix.MSG_CTRUNC != 0 && err == nil {
			60	+	// Descriptors were dropped by the kernel, so the connection can no
			61	+	// longer be demarshalled correctly: treat it as a protocol failure so
			62	+	// the caller closes the descriptors that did arrive.
			63	+	err = &ProtocolError{Kind: "control_truncated", Err: ErrMalformedFrame}
			64	+	}
114		return n, err	65		return n, err
115		}	66		}
116			67
117	-	// parseControl queues the file descriptors carried by socket control messages.	68	+	// parseFDs extracts the file descriptors carried by socket control messages.
118	-	func (q *fdQueue) parseControl(oob []byte) error {	69	+	// Descriptors parsed before an error are returned so the caller can close them.
			70	+	func parseFDs(oob []byte) ([]int, error) {
119		scms, err := syscall.ParseSocketControlMessage(oob)	71		scms, err := syscall.ParseSocketControlMessage(oob)
120		if err != nil {	72		if err != nil {
121	-	return fmt.Errorf("parse control message: %w", err)	73	+	return nil, fmt.Errorf("parse control message: %w", err)
122		}	74		}
123	-	for _, scm := range scms {	75	+
			76	+	var fds []int
			77	+	for i := range scms {
			78	+	scm := &scms[i]
124		if scm.Header.Type != syscall.SCM_RIGHTS {	79		if scm.Header.Type != syscall.SCM_RIGHTS {
125		continue	80		continue
126		}	81		}
127	-	parsedFDs, err := syscall.ParseUnixRights(&scm)	82	+	parsed, err := syscall.ParseUnixRights(scm)
128		if err != nil {	83		if err != nil {
129	-	return fmt.Errorf("parse unix rights: %w", err)	84	+	return fds, fmt.Errorf("parse unix rights: %w", err)
130		}	85		}
131	-	for _, fd := range parsedFDs {	86	+	fds = append(fds, parsed...)
132	-	q.enqueueFD(fd)	87	+	}
			88	+	return fds, nil
			89	+	}
			90	+
			91	+	// queueFDs appends descriptors to this connection's pending list.
			92	+	func (d *Display) queueFDs(fds []int) {
			93	+	if len(fds) == 0 {
			94	+	return
			95	+	}
			96	+	d.fdMu.Lock()
			97	+	d.pendingFDs = append(d.pendingFDs, fds...)
			98	+	d.fdMu.Unlock()
			99	+	}
			100	+
			101	+	// nextFD removes and returns the oldest pending descriptor.
			102	+	func (d *Display) nextFD() (int, bool) {
			103	+	d.fdMu.Lock()
			104	+	defer d.fdMu.Unlock()
			105	+	if len(d.pendingFDs) == 0 {
			106	+	return -1, false
			107	+	}
			108	+	fd := d.pendingFDs[0]
			109	+	d.pendingFDs = d.pendingFDs[1:]
			110	+	return fd, true
			111	+	}
			112	+
			113	+	// closePendingFDs closes every descriptor this connection still holds.
			114	+	func (d *Display) closePendingFDs() {
			115	+	d.fdMu.Lock()
			116	+	fds := d.pendingFDs
			117	+	d.pendingFDs = nil
			118	+	d.fdMu.Unlock()
			119	+
			120	+	for _, fd := range fds {
			121	+	if fd >= 0 {
			122	+	_ = unix.Close(fd)
133		}	123		}
134		}	124		}
135	-	return nil
136		}	125		}
137			126
138	-	// sendmsgWithFDs sends a message potentially containing file descriptors	127	+	// sendmsgWithFDs sends a message, attaching file descriptors when present.
139		func (d *Display) sendmsgWithFDs(buf []byte, fds []int) error {	128		func (d *Display) sendmsgWithFDs(buf []byte, fds []int) error {
140		d.sendMu.Lock()	129		d.sendMu.Lock()
141		defer d.sendMu.Unlock()	130		defer d.sendMu.Unlock()
142			131
143		if len(fds) == 0 {	132		if len(fds) == 0 {
144	-	// Fast path: no FDs to send
145		_, err := d.conn.Write(buf)	133		_, err := d.conn.Write(buf)
146		return err	134		return err
147		}	135		}
			136	+	if d.unix == nil {
			137	+	return errors.New("wlturbo: cannot send file descriptors over a non-Unix connection")
			138	+	}
148			139
149	-	// Create control message with file descriptors	140	+	_, _, err := d.unix.WriteMsgUnix(buf, unix.UnixRights(fds...), nil)
150	-	oob := syscall.UnixRights(fds...)
151	-
152	-	// Send message with control data
153	-	_, _, err := d.conn.(*net.UnixConn).WriteMsgUnix(buf, oob, nil)
154		return err	141		return err
155		}	142		}
156			143
157	-	// GetNextFD retrieves the next file descriptor from the queue
158	-	func GetNextFD() (int, bool) {
159	-	return globalFDQueue.dequeueFD()
160	-	}
161	-
162		// Memory mapping helpers for shared memory buffers	144		// Memory mapping helpers for shared memory buffers
163			145
164		// CreateAnonymousFile creates an anonymous file for shared memory	146		// CreateAnonymousFile creates an anonymous file for shared memory

M wire.go +2 -3


@@ -88,13 +88,12 @@ func (e *DisplayError) Error() string {
88	88		func (e *DisplayError) Unwrap() error { return ErrDisplayError }
89	89
90	90		// receivedFrame is one complete Wayland message read from the connection.
91		-	// fds holds the ancillary descriptors that arrived while the frame's bytes
92		-	// were read, in arrival order.
	91	+	// Ancillary file descriptors are not part of the frame: they are owned by the
	92	+	// connection and consumed in arrival order by Event.Fd.
93	93		type receivedFrame struct {
94	94		object uint32
95	95		opcode uint16
96	96		body []byte
97		-	fds []int
98	97		}
99	98
100	99		// parseHeader decodes and validates an 8-byte Wayland message header.

 func (e *DisplayError) Unwrap() error { return ErrDisplayError }
 // receivedFrame is one complete Wayland message read from the connection.
-// fds holds the ancillary descriptors that arrived while the frame's bytes
+// Ancillary file descriptors are not part of the frame: they are owned by the
-// were read, in arrival order.
+// connection and consumed in arrival order by Event.Fd.
 type receivedFrame struct {
 	object uint32
 	opcode uint16
 	body   []byte
-	fds    []int
 }
 // parseHeader decodes and validates an 8-byte Wayland message header.