DEV Community: André Paris

That was a challenge we faced and you may be interested if you are facing something similar in your company

André Paris — Sat, 07 Mar 2026 14:04:24 +0000

in case of fire put your heap on a cage - how we reduced our mem usage

Marcos Henrique for AWS Community Builders ・ Mar 6

Queue-Based Exponential Backoff: A Resilient Retry Pattern for Distributed Systems

André Paris — Mon, 20 Oct 2025 22:10:39 +0000

Introduction

When building distributed systems, handling transient failures gracefully is crucial for maintaining reliability and user experience. Traditional retry mechanisms using setTimeout or in-process schedulers have significant limitations in production environments. This article explores a more robust alternative: Queue-Based Exponential Backoff, a pattern that leverages message queues to implement resilient retry logic with exponential backoff and jitter.

The Problem with setTimeout

Before diving into the solution, let's understand why setTimeout falls short for retry logic in production:

// ❌ Anti-pattern: Using setTimeout for retries
async function processWithRetry(data: any) {
  try {
    await externalApiCall(data);
  } catch (error) {
    // Problems:
    // 1. Lost if process crashes
    // 2. Not distributed across instances
    // 3. No persistence
    // 4. Memory accumulates with many retries
    // 5. Hard to monitor and observe
    setTimeout(() => {
      processWithRetry(data);
    }, 5000);
  }
}

Key Issues:

No Durability: Retries are lost if the process crashes or restarts
Memory Leaks: Long-running timers accumulate in memory
Poor Scalability: Cannot distribute work across multiple workers
Limited Observability: Hard to track retry attempts and failures
No Backpressure: Can't throttle based on system load

The Solution: Queue-Based Retry

The Queue-Based Exponential Backoff pattern addresses these issues by:

Using a message queue's native delay feature (e.g., SQS DelaySeconds)
Deleting the original message and re-sending with a calculated delay
Implementing exponential backoff with jitter
Tracking retry count in the message payload

Architecture Overview

graph LR
    A[Producer Service] -->|1. Send Message| B[SNS Topic]
    B -->|2. Fan Out| C[SQS Queue]
    C -->|3. Poll| D[Worker Service]
    D -->|4a. Success| E[Delete Message]
    D -->|4b. Error| F{Retryable?}
    F -->|Yes| G[Calculate Backoff]
    G --> H[Delete Original]
    H --> I[Requeue with Delay]
    I --> C
    F -->|No| J[DLQ]
    D -->|Max Retries| J

Pattern Name

Queue-Based Exponential Backoff Pattern (also known as Message Queue Retry Pattern or Delayed Requeue Pattern)

Implementation

1. Worker Service with Retry Logic

import {
  SQSClient,
  ReceiveMessageCommand,
  DeleteMessageCommand,
  SendMessageCommand,
  Message,
} from '@aws-sdk/client-sqs';
import { Injectable } from '@nestjs/common';
import { Logger } from './common/logging';

@Injectable()
export class QueueWorkerService {
  private readonly logger = new Logger(QueueWorkerService.name);
  private readonly sqs = new SQSClient({});

  // Retry configuration based on industry best practices
  private readonly RETRY_CONFIG = {
    maxRetries: 50,                   // Maximum retry attempts
    baseDelayMultiplier: 3.0,         // Exponential backoff multiplier
    maxDelaySeconds: 60,              // Cap at 60 seconds (SQS max: 900s)
    jitterPercentage: 0.1,            // 10% jitter to prevent thundering herd
    totalTimeoutSeconds: 500,         // Total timeout for all attempts
  };

  // Error-specific retry configurations
  private readonly ERROR_RETRY_CONFIG = {
    RATE_LIMIT_EXCEEDED: {
      baseDelay: 60,    // Start with longer delay for rate limits
      maxDelay: 300,    // 5 minutes max
    },
    TEMPORARY_ERROR: {
      baseDelay: 2,     // Quick retry for transient errors
      maxDelay: 60,     // 1 minute max
    },
    QUOTA_EXCEEDED: {
      baseDelay: 120,   // Wait longer for quota replenishment
      maxDelay: 600,    // 10 minutes max
    },
    DEFAULT: {
      baseDelay: 2,     // Balanced default
      maxDelay: 60,     // 1 minute max
    }
  };

  /**
   * Main polling loop - continuously receives and processes messages
   */
  private async poll(handler: MessageHandler, queueUrl: string): Promise<void> {
    while (true) {
      try {
        const { Messages } = await this.sqs.send(
          new ReceiveMessageCommand({
            QueueUrl: queueUrl,
            MaxNumberOfMessages: 1,
            WaitTimeSeconds: 20,  // Long polling
          }),
        );

        if (!Messages || Messages.length === 0) {
          continue; // No messages, continue polling
        }

        for (const msg of Messages) {
          try {
            // Process the message
            await handler.handle(JSON.parse(msg.Body ?? '{}'));

            // Success - delete the message
            await this.sqs.send(
              new DeleteMessageCommand({
                QueueUrl: queueUrl,
                ReceiptHandle: msg.ReceiptHandle!,
              }),
            );
          } catch (err) {
            // Handle retry logic for failures
            await this.retryWithBackoff(err, queueUrl, msg);
          }
        }
      } catch (err) {
        this.logger.error(`Error polling queue`, err as any);
        await new Promise(r => setTimeout(r, 5_000)); // Back-off before retrying poll
      }
    }
  }

  /**
   * Core retry logic: Delete message and requeue with exponential backoff
   */
  private async retryWithBackoff(
    err: any, 
    queueUrl: string, 
    msg: Message
  ): Promise<void> {
    // Only retry if error is retryable
    const isRetryable = this.isRetryableError(err);

    if (!isRetryable) {
      this.logger.error(`Non-retryable error, letting message go to DLQ`, {
        messageId: msg.MessageId,
        error: err.message,
      });
      return; // Let message visibility timeout expire -> DLQ
    }

    // Extract retry count from message
    const retryCount = this.getRetryCount(msg);

    // Check if max retries exceeded
    if (retryCount >= this.RETRY_CONFIG.maxRetries) {
      this.logger.error(`Max retries exceeded, letting message go to DLQ`, {
        messageId: msg.MessageId,
        retryCount,
        maxRetries: this.RETRY_CONFIG.maxRetries,
      });
      return; // Let message go to DLQ
    }

    // Calculate delay with exponential backoff + jitter
    const delaySeconds = this.calculateRequeueDelay(err, retryCount);
    const nextRetryCount = retryCount + 1;

    this.logger.info(`Requeuing message with backoff`, {
      messageId: msg.MessageId,
      delaySeconds,
      retryCount: nextRetryCount,
      maxRetries: this.RETRY_CONFIG.maxRetries,
      error: err.message,
    });

    try {
      // Step 1: Delete the original message to prevent it from going to DLQ
      await this.sqs.send(new DeleteMessageCommand({
        QueueUrl: queueUrl,
        ReceiptHandle: msg.ReceiptHandle!,
      }));

      // Step 2: Re-send with delay and updated retry metadata
      await this.sqs.send(new SendMessageCommand({
        QueueUrl: queueUrl,
        MessageBody: JSON.stringify({
          ...JSON.parse(msg.Body!),
          retryCount: nextRetryCount,
          originalMessageId: msg.MessageId,
          retryTimestamp: new Date().toISOString(),
        }),
        DelaySeconds: Math.min(delaySeconds, 900), // SQS max delay is 900s
      }));

      this.logger.info(`Successfully requeued message`, {
        messageId: msg.MessageId,
        delaySeconds,
        retryProgress: `${nextRetryCount}/${this.RETRY_CONFIG.maxRetries}`,
      });
    } catch (requeueError) {
      this.logger.error(`Failed to requeue message`, {
        messageId: msg.MessageId,
        requeueError: requeueError instanceof Error 
          ? requeueError.message 
          : String(requeueError),
      });
      throw requeueError; // Let original message go to DLQ
    }
  }

  /**
   * Calculate exponential backoff delay with jitter
   */
  private calculateRequeueDelay(error: any, retryCount: number): number {
    // Determine base delay based on error type
    const errorType = this.classifyError(error);
    const config = this.ERROR_RETRY_CONFIG[errorType] || this.ERROR_RETRY_CONFIG.DEFAULT;
    const baseDelay = config.baseDelay;

    // Exponential backoff formula: baseDelay * (multiplier ^ retryCount)
    const exponentialDelay = baseDelay * Math.pow(
      this.RETRY_CONFIG.baseDelayMultiplier, 
      retryCount
    );

    // Add proportional jitter to prevent thundering herd
    // Jitter scales with retry count to maintain randomness
    const jitter = Math.random() * (exponentialDelay * this.RETRY_CONFIG.jitterPercentage);

    // Cap at both error-specific max and global max
    const maxDelay = Math.min(
      config.maxDelay,
      this.RETRY_CONFIG.maxDelaySeconds
    );

    const finalDelay = Math.min(exponentialDelay + jitter, maxDelay);

    this.logger.debug(`Calculated backoff delay`, {
      errorType,
      retryCount,
      baseDelay,
      exponentialDelay,
      jitter: jitter.toFixed(2),
      finalDelay: finalDelay.toFixed(2),
    });

    return Math.ceil(finalDelay); // Round up for SQS
  }

  /**
   * Extract retry count from message body
   */
  private getRetryCount(msg: Message): number {
    try {
      const body = JSON.parse(msg.Body!);
      return body.retryCount || 0;
    } catch {
      return 0;
    }
  }

  /**
   * Classify error to determine appropriate retry strategy
   */
  private classifyError(error: any): keyof typeof this.ERROR_RETRY_CONFIG {
    const message = error.message?.toLowerCase() || '';

    if (message.includes('rate limit') || message.includes('429')) {
      return 'RATE_LIMIT_EXCEEDED';
    }
    if (message.includes('quota exceeded') || message.includes('503')) {
      return 'QUOTA_EXCEEDED';
    }
    if (message.includes('timeout') || message.includes('temporary')) {
      return 'TEMPORARY_ERROR';
    }

    return 'DEFAULT';
  }

  /**
   * Determine if error is retryable
   */
  private isRetryableError(error: any): boolean {
    // Examples of non-retryable errors
    const nonRetryablePatterns = [
      'validation error',
      'not found',
      'unauthorized',
      'forbidden',
      '400',
      '401',
      '403',
      '404',
    ];

    const message = error.message?.toLowerCase() || '';
    return !nonRetryablePatterns.some(pattern => message.includes(pattern));
  }
}

2. CDK Infrastructure Setup

import * as cdk from 'aws-cdk-lib';
import * as sqs from 'aws-cdk-lib/aws-sqs';
import * as sns from 'aws-cdk-lib/aws-sns';
import * as snsSubscriptions from 'aws-cdk-lib/aws-sns-subscriptions';
import { Construct } from 'constructs';

export class QueueRetryStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // Create Dead Letter Queue (DLQ)
    const deadLetterQueue = new sqs.Queue(this, 'ProcessingDLQ', {
      queueName: 'my-service-processing-dlq',
      // Retain messages for 14 days for investigation
      retentionPeriod: cdk.Duration.days(14),
    });

    // Create main processing queue
    const processingQueue = new sqs.Queue(this, 'ProcessingQueue', {
      queueName: 'my-service-processing-queue',

      // Visibility timeout should be longer than processing time
      // This prevents other workers from receiving the message while it's being processed
      visibilityTimeout: cdk.Duration.seconds(60),

      // Message retention period
      retentionPeriod: cdk.Duration.days(14),

      // Configure DLQ - messages move here after maxReceiveCount
      deadLetterQueue: {
        queue: deadLetterQueue,
        // Set to 3 to prevent premature DLQ before our custom retry logic
        // Our custom logic handles up to 50 retries via delete+requeue
        maxReceiveCount: 3,
      },

      // Enable long polling to reduce costs and improve efficiency
      receiveMessageWaitTime: cdk.Duration.seconds(20),
    });

    // Optional: Create SNS topic for fanout pattern
    const topic = new sns.Topic(this, 'ProcessingTopic', {
      displayName: 'My Service Processing Topic',
    });

    // Subscribe queue to topic
    topic.addSubscription(
      new snsSubscriptions.SqsSubscription(processingQueue, {
        rawMessageDelivery: true,
        // Optional: Filter messages by attributes
        filterPolicy: {
          operation: sns.SubscriptionFilter.stringFilter({
            allowlist: ['create', 'update', 'delete'],
          }),
        },
      })
    );

    // Output queue URLs for application configuration
    new cdk.CfnOutput(this, 'QueueUrl', {
      value: processingQueue.queueUrl,
      exportName: 'ProcessingQueueUrl',
    });

    new cdk.CfnOutput(this, 'DLQUrl', {
      value: deadLetterQueue.queueUrl,
      exportName: 'ProcessingDLQUrl',
    });
  }
}

3. Custom Error Handler

/**
 * Custom error class for retryable errors
 */
export class RetryableError extends Error {
  constructor(
    message: string,
    public readonly errorType: 'RATE_LIMIT' | 'QUOTA' | 'TEMPORARY' | 'DEFAULT'
  ) {
    super(message);
    this.name = 'RetryableError';
  }
}

// Usage in your business logic
async function processExternalApiCall(data: any): Promise<void> {
  try {
    await externalApi.call(data);
  } catch (error: any) {
    if (error.statusCode === 429) {
      throw new RetryableError('Rate limit exceeded', 'RATE_LIMIT');
    }
    if (error.statusCode === 503) {
      throw new RetryableError('Service temporarily unavailable', 'TEMPORARY');
    }
    // Non-retryable errors
    throw error;
  }
}

Retry Flow Sequence

sequenceDiagram
    participant P as Producer
    participant Q as SQS Queue
    participant W as Worker
    participant H as Handler
    participant DLQ as Dead Letter Queue

    P->>Q: Send Message
    activate Q

    loop Polling
        W->>Q: Receive Message (Long Poll)
        Q->>W: Message
        deactivate Q

        activate W
        W->>H: Process Message
        activate H

        alt Success
            H-->>W: Success
            W->>Q: Delete Message
            Note over W,Q: Processing Complete
        else Retryable Error
            H-->>W: Retryable Error

            alt Under Max Retries
                W->>W: Calculate Backoff
                W->>Q: Delete Original
                W->>Q: Send with Delay
                Note over W,Q: Retry Scheduled
            else Max Retries Exceeded
                W->>Q: Let Expire
                Q->>DLQ: Move to DLQ
                Note over Q,DLQ: Manual Investigation
            end
        else Non-Retryable Error
            H-->>W: Non-Retryable Error
            W->>Q: Let Expire
            Q->>DLQ: Move to DLQ
        end
        deactivate H
        deactivate W
    end

Key Components Explained

1. Delete + Requeue Pattern

The core insight is to delete the original message and send a new one with a delay. This approach:

Prevents the message from going to DLQ prematurely
Allows precise control over retry timing
Enables retry count tracking in the message payload

// Step 1: Delete original
await sqs.deleteMessage({ ReceiptHandle });

// Step 2: Send new message with delay
await sqs.sendMessage({ 
  MessageBody,
  DelaySeconds: calculatedDelay 
});

2. Exponential Backoff Formula

delay = baseDelay × (multiplier ^ retryCount)

Example progression with baseDelay=2s and multiplier=3:

Retry	Calculation	Delay	Cumulative
1	2 × 3^0	2s	2s
2	2 × 3^1	6s	8s
3	2 × 3^2	18s	26s
4	2 × 3^3	54s	80s
5	2 × 3^4	162s → 60s*	140s

*Capped at maxDelay

3. Jitter

Jitter adds randomness to prevent multiple retries from happening simultaneously (thundering herd):

jitter = random() × (exponentialDelay × jitterPercentage)
finalDelay = exponentialDelay + jitter

4. Visibility Timeout vs. Delay

Visibility Timeout: How long a message is hidden after being received (prevents duplicate processing)
DelaySeconds: Initial delay before message becomes visible (used for retry scheduling)

Exponential Backoff Visualization

graph TD
    A[Attempt 1: Immediate] -->|Success| B[Done]
    A -->|Fail| C[Wait 2s]
    C --> D[Attempt 2]
    D -->|Success| B
    D -->|Fail| E[Wait 6s]
    E --> F[Attempt 3]
    F -->|Success| B
    F -->|Fail| G[Wait 18s]
    G --> H[Attempt 4]
    H -->|Success| B
    H -->|Fail| I[Wait 54s]
    I --> J[Attempt 5]
    J -->|Success| B
    J -->|Fail| K[Wait 60s - Capped]
    K --> L[Attempt 6]
    L -->|Success| B
    L -->|Fail| M[Continue with 60s delay...]
    M --> N[Eventually DLQ after Max Retries]

Error Classification Flow

flowchart TD
    Start([Error Occurred]) --> A{Is Retryable?}
    A -->|No| B[Let Message Expire]
    B --> C[Goes to DLQ]

    A -->|Yes| D{Retry Count < Max?}
    D -->|No| B

    D -->|Yes| E[Classify Error Type]
    E --> F[Get Base Delay for Error Type]

    F --> G[Calculate Exponential Delay<br/>delay = base × multiplier^retryCount]
    G --> H[Add Jitter<br/>jitter = random × delay × 0.1]
    H --> I[Apply Max Cap<br/>finalDelay = min delay + jitter, maxDelay]

    I --> J[Delete Original Message]
    J --> K[Increment Retry Count]
    K --> L[Requeue with DelaySeconds]
    L --> End([Message Requeued])

Pros and Cons

✅ Advantages

Durability: Retries survive process crashes and restarts
Scalability: Distributes work across multiple worker instances
Observability: Full visibility into queue metrics (depth, age, DLQ count)
Backpressure: Natural throttling based on queue depth
Cost-Effective: Pay only for messages processed, not idle timers
Battle-Tested: Leverages proven queue infrastructure
Flexible Retry Logic: Different strategies per error type
No Memory Leaks: Queue manages message lifecycle
Dead Letter Queue: Automatic handling of permanent failures
Distributed: Works seamlessly in multi-instance deployments

⚠️ Disadvantages

Complexity: More moving parts than simple setTimeout
Latency: Minimum delay granularity (1 second for SQS)
Cost: Queue operations have monetary cost (though minimal)
Message Duplication: Requires idempotent message handling
Debugging: Harder to debug compared to in-process retries
Max Delay Limit: SQS caps at 900 seconds (15 minutes)
Infrastructure Dependency: Requires queue service availability
Testing: More complex to test than in-memory retries

When to Use This Pattern

✅ Good Fit:

Integrating with rate-limited external APIs
Processing bulk operations with potential failures
Handling transient network or service errors
Systems requiring high availability and reliability
Distributed microservices architectures
Long-running background jobs

❌ Not Ideal:

Sub-second retry requirements
Single-process applications
Real-time user-facing operations (use circuit breakers instead)
Simple scripts or one-off jobs

Monitoring and Observability

Key metrics to track:

CloudWatch Metrics

ApproximateNumberOfMessagesVisible - Queue depth
ApproximateNumberOfMessagesDelayed - Retries in progress
ApproximateAgeOfOldestMessage - Processing lag
NumberOfMessagesSent - Throughput
NumberOfMessagesDeleted - Success rate
NumberOfMessagesInDLQ - Failure rate

Custom Application Metrics

Retry count distribution
Error types causing retries
Average delay per retry
Success rate by retry attempt
Time to success (first try vs. after retries)

Sample CloudWatch Dashboard

graph TB
    subgraph "CloudWatch Dashboard"
        subgraph "Queue Metrics"
            A[Messages Visible]
            B[Messages Delayed]
            C[Messages in DLQ]
            D[Oldest Message Age]
        end

        subgraph "Processing Metrics"
            E[Messages Received/sec]
            F[Messages Deleted/sec]
            G[Success Rate %]
            H[Avg Processing Time]
        end

        subgraph "Retry Metrics"
            I[Retry Count Distribution]
            J[Error Types]
            K[Backoff Delays]
            L[Time to Success]
        end

        subgraph "Alarms"
            M[🚨 Queue Depth > 5000]
            N[🚨 DLQ > 100]
            O[🚨 Msg Age > 15min]
            P[🚨 Error Rate > 30%]
        end
    end

Best Practices

Idempotency: Always design message handlers to be idempotent
Message Deduplication: Use message IDs to detect and skip duplicates
Timeout Management: Set visibility timeout > processing time + retry delay
DLQ Monitoring: Set up alerts for DLQ depth
Error Classification: Distinguish between retryable and permanent errors
Retry Limits: Set reasonable max retry counts (e.g., 50)
Logging: Include retry count, delay, and error type in logs
Testing: Test retry logic with chaos engineering
Cost Optimization: Use long polling to reduce API calls
Documentation: Document retry behavior for operations team

Configuration Examples

Standard API Integration

{
  maxRetries: 50,
  baseDelayMultiplier: 3.0,
  maxDelaySeconds: 60,
  jitterPercentage: 0.1,

  errorConfigs: {
    RATE_LIMIT: { baseDelay: 60, maxDelay: 300 },
    TEMPORARY: { baseDelay: 2, maxDelay: 60 },
    DEFAULT: { baseDelay: 2, maxDelay: 60 },
  }
}

High-Volume Processing

{
  maxRetries: 10,
  baseDelayMultiplier: 2.0,
  maxDelaySeconds: 30,

  queueConfig: {
    visibilityTimeout: 30,
    messageRetentionPeriod: 7,
    maxReceiveCount: 3,
  }
}

Critical Operations

{
  maxRetries: 5,
  baseDelayMultiplier: 2.0,
  maxDelaySeconds: 10,
  jitterPercentage: 0.2,

  errorConfigs: {
    TEMPORARY: { baseDelay: 1, maxDelay: 5 },
    DEFAULT: { baseDelay: 1, maxDelay: 5 },
  }
}

Comparison with Other Patterns

Pattern	Durability	Scalability	Observability	Complexity	Cost
setTimeout	❌ Low	❌ Poor	❌ Limited	✅ Simple	✅ Free
Queue-Based	✅ High	✅ Excellent	✅ Rich	⚠️ Moderate	⚠️ Low $
Cron Jobs	⚠️ Medium	⚠️ Fair	⚠️ Fair	⚠️ Moderate	✅ Free
Temporal/Conductor	✅ High	✅ Excellent	✅ Rich	❌ Complex	❌ High $

Real-World Use Cases

1. API Rate Limit Handling

When integrating with third-party APIs (Google Calendar, Stripe, etc.), rate limits are common. Queue-based retries with increasing delays naturally handle rate limiting without complex in-memory state.

Example: Syncing appointments to a CRM system

// When rate limited (HTTP 429), automatically retries with:
// Attempt 1: 60s delay
// Attempt 2: 180s delay (60 × 3^1)
// Attempt 3: 300s delay (capped at max)

2. Bulk Data Synchronization

Syncing large datasets between systems (e.g., appointments to a CRM) benefits from queue-based retries when individual records fail due to validation or temporary service issues.

Example: Processing 10,000 customer records

95% succeed on first attempt
4% succeed after 1-2 retries
1% go to DLQ for manual review

3. Webhook Processing

Processing incoming webhooks with external service calls can fail transiently. Queue-based retries ensure webhooks are eventually processed without losing data.

Example: Payment webhook from Stripe

Temporary database connection issue → Retry after 2s
Success on second attempt
Total processing time: 2s delay + processing time

Cost Estimation

AWS SQS Pricing (as of 2025)

Standard Queue: $0.40 per million requests (after 1M free tier)
Free Tier: 1 million requests per month

Example Calculation

Scenario: 10M messages/month with 20% retry rate

Total requests = 10M original + (10M × 20% retries)
               = 10M + 2M
               = 12M requests

Cost = (12M - 1M free tier) × $0.40/1M
     = 11M × $0.40/1M
     = $4.40/month

Comparison: Running a dedicated retry service would cost significantly more in infrastructure and operational overhead.

Testing Checklist

[ ] Message handlers are idempotent
[ ] Retry logic works for retryable errors
[ ] Non-retryable errors go to DLQ
[ ] Max retry count is enforced
[ ] Delays increase exponentially
[ ] Jitter prevents thundering herd
[ ] DLQ alerts are configured
[ ] Visibility timeout > processing time
[ ] Load test with high failure rate
[ ] Cost estimation reviewed
[ ] Monitoring dashboard created
[ ] Runbook documentation complete

Infrastructure Component Diagram

graph TB
    subgraph "Application Layer"
        A[Producer Service]
        B[Worker Service 1]
        C[Worker Service 2]
        D[Worker Service N]
    end

    subgraph "AWS Infrastructure"
        E[SNS Topic]
        F[SQS Queue]
        G[Dead Letter Queue]
        H[CloudWatch Metrics]
        I[CloudWatch Alarms]
    end

    subgraph "Monitoring"
        J[Queue Depth Alert]
        K[DLQ Depth Alert]
        L[Message Age Alert]
        M[Processing Time Alert]
    end

    A -->|Publish| E
    E -->|Subscribe| F
    F -->|Poll| B
    F -->|Poll| C
    F -->|Poll| D
    F -->|Failed Messages| G

    F -->|Metrics| H
    G -->|Metrics| H
    H --> I

    I --> J
    I --> K
    I --> L
    I --> M

Comparison: setTimeout vs Queue-Based

graph LR
    subgraph "setTimeout Pattern ❌"
        A1[Error] --> B1[setTimeout]
        B1 --> C1[In-Memory Timer]
        C1 --> D1{Process Alive?}
        D1 -->|Yes| E1[Retry]
        D1 -->|No| F1[Lost Forever]
    end

    subgraph "Queue-Based Pattern ✅"
        A2[Error] --> B2[Calculate Backoff]
        B2 --> C2[Delete + Requeue]
        C2 --> D2[Persistent Queue]
        D2 --> E2{Worker Available?}
        E2 -->|Yes| F2[Retry Immediately]
        E2 -->|No| G2[Wait in Queue]
        G2 --> F2
    end

References and Further Reading

Official Documentation

Industry Best Practices

Academic and Technical Papers

Marc Brooker et al. (2021). "Exponential Backoff And Jitter" - AWS Architecture Blog
Distributed Systems Observability - O'Reilly
Building Microservices - Sam Newman

Related Patterns

Bulkhead Pattern - Isolate failures
Dead Letter Queue Pattern - Handle permanent failures
Saga Pattern - Distributed transactions
CQRS Pattern - Command Query Responsibility Segregation

Open Source Examples

Bull Queue (Node.js) - Redis-based queue with retry
Celery (Python) - Distributed task queue
Hangfire (.NET) - Background job processing
Apache Kafka - Distributed streaming with retries

Tools and Libraries

AWS SDK for JavaScript v3
SQS Extended Client - For large messages (via S3)
LocalStack - Local development with SQS
AWS CDK - Infrastructure as Code

Monitoring and Observability

Conclusion

The Queue-Based Exponential Backoff pattern is a powerful, production-ready approach to handling transient failures in distributed systems. While it adds infrastructure complexity, the benefits of durability, scalability, and observability make it the right choice for most production workloads that require reliable retry logic.

By leveraging message queues' native delay capabilities and implementing proper exponential backoff with jitter, you can build resilient systems that gracefully handle failures without the pitfalls of in-process retry mechanisms.

Key Takeaways

Delete, Calculate, Requeue - The three steps to reliable retries
Exponential Backoff - Prevents overwhelming failed services
Jitter - Prevents thundering herd problems
Error Classification - Different strategies for different error types
Observability - Monitor queue metrics and retry patterns
Idempotency - Essential for reliable message processing
Dead Letter Queue - Catches permanent failures for investigation

Production Readiness

This pattern has been tested in production systems processing millions of messages daily, handling integrations with rate-limited APIs like Google Calendar, Kustomer CRM, Stripe, and other third-party services. The implementation has proven robust in handling various failure scenarios while maintaining high availability and reliability.

Next Steps

Implement the pattern in your system using the code examples
Monitor queue metrics and retry patterns
Tune configuration based on your specific use case
Document your retry strategy for the operations team
Test thoroughly with chaos engineering

Quick Reference

Minimal Implementation

async function retryWithBackoff(err: any, queueUrl: string, msg: Message) {
  const retryCount = getRetryCount(msg);
  const delay = 2 * Math.pow(3, retryCount); // Exponential backoff

  // 1. Delete original
  await sqs.deleteMessage({ ReceiptHandle: msg.ReceiptHandle });

  // 2. Requeue with delay
  await sqs.sendMessage({
    MessageBody: JSON.stringify({
      ...JSON.parse(msg.Body),
      retryCount: retryCount + 1,
    }),
    DelaySeconds: Math.min(delay, 900),
  });
}

CDK Quick Setup

const queue = new sqs.Queue(this, 'Queue', {
  visibilityTimeout: cdk.Duration.seconds(60),
  deadLetterQueue: { queue: dlq, maxReceiveCount: 3 },
  receiveMessageWaitTime: cdk.Duration.seconds(20),
});

Recommended Configuration

{
  maxRetries: 50,
  baseDelayMultiplier: 3.0,
  maxDelaySeconds: 60,
  jitterPercentage: 0.1,
}

Keywords: retry pattern, exponential backoff, message queue, SQS, distributed systems, resilience, AWS CDK, TypeScript, Node.js, microservices, error handling, queue-based retry, fault tolerance, scalability, observability

Difficulty Level: Intermediate to Advanced

Handling Large Payloads in Event-Driven Architectures on AWS (with CDK + TypeScript)

André Paris — Fri, 17 Oct 2025 03:02:19 +0000

When building event-driven systems in AWS, you’ll eventually hit a common problem — payload size limits.

Services like SNS, SQS, and EventBridge (256 KB per event) can become bottlenecks or point of failure if you send large JSON or binary payloads.

Fortunately, established architectural patterns help you manage large payloads gracefully. Below are five key patterns — with AWS + CDK examples — along with references you can use to dig deeper.

Patterns Overview

Here are the key patterns for managing large payloads in event-driven architectures:

Claim Check Pattern — Store the payload externally and send only a reference
Payload Compression — Compress before sending
Data Partitioning / Chunking — Split the payload into smaller messages
Enrichment Pattern — Send IDs; consumers fetch data as needed
Event Stream Filtering / Projection — Emit only the truly needed fields

1) Claim Check Pattern

Concept

Store the full payload outside the event (e.g. in Amazon S3) and publish a lightweight event containing only a “claim check” (e.g. object key, pre-signed URL, or pointer).

Process

Producer persists the payload in S3.
Producer publishes an event containing the S3 reference (or pointer).
Consumer reads that reference, fetches the payload from S3, and continues processing.

Benefits

Keeps event messages small
Reduces the load on the event bus
Decouples event routing from heavy data transfer

CDK + TypeScript Example

import * as cdk from "aws-cdk-lib";
import * as s3 from "aws-cdk-lib/aws-s3";
import * as sns from "aws-cdk-lib/aws-sns";
import * as subs from "aws-cdk-lib/aws-sns-subscriptions";
import * as lambda from "aws-cdk-lib/aws-lambda";

export class ClaimCheckStack extends cdk.Stack {
  constructor(scope: cdk.App, id: string) {
    super(scope, id);

    const bucket = new s3.Bucket(this, "PayloadBucket");
    const topic = new sns.Topic(this, "ClaimCheckTopic");

    const consumer = new lambda.Function(this, "ConsumerHandler", {
      runtime: lambda.Runtime.NODEJS_20_X,
      handler: "index.handler",
      code: lambda.Code.fromAsset("lambda"),
      environment: { BUCKET: bucket.bucketName },
    });

    topic.addSubscription(new subs.LambdaSubscription(consumer));
    bucket.grantRead(consumer);
  }
}

Event Format Example

{
  "type": "UserReportGenerated",
  "payloadLocation": "s3://my-bucket/reports/report-1234.json"
}

Real-World References

2) Payload Compression

Concept

Compress the payload (e.g. with Gzip, Snappy, LZ4) before embedding it in the event; the consumer decompresses it on receipt.

Process

Producer serializes and compresses the payload, encodes (e.g. Base64), then sends.
Consumer decodes and decompresses before processing.

Benefits

Reduces network usage and message size
Keeps everything in a single event
Adds CPU overhead and latency

Example (Node.js / Lambda)

import * as zlib from "zlib";

export const handler = async (event: any) => {
  // On producer side:
  const compressed = zlib.gzipSync(JSON.stringify(event.data)).toString("base64");

  // On consumer side:
  const decompressed = JSON.parse(
    zlib.gunzipSync(Buffer.from(compressed, "base64")).toString()
  );
  console.log("Received data:", decompressed);
};

Use compression when your payload is moderately large (e.g. 100 KB–250 KB) and still under the service limits.

3) Data Partitioning / Chunking

Concept

Divide a massive payload into manageable chunks, each sent as a separate event, with metadata to allow reassembly.

Process

Producer splits the payload into N parts.
Each event includes: correlationId, partNumber, totalParts, dataChunk.
Consumer receives all parts (possibly out of order), then reconstructs.

Benefits

Supports parallel and incremental processing
Avoids single large event breaches
Requires coordination and tracking

Event Payload Example

{
  "correlationId": "file-xyz",
  "partNumber": 4,
  "totalParts": 10,
  "dataChunk": "base64encodedSegment"
}

AWS Tips

Use SQS, Kinesis, or EventBridge Pipes as the transport for chunks
After full reassembly, persist result to S3 or DynamoDB

4) Enrichment Pattern (Contextual Data Fetching)

Concept

Emit minimal identifiers or context in the event. Consumers decide whether and when to fetch full data via APIs, databases, or external systems.

Process

Producer emits event containing just IDs or essential fields.
Consumer decides to call the service or DB to retrieve full details (lazily).

Benefits

Event bus stays lightweight
Consumers fetch only what they need
Reduces coupling between producers and consumers

Example Event

{
  "type": "OrderCreated",
  "orderId": "ORD-12345",
  "customerId": "CUST-98765"
}

Consumer Logic Example

const orderData = await ordersTable.get({ id: event.orderId });

5) Event Stream Filtering / Projection

Concept

Design event schemas and routing rules so that consumers receive only the subset of data they need, not a monolithic “everything” message.

Process

Producers emit minimal, well-defined event schemas.
Routers or event buses use filters or projections to trim fields.
Consumers get only relevant fields.

Benefits

Reduces data volume over time
Encourages clear schema contracts
Easier evolvability and consumer isolation

AWS (EventBridge) Example

import * as events from "aws-cdk-lib/aws-events";
import * as targets from "aws-cdk-lib/aws-events-targets";

new events.Rule(this, "FilteredRule", {
  eventPattern: {
    source: ["orders.service"],
    detailType: ["OrderCreated"],
    detail: {
      status: ["PAID"]
    }
  },
  targets: [new targets.LambdaFunction(handler)],
});

EventBridge Event Patterns Documentation

Choosing the Right Pattern

Scenario	Recommended Pattern	Why
Very large binary/data files	Claim Check	Offload storage to S3, send reference
Large but compressible message	Compression	Reduce size, maintain single event
Streaming / chunked content	Chunking / Partitioning	Process incrementally
Many consumers, but only some need full data	Enrichment	Fetch data on demand
Many consumers with different needs	Filtering / Projection	Route minimal fields to each consumer

What if the payload is not huge?

When the payload is small (well under 256 KB), you can keep it inline and select a simple event pattern based on fan-out and routing needs.

Use case	Recommended Pattern	Why
1→many fan-out	SNS → (Lambda/SQS)	Simple pub/sub
Cross-domain routing	EventBridge Bus	Filtering, schema registry
Point-to-point, retries	SQS → Lambda	Durable queue, DLQ
Ordered stream	Kinesis	Sharded, ordered ingestion

Inline Event Example

{
  "detail-type": "OrderCreated",
  "source": "orders.service",
  "detail": { "orderId": "ORD-12345", "amount": 129.99 }
}

TL;DR

If the payload is not large:

Keep it inline.
Use SNS for broadcast, SQS for queues, EventBridge for routing, Kinesis for ordered streams.
Always include idempotency keys, DLQs, and minimal schemas.

Building a Reusable AWS Governance Library with CDK: Constructs, Blueprints, and Aspects

André Paris — Tue, 14 Oct 2025 18:38:43 +0000

Introduction

As organizations scale their AWS infrastructure, maintaining consistent security, compliance, and best practices across multiple CDK projects becomes challenging. Each team might implement their own governance rules, leading to inconsistencies, security gaps, and compliance violations.

The solution? Build a reusable governance library that can be shared across all your CDK projects.

In this article, I'll show you how to build a production-ready AWS governance library using three powerful CDK concepts:

Constructs - For organizational structure
Blueprints - For property injection (enforcing defaults)
Aspects - For validation and compliance checking

Why a Governance Library?

Before diving into implementation, let's understand the problem:

Without a Governance Library

// Project A
const bucket = new s3.Bucket(this, 'Bucket', {
  encryption: s3.BucketEncryption.S3_MANAGED,
  blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
  // ... repeated security config
});

// Project B (different team)
const bucket = new s3.Bucket(this, 'Bucket', {
  // Oops! Forgot encryption, no public access block
});

// Project C
const bucket = new s3.Bucket(this, 'Bucket', {
  encryption: s3.BucketEncryption.KMS,
  // Inconsistent encryption method
});

Problems:

❌ Inconsistent security configurations
❌ Copy-paste errors
❌ No centralized compliance enforcement
❌ Difficult to audit
❌ Hard to update org-wide

With a Governance Library

// Just one line - consistent governance across all projects
CommonGovernancePatterns.soc2(app, 'myorg', 'my-project', 'production');

// All S3 buckets automatically get:
// ✅ Encryption enabled
// ✅ Public access blocked
// ✅ SSL enforced
// ✅ Versioning enabled (production)
// ✅ Organizational tags applied

Architecture Overview

Our governance library uses a three-layer architecture:

┌─────────────────────────────────────┐
│   GovernanceFactory (Entry Point)  │
│   - applyToApp()                    │
│   - applyToStack()                  │
└──────────────┬──────────────────────┘
               │
       ┌───────┴────────┐
       │                │
       ▼                ▼
┌─────────────┐  ┌──────────────┐
│  Blueprints │  │   Aspects    │
│  (Inject)   │  │  (Validate)  │
└─────────────┘  └──────────────┘
       │                │
       └────────┬───────┘
                ▼
    ┌──────────────────────┐
    │   Your CDK Resources │
    │   (S3, IAM, RDS...)  │
    └──────────────────────┘

Core Concepts Explained

1. Constructs

Constructs are CDK's building blocks. In our governance library, we use Constructs to organize and apply tags.

export class OrganizationalTags extends Construct {
  constructor(scope: Construct, id: string, private readonly config: GovernanceConfig) {
    super(scope, id);

    // Apply tags at scope level - they inherit to all children
    Tags.of(scope).add('Organization', config.organization);
    Tags.of(scope).add('Project', config.project);
    Tags.of(scope).add('Environment', config.environment);
    Tags.of(scope).add('ManagedBy', 'CDK');
    Tags.of(scope).add('CostCenter', `${config.organization}-${config.project}-${config.environment}`);

    // Apply additional organizational tags
    if (config.organizationalTags) {
      Object.entries(config.organizationalTags).forEach(([key, value]) => {
        Tags.of(scope).add(key, value);
      });
    }
  }
}

Key Points:

Constructs organize code into logical units
Tags applied at App level propagate to all stacks and resources
Single source of truth for organizational metadata

2. Blueprints (Property Injection)

Blueprints use CDK's Property Injection feature to enforce defaults before resources are created. They implement the IPropertyInjector interface.

How Property Injection Works

When you create a resource like new s3.Bucket(this, 'MyBucket', props), CDK:

Takes your props
Passes them through all registered property injectors
Injectors modify the props
Creates the resource with modified props

Example: S3 Security Blueprint

export class SecureS3Blueprint implements IPropertyInjector {
  public readonly constructUniqueId: string;

  constructor(private readonly config: GovernanceConfig) {
    // This ID tells CDK which construct type to inject into
    this.constructUniqueId = Bucket.PROPERTY_INJECTION_ID;
  }

  public inject(originalProps: BucketProps, context: InjectionContext): BucketProps {
    const enforcedProps: Partial<BucketProps> = {};

    // Block public access (override user's setting if needed)
    if (this.config.security?.blockPublicAccess !== false) {
      enforcedProps.blockPublicAccess = BlockPublicAccess.BLOCK_ALL;

      // Warn if we're overriding
      if (originalProps.blockPublicAccess !== BlockPublicAccess.BLOCK_ALL) {
        Annotations.of(context.scope).addWarning(
          `S3 bucket public access setting overridden for security compliance`
        );
      }
    }

    // Enforce SSL
    if (this.config.security?.enforceSSL !== false) {
      enforcedProps.enforceSSL = true;
    }

    // Enforce encryption
    if (this.config.security?.enforceEncryption !== false) {
      enforcedProps.encryption = BucketEncryption.S3_MANAGED;
    }

    // Enable versioning in production
    if (this.config.environment === 'production' && this.config.security?.enableVersioning !== false) {
      enforcedProps.versioned = true;
    }

    // Merge original props with enforced props
    return {
      ...originalProps,
      ...enforcedProps,
    };
  }
}

Example: IAM Role Blueprint

export class SecureIAMRoleBlueprint implements IPropertyInjector {
  public readonly constructUniqueId: string;

  constructor(private readonly config: GovernanceConfig) {
    this.constructUniqueId = Role.PROPERTY_INJECTION_ID;
  }

  public inject(originalProps: RoleProps, context: InjectionContext): RoleProps {
    const additionalPolicies: any[] = [];

    // Add CloudWatch logging in production
    if (this.config.environment === 'production') {
      additionalPolicies.push(
        ManagedPolicy.fromAwsManagedPolicyName('CloudWatchAgentServerPolicy')
      );
    }

    // Enforce session duration limits based on compliance
    const maxSessionDuration = this.getMaxSessionDuration();

    return {
      ...originalProps,
      managedPolicies: [
        ...(originalProps.managedPolicies || []),
        ...additionalPolicies
      ],
      maxSessionDuration: originalProps.maxSessionDuration || maxSessionDuration,
    };
  }

  private getMaxSessionDuration(): Duration {
    switch (this.config.complianceProfile) {
      case 'SOC2':
      case 'HIPAA':
        return Duration.hours(1); // Strict compliance
      case 'PCI':
        return Duration.hours(2);
      default:
        return this.config.environment === 'production' 
          ? Duration.hours(4) 
          : Duration.hours(12);
    }
  }
}

Blueprint Benefits:

✅ Enforces defaults before resource creation
✅ Can override user-provided properties
✅ Works transparently - developers don't need to change code
✅ Centralized security enforcement

3. Aspects (Validation)

Aspects validate resources after they're created but before synthesis. They implement the IAspect interface and visit every construct in the tree.

How Aspects Work

Aspects use the Visitor pattern:

CDK calls visit(node) for every construct
You check if the construct is a resource you care about
You validate and add errors/warnings/info annotations
Annotations appear in CDK output during synthesis

Example: Security Compliance Aspect

export class SecurityComplianceAspect implements IAspect {
  constructor(private readonly config: GovernanceConfig) {}

  public visit(node: IConstruct): void {
    // Check different resource types
    if (node instanceof s3.Bucket) {
      this.validateS3Security(node);
    }

    if (node instanceof iam.Role) {
      this.validateIAMRoleSecurity(node);
    }

    if (node instanceof rds.DatabaseInstance) {
      this.validateRDSInstanceSecurity(node);
    }

    if (node instanceof ec2.SecurityGroup) {
      this.validateSecurityGroupRules(node);
    }
  }

  private validateS3Security(bucket: s3.Bucket): void {
    const bucketResource = bucket.node.defaultChild as s3.CfnBucket;

    // Check encryption
    if (this.config.security?.enforceEncryption && !bucketResource.bucketEncryption) {
      Annotations.of(bucket).addError(
        `S3 bucket "${bucket.node.id}" must have encryption enabled`
      );
    }

    // Check versioning in production
    if (this.config.environment === 'production') {
      const versioningConfig = bucketResource.versioningConfiguration;
      if (!versioningConfig || versioningConfig.status !== 'Enabled') {
        Annotations.of(bucket).addError(
          `S3 bucket "${bucket.node.id}" must have versioning enabled in production`
        );
      }
    }
  }

  private validateIAMRoleSecurity(role: iam.Role): void {
    const roleResource = role.node.defaultChild as iam.CfnRole;

    // Check for overly permissive policies
    const policies = roleResource.policies;
    if (policies && Array.isArray(policies)) {
      policies.forEach((policy: any) => {
        const statements = policy.policyDocument?.Statement;
        if (Array.isArray(statements)) {
          statements.forEach((statement: any) => {
            if (statement.Effect === 'Allow' && 
                statement.Resource === '*' && 
                statement.Action === '*') {
              Annotations.of(role).addError(
                `IAM role "${role.roleName}" has overly permissive policy with Action: "*" and Resource: "*"`
              );
            }
          });
        }
      });
    }
  }

  private validateSecurityGroupRules(sg: ec2.SecurityGroup): void {
    const sgResource = sg.node.defaultChild as ec2.CfnSecurityGroup;

    if (sgResource.securityGroupIngress) {
      const ingress = Array.isArray(sgResource.securityGroupIngress) 
        ? sgResource.securityGroupIngress 
        : [sgResource.securityGroupIngress];

      ingress.forEach((rule: any) => {
        // Check for SSH/RDP from 0.0.0.0/0
        if (rule.cidrIp === '0.0.0.0/0') {
          if (rule.fromPort === 22 || rule.fromPort === 3389) {
            Annotations.of(sg).addError(
              `Security group allows SSH/RDP access from 0.0.0.0/0 - critical security risk!`
            );
          }
        }
      });
    }
  }
}

Example: Naming Convention Aspect

export class NamingConventionAspect implements IAspect {
  constructor(private readonly config: GovernanceConfig) {}

  public visit(node: IConstruct): void {
    if (!this.config.naming?.enforce) return;

    const expectedPrefix = this.config.naming.prefix || 
      `${this.config.organization}-${this.config.environment}`;

    // Validate S3 bucket names
    if (node instanceof s3.Bucket) {
      const bucketName = node.bucketName;

      if (bucketName && !bucketName.startsWith(expectedPrefix)) {
        Annotations.of(node).addError(
          `S3 bucket name "${bucketName}" must start with "${expectedPrefix}"`
        );
      }
    }

    // Validate IAM role names
    if (node instanceof iam.Role) {
      const roleName = node.roleName;

      if (roleName && !roleName.includes(this.config.organization)) {
        Annotations.of(node).addWarning(
          `IAM role name "${roleName}" should include organization "${this.config.organization}"`
        );
      }
    }
  }
}

Aspect Benefits:

✅ Validates resources after creation
✅ Catches configuration issues before deployment
✅ Provides clear error messages
✅ Non-invasive - doesn't modify resources

Configuration System

Create a flexible configuration interface:

export interface GovernanceConfig {
  // Core identifiers
  organization: string;
  project: string;
  environment: 'development' | 'staging' | 'production';

  // Compliance profiles
  complianceProfile?: 'SOC2' | 'HIPAA' | 'PCI' | 'GDPR' | 'basic';

  // Custom tags
  requiredTags?: string[];
  organizationalTags?: Record<string, string>;

  // Security settings
  security?: {
    enforceEncryption?: boolean;
    enforceSSL?: boolean;
    blockPublicAccess?: boolean;
    enableVersioning?: boolean;
  };

  // Naming conventions
  naming?: {
    prefix?: string;
    pattern?: RegExp;
    enforce?: boolean;
  };

  // Tagging strategy
  tagging?: {
    autoApply?: boolean;
    enforceRequired?: boolean;
  };
}

Compliance Profiles

Pre-configure settings for common compliance frameworks:

export const COMPLIANCE_PROFILES: Record<string, Partial<GovernanceConfig>> = {
  basic: {
    security: {
      enforceEncryption: true,
      enforceSSL: true,
      blockPublicAccess: true,
    },
    requiredTags: ['Owner', 'Environment'],
  },

  SOC2: {
    security: {
      enforceEncryption: true,
      enforceSSL: true,
      blockPublicAccess: true,
      enableVersioning: true,
    },
    requiredTags: ['Owner', 'Team', 'DataClassification', 'Compliance'],
    tagging: {
      enforceRequired: true,
    },
  },

  HIPAA: {
    security: {
      enforceEncryption: true,
      enforceSSL: true,
      blockPublicAccess: true,
      enableVersioning: true,
    },
    requiredTags: ['Owner', 'Team', 'DataClassification', 'PHI', 'Compliance'],
    tagging: {
      enforceRequired: true,
    },
  },
};

export function mergeGovernanceConfig(config: GovernanceConfig): GovernanceConfig {
  const profile = config.complianceProfile 
    ? COMPLIANCE_PROFILES[config.complianceProfile] 
    : {};

  return {
    ...DEFAULT_GOVERNANCE_CONFIG,
    ...profile,
    ...config,
    security: {
      ...DEFAULT_GOVERNANCE_CONFIG.security,
      ...profile.security,
      ...config.security,
    },
  } as GovernanceConfig;
}

Governance Factory

Create a factory class to orchestrate everything:

export class GovernanceFactory {
  /**
   * Apply governance to an entire CDK App
   */
  static applyToApp(app: App, config: GovernanceConfig): void {
    const mergedConfig = mergeGovernanceConfig(config);

    // 1. Apply blueprints (property injection)
    this.applyBlueprints(app, mergedConfig);

    // 2. Apply organizational tags
    new OrganizationalTags(app, 'AppGovernance', mergedConfig);

    // 3. Apply aspects (validation)
    this.applyAspects(app, mergedConfig);
  }

  /**
   * Apply governance to a specific stack
   */
  static applyToStack(stack: Stack, config: GovernanceConfig): void {
    const mergedConfig = mergeGovernanceConfig(config);

    new OrganizationalTags(stack, 'StackGovernance', mergedConfig);
    this.applyAspects(stack, mergedConfig);
  }

  /**
   * Apply blueprints via property injection
   */
  private static applyBlueprints(app: App, config: GovernanceConfig): void {
    const blueprints = [
      new SecureS3Blueprint(config),
      new SecureIAMRoleBlueprint(config),
      new SecureRDSBlueprint(config),
    ];

    // Register blueprints with the App
    const existingInjectors = (app as any).propertyInjectors || [];
    (app as any).propertyInjectors = [...existingInjectors, ...blueprints];
  }

  /**
   * Apply aspects for validation
   */
  private static applyAspects(scope: Construct, config: GovernanceConfig): void {
    // Security compliance
    Aspects.of(scope).add(new SecurityComplianceAspect(config));

    // Naming conventions
    if (config.naming?.enforce) {
      Aspects.of(scope).add(new NamingConventionAspect(config));
    }

    // Required tags validation
    if (config.requiredTags && config.requiredTags.length > 0) {
      Aspects.of(scope).add(new RequiredTagsValidationAspect(config));
    }

    // Compliance tag enforcement
    if (config.complianceProfile && config.complianceProfile !== 'basic') {
      Aspects.of(scope).add(new ComplianceTagEnforcementAspect(config));
    }
  }
}

Convenience Patterns

Add common usage patterns:

export class CommonGovernancePatterns {
  /**
   * Apply basic governance (suitable for most projects)
   */
  static basic(
    app: App, 
    organization: string, 
    project: string, 
    environment: 'development' | 'staging' | 'production'
  ): void {
    GovernanceFactory.applyToApp(app, {
      organization,
      project,
      environment,
      complianceProfile: 'basic',
    });
  }

  /**
   * Apply SOC2 compliance governance
   */
  static soc2(
    app: App, 
    organization: string, 
    project: string, 
    environment: 'development' | 'staging' | 'production'
  ): void {
    GovernanceFactory.applyToApp(app, {
      organization,
      project,
      environment,
      complianceProfile: 'SOC2',
    });
  }

  /**
   * Development-friendly governance (less strict)
   */
  static development(app: App, organization: string, project: string): void {
    GovernanceFactory.applyToApp(app, {
      organization,
      project,
      environment: 'development',
      complianceProfile: 'basic',
      security: {
        enforceEncryption: false, // More flexible for dev
        enforceSSL: true,
        blockPublicAccess: true,
        enableVersioning: false,
      },
      tagging: {
        enforceRequired: false, // Warnings instead of errors
      },
    });
  }

  /**
   * Production-ready governance (strict)
   */
  static production(
    app: App, 
    organization: string, 
    project: string, 
    complianceProfile: 'SOC2' | 'HIPAA' | 'PCI' | 'GDPR' = 'SOC2'
  ): void {
    GovernanceFactory.applyToApp(app, {
      organization,
      project,
      environment: 'production',
      complianceProfile,
      security: {
        enforceEncryption: true,
        enforceSSL: true,
        blockPublicAccess: true,
        enableVersioning: true,
      },
      tagging: {
        enforceRequired: true, // Strict enforcement
      },
      naming: {
        enforce: true,
      },
    });
  }
}

Package Structure

Organize your library for reusability:

aws-governance-lib/
├── src/
│   ├── config/
│   │   └── governance-config.ts      # Configuration interfaces
│   ├── constructs/
│   │   └── governance-factory.ts     # Main factory class
│   ├── blueprints/
│   │   └── security-blueprints.ts    # Property injectors
│   ├── aspects/
│   │   ├── core-governance.ts        # Tags, naming, etc.
│   │   ├── security-compliance.ts    # Security validation
│   │   └── infrastructure-validation.ts  # Resource-specific
│   ├── utils/
│   │   └── blueprint-helpers.ts      # Helper functions
│   └── index.ts                      # Public exports
├── test/
│   └── governance.test.ts
├── package.json
├── tsconfig.json
└── README.md

package.json

{
  "name": "@myorg/aws-governance-lib",
  "version": "1.0.0",
  "description": "Reusable AWS CDK governance library",
  "main": "lib/index.js",
  "types": "lib/index.d.ts",
  "scripts": {
    "build": "tsc",
    "test": "jest",
    "prepublishOnly": "npm run build"
  },
  "peerDependencies": {
    "aws-cdk-lib": "^2.100.0",
    "constructs": "^10.0.0"
  },
  "keywords": [
    "aws",
    "cdk",
    "governance",
    "compliance",
    "security"
  ]
}

Using the Library

Installation

npm install @myorg/aws-governance-lib

Basic Usage

import { App } from 'aws-cdk-lib';
import { CommonGovernancePatterns } from '@myorg/aws-governance-lib';
import { MyStack } from './stacks/MyStack';

const app = new App();

// Apply governance with one line
CommonGovernancePatterns.basic(app, 'myorg', 'my-project', 'production');

// Create your stacks normally
new MyStack(app, 'MyStack', {});

app.synth();

Advanced Usage

import { App } from 'aws-cdk-lib';
import { GovernanceFactory } from '@myorg/aws-governance-lib';

const app = new App();

// Custom configuration
GovernanceFactory.applyToApp(app, {
  organization: 'myorg',
  project: 'data-platform',
  environment: 'production',
  complianceProfile: 'SOC2',
  organizationalTags: {
    'BusinessUnit': 'Engineering',
    'Department': 'Platform',
    'CostCenter': 'CC-1234'
  },
  security: {
    enforceEncryption: true,
    enforceSSL: true,
    blockPublicAccess: true,
    enableVersioning: true,
  },
  naming: {
    prefix: 'myorg-prod',
    enforce: true,
  }
});

Per-Stack Governance

const app = new App();

// Different governance per stack
const devStack = new MyStack(app, 'DevStack', { env: 'dev' });
const prodStack = new MyStack(app, 'ProdStack', { env: 'prod' });

GovernanceFactory.applyToStack(devStack, {
  organization: 'myorg',
  project: 'my-project',
  environment: 'development',
  complianceProfile: 'basic',
});

GovernanceFactory.applyToStack(prodStack, {
  organization: 'myorg',
  project: 'my-project',
  environment: 'production',
  complianceProfile: 'SOC2',
});

Real-World Example

Let's see a complete example with validation output:

// bin/app.ts
import { App } from 'aws-cdk-lib';
import { CommonGovernancePatterns } from '@myorg/aws-governance-lib';
import { DataPipelineStack } from './stacks/DataPipelineStack';

const app = new App();

// Apply SOC2 governance
CommonGovernancePatterns.soc2(app, 'myorg', 'data-platform', 'production');

// Create stack
new DataPipelineStack(app, 'DataPipelineStack', {
  env: {
    account: process.env.CDK_DEFAULT_ACCOUNT,
    region: process.env.CDK_DEFAULT_REGION,
  },
});

app.synth();

// stacks/DataPipelineStack.ts
import { Stack, StackProps } from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as iam from 'aws-cdk-lib/aws-iam';

export class DataPipelineStack extends Stack {
  constructor(scope: Construct, id: string, props?: StackProps) {
    super(scope, id, props);

    // Create S3 bucket - governance applied automatically!
    const bucket = new s3.Bucket(this, 'DataLake', {
      // You don't need to specify security settings
      // The governance library handles it
    });

    // Blueprint will inject:
    // - blockPublicAccess: BlockPublicAccess.BLOCK_ALL
    // - encryption: BucketEncryption.S3_MANAGED
    // - enforceSSL: true
    // - versioned: true (production)

    // Aspect will validate:
    // - All security settings are correct
    // - Naming convention follows org standards
    // - Required tags are present

    // Create IAM role
    const role = new iam.Role(this, 'DataPipelineRole', {
      assumedBy: new iam.ServicePrincipal('glue.amazonaws.com'),
    });

    // Blueprint will inject:
    // - CloudWatch managed policies (production)
    // - Maximum session duration (1 hour for SOC2)
  }
}

Synthesis Output

When you run cdk synth, you'll see:

[WARNING] S3 bucket "DataLake" missing required tag "DataClassification" - required for SOC2 compliance
[INFO] Ensure S3 bucket "DataLake" has SSL enforcement via bucket policy
✅ Successfully applied organizational tags: Organization=myorg, Project=data-platform, Environment=production
✅ Security compliance checks passed for 5 resources

Testing the Library

Create comprehensive tests:

import { App, Stack } from 'aws-cdk-lib';
import { Template, Annotations, Match } from 'aws-cdk-lib/assertions';
import * as s3 from 'aws-cdk-lib/aws-s3';
import { GovernanceFactory } from '../src';

describe('Governance Library', () => {
  test('applies S3 security defaults', () => {
    const app = new App();
    const stack = new Stack(app, 'TestStack');

    GovernanceFactory.applyToStack(stack, {
      organization: 'test',
      project: 'test-project',
      environment: 'production',
      complianceProfile: 'SOC2',
    });

    new s3.Bucket(stack, 'TestBucket', {});

    const template = Template.fromStack(stack);

    // Verify security settings were injected
    template.hasResourceProperties('AWS::S3::Bucket', {
      PublicAccessBlockConfiguration: {
        BlockPublicAcls: true,
        BlockPublicPolicy: true,
        IgnorePublicAcls: true,
        RestrictPublicBuckets: true,
      },
      BucketEncryption: {
        ServerSideEncryptionConfiguration: Match.anyValue(),
      },
      VersioningConfiguration: {
        Status: 'Enabled',
      },
    });
  });

  test('validates naming conventions', () => {
    const app = new App();
    const stack = new Stack(app, 'TestStack');

    GovernanceFactory.applyToStack(stack, {
      organization: 'test',
      project: 'test-project',
      environment: 'production',
      naming: {
        prefix: 'test-prod',
        enforce: true,
      },
    });

    new s3.Bucket(stack, 'TestBucket', {
      bucketName: 'wrong-prefix-bucket',
    });

    app.synth();

    // Check for naming convention errors
    const annotations = Annotations.fromStack(stack).findError(
      '*',
      Match.stringLikeRegexp('.*must start with "test-prod".*')
    );

    expect(annotations.length).toBeGreaterThan(0);
  });

  test('applies organizational tags', () => {
    const app = new App();
    const stack = new Stack(app, 'TestStack');

    GovernanceFactory.applyToStack(stack, {
      organization: 'test',
      project: 'test-project',
      environment: 'development',
    });

    new s3.Bucket(stack, 'TestBucket', {});

    const template = Template.fromStack(stack);

    // Verify tags were applied
    template.hasResourceProperties('AWS::S3::Bucket', {
      Tags: Match.arrayWith([
        { Key: 'Organization', Value: 'test' },
        { Key: 'Project', Value: 'test-project' },
        { Key: 'Environment', Value: 'development' },
        { Key: 'ManagedBy', Value: 'CDK' },
      ]),
    });
  });
});

Best Practices

1. Start Simple, Expand Gradually

// Phase 1: Just tagging
GovernanceFactory.applyToApp(app, {
  organization: 'myorg',
  project: 'my-project',
  environment: 'production',
});

// Phase 2: Add basic security
// (update library, users automatically get new features)

// Phase 3: Add compliance profiles
// (rollout with migration guide)

2. Make Warnings Non-Breaking

// Use info/warnings for new validations
if (config.strictMode) {
  Annotations.of(node).addError('Must have encryption');
} else {
  Annotations.of(node).addWarning('Should have encryption');
}

3. Allow Opt-Out Mechanisms

// Let teams opt-out if needed
if (this.config.security?.enforceEncryption !== false) {
  // Apply encryption
}

// Usage:
GovernanceFactory.applyToApp(app, {
  // ...
  security: {
    enforceEncryption: false, // Opt out
  },
});

4. Document Governance Decisions

/**
 * Enforces S3 bucket encryption for SOC2 compliance
 * 
 * Required by: SOC2 CC6.1 - Encryption at Rest
 * Exceptions: Must be approved by Security team
 * Contact: security-team@myorg.com
 */
if (this.config.complianceProfile === 'SOC2') {
  enforcedProps.encryption = BucketEncryption.S3_MANAGED;
}

5. Version Your Library Carefully

Use semantic versioning:

Patch (1.0.1): Bug fixes, non-breaking
Minor (1.1.0): New features, non-breaking
Major (2.0.0): Breaking changes (new enforcements)

Migration Strategy

Step 1: Install Library

npm install @myorg/aws-governance-lib

Step 2: Replace Existing Governance

// BEFORE
import { applyStackGovernance } from './factories/stack-factory';
applyStackGovernance(stack, 'production');

// AFTER
import { CommonGovernancePatterns } from '@myorg/aws-governance-lib';
CommonGovernancePatterns.production(app, 'myorg', 'my-project', 'SOC2');

Step 3: Keep Project-Specific Aspects

import { GovernanceFactory } from '@myorg/aws-governance-lib';
import { MyCustomAspect } from './aspects/custom-aspects';

// Apply library governance
GovernanceFactory.applyToApp(app, config);

// Add project-specific aspects
Aspects.of(app).add(new MyCustomAspect());

Advanced: Resource-Specific Blueprints

Add blueprints for other AWS services:

// Lambda function blueprint
export class SecureLambdaBlueprint implements IPropertyInjector {
  public readonly constructUniqueId = Function.PROPERTY_INJECTION_ID;

  public inject(originalProps: FunctionProps, context: InjectionContext): FunctionProps {
    return {
      ...originalProps,
      // Enable X-Ray tracing
      tracing: Tracing.ACTIVE,
      // Set environment encryption
      environmentEncryption: new kms.Key(context.scope, 'LambdaKey'),
      // Set timeout limits
      timeout: originalProps.timeout || Duration.seconds(30),
    };
  }
}

// RDS blueprint
export class SecureRDSBlueprint implements IPropertyInjector {
  public readonly constructUniqueId = DatabaseInstance.PROPERTY_INJECTION_ID;

  public inject(originalProps: DatabaseInstanceProps, context: InjectionContext): DatabaseInstanceProps {
    return {
      ...originalProps,
      storageEncrypted: true,
      backupRetention: Duration.days(7),
      multiAz: this.config.environment === 'production',
      deletionProtection: this.config.environment === 'production',
      publiclyAccessible: false,
    };
  }
}

Conclusion

Building a reusable governance library transforms how you manage AWS infrastructure:

Benefits Recap

✅ Consistency - All projects follow the same standards
✅ Security - Enforce security by default
✅ Compliance - Built-in SOC2, HIPAA, PCI, GDPR profiles
✅ Maintainability - Update governance in one place
✅ Developer Experience - One-line setup, transparent enforcement
✅ Auditability - Clear validation output
✅ Scalability - Works across unlimited projects

Key Takeaways

Use Blueprints for property injection (enforce defaults)
Use Aspects for validation (catch issues)
Use Constructs for organization (apply tags)
Make it configurable - different needs for dev/prod
Version carefully - breaking changes need migration guides
Document everything - explain why, not just what

Next Steps

Start with tagging and basic security
Add compliance profiles based on your needs
Expand to cover more AWS services
Get feedback from teams
Iterate and improve

With this architecture, you can ensure every CDK project in your organization follows best practices automatically, freeing developers to focus on building features instead of configuring security.

Resources

This architecture is battle-tested across multiple production environments with 100+ stacks and ensures consistent governance at scale.

Building a Production-Ready Data Lake: PostgreSQL to S3 with AWS DMS, Glue, and Athena using CDK

André Paris — Tue, 14 Oct 2025 18:20:17 +0000

Introduction

When building modern data platforms, one of the most common challenges is replicating relational database data into a data lake for analytics. AWS Database Migration Service (DMS) provides a powerful solution for continuous data replication from PostgreSQL to S3, complete with Change Data Capture (CDC) capabilities.

In this article, I'll show you how to build a production-ready data pipeline using AWS CDK (TypeScript) that:

Replicates PostgreSQL data to S3 in Parquet format
Automatically catalogs tables using AWS Glue
Enables SQL queries via Amazon Athena
Supports both full load and ongoing CDC

Architecture Overview

Our solution follows a layered architecture pattern:

PostgreSQL (RDS) 
    ↓
AWS DMS (Serverless)
    ↓
S3 Data Lake (Parquet)
    ↓
AWS Glue Catalog
    ↓
Amazon Athena

Key Components

DMS Replication: Handles full load and CDC from PostgreSQL
S3 Data Lake: Stores data in compressed Parquet format
Glue Data Catalog: Provides schema discovery and metadata management
Athena Workgroups: Enables SQL queries on the data lake

Prerequisites

Before starting, ensure you have:

AWS CDK installed (npm install -g aws-cdk)
An existing VPC with private subnets
A PostgreSQL database with logical replication enabled
Database credentials stored in AWS Secrets Manager
TypeScript knowledge

Implementation

Step 1: Base S3 Construct

First, we create a secure S3 bucket for our data lake:

import { Construct } from 'constructs';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as iam from 'aws-cdk-lib/aws-iam';
import { RemovalPolicy } from 'aws-cdk-lib';

export interface S3BaseProps {
    bucketName?: string;
    removalPolicy?: RemovalPolicy;
    iamRolesProps?: iam.RoleProps[];
}

export class S3Base extends Construct {
    public readonly bucket: s3.Bucket;
    public readonly iamRoles: iam.Role[];

    constructor(scope: Construct, id: string, props: S3BaseProps = {}) {
        super(scope, id);

        this.iamRoles = [];
        this.bucket = new s3.Bucket(this, `instance-${id}`, {
            bucketName: props.bucketName,
            versioned: true,
            blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
            encryption: s3.BucketEncryption.S3_MANAGED,
            enforceSSL: true,
            removalPolicy: props.removalPolicy ?? RemovalPolicy.RETAIN,
            autoDeleteObjects: false,
        });

        // Create IAM roles for DMS access
        if (props.iamRolesProps && props.iamRolesProps.length > 0) {
            props.iamRolesProps.forEach((iamRolesProp, index) => {
                const role = new iam.Role(this, `bucket-${id}-${index}-role`, iamRolesProp);
                this.bucket.grantReadWrite(role);
                this.iamRoles.push(role);
            });
        }
    }
}

Step 2: DMS Base Construct

The DMS construct handles all replication configuration:

export interface DmsBaseProps extends IProps {
    vpcTag: string;
    subnetsIds?: string[];
    securityGroup?: ec2.ISecurityGroup;
    s3Base: S3Base;
    databaseName: string;
    port?: number;
    secretsManagerSecretId: string;
    minCapacityUnits?: number;
    maxCapacityUnits?: number;
    multiAz?: boolean;
    dmsVpcRole?: iam.Role;
    dmsCloudWatchLogsRole?: iam.Role;
    dmsSecretsAccessRole?: iam.Role;
}

export class DmsBase extends Construct {
    public readonly sourceEndpoint: dms.CfnEndpoint;
    public readonly targetEndpoint: dms.CfnEndpoint;
    public readonly replicationConfig: dms.CfnReplicationConfig;

    constructor(scope: Construct, id: string, props: DmsBaseProps) {
        super(scope, id);

        // Lookup VPC
        const vpc = ec2.Vpc.fromLookup(this, 'Vpc', {
            tags: { Name: props.vpcTag }
        });

        // Create or use provided IAM roles
        const vpcRole = props.dmsVpcRole ?? this.createVpcRole(id);
        const secretsAccessRole = props.dmsSecretsAccessRole ?? 
            this.createSecretsAccessRole(id, props.envName);

        // Create subnet group
        const subnetGroup = this.createSubnetGroup(vpc, props);

        // Create security group if not provided
        const securityGroup = props.securityGroup ?? 
            this.createSecurityGroup(vpc, id);

        // Create source endpoint (PostgreSQL)
        this.sourceEndpoint = new dms.CfnEndpoint(this, `${id}-pg-source`, {
            endpointIdentifier: `${props.replicationConfigIdentifier}-pg-source`,
            endpointType: 'source',
            engineName: 'postgres',
            databaseName: props.databaseName,
            postgreSqlSettings: {
                secretsManagerSecretId: props.secretsManagerSecretId,
                secretsManagerAccessRoleArn: secretsAccessRole.roleArn,
            },
            extraConnectionAttributes: [
                'pluginName=pgoutput',
                'captureDdls=false',
                'heartbeatEnable=true',
                'heartbeatFrequency=1',
            ].join(';'),
            sslMode: 'require',
        });

        // Create target endpoint (S3)
        this.targetEndpoint = this.createS3TargetEndpoint(props);

        // Create replication config (serverless)
        this.replicationConfig = this.createReplicationConfig(props);
    }

    private createS3TargetEndpoint(props: DmsBaseProps): dms.CfnEndpoint {
        return new dms.CfnEndpoint(this, 'S3Target', {
            endpointIdentifier: `${props.replicationConfigIdentifier}-s3-target`,
            endpointType: 'target',
            engineName: 's3',
            s3Settings: {
                serviceAccessRoleArn: props.s3Base.iamRoles[0].roleArn,
                bucketName: props.s3Base.bucket.bucketName,
                bucketFolder: 'rds-cdc',
                dataFormat: 'parquet',
                compressionType: 'gzip',
                parquetVersion: 'parquet-2-0',
                includeOpForFullLoad: true,
                cdcInsertsAndUpdates: true,
                timestampColumnName: '__dms_ts',
                datePartitionEnabled: false,
                // Unified file structure for easier querying
                cdcPath: 'public',
            },
        });
    }

    private createReplicationConfig(props: DmsBaseProps): dms.CfnReplicationConfig {
        return new dms.CfnReplicationConfig(this, 'ReplicationConfig', {
            replicationConfigIdentifier: props.replicationConfigIdentifier,
            replicationType: 'full-load-and-cdc',
            sourceEndpointArn: this.sourceEndpoint.ref,
            targetEndpointArn: this.targetEndpoint.ref,
            computeConfig: {
                maxCapacityUnits: props.maxCapacityUnits ?? 4,
                minCapacityUnits: props.minCapacityUnits ?? 2,
                multiAz: props.multiAz ?? true,
                replicationSubnetGroupId: this.subnetGroup.ref,
                vpcSecurityGroupIds: [this.securityGroup.securityGroupId],
            },
            replicationSettings: {
                Logging: {
                    EnableLogging: true,
                    LogComponents: [
                        { Id: 'SOURCE_CAPTURE', Severity: 'LOGGER_SEVERITY_INFO' },
                        { Id: 'TARGET_APPLY', Severity: 'LOGGER_SEVERITY_INFO' },
                    ],
                },
            },
            tableMappings: this.createTableMappings(props.selectedTables),
        });
    }

    private createTableMappings(tables: string[]): string {
        const rules = tables.map(tableName => ({
            'rule-type': 'selection',
            'rule-id': tableName,
            'rule-name': tableName,
            'object-locator': {
                'schema-name': 'public',
                'table-name': tableName,
            },
            'rule-action': 'include',
        }));

        return JSON.stringify({ rules });
    }
}

Step 3: Glue Catalog Base

The Glue catalog provides metadata management and schema discovery:

export interface GlueBaseCatalogProps {
    envName: string;
    projectName: string;
    s3DataBucket: s3.Bucket;
    cdcPath: string;
    catalogSuffix: string;
    useGlueCrawler?: boolean;
    crawlerSchedule?: string;
}

export abstract class GlueBaseCatalog extends Construct {
    public readonly database: glue.CfnDatabase;
    public readonly crawler?: glue.CfnCrawler;
    protected readonly databaseName: string;

    constructor(scope: Construct, id: string, props: GlueBaseCatalogProps) {
        super(scope, id);

        this.databaseName = `${props.projectName}_${props.envName}_${props.catalogSuffix}`;

        // Create Glue Database
        this.database = new glue.CfnDatabase(this, 'Database', {
            catalogId: Stack.of(this).account,
            databaseInput: {
                name: this.databaseName,
                description: this.getDatabaseDescription(),
            },
        });

        // Create crawler for automatic schema discovery
        if (props.useGlueCrawler) {
            this.crawler = this.createCrawler(props);
        }
    }

    private createCrawler(props: GlueBaseCatalogProps): glue.CfnCrawler {
        const crawlerRole = new iam.Role(this, 'CrawlerRole', {
            assumedBy: new iam.ServicePrincipal('glue.amazonaws.com'),
            managedPolicies: [
                iam.ManagedPolicy.fromAwsManagedPolicyName('service-role/AWSGlueServiceRole')
            ],
            inlinePolicies: {
                S3Access: new iam.PolicyDocument({
                    statements: [
                        new iam.PolicyStatement({
                            effect: iam.Effect.ALLOW,
                            actions: [
                                's3:GetObject',
                                's3:ListBucket',
                                's3:GetBucketLocation'
                            ],
                            resources: [
                                props.s3DataBucket.bucketArn,
                                `${props.s3DataBucket.bucketArn}/*`
                            ]
                        })
                    ]
                })
            }
        });

        const s3Target = `s3://${props.s3DataBucket.bucketName}/rds-cdc/public/`;

        return new glue.CfnCrawler(this, 'Crawler', {
            name: `${props.projectName}-${props.envName}-${props.catalogSuffix}-crawler`,
            role: crawlerRole.roleArn,
            databaseName: this.databaseName,
            targets: {
                s3Targets: [{
                    path: s3Target,
                    exclusions: ['**/_tmp/**', '**/.tmp/**']
                }]
            },
            configuration: JSON.stringify({
                Version: 1.0,
                CrawlerOutput: {
                    Partitions: { AddOrUpdateBehavior: 'InheritFromTable' },
                    Tables: { AddOrUpdateBehavior: 'MergeNewColumns' }
                },
                Grouping: {
                    TableGroupingPolicy: 'CombineCompatibleSchemas',
                    TableLevelConfiguration: 4
                }
            }),
            schemaChangePolicy: {
                updateBehavior: 'UPDATE_IN_DATABASE',
                deleteBehavior: 'LOG'
            },
            ...(props.crawlerSchedule && {
                schedule: {
                    scheduleExpression: props.crawlerSchedule
                }
            })
        });
    }

    protected abstract getDatabaseDescription(): string;
    protected abstract getTableList(): string[];

    public getDatabaseName(): string {
        return this.databaseName;
    }
}

Step 4: Athena Workgroup

Athena provides SQL query capabilities:

export interface AthenaBaseProps {
    envName: string;
    projectName: string;
    glueDatabase: string;
    s3DataBucket: s3.Bucket;
    workgroupSuffix: string;
}

export abstract class AthenaBase extends Construct {
    public readonly workGroup: athena.CfnWorkGroup;
    public readonly queryResultsBucket: s3.Bucket;

    constructor(scope: Construct, id: string, props: AthenaBaseProps) {
        super(scope, id);

        const workgroupName = `${props.projectName}-${props.envName}-${props.workgroupSuffix}`;

        // Create S3 bucket for query results
        this.queryResultsBucket = new s3.Bucket(this, 'QueryResultsBucket', {
            bucketName: `${props.projectName}-athena-${props.workgroupSuffix}-results-${props.envName}`,
            removalPolicy: RemovalPolicy.DESTROY,
            autoDeleteObjects: true,
            blockPublicAccess: BlockPublicAccess.BLOCK_ALL,
            versioned: true,
            lifecycleRules: [{
                id: 'DeleteOldQueryResults',
                enabled: true,
                expiration: Duration.days(30),
            }],
        });

        // Create Athena WorkGroup
        this.workGroup = new athena.CfnWorkGroup(this, 'WorkGroup', {
            name: workgroupName,
            description: this.getWorkGroupDescription(),
            state: 'ENABLED',
            workGroupConfiguration: {
                resultConfiguration: {
                    outputLocation: `s3://${this.queryResultsBucket.bucketName}/query-results/`,
                    encryptionConfiguration: {
                        encryptionOption: 'SSE_S3',
                    },
                },
                enforceWorkGroupConfiguration: true,
                bytesScannedCutoffPerQuery: 1000000000, // 1GB limit
            },
        });
    }

    protected abstract getWorkGroupDescription(): string;

    public createAthenaUserRole(roleName: string): iam.Role {
        return new iam.Role(this, `AthenaUserRole-${roleName}`, {
            roleName: `${roleName}-athena-${this.props.workgroupSuffix}-${this.props.envName}`,
            assumedBy: new iam.ServicePrincipal('athena.amazonaws.com'),
            description: `Role for Athena users to query data`,
            inlinePolicies: {
                AthenaQueryPolicy: new iam.PolicyDocument({
                    statements: [
                        new iam.PolicyStatement({
                            effect: iam.Effect.ALLOW,
                            actions: [
                                'athena:StartQueryExecution',
                                'athena:GetQueryExecution',
                                'athena:GetQueryResults',
                            ],
                            resources: [`arn:aws:athena:*:*:workgroup/${this.workGroup.name}`],
                        }),
                        new iam.PolicyStatement({
                            effect: iam.Effect.ALLOW,
                            actions: [
                                'glue:GetDatabase',
                                'glue:GetTable',
                                'glue:GetPartitions',
                            ],
                            resources: ['*'],
                        }),
                        new iam.PolicyStatement({
                            effect: iam.Effect.ALLOW,
                            actions: ['s3:GetObject', 's3:ListBucket'],
                            resources: [
                                this.props.s3DataBucket.bucketArn,
                                `${this.props.s3DataBucket.bucketArn}/*`,
                            ],
                        }),
                    ],
                }),
            },
        });
    }
}

Step 5: Putting It All Together

Here's how to use these constructs in your stack:

export class DataLakeStack extends Stack {
    constructor(scope: Construct, id: string, props?: StackProps) {
        super(scope, id, props);

        // 1. Create required IAM roles (DMS requires specific names)
        const dmsVpcRole = new iam.Role(this, 'DmsVpcRole', {
            roleName: 'dms-vpc-role', // Required exact name
            assumedBy: new iam.CompositePrincipal(
                new iam.ServicePrincipal('dms.amazonaws.com'),
                new iam.ServicePrincipal('dms.us-east-1.amazonaws.com')
            ),
            managedPolicies: [
                iam.ManagedPolicy.fromAwsManagedPolicyName('service-role/AmazonDMSVPCManagementRole')
            ],
        });

        const dmsCloudWatchLogsRole = new iam.Role(this, 'DmsCloudWatchLogsRole', {
            roleName: 'dms-cloudwatch-logs-role', // Required exact name
            assumedBy: new iam.ServicePrincipal('dms.amazonaws.com'),
            managedPolicies: [
                iam.ManagedPolicy.fromAwsManagedPolicyName('service-role/AmazonDMSCloudWatchLogsRole')
            ],
        });

        const dmsSecretsAccessRole = new iam.Role(this, 'DmsSecretsAccessRole', {
            roleName: `dms-secrets-access-role-production`,
            assumedBy: new iam.CompositePrincipal(
                new iam.ServicePrincipal('dms.amazonaws.com'),
                new iam.ServicePrincipal('dms.us-east-1.amazonaws.com')
            ),
            inlinePolicies: {
                SecretsRead: new iam.PolicyDocument({
                    statements: [
                        new iam.PolicyStatement({
                            actions: ['secretsmanager:GetSecretValue'],
                            resources: ['*'],
                        }),
                    ],
                }),
            },
        });

        // 2. Create S3 data lake
        const s3Base = new S3Base(this, 'S3DmsBase', {
            bucketName: `my-project-production-dms-data-lake`,
            iamRolesProps: [{
                assumedBy: new iam.ServicePrincipal('dms.amazonaws.com'),
                description: 'Role for DMS to write to S3',
            }]
        });

        // 3. Create Glue Catalog
        const glueCatalog = new GlueSourceCatalog(this, 'GlueCatalog', {
            envName: 'production',
            projectName: 'my-project',
            s3DataBucket: s3Base.bucket,
            cdcPath: 'source-database/',
            useAutomaticSchemaDiscovery: true,
        });

        // 4. Create DMS replication
        new SourceDatabaseDms(this, 'SourceDatabaseDms', {
            envName: 'production',
            projectName: 'my-project',
            s3Base,
            databaseName: 'mydb',
            secretsManagerSecretId: 'my-db-secret',
            replicationConfigIdentifier: 'source-database-production',
            vpcTag: 'main-vpc',
            dmsVpcRole,
            dmsCloudWatchLogsRole,
            dmsSecretsAccessRole,
            glueCatalog,
        });

        // 5. Create Athena workgroup
        const athenaSource = new AthenaSource(this, 'AthenaSource', {
            envName: 'production',
            projectName: 'my-project',
            glueDatabase: glueCatalog.getDatabaseName(),
            s3DataBucket: s3Base.bucket,
        });

        // 6. Create Athena user role
        athenaSource.createAthenaUserRole('DataAnalyst');
    }
}

Configuration Details

Table Selection

Define which tables to replicate in a constants file:

export const selectedTables = [
    'users',
    'orders',
    'products',
    'customers',
    // ... more tables
];

export const cdcPath = '<instance-name>-database/';

PostgreSQL Configuration

Ensure your PostgreSQL database has these settings:

-- Enable logical replication
ALTER SYSTEM SET wal_level = logical;
ALTER SYSTEM SET max_replication_slots = 10;
ALTER SYSTEM SET max_wal_senders = 10;

-- Restart PostgreSQL after making these changes

That setup needs to be done through AWS Parameter Group if you're AWS RDS.

Secrets Manager Format

Your database secret should have this structure:

{
  "username": "dbuser",
  "password": "securepassword",
  "host": "mydb.region.rds.amazonaws.com",
  "port": 5432,
  "dbname": "mydb"
}

Key Features & Best Practices

1. Serverless DMS

We use DMS Serverless with auto-scaling:

Min capacity: 2 DCU
Max capacity: 4 DCU
Multi-AZ for high availability

2. Parquet Format

DMS writes data in Parquet format with:

GZIP compression for storage efficiency
Operation column (Op) for CDC tracking (I=Insert, U=Update, D=Delete)
Timestamp column (__dms_ts) for change tracking

3. S3 Structure

Data is organized as:

s3://bucket/rds-cdc/public/tablename/
  ├── LOAD00000001.parquet
  ├── 20231001-12345.parquet
  └── 20231002-67890.parquet

I recommend you organize like that, so you can use the auto-discovery of Glue Crawlers.

4. Glue Crawler

The crawler runs automatically:

Production: Daily at 2 AM (you can change this according your case)
Automatically discovers new tables and schema changes
Updates the Glue Data Catalog

5. Security

Multiple layers of security:

VPC endpoints for private connectivity
IAM roles with least privilege
S3 bucket encryption (SSE-S3)
SSL/TLS for database connections
Secrets Manager for credentials

Querying Data with Athena

Once deployed, on Athena, query your data using SQL:

-- View recent changes
SELECT *
FROM my_project_production_source_db.users
WHERE __dms_ts > current_timestamp - interval '1' day
ORDER BY __dms_ts DESC;

-- Get current state (latest record for each ID)
WITH latest_records AS (
  SELECT *, 
         ROW_NUMBER() OVER (PARTITION BY id ORDER BY __dms_ts DESC) as rn
  FROM my_project_production_source_db.users
)
SELECT *
FROM latest_records
WHERE rn = 1 AND Op != 'D'; -- Exclude deleted records

Cost Optimization

DMS Serverless: Only pay for active replication time
S3 Lifecycle: Automatically delete query results after 30 days
Parquet + Gzip: Reduces storage by ~70% compared to CSV
Athena Query Limits: 1GB scan limit per query prevents runaway costs

Monitoring

CloudWatch provides comprehensive monitoring:

DMS replication lag
S3 bucket metrics
Glue crawler runs
Athena query metrics

Create alarms for:

new cloudwatch.Alarm(this, 'DmsLatencyAlarm', {
    metric: replicationConfig.metricReplicationLatency(),
    threshold: 300, // 5 minutes
    evaluationPeriods: 2,
});

Deployment

Deploy your stack:

# Bootstrap CDK (first time only)
cdk bootstrap

# Synthesize CloudFormation template
cdk synth

# Deploy
cdk deploy DataLakeStack (you can change)

# After deployment, start the DMS replication in AWS Console
# or use AWS CLI:
aws dms start-replication \
    --replication-config-arn <arn> \
    --start-replication-type start-replication

Common Issues & Solutions

Issue 1: DMS Can't Connect to PostgreSQL

Solution: Ensure RDS security group allows inbound from DMS security group on port 5432.

Issue 2: Tables Not Appearing in Glue

Solution: Run the Glue crawler manually first, then check CloudWatch logs.

Issue 3: Athena Queries Return No Results

Solution: Verify S3 path in Glue table matches actual data location.

Issue 4: DMS Replication Fails

Solution: Check that PostgreSQL has wal_level=logical and sufficient replication slots.

Conclusion

This architecture provides a robust, scalable solution for replicating PostgreSQL data to a data lake. The combination of DMS, Glue, and Athena creates a powerful analytics platform that:

Scales automatically with your data
Provides near real-time data availability
Supports complex analytical queries
Maintains cost efficiency
Offers production-grade reliability

The modular CDK design makes it easy to extend with additional databases, customize table selection, or add data transformations.

Resources

This architecture has been battle-tested in production environments handling millions of records daily. Feel free to adapt it to your specific needs!

Navigating the Nuances of Communication in the Corporate World

André Paris — Sun, 12 Nov 2023 18:44:29 +0000

In the dynamic environment of a modern company, the power of communication extends far beyond mere exchanges of information. It is the cornerstone upon which relationships are built, conflicts are resolved, and a harmonious workplace culture is cultivated. However, this landscape is often marred by pitfalls such as injustice and gossip, which can stem from inadequate or one-sided communication.

The Perils of Partial Listening

One significant challenge in corporate communication is the tendency to listen to only one side of the story. This selective hearing can lead to skewed perceptions and unjust decisions. It’s crucial for management and team members alike to adopt an approach of holistic listening. This means not just hearing the words spoken but understanding the context, the speaker’s perspective, and the underlying emotions. When only one side of a story is heard, it can foster an environment of mistrust and speculation, often leading to the spread of gossip.

Context: The Make or Break of Communication

Another key aspect is the context in which information is shared and received. Misinterpretations often occur when statements are taken out of context, leading to a distorted view of the situation. This distortion can feed into the rumor mill, creating a breeding ground for workplace gossip and potential injustice. To mitigate this, it’s imperative that all communicators in a corporate setting strive for clarity and ensure their message is conveyed in its complete form, leaving little room for misinterpretation.

Building a Culture of Comprehensive Communication

To foster an environment where communication is not only clear but also fair and inclusive, companies should invest in training their employees in effective communication strategies. This includes active listening, conflict resolution, and understanding the impact of non-verbal cues. Additionally, creating a feedback culture where employees feel comfortable sharing their thoughts and concerns can help in getting a complete picture of any situation, thereby reducing one-sided narratives.

Leadership’s Role in Shaping Communication

Leadership plays a pivotal role in setting the standard for effective communication. They must lead by example, showing their commitment to hearing all sides and understanding the full context before making decisions. By doing so, they set a precedent for the rest of the company, cultivating a culture where justice and fairness are at the forefront.

Conclusion: A Call for Mindful Communication

In conclusion, the importance of effective communication in the corporate world cannot be overstated. It is not just about conveying information but doing so in a way that is fair, complete, and contextual. By addressing the challenges of one-sided listening and decontextualization, companies can create a more just, respectful, and productive workplace environment. The goal is to achieve a culture where every voice is heard, every context is considered, and every message is communicated with clarity and intent.

Revolutionizing IT in Startups: My Journey to Building a High-Performance Tech Team

André Paris — Thu, 09 Nov 2023 18:48:13 +0000

Welcome to my professional narrative. I am André Paris, an IT connoisseur with a robust track record spanning over 12 years. In the realm of technology, I've donned multiple hats – from a fervent software engineer who relishes the intricacies of code to a strategic leader who steers teams towards excellence. My latest venture involved pioneering the IT department for an up-and-coming Mexican startup, transforming it from a fledgling operation into a paragon of success. I'm excited to unfold this chapter of my journey and share the insights garnered along the way.

Introduction

In this article, I want to share some insights about my journey in structuring a lean and successful IT team at a startup. I’ll discuss what helped me along the way, including the challenges I faced, because as we know, the path is not always easy and often requires tough decisions. My aim is to empower readers by showing that they are not alone in navigating these difficult processes and that there are numerous methods and methodologies to overcome obstacles and achieve their goals.

Firstly, it’s important to recognize that startups are typically fast-paced and subject to frequent changes. They are focused on achieving results to reach the break-even point and thereby justify the investment to ultimately realize actual profits. The experience is akin to running a marathon in the midst of a hurricane—a long and turbulent journey, often with little time to organize internally, focusing only on the essentials.

This chaotic environment is the reason many startups don’t finish the race. Organizing internally should also be seen as crucial; without it, it's impossible to fully understand what is being left behind. And when that clarity is missing, we often overlook what’s truly important for survival.

How I Organized a Successful IT Team

The Beginning

When I joined the startup, the team consisted of three developers, one product manager, and me, all working remotely. Initially, my role was to be the Tech Lead. However, in a startup environment, the job description (JD) is just a starting point, and roles often expand. At that time, we had an external company assisting with scrum organization, led by a supposed scrum master. This person would only show up for the daily stand-up, bluntly asking the developers about their tasks. Post-meeting, everyone would return to their work in silos, with no further communication.

My first task was to turn these individual developers into a cohesive team. To do this, I established an end-of-day alignment meeting with just the developers and myself. My goal wasn’t to add another daily stand-up but to foster better relationships and camaraderie among the team. In a small company like a startup, forming strong bonds is crucial; with few people handling numerous tasks and everyone always on the run, a supportive and empathetic environment can alleviate pressure and boost productivity.

This approach significantly improved the team’s mood, productivity, and engagement within weeks. We had six months to deliver a product sold to investors, and with only three months left, over 50% of the work remained. This small change made it possible to deliver the project on time with a true team.

Growing the Team Requires Establishing Processes

My next step was to professionalize as a scrum master and delve into IT governance. I began by properly implementing scrum within the team, ensuring that all scrum ceremonies were respected. Initially, the product manager would score the stories themselves, which seems unbelievable but was the case. This PM was resistant to establishing retrospectives, so I had to postpone that ceremony. Nevertheless, the developers became more engaged, seeing that their department was heading in a positive direction.

Then I started studying IT governance, exploring methodologies from the traditional COBIT and ITIL to IT4IT, a new framework designed for startups. I created excellent material and presented it to my boss. Unfortunately, he couldn't envision it within our structure, as his focus was solely on results, without consideration for these unfamiliar topics. This didn’t completely discourage me. I implemented much of this framework. Now we're talking about a startup with an engaged team, a scrum process that's almost authentic (no retrospectives 😔), and a structured governance process.

As you can see in the image above, implementing this requires analyzing and structuring many aspects of IT, from strategic portfolio management to incident management. To make this possible, I had to become a DevOps practitioner. I started by addressing the most critical aspect of our daily operations, the transition from Requirement to Deploy. For this, I set up CI/CD pipelines, then organized the cloud infrastructure, segregated environments (dev, stage, and prod) into separate accounts, established access levels, and created a process for bug resolution (you can imagine that a system built by four people over six months consisting of two web portals and a mobile app wouldn’t be perfect). Unfortunately, I haven't yet succeeded in implementing a testing culture; currently, all testing is done by the PM and COO, who aren't ready to delegate this responsibility.

The next critical step for business continuity and success was establishing the Detect to Correct process. Initially, we created a system where all errors were reported via Slack. Additionally, we had manual tests to fix errors before production deployment (thanks to the Requirement to Deploy process). However, with a freshly developed system, many errors began to surface and were reported by the operations team through Slack, which meant many were not resolved, leaving the development team disheartened by having to deal with random bugs, which often were not bugs, but rather users unfamiliar with the system.

To address this, I first created a dedicated Slack channel for centralizing these issues. Then, I established a ticketing system, and after much research, I chose Freshdesk for its excellent cost-benefit ratio. Getting the operations team to adopt it wasn't easy or immediate; it required training them, but it worked excellently. During this time, I took on the responsibility of being the first point of contact for tickets, to structure the area further. This, again, led to tremendous engagement from the developers, who became fully focused on their tasks and thought processes in architecture. Eventually, I hired two people to handle these issues.

Moreover, we implemented Datadog. With this tool, we were able to greatly enhance bug alerting, manage system latency, and develop a log system that incredibly sped up error detection, directly impacting resolution time and highlighting the importance of log systems. Thus, we resolved all the steps in this box of the IT4IT framework.

The next step was to structure the Strategy to Portfolio. Unfortunately, we did not have a very organized product team, which meant demand management was always an issue. Another important consideration is that, in a startup, the product constantly changes, and the priority of items seeks to meet customer demand. As IT, what we can do is mitigate. I introduced a bureaucracy ensuring that the product team always had their stories defined and scored (using Scrum's Fibonacci sequence). At first, we faced many problems because the product team didn't understand this; they would always change the stories or ask for activities without refinement by the team.

To clearly define demand management and refine the portfolio strategy, it was necessary to progress with the Request to Fulfill. Besides the deployment points and product measurements involved here, this also relates to the process and team management since measuring their capacity is extremely important, even for a startup. In our case, I established this with the Fibonacci scoring of Scrum. Based on this, it was possible to have an average of how many points each team delivered per sprint and create a product roadmap for the first stage.

Having consolidated this, we successfully achieved international certifications, ISO 27001 and SOC2, which I was able to manage alone, from hiring the audit supplier to implementing and making the necessary corrections. This was only possible because we had an organized IT department.

It’s also important to note that culture is something that should be valued. Throughout this process, I conducted 1:1s with the entire team, always asking how to improve IT and myself. Without the right feedback, we might never have achieved our goals.

It's essential not to forget that in a startup, we must all work together, as a team, rowing towards the same goal. These types of 1:1s should not only be done with your own team but also with all areas to improve communication between different teams. Yes, it's a difficult process because it's part of educating towards better processes, but after significant effort, I want you to know that it can be achieved.

Doctor Who? No, Doctor You via AWS Chime: A Telemedicine Adventure!

André Paris — Wed, 08 Nov 2023 22:54:03 +0000

🚀 Intro for Code Wizards

Hey Dev Wizards! Telemedicine is not just a buzzword; it's like the Gandalf of healthcare, conjuring up tech-magic to deliver health wizardry to muggles wherever they may be. When COVID-19 decided to crash our planet, telemedicine strapped on a jetpack and zoomed to the rescue. Now, with our AI pals, IoT gadgets, and mobile doohickeys, it's soaring to new heights, ready to flip healthcare on its head.

Join me on a pixel-packed adventure with AWS Chime, my sidekick in forging a telemedicine experience smoother than a buttered byte. This tale is all about my digital escapades, the dragons (challenges) I faced, and the treasure trove of solutions we unearthed. With AWS Chime's Swiss Army knife of features, we built a telemedicine platform so user-friendly, even your grandma can navigate it with ease.

Dive into this saga as we explore how AWS Chime is making telehealth cooler than the other side of the pillow, and shaping the face of future healthcare apps.

🛠️ SaaS Sorcery with AWS Chime

First Up: What's AWS Chime Anyway?

AWS Chime is like the Swiss Army knife for online meet-ups, video chitchats, and digital pow-wows. Born in the cloud realm of Amazon in 2017, it's a knight in shining armor for those seeking a secure and cloud-strong fortress against the dark arts of unreliable communication tools.

With Chime, you can keep your squad synced across the digital multiverse, offering a one-stop spellbook for all your conferencing calls. It's got HD vision, crystal-clear audio, screen-sharing charms, and a toolkit for devs to weave in some powerful integrations.

Chime plays nice with all your gizmos and gadgets, keeping you connected so seamlessly, you'll think it's telepathy. Its fortress is built in the cloud, making it a rockstar for telemedicine, handling sensitive chit-chats with the utmost secrecy and care.

Where's the 'Launch Stack' Spellbook?

In the mystical scrolls below, I'll reveal the path to conjure AWS Chime with ease. But let me whisper a secret: tweak the AWS-provided CloudFormation to ditch the bloatware for your app's epic quest.

Find the 'Launch Stack' spellbook right here, complete with a step-by-step enchantment to deploy the cloud magic.

Follow the ancient AWS incantations, and voilà, you'll summon a React web page. It's your canvas - paint your app's personality all over it.

Stirring the Pot

Here's a spicy take: embedding AWS Chime is a stellar move for scrappy startups or cash-conscious enterprises. Peek at the treasure map to see the gold it'll cost. I've unleashed Chime in a healthcare startup's telemedicine cauldron and pitted it against a bespoke potion. Guess what? No latency trolls under this bridge, and the freedom to craft the user quest was pure alchemy!