1use std::{
2 collections::{HashMap, HashSet},
3 error::Error,
4 fmt::{Display, Formatter},
5 os::unix::fs::FileTypeExt,
6 path::{Path, PathBuf},
7 sync::{
8 Arc, Mutex,
9 atomic::{AtomicU64, Ordering},
10 },
11};
12
13use anyhow::{Context, Result, ensure};
14use async_trait::async_trait;
15use serde::{Deserialize, Serialize};
16use serde_json::Value;
17use tokio::{
18 io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
19 net::{UnixListener, unix::OwnedWriteHalf},
20 sync::{Mutex as AsyncMutex, oneshot},
21 task::JoinHandle,
22};
23use tokio_util::sync::CancellationToken;
24use tracing::{debug, info};
25
26use super::{ActorInvocationFailure, ActorKey};
27
28const ACTOR_EXECUTOR_PROTOCOL_VERSION: u32 = 13;
29pub(crate) const MAX_ACTOR_EXECUTOR_MESSAGE_BYTES: usize = 32 * 1024 * 1024;
30
31#[derive(Debug, Serialize)]
32pub struct ActorMethodInvocation {
33 pub request_id: String,
34 pub actor: ActorKey,
35 pub method: String,
36 pub args: Vec<Value>,
37 pub connections: Vec<ActorSocketConnection>,
38}
39
40#[derive(Debug, Serialize)]
41pub struct ActorMethodEviction {
42 pub actor: ActorKey,
43}
44
45#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
46pub struct ActorSocketConnection {
47 pub id: String,
48 pub metadata: Value,
49 pub tags: Vec<String>,
50}
51
52#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
53#[serde(tag = "type", rename_all = "snake_case")]
54pub enum ActorSocketMessage {
55 Text { data: String },
56 Binary { data: String },
57}
58
59#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
60#[serde(tag = "type", rename_all = "snake_case")]
61pub enum ActorSocketEvent {
62 Connect {
63 connection: ActorSocketConnection,
64 },
65 Message {
66 connection_id: String,
67 message: ActorSocketMessage,
68 },
69 Disconnect {
70 connection: ActorSocketConnection,
71 code: u16,
72 reason: String,
73 was_clean: bool,
74 },
75}
76
77#[derive(Debug, Serialize)]
78pub struct ActorSocketInvocation {
79 pub request_id: String,
80 pub actor: ActorKey,
81 pub event: ActorSocketEvent,
82 pub connections: Vec<ActorSocketConnection>,
83}
84
85#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
86#[serde(tag = "type", rename_all = "snake_case")]
87pub enum ActorSocketEffect {
88 Send {
89 connection_id: String,
90 message: ActorSocketMessage,
91 },
92 Broadcast {
93 message: ActorSocketMessage,
94 except_connection_ids: Vec<String>,
95 tags: Vec<String>,
96 },
97 Close {
98 connection_id: String,
99 code: u16,
100 reason: String,
101 },
102 Reject {
103 connection_id: String,
104 code: u16,
105 reason: String,
106 },
107 SetMetadata {
108 connection_id: String,
109 metadata: Value,
110 },
111 SetTags {
112 connection_id: String,
113 tags: Vec<String>,
114 },
115}
116
117#[derive(Debug, PartialEq)]
118pub enum ActorMethodOutcome {
119 Completed {
120 result: Value,
121 state: Value,
122 effects: Vec<ActorSocketEffect>,
123 },
124 Failed(ActorInvocationFailure),
125}
126
127#[derive(Debug, PartialEq)]
128pub enum ActorSocketOutcome {
129 Handled {
130 state: Value,
131 effects: Vec<ActorSocketEffect>,
132 },
133 Failed(ActorInvocationFailure),
134}
135
136#[async_trait]
137pub trait ActorExecutor: Send + Sync {
138 fn supports(&self, actor_type: &str) -> bool;
139
140 async fn invoke(
141 &self,
142 invocation: ActorMethodInvocation,
143 state: Option<&Value>,
144 ) -> Result<ActorMethodOutcome>;
145
146 async fn handle_socket(
147 &self,
148 _invocation: ActorSocketInvocation,
149 _state: Option<&Value>,
150 ) -> Result<ActorSocketOutcome> {
151 Ok(ActorSocketOutcome::Failed(ActorInvocationFailure {
152 code: "socket_not_supported".into(),
153 message: "actor executor does not support sockets".into(),
154 }))
155 }
156
157 async fn evict(&self, _eviction: ActorMethodEviction) -> Result<()> {
158 Ok(())
159 }
160}
161
162pub(crate) struct ActorExecutorListener {
163 listener: UnixListener,
164 socket_path: PathBuf,
165}
166
167impl ActorExecutorListener {
168 pub(crate) async fn bind(socket_path: impl Into<PathBuf>) -> Result<Self> {
169 let socket_path = socket_path.into();
170 prepare_socket_path(&socket_path).await?;
171 if let Some(parent) = socket_path.parent() {
172 tokio::fs::create_dir_all(parent)
173 .await
174 .with_context(|| format!("create actor executor directory {}", parent.display()))?;
175 }
176 let listener = UnixListener::bind(&socket_path)
177 .with_context(|| format!("bind actor executor socket {}", socket_path.display()))?;
178 Ok(Self {
179 listener,
180 socket_path,
181 })
182 }
183
184 pub(crate) async fn accept(self) -> Result<ActorExecutorConnection> {
185 let result = self.accept_connection().await;
186 let cleanup = remove_socket(&self.socket_path).await;
187 match (result, cleanup) {
188 (Ok(connection), Ok(())) => Ok(connection),
189 (Err(error), _) => Err(error),
190 (Ok(_), Err(error)) => Err(error),
191 }
192 }
193
194 async fn accept_connection(&self) -> Result<ActorExecutorConnection> {
195 let (stream, _) =
196 self.listener.accept().await.with_context(|| {
197 format!("accept actor executor at {}", self.socket_path.display())
198 })?;
199 let (reader, writer) = stream.into_split();
200 let mut reader = BufReader::new(reader);
201 let attach = match read_client_message(&mut reader).await? {
202 Some(ActorExecutorClientMessage::Attach {
203 protocol,
204 actor_types,
205 }) => {
206 ensure!(
207 protocol == ACTOR_EXECUTOR_PROTOCOL_VERSION,
208 "customer actor executor uses unsupported protocol version {protocol}"
209 );
210 ensure!(
211 !actor_types.is_empty(),
212 "customer actor executor did not advertise any actor types"
213 );
214 actor_types
215 }
216 Some(_) => {
217 anyhow::bail!("first customer actor executor message must attach the process")
218 }
219 None => anyhow::bail!("customer actor executor disconnected before attaching"),
220 };
221
222 let executor = Arc::new(JsActorExecutor::new(writer, attach));
223 let task = tokio::spawn(read_executor_messages(reader, executor.clone()));
224 debug!(
225 socket = %self.socket_path.display(),
226 actor_types = ?executor.actor_types,
227 "customer JavaScript process connected to actor executor"
228 );
229 Ok(ActorExecutorConnection { executor, task })
230 }
231}
232
233pub(crate) struct ActorExecutorConnection {
234 executor: Arc<JsActorExecutor>,
235 task: JoinHandle<Result<()>>,
236}
237
238impl ActorExecutorConnection {
239 pub(crate) fn executor(&self) -> Arc<dyn ActorExecutor> {
240 self.executor.clone()
241 }
242
243 pub(crate) async fn mark_ready(&self) -> Result<()> {
244 self.executor
245 .send(&ActorExecutorServerMessage::Attached {
246 protocol: ACTOR_EXECUTOR_PROTOCOL_VERSION,
247 })
248 .await?;
249 info!(
250 actor_types = ?self.executor.actor_types,
251 "customer JavaScript process attached to actor executor"
252 );
253 Ok(())
254 }
255
256 pub(crate) async fn run(mut self, shutdown: CancellationToken) -> Result<()> {
257 tokio::select! {
258 result = &mut self.task => {
259 match result {
260 Ok(result) => result,
261 Err(error) => Err(error.into()),
262 }
263 }
264 _ = shutdown.cancelled() => {
265 self.executor.close().await;
266 self.task.abort();
267 let _ = (&mut self.task).await;
268 Ok(())
269 }
270 }
271 }
272}
273
274impl Drop for ActorExecutorConnection {
275 fn drop(&mut self) {
276 self.task.abort();
277 }
278}
279
280struct JsActorExecutor {
281 actor_types: HashSet<String>,
282 next_message_id: AtomicU64,
283 pending: Mutex<HashMap<u64, oneshot::Sender<ExecutorReply>>>,
284 writer: AsyncMutex<OwnedWriteHalf>,
285 residents: Mutex<HashSet<ActorKey>>,
286}
287
288#[async_trait]
289impl ActorExecutor for JsActorExecutor {
290 fn supports(&self, actor_type: &str) -> bool {
291 self.actor_types.contains(actor_type)
292 }
293
294 async fn invoke(
295 &self,
296 invocation: ActorMethodInvocation,
297 state: Option<&Value>,
298 ) -> Result<ActorMethodOutcome> {
299 match self
300 .exchange_with_state(ExecutorCommand::Invoke(invocation), state)
301 .await?
302 {
303 ExecutorReply::Invoked {
304 result,
305 state,
306 effects,
307 } => Ok(ActorMethodOutcome::Completed {
308 result,
309 state,
310 effects,
311 }),
312 ExecutorReply::Failed { code, message } => {
313 Ok(ActorMethodOutcome::Failed(ActorInvocationFailure {
314 code,
315 message,
316 }))
317 }
318 ExecutorReply::Evicted | ExecutorReply::StateRequired => {
319 anyhow::bail!("actor executor returned eviction reply to invocation")
320 }
321 ExecutorReply::WebsocketHandled { .. } => {
322 anyhow::bail!("actor executor returned socket reply to invocation")
323 }
324 }
325 }
326
327 async fn handle_socket(
328 &self,
329 invocation: ActorSocketInvocation,
330 state: Option<&Value>,
331 ) -> Result<ActorSocketOutcome> {
332 match self
333 .exchange_with_state(ExecutorCommand::WebsocketEvent(invocation), state)
334 .await?
335 {
336 ExecutorReply::WebsocketHandled { state, effects } => {
337 Ok(ActorSocketOutcome::Handled { state, effects })
338 }
339 ExecutorReply::Failed { code, message } => {
340 Ok(ActorSocketOutcome::Failed(ActorInvocationFailure {
341 code,
342 message,
343 }))
344 }
345 ExecutorReply::Invoked { .. }
346 | ExecutorReply::Evicted
347 | ExecutorReply::StateRequired => {
348 anyhow::bail!("actor executor returned the wrong reply to socket event")
349 }
350 }
351 }
352
353 async fn evict(&self, eviction: ActorMethodEviction) -> Result<()> {
354 self.residents
355 .lock()
356 .map_err(|_| anyhow::anyhow!("actor residency lock poisoned"))?
357 .remove(&eviction.actor);
358 match self
359 .exchange(&ExecutorCommand::Evict(eviction), None, false)
360 .await?
361 {
362 ExecutorReply::Evicted => Ok(()),
363 ExecutorReply::Failed { code, message } => {
364 anyhow::bail!("actor executor rejected eviction ({code}): {message}")
365 }
366 ExecutorReply::Invoked { .. } => {
367 anyhow::bail!("actor executor returned the wrong reply to eviction")
368 }
369 ExecutorReply::WebsocketHandled { .. } | ExecutorReply::StateRequired => {
370 anyhow::bail!("actor executor returned socket reply to eviction")
371 }
372 }
373 }
374}
375
376impl JsActorExecutor {
377 fn new(writer: OwnedWriteHalf, actor_types: Vec<String>) -> Self {
378 Self {
379 actor_types: actor_types.into_iter().collect(),
380 next_message_id: AtomicU64::new(1),
381 pending: Mutex::new(HashMap::new()),
382 writer: AsyncMutex::new(writer),
383 residents: Mutex::new(HashSet::new()),
384 }
385 }
386
387 async fn exchange_with_state(
388 &self,
389 command: ExecutorCommand,
390 state: Option<&Value>,
391 ) -> Result<ExecutorReply> {
392 let actor = match &command {
393 ExecutorCommand::Invoke(invocation) => &invocation.actor,
394 ExecutorCommand::WebsocketEvent(invocation) => &invocation.actor,
395 ExecutorCommand::Evict(_) => unreachable!("eviction does not hydrate"),
396 };
397 let resident = self
398 .residents
399 .lock()
400 .map_err(|_| anyhow::anyhow!("actor residency lock poisoned"))?
401 .remove(actor);
402 let mut reply = self
403 .exchange(
404 &command,
405 if resident {
406 None
407 } else {
408 Some(state.unwrap_or(&Value::Null))
409 },
410 resident,
411 )
412 .await?;
413 if resident && matches!(reply, ExecutorReply::StateRequired) {
414 reply = self
415 .exchange(&command, Some(state.unwrap_or(&Value::Null)), false)
416 .await?;
417 }
418 ensure!(
419 !matches!(reply, ExecutorReply::StateRequired),
420 "actor executor refused explicit hydration"
421 );
422 if matches!(
423 reply,
424 ExecutorReply::Invoked { .. } | ExecutorReply::WebsocketHandled { .. }
425 ) {
426 let mut residents = self
427 .residents
428 .lock()
429 .map_err(|_| anyhow::anyhow!("actor residency lock poisoned"))?;
430 if residents.len() >= 4096 {
431 residents.clear();
432 }
433 residents.insert(actor.clone());
434 }
435 Ok(reply)
436 }
437
438 async fn exchange(
439 &self,
440 command: &ExecutorCommand,
441 state: Option<&Value>,
442 resident_only: bool,
443 ) -> Result<ExecutorReply> {
444 let message_id = self.next_message_id.fetch_add(1, Ordering::Relaxed);
445 let (reply_tx, reply_rx) = oneshot::channel();
446 self.pending
447 .lock()
448 .map_err(|_| anyhow::anyhow!("actor executor pending-reply lock poisoned"))?
449 .insert(message_id, reply_tx);
450
451 let write_result = self
452 .send(&ActorExecutorServerMessage::Command {
453 message_id,
454 command: ExecutorCommandEnvelope {
455 command,
456 state,
457 resident_only,
458 },
459 })
460 .await;
461 if let Err(error) = write_result {
462 self.remove_pending(message_id)?;
463 if let Some(limit) = error.downcast_ref::<ActorExecutorMessageTooLarge>() {
464 return Ok(ExecutorReply::Failed {
465 code: "resource_exhausted".into(),
466 message: limit.to_string(),
467 });
468 }
469 return Err(error.context("send command to customer actor executor"));
470 }
471
472 reply_rx
473 .await
474 .context("customer actor executor disconnected before replying")
475 }
476
477 async fn send(&self, message: &ActorExecutorServerMessage<'_>) -> Result<()> {
478 write_server_message(&mut *self.writer.lock().await, message).await
479 }
480
481 fn deliver(&self, message_id: u64, reply: ExecutorReply) -> Result<()> {
482 let sender = self
483 .pending
484 .lock()
485 .map_err(|_| anyhow::anyhow!("actor executor pending-reply lock poisoned"))?
486 .remove(&message_id)
487 .with_context(|| format!("actor executor replied to unknown message {message_id}"))?;
488 let _ = sender.send(reply);
489 Ok(())
490 }
491
492 fn remove_pending(&self, message_id: u64) -> Result<()> {
493 self.pending
494 .lock()
495 .map_err(|_| anyhow::anyhow!("actor executor pending-reply lock poisoned"))?
496 .remove(&message_id);
497 Ok(())
498 }
499
500 async fn close(&self) {
501 let _ = self.writer.lock().await.shutdown().await;
502 self.disconnect();
503 }
504
505 fn disconnect(&self) {
506 if let Ok(mut pending) = self.pending.lock() {
507 pending.clear();
508 }
509 }
510}
511
512async fn read_executor_messages(
513 mut reader: BufReader<tokio::net::unix::OwnedReadHalf>,
514 executor: Arc<JsActorExecutor>,
515) -> Result<()> {
516 let result = async {
517 loop {
518 match read_client_message(&mut reader).await? {
519 None => {
520 anyhow::bail!("customer JavaScript actor executor disconnected")
521 }
522 Some(ActorExecutorClientMessage::Reply { message_id, reply }) => {
523 executor.deliver(message_id, reply)?;
524 }
525 Some(ActorExecutorClientMessage::Attach { .. }) => {
526 anyhow::bail!("customer actor executor attached more than once")
527 }
528 }
529 }
530 }
531 .await;
532 executor.disconnect();
533 result
534}
535
536async fn read_client_message(
537 reader: &mut BufReader<tokio::net::unix::OwnedReadHalf>,
538) -> Result<Option<ActorExecutorClientMessage>> {
539 let mut document = Vec::new();
540 let bytes = reader
541 .take((MAX_ACTOR_EXECUTOR_MESSAGE_BYTES + 1) as u64)
542 .read_until(b'\n', &mut document)
543 .await?;
544 if bytes == 0 {
545 return Ok(None);
546 }
547 ensure!(
548 bytes <= MAX_ACTOR_EXECUTOR_MESSAGE_BYTES,
549 "customer actor executor message exceeds {MAX_ACTOR_EXECUTOR_MESSAGE_BYTES} bytes"
550 );
551 serde_json::from_slice(trim_ascii_end(&document))
552 .map(Some)
553 .context("decode customer actor executor message")
554}
555
556fn trim_ascii_end(mut document: &[u8]) -> &[u8] {
557 while document.last().is_some_and(u8::is_ascii_whitespace) {
558 document = &document[..document.len() - 1];
559 }
560 document
561}
562
563async fn write_server_message(
564 writer: &mut OwnedWriteHalf,
565 message: &ActorExecutorServerMessage<'_>,
566) -> Result<()> {
567 let mut bytes = serde_json::to_vec(message)?;
568 bytes.push(b'\n');
569 if bytes.len() > MAX_ACTOR_EXECUTOR_MESSAGE_BYTES {
570 return Err(ActorExecutorMessageTooLarge.into());
571 }
572 writer.write_all(&bytes).await?;
573 Ok(())
574}
575
576#[derive(Debug)]
577struct ActorExecutorMessageTooLarge;
578
579impl Display for ActorExecutorMessageTooLarge {
580 fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
581 write!(
582 formatter,
583 "actor executor command exceeds {MAX_ACTOR_EXECUTOR_MESSAGE_BYTES} bytes"
584 )
585 }
586}
587
588impl Error for ActorExecutorMessageTooLarge {}
589
590async fn prepare_socket_path(path: &Path) -> Result<()> {
591 match tokio::fs::symlink_metadata(path).await {
592 Ok(metadata) => {
593 ensure!(
594 metadata.file_type().is_socket(),
595 "refusing to replace non-socket actor executor path {}",
596 path.display()
597 );
598 tokio::fs::remove_file(path).await?;
599 }
600 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
601 Err(error) => return Err(error.into()),
602 }
603 Ok(())
604}
605
606async fn remove_socket(path: &Path) -> Result<()> {
607 match tokio::fs::remove_file(path).await {
608 Ok(()) => {
609 debug!(socket = %path.display(), "actor executor socket removed");
610 Ok(())
611 }
612 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
613 Err(error) => Err(error.into()),
614 }
615}
616
617#[derive(Debug, Serialize)]
618#[serde(tag = "type", rename_all = "snake_case")]
619enum ActorExecutorServerMessage<'a> {
620 Attached {
621 protocol: u32,
622 },
623 Command {
624 message_id: u64,
625 command: ExecutorCommandEnvelope<'a>,
626 },
627}
628
629#[derive(Debug, Serialize)]
630struct ExecutorCommandEnvelope<'a> {
631 #[serde(flatten)]
632 command: &'a ExecutorCommand,
633 #[serde(skip_serializing_if = "Option::is_none")]
634 state: Option<&'a Value>,
635 resident_only: bool,
636}
637
638#[derive(Debug, Deserialize)]
639#[serde(tag = "type", rename_all = "snake_case")]
640enum ActorExecutorClientMessage {
641 Attach {
642 protocol: u32,
643 actor_types: Vec<String>,
644 },
645 Reply {
646 message_id: u64,
647 reply: ExecutorReply,
648 },
649}
650
651#[derive(Debug, Serialize)]
652#[serde(tag = "type", rename_all = "snake_case")]
653enum ExecutorCommand {
654 Invoke(ActorMethodInvocation),
655 WebsocketEvent(ActorSocketInvocation),
656 Evict(ActorMethodEviction),
657}
658
659#[derive(Debug, Deserialize)]
660#[serde(tag = "type", rename_all = "snake_case")]
661enum ExecutorReply {
662 StateRequired,
663 Invoked {
664 result: Value,
665 state: Value,
666 #[serde(default)]
667 effects: Vec<ActorSocketEffect>,
668 },
669 WebsocketHandled {
670 state: Value,
671 effects: Vec<ActorSocketEffect>,
672 },
673 Failed {
674 code: String,
675 message: String,
676 },
677 Evicted,
678}
679
680#[cfg(test)]
681mod tests {
682 use super::*;
683 use serde_json::json;
684 use tempfile::TempDir;
685 use tokio::{
686 io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
687 net::UnixStream,
688 time::{Duration, timeout},
689 };
690
691 #[tokio::test]
692 async fn one_javascript_executor_runs_until_host_shutdown() -> Result<()> {
693 let root = TempDir::new_in("/tmp")?;
694 let socket = root.path().join("actor-executor.sock");
695 let host = ActorExecutorListener::bind(&socket).await?;
696 let customer = tokio::spawn(run_incrementing_customer(socket.clone()));
697 let connection = host.accept().await?;
698 let executor = connection.executor();
699 connection.mark_ready().await?;
700 assert!(executor.supports("counter"));
701
702 let shutdown = CancellationToken::new();
703 let connection_task = tokio::spawn(connection.run(shutdown.clone()));
704 let outcome = executor
705 .invoke(
706 ActorMethodInvocation {
707 request_id: "request-1".into(),
708 actor: ActorKey {
709 namespace_id: "namespace-1".into(),
710 actor_type: "counter".into(),
711 actor_id: "counter-1".into(),
712 },
713 method: "increment".into(),
714 args: vec![json!(2)],
715 connections: Vec::new(),
716 },
717 None,
718 )
719 .await?;
720 assert_eq!(
721 outcome,
722 ActorMethodOutcome::Completed {
723 result: json!(2),
724 state: json!({ "count": 2 }),
725 effects: Vec::new(),
726 }
727 );
728 let socket_outcome = executor
729 .handle_socket(
730 ActorSocketInvocation {
731 request_id: "socket-request-1".into(),
732 actor: ActorKey {
733 namespace_id: "namespace-1".into(),
734 actor_type: "counter".into(),
735 actor_id: "counter-1".into(),
736 },
737 event: ActorSocketEvent::Connect {
738 connection: ActorSocketConnection {
739 id: "socket-1".into(),
740 metadata: json!({ "userId": "user-1" }),
741 tags: Vec::new(),
742 },
743 },
744 connections: vec![ActorSocketConnection {
745 id: "socket-1".into(),
746 metadata: json!({ "userId": "user-1" }),
747 tags: Vec::new(),
748 }],
749 },
750 Some(&json!({ "count": 2 })),
751 )
752 .await?;
753 assert_eq!(
754 socket_outcome,
755 ActorSocketOutcome::Handled {
756 state: json!({ "count": 3 }),
757 effects: vec![ActorSocketEffect::Send {
758 connection_id: "socket-1".into(),
759 message: ActorSocketMessage::Text {
760 data: "ready".into()
761 },
762 }],
763 }
764 );
765 shutdown.cancel();
766 connection_task.await??;
767 customer.await??;
768 Ok(())
769 }
770
771 #[tokio::test]
772 async fn resident_commands_omit_state_and_retry_only_an_explicit_hydration_request()
773 -> Result<()> {
774 let (host, customer) = UnixStream::pair()?;
775 let (_, writer) = host.into_split();
776 let executor = Arc::new(JsActorExecutor::new(writer, vec!["counter".into()]));
777 let mut reader = BufReader::new(customer);
778 let customer = async {
779 let first = read_json_line(&mut reader).await?;
780 assert_eq!(first["command"]["state"], json!({"count": 9}));
781 executor.deliver(
782 first["message_id"].as_u64().unwrap(),
783 serde_json::from_value(
784 json!({"type":"invoked", "result":10,"state":{"count":10}}),
785 )?,
786 )?;
787 let warm = read_json_line(&mut reader).await?;
788 assert!(warm["command"].get("state").is_none());
789 assert_eq!(warm["command"]["resident_only"], true);
790 executor.deliver(
791 warm["message_id"].as_u64().unwrap(),
792 serde_json::from_value(json!({"type":"state_required"}))?,
793 )?;
794 let retry = read_json_line(&mut reader).await?;
795 assert_eq!(
796 retry["command"]["request_id"],
797 warm["command"]["request_id"]
798 );
799 assert_eq!(retry["command"]["state"], json!({"count": 10}));
800 assert_eq!(retry["command"]["resident_only"], false);
801 executor.deliver(
802 retry["message_id"].as_u64().unwrap(),
803 serde_json::from_value(
804 json!({"type":"invoked", "result":11,"state":{"count":11}}),
805 )?,
806 )?;
807 anyhow::Ok(())
808 };
809 let invoke = async {
810 for count in [9, 10] {
811 let outcome = executor
812 .invoke(
813 ActorMethodInvocation {
814 request_id: format!("request-{count}"),
815 actor: ActorKey {
816 namespace_id: "test".into(),
817 actor_type: "counter".into(),
818 actor_id: "one".into(),
819 },
820 method: "increment".into(),
821 args: vec![],
822 connections: vec![],
823 },
824 Some(&json!({"count":count})),
825 )
826 .await?;
827 assert!(
828 matches!(outcome, ActorMethodOutcome::Completed {result, ..} if result == json!(count + 1))
829 );
830 }
831 anyhow::Ok(())
832 };
833 tokio::try_join!(customer, invoke)?;
834 Ok(())
835 }
836
837 #[tokio::test]
838 async fn oversized_commands_are_reported_as_resource_exhausted() -> Result<()> {
839 let root = TempDir::new_in("/tmp")?;
840 let socket = root.path().join("actor-executor.sock");
841 let host = ActorExecutorListener::bind(&socket).await?;
842 let customer = tokio::spawn(run_attached_customer(socket.clone()));
843 let connection = host.accept().await?;
844 let executor = connection.executor();
845 connection.mark_ready().await?;
846
847 let shutdown = CancellationToken::new();
848 let connection_task = tokio::spawn(connection.run(shutdown.clone()));
849 let outcome = executor
850 .invoke(
851 ActorMethodInvocation {
852 request_id: "request-1".into(),
853 actor: ActorKey {
854 namespace_id: "namespace-1".into(),
855 actor_type: "counter".into(),
856 actor_id: "counter-1".into(),
857 },
858 method: "accept".into(),
859 args: vec![json!("x".repeat(MAX_ACTOR_EXECUTOR_MESSAGE_BYTES))],
860 connections: Vec::new(),
861 },
862 None,
863 )
864 .await?;
865
866 assert!(matches!(
867 outcome,
868 ActorMethodOutcome::Failed(ref failure) if failure.code == "resource_exhausted"
869 ));
870 shutdown.cancel();
871 connection_task.await??;
872 customer.await??;
873 Ok(())
874 }
875
876 #[tokio::test]
877 async fn oversized_client_messages_are_rejected_before_newline() -> Result<()> {
878 let (host, mut customer) = UnixStream::pair()?;
879 let (reader, _) = host.into_split();
880 let mut reader = BufReader::new(reader);
881 let customer = tokio::spawn(async move {
882 let chunk = vec![b'x'; 64 * 1024];
883 for _ in 0..=MAX_ACTOR_EXECUTOR_MESSAGE_BYTES / chunk.len() {
884 customer.write_all(&chunk).await?;
885 }
886 std::future::pending::<()>().await;
887 #[allow(unreachable_code)]
888 Ok::<(), anyhow::Error>(())
889 });
890
891 let result = timeout(Duration::from_secs(5), read_client_message(&mut reader)).await;
892 customer.abort();
893 let error = result
894 .context("oversized actor executor message was not rejected before newline")?
895 .expect_err("oversized actor executor message should fail");
896 assert!(error.to_string().contains("exceeds"));
897 Ok(())
898 }
899
900 async fn run_incrementing_customer(socket: PathBuf) -> Result<()> {
901 let stream = UnixStream::connect(socket).await?;
902 let (reader, mut writer) = stream.into_split();
903 let mut reader = BufReader::new(reader);
904 writer
905 .write_all(b"{\"type\":\"attach\",\"protocol\":13,\"actor_types\":[\"counter\"]}\n")
906 .await?;
907 ensure!(
908 read_json_line(&mut reader).await? == json!({ "type": "attached", "protocol": 13 })
909 );
910
911 let invocation = read_json_line(&mut reader).await?;
912 let invocation_id = invocation["message_id"]
913 .as_u64()
914 .context("invocation message ID")?;
915 ensure!(invocation["command"]["type"] == "invoke");
916 ensure!(invocation["command"].get("timeout_ms").is_none());
917 write_json_line(
918 &mut writer,
919 &json!({
920 "type": "reply",
921 "message_id": invocation_id,
922 "reply": {
923 "type": "invoked",
924 "result": 2,
925 "state": { "count": 2 }
926 }
927 }),
928 )
929 .await?;
930
931 let socket_event = read_json_line(&mut reader).await?;
932 let socket_event_id = socket_event["message_id"]
933 .as_u64()
934 .context("socket event message ID")?;
935 ensure!(socket_event["command"]["type"] == "websocket_event");
936 ensure!(socket_event["command"]["event"]["type"] == "connect");
937 write_json_line(
938 &mut writer,
939 &json!({
940 "type": "reply",
941 "message_id": socket_event_id,
942 "reply": {
943 "type": "websocket_handled",
944 "state": { "count": 3 },
945 "effects": [{
946 "type": "send",
947 "connection_id": "socket-1",
948 "message": { "type": "text", "data": "ready" }
949 }]
950 }
951 }),
952 )
953 .await?;
954
955 let mut trailing = String::new();
956 ensure!(
957 reader.read_line(&mut trailing).await? == 0,
958 "expected Rust host to close the actor executor"
959 );
960 Ok(())
961 }
962
963 async fn run_attached_customer(socket: PathBuf) -> Result<()> {
964 let stream = UnixStream::connect(socket).await?;
965 let (reader, mut writer) = stream.into_split();
966 let mut reader = BufReader::new(reader);
967 writer
968 .write_all(b"{\"type\":\"attach\",\"protocol\":13,\"actor_types\":[\"counter\"]}\n")
969 .await?;
970 ensure!(
971 read_json_line(&mut reader).await? == json!({ "type": "attached", "protocol": 13 })
972 );
973 let mut trailing = String::new();
974 ensure!(
975 reader.read_line(&mut trailing).await? == 0,
976 "oversized command reached the customer actor executor"
977 );
978 Ok(())
979 }
980
981 async fn read_json_line<R>(reader: &mut R) -> Result<Value>
982 where
983 R: tokio::io::AsyncBufRead + Unpin,
984 {
985 let mut line = String::new();
986 ensure!(reader.read_line(&mut line).await? > 0, "expected JSON line");
987 Ok(serde_json::from_str(line.trim_end())?)
988 }
989
990 async fn write_json_line<W>(writer: &mut W, value: &Value) -> Result<()>
991 where
992 W: tokio::io::AsyncWrite + Unpin,
993 {
994 writer
995 .write_all(serde_json::to_string(value)?.as_bytes())
996 .await?;
997 writer.write_all(b"\n").await?;
998 Ok(())
999 }
1000}