Handled publishing jobs for signers

pkg/chain/ethereum/tbtc.go

@@ -995,15 +995,19 @@ func (tc *TbtcChain) DKGParameters() (*tbtc.DKGParameters, error) {
 }
 
 func (tc *TbtcChain) CalculateInactivityClaimSignatureHash(
-	nonce *big.Int,
-	walletPublicKey *ecdsa.PublicKey,
-	inactiveMembersIndexes []group.MemberIndex,
-	heartbeatFailed bool,
+	claim *inactivity.Claim,
 ) (inactivity.ClaimSignatureHash, error) {
 	// TODO: Implement
 	return inactivity.ClaimSignatureHash{}, nil
 }
 
+func (tc *TbtcChain) GetInactivityClaimNonce(
+	walletID [32]byte,
+) (*big.Int, error) {
+	// TODO: Implement
+	return nil, nil
+}
+
 func (tc *TbtcChain) PastDepositRevealedEvents(
 	filter *tbtc.DepositRevealedEventFilter,
 ) ([]*tbtc.DepositRevealedEvent, error) {

pkg/tbtc/chain.go

@@ -107,13 +107,6 @@ type DistributedKeyGenerationChain interface {
 		startBlock uint64,
 	) (dkg.ResultSignatureHash, error)
 
-	CalculateInactivityClaimSignatureHash(
-		nonce *big.Int,
-		walletPublicKey *ecdsa.PublicKey,
-		inactiveMembersIndexes []group.MemberIndex,
-		heartbeatFailed bool,
-	) (inactivity.ClaimSignatureHash, error)
-
 	// IsDKGResultValid checks whether the submitted DKG result is valid from
 	// the on-chain contract standpoint.
 	IsDKGResultValid(dkgResult *DKGChainResult) (bool, error)
@@ -128,6 +121,18 @@ type DistributedKeyGenerationChain interface {
 	DKGParameters() (*DKGParameters, error)
 }
 
+type InactivityClaimChain interface {
+	// CalculateInactivityClaimSignatureHash calculates hash for the given
+	// inactivity claim.
+	CalculateInactivityClaimSignatureHash(
+		claim *inactivity.Claim,
+	) (inactivity.ClaimSignatureHash, error)
+
+	// GetInactivityClaimNonce returns inactivity claim nonce for the given
+	// wallet.
+	GetInactivityClaimNonce(walletID [32]byte) (*big.Int, error)
+}
+
 // DKGChainResultHash represents a hash of the DKGChainResult. The algorithm
 // used is specific to the chain.
 type DKGChainResultHash [32]byte
@@ -461,6 +466,7 @@ type Chain interface {
 	sortition.Chain
 	GroupSelectionChain
 	DistributedKeyGenerationChain
+	InactivityClaimChain
 	BridgeChain
 	WalletProposalValidatorChain
 }

pkg/tbtc/chain_test.go

@@ -553,14 +553,15 @@ func (lc *localChain) DKGParameters() (*DKGParameters, error) {
 }
 
 func (lc *localChain) CalculateInactivityClaimSignatureHash(
-	nonce *big.Int,
-	walletPublicKey *ecdsa.PublicKey,
-	inactiveMembersIndexes []group.MemberIndex,
-	heartbeatFailed bool,
+	claim *inactivity.Claim,
 ) (inactivity.ClaimSignatureHash, error) {
 	panic("unsupported")
 }
 
+func (lc *localChain) GetInactivityClaimNonce(walletID [32]byte) (*big.Int, error) {
+	panic("unsupported")
+}
+
 func (lc *localChain) PastDepositRevealedEvents(
 	filter *DepositRevealedEventFilter,
 ) ([]*DepositRevealedEvent, error) {

pkg/tbtc/heartbeat.go

@@ -189,8 +189,12 @@ func (ha *heartbeatAction) execute() error {
 	// The value of consecutive heartbeat failures exceeds the threshold.
 	// Proceed with operator inactivity notification.
 	err = ha.inactivityClaimExecutor.publishClaim(
+		// Leave the list empty. Some operators were inactive during the
+		// heartbeat because they were simply unstaking and therefore should not
+		// be punished.
 		[]group.MemberIndex{},
 		true,
+		messageToSign,
 	)
 	if err != nil {
 		return fmt.Errorf(

pkg/tbtc/heartbeat_test.go

@@ -42,10 +42,7 @@ func TestHeartbeatAction_HappyPath(t *testing.T) {
 	hostChain.setHeartbeatProposalValidationResult(proposal, true)
 
 	mockExecutor := &mockHeartbeatSigningExecutor{}
-	inactivityNotifier := newInactivityClaimExecutor(
-		hostChain,
-		[]*signer{},
-	)
+	inactivityClaimExecutor := &inactivityClaimExecutor{}
 	action := newHeartbeatAction(
 		logger,
 		hostChain,
@@ -55,7 +52,7 @@ func TestHeartbeatAction_HappyPath(t *testing.T) {
 		mockExecutor,
 		proposal,
 		&heartbeatFailureCounter,
-		inactivityNotifier,
+		inactivityClaimExecutor,
 		startBlock,
 		expiryBlock,
 		func(ctx context.Context, blockHeight uint64) error {
@@ -109,10 +106,7 @@ func TestHeartbeatAction_SigningError(t *testing.T) {
 	mockExecutor := &mockHeartbeatSigningExecutor{}
 	mockExecutor.shouldFail = true
 
-	inactivityNotifier := newInactivityClaimExecutor(
-		hostChain,
-		[]*signer{},
-	)
+	inactivityClaimExecutor := &inactivityClaimExecutor{}
 
 	action := newHeartbeatAction(
 		logger,
@@ -123,7 +117,7 @@ func TestHeartbeatAction_SigningError(t *testing.T) {
 		mockExecutor,
 		proposal,
 		&heartbeatFailureCounter,
-		inactivityNotifier,
+		inactivityClaimExecutor,
 		startBlock,
 		expiryBlock,
 		func(ctx context.Context, blockHeight uint64) error {

pkg/tbtc/inactivity.go

@@ -2,41 +2,101 @@ package tbtc
 
 import (
 	"context"
+	"fmt"
 	"math/big"
 	"sync"
 
 	"github.com/ipfs/go-log/v2"
+	"go.uber.org/zap"
+	"golang.org/x/sync/semaphore"
+
+	"github.com/keep-network/keep-core/pkg/bitcoin"
 	"github.com/keep-network/keep-core/pkg/generator"
 	"github.com/keep-network/keep-core/pkg/net"
 	"github.com/keep-network/keep-core/pkg/protocol/group"
 	"github.com/keep-network/keep-core/pkg/tecdsa/inactivity"
 )
 
+// errInactivityClaimExecutorBusy is an error returned when the inactivity claim
+// executor cannot execute the inactivity claim due to another inactivity claim
+// execution in progress.
+var errInactivityClaimExecutorBusy = fmt.Errorf("inactivity claim executor is busy")
+
 type inactivityClaimExecutor struct {
-	chain   Chain
-	signers []*signer
+	lock *semaphore.Weighted
 
-	protocolLatch *generator.ProtocolLatch
+	chain               Chain
+	signers             []*signer
+	broadcastChannel    net.BroadcastChannel
+	membershipValidator *group.MembershipValidator
+	groupParameters     *GroupParameters
+	protocolLatch 		*generator.ProtocolLatch
+
+	waitForBlockFn waitForBlockFn
 }
 
 // TODO Consider moving all inactivity-related code to pkg/protocol/inactivity.
 func newInactivityClaimExecutor(
 	chain Chain,
 	signers []*signer,
+	broadcastChannel net.BroadcastChannel,
+	membershipValidator *group.MembershipValidator,
+	groupParameters *GroupParameters,
+	protocolLatch *generator.ProtocolLatch,
+	waitForBlockFn waitForBlockFn,
 ) *inactivityClaimExecutor {
 	return &inactivityClaimExecutor{
-		chain:   chain,
-		signers: signers,
+		lock:                semaphore.NewWeighted(1),
+		chain:               chain,
+		signers:             signers,
+		broadcastChannel:    broadcastChannel,
+		membershipValidator: membershipValidator,
+		groupParameters:     groupParameters,
+		protocolLatch:       protocolLatch,
+		waitForBlockFn:      waitForBlockFn,
 	}
 }
 
 func (ice *inactivityClaimExecutor) publishClaim(
 	inactiveMembersIndexes []group.MemberIndex,
 	heartbeatFailed bool,
+	message *big.Int,
 ) error {
-	// TODO: Build a claim and launch the publish function for all
-	//       the signers. The value of `heartbeat` should be true and
-	//       `inactiveMembersIndices` should be empty.
+	if lockAcquired := ice.lock.TryAcquire(1); !lockAcquired {
+		return errInactivityClaimExecutorBusy
+	}
+	defer ice.lock.Release(1)
+
+	wallet := ice.wallet()
+
+	walletPublicKeyHash := bitcoin.PublicKeyHash(wallet.publicKey)
+	walletPublicKeyBytes, err := marshalPublicKey(wallet.publicKey)
+	if err != nil {
+		return fmt.Errorf("cannot marshal wallet public key: [%v]", err)
+	}
+
+	execLogger := logger.With(
+		zap.String("wallet", fmt.Sprintf("0x%x", walletPublicKeyBytes)),
+	)
+
+	walletRegistryData, err := ice.chain.GetWallet(walletPublicKeyHash)
+	if err != nil {
+		return fmt.Errorf("could not get registry data on wallet: [%v]", err)
+	}
+
+	nonce, err := ice.chain.GetInactivityClaimNonce(
+		walletRegistryData.EcdsaWalletID,
+	)
+	if err != nil {
+		return fmt.Errorf("could not get nonce for wallet: [%v]", err)
+	}
+
+	claim := &inactivity.Claim{
+		Nonce:                  nonce,
+		WalletPublicKey:        wallet.publicKey,
+		InactiveMembersIndexes: inactiveMembersIndexes,
+		HeartbeatFailed:        heartbeatFailed,
+	}
 
 	wg := sync.WaitGroup{}
 	wg.Add(len(ice.signers))
@@ -45,11 +105,37 @@ func (ice *inactivityClaimExecutor) publishClaim(
 		ice.protocolLatch.Lock()
 		defer ice.protocolLatch.Unlock()
 
+		defer wg.Done()
+
+		inactivityClaimTimeoutBlock := uint64(0) // TODO: Set the value of timeout block
+
 		go func(signer *signer) {
-			// TODO: Launch claim publishing for members.
+			ctx, cancelCtx := withCancelOnBlock(
+				context.Background(),
+				inactivityClaimTimeoutBlock,
+				ice.waitForBlockFn,
+			)
+			defer cancelCtx()
+
+			ice.publish(
+				ctx,
+				execLogger,
+				message,
+				signer.signingGroupMemberIndex,
+				wallet.groupSize(),
+				wallet.groupDishonestThreshold(
+					ice.groupParameters.HonestThreshold,
+				),
+				ice.membershipValidator,
+				claim,
+			)
+
 		}(currentSigner)
 	}
 
+	// Wait until all controlled signers complete their routine.
+	wg.Wait()
+
 	return nil
 }
 
@@ -58,7 +144,6 @@ func (ice *inactivityClaimExecutor) publish(
 	inactivityLogger log.StandardLogger,
 	seed *big.Int,
 	memberIndex group.MemberIndex,
-	broadcastChannel net.BroadcastChannel,
 	groupSize int,
 	dishonestThreshold int,
 	membershipValidator *group.MembershipValidator,
@@ -69,7 +154,7 @@ func (ice *inactivityClaimExecutor) publish(
 		inactivityLogger,
 		seed.Text(16),
 		memberIndex,
-		broadcastChannel,
+		ice.broadcastChannel,
 		groupSize,
 		dishonestThreshold,
 		membershipValidator,
@@ -78,3 +163,9 @@ func (ice *inactivityClaimExecutor) publish(
 		inactivityClaim,
 	)
 }
+
+func (ice *inactivityClaimExecutor) wallet() wallet {
+	// All signers belong to one wallet. Take that wallet from the
+	// first signer.
+	return ice.signers[0].wallet
+}

pkg/tbtc/inactivity_submit.go

@@ -30,12 +30,7 @@ func (ics *inactivityClaimSigner) SignClaim(claim *inactivity.Claim) (
 		return nil, fmt.Errorf("result is nil")
 	}
 
-	claimHash, err := ics.chain.CalculateInactivityClaimSignatureHash(
-		claim.Nonce,
-		claim.WalletPublicKey,
-		claim.InactiveMembersIndexes,
-		claim.HeartbeatFailed,
-	)
+	claimHash, err := ics.chain.CalculateInactivityClaimSignatureHash(claim)
 	if err != nil {
 		return nil, fmt.Errorf(
 			"inactivity claim hash calculation failed [%w]",

pkg/tbtc/node.go

@@ -524,6 +524,11 @@ func (n *node) getInactivityClaimExecutor(
 	executor := newInactivityClaimExecutor(
 		n.chain,
 		signers,
+		broadcastChannel,
+		membershipValidator,
+		n.groupParameters,
+		n.protocolLatch,
+		n.waitForBlockHeight,
 	)
 
 	n.inactivityClaimExecutors[executorKey] = executor