Improve this page

Conditional Updates

Every write to a Prague cache flows through AddOrUpdate(value) or AddOrUpdate(value, timestampMs). The cache decides whether the write is a logical no-op, an update, or an insert by comparing the incoming value to the resident one — and surfaces that decision to the caller and to downstream observers. The out overload hands you both halves: true for Add or Update, false for Same, and the previous row when there was a change. Every block on this page is compiled and asserted in samples/Prague.Samples/Concepts/ConditionalUpdates.cs.

var changed = cache.AddOrUpdate(NewOrder("ORD-001", 99.99m), out var previous);   // false: structurally equal → Same

On Same the resident instance stays in place — the reference in storage is not swapped, no index is re-keyed, and previous is null.

The `UpdateType` enum

Prague.Kafka.UpdateType names the five outcomes a write can have on the Kafka path:

Value		Meaning
`Filtered`	0	Rejected by a header, key or value filter (live phase only). Carries no key and no values — `key` is `default`, both values `null`.
`Same`	1	Key exists; the value is structurally equal to the resident one. The cache was not changed.
`Add`	2	New key. `oldValue` is `null`.
`Update`	3	Key exists; the value differs. Both `newValue` and `oldValue` are set.
`Delete`	4	Key removed — by a tombstone, or by a filter registered with `treatAsDelete`. `newValue` is `null`, `oldValue` the removed row. Fires only when the key was resident.

Same is the conditional-update outcome. When a producer re-emits the current state of an entity (a common pattern in change-feed pipelines), the cache compares the incoming POCO to the stored one and reports Same instead of Update. No after-handler sees a phantom update.

How equality is decided

Prague does not call object.Equals. Every cached value type implements ICacheEquatable<T> — bool CacheEquals(T? other) and int CacheGetHashCode() — and the generator writes the implementation as a partial of your [DataCache] type; do not write it by hand. It walks every public property and compares property-by-property, recursing into nested objects and collections:

var a = NewOrder("ORD-001", 99.99m);
var b = NewOrder("ORD-001", 99.99m);

var structurallyEqual = a.CacheEquals(b);          // true: every property compared, nested objects and collections recursed
var sameReference = a.Equals(b);                   // false: object.Equals is untouched

Members marked [DataCacheIgnoreEquality] are skipped — useful for monotonic timestamps and audit fields that change per emit but don't represent a state change. Product.LastSeenAtUnixMs in the sample domain is one:

// docs: the index-type showcase — Unique (Sku), explicit symmetric Many (CategoryId), Many with a
// bucket-capacity hint (Brand), Range over a decimal (Price), a named value index (InStock), has-value
// indices on a nullable and on a bool, [DataCacheIgnoreEquality] for conditional updates, and a
// [DataCacheSort] comparer for paging. OrderLine points at it with a ManyToOne foreign key.
namespace Prague.Samples.Models;

using Prague.Core;

[DataCache]
[DataCacheSort("ByPrice", "Price:1", "Id:1")]                   // total order: Id breaks price ties
public partial class Product {
    [DataCacheKey]
    public required int Id { get; init; }

    [DataCacheIndex(DataCacheIndexType.Unique)]
    public required string Sku { get; init; }

    [DataCacheIndex(DataCacheIndexType.Many, Symmetric = true)]    // reverse map (Id → CategoryId) for index-driven joins
    public required int CategoryId { get; init; }

    [DataCacheIndex(DataCacheIndexType.Many, 4)]                   // bucket capacity hint: few products per brand
    public required string Brand { get; init; }

    [DataCacheIndex(DataCacheIndexType.Range)]
    public required decimal Price { get; init; }

    [DataCacheValueIndex(CompareOp.GreaterThan, 0, IndexName = "InStock")]   // key set: WithInStock() = Stock > 0
    public required int Stock { get; init; }

    [DataCacheHasValueIndex]                                       // bool: WithIsPublished() = IsPublished == true
    public required bool IsPublished { get; init; }

    [DataCacheHasValueIndex]                                       // nullable: WithDiscontinuedAtUnixMs() = non-null
    public long? DiscontinuedAtUnixMs { get; init; }

    public required string Name { get; init; }

    [DataCacheIgnoreEquality]                                      // changes here never count as an Update
    public long LastSeenAtUnixMs { get; init; }
}

If two writes of the same Product differ only in LastSeenAtUnixMs, the second is reported as Same and the earlier value stays resident:

var changed = products.AddOrUpdate(NewProduct(lastSeenAtUnixMs: Now + 60_000), out _);   // only [DataCacheIgnoreEquality] fields differ → Same

Observing the outcome

ICacheAfterHandler<TKey, TValue> is the per-cache hook: ValueTask Handle(UpdateType updateType, TKey key, TValue? newValue, TValue? oldValue). Which arguments are populated follows the table above:

public sealed class OrderProjector : ICacheAfterHandler<string, Order> {
    public ValueTask Handle(UpdateType updateType, string key, Order? newValue, Order? oldValue) {
        switch (updateType) {
            case UpdateType.Add:
            case UpdateType.Update:
                Project(newValue!);                   // oldValue is null on Add, the previous row on Update
                break;
            case UpdateType.Delete:
                Retract(key, oldValue!);              // newValue is null, oldValue is the removed row
                break;
            case UpdateType.Same:                     // a re-emit of the resident value: a keepalive at most
            case UpdateType.Filtered:                 // a message a filter dropped: carries no key and no values
                break;
        }

        return ValueTask.CompletedTask;
    }

    private static void Project(Order order) => Console.WriteLine($"upsert {order.OrderId}");

    private static void Retract(string key, Order previous) => Console.WriteLine($"delete {key} (was {previous.Total})");
}

Register one per cache via the Kafka handler builder; WithAfterHandler<T>() adds T as a singleton ICacheAfterHandler<TKey, TValue>:

services.AddKafkaCaches(o => o.BootstrapServers = "kafka:9092",
    builder => builder.AddCache<OrderCache, string, Order>()
        .WithAfterHandler<OrderProjector>());

OrderProjector is invoked only during the live phase — i.e. after the consumer has crossed the partition EOF of the initial load. The initial replay does not fire handlers. During live processing a structurally equal re-emit reaches the handler as UpdateType.Same (with the resident value as newValue and oldValue == null), never as a phantom Update, so projectors can treat it as a keepalive or ignore it.

Multiple handlers per cache are allowed (every ICacheAfterHandler<TKey, TValue> registration is resolved). They run sequentially, awaited one after the other; an exception in one is caught and logged and does not skip the others.

Producer-side dispatch

KafkaCacheProducer.Produce(topic, key, value) always writes — there is no conditional skip on the producer. The generated cache.AddOrUpdateAndProduce(value) is the conditional form: it writes locally and produces only when AddOrUpdate reported a change. cache.RemoveAndProduce(key) removes locally and always publishes the tombstone, resident or not. See Kafka integration.

When the current state has to go back onto the topic even though nothing changed locally (a fresh record for the compacted topic, a re-emit for consumers that are rebuilding or joining later), use cache.AddOrUpdateAndProduce(value, force: true). The local write stays conditional; only the produce becomes unconditional. Consumers apply the record through their own equality gate, so a replica that already holds an equal value reports Same, exactly as for any re-emit.

public static void RePublish(OrderCache orders, Order order) {
    orders.AddOrUpdateAndProduce(order, force: true);   // same conditional local write, but the Produce always happens
}

For the whole cache at once there is cache.ForceProduceAll(), which snapshots every resident record and produces each one unconditionally. It is a repair operation (re-seeding a topic, bootstrapping a consumer that cannot replay), not something to call in normal operation; the details are in Kafka integration.

Why this matters

Compacted Kafka topics naturally re-emit the latest state on each rebalance and rebuild. Without conditional updates a downstream projector would see thousands of Update events on startup that are no-ops. With them the count of Update events is exactly the number of state transitions across the topic's lifetime.

Joins — composing cross-cache views.
Kafka integration — wiring filters and after-handlers.