perf(store): save metadata async (#3298)

* perf(store): save metadata async

* cl

* Optimize metadata writes with batching

* feedback

* De-duplicate batched writes by key in cached store

* fix

* updates

Author: julienrbrt

CHANGELOG.md

@@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changes
 
 - Optimization of mutex usage in cache for reaper [#3286](https://github.com/evstack/ev-node/pull/3286)
+- Optimize metadata writes by making it async in cache store [#3298](https://github.com/evstack/ev-node/pull/3298)
 - Reduce tx cache retention to avoid OOM under (really) heavy tx load [#3299](https://github.com/evstack/ev-node/pull/3299)
 
 ## v1.1.1

pkg/store/cached_store.go

@@ -2,8 +2,11 @@ package store
 
 import (
 	"context"
+	"sync"
+	"time"
 
 	lru "github.com/hashicorp/golang-lru/v2"
+	"github.com/rs/zerolog"
 
 	"github.com/evstack/ev-node/types"
 )
@@ -14,15 +17,37 @@ const (
 
 	// DefaultBlockDataCacheSize is the default number of block data entries to cache.
 	DefaultBlockDataCacheSize = 200_000
+
+	asyncWriteBufferSize = 8192
+
+	// batchWindow is the time the write goroutine waits after receiving the first
+	// op before flushing. This allows bursts of metadata writes (e.g. 3-4 per
+	// height in the submitter) to be coalesced into a single Badger WriteBatch.
+	batchWindow = 100 * time.Microsecond
 )
 
+type asyncWriteOp struct {
+	key      string
+	value    []byte
+	isDelete bool
+}
+
 // CachedStore wraps a Store with LRU caching for frequently accessed data.
 // The underlying LRU cache is thread-safe, so no additional synchronization is needed.
+// Metadata writes (SetMetadata, DeleteMetadata) are processed asynchronously via a
+// buffered channel to avoid blocking Badger's write pipeline for critical operations
+// like block production (batch commits).
 type CachedStore struct {
 	Store
 
 	headerCache    *lru.Cache[uint64, *types.SignedHeader]
 	blockDataCache *lru.Cache[uint64, *blockDataEntry]
+
+	writeCh chan asyncWriteOp
+	done    chan struct{}
+	stopMu  sync.RWMutex
+	stopped bool
+	logger  zerolog.Logger
 }
 
 type blockDataEntry struct {
@@ -73,6 +98,9 @@ func NewCachedStore(store Store, opts ...CachedStoreOption) (*CachedStore, error
 		Store:          store,
 		headerCache:    headerCache,
 		blockDataCache: blockDataCache,
+		writeCh:        make(chan asyncWriteOp, asyncWriteBufferSize),
+		done:           make(chan struct{}),
+		logger:         zerolog.Nop(),
 	}
 
 	for _, opt := range opts {
@@ -81,9 +109,58 @@ func NewCachedStore(store Store, opts ...CachedStoreOption) (*CachedStore, error
 		}
 	}
 
+	cs.startWriteLoop()
+
 	return cs, nil
 }
 
+func (cs *CachedStore) startWriteLoop() {
+	go func() {
+		defer close(cs.done)
+		for op := range cs.writeCh {
+			ops := []asyncWriteOp{op}
+
+			timer := time.NewTimer(batchWindow)
+		collect:
+			for {
+				select {
+				case op, ok := <-cs.writeCh:
+					if !ok {
+						timer.Stop()
+						break collect
+					}
+					ops = append(ops, op)
+				case <-timer.C:
+					break collect
+				}
+			}
+
+			last := make(map[string]asyncWriteOp, len(ops))
+			for _, o := range ops {
+				last[o.key] = o
+			}
+
+			var puts []MetadataKV
+			var deletes []string
+			for _, o := range last {
+				if o.isDelete {
+					deletes = append(deletes, o.key)
+				} else {
+					puts = append(puts, MetadataKV{Key: o.key, Value: o.value})
+				}
+			}
+
+			if err := cs.BatchMetadata(context.Background(), puts, deletes); err != nil {
+				for _, o := range ops {
+					cs.logger.Error().Err(err).Str("key", o.key).
+						Bool("delete", o.isDelete).
+						Msg("async metadata batch write failed")
+				}
+			}
+		}
+	}()
+}
+
 // GetHeader returns the header at the given height, using the cache if available.
 func (cs *CachedStore) GetHeader(ctx context.Context, height uint64) (*types.SignedHeader, error) {
 	// Try cache first
@@ -162,7 +239,7 @@ func (cs *CachedStore) Rollback(ctx context.Context, height uint64, aggregator b
 }
 
 // PruneBlocks wraps the underlying store's PruneBlocks and invalidates caches
-// up to the heigh that we purne
+// up to the height that we prune
 func (cs *CachedStore) PruneBlocks(ctx context.Context, height uint64) error {
 	if err := cs.Store.PruneBlocks(ctx, height); err != nil {
 		return err
@@ -173,8 +250,42 @@ func (cs *CachedStore) PruneBlocks(ctx context.Context, height uint64) error {
 	return nil
 }
 
-// Close closes the underlying store.
+// SetMetadata queues an asynchronous metadata write. The write is persisted
+// by the background goroutine via BatchMetadata. If the store has been stopped,
+// the write falls back to synchronous execution on the underlying store.
+func (cs *CachedStore) SetMetadata(ctx context.Context, key string, value []byte) error {
+	cs.stopMu.RLock()
+	defer cs.stopMu.RUnlock()
+
+	if cs.stopped {
+		return cs.Store.SetMetadata(ctx, key, value)
+	}
+	valueCopy := append([]byte(nil), value...)
+	cs.writeCh <- asyncWriteOp{key: key, value: valueCopy}
+	return nil
+}
+
+// DeleteMetadata queues an asynchronous metadata delete. If the store has been
+// stopped, the delete falls back to synchronous execution.
+func (cs *CachedStore) DeleteMetadata(ctx context.Context, key string) error {
+	cs.stopMu.RLock()
+	defer cs.stopMu.RUnlock()
+
+	if cs.stopped {
+		return cs.Store.DeleteMetadata(ctx, key)
+	}
+	cs.writeCh <- asyncWriteOp{key: key, isDelete: true}
+	return nil
+}
+
+// Close drains pending async writes, then closes the underlying store.
 func (cs *CachedStore) Close() error {
+	cs.stopMu.Lock()
+	cs.stopped = true
+	close(cs.writeCh)
+	cs.stopMu.Unlock()
+	<-cs.done
+
 	cs.ClearCache()
 	return cs.Store.Close()
 }

pkg/store/cached_store_test.go

@@ -2,8 +2,12 @@ package store
 
 import (
 	"context"
+	"fmt"
 	"testing"
+	"time"
 
+	ds "github.com/ipfs/go-datastore"
+	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -270,3 +274,129 @@ func TestCachedStore_Close(t *testing.T) {
 	err = cachedStore.Close()
 	require.NoError(t, err)
 }
+
+func TestCachedStore_AsyncSetMetadata(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+
+	kv, err := NewTestInMemoryKVStore()
+	require.NoError(t, err)
+
+	base := New(kv)
+	cs, err := NewCachedStore(base)
+	require.NoError(t, err)
+	t.Cleanup(func() { cs.Close() })
+
+	require.NoError(t, cs.SetMetadata(ctx, "key1", []byte("value1")))
+
+	require.Eventually(t, func() bool {
+		v, err := base.GetMetadata(ctx, "key1")
+		return err == nil && string(v) == "value1"
+	}, time.Second, 10*time.Millisecond)
+}
+
+func TestCachedStore_AsyncDeleteMetadata(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+
+	kv, err := NewTestInMemoryKVStore()
+	require.NoError(t, err)
+
+	base := New(kv)
+	require.NoError(t, base.SetMetadata(ctx, "key1", []byte("value1")))
+
+	cs, err := NewCachedStore(base)
+	require.NoError(t, err)
+	t.Cleanup(func() { cs.Close() })
+
+	require.NoError(t, cs.DeleteMetadata(ctx, "key1"))
+
+	require.Eventually(t, func() bool {
+		_, err := base.GetMetadata(ctx, "key1")
+		return err != nil
+	}, time.Second, 10*time.Millisecond)
+}
+
+func TestCachedStore_Close_FlushesPendingWrites(t *testing.T) {
+	ctx := context.Background()
+
+	dir := t.TempDir()
+	kv, err := NewDefaultKVStore(dir, "", "test-db")
+	require.NoError(t, err)
+
+	base := New(kv)
+	cs, err := NewCachedStore(base)
+	require.NoError(t, err)
+
+	const n = 100
+	for i := range n {
+		k := fmt.Sprintf("key-%d", i)
+		require.NoError(t, cs.SetMetadata(ctx, k, []byte(k)))
+	}
+
+	require.NoError(t, cs.Close())
+
+	kv2, err := NewDefaultKVStore(dir, "", "test-db")
+	require.NoError(t, err)
+	t.Cleanup(func() { kv2.Close() })
+	reopened := New(kv2)
+
+	for i := range n {
+		k := fmt.Sprintf("key-%d", i)
+		v, err := reopened.GetMetadata(ctx, k)
+		require.NoError(t, err)
+		require.Equal(t, []byte(k), v)
+	}
+}
+
+func TestCachedStore_WriteAfterClose_FallsBack(t *testing.T) {
+	kv, err := NewTestInMemoryKVStore()
+	require.NoError(t, err)
+
+	base := New(kv)
+	cs, err := NewCachedStore(base)
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	require.NoError(t, cs.SetMetadata(ctx, "before", []byte("ok")))
+
+	require.NoError(t, cs.Close())
+
+	err = cs.SetMetadata(ctx, "after", []byte("sync"))
+	require.Error(t, err)
+}
+
+func TestCachedStore_CoalescesSameKeyOps(t *testing.T) {
+	ctx := context.Background()
+
+	kv, err := NewTestInMemoryKVStore()
+	require.NoError(t, err)
+
+	require.NoError(t, kv.Put(ctx, ds.NewKey(GetMetaKey("k")), []byte("original")))
+
+	base := New(kv)
+
+	writeCh := make(chan asyncWriteOp, asyncWriteBufferSize)
+	done := make(chan struct{})
+	cs := &CachedStore{
+		Store:   base,
+		writeCh: writeCh,
+		done:    done,
+		logger:  zerolog.Nop(),
+	}
+	cs.startWriteLoop()
+
+	require.NoError(t, cs.SetMetadata(ctx, "k", []byte("v1")))
+	require.NoError(t, cs.DeleteMetadata(ctx, "k"))
+	require.NoError(t, cs.SetMetadata(ctx, "k", []byte("v2")))
+
+	cs.stopMu.Lock()
+	cs.stopped = true
+	close(writeCh)
+	cs.stopMu.Unlock()
+	<-done
+
+	v, err := base.GetMetadata(ctx, "k")
+	require.NoError(t, err)
+	require.Equal(t, []byte("v2"), v, "last write (Set) should win over delete")
+}

pkg/store/store.go

@@ -190,6 +190,30 @@ func (s *DefaultStore) SetMetadata(ctx context.Context, key string, value []byte
 	return nil
 }
 
+func (s *DefaultStore) BatchMetadata(ctx context.Context, puts []MetadataKV, deletes []string) error {
+	if len(puts) == 0 && len(deletes) == 0 {
+		return nil
+	}
+	batch, err := s.db.Batch(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to create metadata batch: %w", err)
+	}
+	for _, kv := range puts {
+		if err := batch.Put(ctx, ds.NewKey(GetMetaKey(kv.Key)), kv.Value); err != nil {
+			return fmt.Errorf("failed to batch-put metadata key '%s': %w", kv.Key, err)
+		}
+	}
+	for _, key := range deletes {
+		if err := batch.Delete(ctx, ds.NewKey(GetMetaKey(key))); err != nil {
+			return fmt.Errorf("failed to batch-delete metadata key '%s': %w", key, err)
+		}
+	}
+	if err := batch.Commit(ctx); err != nil {
+		return fmt.Errorf("failed to commit metadata batch: %w", err)
+	}
+	return nil
+}
+
 // GetMetadata returns values stored for given key with SetMetadata.
 func (s *DefaultStore) GetMetadata(ctx context.Context, key string) ([]byte, error) {
 	data, err := s.db.Get(ctx, ds.NewKey(GetMetaKey(key)))

pkg/store/tracing.go

@@ -211,6 +211,25 @@ func (t *tracedStore) DeleteMetadata(ctx context.Context, key string) error {
 	return nil
 }
 
+func (t *tracedStore) BatchMetadata(ctx context.Context, puts []MetadataKV, deletes []string) error {
+	ctx, span := t.tracer.Start(ctx, "Store.BatchMetadata",
+		trace.WithAttributes(
+			attribute.Int("puts", len(puts)),
+			attribute.Int("deletes", len(deletes)),
+		),
+	)
+	defer span.End()
+
+	err := t.inner.BatchMetadata(ctx, puts, deletes)
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
+		return err
+	}
+
+	return nil
+}
+
 func (t *tracedStore) DeleteStateAtHeight(ctx context.Context, height uint64) error {
 	ctx, span := t.tracer.Start(ctx, "Store.DeleteStateAtHeight",
 		trace.WithAttributes(attribute.Int64("height", int64(height))),