nusadb — native Rust driver + ORM for NusaDB

nusadb is a fast, stable, dependency-light Rust client for NusaDB. It speaks the Nusa Wire Protocol (PROTOCOL_VERSION 1.1) directly over TCP — it does not link the engine, use C bindings, or pull in an async runtime.

• Fast — buffered framing, native value decoding, pooled connections, nodelay sockets.
• Stable — no panics in the library path, bounds-checked decoding, constant-time SCRAM verification, explicit error type.
• Complete — a driver is a SQL-text transport, so it supports every query the server accepts: any DDL/DML/SELECT, joins, CTEs, window functions, subqueries, set-ops, transactions, stored procedures/functions, and all 16 wire value types decoded to native Rust.

Standalone crate: it has its own [workspace], so it builds and publishes independently of the NusaDB engine workspace.

Install

[dependencies]
nusadb = "0.1"                                            # sync, plaintext (default)
# nusadb = { version = "0.1", features = ["tls"] }        # + sync TLS
# nusadb = { version = "0.1", features = ["async"] }      # + Tokio async API
# nusadb = { version = "0.1", features = ["async-tls"] }  # + async over TLS

Features

Quick start

use nusadb::Connection;

let mut conn = Connection::connect("nusadb://nusa-root@127.0.0.1:5678/nusadb")?;

conn.execute("CREATE TABLE users (id INT NOT NULL, name TEXT, PRIMARY KEY (id))")?;
conn.query_params("INSERT INTO users VALUES ($1, $2)", &[&1_i64, &"alice"])?;

let result = conn.query("SELECT id, name FROM users ORDER BY id")?;
for row in &result.rows {
    let id: i64 = row.get(0)?;
    let name: String = row.get_by_name("name")?;
    println!("{id}: {name}");
}
# Ok::<(), nusadb::Error>(())

Parameters & types

Parameters are positional ($1, $2, …) and bind any ToSql type; Option<T> is NULL:

conn.query_params(
    "INSERT INTO t (id, name, active) VALUES ($1, $2, $3)",
    &[&1_i64, &Option::<&str>::None, &true],
)?;

Results decode to native Rust via the protocol-1.1 typed column metadata:

result.column_types exposes each column's TypeTag (like JDBC getColumnTypeName).

BYTEA round-trips as raw bytes: a Vec<u8> / &[u8] parameter binds as the \x<hex> text form (coerced into the BYTEA column), and a BYTES column decodes back to Value::Bytes / Vec<u8>.

Prepared statements

let stmt = conn.prepare("SELECT name FROM users WHERE id = $1")?;
let a = conn.query_prepared(&stmt, &[&1_i64])?;
let b = conn.query_prepared(&stmt, &[&2_i64])?;

Transactions

conn.transaction(|tx| {
    tx.execute("UPDATE accounts SET balance = balance - 10 WHERE id = 1")?;
    tx.execute("UPDATE accounts SET balance = balance + 10 WHERE id = 2")?;
    Ok(())
})?; // COMMIT on Ok, ROLLBACK on Err

Inside an open transaction, savepoint(name) marks a point you can later undo to with rollback_to_savepoint(name) (the transaction stays open) or forget with release_savepoint(name). The async AsyncConnection has the same three methods.

Notifications (LISTEN/NOTIFY)

listen(channel) subscribes the connection; a notify(channel, payload) from any connection on the same database is then delivered asynchronously. poll_notification(timeout) waits for the next one (a buffered one returns immediately; None blocks), or notifications() drains those buffered while reading other responses. AsyncConnection has the same methods.

conn.listen("orders")?;
// ... elsewhere: other.notify("orders", Some("42"))?;
if let Some(note) = conn.poll_notification(Some(Duration::from_secs(5)))? {
    println!("{} {}", note.channel, note.payload);
}
conn.unlisten("orders")?;

Connection pool

use nusadb::{Pool, Config};

let pool = Pool::new(Config::from_url("nusadb://nusa-root@127.0.0.1:5678/nusadb")?, 16)?;
let mut conn = pool.get()?;            // blocks if the pool is saturated
let n = conn.query("SELECT count(*) FROM users")?.rows.len();
// `conn` returns to the pool on drop.

ORM / query builder

A thin, native builder (double-quoted identifiers, $n params, constant LIMIT/OFFSET, RETURNING *). Covers the full SELECT surface — DISTINCT, joins, comparison/IN/LIKE/ BETWEEN/NULL predicates, GROUP BY + HAVING, set operations (UNION/INTERSECT/EXCEPT), and aggregate terminals. Drop to raw SQL any time.

use nusadb::orm::{Select, Insert, Update, Delete, FromRow};
use nusadb::{Row, Result};

Insert::into("users").set("id", 3_i64).set("name", "carol").run(&mut conn)?;

let rows = Select::from("users")
    .columns(&["id", "name"])
    .filter("id", 3_i64)
    .order_by("id", false)
    .limit(10)
    .offset(20)         // pagination — constant OFFSET, inlined for the server
    .fetch(&mut conn)?;

// DISTINCT, joins (inner/left/right/full/cross), and aggregate terminals.
let cities = Select::from("users").distinct().columns(&["city"]).fetch(&mut conn)?;
let joined = Select::from("orders")
    .columns(&["id"])
    .inner_join("users", "\"orders\".\"user_id\" = \"users\".\"id\"")
    .fetch(&mut conn)?;
let n = Select::from("users").count(&mut conn)?;          // also: sum / avg / min / max
let avg_id = Select::from("users").avg("id", &mut conn)?;

// Predicates, GROUP BY + HAVING, and set operations.
let adults = Select::from("users")
    .gte("age", 18_i64)
    .where_in("city", &["NY", "LA"])
    .is_not_null("email")
    .fetch(&mut conn)?;
let by_city = Select::from("users")
    .select_raw(&["city", "count(*) AS n"])
    .group_by(&["city"])
    .having("count(*) > $1", &[1_i64.into()])
    .fetch(&mut conn)?;
let combined = Select::from("a").columns(&["id"])
    .union(Select::from("b").columns(&["id"]))
    .order_by("id", false)
    .fetch(&mut conn)?;

Update::table("users").set("name", "carol2").filter("id", 3_i64).run(&mut conn)?;
Delete::from("users").filter("id", 3_i64).run(&mut conn)?;

// Map rows into your own structs:
struct User { id: i64, name: String }
impl FromRow for User {
    fn from_row(row: &Row) -> Result<Self> {
        Ok(Self { id: row.get(0)?, name: row.get(1)? })
    }
}
let users: Vec<User> = Select::from("users").fetch_as(&mut conn)?;

TLS (feature tls)

The Nusa Wire Protocol uses implicit TLS 1.3 (rustls, ring provider): the client opens TLS immediately, then handshakes inside the encrypted stream. Use the nusadbs:// scheme, or set Config::tls:

use nusadb::{Config, Connection, TlsConfig};

// Public roots (Mozilla/webpki):
let mut conn = Connection::connect("nusadbs://app@db.example.com:5678/nusadb")?;

// Private CA:
let mut cfg = Config::from_url("nusadb://app@db.internal:5678/nusadb")?;
cfg.tls = Some(TlsConfig::with_ca_pem(std::fs::read("ca.pem")?));
let mut conn = Connection::connect_config(&cfg)?;

// Dev only — skip verification (never in production):
cfg.tls = Some(TlsConfig::danger_insecure());
# Ok::<(), nusadb::Error>(())

Async (feature async)

AsyncConnection / AsyncPool mirror the sync API on Tokio (add async-tls for TLS):

# async fn demo() -> nusadb::Result<()> {
use nusadb::{AsyncConnection, AsyncPool, Config};

let mut conn = AsyncConnection::connect("nusadb://nusa-root@127.0.0.1:5678/nusadb").await?;
conn.execute("CREATE TABLE t (id INT NOT NULL, PRIMARY KEY (id))").await?;
conn.query_params("INSERT INTO t VALUES ($1)", &[&1_i64]).await?;
let rows = conn.query("SELECT id FROM t").await?;

conn.begin().await?;
conn.execute("INSERT INTO t VALUES (2)").await?;
conn.commit().await?;

let pool = AsyncPool::new(Config::from_url("nusadb://nusa-root@127.0.0.1:5678/nusadb")?, 16)?;
let mut c = pool.get().await?;            // awaits a free slot; returns to the pool on drop
let _ = c.query("SELECT 1").await?;
# Ok(()) }

Authentication

• Trust (dev/local): no password — pass password: None.
• SCRAM-SHA-256 (RFC 7677): set password; the driver runs the full SASL exchange and verifies the server's signature in constant time (mutual auth).

Cancellation

let handle = conn.cancel_handle();   // (pid, secret) from BackendKeyData
// from another thread/connection, to abort an in-flight statement:
handle.cancel()?;

Bulk load / export (COPY)

copy_in / copy_out stream the COPY sub-protocol — one round-trip for the whole dataset instead of a row per insert. You write the COPY statement (with any WITH (...) options) and move bytes in the server's text format: tab-delimited fields, \N for SQL NULL, newline per row.

use std::io::Cursor;

// Bulk load: stream any Read of text-format rows.
let mut rows = Cursor::new("1\talice\n2\t\\N\n3\tcarol\n");
let loaded = conn.copy_in("COPY users (id, name) FROM STDIN", &mut rows)?;
assert_eq!(loaded, 3);

// Bulk export: collect the rows into any Write.
let mut out = Vec::new();
let exported = conn.copy_out("COPY users TO STDOUT", &mut out)?;
# Ok::<(), nusadb::Error>(())

If the source fails mid-load the driver sends CopyFail, and a COPY the server refuses (bad SQL, an RLS-protected table) returns an error — either way the connection is left ready for the next statement. AsyncConnection exposes the same copy_in / copy_out over Tokio (an AsyncRead source, an AsyncWrite sink) with the async feature.

Status & roadmap

Supported today: trust + SCRAM auth, TLS 1.3 (feature tls), simple & extended (parameterized/prepared) queries, all value types, transactions (BEGIN/COMMIT/ROLLBACK), connection pooling, cancellation, COPY FROM/TO (bulk load/export, sync and async), the ORM builders, and a full async API on Tokio (features async / async-tls).

Not yet: binary result formats (text is used throughout — lossless for every type). An additive follow-up; it does not change the surface above.

Testing

cargo build -p nusadb-server                    # from the repo root, build the server the tests boot
cd drivers/rust
cargo test                                      # sync tests (skip cleanly if the binary is absent)
cargo test --features "async tls"               # also run the async + TLS-config tests

License

Apache-2.0.