-
Notifications
You must be signed in to change notification settings - Fork 1.7k
/
transmitter.go
153 lines (129 loc) · 4.36 KB
/
transmitter.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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
package llo
import (
"context"
"crypto/ed25519"
"fmt"
"github.com/ethereum/go-ethereum/accounts/abi"
"github.com/smartcontractkit/libocr/offchainreporting2/chains/evmutil"
"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
"github.com/smartcontractkit/libocr/offchainreporting2plus/types"
ocr2types "github.com/smartcontractkit/libocr/offchainreporting2plus/types"
"github.com/smartcontractkit/chainlink-common/pkg/services"
llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo"
"github.com/smartcontractkit/chainlink/v2/core/logger"
"github.com/smartcontractkit/chainlink/v2/core/services/keystore/keys/ocr2key"
"github.com/smartcontractkit/chainlink/v2/core/services/relay/evm/mercury/wsrpc"
"github.com/smartcontractkit/chainlink/v2/core/services/relay/evm/mercury/wsrpc/pb"
)
// LLO Transmitter implementation, based on
// core/services/relay/evm/mercury/transmitter.go
// TODO: prom metrics (common with mercury/transmitter.go?)
// https://smartcontract-it.atlassian.net/browse/MERC-3659
const (
// Mercury server error codes
DuplicateReport = 2
// TODO: revisit these values in light of parallel composition
// https://smartcontract-it.atlassian.net/browse/MERC-3659
// maxTransmitQueueSize = 10_000
// maxDeleteQueueSize = 10_000
// transmitTimeout = 5 * time.Second
)
var PayloadTypes = getPayloadTypes()
func getPayloadTypes() abi.Arguments {
mustNewType := func(t string) abi.Type {
result, err := abi.NewType(t, "", []abi.ArgumentMarshaling{})
if err != nil {
panic(fmt.Sprintf("Unexpected error during abi.NewType: %s", err))
}
return result
}
return abi.Arguments([]abi.Argument{
{Name: "reportContext", Type: mustNewType("bytes32[2]")},
{Name: "report", Type: mustNewType("bytes")},
{Name: "rawRs", Type: mustNewType("bytes32[]")},
{Name: "rawSs", Type: mustNewType("bytes32[]")},
{Name: "rawVs", Type: mustNewType("bytes32")},
})
}
type Transmitter interface {
llotypes.Transmitter
services.Service
}
type transmitter struct {
services.StateMachine
lggr logger.Logger
rpcClient wsrpc.Client
fromAccount string
}
func NewTransmitter(lggr logger.Logger, rpcClient wsrpc.Client, fromAccount ed25519.PublicKey) Transmitter {
return &transmitter{
services.StateMachine{},
lggr,
rpcClient,
fmt.Sprintf("%x", fromAccount),
}
}
func (t *transmitter) Start(ctx context.Context) error {
return nil
}
func (t *transmitter) Close() error {
return nil
}
func (t *transmitter) HealthReport() map[string]error {
report := map[string]error{t.Name(): t.Healthy()}
services.CopyHealth(report, t.rpcClient.HealthReport())
return report
}
func (t *transmitter) Name() string { return t.lggr.Name() }
func (t *transmitter) Transmit(
ctx context.Context,
digest types.ConfigDigest,
seqNr uint64,
report ocr3types.ReportWithInfo[llotypes.ReportInfo],
sigs []types.AttributedOnchainSignature,
) (err error) {
var payload []byte
switch report.Info.ReportFormat {
case llotypes.ReportFormatJSON:
fallthrough
case llotypes.ReportFormatEVM:
payload, err = encodeEVM(digest, seqNr, report.Report, sigs)
default:
return fmt.Errorf("Transmit failed; unsupported report format: %q", report.Info.ReportFormat)
}
if err != nil {
return fmt.Errorf("Transmit: encode failed; %w", err)
}
req := &pb.TransmitRequest{
Payload: payload,
ReportFormat: uint32(report.Info.ReportFormat),
}
// TODO: persistenceManager and queueing, error handling, retry etc
// https://smartcontract-it.atlassian.net/browse/MERC-3659
_, err = t.rpcClient.Transmit(ctx, req)
return err
}
func encodeEVM(digest types.ConfigDigest, seqNr uint64, report ocr2types.Report, sigs []types.AttributedOnchainSignature) ([]byte, error) {
var rs [][32]byte
var ss [][32]byte
var vs [32]byte
for i, as := range sigs {
r, s, v, err := evmutil.SplitSignature(as.Signature)
if err != nil {
return nil, fmt.Errorf("eventTransmit(ev): error in SplitSignature: %w", err)
}
rs = append(rs, r)
ss = append(ss, s)
vs[i] = v
}
rawReportCtx := ocr2key.RawReportContext3(digest, seqNr)
payload, err := PayloadTypes.Pack(rawReportCtx, []byte(report), rs, ss, vs)
if err != nil {
return nil, fmt.Errorf("abi.Pack failed; %w", err)
}
return payload, nil
}
// FromAccount returns the stringified (hex) CSA public key
func (t *transmitter) FromAccount() (ocr2types.Account, error) {
return ocr2types.Account(t.fromAccount), nil
}