Skip to content
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Next Next commit
approach 2
  • Loading branch information
randygrok committed Nov 27, 2025
commit 6d4b4041c3420014a82e6f10d57c99e1ca653a59
302 changes: 163 additions & 139 deletions block/internal/da/client.go

Large diffs are not rendered by default.

518 changes: 102 additions & 416 deletions block/internal/da/client_test.go

Large diffs are not rendered by default.

77 changes: 77 additions & 0 deletions block/internal/da/local_blob_api.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
package da

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: can this file be marked as _tests? or a comment at the top saying its for tests


import (
"context"
"fmt"
"sync"

"github.com/celestiaorg/go-square/v3/share"

"github.com/evstack/ev-node/pkg/blob"
)

// LocalBlobAPI is a simple in-memory BlobAPI implementation for tests.
type LocalBlobAPI struct {
mu sync.Mutex
height uint64
maxSize uint64
byHeight map[uint64][]*blob.Blob
}

// NewLocalBlobAPI creates an in-memory BlobAPI with a max blob size.
func NewLocalBlobAPI(maxSize uint64) *LocalBlobAPI {
return &LocalBlobAPI{
maxSize: maxSize,
byHeight: make(map[uint64][]*blob.Blob),
}
}

func (l *LocalBlobAPI) Submit(ctx context.Context, blobs []*blob.Blob, _ *blob.SubmitOptions) (uint64, error) {
l.mu.Lock()
defer l.mu.Unlock()

for i, b := range blobs {
if uint64(len(b.Data())) > l.maxSize {
return 0, fmt.Errorf("blob %d too big", i)
}
}

l.height++
// store clones to avoid external mutation
stored := make([]*blob.Blob, len(blobs))
for i, b := range blobs {
stored[i] = b
}
l.byHeight[l.height] = append(l.byHeight[l.height], stored...)
return l.height, nil
}

func (l *LocalBlobAPI) GetAll(ctx context.Context, height uint64, namespaces []share.Namespace) ([]*blob.Blob, error) {
l.mu.Lock()
defer l.mu.Unlock()

nsMap := make(map[string]struct{}, len(namespaces))
for _, ns := range namespaces {
nsMap[string(ns.Bytes())] = struct{}{}
}

blobs, ok := l.byHeight[height]
if !ok {
return []*blob.Blob{}, nil
}
var out []*blob.Blob
for _, b := range blobs {
if _, ok := nsMap[string(b.Namespace().Bytes())]; ok {
out = append(out, b)
}
}
return out, nil
}

func (l *LocalBlobAPI) GetProof(ctx context.Context, height uint64, namespace share.Namespace, commitment blob.Commitment) (*blob.Proof, error) {
return &blob.Proof{}, nil
}

func (l *LocalBlobAPI) Included(ctx context.Context, height uint64, namespace share.Namespace, proof *blob.Proof, commitment blob.Commitment) (bool, error) {
return true, nil
}
37 changes: 26 additions & 11 deletions block/internal/submitting/da_submitter.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,9 +13,10 @@ import (
"github.com/evstack/ev-node/block/internal/cache"
"github.com/evstack/ev-node/block/internal/common"
"github.com/evstack/ev-node/block/internal/da"
coreda "github.com/evstack/ev-node/core/da"
"github.com/evstack/ev-node/pkg/blob"
"github.com/evstack/ev-node/pkg/config"
pkgda "github.com/evstack/ev-node/pkg/da"
datypes "github.com/evstack/ev-node/pkg/da/types"
"github.com/evstack/ev-node/pkg/genesis"
"github.com/evstack/ev-node/pkg/rpc/server"
"github.com/evstack/ev-node/pkg/signer"
Expand Down Expand Up @@ -119,7 +120,21 @@ func NewDASubmitter(

if config.RPC.EnableDAVisualization {
visualizerLogger := logger.With().Str("component", "da_visualization").Logger()
server.SetDAVisualizationServer(server.NewDAVisualizationServer(client.GetDA(), visualizerLogger, config.Node.Aggregator))
server.SetDAVisualizationServer(
server.NewDAVisualizationServer(
func(ctx context.Context, id []byte, ns []byte) ([]datypes.Blob, error) {
// minimal fetch: derive height from ID and use namespace provided
height, _ := blob.SplitID(id)
res := client.Retrieve(ctx, height, ns)
if res.Code != datypes.StatusSuccess || len(res.Data) == 0 {
return nil, fmt.Errorf("blob not found")
}
return res.Data, nil
},
visualizerLogger,
config.Node.Aggregator,
),
)
Comment on lines +121 to +135

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we modify the server to not need the function wrapper?

}

// Use NoOp metrics if nil to avoid nil checks throughout the code
Expand Down Expand Up @@ -184,7 +199,7 @@ func (s *DASubmitter) SubmitHeaders(ctx context.Context, cache cache.Manager) er
}
return proto.Marshal(headerPb)
},
func(submitted []*types.SignedHeader, res *coreda.ResultSubmit) {
func(submitted []*types.SignedHeader, res *datypes.ResultSubmit) {
for _, header := range submitted {
cache.SetHeaderDAIncluded(header.Hash().String(), res.Height, header.Height())
}
Expand Down Expand Up @@ -227,7 +242,7 @@ func (s *DASubmitter) SubmitData(ctx context.Context, cache cache.Manager, signe
func(signedData *types.SignedData) ([]byte, error) {
return signedData.MarshalBinary()
},
func(submitted []*types.SignedData, res *coreda.ResultSubmit) {
func(submitted []*types.SignedData, res *datypes.ResultSubmit) {
for _, sd := range submitted {
cache.SetDataDAIncluded(sd.Data.DACommitment().String(), res.Height, sd.Height())
}
Expand Down Expand Up @@ -343,7 +358,7 @@ func submitToDA[T any](
ctx context.Context,
items []T,
marshalFn func(T) ([]byte, error),
postSubmit func([]T, *coreda.ResultSubmit),
postSubmit func([]T, *datypes.ResultSubmit),
itemType string,
namespace []byte,
options []byte,
Expand Down Expand Up @@ -406,7 +421,7 @@ func submitToDA[T any](

// Perform submission
start := time.Now()
res := s.client.Submit(submitCtx, marshaled, -1, namespace, mergedOptions)
res := s.client.Submit(submitCtx, marshaled, namespace, mergedOptions)
s.logger.Debug().Int("attempts", rs.Attempt).Dur("elapsed", time.Since(start)).Uint64("code", uint64(res.Code)).Msg("got SubmitWithHelpers response from celestia")

// Record submission result for observability
Expand All @@ -415,7 +430,7 @@ func submitToDA[T any](
}

switch res.Code {
case coreda.StatusSuccess:
case datypes.StatusSuccess:
submitted := items[:res.SubmittedCount]
postSubmit(submitted, &res)
s.logger.Info().Str("itemType", itemType).Uint64("count", res.SubmittedCount).Msg("successfully submitted items to DA layer")
Expand All @@ -436,7 +451,7 @@ func submitToDA[T any](
s.metrics.DASubmitterPendingBlobs.Set(float64(getTotalPendingFn()))
}

case coreda.StatusTooBig:
case datypes.StatusTooBig:
// Record failure metric
s.recordFailure(common.DASubmitterFailureReasonTooBig)
// Iteratively halve until it fits or single-item too big
Expand All @@ -460,19 +475,19 @@ func submitToDA[T any](
s.metrics.DASubmitterPendingBlobs.Set(float64(getTotalPendingFn()))
}

case coreda.StatusNotIncludedInBlock:
case datypes.StatusNotIncludedInBlock:
// Record failure metric
s.recordFailure(common.DASubmitterFailureReasonNotIncludedInBlock)
s.logger.Info().Dur("backoff", pol.MaxBackoff).Msg("retrying due to mempool state")
rs.Next(reasonMempool, pol)

case coreda.StatusAlreadyInMempool:
case datypes.StatusAlreadyInMempool:
// Record failure metric
s.recordFailure(common.DASubmitterFailureReasonAlreadyInMempool)
s.logger.Info().Dur("backoff", pol.MaxBackoff).Msg("retrying due to mempool state")
rs.Next(reasonMempool, pol)

case coreda.StatusContextCanceled:
case datypes.StatusContextCanceled:
// Record failure metric
s.recordFailure(common.DASubmitterFailureReasonContextCanceled)
s.logger.Info().Msg("DA layer submission canceled due to context cancellation")
Expand Down
18 changes: 9 additions & 9 deletions block/internal/submitting/da_submitter_integration_test.go
Original file line number Diff line number Diff line change
@@ -1,3 +1,6 @@
//go:build !ignore
// +build !ignore

package submitting

import (
Expand All @@ -16,7 +19,6 @@ import (
"github.com/evstack/ev-node/block/internal/cache"
"github.com/evstack/ev-node/block/internal/common"
"github.com/evstack/ev-node/block/internal/da"
coreda "github.com/evstack/ev-node/core/da"
"github.com/evstack/ev-node/pkg/config"
"github.com/evstack/ev-node/pkg/genesis"
"github.com/evstack/ev-node/pkg/signer/noop"
Expand Down Expand Up @@ -83,15 +85,13 @@ func TestDASubmitter_SubmitHeadersAndData_MarksInclusionAndUpdatesLastSubmitted(
require.NoError(t, batch2.SetHeight(2))
require.NoError(t, batch2.Commit())

// Dummy DA
dummyDA := coreda.NewDummyDA(10_000_000, 10*time.Millisecond)

// Create DA submitter
// Create DA submitter with local blob API
daClient := da.NewClient(da.Config{
DA: dummyDA,
Logger: zerolog.Nop(),
Namespace: cfg.DA.Namespace,
DataNamespace: cfg.DA.DataNamespace,
BlobAPI: da.NewLocalBlobAPI(common.DefaultMaxBlobSize),
Logger: zerolog.Nop(),
Namespace: cfg.DA.Namespace,
DataNamespace: cfg.DA.DataNamespace,
DefaultTimeout: 10 * time.Second,
})
daSubmitter := NewDASubmitter(daClient, cfg, gen, common.DefaultBlockOptions(), common.NopMetrics(), zerolog.Nop())

Expand Down
Loading