Simplex orchestration layer preliminary implementation

adapters.go

@@ -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 epoch, 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, block.SealingBlockInfo().ValidatorSet); err != nil {
+			return err
+		}
+		// We are now in a new epoch, so we update the epoch number to prevent indexing Telocks from the previous epoch.
+		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[common.Digest]cachedBlock
+}
+
+func NewCachedStorage(storage Storage) *CachedStorage {
+	return &CachedStorage{
+		Storage: storage,
+		cache:   make(map[common.Digest]cachedBlock),
+	}
+}
+
+func (cs *CachedStorage) RetrieveBlock(seq uint64, digest common.Digest) (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 common.Digest) (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
+}

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)
 }

common/msg.go

@@ -136,7 +136,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 {

config.go

@@ -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.
+	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)
+
+	// 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)
+	VerifySignature(message []byte, signature []byte, publicKey []byte) error
+	CreateSignatureAggregator([]common.Node) common.SignatureAggregator
+	DeserializeQuorumCertificate(bytes []byte) (common.QuorumCertificate, error)
+}