feat(runtime): dispatch server packets from a growing read ring with widened sequences

commit: 8a907e5a6a57eca3be8cbfe1405bcbcb36d64716
author: brice · 12h ago · 2026-10-22 08:39 UTC
committer: brice · 11h ago · 2026-10-22 09:36 UTC
parent: 5f9aee7
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +758 -26

M conn.go +26 -7


@@ -16,6 +16,18 @@ type Conn struct {
16	16		root uint32
17	17
18	18		xidBase, xidMask uint32 // resource-id allocation, used by NewXID
	19	+
	20	+	// Read side, owned by Dispatch: rb[rpos:rend] is received, unconsumed.
	21	+	rb []byte
	22	+	rpos, rend int
	23	+	maxPacket int // MaxPacket; lowered by tests
	24	+	rerr error // sticky read failure
	25	+	seq Seq // last widened sequence number seen
	26	+
	27	+	// Callback arguments, reused per packet so nothing escapes.
	28	+	ev Event
	29	+	reply Reply
	30	+	xerr XError
19	31		}
20	32
21	33		// NewConn takes ownership of fd, a connected stream socket to an X server,

@@ -46,16 +58,23 @@ func setupConn(fd int, authName string, authData []byte) (*Conn, error) {
46	58		if err := unix.SetNonblock(fd, true); err != nil {
47	59		return nil, os.NewSyscallError("fcntl", err)
48	60		}
	61	+	return newConnState(fd, setup, info), nil
	62	+	}
	63	+
	64	+	// newConnState builds a Conn over an fd whose setup already completed.
	65	+	func newConnState(fd int, setup []byte, info setupInfo) *Conn {
49	66		c := &Conn{
50		-	fd: fd,
51		-	buf: NewBuf(0),
52		-	setup: setup,
53		-	root: info.root,
54		-	xidBase: info.xidBase,
55		-	xidMask: info.xidMask,
	67	+	fd: fd,
	68	+	buf: NewBuf(0),
	69	+	setup: setup,
	70	+	root: info.root,
	71	+	xidBase: info.xidBase,
	72	+	xidMask: info.xidMask,
	73	+	rb: make([]byte, readBufSize),
	74	+	maxPacket: MaxPacket,
56	75		}
57	76		c.buf.Ext = &c.ext
58		-	return c, nil
	77	+	return c
59	78		}
60	79
61	80		// Close closes the socket. Later calls, and Flush after Close, return

 	root  uint32
 	xidBase, xidMask uint32 // resource-id allocation, used by NewXID
+	// Read side, owned by Dispatch: rb[rpos:rend] is received, unconsumed.
+	rb         []byte
+	rpos, rend int
+	maxPacket  int   // MaxPacket; lowered by tests
+	rerr       error // sticky read failure
+	seq        Seq   // last widened sequence number seen
+	// Callback arguments, reused per packet so nothing escapes.
+	ev    Event
+	reply Reply
+	xerr  XError
 }
 // NewConn takes ownership of fd, a connected stream socket to an X server,
 	if err := unix.SetNonblock(fd, true); err != nil {
 		return nil, os.NewSyscallError("fcntl", err)
 	}
+	return newConnState(fd, setup, info), nil
+}
+// newConnState builds a Conn over an fd whose setup already completed.
+func newConnState(fd int, setup []byte, info setupInfo) *Conn {
 	c := &Conn{
-		fd:      fd,
+		fd:        fd,
-		buf:     NewBuf(0),
+		buf:       NewBuf(0),
-		setup:   setup,
+		setup:     setup,
-		root:    info.root,
+		root:      info.root,
-		xidBase: info.xidBase,
+		xidBase:   info.xidBase,
-		xidMask: info.xidMask,
+		xidMask:   info.xidMask,
+		rb:        make([]byte, readBufSize),
+		maxPacket: MaxPacket,
 	}
 	c.buf.Ext = &c.ext
-	return c, nil
+	return c
 }
 // Close closes the socket. Later calls, and Flush after Close, return

M conn_xserver_test.go +14 -19


@@ -120,7 +120,7 @@ func TestXserverDial(t *testing.T) {
120	120		t.Fatalf("Root %#x, setup view says otherwise", c.Root())
121	121		}
122	122
123		-	// A trivial round trip: GetInputFocus, then read its reply.
	123	+	// A round trip through Dispatch: GetInputFocus, then its reply.
124	124		seq := xproto.GetInputFocus(c.Buf())
125	125		if seq != 1 {
126	126		t.Fatalf("seq %d", seq)

@@ -128,34 +128,29 @@ func TestXserverDial(t *testing.T) {
128	128		if err := c.Flush(); err != nil {
129	129		t.Fatal(err)
130	130		}
131		-	reply := readPacket(t, c.Fd())
132		-	if reply[0] != 1 \|\| binary.LittleEndian.Uint16(reply[2:]) != uint16(seq) {
133		-	t.Fatalf("reply %x", reply)
	131	+	var got xturbo.Seq
	132	+	var replyLen int
	133	+	h := xturbo.Handler{OnReply: func(r *xturbo.Reply) { got, replyLen = r.Seq, len(r.Raw) }}
	134	+	dispatchUntil(t, c, &h, func() bool { return got != 0 })
	135	+	if got != seq \|\| replyLen != 32 {
	136	+	t.Fatalf("reply seq %d, %d bytes", got, replyLen)
134	137		}
135	138		}
136	139
137		-	// readPacket polls the non-blocking fd and reads one 32-byte packet.
138		-	func readPacket(t *testing.T, fd int) [32]byte {
	140	+	// dispatchUntil polls the connection and dispatches until done reports true.
	141	+	func dispatchUntil(t testing.T, c xturbo.Conn, h *xturbo.Handler, done func() bool) {
139	142		t.Helper()
140		-	var p [32]byte
141		-	got := 0
142	143		deadline := time.Now().Add(5 * time.Second)
143		-	for got < len(p) {
	144	+	for !done() {
144	145		if time.Now().After(deadline) {
145		-	t.Fatal("timeout reading reply")
	146	+	t.Fatal("timeout waiting for the server")
146	147		}
147		-	fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
	148	+	fds := []unix.PollFd{{Fd: int32(c.Fd()), Events: unix.POLLIN}}
148	149		if _, err := unix.Poll(fds, 1000); err != nil && err != unix.EINTR {
149	150		t.Fatal(err)
150	151		}
151		-	n, err := unix.Read(fd, p[got:])
152		-	if err == unix.EAGAIN \|\| err == unix.EINTR {
153		-	continue
154		-	}
155		-	if err != nil \|\| n == 0 {
156		-	t.Fatalf("read: %d %v", n, err)
	152	+	if err := c.Dispatch(h); err != nil {
	153	+	t.Fatal(err)
157	154		}
158		-	got += n
159	155		}
160		-	return p
161	156		}

 		t.Fatalf("Root %#x, setup view says otherwise", c.Root())
 	}
-	// A trivial round trip: GetInputFocus, then read its reply.
+	// A round trip through Dispatch: GetInputFocus, then its reply.
 	seq := xproto.GetInputFocus(c.Buf())
 	if seq != 1 {
 		t.Fatalf("seq %d", seq)
 	if err := c.Flush(); err != nil {
 		t.Fatal(err)
 	}
-	reply := readPacket(t, c.Fd())
+	var got xturbo.Seq
-	if reply[0] != 1 || binary.LittleEndian.Uint16(reply[2:]) != uint16(seq) {
+	var replyLen int
-		t.Fatalf("reply %x", reply)
+	h := xturbo.Handler{OnReply: func(r *xturbo.Reply) { got, replyLen = r.Seq, len(r.Raw) }}
+	dispatchUntil(t, c, &h, func() bool { return got != 0 })
+	if got != seq || replyLen != 32 {
+		t.Fatalf("reply seq %d, %d bytes", got, replyLen)
 	}
 }
-// readPacket polls the non-blocking fd and reads one 32-byte packet.
+// dispatchUntil polls the connection and dispatches until done reports true.
-func readPacket(t *testing.T, fd int) [32]byte {
+func dispatchUntil(t *testing.T, c *xturbo.Conn, h *xturbo.Handler, done func() bool) {
 	t.Helper()
-	var p [32]byte
-	got := 0
 	deadline := time.Now().Add(5 * time.Second)
-	for got < len(p) {
+	for !done() {
 		if time.Now().After(deadline) {
-			t.Fatal("timeout reading reply")
+			t.Fatal("timeout waiting for the server")
 		}
-		fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
+		fds := []unix.PollFd{{Fd: int32(c.Fd()), Events: unix.POLLIN}}
 		if _, err := unix.Poll(fds, 1000); err != nil && err != unix.EINTR {
 			t.Fatal(err)
 		}
-		n, err := unix.Read(fd, p[got:])
+		if err := c.Dispatch(h); err != nil {
-		if err == unix.EAGAIN || err == unix.EINTR {
+			t.Fatal(err)
-			continue
-		}
-		if err != nil || n == 0 {
-			t.Fatalf("read: %d %v", n, err)
 		}
-		got += n
 	}
-	return p
 }

A dispatch.go +207 -0

@@ -0,0 +1,207 @@
	1	+	package xturbo
	2	+
	3	+	import (
	4	+	"encoding/binary"
	5	+	"errors"
	6	+	"io"
	7	+	"os"
	8	+
	9	+	"golang.org/x/sys/unix"
	10	+	)
	11	+
	12	+	// MaxPacket bounds one server packet (a reply or an XGE event) in bytes. The
	13	+	// read ring grows up to this size; a larger packet fails Dispatch with
	14	+	// ErrPacketTooLarge and the connection is dead.
	15	+	const MaxPacket = 16 << 20
	16	+
	17	+	// ErrPacketTooLarge reports a server packet longer than MaxPacket.
	18	+	var ErrPacketTooLarge = errors.New("xturbo: packet too large")
	19	+
	20	+	const (
	21	+	readBufSize = 64 << 10
	22	+	packetHeadSize = 32
	23	+	)
	24	+
	25	+	// Handler receives decoded packets from Dispatch, in wire order, on the
	26	+	// goroutine calling Dispatch. A nil field drops that packet class. Fill it
	27	+	// once at init; the pointers passed to callbacks are owned by the Conn and are
	28	+	// reused for the next packet, so callbacks must not retain them, nor the
	29	+	// views they hold (Reply.Raw, Event.Tail). Callbacks may queue requests and
	30	+	// call Expect but must not call Dispatch.
	31	+	type Handler struct {
	32	+	OnEvent func(ev *Event)
	33	+	OnReply func(r *Reply)
	34	+	OnError func(e *XError)
	35	+	}
	36	+
	37	+	// Reply is one reply packet. Seq is widened; Kind and Args come from the
	38	+	// Expect slot of Seq (Kind 0 when none matched). Raw is the whole reply,
	39	+	// 32+4*length bytes, valid only until the callback returns.
	40	+	type Reply struct {
	41	+	Seq Seq
	42	+	Kind uint16
	43	+	Args [4]uint32
	44	+	Raw []byte
	45	+	}
	46	+
	47	+	var noHandler Handler
	48	+
	49	+	// Dispatch reads everything the server has sent without blocking and calls h
	50	+	// for every complete packet, in wire order. It returns nil once the socket
	51	+	// would block, io.EOF when the server closed the connection and
	52	+	// ErrPacketTooLarge on a packet beyond MaxPacket. Read errors are sticky and
	53	+	// also fail the Buf. Dispatch never flushes. A nil h drops every packet.
	54	+	func (c Conn) Dispatch(h Handler) error {
	55	+	if c.fd < 0 {
	56	+	return ErrClosed
	57	+	}
	58	+	if c.rerr != nil {
	59	+	return c.rerr
	60	+	}
	61	+	if h == nil {
	62	+	h = &noHandler
	63	+	}
	64	+	for {
	65	+	if err := c.drain(h); err != nil {
	66	+	return c.failRead(err)
	67	+	}
	68	+	if c.fd < 0 { // a callback closed the Conn
	69	+	return ErrClosed
	70	+	}
	71	+	n, err := unix.Read(c.fd, c.rb[c.rend:])
	72	+	switch {
	73	+	case err == unix.EINTR:
	74	+	continue
	75	+	case err == unix.EAGAIN:
	76	+	return nil
	77	+	case err == unix.ECONNRESET:
	78	+	return c.failRead(io.EOF)
	79	+	case err != nil:
	80	+	return c.failRead(os.NewSyscallError("read", err))
	81	+	case n == 0:
	82	+	return c.failRead(io.EOF)
	83	+	}
	84	+	c.rend += n
	85	+	}
	86	+	}
	87	+
	88	+	func (c *Conn) failRead(err error) error {
	89	+	c.rerr = err
	90	+	if err == io.EOF {
	91	+	c.buf.Fail(ErrClosed)
	92	+	} else {
	93	+	c.buf.Fail(err)
	94	+	}
	95	+	return err
	96	+	}
	97	+
	98	+	// drain handles every complete packet in the ring, then makes room for the
	99	+	// next one (compacting, growing up to maxPacket).
	100	+	func (c Conn) drain(h Handler) error {
	101	+	for c.fd >= 0 {
	102	+	avail := c.rend - c.rpos
	103	+	need := packetHeadSize
	104	+	if avail >= packetHeadSize {
	105	+	size, ok := packetSize(c.rb[c.rpos:], c.maxPacket)
	106	+	if !ok {
	107	+	return ErrPacketTooLarge
	108	+	}
	109	+	need = size
	110	+	if avail >= need {
	111	+	end := c.rpos + need
	112	+	p := c.rb[c.rpos:end:end]
	113	+	c.packet(h, p)
	114	+	if poison {
	115	+	fillPoison(p)
	116	+	}
	117	+	c.rpos = end
	118	+	continue
	119	+	}
	120	+	}
	121	+	c.makeRoom(need)
	122	+	break
	123	+	}
	124	+	return nil
	125	+	}
	126	+
	127	+	// packetSize returns the length of the packet starting at p (len(p) >= 32):
	128	+	// 32 for errors and events, 32+4*length for replies and XGE events.
	129	+	func packetSize(p []byte, limit int) (int, bool) {
	130	+	n := uint64(packetHeadSize)
	131	+	if p[0] == 1 \|\| p[0]&0x7f == codeGenericEvent {
	132	+	n += 4 * uint64(binary.LittleEndian.Uint32(p[4:]))
	133	+	}
	134	+	if n > uint64(limit) {
	135	+	return 0, false
	136	+	}
	137	+	return int(n), true
	138	+	}
	139	+
	140	+	// makeRoom moves the unconsumed bytes to the start of the ring and grows it
	141	+	// by doubling until need bytes fit. need <= maxPacket.
	142	+	func (c *Conn) makeRoom(need int) {
	143	+	if c.rpos > 0 {
	144	+	k := copy(c.rb, c.rb[c.rpos:c.rend])
	145	+	if poison {
	146	+	fillPoison(c.rb[k:c.rend])
	147	+	}
	148	+	c.rpos, c.rend = 0, k
	149	+	}
	150	+	if need <= len(c.rb) {
	151	+	return
	152	+	}
	153	+	size := max(len(c.rb), packetHeadSize)
	154	+	for size < need {
	155	+	size *= 2
	156	+	}
	157	+	grown := make([]byte, min(size, c.maxPacket))
	158	+	copy(grown, c.rb[:c.rend])
	159	+	if poison {
	160	+	fillPoison(c.rb)
	161	+	}
	162	+	c.rb = grown
	163	+	}
	164	+
	165	+	// packet decodes one complete packet and calls h.
	166	+	func (c Conn) packet(h Handler, p []byte) {
	167	+	switch p[0] {
	168	+	case 0:
	169	+	decodeError(p, &c.ext, &c.xerr)
	170	+	c.xerr.Seq = c.widen(c.xerr.Seq)
	171	+	if h.OnError != nil {
	172	+	h.OnError(&c.xerr)
	173	+	}
	174	+	case 1:
	175	+	c.reply = Reply{Seq: c.widen(Seq(binary.LittleEndian.Uint16(p[2:]))), Raw: p}
	176	+	if h.OnReply != nil {
	177	+	h.OnReply(&c.reply)
	178	+	}
	179	+	default:
	180	+	ev, _, ok := DecodeEvent(p, &c.ext)
	181	+	if !ok { // a sent event claiming code 0 or 1
	182	+	ev = Event{Kind: KindUnknown, SendEvent: true, Seq: Seq(binary.LittleEndian.Uint16(p[2:]))}
	183	+	copy(ev.Raw[:], p)
	184	+	}
	185	+	if p[0]&0x7f == codeKeymapNotify {
	186	+	ev.Seq = c.seq
	187	+	} else {
	188	+	ev.Seq = c.widen(ev.Seq)
	189	+	}
	190	+	c.ev = ev
	191	+	if h.OnEvent != nil {
	192	+	h.OnEvent(&c.ev)
	193	+	}
	194	+	}
	195	+	}
	196	+
	197	+	// widen turns a 16-bit wire sequence into the full Seq, relative to the last
	198	+	// one seen, and acknowledges it to the Buf.
	199	+	func (c *Conn) widen(wire Seq) Seq {
	200	+	full := c.seq&^0xffff \| wire&0xffff
	201	+	if full < c.seq {
	202	+	full += 0x10000
	203	+	}
	204	+	c.seq = full
	205	+	c.buf.ack(full)
	206	+	return full
	207	+	}

@@ -0,0 +1,207 @@
	1	+	package xturbo
	2	+
	3	+	import (
	4	+	"encoding/binary"
	5	+	"errors"
	6	+	"io"
	7	+	"os"
	8	+
	9	+	"golang.org/x/sys/unix"
	10	+	)
	11	+
	12	+	// MaxPacket bounds one server packet (a reply or an XGE event) in bytes. The
	13	+	// read ring grows up to this size; a larger packet fails Dispatch with
	14	+	// ErrPacketTooLarge and the connection is dead.
	15	+	const MaxPacket = 16 << 20
	16	+
	17	+	// ErrPacketTooLarge reports a server packet longer than MaxPacket.
	18	+	var ErrPacketTooLarge = errors.New("xturbo: packet too large")
	19	+
	20	+	const (
	21	+	readBufSize = 64 << 10
	22	+	packetHeadSize = 32
	23	+	)
	24	+
	25	+	// Handler receives decoded packets from Dispatch, in wire order, on the
	26	+	// goroutine calling Dispatch. A nil field drops that packet class. Fill it
	27	+	// once at init; the pointers passed to callbacks are owned by the Conn and are
	28	+	// reused for the next packet, so callbacks must not retain them, nor the
	29	+	// views they hold (Reply.Raw, Event.Tail). Callbacks may queue requests and
	30	+	// call Expect but must not call Dispatch.
	31	+	type Handler struct {
	32	+	OnEvent func(ev *Event)
	33	+	OnReply func(r *Reply)
	34	+	OnError func(e *XError)
	35	+	}
	36	+
	37	+	// Reply is one reply packet. Seq is widened; Kind and Args come from the
	38	+	// Expect slot of Seq (Kind 0 when none matched). Raw is the whole reply,
	39	+	// 32+4*length bytes, valid only until the callback returns.
	40	+	type Reply struct {
	41	+	Seq Seq
	42	+	Kind uint16
	43	+	Args [4]uint32
	44	+	Raw []byte
	45	+	}
	46	+
	47	+	var noHandler Handler
	48	+
	49	+	// Dispatch reads everything the server has sent without blocking and calls h
	50	+	// for every complete packet, in wire order. It returns nil once the socket
	51	+	// would block, io.EOF when the server closed the connection and
	52	+	// ErrPacketTooLarge on a packet beyond MaxPacket. Read errors are sticky and
	53	+	// also fail the Buf. Dispatch never flushes. A nil h drops every packet.
	54	+	func (c Conn) Dispatch(h Handler) error {
	55	+	if c.fd < 0 {
	56	+	return ErrClosed
	57	+	}
	58	+	if c.rerr != nil {
	59	+	return c.rerr
	60	+	}
	61	+	if h == nil {
	62	+	h = &noHandler
	63	+	}
	64	+	for {
	65	+	if err := c.drain(h); err != nil {
	66	+	return c.failRead(err)
	67	+	}
	68	+	if c.fd < 0 { // a callback closed the Conn
	69	+	return ErrClosed
	70	+	}
	71	+	n, err := unix.Read(c.fd, c.rb[c.rend:])
	72	+	switch {
	73	+	case err == unix.EINTR:
	74	+	continue
	75	+	case err == unix.EAGAIN:
	76	+	return nil
	77	+	case err == unix.ECONNRESET:
	78	+	return c.failRead(io.EOF)
	79	+	case err != nil:
	80	+	return c.failRead(os.NewSyscallError("read", err))
	81	+	case n == 0:
	82	+	return c.failRead(io.EOF)
	83	+	}
	84	+	c.rend += n
	85	+	}
	86	+	}
	87	+
	88	+	func (c *Conn) failRead(err error) error {
	89	+	c.rerr = err
	90	+	if err == io.EOF {
	91	+	c.buf.Fail(ErrClosed)
	92	+	} else {
	93	+	c.buf.Fail(err)
	94	+	}
	95	+	return err
	96	+	}
	97	+
	98	+	// drain handles every complete packet in the ring, then makes room for the
	99	+	// next one (compacting, growing up to maxPacket).
	100	+	func (c Conn) drain(h Handler) error {
	101	+	for c.fd >= 0 {
	102	+	avail := c.rend - c.rpos
	103	+	need := packetHeadSize
	104	+	if avail >= packetHeadSize {
	105	+	size, ok := packetSize(c.rb[c.rpos:], c.maxPacket)
	106	+	if !ok {
	107	+	return ErrPacketTooLarge
	108	+	}
	109	+	need = size
	110	+	if avail >= need {
	111	+	end := c.rpos + need
	112	+	p := c.rb[c.rpos:end:end]
	113	+	c.packet(h, p)
	114	+	if poison {
	115	+	fillPoison(p)
	116	+	}
	117	+	c.rpos = end
	118	+	continue
	119	+	}
	120	+	}
	121	+	c.makeRoom(need)
	122	+	break
	123	+	}
	124	+	return nil
	125	+	}
	126	+
	127	+	// packetSize returns the length of the packet starting at p (len(p) >= 32):
	128	+	// 32 for errors and events, 32+4*length for replies and XGE events.
	129	+	func packetSize(p []byte, limit int) (int, bool) {
	130	+	n := uint64(packetHeadSize)
	131	+	if p[0] == 1 \|\| p[0]&0x7f == codeGenericEvent {
	132	+	n += 4 * uint64(binary.LittleEndian.Uint32(p[4:]))
	133	+	}
	134	+	if n > uint64(limit) {
	135	+	return 0, false
	136	+	}
	137	+	return int(n), true
	138	+	}
	139	+
	140	+	// makeRoom moves the unconsumed bytes to the start of the ring and grows it
	141	+	// by doubling until need bytes fit. need <= maxPacket.
	142	+	func (c *Conn) makeRoom(need int) {
	143	+	if c.rpos > 0 {
	144	+	k := copy(c.rb, c.rb[c.rpos:c.rend])
	145	+	if poison {
	146	+	fillPoison(c.rb[k:c.rend])
	147	+	}
	148	+	c.rpos, c.rend = 0, k
	149	+	}
	150	+	if need <= len(c.rb) {
	151	+	return
	152	+	}
	153	+	size := max(len(c.rb), packetHeadSize)
	154	+	for size < need {
	155	+	size *= 2
	156	+	}
	157	+	grown := make([]byte, min(size, c.maxPacket))
	158	+	copy(grown, c.rb[:c.rend])
	159	+	if poison {
	160	+	fillPoison(c.rb)
	161	+	}
	162	+	c.rb = grown
	163	+	}
	164	+
	165	+	// packet decodes one complete packet and calls h.
	166	+	func (c Conn) packet(h Handler, p []byte) {
	167	+	switch p[0] {
	168	+	case 0:
	169	+	decodeError(p, &c.ext, &c.xerr)
	170	+	c.xerr.Seq = c.widen(c.xerr.Seq)
	171	+	if h.OnError != nil {
	172	+	h.OnError(&c.xerr)
	173	+	}
	174	+	case 1:
	175	+	c.reply = Reply{Seq: c.widen(Seq(binary.LittleEndian.Uint16(p[2:]))), Raw: p}
	176	+	if h.OnReply != nil {
	177	+	h.OnReply(&c.reply)
	178	+	}
	179	+	default:
	180	+	ev, _, ok := DecodeEvent(p, &c.ext)
	181	+	if !ok { // a sent event claiming code 0 or 1
	182	+	ev = Event{Kind: KindUnknown, SendEvent: true, Seq: Seq(binary.LittleEndian.Uint16(p[2:]))}
	183	+	copy(ev.Raw[:], p)
	184	+	}
	185	+	if p[0]&0x7f == codeKeymapNotify {
	186	+	ev.Seq = c.seq
	187	+	} else {
	188	+	ev.Seq = c.widen(ev.Seq)
	189	+	}
	190	+	c.ev = ev
	191	+	if h.OnEvent != nil {
	192	+	h.OnEvent(&c.ev)
	193	+	}
	194	+	}
	195	+	}
	196	+
	197	+	// widen turns a 16-bit wire sequence into the full Seq, relative to the last
	198	+	// one seen, and acknowledges it to the Buf.
	199	+	func (c *Conn) widen(wire Seq) Seq {
	200	+	full := c.seq&^0xffff \| wire&0xffff
	201	+	if full < c.seq {
	202	+	full += 0x10000
	203	+	}
	204	+	c.seq = full
	205	+	c.buf.ack(full)
	206	+	return full
	207	+	}

A dispatch_poison_test.go +36 -0

@@ -0,0 +1,36 @@
	1	+	//go:build xturbo_poison
	2	+
	3	+	package xturbo
	4	+
	5	+	import "testing"
	6	+
	7	+	// TestPoisonRetainedViews keeps Reply.Raw and Event.Tail past their callback
	8	+	// (a lifetime bug) and checks the runtime overwrote them.
	9	+	func TestPoisonRetainedViews(t *testing.T) {
	10	+	c, srv := dispatchConn(t)
	11	+	var raw, tail []byte
	12	+	h := Handler{
	13	+	OnReply: func(r *Reply) { raw = r.Raw },
	14	+	OnEvent: func(ev *Event) { tail = ev.Tail },
	15	+	}
	16	+	stream := append(pktReply(1, 64), pktXGE(2, 1, 64)...)
	17	+	feed(t, c, srv, stream, &h)
	18	+	if len(raw) != 96 \|\| len(tail) != 64 {
	19	+	t.Fatal(len(raw), len(tail))
	20	+	}
	21	+	for _, p := range [][]byte{raw, tail} {
	22	+	for i, v := range p {
	23	+	if v != 0xa5 {
	24	+	t.Fatalf("retained byte %d = %#x, want poison", i, v)
	25	+	}
	26	+	}
	27	+	}
	28	+
	29	+	// Growth poisons the old ring too.
	30	+	feed(t, c, srv, pktReply(3, 8), &h)
	31	+	old := c.rb
	32	+	feed(t, c, srv, pktReply(4, 200<<10), &h)
	33	+	if &old[0] == &c.rb[0] \|\| old[0] != 0xa5 \|\| old[len(old)-1] != 0xa5 {
	34	+	t.Fatal("old ring not poisoned after growth")
	35	+	}
	36	+	}

@@ -0,0 +1,36 @@
	1	+	//go:build xturbo_poison
	2	+
	3	+	package xturbo
	4	+
	5	+	import "testing"
	6	+
	7	+	// TestPoisonRetainedViews keeps Reply.Raw and Event.Tail past their callback
	8	+	// (a lifetime bug) and checks the runtime overwrote them.
	9	+	func TestPoisonRetainedViews(t *testing.T) {
	10	+	c, srv := dispatchConn(t)
	11	+	var raw, tail []byte
	12	+	h := Handler{
	13	+	OnReply: func(r *Reply) { raw = r.Raw },
	14	+	OnEvent: func(ev *Event) { tail = ev.Tail },
	15	+	}
	16	+	stream := append(pktReply(1, 64), pktXGE(2, 1, 64)...)
	17	+	feed(t, c, srv, stream, &h)
	18	+	if len(raw) != 96 \|\| len(tail) != 64 {
	19	+	t.Fatal(len(raw), len(tail))
	20	+	}
	21	+	for _, p := range [][]byte{raw, tail} {
	22	+	for i, v := range p {
	23	+	if v != 0xa5 {
	24	+	t.Fatalf("retained byte %d = %#x, want poison", i, v)
	25	+	}
	26	+	}
	27	+	}
	28	+
	29	+	// Growth poisons the old ring too.
	30	+	feed(t, c, srv, pktReply(3, 8), &h)
	31	+	old := c.rb
	32	+	feed(t, c, srv, pktReply(4, 200<<10), &h)
	33	+	if &old[0] == &c.rb[0] \|\| old[0] != 0xa5 \|\| old[len(old)-1] != 0xa5 {
	34	+	t.Fatal("old ring not poisoned after growth")
	35	+	}
	36	+	}

A dispatch_test.go +475 -0

@@ -0,0 +1,475 @@
	1	+	package xturbo
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"encoding/binary"
	6	+	"errors"
	7	+	"fmt"
	8	+	"io"
	9	+	"testing"
	10	+
	11	+	"bnema.dev/xturbo/internal/xtest"
	12	+	"golang.org/x/sys/unix"
	13	+	)
	14	+
	15	+	// dispatchConn returns a Conn over one end of a non-blocking socketpair with
	16	+	// default kernel buffers, and the other end acting as the server. The XFIXES
	17	+	// and Generic Event extensions are negotiated at the golden capture's bases.
	18	+	func dispatchConn(t testing.TB) (*Conn, int) {
	19	+	t.Helper()
	20	+	fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM\|unix.SOCK_NONBLOCK\|unix.SOCK_CLOEXEC, 0)
	21	+	if err != nil {
	22	+	t.Fatal(err)
	23	+	}
	24	+	c := newConnState(fds[0], nil, setupInfo{})
	25	+	t.Cleanup(func() { _ = c.Close(); _ = unix.Close(fds[1]) })
	26	+	c.ext.Set(ExtXFixes, 138, 87, 140)
	27	+	c.ext.Set(ExtGenericEvent, 0x83, 0, 0)
	28	+	c.buf.next = 1 << 40 // every widened seq in the tests is acknowledgeable
	29	+	return c, fds[1]
	30	+	}
	31	+
	32	+	// Packet builders. seq is the 16-bit wire value.
	33	+	func pktEvent(code uint8, seq uint16, fill byte) []byte {
	34	+	p := bytes.Repeat([]byte{fill}, 32)
	35	+	p[0] = code
	36	+	binary.LittleEndian.PutUint16(p[2:], seq)
	37	+	return p
	38	+	}
	39	+
	40	+	func pktReply(seq uint16, data int) []byte {
	41	+	words := (data + 3) / 4
	42	+	p := make([]byte, 32+4*words)
	43	+	p[0], p[1] = 1, 7
	44	+	binary.LittleEndian.PutUint16(p[2:], seq)
	45	+	binary.LittleEndian.PutUint32(p[4:], uint32(words))
	46	+	for i := 8; i < len(p); i++ {
	47	+	p[i] = byte(i * 7)
	48	+	}
	49	+	return p
	50	+	}
	51	+
	52	+	func pktError(code uint8, seq uint16, major uint8) []byte {
	53	+	p := make([]byte, 32)
	54	+	p[1] = code
	55	+	binary.LittleEndian.PutUint16(p[2:], seq)
	56	+	binary.LittleEndian.PutUint32(p[4:], 0xdead)
	57	+	p[10] = major
	58	+	return p
	59	+	}
	60	+
	61	+	func pktXGE(seq uint16, evtype uint16, tail int) []byte {
	62	+	words := (tail + 3) / 4
	63	+	p := make([]byte, 32+4*words)
	64	+	p[0], p[1] = codeGenericEvent, 0x83
	65	+	binary.LittleEndian.PutUint16(p[2:], seq)
	66	+	binary.LittleEndian.PutUint32(p[4:], uint32(words))
	67	+	binary.LittleEndian.PutUint16(p[8:], evtype)
	68	+	for i := 32; i < len(p); i++ {
	69	+	p[i] = byte(i)
	70	+	}
	71	+	return p
	72	+	}
	73	+
	74	+	// record is one callback as seen by a test handler; data is a copy of the
	75	+	// packet bytes (Raw, plus Tail for events).
	76	+	type record struct {
	77	+	class byte // 'e', 'r', 'x'
	78	+	seq Seq
	79	+	kind uint32
	80	+	data []byte
	81	+	}
	82	+
	83	+	func (r record) String() string {
	84	+	return fmt.Sprintf("%c seq %d kind %#x %d bytes", r.class, r.seq, r.kind, len(r.data))
	85	+	}
	86	+
	87	+	type recorder struct {
	88	+	got []record
	89	+	h Handler
	90	+	}
	91	+
	92	+	func newRecorder() *recorder {
	93	+	r := &recorder{}
	94	+	r.h = Handler{
	95	+	OnEvent: func(ev *Event) {
	96	+	r.got = append(r.got, record{'e', ev.Seq, uint32(ev.Kind), append(bytes.Clone(ev.Raw[:]), ev.Tail...)})
	97	+	},
	98	+	OnReply: func(rp *Reply) {
	99	+	r.got = append(r.got, record{'r', rp.Seq, uint32(rp.Kind), bytes.Clone(rp.Raw)})
	100	+	},
	101	+	OnError: func(e *XError) {
	102	+	r.got = append(r.got, record{'x', e.Seq, uint32(e.Code), bytes.Clone(e.Raw[:])})
	103	+	},
	104	+	}
	105	+	return r
	106	+	}
	107	+
	108	+	// feed writes data to the server end, dispatching whenever the socket fills,
	109	+	// then dispatches what remains.
	110	+	func feed(t testing.TB, c Conn, srv int, data []byte, h Handler) {
	111	+	t.Helper()
	112	+	for len(data) > 0 {
	113	+	n, err := unix.Write(srv, data)
	114	+	if err == unix.EAGAIN {
	115	+	n = 0
	116	+	} else if err != nil {
	117	+	t.Fatal(err)
	118	+	}
	119	+	data = data[n:]
	120	+	if err := c.Dispatch(h); err != nil {
	121	+	t.Fatal(err)
	122	+	}
	123	+	}
	124	+	if err := c.Dispatch(h); err != nil {
	125	+	t.Fatal(err)
	126	+	}
	127	+	}
	128	+
	129	+	func golden(t testing.TB, name string) []byte {
	130	+	t.Helper()
	131	+	raw, err := xtest.LoadHex("testdata/golden/" + name + ".hex")
	132	+	if err != nil {
	133	+	t.Fatal(err)
	134	+	}
	135	+	return raw
	136	+	}
	137	+
	138	+	// scriptedStream is a mixed stream of goldens and synthetic packets with
	139	+	// increasing wire sequence numbers, and the callbacks it must produce.
	140	+	func scriptedStream(t testing.TB) ([]byte, []record) {
	141	+	var stream []byte
	142	+	var want []record
	143	+	add := func(class byte, kind uint32, p []byte, seq uint16) {
	144	+	binary.LittleEndian.PutUint16(p[2:], seq)
	145	+	stream = append(stream, p...)
	146	+	want = append(want, record{class, Seq(seq), kind, p})
	147	+	}
	148	+	add('e', 20, golden(t, "xproto/MapRequest.event"), 3)
	149	+	add('r', 0, golden(t, "xproto/GetGeometry.reply"), 4)
	150	+	add('x', 3, golden(t, "xproto/BadWindow.error"), 5)
	151	+	add('e', uint32(ExtGenericEvent)<<8\|0x81, golden(t, "xproto/GeGeneric.event"), 6)
	152	+	add('e', uint32(ExtXFixes)<<8, golden(t, "xfixes/SelectionNotify.event"), 6)
	153	+	add('r', 0, golden(t, "xproto/GetProperty.reply"), 7)
	154	+	km := golden(t, "xproto/KeymapNotify.event") // no sequence: inherits 7
	155	+	stream = append(stream, km...)
	156	+	want = append(want, record{'e', 7, codeKeymapNotify, km})
	157	+	add('e', uint32(KindUnknown), pktEvent(80, 0, 0x11), 8) // unbound code
	158	+	add('e', uint32(ExtGenericEvent)<<8\|0x80, pktXGE(0, 0, 0), 8)
	159	+	add('r', 0, pktReply(0, 0), 9)
	160	+	sent := pktEvent(0x80\|33, 0, 0x22) // SendEvent ClientMessage
	161	+	add('e', 33, sent, 9)
	162	+	return stream, want
	163	+	}
	164	+
	165	+	func checkRecords(t testing.TB, got, want []record) {
	166	+	t.Helper()
	167	+	if len(got) != len(want) {
	168	+	t.Fatalf("%d callbacks, want %d: %v", len(got), len(want), got)
	169	+	}
	170	+	for i := range want {
	171	+	g, w := got[i], want[i]
	172	+	if g.class != w.class \|\| g.seq != w.seq \|\| g.kind != w.kind \|\| !bytes.Equal(g.data, w.data) {
	173	+	t.Fatalf("callback %d: %v, want %v", i, g, w)
	174	+	}
	175	+	}
	176	+	}
	177	+
	178	+	func TestDispatchStream(t *testing.T) {
	179	+	stream, want := scriptedStream(t)
	180	+	t.Run("one read", func(t *testing.T) {
	181	+	c, srv := dispatchConn(t)
	182	+	r := newRecorder()
	183	+	feed(t, c, srv, stream, &r.h)
	184	+	checkRecords(t, r.got, want)
	185	+	if c.buf.acked != 9 \|\| c.seq != 9 {
	186	+	t.Fatalf("acked %d seq %d", c.buf.acked, c.seq)
	187	+	}
	188	+	})
	189	+	t.Run("split at every byte", func(t *testing.T) {
	190	+	for k := range len(stream) + 1 {
	191	+	c, srv := dispatchConn(t)
	192	+	r := newRecorder()
	193	+	feed(t, c, srv, stream[:k], &r.h)
	194	+	feed(t, c, srv, stream[k:], &r.h)
	195	+	checkRecords(t, r.got, want)
	196	+	}
	197	+	})
	198	+	t.Run("byte by byte", func(t *testing.T) {
	199	+	c, srv := dispatchConn(t)
	200	+	r := newRecorder()
	201	+	for i := range stream {
	202	+	feed(t, c, srv, stream[i:i+1], &r.h)
	203	+	}
	204	+	checkRecords(t, r.got, want)
	205	+	})
	206	+	t.Run("nil handler and nil fields", func(t *testing.T) {
	207	+	c, srv := dispatchConn(t)
	208	+	feed(t, c, srv, stream, nil)
	209	+	feed(t, c, srv, stream, &Handler{})
	210	+	if c.rpos != c.rend {
	211	+	t.Fatal("stream not consumed")
	212	+	}
	213	+	})
	214	+	}
	215	+
	216	+	func TestDispatchSeqWidening(t *testing.T) {
	217	+	c, srv := dispatchConn(t)
	218	+	r := newRecorder()
	219	+	// Wire seqs climbing through two 16-bit wraps.
	220	+	wires := []uint16{1, 0x7fff, 0xfffe, 0xffff, 0, 3, 0x8000, 0xffff, 2}
	221	+	wantSeq := []Seq{1, 0x7fff, 0xfffe, 0xffff, 0x10000, 0x10003, 0x18000, 0x1ffff, 0x20002}
	222	+	var stream []byte
	223	+	for i, w := range wires {
	224	+	switch i % 3 {
	225	+	case 0:
	226	+	stream = append(stream, pktReply(w, 4)...)
	227	+	case 1:
	228	+	stream = append(stream, pktEvent(22, w, 0)...)
	229	+	default:
	230	+	stream = append(stream, pktError(3, w, 12)...)
	231	+	}
	232	+	}
	233	+	stream = append(stream, golden(t, "xproto/KeymapNotify.event")...)
	234	+	feed(t, c, srv, stream, &r.h)
	235	+	if len(r.got) != len(wires)+1 {
	236	+	t.Fatal(r.got)
	237	+	}
	238	+	for i, s := range wantSeq {
	239	+	if r.got[i].seq != s {
	240	+	t.Fatalf("packet %d: seq %#x, want %#x", i, r.got[i].seq, s)
	241	+	}
	242	+	}
	243	+	if last := r.got[len(wires)]; last.seq != 0x20002 \|\| c.buf.acked != 0x20002 {
	244	+	t.Fatalf("KeymapNotify seq %#x, acked %#x", last.seq, c.buf.acked)
	245	+	}
	246	+
	247	+	// ack never passes requests that were not sent.
	248	+	c2, srv2 := dispatchConn(t)
	249	+	c2.buf.next = 5
	250	+	feed(t, c2, srv2, pktReply(9, 0), nil)
	251	+	if c2.buf.acked != 0 \|\| c2.seq != 9 {
	252	+	t.Fatalf("acked %d seq %d", c2.buf.acked, c2.seq)
	253	+	}
	254	+	}
	255	+
	256	+	func TestDispatchGrowth(t *testing.T) {
	257	+	c, srv := dispatchConn(t)
	258	+	r := newRecorder()
	259	+	big := pktReply(1, 300<<10) // larger than the 64 KiB ring
	260	+	xge := pktXGE(2, 5, 100<<10)
	261	+	stream := append(append(append([]byte{}, pktEvent(22, 1, 0)...), big...), xge...)
	262	+	feed(t, c, srv, stream, &r.h)
	263	+	if len(r.got) != 3 \|\| !bytes.Equal(r.got[1].data, big) \|\| !bytes.Equal(r.got[2].data, xge) {
	264	+	t.Fatalf("%v", r.got)
	265	+	}
	266	+	if len(c.rb) != 512<<10 {
	267	+	t.Fatalf("ring %d bytes, want 512 KiB", len(c.rb))
	268	+	}
	269	+	}
	270	+
	271	+	func TestDispatchMaxPacket(t *testing.T) {
	272	+	t.Run("exactly MaxPacket", func(t *testing.T) {
	273	+	c, srv := dispatchConn(t)
	274	+	c.maxPacket = 256 << 10
	275	+	r := newRecorder()
	276	+	p := pktReply(1, c.maxPacket-32)
	277	+	feed(t, c, srv, p, &r.h)
	278	+	if len(r.got) != 1 \|\| len(r.got[0].data) != c.maxPacket \|\| len(c.rb) != c.maxPacket {
	279	+	t.Fatalf("%v ring %d", r.got, len(c.rb))
	280	+	}
	281	+	})
	282	+	for _, tc := range []struct {
	283	+	name string
	284	+	p []byte
	285	+	}{
	286	+	{"reply", pktReply(1, 0)},
	287	+	{"XGE", pktXGE(1, 0, 0)},
	288	+	} {
	289	+	t.Run(tc.name+" beyond MaxPacket", func(t *testing.T) {
	290	+	c, srv := dispatchConn(t)
	291	+	r := newRecorder()
	292	+	p := tc.p
	293	+	binary.LittleEndian.PutUint32(p[4:], (MaxPacket-32)/4+1)
	294	+	stream := append(pktEvent(22, 1, 0), p...)
	295	+	if _, err := unix.Write(srv, stream); err != nil {
	296	+	t.Fatal(err)
	297	+	}
	298	+	if err := c.Dispatch(&r.h); !errors.Is(err, ErrPacketTooLarge) {
	299	+	t.Fatal(err)
	300	+	}
	301	+	if len(r.got) != 1 \|\| !errors.Is(c.Dispatch(&r.h), ErrPacketTooLarge) \|\| !errors.Is(c.buf.Err(), ErrPacketTooLarge) {
	302	+	t.Fatal("not sticky")
	303	+	}
	304	+	})
	305	+	}
	306	+	t.Run("length overflowing 32 bits", func(t *testing.T) {
	307	+	if n, ok := packetSize(bytes.Repeat([]byte{1, 0, 0, 0, 0xff, 0xff, 0xff, 0xff}, 4), MaxPacket); ok {
	308	+	t.Fatal(n)
	309	+	}
	310	+	})
	311	+	}
	312	+
	313	+	func TestDispatchEOF(t *testing.T) {
	314	+	for _, tail := range [][]byte{nil, pktReply(2, 64)[:40]} {
	315	+	c, srv := dispatchConn(t)
	316	+	r := newRecorder()
	317	+	data := append(pktEvent(22, 1, 0), tail...)
	318	+	if _, err := unix.Write(srv, data); err != nil {
	319	+	t.Fatal(err)
	320	+	}
	321	+	unix.Shutdown(srv, unix.SHUT_WR)
	322	+	if err := c.Dispatch(&r.h); err != io.EOF {
	323	+	t.Fatal(err)
	324	+	}
	325	+	if len(r.got) != 1 \|\| c.Dispatch(&r.h) != io.EOF \|\| !errors.Is(c.buf.Err(), ErrClosed) {
	326	+	t.Fatal("EOF not sticky")
	327	+	}
	328	+	}
	329	+	c, _ := dispatchConn(t)
	330	+	c.Close()
	331	+	if !errors.Is(c.Dispatch(nil), ErrClosed) {
	332	+	t.Fatal("Dispatch after Close")
	333	+	}
	334	+	}
	335	+
	336	+	func TestDispatchCloseInCallback(t *testing.T) {
	337	+	c, srv := dispatchConn(t)
	338	+	n := 0
	339	+	h := Handler{OnEvent: func(*Event) { n++; c.Close() }}
	340	+	if _, err := unix.Write(srv, append(pktEvent(22, 1, 0), pktEvent(22, 2, 0)...)); err != nil {
	341	+	t.Fatal(err)
	342	+	}
	343	+	if err := c.Dispatch(&h); !errors.Is(err, ErrClosed) \|\| n != 1 {
	344	+	t.Fatal(err, n)
	345	+	}
	346	+	}
	347	+
	348	+	// recordedStream repeats the scripted stream (events, replies, errors) up to
	349	+	// about size bytes.
	350	+	func recordedStream(t testing.TB, size int) ([]byte, int) {
	351	+	unit, want := scriptedStream(t)
	352	+	var s []byte
	353	+	n := 0
	354	+	for len(s) < size {
	355	+	s = append(s, unit...)
	356	+	n += len(want)
	357	+	}
	358	+	return s, n
	359	+	}
	360	+
	361	+	func TestAllocDispatch(t *testing.T) {
	362	+	stream, packets := recordedStream(t, 1<<20)
	363	+	c, srv := dispatchConn(t)
	364	+	var events, replies, errs int
	365	+	var sink byte
	366	+	h := Handler{
	367	+	OnEvent: func(ev *Event) { events++; sink ^= ev.Raw[1]; sink ^= byte(len(ev.Tail)) },
	368	+	OnReply: func(r *Reply) { replies++; sink ^= r.Raw[len(r.Raw)-1] },
	369	+	OnError: func(e *XError) { errs++; sink ^= e.Code },
	370	+	}
	371	+	feed(t, c, srv, stream, &h) // warm the ring
	372	+	if n := testing.AllocsPerRun(5, func() { feed(t, c, srv, stream, &h) }); n != 0 {
	373	+	t.Fatalf("Dispatch allocates %v times per 1 MiB", n)
	374	+	}
	375	+	if events+replies+errs != 7*packets \|\| errs == 0 \|\| replies == 0 {
	376	+	t.Fatal(events, replies, errs)
	377	+	}
	378	+	_ = sink
	379	+	}
	380	+
	381	+	// mixedPackets is a 10k-packet stream: core events, replies, errors, XFIXES
	382	+	// and XGE events.
	383	+	func mixedPackets(t testing.TB) []byte {
	384	+	var s []byte
	385	+	for i := range 10000 {
	386	+	seq := uint16(i)
	387	+	switch i % 5 {
	388	+	case 0:
	389	+	s = append(s, pktEvent(22, seq, 0)...)
	390	+	case 1:
	391	+	s = append(s, pktReply(seq, 24)...)
	392	+	case 2:
	393	+	s = append(s, pktError(3, seq, 12)...)
	394	+	case 3:
	395	+	s = append(s, pktEvent(87, seq, 1)...)
	396	+	default:
	397	+	s = append(s, pktXGE(seq, 1, 8)...)
	398	+	}
	399	+	}
	400	+	return s
	401	+	}
	402	+
	403	+	func BenchmarkDispatch10k(b *testing.B) {
	404	+	stream := mixedPackets(b)
	405	+	c, srv := dispatchConn(b)
	406	+	var sink int
	407	+	h := Handler{
	408	+	OnEvent: func(ev *Event) { sink += int(ev.Kind) },
	409	+	OnReply: func(r *Reply) { sink += len(r.Raw) },
	410	+	OnError: func(e *XError) { sink += int(e.Code) },
	411	+	}
	412	+	feed(b, c, srv, stream, &h)
	413	+	b.SetBytes(int64(len(stream)))
	414	+	b.ReportAllocs()
	415	+	for b.Loop() {
	416	+	feed(b, c, srv, stream, &h)
	417	+	}
	418	+	_ = sink
	419	+	}
	420	+
	421	+	// FuzzDispatchStream feeds an arbitrary byte stream in arbitrary chunks.
	422	+	// Invariants: no panic; the callbacks' bytes concatenate to a prefix of the
	423	+	// stream covering every complete packet; Seq never decreases; Dispatch ends
	424	+	// with io.EOF (or ErrPacketTooLarge) after the server hangs up.
	425	+	func FuzzDispatchStream(f *testing.F) {
	426	+	stream, _ := scriptedStream(f)
	427	+	f.Add(stream, uint16(7))
	428	+	f.Add(stream, uint16(0))
	429	+	f.Add(append(pktReply(1, 1024), pktXGE(2, 1, 64)...), uint16(33))
	430	+	f.Fuzz(func(t *testing.T, stream []byte, chunk uint16) {
	431	+	if len(stream) > 1<<16 {
	432	+	return
	433	+	}
	434	+	c, srv := dispatchConn(t)
	435	+	c.maxPacket = 8 << 10
	436	+	var got []byte
	437	+	var last Seq
	438	+	mono := func(s Seq) {
	439	+	if s < last {
	440	+	t.Fatalf("seq %d after %d", s, last)
	441	+	}
	442	+	last = s
	443	+	}
	444	+	h := Handler{
	445	+	OnEvent: func(ev *Event) { mono(ev.Seq); got = append(append(got, ev.Raw[:]...), ev.Tail...) },
	446	+	OnReply: func(r *Reply) { mono(r.Seq); got = append(got, r.Raw...) },
	447	+	OnError: func(e *XError) { mono(e.Seq); got = append(got, e.Raw[:]...) },
	448	+	}
	449	+	step := int(chunk)%97 + 1
	450	+	var err error
	451	+	for p := stream; len(p) > 0 && err == nil; {
	452	+	k := min(step, len(p))
	453	+	if _, werr := unix.Write(srv, p[:k]); werr != nil {
	454	+	t.Fatal(werr)
	455	+	}
	456	+	p = p[k:]
	457	+	err = c.Dispatch(&h)
	458	+	}
	459	+	unix.Shutdown(srv, unix.SHUT_WR)
	460	+	for err == nil {
	461	+	err = c.Dispatch(&h)
	462	+	}
	463	+	if err != io.EOF && !errors.Is(err, ErrPacketTooLarge) {
	464	+	t.Fatal(err)
	465	+	}
	466	+	if !bytes.HasPrefix(stream, got) {
	467	+	t.Fatal("callbacks diverge from the stream")
	468	+	}
	469	+	if rest := stream[len(got):]; err == io.EOF && len(rest) >= 32 {
	470	+	if n, ok := packetSize(rest, c.maxPacket); ok && n <= len(rest) {
	471	+	t.Fatalf("complete packet of %d bytes left undelivered", n)
	472	+	}
	473	+	}
	474	+	})
	475	+	}

@@ -0,0 +1,475 @@
	1	+	package xturbo
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"encoding/binary"
	6	+	"errors"
	7	+	"fmt"
	8	+	"io"
	9	+	"testing"
	10	+
	11	+	"bnema.dev/xturbo/internal/xtest"
	12	+	"golang.org/x/sys/unix"
	13	+	)
	14	+
	15	+	// dispatchConn returns a Conn over one end of a non-blocking socketpair with
	16	+	// default kernel buffers, and the other end acting as the server. The XFIXES
	17	+	// and Generic Event extensions are negotiated at the golden capture's bases.
	18	+	func dispatchConn(t testing.TB) (*Conn, int) {
	19	+	t.Helper()
	20	+	fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM\|unix.SOCK_NONBLOCK\|unix.SOCK_CLOEXEC, 0)
	21	+	if err != nil {
	22	+	t.Fatal(err)
	23	+	}
	24	+	c := newConnState(fds[0], nil, setupInfo{})
	25	+	t.Cleanup(func() { _ = c.Close(); _ = unix.Close(fds[1]) })
	26	+	c.ext.Set(ExtXFixes, 138, 87, 140)
	27	+	c.ext.Set(ExtGenericEvent, 0x83, 0, 0)
	28	+	c.buf.next = 1 << 40 // every widened seq in the tests is acknowledgeable
	29	+	return c, fds[1]
	30	+	}
	31	+
	32	+	// Packet builders. seq is the 16-bit wire value.
	33	+	func pktEvent(code uint8, seq uint16, fill byte) []byte {
	34	+	p := bytes.Repeat([]byte{fill}, 32)
	35	+	p[0] = code
	36	+	binary.LittleEndian.PutUint16(p[2:], seq)
	37	+	return p
	38	+	}
	39	+
	40	+	func pktReply(seq uint16, data int) []byte {
	41	+	words := (data + 3) / 4
	42	+	p := make([]byte, 32+4*words)
	43	+	p[0], p[1] = 1, 7
	44	+	binary.LittleEndian.PutUint16(p[2:], seq)
	45	+	binary.LittleEndian.PutUint32(p[4:], uint32(words))
	46	+	for i := 8; i < len(p); i++ {
	47	+	p[i] = byte(i * 7)
	48	+	}
	49	+	return p
	50	+	}
	51	+
	52	+	func pktError(code uint8, seq uint16, major uint8) []byte {
	53	+	p := make([]byte, 32)
	54	+	p[1] = code
	55	+	binary.LittleEndian.PutUint16(p[2:], seq)
	56	+	binary.LittleEndian.PutUint32(p[4:], 0xdead)
	57	+	p[10] = major
	58	+	return p
	59	+	}
	60	+
	61	+	func pktXGE(seq uint16, evtype uint16, tail int) []byte {
	62	+	words := (tail + 3) / 4
	63	+	p := make([]byte, 32+4*words)
	64	+	p[0], p[1] = codeGenericEvent, 0x83
	65	+	binary.LittleEndian.PutUint16(p[2:], seq)
	66	+	binary.LittleEndian.PutUint32(p[4:], uint32(words))
	67	+	binary.LittleEndian.PutUint16(p[8:], evtype)
	68	+	for i := 32; i < len(p); i++ {
	69	+	p[i] = byte(i)
	70	+	}
	71	+	return p
	72	+	}
	73	+
	74	+	// record is one callback as seen by a test handler; data is a copy of the
	75	+	// packet bytes (Raw, plus Tail for events).
	76	+	type record struct {
	77	+	class byte // 'e', 'r', 'x'
	78	+	seq Seq
	79	+	kind uint32
	80	+	data []byte
	81	+	}
	82	+
	83	+	func (r record) String() string {
	84	+	return fmt.Sprintf("%c seq %d kind %#x %d bytes", r.class, r.seq, r.kind, len(r.data))
	85	+	}
	86	+
	87	+	type recorder struct {
	88	+	got []record
	89	+	h Handler
	90	+	}
	91	+
	92	+	func newRecorder() *recorder {
	93	+	r := &recorder{}
	94	+	r.h = Handler{
	95	+	OnEvent: func(ev *Event) {
	96	+	r.got = append(r.got, record{'e', ev.Seq, uint32(ev.Kind), append(bytes.Clone(ev.Raw[:]), ev.Tail...)})
	97	+	},
	98	+	OnReply: func(rp *Reply) {
	99	+	r.got = append(r.got, record{'r', rp.Seq, uint32(rp.Kind), bytes.Clone(rp.Raw)})
	100	+	},
	101	+	OnError: func(e *XError) {
	102	+	r.got = append(r.got, record{'x', e.Seq, uint32(e.Code), bytes.Clone(e.Raw[:])})
	103	+	},
	104	+	}
	105	+	return r
	106	+	}
	107	+
	108	+	// feed writes data to the server end, dispatching whenever the socket fills,
	109	+	// then dispatches what remains.
	110	+	func feed(t testing.TB, c Conn, srv int, data []byte, h Handler) {
	111	+	t.Helper()
	112	+	for len(data) > 0 {
	113	+	n, err := unix.Write(srv, data)
	114	+	if err == unix.EAGAIN {
	115	+	n = 0
	116	+	} else if err != nil {
	117	+	t.Fatal(err)
	118	+	}
	119	+	data = data[n:]
	120	+	if err := c.Dispatch(h); err != nil {
	121	+	t.Fatal(err)
	122	+	}
	123	+	}
	124	+	if err := c.Dispatch(h); err != nil {
	125	+	t.Fatal(err)
	126	+	}
	127	+	}
	128	+
	129	+	func golden(t testing.TB, name string) []byte {
	130	+	t.Helper()
	131	+	raw, err := xtest.LoadHex("testdata/golden/" + name + ".hex")
	132	+	if err != nil {
	133	+	t.Fatal(err)
	134	+	}
	135	+	return raw
	136	+	}
	137	+
	138	+	// scriptedStream is a mixed stream of goldens and synthetic packets with
	139	+	// increasing wire sequence numbers, and the callbacks it must produce.
	140	+	func scriptedStream(t testing.TB) ([]byte, []record) {
	141	+	var stream []byte
	142	+	var want []record
	143	+	add := func(class byte, kind uint32, p []byte, seq uint16) {
	144	+	binary.LittleEndian.PutUint16(p[2:], seq)
	145	+	stream = append(stream, p...)
	146	+	want = append(want, record{class, Seq(seq), kind, p})
	147	+	}
	148	+	add('e', 20, golden(t, "xproto/MapRequest.event"), 3)
	149	+	add('r', 0, golden(t, "xproto/GetGeometry.reply"), 4)
	150	+	add('x', 3, golden(t, "xproto/BadWindow.error"), 5)
	151	+	add('e', uint32(ExtGenericEvent)<<8\|0x81, golden(t, "xproto/GeGeneric.event"), 6)
	152	+	add('e', uint32(ExtXFixes)<<8, golden(t, "xfixes/SelectionNotify.event"), 6)
	153	+	add('r', 0, golden(t, "xproto/GetProperty.reply"), 7)
	154	+	km := golden(t, "xproto/KeymapNotify.event") // no sequence: inherits 7
	155	+	stream = append(stream, km...)
	156	+	want = append(want, record{'e', 7, codeKeymapNotify, km})
	157	+	add('e', uint32(KindUnknown), pktEvent(80, 0, 0x11), 8) // unbound code
	158	+	add('e', uint32(ExtGenericEvent)<<8\|0x80, pktXGE(0, 0, 0), 8)
	159	+	add('r', 0, pktReply(0, 0), 9)
	160	+	sent := pktEvent(0x80\|33, 0, 0x22) // SendEvent ClientMessage
	161	+	add('e', 33, sent, 9)
	162	+	return stream, want
	163	+	}
	164	+
	165	+	func checkRecords(t testing.TB, got, want []record) {
	166	+	t.Helper()
	167	+	if len(got) != len(want) {
	168	+	t.Fatalf("%d callbacks, want %d: %v", len(got), len(want), got)
	169	+	}
	170	+	for i := range want {
	171	+	g, w := got[i], want[i]
	172	+	if g.class != w.class \|\| g.seq != w.seq \|\| g.kind != w.kind \|\| !bytes.Equal(g.data, w.data) {
	173	+	t.Fatalf("callback %d: %v, want %v", i, g, w)
	174	+	}
	175	+	}
	176	+	}
	177	+
	178	+	func TestDispatchStream(t *testing.T) {
	179	+	stream, want := scriptedStream(t)
	180	+	t.Run("one read", func(t *testing.T) {
	181	+	c, srv := dispatchConn(t)
	182	+	r := newRecorder()
	183	+	feed(t, c, srv, stream, &r.h)
	184	+	checkRecords(t, r.got, want)
	185	+	if c.buf.acked != 9 \|\| c.seq != 9 {
	186	+	t.Fatalf("acked %d seq %d", c.buf.acked, c.seq)
	187	+	}
	188	+	})
	189	+	t.Run("split at every byte", func(t *testing.T) {
	190	+	for k := range len(stream) + 1 {
	191	+	c, srv := dispatchConn(t)
	192	+	r := newRecorder()
	193	+	feed(t, c, srv, stream[:k], &r.h)
	194	+	feed(t, c, srv, stream[k:], &r.h)
	195	+	checkRecords(t, r.got, want)
	196	+	}
	197	+	})
	198	+	t.Run("byte by byte", func(t *testing.T) {
	199	+	c, srv := dispatchConn(t)
	200	+	r := newRecorder()
	201	+	for i := range stream {
	202	+	feed(t, c, srv, stream[i:i+1], &r.h)
	203	+	}
	204	+	checkRecords(t, r.got, want)
	205	+	})
	206	+	t.Run("nil handler and nil fields", func(t *testing.T) {
	207	+	c, srv := dispatchConn(t)
	208	+	feed(t, c, srv, stream, nil)
	209	+	feed(t, c, srv, stream, &Handler{})
	210	+	if c.rpos != c.rend {
	211	+	t.Fatal("stream not consumed")
	212	+	}
	213	+	})
	214	+	}
	215	+
	216	+	func TestDispatchSeqWidening(t *testing.T) {
	217	+	c, srv := dispatchConn(t)
	218	+	r := newRecorder()
	219	+	// Wire seqs climbing through two 16-bit wraps.
	220	+	wires := []uint16{1, 0x7fff, 0xfffe, 0xffff, 0, 3, 0x8000, 0xffff, 2}
	221	+	wantSeq := []Seq{1, 0x7fff, 0xfffe, 0xffff, 0x10000, 0x10003, 0x18000, 0x1ffff, 0x20002}
	222	+	var stream []byte
	223	+	for i, w := range wires {
	224	+	switch i % 3 {
	225	+	case 0:
	226	+	stream = append(stream, pktReply(w, 4)...)
	227	+	case 1:
	228	+	stream = append(stream, pktEvent(22, w, 0)...)
	229	+	default:
	230	+	stream = append(stream, pktError(3, w, 12)...)
	231	+	}
	232	+	}
	233	+	stream = append(stream, golden(t, "xproto/KeymapNotify.event")...)
	234	+	feed(t, c, srv, stream, &r.h)
	235	+	if len(r.got) != len(wires)+1 {
	236	+	t.Fatal(r.got)
	237	+	}
	238	+	for i, s := range wantSeq {
	239	+	if r.got[i].seq != s {
	240	+	t.Fatalf("packet %d: seq %#x, want %#x", i, r.got[i].seq, s)
	241	+	}
	242	+	}
	243	+	if last := r.got[len(wires)]; last.seq != 0x20002 \|\| c.buf.acked != 0x20002 {
	244	+	t.Fatalf("KeymapNotify seq %#x, acked %#x", last.seq, c.buf.acked)
	245	+	}
	246	+
	247	+	// ack never passes requests that were not sent.
	248	+	c2, srv2 := dispatchConn(t)
	249	+	c2.buf.next = 5
	250	+	feed(t, c2, srv2, pktReply(9, 0), nil)
	251	+	if c2.buf.acked != 0 \|\| c2.seq != 9 {
	252	+	t.Fatalf("acked %d seq %d", c2.buf.acked, c2.seq)
	253	+	}
	254	+	}
	255	+
	256	+	func TestDispatchGrowth(t *testing.T) {
	257	+	c, srv := dispatchConn(t)
	258	+	r := newRecorder()
	259	+	big := pktReply(1, 300<<10) // larger than the 64 KiB ring
	260	+	xge := pktXGE(2, 5, 100<<10)
	261	+	stream := append(append(append([]byte{}, pktEvent(22, 1, 0)...), big...), xge...)
	262	+	feed(t, c, srv, stream, &r.h)
	263	+	if len(r.got) != 3 \|\| !bytes.Equal(r.got[1].data, big) \|\| !bytes.Equal(r.got[2].data, xge) {
	264	+	t.Fatalf("%v", r.got)
	265	+	}
	266	+	if len(c.rb) != 512<<10 {
	267	+	t.Fatalf("ring %d bytes, want 512 KiB", len(c.rb))
	268	+	}
	269	+	}
	270	+
	271	+	func TestDispatchMaxPacket(t *testing.T) {
	272	+	t.Run("exactly MaxPacket", func(t *testing.T) {
	273	+	c, srv := dispatchConn(t)
	274	+	c.maxPacket = 256 << 10
	275	+	r := newRecorder()
	276	+	p := pktReply(1, c.maxPacket-32)
	277	+	feed(t, c, srv, p, &r.h)
	278	+	if len(r.got) != 1 \|\| len(r.got[0].data) != c.maxPacket \|\| len(c.rb) != c.maxPacket {
	279	+	t.Fatalf("%v ring %d", r.got, len(c.rb))
	280	+	}
	281	+	})
	282	+	for _, tc := range []struct {
	283	+	name string
	284	+	p []byte
	285	+	}{
	286	+	{"reply", pktReply(1, 0)},
	287	+	{"XGE", pktXGE(1, 0, 0)},
	288	+	} {
	289	+	t.Run(tc.name+" beyond MaxPacket", func(t *testing.T) {
	290	+	c, srv := dispatchConn(t)
	291	+	r := newRecorder()
	292	+	p := tc.p
	293	+	binary.LittleEndian.PutUint32(p[4:], (MaxPacket-32)/4+1)
	294	+	stream := append(pktEvent(22, 1, 0), p...)
	295	+	if _, err := unix.Write(srv, stream); err != nil {
	296	+	t.Fatal(err)
	297	+	}
	298	+	if err := c.Dispatch(&r.h); !errors.Is(err, ErrPacketTooLarge) {
	299	+	t.Fatal(err)
	300	+	}
	301	+	if len(r.got) != 1 \|\| !errors.Is(c.Dispatch(&r.h), ErrPacketTooLarge) \|\| !errors.Is(c.buf.Err(), ErrPacketTooLarge) {
	302	+	t.Fatal("not sticky")
	303	+	}
	304	+	})
	305	+	}
	306	+	t.Run("length overflowing 32 bits", func(t *testing.T) {
	307	+	if n, ok := packetSize(bytes.Repeat([]byte{1, 0, 0, 0, 0xff, 0xff, 0xff, 0xff}, 4), MaxPacket); ok {
	308	+	t.Fatal(n)
	309	+	}
	310	+	})
	311	+	}
	312	+
	313	+	func TestDispatchEOF(t *testing.T) {
	314	+	for _, tail := range [][]byte{nil, pktReply(2, 64)[:40]} {
	315	+	c, srv := dispatchConn(t)
	316	+	r := newRecorder()
	317	+	data := append(pktEvent(22, 1, 0), tail...)
	318	+	if _, err := unix.Write(srv, data); err != nil {
	319	+	t.Fatal(err)
	320	+	}
	321	+	unix.Shutdown(srv, unix.SHUT_WR)
	322	+	if err := c.Dispatch(&r.h); err != io.EOF {
	323	+	t.Fatal(err)
	324	+	}
	325	+	if len(r.got) != 1 \|\| c.Dispatch(&r.h) != io.EOF \|\| !errors.Is(c.buf.Err(), ErrClosed) {
	326	+	t.Fatal("EOF not sticky")
	327	+	}
	328	+	}
	329	+	c, _ := dispatchConn(t)
	330	+	c.Close()
	331	+	if !errors.Is(c.Dispatch(nil), ErrClosed) {
	332	+	t.Fatal("Dispatch after Close")
	333	+	}
	334	+	}
	335	+
	336	+	func TestDispatchCloseInCallback(t *testing.T) {
	337	+	c, srv := dispatchConn(t)
	338	+	n := 0
	339	+	h := Handler{OnEvent: func(*Event) { n++; c.Close() }}
	340	+	if _, err := unix.Write(srv, append(pktEvent(22, 1, 0), pktEvent(22, 2, 0)...)); err != nil {
	341	+	t.Fatal(err)
	342	+	}
	343	+	if err := c.Dispatch(&h); !errors.Is(err, ErrClosed) \|\| n != 1 {
	344	+	t.Fatal(err, n)
	345	+	}
	346	+	}
	347	+
	348	+	// recordedStream repeats the scripted stream (events, replies, errors) up to
	349	+	// about size bytes.
	350	+	func recordedStream(t testing.TB, size int) ([]byte, int) {
	351	+	unit, want := scriptedStream(t)
	352	+	var s []byte
	353	+	n := 0
	354	+	for len(s) < size {
	355	+	s = append(s, unit...)
	356	+	n += len(want)
	357	+	}
	358	+	return s, n
	359	+	}
	360	+
	361	+	func TestAllocDispatch(t *testing.T) {
	362	+	stream, packets := recordedStream(t, 1<<20)
	363	+	c, srv := dispatchConn(t)
	364	+	var events, replies, errs int
	365	+	var sink byte
	366	+	h := Handler{
	367	+	OnEvent: func(ev *Event) { events++; sink ^= ev.Raw[1]; sink ^= byte(len(ev.Tail)) },
	368	+	OnReply: func(r *Reply) { replies++; sink ^= r.Raw[len(r.Raw)-1] },
	369	+	OnError: func(e *XError) { errs++; sink ^= e.Code },
	370	+	}
	371	+	feed(t, c, srv, stream, &h) // warm the ring
	372	+	if n := testing.AllocsPerRun(5, func() { feed(t, c, srv, stream, &h) }); n != 0 {
	373	+	t.Fatalf("Dispatch allocates %v times per 1 MiB", n)
	374	+	}
	375	+	if events+replies+errs != 7*packets \|\| errs == 0 \|\| replies == 0 {
	376	+	t.Fatal(events, replies, errs)
	377	+	}
	378	+	_ = sink
	379	+	}
	380	+
	381	+	// mixedPackets is a 10k-packet stream: core events, replies, errors, XFIXES
	382	+	// and XGE events.
	383	+	func mixedPackets(t testing.TB) []byte {
	384	+	var s []byte
	385	+	for i := range 10000 {
	386	+	seq := uint16(i)
	387	+	switch i % 5 {
	388	+	case 0:
	389	+	s = append(s, pktEvent(22, seq, 0)...)
	390	+	case 1:
	391	+	s = append(s, pktReply(seq, 24)...)
	392	+	case 2:
	393	+	s = append(s, pktError(3, seq, 12)...)
	394	+	case 3:
	395	+	s = append(s, pktEvent(87, seq, 1)...)
	396	+	default:
	397	+	s = append(s, pktXGE(seq, 1, 8)...)
	398	+	}
	399	+	}
	400	+	return s
	401	+	}
	402	+
	403	+	func BenchmarkDispatch10k(b *testing.B) {
	404	+	stream := mixedPackets(b)
	405	+	c, srv := dispatchConn(b)
	406	+	var sink int
	407	+	h := Handler{
	408	+	OnEvent: func(ev *Event) { sink += int(ev.Kind) },
	409	+	OnReply: func(r *Reply) { sink += len(r.Raw) },
	410	+	OnError: func(e *XError) { sink += int(e.Code) },
	411	+	}
	412	+	feed(b, c, srv, stream, &h)
	413	+	b.SetBytes(int64(len(stream)))
	414	+	b.ReportAllocs()
	415	+	for b.Loop() {
	416	+	feed(b, c, srv, stream, &h)
	417	+	}
	418	+	_ = sink
	419	+	}
	420	+
	421	+	// FuzzDispatchStream feeds an arbitrary byte stream in arbitrary chunks.
	422	+	// Invariants: no panic; the callbacks' bytes concatenate to a prefix of the
	423	+	// stream covering every complete packet; Seq never decreases; Dispatch ends
	424	+	// with io.EOF (or ErrPacketTooLarge) after the server hangs up.
	425	+	func FuzzDispatchStream(f *testing.F) {
	426	+	stream, _ := scriptedStream(f)
	427	+	f.Add(stream, uint16(7))
	428	+	f.Add(stream, uint16(0))
	429	+	f.Add(append(pktReply(1, 1024), pktXGE(2, 1, 64)...), uint16(33))
	430	+	f.Fuzz(func(t *testing.T, stream []byte, chunk uint16) {
	431	+	if len(stream) > 1<<16 {
	432	+	return
	433	+	}
	434	+	c, srv := dispatchConn(t)
	435	+	c.maxPacket = 8 << 10
	436	+	var got []byte
	437	+	var last Seq
	438	+	mono := func(s Seq) {
	439	+	if s < last {
	440	+	t.Fatalf("seq %d after %d", s, last)
	441	+	}
	442	+	last = s
	443	+	}
	444	+	h := Handler{
	445	+	OnEvent: func(ev *Event) { mono(ev.Seq); got = append(append(got, ev.Raw[:]...), ev.Tail...) },
	446	+	OnReply: func(r *Reply) { mono(r.Seq); got = append(got, r.Raw...) },
	447	+	OnError: func(e *XError) { mono(e.Seq); got = append(got, e.Raw[:]...) },
	448	+	}
	449	+	step := int(chunk)%97 + 1
	450	+	var err error
	451	+	for p := stream; len(p) > 0 && err == nil; {
	452	+	k := min(step, len(p))
	453	+	if _, werr := unix.Write(srv, p[:k]); werr != nil {
	454	+	t.Fatal(werr)
	455	+	}
	456	+	p = p[k:]
	457	+	err = c.Dispatch(&h)
	458	+	}
	459	+	unix.Shutdown(srv, unix.SHUT_WR)
	460	+	for err == nil {
	461	+	err = c.Dispatch(&h)
	462	+	}
	463	+	if err != io.EOF && !errors.Is(err, ErrPacketTooLarge) {
	464	+	t.Fatal(err)
	465	+	}
	466	+	if !bytes.HasPrefix(stream, got) {
	467	+	t.Fatal("callbacks diverge from the stream")
	468	+	}
	469	+	if rest := stream[len(got):]; err == io.EOF && len(rest) >= 32 {
	470	+	if n, ok := packetSize(rest, c.maxPacket); ok && n <= len(rest) {
	471	+	t.Fatalf("complete packet of %d bytes left undelivered", n)
	472	+	}
	473	+	}
	474	+	})
	475	+	}