From 750eaa3bd11c1fd638c94b78a3253760095deb3d Mon Sep 17 00:00:00 2001 From: Kostis Karantias Date: Fri, 30 Oct 2026 18:21:59 +0100 Subject: [PATCH] Improvements: - Allow OCR3.1 plugins to read replicated key-value state outside ReportingPlugin methods using ReadOnlyKeyValueState - Add per-oracle liveness metrics for OCR3 and OCR3.1 - Add determinism checking wrappers for OCR3 plugin tests in the ocr3plugintesthelpers package - Prevent repeated Start calls from tearing down running networking components - Avoid mutating Merkle subrange proofs during verification - Improve documentation for the ocr3_1types package - Fix transaction and iterator cleanup in the in-memory key-value database test helper - Reduce ragep2p warning noise when peers disconnect before sending a knock Based on 1e6df1e666a58da08e3a98c37e8c61665bc97689 Co-authored-by: kaleofduty <59616916+kaleofduty@users.noreply.github.com> Co-authored-by: stchrysa Co-authored-by: Yan Ji Co-authored-by: mvidigueira Co-authored-by: Ryan Tinianov Co-authored-by: Jordan Krage --- internal/jmt/verify.go | 27 +- networking/bootstrapper_v2.go | 9 +- networking/ocr_endpoint_v2.go | 9 +- networking/ocr_endpoint_v3.go | 9 +- networking/ragedisco/discovery_protocol.go | 9 +- networking/ragedisco/ragep2p_discoverer.go | 9 +- .../reportingplugin/median/median.go | 54 +-- .../titlerequest/titlerequest.go | 36 +- .../internal/common/list/list.go | 61 ++++ .../internal/managed/managed_ocr3_1_oracle.go | 44 ++- .../internal/ocr3/protocol/metrics.go | 57 +++- .../ocr3/protocol/outcome_generation.go | 2 +- .../protocol/outcome_generation_follower.go | 3 + .../internal/ocr3/protocol/pacemaker.go | 4 +- .../internal/ocr3_1/protocol/blob_endpoint.go | 5 + .../internal/ocr3_1/protocol/blob_exchange.go | 2 + .../internal/ocr3_1/protocol/metrics.go | 79 ++++- .../ocr3_1/protocol/outcome_generation.go | 2 +- .../protocol/outcome_generation_follower.go | 3 + .../internal/ocr3_1/protocol/pacemaker.go | 4 +- .../internal/ocr3_1/protocol/state_sync.go | 3 +- .../internal/shim/metrics.go | 121 +++++++ ...ocr3_1_context_marking_reporting_plugin.go | 94 ++++++ .../shim/ocr3_1_read_only_key_value_state.go | 314 ++++++++++++++++++ .../pebble_key_value_database.go | 4 - .../ocr3_1shims/reporting_plugin_factory.go | 35 ++ offchainreporting2plus/ocr3_1types/blob.go | 2 +- offchainreporting2plus/ocr3_1types/kvdb.go | 54 ++- offchainreporting2plus/ocr3_1types/plugin.go | 163 +++++++-- .../plugin_determinism_wrapper.go | 210 ++++++++++++ .../in_memory_key_value_database.go | 37 ++- offchainreporting2plus/oracle.go | 49 ++- ragep2p/ragep2p.go | 15 +- ragep2p/ragep2pnew/ragep2p.go | 15 +- 34 files changed, 1390 insertions(+), 154 deletions(-) create mode 100644 offchainreporting2plus/internal/common/list/list.go create mode 100644 offchainreporting2plus/internal/shim/ocr3_1_context_marking_reporting_plugin.go create mode 100644 offchainreporting2plus/internal/shim/ocr3_1_read_only_key_value_state.go create mode 100644 offchainreporting2plus/ocr3_1shims/reporting_plugin_factory.go create mode 100644 offchainreporting2plus/ocr3plugintesthelpers/plugin_determinism_wrapper.go diff --git a/internal/jmt/verify.go b/internal/jmt/verify.go index 8f62a40d..11036c31 100644 --- a/internal/jmt/verify.go +++ b/internal/jmt/verify.go @@ -227,11 +227,18 @@ func verifySubrangePadded( ) error { expectedLeftSiblingsAtDepth := make([]int, bitLength(Digest{})) expectedRightSiblingsAtDepth := make([]int, bitLength(Digest{})) + // The bounding leaves' siblings are consumed as we bubble up the layers. + // We track the remaining siblings in local variables so that we don't + // mutate the caller's bounding leaves. A nil bounding leaf yields an empty + // local slice, so below we only need to consult the local slices. + var leftSiblings, rightSiblings []Digest if leftBoundingLeafOrNil != nil { expectedLeftSiblingsAtDepth = cumulativeOccurrencesOfBit(leftBoundingLeafOrNil.Leaf.KeyDigest, true) + leftSiblings = leftBoundingLeafOrNil.Siblings } if rightBoundingLeafOrNil != nil { expectedRightSiblingsAtDepth = cumulativeOccurrencesOfBit(rightBoundingLeafOrNil.Leaf.KeyDigest, false) + rightSiblings = rightBoundingLeafOrNil.Siblings } layer := bottommostLayer @@ -253,10 +260,10 @@ func verifySubrangePadded( // Is this the rightmost node in layer? if i+1 == len(layer) { - if rightBoundingLeafOrNil != nil && len(rightBoundingLeafOrNil.Siblings) > 0 && expectedRightSiblingsAtDepth[depth] == len(rightBoundingLeafOrNil.Siblings) { + if len(rightSiblings) > 0 && expectedRightSiblingsAtDepth[depth] == len(rightSiblings) { // Either our right siblings cover this level. - rightSibling = sparseMerkleNode{false, rightBoundingLeafOrNil.Siblings[0]} - rightBoundingLeafOrNil.Siblings = rightBoundingLeafOrNil.Siblings[1:] + rightSibling = sparseMerkleNode{false, rightSiblings[0]} + rightSiblings = rightSiblings[1:] } else { // Or we put a placeholder node. rightSibling = sparseMerkleNode{false, SparseMerklePlaceholderDigest} @@ -278,10 +285,10 @@ func verifySubrangePadded( // Is this the leftmost node in layer? if i == 0 { - if leftBoundingLeafOrNil != nil && len(leftBoundingLeafOrNil.Siblings) > 0 && expectedLeftSiblingsAtDepth[depth] == len(leftBoundingLeafOrNil.Siblings) { + if len(leftSiblings) > 0 && expectedLeftSiblingsAtDepth[depth] == len(leftSiblings) { // Either our left siblings cover this level. - leftSibling = sparseMerkleNode{false, leftBoundingLeafOrNil.Siblings[0]} - leftBoundingLeafOrNil.Siblings = leftBoundingLeafOrNil.Siblings[1:] + leftSibling = sparseMerkleNode{false, leftSiblings[0]} + leftSiblings = leftSiblings[1:] } else { // Or we put a placeholder node. leftSibling = sparseMerkleNode{false, SparseMerklePlaceholderDigest} @@ -310,11 +317,11 @@ func verifySubrangePadded( if len(layer) != 1 { return fmt.Errorf("unexpectedly ended with a non-singleton layer of %v nodes at the top", len(layer)) } - if leftBoundingLeafOrNil != nil && len(leftBoundingLeafOrNil.Siblings) != 0 { - return fmt.Errorf("unexpectedly ended with left siblings remaining: %v", len(leftBoundingLeafOrNil.Siblings)) + if len(leftSiblings) != 0 { + return fmt.Errorf("unexpectedly ended with left siblings remaining: %v", len(leftSiblings)) } - if rightBoundingLeafOrNil != nil && len(rightBoundingLeafOrNil.Siblings) != 0 { - return fmt.Errorf("unexpectedly ended with right siblings remaining: %v", len(rightBoundingLeafOrNil.Siblings)) + if len(rightSiblings) != 0 { + return fmt.Errorf("unexpectedly ended with right siblings remaining: %v", len(rightSiblings)) } if layer[0].sparseMerkleNode.digest != expectedRootDigest { return fmt.Errorf("computed root digest mismatch: computed %x, expected %x", layer[0].sparseMerkleNode.digest, expectedRootDigest) diff --git a/networking/bootstrapper_v2.go b/networking/bootstrapper_v2.go index 75df7ac3..acdd145e 100644 --- a/networking/bootstrapper_v2.go +++ b/networking/bootstrapper_v2.go @@ -61,9 +61,11 @@ func newBootstrapperV2( } func (b *bootstrapperV2) Start() error { - succeeded := false + // Armed only once we've transitioned to started, so that a rejected Start() + // never tears down a bootstrapperV2 that someone else started. + needsTeardown := false defer func() { - if !succeeded { + if needsTeardown { b.Close() } }() @@ -76,9 +78,10 @@ func (b *bootstrapperV2) Start() error { } b.state = bootstrapperStarted + needsTeardown = true b.logger.Info("BootstrapperV2: Started listening", nil) - succeeded = true + needsTeardown = false return nil } diff --git a/networking/ocr_endpoint_v2.go b/networking/ocr_endpoint_v2.go index 54dbf71e..d19e51f2 100644 --- a/networking/ocr_endpoint_v2.go +++ b/networking/ocr_endpoint_v2.go @@ -148,9 +148,11 @@ func streamNameFromConfigDigest(cd ocr2types.ConfigDigest) string { // Start the ocrEndpointV2. Should only be called once. func (o *ocrEndpointV2) Start() error { - succeeded := false + // Armed only once we've transitioned to started, so that a rejected Start() + // never tears down an ocrEndpointV2 that someone else started. + needsTeardown := false defer func() { - if !succeeded { + if needsTeardown { o.Close() } }() @@ -162,6 +164,7 @@ func (o *ocrEndpointV2) Start() error { return fmt.Errorf("cannot start ocrEndpointV2 that is not unstarted, state was: %d", o.state) } o.state = ocrEndpointStarted + needsTeardown = true for oid, pid := range o.peerMapping { if oid == o.ownOracleID { @@ -199,7 +202,7 @@ func (o *ocrEndpointV2) Start() error { }) o.logger.Info("OCREndpointV2: Started listening", nil) - succeeded = true + needsTeardown = false return nil } diff --git a/networking/ocr_endpoint_v3.go b/networking/ocr_endpoint_v3.go index a8b6f890..11a51679 100644 --- a/networking/ocr_endpoint_v3.go +++ b/networking/ocr_endpoint_v3.go @@ -132,9 +132,11 @@ func newOCREndpointV3( // Start the ocrEndpointV3. Called once at the end of the initialization code. func (o *ocrEndpointV3) start() error { - succeeded := false + // Armed only once we've transitioned to started, so that a rejected start() + // never tears down an ocrEndpointV3 that someone else started. + needsTeardown := false defer func() { - if !succeeded { + if needsTeardown { o.Close() } }() @@ -146,6 +148,7 @@ func (o *ocrEndpointV3) start() error { return fmt.Errorf("cannot start ocrEndpointV3 that is not unstarted, state was: %d", o.state) } o.state = ocrEndpointStarted + needsTeardown = true for oid, pid := range o.peerMapping { if oid == o.ownOracleID { @@ -209,7 +212,7 @@ func (o *ocrEndpointV3) start() error { }) o.logger.Info("OCREndpointV3: Started listening", nil) - succeeded = true + needsTeardown = false return nil } diff --git a/networking/ragedisco/discovery_protocol.go b/networking/ragedisco/discovery_protocol.go index 99b4798d..f065c420 100644 --- a/networking/ragedisco/discovery_protocol.go +++ b/networking/ragedisco/discovery_protocol.go @@ -129,9 +129,11 @@ func newDiscoveryProtocol( } func (p *discoveryProtocol) Start() error { - succeeded := false + // Armed only once we've transitioned to started, so that a rejected Start() + // never tears down a discoveryProtocol that someone else started. + needsTeardown := false defer func() { - if !succeeded { + if needsTeardown { p.Close() } }() @@ -142,6 +144,7 @@ func (p *discoveryProtocol) Start() error { return fmt.Errorf("cannot start discoveryProtocol that is not unstarted, state was: %v", p.state) } p.state = discoveryProtocolStarted + needsTeardown = true p.lock.Lock() defer p.lock.Unlock() @@ -153,7 +156,7 @@ func (p *discoveryProtocol) Start() error { p.processes.Go(p.sendLoop) p.processes.Go(p.saveLoop) p.processes.Go(p.statusReportLoop) - succeeded = true + needsTeardown = false return nil } diff --git a/networking/ragedisco/ragep2p_discoverer.go b/networking/ragedisco/ragep2p_discoverer.go index 6c8ad2e5..3b1a8d4c 100644 --- a/networking/ragedisco/ragep2p_discoverer.go +++ b/networking/ragedisco/ragep2p_discoverer.go @@ -79,9 +79,11 @@ func NewRagep2pDiscoverer( } func (r *Ragep2pDiscoverer) Start(host ragep2pwrapper.Host, keyring ragetypes.PeerKeyring, logger loghelper.LoggerWithContext) error { - succeeded := false + // Armed only once we've transitioned to started, so that a rejected Start() + // never tears down a Ragep2pDiscoverer that someone else started. + needsTeardown := false defer func() { - if !succeeded { + if needsTeardown { r.Close() } }() @@ -93,6 +95,7 @@ func (r *Ragep2pDiscoverer) Start(host ragep2pwrapper.Host, keyring ragetypes.Pe return fmt.Errorf("cannot start Ragep2pDiscoverer that is not unstarted, state was: %v", r.state) } r.state = ragep2pDiscovererStarted + needsTeardown = true r.host = host announceAddresses, ok := combinedAnnounceAddrsForDiscoverer(r.logger, r.announceAddresses) if !ok { @@ -120,7 +123,7 @@ func (r *Ragep2pDiscoverer) Start(host ragep2pwrapper.Host, keyring ragetypes.Pe r.proc.Go(r.connectivityLoop) r.proc.Go(r.writeLoop) - succeeded = true + needsTeardown = false return nil } diff --git a/offchainreporting2/reportingplugin/median/median.go b/offchainreporting2/reportingplugin/median/median.go index 473f00a6..43abf94f 100644 --- a/offchainreporting2/reportingplugin/median/median.go +++ b/offchainreporting2/reportingplugin/median/median.go @@ -347,34 +347,34 @@ func (fac NumericalMedianFactory) NewReportingPlugin(ctx context.Context, config }) return &numericalMedian{ - offchainConfig, - onchainConfig, - fac.ContractTransmitter, - fac.DataSource, - fac.JuelsPerFeeCoinDataSource, - fac.GasPriceSubunitsDataSource, - fac.IncludeGasPriceSubunitsInObservation, - logger, - fac.ReportCodec, - deviationFunc, - fac.AcceptAfterFullTransmissionScheduleElapsed || offchainConfig.AcceptAfterFullTransmissionScheduleElapsed, - - configuration.ConfigDigest, - configuration.F, - configuration.DurationAllTransmissionStages, - epochRound{}, - new(big.Int), - time.Now(), + offchainConfig, + onchainConfig, + fac.ContractTransmitter, + fac.DataSource, + fac.JuelsPerFeeCoinDataSource, + fac.GasPriceSubunitsDataSource, + fac.IncludeGasPriceSubunitsInObservation, + logger, + fac.ReportCodec, + deviationFunc, + fac.AcceptAfterFullTransmissionScheduleElapsed || offchainConfig.AcceptAfterFullTransmissionScheduleElapsed, + + configuration.ConfigDigest, + configuration.F, + configuration.DurationAllTransmissionStages, + epochRound{}, + new(big.Int), + time.Now(), + maxReportLength, + }, types.ReportingPluginInfo{ + "NumericalMedian", + false, + types.ReportingPluginLimits{ + 0, + maxObservationLength, maxReportLength, - }, types.ReportingPluginInfo{ - "NumericalMedian", - false, - types.ReportingPluginLimits{ - 0, - maxObservationLength, - maxReportLength, - }, - }, nil + }, + }, nil } func DefaultDeviationFunc(_ context.Context, thresholdPPB uint64, old *big.Int, new *big.Int) (bool, error) { diff --git a/offchainreporting2/reportingplugin/titlerequest/titlerequest.go b/offchainreporting2/reportingplugin/titlerequest/titlerequest.go index d85799fe..967285a4 100644 --- a/offchainreporting2/reportingplugin/titlerequest/titlerequest.go +++ b/offchainreporting2/reportingplugin/titlerequest/titlerequest.go @@ -40,24 +40,24 @@ type TitleRequestPluginFactory struct { func (fac *TitleRequestPluginFactory) NewReportingPlugin(_ context.Context, config types.ReportingPluginConfig) (types.ReportingPlugin, types.ReportingPluginInfo, error) { return &TitleRequestPlugin{ - config.F, - fac.Client, - fac.Contract, - map[[32]byte]bool{}, - map[[32]byte]time.Time{}, - }, types.ReportingPluginInfo{ - "Title Request ReportingPlugin", - false, - types.ReportingPluginLimits{ - // queries are empty - 0, - // observations are at most 32 (request id) + 32 (title offset) + 32 - // (title len) + maxTitleLen chars, let's generously round to 1000 - 1_000, - // reports follow the same format as observations - 1_000, - }, - }, nil + config.F, + fac.Client, + fac.Contract, + map[[32]byte]bool{}, + map[[32]byte]time.Time{}, + }, types.ReportingPluginInfo{ + "Title Request ReportingPlugin", + false, + types.ReportingPluginLimits{ + // queries are empty + 0, + // observations are at most 32 (request id) + 32 (title offset) + 32 + // (title len) + maxTitleLen chars, let's generously round to 1000 + 1_000, + // reports follow the same format as observations + 1_000, + }, + }, nil } var _ types.ReportingPlugin = (*TitleRequestPlugin)(nil) diff --git a/offchainreporting2plus/internal/common/list/list.go b/offchainreporting2plus/internal/common/list/list.go new file mode 100644 index 00000000..1aefbed5 --- /dev/null +++ b/offchainreporting2plus/internal/common/list/list.go @@ -0,0 +1,61 @@ +package list + +import ( + stdlist "container/list" + "iter" +) + +// List is a type-safe wrapper around the standard library's container/list, +// scoped to what we need: insertion at the back, removal of an arbitrary +// element, and access to the front. All operations are O(1). +type List[T any] struct { + internal stdlist.List +} + +func NewList[T any]() *List[T] { + return &List[T]{} +} + +// Element is a handle to an item in a List, needed to Remove it. +type Element[T any] struct { + internal *stdlist.Element +} + +func (e Element[T]) Value() T { + return e.internal.Value.(T) +} + +func (l *List[T]) PushBack(item T) Element[T] { + return Element[T]{l.internal.PushBack(item)} +} + +// Remove removes e from the list. Removing an element that has already been +// removed is a no-op. +func (l *List[T]) Remove(e Element[T]) { + l.internal.Remove(e.internal) +} + +func (l *List[T]) Front() (T, bool) { + if e := l.internal.Front(); e != nil { + return e.Value.(T), true + } else { + var zero T + return zero, false + } +} + +func (l *List[T]) Len() int { + return l.internal.Len() +} + +// All iterates from front to back. The list must not be modified during +// iteration. +func (l *List[T]) All() iter.Seq[T] { + return func(yield func(T) bool) { + for e := l.internal.Front(); e != nil; e = e.Next() { + if !yield(e.Value.(T)) { + return + } + } + } +} diff --git a/offchainreporting2plus/internal/managed/managed_ocr3_1_oracle.go b/offchainreporting2plus/internal/managed/managed_ocr3_1_oracle.go index 6d3aa6f7..8a0591a0 100644 --- a/offchainreporting2plus/internal/managed/managed_ocr3_1_oracle.go +++ b/offchainreporting2plus/internal/managed/managed_ocr3_1_oracle.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "sync" "github.com/smartcontractkit/libocr/offchainreporting2plus/internal/common" @@ -44,7 +45,7 @@ func RunManagedOCR3_1Oracle[RI any]( offchainConfigDigester types.OffchainConfigDigester, offchainKeyring types.OffchainKeyring, onchainKeyring ocr3types.OnchainKeyring2[RI], - reportingPluginFactory ocr3_1types.ReportingPluginFactory[RI], + reportingPluginFactory ocr3_1types.ReportingPluginFactory2[RI], ) { subs := subprocesses.Subprocesses{} defer subs.Wait() @@ -106,6 +107,12 @@ func RunManagedOCR3_1Oracle[RI any]( }) blobEndpointWrapper := protocol.BlobEndpointWrapper{} + readOnlyKeyValueStateWrapper := shim.NewReadOnlyKeyValueStateWrapper(childLogger, registerer) + // Close releases the goroutine and metrics the wrapper acquired on + // construction, on the early returns below as much as on the happy path. + // Disabling it, which has to happen between closing the plugin and + // closing the database, is deferred separately further down. + defer readOnlyKeyValueStateWrapper.Close() maxDurationInitialization := sharedConfig.MaxDurationInitialization initCtx, initCancel := context.WithTimeout(ctx, maxDurationInitialization) @@ -132,18 +139,26 @@ func RunManagedOCR3_1Oracle[RI any]( sharedConfig.WarnDurationObservation, sharedConfig.MaxDurationShouldAcceptAttestedReport, sharedConfig.MaxDurationShouldTransmitAcceptedReport, - }, &blobEndpointWrapper) + }, &blobEndpointWrapper, readOnlyKeyValueStateWrapper) ins.Stop() if err != nil { return fmt.Errorf("ManagedOCR3_1Oracle: error during NewReportingPlugin(): %w", err), true } - defer loghelper.CloseLogError( - reportingPlugin, - logger, - "ManagedOCR3_1Oracle: error during reportingPlugin.Close()", - ) + // Deferred twice: here as a safety net for the early returns below, + // and again further down so that on the happy path the plugin closes + // before the key value database does. The Once prevents a second + // Close, which per the docs may return an error and cause a spurious + // warning log. + closeLogErrorReportingPluginOnce := sync.OnceFunc(func() { + loghelper.CloseLogError( + reportingPlugin, + logger, + "ManagedOCR3_1Oracle: error during reportingPlugin.Close()", + ) + }) + defer closeLogErrorReportingPluginOnce() var reportingPluginInfo ocr3_1types.ReportingPluginInfo1 switch rpi := reportingPluginInfo_.(type) { @@ -270,6 +285,16 @@ func RunManagedOCR3_1Oracle[RI any]( "ManagedOCR3_1Oracle: error during semanticOCR3_1KeyValueDatabase.Close()", ) + // Shutdown order, by virtue of the defers below running in reverse: + // first the reporting plugin is closed, which is where a well-behaved + // plugin stops its goroutines and discards its read transactions; + // then the key value state is disabled, which logs and discards any + // read transactions a misbehaving plugin leaked; and only then, via + // the earlier defers, is the database closed. + readOnlyKeyValueStateWrapper.Enable(semanticOCR3_1KeyValueDatabase) + defer readOnlyKeyValueStateWrapper.Disable() + defer closeLogErrorReportingPluginOnce() + protocol.RunOracle[RI]( ctx, &blobEndpointWrapper, @@ -285,7 +310,10 @@ func RunManagedOCR3_1Oracle[RI any]( netEndpoint, offchainKeyring, onchainKeyring, - shim.LimitCheckOCR3_1ReportingPlugin[RI]{reportingPlugin, reportingPluginInfo.Limits}, + shim.LimitCheckOCR3_1ReportingPlugin[RI]{ + shim.ContextMarkingOCR3_1ReportingPlugin[RI]{reportingPlugin}, + reportingPluginInfo.Limits, + }, shim.NewOCR3_1TelemetrySender(chTelemetrySend, childLogger, localConfig.EnableTransmissionTelemetry), ) diff --git a/offchainreporting2plus/internal/ocr3/protocol/metrics.go b/offchainreporting2plus/internal/ocr3/protocol/metrics.go index 3a0d7588..b4fdcc0a 100644 --- a/offchainreporting2plus/internal/ocr3/protocol/metrics.go +++ b/offchainreporting2plus/internal/ocr3/protocol/metrics.go @@ -1,19 +1,22 @@ package protocol import ( + "fmt" + "github.com/prometheus/client_golang/prometheus" "github.com/smartcontractkit/libocr/commontypes" "github.com/smartcontractkit/libocr/internal/metricshelper" ) type pacemakerMetrics struct { - registerer prometheus.Registerer - epoch prometheus.Gauge - leader prometheus.Gauge + registerer prometheus.Registerer + epoch prometheus.Gauge + leader prometheus.Gauge + receivedNewEpochWishesTotal []prometheus.Counter } func newPacemakerMetrics(registerer prometheus.Registerer, - logger commontypes.Logger) *pacemakerMetrics { + logger commontypes.Logger, n int) *pacemakerMetrics { epoch := prometheus.NewGauge(prometheus.GaugeOpts{ Name: "ocr3_epoch", @@ -27,28 +30,45 @@ func newPacemakerMetrics(registerer prometheus.Registerer, }) metricshelper.RegisterOrLogError(logger, registerer, leader, "ocr3_experimental_leader_oid") + receivedNewEpochWishesTotal := make([]prometheus.Counter, n) + for i := range n { + receivedNewEpochWishesTotal[i] = prometheus.NewCounter(prometheus.CounterOpts{ + Name: "ocr3_received_new_epoch_wishes_total", + Help: "The total number of NewEpochWish messages received from the sender oracle. Note that " + + "oracles periodically resend their latest NewEpochWish every DeltaResend. This metric is " + + "useful for checking whether the sender oracle is up and participating in the protocol.", + ConstLabels: prometheus.Labels{"sender_oracle_id": fmt.Sprint(i)}, + }) + metricshelper.RegisterOrLogError(logger, registerer, receivedNewEpochWishesTotal[i], "ocr3_received_new_epoch_wishes_total") + } + return &pacemakerMetrics{ registerer, epoch, leader, + receivedNewEpochWishesTotal, } } func (pm *pacemakerMetrics) Close() { pm.registerer.Unregister(pm.epoch) pm.registerer.Unregister(pm.leader) + for _, c := range pm.receivedNewEpochWishesTotal { + pm.registerer.Unregister(c) + } } type outcomeGenerationMetrics struct { - registerer prometheus.Registerer - committedSeqNr prometheus.Gauge - sentObservationsTotal prometheus.Counter - includedObservationsTotal prometheus.Counter - ledCommittedRoundsTotal prometheus.Counter + registerer prometheus.Registerer + committedSeqNr prometheus.Gauge + sentObservationsTotal prometheus.Counter + includedObservationsTotal prometheus.Counter + ledCommittedRoundsTotal prometheus.Counter + theirIncludedObservationsTotal []prometheus.Counter } func newOutcomeGenerationMetrics(registerer prometheus.Registerer, - logger commontypes.Logger) *outcomeGenerationMetrics { + logger commontypes.Logger, n int) *outcomeGenerationMetrics { committedSeqNr := prometheus.NewGauge(prometheus.GaugeOpts{ Name: "ocr3_committed_sequence_number", @@ -83,12 +103,26 @@ func newOutcomeGenerationMetrics(registerer prometheus.Registerer, }) metricshelper.RegisterOrLogError(logger, registerer, ledCommittedRoundsTotal, "ocr3_led_committed_rounds_total") + theirIncludedObservationsTotal := make([]prometheus.Counter, n) + for i := range n { + theirIncludedObservationsTotal[i] = prometheus.NewCounter(prometheus.CounterOpts{ + Name: "ocr3_their_included_observations_total", + Help: "The total number of (valid) observations by the observer oracle included in a proposal " + + "from the leader. Note that there is no guarantee that the proposal will actually get " + + "committed. This metric is useful for checking whether the observer oracle is able to " + + "regularly make observations that are included in proposals.", + ConstLabels: prometheus.Labels{"observer_oracle_id": fmt.Sprint(i)}, + }) + metricshelper.RegisterOrLogError(logger, registerer, theirIncludedObservationsTotal[i], "ocr3_their_included_observations_total") + } + return &outcomeGenerationMetrics{ registerer, committedSeqNr, sentObservationsTotal, includedObservationsTotal, ledCommittedRoundsTotal, + theirIncludedObservationsTotal, } } @@ -97,4 +131,7 @@ func (om *outcomeGenerationMetrics) Close() { om.registerer.Unregister(om.sentObservationsTotal) om.registerer.Unregister(om.includedObservationsTotal) om.registerer.Unregister(om.ledCommittedRoundsTotal) + for _, c := range om.theirIncludedObservationsTotal { + om.registerer.Unregister(c) + } } diff --git a/offchainreporting2plus/internal/ocr3/protocol/outcome_generation.go b/offchainreporting2plus/internal/ocr3/protocol/outcome_generation.go index 84be6773..69c0f02d 100644 --- a/offchainreporting2plus/internal/ocr3/protocol/outcome_generation.go +++ b/offchainreporting2plus/internal/ocr3/protocol/outcome_generation.go @@ -60,7 +60,7 @@ func RunOutcomeGeneration[RI any]( id: id, localConfig: localConfig, logger: logger.MakeUpdated(commontypes.LogFields{"proto": "outgen"}), - metrics: newOutcomeGenerationMetrics(metricsRegisterer, logger), + metrics: newOutcomeGenerationMetrics(metricsRegisterer, logger, config.N()), netSender: netSender, offchainKeyring: offchainKeyring, reportingPlugin: reportingPlugin, diff --git a/offchainreporting2plus/internal/ocr3/protocol/outcome_generation_follower.go b/offchainreporting2plus/internal/ocr3/protocol/outcome_generation_follower.go index 21094f9c..5f762790 100644 --- a/offchainreporting2plus/internal/ocr3/protocol/outcome_generation_follower.go +++ b/offchainreporting2plus/internal/ocr3/protocol/outcome_generation_follower.go @@ -492,6 +492,9 @@ func (outgen *outcomeGenerationState[RI]) backgroundProposalOutcome( if seen[outgen.id] { outgen.metrics.includedObservationsTotal.Inc() } + for observer := range seen { + outgen.metrics.theirIncludedObservationsTotal[observer].Inc() + } } outcomeInputsDigest := MakeOutcomeInputsDigest( diff --git a/offchainreporting2plus/internal/ocr3/protocol/pacemaker.go b/offchainreporting2plus/internal/ocr3/protocol/pacemaker.go index bdbc1e85..ffb9779b 100644 --- a/offchainreporting2plus/internal/ocr3/protocol/pacemaker.go +++ b/offchainreporting2plus/internal/ocr3/protocol/pacemaker.go @@ -70,7 +70,7 @@ func makePacemakerState[RI any]( id: id, localConfig: localConfig, logger: logger.MakeUpdated(commontypes.LogFields{"proto": "pacemaker"}), - metrics: newPacemakerMetrics(metricsRegisterer, logger), + metrics: newPacemakerMetrics(metricsRegisterer, logger, config.N()), netSender: netSender, offchainKeyring: offchainKeyring, telemetrySender: telemetrySender, @@ -231,6 +231,8 @@ func (pace *pacemakerState[RI]) eventNewEpochRequest() { } func (pace *pacemakerState[RI]) messageNewEpochWish(msg MessageNewEpochWish[RI], sender commontypes.OracleID) { + pace.metrics.receivedNewEpochWishesTotal[sender].Inc() + if pace.newEpochWishes[sender] < msg.Epoch { pace.newEpochWishes[sender] = msg.Epoch } diff --git a/offchainreporting2plus/internal/ocr3_1/protocol/blob_endpoint.go b/offchainreporting2plus/internal/ocr3_1/protocol/blob_endpoint.go index a38ec59a..cc1d7c94 100644 --- a/offchainreporting2plus/internal/ocr3_1/protocol/blob_endpoint.go +++ b/offchainreporting2plus/internal/ocr3_1/protocol/blob_endpoint.go @@ -182,6 +182,11 @@ func (r *RoundBlobBroadcastFetcher) FetchBlob(ctx context.Context, handle ocr3_1 blobHandleSumType := blobtypes.ExtractBlobHandleSumType(handle) switch cert := blobHandleSumType.(type) { case *blobtypes.LightCertifiedBlob: + // This expiry comparison differs with blobExchangeState.hasBlobExpired + // (< here vs <= there) because the latter checks against an + // already-committed sequence number, while here we check against the + // sequence number that we might commit at the end of the round, but + // which we have not already committed. if cert != nil && cert.ExpirySeqNr < r.seqNr { return nil, fmt.Errorf("blob expired") } diff --git a/offchainreporting2plus/internal/ocr3_1/protocol/blob_exchange.go b/offchainreporting2plus/internal/ocr3_1/protocol/blob_exchange.go index b02062ff..9fc33216 100644 --- a/offchainreporting2plus/internal/ocr3_1/protocol/blob_exchange.go +++ b/offchainreporting2plus/internal/ocr3_1/protocol/blob_exchange.go @@ -682,6 +682,8 @@ func (bex *blobExchangeState[RI]) eventHighestCommittedSeqNr(highestCommittedSeq } func (bex *blobExchangeState[RI]) hasBlobExpired(expirySeqNr uint64) bool { + // Expiration semantics: Once the blob's ExpirySeqNr is committed, oracles + // are not obliged to carry the blob anymore. return expirySeqNr <= bex.highestCommittedSeqNr } diff --git a/offchainreporting2plus/internal/ocr3_1/protocol/metrics.go b/offchainreporting2plus/internal/ocr3_1/protocol/metrics.go index 6b2b1c72..20ef5bbd 100644 --- a/offchainreporting2plus/internal/ocr3_1/protocol/metrics.go +++ b/offchainreporting2plus/internal/ocr3_1/protocol/metrics.go @@ -39,14 +39,15 @@ func newAttestedStateTransitionBlocksWrittenTotal( } type pacemakerMetrics struct { - registerer prometheus.Registerer - epoch prometheus.Gauge - leader prometheus.Gauge - tProgressTimeoutsTotal prometheus.Counter + registerer prometheus.Registerer + epoch prometheus.Gauge + leader prometheus.Gauge + tProgressTimeoutsTotal prometheus.Counter + receivedNewEpochWishesTotal []prometheus.Counter } func newPacemakerMetrics(registerer prometheus.Registerer, - logger commontypes.Logger) *pacemakerMetrics { + logger commontypes.Logger, n int) *pacemakerMetrics { epoch := prometheus.NewGauge(prometheus.GaugeOpts{ Name: "ocr3_1_epoch", @@ -66,11 +67,24 @@ func newPacemakerMetrics(registerer prometheus.Registerer, }) metricshelper.RegisterOrLogError(logger, registerer, tProgressTimeoutsTotal, "ocr3_1_experimental_t_progress_timeouts_total") + receivedNewEpochWishesTotal := make([]prometheus.Counter, n) + for i := range n { + receivedNewEpochWishesTotal[i] = prometheus.NewCounter(prometheus.CounterOpts{ + Name: "ocr3_1_received_new_epoch_wishes_total", + Help: "The total number of NewEpochWish messages received from the sender oracle. Note that " + + "oracles periodically resend their latest NewEpochWish every DeltaResend. This metric is " + + "useful for checking whether the sender oracle is up and participating in the protocol.", + ConstLabels: prometheus.Labels{"sender_oracle_id": fmt.Sprint(i)}, + }) + metricshelper.RegisterOrLogError(logger, registerer, receivedNewEpochWishesTotal[i], "ocr3_1_received_new_epoch_wishes_total") + } + return &pacemakerMetrics{ registerer, epoch, leader, tProgressTimeoutsTotal, + receivedNewEpochWishesTotal, } } @@ -78,19 +92,23 @@ func (pm *pacemakerMetrics) Close() { pm.registerer.Unregister(pm.epoch) pm.registerer.Unregister(pm.leader) pm.registerer.Unregister(pm.tProgressTimeoutsTotal) + for _, c := range pm.receivedNewEpochWishesTotal { + pm.registerer.Unregister(c) + } } type outcomeGenerationMetrics struct { - registerer prometheus.Registerer - committedSeqNr prometheus.Gauge - sentObservationsTotal prometheus.Counter - includedObservationsTotal prometheus.Counter - ledCommittedRoundsTotal prometheus.Counter - attestedBlocksWrittenTotal prometheus.Counter + registerer prometheus.Registerer + committedSeqNr prometheus.Gauge + sentObservationsTotal prometheus.Counter + includedObservationsTotal prometheus.Counter + ledCommittedRoundsTotal prometheus.Counter + attestedBlocksWrittenTotal prometheus.Counter + theirIncludedObservationsTotal []prometheus.Counter } func newOutcomeGenerationMetrics(registerer prometheus.Registerer, - logger commontypes.Logger) *outcomeGenerationMetrics { + logger commontypes.Logger, n int) *outcomeGenerationMetrics { committedSeqNr := prometheus.NewGauge(prometheus.GaugeOpts{ Name: "ocr3_1_committed_sequence_number", @@ -127,6 +145,19 @@ func newOutcomeGenerationMetrics(registerer prometheus.Registerer, attestedBlocksWrittenTotal := newAttestedStateTransitionBlocksWrittenTotal(registerer, logger, attestedStateTransitionBlockWriterOutcomeGeneration) + theirIncludedObservationsTotal := make([]prometheus.Counter, n) + for i := range n { + theirIncludedObservationsTotal[i] = prometheus.NewCounter(prometheus.CounterOpts{ + Name: "ocr3_1_their_included_observations_total", + Help: "The total number of (valid) observations by the observer oracle included in a proposal " + + "from the leader. Note that there is no guarantee that the proposal will actually get " + + "committed. This metric is useful for checking whether the observer oracle is able to " + + "regularly make observations that are included in proposals.", + ConstLabels: prometheus.Labels{"observer_oracle_id": fmt.Sprint(i)}, + }) + metricshelper.RegisterOrLogError(logger, registerer, theirIncludedObservationsTotal[i], "ocr3_1_their_included_observations_total") + } + return &outcomeGenerationMetrics{ registerer, committedSeqNr, @@ -134,6 +165,7 @@ func newOutcomeGenerationMetrics(registerer prometheus.Registerer, includedObservationsTotal, ledCommittedRoundsTotal, attestedBlocksWrittenTotal, + theirIncludedObservationsTotal, } } @@ -143,6 +175,9 @@ func (om *outcomeGenerationMetrics) Close() { om.registerer.Unregister(om.includedObservationsTotal) om.registerer.Unregister(om.ledCommittedRoundsTotal) om.registerer.Unregister(om.attestedBlocksWrittenTotal) + for _, c := range om.theirIncludedObservationsTotal { + om.registerer.Unregister(c) + } } type stateSyncMetrics struct { @@ -150,10 +185,11 @@ type stateSyncMetrics struct { attestedBlocksWrittenTotal prometheus.Counter attestedBlocksReplayedTotal prometheus.Counter treeSyncCompletedTotal prometheus.Counter + theirCommittedSeqNr []prometheus.Gauge } func newStateSyncMetrics(registerer prometheus.Registerer, - logger commontypes.Logger) *stateSyncMetrics { + logger commontypes.Logger, n int) *stateSyncMetrics { attestedBlocksWrittenTotal := newAttestedStateTransitionBlocksWrittenTotal(registerer, logger, attestedStateTransitionBlockWriterStateSync) @@ -169,11 +205,25 @@ func newStateSyncMetrics(registerer prometheus.Registerer, }) metricshelper.RegisterOrLogError(logger, registerer, treeSyncCompletedTotal, "ocr3_1_experimental_tree_sync_completed_total") + theirCommittedSeqNr := make([]prometheus.Gauge, n) + for i := range n { + theirCommittedSeqNr[i] = prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "ocr3_1_their_committed_sequence_number", + Help: "The highest committed sequence number reported by the sender oracle in its latest " + + "state sync summary. Note that state sync summaries are not authenticated, so a faulty " + + "oracle may report arbitrary values. This metric is useful in comparison with " + + "ocr3_1_committed_sequence_number to check whether the sender oracle is keeping up.", + ConstLabels: prometheus.Labels{"sender_oracle_id": fmt.Sprint(i)}, + }) + metricshelper.RegisterOrLogError(logger, registerer, theirCommittedSeqNr[i], "ocr3_1_their_committed_sequence_number") + } + return &stateSyncMetrics{ registerer, attestedBlocksWrittenTotal, attestedBlocksReplayedTotal, treeSyncCompletedTotal, + theirCommittedSeqNr, } } @@ -181,6 +231,9 @@ func (sm *stateSyncMetrics) Close() { sm.registerer.Unregister(sm.attestedBlocksWrittenTotal) sm.registerer.Unregister(sm.attestedBlocksReplayedTotal) sm.registerer.Unregister(sm.treeSyncCompletedTotal) + for _, g := range sm.theirCommittedSeqNr { + sm.registerer.Unregister(g) + } } type blobExchangeMetrics struct { diff --git a/offchainreporting2plus/internal/ocr3_1/protocol/outcome_generation.go b/offchainreporting2plus/internal/ocr3_1/protocol/outcome_generation.go index be6f99bc..230db35c 100644 --- a/offchainreporting2plus/internal/ocr3_1/protocol/outcome_generation.go +++ b/offchainreporting2plus/internal/ocr3_1/protocol/outcome_generation.go @@ -68,7 +68,7 @@ func RunOutcomeGeneration[RI any]( kvDb: kvDb, localConfig: localConfig, logger: logger.MakeUpdated(commontypes.LogFields{"proto": "outgen"}), - metrics: newOutcomeGenerationMetrics(metricsRegisterer, logger), + metrics: newOutcomeGenerationMetrics(metricsRegisterer, logger, config.N()), netSender: netSender, offchainKeyring: offchainKeyring, reportingPlugin: reportingPlugin, diff --git a/offchainreporting2plus/internal/ocr3_1/protocol/outcome_generation_follower.go b/offchainreporting2plus/internal/ocr3_1/protocol/outcome_generation_follower.go index 9b6e18dc..975f3c33 100644 --- a/offchainreporting2plus/internal/ocr3_1/protocol/outcome_generation_follower.go +++ b/offchainreporting2plus/internal/ocr3_1/protocol/outcome_generation_follower.go @@ -1342,6 +1342,9 @@ func (outgen *outcomeGenerationState[RI]) backgroundCheckAttributedSignedObserva if myObservationIncluded { outgen.metrics.includedObservationsTotal.Inc() } + for _, aso := range asos { + outgen.metrics.theirIncludedObservationsTotal[aso.Observer].Inc() + } return attributedObservations, true } diff --git a/offchainreporting2plus/internal/ocr3_1/protocol/pacemaker.go b/offchainreporting2plus/internal/ocr3_1/protocol/pacemaker.go index d74ddf79..ce9c586d 100644 --- a/offchainreporting2plus/internal/ocr3_1/protocol/pacemaker.go +++ b/offchainreporting2plus/internal/ocr3_1/protocol/pacemaker.go @@ -70,7 +70,7 @@ func makePacemakerState[RI any]( id: id, localConfig: localConfig, logger: logger.MakeUpdated(commontypes.LogFields{"proto": "pacemaker"}), - metrics: newPacemakerMetrics(metricsRegisterer, logger), + metrics: newPacemakerMetrics(metricsRegisterer, logger, config.N()), netSender: netSender, offchainKeyring: offchainKeyring, telemetrySender: telemetrySender, @@ -279,6 +279,8 @@ func (pace *pacemakerState[RI]) requestNewEpoch() { } func (pace *pacemakerState[RI]) messageNewEpochWish(msg MessageNewEpochWish[RI], sender commontypes.OracleID) { + pace.metrics.receivedNewEpochWishesTotal[sender].Inc() + if pace.newEpochWishes[sender] < msg.Epoch { pace.newEpochWishes[sender] = msg.Epoch } diff --git a/offchainreporting2plus/internal/ocr3_1/protocol/state_sync.go b/offchainreporting2plus/internal/ocr3_1/protocol/state_sync.go index b71d4498..add6f4b6 100644 --- a/offchainreporting2plus/internal/ocr3_1/protocol/state_sync.go +++ b/offchainreporting2plus/internal/ocr3_1/protocol/state_sync.go @@ -31,7 +31,7 @@ func RunStateSync[RI any]( reportingPlugin ocr3_1types.ReportingPlugin[RI], ) { - metrics := newStateSyncMetrics(metricsRegisterer, logger) + metrics := newStateSyncMetrics(metricsRegisterer, logger, config.N()) defer metrics.Close() subs := subprocesses.Subprocesses{} @@ -291,6 +291,7 @@ func (stasy *stateSyncState[RI]) messageStateSyncSummary(msg MessageStateSyncSum msg.HighestCommittedSeqNr, time.Now(), } + stasy.metrics.theirCommittedSeqNr[sender].Set(float64(msg.HighestCommittedSeqNr)) stasy.updateHighestHeardFromSummaries() stasy.tryToKickStartSync() diff --git a/offchainreporting2plus/internal/shim/metrics.go b/offchainreporting2plus/internal/shim/metrics.go index 3dc12ce4..5266c9f4 100644 --- a/offchainreporting2plus/internal/shim/metrics.go +++ b/offchainreporting2plus/internal/shim/metrics.go @@ -159,3 +159,124 @@ func (m *keyValueDatabaseMetrics) Close() { m.registerer.Unregister(m.committedReadWriteTransactionsTotal) m.registerer.Unregister(m.discardedReadWriteTransactionsTotal) } + +type readOnlyKeyValueStateMetrics struct { + registerer prometheus.Registerer + oldestOpenTransactionAge prometheus.GaugeFunc + openedTransactionsTotal prometheus.Counter + discardedTransactionsTotal prometheus.Counter + refusedInsidePluginCallTotal prometheus.Counter + refusedUnavailableTotal prometheus.Counter + transactionOpeningDuration prometheus.Histogram + transactionAgeAtDiscard prometheus.Histogram + transactionReads prometheus.Histogram + transactionReadBytes prometheus.Histogram +} + +func newReadOnlyKeyValueStateMetrics( + registerer prometheus.Registerer, + logger commontypes.Logger, + oldestOpenTransactionAgeSeconds func() float64, +) *readOnlyKeyValueStateMetrics { + // The number that predicts trouble. A read transaction keeps the database + // from reclaiming the versions of the data it can still see, and that cost + // is driven by how long the oldest one has been open rather than by how many + // are open. + oldestOpenTransactionAge := prometheus.NewGaugeFunc(prometheus.GaugeOpts{ + Name: "ocr3_1_read_only_key_value_state_oldest_open_transaction_age_seconds", + Help: "Seconds since creation of the oldest KeyValueStateReadTransaction returned by " + + "ReadOnlyKeyValueState.NewReadTransaction that has not been discarded. Zero if none.", + }, oldestOpenTransactionAgeSeconds) + metricshelper.RegisterOrLogError(logger, registerer, oldestOpenTransactionAge, "ocr3_1_read_only_key_value_state_oldest_open_transaction_age_seconds") + + openedTransactionsTotal := prometheus.NewCounter(prometheus.CounterOpts{ + Name: "ocr3_1_read_only_key_value_state_opened_transactions_total", + Help: "The number of successful ReadOnlyKeyValueState.NewReadTransaction calls. Subtract " + + "ocr3_1_read_only_key_value_state_discarded_transactions_total for matching label sets to obtain the number currently open.", + }) + metricshelper.RegisterOrLogError(logger, registerer, openedTransactionsTotal, "ocr3_1_read_only_key_value_state_opened_transactions_total") + + discardedTransactionsTotal := prometheus.NewCounter(prometheus.CounterOpts{ + Name: "ocr3_1_read_only_key_value_state_discarded_transactions_total", + Help: "The number of transactions returned by ReadOnlyKeyValueState.NewReadTransaction that have been " + + "discarded through KeyValueStateReadTransaction.Discard. " + + "Repeated Discard calls count once.", + }) + metricshelper.RegisterOrLogError(logger, registerer, discardedTransactionsTotal, "ocr3_1_read_only_key_value_state_discarded_transactions_total") + + // The vec itself is deliberately not kept: the per-reason counters are all + // the callers need, and unregistering any one of them unregisters the vec, + // see Close. + refusedTransactionsTotal := prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "ocr3_1_read_only_key_value_state_refused_transactions_total", + Help: "The number of ReadOnlyKeyValueState.NewReadTransaction calls returning an error, labeled by reason: " + + "inside_plugin_call for use from a prohibited ReportingPlugin method, " + + "unavailable for unavailable state or failure to create a transaction.", + }, []string{"reason"}) + metricshelper.RegisterOrLogError(logger, registerer, refusedTransactionsTotal, "ocr3_1_read_only_key_value_state_refused_transactions_total") + + transactionOpeningDuration := prometheus.NewHistogram(prometheus.HistogramOpts{ + Name: "ocr3_1_experimental_read_only_key_value_state_transaction_opening_duration_seconds", + Help: "The duration in seconds of successful ReadOnlyKeyValueState.NewReadTransaction calls. " + + "Calls returning an error are excluded.", + // Explicit boundaries keep integer-second le labels exact. + Buckets: []float64{1e-6, 1e-5, 1e-4, 1e-3, 1e-2, 1e-1, 1, 2, 5, 10}, + }) + metricshelper.RegisterOrLogError(logger, registerer, transactionOpeningDuration, "ocr3_1_experimental_read_only_key_value_state_transaction_opening_duration_seconds") + + transactionAgeAtDiscard := prometheus.NewHistogram(prometheus.HistogramOpts{ + Name: "ocr3_1_experimental_read_only_key_value_state_transaction_age_at_discard_seconds", + Help: "The age in seconds of each KeyValueStateReadTransaction returned by ReadOnlyKeyValueState.NewReadTransaction " + + "when discarded. Observed once per transaction on KeyValueStateReadTransaction.Discard.", + Buckets: []float64{1e-4, 1e-3, 1e-2, 1e-1, 1, 2, 5, 10, 100, 1000}, + }) + metricshelper.RegisterOrLogError(logger, registerer, transactionAgeAtDiscard, "ocr3_1_experimental_read_only_key_value_state_transaction_age_at_discard_seconds") + + transactionReads := prometheus.NewHistogram(prometheus.HistogramOpts{ + Name: "ocr3_1_experimental_read_only_key_value_state_transaction_reads", + Help: "The number of Read calls on each KeyValueStateReadTransaction returned by ReadOnlyKeyValueState.NewReadTransaction " + + "before discard, including errors and missing keys. Observed once on KeyValueStateReadTransaction.Discard. " + + "Calls after discard are excluded.", + Buckets: append([]float64{0}, prometheus.ExponentialBuckets(1, 10, 6)...), + }) + metricshelper.RegisterOrLogError(logger, registerer, transactionReads, "ocr3_1_experimental_read_only_key_value_state_transaction_reads") + + transactionReadBytes := prometheus.NewHistogram(prometheus.HistogramOpts{ + Name: "ocr3_1_experimental_read_only_key_value_state_transaction_read_bytes", + Help: "The sum of value lengths in bytes returned without error by Read on each KeyValueStateReadTransaction " + + "returned by ReadOnlyKeyValueState.NewReadTransaction. Repeated reads count separately. Observed once on " + + "KeyValueStateReadTransaction.Discard.", + Buckets: []float64{0, 100, 1e3, 1e4, 1e5, 1e6, 1e7, 1e8, 1e9}, + }) + metricshelper.RegisterOrLogError(logger, registerer, transactionReadBytes, "ocr3_1_experimental_read_only_key_value_state_transaction_read_bytes") + + return &readOnlyKeyValueStateMetrics{ + registerer, + oldestOpenTransactionAge, + openedTransactionsTotal, + discardedTransactionsTotal, + refusedTransactionsTotal.WithLabelValues("inside_plugin_call"), + refusedTransactionsTotal.WithLabelValues("unavailable"), + transactionOpeningDuration, + transactionAgeAtDiscard, + transactionReads, + transactionReadBytes, + } +} + +func (m *readOnlyKeyValueStateMetrics) Close() { + m.registerer.Unregister(m.oldestOpenTransactionAge) + m.registerer.Unregister(m.openedTransactionsTotal) + m.registerer.Unregister(m.discardedTransactionsTotal) + m.registerer.Unregister(m.transactionOpeningDuration) + m.registerer.Unregister(m.transactionAgeAtDiscard) + m.registerer.Unregister(m.transactionReads) + m.registerer.Unregister(m.transactionReadBytes) + // A child of a vec describes the vec's descriptor, and prometheus.Registerer + // documents that Unregister matches collectors by the descriptors they + // describe. So unregistering the first child unregisters the whole vec, and + // the remaining ones are no-ops. Unregistering all of them keeps this + // method oblivious to which is which. + m.registerer.Unregister(m.refusedInsidePluginCallTotal) + m.registerer.Unregister(m.refusedUnavailableTotal) +} diff --git a/offchainreporting2plus/internal/shim/ocr3_1_context_marking_reporting_plugin.go b/offchainreporting2plus/internal/shim/ocr3_1_context_marking_reporting_plugin.go new file mode 100644 index 00000000..d7d0a966 --- /dev/null +++ b/offchainreporting2plus/internal/shim/ocr3_1_context_marking_reporting_plugin.go @@ -0,0 +1,94 @@ +package shim + +import ( + "context" + + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3_1types" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" + "github.com/smartcontractkit/libocr/offchainreporting2plus/types" +) + +// ContextMarkingOCR3_1ReportingPlugin marks the context it passes into the +// wrapped plugin's methods, so that facilities the plugin was handed for use +// *outside* its methods can tell when they are being misused from *inside* one. +// +// Currently the only such facility is [ocr3_1types.ReadOnlyKeyValueState]: the +// methods that are handed a KeyValueStateReader must use that one, since it +// reads at the round's sequence number, and reading anything else would +// silently lose determinism and break agreement between oracles. See +// withOutsideKeyValueStateReadsDisallowed for which methods are marked. +// + +type ContextMarkingOCR3_1ReportingPlugin[RI any] struct { + Plugin ocr3_1types.ReportingPlugin[RI] +} + +var _ ocr3_1types.ReportingPlugin[struct{}] = ContextMarkingOCR3_1ReportingPlugin[struct{}]{} + +func (rp ContextMarkingOCR3_1ReportingPlugin[RI]) Query(ctx context.Context, seqNr uint64, kvReader ocr3_1types.KeyValueStateReader, blobBroadcastFetcher ocr3_1types.BlobBroadcastFetcher) (types.Query, error) { + return rp.Plugin.Query(withOutsideKeyValueStateReadsDisallowed(ctx), seqNr, kvReader, blobBroadcastFetcher) +} + +func (rp ContextMarkingOCR3_1ReportingPlugin[RI]) ObservationQuorum(ctx context.Context, seqNr uint64, aq types.AttributedQuery, aos []types.AttributedObservation, kvReader ocr3_1types.KeyValueStateReader, blobFetcher ocr3_1types.BlobFetcher) (bool, error) { + return rp.Plugin.ObservationQuorum(withOutsideKeyValueStateReadsDisallowed(ctx), seqNr, aq, aos, kvReader, blobFetcher) +} + +func (rp ContextMarkingOCR3_1ReportingPlugin[RI]) Observation(ctx context.Context, seqNr uint64, aq types.AttributedQuery, kvReader ocr3_1types.KeyValueStateReader, blobBroadcastFetcher ocr3_1types.BlobBroadcastFetcher) (types.Observation, error) { + return rp.Plugin.Observation(withOutsideKeyValueStateReadsDisallowed(ctx), seqNr, aq, kvReader, blobBroadcastFetcher) +} + +func (rp ContextMarkingOCR3_1ReportingPlugin[RI]) ValidateObservation(ctx context.Context, seqNr uint64, aq types.AttributedQuery, ao types.AttributedObservation, kvReader ocr3_1types.KeyValueStateReader, blobFetcher ocr3_1types.BlobFetcher) error { + return rp.Plugin.ValidateObservation(withOutsideKeyValueStateReadsDisallowed(ctx), seqNr, aq, ao, kvReader, blobFetcher) +} + +func (rp ContextMarkingOCR3_1ReportingPlugin[RI]) StateTransition(ctx context.Context, seqNr uint64, aq types.AttributedQuery, aos []types.AttributedObservation, kvReadWriter ocr3_1types.KeyValueStateReadWriter, blobFetcher ocr3_1types.BlobFetcher) (ocr3_1types.ReportsPlusPrecursor, error) { + return rp.Plugin.StateTransition(withOutsideKeyValueStateReadsDisallowed(ctx), seqNr, aq, aos, kvReadWriter, blobFetcher) +} + +func (rp ContextMarkingOCR3_1ReportingPlugin[RI]) Committed(ctx context.Context, seqNr uint64, keyValueReader ocr3_1types.KeyValueStateReader) error { + return rp.Plugin.Committed(withOutsideKeyValueStateReadsDisallowed(ctx), seqNr, keyValueReader) +} + +func (rp ContextMarkingOCR3_1ReportingPlugin[RI]) Reports(ctx context.Context, seqNr uint64, reportsPlusPrecursor ocr3_1types.ReportsPlusPrecursor) ([]ocr3types.ReportPlus[RI], error) { + return rp.Plugin.Reports(withOutsideKeyValueStateReadsDisallowed(ctx), seqNr, reportsPlusPrecursor) +} + +func (rp ContextMarkingOCR3_1ReportingPlugin[RI]) ShouldAcceptAttestedReport(ctx context.Context, seqNr uint64, reportWithInfo ocr3types.ReportWithInfo[RI]) (bool, error) { + return rp.Plugin.ShouldAcceptAttestedReport(ctx, seqNr, reportWithInfo) +} + +func (rp ContextMarkingOCR3_1ReportingPlugin[RI]) ShouldTransmitAcceptedReport(ctx context.Context, seqNr uint64, reportWithInfo ocr3types.ReportWithInfo[RI]) (bool, error) { + return rp.Plugin.ShouldTransmitAcceptedReport(ctx, seqNr, reportWithInfo) +} + +func (rp ContextMarkingOCR3_1ReportingPlugin[RI]) Close() error { + return rp.Plugin.Close() +} + +// outsideKeyValueStateReadsDisallowedKey marks a context as belonging to the +// execution of a ReportingPlugin method that must not read the KeyValueState +// through anything but the KeyValueStateReader it was given. The type is +// unexported so that nobody but libocr can produce the marker. +type outsideKeyValueStateReadsDisallowedKey struct{} + +// withOutsideKeyValueStateReadsDisallowed marks ctx so that +// [ReadOnlyKeyValueStateWrapper] refuses to open a read transaction from it. +// +// Marked: Query, Observation, ValidateObservation, ObservationQuorum, +// StateTransition, Committed and Reports. They run at a specific sequence +// number, and those documented as pure would otherwise silently lose +// determinism. +// +// Deliberately not marked: ShouldAcceptAttestedReport and +// ShouldTransmitAcceptedReport. They are node-local transmission decisions that +// are not required to be pure, and they receive no KeyValueStateReader of their +// own, so the outside ReadOnlyKeyValueState is their only way to consult the +// replicated state. Consulting it there is legitimate. +func withOutsideKeyValueStateReadsDisallowed(ctx context.Context) context.Context { + + return context.WithValue(ctx, outsideKeyValueStateReadsDisallowedKey{}, struct{}{}) +} + +func outsideKeyValueStateReadsDisallowed(ctx context.Context) bool { + return ctx.Value(outsideKeyValueStateReadsDisallowedKey{}) != nil +} diff --git a/offchainreporting2plus/internal/shim/ocr3_1_read_only_key_value_state.go b/offchainreporting2plus/internal/shim/ocr3_1_read_only_key_value_state.go new file mode 100644 index 00000000..08e639d1 --- /dev/null +++ b/offchainreporting2plus/internal/shim/ocr3_1_read_only_key_value_state.go @@ -0,0 +1,314 @@ +package shim + +import ( + "context" + "errors" + "fmt" + "slices" + "sync" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/smartcontractkit/libocr/commontypes" + "github.com/smartcontractkit/libocr/offchainreporting2plus/internal/common/list" + "github.com/smartcontractkit/libocr/offchainreporting2plus/internal/ocr3_1/protocol" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3_1types" + "github.com/smartcontractkit/libocr/subprocesses" +) + +const ( + // A read transaction held for longer than this is almost certainly a bug in + // the plugin: reads take microseconds, so anything this slow is blocking on + // something else while keeping the transaction open. We only complain, we + // never discard it out from under a running plugin. + recommendedMaxKeyValueStateReadTransactionDuration = 10 * time.Second + longHeldKeyValueStateReadTransactionCheckInterval = 10 * time.Second +) + +var ( + errKeyValueStateUnavailable = errors.New("key value state is unavailable") + errKeyValueStateReadInsidePluginCall = errors.New("key value state must not be read via ReadOnlyKeyValueState inside this ReportingPlugin method; use the method's KeyValueStateReader or KeyValueStateReadWriter argument if available") + errKeyValueStateReadTransactionDiscarded = errors.New("key value state read transaction has been discarded") +) + +// ReadOnlyKeyValueStateWrapper implements [ocr3_1types.ReadOnlyKeyValueState], +// the plugin's read access to the KeyValueState from outside its own methods. +// +// It exists to enable deferred initialization: the reporting plugin is +// constructed before the key value database is opened (the database needs the +// limits that the plugin's factory reports), so the plugin is handed the +// wrapper up front and the database is supplied later through Enable. +type ReadOnlyKeyValueStateWrapper struct { + logger commontypes.Logger + metrics *readOnlyKeyValueStateMetrics + + subs subprocesses.Subprocesses + chStop chan struct{} + stopOnce sync.Once + + mu sync.Mutex + nilOrKeyValueDatabase protocol.KeyValueDatabase + // openReadTransactions is ordered from oldest to newest: a transaction's + // openedAt is assigned and the transaction registered while holding mu, so + // each registration is newer than the previous one. The front is therefore + // the oldest open transaction, in O(1). + openReadTransactions *list.List[*keyValueStateReadTransaction] +} + +var _ ocr3_1types.ReadOnlyKeyValueState = &ReadOnlyKeyValueStateWrapper{} + +func NewReadOnlyKeyValueStateWrapper(logger commontypes.Logger, registerer prometheus.Registerer) *ReadOnlyKeyValueStateWrapper { + w := &ReadOnlyKeyValueStateWrapper{ + logger, + nil, + subprocesses.Subprocesses{}, + make(chan struct{}), + sync.Once{}, + sync.Mutex{}, + nil, + list.NewList[*keyValueStateReadTransaction](), + } + w.metrics = newReadOnlyKeyValueStateMetrics( + registerer, + logger, + func() float64 { return w.openReadTransactionsSummary().oldestDuration.Seconds() }, + ) + w.subs.Go(w.complainAboutLongHeldReadTransactions) + return w +} + +type openReadTransactionsSummary struct { + count int + // oldestSeqNr and oldestDuration are zero if count is zero. + oldestSeqNr uint64 + oldestDuration time.Duration +} + +func (w *ReadOnlyKeyValueStateWrapper) openReadTransactionsSummary() openReadTransactionsSummary { + w.mu.Lock() + defer w.mu.Unlock() + + summary := openReadTransactionsSummary{w.openReadTransactions.Len(), 0, 0} + if oldest, ok := w.openReadTransactions.Front(); ok { + summary.oldestSeqNr = oldest.seqNr + summary.oldestDuration = time.Since(oldest.openedAt) + } + return summary +} + +// complainAboutLongHeldReadTransactions polls rather than arming a timer per +// read transaction, both to keep the plugin's read path free of per-transaction +// overhead and so that a plugin holding one forever keeps being complained +// about, not just once. +func (w *ReadOnlyKeyValueStateWrapper) complainAboutLongHeldReadTransactions() { + ticker := time.NewTicker(longHeldKeyValueStateReadTransactionCheckInterval) + defer ticker.Stop() + for { + select { + case <-w.chStop: + return + case <-ticker.C: + summary := w.openReadTransactionsSummary() + if summary.oldestDuration <= recommendedMaxKeyValueStateReadTransactionDuration { + continue + } + w.logger.Warn("KeyValueStateReadTransaction opened by ReportingPlugin is being held open for longer than recommended", commontypes.LogFields{ + "oldestOpenReadTransactionSeqNr": summary.oldestSeqNr, + "oldestOpenReadTransactionDuration": summary.oldestDuration.String(), + "recommendedMaxDuration": recommendedMaxKeyValueStateReadTransactionDuration.String(), + "currentlyOpenReadTransactions": summary.count, + }) + } + } +} + +func (w *ReadOnlyKeyValueStateWrapper) Enable(keyValueDatabase protocol.KeyValueDatabase) { + w.mu.Lock() + defer w.mu.Unlock() + w.nilOrKeyValueDatabase = keyValueDatabase +} + +// Disable stops handing out read transactions and discards those still open. +// It undoes Enable; it does not release the resources acquired by +// NewReadOnlyKeyValueStateWrapper, that is what Close is for. +// +// It is called after the reporting plugin has been closed and before the +// underlying key value database is closed. A well-behaved plugin has stopped +// its goroutines and discarded its transactions by the time Close returns, so +// anything still open here is a leak: we log it and discard it on the plugin's +// behalf, which releases the database's resources and makes the plugin's next +// Read fail with a clear error rather than reaching a closed database. +// +// Read holds the transaction's shared lock across the call into the database. +// Discard takes the exclusive lock, waiting for in-flight reads to finish before +// releasing the transaction. Subsequent reads fail with +// errKeyValueStateReadTransactionDiscarded. We do not wait for the plugin to +// discard the transaction itself. +func (w *ReadOnlyKeyValueStateWrapper) Disable() { + w.mu.Lock() + w.nilOrKeyValueDatabase = nil + leakedReadTransactions := slices.Collect(w.openReadTransactions.All()) + w.mu.Unlock() + + if len(leakedReadTransactions) > 0 { + oldest := leakedReadTransactions[0] + w.logger.Error("ReportingPlugin has been closed but is still holding read transactions open. This is a bug in the ReportingPlugin.", commontypes.LogFields{ + "oldestOpenReadTransactionSeqNr": oldest.seqNr, + "oldestOpenReadTransactionDuration": time.Since(oldest.openedAt).String(), + "currentlyOpenReadTransactions": len(leakedReadTransactions), + }) + } + + // Discard takes w.mu to unregister, so we must not be holding it here. + for _, readTransaction := range leakedReadTransactions { + readTransaction.Discard() + } +} + +// Close releases everything NewReadOnlyKeyValueStateWrapper acquired: it stops +// the goroutine that complains about long-held read transactions and +// unregisters the metrics. It Disables first, so that on its own it is a +// complete cleanup and can be deferred right after construction as a safety +// net, regardless of whether Enable was ever called. Closing more than once is +// a no-op. +// +// Close inherits Disable's synchronization when discarding transactions. +func (w *ReadOnlyKeyValueStateWrapper) Close() { + // Idempotent, so this is a no-op if the caller has already Disabled, as the + // managed oracle does on its happy path. + w.Disable() + w.stopOnce.Do(func() { close(w.chStop) }) + w.subs.Wait() + w.metrics.Close() +} + +func (w *ReadOnlyKeyValueStateWrapper) NewReadTransaction(ctx context.Context) (ocr3_1types.KeyValueStateReadTransaction, error) { + if outsideKeyValueStateReadsDisallowed(ctx) { + w.metrics.refusedInsidePluginCallTotal.Inc() + return nil, errKeyValueStateReadInsidePluginCall + } + + startedAt := time.Now() + readTransaction, err := w.newReadTransaction() + if err != nil { + w.metrics.refusedUnavailableTotal.Inc() + return nil, err + } + w.metrics.transactionOpeningDuration.Observe(time.Since(startedAt).Seconds()) + return readTransaction, nil +} + +func (w *ReadOnlyKeyValueStateWrapper) newReadTransaction() (*keyValueStateReadTransaction, error) { + w.mu.Lock() + defer w.mu.Unlock() + + if w.nilOrKeyValueDatabase == nil { + return nil, errKeyValueStateUnavailable + } + + rawReadTransaction, err := w.nilOrKeyValueDatabase.NewReadTransactionUnchecked() + if err != nil { + return nil, fmt.Errorf("%w: failed to create read transaction: %w", errKeyValueStateUnavailable, err) + } + + // Both of the following read from the very transaction we are about to hand + // to the plugin, so the state it sees and the sequence number we report for + // it are guaranteed to agree. + if err := checkNotClobbered(rawReadTransaction); err != nil { + rawReadTransaction.Discard() + return nil, fmt.Errorf("%w: %w", errKeyValueStateUnavailable, err) + } + seqNr, err := rawReadTransaction.ReadHighestCommittedSeqNr() + if err != nil { + rawReadTransaction.Discard() + return nil, fmt.Errorf("%w: failed to read highest committed seq nr: %w", errKeyValueStateUnavailable, err) + } + + readTransaction := &keyValueStateReadTransaction{ + seqNr, + time.Now(), + list.Element[*keyValueStateReadTransaction]{}, + nil, + sync.RWMutex{}, + rawReadTransaction, + sync.Mutex{}, + 0, + 0, + } + readTransaction.element = w.openReadTransactions.PushBack(readTransaction) + readTransaction.onDiscard = func() { + w.unregisterReadTransaction(readTransaction) + } + w.metrics.openedTransactionsTotal.Inc() + return readTransaction, nil +} + +func (w *ReadOnlyKeyValueStateWrapper) unregisterReadTransaction(readTransaction *keyValueStateReadTransaction) { + w.mu.Lock() + defer w.mu.Unlock() + w.openReadTransactions.Remove(readTransaction.element) + w.metrics.discardedTransactionsTotal.Inc() + w.metrics.transactionAgeAtDiscard.Observe(time.Since(readTransaction.openedAt).Seconds()) + w.metrics.transactionReads.Observe(float64(readTransaction.reads)) + w.metrics.transactionReadBytes.Observe(float64(readTransaction.readBytes)) +} + +type keyValueStateReadTransaction struct { + seqNr uint64 + // openedAt and element are only accessed by the wrapper, under its mu. + openedAt time.Time + element list.Element[*keyValueStateReadTransaction] + // onDiscard is invoked exactly once, by Discard. + onDiscard func() + + // mu makes the transaction safe for concurrent use, as promised by + // [ocr3_1types.KeyValueStateReadTransaction]. Reads take it for reading and + // run in parallel; the underlying database read is itself safe for that. + // Discard takes it for writing so that it cannot close the transaction + // underneath an in-flight read. Disable calls Discard from another + // goroutine, so this is not merely defensive. + mu sync.RWMutex + // nilOrRawReadTransaction is nil once discarded. + nilOrRawReadTransaction protocol.KeyValueDatabaseReadTransaction + + // readStatsMu serializes counter updates without serializing database + // reads. Read also holds mu for reading, so Discard's exclusive mu lock + // ensures all updates finish before onDiscard observes the totals. + readStatsMu sync.Mutex + reads uint64 + readBytes uint64 +} + +var _ ocr3_1types.KeyValueStateReadTransaction = &keyValueStateReadTransaction{} + +func (r *keyValueStateReadTransaction) Read(key []byte) ([]byte, error) { + r.mu.RLock() + defer r.mu.RUnlock() + if r.nilOrRawReadTransaction == nil { + return nil, errKeyValueStateReadTransactionDiscarded + } + value, err := r.nilOrRawReadTransaction.Read(key) + r.readStatsMu.Lock() + defer r.readStatsMu.Unlock() + r.reads++ + if err != nil { + return value, err + } + r.readBytes += uint64(len(value)) + return value, nil +} + +func (r *keyValueStateReadTransaction) SeqNr() uint64 { + return r.seqNr +} + +func (r *keyValueStateReadTransaction) Discard() { + r.mu.Lock() + defer r.mu.Unlock() + if r.nilOrRawReadTransaction == nil { + return + } + r.nilOrRawReadTransaction.Discard() + r.nilOrRawReadTransaction = nil + r.onDiscard() +} diff --git a/offchainreporting2plus/keyvaluedatabase/pebble_key_value_database.go b/offchainreporting2plus/keyvaluedatabase/pebble_key_value_database.go index 9922bda4..666a4f29 100644 --- a/offchainreporting2plus/keyvaluedatabase/pebble_key_value_database.go +++ b/offchainreporting2plus/keyvaluedatabase/pebble_key_value_database.go @@ -23,8 +23,6 @@ const ( pebbleBetterDefaultMaxMemTableSize = 64 << 20 // 64 MiB ) -var ErrPebbleReadWriteTransactionAlreadyOpen = errors.New("a read-write transaction is already open") - // NewPebbleKeyValueDatabaseFactory produces a // [ocr3_1types.KeyValueDatabaseFactory] that creates Pebble databases under // the directory indicated by baseDir. The directory must exist and be @@ -112,8 +110,6 @@ func (p *pebbleKeyValueDatabase) Close() error { return err } -// The resulting transaction is NOT thread-safe. - func (p *pebbleKeyValueDatabase) NewReadTransaction() (ocr3_1types.KeyValueDatabaseReadTransaction, error) { snapshot := p.db.NewSnapshot() return &pebbleReadTransaction{ diff --git a/offchainreporting2plus/ocr3_1shims/reporting_plugin_factory.go b/offchainreporting2plus/ocr3_1shims/reporting_plugin_factory.go new file mode 100644 index 00000000..bfe72b25 --- /dev/null +++ b/offchainreporting2plus/ocr3_1shims/reporting_plugin_factory.go @@ -0,0 +1,35 @@ +package ocr3_1shims + +import ( + "context" + + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3_1types" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" +) + +type reportingPluginFactory2Shim[RI any] struct { + wrapped ocr3_1types.ReportingPluginFactory[RI] +} + +func (f *reportingPluginFactory2Shim[RI]) NewReportingPlugin( + ctx context.Context, + config ocr3types.ReportingPluginConfig, + blobBroadcastFetcher ocr3_1types.BlobBroadcastFetcher, + _ ocr3_1types.ReadOnlyKeyValueState, +) (ocr3_1types.ReportingPlugin[RI], ocr3_1types.ReportingPluginInfo, error) { + return f.wrapped.NewReportingPlugin(ctx, config, blobBroadcastFetcher) +} + +// ReportingPluginFactoryAsReportingPluginFactory2 wraps an +// [ocr3_1types.ReportingPluginFactory] and returns an +// [ocr3_1types.ReportingPluginFactory2] that drops the +// [ocr3_1types.ReadOnlyKeyValueState]. +// +// Unlike the other shims there is no "already implements the newer interface" +// fast path: both interfaces declare NewReportingPlugin with different +// signatures, so no concrete type can satisfy both. +func ReportingPluginFactoryAsReportingPluginFactory2[RI any]( + f ocr3_1types.ReportingPluginFactory[RI], +) ocr3_1types.ReportingPluginFactory2[RI] { + return &reportingPluginFactory2Shim[RI]{f} +} diff --git a/offchainreporting2plus/ocr3_1types/blob.go b/offchainreporting2plus/ocr3_1types/blob.go index 50255100..a7e8f680 100644 --- a/offchainreporting2plus/ocr3_1types/blob.go +++ b/offchainreporting2plus/ocr3_1types/blob.go @@ -26,7 +26,7 @@ type BlobBroadcaster interface { // future (e.g. in a query or observation). Note that BroadcastBlob may // return an error, and callers must gracefully handle such an error. Errors // might be transient, and related to resource exhaustion prevention, so - // callers are encouranged to employ a backoff when retrying. An error in + // callers are encouraged to employ a backoff when retrying. An error in // BroadcastBlob should not cause Observation to error also. BroadcastBlob // is a blocking call, and callers must ensure that it fits into their // runtime envelope. diff --git a/offchainreporting2plus/ocr3_1types/kvdb.go b/offchainreporting2plus/ocr3_1types/kvdb.go index ada3c715..4602b7f4 100644 --- a/offchainreporting2plus/ocr3_1types/kvdb.go +++ b/offchainreporting2plus/ocr3_1types/kvdb.go @@ -9,6 +9,15 @@ import ( // Deprecated: Use KeyValueDatabaseReadWriteTransaction instead. type KeyValueReadWriteTransaction = KeyValueDatabaseReadWriteTransaction +// A KeyValueDatabaseReadWriteTransaction is a read-write transaction over the +// database. Writes and deletes are visible to subsequent reads within the +// same transaction. See [KeyValueDatabaseReadTransaction] for the lifetime +// rules. +// +// # Concurrency +// +// Unlike a read transaction, a read-write transaction is not safe for +// concurrent use. type KeyValueDatabaseReadWriteTransaction interface { KeyValueDatabaseReadTransaction // A value of nil is interpreted as an empty slice, and does *not* delete @@ -16,12 +25,33 @@ type KeyValueDatabaseReadWriteTransaction interface { Write(key []byte, value []byte) error Delete(key []byte) error + // Commit commits the transaction. Whether it succeeds or fails, no + // further calls other than Discard may be made on the transaction once + // Commit returns. Commit() error } // Deprecated: Use KeyValueDatabaseReadTransaction instead. type KeyValueReadTransaction = KeyValueDatabaseReadTransaction +// A KeyValueDatabaseReadTransaction is a read transaction over the database. +// +// # Lifetime +// +// Every transaction must be discarded — or, for a +// [KeyValueDatabaseReadWriteTransaction], committed (or discarded to abort +// it). All iterators obtained from Range must be closed before the +// transaction is discarded or committed. After Discard or Commit return, no +// further calls may be made on the transaction, except that Discard may be +// called multiple times. +// +// # Concurrency +// +// Read and Range may be called concurrently with each other and themselves. +// Iterators from different Range calls may be used concurrently with each +// other and with Read and Range calls, but an individual iterator is not +// safe for concurrent use. Discard must not be called concurrently with any +// other call on the transaction. type KeyValueDatabaseReadTransaction interface { // If the key exists, the returned value must not be nil! Read(key []byte) ([]byte, error) @@ -34,15 +64,19 @@ type KeyValueDatabaseReadTransaction interface { // WARNING: DO NOT perform any writes/deletes to the key-value store while // the iterator is opened. Range(loKey []byte, hiKeyExcl []byte) KeyValueDatabaseIterator + // Discard discards the transaction and releases its resources. See + // [KeyValueDatabaseReadTransaction] for the lifetime rules. Discard() } // Deprecated: Use KeyValueDatabaseIterator instead. type KeyValueIterator = KeyValueDatabaseIterator -// KeyValueDatabaseIterator is a iterator over key-value pairs, in ascending order of -// keys. +// KeyValueDatabaseIterator is an iterator over key-value pairs, in ascending +// order of keys. See [KeyValueDatabaseReadTransaction] for the lifetime and +// concurrency rules that apply to iterators. // + // Example usage: // // it := kvReader.Range(loKey, hiKeyExcl) @@ -76,12 +110,24 @@ type KeyValueDatabaseIterator interface { // and will not cause a non-nil error. Err() error // Close closes the iterator and releases any resources associated with it. - // Further iteration is prevented, i.e., [KeyValueDatabaseIterator.Next] will return - // false. Must be called in any case, even if the iteration encountered any + // Must be called in any case, even if the iteration encountered any // error through [KeyValueDatabaseIterator.Value] or [KeyValueDatabaseIterator.Err]. + // After Close returns, no further calls may be made on the iterator, + // including another Close. Close() error } +// KeyValueDatabase is a key-value database. +// +// NewReadTransaction and NewReadWriteTransaction may be called concurrently +// with each other and themselves. At most one read-write transaction is +// active at a time: NewReadWriteTransaction blocks until the previous +// read-write transaction has been committed or discarded. +// +// Every transaction must be committed or discarded (see +// [KeyValueDatabaseReadTransaction]) before Close is called. Close must not +// be called concurrently with any other method, and no method may be called +// after Close. type KeyValueDatabase interface { NewReadWriteTransaction() (KeyValueDatabaseReadWriteTransaction, error) NewReadTransaction() (KeyValueDatabaseReadTransaction, error) diff --git a/offchainreporting2plus/ocr3_1types/plugin.go b/offchainreporting2plus/ocr3_1types/plugin.go index 0eb73d7c..f52486d6 100644 --- a/offchainreporting2plus/ocr3_1types/plugin.go +++ b/offchainreporting2plus/ocr3_1types/plugin.go @@ -23,10 +23,26 @@ type ReportingPluginFactory[RI any] interface { ) (ReportingPlugin[RI], ReportingPluginInfo, error) } +// ReportingPluginFactory2 is like ReportingPluginFactory but additionally +// provides a [ReadOnlyKeyValueState]. See its documentation for transaction +// ownership and lifetime requirements. +type ReportingPluginFactory2[RI any] interface { + // Creates a new reporting plugin instance. The instance may have + // associated goroutines or hold system resources, which should be + // released when its Close() function is called. + NewReportingPlugin( + context.Context, + ocr3types.ReportingPluginConfig, + BlobBroadcastFetcher, + ReadOnlyKeyValueState, + ) (ReportingPlugin[RI], ReportingPluginInfo, error) +} + // Deprecated: Use KeyValueStateReader instead. type KeyValueReader = KeyValueStateReader // KeyValueStateReader provides read access to the replicated KeyValueState. +// Read is safe for concurrent use. type KeyValueStateReader interface { // Read reads the value associated with the given key. // A nil key is interpreted as an empty slice. @@ -35,10 +51,104 @@ type KeyValueStateReader interface { Read(key []byte) ([]byte, error) } +// ReadOnlyKeyValueState gives you read access to the replicated KeyValueState, +// for example to answer requests from your own clients. +// +// # ReportingPlugin method restrictions +// +// Do not use it, or any transaction obtained from it, inside [ReportingPlugin] +// methods that: +// - are marked pure‡; or +// - receive a [KeyValueStateReader] or [KeyValueStateReadWriter] argument. +// +// Pure‡ methods must return results determined solely by their arguments, but +// the state read through ReadOnlyKeyValueState is not. Methods that receive a +// state reader or read-writer argument must use that argument instead, so that +// they access state at the appropriate sequence number. +// +// # Transaction ownership and resource usage +// +// You are responsible for bounding the number of transactions you keep open +// concurrently. The library does not enforce a cap: excessive concurrent usage +// or leaked transactions can exhaust resources and degrade or stall the oracle. +// +// You must discard every transaction you open, promptly after reading what you +// need. An open transaction prevents the database from reclaiming older versions +// of the data it can still see. Long-held or leaked transactions can cause +// unbounded storage growth. Do not call out over the network, wait on a lock, or +// block on other work while holding a transaction open. See +// [ReadOnlyKeyValueState.NewReadTransaction] for a deferred-discard example. +// +// # Plugin shutdown +// +// Before [ReportingPlugin.Close] returns, the plugin must discard all its +// transactions and stop any goroutines that could open or use transactions. +// Do not use ReadOnlyKeyValueState or its transactions after +// [ReportingPlugin.Close] returns. +// +// # Monitoring +// +// Use the following Prometheus metrics to monitor transaction usage. Subtract +// ocr3_1_read_only_key_value_state_discarded_transactions_total from +// ocr3_1_read_only_key_value_state_opened_transactions_total for matching label +// sets to track the number currently open. Monitor +// ocr3_1_read_only_key_value_state_oldest_open_transaction_age_seconds +// to detect long-held transactions and leaks. +type ReadOnlyKeyValueState interface { + // NewReadTransaction returns a [KeyValueStateReadTransaction] over the + // replicated KeyValueState. See [ReadOnlyKeyValueState] for transaction + // ownership, lifetime, resource usage, and monitoring guidance. + // + // ctx applies only to this call, not to the returned transaction. + // Canceling ctx does not discard the transaction or prevent further reads; + // you must still call Discard. + // + // Defer Discard immediately after successfully opening a transaction: + // + // readTransaction, err := keyValueState.NewReadTransaction(ctx) + // if err != nil { + // return err + // } + // defer readTransaction.Discard() + // + // Errors can occur during normal operation, for example when the + // KeyValueState is temporarily unavailable while the oracle is starting up. + // Fail the request you were serving and try again later. + // See [ReadOnlyKeyValueState] for restrictions on use from ReportingPlugin + // methods. + NewReadTransaction(ctx context.Context) (KeyValueStateReadTransaction, error) +} + +// KeyValueStateReadTransaction provides read access to the replicated +// KeyValueState at a fixed point in time. +// +// See [ReadOnlyKeyValueState] for transaction ownership, lifetime, resource +// usage, and monitoring guidance. A transaction is safe for concurrent use. +// +// See [ReadOnlyKeyValueState.NewReadTransaction] for a deferred-discard example. +type KeyValueStateReadTransaction interface { + KeyValueStateReader + + // SeqNr returns the sequence number this transaction reads at: it shows you + // the KeyValueState as it was once SeqNr() had been committed. Newer + // sequence numbers may be committed while you hold the transaction; what it + // shows you will not change. + SeqNr() uint64 + + // Discard releases the transaction. Calling it more than once is fine. + // After Discard returns, no further calls other than Discard may be made + // on the transaction. + Discard() +} + // Deprecated: Use KeyValueStateReadWriter instead. type KeyValueReadWriter = KeyValueStateReadWriter // KeyValueStateReadWriter provides read and write access to the replicated KeyValueState. +// +// Write and Delete are not safe for concurrent use: interleaved concurrent +// writes and deletes may corrupt the resulting state transition. Perform all +// writes and deletes from a single goroutine. type KeyValueStateReadWriter interface { KeyValueStateReader // Write writes the value for the provided key. @@ -59,13 +169,13 @@ type KeyValueStateReadWriter interface { // To do so, the ReportingPlugin defines a number of callbacks that are called // by the OCR protocol logic at certain points in the protocol's execution flow. // The report generated by the ReportingPlugin must be in a format understood by -// contract (or offchain target system) that the reports are transmitted to. +// the contract (or offchain target system) that the reports are transmitted to. // // # Byzantine Fault Tolerance // // We assume that each correct node participating in the protocol instance will // be running the same ReportingPlugin implementation. However, not all nodes -// may be correct; up to f nodes be faulty in arbitrary ways (aka byzantine +// may be correct; up to f nodes may be faulty in arbitrary ways (aka byzantine // faults). For example, faulty nodes could be down, have intermittent // connectivity issues, send garbage messages, or be controlled by an adversary. // @@ -85,7 +195,7 @@ type KeyValueStateReadWriter interface { // given round // - the observation returned by Observation is not included in the list of // AttributedObservations passed to StateTransition -// - a [Query] or [Observation] is malformed. (For defense in depth, it is also +// - a [types.Query] or [types.Observation] is malformed. (For defense in depth, it is also // recommended that malformed elements in the key value state are handled gracefully.) // - instances of the ReportingPlugin run by different oracles have different // call traces. E.g., the ReportingPlugin's Observation function may have been @@ -140,8 +250,8 @@ type ReportingPlugin[RI any] interface { // consistent view to different oracles (e.g. in case of observing a // blockchain). // - // You may assume that the seqNr is increasing strictly monotonically - // across the lifetime of a protocol instance. + // You may assume that the seqNr is increasing monotonically (though *not* + // strictly) across the lifetime of a protocol instance. // // The keyValueStateReader gives read access to the replicated KeyValueState // at the point after seqNr - 1 is committed. It must not be used outside the execution @@ -165,8 +275,8 @@ type ReportingPlugin[RI any] interface { // Observation gets an observation from the underlying data source. Returns // a value or an error. // - // You may assume that the seqNr is increasing strictly monotonically - // across the lifetime of a protocol instance. + // You may assume that the seqNr is increasing monotonically (though *not* + // strictly) across the lifetime of a protocol instance. // // The keyValueStateReader gives read access to the replicated KeyValueState // at the point after seqNr - 1 is committed. It must not be used outside the execution @@ -193,8 +303,11 @@ type ReportingPlugin[RI any] interface { // // This function must be pure‡. // - // You may assume that the seqNr is increasing strictly monotonically - // across the lifetime of a protocol instance. + // You may assume that the seqNr is increasing monotonically (though *not* + // strictly) across the lifetime of a protocol instance. + // + // This function may be called multiple times for the same seqNr, e.g. once + // for each observation. // // The keyValueStateReader gives read access to the replicated KeyValueState // at the point after seqNr - 1 is committed. It must not be used outside the execution @@ -220,6 +333,9 @@ type ReportingPlugin[RI any] interface { // // This function must be pure‡. // + // This function may be called multiple times for the same seqNr, e.g. on + // the leader as more observations arrive. + // // This is an advanced feature. The "default" approach (what OCR1 & OCR2 // did) is to have this function call // quorumhelper.ObservationCountReachesObservationQuorum(QuorumTwoFPlusOne, ...) @@ -258,8 +374,8 @@ type ReportingPlugin[RI any] interface { // // This function must be pure‡. // - // You may assume that the seqNr is increasing strictly monotonically - // across the lifetime of a protocol instance. + // You may assume that the seqNr is increasing monotonically (though *not* + // strictly) across the lifetime of a protocol instance. // // You may assume that the provided list of attributed observations has been // (1) validated by ValidateObservation on each element, (2) checked @@ -271,7 +387,8 @@ type ReportingPlugin[RI any] interface { // a ReportingPlugin to modify the replicated KeyValueState. The keyValueStateReadWriter must not be // used outside the execution of this function. (It's okay to use it anywhere in the call tree rooted // at this function and to pass it to separate goroutines, but they must stop using it before this - // function returns.) + // function returns. Do not call Write or Delete concurrently from multiple goroutines; perform all + // writes and deletes from a single goroutine.) // // The blobFetcher enables fetching blobs. It must not be used outside the execution // of this function. (It's okay to use it anywhere in the call tree rooted at this function @@ -288,12 +405,16 @@ type ReportingPlugin[RI any] interface { // Committed notifies the plugin that seqNr has been committed. // It might not be called for all sequence numbers. - // This method return an error != nil will not lead to the round being aborted/not committed. + // Returning an error != nil from this method will not lead to the round being aborted/not committed. // // Don't do anything slow in here. // // The keyValueStateReader gives read access to the replicated KeyValueState // in the state that it is after the StateTransition for seqNr is computed. + // It must not be used outside the execution of this function. (It's okay to + // use it anywhere in the call tree rooted at this function and to pass it to + // separate goroutines, but they must stop using it before this function + // returns.) Committed( ctx context.Context, seqNr uint64, @@ -351,6 +472,9 @@ type ReportingPlugin[RI any] interface { // needed, e.g. on shutdown of the protocol instance or shutdown of the // oracle node. This will only be called after any calls to other functions // of the plugin have completed. + // + // See [ReadOnlyKeyValueState] for transaction and goroutine cleanup + // requirements that must be satisfied before Close returns. Close() error } @@ -380,6 +504,12 @@ const ( // Messages are checked against these values during (de)serialization. Be // careful when changing these values, they could lead to different versions // of a ReportingPlugin being unable to communicate with each other. +// +// MaxKeyValueModifiedKeys and MaxKeyValueModifiedKeysPlusValuesBytes concern +// modifications to key-values inside the StateTransition method, rather than +// returned data. Write(k, v) and Delete(k) count as modifications. A +// modification that resets the value of a key to its original value at the +// start of StateTransition will still count towards the limit. type ReportingPluginLimits struct { MaxQueryBytes int MaxObservationBytes int @@ -387,11 +517,6 @@ type ReportingPluginLimits struct { MaxReportBytes int MaxReportCount int - // These limits concern modifications to key-values inside the - // StateTransition method. Write(k, v) and Delete(k) count as modifications. - // A modification that resets the value of a key to its original value at - // the start of StateTransition will still count towards the limit. - // MaxKeyValueModifiedKeys upper bounds the number of modified keys inside // the StateTransition method. A key might be modified multiple times within // a single StateTransition, but will only count once towards the limit. A @@ -420,7 +545,7 @@ type ReportingPluginLimits struct { // multiplying by a factor that accounts for a blob reaping interval in the // tens of seconds. MaxPerOracleUnexpiredBlobCumulativePayloadBytes int - // MaxPerOracleUnexpiredBlobCount is upper bounds the number of unreaped + // MaxPerOracleUnexpiredBlobCount upper bounds the number of unreaped // blobs from a single oracle. Inbound broadcasts that would violate this // upper bound will be rejected. Due to the asynchronous nature of blob // reaping, an expired blob might not be reaped yet, and thus might still be diff --git a/offchainreporting2plus/ocr3plugintesthelpers/plugin_determinism_wrapper.go b/offchainreporting2plus/ocr3plugintesthelpers/plugin_determinism_wrapper.go new file mode 100644 index 00000000..7b8c0374 --- /dev/null +++ b/offchainreporting2plus/ocr3plugintesthelpers/plugin_determinism_wrapper.go @@ -0,0 +1,210 @@ +package ocr3plugintesthelpers + +import ( + "context" + "fmt" + "reflect" + + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" + "github.com/smartcontractkit/libocr/offchainreporting2plus/types" +) + +var _ ocr3types.ReportingPlugin[any] = (*reportingPluginDeterminismWrapper[any])(nil) + +// Result is the (value, error) returned by a single wrapped call. +type Result struct { + Value any + Err error +} + +// Equal reports whether r and other are equivalent for the purposes of determinism checking: +// errors must match (by message), and when there is no error the values must be deeply equal. +// When an error is present the values are ignored, mirroring the rule used by checkDeterminism. +func (r Result) Equal(other Result) bool { + if !errorsEqualByString(r.Err, other.Err) { + return false + } + if r.Err != nil { + return true + } + return reflect.DeepEqual(r.Value, other.Value) +} + +// Call is a single invocation of a wrapped method, tagged with its zero-based index within the +// repetition sequence and the Result it returned. +type Call struct { + Index int // zero-based index of the call within the repetition sequence + Result Result // value and error returned by that call +} + +// ReportingPluginNondeterminismDetected is the value panicked with when a wrapped method returns different +// results across repetitions. It carries the method name, the arguments of the call, and the two +// Calls that disagreed — CallA is always the first call (index 0) and CallB is the first +// mismatching call — so the failure can be diagnosed without re-running. Args is the call's +// non-context arguments as a []any, in the order they were passed to the wrapped method. +type ReportingPluginNondeterminismDetected struct { + FuncName string // name of the ReportingPlugin method that was nondeterministic + Args []any // the non-context arguments passed to the wrapped call, in order + CallA Call // the first call (index 0) + CallB Call // the first mismatching call +} + +func (n ReportingPluginNondeterminismDetected) String() string { + return fmt.Sprintf( + "non-deterministic %s: args=%+v; call %d = (%+v, %v); call %d = (%+v, %v)", + n.FuncName, n.Args, + n.CallA.Index, n.CallA.Result.Value, n.CallA.Result.Err, + n.CallB.Index, n.CallB.Result.Value, n.CallB.Result.Err, + ) +} + +type reportingPluginDeterminismWrapper[RI any] struct { + underlying ocr3types.ReportingPlugin[RI] + numAdditionalCallRepetitions int +} + +// NewReportingPluginDeterminismWrapperForTests wraps the four ReportingPlugin functions documented +// as pure to ensure determinism, calling each one numAdditionalCallRepetitions extra times and +// comparing the results. Upon detecting nondeterminism it panics with a +// ReportingPluginNondeterminismDetected value, which fails the surrounding test. +// +// Not all forms of nondeterminism can be captured by this wrapper; for example race conditions +// or nondeterminism derived from node state are not always caught. It targets nondeterminism +// such as that caused by iterating a map, which is common in serialization. +// Returned values are not snapshotted, so a plugin that mutates memory backing a previously +// returned value may evade detection. +// +// Typical usage in a test: +// +// wrapped := ocr3plugintesthelpers.NewReportingPluginDeterminismWrapperForTests(plugin, 10) +// _, err := wrapped.Outcome(ctx, ocr3types.OutcomeContext{}, types.Query{}, nil) +// // If Outcome is nondeterministic, the call panics with a ReportingPluginNondeterminismDetected +// // value, failing the test. +// +// ONLY FOR TESTING. NEVER IN PRODUCTION. +func NewReportingPluginDeterminismWrapperForTests[RI any](underlying ocr3types.ReportingPlugin[RI], numAdditionalCallRepetitions int) ocr3types.ReportingPlugin[RI] { + if numAdditionalCallRepetitions < 1 { + panic(fmt.Sprintf("numAdditionalCallRepetitions must be >= 1, got %d", numAdditionalCallRepetitions)) + } + return &reportingPluginDeterminismWrapper[RI]{ + underlying, + numAdditionalCallRepetitions, + } +} + +func (w *reportingPluginDeterminismWrapper[RI]) Query(ctx context.Context, outctx ocr3types.OutcomeContext) (types.Query, error) { + return w.underlying.Query(ctx, outctx) +} + +func (w *reportingPluginDeterminismWrapper[RI]) Observation(ctx context.Context, outctx ocr3types.OutcomeContext, query types.Query) (types.Observation, error) { + return w.underlying.Observation(ctx, outctx, query) +} + +func (w *reportingPluginDeterminismWrapper[RI]) ValidateObservation(ctx context.Context, outctx ocr3types.OutcomeContext, query types.Query, ao types.AttributedObservation) error { + _, err := checkDeterminism("ValidateObservation", []any{outctx, query, ao}, w.numAdditionalCallRepetitions, func() (struct{}, error) { + return struct{}{}, w.underlying.ValidateObservation(ctx, outctx, query, ao) + }) + return err +} + +func (w *reportingPluginDeterminismWrapper[RI]) ObservationQuorum(ctx context.Context, outctx ocr3types.OutcomeContext, query types.Query, aos []types.AttributedObservation) (bool, error) { + return checkDeterminism("ObservationQuorum", []any{outctx, query, aos}, w.numAdditionalCallRepetitions, func() (bool, error) { + return w.underlying.ObservationQuorum(ctx, outctx, query, aos) + }) +} + +func (w *reportingPluginDeterminismWrapper[RI]) Outcome(ctx context.Context, outctx ocr3types.OutcomeContext, query types.Query, aos []types.AttributedObservation) (ocr3types.Outcome, error) { + return checkDeterminism("Outcome", []any{outctx, query, aos}, w.numAdditionalCallRepetitions, func() (ocr3types.Outcome, error) { + return w.underlying.Outcome(ctx, outctx, query, aos) + }) +} + +func (w *reportingPluginDeterminismWrapper[RI]) Reports(ctx context.Context, seqNr uint64, outcome ocr3types.Outcome) ([]ocr3types.ReportPlus[RI], error) { + return checkDeterminism("Reports", []any{seqNr, outcome}, w.numAdditionalCallRepetitions, func() ([]ocr3types.ReportPlus[RI], error) { + return w.underlying.Reports(ctx, seqNr, outcome) + }) +} + +func (w *reportingPluginDeterminismWrapper[RI]) ShouldAcceptAttestedReport(ctx context.Context, seqNr uint64, reportWithInfo ocr3types.ReportWithInfo[RI]) (bool, error) { + return w.underlying.ShouldAcceptAttestedReport(ctx, seqNr, reportWithInfo) +} + +func (w *reportingPluginDeterminismWrapper[RI]) ShouldTransmitAcceptedReport(ctx context.Context, seqNr uint64, reportWithInfo ocr3types.ReportWithInfo[RI]) (bool, error) { + return w.underlying.ShouldTransmitAcceptedReport(ctx, seqNr, reportWithInfo) +} + +func (w *reportingPluginDeterminismWrapper[RI]) Close() error { + return w.underlying.Close() +} + +type reportingPluginFactoryDeterminismWrapper[RI any] struct { + underlying ocr3types.ReportingPluginFactory[RI] + numAdditionalCallRepetitions int +} + +// NewReportingPluginFactoryDeterminismWrapperForTests wraps the given ReportingPluginFactory so +// that each plugin instance it creates verifies determinism by performing +// numAdditionalCallRepetitions additional calls to each pure method and comparing their results. +// Detected nondeterminism is surfaced as a ReportingPluginNondeterminismDetected panic. +// +// Use it to produce a ReportingPluginFactory that can be wired into OCR3OracleArgs (or +// OCR3OracleArgs2), so a real OCR instance drives the plugin and the determinism check runs +// against actual protocol traffic — handy for smoke testing. See NewReportingPluginDeterminismWrapperForTests +// for a usage example of the underlying wrapper. +// +// ONLY FOR TESTING. NEVER IN PRODUCTION. +func NewReportingPluginFactoryDeterminismWrapperForTests[RI any](underlying ocr3types.ReportingPluginFactory[RI], numAdditionalCallRepetitions int) ocr3types.ReportingPluginFactory[RI] { + if numAdditionalCallRepetitions < 1 { + panic(fmt.Sprintf("numAdditionalCallRepetitions must be >= 1, got %d", numAdditionalCallRepetitions)) + } + return &reportingPluginFactoryDeterminismWrapper[RI]{ + underlying, + numAdditionalCallRepetitions, + } +} + +func (f *reportingPluginFactoryDeterminismWrapper[RI]) NewReportingPlugin(ctx context.Context, config ocr3types.ReportingPluginConfig) (ocr3types.ReportingPlugin[RI], ocr3types.ReportingPluginInfo, error) { + plugin, info, err := f.underlying.NewReportingPlugin(ctx, config) + if err != nil { + return nil, ocr3types.ReportingPluginInfo{}, err + } + return &reportingPluginDeterminismWrapper[RI]{ + plugin, + f.numAdditionalCallRepetitions, + }, info, nil +} + +// checkDeterminism calls the function numAdditionalCallRepetitions extra times and compares each +// result against the first via Result.Equal. On a mismatch it panics with a +// ReportingPluginNondeterminismDetected value carrying the method name, the call's arguments, and the two +// Calls that disagreed. On success, the first call's (result, error) is returned unchanged so +// regular plugin errors propagate as usual. +func checkDeterminism[T any](funcName string, args []any, numAdditionalCallRepetitions int, call func() (T, error)) (T, error) { + v0, e0 := call() + a := Call{0, Result{v0, e0}} + for i := 1; i <= numAdditionalCallRepetitions; i++ { + vi, ei := call() + b := Call{i, Result{vi, ei}} + if !a.Result.Equal(b.Result) { + panic(ReportingPluginNondeterminismDetected{ + funcName, + args, + a, + b, + }) + } + } + return v0, e0 +} + +func errorsEqualByString(err1, err2 error) bool { + if err1 == nil && err2 == nil { + return true + } + + if err1 == nil || err2 == nil { + return false + } + + return err1.Error() == err2.Error() +} diff --git a/offchainreporting2plus/ocrintegrationtesthelpers/in_memory_key_value_database.go b/offchainreporting2plus/ocrintegrationtesthelpers/in_memory_key_value_database.go index 16b9ce38..8f6c8d8d 100644 --- a/offchainreporting2plus/ocrintegrationtesthelpers/in_memory_key_value_database.go +++ b/offchainreporting2plus/ocrintegrationtesthelpers/in_memory_key_value_database.go @@ -248,28 +248,32 @@ func (tx *inMemoryReadWriteTransaction) Range(loKey []byte, hiKeyExcl []byte) oc nil, nil, nil, + nil, } } ctx, cancel := context.WithCancel(tx.ctx) ch := make(chan item) + done := make(chan struct{}) // Btree is not safe to share across goroutines with concurrent writes. We // intentionally do not clone because as per the - // ocr3_1types.KeyValueDatabaseIterator interface, users must not write or - // delete keys while an iterator is opened. We want this to cause a nil - // dereference, panic, or other problematic behavior, so that we'll know we - // misuse the interface in protocol code. + // ocr3_1types.KeyValueDatabaseReadTransaction.Range contract, users must + // not write or delete keys while an iterator is opened. We want this to + // cause a nil dereference, panic, or other problematic behavior, so that + // we'll know we misuse the interface in protocol code. tree := tx.updatedTree iter := &inMemoryRangeIterator{ nil, cancel, + done, ch, tx, nil, } tx.db.subs.Go(func() { + defer close(done) defer close(ch) tree.AscendGreaterOrEqual(item{loKey, nil}, func(i item) bool { if len(hiKeyExcl) > 0 && bytes.Compare(i.key, hiKeyExcl) >= 0 { @@ -290,8 +294,13 @@ func (tx *inMemoryReadWriteTransaction) Range(loKey []byte, hiKeyExcl []byte) oc type inMemoryRangeIterator struct { err error cancel context.CancelFunc - ch chan item - tx *inMemoryReadWriteTransaction + // done is closed once the iterator goroutine has stopped touching the + // tree. Close waits for it, so that after Close returns the iterator + // holds no resources anymore, and e.g. writes/deletes to the transaction + // may safely resume. + done chan struct{} + ch chan item + tx *inMemoryReadWriteTransaction currentKey []byte } @@ -302,6 +311,9 @@ func (i *inMemoryRangeIterator) Close() error { if i.cancel != nil { i.cancel() } + if i.done != nil { + <-i.done + } return nil } @@ -345,13 +357,16 @@ func (tx *inMemoryReadWriteTransaction) Commit() error { } defer tx.db.rwSerializationLock.Unlock() - if err := tx.db.commit(tx.updatedTree); err != nil { - return err - } - + // Discard the transaction regardless of the commit outcome, as the Pebble + // implementation does. Otherwise a failed commit would leave the + // transaction looking live, and a subsequent Discard would release + // rwSerializationLock a second time, which is a fatal "unlock of unlocked + // mutex". + updatedTree := tx.updatedTree tx.updatedTree = nil tx.cancel() - return nil + + return tx.db.commit(updatedTree) } func (tx *inMemoryReadWriteTransaction) Delete(key []byte) error { diff --git a/offchainreporting2plus/oracle.go b/offchainreporting2plus/oracle.go index f05b0014..8ec2f798 100644 --- a/offchainreporting2plus/oracle.go +++ b/offchainreporting2plus/oracle.go @@ -5,6 +5,7 @@ import ( "fmt" "sync" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3_1shims" "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3_1types" "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3shims" @@ -364,7 +365,7 @@ func (args OCR3_1OracleArgs[RI]) runManaged(ctx context.Context) { args.OffchainConfigDigester, args.OffchainKeyring, ocr3shims.OnchainKeyringAsOnchainKeyring2(args.OnchainKeyring), - args.ReportingPluginFactory, + ocr3_1shims.ReportingPluginFactoryAsReportingPluginFactory2(args.ReportingPluginFactory), ) } @@ -393,6 +394,52 @@ func (args OCR3_1OracleArgs2[RI]) localConfig() types.LocalConfig { return args. func (args OCR3_1OracleArgs2[RI]) runManaged(ctx context.Context) { logger := loghelper.MakeRootLoggerWithContext(args.Logger) + managed.RunManagedOCR3_1Oracle( + ctx, + + args.V2Bootstrappers, + args.ContractConfigTracker, + args.ContractTransmitter, + args.Database, + args.KeyValueDatabaseFactory, + args.LocalConfig, + logger, + args.MetricsRegisterer, + args.MonitoringEndpoint, + args.BinaryNetworkEndpointFactory, + args.OffchainConfigDigester, + args.OffchainKeyring, + args.OnchainKeyring, + ocr3_1shims.ReportingPluginFactoryAsReportingPluginFactory2(args.ReportingPluginFactory), + ) +} + +// OCR3_1OracleArgs3 is like OCR3_1OracleArgs2 but accepts a +// ReportingPluginFactory2, whose plugins additionally get a ReadOnlyKeyValueState. +type OCR3_1OracleArgs3[RI any] struct { + BinaryNetworkEndpointFactory types.BinaryNetworkEndpoint2Factory + V2Bootstrappers []commontypes.BootstrapperLocator + ContractConfigTracker types.ContractConfigTracker + ContractTransmitter ocr3types.ContractTransmitter[RI] + Database ocr3_1types.Database + KeyValueDatabaseFactory ocr3_1types.KeyValueDatabaseFactory + LocalConfig types.LocalConfig + Logger commontypes.Logger + MetricsRegisterer prometheus.Registerer + MonitoringEndpoint commontypes.MonitoringEndpoint + OffchainConfigDigester types.OffchainConfigDigester + OffchainKeyring types.OffchainKeyring + OnchainKeyring ocr3types.OnchainKeyring2[RI] + ReportingPluginFactory ocr3_1types.ReportingPluginFactory2[RI] +} + +func (OCR3_1OracleArgs3[RI]) oracleArgsMarker() {} + +func (args OCR3_1OracleArgs3[RI]) localConfig() types.LocalConfig { return args.LocalConfig } + +func (args OCR3_1OracleArgs3[RI]) runManaged(ctx context.Context) { + logger := loghelper.MakeRootLoggerWithContext(args.Logger) + managed.RunManagedOCR3_1Oracle( ctx, diff --git a/ragep2p/ragep2p.go b/ragep2p/ragep2p.go index 3ac17331..082880f1 100644 --- a/ragep2p/ragep2p.go +++ b/ragep2p/ragep2p.go @@ -235,9 +235,11 @@ func NewHost( // Start listening on the network interfaces and dialling peers. func (ho *Host) Start() error { - succeeded := false + // Armed only once we've transitioned to open, so that a rejected Start() + // never tears down a Host that someone else started. + needsTeardown := false defer func() { - if !succeeded { + if needsTeardown { ho.Close() } }() @@ -249,6 +251,7 @@ func (ho *Host) Start() error { return fmt.Errorf("cannot Start() host that has already been started") } ho.state = hostStateOpen + needsTeardown = true ho.subprocesses.Go(func() { ho.dialLoop() @@ -268,7 +271,7 @@ func (ho *Host) Start() error { return fmt.Errorf("failed to start discoverer: %w", err) } - succeeded = true + needsTeardown = false return nil } @@ -776,7 +779,11 @@ func (ho *Host) handleIncomingConnection(conn net.Conn) { } n, err := conn.Read(knck) if err != nil { - logger.Warn("Error while reading knock", commontypes.LogFields{"error": err}) + if !errors.Is(err, io.EOF) { + logger.Warn("Error while reading knock", commontypes.LogFields{"error": err}) + } else { + logger.Debug("Connection closed before knock was received", nil) + } return } if n != knock.KnockSize { diff --git a/ragep2p/ragep2pnew/ragep2p.go b/ragep2p/ragep2pnew/ragep2p.go index 7a214e86..8a7640f5 100644 --- a/ragep2p/ragep2pnew/ragep2p.go +++ b/ragep2p/ragep2pnew/ragep2p.go @@ -234,9 +234,11 @@ func NewHost( // Start listening on the network interfaces and dialling peers. func (ho *Host) Start() error { - succeeded := false + // Armed only once we've transitioned to open, so that a rejected Start() + // never tears down a Host that someone else started. + needsTeardown := false defer func() { - if !succeeded { + if needsTeardown { ho.Close() } }() @@ -248,6 +250,7 @@ func (ho *Host) Start() error { return fmt.Errorf("cannot Start() host that has already been started") } ho.state = hostStateOpen + needsTeardown = true ho.subprocesses.Go(func() { ho.dialLoop() @@ -267,7 +270,7 @@ func (ho *Host) Start() error { return fmt.Errorf("failed to start discoverer: %w", err) } - succeeded = true + needsTeardown = false return nil } @@ -865,7 +868,11 @@ func (ho *Host) handleIncomingConnection(conn net.Conn) { } n, err := conn.Read(knck) if err != nil { - logger.Warn("Error while reading knock", commontypes.LogFields{"error": err}) + if !errors.Is(err, io.EOF) { + logger.Warn("Error while reading knock", commontypes.LogFields{"error": err}) + } else { + logger.Debug("Connection closed before knock was received", nil) + } return } if n != knock.KnockSize {