server-connection-rust

name	server-connection-rust
summary	Configure and optimize Couchbase SDK connections in Rust — async/tokio cluster setup, singleton pattern, wait_until_ready, timeouts, KV CRUD (upsert/insert/get/replace/remove), expiry, durability, sub-document operations
description	Configure and optimize Couchbase SDK connections in Rust — async/tokio cluster setup, singleton pattern, wait_until_ready, timeouts, KV CRUD (upsert/insert/get/replace/remove), expiry, durability, sub-document operations
compatibility	Rust SDK 1.0+. Requires tokio async runtime. No transactions or KV range scan support yet.
metadata	{"last_verified":"2026-05","min_server_version":"7.0","handoff":[{"condition":"user asks about SQL++ queries","skill":"server-querying-rust"},{"condition":"user asks about CAS, bulk ops, sub-document, or SDK patterns","skill":"sdk-patterns-rust"},{"condition":"user asks about full-text search or vector search","skill":"search-rust"},{"condition":"user asks about testing or mocking Couchbase","skill":"testing-patterns-rust"},{"condition":"user asks about field-level encryption","skill":"fle-rust"},{"condition":"user asks about transactions or atomic multi-document operations","skill":"transactions-rust"},{"condition":"user asks about the Analytics service or OLAP queries","skill":"analytics-rust"}]}

Couchbase Server Connection — Rust

Platform-agnostic concepts (connection string formats, timeout semantics, durability, sub-document, troubleshooting): shared/server/sdk-connection-concepts.md

See templates/sdk-connection-rust.rs for the full standalone connection template. Rust SDK 1.x is async-first, built on tokio. Most operations return Result<T, couchbase::error::Error>.

Cargo.toml

[dependencies]
couchbase = "1"
tokio = { version = "1", features = ["full"] }
serde_json = "1"
serde = { version = "1", features = ["derive"] }
uuid = { version = "1", features = ["v4"] }

Connect

use couchbase::authenticator::{Authenticator, PasswordAuthenticator};
use couchbase::cluster::Cluster;
use couchbase::collection::Collection;
use couchbase::options::cluster_options::ClusterOptions;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let cluster = Cluster::connect(
        "couchbase://localhost",
        ClusterOptions::new(Authenticator::PasswordAuthenticator(
            PasswordAuthenticator::new(
                &std::env::var("CB_USERNAME")?,
                &std::env::var("CB_PASSWORD")?,
            ),
        )),
    )
    .await?;

    let bucket = cluster.bucket("travel-sample");
    bucket.wait_until_ready(None).await?;

    let collection: Collection = bucket.scope("inventory").collection("airline");
    Ok(())
}

Capella (TLS)

// Use couchbases:// — Capella certificates are bundled in the SDK
let cluster = Cluster::connect(
    "couchbases://cb.<your-endpoint>.cloud.couchbase.com",
    ClusterOptions::new(Authenticator::PasswordAuthenticator(
        PasswordAuthenticator::new(
            &std::env::var("CB_USERNAME")?,
            &std::env::var("CB_PASSWORD")?,
        ),
    )),
)
.await?;

Singleton pattern

Wrap in Arc for sharing across tasks:

use std::sync::Arc;

pub struct CouchbasePool {
    pub cluster: Arc<Cluster>,
}

impl CouchbasePool {
    pub async fn new(conn_str: &str, user: &str, pass: &str) -> Result<Self, couchbase::error::Error> {
        let cluster = Cluster::connect(
            conn_str,
            ClusterOptions::new(Authenticator::PasswordAuthenticator(
                PasswordAuthenticator::new(user, pass),
            )),
        )
        .await?;
        Ok(Self { cluster: Arc::new(cluster) })
    }
}

KV Operations

Upsert / Insert / Get / Replace / Remove

use couchbase::options::kv_options::ReplaceOptions;
use serde_json::json;

// Upsert — insert or overwrite
collection.upsert("airline::1001", json!({"name": "Rusty Air", "country": "US"}), None).await?;

// Insert — fails if key exists
collection.insert("airline::1002", json!({"name": "Oxide Airways"}), None).await?;

// Get
let result = collection.get("airline::1001", None).await?;
let doc: serde_json::Value = result.content_as()?;

// Replace with CAS (optimistic locking)
let get_result = collection.get("airline::1001", None).await?;
let mut doc: serde_json::Value = get_result.content_as()?;
doc["country"] = json!("CA");
collection.replace(
    "airline::1001",
    doc,
    ReplaceOptions::new().cas(get_result.cas()),
).await?;

// Remove
collection.remove("airline::1001", None).await?;

Expiry (TTL)

use couchbase::options::kv_options::{UpsertOptions, GetOptions};

collection.upsert(
    "session::abc",
    json!({"user": "alice"}),
    UpsertOptions::new().expiry(Duration::from_secs(3600)),
).await?;

// Get with expiry reset
collection.get(
    "session::abc",
    GetOptions::new().with_expiry(true),
).await?;

Durability

use couchbase::durability_level::DurabilityLevel;

collection.upsert(
    "order::99",
    json!({"status": "confirmed"}),
    UpsertOptions::new().durability_level(DurabilityLevel::MAJORITY),
).await?;

See shared/server/durability.md for guidance on when to use each level.

Sub-Document Operations

use couchbase::kv::MutateInSpec;
use couchbase::kv::LookupInSpec;

// Read specific fields
let result = collection.lookup_in("airline::1001", &[
    LookupInSpec::get("name", None)?,
    LookupInSpec::exists("country", None)?,
], None).await?;

let name: String = result.content_as(0)?;
let has_country = result.exists(1);

// Mutate specific fields
collection.mutate_in("airline::1001", &[
    MutateInSpec::upsert("status", "active", None)?,
    MutateInSpec::increment("view_count", 1, None)?,
], None).await?;

Error Handling

use couchbase::error::ErrorKind;

match collection.get("missing::key", None).await {
    Ok(result) => {
        // use result
    }
    Err(e) => match e.kind() {
        ErrorKind::DocumentNotFound => println!("Document does not exist"),
        ErrorKind::DocumentExists => println!("Document already exists"),
        ErrorKind::CasMismatch => println!("Concurrent modification, retry"),
        ErrorKind::ServerTimeout => println!("Operation timed out"),
        _ => println!("Error: {e}"),
    },
}

Capella / WAN Profile

When connecting to Capella or any remote cluster over WAN, apply the wan_development config profile to increase all timeouts for cloud latency:

let mut opts = couchbase::ClusterOptions::default();
opts.apply_profile("wan_development");

let cluster = couchbase::Cluster::connect(
    "couchbases://cb.<your-endpoint>.cloud.couchbase.com",
    opts,
).await?;

Timeouts

The Rust SDK does not expose a generic per-operation timeout field. Wrap the future with tokio::time::timeout if you want an application deadline:

use tokio::time::timeout;

let result = timeout(
    Duration::from_millis(2500),
    collection.get("airline::1001", None),
).await??;  // outer ? = timeout error, inner ? = SDK error

See references/deployment-scenarios.md for scenario-specific timeout values.

Common Errors

Error	Cause	Action
ErrorKind::DocumentNotFound	Document does not exist	Check key; treat it as a missing document
ErrorKind::DocumentExists	Insert on existing key	Use upsert or check first
ErrorKind::CasMismatch	Concurrent modification	Re-fetch and retry
ErrorKind::ServerTimeout	Server did not respond before timeout	Retry; check cluster health
ErrorKind::AuthenticationFailure	Wrong credentials or missing RBAC role	Check user roles in Couchbase UI

use couchbase::error::ErrorKind;

match collection.get("key", None).await {
    Ok(result) => {
        // use result
    }
    Err(e) => match e.kind() {
        ErrorKind::DocumentNotFound => {
            // handle missing
        }
        ErrorKind::ServerTimeout => {
            // retry after checking cluster health
        }
        _ => return Err(e.into()),
    },
}

Common Mistakes

• Import Collection from couchbase::collection::Collection, not from the crate root.
• Use bucket.wait_until_ready(None).await? for the default readiness check.
• ErrorKind::DocumentNotFound is the KV miss case; ErrorKind::UserNotFound is for management APIs.
• For session TTLs, set expiry(...) on write and use a separate sessions collection.

Transactions

Rust SDK 1.0 does not support distributed ACID transactions. There is no cluster.transactions() API.

For atomic multi-document updates, use optimistic locking with CAS — read a document, modify it, then replace using the CAS value from the read. If another writer modified the document concurrently, the replace fails with ErrorKind::CasMismatch and you retry.

See sdk-patterns-rust for the full CAS pattern.