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