fix: server tx dies after getting

AWeirdDev · AWeirdDev · commit d65c158ddb9f · 2026-05-23T19:23:22.000+08:00
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/svld-binding-kv/src/main.rs b/crates/svld-binding-kv/src/main.rs
@@ -67,6 +67,8 @@ async fn main() -> Result<(), Box<dyn core::error::Error>> {
             }
         };
 
+        tracing::info!("worker {:?}: {}(...)", worker_name, function_name);
+
         let tree = match db.open_tree(&worker_name) {
             Ok(t) => t,
             Err(e) => {
@@ -122,14 +124,15 @@ async fn main() -> Result<(), Box<dyn core::error::Error>> {
 
         match &payload {
             KvPayload::Get { key } => {
+                tracing::info!("sending...");
                 match tree.get(&key) {
                     Ok(Some(data)) => {
                         let s = String::from_utf8_lossy(&data);
                         client.send_ok(id, s).await?;
                     }
 
                     Ok(None) => {
-                        client.send_ok(id, payload).await?;
+                        client.send_ok(id, ijson::ijson!(null)).await?;
                     }
 
                     Err(e) => {
diff --git a/crates/svld-ipc-client/Cargo.toml b/crates/svld-ipc-client/Cargo.toml
@@ -12,3 +12,4 @@ serde = "1.0.228"
 serde_json = "1.0.150"
 thiserror = "2.0.18"
 tokio = { version = "1.52.3" }
+tracing = "0.1.44"
diff --git a/crates/svld-ipc-client/src/core.rs b/crates/svld-ipc-client/src/core.rs
@@ -4,12 +4,7 @@
 //! let mut client = BindingClient::connect(PathBuf::from(".serverlessd/bindings.sock")).await?;
 //! ```
 
-use std::{
-    io::{self, IoSlice},
-    marker::PhantomData,
-    mem,
-    path::PathBuf,
-};
+use std::{io, marker::PhantomData, mem, path::PathBuf};
 
 use interprocess::local_socket::{
     ConnectOptions, GenericFilePath,
@@ -75,21 +70,17 @@ impl BindingClient<Initialized> {
         &mut self,
         message: ClientMessage<Result<T, String>>,
     ) -> Result<(), BindingClientError> {
-        let id_raw = message.id.to_le_bytes();
         let payload_raw = serde_json::to_vec(&match message.payload {
             Ok(t) => serde_json::json!({"data": &t}),
             Err(e) => serde_json::json!({"error": &e}),
         })?;
 
-        let len_raw = (payload_raw.len() as u32).to_le_bytes();
-
-        let slices = [
-            IoSlice::new(&id_raw),
-            IoSlice::new(&len_raw),
-            IoSlice::new(&payload_raw),
-        ];
+        let mut buf = Vec::with_capacity(size_of::<u32>() * 2 + payload_raw.len());
+        buf.extend_from_slice(&message.id.to_le_bytes());
+        buf.extend_from_slice(&(payload_raw.len() as u32).to_le_bytes());
+        buf.extend_from_slice(&payload_raw);
 
-        self.send.write_vectored(&slices).await?;
+        self.send.write_all(&buf).await?;
 
         Ok(())
     }
@@ -129,14 +120,19 @@ impl BindingClient<Initialized> {
         &mut self,
     ) -> Result<ServerMessage<T>, BindingClientError> {
         let id = self.recv.read_u32_le().await?;
+        tracing::info!("got id={id}");
 
         let data_len = self.recv.read_u32_le().await? as usize;
         let mut data = Box::<[u8]>::new_uninit_slice(data_len);
         let slice =
             unsafe { std::slice::from_raw_parts_mut(data.as_mut_ptr() as *mut u8, data_len) };
         self.recv.read_exact(slice).await?;
 
-        let UnprocessedServerMessage { func, data, worker } = serde_json::from_slice(slice)?;
+        let UnprocessedServerMessage {
+            func,
+            args: data,
+            worker,
+        } = serde_json::from_slice(slice)?;
 
         Ok(ServerMessage {
             id,
@@ -145,6 +141,8 @@ impl BindingClient<Initialized> {
             args: ijson::from_value::<T>(&data)?,
         })
     }
+
+    // pub async fn shutdown(&mut self) {}
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -173,7 +171,7 @@ pub struct ClientMessage<T> {
 #[derive(Deserialize)]
 struct UnprocessedServerMessage {
     func: String,
-    data: ijson::IValue,
+    args: ijson::IValue,
     worker: String,
 }
 
diff --git a/crates/svld-rt/src/bindings/backend.rs b/crates/svld-rt/src/bindings/backend.rs
@@ -28,7 +28,7 @@ pub trait BindingBackend: Send + Sync + 'static {
     fn get_tx(&self) -> BindingBackendTx;
 
     /// Creates a client from the env name for the worker.
-    fn create_client(&self, binding_name: &str) -> Box<dyn BindingClient>;
+    fn create_client(&self) -> Box<dyn BindingClient>;
 }
 
 /// Creates a binding backend channel.
diff --git a/crates/svld-rt/src/bindings/ipc/core.rs b/crates/svld-rt/src/bindings/ipc/core.rs
@@ -30,7 +30,7 @@ impl IpcBindingsServer {
         let binding_backends = binding_types
             .into_iter()
             .map(|type_| {
-                let backend = Arc::new(binding_backend::IpcBindingBackend::new());
+                let backend = Arc::new(binding_backend::IpcBindingBackend::new(type_.clone()));
                 binding_store.push_binding(&type_, backend.clone());
 
                 (type_, backend)
@@ -73,6 +73,7 @@ pub mod binding_backend {
     use crate::bindings::{BindingBackend, BindingBackendTx, backend::BindingClient};
 
     pub struct IpcBindingBackend {
+        type_: String,
         maybe_tx: (
             AtomicBool,  // has data?
             AtomicUsize, // BindingBackendTx
@@ -86,8 +87,9 @@ pub mod binding_backend {
 
     impl IpcBindingBackend {
         #[inline(always)]
-        pub const fn new() -> Self {
+        pub fn new(type_: String) -> Self {
             Self {
+                type_,
                 maybe_tx: (AtomicBool::new(false), AtomicUsize::new(0)),
             }
         }
@@ -137,7 +139,10 @@ pub mod binding_backend {
 
             // SAFETY: we asserted BindingBackendTx is usize-sized;
             // the bool guard ensures this was previously set via set_tx.
-            Some(unsafe { mem::transmute::<usize, BindingBackendTx>(raw) })
+            Some(unsafe {
+                let tx = mem::ManuallyDrop::new(mem::transmute::<usize, BindingBackendTx>(raw));
+                (*tx).clone()
+            })
         }
 
         #[inline(always)]
@@ -175,8 +180,8 @@ pub mod binding_backend {
         }
 
         #[inline]
-        fn create_client(&self, binding_name: &str) -> Box<dyn BindingClient> {
-            Box::new(IpcBindingClient::new(binding_name.to_string()))
+        fn create_client(&self) -> Box<dyn BindingClient> {
+            Box::new(IpcBindingClient::new(self.type_.clone()))
         }
     }
 
@@ -204,13 +209,13 @@ mod binding_client {
     };
 
     pub struct IpcBindingClient {
-        binding_name: String,
+        binding_type: String,
     }
 
     impl IpcBindingClient {
         #[inline(always)]
-        pub fn new(binding_name: String) -> Self {
-            Self { binding_name }
+        pub fn new(binding_type: String) -> Self {
+            Self { binding_type }
         }
     }
 
@@ -232,7 +237,7 @@ mod binding_client {
                               mut rv: v8::ReturnValue| {
                                  let arr = args.data().cast::<v8::Array>();
 
-                                 let binding_name = arr
+                                 let binding_type = arr
                                      .get_index(scope, 0)
                                      .unwrap()
                                      .cast::<v8::String>()
@@ -244,7 +249,7 @@ mod binding_client {
                                      to_rust_string_lossy(scope);
 
                                  let state = WorkerState::get_from_isolate(scope);
-                                 let tx = state.get_binding(&binding_name).unwrap();
+                                 let tx = state.get_binding_tx(&binding_type).unwrap();
 
                                  let args_len = args.length();
                                  let arr = v8::Array::new(scope, args_len);
@@ -331,7 +336,7 @@ mod binding_client {
                     rv.set(fnk.cast());
                 },
             )
-            .data(v8::String::new(scope, &self.binding_name)?.cast())
+            .data(v8::String::new(scope, &self.binding_type)?.cast())
             .build(scope)?;
 
             handler.set(scope, v8::String::new(scope, "get")?.cast(), get_fn.cast());
@@ -348,7 +353,7 @@ mod binding_client {
 mod task {
     use std::{
         collections::HashMap,
-        io::{self, IoSlice},
+        io,
         string::FromUtf8Error,
         sync::{
             Arc,
@@ -432,6 +437,28 @@ mod task {
         let (tx, mut rx) = binding_backend_channel();
         backend.set_tx(tx)?;
 
+        // spawn dedicated reader so it's never cancelled
+        let (external_tx, mut external_rx) = tokio::sync::mpsc::channel::<Message>(32);
+
+        tokio::spawn(async move {
+            loop {
+                match recv.read_parse_message().await {
+                    Ok(msg) => {
+                        if external_tx.send(msg).await.is_err() {
+                            tracing::error!("errored on sending external tx");
+                            break; // main task dropped, exit
+                        }
+                    }
+                    Err(e) => {
+                        tracing::error!(
+                            "failed to read & parse binding client message, breaking: {e:?}"
+                        );
+                        break;
+                    }
+                }
+            }
+        });
+
         let mut resolutions = HashMap::new();
         let mut roll_id = 0_u32;
 
@@ -445,22 +472,20 @@ mod task {
             }
 
             let event = tokio::select! {
-                msg = rx.recv() => {
-                    match msg {
-                        Some(t) => Event::Internal(t),
-                        None => break,
+                msg = rx.recv() => match msg {
+                    Some(t) => Event::Internal(t),
+                    None => {
+                        tracing::error!("internal channel closed: backend dropped?");
+                        break;
                     }
                 },
-
-                msg = recv.read_parse_message() => {
-                    match msg {
-                        Ok(t) => Event::External(t),
-                        Err(e) => {
-                            tracing::error!("failed to read & parse binding sclient message, breaking: {e:?}");
-                            break;
-                        }
+                msg = external_rx.recv() => match msg {
+                    Some(t) => Event::External(t),
+                    None => {
+                        tracing::error!("reader task died");
+                        break;
                     }
-                }
+                },
             };
 
             match event {
@@ -480,6 +505,7 @@ mod task {
                 }
 
                 Event::External(Message { id, payload }) => {
+                    tracing::debug!("server finished reading payload for id={id}");
                     if let Some(replier) = resolutions.remove(&id) {
                         let _ = replier.send(payload);
                     }
@@ -536,6 +562,7 @@ mod task {
             // get this over with, and we're just reading from them
 
             let len = self.read_u32_le().await? as usize;
+
             let mut buf = Box::<[u8]>::new_uninit_slice(len);
 
             let slice = unsafe { std::slice::from_raw_parts_mut(buf.as_mut_ptr() as *mut u8, len) };
@@ -557,6 +584,7 @@ mod task {
 
             // then we'll get the payload
             let raw_payload = self.read_parse_to_boxed_arr().await?;
+
             let payload = serde_json::from_slice::<ijson::IValue>(&raw_payload)?;
 
             // good. fuck you and eat it up
@@ -572,17 +600,15 @@ mod task {
     #[async_trait]
     impl SendExt for SendHalf {
         async fn send_message(&mut self, message: Message) -> Result<(), SingleTaskError> {
-            let id_raw = message.id.to_le_bytes();
-
             let payload_raw = serde_json::to_vec(&message.payload)?;
-            let len_raw = (payload_raw.len() as u32).to_le_bytes();
-
-            let slices = [
-                IoSlice::new(&id_raw),
-                IoSlice::new(&len_raw),
-                IoSlice::new(&payload_raw),
-            ];
-            self.write_vectored(&slices).await?;
+
+            let mut buf = Vec::with_capacity(size_of::<u32>() * 2 + payload_raw.len());
+
+            buf.extend_from_slice(&message.id.to_le_bytes());
+            buf.extend_from_slice(&(payload_raw.len() as u32).to_le_bytes());
+            buf.extend_from_slice(&payload_raw);
+
+            self.write_all(&buf).await?;
 
             // fah
             Ok(())
diff --git a/crates/svld-rt/src/bindings/store.rs b/crates/svld-rt/src/bindings/store.rs
@@ -106,8 +106,8 @@ pub struct BindingItem {
 
 impl BindingItem {
     #[inline(always)]
-    pub fn create_client(&self, name: &str) -> Box<dyn BindingClient> {
-        self.backend.create_client(name)
+    pub fn create_client(&self) -> Box<dyn BindingClient> {
+        self.backend.create_client()
     }
 
     #[inline(always)]
diff --git a/crates/svld-rt/src/env.rs b/crates/svld-rt/src/env.rs
@@ -16,7 +16,7 @@ pub fn create_js_env<'s>(
         tracing::info!("js env got name={name:?}, type={type_:?}");
         let binding = state.binding_store.get_binding(&type_)?;
 
-        let client = binding.create_client(&name);
+        let client = binding.create_client();
 
         let interface = client.create_interface(scope)?;
 
diff --git a/crates/svld-rt/src/worker/state.rs b/crates/svld-rt/src/worker/state.rs
@@ -201,7 +201,7 @@ impl WorkerState {
     }
 
     #[inline(always)]
-    pub fn get_binding(&self, name: &str) -> Option<BindingBackendTx> {
-        self.binding_store.get_binding_tx(name)
+    pub fn get_binding_tx(&self, type_: &str) -> Option<BindingBackendTx> {
+        self.binding_store.get_binding_tx(type_)
     }
 }
diff --git a/crates/svld-rt/src/worker/task.rs b/crates/svld-rt/src/worker/task.rs
@@ -295,7 +295,6 @@ async fn create_task(rx: &mut WorkerRx, args: InitWorkerArgs<'_>) -> Result<bool
             tracing::info!("resolving event loop");
 
             let isolate = unsafe { state.get_isolate() };
-            tracing::info!("creating scope with context");
             scope_with_context!(
                 isolate: isolate,
                 let &mut scope,
diff --git a/tests/kv.js b/tests/kv.js
@@ -1,6 +1,7 @@
 export default {
   async fetch(_request, env) {
-    return new Response(JSON.stringify(Object.keys(env.KV)), {
+    let hello = await env.KV.get("hello");
+    return new Response(JSON.stringify(hello), {
       status: 200,
     });
   },

Original file line number	Diff line number	Diff line change
`@@ -28,7 +28,7 @@ pub trait BindingBackend: Send + Sync + 'static {`
`28`	`28`	`fn get_tx(&self) -> BindingBackendTx;`
`29`	`29`
`30`	`30`	`/// Creates a client from the env name for the worker.`
`31`		`- fn create_client(&self, binding_name: &str) -> Box<dyn BindingClient>;`
	`31`	`+ fn create_client(&self) -> Box<dyn BindingClient>;`
`32`	`32`	`}`
`33`	`33`
`34`	`34`	`/// Creates a binding backend channel.`
Original file line number	Diff line number	Diff line change
`@@ -106,8 +106,8 @@ pub struct BindingItem {`
`106`	`106`
`107`	`107`	`impl BindingItem {`
`108`	`108`	`#[inline(always)]`
`109`		`- pub fn create_client(&self, name: &str) -> Box<dyn BindingClient> {`
`110`		`- self.backend.create_client(name)`
	`109`	`+ pub fn create_client(&self) -> Box<dyn BindingClient> {`
	`110`	`+ self.backend.create_client()`
`111`	`111`	`}`
`112`	`112`
`113`	`113`	`#[inline(always)]`
Original file line number	Diff line number	Diff line change
`@@ -201,7 +201,7 @@ impl WorkerState {`
`201`	`201`	`}`
`202`	`202`
`203`	`203`	`#[inline(always)]`
`204`		`- pub fn get_binding(&self, name: &str) -> Option<BindingBackendTx> {`
`205`		`- self.binding_store.get_binding_tx(name)`
	`204`	`+ pub fn get_binding_tx(&self, type_: &str) -> Option<BindingBackendTx> {`
	`205`	`+ self.binding_store.get_binding_tx(type_)`
`206`	`206`	`}`
`207`	`207`	`}`