fix(tbtc): separate FROST parameters and retry DKG events

pkg/chain/ethereum/ecdsa_dkg_validator_chain.go

@@ -102,17 +102,33 @@ func ecdsaWalletGroupParametersFromValidator(
 		return nil, err
 	}
 
-	groupSize, err := bigIntPositiveIntLimited("groupSize", gsBig, maxReasonableTBTCGroupSize)
+	return walletGroupParametersFromValidatorValues(gsBig, atBig, gtBig)
+}
+
+func walletGroupParametersFromValidatorValues(
+	groupSizeValue *big.Int,
+	activeThresholdValue *big.Int,
+	groupThresholdValue *big.Int,
+) (*tbtc.GroupParameters, error) {
+	groupSize, err := bigIntPositiveIntLimited(
+		"groupSize",
+		groupSizeValue,
+		maxReasonableTBTCGroupSize,
+	)
 	if err != nil {
 		return nil, err
 	}
-	groupQuorum, err := bigIntPositiveIntLimited("activeThreshold", atBig, maxReasonableTBTCGroupSize)
+	groupQuorum, err := bigIntPositiveIntLimited(
+		"activeThreshold",
+		activeThresholdValue,
+		maxReasonableTBTCGroupSize,
+	)
 	if err != nil {
 		return nil, err
 	}
 	honestThreshold, err := bigIntPositiveIntLimited(
 		"groupThreshold",
-		gtBig,
+		groupThresholdValue,
 		maxReasonableTBTCGroupSize,
 	)
 	if err != nil {

pkg/chain/ethereum/frost_group_parameters_test.go

@@ -0,0 +1,29 @@
+package ethereum
+
+import (
+	"math/big"
+	"testing"
+)
+
+func TestWalletGroupParametersFromValidatorValuesMapsFrostConstants(
+	t *testing.T,
+) {
+	parameters, err := walletGroupParametersFromValidatorValues(
+		big.NewInt(5), // FrostDkgValidator.groupSize()
+		big.NewInt(4), // FrostDkgValidator.activeThreshold()
+		big.NewInt(3), // FrostDkgValidator.groupThreshold()
+	)
+	if err != nil {
+		t.Fatalf("unexpected error: [%v]", err)
+	}
+
+	if parameters.GroupSize != 5 {
+		t.Fatalf("unexpected group size: [%d]", parameters.GroupSize)
+	}
+	if parameters.GroupQuorum != 4 {
+		t.Fatalf("unexpected group quorum: [%d]", parameters.GroupQuorum)
+	}
+	if parameters.HonestThreshold != 3 {
+		t.Fatalf("unexpected honest threshold: [%d]", parameters.HonestThreshold)
+	}
+}

pkg/chain/ethereum/tbtc.go

@@ -423,7 +423,8 @@ func connectFrostDkgValidator(
 	if frostDkgValidatorAddress == (common.Address{}) {
 		logger.Infof(
 			"%s contract address not configured; pre-submit FROST digest "+
-				"view checks disabled",
+				"view checks disabled (the address is required when the FROST "+
+				"wallet registry is enabled)",
 			FrostDkgValidatorContractName,
 		)
 		return nil, nil
@@ -460,6 +461,43 @@ func (tc *TbtcChain) EcdsaWalletGroupParametersFromChain(
 	)
 }
 
+// FrostWalletGroupParametersFromChain mirrors FrostDkgValidator sizing
+// constants. FROST and ECDSA validators are independent and may deliberately
+// use different group sizes and thresholds, so callers must not reuse the
+// legacy ECDSA parameters for FROST protocols.
+func (tc *TbtcChain) FrostWalletGroupParametersFromChain(
+	ctx context.Context,
+) (*tbtc.GroupParameters, error) {
+	if tc.frostWalletRegistry == nil {
+		return nil, nil
+	}
+	if tc.frostDkgValidator == nil {
+		return nil, fmt.Errorf(
+			"FrostDkgValidator is required when FrostWalletRegistry is configured",
+		)
+	}
+
+	callOpts := &bind.CallOpts{Context: ctx, From: tc.key.Address}
+	groupSize, err := tc.frostDkgValidator.GroupSize(callOpts)
+	if err != nil {
+		return nil, fmt.Errorf("read FrostDkgValidator groupSize: %w", err)
+	}
+	activeThreshold, err := tc.frostDkgValidator.ActiveThreshold(callOpts)
+	if err != nil {
+		return nil, fmt.Errorf("read FrostDkgValidator activeThreshold: %w", err)
+	}
+	groupThreshold, err := tc.frostDkgValidator.GroupThreshold(callOpts)
+	if err != nil {
+		return nil, fmt.Errorf("read FrostDkgValidator groupThreshold: %w", err)
+	}
+
+	return walletGroupParametersFromValidatorValues(
+		groupSize,
+		activeThreshold,
+		groupThreshold,
+	)
+}
+
 // Staking returns address of the TokenStaking contract the WalletRegistry is
 // connected to.
 func (tc *TbtcChain) Staking() (chain.Address, error) {

pkg/tbtc/deduplicator.go

@@ -4,6 +4,7 @@ import (
 	"encoding/hex"
 	"math/big"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/keep-network/keep-common/pkg/cache"
@@ -39,13 +40,95 @@ type deduplicator struct {
 	dkgSeedCache       *cache.TimeCache
 	dkgResultHashCache *cache.TimeCache
 	walletClosedCache  *cache.TimeCache
+
+	mutex      sync.Mutex
+	inProgress map[string]struct{}
 }
 
 func newDeduplicator() *deduplicator {
 	return &deduplicator{
 		dkgSeedCache:       cache.NewTimeCache(DKGSeedCachePeriod),
 		dkgResultHashCache: cache.NewTimeCache(DKGResultHashCachePeriod),
 		walletClosedCache:  cache.NewTimeCache(WalletClosedCachePeriod),
+		inProgress:         make(map[string]struct{}),
+	}
+}
+
+// deduplicationLease suppresses concurrent handling of an event without
+// marking it permanently handled up front. Completing the lease moves the key
+// to the long-lived cache; releasing it after an error lets a subscription or
+// recovery replay retry the work.
+type deduplicationLease struct {
+	owner         *deduplicator
+	cache         *cache.TimeCache
+	cacheKey      string
+	inProgressKey string
+}
+
+func (d *deduplicator) beginDKGStarted(
+	seed *big.Int,
+) (*deduplicationLease, bool) {
+	cacheKey := seed.Text(16)
+	return d.begin(
+		d.dkgSeedCache,
+		cacheKey,
+		"dkg-started:"+cacheKey,
+	)
+}
+
+func (d *deduplicator) beginDKGResultSubmitted(
+	seed *big.Int,
+	resultHash DKGChainResultHash,
+	block uint64,
+) (*deduplicationLease, bool) {
+	cacheKey := dkgResultSubmittedCacheKey(seed, resultHash, block)
+	return d.begin(
+		d.dkgResultHashCache,
+		cacheKey,
+		"dkg-result-submitted:"+cacheKey,
+	)
+}
+
+func (d *deduplicator) begin(
+	completedCache *cache.TimeCache,
+	cacheKey string,
+	inProgressKey string,
+) (*deduplicationLease, bool) {
+	completedCache.Sweep()
+
+	d.mutex.Lock()
+	defer d.mutex.Unlock()
+
+	if completedCache.Has(cacheKey) {
+		return nil, false
+	}
+	if _, ok := d.inProgress[inProgressKey]; ok {
+		return nil, false
+	}
+	if d.inProgress == nil {
+		d.inProgress = make(map[string]struct{})
+	}
+
+	d.inProgress[inProgressKey] = struct{}{}
+	return &deduplicationLease{
+		owner:         d,
+		cache:         completedCache,
+		cacheKey:      cacheKey,
+		inProgressKey: inProgressKey,
+	}, true
+}
+
+func (dl *deduplicationLease) finish(completed bool) {
+	dl.owner.mutex.Lock()
+	defer dl.owner.mutex.Unlock()
+
+	if _, ok := dl.owner.inProgress[dl.inProgressKey]; !ok {
+		return
+	}
+
+	delete(dl.owner.inProgress, dl.inProgressKey)
+	if completed {
+		dl.cache.Add(dl.cacheKey)
 	}
 }
 
@@ -59,16 +142,9 @@ func (d *deduplicator) notifyDKGStarted(
 
 	// The cache key is the hexadecimal representation of the seed.
 	cacheKey := newDKGSeed.Text(16)
-	// If the key is not in the cache, that means the seed was not handled
-	// yet and the client should proceed with the execution.
-	if !d.dkgSeedCache.Has(cacheKey) {
-		d.dkgSeedCache.Add(cacheKey)
-		return true
-	}
 
-	// Otherwise, the DKG seed is a duplicate and the client should not proceed
-	// with the execution.
-	return false
+	// Add performs the presence check and insertion atomically.
+	return d.dkgSeedCache.Add(cacheKey)
 }
 
 // notifyDKGResultSubmitted notifies the client wants to start some actions
@@ -81,20 +157,24 @@ func (d *deduplicator) notifyDKGResultSubmitted(
 ) bool {
 	d.dkgResultHashCache.Sweep()
 
-	cacheKey := newDKGResultSeed.Text(16) +
-		hex.EncodeToString(newDKGResultHash[:]) +
-		strconv.Itoa(int(newDKGResultBlock))
+	cacheKey := dkgResultSubmittedCacheKey(
+		newDKGResultSeed,
+		newDKGResultHash,
+		newDKGResultBlock,
+	)
 
-	// If the key is not in the cache, that means the result was not handled
-	// yet and the client should proceed with the execution.
-	if !d.dkgResultHashCache.Has(cacheKey) {
-		d.dkgResultHashCache.Add(cacheKey)
-		return true
-	}
+	// Add performs the presence check and insertion atomically.
+	return d.dkgResultHashCache.Add(cacheKey)
+}
 
-	// Otherwise, the DKG result is a duplicate and the client should not
-	// proceed with the execution.
-	return false
+func dkgResultSubmittedCacheKey(
+	seed *big.Int,
+	resultHash DKGChainResultHash,
+	block uint64,
+) string {
+	return seed.Text(16) +
+		hex.EncodeToString(resultHash[:]) +
+		strconv.FormatUint(block, 10)
 }
 
 func (d *deduplicator) notifyWalletClosed(

pkg/tbtc/deduplicator_test.go

@@ -51,6 +51,31 @@ func TestNotifyDKGStarted(t *testing.T) {
 	}
 }
 
+func TestDKGStartedDeduplicationLease(t *testing.T) {
+	deduplicator := newDeduplicator()
+	seed := big.NewInt(100)
+
+	lease, ok := deduplicator.beginDKGStarted(seed)
+	if !ok {
+		t.Fatal("expected first handler to acquire the event")
+	}
+
+	if _, ok := deduplicator.beginDKGStarted(seed); ok {
+		t.Fatal("expected in-progress duplicate to be suppressed")
+	}
+
+	lease.finish(false)
+	lease, ok = deduplicator.beginDKGStarted(seed)
+	if !ok {
+		t.Fatal("expected released event to be retryable")
+	}
+
+	lease.finish(true)
+	if _, ok := deduplicator.beginDKGStarted(seed); ok {
+		t.Fatal("expected completed event to be suppressed")
+	}
+}
+
 func TestNotifyDKGResultSubmitted(t *testing.T) {
 	deduplicator := deduplicator{
 		dkgResultHashCache: cache.NewTimeCache(testDKGResultHashCachePeriod),