char/avril

(wip) why'd you have to go and make things so complicaaaaated?

git clone https://git.t4t.associates/char/avril

Charlotte Somcore: share message bytes across broadcast recipientscc2165a

main
branchesmainall refs →
11.3 KiB357 linesraw
1use std::any::Any;
2use std::cell::RefCell;
3use std::io;
4use std::rc::Rc;
5use std::sync::Arc;
6
7use bytes::Bytes;
8use v8::{ValueDeserializerHelper, ValueSerializerHelper};
9
10#[derive(Default)]
11pub struct Message {
12    bytes: Bytes,
13    has_transfers: bool,
14    buffers: Vec<v8::SharedRef<v8::BackingStore>>,
15    shared: Vec<v8::SharedRef<v8::BackingStore>>,
16    modules: Vec<Arc<v8::CompiledWasmModule>>,
17    host_payloads: Vec<Arc<dyn Any + Send + Sync>>,
18}
19
20impl Message {
21    pub fn try_clone(&self) -> Option<Self> {
22        if self.has_transfers {
23            return None;
24        }
25        Some(Self {
26            bytes: self.bytes.clone(),
27            has_transfers: false,
28            buffers: Vec::new(),
29            shared: self.shared.clone(),
30            modules: self.modules.clone(),
31            host_payloads: self.host_payloads.clone(),
32        })
33    }
34}
35
36#[derive(Clone, Copy, PartialEq, Eq)]
37enum Direction {
38    Encode,
39    Decode,
40}
41
42#[derive(Clone)]
43struct HostContext {
44    message: Rc<RefCell<Message>>,
45    direction: Direction,
46}
47
48type HostSlot = Rc<RefCell<Option<HostContext>>>;
49
50struct HostScope {
51    slot: HostSlot,
52    previous: Option<HostContext>,
53}
54
55impl Drop for HostScope {
56    fn drop(&mut self) {
57        self.slot.replace(self.previous.take());
58    }
59}
60
61fn current_message(scope: &v8::PinScope, direction: Direction) -> io::Result<Rc<RefCell<Message>>> {
62    scope
63        .get_slot::<HostSlot>()
64        .and_then(|slot| {
65            slot.borrow()
66                .as_ref()
67                .filter(|context| context.direction == direction)
68                .map(|context| context.message.clone())
69        })
70        .ok_or_else(|| io::Error::other("No matching host serialization callback is active"))
71}
72
73/// Carry an opaque Rust payload outside the byte stream during host encoding.
74pub fn serialize_host<T: Any + Send + Sync>(
75    scope: &v8::PinScope,
76    payload: Arc<T>,
77) -> io::Result<u32> {
78    let message = current_message(scope, Direction::Encode)?;
79    let mut message = message.borrow_mut();
80    let id = u32::try_from(message.host_payloads.len())
81        .map_err(|_| io::Error::other("Too many host payloads"))?;
82    message.host_payloads.push(payload);
83    Ok(id)
84}
85
86/// Recover an opaque Rust payload during host decoding, checking its type.
87pub fn deserialize_host<T: Any + Send + Sync>(scope: &v8::PinScope, id: u32) -> io::Result<Arc<T>> {
88    current_message(scope, Direction::Decode)?
89        .borrow()
90        .host_payloads
91        .get(id as usize)
92        .cloned()
93        .and_then(|payload| payload.downcast().ok())
94        .ok_or_else(|| io::Error::other("Invalid host payload"))
95}
96
97struct Policy {
98    direction: Direction,
99    classify: Option<v8::Global<v8::Function>>,
100    host: v8::Global<v8::Function>,
101    error: v8::Global<v8::Function>,
102    message: Rc<RefCell<Message>>,
103}
104
105impl Policy {
106    fn call_host<'s>(
107        &self,
108        scope: &mut v8::PinScope<'s, '_>,
109        value: v8::Local<'s, v8::Value>,
110    ) -> Option<v8::Local<'s, v8::Value>> {
111        let slot = scope.get_slot::<HostSlot>().cloned().unwrap_or_else(|| {
112            let slot = HostSlot::default();
113            scope.set_slot(slot.clone());
114            slot
115        });
116        let previous = slot.replace(Some(HostContext {
117            message: self.message.clone(),
118            direction: self.direction,
119        }));
120        // Nested messaging must restore the outer callback's payload list, even on failure.
121        let _guard = HostScope { slot, previous };
122        let callback = v8::Local::new(scope, &self.host);
123        let receiver = v8::undefined(scope).into();
124        callback.call(scope, receiver, &[value])
125    }
126
127    fn fail(&self, scope: &mut v8::PinScope, message: &str) {
128        let message = v8::String::new(scope, message).unwrap();
129        self.throw(scope, message);
130    }
131
132    fn throw(&self, scope: &mut v8::PinScope, message: v8::Local<v8::String>) {
133        let callback = v8::Local::new(scope, &self.error);
134        let receiver = v8::undefined(scope).into();
135        if let Some(error) = callback.call(scope, receiver, &[message.into()]) {
136            scope.throw_exception(error);
137        }
138    }
139}
140
141impl v8::ValueSerializerImpl for &Policy {
142    fn throw_data_clone_error<'s>(
143        &self,
144        scope: &mut v8::PinScope<'s, '_>,
145        message: v8::Local<'s, v8::String>,
146    ) {
147        self.throw(scope, message);
148    }
149
150    fn has_custom_host_object(&self, _: &v8::Isolate) -> bool {
151        true
152    }
153
154    fn is_host_object<'s>(
155        &self,
156        scope: &mut v8::PinScope<'s, '_>,
157        object: v8::Local<'s, v8::Object>,
158    ) -> Option<bool> {
159        let callback = v8::Local::new(scope, self.classify.as_ref()?);
160        let receiver = v8::undefined(scope).into();
161        Some(
162            callback
163                .call(scope, receiver, &[object.into()])?
164                .uint32_value(scope)?
165                != 0,
166        )
167    }
168
169    fn write_host_object<'s>(
170        &self,
171        scope: &mut v8::PinScope<'s, '_>,
172        object: v8::Local<'s, v8::Object>,
173        serializer: &dyn ValueSerializerHelper,
174    ) -> Option<bool> {
175        let value = self.call_host(scope, object.into())?;
176        serializer.write_value(scope.get_current_context(), value)
177    }
178
179    fn get_shared_array_buffer_id<'s>(
180        &self,
181        scope: &mut v8::PinScope<'s, '_>,
182        buffer: v8::Local<'s, v8::SharedArrayBuffer>,
183    ) -> Option<u32> {
184        let mut message = self.message.borrow_mut();
185        let Ok(id) = u32::try_from(message.shared.len()) else {
186            drop(message);
187            self.fail(scope, "Too many shared buffers");
188            return None;
189        };
190        message.shared.push(buffer.get_backing_store());
191        Some(id)
192    }
193
194    fn get_wasm_module_transfer_id(
195        &self,
196        scope: &mut v8::PinScope<'_, '_>,
197        module: v8::Local<v8::WasmModuleObject>,
198    ) -> Option<u32> {
199        let mut message = self.message.borrow_mut();
200        let Ok(id) = u32::try_from(message.modules.len()) else {
201            drop(message);
202            self.fail(scope, "Too many WebAssembly modules");
203            return None;
204        };
205        message.modules.push(Arc::new(module.get_compiled_module()));
206        Some(id)
207    }
208}
209
210impl v8::ValueDeserializerImpl for &Policy {
211    fn read_host_object<'s>(
212        &self,
213        scope: &mut v8::PinScope<'s, '_>,
214        deserializer: &dyn ValueDeserializerHelper,
215    ) -> Option<v8::Local<'s, v8::Object>> {
216        let value = deserializer.read_value(scope.get_current_context())?;
217        // Host reconstruction calls trusted bootstrap code, not guest constructors.
218        let allowed = std::pin::pin!(v8::AllowJavascriptExecutionScope::new(scope));
219        let scope = &mut allowed.init();
220        v8::Local::try_from(self.call_host(scope, value)?).ok()
221    }
222
223    fn get_shared_array_buffer_from_id<'s>(
224        &self,
225        scope: &mut v8::PinScope<'s, '_>,
226        id: u32,
227    ) -> Option<v8::Local<'s, v8::SharedArrayBuffer>> {
228        Some(v8::SharedArrayBuffer::with_backing_store(
229            scope,
230            self.message.borrow().shared.get(id as usize)?,
231        ))
232    }
233
234    fn get_wasm_module_from_id<'s>(
235        &self,
236        scope: &mut v8::PinScope<'s, '_>,
237        id: u32,
238    ) -> Option<v8::Local<'s, v8::WasmModuleObject>> {
239        v8::WasmModuleObject::from_compiled_module(
240            scope,
241            self.message.borrow().modules.get(id as usize)?,
242        )
243    }
244}
245
246/// Callbacks are isolate-local; the returned message contains no V8 handles.
247pub fn serialize<'s>(
248    scope: &mut v8::PinScope<'s, '_>,
249    value: v8::Local<'s, v8::Value>,
250    transfer: v8::Local<'s, v8::Array>,
251    callbacks: [v8::Local<'s, v8::Function>; 5],
252) -> Option<Message> {
253    let [classify, host, error, validate, commit] = callbacks;
254    let policy = Policy {
255        direction: Direction::Encode,
256        classify: Some(v8::Global::new(scope, classify)),
257        host: v8::Global::new(scope, host),
258        error: v8::Global::new(scope, error),
259        message: Rc::new(RefCell::new(Message {
260            has_transfers: transfer.length() != 0,
261            ..Default::default()
262        })),
263    };
264    let mut buffers = Vec::new();
265    let seen = v8::Set::new(scope);
266    for index in 0..transfer.length() {
267        let value = transfer.get_index(scope, index)?;
268        if seen.has(scope, value)? {
269            policy.fail(scope, "Transfer list contains a duplicate value");
270            return None;
271        }
272        seen.add(scope, value)?;
273        if let Ok(buffer) = v8::Local::<v8::ArrayBuffer>::try_from(value) {
274            buffers.push(buffer);
275        } else {
276            let receiver = v8::undefined(scope).into();
277            if classify
278                .call(scope, receiver, &[value])?
279                .uint32_value(scope)?
280                != 2
281            {
282                policy.fail(scope, "Transfer list contains a non-transferable value");
283                return None;
284            }
285        }
286    }
287    let receiver = v8::undefined(scope).into();
288    validate.call(scope, receiver, &[])?;
289    let serializer = v8::ValueSerializer::new(scope, Box::new(&policy));
290    for (id, buffer) in buffers.iter().enumerate() {
291        serializer.transfer_array_buffer(id as u32, *buffer);
292    }
293    serializer.write_header();
294    if serializer.write_value(scope.get_current_context(), value) != Some(true) {
295        return None;
296    }
297    validate.call(scope, receiver, &[])?;
298    policy.message.borrow_mut().bytes = serializer.release().into();
299    for buffer in buffers {
300        if buffer.was_detached() {
301            policy.fail(scope, "Cannot transfer a detached ArrayBuffer");
302            return None;
303        }
304        if !buffer.is_detachable() {
305            let message = v8::String::new(scope, "Cannot detach this ArrayBuffer")?;
306            scope.throw_exception(v8::Exception::type_error(scope, message));
307            return None;
308        }
309        let store = buffer.get_backing_store();
310        if buffer.detach(None) != Some(true) {
311            return None;
312        }
313        policy.message.borrow_mut().buffers.push(store);
314    }
315    commit.call(scope, receiver, &[])?;
316    drop(serializer);
317    Some(policy.message.take())
318}
319
320pub fn deserialize<'s>(
321    scope: &mut v8::PinScope<'s, '_>,
322    mut message: Message,
323    host: v8::Local<'s, v8::Function>,
324    error: v8::Local<'s, v8::Function>,
325) -> Option<v8::Local<'s, v8::Value>> {
326    let bytes = std::mem::take(&mut message.bytes);
327    let buffers = std::mem::take(&mut message.buffers);
328    let policy = Policy {
329        direction: Direction::Decode,
330        classify: None,
331        host: v8::Global::new(scope, host),
332        error: v8::Global::new(scope, error),
333        message: Rc::new(RefCell::new(message)),
334    };
335    let deserializer = v8::ValueDeserializer::new(scope, Box::new(&policy), &bytes);
336    for (id, store) in buffers.iter().enumerate() {
337        let buffer = v8::ArrayBuffer::with_backing_store(scope, store);
338        deserializer.transfer_array_buffer(id as u32, buffer);
339    }
340    let context = scope.get_current_context();
341    let value = {
342        v8::tc_scope!(let caught, scope);
343        let value = if deserializer.read_header(context) == Some(true) {
344            deserializer.read_value(context)
345        } else {
346            None
347        };
348        if value.is_none() {
349            caught.reset();
350        }
351        value
352    };
353    if value.is_none() {
354        policy.fail(scope, "Cannot deserialize this message");
355    }
356    value
357}