diff --git a/pkg/durableemitter/durable_emitter_log_contract_test.go b/pkg/durableemitter/durable_emitter_log_contract_test.go new file mode 100644 index 0000000000..2b86d7df27 --- /dev/null +++ b/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") +} diff --git a/pkg/workflows/utils_test.go b/pkg/workflows/utils_test.go index 284b6e6638..366e976ebf 100644 --- a/pkg/workflows/utils_test.go +++ b/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)