fix: propagate context through event listener goroutines and delayed queue

All background polling goroutines (5 listener loops across epoch and
finalizer packages) now receive ctx and exit on cancellation instead of
running forever. DelayedQueueManager.Add takes ctx so timer goroutines
also respect cancellation. FetchSigningPolicies and processDelayedQueue
no longer use context.Background().

Author: GrePod

client/epoch/epoch_client.go

@@ -142,20 +142,20 @@ func (c *client) Run(ctx context.Context) error {
 	var uptimeSignedListener <-chan *system.FlareSystemsManagerUptimeVoteSigned
 
 	if c.preregistrationEnabled {
-		epochStartedListener = c.systemsManagerClient.RewardEpochStartedListener(c.db, rewardEpochTiming)
+		epochStartedListener = c.systemsManagerClient.RewardEpochStartedListener(ctx, c.db, rewardEpochTiming)
 	}
 	if c.registrationEnabled {
 		logger.Info("Waiting for VotePowerBlockSelected event to start registration")
-		vpbsListener = c.systemsManagerClient.VotePowerBlockSelectedListener(c.db, rewardEpochTiming)
-		policyListener = c.relayClient.SigningPolicyInitializedListener(c.db, rewardEpochTiming)
+		vpbsListener = c.systemsManagerClient.VotePowerBlockSelectedListener(ctx, c.db, rewardEpochTiming)
+		policyListener = c.relayClient.SigningPolicyInitializedListener(ctx, c.db, rewardEpochTiming)
 	}
 	if c.uptimeVotingEnabled {
 		logger.Info("Waiting for SignUptimeVoteEnabled event to start uptime vote signing")
-		uptimeEnabledListener = c.systemsManagerClient.SignUptimeVoteEnabledListener(c.db, rewardEpochTiming)
+		uptimeEnabledListener = c.systemsManagerClient.SignUptimeVoteEnabledListener(ctx, c.db, rewardEpochTiming)
 	}
 	if c.rewardsSigningEnabled {
 		logger.Info("Waiting for UptimeVoteSigned event to start rewards signing")
-		uptimeSignedListener = c.systemsManagerClient.UptimeVoteSignedListener(c.db, rewardEpochTiming)
+		uptimeSignedListener = c.systemsManagerClient.UptimeVoteSignedListener(ctx, c.db, rewardEpochTiming)
 	}
 
 	for {

client/epoch/epoch_client_test.go

@@ -291,7 +291,7 @@ func (c testSystemsManagerClient) RewardEpochTimingFromChain() (*utils.EpochTimi
 }
 
 func (c testSystemsManagerClient) VotePowerBlockSelectedListener(
-	db epochClientDB, rewardEpochTiming *utils.EpochTimingConfig,
+	_ context.Context, db epochClientDB, rewardEpochTiming *utils.EpochTimingConfig,
 ) <-chan *system.FlareSystemsManagerVotePowerBlockSelected {
 	return c.vpbsChan
 }
@@ -341,7 +341,7 @@ func (c testRelayClient) sendTestPolicy(policy *relay.RelaySigningPolicyInitiali
 }
 
 func (c testRelayClient) SigningPolicyInitializedListener(
-	db epochClientDB, rewardEpochTiming *utils.EpochTimingConfig,
+	_ context.Context, db epochClientDB, rewardEpochTiming *utils.EpochTimingConfig,
 ) <-chan *relay.RelaySigningPolicyInitialized {
 	return c.policyChan
 }
@@ -381,7 +381,7 @@ func (c testRegistryClient) RegisterVoter(
 	}, 1, 0)
 }
 
-func (c testSystemsManagerClient) RewardEpochStartedListener(db epochClientDB, config *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerRewardEpochStarted {
+func (c testSystemsManagerClient) RewardEpochStartedListener(_ context.Context, db epochClientDB, config *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerRewardEpochStarted {
 	return make(chan *system.FlareSystemsManagerRewardEpochStarted)
 }
 
@@ -391,7 +391,7 @@ func (c testRegistryClient) PreregisterVoter(_ context.Context, nextRewardEpochI
 	}, 1, 0)
 }
 
-func (c testSystemsManagerClient) SignUptimeVoteEnabledListener(db epochClientDB, epoch *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerSignUptimeVoteEnabled {
+func (c testSystemsManagerClient) SignUptimeVoteEnabledListener(_ context.Context, db epochClientDB, epoch *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerSignUptimeVoteEnabled {
 	return make(chan *system.FlareSystemsManagerSignUptimeVoteEnabled)
 }
 
@@ -401,7 +401,7 @@ func (c testSystemsManagerClient) SignUptimeVote(_ context.Context, b *big.Int)
 	}, 1, 0)
 }
 
-func (c testSystemsManagerClient) UptimeVoteSignedListener(db epochClientDB, epoch *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerUptimeVoteSigned {
+func (c testSystemsManagerClient) UptimeVoteSignedListener(_ context.Context, db epochClientDB, epoch *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerUptimeVoteSigned {
 	return make(chan *system.FlareSystemsManagerUptimeVoteSigned)
 }
 

client/epoch/relay_utils.go

@@ -18,7 +18,7 @@ import (
 )
 
 type relayContractClient interface {
-	SigningPolicyInitializedListener(epochClientDB, *utils.EpochTimingConfig) <-chan *relay.RelaySigningPolicyInitialized
+	SigningPolicyInitializedListener(ctx context.Context, db epochClientDB, config *utils.EpochTimingConfig) <-chan *relay.RelaySigningPolicyInitialized
 }
 
 type relayContractClientImpl struct {
@@ -42,7 +42,7 @@ func NewRelayContractClient(
 	}, nil
 }
 
-func (r *relayContractClientImpl) SigningPolicyInitializedListener(db epochClientDB, rewardEpochTiming *utils.EpochTimingConfig) <-chan *relay.RelaySigningPolicyInitialized {
+func (r *relayContractClientImpl) SigningPolicyInitializedListener(ctx context.Context, db epochClientDB, rewardEpochTiming *utils.EpochTimingConfig) <-chan *relay.RelaySigningPolicyInitialized {
 	topic0, err := chain.EventIDFromMetadata(relay.RelayMetaData, "SigningPolicyInitialized")
 	if err != nil {
 		// panic, this error is fatal
@@ -55,9 +55,13 @@ func (r *relayContractClientImpl) SigningPolicyInitializedListener(db epochClien
 		ticker := time.NewTicker(shared.EventListenerInterval)
 		eventRangeStart := rewardEpochTiming.StartTime(rewardEpochTiming.EpochIndex(time.Now()) - 1).Unix()
 		for {
-			<-ticker.C
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
 			now := time.Now().Unix()
-			logs, err := db.FetchLogsByAddressAndTopic0Timestamp(context.Background(), r.address, topic0, eventRangeStart, now)
+			logs, err := db.FetchLogsByAddressAndTopic0Timestamp(ctx, r.address, topic0, eventRangeStart, now)
 			if err != nil {
 				logger.Errorf("Error fetching logs %v", err)
 				continue

client/epoch/system_manager_utils.go

@@ -52,15 +52,15 @@ func init() {
 type systemsManagerContractClient interface {
 	RewardEpochTimingFromChain() (*utils.EpochTimingConfig, error)
 
-	RewardEpochStartedListener(epochClientDB, *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerRewardEpochStarted
+	RewardEpochStartedListener(ctx context.Context, db epochClientDB, config *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerRewardEpochStarted
 
-	VotePowerBlockSelectedListener(epochClientDB, *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerVotePowerBlockSelected
+	VotePowerBlockSelectedListener(ctx context.Context, db epochClientDB, config *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerVotePowerBlockSelected
 	SignNewSigningPolicy(ctx context.Context, epochID *big.Int, policy []byte) <-chan shared.ExecuteStatus[any]
 
-	SignUptimeVoteEnabledListener(epochClientDB, *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerSignUptimeVoteEnabled
+	SignUptimeVoteEnabledListener(ctx context.Context, db epochClientDB, config *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerSignUptimeVoteEnabled
 	SignUptimeVote(ctx context.Context, epochID *big.Int) <-chan shared.ExecuteStatus[any]
 
-	UptimeVoteSignedListener(epochClientDB, *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerUptimeVoteSigned
+	UptimeVoteSignedListener(ctx context.Context, db epochClientDB, config *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerUptimeVoteSigned
 	SignRewards(ctx context.Context, epochID *big.Int, rewardHash *common.Hash, weightClaims int) <-chan shared.ExecuteStatus[any]
 	IsRewardHashSigned(*big.Int) bool
 
@@ -190,7 +190,7 @@ func (s *systemsManagerContractClientImpl) GetCurrentRewardEpochID() <-chan shar
 	}, shared.MaxTxSendRetries, shared.TxRetryInterval)
 }
 
-func (s *systemsManagerContractClientImpl) RewardEpochStartedListener(db epochClientDB, rewardEpochTiming *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerRewardEpochStarted {
+func (s *systemsManagerContractClientImpl) RewardEpochStartedListener(ctx context.Context, db epochClientDB, rewardEpochTiming *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerRewardEpochStarted {
 	out := make(chan *system.FlareSystemsManagerRewardEpochStarted)
 	topic0, err := chain.EventIDFromMetadata(system.FlareSystemsManagerMetaData, "RewardEpochStarted")
 	if err != nil {
@@ -202,9 +202,13 @@ func (s *systemsManagerContractClientImpl) RewardEpochStartedListener(db epochCl
 		ticker := time.NewTicker(shared.EventListenerInterval)
 		eventRangeStart := rewardEpochTiming.StartTime(rewardEpochTiming.EpochIndex(time.Now())).Unix() - 60*60 // Expected epoch start - 1h
 		for {
-			<-ticker.C
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
 			now := time.Now().Unix()
-			logs, err := db.FetchLogsByAddressAndTopic0Timestamp(context.Background(), s.address, topic0, eventRangeStart, now)
+			logs, err := db.FetchLogsByAddressAndTopic0Timestamp(ctx, s.address, topic0, eventRangeStart, now)
 			if err != nil {
 				logger.Errorf("Error fetching logs %v", err)
 				continue
@@ -231,7 +235,7 @@ func (s *systemsManagerContractClientImpl) parseRewardEpochStartedEvent(dbLog da
 	return s.flareSystemsManager.ParseRewardEpochStarted(*contractLog)
 }
 
-func (s *systemsManagerContractClientImpl) VotePowerBlockSelectedListener(db epochClientDB, rewardEpochTiming *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerVotePowerBlockSelected {
+func (s *systemsManagerContractClientImpl) VotePowerBlockSelectedListener(ctx context.Context, db epochClientDB, rewardEpochTiming *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerVotePowerBlockSelected {
 	out := make(chan *system.FlareSystemsManagerVotePowerBlockSelected)
 	topic0, err := chain.EventIDFromMetadata(system.FlareSystemsManagerMetaData, "VotePowerBlockSelected")
 	if err != nil {
@@ -243,9 +247,13 @@ func (s *systemsManagerContractClientImpl) VotePowerBlockSelectedListener(db epo
 		ticker := time.NewTicker(shared.EventListenerInterval)
 		eventRangeStart := rewardEpochTiming.StartTime(rewardEpochTiming.EpochIndex(time.Now()) - 1).Unix()
 		for {
-			<-ticker.C
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
 			now := time.Now().Unix()
-			logs, err := db.FetchLogsByAddressAndTopic0Timestamp(context.Background(), s.address, topic0, eventRangeStart, now)
+			logs, err := db.FetchLogsByAddressAndTopic0Timestamp(ctx, s.address, topic0, eventRangeStart, now)
 			if err != nil {
 				logger.Errorf("Error fetching logs %v", err)
 				continue
@@ -276,7 +284,7 @@ func (s *systemsManagerContractClientImpl) RewardEpochTimingFromChain() (*utils.
 	return shared.RewardEpochTimingFromChain(s.flareSystemsManager)
 }
 
-func (s *systemsManagerContractClientImpl) SignUptimeVoteEnabledListener(db epochClientDB, epoch *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerSignUptimeVoteEnabled {
+func (s *systemsManagerContractClientImpl) SignUptimeVoteEnabledListener(ctx context.Context, db epochClientDB, epoch *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerSignUptimeVoteEnabled {
 	out := make(chan *system.FlareSystemsManagerSignUptimeVoteEnabled)
 	topic0, err := chain.EventIDFromMetadata(system.FlareSystemsManagerMetaData, "SignUptimeVoteEnabled")
 	if err != nil {
@@ -289,11 +297,15 @@ func (s *systemsManagerContractClientImpl) SignUptimeVoteEnabledListener(db epoc
 		startEpoch := epoch.EpochIndex(time.Now())
 		eventRangeStart := epoch.StartTime(startEpoch).Unix()
 		for {
-			<-ticker.C
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
 			now := time.Now()
 			currentEpoch := epoch.EpochIndex(now)
 
-			logs, err := db.FetchLogsByAddressAndTopic0Timestamp(context.Background(), s.address, topic0, eventRangeStart, now.Unix())
+			logs, err := db.FetchLogsByAddressAndTopic0Timestamp(ctx, s.address, topic0, eventRangeStart, now.Unix())
 			if err != nil {
 				logger.Errorf("Error fetching logs %v", err)
 				continue
@@ -377,7 +389,7 @@ func (s *systemsManagerContractClientImpl) sendSignUptimeVote(ctx context.Contex
 	return nil
 }
 
-func (s *systemsManagerContractClientImpl) UptimeVoteSignedListener(db epochClientDB, epoch *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerUptimeVoteSigned {
+func (s *systemsManagerContractClientImpl) UptimeVoteSignedListener(ctx context.Context, db epochClientDB, epoch *utils.EpochTimingConfig) <-chan *system.FlareSystemsManagerUptimeVoteSigned {
 	out := make(chan *system.FlareSystemsManagerUptimeVoteSigned)
 	topic0, err := chain.EventIDFromMetadata(system.FlareSystemsManagerMetaData, "UptimeVoteSigned")
 	if err != nil {
@@ -390,11 +402,15 @@ func (s *systemsManagerContractClientImpl) UptimeVoteSignedListener(db epochClie
 		startEpoch := epoch.EpochIndex(time.Now())
 		eventRangeStart := epoch.StartTime(startEpoch).Unix()
 		for {
-			<-ticker.C
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
 			now := time.Now()
 			currentEpoch := epoch.EpochIndex(now)
 
-			logs, err := db.FetchLogsByAddressAndTopic0Timestamp(context.Background(), s.address, topic0, eventRangeStart, now.Unix())
+			logs, err := db.FetchLogsByAddressAndTopic0Timestamp(ctx, s.address, topic0, eventRangeStart, now.Unix())
 			if err != nil {
 				logger.Errorf("Error fetching logs %v", err)
 				continue

client/finalizer/finalizer_client.go

@@ -124,10 +124,10 @@ func (c *client) Run(ctx context.Context) error {
 }
 
 func (c *client) fetchExistingSigningPolicies(
-	_ context.Context, startTime time.Time,
+	ctx context.Context, startTime time.Time,
 ) (time.Time, error) {
 	// Read current signing policies from the database and add them to the storage
-	spList, err := c.relayClient.FetchSigningPolicies(c.db, startTime.Unix(), time.Now().Unix())
+	spList, err := c.relayClient.FetchSigningPolicies(ctx, c.db, startTime.Unix(), time.Now().Unix())
 	if err != nil {
 		return startTime, err
 	}
@@ -150,7 +150,7 @@ func (c *client) fetchExistingSigningPolicies(
 }
 
 func (c *client) runSigningPolicyInitializedListener(ctx context.Context, startTime time.Time) error {
-	spListener := c.relayClient.SigningPolicyInitializedListener(c.db, startTime)
+	spListener := c.relayClient.SigningPolicyInitializedListener(ctx, c.db, startTime)
 	for {
 		var dbPolicy signingPolicyListenerResponse
 		select {

client/finalizer/finalizer_queue.go

@@ -147,7 +147,7 @@ func (p *finalizerQueueProcessor) Run(ctx context.Context) error {
 					logger.Debugf("Finalizer will send now for voting round %v for protocol %v", item.votingRoundID, item.protocolID)
 					p.processItem(ctx, item, true)
 				}
-				p.delayedQueues.Add(st, item)
+				p.delayedQueues.Add(ctx, st, item)
 			} else {
 				logger.Errorf("Finalizer missing finalization data for protocol %v in votingRound %v", item.protocolID, item.votingRoundID)
 			}
@@ -201,12 +201,12 @@ func (p *finalizerQueueProcessor) processItem(ctx context.Context, item *queueIt
 	p.relayClient.SubmitPayloads(ctx, txInput, isDelayed, item.protocolID)
 }
 
-func (p *finalizerQueueProcessor) processDelayedQueue(items []*queueItem) error {
+func (p *finalizerQueueProcessor) processDelayedQueue(ctx context.Context, items []*queueItem) error {
 	now := time.Now()
 	currentEpoch := p.finalizerContext.votingRoundTiming.EpochIndex(now)
 	startTime := p.finalizerContext.votingRoundTiming.StartTime(currentEpoch)
 
-	relayedItems, err := p.relayClient.ProtocolMessageRelayed(context.Background(), p.db, startTime, now)
+	relayedItems, err := p.relayClient.ProtocolMessageRelayed(ctx, p.db, startTime, now)
 	if err != nil {
 		return err
 	}
@@ -216,7 +216,7 @@ func (p *finalizerQueueProcessor) processDelayedQueue(items []*queueItem) error
 			continue
 		}
 		logger.Infof("Finalizer processes delayed queue item for round %v for protocol %v", item.votingRoundID, item.protocolID)
-		p.processItem(context.TODO(), item, true)
+		p.processItem(ctx, item, true)
 	}
 	return nil
 }

client/finalizer/relay_client.go

@@ -111,8 +111,8 @@ func NewRelayContractClient(
 }
 
 // FetchSigningPolicies fetches signing policies emitted by in SigningPolicyInitialized events from Relay smart contract with timestamps in the interval (from,to].
-func (r *relayContractClient) FetchSigningPolicies(db finalizerDB, from, to int64) ([]signingPolicyListenerResponse, error) {
-	logs, err := db.FetchLogsByAddressAndTopic0(context.Background(), r.address, r.topic0SPI, from, to)
+func (r *relayContractClient) FetchSigningPolicies(ctx context.Context, db finalizerDB, from, to int64) ([]signingPolicyListenerResponse, error) {
+	logs, err := db.FetchLogsByAddressAndTopic0(ctx, r.address, r.topic0SPI, from, to)
 	if err != nil {
 		return nil, err
 	}
@@ -129,16 +129,20 @@ func (r *relayContractClient) FetchSigningPolicies(db finalizerDB, from, to int6
 	return result, nil
 }
 
-func (r *relayContractClient) SigningPolicyInitializedListener(db finalizerDB, startTime time.Time) <-chan signingPolicyListenerResponse {
+func (r *relayContractClient) SigningPolicyInitializedListener(ctx context.Context, db finalizerDB, startTime time.Time) <-chan signingPolicyListenerResponse {
 	out := make(chan signingPolicyListenerResponse, listenerBufferSize)
 	go func() {
 		ticker := time.NewTicker(shared.EventListenerInterval)
 		eventRangeStart := startTime.Unix()
 		for {
-			<-ticker.C
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
 			now := time.Now().Unix()
 
-			logs, err := db.FetchLogsByAddressAndTopic0(context.Background(), r.address, r.topic0SPI, eventRangeStart, now)
+			logs, err := db.FetchLogsByAddressAndTopic0(ctx, r.address, r.topic0SPI, eventRangeStart, now)
 			if err != nil {
 				logger.Errorf("Error fetching logs %v", err)
 				continue

utils/queue.go

@@ -1,13 +1,14 @@
 package utils
 
 import (
+	"context"
 	"sync"
 	"time"
 
 	"github.com/flare-foundation/go-flare-common/pkg/logger"
 )
 
-type QueueProcessorFunc[T any] func([]T) error
+type QueueProcessorFunc[T any] func(context.Context, []T) error
 
 type DelayedQueueManager[T any] struct {
 	timeMap map[time.Time][]T
@@ -24,7 +25,7 @@ func NewDelayedQueueManager[T any](processor QueueProcessorFunc[T]) *DelayedQueu
 	}
 }
 
-func (l *DelayedQueueManager[T]) Add(t time.Time, item T) {
+func (l *DelayedQueueManager[T]) Add(ctx context.Context, t time.Time, item T) {
 	if t.Before(time.Now()) {
 		return
 	}
@@ -33,7 +34,7 @@ func (l *DelayedQueueManager[T]) Add(t time.Time, item T) {
 	defer l.Unlock()
 
 	if _, ok := l.timeMap[t]; !ok {
-		l.createTimer(t)
+		l.createTimer(ctx, t)
 	}
 	l.timeMap[t] = append(l.timeMap[t], item)
 }
@@ -47,13 +48,17 @@ func (l *DelayedQueueManager[T]) Get(t time.Time) []T {
 	return items
 }
 
-func (l *DelayedQueueManager[T]) createTimer(t time.Time) {
+func (l *DelayedQueueManager[T]) createTimer(ctx context.Context, t time.Time) {
 	go func() {
 		timer := time.NewTimer(time.Until(t))
-		<-timer.C
-		items := l.Get(t)
-		if err := l.processor(items); err != nil {
-			logger.Errorf("DelayedQueueManager processor error: %s", err)
+		select {
+		case <-timer.C:
+			items := l.Get(t)
+			if err := l.processor(ctx, items); err != nil {
+				logger.Errorf("DelayedQueueManager processor error: %s", err)
+			}
+		case <-ctx.Done():
+			l.Get(t) // remove from map
 		}
 	}()
 }