liven 0.1.0

LIVEN is a fast, lightweight database built to capture, store, and stream data in real time.
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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200

// Data Safety Tests
// Tests for critical data safety, consistency, and correctness guarantees

use liven::executor::execute_query;
use liven::storage::StorageEngine;
use liven::types::Query;
use std::sync::Arc;
use tempfile::tempdir;

#[test]
fn test_flusher_shutdown_with_pending_writes() {
    let temp_dir = tempdir().unwrap();
    let data_dir = temp_dir.path();

    // Create engine and append data
    let engine = StorageEngine::new(data_dir, 16 * 1024 * 1024).unwrap();

    // Append multiple records to ensure flusher has work
    for i in 0..10 {
        let _ = engine.append(
            "test_stream",
            &format!("key_{}", i),
            liven::types::DataValue::String(format!("value_{}", i)),
            false,
        );
    }

    // Drop engine - this should trigger clean shutdown
    drop(engine);

    // Reopen engine and verify data persisted
    let engine2 = StorageEngine::new(data_dir, 16 * 1024 * 1024).unwrap();
    let records = engine2.scan_historical().unwrap();

    // Should have all 10 records (no data loss)
    assert_eq!(
        records.len(),
        10,
        "Data should persist after clean shutdown"
    );

    // Verify content
    for i in 0..10 {
        let key = format!("key_{}", i);
        let record = engine2.get("test_stream", &key).unwrap();
        assert!(record.is_some(), "Record {} should exist", i);
    }
}

#[test]
fn test_write_lock_prevents_duplicate_keys() {
    let temp_dir = tempdir().unwrap();
    let data_dir = temp_dir.path();
    let engine = Arc::new(StorageEngine::new(data_dir, 16 * 1024 * 1024).unwrap());

    // First insert should succeed
    let query1 = Query::Insert {
        stream_name: "test_stream".to_string(),
        key: "test_key".to_string(),
        value: serde_json::json!({"field": "value1"}),
    };

    let result1 = execute_query(&engine, &query1);
    assert!(result1.is_ok(), "First insert should succeed");
    assert_eq!(result1.unwrap().len(), 1);

    // Second insert with same key should fail
    let query2 = Query::Insert {
        stream_name: "test_stream".to_string(),
        key: "test_key".to_string(),
        value: serde_json::json!({"field": "value2"}),
    };

    let result2 = execute_query(&engine, &query2);
    assert!(result2.is_err(), "Second insert should fail");

    // Verify error message contains guidance
    if let Err(e) = result2 {
        let error_msg = e.to_string();
        assert!(error_msg.contains("already exists"));
        assert!(error_msg.contains("upsert"));
        assert!(error_msg.contains("update"));
    }

    // Verify only one record exists
    let records = engine.get("test_stream", "test_key").unwrap();
    assert!(records.is_some());
}

#[test]
fn test_upsert_tombstones_old_records() {
    let temp_dir = tempdir().unwrap();
    let data_dir = temp_dir.path();
    let engine = Arc::new(StorageEngine::new(data_dir, 16 * 1024 * 1024).unwrap());

    // Insert initial record
    let insert_query = Query::Insert {
        stream_name: "test_stream".to_string(),
        key: "test_key".to_string(),
        value: serde_json::json!({"version": 1}),
    };

    execute_query(&engine, &insert_query).unwrap();

    // Upsert should tombstone old record and create new one
    let upsert_query = Query::Upsert {
        stream_name: "test_stream".to_string(),
        key: "test_key".to_string(),
        value: serde_json::json!({"version": 2}),
    };

    execute_query(&engine, &upsert_query).unwrap();

    // Should have exactly one live record (the new one)
    let records = engine.get("test_stream", "test_key").unwrap();
    assert!(records.is_some());

    let record = records.unwrap();
    if let liven::types::DataValue::String(json_str) = &record.value {
        let value: serde_json::Value = serde_json::from_str(json_str).unwrap();
        assert_eq!(value["version"], 2, "Should have the new version");
    } else {
        panic!("Expected string JSON value");
    }
}

#[test]
fn test_update_tombstones_old_records() {
    let temp_dir = tempdir().unwrap();
    let data_dir = temp_dir.path();
    let engine = Arc::new(StorageEngine::new(data_dir, 16 * 1024 * 1024).unwrap());

    // Insert initial record
    let insert_query = Query::Insert {
        stream_name: "test_stream".to_string(),
        key: "test_key".to_string(),
        value: serde_json::json!({"name": "original", "count": 1}),
    };

    execute_query(&engine, &insert_query).unwrap();

    // Update should tombstone old record and merge
    let update_query = Query::Update {
        stream_name: "test_stream".to_string(),
        key: "test_key".to_string(),
        value: serde_json::json!({"count": 2, "new_field": "added"}),
    };

    execute_query(&engine, &update_query).unwrap();

    // Should have exactly one live record (the merged one)
    let records = engine.get("test_stream", "test_key").unwrap();
    assert!(records.is_some());

    let record = records.unwrap();
    if let liven::types::DataValue::String(json_str) = &record.value {
        let value: serde_json::Value = serde_json::from_str(json_str).unwrap();
        assert_eq!(value["name"], "original", "Should preserve original field");
        assert_eq!(value["count"], 2, "Should update count");
        assert_eq!(value["new_field"], "added", "Should add new field");
    } else {
        panic!("Expected string JSON value");
    }
}

#[test]
fn test_key_length_validation() {
    let temp_dir = tempdir().unwrap();
    let data_dir = temp_dir.path();
    let engine = Arc::new(StorageEngine::new(data_dir, 16 * 1024 * 1024).unwrap());

    // 33-byte key should fail
    let long_key = "a".repeat(33);
    let query = Query::Insert {
        stream_name: "test_stream".to_string(),
        key: long_key,
        value: serde_json::json!({"test": "data"}),
    };

    let result = execute_query(&engine, &query);
    assert!(result.is_err(), "33-byte key should be rejected");

    if let Err(e) = result {
        let error_msg = e.to_string();
        assert!(error_msg.contains("33 bytes"));
        assert!(error_msg.contains("maximum is 32 bytes"));
        assert!(error_msg.contains("Shorten the key"));
    }

    // 32-byte key should succeed
    let valid_key = "a".repeat(32);
    let query = Query::Insert {
        stream_name: "test_stream".to_string(),
        key: valid_key,
        value: serde_json::json!({"test": "data"}),
    };

    let result = execute_query(&engine, &query);
    assert!(result.is_ok(), "32-byte key should be accepted");
}