chore: add op name to logs (#3564)

Author: MasterPtato

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

@@ -109,8 +109,8 @@ where
 
 	let res = tokio::time::timeout(I::Operation::TIMEOUT, I::Operation::run(&ctx, &input))
 		.await
-		.map_err(|_| WorkflowError::OperationTimeout(0))
-		.map(|res| res.map_err(WorkflowError::OperationFailure));
+		.map_err(|_| WorkflowError::OperationTimeout(I::Operation::NAME, 0))
+		.map(|res| res.map_err(|err| WorkflowError::OperationFailure(I::Operation::NAME, err)));
 
 	// Record metrics
 	{

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

@@ -618,12 +618,12 @@ impl WorkflowCtx {
 									WorkflowError::ActivityTimeout(error_count)
 								}
 							}
-							WorkflowError::OperationTimeout(_) => {
+							WorkflowError::OperationTimeout(op_name, _) => {
 								if error_count.saturating_add(1) >= I::Activity::MAX_RETRIES {
 									WorkflowError::ActivityMaxFailuresReached(err.into())
 								} else {
 									// Add error count to the error for backoff calculation
-									WorkflowError::OperationTimeout(error_count)
+									WorkflowError::OperationTimeout(op_name, error_count)
 								}
 							}
 							_ => err,

engine/packages/gasoline/src/error.rs

@@ -19,8 +19,8 @@ pub enum WorkflowError {
 	#[error("activity failure, max retries reached: {0:?}")]
 	ActivityMaxFailuresReached(#[source] anyhow::Error),
 
-	#[error("operation failure: {0:?}")]
-	OperationFailure(#[source] anyhow::Error),
+	#[error("operation failure ({0}): {1:?}")]
+	OperationFailure(&'static str, #[source] anyhow::Error),
 
 	#[error("workflow missing from registry: {0}")]
 	WorkflowMissingFromRegistry(String),
@@ -150,8 +150,8 @@ pub enum WorkflowError {
 	ActivityTimeout(usize),
 
 	// Includes error count
-	#[error("operation timed out")]
-	OperationTimeout(usize),
+	#[error("operation {0} timed out")]
+	OperationTimeout(&'static str, usize),
 
 	#[error("duplicate registered workflow: {0}")]
 	DuplicateRegisteredWorkflow(String),
@@ -188,7 +188,7 @@ impl WorkflowError {
 		match self {
 			WorkflowError::ActivityFailure(_, error_count)
 			| WorkflowError::ActivityTimeout(error_count)
-			| WorkflowError::OperationTimeout(error_count) => {
+			| WorkflowError::OperationTimeout(_, error_count) => {
 				// NOTE: Max retry is handled in `WorkflowCtx::activity`
 				let mut backoff = rivet_util::backoff::Backoff::new_at(
 					8,
@@ -220,7 +220,7 @@ impl WorkflowError {
 		match self {
 			WorkflowError::ActivityFailure(_, _)
 			| WorkflowError::ActivityTimeout(_)
-			| WorkflowError::OperationTimeout(_)
+			| WorkflowError::OperationTimeout(_, _)
 			| WorkflowError::NoSignalFound(_)
 			| WorkflowError::NoSignalFoundAndSleep(_, _)
 			| WorkflowError::SubWorkflowIncomplete(_)
@@ -235,7 +235,7 @@ impl WorkflowError {
 		match self {
 			WorkflowError::ActivityFailure(_, _)
 			| WorkflowError::ActivityTimeout(_)
-			| WorkflowError::OperationTimeout(_) => true,
+			| WorkflowError::OperationTimeout(_, _) => true,
 			_ => false,
 		}
 	}

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

@@ -61,6 +61,7 @@ pub async fn start(config: rivet_config::Config, pools: rivet_pools::Pools) -> R
 	}
 }
 
+#[tracing::instrument(skip_all)]
 async fn tick(
 	ctx: &StandaloneCtx,
 	outbound_connections: &mut HashMap<(Id, String), Vec<OutboundConnection>>,
@@ -157,6 +158,7 @@ async fn tick(
 	Ok(())
 }
 
+#[tracing::instrument(skip_all)]
 async fn tick_runner_config(
 	ctx: &StandaloneCtx,
 	ns_id: Id,
@@ -264,32 +266,35 @@ fn spawn_connection(
 	let draining = Arc::new(AtomicBool::new(false));
 
 	let draining2 = draining.clone();
-	let handle = tokio::spawn(async move {
-		if let Err(err) = outbound_handler(
-			&ctx,
-			url,
-			headers,
-			request_lifespan,
-			slots_per_runner,
-			runner_name,
-			namespace_name,
-			shutdown_rx,
-			draining2,
-		)
-		.await
-		{
-			tracing::warn!(?err, "outbound req failed");
+	let handle = tokio::spawn(
+		async move {
+			if let Err(err) = outbound_handler(
+				&ctx,
+				url,
+				headers,
+				request_lifespan,
+				slots_per_runner,
+				runner_name,
+				namespace_name,
+				shutdown_rx,
+				draining2,
+			)
+			.await
+			{
+				tracing::warn!(?err, "outbound req failed");
 
-			// TODO: Add backoff
-			tokio::time::sleep(Duration::from_secs(1)).await;
+				// TODO: Add backoff
+				tokio::time::sleep(Duration::from_secs(1)).await;
 
-			// On error, bump the autoscaler loop again
-			let _ = ctx
-				.msg(rivet_types::msgs::pegboard::BumpServerlessAutoscaler {})
-				.send()
-				.await;
+				// On error, bump the autoscaler loop again
+				let _ = ctx
+					.msg(rivet_types::msgs::pegboard::BumpServerlessAutoscaler {})
+					.send()
+					.await;
+			}
 		}
-	});
+		.custom_instrument(tracing::info_span!("outbound_req_task")),
+	);
 
 	OutboundConnection {
 		handle,
@@ -298,6 +303,7 @@ fn spawn_connection(
 	}
 }
 
+#[tracing::instrument(skip_all)]
 async fn outbound_handler(
 	ctx: &StandaloneCtx,
 	url: String,
@@ -478,6 +484,7 @@ async fn outbound_handler(
 	Ok(())
 }
 
+#[tracing::instrument(skip_all)]
 async fn drain_runner(ctx: &StandaloneCtx, runner_id: Id) -> Result<()> {
 	let res = ctx
 		.signal(pegboard::workflows::runner::Stop {
@@ -498,7 +505,7 @@ async fn drain_runner(ctx: &StandaloneCtx, runner_id: Id) -> Result<()> {
 			"runner workflow not found, likely already stopped"
 		);
 	} else {
-		res?;
+		res.context("failed sending drain signal")?;
 	}
 
 	Ok(())
@@ -507,6 +514,7 @@ async fn drain_runner(ctx: &StandaloneCtx, runner_id: Id) -> Result<()> {
 /// Send a stop message to the client.
 ///
 /// This will close the runner's WebSocket.
+#[tracing::instrument(skip_all)]
 async fn publish_to_client_stop(ctx: &StandaloneCtx, runner_id: Id) -> Result<()> {
 	let receiver_subject =
 		pegboard::pubsub_subjects::RunnerReceiverSubject::new(runner_id).to_string();