Implement report telemetry for LLO

core/scripts/go.mod

@@ -33,7 +33,7 @@ require (
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/chainlink-automation v0.8.1
 	github.com/smartcontractkit/chainlink-common v0.4.2-0.20250228151513-8777dbcefd5c
-	github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250228145850-f846693a6fd5
+	github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250311123514-b4881a96af5a
 	github.com/smartcontractkit/chainlink-integrations/evm v0.0.0-20250304212534-9eff2fa32bea
 	github.com/smartcontractkit/chainlink-testing-framework/lib v1.50.22
 	github.com/smartcontractkit/libocr v0.0.0-20250220133800-f3b940c4f298

core/scripts/go.sum

@@ -1124,8 +1124,8 @@ github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250307105933-7
 github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250307105933-7912a5e97ad0/go.mod h1:2Y6JRpvj9EkgKgymvenuqCnra07+NYVB6+0rQGETSGA=
 github.com/smartcontractkit/chainlink-common v0.4.2-0.20250228151513-8777dbcefd5c h1:pI4Y9C+EXpZmAcmvcvrCA4REWSyvCO3LQw4pGU1a/4M=
 github.com/smartcontractkit/chainlink-common v0.4.2-0.20250228151513-8777dbcefd5c/go.mod h1:YQuXIqQpmpAqstWV0LHaDTJ5nsSWuip5ivEM+Fisb+4=
-github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250228145850-f846693a6fd5 h1:cJpPJ5hEwc6vlMoxmATS60uWPUi62ydnTVBabu6WKEE=
-github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250228145850-f846693a6fd5/go.mod h1:2yUpKW1/jFxpozO/Zkh3fKDzI0jthXoEcU2xuDq+vlo=
+github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250311123514-b4881a96af5a h1:9TAVhJT/mWsJjQxiLoJvXdapzcViH5uxwxRhA0qtalU=
+github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250311123514-b4881a96af5a/go.mod h1:2yUpKW1/jFxpozO/Zkh3fKDzI0jthXoEcU2xuDq+vlo=
 github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 h1:8u9xUrC+yHrTDexOKDd+jrA6LCzFFHeX1G82oj2fsSI=
 github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135/go.mod h1:NkvE4iQgiT7dMCP6U3xPELHhWhN5Xr6rHC0axRebyMU=
 github.com/smartcontractkit/chainlink-framework/chains v0.0.0-20250227163723-3c71fefea680 h1:Es/V5imh4at3NdHDWlbDMjljd24TrJQCy/+8xAVajl0=

core/services/llo/data_source.go

@@ -119,7 +119,7 @@ func (d *dataSource) Observe(ctx context.Context, streamValues llo.StreamValues,
 		// Size needs to accommodate the max number of telemetry events that could be generated
 		// Standard case might be about 3 bridge requests per spec and one stream<=>spec
 		// Overallocate for safety (to avoid dropping packets)
-		telemCh := d.t.MakeTelemChannel(opts, 10*len(streamValues))
+		telemCh := d.t.MakeObservationTelemetryCh(opts, 10*len(streamValues))
 		if telemCh != nil {
 			ctx = pipeline.WithTelemetryCh(ctx, telemCh)
 			// After all Observations have returned, nothing else will be sent to the

core/services/llo/data_source_test.go

@@ -23,6 +23,7 @@ import (
 	llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo"
 
 	"github.com/smartcontractkit/chainlink-data-streams/llo"
+	datastreamsllo "github.com/smartcontractkit/chainlink-data-streams/llo"
 	"github.com/smartcontractkit/chainlink/v2/core/bridges"
 	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils"
 	clhttptest "github.com/smartcontractkit/chainlink/v2/core/internal/testutils/httptest"
@@ -117,11 +118,15 @@ func (m *mockTelemeter) EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.
 	m.v3PremiumLegacyPackets = append(m.v3PremiumLegacyPackets, v3PremiumLegacyPacket{run, trrs, streamID, opts, val, err})
 }
 
-func (m *mockTelemeter) MakeTelemChannel(opts llo.DSOpts, size int) (ch chan<- interface{}) {
+func (m *mockTelemeter) MakeObservationTelemetryCh(opts llo.DSOpts, size int) (ch chan<- interface{}) {
 	m.ch = make(chan interface{}, size)
 	return m.ch
 }
 
+func (m *mockTelemeter) GetReportTelemetryCh() chan<- *datastreamsllo.LLOReportTelemetry {
+	return nil
+}
+
 func Test_DataSource(t *testing.T) {
 	lggr := logger.TestLogger(t)
 	reg := &mockRegistry{make(map[streams.StreamID]*mockPipeline)}

core/services/llo/delegate.go

@@ -39,7 +39,7 @@ type delegate struct {
 
 	src   datastreamsllo.ShouldRetireCache
 	ds    datastreamsllo.DataSource
-	telem services.Service
+	telem TelemeterService
 
 	oracles []Closer
 }
@@ -155,15 +155,18 @@ func (d *delegate) Start(ctx context.Context) error {
 				OnchainKeyring:               d.cfg.OnchainKeyring,
 				ReportingPluginFactory: promwrapper.NewReportingPluginFactory(
 					datastreamsllo.NewPluginFactory(
-						d.cfg.ReportingPluginConfig,
-						psrrc,
-						d.src,
-						d.cfg.RetirementReportCodec,
-						d.cfg.ChannelDefinitionCache,
-						d.ds,
-						logger.Named(lggr, "ReportingPlugin"),
-						llo.EVMOnchainConfigCodec{},
-						d.reportCodecs,
+						datastreamsllo.PluginFactoryParams{
+							Config:                           d.cfg.ReportingPluginConfig,
+							PredecessorRetirementReportCache: psrrc,
+							ShouldRetireCache:                d.src,
+							RetirementReportCodec:            d.cfg.RetirementReportCodec,
+							ChannelDefinitionCache:           d.cfg.ChannelDefinitionCache,
+							DataSource:                       d.ds,
+							Logger:                           logger.Named(lggr, "ReportingPlugin"),
+							OnchainConfigCodec:               llo.EVMOnchainConfigCodec{},
+							ReportCodecs:                     d.reportCodecs,
+							ReportTelemetryCh:                d.telem.GetReportTelemetryCh(),
+						},
 					),
 					lggr,
 					d.cfg.ChainID,

core/services/llo/telem/telem_streams.proto

@@ -24,6 +24,8 @@ message LLOBridgeTelemetry {
     int64 observation_timestamp = 15;
 }
 
+// LLOObservationTelemetry packet sent for each stream on every call to
+// Observation (once per round)
 message LLOObservationTelemetry {
     uint32 stream_id = 1;
     int32 stream_value_type = 2;

core/services/llo/telemetry.go

@@ -12,6 +12,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
 	"github.com/smartcontractkit/chainlink-data-streams/llo"
+	datastreamsllo "github.com/smartcontractkit/chainlink-data-streams/llo"
 
 	"github.com/smartcontractkit/chainlink/v2/core/services/llo/evm"
 	"github.com/smartcontractkit/chainlink/v2/core/services/llo/telem"
@@ -26,7 +27,8 @@ const adapterLWBAErrorName = "AdapterLWBAError"
 
 type Telemeter interface {
 	EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.TaskRunResults, streamID uint32, opts llo.DSOpts, val llo.StreamValue, err error)
-	MakeTelemChannel(opts llo.DSOpts, size int) (ch chan<- interface{})
+	MakeObservationTelemetryCh(opts llo.DSOpts, size int) (ch chan<- interface{})
+	GetReportTelemetryCh() chan<- *datastreamsllo.LLOReportTelemetry
 }
 
 type TelemeterService interface {
@@ -51,11 +53,19 @@ func newTelemeter(lggr logger.Logger, monitoringEndpoint commontypes.MonitoringE
 		chTelemetryPipeline: chTelemetryPipeline,
 		monitoringEndpoint:  monitoringEndpoint,
 		donID:               donID,
-		chch:                make(chan telemetryCollectionContext, 1000), // chch should be consumed from very quickly so we don't need a large buffer, but it also won't hurt
+		chch:                make(chan telemetryCollectionContext, 1000),                                        // chch should be consumed from very quickly so we don't need a large buffer, but it also won't hurt
+		chReportTelemetry:   make(chan *datastreamsllo.LLOReportTelemetry, (2+2)*datastreamsllo.MaxReportCount), // 2 instances+2x size safety buffer
 	}
 	t.Service, t.eng = services.Config{
 		Name:  "LLOTelemeterService",
 		Start: t.start,
+		Close: func() error {
+			// Enforce that it's not used after close
+			close(t.chTelemetryPipeline)
+			close(t.chch)
+			close(t.chReportTelemetry)
+			return nil
+		},
 	}.NewServiceEngine(lggr)
 
 	return t
@@ -69,6 +79,7 @@ type telemeter struct {
 	chTelemetryPipeline chan *TelemetryPipeline
 	donID               uint32
 	chch                chan telemetryCollectionContext
+	chReportTelemetry   chan *datastreamsllo.LLOReportTelemetry
 }
 
 func (t *telemeter) EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.TaskRunResults, streamID uint32, opts llo.DSOpts, val llo.StreamValue, err error) {
@@ -98,13 +109,16 @@ type telemetryCollectionContext struct {
 	opts llo.DSOpts
 }
 
-// CollectTelemetryObserve reads telem packets from the returned channel and
-// sends them to the monitoring endpoint. Stops reading when channel closed or
-// when telemeter is stopped
+// MakeObservationTelemetryCh reads telem packets from the returned channel and sends them
+// to the monitoring endpoint. Stops reading when channel closed or when
+// telemeter is stopped
 //
 // CALLER IS RESPONSIBLE FOR CLOSING THE RETURNED CHANNEL TO AVOID MEMORY
 // LEAKS.
-func (t *telemeter) MakeTelemChannel(opts llo.DSOpts, size int) chan<- interface{} {
+//
+// It is necessary to make a new channel for every Observation call because it
+// closes over DSOpts which is scoped to that call only.
+func (t *telemeter) MakeObservationTelemetryCh(opts llo.DSOpts, size int) chan<- interface{} {
 	ch := make(chan interface{}, size)
 	tcc := telemetryCollectionContext{
 		in:   ch,
@@ -123,6 +137,10 @@ func (t *telemeter) MakeTelemChannel(opts llo.DSOpts, size int) chan<- interface
 	return ch
 }
 
+func (t *telemeter) GetReportTelemetryCh() chan<- *datastreamsllo.LLOReportTelemetry {
+	return t.chReportTelemetry
+}
+
 func (t *telemeter) start(_ context.Context) error {
 	t.eng.Go(func(ctx context.Context) {
 		wg := sync.WaitGroup{}
@@ -139,13 +157,15 @@ func (t *telemeter) start(_ context.Context) error {
 								// channel closed by producer
 								return
 							}
-							t.collectTelemetry(p, tcc.opts)
+							t.collectObservationTelemetry(p, tcc.opts)
 						case <-ctx.Done():
 							return
 						}
 					}
 				}()
 
+			case rt := <-t.chReportTelemetry:
+				t.collectReportTelemetry(rt)
 			case p := <-t.chTelemetryPipeline:
 				t.collectV3PremiumLegacyTelemetry(p)
 			case <-ctx.Done():
@@ -157,7 +177,7 @@ func (t *telemeter) start(_ context.Context) error {
 	return nil
 }
 
-func (t *telemeter) collectTelemetry(p interface{}, opts llo.DSOpts) {
+func (t *telemeter) collectObservationTelemetry(p interface{}, opts llo.DSOpts) {
 	var msg proto.Message
 	switch v := p.(type) {
 	case *pipeline.BridgeTelemetry:
@@ -195,6 +215,15 @@ func (t *telemeter) collectTelemetry(p interface{}, opts llo.DSOpts) {
 	t.monitoringEndpoint.SendLog(bytes)
 }
 
+func (t *telemeter) collectReportTelemetry(rt *datastreamsllo.LLOReportTelemetry) {
+	bytes, err := proto.Marshal(rt)
+	if err != nil {
+		t.eng.Warnf("protobuf marshal failed %v", err.Error())
+		return
+	}
+	t.monitoringEndpoint.SendLog(bytes)
+}
+
 func (t *telemeter) collectV3PremiumLegacyTelemetry(d *TelemetryPipeline) {
 	eaTelemetryValues := ocrcommon.ParseMercuryEATelemetry(t.eng.SugaredLogger, d.trrs, mercuryutils.REPORT_V3)
 	for _, eaTelem := range eaTelemetryValues {
@@ -276,7 +305,10 @@ type nullTelemeter struct{}
 
 func (t *nullTelemeter) EnqueueV3PremiumLegacy(run *pipeline.Run, trrs pipeline.TaskRunResults, streamID uint32, opts llo.DSOpts, val llo.StreamValue, err error) {
 }
-func (t *nullTelemeter) MakeTelemChannel(opts llo.DSOpts, size int) (ch chan<- interface{}) {
+func (t *nullTelemeter) MakeObservationTelemetryCh(opts llo.DSOpts, size int) (ch chan<- interface{}) {
+	return nil
+}
+func (t *nullTelemeter) GetReportTelemetryCh() chan<- *datastreamsllo.LLOReportTelemetry {
 	return nil
 }
 func (t *nullTelemeter) Start(context.Context) error {

core/services/llo/telemetry_test.go

@@ -110,6 +110,10 @@ var trrs = pipeline.TaskRunResults{
 	},
 }
 
+func Test_Telemeter_reportTelemetry(t *testing.T) {
+	t.Fatal("TODO")
+}
+
 func Test_Telemeter_v3PremiumLegacy(t *testing.T) {
 	lggr := logger.TestLogger(t)
 	m := &mockMonitoringEndpoint{}
@@ -233,7 +237,7 @@ func Test_Telemeter_observationTelemetry(t *testing.T) {
 		m := &mockMonitoringEndpoint{chLogs: make(chan []byte, 100)}
 		tm := newTelemeter(lggr, m, donID)
 		servicetest.Run(t, tm)
-		ch := tm.MakeTelemChannel(opts, 100)
+		ch := tm.MakeObservationTelemetryCh(opts, 100)
 
 		ch <- &pipeline.BridgeTelemetry{
 			Name:                   "test-bridge-1",
@@ -277,7 +281,7 @@ func Test_Telemeter_observationTelemetry(t *testing.T) {
 		m := &mockMonitoringEndpoint{chLogs: make(chan []byte, 100)}
 		tm := newTelemeter(lggr, m, donID)
 		servicetest.Run(t, tm)
-		ch := tm.MakeTelemChannel(opts, 100)
+		ch := tm.MakeObservationTelemetryCh(opts, 100)
 
 		ch <- &telem.LLOObservationTelemetry{
 			StreamId:              135,
@@ -315,7 +319,7 @@ func Test_Telemeter_observationTelemetry(t *testing.T) {
 		obsLggr, observedLogs := logger.TestLoggerObserved(t, zapcore.WarnLevel)
 		tm := newTelemeter(obsLggr, m, donID)
 		servicetest.Run(t, tm)
-		ch := tm.MakeTelemChannel(opts, 100)
+		ch := tm.MakeObservationTelemetryCh(opts, 100)
 
 		ch <- struct{}{}
 

deployment/go.mod

@@ -354,7 +354,7 @@ require (
 	github.com/shopspring/decimal v1.4.0 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/smartcontractkit/chainlink-automation v0.8.1 // indirect
-	github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250228145850-f846693a6fd5 // indirect
+	github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250311123514-b4881a96af5a // indirect
 	github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 // indirect
 	github.com/smartcontractkit/chainlink-framework/chains v0.0.0-20250227163723-3c71fefea680 // indirect
 	github.com/smartcontractkit/chainlink-protos/rmn/v1.6/go v0.0.0-20250131130834-15e0d4cde2a6 // indirect

deployment/go.sum

@@ -1170,8 +1170,8 @@ github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250307105933-7
 github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250307105933-7912a5e97ad0/go.mod h1:2Y6JRpvj9EkgKgymvenuqCnra07+NYVB6+0rQGETSGA=
 github.com/smartcontractkit/chainlink-common v0.4.2-0.20250228151513-8777dbcefd5c h1:pI4Y9C+EXpZmAcmvcvrCA4REWSyvCO3LQw4pGU1a/4M=
 github.com/smartcontractkit/chainlink-common v0.4.2-0.20250228151513-8777dbcefd5c/go.mod h1:YQuXIqQpmpAqstWV0LHaDTJ5nsSWuip5ivEM+Fisb+4=
-github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250228145850-f846693a6fd5 h1:cJpPJ5hEwc6vlMoxmATS60uWPUi62ydnTVBabu6WKEE=
-github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250228145850-f846693a6fd5/go.mod h1:2yUpKW1/jFxpozO/Zkh3fKDzI0jthXoEcU2xuDq+vlo=
+github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250311123514-b4881a96af5a h1:9TAVhJT/mWsJjQxiLoJvXdapzcViH5uxwxRhA0qtalU=
+github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250311123514-b4881a96af5a/go.mod h1:2yUpKW1/jFxpozO/Zkh3fKDzI0jthXoEcU2xuDq+vlo=
 github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 h1:8u9xUrC+yHrTDexOKDd+jrA6LCzFFHeX1G82oj2fsSI=
 github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135/go.mod h1:NkvE4iQgiT7dMCP6U3xPELHhWhN5Xr6rHC0axRebyMU=
 github.com/smartcontractkit/chainlink-framework/chains v0.0.0-20250227163723-3c71fefea680 h1:Es/V5imh4at3NdHDWlbDMjljd24TrJQCy/+8xAVajl0=

go.mod

@@ -78,7 +78,7 @@ require (
 	github.com/smartcontractkit/chainlink-ccip v0.0.0-20250307145254-e603dcbe3db1
 	github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250302020946-0f2d5f4a8326
 	github.com/smartcontractkit/chainlink-common v0.4.2-0.20250228151513-8777dbcefd5c
-	github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250228145850-f846693a6fd5
+	github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250311123514-b4881a96af5a
 	github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135
 	github.com/smartcontractkit/chainlink-framework/chains v0.0.0-20250227163723-3c71fefea680
 	github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20250211162441-3d6cea220efb
@@ -383,4 +383,5 @@ replace (
 	github.com/gogo/protobuf => github.com/regen-network/protobuf v1.3.3-alpha.regen.1
 
 	github.com/sourcegraph/sourcegraph/lib => github.com/sourcegraph/sourcegraph-public-snapshot/lib v0.0.0-20240822153003-c864f15af264
+
 )

go.sum

@@ -1023,8 +1023,8 @@ github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250302020946-0
 github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250302020946-0f2d5f4a8326/go.mod h1:ARwstg2HUGjtuZgG/IwxpYk4QdQtcqX69V95FUtlHdE=
 github.com/smartcontractkit/chainlink-common v0.4.2-0.20250228151513-8777dbcefd5c h1:pI4Y9C+EXpZmAcmvcvrCA4REWSyvCO3LQw4pGU1a/4M=
 github.com/smartcontractkit/chainlink-common v0.4.2-0.20250228151513-8777dbcefd5c/go.mod h1:YQuXIqQpmpAqstWV0LHaDTJ5nsSWuip5ivEM+Fisb+4=
-github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250228145850-f846693a6fd5 h1:cJpPJ5hEwc6vlMoxmATS60uWPUi62ydnTVBabu6WKEE=
-github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250228145850-f846693a6fd5/go.mod h1:2yUpKW1/jFxpozO/Zkh3fKDzI0jthXoEcU2xuDq+vlo=
+github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250311123514-b4881a96af5a h1:9TAVhJT/mWsJjQxiLoJvXdapzcViH5uxwxRhA0qtalU=
+github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250311123514-b4881a96af5a/go.mod h1:2yUpKW1/jFxpozO/Zkh3fKDzI0jthXoEcU2xuDq+vlo=
 github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 h1:8u9xUrC+yHrTDexOKDd+jrA6LCzFFHeX1G82oj2fsSI=
 github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135/go.mod h1:NkvE4iQgiT7dMCP6U3xPELHhWhN5Xr6rHC0axRebyMU=
 github.com/smartcontractkit/chainlink-framework/chains v0.0.0-20250227163723-3c71fefea680 h1:Es/V5imh4at3NdHDWlbDMjljd24TrJQCy/+8xAVajl0=

integration-tests/go.mod

@@ -439,7 +439,7 @@ require (
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/smartcontractkit/ccip-owner-contracts v0.1.0 // indirect
 	github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250307105933-7912a5e97ad0 // indirect
-	github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250228145850-f846693a6fd5 // indirect
+	github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250311123514-b4881a96af5a // indirect
 	github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 // indirect
 	github.com/smartcontractkit/chainlink-framework/chains v0.0.0-20250227163723-3c71fefea680 // indirect
 	github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20250211162441-3d6cea220efb // indirect

integration-tests/go.sum

@@ -1428,8 +1428,8 @@ github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250307105933-7
 github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20250307105933-7912a5e97ad0/go.mod h1:2Y6JRpvj9EkgKgymvenuqCnra07+NYVB6+0rQGETSGA=
 github.com/smartcontractkit/chainlink-common v0.4.2-0.20250228151513-8777dbcefd5c h1:pI4Y9C+EXpZmAcmvcvrCA4REWSyvCO3LQw4pGU1a/4M=
 github.com/smartcontractkit/chainlink-common v0.4.2-0.20250228151513-8777dbcefd5c/go.mod h1:YQuXIqQpmpAqstWV0LHaDTJ5nsSWuip5ivEM+Fisb+4=
-github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250228145850-f846693a6fd5 h1:cJpPJ5hEwc6vlMoxmATS60uWPUi62ydnTVBabu6WKEE=
-github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250228145850-f846693a6fd5/go.mod h1:2yUpKW1/jFxpozO/Zkh3fKDzI0jthXoEcU2xuDq+vlo=
+github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250311123514-b4881a96af5a h1:9TAVhJT/mWsJjQxiLoJvXdapzcViH5uxwxRhA0qtalU=
+github.com/smartcontractkit/chainlink-data-streams v0.1.1-0.20250311123514-b4881a96af5a/go.mod h1:2yUpKW1/jFxpozO/Zkh3fKDzI0jthXoEcU2xuDq+vlo=
 github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 h1:8u9xUrC+yHrTDexOKDd+jrA6LCzFFHeX1G82oj2fsSI=
 github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135/go.mod h1:NkvE4iQgiT7dMCP6U3xPELHhWhN5Xr6rHC0axRebyMU=
 github.com/smartcontractkit/chainlink-framework/chains v0.0.0-20250227163723-3c71fefea680 h1:Es/V5imh4at3NdHDWlbDMjljd24TrJQCy/+8xAVajl0=