IncrementalResult
A converging result backed by a join-semilattice — the observation side of monotone (convergent) query execution in :kuilt-warp.
Think of it as a running tally that can only grow: contributions arrive from any number of peers in any order, duplicates are absorbed automatically, and the result refines monotonically toward the least upper bound of all contributions received so far.
Threshold reads (awaitThreshold) are the LVar-style observation primitive. A caller suspends until the result first satisfies a monotone predicate, then receives that stable snapshot. Because the lattice can only grow, once a threshold is crossed it stays crossed — the result cannot fall back below it, so the returned snapshot is permanently valid.
Thread safety
contribute is thread-safe. The internal MutableStateFlow uses a CAS loop so concurrent callers from multiple threads or coroutines converge correctly without external synchronisation. awaitThreshold and state collection are safe across multiple concurrent consumers.
Lattice contract
L must satisfy the three laws that Quilted.piece requires:
Idempotent —
a.piece(a) == aCommutative —
a.piece(b) == b.piece(a)Associative —
a.piece(b.piece(c)) == (a.piece(b)).piece(c)
These laws guarantee convergence regardless of contribution order, duplication, or delivery gaps — the same properties that make a kuilt fabric safe to use as a transport.
Parameters
the lattice bottom — the result before any contributions arrive.
Type Parameters
the lattice type — must be Quilted (idempotent/commutative/associative join).
Samples
val alice = ReplicaId("alice")
val bob = ReplicaId("bob")
// Contribute is synchronous and thread-safe.
val result = IncrementalResult(GCounter.ZERO)
result.contribute(GCounter.of(alice to 3L))
result.contribute(GCounter.of(bob to 2L))
// The lattice only grows: current value is the join of all contributions so far.
check(result.state.value.value == 5L)
// Duplication is absorbed — same delta twice changes nothing.
result.contribute(GCounter.of(alice to 3L))
check(result.state.value.value == 5L)
// awaitThreshold: in a suspend context, suspends until the predicate is first true.
// val crossed: GCounter = result.awaitThreshold { it.value >= 5L }
// Returns the first value that satisfies the predicate; the lattice cannot fall below it.