DynamoDB Streams Advanced Patterns for MCP Server Probe Pipelines: Filter Criteria, Global Tables Replication, PITR Export, and KCL Adapter

Pattern 1: Filter criteria — selective Lambda invocation for probe state transitions

Lambda ESM's FilterCriteria feature lets you declare which DynamoDB Streams records should invoke your function without writing a single line of application code to filter them. For an AliveMCP probe state table receiving 1,667 MODIFY events per second (100,000 endpoints probed every 60 seconds), the difference between filtered and unfiltered alerter Lambdas is 144 million invocations per day versus 1.44 million — a 100× cost reduction driven entirely by filter configuration.

Filter evaluation model

The filter evaluation happens in the Lambda service layer, after Lambda ESM reads the records from the DynamoDB Streams shard, before deciding whether to invoke the function. The key invariant: filtered records are still read from the shard and consumed toward the shard's read throughput — filtering reduces Lambda invocations and compute cost, not DynamoDB Streams capacity consumption. Multiple filters in the same FilterCriteria are OR-combined: a record invokes the function if it matches any one filter. Within a single filter, all patterns are AND-combined: every specified pattern must match.

The full DynamoDB Streams record structure that filters act on:

{
  "eventName": "MODIFY",        // INSERT | MODIFY | REMOVE
  "dynamodb": {
    "Keys": { "PK": { "S": "SERVER#alivemcp.com" }, "SK": { "S": "STATUS" } },
    "NewImage": {
      "status": { "S": "DEGRADED" },
      "consecutiveFailures": { "N": "5" }
    },
    "OldImage": {
      "status": { "S": "HEALTHY" },
      "consecutiveFailures": { "N": "0" }
    }
  }
}

Supported pattern types for Lambda ESM filter criteria:

Practical patterns for MCP probe state tables

Three filters cover the core alerting use cases for a probe state table, and all three can be attached to the same ESM (OR-combined) or split across dedicated Lambda functions:

// CDK: attach three filter patterns to the degradation alerter
alerterFn.addEventSource(new eventsources.DynamoEventSource(probeTable, {
  startingPosition: lambda.StartingPosition.LATEST,
  batchSize: 100,
  bisectBatchOnError: true,
  streamViewType: dynamodb.StreamViewType.NEW_AND_OLD_IMAGES,
  filters: [
    // 1. Status degradation: HEALTHY → DEGRADED or DOWN
    lambda.FilterCriteria.filter({
      eventName: lambda.FilterRule.isEqual('MODIFY'),
      dynamodb: {
        NewImage: { status: { S: lambda.FilterRule.notEquals('HEALTHY') } },
        OldImage: { status: { S: lambda.FilterRule.isEqual('HEALTHY') } }
      }
    }),
    // 2. New server registration (INSERT of canonical metadata record)
    lambda.FilterCriteria.filter({
      eventName: lambda.FilterRule.isEqual('INSERT'),
      dynamodb: { Keys: { SK: { S: lambda.FilterRule.isEqual('METADATA') } } }
    }),
    // 3. Recovery: non-healthy → HEALTHY (log recovery duration metric)
    lambda.FilterCriteria.filter({
      eventName: lambda.FilterRule.isEqual('MODIFY'),
      dynamodb: {
        NewImage: { status: { S: lambda.FilterRule.isEqual('HEALTHY') } },
        OldImage: { status: { S: lambda.FilterRule.notEquals('HEALTHY') } }
      }
    })
  ]
}));

StreamViewType requirements and the silent failure

Filters that reference OldImage require StreamViewType=NEW_AND_OLD_IMAGES. If the stream is configured with NEW_IMAGE only, OldImage is absent from every record — OldImage-based filter patterns never match, Lambda invocation drops to zero, and alerts stop firing with no error surfaced anywhere. The diagnosis is always the same: zero Lambda invocations during a known degradation event, and a filter match rate of 0% in ESM CloudWatch metrics. Changing StreamViewType on a live stream resets the shard sequence; plan this change during a maintenance window and verify stream continuity by comparing the first record in the new stream against the last record in the old stream. Full filter pattern reference at DynamoDB Streams Lambda Filter Criteria for MCP Servers.

Pattern 2: Global Tables replication — monitoring and conflict prevention

DynamoDB Global Tables v2 replicates writes across regions using multi-master active-active replication with eventual cross-region consistency. For AliveMCP's multi-region probe deployment, this means probe results written in us-east-1 may not appear in eu-west-1 for up to several seconds under normal conditions and up to 30+ seconds during regional network events — which affects alerter latency, dashboard freshness, and the validity of read-your-writes assumptions in the billing engine.

ReplicationLatency metric and alarm setup

DynamoDB publishes the ReplicationLatency metric to CloudWatch in the AWS/DynamoDB namespace. The metric captures the time in milliseconds for a write in the source region to appear in each destination region. For a 3-region setup, create 6 alarms (one per source-destination pair), set p99 alarm threshold at 30,000 ms, and add a maximum-statistic alarm at 60,000 ms to detect replication stalls:

// CDK: alarm on replication lag exceeding 30s p99 (us-east-1 → eu-west-1)
new cloudwatch.Alarm(this, 'ReplicationLagAlarm', {
  metric: new cloudwatch.Metric({
    namespace: 'AWS/DynamoDB',
    metricName: 'ReplicationLatency',
    dimensionsMap: { TableName: probeTable.tableName, ReceivingRegion: 'eu-west-1' },
    statistic: 'p99',
    period: cdk.Duration.minutes(1),
    region: 'us-east-1'  // Metric published in SOURCE region
  }),
  threshold: 30000,
  evaluationPeriods: 3,
  datapointsToAlarm: 2,
  treatMissingData: cloudwatch.TreatMissingData.NOT_BREACHING  // Absent when source has no writes
});

The treatMissingData=NOT_BREACHING setting is critical: if the source region has no writes for several minutes (e.g., during overnight low-traffic periods), the metric stops publishing and an alarm with BREACHING will false-fire. Implement synthetic heartbeat writes in each region if you need continuous metric flow.

Conflict resolution and write routing

When two regions write to the same item concurrently, Global Tables resolves the conflict by keeping the write with the higher aws:rep:updatetime value — a microsecond-precision timestamp that DynamoDB assigns internally. The losing write is silently discarded with no error returned to the writer. This attribute is visible in DynamoDB Streams NEW_AND_OLD_IMAGES records and can be used to detect replication overwrites by monitoring large jumps between OldImage and NewImage timestamps.

The most reliable way to prevent conflicts is home-region write routing — assign each MCP server slug to one region that owns status writes for that slug, and route writes from all other regions to the home region:

// Home region assignment via FNV-1a hash of server slug
function getHomeRegion(serverSlug: string): string {
  const regions = ['us-east-1', 'eu-west-1', 'ap-southeast-1'];
  let hash = 2166136261;
  for (let i = 0; i < serverSlug.length; i++) {
    hash ^= serverSlug.charCodeAt(i);
    hash = (hash * 16777619) >>> 0;
  }
  return regions[hash % regions.length];
}

// Route probe writes to home region; other regions add ~50-150ms round-trip
async function writeProbeResult(slug: string, status: string, latencyMs: number) {
  const homeRegion = getHomeRegion(slug);
  const client = homeRegion === process.env.AWS_REGION
    ? localDynamoClient
    : getClientForRegion(homeRegion);
  await client.updateItem({ /* UpdateExpression for status + latencyMs + lastCheckedAt */ });
}

Read routing is independent: use the local replica (eventually consistent) for real-time alerting and user dashboards; route to the home region (with ConsistentRead: true) for billing counts and subscription checks where stale data would cause incorrect results. Note that ConsistentRead: true only guarantees consistency within a single region — it does not provide cross-region consistency. Full replication monitoring patterns at DynamoDB Global Tables Replication Monitoring for MCP Servers.

Pattern 3: PITR and S3 export — 35-day recovery and zero-impact analytics

PITR and ExportTableToPointInTime serve distinct purposes for the probe state table. PITR is the disaster recovery layer: it provides a continuous 35-day rolling backup window at per-item granularity, so any application-layer corruption (bad migration, IAM misconfiguration, probe runner bug that overwrites all status records) can be restored to any second within the past 35 days. The export API is the analytics layer: it creates an S3 snapshot at a specific point in time that Athena can query with standard SQL without consuming a single DynamoDB read capacity unit from the production table.

PITR mechanics and restore

Enabling PITR adds $0.20/GB/month to table storage cost and has zero impact on table throughput. Restore creates a new table alongside the original — RestoreTableToPointInTime cannot overwrite the source table. Restore time scales with table size: minutes for tables under 1 GB, hours for tables over 100 GB.

# Restore to 24 hours ago (before a bad migration ran)
aws dynamodb restore-table-to-point-in-time \
  --source-table-name alivemcp-probe-state \
  --target-table-name alivemcp-probe-state-restored-20261031 \
  --restore-date-time "2026-10-30T14:00:00Z" \
  --billing-mode-override PAY_PER_REQUEST \
  --region us-east-1
# target table must not exist; GSIs restored automatically

ExportTableToPointInTime for monthly SLA reports

The export API creates a consistent snapshot at a specified time with zero impact on read/write capacity. Automate month-end exports with EventBridge Scheduler and a Lambda trigger:

// Lambda: trigger monthly export on first of each month
const exportPoint = new Date(now.getFullYear(), now.getMonth(), 1);

await dynamodb.exportTableToPointInTime({
  TableArn: process.env.TABLE_ARN!,
  ExportTime: exportPoint,
  S3Bucket: process.env.EXPORT_BUCKET!,
  S3Prefix: `exports/monthly/${exportPoint.toISOString().slice(0, 7)}/`,
  ExportFormat: 'DYNAMODB_JSON',  // or 'ION' (~30% smaller, less tooling support)
  ExportType: 'FULL_EXPORT'
});

Athena cannot query DynamoDB JSON exports with a standard JSON SerDe because DynamoDB JSON uses nested type descriptors ({"S":"value"} instead of "value"). Use the DynamoDB Ion SerDe or a Glue ETL job to flatten type descriptors before Athena queries.

Incremental export for continuous probe history

Incremental export (ExportType: INCREMENTAL_EXPORT) captures only items changed within a specified time window, enabling CDC-style probe history to S3 without a continuous Streams pipeline. The constraints that create silent coverage gaps:

An export window from 14:00 to 14:55 misses items modified between 14:55 and 15:00 if the next export starts at 15:00. To avoid gaps, overlap windows by 5–10 minutes and deduplicate by item-level timestamp downstream. For sub-minute CDC latency, use DynamoDB Streams instead of incremental export. Full PITR and export reference at DynamoDB PITR and S3 Export for MCP Servers.

Pattern 4: KCL Kinesis Adapter — distributed stream processing above Lambda ESM scale

Lambda ESM is the right default for DynamoDB Streams consumers — serverless, no fleet management, any Lambda runtime. The KCL Kinesis Adapter becomes necessary when Lambda ESM's constraints bind: the stream has more than ~50 shards with sustained high-throughput writes where per-shard Lambda invocation overhead dominates cost, or when multiple independent consumers need to read the same stream (DynamoDB Streams supports only 2 simultaneous readers; KCL allows multiple independent applications each with their own lease table).

Architecture: adapter API translation

DynamoDB Streams exposes the same GetShardIterator and GetRecords API surface as Kinesis, and the adapter bridges the two. Your KCL application calls standard Kinesis operations; the adapter translates them to DynamoDB Streams equivalents transparently:

ShardRecordProcessor and the shardEnded() checkpoint requirement

Your ShardRecordProcessor receives KCL Record objects whose data payload is the UTF-8 JSON of the DynamoDB Streams change event. In processRecords(), wrap per-record logic in try-catch and never rethrow — rethrowing halts all shard processing. Checkpoint after each batch for at-least-once semantics:

@Override
public void processRecords(ProcessRecordsInput input) {
  for (var record : input.records()) {
    try {
      String json = StandardCharsets.UTF_8.decode(record.data()).toString();
      Map<String, Object> streamRecord = MAPPER.readValue(json, Map.class);
      String eventName = (String) streamRecord.get("eventName");
      // dispatch on eventName: INSERT → provision, MODIFY → alert, REMOVE → cleanup
    } catch (Exception e) {
      // Log and continue — do NOT rethrow
      log.error("Failed to process record: {}", e.getMessage());
    }
  }
  try { input.checkpointer().checkpoint(); }
  catch (Exception e) { log.error("Checkpoint failed: {}", e.getMessage()); }
}

@Override
public void shardEnded(ShardEndedInput input) {
  // CRITICAL: must checkpoint SHARD_END or child shards never start
  try { input.checkpointer().checkpoint(); }
  catch (Exception e) { log.error("SHARD_END checkpoint failed: {}", e.getMessage()); }
}

24-hour shard lifecycle — the silent stoppage

DynamoDB Streams shards have a fixed 24-hour lifetime. When a shard expires, KCL calls shardEnded() on the processor owning that shard. If the processor does not checkpoint the SHARD_END sentinel sequence number, KCL considers the parent shard incomplete and will never start processing the child shard. This failure mode is silent: the processor continues to run and process other shards, but the affected partition stops being monitored without any error or alarm. Since every DynamoDB Streams shard expires every 24 hours, missing shardEnded() checkpoints means the processor progressively loses coverage over all partitions within 24 hours of deployment.

When choosing between Lambda ESM and the KCL adapter, the decision framework is primarily about shard count and write rate: Lambda ESM handles 100–200 shards at 1,667 writes/second (AliveMCP at 100k endpoints) without difficulty; KCL adapter is compelling above 50 shards with sustained >10k writes/minute, latency requirements below 2 seconds, or stateful aggregation that exceeds Lambda's 15-minute timeout.

Composing the four patterns: complete production checklist

The four patterns are independent in configuration but interdependent in guarantees — filter criteria cannot compensate for a missing StreamViewType, Global Tables conflict routing cannot compensate for a missing ReplicationLatency alarm, and the KCL adapter cannot compensate for a missing shardEnded() checkpoint. Each pattern has one configuration item that silently breaks the whole pattern if omitted.

For new AliveMCP DynamoDB Streams deployments, the correct configuration order minimizes risk: configure StreamViewType first (stream reset is disruptive — do it before wiring any consumers), then attach Lambda ESM with filter criteria (filter syntax errors surface immediately in ESM configuration validation), then enable PITR and test an export to verify EarliestRestorableDateTime, then add Global Tables replication alarms (alarm misconfiguration produces false alarms but no data loss), and finally configure home-region write routing (can be deployed behind a feature flag that routes 1% of traffic to verify no conflicts before full rollout).