feat: Expose Rollkit height to DA Height mapping (#2363)

<!--
Please read and fill out this form before submitting your PR.

Please make sure you have reviewed our contributors guide before
submitting your
first PR.

NOTE: PR titles should follow semantic commits:
https://www.conventionalcommits.org/en/v1.0.0/
-->

## Overview

Expose Rollkit height to DA Height mapping

Closes: #2360

<!-- 
Please provide an explanation of the PR, including the appropriate
context,
background, goal, and rationale. If there is an issue with this
information,
please provide a tl;dr and link the issue. 

Ex: Closes #<issue number>
-->


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Tracks and stores the specific height at which data is included in the
DA layer, enabling precise synchronization and mapping between Rollkit
and DA block heights.
- Enhanced logging to provide more detailed context on DA inclusion
events.

- **Bug Fixes**
- Ensures correct association of DA inclusion heights with headers and
data, improving block processing accuracy and error handling.

- **Documentation**
- Updated block validity specifications to clarify focus on data
validation and field requirements.

- **Tests**
- Expanded tests to cover DA height tracking, error scenarios,
concurrency, and metadata persistence, ensuring robust and reliable
behavior.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: Manav-Aggarwal

block/da_includer.go

@@ -28,6 +28,11 @@ func (m *Manager) DAIncluderLoop(ctx context.Context, errCh chan<- error) {
 				break
 			}
 			if daIncluded {
+				m.logger.Debug("both header and data are DA-included, advancing height", "height", nextHeight)
+				if err := m.SetRollkitHeightToDAHeight(ctx, nextHeight); err != nil {
+					errCh <- fmt.Errorf("failed to set rollkit height to DA height: %w", err)
+					return
+				}
 				// Both header and data are DA-included, so we can advance the height
 				if err := m.incrementDAIncludedHeight(ctx); err != nil {
 					errCh <- fmt.Errorf("error while incrementing DA included height: %w", err)

block/da_includer_test.go

@@ -3,6 +3,7 @@ package block
 import (
 	"context"
 	"encoding/binary"
+	"fmt"
 	"sync"
 	"testing"
 	"time"
@@ -53,11 +54,19 @@ func TestDAIncluderLoop_AdvancesHeightWhenBothDAIncluded(t *testing.T) {
 	header, data := types.GetRandomBlock(5, 1, "testchain")
 	headerHash := header.Hash().String()
 	dataHash := data.DACommitment().String()
-	m.headerCache.SetDAIncluded(headerHash)
-	m.dataCache.SetDAIncluded(dataHash)
+	m.headerCache.SetDAIncluded(headerHash, uint64(1))
+	m.dataCache.SetDAIncluded(dataHash, uint64(1))
 
-	store.On("GetBlockData", mock.Anything, uint64(5)).Return(header, data, nil).Once()
+	store.On("GetBlockData", mock.Anything, uint64(5)).Return(header, data, nil).Times(2)
 	store.On("GetBlockData", mock.Anything, uint64(6)).Return(nil, nil, assert.AnError).Once()
+	// Mock expectations for SetRollkitHeightToDAHeight method
+	headerHeightBytes := make([]byte, 8)
+	binary.LittleEndian.PutUint64(headerHeightBytes, uint64(1))
+	store.On("SetMetadata", mock.Anything, fmt.Sprintf("%s/%d/h", RollkitHeightToDAHeightKey, uint64(5)), headerHeightBytes).Return(nil).Once()
+	dataHeightBytes := make([]byte, 8)
+	binary.LittleEndian.PutUint64(dataHeightBytes, uint64(1))
+	store.On("SetMetadata", mock.Anything, fmt.Sprintf("%s/%d/d", RollkitHeightToDAHeightKey, uint64(5)), dataHeightBytes).Return(nil).Once()
+	// Mock expectations for incrementDAIncludedHeight method
 	heightBytes := make([]byte, 8)
 	binary.LittleEndian.PutUint64(heightBytes, expectedDAIncludedHeight)
 	store.On("SetMetadata", mock.Anything, DAIncludedHeightKey, heightBytes).Return(nil).Once()
@@ -93,7 +102,7 @@ func TestDAIncluderLoop_StopsWhenHeaderNotDAIncluded(t *testing.T) {
 
 	header, data := types.GetRandomBlock(5, 1, "testchain")
 	// m.headerCache.SetDAIncluded(headerHash) // Not set
-	m.dataCache.SetDAIncluded(data.DACommitment().String())
+	m.dataCache.SetDAIncluded(data.DACommitment().String(), uint64(1))
 
 	store.On("GetBlockData", mock.Anything, uint64(5)).Return(header, data, nil).Once()
 
@@ -124,7 +133,7 @@ func TestDAIncluderLoop_StopsWhenDataNotDAIncluded(t *testing.T) {
 
 	header, data := types.GetRandomBlock(5, 1, "testchain")
 	headerHash := header.Hash().String()
-	m.headerCache.SetDAIncluded(headerHash)
+	m.headerCache.SetDAIncluded(headerHash, uint64(1))
 	// m.dataCache.SetDAIncluded(data.DACommitment().String()) // Not set
 
 	store.On("GetBlockData", mock.Anything, uint64(5)).Return(header, data, nil).Once()
@@ -272,12 +281,22 @@ func TestDAIncluderLoop_MultipleConsecutiveHeightsDAIncluded(t *testing.T) {
 		headers[i], dataBlocks[i] = types.GetRandomBlock(height, numTxs, "testchain")
 		headerHash := headers[i].Hash().String()
 		dataHash := dataBlocks[i].DACommitment().String()
-		m.headerCache.SetDAIncluded(headerHash)
-		m.dataCache.SetDAIncluded(dataHash)
-		store.On("GetBlockData", mock.Anything, height).Return(headers[i], dataBlocks[i], nil).Once()
+		m.headerCache.SetDAIncluded(headerHash, uint64(i+1))
+		m.dataCache.SetDAIncluded(dataHash, uint64(i+1))
+		store.On("GetBlockData", mock.Anything, height).Return(headers[i], dataBlocks[i], nil).Times(2) // Called by IsDAIncluded and SetRollkitHeightToDAHeight
 	}
 	// Next height returns error
 	store.On("GetBlockData", mock.Anything, startDAIncludedHeight+uint64(numConsecutive+1)).Return(nil, nil, assert.AnError).Once()
+	// Mock expectations for SetRollkitHeightToDAHeight method calls
+	for i := 0; i < numConsecutive; i++ {
+		height := startDAIncludedHeight + uint64(i+1)
+		headerHeightBytes := make([]byte, 8)
+		binary.LittleEndian.PutUint64(headerHeightBytes, uint64(i+1))
+		store.On("SetMetadata", mock.Anything, fmt.Sprintf("%s/%d/h", RollkitHeightToDAHeightKey, height), headerHeightBytes).Return(nil).Once()
+		dataHeightBytes := make([]byte, 8)
+		binary.LittleEndian.PutUint64(dataHeightBytes, uint64(i+1))
+		store.On("SetMetadata", mock.Anything, fmt.Sprintf("%s/%d/d", RollkitHeightToDAHeightKey, height), dataHeightBytes).Return(nil).Once()
+	}
 	store.On("SetMetadata", mock.Anything, DAIncludedHeightKey, mock.Anything).Return(nil).Times(numConsecutive)
 	exec.On("SetFinal", mock.Anything, mock.Anything).Return(nil).Times(numConsecutive)
 
@@ -312,11 +331,25 @@ func TestDAIncluderLoop_AdvancesHeightWhenDataHashIsEmptyAndHeaderDAIncluded(t *
 
 	header, data := types.GetRandomBlock(5, 0, "testchain")
 	headerHash := header.Hash().String()
-	m.headerCache.SetDAIncluded(headerHash)
+	m.headerCache.SetDAIncluded(headerHash, uint64(1))
 	// Do NOT set data as DA-included
 
-	store.On("GetBlockData", mock.Anything, uint64(5)).Return(header, data, nil).Once()
+	store.On("GetBlockData", mock.Anything, uint64(5)).Return(header, data, nil).Times(2) // Called by IsDAIncluded and SetRollkitHeightToDAHeight
 	store.On("GetBlockData", mock.Anything, uint64(6)).Return(nil, nil, assert.AnError).Once()
+	// Mock expectations for SetRollkitHeightToDAHeight method
+	headerHeightBytes := make([]byte, 8)
+	binary.LittleEndian.PutUint64(headerHeightBytes, uint64(1))
+	store.On("SetMetadata", mock.Anything, fmt.Sprintf("%s/%d/h", RollkitHeightToDAHeightKey, uint64(5)), headerHeightBytes).Return(nil).Once()
+	// Note: For empty data, data SetMetadata call should still be made but data won't be marked as DA-included in cache,
+	// so SetRollkitHeightToDAHeight will fail when trying to get the DA height for data
+	// Actually, let's check if this case is handled differently for empty txs
+	// Let me check what happens with empty txs by adding the data cache entry as well
+	dataHash := data.DACommitment().String()
+	m.dataCache.SetDAIncluded(dataHash, uint64(1))
+	dataHeightBytes := make([]byte, 8)
+	binary.LittleEndian.PutUint64(dataHeightBytes, uint64(1))
+	store.On("SetMetadata", mock.Anything, fmt.Sprintf("%s/%d/d", RollkitHeightToDAHeightKey, uint64(5)), dataHeightBytes).Return(nil).Once()
+	// Mock expectations for incrementDAIncludedHeight method
 	heightBytes := make([]byte, 8)
 	binary.LittleEndian.PutUint64(heightBytes, expectedDAIncludedHeight)
 	store.On("SetMetadata", mock.Anything, DAIncludedHeightKey, heightBytes).Return(nil).Once()

block/manager.go

@@ -55,6 +55,9 @@ const (
 
 	// LastBatchDataKey is the key used for persisting the last batch data in store.
 	LastBatchDataKey = "l"
+
+	// RollkitHeightToDAHeightKey is the key used for persisting the mapping rollkit height to da height in store.
+	RollkitHeightToDAHeightKey = "rhb"
 )
 
 var (
@@ -486,6 +489,45 @@ func (m *Manager) IsDAIncluded(ctx context.Context, height uint64) (bool, error)
 	return isIncluded, nil
 }
 
+// SetRollkitHeightToDAHeight stores the mapping from a Rollkit block height to the corresponding
+// DA (Data Availability) layer heights where the block's header and data were included.
+// This mapping is persisted in the store metadata and is used to track which DA heights
+// contain the block components for a given Rollkit height.
+//
+// For blocks with empty transactions, both header and data use the same DA height since
+// empty transaction data is not actually published to the DA layer.
+func (m *Manager) SetRollkitHeightToDAHeight(ctx context.Context, height uint64) error {
+	header, data, err := m.store.GetBlockData(ctx, height)
+	if err != nil {
+		return err
+	}
+	headerHash, dataHash := header.Hash(), data.DACommitment()
+	headerHeightBytes := make([]byte, 8)
+	daHeightForHeader, ok := m.headerCache.GetDAIncludedHeight(headerHash.String())
+	if !ok {
+		return fmt.Errorf("header hash %s not found in cache", headerHash)
+	}
+	binary.LittleEndian.PutUint64(headerHeightBytes, daHeightForHeader)
+	if err := m.store.SetMetadata(ctx, fmt.Sprintf("%s/%d/h", RollkitHeightToDAHeightKey, height), headerHeightBytes); err != nil {
+		return err
+	}
+	dataHeightBytes := make([]byte, 8)
+	// For empty transactions, use the same DA height as the header
+	if bytes.Equal(dataHash, dataHashForEmptyTxs) {
+		binary.LittleEndian.PutUint64(dataHeightBytes, daHeightForHeader)
+	} else {
+		daHeightForData, ok := m.dataCache.GetDAIncludedHeight(dataHash.String())
+		if !ok {
+			return fmt.Errorf("data hash %s not found in cache", dataHash.String())
+		}
+		binary.LittleEndian.PutUint64(dataHeightBytes, daHeightForData)
+	}
+	if err := m.store.SetMetadata(ctx, fmt.Sprintf("%s/%d/d", RollkitHeightToDAHeightKey, height), dataHeightBytes); err != nil {
+		return err
+	}
+	return nil
+}
+
 // GetExecutor returns the executor used by the manager.
 //
 // Note: this is a temporary method to allow testing the manager.

block/manager_test.go

@@ -194,11 +194,11 @@ func TestIsDAIncluded(t *testing.T) {
 	require.False(m.IsDAIncluded(ctx, height))
 
 	// Set the hash as DAIncluded and verify IsDAIncluded returns true
-	m.headerCache.SetDAIncluded(header.Hash().String())
+	m.headerCache.SetDAIncluded(header.Hash().String(), uint64(1))
 	require.False(m.IsDAIncluded(ctx, height))
 
 	// Set the data as DAIncluded and verify IsDAIncluded returns true
-	m.dataCache.SetDAIncluded(data.DACommitment().String())
+	m.dataCache.SetDAIncluded(data.DACommitment().String(), uint64(1))
 	require.True(m.IsDAIncluded(ctx, height))
 }
 
@@ -1065,3 +1065,138 @@ func TestConfigurationDefaults(t *testing.T) {
 		require.Contains(err.Error(), "insufficient bytes for length prefix")
 	})
 }
+
+// TestSetRollkitHeightToDAHeight tests the SetRollkitHeightToDAHeight method which maps
+// rollkit block heights to their corresponding DA heights for both headers and data.
+// This method is critical for tracking which DA heights contain specific rollkit blocks.
+func TestSetRollkitHeightToDAHeight(t *testing.T) {
+	t.Run("Success_WithTransactions", func(t *testing.T) {
+		require := require.New(t)
+		ctx := context.Background()
+		m, mockStore := getManager(t, nil, 0, 0)
+
+		// Create a block with transactions
+		header, data := types.GetRandomBlock(5, 3, "testchain")
+		height := uint64(5)
+
+		// Mock store expectations
+		mockStore.On("GetBlockData", mock.Anything, height).Return(header, data, nil)
+
+		// Set DA included heights in cache
+		headerHeight := uint64(10)
+		dataHeight := uint64(20)
+		m.headerCache.SetDAIncluded(header.Hash().String(), headerHeight)
+		m.dataCache.SetDAIncluded(data.DACommitment().String(), dataHeight)
+
+		// Mock metadata storage
+		headerKey := fmt.Sprintf("%s/%d/h", RollkitHeightToDAHeightKey, height)
+		dataKey := fmt.Sprintf("%s/%d/d", RollkitHeightToDAHeightKey, height)
+		mockStore.On("SetMetadata", mock.Anything, headerKey, mock.Anything).Return(nil)
+		mockStore.On("SetMetadata", mock.Anything, dataKey, mock.Anything).Return(nil)
+
+		// Call the method
+		err := m.SetRollkitHeightToDAHeight(ctx, height)
+		require.NoError(err)
+
+		mockStore.AssertExpectations(t)
+	})
+
+	t.Run("Success_EmptyTransactions", func(t *testing.T) {
+		require := require.New(t)
+		ctx := context.Background()
+		m, mockStore := getManager(t, nil, 0, 0)
+
+		// Create a block with no transactions
+		header, data := types.GetRandomBlock(5, 0, "testchain")
+		height := uint64(5)
+
+		// Mock store expectations
+		mockStore.On("GetBlockData", mock.Anything, height).Return(header, data, nil)
+
+		// Only set header as DA included (data uses special empty hash)
+		headerHeight := uint64(10)
+		m.headerCache.SetDAIncluded(header.Hash().String(), headerHeight)
+		// Note: we don't set data in cache for empty transactions
+
+		// Mock metadata storage - both should use header height for empty transactions
+		headerKey := fmt.Sprintf("%s/%d/h", RollkitHeightToDAHeightKey, height)
+		dataKey := fmt.Sprintf("%s/%d/d", RollkitHeightToDAHeightKey, height)
+		mockStore.On("SetMetadata", mock.Anything, headerKey, mock.Anything).Return(nil)
+		mockStore.On("SetMetadata", mock.Anything, dataKey, mock.Anything).Return(nil)
+
+		// Call the method
+		err := m.SetRollkitHeightToDAHeight(ctx, height)
+		require.NoError(err)
+
+		mockStore.AssertExpectations(t)
+	})
+
+	t.Run("Error_HeaderNotInCache", func(t *testing.T) {
+		require := require.New(t)
+		ctx := context.Background()
+		m, mockStore := getManager(t, nil, 0, 0)
+
+		// Create a block
+		header, data := types.GetRandomBlock(5, 3, "testchain")
+		height := uint64(5)
+
+		// Mock store expectations
+		mockStore.On("GetBlockData", mock.Anything, height).Return(header, data, nil)
+
+		// Don't set header in cache, but set data
+		m.dataCache.SetDAIncluded(data.DACommitment().String(), uint64(11))
+
+		// Call the method - should fail
+		err := m.SetRollkitHeightToDAHeight(ctx, height)
+		require.Error(err)
+		require.Contains(err.Error(), "header hash")
+		require.Contains(err.Error(), "not found in cache")
+
+		mockStore.AssertExpectations(t)
+	})
+
+	t.Run("Error_DataNotInCache_NonEmptyTxs", func(t *testing.T) {
+		require := require.New(t)
+		ctx := context.Background()
+		m, mockStore := getManager(t, nil, 0, 0)
+
+		// Create a block with transactions
+		header, data := types.GetRandomBlock(5, 3, "testchain")
+		height := uint64(5)
+
+		// Mock store expectations
+		mockStore.On("GetBlockData", mock.Anything, height).Return(header, data, nil)
+
+		// Set header but not data in cache
+		m.headerCache.SetDAIncluded(header.Hash().String(), uint64(10))
+
+		// Mock metadata storage for header (should succeed)
+		headerKey := fmt.Sprintf("%s/%d/h", RollkitHeightToDAHeightKey, height)
+		mockStore.On("SetMetadata", mock.Anything, headerKey, mock.Anything).Return(nil)
+
+		// Call the method - should fail on data lookup
+		err := m.SetRollkitHeightToDAHeight(ctx, height)
+		require.Error(err)
+		require.Contains(err.Error(), "data hash")
+		require.Contains(err.Error(), "not found in cache")
+
+		mockStore.AssertExpectations(t)
+	})
+
+	t.Run("Error_BlockNotFound", func(t *testing.T) {
+		require := require.New(t)
+		ctx := context.Background()
+		m, mockStore := getManager(t, nil, 0, 0)
+
+		height := uint64(999) // Non-existent height
+
+		// Mock store expectations
+		mockStore.On("GetBlockData", mock.Anything, height).Return(nil, nil, errors.New("block not found"))
+
+		// Call the method - should fail
+		err := m.SetRollkitHeightToDAHeight(ctx, height)
+		require.Error(err)
+
+		mockStore.AssertExpectations(t)
+	})
+}

block/retriever.go

@@ -133,7 +133,7 @@ func (m *Manager) handlePotentialHeader(ctx context.Context, bz []byte, daHeight
 		return true
 	}
 	headerHash := header.Hash().String()
-	m.headerCache.SetDAIncluded(headerHash)
+	m.headerCache.SetDAIncluded(headerHash, daHeight)
 	m.sendNonBlockingSignalToDAIncluderCh()
 	m.logger.Info("header marked as DA included", "headerHeight", header.Height(), "headerHash", headerHash)
 	if !m.headerCache.IsSeen(headerHash) {
@@ -168,7 +168,7 @@ func (m *Manager) handlePotentialData(ctx context.Context, bz []byte, daHeight u
 	}
 
 	dataHashStr := signedData.Data.DACommitment().String()
-	m.dataCache.SetDAIncluded(dataHashStr)
+	m.dataCache.SetDAIncluded(dataHashStr, daHeight)
 	m.sendNonBlockingSignalToDAIncluderCh()
 	m.logger.Info("signed data marked as DA included", "dataHash", dataHashStr, "daHeight", daHeight, "height", signedData.Height())
 	if !m.dataCache.IsSeen(dataHashStr) {

block/retriever_test.go

@@ -528,9 +528,9 @@ func TestProcessNextDAHeader_HeaderAndDataAlreadySeen(t *testing.T) {
 
 	// Mark both header and data as seen and DA included
 	headerCache.SetSeen(headerHash)
-	headerCache.SetDAIncluded(headerHash)
+	headerCache.SetDAIncluded(headerHash, uint64(10))
 	dataCache.SetSeen(dataHash)
-	dataCache.SetDAIncluded(dataHash)
+	dataCache.SetDAIncluded(dataHash, uint64(10))
 
 	// Set up mocks with explicit logging
 	mockDAClient.On("GetIDs", mock.Anything, daHeight, mock.Anything).Return(&coreda.GetIDsResult{

block/submitter.go

@@ -172,7 +172,7 @@ func (m *Manager) submitHeadersToDA(ctx context.Context, headersToSubmit []*type
 		},
 		func(submitted []*types.SignedHeader, res *coreda.ResultSubmit, gasPrice float64) {
 			for _, header := range submitted {
-				m.headerCache.SetDAIncluded(header.Hash().String())
+				m.headerCache.SetDAIncluded(header.Hash().String(), res.Height)
 			}
 			lastSubmittedHeaderHeight := uint64(0)
 			if l := len(submitted); l > 0 {
@@ -197,7 +197,7 @@ func (m *Manager) submitDataToDA(ctx context.Context, signedDataToSubmit []*type
 		},
 		func(submitted []*types.SignedData, res *coreda.ResultSubmit, gasPrice float64) {
 			for _, signedData := range submitted {
-				m.dataCache.SetDAIncluded(signedData.DACommitment().String())
+				m.dataCache.SetDAIncluded(signedData.Data.DACommitment().String(), res.Height)
 			}
 			lastSubmittedDataHeight := uint64(0)
 			if l := len(submitted); l > 0 {