blob_pager_and_verifier/
delivery.rs

1// Copyright 2026 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use anyhow::{Error, anyhow, bail};
6use fuchsia_sync::Mutex;
7use mapping::{DELIVERY_VMO_SIZE, DeliveryCommand, RawDeliveryCommand};
8use std::collections::HashMap;
9use std::sync::Arc;
10use storage_ptr_slice::PtrByteSlice;
11use zx;
12
13/// Unverified data pages delivered from the driver to be verified.
14#[derive(Copy, Clone)]
15pub struct UnverifiedPages<'a> {
16    slice: PtrByteSlice<'a>,
17    /// Byte offset within the shared delivery VMO.
18    delivery_offset: u64,
19}
20
21impl<'a> UnverifiedPages<'a> {
22    pub(crate) fn new(slice: PtrByteSlice<'a>, delivery_offset: u64) -> Result<Self, zx::Status> {
23        let page_size = zx::system_get_page_size() as usize;
24        if slice.len() % page_size != 0 || delivery_offset % (page_size as u64) != 0 {
25            return Err(zx::Status::INVALID_ARGS);
26        }
27        Ok(Self { slice, delivery_offset })
28    }
29
30    /// Returns the length of the payload in bytes.
31    pub fn len_in_bytes(&self) -> usize {
32        self.slice.len()
33    }
34
35    /// Returns the length of the payload in pages.
36    pub fn len_in_pages(&self) -> usize {
37        self.slice.len() / (zx::system_get_page_size() as usize)
38    }
39
40    /// Returns the byte offset within the shared delivery VMO.
41    pub fn delivery_offset(&self) -> u64 {
42        self.delivery_offset
43    }
44
45    /// Returns the underlying raw pointer byte slice to the unverified data.
46    pub fn as_ptr_byte_slice(&self) -> PtrByteSlice<'a> {
47        self.slice
48    }
49}
50
51/// Trait providing data delivery and blob metadata registration for incoming delivery commands.
52pub trait DeliveryQueueProvider: Send + Sync + 'static {
53    /// Delivers data directly from the delivery VMO into the target blob.
54    fn deliver_pages(
55        &self,
56        key: u64,
57        target_offset: u64,
58        unverified_pages: UnverifiedPages<'_>,
59    ) -> Result<(), Error>;
60
61    /// Handles a RegisterBlob command using a pointer slice to shared memory.
62    fn register_blob(
63        &self,
64        key: u64,
65        leaf_data: PtrByteSlice<'_>,
66        read_alignment: usize,
67    ) -> Result<(), Error> {
68        let _ = (key, leaf_data, read_alignment);
69        Ok(())
70    }
71}
72
73/// A simple in-memory `DeliveryQueueProvider` for testing delivery without Merkle verification.
74pub struct TestVmoProvider {
75    pager: Arc<zx::Pager>,
76    delivery_vmo: zx::Vmo,
77    vmos: Mutex<HashMap<u64, zx::Vmo>>,
78}
79
80impl TestVmoProvider {
81    pub fn new(pager: Arc<zx::Pager>, delivery_vmo: zx::Vmo) -> Self {
82        Self { pager, delivery_vmo, vmos: Mutex::new(HashMap::new()) }
83    }
84
85    pub fn delivery_vmo(&self) -> &zx::Vmo {
86        &self.delivery_vmo
87    }
88
89    pub fn register_vmo(&self, key: u64, vmo: zx::Vmo) {
90        self.vmos.lock().insert(key, vmo);
91    }
92
93    pub fn unregister_vmo(&self, key: u64) {
94        self.vmos.lock().remove(&key);
95    }
96}
97
98impl DeliveryQueueProvider for TestVmoProvider {
99    fn deliver_pages(
100        &self,
101        key: u64,
102        target_offset: u64,
103        unverified_pages: UnverifiedPages<'_>,
104    ) -> Result<(), Error> {
105        let vmos = self.vmos.lock();
106        if let Some(target_vmo) = vmos.get(&key) {
107            let length = unverified_pages.len_in_bytes() as u64;
108            if length == 0 {
109                return Ok(());
110            }
111            self.pager.supply_pages(
112                target_vmo,
113                target_offset..target_offset + length,
114                &self.delivery_vmo,
115                unverified_pages.delivery_offset(),
116            )?;
117            Ok(())
118        } else {
119            bail!("Unknown or expired key {key} in deliver_pages");
120        }
121    }
122}
123
124const NUM_DELIVERY_THREADS: usize = 2;
125
126pub struct DeliveryQueueProcessor {
127    delivery_vmo: zx::Vmo,
128    delivery_threads: Vec<std::thread::JoinHandle<()>>,
129}
130
131impl DeliveryQueueProcessor {
132    pub fn spawn(
133        receiver: vmo_fifo::Receiver<RawDeliveryCommand>,
134        provider: Arc<dyn DeliveryQueueProvider>,
135        delivery_vmo: zx::Vmo,
136    ) -> Result<Self, Error> {
137        let delivery_vmo_clone = delivery_vmo.duplicate_handle(zx::Rights::SAME_RIGHTS)?;
138        let receiver = Arc::new(receiver);
139        let mut delivery_threads = Vec::with_capacity(NUM_DELIVERY_THREADS);
140
141        for _ in 0..NUM_DELIVERY_THREADS {
142            let receiver = receiver.clone();
143            let provider = provider.clone();
144            let thread = std::thread::spawn(move || {
145                while let Ok((msg, maybe_cmd)) = receiver.claim_with(|msg| {
146                    match DeliveryCommand::try_from(**msg) {
147                        // Process `RegisterBlob` before releasing the claim lock so that any
148                        // subsequent `Data` commands for this blob claimed by other worker threads
149                        // observe the initialized Merkle verifier.
150                        Ok(cmd @ DeliveryCommand::RegisterBlob { .. }) => {
151                            if let Err(e) = Self::handle_delivery(cmd, msg, &*provider) {
152                                log::warn!("Failed to handle delivery command: {:?}", e);
153                            }
154                            None
155                        }
156                        Ok(cmd) => Some(cmd),
157                        Err(e) => {
158                            log::warn!("Invalid delivery command opcode: {:?}", e);
159                            None
160                        }
161                    }
162                }) {
163                    if let Some(cmd) = maybe_cmd {
164                        if let Err(e) = Self::handle_delivery(cmd, &msg, &*provider) {
165                            log::warn!("Failed to handle delivery command: {:?}", e);
166                        }
167                    }
168                    if let Err(e) = msg.pop() {
169                        log::warn!("Failed to pop delivery message: {:?}", e);
170                        break;
171                    }
172                }
173            });
174            delivery_threads.push(thread);
175        }
176
177        Ok(Self { delivery_vmo: delivery_vmo_clone, delivery_threads })
178    }
179
180    pub fn delivery_vmo(&self) -> &zx::Vmo {
181        &self.delivery_vmo
182    }
183
184    fn handle_delivery(
185        cmd: DeliveryCommand,
186        raw_msg: &vmo_fifo::Message<'_, RawDeliveryCommand>,
187        provider: &dyn DeliveryQueueProvider,
188    ) -> Result<(), Error> {
189        match cmd {
190            DeliveryCommand::Data { key, target_offset, length, offset } => {
191                if offset.checked_add(length).unwrap_or(u32::MAX) > DELIVERY_VMO_SIZE as u32 {
192                    bail!("Data payload out of bounds");
193                }
194                let vmo_offset = raw_msg.payload_region_offset() as u64 + offset as u64;
195                // The sender is expected to push page-aligned payloads. If the payload is the
196                // final chunk of a blob and the data does not end on a page boundary, the sender
197                // must extend `length` to the next page boundary and pad the trailing bytes with
198                // zeros. This is required because `fuchsia-merkle`'s `verify_aligned` expects
199                // the buffer length to be a multiple of the system page size.
200                let page_size = zx::system_get_page_size() as u32;
201                if offset % page_size != 0 {
202                    bail!("Delivery payload offset must be page aligned: {offset}");
203                }
204                if length % page_size != 0 {
205                    bail!("Delivery payload length must be page aligned: {length}");
206                }
207                if target_offset % (page_size as u64) != 0 {
208                    bail!("Delivery payload target_offset must be page aligned: {target_offset}");
209                }
210                let unverified_pages =
211                    UnverifiedPages::new(raw_msg.payload_slice(offset, length), vmo_offset)
212                        .map_err(|s| anyhow!("Invalid unverified pages: {s}"))?;
213                provider.deliver_pages(key, target_offset, unverified_pages)?;
214            }
215            DeliveryCommand::RegisterBlob { key, offset, length, read_alignment } => {
216                if offset.checked_add(length).unwrap_or(u32::MAX) > DELIVERY_VMO_SIZE as u32 {
217                    bail!("RegisterBlob payload out of bounds");
218                }
219                let leaf_data = raw_msg.payload_slice(offset, length);
220                provider.register_blob(key, leaf_data, read_alignment as usize)?;
221            }
222        }
223        Ok(())
224    }
225}
226
227impl Drop for DeliveryQueueProcessor {
228    fn drop(&mut self) {
229        let _ = self.delivery_vmo.signal(zx::Signals::empty(), vmo_fifo::SIG_SHUTDOWN);
230        for thread in self.delivery_threads.drain(..) {
231            let _ = thread.join();
232        }
233    }
234}
blob_pager_and_verifier/delivery.rs

blob_pager_and_verifier/
delivery.rs
