Generalize expiry based de-duplication, dsmr

chain/processor.go

@@ -53,7 +53,7 @@ func NewExecutionBlock(block *StatelessBlock) (*ExecutionBlock, error) {
 	}, nil
 }
 
-func (b *ExecutionBlock) ContainsTx(id ids.ID) bool {
+func (b *ExecutionBlock) Contains(id ids.ID) bool {
 	return b.txsSet.Contains(id)
 }
 
@@ -69,7 +69,7 @@ func (b *ExecutionBlock) Timestamp() int64 {
 	return b.Tmstmp
 }
 
-func (b *ExecutionBlock) Txs() []*Transaction {
+func (b *ExecutionBlock) Containers() []*Transaction {
 	return b.StatelessBlock.Txs
 }
 

internal/validitywindow/dependencies.go

@@ -15,8 +15,8 @@ type ExecutionBlock[Container emap.Item] interface {
 	Parent() ids.ID
 	Timestamp() int64
 	Height() uint64
-	Txs() []Container
-	ContainsTx(ids.ID) bool
+	Containers() []Container
+	Contains(ids.ID) bool
 }
 
 type ChainIndex[Container emap.Item] interface {

internal/validitywindow/validitywindow.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/ava-labs/avalanchego/ids"
 	"github.com/ava-labs/avalanchego/trace"
 	"github.com/ava-labs/avalanchego/utils/logging"
 	"github.com/ava-labs/avalanchego/utils/set"
@@ -45,7 +46,7 @@ func (v *TimeValidityWindow[Container]) Accept(blk ExecutionBlock[Container]) {
 
 	evicted := v.seen.SetMin(blk.Timestamp())
 	v.log.Debug("txs evicted from seen", zap.Int("len", len(evicted)))
-	v.seen.Add(blk.Txs())
+	v.seen.Add(blk.Containers())
 	v.lastAcceptedBlockHeight = blk.Height()
 }
 
@@ -62,13 +63,23 @@ func (v *TimeValidityWindow[Container]) VerifyExpiryReplayProtection(
 		return err
 	}
 
-	dup, err := v.isRepeat(ctx, parent, oldestAllowed, blk.Txs(), true)
+	dup, err := v.isRepeat(ctx, parent, oldestAllowed, blk.Containers(), true)
 	if err != nil {
 		return err
 	}
 	if dup.Len() > 0 {
 		return fmt.Errorf("%w: duplicate in ancestry", ErrDuplicateContainer)
 	}
+	// make sure we have no repeats within the block itself.
+	// set.Set
+	blkContainerIDs := set.NewSet[ids.ID](len(blk.Containers()))
+	for _, container := range blk.Containers() {
+		id := container.GetID()
+		if blkContainerIDs.Contains(id) {
+			return fmt.Errorf("%w: duplicate in block", ErrDuplicateContainer)
+		}
+		blkContainerIDs.Add(id)
+	}
 	return nil
 }
 
@@ -85,7 +96,7 @@ func (v *TimeValidityWindow[Container]) isRepeat(
 	ctx context.Context,
 	ancestorBlk ExecutionBlock[Container],
 	oldestAllowed int64,
-	txs []Container,
+	containers []Container,
 	stop bool,
 ) (set.Bits, error) {
 	marker := set.NewBits()
@@ -103,14 +114,14 @@ func (v *TimeValidityWindow[Container]) isRepeat(
 		}
 
 		if ancestorBlk.Height() <= v.lastAcceptedBlockHeight || ancestorBlk.Height() == 0 {
-			return v.seen.Contains(txs, marker, stop), nil
+			return v.seen.Contains(containers, marker, stop), nil
 		}
 
-		for i, tx := range txs {
+		for i, container := range containers {
 			if marker.Contains(i) {
 				continue
 			}
-			if ancestorBlk.ContainsTx(tx.GetID()) {
+			if ancestorBlk.Contains(container.GetID()) {
 				marker.Add(i)
 				if stop {
 					return marker, nil

x/dsmr/block.go

@@ -6,6 +6,7 @@ package dsmr
 import (
 	"github.com/ava-labs/avalanchego/ids"
 	"github.com/ava-labs/avalanchego/utils/crypto/bls"
+	"github.com/ava-labs/avalanchego/utils/set"
 	"github.com/ava-labs/avalanchego/utils/wrappers"
 	"github.com/ava-labs/avalanchego/vms/platformvm/warp"
 
@@ -105,6 +106,47 @@ func ParseChunk[T Tx](chunkBytes []byte) (Chunk[T], error) {
 	return c, c.init()
 }
 
+// ExecutionBlock bridge the gap between the dsmr's block implementation and the validity window's execution block interface.
+type ExecutionBlock struct {
+	innerBlock Block
+	certSet    set.Set[ids.ID]
+}
+
+func (e ExecutionBlock) Timestamp() int64 {
+	return e.innerBlock.Timestamp
+}
+
+func (e ExecutionBlock) Height() uint64 {
+	return e.innerBlock.Height
+}
+
+func (e ExecutionBlock) Contains(id ids.ID) bool {
+	return e.certSet.Contains(id)
+}
+
+func (e ExecutionBlock) Parent() ids.ID {
+	return e.innerBlock.ParentID
+}
+
+func (e ExecutionBlock) Containers() []*emapChunkCertificate {
+	emapChunkCert := make([]*emapChunkCertificate, len(e.innerBlock.ChunkCerts))
+	for i := range emapChunkCert {
+		emapChunkCert[i] = &emapChunkCertificate{*e.innerBlock.ChunkCerts[i]}
+	}
+	return emapChunkCert
+}
+
+func NewExecutionBlock(innerBlock Block) ExecutionBlock {
+	certSet := set.Set[ids.ID]{}
+	for _, c := range innerBlock.ChunkCerts {
+		certSet.Add(c.ChunkID)
+	}
+	return ExecutionBlock{
+		innerBlock: innerBlock,
+		certSet:    certSet,
+	}
+}
+
 type Block struct {
 	ParentID  ids.ID `serialize:"true"`
 	Height    uint64 `serialize:"true"`
@@ -116,6 +158,15 @@ type Block struct {
 	blkBytes []byte
 }
 
+func NewBlock(parentID ids.ID, height uint64, timestamp int64, chunkCerts []*ChunkCertificate) Block {
+	return Block{
+		ParentID:   parentID,
+		Height:     height,
+		Timestamp:  timestamp,
+		ChunkCerts: chunkCerts,
+	}
+}
+
 func (b Block) GetID() ids.ID {
 	return b.blkID
 }

x/dsmr/certificate.go

@@ -46,6 +46,14 @@ type ChunkReference struct {
 	Expiry   int64      `serialize:"true"`
 }
 
+type emapChunkCertificate struct {
+	ChunkCertificate
+}
+
+func (e emapChunkCertificate) GetID() ids.ID { return e.ChunkID }
+
+func (e emapChunkCertificate) GetExpiry() int64 { return e.Expiry }
+
 type ChunkCertificate struct {
 	ChunkReference `serialize:"true"`
 	Signature      *warp.BitSetSignature `serialize:"true"`

x/dsmr/node.go

@@ -14,13 +14,15 @@ import (
 	"github.com/ava-labs/avalanchego/ids"
 	"github.com/ava-labs/avalanchego/network/p2p"
 	"github.com/ava-labs/avalanchego/network/p2p/acp118"
+	"github.com/ava-labs/avalanchego/trace"
 	"github.com/ava-labs/avalanchego/utils/crypto/bls"
 	"github.com/ava-labs/avalanchego/utils/logging"
 	"github.com/ava-labs/avalanchego/utils/wrappers"
 	"github.com/ava-labs/avalanchego/vms/platformvm/warp"
 
 	"github.com/ava-labs/hypersdk/codec"
 	"github.com/ava-labs/hypersdk/consts"
+	"github.com/ava-labs/hypersdk/internal/validitywindow"
 	"github.com/ava-labs/hypersdk/proto/pb/dsmr"
 	"github.com/ava-labs/hypersdk/utils"
 
@@ -36,6 +38,7 @@ var (
 
 	ErrEmptyChunk                          = errors.New("empty chunk")
 	ErrNoAvailableChunkCerts               = errors.New("no available chunk certs")
+	ErrAllChunkCertsDuplicate              = errors.New("all chunk certs are duplicated")
 	ErrTimestampNotMonotonicallyIncreasing = errors.New("block timestamp must be greater than parent timestamp")
 	ErrEmptyBlock                          = errors.New("block must reference chunks")
 	ErrInvalidBlockParent                  = errors.New("invalid referenced block parent")
@@ -54,6 +57,7 @@ type Validator struct {
 
 func New[T Tx](
 	log logging.Logger,
+	tracer trace.Tracer,
 	nodeID ids.NodeID,
 	networkID uint32,
 	chainID ids.ID,
@@ -70,6 +74,8 @@ func New[T Tx](
 	lastAccepted Block,
 	quorumNum uint64,
 	quorumDen uint64,
+	chainIndex validitywindow.ChainIndex[*emapChunkCertificate],
+	validityWindowDuration time.Duration,
 ) (*Node[T], error) {
 	return &Node[T]{
 		ID:                            nodeID,
@@ -88,6 +94,10 @@ func New[T Tx](
 		GetChunkSignatureHandler:      getChunkSignatureHandler,
 		ChunkCertificateGossipHandler: chunkCertificateGossipHandler,
 		storage:                       chunkStorage,
+		log:                           log,
+		tracer:                        tracer,
+		validityWindow:                validitywindow.NewTimeValidityWindow(log, tracer, chainIndex),
+		validityWindowDuration:        validityWindowDuration,
 	}, nil
 }
 
@@ -109,6 +119,10 @@ type Node[T Tx] struct {
 	GetChunkSignatureHandler      p2p.Handler
 	ChunkCertificateGossipHandler p2p.Handler
 	storage                       *ChunkStorage[T]
+	log                           logging.Logger
+	tracer                        trace.Tracer
+	validityWindowDuration        time.Duration
+	validityWindow                *validitywindow.TimeValidityWindow[*emapChunkCertificate]
 }
 
 // BuildChunk builds transactions into a Chunk
@@ -217,31 +231,41 @@ func (n *Node[T]) BuildChunk(
 	return chunk, chunkCert, n.storage.AddLocalChunkWithCert(chunk, &chunkCert)
 }
 
-func (n *Node[T]) BuildBlock(parent Block, timestamp int64) (Block, error) {
+func (n *Node[T]) BuildBlock(ctx context.Context, parent Block, timestamp int64) (Block, error) {
 	if timestamp <= parent.Timestamp {
 		return Block{}, ErrTimestampNotMonotonicallyIncreasing
 	}
 
 	chunkCerts := n.storage.GatherChunkCerts()
+	oldestAllowed := max(0, timestamp-int64(n.validityWindowDuration))
+
+	emapChunkCert := make([]*emapChunkCertificate, len(chunkCerts))
+	for i := range emapChunkCert {
+		emapChunkCert[i] = &emapChunkCertificate{*chunkCerts[i]}
+	}
+	dup, err := n.validityWindow.IsRepeat(ctx, NewExecutionBlock(parent), emapChunkCert, oldestAllowed)
+	if err != nil {
+		return Block{}, err
+	}
+
 	availableChunkCerts := make([]*ChunkCertificate, 0)
-	for _, chunkCert := range chunkCerts {
-		// avoid building blocks with expired chunk certs
-		if chunkCert.Expiry < timestamp {
+	for i, chunkCert := range chunkCerts {
+		// avoid building blocks with duplicate or expired chunk certs
+		if chunkCert.Expiry < timestamp || dup.Contains(i) {
 			continue
 		}
-
 		availableChunkCerts = append(availableChunkCerts, chunkCert)
 	}
 	if len(availableChunkCerts) == 0 {
 		return Block{}, ErrNoAvailableChunkCerts
 	}
 
-	blk := Block{
-		ParentID:   parent.GetID(),
-		Height:     parent.Height + 1,
-		Timestamp:  timestamp,
-		ChunkCerts: availableChunkCerts,
-	}
+	blk := NewBlock(
+		parent.GetID(),
+		parent.Height+1,
+		timestamp,
+		availableChunkCerts,
+	)
 
 	packer := wrappers.Packer{Bytes: make([]byte, 0, InitialChunkSize), MaxSize: consts.NetworkSizeLimit}
 	if err := codec.LinearCodec.MarshalInto(blk, &packer); err != nil {
@@ -281,6 +305,13 @@ func (n *Node[T]) Verify(ctx context.Context, parent Block, block Block) error {
 		return fmt.Errorf("%w: %s", ErrEmptyBlock, block.GetID())
 	}
 
+	// Find repeats
+	oldestAllowed := max(0, block.Timestamp-int64(n.validityWindowDuration))
+
+	if err := n.validityWindow.VerifyExpiryReplayProtection(ctx, NewExecutionBlock(block), oldestAllowed); err != nil {
+		return err
+	}
+
 	for _, chunkCert := range block.ChunkCerts {
 		if err := chunkCert.Verify(
 			ctx,
@@ -340,6 +371,8 @@ func (n *Node[T]) Accept(ctx context.Context, block Block) error {
 			}
 		}
 	}
+	// update the validity window with the accepted block.
+	n.validityWindow.Accept(NewExecutionBlock(block))
 
 	if err := n.storage.SetMin(block.Timestamp, chunkIDs); err != nil {
 		return fmt.Errorf("failed to prune chunks: %w", err)