Skip to content

chore: remove DAClient #2221

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tac0turtle merged 15 commits into from
May 5, 2025
Merged

chore: remove DAClient #2221

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
11e6856
remove dac
Apr 26, 2025
f5ef691
fix build
Apr 26, 2025
4ee676c
flatten da
Apr 26, 2025
5c2492c
fix block tests
Apr 26, 2025
5c9f02f
remove namespace being passed to sequecner as its already in DA
Apr 27, 2025
726f10f
revert based change and remove namespace from being passed
Apr 28, 2025
e28a7a5
fix single test
Apr 28, 2025
4fbad7d
address some coderabbit issues
Apr 28, 2025
454299e
lift placeholder and add todo
Apr 28, 2025
2cb673b
Merge branch 'main' into marko/remove_dac
Apr 29, 2025
d23ece0
fix tests
Apr 29, 2025
36845dc
add some better error handling
Apr 29, 2025
4985e5a
Merge branch 'main' into marko/remove_dac
tac0turtle May 2, 2025
14ee9d2
fix da test
May 5, 2025
8ae19ce
remove deadcode
May 5, 2025
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
70 changes: 18 additions & 52 deletions block/manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import (
"github.com/rollkit/rollkit/pkg/genesis"
"github.com/rollkit/rollkit/pkg/signer"
"github.com/rollkit/rollkit/pkg/store"
"github.com/rollkit/rollkit/types"
"github.com/rollkit/rollkit/types" // Import the types package where helpers now reside
pb "github.com/rollkit/rollkit/types/pb/rollkit/v1"
)

Expand All @@ -45,10 +45,6 @@ const (
// defaultMempoolTTL is the number of blocks until transaction is dropped from mempool
defaultMempoolTTL = 25

// blockProtocolOverhead is the protocol overhead when marshaling the block to blob
// see: https://gist.github.com/tuxcanfly/80892dde9cdbe89bfb57a6cb3c27bae2
blockProtocolOverhead = 1 << 16

// maxSubmitAttempts defines how many times Rollkit will re-try to publish block to DA layer.
// This is temporary solution. It will be removed in future versions.
maxSubmitAttempts = 30
Expand Down Expand Up @@ -147,7 +143,7 @@ type Manager struct {
// daIncludedHeight is rollup height at which all blocks have been included
// in the DA
daIncludedHeight atomic.Uint64
dalc coreda.Client
da coreda.DA
gasPrice float64
gasMultiplier float64

Expand Down Expand Up @@ -250,7 +246,7 @@ func NewManager(
store store.Store,
exec coreexecutor.Executor,
sequencer coresequencer.Sequencer,
dalc coreda.Client,
da coreda.DA,
logger log.Logger,
headerStore *goheaderstore.Store[*types.SignedHeader],
dataStore *goheaderstore.Store[*types.Data],
Expand Down Expand Up @@ -293,16 +289,6 @@ func NewManager(
config.DA.MempoolTTL = defaultMempoolTTL
}

//proposerAddress := s.Validators.Proposer.Address.Bytes()

maxBlobSize, err := dalc.MaxBlobSize(ctx)
if err != nil {
return nil, err
}
// allow buffer for the block header and protocol encoding
//nolint:ineffassign // This assignment is needed
maxBlobSize -= blockProtocolOverhead

isProposer, err := isProposer(proposerKey, pubkey)
if err != nil {
logger.Error("error while checking if proposer", "error", err)
Expand Down Expand Up @@ -331,7 +317,6 @@ func NewManager(
genesis: genesis,
lastState: s,
store: store,
dalc: dalc,
daHeight: s.DAHeight,
// channels are buffered to avoid blocking on input/output operations, buffer sizes are arbitrary
HeaderCh: make(chan *types.SignedHeader, channelLength),
Expand All @@ -354,18 +339,14 @@ func NewManager(
isProposer: isProposer,
sequencer: sequencer,
exec: exec,
da: da,
gasPrice: gasPrice,
gasMultiplier: gasMultiplier,
}
agg.init(ctx)
return agg, nil
}

// DALCInitialized returns true if DALC is initialized.
func (m *Manager) DALCInitialized() bool {
return m.dalc != nil
}

// PendingHeaders returns the pending headers.
func (m *Manager) PendingHeaders() *PendingHeaders {
return m.pendingHeaders
Expand Down Expand Up @@ -414,11 +395,6 @@ func (m *Manager) GetDAIncludedHeight() uint64 {
return m.daIncludedHeight.Load()
}

// SetDALC is used to set DataAvailabilityLayerClient used by Manager.
func (m *Manager) SetDALC(dalc coreda.Client) {
m.dalc = dalc
}

// isProposer returns whether or not the manager is a proposer
func isProposer(signer signer.Signer, pubkey crypto.PubKey) (bool, error) {
if signer == nil {
Expand Down Expand Up @@ -1086,14 +1062,12 @@ func (m *Manager) areAllErrorsHeightFromFuture(err error) bool {
}

func (m *Manager) fetchHeaders(ctx context.Context, daHeight uint64) (coreda.ResultRetrieve, error) {
res := types.RetrieveWithHelpers(ctx, m.da, m.logger, daHeight)
var err error
ctx, cancel := context.WithTimeout(ctx, 60*time.Second) //TODO: make this configurable
defer cancel()
headerRes := m.dalc.Retrieve(ctx, daHeight)
if headerRes.Code == coreda.StatusError {
err = fmt.Errorf("failed to retrieve block: %s", headerRes.Message)
if res.Code == coreda.StatusError {
err = fmt.Errorf("failed to retrieve block: %s", res.Message)
}
return headerRes, err
return res, err
}

func (m *Manager) getSignature(header types.Header) (types.Signature, error) {
Expand Down Expand Up @@ -1342,11 +1316,6 @@ func (m *Manager) submitHeadersToDA(ctx context.Context) error {
}
numSubmittedHeaders := 0
attempt := 0
maxBlobSize, err := m.dalc.MaxBlobSize(ctx)
if err != nil {
return err
}
initialMaxBlobSize := maxBlobSize
gasPrice := m.gasPrice
initialGasPrice := gasPrice

Expand All @@ -1371,10 +1340,9 @@ daSubmitRetryLoop:
return fmt.Errorf("failed to marshal header: %w", err)
}
}

ctx, cancel := context.WithTimeout(ctx, 60*time.Second) //TODO: make this configurable
defer cancel()
res := m.dalc.Submit(ctx, headersBz, maxBlobSize, gasPrice)
submitCtx, cancel := context.WithTimeout(ctx, 60*time.Second)
res := types.SubmitWithHelpers(submitCtx, m.da, m.logger, headersBz, gasPrice, nil)
cancel()
switch res.Code {
case coreda.StatusSuccess:
m.logger.Info("successfully submitted Rollkit headers to DA layer", "gasPrice", gasPrice, "daHeight", res.Height, "headerCount", res.SubmittedCount)
Expand All @@ -1396,28 +1364,26 @@ daSubmitRetryLoop:
}
m.pendingHeaders.setLastSubmittedHeight(ctx, lastSubmittedHeight)
headersToSubmit = notSubmittedBlocks
// reset submission options when successful
// scale back gasPrice gradually
backoff = 0
maxBlobSize = initialMaxBlobSize
if m.gasMultiplier > 0 && gasPrice != -1 {
gasPrice = gasPrice / m.gasMultiplier
if gasPrice < initialGasPrice {
gasPrice = initialGasPrice
}
}
m.logger.Debug("resetting DA layer submission options", "backoff", backoff, "gasPrice", gasPrice, "maxBlobSize", maxBlobSize)
m.logger.Debug("resetting DA layer submission options", "backoff", backoff, "gasPrice", gasPrice)
case coreda.StatusNotIncludedInBlock, coreda.StatusAlreadyInMempool:
m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt)
backoff = m.config.DA.BlockTime.Duration * time.Duration(m.config.DA.MempoolTTL) //nolint:gosec
backoff = m.config.DA.BlockTime.Duration * time.Duration(m.config.DA.MempoolTTL)
if m.gasMultiplier > 0 && gasPrice != -1 {
gasPrice = gasPrice * m.gasMultiplier
}
m.logger.Info("retrying DA layer submission with", "backoff", backoff, "gasPrice", gasPrice, "maxBlobSize", maxBlobSize)
m.logger.Info("retrying DA layer submission with", "backoff", backoff, "gasPrice", gasPrice)

case coreda.StatusTooBig:
maxBlobSize = maxBlobSize / 4
fallthrough
// StatusTooBig is now handled within the helper/DA implementation
// case coreda.StatusTooBig:
// maxBlobSize = maxBlobSize / 4
// fallthrough
default:
m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt)
backoff = m.exponentialBackoff(backoff)
Expand Down
29 changes: 12 additions & 17 deletions block/manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,10 +41,10 @@ func WithinDuration(t *testing.T, expected, actual, tolerance time.Duration) boo
}

// Returns a minimalistic block manager using a mock DA Client
func getManager(t *testing.T, dac da.Client, gasPrice float64, gasMultiplier float64) *Manager {
func getManager(t *testing.T, da da.DA, gasPrice float64, gasMultiplier float64) *Manager {
logger := log.NewTestLogger(t)
return &Manager{
dalc: dac,
da: da,
headerCache: cache.NewCache[types.SignedHeader](),
logger: logger,
gasPrice: gasPrice,
Expand Down Expand Up @@ -188,8 +188,8 @@ func TestInitialStateUnexpectedHigherGenesis(t *testing.T) {

func TestSignVerifySignature(t *testing.T) {
require := require.New(t)
mockDAC := mocks.NewClient(t) // Use mock DA Client
m := getManager(t, mockDAC, -1, -1) // Pass mock DA Client
mockDAC := mocks.NewDA(t)
m := getManager(t, mockDAC, -1, -1)
payload := []byte("test")
privKey, _, err := crypto.GenerateKeyPair(crypto.Ed25519, 256)
require.NoError(err)
Expand Down Expand Up @@ -239,8 +239,8 @@ func Test_submitBlocksToDA_BlockMarshalErrorCase1(t *testing.T) {
require := require.New(t)
ctx := context.Background()

mockDA := mocks.NewClient(t) // Use mock DA Client
m := getManager(t, mockDA, -1, -1) // Pass mock DA Client
mockDA := mocks.NewDA(t)
m := getManager(t, mockDA, -1, -1)

header1, data1 := types.GetRandomBlock(uint64(1), 5, chainID)
header2, data2 := types.GetRandomBlock(uint64(2), 5, chainID)
Expand All @@ -254,8 +254,6 @@ func Test_submitBlocksToDA_BlockMarshalErrorCase1(t *testing.T) {
store.On("GetBlockData", mock.Anything, uint64(3)).Return(header3, data3, nil)
store.On("Height", mock.Anything).Return(uint64(3), nil)

mockDA.On("MaxBlobSize", mock.Anything).Return(uint64(100000), nil).Once()

m.store = store

var err error
Expand All @@ -276,15 +274,15 @@ func Test_submitBlocksToDA_BlockMarshalErrorCase2(t *testing.T) {
require := require.New(t)
ctx := context.Background()

mockDA := mocks.NewClient(t) // Use mock DA Client
m := getManager(t, mockDA, -1, -1) // Pass mock DA Client
mockDA := mocks.NewDA(t)
m := getManager(t, mockDA, -1, -1)

header1, data1 := types.GetRandomBlock(uint64(1), 5, chainID)
header2, data2 := types.GetRandomBlock(uint64(2), 5, chainID)
header3, data3 := types.GetRandomBlock(uint64(3), 5, chainID)

store := mocks.NewStore(t)
invalidateBlockHeader(header3) // C fails marshal
invalidateBlockHeader(header3)
store.On("GetMetadata", mock.Anything, LastSubmittedHeightKey).Return(nil, ds.ErrNotFound)
store.On("GetBlockData", mock.Anything, uint64(1)).Return(header1, data1, nil)
store.On("GetBlockData", mock.Anything, uint64(2)).Return(header2, data2, nil)
Expand All @@ -293,14 +291,11 @@ func Test_submitBlocksToDA_BlockMarshalErrorCase2(t *testing.T) {

m.store = store

// Expect MaxBlobSize call (likely happens before marshalling loop)
mockDA.On("MaxBlobSize", mock.Anything).Return(uint64(100000), nil).Once()

var err error
m.pendingHeaders, err = NewPendingHeaders(store, m.logger)
require.NoError(err)
err = m.submitHeadersToDA(ctx)
assert.ErrorContains(err, "failed to transform header to proto") // Error from C is expected
assert.ErrorContains(err, "failed to transform header to proto")
blocks, err := m.pendingHeaders.getPendingHeaders(ctx)
assert.NoError(err)
// Expect all blocks to remain pending because the batch submission was halted
Expand Down Expand Up @@ -376,8 +371,8 @@ func Test_isProposer(t *testing.T) {

func Test_publishBlock_ManagerNotProposer(t *testing.T) {
require := require.New(t)
mockDA := mocks.NewClient(t) // Use mock DA Client
m := getManager(t, mockDA, -1, -1) // Pass mock DA Client
mockDA := mocks.NewDA(t)
m := getManager(t, mockDA, -1, -1)
m.isProposer = false
err := m.publishBlock(context.Background())
require.ErrorIs(err, ErrNotProposer)
Expand Down
98 changes: 0 additions & 98 deletions core/da/client.go
View file
Open in desktop

This file was deleted.

Loading