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    type_id_index_map: HashMap<std::any::TypeId, usize>,
34    type_index_index_map: Vec<usize>,
35    next_index: usize,
36}
37
38const MIN_REMOTE_TYPE_META_LIMIT: usize = 8192;
39const NO_WRITTEN_TYPE_INDEX: usize = usize::MAX;
40
41#[allow(dead_code)]
42impl MetaWriterResolver {
43    /// Write type meta inline using streaming protocol.
44    /// Returns the index assigned to this type.
45    #[inline(always)]
46    pub fn write_type_meta(
47        &mut self,
48        writer: &mut Writer,
49        type_id: std::any::TypeId,
50        type_resolver: &TypeResolver,
51    ) -> Result<(), Error> {
52        match self.type_id_index_map.get(&type_id) {
53            Some(&index) => {
54                // Reference to previously written type: (index << 1) | 1, LSB=1
55                writer.write_var_u32(((index as u32) << 1) | 1);
56            }
57            None => {
58                // New type: index << 1, LSB=0, followed by TypeMeta bytes inline
59                let index = self.next_index;
60                self.next_index += 1;
61                writer.write_var_u32((index as u32) << 1);
62                self.type_id_index_map.insert(type_id, index);
63                // Write TypeMeta bytes inline
64                let type_def = type_resolver.get_type_info(&type_id)?.get_type_def();
65                writer.write_bytes(&type_def);
66            }
67        }
68        Ok(())
69    }
70
71    /// Write type meta by generated struct type index, avoiding Rust TypeId hash lookup.
72    #[inline(always)]
73    pub fn write_type_meta_fast(
74        &mut self,
75        writer: &mut Writer,
76        type_id: std::any::TypeId,
77        type_index: u32,
78        type_resolver: &TypeResolver,
79    ) -> Result<(), Error> {
80        let type_index = type_index as usize;
81        if let Some(&index) = self.type_index_index_map.get(type_index) {
82            if index != NO_WRITTEN_TYPE_INDEX {
83                writer.write_var_u32(((index as u32) << 1) | 1);
84                return Ok(());
85            }
86        }
87
88        let index = self.next_index;
89        self.next_index += 1;
90        writer.write_var_u32((index as u32) << 1);
91        if type_index >= self.type_index_index_map.len() {
92            self.type_index_index_map
93                .resize(type_index + 1, NO_WRITTEN_TYPE_INDEX);
94        }
95        self.type_index_index_map[type_index] = index;
96        let type_meta = type_resolver.get_type_meta_by_index_ref(&type_id, type_index as u32)?;
97        writer.write_bytes(type_meta.get_bytes());
98        Ok(())
99    }
100
101    #[inline(always)]
102    pub fn reset(&mut self) {
103        self.type_id_index_map.clear();
104        self.type_index_index_map.clear();
105        self.next_index = 0;
106    }
107}
108
109/// Streaming meta reader that reads TypeMeta inline during deserialization.
110/// Uses the streaming protocol:
111/// - (index << 1) | 0 for new type definition (followed by TypeMeta bytes)
112/// - (index << 1) | 1 for reference to previously read type
113#[derive(Default)]
114pub struct MetaReaderResolver {
115    pub reading_type_infos: Vec<Rc<TypeInfo>>,
116    parsed_type_infos: HashMap<i64, Rc<TypeInfo>>,
117    remote_schema_versions_by_type: HashMap<String, usize>,
118    total_accepted_schema_versions: usize,
119    cached_meta_header: i64,
120    cached_type_info: Option<Rc<TypeInfo>>,
121}
122
123impl MetaReaderResolver {
124    #[inline(always)]
125    pub fn get(&self, index: usize) -> Option<&Rc<TypeInfo>> {
126        self.reading_type_infos.get(index)
127    }
128
129    /// Read type meta inline using streaming protocol.
130    /// Returns the TypeInfo for this type.
131    #[inline(always)]
132    pub fn read_type_meta(
133        &mut self,
134        reader: &mut Reader,
135        type_resolver: &TypeResolver,
136        config: &Config,
137    ) -> Result<Rc<TypeInfo>, Error> {
138        let index_marker = reader.read_var_u32()?;
139        let is_ref = (index_marker & 1) == 1;
140        let index = (index_marker >> 1) as usize;
141
142        if is_ref {
143            // Reference to previously read type
144            self.reading_type_infos.get(index).cloned().ok_or_else(|| {
145                Error::type_error(format!("TypeInfo not found for type index: {}", index))
146            })
147        } else {
148            // New type - read TypeMeta inline
149            let meta_header = reader.read_i64()?;
150            if let Some(type_info) = self
151                .cached_type_info
152                .as_ref()
153                .filter(|_| self.cached_meta_header == meta_header)
154            {
155                // Header-cache hits intentionally skip without rehashing. Entries reach this cache
156                // only after a successful TypeMeta parse and 52-bit metadata-hash validation. Do
157                // not add body/hash/schema-limit/exact-local checks here; the miss path owns them
158                // before publish.
159                self.reading_type_infos.push(type_info.clone());
160                TypeMeta::skip_bytes_for_validated_header(reader, meta_header)?;
161                return Ok(type_info.clone());
162            }
163            if let Some(type_info) = self.parsed_type_infos.get(&meta_header) {
164                // Header-cache hits intentionally skip without rehashing. Entries reach this cache
165                // only after a successful TypeMeta parse and 52-bit metadata-hash validation. Do
166                // not add body/hash/schema-limit/exact-local checks here; the miss path owns them
167                // before publish.
168                self.cached_meta_header = meta_header;
169                self.cached_type_info = Some(type_info.clone());
170                self.reading_type_infos.push(type_info.clone());
171                TypeMeta::skip_bytes_for_validated_header(reader, meta_header)?;
172                Ok(type_info.clone())
173            } else {
174                let type_def_start = reader.get_cursor() - std::mem::size_of::<i64>();
175                self.read_remote_type_meta(
176                    reader,
177                    type_resolver,
178                    config,
179                    meta_header,
180                    type_def_start,
181                )
182            }
183        }
184    }
185
186    #[cold]
187    #[inline(never)]
188    fn read_remote_type_meta(
189        &mut self,
190        reader: &mut Reader,
191        type_resolver: &TypeResolver,
192        config: &Config,
193        meta_header: i64,
194        type_def_start: usize,
195    ) -> Result<Rc<TypeInfo>, Error> {
196        let type_meta = Rc::new(TypeMeta::from_bytes_with_header(
197            reader,
198            type_resolver,
199            meta_header,
200            config.max_type_fields(),
201            config.max_type_meta_bytes(),
202        )?);
203        let remote_type_def = reader.sub_slice(type_def_start, reader.get_cursor())?;
204
205        let namespace = type_meta.get_namespace();
206        let type_name = type_meta.get_type_name();
207        let register_by_name = !namespace.original.is_empty() || !type_name.original.is_empty();
208        let mut remote_schema_key = None;
209        let type_info = if register_by_name {
210            if let Some(local_type_info) =
211                type_resolver.get_type_info_by_name(&namespace.original, &type_name.original)
212            {
213                if local_type_info.get_type_meta_ref().get_bytes() == remote_type_def {
214                    local_type_info
215                } else {
216                    remote_schema_key =
217                        Some(self.check_remote_type_meta_limit(&type_meta, config)?);
218                    Rc::new(TypeInfo::from_remote_meta(
219                        type_meta.clone(),
220                        Some(local_type_info.get_harness()),
221                        Some(local_type_info.get_type_id() as u32),
222                        Some(local_type_info.get_user_type_id()),
223                    ))
224                }
225            } else {
226                remote_schema_key = Some(self.check_remote_type_meta_limit(&type_meta, config)?);
227                Rc::new(TypeInfo::from_remote_meta(
228                    type_meta.clone(),
229                    None,
230                    None,
231                    None,
232                ))
233            }
234        } else {
235            let type_id = type_meta.get_type_id();
236            let user_type_id = type_meta.get_user_type_id();
237            let local_type_info = if user_type_id != NO_USER_TYPE_ID {
238                type_resolver.get_user_type_info_by_id(user_type_id)
239            } else {
240                type_resolver.get_type_info_by_id(type_id)
241            };
242            if let Some(local_type_info) = local_type_info {
243                if local_type_info.get_type_meta_ref().get_bytes() == remote_type_def {
244                    local_type_info
245                } else {
246                    remote_schema_key =
247                        Some(self.check_remote_type_meta_limit(&type_meta, config)?);
248                    Rc::new(TypeInfo::from_remote_meta(
249                        type_meta.clone(),
250                        Some(local_type_info.get_harness()),
251                        Some(local_type_info.get_type_id() as u32),
252                        Some(local_type_info.get_user_type_id()),
253                    ))
254                }
255            } else {
256                remote_schema_key = Some(self.check_remote_type_meta_limit(&type_meta, config)?);
257                Rc::new(TypeInfo::from_remote_meta(
258                    type_meta.clone(),
259                    None,
260                    None,
261                    None,
262                ))
263            }
264        };
265
266        self.parsed_type_infos
267            .insert(meta_header, type_info.clone());
268        self.cached_meta_header = meta_header;
269        self.cached_type_info = Some(type_info.clone());
270        self.reading_type_infos.push(type_info.clone());
271        if let Some(remote_schema_key) = remote_schema_key {
272            self.record_remote_type_meta(remote_schema_key);
273        }
274        Ok(type_info)
275    }
276
277    #[cold]
278    #[inline(never)]
279    fn check_remote_type_meta_limit(
280        &self,
281        type_meta: &TypeMeta,
282        config: &Config,
283    ) -> Result<String, Error> {
284        let namespace = type_meta.get_namespace();
285        let type_name = type_meta.get_type_name();
286        let key = if !namespace.original.is_empty() || !type_name.original.is_empty() {
287            format!("n{}\0{}", namespace.original, type_name.original)
288        } else {
289            format!("i{}", type_meta.get_user_type_id())
290        };
291
292        let versions_for_type = self
293            .remote_schema_versions_by_type
294            .get(&key)
295            .copied()
296            .unwrap_or(0);
297        if versions_for_type >= config.max_schema_versions_per_type() {
298            return Err(Error::invalid_data(format!(
299                "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={}",
300                config.max_schema_versions_per_type()
301            )));
302        }
303
304        let accepted_type_count =
305            self.remote_schema_versions_by_type.len() + if versions_for_type == 0 { 1 } else { 0 };
306        let global_limit = usize::max(
307            MIN_REMOTE_TYPE_META_LIMIT,
308            accepted_type_count * config.max_average_schema_versions_per_type(),
309        );
310        if self.total_accepted_schema_versions >= global_limit {
311            return Err(Error::invalid_data(format!(
312                "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={}",
313                config.max_average_schema_versions_per_type()
314            )));
315        }
316
317        Ok(key)
318    }
319
320    fn record_remote_type_meta(&mut self, key: String) {
321        let versions_for_type = self
322            .remote_schema_versions_by_type
323            .get(&key)
324            .copied()
325            .unwrap_or(0);
326        self.remote_schema_versions_by_type
327            .insert(key, versions_for_type + 1);
328        self.total_accepted_schema_versions += 1;
329    }
330
331    #[inline(always)]
332    pub fn reset(&mut self) {
333        self.reading_type_infos.clear();
334    }
335}
336
337#[cfg(test)]
338mod tests {
339    use super::*;
340    use crate::config::Config;
341    use crate::context::{ReadContext, WriteContext};
342    use crate::meta::{
343        FieldInfo, FieldType, MetaString, NAMESPACE_ENCODER, NAMESPACE_ENCODINGS,
344        TYPE_NAME_ENCODER, TYPE_NAME_ENCODINGS,
345    };
346    use crate::serializer::{ForyDefault, Serializer};
347    use crate::TypeId;
348
349    struct LocalExt;
350
351    impl ForyDefault for LocalExt {
352        fn fory_default() -> Self {
353            LocalExt
354        }
355    }
356
357    impl Serializer for LocalExt {
358        fn fory_write_data(&self, _context: &mut WriteContext) -> Result<(), Error> {
359            Ok(())
360        }
361
362        fn fory_read_data(_context: &mut ReadContext) -> Result<Self, Error> {
363            Ok(LocalExt)
364        }
365
366        fn fory_type_id_dyn(&self, type_resolver: &TypeResolver) -> Result<TypeId, Error> {
367            Self::fory_get_type_id(type_resolver)
368        }
369
370        fn as_any(&self) -> &dyn std::any::Any {
371            self
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
