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

internal/validitywindow/dependencies.go

@@ -16,7 +16,7 @@ type ExecutionBlock[Container emap.Item] interface {
 	Timestamp() int64
 	Height() uint64
 	Txs() []Container
-	ContainsTx(ids.ID) bool
+	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"
@@ -69,6 +70,15 @@ func (v *TimeValidityWindow[Container]) VerifyExpiryReplayProtection(
 	if dup.Len() > 0 {
 		return fmt.Errorf("%w: duplicate in ancestry", ErrDuplicateContainer)
 	}
+	// make sure we have no repeats within the block itself.
+	blkTxsIDs := make(map[ids.ID]bool, len(blk.Txs()))
+	for _, tx := range blk.Txs() {
+		id := tx.GetID()
+		if _, has := blkTxsIDs[id]; has {
+			return fmt.Errorf("%w: duplicate in block", ErrDuplicateContainer)
+		}
+		blkTxsIDs[id] = true
+	}
 	return nil
 }
 
@@ -110,7 +120,7 @@ func (v *TimeValidityWindow[Container]) isRepeat(
 			if marker.Contains(i) {
 				continue
 			}
-			if ancestorBlk.ContainsTx(tx.GetID()) {
+			if ancestorBlk.Contains(tx.GetID()) {
 				marker.Add(i)
 				if stop {
 					return marker, nil

x/dsmr/assembler.go

@@ -57,7 +57,7 @@ func (b *BlockHandler[T, S, B, R]) Accept(ctx context.Context, block *Block) err
 	}
 
 	// Assemble and execute the block
-	innerBlock, result, state, err := b.Assembler.AssembleBlock(ctx, b.lastAcceptedState, b.lastAcceptedBlock, block.Timestamp, block.Height+1, txs)
+	innerBlock, result, state, err := b.Assembler.AssembleBlock(ctx, b.lastAcceptedState, b.lastAcceptedBlock, block.Tmstmp, block.Hght+1, txs)
 	if err != nil {
 		return err
 	}

x/dsmr/block.go

@@ -11,14 +11,14 @@ import (
 
 	"github.com/ava-labs/hypersdk/codec"
 	"github.com/ava-labs/hypersdk/consts"
+	"github.com/ava-labs/hypersdk/internal/emap"
 	"github.com/ava-labs/hypersdk/utils"
 )
 
 const InitialChunkSize = 250 * 1024
 
 type Tx interface {
-	GetID() ids.ID
-	GetExpiry() int64
+	emap.Item
 	GetSponsor() codec.Address
 }
 
@@ -51,6 +51,14 @@ func (c *Chunk[T]) init() error {
 	return nil
 }
 
+func (c Chunk[T]) GetID() ids.ID {
+	return c.id
+}
+
+func (c Chunk[T]) GetExpiry() int64 {
+	return c.Expiry
+}
+
 func signChunk[T Tx](
 	chunk UnsignedChunk[T],
 	networkID uint32,
@@ -106,9 +114,9 @@ func ParseChunk[T Tx](chunkBytes []byte) (Chunk[T], error) {
 }
 
 type Block struct {
-	ParentID  ids.ID `serialize:"true"`
-	Height    uint64 `serialize:"true"`
-	Timestamp int64  `serialize:"true"`
+	ParentID ids.ID `serialize:"true"`
+	Hght     uint64 `serialize:"true"`
+	Tmstmp   int64  `serialize:"true"`
 
 	ChunkCerts []*ChunkCertificate `serialize:"true"`
 
@@ -119,3 +127,28 @@ type Block struct {
 func (b Block) GetID() ids.ID {
 	return b.blkID
 }
+
+func (b Block) Parent() ids.ID {
+	return b.ParentID
+}
+
+func (b Block) Timestamp() int64 {
+	return b.Tmstmp
+}
+
+func (b Block) Height() uint64 {
+	return b.Hght
+}
+
+func (b Block) Txs() []*ChunkCertificate {
+	return b.ChunkCerts
+}
+
+func (b Block) Contains(id ids.ID) bool {
+	for _, c := range b.ChunkCerts {
+		if c.ChunkID == id {
+			return true
+		}
+	}
+	return false
+}

x/dsmr/certificate.go

@@ -54,7 +54,8 @@ type ChunkCertificate struct {
 	Signature NoVerifyChunkSignature `serialize:"true"`
 }
 
-func (c *ChunkCertificate) GetChunkID() ids.ID { return c.ChunkID }
+func (c ChunkCertificate) GetID() ids.ID    { return c.ChunkID }
+func (c ChunkCertificate) GetExpiry() int64 { return c.Expiry }
 
 func (c *ChunkCertificate) GetSlot() int64 { return c.Expiry }
 
@@ -209,7 +210,7 @@ func ParseWarpChunkCertificate(b []byte) (*WarpChunkCertificate, error) {
 	}, nil
 }
 
-func (c *WarpChunkCertificate) GetChunkID() ids.ID { return c.UnsignedCertificate.ChunkID() }
+func (c *WarpChunkCertificate) GetID() ids.ID { return c.UnsignedCertificate.ChunkID() }
 
 func (c *WarpChunkCertificate) GetSlot() int64 { return c.UnsignedCertificate.Slot() }
 

x/dsmr/node.go

@@ -14,22 +14,31 @@ 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"
 )
 
 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")
 )
 
+type (
+	ChainIndex         = validitywindow.ChainIndex[*ChunkCertificate]
+	timeValidityWindow = *validitywindow.TimeValidityWindow[*ChunkCertificate]
+)
+
 type Validator struct {
 	NodeID ids.NodeID
 	Weight uint64
@@ -46,6 +55,10 @@ func New[T Tx](
 	getChunkSignatureClient *p2p.Client,
 	chunkCertificateGossipClient *p2p.Client,
 	validators []Validator,
+	log logging.Logger,
+	tracer trace.Tracer,
+	chainIndex ChainIndex,
+	validityWindowDuration int64,
 ) (*Node[T], error) {
 	storage, err := newChunkStorage[T](NoVerifier[T]{}, memdb.New())
 	if err != nil {
@@ -79,26 +92,36 @@ func New[T Tx](
 		ChunkCertificateGossipHandler: &ChunkCertificateGossipHandler[T]{
 			storage: storage,
 		},
-		storage: storage,
+		storage:                storage,
+		log:                    log,
+		tracer:                 tracer,
+		validityWindow:         validitywindow.NewTimeValidityWindow(log, tracer, chainIndex),
+		validityWindowDuration: validityWindowDuration,
 	}, nil
 }
 
-type Node[T Tx] struct {
-	nodeID                       ids.NodeID
-	networkID                    uint32
-	chainID                      ids.ID
-	pk                           *bls.PublicKey
-	signer                       warp.Signer
-	getChunkClient               *TypedClient[*dsmr.GetChunkRequest, Chunk[T], []byte]
-	getChunkSignatureClient      *TypedClient[*dsmr.GetChunkSignatureRequest, *dsmr.GetChunkSignatureResponse, []byte]
-	chunkCertificateGossipClient *TypedClient[[]byte, []byte, *dsmr.ChunkCertificateGossip]
-	validators                   []Validator
-
-	GetChunkHandler               *GetChunkHandler[T]
-	GetChunkSignatureHandler      *acp118.Handler
-	ChunkCertificateGossipHandler *ChunkCertificateGossipHandler[T]
-	storage                       *chunkStorage[T]
-}
+type (
+	Node[T Tx] struct {
+		nodeID                       ids.NodeID
+		networkID                    uint32
+		chainID                      ids.ID
+		pk                           *bls.PublicKey
+		signer                       warp.Signer
+		getChunkClient               *TypedClient[*dsmr.GetChunkRequest, Chunk[T], []byte]
+		getChunkSignatureClient      *TypedClient[*dsmr.GetChunkSignatureRequest, *dsmr.GetChunkSignatureResponse, []byte]
+		chunkCertificateGossipClient *TypedClient[[]byte, []byte, *dsmr.ChunkCertificateGossip]
+		validators                   []Validator
+		validityWindow               timeValidityWindow
+
+		GetChunkHandler               *GetChunkHandler[T]
+		GetChunkSignatureHandler      *acp118.Handler
+		ChunkCertificateGossipHandler *ChunkCertificateGossipHandler[T]
+		storage                       *chunkStorage[T]
+		log                           logging.Logger
+		tracer                        trace.Tracer
+		validityWindowDuration        int64
+	}
+)
 
 // BuildChunk builds transactions into a Chunk
 // TODO handle frozen sponsor + validator assignments
@@ -172,29 +195,40 @@ func (n *Node[T]) BuildChunk(
 	return chunk, n.storage.AddLocalChunkWithCert(chunk, &chunkCert)
 }
 
-func (n *Node[T]) BuildBlock(parent Block, timestamp int64) (Block, error) {
-	if timestamp <= parent.Timestamp {
+func (n *Node[T]) BuildBlock(ctx context.Context, parent Block, timestamp int64) (Block, error) {
+	if timestamp <= parent.Tmstmp {
 		return Block{}, ErrTimestampNotMonotonicallyIncreasing
 	}
 
 	chunkCerts := n.storage.GatherChunkCerts()
+	oldestAllowed := timestamp - n.validityWindowDuration
+	if oldestAllowed < 0 {
+		oldestAllowed = 0
+	}
+	dup, err := n.validityWindow.IsRepeat(ctx, parent, chunkCerts, 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 {
+		if dup.Len() == len(chunkCerts) && len(chunkCerts) > 0 {
+			return Block{}, ErrAllChunkCertsDuplicate
+		}
 		return Block{}, ErrNoAvailableChunkCerts
 	}
 
 	blk := Block{
 		ParentID:   parent.GetID(),
-		Height:     parent.Height + 1,
-		Timestamp:  timestamp,
+		Hght:       parent.Hght + 1,
+		Tmstmp:     timestamp,
 		ChunkCerts: availableChunkCerts,
 	}
 
@@ -208,9 +242,22 @@ func (n *Node[T]) BuildBlock(parent Block, timestamp int64) (Block, error) {
 	return blk, nil
 }
 
-func (*Node[T]) Execute(ctx context.Context, _ Block, block Block) error {
+func (n *Node[T]) Execute(ctx context.Context, parentBlock Block, block Block) error {
 	// TODO: Verify header fields
-	// TODO: de-duplicate chunk certificates (internal to block and across history)
+	if block.Tmstmp <= parentBlock.Tmstmp && parentBlock.Hght > 0 {
+		return ErrTimestampNotMonotonicallyIncreasing
+	}
+
+	// Find repeats
+
+	oldestAllowed := block.Timestamp() - n.validityWindowDuration
+	if oldestAllowed < 0 {
+		oldestAllowed = 0
+	}
+	if err := n.validityWindow.VerifyExpiryReplayProtection(ctx, block, oldestAllowed); err != nil {
+		return err
+	}
+
 	for _, chunkCert := range block.ChunkCerts {
 		// TODO: verify chunks within a provided context
 		if err := chunkCert.Verify(ctx, struct{}{}); err != nil {
@@ -263,6 +310,8 @@ func (n *Node[T]) Accept(ctx context.Context, block Block) error {
 			}
 		}
 	}
+	// update the validity window with the accepted block.
+	n.validityWindow.Accept(block)
 
-	return n.storage.SetMin(block.Timestamp, chunkIDs)
+	return n.storage.SetMin(block.Tmstmp, chunkIDs)
 }