diff --git a/adapters.go b/adapters.go index 76f083c6..dad6124f 100644 --- a/adapters.go +++ b/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 } diff --git a/common/api.go b/common/api.go index c12dcd2b..44e51232 100644 --- a/common/api.go +++ b/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 } diff --git a/common/metadata.go b/common/metadata.go index 5c3f3c5f..06e9f941 100644 --- a/common/metadata.go +++ b/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 { diff --git a/common/msg.go b/common/msg.go index 19e2ba6e..cdf9bb68 100644 --- a/common/msg.go +++ b/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 diff --git a/common/msg_test.go b/common/msg_test.go index 32567e68..c18fd3f0 100644 --- a/common/msg_test.go +++ b/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()) + +} diff --git a/external.go b/external.go index 22d65ace..13f59853 100644 --- a/external.go +++ b/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 + // lock guards size, so Size() can be invoked concurrently + lock 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.lock.Lock() + defer p.lock.Unlock() + if p.size == 0 { + bytes, err := p.Bytes() + if err != nil { + return 0 + } + p.size = len(bytes) + } + return p.size +} diff --git a/instance.go b/instance.go index d8666232..a306ab7e 100644 --- a/instance.go +++ b/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 { diff --git a/instance_test.go b/instance_test.go index 24d3de74..386463b7 100644 --- a/instance_test.go +++ b/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 } diff --git a/external.canoto.go b/msm/block.canoto.go similarity index 98% rename from external.canoto.go rename to msm/block.canoto.go index 2c3df92f..960373f3 100644 --- a/external.canoto.go +++ b/msm/block.canoto.go @@ -1,9 +1,9 @@ // Code generated by canoto. DO NOT EDIT. // versions: // canoto v0.19.0 -// source: external.go +// source: block.go -package simplex +package metadata import ( "io" @@ -63,7 +63,7 @@ func (*RawBlock) CanotoSpec(types ...reflect.Type) *canoto.Spec { return s } -// UnmarshalCanoto unmarshals a Canoto-rawBlock byte slice into the struct. +// UnmarshalCanoto unmarshals a Canoto-encoded byte slice into the struct. // // During parsing, the canoto cache is saved. func (c *RawBlock) UnmarshalCanoto(bytes []byte) error { diff --git a/msm/block.go b/msm/block.go index 944638ec..dc4aeebd 100644 --- a/msm/block.go +++ b/msm/block.go @@ -11,6 +11,8 @@ import ( "github.com/ava-labs/simplex/common" ) +//go:generate go run github.com/StephenButtolph/canoto/canoto block.go + type BlockType uint8 // A StateMachineBlock is a representation of a parsed OuterBlock, containing the inner block and the metadata. @@ -21,6 +23,14 @@ type StateMachineBlock struct { Metadata StateMachineMetadata } +// RawBlock is the serialized form of a StateMachineBlock. +type RawBlock struct { + Metadata StateMachineMetadata `canoto:"value,1"` + InnerBlockBytes []byte `canoto:"bytes,2"` + + canotoData canotoData_RawBlock +} + // Clone returns a shallow copy of the block, skipping the canoto caches // so it is safe to call while the original is being marshaled. func (smb *StateMachineBlock) Clone() StateMachineBlock { @@ -120,3 +130,19 @@ func (smb *StateMachineBlock) SealingBlockInfo() *common.SealingBlockInfo { PrevSealingBlockHash: smb.Metadata.SimplexEpochInfo.PrevSealingBlockHash, } } + +func (smb *StateMachineBlock) Bytes() ([]byte, error) { + var innerBlockBytes []byte + if smb.InnerBlock != nil { + rawInnerBlock, err := smb.InnerBlock.Bytes() + if err != nil { + return nil, err + } + innerBlockBytes = rawInnerBlock + } + rawBlock := &RawBlock{ + Metadata: smb.Metadata.Clone(), + InnerBlockBytes: innerBlockBytes, + } + return rawBlock.MarshalCanoto(), nil +} diff --git a/simplex/epoch.go b/simplex/epoch.go index ab58f7f4..0a2bc928 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -31,6 +31,9 @@ const ( DefaultFinalizeVoteRebroadcastTimeout = 6 * time.Second EmptyVoteTimeoutID = "rebroadcast_empty_vote" maxItemCountPerRequest = 10 // max number of rounds or sequences that fit in one replication request + // DefaultMaxReplicationResponseSize is the max size of a replication response. avalanchego rejects messages larger + // than 2 MiB. we cap at 80% (4/5) same as avalanchego (see utils/constants/network.go) + DefaultMaxReplicationResponseSize = 2 * 1024 * 1024 * 4 / 5 ) type EmptyVoteSet struct { @@ -61,6 +64,7 @@ func NewRound(block common.VerifiedBlock) *Round { type EpochConfig struct { MaxProposalWait time.Duration MaxRoundWindow uint64 + MaxReplicationResponseSize int MaxRebroadcastWait time.Duration FinalizeRebroadcastTimeout time.Duration QCDeserializer common.QCDeserializer @@ -265,6 +269,9 @@ func (e *Epoch) maybeAssignDefaultConfig() error { if e.MaxRebroadcastWait == 0 { e.MaxRebroadcastWait = DefaultEmptyVoteRebroadcastTimeout } + if e.MaxReplicationResponseSize == 0 { + e.MaxReplicationResponseSize = DefaultMaxReplicationResponseSize + } if e.RandomSource == nil { source, err := NewRandomSource() if err != nil { @@ -3089,19 +3096,26 @@ func (e *Epoch) handleReplicationRequest(req *common.ReplicationRequest, from co zap.Uint64("max round window", e.MaxRoundWindow)) rounds = rounds[:e.MaxRoundWindow] } + remainingBytes := e.MaxReplicationResponseSize if req.LatestRound > 0 { latestRound := e.getLatestVerifiedQuorumRound() if latestRound != nil && latestRound.GetRound() > req.LatestRound { + size := latestRound.Size() response.LatestRound = latestRound + remainingBytes -= size } } if req.LatestFinalizedSeq > 0 { if e.lastBlock != nil && e.lastBlock.Finalization.Finalization.Seq > req.LatestFinalizedSeq { - response.LatestFinalizedSeq = &common.VerifiedQuorumRound{ + latestFinalizedSeq := &common.VerifiedQuorumRound{ VerifiedBlock: e.lastBlock.VerifiedBlock, Finalization: &e.lastBlock.Finalization, } + size := latestFinalizedSeq.Size() + response.LatestFinalizedSeq = latestFinalizedSeq + remainingBytes -= size + } } @@ -3113,6 +3127,16 @@ func (e *Epoch) handleReplicationRequest(req *common.ReplicationRequest, from co seqData = seqData[:i] break } + size := quorumRound.Size() + if size > remainingBytes { + e.Logger.Debug("Replication response reached size limit", + zap.Stringer("from", from), + zap.Uint64("Stopped at Seq", seq), + zap.Int("max size", e.MaxReplicationResponseSize)) + seqData = seqData[:i] + break + } + remainingBytes -= size seqData[i] = *quorumRound } @@ -3124,6 +3148,16 @@ func (e *Epoch) handleReplicationRequest(req *common.ReplicationRequest, from co // we cannot break early since empty votes may continue } + size := quorumRound.Size() + if size > remainingBytes { + e.Logger.Debug("Replication response reached size limit", + zap.Stringer("from", from), + zap.Uint64("Stopped at Round", roundNum), + zap.Int("max size", e.MaxReplicationResponseSize)) + break + } + remainingBytes -= size + roundData = append(roundData, *quorumRound) } diff --git a/simplex/replication_request_test.go b/simplex/replication_request_test.go index b5a66303..834e830a 100644 --- a/simplex/replication_request_test.go +++ b/simplex/replication_request_test.go @@ -492,3 +492,96 @@ func TestReplicationRequestSeqsAndRoundsTruncated(t *testing.T) { require.Equal(t, expected, resp.Data) } + +// TestReplicationRequestSizeLimited ensures a replication response is capped +// at MaxReplicationResponseSize estimated bytes, starting wit the lowest seqs first. +// Sequences that do not fit are requested again by the requester +// so a partial response is not lost. +func TestReplicationRequestSizeLimited(t *testing.T) { + bb := testutil.NewTestBlockBuilder() + nodes := []common.NodeID{{1}, {2}, {3}, {4}} + comm := NewListenerComm(nodes) + ctx := context.Background() + conf, _, _ := testutil.DefaultTestNodeEpochConfig(t, nodes[0], comm, bb) + conf.ReplicationEnabled = true + + numBlocks := uint64(8) // within MaxRoundWindow + seqs := createBlocks(t, nodes, numBlocks) + for _, data := range seqs { + require.NoError(t, conf.Storage.Index(ctx, data.VerifiedBlock, data.Finalization)) + } + + // measure one quorum round and budget for roughly three of them + oneRound := (&common.VerifiedQuorumRound{ + VerifiedBlock: seqs[0].VerifiedBlock, + Finalization: &seqs[0].Finalization, + }).Size() + conf.MaxReplicationResponseSize = 3*oneRound + oneRound/2 + + e, err := simplex.NewEpoch(conf) + require.NoError(t, err) + t.Cleanup(e.Stop) + require.NoError(t, e.Start()) + + requested := make([]uint64, numBlocks) + for i := range requested { + requested[i] = uint64(i) + } + require.NoError(t, e.HandleMessage(&common.Message{ + ReplicationRequest: &common.ReplicationRequest{ + Seqs: requested, + }, + }, nodes[1])) + + msg := <-comm.in + resp := msg.VerifiedReplicationResponse + + // only the lowest seqs that fit under the budget are sent + require.Len(t, resp.Data, 3) + total := 0 + for i, data := range resp.Data { + require.Equal(t, uint64(i), data.VerifiedBlock.BlockHeader().Seq) + size := data.Size() + require.NoError(t, err) + total += size + } + require.LessOrEqual(t, total, conf.MaxReplicationResponseSize) +} + +// TestReplicationRequestSizeLimitedLatestFinalizedSeq ensures the latest-seq/round are +// sent even when they alone exceed the response size limit. A single item that +// cannot fit in a message could not have been sent through the network in the +// first place. +func TestReplicationRequestSizeLimitedLatestFinalizedSeq(t *testing.T) { + bb := testutil.NewTestBlockBuilder() + nodes := []common.NodeID{{1}, {2}, {3}, {4}} + comm := NewListenerComm(nodes) + ctx := context.Background() + conf, _, _ := testutil.DefaultTestNodeEpochConfig(t, nodes[0], comm, bb) + conf.ReplicationEnabled = true + conf.MaxReplicationResponseSize = 1 + + numBlocks := uint64(4) + seqs := createBlocks(t, nodes, numBlocks) + for _, data := range seqs { + require.NoError(t, conf.Storage.Index(ctx, data.VerifiedBlock, data.Finalization)) + } + + e, err := simplex.NewEpoch(conf) + require.NoError(t, err) + t.Cleanup(e.Stop) + require.NoError(t, e.Start()) + + // ask for data and the latest finalized seq: the latest finalized seq must arrive, the data must not + require.NoError(t, e.HandleMessage(&common.Message{ + ReplicationRequest: &common.ReplicationRequest{ + Seqs: []uint64{0, 1, 2}, + LatestFinalizedSeq: 1, + }, + }, nodes[1])) + + msg := <-comm.in + resp := msg.VerifiedReplicationResponse + require.NotNil(t, resp.LatestFinalizedSeq) + require.Empty(t, resp.Data) +} diff --git a/testutil/block.go b/testutil/block.go index 8df92e5a..26950f69 100644 --- a/testutil/block.go +++ b/testutil/block.go @@ -108,6 +108,14 @@ func (t *TestBlock) Bytes() ([]byte, error) { return rawBytes, nil } +func (t *TestBlock) Size() int { + bytes, err := t.Bytes() + if err != nil { + return 0 + } + return len(bytes) +} + type EncodedTestBlock struct { Data []byte Metadata []byte diff --git a/testutil/random_network/block.go b/testutil/random_network/block.go index 0a9ef623..2c5a4b70 100644 --- a/testutil/random_network/block.go +++ b/testutil/random_network/block.go @@ -84,6 +84,14 @@ func (b *Block) Bytes() ([]byte, error) { return asn1.Marshal(encodedB) } +func (b *Block) Size() int { + bytes, err := b.Bytes() + if err != nil { + return 0 + } + return len(bytes) +} + func (b *Block) containsTX(txID txID) bool { for _, tx := range b.txs { if tx.ID == txID { diff --git a/testutil/util.go b/testutil/util.go index b4694d86..1150d2ed 100644 --- a/testutil/util.go +++ b/testutil/util.go @@ -176,6 +176,9 @@ func (t TestQC) Bytes() []byte { } return bytes } +func (t TestQC) Size() int { + return len(t.Bytes()) +} type TestSigner struct { }