use tokio::sync::oneshot;
use super::LIFECYCLE_DISPATCH_ADMISSION_WAIT;
pub(super) async fn wait_without_unbounded_caller_delay(completion: oneshot::Receiver<()>) {
let deadline = tokio::time::Instant::now() + LIFECYCLE_DISPATCH_ADMISSION_WAIT;
let _ = tokio::time::timeout_at(deadline, completion).await;
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use rmux_core::LifecycleEvent;
use rmux_proto::{
HookLifecycle, HookName, NewSessionRequest, Request, Response, ScopeSelector, SessionName,
SetHookMutationRequest, TerminalSize,
};
use tokio::sync::oneshot;
use super::super::prepare_lifecycle_event;
use crate::handler::RequestHandler;
fn session_name(value: &str) -> SessionName {
SessionName::new(value).expect("valid session name")
}
async fn create_session(handler: &RequestHandler, name: &str) -> SessionName {
let session = session_name(name);
let response = handler
.handle(Request::NewSession(NewSessionRequest {
session_name: session.clone(),
detached: true,
size: Some(TerminalSize { cols: 80, rows: 24 }),
environment: None,
}))
.await;
assert!(matches!(response, Response::NewSession(_)), "{response:?}");
session
}
async fn set_focus_hook(handler: &RequestHandler, command: &str) {
let response = handler
.handle(Request::SetHookMutation(SetHookMutationRequest {
scope: ScopeSelector::Global,
hook: HookName::ClientFocusIn,
command: Some(command.to_owned()),
lifecycle: HookLifecycle::Persistent,
append: false,
unset: false,
run_immediately: false,
index: None,
}))
.await;
assert!(matches!(response, Response::SetHook(_)), "{response:?}");
}
async fn prepared_focus_event(
handler: &RequestHandler,
session_name: SessionName,
) -> super::super::QueuedLifecycleEvent {
let mut state = handler.state.lock().await;
prepare_lifecycle_event(
&mut state,
&LifecycleEvent::ClientFocusIn {
session_name,
client_name: Some("ordered-lifecycle-test".to_owned()),
},
)
}
async fn spawn_lifecycle_consumer(
handler: &RequestHandler,
) -> (oneshot::Sender<()>, tokio::task::JoinHandle<()>) {
let events = handler
.take_lifecycle_dispatch_receiver()
.expect("test activates the lifecycle dispatch receiver once");
let (shutdown, shutdown_rx) = oneshot::channel();
let consumer_handler = handler.clone();
let consumer = tokio::spawn(async move {
consumer_handler
.consume_lifecycle_hooks(events, shutdown_rx)
.await;
});
(shutdown, consumer)
}
async fn stop_lifecycle_consumer(
handler: &RequestHandler,
shutdown: oneshot::Sender<()>,
consumer: tokio::task::JoinHandle<()>,
) {
let _ = shutdown.send(());
handler.shutdown_wait_for();
tokio::time::timeout(Duration::from_secs(2), consumer)
.await
.expect("lifecycle consumer stops after draining shutdown")
.expect("lifecycle consumer task joins");
}
async fn wait_for_hook_block(handler: &RequestHandler, channel: &str) {
tokio::time::timeout(Duration::from_secs(2), async {
loop {
if handler.wait_for_counts(channel) == (1, 0, false) {
return;
}
tokio::task::yield_now().await;
}
})
.await
.expect("hook reaches its deterministic wait-for seam");
}
#[tokio::test]
async fn ordered_lifecycle_wait_observes_a_fast_mutating_hook_before_returning() {
let handler = RequestHandler::new();
let session = create_session(&handler, "ordered-fast-hook").await;
set_focus_hook(&handler, "set-buffer -b ordered-lifecycle-fast completed").await;
let (shutdown, consumer) = spawn_lifecycle_consumer(&handler).await;
let event = prepared_focus_event(&handler, session).await;
handler.emit_prepared_and_wait(event).await;
let state = handler.state.lock().await;
let (_, content) = state
.buffers
.show(Some("ordered-lifecycle-fast"))
.expect("fast hook mutation is visible before the caller resumes");
assert_eq!(content, b"completed");
drop(state);
stop_lifecycle_consumer(&handler, shutdown, consumer).await;
}
#[tokio::test]
async fn ordered_lifecycle_wait_releases_the_caller_but_drains_the_blocked_hook() {
let handler = RequestHandler::new();
let session = create_session(&handler, "ordered-blocked-hook").await;
set_focus_hook(&handler, "wait-for ordered-lifecycle-block").await;
let (shutdown, consumer) = spawn_lifecycle_consumer(&handler).await;
let event = prepared_focus_event(&handler, session).await;
tokio::time::timeout(
Duration::from_secs(2),
handler.emit_prepared_and_wait(event),
)
.await
.expect("a blocked hook must not keep the latency-sensitive caller indefinitely");
wait_for_hook_block(&handler, "ordered-lifecycle-block").await;
stop_lifecycle_consumer(&handler, shutdown, consumer).await;
assert_eq!(
handler.wait_for_counts("ordered-lifecycle-block"),
(0, 0, false),
"shutdown drains the queued hook and its completion"
);
}
#[tokio::test]
async fn ordered_dispatch_accepts_before_bounding_the_callers_completion_wait() {
let handler = RequestHandler::new();
let session = create_session(&handler, "ordered-saturated-hook").await;
set_focus_hook(&handler, "wait-for ordered-outbox-block").await;
let (shutdown, consumer) = spawn_lifecycle_consumer(&handler).await;
let event = prepared_focus_event(&handler, session).await;
let started = tokio::time::Instant::now();
handler.emit_prepared_and_wait(event).await;
let elapsed = started.elapsed();
assert!(
elapsed >= super::LIFECYCLE_DISPATCH_ADMISSION_WAIT && elapsed < Duration::from_secs(2),
"only hook completion waiting is bounded: {elapsed:?}"
);
wait_for_hook_block(&handler, "ordered-outbox-block").await;
stop_lifecycle_consumer(&handler, shutdown, consumer).await;
}
#[tokio::test]
async fn lifecycle_hook_backpressure_releases_the_committed_post_commit_turn() {
let handler = RequestHandler::new();
let session = create_session(&handler, "post-commit-hook-cycle").await;
set_focus_hook(&handler, "wait-for post-commit-hook-cycle").await;
let (shutdown, consumer) = spawn_lifecycle_consumer(&handler).await;
let event = prepared_focus_event(&handler, session).await;
let first = handler
.pane_mode_post_commit
.acquire_capacity()
.await
.expect("first post-commit capacity")
.sequence();
let second = handler
.pane_mode_post_commit
.acquire_capacity()
.await
.expect("second post-commit capacity")
.sequence();
let event_handler = handler.clone();
let mut first_task = tokio::spawn(first.run(async move {
event_handler.emit_prepared_and_wait(event).await;
}));
wait_for_hook_block(&handler, "post-commit-hook-cycle").await;
tokio::time::timeout(Duration::from_millis(100), second.run(async {}))
.await
.expect("hook backpressure must not retain the committed post-commit turn");
assert!(
!first_task.is_finished(),
"the lifecycle hook remains blocked independently of the sequencer"
);
stop_lifecycle_consumer(&handler, shutdown, consumer).await;
tokio::time::timeout(Duration::from_secs(1), &mut first_task)
.await
.expect("ordered publication finishes after hook shutdown")
.expect("first post-commit task joins");
}
}