Crate chorus_client 

Database-ready quorum WAL over 1, 3, or 5 GCS Rapid zonal buckets (typically three).

Start with SegmentedVolume::recover using the database’s durable checkpoint boundary. Consume the returned Recovery stream through its fixed end, then call Recovery::start. Submit caller-numbered opaque records through WalHandle::enqueue_append; admission returns an AppendCompletion without waiting for durability. A completion success is durable on a strict-majority quorum of the configured zones. Use Error::may_have_committed on a completion failure: ambiguous outcomes may replay after takeover, so delivery is at-least-once. Error::ActiveSegmentFull is different: it is definitive admission backpressure, consumes no sequence number, and leaves the writer healthy. Truncate retained sealed history so a deferred rotation can proceed, then retry the same record.

One sequence number identifies one application-encoded transaction record; the WAL never adds another application-level batching layer. Startup replay uses record boundaries through WalSeqNo and is available only on Recovery. Once the database has durably checkpointed its own state, call WalHandle::truncate_before to delete whole sealed segments below that checkpoint. Startup and periodic maintenance retry already-authorized deletion tombstones before repairing missing immutable sealed copies; degraded rotations also schedule a targeted repair. Active segments are never repaired in place, and maintenance never advances the database’s checkpoint floor autonomously. See the database_wal example for the complete lifecycle. Every fallible public operation returns Error.

Operational constraints

A typical production volume uses three Rapid buckets in distinct zones of one region plus a regional bucket for the default GCS manifest store. Bucket arguments are full v2 resource names, and zonal list order is durable replica identity. Current manifests bind that ordered set in chorus.buckets; its length is the replica count, and missing, duplicate, or later mismatched bindings are rejected. Physical zone placement is not discoverable through this API and remains an operator check.

Rapid zonal buckets have neither Object Versioning nor soft delete, so replacement and truncation deletes are permanent. Archive immutable sealed segments before truncation when the database needs point-in-time recovery. The default GCS manifest directory retains roughly 115 current checksummed segments before rotation defers; custom ManifestStore implementations may report a larger budget. WalEngineConfig::max_active_segment_bytes bounds that deferral with non-poisoning Error::ActiveSegmentFull backpressure.

The default GCS manifest is one repeatedly updated object. Treat WalEngineConfig::max_segment_bytes as the single-pending refill floor: for encoded throughput T bytes/s and worst-case provision-plus-fold latency L, size it above T * L with operational headroom. Exceeding that bound is fail-closed: append dispatch pauses before an unregistered segment can receive records.

Complete example

This is the same source built as the database_wal Cargo example.

//! End-to-end database integration using three GCS Rapid zonal buckets for
//! segment data plus one regional bucket hosting the manifest register.

use std::env;

use anyhow::{Context, Result};
use bytes::{BufMut, Bytes, BytesMut};
use chorus_client::{
    BearerAuth, ClientConfig, GrpcReplicaFactory, RefreshingAuthConfig, SegmentedVolume,
    WalEngineConfig, WalSeqNo,
};
use futures::TryStreamExt;

#[tokio::main]
async fn main() -> Result<()> {
    // Use the Cloud Storage gRPC endpoint (`https://storage.googleapis.com`) for
    // every factory unless the deployment has validated another gRPC-compatible
    // endpoint. Regional JSON/XML endpoints do not support gRPC. Each zonal
    // factory must use a distinct bucket, passed as a full v2 resource name such
    // as `projects/_/buckets/my-rapid-zone-a`.
    let endpoints = required_csv("CHORUS_GCS_ENDPOINTS")?;
    let buckets = required_csv("CHORUS_GCS_BUCKETS")?;
    let manifest_endpoint = required_var("CHORUS_GCS_MANIFEST_ENDPOINT")?;
    let manifest_bucket = required_var("CHORUS_GCS_MANIFEST_BUCKET")?;
    anyhow::ensure!(endpoints.len() == 3 && buckets.len() == 3);

    // ADC is the normal production path. All factories share this ArcSwap-backed
    // auth handle, so refreshed tokens reach existing gRPC clients without
    // rebuilding channels. Use GrpcReplicaFactory::connect(..., Some(token)) for
    // a controlled static-token deployment, or None for a local anonymous fake.
    let auth = BearerAuth::google_adc(RefreshingAuthConfig::default()).await?;
    let mut factories = Vec::with_capacity(3);
    for zone in 0..3 {
        factories.push(
            GrpcReplicaFactory::connect_with_auth(
                zone,
                &endpoints[zone],
                buckets[zone].clone(),
                auth.clone(),
            )
            .await?,
        );
    }

    // the regional bucket carries only the manifest control register
    let manifest_factory =
        GrpcReplicaFactory::connect_with_auth(3, &manifest_endpoint, manifest_bucket, auth.clone())
            .await?;

    let volume = SegmentedVolume::new(
        factories,
        manifest_factory,
        "databases/orders/wal",
        ClientConfig::default(),
    )?;

    // Load this boundary from the database's own durable checkpoint. Recovery
    // starts replay there. The manifest independently prevents objects below a
    // committed truncation floor from resurrecting history.
    let checkpoint_resume = WalSeqNo::record(0);

    // Recovery fences and seals the manifest's active tail, then directly
    // streams the fixed replay range. It does not create the next appendable
    // segment until this stream has been consumed successfully.
    let mut recovery = volume.recover(checkpoint_resume).await?;
    let replay_end = recovery.end;
    while let Some(record) = recovery.try_next().await? {
        apply_to_database(record.payload.as_ref())?;
        let _next_durable_resume = record.next_seqno();
    }
    // The default queue is 256 records; size max_segment_bytes separately from
    // encoded throughput and the manifest CAS-rate budget.
    let mut wal = recovery.start(WalEngineConfig::default()).await?;

    // The database owns sequence allocation. `replay_end` is the first sequence
    // available after startup. Admission verifies each contiguous number and
    // returns before GCS durability, allowing the completions to flow through a
    // separate transactional apply pipeline.
    let first_seqno = replay_end;
    let completion_a = wal
        .enqueue_append(
            first_seqno,
            encode_transaction(&[b"put customer/7 alice", b"update customer-index/7"]),
        )
        .await?;
    let completion_b = wal
        .enqueue_append(
            WalSeqNo::record(first_seqno.record_index + 1),
            encode_transaction(&[b"debit account/9 50"]),
        )
        .await?;

    // The appends are now owned by the WAL and may be concurrently in flight.
    // Await durability before applying each transaction to database state. On
    // failure, Error::may_have_committed() distinguishes ambiguous outcomes
    // that require recovery before reusing or advancing the sequence number.
    let (receipt_a, receipt_b) = match tokio::try_join!(completion_a, completion_b) {
        Ok(receipts) => receipts,
        Err(error) => {
            if error.may_have_committed() {
                anyhow::bail!(
                    "append outcome may have committed; restart recovery before reusing the \
                     sequence number: {error}"
                );
            }
            return Err(error.into());
        }
    };
    println!(
        "committed appends {:?} and {:?}; durable resumes {:?} and {:?}",
        receipt_a.seqno,
        receipt_b.seqno,
        receipt_a.next_seqno(),
        receipt_b.next_seqno()
    );

    println!(
        "replayed through exclusive record {}",
        replay_end.record_index
    );

    // Sealed-segment repair is automatic at engine startup, after each
    // rotation, and periodically. It never gap-fills or mutates the active
    // appendable segment, and the engine serializes it with truncation.

    // After the database has durably checkpointed all effects before
    // `replay_end`, request whole-segment deletion. Replay exists only during
    // startup recovery, so a running WAL has no concurrent reader pins. Persist
    // `replay_end` as the database checkpoint regardless of deletion counts;
    // Chorus does not manufacture a replacement checkpoint.
    let report = wal.truncate_before(replay_end).await?;
    println!(
        "truncation deleted_segments={} deleted_objects={}",
        report.deleted_segments, report.deleted_objects
    );

    // Graceful shutdown rejects new appends and allows the accepted-work drain
    // and owned-task joins up to `shutdown_timeout`. On expiry it aborts the
    // remaining tasks, bounds the cleanup join by the same interval, and returns
    // `Error::ShutdownTimeout`; `abort` remains for exceptional termination.
    wal.shutdown().await?;
    Ok(())
}

fn required_var(name: &str) -> Result<String> {
    env::var(name).with_context(|| format!("set {name}"))
}

fn required_csv(name: &str) -> Result<Vec<String>> {
    env::var(name)
        .with_context(|| format!("{name} is required"))
        .map(|value| value.split(',').map(str::to_owned).collect())
}

fn apply_to_database(record: &[u8]) -> Result<()> {
    println!("apply {}", String::from_utf8_lossy(record));
    Ok(())
}

// The database, not the WAL, defines transaction encoding. This example uses a
// tiny length-prefixed envelope; production code would normally call its
// existing transaction codec and submit the resulting bytes once.
fn encode_transaction(operations: &[&[u8]]) -> Bytes {
    let mut encoded = BytesMut::new();
    encoded.put_u32(operations.len() as u32);
    for operation in operations {
        encoded.put_u32(operation.len() as u32);
        encoded.extend_from_slice(operation);
    }
    encoded.freeze()
}

Structs

AppendCompletion

Owned durability future returned after a record is admitted in sequence.

AppendReceipt

Durability evidence for one caller-numbered record.

BearerAuth

Cloneable authentication handle shared by existing gRPC clients.

ClientConfig

Retry policy for transport operations used by recovery and writes.

GrpcReplicaFactory

Reusable Google Storage v2 channel, authentication handle, and zonal routing token for one Rapid bucket.

ManifestVersion

Opaque optimistic-concurrency token for one observed register state.

NoopMetricsRecorder

Recorder used when the application does not configure metrics.

Recovery

Startup recovery stream and capability to resume from a fenced frontier.

RecoveryTimings

Wall-clock cost of the recovery phases that finish before the replay stream is handed back. epoch_claim is the manifest CAS that fences prior writers; prepare covers directory adoption and repair, committed-seal enforcement, and the appendable-candidate takeover walk. Replay and Recovery::start are timed by the caller. This is diagnostic only and is never read on the correctness path.

RefreshingAuthConfig

Controls how aggressively a refreshing credential is renewed.

RepairReport

Result of one best-effort immutable sealed-segment repair pass.

SegmentedVolume

WAL namespace: the zonal replica factories for segment data (one per zone) plus one regional factory hosting the manifest control register.

TruncationReport

Result of one application-triggered truncation pass.

VersionedManifest

One consistent read of the register: its fields and the version token guarding the next conditional update.

WalEngineConfig

Capacity controls for the in-process transactional WAL pipeline.

WalHandle

Control plane for a running WAL engine.

WalRecord

One opaque application record emitted by startup replay.

WalSeqNo

Application-assigned record number and durable checkpoint boundary.

Enums

Error

Failure returned by every fallible chorus-client public operation.

ManifestStoreError

Why a register operation did not produce a new state.

TransportCode

Stable transport classification used by retry and fencing logic.

Traits

AccessTokenSource

Supplies bearer tokens to BearerAuth.

CounterFn

Handle for a monotonically increasing metric.

GaugeFn

Handle for an absolute point-in-time metric.

HistogramFn

Handle for a sampled distribution.

ManifestStore

A linearizable compare-and-swap register holding the manifest.

MetricsRecorder

Registers backend-owned metric handles used directly by Chorus.

UpDownCounterFn

Handle for an additive metric that may increase or decrease.