[CM-380] Identical Aggregator

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

@@ -0,0 +1,152 @@
+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"
+	"github.com/smartcontractkit/chainlink-common/pkg/values/pb"
+
+	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
+	// If true, the encoder name and config will be overridden with the values
+	// provided under configured indices of the observations array (as long as consensus is reached).
+	OverrideEncoder               bool
+	EncoderNameObservationIndex   int
+	EncoderConfigObservationIndex int
+}
+
+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)
+	var encoderName string
+	var encoderConfig *pb.Map
+	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
+		}
+		if a.config.OverrideEncoder {
+			if idx == a.config.EncoderNameObservationIndex {
+				// extract encoder name from observation
+				name, ok := highestObservation.(*values.String)
+				if !ok {
+					return nil, fmt.Errorf("expected encoder name under index %d to be a String", idx)
+				}
+				encoderName = name.Underlying
+			} else if idx == a.config.EncoderConfigObservationIndex {
+				// extract encoder config from observation
+				conf, ok := highestObservation.(*values.Map)
+				if !ok {
+					return nil, fmt.Errorf("expected encoder config under index %d to be a Map", idx)
+				}
+				encoderConfig = values.ProtoMap(conf)
+			}
+		}
+	}
+	valMap, err := values.NewMap(outcome)
+	if err != nil {
+		return nil, err
+	}
+	return &types.AggregationOutcome{
+		EncodableOutcome: values.ProtoMap(valMap),
+		Metadata:         nil,
+		ShouldReport:     true,
+		EncoderName:      encoderName,
+		EncoderConfig:    encoderConfig,
+	}, 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,38 @@
+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_TwoRounds(t *testing.T) {
+	config := getConfig(t)
+	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)
+}
+
+func getConfig(t *testing.T) *values.Map {
+	unwrappedConfig := map[string]any{
+		"expectedObservationsLen": 1,
+	}
+	config, err := values.NewMap(unwrappedConfig)
+	require.NoError(t, err)
+	return config
+}

pkg/capabilities/consensus/ocr3/capability.go

@@ -125,7 +125,7 @@ func (o *capability) RegisterToWorkflow(ctx context.Context, request capabilitie
 	}
 	o.aggregators[request.Metadata.WorkflowID] = agg
 
-	encoder, err := o.encoderFactory(c.EncoderConfig)
+	encoder, err := o.encoderFactory(c.Encoder, c.EncoderConfig, o.lggr)
 	if err != nil {
 		return err
 	}
@@ -143,7 +143,7 @@ func (o *capability) getAggregator(workflowID string) (types.Aggregator, error)
 	return agg, nil
 }
 
-func (o *capability) getEncoder(workflowID string) (types.Encoder, error) {
+func (o *capability) getEncoderByWorkflowID(workflowID string) (types.Encoder, error) {
 	enc, ok := o.encoders[workflowID]
 	if !ok {
 		return nil, fmt.Errorf("no aggregator found for workflowID %s", workflowID)
@@ -152,6 +152,10 @@ func (o *capability) getEncoder(workflowID string) (types.Encoder, error) {
 	return enc, nil
 }
 
+func (o *capability) getEncoderByName(encoderName string, config *values.Map) (types.Encoder, error) {
+	return o.encoderFactory(encoderName, config, o.lggr)
+}
+
 func (o *capability) getRegisteredWorkflowsIDs() []string {
 	o.mu.RLock()
 	defer o.mu.RUnlock()

pkg/capabilities/consensus/ocr3/capability_test.go

@@ -40,7 +40,7 @@ type encoder struct {
 	types.Encoder
 }
 
-func mockEncoderFactory(_ *values.Map) (types.Encoder, error) {
+func mockEncoderFactory(_ string, _ *values.Map, _ logger.Logger) (types.Encoder, error) {
 	return &encoder{}, nil
 }
 

pkg/capabilities/consensus/ocr3/reporting_plugin.go

@@ -27,7 +27,8 @@ var _ ocr3types.ReportingPlugin[[]byte] = (*reportingPlugin)(nil)
 
 type capabilityIface interface {
 	getAggregator(workflowID string) (pbtypes.Aggregator, error)
-	getEncoder(workflowID string) (pbtypes.Encoder, error)
+	getEncoderByWorkflowID(workflowID string) (pbtypes.Encoder, error)
+	getEncoderByName(encoderName string, config *values.Map) (pbtypes.Encoder, error)
 	getRegisteredWorkflowsIDs() []string
 	unregisterWorkflowID(workflowID string)
 }
@@ -390,7 +391,7 @@ func (r *reportingPlugin) Reports(seqNr uint64, outcome ocr3types.Outcome) ([]oc
 			ShouldReport: outcome.ShouldReport,
 		}
 
-		var report []byte
+		var rawReport []byte
 		if info.ShouldReport {
 			meta := &pbtypes.Metadata{
 				Version:          1,
@@ -409,10 +410,26 @@ func (r *reportingPlugin) Reports(seqNr uint64, outcome ocr3types.Outcome) ([]oc
 				continue
 			}
 
-			enc, err := r.r.getEncoder(id.WorkflowId)
-			if err != nil {
-				lggr.Errorw("could not retrieve encoder for workflow", "error", err)
-				continue
+			var encoder pbtypes.Encoder
+			if newOutcome.EncoderName != "" {
+				lggr.Debugw("using encoder from outcome", "encoderName", newOutcome.EncoderName, "executionID", report.Id.WorkflowExecutionId)
+				encoderConfig, err2 := values.FromMapValueProto(newOutcome.EncoderConfig)
+				if err2 != nil {
+					lggr.Errorw("could not convert desired encoder config to values.Map", "error", err2, "executionID", report.Id.WorkflowExecutionId)
+				} else {
+					encoder, err2 = r.r.getEncoderByName(newOutcome.EncoderName, encoderConfig)
+					if err2 != nil {
+						lggr.Errorw("could not retrieve desired encoder, will use per-workflow default", "error", err2, "executionID", report.Id.WorkflowExecutionId)
+					}
+				}
+			}
+
+			if encoder == nil {
+				encoder, err = r.r.getEncoderByWorkflowID(id.WorkflowId)
+				if err != nil {
+					lggr.Errorw("could not retrieve encoder for workflow", "error", err)
+					continue
+				}
 			}
 
 			mv, err := values.FromMapValueProto(newOutcome.EncodableOutcome)
@@ -421,7 +438,7 @@ func (r *reportingPlugin) Reports(seqNr uint64, outcome ocr3types.Outcome) ([]oc
 				continue
 			}
 
-			report, err = enc.Encode(context.Background(), *mv)
+			rawReport, err = encoder.Encode(context.Background(), *mv)
 			if err != nil {
 				r.lggr.Errorw("could not encode report for workflow", "error", err)
 				continue
@@ -436,7 +453,7 @@ func (r *reportingPlugin) Reports(seqNr uint64, outcome ocr3types.Outcome) ([]oc
 
 		// Append every report, even if shouldReport = false, to let the transmitter mark the step as complete.
 		reports = append(reports, ocr3types.ReportWithInfo[[]byte]{
-			Report: report,
+			Report: rawReport,
 			Info:   infob,
 		})
 	}

pkg/capabilities/consensus/ocr3/reporting_plugin_test.go

@@ -75,9 +75,11 @@ func TestReportingPlugin_Query(t *testing.T) {
 }
 
 type mockCapability struct {
+	t                   *testing.T
 	aggregator          *aggregator
 	encoder             *enc
 	registeredWorkflows map[string]bool
+	expectedEncoderName string
 }
 
 type aggregator struct {
@@ -114,7 +116,12 @@ func (mc *mockCapability) getAggregator(workflowID string) (pbtypes.Aggregator,
 	return mc.aggregator, nil
 }
 
-func (mc *mockCapability) getEncoder(workflowID string) (pbtypes.Encoder, error) {
+func (mc *mockCapability) getEncoderByWorkflowID(workflowID string) (pbtypes.Encoder, error) {
+	return mc.encoder, nil
+}
+
+func (mc *mockCapability) getEncoderByName(encoderName string, config *values.Map) (pbtypes.Encoder, error) {
+	require.Equal(mc.t, mc.expectedEncoderName, encoderName)
 	return mc.encoder, nil
 }
 
@@ -460,10 +467,13 @@ func TestReportingPlugin_Reports_NilDerefs(t *testing.T) {
 
 func TestReportingPlugin_Reports_ShouldReportTrue(t *testing.T) {
 	lggr := logger.Test(t)
+	dynamicEncoderName := "special_encoder"
 	s := requests.NewStore()
 	mcap := &mockCapability{
-		aggregator: &aggregator{},
-		encoder:    &enc{},
+		t:                   t,
+		aggregator:          &aggregator{},
+		encoder:             &enc{},
+		expectedEncoderName: dynamicEncoderName,
 	}
 	rp, err := newReportingPlugin(s, mcap, defaultBatchSize, ocr3types.ReportingPluginConfig{}, defaultOutcomePruningThreshold, lggr)
 	require.NoError(t, err)
@@ -494,6 +504,7 @@ func TestReportingPlugin_Reports_ShouldReportTrue(t *testing.T) {
 				Outcome: &pbtypes.AggregationOutcome{
 					EncodableOutcome: nmp,
 					ShouldReport:     true,
+					EncoderName:      dynamicEncoderName,
 				},
 			},
 		},

pkg/capabilities/consensus/ocr3/transmitter_test.go

@@ -41,7 +41,9 @@ func TestTransmitter(t *testing.T) {
 		clockwork.NewFakeClock(),
 		10*time.Second,
 		mockAggregatorFactory,
-		func(config *values.Map) (pbtypes.Encoder, error) { return &encoder{}, nil },
+		func(_ string, _ *values.Map, _ logger.Logger) (pbtypes.Encoder, error) {
+			return &encoder{}, nil
+		},
 		lggr,
 		10,
 	)
@@ -127,7 +129,9 @@ func TestTransmitter_ShouldReportFalse(t *testing.T) {
 		clockwork.NewFakeClock(),
 		10*time.Second,
 		mockAggregatorFactory,
-		func(config *values.Map) (pbtypes.Encoder, error) { return &encoder{}, nil },
+		func(_ string, _ *values.Map, _ logger.Logger) (pbtypes.Encoder, error) {
+			return &encoder{}, nil
+		},
 		lggr,
 		10,
 	)

pkg/capabilities/consensus/ocr3/types/encoder.go

@@ -3,14 +3,15 @@ package types
 import (
 	"context"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/values"
 )
 
 type Encoder interface {
 	Encode(ctx context.Context, input values.Map) ([]byte, error)
 }
 
-type EncoderFactory func(config *values.Map) (Encoder, error)
+type EncoderFactory func(name string, config *values.Map, lggr logger.Logger) (Encoder, error)
 
 type SignedReport struct {
 	Report []byte