fix(pegboard-envoy): retry failed actor event dispatch instead of dropping events

engine/packages/pegboard-envoy/Cargo.toml

@@ -52,3 +52,7 @@ xxhash-rust.workspace = true
 
 pegboard.workspace = true
 namespace.workspace = true
+
+[dev-dependencies]
+rivet-cache.workspace = true
+rivet-test-deps.workspace = true

engine/packages/pegboard-envoy/src/actor_event_demuxer.rs

@@ -114,6 +114,10 @@ impl ActorEventDemuxer {
 			.map(|(_, channel)| channel.handle)
 			.collect::<Vec<_>>();
 
+		// Handlers still drain buffered events, but stop retrying failed dispatches.
+		// The envoy resends unacknowledged events when it reconnects.
+		self.cancellation_token.cancel();
+
 		// Await remaining tasks
 		for handle in handles {
 			let _ = handle.await;
@@ -134,19 +138,33 @@ async fn channel_handler(
 	loop {
 		let mut buffer = Vec::new();
 
-		// Batch process events
-		tokio::select! {
-			count = rx.recv_many(&mut buffer, 1024) => {
-				if count == 0 {
-					break;
-				}
+		// Batch process events. This returns 0 only once every sender is dropped,
+		// so events buffered before shutdown still dispatch.
+		if rx.recv_many(&mut buffer, 1024).await == 0 {
+			break;
+		}
 
-				if let Err(err) = dispatch_events(&ctx, &envoy_key, actor_id, buffer).await {
-					tracing::error!(%envoy_key, ?actor_id, ?err, "actor event processor failed");
-					break;
+		// Retry the same batch until it lands so later events never skip past it.
+		// The actor workflow ignores events it already received.
+		let mut backoff = rivet_util::throttle::Backoff::new(5, None, 250, 250);
+		loop {
+			match dispatch_events(&ctx, &envoy_key, actor_id, &buffer).await {
+				Ok(()) => break,
+				Err(err) => {
+					tracing::warn!(
+						%envoy_key,
+						?actor_id,
+						?err,
+						attempt = backoff.tick_index(),
+						count = buffer.len(),
+						"actor event dispatch failed, retrying, storage may be unavailable"
+					);
+					tokio::select! {
+						_ = backoff.tick() => {}
+						_ = cancellation_token.cancelled() => return,
+					}
 				}
 			}
-			_ = cancellation_token.cancelled() => break,
 		}
 	}
 }
@@ -156,14 +174,14 @@ async fn dispatch_events(
 	ctx: &StandaloneCtx,
 	envoy_key: &str,
 	actor_id: Id,
-	events: Vec<protocol::EventWrapper>,
+	events: &[protocol::EventWrapper],
 ) -> Result<()> {
 	tracing::debug!(%envoy_key, ?actor_id, count=?events.len(), "actor demuxer dispatch");
 
 	let res = ctx
 		.signal(pegboard::workflows::actor2::Events {
 			envoy_key: envoy_key.to_string(),
-			events,
+			events: events.to_vec(),
 		})
 		.to_workflow::<pegboard::workflows::actor2::Workflow>()
 		.tag("actor_id", actor_id)

engine/packages/pegboard-envoy/src/lib.rs

@@ -13,7 +13,7 @@ use tokio::sync::watch;
 use tokio_tungstenite::tungstenite::protocol::frame::CloseFrame;
 use universalpubsub::PublishOpts;
 
-mod actor_event_demuxer;
+pub mod actor_event_demuxer;
 mod actor_kv_task;
 mod actor_lifecycle;
 mod actor_remote_sqlite_task;

engine/packages/pegboard-envoy/tests/actor_event_demuxer.rs

@@ -0,0 +1,162 @@
+use std::{net::SocketAddr, time::Duration};
+
+use anyhow::{Context, Result};
+use gas::db::debug::DatabaseDebug;
+use gas::prelude::*;
+use pegboard_envoy::actor_event_demuxer::ActorEventDemuxer;
+use rivet_envoy_protocol as protocol;
+
+const ENVOY_KEY: &str = "demuxer-test-envoy";
+
+fn state_event(actor_id: Id, index: i64, state: protocol::ActorState) -> protocol::EventWrapper {
+	protocol::EventWrapper {
+		checkpoint: protocol::ActorCheckpoint {
+			actor_id: actor_id.to_string(),
+			generation: 1,
+			index,
+		},
+		inner: protocol::Event::EventActorStateUpdate(protocol::EventActorStateUpdate { state }),
+	}
+}
+
+async fn build_ctx(
+	config: rivet_config::Config,
+	pools: rivet_pools::Pools,
+) -> Result<StandaloneCtx> {
+	let cache = rivet_cache::CacheInner::from_env(&config, pools.clone())?;
+	Ok(StandaloneCtx::new(
+		db::DatabaseKv::new(config.clone(), pools.clone()).await?,
+		config.clone(),
+		pools,
+		cache,
+		"actor_event_demuxer_test",
+		Id::new_v1(config.dc_label()),
+		Id::new_v1(config.dc_label()),
+	)?)
+}
+
+/// The event indices the actor workflow has received, in signal order.
+async fn delivered_indices(debug_db: &db::DatabaseKv, workflow_id: Id) -> Result<Vec<i64>> {
+	let mut signals = debug_db
+		.find_signals(&[], Some(workflow_id), Some("pegboard_actor2_events"), None)
+		.await?;
+	// Signal ids are random, so the debug query does not return signals in send order.
+	signals.sort_by_key(|signal| signal.create_ts);
+	let mut indices = Vec::new();
+	for signal in signals {
+		for event in signal.body["events"].as_array().context("events array")? {
+			indices.push(
+				event["checkpoint"]["index"]
+					.as_i64()
+					.context("event index")?,
+			);
+		}
+	}
+	Ok(indices)
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn events_dispatched_while_storage_is_unreachable_still_reach_the_actor() -> Result<()> {
+	// Storage must be reachable over TCP so the test can cut the connection.
+	// SAFETY: set before any other thread reads the environment.
+	unsafe { std::env::set_var("RIVET_TEST_DATABASE", "postgres") };
+	let test_deps = rivet_test_deps::TestDeps::new().await?;
+	// Single-node Postgres allows one leader per database. Hand the lease to the
+	// proxied pools below, which carry every storage call in this test.
+	test_deps.pools().udb()?.shutdown().await;
+
+	// Every storage call goes through a proxy that the test can disable.
+	let rivet_config::config::Database::Postgres(postgres) = test_deps.config().database() else {
+		anyhow::bail!("test database must be postgres");
+	};
+	let url = postgres.url.read().clone();
+	let port: u16 = url
+		.split_whitespace()
+		.find_map(|part| part.strip_prefix("port="))
+		.context("postgres port")?
+		.parse()?;
+	let network_faults = rivet_test_deps::ToxiproxyTestServer::start().await?;
+	let proxy = network_faults
+		.proxy("postgres", SocketAddr::from(([127, 0, 0, 1], port)))
+		.await?;
+	let mut root = (**test_deps.config()).clone();
+	root.database = Some(rivet_config::config::Database::Postgres(
+		rivet_config::config::db::Postgres {
+			url: rivet_config::secret::Secret::new(url.replace(
+				&format!("port={port}"),
+				&format!("port={}", proxy.listen_addr().port()),
+			)),
+			..postgres.clone()
+		},
+	));
+	let proxied_config = rivet_config::Config::from_root(root);
+	let proxied_pools = rivet_pools::Pools::new(proxied_config.clone()).await?;
+	let ctx = build_ctx(proxied_config.clone(), proxied_pools.clone()).await?;
+
+	// A workflow with no worker keeps every signal it receives, so the test can read them.
+	let actor_id = Id::new_v1(test_deps.config().dc_label());
+	let workflow_id = ctx
+		.workflow(pegboard::workflows::actor2::Input {
+			actor_id,
+			name: "demuxer-test".to_string(),
+			pool_name: "demuxer-test".to_string(),
+			key: None,
+			namespace_id: Id::new_v1(test_deps.config().dc_label()),
+			input: None,
+			from_v1: false,
+		})
+		.tag("actor_id", actor_id)
+		.dispatch()
+		.await?;
+
+	let mut demuxer = ActorEventDemuxer::new(ctx, ENVOY_KEY.to_string());
+
+	proxy.disable().await?;
+	demuxer.ingest(
+		actor_id,
+		state_event(actor_id, 1, protocol::ActorState::ActorStateRunning),
+	);
+	// Keep storage unreachable long enough for the dispatch of event 1 to fail.
+	// A refused connection fails the transaction at once. The outage stays well
+	// under the 10 second leader lease, so the leader keeps running.
+	tokio::time::sleep(Duration::from_secs(2)).await;
+	proxy.enable().await?;
+
+	demuxer.ingest(
+		actor_id,
+		state_event(
+			actor_id,
+			2,
+			protocol::ActorState::ActorStateStopped(protocol::ActorStateStopped {
+				code: protocol::StopCode::Ok,
+				message: None,
+			}),
+		),
+	);
+
+	let debug_db = db::DatabaseKv::new(proxied_config, proxied_pools).await?;
+	// Delivery is asynchronous and has no completion signal visible to the test,
+	// so poll the workflow's signals until both events arrive.
+	let delivered = tokio::time::timeout(Duration::from_secs(20), async {
+		loop {
+			let indices = delivered_indices(&debug_db, workflow_id).await?;
+			if indices.len() >= 2 {
+				return Ok::<_, anyhow::Error>(indices);
+			}
+			tokio::time::sleep(Duration::from_millis(100)).await;
+		}
+	})
+	.await;
+
+	demuxer.shutdown().await;
+	let delivered = match delivered {
+		Ok(indices) => indices?,
+		Err(_) => delivered_indices(&debug_db, workflow_id).await?,
+	};
+	assert_eq!(
+		delivered,
+		vec![1, 2],
+		"every event must reach the actor, in order"
+	);
+	Ok(())
+}