fix(fdwait): check readiness before parking and fail on lost watches

Poll checks fds once before registering them, so an expired deadline
never skips the readiness check (Poll(fds, 0) now matches poll(2)) and
ready fds cost no epoll registration. A descriptor listed twice is
watched once. A registration that cannot be removed marks the poller
broken instead of leaving a stale readable watch. Errors name the fd.

syncWaiter drops its dead fixture branch and duplicate pipe fd; tests
close the waiter's epoll fd. The recycled-fd test, which could skip and
did not guard the regression, is replaced by edge-case tests.

commit: bda62c7a29ccf19892571d808740cd3e51af7e84
author: brice · 9d ago · 2026-10-14 02:42 UTC
committer: brice · 9d ago · 2026-10-14 04:35 UTC
parent: cef9892
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +208 -94

A internal/adapters/fdwait/edges_test.go +129 -0

@@ -0,0 +1,129 @@
	1	+	package fdwait
	2	+
	3	+	import (
	4	+	"testing"
	5	+
	6	+	"golang.org/x/sys/unix"
	7	+	)
	8	+
	9	+	func eventfd(t *testing.T) int {
	10	+	t.Helper()
	11	+	fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
	12	+	if err != nil {
	13	+	t.Fatal(err)
	14	+	}
	15	+	t.Cleanup(func() { unix.Close(fd) })
	16	+	return fd
	17	+	}
	18	+
	19	+	func signal(t *testing.T, fd int) {
	20	+	t.Helper()
	21	+	if _, err := unix.Write(fd, []byte{1, 0, 0, 0, 0, 0, 0, 0}); err != nil {
	22	+	t.Fatal(err)
	23	+	}
	24	+	}
	25	+
	26	+	// assertNoWatches fails if Poll left a registration behind, by making every
	27	+	// test descriptor readable and checking epoll reports nothing.
	28	+	func assertNoWatches(t testing.T, p Poller, fds ...int) {
	29	+	t.Helper()
	30	+	for _, fd := range fds {
	31	+	signal(t, fd)
	32	+	}
	33	+	if err := p.raw.Control(func(ep uintptr) {
	34	+	var events [4]unix.EpollEvent
	35	+	if n, err := unix.EpollWait(int(ep), events[:], 0); n != 0 \|\| err != nil {
	36	+	t.Errorf("registrations left after Poll returned: %d %v", n, err)
	37	+	}
	38	+	}); err != nil {
	39	+	t.Fatal(err)
	40	+	}
	41	+	}
	42	+
	43	+	// Poll(fds, 0) reports existing readiness like poll(2): an expired netpoll
	44	+	// deadline never runs the readiness check.
	45	+	func TestPollZeroTimeoutReportsReady(t *testing.T) {
	46	+	p, err := New()
	47	+	if err != nil {
	48	+	t.Fatal(err)
	49	+	}
	50	+	defer p.Close()
	51	+	fd := eventfd(t)
	52	+	fds := []unix.PollFd{{Fd: -1}, {Fd: int32(fd), Events: unix.POLLIN}}
	53	+	if n, err := p.Poll(fds, 0); n != 0 \|\| err != nil {
	54	+	t.Fatalf("idle: %d %v", n, err)
	55	+	}
	56	+	signal(t, fd)
	57	+	if n, err := p.Poll(fds, 0); n != 1 \|\| err != nil \|\| fds[1].Revents&unix.POLLIN == 0 {
	58	+	t.Fatalf("ready: %d %v %+v", n, err, fds)
	59	+	}
	60	+	}
	61	+
	62	+	// A descriptor listed twice is watched once and removed once.
	63	+	func TestPollDuplicateDescriptor(t *testing.T) {
	64	+	p, err := New()
	65	+	if err != nil {
	66	+	t.Fatal(err)
	67	+	}
	68	+	defer p.Close()
	69	+	fd := eventfd(t)
	70	+	fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}, {Fd: int32(fd), Events: unix.POLLIN}}
	71	+	if n, err := p.Poll(fds, 1); n != 0 \|\| err != nil {
	72	+	t.Fatalf("duplicate: %d %v", n, err)
	73	+	}
	74	+	assertNoWatches(t, p, fd)
	75	+	}
	76	+
	77	+	// A failed registration removes the ones already added in the same call.
	78	+	func TestPollFailedWatchRemovesAdded(t *testing.T) {
	79	+	p, err := New()
	80	+	if err != nil {
	81	+	t.Fatal(err)
	82	+	}
	83	+	defer p.Close()
	84	+	fd := eventfd(t)
	85	+	var self int32
	86	+	if err := p.raw.Control(func(ep uintptr) { self = int32(ep) }); err != nil {
	87	+	t.Fatal(err)
	88	+	}
	89	+	// epoll refuses to watch itself (EINVAL) and is not ready.
	90	+	fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}, {Fd: self, Events: unix.POLLIN}}
	91	+	if _, err := p.Poll(fds, 1); err == nil {
	92	+	t.Fatal("watching the epoll fd itself succeeded")
	93	+	}
	94	+	assertNoWatches(t, p, fd)
	95	+	}
	96	+
	97	+	// A registration whose fd was closed while another reference keeps its file
	98	+	// open cannot be removed: Poll fails from then on instead of spinning.
	99	+	func TestPollFailsAfterLostRegistration(t *testing.T) {
	100	+	p, err := New()
	101	+	if err != nil {
	102	+	t.Fatal(err)
	103	+	}
	104	+	defer p.Close()
	105	+	fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
	106	+	if err != nil {
	107	+	t.Fatal(err)
	108	+	}
	109	+	dup, err := unix.Dup(fd)
	110	+	if err != nil {
	111	+	unix.Close(fd)
	112	+	t.Fatal(err)
	113	+	}
	114	+	defer unix.Close(dup)
	115	+	if err := p.raw.Control(func(ep uintptr) {
	116	+	ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: int32(fd)}
	117	+	if err := unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, fd, &ev); err != nil {
	118	+	t.Fatal(err)
	119	+	}
	120	+	}); err != nil {
	121	+	t.Fatal(err)
	122	+	}
	123	+	unix.Close(fd) // what a caller closing fd during a wait does
	124	+	p.unwatch([]unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}})
	125	+	other := eventfd(t)
	126	+	if _, err := p.Poll([]unix.PollFd{{Fd: int32(other), Events: unix.POLLIN}}, 1); err == nil {
	127	+	t.Fatal("Poll succeeded with a registration it could not remove")
	128	+	}
	129	+	}

@@ -0,0 +1,129 @@
	1	+	package fdwait
	2	+
	3	+	import (
	4	+	"testing"
	5	+
	6	+	"golang.org/x/sys/unix"
	7	+	)
	8	+
	9	+	func eventfd(t *testing.T) int {
	10	+	t.Helper()
	11	+	fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
	12	+	if err != nil {
	13	+	t.Fatal(err)
	14	+	}
	15	+	t.Cleanup(func() { unix.Close(fd) })
	16	+	return fd
	17	+	}
	18	+
	19	+	func signal(t *testing.T, fd int) {
	20	+	t.Helper()
	21	+	if _, err := unix.Write(fd, []byte{1, 0, 0, 0, 0, 0, 0, 0}); err != nil {
	22	+	t.Fatal(err)
	23	+	}
	24	+	}
	25	+
	26	+	// assertNoWatches fails if Poll left a registration behind, by making every
	27	+	// test descriptor readable and checking epoll reports nothing.
	28	+	func assertNoWatches(t testing.T, p Poller, fds ...int) {
	29	+	t.Helper()
	30	+	for _, fd := range fds {
	31	+	signal(t, fd)
	32	+	}
	33	+	if err := p.raw.Control(func(ep uintptr) {
	34	+	var events [4]unix.EpollEvent
	35	+	if n, err := unix.EpollWait(int(ep), events[:], 0); n != 0 \|\| err != nil {
	36	+	t.Errorf("registrations left after Poll returned: %d %v", n, err)
	37	+	}
	38	+	}); err != nil {
	39	+	t.Fatal(err)
	40	+	}
	41	+	}
	42	+
	43	+	// Poll(fds, 0) reports existing readiness like poll(2): an expired netpoll
	44	+	// deadline never runs the readiness check.
	45	+	func TestPollZeroTimeoutReportsReady(t *testing.T) {
	46	+	p, err := New()
	47	+	if err != nil {
	48	+	t.Fatal(err)
	49	+	}
	50	+	defer p.Close()
	51	+	fd := eventfd(t)
	52	+	fds := []unix.PollFd{{Fd: -1}, {Fd: int32(fd), Events: unix.POLLIN}}
	53	+	if n, err := p.Poll(fds, 0); n != 0 \|\| err != nil {
	54	+	t.Fatalf("idle: %d %v", n, err)
	55	+	}
	56	+	signal(t, fd)
	57	+	if n, err := p.Poll(fds, 0); n != 1 \|\| err != nil \|\| fds[1].Revents&unix.POLLIN == 0 {
	58	+	t.Fatalf("ready: %d %v %+v", n, err, fds)
	59	+	}
	60	+	}
	61	+
	62	+	// A descriptor listed twice is watched once and removed once.
	63	+	func TestPollDuplicateDescriptor(t *testing.T) {
	64	+	p, err := New()
	65	+	if err != nil {
	66	+	t.Fatal(err)
	67	+	}
	68	+	defer p.Close()
	69	+	fd := eventfd(t)
	70	+	fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}, {Fd: int32(fd), Events: unix.POLLIN}}
	71	+	if n, err := p.Poll(fds, 1); n != 0 \|\| err != nil {
	72	+	t.Fatalf("duplicate: %d %v", n, err)
	73	+	}
	74	+	assertNoWatches(t, p, fd)
	75	+	}
	76	+
	77	+	// A failed registration removes the ones already added in the same call.
	78	+	func TestPollFailedWatchRemovesAdded(t *testing.T) {
	79	+	p, err := New()
	80	+	if err != nil {
	81	+	t.Fatal(err)
	82	+	}
	83	+	defer p.Close()
	84	+	fd := eventfd(t)
	85	+	var self int32
	86	+	if err := p.raw.Control(func(ep uintptr) { self = int32(ep) }); err != nil {
	87	+	t.Fatal(err)
	88	+	}
	89	+	// epoll refuses to watch itself (EINVAL) and is not ready.
	90	+	fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}, {Fd: self, Events: unix.POLLIN}}
	91	+	if _, err := p.Poll(fds, 1); err == nil {
	92	+	t.Fatal("watching the epoll fd itself succeeded")
	93	+	}
	94	+	assertNoWatches(t, p, fd)
	95	+	}
	96	+
	97	+	// A registration whose fd was closed while another reference keeps its file
	98	+	// open cannot be removed: Poll fails from then on instead of spinning.
	99	+	func TestPollFailsAfterLostRegistration(t *testing.T) {
	100	+	p, err := New()
	101	+	if err != nil {
	102	+	t.Fatal(err)
	103	+	}
	104	+	defer p.Close()
	105	+	fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
	106	+	if err != nil {
	107	+	t.Fatal(err)
	108	+	}
	109	+	dup, err := unix.Dup(fd)
	110	+	if err != nil {
	111	+	unix.Close(fd)
	112	+	t.Fatal(err)
	113	+	}
	114	+	defer unix.Close(dup)
	115	+	if err := p.raw.Control(func(ep uintptr) {
	116	+	ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: int32(fd)}
	117	+	if err := unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, fd, &ev); err != nil {
	118	+	t.Fatal(err)
	119	+	}
	120	+	}); err != nil {
	121	+	t.Fatal(err)
	122	+	}
	123	+	unix.Close(fd) // what a caller closing fd during a wait does
	124	+	p.unwatch([]unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}})
	125	+	other := eventfd(t)
	126	+	if _, err := p.Poll([]unix.PollFd{{Fd: int32(other), Events: unix.POLLIN}}, 1); err == nil {
	127	+	t.Fatal("Poll succeeded with a registration it could not remove")
	128	+	}
	129	+	}

M internal/adapters/fdwait/poller.go +73 -27


@@ -3,74 +3,89 @@ package fdwait
3	3
4	4		import (
5	5		"errors"
	6	+	"fmt"
6	7		"os"
	8	+	"slices"
7	9		"syscall"
8	10		"time"
9	11
10	12		"golang.org/x/sys/unix"
11	13		)
12	14
13		-	// Poller owns an epoll descriptor and is used by one goroutine.
14		-	// Watched descriptors remain owned by the caller.
	15	+	// Poller owns an epoll descriptor and is used by one goroutine; Close may
	16	+	// also run from another goroutine to end a wait. Watched descriptors remain
	17	+	// owned by the caller, who must not close them while a Poll waits.
15	18		type Poller struct {
16	19		file *os.File
17	20		raw syscall.RawConn
	21	+	// broken is set when a registration could not be removed: it may keep
	22	+	// the epoll descriptor readable forever, so every later Poll fails
	23	+	// instead of spinning.
	24	+	broken error
18	25		}
19	26
20	27		func New() (*Poller, error) {
21	28		fd, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)
22	29		if err != nil {
23		-	return nil, err
	30	+	return nil, fmt.Errorf("fdwait epoll create: %w", err)
24	31		}
25	32		if err = unix.SetNonblock(fd, true); err != nil {
26	33		unix.Close(fd)
27		-	return nil, err
	34	+	return nil, fmt.Errorf("fdwait epoll nonblock: %w", err)
28	35		}
29	36		f := os.NewFile(uintptr(fd), "fdwait")
30	37		raw, err := f.SyscallConn()
31	38		if err != nil {
32	39		f.Close()
33		-	return nil, err
	40	+	return nil, fmt.Errorf("fdwait epoll conn: %w", err)
34	41		}
35	42		return &Poller{file: f, raw: raw}, nil
36	43		}
	44	+
37	45		func (p *Poller) Close() error { return p.file.Close() }
38	46
39		-	// Poll waits for readable/error events. It never consumes caller events.
40		-	// Each call explicitly removes its epoll registrations before the caller can
41		-	// close or recycle watched fds. Relying on close to remove them is unsafe:
42		-	// another fd or a kernel fence may retain the open file description.
	47	+	// Poll waits like poll(2) for readable/error events on fds and fills their
	48	+	// Revents. It never consumes caller events. Negative fds are ignored and a
	49	+	// descriptor listed twice is watched once. timeout is in milliseconds; a
	50	+	// negative one waits until an event or Close.
	51	+	//
	52	+	// Each call explicitly removes its epoll registrations before returning.
	53	+	// Relying on close to remove them is unsafe: another fd or a kernel fence
	54	+	// may retain the open file description and keep the epoll fd readable.
	55	+	// When nothing is ready yet, a descriptor epoll cannot watch (EPERM,
	56	+	// EINVAL, ELOOP for epoll nested too deep) fails the call.
43	57		func (p *Poller) Poll(fds []unix.PollFd, timeout int) (int, error) {
	58	+	if p.broken != nil {
	59	+	return 0, p.broken
	60	+	}
	61	+	// Ready descriptors need no registration, and an expired deadline
	62	+	// would skip the readiness check in RawConn.Read.
	63	+	if n, err := pollNow(fds); n != 0 \|\| err != nil \|\| timeout == 0 {
	64	+	return n, err
	65	+	}
44	66		var ctlErr error
45	67		added := 0
46	68		err := p.raw.Control(func(ep uintptr) {
47	69		for _, f := range fds {
48		-	if f.Fd < 0 {
49		-	added++
50		-	continue
51		-	}
52		-	ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: f.Fd}
53		-	if e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, int(f.Fd), &ev); e != nil {
54		-	ctlErr = e
55		-	return
56		-	}
57		-	added++
58		-	}
59		-	})
60		-	defer p.raw.Control(func(ep uintptr) {
61		-	for _, f := range fds[:added] {
62	70		if f.Fd >= 0 {
63		-	unix.EpollCtl(int(ep), unix.EPOLL_CTL_DEL, int(f.Fd), nil)
	71	+	ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: f.Fd}
	72	+	e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, int(f.Fd), &ev)
	73	+	if e != nil && !(errors.Is(e, unix.EEXIST) && listedBefore(fds, added)) {
	74	+	ctlErr = fmt.Errorf("fdwait watch fd %d: %w", f.Fd, e)
	75	+	return
	76	+	}
64	77		}
	78	+	added++
65	79		}
66	80		})
	81	+	defer p.unwatch(fds[:added])
67	82		if err != nil {
68	83		return 0, err
69	84		}
70	85		if ctlErr != nil {
71	86		return 0, ctlErr
72	87		}
73		-	if timeout >= 0 {
	88	+	if timeout > 0 {
74	89		err = p.file.SetReadDeadline(time.Now().Add(time.Duration(timeout) * time.Millisecond))
75	90		} else {
76	91		err = p.file.SetReadDeadline(time.Time{})

@@ -81,8 +96,8 @@ func (p *Poller) Poll(fds []unix.PollFd, timeout int) (int, error) {
81	96		var n int
82	97		var pollErr error
83	98		err = p.raw.Read(func(_ uintptr) bool {
84		-	n, pollErr = unix.Poll(fds, 0)
85		-	return n > 0 \|\| (pollErr != nil && !errors.Is(pollErr, unix.EINTR))
	99	+	n, pollErr = pollNow(fds)
	100	+	return n != 0 \|\| pollErr != nil
86	101		})
87	102		if errors.Is(err, os.ErrDeadlineExceeded) {
88	103		return 0, nil

@@ -92,3 +107,34 @@ func (p *Poller) Poll(fds []unix.PollFd, timeout int) (int, error) {
92	107		}
93	108		return n, pollErr
94	109		}
	110	+
	111	+	// unwatch removes the registrations Poll added for fds. A failure means a
	112	+	// watched fd was closed during the wait: its registration may outlive it.
	113	+	func (p *Poller) unwatch(fds []unix.PollFd) {
	114	+	p.raw.Control(func(ep uintptr) {
	115	+	for i, f := range fds {
	116	+	if f.Fd < 0 \|\| listedBefore(fds, i) {
	117	+	continue
	118	+	}
	119	+	if e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_DEL, int(f.Fd), nil); e != nil && p.broken == nil {
	120	+	p.broken = fmt.Errorf("fdwait unwatch fd %d (closed during a wait?): %w", f.Fd, e)
	121	+	}
	122	+	}
	123	+	})
	124	+	}
	125	+
	126	+	// listedBefore reports whether fds[i].Fd also appears earlier in fds.
	127	+	func listedBefore(fds []unix.PollFd, i int) bool {
	128	+	fd := fds[i].Fd
	129	+	return slices.ContainsFunc(fds[:i], func(f unix.PollFd) bool { return f.Fd == fd })
	130	+	}
	131	+
	132	+	// pollNow checks fds without blocking, retrying an interrupted check.
	133	+	func pollNow(fds []unix.PollFd) (int, error) {
	134	+	for {
	135	+	n, err := unix.Poll(fds, 0)
	136	+	if !errors.Is(err, unix.EINTR) {
	137	+	return n, err
	138	+	}
	139	+	}
	140	+	}

 import (
 	"errors"
+	"fmt"
 	"os"
+	"slices"
 	"syscall"
 	"time"
 	"golang.org/x/sys/unix"
 )
-// Poller owns an epoll descriptor and is used by one goroutine.
+// Poller owns an epoll descriptor and is used by one goroutine; Close may
-// Watched descriptors remain owned by the caller.
+// also run from another goroutine to end a wait. Watched descriptors remain
+// owned by the caller, who must not close them while a Poll waits.
 type Poller struct {
 	file *os.File
 	raw  syscall.RawConn
+	// broken is set when a registration could not be removed: it may keep
+	// the epoll descriptor readable forever, so every later Poll fails
+	// instead of spinning.
+	broken error
 }
 func New() (*Poller, error) {
 	fd, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("fdwait epoll create: %w", err)
 	}
 	if err = unix.SetNonblock(fd, true); err != nil {
 		unix.Close(fd)
-		return nil, err
+		return nil, fmt.Errorf("fdwait epoll nonblock: %w", err)
 	}
 	f := os.NewFile(uintptr(fd), "fdwait")
 	raw, err := f.SyscallConn()
 	if err != nil {
 		f.Close()
-		return nil, err
+		return nil, fmt.Errorf("fdwait epoll conn: %w", err)
 	}
 	return &Poller{file: f, raw: raw}, nil
 }
 func (p *Poller) Close() error { return p.file.Close() }
-// Poll waits for readable/error events. It never consumes caller events.
+// Poll waits like poll(2) for readable/error events on fds and fills their
-// Each call explicitly removes its epoll registrations before the caller can
+// Revents. It never consumes caller events. Negative fds are ignored and a
-// close or recycle watched fds. Relying on close to remove them is unsafe:
+// descriptor listed twice is watched once. timeout is in milliseconds; a
-// another fd or a kernel fence may retain the open file description.
+// negative one waits until an event or Close.
+//
+// Each call explicitly removes its epoll registrations before returning.
+// Relying on close to remove them is unsafe: another fd or a kernel fence
+// may retain the open file description and keep the epoll fd readable.
+// When nothing is ready yet, a descriptor epoll cannot watch (EPERM,
+// EINVAL, ELOOP for epoll nested too deep) fails the call.
 func (p *Poller) Poll(fds []unix.PollFd, timeout int) (int, error) {
+	if p.broken != nil {
+		return 0, p.broken
+	}
+	// Ready descriptors need no registration, and an expired deadline
+	// would skip the readiness check in RawConn.Read.
+	if n, err := pollNow(fds); n != 0 || err != nil || timeout == 0 {
+		return n, err
+	}
 	var ctlErr error
 	added := 0
 	err := p.raw.Control(func(ep uintptr) {
 		for _, f := range fds {
-			if f.Fd < 0 {
-				added++
-				continue
-			}
-			ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: f.Fd}
-			if e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, int(f.Fd), &ev); e != nil {
-				ctlErr = e
-				return
-			}
-			added++
-		}
-	})
-	defer p.raw.Control(func(ep uintptr) {
-		for _, f := range fds[:added] {
 			if f.Fd >= 0 {
-				unix.EpollCtl(int(ep), unix.EPOLL_CTL_DEL, int(f.Fd), nil)
+				ev := unix.EpollEvent{Events: unix.EPOLLIN, Fd: f.Fd}
+				e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_ADD, int(f.Fd), &ev)
+				if e != nil && !(errors.Is(e, unix.EEXIST) && listedBefore(fds, added)) {
+					ctlErr = fmt.Errorf("fdwait watch fd %d: %w", f.Fd, e)
+					return
+				}
 			}
+			added++
 		}
 	})
+	defer p.unwatch(fds[:added])
 	if err != nil {
 		return 0, err
 	}
 	if ctlErr != nil {
 		return 0, ctlErr
 	}
-	if timeout >= 0 {
+	if timeout > 0 {
 		err = p.file.SetReadDeadline(time.Now().Add(time.Duration(timeout) * time.Millisecond))
 	} else {
 		err = p.file.SetReadDeadline(time.Time{})
 	var n int
 	var pollErr error
 	err = p.raw.Read(func(_ uintptr) bool {
-		n, pollErr = unix.Poll(fds, 0)
+		n, pollErr = pollNow(fds)
-		return n > 0 || (pollErr != nil && !errors.Is(pollErr, unix.EINTR))
+		return n != 0 || pollErr != nil
 	})
 	if errors.Is(err, os.ErrDeadlineExceeded) {
 		return 0, nil
 	}
 	return n, pollErr
 }
+// unwatch removes the registrations Poll added for fds. A failure means a
+// watched fd was closed during the wait: its registration may outlive it.
+func (p *Poller) unwatch(fds []unix.PollFd) {
+	p.raw.Control(func(ep uintptr) {
+		for i, f := range fds {
+			if f.Fd < 0 || listedBefore(fds, i) {
+				continue
+			}
+			if e := unix.EpollCtl(int(ep), unix.EPOLL_CTL_DEL, int(f.Fd), nil); e != nil && p.broken == nil {
+				p.broken = fmt.Errorf("fdwait unwatch fd %d (closed during a wait?): %w", f.Fd, e)
+			}
+		}
+	})
+}
+// listedBefore reports whether fds[i].Fd also appears earlier in fds.
+func listedBefore(fds []unix.PollFd, i int) bool {
+	fd := fds[i].Fd
+	return slices.ContainsFunc(fds[:i], func(f unix.PollFd) bool { return f.Fd == fd })
+}
+// pollNow checks fds without blocking, retrying an interrupted check.
+func pollNow(fds []unix.PollFd) (int, error) {
+	for {
+		n, err := unix.Poll(fds, 0)
+		if !errors.Is(err, unix.EINTR) {
+			return n, err
+		}
+	}
+}

D internal/adapters/fdwait/reuse_test.go +0 -58

@@ -1,58 +0,0 @@
1	-	package fdwait
2	-
3	-	import (
4	-	"golang.org/x/sys/unix"
5	-	"testing"
6	-	)
7	-
8	-	// Closing a watched file removes it from epoll. Reusing its number must
9	-	// register the new file, not leave it waiting on the old registration.
10	-	func TestPollRecycledDescriptor(t *testing.T) {
11	-	p, err := New()
12	-	if err != nil {
13	-	t.Fatal(err)
14	-	}
15	-	defer p.Close()
16	-	fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
17	-	if err != nil {
18	-	t.Fatal(err)
19	-	}
20	-	fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
21	-	if _, err := p.Poll(fds, 1); err != nil {
22	-	unix.Close(fd)
23	-	t.Fatal(err)
24	-	}
25	-	unix.Close(fd)
26	-	replacement, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
27	-	if err != nil {
28	-	t.Fatal(err)
29	-	}
30	-	defer unix.Close(replacement)
31	-	if replacement != fd {
32	-	t.Skip("descriptor number not reused")
33	-	}
34	-	if _, err := unix.Write(replacement, []byte{1, 0, 0, 0, 0, 0, 0, 0}); err != nil {
35	-	t.Fatal(err)
36	-	}
37	-	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil {
38	-	t.Fatalf("reused descriptor: %d %v", n, err)
39	-	}
40	-	}
41	-
42	-	func TestPollHangup(t *testing.T) {
43	-	p, err := New()
44	-	if err != nil {
45	-	t.Fatal(err)
46	-	}
47	-	defer p.Close()
48	-	var pipe [2]int
49	-	if err := unix.Pipe2(pipe[:], unix.O_NONBLOCK\|unix.O_CLOEXEC); err != nil {
50	-	t.Fatal(err)
51	-	}
52	-	defer unix.Close(pipe[0])
53	-	unix.Close(pipe[1])
54	-	fds := []unix.PollFd{{Fd: int32(pipe[0]), Events: unix.POLLIN}}
55	-	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil \|\| fds[0].Revents&unix.POLLHUP == 0 {
56	-	t.Fatalf("hangup: %d %v %+v", n, err, fds)
57	-	}
58	-	}

@@ -1,58 +0,0 @@
1	-	package fdwait
2	-
3	-	import (
4	-	"golang.org/x/sys/unix"
5	-	"testing"
6	-	)
7	-
8	-	// Closing a watched file removes it from epoll. Reusing its number must
9	-	// register the new file, not leave it waiting on the old registration.
10	-	func TestPollRecycledDescriptor(t *testing.T) {
11	-	p, err := New()
12	-	if err != nil {
13	-	t.Fatal(err)
14	-	}
15	-	defer p.Close()
16	-	fd, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
17	-	if err != nil {
18	-	t.Fatal(err)
19	-	}
20	-	fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
21	-	if _, err := p.Poll(fds, 1); err != nil {
22	-	unix.Close(fd)
23	-	t.Fatal(err)
24	-	}
25	-	unix.Close(fd)
26	-	replacement, err := unix.Eventfd(0, unix.EFD_NONBLOCK\|unix.EFD_CLOEXEC)
27	-	if err != nil {
28	-	t.Fatal(err)
29	-	}
30	-	defer unix.Close(replacement)
31	-	if replacement != fd {
32	-	t.Skip("descriptor number not reused")
33	-	}
34	-	if _, err := unix.Write(replacement, []byte{1, 0, 0, 0, 0, 0, 0, 0}); err != nil {
35	-	t.Fatal(err)
36	-	}
37	-	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil {
38	-	t.Fatalf("reused descriptor: %d %v", n, err)
39	-	}
40	-	}
41	-
42	-	func TestPollHangup(t *testing.T) {
43	-	p, err := New()
44	-	if err != nil {
45	-	t.Fatal(err)
46	-	}
47	-	defer p.Close()
48	-	var pipe [2]int
49	-	if err := unix.Pipe2(pipe[:], unix.O_NONBLOCK\|unix.O_CLOEXEC); err != nil {
50	-	t.Fatal(err)
51	-	}
52	-	defer unix.Close(pipe[0])
53	-	unix.Close(pipe[1])
54	-	fds := []unix.PollFd{{Fd: int32(pipe[0]), Events: unix.POLLIN}}
55	-	if n, err := p.Poll(fds, 1000); n != 1 \|\| err != nil \|\| fds[0].Revents&unix.POLLHUP == 0 {
56	-	t.Fatalf("hangup: %d %v %+v", n, err, fds)
57	-	}
58	-	}

M internal/adapters/wayland/alloc_hotpath_test.go +1 -0


@@ -65,6 +65,7 @@ func TestCommitSyncStorageIsPerUpdate(t *testing.T) {
65	65		t.Fatal(err)
66	66		}
67	67		defer sw.pipeR.Close()
	68	+	defer sw.waiter.Close()
68	69		defer sw.close()
69	70		srv.syncWait = sw
70	71		tl := &timeline{dev: dev, handle: 7, alive: true}


@@ -65,6 +65,7 @@ func TestCommitSyncStorageIsPerUpdate(t *testing.T) {
65	t.Fatal(err)	65		t.Fatal(err)
66	}	66		}
67	defer sw.pipeR.Close()	67		defer sw.pipeR.Close()
		68	+	defer sw.waiter.Close()
68	defer sw.close()	69		defer sw.close()
69	srv.syncWait = sw	70		srv.syncWait = sw
70	tl := &timeline{dev: dev, handle: 7, alive: true}	71		tl := &timeline{dev: dev, handle: 7, alive: true}

M internal/adapters/wayland/syncobj.go +3 -9


@@ -207,7 +207,6 @@ func (p unixAcquirePoller) poll(fds []unix.PollFd, timeout int) (int, error) {
207	207		type syncWaiter struct {
208	208		poller acquirePoller
209	209		waiter *fdwait.Poller
210		-	pipeFD int
211	210		terminal bool // polling ended in failure; protected by mu
212	211		mu sync.Mutex
213	212		ready map[*syncWait]bool

@@ -238,7 +237,7 @@ func newSyncWaiter(wake func()) (*syncWaiter, error) {
238	237		w.Close()
239	238		return nil, err
240	239		}
241		-	return &syncWaiter{poller: unixAcquirePoller{p}, waiter: p, pipeFD: int(r.Fd()), ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil
	240	+	return &syncWaiter{poller: unixAcquirePoller{p}, waiter: p, ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil
242	241		}
243	242
244	243		// watch registers an eventfd for p; the wait is ready once it fires.

@@ -384,13 +383,8 @@ func (sw syncWaiter) fail(waits []syncWait) {
384	383		// run polls every watched eventfd; it owns and closes them.
385	384		func (sw *syncWaiter) run(ctx context.Context) {
386	385		defer sw.pipeR.Close()
387		-	if sw.waiter != nil {
388		-	defer sw.waiter.Close()
389		-	}
390		-	pipeFD := sw.pipeFD
391		-	if sw.waiter == nil { // bare test fixtures own their pipe
392		-	pipeFD = int(sw.pipeR.Fd())
393		-	}
	386	+	defer sw.waiter.Close()
	387	+	pipeFD := int(sw.pipeR.Fd())
394	388		var waits []*syncWait
395	389		defer func() {
396	390		sw.mu.Lock()

 type syncWaiter struct {
 	poller    acquirePoller
 	waiter    *fdwait.Poller
-	pipeFD    int
 	terminal  bool // polling ended in failure; protected by mu
 	mu        sync.Mutex
 	ready     map[*syncWait]bool
 		w.Close()
 		return nil, err
 	}
-	return &syncWaiter{poller: unixAcquirePoller{p}, waiter: p, pipeFD: int(r.Fd()), ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil
+	return &syncWaiter{poller: unixAcquirePoller{p}, waiter: p, ready: map[*syncWait]bool{}, wake: wake, pipeR: r, pipeW: w, log: zerowrap.Default()}, nil
 }
 // watch registers an eventfd for p; the wait is ready once it fires.
 // run polls every watched eventfd; it owns and closes them.
 func (sw *syncWaiter) run(ctx context.Context) {
 	defer sw.pipeR.Close()
-	if sw.waiter != nil {
+	defer sw.waiter.Close()
-		defer sw.waiter.Close()
+	pipeFD := int(sw.pipeR.Fd())
-	}
-	pipeFD := sw.pipeFD
-	if sw.waiter == nil { // bare test fixtures own their pipe
-		pipeFD = int(sw.pipeR.Fd())
-	}
 	var waits []*syncWait
 	defer func() {
 		sw.mu.Lock()

M internal/adapters/wayland/syncwait_test.go +2 -0


@@ -20,6 +20,7 @@ func TestSyncWaitAlreadySignalled(t *testing.T) {
20	20		t.Fatal(err)
21	21		}
22	22		defer sw.pipeR.Close()
	23	+	defer sw.waiter.Close()
23	24		defer sw.close()
24	25		w, err := sw.watch(syncPoint{tl: &timeline{dev: dev, handle: 7}, point: 4})
25	26		if err != nil \|\| !sw.fired(w) {

@@ -34,6 +35,7 @@ func TestImplicitPollErrorRejectedAndWarnOnce(t *testing.T) {
34	35		t.Fatal(err)
35	36		}
36	37		defer sw.pipeR.Close()
	38	+	defer sw.waiter.Close()
37	39		defer sw.close()
38	40		var output bytes.Buffer
39	41		sw.log = zerowrap.New(zerowrap.Config{Output: &output})


@@ -20,6 +20,7 @@ func TestSyncWaitAlreadySignalled(t *testing.T) {
20	t.Fatal(err)	20		t.Fatal(err)
21	}	21		}
22	defer sw.pipeR.Close()	22		defer sw.pipeR.Close()
		23	+	defer sw.waiter.Close()
23	defer sw.close()	24		defer sw.close()
24	w, err := sw.watch(syncPoint{tl: &timeline{dev: dev, handle: 7}, point: 4})	25		w, err := sw.watch(syncPoint{tl: &timeline{dev: dev, handle: 7}, point: 4})
25	if err != nil \|\| !sw.fired(w) {	26		if err != nil \|\| !sw.fired(w) {

@@ -34,6 +35,7 @@ func TestImplicitPollErrorRejectedAndWarnOnce(t *testing.T) {
34	t.Fatal(err)	35		t.Fatal(err)
35	}	36		}
36	defer sw.pipeR.Close()	37		defer sw.pipeR.Close()
		38	+	defer sw.waiter.Close()
37	defer sw.close()	39		defer sw.close()
38	var output bytes.Buffer	40		var output bytes.Buffer
39	sw.log = zerowrap.New(zerowrap.Config{Output: &output})	41		sw.log = zerowrap.New(zerowrap.Config{Output: &output})