Skip to content

Commit 7ab8191

Browse files
gtklockergtklocker
authored
Merge pull request #125 from smartcontractkit/copy-4cca1fb852fefbaa619fb7df6a6a5b45c2d1b375
Improvements:
2 parents bdc84e1 + 43ab92f commit 7ab8191

27 files changed

Lines changed: 1424 additions & 799 deletions

‎networking/ocr_endpoint_v3.go‎

Lines changed: 24 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import (
1313
"github.com/smartcontractkit/libocr/subprocesses"
1414

1515
"github.com/smartcontractkit/libocr/internal/loghelper"
16+
"github.com/smartcontractkit/libocr/internal/util"
1617
)
1718

1819
var (
@@ -22,8 +23,8 @@ var (
2223
// ocrEndpointV3 represents a member of a particular feed oracle group
2324
type ocrEndpointV3 struct {
2425
// configuration and settings
25-
defaultPriorityConfig ocr2types.BinaryNetworkEndpoint2Config
26-
lowPriorityConfig ocr2types.BinaryNetworkEndpoint2Config
26+
defaultPriorityConfig binaryNetworkEndpoint2ConfigNoNils
27+
lowPriorityConfig binaryNetworkEndpoint2ConfigNoNils
2728
peerMapping map[commontypes.OracleID]ragetypes.PeerID
2829
host *ragep2pnew.Host
2930
configDigest ocr2types.ConfigDigest
@@ -50,6 +51,21 @@ type priorityStreamGroup struct {
5051
Default ragep2pnew.Stream2
5152
}
5253

54+
// copy of ocr2types.BinaryNetworkEndpoint2Config without nils
55+
type binaryNetworkEndpoint2ConfigNoNils struct {
56+
ocr2types.BinaryNetworkEndpointLimits
57+
IncomingMessageBufferSize int
58+
OutgoingMessageBufferSize int
59+
}
60+
61+
func binaryNetworkEndpoint2ConfigNilCoalesceWithPeerConfig(config ocr2types.BinaryNetworkEndpoint2Config, peer *concretePeerV2) binaryNetworkEndpoint2ConfigNoNils {
62+
return binaryNetworkEndpoint2ConfigNoNils{
63+
config.BinaryNetworkEndpointLimits,
64+
util.NilCoalesce(config.OverrideIncomingMessageBufferSize, peer.endpointConfig.IncomingMessageBufferSize),
65+
util.NilCoalesce(config.OverrideOutgoingMessageBufferSize, peer.endpointConfig.OutgoingMessageBufferSize),
66+
}
67+
}
68+
5369
//nolint:unused
5470
func newOCREndpointV3(
5571
logger loghelper.LoggerWithContext,
@@ -94,8 +110,8 @@ func newOCREndpointV3(
94110
}
95111

96112
o := &ocrEndpointV3{
97-
defaultPriorityConfig,
98-
lowPriorityConfig,
113+
binaryNetworkEndpoint2ConfigNilCoalesceWithPeerConfig(defaultPriorityConfig, peer),
114+
binaryNetworkEndpoint2ConfigNilCoalesceWithPeerConfig(lowPriorityConfig, peer),
99115
peerMapping,
100116
host,
101117
configDigest,
@@ -141,8 +157,8 @@ func (o *ocrEndpointV3) start() error {
141157
pid,
142158
streamNameFromConfigDigestAndPriority(o.configDigest, ragep2pnew.StreamPriorityLow),
143159
ragep2pnew.StreamPriorityLow,
144-
ragep2pnew.Stream2Limits{o.lowPriorityConfig.OverrideOutgoingMessageBufferSize,
145-
o.lowPriorityConfig.OverrideIncomingMessageBufferSize,
160+
ragep2pnew.Stream2Limits{o.lowPriorityConfig.OutgoingMessageBufferSize,
161+
o.lowPriorityConfig.IncomingMessageBufferSize,
146162
o.lowPriorityConfig.MaxMessageLength,
147163
ragetypes.TokenBucketParams{
148164
o.lowPriorityConfig.MessagesRatePerOracle,
@@ -163,8 +179,8 @@ func (o *ocrEndpointV3) start() error {
163179
streamNameFromConfigDigestAndPriority(o.configDigest, ragep2pnew.StreamPriorityDefault),
164180
ragep2pnew.StreamPriorityDefault,
165181
ragep2pnew.Stream2Limits{
166-
o.defaultPriorityConfig.OverrideOutgoingMessageBufferSize,
167-
o.defaultPriorityConfig.OverrideIncomingMessageBufferSize,
182+
o.defaultPriorityConfig.OutgoingMessageBufferSize,
183+
o.defaultPriorityConfig.IncomingMessageBufferSize,
168184
o.defaultPriorityConfig.MaxMessageLength,
169185
ragetypes.TokenBucketParams{
170186
o.defaultPriorityConfig.MessagesRatePerOracle,

‎offchainreporting2plus/internal/managed/limits/ocr3_1_limits.go‎

Lines changed: 5 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -71,15 +71,15 @@ func OCR3_1Limits(cfg ocr3_1config.PublicConfig, pluginLimits ocr3_1types.Report
7171
const sigOverhead = 10
7272
const overhead = 256
7373

74-
maxLenStateTransitionOutputs := add(
74+
maxLenStateWriteSet := add(
7575
pluginLimits.MaxKeyValueModifiedKeysPlusValuesBytes,
7676
mul(
7777
pluginLimits.MaxKeyValueModifiedKeys,
7878
repeatedOverhead,
7979
),
8080
)
8181
maxLenCertifiedPrepareOrCommit := add(mul(ed25519.SignatureSize+sigOverhead, cfg.ByzQuorumSize()),
82-
sha256.Size*3,
82+
sha256.Size*5,
8383
overhead)
8484
maxLenMsgNewEpoch := overhead
8585
maxLenMsgEpochStartRequest := add(maxLenCertifiedPrepareOrCommit, overhead)
@@ -132,7 +132,7 @@ func OCR3_1Limits(cfg ocr3_1config.PublicConfig, pluginLimits ocr3_1types.Report
132132

133133
// block sync messages
134134
maxLenMsgBlockSyncRequest := overhead
135-
maxLenAttestedStateTransitionBlock := add(maxLenCertifiedPrepareOrCommit, maxLenStateTransitionOutputs)
135+
maxLenAttestedStateTransitionBlock := add(maxLenCertifiedPrepareOrCommit, maxLenStateWriteSet)
136136
maxLenMsgBlockSyncResponse := add(mul(cfg.GetMaxBlocksPerBlockSyncResponse(), maxLenAttestedStateTransitionBlock), overhead)
137137

138138
// blob exchange messages
@@ -151,13 +151,11 @@ func OCR3_1Limits(cfg ocr3_1config.PublicConfig, pluginLimits ocr3_1types.Report
151151
maxLenMsgEpochStartRequest,
152152
maxLenMsgEpochStart,
153153
maxLenMsgRoundStart,
154-
maxLenMsgObservation,
155154
maxLenMsgProposal,
156155
maxLenMsgPrepare,
157156
maxLenMsgCommit,
158157
maxLenMsgReportSignatures,
159158
maxLenMsgReportsPlusPrecursorRequest,
160-
maxLenMsgReportsPlusPrecursor,
161159
maxLenMsgBlobOffer,
162160
maxLenMsgBlobChunkRequest,
163161
)
@@ -174,15 +172,15 @@ func OCR3_1Limits(cfg ocr3_1config.PublicConfig, pluginLimits ocr3_1types.Report
174172

175173
defaultPriorityMessagesRate := (1.0*float64(time.Second)/float64(cfg.GetDeltaResend()) +
176174
3.0*float64(time.Second)/minEpochInterval +
177-
8.0*float64(time.Second)/float64(minRoundInterval) +
175+
6.0*float64(time.Second)/float64(minRoundInterval) +
178176
2.0*float64(time.Second)/float64(cfg.GetDeltaBlobOfferMinRequestToSameOracleInterval()) +
179177
1.0*float64(time.Second)/float64(cfg.GetDeltaBlobChunkMinRequestToSameOracleInterval())) * 1.2
180178

181179
lowPriorityMessagesRate := (1.0*float64(time.Second)/float64(cfg.GetDeltaBlockSyncMinRequestToSameOracleInterval()) +
182180
1.0*float64(time.Second)/float64(cfg.GetDeltaTreeSyncMinRequestToSameOracleInterval()) +
183181
1.0*float64(time.Second)/float64(cfg.GetDeltaStateSyncSummaryInterval())) * 1.2
184182

185-
defaultPriorityMessagesCapacity := mul(15, 3)
183+
defaultPriorityMessagesCapacity := mul(13, 3)
186184
lowPriorityMessagesCapacity := mul(3, 3)
187185

188186
// we don't multiply bytesRate by a safetyMargin since we already have a generous overhead on each message
@@ -196,9 +194,7 @@ func OCR3_1Limits(cfg ocr3_1config.PublicConfig, pluginLimits ocr3_1types.Report
196194
float64(time.Second)/float64(minRoundInterval)*float64(maxLenMsgRoundStart) +
197195
float64(time.Second)/float64(minRoundInterval)*float64(maxLenMsgProposal) +
198196
float64(time.Second)/float64(minEpochInterval)*float64(maxLenMsgEpochStartRequest) +
199-
float64(time.Second)/float64(minRoundInterval)*float64(maxLenMsgObservation) +
200197
float64(time.Second)/float64(minRoundInterval)*float64(maxLenMsgReportsPlusPrecursorRequest) +
201-
float64(time.Second)/float64(minRoundInterval)*float64(maxLenMsgReportsPlusPrecursor) +
202198
float64(time.Second)/float64(cfg.GetDeltaBlobOfferMinRequestToSameOracleInterval())*float64(maxLenMsgBlobOffer) + // blob-related messages
203199
float64(time.Second)/float64(cfg.GetDeltaBlobChunkMinRequestToSameOracleInterval())*float64(maxLenMsgBlobChunkRequest)
204200

@@ -212,16 +208,13 @@ func OCR3_1Limits(cfg ocr3_1config.PublicConfig, pluginLimits ocr3_1types.Report
212208
maxLenMsgEpochStartRequest,
213209
maxLenMsgEpochStart,
214210
maxLenMsgRoundStart,
215-
maxLenMsgObservation,
216211
maxLenMsgProposal,
217212
maxLenMsgPrepare,
218213
maxLenMsgCommit,
219214
maxLenMsgReportSignatures,
220215
maxLenMsgReportsPlusPrecursorRequest,
221-
maxLenMsgReportsPlusPrecursor,
222216
maxLenMsgBlobOffer,
223217
maxLenMsgBlobChunkRequest,
224-
maxLenMsgBlobOfferResponse,
225218
), 3)
226219

227220
lowPriorityBytesCapacity := mul(add(

‎offchainreporting2plus/internal/managed/managed_ocr3_1_oracle.go‎

Lines changed: 20 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -22,13 +22,6 @@ import (
2222
"github.com/smartcontractkit/libocr/subprocesses"
2323
)
2424

25-
const (
26-
defaultIncomingMessageBufferSize = 10
27-
defaultOutgoingMessageBufferSize = 10
28-
lowPriorityIncomingMessageBufferSize = 10
29-
lowPriorityOutgoingMessageBufferSize = 10
30-
)
31-
3225
// RunManagedOCR3_1Oracle runs a "managed" version of protocol.RunOracle. It handles
3326
// setting up telemetry, garbage collection, configuration updates, translating
3427
// from types.BinaryNetworkEndpoint2 to protocol.NetworkEndpoint, and
@@ -175,15 +168,16 @@ func RunManagedOCR3_1Oracle[RI any](
175168
})
176169
return fmt.Errorf("ManagedOCR3_1Oracle: error during limits"), false
177170
}
171+
178172
defaultPriorityConfig := types.BinaryNetworkEndpoint2Config{
179173
defaultLims,
180-
defaultIncomingMessageBufferSize,
181-
defaultOutgoingMessageBufferSize,
174+
nil,
175+
nil,
182176
}
183177
lowPriorityConfig := types.BinaryNetworkEndpoint2Config{
184178
lowPriorityLimits,
185-
lowPriorityIncomingMessageBufferSize,
186-
lowPriorityOutgoingMessageBufferSize,
179+
nil,
180+
nil,
187181
}
188182

189183
binNetEndpoint, err := messageNetEndpointFactory.NewEndpoint(
@@ -237,7 +231,21 @@ func RunManagedOCR3_1Oracle[RI any](
237231
logger,
238232
"ManagedOCR3_1Oracle: error during keyValueDatabase.Close()",
239233
)
240-
semanticOCR3_1KeyValueDatabase := shim.NewSemanticOCR3_1KeyValueDatabase(keyValueDatabase, reportingPluginInfo.Limits, sharedConfig.PublicConfig, logger, metricsRegisterer)
234+
keyValueDatabaseWithMetrics := shim.NewKeyValueDatabaseWithMetrics(keyValueDatabase, metricsRegisterer, logger)
235+
defer loghelper.CloseLogError(
236+
keyValueDatabaseWithMetrics,
237+
logger,
238+
"ManagedOCR3_1Oracle: error during keyValueDatabaseWithMetrics.Close()",
239+
)
240+
semanticOCR3_1KeyValueDatabase, err := shim.NewSemanticOCR3_1KeyValueDatabase(keyValueDatabaseWithMetrics, reportingPluginInfo.Limits, sharedConfig.PublicConfig, logger, metricsRegisterer)
241+
if err != nil {
242+
return fmt.Errorf("ManagedOCR3_1Oracle: error during NewSemanticOCR3_1KeyValueDatabase: %w", err), false
243+
}
244+
defer loghelper.CloseLogError(
245+
semanticOCR3_1KeyValueDatabase,
246+
logger,
247+
"ManagedOCR3_1Oracle: error during semanticOCR3_1KeyValueDatabase.Close()",
248+
)
241249

242250
protocol.RunOracle[RI](
243251
ctx,

‎offchainreporting2plus/internal/ocr3_1/maxmaxserializationlimits/max_max_serialization_limits.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ const (
55
MaxMaxEpochStartBytes = 4825
66
MaxMaxReportsPlusPrecursorRequestBytes = 18
77
MaxMaxReportsPlusPrecursorBytes = 6815772
8-
MaxMaxBlockSyncRequestBytes = 32
8+
MaxMaxBlockSyncRequestBytes = 40
99
MaxMaxBlockSyncResponseBytes = 27371305
1010
MaxMaxTreeSyncChunkRequestBytes = 106
1111
MaxMaxTreeSyncChunkResponseBytes = 65049225

‎offchainreporting2plus/internal/ocr3_1/protocol/blob_exchange.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1815,9 +1815,9 @@ func (bex *blobExchangeState[RI]) verifyCert(cert *LightCertifiedBlob) error {
18151815
}
18161816

18171817
func staleBlob(expirySeqNr uint64, blobDigest BlobDigest) StaleBlob {
1818-
return StaleBlob{expirySeqNr + 1, blobDigest}
1818+
return StaleBlob{expirySeqNr, blobDigest}
18191819
}
18201820

18211821
func hasBlobExpired(expirySeqNr uint64, committedSeqNr uint64) bool {
1822-
return expirySeqNr < committedSeqNr
1822+
return expirySeqNr <= committedSeqNr
18231823
}

‎offchainreporting2plus/internal/ocr3_1/protocol/blob_reap.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ import (
1010
)
1111

1212
const (
13-
blobReapInterval = 10 * time.Second
13+
blobReapInterval = 3 * time.Second
1414
maxBlobsToReapInSingleTransaction = 100
1515
)
1616

‎offchainreporting2plus/internal/ocr3_1/protocol/event.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,7 @@ func (ev EventComputedObservationQuorumSuccess[RI]) processOutcomeGeneration(out
7878
type EventComputedObservation[RI any] struct {
7979
Epoch uint64
8080
SeqNr uint64
81+
RequestHandle types.RequestHandle
8182
AttributedQuery types.AttributedQuery
8283
Observation types.Observation
8384
}

‎offchainreporting2plus/internal/ocr3_1/protocol/kvdb.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ type KeyValueDatabaseSemanticRead interface {
4040
toSeqNr uint64,
4141
startIndex jmt.Digest,
4242
requestEndInclIndex jmt.Digest,
43+
maxCumulativeKeysPlusValuesBytes int,
4344
) (
4445
endInclIndex jmt.Digest,
4546
boundingLeaves []jmt.BoundingLeaf,

‎offchainreporting2plus/internal/ocr3_1/protocol/message.go‎

Lines changed: 18 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -179,9 +179,10 @@ func (msg MessageEpochStart[RI]) epoch() uint64 {
179179
}
180180

181181
type MessageRoundStart[RI any] struct {
182-
Epoch uint64
183-
SeqNr uint64
184-
Query types.Query
182+
RequestHandle types.RequestHandle // actual handle for inbound message, sentinel for outbound
183+
Epoch uint64
184+
SeqNr uint64
185+
Query types.Query
185186
}
186187

187188
var _ MessageToOutcomeGeneration[struct{}] = (*MessageRoundStart[struct{}])(nil)
@@ -206,6 +207,7 @@ func (msg MessageRoundStart[RI]) epoch() uint64 {
206207
}
207208

208209
type MessageObservation[RI any] struct {
210+
RequestHandle types.RequestHandle // actual handle for outbound message, sentinel for inbound
209211
Epoch uint64
210212
SeqNr uint64
211213
SignedObservation SignedObservation
@@ -354,7 +356,8 @@ func (msg MessageReportSignatures[RI]) processReportAttestation(repatt *reportAt
354356
}
355357

356358
type MessageReportsPlusPrecursorRequest[RI any] struct {
357-
SeqNr uint64
359+
RequestHandle types.RequestHandle // actual handle for inbound message, sentinel for outbound
360+
SeqNr uint64
358361
}
359362

360363
var _ MessageToReportAttestation[struct{}] = MessageReportsPlusPrecursorRequest[struct{}]{}
@@ -372,6 +375,7 @@ func (msg MessageReportsPlusPrecursorRequest[RI]) processReportAttestation(repat
372375
}
373376

374377
type MessageReportsPlusPrecursor[RI any] struct {
378+
RequestHandle types.RequestHandle // actual handle for outbound message, sentinel for inbound
375379
SeqNr uint64
376380
ReportsPlusPrecursor ocr3_1types.ReportsPlusPrecursor
377381
}
@@ -394,10 +398,11 @@ func (msg MessageReportsPlusPrecursor[RI]) processReportAttestation(repatt *repo
394398
}
395399

396400
type MessageBlockSyncRequest[RI any] struct {
397-
RequestHandle types.RequestHandle // actual handle for outbound message, sentinel for inbound
398-
RequestInfo *types.RequestInfo
399-
StartSeqNr uint64 // a successful response must contain at least the block with this sequence number
400-
EndExclSeqNr uint64 // the response may only contain sequence numbers less than this
401+
RequestHandle types.RequestHandle // actual handle for inbound message, sentinel for outbound
402+
RequestInfo *types.RequestInfo
403+
StartSeqNr uint64 // a successful response must contain at least the block with this sequence number
404+
EndExclSeqNr uint64 // the response may only contain sequence numbers less than this
405+
MaxCumulativeWriteSetBytes int
401406
}
402407

403408
var _ MessageToStateSync[struct{}] = MessageBlockSyncRequest[struct{}]{}
@@ -463,11 +468,13 @@ func (msg MessageBlockSyncResponse[RI]) processStateSync(stasy *stateSyncState[R
463468
}
464469

465470
type MessageTreeSyncChunkRequest[RI any] struct {
466-
RequestHandle types.RequestHandle // actual handle for outbound message, sentinel for inbound
471+
RequestHandle types.RequestHandle // actual handle for inbound message, sentinel for outbound
467472
RequestInfo *types.RequestInfo
468473
ToSeqNr uint64
469474
StartIndex jmt.Digest
470475
EndInclIndex jmt.Digest
476+
477+
MaxCumulativeKeysPlusValuesBytes int
471478
}
472479

473480
var _ MessageToStateSync[struct{}] = MessageTreeSyncChunkRequest[struct{}]{}
@@ -534,7 +541,7 @@ func (msg MessageTreeSyncChunkResponse[RI]) processStateSync(stasy *stateSyncSta
534541
}
535542

536543
type MessageBlobOffer[RI any] struct {
537-
RequestHandle types.RequestHandle // actual handle for outbound message, sentinel for inbound
544+
RequestHandle types.RequestHandle // actual handle for inbound message, sentinel for outbound
538545
RequestInfo *types.RequestInfo
539546
ChunkDigestsRoot mt.Digest
540547
PayloadLength uint64
@@ -581,7 +588,7 @@ func (msg MessageBlobOfferResponse[RI]) processBlobExchange(bex *blobExchangeSta
581588
}
582589

583590
type MessageBlobChunkRequest[RI any] struct {
584-
RequestHandle types.RequestHandle // actual handle for outbound message, sentinel for inbound
591+
RequestHandle types.RequestHandle // actual handle for inbound message, sentinel for outbound
585592
RequestInfo *types.RequestInfo
586593
BlobDigest BlobDigest
587594
ChunkIndex uint64

0 commit comments

Comments
 (0)