feltdb 0.6.1

Embedded durable state store: an append-only log with atomic multi-record transactions, version-conditional writes, and idempotent retries
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

//! What a change costs a Rust live subscription, as the subscribed collection
//! grows: the time from a write to every subscriber holding its new result.
//! `rescan_ms` is one full recomputation of a subscription's result -- what
//! every change cost every subscription before results were kept by record.
//!
//! ```text
//! cargo run --release -p feltdb --example subscription_cost
//! ```
use feltdb::FeltDb;
use serde::{Deserialize, Serialize};
use std::time::Instant;

#[derive(Serialize, Deserialize, Clone)]
struct Item {
    n: u64,
    active: bool,
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    for size in [1_000u64, 10_000, 100_000] {
        let directory = std::env::temp_dir().join(format!("feltdb-subscription-cost-{size}-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&directory);
        std::fs::create_dir_all(&directory).unwrap();
        let db = FeltDb::open(directory.join("state.log")).unwrap();
        for n in 0..size {
            db.insert(&format!("items:{n}"), Item { n, active: n % 10 == 0 }).unwrap();
        }
        let subscribers = 10;
        let mut live = Vec::new();
        for _ in 0..subscribers {
            live.push(db.capability("items").live_find::<Item, _>(|item| item.active).await.unwrap());
        }
        // One full recomputation, as a new subscription performs it.
        let began = Instant::now();
        let _probe = db.capability("items").live_find::<Item, _>(|item| item.active).await.unwrap();
        let rescan_ms = began.elapsed().as_secs_f64() * 1000.0;

        let writes = 200;
        let began = Instant::now();
        for w in 0..writes {
            // A matching write: every subscriber's result changes.
            db.insert(&format!("items:new{w}"), Item { n: size + w, active: true }).unwrap();
            for subscription in &mut live {
                subscription.next().await.unwrap();
            }
        }
        let per_change_ms = began.elapsed().as_secs_f64() * 1000.0 / writes as f64;
        println!(
            "{{\"size\":{size},\"subscribers\":{subscribers},\"matches\":{},\"change_all_subscribers_ms\":{per_change_ms:.3},\"rescan_one_subscription_ms\":{rescan_ms:.3}}}",
            size / 10 + writes
        );
        drop(live);
        drop(db);
        let _ = std::fs::remove_dir_all(&directory);
    }
}