tephra 0.1.0

A DCB-compliant, immutable event store with global ordering.
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

//! The caller-facing handle.

use flume::{self as channel, Sender};

use crate::Position;
use crate::event::Event;
use crate::log::set::PositionRange;
use crate::query::{AppendCondition, Query};
use crate::read::{ReadHandle, Reads, Subscription};

use super::{AppendError, Message, Request};

/// A cloneable, `Send` handle to the write coordinator. Every clone feeds the same
/// single writer; dropping the last one (and the owning
/// [`WriteCoordinator`](super::WriteCoordinator)) shuts the coordinator down.
///
/// Also carries a [`ReadHandle`] so appends and reads share one handle, but the two are
/// independent: [`read`](Self::read) runs on the caller's thread over the published
/// snapshot and never touches the writer.
#[derive(Clone)]
pub struct WriteHandle {
    pub(super) tx: Sender<Message>,
    pub(super) reader: ReadHandle,
}

impl WriteHandle {
    /// Appends `events` as one atomic unit, blocking until the batch is durable or the
    /// condition fails.
    ///
    /// `events` are already-encoded [`Event`]s (encoding and validation happen on the
    /// caller thread, off the writer). Pass `None` for an unconditional append, or a
    /// condition to guard the write against concurrent conflicting events. On success
    /// the returned [`PositionRange`] covers the assigned positions, dense and in order.
    ///
    /// Blocks if the request queue is full (backpressure). A
    /// [`ConflictSite::SameBatch`](super::ConflictSite::SameBatch)
    /// conflict is retryable; see [`AppendError`] and
    /// [`ConflictSite`](super::ConflictSite).
    pub fn append(
        &self,
        events: Vec<Event>,
        condition: Option<AppendCondition>,
    ) -> Result<PositionRange, AppendError> {
        if events.is_empty() {
            return Err(AppendError::Empty);
        }
        let (reply, response) = channel::unbounded();
        let request = Request {
            events,
            condition,
            reply,
        };
        // `send` on a full bounded channel blocks; an error means the coordinator is
        // gone (channel disconnected).
        self.tx
            .send(Message::Append(request))
            .map_err(|_| AppendError::Shutdown)?;
        // A dropped reply sender (coordinator died mid-flight) surfaces as shutdown.
        response.recv().map_err(|_| AppendError::Shutdown)?
    }

    /// Reads events matching `query`, ascending, strictly after `after`, up to the
    /// watermark pinned now. Runs on the **caller's own thread** over the published read
    /// snapshot: it never touches the writer thread, and read-your-writes still holds (the
    /// writer publishes the watermark before replying to an append). See
    /// [`ReadHandle::read`] and [`Reads`].
    pub fn read(&self, query: Query, after: Position) -> Reads {
        self.reader.read(query, after)
    }

    /// Starts a [`Subscription`] over `query`, resuming strictly after `after`: it catches up
    /// on all durable events, then tails live ones with no gap and no duplicate at the
    /// boundary. Runs on the caller's thread over the published read state, like
    /// [`read`](Self::read); it blocks (on a condvar the writer signals at each commit) only
    /// while waiting for new events. See [`Subscription`].
    pub fn subscribe(&self, query: Query, after: Position) -> Subscription {
        self.reader.subscribe(query, after)
    }

    /// A standalone [`ReadHandle`] for pure readers, sharing this handle's published read
    /// state without the ability to append.
    pub fn reader(&self) -> ReadHandle {
        self.reader.clone()
    }

    /// The `async` counterpart of [`append`](Self::append): identical semantics, but it
    /// yields to the executor instead of blocking the thread while the request queue is
    /// full (backpressure) and while awaiting the durable reply.
    #[cfg(feature = "async")]
    pub async fn append_async(
        &self,
        events: Vec<Event>,
        condition: Option<AppendCondition>,
    ) -> Result<PositionRange, AppendError> {
        if events.is_empty() {
            return Err(AppendError::Empty);
        }
        let (reply, response) = channel::unbounded();
        let request = Request {
            events,
            condition,
            reply,
        };
        // A full request queue awaits rather than blocks; an error means the coordinator
        // is gone (channel disconnected).
        self.tx
            .send_async(Message::Append(request))
            .await
            .map_err(|_| AppendError::Shutdown)?;
        // A dropped reply sender (coordinator died mid-flight) surfaces as shutdown.
        response
            .recv_async()
            .await
            .map_err(|_| AppendError::Shutdown)?
    }
}