-
Notifications
You must be signed in to change notification settings - Fork 1.7k
/
aggregation.go
119 lines (112 loc) · 3.83 KB
/
aggregation.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
package functions
import (
"bytes"
"encoding/binary"
"fmt"
"sort"
"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/functions/config"
"github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/functions/encoding"
)
func CanAggregate(N int, F int, observations []*encoding.ProcessedRequest) bool {
return N > 0 && F >= 0 && len(observations) > 0 && len(observations) <= N && len(observations) >= 2*F+1
}
func Aggregate(aggMethod config.AggregationMethod, observations []*encoding.ProcessedRequest) (*encoding.ProcessedRequest, error) {
if len(observations) == 0 {
return nil, fmt.Errorf("empty observation list passed for aggregation")
}
var errored []*encoding.ProcessedRequest
var successful []*encoding.ProcessedRequest
reqId := observations[0].RequestID
finalResult := encoding.ProcessedRequest{
RequestID: reqId,
Result: []byte{},
Error: []byte{},
}
for _, obs := range observations {
if !bytes.Equal(obs.RequestID, reqId) {
return nil, fmt.Errorf("inconsistent request IDs in aggregated observations %v vs %v", formatRequestId(obs.RequestID), formatRequestId(reqId))
}
if obs.GetError() != nil && len(obs.GetError()) > 0 {
errored = append(errored, obs)
} else {
successful = append(successful, obs)
}
}
resultIsError := len(errored) > len(successful)
var toAggregate []*encoding.ProcessedRequest
var rawData [][]byte
if resultIsError {
toAggregate = errored
for _, item := range errored {
rawData = append(rawData, item.Error)
}
} else {
toAggregate = successful
for _, item := range successful {
rawData = append(rawData, item.Result)
}
}
// Metadata (CallbackGasLimit, CoordinatorContract and OnchainMetadata) is aggregated using MODE method
finalResult.CallbackGasLimit, finalResult.CoordinatorContract, finalResult.OnchainMetadata = aggregateMetadata(toAggregate)
if resultIsError {
// Errors are always aggregated using MODE method
finalResult.Error = aggregateMode(rawData)
} else {
switch aggMethod {
case config.AggregationMethod_AGGREGATION_MODE:
finalResult.Result = aggregateMode(rawData)
case config.AggregationMethod_AGGREGATION_MEDIAN:
finalResult.Result = aggregateMedian(rawData)
default:
return nil, fmt.Errorf("unsupported aggregation method: %s", aggMethod)
}
}
return &finalResult, nil
}
func aggregateMetadata(items []*encoding.ProcessedRequest) (uint32, []byte, []byte) {
gasLimitBytes := make([][]byte, len(items))
coordinatorContracts := make([][]byte, len(items))
onchainMetadata := make([][]byte, len(items))
for i, item := range items {
gasLimitBytes[i] = make([]byte, 4)
binary.BigEndian.PutUint32(gasLimitBytes[i], item.CallbackGasLimit)
coordinatorContracts[i] = item.CoordinatorContract
if coordinatorContracts[i] == nil {
coordinatorContracts[i] = []byte{}
}
onchainMetadata[i] = item.OnchainMetadata
if onchainMetadata[i] == nil {
onchainMetadata[i] = []byte{}
}
}
aggGasLimitBytes := aggregateMode(gasLimitBytes)
aggGasLimitUint32 := binary.BigEndian.Uint32(aggGasLimitBytes)
aggCoordinatorContract := aggregateMode(coordinatorContracts)
aggOnchainMetadata := aggregateMode(onchainMetadata)
return aggGasLimitUint32, aggCoordinatorContract, aggOnchainMetadata
}
func aggregateMode(items [][]byte) []byte {
counts := make(map[string]int)
var mostFrequent []byte
highestFreq := 0
for _, item := range items {
str := string(item)
currCount := counts[str] + 1
counts[str] = currCount
if currCount > highestFreq {
highestFreq = currCount
mostFrequent = item
}
}
return mostFrequent
}
func aggregateMedian(items [][]byte) []byte {
sort.Slice(items, func(i, j int) bool {
if len(items[i]) != len(items[j]) {
// NOTE: this doesn't account for extra leading zeros
return len(items[i]) < len(items[j])
}
return bytes.Compare(items[i], items[j]) < 0
})
return items[(len(items)-1)/2]
}