use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::time::Duration;
use modrun::{Error, Lifecycle, Modrun, Shutdowner, hook, task, task_with};
use std::future::pending;
use std::net::SocketAddr;
use std::time::Instant;
use tokio::net::TcpListener;
use tokio::task::yield_now;
use tokio::time::sleep;
use tokio::time::timeout;
#[tokio::test]
async fn task_runs_until_stop() {
let hits = Arc::new(AtomicUsize::new(0));
let n = Arc::clone(&hits);
fn boot(lc: Lifecycle, hits: Arc<AtomicUsize>) -> modrun::Result<()> {
lc.append(task("worker", move |stopped| async move {
hits.fetch_add(1, Ordering::SeqCst);
stopped.await;
hits.fetch_add(10, Ordering::SeqCst);
Ok(())
}))
}
Modrun::builder()
.no_banner()
.supply(n)
.invoke(boot)
.start()
.await
.unwrap()
.stop()
.await
.unwrap();
assert_eq!(hits.load(Ordering::SeqCst), 11);
}
#[tokio::test]
async fn task_error_surfaces_on_stop() {
fn boot(lc: Lifecycle) -> modrun::Result<()> {
lc.append(task("boom", |_stopped| async {
Err(Error::hook("task-boom"))
}))
}
let err = Modrun::builder()
.no_banner()
.invoke(boot)
.start()
.await
.unwrap()
.stop()
.await
.unwrap_err();
assert!(err.to_string().contains("task-boom"), "{err}");
}
#[tokio::test]
async fn task_abort_on_stop_timeout() {
let dropped = Arc::new(AtomicBool::new(false));
let flag = Arc::clone(&dropped);
struct Guard(Arc<AtomicBool>);
impl Drop for Guard {
fn drop(&mut self) {
self.0.store(true, Ordering::SeqCst);
}
}
fn boot(lc: Lifecycle, dropped: Arc<AtomicBool>) -> modrun::Result<()> {
lc.append(task("hang", move |_stopped| async move {
let _guard = Guard(dropped);
pending::<()>().await;
Ok(())
}))
}
let err = Modrun::builder()
.no_banner()
.stop_timeout(Duration::from_millis(30))
.supply(flag)
.invoke(boot)
.start()
.await
.unwrap()
.stop()
.await
.unwrap_err();
assert!(err.to_string().contains("timed out"), "{err}");
timeout(Duration::from_secs(1), async {
while !dropped.load(Ordering::SeqCst) {
yield_now().await;
}
})
.await
.expect("aborted task should drop");
}
#[tokio::test]
async fn task_error_unblocks_run_without_manual_shutdown() {
fn boot(lc: Lifecycle) -> modrun::Result<()> {
lc.append(task("die", |_stopped| async { Err(Error::hook("died")) }))
}
let err = Modrun::builder()
.no_banner()
.invoke(boot)
.run()
.await
.unwrap_err();
let msg = err.to_string();
assert!(msg.contains("died"), "{err}");
assert!(msg.contains("die"), "{err}");
}
#[tokio::test]
async fn task_panic_unblocks_run() {
fn boot(lc: Lifecycle) -> modrun::Result<()> {
lc.append(task("worker", |_stopped| async {
panic!("task-boom");
}))
}
let err = timeout(
Duration::from_secs(2),
Modrun::builder().no_banner().invoke(boot).run(),
)
.await
.expect("run should unblock after task panic")
.unwrap_err();
let msg = err.to_string();
assert!(msg.contains("worker"), "{msg}");
assert!(
msg.contains("panicked") || msg.contains("task-boom"),
"{msg}"
);
}
#[tokio::test]
async fn task_success_does_not_request_shutdown() {
fn boot(
lc: Lifecycle,
shutdown: Shutdowner,
slot: Arc<Mutex<Option<Shutdowner>>>,
) -> modrun::Result<()> {
*slot.lock().unwrap() = Some(shutdown.clone());
lc.append(task("quick", |_stopped| async { Ok(()) }))
}
let slot = Arc::new(Mutex::new(None::<Shutdowner>));
let app = Modrun::builder()
.no_banner()
.supply(Arc::clone(&slot))
.invoke(boot)
.start()
.await
.unwrap();
sleep(Duration::from_millis(50)).await;
let requested = slot
.lock()
.unwrap()
.as_ref()
.expect("shutdown captured")
.is_requested();
assert!(!requested, "Ok(()) must not shut the app down");
app.stop().await.unwrap();
}
#[tokio::test]
async fn task_with_run_error_unblocks_run() {
fn boot(lc: Lifecycle) -> modrun::Result<()> {
lc.append(task_with(
"serve",
|| async { Ok(()) },
|(), _stopped| async { Err(Error::hook("serve-boom")) },
))
}
let err = timeout(
Duration::from_secs(2),
Modrun::builder().no_banner().invoke(boot).run(),
)
.await
.expect("run should unblock after task_with run error")
.unwrap_err();
let msg = err.to_string();
assert!(msg.contains("serve-boom"), "{msg}");
assert!(msg.contains("serve"), "{msg}");
}
#[tokio::test]
async fn task_failure_during_later_start_is_not_ok() {
fn boot(lc: Lifecycle) -> modrun::Result<()> {
lc.append(task("die", |_stopped| async { Err(Error::hook("died")) }))?;
lc.append(hook().on_start(|| async {
sleep(Duration::from_secs(30)).await;
Ok(())
}))
}
let started = Instant::now();
let err = timeout(
Duration::from_secs(3),
Modrun::builder()
.no_banner()
.start_timeout(Duration::from_secs(5))
.invoke(boot)
.run(),
)
.await
.expect("run should unblock after task failure during start")
.unwrap_err();
assert!(
started.elapsed() < Duration::from_secs(2),
"should not wait for the later start hook"
);
let msg = err.to_string();
assert!(msg.contains("died"), "{msg}");
assert!(
!msg.contains("cleanup failed"),
"join error should stay the primary error: {msg}"
);
}
#[tokio::test]
async fn task_with_prepare_error_fails_start() {
fn boot(lc: Lifecycle) -> modrun::Result<()> {
lc.append(task_with(
"listen",
|| async { Err(Error::hook("addr in use")) },
|(), _stopped| async { Ok(()) },
))
}
let err = Modrun::builder()
.no_banner()
.invoke(boot)
.start()
.await
.unwrap_err();
assert!(err.to_string().contains("addr in use"), "{err}");
}
#[tokio::test]
async fn task_with_bind_conflict_fails_start() {
let held = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = held.local_addr().unwrap();
fn boot(lc: Lifecycle, addr: SocketAddr) -> modrun::Result<()> {
lc.append(task_with(
"http.serve",
move || async move {
TcpListener::bind(addr)
.await
.map(|_| ())
.map_err(|e| Error::io(format!("bind {addr}"), e))
},
|(), stopped| async move {
stopped.await;
Ok(())
},
))
}
let err = Modrun::builder()
.no_banner()
.supply(addr)
.invoke(boot)
.start()
.await
.unwrap_err();
let _ = held;
let msg = err.to_string();
assert!(msg.contains("bind"), "{msg}");
assert!(
msg.contains("Address already in use")
|| msg.contains("addr in use")
|| msg.contains("os error"),
"{msg}"
);
}
#[tokio::test]
async fn task_with_prepare_value_is_passed_to_run() {
let hits = Arc::new(AtomicUsize::new(0));
let n = Arc::clone(&hits);
fn boot(lc: Lifecycle, hits: Arc<AtomicUsize>) -> modrun::Result<()> {
let prepared = Arc::clone(&hits);
let running = Arc::clone(&hits);
lc.append(task_with(
"worker",
move || {
let prepared = Arc::clone(&prepared);
async move {
prepared.fetch_add(1, Ordering::SeqCst);
Ok(7u8)
}
},
move |value, stopped| {
let running = Arc::clone(&running);
async move {
running.fetch_add(usize::from(value), Ordering::SeqCst);
stopped.await;
Ok(())
}
},
))
}
Modrun::builder()
.no_banner()
.supply(n)
.invoke(boot)
.start()
.await
.unwrap()
.stop()
.await
.unwrap();
assert_eq!(hits.load(Ordering::SeqCst), 8);
}
#[tokio::test]
async fn task_panic_error_includes_name() {
fn boot(lc: Lifecycle) -> modrun::Result<()> {
lc.append(task("worker", |_stopped| async {
panic!("task-boom");
}))
}
let err = Modrun::builder()
.no_banner()
.invoke(boot)
.start()
.await
.unwrap()
.stop()
.await
.unwrap_err();
let msg = err.to_string();
assert!(msg.contains("worker"), "{msg}");
assert!(
msg.contains("panicked") || msg.contains("task-boom"),
"{msg}"
);
}