llo/v31: attributed observation telemetry and per job plugin telemetry

llo/dev/v31/doc.go

@@ -335,17 +335,11 @@
 //  2. Update the job spec on every node: add the new config tracker as instance
 //     1 (Green) and set pluginVersions to ["v30", "v31"], so instance 1 runs
 //     pluginconfig.PluginVersionV31 while instance 0 stays on
-//     PluginVersionV30. Green bootstraps into the staging stage.
-//  3. Let it run, and verify Green from TELEMETRY, not from the Mercury server.
-//     A staging instance marks its reports Specimen = true and the EVM codecs
-//     refuse to encode those, so nothing it produces is transmitted and transmit
-//     volume does not rise. captureReportTelemetry runs before the encode, so
-//     ReportTelemetryCh and OutcomeTelemetryCh do see what Green would have
-//     emitted. The readiness gate is Green's report telemetry covering the same
-//     channel set as Blue's production output, with sane values, plus the blob
-//     pump's Misses/Cycles low and uncorrelated. There is no warm-up minimum to
-//     wait out (see above), so the overlap is however many rounds of that
-//     evidence you want.
+//     PluginVersionV30. Green bootstraps into the staging stage. Enable its
+//     telemetry in the job's [v31] block: captureStagingTelemetry plus the
+//     captureOutcomeTelemetry, captureReportTelemetry and
+//     captureAttributedObservationTelemetry types to verify it from.
+//  3. Let it run, and verify Green from TELEMETRY.
 //  4. Vote to retire v3.0: set shouldRetire for the v3.0 config digest in the
 //     ConfigurationStore. Once more than f oracles observe it, instance 0 moves
 //     to the retired stage in the round it is agreed (retirement is NOT

llo/dev/v31/factory.go

@@ -30,10 +30,18 @@ type PluginFactoryParams struct {
 	logger.Logger
 	protocol.OnchainConfigCodec
 	ReportCodecs map[llotypes.ReportFormat]protocol.ReportCodec
-	// OutcomeTelemetryCh, if set, receives one telemetry struct per StateTransition.
+	// OutcomeTelemetryCh, if set, receives one telemetry struct per production
+	// StateTransition (and staging, with CaptureStagingTelemetry) in rounds
+	// where this oracle is one of the f+1 rotating emitters.
 	OutcomeTelemetryCh chan<- *protocol.LLOOutcomeTelemetry
-	// ReportTelemetryCh, if set, receives one telemetry struct per emitted report.
+	// ReportTelemetryCh, if set, receives one telemetry struct per emitted
+	// production report (and staging, with CaptureStagingTelemetry) in rounds
+	// where this oracle is one of the f+1 rotating emitters.
 	ReportTelemetryCh chan<- *protocol.LLOReportTelemetry
+	// AttributedObservationTelemetryCh, if set, receives the decoded observation
+	// of every oracle in production rounds (and staging, with
+	// CaptureStagingTelemetry) where this oracle is one of emitter.
+	AttributedObservationTelemetryCh chan<- *protocol.LLOAttributedObservationTelemetry
 	// DonID is optional and used only for telemetry and logging.
 	DonID uint32
 	// MaxSnapshotRounds overrides DefaultMaxSnapshotRounds if non-zero. Bounds
@@ -142,6 +150,7 @@ func (f *PluginFactory) NewReportingPlugin(ctx context.Context, cfg ocr3types.Re
 		Logger:                              l,
 		N:                                   cfg.N,
 		F:                                   cfg.F,
+		OracleID:                            cfg.OracleID,
 		RetirementReportCodec:               f.RetirementReportCodec,
 		ReportCodecs:                        f.ReportCodecs,
 		DonID:                               f.DonID,
@@ -230,6 +239,11 @@ func (f *PluginFactory) NewReportingPlugin(ctx context.Context, cfg ocr3types.Re
 		return nil, nil, fmt.Errorf("invalid reporting plugin limits: %w", err)
 	}
 
+	if f.AttributedObservationTelemetryCh != nil {
+		p.attributedObservationTelemeter = newAttributedObservationTelemeter(l, f.AttributedObservationTelemetryCh, cfg.ConfigDigest, f.DonID, cfg.OracleID)
+		p.attributedObservationTelemeter.start()
+	}
+
 	return p, info, nil
 }
 

llo/dev/v31/handover_test.go

@@ -155,6 +155,8 @@ type v31Instance struct {
 	reports []emittedReport
 	// retirementRR is the raw retirement report, once retired.
 	retirementRR []byte
+	// markers counts the staging markers emitted.
+	markers int
 }
 
 func newV31Instance(t *testing.T, digest ocrtypes.ConfigDigest, predecessor *ocrtypes.ConfigDigest, prrc protocol.PredecessorRetirementReportCache) *v31Instance {
@@ -203,6 +205,10 @@ func (v *v31Instance) tryRound(obs ...Observation) error {
 			v.retirementRR = rwi.ReportWithInfo.Report
 			continue
 		}
+		if protocol.IsStagingMarker(rwi.ReportWithInfo.Info, rwi.ReportWithInfo.Report) {
+			v.markers++
+			continue
+		}
 		v.reports = append(v.reports, decodeJSONReport(v.t, "v31", v.seqNr, rwi))
 	}
 	return nil

llo/dev/v31/plugin.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/smartcontractkit/chainlink-data-streams/llo/protocol"
 
+	"github.com/smartcontractkit/libocr/commontypes"
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3_1types"
 	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
 	ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types"
@@ -24,6 +25,10 @@ import (
 type Config struct {
 	// VerboseLogging enables additional, potentially expensive logging.
 	VerboseLogging bool
+	// CaptureStagingTelemetry makes a staging instance emit plugin telemetry
+	// (outcome, report, attributed observation) as a production one does. Off,
+	// only production rounds emit.
+	CaptureStagingTelemetry bool
 }
 
 var _ ocr3_1types.ReportingPlugin[llotypes.ReportInfo] = &Plugin{}
@@ -40,6 +45,7 @@ type Plugin struct {
 	Logger                           logger.Logger
 	N                                int
 	F                                int
+	OracleID                         commontypes.OracleID
 	RetirementReportCodec            protocol.RetirementReportCodec
 	ReportCodecs                     map[llotypes.ReportFormat]protocol.ReportCodec
 	DonID                            uint32
@@ -70,6 +76,10 @@ type Plugin struct {
 	OutcomeTelemetryCh chan<- *protocol.LLOOutcomeTelemetry
 	ReportTelemetryCh  chan<- *protocol.LLOReportTelemetry
 
+	// attributedObservationTelemeter, if set, emits the decoded observation of
+	// every oracle in rounds that emit telemetry and this oracle is an emitter for.
+	attributedObservationTelemeter *attributedObservationTelemeter
+
 	// pump gathers stream observations and broadcasts them as blobs off the OCR
 	// critical path. Observation only picks up the handle it parked.
 	pump *blobPump
@@ -496,6 +506,9 @@ func (p *Plugin) ShouldTransmitAcceptedReport(context.Context, uint64, ocr3types
 }
 
 func (p *Plugin) Close() error {
+	if p.attributedObservationTelemeter != nil {
+		p.attributedObservationTelemeter.Close()
+	}
 	if p.pump != nil && !p.pump.Close() {
 		return fmt.Errorf("blob pump did not stop within %s", p.pump.closeTimeout)
 	}

llo/dev/v31/plugin_test.go

@@ -752,6 +752,8 @@ func Test_Telemetry(t *testing.T) {
 	p.DonID = 7
 	p.OutcomeTelemetryCh = otCh
 	p.ReportTelemetryCh = rtCh
+	// Emitters with N=4, F=1: seqNr 2 is {2, 3}, seqNr 3 is {3, 0}, seqNr 4 is {0, 1}.
+	p.OracleID = 0
 	kv := newMemKV()
 
 	channelDef := llotypes.ChannelDefinition{ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}}
@@ -780,7 +782,7 @@ func Test_Telemetry(t *testing.T) {
 	prec4, err := p.StateTransition(ctx, 4, ocrtypes.AttributedQuery{}, obs(3000, true), kv, testBlobs)
 	require.NoError(t, err)
 
-	require.Len(t, otCh, 3, "one outcome telemetry per non-bootstrap StateTransition")
+	require.Len(t, otCh, 2, "one outcome telemetry per non-bootstrap StateTransition this oracle emits for")
 	ot := <-otCh
 	require.Equal(t, uint32(7), ot.DonId)
 

llo/dev/v31/reports.go

@@ -107,7 +107,7 @@ func (p *Plugin) Reports(ctx context.Context, seqNr uint64, rawPrecursor ocr3_1t
 			reportForEncode := report
 			reportForEncode.ChannelID = opts.TargetChannelID
 
-			p.captureReportTelemetry(reportForEncode, targetCD)
+			p.captureReportTelemetry(reportForEncode, targetCD, out.LifeCycleStage)
 			codec, exists := p.ReportCodecs[targetCD.ReportFormat]
 			if !exists {
 				p.Logger.Warnw("Error encoding backfill report; codec missing for target ReportFormat", "reportFormat", targetCD.ReportFormat, "channelID", cid, "targetChannelID", opts.TargetChannelID, "stage", "Report", "seqNr", seqNr)
@@ -154,7 +154,7 @@ func (p *Plugin) Reports(ctx context.Context, seqNr uint64, rawPrecursor ocr3_1t
 			Specimen:                        out.LifeCycleStage != protocol.LifeCycleStageProduction,
 		}
 
-		p.captureReportTelemetry(report, cd)
+		p.captureReportTelemetry(report, cd, out.LifeCycleStage)
 
 		codec, exists := p.ReportCodecs[cd.ReportFormat]
 		if !exists {
@@ -177,9 +177,39 @@ func (p *Plugin) Reports(ctx context.Context, seqNr uint64, rawPrecursor ocr3_1t
 		})
 	}
 
+	// The staging marker goes last, so the other reports keep their positions,
+	// and only within the report count limit.
+	if out.LifeCycleStage == protocol.LifeCycleStageStaging && len(rwis) < protocol.MaxReportCount {
+		marker, err := p.stagingMarker(seqNr)
+		if err != nil {
+			return nil, err
+		}
+		rwis = append(rwis, marker)
+	}
+
 	return rwis, nil
 }
 
+// stagingMarker is the marker report of a staging round (see
+// protocol.EncodeStagingMarker). It is emitted in every staging round,
+// independent of node local config, so every oracle returns the same report
+// list.
+func (p *Plugin) stagingMarker(seqNr uint64) (ocr3types.ReportPlus[llotypes.ReportInfo], error) {
+	encoded, err := protocol.EncodeStagingMarker(p.ConfigDigest, seqNr)
+	if err != nil {
+		return ocr3types.ReportPlus[llotypes.ReportInfo]{}, fmt.Errorf("error encoding staging marker: %w", err)
+	}
+	return ocr3types.ReportPlus[llotypes.ReportInfo]{
+		ReportWithInfo: ocr3types.ReportWithInfo[llotypes.ReportInfo]{
+			Report: encoded,
+			Info: llotypes.ReportInfo{
+				LifeCycleStage: protocol.LifeCycleStageStaging,
+				ReportFormat:   llotypes.ReportFormatJSON,
+			},
+		},
+	}, nil
+}
+
 // maxUnreportableSamples bounds the channel IDs carried per reason, so one
 // summary line stays readable on a DON with hundreds of channels.
 const maxUnreportableSamples = 10

llo/dev/v31/statetransition.go

@@ -74,7 +74,7 @@ func (p *Plugin) StateTransition(ctx context.Context, seqNr uint64, _ ocrtypes.A
 		return nil, fmt.Errorf("failed to load KV state: %w", err)
 	}
 
-	tally, err := p.decodeObservations(ctx, aos, bf, p.BlobPayloads.round(seqNr))
+	tally, err := p.decodeObservations(ctx, aos, bf, p.BlobPayloads.round(seqNr), p.collectAttributedObservations(seqNr))
 	if err != nil {
 		return nil, err
 	}
@@ -242,6 +242,7 @@ func (p *Plugin) StateTransition(ctx context.Context, seqNr uint64, _ ocrtypes.A
 	p.captureHistoryTelemetry(history, requirements)
 	p.captureInsufficientHistory(effective, prev.opts, history)
 	p.captureOutcomeTelemetry(out, seqNr)
+	p.captureAttributedObservationTelemetry(out, seqNr, tally.attributed)
 	return encodePrecursor(out)
 }
 
@@ -265,9 +266,12 @@ type observationTally struct {
 	// per-oracle record rather than counted here: see writeCodecSupport.
 	supportedFormatsByOracle map[commontypes.OracleID]map[llotypes.ReportFormat]struct{}
 	streamObservations       map[llotypes.StreamID][]protocol.StreamValue
+	// attributed is every observation as decoded, in aos order. Collected only
+	// for attributed observation telemetry.
+	attributed []attributedObservation
 }
 
-func (p *Plugin) decodeObservations(ctx context.Context, aos []ocrtypes.AttributedObservation, bf ocr3_1types.BlobFetcher, memo *roundBlobPayloads) (observationTally, error) {
+func (p *Plugin) decodeObservations(ctx context.Context, aos []ocrtypes.AttributedObservation, bf ocr3_1types.BlobFetcher, memo *roundBlobPayloads, collect bool) (observationTally, error) {
 	tally := observationTally{
 		predConfigsByHash:              make(map[[32]byte]predecessorConfig),
 		predConfigVotesByHash:          make(map[[32]byte]int),
@@ -309,8 +313,14 @@ func (p *Plugin) decodeObservations(ctx context.Context, aos []ocrtypes.Attribut
 			// Deterministic decode failure (same bytes on every oracle): safe to
 			// drop just this observation.
 			p.Logger.Warnw("ignoring invalid observation", "oracleID", ao.Observer, "error", derr)
+			if collect {
+				tally.attributed = append(tally.attributed, attributedObservation{observer: ao.Observer, decodeErr: derr})
+			}
 			continue
 		}
+		if collect {
+			tally.attributed = append(tally.attributed, attributedObservation{observer: ao.Observer, obs: observation})
+		}
 
 		if p.PredecessorConfigDigest != nil {
 			if len(observation.AttestedPredecessorRetirement) != 0 {