WarpOtlpBridge
Drains converged CRDTs — spans, logs, and metrics — to an OTLP-capable OtlpEdge, reconciling each signal by its own producer-local digest so a re-drain sends only what the endpoint lacks.
On each drain call, per signal, the bridge:
Snapshots the local CRDT.
Fetches a compact digest from the edge (what this producer already delivered).
Computes the delta — records present locally but absent from the digest (for metrics: series whose value-hash advanced).
Sends only the delta.
Spans additionally carry inferred causal SpanLinks (#846): the bridge runs inferCausalLinks over the full span snapshot (so a predecessor already delivered on an earlier drain still resolves), filters the links to those whose fromSpanId is in the delta, and threads them to OtlpEdge.send for emission on the OTLP Span.links wire field.
Why reconcile-by-digest?
A naive replay would retransmit the entire buffer on every reconnect. A digest-gated delta means the common case (edge already has the record) never touches the wire, offline-then-reconnect sends only the gap, and the CRDT's idempotent merge makes retries harmless.
Best-effort, per-signal isolation
Each signal is drained independently and best-effort: a failing signal never aborts the others, and its CRDT is left intact for the next attempt. A drain is a DrainResult.Failure only when every attempted signal failed; a partial success reports the counts that got through.
Parameters
the WarpTelemetry whose span/log/metric exporters are drained.
observation time stamped onto metric points. Required — inject a fixed clock in tests; never a real-dispatcher/wall-clock default that decouples from virtual time.
Samples
val telemetry = WarpTelemetry(
replica = us.tractat.kuilt.crdt.ReplicaId("device-uuid-abc123"),
store = InMemoryDurableStore(),
)
telemetry.recover()
// Export spans while possibly offline — export() succeeds on durable local write.
val span = SpanRecord(
traceId = kotlinx.io.bytestring.ByteString(ByteArray(16) { it.toByte() }),
spanId = kotlinx.io.bytestring.ByteString(ByteArray(8) { it.toByte() }),
parentSpanId = null,
name = "checkout",
kind = SpanKind.CLIENT,
startEpochNanos = 1_000_000_000L,
endEpochNanos = 2_000_000_000L,
)
telemetry.spans.export(span)
// When connectivity returns, drain to the backend. WarpOtlpBridge reconciles each
// signal (spans, logs, metrics) by digest: only records the edge doesn't have are
// sent, and a resend on reconnect cannot double-count. Spans also carry inferred
// causal links (#846). The clock stamps metric observation time.
val bridge = WarpOtlpBridge(telemetry, kotlin.time.Clock.System)
// Wire your OtlpEdge implementation and call drain whenever connectivity returns.
// DrainResult.Success(spansSent, logsSent, metricPointsSent) — 0s mean up to date.
// DrainResult.Failure(cause) — every attempted signal failed; retry later.
//
// val edge: OtlpEdge = MyKtorOtlpEdge(endpoint = "https://otel-collector.example.com")
// val result: DrainResult = bridge.drain(edge)
println("bridge ready: $bridge")