liminal-rs 0.16.1

A conversation-based messaging bus built on beamr
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

use std::sync::Arc;

use super::{
    ConsumerCursor, DurabilityError, DurableChannel, DurableConversation, DurableStore,
    MessageEnvelope, head_checkpoint::recover_partition_head, replay_from,
};

#[cfg(test)]
mod tests;

/// Cursor state plus missed messages replayed from the recovered cursor offset.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RecoveredCursor {
    /// Recovered cursor positioned at the persisted offset, or zero when absent.
    pub cursor: ConsumerCursor,
    /// Durable channel messages replayed from the cursor position to the partition head.
    pub replayed_messages: Vec<MessageEnvelope>,
}

/// Recovers all durable channel partition next-sequence counters from haematite logs.
///
/// Each partition resumes from its recorded head and reads only the rows appended after
/// it; a missing or refused record is walked in full and recorded again
/// ([`super::head_checkpoint`]).
///
/// # Errors
///
/// Propagates store read and write errors and returns [`DurabilityError::ConfigError`] on
/// offset or sequence overflow.
pub async fn recover_partition_sequences(
    channel_id: &str,
    partition_count: usize,
    store: &dyn DurableStore,
) -> Result<Vec<u64>, DurabilityError> {
    let mut next_sequences = Vec::with_capacity(partition_count);
    for partition_index in 0..partition_count {
        let stream_key = partition_stream_key(channel_id, partition_index);
        next_sequences.push(recover_partition_head(store, &stream_key).await?);
    }
    Ok(next_sequences)
}

/// Recovers a durable channel by reconstructing each partition sequence from storage.
///
/// # Errors
///
/// Propagates store read errors and channel recovery validation errors.
pub async fn recover_durable_channel(
    channel_id: impl Into<String>,
    partition_count: usize,
    store: Arc<dyn DurableStore>,
) -> Result<DurableChannel, DurabilityError> {
    let channel_id = channel_id.into();
    let next_sequences =
        recover_partition_sequences(&channel_id, partition_count, store.as_ref()).await?;
    DurableChannel::from_recovered_sequences(channel_id, partition_count, store, next_sequences)
}

/// Recovers a durable conversation by delegating to the event-log replay path.
///
/// # Errors
///
/// Propagates store read errors, event deserialization errors, and replay validation errors.
pub async fn recover_conversation(
    conversation_id: impl Into<String>,
    store: Arc<dyn DurableStore>,
) -> Result<DurableConversation, DurabilityError> {
    DurableConversation::recover(conversation_id, store).await
}

/// Recovers a consumer cursor by delegating to the cursor resume path.
///
/// # Errors
///
/// Propagates store read errors from cursor resume.
pub async fn recover_cursor(
    consumer_id: impl Into<String>,
    partition_key: impl Into<String>,
    store: &dyn DurableStore,
) -> Result<ConsumerCursor, DurabilityError> {
    ConsumerCursor::resume(consumer_id, partition_key, store).await
}

/// Recovers a consumer cursor and replays missed durable messages from that offset.
///
/// # Errors
///
/// Propagates cursor resume, store replay, and envelope deserialization errors.
pub async fn recover_cursor_with_replay(
    consumer_id: impl Into<String>,
    partition_key: impl Into<String>,
    store: &dyn DurableStore,
) -> Result<RecoveredCursor, DurabilityError> {
    let cursor = recover_cursor(consumer_id, partition_key, store).await?;
    let replayed_messages =
        replay_from(store, cursor.partition_key(), cursor.current_offset()).await?;
    Ok(RecoveredCursor {
        cursor,
        replayed_messages,
    })
}

fn partition_stream_key(channel_id: &str, partition_index: usize) -> String {
    format!("{channel_id}:{partition_index}")
}