From ed41178f1e85a6678ae8b4cd3eebaea080b3ed09 Mon Sep 17 00:00:00 2001 From: Yacov Manevich Date: Mon, 22 Jun 2026 22:12:37 +0200 Subject: [PATCH] ... Signed-off-by: Yacov Manevich --- adapters.go | 249 +++++++++++++ common/api.go | 4 +- common/msg.go | 2 +- config.go | 103 ++++++ external.canoto.go | 217 +++++++++++ external.go | 92 +++++ instance.go | 442 +++++++++++++++++++++++ instance_test.go | 722 +++++++++++++++++++++++++++++++++++++ msm/approvals.go | 55 ++- msm/approvals_test.go | 146 ++++++++ msm/build_decision.go | 26 +- msm/build_decision_test.go | 30 +- msm/fake_node_test.go | 85 ++--- msm/fuzz_test.go | 37 +- msm/misc.go | 3 + msm/msm.go | 98 ++++- msm/msm_test.go | 146 ++++---- msm/util_test.go | 75 ++-- simplex/epoch.go | 10 + simplex/epoch_test.go | 37 ++ testutil/block.go | 27 ++ testutil/util.go | 8 +- wal/gc.go | 21 +- wal/gc_test.go | 4 +- 24 files changed, 2387 insertions(+), 252 deletions(-) create mode 100644 adapters.go create mode 100644 config.go create mode 100644 external.canoto.go create mode 100644 external.go create mode 100644 instance.go create mode 100644 instance_test.go diff --git a/adapters.go b/adapters.go new file mode 100644 index 00000000..d7e033c9 --- /dev/null +++ b/adapters.go @@ -0,0 +1,249 @@ +// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved. +// See the file LICENSE for licensing terms. + +package simplex + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + + "github.com/ava-labs/simplex/common" + metadata "github.com/ava-labs/simplex/msm" +) + +type Communication struct { + blocked *atomic.Bool + nodes common.Nodes + Sender +} + +func (c *Communication) Validators() common.Nodes { + return c.nodes +} + +func (c *Communication) Broadcast(msg *common.Message) { + if c.blocked.Load() { + return + } + for _, node := range c.nodes { + c.Sender.Send(msg, node.Id) + } +} + +type EpochAwareStorage struct { + msm *metadata.StateMachine + OnEpochChange func(seq uint64) error + Storage + Epoch uint64 +} + +func (e *EpochAwareStorage) Retrieve(seq uint64) (common.VerifiedBlock, common.Finalization, error) { + block, finalization, err := e.Storage.GetBlock(seq) + if err != nil { + return nil, common.Finalization{}, err + } + parsedBlock := &ParsedBlock{ + msm: e.msm, + StateMachineBlock: block, + } + return parsedBlock, *finalization, nil +} + +func (e *EpochAwareStorage) Index(ctx context.Context, block common.VerifiedBlock, certificate common.Finalization) error { + if block.BlockHeader().Epoch < e.Epoch { + // This is a Telock from a previous h, so we ignore it and do not index it. + return nil + } + if err := e.Storage.Index(ctx, block, certificate); err != nil { + return err + } + if block.SealingBlockInfo() != nil { + if err := e.OnEpochChange(block.BlockHeader().Seq); err != nil { + return err + } + // We are now in a new h, so we update the h number to prevent indexing Telocks from the previous h. + e.Epoch = block.BlockHeader().Seq + } + return nil +} + +type cachedBlock struct { + cache *CachedStorage + *ParsedBlock +} + +func (cb *cachedBlock) Verify(ctx context.Context) (common.VerifiedBlock, error) { + vb, err := cb.ParsedBlock.Verify(ctx) + if err == nil { + cb.cache.insertBlock(cb.ParsedBlock) + } + return vb, err +} + +type CachedStorage struct { + msm *metadata.StateMachine + lock sync.RWMutex + Storage + cache map[[32]byte]cachedBlock +} + +func NewCachedStorage(storage Storage) *CachedStorage { + return &CachedStorage{ + Storage: storage, + cache: make(map[[32]byte]cachedBlock), + } +} + +func (cs *CachedStorage) RetrieveBlock(seq uint64, digest [32]byte) (metadata.StateMachineBlock, *common.Finalization, error) { + block, finalization, err := cs.Retrieve(seq, digest) + if err != nil { + return metadata.StateMachineBlock{}, nil, err + } + + return block.(*ParsedBlock).StateMachineBlock, finalization, nil +} + +func (cs *CachedStorage) Retrieve(seq uint64, digest [32]byte) (common.VerifiedBlock, *common.Finalization, error) { + cs.lock.RLock() + item, exists := cs.cache[digest] + if exists { + cs.lock.RUnlock() + // If the block is cached, it means it's not finalized yet, because upon finalizing the block (indexing) + // we also remove it from the cache. Therefore, we return nil for the finalization. + return item.ParsedBlock, nil, nil + } + cs.lock.RUnlock() + + // We don't populate the cache here because we populate it externally. + + block, finalization, err := cs.Storage.GetBlock(seq) + if err != nil { + return nil, nil, err + } + + return &ParsedBlock{ + StateMachineBlock: block, + msm: cs.msm, + }, finalization, nil +} + +func (cs *CachedStorage) Index(ctx context.Context, block common.VerifiedBlock, certificate common.Finalization) error { + err := cs.Storage.Index(ctx, block, certificate) + + if err == nil { + // We delete the block from the cache after it has been indexed because now that it is persisted, + // we can just lookup by sequence number instead of digest. + cs.lock.Lock() + defer cs.lock.Unlock() + delete(cs.cache, block.BlockHeader().Digest) + } + + return err +} + +func (cs *CachedStorage) insertBlock(block *ParsedBlock) { + cs.lock.Lock() + defer cs.lock.Unlock() + + cs.cache[block.Digest()] = cachedBlock{ + ParsedBlock: block, + } +} + +type NoopAuxiliaryInfoApp struct{} + +func (n *NoopAuxiliaryInfoApp) IsLegalAppend(versionID metadata.VersionID, nodes metadata.NodeBLSMappings, history [][]byte, x []byte) error { + if len(x) > 0 { + return fmt.Errorf("input should be empty") + } + return nil +} + +func (n *NoopAuxiliaryInfoApp) IsSufficient(versionID metadata.VersionID, nodes metadata.NodeBLSMappings, history [][]byte) (bool, error) { + return true, nil +} + +func (n *NoopAuxiliaryInfoApp) Generate(versionID metadata.VersionID, nodes metadata.NodeBLSMappings, history [][]byte) ([]byte, error) { + return nil, nil +} + +func (n *NoopAuxiliaryInfoApp) DefaultVersionID() metadata.VersionID { + return 0 +} + +type BlockBuilderWaiter struct { + blocked *atomic.Bool + lock sync.Mutex + cancel context.CancelFunc + msm *metadata.StateMachine + vm VM +} + +func (bw *BlockBuilderWaiter) stop() { + bw.lock.Lock() + defer bw.lock.Unlock() + if bw.cancel != nil { + bw.cancel() + bw.cancel = nil + } +} + +func (bw *BlockBuilderWaiter) WaitForPendingBlock(ctx context.Context) { + if bw.blocked.Load() { + return + } + + bw.lock.Lock() + if bw.cancel != nil { + bw.cancel() + } + ctx, cancel := context.WithCancel(ctx) + bw.cancel = cancel + bw.lock.Unlock() + defer cancel() + bw.vm.WaitForPendingBlock(ctx) +} + +func (bw *BlockBuilderWaiter) BuildBlock(ctx context.Context, metadata common.ProtocolMetadata, blacklist common.Blacklist) (common.VerifiedBlock, bool) { + if bw.blocked.Load() { + return nil, false + } + + block, err := bw.msm.BuildBlock(ctx, metadata, &blacklist) + if err != nil { + return nil, false + } + + pb := ParsedBlock{ + StateMachineBlock: *block, + msm: bw.msm, + } + + return &pb, true +} + +type blockDeserializer struct { + vm VM + msm *metadata.StateMachine +} + +func (bp *blockDeserializer) DeserializeBlock(ctx context.Context, bytes []byte) (common.Block, error) { + var rawBlock RawBlock + if err := rawBlock.UnmarshalCanoto(bytes); err != nil { + return nil, err + } + + block, err := bp.vm.ParseBlock(ctx, rawBlock.InnerBlockBytes) + if err != nil { + return nil, err + } + return &ParsedBlock{ + StateMachineBlock: metadata.StateMachineBlock{ + InnerBlock: block, + Metadata: rawBlock.Metadata, + }, + msm: bp.msm, + }, nil +} diff --git a/common/api.go b/common/api.go index cee833e4..c12dcd2b 100644 --- a/common/api.go +++ b/common/api.go @@ -75,7 +75,7 @@ type Signer interface { } type SignatureVerifier interface { - Verify(message []byte, signature []byte, publicKey []byte) error + VerifySignature(message []byte, signature []byte, publicKey []byte) error } type WriteAheadLog interface { @@ -126,7 +126,7 @@ type VerifiedBlock interface { // BlockDeserializer deserializes blocks according to formatting // enforced by the application. type BlockDeserializer interface { - // DeserializeBlock parses the given bytes and initializes a VerifiedBlock. + // DeserializeBlock deserializes the given bytes and initializes a VerifiedBlock. // Returns an error upon failure. DeserializeBlock(ctx context.Context, bytes []byte) (Block, error) } diff --git a/common/msg.go b/common/msg.go index c6fc97d7..0d56ed75 100644 --- a/common/msg.go +++ b/common/msg.go @@ -121,7 +121,7 @@ func verifyContext(signature []byte, verifier SignatureVerifier, msg []byte, con if err != nil { return err } - return verifier.Verify(toBeSigned, signature, pk) + return verifier.VerifySignature(toBeSigned, signature, pk) } func verifyContextQC(qc QuorumCertificate, msg []byte, context string, nodes Nodes) error { diff --git a/config.go b/config.go new file mode 100644 index 00000000..0646127d --- /dev/null +++ b/config.go @@ -0,0 +1,103 @@ +// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved. +// See the file LICENSE for licensing terms. + +package simplex + +import ( + "context" + "time" + + "github.com/ava-labs/simplex/common" + metadata "github.com/ava-labs/simplex/msm" + "github.com/ava-labs/simplex/wal" +) + +type ParameterConfig struct { + // WalMaxEntryCount is the maximum number of entries in the write-ahead log before it is closed. + WALMaxEntryCount int + // MaxnetworkDelay is the assumed upper bound on the network delay for messages to be delivered. + MaxNetworkDelay time.Duration + // MaxRoundWindow is the maximum number of rounds that can be stored in memory. + MaxRoundWindow uint64 +} + +// PlatformChain is an interface that abstracts the interaction with the P-chain. +type PlatformChain interface { + GetValidatorSet(uint64) (metadata.NodeBLSMappings, error) + // GenesisValidatorSet returns the first ever validator set for this network. + GenesisValidatorSet() metadata.NodeBLSMappings + // GetMinimumHeight returns the minimum height of the block still in the proposal window. + GetMinimumHeight(context.Context) (uint64, error) + // GetCurrentHeight returns the current height of the P-chain. + GetCurrentHeight(context.Context) (uint64, error) + // WaitForProgress should block until either the context is cancelled, or the P-chain height has increased from the provided pChainHeight. + WaitForProgress(ctx context.Context, pChainHeight uint64) error + // LastNonSimplexBlockPChainHeight returns the P-chain height of the last non-simplex block in the chain. + LastNonSimplexBlockPChainHeight() uint64 +} + +// Sender is an interface that defines the ability to send messages to other nodes in the network. +type Sender interface { + // Send sends a message to the given destination node + Send(msg *common.Message, destination common.NodeID) +} + +type VM interface { + // BuildBlock builds a block given the current context and the P-chain height. + BuildBlock(ctx context.Context, pChainHeight uint64) (metadata.VMBlock, error) + + // WaitForPendingBlock returns when either the given context is cancelled, + // or when the VM signals that a block should be built. + WaitForPendingBlock(ctx context.Context) + + // ParseBlock parses the given block bytes into a VMBlock. + ParseBlock(context.Context, []byte) (metadata.VMBlock, error) + + // ComputeICMEpoch computes the ICM epoch transition given the input parameters. + ComputeICMEpoch(input metadata.ICMEpochInput) metadata.ICMEpochInfo +} + +type Storage interface { + // GetBlock retrieves the block and finalization at [seq] with the given digest. + // If the digest is nil, the block with the given sequence number is returned. + // If [seq] the block cannot be found, returns ErrBlockNotFound. + GetBlock(seq uint64) (metadata.StateMachineBlock, *common.Finalization, error) + + // NumBlocks returns the number of blocks stored in the storage. + NumBlocks() uint64 + + // Index indexes the given block and finalization in the storage. + Index(ctx context.Context, block common.VerifiedBlock, certificate common.Finalization) error + + // CreateWAL creates a new Write-Ahead Log (WAL). + CreateWAL() (wal.DeletableWAL, error) +} + +type CryptoOps interface { + Sign(message []byte) ([]byte, error) + AggregateKeys(keys ...[]byte) ([]byte, error) + VerifySignature(message []byte, signature []byte, publicKey []byte) error + CreateSignatureAggregator([]common.Node) common.SignatureAggregator + DeserializeQuorumCertificate(bytes []byte) (common.QuorumCertificate, error) +} + +type MSMConfig struct { + LastNonSimplexInnerBlock metadata.VMBlock + ComputeICMEpoch metadata.ICMEpochTransition +} + +type EpochParams struct { + ID common.NodeID + Sender Sender + Storage Storage + QCDeserializer common.QCDeserializer + BlockDeserializer common.BlockDeserializer + Verifier common.SignatureVerifier + MaxProposalWait time.Duration + MaxRoundWindow uint64 + MaxRebroadcastWait time.Duration + FinalizeRebroadcastTimeout time.Duration + MaxWALSize int + WALCreator wal.Creator + WALs []wal.DeletableWAL +} diff --git a/external.canoto.go b/external.canoto.go new file mode 100644 index 00000000..0286d2f2 --- /dev/null +++ b/external.canoto.go @@ -0,0 +1,217 @@ +// Code generated by canoto. DO NOT EDIT. +// versions: +// canoto v0.19.0 +// source: external.go + +package simplex + +import ( + "io" + "reflect" + "sync/atomic" + + "github.com/StephenButtolph/canoto" +) + +// Ensure that the generated code is compatible with the library version. +const ( + _ uint = canoto.VersionCompatibility - 1 + _ uint = 1 - canoto.VersionCompatibility +) + +// Ensure that unused imports do not error +var _ = io.ErrUnexpectedEOF + +const ( + canotoNumber_RawBlock__Metadata = 1 + canotoNumber_RawBlock__InnerBlockBytes = 2 + + canotoTag_RawBlock__Metadata = "\x0a" // canoto.Tag(canotoNumber_RawBlock__Metadata, canoto.Len) + canotoTag_RawBlock__InnerBlockBytes = "\x12" // canoto.Tag(canotoNumber_RawBlock__InnerBlockBytes, canoto.Len) +) + +type canotoData_RawBlock struct { + size uint64 +} + +// CanotoSpec returns the specification of this canoto message. +func (*RawBlock) CanotoSpec(types ...reflect.Type) *canoto.Spec { + types = append(types, reflect.TypeFor[RawBlock]()) + var zero RawBlock + s := &canoto.Spec{ + Name: "RawBlock", + Fields: []canoto.FieldType{ + canoto.FieldTypeFromField( + /*type inference:*/ (&zero.Metadata), + /*FieldNumber: */ canotoNumber_RawBlock__Metadata, + /*Name: */ "Metadata", + /*FixedLength: */ 0, + /*Repeated: */ false, + /*OneOf: */ "", + /*Pointer: */ false, + /*types: */ types, + ), + { + FieldNumber: canotoNumber_RawBlock__InnerBlockBytes, + Name: "InnerBlockBytes", + OneOf: "", + TypeBytes: true, + }, + }, + } + s.CalculateCanotoCache() + return s +} + +// UnmarshalCanoto unmarshals a Canoto-encoded byte slice into the struct. +// +// During parsing, the canoto cache is saved. +func (c *RawBlock) UnmarshalCanoto(bytes []byte) error { + r := canoto.Reader{ + B: bytes, + } + return c.UnmarshalCanotoFrom(r) +} + +// UnmarshalCanotoFrom populates the struct from a [canoto.Reader]. Most users +// should just use UnmarshalCanoto. +// +// During parsing, the canoto cache is saved. +// +// This function enables configuration of reader options. +func (c *RawBlock) UnmarshalCanotoFrom(r canoto.Reader) error { + // Zero the struct before unmarshaling. + *c = RawBlock{} + atomic.StoreUint64(&c.canotoData.size, uint64(len(r.B))) + + var minField uint32 + for canoto.HasNext(&r) { + field, wireType, err := canoto.ReadTag(&r) + if err != nil { + return err + } + if field < minField { + return canoto.ErrInvalidFieldOrder + } + + switch field { + case canotoNumber_RawBlock__Metadata: + if wireType != canoto.Len { + return canoto.ErrUnexpectedWireType + } + + // Read the bytes for the field. + originalUnsafe := r.Unsafe + r.Unsafe = true + var msgBytes []byte + if err := canoto.ReadBytes(&r, &msgBytes); err != nil { + return err + } + if len(msgBytes) == 0 { + return canoto.ErrZeroValue + } + r.Unsafe = originalUnsafe + + // Unmarshal the field from the bytes. + remainingBytes := r.B + r.B = msgBytes + if err := (&c.Metadata).UnmarshalCanotoFrom(r); err != nil { + return err + } + r.B = remainingBytes + case canotoNumber_RawBlock__InnerBlockBytes: + if wireType != canoto.Len { + return canoto.ErrUnexpectedWireType + } + + if err := canoto.ReadBytes(&r, &c.InnerBlockBytes); err != nil { + return err + } + if len(c.InnerBlockBytes) == 0 { + return canoto.ErrZeroValue + } + default: + return canoto.ErrUnknownField + } + + minField = field + 1 + } + return nil +} + +// ValidCanoto validates that the struct can be correctly marshaled into the +// Canoto format. +// +// Specifically, ValidCanoto ensures: +// 1. All OneOfs are specified at most once. +// 2. All strings are valid utf-8. +// 3. All custom fields are ValidCanoto. +func (c *RawBlock) ValidCanoto() bool { + if !(&c.Metadata).ValidCanoto() { + return false + } + return true +} + +// CalculateCanotoCache populates size and OneOf caches based on the current +// values in the struct. +// +// It is not safe to copy this struct concurrently. +func (c *RawBlock) CalculateCanotoCache() { + var size uint64 + (&c.Metadata).CalculateCanotoCache() + if fieldSize := (&c.Metadata).CachedCanotoSize(); fieldSize != 0 { + size += uint64(len(canotoTag_RawBlock__Metadata)) + canoto.SizeUint(fieldSize) + fieldSize + } + if len(c.InnerBlockBytes) != 0 { + size += uint64(len(canotoTag_RawBlock__InnerBlockBytes)) + canoto.SizeBytes(c.InnerBlockBytes) + } + atomic.StoreUint64(&c.canotoData.size, size) +} + +// CachedCanotoSize returns the previously calculated size of the Canoto +// representation from CalculateCanotoCache. +// +// If CalculateCanotoCache has not yet been called, it will return 0. +// +// If the struct has been modified since the last call to CalculateCanotoCache, +// the returned size may be incorrect. +func (c *RawBlock) CachedCanotoSize() uint64 { + return atomic.LoadUint64(&c.canotoData.size) +} + +// MarshalCanoto returns the Canoto representation of this struct. +// +// It is assumed that this struct is ValidCanoto. +// +// It is not safe to copy this struct concurrently. +func (c *RawBlock) MarshalCanoto() []byte { + c.CalculateCanotoCache() + w := canoto.Writer{ + B: make([]byte, 0, c.CachedCanotoSize()), + } + w = c.MarshalCanotoInto(w) + return w.B +} + +// MarshalCanotoInto writes the struct into a [canoto.Writer] and returns the +// resulting [canoto.Writer]. Most users should just use MarshalCanoto. +// +// It is assumed that CalculateCanotoCache has been called since the last +// modification to this struct. +// +// It is assumed that this struct is ValidCanoto. +// +// It is not safe to copy this struct concurrently. +func (c *RawBlock) MarshalCanotoInto(w canoto.Writer) canoto.Writer { + if fieldSize := (&c.Metadata).CachedCanotoSize(); fieldSize != 0 { + canoto.Append(&w, canotoTag_RawBlock__Metadata) + canoto.AppendUint(&w, fieldSize) + w = (&c.Metadata).MarshalCanotoInto(w) + } + if len(c.InnerBlockBytes) != 0 { + canoto.Append(&w, canotoTag_RawBlock__InnerBlockBytes) + canoto.AppendBytes(&w, c.InnerBlockBytes) + } + return w +} diff --git a/external.go b/external.go new file mode 100644 index 00000000..3ceb0f18 --- /dev/null +++ b/external.go @@ -0,0 +1,92 @@ +// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved. +// See the file LICENSE for licensing terms. + +package simplex + +import ( + "context" + + "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 +} + +func (p *ParsedBlock) BlockHeader() common.BlockHeader { + var md *common.ProtocolMetadata + var err error + if len(p.Metadata.SimplexProtocolMetadata) > 0 { + md, err = common.ProtocolMetadataFromBytes(p.Metadata.SimplexProtocolMetadata) + if err != nil { + panic(err) // TODO: handle error + } + } else { + md = &common.ProtocolMetadata{} + } + + digest := p.StateMachineBlock.Digest() + return common.BlockHeader{ + ProtocolMetadata: *md, + Digest: digest, + } +} + +func (p *ParsedBlock) Blacklist() common.Blacklist { + var blacklist common.Blacklist + _ = blacklist.FromBytes(p.Metadata.SimplexBlacklist) // TODO: encode blacklist with Canoto + return blacklist +} + +func (p *ParsedBlock) Verify(ctx context.Context) (common.VerifiedBlock, error) { + if err := p.msm.VerifyBlock(ctx, &p.StateMachineBlock); err != nil { + return nil, err + } + return p, nil +} + +func (p *ParsedBlock) SealingBlockInfo() *common.SealingBlockInfo { + if p.Metadata.SimplexEpochInfo.BlockValidationDescriptor == nil { + return nil + } + + bdc := p.Metadata.SimplexEpochInfo.BlockValidationDescriptor + var nodes common.Nodes + + for _, vdr := range bdc.AggregatedMembership.Members { + nodes = append(nodes, common.Node{ + Id: vdr.NodeID[:], + Weight: vdr.Weight, + PK: vdr.BLSKey, + }) + } + + return &common.SealingBlockInfo{ + ValidatorSet: nodes, + } +} diff --git a/instance.go b/instance.go new file mode 100644 index 00000000..77275bb4 --- /dev/null +++ b/instance.go @@ -0,0 +1,442 @@ +// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved. +// See the file LICENSE for licensing terms. + +package simplex + +import ( + "context" + "fmt" + "math" + "sync" + "sync/atomic" + "time" + + "github.com/ava-labs/simplex/common" + metadata "github.com/ava-labs/simplex/msm" + "github.com/ava-labs/simplex/nonvalidator" + "github.com/ava-labs/simplex/simplex" + "github.com/ava-labs/simplex/wal" + "go.uber.org/zap" +) + +const ( + tickInterval = time.Millisecond * 100 +) + +type Config struct { + LastNonSimplexInnerBlock metadata.VMBlock + ParameterConfig ParameterConfig + PlatformChain PlatformChain + CryptoOps CryptoOps + Storage Storage + Logger common.Logger + Sender Sender + WALs []wal.DeletableWAL + VM VM + ID common.NodeID +} + +type MsgHandler interface { + HandleMessage(msg *common.Message, from common.NodeID) error +} + +type Instance struct { + Config Config + lock sync.Mutex + cs *CachedStorage + wal *wal.GarbageCollectedWAL + msm *metadata.StateMachine + e *simplex.Epoch + epochChanges chan uint64 + stopCh chan struct{} + duringEpochChange atomic.Bool +} + +func (i *Instance) Start() error { + i.stopCh = make(chan struct{}) + i.epochChanges = make(chan uint64) + + + lastBlock, numBlocks, err := i.lastBlock() + + lastNonSimplexHeight := i.Config.LastNonSimplexInnerBlock.Height() + genesisValidatorSet := i.Config.PlatformChain.GenesisValidatorSet() + nodes, epochNum, err := constructEpochAndValidatorSet(lastNonSimplexHeight, genesisValidatorSet, numBlocks, ParsedBlock{StateMachineBlock: lastBlock,}, i.Config.Storage) + if err != nil { + return fmt.Errorf("error starting instance: %w", err) + } + + iAmValidator := i.determineValidatorOrNot(nodes) + + if iAmValidator { + if err := i.startValidator(); err != nil { + return fmt.Errorf("error starting validator: %w", err) + } + } else { + if err := i.startNonValidator(); err != nil { + return fmt.Errorf("error starting non-validator: %w", err) + } + } + + go i.tick() + go i.listenForEpochChanges() + + return nil +} + +func (i *Instance) startValidator() error { + epochConfig, err := i.createEpochConfig() + if err != nil { + return err + } + + if err := i.startEpoch(epochConfig); err != nil { + return err + } +} + +func (i *Instance) startNonValidator(epochNum uint64) error { + source, err := simplex.NewRandomSource() + if err != nil { + return err + } + + epochAwareStorage := &EpochAwareStorage{ + Epoch: epochNum, + Storage: i.Config.Storage, + OnEpochChange: func(epoch uint64) error { + i.duringEpochChange.Store(true) + i.epochChanges <- epoch + return nil + }, + } + + nonvalidator.NewNonValidator(nonvalidator.Config{ + ID: i.Config.ID, + RandomSource: source, + Storage: epochAwareStorage, + Comm: &Communication{nodes: nodes, Sender: i.Config.Sender, blocked: &i.duringEpochChange} + }) +} + +func (i *Instance) tick() { + ticker := time.NewTicker(tickInterval) + for { + select { + case now := <-ticker.C: + i.lock.Lock() + epoch := i.e + i.lock.Unlock() + if epoch == nil { + // Stopped, so we exit the tick loop. + return + } + epoch.AdvanceTime(now) + case <-i.stopCh: + return + } + } +} + +func (i *Instance) Stop() { + i.lock.Lock() + defer i.lock.Unlock() + + if i.e != nil { + i.e.Stop() + close(i.stopCh) + i.e = nil + } +} + +func (i *Instance) HandleMessage(msg *common.Message, from common.NodeID) error { + i.lock.Lock() + defer i.lock.Unlock() + + if i.duringEpochChange.Load() || i.e == nil { + return nil + } + + return i.e.HandleMessage(msg, from) +} + +func (i *Instance) HandleBlockMessage(ctx context.Context, block *RawBlock, vote *common.Vote, from common.NodeID) error { + i.lock.Lock() + defer i.lock.Unlock() + + if i.duringEpochChange.Load() || i.e == nil { + return nil + } + + if block == nil || vote == nil { + i.Config.Logger.Debug("Received nil block or vote") + return nil + } + + // TODO: use a real context + vmBlock, err := i.Config.VM.ParseBlock(ctx, block.InnerBlockBytes) + if err != nil { + i.Config.Logger.Debug("Failed to parse inner block", zap.Error(err)) + return nil + } + + parsedBlock := ParsedBlock{ + StateMachineBlock: metadata.StateMachineBlock{ + InnerBlock: vmBlock, + Metadata: block.Metadata, + }, + msm: i.msm, + } + + var cb = &cachedBlock{ + cache: i.cs, + ParsedBlock: &parsedBlock, + } + + msg := &common.Message{ + BlockMessage: &common.BlockMessage{ + Block: cb, + Vote: *vote, + }, + } + + return i.e.HandleMessage(msg, from) +} + +func (i *Instance) listenForEpochChanges() { + for { + select { + case epochNum := <-i.epochChanges: + i.transitionEpoch(epochNum) + case <-i.stopCh: + return + } + } +} + +// startEpoch starts a new epoch with the given configuration. +// Must be called under the lock, and assumes that the previous epoch has been stopped (if any). +func (i *Instance) startEpoch(epochConfig simplex.EpochConfig) error { + epoch, err := simplex.NewEpoch(epochConfig) + if err != nil { + return fmt.Errorf("error creating simplex epoch: %w", err) + } + epoch.Epoch = epochConfig.Epoch + i.e = epoch + + return epoch.Start() +} + +func (i *Instance) lastBlock() (metadata.StateMachineBlock, uint64 , error) { + numBlocks := i.Config.Storage.NumBlocks() + if numBlocks == 0 { + return metadata.StateMachineBlock{}, 0, fmt.Errorf("no genesis block found in storage") + } + + lastBlock, _, err := i.Config.Storage.GetBlock(numBlocks - 1) + if err != nil { + return metadata.StateMachineBlock{}, 0, fmt.Errorf("error retrieving last block from storage: %w", err) + } + + return lastBlock, numBlocks, nil +} + +func (i *Instance) determineValidatorOrNot(nodes common.Nodes) bool { + for _, node := range nodes { + if i.Config.ID.Equals(node.Id) { + return true + } + } + return false +} + +func (i *Instance) createEpochConfig() (simplex.EpochConfig, error) { + lastBlock, numBlocks, err := i.lastBlock() + + lastNonSimplexHeight := i.Config.LastNonSimplexInnerBlock.Height() + genesisValidatorSet := i.Config.PlatformChain.GenesisValidatorSet() + nodes, epochNum, err := constructEpochAndValidatorSet(lastNonSimplexHeight, genesisValidatorSet, numBlocks, ParsedBlock{StateMachineBlock: lastBlock,}, i.Config.Storage) + if err != nil { + return simplex.EpochConfig{}, err + } + + wal, err := wal.NewGarbageCollectedWAL(i.Config.WALs, i.Config.Storage.CreateWAL, &common.WALRetentionReader{}, i.Config.ParameterConfig.WALMaxEntryCount) + if err != nil { + return simplex.EpochConfig{}, fmt.Errorf("error creating garbage collected wal: %w", err) + } + i.wal = wal + + if lastBlock.Metadata.SimplexEpochInfo.BlockValidationDescriptor != nil { + i.Config.Logger.Info("Last block is a sealing block, garbage collecting all WALs to start a new epoch") + if err := i.wal.GarbageCollect(math.MaxUint64); err != nil { + return simplex.EpochConfig{}, fmt.Errorf("error garbage collecting WALs: %w", err) + } + } + + cachedStorage := NewCachedStorage(i.Config.Storage) + i.cs = cachedStorage + + msm, err := metadata.NewStateMachine(&metadata.Config{ + GetTime: time.Now, + MyNodeID: i.Config.ID, + KeyAggregator: i.Config.CryptoOps, + GetValidatorSet: i.Config.PlatformChain.GetValidatorSet, + SignatureVerifier: i.Config.CryptoOps, + PChainProgressListener: i.Config.PlatformChain, + LatestPersistedHeight: i.Config.Storage.NumBlocks(), + MaxBlockBuildingWaitTime: i.Config.ParameterConfig.MaxNetworkDelay, + Logger: i.Config.Logger, + Signer: i.Config.CryptoOps, + GenesisValidatorSet: genesisValidatorSet, + LastNonSimplexBlockPChainHeight: lastNonSimplexHeight, + SignatureAggregatorCreator: i.Config.CryptoOps.CreateSignatureAggregator, + BlockBuilder: i.Config.VM, + LastNonSimplexInnerBlock: i.Config.LastNonSimplexInnerBlock, + GetPChainHeightForProposing: i.Config.PlatformChain.GetMinimumHeight, + GetPChainHeightForVerifying: i.Config.PlatformChain.GetCurrentHeight, + AuxiliaryInfoApp: &NoopAuxiliaryInfoApp{}, + ComputeICMEpoch: i.Config.VM.ComputeICMEpoch, + GetBlock: cachedStorage.RetrieveBlock, + }) + if err != nil { + return simplex.EpochConfig{}, fmt.Errorf("error creating metadata state machine: %w", err) + } + + i.msm = msm + + // Wire the MSM into cached storage. + // We do this because otherwise we have a chicken and egg problem - MSM needs the cached storage, + // but the cached storage also needs the MSM + cachedStorage.msm = msm + + source, err := simplex.NewRandomSource() + if err != nil { + return simplex.EpochConfig{}, err + } + + blockBuilder := &BlockBuilderWaiter{vm: i.Config.VM, msm: msm, blocked: &i.duringEpochChange} + + epochAwareStorage := &EpochAwareStorage{ + Epoch: epochNum, + Storage: cachedStorage, + OnEpochChange: func(epoch uint64) error { + i.duringEpochChange.Store(true) + blockBuilder.stop() + i.epochChanges <- epoch + return nil + }, + } + + epochConfig := simplex.EpochConfig{ + // Parameter config + Epoch: epochNum, + ReplicationEnabled: true, + StartTime: time.Now(), + // TODO: For simpicity, we use the same value for all timeouts. If needed we can expand the config. + MaxProposalWait: i.Config.ParameterConfig.MaxNetworkDelay * 2, // 1 proposal + 1 vote + MaxRebroadcastWait: i.Config.ParameterConfig.MaxNetworkDelay * 2, + FinalizeRebroadcastTimeout: i.Config.ParameterConfig.MaxNetworkDelay * 2, + MaxRoundWindow: i.Config.ParameterConfig.MaxRoundWindow, + ID: i.Config.ID, + RandomSource: source, // Seed the random source from crypto/rand + WAL: wal, + Logger: i.Config.Logger, + SignatureAggregatorCreator: i.Config.CryptoOps.CreateSignatureAggregator, + QCDeserializer: i.Config.CryptoOps, + Signer: i.Config.CryptoOps, + Verifier: i.Config.CryptoOps, + Storage: epochAwareStorage, + Comm: &Communication{nodes: nodes, Sender: i.Config.Sender, blocked: &i.duringEpochChange}, + BlockBuilder: blockBuilder, + BlockDeserializer: &blockDeserializer{vm: i.Config.VM, msm: msm}, + } + return epochConfig, nil +} + +func (i *Instance) transitionEpoch(epochNum uint64) { + i.lock.Lock() + defer i.lock.Unlock() + + // Stop the epoch before doing anything else, so that we don't process any more messages while we are changing epochs. + i.e.Stop() + // Wipe out the WALs from the config so we won't try to load them again + i.Config.WALs = nil + // On epoch change, garbage collect the WAL to remove all entries from previous epochs. + if err := i.wal.GarbageCollect(math.MaxUint64); err != nil { + i.Config.Logger.Error("Error garbage collecting epoch config on epoch change", zap.Error(err)) + } + config, err := i.createEpochConfig() + if err != nil { + i.Config.Logger.Error("Error creating epoch config on epoch change", zap.Error(err)) + } + + if config.Epoch != epochNum { + i.Config.Logger.Error("Epoch number mismatch on epoch change", zap.Uint64("expected", epochNum), zap.Uint64("actual", config.Epoch)) + return + } + + i.duringEpochChange.Store(false) + + if err := i.startEpoch(config); err != nil { + i.Config.Logger.Error("Error starting new epoch on epoch change", zap.Error(err)) + } +} + +func constructEpochAndValidatorSet(lastNonSimplexInnerBlockHeight uint64, genesisValidatorSet metadata.NodeBLSMappings, numBlocks uint64, lastBlock ParsedBlock, storage Storage) (common.Nodes, uint64, error) { + epochNum := lastBlock.BlockHeader().Epoch + + var validatorSet metadata.NodeBLSMappings + var nodes common.Nodes + + switch { + // If all we have in the ledger is non-Simplex blocks, load the validator set from genesis + case lastNonSimplexInnerBlockHeight + 1 == numBlocks: + validatorSet = genesisValidatorSet + nodes = validatorSetToNodes(genesisValidatorSet) + epochNum = lastNonSimplexInnerBlockHeight + 1 + // If the last block persisted is a sealing block, then we are in the next h. + case lastBlock.SealingBlockInfo() != nil: + epochNum = lastBlock.BlockHeader().Seq + validatorSet = constructValidatorSetFromSealingBlock(lastBlock) + nodes = lastBlock.SealingBlockInfo().ValidatorSet + // Else, we have at least one Simplex block in the ledger, and it's not a sealing block. + default: + // Therefore, the sequence of the sealing block is the h number. + sealingBlockSeq := lastBlock.BlockHeader().Epoch + sealingBlock, _, err := storage.GetBlock(sealingBlockSeq) + if err != nil { + return nil, 0, fmt.Errorf("error retrieving sealing block from storage: %w", err) + } + 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}) + nodes = validatorSetToNodes(validatorSet) + } + return nodes, epochNum, nil +} + +func validatorSetToNodes(genesisValidatorSet metadata.NodeBLSMappings) common.Nodes { + var nodes common.Nodes + for _, vdr := range genesisValidatorSet { + nodes = append(nodes, common.Node{ + Id: vdr.NodeID[:], + Weight: vdr.Weight, + PK: vdr.BLSKey, + }) + } + return nodes +} + +func constructValidatorSetFromSealingBlock(lastBlock ParsedBlock) metadata.NodeBLSMappings { + var validatorSet metadata.NodeBLSMappings + vdrs := lastBlock.Metadata.SimplexEpochInfo.BlockValidationDescriptor.AggregatedMembership.Members + for _, vdr := range vdrs { + validatorSet = append(validatorSet, metadata.NodeBLSMapping{ + NodeID: vdr.NodeID, + BLSKey: vdr.BLSKey, + Weight: vdr.Weight, + }) + } + return validatorSet +} diff --git a/instance_test.go b/instance_test.go new file mode 100644 index 00000000..37db9a4f --- /dev/null +++ b/instance_test.go @@ -0,0 +1,722 @@ +// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved. +// See the file LICENSE for licensing terms. + +package simplex + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/sha256" + "encoding/asn1" + "encoding/binary" + "sort" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/ava-labs/simplex/common" + metadata "github.com/ava-labs/simplex/msm" + "github.com/ava-labs/simplex/testutil" + "github.com/ava-labs/simplex/wal" + + "github.com/stretchr/testify/require" +) + +func TestInstance(t *testing.T) { + // Two-validator last epoch, end to end: + // - The node under test bootstraps from a non-Simplex genesis block and + // commits a series of blocks alone in the first (single-validator) epoch. + // - It observes a validator-set change on the P-chain that both bumps the + // weight and adds a second validator, then seals the epoch and enters the + // two-validator epoch. + // - A second, real Instance (the peer) is started from a copy of the node's + // ledger, and the two validators drive the two-validator epoch together + // over an in-memory network: block proposals are delivered through + // HandleBlockMessage and votes/finalizations through HandleMessage. + // + // Sealing the epoch needs a quorum of approvals of the new (two-validator) set. + // The peer is not running yet during epoch 1, so its approval is injected + // directly into the node's approval store (HandleApproval) until the sealing + // block appears. + const ( + basePChainHeight = uint64(1) + epochChangePChainHeight = uint64(100) + ) + + var id [20]byte + rand.Read(id[:]) + nodeID := common.NodeID(id[:]) + + // The peer that joins the validator set in the last epoch. Its ID is chosen + // to differ from the (random) node under test. + var peerID [20]byte + rand.Read(peerID[:]) + peerNodeID := common.NodeID(peerID[:]) + + // Epoch 1 is single-validator (just the node under test). The last epoch is + // expanded to two validators (the node + the peer). + validatorSetsAtHeight := map[uint64]metadata.NodeBLSMappings{ + basePChainHeight: { + {NodeID: id, BLSKey: []byte{0xaa}, Weight: 1}, + }, + epochChangePChainHeight: { + {NodeID: id, BLSKey: []byte{0xaa}, Weight: 2}, + {NodeID: peerID, BLSKey: []byte{0xbb}, Weight: 2}, + }, + } + + pc := newTestPlatformChain(basePChainHeight, validatorSetsAtHeight) + cops := &testCryptoOps{} + + // The last non-Simplex block: VM height 1, so the zero (first Simplex) block + // is built at storage seq 1 on top of the genesis at seq 0. + lastNonSimplex := &testInnerBlock{Height_: 0, TS: time.Now(), Payload: []byte("genesis")} + + // net routes messages between the two instances asynchronously (see + // inMemNetwork) so proposals reach HandleBlockMessage and everything else + // reaches HandleMessage. + net := newInMemNetwork(t) + t.Cleanup(net.stop) + + // newInstance builds an Instance sharing the common test dependencies but with + // its own ID, storage and VM, wired to the in-memory network. Each instance + // has an independent VM: the inner-block heights it produces are carried + // verbatim into blocks and never cross-checked, so they need not agree. + newInstance := func(logID int, nodeID common.NodeID, storage *MockStorage) *Instance { + return &Instance{ + Config: Config{ + Logger: testutil.MakeLogger(t, logID), + ID: nodeID, + VM: newTestVM(), + Storage: storage, + Sender: net.senderFor(nodeID), + PlatformChain: pc, + CryptoOps: cops, + LastNonSimplexInnerBlock: lastNonSimplex, + ParameterConfig: ParameterConfig{ + MaxNetworkDelay: 500 * time.Millisecond, + MaxRoundWindow: 100, + WALMaxEntryCount: 1024, + }, + }, + } + } + + storage := NewMockStorage(t) + smb := metadata.StateMachineBlock{InnerBlock: lastNonSimplex} + require.NoError(t, storage.Index(context.Background(), &ParsedBlock{StateMachineBlock: smb}, common.Finalization{})) + + firstInstance := newInstance(0, nodeID, storage) + net.register(nodeID, firstInstance) + require.NoError(t, firstInstance.Start()) + t.Cleanup(firstInstance.Stop) + + // Epoch 1: wait until the node has bootstrapped (genesis + zero block) and + // committed a series of normal blocks on its own. + const epoch1Target = uint64(5) // genesis(0) + zero block(1) + 3 normal blocks + waitForNumBlocks(t, storage, epoch1Target) + + epoch1Blocks := storage.NumBlocks() + require.GreaterOrEqual(t, epoch1Blocks, epoch1Target) + + // The validator set in force is the one introduced by the most recent block + // that carries a BlockValidationDescriptor (the zero block in epoch 1). + require.Equal(t, uint64(1), latestValidatorWeight(t, storage), "epoch 1 should use the original validator weight") + + // Trigger the epoch change: the validator set changes at epochChangePChainHeight, + // growing from one validator to two. + pc.advanceTo(epochChangePChainHeight) + approval := &metadata.ValidatorSetApproval{ + NodeID: peerID, + PChainHeight: epochChangePChainHeight, + AuxInfoDigest: sha256.Sum256(nil), + Signature: []byte{1, 2, 3}, + } + + // The node seals the epoch once it has a quorum of approvals of the new + // (two-validator) set. With two validators the node's self-approval is no longer + // a quorum and the peer is not running yet, so waitForSealingBlock injects the + // peer's approval on each poll until the sealing block is committed. + waitForSealingBlock(t, firstInstance, approval, epoch1Blocks) + + // The node is now in the two-validator epoch but cannot make progress alone + // (a single vote/empty-vote is not a quorum). Start the real peer from a copy + // of the node's ledger so the two validators drive the epoch together. + peerStorage := cloneStorageDeep(t, storage) + secondInstance := newInstance(1, peerNodeID, peerStorage) + net.register(peerNodeID, secondInstance) + require.NoError(t, secondInstance.Start()) + t.Cleanup(secondInstance.Stop) + + // With both validators live, the two-validator epoch commits more blocks. + sealedAt := storage.NumBlocks() + const epoch2Extra = uint64(3) + waitForNumBlocks(t, storage, sealedAt+epoch2Extra) + + // Confirm the post-change validator set has the new weight and now contains + // two validators. + require.Equal(t, uint64(2), latestValidatorWeight(t, storage), "epoch 2 should use the new validator weight") + require.Equal(t, 2, latestValidatorSetSize(t, storage), "the last epoch should have two validators") +} + +func latestValidatorWeight(t *testing.T, storage *MockStorage) uint64 { + t.Helper() + num := storage.NumBlocks() + for seq := int64(num) - 1; seq >= 0; seq-- { + block, ok := storage.blockSnapshot(uint64(seq)) + if !ok { + continue + } + bvd := block.Metadata.SimplexEpochInfo.BlockValidationDescriptor + if bvd != nil && len(bvd.AggregatedMembership.Members) > 0 { + return bvd.AggregatedMembership.Members[0].Weight + } + } + t.Fatalf("no block with a BlockValidationDescriptor found in storage") + return 0 +} + +// latestValidatorSetSize returns the number of validators recorded in the most +// recent block that carries a BlockValidationDescriptor. +func latestValidatorSetSize(t *testing.T, storage *MockStorage) int { + t.Helper() + num := storage.NumBlocks() + for seq := int64(num) - 1; seq >= 0; seq-- { + block, ok := storage.blockSnapshot(uint64(seq)) + if !ok { + continue + } + bvd := block.Metadata.SimplexEpochInfo.BlockValidationDescriptor + if bvd != nil && len(bvd.AggregatedMembership.Members) > 0 { + return len(bvd.AggregatedMembership.Members) + } + } + t.Fatalf("no block with a BlockValidationDescriptor found in storage") + return 0 +} + +func waitForNumBlocks(t *testing.T, storage *MockStorage, target uint64) { + t.Helper() + deadline := time.Now().Add(20 * time.Second) + for time.Now().Before(deadline) { + if storage.NumBlocks() >= target { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("timed out waiting for %d blocks, have %d", target, storage.NumBlocks()) +} + +// waitForSealingBlock waits until a two-validator sealing block (a block carrying a +// BlockValidationDescriptor with the new weight) is committed at or after fromSeq. +// +// Sealing needs a quorum of approvals of the new (two-validator) set, but the peer +// is not running yet, so on every poll it injects the peer's approval into the +// node's approval store. Injecting before the store is initialized is a harmless +// no-op, and re-injecting an existing approval is deduplicated, so this simply +// guarantees the approval lands before the node seals. +func waitForSealingBlock(t *testing.T, inst *Instance, approval *metadata.ValidatorSetApproval, fromSeq uint64) { + t.Helper() + storage := inst.Config.Storage.(*MockStorage) + deadline := time.Now().Add(20 * time.Second) + for time.Now().Before(deadline) { + inst.lock.Lock() + msm := inst.msm + inst.lock.Unlock() + if msm != nil { + _ = msm.HandleApproval(approval, 1) + } + + num := storage.NumBlocks() + for seq := fromSeq; seq < num; seq++ { + block, ok := storage.blockSnapshot(seq) + if !ok { + continue + } + bvd := block.Metadata.SimplexEpochInfo.BlockValidationDescriptor + if bvd != nil && len(bvd.AggregatedMembership.Members) > 0 && + bvd.AggregatedMembership.Members[0].Weight == 2 { + return + } + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("timed out waiting for sealing block after seq %d", fromSeq) +} + +type testInnerBlock struct { + Height_ uint64 + TS time.Time + Payload []byte +} + +func (b *testInnerBlock) Bytes() ([]byte, error) { + out := make([]byte, 16, 16+len(b.Payload)) + binary.BigEndian.PutUint64(out[0:8], b.Height_) + binary.BigEndian.PutUint64(out[8:16], uint64(b.TS.UnixMilli())) + out = append(out, b.Payload...) + return out, nil +} + +func (b *testInnerBlock) Digest() [32]byte { + bytes, _ := b.Bytes() + return sha256.Sum256(bytes) +} + +func (b *testInnerBlock) Height() uint64 { return b.Height_ } +func (b *testInnerBlock) Timestamp() time.Time { return b.TS } +func (b *testInnerBlock) Verify(context.Context, uint64) error { return nil } + +func parseTestInnerBlock(buff []byte) (*testInnerBlock, error) { + b := &testInnerBlock{} + b.Height_ = binary.BigEndian.Uint64(buff[0:8]) + b.TS = time.UnixMilli(int64(binary.BigEndian.Uint64(buff[8:16]))) + b.Payload = append([]byte(nil), buff[16:]...) + return b, nil +} + +type testVM struct { + nextHeight atomic.Uint64 +} + +func newTestVM() *testVM { + vm := &testVM{} + vm.nextHeight.Store(2) // genesis inner block is height 1 + return vm +} + +func (vm *testVM) BuildBlock(_ context.Context, _ uint64) (metadata.VMBlock, error) { + h := vm.nextHeight.Add(1) - 1 + payload := make([]byte, 8) + binary.BigEndian.PutUint64(payload, h) + return &testInnerBlock{Height_: h, TS: time.Now(), Payload: payload}, nil +} + +func (vm *testVM) WaitForPendingBlock(ctx context.Context) { + select { + case <-ctx.Done(): + case <-time.After(10 * time.Millisecond): + } +} + +func (vm *testVM) ParseBlock(_ context.Context, b []byte) (metadata.VMBlock, error) { + return parseTestInnerBlock(b) +} + +func (vm *testVM) ComputeICMEpoch(input metadata.ICMEpochInput) metadata.ICMEpochInfo { + // ACP-181-style transition (mirrors the msm test helper). + var zero metadata.ICMEpochInfo + if input.ParentEpoch == zero { + return metadata.ICMEpochInfo{ + PChainEpochHeight: input.ParentPChainHeight, + EpochNumber: 1, + EpochStartTime: uint64(input.ParentTimestamp.Unix()), + } + } + endTime := time.Unix(int64(input.ParentEpoch.EpochStartTime), 0).Add(time.Second) + if input.ParentTimestamp.Before(endTime) { + return input.ParentEpoch + } + return metadata.ICMEpochInfo{ + PChainEpochHeight: input.ParentPChainHeight, + EpochNumber: input.ParentEpoch.EpochNumber + 1, + EpochStartTime: uint64(input.ParentTimestamp.Unix()), + } +} + +type testPlatformChain struct { + baseHeight uint64 + validatorSetAtHeight map[uint64]metadata.NodeBLSMappings // height --> validator set + lock sync.Mutex + cond *sync.Cond + height uint64 +} + +func newTestPlatformChain(baseHeight uint64, validatorSetsAtHeight map[uint64]metadata.NodeBLSMappings) *testPlatformChain { + pc := &testPlatformChain{ + baseHeight: baseHeight, + validatorSetAtHeight: validatorSetsAtHeight, + height: baseHeight, + } + pc.cond = sync.NewCond(&pc.lock) + return pc +} + +func (pc *testPlatformChain) advanceTo(h uint64) { + pc.lock.Lock() + defer pc.lock.Unlock() + pc.height = h + pc.cond.Broadcast() // wake any WaitForProgress waiters +} + +func (pc *testPlatformChain) currentHeight() uint64 { + pc.lock.Lock() + defer pc.lock.Unlock() + return pc.height +} + +func (pc *testPlatformChain) validatorSet(height uint64) metadata.NodeBLSMappings { + heights := make([]uint64, 0, len(pc.validatorSetAtHeight)) + for h := range pc.validatorSetAtHeight { + heights = append(heights, h) + } + sort.Slice(heights, func(i, j int) bool { return heights[i] < heights[j] }) + + var lastCheckpoint uint64 + for _, h := range heights { + if h > height { + break + } + lastCheckpoint = h + } + return pc.validatorSetAtHeight[lastCheckpoint] +} + +func (pc *testPlatformChain) GetValidatorSet(height uint64) (metadata.NodeBLSMappings, error) { + return pc.validatorSet(height), nil +} + +func (pc *testPlatformChain) GenesisValidatorSet() metadata.NodeBLSMappings { + return pc.validatorSet(pc.baseHeight) +} + +func (pc *testPlatformChain) GetMinimumHeight(context.Context) (uint64, error) { + return pc.currentHeight(), nil +} + +func (pc *testPlatformChain) GetCurrentHeight(context.Context) (uint64, error) { + return pc.currentHeight(), nil +} + +func (pc *testPlatformChain) WaitForProgress(ctx context.Context, pChainHeight uint64) error { + // sync.Cond has no notion of cancellation, so wake the waiter when the context + // is done and let the loop below observe ctx.Err(). + stop := pc.signalWhenContextFinished(ctx) + defer stop() + + pc.lock.Lock() + defer pc.lock.Unlock() + for pc.height == pChainHeight { + if err := ctx.Err(); err != nil { + return err + } + pc.cond.Wait() + } + return nil +} + +func (pc *testPlatformChain) signalWhenContextFinished(ctx context.Context) func() bool { + stop := context.AfterFunc(ctx, func() { + pc.lock.Lock() + defer pc.lock.Unlock() + pc.cond.Broadcast() + }) + return stop +} + +func (pc *testPlatformChain) LastNonSimplexBlockPChainHeight() uint64 { + return pc.baseHeight +} + +// --------------------------------------------------------------------------- +// testCryptoOps: permissive crypto that accepts all signatures and forms +// quorums based on node count (reusing testutil aggregation/QC helpers). +// --------------------------------------------------------------------------- + +type testCryptoOps struct{} + +func (c *testCryptoOps) Sign(message []byte) ([]byte, error) { + // A deterministic, non-empty placeholder signature. + d := sha256.Sum256(message) + return d[:], nil +} + +func (c *testCryptoOps) AggregateKeys(keys ...[]byte) ([]byte, error) { + var out []byte + for _, k := range keys { + out = append(out, k...) + } + return out, nil +} + +func (c *testCryptoOps) VerifySignature(_ []byte, _ []byte, _ []byte) error { + return nil +} + +func (c *testCryptoOps) CreateSignatureAggregator(nodes []common.Node) common.SignatureAggregator { + return &testutil.TestSignatureAggregator{N: len(nodes)} +} + +func (c *testCryptoOps) DeserializeQuorumCertificate(bytes []byte) (common.QuorumCertificate, error) { + var qc []common.Signature + if _, err := asn1.Unmarshal(bytes, &qc); err != nil { + return nil, err + } + return testutil.TestQC(qc), nil +} + +// --------------------------------------------------------------------------- +// MockStorage: in-memory Storage with a real (in-memory) WAL. +// --------------------------------------------------------------------------- + +type MockStorage struct { + t *testing.T + *testutil.InMemStorage + + // snapLock guards snapshots. Each committed block is snapshotted (its metadata + // copied into an object the instance never touches again) at Index time, so the + // test's polling helpers and the peer's storage clone can read block metadata + // without racing on the block's canoto digest cache — which the running + // instance keeps mutating as it re-digests blocks (e.g. as proposal parents). + snapLock sync.Mutex + snapshots map[uint64]storedBlock +} + +type storedBlock struct { + encoded []byte + fin common.Finalization +} + +func NewMockStorage(t *testing.T) *MockStorage { + return &MockStorage{ + t: t, + InMemStorage: testutil.NewInMemStorage(), + snapshots: make(map[uint64]storedBlock), + } +} + +func (m *MockStorage) Index(ctx context.Context, block common.VerifiedBlock, certificate common.Finalization) error { + // Serialize the block on the committing goroutine (where it isn't racing with + // the instance's own use of it) and keep only the bytes. Test readers then + // reconstruct fully independent block objects from these bytes rather than + // touching the instance's live block, whose canoto digest cache — including the + // pointer-shared BlockValidationDescriptor — the instance keeps mutating. + encoded, err := block.Bytes() + if err != nil { + return err + } + seq := m.InMemStorage.NumBlocks() + m.snapLock.Lock() + m.snapshots[seq] = storedBlock{encoded: encoded, fin: certificate} + m.snapLock.Unlock() + return m.InMemStorage.Index(ctx, block, certificate) +} + +func (m *MockStorage) GetBlock(seq uint64) (metadata.StateMachineBlock, *common.Finalization, error) { + vb, f, err := m.InMemStorage.Retrieve(seq) + if err != nil { + return metadata.StateMachineBlock{}, nil, err + } + return vb.(*ParsedBlock).StateMachineBlock, &f, nil +} + +// blockSnapshot reconstructs an independent copy of the block at seq from its +// stored bytes. Test-only readers use it instead of GetBlock so they never touch +// the instance's live block objects (whose canoto digest cache the instance keeps +// mutating). +func (m *MockStorage) blockSnapshot(seq uint64) (metadata.StateMachineBlock, bool) { + m.snapLock.Lock() + sb, ok := m.snapshots[seq] + m.snapLock.Unlock() + if !ok { + return metadata.StateMachineBlock{}, false + } + return m.parseStored(sb.encoded), true +} + +// parseStored reconstructs a StateMachineBlock from the wire bytes recorded at +// Index. Each call yields a fresh object with no pointers shared with any other +// block, so concurrent readers never race. +func (m *MockStorage) parseStored(encoded []byte) metadata.StateMachineBlock { + raw := &RawBlock{} + require.NoError(m.t, raw.UnmarshalCanoto(encoded)) + var inner metadata.VMBlock + if len(raw.InnerBlockBytes) > 0 { + parsed, err := parseTestInnerBlock(raw.InnerBlockBytes) + require.NoError(m.t, err) + inner = parsed + } + return metadata.StateMachineBlock{InnerBlock: inner, Metadata: raw.Metadata} +} + +func (m *MockStorage) CreateWAL() (wal.DeletableWAL, error) { + return testutil.NewTestWAL(m.t), nil +} + +// cloneStorageDeep copies src into a fresh MockStorage holding independent block +// objects, reconstructed from the bytes src recorded at Index. It never touches +// the source instance's live block objects (which would race on their canoto +// digest cache), and the peer gets its own object per block so its instance +// re-digests objects independent of the source node's. +func cloneStorageDeep(t *testing.T, src *MockStorage) *MockStorage { + dst := NewMockStorage(t) + num := src.NumBlocks() + for seq := uint64(0); seq < num; seq++ { + src.snapLock.Lock() + sb, ok := src.snapshots[seq] + src.snapLock.Unlock() + require.True(t, ok, "missing snapshot for committed block %d", seq) + + pb := &ParsedBlock{StateMachineBlock: src.parseStored(sb.encoded)} + require.NoError(t, dst.Index(context.Background(), pb, sb.fin)) + } + return dst +} + +// --------------------------------------------------------------------------- +// inMemNetwork: a tiny in-memory network that routes messages between Instances. +// Delivery happens on a per-node goroutine rather than inline in Send, because +// Send is invoked while the sending epoch holds its lock; delivering inline +// (which would take the destination's lock, whose epoch may in turn Send back) +// could deadlock. Block proposals are delivered through HandleBlockMessage and +// all other messages through HandleMessage, matching the real inbound paths. +// --------------------------------------------------------------------------- + +type netMsg struct { + from common.NodeID + // Exactly one of {msg} or {rawBlock,vote} is set. Proposals are serialized to + // rawBlock eagerly in Send (under the sender's epoch lock) so the delivery + // goroutine never touches the sender's live block object, which would race on + // its lazily-populated canoto digest cache. + msg *common.Message + rawBlock *RawBlock + vote *common.Vote +} + +type netNode struct { + inst *Instance + // in is a buffered inbox drained by the delivery goroutine. The channel itself + // signals that work is available, so no separate wake signal is needed. Sends + // never block (see enqueue); on the rare chance the buffer fills, a dropped + // message costs at most an empty round the epoch recovers from. + in chan netMsg + done chan struct{} + stopped chan struct{} +} + +type inMemNetwork struct { + t *testing.T + lock sync.Mutex + nodes map[string]*netNode +} + +func newInMemNetwork(t *testing.T) *inMemNetwork { + return &inMemNetwork{t: t, nodes: make(map[string]*netNode)} +} + +// senderFor returns the Sender an instance uses to broadcast; it enqueues to the +// destination's delivery goroutine (skipping the loopback copy to itself). +func (n *inMemNetwork) senderFor(self common.NodeID) Sender { + return &networkSender{net: n, self: self} +} + +// register wires inst into the network and starts delivering messages to it. +// Messages that arrive before the epoch exists are dropped by the instance's +// nil-epoch guard, which at worst costs a few empty rounds the epoch recovers from. +func (n *inMemNetwork) register(id common.NodeID, inst *Instance) { + node := &netNode{ + inst: inst, + in: make(chan netMsg, 1024), + done: make(chan struct{}), + stopped: make(chan struct{}), + } + n.lock.Lock() + n.nodes[string(id)] = node + n.lock.Unlock() + go n.deliver(node) +} + +func (n *inMemNetwork) stop() { + n.lock.Lock() + nodes := make([]*netNode, 0, len(n.nodes)) + for _, node := range n.nodes { + nodes = append(nodes, node) + } + n.nodes = make(map[string]*netNode) + n.lock.Unlock() + for _, node := range nodes { + close(node.done) + <-node.stopped + } +} + +func (n *inMemNetwork) enqueue(dest common.NodeID, m netMsg) { + n.lock.Lock() + node := n.nodes[string(dest)] + n.lock.Unlock() + if node == nil { + // Destination not registered; drop. This only happens before an instance + // is registered, never mid-run. + return + } + select { + case node.in <- m: + default: + // Never block the sender (Send runs under the epoch lock). A dropped message + // costs at most an empty round the epoch recovers from. + } +} + +func (n *inMemNetwork) deliver(node *netNode) { + defer close(node.stopped) + for { + select { + case <-node.done: + return + case m := <-node.in: + n.dispatch(node.inst, m) + } + } +} + +func (n *inMemNetwork) dispatch(inst *Instance, m netMsg) { + // Proposals are delivered through HandleBlockMessage (which re-parses the raw + // block), matching the real inbound path; everything else via HandleMessage. + if m.rawBlock != nil { + if err := inst.HandleBlockMessage(context.Background(), m.rawBlock, m.vote, m.from); err != nil { + n.t.Logf("HandleBlockMessage from %x failed: %v", m.from, err) + } + return + } + if err := inst.HandleMessage(m.msg, m.from); err != nil { + n.t.Logf("HandleMessage from %x failed: %v", m.from, err) + } +} + +// 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 { + bytes, err := vb.Bytes() + require.NoError(t, err) + raw := &RawBlock{} + require.NoError(t, raw.UnmarshalCanoto(bytes)) + return raw +} + +type networkSender struct { + net *inMemNetwork + self common.NodeID +} + +func (s *networkSender) Send(msg *common.Message, dest common.NodeID) { + // An instance broadcasts to every validator including itself; ignore the + // loopback copy. + if bytes.Equal(s.self, dest) { + return + } + m := netMsg{from: s.self} + // Send runs under the sender's epoch lock, so serializing the proposal block + // here is ordered with the sender's own accesses to it. Shipping the encoded + // bytes (rather than the live block) keeps the delivery goroutine off the + // sender's block object. + if vbm := msg.VerifiedBlockMessage; vbm != nil { + vote := vbm.Vote + m.rawBlock = toRawBlock(s.net.t, vbm.VerifiedBlock) + m.vote = &vote + } else { + m.msg = msg + } + s.net.enqueue(dest, m) +} diff --git a/msm/approvals.go b/msm/approvals.go index 5c6fe3cb..6b666988 100644 --- a/msm/approvals.go +++ b/msm/approvals.go @@ -5,6 +5,7 @@ package metadata import ( "fmt" + "sync" "github.com/ava-labs/simplex/common" "go.uber.org/zap" @@ -27,8 +28,12 @@ type ApprovalStore struct { validators NodeBLSMappings logger common.Logger pkByNodeID map[nodeID][]byte - approvalsByNodes map[nodeID]approvalsByPChainHeightAndAuxInfoDigest - storedCount int + // lock guards the mutable state below (approvalsByNodes, storedCount). The + // store is accessed concurrently: approvals are handled as they arrive while + // the block builder reads the accumulated approvals when building a block. + lock sync.Mutex + approvalsByNodes map[nodeID]approvalsByPChainHeightAndAuxInfoDigest + storedCount int } func NewApprovalStore(signatureVerifier SignatureVerifier, validators NodeBLSMappings, logger common.Logger) *ApprovalStore { @@ -52,6 +57,9 @@ func NewApprovalStore(signatureVerifier SignatureVerifier, validators NodeBLSMap } func (as *ApprovalStore) Approvals() ValidatorSetApprovals { + as.lock.Lock() + defer as.lock.Unlock() + approvals := make(ValidatorSetApprovals, 0, as.storedCount) for _, approvalsByHeight := range as.approvalsByNodes { for _, approval := range approvalsByHeight { @@ -70,17 +78,21 @@ func (as *ApprovalStore) HandleApproval(approval *ValidatorSetApproval, timestam return nil } - // Second thing we check is if we already have an approval for this height from this node. - if as.approvalExistsAndUpToDate(approval, timestamp) { - as.logger.Debug("Already have an approval from the node", zap.String("nodeID", + // Second thing we check is if the signature of the approval is valid. + // We need it to be valid in order for nodes to be able to aggregate it later on along with other approvals. + // This is checked before taking the lock, as it only reads immutable state. + if err := as.checkApprovalSignature(approval, pk); err != nil { + as.logger.Debug("Received an approval with an invalid signature", zap.String("nodeID", fmt.Sprintf("%x", approval.NodeID)), zap.Uint64("pChainHeight", approval.PChainHeight)) return nil } - // Third thing we check is if the signature of the approval is valid. - // We need it to be valid in order for nodes to be able to aggregate it later on along with other approvals. - if err := as.checkApprovalSignature(approval, pk); err != nil { - as.logger.Debug("Received an approval with an invalid signature", zap.String("nodeID", + as.lock.Lock() + defer as.lock.Unlock() + + // Third thing we check is if we already have an approval for this height from this node. + if as.approvalExistsAndUpToDate(approval, timestamp) { + as.logger.Debug("Already have an approval from the node", zap.String("nodeID", fmt.Sprintf("%x", approval.NodeID)), zap.Uint64("pChainHeight", approval.PChainHeight)) return nil } @@ -167,3 +179,28 @@ func (as *ApprovalStore) approvalExistsAndUpToDate(approval *ValidatorSetApprova return existingApproval.Timestamp >= timestamp } + +func (as *ApprovalStore) PutApprovals(approvalStore *ApprovalStore) { + // Snapshot the approvals under our lock, then hand them to the destination + // store (which takes its own lock). Copying first avoids holding two store + // locks at once. + as.lock.Lock() + type approvalWithTimestamp struct { + approval ValidatorSetApproval + timestamp uint64 + } + snapshot := make([]approvalWithTimestamp, 0, as.storedCount) + for _, approvalsByHeight := range as.approvalsByNodes { + for _, approval := range approvalsByHeight { + snapshot = append(snapshot, approvalWithTimestamp{ + approval: approval.ValidatorSetApproval, + timestamp: approval.Timestamp, + }) + } + } + as.lock.Unlock() + + for _, a := range snapshot { + approvalStore.HandleApproval(&a.approval, a.timestamp) + } +} diff --git a/msm/approvals_test.go b/msm/approvals_test.go index c24e5f37..865bf6c7 100644 --- a/msm/approvals_test.go +++ b/msm/approvals_test.go @@ -222,6 +222,152 @@ func TestApprovalStoreHandleApproval(t *testing.T) { } } +func TestApprovalStorePutApprovals(t *testing.T) { + // findApproval locates the approval for a given (NodeID, PChainHeight) in a slice returned by + // Approvals(). It is used to assert which of two competing approvals (source vs. destination) + // survived a merge, since Approvals() does not carry the timestamp but does carry the Signature, + // and signApproval produces a distinct signature per call. + findApproval := func(got ValidatorSetApprovals, node nodeID, height uint64) (ValidatorSetApproval, bool) { + for _, a := range got { + if a.NodeID == node && a.PChainHeight == height { + return a, true + } + } + return ValidatorSetApproval{}, false + } + + for _, tc := range []struct { + name string + // srcValidators/dstValidators size the two stores via makeValidators; NodeIDs are makeNodeID(i+1). + srcValidators int + dstValidators int + // srcApprovals are loaded into the source store; dstApprovals are pre-loaded into the destination + // store before PutApprovals is called. + srcApprovals []approvalAndTimestamp + dstApprovals []approvalAndTimestamp + // verify asserts on the destination (and source) store after src.PutApprovals(dst). + verify func(t *testing.T, dst, src *ApprovalStore, srcSent, dstSent []approvalAndTimestamp) + }{ + { + // Copies every source approval into an empty destination that shares the same validator set, + // and leaves the source store untouched (PutApprovals copies, it does not move). + name: "copies all approvals into empty destination", + srcValidators: 3, + dstValidators: 3, + srcApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 7, Signature: signApproval(7, [32]byte{})}, 100}, + {ValidatorSetApproval{NodeID: makeNodeID(2), PChainHeight: 8, Signature: signApproval(8, [32]byte{})}, 100}, + }, + verify: func(t *testing.T, dst, src *ApprovalStore, srcSent, _ []approvalAndTimestamp) { + got := dst.Approvals() + require.Len(t, got, 2) + require.Equal(t, 2, dst.storedCount) + require.ElementsMatch(t, []ValidatorSetApproval{srcSent[0].ValidatorSetApproval, srcSent[1].ValidatorSetApproval}, got) + // The source store is unchanged. + require.Len(t, src.Approvals(), 2) + require.Equal(t, 2, src.storedCount) + }, + }, + { + // Approvals from nodes absent from the destination's (smaller) validator set are dropped on + // carry-over. This is the epoch-change case: a validator that is not part of the new set does + // not have its approval carried into the new store. + name: "drops approvals from nodes not in destination validator set", + srcValidators: 3, + dstValidators: 2, + srcApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 1, Signature: signApproval(1, [32]byte{})}, 10}, + {ValidatorSetApproval{NodeID: makeNodeID(2), PChainHeight: 1, Signature: signApproval(1, [32]byte{})}, 10}, + {ValidatorSetApproval{NodeID: makeNodeID(3), PChainHeight: 1, Signature: signApproval(1, [32]byte{})}, 10}, + }, + verify: func(t *testing.T, dst, _ *ApprovalStore, _, _ []approvalAndTimestamp) { + got := dst.Approvals() + require.Len(t, got, 2, "only approvals from nodes in the destination set carry over") + require.Equal(t, 2, dst.storedCount) + _, ok := findApproval(got, makeNodeID(3), 1) + require.False(t, ok, "node 3 is not in the destination validator set") + }, + }, + { + // Merges into a non-empty destination: a pre-existing destination approval with a newer + // timestamp is kept over an older one carried from the source, while a brand-new node's + // approval from the source is added. + name: "keeps newer destination approval and adds new node", + srcValidators: 2, + dstValidators: 2, + dstApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 7, Signature: signApproval(7, [32]byte{})}, 200}, // newer, already present + }, + srcApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 7, Signature: signApproval(7, [32]byte{})}, 100}, // older, must not overwrite + {ValidatorSetApproval{NodeID: makeNodeID(2), PChainHeight: 7, Signature: signApproval(7, [32]byte{})}, 100}, // new node, added + }, + verify: func(t *testing.T, dst, _ *ApprovalStore, _, dstSent []approvalAndTimestamp) { + got := dst.Approvals() + require.Len(t, got, 2) + require.Equal(t, 2, dst.storedCount) + kept, ok := findApproval(got, makeNodeID(1), 7) + require.True(t, ok) + require.Equal(t, dstSent[0].ValidatorSetApproval, kept, "the newer destination approval is retained") + _, ok = findApproval(got, makeNodeID(2), 7) + require.True(t, ok, "the new node's approval is carried over") + }, + }, + { + // A source approval with a newer timestamp replaces the stale destination approval at the + // same (NodeID, PChainHeight). + name: "newer source approval replaces stale destination approval", + srcValidators: 2, + dstValidators: 2, + dstApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 7, Signature: signApproval(7, [32]byte{})}, 100}, // older, present + }, + srcApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 7, Signature: signApproval(7, [32]byte{})}, 200}, // newer, replaces + }, + verify: func(t *testing.T, dst, _ *ApprovalStore, srcSent, _ []approvalAndTimestamp) { + got := dst.Approvals() + require.Len(t, got, 1) + require.Equal(t, 1, dst.storedCount) + require.Equal(t, srcSent[0].ValidatorSetApproval, got[0], "the newer source approval replaces the stale one") + }, + }, + { + // An empty source store is a no-op: the destination is left exactly as it was. + name: "empty source is a no-op", + srcValidators: 2, + dstValidators: 2, + dstApprovals: []approvalAndTimestamp{ + {ValidatorSetApproval{NodeID: makeNodeID(1), PChainHeight: 1, Signature: signApproval(1, [32]byte{})}, 10}, + }, + verify: func(t *testing.T, dst, _ *ApprovalStore, _, dstSent []approvalAndTimestamp) { + got := dst.Approvals() + require.Len(t, got, 1) + require.Equal(t, 1, dst.storedCount) + require.Equal(t, dstSent[0].ValidatorSetApproval, got[0]) + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + src := NewApprovalStore(&signatureVerifier{}, makeValidators(tc.srcValidators), testutil.MakeLogger(t)) + dst := NewApprovalStore(&signatureVerifier{}, makeValidators(tc.dstValidators), testutil.MakeLogger(t)) + + for _, a := range tc.srcApprovals { + require.NoError(t, src.HandleApproval(&a.ValidatorSetApproval, a.Timestamp)) + } + for _, a := range tc.dstApprovals { + require.NoError(t, dst.HandleApproval(&a.ValidatorSetApproval, a.Timestamp)) + } + + src.PutApprovals(dst) + + // storedCount must always stay in sync with the number of retrievable approvals. + require.Len(t, dst.Approvals(), dst.storedCount) + tc.verify(t, dst, src, tc.srcApprovals, tc.dstApprovals) + }) + } +} + func TestApprovalStoreHandleApprovalStoredCountStaysConsistent(t *testing.T) { // Runs a mixed workload (insert, duplicate, replace, new height, prune) // and asserts that storedCount equals len(Approvals()) after every step. diff --git a/msm/build_decision.go b/msm/build_decision.go index cfac2d98..512a62cf 100644 --- a/msm/build_decision.go +++ b/msm/build_decision.go @@ -18,6 +18,7 @@ type blockBuildingDecision struct { buildInnerBlock bool transitionEpoch bool pChainHeight uint64 + validatorSet NodeBLSMappings } // PChainProgressListener listens for changes in the P-chain height. @@ -33,8 +34,8 @@ type blockBuildingDecider struct { waitForPendingBlock func(ctx context.Context) // hasValidatorSetChanged should return whether the validator set has changed since the // P-chain height referenced by the last block in the chain and until the provided P-chain height. - hasValidatorSetChanged func(pChainHeight uint64) (bool, error) - getPChainHeight func() uint64 + hasValidatorSetChanged func(pChainHeight uint64) (bool, NodeBLSMappings, error) + getPChainHeight func(ctx context.Context) (uint64, error) } // shouldBuildBlock determines whether we should build a block at the current time, @@ -46,16 +47,19 @@ func (bbd *blockBuildingDecider) shouldBuildBlock( ctx context.Context, ) (blockBuildingDecision, error) { for { - pChainHeight := bbd.getPChainHeight() + pChainHeight, err := bbd.getPChainHeight(ctx) + if err != nil { + return blockBuildingDecision{}, err + } - shouldTransitionEpoch, err := bbd.hasValidatorSetChanged(pChainHeight) + shouldTransitionEpoch, newValidatorSet, err := bbd.hasValidatorSetChanged(pChainHeight) if err != nil { return blockBuildingDecision{}, err } if shouldTransitionEpoch { // If we should transition to a new epoch, maybe we can also build a block along the way. - return bbd.buildBlockWithEpochTransition(ctx, pChainHeight) + return bbd.buildBlockWithEpochTransition(ctx, pChainHeight, newValidatorSet) } // Else, we don't need to transition to a new epoch, but maybe we should build a block. @@ -71,7 +75,11 @@ func (bbd *blockBuildingDecider) shouldBuildBlock( // If the P-chain height changed, re-evaluate again whether we should transition to a new epoch, // or continue waiting to build a block. - if bbd.getPChainHeight() != pChainHeight { + h, err := bbd.getPChainHeight(ctx) + if err != nil { + return blockBuildingDecision{}, err + } + if h != pChainHeight { continue } @@ -109,7 +117,7 @@ func (bbd *blockBuildingDecider) waitForPChainChangeOrPendingBlock(ctx context.C // It waits up to a limited amount of time (bbd.maxBlockBuildingWaitTime) for a block to be ready to be built, // and if no block is ready by then, it returns the decision to transition epoch without building a block. // Otherwise, it returns the decision to build a block and transition epoch along the way. -func (bbd *blockBuildingDecider) buildBlockWithEpochTransition(ctx context.Context, pChainHeight uint64) (blockBuildingDecision, error) { +func (bbd *blockBuildingDecider) buildBlockWithEpochTransition(ctx context.Context, pChainHeight uint64, validatorSet NodeBLSMappings) (blockBuildingDecision, error) { impatientContext, cancel := context.WithTimeout(ctx, bbd.maxBlockBuildingWaitTime) defer cancel() @@ -124,9 +132,9 @@ func (bbd *blockBuildingDecider) buildBlockWithEpochTransition(ctx context.Conte if impatientContext.Err() != nil { // We have returned from waitForPendingBlock because impatientContext has timed out, // which means we don't need to build a block. - return blockBuildingDecision{transitionEpoch: true, pChainHeight: pChainHeight}, nil + return blockBuildingDecision{transitionEpoch: true, pChainHeight: pChainHeight, validatorSet: validatorSet}, nil } // Block is ready to be built - return blockBuildingDecision{buildInnerBlock: true, transitionEpoch: true, pChainHeight: pChainHeight}, nil + return blockBuildingDecision{buildInnerBlock: true, transitionEpoch: true, pChainHeight: pChainHeight, validatorSet: validatorSet}, nil } diff --git a/msm/build_decision_test.go b/msm/build_decision_test.go index b1b13402..33694802 100644 --- a/msm/build_decision_test.go +++ b/msm/build_decision_test.go @@ -31,8 +31,8 @@ func TestShouldBuildBlock_VMSignalsBlock(t *testing.T) { }, }, waitForPendingBlock: func(ctx context.Context) {}, - hasValidatorSetChanged: func(uint64) (bool, error) { return false, nil }, - getPChainHeight: func() uint64 { return 100 }, + hasValidatorSetChanged: func(uint64) (bool, NodeBLSMappings, error) { return false, nil, nil }, + getPChainHeight: func(context.Context) (uint64, error) { return 100, nil }, } decision, err := bbd.shouldBuildBlock(t.Context()) @@ -55,8 +55,8 @@ func TestShouldBuildBlock_ContextCanceled(t *testing.T) { cancel() <-ctx.Done() }, - hasValidatorSetChanged: func(uint64) (bool, error) { return false, nil }, - getPChainHeight: func() uint64 { return 100 }, + hasValidatorSetChanged: func(uint64) (bool, NodeBLSMappings, error) { return false, nil, nil }, + getPChainHeight: func(context.Context) (uint64, error) { return 100, nil }, } decision, err := bbd.shouldBuildBlock(ctx) @@ -85,10 +85,10 @@ func TestShouldBuildBlock_PChainHeightChangeTriggersEpochTransition(t *testing.T waitForPendingBlock: func(ctx context.Context) { <-ctx.Done() }, - hasValidatorSetChanged: func(height uint64) (bool, error) { - return height == 200, nil + hasValidatorSetChanged: func(height uint64) (bool, NodeBLSMappings, error) { + return height == 200, nil, nil }, - getPChainHeight: func() uint64 { return pChainHeight.Load() }, + getPChainHeight: func(context.Context) (uint64, error) { return pChainHeight.Load(), nil }, } decision, err := bbd.shouldBuildBlock(t.Context()) @@ -122,8 +122,8 @@ func TestShouldBuildBlock_PChainHeightChangeButNoEpochTransition(t *testing.T) { <-ctx.Done() } }, - hasValidatorSetChanged: func(uint64) (bool, error) { return false, nil }, - getPChainHeight: func() uint64 { return pChainHeight.Load() }, + hasValidatorSetChanged: func(uint64) (bool, NodeBLSMappings, error) { return false, nil, nil }, + getPChainHeight: func(context.Context) (uint64, error) { return pChainHeight.Load(), nil }, } decision, err := bbd.shouldBuildBlock(t.Context()) @@ -141,8 +141,8 @@ func TestShouldBuildBlock_EpochTransitionWithVMBlock(t *testing.T) { }, }, waitForPendingBlock: func(ctx context.Context) {}, - hasValidatorSetChanged: func(uint64) (bool, error) { return true, nil }, - getPChainHeight: func() uint64 { return 100 }, + hasValidatorSetChanged: func(uint64) (bool, NodeBLSMappings, error) { return true, nil, nil }, + getPChainHeight: func(context.Context) (uint64, error) { return 100, nil }, } decision, err := bbd.shouldBuildBlock(t.Context()) @@ -162,8 +162,8 @@ func TestShouldBuildBlock_EpochTransitionWithoutVMBlock(t *testing.T) { waitForPendingBlock: func(ctx context.Context) { <-ctx.Done() }, - hasValidatorSetChanged: func(uint64) (bool, error) { return true, nil }, - getPChainHeight: func() uint64 { return 100 }, + hasValidatorSetChanged: func(uint64) (bool, NodeBLSMappings, error) { return true, nil, nil }, + getPChainHeight: func(context.Context) (uint64, error) { return 100, nil }, } decision, err := bbd.shouldBuildBlock(t.Context()) @@ -187,8 +187,8 @@ func TestShouldBuildBlock_EpochTransitionContextCanceled(t *testing.T) { cancel() <-ctx.Done() }, - hasValidatorSetChanged: func(uint64) (bool, error) { return true, nil }, - getPChainHeight: func() uint64 { return 100 }, + hasValidatorSetChanged: func(uint64) (bool, NodeBLSMappings,error) { return true, nil, nil }, + getPChainHeight: func(context.Context) (uint64, error) { return 100, nil }, } decision, err := bbd.shouldBuildBlock(ctx) diff --git a/msm/fake_node_test.go b/msm/fake_node_test.go index 9e8c69b9..07922e70 100644 --- a/msm/fake_node_test.go +++ b/msm/fake_node_test.go @@ -13,6 +13,7 @@ import ( "time" "github.com/ava-labs/simplex/common" + "github.com/ava-labs/simplex/testutil" "github.com/stretchr/testify/require" ) @@ -33,13 +34,15 @@ func TestFakeNodeEpochChangesDespiteEmptyMempool(t *testing.T) { var pChainHeight atomic.Uint64 pChainHeight.Store(100) node := newFakeNode(t) + myID := [20]byte{1} + node.sm.MyNodeID = myID[:] node.sm.AuxiliaryInfoApp = &noopTestAuxInfoApp{} node.sm.GetValidatorSet = validatorSetRetriever.getValidatorSet - node.sm.GetPChainHeightForProposing = func() uint64 { - return pChainHeight.Load() + node.sm.GetPChainHeightForProposing = func(context.Context) (uint64, error) { + return pChainHeight.Load(), nil } - node.sm.GetPChainHeightForVerifying = func() uint64 { - return pChainHeight.Load() + node.sm.GetPChainHeightForVerifying = func(context.Context) (uint64, error) { + return pChainHeight.Load(), nil } node.mempoolEmpty = true node.sm.MaxBlockBuildingWaitTime = 100 * time.Millisecond @@ -60,13 +63,9 @@ func TestFakeNodeEpochChangesDespiteEmptyMempool(t *testing.T) { node.tryFinalizeNextBlock() } if flipCoin() { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{1}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{1}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } else { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{2}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{2}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } if node.isLastBlockSealing() { @@ -90,13 +89,15 @@ func TestFakeNode(t *testing.T) { var pChainHeight atomic.Uint64 pChainHeight.Store(100) node := newFakeNode(t) + myID := [20]byte{1} + node.sm.MyNodeID = myID[:] node.sm.AuxiliaryInfoApp = &noopTestAuxInfoApp{} node.sm.GetValidatorSet = validatorSetRetriever.getValidatorSet - node.sm.GetPChainHeightForProposing = func() uint64 { - return pChainHeight.Load() + node.sm.GetPChainHeightForProposing = func(context.Context) (uint64, error) { + return pChainHeight.Load(), nil } - node.sm.GetPChainHeightForVerifying = func() uint64 { - return pChainHeight.Load() + node.sm.GetPChainHeightForVerifying = func(context.Context) (uint64, error) { + return pChainHeight.Load(), nil } // Create some blocks and finalize them, until we reach height 10 @@ -111,13 +112,9 @@ func TestFakeNode(t *testing.T) { for node.Epoch() == epoch { node.act() if flipCoin() { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{1}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{1}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } else { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{2}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{2}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } } @@ -132,13 +129,9 @@ func TestFakeNode(t *testing.T) { for node.Epoch() == epoch { node.act() if flipCoin() { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{2}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{2}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } else { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{3}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{3}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } } @@ -160,14 +153,16 @@ func TestFakeNodeEmptyMempool(t *testing.T) { var pChainHeight uint64 = 100 node := newFakeNode(t) + myID := [20]byte{1} + node.sm.MyNodeID = myID[:] node.sm.AuxiliaryInfoApp = &noopTestAuxInfoApp{} node.sm.MaxBlockBuildingWaitTime = 100 * time.Millisecond node.sm.GetValidatorSet = validatorSetRetriever.getValidatorSet - node.sm.GetPChainHeightForProposing = func() uint64 { - return pChainHeight + node.sm.GetPChainHeightForProposing = func(context.Context) (uint64, error) { + return pChainHeight, nil } - node.sm.GetPChainHeightForVerifying = func() uint64 { - return pChainHeight + node.sm.GetPChainHeightForVerifying = func(context.Context) (uint64, error) { + return pChainHeight, nil } // Create some blocks and finalize them, until we reach height 10 @@ -185,13 +180,9 @@ func TestFakeNodeEmptyMempool(t *testing.T) { for node.lastFinalizedBlock().Metadata.SimplexEpochInfo.BlockValidationDescriptor == nil { node.act() if flipCoin() { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{1}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{1}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } else { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{2}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{2}, PChainHeight: 200, Signature: signApproval(200, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } } @@ -218,13 +209,9 @@ func TestFakeNodeEmptyMempool(t *testing.T) { for node.Height() < 30 { node.act() if flipCoin() { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{2}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{2}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } else { - node.sm.ApprovalsRetriever = &approvalsRetriever{ - result: []ValidatorSetApproval{{NodeID: [20]byte{3}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}}, - } + require.NoError(t, node.sm.HandleApproval(&ValidatorSetApproval{NodeID: [20]byte{3}, PChainHeight: 300, Signature: signApproval(300, emptyAuxInfoDigest), AuxInfoDigest: emptyAuxInfoDigest}, 1)) } } @@ -234,7 +221,7 @@ func TestFakeNodeEmptyMempool(t *testing.T) { } type innerBlock struct { - InnerBlock + testutil.InnerBlock Prev [32]byte } @@ -260,7 +247,11 @@ func (fn *fakeNode) WaitForProgress(ctx context.Context, pChainHeight uint64) er case <-ctx.Done(): return ctx.Err() case <-time.After(10 * time.Millisecond): - if fn.sm.GetPChainHeightForProposing() != pChainHeight { + height, err := fn.sm.GetPChainHeightForProposing(ctx) + if err != nil { + return err + } + if height != pChainHeight { return nil } } @@ -474,8 +465,8 @@ func (fn *fakeNode) BuildBlock(ctx context.Context, _ uint64) (VMBlock, error) { vmBlock := &innerBlock{ Prev: fn.getLastVMBlockDigest(), - InnerBlock: InnerBlock{ - Bytes: randomBuff(10), + InnerBlock: testutil.InnerBlock{ + Content: randomBuff(10), TS: time.Now(), BlockHeight: uint64(count), }, @@ -487,7 +478,7 @@ func (fn *fakeNode) getParentBlock() StateMachineBlock { if len(fn.blocks) > 0 { return fn.blocks[len(fn.blocks)-1].block } - gb := genesisBlock.InnerBlock.(*InnerBlock) + gb := genesisBlock.InnerBlock.(*testutil.InnerBlock) return StateMachineBlock{ InnerBlock: &innerBlock{ InnerBlock: *gb, diff --git a/msm/fuzz_test.go b/msm/fuzz_test.go index 12a4e489..11cc4de9 100644 --- a/msm/fuzz_test.go +++ b/msm/fuzz_test.go @@ -105,8 +105,8 @@ func FuzzVerifyBlock(f *testing.F) { // Model the verifier as knowing the P-chain exactly up to the height the block // references — the minimal knowledge required to verify it, mirroring production // where GetPChainHeightForVerifying returns the verifier's latest observed height. - sm.GetPChainHeightForProposing = func() uint64 { return block.Metadata.PChainHeight } - sm.GetPChainHeightForVerifying = func() uint64 { return block.Metadata.PChainHeight } + sm.GetPChainHeightForProposing = func(context.Context) (uint64, error) { return block.Metadata.PChainHeight, nil } + sm.GetPChainHeightForVerifying = func(context.Context) (uint64, error) { return block.Metadata.PChainHeight, nil } // The unfuzzed block built by the chain MSM must verify. require.NoError(t, sm.VerifyBlock(context.Background(), block), @@ -184,7 +184,7 @@ func buildEpochChain(tb testing.TB, logger common.Logger) ([]*StateMachineBlock, // Use a genesis block anchored at startTime: the zero block carries over the last // non-Simplex block's timestamp, so it must not be ahead of the blocks built after it. - genesis := StateMachineBlock{InnerBlock: &InnerBlock{BlockHeight: 0, TS: startTime, Bytes: []byte{0}}} + genesis := StateMachineBlock{InnerBlock: &InnerBlock{BlockHeight: 0, TS: startTime, bytes: []byte{0}}} currentPChainHeight := pChainHeight1 currentTime := startTime @@ -195,17 +195,14 @@ func buildEpochChain(tb testing.TB, logger common.Logger) ([]*StateMachineBlock, // round (the builder only collects approvals once the aux info history is ready). sm.AuxiliaryInfoApp = &noopTestAuxInfoApp{} sm.GetValidatorSet = getValidatorSet - sm.GetPChainHeightForProposing = func() uint64 { return currentPChainHeight } - sm.GetPChainHeightForVerifying = func() uint64 { return currentPChainHeight } + sm.GetPChainHeightForProposing = func(context.Context) (uint64, error) { return currentPChainHeight, nil } + sm.GetPChainHeightForVerifying = func(context.Context) (uint64, error) { return currentPChainHeight, nil } sm.GetTime = func() time.Time { return currentTime } sm.GenesisValidatorSet = validatorSet1 sm.LastNonSimplexBlockPChainHeight = pChainHeight1 sm.LastNonSimplexInnerBlock = genesis.InnerBlock tc.blockStore[0] = &outerBlock{block: genesis} - var approvalsResult ValidatorSetApprovals - sm.ApprovalsRetriever = &dynamicApprovalsRetriever{approvals: &approvalsResult} - ctx := context.Background() addBlock := func(seq uint64, b *StateMachineBlock, fin *common.Finalization) { @@ -215,7 +212,7 @@ func buildEpochChain(tb testing.TB, logger common.Logger) ([]*StateMachineBlock, return &InnerBlock{ TS: startTime.Add(time.Duration(h) * time.Millisecond), BlockHeight: h, - Bytes: []byte{byte(h)}, + bytes: []byte{byte(h)}, } } build := func(seq, round, epoch uint64, prev *StateMachineBlock) *StateMachineBlock { @@ -235,14 +232,14 @@ func buildEpochChain(tb testing.TB, logger common.Logger) ([]*StateMachineBlock, // block1: the zero block, finalized so it can later serve as the epoch's reference for // the validator-set change and the sealing-block computation. - tc.blockBuilder.block = nextInner(1) + tc.blockBuilder.Block = nextInner(1) block1 := build(1, 0, 1, nil) addBlock(1, block1, &common.Finalization{}) sm.LatestPersistedHeight = 1 // block2: a normal in-epoch block. currentTime = startTime.Add(2 * time.Millisecond) - tc.blockBuilder.block = nextInner(2) + tc.blockBuilder.Block = nextInner(2) block2 := build(2, 1, 1, block1) addBlock(2, block2, nil) @@ -250,7 +247,7 @@ func buildEpochChain(tb testing.TB, logger common.Logger) ([]*StateMachineBlock, // collect approvals for the next epoch. currentPChainHeight = pChainHeight2 currentTime = startTime.Add(time.Second + 3*time.Millisecond) - tc.blockBuilder.block = nextInner(3) + tc.blockBuilder.Block = nextInner(3) block3 := build(3, 2, 1, block2) addBlock(3, block3, nil) @@ -260,22 +257,22 @@ func buildEpochChain(tb testing.TB, logger common.Logger) ([]*StateMachineBlock, auxInfoDigest := sha256.Sum256(nil) // block4 & block5: collecting-approvals blocks (1/3 then 2/3, not enough to seal). - approvalsResult = ValidatorSetApprovals{{NodeID: node1, PChainHeight: pChainHeight2, AuxInfoDigest: auxInfoDigest, Signature: signApproval(pChainHeight2, auxInfoDigest)}} + require.NoError(tb, sm.HandleApproval(&ValidatorSetApproval{NodeID: node1, PChainHeight: pChainHeight2, AuxInfoDigest: auxInfoDigest, Signature: signApproval(pChainHeight2, auxInfoDigest)}, 1)) currentTime = startTime.Add(time.Second + 4*time.Millisecond) - tc.blockBuilder.block = nextInner(4) + tc.blockBuilder.Block = nextInner(4) block4 := build(4, 3, 1, block3) addBlock(4, block4, nil) - approvalsResult = ValidatorSetApprovals{{NodeID: node2, PChainHeight: pChainHeight2, AuxInfoDigest: auxInfoDigest, Signature: signApproval(pChainHeight2, auxInfoDigest)}} + require.NoError(tb, sm.HandleApproval(&ValidatorSetApproval{NodeID: node2, PChainHeight: pChainHeight2, AuxInfoDigest: auxInfoDigest, Signature: signApproval(pChainHeight2, auxInfoDigest)}, 2)) currentTime = startTime.Add(time.Second + 5*time.Millisecond) - tc.blockBuilder.block = nextInner(5) + tc.blockBuilder.Block = nextInner(5) block5 := build(5, 4, 1, block4) addBlock(5, block5, nil) // block6: the sealing block (3/3 approvals). Its successor is in stateBuildBlockEpochSealed. - approvalsResult = ValidatorSetApprovals{{NodeID: node3, PChainHeight: pChainHeight2, AuxInfoDigest: auxInfoDigest, Signature: signApproval(pChainHeight2, auxInfoDigest)}} + require.NoError(tb, sm.HandleApproval(&ValidatorSetApproval{NodeID: node3, PChainHeight: pChainHeight2, AuxInfoDigest: auxInfoDigest, Signature: signApproval(pChainHeight2, auxInfoDigest)}, 3)) currentTime = startTime.Add(time.Second + 6*time.Millisecond) - tc.blockBuilder.block = nextInner(6) + tc.blockBuilder.Block = nextInner(6) block6 := build(6, 5, 1, block5) require.Equal(tb, stateBuildBlockEpochSealed, block6.Metadata.SimplexEpochInfo.NextState()) // Finalize the sealing block so the epoch transition can proceed. @@ -286,13 +283,13 @@ func buildEpochChain(tb testing.TB, logger common.Logger) ([]*StateMachineBlock, // block7: the first block of the new epoch, built in stateBuildBlockEpochSealed. sealingSeq := uint64(6) currentTime = startTime.Add(time.Second + 7*time.Millisecond) - tc.blockBuilder.block = nextInner(7) + tc.blockBuilder.Block = nextInner(7) block7 := build(7, 6, sealingSeq, block6) addBlock(7, block7, nil) // block8: a normal in-epoch block in the second epoch. currentTime = startTime.Add(time.Second + 8*time.Millisecond) - tc.blockBuilder.block = nextInner(8) + tc.blockBuilder.Block = nextInner(8) block8 := build(8, 7, sealingSeq, block7) addBlock(8, block8, nil) diff --git a/msm/misc.go b/msm/misc.go index c0f13239..920bee8d 100644 --- a/msm/misc.go +++ b/msm/misc.go @@ -49,6 +49,9 @@ type VMBlock interface { // If nil is returned, it is guaranteed that either Accept or Reject will be // called on this block, unless the VM is shut down. Verify(ctx context.Context, pChainHeight uint64) error + + // Bytes returns the byte representation of this block. + Bytes() ([]byte, error) } type bitmask big.Int diff --git a/msm/msm.go b/msm/msm.go index bb7797a2..ab8ae21a 100644 --- a/msm/msm.go +++ b/msm/msm.go @@ -12,6 +12,7 @@ import ( "fmt" "math" "slices" + "sync" "time" "github.com/ava-labs/simplex/common" @@ -191,6 +192,9 @@ type AuxiliaryInfoGenVerifier interface { // StateMachine manages block building and verification across epoch transitions. type StateMachine struct { *Config + lock sync.RWMutex + approvalStore *ApprovalStore + approvalStoreValidatorSet NodeBLSMappings } // Config contains the dependencies and configuration parameters needed to initialize the StateMachine. @@ -205,9 +209,9 @@ type Config struct { // GetTime returns the current time. GetTime func() time.Time // GetPChainHeightForProposing returns the latest known P-chain height to be used when building a block. - GetPChainHeightForProposing func() uint64 + GetPChainHeightForProposing func(context.Context) (uint64, error) // GetPChainHeightForVerifying returns the latest known P-chain height to be used when verifying a block. - GetPChainHeightForVerifying func() uint64 + GetPChainHeightForVerifying func(context.Context) (uint64, error) // BlockBuilder builds new VM blocks. BlockBuilder BlockBuilder // Logger is used for logging state machine operations. @@ -216,8 +220,6 @@ type Config struct { GetValidatorSet ValidatorSetRetriever // GetBlock retrieves a previously built or finalized block. GetBlock BlockRetriever - // ApprovalsRetriever retrieves validator approvals for epoch transitions. - ApprovalsRetriever ApprovalsRetriever // SignatureAggregatorCreator creates a new SignatureAggregator for aggregating validator signatures for epoch transitions. SignatureAggregatorCreator common.SignatureAggregatorCreator // KeyAggregator aggregates public keys from validators. @@ -264,6 +266,40 @@ func NewStateMachine(config *Config) (*StateMachine, error) { return &sm, nil } +func (sm *StateMachine) HandleApproval(approval *ValidatorSetApproval, timestamp uint64) error { + sm.lock.Lock() + approvalStore := sm.approvalStore + sm.lock.Unlock() + + if approvalStore == nil { + sm.Logger.Debug("Approval store is not initialized, ignoring approval", + zap.String("nodeID", fmt.Sprintf("%x", approval.NodeID)), + zap.Uint64("pChainHeight", approval.PChainHeight)) + return nil + } + + return approvalStore.HandleApproval(approval, timestamp) +} + +func (sm *StateMachine) maybeInitializeApprovalStore(validatorSet NodeBLSMappings) error { + sm.lock.Lock() + defer sm.lock.Unlock() + + // If the approval store is not initialized or the validator set has changed, create a new approval store. + if sm.approvalStore == nil || ! validatorSet.Equal(sm.approvalStoreValidatorSet) { + // We first save the old approval store to copy over any existing approvals to the new approval store. + oldApprovalStore := sm.approvalStore + sm.approvalStore = NewApprovalStore(sm.SignatureVerifier, validatorSet, sm.Logger) + sm.approvalStoreValidatorSet = validatorSet + if oldApprovalStore != nil { + oldApprovalStore.PutApprovals(sm.approvalStore) + } + return nil + } + + return nil +} + // BuildBlock constructs the next block on top of the given parent block, and passes in the provided simplex metadata and blacklist. func (sm *StateMachine) BuildBlock(ctx context.Context, metadata common.ProtocolMetadata, blacklist *common.Blacklist) (*StateMachineBlock, error) { // The zero sequence number is reserved for the genesis block, which should never be built. @@ -365,7 +401,10 @@ func (sm *StateMachine) verifyNonZeroBlock(ctx context.Context, block, prevBlock return fmt.Errorf("failed to verify timestamp: %w", err) } - currentPChainHeight := sm.GetPChainHeightForVerifying() + currentPChainHeight, err := sm.GetPChainHeightForVerifying(ctx) + if err != nil { + return fmt.Errorf("failed to get current P-chain height for verifying: %w", err) + } prevPChainHeight := prevBlockMD.PChainHeight proposedPChainHeight := block.Metadata.PChainHeight @@ -373,7 +412,7 @@ func (sm *StateMachine) verifyNonZeroBlock(ctx context.Context, block, prevBlock return fmt.Errorf("failed to verify P-chain height: %w", err) } - err := sm.verifyEpochNumber(block) + err = sm.verifyEpochNumber(block) if err != nil { return err } @@ -468,6 +507,7 @@ func (sm *StateMachine) buildBlockOrTransitionEpoch(ctx context.Context, parentB if decisionToBuildBlock.transitionEpoch && isSealingBlockFinalized { sm.Logger.Debug("Transitioning epoch after building block", zap.Uint64("newPChainRefHeight", decisionToBuildBlock.pChainHeight)) newSimplexEpochInfo.NextPChainReferenceHeight = decisionToBuildBlock.pChainHeight + sm.maybeInitializeApprovalStore(decisionToBuildBlock.validatorSet) } now := sm.GetTime() @@ -541,7 +581,7 @@ func (sm *StateMachine) verifyNormalBlock(ctx context.Context, parentBlock State icmEpochInfo := computeICMEpochInfo(parentBlock, sm.ComputeICMEpoch, timestamp) - if err := sm.verifyNextPChainRefHeightNormal(parentBlock.Metadata, nextBlock.Metadata.SimplexEpochInfo); err != nil { + if err := sm.verifyNextPChainRefHeightNormal(ctx, parentBlock.Metadata, nextBlock.Metadata.SimplexEpochInfo); err != nil { return fmt.Errorf("failed to verify next P-chain reference height for normal block: %w", err) } newSimplexEpochInfo.NextPChainReferenceHeight = nextBlock.Metadata.SimplexEpochInfo.NextPChainReferenceHeight @@ -562,7 +602,7 @@ func verifyPChainHeight(proposedPChainHeight uint64, currentPChainHeight uint64, return nil } -func (sm *StateMachine) verifyNextPChainRefHeightNormal(prevMD StateMachineMetadata, next SimplexEpochInfo) error { +func (sm *StateMachine) verifyNextPChainRefHeightNormal(ctx context.Context, prevMD StateMachineMetadata, next SimplexEpochInfo) error { prev := prevMD.SimplexEpochInfo // Next P-chain height can only increase, not decrease. if next.NextPChainReferenceHeight > 0 && prev.PChainReferenceHeight > next.NextPChainReferenceHeight { @@ -593,7 +633,10 @@ func (sm *StateMachine) verifyNextPChainRefHeightNormal(prevMD StateMachineMetad } // Make sure we have reached the next P-chain reference height, otherwise we won't be able to validate it. - pChainHeight := sm.GetPChainHeightForVerifying() + pChainHeight, err := sm.GetPChainHeightForVerifying(ctx) + if err != nil { + return fmt.Errorf("failed to get current P-chain height for verifying: %w", err) + } if pChainHeight < next.NextPChainReferenceHeight { return fmt.Errorf("%w: target %d, current %d", errPChainHeightNotReached, next.NextPChainReferenceHeight, pChainHeight) @@ -618,6 +661,12 @@ func (sm *StateMachine) verifyNextPChainRefHeightNormal(prevMD StateMachineMetad errValidatorSetUnchanged, next.NextPChainReferenceHeight, prev.PChainReferenceHeight) } + // Else, ! currentValidatorSet.Equal(newValidatorSet) || next.NextPChainReferenceHeight == 0 + // so if next.NextPChainReferenceHeight > 0, we should initialize the approval store for the new validator set. + if next.NextPChainReferenceHeight > 0 { + sm.maybeInitializeApprovalStore(newValidatorSet) + } + // Else, either the validator set has changed, or the next P-chain reference height is still 0. // Both of these cases are fine. @@ -630,7 +679,7 @@ func (sm *StateMachine) verifyNextPChainRefHeightNormal(prevMD StateMachineMetad // We cannot reuse verifyNextPChainRefHeightNormal here — the baseline // for the validator-set change check is the new epoch's PChainReferenceHeight, not the parent's, // as in verifyNextPChainRefHeightNormal. -func (sm *StateMachine) verifyNextPChainRefHeightForNewEpoch(expectedEpochInfo SimplexEpochInfo, next SimplexEpochInfo) error { +func (sm *StateMachine) verifyNextPChainRefHeightForNewEpoch(ctx context.Context, expectedEpochInfo SimplexEpochInfo, next SimplexEpochInfo) error { // The first block of the epoch doesn't trigger an epoch change, we're all set. if next.NextPChainReferenceHeight == 0 { return nil @@ -646,7 +695,10 @@ func (sm *StateMachine) verifyNextPChainRefHeightForNewEpoch(expectedEpochInfo S // If we haven't reached this P-chain height yet, we cannot accept the next P-chain reference height, // because there is no way of querying the validator set for the next P-chain reference height. - pChainHeight := sm.GetPChainHeightForVerifying() + pChainHeight, err := sm.GetPChainHeightForVerifying(ctx) + if err != nil { + return fmt.Errorf("failed to get current P-chain height for verifying: %w", err) + } if pChainHeight < next.NextPChainReferenceHeight { return fmt.Errorf("%w: target %d, current %d", errPChainHeightNotReached, next.NextPChainReferenceHeight, pChainHeight) } @@ -666,6 +718,8 @@ func (sm *StateMachine) verifyNextPChainRefHeightForNewEpoch(expectedEpochInfo S errValidatorSetUnchanged, next.NextPChainReferenceHeight, expectedEpochInfo.PChainReferenceHeight) } + sm.maybeInitializeApprovalStore(newValidatorSet) + return nil } @@ -676,7 +730,7 @@ func (sm *StateMachine) createBlockBuildingDecider(pChainReferenceHeight uint64) pChainListener: sm.PChainProgressListener, getPChainHeight: sm.GetPChainHeightForProposing, waitForPendingBlock: sm.BlockBuilder.WaitForPendingBlock, - hasValidatorSetChanged: func(pChainHeight uint64) (bool, error) { + hasValidatorSetChanged: func(pChainHeight uint64) (bool, NodeBLSMappings, error) { // The given pChainHeight was sampled by the caller of shouldTransitionEpoch(). // We compare between the current validator set, defined by the P-chain reference height in the parent block, // and the new validator set defined by the given pChainHeight. @@ -684,12 +738,12 @@ func (sm *StateMachine) createBlockBuildingDecider(pChainReferenceHeight uint64) currentValidatorSet, err := sm.GetValidatorSet(pChainReferenceHeight) if err != nil { - return false, err + return false, nil, err } newValidatorSet, err := sm.GetValidatorSet(pChainHeight) if err != nil { - return false, err + return false, nil, err } if !currentValidatorSet.Equal(newValidatorSet) { @@ -698,9 +752,9 @@ func (sm *StateMachine) createBlockBuildingDecider(pChainReferenceHeight uint64) zap.String("newValidatorSet", fmt.Sprintf("%v", newValidatorSet.NodeWeights())), zap.Uint64("currentPChainRefHeight", pChainReferenceHeight), zap.Uint64("newPChainHeight", pChainHeight)) - return true, nil + return true,newValidatorSet, nil } - return false, nil + return false, nil, nil }, } return blockBuildingDecider @@ -926,6 +980,8 @@ func (sm *StateMachine) verifyCollectingApprovalsBlock(ctx context.Context, pare return err } + sm.maybeInitializeApprovalStore(validators) + newApprovals := nextBlock.Metadata.SimplexEpochInfo.NextEpochApprovals expectedAuxInfo, auxInfoDigest, isAuxInfoReady, err := sm.computeExpectedAuxInfoForApprovalCollection(parentBlock, nextBlock, prevBlockSeq, validators) @@ -1069,7 +1125,8 @@ func (sm *StateMachine) computeNewApprovals(parentBlock StateMachineBlock, valid // We retrieve approvals that validators have sent us for the next epoch. // These approvals are signed by validators of the next epoch. - approvalsFromPeers := sm.ApprovalsRetriever.Approvals() + sm.maybeInitializeApprovalStore(validators) + approvalsFromPeers := sm.approvalStore.Approvals() sm.Logger.Debug("Retrieved approvals from peers", zap.Int("numApprovals", len(approvalsFromPeers))) // Optimistically sign the epoch transition even if we have already did so in a previous round. @@ -1370,13 +1427,16 @@ func (sm *StateMachine) verifyBlockEpochSealed(ctx context.Context, parentBlock // the proposed pchain height and (optional) next pchain reference height, mirroring // what buildBlockOrTransitionEpoch does on the build side. proposedPChainHeight := nextBlock.Metadata.PChainHeight - currentPChainHeight := sm.GetPChainHeightForVerifying() + currentPChainHeight, err := sm.GetPChainHeightForVerifying(ctx) + if err != nil { + return fmt.Errorf("failed to get current P-chain height for verifying: %w", err) + } prevPChainHeight := parentBlock.Metadata.PChainHeight if err := verifyPChainHeight(proposedPChainHeight, currentPChainHeight, prevPChainHeight); err != nil { return fmt.Errorf("failed to verify P-chain height: %w", err) } - if err := sm.verifyNextPChainRefHeightForNewEpoch(newSimplexEpochInfo, nextBlock.Metadata.SimplexEpochInfo); err != nil { + if err := sm.verifyNextPChainRefHeightForNewEpoch(ctx, newSimplexEpochInfo, nextBlock.Metadata.SimplexEpochInfo); err != nil { return fmt.Errorf("failed to verify next P-chain reference height for new epoch block: %w", err) } newSimplexEpochInfo.NextPChainReferenceHeight = nextBlock.Metadata.SimplexEpochInfo.NextPChainReferenceHeight diff --git a/msm/msm_test.go b/msm/msm_test.go index 6d7fca94..053e8834 100644 --- a/msm/msm_test.go +++ b/msm/msm_test.go @@ -145,10 +145,10 @@ func TestMSMBuildAndVerifyBlocksAfterGenesis(t *testing.T) { func TestMSMFirstSimplexBlockAfterPreSimplexBlocks(t *testing.T) { preSimplexParent := StateMachineBlock{ - InnerBlock: &InnerBlock{ + InnerBlock: &testutil.InnerBlock{ TS: time.Now(), BlockHeight: 42, - Bytes: []byte{4, 5, 6}, + Content: []byte{4, 5, 6}, }, // Since the height is 42, this can't be a genesis block, so it must be a // pre-Simplex block. It already participates in an ICM epoch, which the zero @@ -181,10 +181,10 @@ func TestMSMFirstSimplexBlockAfterPreSimplexBlocks(t *testing.T) { sm1.LastNonSimplexInnerBlock = testConfig1.blockStore[42].block.InnerBlock sm2.LastNonSimplexInnerBlock = testConfig1.blockStore[42].block.InnerBlock - testConfig1.blockBuilder.block = &InnerBlock{ + testConfig1.blockBuilder.Block = &testutil.InnerBlock{ TS: time.Now(), BlockHeight: 43, - Bytes: []byte{7, 8, 9}, + Content: []byte{7, 8, 9}, } block, err := sm1.BuildBlock(context.Background(), md, nil) @@ -327,8 +327,8 @@ func TestMSMNormalOp(t *testing.T) { tc.validatorSetRetriever.resultMap = map[uint64]NodeBLSMappings{ newPChainHeight: newValidatorSet, } - sm.GetPChainHeightForProposing = func() uint64 { return newPChainHeight } - sm.GetPChainHeightForVerifying = func() uint64 { return newPChainHeight } + sm.GetPChainHeightForProposing = func(context.Context) (uint64, error) { return newPChainHeight, nil } + sm.GetPChainHeightForVerifying = func(context.Context) (uint64, error) { return newPChainHeight, nil } }, expectedPChainHeight: newPChainHeight, expectedNextPChainRefHeight: newPChainHeight, @@ -366,10 +366,10 @@ func TestMSMNormalOp(t *testing.T) { _, err = rand.Read(content) require.NoError(t, err) - testConfig1.blockBuilder.block = &InnerBlock{ + testConfig1.blockBuilder.Block = &testutil.InnerBlock{ TS: blockTime, BlockHeight: lastBlock.InnerBlock.Height(), - Bytes: content, + Content: content, } if testCase.setup != nil { @@ -393,10 +393,10 @@ func TestMSMNormalOp(t *testing.T) { require.NoError(t, err) expected := &StateMachineBlock{ - InnerBlock: &InnerBlock{ + InnerBlock: &testutil.InnerBlock{ TS: blockTime, BlockHeight: lastBlock.InnerBlock.Height(), - Bytes: content, + Content: content, }, Metadata: StateMachineMetadata{ SimplexBlacklist: blacklist.Bytes(), @@ -444,28 +444,28 @@ func TestMSMFullEpochLifecycle(t *testing.T) { // extra ICM epoch transition, making the test flaky. startTime := time.Now().Truncate(time.Second) - nextBlock := func(height uint64) *InnerBlock { - return &InnerBlock{ + nextBlock := func(height uint64) *testutil.InnerBlock { + return &testutil.InnerBlock{ TS: startTime.Add(time.Duration(height) * time.Millisecond), BlockHeight: height, - Bytes: []byte{byte(height)}, + Content: []byte{byte(height)}, } } // ----- Step 0: Building on top of genesis or upgrading to Simplex----- genesis := StateMachineBlock{ - InnerBlock: &InnerBlock{ + InnerBlock: &testutil.InnerBlock{ BlockHeight: 0, // Genesis block has height 0 TS: startTime, - Bytes: []byte{0}, + Content: []byte{0}, }, } notGenesis := StateMachineBlock{ - InnerBlock: &InnerBlock{ + InnerBlock: &testutil.InnerBlock{ BlockHeight: 42, TS: startTime, - Bytes: []byte{0}, + Content: []byte{0}, }, } for _, testCase := range []struct { @@ -511,11 +511,11 @@ func TestMSMFullEpochLifecycle(t *testing.T) { // getVerifyingPChainHeight stays ahead of it, modelling the more up-to-date height // the verifier checks against. const pChainHeightLag = uint64(50) - getProposingPChainHeight := func() uint64 { - return currentPChainHeight + getProposingPChainHeight := func(context.Context) (uint64, error) { + return currentPChainHeight, nil } - getVerifyingPChainHeight := func() uint64 { - return currentPChainHeight + pChainHeightLag + getVerifyingPChainHeight := func(context.Context) (uint64, error) { + return currentPChainHeight + pChainHeightLag, nil } // Since we explicitly compare the built block with an expected value, @@ -577,14 +577,14 @@ func TestMSMFullEpochLifecycle(t *testing.T) { // Each one fails the test if it consults the P-chain height function it must not use, // proving that building reads GetPChainHeightForProposing and verifying reads GetPChainHeightForVerifying. sm.GetPChainHeightForProposing = getProposingPChainHeight - sm.GetPChainHeightForVerifying = func() uint64 { + sm.GetPChainHeightForVerifying = func(context.Context) (uint64, error) { require.FailNow(t, "builder must not use GetPChainHeightForVerifying when proposing") - return 0 + return 0, nil } - smVerify.GetPChainHeightForProposing = func() uint64 { + smVerify.GetPChainHeightForProposing = func(context.Context) (uint64, error) { require.FailNow(t, "verifier must not use GetPChainHeightForProposing when verifying") - return 0 + return 0, nil } smVerify.GetPChainHeightForVerifying = getVerifyingPChainHeight @@ -606,7 +606,7 @@ func TestMSMFullEpochLifecycle(t *testing.T) { aggr := &signatureAggregator{} // ----- Step 1: Build zero epoch block (first simplex block) ----- - tc.blockBuilder.block = nextBlock(1) + tc.blockBuilder.Block = nextBlock(1) md := common.ProtocolMetadata{ Seq: baseSeq + 1, Round: 0, @@ -646,7 +646,7 @@ func TestMSMFullEpochLifecycle(t *testing.T) { // ----- Step 2: Build a normal block (no validator set change) ----- currentTime = startTime.Add(2 * time.Millisecond) - tc.blockBuilder.block = nextBlock(2) + tc.blockBuilder.Block = nextBlock(2) md = common.ProtocolMetadata{Seq: baseSeq + 2, Round: 1, Epoch: testCase.epochNum, Prev: block1.Digest()} block2, err := sm.BuildBlock(context.Background(), md, nil) require.NoError(t, err) @@ -676,7 +676,7 @@ func TestMSMFullEpochLifecycle(t *testing.T) { // block4 (whose parent is block3) sees parentTimestamp >= // epochStart + 1s and transitions ICM to epoch 2. currentTime = startTime.Add(time.Second + 3*time.Millisecond) - tc.blockBuilder.block = nextBlock(3) + tc.blockBuilder.Block = nextBlock(3) md = common.ProtocolMetadata{Seq: baseSeq + 3, Round: 2, Epoch: testCase.epochNum, Prev: block2.Digest()} block3, err := sm.BuildBlock(context.Background(), md, nil) require.NoError(t, err) @@ -701,19 +701,13 @@ func TestMSMFullEpochLifecycle(t *testing.T) { // ----- Step 4: First collecting block (1/3 approvals, not enough to seal) ----- - // Override ApprovalsRetriever to use our dynamic approvals. - var approvalsResult ValidatorSetApprovals - sm.ApprovalsRetriever = &dynamicApprovalsRetriever{approvals: &approvalsResult} - sig1 := signApproval(pChainHeight2, emptyAuxInfoDigest) - approvalsResult = ValidatorSetApprovals{ - { - NodeID: node1, - PChainHeight: pChainHeight2, - AuxInfoDigest: emptyAuxInfoDigest, - Signature: sig1, - }, - } + require.NoError(t, sm.HandleApproval(&ValidatorSetApproval{ + NodeID: node1, + PChainHeight: pChainHeight2, + AuxInfoDigest: emptyAuxInfoDigest, + Signature: sig1, + }, 1)) // node1 is at index 0 in validatorSet2 → bitmask bit 0 → {1} bitmask := []byte{1} @@ -721,7 +715,7 @@ func TestMSMFullEpochLifecycle(t *testing.T) { require.NoError(t, err) currentTime = startTime.Add(time.Second + 4*time.Millisecond) - tc.blockBuilder.block = nextBlock(4) + tc.blockBuilder.Block = nextBlock(4) md = common.ProtocolMetadata{Seq: baseSeq + 4, Round: 3, Epoch: testCase.epochNum, Prev: block3.Digest()} block4, err := sm.BuildBlock(context.Background(), md, nil) require.NoError(t, err) @@ -750,14 +744,12 @@ func TestMSMFullEpochLifecycle(t *testing.T) { // ----- Step 5: Second collecting block (2/3 approvals, still not enough since threshold is strictly > 2/3) ----- sig2 := signApproval(pChainHeight2, emptyAuxInfoDigest) - approvalsResult = ValidatorSetApprovals{ - { - NodeID: node2, - PChainHeight: pChainHeight2, - AuxInfoDigest: emptyAuxInfoDigest, - Signature: sig2, - }, - } + require.NoError(t, sm.HandleApproval(&ValidatorSetApproval{ + NodeID: node2, + PChainHeight: pChainHeight2, + AuxInfoDigest: emptyAuxInfoDigest, + Signature: sig2, + }, 2)) // node2 is at index 1 → bitmask bits 0,1 → {3} sig, err = aggr.AppendSignatures(sig, sig2) @@ -765,7 +757,7 @@ func TestMSMFullEpochLifecycle(t *testing.T) { bitmask = []byte{3} currentTime = startTime.Add(time.Second + 5*time.Millisecond) - tc.blockBuilder.block = nextBlock(5) + tc.blockBuilder.Block = nextBlock(5) md = common.ProtocolMetadata{Seq: baseSeq + 5, Round: 4, Epoch: testCase.epochNum, Prev: block4.Digest()} block5, err := sm.BuildBlock(context.Background(), md, nil) require.NoError(t, err) @@ -794,14 +786,12 @@ func TestMSMFullEpochLifecycle(t *testing.T) { // ----- Step 6: Sealing block (3/3 approvals, enough to seal) ----- sig3 := signApproval(pChainHeight2, emptyAuxInfoDigest) - approvalsResult = ValidatorSetApprovals{ - { - NodeID: node3, - PChainHeight: pChainHeight2, - AuxInfoDigest: emptyAuxInfoDigest, - Signature: sig3, - }, - } + require.NoError(t, sm.HandleApproval(&ValidatorSetApproval{ + NodeID: node3, + PChainHeight: pChainHeight2, + AuxInfoDigest: emptyAuxInfoDigest, + Signature: sig3, + }, 3)) // node3 is at index 2 → bitmask bits 0,1,2 → {7} sig6, err := aggr.AppendSignatures(sig, sig3) @@ -809,7 +799,7 @@ func TestMSMFullEpochLifecycle(t *testing.T) { bitmask = []byte{7} currentTime = startTime.Add(time.Second + 6*time.Millisecond) - tc.blockBuilder.block = nextBlock(6) + tc.blockBuilder.Block = nextBlock(6) md = common.ProtocolMetadata{Seq: baseSeq + 6, Round: 5, Epoch: testCase.epochNum, Prev: block5.Digest()} block6, err := sm.BuildBlock(context.Background(), md, nil) require.NoError(t, err) @@ -874,7 +864,7 @@ func TestMSMFullEpochLifecycle(t *testing.T) { subTestCase.setup() - tc.blockBuilder.block = nextBlock(7) + tc.blockBuilder.Block = nextBlock(7) md = common.ProtocolMetadata{Seq: baseSeq + 7, Round: 6, Epoch: testCase.epochNum, Prev: block6.Digest()} // If the sealing block isn't finalized yet, we expect to build a Telock. @@ -1095,7 +1085,7 @@ func TestVerifyNextPChainRefHeightNormal(t *testing.T) { tt.setup(tc) } - err := sm.verifyNextPChainRefHeightNormal(prevMD, tt.next) + err := sm.verifyNextPChainRefHeightNormal(t.Context(), prevMD, tt.next) if tt.err == nil { require.NoError(t, err) return @@ -1497,16 +1487,12 @@ func TestBuildBlockCollectingApprovalsDedupsOwnApprovalAcrossRounds(t *testing.T } tc.validatorSetRetriever.result = validators - // No peer approvals on either round — only the internal optimistic - // self-sign is contributed. - tc.approvalsRetriever.result = nil - // Parent block: epoch transition has started (NextPChainReferenceHeight > 0) // but no approvals have been collected yet. NextState() returns // stateBuildCollectingApprovals. parentSeq := uint64(10) parent := StateMachineBlock{ - InnerBlock: &InnerBlock{TS: time.Now(), BlockHeight: 1, Bytes: []byte{0xAA}}, + InnerBlock: &testutil.InnerBlock{TS: time.Now(), BlockHeight: 1, Content: []byte{0xAA}}, Metadata: StateMachineMetadata{ PChainHeight: 200, SimplexProtocolMetadata: (&common.ProtocolMetadata{ @@ -1523,7 +1509,7 @@ func TestBuildBlockCollectingApprovalsDedupsOwnApprovalAcrossRounds(t *testing.T tc.blockStore[parentSeq] = &outerBlock{block: parent} // ----- Round 1: first collecting-approvals block ----- - tc.blockBuilder.block = &InnerBlock{TS: time.Now(), BlockHeight: 2, Bytes: []byte{0x01}} + tc.blockBuilder.Block = &testutil.InnerBlock{TS: time.Now(), BlockHeight: 2, Content: []byte{0x01}} md1 := common.ProtocolMetadata{Seq: parentSeq + 1, Round: 6, Epoch: 1, Prev: parent.Digest()} block1, err := sm.BuildBlock(context.Background(), md1, nil) require.NoError(t, err) @@ -1539,7 +1525,7 @@ func TestBuildBlockCollectingApprovalsDedupsOwnApprovalAcrossRounds(t *testing.T tc.blockStore[md1.Seq] = &outerBlock{block: *block1} // ----- Round 2: another collecting-approvals block, still no peer approvals ----- - tc.blockBuilder.block = &InnerBlock{TS: time.Now(), BlockHeight: 3, Bytes: []byte{0x02}} + tc.blockBuilder.Block = &testutil.InnerBlock{TS: time.Now(), BlockHeight: 3, Content: []byte{0x02}} md2 := common.ProtocolMetadata{Seq: md1.Seq + 1, Round: 7, Epoch: 1, Prev: block1.Digest()} block2, err := sm.BuildBlock(context.Background(), md2, nil) require.NoError(t, err) @@ -1572,13 +1558,13 @@ func TestVerifyCollectingApprovalsNotReady(t *testing.T) { sm, tc := newStateMachine(t) // Default app (AuxiliaryInfoGenVerifier) for newStateMachine has threshold 2 // so IsSufficient returns false: the aux info is not ready. - sm.GetPChainHeightForProposing = func() uint64 { return nextPChainRefHeight } - sm.GetPChainHeightForVerifying = func() uint64 { return nextPChainRefHeight } + sm.GetPChainHeightForProposing = func(context.Context) (uint64, error) { return nextPChainRefHeight, nil } + sm.GetPChainHeightForVerifying = func(context.Context) (uint64, error) { return nextPChainRefHeight, nil } // Parent block: epoch transition in progress (NextPChainReferenceHeight > 0), // not yet sealed, so NextState() is stateBuildCollectingApprovals. parent := StateMachineBlock{ - InnerBlock: &InnerBlock{TS: time.Now(), BlockHeight: 1, Bytes: []byte{0xAA}}, + InnerBlock: &testutil.InnerBlock{TS: time.Now(), BlockHeight: 1, Content: []byte{0xAA}}, Metadata: StateMachineMetadata{ PChainHeight: nextPChainRefHeight, SimplexProtocolMetadata: (&common.ProtocolMetadata{ @@ -1598,7 +1584,7 @@ func TestVerifyCollectingApprovalsNotReady(t *testing.T) { } build := func(t *testing.T, sm *StateMachine, tc *testConfig, parent StateMachineBlock) *StateMachineBlock { - tc.blockBuilder.block = &InnerBlock{TS: time.Now(), BlockHeight: 2, Bytes: []byte{0x01}} + tc.blockBuilder.Block = &testutil.InnerBlock{TS: time.Now(), BlockHeight: 2, Content: []byte{0x01}} md := common.ProtocolMetadata{Seq: parentSeq + 1, Round: 6, Epoch: 1, Prev: parent.Digest()} block, err := sm.BuildBlock(context.Background(), md, nil) require.NoError(t, err) @@ -1656,8 +1642,8 @@ func TestCollectingApprovalsAuxInfoGating(t *testing.T) { ) sm, tc := newStateMachine(t) - sm.GetPChainHeightForProposing = func() uint64 { return nextPChainRefHeight } - sm.GetPChainHeightForVerifying = func() uint64 { return nextPChainRefHeight } + sm.GetPChainHeightForProposing = func(context.Context) (uint64, error) { return nextPChainRefHeight, nil } + sm.GetPChainHeightForVerifying = func(context.Context) (uint64, error) { return nextPChainRefHeight, nil } // Deterministic votes so the built auxiliary info is predictable. vote1 := []byte("vote-1") @@ -1683,7 +1669,7 @@ func TestCollectingApprovalsAuxInfoGating(t *testing.T) { tc.validatorSetRetriever.result = validators parent := StateMachineBlock{ - InnerBlock: &InnerBlock{TS: time.Now(), BlockHeight: 1, Bytes: []byte{0xAA}}, + InnerBlock: &testutil.InnerBlock{TS: time.Now(), BlockHeight: 1, Content: []byte{0xAA}}, Metadata: StateMachineMetadata{ PChainHeight: nextPChainRefHeight, SimplexProtocolMetadata: (&common.ProtocolMetadata{ @@ -1702,7 +1688,7 @@ func TestCollectingApprovalsAuxInfoGating(t *testing.T) { // build constructs the next collecting block on top of prev, stores it so it can serve // as a parent (and as a back-pointer target for the aux info history), and verifies it. build := func(seq uint64, prev StateMachineBlock) *StateMachineBlock { - tc.blockBuilder.block = &InnerBlock{TS: time.Now(), BlockHeight: seq, Bytes: []byte{byte(seq)}} + tc.blockBuilder.Block = &testutil.InnerBlock{TS: time.Now(), BlockHeight: seq, Content: []byte{byte(seq)}} md := common.ProtocolMetadata{Seq: seq, Round: seq, Epoch: 1, Prev: prev.Digest()} block, err := sm.BuildBlock(context.Background(), md, nil) require.NoError(t, err) @@ -1781,8 +1767,8 @@ func TestCollectingApprovalsAuxInfoVersionIDIsBackwardCompatible(t *testing.T) { ) sm, tc := newStateMachine(t) - sm.GetPChainHeightForVerifying = func() uint64 { return nextPChainRefHeight } - sm.GetPChainHeightForProposing = func() uint64 { return nextPChainRefHeight } + sm.GetPChainHeightForVerifying = func(context.Context) (uint64, error) { return nextPChainRefHeight, nil } + sm.GetPChainHeightForProposing = func(context.Context) (uint64, error) { return nextPChainRefHeight, nil } // threshold 4 so Generate() runs for the first three collecting blocks built on top of the // pre-seeded parent (history not yet sufficient), giving us one "first" and several "later" @@ -1807,7 +1793,7 @@ func TestCollectingApprovalsAuxInfoVersionIDIsBackwardCompatible(t *testing.T) { // The parent already carries auxiliary info for this epoch, stamped with VersionID 1. // This is the backward-compatibility precondition: the epoch's VersionID is already set. parent := StateMachineBlock{ - InnerBlock: &InnerBlock{TS: time.Now(), BlockHeight: 1, Bytes: []byte{0xAA}}, + InnerBlock: &testutil.InnerBlock{TS: time.Now(), BlockHeight: 1, Content: []byte{0xAA}}, Metadata: StateMachineMetadata{ PChainHeight: nextPChainRefHeight, SimplexProtocolMetadata: (&common.ProtocolMetadata{ @@ -1831,7 +1817,7 @@ func TestCollectingApprovalsAuxInfoVersionIDIsBackwardCompatible(t *testing.T) { // buildAndVerify constructs the next collecting block on top of prev, verifies it, and stores it so it // can serve as the next parent (and as a back-pointer target for the aux info history). buildAndVerify := func(seq uint64, prev StateMachineBlock) *StateMachineBlock { - tc.blockBuilder.block = &InnerBlock{TS: time.Now(), BlockHeight: seq, Bytes: []byte{byte(seq)}} + tc.blockBuilder.Block = &testutil.InnerBlock{TS: time.Now(), BlockHeight: seq, Content: []byte{byte(seq)}} md := common.ProtocolMetadata{Seq: seq, Round: seq, Epoch: 1, Prev: prev.Digest()} block, err := sm.BuildBlock(context.Background(), md, nil) require.NoError(t, err) diff --git a/msm/util_test.go b/msm/util_test.go index c0465a12..1f4402b2 100644 --- a/msm/util_test.go +++ b/msm/util_test.go @@ -42,11 +42,15 @@ func init() { type InnerBlock struct { TS time.Time BlockHeight uint64 - Bytes []byte + bytes []byte +} + +func (i *InnerBlock) Bytes() ([]byte, error) { + return i.bytes, nil } func (i *InnerBlock) Digest() [32]byte { - return sha256.Sum256(i.Bytes) + return sha256.Sum256(i.bytes) } func (i *InnerBlock) Height() uint64 { @@ -66,6 +70,10 @@ type fakeVMBlock struct { height uint64 } +func (f *fakeVMBlock) Bytes() ([]byte, error) { + panic("implement me") +} + func (f *fakeVMBlock) Digest() [32]byte { return [32]byte{} } func (f *fakeVMBlock) Height() uint64 { return f.height } func (f *fakeVMBlock) Timestamp() time.Time { return time.Time{} } @@ -92,14 +100,6 @@ func (bs blockStore) getBlock(seq uint64, _ [32]byte) (StateMachineBlock, *commo return blk.block, blk.finalization, nil } -type approvalsRetriever struct { - result ValidatorSetApprovals -} - -func (a approvalsRetriever) Approvals() ValidatorSetApprovals { - return a.result -} - type signer struct { } @@ -206,19 +206,6 @@ func (n *noOpPChainListener) WaitForProgress(ctx context.Context, _ uint64) erro return ctx.Err() } -type blockBuilder struct { - block VMBlock - err error -} - -func (bb *blockBuilder) WaitForPendingBlock(_ context.Context) { - // Block is always ready in tests. -} - -func (bb *blockBuilder) BuildBlock(_ context.Context, _ uint64) (VMBlock, error) { - return bb.block, bb.err -} - type validatorSetRetriever struct { result NodeBLSMappings resultMap map[uint64]NodeBLSMappings @@ -247,21 +234,13 @@ func (ka *keyAggregator) AggregateKeys(keys ...[]byte) ([]byte, error) { var ( genesisBlock = StateMachineBlock{ // Genesis block metadata has all zero values - InnerBlock: &InnerBlock{ - TS: time.Now(), - Bytes: []byte{1, 2, 3}, + InnerBlock: &testutil.InnerBlock{ + TS: time.Now(), + Content: []byte{1, 2, 3}, }, } ) -type dynamicApprovalsRetriever struct { - approvals *ValidatorSetApprovals -} - -func (d *dynamicApprovalsRetriever) Approvals() ValidatorSetApprovals { - return *d.approvals -} - func makeChain(t *testing.T, simplexStartHeight uint64, endHeight uint64) []StateMachineBlock { startTime := time.Now().Add(-time.Duration(endHeight+2) * time.Second) blocks := make([]StateMachineBlock, 0, endHeight+1) @@ -301,7 +280,7 @@ func makeNormalSimplexBlock(t *testing.T, index int, blocks []StateMachineBlock, InnerBlock: &InnerBlock{ TS: start.Add(time.Duration(h) * time.Second), BlockHeight: h, - Bytes: []byte{1, 2, 3}, + bytes: []byte{1, 2, 3}, }, Metadata: StateMachineMetadata{ PChainHeight: 100, @@ -330,21 +309,32 @@ func makeNonSimplexBlock(t *testing.T, startHeight uint64, start time.Time, h ui InnerBlock: &InnerBlock{ TS: start.Add(time.Duration(h-startHeight) * time.Second), BlockHeight: h, - Bytes: []byte{1, 2, 3}, + bytes: []byte{1, 2, 3}, }, } } type testConfig struct { blockStore blockStore - approvalsRetriever approvalsRetriever signatureVerifier signatureVerifier signatureAggregator signatureAggregator - blockBuilder blockBuilder + blockBuilder mockBlockBuilder keyAggregator keyAggregator validatorSetRetriever validatorSetRetriever } +// mockBlockBuilder is a test BlockBuilder whose BuildBlock returns a preconfigured block/error. +type mockBlockBuilder struct { + Block VMBlock + Err error +} + +func (bb *mockBlockBuilder) BuildBlock(context.Context, uint64) (VMBlock, error) { + return bb.Block, bb.Err +} + +func (bb *mockBlockBuilder) WaitForPendingBlock(context.Context) {} + func newStateMachine(t *testing.T) (*StateMachine, *testConfig) { return newStateMachineWithLogger(t, testutil.MakeLogger(t)) } @@ -369,16 +359,15 @@ func newStateMachineWithLogger(tb testing.TB, logger common.Logger) (*StateMachi Logger: logger, GetBlock: testConfig.blockStore.getBlock, MaxBlockBuildingWaitTime: time.Second, - ApprovalsRetriever: &testConfig.approvalsRetriever, SignatureVerifier: &testConfig.signatureVerifier, SignatureAggregatorCreator: newSignatureAggregatorCreator(), BlockBuilder: &testConfig.blockBuilder, KeyAggregator: &testConfig.keyAggregator, - GetPChainHeightForProposing: func() uint64 { - return 100 + GetPChainHeightForProposing: func(context.Context) (uint64, error) { + return 100, nil }, - GetPChainHeightForVerifying: func() uint64 { - return 100 + GetPChainHeightForVerifying: func(context.Context) (uint64, error) { + return 100, nil }, GetValidatorSet: testConfig.validatorSetRetriever.getValidatorSet, PChainProgressListener: &noOpPChainListener{}, diff --git a/simplex/epoch.go b/simplex/epoch.go index 5d24c7e2..8045bd53 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -2274,6 +2274,11 @@ func (e *Epoch) verifyProposalMetadataAndBlacklist(block common.Block) bool { prevBlacklist = prevBlock.Blacklist() + // If the previous block belongs to an earlier epoch, the blacklist should be empty. + if prevBlock.BlockHeader().Epoch < e.Epoch { + prevBlacklist = common.NewBlacklist(uint16(len(e.validatorNodeIDs))) + } + if prevBlacklist.IsEmpty() { prevBlacklist = common.NewBlacklist(uint16(len(e.validatorNodeIDs))) } @@ -2428,6 +2433,11 @@ func (e *Epoch) retrieveBlacklistOfParentBlock(metadata common.ProtocolMetadata) } blacklist = prevBlock.Blacklist() + + // If the previous block belongs to a previous epoch, we need to reset the blacklist to be empty. + if metadata.Epoch > prevBlock.BlockHeader().Epoch { + blacklist = common.NewBlacklist(uint16(len(e.validatorNodeIDs))) + } } if blacklist.IsEmpty() { diff --git a/simplex/epoch_test.go b/simplex/epoch_test.go index 5485e142..7e017a88 100644 --- a/simplex/epoch_test.go +++ b/simplex/epoch_test.go @@ -737,6 +737,43 @@ func TestEpochSimpleFlow(t *testing.T) { } } +func TestEpochResizesBlacklistOnEpochChange(t *testing.T) { + storage := testutil.NewInMemStorage() + epoch1Block := testutil.NewTestBlock(ProtocolMetadata{Epoch: 1, Round: 0, Seq: 0}, NewBlacklist(1)) + require.NoError(t, storage.Index(context.Background(), epoch1Block, Finalization{})) + require.Equal(t, uint16(1), epoch1Block.Blacklist().NodeCount, + "blacklist must contain exactly one node") + + // Restart the node into epoch 2 with a two-validator set. + nodes := []NodeID{{1}, {2}} + bb := testutil.NewTestBlockBuilder() + conf, wal, _ := testutil.DefaultTestNodeEpochConfig(t, NodeID{2}, testutil.NewNoopComm(nodes), bb) + conf.Storage = storage + conf.Epoch = 2 + + e, err := NewEpoch(conf) + require.NoError(t, err) + // Run as epoch 2 while the last committed block still belongs to epoch 1. + e.Epoch = conf.Epoch + t.Cleanup(e.Stop) + require.NoError(t, e.Start()) + + // The node (leader) builds the next block on top of the epoch-1 block. Its + // blacklist must be sized for the new validator set (2), not inherited from the + // parent (1) — otherwise its blacklist is malformed and the block cannot be + // notarized. + bb.BlockShouldBeBuilt <- struct{}{} + block := bb.GetBuiltBlock() + require.Equal(t, uint16(2), block.Blacklist().NodeCount, + "blacklist must be resized to the new epoch's validator count") + + // The other validator votes for the block; together with the node's own vote + // that is a quorum, so the node persists a notarization for the block's round. + // This proves the blacklist has been verified by this node. + testutil.InjectTestVote(t, e, block, NodeID{1}) + wal.AssertNotarization(block.BlockHeader().Round) +} + func TestEpochStartedTwice(t *testing.T) { bb := testutil.NewTestBlockBuilder() diff --git a/testutil/block.go b/testutil/block.go index 6d6a2d9e..b608bae3 100644 --- a/testutil/block.go +++ b/testutil/block.go @@ -9,6 +9,7 @@ import ( "crypto/sha256" "encoding/asn1" "fmt" + "time" "github.com/ava-labs/simplex/common" ) @@ -147,3 +148,29 @@ func (b *BlockDeserializer) DeserializeBlock(ctx context.Context, buff []byte) ( return &tb, nil } + +type InnerBlock struct { + TS time.Time + BlockHeight uint64 + Content []byte +} + +func (i *InnerBlock) Bytes() ([]byte, error) { + return i.Content, nil +} + +func (i *InnerBlock) Digest() [32]byte { + return sha256.Sum256(i.Content) +} + +func (i *InnerBlock) Height() uint64 { + return i.BlockHeight +} + +func (i *InnerBlock) Timestamp() time.Time { + return i.TS +} + +func (i *InnerBlock) Verify(_ context.Context, _ uint64) error { + return nil +} diff --git a/testutil/util.go b/testutil/util.go index 045d4e88..b4694d86 100644 --- a/testutil/util.go +++ b/testutil/util.go @@ -30,7 +30,7 @@ func DefaultTestNodeEpochConfig(t *testing.T, nodeID common.NodeID, comm common. ID: nodeID, Signer: &TestSigner{}, WAL: wal, - Verifier: &testVerifier{}, + Verifier: &TestVerifier{}, Storage: storage, BlockBuilder: bb, SignatureAggregatorCreator: func(weights []common.Node) common.SignatureAggregator { @@ -184,14 +184,14 @@ func (t *TestSigner) Sign([]byte) ([]byte, error) { return []byte{1, 2, 3}, nil } -type testVerifier struct { +type TestVerifier struct { } -func (t *testVerifier) VerifyBlock(common.VerifiedBlock) error { +func (t *TestVerifier) VerifyBlock(common.VerifiedBlock) error { return nil } -func (t *testVerifier) Verify(_ []byte, _ []byte, pk []byte) error { +func (t *TestVerifier) VerifySignature(_ []byte, _ []byte, pk []byte) error { return nil } diff --git a/wal/gc.go b/wal/gc.go index 601a68f4..cfec03d9 100644 --- a/wal/gc.go +++ b/wal/gc.go @@ -10,6 +10,10 @@ import ( "github.com/ava-labs/simplex/common" ) +const ( + defaultMaxWALSize = 100 * 1024 * 1024 // 100 MB +) + // Creator creates a DeletableWAL. Returns an error upon failure. type Creator func() (DeletableWAL, error) @@ -60,6 +64,11 @@ func NewGarbageCollectedWAL(WALs []DeletableWAL, creator Creator, reader Reader, } wals = append(wals, wal) } + + if maxWALSize == 0 { + maxWALSize = defaultMaxWALSize + } + return &GarbageCollectedWAL{ maxWalSize: maxWALSize, reader: reader, @@ -158,7 +167,7 @@ func (gcw *GarbageCollectedWAL) ReadAll() ([][]byte, error) { return allEntries, nil } -func (gcw *GarbageCollectedWAL) Truncate(retentionTerm uint64) error { +func (gcw *GarbageCollectedWAL) GarbageCollect(retentionTerm uint64) error { newWALs := make([]garbageCollectedWAL, 0, len(gcw.wals)) for i, wal := range gcw.wals { @@ -177,3 +186,13 @@ func (gcw *GarbageCollectedWAL) Truncate(retentionTerm uint64) error { return nil } + +func (gcw *GarbageCollectedWAL) Close() error { + for i, wal := range gcw.wals { + if err := wal.Close(); err != nil { + return fmt.Errorf("error closing WAL %d: %v", i, err) + } + } + gcw.wals = nil + return nil +} diff --git a/wal/gc_test.go b/wal/gc_test.go index fc7d3378..10a834eb 100644 --- a/wal/gc_test.go +++ b/wal/gc_test.go @@ -47,12 +47,12 @@ func TestGarbageCollectedWAL(t *testing.T) { require.NoError(t, err) require.Equal(t, records, walRecords) - require.NoError(t, gcw.Truncate(6)) + require.NoError(t, gcw.GarbageCollect(6)) walRecords, err = gcw.ReadAll() require.NoError(t, err) require.Equal(t, records[6:], walRecords) - require.NoError(t, gcw.Truncate(10)) + require.NoError(t, gcw.GarbageCollect(10)) walRecords, err = gcw.ReadAll() require.NoError(t, err) require.Empty(t, walRecords)