Skip to content

perf(storage): recover the WAL without blocking the async runtime #2011

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zonotope merged 14 commits into from
Oct 6, 2026
Merged

perf(storage): recover the WAL without blocking the async runtime #2011

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
f1698e1
keep file operation guards on the same thread that took them
zonotope Oct 1, 2026
277d274
separate lock acquisition from wal recovery
zonotope Oct 1, 2026
ff22250
add async wal recovery fn without blocking the caller
zonotope Oct 1, 2026
fd4122c
recover the WAL asynchronously when opening an async file connection
zonotope Oct 1, 2026
403c1e0
recover the wal asynchronously on the async build paths
zonotope Oct 1, 2026
5aaa1da
open the cli's file-backed instance with build_async
zonotope Oct 1, 2026
9617f8b
update docs
zonotope Oct 1, 2026
1e1ae69
extract repeated join handling into a helper fn
zonotope Oct 1, 2026
4767b4e
remove unnecessary comments
zonotope Oct 2, 2026
305435d
put logged-write crash setup in a FileStorage test hook
zonotope Oct 2, 2026
6c33627
Merge branch 'fix/cas-deadlock' into perf/async-wal-recovery
zonotope Oct 2, 2026
84f2361
Merge remote-tracking branch 'origin/main' into perf/async-wal-recovery
zonotope Oct 5, 2026
e1ac529
make the compiler ensure that OperationHold is !Send
zonotope Oct 6, 2026
58adfcf
fix file-builder examples in async docs
zonotope Oct 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions docs/contributing/benches.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -362,12 +362,13 @@ The fix is to wrap setup work that touches the runtime in `rt.block_on`:
let rt = bench_runtime();

b.iter_batched(
// setup — wrap in block_on so the reactor is alive while
// FlureeBuilder::file(...).build() runs.
// Setup runs in block_on, so the reactor is alive while
// FlureeBuilder::file(...).build_async() runs.
|| rt.block_on(async {
let dir = tempfile::tempdir().unwrap();
let fluree = FlureeBuilder::file(dir.path().to_string_lossy().to_string())
.build()
.build_async()
.await
.unwrap();
(dir, fluree)
}),
Expand Down
42 changes: 23 additions & 19 deletions docs/getting-started/rust-api.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ use fluree_db_api::{FlureeBuilder, Result};
#[tokio::main]
async fn main() -> Result<()> {
// Use file-backed storage for persistence
let fluree = FlureeBuilder::file("./data").build()?;
let fluree = FlureeBuilder::file("./data").build_async().await?;

// Create a new ledger (or load an existing one)
let ledger = fluree.create_ledger("mydb").await?;
Expand All @@ -84,6 +84,9 @@ async fn main() -> Result<()> {
}
```

`build_async()` replays the storage's write-ahead log without blocking the async runtime.
`build()` builds the same instance from synchronous code that has entered a Tokio runtime (for example with `Runtime::enter`), replaying the log on the calling thread.

### Bulk import (high throughput)

For initial ledger bootstraps (large Turtle or JSON-LD datasets), Fluree exposes a bulk import
Expand All @@ -94,7 +97,7 @@ use fluree_db_api::{FlureeBuilder, Result};

#[tokio::main]
async fn main() -> Result<()> {
let fluree = FlureeBuilder::file("./data").build()?;
let fluree = FlureeBuilder::file("./data").build_async().await?;

// `chunks_dir` can be:
// - a directory containing *.ttl, *.trig, or *.jsonld files (sorted lexicographically), OR
Expand Down Expand Up @@ -423,7 +426,7 @@ use tokio::sync::mpsc;

#[tokio::main]
async fn main() -> Result<()> {
let fluree = Arc::new(FlureeBuilder::file("./data").build()?);
let fluree = Arc::new(FlureeBuilder::file("./data").build_async().await?);

// Plan against a borrowed GraphDb, then move the owned LedgerState into the
// spawned producer (GraphDb borrows the state, so plan first).
Expand Down Expand Up @@ -540,7 +543,7 @@ use fluree_db_api::{

#[tokio::main]
async fn main() -> Result<()> {
let fluree = FlureeBuilder::file("./data").build()?;
let fluree = FlureeBuilder::file("./data").build_async().await?;

// Get a cached ledger handle
let handle = fluree.ledger_cached("mydb:main").await?;
Expand Down Expand Up @@ -660,7 +663,7 @@ use std::fs::File;

#[tokio::main]
async fn main() -> Result<()> {
let fluree = FlureeBuilder::file("./data").build()?;
let fluree = FlureeBuilder::file("./data").build_async().await?;

// Export as Turtle to a file
let file = File::create("backup.ttl").unwrap();
Expand Down Expand Up @@ -751,7 +754,7 @@ use fluree_db_api::{FlureeBuilder, Result};
#[tokio::main]
async fn main() -> Result<()> {
// Caching is on by default — no extra call needed
let fluree = FlureeBuilder::file("./data").build()?;
let fluree = FlureeBuilder::file("./data").build_async().await?;

// First call loads from storage
let ledger = fluree.ledger("mydb:main").await?;
Expand All @@ -768,7 +771,8 @@ To **disable** caching (e.g., for a CLI tool that runs once and exits):
```rust
let fluree = FlureeBuilder::file("./data")
.without_ledger_caching()
.build()?;
.build_async()
.await?;
```

#### Disconnecting Ledgers
Expand All @@ -780,7 +784,7 @@ use fluree_db_api::{FlureeBuilder, Result};

#[tokio::main]
async fn main() -> Result<()> {
let fluree = FlureeBuilder::file("./data").build()?;
let fluree = FlureeBuilder::file("./data").build_async().await?;

// Load and use ledger
let ledger = fluree.ledger("mydb:main").await?;
Expand Down Expand Up @@ -814,7 +818,7 @@ use fluree_db_api::{FlureeBuilder, Result};

#[tokio::main]
async fn main() -> Result<()> {
let fluree = FlureeBuilder::file("./data").build()?;
let fluree = FlureeBuilder::file("./data").build_async().await?;

// Check if ledger exists (lightweight nameservice lookup)
if fluree.ledger_exists("mydb:main").await? {
Expand Down Expand Up @@ -850,7 +854,7 @@ use fluree_db_api::{FlureeBuilder, DropMode, DropStatus, Result};

#[tokio::main]
async fn main() -> Result<()> {
let fluree = FlureeBuilder::file("./data").build()?;
let fluree = FlureeBuilder::file("./data").build_async().await?;

// Soft drop: retract every branch in the nameservice, preserve artifacts
let report = fluree.drop_ledger("mydb", DropMode::Soft).await?;
Expand Down Expand Up @@ -949,7 +953,7 @@ use fluree_db_api::{FlureeBuilder, NotifyResult, RefreshOpts, Result};

#[tokio::main]
async fn main() -> Result<()> {
let fluree = FlureeBuilder::file("./data").build()?;
let fluree = FlureeBuilder::file("./data").build_async().await?;

// Load ledger into cache
let _ledger = fluree.ledger_cached("mydb:main").await?;
Expand Down Expand Up @@ -1022,7 +1026,7 @@ use serde_json::json;

#[tokio::main]
async fn main() -> Result<()> {
let fluree = FlureeBuilder::file("./data").build()?;
let fluree = FlureeBuilder::file("./data").build_async().await?;
let handle = fluree.ledger_cached("mydb:main").await?;

// Transaction returns the commit's t value
Expand Down Expand Up @@ -1399,7 +1403,7 @@ async fn main() -> Result<()> {
"https://acme-fluree.example.com",
Some("eyJhbG...".to_string()),
)
.build()?;
.build_async().await?;

let db = fluree.view("local-ledger:main").await?;

Expand Down Expand Up @@ -1468,7 +1472,7 @@ use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() -> Result<()> {
let fluree = Arc::new(FlureeBuilder::file("./data").build()?);
let fluree = Arc::new(FlureeBuilder::file("./data").build_async().await?);

// Start background indexer
let indexer = BackgroundIndexerWorker::new(
Expand Down Expand Up @@ -1718,7 +1722,7 @@ async fn test_persistence() -> Result<()> {

// Create ledger and write data
{
let fluree = FlureeBuilder::file(path).build()?;
let fluree = FlureeBuilder::file(path).build_async().await?;
let ledger = fluree.create_ledger("test").await?;

let data = json!({"@context": {}, "@graph": [{"@id": "ex:test"}]});
Expand All @@ -1731,7 +1735,7 @@ async fn test_persistence() -> Result<()> {

// Verify persistence by reopening
{
let fluree = FlureeBuilder::file(path).build()?;
let fluree = FlureeBuilder::file(path).build_async().await?;
let ledger = fluree.ledger("test:main").await?;

assert!(ledger.t() > 0);
Expand Down Expand Up @@ -2021,7 +2025,7 @@ use serde_json::json;
#[tokio::main]
async fn main() -> Result<()> {
// Caching is on by default (required for stage)
let fluree = FlureeBuilder::file("./data").build()?;
let fluree = FlureeBuilder::file("./data").build_async().await?;

// Get a cached handle
let handle = fluree.ledger_cached("mydb:main").await?;
Expand Down Expand Up @@ -2150,7 +2154,7 @@ use serde_json::json;

#[tokio::main]
async fn main() -> Result<()> {
let fluree = FlureeBuilder::file("./data").build()?;
let fluree = FlureeBuilder::file("./data").build_async().await?;

// Get ledger info with optional context for IRI compaction
let context = json!({
Expand Down Expand Up @@ -2225,7 +2229,7 @@ use serde_json::json;

#[tokio::main]
async fn main() -> Result<()> {
let fluree = FlureeBuilder::file("./data").build()?;
let fluree = FlureeBuilder::file("./data").build_async().await?;

// Find all ledgers on main branch
let query = json!({
Expand Down
2 changes: 1 addition & 1 deletion docs/graph-sources/overview.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,7 +190,7 @@ Graph sources are created and registered via the `fluree-db-api` Rust API, which
```rust
use fluree_db_api::{FlureeBuilder, R2rmlCreateConfig};

let fluree = FlureeBuilder::default().build().await?;
let fluree = FlureeBuilder::file("/path/to/data").build_async().await?;

let config = R2rmlCreateConfig::new_direct(
"execution-log",
Expand Down
2 changes: 1 addition & 1 deletion docs/graph-sources/r2rml.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ If you use **Direct S3** mode, Fluree resolves the current Iceberg metadata by r
```rust
use fluree_db_api::{FlureeBuilder, R2rmlCreateConfig};

let fluree = FlureeBuilder::default().build().await?;
let fluree = FlureeBuilder::file("/path/to/data").build_async().await?;

let config = R2rmlCreateConfig::new_direct(
"airlines-rdf",
Expand Down
2 changes: 1 addition & 1 deletion docs/indexing-and-search/bm25.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ The CLI (`fluree bm25 create/list/sync/drop`) drives a server when one is reacha
use fluree_db_api::{Bm25CreateConfig, FlureeBuilder};
use serde_json::json;

let fluree = FlureeBuilder::file("/path/to/data").build()?;
let fluree = FlureeBuilder::file("/path/to/data").build_async().await?;

// Create a ledger and insert some data
let ledger = fluree.create_ledger("docs:main").await?;
Expand Down
4 changes: 2 additions & 2 deletions docs/indexing-and-search/reindex.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ use fluree_db_api::{FlureeBuilder, ReindexOptions, ReindexResult};

// Create Fluree instance
let fluree = FlureeBuilder::file("/path/to/data")
.build()
.build_async()
.await?;

// Reindex with default options
Expand All @@ -55,7 +55,7 @@ println!("Root ID: {}", result.root_id);
use fluree_db_api::{FlureeBuilder, ReindexOptions};
use fluree_db_indexer::IndexerConfig;

let fluree = FlureeBuilder::file("/path/to/data").build().await?;
let fluree = FlureeBuilder::file("/path/to/data").build_async().await?;

@aaj3f aaj3f Oct 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Optional — the same build().await shape is in two more pages, and two file-backed build()? examples were missed.

docs/graph-sources/overview.md:193 and docs/graph-sources/r2rml.md:29 both have let fluree = FlureeBuilder::default().build().await?;. Like the reindex examples, .await on a Result doesn't compile — and default() has no storage path, so even a sync build() there returns "File storage requires a path". FlureeBuilder::file("/path/to/data").build_async().await? would match this page.

Two more still call build()? from what reads as async code: the without_ledger_caching() example at docs/getting-started/rust-api.md:772-774, and the remote-connection example at docs/operations/configuration.md:1129-1132 (the same example this PR converted at rust-api.md:1402-1408).

Four small edits, so if you agree, I'd rather see them in this PR than lost in the backlog.

Commenting here because those files are not in this diff.


let result = fluree.reindex("mydb:main", ReindexOptions::default()
// Use custom index node sizes
Expand Down
3 changes: 2 additions & 1 deletion docs/operations/configuration.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1129,7 +1129,8 @@ Register remote connections on the `FlureeBuilder`:
let fluree = FlureeBuilder::file("./data")
.remote_connection("acme", "https://acme-fluree.example.com", Some(token))
.remote_connection("partner", "https://partner.example.com", None)
.build()?;
.build_async()
.await?;
```

Each call registers a named connection. The name is used in SPARQL queries:
Expand Down
4 changes: 2 additions & 2 deletions docs/operations/running-fluree.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -315,7 +315,7 @@ use serde_json::json;
let fluree = FlureeBuilder::memory().build_memory();

// File-based persistence — typed
let fluree = FlureeBuilder::file("./data").build()?;
let fluree = FlureeBuilder::file("./data").build_async().await?;

// AWS S3 (requires `aws` feature) — typed
let fluree = FlureeBuilder::s3("my-bucket", "https://s3.us-east-1.amazonaws.com")
Expand All @@ -337,7 +337,7 @@ Quick setup with typed builders:
use fluree_db_api::FlureeBuilder;

let fluree = FlureeBuilder::memory().build_memory(); // In-memory
let fluree = FlureeBuilder::file("./data").build()?; // File-based
let fluree = FlureeBuilder::file("./data").build_async().await?; // File-based
let fluree = FlureeBuilder::s3("bucket", "endpoint").build_client().await?; // S3
```

Expand Down
5 changes: 3 additions & 2 deletions docs/security/encryption.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,11 +58,12 @@ let client = FlureeBuilder::from_json_ld(&config)?
// and the id that encrypts new writes.
let fluree = FlureeBuilder::file("/data/fluree")
.with_encryption_keys(vec![(1, old_key), (2, new_key)], 2)?
.build()?;
.build_async()
.await?;
```

A key set on the builder (`with_encryption_key*()`, `with_encryption_keys()`,
or `AES256Key` / `AES256Keys` in JSON-LD) is applied by every terminal build method — `build()`, `build_memory()`, `build_s3()`,
or `AES256Key` / `AES256Keys` in JSON-LD) is applied by every terminal build method: `build()`, `build_async()`, `build_memory()`, `build_s3()`,
`build_client()` and friends — on every backend. The `build_*_encrypted()` methods
remain for callers that want the key to be an explicit argument;
`build_encrypted(key)` replaces any configured key set with that one key, as
Expand Down
Loading