Add explicit no block timestamp path

Author: alpe

apps/evm/server/force_inclusion_test.go

@@ -34,6 +34,10 @@ func (m *mockDA) Retrieve(ctx context.Context, height uint64, namespace []byte)
 	return da.ResultRetrieve{}
 }
 
+func (m *mockDA) RetrieveBlobs(ctx context.Context, height uint64, namespace []byte) da.ResultRetrieve {
+	return da.ResultRetrieve{}
+}
+
 func (m *mockDA) RetrieveHeaders(ctx context.Context, height uint64) da.ResultRetrieve {
 	return da.ResultRetrieve{}
 }

block/internal/da/client.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/celestiaorg/go-square/v3/share"
@@ -37,6 +38,7 @@ type client struct {
 	dataNamespaceBz    []byte
 	forcedNamespaceBz  []byte
 	hasForcedNamespace bool
+	timestampCache     *blockTimestampCache
 }
 
 // Ensure client implements the FullClient interface (Client + BlobGetter + Verifier).
@@ -45,8 +47,72 @@ var _ FullClient = (*client)(nil)
 const (
 	blockTimestampFetchMaxAttempts = 3
 	blockTimestampFetchBackoff     = 100 * time.Millisecond
+	blockTimestampCacheWindow      = 2048
 )
 
+type blockTimestampCache struct {
+	mu       sync.RWMutex
+	byHeight map[uint64]time.Time
+	highest  uint64
+	window   uint64
+}
+
+func newBlockTimestampCache(window uint64) *blockTimestampCache {
+	if window == 0 {
+		window = blockTimestampCacheWindow
+	}
+	return &blockTimestampCache{
+		byHeight: make(map[uint64]time.Time),
+		window:   window,
+	}
+}
+
+func (c *blockTimestampCache) get(height uint64) (time.Time, bool) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	blockTime, ok := c.byHeight[height]
+	return blockTime, ok
+}
+
+func (c *blockTimestampCache) put(height uint64, blockTime time.Time) {
+	if c == nil || blockTime.IsZero() {
+		return
+	}
+
+	blockTime = blockTime.UTC()
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	minRetained := c.minRetainedHeightLocked()
+	if minRetained > 0 && height < minRetained {
+		return
+	}
+
+	if height > c.highest {
+		c.highest = height
+	}
+	c.byHeight[height] = blockTime
+
+	minRetained = c.minRetainedHeightLocked()
+	if minRetained == 0 {
+		return
+	}
+	for cachedHeight := range c.byHeight {
+		if cachedHeight < minRetained {
+			delete(c.byHeight, cachedHeight)
+		}
+	}
+}
+
+func (c *blockTimestampCache) minRetainedHeightLocked() uint64 {
+	if c.window == 0 || c.highest < c.window-1 {
+		return 0
+	}
+	return c.highest - c.window + 1
+}
+
 // NewClient creates a new blob client wrapper with pre-calculated namespace bytes.
 func NewClient(cfg Config) FullClient {
 	if cfg.DA == nil {
@@ -71,6 +137,7 @@ func NewClient(cfg Config) FullClient {
 		dataNamespaceBz:    datypes.NamespaceFromString(cfg.DataNamespace).Bytes(),
 		forcedNamespaceBz:  forcedNamespaceBz,
 		hasForcedNamespace: hasForcedNamespace,
+		timestampCache:     newBlockTimestampCache(blockTimestampCacheWindow),
 	}
 }
 
@@ -197,7 +264,9 @@ func (c *client) getBlockTimestamp(ctx context.Context, height uint64) (time.Tim
 		header, err := c.headerAPI.GetByHeight(headerCtx, height)
 		cancel()
 		if err == nil {
-			return header.Time(), nil
+			blockTime := header.Time().UTC()
+			c.storeBlockTimestamp(height, blockTime)
+			return blockTime, nil
 		}
 		lastErr = err
 
@@ -225,10 +294,38 @@ func (c *client) getBlockTimestamp(ctx context.Context, height uint64) (time.Tim
 	return time.Time{}, fmt.Errorf("get header timestamp for block %d after %d attempts: %w", height, blockTimestampFetchMaxAttempts, lastErr)
 }
 
+func (c *client) cachedBlockTimestamp(height uint64) (time.Time, bool) {
+	return c.timestampCache.get(height)
+}
+
+func (c *client) storeBlockTimestamp(height uint64, blockTime time.Time) {
+	c.timestampCache.put(height, blockTime)
+}
+
+func (c *client) resolveBlockTimestamp(ctx context.Context, height uint64, strict bool) (time.Time, error) {
+	if !strict {
+		if blockTime, ok := c.cachedBlockTimestamp(height); ok {
+			return blockTime, nil
+		}
+		return time.Time{}, nil
+	}
+
+	return c.getBlockTimestamp(ctx, height)
+}
+
+// RetrieveBlobs retrieves blobs without blocking on DA header timestamps.
+func (c *client) RetrieveBlobs(ctx context.Context, height uint64, namespace []byte) datypes.ResultRetrieve {
+	return c.retrieve(ctx, height, namespace, false)
+}
+
 // Retrieve retrieves blobs from the DA layer at the specified height and namespace.
 // It uses GetAll to fetch all blobs at once.
 // The timestamp is derived from the DA block header to ensure determinism.
 func (c *client) Retrieve(ctx context.Context, height uint64, namespace []byte) datypes.ResultRetrieve {
+	return c.retrieve(ctx, height, namespace, true)
+}
+
+func (c *client) retrieve(ctx context.Context, height uint64, namespace []byte, strictTimestamp bool) datypes.ResultRetrieve {
 	ns, err := share.NewNamespaceFromBytes(namespace)
 	if err != nil {
 		return datypes.ResultRetrieve{
@@ -250,8 +347,7 @@ func (c *client) Retrieve(ctx context.Context, height uint64, namespace []byte)
 		switch {
 		case strings.Contains(err.Error(), datypes.ErrBlobNotFound.Error()):
 			c.logger.Debug().Uint64("height", height).Msg("No blobs found at height")
-			// Fetch block timestamp for deterministic responses using parent context
-			blockTime, err := c.getBlockTimestamp(ctx, height)
+			blockTime, err := c.resolveBlockTimestamp(ctx, height, strictTimestamp)
 			if err != nil {
 				c.logger.Error().Uint64("height", height).Err(err).Msg("failed to get block timestamp")
 				return datypes.ResultRetrieve{
@@ -293,8 +389,7 @@ func (c *client) Retrieve(ctx context.Context, height uint64, namespace []byte)
 		}
 	}
 
-	// Fetch block timestamp for deterministic responses using parent context
-	blockTime, err := c.getBlockTimestamp(ctx, height)
+	blockTime, err := c.resolveBlockTimestamp(ctx, height, strictTimestamp)
 	if err != nil {
 		c.logger.Error().Uint64("height", height).Err(err).Msg("failed to get block timestamp")
 		return datypes.ResultRetrieve{
@@ -426,14 +521,14 @@ func (c *client) Subscribe(ctx context.Context, namespace []byte, includeTimesta
 				var blockTime time.Time
 				// Use header time if available (celestia-node v0.21.0+)
 				if resp.Header != nil && !resp.Header.Time.IsZero() {
-					blockTime = resp.Header.Time
+					blockTime = resp.Header.Time.UTC()
+					c.storeBlockTimestamp(resp.Height, blockTime)
 				} else if includeTimestamp {
 					// Fallback to fetching timestamp for older nodes
 					blockTime, err = c.getBlockTimestamp(ctx, resp.Height)
 					if err != nil {
 						c.logger.Error().Uint64("height", resp.Height).Err(err).Msg("failed to get DA block timestamp for subscription event")
-						blockTime = time.Now()
-						// TODO: we should retry fetching the timestamp. Current time may mess block time consistency for based sequencers.
+						blockTime = time.Time{}
 					}
 				}
 				select {

block/internal/da/client_test.go

@@ -228,6 +228,133 @@ func TestClient_Retrieve_TimestampFetchRetry(t *testing.T) {
 	}
 }
 
+func TestClient_RetrieveBlobs_TimestampBehavior(t *testing.T) {
+	ns := share.MustNewV0Namespace([]byte("ns"))
+	nsBz := ns.Bytes()
+	fixedTime := time.Date(2024, 1, 3, 12, 0, 0, 0, time.UTC)
+
+	specs := map[string]struct {
+		primeCache      bool
+		getAllErr       error
+		wantStatus      datypes.StatusCode
+		wantTimestamp   time.Time
+		wantHeaderCalls int
+	}{
+		"uncached_skips_header_fetch": {
+			wantStatus:      datypes.StatusSuccess,
+			wantTimestamp:   time.Time{},
+			wantHeaderCalls: 0,
+		},
+		"cached_reuses_timestamp": {
+			primeCache:      true,
+			getAllErr:       datypes.ErrBlobNotFound,
+			wantStatus:      datypes.StatusNotFound,
+			wantTimestamp:   fixedTime,
+			wantHeaderCalls: 1,
+		},
+	}
+
+	for name, spec := range specs {
+		t.Run(name, func(t *testing.T) {
+			blobModule := mocks.NewMockBlobModule(t)
+			headerModule := mocks.NewMockHeaderModule(t)
+
+			payloadBlob, err := blobrpc.NewBlobV0(ns, []byte("payload"))
+			require.NoError(t, err)
+
+			headerCalls := 0
+			headerModule.EXPECT().GetByHeight(mock.Anything, uint64(7)).RunAndReturn(func(context.Context, uint64) (*blobrpc.Header, error) {
+				headerCalls++
+				return &blobrpc.Header{Header: blobrpc.RawHeader{Time: fixedTime}}, nil
+			}).Maybe()
+
+			cl := NewClient(Config{
+				DA:            makeBlobRPCClient(blobModule, headerModule),
+				Logger:        zerolog.Nop(),
+				Namespace:     "ns",
+				DataNamespace: "ns",
+			})
+
+			if spec.primeCache {
+				blobModule.On("GetAll", mock.Anything, uint64(7), mock.Anything).Return([]*blobrpc.Blob{payloadBlob}, nil).Once()
+				res := cl.Retrieve(context.Background(), 7, nsBz)
+				require.Equal(t, datypes.StatusSuccess, res.Code)
+				require.Equal(t, fixedTime, res.Timestamp)
+			}
+
+			if spec.getAllErr != nil {
+				blobModule.On("GetAll", mock.Anything, uint64(7), mock.Anything).Return([]*blobrpc.Blob(nil), spec.getAllErr).Once()
+			} else {
+				blobModule.On("GetAll", mock.Anything, uint64(7), mock.Anything).Return([]*blobrpc.Blob{payloadBlob}, nil).Once()
+			}
+
+			res := cl.RetrieveBlobs(context.Background(), 7, nsBz)
+			require.Equal(t, spec.wantStatus, res.Code)
+			require.Equal(t, spec.wantTimestamp, res.Timestamp)
+			require.Equal(t, spec.wantHeaderCalls, headerCalls)
+		})
+	}
+}
+
+func TestBlockTimestampCache_Bounded(t *testing.T) {
+	cache := newBlockTimestampCache(2)
+	t1 := time.Date(2024, 1, 4, 12, 0, 0, 0, time.UTC)
+	t2 := t1.Add(time.Second)
+	t3 := t2.Add(time.Second)
+
+	cache.put(10, t1)
+	cache.put(11, t2)
+	cache.put(12, t3)
+
+	_, ok := cache.get(10)
+	require.False(t, ok)
+
+	got, ok := cache.get(11)
+	require.True(t, ok)
+	require.Equal(t, t2, got)
+
+	got, ok = cache.get(12)
+	require.True(t, ok)
+	require.Equal(t, t3, got)
+}
+
+func TestClient_Subscribe_PrimesTimestampCache(t *testing.T) {
+	ns := share.MustNewV0Namespace([]byte("ns"))
+	nsBz := ns.Bytes()
+	fixedTime := time.Date(2024, 1, 5, 12, 0, 0, 0, time.UTC)
+
+	blobModule := mocks.NewMockBlobModule(t)
+	subCh := make(chan *blobrpc.SubscriptionResponse, 1)
+	blobModule.On("Subscribe", mock.Anything, mock.Anything).Return((<-chan *blobrpc.SubscriptionResponse)(subCh), nil).Once()
+	blobModule.On("GetAll", mock.Anything, uint64(10), mock.Anything).Return([]*blobrpc.Blob(nil), datypes.ErrBlobNotFound).Once()
+
+	cl := NewClient(Config{
+		DA:            makeBlobRPCClient(blobModule, nil),
+		Logger:        zerolog.Nop(),
+		Namespace:     "ns",
+		DataNamespace: "ns",
+	})
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	events, err := cl.Subscribe(ctx, nsBz, false)
+	require.NoError(t, err)
+
+	subCh <- &blobrpc.SubscriptionResponse{
+		Height: 10,
+		Header: &blobrpc.RawHeader{Time: fixedTime},
+	}
+
+	ev := <-events
+	require.Equal(t, uint64(10), ev.Height)
+	require.Equal(t, fixedTime, ev.Timestamp)
+
+	res := cl.RetrieveBlobs(t.Context(), 10, nsBz)
+	require.Equal(t, datypes.StatusNotFound, res.Code)
+	require.Equal(t, fixedTime, res.Timestamp)
+}
+
 func TestClient_SubmitOptionsMerge(t *testing.T) {
 	ns := share.MustNewV0Namespace([]byte("ns")).Bytes()
 	blobModule := mocks.NewMockBlobModule(t)

block/internal/da/interface.go

@@ -14,6 +14,11 @@ type Client interface {
 	// Retrieve retrieves blobs from the DA layer at the specified height and namespace.
 	Retrieve(ctx context.Context, height uint64, namespace []byte) datypes.ResultRetrieve
 
+	// RetrieveBlobs retrieves blobs from the DA layer at the specified height and namespace
+	// without requiring a DA header timestamp. Callers that need deterministic DA time should
+	// use Retrieve instead.
+	RetrieveBlobs(ctx context.Context, height uint64, namespace []byte) datypes.ResultRetrieve
+
 	// Get retrieves blobs by their IDs. Used for visualization and fetching specific blobs.
 	Get(ctx context.Context, ids []datypes.ID, namespace []byte) ([]datypes.Blob, error)
 

block/internal/da/tracing.go

@@ -67,6 +67,26 @@ func (t *tracedClient) Retrieve(ctx context.Context, height uint64, namespace []
 	return res
 }
 
+func (t *tracedClient) RetrieveBlobs(ctx context.Context, height uint64, namespace []byte) datypes.ResultRetrieve {
+	ctx, span := t.tracer.Start(ctx, "DA.RetrieveBlobs",
+		trace.WithAttributes(
+			attribute.Int("ns.length", len(namespace)),
+			attribute.String("da.namespace", hex.EncodeToString(namespace)),
+		),
+	)
+	defer span.End()
+
+	res := t.inner.RetrieveBlobs(ctx, height, namespace)
+
+	if res.Code != datypes.StatusSuccess && res.Code != datypes.StatusNotFound {
+		span.RecordError(&submitError{msg: res.Message})
+		span.SetStatus(codes.Error, res.Message)
+	} else {
+		span.SetAttributes(attribute.Int("blob.count", len(res.Data)))
+	}
+	return res
+}
+
 func (t *tracedClient) Get(ctx context.Context, ids []datypes.ID, namespace []byte) ([]datypes.Blob, error) {
 	ctx, span := t.tracer.Start(ctx, "DA.Get",
 		trace.WithAttributes(