Aggregation Persistence with Change Feed

Aggregation Persistence with Azure Cosmos DB Change Feed

Introduction: The Power of Event-Driven Aggregations

In modern distributed systems, the way we handle data has shifted from simple request-response patterns to event-driven architectures. When working with Azure Cosmos DB, one of the most powerful tools at your disposal is the Change Feed. The Change Feed acts as a persistent, ordered log of all modifications made to the documents within a container. Instead of querying for the current state of a database, the Change Feed allows you to react to the evolution of data in real-time.

Aggregation persistence—the process of calculating and storing running totals, averages, or state snapshots as data flows into the system—is a classic use case for this technology. Imagine an e-commerce platform where you need to track the total value of orders placed by a specific user or calculate the inventory level of a product across multiple warehouse regions. If you perform these calculations on every read request, your application will struggle with performance bottlenecks and high Request Unit (RU) costs.

By using the Change Feed to perform aggregations asynchronously, you shift the computational burden away from the user-facing request path. You transform the database from a passive storage bin into an active, event-driven engine. This lesson will guide you through the architectural patterns, implementation strategies, and operational best practices required to build efficient aggregation persistence solutions using the Cosmos DB Change Feed.

Understanding the Change Feed Mechanism

At its core, the Azure Cosmos DB Change Feed is a sorted list of documents in the order in which they were modified. When you enable the Change Feed on a container, you are essentially subscribing to a stream of events. Each event represents a document update, insertion, or deletion. Because the Change Feed is durable, if your processing application goes offline, it can resume exactly where it left off once it restarts.

The Change Feed provides the "what happened" component of your system. To turn this into "what is the current state," you need a processor. In the Azure ecosystem, the most common way to consume this feed is through the Change Feed Processor (CFP) library. This library manages the complex state of your processing progress, handles load balancing if you scale out your compute instances, and ensures that your aggregation logic runs reliably.

Callout: Change Feed vs. Querying Querying your database directly for aggregations (e.g., SELECT SUM(c.price) FROM c WHERE c.userId = '123') requires the database engine to scan potentially thousands of documents every time a user refreshes their dashboard. This is expensive and slow. Using the Change Feed, you calculate the sum incrementally as documents arrive. When a user requests their total, you simply read a single pre-aggregated document. This changes an O(N) operation into an O(1) operation.

Architectural Patterns for Aggregation

Before writing code, you must decide on the granularity of your aggregations. There are two primary patterns for storing these results:

1. The Single-Document Aggregate

In this pattern, you maintain a single document that holds the current state for a specific entity. For example, a document with the ID user-123-summary stores totalSpent, orderCount, and lastOrderDate. Every time the Change Feed triggers for an order belonging to user-123, your processor reads this summary document, updates the values, and saves it back to the database.

2. The Time-Series Windowing

If you need to track metrics over time (e.g., hourly sales), you use windowing. The Change Feed processor identifies the time bucket for an incoming order and updates a document corresponding to that bucket (e.g., sales-2023-10-27-14). This is excellent for telemetry and analytics where you need to see trends rather than just a single running total.

Note: When using the Single-Document Aggregate pattern, remember that you are creating a "hot spot" for that specific document. If a single user places orders at an extremely high frequency, updating the same document repeatedly can lead to partition key contention. Ensure your document design allows for high-frequency writes or consider a buffer layer if the throughput is exceptionally high.

Implementing the Change Feed Processor

To implement aggregation persistence, you need a .NET application (or another supported language) that acts as the listener. Below is a step-by-step approach to setting up a basic processor.

Step 1: Dependencies

You will need the Microsoft.Azure.Cosmos NuGet package. This package contains the Change Feed Processor library, which simplifies the management of lease documents.

Step 2: Configure the Lease Container

The Change Feed Processor needs a place to store "lease" documents. These documents track which instances are processing which parts of the feed and where they are in the timeline. You should create a separate, low-throughput container in your Cosmos DB account specifically for these leases.

Step 3: Define the Processor

The processor requires two main components: a Container instance for the source data and a Container instance for the leases. You will also define a delegate that handles the actual aggregation logic.

// Setting up the Change Feed Processor
Container leaseContainer = database.GetContainer("leases");
Container sourceContainer = database.GetContainer("orders");

ChangeFeedProcessor processor = sourceContainer
    .GetChangeFeedProcessorBuilder("orderAggregationProcessor", HandleChangesAsync)
    .WithInstanceName("ProcessorInstance1")
    .WithLeaseContainer(leaseContainer)
    .Build();

await processor.StartAsync();

Step 4: The Aggregation Logic (The HandleChangesAsync Method)

This is where the magic happens. Your delegate receives a list of changes. You must iterate through these changes, calculate the new state, and update your aggregation document.

async Task HandleChangesAsync(
    IReadOnlyCollection<Order> changes, 
    CancellationToken cancellationToken)
{
    foreach (var order in changes)
    {
        // 1. Retrieve the existing aggregate document
        var aggregateDoc = await GetAggregateDocument(order.UserId);

        // 2. Perform the aggregation
        aggregateDoc.TotalSpent += order.Amount;
        aggregateDoc.OrderCount++;

        // 3. Persist the updated aggregate
        await summaryContainer.UpsertItemAsync(aggregateDoc);
    }
}

Handling Concurrency and Consistency

When multiple instances of your processor are running, or when multiple orders for the same user arrive simultaneously, you run the risk of race conditions. If two instances read the same aggregate document and attempt to update it, the last one to write will overwrite the changes of the first.

Optimistic Concurrency Control (OCC)

Cosmos DB supports ETag-based optimistic concurrency. Every document has an _etag property. When you read a document, you get its current ETag. When you save it back, you send that ETag in the AccessCondition header. If the document has changed since you read it, the write will fail with a 412 Precondition Failed error.

Tip: In your aggregation logic, always use the ItemRequestOptions to specify the IfMatchEtag condition. If the update fails due to a conflict, you should implement a retry loop that re-reads the latest version of the aggregate document, re-applies the calculation, and attempts the write again.

Transactional Batches

If you are updating multiple related documents (e.g., updating a user summary and a global company sales total), you can use the TransactionalBatch feature. This ensures that either both updates succeed or neither does, maintaining total integrity across your aggregated data.

Best Practices for Production

Building a system that runs reliably in production requires more than just functional code. You must account for failures, scaling, and monitoring.

1. Idempotency is Mandatory

The Change Feed guarantees "at-least-once" delivery. This means that under certain conditions, such as a network hiccup or a worker restart, your HandleChangesAsync method might receive the same document twice. Your aggregation logic must be idempotent. Instead of aggregateDoc.TotalSpent += order.Amount, consider maintaining a list of processed OrderIds within the aggregate document to ensure you never count the same order twice.

2. Monitoring the Lease Container

The lease container is the heartbeat of your Change Feed Processor. If the lease container is under-provisioned, the processor will lag, and your aggregates will fall behind. Use Azure Monitor to track the RU consumption of your lease container and set alerts for high latency or throttling.

3. Scaling Your Processors

The Change Feed Processor automatically balances load across multiple instances. If your processing volume increases, you can simply spin up more instances of your application. The library will detect the new instances and redistribute the lease documents accordingly. Do not manually assign work; let the library handle the orchestration.

4. Handling Poison Messages

What happens if a specific document is malformed and causes your HandleChangesAsync logic to throw an exception? The processor will retry the batch repeatedly, potentially stalling the entire feed. Implement a try-catch block within your delegate. If an error is unrecoverable, log the offending document ID to a "dead-letter" container and move on, ensuring the pipeline continues to flow.

Warning: Avoid long-running tasks inside your HandleChangesAsync method. The Change Feed Processor expects this method to complete quickly. If you have complex tasks, such as sending emails or calling external APIs, push those tasks to a secondary queue (like Azure Service Bus or Storage Queues) and return from the processor immediately.

Common Pitfalls and How to Avoid Them

Pitfall 1: Ignoring the Partition Key

When designing your aggregation documents, ensure they are partitioned in a way that aligns with your query patterns. If your aggregation document is partitioned by UserId, and you are constantly updating it, make sure the UserId is also part of your query path. Avoid cross-partition queries for your aggregate lookups, as they are significantly more expensive.

Pitfall 2: Over-provisioning the Feed Container

Many developers make the mistake of setting the throughput on the source container and the aggregation container to be the same. Often, the aggregation container is read-heavy while the source is write-heavy. Use auto-scale throughput on your containers to handle unpredictable traffic spikes without paying for idle capacity.

Pitfall 3: Neglecting "Deleted" Documents

By default, the Change Feed includes inserts and updates. If you need to handle deletions (e.g., if a user cancels an order, you need to subtract that amount from the total), you must explicitly configure the ChangeFeedProcessorOptions to include deletes.

ChangeFeedProcessorOptions options = new ChangeFeedProcessorOptions
{
    StartFromBeginning = true,
    PeekMode = false,
    // Ensure we process deletions to keep aggregates accurate
    // Note: This requires specific configuration on the ChangeFeedProcessorBuilder
};

Step-by-Step Implementation Checklist

1. Analyze Data Flow: Identify the source container and the target aggregate structure.
2. Define Schema: Design the JSON structure for the aggregate document. Include fields for metadata like LastUpdatedTimestamp and ETag.
3. Provision Infrastructure: Create the source, lease, and target containers. Ensure they are in the same region for lowest latency.
4. Develop Consumer: Write the .NET logic using the ChangeFeedProcessorBuilder.
5. Implement Idempotency: Add logic to check if a specific change has already been applied.
6. Add Error Handling: Use try-catch blocks and log failures to a dedicated diagnostic store.
7. Performance Test: Use a load testing tool to simulate high-volume updates and observe how the processor handles lag.
8. Monitor: Set up alerts for "Change Feed Lag," which indicates the time difference between the latest change in the source and the current position of the processor.

Comparison of Aggregation Techniques

To choose the right approach for your specific scenario, consider the trade-offs in the table below:

As shown, the Change Feed approach strikes an excellent balance between cost and performance for most standard aggregation tasks. It is significantly cheaper than running an always-on stream processing engine while providing far lower latency than on-demand queries.

Advanced Topic: Managing Throughput with Feed Lag

One of the most important metrics to watch in a production Change Feed implementation is "Feed Lag." This metric represents the delay between when a document is written to the source container and when it is processed by your function. If your lag starts to grow indefinitely, it means your consumer cannot keep up with the rate of incoming changes.

To mitigate this, you have three primary levers:

1. Scale Out: Add more instances of your processor. The Change Feed Processor will automatically distribute the work.
2. Optimize the Delegate: If your HandleChangesAsync is doing too much work, simplify it. Move non-critical tasks out of the loop.
3. Increase Throughput: If your database is being throttled, you may need to increase the Request Units (RUs) on your source or lease containers.

Callout: The "Lease Container" Secret Many developers view the lease container as a simple implementation detail. In reality, it is your primary tool for monitoring. By inspecting the lease documents, you can see exactly which instance is responsible for which partition and how far behind they are. If you ever need to "reset" the processing, you can delete the leases (with caution) to force the processor to start from the beginning of the feed.

Testing Your Implementation

You cannot rely on unit tests alone for Change Feed implementations. You need integration tests that actually push documents into a test container and verify that the aggregate document is updated correctly.

Use the following strategy for testing:

• Sequential Tests: Push documents one by one and verify the aggregate.
• Burst Tests: Push 100 documents simultaneously and verify the aggregate reflects the total.
• Failure Simulation: Stop your processor mid-stream, wait, and restart it to ensure it resumes correctly without missing or duplicating data.

When writing your test code, use the CosmosClient to clean up the containers between runs. A clean slate is essential for verifying that your StartFromBeginning logic works as expected.

// Example integration test snippet
[Fact]
public async Task Aggregation_ShouldUpdateTotal_WhenOrderIsCreated()
{
    // Arrange
    var order = new Order { Id = Guid.NewGuid().ToString(), UserId = "U1", Amount = 100 };
    await sourceContainer.CreateItemAsync(order);

    // Act
    // Give the processor a moment to react
    await Task.Delay(2000); 

    // Assert
    var aggregate = await summaryContainer.ReadItemAsync<UserSummary>("U1", new PartitionKey("U1"));
    Assert.Equal(100, aggregate.Resource.TotalSpent);
}

Key Takeaways for Success

Implementing aggregation persistence with the Cosmos DB Change Feed is a foundational skill for building scalable, responsive cloud applications. By following these principles, you ensure your system remains performant as it grows.

• Shift from Pull to Push: Stop calculating aggregates on read. Use the Change Feed to maintain pre-calculated states incrementally.
• Design for Idempotency: Because the Change Feed guarantees "at-least-once" delivery, your code must handle duplicate events without corrupting the state.
• Use Optimistic Concurrency: Protect your aggregate documents from race conditions by using ETags and IfMatch headers.
• Monitor Feed Lag: Keep a close eye on the latency between the source update and the aggregate update. If lag increases, scale your processor instances.
• Separate Lease Containers: Always use a dedicated, low-throughput container for lease management to prevent interference with your primary data.
• Handle Poison Messages: Implement robust error handling within your processor to ensure that one bad document doesn't block the entire pipeline.
• Keep Logic Lean: The Change Feed Processor delegate should be fast. Offload heavy lifting or external service calls to background queues to maintain throughput.

By adopting these patterns, you turn your database into a dynamic engine capable of handling complex state management with minimal overhead. As you move forward, continue to refine your partition strategies and monitor your RU usage, as these will be the primary drivers of your operational costs and system performance.