reddb-io-server 1.28.28

RedDB server-side engine: storage, runtime, replication, MCP, AI, and the gRPC/HTTP/RedWire/PG-wire dispatchers. Re-exported by the umbrella `reddb` crate.
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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

use reddb_file::OperationalManifest;
use reddb_server::{RedDBOptions, RedDBRuntime};
use reddb_types::Value;

#[test]
fn append_only_checkpoints_keep_rows_inside_the_database() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("append_only.rdb");
    {
        let runtime = RedDBRuntime::with_options(RedDBOptions::persistent(&path)).unwrap();
        runtime
            .execute_query("CREATE TABLE events (id INT, msg TEXT) APPEND ONLY")
            .unwrap();
        runtime
            .execute_query("INSERT INTO events (id, msg) VALUES (1, 'a'), (2, 'b')")
            .unwrap();
        runtime.flush().unwrap();
    }
    for expected in [2, 3] {
        let runtime = RedDBRuntime::with_options(RedDBOptions::persistent(&path)).unwrap();
        assert_eq!(
            runtime
                .execute_query("SELECT id FROM events")
                .unwrap()
                .result
                .records
                .len(),
            expected
        );
        assert!(runtime.execute_query("DELETE FROM events").is_err());
        assert!(runtime
            .execute_query("UPDATE events SET msg = 'denied'")
            .is_err());
        if expected == 2 {
            runtime
                .execute_query("INSERT INTO events (id, msg) VALUES (3, 'c')")
                .unwrap();
        }
        runtime.flush().unwrap();
        let manifest = OperationalManifest::for_db_path(&path);
        assert!(!manifest.current_manifest_path_for_test().exists());
        assert!(manifest.append_only_segments_for_test().unwrap().is_empty());
    }
}

#[test]
fn append_only_retention_deletes_expired_rows_without_external_segments() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("retention.rdb");
    let runtime = RedDBRuntime::with_options(RedDBOptions::persistent(&path)).unwrap();
    runtime
        .execute_query("CREATE TABLE events (id INT, msg TEXT) WITH timestamps = true APPEND ONLY")
        .unwrap();
    runtime
        .execute_query("INSERT INTO events (id, msg) VALUES (1, 'expired')")
        .unwrap();
    let manager = runtime.db().store().get_collection("events").unwrap();
    let mut entity = manager.query_all(|_| true).remove(0);
    entity.created_at = 1;
    if let reddb_server::storage::EntityData::Row(row) = &mut entity.data {
        if let Some(named) = &mut row.named {
            named.insert("created_at".into(), Value::UnsignedInteger(1));
        }
        if let Some(schema) = &row.schema {
            let index = schema.iter().position(|name| name == "created_at").unwrap();
            row.columns[index] = Value::UnsignedInteger(1);
        }
    }
    manager.update(entity).unwrap();
    runtime.flush().unwrap();
    runtime
        .execute_query("ALTER COLLECTION events SET RETENTION 1 d")
        .unwrap();
    runtime
        .execute_query("INSERT INTO events (id, msg) VALUES (2, 'live')")
        .unwrap();
    // A tick deliberately yields when a background checkpoint owns the
    // publication fence. Retry bounded work instead of assuming immediate entry.
    for _ in 0..100 {
        runtime.sweep_retention_tick(1000);
        if manager.count() == 1 {
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(10));
    }
    assert_eq!(manager.count(), 1);
    assert!(runtime.execute_query("DELETE FROM events").is_err());
    let result = runtime.execute_query("SELECT rows_swept_total, segments_retired_total FROM red.retention WHERE name = 'events'").unwrap();
    let row = &result.result.records[0];
    assert_eq!(
        row.get("rows_swept_total"),
        Some(&Value::UnsignedInteger(1))
    );
    assert_eq!(
        row.get("segments_retired_total"),
        Some(&Value::UnsignedInteger(0))
    );
    runtime.flush().unwrap();
    drop(manager);
    drop(runtime);
    let cold = RedDBRuntime::with_options(RedDBOptions::persistent(&path)).unwrap();
    assert_eq!(
        cold.db().store().get_collection("events").unwrap().count(),
        1
    );
    assert_eq!(
        cold.execute_query("SELECT id FROM events")
            .unwrap()
            .result
            .records[0]
            .get("id"),
        Some(&Value::Integer(2))
    );
    assert!(!OperationalManifest::for_db_path(&path)
        .current_manifest_path_for_test()
        .exists());
}