use super::*;
#[path = "drainer/fixture.rs"]
mod fixture;
use fixture::DrainerCase;
fn prove_drainer_last_owner(pending: usize) -> Result<(), Box<dyn Error>> {
use liminal::durability::bridge::block_on;
let mut case = DrainerCase::new(false)?;
let starting_head = case
.fixture()?
.services
.self_stream_head("liminal.events")?;
let before = case.counts.snapshot();
let first = case.start_event()?;
let mut sequences = Vec::with_capacity(pending);
for connection in 2..u64::try_from(pending)? + 2 {
sequences.push(
case.fixture()?
.sink
.enqueue_event_for_test(ServerEvent::ConnectionOpened { connection })?,
);
}
assert_eq!(case.queue.measurements()?.0, pending);
case.release_services()?;
let released = std::time::Instant::now();
case.pause.release()?;
let log = case.finish()?;
let elapsed_ms = released.elapsed().as_millis();
let counts = case.counts.snapshot().since(&before);
let failure = case.queue.failure();
let mut failed = 0;
for sequence in sequences {
let error = case
.queue
.await_append(sequence)
.err()
.ok_or("last-owner backlog silently passed without storage")?;
assert!(
error
.to_string()
.contains("receipt batch services owner is gone"),
"{error}"
);
failed += 1;
}
eprintln!(
"DRAINER_LAST_OWNER pending={pending} failed={failed} elapsed_ms={elapsed_ms} counts={counts:?} failure={failure:?}"
);
assert!(
elapsed_ms < 2000,
"drainer last-owner release exceeded the 2s ceiling"
);
assert!(
log.contains("self-stream drainer drops its final services owner and is stopping"),
"{log}"
);
assert_eq!(counts.append_rows, 1);
assert_eq!(usize::try_from(counts.append_rows - 1)? + failed, pending);
if pending == 0 {
assert!(failure.is_none(), "last-owner drop panicked: {failure:?}");
case.queue.await_append(first)?;
}
let refusal = case
.queue
.enqueue(crate::server::self_streams::queue::QueuedRecord {
sequence: 0,
channel: "liminal.events",
payload: crate::server::self_streams::queue::RecordPayload::Single(b"{}".to_vec()),
connection_opened_count: 0,
})
.err()
.ok_or("last-owner drop did not stop its queue")?;
assert!(refusal.to_string().contains(if pending == 0 {
"record queue is closed"
} else {
"receipt batch services owner is gone"
}));
let row = block_on(case.store.read_at("liminal.events:0", starting_head))??;
let bytes = row.ok_or("last-owner committed row absent")?.payload.len();
assert_eq!(
counts,
StoreCounts {
apply_routed_batch_calls: 1,
apply_routed_batch_items: 1,
append_calls: 1,
append_rows: 1,
append_bytes: u64::try_from(bytes)?,
..StoreCounts::default()
}
);
assert!(case.root.as_ref().is_some_and(|root| root.path().exists()));
Ok(())
}
#[test]
fn drainer_last_owner_returns_and_stops_its_queue() -> Result<(), Box<dyn Error>> {
prove_drainer_last_owner(0)
}
#[test]
fn drainer_last_owner_accounts_for_the_full_backlog() -> Result<(), Box<dyn Error>> {
prove_drainer_last_owner(crate::server::self_streams::queue::CAPACITY)
}
#[test]
fn drainer_panic_names_the_next_publish_and_reaps_the_watch() -> Result<(), Box<dyn Error>> {
let mut case = DrainerCase::new(true)?;
let before = case.counts.snapshot();
case.start_event()?;
case.pause.release()?;
case.finish()?;
let error = case
.fixture()?
.sink
.enqueue_event_for_test(ServerEvent::ConnectionOpened { connection: 2 })
.err()
.ok_or("panicked drainer silently admitted the next publish")?;
assert!(
error.to_string().contains("record drainer panicked"),
"{error}"
);
let joined = case
.fixture()?
.sink
.join_drainer_for_test()
.err()
.ok_or("panicked drainer join was reported successful")?;
assert!(
joined.to_string().contains("record drainer join panicked"),
"{joined}"
);
let counts = case.counts.snapshot().since(&before);
eprintln!("DRAINER_PANIC counts={counts:?} publish_error={error} join_error={joined}");
assert_eq!(
counts,
StoreCounts {
append_calls: 1,
..StoreCounts::default()
}
);
Ok(())
}
#[test]
fn drainer_panicked_join_keeps_a_named_failure_after_external_drop() -> Result<(), Box<dyn Error>> {
let mut case = DrainerCase::new(true)?;
let before = case.counts.snapshot();
case.start_event()?;
case.pause.release()?;
case.finish()?;
case.release_services()?;
let error = case
.queue
.failure()
.ok_or("panicked join was silently discarded")?;
assert!(
error.to_string().contains("record drainer panicked")
|| error.to_string().contains("record drainer join panicked"),
"{error}"
);
let counts = case.counts.snapshot().since(&before);
eprintln!("DRAINER_PANICKED_DROP counts={counts:?} error={error}");
assert_eq!(
counts,
StoreCounts {
apply_routed_batch_calls: 1,
apply_routed_batch_items: 1,
append_calls: 1,
..StoreCounts::default()
}
);
Ok(())
}
#[test]
fn drainer_abandoned_unfinished_drop_returns_and_names_the_unconfirmed_thread()
-> Result<(), Box<dyn Error>> {
let mut case = DrainerCase::new(false)?;
let fixture = case.fixture()?;
let shutdown = crate::server::shutdown::ShutdownHandle::new();
let sink = Arc::new(crate::server::self_streams::sink::SelfStreams::new(
ServerMetadata::new("local".to_owned(), fixture.incarnation)?,
&fixture.services,
Arc::clone(&fixture.wake),
shutdown.clone(),
crate::config::types::SelfStreamsConfig::default(),
));
let queue = sink.queue_for_test();
let mut publisher = crate::server::self_streams::owner::PublisherRuntime::start(
Arc::clone(&fixture.wake),
Arc::clone(&sink),
shutdown,
)?;
let before = case.counts.snapshot();
publisher.request_stop_for_test();
case.pause.entered()?;
let bound = publisher
.finish_with_bound(std::time::Duration::ZERO)
.err()
.ok_or("gated owned drain unexpectedly completed")?;
assert!(
bound
.to_string()
.contains("self-stream owned durable drain bound failed"),
"{bound}"
);
assert!(publisher.join_completed_for_test().is_err());
assert!(queue.was_abandoned());
let log = crate::test_log::CapturedLog::default();
let began = std::time::Instant::now();
log.capture(|| drop(sink));
let drop_ms = began.elapsed().as_millis();
assert!(
drop_ms < 2000,
"unfinished abandoned drainer drop exceeded the 2s ceiling"
);
assert!(log.contents().contains(
"self-stream drainer completion remains unconfirmed after owned drain abandonment"
));
let pending = case.counts.snapshot().since(&before);
assert_eq!(
pending,
StoreCounts {
append_calls: 1,
..StoreCounts::default()
}
);
case.pause.release()?;
case.finish()?;
let complete = case.counts.snapshot().since(&before);
eprintln!(
"DRAINER_ABANDONED drop_ms={drop_ms} pending={pending:?} complete={complete:?} error={bound}"
);
assert_eq!(complete.append_rows, 1);
Ok(())
}