merging new features to main

chunk_manager/chunk_manager.go

@@ -97,10 +97,14 @@ func NewChunkManager(cfg *ChunkManagerConfig) (*ChunkManager, error) {
 		return nil, errors.New("provided state tob nonce greater than highest settled tob nonce")
 	}
 
-	if cfg.ExpirationTime < 0 || cfg.GCInterval < 0 || cfg.LayerSubmissionCheckInterval < 0 {
+	if cfg.ExpirationTime < 0 || cfg.GCInterval < 0 || cfg.LayerSubmissionCheckInterval <= 0 {
 		return nil, fmt.Errorf("invalid duration variables, ExpirationTime: %s, GCInterval: %s, LayerSubmissionCheckInterval: %s", cfg.ExpirationTime, cfg.GCInterval, cfg.LayerSubmissionCheckInterval)
 	}
 
+	if cfg.SEQClient == nil {
+		return nil, errors.New("SEQClient is required")
+	}
+
 	// when cfg.StateLowestToBNonce =0, after pushing the first element to both PQs, we add a place holder layer as the first layer
 	// the tob nonce of which is headTobNonce+1
 	// when cfg.StateLowestToBNonce >0, chunks will be replayed with the base headToBNonce to be cfg.StateLowestToBNonce-1, hence guarantees

common/test_utils.go

@@ -250,12 +250,45 @@ func DisplayEthTxs(txs map[string]ethtypes.Transactions) {
 	for domain, domainTxs := range txs {
 		fmt.Printf("domain: %s\n", domain)
 		for _, tx := range domainTxs {
-			fmt.Printf("tx: %s\n", tx.Hash().Hex())
+			sender, err := ExtractSender(tx)
+			if err != nil {
+				panic(err)
+			}
+			fmt.Printf("sender: %s tx: %s:%d\n", sender, tx.Hash().Hex(), tx.Nonce())
 		}
 	}
 	fmt.Printf("========txs info end=======\n")
 }
 
+func FindTxHash(txLhs *ethtypes.Transaction, txs []*ethtypes.Transaction) bool {
+	if txLhs == nil {
+		panic("txLhs is nil")
+	}
+	lhsHash := txLhs.Hash()
+	for _, tx := range txs {
+		rhsHash := tx.Hash()
+		if lhsHash == rhsHash {
+			return true
+		}
+	}
+	return false
+}
+
+func TxsHashUnorderedMatch(txsLhs []*ethtypes.Transaction, txsRhs []*ethtypes.Transaction) bool {
+	if len(txsLhs) != len(txsRhs) {
+		return false
+	}
+
+	for _, tx := range txsLhs {
+		if !FindTxHash(tx, txsRhs) {
+			fmt.Printf("could not find tx hash " + tx.Hash().Hex())
+			return false
+		}
+	}
+
+	return true
+}
+
 func SyncMapLen(m *sync.Map) int {
 	var length int
 	m.Range(func(_, _ interface{}) bool {
@@ -264,3 +297,11 @@ func SyncMapLen(m *sync.Map) int {
 	})
 	return length
 }
+
+func CollectChunksFromRequests(reqs ...*SubmitNewBlockRequest) []*ArcadiaChunk {
+	chunks := make([]*ArcadiaChunk, len(reqs))
+	for i, req := range reqs {
+		chunks[i] = &req.Chunk
+	}
+	return chunks
+}

common/types.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	srpc "github.com/AnomalyFi/nodekit-seq/rpc"
+	"github.com/rollkit/go-da"
 
 	hactions "github.com/AnomalyFi/hypersdk/actions"
 	"github.com/gorilla/websocket"
@@ -205,10 +206,12 @@ type ToBChunk struct {
 	txHash2BundleHash map[string]string
 	revertingTxHashes map[string]struct{}
 
+	// refers to whether bundle at idx has been removed
 	removedBitSet *bitset.BitSet
-	domains       map[string]struct{} // chain ids or rollup ids
-	seqTxs        []*chain.Transaction
-	initialized   bool
+
+	domains     map[string]struct{} // chain ids or rollup ids
+	seqTxs      []*chain.Transaction
+	initialized bool
 
 	l sync.RWMutex
 }
@@ -229,6 +232,17 @@ func (tob *ToBChunk) GetBundles() []*CrossRollupBundle {
 	return ret
 }
 
+func (tob *ToBChunk) IsBundleIdxFiltered(idx int) bool {
+	tob.l.RLock()
+	defer tob.l.RUnlock()
+
+	if idx >= len(tob.Bundles) {
+		return false
+	}
+
+	return tob.removedBitSet.Test(uint(idx))
+}
+
 func (tob *ToBChunk) GetTxs() map[string]ethtypes.Transactions {
 	tob.l.RLock()
 	defer tob.l.RUnlock()
@@ -344,10 +358,45 @@ func (tob *ToBChunk) removeBundleContainTx(txHash string) (*CrossRollupBundle, e
 	bundleIdx2remove := slices.IndexFunc(tob.Bundles, func(crb *CrossRollupBundle) bool {
 		return crb.BundleHash == bundleHash
 	})
+
 	// mark as removed
 	tob.removedBitSet = tob.removedBitSet.Set(uint(bundleIdx2remove))
 
 	// re-populate [tob.txs]
+	tob.repopulateToBTxs()
+
+	return tob.Bundles[bundleIdx2remove], nil
+}
+
+func (tob *ToBChunk) FilterBundleWithHash(bundleHash string) (*CrossRollupBundle, error) {
+	tob.l.Lock()
+	defer tob.l.Unlock()
+
+	var foundIdx int
+	var foundBundle *CrossRollupBundle
+
+	for bundleIdx, bundle := range tob.Bundles {
+		if bundle.BundleHash == bundleHash {
+			foundIdx = bundleIdx
+			foundBundle = bundle
+			break
+		}
+	}
+
+	if foundBundle == nil {
+		return nil, fmt.Errorf("filterBundleWithHash found no bundle hash [%s]", bundleHash)
+	}
+
+	// mark as removed
+	tob.removedBitSet = tob.removedBitSet.Set(uint(foundIdx))
+
+	// re-populate [tob.txs]
+	tob.repopulateToBTxs()
+
+	return tob.Bundles[foundIdx], nil
+}
+
+func (tob *ToBChunk) repopulateToBTxs() {
 	tob.txs = make(map[string]ethtypes.Transactions)
 	for bundleIdx, bundle := range tob.Bundles {
 		// continue as this bundle was removed
@@ -360,13 +409,12 @@ func (tob *ToBChunk) removeBundleContainTx(txHash string) (*CrossRollupBundle, e
 			tob.txs[domain] = l
 		}
 	}
+
 	// track domains that contain txs in it
 	tob.domains = make(map[string]struct{})
 	for domain := range tob.txs {
 		tob.domains[domain] = struct{}{}
 	}
-
-	return tob.Bundles[bundleIdx2remove], nil
 }
 
 // LowestBlockNumber return the tracked lowest heights for domains, this prevents the situation that
@@ -527,7 +575,7 @@ type RoBChunk struct {
 	BlockNumber uint64   `json:"block_number"`
 
 	// following fields will be populated after initialization
-	removedBitSet *bitset.BitSet
+	removedBitSet *bitset.BitSet // refers to whether tx within txs at idx has been removed
 	initialized   bool
 	txs           ethtypes.Transactions
 	seqTxs        []*chain.Transaction
@@ -1541,6 +1589,10 @@ func (b *CrossRollupBundle) Domains() []string {
 	return maps.Keys(b.txs)
 }
 
+func (b *CrossRollupBundle) HasTxs() bool {
+	return len(b.txs) > 0
+}
+
 func (b *CrossRollupBundle) ContainTx(txHash string) bool {
 	for _, txs := range b.txs {
 		contain := slices.ContainsFunc(txs, func(t *ethtypes.Transaction) bool {
@@ -1645,3 +1697,13 @@ type CertInfo struct {
 	Cert        []byte
 	PlaceHolder bool
 }
+
+type DACertificate = []byte
+
+type DAPayload = []byte
+
+type BlobInfo struct {
+	BlobID      []DACertificate
+	Proof       []da.Proof
+	IsFinalized bool
+}

database/mockdb.go

@@ -29,7 +29,7 @@ type MockDB struct {
 	RoBChunkMap                       map[string]map[uint64]*common.RoBChunkDB
 	RoBChunkAcceptedMap               map[string]*common.RoBChunkAcceptedDB
 	LastFetchedBlockNum               common.LastFetchedBlockNumberDB
-	PayloadResp                       common.PayloadDB
+	PayloadResp                       *common.PayloadDB
 	Epoch                             uint64
 	EpochLowestToBNonce               map[uint64]uint64
 	PayloadTxsToB                     map[string]*common.PayloadTxs
@@ -54,7 +54,6 @@ func NewMockDB() *MockDB {
 		RoBChunkMap:               make(map[string]map[uint64]*common.RoBChunkDB),
 		RoBChunkAcceptedMap:       make(map[string]*common.RoBChunkAcceptedDB),
 		LastFetchedBlockNum:       common.LastFetchedBlockNumberDB{},
-		PayloadResp:               common.PayloadDB{},
 		Epoch:                     0,
 		EpochLowestToBNonce:       make(map[uint64]uint64),
 		PayloadTxsToB:             make(map[string]*common.PayloadTxs),
@@ -247,8 +246,10 @@ func (db *MockDB) RemoveBestAuctionBid(epoch uint64) error {
 func (db *MockDB) GetPayloadResp(chainID string, blockNumber uint64) (*common.PayloadDB, error) {
 	db.l.Lock()
 	defer db.l.Unlock()
-
-	return &db.PayloadResp, nil
+	if db.PayloadResp == nil {
+		return nil, nil
+	}
+	return db.PayloadResp, nil
 }
 
 func (db *MockDB) SetPayloadResp(chainID string, blockNumber uint64, txs *common.GetPayloadResponse) error {
@@ -260,7 +261,7 @@ func (db *MockDB) SetPayloadResp(chainID string, blockNumber uint64, txs *common
 		return err
 	}
 
-	db.PayloadResp = *payloadTxs
+	db.PayloadResp = payloadTxs
 	return nil
 }
 

datalayer/da_submitter.go

@@ -21,6 +21,7 @@ const (
 
 type IDASubmitter interface {
 	ChunksChan() chan *common.ArcadiaToSEQChunkMessage
+	SubmitAndFinalizeBlob(ctx context.Context, payload common.DAPayload) (*common.BlobInfo, error)
 }
 
 type DASubmitterOpts struct {
@@ -34,16 +35,6 @@ type DASubmitterOpts struct {
 	Mocking          bool
 }
 
-type DACertificate = []byte
-
-type DAPayload = []byte
-
-type BlobInfo struct {
-	BlobID      []DACertificate
-	Proof       []da.Proof
-	IsFinalized bool
-}
-
 var _ IDASubmitter = (*DASubmitter)(nil)
 
 // CelestiaConfig TODO: check fields below
@@ -120,7 +111,7 @@ func NewDASubmitter(opts DASubmitterOpts) (*DASubmitter, error) {
 					chunkCancel()
 					continue
 				}
-				cert, err := submitter.submitAndFinalizeBlob(chunkCtx, blob)
+				cert, err := submitter.SubmitAndFinalizeBlob(chunkCtx, blob)
 				if err != nil {
 					log.WithError(err).Error("unable to submit blob to DA")
 					chunkCancel()
@@ -152,14 +143,14 @@ func NewDASubmitter(opts DASubmitterOpts) (*DASubmitter, error) {
 	return submitter, nil
 }
 
-// TODO: submission and finalization check should be separated in the future
 // SubmitAndFinalizeBlob Submit does submission of the given payload to the data availability layer.
 // It will retry maxRetries number of times with increasing backoff until it succeeds.
 // On success, returns the DA certificate as []byte.
 // On failure, returns nil byte slice and error
-func (da *DASubmitter) submitAndFinalizeBlob(ctx context.Context, payload DAPayload) (*BlobInfo, error) {
+// TODO: submission and finalization check should be separated in the future
+func (da *DASubmitter) SubmitAndFinalizeBlob(ctx context.Context, payload common.DAPayload) (*common.BlobInfo, error) {
 	var numRetries int
-	var certificateDA DACertificate
+	var certificateDA common.DACertificate
 
 	for {
 		numRetries++
@@ -203,7 +194,7 @@ func (da *DASubmitter) submitAndFinalizeBlob(ctx context.Context, payload DAPayl
 	}
 	da.log.Infof("Retrieved proofs for blobs: %v", proofs)
 
-	cert := &BlobInfo{
+	cert := &common.BlobInfo{
 		BlobID:      [][]byte{certificateDA},
 		Proof:       proofs,
 		IsFinalized: true,

datalayer/da_submitter_test.go

@@ -62,7 +62,7 @@ func TestDAClientCanPushCertToChunkManager(t *testing.T) {
 	mockDA.EXPECT().Submit(mock.Anything, [][]byte{blob}, celestiaConfig.GasPrice, celestiaConfig.NamespaceID).Return([][]byte{testBlobID}, nil)
 	mockDA.EXPECT().GetProofs(mock.Anything, [][]byte{testBlobID}, celestiaConfig.NamespaceID).Return([][]byte{testProof}, nil)
 
-	expectedCert := BlobInfo{
+	expectedCert := common.BlobInfo{
 		BlobID:      [][]byte{testBlobID},
 		Proof:       [][]byte{testProof},
 		IsFinalized: true,