Tracking: high-throughput Rust ingestion with a three-replica WAL

Problem

Object-store WAL writes can put remote request latency and small-object overhead on the ingestion ACK path. A memory-only WAL removes that cost but cannot preserve acknowledged writes across process or node loss. We want an optional, high-throughput durable ingestion path implemented in Rust within lance-context, with minimal changes to Lance OSS.

This issue tracks incremental implementation and validation. The foundational PRs below do not yet enable a distributed writer in production or replace the existing ingestion path.

Proposed architecture

flowchart LR
    I["Ingress: encode Arrow once"] --> R["Session key to virtual bucket"]
    R --> L
    subgraph G["One replication group: three durable replicas"]
        L["Leader + persistent WAL"]
        F1["Follower + persistent WAL"]
        F2["Follower + persistent WAL"]
        L --> F1
        L --> F2
    end
    L -. "ACK after durable quorum commit" .-> I
    L -->|Committed records| P["Fenced Lance publisher"]
    P --> B["Blob: Lance data files + manifest"]
    B --> V["Visible to Lance readers"]

The diagram shows one replication group. Independent groups can ingest in
parallel. A WAL ACK precedes Lance visibility; safe reclamation additionally
requires durable publication checkpoints and protection for active readers and
snapshot transfers, as described below.

Stable session sharding. Route a canonical tenant/table/session key to a virtual bucket, then to a physical replication group. Multiple buckets can share a group; multiple groups can share a node and log engine. Preserve ordering within a session without imposing a global order across independent groups. Layout changes require an explicit migration and fencing protocol.
Three durable replicas per group. Use a leader and two followers on independent persistent storage. The proposed ACK boundary is a committed entry persisted on a majority (two of three), with durable leader-election state and recovery. Memory buffers alone are insufficient. Raft supplies ordering, election and committed-prefix safety; it does not publish Lance manifests. This tolerates a single replica failure under the stated storage assumptions, not simultaneous loss of a quorum's durable data.
Opaque payload replication. Encode Arrow batches once at ingress and replicate bytes without decoding Arrow on each follower. Decode at materialization using shared buffers where possible. Alignment repair, framing, checksums, kernel I/O and the log engine can still copy data; measure allocations and copies instead of claiming end-to-end zero-copy.
Asynchronous Lance publication. Materialize committed WAL records into Lance files and publish the corresponding manifest separately. A durable publisher fence and per-group source positions must make retries, concurrent commits and publisher recovery safe. A WAL ACK and visibility in a Lance reader are distinct boundaries and need an explicit API contract.
Checkpoint, replay and reclamation. Recover from durable snapshots plus a committed log suffix. Retire source payloads only after their Lance publication is durably established. Keep producer deduplication/fencing state independently. Physical collection must preserve active snapshot transfers, live entries and recovery dependencies.
Bounded resource use and progress. Share memory/disk admission across groups on a node, reserve temporary space for snapshots and log rewrites, and apply backpressure when publication or a replica falls behind. Maintenance must still make progress after producer admission closes. Local proposal limits alone do not bound replication traffic.

The implementation is evaluating established Rust consensus and log-engine components, including raft-rs and raft-engine. Useful references are Flink's checkpoint/replay and credit-based flow control, and Apache Fluss's bucket ownership, batched replication and asynchronous lake publication. Fluss uses leader/follower replication with ISR; this proposal does not imply that Fluss uses Raft.

Incremental delivery

Use small PRs with Refs #378 in their descriptions. Keep this tracking issue open across individual implementation PRs; merge each only after its checks pass.

Validation before enabling the path

Verify acknowledged records after leader crashes, follower lag, snapshot transfer interruption and full process restarts using real Rust processes and durable storage.
Exercise ambiguous replies, stale publishers, concurrent manifest commits and retries; verify complete Lance payloads and source positions after recovery.
Run sustained ingestion through publication and repeated reclamation, including controlled publication pauses. Report ACK latency, published throughput, backlog, RSS, allocated disk space and physical write amplification separately.
Compare batching, serialization/deserialization, shared-buffer ownership, bounded parallel reads and replication-group placement using reproducible workloads. Short component benchmarks are not evidence of sustained end-to-end throughput.
Publish only synthetic, portable workloads and sanitized results. Do not include private deployment identifiers, storage locations, identity configuration, business workloads or raw operational artifacts.

Related: #333 tracks maintenance scheduling in the existing system; this issue focuses on the optional durable ingestion path and its correctness/performance contract.

References

Flink state backends and incremental checkpoints.
Flink network stack and credit-based flow control (historical design explanation).
Fluss LogTablet replication and log lifecycle and Lance lake connector documentation, pinned to the researched revision.

feat: add shared WAL proposal admission budget (#377)

feat: reserve configurable WAL maintenance headroom (#379)

feat: share validated Arrow IPC decoding for WAL materialization (#380)

feat: add shared WAL record envelopes and payload validation (#381)

perf: move owned rollout upload buffers into core records (#382)

fix: honor registered rollout dataset locations (#383)

feat: define immutable WAL ingestion contracts (#384)

Progress update: the foundational PRs through #384 are merged. They provide the Lance 13 upgrade, allocation/ownership improvements, session placement, admission controls and immutable ingestion contracts. The distributed ingestion path remains experimental and is not enabled by these PRs.

A synthetic Rust experiment compared 32 MiB and 64 MiB publication source caps in an ABBA sequence. Each case wrote 8,192 distinct 128 KiB payloads (1 GiB), using 32 writers, 8,000 session keys and three RF3 groups across three nodes, then materialized into Lance on object storage. Other admission, fetch and maintenance settings were held fixed within the cohort.

Source cap / run	Published payload MiB/s	Encode-to-ACK p99 ms	Publications
32 MiB / A	36.78	3301	44
64 MiB / A	38.55	2493	30
64 MiB / B	41.95	3168	28
32 MiB / B	38.42	3157	46

All four cases passed complete payload verification, maintenance audits and process restart recovery: all nine group replicas restored their checkpoints, the Lance version/data stayed unchanged, and recovery produced zero additional publications.

Pooled published throughput increased from 37.59 to 40.18 MiB/s, about 6.9%. Two short runs per setting are insufficient to establish sustained capacity or a dependable p99 improvement, so this does not justify a default change yet. These measurements exclude HTTP, catalog resolution, receipt authentication and fixture generation; publisher memory includes a resident 1 GiB synthetic fixture pool. They test process recovery on the original volumes, not volume loss or replacement.

Next steps remain service integration, explicit retry/visibility semantics, replica-pressure coordination and longer bounded-retention tests. Future implementation PRs will continue to use Refs #378; this issue stays open. Only portable synthetic parameters and aggregate results are included here; operational artifacts remain outside the public repository.

fix: prevent racing dataset creation from overwriting data (#385)

Concurrent provisioning exposed a data-loss race that is addressed by #385: Lance checks an absent dataset before creation, then reopens its URI at commit time. A delayed creator can otherwise overwrite a second writer's newly appended data. The context-side guard permits only manifest version 1 during creation and preserves normal appends through the returned handle.

The deterministic regression delays an actual empty storage-listing result while another creator creates and appends. Removing the guard makes the test fail with zero rows instead of one; restoring it preserves the row and version. A barrier at manifest authorization was too late to reproduce this race, so it was replaced before merging.

A separate synthetic admission-window experiment stopped at its second case because it did not meet the predeclared maintenance-exposure gate. Both completed cases verified all 4 GiB of distinct payloads and passed process restart recovery on their original volumes, but the larger-window case completed only 12 maintenance rounds against a required minimum of 16. The remaining two ABBA cases were not launched. This is not evidence of an accepted sustained-throughput improvement, and no defaults changed.

The next performance comparison needs longer workloads with bounded fixture generation and explicit generation/encoding/admission timing. Service integration, distributed replica disk credits and producer-state lifecycle remain open; the foundational PRs do not enable the distributed writer in production.

A four-case synthetic Rust ingestion/publication/recovery comparison has passed all predeclared gates. Each case ingested 98,304 distinct 128 KiB payloads (12 GiB), with 32 writers, 8,000 session keys and three RF3 groups across three nodes. We held the total node proposal budget at 4 GiB and varied only the producer ceiling in an ABBA sequence.

Producer ceiling / run	Published payload MiB/s	Request-to-ACK p99 ms	Successful maintenance rounds
252 MiB / A	38.98	3040	98
1020 MiB / A	47.37	2326	35
1020 MiB / B	45.04	3582	35
252 MiB / B	38.87	2633	99

All cases verified every payload before and after a full WAL-process restart. All nine group replicas restored checkpoints; Lance versions stayed unchanged, with zero additional recovery publications and no maintenance failures. All WAL node process incarnations were explicitly stopped and their recorded identities checked.

Pooled published throughput was 38.92 versus 46.18 MiB/s (+18.6%) in this cohort. Tail latency did not consistently improve. External samples showed peak per-node allocated WAL footprints of 426 and 1412 MiB, respectively; peak sampled acknowledged publication backlogs were 198 and 631 MiB. Two trials per setting are insufficient to establish production capacity or a reliable tail-latency gain; defaults remain unchanged.

The streamed fixture generator retains no workload-wide payload pool and bounds concurrently generated raw payloads to 4 MiB for this workload. Generation is included in request/throughput timing. Maximum observed publisher process RSS high-water was approximately 335 MiB; RSS is not total Pod/cgroup memory, which also accounts for cache and other processes. External resource sampling actually ran roughly every 5–8 seconds, so sampled disk/RSS peaks are observations, not hard bounds. HTTP, authentication, catalog lookup and writer connection setup are outside timing; full readback and restart audits follow the timed publication drain. These results are separate from earlier precomputed-fixture or failed cohorts.

The Arrow schema-contract extraction in #386 (Refs #378) is now merged. Its local Rust/Python checks and all five GitHub workflows, including all packaging platforms and final distribution validation, passed at the reviewed head before merge. Replica disk credits, producer-state lifecycle, incremental checkpoints and production service integration remain open work.

feat: share canonical Arrow ingestion schema fingerprints (#386)

The local experimental Rust ingestion path now has a separate automatic materializer service. It discovers table generations through bounded catalog pages and claims work with expiring leases. Replicated publisher epochs and pinned Lance versions fence publication; the scheduling lease alone is not a storage fence. Active-table slots, a shared source-byte budget and a publication quantum bound work. There is no worker per session.

Real-process tests with etcd and WAL nodes cover competing workers, catalog changes during discovery, prepared-publication recovery after owner loss, stale-owner rejection, complete payload comparisons, catalog/WAL restarts and interrupted table closure. A separate HTTP/materializer test verifies that writes receive durable ACKs while materialization is paused, a replacement automatically publishes the backlog, and full ingestion/materializer/WAL process restart plus request retries preserves data without duplicate publication. All 36 columns of 34 synthetic rows, each with a 128 KiB binary payload, are compared; graceful service exit also passes. This is local correctness coverage, not sustained HTTP performance or a released service feature.

The object-storage experiment intended to compare serial and parallel file materialization stopped after its first serial case: two of 960 node-budget observations exceeded the predeclared 800 ms timeout. All 98,304 distinct 128 KiB payloads (12 GiB) passed initial readback, but the observation gate failed. The remaining three cases were never started, the workload was not repeated to obtain acceptance, and defaults remain unchanged. There is no accepted concurrency comparison.

Diagnostic recovery of that original data passed: all nine group replicas restored checkpoints, every payload was verified, the Lance version stayed unchanged, and recovery made zero additional publications. The original failed gate remains recorded; the status-timeout cause is still under investigation.

The materializer currently performs logical retirement. Physical checkpoint/compaction/collection orchestration, efficient discovery with many idle tables, replica disk credits, producer-state lifecycle, authorization/migration and sustained HTTP throughput remain open. Upcoming implementation PRs will continue to use Refs #378 and remain small; this issue stays open.

The local experimental WAL node now has opt-in automatic checkpoint, compaction and physical collection. One scheduler covers the groups sharing a node's engine/budget, including followers. It uses local durable applied cuts, recovers installed-but-uncompacted checkpoints after restart, and avoids rebuilding an unchanged group. Payload retirement still requires the existing replicated Lance publication proofs; scheduling alone does not authorize deletion.

Real-process tests cover interrupted compaction, unchanged idle groups, a held checkpoint builder across observation timeouts, and full process loss. An existing checkpoint cancellation/recovery regression also passes. Scheduling retains the current job identity after an uncertain observation, while existing builder limits, byte reservations and file/read pins protect concurrent work.

A separate local HTTP/materializer/node test writes 1,024 synthetic rows with 128 KiB binary payloads (128 MiB total), in four publication/drain windows. Eight concurrent upload futures bound generated bodies. Each node has a 32 MiB producer ceiling within a 256 MiB total proposal budget. The test never manually invokes checkpoint, compaction or collection. It compares all 36 columns, requires zero retained source payloads in the final compacted checkpoints and WAL footprints below 32 MiB, then restarts all HTTP/materializer/WAL processes and retries every request without changing the Lance version or duplicating rows.

This validates finite local service/reclamation/recovery behavior. It is not a sustained-throughput measurement, a hard filesystem quota, a volume-loss test, or a result on remote blob storage. Full-copy checkpoint cost, producer-state growth, replica disk credits, idle-table discovery, authorization/migration and longer service workloads remain open. The earlier object-storage concurrency cohort remains failed and has not been rerun. The changes remain local experimental work; implementation PRs will continue to use Refs #378.

The experimental Rust HTTP ingestion path now has a bounded workload driver with separate prepare, ingest, verify and recovery modes. It streams distinct Arrow payloads, preserves producer/sequence/body across retries, refuses a second ingestion start, and checks complete Lance rows. Request-to-ACK timing includes generation and encoding; durable ACK and Lance visibility are measured separately.

One new K8s/object-storage cohort completed 8,192 synthetic rows with 128 KiB binary payloads (1 GiB), eight writers, 31 sessions and three RF3 groups. All data became visible in 26.620 s: 38.467 MiB/s of binary payload. Request-to-ACK p50/p99/max were 16.167 / 324.863 / 1,121.279 ms. All 153 node-budget and 50 receipt observations passed their declared deadlines, and all 36 columns passed readback. This is one finite initial measurement, not a sustained-capacity or scaling result.

All HTTP, materializer, three WAL and experimental metadata processes were then explicitly killed. New process incarnations reused the same durable directories. Retrying all 248 latest producer/session requests and verifying every row again left Lance version 38 unchanged. Node-owned checkpoint/compaction/collection ran throughout with zero recorded maintenance errors; final observed WAL footprints after draining were below the 256 MiB producer ceiling. Some ingestion-time footprints exceeded that ceiling, which is admission control rather than a hard disk quota. All experiment processes have been stopped.

Each WAL Pod had a 4 CPU / 8 GiB limit on a separate host. HTTP, materializer, load generator and one experimental etcd shared another 4 CPU / 8 GiB Pod. Process resource samples were sparse, and shared cgroup memory reached about 7.08 GiB; without a before-run baseline this does not establish a memory bound or component attribution. The recovery test preserves volumes and does not establish volume-loss recovery or HA metadata.

The earlier failed file-materialization concurrency cohort remains failed and was not rerun. Service-side ACK latency attribution, efficient idle-table discovery, replica disk credits, producer-state lifecycle, incremental checkpoints and production authorization/migration remain open. These service changes are still experimental local work; implementation PRs will continue to use Refs #378.

feat(wal): persist immutable prepared Lance descriptors (#410)

feat(wal): select immutable manifests through the catalog quorum (#412)

feat(wal): persist publication outcomes from selected manifests (#414)

feat(wal): close source pins with durable publication fences (#415)

feat(wal): retire closed source payloads while retaining retry receip…

perf(wal): use optimized SHA-256 for opaque payload validation (#417)

feat(wal): attest committed source closures (#423)

feat(metrics): expose opt-in Lance object-store diagnostics (#421)

feat(wal): advance catalog cursors after source closure (#424)

The corrected materializer scheduler now has two fully accepted synthetic end-to-end results. Both write 32 GiB across four tables through three disjoint three-replica WAL groups, with four client processes in separate Pods:

Materializer deployment	Total table slots	Aggregate published payload throughput
One Pod, four slots	4	122.5–132.2 MiB/s
Four Pods, one slot each	4	126.1–136.5 MiB/s

The scheduler previously left available slots empty while another table task was active. Refilling those slots raised the single-Pod result from 44.9–46.1 MiB/s to 122.5–132.2 MiB/s at the same resource limits. The two corrected deployments above have the same total table concurrency: they do not establish scale-out with additional slots or WAL shards. The four-Pod configuration also has four times the aggregate materializer CPU/memory limits.

Each corrected case passed complete 36-column Lance readback, restart of all service processes on retained storage, 512 exact current-request retries, and the original resource and reclamation checks. Recovery added no publications and uploaded no WAL payload bodies. Throughput bounds come from one controller's monotonic observations; they are not statistical confidence intervals or a sustained-capacity result. These tests do not simulate volume or power loss.

Latency remains a limitation: individual clients' resolve p99 values range from 4.06 to 4.71 seconds across these cases, including admission and durable client journaling. They are not server-only quorum ACK latencies. Recorded materializer stage durations prioritize investigation of file writing and publication/cleanup waits over assuming IPC decoding dominates; the timers overlap and cannot be treated as CPU shares or an exclusive critical path.

A checkpoint-cadence comparison is running. A separate comparison will hold four materializer Pods and sixteen tables fixed while changing total slots from four to sixteen. Neither has a completed performance conclusion yet. Bounded sustained operation, production adapters, metadata availability and volume-loss recovery remain unfinished.

For the incremental APIs, #424 merged catalog advancement after source-closure proofs. #426 adds bounded collection of obsolete application snapshot files; its existing head passed CI, and integration with the latest main is being validated before updating and merging it. Physical log maintenance and explicit manifest-history retirement are subsequent increments. These component PRs do not by themselves enable the experimental ingestion service in production.

feat(wal): collect obsolete application snapshot files (#426)

feat(wal): reclaim obsolete physical log segments (#427)

feat(wal): retire bounded manifest history (#428)

The checkpoint-cadence comparison now has four fully accepted synthetic cloud cases. Each publishes 32 GiB across four tables with one materializer Pod / four table slots, four client Pods, nine WAL Pods in three disjoint RF3 groups, and one HTTP Pod. The Rust binaries, durable quorum ACK boundary, workload and resource limits are unchanged between settings.

Case	Checkpoint entries / maximum age	Published payload MiB/s	Process-accounted WAL writes / payload
Frequent A	256 / 1 s	122.382–132.042	6.921
Deferred A	4096 / 10 s	137.994–150.625	3.765
Deferred B	4096 / 10 s	142.721–156.306	3.753
Frequent B, resumed	256 / 1 s	113.791–139.114	6.870

Total payload divided by summed elapsed bounds gives 140.318–153.413 MiB/s for the two deferred cases (64 GiB), versus 117.930–135.486 MiB/s for the frequent pair. These are controller observation bounds, not confidence intervals or sustained capacity. Entry and age thresholds changed together; pressure-triggered checkpointing remained unchanged.

The final case encountered control-plane observation failures before ingestion. Its original services were reconciled and continued using a separately recorded ingestion clock; all original failed terminals remain failures. Thus this is not an uninterrupted ABBA run. Process write accounting spans service lifetime, including setup/readback and additional pre-ingestion idle time in the resumed case. It is not media write amplification, device utilization or an ingestion-only counter.

All four cases passed full Lance readback, all-service restart on retained storage, 512 exact latest-request retries per case, and resource/reclamation checks. Recovery added no publications and no WAL body uploads. Deferred-case client resolve p99 remained 3.55–4.69 seconds, including admission and durable client journaling; it is not server-only quorum ACK latency.

The fixed-four-worker/sixteen-table comparison is still in configuration staging and has no workload result. Physical replica scaling, sustained capacity and production integration remain unproven. Separately, #426, #427 and #428 have merged after their final-head checks passed, adding snapshot-file collection, physical log reclamation and bounded managed-manifest history retirement. These component APIs do not enable the ingestion service in production by themselves.

fix: reclaim client journal rewrite bytes before fencing (#430)

Navigation Menu

Repository navigation

Tracking: high-throughput Rust ingestion with a three-replica WAL #378

Problem

Proposed architecture

Incremental delivery

Validation before enabling the path

References

47 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Tracking: high-throughput Rust ingestion with a three-replica WAL #378

Description

Problem

Proposed architecture

Incremental delivery

Validation before enabling the path

References

Activity

beinan commented on Oct 19, 2026

beinan commented on Oct 19, 2026

beinan commented on Oct 19, 2026

beinan commented on Oct 19, 2026

beinan commented on Oct 19, 2026

beinan commented on Oct 19, 2026

47 remaining items

beinan commented on Oct 21, 2026

beinan commented on Oct 21, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions