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