fix(envoy-client): keep the ping silence deadline bounded across reconnects

docs-internal/engine/sleep-sequence.md

@@ -57,7 +57,7 @@ When the grace deadline elapses before `can_finalize_sleep()` returns true:
 - The final `ActorStateStopped` for a lost generation is always `StopCode::Error` with "envoy connection lost", including when Lost escalates a pending graceful stop. `StopCode::Ok` would make the engine destroy the actor.
 - Already-submitted SQLite commits cannot be recalled. Depot generation checks are the backstop for those.
 - A lost generation starts no new user callback. Every runtime callback passes through `refuseCallbacksAfterLost` in `registry/native.ts` when JS enters it, except the `onSleep` and `onDestroy` cleanup wrappers, which release runtime state and skip the user hook themselves. Callbacks the framework starts on its own (workflow steps, run, inspector workflow calls, deferred `onStateChange`, WebSocket open and message listeners, database provider hooks) check the lost state first. Continuations inside a user call that is already running, and abort or close listeners, are not blocked.
-- envoy-client declares all its actors lost once no engine ping has arrived for `envoy_lost_threshold` minus a 3s margin. The engine refreshes its liveness timestamp just before each ping and reallocates actors once that timestamp is older than the threshold, so the envoy normally stops first, including on half-open connections that never report a close. This is best effort: a ping delayed in transit by more than the margin can still let the engine give up first.
+- envoy-client declares all its actors lost once no engine ping has arrived for `envoy_lost_threshold` minus a 3s margin. The engine refreshes its liveness timestamp just before each ping and reallocates actors once that timestamp is older than the threshold, so the envoy normally stops first, including on half-open connections that never report a close. This is best effort: a ping delayed in transit by more than the margin can still let the engine give up first. The deadline counts from `EngineLiveness::baseline` on a monotonic clock. After a reconnect the engine refreshes its liveness timestamp only when it claims the connection, after sending `ToEnvoyInit`, so the baseline keeps the last proven refresh (a ping, or the install time of a connection claimed by commands) until a ping or WebSocket commands arrive on the new connection, then counts from when the client started opening that connection. Serverless start payloads injected over HTTP do not count as a claim. A generation created after the baseline counts from its creation instead, since the engine only assigns it over a claimed connection, and only generations past their own deadline are stopped.
 
 ## Guarding lifecycle requests
 

engine/sdks/rust/envoy-client/src/actor.rs

@@ -1695,6 +1695,7 @@ mod tests {
 			protocol_metadata: Arc::new(tokio::sync::Mutex::new(None)),
 			shutting_down: std::sync::atomic::AtomicBool::new(false),
 			last_ping_ts: std::sync::atomic::AtomicI64::new(0),
+			engine_liveness: Default::default(),
 			stopped_tx: tokio::sync::watch::channel(true).0,
 		});
 		(shared, envoy_rx)

engine/sdks/rust/envoy-client/src/connection/mod.rs

@@ -49,24 +49,34 @@ pub(crate) async fn install_connection(
 			let _ = message.written.send(result);
 		}
 	});
-	install_connection_with_http(shared, tx, http_tx, http_byte_budget).await
+	install_connection_with_http(
+		shared,
+		tx,
+		http_tx,
+		http_byte_budget,
+		crate::time::Instant::now(),
+	)
+	.await
 }
 
 pub(crate) async fn install_connection_with_http(
 	shared: &SharedContext,
 	tx: tokio::sync::mpsc::UnboundedSender<WsTxMessage>,
 	http_tx: tokio::sync::mpsc::Sender<HttpWsTxMessage>,
 	http_byte_budget: std::sync::Arc<tokio::sync::Semaphore>,
+	// When the client started opening this connection. The engine can only claim the connection
+	// after that, so it is a safe lower bound for the claim even when the install itself is late.
+	connect_started_at: crate::time::Instant,
 ) -> u64 {
 	let mut guard = shared.ws_tx.lock().await;
 	let mut http_guard = shared.http_ws_tx.lock().await;
 	let session = shared
 		.next_connection_session
 		.fetch_add(1, Ordering::AcqRel)
 		.saturating_add(1);
-	// A ping from an earlier connection says nothing about this one. Clearing it keeps the engine
-	// ping silence check disabled until this connection receives its first ping.
-	shared.last_ping_ts.store(0, Ordering::Release);
+	shared
+		.engine_liveness
+		.connection_installed(connect_started_at);
 	shared.connection_session.store(session, Ordering::Release);
 	*guard = Some(tx);
 	*http_guard = Some(HttpConnectionTx {
@@ -240,18 +250,20 @@ async fn forward_to_envoy(shared: &SharedContext, session: u64, message: protoco
 
 	match message {
 		protocol::ToEnvoy::ToEnvoyPing(ping) => {
-			let previous_ping_ts = shared
+			shared
 				.last_ping_ts
-				.swap(crate::time::now_millis(), Ordering::AcqRel);
+				.store(crate::time::now_millis(), Ordering::Release);
+			let newly_claimed = shared
+				.engine_liveness
+				.ping_received(crate::time::Instant::now());
 			ws_send(
 				shared,
 				protocol::ToRivet::ToRivetPong(protocol::ToRivetPong { ts: ping.ts }),
 			)
 			.await;
-			// The envoy loop does not arm the engine ping silence check while no ping is
-			// recorded. Wake it on the first ping so the check arms without waiting for an
-			// unrelated event.
-			if previous_ping_ts == 0 {
+			// The first ping on a connection moves the engine ping silence baseline. The envoy
+			// loop computes the silence wait only when a turn starts, so wake it.
+			if newly_claimed {
 				let _ = crate::envoy::send_to_envoy_tx(
 					shared,
 					ToEnvoyMessage::ConnMessage {
@@ -262,6 +274,11 @@ async fn forward_to_envoy(shared: &SharedContext, session: u64, message: protoco
 			}
 		}
 		other => {
+			if matches!(other, protocol::ToEnvoy::ToEnvoyCommands(_)) {
+				// Recorded before forwarding so the envoy loop sees the claim when it handles the
+				// commands.
+				shared.engine_liveness.commands_received();
+			}
 			let _ = crate::envoy::send_to_envoy_tx(
 				shared,
 				ToEnvoyMessage::ConnMessage {

engine/sdks/rust/envoy-client/src/connection/native.rs

@@ -125,6 +125,7 @@ async fn single_connection(
 		.body(())
 		.map_err(|e| anyhow::anyhow!("failed to build ws request: {e}"))?;
 
+	let connect_started_at = crate::time::Instant::now();
 	let (ws_stream, _) =
 		tokio_tungstenite::connect_async_with_config(request, Some(websocket_config()), false)
 			.await?;
@@ -136,8 +137,14 @@ async fn single_connection(
 	let (http_ws_tx, mut http_ws_rx) =
 		mpsc::channel::<HttpWsTxMessage>(super::HTTP_WS_MESSAGE_CAPACITY);
 	let http_byte_budget = Arc::new(Semaphore::new(super::HTTP_WS_BYTE_CAPACITY));
-	let session =
-		super::install_connection_with_http(shared, ws_tx, http_ws_tx, http_byte_budget).await;
+	let session = super::install_connection_with_http(
+		shared,
+		ws_tx,
+		http_ws_tx,
+		http_byte_budget,
+		connect_started_at,
+	)
+	.await;
 
 	tracing::info!(
 		endpoint = %shared.config.endpoint,

engine/sdks/rust/envoy-client/src/connection/wasm.rs

@@ -104,6 +104,7 @@ mod imp {
 	) -> anyhow::Result<(u64, Option<crate::utils::ParsedCloseReason>)> {
 		let url = super::super::ws_url(shared);
 		let protocols = protocols(&shared.config.token);
+		let connect_started_at = crate::time::Instant::now();
 		let ws = WebSocket::new_with_str_sequence(&url, protocols.as_ref())
 			.map_err(|error| anyhow::anyhow!("failed to create websocket: {}", js_error(error)))?;
 		ws.set_binary_type(BinaryType::Arraybuffer);
@@ -177,9 +178,14 @@ mod imp {
 		let (http_ws_tx, mut http_ws_rx) =
 			mpsc::channel::<HttpWsTxMessage>(super::super::HTTP_WS_MESSAGE_CAPACITY);
 		let http_byte_budget = Arc::new(Semaphore::new(super::super::HTTP_WS_BYTE_CAPACITY));
-		let session =
-			super::super::install_connection_with_http(shared, ws_tx, http_ws_tx, http_byte_budget)
-				.await;
+		let session = super::super::install_connection_with_http(
+			shared,
+			ws_tx,
+			http_ws_tx,
+			http_byte_budget,
+			connect_started_at,
+		)
+		.await;
 
 		tracing::info!(
 			endpoint = %shared.config.endpoint,

engine/sdks/rust/envoy-client/src/context.rs

@@ -4,6 +4,7 @@ use std::sync::Mutex as StdMutex;
 use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64};
 
 use crate::async_counter::AsyncCounter;
+use crate::time::Instant;
 use rivet_envoy_protocol as protocol;
 use tokio::sync::{Mutex, Notify, Semaphore, mpsc, oneshot, watch};
 
@@ -43,6 +44,9 @@ pub struct SharedContext {
 	/// `EnvoyHandle::is_ping_healthy` to surface a dead engine link to upstream health checks.
 	/// Zero means no ping has been received yet.
 	pub last_ping_ts: AtomicI64,
+	/// What this envoy can rely on about the engine's liveness clock. Drives the engine ping
+	/// silence check. Only the connection task updates it.
+	pub engine_liveness: EngineLiveness,
 	// Latched signal fired by `envoy_loop` after its cleanup block completes.
 	// Waiters observing `true` are guaranteed that the loop has exited and
 	// every pending KV/SQLite request has been resolved (with `EnvoyShutdownError`
@@ -68,3 +72,71 @@ pub struct HttpWsTxMessage {
 	pub _byte_permit: tokio::sync::OwnedSemaphorePermit,
 	pub written: oneshot::Sender<anyhow::Result<()>>,
 }
+
+/// Tracks the earliest time the engine could have last refreshed this envoy's liveness timestamp.
+///
+/// The engine declares the envoy's actors lost once that timestamp is older than
+/// `envoy_lost_threshold`. It refreshes the timestamp when it claims a connection, which happens
+/// after it sends `ToEnvoyInit`, and before each ping. The engine sends commands and pings on a
+/// connection only after claiming it, so either one proves the claim. Until then the engine may
+/// still be counting from a ping on an earlier connection, for example when the claim fails.
+/// Times are monotonic so wall clock jumps cannot trigger or delay the check.
+#[derive(Default)]
+pub struct EngineLiveness(parking_lot::Mutex<EngineLivenessState>);
+
+#[derive(Default, Clone, Copy)]
+struct EngineLivenessState {
+	/// Latest time the engine is known to have refreshed its liveness timestamp, on this or an
+	/// earlier connection: a ping, or the install time of a connection the engine claimed.
+	last_refresh: Option<Instant>,
+	/// When the client started opening the current connection.
+	connection_installed_at: Option<Instant>,
+	/// Whether the engine has claimed the current connection.
+	connection_claimed: bool,
+}
+
+impl EngineLiveness {
+	/// A new connection was installed. `connect_started_at` is when the client started opening it,
+	/// which precedes any claim even when the engine claims it before the client finishes the
+	/// handshake. The engine has not proven a claim yet, so the last refresh is kept until it does.
+	pub fn connection_installed(&self, connect_started_at: Instant) {
+		let mut state = self.0.lock();
+		state.connection_installed_at = Some(connect_started_at);
+		state.connection_claimed = false;
+	}
+
+	/// A ping arrived on the current connection. Returns true when this is the first evidence that
+	/// the engine claimed the connection.
+	pub fn ping_received(&self, now: Instant) -> bool {
+		let mut state = self.0.lock();
+		state.last_refresh = Some(now);
+		let newly_claimed = !state.connection_claimed;
+		state.connection_claimed = true;
+		newly_claimed
+	}
+
+	/// Commands arrived on the current connection's WebSocket. Returns true when this is the first
+	/// evidence that the engine claimed the connection. Serverless start payloads injected over
+	/// HTTP do not count.
+	pub fn commands_received(&self) -> bool {
+		let mut state = self.0.lock();
+		let newly_claimed = !state.connection_claimed;
+		state.connection_claimed = true;
+		// The claim refreshed the engine's timestamp after the install, so the install time is a
+		// safe lower bound for it.
+		state.last_refresh = match (state.last_refresh, state.connection_installed_at) {
+			(Some(refresh), Some(installed)) => Some(refresh.max(installed)),
+			(refresh, installed) => refresh.or(installed),
+		};
+		newly_claimed
+	}
+
+	/// The time the engine ping silence deadline counts from. Until the engine claims a
+	/// connection it may still be counting from the last refresh, so a reconnect never moves the
+	/// baseline forward on its own. An envoy the engine never refreshed counts from the install,
+	/// which is earlier than any timestamp the engine could hold for it.
+	pub fn baseline(&self) -> Option<Instant> {
+		let state = *self.0.lock();
+		state.last_refresh.or(state.connection_installed_at)
+	}
+}