Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion docs/fault-tolerance/graceful-shutdown.md
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,11 @@ The `graceful_shutdown()` function:
5. Waits for in-flight requests (based on `graceful_shutdown` per endpoint)
6. Returns to allow cleanup to proceed

The aggregate wait in `runtime.shutdown()` is bounded by
`DYN_RUNTIME_GRACEFUL_SHUTDOWN_TIMEOUT_SECS`, which defaults to 900 seconds
(15 minutes). If endpoint draining exceeds this timeout, Dynamo logs the
remaining graceful endpoint count and proceeds with runtime teardown.

## Endpoint Draining

After the grace period, `runtime.shutdown()` invalidates endpoints so no new requests are accepted. The behavior for in-flight requests depends on the `graceful_shutdown` parameter when serving the endpoint.
Expand All @@ -64,7 +69,7 @@ generate_endpoint.serve_endpoint(

| `graceful_shutdown` | Behavior |
|---------------------|----------|
| `True` | Wait for all in-flight requests to complete before returning |
| `True` | Wait for all in-flight requests to complete, bounded by `DYN_RUNTIME_GRACEFUL_SHUTDOWN_TIMEOUT_SECS` |
| `False` | Return immediately without waiting for requests |

### Component-Specific Behavior
Expand Down
6 changes: 6 additions & 0 deletions lib/runtime/src/config/environment_names.rs
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,10 @@ pub mod runtime {
/// Maximum number of blocking threads for Tokio runtime
pub const DYN_RUNTIME_MAX_BLOCKING_THREADS: &str = "DYN_RUNTIME_MAX_BLOCKING_THREADS";

/// Maximum time to wait for graceful endpoint drain during runtime shutdown.
pub const DYN_RUNTIME_GRACEFUL_SHUTDOWN_TIMEOUT_SECS: &str =
"DYN_RUNTIME_GRACEFUL_SHUTDOWN_TIMEOUT_SECS";

/// Enable Tokio task poll-time histogram (calls enable_metrics_poll_time_histogram on builder).
/// Set to "1", "true", or "yes" to enable. Adds ~2× overhead of Instant::now() per task poll.
pub const DYN_ENABLE_POLL_HISTOGRAM: &str = "DYN_ENABLE_POLL_HISTOGRAM";
Expand Down Expand Up @@ -654,6 +658,7 @@ mod tests {
// Runtime
runtime::DYN_RUNTIME_NUM_WORKER_THREADS,
runtime::DYN_RUNTIME_MAX_BLOCKING_THREADS,
runtime::DYN_RUNTIME_GRACEFUL_SHUTDOWN_TIMEOUT_SECS,
Comment thread
michaelfeil marked this conversation as resolved.
runtime::system::DYN_SYSTEM_ENABLED,
runtime::system::DYN_SYSTEM_HOST,
runtime::system::DYN_SYSTEM_PORT,
Expand Down Expand Up @@ -791,6 +796,7 @@ mod tests {
fn test_naming_conventions() {
// Dynamo-specific vars should start with DYN_
assert!(runtime::DYN_RUNTIME_NUM_WORKER_THREADS.starts_with("DYN_"));
assert!(runtime::DYN_RUNTIME_GRACEFUL_SHUTDOWN_TIMEOUT_SECS.starts_with("DYN_"));
assert!(runtime::system::DYN_SYSTEM_ENABLED.starts_with("DYN_"));
assert!(kvbm::DYN_KVBM_METRICS.starts_with("DYN_"));
assert!(worker::DYN_WORKER_GRACEFUL_SHUTDOWN_TIMEOUT.starts_with("DYN_"));
Expand Down
73 changes: 69 additions & 4 deletions lib/runtime/src/runtime.rs
Original file line number Diff line number Diff line change
Expand Up @@ -24,11 +24,25 @@ use once_cell::sync::OnceCell;
use std::{
mem::ManuallyDrop,
sync::{Arc, atomic::Ordering},
time::Duration,
};
use tokio::{signal, sync::Mutex, task::JoinHandle};

pub use tokio_util::sync::CancellationToken;

const DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_SECS: u64 = 15 * 60;

fn graceful_shutdown_timeout() -> Duration {
let timeout_secs = std::env::var(
config::environment_names::runtime::DYN_RUNTIME_GRACEFUL_SHUTDOWN_TIMEOUT_SECS,
)
.ok()
.and_then(|s| s.parse::<u64>().ok())
.unwrap_or(DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_SECS);

Duration::from_secs(timeout_secs)
}

/// Types of Tokio runtimes that can be used to construct a Dynamo [Runtime].
#[derive(Clone, Debug)]
enum RuntimeType {
Expand Down Expand Up @@ -330,13 +344,22 @@ impl Runtime {
tracing::info!("Active graceful endpoints: {count}");

if count != 0 {
tracker.wait_for_completion().await;
let timeout = graceful_shutdown_timeout();
if tokio::time::timeout(timeout, tracker.wait_for_completion())
.await
.is_err()
{
let remaining = tracker.get_count();
tracing::error!(
timeout_secs = timeout.as_secs(),
remaining_endpoints = remaining,
"Graceful endpoint shutdown timed out; proceeding with runtime teardown"
);
}
}

// Phase 3: Now connections will be disconnected to backend services (e.g. NATS/ETCD) by cancelling the main token
tracing::info!(
"Phase 3: All endpoints ended gracefully. Connections to backend services will now be disconnected"
);
tracing::info!("Phase 3: Connections to backend services will now be disconnected");
main_token.cancel();
});
}
Expand Down Expand Up @@ -391,3 +414,45 @@ impl Drop for RuntimeType {
}
}
}

#[cfg(test)]
mod tests {
use super::*;
use crate::config::environment_names::runtime as env_runtime;

#[tokio::test(start_paused = true)]
async fn shutdown_cancels_main_token_after_graceful_timeout() {
temp_env::async_with_vars(
[(
env_runtime::DYN_RUNTIME_GRACEFUL_SHUTDOWN_TIMEOUT_SECS,
Some("5"),
)],
async {
let runtime = Runtime::from_current().unwrap();
let tracker = runtime.graceful_shutdown_tracker();
let _guard = tracker.register_task();
let main_token = runtime.primary_token();
let endpoint_token = runtime.child_token();

runtime.shutdown();
tokio::task::yield_now().await;

assert!(endpoint_token.is_cancelled());
assert!(!main_token.is_cancelled());
assert_eq!(tracker.get_count(), 1);

tokio::time::advance(Duration::from_secs(4)).await;
tokio::task::yield_now().await;

assert!(!main_token.is_cancelled());

tokio::time::advance(Duration::from_secs(1)).await;
tokio::task::yield_now().await;

assert!(main_token.is_cancelled());
assert_eq!(tracker.get_count(), 1);
},
)
.await;
}
}
Loading