diff --git a/core/scripts/go.mod b/core/scripts/go.mod index fea911afc19..b5389eaee37 100644 --- a/core/scripts/go.mod +++ b/core/scripts/go.mod @@ -49,7 +49,7 @@ require ( github.com/smartcontractkit/chainlink-ccip/chains/evm v0.0.0-20260624154507-ea7ff77a0ddb github.com/smartcontractkit/chainlink-common v0.11.2-0.20260727165036-9952de44dbab github.com/smartcontractkit/chainlink-common/keystore v1.3.0 - github.com/smartcontractkit/chainlink-data-streams v1.0.0 + github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260728111445-96c471be2872 github.com/smartcontractkit/chainlink-evm/gethwrappers v0.0.0-20260713161920-de075095648b diff --git a/core/scripts/go.sum b/core/scripts/go.sum index 493f5598b38..81d0493a190 100644 --- a/core/scripts/go.sum +++ b/core/scripts/go.sum @@ -1586,8 +1586,8 @@ github.com/smartcontractkit/chainlink-common/keystore v1.3.0 h1:V05Rp9/dTc4Wyips github.com/smartcontractkit/chainlink-common/keystore v1.3.0/go.mod h1:vHV8BGm6TN7jBbMsWxq1Hqm3HbCtYFwzvKS0CCczxG8= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 h1:uWEwl7i2ryuRVoV4DmIKm6mqYevf1lH/8cQYhw/JXko= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= -github.com/smartcontractkit/chainlink-data-streams v1.0.0 h1:W+RfzuZHVt50ihlsB7+XpXiBw1v5hRks6DUK2HHH7A4= -github.com/smartcontractkit/chainlink-data-streams v1.0.0/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= +github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf h1:ovwq9+AKlW1DxXVT7Go1tmX567usqAgA+ON2WfzISig= +github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 h1:mzbvXxdbE/96Pdj1zyPKzf25ZlDR48+iTTDTbaITvmk= github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54/go.mod h1:sz/YCiLs8i/V57WISALB7ywNjxW24sj0hi+DE4kzv6A= github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260728111445-96c471be2872 h1:0vivai99gZjV92RLlxnIhUBTODg08JfdBD8wLfN8A5A= diff --git a/core/services/llo/delegate.go b/core/services/llo/delegate.go index 142a51ff2f1..736a9a377e0 100644 --- a/core/services/llo/delegate.go +++ b/core/services/llo/delegate.go @@ -7,11 +7,6 @@ import ( "strconv" "github.com/prometheus/client_golang/prometheus" - ocrcommontypes "github.com/smartcontractkit/libocr/commontypes" - ocr2plus "github.com/smartcontractkit/libocr/offchainreporting2plus" - "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3shims" - "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" - ocr2types "github.com/smartcontractkit/libocr/offchainreporting2plus/types" "gopkg.in/guregu/null.v4" "github.com/smartcontractkit/chainlink-common/pkg/logger" @@ -22,12 +17,20 @@ import ( "github.com/smartcontractkit/chainlink-data-streams/llo/retirement" "github.com/smartcontractkit/chainlink-data-streams/llo/transmitter" llov30 "github.com/smartcontractkit/chainlink-data-streams/llo/v30" + llov31 "github.com/smartcontractkit/chainlink-data-streams/llo/v31" + ocrcommontypes "github.com/smartcontractkit/libocr/commontypes" + ocr2plus "github.com/smartcontractkit/libocr/offchainreporting2plus" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3_1types" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3shims" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" + ocr2types "github.com/smartcontractkit/libocr/offchainreporting2plus/types" corelogger "github.com/smartcontractkit/chainlink/v2/core/logger" "github.com/smartcontractkit/chainlink/v2/core/services/job" "github.com/smartcontractkit/chainlink/v2/core/services/llo/observation" "github.com/smartcontractkit/chainlink/v2/core/services/llo/telem" "github.com/smartcontractkit/chainlink/v2/core/services/ocr3/promwrapper" + promwrapper31 "github.com/smartcontractkit/chainlink/v2/core/services/ocr3_1/promwrapper" "github.com/smartcontractkit/chainlink/v2/core/services/streams" "github.com/smartcontractkit/chainlink/v2/core/services/telemetry" ) @@ -44,8 +47,13 @@ type delegate struct { cfg DelegateConfig reportCodecs map[llotypes.ReportFormat]llocommon.ReportCodec - src llov30.ShouldRetireCache - ds llov30.DataSource + // src is the shared ShouldRetireCache. llov30.ShouldRetireCache and + // llov31.ShouldRetireCache have identical method sets, so this value serves + // both versions. + src llov30.ShouldRetireCache + // ds is the shared LLO data source (llocommon.DataSource); v30 and v31 both + // consume it, lifecycle gating is driven by the round's DSOpts. + ds llocommon.DataSource telem telem.TelemeterService oracles []Closer @@ -86,6 +94,16 @@ type DelegateConfig struct { OnchainKeyring ocr3types.OnchainKeyring[llotypes.ReportInfo] LocalConfig ocr2types.LocalConfig NewOCR3DB func(pluginID int32) ocr3types.Database + + // OCR3.1 (only required when OCR31 is true; see chainlink-data-streams + // llo/config.PluginConfig.OCRVersion) + OCR31 bool + // BinaryNetworkEndpoint2Factory is the OCR3.1 ("2") network endpoint factory + // (peerWrapper.Peer3_1). Required when OCR31 is true. + BinaryNetworkEndpoint2Factory ocr2types.BinaryNetworkEndpoint2Factory + // KeyValueDatabaseFactory provides the replicated per-configDigest key-value + // store the OCR3.1 protocol requires. Required when OCR31 is true. + KeyValueDatabaseFactory ocr3_1types.KeyValueDatabaseFactory } func NewDelegate(cfg DelegateConfig) (job.ServiceCtx, error) { @@ -105,6 +123,14 @@ func NewDelegate(cfg DelegateConfig) (job.ServiceCtx, error) { if cfg.ShouldRetireCache == nil { return nil, errors.New("ShouldRetireCache must not be nil") } + if cfg.OCR31 { + if cfg.KeyValueDatabaseFactory == nil { + return nil, errors.New("KeyValueDatabaseFactory must not be nil when running OCR3.1") + } + if cfg.BinaryNetworkEndpoint2Factory == nil { + return nil, errors.New("BinaryNetworkEndpoint2Factory must not be nil when running OCR3.1") + } + } var codecLggr logger.Logger if cfg.ReportingPluginConfig.VerboseLogging { codecLggr = logger.Named(lggr, "ReportCodecs") @@ -124,11 +150,7 @@ func NewDelegate(cfg DelegateConfig) (job.ServiceCtx, error) { SampleTelemetry: cfg.SampleTelemetry, }) - ds := observation.NewDataSource( - logger.Named(lggr, "DataSource"), - cfg.Registry, - t, - ) + ds := observation.NewDataSource(logger.Named(lggr, "DataSource"), cfg.Registry, t) notifier, ok := cfg.ContractTransmitter.(transmitter.TransmitNotifier) if ok { @@ -166,43 +188,13 @@ func (d *delegate) Start(ctx context.Context) error { // This is a performance optimization }) - oracle, err := ocr2plus.NewOracle(ocr2plus.OCR3OracleArgs2[llotypes.ReportInfo]{ - BinaryNetworkEndpointFactory: d.cfg.BinaryNetworkEndpointFactory, - V2Bootstrappers: d.cfg.V2Bootstrappers, - ContractConfigTracker: configTracker, - ContractTransmitter: d.cfg.ContractTransmitter, - Database: d.cfg.NewOCR3DB(int32(i)), // //nolint:gosec // G115 // impossible due to check on line 119 - LocalConfig: d.cfg.LocalConfig, - Logger: ocrLogger, - MonitoringEndpoint: d.cfg.OCR3MonitoringEndpoint, - OffchainConfigDigester: d.cfg.OffchainConfigDigester, - OffchainKeyring: d.cfg.OffchainKeyring, - OnchainKeyring: ocr3shims.OnchainKeyringAsOnchainKeyring2(d.cfg.OnchainKeyring), - ReportingPluginFactory: promwrapper.NewReportingPluginFactory( - llov30.NewPluginFactory( - llov30.PluginFactoryParams{ - Config: d.cfg.ReportingPluginConfig, - PredecessorRetirementReportCache: psrrc, - ShouldRetireCache: d.src, - RetirementReportCodec: d.cfg.RetirementReportCodec, - ChannelDefinitionCache: d.cfg.ChannelDefinitionCache, - DataSource: d.ds, - Logger: logger.Named(lggr, "ReportingPlugin"), - OnchainConfigCodec: llocommon.EVMOnchainConfigCodec{}, - ReportCodecs: d.reportCodecs, - OutcomeTelemetryCh: d.telem.GetOutcomeTelemetryCh(), - ReportTelemetryCh: d.telem.GetReportTelemetryCh(), - DonID: d.cfg.DonID, - }, - ), - lggr, - "", - d.cfg.ChainID, - "llo", - ), - MetricsRegisterer: prometheus.WrapRegistererWith(map[string]string{"job_name": d.cfg.JobName.ValueOrZero()}, prometheus.DefaultRegisterer), - }) - + var oracle ocr2plus.Oracle + var err error + if d.cfg.OCR31 { + oracle, err = d.newOracleV31(i, configTracker, lggr, ocrLogger, psrrc) + } else { + oracle, err = d.newOracleV30(i, configTracker, lggr, ocrLogger, psrrc) + } if err != nil { return fmt.Errorf("%w: failed to create new OCR oracle", err) } @@ -216,6 +208,89 @@ func (d *delegate) Start(ctx context.Context) error { }) } +// newOracleV30 builds an OCR3.0 oracle running the llo/v30 reporting plugin. +func (d *delegate) newOracleV30(i int, configTracker ocr2types.ContractConfigTracker, lggr logger.Logger, ocrLogger ocrcommontypes.Logger, psrrc llocommon.PredecessorRetirementReportCache) (ocr2plus.Oracle, error) { + return ocr2plus.NewOracle(ocr2plus.OCR3OracleArgs2[llotypes.ReportInfo]{ + BinaryNetworkEndpointFactory: d.cfg.BinaryNetworkEndpointFactory, + V2Bootstrappers: d.cfg.V2Bootstrappers, + ContractConfigTracker: configTracker, + ContractTransmitter: d.cfg.ContractTransmitter, + Database: d.cfg.NewOCR3DB(int32(i)), //nolint:gosec // G115 // impossible due to ContractConfigTrackers length check + LocalConfig: d.cfg.LocalConfig, + Logger: ocrLogger, + MonitoringEndpoint: d.cfg.OCR3MonitoringEndpoint, + OffchainConfigDigester: d.cfg.OffchainConfigDigester, + OffchainKeyring: d.cfg.OffchainKeyring, + OnchainKeyring: ocr3shims.OnchainKeyringAsOnchainKeyring2(d.cfg.OnchainKeyring), + ReportingPluginFactory: promwrapper.NewReportingPluginFactory( + llov30.NewPluginFactory( + llov30.PluginFactoryParams{ + Config: d.cfg.ReportingPluginConfig, + PredecessorRetirementReportCache: psrrc, + ShouldRetireCache: d.src, + RetirementReportCodec: d.cfg.RetirementReportCodec, + ChannelDefinitionCache: d.cfg.ChannelDefinitionCache, + DataSource: d.ds, + Logger: logger.Named(lggr, "ReportingPlugin"), + OnchainConfigCodec: llocommon.EVMOnchainConfigCodec{}, + ReportCodecs: d.reportCodecs, + OutcomeTelemetryCh: d.telem.GetOutcomeTelemetryCh(), + ReportTelemetryCh: d.telem.GetReportTelemetryCh(), + DonID: d.cfg.DonID, + }, + ), + lggr, + "", + d.cfg.ChainID, + "llo", + ), + MetricsRegisterer: prometheus.WrapRegistererWith(map[string]string{"job_name": d.cfg.JobName.ValueOrZero()}, prometheus.DefaultRegisterer), + }) +} + +// newOracleV31 builds an OCR3.1 oracle running the llo/v31 reporting plugin. It +// differs from v30 by the OCR3.1 oracle args (OCR3_1OracleArgs2), the "2" +// network endpoint factory, and the required replicated KeyValueDatabaseFactory. +func (d *delegate) newOracleV31(i int, configTracker ocr2types.ContractConfigTracker, lggr logger.Logger, ocrLogger ocrcommontypes.Logger, psrrc llocommon.PredecessorRetirementReportCache) (ocr2plus.Oracle, error) { + factory := promwrapper31.NewReportingPluginFactory( + llov31.NewPluginFactory(llov31.PluginFactoryParams{ + Config: llov31.Config{VerboseLogging: d.cfg.ReportingPluginConfig.VerboseLogging}, + PredecessorRetirementReportCache: psrrc, + ShouldRetireCache: d.src, + RetirementReportCodec: d.cfg.RetirementReportCodec, + ChannelDefinitionCache: d.cfg.ChannelDefinitionCache, + DataSource: d.ds, + Logger: logger.Named(lggr, "ReportingPlugin"), + OnchainConfigCodec: llocommon.EVMOnchainConfigCodec{}, + ReportCodecs: d.reportCodecs, + OutcomeTelemetryCh: d.telem.GetOutcomeTelemetryCh(), + ReportTelemetryCh: d.telem.GetReportTelemetryCh(), + DonID: d.cfg.DonID, + BlobThreshold: 0, // 0 => llov31.DefaultBlobThreshold + }), + lggr, + "", + d.cfg.ChainID, + "llo", + ) + return ocr2plus.NewOracle(ocr2plus.OCR3_1OracleArgs2[llotypes.ReportInfo]{ + BinaryNetworkEndpointFactory: d.cfg.BinaryNetworkEndpoint2Factory, + V2Bootstrappers: d.cfg.V2Bootstrappers, + ContractConfigTracker: configTracker, + ContractTransmitter: d.cfg.ContractTransmitter, + Database: d.cfg.NewOCR3DB(int32(i)), //nolint:gosec // G115 // impossible due to ContractConfigTrackers length check + KeyValueDatabaseFactory: d.cfg.KeyValueDatabaseFactory, + LocalConfig: d.cfg.LocalConfig, + Logger: ocrLogger, + MonitoringEndpoint: d.cfg.OCR3MonitoringEndpoint, + OffchainConfigDigester: d.cfg.OffchainConfigDigester, + OffchainKeyring: d.cfg.OffchainKeyring, + OnchainKeyring: ocr3shims.OnchainKeyringAsOnchainKeyring2(d.cfg.OnchainKeyring), + ReportingPluginFactory: factory, + MetricsRegisterer: prometheus.WrapRegistererWith(map[string]string{"job_name": d.cfg.JobName.ValueOrZero()}, prometheus.DefaultRegisterer), + }) +} + func (d *delegate) Close() error { return d.StopOnce("LLODelegate", func() (merr error) { for _, oracle := range d.oracles { diff --git a/core/services/llo/observation/data_source.go b/core/services/llo/observation/data_source.go index dfb80a0c410..7fae4236894 100644 --- a/core/services/llo/observation/data_source.go +++ b/core/services/llo/observation/data_source.go @@ -17,7 +17,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink-common/pkg/services" llocommon "github.com/smartcontractkit/chainlink-data-streams/llo/common" - llov30 "github.com/smartcontractkit/chainlink-data-streams/llo/v30" + "github.com/smartcontractkit/chainlink/v2/core/services/llo/telem" "github.com/smartcontractkit/chainlink/v2/core/services/pipeline" "github.com/smartcontractkit/chainlink/v2/core/services/streams" ) @@ -147,8 +147,6 @@ func (e *ObservationFailedError) Unwrap() error { return e.inner } -var _ llov30.DataSource = &dataSource{} - type dataSource struct { wg sync.WaitGroup lggr logger.Logger @@ -165,7 +163,12 @@ type dataSource struct { loopWakeCh chan struct{} } -func NewDataSource(lggr logger.Logger, registry Registry, t Telemeter) llov30.DataSource { +var _ llocommon.DataSource = &dataSource{} + +// NewDataSource returns the shared LLO data source. llo/v30 and llo/v31 both +// consume llocommon.DataSource, so a single implementation serves both OCR +// protocol versions; lifecycle gating is driven by opts.LifeCycleStage(). +func NewDataSource(lggr logger.Logger, registry Registry, t Telemeter) llocommon.DataSource { return newDataSource(lggr, registry, t) } @@ -192,13 +195,39 @@ func (d *dataSource) signalObservationLoopWake() { } } -// Observe starts or refreshes the background observation loop for the plugin's stream set, then fills streamValues -// from the in-memory cache (backed by pipeline observations registered for each stream ID). -func (d *dataSource) Observe(ctx context.Context, streamValues llocommon.StreamValues, opts llov30.DSOpts) error { +// Observe gates the background observation loop on the round's lifecycle stage +// (only a Production instance runs pipeline observations), then fills +// streamValues from the in-memory cache. The stage is carried by opts and is +// derived by each plugin version from its own state (v30 from the previous +// outcome, v31 from the KeyValueState), so a single implementation serves both. +func (d *dataSource) Observe(ctx context.Context, streamValues llocommon.StreamValues, opts llocommon.DSOpts) error { + return d.observe(ctx, streamValues, opts, d.inProduction(opts)) +} + +// inProduction reports whether this OCR instance is the Production instance (the +// only one that should run pipeline observations). +func (d *dataSource) inProduction(opts llocommon.DSOpts) bool { + if opts == nil { + // setObservableStreams logs the nil-opts case; stay silent here to avoid + // a duplicate warning per round. + return false + } + if opts.LifeCycleStage() != llocommon.LifeCycleStageProduction { + d.lggr.Debugw("Observe: LLO OCR instance is not in production lifecycle stage", + "configDigest", opts.ConfigDigest().String(), "stage", opts.LifeCycleStage()) + return false + } + return true +} + +// observe starts or refreshes the background observation loop for the plugin's stream set, then fills streamValues +// from the in-memory cache (backed by pipeline observations registered for each stream ID). inProduction gates the +// loop: when false the observable stream set is cleared and no pipelines run this round. +func (d *dataSource) observe(ctx context.Context, streamValues llocommon.StreamValues, opts telem.DSOpts, inProduction bool) error { // Observation loop logic { // setObservableStreams copies stream IDs and deadline into internal state (the plugin's map is not retained). - d.setObservableStreams(ctx, streamValues, opts) + d.setObservableStreams(ctx, streamValues, opts, inProduction) if !d.observationLoopStarted.Load() { loopStartedCh := make(chan struct{}) @@ -283,7 +312,7 @@ func (d *dataSource) startObservationLoop(loopStartedCh chan struct{}) { startTS := time.Now() ctx, cancel := context.WithTimeout(stopChanCtx, osv.observationTimeout) - lggr := logger.With(d.lggr, "observationTimestamp", osv.opts.ObservationTimestamp(), "configDigest", osv.opts.ConfigDigest(), "seqNr", osv.opts.OutCtx().SeqNr) + lggr := logger.With(d.lggr, "observationTimestamp", osv.opts.ObservationTimestamp(), "configDigest", osv.opts.ConfigDigest(), "seqNr", osv.opts.SeqNr()) var mu sync.Mutex var wg sync.WaitGroup @@ -490,30 +519,24 @@ func (d *dataSource) Close() error { } type observableStreamValues struct { - opts llov30.DSOpts + opts telem.DSOpts streamValues llocommon.StreamValues observationTimeout time.Duration } -// setObservableStreams updates the stream set and observation deadline (T) used by the background loop when in production. -func (d *dataSource) setObservableStreams(ctx context.Context, streamValues llocommon.StreamValues, opts llov30.DSOpts) { +// setObservableStreams updates the stream set and observation deadline (T) used by the background loop. When +// inProduction is false (v30 non-production instance) the observable set is left unchanged/empty so no pipelines run. +func (d *dataSource) setObservableStreams(ctx context.Context, streamValues llocommon.StreamValues, opts telem.DSOpts, inProduction bool) { if opts == nil || len(streamValues) == 0 { d.lggr.Warnw("setObservableStreams: no observable streams to set", "opts", opts, "observable_streams", len(streamValues)) return } - outCtx := opts.OutCtx() - outcome, err := opts.OutcomeCodec().Decode(outCtx.PreviousOutcome) - if err != nil { - d.lggr.Errorw("setObservableStreams: failed to decode outcome", "error", err) - return - } - - if outcome.LifeCycleStage != llocommon.LifeCycleStageProduction { + if !inProduction { d.lggr.Debugw( "setObservableStreams: LLO OCR instance is not in production lifecycle stage", - "configDigest", opts.ConfigDigest().String(), "stage", outcome.LifeCycleStage) + "configDigest", opts.ConfigDigest().String()) return } diff --git a/core/services/llo/observation/data_source_test.go b/core/services/llo/observation/data_source_test.go index 0be17119d5f..d7fd571e418 100644 --- a/core/services/llo/observation/data_source_test.go +++ b/core/services/llo/observation/data_source_test.go @@ -16,7 +16,6 @@ import ( promtest "github.com/prometheus/client_golang/prometheus/testutil" "github.com/shopspring/decimal" - "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" ocr2types "github.com/smartcontractkit/libocr/offchainreporting2plus/types" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -24,7 +23,6 @@ import ( llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" llocommon "github.com/smartcontractkit/chainlink-data-streams/llo/common" - llov30 "github.com/smartcontractkit/chainlink-data-streams/llo/v30" "github.com/smartcontractkit/chainlink/v2/core/bridges" clhttptest "github.com/smartcontractkit/chainlink/v2/core/internal/testutils/httptest" "github.com/smartcontractkit/chainlink/v2/core/internal/testutils/pgtest" @@ -100,9 +98,9 @@ func makeStreamValues(streamIDs ...llotypes.StreamID) llocommon.StreamValues { type mockOpts struct { verboseLogging bool seqNr uint64 - outCtx ocr3types.OutcomeContext configDigest ocr2types.ConfigDigest observationTimestamp time.Time + lifeCycleStage llotypes.LifeCycleStage } func (m *mockOpts) VerboseLogging() bool { return m.verboseLogging } @@ -112,12 +110,6 @@ func (m *mockOpts) SeqNr() uint64 { } return m.seqNr } -func (m *mockOpts) OutCtx() ocr3types.OutcomeContext { - if m.outCtx.SeqNr == 0 { - return ocr3types.OutcomeContext{SeqNr: 1042, PreviousOutcome: []byte("foo")} - } - return m.outCtx -} func (m *mockOpts) ConfigDigest() ocr2types.ConfigDigest { if m.configDigest.Hex() == "" { return ocr2types.ConfigDigest{6, 5, 4} @@ -130,19 +122,11 @@ func (m *mockOpts) ObservationTimestamp() time.Time { } return m.observationTimestamp } -func (m *mockOpts) OutcomeCodec() llov30.OutcomeCodec { - return mockOutputCodec{} -} - -type mockOutputCodec struct{} - -func (oc mockOutputCodec) Encode(outcome llov30.Outcome) (ocr3types.Outcome, error) { - return ocr3types.Outcome{}, nil -} -func (oc mockOutputCodec) Decode(encoded ocr3types.Outcome) (outcome llov30.Outcome, err error) { - return llov30.Outcome{ - LifeCycleStage: llocommon.LifeCycleStageProduction, - }, nil +func (m *mockOpts) LifeCycleStage() llotypes.LifeCycleStage { + if m.lifeCycleStage == "" { + return llocommon.LifeCycleStageProduction + } + return m.lifeCycleStage } type mockTelemeter struct { @@ -155,19 +139,19 @@ type v3PremiumLegacyPacket struct { run *pipeline.Run trrs pipeline.TaskRunResults streamID uint32 - opts llov30.DSOpts + opts telem.DSOpts val llocommon.StreamValue err error } var _ Telemeter = &mockTelemeter{} -func (m *mockTelemeter) EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.TaskRunResults, streamID uint32, opts llov30.DSOpts, val llocommon.StreamValue, err error) { +func (m *mockTelemeter) EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.TaskRunResults, streamID uint32, opts telem.DSOpts, val llocommon.StreamValue, err error) { m.mu.Lock() defer m.mu.Unlock() m.v3PremiumLegacyPackets = append(m.v3PremiumLegacyPackets, v3PremiumLegacyPacket{run, trrs, streamID, opts, val, err}) } -func (m *mockTelemeter) MakeObservationScopedTelemetryCh(opts llov30.DSOpts, size int) (ch chan<- any) { +func (m *mockTelemeter) MakeObservationScopedTelemetryCh(opts telem.DSOpts, size int) (ch chan<- any) { m.mu.Lock() defer m.mu.Unlock() m.ch = make(chan any, size) @@ -941,3 +925,35 @@ result3 -> result3_parse -> multiply3; require.NoError(b, err) ds.Close() } + +// Test_DataSource_inProduction covers the lifecycle gate: only a Production +// instance observes; staging/retired/unknown and nil opts do not. +func Test_DataSource_inProduction(t *testing.T) { + t.Parallel() + reg := &mockRegistry{pipelines: make(map[streams.StreamID]*mockPipeline)} + ds := newDataSource(logger.NullLogger, reg, telem.NullTelemeter) + defer ds.Close() + + require.True(t, ds.inProduction(&mockOpts{lifeCycleStage: llocommon.LifeCycleStageProduction})) + require.False(t, ds.inProduction(&mockOpts{lifeCycleStage: llocommon.LifeCycleStageStaging})) + require.False(t, ds.inProduction(&mockOpts{lifeCycleStage: llocommon.LifeCycleStageRetired})) + require.False(t, ds.inProduction(nil)) +} + +// Test_DataSource_StagingDoesNotObserve asserts a non-Production instance runs +// no pipelines and returns unset stream values. +func Test_DataSource_StagingDoesNotObserve(t *testing.T) { + t.Parallel() + reg := &mockRegistry{pipelines: make(map[streams.StreamID]*mockPipeline)} + reg.pipelines[1] = pipelineForStream(1, 1, big.NewInt(42), nil) + ds := newDataSource(logger.NullLogger, reg, telem.NullTelemeter) + defer ds.Close() + + ctx, cancel := context.WithTimeout(t.Context(), observationTimeout) + defer cancel() + vals := makeStreamValues(1) + require.NoError(t, ds.Observe(ctx, vals, &mockOpts{lifeCycleStage: llocommon.LifeCycleStageStaging})) + + require.Nil(t, vals[1], "staging instance must not populate stream values") + require.Zero(t, reg.pipelines[1].runCount.Load(), "staging instance must not run pipelines") +} diff --git a/core/services/llo/observation/observation_context.go b/core/services/llo/observation/observation_context.go index 52c286955b0..d0b80242b58 100644 --- a/core/services/llo/observation/observation_context.go +++ b/core/services/llo/observation/observation_context.go @@ -14,7 +14,6 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/logger" llocommon "github.com/smartcontractkit/chainlink-data-streams/llo/common" - llov30 "github.com/smartcontractkit/chainlink-data-streams/llo/v30" "github.com/smartcontractkit/chainlink/v2/core/services/llo/telem" "github.com/smartcontractkit/chainlink/v2/core/services/pipeline" @@ -30,7 +29,7 @@ import ( var _ ObservationContext = (*observationContext)(nil) type ObservationContext interface { //nolint:revive // ObservationContext is the established interface name in this package - Observe(ctx context.Context, streamID streams.StreamID, opts llov30.DSOpts) (val llocommon.StreamValue, err error) + Observe(ctx context.Context, streamID streams.StreamID, opts telem.DSOpts) (val llocommon.StreamValue, err error) } type execution struct { @@ -59,7 +58,7 @@ func newObservationContext(l logger.Logger, r Registry, t Telemeter) *observatio return &observationContext{l, r, t, sync.Mutex{}, make(map[streams.Pipeline]*execution)} } -func (oc *observationContext) Observe(ctx context.Context, streamID streams.StreamID, opts llov30.DSOpts) (val llocommon.StreamValue, err error) { +func (oc *observationContext) Observe(ctx context.Context, streamID streams.StreamID, opts telem.DSOpts) (val llocommon.StreamValue, err error) { run, trrs, err := oc.run(ctx, streamID) observationFinishedAt := time.Now() if err != nil { diff --git a/core/services/llo/observation/types.go b/core/services/llo/observation/types.go index 6b5e1e14de1..656aad3eb3f 100644 --- a/core/services/llo/observation/types.go +++ b/core/services/llo/observation/types.go @@ -4,7 +4,7 @@ import ( "context" llocommon "github.com/smartcontractkit/chainlink-data-streams/llo/common" - llov30 "github.com/smartcontractkit/chainlink-data-streams/llo/v30" + "github.com/smartcontractkit/chainlink/v2/core/services/llo/telem" "github.com/smartcontractkit/chainlink/v2/core/services/pipeline" "github.com/smartcontractkit/chainlink/v2/core/services/streams" ) @@ -14,8 +14,8 @@ type Registry interface { } type Telemeter interface { - EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.TaskRunResults, streamID uint32, opts llov30.DSOpts, val llocommon.StreamValue, err error) - MakeObservationScopedTelemetryCh(opts llov30.DSOpts, size int) (ch chan<- any) + EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.TaskRunResults, streamID uint32, opts telem.DSOpts, val llocommon.StreamValue, err error) + MakeObservationScopedTelemetryCh(opts telem.DSOpts, size int) (ch chan<- any) CaptureEATelemetry() bool CaptureObservationTelemetry() bool } diff --git a/core/services/llo/telem/telemetry.go b/core/services/llo/telem/telemetry.go index b956e9126e7..db50172f1bd 100644 --- a/core/services/llo/telem/telemetry.go +++ b/core/services/llo/telem/telemetry.go @@ -15,7 +15,6 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/types/mercury" llocommon "github.com/smartcontractkit/chainlink-data-streams/llo/common" "github.com/smartcontractkit/chainlink-data-streams/llo/reportcodecs/evm" - llov30 "github.com/smartcontractkit/chainlink-data-streams/llo/v30" "github.com/smartcontractkit/chainlink/v2/core/services/ocrcommon" "github.com/smartcontractkit/chainlink/v2/core/services/pipeline" @@ -27,9 +26,14 @@ import ( const adapterLWBAErrorName = "AdapterLWBAError" +// DSOpts is the shared, version-agnostic LLO data-source options (llo/v30 and +// llo/v31 both use llocommon.DSOpts). Aliased here so the telemetry and +// observation paths keep referring to telem.DSOpts. +type DSOpts = llocommon.DSOpts + type Telemeter interface { - EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.TaskRunResults, streamID uint32, opts llov30.DSOpts, val llocommon.StreamValue, err error) - MakeObservationScopedTelemetryCh(opts llov30.DSOpts, size int) (ch chan<- any) + EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.TaskRunResults, streamID uint32, opts DSOpts, val llocommon.StreamValue, err error) + MakeObservationScopedTelemetryCh(opts DSOpts, size int) (ch chan<- any) GetOutcomeTelemetryCh() chan<- *llocommon.LLOOutcomeTelemetry GetReportTelemetryCh() chan<- *llocommon.LLOReportTelemetry CaptureEATelemetry() bool @@ -146,7 +150,7 @@ type telemeter struct { sampler *sampler } -func (t *telemeter) EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.TaskRunResults, streamID uint32, opts llov30.DSOpts, val llocommon.StreamValue, err error) { +func (t *telemeter) EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.TaskRunResults, streamID uint32, opts DSOpts, val llocommon.StreamValue, err error) { if t.Ready() != nil { // This should never happen, telemeter should always be started BEFORE // the oracle and closed AFTER it @@ -170,7 +174,7 @@ func (t *telemeter) EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.Task type telemetryCollectionContext struct { in <-chan any - opts llov30.DSOpts + opts DSOpts } // MakeObservationScopedTelemetryCh reads telem packets from the returned channel and sends them @@ -182,7 +186,7 @@ type telemetryCollectionContext struct { // // It is necessary to make a new channel for every Observation call because it // closes over DSOpts which is scoped to that call only. -func (t *telemeter) MakeObservationScopedTelemetryCh(opts llov30.DSOpts, size int) chan<- any { +func (t *telemeter) MakeObservationScopedTelemetryCh(opts DSOpts, size int) chan<- any { if !t.captureObservationTelemetry && !t.captureEATelemetry { return nil } @@ -362,7 +366,7 @@ func (t *telemeter) enqueueTelemetry(digest string, seqNr uint64, typ synchroniz } } -func (t *telemeter) prepareObservationTelemetry(p any, opts llov30.DSOpts) { +func (t *telemeter) prepareObservationTelemetry(p any, opts DSOpts) { var telemType synchronization.TelemetryType var msg proto.Message switch v := p.(type) { @@ -447,7 +451,7 @@ func (t *telemeter) prepareV3PremiumLegacyTelemetry(d *TelemetryPipeline) { Version: uint32(1000 + mercury.REPORT_V3), // add 1000 to distinguish between legacy feeds, this can be changed if necessary DonId: t.donID, } - epoch, round, err := evm.SeqNrToEpochAndRound(d.opts.OutCtx().SeqNr) + epoch, round, err := evm.SeqNrToEpochAndRound(d.opts.SeqNr()) if err != nil { t.eng.Warnw("Failed to convert sequence number to epoch and round", "err", err) } else { @@ -470,7 +474,7 @@ func (t *telemeter) TrackSeqNr(digest types.ConfigDigest, seqNr uint64) { } type TelemetryObserve struct { - Opts llov30.DSOpts + Opts DSOpts Telemetry any } @@ -478,7 +482,7 @@ type TelemetryPipeline struct { run *pipeline.Run trrs pipeline.TaskRunResults streamID uint32 - opts llov30.DSOpts + opts DSOpts val llocommon.StreamValue dpInvariantViolationDetected bool } @@ -487,9 +491,9 @@ var NullTelemeter TelemeterService = &nullTelemeter{} type nullTelemeter struct{} -func (t *nullTelemeter) EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.TaskRunResults, streamID uint32, opts llov30.DSOpts, val llocommon.StreamValue, err error) { +func (t *nullTelemeter) EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.TaskRunResults, streamID uint32, opts DSOpts, val llocommon.StreamValue, err error) { } -func (t *nullTelemeter) MakeObservationScopedTelemetryCh(opts llov30.DSOpts, size int) (ch chan<- any) { +func (t *nullTelemeter) MakeObservationScopedTelemetryCh(opts DSOpts, size int) (ch chan<- any) { return nil } func (t *nullTelemeter) GetOutcomeTelemetryCh() chan<- *llocommon.LLOOutcomeTelemetry { diff --git a/core/services/llo/telem/telemetry_test.go b/core/services/llo/telem/telemetry_test.go index 682c80374a9..c6036c3d85f 100644 --- a/core/services/llo/telem/telemetry_test.go +++ b/core/services/llo/telem/telemetry_test.go @@ -13,11 +13,10 @@ import ( "google.golang.org/protobuf/proto" "gopkg.in/guregu/null.v4" - "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" ocr2types "github.com/smartcontractkit/libocr/offchainreporting2plus/types" + llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" llocommon "github.com/smartcontractkit/chainlink-data-streams/llo/common" - llov30 "github.com/smartcontractkit/chainlink-data-streams/llo/v30" "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" "github.com/smartcontractkit/chainlink/v2/core/logger" "github.com/smartcontractkit/chainlink/v2/core/services/pipeline" @@ -50,18 +49,14 @@ type mockOpts struct { func (m *mockOpts) VerboseLogging() bool { return m.verboseLogging } func (m *mockOpts) SeqNr() uint64 { return 1042 } -func (m *mockOpts) OutCtx() ocr3types.OutcomeContext { - return ocr3types.OutcomeContext{SeqNr: 1042, PreviousOutcome: ocr3types.Outcome([]byte("foo"))} -} func (m *mockOpts) ConfigDigest() ocr2types.ConfigDigest { return ocr2types.ConfigDigest{6, 5, 4} } func (m *mockOpts) ObservationTimestamp() time.Time { return time.Unix(1737936858, 0) } - -func (m *mockOpts) OutcomeCodec() llov30.OutcomeCodec { - return nil +func (m *mockOpts) LifeCycleStage() llotypes.LifeCycleStage { + return llocommon.LifeCycleStageProduction } const bridgeResponse = `{ diff --git a/core/services/ocr2/delegate.go b/core/services/ocr2/delegate.go index 72c4eb7dc17..5f3f7f1ff6a 100644 --- a/core/services/ocr2/delegate.go +++ b/core/services/ocr2/delegate.go @@ -1571,7 +1571,22 @@ func (d *Delegate) newServicesLLO( NewOCR3DB: func(pluginID int32) ocr3types.Database { return NewDB(d.ds, spec.ID, pluginID, lggr) }, + + OCR31: pluginCfg.IsOCR31(), } + + // OCR3.1 (llo/v31) additionally requires the "2" network endpoint factory and + // a persistent replicated key-value store, wired here the same way the vault + // and DKG OCR3.1 plugins are (pebble under OCR2().KeyValueStoreRootDir()). + if pluginCfg.IsOCR31() { + fullPath := filepath.Join(d.cfg.OCR2().KeyValueStoreRootDir(), jb.ExternalJobID.String()) + if err = utils.EnsureDirAndMaxPerms(fullPath, os.FileMode(0700)); err != nil { + return nil, fmt.Errorf("failed to create LLO key value store directory: %w", err) + } + cfg.BinaryNetworkEndpoint2Factory = d.peerWrapper.Peer3_1 + cfg.KeyValueDatabaseFactory = kvdb.NewPebbleKeyValueDatabaseFactory(fullPath) + } + oracle, err := llo.NewDelegate(cfg) if err != nil { return nil, err diff --git a/core/services/ocr2/plugins/llo/bench/bench_baseline.txt b/core/services/ocr2/plugins/llo/bench/bench_baseline.txt new file mode 100644 index 00000000000..0684e759e16 --- /dev/null +++ b/core/services/ocr2/plugins/llo/bench/bench_baseline.txt @@ -0,0 +1,222 @@ +goos: darwin +goarch: arm64 +pkg: github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/llo/bench +cpu: Apple M5 Max + │ core/services/ocr2/plugins/llo/bench/bench_results.txt │ + │ sec/op │ +FullRound/ch=10_str=1/v30-18 34.72µ ± 3% +FullRound/ch=10_str=1/v31-18 39.77µ ± 1% +FullRound/ch=100_str=1/v30-18 315.7µ ± 0% +FullRound/ch=100_str=1/v31-18 397.1µ ± 1% +FullRound/ch=100_str=10/v30-18 2.205m ± 1% +FullRound/ch=100_str=10/v31-18 2.236m ± 0% +FullRound/ch=1000_str=1/v30-18 3.229m ± 0% +FullRound/ch=1000_str=1/v31-18 4.317m ± 0% +FullRound/ch=1000_str=10/v30-18 22.96m ± 1% +FullRound/ch=1000_str=10/v31-18 23.15m ± 0% +Observation/ch=10_str=1/v30-18 6.466µ ± 0% +Observation/ch=10_str=1/v31-18 8.004µ ± 1% +Observation/ch=100_str=1/v30-18 60.00µ ± 1% +Observation/ch=100_str=1/v31-18 81.50µ ± 0% +Observation/ch=100_str=10/v30-18 432.4µ ± 0% +Observation/ch=100_str=10/v31-18 383.4µ ± 0% +Observation/ch=1000_str=1/v30-18 613.5µ ± 0% +Observation/ch=1000_str=1/v31-18 946.5µ ± 1% +Observation/ch=1000_str=10/v30-18 4.225m ± 1% +Observation/ch=1000_str=10/v31-18 3.807m ± 2% +StateAdvance/ch=10_str=1/v30_Outcome-18 19.90µ ± 0% +StateAdvance/ch=10_str=1/v31_StateTransition-18 21.80µ ± 12% +StateAdvance/ch=100_str=1/v30_Outcome-18 183.5µ ± 0% +StateAdvance/ch=100_str=1/v31_StateTransition-18 219.6µ ± 1% +StateAdvance/ch=100_str=10/v30_Outcome-18 1.473m ± 0% +StateAdvance/ch=100_str=10/v31_StateTransition-18 1.481m ± 0% +StateAdvance/ch=1000_str=1/v30_Outcome-18 1.861m ± 0% +StateAdvance/ch=1000_str=1/v31_StateTransition-18 2.335m ± 0% +StateAdvance/ch=1000_str=10/v30_Outcome-18 15.30m ± 1% +StateAdvance/ch=1000_str=10/v31_StateTransition-18 15.55m ± 1% +Reports/ch=10_str=1/v30-18 8.476µ ± 0% +Reports/ch=10_str=1/v31-18 8.777µ ± 0% +Reports/ch=100_str=1/v30-18 75.05µ ± 0% +Reports/ch=100_str=1/v31-18 79.41µ ± 1% +Reports/ch=100_str=10/v30-18 357.8µ ± 0% +Reports/ch=100_str=10/v31-18 383.8µ ± 1% +Reports/ch=1000_str=1/v30-18 756.3µ ± 1% +Reports/ch=1000_str=1/v31-18 819.7µ ± 0% +Reports/ch=1000_str=10/v30-18 3.424m ± 2% +Reports/ch=1000_str=10/v31-18 3.660m ± 2% +geomean 464.1µ + + │ core/services/ocr2/plugins/llo/bench/bench_results.txt │ + │ obs_B/op │ +FullRound/ch=10_str=1/v30-18 180.0 ± 0% +FullRound/ch=10_str=1/v31-18 172.0 ± 0% +FullRound/ch=100_str=1/v30-18 1.620k ± 0% +FullRound/ch=100_str=1/v31-18 1.612k ± 0% +FullRound/ch=100_str=10/v30-18 16.89k ± 0% +FullRound/ch=100_str=10/v31-18 16.89k ± 0% +FullRound/ch=1000_str=1/v30-18 16.89k ± 0% +FullRound/ch=1000_str=1/v31-18 16.89k ± 0% +FullRound/ch=1000_str=10/v30-18 169.9k ± 0% +FullRound/ch=1000_str=10/v31-18 169.9k ± 0% +geomean 6.727k + + │ core/services/ocr2/plugins/llo/bench/bench_results.txt │ + │ outcome_B/op │ +FullRound/ch=10_str=1/v30-18 482.0 ± 0% +FullRound/ch=100_str=1/v30-18 4.622k ± 0% +FullRound/ch=100_str=10/v30-18 27.97k ± 0% +FullRound/ch=1000_str=1/v30-18 49.51k ± 0% +FullRound/ch=1000_str=10/v30-18 283.5k ± 0% +geomean 15.43k + + │ core/services/ocr2/plugins/llo/bench/bench_results.txt │ + │ report_B/op │ +FullRound/ch=10_str=1/v30-18 2.541k ± 0% +FullRound/ch=10_str=1/v31-18 2.541k ± 0% +FullRound/ch=100_str=1/v30-18 25.59k ± 0% +FullRound/ch=100_str=1/v31-18 25.59k ± 0% +FullRound/ch=100_str=10/v30-18 44.49k ± 0% +FullRound/ch=100_str=10/v31-18 44.49k ± 0% +FullRound/ch=1000_str=1/v30-18 257.9k ± 0% +FullRound/ch=1000_str=1/v31-18 257.9k ± 0% +FullRound/ch=1000_str=10/v30-18 446.9k ± 0% +FullRound/ch=1000_str=10/v31-18 446.9k ± 0% +geomean 50.65k + + │ core/services/ocr2/plugins/llo/bench/bench_results.txt │ + │ reports/op │ +FullRound/ch=10_str=1/v30-18 10.00 ± 0% +FullRound/ch=10_str=1/v31-18 10.00 ± 0% +FullRound/ch=100_str=1/v30-18 100.0 ± 0% +FullRound/ch=100_str=1/v31-18 100.0 ± 0% +FullRound/ch=100_str=10/v30-18 100.0 ± 0% +FullRound/ch=100_str=10/v31-18 100.0 ± 0% +FullRound/ch=1000_str=1/v30-18 1.000k ± 0% +FullRound/ch=1000_str=1/v31-18 1.000k ± 0% +FullRound/ch=1000_str=10/v30-18 1.000k ± 0% +FullRound/ch=1000_str=10/v31-18 1.000k ± 0% +geomean 158.5 + + │ core/services/ocr2/plugins/llo/bench/bench_results.txt │ + │ B/op │ +FullRound/ch=10_str=1/v30-18 64.18Ki ± 0% +FullRound/ch=10_str=1/v31-18 64.38Ki ± 0% +FullRound/ch=100_str=1/v30-18 617.4Ki ± 0% +FullRound/ch=100_str=1/v31-18 586.4Ki ± 0% +FullRound/ch=100_str=10/v30-18 4.477Mi ± 0% +FullRound/ch=100_str=10/v31-18 3.788Mi ± 0% +FullRound/ch=1000_str=1/v30-18 6.626Mi ± 0% +FullRound/ch=1000_str=1/v31-18 6.669Mi ± 0% +FullRound/ch=1000_str=10/v30-18 43.71Mi ± 0% +FullRound/ch=1000_str=10/v31-18 36.53Mi ± 0% +Observation/ch=10_str=1/v30-18 11.46Ki ± 0% +Observation/ch=10_str=1/v31-18 9.344Ki ± 0% +Observation/ch=100_str=1/v30-18 119.9Ki ± 0% +Observation/ch=100_str=1/v31-18 80.11Ki ± 0% +Observation/ch=100_str=10/v30-18 946.7Ki ± 0% +Observation/ch=100_str=10/v31-18 517.1Ki ± 0% +Observation/ch=1000_str=1/v30-18 1.328Mi ± 0% +Observation/ch=1000_str=1/v31-18 1.029Mi ± 0% +Observation/ch=1000_str=10/v30-18 9.117Mi ± 0% +Observation/ch=1000_str=10/v31-18 4.753Mi ± 0% +StateAdvance/ch=10_str=1/v30_Outcome-18 33.69Ki ± 0% +StateAdvance/ch=10_str=1/v31_StateTransition-18 34.02Ki ± 0% +StateAdvance/ch=100_str=1/v30_Outcome-18 320.3Ki ± 0% +StateAdvance/ch=100_str=1/v31_StateTransition-18 297.4Ki ± 0% +StateAdvance/ch=100_str=10/v30_Outcome-18 2.748Mi ± 0% +StateAdvance/ch=100_str=10/v31_StateTransition-18 2.417Mi ± 0% +StateAdvance/ch=1000_str=1/v30_Outcome-18 3.496Mi ± 0% +StateAdvance/ch=1000_str=1/v31_StateTransition-18 3.435Mi ± 0% +StateAdvance/ch=1000_str=10/v30_Outcome-18 26.46Mi ± 0% +StateAdvance/ch=1000_str=10/v31_StateTransition-18 23.00Mi ± 0% +Reports/ch=10_str=1/v30-18 18.90Ki ± 0% +Reports/ch=10_str=1/v31-18 19.80Ki ± 0% +Reports/ch=100_str=1/v30-18 177.1Ki ± 0% +Reports/ch=100_str=1/v31-18 190.9Ki ± 0% +Reports/ch=100_str=10/v30-18 823.5Ki ± 0% +Reports/ch=100_str=10/v31-18 871.7Ki ± 0% +Reports/ch=1000_str=1/v30-18 1.803Mi ± 0% +Reports/ch=1000_str=1/v31-18 2.040Mi ± 0% +Reports/ch=1000_str=10/v30-18 8.135Mi ± 0% +Reports/ch=1000_str=10/v31-18 8.619Mi ± 0% +geomean 842.7Ki + + │ core/services/ocr2/plugins/llo/bench/bench_results.txt │ + │ allocs/op │ +FullRound/ch=10_str=1/v30-18 1.209k ± 0% +FullRound/ch=10_str=1/v31-18 1.292k ± 0% +FullRound/ch=100_str=1/v30-18 10.92k ± 0% +FullRound/ch=100_str=1/v31-18 11.05k ± 0% +FullRound/ch=100_str=10/v30-18 86.50k ± 0% +FullRound/ch=100_str=10/v31-18 76.57k ± 0% +FullRound/ch=1000_str=1/v30-18 110.3k ± 0% +FullRound/ch=1000_str=1/v31-18 110.6k ± 0% +FullRound/ch=1000_str=10/v30-18 878.4k ± 0% +FullRound/ch=1000_str=10/v31-18 771.4k ± 0% +Observation/ch=10_str=1/v30-18 205.0 ± 0% +Observation/ch=10_str=1/v31-18 223.0 ± 0% +Observation/ch=100_str=1/v30-18 1.933k ± 0% +Observation/ch=100_str=1/v31-18 1.777k ± 0% +Observation/ch=100_str=10/v30-18 14.61k ± 0% +Observation/ch=100_str=10/v31-18 9.412k ± 0% +Observation/ch=1000_str=1/v30-18 19.07k ± 0% +Observation/ch=1000_str=1/v31-18 17.13k ± 0% +Observation/ch=1000_str=10/v30-18 148.9k ± 0% +Observation/ch=1000_str=10/v31-18 93.33k ± 0% +StateAdvance/ch=10_str=1/v30_Outcome-18 683.0 ± 0% +StateAdvance/ch=10_str=1/v31_StateTransition-18 698.0 ± 0% +StateAdvance/ch=100_str=1/v30_Outcome-18 6.131k ± 0% +StateAdvance/ch=100_str=1/v31_StateTransition-18 5.984k ± 0% +StateAdvance/ch=100_str=10/v30_Outcome-18 55.12k ± 0% +StateAdvance/ch=100_str=10/v31_StateTransition-18 49.95k ± 0% +StateAdvance/ch=1000_str=1/v30_Outcome-18 63.20k ± 0% +StateAdvance/ch=1000_str=1/v31_StateTransition-18 61.32k ± 0% +StateAdvance/ch=1000_str=10/v30_Outcome-18 562.3k ± 0% +StateAdvance/ch=1000_str=10/v31_StateTransition-18 507.0k ± 0% +Reports/ch=10_str=1/v30-18 320.0 ± 0% +Reports/ch=10_str=1/v31-18 320.0 ± 0% +Reports/ch=100_str=1/v30-18 2.855k ± 0% +Reports/ch=100_str=1/v31-18 2.870k ± 0% +Reports/ch=100_str=10/v30-18 16.76k ± 0% +Reports/ch=100_str=10/v31-18 16.78k ± 0% +Reports/ch=1000_str=1/v30-18 28.08k ± 0% +Reports/ch=1000_str=1/v31-18 28.11k ± 0% +Reports/ch=1000_str=10/v30-18 167.1k ± 0% +Reports/ch=1000_str=10/v31-18 167.2k ± 0% +geomean 15.35k + + │ core/services/ocr2/plugins/llo/bench/bench_results.txt │ + │ kvkeys/op │ +FullRound/ch=10_str=1/v31-18 11.00 ± 0% +FullRound/ch=100_str=1/v31-18 101.0 ± 0% +FullRound/ch=100_str=10/v31-18 101.0 ± 0% +FullRound/ch=1000_str=1/v31-18 1.001k ± 0% +FullRound/ch=1000_str=10/v31-18 1.001k ± 0% +geomean 162.2 + + │ core/services/ocr2/plugins/llo/bench/bench_results.txt │ + │ kvread_B/op │ +FullRound/ch=10_str=1/v31-18 456.0 ± 0% +FullRound/ch=100_str=1/v31-18 4.236k ± 0% +FullRound/ch=100_str=10/v31-18 16.78k ± 0% +FullRound/ch=1000_str=1/v31-18 43.78k ± 0% +FullRound/ch=1000_str=10/v31-18 169.8k ± 0% +geomean 11.92k + + │ core/services/ocr2/plugins/llo/bench/bench_results.txt │ + │ kvwrite_B/op │ +FullRound/ch=10_str=1/v31-18 152.0 ± 0% +FullRound/ch=100_str=1/v31-18 1.412k ± 0% +FullRound/ch=100_str=10/v31-18 1.412k ± 0% +FullRound/ch=1000_str=1/v31-18 14.01k ± 0% +FullRound/ch=1000_str=10/v31-18 14.01k ± 0% +geomean 2.264k + + │ core/services/ocr2/plugins/llo/bench/bench_results.txt │ + │ precursor_B/op │ +FullRound/ch=10_str=1/v31-18 482.0 ± 0% +FullRound/ch=100_str=1/v31-18 4.622k ± 0% +FullRound/ch=100_str=10/v31-18 27.97k ± 0% +FullRound/ch=1000_str=1/v31-18 49.51k ± 0% +FullRound/ch=1000_str=10/v31-18 283.5k ± 0% +geomean 15.43k diff --git a/core/services/ocr2/plugins/llo/bench/harness_test.go b/core/services/ocr2/plugins/llo/bench/harness_test.go new file mode 100644 index 00000000000..6bbc79c1fd8 --- /dev/null +++ b/core/services/ocr2/plugins/llo/bench/harness_test.go @@ -0,0 +1,428 @@ +// Package bench contains a comparative micro-benchmark between the OCR3.0 LLO +// plugin (chainlink-data-streams/llo/v30) and the OCR3.1 LLO plugin +// (chainlink-data-streams/llo/v31). +// +// The two plugins implement identical LLO application logic on top of different +// OCR protocols. The performance question this benchmark answers is what that +// protocol difference costs at steady state: +// +// - v30 (OCR3.0) carries all of its state in a single Outcome blob that is +// decoded from the previous round and re-encoded every round. Its per-round +// cost therefore scales with the *total* state size (channels + streams). +// - v31 (OCR3.1) keeps its state in a replicated KeyValueState (a pebble +// database in production) and only reads/writes the keys that change. Its +// per-round cost scales with per-round *churn*, not total state. +// +// Both plugins are driven exclusively through their exported ReportingPlugin +// APIs, so nothing in the read-only chainlink-data-streams or libocr modules is +// modified. v31's KeyValueState is backed by libocr's in-memory +// KeyValueDatabase (offchainreporting2plus/ocrintegrationtesthelpers) rather +// than the production pebble factory: pebble commits with fsync every round, +// which dwarfs and obscures the plugin's own work. The in-memory store isolates +// plugin CPU/allocation cost, making it directly comparable to v30's in-memory +// Outcome blob. Neither plugin's oracle-level OCR3 protocol Database +// (core/services/llo/delegate.go) is modeled here. +// +// Scope note: blob offloading (v31 offloads large observation payloads to +// libocr blobs above BlobThreshold) cannot be exercised outside libocr's oracle +// runtime — a BlobHandle cannot be constructed by application code (see the +// comment in v31/plugin_test.go). We therefore disable blob offloading here so +// observations always inline, which also makes the observation-transport cost +// directly comparable to v30 (which is always inline). Blob behavior is covered +// by the integration tests. +package bench + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/shopspring/decimal" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" + llocommon "github.com/smartcontractkit/chainlink-data-streams/llo/common" + llov30 "github.com/smartcontractkit/chainlink-data-streams/llo/v30" + llov31 "github.com/smartcontractkit/chainlink-data-streams/llo/v31" + "github.com/smartcontractkit/libocr/commontypes" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3_1types" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" + memkvdb "github.com/smartcontractkit/libocr/offchainreporting2plus/ocrintegrationtesthelpers" + ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" + + corello "github.com/smartcontractkit/chainlink/v2/core/services/llo" +) + +// benchConfigDigest is a fixed config digest shared by both plugins. v31 passes +// it to the KeyValueDatabase factory. +var benchConfigDigest = ocrtypes.ConfigDigest{'b', 'e', 'n', 'c', 'h'} + +const ( + // maxDurationObservation bounds the DataSource.Observe context in both + // plugins. It must be > 0 or the observation context is already expired. + maxDurationObservation = 5 * time.Second + // channelsPerRound is the protocol cap on how many channel definitions an + // observation may vote to add per round (MaxObservationUpdateChannelDefinitionsLength). + // Establishing C channels therefore takes ~ceil(C/channelsPerRound) rounds. + channelsPerRound = 5 + // warmupRoundSlack is added on top of the minimum rounds needed to add every + // channel, to allow the last batch to become reportable and to absorb the + // bootstrap round. + warmupRoundSlack = 32 +) + +// --------------------------------------------------------------------------- +// Workload +// --------------------------------------------------------------------------- + +// workload describes the size of the report-production problem: numChannels +// channels, each observing streamsPerChannel distinct streams via the median +// aggregator, emitting a JSON report. +// +// Stream IDs are unique across the whole workload, so the total number of +// distinct observed streams is numChannels*streamsPerChannel. Report format is +// held constant (JSON) because it is orthogonal to the v30-vs-v31 delta: both +// plugins use the identical report codec, so its cost cancels out of the +// comparison. +type workload struct { + numChannels int + streamsPerChannel int +} + +func (w workload) String() string { + return fmt.Sprintf("ch=%d_str=%d", w.numChannels, w.streamsPerChannel) +} + +// channelDefinitions builds the workload's channel definitions and returns the +// full set of stream IDs referenced. +func (w workload) channelDefinitions() (llotypes.ChannelDefinitions, []llotypes.StreamID) { + defs := make(llotypes.ChannelDefinitions, w.numChannels) + var streamIDs []llotypes.StreamID + var sid llotypes.StreamID + for c := 0; c < w.numChannels; c++ { + streams := make([]llotypes.Stream, 0, w.streamsPerChannel) + for s := 0; s < w.streamsPerChannel; s++ { + sid++ + streams = append(streams, llotypes.Stream{StreamID: sid, Aggregator: llotypes.AggregatorMedian}) + streamIDs = append(streamIDs, sid) + } + defs[llotypes.ChannelID(c+1)] = llotypes.ChannelDefinition{ + ReportFormat: llotypes.ReportFormatJSON, + Streams: streams, + } + } + return defs, streamIDs +} + +// --------------------------------------------------------------------------- +// Mocks (shared, version-agnostic) +// --------------------------------------------------------------------------- + +type mockChannelDefinitionCache struct{ defs llotypes.ChannelDefinitions } + +func (m *mockChannelDefinitionCache) Definitions(llotypes.ChannelDefinitions) llotypes.ChannelDefinitions { + return m.defs +} +func (m *mockChannelDefinitionCache) Start(context.Context) error { return nil } +func (m *mockChannelDefinitionCache) Close() error { return nil } +func (m *mockChannelDefinitionCache) Ready() error { return nil } +func (m *mockChannelDefinitionCache) HealthReport() map[string]error { return nil } +func (m *mockChannelDefinitionCache) Name() string { return "benchChannelDefinitionCache" } + +// staticDataSource fills every requested stream with a fixed decimal value. +// This keeps the DataSource out of the measured critical path (no I/O, no +// allocation-heavy pipeline) so the benchmark isolates plugin cost. +type staticDataSource struct{ value *llocommon.Decimal } + +func newStaticDataSource() *staticDataSource { + return &staticDataSource{value: llocommon.ToDecimal(decimal.NewFromInt(123456))} +} + +func (d *staticDataSource) Observe(_ context.Context, sv llocommon.StreamValues, _ llocommon.DSOpts) error { + for k := range sv { + sv[k] = d.value + } + return nil +} + +type mockShouldRetireCache struct{} + +func (mockShouldRetireCache) ShouldRetire(ocrtypes.ConfigDigest) (bool, error) { return false, nil } + +type mockOnchainConfigCodec struct{} + +func (mockOnchainConfigCodec) Decode([]byte) (llocommon.OnchainConfig, error) { + return llocommon.OnchainConfig{}, nil +} +func (mockOnchainConfigCodec) Encode(llocommon.OnchainConfig) ([]byte, error) { return nil, nil } + +// --------------------------------------------------------------------------- +// Plugin construction +// --------------------------------------------------------------------------- + +func reportCodecs() map[llotypes.ReportFormat]llocommon.ReportCodec { + // The same production codec set both plugins use (delegate.go). Only the + // JSON codec is exercised by this workload. + return corello.NewReportCodecs(logger.Nop(), 0) +} + +// benchOffchainConfig selects protocol version 1 with a 1ns minimum report +// interval. Version 1 makes both plugins use full nanosecond timestamp +// resolution for the JSON report format (version 0 truncates v30's timestamps +// to whole seconds, which would prevent reporting within a single wall-clock +// second and diverge from v31). The 1ns interval effectively reports every +// round while keeping both plugins on identical reportability rules. +func benchOffchainConfig() []byte { + b, err := llocommon.OffchainConfig{ + ProtocolVersion: 1, + DefaultMinReportIntervalNanoseconds: 1, + }.Encode() + if err != nil { + panic(err) + } + return b +} + +func pluginConfig(n, f int) ocr3types.ReportingPluginConfig { + return ocr3types.ReportingPluginConfig{ + ConfigDigest: benchConfigDigest, + N: n, + F: f, + MaxDurationObservation: maxDurationObservation, + OffchainConfig: benchOffchainConfig(), + } +} + +func buildV30(tb testing.TB, defs llotypes.ChannelDefinitions, n, f int) ocr3types.ReportingPlugin[llotypes.ReportInfo] { + tb.Helper() + factory := llov30.NewPluginFactory(llov30.PluginFactoryParams{ + Config: llov30.Config{VerboseLogging: false}, + ShouldRetireCache: mockShouldRetireCache{}, + RetirementReportCodec: llocommon.StandardRetirementReportCodec{}, + ChannelDefinitionCache: &mockChannelDefinitionCache{defs: defs}, + DataSource: newStaticDataSource(), + Logger: logger.Nop(), + OnchainConfigCodec: mockOnchainConfigCodec{}, + ReportCodecs: reportCodecs(), + }) + p, _, err := factory.NewReportingPlugin(context.Background(), pluginConfig(n, f)) + require.NoError(tb, err) + return p +} + +func buildV31(tb testing.TB, defs llotypes.ChannelDefinitions, n, f int) (ocr3_1types.ReportingPlugin[llotypes.ReportInfo], ocr3_1types.KeyValueDatabase) { + tb.Helper() + factory := llov31.NewPluginFactory(llov31.PluginFactoryParams{ + Config: llov31.Config{VerboseLogging: false}, + ShouldRetireCache: mockShouldRetireCache{}, + RetirementReportCodec: llocommon.StandardRetirementReportCodec{}, + ChannelDefinitionCache: &mockChannelDefinitionCache{defs: defs}, + DataSource: newStaticDataSource(), + Logger: logger.Nop(), + OnchainConfigCodec: mockOnchainConfigCodec{}, + ReportCodecs: reportCodecs(), + // Negative disables blob offloading; observations always inline. See the + // package-level scope note. + BlobThreshold: -1, + }) + p, _, err := factory.NewReportingPlugin(context.Background(), pluginConfig(n, f), nil) + require.NoError(tb, err) + + // libocr's in-memory KeyValueDatabase (the same helper v31's integration + // tests use): a btree behind the production KeyValueDatabaseFactory + // interface, whose Commit applies to memory with no WAL/fsync. This isolates + // the plugin's CPU/allocation cost from storage-engine cost, so the v31 + // numbers are directly comparable to v30's in-memory Outcome blob. + dbFactory := memkvdb.NewStatelessInMemoryKeyValueDatabaseFactory() + db, err := dbFactory.NewKeyValueDatabase(benchConfigDigest) + require.NoError(tb, err) + tb.Cleanup(func() { _ = db.Close() }) + return p, db +} + +// --------------------------------------------------------------------------- +// Round drivers +// --------------------------------------------------------------------------- + +func attributedObservation(observer int, obs []byte) ocrtypes.AttributedObservation { + return ocrtypes.AttributedObservation{Observer: commontypes.OracleID(observer), Observation: obs} //nolint:gosec // G115: observer is a small oracle index +} + +// replicate builds n AttributedObservations from a single serialized +// observation. Real oracles observe slightly different values; identical copies +// are representative for a benchmark and keep the workload deterministic (the +// aggregation still processes n observations per stream). +func replicate(obs []byte, n int) []ocrtypes.AttributedObservation { + aos := make([]ocrtypes.AttributedObservation, 0, n) + for i := range n { + aos = append(aos, attributedObservation(i, obs)) + } + return aos +} + +// v30Round drives one full v30 round (Observation → Outcome → Reports) against +// a fixed previous outcome. It returns the produced outcome, the reports, and +// the serialized observation. +func v30Round(tb testing.TB, p ocr3types.ReportingPlugin[llotypes.ReportInfo], seqNr uint64, prevOutcome []byte, n int) (ocr3types.Outcome, []ocr3types.ReportPlus[llotypes.ReportInfo], []byte) { + ctx := context.Background() + outctx := ocr3types.OutcomeContext{SeqNr: seqNr, PreviousOutcome: prevOutcome} + obs, err := p.Observation(ctx, outctx, nil) + require.NoError(tb, err) + aos := replicate(obs, n) + outcome, err := p.Outcome(ctx, outctx, nil, aos) + require.NoError(tb, err) + reports, err := p.Reports(ctx, seqNr, outcome) + require.NoError(tb, err) + return outcome, reports, obs +} + +// v31Round drives one full v31 round (Observation → StateTransition → Reports) +// against the KeyValueDatabase, mirroring libocr's per-seqNr transaction +// lifecycle: Observation reads a snapshot of the state committed after seqNr-1; +// StateTransition mutates a batch that is committed to advance the state to +// seqNr. It mutates db. +func v31Round(tb testing.TB, p ocr3_1types.ReportingPlugin[llotypes.ReportInfo], db ocr3_1types.KeyValueDatabase, seqNr uint64, n int) ([]ocr3types.ReportPlus[llotypes.ReportInfo], []byte) { + ctx := context.Background() + + var obs []byte + if seqNr > 1 { + rtx, err := db.NewReadTransaction() + require.NoError(tb, err) + obs, err = p.Observation(ctx, seqNr, ocrtypes.AttributedQuery{}, rtx, nil) + rtx.Discard() + require.NoError(tb, err) + } + aos := bootOrReplicate(obs, n, seqNr) + + wtx, err := db.NewReadWriteTransaction() + require.NoError(tb, err) + prec, err := p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, aos, wtx, nil) + if err != nil { + wtx.Discard() + require.NoError(tb, err) + } + require.NoError(tb, wtx.Commit()) + + reports, err := p.Reports(ctx, seqNr, prec) + require.NoError(tb, err) + return reports, obs +} + +// bootOrReplicate returns the bootstrap observation set (empty observations) +// for the first round, otherwise n copies of obs. +func bootOrReplicate(obs []byte, n int, seqNr uint64) []ocrtypes.AttributedObservation { + if seqNr == 1 { + // First round: empty observations, only 2f+1 are needed; n is fine. + return replicate(nil, n) + } + return replicate(obs, n) +} + +// --------------------------------------------------------------------------- +// Warmup +// --------------------------------------------------------------------------- + +// warmupRounds returns the round budget needed to establish `channels` +// channels (5 per round) plus slack for reportability and bootstrap. +func warmupRounds(channels int) uint64 { + return uint64(channels/channelsPerRound + warmupRoundSlack) //nolint:gosec // G115: small non-negative round budget +} + +// warmV30 drives rounds until all `channels` channels are established and +// reportable, then returns the previous-outcome and seqNr that reliably yield a +// full report set. That (prevOutcome, seqNr) pair is reused for every measured +// iteration: the stored watermarks are in the past, so each measured round +// (with a fresh wall-clock observation timestamp) reports every channel. +func warmV30(tb testing.TB, p ocr3types.ReportingPlugin[llotypes.ReportInfo], n, channels int) (steadyPrev ocr3types.Outcome, steadySeq uint64) { + tb.Helper() + // seqNr 1 bootstraps with empty observations and no previous outcome. + prev, _, _ := v30Round(tb, p, 1, nil, n) + maxRounds := warmupRounds(channels) + for seqNr := uint64(2); seqNr <= maxRounds; seqNr++ { + outcome, reports, _ := v30Round(tb, p, seqNr, prev, n) + if len(reports) >= channels { + return prev, seqNr + } + prev = outcome + } + tb.Fatalf("v30 did not reach %d reportable channels within %d warmup rounds", channels, maxRounds) + return nil, 0 +} + +// warmV31 drives rounds until all `channels` channels are established and +// reportable, returning the next seqNr to use. Unlike v30, v31 must chain +// (state lives in the mutated KeyValueDatabase), so measured iterations continue from +// this seqNr. +func warmV31(tb testing.TB, p ocr3_1types.ReportingPlugin[llotypes.ReportInfo], db ocr3_1types.KeyValueDatabase, n, channels int) (nextSeq uint64) { + tb.Helper() + maxRounds := warmupRounds(channels) + for seqNr := uint64(1); seqNr <= maxRounds; seqNr++ { + reports, _ := v31Round(tb, p, db, seqNr, n) + if len(reports) >= channels { + return seqNr + 1 + } + } + tb.Fatalf("v31 did not reach %d reportable channels within %d warmup rounds", channels, maxRounds) + return 0 +} + +// reportBytes sums the serialized report payloads. +func reportBytes(reports []ocr3types.ReportPlus[llotypes.ReportInfo]) int { + var total int + for _, r := range reports { + total += len(r.ReportWithInfo.Report) + } + return total +} + +// --------------------------------------------------------------------------- +// Counting KV wrappers (size instrumentation, not used on the hot path) +// --------------------------------------------------------------------------- + +// countingRW wraps a KeyValueState read-write transaction and tallies the volume of +// KeyValueState I/O a single v31 round performs. It is used only for the +// one-off size probe, never inside the timed loop, so its counter overhead +// does not pollute latency measurements. +type countingRW struct { + inner ocr3_1types.KeyValueStateReadWriter + reads, readBytes, writes, writeBytes, deletes, keys int +} + +func (c *countingRW) Read(key []byte) ([]byte, error) { + v, err := c.inner.Read(key) + c.reads++ + c.readBytes += len(v) + return v, err +} + +func (c *countingRW) Write(key, value []byte) error { + c.writes++ + c.keys++ + c.writeBytes += len(key) + len(value) + return c.inner.Write(key, value) +} + +func (c *countingRW) Delete(key []byte) error { + c.deletes++ + c.keys++ + return c.inner.Delete(key) +} + +var _ ocr3_1types.KeyValueStateReadWriter = (*countingRW)(nil) + +// countingReader wraps a read transaction and tallies read volume. +type countingReader struct { + inner ocr3_1types.KeyValueStateReader + reads, readBytes int +} + +func (c *countingReader) Read(key []byte) ([]byte, error) { + v, err := c.inner.Read(key) + c.reads++ + c.readBytes += len(v) + return v, err +} + +var _ ocr3_1types.KeyValueStateReader = (*countingReader)(nil) diff --git a/core/services/ocr2/plugins/llo/bench/plugin_bench_test.go b/core/services/ocr2/plugins/llo/bench/plugin_bench_test.go new file mode 100644 index 00000000000..8f0c560d784 --- /dev/null +++ b/core/services/ocr2/plugins/llo/bench/plugin_bench_test.go @@ -0,0 +1,322 @@ +package bench + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + + llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3_1types" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" + ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" +) + +const ( + benchN = 4 // total oracles + benchF = 1 // fault tolerance (2f+1 = 3 quorum) +) + +// benchWorkloads is the scaling matrix. Total observed streams per round is +// numChannels*streamsPerChannel and is kept at or below the protocol limit +// (MaxObservationStreamValuesLength = 10_000). +var benchWorkloads = []workload{ + {numChannels: 10, streamsPerChannel: 1}, + {numChannels: 100, streamsPerChannel: 1}, + {numChannels: 100, streamsPerChannel: 10}, + {numChannels: 1000, streamsPerChannel: 1}, + {numChannels: 1000, streamsPerChannel: 10}, +} + +// --------------------------------------------------------------------------- +// Correctness gate +// --------------------------------------------------------------------------- + +// TestParity asserts that, for the same workload, both plugins reach a +// report-producing steady state and emit the same number of reports in the +// same format. This guards the benchmark: if the two drivers diverge, the +// latency numbers are not comparing like for like. +func TestParity(t *testing.T) { + t.Parallel() + for _, w := range benchWorkloads { + t.Run(w.String(), func(t *testing.T) { + t.Parallel() + defs, _ := w.channelDefinitions() + + p30 := buildV30(t, defs, benchN, benchF) + prev, seq := warmV30(t, p30, benchN, w.numChannels) + _, reports30, _ := v30Round(t, p30, seq, prev, benchN) + + p31, db := buildV31(t, defs, benchN, benchF) + seq31 := warmV31(t, p31, db, benchN, w.numChannels) + reports31, _ := v31Round(t, p31, db, seq31, benchN) + + require.NotEmpty(t, reports30, "v30 produced no reports") + require.Len(t, reports30, w.numChannels, "v30 should report every channel") + require.Len(t, reports31, len(reports30), "v30 and v31 must produce the same number of reports") + + for i := range reports30 { + require.Equal(t, llotypes.ReportFormatJSON, reports30[i].ReportWithInfo.Info.ReportFormat) + } + for i := range reports31 { + require.Equal(t, llotypes.ReportFormatJSON, reports31[i].ReportWithInfo.Info.ReportFormat) + } + }) + } +} + +// --------------------------------------------------------------------------- +// Full-round benchmark +// --------------------------------------------------------------------------- + +// BenchmarkFullRound measures a complete steady-state round for each plugin: +// Observation → (Outcome | StateTransition) → Reports. Besides ns/op and the +// -benchmem allocation metrics, it reports per-round size characteristics: +// - v30: outcome_B (the re-serialized state blob), obs_B, report_B, reports +// - v31: precursor_B, obs_B, report_B, reports, and KeyValueState I/O volume +// (kvread_B, kvwrite_B, kvkeys) — the incremental state cost v30 lacks. +func BenchmarkFullRound(b *testing.B) { + for _, w := range benchWorkloads { + defs, _ := w.channelDefinitions() + + b.Run(w.String()+"/v30", func(b *testing.B) { + p := buildV30(b, defs, benchN, benchF) + prev, seq := warmV30(b, p, benchN, w.numChannels) + + // Size probe (untimed). Reported after the loop because + // b.ResetTimer() deletes user metrics. + outcome, reports, obs := v30Round(b, p, seq, prev, benchN) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + v30Round(b, p, seq, prev, benchN) + } + b.StopTimer() + b.ReportMetric(float64(len(outcome)), "outcome_B/op") + b.ReportMetric(float64(len(obs)), "obs_B/op") + b.ReportMetric(float64(reportBytes(reports)), "report_B/op") + b.ReportMetric(float64(len(reports)), "reports/op") + }) + + b.Run(w.String()+"/v31", func(b *testing.B) { + p, db := buildV31(b, defs, benchN, benchF) + seq := warmV31(b, p, db, benchN, w.numChannels) + + // Size/IO probe (untimed). Reported after the loop because + // b.ResetTimer() deletes user metrics. + m := probeV31(b, p, db, seq, benchN) + seq++ + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + v31Round(b, p, db, seq, benchN) + seq++ + } + b.StopTimer() + b.ReportMetric(float64(m.precBytes), "precursor_B/op") + b.ReportMetric(float64(m.obsBytes), "obs_B/op") + b.ReportMetric(float64(m.reportBytesTotal), "report_B/op") + b.ReportMetric(float64(m.reports), "reports/op") + b.ReportMetric(float64(m.kvReadBytes), "kvread_B/op") + b.ReportMetric(float64(m.kvWriteBytes), "kvwrite_B/op") + b.ReportMetric(float64(m.kvKeys), "kvkeys/op") + }) + } +} + +// --------------------------------------------------------------------------- +// Per-stage benchmarks (localize where the cost is spent) +// --------------------------------------------------------------------------- + +// BenchmarkObservation measures only the Observation stage (data-source gather +// + observation encode; for v31 also the KeyValueState read). +func BenchmarkObservation(b *testing.B) { + for _, w := range benchWorkloads { + defs, _ := w.channelDefinitions() + + b.Run(w.String()+"/v30", func(b *testing.B) { + p := buildV30(b, defs, benchN, benchF) + prev, seq := warmV30(b, p, benchN, w.numChannels) + outctx := ocr3types.OutcomeContext{SeqNr: seq, PreviousOutcome: prev} + ctx := context.Background() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := p.Observation(ctx, outctx, nil); err != nil { + b.Fatal(err) + } + } + }) + + b.Run(w.String()+"/v31", func(b *testing.B) { + p, db := buildV31(b, defs, benchN, benchF) + seq := warmV31(b, p, db, benchN, w.numChannels) + ctx := context.Background() + rtx, err := db.NewReadTransaction() + require.NoError(b, err) + defer rtx.Discard() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := p.Observation(ctx, seq, ocrtypes.AttributedQuery{}, rtx, nil); err != nil { + b.Fatal(err) + } + } + }) + } +} + +// BenchmarkStateAdvance measures the state-generation stage: v30's Outcome +// (decode previous outcome, aggregate, re-encode full outcome) versus v31's +// StateTransition (incremental KeyValueState reads/writes) followed by the +// KeyValueDatabase commit that libocr performs after every StateTransition. +func BenchmarkStateAdvance(b *testing.B) { + for _, w := range benchWorkloads { + defs, _ := w.channelDefinitions() + + b.Run(w.String()+"/v30_Outcome", func(b *testing.B) { + p := buildV30(b, defs, benchN, benchF) + prev, seq := warmV30(b, p, benchN, w.numChannels) + ctx := context.Background() + outctx := ocr3types.OutcomeContext{SeqNr: seq, PreviousOutcome: prev} + obs, err := p.Observation(ctx, outctx, nil) + require.NoError(b, err) + aos := replicate(obs, benchN) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := p.Outcome(ctx, outctx, nil, aos); err != nil { + b.Fatal(err) + } + } + }) + + b.Run(w.String()+"/v31_StateTransition", func(b *testing.B) { + p, db := buildV31(b, defs, benchN, benchF) + seq := warmV31(b, p, db, benchN, w.numChannels) + ctx := context.Background() + rtx, err := db.NewReadTransaction() + require.NoError(b, err) + obs, err := p.Observation(ctx, seq, ocrtypes.AttributedQuery{}, rtx, nil) + rtx.Discard() + require.NoError(b, err) + aos := replicate(obs, benchN) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + wtx, err := db.NewReadWriteTransaction() + if err != nil { + b.Fatal(err) + } + if _, err := p.StateTransition(ctx, seq, ocrtypes.AttributedQuery{}, aos, wtx, nil); err != nil { + b.Fatal(err) + } + if err := wtx.Commit(); err != nil { + b.Fatal(err) + } + seq++ + } + }) + } +} + +// BenchmarkReports measures only the Reports stage (turning a committed +// outcome/precursor into signed report payloads). +func BenchmarkReports(b *testing.B) { + for _, w := range benchWorkloads { + defs, _ := w.channelDefinitions() + + b.Run(w.String()+"/v30", func(b *testing.B) { + p := buildV30(b, defs, benchN, benchF) + prev, seq := warmV30(b, p, benchN, w.numChannels) + outcome, reports, _ := v30Round(b, p, seq, prev, benchN) + require.NotEmpty(b, reports) + ctx := context.Background() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := p.Reports(ctx, seq, outcome); err != nil { + b.Fatal(err) + } + } + }) + + b.Run(w.String()+"/v31", func(b *testing.B) { + p, db := buildV31(b, defs, benchN, benchF) + seq := warmV31(b, p, db, benchN, w.numChannels) + // Produce a committed precursor to feed Reports repeatedly. + ctx := context.Background() + rtx, err := db.NewReadTransaction() + require.NoError(b, err) + obs, err := p.Observation(ctx, seq, ocrtypes.AttributedQuery{}, rtx, nil) + rtx.Discard() + require.NoError(b, err) + wtx, err := db.NewReadWriteTransaction() + require.NoError(b, err) + prec, err := p.StateTransition(ctx, seq, ocrtypes.AttributedQuery{}, replicate(obs, benchN), wtx, nil) + require.NoError(b, err) + require.NoError(b, wtx.Commit()) + reports, err := p.Reports(ctx, seq, prec) + require.NoError(b, err) + require.NotEmpty(b, reports) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := p.Reports(ctx, seq, prec); err != nil { + b.Fatal(err) + } + } + }) + } +} + +// --------------------------------------------------------------------------- +// v31 size probe +// --------------------------------------------------------------------------- + +type v31Sizes struct { + precBytes int + obsBytes int + reportBytesTotal int + reports int + kvReadBytes int + kvWriteBytes int + kvKeys int +} + +// probeV31 runs one instrumented round to capture per-round size and +// KeyValueState I/O characteristics. It commits, advancing the state by one +// seqNr. +func probeV31(tb testing.TB, p ocr3_1types.ReportingPlugin[llotypes.ReportInfo], db ocr3_1types.KeyValueDatabase, seqNr uint64, n int) v31Sizes { + tb.Helper() + ctx := context.Background() + + rtx, err := db.NewReadTransaction() + require.NoError(tb, err) + cr := &countingReader{inner: rtx} + obs, err := p.Observation(ctx, seqNr, ocrtypes.AttributedQuery{}, cr, nil) + rtx.Discard() + require.NoError(tb, err) + + wtx, err := db.NewReadWriteTransaction() + require.NoError(tb, err) + crw := &countingRW{inner: wtx} + prec, err := p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, replicate(obs, n), crw, nil) + require.NoError(tb, err) + require.NoError(tb, wtx.Commit()) + + reports, err := p.Reports(ctx, seqNr, prec) + require.NoError(tb, err) + + return v31Sizes{ + precBytes: len(prec), + obsBytes: len(obs), + reportBytesTotal: reportBytes(reports), + reports: len(reports), + kvReadBytes: cr.readBytes + crw.readBytes, + kvWriteBytes: crw.writeBytes, + kvKeys: crw.keys, + } +} diff --git a/core/services/ocr2/plugins/llo/bench/run.sh b/core/services/ocr2/plugins/llo/bench/run.sh new file mode 100755 index 00000000000..6b305c47cdd --- /dev/null +++ b/core/services/ocr2/plugins/llo/bench/run.sh @@ -0,0 +1,41 @@ +#!/usr/bin/env bash +# +# Runs the v30-vs-v31 LLO plugin benchmark matrix and (if benchstat is present) +# prints a summarized table with per-benchmark variance. +# +# Usage: +# ./run.sh [bench-regex] [count] [benchtime] +# +# Examples: +# ./run.sh # full matrix, count=6, benchtime=1s +# ./run.sh BenchmarkFullRound 10 2s # just the full-round bench, more samples +# ./run.sh 'FullRound/ch=10' 6 200x # a single workload, fixed iterations +# +# Read the /v30 and /v31 rows for the same workload side by side; the v31 rows +# additionally carry precursor_B, kvread_B, kvwrite_B and kvkeys per op. +set -euo pipefail + +BENCH="${1:-Benchmark}" +COUNT="${2:-6}" +BENCHTIME="${3:-1s}" + +cd "$(dirname "$0")" + +OUT="bench_results.txt" + +echo "running: -bench '${BENCH}' -count=${COUNT} -benchtime=${BENCHTIME}" +go test . \ + -run '^$' \ + -bench "${BENCH}" \ + -benchmem \ + -count="${COUNT}" \ + -benchtime="${BENCHTIME}" \ + -timeout=60m | tee "${OUT}" + +echo +if command -v benchstat >/dev/null 2>&1; then + echo "=== benchstat (${OUT}) ===" + benchstat "${OUT}" +else + echo "benchstat not found; install with: go install golang.org/x/perf/cmd/benchstat@latest" +fi diff --git a/core/services/ocr2/plugins/llo/helpers_test.go b/core/services/ocr2/plugins/llo/helpers_test.go index 683a0a39e56..a36aad19d3b 100644 --- a/core/services/ocr2/plugins/llo/helpers_test.go +++ b/core/services/ocr2/plugins/llo/helpers_test.go @@ -181,10 +181,16 @@ func setupNode( // [OCR2] c.OCR2.Enabled = new(true) c.OCR2.ContractPollInterval = commonconfig.MustNewDuration(100 * time.Millisecond) + // Unique per-node root for the OCR3.1 (llo/v31) pebble key-value store so the + // nodes in this process don't share state and nothing is written to ~/.chainlink-data. + c.OCR2.KeyValueStoreRootDir = new(t.TempDir()) // [P2P] c.P2P.PeerID = new(p2pKey.PeerID()) c.P2P.TraceLogging = new(true) + // Required for OCR3.1 (llo/v31) networking (the "2" endpoint factory needs + // the experimental ragep2p host). Backward-compatible with OCR3.0. + c.P2P.EnableExperimentalRageP2P = new(true) // [P2P.V2] c.P2P.V2.Enabled = new(true) diff --git a/core/services/ocr2/plugins/llo/history_backfill_integration_test.go b/core/services/ocr2/plugins/llo/history_backfill_integration_test.go index d615af19b51..96e6828d256 100644 --- a/core/services/ocr2/plugins/llo/history_backfill_integration_test.go +++ b/core/services/ocr2/plugins/llo/history_backfill_integration_test.go @@ -105,7 +105,19 @@ func quoteBackfillString(benchmark float64) string { func TestIntegration_LLO_history_backfill(t *testing.T) { t.Parallel() + for _, ocr31 := range []bool{false, true} { + name := "OCR3.0/v30" + if ocr31 { + name = "OCR3.1/v31" + } + t.Run(name, func(t *testing.T) { + t.Parallel() + testIntegrationLLOHistoryBackfill(t, ocr31) + }) + } +} +func testIntegrationLLOHistoryBackfill(t *testing.T, ocr31 bool) { const ( salt = 600 donID = uint32(776655) @@ -167,6 +179,9 @@ lloConfigMode = "bluegreen" donID = %d channelDefinitionsContractAddress = "0x%x" channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, configStoreAddress, fromBlock) + if ocr31 { + pluginConfig += "\nocrVersion = \"3.1\"" + } nativeStrm := Stream{ id: streamNative, @@ -208,9 +223,13 @@ channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, confi require.NoError(t, err) backend.Commit() + productionConfigOpts := []OCRConfigOption{WithOracles(oracles), WithOffchainConfig(offchainConfig)} + if ocr31 { + productionConfigOpts = append(productionConfigOpts, WithOCR31()) + } setProductionConfig( t, donID, steve, backend, configurator, configuratorAddress, nodes, - WithOracles(oracles), WithOffchainConfig(offchainConfig), + productionConfigOpts..., ) signerAddresses := make([]common.Address, len(oracles)) diff --git a/core/services/ocr2/plugins/llo/integration_test.go b/core/services/ocr2/plugins/llo/integration_test.go index 0068209c5c6..134aeb37e41 100644 --- a/core/services/ocr2/plugins/llo/integration_test.go +++ b/core/services/ocr2/plugins/llo/integration_test.go @@ -32,6 +32,7 @@ import ( "github.com/smartcontractkit/freeport" "github.com/smartcontractkit/libocr/offchainreporting2/types" "github.com/smartcontractkit/libocr/offchainreporting2plus/confighelper" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3_1confighelper" "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3confighelper" ocr2types "github.com/smartcontractkit/libocr/offchainreporting2plus/types" @@ -225,6 +226,12 @@ type OCRConfig struct { MaxDurationShouldTransmitAcceptedReport time.Duration F int OnchainConfig []byte + + // ocr31 selects the OCR3.1 (llo/v31) config format when true. It changes the + // confighelper used by generateConfig (ocr3_1confighelper => offchainConfigVersion + // 310) so the on-chain config matches what an OCR3.1 node validates. The + // config digest prefix (LLO 0x0009) is identical across OCR3.0/3.1. + ocr31 bool } func makeDefaultOCRConfig() *OCRConfig { @@ -285,6 +292,13 @@ func WithOracles(oracles []confighelper.OracleIdentityExtra) OCRConfigOption { } } +// WithOCR31 switches config generation to the OCR3.1 (llo/v31) format. +func WithOCR31() OCRConfigOption { + return func(cfg *OCRConfig) { + cfg.ocr31 = true + } +} + type OCRConfigOption func(*OCRConfig) func generateConfig(t *testing.T, opts ...OCRConfigOption) (signers []types.OnchainPublicKey, transmitters []types.Account, f uint8, outOnchainConfig []byte, offchainConfigVersion uint64, offchainConfig []byte) { @@ -295,30 +309,85 @@ func generateConfig(t *testing.T, opts ...OCRConfigOption) (signers []types.Onch } t.Logf("Using OCR config: %+v\n", cfg) var err error - signers, transmitters, f, outOnchainConfig, offchainConfigVersion, offchainConfig, err = ocr3confighelper.ContractSetConfigArgsForTests( + if cfg.ocr31 { + signers, transmitters, f, outOnchainConfig, offchainConfigVersion, offchainConfig, err = generateOCR31Config(cfg) + } else { + signers, transmitters, f, outOnchainConfig, offchainConfigVersion, offchainConfig, err = ocr3confighelper.ContractSetConfigArgsForTests( + cfg.DeltaProgress, + cfg.DeltaResend, + cfg.DeltaInitial, + cfg.DeltaRound, + cfg.DeltaGrace, + cfg.DeltaCertifiedCommitRequest, + cfg.DeltaStage, + cfg.RMax, + cfg.S, + cfg.Oracles, + cfg.ReportingPluginConfig, + cfg.MaxDurationInitialization, + cfg.MaxDurationQuery, + cfg.MaxDurationObservation, + cfg.MaxDurationShouldAcceptAttestedReport, + cfg.MaxDurationShouldTransmitAcceptedReport, + cfg.F, + cfg.OnchainConfig, + ) + } + + require.NoError(t, err) + + return +} + +// generateOCR31Config maps the shared OCRConfig to ocr3_1confighelper's OCR3.1 +// argument shape (offchainConfigVersion 310). The v3.0 MaxDurationQuery/Observation +// become OCR3.1 warn durations; DeltaResend/DeltaInitial/DeltaCertifiedCommitRequest +// move into the optional-config struct. +func generateOCR31Config(cfg *OCRConfig) (signers []types.OnchainPublicKey, transmitters []types.Account, f uint8, outOnchainConfig []byte, offchainConfigVersion uint64, offchainConfig []byte, err error) { + // OCR3.1 requires a positive MaxDurationInitialization (unlike OCR3.0 where it + // is optional); default it when the shared config leaves it unset. + maxDurationInitialization := time.Second + if cfg.MaxDurationInitialization != nil && *cfg.MaxDurationInitialization > 0 { + maxDurationInitialization = *cfg.MaxDurationInitialization + } + deltaResend := cfg.DeltaResend + deltaInitial := cfg.DeltaInitial + deltaReportsPlusPrecursorRequest := cfg.DeltaCertifiedCommitRequest + // OCR3.1 requires every warn/max duration to be positive. The shared OCRConfig + // leaves several at 0 (valid for OCR3.0); positive-default them here. + positive := func(d time.Duration) time.Duration { + if d > 0 { + return d + } + return time.Second + } + return ocr3_1confighelper.ContractSetConfigArgsForTests( + ocr3_1confighelper.CheckPublicConfigLevelDefault, + cfg.Oracles, + cfg.F, cfg.DeltaProgress, - cfg.DeltaResend, - cfg.DeltaInitial, cfg.DeltaRound, cfg.DeltaGrace, - cfg.DeltaCertifiedCommitRequest, - cfg.DeltaStage, cfg.RMax, + cfg.DeltaStage, cfg.S, - cfg.Oracles, cfg.ReportingPluginConfig, - cfg.MaxDurationInitialization, - cfg.MaxDurationQuery, - cfg.MaxDurationObservation, - cfg.MaxDurationShouldAcceptAttestedReport, - cfg.MaxDurationShouldTransmitAcceptedReport, - cfg.F, cfg.OnchainConfig, + maxDurationInitialization, + positive(cfg.MaxDurationQuery), // warnDurationQuery + positive(cfg.MaxDurationObservation), // warnDurationObservation + time.Second, // warnDurationValidateObservation + time.Second, // warnDurationObservationQuorum + time.Second, // warnDurationStateTransition + time.Second, // warnDurationCommitted + positive(cfg.MaxDurationShouldAcceptAttestedReport), + positive(cfg.MaxDurationShouldTransmitAcceptedReport), + ocr3_1confighelper.ContractSetConfigArgsOptionalConfig{ + DeltaResend: &deltaResend, + DeltaInitial: &deltaInitial, + DeltaReportsPlusPrecursorRequest: &deltaReportsPlusPrecursorRequest, + }, ) - - require.NoError(t, err) - - return } func setLegacyConfig(t *testing.T, donID uint32, steve *bind.TransactOpts, backend evmtypes.Backend, legacyVerifier *verifier.Verifier, legacyVerifierAddr common.Address, nodes []Node, oracles []confighelper.OracleIdentityExtra, inOffchainConfig llocommon.OffchainConfig) ocr2types.ConfigDigest { @@ -666,15 +735,24 @@ func TestIntegration_LLO_multi_formats(t *testing.T) { DefaultMinReportIntervalNanoseconds: 1, }, } + ocrVersions := []struct { + name string + ocr31 bool + }{ + {"OCR3.0/v30", false}, + {"OCR3.1/v31", true}, + } for _, offchainConfig := range offchainConfigs { - t.Run(fmt.Sprintf("offchainConfig=%+v", offchainConfig), func(t *testing.T) { - t.Parallel() - testIntegrationLLOMultiFormats(t, offchainConfig) - }) + for _, ov := range ocrVersions { + t.Run(fmt.Sprintf("%s/offchainConfig=%+v", ov.name, offchainConfig), func(t *testing.T) { + t.Parallel() + testIntegrationLLOMultiFormats(t, offchainConfig, ov.ocr31) + }) + } } } -func testIntegrationLLOMultiFormats(t *testing.T, offchainConfig llocommon.OffchainConfig) { +func testIntegrationLLOMultiFormats(t *testing.T, offchainConfig llocommon.OffchainConfig, ocr31 bool) { testStartTimeStamp := time.Now() expirationWindow := uint32(3600) @@ -1010,6 +1088,9 @@ lloConfigMode = "bluegreen" donID = %d channelDefinitionsContractAddress = "0x%x" channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, configStoreAddress, fromBlock) + if ocr31 { + pluginConfig += "\nocrVersion = \"3.1\"" + } bridgeName := "superbridge" @@ -1197,8 +1278,12 @@ dp -> deribit_funding_interval_hours_parse -> deribit_funding_interval_hours_dec } // Set config on configurator + productionConfigOpts := []OCRConfigOption{WithOracles(oracles), WithOffchainConfig(offchainConfig)} + if ocr31 { + productionConfigOpts = append(productionConfigOpts, WithOCR31()) + } digest := setProductionConfig( - t, donID, steve, backend, configurator, configuratorAddress, nodes, WithOracles(oracles), WithOffchainConfig(offchainConfig), + t, donID, steve, backend, configurator, configuratorAddress, nodes, productionConfigOpts..., ) // NOTE: Wait for one of each type of report @@ -1831,10 +1916,26 @@ func TestIntegration_LLO_blue_green_lifecycle(t *testing.T) { ProtocolVersion: 0, DefaultMinReportIntervalNanoseconds: 0, EnableObservationCompression: false} - testIntegrationLLOBlueGreenLifecycle(t, offchainConfig) + for _, ocr31 := range []bool{false, true} { + name := "OCR3.0/v30" + if ocr31 { + name = "OCR3.1/v31" + } + t.Run(name, func(t *testing.T) { + t.Parallel() + testIntegrationLLOBlueGreenLifecycle(t, offchainConfig, ocr31) + }) + } } -func testIntegrationLLOBlueGreenLifecycle(t *testing.T, offchainConfig llocommon.OffchainConfig) { +func testIntegrationLLOBlueGreenLifecycle(t *testing.T, offchainConfig llocommon.OffchainConfig, ocr31 bool) { + // withVersion appends WithOCR31() to config options when running the v31 variant. + withVersion := func(opts ...OCRConfigOption) []OCRConfigOption { + if ocr31 { + return append(opts, WithOCR31()) + } + return opts + } clientCSAKeys := make([]csakey.KeyV2, nNodes) clientPubKeys := make([]ed25519.PublicKey, nNodes) @@ -1909,6 +2010,9 @@ lloConfigMode = "bluegreen" donID = %d channelDefinitionsContractAddress = "0x%x" channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, configStoreAddress, fromBlock) + if ocr31 { + pluginConfig += "\nocrVersion = \"3.1\"" + } addOCRJobsEVMPremiumLegacy(t, streams, serverPubKey, serverURL, configuratorAddress, bootstrapPeerID, bootstrapNodePort, nodes, configStoreAddress, clientPubKeys, pluginConfig, relayType, relayConfig) var blueDigest ocr2types.ConfigDigest @@ -1919,7 +2023,7 @@ channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, confi { // Set config on configurator blueDigest = setProductionConfig( - t, donID, steve, backend, configurator, configuratorAddress, nodes, WithOracles(oracles), WithOffchainConfig(offchainConfig), + t, donID, steve, backend, configurator, configuratorAddress, nodes, withVersion(WithOracles(oracles), WithOffchainConfig(offchainConfig))..., ) // NOTE: Wait until blue produces a report @@ -1945,7 +2049,7 @@ channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, confi { offchainConfig.EnableObservationCompression = true greenDigest = setStagingConfig( - t, donID, steve, backend, configurator, configuratorAddress, nodes, WithPredecessorConfigDigest(blueDigest), WithOracles(oracles), WithOffchainConfig(offchainConfig), + t, donID, steve, backend, configurator, configuratorAddress, nodes, withVersion(WithPredecessorConfigDigest(blueDigest), WithOracles(oracles), WithOffchainConfig(offchainConfig))..., ) // NOTE: Wait until green produces the first "specimen" report @@ -2084,7 +2188,7 @@ channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, confi offchainConfig.ProtocolVersion = 1 offchainConfig.DefaultMinReportIntervalNanoseconds = 1 blueDigest = setStagingConfig( - t, donID, steve, backend, configurator, configuratorAddress, nodes, WithPredecessorConfigDigest(greenDigest), WithOracles(oracles), WithOffchainConfig(offchainConfig), + t, donID, steve, backend, configurator, configuratorAddress, nodes, withVersion(WithPredecessorConfigDigest(greenDigest), WithOracles(oracles), WithOffchainConfig(offchainConfig))..., ) // NOTE: Wait until blue produces the first "specimen" report @@ -2188,7 +2292,19 @@ channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, confi func TestIntegration_LLO_channel_merging_owners_adders(t *testing.T) { t.Parallel() + for _, ocr31 := range []bool{false, true} { + name := "OCR3.0/v30" + if ocr31 { + name = "OCR3.1/v31" + } + t.Run(name, func(t *testing.T) { + t.Parallel() + testIntegrationLLOChannelMerging(t, ocr31) + }) + } +} +func testIntegrationLLOChannelMerging(t *testing.T, ocr31 bool) { offchainConfig := llocommon.OffchainConfig{ ProtocolVersion: 1, DefaultMinReportIntervalNanoseconds: uint64(1 * time.Second), @@ -2270,6 +2386,9 @@ lloConfigMode = "bluegreen" donID = %d channelDefinitionsContractAddress = "0x%x" channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, configStoreAddress, fromBlock) + if ocr31 { + pluginConfig += "\nocrVersion = \"3.1\"" + } // Add stream specs and LLO jobs to all nodes for i, node := range nodes { @@ -2289,8 +2408,12 @@ channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, confi } // Set initial OCR config + mergeConfigOpts := []OCRConfigOption{WithOracles(oracles), WithOffchainConfig(offchainConfig)} + if ocr31 { + mergeConfigOpts = append(mergeConfigOpts, WithOCR31()) + } digest := setProductionConfig( - t, donID, steve, backend, configurator, configuratorAddress, nodes, WithOracles(oracles), WithOffchainConfig(offchainConfig), + t, donID, steve, backend, configurator, configuratorAddress, nodes, mergeConfigOpts..., ) // Track reports by channel ID @@ -2694,7 +2817,19 @@ channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, confi // and no longer transmits reports for that channel. func TestIntegration_LLO_tombstone_stops_observations_and_reports(t *testing.T) { t.Parallel() + for _, ocr31 := range []bool{false, true} { + name := "OCR3.0/v30" + if ocr31 { + name = "OCR3.1/v31" + } + t.Run(name, func(t *testing.T) { + t.Parallel() + testIntegrationLLOTombstone(t, ocr31) + }) + } +} +func testIntegrationLLOTombstone(t *testing.T, ocr31 bool) { const ( salt = 500 donID = uint32(777666) @@ -2749,6 +2884,9 @@ lloConfigMode = "bluegreen" donID = %d channelDefinitionsContractAddress = "0x%x" channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, configStoreAddress, fromBlock) + if ocr31 { + pluginConfig += "\nocrVersion = \"3.1\"" + } var streamACalls, streamBCalls atomic.Uint64 priceA := decimal.NewFromFloat(111.1) @@ -2791,9 +2929,13 @@ channelDefinitionsContractFromBlock = %d`, serverURL, serverPubKey, donID, confi require.NoError(t, err) backend.Commit() + tombstoneConfigOpts := []OCRConfigOption{WithOracles(oracles), WithOffchainConfig(offchainConfig)} + if ocr31 { + tombstoneConfigOpts = append(tombstoneConfigOpts, WithOCR31()) + } setProductionConfig( t, donID, steve, backend, configurator, configuratorAddress, nodes, - WithOracles(oracles), WithOffchainConfig(offchainConfig), + tombstoneConfigOpts..., ) seenChannels := make(map[uint32]bool) diff --git a/core/services/ocr3_1/promwrapper/factory.go b/core/services/ocr3_1/promwrapper/factory.go new file mode 100644 index 00000000000..5d1aeb72405 --- /dev/null +++ b/core/services/ocr3_1/promwrapper/factory.go @@ -0,0 +1,61 @@ +package promwrapper + +import ( + "context" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3_1types" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" +) + +var _ ocr3_1types.ReportingPluginFactory[any] = &ReportingPluginFactory[any]{} + +// ReportingPluginFactory wraps an ocr3_1types.ReportingPluginFactory so the +// produced plugin reports prometheus metrics. It is the OCR3.1 counterpart of +// core/services/ocr3/promwrapper.ReportingPluginFactory. +type ReportingPluginFactory[RI any] struct { + origin ocr3_1types.ReportingPluginFactory[RI] + lggr logger.Logger + chainFamily string + chainID string + plugin string +} + +func NewReportingPluginFactory[RI any]( + origin ocr3_1types.ReportingPluginFactory[RI], + lggr logger.Logger, + chainFamily string, + chainID string, + plugin string, +) *ReportingPluginFactory[RI] { + return &ReportingPluginFactory[RI]{ + origin: origin, + lggr: lggr, + chainFamily: chainFamily, + chainID: chainID, + plugin: plugin, + } +} + +func (r ReportingPluginFactory[RI]) NewReportingPlugin(ctx context.Context, config ocr3types.ReportingPluginConfig, bbf ocr3_1types.BlobBroadcastFetcher) (ocr3_1types.ReportingPlugin[RI], ocr3_1types.ReportingPluginInfo, error) { + plugin, info, err := r.origin.NewReportingPlugin(ctx, config, bbf) + if err != nil { + return nil, nil, err + } + r.lggr.Infow("Wrapping OCR3.1 ReportingPlugin with prometheus metrics reporter", + "configDigest", config.ConfigDigest, + "oracleID", config.OracleID, + ) + wrapped := newReportingPlugin( + plugin, + r.chainFamily, + r.chainID, + r.plugin, + config.ConfigDigest.String(), + promOCR3ReportsGenerated, + promOCR3Durations, + promOCR3Sizes, + promOCR3PluginStatus, + ) + return wrapped, info, err +} diff --git a/core/services/ocr3_1/promwrapper/metrics.go b/core/services/ocr3_1/promwrapper/metrics.go new file mode 100644 index 00000000000..4768c31d566 --- /dev/null +++ b/core/services/ocr3_1/promwrapper/metrics.go @@ -0,0 +1,85 @@ +// Package promwrapper instruments an OCR3.1 ReportingPlugin with prometheus +// metrics. It is the OCR3.1 counterpart of core/services/ocr3/promwrapper and +// mirrors that package's structure (see also the ocr3 / ocr3_1 split of +// beholderwrapper). +// +// Metrics are emitted under their own ocr3_1_reporting_plugin_* names (distinct +// from the OCR3.0 wrapper's ocr3_reporting_plugin_* series) so the two protocol +// versions are independently observable and so the two packages never contend +// for the same collector registration. +package promwrapper + +import ( + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +type functionType string + +const ( + query functionType = "query" + observation functionType = "observation" + validateObservation functionType = "validateObservation" + observationQuorum functionType = "observationQuorum" + stateTransition functionType = "stateTransition" + committed functionType = "committed" + reports functionType = "reports" + shouldAccept functionType = "shouldAccept" + shouldTransmit functionType = "shouldTransmit" +) + +var ( + buckets = []float64{ + float64(10 * time.Millisecond), + float64(50 * time.Millisecond), + float64(100 * time.Millisecond), + float64(200 * time.Millisecond), + float64(500 * time.Millisecond), + float64(700 * time.Millisecond), + float64(time.Second), + float64(2 * time.Second), + float64(5 * time.Second), + float64(10 * time.Second), + float64(20 * time.Second), + float64(30 * time.Second), + } + + promOCR3ReportsGenerated = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "ocr3_1_reporting_plugin_reports_processed", + Help: "Tracks number of reports processed/generated by different OCR3.1 functions", + }, + []string{"chainFamily", "chainID", "plugin", "function"}, + ) + promOCR3Durations = promauto.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "ocr3_1_reporting_plugin_duration", + Help: "The amount of time elapsed during the OCR3.1 plugin's function", + Buckets: buckets, + }, + []string{"chainFamily", "chainID", "plugin", "function", "success"}, + ) + promOCR3Sizes = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "ocr3_1_reporting_plugin_data_sizes", + Help: "Tracks the size of the data produced by the OCR3.1 plugin in bytes (e.g. reports, observations etc.)", + }, + []string{"chainFamily", "chainID", "plugin", "function"}, + ) + promOCR3PluginStatus = promauto.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "ocr3_1_reporting_plugin_status", + Help: "Gauge indicating whether the OCR3.1 plugin is up and running or not", + }, + []string{"chainFamily", "chainID", "plugin", "configDigest"}, + ) +) + +func boolToInt(arg bool) int { + if arg { + return 1 + } + return 0 +} diff --git a/core/services/ocr3_1/promwrapper/plugin.go b/core/services/ocr3_1/promwrapper/plugin.go new file mode 100644 index 00000000000..5785101e6ae --- /dev/null +++ b/core/services/ocr3_1/promwrapper/plugin.go @@ -0,0 +1,171 @@ +package promwrapper + +import ( + "context" + "strconv" + "time" + + "github.com/prometheus/client_golang/prometheus" + + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3_1types" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" + ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" +) + +var _ ocr3_1types.ReportingPlugin[any] = &reportingPlugin[any]{} + +// reportingPlugin wraps an OCR3.1 ReportingPlugin, instrumenting the same +// prometheus metrics as the OCR3.0 wrapper (durations, data sizes, reports +// generated, up/down status). +type reportingPlugin[RI any] struct { + ocr3_1types.ReportingPlugin[RI] + chainFamily string + chainID string + plugin string + + configDigest string + // Prometheus components for tracking metrics + reportsGenerated *prometheus.CounterVec + durations *prometheus.HistogramVec + sizes *prometheus.CounterVec + status *prometheus.GaugeVec +} + +func newReportingPlugin[RI any]( + origin ocr3_1types.ReportingPlugin[RI], + chainFamily string, + chainID string, + plugin string, + configDigest string, + reportsGenerated *prometheus.CounterVec, + durations *prometheus.HistogramVec, + sizes *prometheus.CounterVec, + status *prometheus.GaugeVec, +) *reportingPlugin[RI] { + return &reportingPlugin[RI]{ + ReportingPlugin: origin, + chainFamily: chainFamily, + chainID: chainID, + plugin: plugin, + configDigest: configDigest, + reportsGenerated: reportsGenerated, + durations: durations, + sizes: sizes, + status: status, + } +} + +func (p *reportingPlugin[RI]) Query(ctx context.Context, seqNr uint64, kvr ocr3_1types.KeyValueStateReader, bbf ocr3_1types.BlobBroadcastFetcher) (ocrtypes.Query, error) { + result, err := withObservedExecution(p, query, func() (ocrtypes.Query, error) { + return p.ReportingPlugin.Query(ctx, seqNr, kvr, bbf) + }) + p.trackSize(query, len(result), err) + return result, err +} + +func (p *reportingPlugin[RI]) Observation(ctx context.Context, seqNr uint64, aq ocrtypes.AttributedQuery, kvr ocr3_1types.KeyValueStateReader, bbf ocr3_1types.BlobBroadcastFetcher) (ocrtypes.Observation, error) { + result, err := withObservedExecution(p, observation, func() (ocrtypes.Observation, error) { + return p.ReportingPlugin.Observation(ctx, seqNr, aq, kvr, bbf) + }) + p.trackSize(observation, len(result), err) + return result, err +} + +func (p *reportingPlugin[RI]) ValidateObservation(ctx context.Context, seqNr uint64, aq ocrtypes.AttributedQuery, ao ocrtypes.AttributedObservation, kvr ocr3_1types.KeyValueStateReader, bf ocr3_1types.BlobFetcher) error { + _, err := withObservedExecution(p, validateObservation, func() (any, error) { + err := p.ReportingPlugin.ValidateObservation(ctx, seqNr, aq, ao, kvr, bf) + return nil, err + }) + return err +} + +func (p *reportingPlugin[RI]) ObservationQuorum(ctx context.Context, seqNr uint64, aq ocrtypes.AttributedQuery, aos []ocrtypes.AttributedObservation, kvr ocr3_1types.KeyValueStateReader, bf ocr3_1types.BlobFetcher) (bool, error) { + return withObservedExecution(p, observationQuorum, func() (bool, error) { + return p.ReportingPlugin.ObservationQuorum(ctx, seqNr, aq, aos, kvr, bf) + }) +} + +func (p *reportingPlugin[RI]) StateTransition(ctx context.Context, seqNr uint64, aq ocrtypes.AttributedQuery, aos []ocrtypes.AttributedObservation, kvrw ocr3_1types.KeyValueStateReadWriter, bf ocr3_1types.BlobFetcher) (ocr3_1types.ReportsPlusPrecursor, error) { + result, err := withObservedExecution(p, stateTransition, func() (ocr3_1types.ReportsPlusPrecursor, error) { + return p.ReportingPlugin.StateTransition(ctx, seqNr, aq, aos, kvrw, bf) + }) + p.trackSize(stateTransition, len(result), err) + return result, err +} + +func (p *reportingPlugin[RI]) Committed(ctx context.Context, seqNr uint64, kvr ocr3_1types.KeyValueStateReader) error { + _, err := withObservedExecution(p, committed, func() (any, error) { + err := p.ReportingPlugin.Committed(ctx, seqNr, kvr) + return nil, err + }) + return err +} + +func (p *reportingPlugin[RI]) Reports(ctx context.Context, seqNr uint64, rpp ocr3_1types.ReportsPlusPrecursor) ([]ocr3types.ReportPlus[RI], error) { + result, err := withObservedExecution(p, reports, func() ([]ocr3types.ReportPlus[RI], error) { + return p.ReportingPlugin.Reports(ctx, seqNr, rpp) + }) + p.trackReports(reports, len(result)) + return result, err +} + +func (p *reportingPlugin[RI]) ShouldAcceptAttestedReport(ctx context.Context, seqNr uint64, reportWithInfo ocr3types.ReportWithInfo[RI]) (bool, error) { + result, err := withObservedExecution(p, shouldAccept, func() (bool, error) { + return p.ReportingPlugin.ShouldAcceptAttestedReport(ctx, seqNr, reportWithInfo) + }) + p.trackReports(shouldAccept, boolToInt(result)) + return result, err +} + +func (p *reportingPlugin[RI]) ShouldTransmitAcceptedReport(ctx context.Context, seqNr uint64, reportWithInfo ocr3types.ReportWithInfo[RI]) (bool, error) { + result, err := withObservedExecution(p, shouldTransmit, func() (bool, error) { + return p.ReportingPlugin.ShouldTransmitAcceptedReport(ctx, seqNr, reportWithInfo) + }) + p.trackReports(shouldTransmit, boolToInt(result)) + return result, err +} + +func (p *reportingPlugin[RI]) Close() error { + p.updateStatus(false) + return p.ReportingPlugin.Close() +} + +func (p *reportingPlugin[RI]) trackReports(function functionType, count int) { + p.reportsGenerated. + WithLabelValues(p.chainFamily, p.chainID, p.plugin, string(function)). + Add(float64(count)) +} + +func (p *reportingPlugin[RI]) updateStatus(status bool) { + p.status. + WithLabelValues(p.chainFamily, p.chainID, p.plugin, p.configDigest). + Set(float64(boolToInt(status))) +} + +func (p *reportingPlugin[RI]) trackSize(function functionType, size int, err error) { + if err != nil { + return + } + p.sizes. + WithLabelValues(p.chainFamily, p.chainID, p.plugin, string(function)). + Add(float64(size)) +} + +func withObservedExecution[RI, R any]( + p *reportingPlugin[RI], + function functionType, + exec func() (R, error), +) (R, error) { + start := time.Now() + result, err := exec() + + success := err == nil + + p.durations. + WithLabelValues(p.chainFamily, p.chainID, p.plugin, string(function), strconv.FormatBool(success)). + Observe(float64(time.Since(start))) + + p.updateStatus(true) + + return result, err +} diff --git a/core/services/ocr3_1/promwrapper/plugin_test.go b/core/services/ocr3_1/promwrapper/plugin_test.go new file mode 100644 index 00000000000..6a3ae563fdc --- /dev/null +++ b/core/services/ocr3_1/promwrapper/plugin_test.go @@ -0,0 +1,172 @@ +package promwrapper + +import ( + "context" + "testing" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + io_prometheus_client "github.com/prometheus/client_model/go" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3_1types" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" + ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" + + "github.com/smartcontractkit/chainlink/v2/core/logger" +) + +// Test_Plugin_FunctionLabels guards the OCR3.1 metric labelling: the three +// OCR3.1-only phases must be recorded under their own distinct function labels +// (observationQuorum, stateTransition, committed) and NOT conflated with one +// another. Each method is invoked once and the per-label duration histogram +// sample count is asserted to increment by exactly one. +func Test_Plugin_FunctionLabels(t *testing.T) { + t.Parallel() + const ( + fam = "evm" + id = "1" + plug = "llo" + ) + funcs := []functionType{query, observation, validateObservation, observationQuorum, stateTransition, committed, reports, shouldAccept, shouldTransmit} + + init := map[functionType]int{} + for _, f := range funcs { + init[f] = counterFromHistogramByLabels(t, promOCR3Durations, fam, id, plug, string(f), "true") + } + + p := newReportingPlugin( + fakePlugin[uint]{reports: make([]ocr3types.ReportPlus[uint], 2), stateTransitionSize: 4}, + fam, id, plug, "abc", + promOCR3ReportsGenerated, promOCR3Durations, promOCR3Sizes, promOCR3PluginStatus, + ) + + ctx := t.Context() + _, err := p.Query(ctx, 1, nil, nil) + require.NoError(t, err) + _, err = p.Observation(ctx, 1, ocrtypes.AttributedQuery{}, nil, nil) + require.NoError(t, err) + require.NoError(t, p.ValidateObservation(ctx, 1, ocrtypes.AttributedQuery{}, ocrtypes.AttributedObservation{}, nil, nil)) + _, err = p.ObservationQuorum(ctx, 1, ocrtypes.AttributedQuery{}, nil, nil, nil) + require.NoError(t, err) + _, err = p.StateTransition(ctx, 1, ocrtypes.AttributedQuery{}, nil, nil, nil) + require.NoError(t, err) + require.NoError(t, p.Committed(ctx, 1, nil)) + _, err = p.Reports(ctx, 1, nil) + require.NoError(t, err) + _, err = p.ShouldAcceptAttestedReport(ctx, 1, ocr3types.ReportWithInfo[uint]{}) + require.NoError(t, err) + _, err = p.ShouldTransmitAcceptedReport(ctx, 1, ocr3types.ReportWithInfo[uint]{}) + require.NoError(t, err) + + // Every phase recorded exactly one duration sample under its own label. + for _, f := range funcs { + got := counterFromHistogramByLabels(t, promOCR3Durations, fam, id, plug, string(f), "true") - init[f] + require.Equalf(t, 1, got, "duration label %q should increment once", f) + } + + // StateTransition precursor size tracked under the stateTransition label. + require.Equal(t, 4, int(testutil.ToFloat64(promOCR3Sizes.WithLabelValues(fam, id, plug, string(stateTransition))))) + // Reports counter under the reports label. + require.Equal(t, 2, int(testutil.ToFloat64(promOCR3ReportsGenerated.WithLabelValues(fam, id, plug, string(reports))))) +} + +// Test_Factory covers NewReportingPluginFactory + NewReportingPlugin: the +// factory wraps the origin plugin and the wrapper reports metrics. +func Test_Factory(t *testing.T) { + t.Parallel() + factory := NewReportingPluginFactory( + fakeFactory[uint]{plugin: fakePlugin[uint]{}}, + logger.TestLogger(t), "aptos", "1", "llo", + ) + + cd := ocrtypes.ConfigDigest{9} + p, info, err := factory.NewReportingPlugin(t.Context(), ocr3types.ReportingPluginConfig{ConfigDigest: cd}, nil) + require.NoError(t, err) + require.NotNil(t, p) + require.NotNil(t, info) + + _, err = p.Query(t.Context(), 1, nil, nil) + require.NoError(t, err) + require.Equal(t, 1, int(testutil.ToFloat64(promOCR3PluginStatus.WithLabelValues("aptos", "1", "llo", cd.String())))) + + require.NoError(t, p.Close()) + require.Equal(t, 0, int(testutil.ToFloat64(promOCR3PluginStatus.WithLabelValues("aptos", "1", "llo", cd.String())))) +} + +func counterFromHistogramByLabels(t *testing.T, histogramVec *prometheus.HistogramVec, labels ...string) int { + observer, err := histogramVec.GetMetricWithLabelValues(labels...) + require.NoError(t, err) + + metricCh := make(chan prometheus.Metric, 1) + observer.(prometheus.Histogram).Collect(metricCh) + close(metricCh) + + metric := <-metricCh + pb := &io_prometheus_client.Metric{} + err = metric.Write(pb) + require.NoError(t, err) + + //nolint:gosec // we don't care about that in tests + return int(pb.GetHistogram().GetSampleCount()) +} + +type fakeFactory[RI any] struct{ plugin fakePlugin[RI] } + +func (f fakeFactory[RI]) NewReportingPlugin(context.Context, ocr3types.ReportingPluginConfig, ocr3_1types.BlobBroadcastFetcher) (ocr3_1types.ReportingPlugin[RI], ocr3_1types.ReportingPluginInfo, error) { + return f.plugin, ocr3_1types.ReportingPluginInfo1{}, nil +} + +type fakePlugin[RI any] struct { + reports []ocr3types.ReportPlus[RI] + observationSize int + stateTransitionSize int + err error +} + +func (f fakePlugin[RI]) Query(context.Context, uint64, ocr3_1types.KeyValueStateReader, ocr3_1types.BlobBroadcastFetcher) (ocrtypes.Query, error) { + return ocrtypes.Query{}, f.err +} + +func (f fakePlugin[RI]) Observation(context.Context, uint64, ocrtypes.AttributedQuery, ocr3_1types.KeyValueStateReader, ocr3_1types.BlobBroadcastFetcher) (ocrtypes.Observation, error) { + if f.err != nil { + return nil, f.err + } + return make([]byte, f.observationSize), nil +} + +func (f fakePlugin[RI]) ValidateObservation(context.Context, uint64, ocrtypes.AttributedQuery, ocrtypes.AttributedObservation, ocr3_1types.KeyValueStateReader, ocr3_1types.BlobFetcher) error { + return f.err +} + +func (f fakePlugin[RI]) ObservationQuorum(context.Context, uint64, ocrtypes.AttributedQuery, []ocrtypes.AttributedObservation, ocr3_1types.KeyValueStateReader, ocr3_1types.BlobFetcher) (bool, error) { + return false, f.err +} + +func (f fakePlugin[RI]) StateTransition(context.Context, uint64, ocrtypes.AttributedQuery, []ocrtypes.AttributedObservation, ocr3_1types.KeyValueStateReadWriter, ocr3_1types.BlobFetcher) (ocr3_1types.ReportsPlusPrecursor, error) { + if f.err != nil { + return nil, f.err + } + return make([]byte, f.stateTransitionSize), nil +} + +func (f fakePlugin[RI]) Committed(context.Context, uint64, ocr3_1types.KeyValueStateReader) error { + return f.err +} + +func (f fakePlugin[RI]) Reports(context.Context, uint64, ocr3_1types.ReportsPlusPrecursor) ([]ocr3types.ReportPlus[RI], error) { + if f.err != nil { + return nil, f.err + } + return f.reports, nil +} + +func (f fakePlugin[RI]) ShouldAcceptAttestedReport(context.Context, uint64, ocr3types.ReportWithInfo[RI]) (bool, error) { + return true, f.err +} + +func (f fakePlugin[RI]) ShouldTransmitAcceptedReport(context.Context, uint64, ocr3types.ReportWithInfo[RI]) (bool, error) { + return true, f.err +} + +func (f fakePlugin[RI]) Close() error { return f.err } diff --git a/deployment/go.mod b/deployment/go.mod index 5044638f69c..c68b1a8a869 100644 --- a/deployment/go.mod +++ b/deployment/go.mod @@ -50,7 +50,7 @@ require ( github.com/smartcontractkit/chainlink-ccip/deployment v0.0.0-20260624154507-ea7ff77a0ddb github.com/smartcontractkit/chainlink-common v0.11.2-0.20260727165036-9952de44dbab github.com/smartcontractkit/chainlink-common/keystore v1.3.0 - github.com/smartcontractkit/chainlink-data-streams v1.0.0 + github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260728111445-96c471be2872 github.com/smartcontractkit/chainlink-evm/contracts/cre/gobindings v0.0.0-20260403151002-2c91155b5501 diff --git a/deployment/go.sum b/deployment/go.sum index cc9926bfd65..e19696803f9 100644 --- a/deployment/go.sum +++ b/deployment/go.sum @@ -1420,8 +1420,8 @@ github.com/smartcontractkit/chainlink-common/keystore v1.3.0 h1:V05Rp9/dTc4Wyips github.com/smartcontractkit/chainlink-common/keystore v1.3.0/go.mod h1:vHV8BGm6TN7jBbMsWxq1Hqm3HbCtYFwzvKS0CCczxG8= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 h1:uWEwl7i2ryuRVoV4DmIKm6mqYevf1lH/8cQYhw/JXko= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= -github.com/smartcontractkit/chainlink-data-streams v1.0.0 h1:W+RfzuZHVt50ihlsB7+XpXiBw1v5hRks6DUK2HHH7A4= -github.com/smartcontractkit/chainlink-data-streams v1.0.0/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= +github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf h1:ovwq9+AKlW1DxXVT7Go1tmX567usqAgA+ON2WfzISig= +github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 h1:mzbvXxdbE/96Pdj1zyPKzf25ZlDR48+iTTDTbaITvmk= github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54/go.mod h1:sz/YCiLs8i/V57WISALB7ywNjxW24sj0hi+DE4kzv6A= github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260728111445-96c471be2872 h1:0vivai99gZjV92RLlxnIhUBTODg08JfdBD8wLfN8A5A= diff --git a/go.mod b/go.mod index 64630a47aec..eaafbc98448 100644 --- a/go.mod +++ b/go.mod @@ -88,7 +88,7 @@ require ( github.com/smartcontractkit/chainlink-common v0.11.2-0.20260727165036-9952de44dbab github.com/smartcontractkit/chainlink-common/keystore v1.3.0 github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 - github.com/smartcontractkit/chainlink-data-streams v1.0.0 + github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260728111445-96c471be2872 github.com/smartcontractkit/chainlink-evm/contracts/cre/gobindings v0.0.0-20260403151002-2c91155b5501 github.com/smartcontractkit/chainlink-evm/gethwrappers v0.0.0-20260512150409-b4068bf735e6 diff --git a/go.sum b/go.sum index 6b6b5639441..b631a77ddca 100644 --- a/go.sum +++ b/go.sum @@ -1165,8 +1165,8 @@ github.com/smartcontractkit/chainlink-common/keystore v1.3.0 h1:V05Rp9/dTc4Wyips github.com/smartcontractkit/chainlink-common/keystore v1.3.0/go.mod h1:vHV8BGm6TN7jBbMsWxq1Hqm3HbCtYFwzvKS0CCczxG8= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 h1:uWEwl7i2ryuRVoV4DmIKm6mqYevf1lH/8cQYhw/JXko= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= -github.com/smartcontractkit/chainlink-data-streams v1.0.0 h1:W+RfzuZHVt50ihlsB7+XpXiBw1v5hRks6DUK2HHH7A4= -github.com/smartcontractkit/chainlink-data-streams v1.0.0/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= +github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf h1:ovwq9+AKlW1DxXVT7Go1tmX567usqAgA+ON2WfzISig= +github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260728111445-96c471be2872 h1:0vivai99gZjV92RLlxnIhUBTODg08JfdBD8wLfN8A5A= github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260728111445-96c471be2872/go.mod h1:xxdndutJLTdMDdJwazJDCmBh8H/d7nwc4aj7ZqVbzN4= github.com/smartcontractkit/chainlink-evm/contracts/cre/gobindings v0.0.0-20260403151002-2c91155b5501 h1:QJiXTG9CmaQAuMRn5JGi+Jhji7fSkehVnKpjc8oNJJY= diff --git a/integration-tests/go.mod b/integration-tests/go.mod index dc110b47677..96f98470d35 100644 --- a/integration-tests/go.mod +++ b/integration-tests/go.mod @@ -422,7 +422,7 @@ require ( github.com/smartcontractkit/chainlink-ccip/deployment v0.0.0-20260624154507-ea7ff77a0ddb // indirect github.com/smartcontractkit/chainlink-ccv v0.1.1-0.20260716164331-d938b371c5d6 // indirect github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 // indirect - github.com/smartcontractkit/chainlink-data-streams v1.0.0 // indirect + github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf // indirect github.com/smartcontractkit/chainlink-evm/contracts/cre/gobindings v0.0.0-20260403151002-2c91155b5501 // indirect github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 // indirect github.com/smartcontractkit/chainlink-framework/capabilities v0.0.0-20260423135514-5b1a7565a99c // indirect diff --git a/integration-tests/go.sum b/integration-tests/go.sum index 8ad3efb2e67..abd113de9b3 100644 --- a/integration-tests/go.sum +++ b/integration-tests/go.sum @@ -1407,8 +1407,8 @@ github.com/smartcontractkit/chainlink-common/keystore v1.3.0 h1:V05Rp9/dTc4Wyips github.com/smartcontractkit/chainlink-common/keystore v1.3.0/go.mod h1:vHV8BGm6TN7jBbMsWxq1Hqm3HbCtYFwzvKS0CCczxG8= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 h1:uWEwl7i2ryuRVoV4DmIKm6mqYevf1lH/8cQYhw/JXko= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= -github.com/smartcontractkit/chainlink-data-streams v1.0.0 h1:W+RfzuZHVt50ihlsB7+XpXiBw1v5hRks6DUK2HHH7A4= -github.com/smartcontractkit/chainlink-data-streams v1.0.0/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= +github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf h1:ovwq9+AKlW1DxXVT7Go1tmX567usqAgA+ON2WfzISig= +github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 h1:mzbvXxdbE/96Pdj1zyPKzf25ZlDR48+iTTDTbaITvmk= github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54/go.mod h1:sz/YCiLs8i/V57WISALB7ywNjxW24sj0hi+DE4kzv6A= github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260728111445-96c471be2872 h1:0vivai99gZjV92RLlxnIhUBTODg08JfdBD8wLfN8A5A= diff --git a/integration-tests/load/go.mod b/integration-tests/load/go.mod index cfe217a2f8e..2386fa7bf87 100644 --- a/integration-tests/load/go.mod +++ b/integration-tests/load/go.mod @@ -485,7 +485,7 @@ require ( github.com/smartcontractkit/chainlink-ccv v0.1.1-0.20260716164331-d938b371c5d6 // indirect github.com/smartcontractkit/chainlink-common/keystore v1.3.0 // indirect github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 // indirect - github.com/smartcontractkit/chainlink-data-streams v1.0.0 // indirect + github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf // indirect github.com/smartcontractkit/chainlink-evm/contracts/cre/gobindings v0.0.0-20260403151002-2c91155b5501 // indirect github.com/smartcontractkit/chainlink-evm/gethwrappers v0.0.0-20260713161920-de075095648b // indirect github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 // indirect diff --git a/integration-tests/load/go.sum b/integration-tests/load/go.sum index 8d730cecfc2..fda6dbeadd0 100644 --- a/integration-tests/load/go.sum +++ b/integration-tests/load/go.sum @@ -1643,8 +1643,8 @@ github.com/smartcontractkit/chainlink-common/keystore v1.3.0 h1:V05Rp9/dTc4Wyips github.com/smartcontractkit/chainlink-common/keystore v1.3.0/go.mod h1:vHV8BGm6TN7jBbMsWxq1Hqm3HbCtYFwzvKS0CCczxG8= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 h1:uWEwl7i2ryuRVoV4DmIKm6mqYevf1lH/8cQYhw/JXko= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= -github.com/smartcontractkit/chainlink-data-streams v1.0.0 h1:W+RfzuZHVt50ihlsB7+XpXiBw1v5hRks6DUK2HHH7A4= -github.com/smartcontractkit/chainlink-data-streams v1.0.0/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= +github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf h1:ovwq9+AKlW1DxXVT7Go1tmX567usqAgA+ON2WfzISig= +github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 h1:mzbvXxdbE/96Pdj1zyPKzf25ZlDR48+iTTDTbaITvmk= github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54/go.mod h1:sz/YCiLs8i/V57WISALB7ywNjxW24sj0hi+DE4kzv6A= github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260728111445-96c471be2872 h1:0vivai99gZjV92RLlxnIhUBTODg08JfdBD8wLfN8A5A= diff --git a/plugins/plugins.public.yaml b/plugins/plugins.public.yaml index 069b6bf134c..1a72212fa5c 100644 --- a/plugins/plugins.public.yaml +++ b/plugins/plugins.public.yaml @@ -51,7 +51,7 @@ plugins: streams: - moduleURI: "github.com/smartcontractkit/chainlink-data-streams" - gitRef: "v1.0.0" + gitRef: "v1.0.1-0.20260730163401-4a7d4b607ebf" installPath: "./mercury/cmd/chainlink-mercury" ton: diff --git a/system-tests/lib/go.mod b/system-tests/lib/go.mod index 50e2f08c393..bba43c820bc 100644 --- a/system-tests/lib/go.mod +++ b/system-tests/lib/go.mod @@ -460,7 +460,7 @@ require ( github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260624154507-ea7ff77a0ddb // indirect github.com/smartcontractkit/chainlink-ccv v0.1.1-0.20260716164331-d938b371c5d6 // indirect github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 // indirect - github.com/smartcontractkit/chainlink-data-streams v1.0.0 // indirect + github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf // indirect github.com/smartcontractkit/chainlink-evm/contracts/cre/gobindings v0.0.0-20260403151002-2c91155b5501 // indirect github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 // indirect github.com/smartcontractkit/chainlink-framework/capabilities v0.0.0-20260423135514-5b1a7565a99c // indirect diff --git a/system-tests/lib/go.sum b/system-tests/lib/go.sum index 664dbe2cae6..6c1c4134963 100644 --- a/system-tests/lib/go.sum +++ b/system-tests/lib/go.sum @@ -1557,8 +1557,8 @@ github.com/smartcontractkit/chainlink-common/keystore v1.3.0 h1:V05Rp9/dTc4Wyips github.com/smartcontractkit/chainlink-common/keystore v1.3.0/go.mod h1:vHV8BGm6TN7jBbMsWxq1Hqm3HbCtYFwzvKS0CCczxG8= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 h1:uWEwl7i2ryuRVoV4DmIKm6mqYevf1lH/8cQYhw/JXko= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= -github.com/smartcontractkit/chainlink-data-streams v1.0.0 h1:W+RfzuZHVt50ihlsB7+XpXiBw1v5hRks6DUK2HHH7A4= -github.com/smartcontractkit/chainlink-data-streams v1.0.0/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= +github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf h1:ovwq9+AKlW1DxXVT7Go1tmX567usqAgA+ON2WfzISig= +github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 h1:mzbvXxdbE/96Pdj1zyPKzf25ZlDR48+iTTDTbaITvmk= github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54/go.mod h1:sz/YCiLs8i/V57WISALB7ywNjxW24sj0hi+DE4kzv6A= github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260728111445-96c471be2872 h1:0vivai99gZjV92RLlxnIhUBTODg08JfdBD8wLfN8A5A= diff --git a/system-tests/tests/go.mod b/system-tests/tests/go.mod index 330bc52784f..369f75141e6 100644 --- a/system-tests/tests/go.mod +++ b/system-tests/tests/go.mod @@ -238,7 +238,7 @@ require ( github.com/smartcontractkit/chainlink-ccip/chains/evm v0.0.0-20260624154507-ea7ff77a0ddb // indirect github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260624154507-ea7ff77a0ddb // indirect github.com/smartcontractkit/chainlink-ccv v0.1.1-0.20260716164331-d938b371c5d6 // indirect - github.com/smartcontractkit/chainlink-data-streams v1.0.0 // indirect + github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf // indirect github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 // indirect github.com/smartcontractkit/chainlink-protos/chainlink-ccv/committee-verifier v0.0.0-20251211142334-5c3421fe2c8d // indirect github.com/smartcontractkit/chainlink-protos/chainlink-ccv/heartbeat v0.0.0-20260115142640-f6b99095c12e // indirect diff --git a/system-tests/tests/go.sum b/system-tests/tests/go.sum index 3a912134784..0dbb1280035 100644 --- a/system-tests/tests/go.sum +++ b/system-tests/tests/go.sum @@ -1762,8 +1762,8 @@ github.com/smartcontractkit/chainlink-common/keystore v1.3.0 h1:V05Rp9/dTc4Wyips github.com/smartcontractkit/chainlink-common/keystore v1.3.0/go.mod h1:vHV8BGm6TN7jBbMsWxq1Hqm3HbCtYFwzvKS0CCczxG8= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 h1:uWEwl7i2ryuRVoV4DmIKm6mqYevf1lH/8cQYhw/JXko= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= -github.com/smartcontractkit/chainlink-data-streams v1.0.0 h1:W+RfzuZHVt50ihlsB7+XpXiBw1v5hRks6DUK2HHH7A4= -github.com/smartcontractkit/chainlink-data-streams v1.0.0/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= +github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf h1:ovwq9+AKlW1DxXVT7Go1tmX567usqAgA+ON2WfzISig= +github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260730163401-4a7d4b607ebf/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54 h1:mzbvXxdbE/96Pdj1zyPKzf25ZlDR48+iTTDTbaITvmk= github.com/smartcontractkit/chainlink-deployments-framework v0.111.1-0.20260612191326-e31c0ae4cd54/go.mod h1:sz/YCiLs8i/V57WISALB7ywNjxW24sj0hi+DE4kzv6A= github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260728111445-96c471be2872 h1:0vivai99gZjV92RLlxnIhUBTODg08JfdBD8wLfN8A5A=