Skip to content
Prague
Improve this page

Incremental Sync

Goal: a client calls with the watermark you handed it last time and gets only the customers whose data changed since, including changes that happened on their orders, plus a new watermark.

Prerequisite: Customer declares CustomerActivityIndex, an IDataCacheGlobalLastUpdateIndex<int> keyed by customer id, and Order reports into it through CustomerId. How that is declared, and what the index stores, is in Global Last-Update Index. The code is in IncrementalSync.cs; every block runs as a test.

1. Single cache: UpdatedAfter

On the cache whose primary key is the group key (Customer), the query builder exposes UpdatedAfter. The out long max overload gives you the next watermark in the same pass.

using var changed = customers.Query()
    .UpdatedAfter(lastSyncMs, out var newWatermarkMs)   // strictly greater than lastSyncMs
    .ExecutePooled();

// ... send `changed` to the client together with newWatermarkMs; it echoes it back next time.

UpdatedAfter is strictly greater-than, so echoing the watermark back yields nothing until something changes. The timestamp is the one passed to AddOrUpdate; on the Kafka path that is the message timestamp.

2. An order write bumps its customer

This is the point of a global index: writing ORD-003 for customer 42 moves customer 42's timestamp even though the Customer row did not change.

orders.AddOrUpdate(NewOrder("ORD-003", 42, 5m), T0 + 10_000);   // Ada's row did not change, her activity did

activity.Index.TryGetLastUpdated(42, out var adaActivityMs);   // T0 + 10_000
var changedCustomers = activity.Index.GetValuesGt(T0 + 4_000);  // [42]

3. Cross-cache response

Order cannot call UpdatedAfter (its key is not the group key), so the sync reads the index directly: GetValuesGt returns the changed customer ids, each cache is then read by key or by lane, and TryGetMax over the returned ids yields the watermark.

public sealed record CustomerChanges(Customer Customer, Order[] Orders);

public sealed record SyncResponse(CustomerChanges[] Changed, long Watermark);

public SyncResponse ChangesSince(long watermarkMs) {
    // Which customers changed — through their own row OR through any of their orders. Strictly after the watermark.
    var changedIds = _activity.Index.GetValuesGt(watermarkMs);
    if (changedIds.Count == 0)
        return new SyncResponse([], watermarkMs);

    var ids = changedIds.ToArray();
    var changes = new List<CustomerChanges>(ids.Length);
    foreach (var customerId in ids) {
        if (!_customers.Cache.TryGet(customerId, out var customer))
            continue;                                           // the customer row itself is gone: see "Deletes"
        var orders = _orders.Query().WithCustomerId(customerId).Execute();   // allocating: the array goes into the DTO
        changes.Add(new CustomerChanges(customer, orders.ToArray()));
    }

    // The watermark to hand back: the newest timestamp among the groups we actually returned.
    _activity.TryGetMax(ids, out var newWatermark);
    return new SyncResponse(changes.ToArray(), newWatermark);
}

CrossCacheSyncRoundTrips checks the three states a client sees: a first call with watermark 0 returns everyone, an idle call returns nothing and the same watermark, and a call after one order write returns that one customer with all her orders.

4. Deletes

Two cases, and they differ:

// Deleting one of several entities in a group bumps the group to the delete time: the client re-reads
// customer 42 and sees that ORD-003 is gone from the list.
orders.Cache.Remove("ORD-003", T0 + 20_000);
var bumped = activity.Index.GetValuesGt(T0 + 5_000);               // [42]

// Deleting the LAST entity of a group drops the group from the index: nothing is "after" any watermark.
orders.Cache.Remove("ORD-002", T0 + 21_000);                        // Bob's only order
customers.Cache.Remove(7, T0 + 22_000);                             // and Bob himself
var bobStillTracked = activity.Index.TryGetLastUpdated(7, out _);  // false
  • Removing one of several entities in a group bumps the group to the delete timestamp, so the client re-reads the customer and sees the shorter order list.
  • Removing the last entity drops the group from the index. The deletion is invisible to GetValuesGt, so a client cannot learn about it from the watermark alone. If your protocol must propagate "customer 7 is gone", keep a tombstone list or compare id sets on the client.

5. Prepared form

The same narrowing exists on Prepare<TArgs>(), with the watermark as the argument.

var changedSince = customers.Prepare<long>()
    .UpdatedAfter(static since => since)                // parameterized twin; also (from, to) and DateTimeOffset forms
    .BuildFrozen();

using var rows = changedSince.ExecutePooled(T0 + 3_000);

Pitfalls

  • Wall-clock timestamps in tests. AddOrUpdate(entity) stamps UtcNow; two writes in the same millisecond tie. Pass explicit timestamps when the order matters, as the tests do.
  • Equal timestamps at the watermark. UpdatedAfter is exclusive, so a change stamped exactly at the watermark you return is not delivered next time. Message timestamps make this rare but not impossible; if you need at-least-once, subtract one millisecond from the watermark you hand out and dedupe on the client.
  • Watermark from a filtered subset. TryGetMax(ids, out max) over the ids you actually return is right; the global maximum of the index would skip changes in groups you filtered out this round.
Prague documentation