Add on-chain reward pipeline observability

x/emissions/metrics/labels.go

@@ -67,6 +67,7 @@ const (
 	REPUTER_LAST_COMMIT_EVENT                               = "reputer_last_commit_event"
 	FORECAST_TASK_SCORE_EVENT                               = "forecast_task_score_event"
 	TOPIC_REWARD_EVENT                                      = "topic_reward_event"
+	REWARD_REDIRECTED_TO_ECOSYSTEM_EVENT                    = "reward_redirected_to_ecosystem_event"
 	WORKER_EMA_SCORE_EVENT                                  = "worker_ema_score_event"
 	LISTENING_COEFFICIENTS_EVENT                            = "listening_coefficients_event"
 	INFERER_NETWORK_REGRET_EVENT                            = "inferer_network_regret_event"

x/emissions/module/rewards/rewards.go

@@ -124,18 +124,24 @@ func EmitRewards(args EmitRewardsArgs) error {
 	// Process rewards for each topic, pruning at the end of epoch
 	totalMonthlyReputerRewards := cosmosMath.ZeroInt()
 	totalMonthlyTopicRewards := cosmosMath.ZeroInt()
+	// Track topic IDs whose rewards were redirected to ecosystem
+	redirectedTopicIds := make([]uint64, 0)
+
 	for _, topicId := range sortedRewardableTopics {
 		topicRewardNonce, err := args.K.GetTopicKeeper().GetTopicRewardNonce(args.Ctx, topicId)
 		if err != nil || topicRewardNonce == 0 {
 			// return reward to ecosystem account
-			err = args.K.GetBankingKeeper().MoveCoinsFromAlloraRewardsToEcosystem(args.Ctx, *topicRewards[topicId])
+			rewardAmount := *topicRewards[topicId]
+			err = args.K.GetBankingKeeper().MoveCoinsFromAlloraRewardsToEcosystem(args.Ctx, rewardAmount)
 			if err != nil {
 				Logger(args.Ctx).Error("Failed to move coins from allora rewards to ecosystem", "topicId", topicId, "error", err)
 				panic(err)
 			}
 			*topicRewards[topicId] = alloraMath.ZeroDec()
 
-			Logger(args.Ctx).Info("Topic has no valid reward nonce, skipping", "topicId", topicId)
+			Logger(args.Ctx).Warn("Topic has no valid reward nonce, redirecting reward to ecosystem", "topicId", topicId, "rewardAmount", rewardAmount.String())
+			types.EmitNewRewardRedirectedToEcosystemEvent(args.Ctx, topicId, rewardAmount, "no_valid_reward_nonce", args.BlockHeight)
+			redirectedTopicIds = append(redirectedTopicIds, topicId)
 			continue
 		}
 		// Defer pruning records after rewards payout
@@ -156,14 +162,17 @@ func EmitRewards(args EmitRewardsArgs) error {
 		})
 		if err != nil {
 			// return reward to ecosystem account
-			errMC := args.K.GetBankingKeeper().MoveCoinsFromAlloraRewardsToEcosystem(args.Ctx, *topicRewards[topicId])
+			rewardAmount := *topicRewards[topicId]
+			errMC := args.K.GetBankingKeeper().MoveCoinsFromAlloraRewardsToEcosystem(args.Ctx, rewardAmount)
 			if errMC != nil {
 				Logger(args.Ctx).Error("Failed to move coins from allora rewards to ecosystem", "topicId", topicId, "error", errMC)
 				panic(errMC)
 			}
 			*topicRewards[topicId] = alloraMath.ZeroDec()
 
-			Logger(args.Ctx).Error("Failed to process rewards", "topicId", topicId, "error", err)
+			Logger(args.Ctx).Error("Failed to process rewards, redirecting to ecosystem", "topicId", topicId, "rewardAmount", rewardAmount.String(), "error", err)
+			types.EmitNewRewardRedirectedToEcosystemEvent(args.Ctx, topicId, rewardAmount, "reward_distribution_failed", args.BlockHeight)
+			redirectedTopicIds = append(redirectedTopicIds, topicId)
 			continue
 		}
 
@@ -188,8 +197,8 @@ func EmitRewards(args EmitRewardsArgs) error {
 		return errors.Wrapf(err, "failed to add monthly rewards")
 	}
 
-	// Emit reward of each topic
-	types.EmitNewTopicRewardSetEvent(args.Ctx, topicRewards)
+	// Emit reward of each topic, including redirected topic IDs
+	types.EmitNewTopicRewardSetEvent(args.Ctx, topicRewards, redirectedTopicIds)
 	return nil
 }
 

x/emissions/proto/emissions/v9/events.proto

@@ -382,6 +382,21 @@ message EventTopicRewardsSet {
     (gogoproto.customtype) = "github.com/allora-network/allora-chain/math.Dec",
     (gogoproto.nullable) = false
   ];
+  // Topic IDs whose rewards were redirected to the ecosystem account this epoch.
+  repeated uint64 redirected_topic_ids = 3;
+}
+
+// Emitted when a topic's rewards are redirected to the ecosystem module account
+// instead of being distributed to participants.
+message EventRewardRedirectedToEcosystem {
+  uint64 topic_id = 1;
+  string reward_amount = 2 [
+    (gogoproto.customtype) = "github.com/allora-network/allora-chain/math.Dec",
+    (gogoproto.nullable) = false
+  ];
+  // "no_valid_reward_nonce" or "reward_distribution_failed"
+  string reason = 3;
+  int64 block_height = 4;
 }
 
 message EventEMAScoresSet {

x/emissions/types/events_emitters.go

@@ -455,15 +455,24 @@ func EmitNewReputerAndDelegatorRewardsSettledEvent(ctx context.Context, blockHei
 	}
 }
 
-func EmitNewTopicRewardSetEvent(ctx context.Context, topicRewards map[uint64]*alloraMath.Dec) {
+func EmitNewTopicRewardSetEvent(ctx context.Context, topicRewards map[uint64]*alloraMath.Dec, redirectedTopicIds []uint64) {
 	metrics.IncrProducerEventCount(metrics.TOPIC_REWARD_EVENT)
 	sdkCtx := sdk.UnwrapSDKContext(ctx)
-	err := sdkCtx.EventManager().EmitTypedEvent(NewTopicRewardSetEventBase(topicRewards))
+	err := sdkCtx.EventManager().EmitTypedEvent(NewTopicRewardSetEventBase(topicRewards, redirectedTopicIds))
 	if err != nil {
 		sdkCtx.Logger().Warn("Error emitting NewTopicRewardSetEvent", "error", err)
 	}
 }
 
+func EmitNewRewardRedirectedToEcosystemEvent(ctx context.Context, topicId TopicId, rewardAmount alloraMath.Dec, reason string, blockHeight BlockHeight) {
+	metrics.IncrProducerEventCount(metrics.REWARD_REDIRECTED_TO_ECOSYSTEM_EVENT)
+	sdkCtx := sdk.UnwrapSDKContext(ctx)
+	err := sdkCtx.EventManager().EmitTypedEvent(NewRewardRedirectedToEcosystemEventBase(topicId, rewardAmount, reason, blockHeight))
+	if err != nil {
+		sdkCtx.Logger().Warn("Error emitting RewardRedirectedToEcosystemEvent", "error", err)
+	}
+}
+
 // Delegate rewards share updated event
 
 func EmitNewDelegateRewardShareUpdatedEvent(ctx context.Context, topicId TopicId, reputer string, rewardPerShare alloraMath.Dec) {

x/emissions/types/events_test.go

@@ -1032,7 +1032,7 @@ func TestEmitNewTopicRewardsSetEvent(t *testing.T) {
 		topicRewards[id] = &reward
 	}
 
-	types.EmitNewTopicRewardSetEvent(ctx, topicRewards)
+	types.EmitNewTopicRewardSetEvent(ctx, topicRewards, nil)
 	events := ctx.EventManager().Events()
 	require.Len(t, events, 1)
 

x/emissions/types/events_utils.go

@@ -442,15 +442,25 @@
 	}
 }
 
-func NewTopicRewardSetEventBase(topicRewards map[uint64]*alloraMath.Dec) proto.Message {
+func NewTopicRewardSetEventBase(topicRewards map[uint64]*alloraMath.Dec, redirectedTopicIds []uint64) proto.Message {
 	ids := alloraMath.GetSortedKeys(topicRewards)
 	rewardValues := make([]alloraMath.Dec, 0)
 	for _, id := range ids {
 		rewardValues = append(rewardValues, *topicRewards[id])
 	}
 	return &EventTopicRewardsSet{
-		TopicIds: ids,
-		Rewards:  rewardValues,
+		TopicIds:           ids,
+		Rewards:            rewardValues,
+		RedirectedTopicIds: redirectedTopicIds,
+	}
+}
+
+func NewRewardRedirectedToEcosystemEventBase(topicId TopicId, rewardAmount alloraMath.Dec, reason string, blockHeight BlockHeight) proto.Message {
+	return &EventRewardRedirectedToEcosystem{
+		TopicId:      topicId,
+		RewardAmount: rewardAmount,
+		Reason:       reason,
+		BlockHeight:  blockHeight,
 	}
 }