CRE integrity tests: execution-ID + DurableEmitter logs

pkg/durableemitter/durable_emitter_log_contract_test.go

@@ -0,0 +1,114 @@
+package durableemitter
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap/zapcore"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/services/servicetest"
+)
+
+// Log-message contract pins.
+//
+// DO NOT change these strings to make a failing test pass without coordinating
+// with the CRE observability team.
+//
+// WHO DEPENDS ON THESE: the CRE E2E-integrity monitoring system (epic
+// CRE-5693). Its integrity detectors alert on these exact DurableEmitter warn
+// messages to distinguish "event dropped but will be retransmitted" from a
+// genuine delivery gap. If the wording in durable_emitter.go changes, those
+// log-based detectors silently stop matching and E2E integrity monitoring
+// goes blind. Treat a failure here as a breaking change to an external
+// contract: update the detectors in lockstep, or keep the message as-is.
+const (
+	// Emitted from DurableEmitter.deliveryCallback when a batch publish fails
+	// and the event is left in the DB for the retransmit loop.
+	logMsgDeliveryFailed = "DurableEmitter: failed to deliver event. Relying on retransmit."
+	// Emitted from DurableEmitter.Emit when BatchEmitter.QueueMessage rejects
+	// the event (buffer full / stopped) and delivery falls back to retransmit.
+	logMsgBufferFull = "DurableEmitter: batch emitter buffer full, relying on retransmit"
+)
+
+// TestDurableEmitter_LogContract_DeliveryFailure drives a real publish failure
+// through the delivery callback (via testBatchEmitter.setPublishErr) and
+// asserts the exact warn message the CRE integrity detectors match on.
+func TestDurableEmitter_LogContract_DeliveryFailure(t *testing.T) {
+	lggr, observed := logger.TestObserved(t, zapcore.WarnLevel)
+
+	store := NewMemDurableEventStore()
+	be := newTestBatchEmitter()
+	be.setPublishErr(errors.New("chip ingress unavailable"))
+
+	cfg := DefaultConfig()
+	cfg.DisablePruning = true
+
+	// retransmitEnabled=false keeps background loops from adding extra log
+	// noise; the delivery callback path under test is unaffected.
+	em, err := NewDurableEmitter(store, be, false, cfg, lggr, nil)
+	require.NoError(t, err)
+	servicetest.Run(t, em)
+
+	require.NoError(t, em.Emit(t.Context(), []byte("log-contract"), testEmitAttrs()...))
+
+	// The delivery callback fires asynchronously; FilterMessage is an exact
+	// (full-string) match, so this pins the message verbatim.
+	require.Eventually(t, func() bool {
+		return observed.FilterMessage(logMsgDeliveryFailed).Len() >= 1
+	}, 2*time.Second, 10*time.Millisecond,
+		"exact warn message %q not observed: the DurableEmitter delivery-failure log contract changed; this breaks the CRE E2E-integrity detectors (CRE-5693)",
+		logMsgDeliveryFailed)
+
+	entry := observed.FilterMessage(logMsgDeliveryFailed).All()[0]
+	require.Equal(t, zapcore.WarnLevel, entry.Level,
+		"delivery-failure log must stay at Warn level; the CRE integrity detectors filter on it")
+
+	// The event must remain persisted for the retransmit loop — that is the
+	// promise the message makes.
+	require.Equal(t, 1, store.Len(), "event must remain in the store for retransmit after a delivery failure")
+}
+
+// rejectingBatchEmitter is a BatchEmitter whose QueueMessage always rejects,
+// simulating a full internal buffer. This forces Emit down the
+// "buffer full, relying on retransmit" branch.
+type rejectingBatchEmitter struct{}
+
+func (rejectingBatchEmitter) QueueMessage(*chipingress.CloudEventPb, func(error)) error {
+	return errors.New("buffer full")
+}
+func (rejectingBatchEmitter) Start(context.Context) {}
+func (rejectingBatchEmitter) Stop()                 {}
+
+// TestDurableEmitter_LogContract_BufferFull forces QueueMessage to reject and
+// asserts the exact warn message the CRE integrity detectors match on.
+func TestDurableEmitter_LogContract_BufferFull(t *testing.T) {
+	lggr, observed := logger.TestObserved(t, zapcore.WarnLevel)
+
+	store := NewMemDurableEventStore()
+
+	cfg := DefaultConfig()
+	cfg.DisablePruning = true
+
+	em, err := NewDurableEmitter(store, rejectingBatchEmitter{}, false, cfg, lggr, nil)
+	require.NoError(t, err)
+	servicetest.Run(t, em)
+
+	// Emit succeeds (the insert is durable) even though enqueueing fails;
+	// the warn is logged synchronously before Emit returns.
+	require.NoError(t, em.Emit(t.Context(), []byte("log-contract"), testEmitAttrs()...))
+
+	require.Equal(t, 1, observed.FilterMessage(logMsgBufferFull).Len(),
+		"exact warn message %q not observed: the DurableEmitter buffer-full log contract changed; this breaks the CRE E2E-integrity detectors (CRE-5693)",
+		logMsgBufferFull)
+
+	entry := observed.FilterMessage(logMsgBufferFull).All()[0]
+	require.Equal(t, zapcore.WarnLevel, entry.Level,
+		"buffer-full log must stay at Warn level; the CRE integrity detectors filter on it")
+
+	require.Equal(t, 1, store.Len(), "event must remain in the store for retransmit when the buffer is full")
+}

pkg/workflows/utils_test.go

@@ -42,6 +42,77 @@ func Test_EncodeExecutionID(t *testing.T) {
 	assert.NotEqual(t, reversed, actual)
 }
 
+// Test_ExecutionID_GoldenValues pins the exact execution-ID derivation:
+//
+//	executionID = hex(sha256(workflowID || triggerEventID || strconv.Itoa(triggerIndex)))
+//
+// DO NOT update these expected values to make a failing test pass without
+// coordinating with the CRE observability team.
+//
+// WHO DEPENDS ON THIS: the CRE E2E-integrity monitoring system (epic CRE-5693)
+// — specifically the integrity detectors and the synthetic freshness probe —
+// predicts execution IDs off-platform from (workflowID, triggerEventID,
+// triggerIndex) using this exact derivation. Changing the hash function, the
+// field order, the trigger-index encoding, or the hex encoding will:
+//   - break E2E integrity monitoring (the probe's predicted IDs will never
+//     match emitted events), and
+//   - change every workflow execution ID platform-wide.
+//
+// If this test fails, the derivation changed; treat it as a breaking change to
+// an external contract, not a test to be updated in place.
+func Test_ExecutionID_GoldenValues(t *testing.T) {
+	t.Run("GenerateExecutionIDWithTriggerIndex", func(t *testing.T) {
+		cases := []struct {
+			name           string
+			workflowID     string
+			triggerEventID string
+			triggerIndex   int
+			expected       string
+		}{
+			{
+				name:           "simple inputs, index 0",
+				workflowID:     "workflowID",
+				triggerEventID: "triggerEventID",
+				triggerIndex:   0,
+				expected:       "a47b654c76a2d715a0c5c814ba51fb036a1d467f25e7f56a101066ba6fafeb77",
+			},
+			{
+				name:           "realistic 64-hex workflow ID, index 1",
+				workflowID:     "15c631d295ef5e32deb99a10ee6804bc4af1385568f9b3363f6552ac6dbb2cef",
+				triggerEventID: "trigger_event_id_abc123",
+				triggerIndex:   1,
+				expected:       "f47b82ecfba846d5454e2db711f41240dc9fbffae86d9dc8c1c9e3dc57466b5b",
+			},
+			{
+				name:           "empty IDs, multi-digit index",
+				workflowID:     "",
+				triggerEventID: "",
+				triggerIndex:   42,
+				expected:       "73475cb40a568e8da8a045ced110137e159f890ac4da883b6b17dc651b3a8049",
+			},
+		}
+		for _, tc := range cases {
+			t.Run(tc.name, func(t *testing.T) {
+				got, err := GenerateExecutionIDWithTriggerIndex(tc.workflowID, tc.triggerEventID, tc.triggerIndex)
+				require.NoError(t, err)
+				require.Len(t, got, 64, "execution ID must be a 64-char hex string")
+				require.Equal(t, tc.expected, got,
+					"execution-ID derivation changed: this breaks the CRE E2E-integrity synthetic freshness probe (CRE-5693) and changes every execution ID platform-wide")
+			})
+		}
+	})
+
+	// EncodeExecutionID is the deprecated no-trigger-index variant:
+	// hex(sha256(workflowID || eventID)). Still pinned because existing
+	// consumers (and the integrity tooling's backwards-compat path) rely on it.
+	t.Run("EncodeExecutionID (deprecated variant)", func(t *testing.T) {
+		got, err := EncodeExecutionID("workflowID", "eventID")
+		require.NoError(t, err)
+		require.Equal(t, "8ea89f1a81727293b00e7abea61a256f4fb33ef10a09ec79b2a526e0e2b5d250", got,
+			"EncodeExecutionID derivation changed: this breaks CRE E2E-integrity monitoring (CRE-5693)")
+	})
+}
+
 func Test_GetTriggerReferenceID_RoundTrip(t *testing.T) {
 	for _, idx := range []int{0, 1, 5, 100} {
 		refID := GetTriggerReferenceID(idx)