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