use super::carriers::{ChoiceOutcome, ConfiguredChooser};
use super::{Configuration, ServerError, refusal};
use crate::config::types::OutboundLinkConfig;
use crate::link_bridge::{BridgeCounters, BridgeRegistry, LinkBridge};
use crate::link_open::Stop;
use crate::server::ConnectionSupervisor;
use std::sync::{Arc, Mutex, TryLockError};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};
#[path = "retry/wake.rs"]
mod wake;
use wake::Wake;
#[cfg(test)]
#[path = "retry/tests.rs"]
mod tests;
const FIRST_RETRY: Duration = Duration::from_secs(1);
const MAX_RETRY: Duration = Duration::from_secs(30);
#[derive(Debug)]
struct Slot {
entry: OutboundLinkConfig,
bridge: Option<LinkBridge>,
retry_at: Option<Instant>,
backoff: Duration,
stopped: bool,
}
impl Slot {
fn schedule(&mut self) -> Result<(), ServerError> {
self.schedule_at(Instant::now())
}
fn schedule_at(&mut self, now: Instant) -> Result<(), ServerError> {
self.retry_at = Some(
now.checked_add(self.backoff)
.ok_or_else(|| failure("configured retry deadline overflow"))?,
);
self.backoff = self.backoff.saturating_mul(2).min(MAX_RETRY);
Ok(())
}
}
type Slots = Vec<Arc<Mutex<Slot>>>;
pub(super) struct RetryLinks {
wake: Arc<Wake>,
slots: Slots,
#[cfg(feature = "test-support")]
registry: BridgeRegistry,
stop: Stop,
worker: Option<JoinHandle<Result<(), ServerError>>>,
}
impl std::fmt::Debug for RetryLinks {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("RetryLinks")
.field("slots", &self.slots)
.finish_non_exhaustive()
}
}
impl RetryLinks {
pub(super) fn start(
configuration: &Configuration,
registry: &BridgeRegistry,
supervisor: &ConnectionSupervisor,
) -> Result<Self, ServerError> {
let mut chooser = ConfiguredChooser::new(&configuration.host, &configuration.machine)
.with_topology(configuration.services.clone())
.with_endpoint(supervisor)?;
let wake = Arc::new(Wake::default());
let mut slots = Vec::new();
for entry in &configuration.machine.links {
registry.remember(entry.link_id.clone(), BridgeCounters::default())?;
let mut slot = Slot {
entry: entry.clone(),
bridge: None,
retry_at: None,
backoff: FIRST_RETRY,
stopped: false,
};
attempt(
&mut slot,
&mut chooser,
&configuration.services,
registry,
&wake,
true,
)?;
slots.push(Arc::new(Mutex::new(slot)));
}
let stop = chooser.stop_handle();
let worker_slots = slots.clone();
let worker_wake = wake.clone();
let worker_registry = registry.clone();
let services = configuration.services.clone();
let dispatcher = tracing::dispatcher::get_default(Clone::clone);
let worker = std::thread::Builder::new()
.name("liminal-link-retry".to_owned())
.spawn(move || {
tracing::dispatcher::with_default(&dispatcher, || {
let result = run(
&worker_slots,
&worker_wake,
&mut chooser,
&services,
&worker_registry,
);
if let Err(error) = &result {
tracing::error!(%error, "configured link retry owner failed");
}
result
})
})
.map_err(|error| failure(format!("configured retry worker: {error}")))?;
Ok(Self {
wake,
slots,
stop,
worker: Some(worker),
#[cfg(feature = "test-support")]
registry: registry.clone(),
})
}
pub(super) fn finish(&mut self) -> Result<(), ServerError> {
let mut failure = None;
if let Err(error) = self.wake.stop() {
retain_failure(&mut failure, error);
}
for slot in &self.slots {
match slot.try_lock() {
Ok(slot) => {
if let Some(bridge) = &slot.bridge
&& let Err(error) = bridge.request_shutdown()
{
retain_failure(&mut failure, error);
}
}
Err(TryLockError::WouldBlock) => {}
Err(TryLockError::Poisoned(_)) => {
retain_failure(&mut failure, self::failure("configured link slot poisoned"));
}
}
}
if let Err(error) = (self.stop)() {
retain_failure(&mut failure, self::failure(error.0));
}
if let Some(worker) = self.worker.take() {
match worker.join() {
Ok(Ok(())) => {}
Ok(Err(error)) => {
if failure.is_none() {
retain_failure(&mut failure, error);
}
}
Err(_) => retain_failure(
&mut failure,
self::failure("configured retry worker panicked"),
),
}
}
failure.map_or(Ok(()), Err)
}
#[cfg(feature = "test-support")]
pub(super) fn observe_answers(&self, count: u64, bound: Duration) -> Result<bool, ServerError> {
let mut ids = Vec::new();
for slot in &self.slots {
let slot = lock(slot)?;
if !slot.stopped {
ids.push(slot.entry.link_id.clone());
}
}
self.registry.observe_answers(&ids, count, bound)
}
#[cfg(feature = "test-support")]
pub(super) fn finish_first(&self) -> Result<(), ServerError> {
let mut slot = lock(
self.slots
.first()
.ok_or_else(|| failure("configured bridge absent"))?,
)?;
slot.stopped = true;
slot.retry_at = None;
slot.bridge
.take()
.ok_or_else(|| failure("configured bridge absent"))?
.finish()
}
}
impl Drop for RetryLinks {
fn drop(&mut self) {
if self.worker.is_some()
&& let Err(error) = self.finish()
{
tracing::error!(%error, "configured link retry shutdown failed");
}
}
}
fn retain_failure(first: &mut Option<ServerError>, error: ServerError) {
tracing::error!(%error, "configured link shutdown failed");
if first.is_none() {
*first = Some(error);
}
}
fn failure(message: impl Into<String>) -> ServerError {
ServerError::ListenerAccept {
message: message.into(),
}
}
fn lock(slot: &Mutex<Slot>) -> Result<std::sync::MutexGuard<'_, Slot>, ServerError> {
slot.lock()
.map_err(|_| failure("configured link slot poisoned"))
}
fn attempt(
slot: &mut Slot,
chooser: &mut ConfiguredChooser,
services: &Arc<crate::server::connection::services::LiminalConnectionServices>,
registry: &BridgeRegistry,
wake: &Arc<Wake>,
initial: bool,
) -> Result<(), ServerError> {
let outcome = chooser.open(&slot.entry);
if wake.stopping()? {
slot.stopped = true;
return Ok(());
}
let outcome = outcome.map_err(|error| refusal(error.0))?;
answered(
slot,
outcome,
services,
registry,
wake,
initial,
Instant::now(),
)
}
fn answered(
slot: &mut Slot,
outcome: ChoiceOutcome,
services: &Arc<crate::server::connection::services::LiminalConnectionServices>,
registry: &BridgeRegistry,
wake: &Arc<Wake>,
initial: bool,
now: Instant,
) -> Result<(), ServerError> {
match outcome {
ChoiceOutcome::Opened(link) => {
let counters = registry
.snapshot()
.remove(&slot.entry.link_id)
.unwrap_or_default();
let bridge = LinkBridge::start_resuming(
*link,
services.clone(),
slot.entry.peer_holder.clone(),
counters,
)?;
let signal = Arc::downgrade(wake);
bridge.notify_end(Arc::new(move || {
if let Some(signal) = signal.upgrade() {
signal.notify();
}
}))?;
bridge.install_observations(registry);
slot.bridge = Some(bridge);
slot.retry_at = None;
slot.backoff = FIRST_RETRY;
}
ChoiceOutcome::DialRefused { reason } => {
registry.dial_refused(&slot.entry.link_id, reason.clone())?;
if initial {
tracing::warn!(link = %slot.entry.link_id, %reason, "outbound link down; dial will be retried");
}
slot.schedule_at(now)?;
}
ChoiceOutcome::Held { grant, draft } => {
tracing::warn!(link = %slot.entry.link_id, %grant, %draft, "outbound link held; no bridge started");
slot.stopped = true;
}
ChoiceOutcome::Refused { reason } => {
let changed = registry.open_refused(&slot.entry.link_id, reason.clone())?;
if initial || changed {
tracing::warn!(link = %slot.entry.link_id, %reason, "outbound link refused; dial will be retried");
}
slot.schedule_at(now)?;
}
}
Ok(())
}
fn run(
slots: &Slots,
wake: &Arc<Wake>,
chooser: &mut ConfiguredChooser,
services: &Arc<crate::server::connection::services::LiminalConnectionServices>,
registry: &BridgeRegistry,
) -> Result<(), ServerError> {
while !wake.stopping()? {
let mut deadline = None;
for slot in slots {
if wake.stopping()? {
break;
}
let mut slot = lock(slot)?;
if slot.stopped {
continue;
}
if let Some(bridge) = &slot.bridge
&& bridge.end_reason()?.as_deref() == Some("peer_gone")
{
let bridge = slot
.bridge
.take()
.ok_or_else(|| failure("ended configured bridge absent"))?;
registry.remember(slot.entry.link_id.clone(), bridge.finish_retaining()?)?;
slot.backoff = FIRST_RETRY;
slot.schedule()?;
}
if slot.retry_at.is_some_and(|retry| retry <= Instant::now()) {
attempt(&mut slot, chooser, services, registry, wake, false)?;
}
if let Some(retry) = slot.retry_at {
deadline = Some(deadline.map_or(retry, |existing: Instant| existing.min(retry)));
}
}
wake.wait(deadline)?;
}
let mut failure = None;
for slot in slots {
if let Some(bridge) = lock(slot)?.bridge.take()
&& let Err(error) = bridge.finish()
{
retain_failure(&mut failure, error);
}
}
if let Err(error) = chooser.finish() {
retain_failure(&mut failure, error);
}
failure.map_or(Ok(()), Err)
}