fix: propagate context through tx, DB, and signing call chains

- Thread ctx through Submitter.RunEpoch/GetPayload so channel reads
  respect cancellation (mirrors fix already applied to SignatureSubmitter)
- Add ctx to chain.Client interface (SendRawTx, Nonce) and all tx_utils
  functions so blockchain operations derive timeouts from the parent
  context instead of context.Background()
- Add ctx to finalizerDB and epochClientDB interfaces; pass through to
  database calls replacing context.TODO()
- Add ctx to SetGas, RegisterVoter, PreregisterVoter, SignNewSigningPolicy,
  SignUptimeVote, SignRewards so the epoch signing chain is cancellable

Author: GrePod

client/epoch/db.go

@@ -10,17 +10,17 @@ import (
 )
 
 type epochClientDB interface {
-	FetchLogsByAddressAndTopic0Timestamp(common.Address, common.Hash, int64, int64) ([]database.Log, error)
+	FetchLogsByAddressAndTopic0Timestamp(ctx context.Context, address common.Address, topic0 common.Hash, from, to int64) ([]database.Log, error)
 }
 
 type epochClientDBGorm struct {
 	db *gorm.DB
 }
 
 func (g epochClientDBGorm) FetchLogsByAddressAndTopic0Timestamp(
-	address common.Address, topic0 common.Hash, fromTimestamp int64, toTimestamp int64,
+	ctx context.Context, address common.Address, topic0 common.Hash, fromTimestamp int64, toTimestamp int64,
 ) ([]database.Log, error) {
-	return database.FetchLogsByAddressAndTopic0Timestamp(context.TODO(), g.db, database.LogsParams{
+	return database.FetchLogsByAddressAndTopic0Timestamp(ctx, g.db, database.LogsParams{
 		Address: address,
 		Topic0:  topic0,
 		From:    fromTimestamp,

client/epoch/epoch_client.go

@@ -162,58 +162,58 @@ func (c *client) Run(ctx context.Context) error {
 		select {
 		case rewardEpochStarted := <-epochStartedListener:
 			logger.Debugf("RewardEpochStarted event emitted for epoch %v", rewardEpochStarted.RewardEpochId)
-			c.preregisterVoter(new(big.Int).Add(rewardEpochStarted.RewardEpochId, big.NewInt(1)))
+			c.preregisterVoter(ctx, new(big.Int).Add(rewardEpochStarted.RewardEpochId, big.NewInt(1)))
 		case powerBlockData := <-vpbsListener:
 			logger.Debugf("VotePowerBlockSelected event emitted for epoch %v", powerBlockData.RewardEpochId)
-			c.registerVoter(powerBlockData.RewardEpochId)
+			c.registerVoter(ctx, powerBlockData.RewardEpochId)
 		case signingPolicy := <-policyListener:
 			logger.Debugf("SigningPolicyInitialized event emitted for epoch %v", signingPolicy.RewardEpochId)
-			c.signPolicy(signingPolicy.RewardEpochId, signingPolicy.SigningPolicyBytes)
+			c.signPolicy(ctx, signingPolicy.RewardEpochId, signingPolicy.SigningPolicyBytes)
 		case uptimeVoteEnabled := <-uptimeEnabledListener:
 			logger.Debugf("SignUptimeVoteEnabled event emitted for epoch %v", uptimeVoteEnabled.RewardEpochId)
-			c.signUptimeVote(uptimeVoteEnabled.RewardEpochId)
+			c.signUptimeVote(ctx, uptimeVoteEnabled.RewardEpochId)
 		case uptimeVoteSigned := <-uptimeSignedListener:
 			logger.Infof("Uptime vote threshold reached for epoch %v, signing rewards", uptimeVoteSigned.RewardEpochId)
-			c.signRewards(uptimeVoteSigned.RewardEpochId)
+			c.signRewards(ctx, uptimeVoteSigned.RewardEpochId)
 		case <-ctx.Done():
 			return ctx.Err()
 		}
 	}
 }
 
-func (c *client) registerVoter(epochID *big.Int) {
+func (c *client) registerVoter(ctx context.Context, epochID *big.Int) {
 	if !c.isFutureEpoch(epochID) {
 		logger.Debugf("Skipping registration process for old epoch %v", epochID)
 		return
 	}
 
 	logger.Infof("VotePowerBlockSelected event emitted for next epoch %v, starting registration", epochID)
-	registerResult := <-c.registryClient.RegisterVoter(epochID, c.identityAddress)
+	registerResult := <-c.registryClient.RegisterVoter(ctx, epochID, c.identityAddress)
 	if !registerResult.Success {
 		logger.Errorf("RegisterVoter failed %s", registerResult.Message)
 	}
 }
 
-func (c *client) preregisterVoter(epochID *big.Int) {
+func (c *client) preregisterVoter(ctx context.Context, epochID *big.Int) {
 	if !c.isFutureEpoch(epochID) {
 		logger.Debugf("Skipping pre-registration process for old epoch %v", epochID)
 		return
 	}
 
-	registerResult := <-c.registryClient.PreregisterVoter(epochID, c.identityAddress)
+	registerResult := <-c.registryClient.PreregisterVoter(ctx, epochID, c.identityAddress)
 	if !registerResult.Success {
 		logger.Errorf("PreregisterVoter failed %s", registerResult.Message)
 	}
 }
 
-func (c *client) signPolicy(epochID *big.Int, policy []byte) {
+func (c *client) signPolicy(ctx context.Context, epochID *big.Int, policy []byte) {
 	if !c.isFutureEpoch(epochID) {
 		logger.Debugf("Skipping policy signing for old epoch %v", epochID)
 		return
 	}
 
 	logger.Infof("SigningPolicyInitialized event emitted for next epoch %v, signing new policy", epochID)
-	signingResult := <-c.systemsManagerClient.SignNewSigningPolicy(epochID, policy)
+	signingResult := <-c.systemsManagerClient.SignNewSigningPolicy(ctx, epochID, policy)
 	if signingResult.Success {
 		logger.Info("SignNewSigningPolicy success")
 	} else {
@@ -222,8 +222,8 @@ func (c *client) signPolicy(epochID *big.Int, policy []byte) {
 	}
 }
 
-func (c *client) signUptimeVote(epochId *big.Int) {
-	signUptimeVoteResult := <-c.systemsManagerClient.SignUptimeVote(epochId)
+func (c *client) signUptimeVote(ctx context.Context, epochId *big.Int) {
+	signUptimeVoteResult := <-c.systemsManagerClient.SignUptimeVote(ctx, epochId)
 	if signUptimeVoteResult.Success {
 		logger.Info("SignUptimeVote completed")
 	} else {
@@ -261,7 +261,7 @@ func (c *client) isFutureEpoch(epochID *big.Int) bool {
 //
 // Since reward claim data is currently published manually, and it might take a day or so for the data to be available,
 // a retry mechanism is employed with a large retry interval (configurable).
-func (c *client) signRewards(epochId *big.Int) {
+func (c *client) signRewards(ctx context.Context, epochId *big.Int) {
 	res := shared.ExecuteWithRetryAttempts(func(i int) (*struct{}, error) {
 		if c.systemsManagerClient.IsRewardHashSigned(epochId) {
 			return nil, nil
@@ -280,7 +280,7 @@ func (c *client) signRewards(epochId *big.Int) {
 		if err != nil {
 			return nil, errors.Wrapf(err, "reward data verification for epoch %d failed", epochId)
 		}
-		signingResult := <-c.systemsManagerClient.SignRewards(epochId, hash, weightClaims)
+		signingResult := <-c.systemsManagerClient.SignRewards(ctx, epochId, hash, weightClaims)
 		if !signingResult.Success {
 			return nil, errors.Errorf("unable to send reward signature")
 		}

client/epoch/epoch_client_test.go

@@ -258,7 +258,7 @@ func TestEpochClientRegisterErr(t *testing.T) {
 type testDB struct{}
 
 func (db testDB) FetchLogsByAddressAndTopic0Timestamp(
-	address common.Address, topic0 common.Hash, from, to int64,
+	ctx context.Context, address common.Address, topic0 common.Hash, from, to int64,
 ) ([]database.Log, error) {
 	return nil, errors.New("not implemented")
 }
@@ -297,7 +297,7 @@ func (c testSystemsManagerClient) VotePowerBlockSelectedListener(
 }
 
 func (c testSystemsManagerClient) SignNewSigningPolicy(
-	epochID *big.Int, policy []byte,
+	_ context.Context, epochID *big.Int, policy []byte,
 ) <-chan shared.ExecuteStatus[any] {
 	return shared.ExecuteWithRetryChan(func() (any, error) {
 		if c.signingErr != nil {
@@ -358,7 +358,7 @@ func newTestRegistryClient() testRegistryClient {
 }
 
 func (c testRegistryClient) RegisterVoter(
-	epochID *big.Int, address common.Address,
+	_ context.Context, epochID *big.Int, address common.Address,
 ) <-chan shared.ExecuteStatus[any] {
 	return shared.ExecuteWithRetryChan(func() (any, error) {
 		if c.registerErr != nil {
@@ -385,7 +385,7 @@ func (c testSystemsManagerClient) RewardEpochStartedListener(db epochClientDB, c
 	return make(chan *system.FlareSystemsManagerRewardEpochStarted)
 }
 
-func (c testRegistryClient) PreregisterVoter(nextRewardEpochId *big.Int, address common.Address) <-chan shared.ExecuteStatus[any] {
+func (c testRegistryClient) PreregisterVoter(_ context.Context, nextRewardEpochId *big.Int, address common.Address) <-chan shared.ExecuteStatus[any] {
 	return shared.ExecuteWithRetryChan(func() (any, error) {
 		return nil, nil
 	}, 1, 0)
@@ -395,7 +395,7 @@ func (c testSystemsManagerClient) SignUptimeVoteEnabledListener(db epochClientDB
 	return make(chan *system.FlareSystemsManagerSignUptimeVoteEnabled)
 }
 
-func (c testSystemsManagerClient) SignUptimeVote(b *big.Int) <-chan shared.ExecuteStatus[any] {
+func (c testSystemsManagerClient) SignUptimeVote(_ context.Context, b *big.Int) <-chan shared.ExecuteStatus[any] {
 	return shared.ExecuteWithRetryChan(func() (any, error) {
 		return nil, nil
 	}, 1, 0)
@@ -405,7 +405,7 @@ func (c testSystemsManagerClient) UptimeVoteSignedListener(db epochClientDB, epo
 	return make(chan *system.FlareSystemsManagerUptimeVoteSigned)
 }
 
-func (c testSystemsManagerClient) SignRewards(b *big.Int, hash *common.Hash, claims int) <-chan shared.ExecuteStatus[any] {
+func (c testSystemsManagerClient) SignRewards(_ context.Context, b *big.Int, hash *common.Hash, claims int) <-chan shared.ExecuteStatus[any] {
 	return shared.ExecuteWithRetryChan(func() (any, error) {
 		return nil, nil
 	}, 1, 0)

client/epoch/register_utils_test.go

@@ -1,6 +1,7 @@
 package epoch_test
 
 import (
+	"context"
 	"math/big"
 	"testing"
 
@@ -45,7 +46,7 @@ func TestSetGas(t *testing.T) {
 		txOptions := new(bind.TransactOpts)
 		txOptions.From = address
 
-		err = epoch.SetGas(txOptions, cl, &test.gasConfig)
+		err = epoch.SetGas(context.Background(), txOptions, cl, &test.gasConfig)
 
 		require.Equal(t, address, txOptions.From)
 

client/epoch/registry_utils.go

@@ -103,8 +103,8 @@ func init() {
 }
 
 type registryContractClient interface {
-	RegisterVoter(nextRewardEpochId *big.Int, address common.Address) <-chan shared.ExecuteStatus[any]
-	PreregisterVoter(nextRewardEpochId *big.Int, address common.Address) <-chan shared.ExecuteStatus[any]
+	RegisterVoter(ctx context.Context, nextRewardEpochId *big.Int, address common.Address) <-chan shared.ExecuteStatus[any]
+	PreregisterVoter(ctx context.Context, nextRewardEpochId *big.Int, address common.Address) <-chan shared.ExecuteStatus[any]
 }
 
 type registryContractClientImpl struct {
@@ -153,9 +153,9 @@ func NewRegistryContractClient(
 }
 
 // RegisterVoter tries to register voter on VoterRegistry smart contract.
-func (r *registryContractClientImpl) RegisterVoter(nextRewardEpochID *big.Int, address common.Address) <-chan shared.ExecuteStatus[any] {
+func (r *registryContractClientImpl) RegisterVoter(ctx context.Context, nextRewardEpochID *big.Int, address common.Address) <-chan shared.ExecuteStatus[any] {
 	return shared.ExecuteWithRetryChan(func() (any, error) {
-		err := r.sendRegisterVoter(nextRewardEpochID, address)
+		err := r.sendRegisterVoter(ctx, nextRewardEpochID, address)
 		if err != nil {
 			if shared.ExistsAsSubstring(nonFatalRegisterErrors, err.Error()) {
 				logger.Debugf("Non fatal error sending register voter: %v", err)
@@ -167,7 +167,7 @@ func (r *registryContractClientImpl) RegisterVoter(nextRewardEpochID *big.Int, a
 	}, shared.MaxTxSendRetriesLong, shared.TxRetryIntervalLong)
 }
 
-func (r *registryContractClientImpl) sendRegisterVoter(nextRewardEpochID *big.Int, address common.Address) error {
+func (r *registryContractClientImpl) sendRegisterVoter(ctx context.Context, nextRewardEpochID *big.Int, address common.Address) error {
 	epochID := uint32(nextRewardEpochID.Uint64())
 
 	var (
@@ -193,12 +193,13 @@ func (r *registryContractClientImpl) sendRegisterVoter(nextRewardEpochID *big.In
 		V: signature[64] + 27,
 	}
 
-	err = SetGas(r.senderTxOpts, r.ethClient, r.gasCfg)
+	err = SetGas(ctx, r.senderTxOpts, r.ethClient, r.gasCfg)
 	if err != nil {
 		return err
 	}
 
 	estimatedGasLimit, err := chain.DryRunTxAbi(
+		ctx,
 		r.ethClient,
 		chain.DefaultTxTimeout,
 		r.senderTxOpts.From,
@@ -224,7 +225,7 @@ func (r *registryContractClientImpl) sendRegisterVoter(nextRewardEpochID *big.In
 		return fmt.Errorf("sending registry tx: %w", err)
 	}
 
-	err = r.txVerifier.WaitUntilMined(r.senderTxOpts.From, tx, chain.DefaultTxTimeout)
+	err = r.txVerifier.WaitUntilMined(ctx, r.senderTxOpts.From, tx, chain.DefaultTxTimeout)
 	if err != nil {
 		return err
 	}
@@ -233,9 +234,9 @@ func (r *registryContractClientImpl) sendRegisterVoter(nextRewardEpochID *big.In
 }
 
 // PreregisterVoter tries to pre-register voter on VoterPreRegistry smart contract.
-func (r *registryContractClientImpl) PreregisterVoter(nextRewardEpochId *big.Int, address common.Address) <-chan shared.ExecuteStatus[any] {
+func (r *registryContractClientImpl) PreregisterVoter(ctx context.Context, nextRewardEpochId *big.Int, address common.Address) <-chan shared.ExecuteStatus[any] {
 	return shared.ExecuteWithRetryChan(func() (any, error) {
-		err := r.sendPreRegisterVoter(nextRewardEpochId, address)
+		err := r.sendPreRegisterVoter(ctx, nextRewardEpochId, address)
 		if err != nil {
 			if shared.ExistsAsSubstring(nonFatalPreregisterErrors, err.Error()) {
 				logger.Debugf("Non fatal error sending pre-register voter: %v", err)
@@ -247,7 +248,7 @@ func (r *registryContractClientImpl) PreregisterVoter(nextRewardEpochId *big.Int
 	}, shared.MaxTxSendRetries, shared.TxRetryInterval)
 }
 
-func (r *registryContractClientImpl) sendPreRegisterVoter(nextRewardEpochID *big.Int, address common.Address) error {
+func (r *registryContractClientImpl) sendPreRegisterVoter(ctx context.Context, nextRewardEpochID *big.Int, address common.Address) error {
 	epochID := uint32(nextRewardEpochID.Uint64())
 
 	var (
@@ -273,12 +274,13 @@ func (r *registryContractClientImpl) sendPreRegisterVoter(nextRewardEpochID *big
 		V: signature[64] + 27,
 	}
 
-	err = SetGas(r.senderTxOpts, r.ethClient, r.gasCfg)
+	err = SetGas(ctx, r.senderTxOpts, r.ethClient, r.gasCfg)
 	if err != nil {
 		return fmt.Errorf("setting gas pre registry:%w", err)
 	}
 
 	estimatedGasLimit, err := chain.DryRunTxAbi(
+		ctx,
 		r.ethClient,
 		chain.DefaultTxTimeout,
 		r.senderTxOpts.From,
@@ -304,7 +306,7 @@ func (r *registryContractClientImpl) sendPreRegisterVoter(nextRewardEpochID *big
 		return fmt.Errorf("sending preregistry tx: %w", err)
 	}
 
-	err = r.txVerifier.WaitUntilMined(r.senderTxOpts.From, tx, chain.DefaultTxTimeout)
+	err = r.txVerifier.WaitUntilMined(ctx, r.senderTxOpts.From, tx, chain.DefaultTxTimeout)
 	if err != nil {
 		return err
 	}
@@ -335,19 +337,19 @@ func (r *registryContractClientImpl) createSignatureNew(chainID int, nextRewardE
 }
 
 // SetGas sets gas parameters in txOptions according to the gasConfig.
-func SetGas(txOptions *bind.TransactOpts, client *ethclient.Client, gasConfig *config.Gas) error {
+func SetGas(ctx context.Context, txOptions *bind.TransactOpts, client *ethclient.Client, gasConfig *config.Gas) error {
 	switch gasConfig.TxType {
 	case 0:
-		gasPrice, err := chain.GetGasPrice(gasConfig, client, chain.DefaultTxTimeout)
+		gasPrice, err := chain.GetGasPrice(ctx, gasConfig, client, chain.DefaultTxTimeout)
 		if err != nil {
 			logger.Warnf("Unable to obtain gas price: %v, using fallback %d", err, fallbackGasPrice)
 			gasPrice = new(big.Int).Set(fallbackGasPrice)
 		}
 		txOptions.GasPrice = gasPrice
 		return nil
 	case 2:
-		ctx, cancelFunc := context.WithTimeout(context.Background(), timeout)
-		baseFeePerGas, err := chain.BaseFee(ctx, client)
+		feeCtx, cancelFunc := context.WithTimeout(ctx, chain.DefaultTxTimeout)
+		baseFeePerGas, err := chain.BaseFee(feeCtx, client)
 		cancelFunc()
 
 		if err != nil {

client/epoch/relay_utils.go

@@ -1,6 +1,7 @@
 package epoch
 
 import (
+	"context"
 	"time"
 
 	"github.com/flare-foundation/flare-system-client/client/shared"
@@ -56,7 +57,7 @@ func (r *relayContractClientImpl) SigningPolicyInitializedListener(db epochClien
 		for {
 			<-ticker.C
 			now := time.Now().Unix()
-			logs, err := db.FetchLogsByAddressAndTopic0Timestamp(r.address, topic0, eventRangeStart, now)
+			logs, err := db.FetchLogsByAddressAndTopic0Timestamp(context.Background(), r.address, topic0, eventRangeStart, now)
 			if err != nil {
 				logger.Errorf("Error fetching logs %v", err)
 				continue