Skip to content

New consensus aggregator: FREQUENCY_LIST #679

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
bolekk wants to merge 1 commit into
base: main
Choose a base branch
from
+152 −8
Draft

New consensus aggregator: FREQUENCY_LIST #679

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions consensus/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ require (
github.com/shopspring/decimal v1.4.0
github.com/smartcontractkit/capabilities/libs v0.0.0-20260210010829-97eb42ca2924
github.com/smartcontractkit/chainlink-common v0.11.2-0.20260529092756-a94bc8ce96d6
github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260526195338-adcf8013a1b7
github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260707203317-661b54b51a33
github.com/smartcontractkit/cre-sdk-go v0.9.0
github.com/smartcontractkit/libocr v0.0.0-20250912173940-f3ab0246e23d
github.com/stretchr/testify v1.11.1
Expand All @@ -22,6 +22,7 @@ require (

require (
github.com/Masterminds/semver/v3 v3.4.0 // indirect
github.com/ProjectZKM/Ziren/crates/go-runtime/zkvm_runtime v0.0.0-20251001021608-1fe7b43fc4d6 // indirect
github.com/XSAM/otelsql v0.37.0 // indirect
github.com/bahlo/generic-list-go v0.2.0 // indirect
github.com/beorn7/perks v1.0.1 // indirect
Expand All @@ -30,6 +31,8 @@ require (
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/cloudevents/sdk-go/binding/format/protobuf/v2 v2.16.1 // indirect
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0 // indirect
github.com/ethereum/go-ethereum v1.17.2 // indirect
github.com/fatih/color v1.18.0 // indirect
github.com/fxamacker/cbor/v2 v2.9.0 // indirect
github.com/gabriel-vasile/mimetype v1.4.8 // indirect
Expand All @@ -51,6 +54,7 @@ require (
github.com/hashicorp/go-hclog v1.6.3 // indirect
github.com/hashicorp/go-plugin v1.8.0 // indirect
github.com/hashicorp/yamux v0.1.2 // indirect
github.com/holiman/uint256 v1.3.2 // indirect
github.com/invopop/jsonschema v0.13.0 // indirect
github.com/jackc/pgpassfile v1.0.0 // indirect
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
Expand Down Expand Up @@ -80,7 +84,7 @@ require (
github.com/prometheus/procfs v0.20.1 // indirect
github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 // indirect
github.com/scylladb/go-reflectx v1.0.1 // indirect
github.com/smartcontractkit/chain-selectors v1.0.100 // indirect
github.com/smartcontractkit/chain-selectors v1.0.104 // indirect
github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260529092756-a94bc8ce96d6 // indirect
github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b // indirect
github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260205130626-db2a2aab956b // indirect
Expand Down
18 changes: 12 additions & 6 deletions consensus/go.sum
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

90 changes: 90 additions & 0 deletions consensus/oracle/consensus_execution.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,8 @@ func CalculateOutcomeForObservations(
return handleCommonPrefixAggregation(lggr, observations, f, errorsMigrationFlag)
case sdk.AggregationType_AGGREGATION_TYPE_COMMON_SUFFIX:
return handleCommonSuffixAggregation(lggr, observations, f, errorsMigrationFlag)
case sdk.AggregationType_AGGREGATION_TYPE_FREQUENCY_LIST:
return handleValueCountsAggregation(lggr, observations, f)
default:
return nil, fmt.Errorf("unknown aggregation type: %s", aggregation)
}
Expand Down Expand Up @@ -338,6 +340,94 @@ func handleIdenticalAggregation(_ logger.Logger, values []*valuespb.Value, f int
return uniqueCandidate, nil
}

// handleValueCountsAggregation returns a list of all distinct observation values
// together with how many times each was seen. Each list element is a map with
// "value" and "count" fields. Results are sorted by count descending, then by
// serialized value ascending for determinism.
func handleValueCountsAggregation(
_ logger.Logger,
observations []*valuespb.Value,
f int,
) (*valuespb.Value, error) {
if len(observations) < f+1 {
return nil, ErrInsufficientObservations
}

type valueOccurrence struct {
count int64
value *valuespb.Value
}

var (
marshaler = &proto.MarshalOptions{Deterministic: true}
occurrences = make(map[string]valueOccurrence)
)

for _, currentValue := range observations {
if currentValue == nil || currentValue.Value == nil {
continue
}

b, err := marshaler.Marshal(currentValue)
if err != nil {
return nil, fmt.Errorf("unable to marshal value: %w", err)
}
key := string(b)

occurrence := occurrences[key]
occurrence.count++
if occurrence.value == nil {
occurrence.value = currentValue
}
occurrences[key] = occurrence
}

if len(occurrences) == 0 {
return nil, ErrInsufficientObservations
}

type countedValue struct {
key string
count int64
value *valuespb.Value
}

counted := make([]countedValue, 0, len(occurrences))
for key, occurrence := range occurrences {
counted = append(counted, countedValue{
key: key,
count: occurrence.count,
value: occurrence.value,
})
}

slices.SortFunc(counted, func(a, b countedValue) int {
if a.count != b.count {
if a.count > b.count {
return -1
}
return 1
}
if a.key < b.key {
return -1
}
if a.key > b.key {
return 1
}
return 0
})

result := make([]*valuespb.Value, 0, len(counted))
for _, entry := range counted {
result = append(result, valuespb.NewMapValue(map[string]*valuespb.Value{
"value": entry.value,
"count": valuespb.NewInt64Value(entry.count),
}))
}

return valuespb.NewListValue(result), nil
}

// handleCommonSuffixAggregation reverses the underlying lists in the slice of
// observations and delegates logic to handleCommonPrefixAggregation and then
// reverses the result a final time.
Expand Down
44 changes: 44 additions & 0 deletions consensus/oracle/consensus_execution_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,6 +204,50 @@ func Test_CalculateOutcomeForObservations(t *testing.T) {
},
expectedOutcome: mustNewList("7", "8", "9"),
},
{
name: "value counts aggregation: distinct values with counts",
observations: []*valuespb.Value{
values.Proto(values.NewInt64(42)),
values.Proto(values.NewInt64(42)),
values.Proto(values.NewInt64(42)),
values.Proto(values.NewInt64(50)),
values.Proto(values.NewString("malicious")),
},
descriptor: &sdk.ConsensusDescriptor{
Descriptor_: &sdk.ConsensusDescriptor_Aggregation{
Aggregation: sdk.AggregationType_AGGREGATION_TYPE_FREQUENCY_LIST,
},
},
f: 2,
expectedOutcome: valuespb.NewListValue([]*valuespb.Value{
valuespb.NewMapValue(map[string]*valuespb.Value{
"value": values.Proto(values.NewInt64(42)),
"count": valuespb.NewInt64Value(3),
}),
valuespb.NewMapValue(map[string]*valuespb.Value{
"value": values.Proto(values.NewString("malicious")),
"count": valuespb.NewInt64Value(1),
}),
valuespb.NewMapValue(map[string]*valuespb.Value{
"value": values.Proto(values.NewInt64(50)),
"count": valuespb.NewInt64Value(1),
}),
}),
},
{
name: "value counts aggregation: insufficient observations",
observations: []*valuespb.Value{
values.Proto(values.NewInt64(10)),
values.Proto(values.NewInt64(20)),
},
descriptor: &sdk.ConsensusDescriptor{
Descriptor_: &sdk.ConsensusDescriptor_Aggregation{
Aggregation: sdk.AggregationType_AGGREGATION_TYPE_FREQUENCY_LIST,
},
},
f: 2,
expectedError: ErrInsufficientObservations,
},
{
name: "unknown aggregation type (UNSPECIFIED)",
observations: []*valuespb.Value{
Expand Down
Loading