Skip to content

[DS-3708] LLO OCR3.1 wiring #23303

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
brunotm wants to merge 7 commits into
base: develop
Choose a base branch
from
Draft

[DS-3708] LLO OCR3.1 wiring #23303

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
16d062a
llo: OCR3.1 promwrapper
brunotm Aug 3, 2026
56932b3
llo: OCR3.1 implementation
brunotm Aug 3, 2026
da3be37
llo: OCR3.1 integration tests
brunotm Aug 3, 2026
2c9b7c7
llo: OCR3.1, OCR3.0 benchmarks
brunotm Aug 3, 2026
dc8c7e4
ocr3_1 metric namespace
brunotm Aug 3, 2026
fc96111
llo: linting fixes
brunotm Aug 3, 2026
3217eda
uncommited test bump chainlink-datastreams
brunotm Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion core/scripts/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions core/scripts/go.sum
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

173 changes: 124 additions & 49 deletions core/services/llo/delegate.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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"
)
Expand All @@ -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
Expand Down Expand Up @@ -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) {
Expand All @@ -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")
Expand All @@ -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 {
Expand Down Expand Up @@ -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)
}
Expand All @@ -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 {
Expand Down
65 changes: 44 additions & 21 deletions core/services/llo/observation/data_source.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down Expand Up @@ -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
Expand All @@ -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)
}

Expand All @@ -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{})
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
}

Expand Down
Loading
Loading