diff --git a/blockchain/blockchain.go b/blockchain/blockchain.go index 18ecfa6a44..96f655a4dc 100644 --- a/blockchain/blockchain.go +++ b/blockchain/blockchain.go @@ -3,6 +3,7 @@ package blockchain import ( "errors" "iter" + "sync/atomic" "github.com/NethermindEth/juno/blockchain/networks" "github.com/NethermindEth/juno/blockchain/statebackend" @@ -104,6 +105,14 @@ type Blockchain struct { cachedFilters *AggregatedBloomFilterCache runningFilter *core.RunningEventFilter stateBackend statebackend.StateBackend + + // chainHeight and l1Head mirror their database entries. Nil means "unknown", so a failed + // refresh sends readers to the database instead of serving a stale value. Only valid while + // db.ChainHeight and db.L1Height are written through Blockchain by one writer at a time: a + // migration on the raw db.KeyValueStore, or a concurrent store and revert, leave it stale. + chainHeight atomic.Pointer[uint64] + l1Head atomic.Pointer[core.L1Head] + cacheHeads bool } // options holds configuration for constructing a Blockchain. @@ -112,6 +121,7 @@ type options struct { stateVersion bool runningFilterInitialize core.RunningEventFilterInitializer retentionFloor *pruner.RetentionFloor + remoteDatabase bool } // Option is a functional option for configuring Blockchain options. @@ -139,6 +149,15 @@ func WithRunningEventFilterInitializer(initialize core.RunningEventFilterInitial } } +// WithRemoteDatabase marks the database as one another process writes, as `--remote-db` followers +// do. Such a node never stores or reverts, so it cannot know when the heads move and must read +// them back instead of caching them. +func WithRemoteDatabase() Option { + return func(o *options) { + o.remoteDatabase = true + } +} + // WithRetentionFloor shares a seeded retention floor (see // [pruner.NewRetentionFloor]) with the state backend, so retention checks // skip the database. The default unseeded floor probes the database instead. @@ -167,7 +186,7 @@ func New(database db.KeyValueStore, network *networks.Network, opts ...Option) * runningFilter := core.NewRunningEventFilterLazy(database, o.runningFilterInitialize) - return &Blockchain{ + chain := &Blockchain{ database: database, network: network, listener: o.listener, @@ -181,7 +200,16 @@ func New(database db.KeyValueStore, network *networks.Network, opts ...Option) * o.retentionFloor, o.stateVersion, ), + cacheHeads: !o.remoteDatabase, + } + if chain.cacheHeads { + chain.cacheChainHeight() + if l1Head, err := core.GetL1Head(database); err == nil { + chain.l1Head.Store(&l1Head) + } } + + return chain } func (b *Blockchain) Network() *networks.Network { @@ -191,12 +219,36 @@ func (b *Blockchain) Network() *networks.Network { // Height returns the latest block height. If blockchain is empty nil is returned. func (b *Blockchain) Height() (uint64, error) { b.listener.OnRead("Height") + return b.height() +} + +func (b *Blockchain) height() (uint64, error) { + if height := b.chainHeight.Load(); height != nil { + return *height, nil + } return core.GetChainHeight(b.database) } +// cacheChainHeight refreshes the cached height from the database rather than from the caller's +// block, so the cache stays derived from what was committed: reverting the genesis block removes +// the entry entirely instead of decrementing it. A read failure caches "unknown", which sends +// readers to the database. +func (b *Blockchain) cacheChainHeight() { + if !b.cacheHeads { + return + } + + height, err := core.GetChainHeight(b.database) + if err != nil { + b.chainHeight.Store(nil) + return + } + b.chainHeight.Store(&height) +} + func (b *Blockchain) Head() (*core.Block, error) { b.listener.OnRead("Head") - curHeight, err := core.GetChainHeight(b.database) + curHeight, err := b.height() if err != nil { return nil, err } @@ -206,7 +258,7 @@ func (b *Blockchain) Head() (*core.Block, error) { func (b *Blockchain) HeadsHeader() (*core.Header, error) { b.listener.OnRead("HeadsHeader") - height, err := core.GetChainHeight(b.database) + height, err := b.height() if err != nil { return nil, err } @@ -358,14 +410,47 @@ func (b *Blockchain) SubscribeL1Head() L1HeadSubscription { return L1HeadSubscription{b.l1HeadFeed.Subscribe()} } +// L1Head returns the latest L1 head. The returned felts are shared with every other caller and +// must not be mutated in place. func (b *Blockchain) L1Head() (core.L1Head, error) { b.listener.OnRead("L1Head") + if l1Head := b.l1Head.Load(); l1Head != nil { + return *l1Head, nil + } return core.GetL1Head(b.database) } func (b *Blockchain) SetL1Head(update *core.L1Head) error { + if err := core.WriteL1Head(b.database, update); err != nil { + return err + } + b.cacheL1Head(update) b.l1HeadFeed.Send(update) - return core.WriteL1Head(b.database, update) + + return nil +} + +func (b *Blockchain) cacheL1Head(update *core.L1Head) { + if !b.cacheHeads { + return + } + + if update == nil { + b.l1Head.Store(nil) + return + } + + // Deep copy: update and the felts it points at are shared with the feed's subscribers and + // outlive this call, while every later L1Head reader hands out what is cached here. Both + // felts are optional, so neither can be cloned unconditionally. + cached := core.L1Head{BlockNumber: update.BlockNumber} + if update.BlockHash != nil { + cached.BlockHash = update.BlockHash.Clone() + } + if update.StateRoot != nil { + cached.StateRoot = update.StateRoot.Clone() + } + b.l1Head.Store(&cached) } // Store takes a block and state update and performs sanity checks before putting in the database. @@ -375,6 +460,8 @@ func (b *Blockchain) Store( stateUpdate *core.StateUpdate, newClasses map[felt.Felt]core.ClassDefinition, ) error { + defer b.cacheChainHeight() + return b.stateBackend.Store(block, blockCommitments, stateUpdate, newClasses) } @@ -433,6 +520,8 @@ func (b *Blockchain) EventFilter( preConfirmedFn func() (PreConfirmedReader, error), ) (EventFilterer, error) { b.listener.OnRead("EventFilter") + // Do not use b.height() here. Events reads the height from the database on each call. Thus + // this bound and the range logic in Events use the same source. latest, err := core.GetChainHeight(b.database) if err != nil { return nil, err @@ -452,6 +541,12 @@ func (b *Blockchain) EventFilter( // RevertHead reverts the head block func (b *Blockchain) RevertHead() error { + defer b.cacheChainHeight() + + // Drop the cached height before the batch commits. A stale height outlives the block it names + // and would point readers at one that is already deleted; an unknown height sends them to the + // database, which is correct on both sides of the commit. + b.chainHeight.Store(nil) return b.stateBackend.RevertHead() } @@ -477,6 +572,8 @@ func (b *Blockchain) Finalise( newClasses map[felt.Felt]core.ClassDefinition, sign core.BlockSignFunc, ) error { + defer b.cacheChainHeight() + return b.stateBackend.Finalise(block, stateUpdate, newClasses, sign) } diff --git a/blockchain/blockchain_test.go b/blockchain/blockchain_test.go index 61cd151250..932f6bc765 100644 --- a/blockchain/blockchain_test.go +++ b/blockchain/blockchain_test.go @@ -1,7 +1,10 @@ package blockchain_test import ( + "bytes" + "errors" "fmt" + "sync/atomic" "testing" "github.com/NethermindEth/juno/blockchain" @@ -1043,8 +1046,276 @@ func TestEventsMultiPreConfirmed(t *testing.T) { }) } -func TestRevert(t *testing.T) { +type headReadCounter struct { + db.KeyValueStore + chainHeight atomic.Int64 + l1Head atomic.Int64 +} + +func (c *headReadCounter) Get(key []byte, cb func([]byte) error) error { + switch { + case bytes.Equal(key, db.ChainHeight.Key()): + c.chainHeight.Add(1) + case bytes.Equal(key, db.L1Height.Key()): + c.l1Head.Add(1) + } + return c.KeyValueStore.Get(key, cb) +} + +func TestHeightAndL1HeadAreServedWithoutReadingTheDatabase(t *testing.T) { + counter := &headReadCounter{KeyValueStore: memory.New()} + chain := blockchain.New( + counter, + &networks.Mainnet, + blockchain.WithNewState(statetestutils.UseNewState()), + ) + + client := feeder.NewTestClient(t, &networks.Mainnet) + gw := adaptfeeder.New(client) + + block, err := gw.BlockByNumber(t.Context(), 0) + require.NoError(t, err) + stateUpdate, err := gw.StateUpdate(t.Context(), 0) + require.NoError(t, err) + require.NoError(t, chain.Store(block, &emptyCommitments, stateUpdate, nil)) + + wantL1Head := core.L1Head{ + BlockNumber: block.Number, + BlockHash: block.Hash, + StateRoot: block.GlobalStateRoot, + } + require.NoError(t, chain.SetL1Head(&wantL1Head)) + + uncached := blockchain.New( + counter, + &networks.Mainnet, + blockchain.WithRemoteDatabase(), + blockchain.WithNewState(statetestutils.UseNewState()), + ) + counter.chainHeight.Store(0) + counter.l1Head.Store(0) + _, err = uncached.Height() + require.NoError(t, err) + _, err = uncached.L1Head() + require.NoError(t, err) + require.Equal(t, int64(1), counter.chainHeight.Load()) + require.Equal(t, int64(1), counter.l1Head.Load()) + + counter.chainHeight.Store(0) + counter.l1Head.Store(0) + + for range 3 { + height, err := chain.Height() + require.NoError(t, err) + assert.Equal(t, block.Number, height) + + l1Head, err := chain.L1Head() + require.NoError(t, err) + assert.Equal(t, wantL1Head, l1Head) + } + + assert.Zero(t, counter.chainHeight.Load()) + assert.Zero(t, counter.l1Head.Load()) +} + +func TestCachedL1HeadIsIsolatedFromTheCaller(t *testing.T) { + chain := blockchain.New( + memory.New(), + &networks.Mainnet, + blockchain.WithNewState(statetestutils.UseNewState()), + ) + + blockHash := new(felt.Felt).SetUint64(9) + stateRoot := new(felt.Felt).SetUint64(10) + require.NoError(t, chain.SetL1Head(&core.L1Head{ + BlockNumber: 3, + BlockHash: blockHash, + StateRoot: stateRoot, + })) + + // The L1 client keeps the felts it handed over, so mutating them must not reach the cache. + blockHash.SetUint64(0xdead) + stateRoot.SetUint64(0xbeef) + + l1Head, err := chain.L1Head() + require.NoError(t, err) + assert.Equal(t, new(felt.Felt).SetUint64(9), l1Head.BlockHash) + assert.Equal(t, new(felt.Felt).SetUint64(10), l1Head.StateRoot) +} + +func TestClearingTheL1HeadEmptiesTheCache(t *testing.T) { + chain := blockchain.New( + memory.New(), + &networks.Mainnet, + blockchain.WithNewState(statetestutils.UseNewState()), + ) + + require.NoError(t, chain.SetL1Head(&core.L1Head{ + BlockNumber: 3, + BlockHash: new(felt.Felt).SetUint64(9), + StateRoot: new(felt.Felt).SetUint64(10), + })) + require.NoError(t, chain.SetL1Head(nil)) + + // A nil update is written as an unknown head, so the cache must not keep serving the old one. + l1Head, err := chain.L1Head() + require.NoError(t, err) + assert.Equal(t, core.L1Head{}, l1Head) +} + +// failingL1HeadWriter returns an error for each write of the L1 head key. It sends all other +// writes to the wrapped store. +type failingL1HeadWriter struct { + db.KeyValueStore +} + +func (w failingL1HeadWriter) Put(key, value []byte) error { + if bytes.Equal(key, db.L1Height.Key()) { + return errors.New("write failed") + } + return w.KeyValueStore.Put(key, value) +} + +func TestFailedL1HeadWriteReachesNeitherCacheNorSubscribers(t *testing.T) { + chain := blockchain.New( + failingL1HeadWriter{KeyValueStore: memory.New()}, + &networks.Mainnet, + blockchain.WithNewState(statetestutils.UseNewState()), + ) + sub := chain.SubscribeL1Head() + t.Cleanup(sub.Unsubscribe) + + require.Error(t, chain.SetL1Head(&core.L1Head{ + BlockNumber: 3, + BlockHash: new(felt.Felt).SetUint64(9), + StateRoot: new(felt.Felt).SetUint64(10), + })) + + // The pruner deletes historical state when it receives an update from this feed. Thus + // SetL1Head must not send a head that the database did not accept. Each reader must + // continue to use the database. + select { + case got := <-sub.Recv(): + t.Fatalf("published an L1 head that was never committed: %v", got) + default: + } + _, err := chain.L1Head() + require.ErrorIs(t, err, db.ErrKeyNotFound) +} + +func TestCachedHeightAdvancesWithEachStoredBlock(t *testing.T) { + client := feeder.NewTestClient(t, &networks.Mainnet) + gw := adaptfeeder.New(client) + counter := &headReadCounter{KeyValueStore: memory.New()} + chain := blockchain.New( + counter, + &networks.Mainnet, + blockchain.WithNewState(statetestutils.UseNewState()), + ) + + for blockNumber := range uint64(2) { + block, err := gw.BlockByNumber(t.Context(), blockNumber) + require.NoError(t, err) + stateUpdate, err := gw.StateUpdate(t.Context(), blockNumber) + require.NoError(t, err) + require.NoError(t, chain.Store(block, &emptyCommitments, stateUpdate, nil)) + + counter.chainHeight.Store(0) + height, err := chain.Height() + require.NoError(t, err) + assert.Equal(t, blockNumber, height) + assert.Zero(t, counter.chainHeight.Load()) + } +} + +func TestStoreGenesisCachesTheHeight(t *testing.T) { + counter := &headReadCounter{KeyValueStore: memory.New()} + chain := blockchain.New( + counter, + &networks.Mainnet, + blockchain.WithNewState(statetestutils.UseNewState()), + ) + + genesisDiff := core.EmptyStateDiff() + require.NoError(t, chain.StoreGenesis(&genesisDiff, nil)) + + counter.chainHeight.Store(0) + height, err := chain.Height() + require.NoError(t, err) + assert.Zero(t, height) + assert.Zero(t, counter.chainHeight.Load()) +} + +func TestHeadsAreCachedWhenTheDatabaseIsAlreadyPopulated(t *testing.T) { + counter := &headReadCounter{KeyValueStore: memory.New()} + require.NoError(t, core.WriteChainHeight(counter, 7)) + wantL1Head := core.L1Head{ + BlockNumber: 3, + BlockHash: new(felt.Felt).SetUint64(9), + StateRoot: new(felt.Felt).SetUint64(10), + } + require.NoError(t, core.WriteL1Head(counter, &wantL1Head)) + + chain := blockchain.New( + counter, + &networks.Mainnet, + blockchain.WithNewState(statetestutils.UseNewState()), + ) + + // Nothing was stored through chain, so only the constructor can have filled the cache. This is + // the restarted-node case: heads are served from memory before the first block arrives. + counter.chainHeight.Store(0) + counter.l1Head.Store(0) + + height, err := chain.Height() + require.NoError(t, err) + assert.Equal(t, uint64(7), height) + + l1Head, err := chain.L1Head() + require.NoError(t, err) + assert.Equal(t, wantL1Head, l1Head) + + assert.Zero(t, counter.chainHeight.Load()) + assert.Zero(t, counter.l1Head.Load()) +} + +func TestHeadsAreReadFromTheDatabaseWhenAnotherProcessWritesIt(t *testing.T) { testDB := memory.New() + require.NoError(t, core.WriteChainHeight(testDB, 7)) + firstL1Head := core.L1Head{ + BlockNumber: 3, + BlockHash: new(felt.Felt).SetUint64(9), + StateRoot: new(felt.Felt).SetUint64(10), + } + require.NoError(t, core.WriteL1Head(testDB, &firstL1Head)) + + chain := blockchain.New( + testDB, + &networks.Mainnet, + blockchain.WithRemoteDatabase(), + blockchain.WithNewState(statetestutils.UseNewState()), + ) + + require.NoError(t, core.WriteChainHeight(testDB, 8)) + secondL1Head := core.L1Head{ + BlockNumber: 4, + BlockHash: new(felt.Felt).SetUint64(11), + StateRoot: new(felt.Felt).SetUint64(12), + } + require.NoError(t, core.WriteL1Head(testDB, &secondL1Head)) + + height, err := chain.Height() + require.NoError(t, err) + assert.Equal(t, uint64(8), height) + + l1Head, err := chain.L1Head() + require.NoError(t, err) + assert.Equal(t, secondL1Head, l1Head) +} + +func TestRevert(t *testing.T) { + counter := &headReadCounter{KeyValueStore: memory.New()} + testDB := counter chain := blockchain.New( testDB, &networks.Mainnet, @@ -1071,6 +1342,15 @@ func TestRevert(t *testing.T) { require.NoError(t, err) assert.Equal(t, uint64(1), height) }) + t.Run("height is cached after reverting onto a block that still exists", func(t *testing.T) { + // Zero reads means the revert refreshed the cache. Invalidating alone would leave it nil + // and send every later reader back to the database for the rest of the process's life. + counter.chainHeight.Store(0) + height, err := chain.Height() + require.NoError(t, err) + assert.Equal(t, uint64(1), height) + assert.Zero(t, counter.chainHeight.Load()) + }) t.Run("head should revert", func(t *testing.T) { block, err := chain.Head() require.NoError(t, err) @@ -1114,8 +1394,16 @@ func TestRevert(t *testing.T) { require.NoError(t, it.Close()) }) + t.Run("height should report an empty chain once every block is reverted", func(t *testing.T) { + _, err := chain.Height() + require.ErrorIs(t, err, db.ErrKeyNotFound) + }) + t.Run("cannot revert on empty chain", func(t *testing.T) { require.Error(t, chain.RevertHead()) + + _, err := chain.Height() + require.ErrorIs(t, err, db.ErrKeyNotFound) }) } diff --git a/node/node.go b/node/node.go index f5d1d8625a..467d9bff57 100644 --- a/node/node.go +++ b/node/node.go @@ -255,6 +255,9 @@ func New(cfg *Config, version string, logLevel *log.Level) (*Node, error) { blockchain.WithRunningEventFilterInitializer(pruner.InitializeRunningEventFilter), ) } + if dbIsRemote { + opts = append(opts, blockchain.WithRemoteDatabase()) + } chain := blockchain.New(database, &cfg.Network, opts...) // Verify that cfg.Network is compatible with the database.