Simplex orchestration layer preliminary implementation #427
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,252 @@ | ||
| // 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 { | ||
| nodes atomic.Value // common.Nodes | ||
| Sender | ||
| Broadcaster | ||
| } | ||
|
|
||
| func (c *Communication) SetValidators(nodes common.Nodes) { | ||
| c.nodes.Store(nodes) | ||
| } | ||
|
|
||
| func (c *Communication) Validators() common.Nodes { | ||
| nodes, ok := c.nodes.Load().(common.Nodes) | ||
| if !ok { | ||
| return nil | ||
| } | ||
| return nodes | ||
| } | ||
|
|
||
| // EpochAwareStorage is a wrapper around Storage that is aware of epoch changes. | ||
| // Upon an epoch change, it will ignore blocks from previous epochs | ||
| // and will call the onEpochChange callback when a new epoch is detected. | ||
| type EpochAwareStorage struct { | ||
| msm *metadata.StateMachine | ||
| onEpochChange func(seq uint64, validators common.Nodes) 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. | ||
|
Collaborator
There was a problem hiding this comment. how does this mean the block is a telock? don't telocks have the same epoch as the sealing block?
Collaborator
Author
There was a problem hiding this comment. this is a Telock from the previous epoch. We may collect a finalization on a Telock and then we first index the sealing block and then the Telocks. Once we index the sealing block, we increment our epoch, right? |
||
| 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, block.SealingBlockInfo().ValidatorSet); 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 | ||
| } | ||
|
|
||
| // cachedBlock is a wrapper around ParsedBlock that caches the block in the CachedStorage upon verification. | ||
| // It is needed for the MSM because the MSM needs to be able to retrieve blocks that aren't finalized during its execution. | ||
| // These blocks are cached in the CachedStorage upon verification, and removed from the cache upon finalization (indexing). | ||
| 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) | ||
|
|
||
| // We also delete all blocks that are older than the indexed block, because they are now finalized and persisted. | ||
| for digest, cachedBlock := range cs.cache { | ||
| if cachedBlock.BlockHeader().Seq < block.BlockHeader().Seq { | ||
| delete(cs.cache, 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(metadata.VersionID, metadata.NodeBLSMappings, [][]byte) ([]byte, error) { | ||
| return nil, nil | ||
| } | ||
|
|
||
| func (n *NoopAuxiliaryInfoApp) DefaultVersionID() metadata.VersionID { | ||
| return 0 | ||
| } | ||
|
|
||
| type BlockBuilderWaiter struct { | ||
| 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) { | ||
| 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) { | ||
| 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 | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
| @@ -0,0 +1,82 @@ | ||||||
| // 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" | ||||||
| ) | ||||||
|
|
||||||
| 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. | ||||||
|
Collaborator
There was a problem hiding this comment.
Suggested change
|
||||||
| 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() uint64 | ||||||
| // GetCurrentHeight returns the current height of the P-chain. | ||||||
| GetCurrentHeight() uint64 | ||||||
| // 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 | ||||||
| } | ||||||
|
|
||||||
| type Broadcaster interface { | ||||||
| Broadcast(msg *common.Message) | ||||||
| } | ||||||
|
|
||||||
| // 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) | ||||||
|
Collaborator
There was a problem hiding this comment. can we use the same terminology as simplex.Storage? maybe we can embed simplex.Storage in this API? type Storage interface {
simplex.Storage
// CreateWAL creates a new Write-Ahead Log (WAL).
CreateWAL() (wal.DeletableWAL, error)
Collaborator
There was a problem hiding this comment. why do couple the WAL with the storage anyways?
Collaborator
Author
There was a problem hiding this comment.
We can't,
Yeah, so i was trying to minimize the number of dependencies. I was too frugal to dedicate an entire interface for a single method. I can do it though
Collaborator
There was a problem hiding this comment. yea i think it makes sense to separate it
Collaborator
Author
There was a problem hiding this comment. |
||||||
|
|
||||||
| // 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 | ||||||
| } | ||||||
|
|
||||||
| type CryptoOps interface { | ||||||
| Sign(message []byte) ([]byte, error) | ||||||
| AggregateKeys(keys ...[]byte) ([]byte, error) | ||||||
|
Collaborator
There was a problem hiding this comment. instead of redefining the functions should we just inline the interfaces from msm
Suggested change
same with the others common.Signer
common.SignatureVerifier
common.QCDeserializer
metadata.KeyAggregator |
||||||
| VerifySignature(message []byte, signature []byte, publicKey []byte) error | ||||||
| CreateSignatureAggregator([]common.Node) common.SignatureAggregator | ||||||
| DeserializeQuorumCertificate(bytes []byte) (common.QuorumCertificate, error) | ||||||
| } | ||||||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
why are some of these fields exported while others aren't?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Fixed.