client.go raw
1 package neutrino
2
3 import (
4 "context"
5 "errors"
6 "fmt"
7 "net"
8 "sync"
9 "time"
10
11 "git.smesh.lol/nostrino/pkg/btcwire"
12 "git.smesh.lol/nostrino/pkg/chain"
13 "git.smesh.lol/nostrino/pkg/crypto/chainhash"
14 "git.smesh.lol/nostrino/pkg/filter"
15 "git.smesh.lol/nostrino/pkg/p2p"
16 )
17
18 // Config configures a light client.
19 type Config struct {
20 // Profile is the chain being followed; required.
21 Profile *chain.Profile
22 // Store is the durable state; required.
23 Store Store
24 // Dial opens the transport. When nil, Address is dialed over TCP.
25 Dial func(ctx context.Context) (net.Conn, error)
26 // Address is a TCP P2P address used when Dial is nil.
27 Address string
28 // UserAgent is advertised in version. Empty means p2p.DefaultUserAgent.
29 UserAgent string
30 // HandshakeTimeout bounds connect and version/verack. Zero means 30s.
31 HandshakeTimeout time.Duration
32 // ReadTimeout bounds one frame read. Zero means 60s.
33 ReadTimeout time.Duration
34 // RequireCompactFilters refuses a peer without the BIP157 service bit.
35 RequireCompactFilters bool
36 // Logf receives diagnostics; nil discards them.
37 Logf func(string, ...any)
38 }
39
40 // Client is a compact-filter light client for one chain over one peer
41 // connection.
42 type Client struct {
43 cfg Config
44 store Store
45
46 connectMu sync.Mutex
47 mu sync.Mutex
48 peer *p2p.Conn
49 pending net.Conn
50 closed bool
51 // peerTip is the peer's last block height from version.
52 peerTip int32
53 }
54
55 // New validates cfg and returns a client. It does not connect.
56 func New(cfg Config) (*Client, error) {
57 if cfg.Profile == nil {
58 return nil, errors.New("neutrino: nil profile")
59 }
60 if err := cfg.Profile.Validate(); err != nil {
61 return nil, err
62 }
63 if cfg.Store == nil {
64 return nil, errors.New("neutrino: nil store")
65 }
66 if cfg.Dial == nil && cfg.Address == "" {
67 return nil, errors.New("neutrino: no address or dialer")
68 }
69 if cfg.HandshakeTimeout == 0 {
70 cfg.HandshakeTimeout = 30 * time.Second
71 }
72 if cfg.ReadTimeout == 0 {
73 cfg.ReadTimeout = 60 * time.Second
74 }
75 if cfg.Logf == nil {
76 cfg.Logf = func(string, ...any) {}
77 }
78 return &Client{cfg: cfg, store: cfg.Store}, nil
79 }
80
81 // MaxBlockPayload is the largest block this client will accept.
82 //
83 // It is comfortably above the 4 MB serialised maximum for a block and below the wire package's
84 // 32 MiB frame limit, so a real block always fits and a peer that claims otherwise is refused
85 // before its bytes are read.
86 const MaxBlockPayload = 8 << 20
87
88 // ErrNoHeader is a block requested at a height this store has no header for.
89 var ErrNoHeader = errors.New("neutrino: no header at that height")
90
91 // ErrBlockTooLarge is a block over the cap, or one the peer answered with something else.
92 var ErrBlockTooLarge = errors.New("neutrino: the block is larger than the cap")
93
94 // MaxTransactionPayload bounds a broadcast. Bitcoin policy caps a standard transaction well
95 // below this; the bound exists so a caller cannot hand the peer an arbitrary frame.
96 const MaxTransactionPayload = 1 << 20
97
98 // ErrTransactionTooLarge is a broadcast over the cap.
99 var ErrTransactionTooLarge = errors.New("neutrino: the transaction is larger than the cap")
100
101 // rawPayload is a payload whose bytes are already encoded.
102 //
103 // Broadcast sends a transaction exactly as the wallet built and signed it: re-encoding it
104 // through a decoder here would be a second place that could disagree with the signer.
105 type rawPayload []byte
106
107 // EncodePayload returns the bytes unchanged.
108 func (p rawPayload) EncodePayload() ([]byte, error) { return p, nil }
109
110 // DecodePayload is not part of what a broadcast does.
111 func (p rawPayload) DecodePayload([]byte) error {
112 return errors.New("neutrino: a raw payload is encode-only")
113 }
114
115 // Broadcast sends a raw transaction to the peer over the connected tunnel.
116 //
117 // A tx message has no reply, so a nil error means the frame reached the peer, not that the
118 // transaction was accepted into its mempool. A caller that needs the second fact must watch
119 // for the transaction; claiming it here would be a claim without evidence.
120 func (c *Client) Broadcast(ctx context.Context, raw []byte) error {
121 if len(raw) == 0 {
122 return errors.New("neutrino: refusing to broadcast an empty transaction")
123 }
124
125 if len(raw) > MaxTransactionPayload {
126 return fmt.Errorf(
127 "%w: transaction is %d bytes, maximum is %d",
128 ErrTransactionTooLarge, len(raw), MaxTransactionPayload,
129 )
130 }
131
132 c.mu.Lock()
133 peer := c.peer
134 c.mu.Unlock()
135
136 if peer == nil {
137 return errors.New("neutrino: not connected")
138 }
139
140 if err := peer.Send(btcwire.CmdTx, rawPayload(raw)); err != nil {
141 return fmt.Errorf("neutrino: tx: %w", err)
142 }
143
144 return nil
145 }
146
147 // Block fetches the raw block at a height.
148 //
149 // The height is resolved to a hash from the header store, so the request names what the wallet
150 // has already verified. The first 80 bytes of the answer must hash to that name: a peer can send
151 // any bytes it likes, and this is the cheap check before the expensive one in pkg/block.
152 func (c *Client) Block(ctx context.Context, height uint32) ([]byte, error) {
153 header, ok := c.store.Header(height)
154
155 if !ok {
156 return nil, fmt.Errorf("%w: %d", ErrNoHeader, height)
157 }
158
159 hash := header.BlockHash()
160
161 c.mu.Lock()
162 peer := c.peer
163 c.mu.Unlock()
164
165 if peer == nil {
166 return nil, errors.New("neutrino: not connected")
167 }
168
169 request := &btcwire.GetData{InvList: []btcwire.InvVect{{Type: btcwire.InvVectTypeBlock, Hash: hash}}}
170
171 if err := peer.Send(btcwire.CmdGetData, request); err != nil {
172 return nil, fmt.Errorf("neutrino: getdata: %w", err)
173 }
174
175 message, err := peer.ReadUntilMax(ctx, btcwire.CmdBlock, MaxBlockPayload)
176
177 if err != nil {
178 return nil, fmt.Errorf("neutrino: reading block %d: %w", height, err)
179 }
180
181 if len(message.Payload) > MaxBlockPayload {
182 return nil, fmt.Errorf("%w: %d bytes", ErrBlockTooLarge, len(message.Payload))
183 }
184
185 if len(message.Payload) < btcwire.BlockHeaderSize {
186 return nil, fmt.Errorf("%w: %d bytes is not a block", ErrBlockTooLarge, len(message.Payload))
187 }
188
189 var got btcwire.BlockHeader
190
191 if err := got.Deserialize(message.Payload[:btcwire.BlockHeaderSize]); err != nil {
192 return nil, fmt.Errorf("neutrino: block %d header: %w", height, err)
193 }
194
195 if got.BlockHash() != hash {
196 return nil, fmt.Errorf("%w: the peer sent %v, not %v", ErrBlockTooLarge, got.BlockHash(), hash)
197 }
198
199 return message.Payload, nil
200 }
201
202 // Store returns the client's store.
203 func (c *Client) Store() Store { return c.store }
204
205 // Profile returns the chain profile.
206 func (c *Client) Profile() *chain.Profile { return c.cfg.Profile }
207
208 // Connect dials and handshakes if not already connected. Close can interrupt
209 // a Connect in progress: it closes the pending socket, not just the finished
210 // peer.
211 func (c *Client) Connect(ctx context.Context) error {
212 c.connectMu.Lock()
213 defer c.connectMu.Unlock()
214 c.mu.Lock()
215 if c.peer != nil {
216 c.mu.Unlock()
217 return nil
218 }
219 if c.closed {
220 c.mu.Unlock()
221 return errors.New("neutrino: closed")
222 }
223 c.mu.Unlock()
224 if err := c.ensureAnchor(); err != nil {
225 return err
226 }
227 tip, _, ok := c.store.HeaderTip()
228 var last int32
229 if ok {
230 last = int32(tip)
231 }
232 var conn net.Conn
233 var err error
234 if c.cfg.Dial != nil {
235 conn, err = c.cfg.Dial(ctx)
236 if err != nil {
237 return fmt.Errorf("neutrino: dial: %w", err)
238 }
239 } else {
240 var d net.Dialer
241 conn, err = d.DialContext(ctx, "tcp", c.cfg.Address)
242 if err != nil {
243 return fmt.Errorf("neutrino: dial %s: %w", c.cfg.Address, err)
244 }
245 }
246 c.mu.Lock()
247 if c.closed {
248 c.mu.Unlock()
249 conn.Close()
250 return errors.New("neutrino: closed")
251 }
252 c.pending = conn
253 c.mu.Unlock()
254
255 peer := p2p.New(conn, p2p.Config{
256 Magic: c.cfg.Profile.Magic,
257 UserAgent: c.cfg.UserAgent,
258 LastBlock: last,
259 ReadTimeout: c.cfg.ReadTimeout,
260 RequireCompactFilters: c.cfg.RequireCompactFilters,
261 Logf: c.cfg.Logf,
262 })
263 hctx, cancel := context.WithTimeout(ctx, c.cfg.HandshakeTimeout)
264 defer cancel()
265 version, err := peer.Handshake(hctx)
266 if err != nil {
267 c.mu.Lock()
268 c.pending = nil
269 c.mu.Unlock()
270 conn.Close()
271 return fmt.Errorf("neutrino: handshake: %w", err)
272 }
273 c.mu.Lock()
274 if c.closed {
275 c.mu.Unlock()
276 peer.Close()
277 return errors.New("neutrino: closed")
278 }
279 c.pending = nil
280 c.peer = peer
281 c.peerTip = version.LastBlock
282 c.mu.Unlock()
283 c.cfg.Logf("neutrino: peer %s protocol=%d services=%#x tip=%d user=%s",
284 c.cfg.Profile.Name, version.ProtocolVersion, version.Services, version.LastBlock, version.UserAgent)
285 return nil
286 }
287
288 // ensureAnchor installs the trusted starting point on a fresh store. A
289 // profile checkpoint pins both the block and its filter header, which is what
290 // makes a first mainnet sync small; without one, the anchor is genesis.
291 func (c *Client) ensureAnchor() error {
292 if _, ok := c.store.Anchor(); ok {
293 return nil
294 }
295 if h := c.cfg.Profile.GenesisHeader.BlockHash(); h != c.cfg.Profile.GenesisHash {
296 return errors.New("neutrino: profile genesis header does not match its hash")
297 }
298 anchor := Anchor{Height: 0, BlockHash: c.cfg.Profile.GenesisHash, FilterHeight: -1}
299 if cp := c.cfg.Profile.Checkpoint; cp != nil {
300 anchor = Anchor{
301 Height: cp.Height,
302 BlockHash: cp.BlockHash,
303 FilterHeight: int32(cp.Height),
304 FilterHeader: cp.FilterHeader,
305 }
306 }
307 if anchor.BlockHash == (chainhash.Hash{}) {
308 return errors.New("neutrino: profile has no usable anchor")
309 }
310 if err := c.store.SetAnchor(anchor); err != nil {
311 return fmt.Errorf("neutrino: set anchor: %w", err)
312 }
313 if c.cfg.Profile.Checkpoint == nil {
314 if _, ok := c.store.Header(0); !ok {
315 if err := c.store.PutHeaders(0, []btcwire.BlockHeader{c.cfg.Profile.GenesisHeader}); err != nil {
316 return fmt.Errorf("neutrino: seed genesis header: %w", err)
317 }
318 }
319 }
320 return nil
321 }
322
323 // Close ends the connection, including a dial or handshake in progress.
324 func (c *Client) Close() error {
325 c.mu.Lock()
326 c.closed = true
327 pending := c.pending
328 peer := c.peer
329 c.pending = nil
330 c.peer = nil
331 c.mu.Unlock()
332 if pending != nil {
333 pending.Close()
334 }
335 if peer != nil {
336 return peer.Close()
337 }
338 return nil
339 }
340
341 // SelfPeer reports the peer's advertised tip from the handshake.
342 func (c *Client) PeerTip() int32 {
343 c.mu.Lock()
344 defer c.mu.Unlock()
345 return c.peerTip
346 }
347
348 // Result summarises a sync.
349 type Result struct {
350 StartHeight uint32
351 TipHeight uint32
352 HeadersAdded int
353 FilterHeadersAdded int
354 FiltersAdded int
355 FilterBytes int
356 Elapsed time.Duration
357 }
358
359 // Sync runs header, filter header and filter sync to the peer's tip.
360 func (c *Client) Sync(ctx context.Context) (*Result, error) {
361 return c.SyncFrom(ctx, 0, 0)
362 }
363
364 // SyncFrom is Sync with a first filter height and a filter count bound. A
365 // wallet that scans only recent history passes its birth height here; header
366 // and filter-header verification is always complete.
367 func (c *Client) SyncFrom(ctx context.Context, from uint32, max int) (*Result, error) {
368 start := time.Now()
369 res := &Result{}
370 if err := c.Connect(ctx); err != nil {
371 return res, err
372 }
373 if first, _, ok := c.store.HeaderTip(); ok {
374 res.StartHeight = first
375 }
376 n, err := c.SyncHeaders(ctx)
377 if err != nil {
378 return res, err
379 }
380 res.HeadersAdded = n
381 m, err := c.SyncFilterHeaders(ctx)
382 if err != nil {
383 return res, err
384 }
385 res.FilterHeadersAdded = m
386 k, bytes, err := c.SyncFilters(ctx, from, max)
387 if err != nil {
388 return res, err
389 }
390 res.FiltersAdded = k
391 res.FilterBytes = bytes
392 tip, _, _ := c.store.HeaderTip()
393 res.TipHeight = tip
394 res.Elapsed = time.Since(start)
395 return res, nil
396 }
397
398 // SyncHeaders requests headers until the peer's advertised tip is reached or
399 // the peer has nothing more. It returns the number of headers added.
400 func (c *Client) SyncHeaders(ctx context.Context) (int, error) {
401 c.mu.Lock()
402 peer := c.peer
403 peerTip := c.peerTip
404 c.mu.Unlock()
405 if peer == nil {
406 return 0, errors.New("neutrino: not connected")
407 }
408 if err := c.ensureAnchor(); err != nil {
409 return 0, err
410 }
411 added := 0
412 reorgs := 0
413 for {
414 tip, tipHash, ok := c.store.HeaderTip()
415 if !ok {
416 return added, errors.New("neutrino: no header tip")
417 }
418 req := &btcwire.GetHeaders{
419 ProtocolVersion: btcwire.ProtocolVersion,
420 LocatorHashes: c.locator(tip),
421 StopHash: chainhash.Hash{},
422 }
423 if err := peer.Send(btcwire.CmdGetHeaders, req); err != nil {
424 return added, err
425 }
426 msg, err := peer.ReadUntil(ctx, btcwire.CmdHeaders)
427 if err != nil {
428 return added, fmt.Errorf("neutrino: headers: %w", err)
429 }
430 var resp btcwire.Headers
431 if err := resp.DecodePayload(msg.Payload); err != nil {
432 return added, fmt.Errorf("neutrino: decode headers: %w", err)
433 }
434 if len(resp.Headers) == 0 {
435 break
436 }
437 base := tip
438 prev := tipHash
439 if resp.Headers[0].PrevBlock != tipHash {
440 forkHeight, ok := c.findFork(resp.Headers[0].PrevBlock, tip)
441 if !ok {
442 return added, fmt.Errorf("neutrino: peer chain does not connect to the stored chain")
443 }
444 reorgs++
445 if reorgs > 5 {
446 return added, errors.New("neutrino: too many reorg attempts")
447 }
448 if err := c.store.Truncate(forkHeight + 1); err != nil {
449 return added, fmt.Errorf("neutrino: reorg rollback: %w", err)
450 }
451 c.cfg.Logf("neutrino: reorg at height %d, rolling back from %d", forkHeight, tip)
452 base = forkHeight
453 prev = resp.Headers[0].PrevBlock
454 }
455 for i := range resp.Headers {
456 if err := c.validateHeader(base+uint32(i)+1, &resp.Headers[i], prev); err != nil {
457 return added, err
458 }
459 prev = resp.Headers[i].BlockHash()
460 }
461 if err := c.store.PutHeaders(base+1, resp.Headers); err != nil {
462 return added, err
463 }
464 added += len(resp.Headers)
465 newTip := base + uint32(len(resp.Headers))
466 if peerTip >= 0 && int64(newTip) >= int64(peerTip) {
467 break
468 }
469 if len(resp.Headers) < btcwire.MaxBlockHeadersPerMsg {
470 break
471 }
472 }
473 return added, nil
474 }
475
476 // findFork walks back from tip looking for hash. It is bounded: a reorg deeper
477 // than the bound is refused rather than scanning an entire mainnet history.
478 func (c *Client) findFork(hash chainhash.Hash, tip uint32) (uint32, bool) {
479 anchor, hasAnchor := c.store.Anchor()
480 const maxScan = 10000
481 for i := 0; i < maxScan; i++ {
482 if tip < uint32(i) {
483 return 0, false
484 }
485 h := tip - uint32(i)
486 if hasAnchor && h < anchor.Height {
487 return 0, false
488 }
489 if hasAnchor && h == anchor.Height {
490 return h, hash == anchor.BlockHash
491 }
492 hdr, ok := c.store.Header(h)
493 if !ok {
494 if hasAnchor && h == anchor.Height+1 {
495 return anchor.Height, hash == anchor.BlockHash
496 }
497 return 0, false
498 }
499 if hdr.BlockHash() == hash {
500 return h, true
501 }
502 }
503 return 0, false
504 }
505
506 // locator builds a block locator from the stored chain, with ten single steps
507 // followed by exponentially growing gaps.
508 func (c *Client) locator(tip uint32) []chainhash.Hash {
509 var out []chainhash.Hash
510 anchor, hasAnchor := c.store.Anchor()
511 h := tip
512 step := uint32(1)
513 for {
514 if hasAnchor && h == anchor.Height {
515 out = append(out, anchor.BlockHash)
516 break
517 }
518 hdr, ok := c.store.Header(h)
519 if !ok {
520 if hasAnchor && h > anchor.Height {
521 h = anchor.Height
522 continue
523 }
524 break
525 }
526 out = append(out, hdr.BlockHash())
527 if h == 0 || len(out) >= btcwire.MaxLocatorHashesPerMsg {
528 break
529 }
530 if h < step {
531 h = 0
532 } else {
533 h -= step
534 }
535 if len(out) >= 10 {
536 step *= 2
537 }
538 }
539 return out
540 }
541
542 // filterHeaderStop returns the last height a getcfheaders request that starts
543 // at start may name, given the header tip. BIP157 caps one request at
544 // MaxFilterHashesPerMsg filter hashes, so a taller chain is walked in chunks.
545 func filterHeaderStop(start, tip uint32) uint32 {
546 if tip-start+1 > uint32(btcwire.MaxFilterHashesPerMsg) {
547 return start + uint32(btcwire.MaxFilterHashesPerMsg) - 1
548 }
549 return tip
550 }
551
552 // SyncFilterHeaders requests the compact filter header chain up to the stored
553 // block tip. It returns the number of filter headers added.
554 func (c *Client) SyncFilterHeaders(ctx context.Context) (int, error) {
555 c.mu.Lock()
556 peer := c.peer
557 c.mu.Unlock()
558 if peer == nil {
559 return 0, errors.New("neutrino: not connected")
560 }
561 added := 0
562 for {
563 tip, tipHash, ok := c.store.HeaderTip()
564 if !ok {
565 return added, errors.New("neutrino: no header tip")
566 }
567 start := uint32(0)
568 prev := chainhash.Hash{}
569 if fTip, fHash, ok := c.store.FilterHeaderTip(); ok {
570 start = fTip + 1
571 prev = fHash
572 }
573 if start > tip {
574 break
575 }
576 // One getcfheaders covers at most MaxFilterHashesPerMsg blocks: BIP157
577 // limits the range, and a peer refuses a wider request by dropping the
578 // connection. The chain is walked in as many requests as it takes.
579 stop := filterHeaderStop(start, tip)
580 stopHash := tipHash
581 if stop != tip {
582 stopHeader, ok := c.store.Header(stop)
583 if !ok {
584 return added, fmt.Errorf("neutrino: cfheaders stop block %d missing", stop)
585 }
586 stopHash = stopHeader.BlockHash()
587 }
588 req := &btcwire.GetCFHeaders{
589 FilterType: btcwire.FilterTypeBasic,
590 StartHeight: start,
591 StopHash: stopHash,
592 }
593 if err := peer.Send(btcwire.CmdGetCFHeaders, req); err != nil {
594 return added, err
595 }
596 msg, err := peer.ReadUntil(ctx, btcwire.CmdCFHeaders)
597 if err != nil {
598 return added, fmt.Errorf("neutrino: cfheaders: %w", err)
599 }
600 var resp btcwire.CFHeaders
601 if err := resp.DecodePayload(msg.Payload); err != nil {
602 return added, fmt.Errorf("neutrino: decode cfheaders: %w", err)
603 }
604 if resp.FilterType != btcwire.FilterTypeBasic {
605 return added, fmt.Errorf("neutrino: filter type %d in cfheaders", resp.FilterType)
606 }
607 if resp.PrevFilterHeader != prev {
608 return added, fmt.Errorf("neutrino: cfheaders previous filter header %s, want %s", resp.PrevFilterHeader, prev)
609 }
610 if len(resp.FilterHashes) == 0 {
611 break
612 }
613 end := start + uint32(len(resp.FilterHashes))
614 if end > tip+1 {
615 return added, fmt.Errorf("neutrino: cfheaders returned past the block tip")
616 }
617 endHeader, ok := c.store.Header(end - 1)
618 if !ok {
619 return added, fmt.Errorf("neutrino: cfheaders end block %d missing", end-1)
620 }
621 if resp.StopHash != endHeader.BlockHash() {
622 return added, fmt.Errorf("neutrino: cfheaders stop hash %s, want %s", resp.StopHash, endHeader.BlockHash())
623 }
624 chain := make([]chainhash.Hash, len(resp.FilterHashes))
625 cur := prev
626 for i, fh := range resp.FilterHashes {
627 cur = btcwire.FilterHeader(fh, cur)
628 chain[i] = cur
629 }
630 if err := c.store.PutFilterHeaders(start, resp.FilterHashes, chain); err != nil {
631 return added, err
632 }
633 added += len(chain)
634 }
635 return added, nil
636 }
637
638 // SyncFilters downloads and verifies compact filters. from is the first height
639 // to fetch (0 for the earliest stored), max bounds the count (0 for all). It
640 // returns the number of filters and their total bytes.
641 func (c *Client) SyncFilters(ctx context.Context, from uint32, max int) (int, int, error) {
642 c.mu.Lock()
643 peer := c.peer
644 c.mu.Unlock()
645 if peer == nil {
646 return 0, 0, errors.New("neutrino: not connected")
647 }
648 tip, _, ok := c.store.HeaderTip()
649 if !ok {
650 return 0, 0, errors.New("neutrino: no header tip")
651 }
652 fTip, hasFilter := c.store.FilterTip()
653 start := from
654 if hasFilter && fTip+1 > start {
655 start = fTip + 1
656 }
657 if a, ok := c.store.Anchor(); ok && a.FilterHeight >= 0 && start < uint32(a.FilterHeight)+1 {
658 // Filters below the checkpoint cannot be verified: only the filter
659 // header at the checkpoint itself is trusted, not the filter hashes
660 // before it.
661 start = uint32(a.FilterHeight) + 1
662 }
663 if start > tip {
664 return 0, 0, nil
665 }
666 batch := 500
667 if max > 0 && max < batch {
668 batch = max
669 }
670 count, bytes := 0, 0
671 for start <= tip {
672 end := start + uint32(batch) - 1
673 if end > tip {
674 end = tip
675 }
676 if max > 0 && count+int(end-start+1) > max {
677 end = start + uint32(max-count) - 1
678 }
679 stopHeader, ok := c.store.Header(end)
680 if !ok {
681 return count, bytes, fmt.Errorf("neutrino: filter range end block %d missing", end)
682 }
683 req := &btcwire.GetCFilters{
684 FilterType: btcwire.FilterTypeBasic,
685 StartHeight: start,
686 StopHash: stopHeader.BlockHash(),
687 }
688 if err := peer.Send(btcwire.CmdGetCFilters, req); err != nil {
689 return count, bytes, err
690 }
691 for h := start; h <= end; h++ {
692 msg, err := peer.ReadUntil(ctx, btcwire.CmdCFilter)
693 if err != nil {
694 return count, bytes, fmt.Errorf("neutrino: cfilter %d: %w", h, err)
695 }
696 var cf btcwire.CFilter
697 if err := cf.DecodePayload(msg.Payload); err != nil {
698 return count, bytes, fmt.Errorf("neutrino: decode cfilter %d: %w", h, err)
699 }
700 if cf.FilterType != btcwire.FilterTypeBasic {
701 return count, bytes, fmt.Errorf("neutrino: filter %d has type %d", h, cf.FilterType)
702 }
703 hdr, ok := c.store.Header(h)
704 if !ok {
705 return count, bytes, fmt.Errorf("neutrino: filter %d has no header", h)
706 }
707 blockHash := hdr.BlockHash()
708 if cf.BlockHash != blockHash {
709 return count, bytes, fmt.Errorf("neutrino: filter %d block hash %s, want %s", h, cf.BlockHash, blockHash)
710 }
711 filterHash := chainhash.DoubleHashH(cf.Filter)
712 expected, ok := c.store.FilterHash(h)
713 if !ok {
714 return count, bytes, fmt.Errorf("neutrino: filter %d has no committed filter hash", h)
715 }
716 if filterHash != expected {
717 return count, bytes, fmt.Errorf("neutrino: filter %d hash %s, filter header commits to %s", h, filterHash, expected)
718 }
719 if err := c.store.PutFilter(h, blockHash, filterHash, cf.Filter); err != nil {
720 return count, bytes, err
721 }
722 count++
723 bytes += len(cf.Filter)
724 }
725 start = end + 1
726 }
727 return count, bytes, nil
728 }
729
730 // MatchScripts scans stored filters in [from,to] (to 0 means the filter tip)
731 // for any of scripts, returning the matching block heights. It never contacts
732 // the peer; the match is local, so the gateway does not learn the wallet's
733 // script set.
734 func (c *Client) MatchScripts(from, to uint32, scripts [][]byte) ([]uint32, error) {
735 return MatchStoredScripts(c.store, from, to, scripts)
736 }
737
738 // MatchStoredScripts is MatchScripts without a client, for a caller that
739 // already holds a store (such as the wasm page).
740 func MatchStoredScripts(store Store, from, to uint32, scripts [][]byte) ([]uint32, error) {
741 if len(scripts) == 0 {
742 return nil, nil
743 }
744 tip, ok := store.FilterTip()
745 if !ok {
746 return nil, nil
747 }
748 if to == 0 || to > tip {
749 to = tip
750 }
751 var out []uint32
752 for h := from; h <= to; h++ {
753 f, ok := store.Filter(h)
754 if !ok {
755 continue
756 }
757 hdr, ok := store.Header(h)
758 if !ok {
759 return nil, fmt.Errorf("neutrino: filter %d has no header", h)
760 }
761 blockHash := hdr.BlockHash()
762 match, err := filter.Match(f, [32]byte(blockHash), scripts)
763 if err != nil {
764 return nil, fmt.Errorf("neutrino: filter %d: %w", h, err)
765 }
766 if match {
767 out = append(out, h)
768 }
769 }
770 return out, nil
771 }
772