Epoch orchestration layer

adapters.go

@@ -0,0 +1,300 @@
+// 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 epochChangeSupression struct {
+	lock sync.RWMutex
+	sealingBlockSeq uint64
+	active bool
+}
+
+func (ecs *epochChangeSupression) isSupressionActive() bool {
+	ecs.lock.RLock()
+	defer ecs.lock.RUnlock()
+	return ecs.active
+}
+
+func (ecs *epochChangeSupression) sendProhibited(seq uint64) bool {
+	ecs.lock.RLock()
+	defer ecs.lock.RUnlock()
+	if !ecs.active {
+		return false
+	}
+	return seq > ecs.sealingBlockSeq
+}
+
+func (ecs *epochChangeSupression) setSupression(sealingBlockSeq uint64) {
+	ecs.lock.Lock()
+	defer ecs.lock.Unlock()
+	ecs.sealingBlockSeq = sealingBlockSeq
+	ecs.active = true
+}
+
+func (ecs *epochChangeSupression) clearSupression() {
+	ecs.lock.Lock()
+	defer ecs.lock.Unlock()
+	ecs.active = false
+}
+
+type Communication struct {
+	epochChangeSupression *epochChangeSupression
+	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
+}
+
+func (c *Communication) Broadcast(msg *common.Message) {
+	if c.epochChangeSupression.sendProhibited(msg.Seq()) {
+		return
+	}
+
+	c.Broadcaster.Broadcast(msg)
+}
+
+func (c *Communication) Send(msg *common.Message, destination common.NodeID) {
+	if c.epochChangeSupression.sendProhibited(msg.Seq()) {
+		return
+	}
+
+	c.Sender.Send(msg, destination)
+}
+
+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.
+		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
+}
+
+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 {
+	epochChangeSupression *epochChangeSupression
+	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.epochChangeSupression.isSupressionActive() {
+		<- ctx.Done() // We wait for the context to be cancelled once the epoch is tore down.
+		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.epochChangeSupression.isSupressionActive() {
+		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
+}

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

@@ -25,6 +25,31 @@ type Message struct {
 	BlockDigestRequest          *BlockDigestRequest
 }
 
+func (m *Message) Seq() uint64 {
+	switch {
+	case m.BlockMessage != nil:
+		return m.BlockMessage.Block.BlockHeader().Seq
+	case m.VerifiedBlockMessage != nil:
+		return m.VerifiedBlockMessage.VerifiedBlock.BlockHeader().Seq
+	case m.EmptyNotarization != nil:
+		// Empty notarizations have no sequence, only a round.
+		return m.EmptyNotarization.Vote.Round
+	case m.VoteMessage != nil:
+		return m.VoteMessage.Vote.Seq
+	case m.EmptyVoteMessage != nil:
+		// Empty votes have no sequence, only a round.
+		return m.EmptyVoteMessage.Vote.Round
+	case m.Notarization != nil:
+		return m.Notarization.Vote.Seq
+	case m.FinalizeVote != nil:
+		return m.FinalizeVote.Finalization.Seq
+	case m.Finalization != nil:
+		return m.Finalization.Finalization.Seq
+	default:
+		return 0
+	}
+}
+
 type EmptyVoteMetadata struct {
 	Round uint64
 	Epoch uint64
@@ -121,7 +146,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 {