Merge branch 'main' into alex/cache_io

* main:
  refactor: use state da height as well (#2872)
  refactor: retrieve highest da height in cache (#2870)
  chore: change from event count to start and end height (#2871)

Author: alpe

block/internal/cache/generic_cache.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 
 	"golang.org/x/sync/errgroup"
+	"sync/atomic"
 )
 
 // Cache is a generic cache that maintains items that are seen and hard confirmed
@@ -22,6 +23,8 @@ type Cache[T any] struct {
 	daIncluded *sync.Map
 	// hashByHeight tracks the hash associated with each height for pruning
 	hashByHeight *sync.Map
+	// maxDAHeight tracks the maximum DA height seen
+	maxDAHeight *atomic.Uint64
 }
 
 // NewCache returns a new Cache struct
@@ -31,6 +34,7 @@ func NewCache[T any]() *Cache[T] {
 		hashes:        new(sync.Map),
 		daIncluded:    new(sync.Map),
 		hashByHeight:  new(sync.Map),
+		maxDAHeight:   &atomic.Uint64{},
 	}
 }
 
@@ -95,13 +99,25 @@ func (c *Cache[T]) getDAIncluded(hash string) (uint64, bool) {
 func (c *Cache[T]) setDAIncluded(hash string, daHeight uint64, blockHeight uint64) {
 	c.daIncluded.Store(hash, daHeight)
 	c.hashByHeight.Store(blockHeight, hash)
+
+	// Update max DA height if necessary
+	current := c.maxDAHeight.Load()
+	if daHeight >= current {
+		_ = c.maxDAHeight.CompareAndSwap(current, daHeight)
+	}
 }
 
 // removeDAIncluded removes the DA-included status of the hash
 func (c *Cache[T]) removeDAIncluded(hash string) {
 	c.daIncluded.Delete(hash)
 }
 
+// daHeight returns the maximum DA height from all DA-included items.
+// Returns 0 if no items are DA-included.
+func (c *Cache[T]) daHeight() uint64 {
+	return c.maxDAHeight.Load()
+}
+
 // deleteAllForHeight removes all items and their associated data from the cache at the given height
 func (c *Cache[T]) deleteAllForHeight(height uint64) {
 	c.itemsByHeight.Delete(height)
@@ -255,8 +271,12 @@ func (c *Cache[T]) LoadFromDisk(folderPath string) error {
 		}
 		for k, v := range daIncludedMap {
 			c.daIncluded.Store(k, v)
+			// Update max DA height during load
+			current := c.maxDAHeight.Load()
+			if v > current {
+				_ = c.maxDAHeight.CompareAndSwap(current, v)
+			}
 		}
-
 		return nil
 	})
 	return wg.Wait()

block/internal/cache/generic_cache_test.go

@@ -14,6 +14,47 @@ func init() {
 	gob.Register(&testItem{})
 }
 
+// TestCache_MaxDAHeight verifies that daHeight tracks the maximum DA height
+func TestCache_MaxDAHeight(t *testing.T) {
+	c := NewCache[testItem]()
+
+	// Initially should be 0
+	if got := c.daHeight(); got != 0 {
+		t.Errorf("initial daHeight = %d, want 0", got)
+	}
+
+	// Set items with increasing DA heights
+	c.setDAIncluded("hash1", 100, 1)
+	if got := c.daHeight(); got != 100 {
+		t.Errorf("after setDAIncluded(100): daHeight = %d, want 100", got)
+	}
+
+	c.setDAIncluded("hash2", 50, 2) // Lower height shouldn't change max
+	if got := c.daHeight(); got != 100 {
+		t.Errorf("after setDAIncluded(50): daHeight = %d, want 100", got)
+	}
+
+	c.setDAIncluded("hash3", 200, 3)
+	if got := c.daHeight(); got != 200 {
+		t.Errorf("after setDAIncluded(200): daHeight = %d, want 200", got)
+	}
+
+	// Test persistence
+	dir := t.TempDir()
+	if err := c.SaveToDisk(dir); err != nil {
+		t.Fatalf("SaveToDisk failed: %v", err)
+	}
+
+	c2 := NewCache[testItem]()
+	if err := c2.LoadFromDisk(dir); err != nil {
+		t.Fatalf("LoadFromDisk failed: %v", err)
+	}
+
+	if got := c2.daHeight(); got != 200 {
+		t.Errorf("after load: daHeight = %d, want 200", got)
+	}
+}
+
 // TestCache_SaveLoad_ErrorPaths covers SaveToDisk and LoadFromDisk error scenarios.
 func TestCache_SaveLoad_ErrorPaths(t *testing.T) {
 	c := NewCache[testItem]()

block/internal/cache/manager.go

@@ -49,6 +49,7 @@ type Manager interface {
 	GetHeaderDAIncluded(hash string) (uint64, bool)
 	SetHeaderDAIncluded(hash string, daHeight uint64, blockHeight uint64)
 	RemoveHeaderDAIncluded(hash string)
+	DaHeight() uint64
 
 	// Data operations
 	IsDataSeen(hash string) bool
@@ -165,6 +166,11 @@ func (m *implementation) RemoveHeaderDAIncluded(hash string) {
 	m.headerCache.removeDAIncluded(hash)
 }
 
+// DaHeight fetches the heights da height contained in the processed cache.
+func (m *implementation) DaHeight() uint64 {
+	return max(m.headerCache.daHeight(), m.dataCache.daHeight())
+}
+
 // Data operations
 func (m *implementation) IsDataSeen(hash string) bool {
 	return m.dataCache.isSeen(hash)

block/internal/syncing/da_retriever.go

@@ -214,7 +214,18 @@ func (r *DARetriever) processBlobs(ctx context.Context, blobs [][]byte, daHeight
 	}
 
 	if len(events) > 0 {
-		r.logger.Info().Int("count", len(events)).Uint64("da_height", daHeight).Msg("processed blocks from DA")
+		startHeight := events[0].Header.Height()
+		endHeight := events[0].Header.Height()
+		for _, event := range events {
+			h := event.Header.Height()
+			if h < startHeight {
+				startHeight = h
+			}
+			if h > endHeight {
+				endHeight = h
+			}
+		}
+		r.logger.Info().Uint64("da_height", daHeight).Uint64("start_height", startHeight).Uint64("end_height", endHeight).Msg("processed blocks from DA")
 	}
 
 	return events

block/internal/syncing/syncer.go

@@ -177,7 +177,7 @@ func (s *Syncer) SetLastState(state types.State) {
 
 // GetDAHeight returns the current DA height
 func (s *Syncer) GetDAHeight() uint64 {
-	return s.daHeight.Load()
+	return max(s.daHeight.Load(), s.cache.DaHeight())
 }
 
 // SetDAHeight updates the DA height
@@ -217,7 +217,9 @@ func (s *Syncer) initializeState() error {
 	s.SetLastState(state)
 
 	// Set DA height
-	s.SetDAHeight(state.DAHeight)
+	// we get the max from the genesis da height, the state da height and the cache (fetched) da height
+	// if a user has messed up and sync da too far ahead, on restart they can clear the cache (--clear-cache) and the retrieve will restart fetching from the last known block synced and executed from DA or the set genesis da height.
+	s.SetDAHeight(max(s.genesis.DAStartHeight, s.cache.DaHeight(), state.DAHeight))
 
 	s.logger.Info().
 		Uint64("height", state.LastBlockHeight).

block/internal/syncing/syncer_test.go

@@ -433,7 +433,6 @@ func TestSyncLoopPersistState(t *testing.T) {
 
 	t.Log("sync workers on instance1 completed")
 	require.Equal(t, myFutureDAHeight, syncerInst1.GetDAHeight())
-	lastStateDAHeight := syncerInst1.GetLastState().DAHeight
 
 	// wait for all events consumed
 	require.NoError(t, cacheMgr.SaveToDisk())
@@ -470,7 +469,6 @@ func TestSyncLoopPersistState(t *testing.T) {
 		make(chan error, 1),
 	)
 	require.NoError(t, syncerInst2.initializeState())
-	require.Equal(t, lastStateDAHeight, syncerInst2.GetDAHeight())
 
 	ctx, cancel = context.WithCancel(t.Context())
 	t.Cleanup(cancel)
@@ -484,7 +482,7 @@ func TestSyncLoopPersistState(t *testing.T) {
 		Run(func(arg mock.Arguments) {
 			cancel()
 			// retrieve last one again
-			assert.Equal(t, lastStateDAHeight, arg.Get(1).(uint64))
+			assert.Equal(t, syncerInst2.GetDAHeight(), arg.Get(1).(uint64))
 		}).
 		Return(nil, nil)
 
@@ -589,6 +587,11 @@ func TestSyncer_InitializeState_CallsReplayer(t *testing.T) {
 	// This test verifies that initializeState() invokes Replayer.
 	// The detailed replay logic is tested in block/internal/common/replay_test.go
 
+	ds := dssync.MutexWrap(datastore.NewMapDatastore())
+	st := store.New(ds)
+	cm, err := cache.NewManager(config.DefaultConfig(), st, zerolog.Nop())
+	require.NoError(t, err)
+
 	// Create mocks
 	mockStore := testmocks.NewMockStore(t)
 	mockExec := testmocks.NewMockHeightAwareExecutor(t)
@@ -627,17 +630,17 @@ func TestSyncer_InitializeState_CallsReplayer(t *testing.T) {
 		daHeight:  &atomic.Uint64{},
 		logger:    zerolog.Nop(),
 		ctx:       context.Background(),
+		cache:     cm,
 	}
 
 	// Initialize state - this should call Replayer
-	err := syncer.initializeState()
+	err = syncer.initializeState()
 	require.NoError(t, err)
 
 	// Verify state was initialized correctly
 	state := syncer.GetLastState()
 	assert.Equal(t, storeHeight, state.LastBlockHeight)
 	assert.Equal(t, gen.ChainID, state.ChainID)
-	assert.Equal(t, uint64(5), syncer.GetDAHeight())
 
 	// Verify that GetLatestHeight was called (proves Replayer was invoked)
 	mockExec.AssertCalled(t, "GetLatestHeight", mock.Anything)

tools/da-debug/README.md

@@ -74,7 +74,7 @@ da-debug search 100 "0x000000000000000000000000000000000000000000000000000000746
 All commands support these global flags:
 
 <!-- markdown-link-check-disable -->
-- `--da-url string`: DA layer JSON-RPC URL (default: "http://localhost:7980")
+- `--da-url string`: DA layer JSON-RPC URL (default: "<http://localhost:7980>")
 <!-- markdown-link-check-enable -->
 - `--auth-token string`: Authentication token for DA layer
 - `--timeout duration`: Request timeout (default: 30s)