mem engine streams — coverage reconciliation against the seam task's foundation with the two unpinned engine-side pins (mem-engine-streams): the read_from_consumer cursor-read surface (absent-consumer head read, checkpoint-forward resume, tail-empty, extent clamps, name validation) and the ADR-015 trim property (saved sub-horizon checkpoints untouched — offsets never renumbered; consumer reads resume at the horizon's first remaining row). No production code changed; tests only (+2, mem suite at 99). Task file: status completed, Notes/Summary filled

This commit is contained in:
glm-5.3-flash committed 2026-10-10 12:17:10 +00:00
1 parent f37fb0d17d
commit 3eca862f84
2 files changed
+128 -8

No files matched your search

+66
View File
@@ -644,6 +644,72 @@ mod tests {
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();