diff --git a/llo/dev/v31/doc.go b/llo/dev/v31/doc.go index ee104e1..b7469a1 100644 --- a/llo/dev/v31/doc.go +++ b/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 diff --git a/llo/dev/v31/factory.go b/llo/dev/v31/factory.go index 1c72f5c..395adfa 100644 --- a/llo/dev/v31/factory.go +++ b/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 } diff --git a/llo/dev/v31/handover_test.go b/llo/dev/v31/handover_test.go index aaefc80..c7837dd 100644 --- a/llo/dev/v31/handover_test.go +++ b/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 diff --git a/llo/dev/v31/plugin.go b/llo/dev/v31/plugin.go index 51a6ae2..eacd988 100644 --- a/llo/dev/v31/plugin.go +++ b/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) } diff --git a/llo/dev/v31/plugin_test.go b/llo/dev/v31/plugin_test.go index 73878b7..a71f6bb 100644 --- a/llo/dev/v31/plugin_test.go +++ b/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) diff --git a/llo/dev/v31/reports.go b/llo/dev/v31/reports.go index faf29b6..f06c241 100644 --- a/llo/dev/v31/reports.go +++ b/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 diff --git a/llo/dev/v31/statetransition.go b/llo/dev/v31/statetransition.go index f558243..9145839 100644 --- a/llo/dev/v31/statetransition.go +++ b/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 { diff --git a/llo/dev/v31/telemetry.go b/llo/dev/v31/telemetry.go index 18fb976..aa6cf51 100644 --- a/llo/dev/v31/telemetry.go +++ b/llo/dev/v31/telemetry.go @@ -2,20 +2,30 @@ package llo import ( "fmt" + "slices" + "sync" + "google.golang.org/protobuf/encoding/protowire" + "google.golang.org/protobuf/proto" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" protocol "github.com/smartcontractkit/chainlink-data-streams/llo/protocol" + "github.com/smartcontractkit/libocr/commontypes" ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" ) -// Telemetry is emitted best-effort on buffered channels; a full channel drops +// Telemetry is emitted best-effort on buffered channels. A full channel drops // the datum rather than blocking the protocol. Emission is an unobservable side // effect and does not affect StateTransition/Reports return values or state. +// captureOutcomeTelemetry emits the outcome of rounds whose stage emits +// telemetry, when this oracle is one of the f+1 telemetry emitters for seqNr. +// Every oracle computes the same outcome, so f+1 copies suffice. func (p *Plugin) captureOutcomeTelemetry(out precursor, seqNr uint64) { - if p.OutcomeTelemetryCh == nil { + if p.OutcomeTelemetryCh == nil || !p.emitsTelemetry(out.LifeCycleStage) || !p.isTelemetryEmitter(seqNr) { return } ot, err := makeOutcomeTelemetry(out, p.ConfigDigest, seqNr, p.DonID, p.minContributions()) @@ -65,8 +75,10 @@ func makeOutcomeTelemetry(out precursor, configDigest ocrtypes.ConfigDigest, seq return ot, nil } -func (p *Plugin) captureReportTelemetry(r protocol.Report, cd llotypes.ChannelDefinition) { - if p.ReportTelemetryCh == nil { +// captureReportTelemetry emits the telemetry of reports produced in a stage +// that emits telemetry, when this oracle is one telemetry emitter for the seqNr. +func (p *Plugin) captureReportTelemetry(r protocol.Report, cd llotypes.ChannelDefinition, stage llotypes.LifeCycleStage) { + if p.ReportTelemetryCh == nil || !p.emitsTelemetry(stage) || !p.isTelemetryEmitter(r.SeqNr) { return } rt, err := makeReportTelemetry(r, cd, p.DonID) @@ -121,3 +133,224 @@ func makeReportTelemetry(r protocol.Report, cd llotypes.ChannelDefinition, donID } return rt, nil } + +// maxAttributedObservationTelemetryBytes bounds the serialized size of one +// attributed observation telemetry message. A decoded observation may reach +// protocol.MaxDecompressedObservationLength, so larger ones are split. +const maxAttributedObservationTelemetryBytes = 1 << 20 + +// attributedObservationTelemetryQueueSize bounds the rounds waiting to be built. +// A full queue drops the round rather than blocking StateTransition. +const attributedObservationTelemetryQueueSize = 8 + +// attributedObservation is one entry of the round attributed observations, as +// decoded by StateTransition. decodeErr is set when the observation was dropped +// as invalid. +type attributedObservation struct { + observer commontypes.OracleID + obs Observation + decodeErr error +} + +type attributedObservationRound struct { + seqNr uint64 + agreedObservationTimestampNanos uint64 + aos []attributedObservation +} + +// isTelemetryEmitter reports whether this oracle is one of the f+1 oracles +// emitting plugin telemetry for this seqNr. +func (p *Plugin) isTelemetryEmitter(seqNr uint64) bool { + if p.N <= 0 { + return false + } + n := uint64(p.N) + // (seqNr+i) % N == OracleID for the smallest i in [0, N). + i := (uint64(p.OracleID) + n - seqNr%n) % n + return i <= uint64(p.F) +} + +// collectAttributedObservations reports whether StateTransition should keep the +// decoded observations of seqNr for telemetry. +func (p *Plugin) collectAttributedObservations(seqNr uint64) bool { + return p.attributedObservationTelemeter != nil && p.isTelemetryEmitter(seqNr) +} + +// emitsTelemetry reports whether a round in stage emits plugin telemetry. +// Production always does. Staging only with Config.CaptureStagingTelemetry, so a +// staging instance can be verified from telemetry before it is promoted. Any +// other stage emits nothing. +func (p *Plugin) emitsTelemetry(stage llotypes.LifeCycleStage) bool { + switch stage { + case protocol.LifeCycleStageProduction: + return true + case protocol.LifeCycleStageStaging: + return p.Config.CaptureStagingTelemetry + default: + return false + } +} + +// captureAttributedObservationTelemetry hands the round decoded observations +// to the telemeter. +func (p *Plugin) captureAttributedObservationTelemetry(out precursor, seqNr uint64, aos []attributedObservation) { + if len(aos) == 0 || !p.emitsTelemetry(out.LifeCycleStage) { + return + } + p.attributedObservationTelemeter.enqueue(attributedObservationRound{ + seqNr: seqNr, + agreedObservationTimestampNanos: out.ObservationTimestampNanoseconds, + aos: aos, + }) +} + +// attributedObservationTelemeter builds attributed observation telemetry off the +// StateTransition path. It only reads the decoded observations, which are shared +// with the round tally and the blob payload memo. +type attributedObservationTelemeter struct { + lggr logger.Logger + configDigest ocrtypes.ConfigDigest + donID uint32 + emitter commontypes.OracleID + maxBytes int + + in chan attributedObservationRound + out chan<- *protocol.LLOAttributedObservationTelemetry + stop chan struct{} + done chan struct{} + started bool + closeOnce sync.Once +} + +func newAttributedObservationTelemeter(lggr logger.Logger, out chan<- *protocol.LLOAttributedObservationTelemetry, configDigest ocrtypes.ConfigDigest, donID uint32, emitter commontypes.OracleID) *attributedObservationTelemeter { + t := &attributedObservationTelemeter{ + lggr: lggr, + configDigest: configDigest, + donID: donID, + emitter: emitter, + maxBytes: maxAttributedObservationTelemetryBytes, + in: make(chan attributedObservationRound, attributedObservationTelemetryQueueSize), + out: out, + stop: make(chan struct{}), + done: make(chan struct{}), + } + return t +} + +// start runs the telemeter until Close. Call it at most once, before Close. +func (t *attributedObservationTelemeter) start() { + t.started = true + go t.run() +} + +func (t *attributedObservationTelemeter) enqueue(r attributedObservationRound) { + select { + case t.in <- r: + default: + t.lggr.Warnw("Attributed observation telemetry queue is full, dropping round", "seqNr", r.seqNr) + } +} + +// Close stops the telemeter and waits for the round in progress to finish. +// Queued rounds are dropped. +func (t *attributedObservationTelemeter) Close() { + t.closeOnce.Do(func() { close(t.stop) }) + if t.started { + <-t.done + } +} + +func (t *attributedObservationTelemeter) run() { + defer close(t.done) + for { + select { + case <-t.stop: + return + case r := <-t.in: + for _, ao := range r.aos { + for _, msg := range t.makeTelemetry(r.seqNr, r.agreedObservationTimestampNanos, ao) { + select { + case t.out <- msg: + default: + t.lggr.Warnw("AttributedObservationTelemetryCh is full, dropping telemetry", "seqNr", r.seqNr, "observer", ao.observer) + } + } + } + } + } +} + +// makeTelemetry builds the telemetry of one observation, split in parts of at +// most maxBytes. Stream values are packed in ascending stream id order. The +// header is repeated in every part, the votes and decode error are set in the +// first only. A single value larger than maxBytes gets a part of its own. +func (t *attributedObservationTelemeter) makeTelemetry(seqNr, agreedTs uint64, ao attributedObservation) []*protocol.LLOAttributedObservationTelemetry { + header := func() *protocol.LLOAttributedObservationTelemetry { + return &protocol.LLOAttributedObservationTelemetry{ + ConfigDigest: t.configDigest[:], + SeqNr: seqNr, + DonId: t.donID, + Observer: uint32(ao.observer), + Emitter: uint32(t.emitter), + OracleObservationTimestampNanoseconds: ao.obs.UnixTimestampNanoseconds, + AgreedObservationTimestampNanoseconds: agreedTs, + } + } + + first := header() + if ao.decodeErr != nil { + first.DecodeError = proto.String(ao.decodeErr.Error()) + return []*protocol.LLOAttributedObservationTelemetry{first} + } + if len(ao.obs.RemoveChannelIDs) > 0 { + first.RemoveChannelIds = make([]uint32, 0, len(ao.obs.RemoveChannelIDs)) + for id := range ao.obs.RemoveChannelIDs { + first.RemoveChannelIds = append(first.RemoveChannelIds, id) + } + slices.Sort(first.RemoveChannelIds) + } + if len(ao.obs.UpdateChannelDefinitions) > 0 { + first.UpdateChannelDefinitions = make(map[uint32]*protocol.LLOChannelDefinitionProto, len(ao.obs.UpdateChannelDefinitions)) + for id, cd := range ao.obs.UpdateChannelDefinitions { + first.UpdateChannelDefinitions[id] = protocol.ChannelDefinitionToProto(cd) + } + } + + ids := make([]llotypes.StreamID, 0, len(ao.obs.StreamValues)) + for id, sv := range ao.obs.StreamValues { + if sv != nil { + ids = append(ids, id) + } + } + slices.Sort(ids) + + headerSize := proto.Size(header()) + parts := []*protocol.LLOAttributedObservationTelemetry{first} + cur, size := first, proto.Size(first) + for _, id := range ids { + pv, err := protocol.StreamValueToProto(ao.obs.StreamValues[id]) + if err != nil { + t.lggr.Warnw("Error making attributed observation telemetry", "seqNr", seqNr, "observer", ao.observer, "streamID", id, "err", err) + continue + } + entry := streamValuesEntrySize(id, pv) + if size+entry > t.maxBytes && size > headerSize { + cur = header() + size = headerSize + parts = append(parts, cur) + } + if cur.StreamValues == nil { + cur.StreamValues = make(map[uint32]*protocol.LLOStreamValue) + } + cur.StreamValues[id] = pv + size += entry + } + return parts +} + +// streamValuesEntrySize is the serialized size of one stream_values map entry. +func streamValuesEntrySize(id llotypes.StreamID, v *protocol.LLOStreamValue) int { + n := protowire.SizeTag(1) + protowire.SizeVarint(uint64(id)) + + protowire.SizeTag(2) + protowire.SizeBytes(proto.Size(v)) + return protowire.SizeTag(9) + protowire.SizeBytes(n) +} diff --git a/llo/dev/v31/telemetry_test.go b/llo/dev/v31/telemetry_test.go new file mode 100644 index 0000000..3ed37ce --- /dev/null +++ b/llo/dev/v31/telemetry_test.go @@ -0,0 +1,672 @@ +package llo + +import ( + "errors" + "sort" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/shopspring/decimal" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" + "github.com/smartcontractkit/chainlink-common/pkg/utils/tests" + + "github.com/smartcontractkit/chainlink-data-streams/llo/dev/v31/llotest" + protocol "github.com/smartcontractkit/chainlink-data-streams/llo/protocol" + + "github.com/smartcontractkit/libocr/commontypes" + ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" +) + +// attachAOTelemeter wires an attributed observation telemeter to p without +// starting it, so tests read the enqueued rounds directly. +func attachAOTelemeter(t *testing.T, p *Plugin) *attributedObservationTelemeter { + t.Helper() + tel := newAttributedObservationTelemeter(logger.Test(t), make(chan *protocol.LLOAttributedObservationTelemetry), p.ConfigDigest, p.DonID, p.OracleID) + p.attributedObservationTelemeter = tel + return tel +} + +// requireAORound asserts whether the last StateTransition enqueued a round. +func requireAORound(t *testing.T, tel *attributedObservationTelemeter, want bool) attributedObservationRound { + t.Helper() + if !want { + require.Empty(t, tel.in, "no attributed observation telemetry expected") + return attributedObservationRound{} + } + require.Len(t, tel.in, 1, "one attributed observation round expected") + return <-tel.in +} + +func Test_IsTelemetryEmitter(t *testing.T) { + for _, tc := range []struct{ n, f int }{{4, 1}, {7, 2}, {16, 5}, {31, 10}} { + for seqNr := uint64(0); seqNr < uint64(3*tc.n); seqNr++ { + var emitters []int + for id := 0; id < tc.n; id++ { + p := &Plugin{N: tc.n, F: tc.f, OracleID: commontypes.OracleID(id)} + if p.isTelemetryEmitter(seqNr) { + emitters = append(emitters, id) + } + } + want := make([]int, 0, tc.f+1) + for i := 0; i <= tc.f; i++ { + want = append(want, int((seqNr+uint64(i))%uint64(tc.n))) + } + sort.Ints(want) + require.Equal(t, want, emitters, "n=%d f=%d seqNr=%d", tc.n, tc.f, seqNr) + } + } +} + +func Test_AttributedObservationTelemetry_StateTransition(t *testing.T) { + ctx := tests.Context(t) + p := testPlugin(t) + p.DonID = 7 + tel := attachAOTelemeter(t, p) + kv := newMemKV() + + channelDef := llotypes.ChannelDefinition{ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}} + obsAt := func(observer int, ts uint64) Observation { + return Observation{ + UnixTimestampNanoseconds: ts, + RemoveChannelIDs: map[llotypes.ChannelID]struct{}{9: {}}, + UpdateChannelDefinitions: llotypes.ChannelDefinitions{1: channelDef}, + StreamValues: protocol.StreamValues{ + 100: protocol.ToDecimal(decimal.NewFromInt(int64(40 + observer))), + 101: protocol.ToDecimal(decimal.NewFromInt(int64(50 + observer))), + }, + } + } + round := func(seqNr uint64, ts uint64) precursor { + t.Helper() + aos := make([]ocrtypes.AttributedObservation, 0, 4) + for i := 0; i < 4; i++ { + aos = append(aos, ao(i, mustEncodeObs(t, obsAt(i, ts+uint64(i))))) + } + b, err := p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, aos, kv, testBlobs) + require.NoError(t, err) + out, err := decodePrecursor(b) + require.NoError(t, err) + return out + } + + t.Run("bootstrap round emits nothing", func(t *testing.T) { + p.OracleID = 1 + _, err := p.StateTransition(ctx, 1, ocrtypes.AttributedQuery{}, []ocrtypes.AttributedObservation{ao(0, nil), ao(1, nil), ao(2, nil)}, kv, testBlobs) + require.NoError(t, err) + requireAORound(t, tel, false) + }) + + t.Run("emitter round carries every observer", func(t *testing.T) { + p.OracleID = 2 + out := round(2, 1000) + r := requireAORound(t, tel, true) + require.Equal(t, uint64(2), r.seqNr) + require.Equal(t, out.ObservationTimestampNanoseconds, r.agreedObservationTimestampNanos) + require.Len(t, r.aos, 4) + + for i, a := range r.aos { + require.Equal(t, commontypes.OracleID(i), a.observer) + msgs := tel.makeTelemetry(r.seqNr, r.agreedObservationTimestampNanos, a) + require.Len(t, msgs, 1) + m := msgs[0] + require.Equal(t, p.ConfigDigest[:], m.ConfigDigest) + require.Equal(t, uint64(2), m.SeqNr) + require.Equal(t, uint32(7), m.DonId) + require.Equal(t, uint32(i), m.Observer) + require.Equal(t, 1000+uint64(i), m.OracleObservationTimestampNanoseconds) + require.Equal(t, out.ObservationTimestampNanoseconds, m.AgreedObservationTimestampNanoseconds) + require.Nil(t, m.DecodeError) + require.Equal(t, []uint32{9}, m.RemoveChannelIds) + require.Contains(t, m.UpdateChannelDefinitions, uint32(1)) + require.Len(t, m.StreamValues, 2) + sv, err := protocol.UnmarshalProtoStreamValue(m.StreamValues[100]) + require.NoError(t, err) + require.Equal(t, protocol.ToDecimal(decimal.NewFromInt(int64(40+i))), sv) + } + }) + + t.Run("non emitter round emits nothing", func(t *testing.T) { + // Emitters for seqNr 3 with N=4, F=1 are oracles 3 and 0. + p.OracleID = 1 + round(3, 2000) + requireAORound(t, tel, false) + }) + + t.Run("agreed timestamp holds when the median regresses", func(t *testing.T) { + p.OracleID = 0 // emitter for seqNr 4 + out := round(4, 500) + r := requireAORound(t, tel, true) + require.Greater(t, out.ObservationTimestampNanoseconds, uint64(500+3), "median must have been held") + require.Equal(t, out.ObservationTimestampNanoseconds, r.agreedObservationTimestampNanos) + m := tel.makeTelemetry(r.seqNr, r.agreedObservationTimestampNanos, r.aos[0])[0] + require.Equal(t, uint64(500), m.OracleObservationTimestampNanoseconds) + require.Equal(t, out.ObservationTimestampNanoseconds, m.AgreedObservationTimestampNanoseconds) + }) + + t.Run("invalid observation carries only the decode error", func(t *testing.T) { + p.OracleID = 1 // emitter for seqNr 5 + aos := []ocrtypes.AttributedObservation{ + ao(0, mustEncodeObs(t, obsAt(0, 3000))), + ao(1, mustEncodeObs(t, obsAt(1, 3000))), + ao(2, mustEncodeObs(t, obsAt(2, 3000))), + ao(3, []byte{0x7f}), + } + _, err := p.StateTransition(ctx, 5, ocrtypes.AttributedQuery{}, aos, kv, testBlobs) + require.NoError(t, err) + r := requireAORound(t, tel, true) + require.Len(t, r.aos, 4) + + bad := r.aos[3] + require.Equal(t, commontypes.OracleID(3), bad.observer) + msgs := tel.makeTelemetry(r.seqNr, r.agreedObservationTimestampNanos, bad) + require.Len(t, msgs, 1) + require.NotNil(t, msgs[0].DecodeError) + require.Contains(t, *msgs[0].DecodeError, "unknown observation wire version") + require.Empty(t, msgs[0].StreamValues) + require.Empty(t, msgs[0].RemoveChannelIds) + require.Empty(t, msgs[0].UpdateChannelDefinitions) + require.Zero(t, msgs[0].OracleObservationTimestampNanoseconds) + }) + + t.Run("aborted round emits nothing", func(t *testing.T) { + p.OracleID = 2 // emitter for seqNr 6 + aos := make([]ocrtypes.AttributedObservation, 0, 4) + for i := 0; i < 4; i++ { + aos = append(aos, ao(i, mustEncodeObs(t, obsAt(i, 4000)))) + } + // A fetcher without the blobs fails the fetch, which aborts the round. + _, err := p.StateTransition(ctx, 6, ocrtypes.AttributedQuery{}, aos, kv, llotest.NewBlobBroadcastFetcher()) + require.Error(t, err) + requireAORound(t, tel, false) + }) +} + +func Test_PluginTelemetry_Lifecycle(t *testing.T) { + t.Run("staging telemetry off", func(t *testing.T) { testPluginTelemetryLifecycle(t, false) }) + t.Run("staging telemetry on", func(t *testing.T) { testPluginTelemetryLifecycle(t, true) }) +} + +// testPluginTelemetryLifecycle drives a v31 to v31 handover and checks outcome +// and attributed observation telemetry at every stage. +func testPluginTelemetryLifecycle(t *testing.T, stagingTelemetry bool) { + oldDigest := ocrtypes.ConfigDigest{0xa1} + newDigest := ocrtypes.ConfigDigest{0xa2} + + // Outcome and attributed observation telemetry share the same gate: emitted + // for production rounds, and staging ones when enabled. An empty stage + // expects nothing. + requireRoundTelemetry := func(t *testing.T, tel *attributedObservationTelemeter, ot chan *protocol.LLOOutcomeTelemetry, stage llotypes.LifeCycleStage) { + t.Helper() + requireAORound(t, tel, stage != "") + if stage == "" { + require.Empty(t, ot, "no outcome telemetry expected") + return + } + require.Len(t, ot, 1, "one outcome telemetry expected") + require.Equal(t, stage, llotypes.LifeCycleStage((<-ot).LifeCycleStage)) + } + + // emitterRound makes the instance an emitter for its next round, so only + // the lifecycle gate decides whether it emits. + emitterRound := func(v *v31Instance) { + v.p.OracleID = commontypes.OracleID((v.seqNr + 1) % uint64(v.p.N)) + } + + old := newV31Instance(t, oldDigest, nil, nil) + oldTel := attachAOTelemeter(t, old.p) + oldOT := make(chan *protocol.LLOOutcomeTelemetry, 8) + old.p.OutcomeTelemetryCh = oldOT + ts := handoverTickNanos + + emitterRound(old) + old.round(Observation{}) + requireRoundTelemetry(t, oldTel, oldOT, "") + require.Equal(t, protocol.LifeCycleStageProduction, old.lifeCycleStage()) + + ts += handoverTickNanos + emitterRound(old) + old.round(Observation{ + UnixTimestampNanoseconds: ts, + UpdateChannelDefinitions: llotypes.ChannelDefinitions{handoverChannelID: handoverChannel()}, + StreamValues: handoverStreamValues(), + }) + requireRoundTelemetry(t, oldTel, oldOT, protocol.LifeCycleStageProduction) + + for range 2 { + ts += handoverTickNanos + emitterRound(old) + old.round(Observation{UnixTimestampNanoseconds: ts, StreamValues: handoverStreamValues()}) + requireRoundTelemetry(t, oldTel, oldOT, protocol.LifeCycleStageProduction) + } + + // Staging instance: emits from its first round after bootstrap, only when + // staging telemetry is enabled. + newInst := newV31Instance(t, newDigest, &oldDigest, newPredecessorCache(t, nil, true)) + newInst.p.Config.CaptureStagingTelemetry = stagingTelemetry + stagingStage := llotypes.LifeCycleStage("") + if stagingTelemetry { + stagingStage = protocol.LifeCycleStageStaging + } + newTel := attachAOTelemeter(t, newInst.p) + newOT := make(chan *protocol.LLOOutcomeTelemetry, 8) + newInst.p.OutcomeTelemetryCh = newOT + stagingTS := ts + + emitterRound(newInst) + newInst.round(Observation{}) + requireRoundTelemetry(t, newTel, newOT, "") + require.Equal(t, protocol.LifeCycleStageStaging, newInst.lifeCycleStage()) + + stagingTS += handoverTickNanos + emitterRound(newInst) + newInst.round(Observation{ + UnixTimestampNanoseconds: stagingTS, + UpdateChannelDefinitions: llotypes.ChannelDefinitions{handoverChannelID: handoverChannel()}, + StreamValues: handoverStreamValues(), + }) + requireRoundTelemetry(t, newTel, newOT, stagingStage) + + stagingTS += handoverTickNanos + emitterRound(newInst) + newInst.round(Observation{UnixTimestampNanoseconds: stagingTS, StreamValues: handoverStreamValues()}) + requireRoundTelemetry(t, newTel, newOT, stagingStage) + + // Retirement round: the outcome is retired, so nothing is emitted. + old.p.ShouldRetireCache = &mockShouldRetireCache{retire: true} + ts += handoverTickNanos + emitterRound(old) + old.round(Observation{UnixTimestampNanoseconds: ts, ShouldRetire: true, StreamValues: handoverStreamValues()}) + require.Equal(t, protocol.LifeCycleStageRetired, old.lifeCycleStage()) + requireRoundTelemetry(t, oldTel, oldOT, "") + + ts += handoverTickNanos + emitterRound(old) + old.round(Observation{UnixTimestampNanoseconds: ts, StreamValues: handoverStreamValues()}) + requireRoundTelemetry(t, oldTel, oldOT, "") + + // Promotion round: the outcome is production, so it emits. + promoted := newPredecessorCache(t, attest(t, old.retirementRR), true) + newInst.p.PredecessorRetirementReportCache = promoted + attestedBytes, err := promoted.AttestedRetirementReport(oldDigest) + require.NoError(t, err) + + stagingTS += handoverTickNanos + emitterRound(newInst) + newInst.round(Observation{ + UnixTimestampNanoseconds: stagingTS, + AttestedPredecessorRetirement: attestedBytes, + PredecessorSigners: handoverPredecessorSigners, + PredecessorF: 1, + StreamValues: handoverStreamValues(), + }) + require.Equal(t, protocol.LifeCycleStageProduction, newInst.lifeCycleStage()) + requireRoundTelemetry(t, newTel, newOT, protocol.LifeCycleStageProduction) + + stagingTS += handoverTickNanos + emitterRound(newInst) + newInst.round(Observation{UnixTimestampNanoseconds: stagingTS, StreamValues: handoverStreamValues()}) + requireRoundTelemetry(t, newTel, newOT, protocol.LifeCycleStageProduction) +} + +// Test_AttributedObservationTelemetry_Deterministic checks that collecting +// observations for telemetry leaves the state transition output unchanged. +func Test_AttributedObservationTelemetry_Deterministic(t *testing.T) { + ctx := tests.Context(t) + plain, withTel := testPlugin(t), testPlugin(t) + tel := newAttributedObservationTelemeter(logger.Test(t), make(chan *protocol.LLOAttributedObservationTelemetry, 1024), withTel.ConfigDigest, 0, 0) + withTel.attributedObservationTelemeter = tel + tel.start() + t.Cleanup(tel.Close) + plainKV, telKV := newMemKV(), newMemKV() + + channelDef := llotypes.ChannelDefinition{ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}} + for seqNr := uint64(1); seqNr <= 6; seqNr++ { + withTel.OracleID = commontypes.OracleID(seqNr % uint64(withTel.N)) + aos := make([]ocrtypes.AttributedObservation, 0, 4) + for i := 0; i < 4; i++ { + var b []byte + if seqNr > 1 { + b = mustEncodeObs(t, Observation{ + UnixTimestampNanoseconds: seqNr * 1000, + UpdateChannelDefinitions: llotypes.ChannelDefinitions{1: channelDef}, + StreamValues: protocol.StreamValues{100: protocol.ToDecimal(decimal.NewFromInt(int64(i)))}, + }) + } + aos = append(aos, ao(i, b)) + } + want, err := plain.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, aos, plainKV, testBlobs) + require.NoError(t, err) + got, err := withTel.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, aos, telKV, testBlobs) + require.NoError(t, err) + require.Equal(t, want, got, "seqNr %d", seqNr) + require.Equal(t, plainKV.m, telKV.m, "seqNr %d", seqNr) + } +} + +func Test_AttributedObservationTelemetry_Split(t *testing.T) { + tel := newAttributedObservationTelemeter(logger.Test(t), nil, ocrtypes.ConfigDigest{1}, 7, 3) + + obs := Observation{ + UnixTimestampNanoseconds: 1234, + RemoveChannelIDs: map[llotypes.ChannelID]struct{}{5: {}, 2: {}}, + UpdateChannelDefinitions: llotypes.ChannelDefinitions{1: {ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 1, Aggregator: llotypes.AggregatorMedian}}}}, + StreamValues: protocol.StreamValues{}, + } + for id := llotypes.StreamID(1); id <= 500; id++ { + obs.StreamValues[id] = protocol.ToDecimal(decimal.NewFromInt(int64(id) * 1_000_003)) + } + obs.StreamValues[501] = nil // skipped + + requireParts := func(t *testing.T, msgs []*protocol.LLOAttributedObservationTelemetry) { + t.Helper() + seen := map[uint32]bool{} + for i, m := range msgs { + require.Equal(t, []byte(tel.configDigest[:]), m.ConfigDigest) + require.Equal(t, uint64(9), m.SeqNr) + require.Equal(t, uint32(7), m.DonId) + require.Equal(t, uint32(4), m.Observer) + require.Equal(t, uint32(3), m.Emitter) + require.Equal(t, uint64(1234), m.OracleObservationTimestampNanoseconds) + require.Equal(t, uint64(5678), m.AgreedObservationTimestampNanoseconds) + if i == 0 { + require.Equal(t, []uint32{2, 5}, m.RemoveChannelIds) + require.Len(t, m.UpdateChannelDefinitions, 1) + } else { + require.Empty(t, m.RemoveChannelIds) + require.Empty(t, m.UpdateChannelDefinitions) + require.NotEmpty(t, m.StreamValues, "parts after the first carry values") + } + for id := range m.StreamValues { + require.False(t, seen[id], "stream %d in two parts", id) + seen[id] = true + } + } + require.Len(t, seen, 500) + } + + t.Run("fits in one part", func(t *testing.T) { + msgs := tel.makeTelemetry(9, 5678, attributedObservation{observer: 4, obs: obs}) + require.Len(t, msgs, 1) + requireParts(t, msgs) + }) + + t.Run("split under the budget", func(t *testing.T) { + tel.maxBytes = 2048 + msgs := tel.makeTelemetry(9, 5678, attributedObservation{observer: 4, obs: obs}) + require.Greater(t, len(msgs), 2) + requireParts(t, msgs) + for _, m := range msgs { + require.LessOrEqual(t, proto.Size(m), tel.maxBytes) + } + // Ascending stream id order across parts. + var last uint32 + for _, m := range msgs { + ids := make([]uint32, 0, len(m.StreamValues)) + for id := range m.StreamValues { + ids = append(ids, id) + } + sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) + for _, id := range ids { + require.Greater(t, id, last) + last = id + } + } + }) + + t.Run("oversized values get a part each", func(t *testing.T) { + tel.maxBytes = 1 + msgs := tel.makeTelemetry(9, 5678, attributedObservation{observer: 4, obs: obs}) + require.Len(t, msgs, 501, "votes part plus one part per value") + require.Empty(t, msgs[0].StreamValues) + requireParts(t, msgs) + }) + + t.Run("decode error", func(t *testing.T) { + msgs := tel.makeTelemetry(9, 5678, attributedObservation{observer: 4, decodeErr: errors.New("boom")}) + require.Len(t, msgs, 1) + require.Equal(t, "boom", msgs[0].GetDecodeError()) + require.Empty(t, msgs[0].StreamValues) + }) +} + +func Test_AttributedObservationTelemetry_NonBlocking(t *testing.T) { + // Nobody reads the output channel and the queue overflows: neither blocks. + tel := newAttributedObservationTelemeter(logger.Test(t), make(chan *protocol.LLOAttributedObservationTelemetry), ocrtypes.ConfigDigest{1}, 0, 0) + tel.start() + + r := attributedObservationRound{seqNr: 1, aos: []attributedObservation{{observer: 0}, {observer: 1}}} + done := make(chan struct{}) + go func() { + defer close(done) + for range 10 * attributedObservationTelemetryQueueSize { + tel.enqueue(r) + } + tel.Close() + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("telemeter blocked") + } +} + +// Test_AttributedObservationTelemetry_Concurrent runs the telemeter alongside +// consecutive state transitions, which share the decoded observations. Run +// with -race. +func Test_AttributedObservationTelemetry_Concurrent(t *testing.T) { + ctx := tests.Context(t) + p := testPlugin(t) + out := make(chan *protocol.LLOAttributedObservationTelemetry, 16) + tel := newAttributedObservationTelemeter(logger.Test(t), out, p.ConfigDigest, 0, 0) + p.attributedObservationTelemeter = tel + tel.start() + + var wg sync.WaitGroup + var received int + wg.Go(func() { + for range out { + received++ + } + }) + + kv := newMemKV() + channelDef := llotypes.ChannelDefinition{ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}} + for seqNr := uint64(1); seqNr <= 30; seqNr++ { + p.OracleID = commontypes.OracleID(seqNr % uint64(p.N)) + aos := make([]ocrtypes.AttributedObservation, 0, 4) + for i := 0; i < 4; i++ { + var b []byte + if seqNr > 1 { + b = mustEncodeObs(t, Observation{ + UnixTimestampNanoseconds: seqNr * 1000, + UpdateChannelDefinitions: llotypes.ChannelDefinitions{1: channelDef}, + StreamValues: protocol.StreamValues{100: protocol.ToDecimal(decimal.NewFromInt(int64(i)))}, + }) + } + aos = append(aos, ao(i, b)) + } + _, err := p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, aos, kv, testBlobs) + require.NoError(t, err) + } + + require.Eventually(t, func() bool { return len(tel.in) == 0 }, 5*time.Second, 10*time.Millisecond) + tel.Close() + close(out) + wg.Wait() + assert.Positive(t, received) +} + +func Test_OutcomeTelemetry_Emitters(t *testing.T) { + ctx := tests.Context(t) + p := testPlugin(t) + otCh := make(chan *protocol.LLOOutcomeTelemetry, 8) + p.OutcomeTelemetryCh = otCh + kv := newMemKV() + + _, err := p.StateTransition(ctx, 1, ocrtypes.AttributedQuery{}, []ocrtypes.AttributedObservation{ao(0, nil), ao(1, nil), ao(2, nil)}, kv, testBlobs) + require.NoError(t, err) + + for seqNr := uint64(2); seqNr <= 9; seqNr++ { + // Alternate between an emitter and a non emitter for seqNr. + emitter := seqNr%2 == 0 + p.OracleID = commontypes.OracleID(seqNr % uint64(p.N)) + if !emitter { + p.OracleID = commontypes.OracleID((seqNr + uint64(p.F) + 1) % uint64(p.N)) + } + require.Equal(t, emitter, p.isTelemetryEmitter(seqNr)) + + aos := make([]ocrtypes.AttributedObservation, 0, 4) + for i := 0; i < 4; i++ { + aos = append(aos, ao(i, mustEncodeObs(t, Observation{UnixTimestampNanoseconds: seqNr * 1000}))) + } + _, err := p.StateTransition(ctx, seqNr, ocrtypes.AttributedQuery{}, aos, kv, testBlobs) + require.NoError(t, err) + + if !emitter { + require.Empty(t, otCh, "seqNr %d", seqNr) + continue + } + require.Len(t, otCh, 1, "seqNr %d", seqNr) + require.Equal(t, seqNr, (<-otCh).SeqNr) + } +} + +func Test_ReportTelemetry_Lifecycle(t *testing.T) { + // runAs drives an instance through five rounds. With emitter, this oracle is + // one of the f+1 telemetry emitters of every round, otherwise of none. + runAs := func(t *testing.T, predecessor *ocrtypes.ConfigDigest, stagingTelemetry, emitter bool, stage llotypes.LifeCycleStage) (*v31Instance, chan *protocol.LLOReportTelemetry) { + t.Helper() + v := newV31Instance(t, ocrtypes.ConfigDigest{0xb1}, predecessor, newPredecessorCache(t, nil, true)) + v.p.Config.CaptureStagingTelemetry = stagingTelemetry + rtCh := make(chan *protocol.LLOReportTelemetry, 16) + v.p.ReportTelemetryCh = rtCh + round := func(o Observation) { + offset := uint64(0) + if !emitter { + offset = uint64(v.p.F) + 1 + } + v.p.OracleID = commontypes.OracleID((v.seqNr + 1 + offset) % uint64(v.p.N)) + v.round(o) + } + + ts := handoverTickNanos + round(Observation{}) + require.Equal(t, stage, v.lifeCycleStage()) + ts += handoverTickNanos + round(Observation{ + UnixTimestampNanoseconds: ts, + UpdateChannelDefinitions: llotypes.ChannelDefinitions{handoverChannelID: handoverChannel()}, + StreamValues: handoverStreamValues(), + }) + for range 3 { + ts += handoverTickNanos + round(Observation{UnixTimestampNanoseconds: ts, StreamValues: handoverStreamValues()}) + } + require.NotEmpty(t, v.reports, "the channel must be reporting") + return v, rtCh + } + run := func(t *testing.T, predecessor *ocrtypes.ConfigDigest, stagingTelemetry bool, stage llotypes.LifeCycleStage) (*v31Instance, chan *protocol.LLOReportTelemetry) { + t.Helper() + return runAs(t, predecessor, stagingTelemetry, true, stage) + } + + // A staging instance emits specimen channel reports and a marker per round. + requireSpecimensAndMarkers := func(t *testing.T, v *v31Instance) { + t.Helper() + require.NotEmpty(t, v.reports) + for _, r := range v.reports { + require.True(t, r.specimen) + } + require.Equal(t, int(v.seqNr)-1, v.markers, "one marker per staging round after bootstrap") + } + + t.Run("staging emits no report telemetry by default", func(t *testing.T) { + v, rtCh := run(t, &ocrtypes.ConfigDigest{0xb0}, false, protocol.LifeCycleStageStaging) + requireSpecimensAndMarkers(t, v) + require.Empty(t, rtCh) + }) + + t.Run("staging emits specimen report telemetry when enabled", func(t *testing.T) { + v, rtCh := run(t, &ocrtypes.ConfigDigest{0xb0}, true, protocol.LifeCycleStageStaging) + requireSpecimensAndMarkers(t, v) + require.Len(t, rtCh, len(v.reports)) + for range len(v.reports) { + require.True(t, (<-rtCh).Specimen) + } + }) + + t.Run("production emits telemetry for every report", func(t *testing.T) { + v, rtCh := run(t, nil, false, protocol.LifeCycleStageProduction) + require.Zero(t, v.markers) + require.Len(t, rtCh, len(v.reports)) + for range len(v.reports) { + require.False(t, (<-rtCh).Specimen) + } + }) + + t.Run("a non emitter emits no report telemetry", func(t *testing.T) { + v, rtCh := runAs(t, nil, false, false, protocol.LifeCycleStageProduction) + require.NotEmpty(t, v.reports) + require.Empty(t, rtCh) + }) +} + +func Test_Reports_StagingMarker(t *testing.T) { + ctx := tests.Context(t) + + t.Run("staging returns its specimens followed by the marker", func(t *testing.T) { + p := testPlugin(t) + p.ConfigDigest = ocrtypes.ConfigDigest{0xc1} + codec := &countingReportCodec{} + p.ReportCodecs = map[llotypes.ReportFormat]protocol.ReportCodec{llotypes.ReportFormatJSON: codec} + b, err := encodePrecursor(precursor{ + LifeCycleStage: protocol.LifeCycleStageStaging, + ObservationTimestampNanoseconds: 2000, + ChannelDefinitions: llotypes.ChannelDefinitions{ + 1: {ReportFormat: llotypes.ReportFormatJSON, Streams: []llotypes.Stream{{StreamID: 100, Aggregator: llotypes.AggregatorMedian}}}, + }, + ValidAfterNanoseconds: map[llotypes.ChannelID]uint64{1: 1000}, + StreamAggregates: protocol.StreamAggregates{100: {llotypes.AggregatorMedian: protocol.ToDecimal(decimal.NewFromInt(1))}}, + SupportByFormat: map[llotypes.ReportFormat]int{llotypes.ReportFormatJSON: 4}, + }) + require.NoError(t, err) + + rwis, err := p.Reports(ctx, 12, b) + require.NoError(t, err) + require.Len(t, rwis, 2) + require.Equal(t, int32(1), codec.calls.Load(), "the channel specimen is encoded") + require.False(t, protocol.IsStagingMarker(rwis[0].ReportWithInfo.Info, rwis[0].ReportWithInfo.Report)) + + marker := rwis[1].ReportWithInfo + require.Equal(t, llotypes.ReportInfo{LifeCycleStage: protocol.LifeCycleStageStaging, ReportFormat: llotypes.ReportFormatJSON}, marker.Info) + require.True(t, protocol.IsStagingMarker(marker.Info, marker.Report)) + want, err := protocol.EncodeStagingMarker(p.ConfigDigest, 12) + require.NoError(t, err) + require.Equal(t, want, []byte(marker.Report)) + }) + + t.Run("production returns no marker", func(t *testing.T) { + p := testPlugin(t) + b, err := encodePrecursor(precursor{LifeCycleStage: protocol.LifeCycleStageProduction}) + require.NoError(t, err) + rwis, err := p.Reports(ctx, 12, b) + require.NoError(t, err) + require.Empty(t, rwis) + }) +} + +type countingReportCodec struct{ calls atomic.Int32 } + +func (c *countingReportCodec) Encode(protocol.Report, llotypes.ChannelDefinition, *protocol.OptsCache) ([]byte, error) { + c.calls.Add(1) + return []byte(`{}`), nil +} + +func (c *countingReportCodec) Verify(llotypes.ChannelDefinition) error { return nil } diff --git a/llo/pluginconfig/v31config.go b/llo/pluginconfig/v31config.go index e266e0a..3e1e73b 100644 --- a/llo/pluginconfig/v31config.go +++ b/llo/pluginconfig/v31config.go @@ -85,6 +85,20 @@ type V31Config struct { // round cannot inflate the derived snapshot age bound. Must be set above the // DON real round cadence. MaxRoundPeriod Duration `json:"maxRoundPeriod" toml:"maxRoundPeriod"` + + // Plugin telemetry. Each is decided here for v31, independently of + // the node CaptureEATelemetry parameter, which keeps driving data source telemetry. + // + // CaptureOutcomeTelemetry emits one outcome per StateTransition. + CaptureOutcomeTelemetry bool `json:"captureOutcomeTelemetry" toml:"captureOutcomeTelemetry"` + // CaptureReportTelemetry emits one telemetry per report. + CaptureReportTelemetry bool `json:"captureReportTelemetry" toml:"captureReportTelemetry"` + // CaptureAttributedObservationTelemetry emits the decoded oracle observations that made into + // the state transition. + CaptureAttributedObservationTelemetry bool `json:"captureAttributedObservationTelemetry" toml:"captureAttributedObservationTelemetry"` + // CaptureStagingTelemetry makes a staging instance emit the plugin telemetry + // enabled above. Off, only a production instance emits it. + CaptureStagingTelemetry bool `json:"captureStagingTelemetry" toml:"captureStagingTelemetry"` } // IsZero reports whether no knob is set, so the caller can tell an absent block diff --git a/llo/pluginconfig/v31config_test.go b/llo/pluginconfig/v31config_test.go index fc66349..c77f891 100644 --- a/llo/pluginconfig/v31config_test.go +++ b/llo/pluginconfig/v31config_test.go @@ -33,7 +33,11 @@ func Test_V31Config_Unmarshal(t *testing.T) { maxDurationBlobObservation = "1s" blobInFlightWaitFactor = 4 maxBlobSnapshotAge = "-1s" - maxRoundPeriod = "3s"` + maxRoundPeriod = "3s" + captureOutcomeTelemetry = true + captureReportTelemetry = true + captureAttributedObservationTelemetry = true + captureStagingTelemetry = true` var pc PluginConfig require.NoError(t, toml.Unmarshal([]byte(rawToml), &pc)) @@ -46,6 +50,10 @@ func Test_V31Config_Unmarshal(t *testing.T) { assert.Equal(t, uint64(4), pc.V31.BlobInFlightWaitFactor) assert.Equal(t, -time.Second, pc.V31.MaxBlobSnapshotAge.Duration()) assert.Equal(t, 3*time.Second, pc.V31.MaxRoundPeriod.Duration()) + assert.True(t, pc.V31.CaptureOutcomeTelemetry) + assert.True(t, pc.V31.CaptureReportTelemetry) + assert.True(t, pc.V31.CaptureAttributedObservationTelemetry) + assert.True(t, pc.V31.CaptureStagingTelemetry) require.NoError(t, pc.V31.Validate()) }) diff --git a/llo/protocol/llo_plugin_telemetry.pb.go b/llo/protocol/llo_plugin_telemetry.pb.go index d1acc11..b50afa3 100644 --- a/llo/protocol/llo_plugin_telemetry.pb.go +++ b/llo/protocol/llo_plugin_telemetry.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.12 -// protoc v7.36.1 +// protoc v7.36.2 // source: llo_plugin_telemetry.proto package protocol @@ -21,8 +21,7 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) -// LLOOutcomeTelemetry sent on every call to Outcome (once per -// round) +// LLOOutcomeTelemetry sent on every call to Outcome (once per round) type LLOOutcomeTelemetry struct { state protoimpl.MessageState `protogen:"open.v1"` LifeCycleStage string `protobuf:"bytes,1,opt,name=life_cycle_stage,json=lifeCycleStage,proto3" json:"life_cycle_stage,omitempty"` @@ -306,6 +305,138 @@ func (x *LLOReportTelemetry) GetDonId() uint32 { return 0 } +// LLOAttributedObservationTelemetry carries the observation of one oracle in a +// production round (or staging, when enabled), as decoded by StateTransition. Large observations are split +// across several messages sharing (config_digest, seq_nr, observer), each with a +// stream values subset. Channel votes and decode_error are set only in the first message. +type LLOAttributedObservationTelemetry struct { + state protoimpl.MessageState `protogen:"open.v1"` + ConfigDigest []byte `protobuf:"bytes,1,opt,name=config_digest,json=configDigest,proto3" json:"config_digest,omitempty"` + SeqNr uint64 `protobuf:"varint,2,opt,name=seq_nr,json=seqNr,proto3" json:"seq_nr,omitempty"` + DonId uint32 `protobuf:"varint,3,opt,name=don_id,json=donId,proto3" json:"don_id,omitempty"` + Observer uint32 `protobuf:"varint,4,opt,name=observer,proto3" json:"observer,omitempty"` // oracle that made the observation + Emitter uint32 `protobuf:"varint,5,opt,name=emitter,proto3" json:"emitter,omitempty"` // oracle that emitted the telemetry + // the oracle and don observation_timestamps + OracleObservationTimestampNanoseconds uint64 `protobuf:"varint,7,opt,name=oracle_observation_timestamp_nanoseconds,json=oracleObservationTimestampNanoseconds,proto3" json:"oracle_observation_timestamp_nanoseconds,omitempty"` + AgreedObservationTimestampNanoseconds uint64 `protobuf:"varint,8,opt,name=agreed_observation_timestamp_nanoseconds,json=agreedObservationTimestampNanoseconds,proto3" json:"agreed_observation_timestamp_nanoseconds,omitempty"` + // stream id => value as observed by this oracle + StreamValues map[uint32]*LLOStreamValue `protobuf:"bytes,9,rep,name=stream_values,json=streamValues,proto3" json:"stream_values,omitempty" protobuf_key:"varint,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + // observation votes to update channels + RemoveChannelIds []uint32 `protobuf:"varint,11,rep,packed,name=remove_channel_ids,json=removeChannelIds,proto3" json:"remove_channel_ids,omitempty"` + UpdateChannelDefinitions map[uint32]*LLOChannelDefinitionProto `protobuf:"bytes,12,rep,name=update_channel_definitions,json=updateChannelDefinitions,proto3" json:"update_channel_definitions,omitempty" protobuf_key:"varint,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + // observation decode error if any + DecodeError *string `protobuf:"bytes,16,opt,name=decode_error,json=decodeError,proto3,oneof" json:"decode_error,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *LLOAttributedObservationTelemetry) Reset() { + *x = LLOAttributedObservationTelemetry{} + mi := &file_llo_plugin_telemetry_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *LLOAttributedObservationTelemetry) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*LLOAttributedObservationTelemetry) ProtoMessage() {} + +func (x *LLOAttributedObservationTelemetry) ProtoReflect() protoreflect.Message { + mi := &file_llo_plugin_telemetry_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use LLOAttributedObservationTelemetry.ProtoReflect.Descriptor instead. +func (*LLOAttributedObservationTelemetry) Descriptor() ([]byte, []int) { + return file_llo_plugin_telemetry_proto_rawDescGZIP(), []int{3} +} + +func (x *LLOAttributedObservationTelemetry) GetConfigDigest() []byte { + if x != nil { + return x.ConfigDigest + } + return nil +} + +func (x *LLOAttributedObservationTelemetry) GetSeqNr() uint64 { + if x != nil { + return x.SeqNr + } + return 0 +} + +func (x *LLOAttributedObservationTelemetry) GetDonId() uint32 { + if x != nil { + return x.DonId + } + return 0 +} + +func (x *LLOAttributedObservationTelemetry) GetObserver() uint32 { + if x != nil { + return x.Observer + } + return 0 +} + +func (x *LLOAttributedObservationTelemetry) GetEmitter() uint32 { + if x != nil { + return x.Emitter + } + return 0 +} + +func (x *LLOAttributedObservationTelemetry) GetOracleObservationTimestampNanoseconds() uint64 { + if x != nil { + return x.OracleObservationTimestampNanoseconds + } + return 0 +} + +func (x *LLOAttributedObservationTelemetry) GetAgreedObservationTimestampNanoseconds() uint64 { + if x != nil { + return x.AgreedObservationTimestampNanoseconds + } + return 0 +} + +func (x *LLOAttributedObservationTelemetry) GetStreamValues() map[uint32]*LLOStreamValue { + if x != nil { + return x.StreamValues + } + return nil +} + +func (x *LLOAttributedObservationTelemetry) GetRemoveChannelIds() []uint32 { + if x != nil { + return x.RemoveChannelIds + } + return nil +} + +func (x *LLOAttributedObservationTelemetry) GetUpdateChannelDefinitions() map[uint32]*LLOChannelDefinitionProto { + if x != nil { + return x.UpdateChannelDefinitions + } + return nil +} + +func (x *LLOAttributedObservationTelemetry) GetDecodeError() string { + if x != nil && x.DecodeError != nil { + return *x.DecodeError + } + return "" +} + var File_llo_plugin_telemetry_proto protoreflect.FileDescriptor const file_llo_plugin_telemetry_proto_rawDesc = "" + @@ -349,7 +480,26 @@ const file_llo_plugin_telemetry_proto_rawDesc = "" + "\x06seq_nr\x18\t \x01(\x04R\x05seqNr\x12#\n" + "\rconfig_digest\x18\n" + " \x01(\fR\fconfigDigest\x12\x15\n" + - "\x06don_id\x18\v \x01(\rR\x05donIdB\fZ\n" + + "\x06don_id\x18\v \x01(\rR\x05donId\"\xe8\x06\n" + + "!LLOAttributedObservationTelemetry\x12#\n" + + "\rconfig_digest\x18\x01 \x01(\fR\fconfigDigest\x12\x15\n" + + "\x06seq_nr\x18\x02 \x01(\x04R\x05seqNr\x12\x15\n" + + "\x06don_id\x18\x03 \x01(\rR\x05donId\x12\x1a\n" + + "\bobserver\x18\x04 \x01(\rR\bobserver\x12\x18\n" + + "\aemitter\x18\x05 \x01(\rR\aemitter\x12W\n" + + "(oracle_observation_timestamp_nanoseconds\x18\a \x01(\x04R%oracleObservationTimestampNanoseconds\x12W\n" + + "(agreed_observation_timestamp_nanoseconds\x18\b \x01(\x04R%agreedObservationTimestampNanoseconds\x12\\\n" + + "\rstream_values\x18\t \x03(\v27.v1.LLOAttributedObservationTelemetry.StreamValuesEntryR\fstreamValues\x12,\n" + + "\x12remove_channel_ids\x18\v \x03(\rR\x10removeChannelIds\x12\x81\x01\n" + + "\x1aupdate_channel_definitions\x18\f \x03(\v2C.v1.LLOAttributedObservationTelemetry.UpdateChannelDefinitionsEntryR\x18updateChannelDefinitions\x12&\n" + + "\fdecode_error\x18\x10 \x01(\tH\x00R\vdecodeError\x88\x01\x01\x1aS\n" + + "\x11StreamValuesEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\rR\x03key\x12(\n" + + "\x05value\x18\x02 \x01(\v2\x12.v1.LLOStreamValueR\x05value:\x028\x01\x1aj\n" + + "\x1dUpdateChannelDefinitionsEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\rR\x03key\x123\n" + + "\x05value\x18\x02 \x01(\v2\x1d.v1.LLOChannelDefinitionProtoR\x05value:\x028\x01B\x0f\n" + + "\r_decode_errorB\fZ\n" + ".;protocolb\x06proto3" var ( @@ -364,34 +514,41 @@ func file_llo_plugin_telemetry_proto_rawDescGZIP() []byte { return file_llo_plugin_telemetry_proto_rawDescData } -var file_llo_plugin_telemetry_proto_msgTypes = make([]protoimpl.MessageInfo, 7) +var file_llo_plugin_telemetry_proto_msgTypes = make([]protoimpl.MessageInfo, 10) var file_llo_plugin_telemetry_proto_goTypes = []any{ - (*LLOOutcomeTelemetry)(nil), // 0: v1.LLOOutcomeTelemetry - (*LLOAggregatorStreamValue)(nil), // 1: v1.LLOAggregatorStreamValue - (*LLOReportTelemetry)(nil), // 2: v1.LLOReportTelemetry - nil, // 3: v1.LLOOutcomeTelemetry.ChannelDefinitionsEntry - nil, // 4: v1.LLOOutcomeTelemetry.ValidAfterNanosecondsEntry - nil, // 5: v1.LLOOutcomeTelemetry.StreamAggregatesEntry - nil, // 6: v1.LLOAggregatorStreamValue.AggregatorValuesEntry - (*LLOStreamDefinition)(nil), // 7: v1.LLOStreamDefinition - (*LLOStreamValue)(nil), // 8: v1.LLOStreamValue - (*LLOChannelDefinitionProto)(nil), // 9: v1.LLOChannelDefinitionProto + (*LLOOutcomeTelemetry)(nil), // 0: v1.LLOOutcomeTelemetry + (*LLOAggregatorStreamValue)(nil), // 1: v1.LLOAggregatorStreamValue + (*LLOReportTelemetry)(nil), // 2: v1.LLOReportTelemetry + (*LLOAttributedObservationTelemetry)(nil), // 3: v1.LLOAttributedObservationTelemetry + nil, // 4: v1.LLOOutcomeTelemetry.ChannelDefinitionsEntry + nil, // 5: v1.LLOOutcomeTelemetry.ValidAfterNanosecondsEntry + nil, // 6: v1.LLOOutcomeTelemetry.StreamAggregatesEntry + nil, // 7: v1.LLOAggregatorStreamValue.AggregatorValuesEntry + nil, // 8: v1.LLOAttributedObservationTelemetry.StreamValuesEntry + nil, // 9: v1.LLOAttributedObservationTelemetry.UpdateChannelDefinitionsEntry + (*LLOStreamDefinition)(nil), // 10: v1.LLOStreamDefinition + (*LLOStreamValue)(nil), // 11: v1.LLOStreamValue + (*LLOChannelDefinitionProto)(nil), // 12: v1.LLOChannelDefinitionProto } var file_llo_plugin_telemetry_proto_depIdxs = []int32{ - 3, // 0: v1.LLOOutcomeTelemetry.channel_definitions:type_name -> v1.LLOOutcomeTelemetry.ChannelDefinitionsEntry - 4, // 1: v1.LLOOutcomeTelemetry.valid_after_nanoseconds:type_name -> v1.LLOOutcomeTelemetry.ValidAfterNanosecondsEntry - 5, // 2: v1.LLOOutcomeTelemetry.stream_aggregates:type_name -> v1.LLOOutcomeTelemetry.StreamAggregatesEntry - 6, // 3: v1.LLOAggregatorStreamValue.aggregator_values:type_name -> v1.LLOAggregatorStreamValue.AggregatorValuesEntry - 7, // 4: v1.LLOReportTelemetry.stream_definitions:type_name -> v1.LLOStreamDefinition - 8, // 5: v1.LLOReportTelemetry.stream_values:type_name -> v1.LLOStreamValue - 9, // 6: v1.LLOOutcomeTelemetry.ChannelDefinitionsEntry.value:type_name -> v1.LLOChannelDefinitionProto - 1, // 7: v1.LLOOutcomeTelemetry.StreamAggregatesEntry.value:type_name -> v1.LLOAggregatorStreamValue - 8, // 8: v1.LLOAggregatorStreamValue.AggregatorValuesEntry.value:type_name -> v1.LLOStreamValue - 9, // [9:9] is the sub-list for method output_type - 9, // [9:9] is the sub-list for method input_type - 9, // [9:9] is the sub-list for extension type_name - 9, // [9:9] is the sub-list for extension extendee - 0, // [0:9] is the sub-list for field type_name + 4, // 0: v1.LLOOutcomeTelemetry.channel_definitions:type_name -> v1.LLOOutcomeTelemetry.ChannelDefinitionsEntry + 5, // 1: v1.LLOOutcomeTelemetry.valid_after_nanoseconds:type_name -> v1.LLOOutcomeTelemetry.ValidAfterNanosecondsEntry + 6, // 2: v1.LLOOutcomeTelemetry.stream_aggregates:type_name -> v1.LLOOutcomeTelemetry.StreamAggregatesEntry + 7, // 3: v1.LLOAggregatorStreamValue.aggregator_values:type_name -> v1.LLOAggregatorStreamValue.AggregatorValuesEntry + 10, // 4: v1.LLOReportTelemetry.stream_definitions:type_name -> v1.LLOStreamDefinition + 11, // 5: v1.LLOReportTelemetry.stream_values:type_name -> v1.LLOStreamValue + 8, // 6: v1.LLOAttributedObservationTelemetry.stream_values:type_name -> v1.LLOAttributedObservationTelemetry.StreamValuesEntry + 9, // 7: v1.LLOAttributedObservationTelemetry.update_channel_definitions:type_name -> v1.LLOAttributedObservationTelemetry.UpdateChannelDefinitionsEntry + 12, // 8: v1.LLOOutcomeTelemetry.ChannelDefinitionsEntry.value:type_name -> v1.LLOChannelDefinitionProto + 1, // 9: v1.LLOOutcomeTelemetry.StreamAggregatesEntry.value:type_name -> v1.LLOAggregatorStreamValue + 11, // 10: v1.LLOAggregatorStreamValue.AggregatorValuesEntry.value:type_name -> v1.LLOStreamValue + 11, // 11: v1.LLOAttributedObservationTelemetry.StreamValuesEntry.value:type_name -> v1.LLOStreamValue + 12, // 12: v1.LLOAttributedObservationTelemetry.UpdateChannelDefinitionsEntry.value:type_name -> v1.LLOChannelDefinitionProto + 13, // [13:13] is the sub-list for method output_type + 13, // [13:13] is the sub-list for method input_type + 13, // [13:13] is the sub-list for extension type_name + 13, // [13:13] is the sub-list for extension extendee + 0, // [0:13] is the sub-list for field type_name } func init() { file_llo_plugin_telemetry_proto_init() } @@ -400,13 +557,14 @@ func file_llo_plugin_telemetry_proto_init() { return } file_plugin_codecs_proto_init() + file_llo_plugin_telemetry_proto_msgTypes[3].OneofWrappers = []any{} type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_llo_plugin_telemetry_proto_rawDesc), len(file_llo_plugin_telemetry_proto_rawDesc)), NumEnums: 0, - NumMessages: 7, + NumMessages: 10, NumExtensions: 0, NumServices: 0, }, diff --git a/llo/protocol/llo_plugin_telemetry.proto b/llo/protocol/llo_plugin_telemetry.proto index 0a321fd..116d84d 100644 --- a/llo/protocol/llo_plugin_telemetry.proto +++ b/llo/protocol/llo_plugin_telemetry.proto @@ -5,8 +5,7 @@ option go_package = ".;protocol"; import "plugin_codecs.proto"; -// LLOOutcomeTelemetry sent on every call to Outcome (once per -// round) +// LLOOutcomeTelemetry sent on every call to Outcome (once per round) message LLOOutcomeTelemetry { string life_cycle_stage = 1; uint64 observation_timestamp_nanoseconds = 2; @@ -45,3 +44,29 @@ message LLOReportTelemetry { bytes config_digest = 10; uint32 don_id = 11; } + +// LLOAttributedObservationTelemetry carries the observation of one oracle in a +// production round (or staging, when enabled), as decoded by StateTransition. Large observations are split +// across several messages sharing (config_digest, seq_nr, observer), each with a +// stream values subset. Channel votes and decode_error are set only in the first message. +message LLOAttributedObservationTelemetry { + bytes config_digest = 1; + uint64 seq_nr = 2; + uint32 don_id = 3; + uint32 observer = 4; // oracle that made the observation + uint32 emitter = 5; // oracle that emitted the telemetry + + // the oracle and don observation_timestamps + uint64 oracle_observation_timestamp_nanoseconds = 7; + uint64 agreed_observation_timestamp_nanoseconds = 8; + + // stream id => value as observed by this oracle + map stream_values = 9; + + // observation votes to update channels + repeated uint32 remove_channel_ids = 11; + map update_channel_definitions = 12; + + // observation decode error if any + optional string decode_error = 16; +} diff --git a/llo/protocol/staging_marker.go b/llo/protocol/staging_marker.go new file mode 100644 index 0000000..542ab65 --- /dev/null +++ b/llo/protocol/staging_marker.go @@ -0,0 +1,43 @@ +package protocol + +import ( + "bytes" + "encoding/json" + + llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" + + ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" +) + +// A staging marker is the last report of a staging round. It is never +// transmitted: the transmitter drops it after notifying the transmit listeners, +// which flushes the telemetry buffered for that round. A staging instance has +// no other report guaranteed to reach the transmitter. +// +// It travels as a JSON report, which the keyring already signs, with its own +// shape so it can never be mistaken for a channel report. + +type stagingMarker struct { + StagingMarker stagingMarkerBody `json:"stagingMarker"` +} + +type stagingMarkerBody struct { + ConfigDigest ocrtypes.ConfigDigest `json:"configDigest"` + SeqNr uint64 `json:"seqNr"` +} + +// stagingMarkerPrefix is how every encoded staging marker starts. JSON channel +// reports start with their first field, ConfigDigest. +var stagingMarkerPrefix = []byte(`{"stagingMarker":`) + +// EncodeStagingMarker encodes the staging marker of seqNr. +func EncodeStagingMarker(configDigest ocrtypes.ConfigDigest, seqNr uint64) ([]byte, error) { + return json.Marshal(stagingMarker{stagingMarkerBody{ConfigDigest: configDigest, SeqNr: seqNr}}) +} + +// IsStagingMarker reports whether report is a staging marker. +func IsStagingMarker(info llotypes.ReportInfo, report []byte) bool { + return info.LifeCycleStage == LifeCycleStageStaging && + info.ReportFormat == llotypes.ReportFormatJSON && + bytes.HasPrefix(report, stagingMarkerPrefix) +} diff --git a/llo/protocol/staging_marker_test.go b/llo/protocol/staging_marker_test.go new file mode 100644 index 0000000..26a1e87 --- /dev/null +++ b/llo/protocol/staging_marker_test.go @@ -0,0 +1,36 @@ +package protocol + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" + + llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" + + ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" +) + +func Test_StagingMarker(t *testing.T) { + staging := llotypes.ReportInfo{LifeCycleStage: LifeCycleStageStaging, ReportFormat: llotypes.ReportFormatJSON} + + marker, err := EncodeStagingMarker(ocrtypes.ConfigDigest{1, 2, 3}, 42) + require.NoError(t, err) + require.True(t, IsStagingMarker(staging, marker)) + + var decoded stagingMarker + require.NoError(t, json.Unmarshal(marker, &decoded)) + require.Equal(t, ocrtypes.ConfigDigest{1, 2, 3}, decoded.StagingMarker.ConfigDigest) + require.Equal(t, uint64(42), decoded.StagingMarker.SeqNr) + + t.Run("requires the staging stage and JSON format", func(t *testing.T) { + require.False(t, IsStagingMarker(llotypes.ReportInfo{LifeCycleStage: LifeCycleStageProduction, ReportFormat: llotypes.ReportFormatJSON}, marker)) + require.False(t, IsStagingMarker(llotypes.ReportInfo{LifeCycleStage: LifeCycleStageStaging, ReportFormat: llotypes.ReportFormatEVMPremiumLegacy}, marker)) + }) + + t.Run("a JSON channel report is not a marker", func(t *testing.T) { + // JSON channel reports start with their first field, ConfigDigest. + report := []byte(`{"ConfigDigest":"0102","SeqNr":42,"ChannelID":0,"Specimen":true}`) + require.False(t, IsStagingMarker(staging, report)) + }) +} diff --git a/llo/transmitter/transmitter.go b/llo/transmitter/transmitter.go index bb68c1d..b2158ff 100644 --- a/llo/transmitter/transmitter.go +++ b/llo/transmitter/transmitter.go @@ -18,6 +18,7 @@ import ( llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" "github.com/smartcontractkit/chainlink-data-streams/llo/pluginconfig" + "github.com/smartcontractkit/chainlink-data-streams/llo/protocol" mercurytransmitter "github.com/smartcontractkit/chainlink-data-streams/llo/transmitter/dataengine" ) @@ -175,6 +176,12 @@ func (t *transmitter) Transmit( } t.notify(digest, seqNr) + if protocol.IsStagingMarker(report.Info, report.Report) { + // The staging marker is never transmitted. It exists to drive the + // notification above, which flushes the telemetry buffered for seqNr. + return nil + } + g := new(errgroup.Group) for _, st := range t.subTransmitters { g.Go(func() error { diff --git a/llo/transmitter/transmitter_staging_test.go b/llo/transmitter/transmitter_staging_test.go new file mode 100644 index 0000000..96ba220 --- /dev/null +++ b/llo/transmitter/transmitter_staging_test.go @@ -0,0 +1,75 @@ +package transmitter + +import ( + "context" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/services" + llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" + "github.com/smartcontractkit/chainlink-common/pkg/utils/tests" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" + "github.com/smartcontractkit/libocr/offchainreporting2plus/types" + + "github.com/smartcontractkit/chainlink-data-streams/llo/protocol" +) + +type countingSubTransmitter struct { + services.Service + calls atomic.Int32 +} + +func (c *countingSubTransmitter) Transmit(context.Context, types.ConfigDigest, uint64, ocr3types.ReportWithInfo[llotypes.ReportInfo], []types.AttributedOnchainSignature) error { + c.calls.Add(1) + return nil +} + +func (c *countingSubTransmitter) FromAccount(context.Context) (types.Account, error) { + return "", nil +} + +func Test_transmitter_StagingMarkerOnlyNotifies(t *testing.T) { + ctx := tests.Context(t) + sub := &countingSubTransmitter{} + tr := &transmitter{ + lggr: logger.Test(t), + subTransmitters: []Transmitter{sub}, + onTransmit: &onTransmit{lggr: logger.Test(t)}, + } + type notification struct { + digest types.ConfigDigest + seqNr uint64 + } + notified := make(chan notification, 4) + tr.OnTransmit(func(digest types.ConfigDigest, seqNr uint64) { notified <- notification{digest, seqNr} }) + + jsonReport := func(stage llotypes.LifeCycleStage, payload []byte) ocr3types.ReportWithInfo[llotypes.ReportInfo] { + return ocr3types.ReportWithInfo[llotypes.ReportInfo]{ + Report: payload, + Info: llotypes.ReportInfo{LifeCycleStage: stage, ReportFormat: llotypes.ReportFormatJSON}, + } + } + marker, err := protocol.EncodeStagingMarker(types.ConfigDigest{2}, 7) + require.NoError(t, err) + + t.Run("staging marker notifies and is not transmitted", func(t *testing.T) { + require.NoError(t, tr.Transmit(ctx, types.ConfigDigest{2}, 7, jsonReport(protocol.LifeCycleStageStaging, marker), nil)) + require.Equal(t, notification{types.ConfigDigest{2}, 7}, <-notified) + require.Zero(t, sub.calls.Load()) + }) + + t.Run("staging specimen notifies and is transmitted", func(t *testing.T) { + require.NoError(t, tr.Transmit(ctx, types.ConfigDigest{2}, 8, jsonReport(protocol.LifeCycleStageStaging, []byte(`{"ConfigDigest":"02","Specimen":true}`)), nil)) + require.Equal(t, notification{types.ConfigDigest{2}, 8}, <-notified) + require.Equal(t, int32(1), sub.calls.Load()) + }) + + t.Run("production report notifies and is transmitted", func(t *testing.T) { + require.NoError(t, tr.Transmit(ctx, types.ConfigDigest{1}, 9, jsonReport(protocol.LifeCycleStageProduction, []byte(`{}`)), nil)) + require.Equal(t, notification{types.ConfigDigest{1}, 9}, <-notified) + require.Equal(t, int32(2), sub.calls.Load()) + }) +}