Skip to content

Separate signed header and data to different blobs#858

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

Closed
gupadhyaya wants to merge 18 commits into from
Closed

Separate signed header and data to different blobs #858

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
c69e412
remove unused namespace id in the executor
Apr 10, 2023
fca31f1
update config to accept two namespace ids
Apr 10, 2023
1ba0ca0
extend da clients to support processing of headers and data separately
Apr 12, 2023
c8d0caa
return data hash and assign to header before signing
Apr 12, 2023
c9ea875
minor fixes and optimizations
Apr 12, 2023
82b0872
fixing da tests and mock da
Apr 12, 2023
4d1a233
fixing mock datastore inserts and queries
Apr 13, 2023
cf68e39
minor fix to TestReceive
Apr 13, 2023
d7e7f5b
adding block validation for data hash
Apr 13, 2023
a2a6fc8
fix golangci-lint
Apr 13, 2023
1d72554
fix datas typo
Apr 19, 2023
814652e
remove code duplication, extract to function
Apr 24, 2023
716966e
minor fixes
Apr 26, 2023
d08e4d6
remove hash dependency, separate validate & apply, populate ISRs befo…
May 1, 2023
d1e1fd1
go mod tidy
May 1, 2023
d99852c
enforcing different header and data namespace ids for mock
May 1, 2023
6ed20a2
fix failing test by making sure that the header and data namespace i…
May 2, 2023
47ce34b
fix test failure due to same empty header and data namespace ids
May 4, 2023
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
108 changes: 67 additions & 41 deletions block/manager.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -120,7 +120,7 @@ func NewManager(
conf.DABlockTime = defaultDABlockTime
}

exec := state.NewBlockExecutor(proposerAddress, conf.NamespaceID, genesis.ChainID, mempool, proxyApp, conf.FraudProofs, eventBus, logger)
exec := state.NewBlockExecutor(proposerAddress, genesis.ChainID, mempool, proxyApp, conf.FraudProofs, eventBus, logger)
if s.LastBlockHeight+1 == genesis.InitialHeight {
res, err := exec.InitChain(genesis)
if err != nil {
Expand DownExpand Up@@ -326,6 +326,10 @@ func (m *Manager) trySyncNextBlock(ctx context.Context, daHeight uint64) error {

if b != nil && commit != nil {
m.logger.Info("Syncing block", "height", b.SignedHeader.Header.Height())
// Validate the received block before applying
if err := m.executor.Validate(m.lastState, b); err != nil {
return fmt.Errorf("failed to validate block: %w", err)
}
newState, responses, err := m.executor.ApplyBlock(ctx, m.lastState, b)
if err != nil {
return fmt.Errorf("failed to ApplyBlock: %w", err)
Expand DownExpand Up@@ -412,33 +416,43 @@ func (m *Manager) processNextDABlock(ctx context.Context) error {
var err error
m.logger.Debug("trying to retrieve block from DA", "daHeight", daHeight)
for r := 0; r < maxRetries; r++ {
blockResp, fetchErr := m.fetchBlock(ctx, daHeight)
headerResp, dataResp, fetchErr := m.fetchBlock(ctx, daHeight)
if fetchErr != nil {
err = multierr.Append(err, fetchErr)
time.Sleep(100 * time.Millisecond)
} else {
if blockResp.Code == da.StatusNotFound {
m.logger.Debug("no block found", "daHeight", daHeight, "reason", blockResp.Message)
if headerResp.Code == da.StatusNotFound {
m.logger.Debug("no block header found", "daHeight", daHeight, "reason", headerResp.Message)
return nil
}
if dataResp.Code == da.StatusNotFound {
m.logger.Debug("no block data found", "daHeight", daHeight, "reason", dataResp.Message)
return nil
}
m.logger.Debug("retrieved potential blocks", "n", len(blockResp.Blocks), "daHeight", daHeight)
for _, block := range blockResp.Blocks {
m.blockInCh <- newBlockEvent{block, daHeight}
m.logger.Debug("retrieved potential blocks", "n", len(headerResp.Headers), "daHeight", daHeight)
// TODO (fix this): assumes both block header and data exists in the same da height
// Fix should cache the headers and data and then match to construct the block and send to channel
for i, header := range headerResp.Headers {
m.blockInCh <- newBlockEvent{&types.Block{SignedHeader: *header, Data: *dataResp.Data[i]}, daHeight}
}
return nil
}
}
return err
}

func (m *Manager) fetchBlock(ctx context.Context, daHeight uint64) (da.ResultRetrieveBlocks, error) {
func (m *Manager) fetchBlock(ctx context.Context, daHeight uint64) (da.ResultRetrieveBlockHeaders, da.ResultRetrieveBlockData, error) {
var err error
blockRes := m.retriever.RetrieveBlocks(ctx, daHeight)
switch blockRes.Code {
case da.StatusError:
err = fmt.Errorf("failed to retrieve block: %s", blockRes.Message)
headerRes := m.retriever.RetrieveBlockHeaders(ctx, daHeight)
if headerRes.Code == da.StatusError {
err = fmt.Errorf("failed to retrieve block: %s", headerRes.Message)
}

dataRes := m.retriever.RetrieveBlockData(ctx, daHeight)
if dataRes.Code == da.StatusError {
err = multierr.Append(err, fmt.Errorf("failed to retrieve block: %s", dataRes.Message))
}
return blockRes, err
return headerRes, dataRes, err
}

func (m *Manager) getRemainingSleep(start time.Time) time.Duration {
Expand DownExpand Up@@ -523,22 +537,6 @@ func (m *Manager) publishBlock(ctx context.Context) error {
m.logger.Info("Creating and publishing block", "height", newHeight)
block = m.executor.CreateBlock(newHeight, lastCommit, lastHeaderHash, m.lastState)
m.logger.Debug("block info", "num_tx", len(block.Data.Txs))

commit, err = m.getCommit(block.SignedHeader.Header)
if err != nil {
return err
}

// set the commit to current block's signed header
block.SignedHeader.Commit = *commit

block.SignedHeader.Validators = m.lastState.Validators

// SaveBlock commits the DB tx
err = m.store.SaveBlock(block, commit)
if err != nil {
return err
}
}

// Apply the block but DONT commit
Expand All@@ -547,11 +545,24 @@ func (m *Manager) publishBlock(ctx context.Context) error {
return err
}

if commit == nil {
commit, err = m.getCommit(block.SignedHeader.Header)
if err != nil {
return err
}
// Before taking the hash, we need updated ISRs, hence after ApplyBlock
block.SignedHeader.Header.DataHash, err = block.Data.Hash()
if err != nil {
return err
}

// Sign the block and set the commit to current block's signed header along with signers
commit, err = m.getCommit(block.SignedHeader.Header)
if err != nil {
return err
}

block.SignedHeader.Commit = *commit
block.SignedHeader.Validators = m.lastState.Validators

// Validate the created block before storing
if err := m.executor.Validate(m.lastState, block); err != nil {
return fmt.Errorf("failed to validate block: %w", err)
}

// SaveBlock commits the DB tx
Expand All@@ -560,8 +571,7 @@ func (m *Manager) publishBlock(ctx context.Context) error {
return err
}

err = m.submitBlockToDA(ctx, block)
if err != nil {
if err := m.submitBlockToDA(ctx, block); err != nil {
m.logger.Error("Failed to submit block to DA Layer")
return err
}
Expand DownExpand Up@@ -613,19 +623,35 @@ func (m *Manager) submitBlockToDA(ctx context.Context, block *types.Block) error
submitted := false
backoff := initialBackoff
for attempt := 1; ctx.Err() == nil && !submitted && attempt <= maxSubmitAttempts; attempt++ {
res := m.dalc.SubmitBlock(ctx, block)
if res.Code == da.StatusSuccess {
m.logger.Info("successfully submitted Rollkit block to DA layer", "rollkitHeight", block.SignedHeader.Header.Height(), "daHeight", res.DAHeight)
headerRes := m.dalc.SubmitBlockHeader(ctx, &block.SignedHeader)
dataRes := m.dalc.SubmitBlockData(ctx, &block.Data)
if headerRes.Code == da.StatusSuccess && dataRes.Code == da.StatusSuccess {
m.logger.Info(
"successfully submitted Rollkit block to DA layer",
"rollkitHeight",
block.SignedHeader.Header.Height(),
"daHeight of the block header",
headerRes.DAHeight,
"daHeight of the block data",
dataRes.DAHeight,
)
submitted = true
} else {
m.logger.Error("DA layer submission failed", "error", res.Message, "attempt", attempt)
var errMsg string
if headerRes.Code == da.StatusError {
errMsg = headerRes.Message
}
if dataRes.Code == da.StatusError {
errMsg += "," + dataRes.Message
}
m.logger.Error("DA layer submission failed", "error", errMsg, "attempt", attempt)
time.Sleep(backoff)
backoff = m.exponentialBackoff(backoff)
}
}

if !submitted {
return fmt.Errorf("Failed to submit block to DA layer after %d attempts", maxSubmitAttempts)
return fmt.Errorf("failed to submit block to DA layer after %d attempts", maxSubmitAttempts)
}

return nil
Expand Down
7 changes: 4 additions & 3 deletions block/manager_test.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -71,8 +71,9 @@ func TestInitialState(t *testing.T) {

key, _, _ := crypto.GenerateEd25519Key(rand.Reader)
conf := config.BlockManagerConfig{
BlockTime: 10 * time.Second,
NamespaceID: types.NamespaceID{1, 2, 3, 4, 5, 6, 7, 8},
BlockTime: 10 * time.Second,
HeaderNamespaceID: types.NamespaceID{1, 2, 3, 4, 5, 6, 7, 8},
DataNamespaceID: types.NamespaceID{8, 7, 6, 5, 4, 3, 2, 1},
}

for _, c := range cases {
Expand All@@ -93,7 +94,7 @@ func TestInitialState(t *testing.T) {

func getMockDALC(logger log.Logger) da.DataAvailabilityLayerClient {
dalc := &mockda.DataAvailabilityLayerClient{}
_ = dalc.Init([8]byte{}, nil, nil, logger)
_ = dalc.Init([8]byte{1, 2, 3, 4, 5, 6, 7, 8}, [8]byte{8, 7, 6, 5, 4, 3, 2, 1}, nil, nil, logger)
_ = dalc.Start()
return dalc
}
Expand Down
56 changes: 37 additions & 19 deletions config/config.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -2,6 +2,7 @@ package config

import (
"encoding/hex"
"fmt"
"time"

"github.com/spf13/cobra"
Expand All@@ -11,17 +12,18 @@ import (
)

const (
flagAggregator = "rollkit.aggregator"
flagDALayer = "rollkit.da_layer"
flagDAConfig = "rollkit.da_config"
flagBlockTime = "rollkit.block_time"
flagDABlockTime = "rollkit.da_block_time"
flagDAStartHeight = "rollkit.da_start_height"
flagNamespaceID = "rollkit.namespace_id"
flagFraudProofs = "rollkit.experimental_insecure_fraud_proofs"
flagLight = "rollkit.light"
flagTrustedHash = "rollkit.trusted_hash"
flagLazyAggregator = "rollkit.lazy_aggregator"
flagAggregator = "rollkit.aggregator"
flagDALayer = "rollkit.da_layer"
flagDAConfig = "rollkit.da_config"
flagBlockTime = "rollkit.block_time"
flagDABlockTime = "rollkit.da_block_time"
flagDAStartHeight = "rollkit.da_start_height"
flagHeaderNamespaceID = "rollkit.header_namespace_id"
flagDataNamespaceID = "rollkit.data_namespace_id"
flagFraudProofs = "rollkit.experimental_insecure_fraud_proofs"
flagLight = "rollkit.light"
flagTrustedHash = "rollkit.trusted_hash"
flagLazyAggregator = "rollkit.lazy_aggregator"
)

// NodeConfig stores Rollkit node configuration.
Expand DownExpand Up@@ -53,9 +55,10 @@ type BlockManagerConfig struct {
// DABlockTime informs about block time of underlying data availability layer
DABlockTime time.Duration `mapstructure:"da_block_time"`
// DAStartHeight allows skipping first DAStartHeight-1 blocks when querying for blocks.
DAStartHeight uint64 `mapstructure:"da_start_height"`
NamespaceID types.NamespaceID `mapstructure:"namespace_id"`
FraudProofs bool `mapstructure:"fraud_proofs"`
DAStartHeight uint64 `mapstructure:"da_start_height"`
HeaderNamespaceID types.NamespaceID `mapstructure:"header_namespace_id"`
DataNamespaceID types.NamespaceID `mapstructure:"data_namespace_id"`
FraudProofs bool `mapstructure:"fraud_proofs"`
}

// GetViperConfig reads configuration parameters from Viper instance.
Expand All@@ -69,14 +72,14 @@ func (nc *NodeConfig) GetViperConfig(v *viper.Viper) error {
nc.DABlockTime = v.GetDuration(flagDABlockTime)
nc.BlockTime = v.GetDuration(flagBlockTime)
nc.LazyAggregator = v.GetBool(flagLazyAggregator)
nsID := v.GetString(flagNamespaceID)
nc.FraudProofs = v.GetBool(flagFraudProofs)
nc.Light = v.GetBool(flagLight)
bytes, err := hex.DecodeString(nsID)
if err != nil {
if err := getHexBytes(v, flagHeaderNamespaceID, nc.HeaderNamespaceID[:]); err != nil {
return err
}
if err := getHexBytes(v, flagDataNamespaceID, nc.DataNamespaceID[:]); err != nil {
return err
}
copy(nc.NamespaceID[:], bytes)
nc.TrustedHash = v.GetString(flagTrustedHash)
return nil
}
Expand All@@ -93,8 +96,23 @@ func AddFlags(cmd *cobra.Command) {
cmd.Flags().Duration(flagBlockTime, def.BlockTime, "block time (for aggregator mode)")
cmd.Flags().Duration(flagDABlockTime, def.DABlockTime, "DA chain block time (for syncing)")
cmd.Flags().Uint64(flagDAStartHeight, def.DAStartHeight, "starting DA block height (for syncing)")
cmd.Flags().BytesHex(flagNamespaceID, def.NamespaceID[:], "namespace identifies (8 bytes in hex)")
cmd.Flags().BytesHex(flagHeaderNamespaceID, def.HeaderNamespaceID[:], "header namespace identifier (8 bytes in hex)")
cmd.Flags().BytesHex(flagDataNamespaceID, def.DataNamespaceID[:], "data namespace identifier (8 bytes in hex)")
cmd.Flags().Bool(flagFraudProofs, def.FraudProofs, "enable fraud proofs (experimental & insecure)")
cmd.Flags().Bool(flagLight, def.Light, "run light client")
cmd.Flags().String(flagTrustedHash, def.TrustedHash, "initial trusted hash to start the header exchange service")
}

func getHexBytes(v *viper.Viper, flag string, dst []byte) error {
str := v.GetString(flag)
bytes, err := hex.DecodeString(str)
if err != nil {
return err
}
if len(bytes) != len(types.NamespaceID{}) {
return fmt.Errorf("invalid length of namespace ID for '%s', expected: %d, got: %d",
flag, len(types.NamespaceID{}), len(bytes))
Comment thread
gupadhyaya marked this conversation as resolved.
}
_ = copy(dst, bytes)
return nil
}
6 changes: 4 additions & 2 deletions config/config_test.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -25,7 +25,8 @@ func TestViperAndCobra(t *testing.T) {
assert.NoError(cmd.Flags().Set(flagDALayer, "foobar"))
assert.NoError(cmd.Flags().Set(flagDAConfig, `{"json":true}`))
assert.NoError(cmd.Flags().Set(flagBlockTime, "1234s"))
assert.NoError(cmd.Flags().Set(flagNamespaceID, "0102030405060708"))
assert.NoError(cmd.Flags().Set(flagHeaderNamespaceID, "0102030405060708"))
assert.NoError(cmd.Flags().Set(flagDataNamespaceID, "0807060504030201"))
assert.NoError(cmd.Flags().Set(flagFraudProofs, "false"))

nc := DefaultNodeConfig
Expand All@@ -35,6 +36,7 @@ func TestViperAndCobra(t *testing.T) {
assert.Equal("foobar", nc.DALayer)
assert.Equal(`{"json":true}`, nc.DAConfig)
assert.Equal(1234*time.Second, nc.BlockTime)
assert.Equal(types.NamespaceID{1, 2, 3, 4, 5, 6, 7, 8}, nc.NamespaceID)
assert.Equal(types.NamespaceID{1, 2, 3, 4, 5, 6, 7, 8}, nc.HeaderNamespaceID)
assert.Equal(types.NamespaceID{8, 7, 6, 5, 4, 3, 2, 1}, nc.DataNamespaceID)
assert.Equal(false, nc.FraudProofs)
}
7 changes: 4 additions & 3 deletions config/defaults.go
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -21,9 +21,10 @@ var DefaultNodeConfig = NodeConfig{
Aggregator: false,
LazyAggregator: false,
BlockManagerConfig: BlockManagerConfig{
BlockTime: 30 * time.Second,
NamespaceID: types.NamespaceID{},
FraudProofs: false,
BlockTime: 30 * time.Second,

@nashqueuenashqueueApr 25, 2023

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.

If we run a mock, do we want to speed up the block time to reduce test times? How does this affect the test speeds in general?

HeaderNamespaceID: types.NamespaceID{},
DataNamespaceID: types.NamespaceID{},
FraudProofs: false,
},
DALayer: "mock",
DAConfig: "",
Expand Down
Loading