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,9 +16,9 @@ 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 {
-	GetExecutionBlock(ctx context.Context, blkID ids.ID) (ExecutionBlock[Container], error)
+	GetExecutionBlock(ctx context.Context, blkID ids.ID) (ExecutionBlock[Container], bool, error)
 }

internal/validitywindow/syncer.go

@@ -39,10 +39,11 @@ func (s *Syncer[Container]) start(ctx context.Context, lastAcceptedBlock Executi
 		seenValidityWindow = false
 		validityWindow     = s.getValidityWindow(lastAcceptedBlock.Timestamp())
 		err                error
+		hasBlock           bool
 	)
 	for {
-		parent, err = s.chainIndex.GetExecutionBlock(ctx, parent.Parent())
-		if err != nil {
+		parent, hasBlock, err = s.chainIndex.GetExecutionBlock(ctx, parent.Parent())
+		if err != nil || !hasBlock {
 			break // If we can't fetch far enough back or we've gone past genesis, execute what we can
 		}
 		parents = append(parents, parent)

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"
@@ -57,8 +58,8 @@ func (v *TimeValidityWindow[Container]) VerifyExpiryReplayProtection(
 	if blk.Height() <= v.lastAcceptedBlockHeight {
 		return nil
 	}
-	parent, err := v.chainIndex.GetExecutionBlock(ctx, blk.Parent())
-	if err != nil {
+	parent, hasBlock, err := v.chainIndex.GetExecutionBlock(ctx, blk.Parent())
+	if err != nil || !hasBlock {
 		return err
 	}
 
@@ -69,6 +70,16 @@ 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.
+	// set.Set
+	blkTxsIDs := set.NewSet[ids.ID](len(blk.Txs()))
+	for _, tx := range blk.Txs() {
+		id := tx.GetID()
+		if blkTxsIDs.Contains(id) {
+			return fmt.Errorf("%w: duplicate in block", ErrDuplicateContainer)
+		}
+		blkTxsIDs.Add(id)
+	}
 	return nil
 }
 
@@ -97,6 +108,7 @@ func (v *TimeValidityWindow[Container]) isRepeat(
 	defer v.lock.Unlock()
 
 	var err error
+	var hasBlock bool
 	for {
 		if ancestorBlk.Timestamp() < oldestAllowed {
 			return marker, nil
@@ -110,16 +122,16 @@ 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
 				}
 			}
 		}
 
-		ancestorBlk, err = v.chainIndex.GetExecutionBlock(ctx, ancestorBlk.Parent())
-		if err != nil {
+		ancestorBlk, hasBlock, err = v.chainIndex.GetExecutionBlock(ctx, ancestorBlk.Parent())
+		if err != nil || !hasBlock {
 			return marker, err
 		}
 	}

vm/resolutions.go

@@ -92,15 +92,15 @@ func (vm *VM) LastAcceptedStatefulBlock() *StatefulBlock {
 	return vm.lastAccepted
 }
 
-func (vm *VM) GetExecutionBlock(ctx context.Context, blkID ids.ID) (validitywindow.ExecutionBlock[*chain.Transaction], error) {
+func (vm *VM) GetExecutionBlock(ctx context.Context, blkID ids.ID) (validitywindow.ExecutionBlock[*chain.Transaction], bool, error) {
 	_, span := vm.tracer.Start(ctx, "VM.GetExecutionBlock")
 	defer span.End()
 
 	blk, err := vm.GetStatefulBlock(ctx, blkID)
 	if err != nil {
-		return nil, err
+		return nil, true, err
 	}
-	return blk.ExecutionBlock, nil
+	return blk.ExecutionBlock, true, nil
 }
 
 func (vm *VM) LastAcceptedBlockResult() *chain.ExecutedBlock {

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

@@ -6,19 +6,20 @@ 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"
 
 	"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 +52,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,16 +115,54 @@ 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"`
 
 	blkID    ids.ID
 	blkBytes []byte
+	certSet  set.Set[ids.ID]
+}
+
+func NewBlock(parentID ids.ID, height uint64, timestamp int64, chunkCerts []*ChunkCertificate) Block {
+	blk := Block{
+		ParentID:   parentID,
+		Hght:       height,
+		Tmstmp:     timestamp,
+		ChunkCerts: chunkCerts,
+	}
+	blk.init()
+	return blk
 }
 
 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 {
+	return b.certSet.Contains(id)
+}
+
+func (b Block) init() {
+	for _, c := range b.ChunkCerts {
+		b.certSet.Add(c.ChunkID)
+	}
+}

x/dsmr/certificate.go

@@ -51,9 +51,8 @@ type ChunkCertificate struct {
 	Signature      *warp.BitSetSignature `serialize:"true"`
 }
 
-func (c *ChunkCertificate) GetChunkID() ids.ID { return c.ChunkID }
-
-func (c *ChunkCertificate) GetSlot() int64 { return c.Expiry }
+func (c ChunkCertificate) GetID() ids.ID    { return c.ChunkID }
+func (c ChunkCertificate) GetExpiry() int64 { return c.Expiry }
 
 func (c *ChunkCertificate) Bytes() []byte {
 	bytes, err := Codec.Marshal(CodecVersion, c)