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
// SPDX-License-Identifier: MIT OR Apache-2.0
// SPDX-FileCopyrightText: 2025-2026 ObzenFlow Contributors
// https://obzenflow.dev
use super::journal_error::JournalError;
use super::journal_owner::JournalOwner;
use super::reader::JournalReader;
use super::{AppendOptions, JournalConfig, ObservabilityPolicy};
use crate::event::journal_record::JournalRecord;
use crate::event::types::EventId;
use crate::event::vector_clock::CausalOrderingService;
use crate::event::JournalEvent;
use crate::id::JournalId;
use async_trait::async_trait;
/// Core journal trait - defines what a journal must do
///
/// Infrastructure will implement this trait with actual storage
/// Generic over T which is the event type (ChainEvent or SystemEvent)
#[async_trait]
pub trait Journal<T>: Send + Sync
where
T: JournalEvent,
{
/// Get the ID of this journal
fn id(&self) -> &JournalId;
/// Get the owner of this journal (if any)
fn owner(&self) -> Option<&JournalOwner>;
fn observation_reader(&self) -> Option<&dyn super::JournalObservationReader> {
None
}
/// Current committed carriers, newest first, for reporting keys and
/// independently stamped observation families. Cost is proportional to
/// retained keys and their carrier frames, never the journal's history.
/// Implementations must not scan, rebuild an index, or write checkpoints
/// here. An unavailable live index returns no values; callers retain their
/// previous buffer. This does not certify physical journal coverage.
async fn read_metrics_tail(&self) -> Result<Vec<JournalRecord<T::Payload>>, JournalError> {
Ok(Vec::new())
}
/// Configure this journal before publication starts. Handles of the same
/// journal share the policy and allowance; other journals are independent.
fn configure(&self, config: JournalConfig) -> Result<(), JournalError> {
match config.observability {
ObservabilityPolicy::EveryRecord => Ok(()),
ObservabilityPolicy::Periodic { .. } => Err(JournalError::Implementation {
message: "This journal does not support sparse observability".into(),
source: "unsupported observability policy".into(),
}),
}
}
/// Append an event to the journal
///
/// The implementation MUST:
/// 1. Generate appropriate vector clock based on writer and parent
/// 2. Ensure atomic append operation
/// 3. Return the complete JournalRecord with causal information
/// 4. Apply journal policy to inherited and deferred optional attachments.
/// Invoke deferred capture only after admission, and preserve historical
/// attachments exactly. Payload and provenance are never sampled.
/// 5. Retain an initiated physical commit through storage bookkeeping if
/// the caller stops waiting. Cancellation is not rollback.
///
/// An error certifies non-commit, except `JournalError::CommitIndeterminate`.
/// That result means storage may have committed and must not be retried.
async fn append(
&self,
event: T,
options: AppendOptions<'_, T>,
) -> Result<JournalRecord<T::Payload>, JournalError>;
/// Atomically append a logical group of events.
///
/// Implementations must make the complete group visible together or leave
/// every member invisible, including after recovery from an interrupted
/// physical write. `group_id` is a deterministic, policy-neutral identity
/// used by durable implementations for framing and recovery diagnostics.
/// The single-append cancellation and indeterminate-error contract applies
/// to the complete group.
///
/// The default rejects multi-event groups. Lightweight journals can rely
/// on this default until they support atomic groups.
async fn append_group(
&self,
group_id: &str,
events: Vec<T>,
options: AppendOptions<'_, T>,
) -> Result<Vec<JournalRecord<T::Payload>>, JournalError> {
match events.len() {
0 => Ok(Vec::new()),
1 => Ok(vec![
self.append(events.into_iter().next().expect("one event"), options)
.await?,
]),
count => Err(JournalError::AtomicAppendUnsupported {
group_id: group_id.to_string(),
member_count: count,
}),
}
}
/// Read every event in raw append/storage order, without causal sorting.
///
/// The single raw-enumeration primitive the causal reads derive from. Disk
/// reads through its full-scan framed reader; memory clones its in-memory
/// vector.
async fn read_all_unordered(&self) -> Result<Vec<JournalRecord<T::Payload>>, JournalError>;
/// Read all events and return them in causal order.
///
/// If A happened-before B then A appears before B; concurrent events are
/// broken by `EventId` ordering, not wall-clock time. Derived from
/// `read_all_unordered`.
async fn read_causally_ordered(&self) -> Result<Vec<JournalRecord<T::Payload>>, JournalError> {
CausalOrderingService::order_envelopes_by_event_id(self.read_all_unordered().await?)
}
/// Read events causally after the given event, in causal order. Empty if the
/// event is absent. Derived from `read_causally_ordered`.
async fn read_causally_after(
&self,
after_event_id: &EventId,
) -> Result<Vec<JournalRecord<T::Payload>>, JournalError> {
let all = self.read_causally_ordered().await?;
Ok(match all.iter().position(|e| e.id() == after_event_id) {
Some(pos) => all.into_iter().skip(pos + 1).collect(),
None => Vec::new(),
})
}
/// Read a specific event by ID
///
/// Returns None if the event doesn't exist
async fn read_event(
&self,
event_id: &EventId,
) -> Result<Option<JournalRecord<T::Payload>>, JournalError>;
/// Create a reader that starts from the beginning. Multiple readers can be
/// created for the same journal. Derived from `reader_from(0)`.
async fn reader(&self) -> Result<Box<dyn JournalReader<T>>, JournalError> {
self.reader_from(0).await
}
/// Create a reader that starts from a specific position.
///
/// The position is the portable append index (`0, 1, 2, ...`), the count of
/// committed records before the reader's first event. A reader created at
/// position N is equivalent to a reader advanced past N records. Useful for
/// resuming from a checkpoint.
async fn reader_from(&self, position: u64) -> Result<Box<dyn JournalReader<T>>, JournalError>;
/// Read the last N events from the journal by scanning backwards from EOF.
///
/// This is useful for efficiently getting recent events without loading the entire journal.
/// Events are returned in reverse order (most recent first).
///
/// Returns an empty vec if the journal is empty.
/// This method should be O(n) where n is the requested count,
/// not O(total_events) in the journal.
async fn read_last_n(
&self,
count: usize,
) -> Result<Vec<JournalRecord<T::Payload>>, JournalError>;
}