WalFlow.Stores.S3 0.1.0-alpha.alpha.20260708090623

WalFlow.Stores.S3

AWS S3-based store for WalFlow CDC processor. Persists replication state as JSON files in Amazon S3 or S3-compatible storage services.

Features

• Flexible Authentication: Supports explicit credentials, AWS profiles, IAM roles, and environment variables
• HttpClient Injection: Optional custom HttpClient support for proxy/retry policies
• S3-Compatible Storage: Works with MinIO, LocalStack, and other S3-compatible services
• Server-Side Encryption: Optional AES256 encryption for providers that support SSE-S3
• Singleton Pattern: Thread-safe operations with semaphore locking
• Error Resilience: Handles missing files and JSON corruption gracefully

Installation

dotnet add package WalFlow.Stores.S3

Configuration

appsettings.json

Using IAM Role / Default Credentials

{
  "S3Store": {
    "BucketName": "my-cdc-bucket",
    "KeyPrefix": "walflow/state/",
    "Region": "us-east-1"
  }
}

Using Explicit Credentials

{
  "S3Store": {
    "BucketName": "my-cdc-bucket",
    "KeyPrefix": "walflow/state/",
    "Region": "us-west-2",
    "AccessKeyId": "AKIAIOSFODNN7EXAMPLE",
    "SecretAccessKey": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
  }
}

Using AWS Profile

{
  "S3Store": {
    "BucketName": "my-cdc-bucket",
    "KeyPrefix": "walflow/state/",
    "ProfileName": "my-aws-profile"
  }
}

Using S3-Compatible Service (MinIO, LocalStack)

{
  "S3Store": {
    "BucketName": "my-cdc-bucket",
    "KeyPrefix": "walflow/state/",
    "ServiceUrl": "http://localhost:9000",
    "ForcePathStyle": true,
    "AccessKeyId": "minioadmin",
    "SecretAccessKey": "minioadmin"
  }
}

Options

Using Cloudflare R2 (S3-Compatible)

{
  "S3Store": {
    "BucketName": "my-r2-bucket",
    "KeyPrefix": "walflow/state/",
    "ServiceUrl": "https://<accountid>.r2.cloudflarestorage.com",
    "AccessKeyId": "<r2-access-key-id>",
    "SecretAccessKey": "<r2-secret-access-key>",
    "EnableServerSideEncryption": false
  }
}

For providers/endpoints that reject x-amz-server-side-encryption, set EnableServerSideEncryption to false.

Usage

Basic Registration (Default Credentials)

services.Configure<S3StoreOptions>(
    configuration.GetSection(S3StoreOptions.SectionName));

services.AddSingleton<IReplicationStore, S3Store>();

With IHttpClientFactory (Recommended)

// Configure named HttpClient with policies
services.AddHttpClient("S3Client")
    .ConfigurePrimaryHttpMessageHandler(() => new HttpClientHandler
    {
        MaxConnectionsPerServer = 10
    })
    .AddPolicyHandler(GetRetryPolicy());

services.AddSingleton<IReplicationStore>(sp =>
{
    var options = sp.GetRequiredService<IOptions<S3StoreOptions>>();
    var logger = sp.GetRequiredService<ILogger<S3Store>>();
    var httpClientFactory = sp.GetRequiredService<IHttpClientFactory>();
    
    return S3Store.CreateWithHttpClientFactory(options, logger, httpClientFactory, "S3Client");
});

With Custom HttpClient Instance

services.AddSingleton<IReplicationStore>(sp =>
{
    var options = sp.GetRequiredService<IOptions<S3StoreOptions>>();
    var logger = sp.GetRequiredService<ILogger<S3Store>>();
    var httpClient = new HttpClient(new HttpClientHandler
    {
        MaxConnectionsPerServer = 10
    });
    
    return S3Store.CreateWithHttpClient(options, logger, httpClient);
});

With External S3 Client

services.AddSingleton<IReplicationStore>(sp =>
{
    var options = sp.GetRequiredService<IOptions<S3StoreOptions>>();
    var logger = sp.GetRequiredService<ILogger<S3Store>>();
    var s3Client = sp.GetRequiredService<IAmazonS3>();
    
    return S3Store.CreateWithS3Client(options, logger, s3Client);
});

Manual Configuration

services.AddSingleton<IReplicationStore>(sp =>
{
    var options = Options.Create(new S3StoreOptions
    {
        BucketName = "my-cdc-bucket",
        Region = "us-east-1",
        KeyPrefix = "production/state/"
    });
    
    var logger = sp.GetRequiredService<ILogger<S3Store>>();
    return new S3Store(options, logger);
});

S3 Object Structure

State files are stored with the following structure:

s3://my-cdc-bucket/
  walflow/state/
    my_slot_name.json
    another_slot.json

Each file contains the full ReplicationState as JSON:

{
  "SlotName": "my_slot_name",
  "LastProcessedLsn": 123456789,
  "LastUpdated": "2024-01-01T12:00:00Z",
  "MessagesProcessed": 50000,
  "SnapshotCompleted": true,
  "CompletedSnapshotTables": ["public.users", "public.orders"],
  "Snapshot": { ... },
  "IncrementalPkCursor": { ... }
}

Authentication Priority

The S3Store uses the following credential priority:

1. Explicit Credentials: AccessKeyId and SecretAccessKey in options
2. AWS Profile: ProfileName in options
3. Default Chain: Environment variables → IAM role → default profile

AWS Permissions Required

The S3 bucket must grant the following permissions:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:PutObject"
      ],
      "Resource": "arn:aws:s3:::my-cdc-bucket/walflow/state/*"
    }
  ]
}

Error Handling

• Bucket Not Found: Throws AmazonS3Exception on first access
• Missing State File: Returns null (normal for first run)
• JSON Corruption: Logs error and returns null
• Network Failures: Throws exception (consider retry policies)

Thread Safety

The S3Store is thread-safe:

• Write operations use SemaphoreSlim to prevent concurrent writes to same slot
• Read operations are lock-free
• Multiple instances can safely access the same bucket

Performance Considerations

• Latency: S3 operations have network latency (~50-200ms per operation)
• Rate Limits: S3 has default rate limits (3,500 PUT/s, 5,500 GET/s per prefix)
• Costs: Charged per request and storage
• Optimization: Consider increasing StateSaveInterval to reduce S3 requests

Using with MinIO

var options = Options.Create(new S3StoreOptions
{
    BucketName = "walflow",
    ServiceUrl = "http://localhost:9000",
    ForcePathStyle = true,
    AccessKeyId = "minioadmin",
    SecretAccessKey = "minioadmin"
});

var store = new S3Store(options, logger);

Monitoring

Use AWS CLI to monitor state:

# List all state files
aws s3 ls s3://my-cdc-bucket/walflow/state/

# Download a specific state file
aws s3 cp s3://my-cdc-bucket/walflow/state/my_slot.json ./state.json

# View state content
aws s3 cp s3://my-cdc-bucket/walflow/state/my_slot.json - | jq .

License

See the LICENSE file in the repository root.