[CM-380] Identical Aggregator

pkg/capabilities/consensus/ocr3/aggregators/identical.go

@@ -0,0 +1,125 @@
+package aggregators
+
+import (
+	"crypto/sha256"
+	"fmt"
+
+	"google.golang.org/protobuf/proto"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/consensus/ocr3/types"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/values"
+
+	ocrcommon "github.com/smartcontractkit/libocr/commontypes"
+)
+
+type identicalAggregator struct {
+	config aggregatorConfig
+	lggr   logger.Logger
+}
+
+type aggregatorConfig struct {
+	// Length of the list of observations that each node is expected to provide.
+	// Aggregator's output (i.e. EncodableOutcome) will be a values.Map with the same
+	// number of elements and keyed by indices 0,1,2,... (unless KeyOverrides are provided).
+	// Defaults to 1.
+	ExpectedObservationsLen int
+	// If non-empty, the keys in the outcome map will be replaced with these values.
+	// If non-empty, must be of length ExpectedObservationsLen.
+	KeyOverrides []string
+}
+
+type counter struct {
+	fullObservation values.Value
+	count           int
+}
+
+var _ types.Aggregator = (*identicalAggregator)(nil)
+
+func (a *identicalAggregator) Aggregate(_ *types.AggregationOutcome, observations map[ocrcommon.OracleID][]values.Value, f int) (*types.AggregationOutcome, error) {
+	counters := []map[[32]byte]*counter{}
+	for i := 0; i < a.config.ExpectedObservationsLen; i++ {
+		counters = append(counters, map[[32]byte]*counter{})
+	}
+	for nodeID, nodeObservations := range observations {
+		if len(nodeObservations) == 0 || nodeObservations[0] == nil {
+			a.lggr.Warnf("node %d contributed with empty observations", nodeID)
+			continue
+		}
+		if len(nodeObservations) != a.config.ExpectedObservationsLen {
+			a.lggr.Warnf("node %d contributed with an incorrect number of observations %d - ignoring them", nodeID, len(nodeObservations))
+			continue
+		}
+		for idx, observation := range nodeObservations {
+			marshalled, err := proto.MarshalOptions{Deterministic: true}.Marshal(values.Proto(observation))
+			if err != nil {
+				return nil, err
+			}
+			sha := sha256.Sum256(marshalled)
+			elem, ok := counters[idx][sha]
+			if !ok {
+				counters[idx][sha] = &counter{
+					fullObservation: observation,
+					count:           1,
+				}
+			} else {
+				elem.count++
+			}
+		}
+	}
+	return a.collectHighestCounts(counters, f)
+}
+
+func (a *identicalAggregator) collectHighestCounts(counters []map[[32]byte]*counter, f int) (*types.AggregationOutcome, error) {
+	useOverrides := len(a.config.KeyOverrides) == len(counters)
+	outcome := make(map[string]any)
+	for idx, shaToCounter := range counters {
+		highestCount := 0
+		var highestObservation values.Value
+		for _, counter := range shaToCounter {
+			if counter.count > highestCount {
+				highestCount = counter.count
+				highestObservation = counter.fullObservation
+			}
+		}
+		if highestCount < 2*f+1 {
+			return nil, fmt.Errorf("can't reach consensus on observations with index %d", idx)
+		}
+		if useOverrides {
+			outcome[a.config.KeyOverrides[idx]] = highestObservation
+		} else {
+			outcome[fmt.Sprintf("%d", idx)] = highestObservation
+		}
+	}
+	valMap, err := values.NewMap(outcome)
+	if err != nil {
+		return nil, err
+	}
+	return &types.AggregationOutcome{
+		EncodableOutcome: values.ProtoMap(valMap),
+		Metadata:         nil,
+		ShouldReport:     true,
+	}, nil
+}
+
+func NewIdenticalAggregator(config values.Map, lggr logger.Logger) (*identicalAggregator, error) {
+	parsedConfig, err := ParseConfig(config)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse config (%+v): %w", config, err)
+	}
+	return &identicalAggregator{
+		config: parsedConfig,
+		lggr:   lggr,
+	}, nil
+}
+
+func ParseConfig(config values.Map) (aggregatorConfig, error) {
+	parsedConfig := aggregatorConfig{}
+	if err := config.UnwrapTo(&parsedConfig); err != nil {
+		return aggregatorConfig{}, err
+	}
+	if parsedConfig.ExpectedObservationsLen == 0 {
+		parsedConfig.ExpectedObservationsLen = 1
+	}
+	return parsedConfig, nil
+}

pkg/capabilities/consensus/ocr3/aggregators/identical_test.go

@@ -0,0 +1,93 @@
+package aggregators_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/libocr/commontypes"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/consensus/ocr3/aggregators"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/values"
+)
+
+func TestDataFeedsAggregator_Aggregate(t *testing.T) {
+	config := getConfig(t, nil)
+	agg, err := aggregators.NewIdenticalAggregator(*config, logger.Nop())
+	require.NoError(t, err)
+
+	observations := map[commontypes.OracleID][]values.Value{
+		0: {values.NewString("a")},
+		1: {values.NewString("a")},
+		2: {values.NewString("a")},
+		3: {values.NewString("a")},
+	}
+	outcome, err := agg.Aggregate(nil, observations, 1)
+	require.NoError(t, err)
+	require.True(t, outcome.ShouldReport)
+	require.Equal(t, "", outcome.EncoderName)
+	require.Nil(t, outcome.EncoderConfig)
+
+	m, err := values.FromMapValueProto(outcome.EncodableOutcome)
+	require.NoError(t, err)
+
+	require.Len(t, m.Underlying, 1)
+	require.Equal(t, m.Underlying["0"], values.NewString("a"))
+}
+
+func TestDataFeedsAggregator_Aggregate_OverrideWithKeys(t *testing.T) {
+	config := getConfig(t, []string{"outcome"})
+	agg, err := aggregators.NewIdenticalAggregator(*config, logger.Nop())
+	require.NoError(t, err)
+
+	observations := map[commontypes.OracleID][]values.Value{
+		0: {values.NewString("a")},
+		1: {values.NewString("a")},
+		2: {values.NewString("a")},
+		3: {values.NewString("a")},
+	}
+	outcome, err := agg.Aggregate(nil, observations, 1)
+	require.NoError(t, err)
+	require.True(t, outcome.ShouldReport)
+	require.Equal(t, "", outcome.EncoderName)
+	require.Nil(t, outcome.EncoderConfig)
+
+	m, err := values.FromMapValueProto(outcome.EncodableOutcome)
+	require.NoError(t, err)
+
+	require.Len(t, m.Underlying, 1)
+	require.Equal(t, m.Underlying["outcome"], values.NewString("a"))
+}
+
+func TestDataFeedsAggregator_Aggregate_NoConsensus(t *testing.T) {
+	config := getConfig(t, []string{"outcome"})
+	agg, err := aggregators.NewIdenticalAggregator(*config, logger.Nop())
+	require.NoError(t, err)
+
+	encoderStr := "evm"
+	encoderName := values.NewString(encoderStr)
+	encoderCfg, err := values.NewMap(map[string]any{"foo": "bar"})
+	require.NoError(t, err)
+
+	observations := map[commontypes.OracleID][]values.Value{
+		0: {values.NewString("a"), encoderName, encoderCfg},
+		1: {values.NewString("b"), encoderName, encoderCfg},
+		2: {values.NewString("b"), encoderName, encoderCfg},
+		3: {values.NewString("a"), encoderName, encoderCfg},
+	}
+	outcome, err := agg.Aggregate(nil, observations, 1)
+	require.Nil(t, outcome)
+	require.ErrorContains(t, err, "can't reach consensus on observations with index 0")
+}
+
+func getConfig(t *testing.T, overrideKeys []string) *values.Map {
+	unwrappedConfig := map[string]any{
+		"expectedObservationsLen": len(overrideKeys),
+		"keyOverrides":            overrideKeys,
+	}
+
+	config, err := values.NewMap(unwrappedConfig)
+	require.NoError(t, err)
+	return config
+}

pkg/capabilities/consensus/ocr3/capability.go

@@ -282,6 +282,8 @@ func (o *capability) queueRequestForProcessing(
 		WorkflowDonID:            metadata.WorkflowDonID,
 		WorkflowDonConfigVersion: metadata.WorkflowDonConfigVersion,
 		Observations:             i.Observations,
+		OverriddenEncoderName:    i.EncoderName,
+		OverriddenEncoderConfig:  i.EncoderConfig,
 		KeyID:                    c.KeyID,
 		ExpiresAt:                o.clock.Now().Add(requestTimeout),
 	}

pkg/capabilities/consensus/ocr3/models.go

@@ -16,5 +16,7 @@ type config struct {
 }
 
 type inputs struct {
-	Observations *values.List `json:"observations"`
+	Observations  *values.List `json:"observations" jsonschema:""`
+	EncoderName   string       `mapstructure:"encoder" json:"encoder,omitempty"`
+	EncoderConfig *values.Map  `mapstructure:"encoder_config" json:"encoder_config,omitempty"`
 }

pkg/capabilities/consensus/ocr3/ocr3cap/identical_consensus.go

@@ -32,17 +32,17 @@ func (c IdenticalConsensusConfig[T]) New(w *sdk.WorkflowSpecFactory, ref string,
 }
 
 type IdenticalConsensusInput[T any] struct {
-	Observations sdk.CapDefinition[T]
-}
-
-type IdenticalConsensusMergedInput[T any] struct {
-	Observations []T
+	Observation   sdk.CapDefinition[T]
+	Encoder       Encoder
+	EncoderConfig EncoderConfig
 }
 
 func (input IdenticalConsensusInput[T]) ToSteps() sdk.StepInputs {
 	return sdk.StepInputs{
 		Mapping: map[string]any{
-			"observations": input.Observations.Ref(),
+			"observations":  sdk.ListOf(input.Observation).Ref(),
+			"encoder":       input.Encoder,
+			"encoderConfig": input.EncoderConfig,
 		},
 	}
 }

pkg/capabilities/consensus/ocr3/ocr3cap/identical_consensus_test.go

@@ -26,7 +26,11 @@ func TestIdenticalConsensus(t *testing.T) {
 		Encoder:       ocr3.EncoderEVM,
 		EncoderConfig: ocr3.EncoderConfig{},
 		ReportID:      "0001",
-	}.New(workflow, "consensus", ocr3.IdenticalConsensusInput[basictrigger.TriggerOutputs]{Observations: trigger})
+	}.New(workflow, "consensus", ocr3.IdenticalConsensusInput[basictrigger.TriggerOutputs]{
+		Observation:   trigger,
+		Encoder:       "evm",
+		EncoderConfig: ocr3.EncoderConfig(map[string]any{"foo": "bar"}),
+	})
 
 	chainwriter.TargetConfig{
 		Address:    "0x1235",
@@ -55,9 +59,13 @@ func TestIdenticalConsensus(t *testing.T) {
 		Actions: []sdk.StepDefinition{},
 		Consensus: []sdk.StepDefinition{
 			{
-				ID:     "offchain_reporting@1.0.0",
-				Ref:    "consensus",
-				Inputs: sdk.StepInputs{Mapping: map[string]any{"observations": "$(trigger.outputs)"}},
+				ID:  "offchain_reporting@1.0.0",
+				Ref: "consensus",
+				Inputs: sdk.StepInputs{Mapping: map[string]any{
+					"observations":  []any{"$(trigger.outputs)"},
+					"encoder":       "evm",
+					"encoderConfig": map[string]any{"foo": "bar"},
+				}},
 				Config: map[string]any{
 					"encoder":            "EVM",
 					"encoder_config":     map[string]any{},