Generalize expiry based de-duplication (#1782)

chain/accepter.go

@@ -11,13 +11,13 @@ import (
 
 type Accepter struct {
 	tracer         trace.Tracer
-	validityWindow *TimeValidityWindow
+	validityWindow ValidityWindow
 	metrics        *chainMetrics
 }
 
 func NewAccepter(
 	tracer trace.Tracer,
-	validityWindow *TimeValidityWindow,
+	validityWindow ValidityWindow,
 	metrics *chainMetrics,
 ) *Accepter {
 	return &Accepter{
@@ -31,7 +31,7 @@ func (a *Accepter) AcceptBlock(ctx context.Context, blk *ExecutionBlock) error {
 	_, span := a.tracer.Start(ctx, "Chain.AcceptBlock")
 	defer span.End()
 
-	a.metrics.txsAccepted.Add(float64(len(blk.Txs)))
+	a.metrics.txsAccepted.Add(float64(len(blk.StatelessBlock.Txs)))
 	a.validityWindow.Accept(blk)
 
 	return nil

chain/assembler.go

@@ -49,6 +49,9 @@ func (a *Assembler) AssembleBlock(
 	if err != nil {
 		return nil, nil, err
 	}
-	executionBlock := NewExecutionBlock(sb)
+	executionBlock, err := NewExecutionBlock(sb)
+	if err != nil {
+		return nil, nil, err
+	}
 	return a.processor.Execute(ctx, parentView, executionBlock)
 }

chain/builder.go

@@ -67,7 +67,7 @@ type Builder struct {
 	metadataManager MetadataManager
 	balanceHandler  BalanceHandler
 	mempool         Mempool
-	validityWindow  *TimeValidityWindow
+	validityWindow  ValidityWindow
 	metrics         *chainMetrics
 	config          Config
 }
@@ -79,7 +79,7 @@ func NewBuilder(
 	metadataManager MetadataManager,
 	balanceHandler BalanceHandler,
 	mempool Mempool,
-	validityWindow *TimeValidityWindow,
+	validityWindow ValidityWindow,
 	metrics *chainMetrics,
 	config Config,
 ) *Builder {
@@ -476,7 +476,11 @@ func (c *Builder) BuildBlock(ctx context.Context, parentView state.View, parent
 		zap.Int64("parent (t)", parent.Tmstmp),
 		zap.Int64("block (t)", timestamp),
 	)
-	return NewExecutionBlock(blk), &ExecutedBlock{
+	execBlock, err := NewExecutionBlock(blk)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	return execBlock, &ExecutedBlock{
 		Block:         blk,
 		Results:       results,
 		UnitPrices:    feeManager.UnitPrices(),

chain/chain.go

@@ -11,10 +11,14 @@ import (
 	"github.com/ava-labs/avalanchego/x/merkledb"
 	"github.com/prometheus/client_golang/prometheus"
 
+	"github.com/ava-labs/hypersdk/internal/validitywindow"
 	"github.com/ava-labs/hypersdk/internal/workers"
 	"github.com/ava-labs/hypersdk/state"
 )
 
+// create a type alias for the concrete TimeWindowWindow type.
+type ValidityWindow = *validitywindow.TimeValidityWindow[*Transaction]
+
 type Chain struct {
 	builder     *Builder
 	processor   *Processor
@@ -34,7 +38,7 @@ func NewChain(
 	balanceHandler BalanceHandler,
 	authVerifiers workers.Workers,
 	authVM AuthVM,
-	validityWindow *TimeValidityWindow,
+	validityWindow ValidityWindow,
 	config Config,
 ) (*Chain, error) {
 	metrics, err := newMetrics(registerer)

chain/parser.go

@@ -32,5 +32,5 @@ func (b *BlockParser) ParseBlock(ctx context.Context, bytes []byte) (*ExecutionB
 	if err != nil {
 		return nil, err
 	}
-	return NewExecutionBlock(blk), nil
+	return NewExecutionBlock(blk)
 }

chain/pre_executor.go

@@ -14,14 +14,14 @@ import (
 
 type PreExecutor struct {
 	ruleFactory     RuleFactory
-	validityWindow  *TimeValidityWindow
+	validityWindow  ValidityWindow
 	metadataManager MetadataManager
 	balanceHandler  BalanceHandler
 }
 
 func NewPreExecutor(
 	ruleFactory RuleFactory,
-	validityWindow *TimeValidityWindow,
+	validityWindow ValidityWindow,
 	metadataManager MetadataManager,
 	balanceHandler BalanceHandler,
 ) *PreExecutor {

chain/processor.go

@@ -35,27 +35,42 @@ type ExecutionBlock struct {
 	sigJob     workers.Job
 }
 
-func NewExecutionBlock(block *StatelessBlock) *ExecutionBlock {
+func NewExecutionBlock(block *StatelessBlock) (*ExecutionBlock, error) {
+	authCounts := make(map[uint8]int)
+	txsSet := set.NewSet[ids.ID](len(block.Txs))
+	for _, tx := range block.Txs {
+		if txsSet.Contains(tx.ID()) {
+			return nil, ErrDuplicateTx
+		}
+		txsSet.Add(tx.ID())
+		authCounts[tx.Auth.GetTypeID()]++
+	}
+
 	return &ExecutionBlock{
+		authCounts:     authCounts,
+		txsSet:         txsSet,
 		StatelessBlock: block,
-	}
+	}, nil
 }
 
-func (b *ExecutionBlock) initTxs() error {
-	if b.txsSet.Len() == len(b.Txs) {
-		return nil
-	}
-	b.authCounts = make(map[uint8]int)
-	b.txsSet = set.NewSet[ids.ID](len(b.Txs))
-	for _, tx := range b.Txs {
-		if b.txsSet.Contains(tx.ID()) {
-			return ErrDuplicateTx
-		}
-		b.txsSet.Add(tx.ID())
-		b.authCounts[tx.Auth.GetTypeID()]++
-	}
+func (b *ExecutionBlock) ContainsTx(id ids.ID) bool {
+	return b.txsSet.Contains(id)
+}
 
-	return nil
+func (b *ExecutionBlock) Height() uint64 {
+	return b.Hght
+}
+
+func (b *ExecutionBlock) Parent() ids.ID {
+	return b.Prnt
+}
+
+func (b *ExecutionBlock) Timestamp() int64 {
+	return b.Tmstmp
+}
+
+func (b *ExecutionBlock) Txs() []*Transaction {
+	return b.StatelessBlock.Txs
 }
 
 type Processor struct {
@@ -66,7 +81,7 @@ type Processor struct {
 	authVM                  AuthVM
 	metadataManager         MetadataManager
 	balanceHandler          BalanceHandler
-	validityWindow          *TimeValidityWindow
+	validityWindow          ValidityWindow
 	metrics                 *chainMetrics
 	config                  Config
 }
@@ -79,7 +94,7 @@ func NewProcessor(
 	authVM AuthVM,
 	metadataManager MetadataManager,
 	balanceHandler BalanceHandler,
-	validityWindow *TimeValidityWindow,
+	validityWindow ValidityWindow,
 	metrics *chainMetrics,
 	config Config,
 ) *Processor {
@@ -150,7 +165,7 @@ func (p *Processor) Execute(
 	if b.Tmstmp < parentTimestamp+r.GetMinBlockGap() {
 		return nil, nil, ErrTimestampTooEarly
 	}
-	if len(b.Txs) == 0 && b.Tmstmp < parentTimestamp+r.GetMinEmptyBlockGap() {
+	if len(b.StatelessBlock.Txs) == 0 && b.Tmstmp < parentTimestamp+r.GetMinEmptyBlockGap() {
 		return nil, nil, ErrTimestampTooEarly
 	}
 
@@ -286,7 +301,7 @@ func (p *Processor) executeTxs(
 	defer span.End()
 
 	var (
-		numTxs = len(b.Txs)
+		numTxs = len(b.StatelessBlock.Txs)
 		t      = b.Tmstmp
 
 		f       = fetcher.New(im, numTxs, p.config.StateFetchConcurrency)
@@ -296,7 +311,7 @@ func (p *Processor) executeTxs(
 	)
 
 	// Fetch required keys and execute transactions
-	for li, ltx := range b.Txs {
+	for li, ltx := range b.StatelessBlock.Txs {
 		i := li
 		tx := ltx
 
@@ -361,7 +376,7 @@ func (p *Processor) executeTxs(
 		return nil, nil, err
 	}
 
-	p.metrics.txsVerified.Add(float64(len(b.Txs)))
+	p.metrics.txsVerified.Add(float64(len(b.StatelessBlock.Txs)))
 
 	// Return tstate that can be used to add block-level keys to state
 	return results, ts, nil
@@ -372,13 +387,10 @@ func (p *Processor) AsyncVerify(ctx context.Context, block *ExecutionBlock) erro
 	ctx, span := p.tracer.Start(ctx, "Chain.AsyncVerify")
 	defer span.End()
 
-	if err := block.initTxs(); err != nil {
-		return err
-	}
 	if block.sigJob != nil {
 		return nil
 	}
-	sigJob, err := p.authVerificationWorkers.NewJob(len(block.Txs))
+	sigJob, err := p.authVerificationWorkers.NewJob(len(block.StatelessBlock.Txs))
 	if err != nil {
 		return err
 	}
@@ -395,7 +407,7 @@ func (p *Processor) AsyncVerify(ctx context.Context, block *ExecutionBlock) erro
 		go batchVerifier.Done(func() { sigVerifySpan.End() })
 	}()
 
-	for _, tx := range block.Txs {
+	for _, tx := range block.StatelessBlock.Txs {
 		unsignedTxBytes, err := tx.UnsignedBytes()
 		if err != nil {
 			return err //nolint:spancheck

chain/stateless_block.go

@@ -147,5 +147,5 @@ func NewGenesisBlock(root ids.ID) (*ExecutionBlock, error) {
 	if err != nil {
 		return nil, err
 	}
-	return NewExecutionBlock(sb), nil
+	return NewExecutionBlock(sb)
 }