Durable Emitter: LOOP Plugin Support

core/cmd/shell.go

@@ -129,6 +129,7 @@ func initGlobals(cfgProm config.Prometheus, cfgTracing config.Tracing, cfgTeleme
 				ChipIngressEmitterEnabled:      cfgTelemetry.ChipIngressEndpoint() != "",
 				ChipIngressEmitterGRPCEndpoint: cfgTelemetry.ChipIngressEndpoint(),
 				ChipIngressInsecureConnection:  cfgTelemetry.ChipIngressInsecureConnection(),
+				ChipIngressLogger:              lggr,
 				LogStreamingEnabled:            cfgTelemetry.LogStreamingEnabled(),
 				LogLevel:                       cfgTelemetry.LogLevel(),
 				LogBatchProcessor:              cfgTelemetry.LogBatchProcessor(),

core/scripts/go.mod

@@ -43,7 +43,7 @@ require (
 	github.com/smartcontractkit/chain-selectors v1.0.100
 	github.com/smartcontractkit/chainlink-automation v0.8.1
 	github.com/smartcontractkit/chainlink-ccip/chains/evm v0.0.0-20260506144252-c100eabfda74
-	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260518100439-9564f35fd264
+	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260520174852-16779f420738
 	github.com/smartcontractkit/chainlink-common/keystore v1.1.0
 	github.com/smartcontractkit/chainlink-data-streams v0.1.14
 	github.com/smartcontractkit/chainlink-deployments-framework v0.105.0
@@ -480,7 +480,7 @@ require (
 	github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260506144252-c100eabfda74 // indirect
 	github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260511195239-0f6e1b177fc7 // indirect
 	github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260428133800-3b1484e8b1fd // indirect
-	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260511200925-b94130438417 // indirect
+	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260518142424-bacfb6ba4146 // indirect
 	github.com/smartcontractkit/chainlink-evm/contracts/cre/gobindings v0.0.0-20260403151002-2c91155b5501 // indirect
 	github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 // indirect
 	github.com/smartcontractkit/chainlink-framework/capabilities v0.0.0-20260423135514-5b1a7565a99c // indirect

core/services/beholder/durable_event_store_orm_test.go

@@ -9,9 +9,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	beholdersvc "github.com/smartcontractkit/chainlink/v2/core/services/beholder"
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder/beholderstore"
 
-	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils"
 	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils/pgtest"
 )
 
@@ -24,11 +23,11 @@ func truncateChipDurableEvents(t *testing.T, db *sqlx.DB) {
 	require.NoError(t, err)
 }
 
-func TestPgDurableEventStore_InsertDeleteRoundTrip(t *testing.T) {
+func TestPGDurableEventStore_InsertDeleteRoundTrip(t *testing.T) {
 	db := pgtest.NewSqlxDB(t)
 	truncateChipDurableEvents(t, db)
 	ctx := t.Context()
-	store := beholdersvc.NewPgDurableEventStore(db)
+	store := beholderstore.New(db)
 
 	id, err := store.Insert(ctx, []byte("test-payload"))
 	require.NoError(t, err)
@@ -47,11 +46,11 @@ func TestPgDurableEventStore_InsertDeleteRoundTrip(t *testing.T) {
 	assert.Empty(t, events)
 }
 
-func TestPgDurableEventStore_ListPending_RespectsCreatedBefore(t *testing.T) {
+func TestPGDurableEventStore_ListPending_RespectsCreatedBefore(t *testing.T) {
 	db := pgtest.NewSqlxDB(t)
 	truncateChipDurableEvents(t, db)
 	ctx := t.Context()
-	store := beholdersvc.NewPgDurableEventStore(db)
+	store := beholderstore.New(db)
 
 	_, err := store.Insert(ctx, []byte("event-1"))
 	require.NoError(t, err)
@@ -67,11 +66,11 @@ func TestPgDurableEventStore_ListPending_RespectsCreatedBefore(t *testing.T) {
 	assert.Len(t, events, 1)
 }
 
-func TestPgDurableEventStore_ListPending_RespectsLimit(t *testing.T) {
+func TestPGDurableEventStore_ListPending_RespectsLimit(t *testing.T) {
 	db := pgtest.NewSqlxDB(t)
 	truncateChipDurableEvents(t, db)
 	ctx := t.Context()
-	store := beholdersvc.NewPgDurableEventStore(db)
+	store := beholderstore.New(db)
 
 	for i := 0; i < 20; i++ {
 		_, err := store.Insert(ctx, []byte(fmt.Sprintf("event-%d", i)))
@@ -83,11 +82,11 @@ func TestPgDurableEventStore_ListPending_RespectsLimit(t *testing.T) {
 	assert.Len(t, events, 5)
 }
 
-func TestPgDurableEventStore_DeleteExpired(t *testing.T) {
+func TestPGDurableEventStore_DeleteExpired(t *testing.T) {
 	db := pgtest.NewSqlxDB(t)
 	truncateChipDurableEvents(t, db)
 	ctx := t.Context()
-	store := beholdersvc.NewPgDurableEventStore(db)
+	store := beholderstore.New(db)
 
 	_, err := store.Insert(ctx, []byte("will-expire"))
 	require.NoError(t, err)
@@ -103,11 +102,11 @@ func TestPgDurableEventStore_DeleteExpired(t *testing.T) {
 	assert.Equal(t, int64(1), deleted)
 }
 
-func TestPgDurableEventStore_ObserveDurableQueue(t *testing.T) {
+func TestPGDurableEventStore_ObserveDurableQueue(t *testing.T) {
 	db := pgtest.NewSqlxDB(t)
 	truncateChipDurableEvents(t, db)
-	ctx := testutils.Context(t)
-	store := beholdersvc.NewPgDurableEventStore(db)
+	ctx := t.Context()
+	store := beholderstore.New(db)
 
 	st, err := store.ObserveDurableQueue(ctx, time.Hour, time.Minute)
 	require.NoError(t, err)
@@ -122,11 +121,11 @@ func TestPgDurableEventStore_ObserveDurableQueue(t *testing.T) {
 	assert.Positive(t, st.OldestPendingAge)
 }
 
-func TestPgDurableEventStore_MarkDeliveredAndPurgeDelivered(t *testing.T) {
+func TestPGDurableEventStore_MarkDeliveredAndPurgeDelivered(t *testing.T) {
 	db := pgtest.NewSqlxDB(t)
 	truncateChipDurableEvents(t, db)
-	ctx := testutils.Context(t)
-	store := beholdersvc.NewPgDurableEventStore(db)
+	ctx := t.Context()
+	store := beholderstore.New(db)
 
 	id, err := store.Insert(ctx, []byte("payload"))
 	require.NoError(t, err)

core/services/chainlink/application.go

@@ -27,6 +27,7 @@ import (
 	"go.uber.org/zap/zapcore"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder/beholderstore"
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
 	"github.com/smartcontractkit/chainlink-common/pkg/loop"
 	nodeauthjwt "github.com/smartcontractkit/chainlink-common/pkg/nodeauth/jwt"
@@ -379,9 +380,13 @@ func NewApplication(ctx context.Context, opts ApplicationOpts) (Application, err
 
 	// Wire DurableEmitter for persistent chip ingress delivery when enabled.
 	if cfg.Telemetry().DurableEmitterEnabled() && cfg.Telemetry().ChipIngressEndpoint() != "" {
-		if err = setupDurableEmitter(ctx, opts.DS, globalLogger, cfg.Telemetry()); err != nil {
+		beholderClient := beholder.GetClient()
+		if err = beholderClient.SetupDurableEmitter(beholderstore.New(opts.DS), true); err != nil {
 			return nil, fmt.Errorf("failed to set up chip durable emitter: %w", err)
 		}
+		if err = beholderClient.StartDurableEmitter(ctx); err != nil {
+			return nil, fmt.Errorf("failed to start chip durable emitter: %w", err)
+		}
 	}
 
 	creServices, err := cre.NewServices(

deployment/go.mod

@@ -42,7 +42,7 @@ require (
 	github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260415165642-49f23e4d76cc
 	github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260511195239-0f6e1b177fc7
 	github.com/smartcontractkit/chainlink-ccip/deployment v0.0.0-20260504204047-af9826978b72
-	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260518100439-9564f35fd264
+	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260520174852-16779f420738
 	github.com/smartcontractkit/chainlink-common/keystore v1.1.0
 	github.com/smartcontractkit/chainlink-deployments-framework v0.105.0
 	github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260519153648-e22912dab374
@@ -417,7 +417,7 @@ require (
 	github.com/smartcontractkit/chainlink-automation v0.8.1 // indirect
 	github.com/smartcontractkit/chainlink-ccip/ccv/chains/evm v0.0.0-20260408145530-22e2d05695cd // indirect
 	github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260428133800-3b1484e8b1fd // indirect
-	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.10 // indirect
+	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260518142424-bacfb6ba4146 // indirect
 	github.com/smartcontractkit/chainlink-data-streams v0.1.14 // indirect
 	github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 // indirect
 	github.com/smartcontractkit/chainlink-framework/capabilities v0.0.0-20260423135514-5b1a7565a99c // indirect

go.mod

@@ -85,9 +85,9 @@ require (
 	github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260415165642-49f23e4d76cc
 	github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260415165642-49f23e4d76cc
 	github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260428133800-3b1484e8b1fd
-	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260518100439-9564f35fd264
+	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260520174852-16779f420738
 	github.com/smartcontractkit/chainlink-common/keystore v1.1.0
-	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.10
+	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260518142424-bacfb6ba4146
 	github.com/smartcontractkit/chainlink-data-streams v0.1.14
 	github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260519153648-e22912dab374
 	github.com/smartcontractkit/chainlink-evm/contracts/cre/gobindings v0.0.0-20260403151002-2c91155b5501

integration-tests/go.mod

@@ -29,7 +29,7 @@ require (
 	github.com/smartcontractkit/chainlink-ccip/chains/evm v0.0.0-20260506144252-c100eabfda74
 	github.com/smartcontractkit/chainlink-ccip/chains/solana v0.0.0-20260506144252-c100eabfda74
 	github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260511195239-0f6e1b177fc7
-	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260518100439-9564f35fd264
+	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260520174852-16779f420738
 	github.com/smartcontractkit/chainlink-common/keystore v1.1.0
 	github.com/smartcontractkit/chainlink-deployments-framework v0.105.0
 	github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260519153648-e22912dab374
@@ -394,7 +394,7 @@ require (
 	github.com/smartcontractkit/chainlink-ccip/ccv/chains/evm v0.0.0-20260408145530-22e2d05695cd // indirect
 	github.com/smartcontractkit/chainlink-ccip/deployment v0.0.0-20260504204047-af9826978b72 // indirect
 	github.com/smartcontractkit/chainlink-ccv v0.0.2-0.20260428133800-3b1484e8b1fd // indirect
-	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.10 // indirect
+	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260518142424-bacfb6ba4146 // indirect
 	github.com/smartcontractkit/chainlink-data-streams v0.1.14 // indirect
 	github.com/smartcontractkit/chainlink-evm/contracts/cre/gobindings v0.0.0-20260403151002-2c91155b5501 // indirect
 	github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 // indirect