Add store-backed FlowStore to datasourcex-util by rossdanderson · Pull Request #36 · caplin/DataSource-Extensions

rossdanderson · 2026-06-02T14:22:06Z

Adds FlowStore to datasourcex-util: a map whose entries are backed by an external store instead of being held entirely in memory.

Bounded in-memory hot set, reading through to the store on a miss.
Mutations exposed as a coroutine Flow of versioned deltas.
MutableFlowStore (owner): each mutation takes the caller's transaction handle (a jOOQ Configuration, a JDBC Connection, …), adapted via TxContext, and enlists the write on it. On commit the store refreshes its cache and emits the delta; on rollback nothing is published.
FlowStore (consumer): follows a delta stream and reads through on a miss, staying coherent via per-key versions. Removals leave a tombstone so a stale read-through cannot resurrect a deleted key.
Store SPI: CacheLoader / CacheWriter / CacheLoaderWriter. The store assigns and returns each write's version (a DB sequence, identity, or version column), so the version is the durable commit order and survives restarts.
VersionedMapEvent with Jackson 2, Jackson 3, and Fory serializers for distribution.
Dokka sample (StoreSamples) showing a MutableFlowStore enlisted on a blocking jOOQ transaction, with a TransactionListener that runs the buffered publish on commit/rollback so callers write plain dsl.transaction { … }.

Intended for values that are aggregate roots — each key is written and published as a whole.

A map whose entries are backed by an external store with a bounded in-memory hot set, read-through on miss, and changes exposed as a coroutine Flow of versioned deltas.

Source versions from the writer (DB), not an app-side VersionSource: the version is now commit-ordered and durable, fixing same-key write ordering and removing the in-memory counter that reset on restart. CacheWriter.write/delete return the assigned version; VersionSource and its implementations are deleted. - All cache writes are version-gated via Caffeine's per-key atomic compute (lock-free), so a read-through, owner commit, or out-of-order delta cannot clobber a newer entry. Owner commits emit the delta before refreshing the cache to close a cancellation gap. - Add tombstones (CacheEntry: Live/Tombstone) so a removal cannot be resurrected by a stale older read-through. - Bound the signal SharedFlow buffer (default 256, tunable) instead of Int.MAX_VALUE. - Reject null/non-integral version in both Jackson deserializers; document the String/JSON-native key/value round-trip limitation. - AutoCommitTxContext is one-shot (reuse throws; rollback-after-commit no-ops). - deleteAll versions per key; document loadAll/loadAllKeys defaults; delegate the CacheLoaderWriter factory overload. - De-flake FlowStoreTest (subscription synchronisation instead of delays); add tombstone, version-parsing, and tx-guard coverage.

- Centralize the VersionedMapEvent -> CacheEntry mapping in one toEntry() helper instead of restating it across the owner and consumer paths. - Add AbstractFlowStore.cacheReflectIfNewer and collapse the consumer's hand-rolled version-gated computeIfPresent onto it. - Drop the redundant loader param from MutableFlowStoreImpl (always the writer). - Rewrite KDoc/sample docs to describe behaviour without rejected-alternative framing, and drop stale references to the removed delta channel.

…fNewer

…sion instead)

Add store-backed FlowStore to datasourcex-util

14e6db6

A map whose entries are backed by an external store with a bounded in-memory hot set, read-through on miss, and changes exposed as a coroutine Flow of versioned deltas.