From b5c6e34a74ffce798784c3ac15b791e295c254ae Mon Sep 17 00:00:00 2001 From: Bruno Moura Date: Fri, 7 Aug 2026 20:14:21 +0100 Subject: [PATCH 1/2] llo: DS-3520 package reorg and experimental dev/v31 --- pkg/cmd/chainlink-evm/main.go | 2 +- .../channel_definition_cache_factory.go | 2 +- .../channel_definition_cache_factory_test.go | 2 +- .../onchain_channel_definition_cache.go | 46 +++++------ .../onchain_channel_definition_cache_test.go | 76 +++++++++---------- pkg/llo/orm.go | 6 +- pkg/llo/orm_test.go | 20 ++--- pkg/relay/evm.go | 2 +- pkg/relay/llo_provider.go | 8 +- 9 files changed, 82 insertions(+), 82 deletions(-) diff --git a/pkg/cmd/chainlink-evm/main.go b/pkg/cmd/chainlink-evm/main.go index c8863e61c4..a271889f41 100644 --- a/pkg/cmd/chainlink-evm/main.go +++ b/pkg/cmd/chainlink-evm/main.go @@ -18,7 +18,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/types/core" "github.com/smartcontractkit/chainlink-common/pkg/utils/mailbox" "github.com/smartcontractkit/chainlink-data-streams/llo/retirement" - mercurytransmitter "github.com/smartcontractkit/chainlink-data-streams/llo/transmitter/de" + mercurytransmitter "github.com/smartcontractkit/chainlink-data-streams/llo/transmitter/dataengine" "github.com/smartcontractkit/chainlink-data-streams/mercury/wsrpc" "github.com/smartcontractkit/chainlink-data-streams/mercury/wsrpc/cache" "github.com/smartcontractkit/chainlink-evm/pkg/chains/legacyevm" diff --git a/pkg/llo/channeldefinitions/channel_definition_cache_factory.go b/pkg/llo/channeldefinitions/channel_definition_cache_factory.go index 0de0f5e468..f98465aae1 100644 --- a/pkg/llo/channeldefinitions/channel_definition_cache_factory.go +++ b/pkg/llo/channeldefinitions/channel_definition_cache_factory.go @@ -6,7 +6,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/logger" llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" - lloconfig "github.com/smartcontractkit/chainlink-data-streams/llo/config" + lloconfig "github.com/smartcontractkit/chainlink-data-streams/llo/pluginconfig" "github.com/smartcontractkit/chainlink-evm/pkg/logpoller" ) diff --git a/pkg/llo/channeldefinitions/channel_definition_cache_factory_test.go b/pkg/llo/channeldefinitions/channel_definition_cache_factory_test.go index 1fc13abd01..59783c3b4b 100644 --- a/pkg/llo/channeldefinitions/channel_definition_cache_factory_test.go +++ b/pkg/llo/channeldefinitions/channel_definition_cache_factory_test.go @@ -8,7 +8,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/logger" - lloconfig "github.com/smartcontractkit/chainlink-data-streams/llo/config" + lloconfig "github.com/smartcontractkit/chainlink-data-streams/llo/pluginconfig" ) func Test_ChannelDefinitionCacheFactory(t *testing.T) { diff --git a/pkg/llo/channeldefinitions/onchain_channel_definition_cache.go b/pkg/llo/channeldefinitions/onchain_channel_definition_cache.go index 6e7b35b490..887217d623 100644 --- a/pkg/llo/channeldefinitions/onchain_channel_definition_cache.go +++ b/pkg/llo/channeldefinitions/onchain_channel_definition_cache.go @@ -34,7 +34,7 @@ import ( llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" "github.com/smartcontractkit/chainlink-common/pkg/types/query" "github.com/smartcontractkit/chainlink-common/pkg/types/query/primitives" - "github.com/smartcontractkit/chainlink-data-streams/llo/types" + "github.com/smartcontractkit/chainlink-data-streams/llo/channelsource" "github.com/smartcontractkit/chainlink-evm/gethwrappers/llo-feeds/generated/channel_config_store" "github.com/smartcontractkit/chainlink-evm/pkg/logpoller" ) @@ -105,7 +105,7 @@ func init() { } type ChannelDefinitionCacheORM interface { - LoadChannelDefinitions(ctx context.Context, addr common.Address, donID uint32) (pd *types.PersistedDefinitions, err error) + LoadChannelDefinitions(ctx context.Context, addr common.Address, donID uint32) (pd *channelsource.PersistedDefinitions, err error) StoreChannelDefinitions(ctx context.Context, addr common.Address, donID, version uint32, dfns json.RawMessage, blockNum int64, format uint32) (err error) CleanupChannelDefinitions(ctx context.Context, addr common.Address, donID uint32) error } @@ -135,15 +135,15 @@ func WithLogPollInterval(d time.Duration) Option { // It tracks the latest block number processed, the version (for owner sources), and // source definitions keyed by source ID. type Definitions struct { - LastBlockNum int64 // The latest block number from which channel definitions were processed - Version uint32 // The version number from the owner source (only updated for SourceOwner) - Sources map[uint32]types.SourceDefinition // Channel definitions grouped by source ID + LastBlockNum int64 // The latest block number from which channel definitions were processed + Version uint32 // The version number from the owner source (only updated for SourceOwner) + Sources map[uint32]channelsource.SourceDefinition // Channel definitions grouped by source ID } // channelDefinitionCache maintains an in-memory cache of channel definitions fetched from on-chain // events and external URLs. It polls the blockchain for new channel definition events, fetches // definitions from URLs, verifies SHA hashes, merges definitions from multiple sources according -// to authority rules, and persists source definitions (map[uint32]types.SourceDefinition) to the database. +// to authority rules, and persists source definitions (map[uint32]channelsource.SourceDefinition) to the database. type channelDefinitionCache struct { services.StateMachine @@ -162,7 +162,7 @@ type channelDefinitionCache struct { lggr logger.SugaredLogger initialBlockNum int64 - fetchTriggerCh chan types.Trigger + fetchTriggerCh chan channelsource.Trigger definitionsMu sync.RWMutex definitions Definitions @@ -196,11 +196,11 @@ func NewChannelDefinitionCache(lggr logger.Logger, orm ChannelDefinitionCacheORM donID: donID, donIDTopic: common.BigToHash(big.NewInt(int64(donID))), lggr: logger.Sugared(lggr).Named("ChannelDefinitionCache").With("addr", addr, "fromBlock", fromBlock), - fetchTriggerCh: make(chan types.Trigger, 1), + fetchTriggerCh: make(chan channelsource.Trigger, 1), initialBlockNum: fromBlock, chStop: make(chan struct{}), definitions: Definitions{ - Sources: make(map[uint32]types.SourceDefinition), + Sources: make(map[uint32]channelsource.SourceDefinition), }, } @@ -251,15 +251,15 @@ func (c *channelDefinitionCache) Start(ctx context.Context) error { return err } - var pd *types.PersistedDefinitions + var pd *channelsource.PersistedDefinitions if pd, err = c.orm.LoadChannelDefinitions(ctx, c.addr, c.donID); err != nil { return err } - c.definitions.Sources = make(map[uint32]types.SourceDefinition) + c.definitions.Sources = make(map[uint32]channelsource.SourceDefinition) if pd != nil { if pd.Format == MultiChannelDefinitionsFormat { - var sources map[uint32]types.SourceDefinition + var sources map[uint32]channelsource.SourceDefinition if sources, err = decodePersistedSourceDefinitions(pd.Definitions); err != nil { return err } @@ -447,7 +447,7 @@ func (c *channelDefinitionCache) scanFromBlockNum() int64 { // the fetch channel for asynchronous processing by fetchLatestLoop. func (c *channelDefinitionCache) processLogs(logs []logpoller.Log) { for _, log := range logs { - var trigger types.Trigger + var trigger channelsource.Trigger switch log.EventSig { case NewChannelDefinition: unpacked, err := c.unpackOwnerLog(log) @@ -456,7 +456,7 @@ func (c *channelDefinitionCache) processLogs(logs []logpoller.Log) { c.lggr.Warnw("Failed to unpack owner log", "err", err, "blockNumber", log.BlockNumber) continue } - trigger = types.Trigger{ + trigger = channelsource.Trigger{ Source: SourceOwner, URL: unpacked.Url, SHA: unpacked.Sha, @@ -472,7 +472,7 @@ func (c *channelDefinitionCache) processLogs(logs []logpoller.Log) { c.lggr.Warnw("Failed to unpack adder log", "err", err, "blockNumber", log.BlockNumber) continue } - trigger = types.Trigger{ + trigger = channelsource.Trigger{ Source: unpacked.ChannelAdderId, URL: unpacked.Url, SHA: unpacked.Sha, @@ -645,7 +645,7 @@ func (c *channelDefinitionCache) mergeDefinitions(source uint32, currentDefiniti func (c *channelDefinitionCache) fetchLatestLoop() { defer c.wg.Done() - var trigger types.Trigger + var trigger channelsource.Trigger for { select { case trigger = <-c.fetchTriggerCh: @@ -663,7 +663,7 @@ func (c *channelDefinitionCache) fetchLatestLoop() { // fetchRetryTimeout is reached or the cache is stopped (context cache shutdown). // This isolates retry logic from the main fetch loop, allowing it to continue processing new triggers // while retries occur in the background. -func (c *channelDefinitionCache) fetchLoop(trigger types.Trigger) { +func (c *channelDefinitionCache) fetchLoop(trigger channelsource.Trigger) { defer c.wg.Done() var err error b := newHTTPFetchBackoff() @@ -702,7 +702,7 @@ func (c *channelDefinitionCache) fetchLoop(trigger types.Trigger) { // Returns an error if fetching, SHA verification, or JSON decoding fails. Note that adder limit // checks occur during merging in Definitions(), where violations are handled by logging warnings // and stopping processing for that source, not by returning errors. -func (c *channelDefinitionCache) fetchAndSetChannelDefinitions(ctx context.Context, trigger types.Trigger) error { +func (c *channelDefinitionCache) fetchAndSetChannelDefinitions(ctx context.Context, trigger channelsource.Trigger) error { defs, err := c.fetchChannelDefinitions(ctx, trigger) if err != nil { return fmt.Errorf("failed to fetch channel definitions: %w", err) @@ -722,7 +722,7 @@ func (c *channelDefinitionCache) fetchAndSetChannelDefinitions(ctx context.Conte } } - c.definitions.Sources[trigger.Source] = types.SourceDefinition{ + c.definitions.Sources[trigger.Source] = channelsource.SourceDefinition{ Trigger: trigger, Definitions: defs, } @@ -746,7 +746,7 @@ func (c *channelDefinitionCache) fetchAndSetChannelDefinitions(ctx context.Conte // the JSON response, and annotates each definition with its source identifier. Returns an // error if the URL is invalid, the HTTP request fails, the hash verification fails, or the // JSON cannot be decoded. -func (c *channelDefinitionCache) fetchChannelDefinitions(ctx context.Context, trigger types.Trigger) (llotypes.ChannelDefinitions, error) { +func (c *channelDefinitionCache) fetchChannelDefinitions(ctx context.Context, trigger channelsource.Trigger) (llotypes.ChannelDefinitions, error) { u, err := url.ParseRequestURI(trigger.URL) if err != nil { return nil, fmt.Errorf("failed to parse URL %s: %w", trigger.URL, err) @@ -921,7 +921,7 @@ func (c *channelDefinitionCache) Definitions(prev llotypes.ChannelDefinitions) l merged = make(llotypes.ChannelDefinitions) } - src := make([]types.SourceDefinition, 0, len(c.definitions.Sources)) + src := make([]channelsource.SourceDefinition, 0, len(c.definitions.Sources)) for _, sourceDefinition := range c.definitions.Sources { src = append(src, sourceDefinition) } @@ -946,8 +946,8 @@ func (c *channelDefinitionCache) Definitions(prev llotypes.ChannelDefinitions) l return merged } -func decodePersistedSourceDefinitions(definitionsJSON json.RawMessage) (map[uint32]types.SourceDefinition, error) { - var sources map[uint32]types.SourceDefinition +func decodePersistedSourceDefinitions(definitionsJSON json.RawMessage) (map[uint32]channelsource.SourceDefinition, error) { + var sources map[uint32]channelsource.SourceDefinition if err := json.Unmarshal(definitionsJSON, &sources); err != nil { return nil, fmt.Errorf("failed to unmarshal persisted definitions: %w", err) } diff --git a/pkg/llo/channeldefinitions/onchain_channel_definition_cache_test.go b/pkg/llo/channeldefinitions/onchain_channel_definition_cache_test.go index 66e9f8b3fe..69e3ac9864 100644 --- a/pkg/llo/channeldefinitions/onchain_channel_definition_cache_test.go +++ b/pkg/llo/channeldefinitions/onchain_channel_definition_cache_test.go @@ -19,7 +19,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/logger" llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" "github.com/smartcontractkit/chainlink-common/pkg/types/query" - "github.com/smartcontractkit/chainlink-data-streams/llo/types" + "github.com/smartcontractkit/chainlink-data-streams/llo/channelsource" "github.com/smartcontractkit/chainlink-evm/pkg/logpoller" "github.com/smartcontractkit/chainlink-evm/pkg/testutils" ) @@ -81,12 +81,12 @@ type mockCDCORM struct { lastPersistedAddr common.Address lastPersistedDonID uint32 lastPersistedVersion uint32 - lastPersistedDfns map[uint32]types.SourceDefinition + lastPersistedDfns map[uint32]channelsource.SourceDefinition lastPersistedBlockNum int64 lastPersistedFormat uint32 } -func (m *mockCDCORM) LoadChannelDefinitions(ctx context.Context, addr common.Address, donID uint32) (pd *types.PersistedDefinitions, err error) { +func (m *mockCDCORM) LoadChannelDefinitions(ctx context.Context, addr common.Address, donID uint32) (pd *channelsource.PersistedDefinitions, err error) { panic("not implemented") } func (m *mockCDCORM) StoreChannelDefinitions(ctx context.Context, addr common.Address, donID, version uint32, dfns json.RawMessage, blockNum int64, format uint32) (err error) { @@ -150,8 +150,8 @@ func drainChannel[T any](ch chan T) { } // collectTriggers collects all available triggers from a channel up to maxCount -func collectTriggers(ch chan types.Trigger, maxCount int) []types.Trigger { - triggers := make([]types.Trigger, 0, maxCount) +func collectTriggers(ch chan channelsource.Trigger, maxCount int) []channelsource.Trigger { + triggers := make([]channelsource.Trigger, 0, maxCount) for range maxCount { select { case trigger := <-ch: @@ -210,7 +210,7 @@ func Test_ChannelDefinitionCache(t *testing.T) { cdc := &channelDefinitionCache{ lggr: logger.TestSugared(t), definitions: Definitions{ - Sources: make(map[uint32]types.SourceDefinition), + Sources: make(map[uint32]channelsource.SourceDefinition), }, orm: &mockCDCORM{}, // Required for persist() call in Definitions() } @@ -227,8 +227,8 @@ func Test_ChannelDefinitionCache(t *testing.T) { Source: adderID, }, } - cdc.definitions.Sources[adderID] = types.SourceDefinition{ - Trigger: types.Trigger{ + cdc.definitions.Sources[adderID] = channelsource.SourceDefinition{ + Trigger: channelsource.Trigger{ Source: adderID, BlockNum: 1000, }, @@ -258,8 +258,8 @@ func Test_ChannelDefinitionCache(t *testing.T) { Tombstone: true, }, } - cdc.definitions.Sources[SourceOwner] = types.SourceDefinition{ - Trigger: types.Trigger{ + cdc.definitions.Sources[SourceOwner] = channelsource.SourceDefinition{ + Trigger: channelsource.Trigger{ Source: SourceOwner, BlockNum: 2000, }, @@ -277,14 +277,14 @@ func Test_ChannelDefinitionCache(t *testing.T) { t.Run("readLogs", func(t *testing.T) { lp := &mockLogPoller{latestBlockErr: sql.ErrNoRows} - fetchTriggerCh := make(chan types.Trigger, 100) + fetchTriggerCh := make(chan channelsource.Trigger, 100) cdc := &channelDefinitionCache{ donID: donID, lp: lp, lggr: logger.TestSugared(t), fetchTriggerCh: fetchTriggerCh, definitions: Definitions{ - Sources: make(map[uint32]types.SourceDefinition), + Sources: make(map[uint32]channelsource.SourceDefinition), }, } @@ -434,7 +434,7 @@ func Test_ChannelDefinitionCache(t *testing.T) { triggers := collectTriggers(fetchTriggerCh, 8) require.Len(t, triggers, 4, "expected 4 triggers") // Find the trigger with version 45 (latest) - var latestTrigger *types.Trigger + var latestTrigger *channelsource.Trigger for i := range triggers { if triggers[i].Version == 45 { latestTrigger = &triggers[i] @@ -612,7 +612,7 @@ func Test_ChannelDefinitionCache(t *testing.T) { // Use a URL with invalid scheme that will fail at HTTP client level // This avoids panic from URL parsing in the HTTP library - trigger := types.Trigger{ + trigger := channelsource.Trigger{ Source: SourceOwner, URL: "http://[::1", SHA: [32]byte{}, @@ -629,7 +629,7 @@ func Test_ChannelDefinitionCache(t *testing.T) { c.resp = nil c.err = errors.New("http request failed") - trigger := types.Trigger{ + trigger := channelsource.Trigger{ Source: SourceOwner, URL: "http://example.com/definitions.json", SHA: [32]byte{}, @@ -646,7 +646,7 @@ func Test_ChannelDefinitionCache(t *testing.T) { c.err = nil c.resp = &http.Response{StatusCode: 500, Body: io.NopCloser(bytes.NewReader([]byte{1, 2, 3}))} - trigger := types.Trigger{ + trigger := channelsource.Trigger{ Source: SourceOwner, URL: "http://example.com/definitions.json", SHA: [32]byte{}, @@ -668,7 +668,7 @@ func Test_ChannelDefinitionCache(t *testing.T) { c.err = nil c.resp = &http.Response{StatusCode: 404, Body: io.NopCloser(bytes.NewReader(largeBody))} - trigger := types.Trigger{ + trigger := channelsource.Trigger{ Source: SourceOwner, URL: "http://example.com/definitions.json", SHA: [32]byte{}, @@ -689,7 +689,7 @@ func Test_ChannelDefinitionCache(t *testing.T) { c.err = nil c.resp = &http.Response{StatusCode: 200, Body: io.NopCloser(bytes.NewReader(hugeBody))} - trigger := types.Trigger{ + trigger := channelsource.Trigger{ Source: SourceOwner, URL: "http://example.com/definitions.json", SHA: [32]byte{}, @@ -707,7 +707,7 @@ func Test_ChannelDefinitionCache(t *testing.T) { c.resp = &http.Response{StatusCode: 200, Body: io.NopCloser(bytes.NewReader([]byte{1, 2, 3}))} expectedSha := common.HexToHash("0xfd1780a6fc9ee0dab26ceb4b3941ab03e66ccd970d1db91612c66df4515b0a0a") - trigger := types.Trigger{ + trigger := channelsource.Trigger{ Source: SourceOwner, URL: "http://example.com/definitions.json", SHA: [32]byte(expectedSha), @@ -724,7 +724,7 @@ func Test_ChannelDefinitionCache(t *testing.T) { c.err = nil c.resp = &http.Response{StatusCode: 200, Body: io.NopCloser(bytes.NewReader([]byte(`{"foo":"bar"}`)))} - trigger := types.Trigger{ + trigger := channelsource.Trigger{ Source: SourceOwner, URL: "http://example.com/definitions.json", SHA: [32]byte{}, @@ -763,7 +763,7 @@ func Test_ChannelDefinitionCache(t *testing.T) { c.resp = &http.Response{StatusCode: 200, Body: io.NopCloser(bytes.NewReader([]byte(valid)))} expectedSha := common.HexToHash("0x367bbc75f7b6c9fc66a98ea99f837ea7ac4a3c2d6a9ee284de018bd02c41b52d") - trigger := types.Trigger{ + trigger := channelsource.Trigger{ Source: SourceOwner, URL: "http://example.com/definitions.json", SHA: [32]byte(expectedSha), @@ -807,9 +807,9 @@ func Test_ChannelDefinitionCache(t *testing.T) { cdc.definitions.Version = 42 cdc.persistedBlockNum = 141 cdc.definitions.LastBlockNum = 142 - cdc.definitions.Sources = map[uint32]types.SourceDefinition{ + cdc.definitions.Sources = map[uint32]channelsource.SourceDefinition{ SourceOwner: { - Trigger: types.Trigger{ + Trigger: channelsource.Trigger{ Source: SourceOwner, BlockNum: 142, Version: 42, @@ -835,9 +835,9 @@ func Test_ChannelDefinitionCache(t *testing.T) { cdc.persistedBlockNum = 141 cdc.definitions.Version = 43 cdc.definitions.LastBlockNum = 143 - cdc.definitions.Sources = map[uint32]types.SourceDefinition{ + cdc.definitions.Sources = map[uint32]channelsource.SourceDefinition{ SourceOwner: { - Trigger: types.Trigger{ + Trigger: channelsource.Trigger{ Source: SourceOwner, BlockNum: 143, Version: 43, @@ -859,9 +859,9 @@ func Test_ChannelDefinitionCache(t *testing.T) { ctx := t.Context() cdc.definitions.Version = 43 cdc.definitions.LastBlockNum = 143 - cdc.definitions.Sources = map[uint32]types.SourceDefinition{ + cdc.definitions.Sources = map[uint32]channelsource.SourceDefinition{ SourceOwner: { - Trigger: types.Trigger{ + Trigger: channelsource.Trigger{ Source: SourceOwner, BlockNum: 143, Version: 43, @@ -1384,9 +1384,9 @@ func Test_decodePersistedSourceDefinitions(t *testing.T) { source2 := uint32(2) source3 := uint32(3) - definitions := map[uint32]types.SourceDefinition{ + definitions := map[uint32]channelsource.SourceDefinition{ source2: { - Trigger: types.Trigger{ + Trigger: channelsource.Trigger{ Source: source2, URL: "http://example.com/source2.json", SHA: [32]byte{1, 2, 3}, @@ -1399,7 +1399,7 @@ func Test_decodePersistedSourceDefinitions(t *testing.T) { }, }, source3: { - Trigger: types.Trigger{ + Trigger: channelsource.Trigger{ Source: source3, URL: "http://example.com/source3.json", SHA: [32]byte{4, 5, 6}, @@ -1431,9 +1431,9 @@ func Test_decodePersistedSourceDefinitions(t *testing.T) { channelID1 := llotypes.ChannelID(100) channelID2 := llotypes.ChannelID(200) - definitions := map[uint32]types.SourceDefinition{ + definitions := map[uint32]channelsource.SourceDefinition{ legacyKey: { - Trigger: types.Trigger{ + Trigger: channelsource.Trigger{ Source: legacyKey, // This should be migrated to SourceOwner URL: "http://example.com/owner.json", SHA: [32]byte{7, 8, 9}, @@ -1517,9 +1517,9 @@ func Test_decodePersistedSourceDefinitions(t *testing.T) { source2 := uint32(2) source3 := uint32(3) - definitions := map[uint32]types.SourceDefinition{ + definitions := map[uint32]channelsource.SourceDefinition{ legacyKey: { - Trigger: types.Trigger{ + Trigger: channelsource.Trigger{ Source: legacyKey, URL: "http://example.com/owner.json", SHA: [32]byte{1, 1, 1}, @@ -1531,7 +1531,7 @@ func Test_decodePersistedSourceDefinitions(t *testing.T) { }, }, source2: { - Trigger: types.Trigger{ + Trigger: channelsource.Trigger{ Source: source2, URL: "http://example.com/source2.json", SHA: [32]byte{2, 2, 2}, @@ -1543,7 +1543,7 @@ func Test_decodePersistedSourceDefinitions(t *testing.T) { }, }, source3: { - Trigger: types.Trigger{ + Trigger: channelsource.Trigger{ Source: source3, URL: "http://example.com/source3.json", SHA: [32]byte{3, 3, 3}, @@ -1589,9 +1589,9 @@ func Test_decodePersistedSourceDefinitions(t *testing.T) { def2 := makeChannelDefinition(200, source2) // Different source def3 := makeChannelDefinition(300, source3) // Different source - definitions := map[uint32]types.SourceDefinition{ + definitions := map[uint32]channelsource.SourceDefinition{ legacyKey: { - Trigger: types.Trigger{ + Trigger: channelsource.Trigger{ Source: legacyKey, URL: "http://example.com/owner.json", SHA: [32]byte{9, 9, 9}, diff --git a/pkg/llo/orm.go b/pkg/llo/orm.go index 41927c796f..b3a3081cc5 100644 --- a/pkg/llo/orm.go +++ b/pkg/llo/orm.go @@ -10,7 +10,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/smartcontractkit/chainlink-common/pkg/sqlutil" - "github.com/smartcontractkit/chainlink-data-streams/llo/types" + "github.com/smartcontractkit/chainlink-data-streams/llo/channelsource" "github.com/smartcontractkit/chainlink-evm/pkg/llo/channeldefinitions" ) @@ -29,8 +29,8 @@ func NewChainScopedORM(ds sqlutil.DataSource, chainSelector uint64) ChainScopedO return &chainScopedORM{ds, chainSelector} } -func (o *chainScopedORM) LoadChannelDefinitions(ctx context.Context, addr common.Address, donID uint32) (pd *types.PersistedDefinitions, err error) { - pd = new(types.PersistedDefinitions) +func (o *chainScopedORM) LoadChannelDefinitions(ctx context.Context, addr common.Address, donID uint32) (pd *channelsource.PersistedDefinitions, err error) { + pd = new(channelsource.PersistedDefinitions) err = o.ds.GetContext(ctx, pd, "SELECT * FROM channel_definitions WHERE chain_selector = $1 AND addr = $2 AND don_id = $3", o.chainSelector, addr, donID) if errors.Is(err, sql.ErrNoRows) { return nil, nil diff --git a/pkg/llo/orm_test.go b/pkg/llo/orm_test.go index 15ad14bb07..f6db7350a1 100644 --- a/pkg/llo/orm_test.go +++ b/pkg/llo/orm_test.go @@ -10,7 +10,7 @@ import ( "github.com/stretchr/testify/require" llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" - "github.com/smartcontractkit/chainlink-data-streams/llo/types" + "github.com/smartcontractkit/chainlink-data-streams/llo/channelsource" "github.com/smartcontractkit/chainlink-evm/pkg/llo/channeldefinitions" "github.com/smartcontractkit/chainlink-evm/pkg/testutils" @@ -105,7 +105,7 @@ func Test_ORM(t *testing.T) { assert.Equal(t, uint32(1), pd.Format) // Unmarshal the definitions from json.RawMessage - var loadedDefs map[uint32]types.SourceDefinition + var loadedDefs map[uint32]channelsource.SourceDefinition err = json.Unmarshal(pd.Definitions, &loadedDefs) require.NoError(t, err) require.Len(t, loadedDefs, 1) @@ -128,7 +128,7 @@ func Test_ORM(t *testing.T) { require.NoError(t, err) // Unmarshal empty definitions - var emptyDefs map[uint32]types.SourceDefinition + var emptyDefs map[uint32]channelsource.SourceDefinition err = json.Unmarshal(pd.Definitions, &emptyDefs) require.NoError(t, err) assert.Empty(t, emptyDefs) @@ -138,7 +138,7 @@ func Test_ORM(t *testing.T) { pd, err = orm.LoadChannelDefinitions(ctx, addr1, donID2) require.NoError(t, err) - assert.Equal(t, (*types.PersistedDefinitions)(nil), pd) + assert.Equal(t, (*channelsource.PersistedDefinitions)(nil), pd) }) }) @@ -148,9 +148,9 @@ func Test_ORM(t *testing.T) { cid2 := rand.Uint32() cid3 := rand.Uint32() cid4 := rand.Uint32() - defs := map[uint32]types.SourceDefinition{ + defs := map[uint32]channelsource.SourceDefinition{ 1: { - Trigger: types.Trigger{ + Trigger: channelsource.Trigger{ Source: 1, BlockNum: 142, Version: 42, @@ -168,7 +168,7 @@ func Test_ORM(t *testing.T) { }, }, 2: { - Trigger: types.Trigger{ + Trigger: channelsource.Trigger{ Source: 2, BlockNum: 142, Version: 42, @@ -205,14 +205,14 @@ func Test_ORM(t *testing.T) { assert.Equal(t, channeldefinitions.MultiChannelDefinitionsFormat, pd.Format) // Unmarshal and compare - var loadedDefs map[uint32]types.SourceDefinition + var loadedDefs map[uint32]channelsource.SourceDefinition err = json.Unmarshal(pd.Definitions, &loadedDefs) require.NoError(t, err) assert.Equal(t, defs, loadedDefs) }) t.Run("does not update if version is older than the database persisted version", func(t *testing.T) { // try to update with an older version - emptyDefsJSON, err := json.Marshal(map[uint32]types.SourceDefinition{}) + emptyDefsJSON, err := json.Marshal(map[uint32]channelsource.SourceDefinition{}) require.NoError(t, err) err = orm.StoreChannelDefinitions(ctx, addr1, donID1, 41, emptyDefsJSON, expectedBlockNum-1, channeldefinitions.MultiChannelDefinitionsFormat) require.NoError(t, err) @@ -222,7 +222,7 @@ func Test_ORM(t *testing.T) { assert.Equal(t, uint32(42), pd.Version) // Unmarshal and verify original definitions are still there - var loadedDefs map[uint32]types.SourceDefinition + var loadedDefs map[uint32]channelsource.SourceDefinition err = json.Unmarshal(pd.Definitions, &loadedDefs) require.NoError(t, err) assert.Equal(t, defs, loadedDefs) diff --git a/pkg/relay/evm.go b/pkg/relay/evm.go index c1b543e960..6300745bf9 100644 --- a/pkg/relay/evm.go +++ b/pkg/relay/evm.go @@ -32,7 +32,7 @@ import ( commontypes "github.com/smartcontractkit/chainlink-common/pkg/types" coretypes "github.com/smartcontractkit/chainlink-common/pkg/types/core" "github.com/smartcontractkit/chainlink-data-streams/llo/retirement" - coreconfig "github.com/smartcontractkit/chainlink-data-streams/llo/transmitter/de" + coreconfig "github.com/smartcontractkit/chainlink-data-streams/llo/transmitter/dataengine" "github.com/smartcontractkit/chainlink-data-streams/mercury" mercuryconfig "github.com/smartcontractkit/chainlink-data-streams/mercury/config" "github.com/smartcontractkit/chainlink-data-streams/mercury/wsrpc" diff --git a/pkg/relay/llo_provider.go b/pkg/relay/llo_provider.go index 78895445ad..f7abbbef2a 100644 --- a/pkg/relay/llo_provider.go +++ b/pkg/relay/llo_provider.go @@ -21,11 +21,11 @@ import ( relaytypes "github.com/smartcontractkit/chainlink-common/pkg/types" coretypes "github.com/smartcontractkit/chainlink-common/pkg/types/core" llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/llo" - lloconfig "github.com/smartcontractkit/chainlink-data-streams/llo/config" + lloconfig "github.com/smartcontractkit/chainlink-data-streams/llo/pluginconfig" "github.com/smartcontractkit/chainlink-data-streams/llo/retirement" llotransmitter "github.com/smartcontractkit/chainlink-data-streams/llo/transmitter" - "github.com/smartcontractkit/chainlink-data-streams/llo/transmitter/bm" - mercurytransmitter "github.com/smartcontractkit/chainlink-data-streams/llo/transmitter/de" + mercurytransmitter "github.com/smartcontractkit/chainlink-data-streams/llo/transmitter/dataengine" + dummytransmitter "github.com/smartcontractkit/chainlink-data-streams/llo/transmitter/dummy" "github.com/smartcontractkit/chainlink-data-streams/mercury" "github.com/smartcontractkit/chainlink-data-streams/mercury/wsrpc" "github.com/smartcontractkit/chainlink-data-streams/rpc" @@ -118,7 +118,7 @@ func NewLLOProvider( var transmitter LLOTransmitter if lloCfg.BenchmarkMode { lggr.Info("Benchmark mode enabled, using dummy transmitter. NOTE: THIS WILL NOT TRANSMIT ANYTHING") - transmitter = bm.NewTransmitter(lggr, csaPub) + transmitter = dummytransmitter.NewTransmitter(lggr, csaPub) } else { clients := make(map[string]rpc.Client) From 6af3e8fc2d12fdd78968c37e5468ceceb771ec60 Mon Sep 17 00:00:00 2001 From: Bruno Moura Date: Sat, 8 Aug 2026 17:55:00 +0100 Subject: [PATCH 2/2] DS-3520 bump chainlink-data-streams --- go.mod | 3 +-- go.sum | 6 ++---- 2 files changed, 3 insertions(+), 6 deletions(-) diff --git a/go.mod b/go.mod index aca158153d..02ce35f443 100644 --- a/go.mod +++ b/go.mod @@ -32,7 +32,7 @@ require ( github.com/smartcontractkit/chainlink-automation v0.8.1 github.com/smartcontractkit/chainlink-common v0.11.2-0.20260729194708-6738a99f5706 github.com/smartcontractkit/chainlink-common/keystore v1.3.0 - github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260806155614-21385fa363af + github.com/smartcontractkit/chainlink-data-streams v1.1.0 github.com/smartcontractkit/chainlink-evm/gethwrappers v0.0.0-20260512150409-b4068bf735e6 github.com/smartcontractkit/chainlink-framework/capabilities v0.0.0-20260423135514-5b1a7565a99c github.com/smartcontractkit/chainlink-framework/chains v0.0.0-20260724153515-bb6a2de39bcb @@ -93,7 +93,6 @@ require ( github.com/esote/minmaxheap v1.0.0 // indirect github.com/ethereum/c-kzg-4844/v2 v2.1.7 // indirect github.com/ethereum/go-bigmodexpfix v0.0.0-20250911101455-f9e208c548ab // indirect - github.com/expr-lang/expr v1.17.8 // indirect github.com/fatih/color v1.19.0 // indirect github.com/ferranbt/fastssz v0.1.4 // indirect github.com/fsnotify/fsnotify v1.10.1 // indirect diff --git a/go.sum b/go.sum index 7d0a6457c2..b104ccf06e 100644 --- a/go.sum +++ b/go.sum @@ -172,8 +172,6 @@ github.com/ethereum/go-bigmodexpfix v0.0.0-20250911101455-f9e208c548ab h1:rvv6MJ github.com/ethereum/go-bigmodexpfix v0.0.0-20250911101455-f9e208c548ab/go.mod h1:IuLm4IsPipXKF7CW5Lzf68PIbZ5yl7FFd74l/E0o9A8= github.com/ethereum/go-ethereum v1.17.3 h1:Ev/sQHH+UdKZHWjuVzhu2pxhi/sXaPZl23Q+Q5LDd4Q= github.com/ethereum/go-ethereum v1.17.3/go.mod h1:f2EhRwqewIZkGoQekywI2Y2RZAMTSavLNkD9qItFy1A= -github.com/expr-lang/expr v1.17.8 h1:W1loDTT+0PQf5YteHSTpju2qfUfNoBt4yw9+wOEU9VM= -github.com/expr-lang/expr v1.17.8/go.mod h1:8/vRC7+7HBzESEqt5kKpYXxrxkr31SaO8r40VO/1IT4= github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4= github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk= github.com/fatih/color v1.19.0 h1:Zp3PiM21/9Ld6FzSKyL5c/BULoe/ONr9KlbYVOfG8+w= @@ -662,8 +660,8 @@ github.com/smartcontractkit/chainlink-common/keystore v1.3.0 h1:V05Rp9/dTc4Wyips github.com/smartcontractkit/chainlink-common/keystore v1.3.0/go.mod h1:vHV8BGm6TN7jBbMsWxq1Hqm3HbCtYFwzvKS0CCczxG8= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 h1:uWEwl7i2ryuRVoV4DmIKm6mqYevf1lH/8cQYhw/JXko= github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72/go.mod h1:UYcRMb4dZcoaIPgZJ3hckCySTqtJc9K4Q+tOKErwTq0= -github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260806155614-21385fa363af h1:JXO7FcXvnNrk1CVxpccQuGwYCfQhO0Mg3gHjnhZ3YuA= -github.com/smartcontractkit/chainlink-data-streams v1.0.1-0.20260806155614-21385fa363af/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= +github.com/smartcontractkit/chainlink-data-streams v1.1.0 h1:O5ngSwpwey7kQ3t4YgFiXzu3L3EvYiPS4c93gMWNRbk= +github.com/smartcontractkit/chainlink-data-streams v1.1.0/go.mod h1:dF5JiHWueHjYguUUUrFeb03MkcDqha/tssEkqTkgzp4= github.com/smartcontractkit/chainlink-evm/gethwrappers v0.0.0-20260512150409-b4068bf735e6 h1:JFo7C3FilwhfwGBLAyj2umbL+P4QxGmVi/b8yt9kqvI= github.com/smartcontractkit/chainlink-evm/gethwrappers v0.0.0-20260512150409-b4068bf735e6/go.mod h1:a260YnLyWq2NHLUN5cSVyMGk9nhO6RguCaTI2rsVqyA= github.com/smartcontractkit/chainlink-framework/capabilities v0.0.0-20260423135514-5b1a7565a99c h1:AYRSQarVw1EJXUrGvHSwmRTtNHHww/i3xwLat5CshUE=