diff --git a/Cargo.lock b/Cargo.lock index 65dc1f5f05..04323efadb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4067,6 +4067,7 @@ dependencies = [ "namespace", "pegboard", "rand 0.8.5", + "rivet-cache", "rivet-config", "rivet-data", "rivet-depot-client", @@ -4077,6 +4078,7 @@ dependencies = [ "rivet-perf", "rivet-pools", "rivet-runtime", + "rivet-test-deps", "rivet-types", "rivet-util", "rusqlite", diff --git a/engine/packages/pegboard-envoy/Cargo.toml b/engine/packages/pegboard-envoy/Cargo.toml index 4d7f5f54ab..d1494a38ff 100644 --- a/engine/packages/pegboard-envoy/Cargo.toml +++ b/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 diff --git a/engine/packages/pegboard-envoy/src/actor_event_demuxer.rs b/engine/packages/pegboard-envoy/src/actor_event_demuxer.rs index 125689ab57..c4498a4a31 100644 --- a/engine/packages/pegboard-envoy/src/actor_event_demuxer.rs +++ b/engine/packages/pegboard-envoy/src/actor_event_demuxer.rs @@ -114,6 +114,10 @@ impl ActorEventDemuxer { .map(|(_, channel)| channel.handle) .collect::>(); + // 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, + 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::() .tag("actor_id", actor_id) diff --git a/engine/packages/pegboard-envoy/src/lib.rs b/engine/packages/pegboard-envoy/src/lib.rs index 21c6e33ca9..de8a798fff 100644 --- a/engine/packages/pegboard-envoy/src/lib.rs +++ b/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; diff --git a/engine/packages/pegboard-envoy/tests/actor_event_demuxer.rs b/engine/packages/pegboard-envoy/tests/actor_event_demuxer.rs new file mode 100644 index 0000000000..6c2d686140 --- /dev/null +++ b/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 { + 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> { + 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(()) +}