Don't restart non-validators that are behind

adapters.go

@@ -19,6 +19,15 @@ type Communication struct {
 	Broadcaster
 }
 
+func newCommunication(sender Sender, broadcaster Broadcaster, validators common.Nodes) *Communication {
+	c := &Communication{
+		Sender:      sender,
+		Broadcaster: broadcaster,
+	}
+	c.SetValidators(validators)
+	return c
+}
+
 func (c *Communication) SetValidators(nodes common.Nodes) {
 	c.nodes.Store(nodes)
 }
@@ -31,45 +40,52 @@ func (c *Communication) Validators() common.Nodes {
 	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
+// 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.
@@ -114,6 +130,16 @@ func (cs *CachedStorage) RetrieveBlock(seq uint64, digest common.Digest) (metada
 func (cs *CachedStorage) Retrieve(seq uint64, digest common.Digest) (common.VerifiedBlock, *common.Finalization, error) {
 	cs.lock.RLock()
 	item, exists := cs.cache[digest]
+	if !exists && digest == (common.Digest{}) {
+		// Seq-only lookups pass a zero digest, so scan the cache by seq.
+		// Otherwise a verified but not yet finalized block is invisible to them.
+		for _, cb := range cs.cache {
+			if cb.Metadata.SimplexProtocolMetadata.Seq == seq {
+				item, exists = cb, true
+				break
+			}
+		}
+	}
 	if exists {
 		cs.lock.RUnlock()
 		// If the block is cached, it means it's not finalized yet, because upon finalizing the block (indexing)
@@ -229,17 +255,17 @@ func (bw *BlockBuilderWaiter) BuildBlock(ctx context.Context, metadata common.Pr
 }
 
 type blockDeserializer struct {
-	vm  VM
-	msm *metadata.StateMachine
+	deserializer BlockDeserializer
+	msm          *metadata.StateMachine
 }
 
-func (bp *blockDeserializer) DeserializeBlock(ctx context.Context, bytes []byte) (common.Block, error) {
+func (bd *blockDeserializer) DeserializeBlock(ctx context.Context, bytes []byte) (common.Block, error) {
 	var rawBlock metadata.RawBlock
 	if err := rawBlock.UnmarshalCanoto(bytes); err != nil {
 		return nil, err
 	}
 
-	block, err := bp.vm.ParseBlock(ctx, rawBlock.InnerBlockBytes)
+	block, err := bd.deserializer.ParseBlock(ctx, rawBlock.InnerBlockBytes)
 	if err != nil {
 		return nil, err
 	}
@@ -248,6 +274,6 @@ func (bp *blockDeserializer) DeserializeBlock(ctx context.Context, bytes []byte)
 			InnerBlock: block,
 			Metadata:   rawBlock.Metadata,
 		},
-		msm: bp.msm,
+		msm: bd.msm,
 	}, nil
 }

common/msg.go

@@ -30,6 +30,10 @@ type Message struct {
 	// Verified Messages
 	VerifiedBlockMessage        *VerifiedBlockMessage
 	VerifiedReplicationResponse *VerifiedReplicationResponse
+
+	// Epoch Transition Messages
+	AuxiliaryInfo           *AuxiliaryInfo
+	EpochTransitionApproval *ValidatorSetApproval
 }
 
 func (m *Message) IsReplicationMessage() bool {
@@ -432,6 +436,23 @@ 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

config.go

@@ -53,12 +53,11 @@ type VM interface {
 	// WaitForPendingBlock returns when either the given context is cancelled,
 	// or when the VM signals that a block should be built.
 	WaitForPendingBlock(ctx context.Context)
+}
 
+type BlockDeserializer interface {
 	// ParseBlock parses the given block bytes into a VMBlock.
 	ParseBlock(context.Context, []byte) (avalanchego.VMBlock, error)
-
-	// ComputeICMEpoch computes the ICM epoch transition given the input parameters.
-	ComputeICMEpoch(input metadata.ICMEpochInput) metadata.ICMEpochInfo
 }
 
 type Storage interface {