diff --git a/.changeset/support_concurrent_chain_manager_reads_via_mvcc_snapshots.md b/.changeset/support_concurrent_chain_manager_reads_via_mvcc_snapshots.md new file mode 100644 index 00000000..3deed854 --- /dev/null +++ b/.changeset/support_concurrent_chain_manager_reads_via_mvcc_snapshots.md @@ -0,0 +1,14 @@ +--- +default: major +--- + +# Support concurrent chain.Manager reads via MVCC snapshots + +The Manager previously serialized all of its methods -- including read-only ones -- with a single mutex, so heavy read load (e.g. from the syncer or subscribers) contended with block processing. The Manager now follows an MVCC scheme: writes accumulate in the Store's "scratchpad" -- which includes the tip state -- and are committed by Flush, at which point they become visible to "snapshots": self-contained, read-only captures of the committed state. Read-only Manager methods operate on snapshots, so they proceed concurrently with each other and with block processing, and never stall writers. bbolt supports snapshots natively via read-only transactions; `MemDB` now implements them with pin-aware copy-on-flush generations (`CacheDB` delegates to its underlying DB). + +This is a breaking change for implementors and consumers of the `chain.DB` and `chain.Store` interfaces: + +- `Store` is now `Snapshot() (StoreSnapshot, func())` + `Scratchpad() StoreScratchpad`. `StoreSnapshot` carries the read methods plus `TipState`; `StoreScratchpad` embeds it and adds the write methods and `Flush`, with `ApplyBlock`/`RevertBlock` updating the scratchpad's tip. Implementations may flush autonomously between operations (e.g. to bound the size or age of the accumulated writes), handling errors internally; the Manager explicitly calls Flush to publish the new tip after processing blocks, before notifying OnReorg listeners. +- `DB` mirrors `Store`: `Snapshot() (DBSnapshot, release func())` + `Scratchpad() DBScratchpad`. The scratchpad carries `Bucket`/`CreateBucket`/`Flush`/`Cancel` and must not flush autonomously; snapshots are read-only and remain valid until released, even across concurrent Flushes. A single `DBBucket` interface serves both sides; its `Put`/`Delete` methods may return errors on read-only (snapshot) buckets. +- `NewManager` no longer takes a `consensus.State` parameter, and `NewDBStore`/`NewDBStoreAtCheckpoint` no longer return one; the tip state is provided by the Store itself (via a snapshot's or scratchpad's `TipState`). +- Reads reflect committed state. Pruned blocks remain visible to read-only methods until the next commit, and blocks added to non-best chains remain *invisible* to them until the next commit. Blocks that extend the best chain are always committed before `AddBlocks` returns and before `OnReorg` listeners fire, so subscribers observe the tip they are notified of. diff --git a/chain/chain_test.go b/chain/chain_test.go index 29a7fb85..41157eb1 100644 --- a/chain/chain_test.go +++ b/chain/chain_test.go @@ -138,11 +138,11 @@ func TestV2Attestations(t *testing.T) { // would be mined in every block, so the txpool must reject it regardless of // any attestations or arbitrary data it carries. t.Run("rejects transactions that spend no elements", func(t *testing.T) { - store, tipState, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) + store, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := chain.NewManager(store, tipState) + cm := chain.NewManager(store) cases := map[string]types.V2Transaction{ "arbitrary data only": { @@ -181,11 +181,11 @@ func TestV2Attestations(t *testing.T) { } for name, fundTxn := range spend { t.Run("accepts funded announcement transaction ("+name+")", func(t *testing.T) { - store, tipState, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) + store, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := chain.NewManager(store, tipState) + cm := chain.NewManager(store) ms := newMemState() // mine until a utxo is spendable diff --git a/chain/db.go b/chain/db.go index a3ab8f23..d8fee1ec 100644 --- a/chain/db.go +++ b/chain/db.go @@ -6,8 +6,10 @@ import ( "errors" "fmt" "iter" + "maps" "math/bits" "sort" + "sync" "time" "go.sia.tech/core/consensus" @@ -59,8 +61,32 @@ func (vs *versionedState) DecodeFrom(d *types.Decoder) { vs.State.DecodeFrom(d) } -// A DB is a generic key-value database. +// A DB is a generic key-value database, comprising a mutable "scratchpad" and +// read-only snapshots of its committed state. type DB interface { + // Snapshot returns a read-only snapshot of the DB's data as of the most + // recent Flush. Snapshots are unaffected by concurrent scratchpad writes + // and Flushes. The release function must be called when the snapshot is + // no longer needed. + Snapshot() (dbs DBSnapshot, release func()) + // Scratchpad returns a handle to the current scratchpad, which observes + // the DB's uncommitted data. + Scratchpad() DBScratchpad +} + +// A DBSnapshot is a read-only snapshot of a DB. +type DBSnapshot interface { + Bucket(name []byte) DBBucket +} + +// A DBScratchpad accumulates writes to a DB, which become durable when Flush +// is called. Its methods observe the accumulated writes. It is not safe for +// concurrent use. +// +// Unlike a StoreScratchpad, a DBScratchpad must not flush autonomously: +// callers batch writes across multiple method calls, so only they know when +// the accumulated writes form a consistent snapshot. +type DBScratchpad interface { Bucket(name []byte) DBBucket CreateBucket(name []byte) (DBBucket, error) Flush() error @@ -70,42 +96,69 @@ type DB interface { // A DBBucket is a set of key-value pairs. type DBBucket interface { Get(key []byte) []byte + Iter() iter.Seq2[[]byte, []byte] + // these methods MAY return errors if the bucket is read-only Put(key, value []byte) error Delete(key []byte) error - Iter() iter.Seq2[[]byte, []byte] +} + +// a memGen is a "generation" of committed MemDB state. Each open snapshot pins +// the generation it was created from; Flush mutates the current generation in +// place if it is unpinned, and otherwise leaves it frozen, applying writes to +// a copy instead. +type memGen struct { + buckets map[string]map[string][]byte + pins int } // MemDB implements DB with an in-memory map. type MemDB struct { - buckets map[string]map[string][]byte - puts map[string]map[string][]byte - dels map[string]map[string]struct{} + mu sync.Mutex // guards gen pointer and pin counts + gen *memGen + puts map[string]map[string][]byte + dels map[string]map[string]struct{} } -// Flush implements DB. +// Flush implements DBScratchpad. func (db *MemDB) Flush() error { + db.mu.Lock() + defer db.mu.Unlock() + if len(db.puts) == 0 && len(db.dels) == 0 { + return nil + } else if db.gen.pins > 0 { + // the current generation is pinned by one or more snapshots; leave it + // frozen, applying our writes to a copy + // + // NOTE: bucket maps must not be shared between generations, as a + // later Flush may mutate buckets that this Flush did not + buckets := make(map[string]map[string][]byte, len(db.gen.buckets)) + for name, kvs := range db.gen.buckets { + buckets[name] = maps.Clone(kvs) + } + db.gen = &memGen{buckets: buckets} + } for bucket, puts := range db.puts { - if db.buckets[bucket] == nil { - db.buckets[bucket] = make(map[string][]byte) + if db.gen.buckets[bucket] == nil { + db.gen.buckets[bucket] = make(map[string][]byte) } for key, val := range puts { - db.buckets[bucket][key] = val + db.gen.buckets[bucket][key] = val } delete(db.puts, bucket) } for bucket, dels := range db.dels { - if db.buckets[bucket] == nil { - db.buckets[bucket] = make(map[string][]byte) + if db.gen.buckets[bucket] == nil { + db.gen.buckets[bucket] = make(map[string][]byte) } for key := range dels { - delete(db.buckets[bucket], key) + delete(db.gen.buckets[bucket], key) } delete(db.dels, bucket) } return nil } -// Cancel implements DB. +// Cancel implements DBScratchpad. func (db *MemDB) Cancel() { for k := range db.puts { delete(db.puts, k) @@ -121,12 +174,12 @@ func (db *MemDB) get(bucket string, key []byte) []byte { } else if _, ok := db.dels[bucket][string(key)]; ok { return nil } - return db.buckets[bucket][string(key)] + return db.gen.buckets[bucket][string(key)] } func (db *MemDB) put(bucket string, key, value []byte) error { if db.puts[bucket] == nil { - if db.buckets[bucket] == nil { + if db.gen.buckets[bucket] == nil { return errors.New("bucket does not exist") } db.puts[bucket] = make(map[string][]byte) @@ -138,7 +191,7 @@ func (db *MemDB) put(bucket string, key, value []byte) error { func (db *MemDB) delete(bucket string, key []byte) error { if db.dels[bucket] == nil { - if db.buckets[bucket] == nil { + if db.gen.buckets[bucket] == nil { return errors.New("bucket does not exist") } db.dels[bucket] = make(map[string]struct{}) @@ -148,9 +201,12 @@ func (db *MemDB) delete(bucket string, key []byte) error { return nil } -// Bucket implements DB. +// Scratchpad implements DB. +func (db *MemDB) Scratchpad() DBScratchpad { return db } + +// Bucket implements DBScratchpad. func (db *MemDB) Bucket(name []byte) DBBucket { - if db.buckets[string(name)] == nil && + if db.gen.buckets[string(name)] == nil && db.puts[string(name)] == nil && db.dels[string(name)] == nil { return nil @@ -158,9 +214,9 @@ func (db *MemDB) Bucket(name []byte) DBBucket { return memBucket{string(name), db} } -// CreateBucket implements DB. +// CreateBucket implements DBScratchpad. func (db *MemDB) CreateBucket(name []byte) (DBBucket, error) { - if db.buckets[string(name)] != nil { + if db.gen.buckets[string(name)] != nil { return nil, errors.New("bucket already exists") } db.puts[string(name)] = make(map[string][]byte) @@ -168,6 +224,55 @@ func (db *MemDB) CreateBucket(name []byte) (DBBucket, error) { return db.Bucket(name), nil } +type memSnapshotBucket struct { + kvs map[string][]byte +} + +func (b memSnapshotBucket) Get(key []byte) []byte { return b.kvs[string(key)] } +func (b memSnapshotBucket) Iter() iter.Seq2[[]byte, []byte] { + return func(yield func([]byte, []byte) bool) { + for key, val := range b.kvs { + if !yield([]byte(key), val) { + return + } + } + } +} +func (b memSnapshotBucket) Put(_, _ []byte) error { return errors.New("bucket is read-only") } +func (b memSnapshotBucket) Delete(_ []byte) error { return errors.New("bucket is read-only") } + +type memDBSnapshot struct { + db *MemDB + gen *memGen + released bool +} + +func (v *memDBSnapshot) Bucket(name []byte) DBBucket { + kvs, ok := v.gen.buckets[string(name)] + if !ok { + return nil + } + return memSnapshotBucket{kvs} +} + +func (v *memDBSnapshot) release() { + v.db.mu.Lock() + defer v.db.mu.Unlock() + if !v.released { + v.released = true + v.gen.pins-- + } +} + +// Snapshot implements DB. +func (db *MemDB) Snapshot() (DBSnapshot, func()) { + db.mu.Lock() + defer db.mu.Unlock() + db.gen.pins++ + s := &memDBSnapshot{db: db, gen: db.gen} + return s, s.release +} + type memBucket struct { name string db *MemDB @@ -178,7 +283,7 @@ func (b memBucket) Put(key, value []byte) error { return b.db.put(b.name, key, v func (b memBucket) Delete(key []byte) error { return b.db.delete(b.name, key) } func (b memBucket) Iter() iter.Seq2[[]byte, []byte] { return func(yield func([]byte, []byte) bool) { - for key, val := range b.db.buckets[b.name] { + for key, val := range b.db.gen.buckets[b.name] { if pval, ok := b.db.puts[b.name][string(key)]; ok { val = pval } else if _, ok := b.db.dels[b.name][string(key)]; ok { @@ -188,15 +293,23 @@ func (b memBucket) Iter() iter.Seq2[[]byte, []byte] { return } } + for key, val := range b.db.puts[b.name] { + if _, ok := b.db.gen.buckets[b.name][key]; ok { + continue // already yielded above + } + if !yield([]byte(key), val) { + return + } + } } } // NewMemDB returns an in-memory DB for use with DBStore. func NewMemDB() *MemDB { return &MemDB{ - buckets: make(map[string]map[string][]byte), - puts: make(map[string]map[string][]byte), - dels: make(map[string]map[string]struct{}), + gen: &memGen{buckets: make(map[string]map[string][]byte)}, + puts: make(map[string]map[string][]byte), + dels: make(map[string]map[string]struct{}), } } @@ -248,9 +361,12 @@ type CacheDB struct { kvs map[string][][2][]byte } -// Bucket implements DB. +// Scratchpad implements DB. +func (db *CacheDB) Scratchpad() DBScratchpad { return db } + +// Bucket implements DBScratchpad. func (db *CacheDB) Bucket(name []byte) DBBucket { - b := db.db.Bucket(name) + b := db.db.Scratchpad().Bucket(name) if b == nil { return nil } else if db.mem.Bucket(name) == nil { @@ -259,16 +375,17 @@ func (db *CacheDB) Bucket(name []byte) DBBucket { return cacheBucket{memBucket{string(name), db.mem}, b} } -// CreateBucket implements DB. +// CreateBucket implements DBScratchpad. func (db *CacheDB) CreateBucket(name []byte) (DBBucket, error) { - if _, err := db.db.CreateBucket(name); err != nil { + if _, err := db.db.Scratchpad().CreateBucket(name); err != nil { return nil, err } return db.mem.CreateBucket(name) } -// Flush implements DB. +// Flush implements DBScratchpad. func (db *CacheDB) Flush() error { + sp := db.db.Scratchpad() // puts for name, puts := range db.mem.puts { bucket := db.kvs[name] @@ -281,7 +398,7 @@ func (db *CacheDB) Flush() error { db.kvs[name] = bucket } for bucket, kvs := range db.kvs { - bucket := db.db.Bucket([]byte(bucket)) + bucket := sp.Bucket([]byte(bucket)) sort.Slice(kvs, func(i, j int) bool { return bytes.Compare(kvs[i][0], kvs[j][0]) < 0 }) @@ -304,7 +421,7 @@ func (db *CacheDB) Flush() error { db.kvs[name] = bucket } for name, kvs := range db.kvs { - bucket := db.db.Bucket([]byte(name)) + bucket := sp.Bucket([]byte(name)) sort.Slice(kvs, func(i, j int) bool { return bytes.Compare(kvs[i][0], kvs[j][0]) < 0 }) @@ -318,7 +435,10 @@ func (db *CacheDB) Flush() error { } // clear MemDB - for _, bucket := range db.mem.buckets { + // + // NOTE: the MemDB is internal to the CacheDB and never has open snapshots, so + // its current generation can be mutated freely + for _, bucket := range db.mem.gen.buckets { clear(bucket) } for _, bucket := range db.mem.puts { @@ -327,13 +447,20 @@ func (db *CacheDB) Flush() error { for _, bucket := range db.mem.dels { clear(bucket) } - return db.db.Flush() + return sp.Flush() } -// Cancel implements DB. +// Cancel implements DBScratchpad. func (db *CacheDB) Cancel() { db.mem.Cancel() - db.db.Cancel() + db.db.Scratchpad().Cancel() +} + +// Snapshot implements DB. +func (db *CacheDB) Snapshot() (DBSnapshot, func()) { + // unflushed writes are cached in memory, so the underlying DB always + // reflects the state as of the last Flush + return db.db.Snapshot() } // NewCacheDB returns a new CacheDB that wraps the given DB. @@ -351,20 +478,19 @@ func check(err error) { } } -// dbBucket is a helper type for implementing Store. -type dbBucket struct { - b DBBucket - db *DBStore +// dbBucketReader is a helper type for implementing the read half of Store. +type dbBucketReader struct { + b DBBucket } -func (b *dbBucket) getRaw(key []byte) []byte { +func (b dbBucketReader) getRaw(key []byte) []byte { if b.b == nil { return nil } return b.b.Get(key) } -func (b *dbBucket) get(key []byte, v types.DecoderFrom) bool { +func (b dbBucketReader) get(key []byte, v types.DecoderFrom) bool { val := b.getRaw(key) if val == nil { return false @@ -378,22 +504,31 @@ func (b *dbBucket) get(key []byte, v types.DecoderFrom) bool { return true } -func (b *dbBucket) putRaw(key, value []byte) { +// dbBucket is a helper type for implementing the write half of Store. +type dbBucket struct { + b DBBucket + sp *dbScratchpad +} + +func (b dbBucket) getRaw(key []byte) []byte { return dbBucketReader{b.b}.getRaw(key) } +func (b dbBucket) get(key []byte, v types.DecoderFrom) bool { return dbBucketReader{b.b}.get(key, v) } + +func (b dbBucket) putRaw(key, value []byte) { check(b.b.Put(key, value)) - b.db.unflushed += len(value) + b.sp.unflushed += len(value) } -func (b *dbBucket) put(key []byte, v types.EncoderTo) { +func (b dbBucket) put(key []byte, v types.EncoderTo) { var buf bytes.Buffer - b.db.enc.Reset(&buf) - v.EncodeTo(&b.db.enc) - b.db.enc.Flush() + b.sp.enc.Reset(&buf) + v.EncodeTo(&b.sp.enc) + b.sp.enc.Flush() b.putRaw(key, buf.Bytes()) } -func (b *dbBucket) delete(key []byte) { +func (b dbBucket) delete(key []byte) { check(b.b.Delete(key)) - b.db.unflushed += len(key) + b.sp.unflushed += len(key) } var ( @@ -412,56 +547,48 @@ var ( // DBStore implements Store using a key-value database. type DBStore struct { - db DB - n *consensus.Network // for getState - enc types.Encoder + db DB + n *consensus.Network + scratchpad *dbScratchpad +} - unflushed int - lastFlush time.Time +// bucket returns a writeable handle for the named bucket. +func (s *dbScratchpad) bucket(name []byte) dbBucket { + return dbBucket{s.sp.Bucket(name), s} } -func (db *DBStore) bucket(name []byte) *dbBucket { - return &dbBucket{db.db.Bucket(name), db} +func readBucket(ss DBSnapshot, name []byte) dbBucketReader { + return dbBucketReader{ss.Bucket(name)} } -func (db *DBStore) encHeight(height uint64) []byte { +func encHeight(height uint64) []byte { var buf [8]byte return binary.BigEndian.AppendUint64(buf[:0], height) } -func (db *DBStore) putBestIndex(index types.ChainIndex) { - db.bucket(bMainChain).put(db.encHeight(index.Height), &index.ID) -} - -func (db *DBStore) deleteBestIndex(height uint64) { - db.bucket(bMainChain).delete(db.encHeight(height)) -} - -func (db *DBStore) getHeight() (height uint64) { - if val := db.bucket(bMainChain).getRaw(keyHeight); len(val) == 8 { +func getHeight(ss DBSnapshot) (height uint64) { + if val := readBucket(ss, bMainChain).getRaw(keyHeight); len(val) == 8 { height = binary.BigEndian.Uint64(val) } return } -func (db *DBStore) putHeight(height uint64) { - db.bucket(bMainChain).putRaw(keyHeight, db.encHeight(height)) -} - -func (db *DBStore) getState(id types.BlockID) (consensus.State, bool) { +func getState(ss DBSnapshot, n *consensus.Network, id types.BlockID) (consensus.State, bool) { var vs versionedState - ok := db.bucket(bStates).get(id[:], &vs) - vs.State.Network = db.n + ok := readBucket(ss, bStates).get(id[:], &vs) + vs.State.Network = n return vs.State, ok } -func (db *DBStore) putState(cs consensus.State) { - db.bucket(bStates).put(cs.Index.ID[:], versionedState{cs}) +func tipState(ss DBSnapshot, n *consensus.Network) consensus.State { + index, _ := bestIndex(ss, getHeight(ss)) + cs, _ := getState(ss, n, index.ID) + return cs } -func (db *DBStore) getBlock(id types.BlockID) (bh types.BlockHeader, b *types.Block, bs *consensus.V1BlockSupplement, _ bool) { +func getBlock(ss DBSnapshot, id types.BlockID) (bh types.BlockHeader, b *types.Block, bs *consensus.V1BlockSupplement, _ bool) { var sb supplementedBlock - if ok := db.bucket(bBlocks).get(id[:], &sb); !ok { + if ok := readBucket(ss, bBlocks).get(id[:], &sb); !ok { return types.BlockHeader{}, nil, nil, false } else if sb.Header == nil { sb.Header = new(types.BlockHeader) @@ -470,13 +597,13 @@ func (db *DBStore) getBlock(id types.BlockID) (bh types.BlockHeader, b *types.Bl return *sb.Header, sb.Block, sb.Supplement, true } -func (db *DBStore) putBlock(bh types.BlockHeader, b *types.Block, bs *consensus.V1BlockSupplement) { +func (s *dbScratchpad) putBlock(bh types.BlockHeader, b *types.Block, bs *consensus.V1BlockSupplement) { id := bh.ID() - db.bucket(bBlocks).put(id[:], supplementedBlock{&bh, b, bs}) + s.bucket(bBlocks).put(id[:], supplementedBlock{&bh, b, bs}) } -func (db *DBStore) getAncestorInfo(id types.BlockID) (parentID types.BlockID, timestamp time.Time, ok bool) { - ok = db.bucket(bBlocks).get(id[:], types.DecoderFunc(func(d *types.Decoder) { +func getAncestorInfo(ss DBSnapshot, id types.BlockID) (parentID types.BlockID, timestamp time.Time, ok bool) { + ok = readBucket(ss, bBlocks).get(id[:], types.DecoderFunc(func(d *types.Decoder) { v := d.ReadUint8() if v != 2 && v != 3 { d.SetErr(fmt.Errorf("incompatible version (%d)", v)) @@ -494,8 +621,8 @@ func (db *DBStore) getAncestorInfo(id types.BlockID) (parentID types.BlockID, ti return } -func (db *DBStore) getBlockHeader(id types.BlockID) (bh types.BlockHeader, ok bool) { - ok = db.bucket(bBlocks).get(id[:], types.DecoderFunc(func(d *types.Decoder) { +func getBlockHeader(ss DBSnapshot, id types.BlockID) (bh types.BlockHeader, ok bool) { + ok = readBucket(ss, bBlocks).get(id[:], types.DecoderFunc(func(d *types.Decoder) { v := d.ReadUint8() if v != 2 && v != 3 { d.SetErr(fmt.Errorf("incompatible version (%d)", v)) @@ -519,7 +646,7 @@ func (db *DBStore) getBlockHeader(id types.BlockID) (bh types.BlockHeader, ok bo return } -func (db *DBStore) treeKey(row, col uint64) []byte { +func treeKey(row, col uint64) []byte { // If we assume that the total number of elements is less than 2^32, we can // pack row and col into one uint32 key. We do this by setting the top 'row' // bits of 'col' to 1. Since each successive row has half as many columns, @@ -528,7 +655,7 @@ func (db *DBStore) treeKey(row, col uint64) []byte { return binary.BigEndian.AppendUint32(buf[:0], uint32(((1<= numLeaves { panic(fmt.Sprintf("leafIndex %v exceeds accumulator size %v", leafIndex, numLeaves)) // should never happen } @@ -540,67 +667,67 @@ func (db *DBStore) getElementProof(leafIndex, numLeaves uint64) (proof []types.H proof = make([]types.Hash256, bits.Len64(leafIndex^numLeaves)-1) for i := range proof { row, col := uint64(i), (leafIndex>>i)^1 - if !db.bucket(bTree).get(db.treeKey(row, col), &proof[i]) { + if !readBucket(ss, bTree).get(treeKey(row, col), &proof[i]) { panic(fmt.Sprintf("missing proof element %v for leaf %v", i, leafIndex)) } } return } -func (db *DBStore) getSiacoinElement(id types.SiacoinOutputID, numLeaves uint64) (sce types.SiacoinElement, ok bool) { - ok = db.bucket(bSiacoinElements).get(id[:], &sce) +func getSiacoinElement(ss DBSnapshot, id types.SiacoinOutputID, numLeaves uint64) (sce types.SiacoinElement, ok bool) { + ok = readBucket(ss, bSiacoinElements).get(id[:], &sce) if ok { - sce.StateElement.MerkleProof = db.getElementProof(sce.StateElement.LeafIndex, numLeaves) + sce.StateElement.MerkleProof = getElementProof(ss, sce.StateElement.LeafIndex, numLeaves) } return } -func (db *DBStore) putSiacoinElement(sce types.SiacoinElement) { +func (s *dbScratchpad) putSiacoinElement(sce types.SiacoinElement) { sce.StateElement.MerkleProof = nil - db.bucket(bSiacoinElements).put(sce.ID[:], sce.Share()) + s.bucket(bSiacoinElements).put(sce.ID[:], sce.Share()) } -func (db *DBStore) deleteSiacoinElement(id types.SiacoinOutputID) { - db.bucket(bSiacoinElements).delete(id[:]) +func (s *dbScratchpad) deleteSiacoinElement(id types.SiacoinOutputID) { + s.bucket(bSiacoinElements).delete(id[:]) } -func (db *DBStore) getSiafundElement(id types.SiafundOutputID, numLeaves uint64) (sfe types.SiafundElement, ok bool) { - ok = db.bucket(bSiafundElements).get(id[:], &sfe) +func getSiafundElement(ss DBSnapshot, id types.SiafundOutputID, numLeaves uint64) (sfe types.SiafundElement, ok bool) { + ok = readBucket(ss, bSiafundElements).get(id[:], &sfe) if ok { - sfe.StateElement.MerkleProof = db.getElementProof(sfe.StateElement.LeafIndex, numLeaves) + sfe.StateElement.MerkleProof = getElementProof(ss, sfe.StateElement.LeafIndex, numLeaves) } return } -func (db *DBStore) putSiafundElement(sfe types.SiafundElement) { +func (s *dbScratchpad) putSiafundElement(sfe types.SiafundElement) { sfe.StateElement.MerkleProof = nil - db.bucket(bSiafundElements).put(sfe.ID[:], sfe.Share()) + s.bucket(bSiafundElements).put(sfe.ID[:], sfe.Share()) } -func (db *DBStore) deleteSiafundElement(id types.SiafundOutputID) { - db.bucket(bSiafundElements).delete(id[:]) +func (s *dbScratchpad) deleteSiafundElement(id types.SiafundOutputID) { + s.bucket(bSiafundElements).delete(id[:]) } -func (db *DBStore) getFileContractElement(id types.FileContractID, numLeaves uint64) (fce types.FileContractElement, ok bool) { - ok = db.bucket(bFileContractElements).get(id[:], &fce) +func getFileContractElement(ss DBSnapshot, id types.FileContractID, numLeaves uint64) (fce types.FileContractElement, ok bool) { + ok = readBucket(ss, bFileContractElements).get(id[:], &fce) if ok { - fce.StateElement.MerkleProof = db.getElementProof(fce.StateElement.LeafIndex, numLeaves) + fce.StateElement.MerkleProof = getElementProof(ss, fce.StateElement.LeafIndex, numLeaves) } return } -func (db *DBStore) putFileContractElement(fce types.FileContractElement) { +func (s *dbScratchpad) putFileContractElement(fce types.FileContractElement) { fce.StateElement.MerkleProof = nil - db.bucket(bFileContractElements).put(fce.ID[:], fce.Share()) + s.bucket(bFileContractElements).put(fce.ID[:], fce.Share()) } -func (db *DBStore) deleteFileContractElement(id types.FileContractID) { - db.bucket(bFileContractElements).delete(id[:]) +func (s *dbScratchpad) deleteFileContractElement(id types.FileContractID) { + s.bucket(bFileContractElements).delete(id[:]) } -func (db *DBStore) putFileContractExpiration(id types.FileContractID, windowEnd uint64, apply bool) { - b := db.bucket(bFileContractElements) - key := db.encHeight(windowEnd) +func (s *dbScratchpad) putFileContractExpiration(id types.FileContractID, windowEnd uint64, apply bool) { + b := s.bucket(bFileContractElements) + key := encHeight(windowEnd) // When applying, we append; when reverting, we prepend. This ensures that // the order of the IDs -- and consequently, the ExpiringFileContracts in a // V1BlockSupplement -- remain stable across reorgs. Without this adjustment, @@ -613,9 +740,8 @@ func (db *DBStore) putFileContractExpiration(id types.FileContractID, windowEnd } } -// ExpiringFileContractIDs returns the expiring file contract IDs at the given height. -func (db *DBStore) ExpiringFileContractIDs(height uint64) []types.FileContractID { - buf := db.bucket(bFileContractElements).getRaw(db.encHeight(height)) +func expiringFileContractIDs(ss DBSnapshot, height uint64) []types.FileContractID { + buf := readBucket(ss, bFileContractElements).getRaw(encHeight(height)) ids := make([]types.FileContractID, 0, len(buf)/32) for i := 0; i < len(buf); i += 32 { ids = append(ids, (types.FileContractID)(buf[i:])) @@ -623,20 +749,9 @@ func (db *DBStore) ExpiringFileContractIDs(height uint64) []types.FileContractID return ids } -// OverwriteExpiringFileContractIDs overwrites the expiring file contract IDs at the given height. -// This should not be called unless the IDs are known to be correct, as it will overwrite -// any existing IDs at that height. -func (db *DBStore) OverwriteExpiringFileContractIDs(height uint64, ids []types.FileContractID) { - buf := make([]byte, len(ids)*32) - for i, id := range ids { - copy(buf[i*32:], id[:]) - } - db.bucket(bFileContractElements).putRaw(db.encHeight(height), buf) -} - -func (db *DBStore) deleteFileContractExpiration(id types.FileContractID, windowEnd uint64) { - b := db.bucket(bFileContractElements) - key := db.encHeight(windowEnd) +func (s *dbScratchpad) deleteFileContractExpiration(id types.FileContractID, windowEnd uint64) { + b := s.bucket(bFileContractElements) + key := encHeight(windowEnd) val := append([]byte(nil), b.getRaw(key)...) for i := 0; i < len(val); i += 32 { if *(*types.FileContractID)(val[i:]) == id { @@ -650,37 +765,27 @@ func (db *DBStore) deleteFileContractExpiration(id types.FileContractID, windowE panic("missing file contract expiration") } -func (db *DBStore) applyState(next consensus.State) { - db.putBestIndex(next.Index) - db.putHeight(next.Index.Height) -} - -func (db *DBStore) revertState(prev consensus.State) { - db.deleteBestIndex(prev.Index.Height + 1) - db.putHeight(prev.Index.Height) -} - -func (db *DBStore) applyElements(cau consensus.ApplyUpdate) { +func (s *dbScratchpad) applyElements(cau consensus.ApplyUpdate) { cau.ForEachTreeNode(func(row, col uint64, h types.Hash256) { - db.bucket(bTree).putRaw(db.treeKey(row, col), h[:]) + s.bucket(bTree).putRaw(treeKey(row, col), h[:]) }) for _, sced := range cau.SiacoinElementDiffs() { if sced.Created && sced.Spent { continue // ephemeral } else if sced.Spent { - db.deleteSiacoinElement(sced.SiacoinElement.ID) + s.deleteSiacoinElement(sced.SiacoinElement.ID) } else { - db.putSiacoinElement(sced.SiacoinElement.Share()) + s.putSiacoinElement(sced.SiacoinElement.Share()) } } for _, sfed := range cau.SiafundElementDiffs() { if sfed.Created && sfed.Spent { continue // ephemeral } else if sfed.Spent { - db.deleteSiafundElement(sfed.SiafundElement.ID) + s.deleteSiafundElement(sfed.SiafundElement.ID) } else { - db.putSiafundElement(sfed.SiafundElement.Share()) + s.putSiafundElement(sfed.SiafundElement.Share()) } } for _, fced := range cau.FileContractElementDiffs() { @@ -688,45 +793,45 @@ func (db *DBStore) applyElements(cau consensus.ApplyUpdate) { if fced.Created && fced.Resolved { continue } else if fced.Resolved { - db.deleteFileContractElement(fce.ID) - db.deleteFileContractExpiration(fce.ID, fce.FileContract.WindowEnd) + s.deleteFileContractElement(fce.ID) + s.deleteFileContractExpiration(fce.ID, fce.FileContract.WindowEnd) } else if fced.Revision != nil { rev := fce.Share() rev.FileContract = *fced.Revision - db.putFileContractElement(rev.Share()) + s.putFileContractElement(rev.Share()) if rev.FileContract.WindowEnd != fce.FileContract.WindowEnd { - db.deleteFileContractExpiration(fce.ID, fce.FileContract.WindowEnd) - db.putFileContractExpiration(fce.ID, rev.FileContract.WindowEnd, true) + s.deleteFileContractExpiration(fce.ID, fce.FileContract.WindowEnd) + s.putFileContractExpiration(fce.ID, rev.FileContract.WindowEnd, true) } } else { - db.putFileContractElement(fce.Share()) - db.putFileContractExpiration(fce.ID, fce.FileContract.WindowEnd, true) + s.putFileContractElement(fce.Share()) + s.putFileContractExpiration(fce.ID, fce.FileContract.WindowEnd, true) } } } -func (db *DBStore) revertElements(cru consensus.RevertUpdate) { +func (s *dbScratchpad) revertElements(cru consensus.RevertUpdate) { for _, fced := range cru.FileContractElementDiffs() { fce := &fced.FileContractElement if fced.Created && fced.Resolved { continue } else if fced.Resolved { // contract no longer resolved; restore it - db.putFileContractElement(fce.Share()) - db.putFileContractExpiration(fce.ID, fce.FileContract.WindowEnd, false) + s.putFileContractElement(fce.Share()) + s.putFileContractExpiration(fce.ID, fce.FileContract.WindowEnd, false) } else if fced.Revision != nil { // contract no longer revised; restore prior revision rev := fce.Share() rev.FileContract = *fced.Revision - db.putFileContractElement(fce.Share()) + s.putFileContractElement(fce.Share()) if rev.FileContract.WindowEnd != fce.FileContract.WindowEnd { - db.deleteFileContractExpiration(fce.ID, rev.FileContract.WindowEnd) - db.putFileContractExpiration(fce.ID, fce.FileContract.WindowEnd, false) + s.deleteFileContractExpiration(fce.ID, rev.FileContract.WindowEnd) + s.putFileContractExpiration(fce.ID, fce.FileContract.WindowEnd, false) } } else { // contract no longer exists; delete it - db.deleteFileContractElement(fce.ID) - db.deleteFileContractExpiration(fce.ID, fce.FileContract.WindowEnd) + s.deleteFileContractElement(fce.ID) + s.deleteFileContractExpiration(fce.ID, fce.FileContract.WindowEnd) } } @@ -735,10 +840,10 @@ func (db *DBStore) revertElements(cru consensus.RevertUpdate) { continue // ephemeral } else if sfed.Spent { // output no longer spent; restore it - db.putSiafundElement(sfed.SiafundElement.Share()) + s.putSiafundElement(sfed.SiafundElement.Share()) } else { // output no longer exists; delete it - db.deleteSiafundElement(sfed.SiafundElement.ID) + s.deleteSiafundElement(sfed.SiafundElement.ID) } } for _, sced := range cru.SiacoinElementDiffs() { @@ -746,15 +851,15 @@ func (db *DBStore) revertElements(cru consensus.RevertUpdate) { continue // ephemeral } else if sced.Spent { // output no longer spent; restore it - db.putSiacoinElement(sced.SiacoinElement.Share()) + s.putSiacoinElement(sced.SiacoinElement.Share()) } else { // output no longer exists; delete it - db.deleteSiacoinElement(sced.SiacoinElement.ID) + s.deleteSiacoinElement(sced.SiacoinElement.ID) } } cru.ForEachTreeNode(func(row, col uint64, h types.Hash256) { - db.bucket(bTree).putRaw(db.treeKey(row, col), h[:]) + s.bucket(bTree).putRaw(treeKey(row, col), h[:]) }) // NOTE: Although the element tree has shrunk, we do not need to explicitly @@ -765,42 +870,38 @@ func (db *DBStore) revertElements(cru consensus.RevertUpdate) { // reclaim it immediately.) } -// BestIndex implements Store. -func (db *DBStore) BestIndex(height uint64) (index types.ChainIndex, ok bool) { +func bestIndex(ss DBSnapshot, height uint64) (index types.ChainIndex, ok bool) { index.Height = height - ok = db.bucket(bMainChain).get(db.encHeight(height), &index.ID) + ok = readBucket(ss, bMainChain).get(encHeight(height), &index.ID) return } -// SupplementTipTransaction implements Store. -func (db *DBStore) SupplementTipTransaction(txn types.Transaction) (ts consensus.V1TransactionSupplement) { - height := db.getHeight() - if height >= db.n.HardforkV2.RequireHeight { +func supplementTipTransaction(ss DBSnapshot, n *consensus.Network, txn types.Transaction) (ts consensus.V1TransactionSupplement) { + if getHeight(ss) >= n.HardforkV2.RequireHeight { return consensus.V1TransactionSupplement{} } // get tip state, for proof-trimming - index, _ := db.BestIndex(height) - cs, _ := db.State(index.ID) + cs := tipState(ss, n) numLeaves := cs.Elements.NumLeaves for _, sci := range txn.SiacoinInputs { - if sce, ok := db.getSiacoinElement(sci.ParentID, numLeaves); ok { + if sce, ok := getSiacoinElement(ss, sci.ParentID, numLeaves); ok { ts.SiacoinInputs = append(ts.SiacoinInputs, sce.Move()) } } for _, sfi := range txn.SiafundInputs { - if sfe, ok := db.getSiafundElement(sfi.ParentID, numLeaves); ok { + if sfe, ok := getSiafundElement(ss, sfi.ParentID, numLeaves); ok { ts.SiafundInputs = append(ts.SiafundInputs, sfe.Move()) } } for _, fcr := range txn.FileContractRevisions { - if fce, ok := db.getFileContractElement(fcr.ParentID, numLeaves); ok { + if fce, ok := getFileContractElement(ss, fcr.ParentID, numLeaves); ok { ts.RevisedFileContracts = append(ts.RevisedFileContracts, fce.Move()) } } for _, sp := range txn.StorageProofs { - if fce, ok := db.getFileContractElement(sp.ParentID, numLeaves); ok { - if windowIndex, ok := db.BestIndex(fce.FileContract.WindowStart - 1); ok { + if fce, ok := getFileContractElement(ss, sp.ParentID, numLeaves); ok { + if windowIndex, ok := bestIndex(ss, fce.FileContract.WindowStart-1); ok { ts.StorageProofs = append(ts.StorageProofs, consensus.V1StorageProofSupplement{ FileContract: fce.Move(), WindowID: windowIndex.ID, @@ -811,27 +912,23 @@ func (db *DBStore) SupplementTipTransaction(txn types.Transaction) (ts consensus return } -// SupplementTipBlock implements Store. -func (db *DBStore) SupplementTipBlock(b types.Block) (bs consensus.V1BlockSupplement) { - height := db.getHeight() - if height >= db.n.HardforkV2.RequireHeight { +func supplementTipBlock(ss DBSnapshot, n *consensus.Network, b types.Block) (bs consensus.V1BlockSupplement) { + if getHeight(ss) >= n.HardforkV2.RequireHeight { return consensus.V1BlockSupplement{Transactions: make([]consensus.V1TransactionSupplement, len(b.Transactions))} } // get tip state, for proof-trimming - index, _ := db.BestIndex(height) - cs, _ := db.State(index.ID) + cs := tipState(ss, n) numLeaves := cs.Elements.NumLeaves bs = consensus.V1BlockSupplement{ Transactions: make([]consensus.V1TransactionSupplement, len(b.Transactions)), } for i, txn := range b.Transactions { - bs.Transactions[i] = db.SupplementTipTransaction(txn) + bs.Transactions[i] = supplementTipTransaction(ss, n, txn) } - ids := db.bucket(bFileContractElements).getRaw(db.encHeight(db.getHeight() + 1)) - for i := 0; i < len(ids); i += 32 { - fce, ok := db.getFileContractElement(*(*types.FileContractID)(ids[i:]), numLeaves) + for _, id := range expiringFileContractIDs(ss, cs.Index.Height+1) { + fce, ok := getFileContractElement(ss, id, numLeaves) if !ok { panic("missing FileContractElement") } @@ -840,16 +937,15 @@ func (db *DBStore) SupplementTipBlock(b types.Block) (bs consensus.V1BlockSupple return bs } -// AncestorTimestamp implements Store. -func (db *DBStore) AncestorTimestamp(id types.BlockID) (t time.Time, ok bool) { - cs, _ := db.State(id) - if cs.Index.Height > db.n.HardforkOak.Height { +func ancestorTimestamp(ss DBSnapshot, n *consensus.Network, id types.BlockID) (t time.Time, ok bool) { + cs, _ := getState(ss, n, id) + if cs.Index.Height > n.HardforkOak.Height { return time.Time{}, true } - getBestID := func(height uint64) (id types.BlockID) { - db.bucket(bMainChain).get(db.encHeight(height), &id) - return + getBestID := func(height uint64) types.BlockID { + index, _ := bestIndex(ss, height) + return index.ID } ancestorID := id for i := uint64(0); i < cs.AncestorDepth() && i < cs.Index.Height; i++ { @@ -861,121 +957,268 @@ func (db *DBStore) AncestorTimestamp(id types.BlockID) (t time.Time, ok bool) { } break } - ancestorID, _, _ = db.getAncestorInfo(ancestorID) + ancestorID, _, _ = getAncestorInfo(ss, ancestorID) } - _, t, ok = db.getAncestorInfo(ancestorID) + _, t, ok = getAncestorInfo(ss, ancestorID) return } -// State implements Store. -func (db *DBStore) State(id types.BlockID) (consensus.State, bool) { - return db.getState(id) +// NOTE: these values were chosen empirically and should constitute a +// sensible default; if necessary, we can make them configurable +const ( + flushSizeThreshold = 100e6 + flushDurationThreshold = 5 * time.Second +) + +func (s *dbScratchpad) shouldFlush() bool { + return s.unflushed >= flushSizeThreshold || time.Since(s.lastFlush) >= flushDurationThreshold } -// AddState implements Store. -func (db *DBStore) AddState(cs consensus.State) { - db.putState(cs) +// flushIfFull flushes the accumulated writes if they exceed the size +// threshold. Unlike shouldFlush, it ignores their age; it is used by methods +// for which an age-triggered flush would be undesirable, either because it +// would commit twice in quick succession (AddState/AddBlock, whose writes are +// shortly followed by an ApplyBlock or an explicit Flush) or because it would +// publish an intermediate reverted tip to snapshot readers (RevertBlock). +func (s *dbScratchpad) flushIfFull() { + if s.unflushed >= flushSizeThreshold { + s.mustFlush() + } } -// Block implements Store. -func (db *DBStore) Block(id types.BlockID) (types.Block, *consensus.V1BlockSupplement, bool) { - _, b, bs, ok := db.getBlock(id) +func (s *dbScratchpad) mustFlush() { check(s.Flush()) } + +// dbScratchpad implements StoreScratchpad. Its read methods observe the +// unflushed writes of the underlying DB scratchpad. +type dbScratchpad struct { + sp DBScratchpad + n *consensus.Network + enc types.Encoder + + unflushed int + lastFlush time.Time + tip consensus.State // updated by ApplyBlock/RevertBlock +} + +// TipState implements StoreSnapshot. +func (s *dbScratchpad) TipState() consensus.State { return s.tip } + +// BestIndex implements StoreSnapshot. +func (s *dbScratchpad) BestIndex(height uint64) (types.ChainIndex, bool) { + return bestIndex(s.sp, height) +} + +// Block implements StoreSnapshot. +func (s *dbScratchpad) Block(id types.BlockID) (types.Block, *consensus.V1BlockSupplement, bool) { + _, b, bs, ok := getBlock(s.sp, id) if !ok || b == nil { return types.Block{}, nil, false } return *b, bs, ok } -// AddBlock implements Store. -func (db *DBStore) AddBlock(b types.Block, bs *consensus.V1BlockSupplement) { - db.putBlock(b.Header(), &b, bs) +// Header implements StoreSnapshot. +func (s *dbScratchpad) Header(id types.BlockID) (types.BlockHeader, bool) { + return getBlockHeader(s.sp, id) } -// PruneBlock implements Store. -func (db *DBStore) PruneBlock(id types.BlockID) { - if bh, _, _, ok := db.getBlock(id); ok { - db.putBlock(bh, nil, nil) - } +// State implements StoreSnapshot. +func (s *dbScratchpad) State(id types.BlockID) (consensus.State, bool) { + return getState(s.sp, s.n, id) +} + +// ExpiringFileContractIDs implements StoreSnapshot. +func (s *dbScratchpad) ExpiringFileContractIDs(height uint64) []types.FileContractID { + return expiringFileContractIDs(s.sp, height) +} + +// AncestorTimestamp implements StoreSnapshot. +func (s *dbScratchpad) AncestorTimestamp(id types.BlockID) (time.Time, bool) { + return ancestorTimestamp(s.sp, s.n, id) +} + +// SupplementTipTransaction implements StoreSnapshot. +func (s *dbScratchpad) SupplementTipTransaction(txn types.Transaction) consensus.V1TransactionSupplement { + return supplementTipTransaction(s.sp, s.n, txn) } -// Header implements Store. -func (db *DBStore) Header(id types.BlockID) (bh types.BlockHeader, exists bool) { - return db.getBlockHeader(id) +// SupplementTipBlock implements StoreSnapshot. +func (s *dbScratchpad) SupplementTipBlock(b types.Block) consensus.V1BlockSupplement { + return supplementTipBlock(s.sp, s.n, b) } -func (db *DBStore) shouldFlush() bool { - // NOTE: these values were chosen empirically and should constitute a - // sensible default; if necessary, we can make them configurable - const flushSizeThreshold = 100e6 - const flushDurationThreshold = 5 * time.Second - return db.unflushed >= flushSizeThreshold || time.Since(db.lastFlush) >= flushDurationThreshold +// AddState implements StoreScratchpad. +func (s *dbScratchpad) AddState(cs consensus.State) { + s.bucket(bStates).put(cs.Index.ID[:], versionedState{cs}) + s.flushIfFull() } -// ApplyBlock implements Store. -func (db *DBStore) ApplyBlock(s consensus.State, cau consensus.ApplyUpdate) { - db.applyState(s) - if s.Index.Height <= db.n.HardforkV2.RequireHeight { - db.applyElements(cau) +// AddBlock implements StoreScratchpad. +func (s *dbScratchpad) AddBlock(b types.Block, bs *consensus.V1BlockSupplement) { + s.putBlock(b.Header(), &b, bs) + s.flushIfFull() +} + +// PruneBlock implements StoreScratchpad. +func (s *dbScratchpad) PruneBlock(id types.BlockID) { + if bh, _, _, ok := getBlock(s.sp, id); ok { + s.putBlock(bh, nil, nil) } - if db.shouldFlush() { - if err := db.Flush(); err != nil { - panic(err) - } + if s.shouldFlush() { + s.mustFlush() + } +} + +// OverwriteExpiringFileContractIDs implements StoreScratchpad. This should +// not be called unless the IDs are known to be correct, as it will overwrite +// any existing IDs at that height. +func (s *dbScratchpad) OverwriteExpiringFileContractIDs(height uint64, ids []types.FileContractID) { + buf := make([]byte, len(ids)*32) + for i, id := range ids { + copy(buf[i*32:], id[:]) + } + s.bucket(bFileContractElements).putRaw(encHeight(height), buf) +} + +// ApplyBlock implements StoreScratchpad. +func (s *dbScratchpad) ApplyBlock(cs consensus.State, cau consensus.ApplyUpdate) { + s.bucket(bMainChain).put(encHeight(cs.Index.Height), &cs.Index.ID) + s.bucket(bMainChain).putRaw(keyHeight, encHeight(cs.Index.Height)) + if cs.Index.Height <= s.n.HardforkV2.RequireHeight { + s.applyElements(cau) + } + s.tip = cs + if s.shouldFlush() { + s.mustFlush() } } -// RevertBlock implements Store. -func (db *DBStore) RevertBlock(s consensus.State, cru consensus.RevertUpdate) { - if s.Index.Height <= db.n.HardforkV2.RequireHeight { - db.revertElements(cru) +// RevertBlock implements StoreScratchpad. +func (s *dbScratchpad) RevertBlock(cs consensus.State, cru consensus.RevertUpdate) { + if cs.Index.Height <= s.n.HardforkV2.RequireHeight { + s.revertElements(cru) } - db.revertState(s) - if db.shouldFlush() { - if err := db.Flush(); err != nil { - panic(err) + s.bucket(bMainChain).delete(encHeight(cs.Index.Height + 1)) + s.bucket(bMainChain).putRaw(keyHeight, encHeight(cs.Index.Height)) + s.tip = cs + s.flushIfFull() +} + +// Flush implements StoreScratchpad. +func (s *dbScratchpad) Flush() error { + if s.unflushed > 0 { + if err := s.sp.Flush(); err != nil { + return err } + s.unflushed = 0 + s.lastFlush = time.Now() } + return nil } -// Flush flushes any uncommitted data to the underlying DB. -func (db *DBStore) Flush() error { - if db.unflushed == 0 { - return nil +// Scratchpad implements Store. +func (db *DBStore) Scratchpad() StoreScratchpad { return db.scratchpad } + +// dbSnapshot implements StoreSnapshot. +type dbSnapshot struct { + ss DBSnapshot + n *consensus.Network + + tipOnce sync.Once + tip consensus.State +} + +// TipState implements StoreSnapshot. +func (s *dbSnapshot) TipState() consensus.State { + // The tip is derived from the snapshot itself, so it is always consistent + // with the snapshot's data; no synchronization with the scratchpad is + // required. + s.tipOnce.Do(func() { + s.tip = tipState(s.ss, s.n) + }) + return s.tip +} + +// BestIndex implements StoreSnapshot. +func (s *dbSnapshot) BestIndex(height uint64) (types.ChainIndex, bool) { + return bestIndex(s.ss, height) +} + +// Block implements StoreSnapshot. +func (s *dbSnapshot) Block(id types.BlockID) (types.Block, *consensus.V1BlockSupplement, bool) { + _, b, bs, ok := getBlock(s.ss, id) + if !ok || b == nil { + return types.Block{}, nil, false } - err := db.db.Flush() - db.unflushed = 0 - db.lastFlush = time.Now() - return err + return *b, bs, ok } -// NewDBStore creates a new DBStore using the provided database. The tip state -// is also returned. The DB will be automatically migrated if necessary. The -// provided logger may be nil. -func NewDBStore(db DB, n *consensus.Network, genesisBlock types.Block, logger MigrationLogger) (_ *DBStore, _ consensus.State, err error) { +// Header implements StoreSnapshot. +func (s *dbSnapshot) Header(id types.BlockID) (types.BlockHeader, bool) { + return getBlockHeader(s.ss, id) +} + +// State implements StoreSnapshot. +func (s *dbSnapshot) State(id types.BlockID) (consensus.State, bool) { + return getState(s.ss, s.n, id) +} + +// ExpiringFileContractIDs implements StoreSnapshot. +func (s *dbSnapshot) ExpiringFileContractIDs(height uint64) []types.FileContractID { + return expiringFileContractIDs(s.ss, height) +} + +// AncestorTimestamp implements StoreSnapshot. +func (s *dbSnapshot) AncestorTimestamp(id types.BlockID) (time.Time, bool) { + return ancestorTimestamp(s.ss, s.n, id) +} + +// SupplementTipTransaction implements StoreSnapshot. +func (s *dbSnapshot) SupplementTipTransaction(txn types.Transaction) consensus.V1TransactionSupplement { + return supplementTipTransaction(s.ss, s.n, txn) +} + +// SupplementTipBlock implements StoreSnapshot. +func (s *dbSnapshot) SupplementTipBlock(b types.Block) consensus.V1BlockSupplement { + return supplementTipBlock(s.ss, s.n, b) +} + +// Snapshot implements Store. +func (db *DBStore) Snapshot() (StoreSnapshot, func()) { + ss, release := db.db.Snapshot() + return &dbSnapshot{ss: ss, n: db.n}, release +} + +// NewDBStore creates a new DBStore using the provided database. The DB will +// be automatically migrated if necessary. The provided logger may be nil. +func NewDBStore(db DB, n *consensus.Network, genesisBlock types.Block, logger MigrationLogger) (_ *DBStore, err error) { + sp := db.Scratchpad() // during initialization, we should return an error instead of panicking defer func() { if r := recover(); r != nil { - db.Cancel() + sp.Cancel() err = fmt.Errorf("panic during database initialization: %v", r) } }() if err := sanityCheckNetwork(n); err != nil { - return nil, consensus.State{}, fmt.Errorf("invalid network: %w", err) + return nil, fmt.Errorf("invalid network: %w", err) } // don't accidentally overwrite a siad database - if db.Bucket([]byte("ChangeLog")) != nil { - return nil, consensus.State{}, errors.New("detected siad database, refusing to proceed") + if sp.Bucket([]byte("ChangeLog")) != nil { + return nil, errors.New("detected siad database, refusing to proceed") } dbs := &DBStore{ - db: db, - n: n, + db: db, + n: n, + scratchpad: &dbScratchpad{sp: sp, n: n, lastFlush: time.Now()}, } + scratch := dbs.scratchpad // if the db is empty, initialize it - if version := dbs.bucket(bVersion).getRaw(bVersion); len(version) != 1 { + if version := readBucket(sp, bVersion).getRaw(bVersion); len(version) != 1 { for _, bucket := range [][]byte{ bVersion, bNetwork, @@ -987,71 +1230,73 @@ func NewDBStore(db DB, n *consensus.Network, genesisBlock types.Block, logger Mi bSiafundElements, bTree, } { - if _, err := db.CreateBucket(bucket); err != nil { + if _, err := sp.CreateBucket(bucket); err != nil { panic(err) } } - dbs.bucket(bVersion).putRaw(bVersion, []byte{4}) - dbs.bucket(bNetwork).putRaw(bNetwork, []byte(n.Name)) + scratch.bucket(bVersion).putRaw(bVersion, []byte{4}) + scratch.bucket(bNetwork).putRaw(bNetwork, []byte(n.Name)) // store genesis state and apply genesis block to it genesisState := n.GenesisState() - dbs.putState(genesisState) + scratch.AddState(genesisState) bs := consensus.V1BlockSupplement{Transactions: make([]consensus.V1TransactionSupplement, len(genesisBlock.Transactions))} cs, cau := consensus.ApplyBlock(genesisState, genesisBlock, bs, time.Time{}) - dbs.putBlock(genesisBlock.Header(), &genesisBlock, &bs) - dbs.putState(cs) - dbs.ApplyBlock(cs, cau) - if err := dbs.Flush(); err != nil { - return nil, consensus.State{}, err + scratch.AddBlock(genesisBlock, &bs) + scratch.AddState(cs) + scratch.ApplyBlock(cs, cau) + if err := scratch.Flush(); err != nil { + return nil, err } } else if version[0] != 4 { if logger == nil { logger = noopLogger{} } if err := migrateDB(dbs, logger); err != nil { - return nil, consensus.State{}, fmt.Errorf("failed to migrate database: %w", err) + return nil, fmt.Errorf("failed to migrate database: %w", err) } } - if network := dbs.bucket(bNetwork).getRaw(bNetwork); len(network) != 0 && string(network) != n.Name { - return nil, consensus.State{}, fmt.Errorf("database previously initialized with different network (%s)", string(network)) + if network := readBucket(sp, bNetwork).getRaw(bNetwork); len(network) != 0 && string(network) != n.Name { + return nil, fmt.Errorf("database previously initialized with different network (%s)", string(network)) } // load tip state - index, _ := dbs.BestIndex(dbs.getHeight()) - cs, _ := dbs.State(index.ID) - return dbs, cs, err + dbs.scratchpad.tip = tipState(sp, n) + return dbs, err } // NewDBStoreAtCheckpoint creates a DBStore initialized at the provided // checkpoint. The checkpoint must be a v2 block. If the DB already exists, the // checkpoint will be set as its new tip. The DB will be automatically migrated // if necessary. The provided logger may be nil. -func NewDBStoreAtCheckpoint(db DB, cs consensus.State, b types.Block, logger MigrationLogger) (_ *DBStore, _ consensus.State, err error) { +func NewDBStoreAtCheckpoint(db DB, cs consensus.State, b types.Block, logger MigrationLogger) (_ *DBStore, err error) { + sp := db.Scratchpad() // during initialization, we should return an error instead of panicking defer func() { if r := recover(); r != nil { - db.Cancel() + sp.Cancel() err = fmt.Errorf("panic during database initialization: %v", r) } }() if err := sanityCheckNetwork(cs.Network); err != nil { - return nil, consensus.State{}, fmt.Errorf("invalid network: %w", err) + return nil, fmt.Errorf("invalid network: %w", err) } // don't accidentally overwrite a siad database - if db.Bucket([]byte("ChangeLog")) != nil { - return nil, consensus.State{}, errors.New("detected siad database, refusing to proceed") + if sp.Bucket([]byte("ChangeLog")) != nil { + return nil, errors.New("detected siad database, refusing to proceed") } dbs := &DBStore{ - db: db, - n: cs.Network, + db: db, + n: cs.Network, + scratchpad: &dbScratchpad{sp: sp, n: cs.Network, lastFlush: time.Now()}, } + scratch := dbs.scratchpad // if the db is empty, initialize it - if version := dbs.bucket(bVersion).getRaw(bVersion); len(version) != 1 { + if version := readBucket(sp, bVersion).getRaw(bVersion); len(version) != 1 { for _, bucket := range [][]byte{ bVersion, bNetwork, @@ -1063,32 +1308,32 @@ func NewDBStoreAtCheckpoint(db DB, cs consensus.State, b types.Block, logger Mig bSiafundElements, bTree, } { - if _, err := db.CreateBucket(bucket); err != nil { + if _, err := sp.CreateBucket(bucket); err != nil { panic(err) } } - dbs.bucket(bVersion).putRaw(bVersion, []byte{4}) - dbs.bucket(bNetwork).putRaw(bNetwork, []byte(cs.Network.Name)) + scratch.bucket(bVersion).putRaw(bVersion, []byte{4}) + scratch.bucket(bNetwork).putRaw(bNetwork, []byte(cs.Network.Name)) } else if version[0] != 4 { if logger == nil { logger = noopLogger{} } if err := migrateDB(dbs, logger); err != nil { - return nil, consensus.State{}, fmt.Errorf("failed to migrate database: %w", err) + return nil, fmt.Errorf("failed to migrate database: %w", err) } } - if network := dbs.bucket(bNetwork).getRaw(bNetwork); len(network) != 0 && string(network) != cs.Network.Name { - return nil, consensus.State{}, fmt.Errorf("database previously initialized with different network (%s)", string(network)) + if network := readBucket(sp, bNetwork).getRaw(bNetwork); len(network) != 0 && string(network) != cs.Network.Name { + return nil, fmt.Errorf("database previously initialized with different network (%s)", string(network)) } - dbs.putState(cs) + scratch.AddState(cs) bs := consensus.V1BlockSupplement{Transactions: make([]consensus.V1TransactionSupplement, len(b.Transactions))} cs, cau := consensus.ApplyBlock(cs, b, bs, time.Time{}) - dbs.putBlock(b.Header(), &b, &bs) - dbs.putState(cs) - dbs.ApplyBlock(cs, cau) - if err := dbs.Flush(); err != nil { - return nil, consensus.State{}, err + scratch.AddBlock(b, &bs) + scratch.AddState(cs) + scratch.ApplyBlock(cs, cau) + if err := scratch.Flush(); err != nil { + return nil, err } - return dbs, cs, nil + return dbs, nil } diff --git a/chain/db_test.go b/chain/db_test.go index d13e1ffa..17be7699 100644 --- a/chain/db_test.go +++ b/chain/db_test.go @@ -5,52 +5,21 @@ import ( "encoding/json" "errors" "math/bits" + "path/filepath" "reflect" "testing" "time" "go.sia.tech/core/consensus" "go.sia.tech/core/types" + "go.sia.tech/coreutils" "go.sia.tech/coreutils/chain" "go.sia.tech/coreutils/testutil" "lukechampine.com/frand" ) -// NOTE: due to a bug in the transaction validation code, calculating payouts -// is way harder than it needs to be. Tax is calculated on the post-tax -// contract payout (instead of the sum of the renter and host payouts). So the -// equation for the payout is: -// -// payout = renterPayout + hostPayout + payout*tax -// ∴ payout = (renterPayout + hostPayout) / (1 - tax) -// -// This would work if 'tax' were a simple fraction, but because the tax must -// be evenly distributed among siafund holders, 'tax' is actually a function -// that multiplies by a fraction and then rounds down to the nearest multiple -// of the siafund count. Thus, when inverting the function, we have to make an -// initial guess and then fix the rounding error. func taxAdjustedPayout(target types.Currency) types.Currency { - // compute initial guess as target * (1 / 1-tax); since this does not take - // the siafund rounding into account, the guess will be up to - // types.SiafundCount greater than the actual payout value. guess := target.Mul64(1000).Div64(961) - - // now, adjust the guess to remove the rounding error. We know that: - // - // (target % types.SiafundCount) == (payout % types.SiafundCount) - // - // therefore, we can simply adjust the guess to have this remainder as - // well. The only wrinkle is that, since we know guess >= payout, if the - // guess remainder is smaller than the target remainder, we must subtract - // an extra types.SiafundCount. - // - // for example, if target = 87654321 and types.SiafundCount = 10000, then: - // - // initial_guess = 87654321 * (1 / (1 - tax)) - // = 91211572 - // target % 10000 = 4321 - // adjusted_guess = 91204321 - mod64 := func(c types.Currency, v uint64) types.Currency { var r uint64 if c.Hi < v { @@ -72,20 +41,22 @@ func taxAdjustedPayout(target types.Currency) types.Currency { func TestGetEmptyBlockID(t *testing.T) { n, genesisBlock := testutil.V2Network() - store, tipState, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) + store, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := chain.NewManager(store, tipState) + cm := chain.NewManager(store) _, _ = cm.Block(types.BlockID{}) } func TestExpiringFileContracts(t *testing.T) { n, genesisBlock := chain.TestnetZen() - store, cs, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) + store, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } + sp := store.Scratchpad() + cs := sp.TipState() // create two file contracts with the same expiration height b := types.Block{ @@ -98,35 +69,35 @@ func TestExpiringFileContracts(t *testing.T) { }, }}, } - bs := store.SupplementTipBlock(b) + bs := sp.SupplementTipBlock(b) var cau consensus.ApplyUpdate cs, cau = consensus.ApplyBlock(cs, b, bs, time.Time{}) - store.AddState(cs) - store.AddBlock(b, &bs) - store.ApplyBlock(cs, cau) + sp.AddState(cs) + sp.AddBlock(b, &bs) + sp.ApplyBlock(cs, cau) // apply another block, causing the expired contracts to be removed b = types.Block{ ParentID: cs.Index.ID, MinerPayouts: []types.SiacoinOutput{{Value: cs.BlockReward()}}, } - bs = store.SupplementTipBlock(b) + bs = sp.SupplementTipBlock(b) if len(bs.ExpiringFileContracts) != 2 { t.Fatalf("expected 2 file contracts, got %d", len(bs.ExpiringFileContracts)) } cs, cau = consensus.ApplyBlock(cs, b, bs, time.Time{}) - store.AddState(cs) - store.AddBlock(b, &bs) - store.ApplyBlock(cs, cau) + sp.AddState(cs) + sp.AddBlock(b, &bs) + sp.ApplyBlock(cs, cau) // revert the block, causing the expired contracts to be re-created - prev, _ := store.State(b.ParentID) + prev, _ := sp.State(b.ParentID) cru := consensus.RevertBlock(prev, b, bs) - store.RevertBlock(prev, cru) + sp.RevertBlock(prev, cru) // the supplement for the next block should contain the same expiring // contracts as before, in the same order - bs2 := store.SupplementTipBlock(types.Block{ParentID: cs.Index.ID}) + bs2 := sp.SupplementTipBlock(types.Block{ParentID: cs.Index.ID}) if !reflect.DeepEqual(bs, bs2) { t.Fatalf("expected supplements to be the same") } @@ -147,11 +118,11 @@ func TestReorgExpiringFileContractOrder(t *testing.T) { giftAmount := types.Siacoins(1000) giftUTXOID := genesis.Transactions[0].SiacoinOutputID(0) - db1, ts1, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) + db1, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) if err != nil { t.Fatal(err) } - cm1 := chain.NewManager(db1, ts1) + cm1 := chain.NewManager(db1) newFileContract := func() types.FileContract { return types.FileContract{ @@ -273,11 +244,11 @@ func TestReorgExpiringFileContractOrder(t *testing.T) { blocks = append(blocks, cau.Block) } - db2, ts2, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) + db2, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) if err != nil { t.Fatal(err) } - cm2 := chain.NewManager(db2, ts2) + cm2 := chain.NewManager(db2) // applying the blocks should fail because the expiring file contract // order is not the same because of the missing revert. @@ -286,7 +257,7 @@ func TestReorgExpiringFileContractOrder(t *testing.T) { } // reinit the chain manager with the correct expiring contract order - cm2 = chain.NewManager(db2, ts2, chain.WithExpiringContractOrder(map[types.BlockID][]types.FileContractID{ + cm2 = chain.NewManager(db2, chain.WithExpiringContractOrder(map[types.BlockID][]types.FileContractID{ expirationIndex.ID: { contractD, contractA, @@ -298,11 +269,11 @@ func TestReorgExpiringFileContractOrder(t *testing.T) { } // init a fresh chain manager with the correct expiring contract order - db3, ts3, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) + db3, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) if err != nil { t.Fatal(err) } - cm3 := chain.NewManager(db3, ts3, chain.WithExpiringContractOrder(map[types.BlockID][]types.FileContractID{ + cm3 := chain.NewManager(db3, chain.WithExpiringContractOrder(map[types.BlockID][]types.FileContractID{ expirationIndex.ID: { contractD, contractA, @@ -342,17 +313,20 @@ func TestReorgExpiringFileContractOrder(t *testing.T) { func TestPruneBlocks(t *testing.T) { n, genesisBlock := testutil.V2Network() - store, tipState, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) + store, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := chain.NewManager(store, tipState) + cm := chain.NewManager(store) // mine a bunch of blocks testutil.MineBlocks(t, cm, types.VoidAddress, 100) // prune up to height 50 cm.PruneBlocks(50) + // mine another block; pruned blocks are only guaranteed to disappear from + // read-only methods at the next commit + testutil.MineBlocks(t, cm, types.VoidAddress, 1) // ensure blocks < 50 are pruned for height := range uint64(50) { @@ -375,3 +349,92 @@ func TestPruneBlocks(t *testing.T) { } } } + +func testDBSnapshot(t *testing.T, db chain.DB) { + t.Helper() + + sp := db.Scratchpad() + b, err := sp.CreateBucket([]byte("test")) + if err != nil { + t.Fatal(err) + } else if err := b.Put([]byte("foo"), []byte("bar")); err != nil { + t.Fatal(err) + } else if err := sp.Flush(); err != nil { + t.Fatal(err) + } + + // unflushed writes should be visible via the scratchpad, but not via + // Snapshot + b = sp.Bucket([]byte("test")) + if err := b.Put([]byte("baz"), []byte("quux")); err != nil { + t.Fatal(err) + } else if err := b.Delete([]byte("foo")); err != nil { + t.Fatal(err) + } else if !bytes.Equal(b.Get([]byte("baz")), []byte("quux")) { + t.Fatal("expected unflushed write to be visible via the scratchpad") + } + var iterated int + for k, v := range b.Iter() { + if !bytes.Equal(k, []byte("baz")) || !bytes.Equal(v, []byte("quux")) { + t.Fatal("unexpected key-value pair:", string(k), string(v)) + } + iterated++ + } + if iterated != 1 { + t.Fatal("expected unflushed write to be visible via scratchpad iteration") + } + snapshot, release := db.Snapshot() + if vb := snapshot.Bucket([]byte("nonexistent")); vb != nil { + t.Fatal("expected nil bucket for nonexistent bucket") + } + vb := snapshot.Bucket([]byte("test")) + if vb == nil { + t.Fatal("expected snapshot bucket to exist") + } else if vb.Get([]byte("baz")) != nil { + t.Fatal("expected unflushed write to not be visible via Snapshot") + } else if !bytes.Equal(vb.Get([]byte("foo")), []byte("bar")) { + t.Fatal("expected flushed write to be visible via Snapshot") + } + + // open snapshots are unaffected by Flush + if err := sp.Flush(); err != nil { + t.Fatal(err) + } else if vb.Get([]byte("baz")) != nil { + t.Fatal("expected open snapshot to be unaffected by Flush") + } else if !bytes.Equal(vb.Get([]byte("foo")), []byte("bar")) { + t.Fatal("expected open snapshot to be unaffected by Flush") + } + release() + + // a new snapshot should see the flushed writes + snapshot, release = db.Snapshot() + defer release() + vb = snapshot.Bucket([]byte("test")) + if !bytes.Equal(vb.Get([]byte("baz")), []byte("quux")) { + t.Fatal("expected flushed write to be visible via Snapshot") + } else if vb.Get([]byte("foo")) != nil { + t.Fatal("expected flushed delete to be visible via Snapshot") + } + for k, v := range vb.Iter() { + if !bytes.Equal(k, []byte("baz")) || !bytes.Equal(v, []byte("quux")) { + t.Fatal("unexpected key-value pair:", string(k), string(v)) + } + } +} + +func TestMemDBSnapshot(t *testing.T) { + testDBSnapshot(t, chain.NewMemDB()) +} + +func TestCacheDBSnapshot(t *testing.T) { + testDBSnapshot(t, chain.NewCacheDB(chain.NewMemDB())) +} + +func TestBoltDBSnapshot(t *testing.T) { + bdb, err := coreutils.OpenBoltChainDB(filepath.Join(t.TempDir(), "test.db")) + if err != nil { + t.Fatal(err) + } + defer bdb.Close() + testDBSnapshot(t, bdb) +} diff --git a/chain/manager.go b/chain/manager.go index 31a26364..31ac2f6b 100644 --- a/chain/manager.go +++ b/chain/manager.go @@ -44,35 +44,62 @@ type RevertUpdate struct { State consensus.State // post-reversion, i.e. pre-application } -// A Store durably commits Manager-related data to storage. I/O errors must be -// handled internally, e.g. by panicking or calling os.Exit. +// A Store durably commits blockchain data to storage. type Store interface { + // Snapshot returns a read-only snapshot of the Store's data as of the most + // recent Flush. Snapshots are unaffected by concurrent scratchpad writes + // and Flushes. The release function must be called when the snapshot is no + // longer needed; failure to do so may pin resources indefinitely and + // prevent the Store from shutting down. + Snapshot() (ss StoreSnapshot, release func()) + // Scratchpad returns a handle to the current scratchpad, which observes + // the Store's uncommitted data. + Scratchpad() StoreScratchpad +} + +// A StoreSnapshot is a consistent view of blockchain state as of a particular +// tip. +type StoreSnapshot interface { + TipState() consensus.State BestIndex(height uint64) (types.ChainIndex, bool) + Block(id types.BlockID) (types.Block, *consensus.V1BlockSupplement, bool) + Header(id types.BlockID) (types.BlockHeader, bool) + State(id types.BlockID) (consensus.State, bool) + SupplementTipTransaction(txn types.Transaction) consensus.V1TransactionSupplement SupplementTipBlock(b types.Block) consensus.V1BlockSupplement - Block(id types.BlockID) (types.Block, *consensus.V1BlockSupplement, bool) - Header(id types.BlockID) (types.BlockHeader, bool) + ExpiringFileContractIDs(height uint64) []types.FileContractID + AncestorTimestamp(id types.BlockID) (time.Time, bool) +} + +// A StoreScratchpad accumulates writes to a Store. Its methods observe the +// accumulated writes; in particular, ApplyBlock and RevertBlock update the +// tip state reported by TipState. It is not safe for concurrent use. +// +// Implementations are allowed to call Flush internally, so long as they do so +// in a way that preserves the consistency of the resulting snapshot. In +// general, flushing after a method is safe, while flushing mid-method is not. +type StoreScratchpad interface { + StoreSnapshot + + AddState(cs consensus.State) AddBlock(b types.Block, bs *consensus.V1BlockSupplement) PruneBlock(id types.BlockID) - State(id types.BlockID) (consensus.State, bool) - AddState(cs consensus.State) - AncestorTimestamp(id types.BlockID) (time.Time, bool) - ExpiringFileContractIDs(height uint64) []types.FileContractID OverwriteExpiringFileContractIDs(height uint64, ids []types.FileContractID) - // ApplyBlock and RevertBlock are free to commit whenever they see fit. ApplyBlock(s consensus.State, cau consensus.ApplyUpdate) RevertBlock(s consensus.State, cru consensus.RevertUpdate) + Flush() error } // blockAndParent returns the block with the specified ID, along with its parent // state. -func blockAndParent(s Store, id types.BlockID) (types.Block, *consensus.V1BlockSupplement, consensus.State, bool) { - b, bs, ok := s.Block(id) - cs, ok2 := s.State(b.ParentID) +func blockAndParent(ss StoreSnapshot, id types.BlockID) (types.Block, *consensus.V1BlockSupplement, consensus.State, bool) { + b, bs, ok := ss.Block(id) + cs, ok2 := ss.State(b.ParentID) return b, bs, cs, ok && ok2 } @@ -80,7 +107,6 @@ func blockAndParent(s Store, id types.BlockID) (types.Block, *consensus.V1BlockS // chain. type Manager struct { store Store - tipState consensus.State onReorg map[[16]byte]func(types.ChainIndex) onPool map[[16]byte]func() expiringFileContractOrder map[types.BlockID][]types.FileContractID @@ -99,14 +125,30 @@ type Manager struct { lastRevertedV2 []types.V2Transaction } + // guards scratchpad and txpool mu sync.Mutex } +// scratchpad is called (and its 'unlock' function deferred) at the top of all +// Manager methods that mutate state. +func (m *Manager) scratchpad() (sp StoreScratchpad, unlock func()) { + m.mu.Lock() + // NOTE: sync.Once makes it possible to unlock before the end of a function + // without causing the 'defer unlock()' to panic + var once sync.Once + return m.store.Scratchpad(), func() { once.Do(m.mu.Unlock) } +} + +// snapshot is called at the top of all Manager methods that read state. +func (m *Manager) snapshot() (ss StoreSnapshot, release func()) { + return m.store.Snapshot() +} + // TipState returns the consensus state for the current tip. func (m *Manager) TipState() consensus.State { - m.mu.Lock() - defer m.mu.Unlock() - return m.tipState + ss, release := m.snapshot() + defer release() + return ss.TipState() } // Tip returns the tip of the best known valid chain. @@ -116,36 +158,36 @@ func (m *Manager) Tip() types.ChainIndex { // Block returns the block with the specified ID. func (m *Manager) Block(id types.BlockID) (types.Block, bool) { - m.mu.Lock() - defer m.mu.Unlock() - b, _, ok := m.store.Block(id) + ss, release := m.snapshot() + defer release() + b, _, ok := ss.Block(id) return b, ok } // State returns the state with the specified ID. func (m *Manager) State(id types.BlockID) (consensus.State, bool) { - m.mu.Lock() - defer m.mu.Unlock() - return m.store.State(id) + ss, release := m.snapshot() + defer release() + return ss.State(id) } // BestIndex returns the index of the block at the specified height within the // best chain. func (m *Manager) BestIndex(height uint64) (types.ChainIndex, bool) { - m.mu.Lock() - defer m.mu.Unlock() - return m.store.BestIndex(height) + ss, release := m.snapshot() + defer release() + return ss.BestIndex(height) } // MinReorgIndex returns the index on the best chain below which the manager // cannot perform a reorg. func (m *Manager) MinReorgIndex() types.ChainIndex { - m.mu.Lock() - defer m.mu.Unlock() - index := m.tipState.Index + ss, release := m.snapshot() + defer release() + index := ss.TipState().Index for index.Height > 0 { - prevIndex, ok := m.store.BestIndex(index.Height - 1) - _, _, ok2 := m.store.Block(prevIndex.ID) + prevIndex, ok := ss.BestIndex(index.Height - 1) + _, _, ok2 := ss.Block(prevIndex.ID) if !ok || !ok2 { break } @@ -158,10 +200,10 @@ func (m *Manager) MinReorgIndex() types.ChainIndex { // the 10 most-recent blocks, and subsequently spaced exponentionally farther // apart until reaching the genesis block. func (m *Manager) History() ([32]types.BlockID, error) { - m.mu.Lock() - defer m.mu.Unlock() + ss, release := m.snapshot() + defer release() - tipHeight := m.tipState.Index.Height + tipHeight := ss.TipState().Index.Height histHeight := func(i int) uint64 { offset := uint64(i) if offset >= 10 { @@ -174,7 +216,7 @@ func (m *Manager) History() ([32]types.BlockID, error) { } var history [32]types.BlockID for i := range history { - index, ok := m.store.BestIndex(histHeight(i)) + index, ok := ss.BestIndex(histHeight(i)) if !ok { break } @@ -187,22 +229,23 @@ func (m *Manager) History() ([32]types.BlockID, error) { // which must be on the best chain. It also returns the number of headers // between the end of the returned slice and the current tip. func (m *Manager) Headers(index types.ChainIndex, maxHeaders uint64) ([]types.BlockHeader, uint64, error) { - m.mu.Lock() - defer m.mu.Unlock() - if bestIndex, ok := m.store.BestIndex(index.Height); !ok || bestIndex != index { + ss, release := m.snapshot() + defer release() + tip := ss.TipState() + if bestIndex, ok := ss.BestIndex(index.Height); !ok || bestIndex != index { return nil, 0, fmt.Errorf("index %v is not on our best chain", index) } - maxHeaders = min(maxHeaders, m.tipState.Index.Height-index.Height) + maxHeaders = min(maxHeaders, tip.Index.Height-index.Height) headers := make([]types.BlockHeader, maxHeaders) for i := range headers { - index, _ := m.store.BestIndex(index.Height + uint64(i) + 1) - bh, ok := m.store.Header(index.ID) + index, _ := ss.BestIndex(index.Height + uint64(i) + 1) + bh, ok := ss.Header(index.ID) if !ok { return nil, 0, fmt.Errorf("missing block header %v", index) } headers[i] = bh } - return headers, m.tipState.Index.Height - (index.Height + maxHeaders), nil + return headers, tip.Index.Height - (index.Height + maxHeaders), nil } // BlocksForHistory returns up to maxBlocks consecutive blocks from the best @@ -211,56 +254,57 @@ func (m *Manager) Headers(index types.ChainIndex, maxHeaders uint64) ([]types.Bl // returns the number of blocks between the end of the returned slice and the // current tip. func (m *Manager) BlocksForHistory(history []types.BlockID, maxBlocks uint64) ([]types.Block, uint64, error) { - m.mu.Lock() - defer m.mu.Unlock() + ss, release := m.snapshot() + defer release() + tip := ss.TipState() var attachHeight uint64 for _, id := range history { - if cs, ok := m.store.State(id); !ok { + if cs, ok := ss.State(id); !ok { continue - } else if index, ok := m.store.BestIndex(cs.Index.Height); ok && index == cs.Index { + } else if index, ok := ss.BestIndex(cs.Index.Height); ok && index == cs.Index { attachHeight = cs.Index.Height break } } - if maxBlocks > m.tipState.Index.Height-attachHeight { - maxBlocks = m.tipState.Index.Height - attachHeight + if maxBlocks > tip.Index.Height-attachHeight { + maxBlocks = tip.Index.Height - attachHeight } blocks := make([]types.Block, maxBlocks) for i := range blocks { - index, ok := m.store.BestIndex(attachHeight + uint64(i) + 1) + index, ok := ss.BestIndex(attachHeight + uint64(i) + 1) if !ok { return nil, 0, fmt.Errorf("unknown block at height %v", attachHeight+uint64(i)+1) } - b, _, ok := m.store.Block(index.ID) + b, _, ok := ss.Block(index.ID) if !ok { return nil, 0, fmt.Errorf("missing block %v", index) } blocks[i] = b } - return blocks, m.tipState.Index.Height - (attachHeight + maxBlocks), nil + return blocks, tip.Index.Height - (attachHeight + maxBlocks), nil } // AddBlocks ingests a chain of blocks. If the blocks are valid, the chain they // belong to may become the new best chain, triggering a reorg. func (m *Manager) AddBlocks(blocks []types.Block) error { - m.mu.Lock() - defer m.mu.Unlock() + sp, unlock := m.scratchpad() + defer unlock() if len(blocks) == 0 { return nil } log := m.log.Named("AddBlocks") - cs := m.tipState + cs := sp.TipState() for _, b := range blocks { bid := b.ID() var ok bool - if _, bs, _ := m.store.Block(bid); bs != nil { + if _, bs, _ := sp.Block(bid); bs != nil { // already have this block - cs, _ = m.store.State(bid) + cs, _ = sp.State(bid) continue } else if b.ParentID != cs.Index.ID { - if cs, ok = m.store.State(b.ParentID); !ok { + if cs, ok = sp.State(b.ParentID); !ok { return fmt.Errorf("missing parent state for block %v", bid) } } @@ -269,28 +313,28 @@ func (m *Manager) AddBlocks(blocks []types.Block) error { } else if err := consensus.ValidateOrphan(cs, b); err != nil { return fmt.Errorf("block %v is invalid: %w", types.ChainIndex{Height: cs.Index.Height + 1, ID: bid}, err) } - ancestorTimestamp, ok := m.store.AncestorTimestamp(b.ParentID) + ancestorTimestamp, ok := sp.AncestorTimestamp(b.ParentID) if !ok { return fmt.Errorf("missing ancestor timestamp for block %v", b.ParentID) } cs = consensus.ApplyHeader(cs, b.Header(), ancestorTimestamp) - m.store.AddState(cs) - m.store.AddBlock(b, nil) + sp.AddState(cs) + sp.AddBlock(b, nil) log.Debug("added block", zap.Uint64("height", cs.Index.Height), zap.Stringer("id", bid)) } // if this chain is now the best chain, trigger a reorg - if cs.SufficientlyHeavierThan(m.tipState) { - oldTip := m.tipState.Index + if cs.SufficientlyHeavierThan(sp.TipState()) { + oldTip := sp.TipState().Index log.Debug("reorging to", zap.Stringer("current", oldTip), zap.Stringer("target", cs.Index)) - if err := m.reorgTo(cs.Index); err != nil { - if err := m.reorgTo(oldTip); err != nil { + if err := m.reorgTo(sp, cs.Index); err != nil { + if err := m.reorgTo(sp, oldTip); err != nil { return fmt.Errorf("failed to revert failed reorg: %w", err) } return fmt.Errorf("reorg failed: %w", err) } - // release lock while notifying listeners - tip := m.tipState.Index + // notify listeners, without holding the write lock + tip := sp.TipState().Index fns := make([]func(), 0, len(m.onReorg)+len(m.onPool)) for _, fn := range m.onReorg { fns = append(fns, func() { fn(tip) }) @@ -298,11 +342,10 @@ func (m *Manager) AddBlocks(blocks []types.Block) error { for _, fn := range m.onPool { fns = append(fns, fn) } - m.mu.Unlock() + unlock() for _, fn := range fns { fn() } - m.mu.Lock() } return nil } @@ -311,37 +354,37 @@ func (m *Manager) AddBlocks(blocks []types.Block) error { // validated, and the first block's parent must be known. If the chain has // sufficient work, it may become the new best chain, triggering a reorg. func (m *Manager) AddValidatedV2Blocks(blocks []types.Block, states []consensus.State) error { - m.mu.Lock() - defer m.mu.Unlock() + sp, unlock := m.scratchpad() + defer unlock() if len(blocks) == 0 { return nil } else if len(states) != len(blocks) { return errors.New("chain: expected same number of blocks and states") } - if _, ok := m.store.State(blocks[0].ParentID); !ok { + if _, ok := sp.State(blocks[0].ParentID); !ok { return fmt.Errorf("missing parent for block %v", blocks[0].ParentID) } for i := range blocks { if blocks[i].V2 == nil { return errors.New("only v2 blocks can be pre-validated") } - m.store.AddBlock(blocks[i], &consensus.V1BlockSupplement{}) - m.store.AddState(states[i]) + sp.AddBlock(blocks[i], &consensus.V1BlockSupplement{}) + sp.AddState(states[i]) } // if this chain is now the best chain, trigger a reorg cs := states[len(states)-1] - if cs.SufficientlyHeavierThan(m.tipState) { - oldTip := m.tipState.Index + if cs.SufficientlyHeavierThan(sp.TipState()) { + oldTip := sp.TipState().Index m.log.Debug("reorging", zap.Stringer("current", oldTip), zap.Stringer("target", cs.Index)) - if err := m.reorgTo(cs.Index); err != nil { - if err := m.reorgTo(oldTip); err != nil { + if err := m.reorgTo(sp, cs.Index); err != nil { + if err := m.reorgTo(sp, oldTip); err != nil { return fmt.Errorf("failed to revert failed reorg: %w", err) } return fmt.Errorf("reorg failed: %w", err) } // release lock while notifying listeners - tip := m.tipState.Index + tip := sp.TipState().Index fns := make([]func(), 0, len(m.onReorg)+len(m.onPool)) for _, fn := range m.onReorg { fns = append(fns, func() { fn(tip) }) @@ -349,11 +392,10 @@ func (m *Manager) AddValidatedV2Blocks(blocks []types.Block, states []consensus. for _, fn := range m.onPool { fns = append(fns, fn) } - m.mu.Unlock() + unlock() for _, fn := range fns { fn() } - m.mu.Lock() } return nil } @@ -382,65 +424,65 @@ func (m *Manager) overwriteExpirations(b types.Block, bs *consensus.V1BlockSuppl } // revertTip reverts the current tip. -func (m *Manager) revertTip() error { - b, bs, cs, ok := blockAndParent(m.store, m.tipState.Index.ID) +func (m *Manager) revertTip(sp StoreScratchpad) error { + tip := sp.TipState() + b, bs, cs, ok := blockAndParent(sp, tip.Index.ID) if !ok { - return fmt.Errorf("%w %v", ErrMissingBlock, m.tipState.Index) + return fmt.Errorf("%w %v", ErrMissingBlock, tip.Index) } cru := consensus.RevertBlock(cs, b, *bs) - m.store.RevertBlock(cs, cru) + sp.RevertBlock(cs, cru) m.revertPoolUpdate(cru, cs) - m.tipState = cs return nil } // applyTip adds a block to the current tip. -func (m *Manager) applyTip(index types.ChainIndex) error { +func (m *Manager) applyTip(sp StoreScratchpad, index types.ChainIndex) error { + tip := sp.TipState() var cs consensus.State var cau consensus.ApplyUpdate - b, bs, ok := m.store.Block(index.ID) + b, bs, ok := sp.Block(index.ID) if !ok { return fmt.Errorf("%w %v", ErrMissingBlock, index) - } else if b.ParentID != m.tipState.Index.ID { + } else if b.ParentID != tip.Index.ID { panic("applyTip called with non-attaching block") } else if bs == nil { bs = new(consensus.V1BlockSupplement) - *bs = m.store.SupplementTipBlock(b) + *bs = sp.SupplementTipBlock(b) if err := m.overwriteExpirations(b, bs); err != nil { return fmt.Errorf("failed to overwrite expiring file contract order in block %v: %w", index, err) - } else if err := consensus.ValidateBlock(m.tipState, b, *bs); err != nil { + } else if err := consensus.ValidateBlock(tip, b, *bs); err != nil { return fmt.Errorf("block %v is invalid: %w", index, err) } - ancestorTimestamp, ok := m.store.AncestorTimestamp(b.ParentID) + ancestorTimestamp, ok := sp.AncestorTimestamp(b.ParentID) if !ok { return fmt.Errorf("missing ancestor timestamp for block %v", b.ParentID) } - cs, cau = consensus.ApplyBlock(m.tipState, b, *bs, ancestorTimestamp) - m.store.AddState(cs) - m.store.AddBlock(b, bs) + cs, cau = consensus.ApplyBlock(tip, b, *bs, ancestorTimestamp) + sp.AddState(cs) + sp.AddBlock(b, bs) } else { - ancestorTimestamp, ok := m.store.AncestorTimestamp(b.ParentID) + ancestorTimestamp, ok := sp.AncestorTimestamp(b.ParentID) if !ok { return fmt.Errorf("missing ancestor timestamp for block %v", b.ParentID) } else if err := m.overwriteExpirations(b, bs); err != nil { return fmt.Errorf("failed to overwrite expiring file contract order in block %v: %w", index, err) } - cs, cau = consensus.ApplyBlock(m.tipState, b, *bs, ancestorTimestamp) + cs, cau = consensus.ApplyBlock(tip, b, *bs, ancestorTimestamp) } - m.store.ApplyBlock(cs, cau) + sp.ApplyBlock(cs, cau) m.applyPoolUpdate(cau, cs) - m.tipState = cs return nil } -func (m *Manager) reorgPath(a, b types.ChainIndex, maxLen int) (revert, apply []types.ChainIndex, err error) { +func (m *Manager) reorgPath(ss StoreSnapshot, a, b types.ChainIndex, maxLen int) (revert, apply []types.ChainIndex, err error) { // helper function for "rewinding" to the parent index rewind := func(index *types.ChainIndex) bool { if len(revert)+len(apply) > maxLen { err = fmt.Errorf("reorg path is too long (-%d +%d, max %d)", len(revert), len(apply), maxLen) return false } - bh, ok := m.store.Header(index.ID) + bh, ok := ss.Header(index.ID) if !ok { err = fmt.Errorf("%w %v", ErrMissingBlock, *index) } else { @@ -465,7 +507,7 @@ func (m *Manager) reorgPath(a, b types.ChainIndex, maxLen int) (revert, apply [] // special case: if a is uninitialized, we're starting from genesis if a == (types.ChainIndex{}) { - a, _ = m.store.BestIndex(0) + a, _ = ss.BestIndex(0) apply = append(apply, a) } @@ -482,22 +524,22 @@ func (m *Manager) reorgPath(a, b types.ChainIndex, maxLen int) (revert, apply [] return } -func (m *Manager) reorgTo(index types.ChainIndex) error { - revert, apply, err := m.reorgPath(m.tipState.Index, index, math.MaxInt) +func (m *Manager) reorgTo(sp StoreScratchpad, index types.ChainIndex) error { + revert, apply, err := m.reorgPath(sp, sp.TipState().Index, index, math.MaxInt) if err != nil { return err } for range revert { - if err := m.revertTip(); err != nil { - return fmt.Errorf("couldn't revert block %v: %w", m.tipState.Index, err) + if err := m.revertTip(sp); err != nil { + return fmt.Errorf("couldn't revert block %v: %w", sp.TipState().Index, err) } } for _, index := range apply { - if err := m.applyTip(index); err != nil { + if err := m.applyTip(sp, index); err != nil { return fmt.Errorf("couldn't apply block %v: %w", index, err) } } - if err := m.store.Flush(); err != nil { + if err := sp.Flush(); err != nil { return err } @@ -505,7 +547,7 @@ func (m *Manager) reorgTo(index types.ChainIndex) error { m.txpool.ms = nil m.txpool.medianFee = nil if len(revert) > 0 { - b, _, _ := m.store.Block(revert[0].ID) + b, _, _ := sp.Block(revert[0].ID) m.txpool.lastReverted = m.txpool.lastReverted[:0] // prevent coinbase transactions from spamming the pool for _, txn := range b.Transactions { @@ -525,45 +567,47 @@ func (m *Manager) reorgTo(index types.ChainIndex) error { return nil } -// PruneBlocks prunes any blocks below the specified height -// from the store. This should only be called after all -// subscribers have processed blocks up to the specified height. +// PruneBlocks prunes any blocks below the specified height from the store. This +// should only be called after all subscribers have processed blocks up to the +// specified height. // -// Once the blocks are removed, they cannot be re-added without -// resyncing from genesis. +// Once the blocks are removed, they cannot be re-added without resyncing from +// genesis. // -// This can take a while depending on the number of blocks -// it is recommended to call this frequently to avoid -// a large backlog. +// This can take a while depending on the number of blocks it is recommended to +// call this frequently to avoid a large backlog. +// +// Pruned blocks are not guaranteed to be removed immediately, only eventually. func (m *Manager) PruneBlocks(height uint64) { - m.mu.Lock() - defer m.mu.Unlock() + sp, unlock := m.scratchpad() + defer unlock() for h := height; h > 0; h-- { - index, ok := m.store.BestIndex(h - 1) + index, ok := sp.BestIndex(h - 1) if !ok { break // block does not exist - } else if _, _, ok := m.store.Block(index.ID); !ok { + } else if _, _, ok := sp.Block(index.ID); !ok { break // block does not exist } - m.store.PruneBlock(index.ID) + sp.PruneBlock(index.ID) } } // UpdatesSince returns at most max updates on the path between index and the // Manager's current tip. func (m *Manager) UpdatesSince(index types.ChainIndex, maxBlocks int) (rus []RevertUpdate, aus []ApplyUpdate, err error) { - m.mu.Lock() - defer m.mu.Unlock() + ss, release := m.snapshot() + defer release() + tip := ss.TipState() onBestChain := func(index types.ChainIndex) bool { - bi, _ := m.store.BestIndex(index.Height) + bi, _ := ss.BestIndex(index.Height) return bi.ID == index.ID || index == types.ChainIndex{} } - for index != m.tipState.Index && len(rus)+len(aus) < maxBlocks { + for index != tip.Index && len(rus)+len(aus) < maxBlocks { // revert until we are on the best chain, then apply if !onBestChain(index) { - b, bs, cs, ok := blockAndParent(m.store, index.ID) + b, bs, cs, ok := blockAndParent(ss, index.ID) if !ok { return nil, nil, fmt.Errorf("%w %v", ErrMissingBlock, index) } else if bs == nil { @@ -575,17 +619,17 @@ func (m *Manager) UpdatesSince(index types.ChainIndex, maxBlocks int) (rus []Rev } else { // special case: if index is uninitialized, we're starting from genesis if index == (types.ChainIndex{}) { - index, _ = m.store.BestIndex(0) + index, _ = ss.BestIndex(0) } else { - index, _ = m.store.BestIndex(index.Height + 1) + index, _ = ss.BestIndex(index.Height + 1) } - b, bs, cs, ok := blockAndParent(m.store, index.ID) + b, bs, cs, ok := blockAndParent(ss, index.ID) if !ok { return nil, nil, fmt.Errorf("%w %v", ErrMissingBlock, index) } else if bs == nil { return nil, nil, fmt.Errorf("missing supplement for block %v", index) } - ancestorTimestamp, ok := m.store.AncestorTimestamp(b.ParentID) + ancestorTimestamp, ok := ss.AncestorTimestamp(b.ParentID) if !ok && index.Height != 0 { return nil, nil, fmt.Errorf("missing ancestor timestamp for block %v", b.ParentID) } @@ -600,13 +644,13 @@ func (m *Manager) UpdatesSince(index types.ChainIndex, maxBlocks int) (rus []Rev // chain changes. The fn is called with the new tip. It returns a function that // removes fn from the set. func (m *Manager) OnReorg(fn func(types.ChainIndex)) (cancel func()) { - m.mu.Lock() - defer m.mu.Unlock() + _, unlock := m.scratchpad() + defer unlock() key := frand.Entropy128() m.onReorg[key] = fn return func() { - m.mu.Lock() - defer m.mu.Unlock() + _, unlock := m.scratchpad() + defer unlock() delete(m.onReorg, key) } } @@ -615,20 +659,21 @@ func (m *Manager) OnReorg(fn func(types.ChainIndex)) (cancel func()) { // transaction pool may have changed. It returns a function that removes fn from // the set. func (m *Manager) OnPoolChange(fn func()) (cancel func()) { - m.mu.Lock() - defer m.mu.Unlock() + _, unlock := m.scratchpad() + defer unlock() key := frand.Entropy128() m.onPool[key] = fn return func() { - m.mu.Lock() - defer m.mu.Unlock() + _, unlock := m.scratchpad() + defer unlock() delete(m.onPool, key) } } -func (m *Manager) revalidatePool() { +func (m *Manager) revalidatePool(ss StoreSnapshot) { log := m.log.Named("revalidatePool") - txpoolMaxWeight := m.tipState.MaxBlockWeight() * 10 + tip := ss.TipState() + txpoolMaxWeight := tip.MaxBlockWeight() * 10 if m.txpool.ms != nil && m.txpool.weight < txpoolMaxWeight { return } @@ -650,14 +695,14 @@ func (m *Manager) revalidatePool() { txnFees = append(txnFees, feeTxn{ index: i, fees: txn.TotalFees(), - weight: m.tipState.TransactionWeight(txn), + weight: tip.TransactionWeight(txn), }) } for i, txn := range m.txpool.v2txns { txnFees = append(txnFees, feeTxn{ index: i, fees: txn.MinerFee, - weight: m.tipState.V2TransactionWeight(txn), + weight: tip.V2TransactionWeight(txn), v2: true, }) } @@ -688,7 +733,7 @@ func (m *Manager) revalidatePool() { for txid := range m.txpool.indices { delete(m.txpool.indices, txid) } - m.txpool.ms = consensus.NewMidState(m.tipState) + m.txpool.ms = consensus.NewMidState(tip) m.txpool.txns = append(m.txpool.txns, m.txpool.lastReverted...) m.txpool.weight = 0 filtered := m.txpool.txns[:0] @@ -698,14 +743,14 @@ func (m *Manager) revalidatePool() { // already in the pool continue } - ts := m.store.SupplementTipTransaction(txn) + ts := ss.SupplementTipTransaction(txn) if err := consensus.ValidateTransaction(m.txpool.ms, txn, ts); err != nil { log.Debug("dropping invalid pool transaction", zap.Stringer("id", txn.ID()), zap.Error(err)) continue } m.txpool.ms.ApplyTransaction(txn, ts) m.txpool.indices[id] = len(filtered) - m.txpool.weight += m.tipState.TransactionWeight(txn) + m.txpool.weight += tip.TransactionWeight(txn) filtered = append(filtered, txn) } m.txpool.txns = filtered @@ -723,13 +768,13 @@ func (m *Manager) revalidatePool() { } m.txpool.ms.ApplyV2Transaction(txn) m.txpool.indices[id] = len(v2filtered) - m.txpool.weight += m.tipState.V2TransactionWeight(txn) + m.txpool.weight += tip.V2TransactionWeight(txn) v2filtered = append(v2filtered, txn) } m.txpool.v2txns = v2filtered } -func (m *Manager) computeMedianFee() types.Currency { +func (m *Manager) computeMedianFee(ss StoreSnapshot) types.Currency { if m.txpool.medianFee != nil { return *m.txpool.medianFee } @@ -765,8 +810,8 @@ func (m *Manager) computeMedianFee() types.Currency { } prevFees := make([]types.Currency, 0, 10) for i := uint64(0); i < 10; i++ { - index, ok1 := m.store.BestIndex(m.tipState.Index.Height - i) - b, _, cs, ok2 := blockAndParent(m.store, index.ID) + index, ok1 := ss.BestIndex(ss.TipState().Index.Height - i) + b, _, cs, ok2 := blockAndParent(ss, index.ID) if ok1 && ok2 { prevFees = append(prevFees, calculateBlockMedianFee(cs, b)) } @@ -1009,9 +1054,9 @@ func (m *Manager) applyPoolUpdate(cau consensus.ApplyUpdate, cs consensus.State) // PoolTransaction returns the transaction with the specified ID, if it is // currently in the pool. func (m *Manager) PoolTransaction(id types.TransactionID) (types.Transaction, bool) { - m.mu.Lock() - defer m.mu.Unlock() - m.revalidatePool() + sp, unlock := m.scratchpad() + defer unlock() + m.revalidatePool(sp) i, ok := m.txpool.indices[id] if !ok { return types.Transaction{}, false @@ -1022,18 +1067,18 @@ func (m *Manager) PoolTransaction(id types.TransactionID) (types.Transaction, bo // PoolTransactions returns the transactions currently in the txpool. Any prefix // of the returned slice constitutes a valid transaction set. func (m *Manager) PoolTransactions() []types.Transaction { - m.mu.Lock() - defer m.mu.Unlock() - m.revalidatePool() + sp, unlock := m.scratchpad() + defer unlock() + m.revalidatePool(sp) return slices.Clone(m.txpool.txns) } // V2PoolTransaction returns the v2 transaction with the specified ID, if it is // currently in the pool. func (m *Manager) V2PoolTransaction(id types.TransactionID) (types.V2Transaction, bool) { - m.mu.Lock() - defer m.mu.Unlock() - m.revalidatePool() + sp, unlock := m.scratchpad() + defer unlock() + m.revalidatePool(sp) i, ok := m.txpool.indices[id] if !ok { return types.V2Transaction{}, false @@ -1044,9 +1089,9 @@ func (m *Manager) V2PoolTransaction(id types.TransactionID) (types.V2Transaction // V2PoolTransactions returns the v2 transactions currently in the txpool. Any // prefix of the returned slice constitutes a valid transaction set. func (m *Manager) V2PoolTransactions() []types.V2Transaction { - m.mu.Lock() - defer m.mu.Unlock() - m.revalidatePool() + sp, unlock := m.scratchpad() + defer unlock() + m.revalidatePool(sp) v2txns := make([]types.V2Transaction, len(m.txpool.v2txns)) for i, txn := range m.txpool.v2txns { v2txns[i] = txn.DeepCopy() @@ -1057,9 +1102,9 @@ func (m *Manager) V2PoolTransactions() []types.V2Transaction { // TransactionsForPartialBlock returns the transactions in the txpool with the // specified hashes. func (m *Manager) TransactionsForPartialBlock(missing []types.Hash256) (txns []types.Transaction, v2txns []types.V2Transaction) { - m.mu.Lock() - defer m.mu.Unlock() - m.revalidatePool() + sp, unlock := m.scratchpad() + defer unlock() + m.revalidatePool(sp) want := make(map[types.Hash256]bool) for _, h := range missing { want[h] = true @@ -1086,11 +1131,11 @@ func (m *Manager) TransactionsForPartialBlock(missing []types.Hash256) (txns []t // RecommendedFee returns the recommended fee (per weight unit) to ensure a high // probability of inclusion in the next block. func (m *Manager) RecommendedFee() types.Currency { - m.mu.Lock() - defer m.mu.Unlock() - m.revalidatePool() + sp, unlock := m.scratchpad() + defer unlock() + m.revalidatePool(sp) - medianFee := m.computeMedianFee() + medianFee := m.computeMedianFee(sp) // calculate a fee relative to the total txpool weight // @@ -1123,9 +1168,9 @@ func (m *Manager) RecommendedFee() types.Currency { // UnconfirmedParents returns the transactions in the txpool that are referenced // by txn. func (m *Manager) UnconfirmedParents(txn types.Transaction) []types.Transaction { - m.mu.Lock() - defer m.mu.Unlock() - m.revalidatePool() + sp, unlock := m.scratchpad() + defer unlock() + m.revalidatePool(sp) parentMap := m.computeParentMap() var parents []types.Transaction @@ -1175,9 +1220,9 @@ func (m *Manager) UnconfirmedParents(txn types.Transaction) []types.Transaction // tip, the transaction will be updated. The transaction set includes the parents // and the transaction itself in an order valid for broadcasting. func (m *Manager) V2TransactionSet(basis types.ChainIndex, txn types.V2Transaction) (types.ChainIndex, []types.V2Transaction, error) { - m.mu.Lock() - defer m.mu.Unlock() - m.revalidatePool() + sp, unlock := m.scratchpad() + defer unlock() + m.revalidatePool(sp) // get the transaction's parents parentMap := m.computeParentMap() @@ -1222,26 +1267,27 @@ func (m *Manager) V2TransactionSet(basis types.ChainIndex, txn types.V2Transacti } // update the transaction's basis to match tip - txns, err := m.updateV2TransactionProofs(append(parents, txn), basis, m.tipState.Index) + tip := sp.TipState().Index + txns, err := m.updateV2TransactionProofs(sp, append(parents, txn), basis, tip) if err != nil { return types.ChainIndex{}, nil, fmt.Errorf("failed to update transaction set basis: %w", err) } - return m.tipState.Index, txns, nil + return tip, txns, nil } -func (m *Manager) checkTxnSet(txns []types.Transaction, v2txns []types.V2Transaction) (bool, error) { +func (m *Manager) checkTxnSet(sp StoreScratchpad, txns []types.Transaction, v2txns []types.V2Transaction) (bool, error) { if err := checkEphemeralOutputs(v2txns); err != nil { return false, err } else if err := checkFileContractRevisions(v2txns); err != nil { return false, err } allInPool := true - ms := consensus.NewMidState(m.tipState) + ms := consensus.NewMidState(sp.TipState()) for _, txn := range txns { if _, ok := m.txpool.indices[txn.ID()]; !ok { allInPool = false } - ts := m.store.SupplementTipTransaction(txn) + ts := sp.SupplementTipTransaction(txn) if err := consensus.ValidateTransaction(ms, txn, ts); err != nil { return false, fmt.Errorf("transaction %v is invalid: %w", txn.ID(), err) } @@ -1264,10 +1310,10 @@ func (m *Manager) checkTxnSet(txns []types.Transaction, v2txns []types.V2Transac return allInPool, nil } -func (m *Manager) updateV2TransactionProofs(txns []types.V2Transaction, from, to types.ChainIndex) (updated []types.V2Transaction, err error) { +func (m *Manager) updateV2TransactionProofs(ss StoreSnapshot, txns []types.V2Transaction, from, to types.ChainIndex) (updated []types.V2Transaction, err error) { // first validate the transaction set against its claimed basis; attempting // to update an invalid proof can cause a panic - basisState, ok := m.store.State(from.ID) + basisState, ok := ss.State(from.ID) if !ok { return nil, fmt.Errorf("couldn't find state for basis %v", from) } @@ -1277,7 +1323,7 @@ func (m *Manager) updateV2TransactionProofs(txns []types.V2Transaction, from, to } } - revert, apply, err := m.reorgPath(from, to, 144) + revert, apply, err := m.reorgPath(ss, from, to, 144) if err != nil { return nil, fmt.Errorf("couldn't determine reorg path from %v to %v: %w", from, to, err) } @@ -1288,7 +1334,7 @@ func (m *Manager) updateV2TransactionProofs(txns []types.V2Transaction, from, to updated = append(updated, txn.DeepCopy()) } for _, index := range revert { - b, bs, cs, ok := blockAndParent(m.store, index.ID) + b, bs, cs, ok := blockAndParent(ss, index.ID) if !ok { return nil, fmt.Errorf("missing reverted block at index %v", index) } else if bs == nil { @@ -1305,7 +1351,7 @@ func (m *Manager) updateV2TransactionProofs(txns []types.V2Transaction, from, to } for _, index := range apply { - b, bs, cs, ok := blockAndParent(m.store, index.ID) + b, bs, cs, ok := blockAndParent(ss, index.ID) if !ok { return nil, fmt.Errorf("missing applied block at index %v", index) } else if bs == nil { @@ -1313,7 +1359,7 @@ func (m *Manager) updateV2TransactionProofs(txns []types.V2Transaction, from, to } else if err := m.overwriteExpirations(b, bs); err != nil { return nil, fmt.Errorf("failed to overwrite expirations for block %v: %w", index, err) } - ancestorTimestamp, _ := m.store.AncestorTimestamp(b.ParentID) + ancestorTimestamp, _ := ss.AncestorTimestamp(b.ParentID) cs, cau := consensus.ApplyBlock(cs, b, *bs, ancestorTimestamp) // get the transactions that were confirmed in this block @@ -1383,20 +1429,21 @@ func (m *Manager) updateV2TransactionProofs(txns []types.V2Transaction, from, to // of the transactions are added to the pool. If all of the transactions are // already known to the pool, AddPoolTransactions returns true. func (m *Manager) AddPoolTransactions(txns []types.Transaction) (known bool, err error) { - m.mu.Lock() - defer m.mu.Unlock() - m.revalidatePool() + sp, unlock := m.scratchpad() + defer unlock() + m.revalidatePool(sp) - if known, err := m.checkTxnSet(txns, nil); known || err != nil { + if known, err := m.checkTxnSet(sp, txns, nil); known || err != nil { return known, err } + tip := sp.TipState() for _, txn := range txns { txid := txn.ID() if _, ok := m.txpool.indices[txid]; ok { continue // skip transactions already in the pool } - ts := m.store.SupplementTipTransaction(txn) + ts := sp.SupplementTipTransaction(txn) if err := consensus.ValidateTransaction(m.txpool.ms, txn, ts); err != nil { m.txpool.ms = nil // force revalidation next time the pool is queried return false, fmt.Errorf("transaction %v conflicts with pool: %w", txid, err) @@ -1404,7 +1451,7 @@ func (m *Manager) AddPoolTransactions(txns []types.Transaction) (known bool, err m.txpool.ms.ApplyTransaction(txn, ts) m.txpool.indices[txid] = len(m.txpool.txns) m.txpool.txns = append(m.txpool.txns, txn) - m.txpool.weight += m.tipState.TransactionWeight(txn) + m.txpool.weight += tip.TransactionWeight(txn) } // invalidate caches m.txpool.medianFee = nil @@ -1414,12 +1461,10 @@ func (m *Manager) AddPoolTransactions(txns []types.Transaction) (known bool, err for _, fn := range m.onPool { fns = append(fns, fn) } - m.mu.Unlock() + unlock() for _, fn := range fns { fn() } - m.mu.Lock() - return false, nil } @@ -1434,9 +1479,9 @@ func (m *Manager) UpdateV2TransactionSet(txns []types.V2Transaction, from, to ty if from == to { return txns, nil } - m.mu.Lock() - defer m.mu.Unlock() - return m.updateV2TransactionProofs(txns, from, to) + ss, unlock := m.snapshot() + defer unlock() + return m.updateV2TransactionProofs(ss, txns, from, to) } // AddV2PoolTransactions validates a transaction set and adds it to the txpool. @@ -1454,23 +1499,24 @@ func (m *Manager) UpdateV2TransactionSet(txns []types.V2Transaction, from, to ty // proofs will be updated accordingly. The original transactions are not // modified and none of their memory is retained. func (m *Manager) AddV2PoolTransactions(basis types.ChainIndex, txns []types.V2Transaction) (known bool, _ error) { - m.mu.Lock() - defer m.mu.Unlock() - m.revalidatePool() + sp, unlock := m.scratchpad() + defer unlock() + m.revalidatePool(sp) // take ownership of Merkle proofs, and update them to the current tip txns = slices.Clone(txns) for i := range txns { txns[i] = txns[i].DeepCopy() } - txns, err := m.updateV2TransactionProofs(txns, basis, m.tipState.Index) + txns, err := m.updateV2TransactionProofs(sp, txns, basis, sp.TipState().Index) if err != nil { return false, fmt.Errorf("failed to update set basis: %w", err) } - if known, err := m.checkTxnSet(nil, txns); known || err != nil { + if known, err := m.checkTxnSet(sp, nil, txns); known || err != nil { return known, err } + tip := sp.TipState() for _, txn := range txns { txid := txn.ID() if _, ok := m.txpool.indices[txid]; ok { @@ -1483,33 +1529,31 @@ func (m *Manager) AddV2PoolTransactions(basis types.ChainIndex, txns []types.V2T m.txpool.ms.ApplyV2Transaction(txn) m.txpool.indices[txid] = len(m.txpool.v2txns) m.txpool.v2txns = append(m.txpool.v2txns, txn) - m.txpool.weight += m.tipState.V2TransactionWeight(txn) + m.txpool.weight += tip.V2TransactionWeight(txn) } // invalidate caches m.txpool.medianFee = nil - // release lock while notifying listeners + // notify listeners, without holding the write lock fns := make([]func(), 0, len(m.onPool)) for _, fn := range m.onPool { fns = append(fns, fn) } - m.mu.Unlock() + unlock() for _, fn := range fns { fn() } - m.mu.Lock() - return false, nil } -// NewManager returns a Manager initialized with the provided Store and State. -func NewManager(store Store, cs consensus.State, opts ...ManagerOption) *Manager { +// NewManager returns a Manager initialized with the provided Store. The Store +// must not have any unflushed writes. +func NewManager(store Store, opts ...ManagerOption) *Manager { m := &Manager{ - log: zap.NewNop(), - store: store, - tipState: cs, - onReorg: make(map[[16]byte]func(types.ChainIndex)), - onPool: make(map[[16]byte]func()), + log: zap.NewNop(), + store: store, + onReorg: make(map[[16]byte]func(types.ChainIndex)), + onPool: make(map[[16]byte]func()), expiringFileContractOrder: defaultExpiringFileContractOrder, } diff --git a/chain/manager_test.go b/chain/manager_test.go index 03cd5114..408f2ce9 100644 --- a/chain/manager_test.go +++ b/chain/manager_test.go @@ -4,6 +4,7 @@ import ( "math" "reflect" "strings" + "sync" "testing" "time" @@ -16,10 +17,13 @@ import ( // the tip is valid or not. This is useful for testing purposes, where we want to // simulate a reorganization of the blockchain. func (m *Manager) ForceRevertTip() error { - m.mu.Lock() - defer m.mu.Unlock() + sp, unlock := m.scratchpad() + defer unlock() - return m.revertTip() + if err := m.revertTip(sp); err != nil { + return err + } + return sp.Flush() } func findBlockNonce(cs consensus.State, b *types.Block) { @@ -73,11 +77,13 @@ func TestRevertedNoElementTransaction(t *testing.T) { n.HardforkFoundation.Height = 0 n.HardforkV2.AllowHeight = 0 - store, tipState, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) + store, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := NewManager(store, tipState) + cm := NewManager(store) + snap, release := store.Snapshot() + defer release() txn := types.V2Transaction{ArbitraryData: frand.Bytes(16)} @@ -103,8 +109,8 @@ func TestRevertedNoElementTransaction(t *testing.T) { } } findBlockNonce(cs, &b) - ancestorTimestamp, _ := store.AncestorTimestamp(b.ParentID) - cs, _ = consensus.ApplyBlock(cs, b, store.SupplementTipBlock(b), ancestorTimestamp) + ancestorTimestamp, _ := snap.AncestorTimestamp(b.ParentID) + cs, _ = consensus.ApplyBlock(cs, b, snap.SupplementTipBlock(b), ancestorTimestamp) return b, cs } @@ -138,12 +144,13 @@ func TestManager(t *testing.T) { n.InitialTarget = types.BlockID{0xFF} - store, tipState, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) + store, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := NewManager(store, tipState) + cm := NewManager(store) + sp := store.Scratchpad() mine := func(cs consensus.State, n int) (blocks []types.Block) { for i := 0; i < n; i++ { b := types.Block{ @@ -155,8 +162,8 @@ func TestManager(t *testing.T) { }}, } findBlockNonce(cs, &b) - ancestorTimestamp, _ := store.AncestorTimestamp(b.ParentID) - cs, _ = consensus.ApplyBlock(cs, b, store.SupplementTipBlock(b), ancestorTimestamp) + ancestorTimestamp, _ := sp.AncestorTimestamp(b.ParentID) + cs, _ = consensus.ApplyBlock(cs, b, sp.SupplementTipBlock(b), ancestorTimestamp) blocks = append(blocks, b) } return @@ -232,11 +239,11 @@ func TestTxPool(t *testing.T) { } genesisBlock.Transactions = []types.Transaction{giftTxn} - store, tipState, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) + store, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := NewManager(store, tipState) + cm := NewManager(store) // add a listener var changeSets [][]types.TransactionID @@ -380,11 +387,12 @@ func TestUpdateV2TransactionSet(t *testing.T) { } // initialize chain manager and mine a mix of v1 and v2 blocks - store, genesisState, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) + store, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := NewManager(store, genesisState) + cm := NewManager(store) + genesisState := cm.TipState() for range 10 { cs := cm.TipState() b := types.Block{ @@ -444,11 +452,11 @@ func TestFullTxPool(t *testing.T) { } genesisBlock.Transactions = []types.Transaction{giftTxn} - store, tipState, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) + store, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := NewManager(store, tipState) + cm := NewManager(store) signTxn := func(txn *types.Transaction) { for _, sci := range txn.SiacoinInputs { @@ -528,44 +536,44 @@ func TestNewDBStoreAtCheckpoint(t *testing.T) { t.Run("DBThenCheckpoint", func(t *testing.T) { db := NewMemDB() - _, _, _ = NewDBStore(db, n, genesisBlock, nil) + _, _ = NewDBStore(db, n, genesisBlock, nil) db.Flush() - _, tipState, err := NewDBStoreAtCheckpoint(db, checkpointParent, checkpointBlock, nil) + store, err := NewDBStoreAtCheckpoint(db, checkpointParent, checkpointBlock, nil) if err != nil { t.Fatal(err) - } else if tipState.Index != checkpointState.Index { + } else if store.Scratchpad().TipState().Index != checkpointState.Index { t.Fatal("DB should be initialized at checkpoint") } }) t.Run("CheckpointThenDB", func(t *testing.T) { db := NewMemDB() - _, _, err := NewDBStoreAtCheckpoint(db, checkpointParent, checkpointBlock, nil) + _, err := NewDBStoreAtCheckpoint(db, checkpointParent, checkpointBlock, nil) if err != nil { t.Fatal(err) } db.Flush() - _, tipState, err := NewDBStore(db, n, genesisBlock, nil) + store, err := NewDBStore(db, n, genesisBlock, nil) if err != nil { t.Fatal(err) } - if tipState.Index != checkpointState.Index { + if store.Scratchpad().TipState().Index != checkpointState.Index { t.Fatal("DB should remain at checkpoint") } }) t.Run("CheckpointTwice", func(t *testing.T) { db := NewMemDB() - _, _, err := NewDBStoreAtCheckpoint(db, checkpointParent, checkpointBlock, nil) + _, err := NewDBStoreAtCheckpoint(db, checkpointParent, checkpointBlock, nil) if err != nil { t.Fatal(err) } db.Flush() - _, tipState, err := NewDBStoreAtCheckpoint(db, checkpointParent, checkpointBlock, nil) + store, err := NewDBStoreAtCheckpoint(db, checkpointParent, checkpointBlock, nil) if err != nil { t.Fatal(err) } - if tipState.Index != checkpointState.Index { + if store.Scratchpad().TipState().Index != checkpointState.Index { t.Fatal("DB should remain at checkpoint") } }) @@ -573,11 +581,11 @@ func TestNewDBStoreAtCheckpoint(t *testing.T) { t.Run("DifferentNetwork", func(t *testing.T) { mainnet, mainnetGenesis := Mainnet() db := NewMemDB() - _, _, err := NewDBStore(db, mainnet, mainnetGenesis, nil) + _, err := NewDBStore(db, mainnet, mainnetGenesis, nil) if err != nil { t.Fatal(err) } - _, _, err = NewDBStoreAtCheckpoint(db, checkpointParent, checkpointBlock, nil) + _, err = NewDBStoreAtCheckpoint(db, checkpointParent, checkpointBlock, nil) if err == nil { t.Fatal("expected error when initializing with different network") } @@ -620,22 +628,22 @@ func TestMinReorgIndex(t *testing.T) { checkpointState, _ := consensus.ApplyBlock(cs, b, consensus.V1BlockSupplement{}, time.Time{}) // initialize manager at checkpoint - store, tipState, err := NewDBStoreAtCheckpoint(NewMemDB(), cs, b, nil) + store, err := NewDBStoreAtCheckpoint(NewMemDB(), cs, b, nil) if err != nil { t.Fatal(err) } - cm := NewManager(store, tipState) + cm := NewManager(store) // min reorg index should be at checkpoint if minReorg := cm.MinReorgIndex(); minReorg != checkpointState.Index { t.Fatal("unexpected min reorg index:", minReorg, "expected:", checkpointState.Index) } // reinitialize at genesis - store, tipState, err = NewDBStore(NewMemDB(), n, genesisBlock, nil) + store, err = NewDBStore(NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm = NewManager(store, tipState) + cm = NewManager(store) if minReorg := cm.MinReorgIndex(); minReorg.ID != genesisBlock.ID() { t.Fatal("unexpected min reorg index:", minReorg, "expected:", genesisBlock.ID()) } @@ -691,11 +699,11 @@ func TestReorgPathMaxLen(t *testing.T) { n.HardforkV2.AllowHeight = 1 n.HardforkV2.RequireHeight = 1 n.HardforkV2.FinalCutHeight = 1 - store, tipState, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) + store, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := NewManager(store, tipState) + cm := NewManager(store) const chainLen = 200 tip := mineEmptyBlocks(t, cm, chainLen) @@ -716,7 +724,9 @@ func TestReorgPathMaxLen(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - _, _, err := cm.reorgPath(genesisIdx, tip, tc.maxLen) + sp, unlock := cm.scratchpad() + defer unlock() + _, _, err := cm.reorgPath(sp, genesisIdx, tip, tc.maxLen) if tc.shouldErr { if err == nil { t.Fatalf("expected error, got nil") @@ -747,11 +757,11 @@ func TestReorgPathBogusBasisBailsFast(t *testing.T) { n.HardforkV2.AllowHeight = 1 n.HardforkV2.RequireHeight = 1 n.HardforkV2.FinalCutHeight = 1 - store, tipState, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) + store, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := NewManager(store, tipState) + cm := NewManager(store) // mine enough real blocks that the in-loop cap fires before the rewind // reaches genesis (and would otherwise error with ErrMissingBlock). @@ -761,7 +771,9 @@ func TestReorgPathBogusBasisBailsFast(t *testing.T) { bogus := types.ChainIndex{Height: math.MaxUint64, ID: tip.ID} const maxLen = 144 - revert, apply, err := cm.reorgPath(bogus, tip, maxLen) + sp, unlock := cm.scratchpad() + defer unlock() + revert, apply, err := cm.reorgPath(sp, bogus, tip, maxLen) if err == nil { t.Fatal("expected error for bogus basis, got nil") } @@ -775,3 +787,170 @@ func TestReorgPathBogusBasisBailsFast(t *testing.T) { t.Fatalf("revert+apply grew to %d, expected ≤ %d (maxLen+2)", pathLen, maxLen+2) } } + +func TestReadsDoNotBlockOnWriters(t *testing.T) { + n, genesisBlock := TestnetZen() + genesisBlock.Transactions = nil + store, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) + if err != nil { + t.Fatal(err) + } + cm := NewManager(store) + tipState := cm.TipState() + + // simulate a long-running writer by holding the write lock; readers + // should still be able to observe the snapshot + _, unlock := cm.scratchpad() + done := make(chan struct{}) + go func() { + defer close(done) + if cm.TipState().Index != tipState.Index { + t.Error("unexpected tip state") + } + if _, ok := cm.Block(genesisBlock.ID()); !ok { + t.Error("missing genesis block") + } + if _, ok := cm.BestIndex(0); !ok { + t.Error("missing genesis index") + } + if _, _, err := cm.UpdatesSince(types.ChainIndex{}, 10); err != nil { + t.Error(err) + } + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("reads blocked while writer held the write lock") + } + unlock() +} + +// TestManagerConcurrentReads exercises the Manager's read methods from many +// goroutines while blocks are concurrently added, verifying (particularly +// under the race detector) that readers observe consistent snapshots without +// blocking on writers. +func TestManagerConcurrentReads(t *testing.T) { + n, genesisBlock := TestnetZen() + genesisBlock.Transactions = nil + n.InitialTarget = types.BlockID{0xFF} + n.BlockInterval = time.Second + n.MaturityDelay = 5 + n.HardforkDevAddr.Height = 1 + n.HardforkTax.Height = 1 + n.HardforkStorageProof.Height = 1 + n.HardforkOak.Height = 1 + n.HardforkASIC.Height = 1 + n.HardforkFoundation.Height = 1 + n.HardforkV2.AllowHeight = 1 + n.HardforkV2.RequireHeight = 1 + n.HardforkV2.FinalCutHeight = 1 + store, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) + if err != nil { + t.Fatal(err) + } + cm := NewManager(store) + + var wg sync.WaitGroup + stop := make(chan struct{}) + for range 8 { + wg.Go(func() { + for { + select { + case <-stop: + return + default: + } + // each of these should observe a consistent snapshot; since + // the chain is extended linearly, blocks and best indices + // visible in one call must remain visible in the next + tip := cm.Tip() + if cs := cm.TipState(); cs.Index.Height < tip.Height { + t.Error("tip height regressed:", cs.Index, tip) + } + if index, ok := cm.BestIndex(tip.Height); !ok || index != tip { + t.Error("tip not on best chain:", tip, index) + } + if b, ok := cm.Block(tip.ID); !ok { + t.Error("missing tip block:", tip) + } else if b.ID() != tip.ID { + t.Error("block ID mismatch:", b.ID(), tip.ID) + } + if _, ok := cm.State(tip.ID); !ok { + t.Error("missing tip state:", tip) + } + if _, err := cm.History(); err != nil { + t.Error(err) + } + if _, _, err := cm.UpdatesSince(tip, 10); err != nil { + t.Error(err) + } + if _, _, err := cm.Headers(tip, 10); err != nil { + t.Error(err) + } + if _, _, err := cm.BlocksForHistory([]types.BlockID{tip.ID}, 10); err != nil { + t.Error(err) + } + cm.MinReorgIndex() + cm.PoolTransactions() + cm.RecommendedFee() + } + }) + } + + mineEmptyBlocks(t, cm, 25) + close(stop) + wg.Wait() + + if cm.Tip().Height != 25 { + t.Fatal("expected tip height 25, got", cm.Tip().Height) + } +} + +// TestSnapshotDoesNotBlockWrites verifies that a long-lived snapshot neither +// stalls writers nor observes their effects. +func TestSnapshotDoesNotBlockWrites(t *testing.T) { + n, genesisBlock := TestnetZen() + genesisBlock.Transactions = nil + n.InitialTarget = types.BlockID{0xFF} + n.BlockInterval = time.Second + n.MaturityDelay = 5 + n.HardforkDevAddr.Height = 1 + n.HardforkTax.Height = 1 + n.HardforkStorageProof.Height = 1 + n.HardforkOak.Height = 1 + n.HardforkASIC.Height = 1 + n.HardforkFoundation.Height = 1 + n.HardforkV2.AllowHeight = 1 + n.HardforkV2.RequireHeight = 1 + n.HardforkV2.FinalCutHeight = 1 + store, err := NewDBStore(NewMemDB(), n, genesisBlock, nil) + if err != nil { + t.Fatal(err) + } + cm := NewManager(store) + tipState := cm.TipState() + + // hold a snapshot across several AddBlocks calls; if snapshots excluded + // writers, this would deadlock + snap, release := store.Snapshot() + defer release() + mineEmptyBlocks(t, cm, 5) + + // the Manager should have advanced... + if cm.Tip().Height != tipState.Index.Height+5 { + t.Fatal("expected tip to advance to height 5, got", cm.Tip().Height) + } + // ...while the held snapshot still reflects the old tip + if snap.TipState().Index != tipState.Index { + t.Fatal("snapshot tip changed:", snap.TipState().Index) + } else if _, ok := snap.BestIndex(tipState.Index.Height + 1); ok { + t.Fatal("snapshot should not see new blocks") + } + + // a snapshot taken now should see the new tip + snap2, release2 := store.Snapshot() + defer release2() + if snap2.TipState().Index != cm.Tip() { + t.Fatal("new snapshot tip mismatch:", snap2.TipState().Index, cm.Tip()) + } +} diff --git a/chain/migrate.go b/chain/migrate.go index 3a385793..6bb9b4a8 100644 --- a/chain/migrate.go +++ b/chain/migrate.go @@ -49,60 +49,63 @@ func NewZapMigrationLogger(log *zap.Logger) MigrationLogger { } func migrateDB(dbs *DBStore, l MigrationLogger) error { - version := dbs.bucket(bVersion).getRaw(bVersion) + scratch := dbs.scratchpad + sp := scratch.sp + version := readBucket(sp, bVersion).getRaw(bVersion) switch version[0] { case 1, 2, 3: l.Printf("Removing sidechain blocks") toDelete := make(map[types.BlockID]bool) - for id := range dbs.db.Bucket(bBlocks).Iter() { + for id := range sp.Bucket(bBlocks).Iter() { toDelete[(types.BlockID)(id)] = true } - for _, id := range dbs.db.Bucket(bMainChain).Iter() { + for _, id := range sp.Bucket(bMainChain).Iter() { if len(id) == 32 { delete(toDelete, (types.BlockID)(id)) } } for id := range toDelete { - dbs.bucket(bBlocks).delete(id[:]) - dbs.bucket(bStates).delete(id[:]) + scratch.bucket(bBlocks).delete(id[:]) + scratch.bucket(bStates).delete(id[:]) } l.Printf("Removing block supplement data") - for id := range dbs.db.Bucket(bFileContractElements).Iter() { - dbs.bucket(bFileContractElements).delete(id) + for id := range sp.Bucket(bFileContractElements).Iter() { + scratch.bucket(bFileContractElements).delete(id) } - for id := range dbs.db.Bucket(bSiacoinElements).Iter() { - dbs.bucket(bSiacoinElements).delete(id) + for id := range sp.Bucket(bSiacoinElements).Iter() { + scratch.bucket(bSiacoinElements).delete(id) } - for id := range dbs.db.Bucket(bSiafundElements).Iter() { - dbs.bucket(bSiafundElements).delete(id) + for id := range sp.Bucket(bSiafundElements).Iter() { + scratch.bucket(bSiafundElements).delete(id) } - if dbs.shouldFlush() { - if err := dbs.Flush(); err != nil { + if scratch.shouldFlush() { + if err := scratch.Flush(); err != nil { return err } } l.Printf("Recomputing main chain") - v1Blocks := min(dbs.getHeight(), dbs.n.HardforkV2.RequireHeight) + 1 - cs := dbs.n.GenesisState() + n := dbs.n + v1Blocks := min(getHeight(sp), n.HardforkV2.RequireHeight) + 1 + cs := n.GenesisState() for height := range v1Blocks { - index, _ := dbs.BestIndex(height) - _, b, _, _ := dbs.getBlock(index.ID) + index, _ := bestIndex(sp, height) + _, b, _, _ := getBlock(sp, index.ID) if b == nil { return errors.New("missing block needed for migration") } - bs := dbs.SupplementTipBlock(*b) - dbs.putBlock(b.Header(), b, &bs) + bs := supplementTipBlock(sp, n, *b) + scratch.AddBlock(*b, &bs) // v2 blocks may be invalid - if height >= dbs.n.HardforkV2.AllowHeight { + if height >= n.HardforkV2.AllowHeight { if err := consensus.ValidateBlock(cs, *b, bs); err != nil && index.Height > 0 { l.Printf("Block %v is invalid (%v), removing it and all subsequent blocks", index, err) for ; height < v1Blocks; height++ { - if index, ok := dbs.BestIndex(height); ok { - dbs.bucket(bBlocks).delete(index.ID[:]) - dbs.bucket(bStates).delete(index.ID[:]) - if dbs.shouldFlush() { - if err := dbs.Flush(); err != nil { + if index, ok := bestIndex(sp, height); ok { + scratch.bucket(bBlocks).delete(index.ID[:]) + scratch.bucket(bStates).delete(index.ID[:]) + if scratch.shouldFlush() { + if err := scratch.Flush(); err != nil { return err } } @@ -112,22 +115,17 @@ func migrateDB(dbs *DBStore, l MigrationLogger) error { } } var cau consensus.ApplyUpdate - ancestorTimestamp, _ := dbs.AncestorTimestamp(b.ParentID) + ancestorTimestamp, _ := ancestorTimestamp(sp, n, b.ParentID) cs, cau = consensus.ApplyBlock(cs, *b, bs, ancestorTimestamp) - dbs.putState(cs) - dbs.ApplyBlock(cs, cau) - if dbs.shouldFlush() { - if err := dbs.Flush(); err != nil { - return err - } - } + scratch.AddState(cs) + scratch.ApplyBlock(cs, cau) // flushes as necessary l.SetProgress(99.9 * float64(height) / float64(v1Blocks)) } - if err := dbs.Flush(); err != nil { + if err := scratch.Flush(); err != nil { return err } - dbs.bucket(bVersion).putRaw(bVersion, []byte{4}) - if err := dbs.Flush(); err != nil { + scratch.bucket(bVersion).putRaw(bVersion, []byte{4}) + if err := scratch.Flush(); err != nil { return err } l.SetProgress(100) diff --git a/chain/pool_test.go b/chain/pool_test.go index f0317349..13500d51 100644 --- a/chain/pool_test.go +++ b/chain/pool_test.go @@ -17,11 +17,11 @@ func TestAddV2PoolTransactionsRecover(t *testing.T) { sp := types.PolicyPublicKey(sk.PublicKey()) addr := sp.Address() - store, genesisState, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) + store, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := chain.NewManager(store, genesisState) + cm := chain.NewManager(store) es := testutil.NewElementStateStore(t, cm) testutil.MineBlocks(t, cm, addr, 20+int(n.MaturityDelay)) @@ -118,11 +118,11 @@ func TestAddV2PoolTransactionsEphemeralValue(t *testing.T) { sp := types.PolicyPublicKey(sk.PublicKey()) addr := sp.Address() - store, genesisState, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) + store, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := chain.NewManager(store, genesisState) + cm := chain.NewManager(store) es := testutil.NewElementStateStore(t, cm) testutil.MineBlocks(t, cm, addr, 5+int(n.MaturityDelay)) diff --git a/db.go b/db.go index ff0eb359..899a78cb 100644 --- a/db.go +++ b/db.go @@ -24,69 +24,111 @@ func (b boltBucket) Iter() iter.Seq2[[]byte, []byte] { // BoltChainDB implements chain.DB with a BoltDB database. type BoltChainDB struct { - tx *bbolt.Tx + db *bbolt.DB + scratchpad *boltDBScratchpad +} + +// boltDBScratchpad implements chain.DBScratchpad. It lazily opens a bbolt +// write transaction, which accumulates writes until Flush commits it. +type boltDBScratchpad struct { db *bbolt.DB + tx *bbolt.Tx } -func (db *BoltChainDB) newTx() (err error) { - if db.tx == nil { - db.tx, err = db.db.Begin(true) +func (s *boltDBScratchpad) newTx() (err error) { + if s.tx == nil { + s.tx, err = s.db.Begin(true) } return } -// Bucket implements chain.DB. -func (db *BoltChainDB) Bucket(name []byte) chain.DBBucket { - if err := db.newTx(); err != nil { +// Bucket implements chain.DBScratchpad. +func (s *boltDBScratchpad) Bucket(name []byte) chain.DBBucket { + if err := s.newTx(); err != nil { panic(err) } - b := db.tx.Bucket(name) + b := s.tx.Bucket(name) if b == nil { return nil } return boltBucket{b} } -// CreateBucket implements chain.DB. -func (db *BoltChainDB) CreateBucket(name []byte) (chain.DBBucket, error) { - if err := db.newTx(); err != nil { +// CreateBucket implements chain.DBScratchpad. +func (s *boltDBScratchpad) CreateBucket(name []byte) (chain.DBBucket, error) { + if err := s.newTx(); err != nil { return nil, err } - b, err := db.tx.CreateBucket(name) + b, err := s.tx.CreateBucket(name) if err != nil { return nil, err } return boltBucket{b}, nil } -// Flush implements chain.DB. -func (db *BoltChainDB) Flush() error { - if db.tx == nil { +// Flush implements chain.DBScratchpad. +func (s *boltDBScratchpad) Flush() error { + if s.tx == nil { return nil } - err := db.tx.Commit() - db.tx = nil + err := s.tx.Commit() + s.tx = nil return err } -// Cancel implements chain.DB. -func (db *BoltChainDB) Cancel() { - if db.tx == nil { +// Cancel implements chain.DBScratchpad. +func (s *boltDBScratchpad) Cancel() { + if s.tx == nil { return } - db.tx.Rollback() - db.tx = nil + s.tx.Rollback() + s.tx = nil +} + +// Scratchpad implements chain.DB. +func (db *BoltChainDB) Scratchpad() chain.DBScratchpad { return db.scratchpad } + +type boltDBSnapshot struct { + tx *bbolt.Tx } -// Close closes the BoltDB database. +// Bucket implements chain.DBSnapshot. +func (v boltDBSnapshot) Bucket(name []byte) chain.DBBucket { + b := v.tx.Bucket(name) + if b == nil { + return nil + } + return boltBucket{b} +} + +func (v boltDBSnapshot) release() { + v.tx.Rollback() +} + +// Snapshot implements chain.DB. +func (db *BoltChainDB) Snapshot() (chain.DBSnapshot, func()) { + tx, err := db.db.Begin(false) + if err != nil { + panic(err) + } + s := boltDBSnapshot{tx} + return s, s.release +} + +// Close flushes any pending writes and closes the BoltDB database. func (db *BoltChainDB) Close() error { - db.Flush() + if err := db.scratchpad.Flush(); err != nil { + return err + } return db.db.Close() } // NewBoltChainDB creates a new BoltChainDB. func NewBoltChainDB(db *bbolt.DB) *BoltChainDB { - return &BoltChainDB{db: db} + return &BoltChainDB{ + db: db, + scratchpad: &boltDBScratchpad{db: db}, + } } // OpenBoltChainDB opens a BoltDB database. diff --git a/internal/cmd/calcswaps/main.go b/internal/cmd/calcswaps/main.go index 9531c15f..874a5ac4 100644 --- a/internal/cmd/calcswaps/main.go +++ b/internal/cmd/calcswaps/main.go @@ -86,42 +86,46 @@ func main() { } defer tipDB.Close() - tipStore, tipState, err := chain.NewDBStore(tipDB, n, genesis, nil) + tipStore, err := chain.NewDBStore(tipDB, n, genesis, nil) if err != nil { log.Panic("failed to create tip store", zap.Error(err)) } + tipSS, release := tipStore.Snapshot() + defer release() + tipState := tipSS.TipState() log.Info("starting expiring file contract order calculation", zap.String("network", network), zap.Stringer("index", tipState.Index)) - cleanStore, cs, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) + cleanStore, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) if err != nil { log.Panic("failed to create clean db", zap.Error(err)) } - cm := chain.NewManager(cleanStore, cs, chain.WithLog(log.Named("chain"))) + cleanSP := cleanStore.Scratchpad() + cm := chain.NewManager(cleanStore, chain.WithLog(log.Named("chain"))) tip := min(tipState.Index.Height, maxCheckHeight) overwriteIDs := make(map[types.BlockID][]types.FileContractID) - for height := range tip - 144 { + for height := range max(tip, 144) - 144 { select { case <-ctx.Done(): return default: } - index, ok := tipStore.BestIndex(height) + index, ok := tipSS.BestIndex(height) if !ok { log.Panic("failed to get best index", zap.Uint64("height", height)) } log := log.With(zap.Stringer("index", index)) - b, bs, ok := tipStore.Block(index.ID) + b, bs, ok := tipSS.Block(index.ID) if !ok { log.Panic("failed to get block") } else if bs == nil { log.Panic("block state is nil") } - cs, ok := tipStore.State(index.ID) + cs, ok := tipSS.State(index.ID) if !ok { log.Panic("failed to get state for block") } @@ -131,7 +135,7 @@ func main() { order = append(order, fc.ID) } - cleanOrder := cleanStore.ExpiringFileContractIDs(height) + cleanOrder := cleanSP.ExpiringFileContractIDs(height) seen := make(map[types.FileContractID]bool) if !slices.Equal(order, cleanOrder) { // ensure that all expiring file contracts in the clean db are also @@ -152,8 +156,8 @@ func main() { zap.Stringers("expected", order), zap.Stringers("actual", cleanOrder)) - cleanStore.OverwriteExpiringFileContractIDs(height, order) - if err := cleanStore.Flush(); err != nil { + cleanSP.OverwriteExpiringFileContractIDs(height, order) + if err := cleanSP.Flush(); err != nil { log.Panic("failed to flush clean db", zap.Error(err)) } overwriteIDs[index.ID] = order diff --git a/internal/cmd/sync/main.go b/internal/cmd/sync/main.go index 5a6dd2ed..2d0d4b79 100644 --- a/internal/cmd/sync/main.go +++ b/internal/cmd/sync/main.go @@ -150,24 +150,23 @@ func main() { } var store *chain.DBStore - var tipState consensus.State if checkpoint != (types.ChainIndex{}) { log.Info("retrieving checkpoint", zap.Stringer("index", checkpoint)) cs, b, err := syncer.RetrieveCheckpoint(ctx, bootstrapPeers, checkpoint, n, genesis.ID()) if err != nil { log.Panic("failed to retrieve checkpoint", zap.Error(err)) } - store, tipState, err = chain.NewDBStoreAtCheckpoint(db, cs, b, chain.NewZapMigrationLogger(log.Named("migrate"))) + store, err = chain.NewDBStoreAtCheckpoint(db, cs, b, chain.NewZapMigrationLogger(log.Named("migrate"))) if err != nil { log.Panic("failed to create store", zap.Error(err)) } } else { - store, tipState, err = chain.NewDBStore(db, n, genesis, chain.NewZapMigrationLogger(log.Named("migrate"))) + store, err = chain.NewDBStore(db, n, genesis, chain.NewZapMigrationLogger(log.Named("migrate"))) if err != nil { log.Panic("failed to create store", zap.Error(err)) } } - cm := chain.NewManager(store, tipState, chain.WithLog(log.Named("chain"))) + cm := chain.NewManager(store, chain.WithLog(log.Named("chain"))) log = log.With(zap.Stringer("start", cm.Tip())) l, err := net.Listen("tcp", ":0") diff --git a/miner_test.go b/miner_test.go index 53309033..c9671eac 100644 --- a/miner_test.go +++ b/miner_test.go @@ -23,11 +23,11 @@ func TestMiner(t *testing.T) { }, }} - store, tipState, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) + store, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := chain.NewManager(store, tipState) + cm := chain.NewManager(store) // create a transaction txn := types.Transaction{ @@ -87,11 +87,11 @@ func TestV2MineBlocks(t *testing.T) { }, }} - store, tipState, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) + store, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := chain.NewManager(store, tipState) + cm := chain.NewManager(store) mineBlocks := func(t *testing.T, n int) { for ; n > 0; n-- { diff --git a/rhp/v4/rpc_test.go b/rhp/v4/rpc_test.go index 0f2d27e6..ca0ac1ba 100644 --- a/rhp/v4/rpc_test.go +++ b/rhp/v4/rpc_test.go @@ -166,11 +166,11 @@ func testRenterHostPairWebTransport(tb testing.TB, hostKey types.PrivateKey, cm } func startTestNode(tb testing.TB, n *consensus.Network, genesis types.Block) (*chain.Manager, *wallet.SingleAddressWallet) { - db, tipstate, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) + db, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) if err != nil { tb.Fatal(err) } - cm := chain.NewManager(db, tipstate) + cm := chain.NewManager(db) syncerListener, err := net.Listen("tcp", ":0") if err != nil { diff --git a/syncer/syncer_test.go b/syncer/syncer_test.go index a0acc931..af52d605 100644 --- a/syncer/syncer_test.go +++ b/syncer/syncer_test.go @@ -74,11 +74,11 @@ func mineBlocks(t *testing.T, s *syncer.Syncer, cm *chain.Manager, n int) { func newTestSyncer(t testing.TB, opts ...syncer.Option) (*syncer.Syncer, *chain.Manager) { n, genesis := testutil.Network() - store, tipState1, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) + store, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) if err != nil { t.Fatal(err) } - cm := chain.NewManager(store, tipState1) + cm := chain.NewManager(store) l, err := net.Listen("tcp", ":0") if err != nil { @@ -265,11 +265,11 @@ func TestInstantSync(t *testing.T) { t.Fatalf("expected checkpoint state %v, got %v", b.ParentID, cs.Index.ID) } // initialize new manager at synced checkpoint - store, newTipState, err := chain.NewDBStoreAtCheckpoint(chain.NewMemDB(), cs, b, nil) + store, err := chain.NewDBStoreAtCheckpoint(chain.NewMemDB(), cs, b, nil) if err != nil { t.Fatal(err) } - cm2 := chain.NewManager(store, newTipState) + cm2 := chain.NewManager(store) if cm2.Tip() != index { t.Fatalf("expected tip %v, got %v", index, cm2.Tip()) @@ -442,11 +442,11 @@ func TestForkPeerSynced(t *testing.T) { // s2 has a shorter fork chain (10 blocks) with a very long sync // interval so it never adopts s1's chain during the test n, genesis := testutil.Network() - store2, tipState2, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) + store2, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) if err != nil { t.Fatal(err) } - cm2 := chain.NewManager(store2, tipState2) + cm2 := chain.NewManager(store2) testutil.MineBlocks(t, cm2, types.VoidAddress, 10) l2, err := net.Listen("tcp", ":0") @@ -516,11 +516,11 @@ func TestParallelSyncStall(t *testing.T) { // s2 has blocks but BlocksForHistory always fails, so blocks // can never be served despite valid headers n, genesis := testutil.Network() - store2, tipState2, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) + store2, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) if err != nil { t.Fatal(err) } - cm2 := chain.NewManager(store2, tipState2) + cm2 := chain.NewManager(store2) testutil.MineBlocks(t, cm2, types.VoidAddress, 10) l2, err := net.Listen("tcp", ":0") @@ -590,11 +590,11 @@ func TestShareNodesMalformed(t *testing.T) { ps1.AddPeer(addr) } ps1.AddPeer("127.0.0.1:65535") - store1, tipState1, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) + store1, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) if err != nil { t.Fatal(err) } - cm1 := chain.NewManager(store1, tipState1) + cm1 := chain.NewManager(store1) l1, err := net.Listen("tcp", ":0") if err != nil { @@ -612,11 +612,11 @@ func TestShareNodesMalformed(t *testing.T) { // s2 connects to s1 and runs peer discovery ps2 := testutil.NewEphemeralPeerStore() - store2, tipState2, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) + store2, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) if err != nil { t.Fatal(err) } - cm2 := chain.NewManager(store2, tipState2) + cm2 := chain.NewManager(store2) l2, err := net.Listen("tcp", ":0") if err != nil { @@ -686,12 +686,12 @@ func (b *blockingManager) unblock() { b.once.Do(func() { close(b.release) }) } func newBlockingSyncer(t *testing.T, log *zap.Logger, opts ...syncer.Option) (*syncer.Syncer, *blockingManager) { t.Helper() n, genesis := testutil.Network() - store, ts, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) + store, err := chain.NewDBStore(chain.NewMemDB(), n, genesis, nil) if err != nil { t.Fatal(err) } bm := &blockingManager{ - Manager: chain.NewManager(store, ts), + Manager: chain.NewManager(store), entered: make(chan struct{}, 64), release: make(chan struct{}), } diff --git a/wallet/wallet_test.go b/wallet/wallet_test.go index 6a8814fd..9510a056 100644 --- a/wallet/wallet_test.go +++ b/wallet/wallet_test.go @@ -226,13 +226,14 @@ func TestWallet(t *testing.T) { // create chain store network, genesis := testutil.Network() - cs, genesisState, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) + cs, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) if err != nil { t.Fatal(err) } // create chain manager and subscribe the wallet - cm := chain.NewManager(cs, genesisState) + cm := chain.NewManager(cs) + genesisState := cm.TipState() // create wallet l := zaptest.NewLogger(t) @@ -384,13 +385,14 @@ func TestWalletLockUnlock(t *testing.T) { // create chain store network, genesis := testutil.Network() - cs, genesisState, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) + cs, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) if err != nil { t.Fatal(err) } // create chain manager and subscribe the wallet - cm := chain.NewManager(cs, genesisState) + cm := chain.NewManager(cs) + genesisState := cm.TipState() // create wallet l := zaptest.NewLogger(t) w, err := wallet.NewSingleAddressWallet(pk, cm, ws, &testutil.MockSyncer{}, wallet.WithLogger(l.Named("wallet"))) @@ -443,13 +445,13 @@ func TestWalletUnconfirmed(t *testing.T) { // create chain store network, genesis := testutil.Network() - cs, tipState, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) + cs, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) if err != nil { t.Fatal(err) } // create chain manager and subscribe the wallet - cm := chain.NewManager(cs, tipState) + cm := chain.NewManager(cs) // create wallet l := zaptest.NewLogger(t) w, err := wallet.NewSingleAddressWallet(pk, cm, ws, &testutil.MockSyncer{}, wallet.WithLogger(l.Named("wallet"))) @@ -546,13 +548,14 @@ func TestWalletRedistribute(t *testing.T) { // create chain store network, genesis := testutil.Network() network.HardforkV2.AllowHeight = 1 // allow V2 transactions from the start - cs, tipState, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) + cs, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) if err != nil { t.Fatal(err) } // create chain manager and subscribe the wallet - cm := chain.NewManager(cs, tipState) + cm := chain.NewManager(cs) + tipState := cm.TipState() // create wallet l := zaptest.NewLogger(t) w, err := wallet.NewSingleAddressWallet(pk, cm, ws, &testutil.MockSyncer{}, wallet.WithLogger(l.Named("wallet"))) @@ -651,13 +654,14 @@ func TestReorg(t *testing.T) { // create chain store network, genesis := testutil.Network() - cs, genesisState, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) + cs, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) if err != nil { t.Fatal(err) } // create chain manager and subscribe the wallet - cm := chain.NewManager(cs, genesisState) + cm := chain.NewManager(cs) + genesisState := cm.TipState() // create wallet l := zaptest.NewLogger(t) w, err := wallet.NewSingleAddressWallet(pk, cm, ws, &testutil.MockSyncer{}, wallet.WithLogger(l.Named("wallet"))) @@ -865,13 +869,14 @@ func TestWalletV2(t *testing.T) { // create chain store network, genesis := testutil.Network() - cs, genesisState, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) + cs, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) if err != nil { t.Fatal(err) } // create chain manager and subscribe the wallet - cm := chain.NewManager(cs, genesisState) + cm := chain.NewManager(cs) + genesisState := cm.TipState() // create wallet l := zaptest.NewLogger(t) w, err := wallet.NewSingleAddressWallet(pk, cm, ws, &testutil.MockSyncer{}, wallet.WithLogger(l.Named("wallet"))) @@ -1042,13 +1047,14 @@ func TestReorgV2(t *testing.T) { // create chain store network, genesis := testutil.V2Network() - cs, genesisState, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) + cs, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) if err != nil { t.Fatal(err) } // create chain manager and subscribe the wallet - cm := chain.NewManager(cs, genesisState) + cm := chain.NewManager(cs) + genesisState := cm.TipState() // create wallet l := zaptest.NewLogger(t) @@ -1225,8 +1231,9 @@ func TestReorgV2(t *testing.T) { if !coreutils.FindBlockNonce(state, &b, time.Second) { t.Fatal("failed to find nonce") } - ancestorTimestamp, _ := cs.AncestorTimestamp(b.ParentID) - state, _ = consensus.ApplyBlock(state, b, cs.SupplementTipBlock(b), ancestorTimestamp) + csSP := cs.Scratchpad() + ancestorTimestamp, _ := csSP.AncestorTimestamp(b.ParentID) + state, _ = consensus.ApplyBlock(state, b, csSP.SupplementTipBlock(b), ancestorTimestamp) reorgBlocks := []types.Block{b} for i := 0; i < 5; i++ { b := types.Block{ @@ -1241,8 +1248,8 @@ func TestReorgV2(t *testing.T) { if !coreutils.FindBlockNonce(state, &b, time.Second) { t.Fatal("failed to find nonce") } - ancestorTimestamp, _ := cs.AncestorTimestamp(b.ParentID) - state, _ = consensus.ApplyBlock(state, b, cs.SupplementTipBlock(b), ancestorTimestamp) + ancestorTimestamp, _ := csSP.AncestorTimestamp(b.ParentID) + state, _ = consensus.ApplyBlock(state, b, csSP.SupplementTipBlock(b), ancestorTimestamp) reorgBlocks = append(reorgBlocks, b) } @@ -1293,13 +1300,13 @@ func TestFundTransaction(t *testing.T) { network.HardforkV2.RequireHeight = 3 // create chain store - cs, tipState, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) + cs, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) if err != nil { t.Fatal(err) } // create chain manager and subscribe the wallet - cm := chain.NewManager(cs, tipState) + cm := chain.NewManager(cs) // create wallet l := zaptest.NewLogger(t) w, err := wallet.NewSingleAddressWallet(pk, cm, ws, &testutil.MockSyncer{}, wallet.WithLogger(l.Named("wallet"))) @@ -1394,11 +1401,12 @@ func TestSingleAddressWalletEventTypes(t *testing.T) { // raise the require height to test v1 events network.HardforkV2.RequireHeight = 100 network.HardforkV2.FinalCutHeight = 200 - store, genesisState, err := chain.NewDBStore(bdb, network, genesisBlock, nil) + store, err := chain.NewDBStore(bdb, network, genesisBlock, nil) if err != nil { t.Fatal(err) } - cm := chain.NewManager(store, genesisState) + cm := chain.NewManager(store) + genesisState := cm.TipState() ws := testutil.NewEphemeralWalletStore() wm, err := wallet.NewSingleAddressWallet(pk, cm, ws, &testutil.MockSyncer{}, wallet.WithLogger(log)) @@ -1781,13 +1789,13 @@ func TestV2TxPoolRace(t *testing.T) { // create chain store network, genesis := testutil.V2Network() - cs, genesisState, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) + cs, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) if err != nil { t.Fatal(err) } // create chain manager and subscribe the wallet - cm := chain.NewManager(cs, genesisState) + cm := chain.NewManager(cs) // create wallet l := zaptest.NewLogger(t) w, err := wallet.NewSingleAddressWallet(pk, cm, ws, &testutil.MockSyncer{}, wallet.WithLogger(l.Named("wallet"))) @@ -1918,13 +1926,13 @@ func TestRebroadcastTransaction(t *testing.T) { ws := testutil.NewEphemeralWalletStore() // create chain store - dbs, genesisState, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) + dbs, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) if err != nil { t.Fatal(err) } // create chain manager and subscribe the wallet - cm := chain.NewManager(dbs, genesisState) + cm := chain.NewManager(dbs) // create wallet l := zaptest.NewLogger(t) @@ -2073,13 +2081,13 @@ func TestReloadBroadcastedSets(t *testing.T) { ws := testutil.NewEphemeralWalletStore() // create chain store - dbs, genesisState, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) + dbs, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) if err != nil { t.Fatal(err) } // create chain manager and subscribe the wallet - cm := chain.NewManager(dbs, genesisState) + cm := chain.NewManager(dbs) // create wallet l := zaptest.NewLogger(t) @@ -2144,13 +2152,13 @@ func TestSplitUTXO(t *testing.T) { ws := testutil.NewEphemeralWalletStore() // create chain store - dbs, genesisState, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) + dbs, err := chain.NewDBStore(chain.NewMemDB(), network, genesis, nil) if err != nil { t.Fatal(err) } // create chain manager and subscribe the wallet - cm := chain.NewManager(dbs, genesisState) + cm := chain.NewManager(dbs) // create wallet l := zaptest.NewLogger(t)