use std::time::Duration;
use armdb::armour::Db;
use armdb::{CollectionMeta, DbError, FixedConfig, NoHook, QueueConfig, TxConfig, queue};
use armour_core::GetType;
use rapira::Rapira;
#[derive(Clone, Debug, PartialEq, Eq, GetType, Rapira)]
pub struct VarJob {
body: String,
}
queue!(VarJobs, payload = VarJob, name = "queue_var_jobs", var);
queue!(
VarJobsSchemaDrift,
payload = VarJob,
name = "queue_var_schema_drift",
var
);
#[derive(Clone, Debug, PartialEq, Eq, GetType, Rapira)]
pub struct ChangedVarJob {
body: String,
priority: i32,
}
queue!(
VarJobsSchemaDriftChanged,
payload = ChangedVarJob,
name = "queue_var_schema_drift",
var
);
queue!(
PayloadNameCollision,
payload = VarJob,
name = "queue_var_collision",
var
);
queue!(
PayloadOccupier,
payload = u64,
name = "queue_var_collision.payload"
);
#[test]
fn open_queue_rejects_unbuilt_invalid_var_policies() {
let assert_invalid = |config| {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
assert!(matches!(
db.open_queue::<VarJobs>(config),
Err(DbError::Config(_))
));
};
assert_invalid(QueueConfig::balanced().max_attempts(0));
assert_invalid(QueueConfig::balanced().backoff_secs(&[]));
assert_invalid(QueueConfig::balanced().lease_timeout(Duration::ZERO));
assert_invalid(QueueConfig::balanced().wakeup_recheck(Duration::ZERO));
}
struct CompatibleStagedHeader;
impl armdb::QueueSpec for CompatibleStagedHeader {
type Payload = VarJob;
type Mode = armdb::VarQueueMode;
const NAME: &'static str = "queue_var_collision";
const PAYLOAD_NAME: &'static str = "queue_var_collision.compatible_payload";
}
#[derive(
Clone,
Copy,
Debug,
PartialEq,
Eq,
GetType,
zerocopy::FromBytes,
zerocopy::IntoBytes,
zerocopy::Immutable,
zerocopy::KnownLayout,
)]
#[repr(transparent)]
struct Effect(u64);
impl CollectionMeta for Effect {
type SelfId = [u8; 8];
const NAME: &'static str = "queue_var_effect";
}
#[test]
fn lease_does_not_rewrite_the_var_payload_and_ack_removes_both_rows() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let q = db
.open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
q.push(
VarJob {
body: "once".into(),
},
None,
)
.unwrap();
let row = q
.lease(1, Some(Duration::from_secs(10)))
.unwrap()
.pop()
.unwrap();
assert_eq!(row.payload().body, "once");
assert_eq!(q.debug_payload_write_count(), 1);
q.ack(row.token()).unwrap();
assert_eq!(q.debug_payload_len(), 0);
assert_eq!(q.len(), 0);
}
#[test]
fn reopen_collects_an_orphan_var_payload() {
let dir = tempfile::tempdir().unwrap();
{
let db = Db::open_test(dir.path()).unwrap();
let q = db
.open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
q.debug_insert_orphan_payload(
42,
VarJob {
body: "orphan".into(),
},
)
.unwrap();
assert_eq!(q.debug_payload_len(), 1);
}
let db = Db::open_test(dir.path()).unwrap();
let q = db
.open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
assert_eq!(q.debug_payload_len(), 0);
}
#[test]
fn nack_retains_the_var_payload_across_retry_and_dead_letter_moves() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let retry = db
.open_queue::<VarJobs>(
QueueConfig::balanced()
.max_attempts(2)
.backoff_secs(&[0])
.build()
.unwrap(),
)
.unwrap();
retry
.push(
VarJob {
body: "retry".into(),
},
None,
)
.unwrap();
let first = retry.lease(1, None).unwrap().pop().unwrap();
retry.nack(first.token()).unwrap();
let second = retry.lease(1, None).unwrap().pop().unwrap();
assert_eq!(second.payload().body, "retry");
retry.nack(second.token()).unwrap();
assert_eq!(retry.len(), 0);
assert_eq!(retry.dead_len(), 1);
assert_eq!(retry.debug_payload_len(), 1);
}
#[test]
fn nack_invalidates_the_token_before_the_var_retry_is_leased_again() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let queue = db
.open_queue::<VarJobs>(
QueueConfig::balanced()
.max_attempts(2)
.backoff_secs(&[0])
.build()
.unwrap(),
)
.unwrap();
queue
.push(
VarJob {
body: "retry".into(),
},
None,
)
.unwrap();
let leased = queue.lease(1, None).unwrap().pop().unwrap();
queue.nack(leased.token()).unwrap();
assert!(matches!(
queue.ack(leased.token()),
Err(DbError::LeaseExpired)
));
}
#[test]
fn transaction_push_and_ack_pair_var_payload_with_the_header() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let q = db
.open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
db.tx()
.push(
&q,
VarJob {
body: "atomic".into(),
},
None,
)
.unwrap()
.commit()
.unwrap();
assert_eq!(q.len(), 1);
assert_eq!(q.debug_payload_len(), 1);
let row = q.lease(1, None).unwrap().pop().unwrap();
db.tx().ack(&q, row.token()).unwrap().commit().unwrap();
assert_eq!(q.len(), 0);
assert_eq!(q.debug_payload_len(), 0);
}
#[test]
fn transactional_var_pushes_respect_max_depth_before_commit() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let q = db
.open_queue::<VarJobs>(QueueConfig::balanced().max_depth(Some(1)).build().unwrap())
.unwrap();
let result = db
.tx()
.push(
&q,
VarJob {
body: "first".into(),
},
None,
)
.unwrap()
.push(
&q,
VarJob {
body: "second".into(),
},
None,
);
assert!(matches!(result, Err(DbError::QueueFull)));
assert_eq!(q.len(), 0);
assert_eq!(q.debug_payload_len(), 0);
}
#[test]
fn payload_schema_drift_is_rejected_on_reopen() {
let dir = tempfile::tempdir().unwrap();
Db::open_test(dir.path())
.unwrap()
.open_queue::<VarJobsSchemaDrift>(QueueConfig::balanced().build().unwrap())
.unwrap();
let result = Db::open_test(dir.path())
.unwrap()
.open_queue::<VarJobsSchemaDriftChanged>(QueueConfig::balanced().build().unwrap());
let Err(err) = result else {
panic!("payload type drift must fail managed reopen");
};
assert!(matches!(err, DbError::SchemaMismatch { .. }), "{err}");
}
#[test]
fn payload_name_collision_does_not_publish_the_staged_header() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
db.open_queue::<PayloadOccupier>(QueueConfig::balanced().build().unwrap())
.unwrap();
assert!(
db.open_queue::<PayloadNameCollision>(QueueConfig::balanced().build().unwrap())
.is_err()
);
db.open_queue::<CompatibleStagedHeader>(QueueConfig::balanced().build().unwrap())
.expect("failed composite Var open must release its unpublished header name");
}
#[test]
fn failed_transaction_with_var_push_keeps_both_rows_and_count() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let q = db
.open_queue::<VarJobs>(QueueConfig::balanced().build().unwrap())
.unwrap();
let effects = db
.open_zero_tree_fixed::<Effect, 8, _>(FixedConfig::test(), NoHook, &[])
.unwrap();
let result = db
.tx()
.push(
&q,
VarJob {
body: "atomic".into(),
},
None,
)
.unwrap()
.guard(&effects, &1_u64.to_be_bytes(), |_| {
Err(DbError::Client("abort var queue tx"))
})
.unwrap()
.commit();
assert!(matches!(result, Err(DbError::Client("abort var queue tx"))));
assert_eq!(q.len(), 0);
assert_eq!(q.debug_payload_len(), 0);
assert!(q.lease(1, None).unwrap().is_empty());
}
#[test]
fn var_ack_and_nack_reject_max_ops_before_any_mutation() {
for nack in [false, true] {
let dir = tempfile::tempdir().unwrap();
let db = Db::builder(dir.path())
.for_tests()
.tx_config(TxConfig {
max_ops: 1,
..TxConfig::default()
})
.open()
.unwrap();
let q = db
.open_queue::<VarJobs>(QueueConfig::balanced().backoff_secs(&[0]).build().unwrap())
.unwrap();
q.push(
VarJob {
body: "limit".into(),
},
None,
)
.unwrap();
let leased = q.lease(1, None).unwrap().pop().unwrap();
let result = if nack {
db.tx().nack(&q, leased.token())
} else {
db.tx().ack(&q, leased.token())
};
assert!(matches!(
result,
Err(DbError::Client("transaction exceeds max_ops"))
));
assert_eq!(q.len(), 1);
assert_eq!(q.debug_payload_len(), 1);
}
}