Skip to content

refactor(cache): save caches to disk#2282

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
julienrbrt merged 5 commits into from
May 21, 2025
Merged

refactor(cache): save caches to disk #2282

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
217d2b9
refactor(cache): save caches to disk
julienrbrt May 16, 2025
f670fcf
updates
julienrbrt May 16, 2025
ca7e6c3
implement feedback
julienrbrt May 19, 2025
c77c15f
Merge branch 'main' into julien/cache
julienrbrt May 20, 2025
4397190
logs consistency
julienrbrt May 21, 2025
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
80 changes: 60 additions & 20 deletions block/manager.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,9 +4,11 @@ import (
"bytes"
"context"
"encoding/binary"
"encoding/gob"
"encoding/hex"
"errors"
"fmt"
"path/filepath"
"sync"
"sync/atomic"
"time"
Expand DownExpand Up@@ -276,8 +278,7 @@ func NewManager(
}

// set block height in store
err = store.SetHeight(ctx, s.LastBlockHeight)
if err != nil {
if err = store.SetHeight(ctx, s.LastBlockHeight); err != nil {
return nil, err
}

Expand All@@ -286,22 +287,22 @@ func NewManager(
}

if config.DA.BlockTime.Duration == 0 {
logger.Info("Using default DA block time", "DABlockTime", defaultDABlockTime)
logger.Info("using default DA block time", "DABlockTime", defaultDABlockTime)
config.DA.BlockTime.Duration = defaultDABlockTime
}

if config.Node.BlockTime.Duration == 0 {
logger.Info("Using default block time", "BlockTime", defaultBlockTime)
logger.Info("using default block time", "BlockTime", defaultBlockTime)
config.Node.BlockTime.Duration = defaultBlockTime
}

if config.Node.LazyBlockInterval.Duration == 0 {
logger.Info("Using default lazy block time", "LazyBlockTime", defaultLazyBlockTime)
logger.Info("using default lazy block time", "LazyBlockTime", defaultLazyBlockTime)
config.Node.LazyBlockInterval.Duration = defaultLazyBlockTime
}

if config.DA.MempoolTTL == 0 {
logger.Info("Using default mempool ttl", "MempoolTTL", defaultMempoolTTL)
logger.Info("using default mempool ttl", "MempoolTTL", defaultMempoolTTL)
config.DA.MempoolTTL = defaultMempoolTTL
}

Expand All@@ -312,7 +313,7 @@ func NewManager(

// If lastBatchHash is not set, retrieve the last batch hash from store
lastBatchDataBytes, err := store.GetMetadata(ctx, LastBatchDataKey)
if err != nil {
if err != nil && s.LastBlockHeight > 0 {
logger.Error("error while retrieving last batch hash", "error", err)
}

Expand All@@ -324,7 +325,7 @@ func NewManager(
daH := atomic.Uint64{}
daH.Store(s.DAHeight)

agg := &Manager{
m := &Manager{
signer: signer,
config: config,
genesis: genesis,
Expand DownExpand Up@@ -358,16 +359,26 @@ func NewManager(
txNotifyCh: make(chan struct{}, 1), // Non-blocking channel
batchSubmissionChan: make(chan coresequencer.Batch, eventInChLength),
}
agg.init(ctx)

// initialize da included height
if height, err := m.store.GetMetadata(ctx, DAIncludedHeightKey); err == nil && len(height) == 8 {
Comment thread
julienrbrt marked this conversation as resolved.
m.daIncludedHeight.Store(binary.LittleEndian.Uint64(height))
}

Comment on lines +363 to +367

@coderabbitaicoderabbitaiBotMay 16, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Missing error-handling for unexpected metadata format

GetMetadata may return a byte‐slice of unexpected length.
If the length is neither 0 nor 8 the current code silently ignores the error case, which could hide data corruption:

ifheight, err:=m.store.GetMetadata(ctx, DAIncludedHeightKey); err==nil {
iflen(height) !=8 {
returnnil, fmt.Errorf("corrupted DAIncludedHeight metadata, length=%d", len(height))
}
m.daIncludedHeight.Store(binary.LittleEndian.Uint64(height))
}

This protects later logic from a poisoned daIncludedHeight.

🤖 Prompt for AI Agents
In block/manager.go around lines 366 to 370, the code does not handle the case
where the metadata byte slice length is neither 0 nor 8, which could indicate
data corruption. Modify the code to check if the length of the returned byte
slice is exactly 8; if not, return an error indicating corrupted
DAIncludedHeight metadata with the actual length. This ensures that invalid
metadata formats are caught early and prevents using a poisoned daIncludedHeight
value.

// Set the default publishBlock implementation
agg.publishBlock = agg.publishBlockInternal
if s, ok := agg.sequencer.(interface {
m.publishBlock = m.publishBlockInternal
if s, ok := m.sequencer.(interface {
SetBatchSubmissionChan(chan coresequencer.Batch)
}); ok {
s.SetBatchSubmissionChan(agg.batchSubmissionChan)
s.SetBatchSubmissionChan(m.batchSubmissionChan)
}

return agg, nil
// fetch caches from disks
if err := m.LoadCache(); err != nil {
return nil, fmt.Errorf("failed to load cache: %w", err)
}

return m, nil
}

// PendingHeaders returns the pending headers.
Expand All@@ -387,13 +398,6 @@ func (m *Manager) GetLastState() types.State {
return m.lastState
}

func (m *Manager) init(ctx context.Context) {
// initialize da included height
if height, err := m.store.GetMetadata(ctx, DAIncludedHeightKey); err == nil && len(height) == 8 {
m.daIncludedHeight.Store(binary.LittleEndian.Uint64(height))
}
}

// GetDAIncludedHeight returns the rollup height at which all blocks have been
// included in the DA
func (m *Manager) GetDAIncludedHeight() uint64 {
Expand DownExpand Up@@ -925,6 +929,12 @@ func (m *Manager) NotifyNewTransactions() {
}
}

var (
cacheDir = "cache"
headerCacheDir = filepath.Join(cacheDir, "header")
dataCacheDir = filepath.Join(cacheDir, "data")
)

// HeaderCache returns the headerCache used by the manager.
func (m *Manager) HeaderCache() *cache.Cache[types.SignedHeader] {
return m.headerCache
Expand All@@ -934,3 +944,33 @@ func (m *Manager) HeaderCache() *cache.Cache[types.SignedHeader] {
func (m *Manager) DataCache() *cache.Cache[types.Data] {
return m.dataCache
}

// LoadCache loads the header and data caches from disk.
func (m *Manager) LoadCache() error {
gob.Register(&types.SignedHeader{})

@tac0turtletac0turtleMay 19, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do we need to use gob here? seems like an extra dependency which could be avoided

gob.Register(&types.Data{})

cfgDir := filepath.Dir(m.config.ConfigPath())

if err := m.headerCache.LoadFromDisk(filepath.Join(cfgDir, headerCacheDir)); err != nil {
return fmt.Errorf("failed to load header cache from disk: %w", err)
}

if err := m.dataCache.LoadFromDisk(filepath.Join(cfgDir, dataCacheDir)); err != nil {
return fmt.Errorf("failed to load data cache from disk: %w", err)
}

return nil
}

// SaveCache saves the header and data caches to disk.
func (m *Manager) SaveCache() error {
if err := m.headerCache.SaveToDisk(filepath.Join(filepath.Dir(m.config.ConfigPath()), headerCacheDir)); err != nil {
return fmt.Errorf("failed to save header cache to disk: %w", err)
}

if err := m.dataCache.SaveToDisk(filepath.Join(filepath.Dir(m.config.ConfigPath()), dataCacheDir)); err != nil {
return fmt.Errorf("failed to save data cache to disk: %w", err)
}
return nil
}
10 changes: 7 additions & 3 deletions block/store_test.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -3,6 +3,7 @@ package block
import (
// ... other necessary imports ...
"context"
"encoding/binary"
"errors"
"sync"
"sync/atomic"
Expand All@@ -13,6 +14,7 @@ import (
ds "github.com/ipfs/go-datastore"
"github.com/rollkit/rollkit/pkg/config"
"github.com/rollkit/rollkit/pkg/signer/noop"

// Use existing store mock if available, or define one
mocksStore "github.com/rollkit/rollkit/test/mocks"
extmocks "github.com/rollkit/rollkit/test/mocks/external"
Expand DownExpand Up@@ -76,8 +78,11 @@ func setupManagerForStoreRetrieveTest(t *testing.T) (
config: nodeConf,
signer: signer,
}
// Initialize daHeight atomic variable
m.init(ctx) // Call init to handle potential DAIncludedHeightKey loading

// initialize da included height
if height, err := m.store.GetMetadata(ctx, DAIncludedHeightKey); err == nil && len(height) == 8 {
m.daIncludedHeight.Store(binary.LittleEndian.Uint64(height))
}

return m, mockStore, mockHeaderStore, mockDataStore, headerStoreCh, dataStoreCh, headerInCh, dataInCh, ctx, cancel
}
Expand DownExpand Up@@ -218,7 +223,6 @@ func TestDataStoreRetrieveLoop_NoNewData(t *testing.T) {

// TestDataStoreRetrieveLoop_HandlesFetchError verifies that the data store retrieve loop handles fetch errors gracefully.
func TestDataStoreRetrieveLoop_HandlesFetchError(t *testing.T) {

m, mockStore, _, mockDataStore, _, dataStoreCh, _, dataInCh, ctx, cancel := setupManagerForStoreRetrieveTest(t)
defer cancel()

Expand Down
2 changes: 1 addition & 1 deletion da/jsonrpc/client.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -239,7 +239,7 @@ func newClient(ctx context.Context, logger log.Logger, addr string, authHeader h
return nil, fmt.Errorf("failed to decode namespace: %w", err)
}
client.DA.Namespace = namespaceBytes
logger.Info("Creating new client", "namespace", namespace)
logger.Info("creating new client", "namespace", namespace)
errs := getKnownErrorsMapping()
for name, module := range moduleMap(&client) {
closer, err := jsonrpc.NewMergeClient(ctx, addr, name, []interface{}{module}, authHeader, jsonrpc.WithErrors(errs))
Expand Down
27 changes: 13 additions & 14 deletions node/full.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -377,13 +377,13 @@ func (n *FullNode) Run(parentCtx context.Context) error {
}

go func() {
if err := n.rpcServer.ListenAndServe(); err != http.ErrServerClosed {
err := n.rpcServer.ListenAndServe()
if err != nil && err != http.ErrServerClosed {
n.Logger.Error("RPC server error", "err", err)
}
n.Logger.Info("started RPC server", "addr", n.nodeConfig.RPC.Address)
}()

n.Logger.Info("Started RPC server", "addr", n.nodeConfig.RPC.Address)

n.Logger.Info("starting P2P client")
err = n.p2pClient.Start(ctx)
if err != nil {
Expand DownExpand Up@@ -432,8 +432,7 @@ func (n *FullNode) Run(parentCtx context.Context) error {
}

// Perform cleanup
n.Logger.Info("halting full node...")
n.Logger.Info("shutting down full node sub services...")
n.Logger.Info("halting full node and its sub services...")

// Use a timeout context to ensure shutdown doesn't hang
shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
Expand All@@ -444,7 +443,6 @@ func (n *FullNode) Run(parentCtx context.Context) error {
// Stop P2P Client
err = n.p2pClient.Close()
if err != nil {
n.Logger.Error("error closing P2P client", "error", err)
multiErr = errors.Join(multiErr, fmt.Errorf("closing P2P client: %w", err))
}

Expand DownExpand Up@@ -479,7 +477,6 @@ func (n *FullNode) Run(parentCtx context.Context) error {
err = n.prometheusSrv.Shutdown(shutdownCtx)
// http.ErrServerClosed is expected on graceful shutdown
if err != nil && !errors.Is(err, http.ErrServerClosed) {
n.Logger.Error("error shutting down Prometheus server", "error", err)
multiErr = errors.Join(multiErr, fmt.Errorf("shutting down Prometheus server: %w", err))
}
}
Expand All@@ -488,7 +485,6 @@ func (n *FullNode) Run(parentCtx context.Context) error {
if n.pprofSrv != nil {
err = n.pprofSrv.Shutdown(shutdownCtx)
if err != nil && !errors.Is(err, http.ErrServerClosed) {
n.Logger.Error("error shutting down pprof server", "error", err)
multiErr = errors.Join(multiErr, fmt.Errorf("shutting down pprof server: %w", err))
}
}
Expand All@@ -497,22 +493,25 @@ func (n *FullNode) Run(parentCtx context.Context) error {
if n.rpcServer != nil {
err = n.rpcServer.Shutdown(shutdownCtx)
if err != nil && !errors.Is(err, http.ErrServerClosed) {
n.Logger.Error("error shutting down RPC server", "error", err)
multiErr = errors.Join(multiErr, fmt.Errorf("shutting down RPC server: %w", err))
}
}

// Ensure Store.Close is called last to maximize chance of data flushing
err = n.Store.Close()
if err != nil {
// Store.Close() might log internally, but log here too for context
n.Logger.Error("error closing store", "error", err)
if err = n.Store.Close(); err != nil {
multiErr = errors.Join(multiErr, fmt.Errorf("closing store: %w", err))
}

// Save caches if needed
if err := n.blockManager.SaveCache(); err != nil {
multiErr = errors.Join(multiErr, fmt.Errorf("saving caches: %w", err))
}

// Log final status
if multiErr != nil {
n.Logger.Error("errors encountered while stopping node", "errors", multiErr)
for _, err := range multiErr.(interface{ Unwrap() []error }).Unwrap() {
n.Logger.Error("error during shutdown", "error", err)
}
} else {
n.Logger.Info("full node halted successfully")
}
Expand Down
Loading