fix(envoy-client): ack start commands immediately instead of waiting for the periodic tick

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

@@ -18,11 +18,10 @@ pub async fn handle_commands(ctx: &mut EnvoyContext, commands: Vec<protocol::Com
 		);
 	}
 
-	// Collect actors with a stop in the raw batch before dedup, so a replayed
-	// (skipped) stop is still re-acked instead of being replayed forever.
-	let stopped_actors: Vec<(String, u32)> = commands
+	// Collect every actor in the raw batch before dedup, so a replayed
+	// (skipped) command is still re-acked instead of being replayed forever.
+	let batch_actors: Vec<(String, u32)> = commands
 		.iter()
-		.filter(|c| matches!(c.inner, protocol::Command::CommandStopActor(_)))
 		.map(|c| (c.checkpoint.actor_id.clone(), c.checkpoint.generation))
 		.collect();
 
@@ -91,18 +90,22 @@ pub async fn handle_commands(ctx: &mut EnvoyContext, commands: Vec<protocol::Com
 		}
 	}
 
-	// Ack stops immediately since their actors are removed before the periodic
-	// tick. Scope to just the stopped actors instead of a full-state ack, and do
-	// not clear dedup; the tick handles full re-acks, recovery, and clearing.
-	if !stopped_actors.is_empty() {
-		send_stop_command_acks(ctx, &stopped_actors).await;
+	// Ack the whole batch immediately. Anything left unacked stays in the
+	// engine's `ActorCommandKey` subspace, and `envoy_conn_prepare` re-streams
+	// that subspace on every reconnect, so a start that waits for the periodic
+	// tick can be replayed for up to `ACK_COMMANDS_INTERVAL_MS` and resurrect a
+	// stopped actor or replace a live one. Scope to just this batch's actors
+	// instead of a full-state ack, and do not clear dedup; the tick handles
+	// full re-acks, recovery, and clearing.
+	if !batch_actors.is_empty() {
+		send_batch_command_acks(ctx, &batch_actors).await;
 	}
 }
 
 /// Ack only the given actors' latest processed command index. Used for the
-/// immediate stop ack. Does not clear dedup (see the race note in
+/// immediate post-batch ack. Does not clear dedup (see the race note in
 /// `send_command_ack`); a failed send is retried by the replayed batch or tick.
-async fn send_stop_command_acks(ctx: &EnvoyContext, actors: &[(String, u32)]) {
+async fn send_batch_command_acks(ctx: &EnvoyContext, actors: &[(String, u32)]) {
 	let mut highest: HashMap<(String, u32), i64> = HashMap::new();
 	for key in actors {
 		if let Some(&index) = ctx.processed_command_idx.get(key) {

engine/sdks/rust/envoy-client/tests/command_dedup.rs

@@ -140,6 +140,26 @@ fn stop_command(actor_id: &str, generation: u32, index: i64) -> protocol::Comman
 	}
 }
 
+fn start_command(actor_id: &str, generation: u32, index: i64) -> protocol::CommandWrapper {
+	protocol::CommandWrapper {
+		checkpoint: protocol::ActorCheckpoint {
+			actor_id: actor_id.to_string(),
+			generation,
+			index,
+		},
+		inner: protocol::Command::CommandStartActor(protocol::CommandStartActor {
+			config: protocol::ActorConfig {
+				name: actor_id.to_string(),
+				key: None,
+				create_ts: 0,
+				input: None,
+			},
+			hibernating_requests: Vec::new(),
+			preloaded_kv: None,
+		}),
+	}
+}
+
 fn execute_request() -> protocol::SqliteExecuteRequest {
 	protocol::SqliteExecuteRequest {
 		namespace_id: "test".to_string(),
@@ -289,6 +309,36 @@ fn decode_ack_checkpoints(msg: WsTxMessage) -> Vec<protocol::ActorCheckpoint> {
 	}
 }
 
+#[tokio::test]
+async fn start_command_is_acked_immediately() {
+	let mut ctx = new_envoy_context();
+	let (ws_tx, mut ws_rx) = mpsc::unbounded_channel();
+	*ctx.shared.ws_tx.lock().await = Some(ws_tx);
+
+	handle_commands(&mut ctx, vec![start_command("actor-a", 1, 1)]).await;
+
+	// A start left unacked stays in the engine's command subspace, which is
+	// re-streamed on every reconnect. Waiting for the periodic tick leaves a
+	// window of `ACK_COMMANDS_INTERVAL_MS` in which a reconnect replays the
+	// start and replaces the live actor.
+	let checkpoints = decode_ack_checkpoints(
+		ws_rx
+			.try_recv()
+			.expect("start should trigger an immediate ack"),
+	);
+	assert_eq!(checkpoints.len(), 1);
+	assert_eq!(checkpoints[0].actor_id, "actor-a");
+	assert_eq!(checkpoints[0].generation, 1);
+	assert_eq!(checkpoints[0].index, 1);
+
+	// Dedup is retained so a replay can still be suppressed in-process until
+	// the tick clears it.
+	assert_eq!(
+		ctx.processed_command_idx.get(&("actor-a".to_string(), 1)),
+		Some(&1)
+	);
+}
+
 #[tokio::test]
 async fn stop_command_is_acked_immediately() {
 	let mut ctx = new_envoy_context();