1 // Package chainsource makes a light client answer on a chaindata channel.
2 //
3 // The wallet never touches the client: it sends a frame and this adapter drives the client.
4 // That is what keeps the two sides independent of each other's code, and it is why this
5 // adapter is the only place that knows both.
6 package chainsource
7 8 import (
9 "context"
10 11 "git.smesh.lol/nostrino/pkg/chaindata"
12 "git.smesh.lol/nostrino/pkg/neutrino"
13 )
14 15 // Nostrino is a chaindata.Source over a BIP157 light client and its store.
16 //
17 // One value per chain. The client is not safe for two concurrent syncs, so the adapter
18 // serialises them: a wallet that asked for two would be two questions about one chain.
19 type Nostrino struct {
20 client *neutrino.Client
21 store neutrino.Store
22 sync chan struct{}
23 }
24 25 // New wraps a client.
26 func New(client *neutrino.Client) (*Nostrino, error) {
27 if client == nil {
28 return nil, errNoClient
29 }
30 31 if client.Store() == nil {
32 return nil, errNoStore
33 }
34 35 return &Nostrino{client: client, store: client.Store(), sync: make(chan struct{}, 1)}, nil
36 }
37 38 // errNoClient is returned for a nil client.
39 var errNoClient = errorString("chainsource: no light client")
40 41 // errNoStore is returned for a client with no store.
42 var errNoStore = errorString("chainsource: the light client has no store")
43 44 // errorString is a plain error, kept here so the package needs no other dependency.
45 type errorString string
46 47 func (e errorString) Error() string { return string(e) }
48 49 // Position reports the store's verified tips.
50 //
51 // It reads the store rather than the peer: a tip the client has not verified is not a
52 // position this wallet may act on.
53 func (n *Nostrino) Position(chaindata.Context) (chaindata.Position, error) {
54 position := chaindata.Position{}
55 56 if tip, _, ok := n.store.HeaderTip(); ok {
57 position.Tip = tip
58 }
59 60 if filters, ok := n.store.FilterTip(); ok {
61 position.Filters = filters
62 }
63 64 return position, nil
65 }
66 67 // Sync advances the chain from a height.
68 //
69 // from is a lower bound, not a demand: the client resumes from its own verified tip, because
70 // re-downloading verified filters is work with no answer in it.
71 func (n *Nostrino) Sync(ctx chaindata.Context, from uint32) (chaindata.Position, error) {
72 release, ok := n.lock(ctx)
73 74 if !ok {
75 return chaindata.Position{}, errorString("chainsource: cancelled before the sync began")
76 }
77 78 defer release()
79 80 runCtx, stop := bridge(ctx)
81 82 defer stop()
83 84 if _, err := n.client.SyncFrom(runCtx, from, 0); err != nil {
85 return chaindata.Position{}, err
86 }
87 88 return n.Position(ctx)
89 }
90 91 // Match returns the heights whose stored filters mention any of the scripts.
92 //
93 // The scan is the client's own local matching: the peer never learns the script set.
94 func (n *Nostrino) Match(ctx chaindata.Context, scripts [][]byte) ([]uint32, error) {
95 if len(scripts) == 0 {
96 return nil, errorString("chainsource: a match needs at least one script")
97 }
98 99 release, ok := n.lock(ctx)
100 101 if !ok {
102 return nil, errorString("chainsource: cancelled before the match began")
103 }
104 105 defer release()
106 107 return n.client.MatchScripts(0, 0, scripts)
108 }
109 110 // Filters returns a contiguous run of stored filters.
111 //
112 // The range is served from the client's own verified store, and the whole run is one slice
113 // of filters: the caller gets a batch, not a round trip per block. A height whose filter is
114 // not stored is a gap, and a gap is an error rather than a shorter answer, because a wallet
115 // that silently skipped a block would miss its own money.
116 //
117 // Block returns the raw block at height.
118 func (n *Nostrino) Block(ctx chaindata.Context, height uint32) ([]byte, error) {
119 release, ok := n.lock(ctx)
120 121 if !ok {
122 return nil, errorString("chainsource: cancelled before the block fetch began")
123 }
124 125 defer release()
126 127 requestCtx, done := bridge(ctx)
128 129 defer done()
130 131 raw, err := n.client.Block(requestCtx, height)
132 133 if err != nil {
134 return nil, errorString("chainsource: " + err.Error())
135 }
136 137 return raw, nil
138 }
139 140 // Broadcast sends one raw transaction to the peer this client follows.
141 //
142 // It is the write half of the channel: a wallet that builds and signs locally hands the bytes
143 // here, and the light client's tunnel carries them to the node. The peer does not acknowledge a
144 // tx message, so a nil error means the frame reached the peer.
145 func (n *Nostrino) Broadcast(ctx chaindata.Context, raw []byte) error {
146 release, ok := n.lock(ctx)
147 148 if !ok {
149 return errorString("chainsource: cancelled before the broadcast began")
150 }
151 152 defer release()
153 154 requestCtx, done := bridge(ctx)
155 156 defer done()
157 158 if err := n.client.Broadcast(requestCtx, raw); err != nil {
159 return errorString("chainsource: " + err.Error())
160 }
161 162 return nil
163 }
164 165 func (n *Nostrino) Filters(
166 ctx chaindata.Context, from, count uint32,
167 ) ([]chaindata.Filter, error) {
168 if count == 0 || count > chaindata.MaxFilters {
169 return nil, errorString("chainsource: a filter read is 1 to 2000 filters")
170 }
171 172 release, ok := n.lock(ctx)
173 174 if !ok {
175 return nil, errorString("chainsource: cancelled before the read began")
176 }
177 178 defer release()
179 180 out := make([]chaindata.Filter, 0, count)
181 182 for offset := uint32(0); offset < count; offset++ {
183 height := from + offset
184 185 body, ok := n.store.Filter(height)
186 187 if !ok {
188 return nil, errorString("chainsource: no filter at height " + itoa(height))
189 }
190 191 filter := chaindata.Filter{Height: height, Filter: body}
192 193 if header, ok := n.store.Header(height); ok {
194 hash := header.BlockHash()
195 196 filter.BlockHash = hash[:]
197 }
198 199 out = append(out, filter)
200 }
201 202 return out, nil
203 }
204 205 // lock takes the one sync slot, or gives up when the request is cancelled.
206 //
207 // The cancellation is checked before the slot is offered, not raced with it: when both cases
208 // of a select are ready Go picks one at random, so a cancelled request that also found the
209 // slot free would start its work half the time. A cancelled request must never begin.
210 func (n *Nostrino) lock(ctx chaindata.Context) (func(), bool) {
211 select {
212 case <-ctx.Done():
213 return nil, false
214 default:
215 }
216 217 select {
218 case n.sync <- struct{}{}:
219 return func() { <-n.sync }, true
220 case <-ctx.Done():
221 return nil, false
222 }
223 }
224 225 // bridge turns a chaindata cancellation into a context cancellation.
226 //
227 // It is one goroutine per request, which is the price of an interface that carries a channel
228 // instead of a context.Context: the client's loops select on the derived context, so the sync
229 // stops at its next read rather than at its next check of a flag.
230 func bridge(ctx chaindata.Context) (context.Context, func()) {
231 runCtx, cancel := context.WithCancel(context.Background())
232 233 stopped := make(chan struct{})
234 235 go func() {
236 defer close(stopped)
237 238 select {
239 case <-ctx.Done():
240 cancel()
241 case <-runCtx.Done():
242 }
243 }()
244 245 return runCtx, func() {
246 cancel()
247 <-stopped
248 }
249 }
250 251 // itoa renders a height without pulling in formatting.
252 func itoa(value uint32) string {
253 if value == 0 {
254 return "0"
255 }
256 257 var digits []byte
258 259 for value > 0 {
260 digits = append([]byte{byte('0' + value%10)}, digits...)
261 value /= 10
262 }
263 264 return string(digits)
265 }
266