Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
105 changes: 101 additions & 4 deletions blockchain/blockchain.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
"errors"
"fmt"
"iter"
"sync/atomic"

"github.com/NethermindEth/juno/blockchain/networks"
"github.com/NethermindEth/juno/blockchain/statebackend"
Expand Down Expand Up @@ -116,6 +117,14 @@
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.
Expand All @@ -124,6 +133,7 @@
stateVersion bool
runningFilterInitialize core.RunningEventFilterInitializer
retentionFloor *pruner.RetentionFloor
remoteDatabase bool
}

// Option is a functional option for configuring Blockchain options.
Expand Down Expand Up @@ -151,6 +161,15 @@
}
}

// 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.
Expand Down Expand Up @@ -179,7 +198,7 @@

runningFilter := core.NewRunningEventFilterLazy(database, o.runningFilterInitialize)

return &Blockchain{
chain := &Blockchain{
database: database,
network: network,
listener: o.listener,
Expand All @@ -193,7 +212,16 @@
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 {
Expand All @@ -203,12 +231,36 @@
// 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)
}
Comment thread
brbrr marked this conversation as resolved.

// 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

Check warning on line 250 in blockchain/blockchain.go

View check run for this annotation

Codecov / codecov/patch

blockchain/blockchain.go#L250

Added line #L250 was not covered by tests
}

height, err := core.GetChainHeight(b.database)
if err != nil {
b.chainHeight.Store(nil)
return
}
b.chainHeight.Store(&height)
}
Comment thread
brbrr marked this conversation as resolved.

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
}
Expand All @@ -218,7 +270,7 @@

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
}
Expand Down Expand Up @@ -396,14 +448,47 @@
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)
}
Comment thread
brbrr marked this conversation as resolved.

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

Check warning on line 473 in blockchain/blockchain.go

View check run for this annotation

Codecov / codecov/patch

blockchain/blockchain.go#L473

Added line #L473 was not covered by tests
}

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}
Comment thread
brbrr marked this conversation as resolved.
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.
Expand All @@ -413,6 +498,8 @@
stateUpdate *core.StateUpdate,
newClasses map[felt.Felt]core.ClassDefinition,
) error {
defer b.cacheChainHeight()
Comment thread
brbrr marked this conversation as resolved.

return b.stateBackend.Store(block, blockCommitments, stateUpdate, newClasses)
}

Expand Down Expand Up @@ -471,6 +558,8 @@
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
Expand All @@ -490,6 +579,12 @@

// 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()
}

Expand All @@ -515,6 +610,8 @@
newClasses map[felt.Felt]core.ClassDefinition,
sign core.BlockSignFunc,
) error {
defer b.cacheChainHeight()

return b.stateBackend.Finalise(block, stateUpdate, newClasses, sign)
}

Expand Down
Loading
Loading