Merge branch 'main' into custom-msg-jsonpb-marshal

Author: skudasov

pkg/capabilities/consensus/ocr3/datafeeds/feeds_aggregator.go

@@ -43,7 +43,7 @@ const (
 )
 
 type aggregatorConfig struct {
-	Feeds map[datastreams.FeedID]feedConfig
+	Feeds map[datastreams.FeedID]FeedConfig
 	// AllowedPartialStaleness is an optional optimization that tries to maximize batching.
 	// Once any deviation or heartbeat threshold hits, we will include all other feeds that are
 	// within the AllowedPartialStaleness range of their own heartbeat.
@@ -52,18 +52,30 @@ type aggregatorConfig struct {
 	AllowedPartialStalenessStr string  `mapstructure:"allowedPartialStaleness"`
 }
 
-type feedConfig struct {
-	Deviation       decimal.Decimal `mapstructure:"-"`
-	Heartbeat       int             // seconds
-	DeviationString string          `mapstructure:"deviation"`
-	RemappedIDHex   string          `mapstructure:"remappedId"`
-	RemappedID      []byte          `mapstructure:"-"`
+// FeedConfig defines the configuration for each individual feed used by the aggregator.
+// It's map representation is used directly in user-defined workflows to specify the configuration for each feed.
+type FeedConfig struct {
+	Heartbeat     int    // seconds
+	Deviation     string `mapstructure:"deviation"`
+	RemappedIDHex string `mapstructure:"remappedId"` // DO NOT CHANGE THIS. It's user facing in existing DataFeeds configurations and should be kept consistent for backward compatibility.
+	// internal fields set by [ParseConfig] after parsing the config
+	// work around mapstructure limitations to allow for decimal.Decimal and byte slices
+	parsedDeviation decimal.Decimal
+	remappedID      []byte
 }
 
-func (c feedConfig) HeartbeatNanos() int64 {
+func (c FeedConfig) HeartbeatNanos() int64 {
 	return int64(c.Heartbeat) * time.Second.Nanoseconds()
 }
 
+func (c FeedConfig) DeviationAsDecimal() decimal.Decimal {
+	return c.parsedDeviation
+}
+
+func (c FeedConfig) RemappedID() []byte {
+	return c.remappedID
+}
+
 type dataFeedsAggregator struct {
 	config      aggregatorConfig
 	reportCodec datastreams.ReportCodec
@@ -146,10 +158,10 @@ func (a *dataFeedsAggregator) Aggregate(lggr logger.Logger, previousOutcome *typ
 			"oldPrice", oldPrice,
 			"newPrice", newPrice,
 			"currDeviation", currDeviation,
-			"deviation", config.Deviation.InexactFloat64(),
+			"deviation", config.DeviationAsDecimal().InexactFloat64(),
 		)
 		if currStaleness > int64(config.Heartbeat) ||
-			currDeviation > config.Deviation.InexactFloat64() {
+			currDeviation > config.DeviationAsDecimal().InexactFloat64() {
 			previousReportInfo.ObservationTimestamp = latestReport.ObservationTimestamp
 			previousReportInfo.BenchmarkPrice = latestReport.BenchmarkPrice
 			reportsNeedingUpdate = append(reportsNeedingUpdate, latestReport)
@@ -177,7 +189,7 @@ func (a *dataFeedsAggregator) Aggregate(lggr logger.Logger, previousOutcome *typ
 	toWrap := make([]any, 0, len(reportsNeedingUpdate))
 	for _, report := range reportsNeedingUpdate {
 		feedID := datastreams.FeedID(report.FeedID).Bytes()
-		remappedID := a.config.Feeds[datastreams.FeedID(report.FeedID)].RemappedID
+		remappedID := a.config.Feeds[datastreams.FeedID(report.FeedID)].RemappedID()
 		if len(remappedID) == 0 { // fall back to original ID
 			remappedID = feedID[:]
 		}
@@ -348,15 +360,15 @@ func ParseConfig(config values.Map) (aggregatorConfig, error) {
 	}
 
 	for feedID, feedCfg := range parsedConfig.Feeds {
-		if feedCfg.DeviationString != "" {
+		if feedCfg.Deviation != "" {
 			if _, err := datastreams.NewFeedID(feedID.String()); err != nil {
 				return aggregatorConfig{}, fmt.Errorf("cannot parse feedID config for feed %s: %w", feedID, err)
 			}
-			dec, err := decimal.NewFromString(feedCfg.DeviationString)
+			dec, err := decimal.NewFromString(feedCfg.Deviation)
 			if err != nil {
 				return aggregatorConfig{}, fmt.Errorf("cannot parse deviation config for feed %s: %w", feedID, err)
 			}
-			feedCfg.Deviation = dec
+			feedCfg.parsedDeviation = dec
 			parsedConfig.Feeds[feedID] = feedCfg
 		}
 		trimmed, nonEmpty := strings.CutPrefix(feedCfg.RemappedIDHex, "0x")
@@ -365,7 +377,7 @@ func ParseConfig(config values.Map) (aggregatorConfig, error) {
 			if err != nil {
 				return aggregatorConfig{}, fmt.Errorf("cannot parse remappedId config for feed %s: %w", feedID, err)
 			}
-			feedCfg.RemappedID = rawRemappedID
+			feedCfg.remappedID = rawRemappedID
 			parsedConfig.Feeds[feedID] = feedCfg
 		}
 	}

pkg/capabilities/consensus/ocr3/datafeeds/feeds_aggregator_test.go

@@ -206,9 +206,9 @@ func TestDataFeedsAggregator_ParseConfig(t *testing.T) {
 		config := getConfig(t, feedIDA.String(), "0.1", heartbeatA)
 		parsedConfig, err := datafeeds.ParseConfig(*config)
 		require.NoError(t, err)
-		require.Equal(t, deviationA, parsedConfig.Feeds[feedIDA].Deviation)
+		require.Equal(t, deviationA, parsedConfig.Feeds[feedIDA].DeviationAsDecimal())
 		require.Equal(t, heartbeatA, parsedConfig.Feeds[feedIDA].Heartbeat)
-		require.Equal(t, deviationB, parsedConfig.Feeds[feedIDB].Deviation)
+		require.Equal(t, deviationB, parsedConfig.Feeds[feedIDB].DeviationAsDecimal())
 		require.Equal(t, heartbeatB, parsedConfig.Feeds[feedIDB].Heartbeat)
 		require.InEpsilon(t, allowedPartialStaleness, parsedConfig.AllowedPartialStaleness, math.SmallestNonzeroFloat64)
 	})

pkg/capabilities/consensus/ocr3/datafeeds/helpers_test.go

@@ -1,47 +1,7 @@
 package datafeeds
 
-import (
-	"strconv"
-	"testing"
-
-	"github.com/stretchr/testify/require"
-
-	"github.com/smartcontractkit/chainlink-common/pkg/values"
-)
-
 // exported for testing only
 var LLOStreamPrices = lloStreamPrices
 
 var DecimalToBigInt = decimalToBigInt
 var BigIntToDecimal = bigIntToDecimal
-
-type FeedConfig = feedConfig
-
-// helper function to create a map of feed configs
-func NewLLOconfig(t *testing.T, m map[uint32]FeedConfig, opts ...lloConfigOpt) values.Map {
-	unwrappedConfig := map[string]any{
-		"streams": map[string]any{},
-	}
-
-	for feedID, cfg := range m {
-		unwrappedConfig["streams"].(map[string]any)[strconv.FormatUint(uint64(feedID), 10)] = map[string]any{
-			"deviation":  cfg.Deviation.String(),
-			"heartbeat":  cfg.Heartbeat,
-			"remappedID": cfg.RemappedIDHex,
-		}
-	}
-	for _, opt := range opts {
-		opt(t, unwrappedConfig)
-	}
-	config, err := values.NewMap(unwrappedConfig)
-	require.NoError(t, err)
-	return *config
-}
-
-type lloConfigOpt = func(t *testing.T, m map[string]any)
-
-func LLOConfigAllowStaleness(staleness float64) lloConfigOpt {
-	return func(t *testing.T, m map[string]any) {
-		m["allowedPartialStaleness"] = strconv.FormatFloat(staleness, 'f', -1, 64)
-	}
-}

pkg/capabilities/consensus/ocr3/datafeeds/llo_aggregator.go

@@ -45,25 +45,48 @@ var (
 type LLOAggregatorConfig struct {
 	// workaround for the fact that mapstructure doesn't support uint32 keys
 	//streams    map[uint32]feedConfig `mapstructure:"-"`
-	StreamsStr map[string]feedConfig `mapstructure:"streams"`
+	Streams map[string]FeedConfig `mapstructure:"streams"`
 	// allowedPartialStaleness is an optional optimization that tries to maximize batching.
 	// Once any deviation or heartbeat threshold hits, we will include all other feeds that are
 	// within the allowedPartialStaleness range of their own heartbeat.
 	// For example, setting 0.2 will include all feeds that are within 20% of their heartbeat.
 	//allowedPartialStaleness float64 `mapstructure:"-"`
 	// workaround for the fact that mapstructure doesn't support float64 keys
-	AllowedPartialStalenessStr string `mapstructure:"allowedPartialStaleness"`
+	AllowedPartialStaleness string `mapstructure:"allowedPartialStaleness"`
+}
+
+// ToMap converts the LLOAggregatorConfig to a values.Map, which is suitable for the
+// [NewAggegator] function in the OCR3 Aggregator interface.
+func (c LLOAggregatorConfig) ToMap() (*values.Map, error) {
+	v, err := values.WrapMap(c)
+	if err != nil {
+		// this should never happen since we are wrapping a struct
+		return &values.Map{}, fmt.Errorf("failed to wrap LLOAggregatorConfig: %w", err)
+	}
+	return v, nil
+}
+
+func NewLLOConfig(m values.Map) (LLOAggregatorConfig, error) {
+	// Create a default LLOAggregatorConfig
+	config := LLOAggregatorConfig{
+		Streams: make(map[string]FeedConfig),
+	}
+	if err := m.UnwrapTo(&config); err != nil {
+		return LLOAggregatorConfig{}, fmt.Errorf("failed to unwrap values.Map to LLOAggregatorConfig: %w", err)
+	}
+
+	return config, nil
 }
 
 func (c LLOAggregatorConfig) convertToInternal() (parsedLLOAggregatorConfig, error) {
 	parsedConfig := parsedLLOAggregatorConfig{
-		streams: make(map[uint32]feedConfig),
+		streams: make(map[uint32]FeedConfig),
 	}
 	cfgErr := func(err error) error {
 		cfgErr := fmt.Errorf("llo aggregator config: %w", ErrInvalidConfig)
 		return errors.Join(cfgErr, err)
 	}
-	for s, cfg := range c.StreamsStr {
+	for s, cfg := range c.Streams {
 		id, err := strconv.ParseUint(s, 10, 32)
 		if err != nil {
 			// this should never happen since we are using a mapstructure-compatible config
@@ -77,25 +100,25 @@ func (c LLOAggregatorConfig) convertToInternal() (parsedLLOAggregatorConfig, err
 		if cfg.RemappedIDHex == "" {
 			return parsedConfig, cfgErr(fmt.Errorf("remappedID is required for stream %d", streamID))
 		}
-		if cfg.DeviationString != "" {
-			dec, err := decimal.NewFromString(cfg.DeviationString)
+		if cfg.Deviation != "" {
+			dec, err := decimal.NewFromString(cfg.Deviation)
 			if err != nil {
 				return parsedConfig, cfgErr(fmt.Errorf("cannot parse deviation config for feed %d: %w", streamID, err))
 			}
-			cfg.Deviation = dec
+			cfg.parsedDeviation = dec
 			parsedConfig.streams[streamID] = cfg
 		}
 		trimmed := strings.TrimPrefix(cfg.RemappedIDHex, "0x")
 		rawRemappedID, err := hex.DecodeString(trimmed)
 		if err != nil {
 			return parsedConfig, cfgErr(fmt.Errorf("cannot parse remappedId config for feed %d: %w", streamID, err))
 		}
-		cfg.RemappedID = rawRemappedID
+		cfg.remappedID = rawRemappedID
 		parsedConfig.streams[streamID] = cfg
 	}
 	// convert allowedPartialStaleness from string to float64
-	if c.AllowedPartialStalenessStr != "" {
-		allowedPartialStaleness, err := decimal.NewFromString(c.AllowedPartialStalenessStr)
+	if c.AllowedPartialStaleness != "" {
+		allowedPartialStaleness, err := decimal.NewFromString(c.AllowedPartialStaleness)
 		if err != nil {
 			return parsedConfig, cfgErr(fmt.Errorf("cannot parse allowedPartialStaleness: %w", err))
 		}
@@ -108,7 +131,7 @@ func (c LLOAggregatorConfig) convertToInternal() (parsedLLOAggregatorConfig, err
 // the separation is because mapstructure only supports string keys.
 // the are exposed in LLOAggregatorConfig for the config which is then processed into this internal representation.
 type parsedLLOAggregatorConfig struct {
-	streams                 map[uint32]feedConfig
+	streams                 map[uint32]FeedConfig
 	allowedPartialStaleness float64
 }
 
@@ -118,6 +141,8 @@ type LLOAggregator struct {
 	config parsedLLOAggregatorConfig
 }
 
+// NewLLOAggregator creates a new LLOAggregator instance based on the provided configuration.
+// The config should be a [values.Map] that has represents from the [LLOAggregatorConfig]. See [LLOAggreagatorConfig.ToMap]
 func NewLLOAggregator(config values.Map) (types.Aggregator, error) {
 	parsedConfig, err := parseLLOConfig(config)
 	if err != nil {
@@ -180,10 +205,10 @@ func (a *LLOAggregator) Aggregate(lggr logger.Logger, previousOutcome *types.Agg
 			"oldPrice", oldPrice,
 			"newPrice", newPrice,
 			"currDeviation", priceDeviation,
-			"deviation", config.Deviation.InexactFloat64(),
+			"deviation", config.DeviationAsDecimal().InexactFloat64(),
 		)
 		if timeDiffNs > config.HeartbeatNanos() ||
-			priceDeviation > config.Deviation.InexactFloat64() {
+			priceDeviation > config.DeviationAsDecimal().InexactFloat64() {
 			// this stream needs an update
 			previousStreamInfo.Timestamp = observationTimestamp.UnixNano()
 			var err2 error
@@ -214,7 +239,7 @@ func (a *LLOAggregator) Aggregate(lggr logger.Logger, previousOutcome *types.Agg
 
 	toWrap := make([]*EVMEncodableStreamUpdate, 0, len(mustUpdateIDs))
 	for _, streamID := range mustUpdateIDs {
-		remappedID := a.config.streams[streamID].RemappedID
+		remappedID := a.config.streams[streamID].RemappedID()
 		newPrice := prices[streamID]
 		w := &EVMEncodableStreamUpdate{
 			StreamID:   streamID,
@@ -413,7 +438,7 @@ func lloStreamPrices(lggr logger.Logger, wantStreamIDs []uint32, lloEvents map[o
 // parseLLOConfig parses the user-facing, type-less, LLO aggregator in the internal typed config.
 func parseLLOConfig(config values.Map) (parsedLLOAggregatorConfig, error) {
 	converter := LLOAggregatorConfig{
-		StreamsStr: make(map[string]feedConfig),
+		Streams: make(map[string]FeedConfig),
 	}
 	if err := config.UnwrapTo(&converter); err != nil {
 		return parsedLLOAggregatorConfig{}, err