Struct AppendStreamBuilder 

pub struct AppendStreamBuilder { /* private fields */ }

Configures a bounded, concurrent table append stream.

Retries temporary rejections by default. Replaying unknown outcomes requires Self::retry_unknown because an unacknowledged attempt may have committed.

Telemetry

For a log or telemetry sink that should keep running after batch failures, use AppendFailurePolicy::Continue and keep the circuit breaker enabled. Start with one concurrent batch per stream when producers run on multiple application replicas; tune the total concurrency across all producers.

use scopedb_client::AppendFailurePolicy;

let stream = client.table("events").append_stream()
    .failure_policy(AppendFailurePolicy::Continue)
    .max_concurrent_batches(1)
    .build()?;

if let Err(error) = stream.try_send(&serde_json::json!({"name": "request.completed"})) {
    eprintln!("telemetry row not admitted: {error}");
}
let report = stream.shutdown().await?;
println!("Write results: {report:?}");

Continue mode is best effort. Monitor reports and AppendStream::stats for failed, unknown, and locally dropped rows. Send failure diagnostics to a separate sink. AppendStream::try_send never waits for capacity; AppendStream::send provides backpressure instead.

Implementations

impl AppendStreamBuilder

pub fn failure_policy(self, failure_policy: AppendFailurePolicy) -> Self

Chooses whether to stop or continue after a batch fails. Defaults to Stop.

Examples found in repository

examples/telemetry.rs (line 51)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Continue mode is explicitly best effort: failed or ambiguous batches are
    // accounted for, released, and do not stop later telemetry batches.
    let telemetry = table
        .append_stream()
        .failure_policy(AppendFailurePolicy::Continue)
        .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
        .target_batch_bytes(1024 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(32 * 1024 * 1024)
        // This limit applies to each stream; include every application replica.
        .max_concurrent_batches(1)
        .attempt_timeout(Duration::from_secs(10))
        .on_batch_failure(|event| {
            // A synchronous observer; use a separate diagnostics sink.
            eprintln!("telemetry append failure: {event:?}");
        })
        .build()?;

    let mut dropped_locally = 0_u64;
    for request in 1..=100 {
        if !track(&telemetry, request) {
            dropped_locally += 1;
        }
    }
    println!("locally dropped: {dropped_locally}");

    // In a service, stop and join producer tasks before this settlement barrier.
    let report = telemetry.shutdown().await?;
    if report.outcome != AppendDeliveryOutcome::Ok {
        eprintln!("telemetry loss or ambiguity: {report:?}");
    }
    println!("final stream stats: {:?}", telemetry.stats());

    Ok(())
}

pub fn target_batch_bytes(self, target_batch_bytes: usize) -> Self

Sets the target NDJSON payload size (default 4 MiB, maximum 8 MiB). One row may exceed this target up to 8 MiB.

Examples found in repository

examples/telemetry.rs (line 53)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Continue mode is explicitly best effort: failed or ambiguous batches are
    // accounted for, released, and do not stop later telemetry batches.
    let telemetry = table
        .append_stream()
        .failure_policy(AppendFailurePolicy::Continue)
        .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
        .target_batch_bytes(1024 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(32 * 1024 * 1024)
        // This limit applies to each stream; include every application replica.
        .max_concurrent_batches(1)
        .attempt_timeout(Duration::from_secs(10))
        .on_batch_failure(|event| {
            // A synchronous observer; use a separate diagnostics sink.
            eprintln!("telemetry append failure: {event:?}");
        })
        .build()?;

    let mut dropped_locally = 0_u64;
    for request in 1..=100 {
        if !track(&telemetry, request) {
            dropped_locally += 1;
        }
    }
    println!("locally dropped: {dropped_locally}");

    // In a service, stop and join producer tasks before this settlement barrier.
    let report = telemetry.shutdown().await?;
    if report.outcome != AppendDeliveryOutcome::Ok {
        eprintln!("telemetry loss or ambiguity: {report:?}");
    }
    println!("final stream stats: {:?}", telemetry.stats());

    Ok(())
}

examples/bulk_append.rs (line 31)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // The default Stop policy gives strict barriers. Local memory and request
    // concurrency are bounded independently; concurrent batches have no
    // defined commit order.
    let stream = table
        .append_stream()
        .target_batch_bytes(32 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(1024 * 1024)
        .max_concurrent_batches(4)
        .attempt_timeout(Duration::from_secs(30))
        .build()?;

    let rows = (1..=10_000).map(|id| {
        serde_json::json!({
            "id": id,
            "occurred_at": jiff::Timestamp::now(),
            "name": format!("example-{id}"),
        })
    });

    let admission_error = match stream.send_all(rows).await {
        Ok(result) => {
            println!("accepted locally: {} rows", result.accepted_rows);
            None
        }
        Err(error) => {
            // There is no stream-wide abort or rollback. Settle the already
            // accepted prefix even if row production or admission stopped.
            eprintln!("row admission stopped; settling the accepted prefix: {error}");
            Some(error)
        }
    };

    match stream.shutdown().await {
        Ok(report) => {
            println!("committed remotely: {} rows", report.committed_rows);
            println!("final stream stats: {:?}", stream.stats());
        }
        Err(error) => {
            if matches!(
                error.append_details().map(|details| details.status),
                Some(AppendState::Unknown)
            ) {
                eprintln!("commit outcome is unknown; reconcile before replaying");
            } else {
                eprintln!("bulk append failed: {error}");
            }
            eprintln!("final stream stats: {:?}", stream.stats());
            return Err(error.into());
        }
    }

    if let Some(error) = admission_error {
        return Err(error.into());
    }
    Ok(())
}

pub fn max_batch_rows(self, max_batch_rows: usize) -> Self

Sets the maximum rows in one HTTP batch, up to the 200,000-row protocol limit.

Examples found in repository

examples/telemetry.rs (line 54)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Continue mode is explicitly best effort: failed or ambiguous batches are
    // accounted for, released, and do not stop later telemetry batches.
    let telemetry = table
        .append_stream()
        .failure_policy(AppendFailurePolicy::Continue)
        .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
        .target_batch_bytes(1024 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(32 * 1024 * 1024)
        // This limit applies to each stream; include every application replica.
        .max_concurrent_batches(1)
        .attempt_timeout(Duration::from_secs(10))
        .on_batch_failure(|event| {
            // A synchronous observer; use a separate diagnostics sink.
            eprintln!("telemetry append failure: {event:?}");
        })
        .build()?;

    let mut dropped_locally = 0_u64;
    for request in 1..=100 {
        if !track(&telemetry, request) {
            dropped_locally += 1;
        }
    }
    println!("locally dropped: {dropped_locally}");

    // In a service, stop and join producer tasks before this settlement barrier.
    let report = telemetry.shutdown().await?;
    if report.outcome != AppendDeliveryOutcome::Ok {
        eprintln!("telemetry loss or ambiguity: {report:?}");
    }
    println!("final stream stats: {:?}", telemetry.stats());

    Ok(())
}

examples/bulk_append.rs (line 32)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // The default Stop policy gives strict barriers. Local memory and request
    // concurrency are bounded independently; concurrent batches have no
    // defined commit order.
    let stream = table
        .append_stream()
        .target_batch_bytes(32 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(1024 * 1024)
        .max_concurrent_batches(4)
        .attempt_timeout(Duration::from_secs(30))
        .build()?;

    let rows = (1..=10_000).map(|id| {
        serde_json::json!({
            "id": id,
            "occurred_at": jiff::Timestamp::now(),
            "name": format!("example-{id}"),
        })
    });

    let admission_error = match stream.send_all(rows).await {
        Ok(result) => {
            println!("accepted locally: {} rows", result.accepted_rows);
            None
        }
        Err(error) => {
            // There is no stream-wide abort or rollback. Settle the already
            // accepted prefix even if row production or admission stopped.
            eprintln!("row admission stopped; settling the accepted prefix: {error}");
            Some(error)
        }
    };

    match stream.shutdown().await {
        Ok(report) => {
            println!("committed remotely: {} rows", report.committed_rows);
            println!("final stream stats: {:?}", stream.stats());
        }
        Err(error) => {
            if matches!(
                error.append_details().map(|details| details.status),
                Some(AppendState::Unknown)
            ) {
                eprintln!("commit outcome is unknown; reconcile before replaying");
            } else {
                eprintln!("bulk append failed: {error}");
            }
            eprintln!("final stream stats: {:?}", stream.stats());
            return Err(error.into());
        }
    }

    if let Some(error) = admission_error {
        return Err(error.into());
    }
    Ok(())
}

pub fn flush_interval(self, flush_interval: Duration) -> Self

Sets the maximum delay between buffering the first row and dispatching its batch.

Examples found in repository

examples/telemetry.rs (line 55)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Continue mode is explicitly best effort: failed or ambiguous batches are
    // accounted for, released, and do not stop later telemetry batches.
    let telemetry = table
        .append_stream()
        .failure_policy(AppendFailurePolicy::Continue)
        .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
        .target_batch_bytes(1024 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(32 * 1024 * 1024)
        // This limit applies to each stream; include every application replica.
        .max_concurrent_batches(1)
        .attempt_timeout(Duration::from_secs(10))
        .on_batch_failure(|event| {
            // A synchronous observer; use a separate diagnostics sink.
            eprintln!("telemetry append failure: {event:?}");
        })
        .build()?;

    let mut dropped_locally = 0_u64;
    for request in 1..=100 {
        if !track(&telemetry, request) {
            dropped_locally += 1;
        }
    }
    println!("locally dropped: {dropped_locally}");

    // In a service, stop and join producer tasks before this settlement barrier.
    let report = telemetry.shutdown().await?;
    if report.outcome != AppendDeliveryOutcome::Ok {
        eprintln!("telemetry loss or ambiguity: {report:?}");
    }
    println!("final stream stats: {:?}", telemetry.stats());

    Ok(())
}

examples/bulk_append.rs (line 33)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // The default Stop policy gives strict barriers. Local memory and request
    // concurrency are bounded independently; concurrent batches have no
    // defined commit order.
    let stream = table
        .append_stream()
        .target_batch_bytes(32 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(1024 * 1024)
        .max_concurrent_batches(4)
        .attempt_timeout(Duration::from_secs(30))
        .build()?;

    let rows = (1..=10_000).map(|id| {
        serde_json::json!({
            "id": id,
            "occurred_at": jiff::Timestamp::now(),
            "name": format!("example-{id}"),
        })
    });

    let admission_error = match stream.send_all(rows).await {
        Ok(result) => {
            println!("accepted locally: {} rows", result.accepted_rows);
            None
        }
        Err(error) => {
            // There is no stream-wide abort or rollback. Settle the already
            // accepted prefix even if row production or admission stopped.
            eprintln!("row admission stopped; settling the accepted prefix: {error}");
            Some(error)
        }
    };

    match stream.shutdown().await {
        Ok(report) => {
            println!("committed remotely: {} rows", report.committed_rows);
            println!("final stream stats: {:?}", stream.stats());
        }
        Err(error) => {
            if matches!(
                error.append_details().map(|details| details.status),
                Some(AppendState::Unknown)
            ) {
                eprintln!("commit outcome is unknown; reconcile before replaying");
            } else {
                eprintln!("bulk append failed: {error}");
            }
            eprintln!("final stream stats: {:?}", stream.stats());
            return Err(error.into());
        }
    }

    if let Some(error) = admission_error {
        return Err(error.into());
    }
    Ok(())
}

pub fn channel_capacity(self, channel_capacity: usize) -> Self

pub fn max_buffered_bytes(self, max_buffered_bytes: usize) -> Self

Sets the maximum bytes admitted locally but not yet settled remotely.

Examples found in repository

examples/telemetry.rs (line 56)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Continue mode is explicitly best effort: failed or ambiguous batches are
    // accounted for, released, and do not stop later telemetry batches.
    let telemetry = table
        .append_stream()
        .failure_policy(AppendFailurePolicy::Continue)
        .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
        .target_batch_bytes(1024 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(32 * 1024 * 1024)
        // This limit applies to each stream; include every application replica.
        .max_concurrent_batches(1)
        .attempt_timeout(Duration::from_secs(10))
        .on_batch_failure(|event| {
            // A synchronous observer; use a separate diagnostics sink.
            eprintln!("telemetry append failure: {event:?}");
        })
        .build()?;

    let mut dropped_locally = 0_u64;
    for request in 1..=100 {
        if !track(&telemetry, request) {
            dropped_locally += 1;
        }
    }
    println!("locally dropped: {dropped_locally}");

    // In a service, stop and join producer tasks before this settlement barrier.
    let report = telemetry.shutdown().await?;
    if report.outcome != AppendDeliveryOutcome::Ok {
        eprintln!("telemetry loss or ambiguity: {report:?}");
    }
    println!("final stream stats: {:?}", telemetry.stats());

    Ok(())
}

examples/bulk_append.rs (line 34)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // The default Stop policy gives strict barriers. Local memory and request
    // concurrency are bounded independently; concurrent batches have no
    // defined commit order.
    let stream = table
        .append_stream()
        .target_batch_bytes(32 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(1024 * 1024)
        .max_concurrent_batches(4)
        .attempt_timeout(Duration::from_secs(30))
        .build()?;

    let rows = (1..=10_000).map(|id| {
        serde_json::json!({
            "id": id,
            "occurred_at": jiff::Timestamp::now(),
            "name": format!("example-{id}"),
        })
    });

    let admission_error = match stream.send_all(rows).await {
        Ok(result) => {
            println!("accepted locally: {} rows", result.accepted_rows);
            None
        }
        Err(error) => {
            // There is no stream-wide abort or rollback. Settle the already
            // accepted prefix even if row production or admission stopped.
            eprintln!("row admission stopped; settling the accepted prefix: {error}");
            Some(error)
        }
    };

    match stream.shutdown().await {
        Ok(report) => {
            println!("committed remotely: {} rows", report.committed_rows);
            println!("final stream stats: {:?}", stream.stats());
        }
        Err(error) => {
            if matches!(
                error.append_details().map(|details| details.status),
                Some(AppendState::Unknown)
            ) {
                eprintln!("commit outcome is unknown; reconcile before replaying");
            } else {
                eprintln!("bulk append failed: {error}");
            }
            eprintln!("final stream stats: {:?}", stream.stats());
            return Err(error.into());
        }
    }

    if let Some(error) = admission_error {
        return Err(error.into());
    }
    Ok(())
}

pub fn max_concurrent_batches(self, max_concurrent_batches: usize) -> Self

Sets the maximum number of append batches sent concurrently per stream (default 4).

Examples found in repository

examples/telemetry.rs (line 58)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Continue mode is explicitly best effort: failed or ambiguous batches are
    // accounted for, released, and do not stop later telemetry batches.
    let telemetry = table
        .append_stream()
        .failure_policy(AppendFailurePolicy::Continue)
        .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
        .target_batch_bytes(1024 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(32 * 1024 * 1024)
        // This limit applies to each stream; include every application replica.
        .max_concurrent_batches(1)
        .attempt_timeout(Duration::from_secs(10))
        .on_batch_failure(|event| {
            // A synchronous observer; use a separate diagnostics sink.
            eprintln!("telemetry append failure: {event:?}");
        })
        .build()?;

    let mut dropped_locally = 0_u64;
    for request in 1..=100 {
        if !track(&telemetry, request) {
            dropped_locally += 1;
        }
    }
    println!("locally dropped: {dropped_locally}");

    // In a service, stop and join producer tasks before this settlement barrier.
    let report = telemetry.shutdown().await?;
    if report.outcome != AppendDeliveryOutcome::Ok {
        eprintln!("telemetry loss or ambiguity: {report:?}");
    }
    println!("final stream stats: {:?}", telemetry.stats());

    Ok(())
}

examples/bulk_append.rs (line 35)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // The default Stop policy gives strict barriers. Local memory and request
    // concurrency are bounded independently; concurrent batches have no
    // defined commit order.
    let stream = table
        .append_stream()
        .target_batch_bytes(32 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(1024 * 1024)
        .max_concurrent_batches(4)
        .attempt_timeout(Duration::from_secs(30))
        .build()?;

    let rows = (1..=10_000).map(|id| {
        serde_json::json!({
            "id": id,
            "occurred_at": jiff::Timestamp::now(),
            "name": format!("example-{id}"),
        })
    });

    let admission_error = match stream.send_all(rows).await {
        Ok(result) => {
            println!("accepted locally: {} rows", result.accepted_rows);
            None
        }
        Err(error) => {
            // There is no stream-wide abort or rollback. Settle the already
            // accepted prefix even if row production or admission stopped.
            eprintln!("row admission stopped; settling the accepted prefix: {error}");
            Some(error)
        }
    };

    match stream.shutdown().await {
        Ok(report) => {
            println!("committed remotely: {} rows", report.committed_rows);
            println!("final stream stats: {:?}", stream.stats());
        }
        Err(error) => {
            if matches!(
                error.append_details().map(|details| details.status),
                Some(AppendState::Unknown)
            ) {
                eprintln!("commit outcome is unknown; reconcile before replaying");
            } else {
                eprintln!("bulk append failed: {error}");
            }
            eprintln!("final stream stats: {:?}", stream.stats());
            return Err(error.into());
        }
    }

    if let Some(error) = admission_error {
        return Err(error.into());
    }
    Ok(())
}

pub fn max_retries(self, max_retries: usize) -> Self

pub fn retry_unknown(self, enabled: bool) -> Self

Opts into replaying transient unknown outcomes. This can duplicate rows. Client timeouts are never replayed automatically, even when enabled.

pub fn initial_backoff(self, initial_backoff: Duration) -> Self

pub fn max_backoff(self, max_backoff: Duration) -> Self

pub fn attempt_timeout(self, attempt_timeout: Duration) -> Self

Sets a timeout for each HTTP attempt, overriding the client’s write timeout. A timeout has an unknown commit outcome. By default the client uses 30 seconds. The duration must be finite and nonzero; this deadline cannot be disabled.

Examples found in repository

examples/telemetry.rs (line 59)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Continue mode is explicitly best effort: failed or ambiguous batches are
    // accounted for, released, and do not stop later telemetry batches.
    let telemetry = table
        .append_stream()
        .failure_policy(AppendFailurePolicy::Continue)
        .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
        .target_batch_bytes(1024 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(32 * 1024 * 1024)
        // This limit applies to each stream; include every application replica.
        .max_concurrent_batches(1)
        .attempt_timeout(Duration::from_secs(10))
        .on_batch_failure(|event| {
            // A synchronous observer; use a separate diagnostics sink.
            eprintln!("telemetry append failure: {event:?}");
        })
        .build()?;

    let mut dropped_locally = 0_u64;
    for request in 1..=100 {
        if !track(&telemetry, request) {
            dropped_locally += 1;
        }
    }
    println!("locally dropped: {dropped_locally}");

    // In a service, stop and join producer tasks before this settlement barrier.
    let report = telemetry.shutdown().await?;
    if report.outcome != AppendDeliveryOutcome::Ok {
        eprintln!("telemetry loss or ambiguity: {report:?}");
    }
    println!("final stream stats: {:?}", telemetry.stats());

    Ok(())
}

examples/bulk_append.rs (line 36)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // The default Stop policy gives strict barriers. Local memory and request
    // concurrency are bounded independently; concurrent batches have no
    // defined commit order.
    let stream = table
        .append_stream()
        .target_batch_bytes(32 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(1024 * 1024)
        .max_concurrent_batches(4)
        .attempt_timeout(Duration::from_secs(30))
        .build()?;

    let rows = (1..=10_000).map(|id| {
        serde_json::json!({
            "id": id,
            "occurred_at": jiff::Timestamp::now(),
            "name": format!("example-{id}"),
        })
    });

    let admission_error = match stream.send_all(rows).await {
        Ok(result) => {
            println!("accepted locally: {} rows", result.accepted_rows);
            None
        }
        Err(error) => {
            // There is no stream-wide abort or rollback. Settle the already
            // accepted prefix even if row production or admission stopped.
            eprintln!("row admission stopped; settling the accepted prefix: {error}");
            Some(error)
        }
    };

    match stream.shutdown().await {
        Ok(report) => {
            println!("committed remotely: {} rows", report.committed_rows);
            println!("final stream stats: {:?}", stream.stats());
        }
        Err(error) => {
            if matches!(
                error.append_details().map(|details| details.status),
                Some(AppendState::Unknown)
            ) {
                eprintln!("commit outcome is unknown; reconcile before replaying");
            } else {
                eprintln!("bulk append failed: {error}");
            }
            eprintln!("final stream stats: {:?}", stream.stats());
            return Err(error.into());
        }
    }

    if let Some(error) = admission_error {
        return Err(error.into());
    }
    Ok(())
}

pub fn circuit_breaker( self, circuit_breaker: Option<AppendCircuitBreakerOptions>, ) -> Self

Configures the continue-mode availability circuit, or disables it with None. Enabled by default: five consecutive availability failures pause dispatch for 30 seconds before probing recovery.

Examples found in repository

examples/telemetry.rs (line 52)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Continue mode is explicitly best effort: failed or ambiguous batches are
    // accounted for, released, and do not stop later telemetry batches.
    let telemetry = table
        .append_stream()
        .failure_policy(AppendFailurePolicy::Continue)
        .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
        .target_batch_bytes(1024 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(32 * 1024 * 1024)
        // This limit applies to each stream; include every application replica.
        .max_concurrent_batches(1)
        .attempt_timeout(Duration::from_secs(10))
        .on_batch_failure(|event| {
            // A synchronous observer; use a separate diagnostics sink.
            eprintln!("telemetry append failure: {event:?}");
        })
        .build()?;

    let mut dropped_locally = 0_u64;
    for request in 1..=100 {
        if !track(&telemetry, request) {
            dropped_locally += 1;
        }
    }
    println!("locally dropped: {dropped_locally}");

    // In a service, stop and join producer tasks before this settlement barrier.
    let report = telemetry.shutdown().await?;
    if report.outcome != AppendDeliveryOutcome::Ok {
        eprintln!("telemetry loss or ambiguity: {report:?}");
    }
    println!("final stream stats: {:?}", telemetry.stats());

    Ok(())
}

pub fn on_batch_failure<F>(self, listener: F) -> Self

where F: Fn(&AppendBatchFailureEvent) + Send + Sync + 'static,

Observes remote batch failures. Listener panics are isolated from the worker.

Examples found in repository

examples/telemetry.rs (lines 60-63)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Continue mode is explicitly best effort: failed or ambiguous batches are
    // accounted for, released, and do not stop later telemetry batches.
    let telemetry = table
        .append_stream()
        .failure_policy(AppendFailurePolicy::Continue)
        .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
        .target_batch_bytes(1024 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(32 * 1024 * 1024)
        // This limit applies to each stream; include every application replica.
        .max_concurrent_batches(1)
        .attempt_timeout(Duration::from_secs(10))
        .on_batch_failure(|event| {
            // A synchronous observer; use a separate diagnostics sink.
            eprintln!("telemetry append failure: {event:?}");
        })
        .build()?;

    let mut dropped_locally = 0_u64;
    for request in 1..=100 {
        if !track(&telemetry, request) {
            dropped_locally += 1;
        }
    }
    println!("locally dropped: {dropped_locally}");

    // In a service, stop and join producer tasks before this settlement barrier.
    let report = telemetry.shutdown().await?;
    if report.outcome != AppendDeliveryOutcome::Ok {
        eprintln!("telemetry loss or ambiguity: {report:?}");
    }
    println!("final stream stats: {:?}", telemetry.stats());

    Ok(())
}

pub fn build(self) -> Result<AppendStream, Error>

Examples found in repository

examples/append_stream.rs (line 21)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;
    let stream = table.append_stream().build()?;

    let admitted = stream
        .send_all([
            serde_json::json!({
                "id": 1,
                "occurred_at": jiff::Timestamp::now(),
                "name": "first",
            }),
            serde_json::json!({
                "id": 2,
                "occurred_at": jiff::Timestamp::now(),
                "name": "second",
            }),
        ])
        .await?;
    println!("accepted locally: {} rows", admitted.accepted_rows);

    // send_all() confirms only local admission. With the default strict policy,
    // successful shutdown confirms at least one commit acknowledgement per row.
    // Retrying an unknown outcome can insert duplicates.
    let report = stream.shutdown().await?;
    println!("committed remotely: {} rows", report.committed_rows);

    Ok(())
}

examples/import_ndjson.rs (line 29)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let path = std::env::args()
        .nth(1)
        .ok_or("usage: import_ndjson <events.ndjson>")?;
    let mut input = BufReader::new(File::open(path)?);
    let client = common::client()?;
    let stream = common::write_table(&client)?.append_stream().build()?;
    let admitted: Result<(), Box<dyn std::error::Error>> = async {
        let mut line_number = 0;
        loop {
            // Bound a single input record as well as the stream's queued bytes.
            let mut line = String::new();
            let bytes = input
                .by_ref()
                .take(8 * 1024 * 1024 + 1)
                .read_line(&mut line)?;
            if bytes == 0 {
                break;
            }
            line_number += 1;
            if bytes > 8 * 1024 * 1024 {
                return Err(format!("line {line_number} exceeds 8 MiB").into());
            }
            if line.trim().is_empty() {
                continue;
            }
            let row: serde_json::Value = serde_json::from_str(&line)
                .map_err(|error| format!("line {line_number}: {error}"))?;
            stream.send(&row).await?;
        }
        Ok(())
    }
    .await;
    // Settle the accepted prefix even if parsing or admission failed midway.
    let settled = stream.shutdown().await;
    println!("delivery: {:?}", stream.stats());
    settled?;
    admitted?;
    Ok(())
}

examples/telemetry.rs (line 64)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // Continue mode is explicitly best effort: failed or ambiguous batches are
    // accounted for, released, and do not stop later telemetry batches.
    let telemetry = table
        .append_stream()
        .failure_policy(AppendFailurePolicy::Continue)
        .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
        .target_batch_bytes(1024 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(32 * 1024 * 1024)
        // This limit applies to each stream; include every application replica.
        .max_concurrent_batches(1)
        .attempt_timeout(Duration::from_secs(10))
        .on_batch_failure(|event| {
            // A synchronous observer; use a separate diagnostics sink.
            eprintln!("telemetry append failure: {event:?}");
        })
        .build()?;

    let mut dropped_locally = 0_u64;
    for request in 1..=100 {
        if !track(&telemetry, request) {
            dropped_locally += 1;
        }
    }
    println!("locally dropped: {dropped_locally}");

    // In a service, stop and join producer tasks before this settlement barrier.
    let report = telemetry.shutdown().await?;
    if report.outcome != AppendDeliveryOutcome::Ok {
        eprintln!("telemetry loss or ambiguity: {report:?}");
    }
    println!("final stream stats: {:?}", telemetry.stats());

    Ok(())
}

examples/bulk_append.rs (line 37)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = common::client()?;
    let table = common::write_table(&client)?;

    // The default Stop policy gives strict barriers. Local memory and request
    // concurrency are bounded independently; concurrent batches have no
    // defined commit order.
    let stream = table
        .append_stream()
        .target_batch_bytes(32 * 1024)
        .max_batch_rows(10_000)
        .flush_interval(Duration::from_secs(1))
        .max_buffered_bytes(1024 * 1024)
        .max_concurrent_batches(4)
        .attempt_timeout(Duration::from_secs(30))
        .build()?;

    let rows = (1..=10_000).map(|id| {
        serde_json::json!({
            "id": id,
            "occurred_at": jiff::Timestamp::now(),
            "name": format!("example-{id}"),
        })
    });

    let admission_error = match stream.send_all(rows).await {
        Ok(result) => {
            println!("accepted locally: {} rows", result.accepted_rows);
            None
        }
        Err(error) => {
            // There is no stream-wide abort or rollback. Settle the already
            // accepted prefix even if row production or admission stopped.
            eprintln!("row admission stopped; settling the accepted prefix: {error}");
            Some(error)
        }
    };

    match stream.shutdown().await {
        Ok(report) => {
            println!("committed remotely: {} rows", report.committed_rows);
            println!("final stream stats: {:?}", stream.stats());
        }
        Err(error) => {
            if matches!(
                error.append_details().map(|details| details.status),
                Some(AppendState::Unknown)
            ) {
                eprintln!("commit outcome is unknown; reconcile before replaying");
            } else {
                eprintln!("bulk append failed: {error}");
            }
            eprintln!("final stream stats: {:?}", stream.stats());
            return Err(error.into());
        }
    }

    if let Some(error) = admission_error {
        return Err(error.into());
    }
    Ok(())
}