feat(runtime): attribute replies and errors through an Expect ring with connection stats

commit: 350da1981575fb3116e0677432801235ab714309
author: brice · 14h ago · 2026-10-22 08:45 UTC
committer: brice · 13h ago · 2026-10-22 09:36 UTC
parent: 8a907e5
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

7 files · +507 -42

M conn.go +5 -0


@@ -24,6 +24,11 @@ type Conn struct {
24	24		rerr error // sticky read failure
25	25		seq Seq // last widened sequence number seen
26	26
	27	+	// Reply ring: Expect registrations indexed by seq&ringMask.
	28	+	ring [ringSize]slot
	29	+	live int // registered slots
	30	+	stats Stats
	31	+
27	32		// Callback arguments, reused per packet so nothing escapes.
28	33		ev Event
29	34		reply Reply


@@ -24,6 +24,11 @@ type Conn struct {
24	rerr error // sticky read failure	24		rerr error // sticky read failure
25	seq Seq // last widened sequence number seen	25		seq Seq // last widened sequence number seen
26		26
		27	+	// Reply ring: Expect registrations indexed by seq&ringMask.
		28	+	ring [ringSize]slot
		29	+	live int // registered slots
		30	+	stats Stats
		31	+
27	// Callback arguments, reused per packet so nothing escapes.	32		// Callback arguments, reused per packet so nothing escapes.
28	ev Event	33		ev Event
29	reply Reply	34		reply Reply

M conn_xserver_test.go +28 -9


@@ -120,20 +120,39 @@ func TestXserverDial(t *testing.T) {
120	120		t.Fatalf("Root %#x, setup view says otherwise", c.Root())
121	121		}
122	122
123		-	// A round trip through Dispatch: GetInputFocus, then its reply.
	123	+	// Round trips through Dispatch: GetInputFocus's reply, then the error
	124	+	// of GetGeometry on window None, both attributed through Expect.
	125	+	const kindFocus, kindGeom = 16, 17
124	126		seq := xproto.GetInputFocus(c.Buf())
125		-	if seq != 1 {
126		-	t.Fatalf("seq %d", seq)
	127	+	bad := xproto.GetGeometry(c.Buf(), xproto.Drawable(xproto.WindowNone))
	128	+	if seq != 1 \|\| bad != 2 {
	129	+	t.Fatalf("seqs %d %d", seq, bad)
	130	+	}
	131	+	if err := c.Expect(seq, kindFocus, 1, 2, 3, 4); err != nil {
	132	+	t.Fatal(err)
	133	+	}
	134	+	if err := c.Expect(bad, kindGeom, 5, 0, 0, 0); err != nil {
	135	+	t.Fatal(err)
127	136		}
128	137		if err := c.Flush(); err != nil {
129	138		t.Fatal(err)
130	139		}
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)
	140	+	var reply xturbo.Reply
	141	+	var xerr xturbo.XError
	142	+	h := xturbo.Handler{
	143	+	OnReply: func(r xturbo.Reply) { reply = r; reply.Raw = nil },
	144	+	OnError: func(e xturbo.XError) { xerr = e },
	145	+	}
	146	+	dispatchUntil(t, c, &h, func() bool { return xerr.Seq != 0 })
	147	+	if reply.Seq != seq \|\| reply.Kind != kindFocus \|\| reply.Args != [4]uint32{1, 2, 3, 4} {
	148	+	t.Fatalf("reply %+v", reply)
	149	+	}
	150	+	if xerr.Seq != bad \|\| xerr.Kind != kindGeom \|\| xerr.Args[0] != 5 \|\| xerr.Code != xproto.BadDrawable \|\|
	151	+	xerr.Major != 14 \|\| xerr.Ext != xturbo.ExtCore \|\| xerr.Bad != 0 {
	152	+	t.Fatalf("error %+v", xerr)
	153	+	}
	154	+	if st := c.Stats(); st.DroppedReplies != 0 \|\| st.UnhandledErrors != 0 {
	155	+	t.Fatalf("%+v", st)
137	156		}
138	157		}
139	158

 		t.Fatalf("Root %#x, setup view says otherwise", c.Root())
 	}
-	// A round trip through Dispatch: GetInputFocus, then its reply.
+	// Round trips through Dispatch: GetInputFocus's reply, then the error
+	// of GetGeometry on window None, both attributed through Expect.
+	const kindFocus, kindGeom = 16, 17
 	seq := xproto.GetInputFocus(c.Buf())
-	if seq != 1 {
+	bad := xproto.GetGeometry(c.Buf(), xproto.Drawable(xproto.WindowNone))
-		t.Fatalf("seq %d", seq)
+	if seq != 1 || bad != 2 {
+		t.Fatalf("seqs %d %d", seq, bad)
+	}
+	if err := c.Expect(seq, kindFocus, 1, 2, 3, 4); err != nil {
+		t.Fatal(err)
+	}
+	if err := c.Expect(bad, kindGeom, 5, 0, 0, 0); err != nil {
+		t.Fatal(err)
 	}
 	if err := c.Flush(); err != nil {
 		t.Fatal(err)
 	}
-	var got xturbo.Seq
+	var reply xturbo.Reply
-	var replyLen int
+	var xerr xturbo.XError
-	h := xturbo.Handler{OnReply: func(r *xturbo.Reply) { got, replyLen = r.Seq, len(r.Raw) }}
+	h := xturbo.Handler{
-	dispatchUntil(t, c, &h, func() bool { return got != 0 })
+		OnReply: func(r *xturbo.Reply) { reply = *r; reply.Raw = nil },
-	if got != seq || replyLen != 32 {
+		OnError: func(e *xturbo.XError) { xerr = *e },
-		t.Fatalf("reply seq %d, %d bytes", got, replyLen)
+	}
+	dispatchUntil(t, c, &h, func() bool { return xerr.Seq != 0 })
+	if reply.Seq != seq || reply.Kind != kindFocus || reply.Args != [4]uint32{1, 2, 3, 4} {
+		t.Fatalf("reply %+v", reply)
+	}
+	if xerr.Seq != bad || xerr.Kind != kindGeom || xerr.Args[0] != 5 || xerr.Code != xproto.BadDrawable ||
+		xerr.Major != 14 || xerr.Ext != xturbo.ExtCore || xerr.Bad != 0 {
+		t.Fatalf("error %+v", xerr)
+	}
+	if st := c.Stats(); st.DroppedReplies != 0 || st.UnhandledErrors != 0 {
+		t.Fatalf("%+v", st)
 	}
 }

M dispatch.go +26 -8


@@ -23,7 +23,9 @@ const (
23	23		)
24	24
25	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
	26	+	// goroutine calling Dispatch. OnReply sees only replies registered with
	27	+	// Expect; OnError sees every error (Kind 0 when unregistered). A nil field
	28	+	// drops that packet class (nil OnError counts Stats.UnhandledErrors). Fill it
27	29		// once at init; the pointers passed to callbacks are owned by the Conn and are
28	30		// reused for the next packet, so callbacks must not retain them, nor the
29	31		// views they hold (Reply.Raw, Event.Tail). Callbacks may queue requests and

@@ -34,8 +36,8 @@ type Handler struct {
34	36		OnError func(e *XError)
35	37		}
36	38
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	+	// Reply is one reply packet to a request registered with Expect. Seq is
	40	+	// widened; Kind and Args are the registration's. Raw is the whole reply,
39	41		// 32+4*length bytes, valid only until the callback returns.
40	42		type Reply struct {
41	43		Seq Seq

@@ -168,12 +170,22 @@ func (c Conn) packet(h Handler, p []byte) {
168	170		case 0:
169	171		decodeError(p, &c.ext, &c.xerr)
170	172		c.xerr.Seq = c.widen(c.xerr.Seq)
	173	+	c.xerr.Kind, c.xerr.Args, _ = c.take(c.xerr.Seq)
171	174		if h.OnError != nil {
172	175		h.OnError(&c.xerr)
	176	+	} else {
	177	+	c.stats.UnhandledErrors++
173	178		}
174	179		case 1:
175		-	c.reply = Reply{Seq: c.widen(Seq(binary.LittleEndian.Uint16(p[2:]))), Raw: p}
176		-	if h.OnReply != nil {
	180	+	seq := c.widen(Seq(binary.LittleEndian.Uint16(p[2:])))
	181	+	kind, args, ok := c.take(seq)
	182	+	c.reply = Reply{Seq: seq, Kind: kind, Args: args, Raw: p}
	183	+	switch {
	184	+	case !ok:
	185	+	c.stats.DroppedReplies++
	186	+	case kind < kindUser:
	187	+	c.consumeReply(&c.reply)
	188	+	case h.OnReply != nil:
177	189		h.OnReply(&c.reply)
178	190		}
179	191		default:

@@ -187,6 +199,9 @@ func (c Conn) packet(h Handler, p []byte) {
187	199		} else {
188	200		ev.Seq = c.widen(ev.Seq)
189	201		}
	202	+	if ev.Kind == KindUnknown {
	203	+	c.stats.UnknownEvents++
	204	+	}
190	205		c.ev = ev
191	206		if h.OnEvent != nil {
192	207		h.OnEvent(&c.ev)

@@ -195,13 +210,16 @@ func (c Conn) packet(h Handler, p []byte) {
195	210		}
196	211
197	212		// widen turns a 16-bit wire sequence into the full Seq, relative to the last
198		-	// one seen, and acknowledges it to the Buf.
	213	+	// one seen, acknowledges it to the Buf and retires the Expect slots of the
	214	+	// requests it proves complete.
199	215		func (c *Conn) widen(wire Seq) Seq {
200		-	full := c.seq&^0xffff \| wire&0xffff
201		-	if full < c.seq {
	216	+	prev := c.seq
	217	+	full := prev&^0xffff \| wire&0xffff
	218	+	if full < prev {
202	219		full += 0x10000
203	220		}
204	221		c.seq = full
205	222		c.buf.ack(full)
	223	+	c.retire(prev, full)
206	224		return full
207	225		}

 )
 // Handler receives decoded packets from Dispatch, in wire order, on the
-// goroutine calling Dispatch. A nil field drops that packet class. Fill it
+// goroutine calling Dispatch. OnReply sees only replies registered with
+// Expect; OnError sees every error (Kind 0 when unregistered). A nil field
+// drops that packet class (nil OnError counts Stats.UnhandledErrors). Fill it
 // once at init; the pointers passed to callbacks are owned by the Conn and are
 // reused for the next packet, so callbacks must not retain them, nor the
 // views they hold (Reply.Raw, Event.Tail). Callbacks may queue requests and
 	OnError func(e *XError)
 }
-// Reply is one reply packet. Seq is widened; Kind and Args come from the
+// Reply is one reply packet to a request registered with Expect. Seq is
-// Expect slot of Seq (Kind 0 when none matched). Raw is the whole reply,
+// widened; Kind and Args are the registration's. Raw is the whole reply,
 // 32+4*length bytes, valid only until the callback returns.
 type Reply struct {
 	Seq  Seq
 	case 0:
 		decodeError(p, &c.ext, &c.xerr)
 		c.xerr.Seq = c.widen(c.xerr.Seq)
+		c.xerr.Kind, c.xerr.Args, _ = c.take(c.xerr.Seq)
 		if h.OnError != nil {
 			h.OnError(&c.xerr)
+		} else {
+			c.stats.UnhandledErrors++
 		}
 	case 1:
-		c.reply = Reply{Seq: c.widen(Seq(binary.LittleEndian.Uint16(p[2:]))), Raw: p}
+		seq := c.widen(Seq(binary.LittleEndian.Uint16(p[2:])))
-		if h.OnReply != nil {
+		kind, args, ok := c.take(seq)
+		c.reply = Reply{Seq: seq, Kind: kind, Args: args, Raw: p}
+		switch {
+		case !ok:
+			c.stats.DroppedReplies++
+		case kind < kindUser:
+			c.consumeReply(&c.reply)
+		case h.OnReply != nil:
 			h.OnReply(&c.reply)
 		}
 	default:
 		} else {
 			ev.Seq = c.widen(ev.Seq)
 		}
+		if ev.Kind == KindUnknown {
+			c.stats.UnknownEvents++
+		}
 		c.ev = ev
 		if h.OnEvent != nil {
 			h.OnEvent(&c.ev)
 }
 // widen turns a 16-bit wire sequence into the full Seq, relative to the last
-// one seen, and acknowledges it to the Buf.
+// one seen, acknowledges it to the Buf and retires the Expect slots of the
+// requests it proves complete.
 func (c *Conn) widen(wire Seq) Seq {
-	full := c.seq&^0xffff | wire&0xffff
+	prev := c.seq
-	if full < c.seq {
+	full := prev&^0xffff | wire&0xffff
+	if full < prev {
 		full += 0x10000
 	}
 	c.seq = full
 	c.buf.ack(full)
+	c.retire(prev, full)
 	return full
 }

M dispatch_poison_test.go +1 -0


@@ -14,6 +14,7 @@ func TestPoisonRetainedViews(t *testing.T) {
14	14		OnEvent: func(ev *Event) { tail = ev.Tail },
15	15		}
16	16		stream := append(pktReply(1, 64), pktXGE(2, 1, 64)...)
	17	+	expect(t, c, 1, 3, 4)
17	18		feed(t, c, srv, stream, &h)
18	19		if len(raw) != 96 \|\| len(tail) != 64 {
19	20		t.Fatal(len(raw), len(tail))


@@ -14,6 +14,7 @@ func TestPoisonRetainedViews(t *testing.T) {
14	OnEvent: func(ev *Event) { tail = ev.Tail },	14		OnEvent: func(ev *Event) { tail = ev.Tail },
15	}	15		}
16	stream := append(pktReply(1, 64), pktXGE(2, 1, 64)...)	16		stream := append(pktReply(1, 64), pktXGE(2, 1, 64)...)
		17	+	expect(t, c, 1, 3, 4)
17	feed(t, c, srv, stream, &h)	18		feed(t, c, srv, stream, &h)
18	if len(raw) != 96 \|\| len(tail) != 64 {	19		if len(raw) != 96 \|\| len(tail) != 64 {
19	t.Fatal(len(raw), len(tail))	20		t.Fatal(len(raw), len(tail))

M dispatch_test.go +77 -25


@@ -89,10 +89,14 @@ type recorder struct {
89	89		h Handler
90	90		}
91	91
92		-	func newRecorder() *recorder {
	92	+	// newRecorder records every callback of c. Replies reach OnReply only when
	93	+	// registered: on each MapRequest event (the first packet of scriptedStream)
	94	+	// it registers the stream's replies, at ev.Seq+1, +4 and +6.
	95	+	func newRecorder(c Conn) recorder {
93	96		r := &recorder{}
94	97		r.h = Handler{
95	98		OnEvent: func(ev *Event) {
	99	+	expectScripted(c, ev)
96	100		r.got = append(r.got, record{'e', ev.Seq, uint32(ev.Kind), append(bytes.Clone(ev.Raw[:]), ev.Tail...)})
97	101		},
98	102		OnReply: func(rp *Reply) {

@@ -146,22 +150,45 @@ func scriptedStream(t testing.TB) ([]byte, []record) {
146	150		want = append(want, record{class, Seq(seq), kind, p})
147	151		}
148	152		add('e', 20, golden(t, "xproto/MapRequest.event"), 3)
149		-	add('r', 0, golden(t, "xproto/GetGeometry.reply"), 4)
	153	+	add('r', uint32(kindUser), golden(t, "xproto/GetGeometry.reply"), 4)
150	154		add('x', 3, golden(t, "xproto/BadWindow.error"), 5)
151	155		add('e', uint32(ExtGenericEvent)<<8\|0x81, golden(t, "xproto/GeGeneric.event"), 6)
152	156		add('e', uint32(ExtXFixes)<<8, golden(t, "xfixes/SelectionNotify.event"), 6)
153		-	add('r', 0, golden(t, "xproto/GetProperty.reply"), 7)
	157	+	add('r', uint32(kindUser), golden(t, "xproto/GetProperty.reply"), 7)
154	158		km := golden(t, "xproto/KeymapNotify.event") // no sequence: inherits 7
155	159		stream = append(stream, km...)
156	160		want = append(want, record{'e', 7, codeKeymapNotify, km})
157	161		add('e', uint32(KindUnknown), pktEvent(80, 0, 0x11), 8) // unbound code
158	162		add('e', uint32(ExtGenericEvent)<<8\|0x80, pktXGE(0, 0, 0), 8)
159		-	add('r', 0, pktReply(0, 0), 9)
	163	+	add('r', uint32(kindUser), pktReply(0, 0), 9)
160	164		sent := pktEvent(0x80\|33, 0, 0x22) // SendEvent ClientMessage
161	165		add('e', 33, sent, 9)
162	166		return stream, want
163	167		}
164	168
	169	+	// expectScripted registers the replies of scriptedStream when its first
	170	+	// event arrives. Allocation-free, so the alloc tests use it too.
	171	+	func expectScripted(c Conn, ev Event) {
	172	+	if ev.Kind != 20 {
	173	+	return
	174	+	}
	175	+	for _, d := range [...]Seq{1, 4, 6} {
	176	+	if err := c.Expect(ev.Seq+d, kindUser, uint32(d), 0, 0, 0); err != nil {
	177	+	panic(err)
	178	+	}
	179	+	}
	180	+	}
	181	+
	182	+	// expect registers seqs with kind kindUser.
	183	+	func expect(t testing.TB, c *Conn, seqs ...Seq) {
	184	+	t.Helper()
	185	+	for _, s := range seqs {
	186	+	if err := c.Expect(s, kindUser, 0, 0, 0, 0); err != nil {
	187	+	t.Fatal(err)
	188	+	}
	189	+	}
	190	+	}
	191	+
165	192		func checkRecords(t testing.TB, got, want []record) {
166	193		t.Helper()
167	194		if len(got) != len(want) {

@@ -179,17 +206,20 @@ func TestDispatchStream(t *testing.T) {
179	206		stream, want := scriptedStream(t)
180	207		t.Run("one read", func(t *testing.T) {
181	208		c, srv := dispatchConn(t)
182		-	r := newRecorder()
	209	+	r := newRecorder(c)
183	210		feed(t, c, srv, stream, &r.h)
184	211		checkRecords(t, r.got, want)
185	212		if c.buf.acked != 9 \|\| c.seq != 9 {
186	213		t.Fatalf("acked %d seq %d", c.buf.acked, c.seq)
187	214		}
	215	+	if st := c.Stats(); st.UnknownEvents != 1 \|\| st.DroppedReplies != 0 \|\| st.UnhandledErrors != 0 \|\| st.MaxRingUse != 3 {
	216	+	t.Fatalf("%+v", st)
	217	+	}
188	218		})
189	219		t.Run("split at every byte", func(t *testing.T) {
190	220		for k := range len(stream) + 1 {
191	221		c, srv := dispatchConn(t)
192		-	r := newRecorder()
	222	+	r := newRecorder(c)
193	223		feed(t, c, srv, stream[:k], &r.h)
194	224		feed(t, c, srv, stream[k:], &r.h)
195	225		checkRecords(t, r.got, want)

@@ -197,7 +227,7 @@ func TestDispatchStream(t *testing.T) {
197	227		})
198	228		t.Run("byte by byte", func(t *testing.T) {
199	229		c, srv := dispatchConn(t)
200		-	r := newRecorder()
	230	+	r := newRecorder(c)
201	231		for i := range stream {
202	232		feed(t, c, srv, stream[i:i+1], &r.h)
203	233		}

@@ -215,7 +245,7 @@ func TestDispatchStream(t *testing.T) {
215	245
216	246		func TestDispatchSeqWidening(t *testing.T) {
217	247		c, srv := dispatchConn(t)
218		-	r := newRecorder()
	248	+	r := newRecorder(c)
219	249		// Wire seqs climbing through two 16-bit wraps.
220	250		wires := []uint16{1, 0x7fff, 0xfffe, 0xffff, 0, 3, 0x8000, 0xffff, 2}
221	251		wantSeq := []Seq{1, 0x7fff, 0xfffe, 0xffff, 0x10000, 0x10003, 0x18000, 0x1ffff, 0x20002}

@@ -231,6 +261,7 @@ func TestDispatchSeqWidening(t *testing.T) {
231	261		}
232	262		}
233	263		stream = append(stream, golden(t, "xproto/KeymapNotify.event")...)
	264	+	expect(t, c, 1, 0xffff, 0x18000)
234	265		feed(t, c, srv, stream, &r.h)
235	266		if len(r.got) != len(wires)+1 {
236	267		t.Fatal(r.got)

@@ -247,6 +278,7 @@ func TestDispatchSeqWidening(t *testing.T) {
247	278		// ack never passes requests that were not sent.
248	279		c2, srv2 := dispatchConn(t)
249	280		c2.buf.next = 5
	281	+	expect(t, c2, 9)
250	282		feed(t, c2, srv2, pktReply(9, 0), nil)
251	283		if c2.buf.acked != 0 \|\| c2.seq != 9 {
252	284		t.Fatalf("acked %d seq %d", c2.buf.acked, c2.seq)

@@ -255,10 +287,11 @@ func TestDispatchSeqWidening(t *testing.T) {
255	287
256	288		func TestDispatchGrowth(t *testing.T) {
257	289		c, srv := dispatchConn(t)
258		-	r := newRecorder()
	290	+	r := newRecorder(c)
259	291		big := pktReply(1, 300<<10) // larger than the 64 KiB ring
260	292		xge := pktXGE(2, 5, 100<<10)
261	293		stream := append(append(append([]byte{}, pktEvent(22, 1, 0)...), big...), xge...)
	294	+	expect(t, c, 1)
262	295		feed(t, c, srv, stream, &r.h)
263	296		if len(r.got) != 3 \|\| !bytes.Equal(r.got[1].data, big) \|\| !bytes.Equal(r.got[2].data, xge) {
264	297		t.Fatalf("%v", r.got)

@@ -272,8 +305,9 @@ func TestDispatchMaxPacket(t *testing.T) {
272	305		t.Run("exactly MaxPacket", func(t *testing.T) {
273	306		c, srv := dispatchConn(t)
274	307		c.maxPacket = 256 << 10
275		-	r := newRecorder()
	308	+	r := newRecorder(c)
276	309		p := pktReply(1, c.maxPacket-32)
	310	+	expect(t, c, 1)
277	311		feed(t, c, srv, p, &r.h)
278	312		if len(r.got) != 1 \|\| len(r.got[0].data) != c.maxPacket \|\| len(c.rb) != c.maxPacket {
279	313		t.Fatalf("%v ring %d", r.got, len(c.rb))

@@ -288,7 +322,7 @@ func TestDispatchMaxPacket(t *testing.T) {
288	322		} {
289	323		t.Run(tc.name+" beyond MaxPacket", func(t *testing.T) {
290	324		c, srv := dispatchConn(t)
291		-	r := newRecorder()
	325	+	r := newRecorder(c)
292	326		p := tc.p
293	327		binary.LittleEndian.PutUint32(p[4:], (MaxPacket-32)/4+1)
294	328		stream := append(pktEvent(22, 1, 0), p...)

@@ -313,7 +347,7 @@ func TestDispatchMaxPacket(t *testing.T) {
313	347		func TestDispatchEOF(t *testing.T) {
314	348		for _, tail := range [][]byte{nil, pktReply(2, 64)[:40]} {
315	349		c, srv := dispatchConn(t)
316		-	r := newRecorder()
	350	+	r := newRecorder(c)
317	351		data := append(pktEvent(22, 1, 0), tail...)
318	352		if _, err := unix.Write(srv, data); err != nil {
319	353		t.Fatal(err)

@@ -364,7 +398,7 @@ func TestAllocDispatch(t *testing.T) {
364	398		var events, replies, errs int
365	399		var sink byte
366	400		h := Handler{
367		-	OnEvent: func(ev *Event) { events++; sink ^= ev.Raw[1]; sink ^= byte(len(ev.Tail)) },
	401	+	OnEvent: func(ev *Event) { expectScripted(c, ev); events++; sink ^= ev.Raw[1]; sink ^= byte(len(ev.Tail)) },
368	402		OnReply: func(r *Reply) { replies++; sink ^= r.Raw[len(r.Raw)-1] },
369	403		OnError: func(e *XError) { errs++; sink ^= e.Code },
370	404		}

@@ -372,8 +406,8 @@ func TestAllocDispatch(t *testing.T) {
372	406		if n := testing.AllocsPerRun(5, func() { feed(t, c, srv, stream, &h) }); n != 0 {
373	407		t.Fatalf("Dispatch allocates %v times per 1 MiB", n)
374	408		}
375		-	if events+replies+errs != 7*packets \|\| errs == 0 \|\| replies == 0 {
376		-	t.Fatal(events, replies, errs)
	409	+	if events+replies+errs != 7*packets \|\| errs == 0 \|\| replies == 0 \|\| c.Stats().DroppedReplies != 0 {
	410	+	t.Fatal(events, replies, errs, c.Stats())
377	411		}
378	412		_ = sink
379	413		}

@@ -405,7 +439,12 @@ func BenchmarkDispatch10k(b *testing.B) {
405	439		c, srv := dispatchConn(b)
406	440		var sink int
407	441		h := Handler{
408		-	OnEvent: func(ev *Event) { sink += int(ev.Kind) },
	442	+	OnEvent: func(ev *Event) {
	443	+	if ev.Kind == 22 {
	444	+	_ = c.Expect(ev.Seq+1, kindUser, 0, 0, 0, 0)
	445	+	}
	446	+	sink += int(ev.Kind)
	447	+	},
409	448		OnReply: func(r *Reply) { sink += len(r.Raw) },
410	449		OnError: func(e *XError) { sink += int(e.Code) },
411	450		}

@@ -415,13 +454,17 @@ func BenchmarkDispatch10k(b *testing.B) {
415	454		for b.Loop() {
416	455		feed(b, c, srv, stream, &h)
417	456		}
	457	+	if c.Stats().DroppedReplies != 0 {
	458	+	b.Fatal(c.Stats())
	459	+	}
418	460		_ = sink
419	461		}
420	462
421	463		// 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.
	464	+	// Invariants: no panic; the callbacks' bytes concatenate to the stream's
	465	+	// complete packets minus the (unregistered, so dropped) replies, which
	466	+	// Stats.DroppedReplies counts; Seq never decreases; Dispatch ends with io.EOF
	467	+	// (or ErrPacketTooLarge) after the server hangs up.
425	468		func FuzzDispatchStream(f *testing.F) {
426	469		stream, _ := scriptedStream(f)
427	470		f.Add(stream, uint16(7))

@@ -463,13 +506,22 @@ func FuzzDispatchStream(f *testing.F) {
463	506		if err != io.EOF && !errors.Is(err, ErrPacketTooLarge) {
464	507		t.Fatal(err)
465	508		}
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)
	509	+	var want []byte
	510	+	var replies uint64
	511	+	for p := stream; len(p) >= 32; {
	512	+	n, ok := packetSize(p, c.maxPacket)
	513	+	if !ok \|\| n > len(p) {
	514	+	break
472	515		}
	516	+	if p[0] == 1 {
	517	+	replies++
	518	+	} else {
	519	+	want = append(want, p[:n]...)
	520	+	}
	521	+	p = p[n:]
	522	+	}
	523	+	if !bytes.Equal(got, want) \|\| c.Stats().DroppedReplies != replies {
	524	+	t.Fatalf("callbacks diverge from the stream: %d/%d bytes, %d dropped of %d replies", len(got), len(want), c.Stats().DroppedReplies, replies)
473	525		}
474	526		})
475	527		}

 	h   Handler
 }
-func newRecorder() *recorder {
+// newRecorder records every callback of c. Replies reach OnReply only when
+// registered: on each MapRequest event (the first packet of scriptedStream)
+// it registers the stream's replies, at ev.Seq+1, +4 and +6.
+func newRecorder(c *Conn) *recorder {
 	r := &recorder{}
 	r.h = Handler{
 		OnEvent: func(ev *Event) {
+			expectScripted(c, ev)
 			r.got = append(r.got, record{'e', ev.Seq, uint32(ev.Kind), append(bytes.Clone(ev.Raw[:]), ev.Tail...)})
 		},
 		OnReply: func(rp *Reply) {
 		want = append(want, record{class, Seq(seq), kind, p})
 	}
 	add('e', 20, golden(t, "xproto/MapRequest.event"), 3)
-	add('r', 0, golden(t, "xproto/GetGeometry.reply"), 4)
+	add('r', uint32(kindUser), golden(t, "xproto/GetGeometry.reply"), 4)
 	add('x', 3, golden(t, "xproto/BadWindow.error"), 5)
 	add('e', uint32(ExtGenericEvent)<<8|0x81, golden(t, "xproto/GeGeneric.event"), 6)
 	add('e', uint32(ExtXFixes)<<8, golden(t, "xfixes/SelectionNotify.event"), 6)
-	add('r', 0, golden(t, "xproto/GetProperty.reply"), 7)
+	add('r', uint32(kindUser), golden(t, "xproto/GetProperty.reply"), 7)
 	km := golden(t, "xproto/KeymapNotify.event") // no sequence: inherits 7
 	stream = append(stream, km...)
 	want = append(want, record{'e', 7, codeKeymapNotify, km})
 	add('e', uint32(KindUnknown), pktEvent(80, 0, 0x11), 8) // unbound code
 	add('e', uint32(ExtGenericEvent)<<8|0x80, pktXGE(0, 0, 0), 8)
-	add('r', 0, pktReply(0, 0), 9)
+	add('r', uint32(kindUser), pktReply(0, 0), 9)
 	sent := pktEvent(0x80|33, 0, 0x22) // SendEvent ClientMessage
 	add('e', 33, sent, 9)
 	return stream, want
 }
+// expectScripted registers the replies of scriptedStream when its first
+// event arrives. Allocation-free, so the alloc tests use it too.
+func expectScripted(c *Conn, ev *Event) {
+	if ev.Kind != 20 {
+		return
+	}
+	for _, d := range [...]Seq{1, 4, 6} {
+		if err := c.Expect(ev.Seq+d, kindUser, uint32(d), 0, 0, 0); err != nil {
+			panic(err)
+		}
+	}
+}
+// expect registers seqs with kind kindUser.
+func expect(t testing.TB, c *Conn, seqs ...Seq) {
+	t.Helper()
+	for _, s := range seqs {
+		if err := c.Expect(s, kindUser, 0, 0, 0, 0); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
 func checkRecords(t testing.TB, got, want []record) {
 	t.Helper()
 	if len(got) != len(want) {
 	stream, want := scriptedStream(t)
 	t.Run("one read", func(t *testing.T) {
 		c, srv := dispatchConn(t)
-		r := newRecorder()
+		r := newRecorder(c)
 		feed(t, c, srv, stream, &r.h)
 		checkRecords(t, r.got, want)
 		if c.buf.acked != 9 || c.seq != 9 {
 			t.Fatalf("acked %d seq %d", c.buf.acked, c.seq)
 		}
+		if st := c.Stats(); st.UnknownEvents != 1 || st.DroppedReplies != 0 || st.UnhandledErrors != 0 || st.MaxRingUse != 3 {
+			t.Fatalf("%+v", st)
+		}
 	})
 	t.Run("split at every byte", func(t *testing.T) {
 		for k := range len(stream) + 1 {
 			c, srv := dispatchConn(t)
-			r := newRecorder()
+			r := newRecorder(c)
 			feed(t, c, srv, stream[:k], &r.h)
 			feed(t, c, srv, stream[k:], &r.h)
 			checkRecords(t, r.got, want)
 	})
 	t.Run("byte by byte", func(t *testing.T) {
 		c, srv := dispatchConn(t)
-		r := newRecorder()
+		r := newRecorder(c)
 		for i := range stream {
 			feed(t, c, srv, stream[i:i+1], &r.h)
 		}
 func TestDispatchSeqWidening(t *testing.T) {
 	c, srv := dispatchConn(t)
-	r := newRecorder()
+	r := newRecorder(c)
 	// Wire seqs climbing through two 16-bit wraps.
 	wires := []uint16{1, 0x7fff, 0xfffe, 0xffff, 0, 3, 0x8000, 0xffff, 2}
 	wantSeq := []Seq{1, 0x7fff, 0xfffe, 0xffff, 0x10000, 0x10003, 0x18000, 0x1ffff, 0x20002}
 		}
 	}
 	stream = append(stream, golden(t, "xproto/KeymapNotify.event")...)
+	expect(t, c, 1, 0xffff, 0x18000)
 	feed(t, c, srv, stream, &r.h)
 	if len(r.got) != len(wires)+1 {
 		t.Fatal(r.got)
 	// ack never passes requests that were not sent.
 	c2, srv2 := dispatchConn(t)
 	c2.buf.next = 5
+	expect(t, c2, 9)
 	feed(t, c2, srv2, pktReply(9, 0), nil)
 	if c2.buf.acked != 0 || c2.seq != 9 {
 		t.Fatalf("acked %d seq %d", c2.buf.acked, c2.seq)
 func TestDispatchGrowth(t *testing.T) {
 	c, srv := dispatchConn(t)
-	r := newRecorder()
+	r := newRecorder(c)
 	big := pktReply(1, 300<<10) // larger than the 64 KiB ring
 	xge := pktXGE(2, 5, 100<<10)
 	stream := append(append(append([]byte{}, pktEvent(22, 1, 0)...), big...), xge...)
+	expect(t, c, 1)
 	feed(t, c, srv, stream, &r.h)
 	if len(r.got) != 3 || !bytes.Equal(r.got[1].data, big) || !bytes.Equal(r.got[2].data, xge) {
 		t.Fatalf("%v", r.got)
 	t.Run("exactly MaxPacket", func(t *testing.T) {
 		c, srv := dispatchConn(t)
 		c.maxPacket = 256 << 10
-		r := newRecorder()
+		r := newRecorder(c)
 		p := pktReply(1, c.maxPacket-32)
+		expect(t, c, 1)
 		feed(t, c, srv, p, &r.h)
 		if len(r.got) != 1 || len(r.got[0].data) != c.maxPacket || len(c.rb) != c.maxPacket {
 			t.Fatalf("%v ring %d", r.got, len(c.rb))
 	} {
 		t.Run(tc.name+" beyond MaxPacket", func(t *testing.T) {
 			c, srv := dispatchConn(t)
-			r := newRecorder()
+			r := newRecorder(c)
 			p := tc.p
 			binary.LittleEndian.PutUint32(p[4:], (MaxPacket-32)/4+1)
 			stream := append(pktEvent(22, 1, 0), p...)
 func TestDispatchEOF(t *testing.T) {
 	for _, tail := range [][]byte{nil, pktReply(2, 64)[:40]} {
 		c, srv := dispatchConn(t)
-		r := newRecorder()
+		r := newRecorder(c)
 		data := append(pktEvent(22, 1, 0), tail...)
 		if _, err := unix.Write(srv, data); err != nil {
 			t.Fatal(err)
 	var events, replies, errs int
 	var sink byte
 	h := Handler{
-		OnEvent: func(ev *Event) { events++; sink ^= ev.Raw[1]; sink ^= byte(len(ev.Tail)) },
+		OnEvent: func(ev *Event) { expectScripted(c, ev); events++; sink ^= ev.Raw[1]; sink ^= byte(len(ev.Tail)) },
 		OnReply: func(r *Reply) { replies++; sink ^= r.Raw[len(r.Raw)-1] },
 		OnError: func(e *XError) { errs++; sink ^= e.Code },
 	}
 	if n := testing.AllocsPerRun(5, func() { feed(t, c, srv, stream, &h) }); n != 0 {
 		t.Fatalf("Dispatch allocates %v times per 1 MiB", n)
 	}
-	if events+replies+errs != 7*packets || errs == 0 || replies == 0 {
+	if events+replies+errs != 7*packets || errs == 0 || replies == 0 || c.Stats().DroppedReplies != 0 {
-		t.Fatal(events, replies, errs)
+		t.Fatal(events, replies, errs, c.Stats())
 	}
 	_ = sink
 }
 	c, srv := dispatchConn(b)
 	var sink int
 	h := Handler{
-		OnEvent: func(ev *Event) { sink += int(ev.Kind) },
+		OnEvent: func(ev *Event) {
+			if ev.Kind == 22 {
+				_ = c.Expect(ev.Seq+1, kindUser, 0, 0, 0, 0)
+			}
+			sink += int(ev.Kind)
+		},
 		OnReply: func(r *Reply) { sink += len(r.Raw) },
 		OnError: func(e *XError) { sink += int(e.Code) },
 	}
 	for b.Loop() {
 		feed(b, c, srv, stream, &h)
 	}
+	if c.Stats().DroppedReplies != 0 {
+		b.Fatal(c.Stats())
+	}
 	_ = sink
 }
 // FuzzDispatchStream feeds an arbitrary byte stream in arbitrary chunks.
-// Invariants: no panic; the callbacks' bytes concatenate to a prefix of the
+// Invariants: no panic; the callbacks' bytes concatenate to the stream's
-// stream covering every complete packet; Seq never decreases; Dispatch ends
+// complete packets minus the (unregistered, so dropped) replies, which
-// with io.EOF (or ErrPacketTooLarge) after the server hangs up.
+// Stats.DroppedReplies counts; Seq never decreases; Dispatch ends with io.EOF
+// (or ErrPacketTooLarge) after the server hangs up.
 func FuzzDispatchStream(f *testing.F) {
 	stream, _ := scriptedStream(f)
 	f.Add(stream, uint16(7))
 		if err != io.EOF && !errors.Is(err, ErrPacketTooLarge) {
 			t.Fatal(err)
 		}
-		if !bytes.HasPrefix(stream, got) {
+		var want []byte
-			t.Fatal("callbacks diverge from the stream")
+		var replies uint64
-		}
+		for p := stream; len(p) >= 32; {
-		if rest := stream[len(got):]; err == io.EOF && len(rest) >= 32 {
+			n, ok := packetSize(p, c.maxPacket)
-			if n, ok := packetSize(rest, c.maxPacket); ok && n <= len(rest) {
+			if !ok || n > len(p) {
-				t.Fatalf("complete packet of %d bytes left undelivered", n)
+				break
 			}
+			if p[0] == 1 {
+				replies++
+			} else {
+				want = append(want, p[:n]...)
+			}
+			p = p[n:]
+		}
+		if !bytes.Equal(got, want) || c.Stats().DroppedReplies != replies {
+			t.Fatalf("callbacks diverge from the stream: %d/%d bytes, %d dropped of %d replies", len(got), len(want), c.Stats().DroppedReplies, replies)
 		}
 	})
 }

A ring.go +120 -0

@@ -0,0 +1,120 @@
	1	+	package xturbo
	2	+
	3	+	import "errors"
	4	+
	5	+	// ErrRingFull reports more than ringSize registered requests outstanding:
	6	+	// the Expect slot of seq still belongs to an earlier request that has not
	7	+	// completed. Treat it as a bug: Sync, then retry.
	8	+	var ErrRingFull = errors.New("xturbo: reply ring full")
	9	+
	10	+	var errReservedKind = errors.New("xturbo: reply kinds below 16 are reserved")
	11	+
	12	+	const (
	13	+	ringSize = 1024
	14	+	ringMask = ringSize - 1
	15	+	)
	16	+
	17	+	// Reply kinds below kindUser are reserved by root and consumed internally;
	18	+	// they never reach Handler.OnReply.
	19	+	const (
	20	+	kindNone uint16 = iota // unregistered
	21	+	kindDiscard // consume and drop (resync GetInputFocus)
	22	+	kindInternAtom // InternAtoms: atom into dst
	23	+	kindQueryExtension // connect: fill the ExtTable
	24	+	kindXIDRange // XC-MISC GetXIDRange refill
	25	+	kindBigreqEnable // BIG-REQUESTS Enable: Buf limit
	26	+	kindUser uint16 = 16
	27	+	)
	28	+
	29	+	// slot is one Expect registration; seq 0 means free.
	30	+	type slot struct {
	31	+	seq Seq
	32	+	kind uint16
	33	+	args [4]uint32
	34	+	}
	35	+
	36	+	// Stats are connection counters, for diagnostics.
	37	+	type Stats struct {
	38	+	DroppedReplies uint64 // replies with no Expect slot
	39	+	UnhandledErrors uint64 // errors dispatched with a nil Handler.OnError
	40	+	UnknownEvents uint64 // events decoded as KindUnknown
	41	+	Resyncs uint64 // GetInputFocus appended by Flush to re-sync sequences
	42	+	MaxRingUse int // most Expect slots outstanding at once
	43	+	}
	44	+
	45	+	// Stats returns the connection counters.
	46	+	func (c *Conn) Stats() Stats { return c.stats }
	47	+
	48	+	// Expect registers the request seq so its reply, or its error, is delivered
	49	+	// with kind and args a0..a3 (Reply.Kind/Args, XError.Kind/Args). Call it
	50	+	// before the next Dispatch. kind must be 16 or more; lower kinds are reserved.
	51	+	// A reply without a registration is dropped (Stats.DroppedReplies); a
	52	+	// registered request that completes without reply or error frees its slot
	53	+	// silently once a later sequence number arrives. seq 0 (a failed request) and
	54	+	// requests already completed are ignored. ErrRingFull: more than 1024
	55	+	// registered requests outstanding.
	56	+	func (c *Conn) Expect(seq Seq, kind uint16, a0, a1, a2, a3 uint32) error {
	57	+	if kind < kindUser {
	58	+	return errReservedKind
	59	+	}
	60	+	return c.expect(seq, kind, [4]uint32{a0, a1, a2, a3})
	61	+	}
	62	+
	63	+	func (c *Conn) expect(seq Seq, kind uint16, args [4]uint32) error {
	64	+	if seq == 0 \|\| seq < c.seq {
	65	+	return nil
	66	+	}
	67	+	s := &c.ring[seq&ringMask]
	68	+	switch s.seq {
	69	+	case seq:
	70	+	case 0:
	71	+	c.live++
	72	+	c.stats.MaxRingUse = max(c.stats.MaxRingUse, c.live)
	73	+	default:
	74	+	return ErrRingFull
	75	+	}
	76	+	*s = slot{seq: seq, kind: kind, args: args}
	77	+	return nil
	78	+	}
	79	+
	80	+	// retire frees the slots of requests in [prev, full): a packet with sequence
	81	+	// full proves they completed. The slot of full itself stays, its reply or
	82	+	// error may still follow its events.
	83	+	func (c *Conn) retire(prev, full Seq) {
	84	+	if c.live == 0 \|\| full <= prev {
	85	+	return
	86	+	}
	87	+	if full-prev >= ringSize {
	88	+	for i := range c.ring {
	89	+	if s := c.ring[i].seq; s != 0 && s < full {
	90	+	c.ring[i] = slot{}
	91	+	c.live--
	92	+	}
	93	+	}
	94	+	return
	95	+	}
	96	+	for seq := max(prev, 1); seq < full; seq++ {
	97	+	if s := &c.ring[seq&ringMask]; s.seq == seq {
	98	+	*s = slot{}
	99	+	c.live--
	100	+	}
	101	+	}
	102	+	}
	103	+
	104	+	// take frees and returns the slot of seq, or ok=false when none is registered.
	105	+	func (c *Conn) take(seq Seq) (kind uint16, args [4]uint32, ok bool) {
	106	+	s := &c.ring[seq&ringMask]
	107	+	if s.seq != seq \|\| seq == 0 {
	108	+	return 0, [4]uint32{}, false
	109	+	}
	110	+	kind, args = s.kind, s.args
	111	+	*s = slot{}
	112	+	c.live--
	113	+	return kind, args, true
	114	+	}
	115	+
	116	+	// consumeReply handles a reply registered with a reserved kind. kindDiscard
	117	+	// replies are dropped; the bootstrap kinds (kindInternAtom,
	118	+	// kindQueryExtension, kindXIDRange, kindBigreqEnable) get their consumers
	119	+	// with the connection bootstrap and are dropped until then.
	120	+	func (c Conn) consumeReply(Reply) {}

@@ -0,0 +1,120 @@
	1	+	package xturbo
	2	+
	3	+	import "errors"
	4	+
	5	+	// ErrRingFull reports more than ringSize registered requests outstanding:
	6	+	// the Expect slot of seq still belongs to an earlier request that has not
	7	+	// completed. Treat it as a bug: Sync, then retry.
	8	+	var ErrRingFull = errors.New("xturbo: reply ring full")
	9	+
	10	+	var errReservedKind = errors.New("xturbo: reply kinds below 16 are reserved")
	11	+
	12	+	const (
	13	+	ringSize = 1024
	14	+	ringMask = ringSize - 1
	15	+	)
	16	+
	17	+	// Reply kinds below kindUser are reserved by root and consumed internally;
	18	+	// they never reach Handler.OnReply.
	19	+	const (
	20	+	kindNone uint16 = iota // unregistered
	21	+	kindDiscard // consume and drop (resync GetInputFocus)
	22	+	kindInternAtom // InternAtoms: atom into dst
	23	+	kindQueryExtension // connect: fill the ExtTable
	24	+	kindXIDRange // XC-MISC GetXIDRange refill
	25	+	kindBigreqEnable // BIG-REQUESTS Enable: Buf limit
	26	+	kindUser uint16 = 16
	27	+	)
	28	+
	29	+	// slot is one Expect registration; seq 0 means free.
	30	+	type slot struct {
	31	+	seq Seq
	32	+	kind uint16
	33	+	args [4]uint32
	34	+	}
	35	+
	36	+	// Stats are connection counters, for diagnostics.
	37	+	type Stats struct {
	38	+	DroppedReplies uint64 // replies with no Expect slot
	39	+	UnhandledErrors uint64 // errors dispatched with a nil Handler.OnError
	40	+	UnknownEvents uint64 // events decoded as KindUnknown
	41	+	Resyncs uint64 // GetInputFocus appended by Flush to re-sync sequences
	42	+	MaxRingUse int // most Expect slots outstanding at once
	43	+	}
	44	+
	45	+	// Stats returns the connection counters.
	46	+	func (c *Conn) Stats() Stats { return c.stats }
	47	+
	48	+	// Expect registers the request seq so its reply, or its error, is delivered
	49	+	// with kind and args a0..a3 (Reply.Kind/Args, XError.Kind/Args). Call it
	50	+	// before the next Dispatch. kind must be 16 or more; lower kinds are reserved.
	51	+	// A reply without a registration is dropped (Stats.DroppedReplies); a
	52	+	// registered request that completes without reply or error frees its slot
	53	+	// silently once a later sequence number arrives. seq 0 (a failed request) and
	54	+	// requests already completed are ignored. ErrRingFull: more than 1024
	55	+	// registered requests outstanding.
	56	+	func (c *Conn) Expect(seq Seq, kind uint16, a0, a1, a2, a3 uint32) error {
	57	+	if kind < kindUser {
	58	+	return errReservedKind
	59	+	}
	60	+	return c.expect(seq, kind, [4]uint32{a0, a1, a2, a3})
	61	+	}
	62	+
	63	+	func (c *Conn) expect(seq Seq, kind uint16, args [4]uint32) error {
	64	+	if seq == 0 \|\| seq < c.seq {
	65	+	return nil
	66	+	}
	67	+	s := &c.ring[seq&ringMask]
	68	+	switch s.seq {
	69	+	case seq:
	70	+	case 0:
	71	+	c.live++
	72	+	c.stats.MaxRingUse = max(c.stats.MaxRingUse, c.live)
	73	+	default:
	74	+	return ErrRingFull
	75	+	}
	76	+	*s = slot{seq: seq, kind: kind, args: args}
	77	+	return nil
	78	+	}
	79	+
	80	+	// retire frees the slots of requests in [prev, full): a packet with sequence
	81	+	// full proves they completed. The slot of full itself stays, its reply or
	82	+	// error may still follow its events.
	83	+	func (c *Conn) retire(prev, full Seq) {
	84	+	if c.live == 0 \|\| full <= prev {
	85	+	return
	86	+	}
	87	+	if full-prev >= ringSize {
	88	+	for i := range c.ring {
	89	+	if s := c.ring[i].seq; s != 0 && s < full {
	90	+	c.ring[i] = slot{}
	91	+	c.live--
	92	+	}
	93	+	}
	94	+	return
	95	+	}
	96	+	for seq := max(prev, 1); seq < full; seq++ {
	97	+	if s := &c.ring[seq&ringMask]; s.seq == seq {
	98	+	*s = slot{}
	99	+	c.live--
	100	+	}
	101	+	}
	102	+	}
	103	+
	104	+	// take frees and returns the slot of seq, or ok=false when none is registered.
	105	+	func (c *Conn) take(seq Seq) (kind uint16, args [4]uint32, ok bool) {
	106	+	s := &c.ring[seq&ringMask]
	107	+	if s.seq != seq \|\| seq == 0 {
	108	+	return 0, [4]uint32{}, false
	109	+	}
	110	+	kind, args = s.kind, s.args
	111	+	*s = slot{}
	112	+	c.live--
	113	+	return kind, args, true
	114	+	}
	115	+
	116	+	// consumeReply handles a reply registered with a reserved kind. kindDiscard
	117	+	// replies are dropped; the bootstrap kinds (kindInternAtom,
	118	+	// kindQueryExtension, kindXIDRange, kindBigreqEnable) get their consumers
	119	+	// with the connection bootstrap and are dropped until then.
	120	+	func (c Conn) consumeReply(Reply) {}

A ring_test.go +250 -0

@@ -0,0 +1,250 @@
	1	+	package xturbo
	2	+
	3	+	import (
	4	+	"encoding/binary"
	5	+	"errors"
	6	+	"testing"
	7	+	)
	8	+
	9	+	const kindGeom = kindUser + 1
	10	+
	11	+	func TestExpectReplyRouting(t *testing.T) {
	12	+	c, srv := dispatchConn(t)
	13	+	r := newRecorder(c)
	14	+	var args [4]uint32
	15	+	r.h.OnReply = func(rp *Reply) {
	16	+	r.got = append(r.got, record{'r', rp.Seq, uint32(rp.Kind), nil})
	17	+	args = rp.Args
	18	+	}
	19	+	if err := c.Expect(2, kindGeom, 10, 11, 12, 13); err != nil {
	20	+	t.Fatal(err)
	21	+	}
	22	+	if err := c.Expect(3, kindGeom+1, 0, 0, 0, 0); err != nil {
	23	+	t.Fatal(err)
	24	+	}
	25	+	if st := c.Stats(); st.MaxRingUse != 2 {
	26	+	t.Fatal(st)
	27	+	}
	28	+	// 1: unregistered reply, dropped. 2: registered. 3: registered no-reply
	29	+	// request, retired by the later reply 4, which is unregistered.
	30	+	stream := append(append(append(pktReply(1, 0), pktReply(2, 8)...), pktEvent(22, 3, 0)...), pktReply(4, 0)...)
	31	+	feed(t, c, srv, stream, &r.h)
	32	+	if len(r.got) != 2 \|\| r.got[0].class != 'r' \|\| r.got[0].seq != 2 \|\| r.got[0].kind != uint32(kindGeom) \|\| args != [4]uint32{10, 11, 12, 13} {
	33	+	t.Fatalf("%v %v", r.got, args)
	34	+	}
	35	+	if st := c.Stats(); st.DroppedReplies != 2 \|\| c.live != 0 {
	36	+	t.Fatalf("%+v live %d", st, c.live)
	37	+	}
	38	+	// A second reply to the same seq (or one after completion) is dropped.
	39	+	feed(t, c, srv, pktReply(4, 0), &r.h)
	40	+	if c.Stats().DroppedReplies != 3 {
	41	+	t.Fatal(c.Stats())
	42	+	}
	43	+	// Expect on a completed request, or seq 0, is ignored.
	44	+	if c.Expect(2, kindGeom, 0, 0, 0, 0) != nil \|\| c.Expect(0, kindGeom, 0, 0, 0, 0) != nil \|\| c.live != 0 {
	45	+	t.Fatal("stale Expect registered")
	46	+	}
	47	+	if err := c.Expect(9, kindDiscard, 0, 0, 0, 0); err == nil {
	48	+	t.Fatal("reserved kind accepted")
	49	+	}
	50	+	}
	51	+
	52	+	func TestExpectErrorRouting(t *testing.T) {
	53	+	c, srv := dispatchConn(t)
	54	+	var got []XError
	55	+	h := Handler{OnError: func(e XError) { got = append(got, e) }}
	56	+	if err := c.Expect(5, kindGeom, 7, 0, 0, 1); err != nil {
	57	+	t.Fatal(err)
	58	+	}
	59	+	stream := append(append(pktError(3, 4, 14), pktError(3, 5, 14)...), pktError(140, 6, 138)...)
	60	+	feed(t, c, srv, stream, &h)
	61	+	if len(got) != 3 {
	62	+	t.Fatal(got)
	63	+	}
	64	+	if got[0].Kind != 0 \|\| got[0].Args != [4]uint32{} \|\| got[0].Seq != 4 {
	65	+	t.Fatalf("unregistered: %+v", got[0])
	66	+	}
	67	+	if got[1].Kind != kindGeom \|\| got[1].Args != [4]uint32{7, 0, 0, 1} \|\| got[1].Seq != 5 \|\| got[1].Major != 14 \|\| got[1].Bad != 0xdead {
	68	+	t.Fatalf("registered: %+v", got[1])
	69	+	}
	70	+	if got[2].Ext != ExtXFixes \|\| got[2].Rel != 0 \|\| got[2].Kind != 0 {
	71	+	t.Fatalf("extension: %+v", got[2])
	72	+	}
	73	+	if c.live != 0 \|\| c.Stats().UnhandledErrors != 0 {
	74	+	t.Fatal(c.live, c.Stats())
	75	+	}
	76	+	// A nil OnError counts.
	77	+	feed(t, c, srv, pktError(3, 7, 14), nil)
	78	+	if st := c.Stats(); st.UnhandledErrors != 1 {
	79	+	t.Fatal(st)
	80	+	}
	81	+	}
	82	+
	83	+	func TestReservedKindsConsumed(t *testing.T) {
	84	+	c, srv := dispatchConn(t)
	85	+	r := newRecorder(c)
	86	+	for i, k := range []uint16{kindDiscard, kindInternAtom, kindQueryExtension, kindXIDRange, kindBigreqEnable, 15} {
	87	+	if err := c.expect(Seq(i+1), k, [4]uint32{}); err != nil {
	88	+	t.Fatal(err)
	89	+	}
	90	+	}
	91	+	var stream []byte
	92	+	for i := range 6 {
	93	+	stream = append(stream, pktReply(uint16(i+1), 4)...)
	94	+	}
	95	+	feed(t, c, srv, stream, &r.h)
	96	+	if len(r.got) != 0 \|\| c.live != 0 \|\| c.Stats().DroppedReplies != 0 {
	97	+	t.Fatalf("%v live %d %+v", r.got, c.live, c.Stats())
	98	+	}
	99	+	}
	100	+
	101	+	func TestExpectRingFull(t *testing.T) {
	102	+	c, srv := dispatchConn(t)
	103	+	for seq := Seq(1); seq <= ringSize; seq++ {
	104	+	if err := c.Expect(seq, kindGeom, 0, 0, 0, 0); err != nil {
	105	+	t.Fatal(seq, err)
	106	+	}
	107	+	}
	108	+	if err := c.Expect(ringSize+1, kindGeom, 0, 0, 0, 0); !errors.Is(err, ErrRingFull) {
	109	+	t.Fatal(err)
	110	+	}
	111	+	// Re-registering the same seq overwrites.
	112	+	if err := c.Expect(1, kindGeom+2, 0, 0, 0, 0); err != nil \|\| c.live != ringSize \|\| c.Stats().MaxRingUse != ringSize {
	113	+	t.Fatal(err, c.live)
	114	+	}
	115	+	// An event with seq 1 does not free slot 1 (its reply may follow); one
	116	+	// with seq 2 does.
	117	+	feed(t, c, srv, pktEvent(22, 1, 0), nil)
	118	+	if !errors.Is(c.Expect(ringSize+1, kindGeom, 0, 0, 0, 0), ErrRingFull) {
	119	+	t.Fatal("slot freed early")
	120	+	}
	121	+	feed(t, c, srv, pktEvent(22, 2, 0), nil)
	122	+	if err := c.Expect(ringSize+1, kindGeom, 0, 0, 0, 0); err != nil \|\| c.live != ringSize {
	123	+	t.Fatal(err, c.live)
	124	+	}
	125	+	// A jump far ahead retires every older slot.
	126	+	feed(t, c, srv, pktEvent(22, 0x9000, 0), nil)
	127	+	if c.live != 0 {
	128	+	t.Fatal(c.live)
	129	+	}
	130	+	}
	131	+
	132	+	func TestAllocExpect(t *testing.T) {
	133	+	c, srv := dispatchConn(t)
	134	+	var sink uint32
	135	+	h := Handler{
	136	+	OnReply: func(r *Reply) { sink += r.Args[0] + uint32(r.Kind) },
	137	+	OnError: func(e *XError) { sink += e.Args[0] + uint32(e.Kind) },
	138	+	}
	139	+	// Each round: 3 registered requests, a reply, an error, a void request
	140	+	// retired by the next round's first packet.
	141	+	round := append(append(pktReply(0, 0), pktError(3, 0, 12)...), pktEvent(22, 0, 0)...)
	142	+	next := Seq(1)
	143	+	run := func() {
	144	+	for i := range Seq(3) {
	145	+	if err := c.Expect(next+i, kindGeom, uint32(i), 0, 0, 0); err != nil {
	146	+	t.Fatal(err)
	147	+	}
	148	+	binary.LittleEndian.PutUint16(round[32*i+2:], uint16(next+i))
	149	+	}
	150	+	next += 3
	151	+	feed(t, c, srv, round, &h)
	152	+	}
	153	+	run()
	154	+	if n := testing.AllocsPerRun(1000, run); n != 0 {
	155	+	t.Fatalf("Expect + reply/error match allocates %v times", n)
	156	+	}
	157	+	if c.Stats().MaxRingUse > 4 {
	158	+	t.Fatal(c.Stats())
	159	+	}
	160	+	_ = sink
	161	+	}
	162	+
	163	+	// FuzzReplyRing drives Expect and packet arrival against a map model. Each op
	164	+	// byte: low 2 bits choose Expect / reply / error / event, the rest a sequence
	165	+	// step. Invariants: Expect fails only with more than ringSize live
	166	+	// registrations of distinct incomplete seqs; each registered reply/error is
	167	+	// delivered once with its kind and args; live count matches the model.
	168	+	func FuzzReplyRing(f *testing.F) {
	169	+	f.Add([]byte{0, 4, 1, 8, 2, 0xff, 3, 0x40})
	170	+	f.Add([]byte{0x80, 0x84, 0x88, 1, 5, 9, 3})
	171	+	f.Fuzz(func(t *testing.T, ops []byte) {
	172	+	c, srv := dispatchConn(t)
	173	+	model := map[Seq]uint32{} // registered seq -> args[0]
	174	+	var gotSeq Seq
	175	+	var gotArg uint32
	176	+	var gotKind uint16
	177	+	delivered := 0
	178	+	h := Handler{
	179	+	OnReply: func(r *Reply) { gotSeq, gotArg, gotKind = r.Seq, r.Args[0], r.Kind; delivered++ },
	180	+	OnError: func(e *XError) { gotSeq, gotArg, gotKind = e.Seq, e.Args[0], e.Kind; delivered++ },
	181	+	}
	182	+	wire := Seq(0) // last sequence sent by the "server"
	183	+	for i, op := range ops {
	184	+	step := Seq(op >> 2)
	185	+	switch op & 3 {
	186	+	case 0: // Expect wire+1+step*13
	187	+	seq := wire + 1 + step*13
	188	+	err := c.Expect(seq, kindGeom, uint32(i), 0, 0, 0)
	189	+	clash := false
	190	+	for s := range model {
	191	+	if s != seq && s&ringMask == seq&ringMask {
	192	+	clash = true
	193	+	}
	194	+	}
	195	+	if clash != errors.Is(err, ErrRingFull) \|\| (err != nil && !clash) {
	196	+	t.Fatalf("Expect(%d) = %v, clash %v", seq, err, clash)
	197	+	}
	198	+	if err == nil {
	199	+	model[seq] = uint32(i)
	200	+	}
	201	+	default: // a reply, error or event for wire+step
	202	+	seq := wire + step
	203	+	if seq == 0 {
	204	+	seq = 1
	205	+	}
	206	+	var p []byte
	207	+	switch op & 3 {
	208	+	case 1:
	209	+	p = pktReply(uint16(seq), 0)
	210	+	case 2:
	211	+	p = pktError(3, uint16(seq), 1)
	212	+	default:
	213	+	p = pktEvent(22, uint16(seq), 0)
	214	+	}
	215	+	delivered = 0
	216	+	gotKind = 0
	217	+	feed(t, c, srv, p, &h)
	218	+	if c.seq != seq {
	219	+	t.Fatalf("widened %d, want %d", c.seq, seq)
	220	+	}
	221	+	arg, reg := model[seq]
	222	+	switch op & 3 {
	223	+	case 1:
	224	+	if reg != (delivered == 1) \|\| reg && (gotSeq != seq \|\| gotArg != arg \|\| gotKind != kindGeom) {
	225	+	t.Fatalf("reply %d: reg %v delivered %d", seq, reg, delivered)
	226	+	}
	227	+	case 2:
	228	+	if delivered != 1 \|\| gotSeq != seq \|\| reg != (gotKind == kindGeom) \|\| reg && gotArg != arg {
	229	+	t.Fatalf("error %d: reg %v kind %d", seq, reg, gotKind)
	230	+	}
	231	+	}
	232	+	if op&3 != 3 {
	233	+	delete(model, seq)
	234	+	}
	235	+	for s := range model {
	236	+	if s < seq {
	237	+	delete(model, s)
	238	+	}
	239	+	}
	240	+	wire = seq
	241	+	}
	242	+	if c.live != len(model) \|\| c.live > ringSize {
	243	+	t.Fatalf("live %d, model %d", c.live, len(model))
	244	+	}
	245	+	if c.Stats().MaxRingUse > ringSize {
	246	+	t.Fatal(c.Stats())
	247	+	}
	248	+	}
	249	+	})
	250	+	}

@@ -0,0 +1,250 @@
	1	+	package xturbo
	2	+
	3	+	import (
	4	+	"encoding/binary"
	5	+	"errors"
	6	+	"testing"
	7	+	)
	8	+
	9	+	const kindGeom = kindUser + 1
	10	+
	11	+	func TestExpectReplyRouting(t *testing.T) {
	12	+	c, srv := dispatchConn(t)
	13	+	r := newRecorder(c)
	14	+	var args [4]uint32
	15	+	r.h.OnReply = func(rp *Reply) {
	16	+	r.got = append(r.got, record{'r', rp.Seq, uint32(rp.Kind), nil})
	17	+	args = rp.Args
	18	+	}
	19	+	if err := c.Expect(2, kindGeom, 10, 11, 12, 13); err != nil {
	20	+	t.Fatal(err)
	21	+	}
	22	+	if err := c.Expect(3, kindGeom+1, 0, 0, 0, 0); err != nil {
	23	+	t.Fatal(err)
	24	+	}
	25	+	if st := c.Stats(); st.MaxRingUse != 2 {
	26	+	t.Fatal(st)
	27	+	}
	28	+	// 1: unregistered reply, dropped. 2: registered. 3: registered no-reply
	29	+	// request, retired by the later reply 4, which is unregistered.
	30	+	stream := append(append(append(pktReply(1, 0), pktReply(2, 8)...), pktEvent(22, 3, 0)...), pktReply(4, 0)...)
	31	+	feed(t, c, srv, stream, &r.h)
	32	+	if len(r.got) != 2 \|\| r.got[0].class != 'r' \|\| r.got[0].seq != 2 \|\| r.got[0].kind != uint32(kindGeom) \|\| args != [4]uint32{10, 11, 12, 13} {
	33	+	t.Fatalf("%v %v", r.got, args)
	34	+	}
	35	+	if st := c.Stats(); st.DroppedReplies != 2 \|\| c.live != 0 {
	36	+	t.Fatalf("%+v live %d", st, c.live)
	37	+	}
	38	+	// A second reply to the same seq (or one after completion) is dropped.
	39	+	feed(t, c, srv, pktReply(4, 0), &r.h)
	40	+	if c.Stats().DroppedReplies != 3 {
	41	+	t.Fatal(c.Stats())
	42	+	}
	43	+	// Expect on a completed request, or seq 0, is ignored.
	44	+	if c.Expect(2, kindGeom, 0, 0, 0, 0) != nil \|\| c.Expect(0, kindGeom, 0, 0, 0, 0) != nil \|\| c.live != 0 {
	45	+	t.Fatal("stale Expect registered")
	46	+	}
	47	+	if err := c.Expect(9, kindDiscard, 0, 0, 0, 0); err == nil {
	48	+	t.Fatal("reserved kind accepted")
	49	+	}
	50	+	}
	51	+
	52	+	func TestExpectErrorRouting(t *testing.T) {
	53	+	c, srv := dispatchConn(t)
	54	+	var got []XError
	55	+	h := Handler{OnError: func(e XError) { got = append(got, e) }}
	56	+	if err := c.Expect(5, kindGeom, 7, 0, 0, 1); err != nil {
	57	+	t.Fatal(err)
	58	+	}
	59	+	stream := append(append(pktError(3, 4, 14), pktError(3, 5, 14)...), pktError(140, 6, 138)...)
	60	+	feed(t, c, srv, stream, &h)
	61	+	if len(got) != 3 {
	62	+	t.Fatal(got)
	63	+	}
	64	+	if got[0].Kind != 0 \|\| got[0].Args != [4]uint32{} \|\| got[0].Seq != 4 {
	65	+	t.Fatalf("unregistered: %+v", got[0])
	66	+	}
	67	+	if got[1].Kind != kindGeom \|\| got[1].Args != [4]uint32{7, 0, 0, 1} \|\| got[1].Seq != 5 \|\| got[1].Major != 14 \|\| got[1].Bad != 0xdead {
	68	+	t.Fatalf("registered: %+v", got[1])
	69	+	}
	70	+	if got[2].Ext != ExtXFixes \|\| got[2].Rel != 0 \|\| got[2].Kind != 0 {
	71	+	t.Fatalf("extension: %+v", got[2])
	72	+	}
	73	+	if c.live != 0 \|\| c.Stats().UnhandledErrors != 0 {
	74	+	t.Fatal(c.live, c.Stats())
	75	+	}
	76	+	// A nil OnError counts.
	77	+	feed(t, c, srv, pktError(3, 7, 14), nil)
	78	+	if st := c.Stats(); st.UnhandledErrors != 1 {
	79	+	t.Fatal(st)
	80	+	}
	81	+	}
	82	+
	83	+	func TestReservedKindsConsumed(t *testing.T) {
	84	+	c, srv := dispatchConn(t)
	85	+	r := newRecorder(c)
	86	+	for i, k := range []uint16{kindDiscard, kindInternAtom, kindQueryExtension, kindXIDRange, kindBigreqEnable, 15} {
	87	+	if err := c.expect(Seq(i+1), k, [4]uint32{}); err != nil {
	88	+	t.Fatal(err)
	89	+	}
	90	+	}
	91	+	var stream []byte
	92	+	for i := range 6 {
	93	+	stream = append(stream, pktReply(uint16(i+1), 4)...)
	94	+	}
	95	+	feed(t, c, srv, stream, &r.h)
	96	+	if len(r.got) != 0 \|\| c.live != 0 \|\| c.Stats().DroppedReplies != 0 {
	97	+	t.Fatalf("%v live %d %+v", r.got, c.live, c.Stats())
	98	+	}
	99	+	}
	100	+
	101	+	func TestExpectRingFull(t *testing.T) {
	102	+	c, srv := dispatchConn(t)
	103	+	for seq := Seq(1); seq <= ringSize; seq++ {
	104	+	if err := c.Expect(seq, kindGeom, 0, 0, 0, 0); err != nil {
	105	+	t.Fatal(seq, err)
	106	+	}
	107	+	}
	108	+	if err := c.Expect(ringSize+1, kindGeom, 0, 0, 0, 0); !errors.Is(err, ErrRingFull) {
	109	+	t.Fatal(err)
	110	+	}
	111	+	// Re-registering the same seq overwrites.
	112	+	if err := c.Expect(1, kindGeom+2, 0, 0, 0, 0); err != nil \|\| c.live != ringSize \|\| c.Stats().MaxRingUse != ringSize {
	113	+	t.Fatal(err, c.live)
	114	+	}
	115	+	// An event with seq 1 does not free slot 1 (its reply may follow); one
	116	+	// with seq 2 does.
	117	+	feed(t, c, srv, pktEvent(22, 1, 0), nil)
	118	+	if !errors.Is(c.Expect(ringSize+1, kindGeom, 0, 0, 0, 0), ErrRingFull) {
	119	+	t.Fatal("slot freed early")
	120	+	}
	121	+	feed(t, c, srv, pktEvent(22, 2, 0), nil)
	122	+	if err := c.Expect(ringSize+1, kindGeom, 0, 0, 0, 0); err != nil \|\| c.live != ringSize {
	123	+	t.Fatal(err, c.live)
	124	+	}
	125	+	// A jump far ahead retires every older slot.
	126	+	feed(t, c, srv, pktEvent(22, 0x9000, 0), nil)
	127	+	if c.live != 0 {
	128	+	t.Fatal(c.live)
	129	+	}
	130	+	}
	131	+
	132	+	func TestAllocExpect(t *testing.T) {
	133	+	c, srv := dispatchConn(t)
	134	+	var sink uint32
	135	+	h := Handler{
	136	+	OnReply: func(r *Reply) { sink += r.Args[0] + uint32(r.Kind) },
	137	+	OnError: func(e *XError) { sink += e.Args[0] + uint32(e.Kind) },
	138	+	}
	139	+	// Each round: 3 registered requests, a reply, an error, a void request
	140	+	// retired by the next round's first packet.
	141	+	round := append(append(pktReply(0, 0), pktError(3, 0, 12)...), pktEvent(22, 0, 0)...)
	142	+	next := Seq(1)
	143	+	run := func() {
	144	+	for i := range Seq(3) {
	145	+	if err := c.Expect(next+i, kindGeom, uint32(i), 0, 0, 0); err != nil {
	146	+	t.Fatal(err)
	147	+	}
	148	+	binary.LittleEndian.PutUint16(round[32*i+2:], uint16(next+i))
	149	+	}
	150	+	next += 3
	151	+	feed(t, c, srv, round, &h)
	152	+	}
	153	+	run()
	154	+	if n := testing.AllocsPerRun(1000, run); n != 0 {
	155	+	t.Fatalf("Expect + reply/error match allocates %v times", n)
	156	+	}
	157	+	if c.Stats().MaxRingUse > 4 {
	158	+	t.Fatal(c.Stats())
	159	+	}
	160	+	_ = sink
	161	+	}
	162	+
	163	+	// FuzzReplyRing drives Expect and packet arrival against a map model. Each op
	164	+	// byte: low 2 bits choose Expect / reply / error / event, the rest a sequence
	165	+	// step. Invariants: Expect fails only with more than ringSize live
	166	+	// registrations of distinct incomplete seqs; each registered reply/error is
	167	+	// delivered once with its kind and args; live count matches the model.
	168	+	func FuzzReplyRing(f *testing.F) {
	169	+	f.Add([]byte{0, 4, 1, 8, 2, 0xff, 3, 0x40})
	170	+	f.Add([]byte{0x80, 0x84, 0x88, 1, 5, 9, 3})
	171	+	f.Fuzz(func(t *testing.T, ops []byte) {
	172	+	c, srv := dispatchConn(t)
	173	+	model := map[Seq]uint32{} // registered seq -> args[0]
	174	+	var gotSeq Seq
	175	+	var gotArg uint32
	176	+	var gotKind uint16
	177	+	delivered := 0
	178	+	h := Handler{
	179	+	OnReply: func(r *Reply) { gotSeq, gotArg, gotKind = r.Seq, r.Args[0], r.Kind; delivered++ },
	180	+	OnError: func(e *XError) { gotSeq, gotArg, gotKind = e.Seq, e.Args[0], e.Kind; delivered++ },
	181	+	}
	182	+	wire := Seq(0) // last sequence sent by the "server"
	183	+	for i, op := range ops {
	184	+	step := Seq(op >> 2)
	185	+	switch op & 3 {
	186	+	case 0: // Expect wire+1+step*13
	187	+	seq := wire + 1 + step*13
	188	+	err := c.Expect(seq, kindGeom, uint32(i), 0, 0, 0)
	189	+	clash := false
	190	+	for s := range model {
	191	+	if s != seq && s&ringMask == seq&ringMask {
	192	+	clash = true
	193	+	}
	194	+	}
	195	+	if clash != errors.Is(err, ErrRingFull) \|\| (err != nil && !clash) {
	196	+	t.Fatalf("Expect(%d) = %v, clash %v", seq, err, clash)
	197	+	}
	198	+	if err == nil {
	199	+	model[seq] = uint32(i)
	200	+	}
	201	+	default: // a reply, error or event for wire+step
	202	+	seq := wire + step
	203	+	if seq == 0 {
	204	+	seq = 1
	205	+	}
	206	+	var p []byte
	207	+	switch op & 3 {
	208	+	case 1:
	209	+	p = pktReply(uint16(seq), 0)
	210	+	case 2:
	211	+	p = pktError(3, uint16(seq), 1)
	212	+	default:
	213	+	p = pktEvent(22, uint16(seq), 0)
	214	+	}
	215	+	delivered = 0
	216	+	gotKind = 0
	217	+	feed(t, c, srv, p, &h)
	218	+	if c.seq != seq {
	219	+	t.Fatalf("widened %d, want %d", c.seq, seq)
	220	+	}
	221	+	arg, reg := model[seq]
	222	+	switch op & 3 {
	223	+	case 1:
	224	+	if reg != (delivered == 1) \|\| reg && (gotSeq != seq \|\| gotArg != arg \|\| gotKind != kindGeom) {
	225	+	t.Fatalf("reply %d: reg %v delivered %d", seq, reg, delivered)
	226	+	}
	227	+	case 2:
	228	+	if delivered != 1 \|\| gotSeq != seq \|\| reg != (gotKind == kindGeom) \|\| reg && gotArg != arg {
	229	+	t.Fatalf("error %d: reg %v kind %d", seq, reg, gotKind)
	230	+	}
	231	+	}
	232	+	if op&3 != 3 {
	233	+	delete(model, seq)
	234	+	}
	235	+	for s := range model {
	236	+	if s < seq {
	237	+	delete(model, s)
	238	+	}
	239	+	}
	240	+	wire = seq
	241	+	}
	242	+	if c.live != len(model) \|\| c.live > ringSize {
	243	+	t.Fatalf("live %d, model %d", c.live, len(model))
	244	+	}
	245	+	if c.Stats().MaxRingUse > ringSize {
	246	+	t.Fatal(c.Stats())
	247	+	}
	248	+	}
	249	+	})
	250	+	}