Replication response size cap

adapters.go

@@ -234,7 +234,7 @@ type blockDeserializer struct {
 }
 
 func (bp *blockDeserializer) DeserializeBlock(ctx context.Context, bytes []byte) (common.Block, error) {
-	var rawBlock RawBlock
+	var rawBlock metadata.RawBlock
 	if err := rawBlock.UnmarshalCanoto(bytes); err != nil {
 		return nil, err
 	}

common/api.go

@@ -119,6 +119,9 @@ type VerifiedBlock interface {
 	// Bytes returns a byte encoding of the block
 	Bytes() ([]byte, error)
 
+	//Size returns the number of the bytes encoding of the block
+	Size() int
+
 	// SealingBlockInfo returns a non-nil value for a block that is not a sealing block and that is not the first ever simplex block.
 	SealingBlockInfo() *SealingBlockInfo
 }

common/metadata.go

@@ -77,6 +77,9 @@ func (bh *BlockHeader) Bytes() []byte {
 
 	return buff
 }
+func (bh *BlockHeader) Size() int {
+	return BlockHeaderLen
+}
 
 func (bh *BlockHeader) FromBytes(buff []byte) error {
 	if len(buff) != BlockHeaderLen {

common/msg.go

@@ -56,15 +56,17 @@ type ToBeSignedEmptyVote struct {
 	EmptyVoteMetadata
 }
 
+const emptyVoteLen = 1 + 8 + 8 // Version + Epoch + Round
+
 func (v *ToBeSignedEmptyVote) Bytes() []byte {
-	bytes := make([]byte, 1+8+8) // Version + Epoch + Round
+	bytes := make([]byte, emptyVoteLen)
 	binary.BigEndian.PutUint64(bytes[1:9], v.EmptyVoteMetadata.Epoch)
 	binary.BigEndian.PutUint64(bytes[9:17], v.EmptyVoteMetadata.Round)
 	return bytes
 }
 
 func (v *ToBeSignedEmptyVote) FromBytes(buff []byte) error {
-	if len(buff) != 17 {
+	if len(buff) != emptyVoteLen {
 		return fmt.Errorf("invalid buffer length, expected 17, got %d", len(buff))
 	}
 
@@ -78,6 +80,10 @@ func (v *ToBeSignedEmptyVote) FromBytes(buff []byte) error {
 	return nil
 }
 
+func (v *ToBeSignedEmptyVote) Size() int {
+	return emptyVoteLen
+}
+
 func (v *ToBeSignedEmptyVote) Sign(signer Signer) ([]byte, error) {
 	context := "ToBeSignedEmptyVote"
 	msg := v.Bytes()
@@ -198,6 +204,10 @@ func (f *Finalization) Verify(nodes Nodes) error {
 	return verifyContextQC(f.QC, f.Finalization.Bytes(), context, nodes)
 }
 
+func (f *Finalization) Size() int {
+	return f.Finalization.Size() + f.QC.Size()
+}
+
 // Notarization represents a block that has reached a quorum of votes.
 type Notarization struct {
 	Vote ToBeSignedVote
@@ -209,6 +219,10 @@ func (n *Notarization) Verify(nodes Nodes) error {
 	return verifyContextQC(n.QC, n.Vote.Bytes(), context, nodes)
 }
 
+func (n *Notarization) Size() int {
+	return n.Vote.Size() + n.QC.Size()
+}
+
 type BlockMessage struct {
 	Block Block
 	Vote  Vote
@@ -228,6 +242,9 @@ func (en *EmptyNotarization) Verify(nodes Nodes) error {
 	context := "ToBeSignedEmptyVote"
 	return verifyContextQC(en.QC, en.Vote.Bytes(), context, nodes)
 }
+func (en *EmptyNotarization) Size() int {
+	return en.Vote.Size() + en.QC.Size()
+}
 
 type SignedMessage struct {
 	Payload []byte
@@ -243,6 +260,8 @@ type QuorumCertificate interface {
 	Verify(msg []byte, nodes Nodes) error
 	// Bytes returns a raw representation of the given QuorumCertificate.
 	Bytes() []byte
+	// Size returns the number of bytes
+	Size() int
 }
 
 type ReplicationRequest struct {
@@ -374,6 +393,24 @@ func (q *VerifiedQuorumRound) GetRound() uint64 {
 	return 0
 }
 
+func (q *VerifiedQuorumRound) Size() int {
+	var size int
+
+	if q.VerifiedBlock != nil {
+		size += q.VerifiedBlock.Size()
+	}
+	if q.Notarization != nil {
+		size += q.Notarization.Size()
+	}
+	if q.Finalization != nil {
+		size += q.Finalization.Size()
+	}
+	if q.EmptyNotarization != nil {
+		size += q.EmptyNotarization.Size()
+	}
+	return size
+}
+
 type VerifiedFinalizedBlock struct {
 	VerifiedBlock VerifiedBlock
 	Finalization  Finalization

common/msg_test.go

@@ -207,3 +207,38 @@ func TestQuorumRoundMalformed(t *testing.T) {
 	}
 
 }
+
+func TestSizeMatchesBytes(t *testing.T) {
+	qc := testutil.TestQC{
+		{Signer: common.NodeID{1}, Value: []byte("signature1")},
+		{Signer: common.NodeID{2}, Value: []byte("signature2")},
+		{Signer: common.NodeID{3}, Value: []byte("signature3")},
+	}
+
+	bh := common.BlockHeader{
+		ProtocolMetadata: common.ProtocolMetadata{
+			Version: 1,
+			Epoch:   2,
+			Round:   3,
+			Seq:     4,
+			Prev:    common.Digest{3},
+		},
+		Digest: common.Digest{6},
+	}
+	require.Equal(t, len(bh.Bytes()), bh.Size())
+
+	emptyVote := common.ToBeSignedEmptyVote{
+		EmptyVoteMetadata: common.EmptyVoteMetadata{Round: 7, Epoch: 8},
+	}
+	require.Equal(t, len(emptyVote.Bytes()), emptyVote.Size())
+
+	notarization := common.Notarization{Vote: common.ToBeSignedVote{BlockHeader: bh}, QC: qc}
+	require.Equal(t, len(notarization.Vote.Bytes())+len(notarization.QC.Bytes()), notarization.Size())
+
+	finalization := common.Finalization{Finalization: common.ToBeSignedFinalization{BlockHeader: bh}, QC: qc}
+	require.Equal(t, len(finalization.Finalization.Bytes())+len(finalization.QC.Bytes()), finalization.Size())
+
+	emptyNotarization := common.EmptyNotarization{Vote: emptyVote, QC: qc}
+	require.Equal(t, len(emptyNotarization.Vote.Bytes())+len(emptyNotarization.QC.Bytes()), emptyNotarization.Size())
+
+}

external.go

@@ -5,37 +5,19 @@ package simplex
 
 import (
 	"context"
+	"sync"
 
 	"github.com/ava-labs/simplex/common"
 	metadata "github.com/ava-labs/simplex/msm"
 )
 
-type RawBlock struct {
-	Metadata        metadata.StateMachineMetadata `canoto:"value,1"`
-	InnerBlockBytes []byte                        `canoto:"bytes,2"`
-
-	canotoData canotoData_RawBlock
-}
-
 type ParsedBlock struct {
 	metadata.StateMachineBlock
 	msm *metadata.StateMachine
-}
-
-func (p *ParsedBlock) Bytes() ([]byte, error) {
-	var innerBlockBytes []byte
-	if p.InnerBlock != nil {
-		rawInnerBlock, err := p.InnerBlock.Bytes()
-		if err != nil {
-			return nil, err
-		}
-		innerBlockBytes = rawInnerBlock
-	}
-	rawBlock := &RawBlock{
-		Metadata:        p.Metadata,
-		InnerBlockBytes: innerBlockBytes,
-	}
-	return rawBlock.MarshalCanoto(), nil
+	// sizeLock guards size, so Size() can be invoked concurrently
+	sizeLock sync.Mutex
+	// size caches the length of the Bytes encoding, computed on first use
+	size int
 }
 
 func (p *ParsedBlock) BlockHeader() common.BlockHeader {
@@ -69,3 +51,15 @@ func (p *ParsedBlock) Verify(ctx context.Context) (common.VerifiedBlock, error)
 	}
 	return p, nil
 }
+func (p *ParsedBlock) Size() int {
+	p.sizeLock.Lock()
+	defer p.sizeLock.Unlock()
+	if p.size == 0 {
+		bytes, err := p.Bytes()
+		if err != nil {
+			return 0
+		}
+		p.size = len(bytes)
+	}
+	return p.size
+}

instance.go

@@ -100,7 +100,7 @@ func (i *Instance) Start(ctx context.Context) error {
 
 	lastNonSimplexHeight := i.Config.LastNonSimplexInnerBlock.Height()
 	genesisValidatorSet := i.Config.PlatformChain.GenesisValidatorSet()
-	nodes, epochNum, err := constructEpochAndValidatorSet(i.Config.Logger, lastNonSimplexHeight, genesisValidatorSet, numBlocks, ParsedBlock{StateMachineBlock: lastBlock}, i.Config.Storage)
+	nodes, epochNum, err := constructEpochAndValidatorSet(i.Config.Logger, lastNonSimplexHeight, genesisValidatorSet, numBlocks, &ParsedBlock{StateMachineBlock: lastBlock}, i.Config.Storage)
 	if err != nil {
 		return fmt.Errorf("error determining latest epoch and validator set: %w", err)
 	}
@@ -417,7 +417,7 @@ func (i *Instance) createEpochConfig() (simplex.EpochConfig, error) {
 
 	lastNonSimplexHeight := i.Config.LastNonSimplexInnerBlock.Height()
 	genesisValidatorSet := i.Config.PlatformChain.GenesisValidatorSet()
-	nodes, epochNum, err := constructEpochAndValidatorSet(i.Config.Logger, lastNonSimplexHeight, genesisValidatorSet, numBlocks, ParsedBlock{StateMachineBlock: lastBlock}, i.Config.Storage)
+	nodes, epochNum, err := constructEpochAndValidatorSet(i.Config.Logger, lastNonSimplexHeight, genesisValidatorSet, numBlocks, &ParsedBlock{StateMachineBlock: lastBlock}, i.Config.Storage)
 	if err != nil {
 		return simplex.EpochConfig{}, err
 	}
@@ -581,7 +581,7 @@ func (i *Instance) transitionEpochValidator(epochChange epochChange) error {
 	return i.startAtEpoch(epochChange.validators, epochChange.epochNum)
 }
 
-func constructEpochAndValidatorSet(logger common.Logger, lastNonSimplexInnerBlockHeight uint64, genesisValidatorSet metadata.NodeBLSMappings, numBlocks uint64, lastBlock ParsedBlock, storage Storage) (common.Nodes, uint64, error) {
+func constructEpochAndValidatorSet(logger common.Logger, lastNonSimplexInnerBlockHeight uint64, genesisValidatorSet metadata.NodeBLSMappings, numBlocks uint64, lastBlock *ParsedBlock, storage Storage) (common.Nodes, uint64, error) {
 	epochNum := lastBlock.BlockHeader().Epoch
 
 	var validatorSet metadata.NodeBLSMappings
@@ -613,7 +613,7 @@ func constructEpochAndValidatorSet(logger common.Logger, lastNonSimplexInnerBloc
 		if sealingBlock.Metadata.SimplexEpochInfo.BlockValidationDescriptor == nil {
 			return nil, 0, fmt.Errorf("expected sealing block at seq %d, but got a non-sealing block", sealingBlockSeq)
 		}
-		validatorSet = constructValidatorSetFromSealingBlock(ParsedBlock{StateMachineBlock: sealingBlock})
+		validatorSet = constructValidatorSetFromSealingBlock(&ParsedBlock{StateMachineBlock: sealingBlock})
 		nodes = validatorSetToNodes(validatorSet)
 		logger.Debug("Determined epoch and validator set from sealing block in storage",
 			zap.Uint64("epoch", epochNum), zap.Uint64("sealingBlockSeq", sealingBlockSeq))
@@ -633,7 +633,7 @@ func validatorSetToNodes(validatorSet metadata.NodeBLSMappings) common.Nodes {
 	return nodes
 }
 
-func constructValidatorSetFromSealingBlock(lastBlock ParsedBlock) metadata.NodeBLSMappings {
+func constructValidatorSetFromSealingBlock(lastBlock *ParsedBlock) metadata.NodeBLSMappings {
 	var validatorSet metadata.NodeBLSMappings
 	vdrs := lastBlock.Metadata.SimplexEpochInfo.BlockValidationDescriptor.AggregatedMembership.Members
 	for _, vdr := range vdrs {

instance_test.go

@@ -385,6 +385,82 @@ func TestInstanceRestartAcrossEpochs(t *testing.T) {
 	// The restarted node keeps extending the chain.
 	waitForNumBlocks(t, storage, storage.NumBlocks()+2)
 }
+func TestParseBlockSizeMatchesBytes(t *testing.T) {
+	// Case 1: Bytes() first, Size() second, size returns the cached length.
+	pb := &ParsedBlock{
+		StateMachineBlock: metadata.StateMachineBlock{
+			Metadata: metadata.StateMachineMetadata{
+				SimplexProtocolMetadata: []byte{1, 2, 3},
+				SimplexBlacklist:        []byte{4, 5},
+				PChainHeight:            6,
+			},
+			InnerBlock: &testInnerBlock{
+				Height_: 7,
+				TS:      time.UnixMilli(8),
+				Payload: []byte("payload"),
+			},
+		},
+	}
+	bytes, err := pb.Bytes()
+	require.NoError(t, err)
+	require.Equal(t, len(bytes), pb.Size())
+
+	// Case 2: Size() first on a non serialized block. it will
+	// compute the size and match a later Byte() call.
+	pb2 := &ParsedBlock{
+		StateMachineBlock: metadata.StateMachineBlock{
+			Metadata: metadata.StateMachineMetadata{
+				SimplexProtocolMetadata: []byte{1, 2, 3},
+				SimplexBlacklist:        []byte{4, 5},
+				PChainHeight:            6,
+			},
+			InnerBlock: &testInnerBlock{
+				Height_: 9,
+				TS:      time.UnixMilli(10),
+				Payload: []byte("other payload"),
+			},
+		},
+	}
+	size := pb2.Size()
+	require.NotZero(t, size)
+	bytes2, err := pb2.Bytes()
+	require.NoError(t, err)
+	require.Equal(t, len(bytes2), size)
+
+	// case 3: cincurrent Size() calls on a block that was never serialized.
+	// the goroutines rase to compute the size, the lock must make this
+	// safe and every call must return the correct value
+
+	pb3 := &ParsedBlock{
+		StateMachineBlock: metadata.StateMachineBlock{
+			Metadata: metadata.StateMachineMetadata{
+				SimplexProtocolMetadata: []byte{1, 2, 3},
+				SimplexBlacklist:        []byte{4, 5},
+				PChainHeight:            6,
+			},
+			InnerBlock: &testInnerBlock{
+				Height_: 11,
+				TS:      time.UnixMilli(12),
+				Payload: []byte("concurrent"),
+			},
+		},
+	}
+	var wg sync.WaitGroup
+	sizes := make([]int, 4)
+	for i := range sizes {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			sizes[i] = pb3.Size()
+		}()
+	}
+	wg.Wait()
+	bytes3, err := pb3.Bytes()
+	require.NoError(t, err)
+	for _, size := range sizes {
+		require.Equal(t, len(bytes3), size)
+	}
+}
 
 // requireTipIsSealing asserts whether the last block in storage is a sealing block.
 func requireTipIsSealing(t *testing.T, storage *MockStorage, want bool) {
@@ -796,7 +872,7 @@ func (m *MockStorage) blockAt(seq uint64) (metadata.StateMachineBlock, bool) {
 }
 
 func (m *MockStorage) parseStored(encoded []byte) metadata.StateMachineBlock {
-	raw := &RawBlock{}
+	raw := &metadata.RawBlock{}
 	require.NoError(m.t, raw.UnmarshalCanoto(encoded))
 	var inner avalanchego.VMBlock
 	if len(raw.InnerBlockBytes) > 0 {
@@ -920,10 +996,10 @@ func (n *inMemNetwork) dispatch(inst *Instance, m netMsg) {
 
 // toRawBlock re-encodes a verified block into the wire RawBlock the receiving
 // instance parses in HandleBlockMessage.
-func toRawBlock(t *testing.T, vb common.VerifiedBlock) *RawBlock {
+func toRawBlock(t *testing.T, vb common.VerifiedBlock) *metadata.RawBlock {
 	bytes, err := vb.Bytes()
 	require.NoError(t, err)
-	raw := &RawBlock{}
+	raw := &metadata.RawBlock{}
 	require.NoError(t, raw.UnmarshalCanoto(bytes))
 	return raw
 }