Skip to content

[DS-3520] llo packages reorg, experimental v31 #521

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
brunotm merged 2 commits into from
Aug 10, 2026
Merged

[DS-3520] llo packages reorg, experimental v31 #521

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b5c6e34
llo: DS-3520 package reorg and experimental dev/v31
brunotm Aug 7, 2026
6af3e8f
DS-3520 bump chainlink-data-streams
brunotm Aug 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
6 changes: 2 additions & 4 deletions go.sum
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion pkg/cmd/chainlink-evm/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
2 changes: 1 addition & 1 deletion pkg/llo/channeldefinitions/channel_definition_cache_factory.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)

Expand Down
2 changes: 1 addition & 1 deletion pkg/llo/channeldefinitions/channel_definition_cache_factory_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
46 changes: 23 additions & 23 deletions pkg/llo/channeldefinitions/onchain_channel_definition_cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -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

Expand All @@ -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
Expand Down Expand Up @@ -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),
},
}

Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -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)
Expand All @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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:
Expand All @@ -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()
Expand Down Expand Up @@ -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)
Expand All @@ -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,
}
Expand All @@ -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)
Expand Down Expand Up @@ -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)
}
Expand All @@ -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)
}
Expand Down
Loading
Loading