771 lines
28 KiB
Rust
771 lines
28 KiB
Rust
//! The stream mechanism's mem arm — the in-memory event log +
|
|
//! per-consumer offsets (engine-mem.md §Mapping the contract: global
|
|
//! FIFO by monotone offset; keyed publish rides the overlay).
|
|
//!
|
|
//! - Publishes append to the per-stream log under its section guard:
|
|
//! engine-assigned offsets (the engine-wide allocator — pg-bigserial
|
|
//! shaped, monotone, immutable, never renumbered; a fresh stream's
|
|
//! first event does not carry offset 1, the ADR-015 §4 note), keys
|
|
//! are carried metadata (a present key must be non-empty —
|
|
//! `InvalidName`, ADR-015 §1), and the commit boundary fires one
|
|
//! wake per event on the stream's engine-internal wake channel.
|
|
//! - Cursor reads are `offset ASC` (global FIFO, ADR-015 §4) with the
|
|
//! extent guard at the trait-impl entry (`limit <= 0` → empty
|
|
//! `Vec`, ADR-023 §2).
|
|
//! - Checkpoints are explicit, monotone saves (a save below the
|
|
//! stored checkpoint is a silent no-op — ADR-019 §6; an absent
|
|
//! consumer reads 0, ADR-019 §1).
|
|
//! - `trim_to` deletes `offset <= horizon` exactly (a negative horizon
|
|
//! deletes nothing, idempotently — the boundary argument is total,
|
|
//! ADR-023 §2), **wakes nothing**, and surviving events keep their
|
|
//! offsets (gaps legal, ADR-015 §5). Reads from a trimmed region
|
|
//! resume at the horizon's first remaining row.
|
|
//! - `subscribe` attaches the receiver's wake feed **before** the
|
|
//! stored-offset read (no commit can fall between — a buffered wake
|
|
//! for a commit already re-read costs an empty re-poll, never a
|
|
//! missed event), drains the log to the tail, then parks on the
|
|
//! feed: wake-driven re-read with no polling requirement; explicit
|
|
//! saves only. The one close path is engine drop — the bus close
|
|
//! disconnects every feed and `recv()` returns the terminal `None`
|
|
//! (ADR-021 §5).
|
|
|
|
use std::collections::VecDeque;
|
|
use std::sync::Arc;
|
|
|
|
use serde_json::Value;
|
|
|
|
use alkstore::{
|
|
BoxedFuture, Error, EventReceiver, Result, StreamEvent, StreamHandle, validate_local_name,
|
|
};
|
|
|
|
use crate::clock::Clock;
|
|
use crate::state::Guard;
|
|
use crate::state::{EventRow, State, StreamState};
|
|
use crate::wakes::WakeBus;
|
|
|
|
/// How many events a wake-driven re-read fetches per page while
|
|
/// draining to the tail.
|
|
const SUBSCRIBE_PAGE: i64 = 256;
|
|
|
|
/// The engine-internal wake channel a stream's subscribers ride:
|
|
/// engine-derived, under the reserved prefix — consumer entry points
|
|
/// reject the prefix, so a consumer `listen` can never attach to it
|
|
/// (and `notify` can never fire into it). The publish-side wake feed
|
|
/// is plumbing, not contract surface.
|
|
pub(crate) fn stream_wake_channel(stream: &str) -> String {
|
|
format!("{}stream_wake:{}", alkstore::RESERVED_PREFIX, stream)
|
|
}
|
|
|
|
/// Insert one committed event under the streams guard (the auto-commit
|
|
/// publish and the commit-merge share this effect). Returns nothing —
|
|
/// the caller owns the wake firing at its commit boundary.
|
|
pub(crate) fn insert_event(s: &mut StreamState, stream: &str, row: EventRow) {
|
|
s.logs
|
|
.entry(stream.to_string())
|
|
.or_default()
|
|
.insert(row.offset, row);
|
|
}
|
|
|
|
/// Read `offset ASC`, `offset > cursor`, up to `limit` (the extent
|
|
/// guard is the caller's — this helper is contract-blind to it).
|
|
pub(crate) fn read_events(s: &StreamState, stream: &str, from: i64, limit: i64) -> Vec<EventRow> {
|
|
s.logs
|
|
.get(stream)
|
|
.map(|log| {
|
|
log.range((from + 1)..)
|
|
.take(limit.max(0) as usize)
|
|
.map(|(_, row)| row.clone())
|
|
.collect()
|
|
})
|
|
.unwrap_or_default()
|
|
}
|
|
|
|
/// The monotone checkpoint save (ADR-019 §6): a save at-or-below the
|
|
/// stored checkpoint is a silent no-op; the direct form, the tx form,
|
|
/// and the receiver's no-arg form all land here.
|
|
pub(crate) fn save_offset_row(s: &mut StreamState, stream: &str, consumer: &str, offset: i64) {
|
|
let key = (stream.to_string(), consumer.to_string());
|
|
match s.offsets.get(&key) {
|
|
Some(current) if offset <= *current => {}
|
|
_ => {
|
|
s.offsets.insert(key, offset);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// The stored checkpoint; absent consumer = 0.
|
|
pub(crate) fn get_offset_row(s: &StreamState, stream: &str, consumer: &str) -> i64 {
|
|
s.offsets
|
|
.get(&(stream.to_string(), consumer.to_string()))
|
|
.copied()
|
|
.unwrap_or(0)
|
|
}
|
|
|
|
/// `trim_to`: delete `offset <= horizon` (exact boundary); surviving
|
|
/// events keep their offsets. Returns the deleted count. Wakes nothing.
|
|
pub(crate) fn trim_events(s: &mut StreamState, stream: &str, horizon: i64) -> i64 {
|
|
let Some(log) = s.logs.get_mut(stream) else {
|
|
return 0;
|
|
};
|
|
let horizon_next = horizon.saturating_add(1);
|
|
let removed: Vec<i64> = log.range(..horizon_next).map(|(k, _)| *k).collect();
|
|
let count = removed.len();
|
|
for offset in removed {
|
|
log.remove(&offset);
|
|
}
|
|
count as i64
|
|
}
|
|
|
|
fn event_from_row(row: EventRow, stream: &str) -> StreamEvent {
|
|
StreamEvent::from_row(
|
|
row.offset,
|
|
stream.to_string(),
|
|
row.key,
|
|
row.payload,
|
|
row.created_at,
|
|
)
|
|
}
|
|
|
|
/// The stream-scoped handle (see the module docs) for one validated
|
|
/// stream name.
|
|
pub struct MemStreamHandle {
|
|
pub(crate) name: String,
|
|
pub(crate) state: Arc<State>,
|
|
pub(crate) clock: Arc<Clock>,
|
|
pub(crate) wakes: Arc<WakeBus>,
|
|
}
|
|
|
|
impl std::fmt::Debug for MemStreamHandle {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("MemStreamHandle")
|
|
.field("name", &self.name)
|
|
.finish_non_exhaustive()
|
|
}
|
|
}
|
|
|
|
impl MemStreamHandle {
|
|
pub(crate) fn new(
|
|
name: String,
|
|
state: Arc<State>,
|
|
clock: Arc<Clock>,
|
|
wakes: Arc<WakeBus>,
|
|
) -> Self {
|
|
Self {
|
|
name,
|
|
state,
|
|
clock,
|
|
wakes,
|
|
}
|
|
}
|
|
|
|
/// The keyed publish is the one publish shape (the plain form is
|
|
/// `key = None`).
|
|
fn stage_publish(&self, key: Option<String>, payload: Value) -> Result<i64> {
|
|
if let Some(k) = &key
|
|
&& (k.is_empty() || k.trim().is_empty())
|
|
{
|
|
return Err(Error::InvalidName { name: k.clone() });
|
|
}
|
|
let bytes = alkstore::encode_payload(&payload)?;
|
|
let now = self.clock.unix_now();
|
|
let offset = self.state.allocate_event_offset();
|
|
let row = EventRow {
|
|
offset,
|
|
key,
|
|
payload: bytes,
|
|
created_at: now,
|
|
};
|
|
{
|
|
let mut s = self.state.streams.guard();
|
|
insert_event(&mut s, &self.name, row);
|
|
}
|
|
// The auto-commit publish's commit boundary is its own
|
|
// completion: one wake per event, per-subscriber FIFO.
|
|
let channel = stream_wake_channel(&self.name);
|
|
self.wakes.fire_commit_wakes([channel]);
|
|
Ok(offset)
|
|
}
|
|
}
|
|
|
|
impl StreamHandle for MemStreamHandle {
|
|
fn name(&self) -> &str {
|
|
&self.name
|
|
}
|
|
|
|
fn publish<'a>(&'a self, payload: Value) -> BoxedFuture<'a, Result<i64>> {
|
|
match self.stage_publish(None, payload) {
|
|
Ok(offset) => Box::pin(async move { Ok(offset) }),
|
|
Err(e) => Box::pin(async move { Err(e) }),
|
|
}
|
|
}
|
|
|
|
fn publish_with_key<'a>(
|
|
&'a self,
|
|
key: Option<String>,
|
|
payload: Value,
|
|
) -> BoxedFuture<'a, Result<i64>> {
|
|
match self.stage_publish(key, payload) {
|
|
Ok(offset) => Box::pin(async move { Ok(offset) }),
|
|
Err(e) => Box::pin(async move { Err(e) }),
|
|
}
|
|
}
|
|
|
|
fn read_since<'a>(
|
|
&'a self,
|
|
offset: i64,
|
|
limit: i64,
|
|
) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
|
|
if limit <= 0 {
|
|
return Box::pin(async { Ok(Vec::new()) });
|
|
}
|
|
let events = {
|
|
let s = self.state.streams.guard();
|
|
read_events(&s, &self.name, offset, limit)
|
|
};
|
|
let name = self.name.clone();
|
|
Box::pin(async move {
|
|
Ok(events
|
|
.into_iter()
|
|
.map(|r| event_from_row(r, &name))
|
|
.collect())
|
|
})
|
|
}
|
|
|
|
fn read_from_consumer<'a>(
|
|
&'a self,
|
|
consumer: &str,
|
|
limit: i64,
|
|
) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
|
|
if let Err(e) = validate_local_name(consumer) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
if limit <= 0 {
|
|
return Box::pin(async { Ok(Vec::new()) });
|
|
}
|
|
let name = self.name.clone();
|
|
let state = self.state.clone();
|
|
let consumer = consumer.to_string();
|
|
Box::pin(async move {
|
|
let events = {
|
|
let s = state.streams.guard();
|
|
let from = get_offset_row(&s, &name, &consumer);
|
|
read_events(&s, &name, from, limit)
|
|
};
|
|
Ok(events
|
|
.into_iter()
|
|
.map(|r| event_from_row(r, &name))
|
|
.collect())
|
|
})
|
|
}
|
|
|
|
fn save_offset<'a>(&'a self, consumer: &str, offset: i64) -> BoxedFuture<'a, Result<()>> {
|
|
if let Err(e) = validate_local_name(consumer) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
let name = self.name.clone();
|
|
let state = self.state.clone();
|
|
let consumer = consumer.to_string();
|
|
Box::pin(async move {
|
|
let mut s = state.streams.guard();
|
|
save_offset_row(&mut s, &name, &consumer, offset);
|
|
Ok(())
|
|
})
|
|
}
|
|
|
|
fn get_offset<'a>(&'a self, consumer: &str) -> BoxedFuture<'a, Result<i64>> {
|
|
if let Err(e) = validate_local_name(consumer) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
let name = self.name.clone();
|
|
let state = self.state.clone();
|
|
let consumer = consumer.to_string();
|
|
Box::pin(async move {
|
|
let s = state.streams.guard();
|
|
Ok(get_offset_row(&s, &name, &consumer))
|
|
})
|
|
}
|
|
|
|
fn trim_to<'a>(&'a self, horizon: i64) -> BoxedFuture<'a, Result<i64>> {
|
|
let name = self.name.clone();
|
|
let state = self.state.clone();
|
|
Box::pin(async move {
|
|
let mut s = state.streams.guard();
|
|
Ok(trim_events(&mut s, &name, horizon))
|
|
})
|
|
}
|
|
|
|
fn subscribe<'a>(&'a self, consumer: &str) -> BoxedFuture<'a, Result<Box<dyn EventReceiver>>> {
|
|
if let Err(e) = validate_local_name(consumer) {
|
|
return Box::pin(async move { Err(e) });
|
|
}
|
|
let name = self.name.clone();
|
|
let consumer = consumer.to_string();
|
|
let state = self.state.clone();
|
|
let wakes = self.wakes.clone();
|
|
Box::pin(async move {
|
|
// Attach first (starts from "now"; attach-after-commit
|
|
// no-replay is structural), then read the stored offset and
|
|
// drain to the tail.
|
|
let feed = wakes.attach(&stream_wake_channel(&name));
|
|
let mut cursor;
|
|
let mut pending = VecDeque::new();
|
|
{
|
|
let s = state.streams.guard();
|
|
cursor = get_offset_row(&s, &name, &consumer);
|
|
loop {
|
|
let rows = read_events(&s, &name, cursor, SUBSCRIBE_PAGE);
|
|
let got = rows.len();
|
|
for row in rows {
|
|
cursor = row.offset;
|
|
pending.push_back(event_from_row(row, &name));
|
|
}
|
|
if got < SUBSCRIBE_PAGE as usize {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
Ok(Box::new(MemEventReceiver {
|
|
stream: name,
|
|
consumer,
|
|
state,
|
|
feed,
|
|
pending,
|
|
cursor,
|
|
position: 0,
|
|
}) as Box<dyn EventReceiver>)
|
|
})
|
|
}
|
|
}
|
|
|
|
/// The durable subscription receiver (see the module docs). Its
|
|
/// position is the last-*yielded* event's offset, so `save_offset()`
|
|
/// never checkpoints an event the consumer did not observe.
|
|
pub struct MemEventReceiver {
|
|
stream: String,
|
|
consumer: String,
|
|
state: Arc<State>,
|
|
feed: crate::wakes::WakeFeed,
|
|
pending: VecDeque<StreamEvent>,
|
|
cursor: i64,
|
|
position: i64,
|
|
}
|
|
|
|
impl std::fmt::Debug for MemEventReceiver {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("MemEventReceiver")
|
|
.field("stream", &self.stream)
|
|
.finish_non_exhaustive()
|
|
}
|
|
}
|
|
|
|
impl MemEventReceiver {
|
|
/// One forward read from the cursor into the pending buffer.
|
|
fn drain_a_page(&mut self) -> Result<()> {
|
|
let events = {
|
|
let s = self.state.streams.guard();
|
|
read_events(&s, &self.stream, self.cursor, SUBSCRIBE_PAGE)
|
|
};
|
|
for row in events {
|
|
self.cursor = row.offset;
|
|
self.pending.push_back(event_from_row(row, &self.stream));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn pop_positional(&mut self) -> Option<StreamEvent> {
|
|
let event = self.pending.pop_front()?;
|
|
self.position = event.offset;
|
|
Some(event)
|
|
}
|
|
}
|
|
|
|
impl EventReceiver for MemEventReceiver {
|
|
fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option<Result<StreamEvent>>> {
|
|
Box::pin(async move {
|
|
loop {
|
|
if let Some(event) = self.pop_positional() {
|
|
return Some(Ok(event));
|
|
}
|
|
// Park on the wake feed: a commit that lands after this
|
|
// park fires a wake that releases it; a commit that
|
|
// landed before it already buffered a wake (attach-first).
|
|
let _wake = match self.feed.rx().recv().await {
|
|
Some(wake) => wake,
|
|
None => return None,
|
|
};
|
|
if let Err(e) = self.drain_a_page() {
|
|
return Some(Err(database_only(e)));
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
fn try_recv(&mut self) -> Result<Option<StreamEvent>> {
|
|
if let Some(event) = self.pop_positional() {
|
|
return Ok(Some(event));
|
|
}
|
|
// A buffered wake is the hint that a commit landed which the
|
|
// drain has not seen yet; the poll drains it and re-reads once.
|
|
let closed = matches!(
|
|
self.feed.rx().try_recv(),
|
|
Err(tokio::sync::mpsc::error::TryRecvError::Disconnected)
|
|
);
|
|
self.drain_a_page()?;
|
|
if let Some(event) = self.pop_positional() {
|
|
return Ok(Some(event));
|
|
}
|
|
if closed {
|
|
// The bus close (engine drop) is the one close path: once
|
|
// the feed is disconnected and nothing is buffered or
|
|
// readable, Closed is the terminal poll arm (never reopens).
|
|
return Err(Error::Closed);
|
|
}
|
|
Ok(None)
|
|
}
|
|
|
|
fn read_since<'a>(
|
|
&'a mut self,
|
|
offset: i64,
|
|
limit: i64,
|
|
) -> BoxedFuture<'a, Result<Vec<StreamEvent>>> {
|
|
if limit <= 0 {
|
|
return Box::pin(async { Ok(Vec::new()) });
|
|
}
|
|
let events = {
|
|
let s = self.state.streams.guard();
|
|
read_events(&s, &self.stream, offset, limit)
|
|
};
|
|
let stream = self.stream.clone();
|
|
Box::pin(async move {
|
|
Ok(events
|
|
.into_iter()
|
|
.map(|r| event_from_row(r, &stream))
|
|
.collect())
|
|
})
|
|
}
|
|
|
|
fn save_offset(&mut self) -> Result<()> {
|
|
let position = self.position;
|
|
if position == 0 {
|
|
return Ok(());
|
|
}
|
|
let mut s = self.state.streams.guard();
|
|
save_offset_row(&mut s, &self.stream, &self.consumer, position);
|
|
Ok(())
|
|
}
|
|
|
|
fn offset(&self) -> i64 {
|
|
self.position
|
|
}
|
|
}
|
|
|
|
/// `recv()`'s `Err` arm carries `Database` only (ADR-021 §5): a read
|
|
/// error mapped elsewhere is remapped into the opaque fallback with
|
|
/// the original preserved as the source (the decode-side posture
|
|
/// guard — mem reads cannot fail in practice).
|
|
fn database_only(e: Error) -> Error {
|
|
match e {
|
|
already @ Error::Database(_) => already,
|
|
other => Error::database(other),
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::store::MemStore;
|
|
use alkstore::{EventReceiver, Store, StreamHandle};
|
|
|
|
fn handle(store: &MemStore, name: &str) -> MemStreamHandle {
|
|
MemStreamHandle::new(
|
|
name.to_string(),
|
|
store.state.clone(),
|
|
store.clock.clone(),
|
|
store.wakes.clone(),
|
|
)
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn publish_read_round_trip_is_global_fifo_by_offset() {
|
|
let store = MemStore::new();
|
|
let handle = handle(&store, "orders");
|
|
let a = handle.publish(serde_json::json!(1)).await.unwrap();
|
|
let b = handle.publish(serde_json::json!(2)).await.unwrap();
|
|
let keyed = handle
|
|
.publish_with_key(Some("cust-7".into()), serde_json::json!(3))
|
|
.await
|
|
.unwrap();
|
|
assert!(a < b && b < keyed, "offsets are monotone per stream");
|
|
let events = handle.read_since(0, 10).await.unwrap();
|
|
assert_eq!(events.len(), 3);
|
|
assert_eq!(
|
|
events.iter().map(|e| e.offset).collect::<Vec<_>>(),
|
|
vec![a, b, keyed],
|
|
"offset ASC — global FIFO regardless of keys"
|
|
);
|
|
assert_eq!(events[0].stream, "orders");
|
|
assert_eq!(events[0].key, None);
|
|
assert_eq!(events[2].key.as_deref(), Some("cust-7"));
|
|
assert_eq!(events[2].payload_as::<i64>().unwrap(), 3);
|
|
assert!(
|
|
events
|
|
.iter()
|
|
.all(|e| e.created_at == store.clock.unix_now())
|
|
);
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn empty_some_key_is_invalid_at_every_entry() {
|
|
let store = MemStore::new();
|
|
let handle = handle(&store, "orders");
|
|
let err = handle
|
|
.publish_with_key(Some(String::new()), serde_json::json!(1))
|
|
.await
|
|
.unwrap_err();
|
|
assert!(matches!(err, alkstore::Error::InvalidName { .. }));
|
|
let err = handle
|
|
.publish_with_key(Some(" ".into()), serde_json::json!(1))
|
|
.await
|
|
.unwrap_err();
|
|
assert!(matches!(err, alkstore::Error::InvalidName { .. }));
|
|
let mut tx = store.begin_tx().await.unwrap();
|
|
let err = tx
|
|
.publish_with_key_tx("orders", Some(String::new()), serde_json::json!(1))
|
|
.await
|
|
.unwrap_err();
|
|
assert!(matches!(err, alkstore::Error::InvalidName { .. }));
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn read_extents_clamp_empty() {
|
|
let store = MemStore::new();
|
|
let handle = handle(&store, "orders");
|
|
handle.publish(serde_json::json!(1)).await.unwrap();
|
|
assert!(handle.read_since(0, 0).await.unwrap().is_empty());
|
|
assert!(
|
|
handle.read_since(0, -3).await.unwrap().is_empty(),
|
|
"the LIMIT -1 dialect artifact is dead"
|
|
);
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn saves_are_monotone_and_absent_consumers_read_zero() {
|
|
let store = MemStore::new();
|
|
let handle = handle(&store, "orders");
|
|
assert_eq!(
|
|
handle.get_offset("c1").await.unwrap(),
|
|
0,
|
|
"an absent consumer reads 0"
|
|
);
|
|
let tail = handle.publish(serde_json::json!(1)).await.unwrap();
|
|
handle.save_offset("c1", tail).await.unwrap();
|
|
assert_eq!(handle.get_offset("c1").await.unwrap(), tail);
|
|
handle.save_offset("c1", tail - 1).await.unwrap();
|
|
assert_eq!(
|
|
handle.get_offset("c1").await.unwrap(),
|
|
tail,
|
|
"a below-checkpoint save is a silent no-op (monotone)"
|
|
);
|
|
handle.save_offset("c1", tail).await.unwrap();
|
|
assert_eq!(
|
|
handle.get_offset("c1").await.unwrap(),
|
|
tail,
|
|
"an equal save is also a no-op"
|
|
);
|
|
handle.save_offset("c1", tail + 5).await.unwrap();
|
|
assert_eq!(handle.get_offset("c1").await.unwrap(), tail + 5);
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn trim_is_exact_boundary_and_wakes_nothing() {
|
|
let store = MemStore::new();
|
|
let handle = handle(&store, "orders");
|
|
let a = handle.publish(serde_json::json!(1)).await.unwrap();
|
|
let b = handle.publish(serde_json::json!(2)).await.unwrap();
|
|
let c = handle.publish(serde_json::json!(3)).await.unwrap();
|
|
assert_eq!(
|
|
handle.trim_to(b - 1).await.unwrap(),
|
|
1,
|
|
"exact boundary: offset <= horizon"
|
|
);
|
|
let events = handle.read_since(0, 10).await.unwrap();
|
|
assert_eq!(
|
|
events.iter().map(|e| e.offset).collect::<Vec<_>>(),
|
|
vec![b, c],
|
|
"surviving rows keep their offsets, never renumbered"
|
|
);
|
|
assert_eq!(
|
|
handle.trim_to(-100).await.unwrap(),
|
|
0,
|
|
"a negative horizon deletes nothing"
|
|
);
|
|
assert_eq!(handle.trim_to(c).await.unwrap(), 2);
|
|
assert_eq!(handle.trim_to(c).await.unwrap(), 0, "idempotent");
|
|
// Reads from a trimmed region resume at the horizon's first
|
|
// remaining row.
|
|
assert!(handle.read_since(0, 10).await.unwrap().is_empty());
|
|
let _ = a;
|
|
assert_eq!(handle.get_offset("late").await.unwrap(), 0);
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn subscriber_replays_from_the_saved_offset_and_ride_wakes() {
|
|
let store = MemStore::new();
|
|
let handle = handle(&store, "orders");
|
|
let _a = handle.publish(serde_json::json!(1)).await.unwrap();
|
|
let b = handle.publish(serde_json::json!(2)).await.unwrap();
|
|
handle.save_offset("c1", b).await.unwrap();
|
|
let mut receiver = handle.subscribe("c1").await.unwrap();
|
|
// Everything after the checkpoint replays on attach.
|
|
let c = handle.publish(serde_json::json!(3)).await.unwrap();
|
|
let first = receiver.recv().await.unwrap().unwrap();
|
|
assert_eq!(
|
|
first.offset, c,
|
|
"replay starts at the stored checkpoint's forward tail"
|
|
);
|
|
assert_eq!(first.payload_as::<i64>().unwrap(), 3);
|
|
// Post-attach publishes deliver without polling.
|
|
let d = handle.publish(serde_json::json!(4)).await.unwrap();
|
|
let second = receiver.recv().await.unwrap().unwrap();
|
|
assert_eq!(second.offset, d);
|
|
// The explicit-save receiver form and the direct form compose.
|
|
receiver.save_offset().unwrap();
|
|
assert_eq!(receiver.offset(), d);
|
|
assert_eq!(
|
|
handle.get_offset("c1").await.unwrap(),
|
|
d,
|
|
"the receiver saves the last-yielded offset"
|
|
);
|
|
// A save before any yield is a no-op on the stored checkpoint.
|
|
let mut fresh = handle.subscribe("c2").await.unwrap();
|
|
assert_eq!(fresh.offset(), 0);
|
|
fresh.save_offset().unwrap();
|
|
assert_eq!(handle.get_offset("c2").await.unwrap(), 0);
|
|
let _ = _a;
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn read_from_consumer_reads_the_cursor_and_clamps_the_extent() {
|
|
let store = MemStore::new();
|
|
let handle = handle(&store, "orders");
|
|
let a = handle.publish(serde_json::json!(1)).await.unwrap();
|
|
let b = handle.publish(serde_json::json!(2)).await.unwrap();
|
|
let c = handle.publish(serde_json::json!(3)).await.unwrap();
|
|
let head = handle.read_from_consumer("c1", 10).await.unwrap();
|
|
assert_eq!(
|
|
head.iter().map(|e| e.offset).collect::<Vec<_>>(),
|
|
vec![a, b, c],
|
|
"an absent consumer reads from the head (checkpoint 0)"
|
|
);
|
|
handle.save_offset("c1", b).await.unwrap();
|
|
let cursor = handle.read_from_consumer("c1", 10).await.unwrap();
|
|
assert_eq!(
|
|
cursor.iter().map(|e| e.offset).collect::<Vec<_>>(),
|
|
vec![c],
|
|
"the cursor read resumes at the saved checkpoint's forward tail"
|
|
);
|
|
assert!(handle.read_from_consumer("c1", 0).await.unwrap().is_empty());
|
|
assert!(
|
|
handle
|
|
.read_from_consumer("c1", -2)
|
|
.await
|
|
.unwrap()
|
|
.is_empty(),
|
|
"the extent guard clamps non-positive limits to empty"
|
|
);
|
|
handle.save_offset("c1", c).await.unwrap();
|
|
assert!(
|
|
handle
|
|
.read_from_consumer("c1", 10)
|
|
.await
|
|
.unwrap()
|
|
.is_empty(),
|
|
"a cursor at the tail reads empty"
|
|
);
|
|
match handle.read_from_consumer("", 10).await {
|
|
Err(alkstore::Error::InvalidName { .. }) => {}
|
|
_other => panic!("an empty consumer name rejects"),
|
|
}
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn saved_sub_horizon_offsets_survive_trim_and_consumer_reads_resume_at_the_horizon() {
|
|
let store = MemStore::new();
|
|
let handle = handle(&store, "orders");
|
|
let a = handle.publish(serde_json::json!(1)).await.unwrap();
|
|
let b = handle.publish(serde_json::json!(2)).await.unwrap();
|
|
let _c = handle.publish(serde_json::json!(3)).await.unwrap();
|
|
handle.save_offset("c1", a).await.unwrap();
|
|
assert_eq!(handle.trim_to(b - 1).await.unwrap(), 1);
|
|
assert_eq!(
|
|
handle.get_offset("c1").await.unwrap(),
|
|
a,
|
|
"the saved sub-horizon checkpoint is untouched by trim — offsets are never renumbered"
|
|
);
|
|
let resumed = handle.read_from_consumer("c1", 10).await.unwrap();
|
|
assert_eq!(
|
|
resumed.iter().map(|e| e.offset).collect::<Vec<_>>(),
|
|
vec![b, _c],
|
|
"reads from a trimmed-away region resume at the horizon's first remaining row"
|
|
);
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn receiver_try_recv_reads_without_await() {
|
|
let store = MemStore::new();
|
|
let handle = handle(&store, "orders");
|
|
let mut receiver = handle.subscribe("c1").await.unwrap();
|
|
assert!(try_none(receiver.as_mut()), "idle reads as Ok(None)");
|
|
let a = handle.publish(serde_json::json!(1)).await.unwrap();
|
|
let got = receiver.try_recv().unwrap().unwrap();
|
|
assert_eq!(got.offset, a);
|
|
assert!(try_none(receiver.as_mut()));
|
|
}
|
|
|
|
fn try_none(receiver: &mut dyn EventReceiver) -> bool {
|
|
matches!(receiver.try_recv(), Ok(None))
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn engine_drop_closes_the_receiver_on_both_arms() {
|
|
let (mut receiver, offset) = {
|
|
let store = MemStore::new();
|
|
let handle = handle(&store, "orders");
|
|
let offset = handle.publish(serde_json::json!(1)).await.unwrap();
|
|
(handle.subscribe("c1").await.unwrap(), offset)
|
|
};
|
|
// The engine is gone; the replayed pre-drop event still
|
|
// delivers (it was committed and visible), then the receiver
|
|
// is closed — terminal on both arms (never reopens).
|
|
let delivered = receiver
|
|
.try_recv()
|
|
.unwrap()
|
|
.expect("the pre-drop event replays");
|
|
assert_eq!(delivered.offset, offset);
|
|
assert!(matches!(receiver.try_recv(), Err(alkstore::Error::Closed)));
|
|
let closed = tokio::time::timeout(std::time::Duration::from_millis(200), receiver.recv())
|
|
.await
|
|
.expect("the closed feed parks no receiver");
|
|
assert!(
|
|
closed.is_none(),
|
|
"the terminal None arm (engine drop is the close cause)"
|
|
);
|
|
}
|
|
|
|
#[tokio::test(flavor = "current_thread")]
|
|
async fn trim_wakes_nothing_for_an_open_subscriber() {
|
|
let store = MemStore::new();
|
|
let handle = handle(&store, "orders");
|
|
let a = handle.publish(serde_json::json!(1)).await.unwrap();
|
|
let b = handle.publish(serde_json::json!(2)).await.unwrap();
|
|
let mut receiver = handle.subscribe("c1").await.unwrap();
|
|
assert_eq!(receiver.recv().await.unwrap().unwrap().offset, a);
|
|
assert_eq!(receiver.recv().await.unwrap().unwrap().offset, b);
|
|
handle.trim_to(b).await.unwrap();
|
|
let quiet =
|
|
tokio::time::timeout(std::time::Duration::from_millis(50), receiver.recv()).await;
|
|
assert!(quiet.is_err(), "trim emits no wake and delivers no event");
|
|
let _ = receiver.read_since(0, 10).await.unwrap();
|
|
}
|
|
}
|