liven 0.0.3

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

use crate::storage::key::StreamKey;
use crate::types::DataValue;
use crate::types::Record;
use std::sync::mpsc::SyncSender;

/// Sentinel value to signal flusher thread shutdown.
#[derive(Debug, Clone, Copy)]
pub struct ShutdownSentinel;

/// Represents write operations enqueued in the group-commit ring buffer.
pub enum LogEntry {
    Single {
        stream_name: String,
        key: StreamKey,
        value: DataValue,
        is_tombstone: bool,
        response_tx: SyncSender<Result<Record, String>>,
    },
    TombstoneBatch {
        stream_name: String,
        keys: Vec<StreamKey>,
        response_tx: SyncSender<Result<Vec<Record>, String>>,
    },
    /// Shutdown signal for clean flusher thread termination.
    Shutdown,
}

/// Thread-safe client interface to enqueue writes into the background flusher.
#[derive(Clone)]
pub struct LogRingBuffer {
    tx: std::sync::mpsc::SyncSender<LogEntry>,
}

impl LogRingBuffer {
    pub fn new(tx: std::sync::mpsc::SyncSender<LogEntry>) -> Self {
        Self { tx }
    }

    /// Enqueues a write transaction and blocks until execution by the background flusher.
    /// This provides a synchronous API with high-throughput group-commits under the hood.
    pub fn append(
        &self,
        stream_name: &str,
        key: StreamKey,
        value: DataValue,
        is_tombstone: bool,
    ) -> crate::error::Result<Record> {
        let (response_tx, rx) = std::sync::mpsc::sync_channel(1);
        let entry = LogEntry::Single {
            stream_name: stream_name.to_string(),
            key,
            value,
            is_tombstone,
            response_tx,
        };

        self.tx
            .send(entry)
            .map_err(|_| crate::error::LivenError::FlusherTerminated)?;

        rx.recv()
            .map_err(|e| {
                crate::error::LivenError::Internal(format!(
                    "LogRingBuffer coordination error: {}",
                    e
                ))
            })?
            .map_err(Into::into)
    }

    /// Enqueues a transaction without waiting for response.
    pub fn tx_send(&self, entry: LogEntry) -> crate::error::Result<()> {
        self.tx
            .send(entry)
            .map_err(|_| crate::error::LivenError::FlusherTerminated)
    }

    /// Sends the shutdown sentinel to the flusher thread.
    pub fn send_shutdown(&self) -> crate::error::Result<()> {
        self.tx
            .send(LogEntry::Shutdown)
            .map_err(|_| crate::error::LivenError::FlusherTerminated)
    }
}