Cappl 1076

consensus/action/capability.go

@@ -121,7 +121,7 @@ func (c *consensusCapability) Initialise(ctx context.Context, dependencies core.
 	}
 
 	reportingPlugin, err := plugin.NewReportingPluginFactory(c.lggr, c.metrics, c.reqStore, c.SetRequestTimeout,
-		c.requestBatchSize)
+		c.requestBatchSize, defaultKeyBundleIDForValueConsensus)
 	if err != nil {
 		return fmt.Errorf("error when creating reporting plugin factory: %w", err)
 	}

consensus/go.mod

@@ -3,6 +3,8 @@ module github.com/smartcontractkit/capabilities/consensus
 go 1.25.3
 
 require (
+	github.com/cloudevents/sdk-go/v2 v2.16.1
+	github.com/google/uuid v1.6.0
 	github.com/jonboulle/clockwork v0.5.0
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/capabilities/libs v0.0.0-20250930133443-a868d4d9dee8
@@ -27,7 +29,6 @@ require (
 	github.com/cenkalti/backoff/v5 v5.0.2 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudevents/sdk-go/binding/format/protobuf/v2 v2.16.1 // indirect
-	github.com/cloudevents/sdk-go/v2 v2.16.1 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/fatih/color v1.18.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
@@ -41,7 +42,6 @@ require (
 	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/go-cmp v0.7.0 // indirect
-	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.1 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.2 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3 // indirect

consensus/oracle/consensus_execution.go

@@ -42,41 +42,25 @@ func CalculateOutcomeForObservations(
 	observations []*valuespb.Value,
 	consensusDescriptor *sdk.ConsensusDescriptor,
 	defaultValue *valuespb.Value,
-	minObservations int,
-	f int,
-) (*valuespb.Value, error) {
-	filtered, _, err := filterObservations(observations, minObservations)
-	if err != nil {
-		return nil, err
-	}
-
-	return handleDescriptor(lggr, consensusDescriptor, filtered, defaultValue, f)
-}
-
-func handleDescriptor(
-	lggr logger.Logger,
-	consensusDescriptor *sdk.ConsensusDescriptor,
-	filtered []*valuespb.Value,
-	defaultValue *valuespb.Value,
 	f int,
 ) (*valuespb.Value, error) {
 	switch desc := consensusDescriptor.GetDescriptor_().(type) {
 	case *sdk.ConsensusDescriptor_Aggregation:
 		aggregation := consensusDescriptor.GetAggregation()
 		switch aggregation {
 		case sdk.AggregationType_AGGREGATION_TYPE_IDENTICAL:
-			return handleIdenticalAggregation(lggr, filtered, f)
+			return handleIdenticalAggregation(lggr, observations, f)
 		case sdk.AggregationType_AGGREGATION_TYPE_MEDIAN:
-			return handleMedianAggregation(lggr, filtered, f)
+			return handleMedianAggregation(lggr, observations, f)
 		case sdk.AggregationType_AGGREGATION_TYPE_COMMON_PREFIX:
-			return handleCommonPrefixAggregation(lggr, filtered, f)
+			return handleCommonPrefixAggregation(lggr, observations, f)
 		case sdk.AggregationType_AGGREGATION_TYPE_COMMON_SUFFIX:
-			return handleCommonSuffixAggregation(lggr, filtered, f)
+			return handleCommonSuffixAggregation(lggr, observations, f)
 		default:
 			return nil, fmt.Errorf("unknown aggregation type: %s", aggregation)
 		}
 	case *sdk.ConsensusDescriptor_FieldsMap:
-		return handleFieldsMapAggregation(lggr, filtered, desc.FieldsMap.GetFields(), defaultValue, f)
+		return handleFieldsMapAggregation(lggr, observations, desc.FieldsMap.GetFields(), defaultValue, f)
 	default:
 		return nil, fmt.Errorf("unknown consensus descriptor type: %T", desc)
 	}
@@ -126,7 +110,7 @@ func handleFieldsMapAggregation(
 			}
 		}
 
-		aggregated, err = handleDescriptor(lggr, d, obsForKey, defaultForKey, f)
+		aggregated, err = CalculateOutcomeForObservations(lggr, obsForKey, d, defaultForKey, f)
 		if err == nil {
 			result[key] = aggregated
 			continue
@@ -456,35 +440,29 @@ func filterObservations(observationProtos []*valuespb.Value, minObservations int
 		return nil, nil, fmt.Errorf("insufficient observations (%d) to meet minimum (%d)", len(observationProtos), minObservations)
 	}
 
-	var dominantType reflect.Type
-	var highestCount int
-	var highestCountEqual bool
-
 	observationsByType := map[reflect.Type][]*valuespb.Value{}
-
 	for _, observation := range observationProtos {
 		if observation.Value == nil {
 			continue
 		}
 
 		tpe := reflect.TypeOf(observation.Value)
 		observationsByType[tpe] = append(observationsByType[tpe], observation)
-		count := len(observationsByType[tpe])
-		if count > highestCount {
-			highestCount = count
-			dominantType = tpe
-			highestCountEqual = false
-		} else if count == highestCount {
-			highestCountEqual = true
-		}
 	}
 
-	if highestCount < minObservations {
-		return nil, nil, fmt.Errorf("no single type met the minimum observation threshold of %d", minObservations)
+	var dominantType reflect.Type
+	for tpe, obsOfType := range observationsByType {
+		if len(obsOfType) >= minObservations {
+			if dominantType != nil {
+				// More than one type meets the threshold
+				return nil, nil, ErrMultipleValuesMetThreshold
+			}
+			dominantType = tpe
+		}
 	}
 
-	if highestCountEqual {
-		return nil, nil, ErrMultipleValuesMetThreshold
+	if dominantType == nil {
+		return nil, nil, ErrNoValuesMetThreshold
 	}
 
 	return observationsByType[dominantType], dominantType, nil

consensus/oracle/consensus_execution_test.go

@@ -25,7 +25,6 @@ func Test_CalculateOutcomeForObservations(t *testing.T) {
 		observations    []*valuespb.Value
 		descriptor      *sdk.ConsensusDescriptor
 		defaultValue    *valuespb.Value
-		minObs          int
 		f               int
 		expectedOutcome *valuespb.Value
 		expectedError   error
@@ -43,7 +42,7 @@ func Test_CalculateOutcomeForObservations(t *testing.T) {
 					Aggregation: sdk.AggregationType_AGGREGATION_TYPE_MEDIAN,
 				},
 			},
-			minObs:          3,
+			f:               2,
 			expectedOutcome: nil,
 			expectedError:   errors.New("insufficient observations"),
 		},
@@ -61,7 +60,7 @@ func Test_CalculateOutcomeForObservations(t *testing.T) {
 					Aggregation: sdk.AggregationType_AGGREGATION_TYPE_MEDIAN,
 				},
 			},
-			minObs:          5,
+			f:               4,
 			expectedOutcome: values.Proto(values.NewInt64(30)),
 			expectedError:   nil,
 		},
@@ -80,13 +79,28 @@ func Test_CalculateOutcomeForObservations(t *testing.T) {
 					Aggregation: sdk.AggregationType_AGGREGATION_TYPE_IDENTICAL,
 				},
 			},
-			minObs:          5,
 			f:               3,
 			expectedOutcome: values.Proto(values.NewInt64(42)),
 			expectedError:   nil,
 		},
 		{
-			name: "median: mixed types, one dominant (int64) - handled by filtering",
+			name: "median: mixed types, two eligible types (int64, float64) - error returned",
+			f:    1,
+			observations: []*valuespb.Value{
+				values.Proto(values.NewInt64(10)), values.Proto(values.NewFloat64(1.0)),
+				values.Proto(values.NewInt64(20)), values.Proto(values.NewFloat64(2.0)),
+				values.Proto(values.NewInt64(30)), values.Proto(values.NewInt64(40)),
+			},
+			descriptor: &sdk.ConsensusDescriptor{
+				Descriptor_: &sdk.ConsensusDescriptor_Aggregation{
+					Aggregation: sdk.AggregationType_AGGREGATION_TYPE_MEDIAN,
+				},
+			},
+			expectedError: ErrMultipleValuesMetThreshold,
+		},
+		{
+			name: "median: mixed types, one eligible type (int64) - handled by filtering",
+			f:    2,
 			observations: []*valuespb.Value{
 				values.Proto(values.NewInt64(10)), values.Proto(values.NewFloat64(1.0)),
 				values.Proto(values.NewInt64(20)), values.Proto(values.NewFloat64(2.0)),
@@ -155,7 +169,6 @@ func Test_CalculateOutcomeForObservations(t *testing.T) {
 				mustWrap(s{Val: 50, OtherField: "common", PrefixSlice: []int64{1, 2, 6}, Nest: s1{Val: 102}, SuffixSlice: []int64{42, 2, 3}}),
 			},
 			descriptor: cre.ConsensusAggregationFromTags[s]().Descriptor(),
-			minObs:     5,
 			f:          2,
 			expectedOutcome: mustWrap(s{
 				Val:         30,
@@ -201,7 +214,6 @@ func Test_CalculateOutcomeForObservations(t *testing.T) {
 					Aggregation: sdk.AggregationType_AGGREGATION_TYPE_UNSPECIFIED,
 				},
 			},
-			minObs:          1,
 			expectedOutcome: nil,
 			expectedError:   errors.New("unknown aggregation type"),
 		},
@@ -214,7 +226,6 @@ func Test_CalculateOutcomeForObservations(t *testing.T) {
 				tc.observations,
 				tc.descriptor,
 				tc.defaultValue,
-				tc.minObs,
 				tc.f,
 			)
 

consensus/oracle/identical_observations_test.go

@@ -254,7 +254,7 @@ func Test_filterObservations(t *testing.T) {
 			minObservations:      3,
 			expectedObservations: nil,
 			expectedType:         nil,
-			expectedError:        errors.New("no single type met the minimum observation threshold of 3"),
+			expectedError:        errors.New("no values met f+1 threshold"),
 		},
 		{
 			name: "dominant type is TypeNil",
@@ -266,7 +266,7 @@ func Test_filterObservations(t *testing.T) {
 			minObservations:      2,
 			expectedObservations: nil,
 			expectedType:         nil,
-			expectedError:        errors.New("no single type met the minimum observation threshold of 2"),
+			expectedError:        errors.New("no values met f+1 threshold"),
 		},
 		{
 			name: "all observations are of dominant type",

consensus/oracle/plugin/batching.go → consensus/oracle/plugin/batching/batching.go

@@ -1,23 +1,14 @@
-package plugin
+package batching
 
 import (
-	"google.golang.org/protobuf/proto"
+	"context"
 
-	"github.com/smartcontractkit/capabilities/consensus/oracle"
+	"google.golang.org/protobuf/proto"
 )
 
-// IDKey represents a unique identifier for a ConsensusRequest used for deduplication
-type IDKey struct {
-	workflowExecutionID string
-	referenceID         string
-}
-
-// GetIDKey creates a unique identifier from a ConsensusRequest for deduplication
-func GetIDKey(rq *oracle.ConsensusRequest) IDKey {
-	return IDKey{
-		workflowExecutionID: rq.Metadata.WorkflowExecutionID,
-		referenceID:         rq.Metadata.ReferenceID,
-	}
+type metrics interface {
+	IncBatchCapacityExceeded(ctx context.Context, step string)
+	IncBatchRequestsTotal(ctx context.Context, step string)
 }
 
 // varintSize calculates the size of a varint encoding
@@ -46,24 +37,23 @@ func varintSize(x uint64) int {
 	}
 }
 
-// CalculateMessageSize calculates the marshalled size of any proto message
-func CalculateMessageSize(message proto.Message) int {
-	if message == nil {
-		return 0
-	}
-
+// calculateMessageSize calculates the marshalled size of any proto message
+func calculateMessageSize(message proto.Message) int {
 	// Use proto.Size which gives us the exact marshalled size
 	return proto.Size(message)
 }
 
-// BatchHasCapacity checks if adding a new proto message would exceed the size limit
-func BatchHasCapacity(cachedSize int, message proto.Message, maxSizeBytes int, incBatchRequestsMetric func(),
-	incBatchSizeExceededMetric func()) (bool, int) {
-	incBatchRequestsMetric()
+func batchHasCapacityForMessageOnSlice(cachedSize int, message proto.Message, maxSizeBytes int) (bool, int) {
+	numBytes := proto.Size(message)
+	return batchHasCapacityForSliceBytes(cachedSize, numBytes, maxSizeBytes)
+}
 
-	// Calculate size if we add one more message
-	newMessageSize := proto.Size(message)
+func batchHasCapacityForStringOnSlice(cachedSize int, message string, maxSizeBytes int) (bool, int) {
+	numBytes := len(message)
+	return batchHasCapacityForSliceBytes(cachedSize, numBytes, maxSizeBytes)
+}
 
+func batchHasCapacityForSliceBytes(cachedSize int, newMessageSize int, maxSizeBytes int) (bool, int) {
 	// Add protobuf field overhead: tag (field number + wire type) + length prefix
 	// For repeated fields in protobuf, each element gets:
 	// - Tag: field number (1 for the repeated field) << 3 | wire type (2 for length-delimited)
@@ -78,7 +68,6 @@ func BatchHasCapacity(cachedSize int, message proto.Message, maxSizeBytes int, i
 
 	// Check against config
 	if totalSizeWithNewMessage > maxSizeBytes {
-		incBatchSizeExceededMetric()
 		// Stop adding more messages
 		return false, cachedSize
 	}

consensus/oracle/plugin/batching/observation_batch.go

@@ -0,0 +1,78 @@
+package batching
+
+import (
+	"context"
+	"fmt"
+
+	"google.golang.org/protobuf/proto"
+
+	oracletypes "github.com/smartcontractkit/capabilities/consensus/oracle/types"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+)
+
+type ObservationBatch struct {
+	oracletypes.Observation
+
+	lggr                       logger.Logger
+	currentSerialisedBatchSize int
+	metrics                    metrics
+	maxObservationLengthBytes  int
+}
+
+func NewObservationBatch(ctx context.Context, lggr logger.Logger, maxObservationLengthBytes int, metrics metrics) *ObservationBatch {
+	metrics.IncBatchRequestsTotal(ctx, "observation")
+	observations := make(map[string]*oracletypes.RequestObservation)
+	obs := &oracletypes.Observation{Observations: observations}
+	messageSize := calculateMessageSize(obs)
+
+	return &ObservationBatch{
+		Observation:                oracletypes.Observation{Observations: observations},
+		lggr:                       lggr,
+		currentSerialisedBatchSize: messageSize,
+		maxObservationLengthBytes:  maxObservationLengthBytes,
+		metrics:                    metrics,
+	}
+}
+
+func (ob *ObservationBatch) AddObservation(ctx context.Context, reqObs *oracletypes.RequestObservation) bool {
+	// Adding an entry to a map is the same as adding a key-value pair to a slice for size calculation purposes
+	mapEntry := oracletypes.ObservationMapEntry{
+		Key:   reqObs.Metadata.RequestId,
+		Value: reqObs,
+	}
+
+	mapEntrySize := proto.Size(&mapEntry)
+	ok, newSize := batchHasCapacityForSliceBytes(ob.currentSerialisedBatchSize, mapEntrySize, ob.maxObservationLengthBytes)
+
+	if !ok {
+		ob.metrics.IncBatchCapacityExceeded(ctx, "observation")
+		return false
+	}
+
+	ob.currentSerialisedBatchSize = newSize
+	ob.Observations[reqObs.Metadata.RequestId] = reqObs
+
+	return true
+}
+
+func (ob *ObservationBatch) CurrentSerialisedBatchSize() int {
+	return ob.currentSerialisedBatchSize
+}
+
+func (ob *ObservationBatch) SerialiseObservationBatch() ([]byte, error) {
+	serialisedBatch, err := proto.MarshalOptions{Deterministic: true}.Marshal(&ob.Observation)
+	if err != nil {
+		return nil, fmt.Errorf("failed to serialise batch of observations: %w", err)
+	}
+
+	ob.lggr.Debugw("serialised observation batch", "numObservations", len(ob.Observations), "actualSizeBytes", len(serialisedBatch),
+		"calculatedSizeBytes", ob.currentSerialisedBatchSize,
+		"maxObservationLengthBytes", ob.maxObservationLengthBytes)
+
+	return serialisedBatch, nil
+}
+
+func (ob *ObservationBatch) NumObservationsInBatch() int {
+	return len(ob.Observations)
+}