1 // Package chaindata is the wallet's channel to a light client: a binary frame protocol over
2 // a tiny-spawn duplex.
3 //
4 // It is the seam, and the seam is the whole design. The wallet sends a frame and gets a
5 // frame back; what sits behind the other end is nostrino's light client today, a full node or a
6 // fixture tomorrow, and neither side of the channel needs to know which. The vocabulary is
7 // the old client's own: status, sync, match, and a batched filter read.
8 //
9 // Three properties are load-bearing:
10 //
11 // - Frames are binary, not JSON. A filter set and a script list are byte strings; encoding
12 // them as base64 would grow the payload and make every message a parsing problem. The
13 // codec is a fixed 16-byte header and a payload, and the payload of a filter read is one
14 // buffer that the browser transport can transfer rather than copy.
15 // - Every request carries an id and every answer echoes it. The wallet may have one
16 // request in flight and a cancel behind it, so an answer with the wrong id is a bug
17 // rather than an answer.
18 // - A close is a frame, and it also means the far end died. A worker that is terminated
19 // without a word closes the port; the domain treats that exactly like a close frame, so
20 // an in-flight request fails instead of waiting for an answer that cannot come.
21 package chaindata
22 23 import (
24 "encoding/binary"
25 "errors"
26 "fmt"
27 "sort"
28 "sync"
29 "time"
30 31 "git.smesh.lol/tiny-spawn"
32 )
33 34 // Frame layout constants. The header is fixed so a decoder can reject anything it does not
35 // understand before it allocates.
36 const (
37 // Magic is 'N','C','D','1' read little-endian.
38 Magic uint32 = 0x3144434e
39 // Version is the frame format. A mismatch is refused rather than guessed at.
40 Version uint8 = 1
41 // HeaderLen is the fixed part of every frame:
42 //
43 // 0..4 magic 4 version 5 kind 6..8 flags
44 // 8..12 id 12..14 method 14..16 status 16..20 payload length
45 //
46 // The length is in the header so a decoder rejects a truncated frame before it
47 // allocates the payload.
48 HeaderLen = 20
49 // MaxPayload bounds one frame. A filter range is the largest thing that crosses this
50 // channel, and a batch is a few hundred kilobytes.
51 MaxPayload = 8 << 20
52 // MaxScripts and MaxScriptLen bound a match request before anything is allocated.
53 MaxScripts = 10000
54 MaxScriptLen = 10000
55 // MaxFilters bounds one filter read.
56 MaxFilters = 2000
57 )
58 59 // Kind is what a frame is for.
60 type Kind uint8
61 62 const (
63 // KindRequest asks the far end to do something.
64 KindRequest Kind = 0
65 // KindResult answers a request.
66 KindResult Kind = 1
67 // KindCancel asks the far end to abandon one request.
68 KindCancel Kind = 2
69 // KindClose says this end is going away. The far end must finish or fail what it is
70 // doing and stop.
71 KindClose Kind = 3
72 )
73 74 // String renders a kind for a log or an error.
75 func (k Kind) String() string {
76 switch k {
77 case KindRequest:
78 return "request"
79 case KindResult:
80 return "result"
81 case KindCancel:
82 return "cancel"
83 case KindClose:
84 return "close"
85 }
86 87 return fmt.Sprintf("kind(%d)", uint8(k))
88 }
89 90 // Method is one command a request may carry.
91 type Method uint16
92 93 const (
94 // MethodStatus reports the chain position.
95 MethodStatus Method = 1
96 // MethodSync advances headers and filters and reports the new position.
97 MethodSync Method = 2
98 // MethodMatch returns the heights whose filters mention any of the scripts.
99 MethodMatch Method = 3
100 // MethodBlock returns one raw block, which is what turns a matched height into coins.
101 MethodBlock Method = 5
102 // MethodBroadcast sends one raw transaction to the peer the light client follows, so a
103 // wallet that signs locally has somewhere to hand the result.
104 MethodBroadcast Method = 6
105 // MethodFilters returns a contiguous run of compact filters, one payload, so a wallet
106 // that needs them fetches a batch rather than a round trip per block.
107 MethodFilters Method = 4
108 )
109 110 // String renders a method.
111 func (m Method) String() string {
112 switch m {
113 case MethodStatus:
114 return "status"
115 case MethodSync:
116 return "sync"
117 case MethodMatch:
118 return "match"
119 case MethodBlock:
120 return "block"
121 case MethodBroadcast:
122 return "broadcast"
123 case MethodFilters:
124 return "filters"
125 }
126 127 return fmt.Sprintf("method(%d)", uint16(m))
128 }
129 130 // Frame is one message.
131 //
132 // Payload is the only variable part. It is a slice rather than a string because it is
133 // usually not text: scripts, filters, heights and status words are binary.
134 type Frame struct {
135 Kind Kind
136 // ID correlates a result or a cancel with its request. Zero is not used for requests.
137 ID uint32
138 // Method is set on requests and on results, so an answer says what it answers even if
139 // its id is all a strict caller checks.
140 Method Method
141 // Status is zero on success and a small code on failure. It is a code rather than a
142 // string so a transport never has to parse an error.
143 Status Status
144 // Payload carries the method's arguments or its answer.
145 Payload []byte
146 }
147 148 // Status is a result code.
149 type Status uint16
150 151 const (
152 // StatusOK is a successful result.
153 StatusOK Status = 0
154 // StatusFailed is a refusal the source reported.
155 StatusFailed Status = 1
156 // StatusCanceled means the request was cancelled before it finished.
157 StatusCanceled Status = 2
158 // StatusUnknownMethod is an unsupported method.
159 StatusUnknownMethod Status = 3
160 // StatusMalformed means the payload could not be read.
161 StatusMalformed Status = 4
162 // StatusClosed means the far end went away.
163 StatusClosed Status = 5
164 )
165 166 // String renders a status.
167 func (s Status) String() string {
168 switch s {
169 case StatusOK:
170 return "ok"
171 case StatusFailed:
172 return "failed"
173 case StatusCanceled:
174 return "canceled"
175 case StatusUnknownMethod:
176 return "unknown method"
177 case StatusMalformed:
178 return "malformed"
179 case StatusClosed:
180 return "closed"
181 }
182 183 return fmt.Sprintf("status(%d)", uint16(s))
184 }
185 186 // ErrPayloadTooLarge is returned by Encode when a payload exceeds MaxPayload.
187 var ErrPayloadTooLarge = errors.New("chaindata: payload is larger than one frame")
188 189 // ErrBadFrame is returned by Decode for anything that is not a frame this version reads.
190 var ErrBadFrame = errors.New("chaindata: not a chaindata frame")
191 192 // Encode writes a frame.
193 //
194 // The header is written first and the payload copied once, so a batch of filters is one
195 // allocation and one pass rather than an encoder walking the items.
196 func Encode(frame Frame) ([]byte, error) {
197 if len(frame.Payload) > MaxPayload {
198 return nil, fmt.Errorf("%w: %d bytes", ErrPayloadTooLarge, len(frame.Payload))
199 }
200 201 out := make([]byte, HeaderLen+len(frame.Payload))
202 203 binary.LittleEndian.PutUint32(out[0:4], Magic)
204 out[4] = Version
205 out[5] = byte(frame.Kind)
206 binary.LittleEndian.PutUint16(out[6:8], 0)
207 binary.LittleEndian.PutUint32(out[8:12], frame.ID)
208 binary.LittleEndian.PutUint16(out[12:14], uint16(frame.Method))
209 binary.LittleEndian.PutUint16(out[14:16], uint16(frame.Status))
210 binary.LittleEndian.PutUint32(out[16:20], uint32(len(frame.Payload)))
211 212 copy(out[HeaderLen:], frame.Payload)
213 214 return out, nil
215 }
216 217 // Decode reads a frame and refuses anything that is not one.
218 func Decode(raw []byte) (Frame, error) {
219 if len(raw) < HeaderLen {
220 return Frame{}, fmt.Errorf("%w: %d bytes is shorter than a header", ErrBadFrame,
221 len(raw))
222 }
223 224 if magic := binary.LittleEndian.Uint32(raw[0:4]); magic != Magic {
225 return Frame{}, fmt.Errorf("%w: wrong magic", ErrBadFrame)
226 }
227 228 if version := raw[4]; version != Version {
229 return Frame{}, fmt.Errorf("%w: version %d", ErrBadFrame, version)
230 }
231 232 kind := Kind(raw[5])
233 234 if kind > KindClose {
235 return Frame{}, fmt.Errorf("%w: %s", ErrBadFrame, kind)
236 }
237 238 length := binary.LittleEndian.Uint32(raw[16:20])
239 240 if length > MaxPayload {
241 return Frame{}, fmt.Errorf("%w: %d-byte payload", ErrBadFrame, length)
242 }
243 244 if uint64(len(raw)) != uint64(HeaderLen)+uint64(length) {
245 return Frame{}, fmt.Errorf("%w: %d bytes of payload, not %d", ErrBadFrame,
246 len(raw)-HeaderLen, length)
247 }
248 249 payload := make([]byte, length)
250 copy(payload, raw[HeaderLen:])
251 252 return Frame{
253 Kind: kind,
254 ID: binary.LittleEndian.Uint32(raw[8:12]),
255 Method: Method(binary.LittleEndian.Uint16(raw[12:14])),
256 Status: Status(binary.LittleEndian.Uint16(raw[14:16])),
257 Payload: payload,
258 }, nil
259 }
260 261 //
262 // Payload codecs. Each one reads and writes exactly one shape, with the count read before
263 // anything is allocated from it.
264 //
265 266 // Position is the chain position a status reports.
267 type Position struct {
268 // Tip is the highest verified header height.
269 Tip uint32
270 // Filters is the highest height whose filter is downloaded and verified.
271 Filters uint32
272 // Running reports whether a sync is in progress.
273 Running bool
274 }
275 276 // PutPosition writes a position: tip, filters, flags.
277 func PutPosition(buffer []byte, position Position) []byte {
278 buffer = binary.LittleEndian.AppendUint32(buffer, position.Tip)
279 buffer = binary.LittleEndian.AppendUint32(buffer, position.Filters)
280 281 flags := byte(0)
282 283 if position.Running {
284 flags = 1
285 }
286 287 return append(buffer, flags)
288 }
289 290 // ReadPosition reads a position.
291 func ReadPosition(payload []byte) (Position, error) {
292 if len(payload) < 9 {
293 return Position{}, fmt.Errorf("%w: a position is 9 bytes", ErrBadFrame)
294 }
295 296 return Position{
297 Tip: binary.LittleEndian.Uint32(payload[0:4]),
298 Filters: binary.LittleEndian.Uint32(payload[4:8]),
299 Running: payload[8]&1 == 1,
300 }, nil
301 }
302 303 // PutHeight writes the four-byte height a block request carries.
304 func PutHeight(buffer []byte, height uint32) []byte {
305 return binary.LittleEndian.AppendUint32(buffer, height)
306 }
307 308 // ReadHeight reads a height, refusing anything that is not exactly four bytes.
309 func ReadHeight(payload []byte) (uint32, error) {
310 if len(payload) != 4 {
311 return 0, fmt.Errorf("%w: a height is 4 bytes, not %d", ErrBadFrame, len(payload))
312 }
313 314 return binary.LittleEndian.Uint32(payload), nil
315 }
316 317 // PutHeights writes a list of heights.
318 func PutHeights(buffer []byte, heights []uint32) []byte {
319 buffer = binary.LittleEndian.AppendUint32(buffer, uint32(len(heights)))
320 321 for _, height := range heights {
322 buffer = binary.LittleEndian.AppendUint32(buffer, height)
323 }
324 325 return buffer
326 }
327 328 // ReadHeights reads a list of heights.
329 func ReadHeights(payload []byte) ([]uint32, error) {
330 if len(payload) < 4 {
331 return nil, fmt.Errorf("%w: a height list starts with a count", ErrBadFrame)
332 }
333 334 count := binary.LittleEndian.Uint32(payload[0:4])
335 336 if count > MaxFilters*8 {
337 return nil, fmt.Errorf("%w: %d heights", ErrBadFrame, count)
338 }
339 340 if uint64(len(payload)) != 4+uint64(count)*4 {
341 return nil, fmt.Errorf("%w: %d heights need %d bytes, not %d", ErrBadFrame,
342 count, 4+count*4, len(payload))
343 }
344 345 out := make([]uint32, count)
346 347 for i := uint32(0); i < count; i++ {
348 out[i] = binary.LittleEndian.Uint32(payload[4+i*4:])
349 }
350 351 return out, nil
352 }
353 354 // PutScripts writes a match request: a count, then each length and its bytes.
355 func PutScripts(buffer []byte, scripts [][]byte) ([]byte, error) {
356 if len(scripts) == 0 {
357 return nil, fmt.Errorf("a match needs at least one script")
358 }
359 360 if len(scripts) > MaxScripts {
361 return nil, fmt.Errorf("%d scripts is more than one match", len(scripts))
362 }
363 364 buffer = binary.LittleEndian.AppendUint32(buffer, uint32(len(scripts)))
365 366 for _, script := range scripts {
367 if len(script) == 0 || len(script) > MaxScriptLen {
368 return nil, fmt.Errorf("a script is 1 to %d bytes, not %d", MaxScriptLen,
369 len(script))
370 }
371 372 buffer = binary.LittleEndian.AppendUint32(buffer, uint32(len(script)))
373 buffer = append(buffer, script...)
374 }
375 376 return buffer, nil
377 }
378 379 // ReadScripts reads a match request.
380 func ReadScripts(payload []byte) ([][]byte, error) {
381 if len(payload) < 4 {
382 return nil, fmt.Errorf("%w: a match starts with a count", ErrBadFrame)
383 }
384 385 count := binary.LittleEndian.Uint32(payload[0:4])
386 387 if count == 0 || count > MaxScripts {
388 return nil, fmt.Errorf("%w: %d scripts", ErrBadFrame, count)
389 }
390 391 out := make([][]byte, 0, count)
392 offset := 4
393 394 for i := uint32(0); i < count; i++ {
395 if offset+4 > len(payload) {
396 return nil, fmt.Errorf("%w: script %d has no length", ErrBadFrame, i)
397 }
398 399 length := binary.LittleEndian.Uint32(payload[offset:])
400 offset += 4
401 402 if length == 0 || length > MaxScriptLen {
403 return nil, fmt.Errorf("%w: script %d is %d bytes", ErrBadFrame, i, length)
404 }
405 406 if offset+int(length) > len(payload) {
407 return nil, fmt.Errorf("%w: script %d runs past the payload", ErrBadFrame, i)
408 }
409 410 script := make([]byte, length)
411 copy(script, payload[offset:offset+int(length)])
412 413 out = append(out, script)
414 offset += int(length)
415 }
416 417 if offset != len(payload) {
418 return nil, fmt.Errorf("%w: %d trailing bytes", ErrBadFrame, len(payload)-offset)
419 }
420 421 return out, nil
422 }
423 424 // PutFilterRange writes a filter read request: from, count.
425 func PutFilterRange(buffer []byte, from, count uint32) []byte {
426 buffer = binary.LittleEndian.AppendUint32(buffer, from)
427 428 return binary.LittleEndian.AppendUint32(buffer, count)
429 }
430 431 // ReadFilterRange reads a filter read request.
432 func ReadFilterRange(payload []byte) (from, count uint32, err error) {
433 if len(payload) != 8 {
434 return 0, 0, fmt.Errorf("%w: a filter range is 8 bytes", ErrBadFrame)
435 }
436 437 from = binary.LittleEndian.Uint32(payload[0:4])
438 count = binary.LittleEndian.Uint32(payload[4:8])
439 440 if count == 0 || count > MaxFilters {
441 return 0, 0, fmt.Errorf("%w: %d filters", ErrBadFrame, count)
442 }
443 444 return from, count, nil
445 }
446 447 // Filter is one compact filter and the block it belongs to.
448 type Filter struct {
449 // Height is the block height.
450 Height uint32
451 // BlockHash is the block's hash, 32 bytes.
452 BlockHash []byte
453 // Filter is the Golomb-coded set.
454 Filter []byte
455 }
456 457 // PutFilters writes a batch: a count, then each height, hash and filter.
458 //
459 // The whole batch is one payload, which is what the transport transfers instead of copying
460 // per filter.
461 func PutFilters(buffer []byte, filters []Filter) ([]byte, error) {
462 if len(filters) > MaxFilters {
463 return nil, fmt.Errorf("%d filters is more than one read", len(filters))
464 }
465 466 buffer = binary.LittleEndian.AppendUint32(buffer, uint32(len(filters)))
467 468 for i, filter := range filters {
469 if len(filter.BlockHash) != 32 {
470 return nil, fmt.Errorf("filter %d has a %d-byte block hash", i,
471 len(filter.BlockHash))
472 }
473 474 if len(filter.Filter) > MaxPayload {
475 return nil, fmt.Errorf("filter %d is larger than one frame", i)
476 }
477 478 buffer = binary.LittleEndian.AppendUint32(buffer, filter.Height)
479 buffer = append(buffer, filter.BlockHash...)
480 buffer = binary.LittleEndian.AppendUint32(buffer, uint32(len(filter.Filter)))
481 buffer = append(buffer, filter.Filter...)
482 }
483 484 return buffer, nil
485 }
486 487 // ReadFilters reads a batch of filters.
488 func ReadFilters(payload []byte) ([]Filter, error) {
489 if len(payload) < 4 {
490 return nil, fmt.Errorf("%w: a filter batch starts with a count", ErrBadFrame)
491 }
492 493 count := binary.LittleEndian.Uint32(payload[0:4])
494 495 if count > MaxFilters {
496 return nil, fmt.Errorf("%w: %d filters", ErrBadFrame, count)
497 }
498 499 out := make([]Filter, 0, count)
500 offset := 4
501 502 for i := uint32(0); i < count; i++ {
503 if offset+4+32+4 > len(payload) {
504 return nil, fmt.Errorf("%w: filter %d is truncated", ErrBadFrame, i)
505 }
506 507 height := binary.LittleEndian.Uint32(payload[offset:])
508 offset += 4
509 510 hash := make([]byte, 32)
511 copy(hash, payload[offset:offset+32])
512 offset += 32
513 514 length := binary.LittleEndian.Uint32(payload[offset:])
515 offset += 4
516 517 if length > MaxPayload || offset+int(length) > len(payload) {
518 return nil, fmt.Errorf("%w: filter %d is %d bytes", ErrBadFrame, i, length)
519 }
520 521 body := make([]byte, length)
522 copy(body, payload[offset:offset+int(length)])
523 524 offset += int(length)
525 526 out = append(out, Filter{Height: height, BlockHash: hash, Filter: body})
527 }
528 529 if offset != len(payload) {
530 return nil, fmt.Errorf("%w: %d trailing bytes", ErrBadFrame, len(payload)-offset)
531 }
532 533 sort.SliceStable(out, func(i, j int) bool { return out[i].Height < out[j].Height })
534 535 return out, nil
536 }
537 538 //
539 // The domain.
540 //
541 542 // Source is the chain behind the channel: a light client, a full node or a fixture.
543 //
544 // Every method takes a context, because every one of them can be cancelled: a sync of ten
545 // thousand filters is the longest thing a wallet does, and a person closing the page should
546 // not have to wait for it.
547 type Source interface {
548 // Position reports the current chain position.
549 Position(ctx Context) (Position, error)
550 // Sync advances the chain and reports where it ends.
551 Sync(ctx Context, from uint32) (Position, error)
552 // Match returns the ascending heights whose filters mention any of the scripts.
553 Match(ctx Context, scripts [][]byte) ([]uint32, error)
554 // Block returns the raw block at height.
555 Block(ctx Context, height uint32) ([]byte, error)
556 // Broadcast sends one raw transaction to the chain's peer.
557 Broadcast(ctx Context, raw []byte) error
558 // Filters returns a contiguous run of filters starting at from.
559 Filters(ctx Context, from, count uint32) ([]Filter, error)
560 }
561 562 // Context is a cancellation handle for one request.
563 //
564 // It is an interface rather than context.Context so this package carries no dependency on
565 // the standard context's tree, and it carries a channel rather than a flag so a source can
566 // select on it instead of polling: a sync that polls a boolean is a sync that cannot stop
567 // promptly, and a source bridging to context.Context needs the channel anyway.
568 type Context interface {
569 // Done is closed when the request is cancelled.
570 Done() <-chan struct{}
571 // Err explains a cancellation, or is empty.
572 Err() string
573 }
574 575 // cancelable is the context a domain hands its source for one request.
576 type cancelable struct {
577 done chan struct{}
578 once sync.Once
579 mu sync.Mutex
580 cancelled bool
581 reason string
582 }
583 584 // newCancelable builds an open context.
585 func newCancelable() *cancelable {
586 return &cancelable{done: make(chan struct{})}
587 }
588 589 // Done implements Context.
590 func (c *cancelable) Done() <-chan struct{} { return c.done }
591 592 // Err implements Context.
593 func (c *cancelable) Err() string {
594 c.mu.Lock()
595 defer c.mu.Unlock()
596 597 return c.reason
598 }
599 600 // cancel stops the request, once.
601 func (c *cancelable) cancel(reason string) {
602 c.once.Do(func() {
603 c.mu.Lock()
604 c.cancelled = true
605 c.reason = reason
606 c.mu.Unlock()
607 608 close(c.done)
609 })
610 }
611 612 // cancelled reports whether the request was cancelled.
613 func (c *cancelable) cancelledNow() bool {
614 select {
615 case <-c.done:
616 return true
617 default:
618 return false
619 }
620 }
621 622 // Data is what the domain needs at spawn.
623 type Data struct {
624 // Source is the chain to serve. Fixed at spawn like every other input.
625 Source Source
626 }
627 628 // Chans is the domain's fixed channel set.
629 type Chans struct {
630 // Link carries frames both ways.
631 Link tinyspawn.Duplex[[]byte]
632 // Stop cancels the domain and everything in flight.
633 Stop tinyspawn.Breaker
634 }
635 636 // Serve is the chain-data domain: an event loop over one duplex.
637 //
638 // Cancellation arrives while a request is in flight, so the frames are read by a small
639 // relay that owns the receive side and hands requests to this loop. The relay cancels a
640 // request's context directly, which is the one thing that must not wait for the loop: a sync
641 // that is being cancelled is by definition the one the loop is inside.
642 func Serve(data Data, chans Chans) {
643 requests := make(chan Frame, 4)
644 relayDone := make(chan struct{})
645 646 inFlight := ®istry{}
647 648 go relay(chans.Link, chans.Stop, requests, inFlight, relayDone)
649 650 for {
651 select {
652 case <-chans.Stop.Done():
653 return
654 655 case <-relayDone:
656 return
657 658 case frame, ok := <-requests:
659 if !ok {
660 return
661 }
662 663 answer := handle(data.Source, frame, inFlight)
664 665 select {
666 case chans.Link.Tx <- answer:
667 case <-chans.Stop.Done():
668 return
669 case <-relayDone:
670 // The peer is gone; there is nobody to answer.
671 return
672 }
673 }
674 }
675 }
676 677 // registry tracks the request currently in flight, so a cancel frame can reach it.
678 type registry struct {
679 mu sync.Mutex
680 current *cancelable
681 id uint32
682 }
683 684 // put registers a request.
685 func (r *registry) put(id uint32, ctx *cancelable) {
686 r.mu.Lock()
687 defer r.mu.Unlock()
688 689 r.current = ctx
690 r.id = id
691 }
692 693 // clear forgets a request.
694 func (r *registry) clear(id uint32) {
695 r.mu.Lock()
696 defer r.mu.Unlock()
697 698 if r.id == id {
699 r.current = nil
700 r.id = 0
701 }
702 }
703 704 // cancel stops the registered request if its id matches.
705 func (r *registry) cancel(id uint32, reason string) {
706 r.mu.Lock()
707 current, held := r.current, r.id
708 r.mu.Unlock()
709 710 if current != nil && held == id {
711 current.cancel(reason)
712 }
713 }
714 715 // relay reads frames and routes them, so cancellation is never blocked behind the work.
716 func relay(
717 link tinyspawn.Duplex[[]byte],
718 stop tinyspawn.Breaker,
719 requests chan<- Frame,
720 inFlight *registry,
721 done chan<- struct{},
722 ) {
723 defer close(done)
724 defer close(requests)
725 726 for {
727 select {
728 case <-stop.Done():
729 return
730 731 case raw, ok := <-link.Recv():
732 if !ok {
733 // The transaction port closed, which is what a terminated worker looks
734 // like. Everything in flight fails.
735 inFlight.cancel(0, "the channel closed")
736 737 return
738 }
739 740 frame, err := Decode(raw)
741 742 if err != nil {
743 // A frame this version cannot read is not a request. Answer it so the
744 // sender is not left waiting, but never block the relay on it.
745 select {
746 case link.Tx <- EncodeOrClose(StatusMalformed, err.Error()):
747 case <-stop.Done():
748 return
749 default:
750 }
751 752 continue
753 }
754 755 switch frame.Kind {
756 case KindCancel:
757 inFlight.cancel(frame.ID, "cancelled by the wallet")
758 759 case KindClose:
760 inFlight.cancel(0, "the wallet closed the channel")
761 762 return
763 764 case KindRequest:
765 requests <- frame
766 }
767 }
768 }
769 }
770 771 // encodeResult builds an answer frame.
772 func encodeResult(frame Frame, status Status, payload []byte, message string) []byte {
773 out := Frame{
774 Kind: KindResult, ID: frame.ID, Method: frame.Method, Status: status,
775 Payload: payload,
776 }
777 778 if status != StatusOK && message != "" {
779 out.Payload = append([]byte(message), 0)
780 }
781 782 encoded, err := Encode(out)
783 784 if err != nil {
785 // The only way this fails is an over-large payload, which the codecs bound before
786 // they build one; answer with the failure rather than sending nothing.
787 fallback, _ := Encode(Frame{
788 Kind: KindResult, ID: frame.ID, Method: frame.Method, Status: StatusMalformed,
789 Payload: []byte("chaindata: the answer was too large\x00"),
790 })
791 792 return fallback
793 }
794 795 return encoded
796 }
797 798 // EncodeOrClose is an answer with no request to answer, for a transport-level failure.
799 func EncodeOrClose(status Status, message string) []byte {
800 encoded, _ := Encode(Frame{
801 Kind: KindResult, Status: status, Payload: append([]byte(message), 0),
802 })
803 804 return encoded
805 }
806 807 // handle runs one request.
808 func handle(source Source, frame Frame, inFlight *registry) []byte {
809 if source == nil {
810 return encodeResult(frame, StatusFailed, nil, "this chain-data domain has no source")
811 }
812 813 ctx := newCancelable()
814 815 inFlight.put(frame.ID, ctx)
816 817 defer inFlight.clear(frame.ID)
818 819 switch frame.Method {
820 case MethodStatus:
821 position, err := source.Position(ctx)
822 823 return positionResult(frame, ctx, position, err)
824 825 case MethodSync:
826 if len(frame.Payload) != 4 {
827 return encodeResult(frame, StatusMalformed, nil, "sync takes a 4-byte height")
828 }
829 830 position, err := source.Sync(ctx, binary.LittleEndian.Uint32(frame.Payload))
831 832 return positionResult(frame, ctx, position, err)
833 834 case MethodMatch:
835 scripts, err := ReadScripts(frame.Payload)
836 837 if err != nil {
838 return encodeResult(frame, StatusMalformed, nil, err.Error())
839 }
840 841 heights, err := source.Match(ctx, scripts)
842 843 if ctx.cancelledNow() {
844 return encodeResult(frame, StatusCanceled, nil, ctx.Err())
845 }
846 847 if err != nil {
848 return encodeResult(frame, StatusFailed, nil, err.Error())
849 }
850 851 return encodeResult(frame, StatusOK, PutHeights(nil, heights), "")
852 853 case MethodBlock:
854 height, err := ReadHeight(frame.Payload)
855 856 if err != nil {
857 return encodeResult(frame, StatusMalformed, nil, err.Error())
858 }
859 860 raw, err := source.Block(ctx, height)
861 862 if ctx.cancelledNow() {
863 return encodeResult(frame, StatusCanceled, nil, ctx.Err())
864 }
865 866 if err != nil {
867 return encodeResult(frame, StatusFailed, nil, err.Error())
868 }
869 870 return encodeResult(frame, StatusOK, raw, "")
871 872 case MethodBroadcast:
873 if len(frame.Payload) == 0 {
874 return encodeResult(frame, StatusMalformed, nil, "broadcast takes one raw transaction")
875 }
876 877 if err := source.Broadcast(ctx, frame.Payload); err != nil {
878 if ctx.cancelledNow() {
879 return encodeResult(frame, StatusCanceled, nil, ctx.Err())
880 }
881 882 return encodeResult(frame, StatusFailed, nil, err.Error())
883 }
884 885 if ctx.cancelledNow() {
886 return encodeResult(frame, StatusCanceled, nil, ctx.Err())
887 }
888 889 return encodeResult(frame, StatusOK, nil, "")
890 891 case MethodFilters:
892 from, count, err := ReadFilterRange(frame.Payload)
893 894 if err != nil {
895 return encodeResult(frame, StatusMalformed, nil, err.Error())
896 }
897 898 filters, err := source.Filters(ctx, from, count)
899 900 if ctx.cancelledNow() {
901 return encodeResult(frame, StatusCanceled, nil, ctx.Err())
902 }
903 904 if err != nil {
905 return encodeResult(frame, StatusFailed, nil, err.Error())
906 }
907 908 payload, err := PutFilters(nil, filters)
909 910 if err != nil {
911 return encodeResult(frame, StatusMalformed, nil, err.Error())
912 }
913 914 return encodeResult(frame, StatusOK, payload, "")
915 }
916 917 return encodeResult(frame, StatusUnknownMethod, nil,
918 "unknown chain-data method: "+frame.Method.String())
919 }
920 921 // positionResult encodes a position, or the cancellation that stopped it.
922 func positionResult(frame Frame, ctx *cancelable, position Position, err error) []byte {
923 if ctx.cancelledNow() {
924 return encodeResult(frame, StatusCanceled, nil, ctx.Err())
925 }
926 927 if err != nil {
928 return encodeResult(frame, StatusFailed, nil, err.Error())
929 }
930 931 return encodeResult(frame, StatusOK, PutPosition(nil, position), "")
932 }
933 934 //
935 // The client: the wallet's end of the channel.
936 //
937 938 // Client is the wallet's end of a chain-data channel.
939 //
940 // It is one value a wallet's domain holds per chain. Answers arrive on a reader goroutine and
941 // are routed by request id, so a request can be cancelled and a close can fail everything in
942 // flight without the caller having to poll.
943 type Client struct {
944 // Name is the chain this client speaks to, for error messages.
945 Name string
946 947 link tinyspawn.Duplex[[]byte]
948 stop tinyspawn.Breaker
949 950 mu sync.Mutex
951 seq uint32
952 pending map[uint32]chan Frame
953 closed bool
954 reason string
955 // timeout is the deadline used when timeoutSet is true; otherwise each method's
956 // default applies.
957 timeout time.Duration
958 timeoutSet bool
959 }
960 961 // NewClient wraps one end of a duplex and starts its reader.
962 func NewClient(name string, link tinyspawn.Duplex[[]byte], stop tinyspawn.Breaker) *Client {
963 client := &Client{
964 Name: name, link: link, stop: stop, pending: map[uint32]chan Frame{},
965 timeout: DefaultRequestTimeout,
966 }
967 968 go client.reader()
969 970 return client
971 }
972 973 // DefaultRequestTimeout bounds how long a request waits for an answer.
974 //
975 // Without it a wallet whose chain channel never came up waits for ever: the page shows
976 // "opening the workers…" (or a spinner on the first status), nothing is logged anywhere, and a
977 // dead channel is indistinguishable from a slow one. A timeout turns that into a reason.
978 //
979 // It deliberately does not apply to Sync: a first sync is minutes of work in one request, and
980 // a deadline meant to catch a dead channel must not kill a live one that is making progress.
981 // Sync is bounded by the channel closing, the caller cancelling it, and the light client's own
982 // read timeout.
983 const DefaultRequestTimeout = 60 * time.Second
984 985 // ErrNoAnswer means the far end did not answer within the timeout.
986 var ErrNoAnswer = errors.New("chaindata: the far end did not answer")
987 988 // SetTimeout overrides the answer deadline for every method, including Sync. Zero restores the
989 // per-method defaults, which means Sync has no answer deadline again.
990 func (c *Client) SetTimeout(timeout time.Duration) {
991 c.mu.Lock()
992 defer c.mu.Unlock()
993 994 if timeout <= 0 {
995 c.timeout = DefaultRequestTimeout
996 c.timeoutSet = false
997 998 return
999 }
1000 1001 c.timeout = timeout
1002 c.timeoutSet = true
1003 }
1004 1005 // reader routes answers to the requests waiting for them.
1006 func (c *Client) reader() {
1007 for {
1008 select {
1009 case <-c.stop.Done():
1010 c.failAll("the wallet stopped")
1011 1012 return
1013 1014 case raw, ok := <-c.link.Recv():
1015 if !ok {
1016 c.failAll("the chain-data channel closed")
1017 1018 return
1019 }
1020 1021 frame, err := Decode(raw)
1022 1023 if err != nil {
1024 // An answer this version cannot read cannot be matched to a request. The
1025 // request it belonged to will time out with the caller's own deadline
1026 // rather than being answered with somebody else's answer.
1027 continue
1028 }
1029 1030 if frame.Kind == KindClose {
1031 c.failAll(reasonOf(frame, "the chain-data domain closed"))
1032 1033 return
1034 }
1035 1036 c.deliver(frame)
1037 }
1038 }
1039 }
1040 1041 // deliver hands one answer to its waiter, if there is one.
1042 func (c *Client) deliver(frame Frame) {
1043 c.mu.Lock()
1044 1045 waiter, ok := c.pending[frame.ID]
1046 1047 if ok {
1048 delete(c.pending, frame.ID)
1049 }
1050 1051 c.mu.Unlock()
1052 1053 if !ok {
1054 // An answer to a request nobody is waiting for: a cancel raced the answer, or the
1055 // request was abandoned. Dropping it is the only honest option.
1056 return
1057 }
1058 1059 select {
1060 case waiter <- frame:
1061 default:
1062 }
1063 }
1064 1065 // failAll fails everything in flight and refuses new requests.
1066 func (c *Client) failAll(reason string) {
1067 c.mu.Lock()
1068 1069 c.closed = true
1070 c.reason = reason
1071 1072 waiting := c.pending
1073 c.pending = map[uint32]chan Frame{}
1074 1075 c.mu.Unlock()
1076 1077 for _, waiter := range waiting {
1078 select {
1079 case waiter <- Frame{Kind: KindClose, Status: StatusClosed,
1080 Payload: append([]byte(reason), 0)}:
1081 default:
1082 }
1083 }
1084 }
1085 1086 // reasonOf reads the message a close frame carries.
1087 func reasonOf(frame Frame, fallback string) string {
1088 if len(frame.Payload) == 0 {
1089 return fallback
1090 }
1091 1092 message := string(frame.Payload)
1093 1094 return trimNul(message)
1095 }
1096 1097 // trimNul cuts the trailing NUL a message payload carries.
1098 func trimNul(text string) string {
1099 if n := len(text); n > 0 && text[n-1] == 0 {
1100 return text[:n-1]
1101 }
1102 1103 return text
1104 }
1105 1106 // Closed reports whether the channel is gone.
1107 func (c *Client) Closed() bool {
1108 c.mu.Lock()
1109 defer c.mu.Unlock()
1110 1111 return c.closed
1112 }
1113 1114 // Do sends one request and waits for its answer.
1115 //
1116 // The wait is bounded by the breaker and by the channel closing, not by a deadline: a caller
1117 // with a deadline should cancel the request, which sends a frame the far end acts on.
1118 func (c *Client) Do(frame Frame) (Frame, error) {
1119 c.mu.Lock()
1120 1121 if c.closed {
1122 reason := c.reason
1123 c.mu.Unlock()
1124 1125 return Frame{}, fmt.Errorf("%s: %s", c.Name, reason)
1126 }
1127 1128 c.seq++
1129 id := c.seq
1130 1131 waiter := make(chan Frame, 1)
1132 c.pending[id] = waiter
1133 1134 c.mu.Unlock()
1135 1136 frame.Kind = KindRequest
1137 frame.ID = id
1138 1139 encoded, err := Encode(frame)
1140 1141 if err != nil {
1142 c.mu.Lock()
1143 delete(c.pending, id)
1144 c.mu.Unlock()
1145 1146 return Frame{}, fmt.Errorf("%s: %w", c.Name, err)
1147 }
1148 1149 select {
1150 case c.link.Tx <- encoded:
1151 case <-c.stop.Done():
1152 c.mu.Lock()
1153 delete(c.pending, id)
1154 c.mu.Unlock()
1155 1156 return Frame{}, fmt.Errorf("%s: the wallet stopped", c.Name)
1157 }
1158 1159 // The deadline is computed once, for this method: a sync may legitimately take minutes,
1160 // while a status that takes a minute is a dead channel. A zero deadline leaves timeoutCh
1161 // nil, which disables that case in the select below.
1162 timeout := c.requestTimeout(frame.Method)
1163 1164 var timeoutCh <-chan time.Time
1165 1166 if timeout > 0 {
1167 timer := time.NewTimer(timeout)
1168 defer timer.Stop()
1169 1170 timeoutCh = timer.C
1171 }
1172 1173 select {
1174 case answer := <-waiter:
1175 if answer.Kind == KindClose {
1176 return Frame{}, fmt.Errorf("%s: %s", c.Name, reasonOf(answer, "closed"))
1177 }
1178 1179 if answer.Status != StatusOK {
1180 return answer, fmt.Errorf("%s: %s", c.Name, statusMessage(answer))
1181 }
1182 1183 return answer, nil
1184 1185 case <-timeoutCh:
1186 c.mu.Lock()
1187 delete(c.pending, id)
1188 c.mu.Unlock()
1189 1190 return Frame{}, fmt.Errorf(
1191 "%s: %w within %s: is the chain channel open?",
1192 c.Name, ErrNoAnswer, timeout.Round(time.Second),
1193 )
1194 1195 case <-c.stop.Done():
1196 c.mu.Lock()
1197 delete(c.pending, id)
1198 c.mu.Unlock()
1199 1200 return Frame{}, fmt.Errorf("%s: the wallet stopped", c.Name)
1201 }
1202 }
1203 1204 // requestTimeout is the answer deadline for one method, or zero for no deadline.
1205 //
1206 // An explicit SetTimeout applies to every method, Sync included. Otherwise only Sync is
1207 // unbounded: it is one long piece of work, and its liveness is the channel's own (a close, the
1208 // caller's cancellation, the light client's read timeout), not an answer deadline. A deadline
1209 // meant to catch a dead channel must not kill a live sync that is making progress.
1210 func (c *Client) requestTimeout(method Method) time.Duration {
1211 c.mu.Lock()
1212 defer c.mu.Unlock()
1213 1214 if c.timeoutSet {
1215 return c.timeout
1216 }
1217 1218 if method == MethodSync {
1219 return 0
1220 }
1221 1222 return DefaultRequestTimeout
1223 }
1224 1225 // Cancel asks the far end to abandon one request.
1226 //
1227 // It is best effort by design: the request may have finished already, and a cancel frame for
1228 // an answer that is on its way is dropped by the far end.
1229 func (c *Client) Cancel(id uint32) {
1230 encoded, err := Encode(Frame{Kind: KindCancel, ID: id})
1231 1232 if err != nil {
1233 return
1234 }
1235 1236 select {
1237 case c.link.Tx <- encoded:
1238 case <-c.stop.Done():
1239 default:
1240 }
1241 }
1242 1243 // CancelPending cancels everything in flight. The clear is done under the same lock a
1244 // finished request takes, so a cancel cannot be sent twice for one id.
1245 func (c *Client) CancelPending() {
1246 c.mu.Lock()
1247 ids := make([]uint32, 0, len(c.pending))
1248 1249 for id := range c.pending {
1250 ids = append(ids, id)
1251 }
1252 1253 c.mu.Unlock()
1254 1255 for _, id := range ids {
1256 c.Cancel(id)
1257 }
1258 }
1259 1260 // Close sends a close frame and stops the client.
1261 //
1262 // The frame is what tells the far end to finish; closing the breaker is what tells this end's
1263 // reader to stop. Both are needed: a domain that only stopped reading would leave the other
1264 // side working.
1265 func (c *Client) Close() {
1266 encoded, err := Encode(Frame{Kind: KindClose, Status: StatusClosed})
1267 1268 if err == nil {
1269 select {
1270 case c.link.Tx <- encoded:
1271 case <-c.stop.Done():
1272 default:
1273 }
1274 }
1275 1276 if c.stop != nil && !c.stop.Closed() {
1277 c.stop.Close()
1278 }
1279 }
1280 1281 // statusMessage reads the message a failed result carries.
1282 func statusMessage(frame Frame) string {
1283 if len(frame.Payload) == 0 {
1284 return frame.Status.String()
1285 }
1286 1287 return trimNul(string(frame.Payload))
1288 }
1289 1290 // Position asks the chain where it is.
1291 func (c *Client) Position() (Position, error) {
1292 answer, err := c.Do(Frame{Method: MethodStatus})
1293 1294 if err != nil {
1295 return Position{}, err
1296 }
1297 1298 return ReadPosition(answer.Payload)
1299 }
1300 1301 // Sync advances the chain from a height.
1302 func (c *Client) Sync(from uint32) (Position, error) {
1303 answer, err := c.Do(Frame{
1304 Method: MethodSync,
1305 Payload: binary.LittleEndian.AppendUint32(nil, from),
1306 })
1307 1308 if err != nil {
1309 return Position{}, err
1310 }
1311 1312 return ReadPosition(answer.Payload)
1313 }
1314 1315 // Match asks which heights mention any of the scripts.
1316 func (c *Client) Match(scripts [][]byte) ([]uint32, error) {
1317 payload, err := PutScripts(nil, scripts)
1318 1319 if err != nil {
1320 return nil, err
1321 }
1322 1323 answer, err := c.Do(Frame{Method: MethodMatch, Payload: payload})
1324 1325 if err != nil {
1326 return nil, err
1327 }
1328 1329 return ReadHeights(answer.Payload)
1330 }
1331 1332 // Filters reads a contiguous run of compact filters.
1333 func (c *Client) Filters(from, count uint32) ([]Filter, error) {
1334 answer, err := c.Do(Frame{
1335 Method: MethodFilters,
1336 Payload: PutFilterRange(nil, from, count),
1337 })
1338 1339 if err != nil {
1340 return nil, err
1341 }
1342 1343 return ReadFilters(answer.Payload)
1344 }
1345 1346 // Block reads one raw block by height.
1347 //
1348 // The bytes are unverified here on purpose: this is the transport, and the proof is the merkle
1349 // root in nostrino/pkg/block, checked by whoever knows which header the block is supposed to
1350 // belong to.
1351 func (c *Client) Block(height uint32) ([]byte, error) {
1352 answer, err := c.Do(Frame{Method: MethodBlock, Payload: PutHeight(nil, height)})
1353 1354 if err != nil {
1355 return nil, err
1356 }
1357 1358 return answer.Payload, nil
1359 }
1360 1361 // Broadcast hands one raw transaction to the chain.
1362 //
1363 // An empty transaction is refused here rather than sent, and a successful answer means the far
1364 // end sent the transaction to its peer; whether the transaction is in a mempool is a different
1365 // fact and has to be checked on chain.
1366 func (c *Client) Broadcast(raw []byte) error {
1367 if len(raw) == 0 {
1368 return errors.New("chaindata: refusing to broadcast an empty transaction")
1369 }
1370 1371 _, err := c.Do(Frame{Method: MethodBroadcast, Payload: raw})
1372 1373 return err
1374 }
1375