ChIP DurableEmitter Core Integration

core/config/docs/core.toml

@@ -870,6 +870,8 @@ AuthHeadersTTL = '0s' # Default
 ChipIngressEndpoint = '' # Default
 # ChipIngressInsecureConnection disables TLS when connecting to CHIP Ingress.
 ChipIngressInsecureConnection = false # Default
+# DurableEmitterEnabled enables persisting outbound CHIP events to Postgres for at-least-once delivery.
+DurableEmitterEnabled = false # Default
 
 # HeartbeatInterval is the interval at which a the application heartbeat is sent to telemetry backends.
 HeartbeatInterval = '1s' # Default

core/config/telemetry_config.go

@@ -18,6 +18,7 @@ type Telemetry interface {
 	EmitterExportTimeout() time.Duration
 	ChipIngressEndpoint() string
 	ChipIngressInsecureConnection() bool
+	DurableEmitterEnabled() bool
 	HeartbeatInterval() time.Duration
 	LogStreamingEnabled() bool
 	LogLevel() zapcore.Level

core/config/toml/types.go

@@ -2858,6 +2858,7 @@ type Telemetry struct {
 	AuthHeadersTTL                *commonconfig.Duration
 	ChipIngressEndpoint           *string
 	ChipIngressInsecureConnection *bool
+	DurableEmitterEnabled         *bool
 	HeartbeatInterval             *commonconfig.Duration
 	LogLevel                      *string
 	LogStreamingEnabled           *bool
@@ -2902,6 +2903,9 @@ func (b *Telemetry) setFrom(f *Telemetry) {
 	if v := f.ChipIngressInsecureConnection; v != nil {
 		b.ChipIngressInsecureConnection = v
 	}
+	if v := f.DurableEmitterEnabled; v != nil {
+		b.DurableEmitterEnabled = v
+	}
 	if v := f.HeartbeatInterval; v != nil {
 		b.HeartbeatInterval = v
 	}

core/scripts/cre/environment/environment/beholder.go

@@ -38,35 +38,25 @@ type moduleInfo struct {
 	Version string `json:"Version"`
 }
 
-// getSchemaSetFromGoMod uses `go list` to extract the version/commit ref
-// from the github.com/smartcontractkit/chainlink-protos/workflows/go dependency.
-// It returns a SchemaSet with hardcoded values matching default.toml config.
-func getSchemaSetFromGoMod(ctx context.Context) ([]chipingressset.SchemaSet, error) {
-	const targetModule = "github.com/smartcontractkit/chainlink-protos/workflows/go"
+const chainlinkProtosGitURI = "https://github.com/smartcontractkit/chainlink-protos"
 
-	// Get the absolute path to the repository root (where go.mod is located)
-	repoRoot, err := filepath.Abs(relativePathToRepoRoot)
-	if err != nil {
-		return nil, errors.Wrap(err, "failed to get absolute path to repository root")
-	}
-
-	// Use `go list -m -json` to get module information
+// schemaCommitRefFromGoMod runs `go list -m -json` for targetModule from repoRoot and returns the ref for FetchAndRegisterProtos.
+func schemaCommitRefFromGoMod(ctx context.Context, repoRoot, targetModule string) (ref string, rawVersion string, err error) {
 	cmd := exec.CommandContext(ctx, "go", "list", "-m", "-json", targetModule)
 	cmd.Dir = repoRoot
 
-	output, err := cmd.Output()
-	if err != nil {
-		return nil, errors.Wrapf(err, "failed to run 'go list -m -json %s'", targetModule)
+	output, cmdErr := cmd.Output()
+	if cmdErr != nil {
+		return "", "", errors.Wrapf(cmdErr, "failed to run 'go list -m -json %s'", targetModule)
 	}
 
-	// Parse JSON output
 	var modInfo moduleInfo
-	if err := json.Unmarshal(output, &modInfo); err != nil {
-		return nil, errors.Wrap(err, "failed to parse go list JSON output")
+	if unmarshalErr := json.Unmarshal(output, &modInfo); unmarshalErr != nil {
+		return "", "", errors.Wrap(unmarshalErr, "failed to parse go list JSON output")
 	}
 
 	if modInfo.Version == "" {
-		return nil, errors.Errorf("no version found for module %s", targetModule)
+		return "", "", errors.Errorf("no version found for module %s", targetModule)
 	}
 
 	// Extract commit ref from version string
@@ -76,18 +66,49 @@ func getSchemaSetFromGoMod(ctx context.Context) ([]chipingressset.SchemaSet, err
 	// 3. 2a35b54f48ae06be4cc81c768dc9cc9e92249571 -> full commit hash, use as-is
 	// 4. v0.0.0-YYYYMMDDHHMMSS-SHORTHASH -> extract short hash
 	commitRef := extractCommitRef(modInfo.Version)
+	return commitRef, modInfo.Version, nil
+}
 
-	framework.L.Info().Msgf("Extracted commit ref for %s: %s (from version: %s)", targetModule, commitRef, modInfo.Version)
+// getSchemaSetFromGoMod resolves SchemaSets from chainlink-protos commits pinned in go.mod:
+//   - workflows (chip-cre.json) for CRE/workflow telemetry
+//   - node-platform (chip-schemas.json) for PluginRelayerConfigEmitter / common.v1.ChainPluginConfig
+func getSchemaSetFromGoMod(ctx context.Context) ([]chipingressset.SchemaSet, error) {
+	const (
+		workflowsModule    = "github.com/smartcontractkit/chainlink-protos/workflows/go"
+		nodePlatformModule = "github.com/smartcontractkit/chainlink-protos/node-platform"
+	)
+
+	repoRoot, err := filepath.Abs(relativePathToRepoRoot)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get absolute path to repository root")
+	}
 
-	// Return SchemaSet with hardcoded values from default.toml
-	schemaSet := chipingressset.SchemaSet{
-		URI:        "https://github.com/smartcontractkit/chainlink-protos",
-		Ref:        commitRef,
-		SchemaDir:  "workflows",
-		ConfigFile: "chip-cre.json", // file with mappings of protobufs to subjects, together with references
+	wfRef, wfVer, err := schemaCommitRefFromGoMod(ctx, repoRoot, workflowsModule)
+	if err != nil {
+		return nil, err
+	}
+	framework.L.Info().Msgf("Extracted commit ref for %s: %s (from version: %s)", workflowsModule, wfRef, wfVer)
+
+	npRef, npVer, err := schemaCommitRefFromGoMod(ctx, repoRoot, nodePlatformModule)
+	if err != nil {
+		return nil, err
 	}
+	framework.L.Info().Msgf("Extracted commit ref for %s: %s (from version: %s)", nodePlatformModule, npRef, npVer)
 
-	return []chipingressset.SchemaSet{schemaSet}, nil
+	return []chipingressset.SchemaSet{
+		{
+			URI:        chainlinkProtosGitURI,
+			Ref:        wfRef,
+			SchemaDir:  "workflows",
+			ConfigFile: "chip-cre.json",
+		},
+		{
+			URI:        chainlinkProtosGitURI,
+			Ref:        npRef,
+			SchemaDir:  "node-platform",
+			ConfigFile: "chip-schemas.json",
+		},
+	}, nil
 }
 
 // extractCommitRef extracts a commit reference from various version formats

core/scripts/cre/environment/observability-overrides/otel.yaml

@@ -0,0 +1,43 @@
+receivers:
+  otlp:
+    protocols:
+      grpc:
+        endpoint: 0.0.0.0:4317
+      http:
+        endpoint: 0.0.0.0:4318
+
+exporters:
+  debug:
+    verbosity: detailed
+    sampling_initial: 5
+    sampling_thereafter: 200
+  otlphttp/logs:
+    endpoint: "http://loki:3100/otlp"
+    tls:
+      insecure: true
+  otlp:
+    endpoint: "http://tempo:4317"
+    tls:
+      insecure: true
+  prometheus:
+    endpoint: "0.0.0.0:8889"
+
+processors:
+  transform:
+    log_statements:
+      - context: log
+        statements:
+          - set(body, Base64Decode(log.body.string))
+
+service:
+  pipelines:
+    traces:
+      receivers: [otlp]
+      exporters: [debug, otlp]
+    logs:
+      receivers: [otlp]
+      processors: [transform]
+      exporters: [debug, otlphttp/logs]
+    metrics:
+      receivers: [otlp]
+      exporters: [debug, prometheus]

core/scripts/go.mod

@@ -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.20251211140724-319861e514c4 // indirect
+	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260511200925-b94130438417 // 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