EventBridge Pipes to Lambda: Enrich, Filter, and Fail Closed

EventBridge Pipes look like free glue: point SQS or Kinesis at Lambda, drop a filter, maybe an enrichment step, and delete the custom “router” Lambda that only existed to fan messages. The unfair advantage is not “use Pipes” — it is treating filters and enrichment as a fail-closed path. A wrong pattern, a 403 on enrichment, or an enrichment timeout that still marks the source record consumed is how you lose orders without a single red CloudWatch alarm on the target.

⚡ TL;DR: Design Pipes as filter → enrich → target with explicit drop vs DLQ semantics. Prefer source-native filters that fail closed (unknown fields do not match). Enrichment must be idempotent, IAM-scoped, and timeout-bounded; never mutate identity keys. On Lambda targets, enable partial-batch failure reporting and wire source DLQs / on-failure destinations. Prove every discard path with metrics (Matched, Invocations, enrichment errors) before cutting over from a custom router. Pair with EventBridge Schema Evolution so filters stay additive.

Model the pipe as a contract, not a convenience

A Pipe has four stages that each can drop or fail: source poll, filter, enrichment, target. Teams that only test the happy path discover in prod that enrichment failures can still advance the source cursor depending on configuration. Write the contract down before CDK.

// contracts/order-pipe.ts — explicit semantics for ops and reviewers
export type PipeContract = {
  source: "sqs" | "kinesis" | "dynamodb-stream" | "kafka";
  filter: {
    // ✅ Match only known event types; unknown → drop (fail closed for enrichment)
    matchEventTypes: string[];
    onUnknownField: "drop" | "dlq"; // never "forward_anyway"
  };
  enrichment?: {
    type: "lambda" | "api-destination" | "step-functions";
    timeoutMs: number;
    onError: "retry_then_dlq" | "dlq" | "fail_pipe"; // ❌ never silent continue
    idempotencyKey: "sourceMessageId" | "eventId";
  };
  target: {
    type: "lambda";
    partialBatchFailure: true; // required for SQS/Kinesis batches
    maxConcurrency?: number;
  };
  dlq: { arn: string; maxReceiveCount: number };
};

// ❌ “We’ll add DLQ later” is how silent loss ships
export const ORDER_PIPE: PipeContract = {
  source: "sqs",
  filter: {
    matchEventTypes: ["order.created", "order.updated"],
    onUnknownField: "drop",
  },
  enrichment: {
    type: "lambda",
    timeoutMs: 3000,
    onError: "retry_then_dlq",
    idempotencyKey: "sourceMessageId",
  },
  target: { type: "lambda", partialBatchFailure: true, maxConcurrency: 50 },
  dlq: { arn: "arn:aws:sqs:REGION:ACCOUNT:orders-pipe-dlq", maxReceiveCount: 3 },
};

If product later adds order.cancelled, the filter must be updated deliberately — see additive consumer patterns in EventBridge Schema Evolution.

Filters: fail closed on shape, not on vibes

EventBridge filter patterns are powerful and easy to overfit. Prefer matching on stable envelope fields (detail-type, source, typed ids). Matching deep optional JSON paths is how a producer field rename silently zeros throughput.

// infra/order-pipe-filter.ts — CDK-ish filter pattern (illustrative)
export const orderCreatedFilter = {
  // ✅ Envelope + required business keys
  "detail-type": ["order.created", "order.updated"],
  source: ["com.cheatcoders.orders"],
  detail: {
    orderId: [{ "exists": true }],
    status: ["PENDING", "CONFIRMED"],
  },
};

// ❌ Fragile: nested optional promo that producers sometimes omit
export const fragilePromoFilter = {
  detail: {
    promo: {
      code: [{ "prefix": "SUMMER" }],
    },
  },
};

// Unit-test patterns against fixtures — Pipe filter changes are deploys
export function assertFilterFixtures(
  pattern: Record<string, unknown>,
  fixtures: { name: string; event: unknown; expectMatch: boolean }[],
) {
  for (const f of fixtures) {
    const matched = eventBridgePatternMatches(pattern, f.event); // your tester
    if (matched !== f.expectMatch) {
      throw new Error(`filter_fixture_failed:${f.name}`);
    }
  }
}

Operational rule: every filter change ships with at least one should-match and one should-drop fixture in CI. Without that, you are editing production drop logic by eye.

Enrichment: IAM, timeouts, and never rewrite identity

Enrichment Lambda/API calls run in the hot path before the target. Give enrichment a dedicated role that can only GetItem / GetSecret for what it needs — not the target’s write permissions. Cap timeout well below the source visibility timeout so retries remain possible.

// enrichment/enrich-order.ts
import { DynamoDBClient, GetItemCommand } from "@aws-sdk/client-dynamodb";
import { createHash } from "node:crypto";

const ddb = new DynamoDBClient({});
const TABLE = process.env.CUSTOMER_TABLE!;

export type PipeRecord = {
  orderId: string;
  customerId: string;
  status: string;
};

export type Enriched = PipeRecord & {
  tier: "free" | "pro" | "enterprise";
  enrichmentVersion: 1;
};

export async function handler(event: { orderId: string; customerId: string; status: string }): Promise<Enriched> {
  // ✅ Preserve source identity keys; only ADD fields
  // ❌ Never: regenerate orderId, coerce types, or drop status
  const out = await ddb.send(
    new GetItemCommand({
      TableName: TABLE,
      Key: { pk: { S: `CUST#${event.customerId}` } },
      ProjectionExpression: "tier",
    }),
  );

  if (!out.Item?.tier?.S) {
    // Fail closed: throw so Pipe retry/DLQ path engages
    // ❌ return { ...event, tier: "free" } on miss — invents business state
    throw new Error(`customer_tier_missing:${event.customerId}`);
  }

  return {
    ...event,
    tier: out.Item.tier.S as Enriched["tier"],
    enrichmentVersion: 1,
  };
}

// Idempotency for enrichment side effects (if any writes):
export function enrichKey(msgId: string) {
  return createHash("sha256").update(`pipe-enrich:v1:${msgId}`).digest("hex");
}

Reuse the same DynamoDB idempotency discipline from Lambda Powertools Idempotency if enrichment ever writes. Prefer read-only enrichment so retries are free.

Target Lambda: partial batch failure or you reprocess ghosts

SQS and Kinesis Pipes can batch. If your target returns success for the whole batch after one record throws, you either lose the bad record or reprocess the good ones forever. Report failures explicitly.

// target/process-orders.ts
import type { SQSBatchResponse, SQSEvent } from "aws-lambda";

export async function handler(event: SQSEvent): Promise<SQSBatchResponse> {
  const batchItemFailures: { itemIdentifier: string }[] = [];

  for (const rec of event.Records) {
    try {
      const body = JSON.parse(rec.body) as { orderId: string; tier: string };
      await applyOrder(body);
    } catch (err) {
      console.error("order_apply_failed", {
        messageId: rec.messageId,
        err: err instanceof Error ? err.message : String(err),
      });
      // ✅ Only failed ids — good records commit
      batchItemFailures.push({ itemIdentifier: rec.messageId });
    }
  }

  return { batchItemFailures };
}

async function applyOrder(_o: { orderId: string; tier: string }) {
  // business write — must be idempotent on orderId
}

In the Pipe/target configuration, enable the report-batch-item-failures feature and set reserved concurrency so a thundering enrichment cache miss does not melt DynamoDB. For cold-path latency budgets on the target, the measurement playbook in Lambda Cold Starts on Node 20 still applies.

Observability: prove drops are intentional

Pipes emit metrics; most teams never chart them next to source depth. Minimum dashboard:

• Source: ApproximateNumberOfMessagesVisible (SQS) or iterator age (Kinesis)
• Pipe: invocations, enrichment errors/timeouts, target failures
• Ratio: matched vs source receive rate (sudden drop = filter regression)
• DLQ depth + age alarm

// alerts/pipe-slo.ts — illustrative thresholds
export const PIPE_ALERTS = [
  { metric: "EnrichmentErrors", window: "5m", threshold: 5, action: "page" },
  { metric: "TargetFailures", window: "5m", threshold: 10, action: "page" },
  { metric: "DlqVisible", window: "1m", threshold: 1, action: "ticket" },
  // ✅ MatchedRate drop >50% vs 24h baseline after deploy → auto-rollback candidate
  { metric: "MatchedRateDropPct", window: "15m", threshold: 50, action: "page" },
];

Before cutover, dual-run: keep the old router Lambda writing to a shadow log, compare event ids processed by Pipe vs router for 24–72h. Only then delete the glue.

Checklist before you delete the router Lambda

• [ ] Pipe contract documents filter drop vs DLQ vs fail for enrichment errors
• [ ] Filter fixtures in CI (match + drop) for every pattern change
• [ ] Enrichment IAM is read-scoped; timeout < source visibility / shard retry budget
• [ ] Enrichment never mutates identity keys; versioned enrichment payload
• [ ] Target reports batchItemFailures; handler idempotent on business key
• [ ] DLQ + alarm on depth/age; enrichment/target error alarms
• [ ] Matched-rate dashboard vs baseline; dual-run comparison before cutover
• [ ] Schema/filter changes follow additive rules (schema evolution guide)

Related reading

• EventBridge Schema Evolution: Additive Changes Without Breaking Consumers
• Lambda Powertools Idempotency: DynamoDB Keys That Survive Retries
• Lambda Cold Starts on Node 20: Measure, Cut, and Keep Cutting
• Human-in-the-Loop Gates: Dual Control for Prod-Touching Agent Tools

Last updated on September 11, 2026