fix(verifier): match header blobs wrapped in DAHeaderEnvelope

README.md

@@ -135,7 +135,7 @@ For an authenticated DA-proxy endpoint, pass its token with `--balance.token`:
 
 The following metrics are exposed:
 
-Blocks without transactions have only their header blob verified (`type="header"`). Blocks with transactions have both their header blob (`type="header"`) and data blob (`type="data"`) verified, and each outcome is recorded under its own type. Tx blocks now count under `type="header"` too, so header submission counts (`submission_*`, `unsubmitted_*`, DA height and duration) rise after upgrading. Each transaction block costs one extra Celestia request for its header blob. If retries are exhausted, only the blob types that are still unverified are recorded as missing.
+Blocks without transactions have only their header blob verified (`type="header"`). Blocks with transactions have both their header blob (`type="header"`) and data blob (`type="data"`) verified, and each outcome is recorded under its own type. Tx blocks now count under `type="header"` too, so header submission counts (`submission_*`, `unsubmitted_*`, DA height and duration) rise after upgrading. Each transaction block costs one extra Celestia request for its header blob. If retries are exhausted, only the blob types that are still unverified are recorded as missing. The header blob is matched by its signed-header content: ev-node publishes it inside a `DAHeaderEnvelope` carrying an extra envelope signature, which is ignored when comparing against the stored block. Because "missing" means a block could not be matched on Celestia after retries, not that a submission transaction failed, the `unsubmitted_*` and failure counts reflect verification outcomes rather than the sequencer's own pending-blob backlog.
 
 ### `ev_metrics_unsubmitted_block_range_start`
 - **Type**: Gauge

internal/clients/celestia/client.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -163,116 +164,66 @@ func (c *Client) fetchBlobs(ctx context.Context, daHeight uint64, ns libshare.Na
 	return result, nil
 }
 
-// VerifyBlobAtHeight verifies that a specific blob exists at the given DA height
-// by computing its commitment and fetching it directly from Celestia.
-func (c *Client) VerifyBlobAtHeight(ctx context.Context, blob []byte, daHeight uint64, namespace []byte) (bool, error) {
-	if len(blob) == 0 {
-		return true, nil // empty blobs are valid (nothing to verify)
-	}
-
-	ns, err := libshare.NewNamespaceFromBytes(namespace)
-	if err != nil {
-		return false, fmt.Errorf("invalid namespace: %w", err)
-	}
-
-	b, err := jsonrpc.NewBlobV0(ns, blob)
-	if err != nil {
-		return false, fmt.Errorf("failed to create blob for commitment: %w", err)
-	}
-
-	c.logger.Debug().
-		Str("commitment", fmt.Sprintf("%x", b.Commitment)).
-		Int("blob_size", len(blob)).
-		Uint64("da_height", daHeight).
-		Msg("fetching blob by commitment from Celestia")
-
-	result, err := c.Blob.Get(ctx, daHeight, ns, b.Commitment)
-	if err != nil {
-		if strings.Contains(err.Error(), "blob: not found") {
-			c.logger.Debug().Uint64("da_height", daHeight).Msg("blob not found at DA height")
-			return false, nil
+// VerifyHeaderBlobAtHeight verifies a header blob, accounting for the DAHeaderEnvelope
+// wrapper. ev-node submits the signed header plus an envelope signature (field 4) whose
+// bytes cannot be recomputed, so the published blob is matched by its signed-header
+// content instead of by commitment. Legacy blobs without an envelope also match.
+func (c *Client) VerifyHeaderBlobAtHeight(ctx context.Context, signedHeaderBlob []byte, daHeight uint64, namespace []byte) (bool, error) {
+	return c.verifyBlobAtHeight(ctx, "header", signedHeaderBlob, daHeight, namespace, func(blob []byte) ([]byte, error) {
+		var sh evnode.SignedHeader
+		if err := proto.Unmarshal(blob, &sh); err != nil {
+			return nil, err
 		}
-		return false, fmt.Errorf("failed to get blob: %w", err)
-	}
-
-	if result != nil {
-		c.logger.Debug().
-			Str("commitment", fmt.Sprintf("%x", b.Commitment)).
-			Msg("blob verified on Celestia")
-	}
-
-	return result != nil, nil
+		// drop the envelope signature (field 4, unknown to SignedHeader) but keep unknown
+		// fields nested inside the header, which are part of the signed bytes.
+		sh.ProtoReflect().SetUnknown(nil)
+		return proto.Marshal(&sh)
+	})
 }
 
 // VerifyDataBlobAtHeight verifies a data blob, accounting for the SignedData wrapper
 // ev-node submits Data wrapped in SignedData, but the Store API returns unwrapped Data
 func (c *Client) VerifyDataBlobAtHeight(ctx context.Context, unwrappedDataBlob []byte, daHeight uint64, namespace []byte) (bool, error) {
-	if len(unwrappedDataBlob) == 0 {
+	return c.verifyBlobAtHeight(ctx, "data", unwrappedDataBlob, daHeight, namespace, func(blob []byte) ([]byte, error) {
+		var signedData evnode.SignedData
+		if err := proto.Unmarshal(blob, &signedData); err != nil {
+			return nil, err
+		}
+		if signedData.Data == nil {
+			return nil, errors.New("SignedData has nil Data")
+		}
+		return proto.Marshal(signedData.Data)
+	})
+}
+
+// verifyBlobAtHeight reports whether any blob in the namespace at daHeight, reduced
+// by unwrap to the form the Store API returns, equals want.
+func (c *Client) verifyBlobAtHeight(ctx context.Context, kind string, want []byte, daHeight uint64, namespace []byte, unwrap func([]byte) ([]byte, error)) (bool, error) {
+	if len(want) == 0 {
 		return true, nil
 	}
 
-	c.logger.Debug().
-		Int("unwrapped_size", len(unwrappedDataBlob)).
-		Uint64("da_height", daHeight).
-		Msg("verifying data blob (will check wrapped SignedData on Celestia)")
-
-	// get all blobs at the DA height
 	blobs, err := c.GetBlobsAtHeight(ctx, daHeight, namespace)
 	if err != nil {
 		return false, fmt.Errorf("failed to get blobs: %w", err)
 	}
 
-	if len(blobs) == 0 {
-		c.logger.Debug().Uint64("da_height", daHeight).Msg("no blobs found")
-		return false, nil
-	}
-
-	// try to unwrap each blob as SignedData and compare the inner Data
 	for i, blob := range blobs {
-		var signedData evnode.SignedData
-		if err := proto.Unmarshal(blob, &signedData); err != nil {
-			c.logger.Debug().
-				Int("blob_index", i).
-				Err(err).
-				Msg("failed to unmarshal as SignedData, skipping")
-			continue
-		}
-
-		if signedData.Data == nil {
-			c.logger.Debug().Int("blob_index", i).Msg("SignedData has nil Data, skipping")
-			continue
-		}
-
-		// marshal the inner Data to compare
-		celestiaData, err := proto.Marshal(signedData.Data)
+		got, err := unwrap(blob)
 		if err != nil {
-			c.logger.Debug().
-				Int("blob_index", i).
-				Err(err).
-				Msg("failed to marshal inner Data")
+			c.logger.Debug().Int("blob_index", i).Str("kind", kind).Err(err).Msg("failed to unwrap blob, skipping")
 			continue
 		}
-
-		equal := bytes.Equal(celestiaData, unwrappedDataBlob)
-
-		c.logger.Debug().
-			Int("blob_index", i).
-			Int("celestia_data_size", len(celestiaData)).
-			Int("evnode_data_size", len(unwrappedDataBlob)).
-			Bool("matches", equal).
-			Msg("comparing unwrapped Data")
-
-		if equal {
-			c.logger.Debug().
-				Int("blob_index", i).
-				Msg("found matching Data")
+		if bytes.Equal(got, want) {
 			return true, nil
 		}
 	}
 
 	c.logger.Warn().
+		Str("kind", kind).
+		Uint64("da_height", daHeight).
 		Int("checked_blobs", len(blobs)).
-		Msg("no matching Data found in any SignedData wrapper")
+		Msg("no matching blob found")
 	return false, nil
 }
 

internal/clients/celestia/client_test.go

@@ -12,9 +12,11 @@ import (
 
 	libshare "github.com/celestiaorg/go-square/v3/share"
 	"github.com/evstack/ev-node/pkg/da/jsonrpc"
+	evnode "github.com/evstack/ev-node/types/pb/evnode/v1"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
 )
 
 func TestGetBlockTimestamp_AuthorizationHeader(t *testing.T) {
@@ -214,3 +216,57 @@ func TestGetBlobsAtHeight_Cache(t *testing.T) {
 		})
 	}
 }
+
+func TestVerifyHeaderBlobAtHeight(t *testing.T) {
+	ns := libshare.MustNewV0Namespace([]byte("ev-metrics"))
+	signed := &evnode.SignedHeader{
+		Header:    &evnode.Header{Height: 7, ChainId: "test"},
+		Signature: []byte("sig"),
+		Signer:    &evnode.Signer{Address: []byte("addr"), PubKey: []byte("pub")},
+	}
+	stored, err := proto.Marshal(signed)
+	require.NoError(t, err)
+	envelope, err := proto.Marshal(&evnode.DAHeaderEnvelope{
+		Header: signed.Header, Signature: signed.Signature, Signer: signed.Signer,
+		EnvelopeSignature: []byte("envelope-sig"),
+	})
+	require.NoError(t, err)
+	require.NotEqual(t, stored, envelope)
+
+	other, err := proto.Marshal(&evnode.DAHeaderEnvelope{Header: &evnode.Header{Height: 8}, EnvelopeSignature: []byte("x")})
+	require.NoError(t, err)
+
+	newClient := func(published ...[]byte) *Client {
+		return &Client{
+			logger:   zerolog.Nop(),
+			cacheTTL: time.Minute,
+			getAll: func(context.Context, uint64, []libshare.Namespace) ([]*jsonrpc.Blob, error) {
+				var blobs []*jsonrpc.Blob
+				for _, p := range published {
+					b, err := jsonrpc.NewBlobV0(ns, p)
+					require.NoError(t, err)
+					blobs = append(blobs, b)
+				}
+				return blobs, nil
+			},
+		}
+	}
+
+	tests := map[string]struct {
+		published [][]byte
+		want      bool
+	}{
+		"envelope blob":          {[][]byte{other, envelope}, true},
+		"legacy signed header":   {[][]byte{stored}, true},
+		"different header":       {[][]byte{other}, false},
+		"nothing at this height": {nil, false},
+		"undecodable blob":       {[][]byte{{0xff, 0xff}, envelope}, true},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			got, err := newClient(tt.published...).VerifyHeaderBlobAtHeight(context.Background(), stored, 5, ns.Bytes())
+			require.NoError(t, err)
+			require.Equal(t, tt.want, got)
+		})
+	}
+}

pkg/exporters/verifier/verifier.go

@@ -21,7 +21,7 @@ type blockSource interface {
 
 type daVerifier interface {
 	GetBlockTimestamp(ctx context.Context, daHeight uint64) (time.Time, error)
-	VerifyBlobAtHeight(ctx context.Context, blob []byte, daHeight uint64, namespace []byte) (bool, error)
+	VerifyHeaderBlobAtHeight(ctx context.Context, signedHeaderBlob []byte, daHeight uint64, namespace []byte) (bool, error)
 	VerifyDataBlobAtHeight(ctx context.Context, unwrappedDataBlob []byte, daHeight uint64, namespace []byte) (bool, error)
 }
 
@@ -413,7 +413,7 @@ func (e *exporter) verifyAttempt(ctx context.Context, m *metrics.Metrics, logger
 
 	if !j.headerDone {
 		j.headerDone = verifyPart("header", blockResultWithBlobs.HeaderDaHeight, func(daHeight uint64) (bool, error) {
-			return e.celestiaClient.VerifyBlobAtHeight(attemptCtx, blockResultWithBlobs.HeaderBlob, daHeight, e.headerNS)
+			return e.celestiaClient.VerifyHeaderBlobAtHeight(attemptCtx, blockResultWithBlobs.HeaderBlob, daHeight, e.headerNS)
 		})
 	}
 

pkg/exporters/verifier/verifier_test.go

@@ -60,7 +60,7 @@ func (f *fakeDA) GetBlockTimestamp(context.Context, uint64) (time.Time, error) {
 	return time.Now(), nil
 }
 
-func (f *fakeDA) VerifyBlobAtHeight(_ context.Context, _ []byte, daHeight uint64, _ []byte) (bool, error) {
+func (f *fakeDA) VerifyHeaderBlobAtHeight(_ context.Context, _ []byte, daHeight uint64, _ []byte) (bool, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	f.verified = append(f.verified, daHeight-daOffset)