1 // Package store is the wallet's durable state: an in-memory cache over an asynchronous
2 // store, with the writes batched behind it.
3 //
4 // The shape is forced by the platform. IndexedDB is asynchronous and every one of its calls
5 // is a round trip to another thread, while the wallet's code wants to read and write its own
6 // state synchronously, the way an embedded database's View and Update do. A shim that awaited
7 // a Put would put a thread-hop inside the wallet's hot path, and a wallet that is matching a
8 // filter batch writes dozens of times in a row.
9 //
10 // So the cache is the truth for the life of the process and the store behind it is a
11 // write-behind copy:
12 //
13 // - Get, Put and Delete are memory operations that never block and never fail;
14 // - a Put marks the key dirty and a coalescing flush writes the dirty set in one batch;
15 // - a failed flush keeps the batch dirty and tries again, and the seed is what makes that
16 // recoverable: a lost flush is a rescan, not a lost wallet.
17 //
18 // What this deliberately does not promise is durability at the moment of a write. It cannot:
19 // the platform cannot promise it either. What it promises is that a write is never lost
20 // silently, and that losing the tail of the log costs a rescan rather than a deposit.
21 package store
22 23 import (
24 "fmt"
25 "sort"
26 "strings"
27 "sync"
28 "time"
29 )
30 31 // Op is one durable change: a value to store, or a key to remove.
32 type Op struct {
33 // Key is the document name.
34 Key string
35 // Value is the new value. Nil with Delete false is an empty value, which is different
36 // from a deletion and is stored as empty.
37 Value []byte
38 // Delete marks a removal.
39 Delete bool
40 }
41 42 // Persistence is the asynchronous side.
43 //
44 // Load is called once, at open, and returns everything the store holds: the wallet's data is
45 // small enough that a full load is simpler and faster than a page-fetching API, and the seed
46 // can rebuild it if it is wrong. Write is called with a batch and must apply all of it or
47 // none of it: a half-applied batch would be a wallet whose cache and store disagree.
48 type Persistence interface {
49 // Load returns every stored key and value.
50 Load() (map[string][]byte, error)
51 // Write applies one batch atomically.
52 Write(batch []Op) error
53 }
54 55 // Cache is the wallet's synchronous state.
56 type Cache struct {
57 mu sync.Mutex
58 items map[string][]byte
59 // dirty holds the keys changed since the last successful flush, with the value to
60 // write and the version it was written at. A second change to the same key replaces
61 // the first, which is the coalescing a filter batch needs, and the version is how a
62 // flush knows whether the value it wrote is still the value the cache holds.
63 dirty map[string]pendingOp
64 version uint64
65 66 persist Persistence
67 // done stops the flusher.
68 done chan struct{}
69 // stopped is closed when the flusher has exited.
70 stopped chan struct{}
71 72 // flushInterval is how long a dirty key may wait before the background flush takes it.
73 // Zero means nothing flushes until the caller asks, which is what a test wants.
74 flushInterval time.Duration
75 76 // failures counts failed flushes, for a caller that wants to say so.
77 failures int
78 lastErr error
79 }
80 81 // DefaultFlushInterval is how long a change may wait before it reaches the store.
82 //
83 // Two seconds is the platform's compromise: every flush is a transaction on another thread,
84 // and a wallet matching a block writes in bursts, so flushing per write would spend the sync
85 // in round trips. What a lost window costs is a rescan of the last couple of seconds, and the
86 // seed is what makes that free.
87 const DefaultFlushInterval = 2 * time.Second
88 89 // Open builds a cache over a persistence, flushing on the default interval.
90 func Open(persist Persistence) (*Cache, error) {
91 return OpenWith(persist, DefaultFlushInterval)
92 }
93 94 // OpenManual builds a cache that flushes only when it is asked.
95 //
96 // It is for tests, and for a caller that flushes at its own boundaries: a sync that has just
97 // verified a batch of filters knows better than a timer when to write.
98 func OpenManual(persist Persistence) (*Cache, error) {
99 return OpenWith(persist, 0)
100 }
101 102 // OpenWith builds a cache over a persistence, loading what it holds.
103 //
104 // A persistence that cannot load is an error: a wallet that silently started empty would
105 // show a zero balance for a funded wallet, which is the one mistake this package must not
106 // make.
107 func OpenWith(persist Persistence, interval time.Duration) (*Cache, error) {
108 if persist == nil {
109 return nil, fmt.Errorf("store: a cache needs a persistence")
110 }
111 112 loaded, err := persist.Load()
113 114 if err != nil {
115 return nil, fmt.Errorf("store: loading: %w", err)
116 }
117 118 cache := &Cache{
119 items: make(map[string][]byte, len(loaded)),
120 dirty: map[string]pendingOp{},
121 persist: persist,
122 done: make(chan struct{}),
123 stopped: make(chan struct{}),
124 flushInterval: interval,
125 }
126 127 for key, value := range loaded {
128 copied := make([]byte, len(value))
129 copy(copied, value)
130 131 cache.items[key] = copied
132 }
133 134 go cache.flusher()
135 136 return cache, nil
137 }
138 139 // Get reads a value.
140 func (c *Cache) Get(key string) ([]byte, bool) {
141 c.mu.Lock()
142 defer c.mu.Unlock()
143 144 value, ok := c.items[key]
145 146 if !ok {
147 return nil, false
148 }
149 150 copied := make([]byte, len(value))
151 copy(copied, value)
152 153 return copied, true
154 }
155 156 // Put stores a value in memory and marks it for the next flush.
157 //
158 // It returns immediately. The value is copied, so a caller reusing its buffer cannot change
159 // what will be written.
160 func (c *Cache) Put(key string, value []byte) error {
161 if key == "" {
162 return fmt.Errorf("store: a value needs a key")
163 }
164 165 if len(key) > 256 {
166 return fmt.Errorf("store: the key %q is longer than 256 bytes", key[:32])
167 }
168 169 copied := make([]byte, len(value))
170 copy(copied, value)
171 172 c.mu.Lock()
173 c.items[key] = copied
174 c.version++
175 c.dirty[key] = pendingOp{op: Op{Key: key, Value: copied}, version: c.version}
176 c.mu.Unlock()
177 178 return nil
179 }
180 181 // Delete removes a value.
182 func (c *Cache) Delete(key string) error {
183 if key == "" {
184 return fmt.Errorf("store: a deletion needs a key")
185 }
186 187 c.mu.Lock()
188 delete(c.items, key)
189 c.version++
190 c.dirty[key] = pendingOp{op: Op{Key: key, Delete: true}, version: c.version}
191 c.mu.Unlock()
192 193 return nil
194 }
195 196 // Keys returns the keys with a prefix, sorted.
197 func (c *Cache) Keys(prefix string) []string {
198 c.mu.Lock()
199 defer c.mu.Unlock()
200 201 out := make([]string, 0, len(c.items))
202 203 for key := range c.items {
204 if strings.HasPrefix(key, prefix) {
205 out = append(out, key)
206 }
207 }
208 209 sort.Strings(out)
210 211 return out
212 }
213 214 // Len reports how many keys are held.
215 func (c *Cache) Len() int {
216 c.mu.Lock()
217 defer c.mu.Unlock()
218 219 return len(c.items)
220 }
221 222 // flusher writes the dirty set whenever it is asked and, with an interval, on a timer.
223 func (c *Cache) flusher() {
224 defer close(c.stopped)
225 226 c.mu.Lock()
227 interval := c.flushInterval
228 c.mu.Unlock()
229 230 if interval <= 0 {
231 // Manual: nothing happens until Close, which flushes the last window itself.
232 <-c.done
233 234 return
235 }
236 237 ticker := time.NewTicker(interval)
238 defer ticker.Stop()
239 240 for {
241 select {
242 case <-c.done:
243 // A last attempt, so a clean shutdown does not leave the cache dirty when the
244 // store is reachable.
245 _ = c.Flush()
246 247 return
248 249 case <-ticker.C:
250 _ = c.Flush()
251 }
252 }
253 }
254 255 // Flush writes the dirty set, and returns what went wrong.
256 //
257 // One batch, one write. A failure keeps every key dirty and is reported; the next call tries
258 // again with whatever is dirty then, so a store that comes back gets everything that changed
259 // while it was away. A value that changed during the write is newer than what was written, so
260 // it stays dirty and the next flush carries it.
261 func (c *Cache) Flush() error {
262 c.mu.Lock()
263 264 if len(c.dirty) == 0 {
265 c.mu.Unlock()
266 267 return nil
268 }
269 270 batch := make([]Op, 0, len(c.dirty))
271 written := make(map[string]uint64, len(c.dirty))
272 273 for key, pending := range c.dirty {
274 batch = append(batch, pending.op)
275 276 written[key] = pending.version
277 }
278 279 c.mu.Unlock()
280 281 sort.Slice(batch, func(i, j int) bool { return batch[i].Key < batch[j].Key })
282 283 err := c.persist.Write(batch)
284 285 c.mu.Lock()
286 defer c.mu.Unlock()
287 288 if err != nil {
289 c.failures++
290 c.lastErr = err
291 292 return fmt.Errorf("store: writing %d change(s): %w", len(batch), err)
293 }
294 295 for key, version := range written {
296 if pending, ok := c.dirty[key]; ok && pending.version == version {
297 delete(c.dirty, key)
298 }
299 }
300 301 return nil
302 }
303 304 // pendingOp is one key's pending change and the version it was made at.
305 type pendingOp struct {
306 op Op
307 version uint64
308 }
309 310 // Failures reports how many flushes have failed, and why the last one did.
311 func (c *Cache) Failures() (int, error) {
312 c.mu.Lock()
313 defer c.mu.Unlock()
314 315 return c.failures, c.lastErr
316 }
317 318 // Close stops the flusher, after one last flush.
319 func (c *Cache) Close() error {
320 select {
321 case <-c.done:
322 // Already closing.
323 default:
324 close(c.done)
325 }
326 327 select {
328 case <-c.stopped:
329 case <-time.After(5 * time.Second):
330 return fmt.Errorf("store: the flusher did not stop")
331 }
332 333 return c.Flush()
334 }
335