fix(gas, pb, api): make .removed events noop if dont match, add check for dead wfs to actor/runner errors, make api error types unknown

engine/packages/api-types/src/runner_configs/mod.rs

@@ -8,5 +8,6 @@ pub struct RunnerConfigResponse {
 	#[serde(flatten)]
 	pub config: rivet_types::runner_configs::RunnerConfig,
 	#[serde(default, skip_serializing_if = "Option::is_none")]
+	#[schema(value_type = Option<Object>, additional_properties = true)]
 	pub runner_pool_error: Option<rivet_types::actor::RunnerPoolError>,
 }

engine/packages/gasoline/src/builder/workflow/signal.rs

@@ -8,7 +8,7 @@ use crate::{
 	builder::BuilderError,
 	ctx::WorkflowCtx,
 	error::WorkflowError,
-	history::{cursor::HistoryResult, event::EventType, removed::Signal as RemovedSignal},
+	history::{cursor::HistoryResult, event::EventType},
 	metrics,
 	signal::Signal,
 	workflow::Workflow,
@@ -113,9 +113,7 @@ impl<'a, T: Signal + Serialize> SignalBuilder<'a, T> {
 		}
 
 		// Check if this signal is being replayed and previously had no target (will have a removed event)
-		if self.graceful_not_found && self.ctx.cursor().is_removed() {
-			self.ctx.cursor().compare_removed::<RemovedSignal<T>>()?;
-
+		if self.graceful_not_found && self.ctx.cursor().current_event_is_removed() {
 			tracing::debug!("replaying gracefully not found signal dispatch");
 
 			// Move to next event

engine/packages/gasoline/src/ctx/workflow.rs

@@ -22,7 +22,7 @@
 	executable::{AsyncResult, Executable},
 	history::{
 		History,
-		cursor::{Cursor, HistoryResult},
+		cursor::{Cursor, HistoryResult, RemovedHistoryResult},
 		event::SleepState,
 		location::Location,
 		removed::Removed,
@@ -1108,27 +1108,35 @@
 	pub async fn removed<T: Removed>(&mut self) -> Result<()> {
 		self.check_stop()?;
 
-		// Existing event
-		if self.cursor.compare_removed::<T>()? {
-			tracing::debug!("skipping removed step");
-		}
-		// New "removed" event
-		else {
-			tracing::debug!("inserting removed step");
+		match self.cursor.compare_removed::<T>() {
+			RemovedHistoryResult::New => {
+				tracing::debug!("inserting removed step");
 
-			self.db
-				.commit_workflow_removed_event(
-					self.workflow_id,
-					&self.cursor.current_location(),
-					T::event_type(),
-					T::name(),
-					self.loop_location(),
-				)
-				.await?;
-		};
+				self.db
+					.commit_workflow_removed_event(
+						self.workflow_id,
+						&self.cursor.current_location(),
+						T::event_type(),
+						T::name(),
+						self.loop_location(),
+					)
+					.await?;
 
-		// Move to next event
-		self.cursor.inc();
+				// Move to next event
+				self.cursor.inc();
+			}
+			RemovedHistoryResult::Skip => {
+				tracing::debug!("skipping removed step");
+
+				// Move to next event
+				self.cursor.inc();
+			}
+			RemovedHistoryResult::Ignore(msg) => {
+				tracing::debug!(
+					"removed event filter doesn't match existing event, ignoring: {msg}"
+				);
+			}
+		}
 
 		Ok(())
 	}
@@ -1147,12 +1155,12 @@
 		}
 
 		let history_res = self.cursor.compare_version_check()?;
 		let check_version_location = self.cursor.current_location_for(&history_res);
 
 		let (version, insert) = match history_res {
 			CheckVersionHistoryResult::New => (latest_version, true),
 			CheckVersionHistoryResult::Event(version) => (version, false),
 			CheckVersionHistoryResult::Insertion(next_event_version) => (next_event_version, true),
 		};
 
 		if insert {

engine/packages/gasoline/src/db/mod.rs

@@ -323,9 +323,13 @@ impl WorkflowData {
 			.map_err(WorkflowError::DeserializeWorkflowOutput)
 	}
 
-	pub fn has_output(&self) -> bool {
+	pub fn is_complete(&self) -> bool {
 		self.output.is_some()
 	}
+
+	pub fn is_dead(&self) -> bool {
+		!self.is_complete() && !self.has_wake_condition
+	}
 }
 
 #[derive(Debug)]