[wip] Disseminate approvals and auxiliary info

adapters.go

@@ -7,69 +7,67 @@ 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)
+	highestValidator func() common.Nodes
 }
 
 func (c *Communication) Validators() common.Nodes {
-	nodes, ok := c.nodes.Load().(common.Nodes)
-	if !ok {
-		return nil
-	}
-	return nodes
+	return c.highestValidator()
 }
 
-// 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
+// InstanceStorage is a wrapper around Storage that skips indexing Telocks
+// and delegates post-index handling to a caller-provided onIndex hook.
+type InstanceStorage struct {
 	Storage
-	epoch uint64
+
+	msm *metadata.StateMachine
+
+	onIndex func(block *ParsedBlock) error
 }
 
-func (e *EpochAwareStorage) Retrieve(seq uint64) (common.VerifiedBlock, common.Finalization, error) {
-	block, finalization, err := e.Storage.GetBlock(seq)
+func NewInstanceStorage(storage Storage, msm *metadata.StateMachine, onIndex func(block *ParsedBlock) error) *InstanceStorage {
+	return &InstanceStorage{
+		Storage: storage,
+		msm:     msm,
+		onIndex: onIndex,
+	}
+}
+
+func (s *InstanceStorage) Retrieve(seq uint64) (common.VerifiedBlock, common.Finalization, error) {
+	block, finalization, err := s.Storage.GetBlock(seq)
 	if err != nil {
 		return nil, common.Finalization{}, err
 	}
 	parsedBlock := &ParsedBlock{
-		msm:               e.msm,
+		msm:               s.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.
+func (s *InstanceStorage) Index(ctx context.Context, block common.VerifiedBlock, certificate common.Finalization) error {
+	pb, ok := block.(*ParsedBlock)
+	if !ok {
+		return fmt.Errorf("expected ParsedBlock, got %T", block)
+	}
+
+	// A Telock only extends time until the epoch transition finalizes, so we never index it.
+	if pb.Type() == metadata.BlockTypeTelock {
 		return nil
 	}
-	if err := e.Storage.Index(ctx, block, certificate); err != nil {
+
+	if err := s.Storage.Index(ctx, block, certificate); err != nil {
 		return err
 	}
-	// This is a sealing block, and it is not the zero block
-	if block.SealingBlockInfo() != nil && block.SealingBlockInfo().PrevSealingBlockHash != [32]byte{} {
-		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
+
+	return s.onIndex(pb)
 }
 
 // cachedBlock is a wrapper around ParsedBlock that caches the block in the CachedStorage upon verification.
@@ -89,9 +87,10 @@ func (cb *cachedBlock) Verify(ctx context.Context) (common.VerifiedBlock, error)
 }
 
 type CachedStorage struct {
-	msm  *metadata.StateMachine
-	lock sync.RWMutex
 	Storage
+
+	lock  sync.RWMutex
+	msm   *metadata.StateMachine
 	cache map[common.Digest]cachedBlock
 }
 

common/api.go

@@ -59,7 +59,7 @@ type Storage interface {
 }
 
 type Communication interface {
-	// Validators returns all nodes that participate in consensus in the epoch.
+	// Validators returns the nodes of the latest validator set.
 	Validators() Nodes
 
 	// Send sends a message to the given destination node
@@ -173,6 +173,12 @@ func (nws Nodes) NodeIDs() []NodeID {
 	return nodes
 }
 
+func (nws Nodes) Contains(id NodeID) bool {
+	return slices.ContainsFunc(nws, func(node Node) bool {
+		return id.Equals(node.Id)
+	})
+}
+
 // Node is a struct that pairs a node ID with its weight and public key.
 type Node struct {
 	Id     NodeID

common/msg.go

@@ -30,6 +30,10 @@ type Message struct {
 	// Verified Messages
 	VerifiedBlockMessage        *VerifiedBlockMessage
 	VerifiedReplicationResponse *VerifiedReplicationResponse
+
+	// Epoch Transition Messages
+	AuxiliaryInfo           *AuxiliaryInfo
+	EpochTransitionApproval *EpochTransitionApproval
 }
 
 func (m *Message) IsReplicationMessage() bool {
@@ -387,9 +391,30 @@ type BlockDigestRequest struct {
 // VersionID is an identifier for applications that care about epoch changes.
 type VersionID uint32
 
+//go:generate go run github.com/StephenButtolph/canoto/canoto msg.go
+
+// AuxiliaryInfo defines application-specific information for applications that might care about epoch change,
+// such as threshold distributed public key generation.
+type AuxiliaryInfo struct {
+	// VersionID is an identifier that identifies the application.
+	// Can be used for backward-compatibility and upgrade purposes.
+	Version VersionID `canoto:"uint,1"`
+
+	// Info is opaque bytes that can be used by applications to encode any information that describes
+	// the current state for the application.
+	Data []byte `canoto:"bytes,2"`
+
+	canotoData canotoData_AuxiliaryInfo
+}
+
+// ValidatorSetApproval is an approval from a validator
 type ValidatorSetApproval struct {
 	NodeID        avalanchego.NodeID
 	AuxInfoDigest [32]byte
 	PChainHeight  uint64
 	Signature     []byte
 }
+
+type EpochTransitionApproval struct {
+	Approval ValidatorSetApproval
+}

common/timeout_handler.go

@@ -149,3 +149,11 @@ func (t *TimeoutHandler[T]) Close() {
 		close(t.close)
 	}
 }
+
+func (t *TimeoutHandler[T]) Has(item T) bool {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+
+	_, ok := t.tasks[item]
+	return ok
+}