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::{MutPtrByteSlice, 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/// Parameters for delivering verified parent blob data (or sparse zeroes) into a child VMO.
52pub struct ChildData<'a> {
53    pub parent_offset: u64,
54    pub offset: u32,
55    pub length: u32,
56    pub payload_region_offset: u64,
57    pub payload: Option<MutPtrByteSlice<'a>>,
58    pub child_key: u64,
59    pub child_target_offset: u64,
60    pub in_chunk_offset: u32,
61    pub child_length: u32,
62}
63
64/// Parameters for delivering an EROFS inline tail directly from the parent blob's VMO in
65/// `pkg-cache`.
66#[derive(Copy, Clone, Debug)]
67pub struct ChildInlineTail {
68    pub parent_byte_offset: u64,
69    pub child_key: u64,
70    pub child_target_offset: u64,
71    pub tail_length: u32,
72}
73
74/// Trait providing data delivery and blob metadata registration for incoming delivery commands.
75pub trait DeliveryQueueProvider: Send + Sync + 'static {
76    /// Delivers data directly from the delivery VMO into the target blob.
77    fn deliver_pages(
78        &self,
79        key: u64,
80        target_offset: u64,
81        unverified_pages: UnverifiedPages<'_>,
82    ) -> Result<(), Error>;
83
84    /// Handles a RegisterBlob command using a pointer slice to shared memory.
85    fn register_blob(
86        &self,
87        key: u64,
88        leaf_data: PtrByteSlice<'_>,
89        read_alignment: usize,
90    ) -> Result<(), Error> {
91        let _ = (key, leaf_data, read_alignment);
92        Ok(())
93    }
94
95    /// Verifies a parent blob chunk (when `data.length > 0`), applies file-tail zeroing in-place
96    /// in the delivery queue VMO, and supplies the resulting pages to the child VMO.
97    fn deliver_child_pages(&self, _data: ChildData<'_>) -> Result<(), Error> {
98        unreachable!()
99    }
100
101    /// Supplies an EROFS inline tail page by reading directly from the parent blob's VMO.
102    fn deliver_child_inline_tail(
103        &self,
104        _data: ChildInlineTail,
105        _receiver: &vmo_fifo::Receiver<RawDeliveryCommand>,
106    ) -> Result<(), Error> {
107        unreachable!()
108    }
109}
110
111/// A simple in-memory `DeliveryQueueProvider` for testing delivery without Merkle verification.
112pub struct TestVmoProvider {
113    pager: Arc<zx::Pager>,
114    delivery_vmo: zx::Vmo,
115    vmos: Mutex<HashMap<u64, zx::Vmo>>,
116}
117
118impl TestVmoProvider {
119    pub fn new(pager: Arc<zx::Pager>, delivery_vmo: zx::Vmo) -> Self {
120        Self { pager, delivery_vmo, vmos: Mutex::new(HashMap::new()) }
121    }
122
123    pub fn delivery_vmo(&self) -> &zx::Vmo {
124        &self.delivery_vmo
125    }
126
127    pub fn register_vmo(&self, key: u64, vmo: zx::Vmo) {
128        self.vmos.lock().insert(key, vmo);
129    }
130
131    pub fn unregister_vmo(&self, key: u64) {
132        self.vmos.lock().remove(&key);
133    }
134}
135
136impl DeliveryQueueProvider for TestVmoProvider {
137    fn deliver_pages(
138        &self,
139        key: u64,
140        target_offset: u64,
141        unverified_pages: UnverifiedPages<'_>,
142    ) -> Result<(), Error> {
143        let vmos = self.vmos.lock();
144        if let Some(target_vmo) = vmos.get(&key) {
145            let length = unverified_pages.len_in_bytes() as u64;
146            if length == 0 {
147                return Ok(());
148            }
149            self.pager.supply_pages(
150                target_vmo,
151                target_offset..target_offset + length,
152                &self.delivery_vmo,
153                unverified_pages.delivery_offset(),
154            )?;
155            Ok(())
156        } else {
157            bail!("Unknown or expired key {key} in deliver_pages");
158        }
159    }
160}
161
162const NUM_DELIVERY_THREADS: usize = 2;
163
164pub struct DeliveryQueueProcessor {
165    delivery_vmo: zx::Vmo,
166    delivery_threads: Vec<std::thread::JoinHandle<()>>,
167}
168
169impl DeliveryQueueProcessor {
170    pub fn spawn(
171        receiver: vmo_fifo::Receiver<RawDeliveryCommand>,
172        provider: Arc<dyn DeliveryQueueProvider>,
173        delivery_vmo: zx::Vmo,
174    ) -> Result<Self, Error> {
175        let delivery_vmo_clone = delivery_vmo.duplicate_handle(zx::Rights::SAME_RIGHTS)?;
176        let receiver = Arc::new(receiver);
177        let mut delivery_threads = Vec::with_capacity(NUM_DELIVERY_THREADS);
178
179        for _ in 0..NUM_DELIVERY_THREADS {
180            let receiver = receiver.clone();
181            let provider = provider.clone();
182            let thread = std::thread::spawn(move || {
183                while let Ok((mut msg, maybe_cmd)) = receiver.claim_with(|msg| {
184                    match DeliveryCommand::try_from(**msg) {
185                        // Process `RegisterBlob` before releasing the claim lock so that any
186                        // subsequent `Data` commands for this blob claimed by other worker threads
187                        // observe the initialized Merkle verifier.
188                        Ok(cmd @ DeliveryCommand::RegisterBlob { .. }) => {
189                            if let Err(e) = Self::handle_delivery(cmd, msg, &*provider) {
190                                log::warn!("Failed to handle delivery command: {:?}", e);
191                            }
192                            None
193                        }
194                        Ok(cmd) => Some(cmd),
195                        Err(e) => {
196                            log::warn!("Invalid delivery command opcode: {:?}", e);
197                            None
198                        }
199                    }
200                }) {
201                    match maybe_cmd {
202                        Some(DeliveryCommand::ChildInlineTail {
203                            parent_byte_offset,
204                            child_key,
205                            child_target_offset,
206                            tail_length,
207                        }) => {
208                            // Pop the header-only command from the FIFO before reading from the
209                            // parent blob's VMO so `read_head` can advance even if the parent VMO
210                            // read faults on an evicted or not-yet-supplied parent page.
211                            if let Err(e) = msg.pop() {
212                                log::warn!("Failed to pop delivery message: {:?}", e);
213                                break;
214                            }
215                            if let Err(e) = provider.deliver_child_inline_tail(
216                                ChildInlineTail {
217                                    parent_byte_offset,
218                                    child_key,
219                                    child_target_offset,
220                                    tail_length,
221                                },
222                                &receiver,
223                            ) {
224                                log::warn!("Failed to handle ChildInlineTail command: {:?}", e);
225                            }
226                        }
227                        Some(cmd) => {
228                            if let Err(e) = Self::handle_delivery(cmd, &mut msg, &*provider) {
229                                log::warn!("Failed to handle delivery command: {:?}", e);
230                            }
231                            if let Err(e) = msg.pop() {
232                                log::warn!("Failed to pop delivery message: {:?}", e);
233                                break;
234                            }
235                        }
236                        None => {
237                            if let Err(e) = msg.pop() {
238                                log::warn!("Failed to pop delivery message: {:?}", e);
239                                break;
240                            }
241                        }
242                    }
243                }
244            });
245            delivery_threads.push(thread);
246        }
247
248        Ok(Self { delivery_vmo: delivery_vmo_clone, delivery_threads })
249    }
250
251    pub fn delivery_vmo(&self) -> &zx::Vmo {
252        &self.delivery_vmo
253    }
254
255    fn handle_delivery(
256        cmd: DeliveryCommand,
257        raw_msg: &mut vmo_fifo::Message<'_, RawDeliveryCommand>,
258        provider: &dyn DeliveryQueueProvider,
259    ) -> Result<(), Error> {
260        match cmd {
261            DeliveryCommand::Data { key, target_offset, length, offset } => {
262                if offset.checked_add(length).unwrap_or(u32::MAX) > DELIVERY_VMO_SIZE as u32 {
263                    bail!("Data payload out of bounds");
264                }
265                let vmo_offset = raw_msg.payload_region_offset() as u64 + offset as u64;
266                // The sender is expected to push page-aligned payloads. If the payload is the
267                // final chunk of a blob and the data does not end on a page boundary, the sender
268                // must extend `length` to the next page boundary and pad the trailing bytes with
269                // zeros. This is required because `fuchsia-merkle`'s `verify_aligned` expects
270                // the buffer length to be a multiple of the system page size.
271                let page_size = zx::system_get_page_size() as u32;
272                if offset % page_size != 0 {
273                    bail!("Delivery payload offset must be page aligned: {offset}");
274                }
275                if length % page_size != 0 {
276                    bail!("Delivery payload length must be page aligned: {length}");
277                }
278                if target_offset % (page_size as u64) != 0 {
279                    bail!("Delivery payload target_offset must be page aligned: {target_offset}");
280                }
281                let unverified_pages =
282                    UnverifiedPages::new(raw_msg.payload_slice(offset, length), vmo_offset)
283                        .map_err(|s| anyhow!("Invalid unverified pages: {s}"))?;
284                provider.deliver_pages(key, target_offset, unverified_pages)?;
285            }
286            DeliveryCommand::RegisterBlob { key, offset, length, read_alignment } => {
287                if offset.checked_add(length).unwrap_or(u32::MAX) > DELIVERY_VMO_SIZE as u32 {
288                    bail!("RegisterBlob payload out of bounds");
289                }
290                let leaf_data = raw_msg.payload_slice(offset, length);
291                provider.register_blob(key, leaf_data, read_alignment as usize)?;
292            }
293            DeliveryCommand::ChildData {
294                parent_offset,
295                offset,
296                length,
297                child_key,
298                child_target_offset,
299                in_chunk_offset,
300                child_length,
301            } => {
302                let payload_region_offset = raw_msg.payload_region_offset() as u64;
303                let payload = (length > 0
304                    && offset
305                        .checked_add(length)
306                        .is_some_and(|end| end <= DELIVERY_VMO_SIZE as u32))
307                .then(|| raw_msg.payload_mut_slice(offset, length));
308                provider.deliver_child_pages(ChildData {
309                    parent_offset,
310                    offset,
311                    length,
312                    payload_region_offset,
313                    payload,
314                    child_key,
315                    child_target_offset,
316                    in_chunk_offset,
317                    child_length,
318                })?;
319            }
320            DeliveryCommand::ChildInlineTail { .. } => {
321                unreachable!("ChildInlineTail is handled before handle_delivery");
322            }
323        }
324        Ok(())
325    }
326}
327
328impl Drop for DeliveryQueueProcessor {
329    fn drop(&mut self) {
330        let _ = self.delivery_vmo.signal(zx::Signals::empty(), vmo_fifo::SIG_SHUTDOWN);
331        for thread in self.delivery_threads.drain(..) {
332            let _ = thread.join();
333        }
334    }
335}
blob_pager_and_verifier/delivery.rs

blob_pager_and_verifier/
delivery.rs
