store_test.go raw
1 package store
2
3 import (
4 "errors"
5 "strings"
6 "sync"
7 "testing"
8 "time"
9 )
10
11 // recorder is a persistence that records what it is asked to write and can be told to fail.
12 type recorder struct {
13 mu sync.Mutex
14 items map[string][]byte
15 batches [][]Op
16 writes int
17 // fail makes the next N writes fail.
18 fail int
19 // loads counts Load calls.
20 loads int
21 // loadErr is returned by Load when set.
22 loadErr error
23 }
24
25 // newRecorder builds a recorder holding the given documents.
26 func newRecorder(items map[string][]byte) *recorder {
27 if items == nil {
28 items = map[string][]byte{}
29 }
30
31 return &recorder{items: items}
32 }
33
34 // Load implements Persistence.
35 func (r *recorder) Load() (map[string][]byte, error) {
36 r.mu.Lock()
37 defer r.mu.Unlock()
38
39 r.loads++
40
41 if r.loadErr != nil {
42 return nil, r.loadErr
43 }
44
45 out := make(map[string][]byte, len(r.items))
46
47 for key, value := range r.items {
48 out[key] = value
49 }
50
51 return out, nil
52 }
53
54 // Write implements Persistence.
55 func (r *recorder) Write(batch []Op) error {
56 r.mu.Lock()
57 defer r.mu.Unlock()
58
59 r.writes++
60
61 if r.fail > 0 {
62 r.fail--
63
64 return errors.New("the store is unavailable")
65 }
66
67 copied := make([]Op, len(batch))
68
69 for i, op := range batch {
70 copied[i] = Op{Key: op.Key, Delete: op.Delete, Value: append([]byte{}, op.Value...)}
71
72 if op.Delete {
73 delete(r.items, op.Key)
74 } else {
75 r.items[op.Key] = append([]byte{}, op.Value...)
76 }
77 }
78
79 r.batches = append(r.batches, copied)
80
81 return nil
82 }
83
84 // snapshot returns a copy of what the store holds.
85 func (r *recorder) snapshot() map[string][]byte {
86 r.mu.Lock()
87 defer r.mu.Unlock()
88
89 out := make(map[string][]byte, len(r.items))
90
91 for key, value := range r.items {
92 out[key] = value
93 }
94
95 return out
96 }
97
98 // batchCount is how many batches have been written.
99 func (r *recorder) batchCount() int {
100 r.mu.Lock()
101 defer r.mu.Unlock()
102
103 return len(r.batches)
104 }
105
106 // open builds a manual-flush cache over a recorder and closes it at the end of the test.
107 //
108 // Manual, because a background flush would make a write-count assertion a race: these tests
109 // count the round trips, so they take the flushes into their own hands.
110 func open(t *testing.T, persist Persistence) *Cache {
111 t.Helper()
112
113 cache, err := OpenManual(persist)
114
115 if err != nil {
116 t.Fatal(err)
117 }
118
119 t.Cleanup(func() { _ = cache.Close() })
120
121 return cache
122 }
123
124 // TestReadsAndWritesAreMemory: the hot path never touches the persistence.
125 //
126 // This is the property the whole package exists for: a wallet matching a filter batch writes
127 // dozens of times in a row, and a thread-hop per write is the difference between a sync that
128 // keeps up and one that does not.
129 func TestReadsAndWritesAreMemory(t *testing.T) {
130 persist := newRecorder(nil)
131 cache, err := OpenManual(persist)
132
133 if err != nil {
134 t.Fatal(err)
135 }
136
137 // Manual flushing, so no background write can land between the writes and the count.
138 for i := 0; i < 200; i++ {
139 if err := cache.Put("addr/"+itoa(i), []byte("value")); err != nil {
140 t.Fatal(err)
141 }
142
143 if value, ok := cache.Get("addr/" + itoa(i)); !ok || string(value) != "value" {
144 t.Fatalf("read back %q, %v", value, ok)
145 }
146 }
147
148 if cache.Len() != 200 {
149 t.Fatalf("the cache holds %d keys", cache.Len())
150 }
151
152 // A flush is one round trip for all of them, not one each.
153 if err := cache.Flush(); err != nil {
154 t.Fatal(err)
155 }
156
157 persist.mu.Lock()
158 writes := persist.writes
159 persist.mu.Unlock()
160
161 if writes != 1 {
162 t.Fatalf("%d writes for 200 changes", writes)
163 }
164
165 if err := cache.Close(); err != nil {
166 t.Fatal(err)
167 }
168
169 if len(persist.snapshot()) != 200 {
170 t.Fatalf("the store holds %d documents", len(persist.snapshot()))
171 }
172 }
173
174 // TestCoalescing: many writes to one key are one document in one batch.
175 func TestCoalescing(t *testing.T) {
176 persist := newRecorder(nil)
177 cache := open(t, persist)
178
179 for i := 0; i < 100; i++ {
180 if err := cache.Put("utxo/one", []byte("value "+itoa(i))); err != nil {
181 t.Fatal(err)
182 }
183 }
184
185 if err := cache.Flush(); err != nil {
186 t.Fatal(err)
187 }
188
189 if err := cache.Close(); err != nil {
190 t.Fatal(err)
191 }
192
193 if persist.batchCount() != 1 {
194 t.Fatalf("%d batches", persist.batchCount())
195 }
196
197 first := persist.batches[0]
198
199 if len(first) != 1 || first[0].Key != "utxo/one" {
200 t.Fatalf("the batch is %+v", first)
201 }
202
203 if string(first[0].Value) != "value 99" {
204 t.Fatalf("the written value is %q", first[0].Value)
205 }
206 }
207
208 // TestFailedWriteKeepsTheDataDirty: a store that is away loses nothing, and the next flush
209 // carries everything that changed while it was away.
210 func TestFailedWriteKeepsTheDataDirty(t *testing.T) {
211 persist := newRecorder(nil)
212 cache, err := OpenManual(persist)
213
214 if err != nil {
215 t.Fatal(err)
216 }
217
218 defer func() { _ = cache.Close() }()
219
220 // The first write fails, so this change must survive.
221 persist.mu.Lock()
222 persist.fail = 1
223 persist.mu.Unlock()
224
225 if err := cache.Put("addr/one", []byte("first")); err != nil {
226 t.Fatal(err)
227 }
228
229 if err := cache.Flush(); err == nil {
230 t.Fatal("the failing store reported success")
231 }
232
233 if failures, last := cache.Failures(); failures != 1 || last == nil {
234 t.Fatalf("failures = %d, %v", failures, last)
235 }
236
237 // A second change, then a flush that succeeds: both are written.
238 if err := cache.Put("addr/two", []byte("second")); err != nil {
239 t.Fatal(err)
240 }
241
242 if err := cache.Flush(); err != nil {
243 t.Fatal(err)
244 }
245
246 if err := cache.Close(); err != nil {
247 t.Fatal(err)
248 }
249
250 stored := persist.snapshot()
251
252 if string(stored["addr/one"]) != "first" || string(stored["addr/two"]) != "second" {
253 t.Fatalf("the store holds %v", stored)
254 }
255 }
256
257 // TestDeletionIsWritten: a removal is an operation, not an absence.
258 func TestDeletionIsWritten(t *testing.T) {
259 persist := newRecorder(map[string][]byte{
260 "addr/gone": []byte("value"),
261 "addr/kept": []byte("value"),
262 })
263
264 cache := open(t, persist)
265
266 if _, ok := cache.Get("addr/gone"); !ok {
267 t.Fatal("the loaded document is not there")
268 }
269
270 if err := cache.Delete("addr/gone"); err != nil {
271 t.Fatal(err)
272 }
273
274 if _, ok := cache.Get("addr/gone"); ok {
275 t.Fatal("the deleted document is still readable")
276 }
277
278 if err := cache.Close(); err != nil {
279 t.Fatal(err)
280 }
281
282 stored := persist.snapshot()
283
284 if _, ok := stored["addr/gone"]; ok {
285 t.Fatal("the store still holds the deleted document")
286 }
287
288 if string(stored["addr/kept"]) != "value" {
289 t.Fatal("the other document was lost")
290 }
291 }
292
293 // TestReopenSeesWhatWasWritten: the store is what survives a restart.
294 func TestReopenSeesWhatWasWritten(t *testing.T) {
295 persist := newRecorder(nil)
296
297 first, err := OpenManual(persist)
298
299 if err != nil {
300 t.Fatal(err)
301 }
302
303 for _, key := range []string{"wallet/seed-id", "addr/0", "addr/1", "utxo/a"} {
304 if err := first.Put(key, []byte(key)); err != nil {
305 t.Fatal(err)
306 }
307 }
308
309 if err := first.Close(); err != nil {
310 t.Fatal(err)
311 }
312
313 second, err := OpenManual(persist)
314
315 if err != nil {
316 t.Fatal(err)
317 }
318
319 defer func() { _ = second.Close() }()
320
321 if second.Len() != 4 {
322 t.Fatalf("the reopened cache holds %d keys", second.Len())
323 }
324
325 keys := second.Keys("addr/")
326
327 if len(keys) != 2 || keys[0] != "addr/0" || keys[1] != "addr/1" {
328 t.Fatalf("keys = %v", keys)
329 }
330 }
331
332 // TestLoadFailureIsAnError: a wallet that cannot read its store must say so rather than
333 // showing a zero balance for a funded wallet.
334 func TestLoadFailureIsAnError(t *testing.T) {
335 persist := newRecorder(nil)
336 persist.loadErr = errors.New("the database is corrupt")
337
338 if _, err := Open(persist); err == nil {
339 t.Fatal("a store that could not load was opened empty")
340 }
341 }
342
343 // TestValuesAreCopied: a caller reusing its buffer cannot change what is stored or written.
344 func TestValuesAreCopied(t *testing.T) {
345 persist := newRecorder(nil)
346 cache := open(t, persist)
347
348 buffer := []byte("original")
349
350 if err := cache.Put("addr/one", buffer); err != nil {
351 t.Fatal(err)
352 }
353
354 buffer[0] = 'X'
355
356 value, _ := cache.Get("addr/one")
357
358 if string(value) != "original" {
359 t.Fatalf("the cached value is %q", value)
360 }
361
362 if err := cache.Close(); err != nil {
363 t.Fatal(err)
364 }
365
366 if string(persist.snapshot()["addr/one"]) != "original" {
367 t.Fatalf("the written value is %q", persist.snapshot()["addr/one"])
368 }
369 }
370
371 // TestKeysAreSorted: iteration order is deterministic on every platform.
372 func TestKeysAreSorted(t *testing.T) {
373 cache := open(t, newRecorder(nil))
374
375 for _, key := range []string{"b", "a", "c", "aa"} {
376 if err := cache.Put(key, nil); err != nil {
377 t.Fatal(err)
378 }
379 }
380
381 keys := cache.Keys("")
382
383 if strings.Join(keys, ",") != "a,aa,b,c" {
384 t.Fatalf("keys = %v", keys)
385 }
386
387 if got := cache.Keys("a"); strings.Join(got, ",") != "a,aa" {
388 t.Fatalf("prefixed keys = %v", got)
389 }
390 }
391
392 // TestBadKeysAreRefused: an empty or huge key is a caller's mistake, named here.
393 func TestBadKeysAreRefused(t *testing.T) {
394 cache := open(t, newRecorder(nil))
395
396 if err := cache.Put("", nil); err == nil {
397 t.Fatal("an empty key was stored")
398 }
399
400 if err := cache.Delete(""); err == nil {
401 t.Fatal("an empty key was deleted")
402 }
403
404 if err := cache.Put(strings.Repeat("k", 300), nil); err == nil {
405 t.Fatal("a 300-byte key was stored")
406 }
407 }
408
409 // TestAChangeDuringAFlushIsKept: the value written is not necessarily the value the cache
410 // ends up with, and the newer one must reach the store.
411 func TestAChangeDuringAFlushIsKept(t *testing.T) {
412 persist := &slowRecorder{
413 recorder: newRecorder(nil),
414 started: make(chan struct{}),
415 release: make(chan struct{}),
416 }
417
418 cache, err := OpenManual(persist)
419
420 if err != nil {
421 t.Fatal(err)
422 }
423
424 defer func() { _ = cache.Close() }()
425
426 if err := cache.Put("addr/one", []byte("first")); err != nil {
427 t.Fatal(err)
428 }
429
430 flushed := make(chan error, 1)
431
432 go func() { flushed <- cache.Flush() }()
433
434 // Wait until the write is in progress, then change the value.
435 <-persist.started
436
437 if err := cache.Put("addr/one", []byte("second")); err != nil {
438 t.Fatal(err)
439 }
440
441 close(persist.release)
442
443 if err := <-flushed; err != nil {
444 t.Fatal(err)
445 }
446
447 if err := cache.Flush(); err != nil {
448 t.Fatal(err)
449 }
450
451 if err := cache.Close(); err != nil {
452 t.Fatal(err)
453 }
454
455 if got := string(persist.snapshot()["addr/one"]); got != "second" {
456 t.Fatalf("the store holds %q, want the newer value", got)
457 }
458 }
459
460 // slowRecorder blocks in Write until it is released, so a change can be made mid-flush.
461 type slowRecorder struct {
462 *recorder
463 started chan struct{}
464 release chan struct{}
465 once sync.Once
466 }
467
468 // Write implements Persistence.
469 func (s *slowRecorder) Write(batch []Op) error {
470 s.once.Do(func() { close(s.started) })
471
472 <-s.release
473
474 return s.recorder.Write(batch)
475 }
476
477 // TestConcurrentWritersAndTheFlusher: memory operations and the background flush do not
478 // corrupt each other. It is the one test that wants -race.
479 func TestConcurrentWritersAndTheFlusher(t *testing.T) {
480 // A real interval, because this test is about the flusher running under memory writes.
481 cache, err := OpenWith(newRecorder(nil), time.Millisecond)
482
483 if err != nil {
484 t.Fatal(err)
485 }
486
487 defer func() { _ = cache.Close() }()
488
489 var group sync.WaitGroup
490
491 for worker := 0; worker < 4; worker++ {
492 group.Add(1)
493
494 go func(id int) {
495 defer group.Done()
496
497 for i := 0; i < 200; i++ {
498 key := "addr/" + itoa(id) + "/" + itoa(i)
499
500 if err := cache.Put(key, []byte("value")); err != nil {
501 t.Errorf("put: %v", err)
502
503 return
504 }
505
506 _ = cache.Keys("addr/")
507 _, _ = cache.Get(key)
508 }
509 }(worker)
510 }
511
512 group.Wait()
513
514 if err := cache.Flush(); err != nil {
515 t.Fatal(err)
516 }
517
518 if cache.Len() != 800 {
519 t.Fatalf("the cache holds %d keys", cache.Len())
520 }
521 }
522
523 // itoa renders a small integer without pulling in formatting.
524 func itoa(value int) string {
525 if value == 0 {
526 return "0"
527 }
528
529 var digits []byte
530
531 for value > 0 {
532 digits = append([]byte{byte('0' + value%10)}, digits...)
533 value /= 10
534 }
535
536 return string(digits)
537 }
538