Ingest That Can Be Replayed

Take a product feed that sends a message every time a price changes. At 10:00 product 1042 drops from 125 kr to 99 kr. That message has version 7. At 10:05 the price goes up to 119 kr. That message has version 8.

The import saves both. Then the 10:00 message arrives again. Now the database says 99 kr, and the site sells the product for 99 kr.

No error was logged. The import saved whatever came last, and this time the old message came last.

Messages come more than once

This is normal. A queue can deliver the same message twice, for example when the import saves the price but crashes before it tells the queue it is done. The second copy can arrive after a newer message.

And sometimes I do it myself. The import was down for an hour, so I run the files from that hour again, starting a little earlier to be sure nothing is missed.

So I assume every message can arrive twice, and in any order.

A key and a version

The key says which product the message is about. Here it is the product id.

The version says how new the message is. It has to come from the source system, because only the source knows the order the changes happened in. A number that goes up by one for each change is best. The time the import received the message does not work, because that is exactly the order that went wrong.

The rule is then simple: "Only save if the version is higher than the one I have." If the version is the same or lower, the message is skipped.

Deleted products

Example: product 1042 is removed from the shop. That message has version 9. The typical way is to delete the row from the database. Then I run some files again, and the version 8 message comes through. The import cannot find product 1042, so it creates it. The product is back in the shop at 119 kr.

Deleting the row also deleted the version. So I do not delete the row. I mark it as deleted and keep the version. The shop only shows products that are not marked. Now the old version 8 message meets version 9, and it is skipped.

If the product is put back in the shop later, the source sends version 10, and it is saved as normal.

The code

public sealed record ProductMessage(
    string ProductId,
    long Version, // set by the source, never by the import
    decimal? Price,
    bool Deleted);

public sealed record StoredProduct(long Version, decimal? Price, bool Deleted);

public enum Result { Saved, Duplicate, Old }

public sealed class ProductStore
{
    readonly Dictionary<string, StoredProduct> _products = new();

    public Result Apply(ProductMessage m)
    {
        if (_products.TryGetValue(m.ProductId, out var stored))
        {
            if (m.Version == stored.Version) return Result.Duplicate;
            if (m.Version < stored.Version) return Result.Old;
        }

        // A delete keeps the row and the version. It only marks the product.
        _products[m.ProductId] = new StoredProduct(m.Version, m.Price, m.Deleted);
        return Result.Saved;
    }

    public IEnumerable<string> ForSale() =>
        _products.Where(p => !p.Value.Deleted).Select(p => p.Key);
}

A delete goes through the same check, so an old delete cannot remove a newer product.

The sample keeps the products in memory to keep it short. In a database I put the check in the update itself, UPDATE Products SET ... WHERE ProductId = @id AND Version < @version, and insert the product if it is not there yet. If I read the version first and compare in C#, two workers can both read version 7 and both save. Whichever saves last wins, even if its message is older.

Where it does not help

Some sources do not send a version. A file with the whole catalogue every night often has no number in it. There I use the time the file was made, and I trust it less. Two changes in the same second get the same timestamp, and the second one is skipped as a duplicate.

A replay also cannot fix data that was saved wrong. If a mapping was wrong for three days and I fix it, the messages from those days come through again with the same versions, and they are all skipped. The code only looks at the version, not the content. To push a fix through, I reset the version on the rows I want to fix first.

Marked rows also pile up. If I clean them out, an old replay can bring the products back. So I keep them for longer than the oldest replay I would ever run.

Final thought

I still catch myself writing an import that just overwrites what is there, because my test only sends each message once, and in order. These days I ask where the version comes from before I write the import.