fix(x/oracle): update latest ids when source is removed from list

Signed-off-by: Artur Troian <troian@users.noreply.github.com>

Author: troian

x/oracle/keeper/abci.go

@@ -42,6 +42,16 @@ func (k *keeper) EndBlocker(ctx context.Context) error {
 	twapDuration := params.TwapWindow
 	twapStart := now.Add(-twapDuration)
 
+	// Build a set of currently-authorized source IDs from params.Sources.
+	// Only these sources should participate in aggregation; latestPriceID
+	// entries for removed sources are ignored.
+	activeSourceIDs := make(map[uint32]bool, len(params.Sources))
+	for _, source := range params.Sources {
+		if id, err := k.sourceID.Get(sctx, source); err == nil {
+			activeSourceIDs[id] = true
+		}
+	}
+
 	// Phase 1: walk latestPriceID to discover sources per denom and their latest timestamps.
 	// latestByDenom maps DataID → list of (source, latestTimestamp) pairs.
 	type sourceInfo struct {
@@ -52,6 +62,11 @@ func (k *keeper) EndBlocker(ctx context.Context) error {
 	latestByDenom := make(map[types.DataID][]sourceInfo)
 
 	err = k.latestPriceID.Walk(sctx, nil, func(key types.PriceDataID, state types.PriceLatestDataState) (bool, error) {
+		// Skip sources that are no longer in params.Sources.
+		if !activeSourceIDs[key.Source] {
+			return false, nil
+		}
+
 		did := types.DataID{
 			Denom:     key.Denom,
 			BaseDenom: key.BaseDenom,

x/oracle/keeper/keeper.go

@@ -389,6 +389,13 @@ func (k *keeper) SetParams(ctx sdk.Context, p types.Params) error {
 		return err
 	}
 
+	// Determine which sources are being removed so we can clean up their
+	// latestPriceID entries. Without this cleanup the EndBlocker would
+	// continue to discover (and skip) stale entries every block, and — more
+	// critically — the orphaned latestPriceID state prevents the aggregator
+	// from recovering after a remove-then-re-add cycle.
+	oldParams, oldErr := k.Params.Get(ctx)
+
 	if err := k.Params.Set(ctx, p); err != nil {
 		return err
 	}
@@ -423,6 +430,34 @@ func (k *keeper) SetParams(ctx sdk.Context, p types.Params) error {
 			}
 		}
 	}
+
+	// Clean up latestPriceID entries for sources that were removed.
+	// This prevents orphaned state from polluting the EndBlocker walk
+	// and ensures a re-added source starts fresh.
+	if oldErr == nil {
+		newSourceSet := make(map[string]struct{}, len(p.Sources))
+		for _, s := range p.Sources {
+			newSourceSet[s] = struct{}{}
+		}
+
+		for _, s := range oldParams.Sources {
+			if _, ok := newSourceSet[s]; ok {
+				continue
+			}
+
+			// Source was removed — resolve its ID and delete latestPriceID entries.
+			sID, err := k.sourceID.Get(ctx, s)
+			if err != nil {
+				// No sourceID mapping means no latestPriceID to clean up.
+				continue
+			}
+
+			if err := k.removeSourceLatestPriceIDs(ctx, sID); err != nil {
+				return err
+			}
+		}
+	}
+
 	// call hooks
 	for _, hook := range k.hooks.onSetParams {
 		hook(ctx, p)
@@ -431,6 +466,31 @@ func (k *keeper) SetParams(ctx sdk.Context, p types.Params) error {
 	return nil
 }
 
+// removeSourceLatestPriceIDs deletes all latestPriceID entries for the given
+// source ID. This is called when a source is removed from params.Sources so
+// that the EndBlocker no longer discovers stale entries
+func (k *keeper) removeSourceLatestPriceIDs(ctx sdk.Context, sourceID uint32) error {
+	var toDelete []types.PriceDataID
+
+	err := k.latestPriceID.Walk(ctx, nil, func(key types.PriceDataID, _ types.PriceLatestDataState) (bool, error) {
+		if key.Source == sourceID {
+			toDelete = append(toDelete, key)
+		}
+		return false, nil
+	})
+	if err != nil {
+		return err
+	}
+
+	for _, key := range toDelete {
+		if err := k.latestPriceID.Remove(ctx, key); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 // GetParams returns the current x/oracle module parameters.
 func (k *keeper) GetParams(ctx sdk.Context) (types.Params, error) {
 	return k.Params.Get(ctx)