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