fix(runtime): bound and harden Dispatch; reject unsent Expect

Decisions:
- M1: Dispatch sets a dispatching flag (cleared by defer). A nested call
  returns unexported errReentrantDispatch and leaves read state untouched;
  nothing in the public contract needs it exported.
- M2: one Dispatch stops after reading about len(c.rb) bytes (checked after
  each drain, so queued packets are always handled). The fd stays readable
  (level-triggered poll), so the caller flushes and calls Dispatch again.
- L1: errors whose slot has a reserved kind (1..15) go to consumeError; they
  reach OnError only if it returns false. The stub consumes (drops) them,
  like consumeReply; T3.5 replaces it.
- L3: Reply.Raw and Event.Tail are cleared after the callback, and for
  dropped replies, so a grown old ring is not retained.
- L4: Expect of seq >= Buf.Next() returns errExpectUnsent (unexported, a
  caller bug); the internal expect used by the bootstrap is unchanged.
- Fuzz: FuzzDispatchStream starts with a tiny read ring to fuzz growth;
  FuzzReplyRing allows large seq jumps (>=1024 fast retire, 16-bit wrap).
- Tests feed via pump (Dispatch until the socket is empty).

commit: 1aebbe1166ae05c2b7408a3c97b9cfe97fbbd7f6
author: brice · 13h ago · 2026-10-22 09:02 UTC
committer: brice · 12h ago · 2026-10-22 09:36 UTC
parent: 350da19
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
checks: 4/4

5 files · +196 -19

M conn.go +6 -5


@@ -18,11 +18,12 @@ type Conn struct {
18	18		xidBase, xidMask uint32 // resource-id allocation, used by NewXID
19	19
20	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
	21	+	rb []byte
	22	+	rpos, rend int
	23	+	maxPacket int // MaxPacket; lowered by tests
	24	+	rerr error // sticky read failure
	25	+	dispatching bool // inside Dispatch: rejects nested calls
	26	+	seq Seq // last widened sequence number seen
26	27
27	28		// Reply ring: Expect registrations indexed by seq&ringMask.
28	29		ring [ringSize]slot


@@ -18,11 +18,12 @@ type Conn struct {
18		xidBase, xidMask uint32 // resource-id allocation, used by NewXID	18		xidBase, xidMask uint32 // resource-id allocation, used by NewXID
19			19
20		// Read side, owned by Dispatch: rb[rpos:rend] is received, unconsumed.	20		// Read side, owned by Dispatch: rb[rpos:rend] is received, unconsumed.
21	-	rb []byte	21	+	rb []byte
22	-	rpos, rend int	22	+	rpos, rend int
23	-	maxPacket int // MaxPacket; lowered by tests	23	+	maxPacket int // MaxPacket; lowered by tests
24	-	rerr error // sticky read failure	24	+	rerr error // sticky read failure
25	-	seq Seq // last widened sequence number seen	25	+	dispatching bool // inside Dispatch: rejects nested calls
			26	+	seq Seq // last widened sequence number seen
26			27
27		// Reply ring: Expect registrations indexed by seq&ringMask.	28		// Reply ring: Expect registrations indexed by seq&ringMask.
28		ring [ringSize]slot	29		ring [ringSize]slot

M dispatch.go +34 -8


@@ -17,6 +17,11 @@ const MaxPacket = 16 << 20
17	17		// ErrPacketTooLarge reports a server packet longer than MaxPacket.
18	18		var ErrPacketTooLarge = errors.New("xturbo: packet too large")
19	19
	20	+	// errReentrantDispatch is returned by a Dispatch called from a Handler
	21	+	// callback. It is unexported: it reports a programming error, not a condition
	22	+	// callers handle. The outer Dispatch is unaffected.
	23	+	var errReentrantDispatch = errors.New("xturbo: Dispatch called from a Handler callback")
	24	+
20	25		const (
21	26		readBufSize = 64 << 10
22	27		packetHeadSize = 32

@@ -29,7 +34,8 @@ const (
29	34		// once at init; the pointers passed to callbacks are owned by the Conn and are
30	35		// reused for the next packet, so callbacks must not retain them, nor the
31	36		// views they hold (Reply.Raw, Event.Tail). Callbacks may queue requests and
32		-	// call Expect but must not call Dispatch.
	37	+	// call Expect but must not call Dispatch: a nested Dispatch returns an error
	38	+	// and changes nothing.
33	39		type Handler struct {
34	40		OnEvent func(ev *Event)
35	41		OnReply func(r *Reply)

@@ -48,12 +54,20 @@ type Reply struct {
48	54
49	55		var noHandler Handler
50	56
51		-	// Dispatch reads everything the server has sent without blocking and calls h
52		-	// for every complete packet, in wire order. It returns nil once the socket
53		-	// would block, io.EOF when the server closed the connection and
54		-	// ErrPacketTooLarge on a packet beyond MaxPacket. Read errors are sticky and
55		-	// also fail the Buf. Dispatch never flushes. A nil h drops every packet.
	57	+	// Dispatch reads what the server has sent without blocking and calls h for
	58	+	// every complete packet, in wire order. It returns nil once the socket would
	59	+	// block or after a bounded amount of input (about one read ring, so a flooding
	60	+	// server cannot starve the caller's event loop): call it again while the fd is
	61	+	// still readable, and Flush requests queued by callbacks in between. It
	62	+	// returns io.EOF when the server closed the connection and ErrPacketTooLarge
	63	+	// on a packet beyond MaxPacket. Read errors are sticky and also fail the Buf.
	64	+	// Dispatch never flushes. A nil h drops every packet.
56	65		func (c Conn) Dispatch(h Handler) error {
	66	+	if c.dispatching {
	67	+	return errReentrantDispatch
	68	+	}
	69	+	c.dispatching = true
	70	+	defer func() { c.dispatching = false }()
57	71		if c.fd < 0 {
58	72		return ErrClosed
59	73		}

@@ -63,6 +77,7 @@ func (c Conn) Dispatch(h Handler) error {
63	77		if h == nil {
64	78		h = &noHandler
65	79		}
	80	+	budget := len(c.rb)
66	81		for {
67	82		if err := c.drain(h); err != nil {
68	83		return c.failRead(err)

@@ -70,6 +85,9 @@ func (c Conn) Dispatch(h Handler) error {
70	85		if c.fd < 0 { // a callback closed the Conn
71	86		return ErrClosed
72	87		}
	88	+	if budget <= 0 {
	89	+	return nil
	90	+	}
73	91		n, err := unix.Read(c.fd, c.rb[c.rend:])
74	92		switch {
75	93		case err == unix.EINTR:

@@ -84,6 +102,7 @@ func (c Conn) Dispatch(h Handler) error {
84	102		return c.failRead(io.EOF)
85	103		}
86	104		c.rend += n
	105	+	budget -= n
87	106		}
88	107		}
89	108

@@ -171,9 +190,14 @@ func (c Conn) packet(h Handler, p []byte) {
171	190		decodeError(p, &c.ext, &c.xerr)
172	191		c.xerr.Seq = c.widen(c.xerr.Seq)
173	192		c.xerr.Kind, c.xerr.Args, _ = c.take(c.xerr.Seq)
174		-	if h.OnError != nil {
	193	+	switch {
	194	+	case c.xerr.Kind != kindNone && c.xerr.Kind < kindUser:
	195	+	if !c.consumeError(&c.xerr) && h.OnError != nil {
	196	+	h.OnError(&c.xerr)
	197	+	}
	198	+	case h.OnError != nil:
175	199		h.OnError(&c.xerr)
176		-	} else {
	200	+	default:
177	201		c.stats.UnhandledErrors++
178	202		}
179	203		case 1:

@@ -188,6 +212,7 @@ func (c Conn) packet(h Handler, p []byte) {
188	212		case h.OnReply != nil:
189	213		h.OnReply(&c.reply)
190	214		}
	215	+	c.reply.Raw = nil // do not retain the ring
191	216		default:
192	217		ev, _, ok := DecodeEvent(p, &c.ext)
193	218		if !ok { // a sent event claiming code 0 or 1

@@ -206,6 +231,7 @@ func (c Conn) packet(h Handler, p []byte) {
206	231		if h.OnEvent != nil {
207	232		h.OnEvent(&c.ev)
208	233		}
	234	+	c.ev.Tail = nil // do not retain the ring
209	235		}
210	236		}
211	237

 // ErrPacketTooLarge reports a server packet longer than MaxPacket.
 var ErrPacketTooLarge = errors.New("xturbo: packet too large")
+// errReentrantDispatch is returned by a Dispatch called from a Handler
+// callback. It is unexported: it reports a programming error, not a condition
+// callers handle. The outer Dispatch is unaffected.
+var errReentrantDispatch = errors.New("xturbo: Dispatch called from a Handler callback")
 const (
 	readBufSize    = 64 << 10
 	packetHeadSize = 32
 // 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
-// call Expect but must not call Dispatch.
+// call Expect but must not call Dispatch: a nested Dispatch returns an error
+// and changes nothing.
 type Handler struct {
 	OnEvent func(ev *Event)
 	OnReply func(r *Reply)
 var noHandler Handler
-// Dispatch reads everything the server has sent without blocking and calls h
+// Dispatch reads what the server has sent without blocking and calls h for
-// for every complete packet, in wire order. It returns nil once the socket
+// every complete packet, in wire order. It returns nil once the socket would
-// would block, io.EOF when the server closed the connection and
+// block or after a bounded amount of input (about one read ring, so a flooding
-// ErrPacketTooLarge on a packet beyond MaxPacket. Read errors are sticky and
+// server cannot starve the caller's event loop): call it again while the fd is
-// also fail the Buf. Dispatch never flushes. A nil h drops every packet.
+// still readable, and Flush requests queued by callbacks in between. It
+// returns io.EOF when the server closed the connection and ErrPacketTooLarge
+// on a packet beyond MaxPacket. Read errors are sticky and also fail the Buf.
+// Dispatch never flushes. A nil h drops every packet.
 func (c *Conn) Dispatch(h *Handler) error {
+	if c.dispatching {
+		return errReentrantDispatch
+	}
+	c.dispatching = true
+	defer func() { c.dispatching = false }()
 	if c.fd < 0 {
 		return ErrClosed
 	}
 	if h == nil {
 		h = &noHandler
 	}
+	budget := len(c.rb)
 	for {
 		if err := c.drain(h); err != nil {
 			return c.failRead(err)
 		if c.fd < 0 { // a callback closed the Conn
 			return ErrClosed
 		}
+		if budget <= 0 {
+			return nil
+		}
 		n, err := unix.Read(c.fd, c.rb[c.rend:])
 		switch {
 		case err == unix.EINTR:
 			return c.failRead(io.EOF)
 		}
 		c.rend += n
+		budget -= n
 	}
 }
 		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 {
+		switch {
+		case c.xerr.Kind != kindNone && c.xerr.Kind < kindUser:
+			if !c.consumeError(&c.xerr) && h.OnError != nil {
+				h.OnError(&c.xerr)
+			}
+		case h.OnError != nil:
 			h.OnError(&c.xerr)
-		} else {
+		default:
 			c.stats.UnhandledErrors++
 		}
 	case 1:
 		case h.OnReply != nil:
 			h.OnReply(&c.reply)
 		}
+		c.reply.Raw = nil // do not retain the ring
 	default:
 		ev, _, ok := DecodeEvent(p, &c.ext)
 		if !ok { // a sent event claiming code 0 or 1
 		if h.OnEvent != nil {
 			h.OnEvent(&c.ev)
 		}
+		c.ev.Tail = nil // do not retain the ring
 	}
 }

M dispatch_test.go +120 -4


@@ -7,6 +7,7 @@ import (
7	7		"fmt"
8	8		"io"
9	9		"testing"
	10	+	"time"
10	11
11	12		"bnema.dev/xturbo/internal/xtest"
12	13		"golang.org/x/sys/unix"

@@ -121,12 +122,23 @@ func feed(t testing.TB, c Conn, srv int, data []byte, h Handler) {
121	122		t.Fatal(err)
122	123		}
123	124		data = data[n:]
	125	+	pump(t, c, h)
	126	+	}
	127	+	pump(t, c, h)
	128	+	}
	129	+
	130	+	// pump dispatches until the socket holds nothing more: one Dispatch reads a
	131	+	// bounded amount.
	132	+	func pump(t testing.TB, c Conn, h Handler) {
	133	+	t.Helper()
	134	+	for {
124	135		if err := c.Dispatch(h); err != nil {
125	136		t.Fatal(err)
126	137		}
127		-	}
128		-	if err := c.Dispatch(h); err != nil {
129		-	t.Fatal(err)
	138	+	fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
	139	+	if n, _ := unix.Poll(fds, 0); n == 0 {
	140	+	return
	141	+	}
130	142		}
131	143		}
132	144

@@ -278,7 +290,9 @@ func TestDispatchSeqWidening(t *testing.T) {
278	290		// ack never passes requests that were not sent.
279	291		c2, srv2 := dispatchConn(t)
280	292		c2.buf.next = 5
281		-	expect(t, c2, 9)
	293	+	if err := c2.expect(9, kindUser, [4]uint32{}); err != nil { // internal: Expect rejects unsent seqs
	294	+	t.Fatal(err)
	295	+	}
282	296		feed(t, c2, srv2, pktReply(9, 0), nil)
283	297		if c2.buf.acked != 0 \|\| c2.seq != 9 {
284	298		t.Fatalf("acked %d seq %d", c2.buf.acked, c2.seq)

@@ -379,6 +393,106 @@ func TestDispatchCloseInCallback(t *testing.T) {
379	393		}
380	394		}
381	395
	396	+	func TestDispatchReentrant(t *testing.T) {
	397	+	c, srv := dispatchConn(t)
	398	+	var nested error
	399	+	var got []Seq
	400	+	h := Handler{OnEvent: func(ev *Event) {
	401	+	got = append(got, ev.Seq)
	402	+	if len(got) == 1 {
	403	+	nested = c.Dispatch(nil)
	404	+	}
	405	+	}}
	406	+	var stream []byte
	407	+	for i := range 5 {
	408	+	stream = append(stream, pktEvent(22, uint16(i+1), 0)...)
	409	+	}
	410	+	if _, err := unix.Write(srv, stream); err != nil {
	411	+	t.Fatal(err)
	412	+	}
	413	+	if err := c.Dispatch(&h); err != nil {
	414	+	t.Fatal(err)
	415	+	}
	416	+	if !errors.Is(nested, errReentrantDispatch) \|\| len(got) != 5 \|\| got[4] != 5 {
	417	+	t.Fatal(nested, got)
	418	+	}
	419	+	if err := c.Dispatch(&h); err != nil \|\| c.dispatching { // usable afterwards
	420	+	t.Fatal(err)
	421	+	}
	422	+	}
	423	+
	424	+	func TestDispatchBounded(t *testing.T) {
	425	+	c, srv := dispatchConn(t)
	426	+	stop := make(chan struct{})
	427	+	done := make(chan struct{})
	428	+	go func() { // keep the socket full until told to stop
	429	+	defer close(done)
	430	+	buf := bytes.Repeat(pktEvent(22, 1, 0), 2048)
	431	+	for {
	432	+	select {
	433	+	case <-stop:
	434	+	return
	435	+	default:
	436	+	}
	437	+	if _, err := unix.Write(srv, buf); err != nil && err != unix.EAGAIN {
	438	+	return
	439	+	}
	440	+	}
	441	+	}()
	442	+	defer func() { close(stop); <-done }()
	443	+	events := 0
	444	+	h := Handler{OnEvent: func(*Event) { events++ }}
	445	+	fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
	446	+	for n, _ := unix.Poll(fds, 1000); n == 0; n, _ = unix.Poll(fds, 1000) {
	447	+	t.Fatal("writer never filled the socket")
	448	+	}
	449	+	res := make(chan error, 1)
	450	+	go func() { res <- c.Dispatch(&h) }()
	451	+	select {
	452	+	case err := <-res:
	453	+	if err != nil {
	454	+	t.Fatal(err)
	455	+	}
	456	+	case <-time.After(5 * time.Second):
	457	+	t.Fatal("Dispatch did not return under a flooding server")
	458	+	}
	459	+	// one budget of reads, plus at most the bytes already in the ring
	460	+	if max := 2 * len(c.rb) / 32; events == 0 \|\| events > max {
	461	+	t.Fatalf("%d events, want 1..%d", events, max)
	462	+	}
	463	+	}
	464	+
	465	+	func TestDispatchReservedKindError(t *testing.T) {
	466	+	c, srv := dispatchConn(t)
	467	+	r := newRecorder(c)
	468	+	if err := c.expect(1, kindDiscard, [4]uint32{}); err != nil {
	469	+	t.Fatal(err)
	470	+	}
	471	+	feed(t, c, srv, pktError(3, 1, 0), &r.h)
	472	+	if len(r.got) != 0 \|\| c.Stats().UnhandledErrors != 0 {
	473	+	t.Fatalf("reserved-kind error reached OnError: %v", r.got)
	474	+	}
	475	+	if c.live != 0 {
	476	+	t.Fatal("slot not freed")
	477	+	}
	478	+	}
	479	+
	480	+	func TestDispatchClearsBorrowedViews(t *testing.T) {
	481	+	c, srv := dispatchConn(t)
	482	+	var h Handler
	483	+	h.OnReply = func(*Reply) {}
	484	+	expect(t, c, 2)
	485	+	stream := append(append(pktReply(1, 64), pktReply(2, 64)...), pktXGE(3, 5, 64)...)
	486	+	feed(t, c, srv, stream, &h) // reply 1 is dropped, 2 delivered, 3 an event with Tail
	487	+	if c.reply.Raw != nil \|\| c.ev.Tail != nil {
	488	+	t.Fatalf("retained views: reply %d ev %d", len(c.reply.Raw), len(c.ev.Tail))
	489	+	}
	490	+	feed(t, c, srv, pktReply(4, 8), nil) // dropped reply
	491	+	if c.reply.Raw != nil {
	492	+	t.Fatal("dropped reply retained")
	493	+	}
	494	+	}
	495	+
382	496		// recordedStream repeats the scripted stream (events, replies, errors) up to
383	497		// about size bytes.
384	498		func recordedStream(t testing.TB, size int) ([]byte, int) {

@@ -468,6 +582,7 @@ func BenchmarkDispatch10k(b *testing.B) {
468	582		func FuzzDispatchStream(f *testing.F) {
469	583		stream, _ := scriptedStream(f)
470	584		f.Add(stream, uint16(7))
	585	+	f.Add(stream, uint16(0x2000\|7))
471	586		f.Add(stream, uint16(0))
472	587		f.Add(append(pktReply(1, 1024), pktXGE(2, 1, 64)...), uint16(33))
473	588		f.Fuzz(func(t *testing.T, stream []byte, chunk uint16) {

@@ -476,6 +591,7 @@ func FuzzDispatchStream(f *testing.F) {
476	591		}
477	592		c, srv := dispatchConn(t)
478	593		c.maxPacket = 8 << 10
	594	+	c.rb = make([]byte, packetHeadSize+int(chunk>>8)%96) // tiny ring: fuzz growth
479	595		var got []byte
480	596		var last Seq
481	597		mono := func(s Seq) {


@@ -7,6 +7,7 @@ import (
7		"fmt"	7		"fmt"
8		"io"	8		"io"
9		"testing"	9		"testing"
			10	+	"time"
10			11
11		"bnema.dev/xturbo/internal/xtest"	12		"bnema.dev/xturbo/internal/xtest"
12		"golang.org/x/sys/unix"	13		"golang.org/x/sys/unix"

@@ -121,12 +122,23 @@ func feed(t testing.TB, c Conn, srv int, data []byte, h Handler) {
121		t.Fatal(err)	122		t.Fatal(err)
122		}	123		}
123		data = data[n:]	124		data = data[n:]
			125	+	pump(t, c, h)
			126	+	}
			127	+	pump(t, c, h)
			128	+	}
			129	+
			130	+	// pump dispatches until the socket holds nothing more: one Dispatch reads a
			131	+	// bounded amount.
			132	+	func pump(t testing.TB, c Conn, h Handler) {
			133	+	t.Helper()
			134	+	for {
124		if err := c.Dispatch(h); err != nil {	135		if err := c.Dispatch(h); err != nil {
125		t.Fatal(err)	136		t.Fatal(err)
126		}	137		}
127	-	}	138	+	fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
128	-	if err := c.Dispatch(h); err != nil {	139	+	if n, _ := unix.Poll(fds, 0); n == 0 {
129	-	t.Fatal(err)	140	+	return
			141	+	}
130		}	142		}
131		}	143		}
132			144

@@ -278,7 +290,9 @@ func TestDispatchSeqWidening(t *testing.T) {
278		// ack never passes requests that were not sent.	290		// ack never passes requests that were not sent.
279		c2, srv2 := dispatchConn(t)	291		c2, srv2 := dispatchConn(t)
280		c2.buf.next = 5	292		c2.buf.next = 5
281	-	expect(t, c2, 9)	293	+	if err := c2.expect(9, kindUser, [4]uint32{}); err != nil { // internal: Expect rejects unsent seqs
			294	+	t.Fatal(err)
			295	+	}
282		feed(t, c2, srv2, pktReply(9, 0), nil)	296		feed(t, c2, srv2, pktReply(9, 0), nil)
283		if c2.buf.acked != 0 \|\| c2.seq != 9 {	297		if c2.buf.acked != 0 \|\| c2.seq != 9 {
284		t.Fatalf("acked %d seq %d", c2.buf.acked, c2.seq)	298		t.Fatalf("acked %d seq %d", c2.buf.acked, c2.seq)

@@ -379,6 +393,106 @@ func TestDispatchCloseInCallback(t *testing.T) {
379		}	393		}
380		}	394		}
381			395
			396	+	func TestDispatchReentrant(t *testing.T) {
			397	+	c, srv := dispatchConn(t)
			398	+	var nested error
			399	+	var got []Seq
			400	+	h := Handler{OnEvent: func(ev *Event) {
			401	+	got = append(got, ev.Seq)
			402	+	if len(got) == 1 {
			403	+	nested = c.Dispatch(nil)
			404	+	}
			405	+	}}
			406	+	var stream []byte
			407	+	for i := range 5 {
			408	+	stream = append(stream, pktEvent(22, uint16(i+1), 0)...)
			409	+	}
			410	+	if _, err := unix.Write(srv, stream); err != nil {
			411	+	t.Fatal(err)
			412	+	}
			413	+	if err := c.Dispatch(&h); err != nil {
			414	+	t.Fatal(err)
			415	+	}
			416	+	if !errors.Is(nested, errReentrantDispatch) \|\| len(got) != 5 \|\| got[4] != 5 {
			417	+	t.Fatal(nested, got)
			418	+	}
			419	+	if err := c.Dispatch(&h); err != nil \|\| c.dispatching { // usable afterwards
			420	+	t.Fatal(err)
			421	+	}
			422	+	}
			423	+
			424	+	func TestDispatchBounded(t *testing.T) {
			425	+	c, srv := dispatchConn(t)
			426	+	stop := make(chan struct{})
			427	+	done := make(chan struct{})
			428	+	go func() { // keep the socket full until told to stop
			429	+	defer close(done)
			430	+	buf := bytes.Repeat(pktEvent(22, 1, 0), 2048)
			431	+	for {
			432	+	select {
			433	+	case <-stop:
			434	+	return
			435	+	default:
			436	+	}
			437	+	if _, err := unix.Write(srv, buf); err != nil && err != unix.EAGAIN {
			438	+	return
			439	+	}
			440	+	}
			441	+	}()
			442	+	defer func() { close(stop); <-done }()
			443	+	events := 0
			444	+	h := Handler{OnEvent: func(*Event) { events++ }}
			445	+	fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
			446	+	for n, _ := unix.Poll(fds, 1000); n == 0; n, _ = unix.Poll(fds, 1000) {
			447	+	t.Fatal("writer never filled the socket")
			448	+	}
			449	+	res := make(chan error, 1)
			450	+	go func() { res <- c.Dispatch(&h) }()
			451	+	select {
			452	+	case err := <-res:
			453	+	if err != nil {
			454	+	t.Fatal(err)
			455	+	}
			456	+	case <-time.After(5 * time.Second):
			457	+	t.Fatal("Dispatch did not return under a flooding server")
			458	+	}
			459	+	// one budget of reads, plus at most the bytes already in the ring
			460	+	if max := 2 * len(c.rb) / 32; events == 0 \|\| events > max {
			461	+	t.Fatalf("%d events, want 1..%d", events, max)
			462	+	}
			463	+	}
			464	+
			465	+	func TestDispatchReservedKindError(t *testing.T) {
			466	+	c, srv := dispatchConn(t)
			467	+	r := newRecorder(c)
			468	+	if err := c.expect(1, kindDiscard, [4]uint32{}); err != nil {
			469	+	t.Fatal(err)
			470	+	}
			471	+	feed(t, c, srv, pktError(3, 1, 0), &r.h)
			472	+	if len(r.got) != 0 \|\| c.Stats().UnhandledErrors != 0 {
			473	+	t.Fatalf("reserved-kind error reached OnError: %v", r.got)
			474	+	}
			475	+	if c.live != 0 {
			476	+	t.Fatal("slot not freed")
			477	+	}
			478	+	}
			479	+
			480	+	func TestDispatchClearsBorrowedViews(t *testing.T) {
			481	+	c, srv := dispatchConn(t)
			482	+	var h Handler
			483	+	h.OnReply = func(*Reply) {}
			484	+	expect(t, c, 2)
			485	+	stream := append(append(pktReply(1, 64), pktReply(2, 64)...), pktXGE(3, 5, 64)...)
			486	+	feed(t, c, srv, stream, &h) // reply 1 is dropped, 2 delivered, 3 an event with Tail
			487	+	if c.reply.Raw != nil \|\| c.ev.Tail != nil {
			488	+	t.Fatalf("retained views: reply %d ev %d", len(c.reply.Raw), len(c.ev.Tail))
			489	+	}
			490	+	feed(t, c, srv, pktReply(4, 8), nil) // dropped reply
			491	+	if c.reply.Raw != nil {
			492	+	t.Fatal("dropped reply retained")
			493	+	}
			494	+	}
			495	+
382		// recordedStream repeats the scripted stream (events, replies, errors) up to	496		// recordedStream repeats the scripted stream (events, replies, errors) up to
383		// about size bytes.	497		// about size bytes.
384		func recordedStream(t testing.TB, size int) ([]byte, int) {	498		func recordedStream(t testing.TB, size int) ([]byte, int) {

@@ -468,6 +582,7 @@ func BenchmarkDispatch10k(b *testing.B) {
468		func FuzzDispatchStream(f *testing.F) {	582		func FuzzDispatchStream(f *testing.F) {
469		stream, _ := scriptedStream(f)	583		stream, _ := scriptedStream(f)
470		f.Add(stream, uint16(7))	584		f.Add(stream, uint16(7))
			585	+	f.Add(stream, uint16(0x2000\|7))
471		f.Add(stream, uint16(0))	586		f.Add(stream, uint16(0))
472		f.Add(append(pktReply(1, 1024), pktXGE(2, 1, 64)...), uint16(33))	587		f.Add(append(pktReply(1, 1024), pktXGE(2, 1, 64)...), uint16(33))
473		f.Fuzz(func(t *testing.T, stream []byte, chunk uint16) {	588		f.Fuzz(func(t *testing.T, stream []byte, chunk uint16) {

@@ -476,6 +591,7 @@ func FuzzDispatchStream(f *testing.F) {
476		}	591		}
477		c, srv := dispatchConn(t)	592		c, srv := dispatchConn(t)
478		c.maxPacket = 8 << 10	593		c.maxPacket = 8 << 10
			594	+	c.rb = make([]byte, packetHeadSize+int(chunk>>8)%96) // tiny ring: fuzz growth
479		var got []byte	595		var got []byte
480		var last Seq	596		var last Seq
481		mono := func(s Seq) {	597		mono := func(s Seq) {

M ring.go +14 -1


@@ -7,6 +7,8 @@ import "errors"
7	7		// completed. Treat it as a bug: Sync, then retry.
8	8		var ErrRingFull = errors.New("xturbo: reply ring full")
9	9
	10	+	var errExpectUnsent = errors.New("xturbo: Expect of a sequence number not yet sent")
	11	+
10	12		var errReservedKind = errors.New("xturbo: reply kinds below 16 are reserved")
11	13
12	14		const (

@@ -51,12 +53,16 @@ func (c *Conn) Stats() Stats { return c.stats }
51	53		// A reply without a registration is dropped (Stats.DroppedReplies); a
52	54		// registered request that completes without reply or error frees its slot
53	55		// silently once a later sequence number arrives. seq 0 (a failed request) and
54		-	// requests already completed are ignored. ErrRingFull: more than 1024
	56	+	// requests already completed are ignored. A seq that was never sent (at or
	57	+	// beyond Buf.Next) is a caller bug and fails. ErrRingFull: more than 1024
55	58		// registered requests outstanding.
56	59		func (c *Conn) Expect(seq Seq, kind uint16, a0, a1, a2, a3 uint32) error {
57	60		if kind < kindUser {
58	61		return errReservedKind
59	62		}
	63	+	if seq >= c.buf.Next() {
	64	+	return errExpectUnsent
	65	+	}
60	66		return c.expect(seq, kind, [4]uint32{a0, a1, a2, a3})
61	67		}
62	68

@@ -113,6 +119,13 @@ func (c *Conn) take(seq Seq) (kind uint16, args [4]uint32, ok bool) {
113	119		return kind, args, true
114	120		}
115	121
	122	+	// consumeError handles an error whose request was registered with a reserved
	123	+	// kind (1..15). It reports whether it consumed the error; an unconsumed error
	124	+	// is delivered to OnError. The bootstrap consumers arrive with the connection
	125	+	// bootstrap; until then every reserved-kind error is consumed and dropped,
	126	+	// like consumeReply drops its replies.
	127	+	func (c Conn) consumeError(XError) bool { return true }
	128	+
116	129		// consumeReply handles a reply registered with a reserved kind. kindDiscard
117	130		// replies are dropped; the bootstrap kinds (kindInternAtom,
118	131		// kindQueryExtension, kindXIDRange, kindBigreqEnable) get their consumers

 // completed. Treat it as a bug: Sync, then retry.
 var ErrRingFull = errors.New("xturbo: reply ring full")
+var errExpectUnsent = errors.New("xturbo: Expect of a sequence number not yet sent")
 var errReservedKind = errors.New("xturbo: reply kinds below 16 are reserved")
 const (
 // A reply without a registration is dropped (Stats.DroppedReplies); a
 // registered request that completes without reply or error frees its slot
 // silently once a later sequence number arrives. seq 0 (a failed request) and
-// requests already completed are ignored. ErrRingFull: more than 1024
+// requests already completed are ignored. A seq that was never sent (at or
+// beyond Buf.Next) is a caller bug and fails. ErrRingFull: more than 1024
 // registered requests outstanding.
 func (c *Conn) Expect(seq Seq, kind uint16, a0, a1, a2, a3 uint32) error {
 	if kind < kindUser {
 		return errReservedKind
 	}
+	if seq >= c.buf.Next() {
+		return errExpectUnsent
+	}
 	return c.expect(seq, kind, [4]uint32{a0, a1, a2, a3})
 }
 	return kind, args, true
 }
+// consumeError handles an error whose request was registered with a reserved
+// kind (1..15). It reports whether it consumed the error; an unconsumed error
+// is delivered to OnError. The bootstrap consumers arrive with the connection
+// bootstrap; until then every reserved-kind error is consumed and dropped,
+// like consumeReply drops its replies.
+func (c *Conn) consumeError(*XError) bool { return true }
 // consumeReply handles a reply registered with a reserved kind. kindDiscard
 // replies are dropped; the bootstrap kinds (kindInternAtom,
 // kindQueryExtension, kindXIDRange, kindBigreqEnable) get their consumers

M ring_test.go +22 -1


@@ -8,6 +8,22 @@ import (
8	8
9	9		const kindGeom = kindUser + 1
10	10
	11	+	func TestExpectUnsentSeq(t *testing.T) {
	12	+	c, _ := dispatchConn(t)
	13	+	c.buf.next = 5 // requests 1..4 were sent
	14	+	if err := c.Expect(4, kindGeom, 0, 0, 0, 0); err != nil {
	15	+	t.Fatal(err)
	16	+	}
	17	+	for _, seq := range []Seq{5, 6, 1 << 20} {
	18	+	if err := c.Expect(seq, kindGeom, 0, 0, 0, 0); !errors.Is(err, errExpectUnsent) {
	19	+	t.Fatalf("Expect(%d) = %v", seq, err)
	20	+	}
	21	+	}
	22	+	if c.live != 1 {
	23	+	t.Fatalf("live %d", c.live)
	24	+	}
	25	+	}
	26	+
11	27		func TestExpectReplyRouting(t *testing.T) {
12	28		c, srv := dispatchConn(t)
13	29		r := newRecorder(c)

@@ -167,6 +183,7 @@ func TestAllocExpect(t *testing.T) {
167	183		// delivered once with its kind and args; live count matches the model.
168	184		func FuzzReplyRing(f *testing.F) {
169	185		f.Add([]byte{0, 4, 1, 8, 2, 0xff, 3, 0x40})
	186	+	f.Add([]byte{0, 4, 8, 0xfd, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1})
170	187		f.Add([]byte{0x80, 0x84, 0x88, 1, 5, 9, 3})
171	188		f.Fuzz(func(t *testing.T, ops []byte) {
172	189		c, srv := dispatchConn(t)

@@ -182,6 +199,10 @@ func FuzzReplyRing(f *testing.F) {
182	199		wire := Seq(0) // last sequence sent by the "server"
183	200		for i, op := range ops {
184	201		step := Seq(op >> 2)
	202	+	jump := step // replies may leap past the ring and the 16-bit wrap
	203	+	if step >= 56 {
	204	+	jump = (step - 55) * 700
	205	+	}
185	206		switch op & 3 {
186	207		case 0: // Expect wire+1+step*13
187	208		seq := wire + 1 + step*13

@@ -199,7 +220,7 @@ func FuzzReplyRing(f *testing.F) {
199	220		model[seq] = uint32(i)
200	221		}
201	222		default: // a reply, error or event for wire+step
202		-	seq := wire + step
	223	+	seq := wire + jump
203	224		if seq == 0 {
204	225		seq = 1
205	226		}

 const kindGeom = kindUser + 1
+func TestExpectUnsentSeq(t *testing.T) {
+	c, _ := dispatchConn(t)
+	c.buf.next = 5 // requests 1..4 were sent
+	if err := c.Expect(4, kindGeom, 0, 0, 0, 0); err != nil {
+		t.Fatal(err)
+	}
+	for _, seq := range []Seq{5, 6, 1 << 20} {
+		if err := c.Expect(seq, kindGeom, 0, 0, 0, 0); !errors.Is(err, errExpectUnsent) {
+			t.Fatalf("Expect(%d) = %v", seq, err)
+		}
+	}
+	if c.live != 1 {
+		t.Fatalf("live %d", c.live)
+	}
+}
 func TestExpectReplyRouting(t *testing.T) {
 	c, srv := dispatchConn(t)
 	r := newRecorder(c)
 // delivered once with its kind and args; live count matches the model.
 func FuzzReplyRing(f *testing.F) {
 	f.Add([]byte{0, 4, 1, 8, 2, 0xff, 3, 0x40})
+	f.Add([]byte{0, 4, 8, 0xfd, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1})
 	f.Add([]byte{0x80, 0x84, 0x88, 1, 5, 9, 3})
 	f.Fuzz(func(t *testing.T, ops []byte) {
 		c, srv := dispatchConn(t)
 		wire := Seq(0) // last sequence sent by the "server"
 		for i, op := range ops {
 			step := Seq(op >> 2)
+			jump := step // replies may leap past the ring and the 16-bit wrap
+			if step >= 56 {
+				jump = (step - 55) * 700
+			}
 			switch op & 3 {
 			case 0: // Expect wire+1+step*13
 				seq := wire + 1 + step*13
 					model[seq] = uint32(i)
 				}
 			default: // a reply, error or event for wire+step
-				seq := wire + step
+				seq := wire + jump
 				if seq == 0 {
 					seq = 1
 				}