use std::sync::Arc;
use armdb::armour::{Db, TypedMigration};
use armdb::schema::{MermaidEr, SchemaFormatter, SchemaRegistry};
use armdb::{
CollectionMeta, Config, ConstMap, DbError, EmitSink, Events, FixedConfig, FixedMap,
FormatMigration, IndexTargets, MigrateAction, NoHook, Projected, Queue, QueueConfig,
QueueTargets, RapiraCodec, TtlHook, TtlPolicy, TypedTree, TypedWriteHook, ZeroHookAdapter,
queue,
};
use armour_core::{CollectionKind, GetType, RelationKind};
use rapira::Rapira;
#[derive(
Clone,
Copy,
Debug,
PartialEq,
Eq,
GetType,
zerocopy::FromBytes,
zerocopy::IntoBytes,
zerocopy::Immutable,
zerocopy::KnownLayout,
)]
#[repr(transparent)]
pub struct FixedJob(u64);
#[derive(Clone, Debug, PartialEq, Eq, armour_core::GetType, rapira::Rapira)]
pub struct VarJob(String);
#[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
struct Source(u64);
impl CollectionMeta for Source {
type SelfId = [u8; 8];
const NAME: &'static str = "queue_stream_source";
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
struct StreamMigrationV1(u64);
impl CollectionMeta for StreamMigrationV1 {
type SelfId = [u8; 8];
const NAME: &'static str = "queue_stream_migration_source";
const VERSION: u16 = 1;
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
struct StreamMigrationV2(u64);
const STREAM_FORMAT: FormatMigration = armdb::format_step!(
1,
StreamMigrationV1 => StreamMigrationV2,
RapiraCodec,
|old| Ok::<_, DbError>(Some(StreamMigrationV2(old.0 + 1)))
);
impl CollectionMeta for StreamMigrationV2 {
type SelfId = [u8; 8];
const NAME: &'static str = "queue_stream_migration_source";
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[STREAM_FORMAT];
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
struct TypedStreamMigrationV1(u64);
impl CollectionMeta for TypedStreamMigrationV1 {
type SelfId = [u8; 8];
const NAME: &'static str = "queue_stream_typed_migration_source";
const VERSION: u16 = 1;
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
struct TypedStreamMigrationV2(u64);
impl CollectionMeta for TypedStreamMigrationV2 {
type SelfId = [u8; 8];
const NAME: &'static str = "queue_stream_typed_migration_source";
const VERSION: u16 = 2;
}
fn bump_typed_stream_migration(
_: &[u8; 8],
row: &TypedStreamMigrationV2,
) -> MigrateAction<TypedStreamMigrationV2> {
MigrateAction::Update(TypedStreamMigrationV2(row.0 + 1))
}
const TYPED_STREAM_MIGRATIONS: &[TypedMigration<[u8; 8], TypedStreamMigrationV2>] =
&[(1, bump_typed_stream_migration)];
#[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
struct TtlSource(u64);
impl CollectionMeta for TtlSource {
type SelfId = [u8; 8];
const NAME: &'static str = "queue_stream_ttl_source";
const TTL: Option<TtlPolicy<Self>> = Some(TtlPolicy {
revision: 1,
deadline: |_, _| None,
});
}
#[derive(
Clone,
Copy,
Debug,
PartialEq,
Eq,
GetType,
Rapira,
zerocopy::FromBytes,
zerocopy::IntoBytes,
zerocopy::Immutable,
zerocopy::KnownLayout,
)]
#[repr(transparent)]
struct MapSource(u64);
impl CollectionMeta for MapSource {
type SelfId = [u8; 8];
const NAME: &'static str = "queue_stream_map_source";
}
queue!(FixedJobs, payload = FixedJob, name = "queue_stream_fixed");
queue!(
OtherFixedJobs,
payload = FixedJob,
name = "queue_stream_other_fixed"
);
queue!(VarJobs, payload = VarJob, name = "queue_stream_var", var);
queue!(SchemaJobs, payload = VarJob, name = "schema_jobs", var);
struct FixedFixture {
queue: Arc<Queue<FixedJobs>>,
_db: Db,
_dir: tempfile::TempDir,
}
fn open_fixed_jobs() -> FixedFixture {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let queue = db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
FixedFixture {
queue,
_db: db,
_dir: dir,
}
}
struct OtherFixedFixture {
queue: Arc<Queue<OtherFixedJobs>>,
_db: Db,
_dir: tempfile::TempDir,
}
fn open_other_fixed_jobs() -> OtherFixedFixture {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let queue = db
.open_queue::<OtherFixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
OtherFixedFixture {
queue,
_db: db,
_dir: dir,
}
}
struct VarFixture {
queue: Arc<Queue<VarJobs>>,
_db: Db,
_dir: tempfile::TempDir,
}
fn open_var() -> VarFixture {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let queue = db
.open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
VarFixture {
queue,
_db: db,
_dir: dir,
}
}
fn open_schema_queue() -> (
SchemaRegistry,
Arc<Queue<SchemaJobs>>,
Db,
tempfile::TempDir,
) {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<SchemaJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
(SchemaRegistry::new(), jobs, db, dir)
}
#[test]
fn queue_registration_classifies_both_physical_names_and_stream_graph_edge() {
let (mut reg, jobs, _db, _dir) = open_schema_queue();
reg.register_queue(&jobs);
let _hook = reg.change_stream::<Source, _>(
QueueTargets::new().with(&jobs),
|_key, _old, _new, _sink| Ok(()),
);
let graph = reg.finish().unwrap().graph();
assert!(
graph
.collections
.iter()
.any(|n| { n.name == "schema_jobs" && n.kind == CollectionKind::Queue })
);
assert!(
graph
.collections
.iter()
.any(|n| { n.name == "schema_jobs.payload" && n.kind == CollectionKind::Queue })
);
assert!(
graph
.relations
.iter()
.any(|r| matches!(r.kind, RelationKind::Stream))
);
assert!(
MermaidEr
.format(&graph)
.contains("||--o{ schema_jobs : \"stream\"")
);
}
#[test]
fn change_stream_on_records_a_stream_graph_edge() {
let (mut reg, jobs, _db, _dir) = open_schema_queue();
reg.register_queue(&jobs);
let base = Projected::<[u8; 8], Source>::new(IndexTargets::new(), |_, _, _| Ok(()));
let _hook = reg.change_stream_on(
base,
QueueTargets::new().with(&jobs),
|_key, _old, _new, _sink| Ok(()),
);
assert!(
reg.finish()
.unwrap()
.graph()
.relations
.iter()
.any(|r| r.from == Source::NAME
&& r.to == "schema_jobs"
&& matches!(r.kind, RelationKind::Stream))
);
}
#[test]
#[should_panic(expected = "already declares a stream")]
fn schema_stream_rejects_a_duplicate_source_declaration() {
let (mut reg, jobs, _db, _dir) = open_schema_queue();
reg.register_queue(&jobs);
let targets = QueueTargets::new().with(&jobs);
let _first = reg.change_stream::<Source, _>(targets, |_key, _old, _new, _sink| Ok(()));
let _second = reg.change_stream::<Source, _>(
QueueTargets::new().with(&jobs),
|_key, _old, _new, _sink| Ok(()),
);
}
#[test]
#[should_panic(expected = "cannot declare TTL")]
fn schema_stream_rejects_a_ttl_source() {
let (mut reg, jobs, _db, _dir) = open_schema_queue();
reg.register_queue(&jobs);
let _hook = reg.change_stream::<TtlSource, _>(
QueueTargets::new().with(&jobs),
|_key, _old, _new, _sink| Ok(()),
);
}
fn key(id: u64) -> [u8; 8] {
id.to_be_bytes()
}
type StreamHook = Projected<[u8; 8], Source, NoHook, Events<[u8; 8], Source>>;
type StreamTree = TypedTree<[u8; 8], Source, RapiraCodec, TtlHook<Source, StreamHook>>;
fn stream_hook(jobs: &Arc<Queue<FixedJobs>>) -> StreamHook {
let targets = QueueTargets::new().with(jobs);
Projected::<[u8; 8], Source>::new(IndexTargets::new(), |_, _, _| Ok(()))
.and_events(targets, |_, _, _, _| {
panic!("non-transactional write invoked stream closure")
})
}
fn bulk_stream_hook(jobs: &Arc<Queue<FixedJobs>>) -> StreamHook {
let stream_jobs = Arc::clone(jobs);
Projected::<[u8; 8], Source>::new(IndexTargets::new(), |_, _, _| Ok(())).and_events(
QueueTargets::new().with(jobs),
move |_key, old, new, sink| {
if let (Some(old), None) = (old, new) {
sink.emit(&stream_jobs, FixedJob(old.0))?;
}
Ok(())
},
)
}
fn map_stream_hook(
jobs: &Arc<Queue<FixedJobs>>,
) -> Projected<[u8; 8], MapSource, NoHook, Events<[u8; 8], MapSource>> {
Projected::<[u8; 8], MapSource>::new(IndexTargets::new(), |_, _, _| Ok(()))
.and_events(QueueTargets::new().with(jobs), |_, _, _, _| Ok(()))
}
fn map_stream_var_hook(
jobs: &Arc<Queue<VarJobs>>,
) -> Projected<[u8; 8], MapSource, NoHook, Events<[u8; 8], MapSource>> {
let stream_jobs = Arc::clone(jobs);
Projected::<[u8; 8], MapSource>::new(IndexTargets::new(), |_, _, _| Ok(())).and_events(
QueueTargets::new().with(jobs),
move |_, _, new, sink| {
if let Some(new) = new {
sink.emit(&stream_jobs, VarJob(format!("map-job-{}", new.0)))?;
}
Ok(())
},
)
}
struct StreamFixture {
source: Arc<StreamTree>,
jobs: Arc<Queue<FixedJobs>>,
_db: Db,
_dir: tempfile::TempDir,
}
fn open_stream_fixture() -> StreamFixture {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let source = db
.open_typed_tree::<Source, RapiraCodec, _>(Config::test(), stream_hook(&jobs), &[])
.unwrap();
StreamFixture {
source,
jobs,
_db: db,
_dir: dir,
}
}
fn open_bulk_stream_fixture() -> StreamFixture {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let source = db
.open_typed_tree::<Source, RapiraCodec, _>(Config::test(), bulk_stream_hook(&jobs), &[])
.unwrap();
StreamFixture {
source,
jobs,
_db: db,
_dir: dir,
}
}
fn seed_stream_source(fixture: &StreamFixture) {
fixture
._db
.tx()
.put(&fixture.source, &key(1), &Source(10))
.unwrap()
.put(&fixture.source, &key(2), &Source(20))
.unwrap()
.commit()
.unwrap();
}
#[test]
fn bulk_stream_refusals_leave_every_source_row_unchanged() {
let fixture = open_bulk_stream_fixture();
seed_stream_source(&fixture);
let before = [
(key(1), *fixture.source.get(&key(1)).unwrap()),
(key(2), *fixture.source.get(&key(2)).unwrap()),
];
let scope_key = key(0);
assert_eq!(fixture.source.scope::<0>(&scope_key).count().unwrap(), 2);
assert!(!fixture.source.scope::<0>(&scope_key).is_empty().unwrap());
assert!(matches!(
fixture.source.scope::<0>(&scope_key).delete_all(),
Err(DbError::ChangeStreamRequiresTx {
collection: "queue_stream_source",
queue: "queue_stream_fixed",
})
));
assert_eq!(
[
(key(1), *fixture.source.get(&key(1)).unwrap()),
(key(2), *fixture.source.get(&key(2)).unwrap()),
],
before
);
assert!(matches!(
fixture.source.scope::<0>(&scope_key).delete_up_to(1),
Err(DbError::ChangeStreamRequiresTx {
collection: "queue_stream_source",
queue: "queue_stream_fixed",
})
));
assert_eq!(
[
(key(1), *fixture.source.get(&key(1)).unwrap()),
(key(2), *fixture.source.get(&key(2)).unwrap()),
],
before
);
assert!(matches!(
fixture.source.scope::<0>(&scope_key).retain(|_, _| false),
Err(DbError::ChangeStreamRequiresTx {
collection: "queue_stream_source",
queue: "queue_stream_fixed",
})
));
assert_eq!(
[
(key(1), *fixture.source.get(&key(1)).unwrap()),
(key(2), *fixture.source.get(&key(2)).unwrap()),
],
before
);
}
#[test]
fn bulk_stream_map_refusal_is_preflight() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let mut config = Config::test();
config.iterable = true;
let source = db
.open_typed_map::<MapSource, RapiraCodec, _>(config, map_stream_hook(&jobs), &[])
.unwrap();
db.tx()
.put(&source, &key(1), &MapSource(1))
.unwrap()
.put(&source, &key(2), &MapSource(2))
.unwrap()
.commit()
.unwrap();
assert!(matches!(
source.scope::<0>(&key(0)).unwrap().delete_all(),
Err(DbError::ChangeStreamRequiresTx { .. })
));
assert_eq!(source.scope::<0>(&key(0)).unwrap().count().unwrap(), 2);
}
#[test]
fn bulk_tx_delete_scope_applies_stream_source_and_queue_rows() {
let fixture = open_bulk_stream_fixture();
seed_stream_source(&fixture);
let scope = fixture.source.tx_scope::<0>(&key(0));
let (tx, selection) = fixture._db.tx().delete_scope(&scope, 2).unwrap();
assert_eq!(selection.queued, 2);
tx.durable().commit().unwrap();
assert_eq!(fixture.source.scope::<0>(&key(0)).count().unwrap(), 0);
assert_eq!(fixture.jobs.len(), 2);
}
#[test]
fn one_source_commit_emits_one_queue_row_and_repeated_source_writes_collapse() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let stream_jobs = Arc::clone(&jobs);
let targets = QueueTargets::new().with(&jobs);
let source = db
.open_typed_tree::<Source, RapiraCodec, _>(
Config::test(),
Projected::<[u8; 8], Source>::new(IndexTargets::new(), |_, _, _| Ok(())).and_events(
targets,
move |_, _old, new, sink| {
if let Some(new) = new {
sink.emit(&stream_jobs, FixedJob(new.0))?;
}
Ok(())
},
),
&[],
)
.unwrap();
db.tx()
.put(&source, &key(1), &Source(1))
.unwrap()
.put(&source, &key(1), &Source(2))
.unwrap()
.commit()
.unwrap();
let leased = jobs.lease(8, None).unwrap();
assert_eq!(leased.len(), 1);
assert_eq!(leased[0].payload(), &FixedJob(2));
}
#[test]
fn stream_emissions_share_the_transactions_max_depth_admission() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<FixedJobs>(QueueConfig::balanced().max_depth(Some(1)).build().unwrap())
.unwrap();
let stream_jobs = Arc::clone(&jobs);
let source = db
.open_typed_tree::<Source, RapiraCodec, _>(
Config::test(),
Projected::<[u8; 8], Source>::new(IndexTargets::new(), |_, _, _| Ok(())).and_events(
QueueTargets::new().with(&jobs),
move |_, _, new, sink| {
if let Some(new) = new {
sink.emit(&stream_jobs, FixedJob(new.0))?;
}
Ok(())
},
),
&[],
)
.unwrap();
let result = db
.tx()
.put(&source, &key(1), &Source(1))
.unwrap()
.put(&source, &key(2), &Source(2))
.unwrap()
.commit();
assert!(matches!(result, Err(DbError::QueueFull)));
assert!(source.get(&key(1)).is_none());
assert!(source.get(&key(2)).is_none());
assert_eq!(jobs.len(), 0);
}
#[test]
fn stream_emission_consumes_a_same_transaction_ack_credit_at_max_depth() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<FixedJobs>(QueueConfig::balanced().max_depth(Some(1)).build().unwrap())
.unwrap();
let stream_jobs = Arc::clone(&jobs);
let source = db
.open_typed_tree::<Source, RapiraCodec, _>(
Config::test(),
Projected::<[u8; 8], Source>::new(IndexTargets::new(), |_, _, _| Ok(())).and_events(
QueueTargets::new().with(&jobs),
move |_, _, new, sink| {
if let Some(new) = new {
sink.emit(&stream_jobs, FixedJob(new.0))?;
}
Ok(())
},
),
&[],
)
.unwrap();
jobs.push(FixedJob(0), None).unwrap();
let token = jobs.lease(1, None).unwrap().pop().unwrap().token();
db.tx()
.put(&source, &key(1), &Source(1))
.unwrap()
.ack(&jobs, token)
.unwrap()
.commit()
.unwrap();
assert_eq!(jobs.len(), 1);
let leased = jobs.lease(1, None).unwrap().pop().unwrap();
assert_eq!(leased.payload(), &FixedJob(1));
let result = db
.tx()
.put(&source, &key(2), &Source(2))
.unwrap()
.put(&source, &key(3), &Source(3))
.unwrap()
.ack(&jobs, leased.token())
.unwrap()
.commit();
assert!(matches!(result, Err(DbError::QueueFull)));
assert!(source.get(&key(2)).is_none());
assert!(source.get(&key(3)).is_none());
assert_eq!(jobs.len(), 1);
jobs.ack(leased.token()).unwrap();
assert_eq!(jobs.len(), 0);
}
#[test]
fn format_migration_emits_no_persistent_change_stream_event() {
let dir = tempfile::tempdir().unwrap();
{
let db = Db::open_test(dir.path()).unwrap();
let source = db
.open_typed_tree::<StreamMigrationV1, RapiraCodec, _>(Config::test(), NoHook, &[])
.unwrap();
source.put(&key(1), StreamMigrationV1(1)).unwrap();
}
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let stream_jobs = Arc::clone(&jobs);
let source = db
.open_typed_tree::<StreamMigrationV2, RapiraCodec, _>(
Config::test(),
Projected::<[u8; 8], StreamMigrationV2>::new(IndexTargets::new(), |_, _, _| Ok(()))
.and_events(QueueTargets::new().with(&jobs), move |_, _, new, sink| {
if let Some(new) = new {
sink.emit(&stream_jobs, FixedJob(new.0))?;
}
Ok(())
}),
&[],
)
.unwrap();
assert_eq!(
source.get(&key(1)).map(|row| *row),
Some(StreamMigrationV2(2))
);
assert_eq!(jobs.len(), 0, "migrations are hook-free");
}
#[test]
fn typed_update_migration_emits_no_persistent_change_stream_event() {
let dir = tempfile::tempdir().unwrap();
{
let db = Db::open_test(dir.path()).unwrap();
let source = db
.open_typed_tree::<TypedStreamMigrationV1, RapiraCodec, _>(Config::test(), NoHook, &[])
.unwrap();
source.put(&key(1), TypedStreamMigrationV1(1)).unwrap();
}
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let stream_jobs = Arc::clone(&jobs);
let source = db
.open_typed_tree::<TypedStreamMigrationV2, RapiraCodec, _>(
Config::test(),
Projected::<[u8; 8], TypedStreamMigrationV2>::new(
IndexTargets::new(),
|_, _, _| Ok(()),
)
.and_events(QueueTargets::new().with(&jobs), move |_, _, new, sink| {
if let Some(new) = new {
sink.emit(&stream_jobs, FixedJob(new.0))?;
}
Ok(())
}),
TYPED_STREAM_MIGRATIONS,
)
.unwrap();
assert_eq!(
source.get(&key(1)).map(|row| *row),
Some(TypedStreamMigrationV2(2))
);
assert_eq!(jobs.len(), 0, "typed migrations are hook-free");
}
#[test]
fn one_source_commit_emits_one_var_header_and_payload() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let stream_jobs = Arc::clone(&jobs);
let source = db
.open_typed_tree::<Source, RapiraCodec, _>(
Config::test(),
Projected::<[u8; 8], Source>::new(IndexTargets::new(), |_, _, _| Ok(())).and_events(
QueueTargets::new().with(&jobs),
move |_, _, new, sink| {
if let Some(new) = new {
sink.emit(&stream_jobs, VarJob(format!("job-{}", new.0)))?;
}
Ok(())
},
),
&[],
)
.unwrap();
db.tx()
.put(&source, &key(1), &Source(7))
.unwrap()
.commit()
.unwrap();
assert_eq!(jobs.len(), 1, "the FixedStore header must be committed");
assert_eq!(
jobs.debug_payload_len(),
1,
"the Var payload must be committed"
);
let leased = jobs.lease(1, None).unwrap();
assert_eq!(leased.len(), 1);
assert_eq!(leased[0].payload(), &VarJob("job-7".into()));
}
#[test]
fn stream_callback_facade_reentry_returns_reentrant_tx_before_locking() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let source_slot = Arc::new(std::sync::Mutex::new(None::<Arc<StreamTree>>));
let callback_source = Arc::clone(&source_slot);
let source = db
.open_typed_tree::<Source, RapiraCodec, _>(
Config::test(),
Projected::<[u8; 8], Source>::new(IndexTargets::new(), |_, _, _| Ok(())).and_events(
QueueTargets::new().with(&jobs),
move |_, _, _, _| {
callback_source
.lock()
.unwrap()
.as_ref()
.unwrap()
.put(&key(2), Source(2))
.map(|_| ())
},
),
&[],
)
.unwrap();
*source_slot.lock().unwrap() = Some(Arc::clone(&source));
let result = db.tx().put(&source, &key(1), &Source(1)).unwrap().commit();
assert!(matches!(result, Err(DbError::ReentrantTx)));
assert!(source.get(&key(1)).is_none());
assert!(source.get(&key(2)).is_none());
}
#[test]
fn pre_log_guard_rejection_leaves_var_stream_source_queue_and_callback_unchanged() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let stream_jobs = Arc::clone(&jobs);
let callbacks = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let callbacks_in_hook = Arc::clone(&callbacks);
let source = db
.open_typed_tree::<Source, RapiraCodec, _>(
Config::test(),
Projected::<[u8; 8], Source>::new(IndexTargets::new(), |_, _, _| Ok(())).and_events(
QueueTargets::new().with(&jobs),
move |_, _, new, sink| {
callbacks_in_hook.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
if let Some(new) = new {
sink.emit(&stream_jobs, VarJob(format!("job-{}", new.0)))?;
}
Ok(())
},
),
&[],
)
.unwrap();
let result = db
.tx()
.put(&source, &key(1), &Source(7))
.unwrap()
.guard(&source, &key(2), |_| {
Err(DbError::Client("reject before log"))
})
.unwrap()
.commit();
assert!(matches!(result, Err(DbError::Client("reject before log"))));
assert!(source.get(&key(1)).is_none(), "source write must roll back");
assert_eq!(jobs.len(), 0, "header count must remain unchanged");
assert_eq!(
jobs.debug_payload_len(),
0,
"payload rows must remain unchanged"
);
assert_eq!(callbacks.load(std::sync::atomic::Ordering::SeqCst), 0);
}
fn open_direct_stream_source()
-> armdb::DbResult<TypedTree<[u8; 8], Source, RapiraCodec, StreamHook>> {
let dir = tempfile::tempdir().unwrap();
TypedTree::open_hooked(
dir.path(),
Config::test(),
RapiraCodec,
stream_hook(&open_fixed_jobs().queue),
)
}
fn open_source_with_other_db_queue() -> armdb::DbResult<Arc<StreamTree>> {
let source_dir = tempfile::tempdir().unwrap();
let target_dir = tempfile::tempdir().unwrap();
let source_db = Db::open_test(source_dir.path()).unwrap();
let target_db = Db::open_test(target_dir.path()).unwrap();
let _same_name_local_queue = source_db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let jobs = target_db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
source_db.open_typed_tree::<Source, RapiraCodec, _>(Config::test(), stream_hook(&jobs), &[])
}
#[test]
fn stream_sink_admits_only_the_declared_queue_by_arc_identity() {
let declared = open_fixed_jobs();
let foreign = open_other_fixed_jobs();
let targets = QueueTargets::new().with(&declared.queue);
let mut sink = EmitSink::debug_new(&targets, 8, usize::MAX);
sink.emit(&declared.queue, FixedJob(1)).unwrap();
assert!(matches!(
sink.emit(&foreign.queue, FixedJob(2)),
Err(DbError::Client("undeclared queue target"))
));
assert_eq!(sink.len(), 1);
}
#[test]
fn stream_sink_counts_fixed_and_var_physical_rows_before_transport() {
let fixed = open_fixed_jobs();
let var = open_var();
let fixed_targets = QueueTargets::new().with(&fixed.queue);
let var_targets = QueueTargets::new().with(&var.queue);
let mut fixed_sink = EmitSink::debug_new(&fixed_targets, 1, usize::MAX);
fixed_sink.emit(&fixed.queue, FixedJob(1)).unwrap();
assert!(matches!(
fixed_sink.emit(&fixed.queue, FixedJob(2)),
Err(DbError::ProjectionBudgetExceeded { rows: 2, .. })
));
let mut var_sink = EmitSink::debug_new(&var_targets, 2, usize::MAX);
var_sink
.emit_at(&var.queue, VarJob("one".into()), std::time::UNIX_EPOCH)
.unwrap();
assert!(matches!(
var_sink.emit(&var.queue, VarJob("two".into())),
Err(DbError::ProjectionBudgetExceeded { rows: 4, .. })
));
}
#[test]
fn stream_sink_rejects_the_exact_fixed_physical_byte_footprint() {
let fixed = open_fixed_jobs();
let targets = QueueTargets::new().with(&fixed.queue);
let mut sink = EmitSink::debug_new(&targets, 1, 4_143);
assert!(matches!(
sink.emit(&fixed.queue, FixedJob(1)),
Err(DbError::ProjectionBudgetExceeded {
rows: 1,
bytes: 4_144
})
));
assert!(sink.is_empty());
}
#[test]
fn stream_sink_counts_var_payload_wrapper_framing_in_its_byte_budget() {
let jobs = open_var();
let targets = QueueTargets::new().with(&jobs.queue);
let wrapper = rapira::serialize(&rapira::serialize(&VarJob("one".into())));
let expected = 56 + wrapper.len();
let mut sink = EmitSink::debug_new(&targets, 2, expected - 1);
assert!(matches!(
sink.emit(&jobs.queue, VarJob("one".into())),
Err(DbError::ProjectionBudgetExceeded { rows: 2, bytes }) if bytes == expected
));
}
#[test]
#[should_panic(expected = "QueueTargets: duplicate physical queue name")]
fn queue_targets_reject_duplicate_physical_names() {
let jobs = open_fixed_jobs();
let _ = QueueTargets::new().with(&jobs.queue).with(&jobs.queue);
}
#[test]
fn stream_sink_preserves_source_order_in_its_buffer() {
let jobs = open_fixed_jobs();
let targets = QueueTargets::new().with(&jobs.queue);
let mut sink = EmitSink::debug_new(&targets, 2, usize::MAX);
let first = std::time::UNIX_EPOCH + std::time::Duration::from_secs(1);
let second = std::time::UNIX_EPOCH + std::time::Duration::from_secs(2);
sink.emit_at(&jobs.queue, FixedJob(1), first).unwrap();
sink.emit_at(&jobs.queue, FixedJob(2), second).unwrap();
assert_eq!(
armdb::testing::stream_visible_at(&sink),
vec![first, second]
);
}
#[test]
fn projected_carrier_exposes_a_transition_stream_once() {
let jobs = open_fixed_jobs();
let targets = QueueTargets::new().with(&jobs.queue);
let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
let seen_by_project = Arc::clone(&seen);
let hook = Projected::<[u8; 8], Source>::new(IndexTargets::new(), |_, _, _| Ok(())).and_events(
targets,
move |_key, old, new, sink| {
seen_by_project
.lock()
.unwrap()
.push((old.map(|value| value.0), new.map(|value| value.0)));
if old != new {
sink.emit(&jobs.queue, FixedJob(new.expect("new snapshot").0))?;
}
Ok(())
},
);
assert!(!<Projected<[u8; 8], Source> as TypedWriteHook<[u8; 8], Source>>::EMITS);
fn emits<H: TypedWriteHook<[u8; 8], Source>>(_: &H) -> bool {
H::EMITS
}
assert!(emits(&hook));
assert!(hook.stream_targets().is_some());
let mut sink = EmitSink::debug_new(hook.stream_targets().unwrap(), 1, usize::MAX);
hook.stream_project(&[0; 8], Some(&Source(4)), Some(&Source(4)), &mut sink)
.unwrap();
assert_eq!(*seen.lock().unwrap(), vec![(Some(4), Some(4))]);
}
#[test]
fn a_stream_source_refuses_non_transactional_writes_without_a_trace() {
let fixture = open_stream_fixture();
let err = match fixture.source.put(&key(1), Source(1)) {
Ok(_) => panic!("stream source write unexpectedly succeeded"),
Err(err) => err,
};
assert!(matches!(err, DbError::ChangeStreamRequiresTx { .. }));
assert!(fixture.source.get(&key(1)).is_none());
assert_eq!(fixture.jobs.len(), 0);
}
#[test]
fn a_stream_source_refuses_atomic_writes_without_a_trace() {
let fixture = open_stream_fixture();
let err = match fixture
.source
.atomic(&key(1), |shard| shard.put(&key(1), Source(1)).map(|_| ()))
{
Ok(_) => panic!("stream source atomic write unexpectedly succeeded"),
Err(err) => err,
};
assert!(matches!(err, DbError::ChangeStreamRequiresTx { .. }));
assert!(fixture.source.get(&key(1)).is_none());
assert_eq!(fixture.jobs.len(), 0);
}
#[test]
fn a_var_typed_stream_source_refuses_atomic_insert_without_a_trace() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let source = db
.open_var_typed_tree::<Source, RapiraCodec, _>(Config::test(), stream_hook(&jobs), &[])
.unwrap();
let err = match source.atomic(&key(1), |shard| shard.insert(&key(1), &Source(1))) {
Ok(_) => panic!("var stream source atomic insert unexpectedly succeeded"),
Err(err) => err,
};
assert!(matches!(err, DbError::ChangeStreamRequiresTx { .. }));
assert!(source.get(&key(1)).is_none());
assert_eq!(jobs.len(), 0);
}
#[test]
fn direct_open_and_cross_db_target_are_rejected() {
assert!(matches!(
open_direct_stream_source(),
Err(DbError::ChangeStreamRequiresManagedOpen)
));
assert!(matches!(
open_source_with_other_db_queue(),
Err(DbError::Client(_))
));
}
#[test]
fn direct_const_map_stream_sources_are_rejected_for_both_backends() {
let jobs = open_fixed_jobs();
let hook = Projected::<[u8; 8], MapSource>::new(IndexTargets::new(), |_, _, _| Ok(()))
.and_events(QueueTargets::new().with(&jobs.queue), |_, _, _, _| Ok(()));
let bitcask_dir = tempfile::tempdir().unwrap();
assert!(matches!(
ConstMap::<[u8; 8], 8, ZeroHookAdapter<[u8; 8], MapSource, _>>::open_hooked(
bitcask_dir.path(),
Config::test(),
ZeroHookAdapter::new(hook),
),
Err(DbError::ChangeStreamRequiresManagedOpen)
));
let hook = Projected::<[u8; 8], MapSource>::new(IndexTargets::new(), |_, _, _| Ok(()))
.and_events(QueueTargets::new().with(&jobs.queue), |_, _, _, _| Ok(()));
let fixed_dir = tempfile::tempdir().unwrap();
assert!(matches!(
FixedMap::<[u8; 8], 8, ZeroHookAdapter<[u8; 8], MapSource, _>>::open_with_hook(
fixed_dir.path(),
FixedConfig::test(),
ZeroHookAdapter::new(hook),
),
Err(DbError::ChangeStreamRequiresManagedOpen)
));
}
#[test]
fn managed_zero_map_stream_sources_refuse_facade_and_atomic_writes_without_a_trace() {
let bitcask_dir = tempfile::tempdir().unwrap();
let bitcask_db = Db::open_test(bitcask_dir.path()).unwrap();
let bitcask_jobs = bitcask_db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let bitcask_source = bitcask_db
.open_zero_map::<MapSource, 8, _>(Config::test(), map_stream_hook(&bitcask_jobs), &[])
.unwrap();
let err = bitcask_source.put(&key(1), &MapSource(1)).unwrap_err();
assert!(matches!(err, DbError::ChangeStreamRequiresTx { .. }));
assert!(bitcask_source.get(&key(1)).is_none());
assert_eq!(bitcask_jobs.len(), 0);
let fixed_dir = tempfile::tempdir().unwrap();
let fixed_db = Db::open_test(fixed_dir.path()).unwrap();
let fixed_jobs = fixed_db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let fixed_source = fixed_db
.open_zero_map_fixed::<MapSource, 8, _>(
FixedConfig::test(),
map_stream_hook(&fixed_jobs),
&[],
)
.unwrap();
let err = match fixed_source.atomic(&key(1), |shard| shard.put(&key(1), &MapSource(1))) {
Ok(_) => panic!("fixed zero-map stream atomic write unexpectedly succeeded"),
Err(err) => err,
};
assert!(matches!(err, DbError::ChangeStreamRequiresTx { .. }));
assert!(fixed_source.get(&key(1)).is_none());
assert_eq!(fixed_jobs.len(), 0);
}
#[test]
fn fixed_zero_map_stream_commit_writes_var_queue_header_and_payload() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let source = db
.open_zero_map_fixed::<MapSource, 8, _>(
FixedConfig::test(),
map_stream_var_hook(&jobs),
&[],
)
.unwrap();
db.tx()
.put(&source, &key(1), &MapSource(7))
.unwrap()
.commit()
.unwrap();
assert_eq!(jobs.len(), 1, "the FixedStore header must be committed");
assert_eq!(
jobs.debug_payload_len(),
1,
"the Var payload must be committed"
);
let leased = jobs.lease(1, None).unwrap();
assert_eq!(leased.len(), 1);
assert_eq!(leased[0].payload(), &VarJob("map-job-7".into()));
}
#[test]
fn var_stream_preflight_drains_payload_cleanup_before_entering_the_commit_guard() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<VarJobs>(
QueueConfig::balanced()
.max_attempts(1)
.dead_retention(Some(std::time::Duration::from_millis(1)))
.build()
.unwrap(),
)
.unwrap();
jobs.push(VarJob("cleanup".into()), None).unwrap();
let leased = jobs.lease(1, None).unwrap().pop().unwrap();
jobs.debug_fail_payload_deletes(2);
jobs.nack(leased.token()).unwrap();
std::thread::sleep(std::time::Duration::from_millis(3));
assert!(db.sweep_expired_now().is_err());
assert_eq!(jobs.debug_payload_len(), 1, "failed cleanup must be queued");
let source = db
.open_typed_tree::<MapSource, RapiraCodec, _>(
Config::test(),
map_stream_var_hook(&jobs),
&[],
)
.unwrap();
db.tx()
.put(&source, &key(1), &MapSource(7))
.unwrap()
.commit()
.unwrap();
assert_eq!(
jobs.debug_payload_len(),
1,
"stale payload is cleaned before stream output"
);
let leased = jobs.lease(1, None).unwrap();
assert_eq!(leased.len(), 1);
assert_eq!(leased[0].payload(), &VarJob("map-job-7".into()));
}
#[test]
fn ttl_stream_source_is_rejected_before_publication() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let jobs = db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let hook = Projected::<[u8; 8], TtlSource>::new(IndexTargets::new(), |_, _, _| Ok(()))
.and_events(QueueTargets::new().with(&jobs), |_, _, _, _| Ok(()));
let err = match db.open_typed_tree::<TtlSource, RapiraCodec, _>(Config::test(), hook, &[]) {
Ok(_) => panic!("TTL stream source unexpectedly opened"),
Err(err) => err,
};
assert!(matches!(err, DbError::Client(_)));
assert!(
db.inspector()
.collections()
.unwrap()
.iter()
.all(|collection| collection.name != TtlSource::NAME)
);
}
armour_core::const_hasher_or!(member_id, "MEM", "dev-key-member_id", brand = "member");
type MemberId = armour_core::Fuid<member_id::Hasher>;
#[derive(
Clone,
Copy,
Debug,
PartialEq,
Eq,
Hash,
zerocopy::FromBytes,
zerocopy::IntoBytes,
zerocopy::Immutable,
zerocopy::KnownLayout,
)]
#[repr(C)]
pub struct HandleKey([u8; 16]);
armdb::impl_key_zerocopy!(
HandleKey,
armour_core::KeyScheme::Typed(&[armour_core::KeyType::Array(16)])
);
impl HandleKey {
fn new(s: &str) -> Self {
let mut b = [0u8; 16];
b[..s.len()].copy_from_slice(s.as_bytes());
Self(b)
}
}
armdb::unique_index!(Handles, key = HandleKey, name = "queue_stream_handles");
#[derive(Clone, Debug, PartialEq, GetType, armour_core::GetRefs, Rapira)]
struct Member {
handle: String,
weight: u64,
}
impl CollectionMeta for Member {
type SelfId = MemberId;
const NAME: &'static str = "queue_stream_members";
}
#[derive(
Clone,
Copy,
Debug,
Default,
PartialEq,
GetType,
armour_core::GetRefs,
zerocopy::FromBytes,
zerocopy::IntoBytes,
zerocopy::Immutable,
zerocopy::KnownLayout,
)]
#[repr(C)]
struct MemberMarker(u8);
impl CollectionMeta for MemberMarker {
type SelfId = MemberId;
const NAME: &'static str = "queue_stream_member_tags";
}
#[derive(Clone, Debug, PartialEq, GetType, armour_core::GetRefs, Rapira)]
struct MemberStats {
total_weight: u64,
}
impl CollectionMeta for MemberStats {
type SelfId = [u8; 8];
const NAME: &'static str = "queue_stream_member_stats";
}
type MemberStatsMap = armdb::typed_map!(MemberStats, RapiraCodec);
type MemberHook = Projected<MemberId, Member, NoHook, Events<MemberId, Member>>;
type MemberMap = armdb::typed_map!(Member, RapiraCodec, MemberHook);
struct FourHalves {
db: Arc<Db>,
members: Arc<MemberMap>,
index: Arc<armdb::zero_tree!(MemberMarker)>,
stats: Arc<MemberStatsMap>,
handles: Arc<armdb::UniqueIndex<Handles>>,
jobs: Arc<Queue<FixedJobs>>,
_dir: tempfile::TempDir,
}
const STATS_KEY: [u8; 8] = [0; 8];
fn open_four_halves() -> FourHalves {
let dir = tempfile::tempdir().unwrap();
let db = Arc::new(Db::open_test(dir.path()).unwrap());
let jobs = db
.open_queue::<FixedJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let index = db
.open_zero_tree::<MemberMarker, { std::mem::size_of::<MemberMarker>() }, _>(
Config::test(),
NoHook,
&[],
)
.unwrap();
let stats = db
.open_typed_map::<MemberStats, RapiraCodec, _>(Config::test(), NoHook, &[])
.unwrap();
stats
.put(&STATS_KEY, MemberStats { total_weight: 0 })
.unwrap();
let handles = db
.open_unique_index::<Handles>(FixedConfig::test())
.unwrap();
let mut reg = SchemaRegistry::new();
reg.register(&index, CollectionKind::SecondaryIndex);
reg.register(&stats, CollectionKind::Counters);
reg.register_external(handles.name(), CollectionKind::UniqueIndex);
reg.register_queue(&jobs);
let idx_for_projection = Arc::clone(&index);
let carrier = reg.secondary_index::<Member, _>(
IndexTargets::new().with_index(&index),
move |k: &MemberId, _v: &Member, sink| sink.row(&idx_for_projection, *k, MemberMarker(1)),
);
let stats_for_counter = Arc::clone(&stats);
let carrier = reg.counted_on(
carrier,
armdb::CounterTargets::new().with_counter(
&stats,
"total_weight",
armdb::OnMissingTarget::Error,
|v: &MemberStats| v.total_weight,
|cur: Option<&MemberStats>, n: u64| match cur {
Some(_) => armdb::Upsert::Put(MemberStats { total_weight: n }),
None => armdb::Upsert::Keep,
},
),
move |_k, v: &Member, sink| sink.add(&stats_for_counter, STATS_KEY, v.weight),
);
let handles_for_claim = Arc::clone(&handles);
let carrier = reg.unique_index_on(
carrier,
armdb::UniqueTargets::new().with(&handles),
move |_k, v: &Member, sink| sink.claim(&handles_for_claim, HandleKey::new(&v.handle)),
);
let jobs_for_stream = Arc::clone(&jobs);
let hook = reg.change_stream_on(
carrier,
QueueTargets::new().with(&jobs),
move |_k, _old, new: Option<&Member>, sink| {
if let Some(new) = new {
sink.emit(&jobs_for_stream, FixedJob(new.weight))?;
}
Ok(())
},
);
assert_eq!(hook.targets().len(), 1, "the index half is non-empty");
assert!(
hook.counter_targets().is_some(),
"the counter half survived"
);
assert!(hook.claim_targets().is_some(), "the claim half survived");
assert!(hook.stream_targets().is_some(), "the stream half survived");
let members = db
.open_typed_map::<Member, RapiraCodec, _>(Config::test(), hook, &[])
.unwrap();
FourHalves {
db,
members,
index,
stats,
handles,
jobs,
_dir: dir,
}
}
#[test]
fn a_commit_through_all_four_carrier_halves_writes_index_counter_claim_and_queue_row() {
let f = open_four_halves();
let id = MemberId::with_shard(1);
f.db.tx()
.put(
&f.members,
&id,
&Member {
handle: "rail".into(),
weight: 7,
},
)
.unwrap()
.commit()
.unwrap();
assert!(f.index.get(&id).is_some(), "the index half wrote its row");
assert_eq!(
f.stats.get(&STATS_KEY).unwrap().total_weight,
7,
"the counter half applied the contribution"
);
assert!(
f.handles.contains(&HandleKey::new("rail")),
"the claim half wrote its owner row"
);
let leased = f.jobs.lease(8, None).unwrap();
assert_eq!(leased.len(), 1, "the stream half emitted exactly one row");
assert_eq!(leased[0].payload(), &FixedJob(7));
}
#[test]
fn a_unique_violation_after_stream_lowering_leaves_no_queue_row_behind() {
let f = open_four_halves();
let first = MemberId::with_shard(1);
f.db.tx()
.put(
&f.members,
&first,
&Member {
handle: "rail".into(),
weight: 7,
},
)
.unwrap()
.commit()
.unwrap();
let leased = f.jobs.lease(8, None).unwrap();
assert_eq!(leased.len(), 1);
f.jobs.ack(leased[0].token()).unwrap();
let weight_before = f.stats.get(&STATS_KEY).unwrap().total_weight;
let second = MemberId::with_shard(2);
let refused = f.db.tx().put(
&f.members,
&second,
&Member {
handle: "rail".into(),
weight: 11,
},
);
let outcome = refused.and_then(armdb::Tx::commit);
assert!(
matches!(outcome, Err(DbError::UniqueViolation { .. })),
"expected a unique violation, got {outcome:?}"
);
assert!(f.members.get(&second).is_none(), "no source row");
assert!(f.index.get(&second).is_none(), "no index row");
assert_eq!(
f.stats.get(&STATS_KEY).unwrap().total_weight,
weight_before,
"no counter delta"
);
assert_eq!(f.jobs.len(), 0, "no queue row");
assert!(f.jobs.lease(8, None).unwrap().is_empty());
}