msgport_test.go raw
1 package msgport
2
3 import (
4 "errors"
5 "sync"
6 "testing"
7 "time"
8
9 "git.smesh.lol/nostrino/pkg/chaindata"
10 "git.smesh.lol/tiny-spawn"
11 )
12
13 // source is a chain that answers without a network, so the test is about the adapter.
14 type source struct {
15 mu sync.Mutex
16 position chaindata.Position
17 matched []uint32
18 scripts int
19 blockHeight uint32
20 }
21
22 func (s *source) Position(chaindata.Context) (chaindata.Position, error) {
23 s.mu.Lock()
24 defer s.mu.Unlock()
25
26 return s.position, nil
27 }
28
29 func (s *source) Sync(_ chaindata.Context, from uint32) (chaindata.Position, error) {
30 s.mu.Lock()
31 defer s.mu.Unlock()
32
33 s.position = chaindata.Position{Tip: from + 10, Filters: from + 10}
34
35 return s.position, nil
36 }
37
38 func (s *source) Match(_ chaindata.Context, scripts [][]byte) ([]uint32, error) {
39 s.mu.Lock()
40 defer s.mu.Unlock()
41
42 s.scripts = len(scripts)
43
44 return append([]uint32{}, s.matched...), nil
45 }
46
47 func (s *source) Block(_ chaindata.Context, height uint32) ([]byte, error) {
48 s.mu.Lock()
49 defer s.mu.Unlock()
50
51 s.blockHeight = height
52
53 raw := make([]byte, 128)
54 raw[0] = byte(height)
55
56 return raw, nil
57 }
58
59 func (s *source) Broadcast(_ chaindata.Context, raw []byte) error {
60 s.mu.Lock()
61 defer s.mu.Unlock()
62
63 s.blockHeight = uint32(len(raw))
64
65 return nil
66 }
67
68 func (s *source) Filters(_ chaindata.Context, from, count uint32) ([]chaindata.Filter, error) {
69 out := make([]chaindata.Filter, 0, count)
70
71 for offset := uint32(0); offset < count; offset++ {
72 hash := make([]byte, 32)
73 hash[0] = byte(offset)
74
75 out = append(out, chaindata.Filter{
76 Height: from + offset,
77 BlockHash: hash,
78 Filter: []byte{byte(offset)},
79 })
80 }
81
82 return out, nil
83 }
84
85 // pipePort is one end of a connected pair, standing in for a MessagePort.
86 type pipePort struct {
87 in chan []byte
88 peer *pipePort
89 mu sync.Mutex
90 dead bool
91 }
92
93 // Send delivers to the peer's receive channel.
94 func (p *pipePort) Send(frame []byte) error {
95 peer := p.peer
96
97 peer.mu.Lock()
98 defer peer.mu.Unlock()
99
100 if peer.dead {
101 return ErrClosed
102 }
103
104 select {
105 case peer.in <- frame:
106 return nil
107 default:
108 return errors.New("the pipe is full")
109 }
110 }
111
112 // Receive is the incoming frames.
113 func (p *pipePort) Receive() <-chan []byte { return p.in }
114
115 // Close ends this end and, as a terminated worker does, its peer with it.
116 func (p *pipePort) Close() error {
117 for _, end := range []*pipePort{p, p.peer} {
118 end.mu.Lock()
119
120 if !end.dead {
121 end.dead = true
122
123 close(end.in)
124 }
125
126 end.mu.Unlock()
127 }
128
129 return nil
130 }
131
132 // pipe returns two connected ports.
133 func pipe() (Port, Port) {
134 first := &pipePort{in: make(chan []byte, Buffer)}
135 second := &pipePort{in: make(chan []byte, Buffer)}
136
137 first.peer = second
138 second.peer = first
139
140 return first, second
141 }
142
143 // start serves a chain over one end and returns a client on the other.
144 func start(t *testing.T, chain *source) (*chaindata.Client, tinyspawn.Breaker, func()) {
145 t.Helper()
146
147 serverPort, walletPort := pipe()
148
149 serverStop := tinyspawn.NewBreaker()
150 served := make(chan struct{})
151
152 go func() {
153 defer close(served)
154
155 chaindata.Serve(chaindata.Data{Source: chain},
156 chaindata.Chans{Link: Duplex(serverPort), Stop: serverStop})
157 }()
158
159 clientStop := tinyspawn.NewBreaker()
160 client := chaindata.NewClient("btc", Duplex(walletPort), clientStop)
161
162 shutdown := func() {
163 client.Close()
164
165 if !serverStop.Closed() {
166 serverStop.Close()
167 }
168
169 select {
170 case <-served:
171 case <-time.After(5 * time.Second):
172 t.Error("the domain did not stop")
173 }
174 }
175
176 return client, clientStop, shutdown
177 }
178
179 // TestAFrameTravelsBothWays: status, sync and match work through ports, which is the whole
180 // claim of this package.
181 func TestAFrameTravelsBothWays(t *testing.T) {
182 chain := &source{position: chaindata.Position{Tip: 42, Filters: 40}, matched: []uint32{11, 12}}
183 client, _, shutdown := start(t, chain)
184
185 defer shutdown()
186
187 position, err := client.Position()
188
189 if err != nil {
190 t.Fatal(err)
191 }
192
193 if position.Tip != 42 || position.Filters != 40 {
194 t.Fatalf("the position is %+v", position)
195 }
196
197 position, err = client.Sync(7)
198
199 if err != nil {
200 t.Fatal(err)
201 }
202
203 if position.Tip != 17 || position.Filters != 17 {
204 t.Fatalf("the synced position is %+v", position)
205 }
206
207 matched, err := client.Match([][]byte{{0x00, 0x14}, {0x00, 0x14, 0x01}})
208
209 if err != nil {
210 t.Fatal(err)
211 }
212
213 if len(matched) != 2 || matched[0] != 11 || matched[1] != 12 {
214 t.Fatalf("the matches are %v", matched)
215 }
216
217 if chain.scripts != 2 {
218 t.Fatalf("the source was asked about %d scripts", chain.scripts)
219 }
220 }
221
222 // TestAFilterRangeArrivesAsOneFrame: the payload the browser transfers is the payload the
223 // wallet reads, with no per-filter message in between.
224 func TestAFilterRangeArrivesAsOneFrame(t *testing.T) {
225 chain := &source{}
226 client, _, shutdown := start(t, chain)
227
228 defer shutdown()
229
230 filters, err := client.Filters(100, 5)
231
232 if err != nil {
233 t.Fatal(err)
234 }
235
236 if len(filters) != 5 {
237 t.Fatalf("%d filters", len(filters))
238 }
239
240 for index, filter := range filters {
241 if filter.Height != uint32(100+index) {
242 t.Fatalf("filter %d is at height %d", index, filter.Height)
243 }
244 }
245 }
246
247 // TestManyRequestsInOrder: ids are matched to answers, not to arrival order.
248 func TestManyRequestsInOrder(t *testing.T) {
249 chain := &source{position: chaindata.Position{Tip: 3, Filters: 3}}
250 client, _, shutdown := start(t, chain)
251
252 defer shutdown()
253
254 for round := 0; round < 20; round++ {
255 position, err := client.Position()
256
257 if err != nil {
258 t.Fatalf("round %d: %v", round, err)
259 }
260
261 if position.Tip != 3 {
262 t.Fatalf("round %d: the position is %+v", round, position)
263 }
264 }
265 }
266
267 // TestAPortThatDiesFailsTheRequest: a worker termination has to arrive as an error, not as an
268 // answer that never comes.
269 func TestAPortThatDiesFailsTheRequest(t *testing.T) {
270 serverPort, walletPort := pipe()
271
272 chain := &source{position: chaindata.Position{Tip: 9, Filters: 9}}
273 serverStop := tinyspawn.NewBreaker()
274
275 go chaindata.Serve(chaindata.Data{Source: chain},
276 chaindata.Chans{Link: Duplex(serverPort), Stop: serverStop})
277
278 defer func() {
279 _ = serverPort.Close()
280
281 if !serverStop.Closed() {
282 serverStop.Close()
283 }
284 }()
285
286 clientStop := tinyspawn.NewBreaker()
287 client := chaindata.NewClient("btc", Duplex(walletPort), clientStop)
288
289 defer client.Close()
290
291 if _, err := client.Position(); err != nil {
292 t.Fatal(err)
293 }
294
295 // The worker is terminated: both ends of the port pair go away.
296 _ = walletPort.Close()
297
298 deadline := time.After(5 * time.Second)
299
300 for {
301 select {
302 case <-deadline:
303 t.Fatal("the client never noticed the port had died")
304 default:
305 }
306
307 if _, err := client.Position(); err != nil {
308 return
309 }
310
311 time.Sleep(5 * time.Millisecond)
312 }
313 }
314
315 // TestADeadServerStopsTheDomain: a closed port ends the serve loop instead of leaving it
316 // waiting for frames that cannot come.
317 func TestADeadServerStopsTheDomain(t *testing.T) {
318 serverPort, _ := pipe()
319
320 serverStop := tinyspawn.NewBreaker()
321 defer serverStop.Close()
322
323 served := make(chan struct{})
324
325 go func() {
326 defer close(served)
327
328 chaindata.Serve(chaindata.Data{Source: &source{}},
329 chaindata.Chans{Link: Duplex(serverPort), Stop: serverStop})
330 }()
331
332 _ = serverPort.Close()
333
334 select {
335 case <-served:
336 case <-time.After(5 * time.Second):
337 t.Fatal("the domain outlived its port")
338 }
339 }
340