From 62d7d3c5e1eebc844151b1b3caeb1db32a1e5b21 Mon Sep 17 00:00:00 2001 From: Bruno Moura Date: Thu, 8 Oct 2026 16:46:34 +0100 Subject: [PATCH 1/5] llo/protocol: add attributed observation telemetry message Add LLOAttributedObservationTelemetry, carrying the decoded observation of oracles in a round: stream values, channel votes, the oracle and agreed observation timestamps, and the decode error if the observation was dropped. Large observations are split across messages sharing (config_digest, seq_nr, observer). Used in v31 instead of the messagePropose. --- llo/protocol/llo_plugin_telemetry.pb.go | 218 ++++++++++++++++++++---- llo/protocol/llo_plugin_telemetry.proto | 29 +++- 2 files changed, 215 insertions(+), 32 deletions(-) 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; +} From c921f4cd318f24445d68782f2e96e598151aeeed Mon Sep 17 00:00:00 2001 From: Bruno Moura Date: Thu, 8 Oct 2026 16:50:39 +0100 Subject: [PATCH 2/5] llo: add staging marker reports Add a staging marker report: a JSON report with its own shape that a staging round appends last, so a staging instance always has a report reaching the transmitter. Allowing the transmitter to notify attached listeners also for a staging instance --- llo/protocol/staging_marker.go | 43 ++++++++++++ llo/protocol/staging_marker_test.go | 36 ++++++++++ llo/transmitter/transmitter.go | 7 ++ llo/transmitter/transmitter_staging_test.go | 75 +++++++++++++++++++++ 4 files changed, 161 insertions(+) create mode 100644 llo/protocol/staging_marker.go create mode 100644 llo/protocol/staging_marker_test.go create mode 100644 llo/transmitter/transmitter_staging_test.go 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()) + }) +} From 1056f9d7aafa11747884c2d952c4d4d7d6b5c465 Mon Sep 17 00:00:00 2001 From: Bruno Moura Date: Thu, 8 Oct 2026 17:00:16 +0100 Subject: [PATCH 3/5] llo/pluginconfig: add v31 plugin telemetry knobs Add captureOutcomeTelemetry, captureReportTelemetry and captureAttributedObservationTelemetry to the v31 job config, so v31 plugin telemetry is enabled per job instead of by the node CaptureEATelemetry setting, which keeps driving data source telemetry. Add captureStagingTelemetry to let a staging instance emit it too. All are off by default. --- llo/pluginconfig/v31config.go | 14 ++++++++++++++ llo/pluginconfig/v31config_test.go | 10 +++++++++- 2 files changed, 23 insertions(+), 1 deletion(-) 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()) }) From 4f1c437bfcb6e162ba020efd87a6fcb81067406f Mon Sep 17 00:00:00 2001 From: Bruno Moura Date: Thu, 8 Oct 2026 17:02:51 +0100 Subject: [PATCH 4/5] llo/dev/v31: emit attributed observation telemetry Emit, from StateTransition, the decoded observation of every oracle in the round as LLOAttributedObservationTelemetry, built and split off the StateTransition path by a plugin owned telemeter. Outcome and attributed observation telemetry are emitted by f+1 oracles rotating with seqNr. Plugin telemetry is emitted for production rounds, and for staging rounds with CaptureStagingTelemetry. --- llo/dev/v31/doc.go | 16 +- llo/dev/v31/factory.go | 17 +- llo/dev/v31/handover_test.go | 6 + llo/dev/v31/plugin.go | 13 + llo/dev/v31/plugin_test.go | 4 +- llo/dev/v31/reports.go | 34 +- llo/dev/v31/statetransition.go | 14 +- llo/dev/v31/telemetry.go | 241 +++++++++++- llo/dev/v31/telemetry_test.go | 652 +++++++++++++++++++++++++++++++++ 9 files changed, 975 insertions(+), 22 deletions(-) create mode 100644 llo/dev/v31/telemetry_test.go 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..f3deaeb 100644 --- a/llo/dev/v31/factory.go +++ b/llo/dev/v31/factory.go @@ -30,10 +30,17 @@ 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). 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 +149,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 +238,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..ffa10a4 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. +func (p *Plugin) captureReportTelemetry(r protocol.Report, cd llotypes.ChannelDefinition, stage llotypes.LifeCycleStage) { + if p.ReportTelemetryCh == nil || !p.emitsTelemetry(stage) { 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 outcome and attributed observation 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..1acfb08 --- /dev/null +++ b/llo/dev/v31/telemetry_test.go @@ -0,0 +1,652 @@ +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) { + run := func(t *testing.T, predecessor *ocrtypes.ConfigDigest, stagingTelemetry 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 + + ts := handoverTickNanos + v.round(Observation{}) + require.Equal(t, stage, v.lifeCycleStage()) + ts += handoverTickNanos + v.round(Observation{ + UnixTimestampNanoseconds: ts, + UpdateChannelDefinitions: llotypes.ChannelDefinitions{handoverChannelID: handoverChannel()}, + StreamValues: handoverStreamValues(), + }) + for range 3 { + ts += handoverTickNanos + v.round(Observation{UnixTimestampNanoseconds: ts, StreamValues: handoverStreamValues()}) + } + require.NotEmpty(t, v.reports, "the channel must be reporting") + return v, rtCh + } + + // 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) + } + }) +} + +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 } From 27c662575a869d4414a76efd8c8736b1a85272a9 Mon Sep 17 00:00:00 2001 From: Bruno Moura Date: Thu, 8 Oct 2026 19:11:55 +0100 Subject: [PATCH 5/5] llo/dev/v31: emit report telemetry in same plugin schedule --- llo/dev/v31/factory.go | 3 ++- llo/dev/v31/telemetry.go | 6 +++--- llo/dev/v31/telemetry_test.go | 28 ++++++++++++++++++++++++---- 3 files changed, 29 insertions(+), 8 deletions(-) diff --git a/llo/dev/v31/factory.go b/llo/dev/v31/factory.go index f3deaeb..395adfa 100644 --- a/llo/dev/v31/factory.go +++ b/llo/dev/v31/factory.go @@ -35,7 +35,8 @@ type PluginFactoryParams struct { // where this oracle is one of the f+1 rotating emitters. OutcomeTelemetryCh chan<- *protocol.LLOOutcomeTelemetry // ReportTelemetryCh, if set, receives one telemetry struct per emitted - // production report (and staging, with CaptureStagingTelemetry). + // 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 diff --git a/llo/dev/v31/telemetry.go b/llo/dev/v31/telemetry.go index ffa10a4..aa6cf51 100644 --- a/llo/dev/v31/telemetry.go +++ b/llo/dev/v31/telemetry.go @@ -76,9 +76,9 @@ func makeOutcomeTelemetry(out precursor, configDigest ocrtypes.ConfigDigest, seq } // captureReportTelemetry emits the telemetry of reports produced in a stage -// that emits telemetry. +// 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) { + if p.ReportTelemetryCh == nil || !p.emitsTelemetry(stage) || !p.isTelemetryEmitter(r.SeqNr) { return } rt, err := makeReportTelemetry(r, cd, p.DonID) @@ -159,7 +159,7 @@ type attributedObservationRound struct { } // isTelemetryEmitter reports whether this oracle is one of the f+1 oracles -// emitting outcome and attributed observation telemetry for this seqNr. +// emitting plugin telemetry for this seqNr. func (p *Plugin) isTelemetryEmitter(seqNr uint64) bool { if p.N <= 0 { return false diff --git a/llo/dev/v31/telemetry_test.go b/llo/dev/v31/telemetry_test.go index 1acfb08..3ed37ce 100644 --- a/llo/dev/v31/telemetry_test.go +++ b/llo/dev/v31/telemetry_test.go @@ -539,29 +539,43 @@ func Test_OutcomeTelemetry_Emitters(t *testing.T) { } func Test_ReportTelemetry_Lifecycle(t *testing.T) { - run := func(t *testing.T, predecessor *ocrtypes.ConfigDigest, stagingTelemetry bool, stage llotypes.LifeCycleStage) (*v31Instance, chan *protocol.LLOReportTelemetry) { + // 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 - v.round(Observation{}) + round(Observation{}) require.Equal(t, stage, v.lifeCycleStage()) ts += handoverTickNanos - v.round(Observation{ + round(Observation{ UnixTimestampNanoseconds: ts, UpdateChannelDefinitions: llotypes.ChannelDefinitions{handoverChannelID: handoverChannel()}, StreamValues: handoverStreamValues(), }) for range 3 { ts += handoverTickNanos - v.round(Observation{UnixTimestampNanoseconds: ts, StreamValues: handoverStreamValues()}) + 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) { @@ -596,6 +610,12 @@ func Test_ReportTelemetry_Lifecycle(t *testing.T) { 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) {