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_VERSIONS: u64 = 8192;
41const MAX_REMOTE_TYPE_META_KEYS: usize = 8192;
42const NO_WRITTEN_TYPE_INDEX: usize = usize::MAX;
43
44#[allow(dead_code)]
45impl MetaWriterResolver {
46    /// Write type meta inline using streaming protocol.
47    /// Returns the index assigned to this type.
48    #[inline(always)]
49    pub fn write_type_meta(
50        &mut self,
51        writer: &mut Writer,
52        provider_type_id: std::any::TypeId,
53        type_resolver: &TypeResolver,
54    ) -> Result<(), Error> {
55        let type_info = type_resolver.get_provider_type_info(&provider_type_id)?;
56        self.write_resolved_type_meta(writer, &type_info)
57    }
58
59    #[inline(always)]
60    pub(crate) fn write_resolved_type_meta(
61        &mut self,
62        writer: &mut Writer,
63        type_info: &Rc<TypeInfo>,
64    ) -> Result<(), Error> {
65        let identity = Rc::as_ptr(type_info);
66        match self.type_info_index_map.get(&identity) {
67            Some(&index) => {
68                // Reference to previously written type: (index << 1) | 1, LSB=1
69                writer.write_var_u32(((index as u32) << 1) | 1);
70            }
71            None => {
72                // New type: index << 1, LSB=0, followed by TypeMeta bytes inline
73                let index = self.next_index;
74                self.next_index += 1;
75                writer.write_var_u32((index as u32) << 1);
76                self.type_info_index_map.insert(identity, index);
77                let type_def = type_info.get_type_def();
78                writer.write_bytes(&type_def);
79            }
80        }
81        Ok(())
82    }
83
84    /// Write type meta by generated struct type index, avoiding Rust TypeId hash lookup.
85    #[inline(always)]
86    pub fn write_type_meta_fast(
87        &mut self,
88        writer: &mut Writer,
89        type_id: std::any::TypeId,
90        type_index: u32,
91        type_resolver: &TypeResolver,
92    ) -> Result<(), Error> {
93        let type_index = type_index as usize;
94        if let Some(&index) = self.type_index_index_map.get(type_index) {
95            if index != NO_WRITTEN_TYPE_INDEX {
96                writer.write_var_u32(((index as u32) << 1) | 1);
97                return Ok(());
98            }
99        }
100
101        let index = self.next_index;
102        self.next_index += 1;
103        writer.write_var_u32((index as u32) << 1);
104        if type_index >= self.type_index_index_map.len() {
105            self.type_index_index_map
106                .resize(type_index + 1, NO_WRITTEN_TYPE_INDEX);
107        }
108        self.type_index_index_map[type_index] = index;
109        let type_meta = type_resolver.get_type_meta_by_index_ref(&type_id, type_index as u32)?;
110        writer.write_bytes(type_meta.get_bytes());
111        Ok(())
112    }
113
114    #[inline(always)]
115    pub fn reset(&mut self) {
116        self.type_info_index_map.clear();
117        self.type_index_index_map.clear();
118        self.next_index = 0;
119    }
120}
121
122/// Streaming meta reader that reads TypeMeta inline during deserialization.
123/// Uses the streaming protocol:
124/// - (index << 1) | 0 for new type definition (followed by TypeMeta bytes)
125/// - (index << 1) | 1 for reference to previously read type
126#[derive(Default)]
127pub struct MetaReaderResolver {
128    pub reading_type_infos: Vec<Rc<TypeInfo>>,
129    parsed_type_infos: HashMap<i64, Rc<TypeInfo>>,
130    remote_schema_versions_by_type: HashMap<String, usize>,
131    total_accepted_schema_versions: u64,
132    cached_meta_hash: i64,
133    cached_type_info: Option<Rc<TypeInfo>>,
134}
135
136#[derive(Clone, Copy)]
137enum TypeInfoExpectation<'a> {
138    Any,
139    Exact(&'a Rc<TypeInfo>),
140    Structural(&'a Rc<TypeInfo>),
141}
142
143impl<'a> TypeInfoExpectation<'a> {
144    #[inline(always)]
145    fn local(self) -> Option<&'a Rc<TypeInfo>> {
146        match self {
147            Self::Any => None,
148            Self::Exact(type_info) | Self::Structural(type_info) => Some(type_info),
149        }
150    }
151
152    #[inline(always)]
153    fn allows_stub(self) -> bool {
154        matches!(self, Self::Structural(_))
155    }
156}
157
158impl MetaReaderResolver {
159    #[inline(always)]
160    pub fn get(&self, index: usize) -> Option<&Rc<TypeInfo>> {
161        self.reading_type_infos.get(index)
162    }
163
164    /// Read type meta inline using streaming protocol.
165    /// Returns the TypeInfo for this type.
166    #[inline(always)]
167    pub fn read_type_meta(
168        &mut self,
169        reader: &mut Reader,
170        type_resolver: &TypeResolver,
171        config: &Config,
172    ) -> Result<Rc<TypeInfo>, Error> {
173        self.read_type_meta_with_expected(reader, type_resolver, config, TypeInfoExpectation::Any)
174    }
175
176    #[inline(always)]
177    pub(crate) fn read_type_meta_for(
178        &mut self,
179        reader: &mut Reader,
180        type_resolver: &TypeResolver,
181        config: &Config,
182        expected: &Rc<TypeInfo>,
183    ) -> Result<Rc<TypeInfo>, Error> {
184        self.read_type_meta_with_expected(
185            reader,
186            type_resolver,
187            config,
188            TypeInfoExpectation::Exact(expected),
189        )
190    }
191
192    #[inline(always)]
193    pub(crate) fn read_struct_type_meta_for(
194        &mut self,
195        reader: &mut Reader,
196        type_resolver: &TypeResolver,
197        config: &Config,
198        expected: &Rc<TypeInfo>,
199    ) -> Result<Rc<TypeInfo>, Error> {
200        self.read_type_meta_with_expected(
201            reader,
202            type_resolver,
203            config,
204            TypeInfoExpectation::Structural(expected),
205        )
206    }
207
208    #[inline(always)]
209    fn read_type_meta_with_expected(
210        &mut self,
211        reader: &mut Reader,
212        type_resolver: &TypeResolver,
213        config: &Config,
214        expected: TypeInfoExpectation<'_>,
215    ) -> Result<Rc<TypeInfo>, Error> {
216        let index_marker = reader.read_var_u32()?;
217        let is_ref = (index_marker & 1) == 1;
218        let index = (index_marker >> 1) as usize;
219
220        if is_ref {
221            // Reference to previously read type
222            let type_info = self.reading_type_infos.get(index).cloned().ok_or_else(|| {
223                Error::type_error(format!("TypeInfo not found for type index: {}", index))
224            })?;
225            Self::check_expected_owner(&type_info, expected)?;
226            Ok(type_info)
227        } else {
228            // New type - read TypeMeta inline
229            let meta_header = reader.read_i64()?;
230            let meta_hash = TypeMeta::header_hash(meta_header);
231            if let Some(type_info) = expected
232                .local()
233                .filter(|type_info| type_info.get_type_meta_ref().get_hash() == meta_hash)
234                .cloned()
235            {
236                // A statically expected local owner has priority over remote checked-cache hints.
237                // The top-52 hash is the schema identity; current low bits only bound this skip.
238                TypeMeta::skip_bytes_for_validated_header(reader, meta_header)?;
239                self.reading_type_infos.push(type_info.clone());
240                return Ok(type_info);
241            }
242            if let Some(type_info) = self
243                .cached_type_info
244                .as_ref()
245                .filter(|_| self.cached_meta_hash == meta_hash)
246                .cloned()
247            {
248                // The 52-bit header hash is the schema identity. Low header bits describe only
249                // this frame, so a checked hit uses them solely to skip its opaque body.
250                Self::check_expected_owner(&type_info, expected)?;
251                TypeMeta::skip_bytes_for_validated_header(reader, meta_header)?;
252                self.reading_type_infos.push(type_info.clone());
253                return Ok(type_info);
254            }
255            if let Some(type_info) = self.parsed_type_infos.get(&meta_hash).cloned() {
256                // Entries reach this cache only after successful TypeMeta parse, body-hash
257                // validation, policy checks, and publication on the miss path.
258                Self::check_expected_owner(&type_info, expected)?;
259                TypeMeta::skip_bytes_for_validated_header(reader, meta_header)?;
260                self.cached_meta_hash = meta_hash;
261                self.cached_type_info = Some(type_info.clone());
262                self.reading_type_infos.push(type_info.clone());
263                Ok(type_info)
264            } else {
265                self.read_remote_type_meta(
266                    reader,
267                    type_resolver,
268                    config,
269                    meta_header,
270                    meta_hash,
271                    expected,
272                )
273            }
274        }
275    }
276
277    #[inline(always)]
278    fn check_expected_owner(
279        type_info: &TypeInfo,
280        expected: TypeInfoExpectation<'_>,
281    ) -> Result<(), Error> {
282        let Some(expected_type_info) = expected.local() else {
283            return Ok(());
284        };
285        let expected_target = expected_type_info
286            .get_harness()
287            .target_type_id()
288            .ok_or_else(|| Error::type_error("expected TypeInfo has no concrete target"))?;
289        let resolved_target = type_info.get_harness().target_type_id();
290        if expected.allows_stub()
291            && !crate::type_id::is_struct_type_id(type_info.get_type_meta_ref().get_type_id())
292        {
293            return Err(Self::type_info_kind_mismatch(
294                type_info.get_type_meta_ref().get_type_id(),
295            ));
296        }
297        if resolved_target != Some(expected_target)
298            && !(expected.allows_stub() && resolved_target.is_none())
299        {
300            return Err(Self::type_info_owner_mismatch(
301                resolved_target,
302                expected_target,
303            ));
304        }
305        Ok(())
306    }
307
308    #[cold]
309    #[inline(never)]
310    fn type_info_owner_mismatch(
311        resolved: Option<std::any::TypeId>,
312        expected: std::any::TypeId,
313    ) -> Error {
314        Error::type_error(format!(
315            "resolved TypeInfo target {:?} does not match declared target {:?}",
316            resolved, expected,
317        ))
318    }
319
320    #[cold]
321    #[inline(never)]
322    fn type_info_kind_mismatch(resolved: u32) -> Error {
323        Error::type_error(format!(
324            "resolved TypeInfo wire kind {} is not structural metadata",
325            resolved,
326        ))
327    }
328
329    #[cold]
330    #[inline(never)]
331    fn read_remote_type_meta(
332        &mut self,
333        reader: &mut Reader,
334        type_resolver: &TypeResolver,
335        config: &Config,
336        meta_header: i64,
337        meta_hash: i64,
338        expected: TypeInfoExpectation<'_>,
339    ) -> Result<Rc<TypeInfo>, Error> {
340        let type_meta = Rc::new(TypeMeta::from_bytes_with_header(
341            reader,
342            type_resolver,
343            meta_header,
344            config.max_type_fields(),
345            config.max_type_meta_bytes(),
346        )?);
347
348        let namespace = type_meta.get_namespace();
349        let type_name = type_meta.get_type_name();
350        let register_by_name = !namespace.original.is_empty() || !type_name.original.is_empty();
351        let remote_schema_key;
352        // The body and its hash are validated above. From this point the top-52 hash alone decides
353        // whether a resolved local TypeMeta owns this schema; body bytes are not a second identity.
354        // A local owner is root-local reading state, not a remote checked-cache publication.
355        let type_info = if register_by_name {
356            if let Some(local_type_info) =
357                type_resolver.get_type_info_by_name(&namespace.original, &type_name.original)
358            {
359                if local_type_info.get_type_meta_ref().get_hash() == meta_hash {
360                    Self::check_expected_owner(&local_type_info, expected)?;
361                    self.reading_type_infos.push(local_type_info.clone());
362                    return Ok(local_type_info);
363                } else {
364                    remote_schema_key = self.check_remote_type_meta_limit(&type_meta, config)?;
365                    Rc::new(TypeInfo::from_remote_meta(
366                        type_meta.clone(),
367                        Some(local_type_info.get_harness()),
368                        Some(local_type_info.get_type_id() as u32),
369                        Some(local_type_info.get_user_type_id()),
370                    ))
371                }
372            } else {
373                remote_schema_key = self.check_remote_type_meta_limit(&type_meta, config)?;
374                Rc::new(TypeInfo::from_remote_meta(
375                    type_meta.clone(),
376                    None,
377                    None,
378                    None,
379                ))
380            }
381        } else {
382            let type_id = type_meta.get_type_id();
383            let user_type_id = type_meta.get_user_type_id();
384            let local_type_info = if user_type_id != NO_USER_TYPE_ID {
385                type_resolver.get_user_type_info_by_id(user_type_id)
386            } else {
387                type_resolver.get_type_info_by_id(type_id)
388            };
389            if let Some(local_type_info) = local_type_info {
390                if local_type_info.get_type_meta_ref().get_hash() == meta_hash {
391                    Self::check_expected_owner(&local_type_info, expected)?;
392                    self.reading_type_infos.push(local_type_info.clone());
393                    return Ok(local_type_info);
394                } else {
395                    remote_schema_key = self.check_remote_type_meta_limit(&type_meta, config)?;
396                    Rc::new(TypeInfo::from_remote_meta(
397                        type_meta.clone(),
398                        Some(local_type_info.get_harness()),
399                        Some(local_type_info.get_type_id() as u32),
400                        Some(local_type_info.get_user_type_id()),
401                    ))
402                }
403            } else {
404                remote_schema_key = self.check_remote_type_meta_limit(&type_meta, config)?;
405                Rc::new(TypeInfo::from_remote_meta(
406                    type_meta.clone(),
407                    None,
408                    None,
409                    None,
410                ))
411            }
412        };
413
414        // A concrete remote harness must own the declared local target. Compatible structural
415        // mapping alone may retain an unregistered stub, and this decision must precede root or
416        // persistent checked-cache publication and schema-version accounting.
417        Self::check_expected_owner(&type_info, expected)?;
418        self.parsed_type_infos.insert(meta_hash, type_info.clone());
419        self.cached_meta_hash = meta_hash;
420        self.cached_type_info = Some(type_info.clone());
421        self.reading_type_infos.push(type_info.clone());
422        self.record_remote_type_meta(remote_schema_key);
423        Ok(type_info)
424    }
425
426    #[cold]
427    #[inline(never)]
428    fn check_remote_type_meta_limit(
429        &self,
430        type_meta: &TypeMeta,
431        config: &Config,
432    ) -> Result<String, Error> {
433        let namespace = type_meta.get_namespace();
434        let type_name = type_meta.get_type_name();
435        let key = if !namespace.original.is_empty() || !type_name.original.is_empty() {
436            format!("n{}\0{}", namespace.original, type_name.original)
437        } else {
438            format!("i{}", type_meta.get_user_type_id())
439        };
440
441        let versions_for_type = self
442            .remote_schema_versions_by_type
443            .get(&key)
444            .copied()
445            .unwrap_or(0);
446        // Reaching the key cap must not disable schema evolution for keys that were already
447        // accepted.
448        if versions_for_type == 0
449            && self.remote_schema_versions_by_type.len() >= MAX_REMOTE_TYPE_META_KEYS
450        {
451            return Err(Error::invalid_data(
452                "remote logical TypeMeta key limit exceeded. The data may be malicious",
453            ));
454        }
455        if versions_for_type >= config.max_schema_versions_per_type() {
456            return Err(Error::invalid_data(format!(
457                "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={}",
458                config.max_schema_versions_per_type()
459            )));
460        }
461
462        let accepted_type_count = (self.remote_schema_versions_by_type.len()
463            + if versions_for_type == 0 { 1 } else { 0 }) as u64;
464        let max_average = config.max_average_schema_versions_per_type() as u64;
465        let reached_average_limit = max_average == 0
466            || self.total_accepted_schema_versions / max_average >= accepted_type_count;
467        if self.total_accepted_schema_versions == u64::MAX
468            || (self.total_accepted_schema_versions >= MIN_REMOTE_TYPE_META_VERSIONS
469                && reached_average_limit)
470        {
471            return Err(Error::invalid_data(format!(
472                "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={}",
473                config.max_average_schema_versions_per_type()
474            )));
475        }
476
477        Ok(key)
478    }
479
480    fn record_remote_type_meta(&mut self, key: String) {
481        let versions_for_type = self
482            .remote_schema_versions_by_type
483            .get(&key)
484            .copied()
485            .unwrap_or(0);
486        self.remote_schema_versions_by_type
487            .insert(key, versions_for_type + 1);
488        // The cold miss check rejects u64::MAX before its caller publishes the TypeInfo and reaches
489        // this mutation.
490        self.total_accepted_schema_versions += 1;
491    }
492
493    #[inline(always)]
494    pub fn reset(&mut self) {
495        self.reading_type_infos.clear();
496    }
497}
498
499#[cfg(test)]
500mod tests {
501    use super::*;
502    use crate::config::Config;
503    use crate::context::{ReadContext, WriteContext};
504    use crate::meta::{
505        FieldInfo, FieldType, MetaString, NAMESPACE_ENCODER, NAMESPACE_ENCODINGS,
506        TYPE_NAME_ENCODER, TYPE_NAME_ENCODINGS,
507    };
508    use crate::serializer::{Serializer, StructSerializer};
509    use crate::TypeId;
510
511    const LOCAL_COLLISION_FIELD: &str = "!AsA3daaa";
512    const REMOTE_COLLISION_FIELD: &str = "!aR5Ocaaaa";
513
514    struct LocalExt;
515
516    impl Serializer for LocalExt {
517        type Target = Self;
518
519        fn write_data(_value: &Self, _context: &mut WriteContext) -> Result<(), Error> {
520            Ok(())
521        }
522
523        fn read_data(_context: &mut ReadContext) -> Result<Self, Error> {
524            Ok(LocalExt)
525        }
526    }
527
528    struct CollisionStruct;
529
530    impl Serializer for CollisionStruct {
531        type Target = Self;
532
533        fn write_data(_value: &Self, _context: &mut WriteContext) -> Result<(), Error> {
534            Ok(())
535        }
536
537        fn read_data(_context: &mut ReadContext) -> Result<Self, Error> {
538            Ok(CollisionStruct)
539        }
540
541        fn static_type_id() -> TypeId {
542            TypeId::STRUCT
543        }
544    }
545
546    impl StructSerializer for CollisionStruct {
547        fn type_index() -> u32 {
548            9001
549        }
550
551        fn fields_info(_type_resolver: &TypeResolver) -> Result<Vec<FieldInfo>, Error> {
552            Ok(vec![FieldInfo::new(
553                LOCAL_COLLISION_FIELD,
554                FieldType::new(crate::type_id::INT32, false, vec![]),
555            )])
556        }
557
558        fn variants_fields_info(
559            _type_resolver: &TypeResolver,
560        ) -> Result<Vec<(String, std::any::TypeId, Vec<FieldInfo>)>, Error> {
561            Ok(vec![])
562        }
563
564        fn sorted_field_names() -> &'static [&'static str] {
565            &[LOCAL_COLLISION_FIELD]
566        }
567
568        fn read_compatible(
569            _context: &mut ReadContext,
570            _type_info: &Rc<TypeInfo>,
571        ) -> Result<Self, Error> {
572            Ok(CollisionStruct)
573        }
574    }
575
576    struct ForeignStruct;
577
578    impl Serializer for ForeignStruct {
579        type Target = Self;
580
581        fn write_data(_value: &Self, _context: &mut WriteContext) -> Result<(), Error> {
582            Ok(())
583        }
584
585        fn read_data(_context: &mut ReadContext) -> Result<Self, Error> {
586            Ok(ForeignStruct)
587        }
588
589        fn static_type_id() -> TypeId {
590            TypeId::STRUCT
591        }
592    }
593
594    impl StructSerializer for ForeignStruct {
595        fn type_index() -> u32 {
596            9002
597        }
598
599        fn fields_info(_type_resolver: &TypeResolver) -> Result<Vec<FieldInfo>, Error> {
600            Ok(vec![FieldInfo::new(
601                "local_b",
602                FieldType::new(crate::type_id::INT32, false, vec![]),
603            )])
604        }
605
606        fn variants_fields_info(
607            _type_resolver: &TypeResolver,
608        ) -> Result<Vec<(String, std::any::TypeId, Vec<FieldInfo>)>, Error> {
609            Ok(vec![])
610        }
611
612        fn sorted_field_names() -> &'static [&'static str] {
613            &["local_b"]
614        }
615
616        fn read_compatible(
617            _context: &mut ReadContext,
618            _type_info: &Rc<TypeInfo>,
619        ) -> Result<Self, Error> {
620            Ok(ForeignStruct)
621        }
622    }
623
624    fn read_type_def(
625        resolver: &mut MetaReaderResolver,
626        config: &Config,
627        type_def: &[u8],
628    ) -> Result<Rc<TypeInfo>, Error> {
629        let type_resolver = TypeResolver::default();
630        read_type_def_with_type_resolver(resolver, config, &type_resolver, type_def)
631    }
632
633    fn read_type_def_with_type_resolver(
634        resolver: &mut MetaReaderResolver,
635        config: &Config,
636        type_resolver: &TypeResolver,
637        type_def: &[u8],
638    ) -> Result<Rc<TypeInfo>, Error> {
639        let mut bytes = vec![];
640        let mut writer = Writer::from_buffer(&mut bytes);
641        writer.write_var_u32(0);
642        writer.write_bytes(type_def);
643        let mut reader = Reader::new(&bytes);
644        resolver.read_type_meta(&mut reader, type_resolver, config)
645    }
646
647    fn remote_struct_meta(user_type_id: u32, field_name: &str) -> TypeMeta {
648        TypeMeta::new(
649            TypeId::STRUCT as u32,
650            user_type_id,
651            MetaString::get_empty().clone(),
652            MetaString::get_empty().clone(),
653            false,
654            vec![FieldInfo::new(
655                field_name,
656                FieldType::new(crate::type_id::INT32, false, vec![]),
657            )],
658        )
659        .unwrap()
660    }
661
662    fn named_struct_meta(namespace: &str, type_name: &str, field_name: &str) -> TypeMeta {
663        named_meta(
664            TypeId::NAMED_STRUCT,
665            namespace,
666            type_name,
667            vec![FieldInfo::new(
668                field_name,
669                FieldType::new(crate::type_id::INT32, false, vec![]),
670            )],
671        )
672    }
673
674    fn named_meta(
675        type_id: TypeId,
676        namespace: &str,
677        type_name: &str,
678        fields: Vec<FieldInfo>,
679    ) -> TypeMeta {
680        TypeMeta::new(
681            type_id as u32,
682            NO_USER_TYPE_ID,
683            NAMESPACE_ENCODER
684                .encode_with_encodings(namespace, NAMESPACE_ENCODINGS)
685                .unwrap(),
686            TYPE_NAME_ENCODER
687                .encode_with_encodings(type_name, TYPE_NAME_ENCODINGS)
688                .unwrap(),
689            true,
690            fields,
691        )
692        .unwrap()
693    }
694
695    fn read_type_def_for(
696        resolver: &mut MetaReaderResolver,
697        config: &Config,
698        type_resolver: &TypeResolver,
699        expected: &Rc<TypeInfo>,
700        type_def: &[u8],
701    ) -> Result<Rc<TypeInfo>, Error> {
702        let mut bytes = vec![];
703        let mut writer = Writer::from_buffer(&mut bytes);
704        writer.write_var_u32(0);
705        writer.write_bytes(type_def);
706        let mut reader = Reader::new(&bytes);
707        resolver.read_type_meta_for(&mut reader, type_resolver, config, expected)
708    }
709
710    fn read_struct_type_def_for(
711        resolver: &mut MetaReaderResolver,
712        config: &Config,
713        type_resolver: &TypeResolver,
714        expected: &Rc<TypeInfo>,
715        type_def: &[u8],
716    ) -> Result<Rc<TypeInfo>, Error> {
717        let mut bytes = vec![];
718        let mut writer = Writer::from_buffer(&mut bytes);
719        writer.write_var_u32(0);
720        writer.write_bytes(type_def);
721        let mut reader = Reader::new(&bytes);
722        resolver.read_struct_type_meta_for(&mut reader, type_resolver, config, expected)
723    }
724
725    fn type_def_frame(meta_hash: i64, flags: i64, body_size: usize, fill: u8) -> Vec<u8> {
726        assert_eq!(flags & !0xf00, 0);
727        let inline_size = body_size.min(0xff);
728        let header = (((meta_hash as u64) << 12) | flags as u64 | inline_size as u64) as i64;
729        let mut type_def = vec![];
730        let mut writer = Writer::from_buffer(&mut type_def);
731        writer.write_i64(header);
732        if body_size >= 0xff {
733            writer.write_var_u32(u32::try_from(body_size - 0xff).unwrap());
734        }
735        writer.write_bytes(&vec![fill; body_size]);
736        type_def
737    }
738
739    fn read_type_def_with_cursor(
740        resolver: &mut MetaReaderResolver,
741        config: &Config,
742        type_def: &[u8],
743    ) -> (Result<Rc<TypeInfo>, Error>, usize, usize) {
744        let mut bytes = vec![];
745        let mut writer = Writer::from_buffer(&mut bytes);
746        writer.write_var_u32(0);
747        writer.write_bytes(type_def);
748        let mut reader = Reader::new(&bytes);
749        let result = resolver.read_type_meta(&mut reader, &TypeResolver::default(), config);
750        (result, reader.get_cursor(), bytes.len())
751    }
752
753    fn fill_remote_schema_keys(resolver: &mut MetaReaderResolver, count: usize, versions: usize) {
754        assert!(count <= MAX_REMOTE_TYPE_META_KEYS);
755        for user_type_id in 0..count {
756            resolver
757                .remote_schema_versions_by_type
758                .insert(format!("i{user_type_id}"), versions);
759        }
760        resolver.total_accepted_schema_versions = count as u64 * versions as u64;
761    }
762
763    #[test]
764    fn logical_type_key_cap() {
765        let config = Config::default();
766        let mut resolver = MetaReaderResolver::default();
767        fill_remote_schema_keys(&mut resolver, MAX_REMOTE_TYPE_META_KEYS - 1, 1);
768
769        let last = remote_struct_meta((MAX_REMOTE_TYPE_META_KEYS - 1) as u32, "a");
770        read_type_def(&mut resolver, &config, last.get_bytes()).unwrap();
771        assert_eq!(
772            resolver.remote_schema_versions_by_type.len(),
773            MAX_REMOTE_TYPE_META_KEYS
774        );
775        assert_eq!(
776            resolver.total_accepted_schema_versions,
777            MAX_REMOTE_TYPE_META_KEYS as u64
778        );
779
780        let parsed_count = resolver.parsed_type_infos.len();
781        let reading_count = resolver.reading_type_infos.len();
782        let cached_hash = resolver.cached_meta_hash;
783        let cached_type_info = resolver.cached_type_info.as_ref().map(Rc::as_ptr);
784        let rejected = remote_struct_meta(MAX_REMOTE_TYPE_META_KEYS as u32, "a");
785        let err = read_type_def(&mut resolver, &config, rejected.get_bytes())
786            .unwrap_err()
787            .to_string();
788
789        assert!(err.contains("logical TypeMeta key limit"));
790        assert_eq!(
791            resolver.remote_schema_versions_by_type.len(),
792            MAX_REMOTE_TYPE_META_KEYS
793        );
794        assert_eq!(
795            resolver.total_accepted_schema_versions,
796            MAX_REMOTE_TYPE_META_KEYS as u64
797        );
798        assert_eq!(resolver.parsed_type_infos.len(), parsed_count);
799        assert_eq!(resolver.reading_type_infos.len(), reading_count);
800        assert_eq!(resolver.cached_meta_hash, cached_hash);
801        assert_eq!(
802            resolver.cached_type_info.as_ref().map(Rc::as_ptr),
803            cached_type_info
804        );
805    }
806
807    #[test]
808    fn existing_key_keeps_limits() {
809        let mut per_type_resolver = MetaReaderResolver::default();
810        fill_remote_schema_keys(&mut per_type_resolver, MAX_REMOTE_TYPE_META_KEYS, 1);
811        let per_type_config = Config {
812            max_schema_versions_per_type: 1,
813            ..Default::default()
814        };
815        let changed = remote_struct_meta(0, "b");
816        let err = read_type_def(
817            &mut per_type_resolver,
818            &per_type_config,
819            changed.get_bytes(),
820        )
821        .unwrap_err()
822        .to_string();
823        assert!(err.contains("max_schema_versions_per_type"));
824
825        let mut average_resolver = MetaReaderResolver::default();
826        fill_remote_schema_keys(&mut average_resolver, MAX_REMOTE_TYPE_META_KEYS, 3);
827        *average_resolver
828            .remote_schema_versions_by_type
829            .get_mut("i0")
830            .unwrap() = 2;
831        average_resolver.total_accepted_schema_versions -= 1;
832        let average_config = Config {
833            max_schema_versions_per_type: 10,
834            max_average_schema_versions_per_type: 3,
835            ..Default::default()
836        };
837
838        let accepted = remote_struct_meta(0, "b");
839        read_type_def(&mut average_resolver, &average_config, accepted.get_bytes()).unwrap();
840        assert_eq!(average_resolver.total_accepted_schema_versions, 24_576);
841
842        let rejected = remote_struct_meta(0, "c");
843        let err = read_type_def(&mut average_resolver, &average_config, rejected.get_bytes())
844            .unwrap_err()
845            .to_string();
846        assert!(err.contains("max_average_schema_versions_per_type"));
847        assert_eq!(average_resolver.total_accepted_schema_versions, 24_576);
848    }
849
850    #[test]
851    fn schema_total_does_not_wrap() {
852        let config = Config {
853            max_schema_versions_per_type: u32::MAX,
854            max_average_schema_versions_per_type: u32::MAX,
855            ..Default::default()
856        };
857        let mut resolver = MetaReaderResolver::default();
858        fill_remote_schema_keys(&mut resolver, 1, 1);
859        resolver.total_accepted_schema_versions = u64::MAX;
860        let meta = remote_struct_meta(0, "b");
861
862        let err = read_type_def(&mut resolver, &config, meta.get_bytes())
863            .unwrap_err()
864            .to_string();
865
866        assert!(err.contains("remote schema version limit exceeded globally"));
867        assert_eq!(resolver.total_accepted_schema_versions, u64::MAX);
868        assert_eq!(resolver.remote_schema_versions_by_type.get("i0"), Some(&1));
869        assert!(resolver.parsed_type_infos.is_empty());
870        assert!(resolver.cached_type_info.is_none());
871        assert!(resolver.reading_type_infos.is_empty());
872    }
873
874    #[test]
875    fn checked_cache_bypasses_key_cap() {
876        let config = Config::default();
877        let mut resolver = MetaReaderResolver::default();
878        fill_remote_schema_keys(&mut resolver, MAX_REMOTE_TYPE_META_KEYS - 1, 1);
879        let meta = remote_struct_meta((MAX_REMOTE_TYPE_META_KEYS - 1) as u32, "a");
880        let first = read_type_def(&mut resolver, &config, meta.get_bytes()).unwrap();
881
882        resolver.reset();
883        resolver.cached_type_info = None;
884        let strict_config = Config {
885            max_schema_versions_per_type: 1,
886            max_average_schema_versions_per_type: 1,
887            ..Default::default()
888        };
889        let cached = read_type_def(&mut resolver, &strict_config, meta.get_bytes()).unwrap();
890
891        assert!(Rc::ptr_eq(&first, &cached));
892        assert_eq!(resolver.reading_type_infos.len(), 1);
893        assert_eq!(
894            resolver.remote_schema_versions_by_type.len(),
895            MAX_REMOTE_TYPE_META_KEYS
896        );
897        assert_eq!(
898            resolver.total_accepted_schema_versions,
899            MAX_REMOTE_TYPE_META_KEYS as u64
900        );
901    }
902
903    #[test]
904    fn checked_hash_ignores_frame_length() {
905        let mut resolver = MetaReaderResolver::default();
906        let meta = remote_struct_meta(9001, "a");
907        let first = read_type_def(&mut resolver, &Config::default(), meta.get_bytes()).unwrap();
908        let parsed_count = resolver.parsed_type_infos.len();
909        let accepted_versions = resolver.total_accepted_schema_versions;
910        let strict_config = Config {
911            max_type_fields: 0,
912            max_type_meta_bytes: 1,
913            max_schema_versions_per_type: 0,
914            max_average_schema_versions_per_type: 0,
915            ..Default::default()
916        };
917
918        resolver.reset();
919        resolver.cached_type_info = None;
920        let inline = type_def_frame(meta.get_hash(), 0b101 << 9, 17, 0xff);
921        let inline_header = i64::from_le_bytes(inline[..8].try_into().unwrap());
922        assert_eq!(TypeMeta::header_hash(inline_header), meta.get_hash());
923        let (inline_result, inline_cursor, inline_len) =
924            read_type_def_with_cursor(&mut resolver, &strict_config, &inline);
925        assert!(Rc::ptr_eq(&first, &inline_result.unwrap()));
926        assert_eq!(inline_cursor, inline_len);
927
928        resolver.reset();
929        let extended = type_def_frame(meta.get_hash(), 0b1 << 8, 257, 0xa5);
930        let extended_header = i64::from_le_bytes(extended[..8].try_into().unwrap());
931        assert_eq!(TypeMeta::header_hash(extended_header), meta.get_hash());
932        let (extended_result, extended_cursor, extended_len) =
933            read_type_def_with_cursor(&mut resolver, &strict_config, &extended);
934        assert!(Rc::ptr_eq(&first, &extended_result.unwrap()));
935        assert_eq!(extended_cursor, extended_len);
936        assert_eq!(resolver.parsed_type_infos.len(), parsed_count);
937        assert_eq!(resolver.total_accepted_schema_versions, accepted_versions);
938    }
939
940    #[test]
941    fn checked_hash_rejects_truncation() {
942        let mut resolver = MetaReaderResolver::default();
943        let meta = remote_struct_meta(9001, "a");
944        read_type_def(&mut resolver, &Config::default(), meta.get_bytes()).unwrap();
945        let parsed_count = resolver.parsed_type_infos.len();
946
947        resolver.reset();
948        resolver.cached_type_info = None;
949        let mut truncated = meta.get_bytes().to_vec();
950        truncated.pop();
951        let (result, _, _) =
952            read_type_def_with_cursor(&mut resolver, &Config::default(), &truncated);
953
954        assert!(result.is_err());
955        assert!(resolver.reading_type_infos.is_empty());
956        assert!(resolver.cached_type_info.is_none());
957        assert_eq!(resolver.parsed_type_infos.len(), parsed_count);
958    }
959
960    #[test]
961    fn exact_local_bypasses_key_cap() {
962        let mut type_resolver = TypeResolver::default();
963        type_resolver
964            .register_serializer_by_name::<LocalExt>("example.SharedExt")
965            .unwrap();
966        let type_resolver = type_resolver.build_final_type_resolver().unwrap();
967        let local_info = type_resolver
968            .get_type_info_by_name("example", "SharedExt")
969            .unwrap();
970        let exact = local_info.get_type_meta_ref().get_bytes().to_vec();
971
972        let mut resolver = MetaReaderResolver::default();
973        fill_remote_schema_keys(&mut resolver, MAX_REMOTE_TYPE_META_KEYS, 1);
974        let strict_config = Config {
975            max_schema_versions_per_type: 1,
976            max_average_schema_versions_per_type: 1,
977            ..Default::default()
978        };
979        let resolved =
980            read_type_def_with_type_resolver(&mut resolver, &strict_config, &type_resolver, &exact)
981                .unwrap();
982
983        assert!(Rc::ptr_eq(&local_info, &resolved));
984        assert_eq!(
985            resolver.remote_schema_versions_by_type.len(),
986            MAX_REMOTE_TYPE_META_KEYS
987        );
988        assert_eq!(
989            resolver.total_accepted_schema_versions,
990            MAX_REMOTE_TYPE_META_KEYS as u64
991        );
992    }
993
994    #[test]
995    fn local_hash_owns_validated_miss() {
996        let mut type_resolver = TypeResolver::default();
997        type_resolver
998            .register_by_name::<CollisionStruct>("e!.C!")
999            .unwrap();
1000        let type_resolver = type_resolver.build_final_type_resolver().unwrap();
1001        let local_info = type_resolver.get_type_info_by_name("e!", "C!").unwrap();
1002        let local_type_def = local_info.get_type_meta_ref().get_bytes();
1003        let namespace = NAMESPACE_ENCODER
1004            .encode_with_encodings("e!", NAMESPACE_ENCODINGS)
1005            .unwrap();
1006        let type_name = TYPE_NAME_ENCODER
1007            .encode_with_encodings("C!", TYPE_NAME_ENCODINGS)
1008            .unwrap();
1009        // These two field names form a valid top-52 TypeMeta hash collision while producing
1010        // different body lengths. The miss path must validate the remote frame, then select the
1011        // local owner by that protocol identity without falling back to full encoded bytes.
1012        let remote_meta = TypeMeta::new(
1013            TypeId::NAMED_STRUCT as u32,
1014            NO_USER_TYPE_ID,
1015            namespace,
1016            type_name,
1017            true,
1018            vec![FieldInfo::new(
1019                REMOTE_COLLISION_FIELD,
1020                FieldType::new(crate::type_id::INT32, false, vec![]),
1021            )],
1022        )
1023        .unwrap();
1024        let remote_type_def = remote_meta.get_bytes();
1025        let local_header = i64::from_le_bytes(local_type_def[..8].try_into().unwrap());
1026        let remote_header = i64::from_le_bytes(remote_type_def[..8].try_into().unwrap());
1027        assert_ne!(local_header & 0xfff, remote_header & 0xfff);
1028        assert_ne!(&local_type_def[8..], &remote_type_def[8..]);
1029        assert_eq!(
1030            TypeMeta::header_hash(remote_header),
1031            local_info.get_type_meta_ref().get_hash()
1032        );
1033
1034        let mut resolver = MetaReaderResolver::default();
1035        fill_remote_schema_keys(&mut resolver, MAX_REMOTE_TYPE_META_KEYS, 1);
1036        let cached_hash = resolver.cached_meta_hash;
1037        let strict_config = Config {
1038            max_schema_versions_per_type: 1,
1039            max_average_schema_versions_per_type: 1,
1040            ..Default::default()
1041        };
1042        let resolved = read_type_def_with_type_resolver(
1043            &mut resolver,
1044            &strict_config,
1045            &type_resolver,
1046            remote_type_def,
1047        )
1048        .unwrap();
1049
1050        assert!(Rc::ptr_eq(&local_info, &resolved));
1051        assert_eq!(resolver.reading_type_infos.len(), 1);
1052        assert!(resolver.parsed_type_infos.is_empty());
1053        assert_eq!(resolver.cached_meta_hash, cached_hash);
1054        assert!(resolver.cached_type_info.is_none());
1055        assert!(!resolver
1056            .remote_schema_versions_by_type
1057            .contains_key("ne!\0C!"));
1058        assert_eq!(
1059            resolver.total_accepted_schema_versions,
1060            MAX_REMOTE_TYPE_META_KEYS as u64
1061        );
1062
1063        // A checked remote owner under the same top-52 identity must not outrank a statically
1064        // expected local owner, nor may that local hit replace or increment remote cache state.
1065        let remote_header_hash = TypeMeta::header_hash(remote_header);
1066        let mut warmed = MetaReaderResolver::default();
1067        let remote_owner = read_type_def_with_type_resolver(
1068            &mut warmed,
1069            &Config::default(),
1070            &TypeResolver::default(),
1071            remote_type_def,
1072        )
1073        .unwrap();
1074        let versions = warmed.total_accepted_schema_versions;
1075        assert_eq!(warmed.parsed_type_infos.len(), 1);
1076        assert!(Rc::ptr_eq(
1077            warmed.parsed_type_infos.get(&remote_header_hash).unwrap(),
1078            &remote_owner,
1079        ));
1080
1081        let mut bytes = vec![];
1082        let mut writer = Writer::from_buffer(&mut bytes);
1083        writer.write_var_u32(0);
1084        writer.write_bytes(local_type_def);
1085        let mut reader = Reader::new(&bytes);
1086        let resolved = warmed
1087            .read_type_meta_for(&mut reader, &type_resolver, &Config::default(), &local_info)
1088            .unwrap();
1089
1090        assert!(Rc::ptr_eq(&local_info, &resolved));
1091        assert_eq!(reader.get_cursor(), bytes.len());
1092        assert_eq!(warmed.total_accepted_schema_versions, versions);
1093        assert_eq!(warmed.parsed_type_infos.len(), 1);
1094        assert!(Rc::ptr_eq(
1095            warmed.parsed_type_infos.get(&remote_header_hash).unwrap(),
1096            &remote_owner,
1097        ));
1098        assert!(Rc::ptr_eq(
1099            warmed.cached_type_info.as_ref().unwrap(),
1100            &remote_owner,
1101        ));
1102    }
1103
1104    #[test]
1105    fn expected_owner_rejects_foreign() {
1106        let mut type_resolver = TypeResolver::default();
1107        type_resolver
1108            .register_by_name::<CollisionStruct>("owner.StructA")
1109            .unwrap();
1110        type_resolver
1111            .register_by_name::<ForeignStruct>("owner.StructB")
1112            .unwrap();
1113        let type_resolver = type_resolver.build_final_type_resolver().unwrap();
1114        let expected = type_resolver
1115            .get_type_info_by_name("owner", "StructA")
1116            .unwrap();
1117        let foreign = type_resolver
1118            .get_type_info_by_name("owner", "StructB")
1119            .unwrap();
1120        let config = Config::default();
1121
1122        let unknown = named_struct_meta("owner", "Unknown", "remote");
1123        let mut unknown_resolver = MetaReaderResolver::default();
1124        let error = read_type_def_for(
1125            &mut unknown_resolver,
1126            &config,
1127            &type_resolver,
1128            &expected,
1129            unknown.get_bytes(),
1130        )
1131        .unwrap_err();
1132        assert!(error.to_string().contains("does not match declared target"));
1133        assert!(unknown_resolver.reading_type_infos.is_empty());
1134        assert!(unknown_resolver.parsed_type_infos.is_empty());
1135        assert!(unknown_resolver.cached_type_info.is_none());
1136        assert!(unknown_resolver.remote_schema_versions_by_type.is_empty());
1137        assert_eq!(unknown_resolver.total_accepted_schema_versions, 0);
1138
1139        let mut structural_stub_resolver = MetaReaderResolver::default();
1140        let structural_stub = read_struct_type_def_for(
1141            &mut structural_stub_resolver,
1142            &config,
1143            &type_resolver,
1144            &expected,
1145            unknown.get_bytes(),
1146        )
1147        .unwrap();
1148        assert!(structural_stub.get_harness().target_type_id().is_none());
1149        assert_eq!(structural_stub_resolver.reading_type_infos.len(), 1);
1150        assert_eq!(structural_stub_resolver.parsed_type_infos.len(), 1);
1151        assert_eq!(structural_stub_resolver.total_accepted_schema_versions, 1);
1152
1153        let non_struct_meta = [
1154            (TypeId::NAMED_EXT, "UnknownExt"),
1155            (TypeId::NAMED_ENUM, "UnknownEnum"),
1156            (TypeId::NAMED_UNION, "UnknownUnion"),
1157        ];
1158        for (type_id, type_name) in non_struct_meta {
1159            let remote = named_meta(type_id, "remote", type_name, vec![]);
1160            let mut resolver = MetaReaderResolver::default();
1161            let error = read_struct_type_def_for(
1162                &mut resolver,
1163                &config,
1164                &type_resolver,
1165                &expected,
1166                remote.get_bytes(),
1167            )
1168            .unwrap_err();
1169            assert!(error.to_string().contains("not structural metadata"));
1170            assert!(resolver.reading_type_infos.is_empty());
1171            assert!(resolver.parsed_type_infos.is_empty());
1172            assert!(resolver.cached_type_info.is_none());
1173            assert!(resolver.remote_schema_versions_by_type.is_empty());
1174            assert_eq!(resolver.total_accepted_schema_versions, 0);
1175        }
1176
1177        let same_name_ext = named_meta(TypeId::NAMED_EXT, "owner", "StructA", vec![]);
1178        for structural in [false, true] {
1179            let mut resolver = MetaReaderResolver::default();
1180            let result = if structural {
1181                read_struct_type_def_for(
1182                    &mut resolver,
1183                    &config,
1184                    &type_resolver,
1185                    &expected,
1186                    same_name_ext.get_bytes(),
1187                )
1188            } else {
1189                read_type_def_with_type_resolver(
1190                    &mut resolver,
1191                    &config,
1192                    &type_resolver,
1193                    same_name_ext.get_bytes(),
1194                )
1195            };
1196            let error = result.unwrap_err();
1197            assert!(
1198                error
1199                    .to_string()
1200                    .contains("kind does not match registered type metadata"),
1201                "{error}"
1202            );
1203            assert!(resolver.reading_type_infos.is_empty());
1204            assert!(resolver.parsed_type_infos.is_empty());
1205            assert!(resolver.cached_type_info.is_none());
1206            assert!(resolver.remote_schema_versions_by_type.is_empty());
1207            assert_eq!(resolver.total_accepted_schema_versions, 0);
1208        }
1209
1210        let unknown_ext = named_meta(TypeId::NAMED_EXT, "remote", "CachedExt", vec![]);
1211        let mut ext_resolver = MetaReaderResolver::default();
1212        let ext_info = read_type_def_with_type_resolver(
1213            &mut ext_resolver,
1214            &config,
1215            &type_resolver,
1216            unknown_ext.get_bytes(),
1217        )
1218        .unwrap();
1219        assert!(ext_info.get_harness().target_type_id().is_none());
1220        let mut ref_bytes = vec![];
1221        Writer::from_buffer(&mut ref_bytes).write_var_u32(1);
1222        let mut ref_reader = Reader::new(&ref_bytes);
1223        let error = ext_resolver
1224            .read_struct_type_meta_for(&mut ref_reader, &type_resolver, &config, &expected)
1225            .unwrap_err();
1226        assert!(error.to_string().contains("not structural metadata"));
1227        assert_eq!(ext_resolver.reading_type_infos.len(), 1);
1228        assert_eq!(ext_resolver.parsed_type_infos.len(), 1);
1229        assert_eq!(ext_resolver.total_accepted_schema_versions, 1);
1230
1231        ext_resolver.reset();
1232        let parsed_count = ext_resolver.parsed_type_infos.len();
1233        let accepted_versions = ext_resolver.total_accepted_schema_versions;
1234        let cached_hash = ext_resolver.cached_meta_hash;
1235        let cached_owner = ext_resolver.cached_type_info.as_ref().map(Rc::as_ptr);
1236        let error = read_struct_type_def_for(
1237            &mut ext_resolver,
1238            &config,
1239            &type_resolver,
1240            &expected,
1241            unknown_ext.get_bytes(),
1242        )
1243        .unwrap_err();
1244        assert!(error.to_string().contains("not structural metadata"));
1245        assert!(ext_resolver.reading_type_infos.is_empty());
1246        assert_eq!(ext_resolver.parsed_type_infos.len(), parsed_count);
1247        assert_eq!(
1248            ext_resolver.total_accepted_schema_versions,
1249            accepted_versions
1250        );
1251        assert_eq!(ext_resolver.cached_meta_hash, cached_hash);
1252        assert_eq!(
1253            ext_resolver.cached_type_info.as_ref().map(Rc::as_ptr),
1254            cached_owner
1255        );
1256
1257        let changed_foreign = named_struct_meta("owner", "StructB", "remote_b");
1258        assert_ne!(
1259            changed_foreign.get_hash(),
1260            foreign.get_type_meta_ref().get_hash()
1261        );
1262        let mut miss_resolver = MetaReaderResolver::default();
1263        let error = read_type_def_for(
1264            &mut miss_resolver,
1265            &config,
1266            &type_resolver,
1267            &expected,
1268            changed_foreign.get_bytes(),
1269        )
1270        .unwrap_err();
1271        assert!(error.to_string().contains("does not match declared target"));
1272        assert!(miss_resolver.reading_type_infos.is_empty());
1273        assert!(miss_resolver.parsed_type_infos.is_empty());
1274        assert!(miss_resolver.cached_type_info.is_none());
1275        assert!(miss_resolver.remote_schema_versions_by_type.is_empty());
1276        assert_eq!(miss_resolver.total_accepted_schema_versions, 0);
1277
1278        let mut structural_miss_resolver = MetaReaderResolver::default();
1279        let error = read_struct_type_def_for(
1280            &mut structural_miss_resolver,
1281            &config,
1282            &type_resolver,
1283            &expected,
1284            changed_foreign.get_bytes(),
1285        )
1286        .unwrap_err();
1287        assert!(error.to_string().contains("does not match declared target"));
1288        assert!(structural_miss_resolver.reading_type_infos.is_empty());
1289        assert!(structural_miss_resolver.parsed_type_infos.is_empty());
1290        assert!(structural_miss_resolver.cached_type_info.is_none());
1291        assert!(structural_miss_resolver
1292            .remote_schema_versions_by_type
1293            .is_empty());
1294        assert_eq!(structural_miss_resolver.total_accepted_schema_versions, 0);
1295
1296        let mut ref_resolver = MetaReaderResolver::default();
1297        let resolved = read_type_def_with_type_resolver(
1298            &mut ref_resolver,
1299            &config,
1300            &type_resolver,
1301            foreign.get_type_meta_ref().get_bytes(),
1302        )
1303        .unwrap();
1304        assert!(Rc::ptr_eq(&resolved, &foreign));
1305        let mut ref_bytes = vec![];
1306        Writer::from_buffer(&mut ref_bytes).write_var_u32(1);
1307        let mut ref_reader = Reader::new(&ref_bytes);
1308        let error = ref_resolver
1309            .read_type_meta_for(&mut ref_reader, &type_resolver, &config, &expected)
1310            .unwrap_err();
1311        assert!(error.to_string().contains("does not match declared target"));
1312        assert_eq!(ref_resolver.reading_type_infos.len(), 1);
1313        assert!(ref_resolver.parsed_type_infos.is_empty());
1314        assert!(ref_resolver.cached_type_info.is_none());
1315        assert_eq!(ref_resolver.total_accepted_schema_versions, 0);
1316
1317        let mut cache_resolver = MetaReaderResolver::default();
1318        let cached = read_type_def_with_type_resolver(
1319            &mut cache_resolver,
1320            &config,
1321            &type_resolver,
1322            changed_foreign.get_bytes(),
1323        )
1324        .unwrap();
1325        assert_eq!(
1326            cached.get_harness().target_type_id(),
1327            foreign.get_harness().target_type_id()
1328        );
1329        cache_resolver.reset();
1330        let parsed_count = cache_resolver.parsed_type_infos.len();
1331        let accepted_versions = cache_resolver.total_accepted_schema_versions;
1332        let cached_hash = cache_resolver.cached_meta_hash;
1333        let cached_owner = cache_resolver.cached_type_info.as_ref().map(Rc::as_ptr);
1334        let error = read_type_def_for(
1335            &mut cache_resolver,
1336            &config,
1337            &type_resolver,
1338            &expected,
1339            changed_foreign.get_bytes(),
1340        )
1341        .unwrap_err();
1342        assert!(error.to_string().contains("does not match declared target"));
1343        assert!(cache_resolver.reading_type_infos.is_empty());
1344        assert_eq!(cache_resolver.parsed_type_infos.len(), parsed_count);
1345        assert_eq!(
1346            cache_resolver.total_accepted_schema_versions,
1347            accepted_versions
1348        );
1349        assert_eq!(cache_resolver.cached_meta_hash, cached_hash);
1350        assert_eq!(
1351            cache_resolver.cached_type_info.as_ref().map(Rc::as_ptr),
1352            cached_owner
1353        );
1354    }
1355
1356    #[test]
1357    fn type_meta_field_limit_rejects_large_struct() {
1358        let meta = TypeMeta::new(
1359            TypeId::STRUCT as u32,
1360            9001,
1361            MetaString::get_empty().clone(),
1362            MetaString::get_empty().clone(),
1363            false,
1364            vec![
1365                FieldInfo::new("a", FieldType::new(crate::type_id::INT32, false, vec![])),
1366                FieldInfo::new("b", FieldType::new(crate::type_id::INT32, false, vec![])),
1367            ],
1368        )
1369        .unwrap();
1370        let config = Config {
1371            max_type_fields: 1,
1372            ..Default::default()
1373        };
1374        let err = read_type_def(
1375            &mut MetaReaderResolver::default(),
1376            &config,
1377            meta.get_bytes(),
1378        )
1379        .unwrap_err()
1380        .to_string();
1381        assert!(err.contains("max_type_fields"));
1382    }
1383
1384    #[test]
1385    fn type_meta_body_limit_rejects_large_metadata() {
1386        let meta = TypeMeta::new(
1387            TypeId::STRUCT as u32,
1388            9001,
1389            MetaString::get_empty().clone(),
1390            MetaString::get_empty().clone(),
1391            false,
1392            vec![FieldInfo::new(
1393                "a",
1394                FieldType::new(crate::type_id::INT32, false, vec![]),
1395            )],
1396        )
1397        .unwrap();
1398        let config = Config {
1399            max_type_meta_bytes: 1,
1400            ..Default::default()
1401        };
1402        let err = read_type_def(
1403            &mut MetaReaderResolver::default(),
1404            &config,
1405            meta.get_bytes(),
1406        )
1407        .unwrap_err()
1408        .to_string();
1409        assert!(err.contains("max_type_meta_bytes"));
1410    }
1411
1412    #[test]
1413    fn schema_limit_tracks_unknown_struct_types_separately() {
1414        fn type_def(user_type_id: u32, field_name: &str) -> Vec<u8> {
1415            TypeMeta::new(
1416                TypeId::STRUCT as u32,
1417                user_type_id,
1418                MetaString::get_empty().clone(),
1419                MetaString::get_empty().clone(),
1420                false,
1421                vec![FieldInfo::new(
1422                    field_name,
1423                    FieldType::new(crate::type_id::INT32, false, vec![]),
1424                )],
1425            )
1426            .unwrap()
1427            .get_bytes()
1428            .to_vec()
1429        }
1430
1431        let config = Config {
1432            max_schema_versions_per_type: 1,
1433            ..Default::default()
1434        };
1435
1436        let mut resolver = MetaReaderResolver::default();
1437        read_type_def(&mut resolver, &config, &type_def(9001, "a")).unwrap();
1438        read_type_def(&mut resolver, &config, &type_def(9002, "a")).unwrap();
1439
1440        let err = read_type_def(&mut resolver, &config, &type_def(9001, "b"))
1441            .unwrap_err()
1442            .to_string();
1443        assert!(err.contains("max_schema_versions_per_type"));
1444    }
1445
1446    #[test]
1447    fn schema_limit_rejects_extra_versions_for_type() {
1448        let meta = TypeMeta::new(
1449            TypeId::STRUCT as u32,
1450            9001,
1451            MetaString::get_empty().clone(),
1452            MetaString::get_empty().clone(),
1453            false,
1454            vec![FieldInfo::new(
1455                "a",
1456                FieldType::new(crate::type_id::INT32, false, vec![]),
1457            )],
1458        )
1459        .unwrap();
1460        let type_def = meta.get_bytes().to_vec();
1461
1462        let config = Config {
1463            max_schema_versions_per_type: 1,
1464            ..Default::default()
1465        };
1466        let mut resolver = MetaReaderResolver::default();
1467        let mut bytes = vec![];
1468        let mut writer = Writer::from_buffer(&mut bytes);
1469        writer.write_var_u32(0);
1470        writer.write_bytes(&type_def);
1471        let mut reader = Reader::new(&bytes);
1472        resolver
1473            .read_type_meta(&mut reader, &TypeResolver::default(), &config)
1474            .unwrap();
1475
1476        let changed = TypeMeta::new(
1477            TypeId::STRUCT as u32,
1478            9001,
1479            MetaString::get_empty().clone(),
1480            MetaString::get_empty().clone(),
1481            false,
1482            vec![FieldInfo::new(
1483                "b",
1484                FieldType::new(crate::type_id::INT32, false, vec![]),
1485            )],
1486        )
1487        .unwrap();
1488        let mut bytes = vec![];
1489        let mut writer = Writer::from_buffer(&mut bytes);
1490        writer.write_var_u32(0);
1491        writer.write_bytes(changed.get_bytes());
1492        let mut reader = Reader::new(&bytes);
1493        let err = resolver
1494            .read_type_meta(&mut reader, &TypeResolver::default(), &config)
1495            .unwrap_err()
1496            .to_string();
1497        assert!(err.contains("max_schema_versions_per_type"));
1498    }
1499
1500    #[test]
1501    fn schema_limit_check_is_not_recorded() {
1502        let config = Config {
1503            max_schema_versions_per_type: 1,
1504            ..Default::default()
1505        };
1506        let mut resolver = MetaReaderResolver::default();
1507        let checked = TypeMeta::new(
1508            TypeId::STRUCT as u32,
1509            9001,
1510            MetaString::get_empty().clone(),
1511            MetaString::get_empty().clone(),
1512            false,
1513            vec![FieldInfo::new(
1514                "a",
1515                FieldType::new(crate::type_id::INT32, false, vec![]),
1516            )],
1517        )
1518        .unwrap();
1519        let accepted = TypeMeta::new(
1520            TypeId::STRUCT as u32,
1521            9001,
1522            MetaString::get_empty().clone(),
1523            MetaString::get_empty().clone(),
1524            false,
1525            vec![FieldInfo::new(
1526                "b",
1527                FieldType::new(crate::type_id::INT32, false, vec![]),
1528            )],
1529        )
1530        .unwrap();
1531
1532        resolver
1533            .check_remote_type_meta_limit(&checked, &config)
1534            .unwrap();
1535
1536        let mut bytes = vec![];
1537        let mut writer = Writer::from_buffer(&mut bytes);
1538        writer.write_var_u32(0);
1539        writer.write_bytes(accepted.get_bytes());
1540        let mut reader = Reader::new(&bytes);
1541        resolver
1542            .read_type_meta(&mut reader, &TypeResolver::default(), &config)
1543            .unwrap();
1544    }
1545
1546    #[test]
1547    fn non_struct_type_meta_uses_limit() {
1548        let config = Config {
1549            max_schema_versions_per_type: 1,
1550            ..Default::default()
1551        };
1552        let mut resolver = MetaReaderResolver::default();
1553        let namespace = NAMESPACE_ENCODER
1554            .encode_with_encodings("example", NAMESPACE_ENCODINGS)
1555            .unwrap();
1556        let type_name = TYPE_NAME_ENCODER
1557            .encode_with_encodings("RemoteEnum", TYPE_NAME_ENCODINGS)
1558            .unwrap();
1559        let first = TypeMeta::new(
1560            TypeId::NAMED_ENUM as u32,
1561            NO_USER_TYPE_ID,
1562            namespace.clone(),
1563            type_name.clone(),
1564            true,
1565            vec![],
1566        )
1567        .unwrap();
1568        let second = TypeMeta::new(
1569            TypeId::NAMED_EXT as u32,
1570            NO_USER_TYPE_ID,
1571            namespace,
1572            type_name,
1573            true,
1574            vec![],
1575        )
1576        .unwrap();
1577
1578        let key = resolver
1579            .check_remote_type_meta_limit(&first, &config)
1580            .unwrap();
1581        resolver.record_remote_type_meta(key);
1582
1583        let err = resolver
1584            .check_remote_type_meta_limit(&second, &config)
1585            .unwrap_err()
1586            .to_string();
1587        assert!(err.contains("max_schema_versions_per_type"));
1588    }
1589
1590    #[test]
1591    fn exact_local_non_struct_type_meta_bypasses_limit() {
1592        let config = Config {
1593            max_schema_versions_per_type: 1,
1594            ..Default::default()
1595        };
1596        let mut type_resolver = TypeResolver::default();
1597        type_resolver
1598            .register_serializer_by_name::<LocalExt>("example.SharedExt")
1599            .unwrap();
1600        let type_resolver = type_resolver.build_final_type_resolver().unwrap();
1601        let local_info = type_resolver
1602            .get_type_info_by_name("example", "SharedExt")
1603            .unwrap();
1604        let exact = local_info.get_type_meta_ref().get_bytes().to_vec();
1605
1606        let mut resolver = MetaReaderResolver::default();
1607        read_type_def_with_type_resolver(&mut resolver, &config, &type_resolver, &exact).unwrap();
1608
1609        let namespace = NAMESPACE_ENCODER
1610            .encode_with_encodings("example", NAMESPACE_ENCODINGS)
1611            .unwrap();
1612        let type_name = TYPE_NAME_ENCODER
1613            .encode_with_encodings("SharedExt", TYPE_NAME_ENCODINGS)
1614            .unwrap();
1615        let second = TypeMeta::new(
1616            TypeId::NAMED_ENUM as u32,
1617            NO_USER_TYPE_ID,
1618            namespace,
1619            type_name,
1620            true,
1621            vec![],
1622        )
1623        .unwrap();
1624        resolver
1625            .check_remote_type_meta_limit(&second, &config)
1626            .unwrap();
1627    }
1628}
fory_core/resolver/meta_resolver.rs

fory_core/resolver/
meta_resolver.rs
