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