fory_core/resolver/
meta_resolver.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use crate::buffer::{Reader, Writer};
19use crate::config::Config;
20use crate::error::Error;
21use crate::meta::TypeMeta;
22use crate::resolver::type_resolver::NO_USER_TYPE_ID;
23use crate::resolver::{TypeInfo, TypeResolver};
24use std::collections::HashMap;
25use std::rc::Rc;
26
27/// Streaming meta writer that writes TypeMeta inline during serialization.
28/// Uses the streaming protocol:
29/// - (index << 1) | 0 for new type definition (followed by TypeMeta bytes)
30/// - (index << 1) | 1 for reference to previously written type
31#[derive(Default)]
32pub struct MetaWriterResolver {
33    // Provider and target indexes share one Rc<TypeInfo>; pointer identity keeps
34    // their streaming metadata references in one sequence without probing maps.
35    type_info_index_map: HashMap<*const TypeInfo, usize>,
36    type_index_index_map: Vec<usize>,
37    next_index: usize,
38}
39
40const MIN_REMOTE_TYPE_META_LIMIT: usize = 8192;
41const NO_WRITTEN_TYPE_INDEX: usize = usize::MAX;
42
43#[allow(dead_code)]
44impl MetaWriterResolver {
45    /// Write type meta inline using streaming protocol.
46    /// Returns the index assigned to this type.
47    #[inline(always)]
48    pub fn write_type_meta(
49        &mut self,
50        writer: &mut Writer,
51        provider_type_id: std::any::TypeId,
52        type_resolver: &TypeResolver,
53    ) -> Result<(), Error> {
54        let type_info = type_resolver.get_provider_type_info(&provider_type_id)?;
55        self.write_resolved_type_meta(writer, &type_info)
56    }
57
58    #[inline(always)]
59    pub(crate) fn write_resolved_type_meta(
60        &mut self,
61        writer: &mut Writer,
62        type_info: &Rc<TypeInfo>,
63    ) -> Result<(), Error> {
64        let identity = Rc::as_ptr(type_info);
65        match self.type_info_index_map.get(&identity) {
66            Some(&index) => {
67                // Reference to previously written type: (index << 1) | 1, LSB=1
68                writer.write_var_u32(((index as u32) << 1) | 1);
69            }
70            None => {
71                // New type: index << 1, LSB=0, followed by TypeMeta bytes inline
72                let index = self.next_index;
73                self.next_index += 1;
74                writer.write_var_u32((index as u32) << 1);
75                self.type_info_index_map.insert(identity, index);
76                let type_def = type_info.get_type_def();
77                writer.write_bytes(&type_def);
78            }
79        }
80        Ok(())
81    }
82
83    /// Write type meta by generated struct type index, avoiding Rust TypeId hash lookup.
84    #[inline(always)]
85    pub fn write_type_meta_fast(
86        &mut self,
87        writer: &mut Writer,
88        type_id: std::any::TypeId,
89        type_index: u32,
90        type_resolver: &TypeResolver,
91    ) -> Result<(), Error> {
92        let type_index = type_index as usize;
93        if let Some(&index) = self.type_index_index_map.get(type_index) {
94            if index != NO_WRITTEN_TYPE_INDEX {
95                writer.write_var_u32(((index as u32) << 1) | 1);
96                return Ok(());
97            }
98        }
99
100        let index = self.next_index;
101        self.next_index += 1;
102        writer.write_var_u32((index as u32) << 1);
103        if type_index >= self.type_index_index_map.len() {
104            self.type_index_index_map
105                .resize(type_index + 1, NO_WRITTEN_TYPE_INDEX);
106        }
107        self.type_index_index_map[type_index] = index;
108        let type_meta = type_resolver.get_type_meta_by_index_ref(&type_id, type_index as u32)?;
109        writer.write_bytes(type_meta.get_bytes());
110        Ok(())
111    }
112
113    #[inline(always)]
114    pub fn reset(&mut self) {
115        self.type_info_index_map.clear();
116        self.type_index_index_map.clear();
117        self.next_index = 0;
118    }
119}
120
121/// Streaming meta reader that reads TypeMeta inline during deserialization.
122/// Uses the streaming protocol:
123/// - (index << 1) | 0 for new type definition (followed by TypeMeta bytes)
124/// - (index << 1) | 1 for reference to previously read type
125#[derive(Default)]
126pub struct MetaReaderResolver {
127    pub reading_type_infos: Vec<Rc<TypeInfo>>,
128    parsed_type_infos: HashMap<i64, Rc<TypeInfo>>,
129    remote_schema_versions_by_type: HashMap<String, usize>,
130    total_accepted_schema_versions: usize,
131    cached_meta_header: i64,
132    cached_type_info: Option<Rc<TypeInfo>>,
133}
134
135impl MetaReaderResolver {
136    #[inline(always)]
137    pub fn get(&self, index: usize) -> Option<&Rc<TypeInfo>> {
138        self.reading_type_infos.get(index)
139    }
140
141    /// Read type meta inline using streaming protocol.
142    /// Returns the TypeInfo for this type.
143    #[inline(always)]
144    pub fn read_type_meta(
145        &mut self,
146        reader: &mut Reader,
147        type_resolver: &TypeResolver,
148        config: &Config,
149    ) -> Result<Rc<TypeInfo>, Error> {
150        let index_marker = reader.read_var_u32()?;
151        let is_ref = (index_marker & 1) == 1;
152        let index = (index_marker >> 1) as usize;
153
154        if is_ref {
155            // Reference to previously read type
156            self.reading_type_infos.get(index).cloned().ok_or_else(|| {
157                Error::type_error(format!("TypeInfo not found for type index: {}", index))
158            })
159        } else {
160            // New type - read TypeMeta inline
161            let meta_header = reader.read_i64()?;
162            if let Some(type_info) = self
163                .cached_type_info
164                .as_ref()
165                .filter(|_| self.cached_meta_header == meta_header)
166            {
167                // Header-cache hits intentionally skip without rehashing. Entries reach this cache
168                // only after a successful TypeMeta parse and 52-bit metadata-hash validation. Do
169                // not add body/hash/schema-limit/exact-local checks here; the miss path owns them
170                // before publish.
171                self.reading_type_infos.push(type_info.clone());
172                TypeMeta::skip_bytes_for_validated_header(reader, meta_header)?;
173                return Ok(type_info.clone());
174            }
175            if let Some(type_info) = self.parsed_type_infos.get(&meta_header) {
176                // Header-cache hits intentionally skip without rehashing. Entries reach this cache
177                // only after a successful TypeMeta parse and 52-bit metadata-hash validation. Do
178                // not add body/hash/schema-limit/exact-local checks here; the miss path owns them
179                // before publish.
180                self.cached_meta_header = meta_header;
181                self.cached_type_info = Some(type_info.clone());
182                self.reading_type_infos.push(type_info.clone());
183                TypeMeta::skip_bytes_for_validated_header(reader, meta_header)?;
184                Ok(type_info.clone())
185            } else {
186                let type_def_start = reader.get_cursor() - std::mem::size_of::<i64>();
187                self.read_remote_type_meta(
188                    reader,
189                    type_resolver,
190                    config,
191                    meta_header,
192                    type_def_start,
193                )
194            }
195        }
196    }
197
198    #[cold]
199    #[inline(never)]
200    fn read_remote_type_meta(
201        &mut self,
202        reader: &mut Reader,
203        type_resolver: &TypeResolver,
204        config: &Config,
205        meta_header: i64,
206        type_def_start: usize,
207    ) -> Result<Rc<TypeInfo>, Error> {
208        let type_meta = Rc::new(TypeMeta::from_bytes_with_header(
209            reader,
210            type_resolver,
211            meta_header,
212            config.max_type_fields(),
213            config.max_type_meta_bytes(),
214        )?);
215        let remote_type_def = reader.sub_slice(type_def_start, reader.get_cursor())?;
216
217        let namespace = type_meta.get_namespace();
218        let type_name = type_meta.get_type_name();
219        let register_by_name = !namespace.original.is_empty() || !type_name.original.is_empty();
220        let mut remote_schema_key = None;
221        let type_info = if register_by_name {
222            if let Some(local_type_info) =
223                type_resolver.get_type_info_by_name(&namespace.original, &type_name.original)
224            {
225                if local_type_info.get_type_meta_ref().get_bytes() == remote_type_def {
226                    local_type_info
227                } else {
228                    remote_schema_key =
229                        Some(self.check_remote_type_meta_limit(&type_meta, config)?);
230                    Rc::new(TypeInfo::from_remote_meta(
231                        type_meta.clone(),
232                        Some(local_type_info.get_harness()),
233                        Some(local_type_info.get_type_id() as u32),
234                        Some(local_type_info.get_user_type_id()),
235                    ))
236                }
237            } else {
238                remote_schema_key = Some(self.check_remote_type_meta_limit(&type_meta, config)?);
239                Rc::new(TypeInfo::from_remote_meta(
240                    type_meta.clone(),
241                    None,
242                    None,
243                    None,
244                ))
245            }
246        } else {
247            let type_id = type_meta.get_type_id();
248            let user_type_id = type_meta.get_user_type_id();
249            let local_type_info = if user_type_id != NO_USER_TYPE_ID {
250                type_resolver.get_user_type_info_by_id(user_type_id)
251            } else {
252                type_resolver.get_type_info_by_id(type_id)
253            };
254            if let Some(local_type_info) = local_type_info {
255                if local_type_info.get_type_meta_ref().get_bytes() == remote_type_def {
256                    local_type_info
257                } else {
258                    remote_schema_key =
259                        Some(self.check_remote_type_meta_limit(&type_meta, config)?);
260                    Rc::new(TypeInfo::from_remote_meta(
261                        type_meta.clone(),
262                        Some(local_type_info.get_harness()),
263                        Some(local_type_info.get_type_id() as u32),
264                        Some(local_type_info.get_user_type_id()),
265                    ))
266                }
267            } else {
268                remote_schema_key = Some(self.check_remote_type_meta_limit(&type_meta, config)?);
269                Rc::new(TypeInfo::from_remote_meta(
270                    type_meta.clone(),
271                    None,
272                    None,
273                    None,
274                ))
275            }
276        };
277
278        self.parsed_type_infos
279            .insert(meta_header, type_info.clone());
280        self.cached_meta_header = meta_header;
281        self.cached_type_info = Some(type_info.clone());
282        self.reading_type_infos.push(type_info.clone());
283        if let Some(remote_schema_key) = remote_schema_key {
284            self.record_remote_type_meta(remote_schema_key);
285        }
286        Ok(type_info)
287    }
288
289    #[cold]
290    #[inline(never)]
291    fn check_remote_type_meta_limit(
292        &self,
293        type_meta: &TypeMeta,
294        config: &Config,
295    ) -> Result<String, Error> {
296        let namespace = type_meta.get_namespace();
297        let type_name = type_meta.get_type_name();
298        let key = if !namespace.original.is_empty() || !type_name.original.is_empty() {
299            format!("n{}\0{}", namespace.original, type_name.original)
300        } else {
301            format!("i{}", type_meta.get_user_type_id())
302        };
303
304        let versions_for_type = self
305            .remote_schema_versions_by_type
306            .get(&key)
307            .copied()
308            .unwrap_or(0);
309        if versions_for_type >= config.max_schema_versions_per_type() {
310            return Err(Error::invalid_data(format!(
311                "remote schema version limit exceeded for one type. The data may be malicious. If the data is not malicious, please increase max_schema_versions_per_type={}",
312                config.max_schema_versions_per_type()
313            )));
314        }
315
316        let accepted_type_count =
317            self.remote_schema_versions_by_type.len() + if versions_for_type == 0 { 1 } else { 0 };
318        let global_limit = usize::max(
319            MIN_REMOTE_TYPE_META_LIMIT,
320            accepted_type_count * config.max_average_schema_versions_per_type(),
321        );
322        if self.total_accepted_schema_versions >= global_limit {
323            return Err(Error::invalid_data(format!(
324                "remote schema version limit exceeded globally. The data may be malicious. If the data is not malicious, please increase max_average_schema_versions_per_type={}",
325                config.max_average_schema_versions_per_type()
326            )));
327        }
328
329        Ok(key)
330    }
331
332    fn record_remote_type_meta(&mut self, key: String) {
333        let versions_for_type = self
334            .remote_schema_versions_by_type
335            .get(&key)
336            .copied()
337            .unwrap_or(0);
338        self.remote_schema_versions_by_type
339            .insert(key, versions_for_type + 1);
340        self.total_accepted_schema_versions += 1;
341    }
342
343    #[inline(always)]
344    pub fn reset(&mut self) {
345        self.reading_type_infos.clear();
346    }
347}
348
349#[cfg(test)]
350mod tests {
351    use super::*;
352    use crate::config::Config;
353    use crate::context::{ReadContext, WriteContext};
354    use crate::meta::{
355        FieldInfo, FieldType, MetaString, NAMESPACE_ENCODER, NAMESPACE_ENCODINGS,
356        TYPE_NAME_ENCODER, TYPE_NAME_ENCODINGS,
357    };
358    use crate::serializer::Serializer;
359    use crate::TypeId;
360
361    struct LocalExt;
362
363    impl Serializer for LocalExt {
364        type Target = Self;
365
366        fn write_data(_value: &Self, _context: &mut WriteContext) -> Result<(), Error> {
367            Ok(())
368        }
369
370        fn read_data(_context: &mut ReadContext) -> Result<Self, Error> {
371            Ok(LocalExt)
372        }
373    }
374
375    fn read_type_def(
376        resolver: &mut MetaReaderResolver,
377        config: &Config,
378        type_def: &[u8],
379    ) -> Result<Rc<TypeInfo>, Error> {
380        let type_resolver = TypeResolver::default();
381        read_type_def_with_type_resolver(resolver, config, &type_resolver, type_def)
382    }
383
384    fn read_type_def_with_type_resolver(
385        resolver: &mut MetaReaderResolver,
386        config: &Config,
387        type_resolver: &TypeResolver,
388        type_def: &[u8],
389    ) -> Result<Rc<TypeInfo>, Error> {
390        let mut bytes = vec![];
391        let mut writer = Writer::from_buffer(&mut bytes);
392        writer.write_var_u32(0);
393        writer.write_bytes(type_def);
394        let mut reader = Reader::new(&bytes);
395        resolver.read_type_meta(&mut reader, type_resolver, config)
396    }
397
398    #[test]
399    fn type_meta_field_limit_rejects_large_struct() {
400        let meta = TypeMeta::new(
401            TypeId::STRUCT as u32,
402            9001,
403            MetaString::get_empty().clone(),
404            MetaString::get_empty().clone(),
405            false,
406            vec![
407                FieldInfo::new("a", FieldType::new(crate::type_id::INT32, false, vec![])),
408                FieldInfo::new("b", FieldType::new(crate::type_id::INT32, false, vec![])),
409            ],
410        )
411        .unwrap();
412        let config = Config {
413            max_type_fields: 1,
414            ..Default::default()
415        };
416        let err = read_type_def(
417            &mut MetaReaderResolver::default(),
418            &config,
419            meta.get_bytes(),
420        )
421        .unwrap_err()
422        .to_string();
423        assert!(err.contains("max_type_fields"));
424    }
425
426    #[test]
427    fn type_meta_body_limit_rejects_large_metadata() {
428        let meta = TypeMeta::new(
429            TypeId::STRUCT as u32,
430            9001,
431            MetaString::get_empty().clone(),
432            MetaString::get_empty().clone(),
433            false,
434            vec![FieldInfo::new(
435                "a",
436                FieldType::new(crate::type_id::INT32, false, vec![]),
437            )],
438        )
439        .unwrap();
440        let config = Config {
441            max_type_meta_bytes: 1,
442            ..Default::default()
443        };
444        let err = read_type_def(
445            &mut MetaReaderResolver::default(),
446            &config,
447            meta.get_bytes(),
448        )
449        .unwrap_err()
450        .to_string();
451        assert!(err.contains("max_type_meta_bytes"));
452    }
453
454    #[test]
455    fn schema_limit_tracks_unknown_struct_types_separately() {
456        fn type_def(user_type_id: u32, field_name: &str) -> Vec<u8> {
457            TypeMeta::new(
458                TypeId::STRUCT as u32,
459                user_type_id,
460                MetaString::get_empty().clone(),
461                MetaString::get_empty().clone(),
462                false,
463                vec![FieldInfo::new(
464                    field_name,
465                    FieldType::new(crate::type_id::INT32, false, vec![]),
466                )],
467            )
468            .unwrap()
469            .get_bytes()
470            .to_vec()
471        }
472
473        let config = Config {
474            max_schema_versions_per_type: 1,
475            ..Default::default()
476        };
477
478        let mut resolver = MetaReaderResolver::default();
479        read_type_def(&mut resolver, &config, &type_def(9001, "a")).unwrap();
480        read_type_def(&mut resolver, &config, &type_def(9002, "a")).unwrap();
481
482        let err = read_type_def(&mut resolver, &config, &type_def(9001, "b"))
483            .unwrap_err()
484            .to_string();
485        assert!(err.contains("max_schema_versions_per_type"));
486    }
487
488    #[test]
489    fn schema_limit_rejects_extra_versions_for_type() {
490        let meta = TypeMeta::new(
491            TypeId::STRUCT as u32,
492            9001,
493            MetaString::get_empty().clone(),
494            MetaString::get_empty().clone(),
495            false,
496            vec![FieldInfo::new(
497                "a",
498                FieldType::new(crate::type_id::INT32, false, vec![]),
499            )],
500        )
501        .unwrap();
502        let type_def = meta.get_bytes().to_vec();
503
504        let config = Config {
505            max_schema_versions_per_type: 1,
506            ..Default::default()
507        };
508        let mut resolver = MetaReaderResolver::default();
509        let mut bytes = vec![];
510        let mut writer = Writer::from_buffer(&mut bytes);
511        writer.write_var_u32(0);
512        writer.write_bytes(&type_def);
513        let mut reader = Reader::new(&bytes);
514        resolver
515            .read_type_meta(&mut reader, &TypeResolver::default(), &config)
516            .unwrap();
517
518        let changed = TypeMeta::new(
519            TypeId::STRUCT as u32,
520            9001,
521            MetaString::get_empty().clone(),
522            MetaString::get_empty().clone(),
523            false,
524            vec![FieldInfo::new(
525                "b",
526                FieldType::new(crate::type_id::INT32, false, vec![]),
527            )],
528        )
529        .unwrap();
530        let mut bytes = vec![];
531        let mut writer = Writer::from_buffer(&mut bytes);
532        writer.write_var_u32(0);
533        writer.write_bytes(changed.get_bytes());
534        let mut reader = Reader::new(&bytes);
535        let err = resolver
536            .read_type_meta(&mut reader, &TypeResolver::default(), &config)
537            .unwrap_err()
538            .to_string();
539        assert!(err.contains("max_schema_versions_per_type"));
540    }
541
542    #[test]
543    fn schema_limit_check_is_not_recorded() {
544        let config = Config {
545            max_schema_versions_per_type: 1,
546            ..Default::default()
547        };
548        let mut resolver = MetaReaderResolver::default();
549        let checked = TypeMeta::new(
550            TypeId::STRUCT as u32,
551            9001,
552            MetaString::get_empty().clone(),
553            MetaString::get_empty().clone(),
554            false,
555            vec![FieldInfo::new(
556                "a",
557                FieldType::new(crate::type_id::INT32, false, vec![]),
558            )],
559        )
560        .unwrap();
561        let accepted = TypeMeta::new(
562            TypeId::STRUCT as u32,
563            9001,
564            MetaString::get_empty().clone(),
565            MetaString::get_empty().clone(),
566            false,
567            vec![FieldInfo::new(
568                "b",
569                FieldType::new(crate::type_id::INT32, false, vec![]),
570            )],
571        )
572        .unwrap();
573
574        resolver
575            .check_remote_type_meta_limit(&checked, &config)
576            .unwrap();
577
578        let mut bytes = vec![];
579        let mut writer = Writer::from_buffer(&mut bytes);
580        writer.write_var_u32(0);
581        writer.write_bytes(accepted.get_bytes());
582        let mut reader = Reader::new(&bytes);
583        resolver
584            .read_type_meta(&mut reader, &TypeResolver::default(), &config)
585            .unwrap();
586    }
587
588    #[test]
589    fn non_struct_type_meta_uses_limit() {
590        let config = Config {
591            max_schema_versions_per_type: 1,
592            ..Default::default()
593        };
594        let mut resolver = MetaReaderResolver::default();
595        let namespace = NAMESPACE_ENCODER
596            .encode_with_encodings("example", NAMESPACE_ENCODINGS)
597            .unwrap();
598        let type_name = TYPE_NAME_ENCODER
599            .encode_with_encodings("RemoteEnum", TYPE_NAME_ENCODINGS)
600            .unwrap();
601        let first = TypeMeta::new(
602            TypeId::NAMED_ENUM as u32,
603            NO_USER_TYPE_ID,
604            namespace.clone(),
605            type_name.clone(),
606            true,
607            vec![],
608        )
609        .unwrap();
610        let second = TypeMeta::new(
611            TypeId::NAMED_EXT as u32,
612            NO_USER_TYPE_ID,
613            namespace,
614            type_name,
615            true,
616            vec![],
617        )
618        .unwrap();
619
620        let key = resolver
621            .check_remote_type_meta_limit(&first, &config)
622            .unwrap();
623        resolver.record_remote_type_meta(key);
624
625        let err = resolver
626            .check_remote_type_meta_limit(&second, &config)
627            .unwrap_err()
628            .to_string();
629        assert!(err.contains("max_schema_versions_per_type"));
630    }
631
632    #[test]
633    fn exact_local_non_struct_type_meta_bypasses_limit() {
634        let config = Config {
635            max_schema_versions_per_type: 1,
636            ..Default::default()
637        };
638        let mut type_resolver = TypeResolver::default();
639        type_resolver
640            .register_serializer_by_name::<LocalExt>("example.SharedExt")
641            .unwrap();
642        let type_resolver = type_resolver.build_final_type_resolver().unwrap();
643        let local_info = type_resolver
644            .get_type_info_by_name("example", "SharedExt")
645            .unwrap();
646        let exact = local_info.get_type_meta_ref().get_bytes().to_vec();
647
648        let mut resolver = MetaReaderResolver::default();
649        read_type_def_with_type_resolver(&mut resolver, &config, &type_resolver, &exact).unwrap();
650
651        let namespace = NAMESPACE_ENCODER
652            .encode_with_encodings("example", NAMESPACE_ENCODINGS)
653            .unwrap();
654        let type_name = TYPE_NAME_ENCODER
655            .encode_with_encodings("SharedExt", TYPE_NAME_ENCODINGS)
656            .unwrap();
657        let second = TypeMeta::new(
658            TypeId::NAMED_ENUM as u32,
659            NO_USER_TYPE_ID,
660            namespace,
661            type_name,
662            true,
663            vec![],
664        )
665        .unwrap();
666        resolver
667            .check_remote_type_meta_limit(&second, &config)
668            .unwrap();
669    }
670}
fory_core/resolver/meta_resolver.rs

fory_core/resolver/
meta_resolver.rs
