feat(x/oracle): add timestamp filtering for stale data

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

Author: troian

go.mod

@@ -47,7 +47,7 @@ require (
 	google.golang.org/grpc v1.76.0
 	gopkg.in/yaml.v3 v3.0.1
 	gotest.tools/v3 v3.5.2
-	pkg.akt.dev/go v0.2.0-b6
+	pkg.akt.dev/go v0.2.0-b7
 	pkg.akt.dev/go/cli v0.2.0-b5
 	pkg.akt.dev/go/sdl v0.2.0-b0
 )

go.sum

@@ -3290,8 +3290,8 @@ nhooyr.io/websocket v1.8.17 h1:KEVeLJkUywCKVsnLIDlD/5gtayKp8VoCkksHCGGfT9Y=
 nhooyr.io/websocket v1.8.17/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c=
 pgregory.net/rapid v0.5.5 h1:jkgx1TjbQPD/feRoK+S/mXw9e1uj6WilpHrXJowi6oA=
 pgregory.net/rapid v0.5.5/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04=
-pkg.akt.dev/go v0.2.0-b6 h1:vWfPdKmMduzqEM9udDB+9jtdlHLlbuV6bzbmVyHuTZo=
-pkg.akt.dev/go v0.2.0-b6/go.mod h1:8cUJoOmylVh0+UGtr0uY5bOvfIZiir1X6iDelXqxv9M=
+pkg.akt.dev/go v0.2.0-b7 h1:QM1vWXUIikPHoUH4a4NWEZ1lB+ZDZOokV9/bo3r+Lu4=
+pkg.akt.dev/go v0.2.0-b7/go.mod h1:8cUJoOmylVh0+UGtr0uY5bOvfIZiir1X6iDelXqxv9M=
 pkg.akt.dev/go/cli v0.2.0-b5 h1:Ds+Y04gprqjzkQEp4SBc9hWUp5iynxEtNnCTHC2BeAA=
 pkg.akt.dev/go/cli v0.2.0-b5/go.mod h1:L+jzPE95i0YQvuEgSDi5WlVeaSe5FpocofH+k4akNio=
 pkg.akt.dev/go/sdl v0.2.0-b0 h1:IOx8FwA57r08fiuTmWNmuTVpQhzb7vGPuCPhMtxi0fo=

testutil/oracle/price_feeder.go

@@ -7,7 +7,7 @@ import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 
 	oraclev1 "pkg.akt.dev/go/node/oracle/v1"
-	sdkutil "pkg.akt.dev/go/sdkutil"
+	"pkg.akt.dev/go/sdkutil"
 
 	oraclekeeper "pkg.akt.dev/node/v2/x/oracle/keeper"
 )
@@ -99,12 +99,13 @@ func (pf *PriceFeeder) FeedPrice(ctx sdk.Context, denom string) error {
 	}
 
 	priceHealth := oraclev1.PriceHealth{
-		Denom:           denom,
-		IsHealthy:       true,
-		HasMinSources:   true,
-		AllSourcesFresh: true,
-		DeviationOk:     true,
-		FailureReason:   []string{},
+		Denom:               denom,
+		IsHealthy:           true,
+		HasMinSources:       true,
+		TotalSources:        1,
+		TotalHealthySources: 1,
+		DeviationOk:         true,
+		FailureReason:       []string{},
 	}
 
 	if err := pf.keeper.SetAggregatedPrice(ctx, dataID, aggregatedPrice); err != nil {

x/oracle/keeper/keeper.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"slices"
 	"sort"
 	"time"
 
@@ -322,10 +323,16 @@ func (k *keeper) EndBlocker(ctx context.Context) error {
 		panic(fmt.Sprintf("failed to walk latest prices: %v", err))
 	}
 
+	cutoffHeight := sctx.BlockHeight() - params.MaxPriceStalenessBlocks
+
 	for id, rid := range rIDs {
 		latestData := make([]types.PriceData, 0, len(rid))
 
 		for _, id := range rid {
+			if id.Height < cutoffHeight {
+				continue
+			}
+
 			state, _ := k.prices.Get(sctx, id)
 
 			latestData = append(latestData, types.PriceData{
@@ -496,18 +503,28 @@ func (k *keeper) calculateAggregatedPrices(ctx sdk.Context, id types.DataID, lat
 		Denom: id.Denom,
 	}
 
+	params, err := k.GetParams(ctx)
+	if err != nil {
+		return aggregated, err
+	}
+
+	// filter out stale sources by time
+	// todo block time is a variable, it should not be hardcoded
+	cutoffTimestamp := ctx.BlockTime().Add(-time.Duration(params.MaxPriceStalenessBlocks) * (time.Second * 6))
+
+	for i := len(latestData) - 1; i >= 0; i-- {
+		if latestData[i].State.Timestamp.Before(cutoffTimestamp) {
+			latestData = slices.Delete(latestData, i, i+1)
+		}
+	}
+
 	if len(latestData) == 0 {
 		return aggregated, errorsmod.Wrap(
 			types.ErrPriceStalled,
 			"all price sources are stale",
 		)
 	}
 
-	params, err := k.GetParams(ctx)
-	if err != nil {
-		return aggregated, err
-	}
-
 	// Calculate TWAP for each source
 	var twaps []sdkmath.LegacyDec //nolint:prealloc
 	for _, source := range latestData {
@@ -624,7 +641,9 @@ func (k *keeper) getAggregatedPrice(ctx sdk.Context, denom string) (types.Aggreg
 // CheckPriceHealth checks if the aggregated price meets health requirements
 func (k *keeper) setPriceHealth(ctx sdk.Context, params types.Params, dataIDs []types.PriceDataRecordID, aggregatedPrice types.AggregatedPrice) types.PriceHealth {
 	health := types.PriceHealth{
-		Denom: aggregatedPrice.Denom,
+		Denom:               aggregatedPrice.Denom,
+		TotalSources:        uint32(len(dataIDs)),
+		TotalHealthySources: aggregatedPrice.NumSources,
 	}
 
 	// Check 1: Minimum number of sources
@@ -647,41 +666,7 @@ func (k *keeper) setPriceHealth(ctx sdk.Context, params types.Params, dataIDs []
 		))
 	}
 
-	// Check 3: All sources are fresh
-	allFresh := true
-	foundSource := false
-	cutoffHeight := ctx.BlockHeight() - params.MaxPriceStalenessBlocks
-
-	for _, did := range dataIDs {
-		foundSource = true
-		allFresh = allFresh && did.Height >= cutoffHeight
-		//return !allFresh, nil
-	}
-
-	//err := k.latestPrices.Walk(ctx, nil, func(key types.PriceDataID, value int64) (bool, error) {
-	//	if key.Denom != aggregatedPrice.Denom || key.BaseDenom != sdkutil.DenomUSD {
-	//		return false, nil
-	//	}
-	//
-	//	foundSource = true
-	//	allFresh = allFresh && value >= cutoffHeight
-	//	return !allFresh, nil
-	//})
-
-	//if err != nil {
-	//	allFresh = false
-	//}
-
-	if !foundSource {
-		allFresh = false
-	}
-
-	if !allFresh {
-		health.FailureReason = append(health.FailureReason, "one or more price sources are stale")
-	}
-
-	health.AllSourcesFresh = allFresh
-	health.IsHealthy = health.HasMinSources && health.DeviationOk && health.AllSourcesFresh
+	health.IsHealthy = health.HasMinSources && health.DeviationOk
 
 	err := k.pricesHealth.Set(ctx, types.DataID{Denom: health.Denom, BaseDenom: sdkutil.DenomUSD}, health)
 	// if there is an error when storing price health, something went horribly wrong