From d816ce5cee9fdd365c4740a366be1647c555d89b Mon Sep 17 00:00:00 2001 From: srt0422 Date: Sat, 28 Mar 2026 05:36:36 -0700 Subject: [PATCH] Add on-chain reward pipeline observability Add EventRewardRedirectedToEcosystem proto message and emitter for visibility when topic rewards are redirected to the ecosystem bucket. Escalate redirect log from Info to Warn. Track redirected_topic_ids in EventTopicRewardsSet. Supports monitoring the reward pipeline to detect outages like the 30-hour mainnet incident. Co-Authored-By: Claude Opus 4.6 --- x/emissions/metrics/labels.go | 1 + x/emissions/module/rewards/rewards.go | 21 +++++++++++++++------ x/emissions/proto/emissions/v9/events.proto | 15 +++++++++++++++ x/emissions/types/events_emitters.go | 13 +++++++++++-- x/emissions/types/events_test.go | 2 +- x/emissions/types/events_utils.go | 16 +++++++++++++--- 6 files changed, 56 insertions(+), 12 deletions(-) diff --git a/x/emissions/metrics/labels.go b/x/emissions/metrics/labels.go index de93f3bc2..22edb3c5b 100644 --- a/x/emissions/metrics/labels.go +++ b/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" diff --git a/x/emissions/module/rewards/rewards.go b/x/emissions/module/rewards/rewards.go index 051615cda..d90f855a5 100644 --- a/x/emissions/module/rewards/rewards.go +++ b/x/emissions/module/rewards/rewards.go @@ -125,18 +125,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.GetTopicRewardNonce(args.Ctx, topicId) if err != nil || topicRewardNonce == 0 { // return reward to ecosystem account - err = args.K.MoveCoinsFromAlloraRewardsToEcosystem(args.Ctx, *topicRewards[topicId]) + rewardAmount := *topicRewards[topicId] + err = args.K.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 @@ -157,14 +163,17 @@ func EmitRewards(args EmitRewardsArgs) error { }) if err != nil { // return reward to ecosystem account - errMC := args.K.MoveCoinsFromAlloraRewardsToEcosystem(args.Ctx, *topicRewards[topicId]) + rewardAmount := *topicRewards[topicId] + errMC := args.K.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 } @@ -189,8 +198,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 } diff --git a/x/emissions/proto/emissions/v9/events.proto b/x/emissions/proto/emissions/v9/events.proto index d63226439..b3b8aa02e 100644 --- a/x/emissions/proto/emissions/v9/events.proto +++ b/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 { diff --git a/x/emissions/types/events_emitters.go b/x/emissions/types/events_emitters.go index e9e7b3c65..218559295 100644 --- a/x/emissions/types/events_emitters.go +++ b/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) { diff --git a/x/emissions/types/events_test.go b/x/emissions/types/events_test.go index bc853dfdc..9afdb20b0 100644 --- a/x/emissions/types/events_test.go +++ b/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) diff --git a/x/emissions/types/events_utils.go b/x/emissions/types/events_utils.go index 3cca16bfd..a59dc55a9 100644 --- a/x/emissions/types/events_utils.go +++ b/x/emissions/types/events_utils.go @@ -442,15 +442,25 @@ func NewRewardsSetEventBase(actorType ActorType, blockHeight, blockHeightTx Bloc } } -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, } }