fix(envoy-client): complete command and request lifecycles

Author: NathanFlurry

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

@@ -558,6 +558,7 @@ fn handle_req_start(
 				}
 				Err(error) => {
 					tracing::error!(?error, "fetch failed");
+					send_fetch_error_response(&shared, gateway_id, request_id).await;
 				}
 			}
 		}
@@ -1343,6 +1344,40 @@ async fn send_response(
 	}
 }
 
+async fn send_fetch_error_response(
+	shared: &SharedContext,
+	gateway_id: protocol::GatewayId,
+	request_id: protocol::RequestId,
+) {
+	let body = br#"{"code":"envoy_fetch_failed","message":"actor fetch failed"}"#.to_vec();
+	let mut headers = HashableMap::new();
+	headers.insert("content-length".to_string(), body.len().to_string());
+	headers.insert(
+		"x-rivet-error".to_string(),
+		"envoy.fetch_failed".to_string(),
+	);
+
+	ws_send(
+		shared,
+		protocol::ToRivet::ToRivetTunnelMessage(protocol::ToRivetTunnelMessage {
+			message_id: protocol::MessageId {
+				gateway_id,
+				request_id,
+				message_index: 0,
+			},
+			message_kind: protocol::ToRivetTunnelMessageKind::ToRivetResponseStart(
+				protocol::ToRivetResponseStart {
+					status: 500,
+					headers,
+					body: Some(body),
+					stream: false,
+				},
+			),
+		}),
+	)
+	.await;
+}
+
 #[cfg(test)]
 mod tests {
 	use std::collections::HashMap;

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

@@ -485,6 +485,7 @@ async fn handle_conn_message(
 		}
 		protocol::ToEnvoy::ToEnvoyCommands(commands) => {
 			handle_commands(ctx, commands).await;
+			send_command_ack(ctx).await;
 		}
 		protocol::ToEnvoy::ToEnvoyAckEvents(ack) => {
 			handle_ack_events(ctx, ack);

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

@@ -23,10 +23,7 @@ pub async fn handle_send_events(ctx: &mut EnvoyContext, events: Vec<protocol::Ev
 					state_update.state,
 					protocol::ActorState::ActorStateStopped(_)
 				) {
-					// If the actor is being stopped by rivet, we don't need the entry anymore
-					if entry.received_stop {
-						remove_after_stop = true;
-					}
+					remove_after_stop = true;
 				}
 			}
 		}
@@ -50,8 +47,6 @@ pub fn handle_ack_events(ctx: &mut EnvoyContext, ack: protocol::ToEnvoyAckEvents
 	}
 }
 
-// TODO: If the envoy disconnects, actor stops, then envoy reconnects, we will send the stop event but there
-// is no mechanism to remove the actor entry afterwards. We only remove the actor entry if rivet stops the actor.
 pub async fn resend_unacknowledged_events(ctx: &EnvoyContext) {
 	let mut events: Vec<protocol::EventWrapper> = Vec::new();