From 38019a378d1f73e73269c7f8a8e501749085df79 Mon Sep 17 00:00:00 2001 From: Kostis Karantias Date: Thu, 17 Sep 2026 16:55:50 +0300 Subject: [PATCH] Alpha support for out-of-consensus key-value state reads --- 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 | 43 ++- .../internal/ocr3_1/protocol/metrics.go | 6 + .../internal/shim/metrics.go | 72 ++++ ...ocr3_1_context_marking_reporting_plugin.go | 94 +++++ .../shim/ocr3_1_read_only_key_value_state.go | 323 ++++++++++++++++++ .../ocr3_1shims/reporting_plugin_factory.go | 35 ++ offchainreporting2plus/ocr3_1types/plugin.go | 86 +++++ offchainreporting2plus/oracle.go | 55 ++- ragep2p/ragep2p.go | 9 +- ragep2p/ragep2pnew/ragep2p.go | 9 +- 18 files changed, 857 insertions(+), 71 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 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..f1f154ac 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,15 +139,21 @@ 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 } + // Closing the plugin is idempotent so that it can be deferred twice: + // once here, as a safety net for the early returns below, and once + // more further down, after the key value database has been set up, + // so that on the happy path the plugin closes *before* the database + // does. Whichever runs second is a no-op. + closeReportingPlugin := closerFunc(sync.OnceValue(reportingPlugin.Close)) defer loghelper.CloseLogError( - reportingPlugin, + closeReportingPlugin, logger, "ManagedOCR3_1Oracle: error during reportingPlugin.Close()", ) @@ -270,6 +283,20 @@ 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 loghelper.CloseLogError( + closeReportingPlugin, + logger, + "ManagedOCR3_1Oracle: error during reportingPlugin.Close()", + ) + protocol.RunOracle[RI]( ctx, &blobEndpointWrapper, @@ -285,7 +312,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), ) @@ -737,3 +767,8 @@ func (d *devNullRegisterer) MustRegister(collectors ...prometheus.Collector) { func (d *devNullRegisterer) Unregister(collector prometheus.Collector) bool { return false } + +// closerFunc adapts a func() error to io.Closer. +type closerFunc func() error + +func (f closerFunc) Close() error { return f() } diff --git a/offchainreporting2plus/internal/ocr3_1/protocol/metrics.go b/offchainreporting2plus/internal/ocr3_1/protocol/metrics.go index 6b2b1c72..2dc3dc5d 100644 --- a/offchainreporting2plus/internal/ocr3_1/protocol/metrics.go +++ b/offchainreporting2plus/internal/ocr3_1/protocol/metrics.go @@ -22,6 +22,12 @@ const ( // outcomeGenerationMetrics and stateSyncMetrics. The two variants are // registered from separate structs against the same registerer, and prometheus // only tolerates that if the name, help and label names match exactly. +// +// TODO: Rewrite this and the other ConstLabels-per-value metrics in the style +// of newReadOnlyKeyValueStateMetrics in internal/shim/metrics.go: one +// CounterVec defined once, WithLabelValues children exposed as plain Counter +// fields, and the children unregistered in Close (which unregisters the vec). +// That keeps name, help and label names in a single place. func newAttestedStateTransitionBlocksWrittenTotal( registerer prometheus.Registerer, logger commontypes.Logger, diff --git a/offchainreporting2plus/internal/shim/metrics.go b/offchainreporting2plus/internal/shim/metrics.go index 3dc12ce4..60753d3e 100644 --- a/offchainreporting2plus/internal/shim/metrics.go +++ b/offchainreporting2plus/internal/shim/metrics.go @@ -159,3 +159,75 @@ func (m *keyValueDatabaseMetrics) Close() { m.registerer.Unregister(m.committedReadWriteTransactionsTotal) m.registerer.Unregister(m.discardedReadWriteTransactionsTotal) } + +type readOnlyKeyValueStateMetrics struct { + registerer prometheus.Registerer + openReadTransactions prometheus.GaugeFunc + oldestOpenReadTransactionDuration prometheus.GaugeFunc + openedReadTransactionsTotal prometheus.Counter + refusedInsidePluginCallTotal prometheus.Counter + refusedTooManyTotal prometheus.Counter + refusedUnavailableTotal prometheus.Counter +} + +func newReadOnlyKeyValueStateMetrics( + registerer prometheus.Registerer, + logger commontypes.Logger, + openReadTransactionCount func() float64, + oldestOpenReadTransactionDurationSeconds func() float64, +) *readOnlyKeyValueStateMetrics { + openReadTransactions := prometheus.NewGaugeFunc(prometheus.GaugeOpts{ + Name: "ocr3_1_experimental_key_value_state_open_read_transactions", + Help: "The number of read transactions the reporting plugin currently holds open.", + }, openReadTransactionCount) + metricshelper.RegisterOrLogError(logger, registerer, openReadTransactions, "ocr3_1_experimental_key_value_state_open_read_transactions") + + // 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. + oldestOpenReadTransactionDuration := prometheus.NewGaugeFunc(prometheus.GaugeOpts{ + Name: "ocr3_1_experimental_key_value_state_oldest_open_read_transaction_duration_seconds", + Help: "How long the oldest read transaction the reporting plugin holds open has been open, zero if none.", + }, oldestOpenReadTransactionDurationSeconds) + metricshelper.RegisterOrLogError(logger, registerer, oldestOpenReadTransactionDuration, "ocr3_1_experimental_key_value_state_oldest_open_read_transaction_duration_seconds") + + openedReadTransactionsTotal := prometheus.NewCounter(prometheus.CounterOpts{ + Name: "ocr3_1_experimental_key_value_state_opened_read_transactions_total", + Help: "The number of read transactions opened by the reporting plugin.", + }) + metricshelper.RegisterOrLogError(logger, registerer, openedReadTransactionsTotal, "ocr3_1_experimental_key_value_state_opened_read_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. + refusedReadTransactionsTotal := prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "ocr3_1_experimental_key_value_state_refused_read_transactions_total", + Help: "The number of read transactions refused to the reporting plugin, by reason.", + }, []string{"reason"}) + metricshelper.RegisterOrLogError(logger, registerer, refusedReadTransactionsTotal, "ocr3_1_experimental_key_value_state_refused_read_transactions_total") + + return &readOnlyKeyValueStateMetrics{ + registerer, + openReadTransactions, + oldestOpenReadTransactionDuration, + openedReadTransactionsTotal, + refusedReadTransactionsTotal.WithLabelValues("inside_plugin_call"), + refusedReadTransactionsTotal.WithLabelValues("too_many"), + refusedReadTransactionsTotal.WithLabelValues("unavailable"), + } +} + +func (m *readOnlyKeyValueStateMetrics) Close() { + m.registerer.Unregister(m.openReadTransactions) + m.registerer.Unregister(m.oldestOpenReadTransactionDuration) + m.registerer.Unregister(m.openedReadTransactionsTotal) + // 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.refusedTooManyTotal) + 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..f572e048 --- /dev/null +++ b/offchainreporting2plus/internal/shim/ocr3_1_read_only_key_value_state.go @@ -0,0 +1,323 @@ +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" +) + +// It's much easier to increase these than to decrease them, so we start with +// conservative values. Talk to the maintainers if you need higher limits for +// your plugin. +const ( + // maxConcurrentKeyValueStateReadTransactions bounds how many read + // transactions the reporting plugin may hold open at once. Opening more + // blocks rather than fails, see + // [ocr3_1types.ReadOnlyKeyValueState.NewReadTransaction]. + // + // What a read transaction costs the database is dominated by how long it is + // held, not by how many are open, so this bound is not there to protect the + // database. It is a backstop against a plugin that leaks transactions: once + // it has leaked this many, further attempts fail loudly instead of quietly + // piling up. + maxConcurrentKeyValueStateReadTransactions = 16 +) + +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 ( + errTooManyKeyValueStateReadTransactions = errors.New("too many key value state read transactions are already open") + errKeyValueStateUnavailable = errors.New("key value state is unavailable") + errKeyValueStateReadInsidePluginCall = errors.New("key value state must not be read inside a ReportingPlugin method, use the method's KeyValueStateReader argument") + 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 + + // slots bounds the number of concurrently open read transactions. Sending + // claims a slot, receiving releases it. + slots chan struct{} + + 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, + make(chan struct{}, maxConcurrentKeyValueStateReadTransactions), + subprocesses.Subprocesses{}, + make(chan struct{}), + sync.Once{}, + sync.Mutex{}, + nil, + list.NewList[*keyValueStateReadTransaction](), + } + w.metrics = newReadOnlyKeyValueStateMetrics( + registerer, + logger, + func() float64 { return float64(w.openReadTransactionsSummary().count) }, + 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. +// +// Discarding out from under the plugin is safe because Read holds the +// transaction's lock across the call into the database, so a concurrent read +// either already completed or fails with +// errKeyValueStateReadTransactionDiscarded. We never wait for the plugin. +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 does not wait for the reporting plugin either, for the same reason +// Disable does not. +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 + } + + select { + case w.slots <- struct{}{}: + case <-ctx.Done(): + w.metrics.refusedTooManyTotal.Inc() + return nil, fmt.Errorf("%w: %w", errTooManyKeyValueStateReadTransactions, ctx.Err()) + } + + readTransaction, err := w.newReadTransaction() + if err != nil { + <-w.slots + w.metrics.refusedUnavailableTotal.Inc() + return nil, err + } + w.metrics.openedReadTransactionsTotal.Inc() + 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, + } + readTransaction.element = w.openReadTransactions.PushBack(readTransaction) + readTransaction.onDiscard = func() { + w.unregisterReadTransaction(readTransaction) + <-w.slots + } + return readTransaction, nil +} + +func (w *ReadOnlyKeyValueStateWrapper) unregisterReadTransaction(readTransaction *keyValueStateReadTransaction) { + w.mu.Lock() + defer w.mu.Unlock() + w.openReadTransactions.Remove(readTransaction.element) +} + +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 +} + +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 + } + return r.nilOrRawReadTransaction.Read(key) +} + +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/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/plugin.go b/offchainreporting2plus/ocr3_1types/plugin.go index 0eb73d7c..3408f9e7 100644 --- a/offchainreporting2plus/ocr3_1types/plugin.go +++ b/offchainreporting2plus/ocr3_1types/plugin.go @@ -23,6 +23,20 @@ type ReportingPluginFactory[RI any] interface { ) (ReportingPlugin[RI], ReportingPluginInfo, error) } +// ReportingPluginFactory2 is like ReportingPluginFactory but additionally +// provides a ReadOnlyKeyValueState. +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 @@ -35,6 +49,78 @@ type KeyValueStateReader interface { Read(key []byte) ([]byte, error) } +// ReadOnlyKeyValueState gives you read access to the replicated KeyValueState +// from outside your ReportingPlugin's methods, for example to answer requests +// from your own clients. +// +// Do not use it inside your ReportingPlugin's methods. Those methods are handed +// a KeyValueStateReader argument and must use that one: they run at a specific +// sequence number, and reading anything else would make them produce different +// results on different oracles, which breaks agreement between them. +type ReadOnlyKeyValueState interface { + // NewReadTransaction returns a read transaction over the replicated + // KeyValueState. You are responsible for discarding it, see + // [KeyValueStateReadTransaction]. + // + // Only a limited number of read transactions may be open at once. Reaching + // that limit is not an error: NewReadTransaction blocks until one of the + // open transactions is discarded or ctx is done, and only in the latter + // case returns an error. Concurrent request handlers therefore queue + // rather than fail. Talk to the maintainers if you need the limit raised. + // + // NewReadTransaction also returns an error if the KeyValueState is + // temporarily unreadable because this oracle is starting up, shutting down + // or catching up with the other oracles, or if you call it from inside one + // of your ReportingPlugin's methods. Every one of these but the last can + // happen in normal operation: fail the request you were serving and try + // again later. + NewReadTransaction(ctx context.Context) (KeyValueStateReadTransaction, error) +} + +// KeyValueStateReadTransaction provides read access to the replicated +// KeyValueState at a fixed point in time. +// +// You must discard every transaction you open. The usual way is to write +// +// readTransaction, err := keyValueState.NewReadTransaction(ctx) +// if err != nil { +// return err +// } +// defer readTransaction.Discard() +// +// immediately, before doing anything else. Until you discard it, this oracle +// cannot reclaim the storage holding the older versions of the data your +// transaction can still see. A transaction you never discard will make this +// oracle's database grow without bound for as long as your plugin runs, and +// will eventually slow down or stall the protocol. Nothing will stop you: this +// is your responsibility, and we will only log about it. +// +// Hold a transaction for as short a time as you can. Read what you need, +// discard it, and do the rest of your work afterwards. In particular, do not +// call out over the network, wait on a lock, or block on anything else while +// you hold one. The reads themselves take microseconds; if you hold a +// transaction for longer than a round, something in your code is waiting when +// it should not be. +// +// A transaction is safe for concurrent use. +// +// There is one case in which a transaction stops working before you discard it: +// when the protocol instance shuts down, because the underlying database is +// closed at that point. Read then returns an error. +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. + // Reading after it returns an error. + Discard() +} + // Deprecated: Use KeyValueStateReadWriter instead. type KeyValueReadWriter = KeyValueStateReadWriter diff --git a/offchainreporting2plus/oracle.go b/offchainreporting2plus/oracle.go index f05b0014..13479ac9 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,58 @@ 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_1OracleArgs109Alpha is like OCR3_1OracleArgs2 but accepts a +// ReportingPluginFactory2, whose plugins additionally get a ReadOnlyKeyValueState. +// +// WARNING: ALPHA. MUST NOT BE USED IN PRODUCTION. +// +// The ReadOnlyKeyValueState this exposes is under active development. Its +// interface, semantics and limits may change or be removed without notice +// and without a deprecation period. +type OCR3_1OracleArgs109Alpha[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_1OracleArgs109Alpha[RI]) oracleArgsMarker() {} + +func (args OCR3_1OracleArgs109Alpha[RI]) localConfig() types.LocalConfig { return args.LocalConfig } + +func (args OCR3_1OracleArgs109Alpha[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..c0617d46 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 } diff --git a/ragep2p/ragep2pnew/ragep2p.go b/ragep2p/ragep2pnew/ragep2p.go index 7a214e86..ce652b2a 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 }