WarpLogRecordExporter

A CRDT-backed log-record exporter.

Log records are stored in an Rga>: an ordered, append-only sequence. Insertion order within a single replica is preserved; cross-replica ordering is resolved by RGA's Lamport tiebreak — deterministic but not wall-clock accurate under clock skew.

Idempotency

Each LogRecord carries a caller-assigned LogRecord.recordId (8 bytes). export tracks which record ids have already been inserted: re-exporting a record with the same LogRecord.recordId returns ExportResult.Success immediately without inserting a duplicate into the Rga. The dedup state is rebuilt from the op-log on recover, so idempotency survives process restarts.

Key inversion

export returns ExportResult.Success the moment the record is durably written to the DurableStore — not when it is delivered to any backend. Delivery is asynchronous and eventually consistent; the CRDT merge guarantees that any replica which receives the record will incorporate it correctly, even if the record arrives out of order or more than once.

Buffer cap

maxRecords bounds how many records stay visible, and bufferPolicy decides which record gives way. Every drop is counted — exactly, per record, on ExporterHealth.dropped and ExporterHealth.refused — and a rate-limited info line reports the running total so the loss is not silent for a consumer who never reads health.

Per-record correlation was given up deliberately (#2218). This class used to log every drop with the evicted record's id and body, detailed enough to line up against a backend's log index — written when eviction was exceptional. At DEFAULT_MAX_LOG_RECORDS the buffer is full permanently, so every exported record evicts one and that line became a per-record narration of a ring buffer doing exactly what it is configured to do, on the export hot path. A signal that fires always is not one. What is lost is which records went; what is kept, and is what an operator actually reads, is how many.

• BufferPolicy.DROP_OLDEST evicts visible index 0 and then inserts the arrival, so the buffer is a sliding window over the most recent maxRecords records.

• BufferPolicy.DROP_NEWEST refuses the arrival. At a full buffer the newest record is the one arriving, so "drop the newest" drops it: on the export path the buffer freezes at the first maxRecords records and appends no further op — no insert, no eviction, no tombstone.

merge is the exception, and it is the intended production path, not a corner: snapshot/merge exist so this exporter can participate in gossip. A merge folds a remote op-log in wholesale, so it can push the visible count past maxRecords — remote inserts interleaving into the visible order, remote tombstones arriving from a peer running a different bufferPolicy — after which the gate simply keeps refusing. Neither policy evicts on the merge path.

So the claim for DROP_NEWEST is about what this replica emits, not about what its buffer holds: it never authors a Remove, which is what makes its own contribution to the shared op-log a downward-closed prefix (#2127). The two exporters that share BufferPolicy resolve "newest" differently — WarpSpanExporter evicts the newest buffered span and admits the arrival — and that is the reason this one does not.

On-disk layout — segmented op-log

An Rga is an op-log, and Rga.piece is an idempotent union of two op-logs. So the exporter does not need one key holding the whole log: it keeps the log in segments of at most segmentOps operations, each under its own key (otel.logs.seg.<n>), plus a small LogSegmentIndex naming the live ones (otel.logs.idx).

export appends the us.tractat.kuilt.crdt.RgaOp.Inserts that Rga.insertAllAfter already returns to the active segment and rewrites only that segment once per turn, so the encode-and-write cost is O(segmentOps) per turn — a constant, independent of how many records the log holds, and amortised across however many records the turn took (#2194). The previous layout re-encoded and rewrote the entire log on every single record, which is O(N) work once per record: Θ(N²) time and Θ(N²) bytes to accumulate N records (#1860).

Recovery reads the segments named by the index and unions them with Rga.piece. Set union is commutative and idempotent, so the reconstruction is exact and order-independent — and the persisted Rga wire form derives its Lamport clock from the op-set, so nothing is lost by not persisting it per segment.

Windowing the in-memory op-log

Eviction only tombstones: the evicted record's Insert op — body and all — stays in the log, so the in-memory op-log grew with the number of records ever exported even though maxRecords held visibility flat. Rga.dropWindow is the fix, and this exporter calls it in passes (windowPass): everything outside the retained window is dropped from log.

How cheaply the drop is recorded depends on who authored the dot, and only one of the two arms is a bound. This replica's own dots fold into a per-author compaction floor — O(authors), not O(elements dropped) — so a log fed only by export settles back to O(maxRecords) after every pass. A foreign author's dot cannot: raising another author's floor entry would annihilate dots it may not have minted yet, so Rga.dropWindow records those in an explicit RgaOp.Compact costing one (RgaId -> RgaId) pair each — and nothing ever prunes them, because a purge retains Compact unconditionally and Rga.piece unions the positions it carries. So the honest in-memory bound is O(maxRecords) on the export path, plus one bodiless pair per foreign element ever windowed away on the merge path.

That second term is a strict improvement on what preceded it — before windowing, a merged-in foreign Insert was retained whole, body included — but it is growth, not a bound, and a replica that gossips accumulates it for as long as the process lives. Bounding it needs the same causal-stability argument segment retirement does, and is not attempted here. WarpLogRecordExporterWindowingTest measures both terms against each other.

Rga.compact is not the mechanism, and could not be. It is the obvious candidate — it is the reclamation this codebase already had — and it reclaims nothing at all here, for a structural reason rather than a tuning one. Its condition 4 refuses to collect a tombstone that is still some live element's after, and this log is an append chain: every record is inserted after the previous one, so every element except the tail is the predecessor of a live successor. BufferPolicy.DROP_OLDEST evicts index 0, which at every maxRecords above one is the element furthest from being the tail — so the one tombstone condition 4 would accept is the one this exporter never produces. At maxRecords = 1 that stops holding: index 0 is the tail, and the record replacing it is appended after RgaId.HEAD, so condition 4 would accept the eviction. It reclaims nothing there either, because condition 3 blocks independently and at every cap: a delivered frontier is a fact about peers, and this class holds a DurableStore, not a us.tractat.kuilt.core.Seam. Windowing exists because forgetting position needs no barrier at all — the next paragraph is why.

Windowing is sound without a causal-stability barrier because it deliberately forgets position, not identity: a dropped dot stays suppressed, so a peer that still holds the raw Insert cannot push the record back in through merge — Rga.piece merges the suppression and re-purges under it. Which suppressor does the work follows the same split as the cost: the floor for this replica's own dots, and the retained RgaOp.Compact's compacted-id set for a foreign author's. What is given up is the stability of a survivor's position when its predecessor is dropped; see Rga.compactedBelow.

Retiring superseded segments

Windowing alone leaves the store growing: a windowed-away Insert leaves log, but it stays in whichever sealed segment it landed in. A sealed segment whose every op the suppression state already covers contributes nothing to what recovery reconstructs — the union re-purges those ops under the floor and the retained Compacts — so its key can be deleted. That is what a windowPass does next, and on the export path it is what keeps the number of keys recovery opens flat instead of growing with the records ever exported — the gossip path's key count is not bound this way; see below.

clear is the second driver, and the only one that retires the whole sealed set at once: it is a pass that retains nothing, so every sealed segment a replica fed by export holds becomes superseded in one step.

It is not simply "delete a key", and two rules make it safe:

• A segment carrying an RgaOp.Compact is never retired. A Compact is the only carrier of Rga's "once compacted, always compacted" guarantee for the ids it names, and nothing prunes one; dropping it lets a peer that never received the compaction re-admit the purged record. The legacy migration's segment and any merge-adopted segment can carry one.

• Unknown content means keep. A segment is retired only on positive evidence that every op it holds is superseded — never on the absence of evidence to the contrary. A segment whose content could not be read keeps its key and its place in the index.

The residue is the merge path's, again — and on disk it is not the bodiless pair the in-memory bound is priced in. A foreign author's dots are covered by an explicit RgaOp.Compact, nothing prunes one, and any segment carrying one is therefore pinned — where pinned means retained entire: every RgaOp.Insert it holds, bodies included, for the life of the store. Two shapes reach it:

• a sealed segment that happened to be active when a pass minted a Compact keeps its full segmentOps ops — ~123 KB at DEFAULT_LOG_SEGMENT_OPS; and

• a merge persists the remote op-log verbatim under a key of its own, so merging from a peer whose log carries a Compact — which any peer that has itself windowed a foreign author's dots does, i.e. any peer in a steady-state mesh — pins that peer's whole log. At DEFAULT_MAX_LOG_RECORDS that is megabytes per merge.

So the on-disk total settles for a replica fed by export and grows in whole records for one fed by gossip — a coarser split than the in-memory bound's. Consolidation — rewriting a pinned segment's Compact forward so the segment itself can go — is the obvious escape and is deliberately absent: §9 of the design declined it, and nothing implements it.

Parameters

Samples

val replica = ReplicaId("device-uuid-abc123")
val store = InMemoryDurableStore()
val exporter = WarpLogRecordExporter(replica = replica, store = store)

// Recover persisted state from a previous session (rebuilds dedup map too).
exporter.recover()

// recordId is an 8-byte caller-assigned identifier unique per record.
val record = LogRecord(
    recordId = ByteString(ByteArray(8) { it.toByte() }),
    body = "user checked out",
    severityNumber = 9, // INFO
    severityText = "INFO",
    observedEpochNanos = 1_000_000_000L,
    traceId = ByteString(ByteArray(16) { it.toByte() }),
    spanId = ByteString(ByteArray(8) { it.toByte() }),
    attributes = mapOf("user.id" to "42"),
)

// Idempotent: re-exporting the same recordId is a no-op (Rga set-union).
exporter.export(record)
val secondResult = exporter.export(record)
check(secondResult == ExportResult.Success)
check(exporter.snapshot().toList().size == 1) { "duplicate was stored" }