detcore/preemptions/
record_serde.rs

1/*
2 * Copyright (c) Meta Platforms, Inc. and affiliates.
3 * All rights reserved.
4 *
5 * This source code is licensed under the BSD-style license found in the
6 * LICENSE file in the root directory of this source tree.
7 */
8
9//! Preserve legacy encoding, but never let an old reader ignore known omissions.
10
11use std::collections::BTreeMap;
12use std::fmt;
13
14use chrono::DateTime;
15use chrono::Utc;
16use detcore_model::host_input::HostInputSet;
17use serde::Deserialize;
18use serde::Serialize;
19use serde::de::DeserializeSeed;
20use serde::de::MapAccess;
21use serde::de::SeqAccess;
22use serde::de::Visitor;
23use serde::de::value::MapAccessDeserializer;
24use serde::de::value::SeqAccessDeserializer;
25use serde::de::value::StringDeserializer;
26use serde::ser::SerializeMap;
27
28use super::HostObservedCount;
29use super::PreemptionRecord;
30use super::ReservedInputMetadata;
31use super::ThreadHistory;
32use super::TimeSkip;
33use super::callback_history;
34use crate::types::DetTid;
35use crate::types::SchedEvent;
36
37const ENVELOPE: &str = "incomplete_preemption_record";
38const DEPARTURE: &str = "replay_departed_at";
39const VERSION: u32 = 1;
40
41// A private remote definition keeps the old field order/omissions without
42// exposing a public raw serializer that could bypass the marked envelope.
43#[derive(Serialize, Deserialize)]
44#[serde(remote = "PreemptionRecord")]
45struct RawRecord {
46    per_thread: BTreeMap<DetTid, ThreadHistory>,
47    global: Vec<SchedEvent>,
48    #[serde(default, skip_serializing_if = "Option::is_none")]
49    epoch: Option<DateTime<Utc>>,
50    #[serde(default, skip_serializing_if = "Option::is_none")]
51    pipe_wake_latency: Option<u64>,
52    #[serde(default, skip_serializing_if = "Option::is_none")]
53    signal_state_copy: Option<bool>,
54    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
55    cap_preemption_charge: bool,
56    #[serde(default, skip_serializing_if = "Option::is_none")]
57    chaos_timeslice: Option<u64>,
58    #[serde(default, skip_serializing_if = "Option::is_none")]
59    guest_ids: Option<detcore_model::pid::GuestIds>,
60    #[serde(default, skip_serializing_if = "ReservedInputMetadata::is_missing")]
61    reserved_input: ReservedInputMetadata,
62    #[serde(skip)]
63    replay_departed_at: Option<u64>,
64    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
65    callback_checkpoints: BTreeMap<DetTid, Vec<callback_history::Checkpoint>>,
66    #[serde(default, skip_serializing_if = "Option::is_none")]
67    rng_seed: Option<u64>,
68    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
69    expired_wait_runs_charged_once: bool,
70    #[serde(default, skip_serializing_if = "Option::is_none")]
71    host_inputs: Option<HostInputSet>,
72    #[serde(default, skip_serializing_if = "HostObservedCount::is_zero")]
73    store_exclusive_failures: HostObservedCount,
74    #[serde(default, skip_serializing_if = "Option::is_none")]
75    format_version: Option<u32>,
76    #[serde(default, skip_serializing_if = "super::lists_no_time_skips")]
77    time_skips: Option<Vec<TimeSkip>>,
78}
79
80struct LegacyFields<'a>(&'a PreemptionRecord);
81
82impl Serialize for LegacyFields<'_> {
83    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
84        RawRecord::serialize(self.0, serializer)
85    }
86}
87
88#[derive(Serialize)]
89struct MarkedRef<'a> {
90    version: u32,
91    replay_departed_at: u64,
92    record: LegacyFields<'a>,
93}
94
95#[derive(Deserialize)]
96#[serde(deny_unknown_fields)]
97struct Marked {
98    version: u32,
99    replay_departed_at: u64,
100    #[serde(deserialize_with = "deserialize_payload")]
101    record: PreemptionRecord,
102}
103
104impl Serialize for PreemptionRecord {
105    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
106        let Some(replay_departed_at) = self.replay_departed_at else {
107            return RawRecord::serialize(self, serializer);
108        };
109        // Fail before invoking any output-producing serializer operation.
110        // There is no versioned binary envelope for this standalone record.
111        if !serializer.is_human_readable() {
112            return Err(serde::ser::Error::custom(
113                "marked preemption records require the incomplete_preemption_record JSON envelope",
114            ));
115        }
116        let mut map = serializer.serialize_map(Some(1))?;
117        map.serialize_entry(
118            ENVELOPE,
119            &MarkedRef {
120                version: VERSION,
121                replay_departed_at,
122                record: LegacyFields(self),
123            },
124        )?;
125        map.end()
126    }
127}
128
129impl<'de> Deserialize<'de> for PreemptionRecord {
130    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
131        if !deserializer.is_human_readable() {
132            // Keep the old positional decoder, including its pre-existing
133            // limitations for fields omitted by conditional serialization.
134            return RawRecord::deserialize(deserializer);
135        }
136        deserializer.deserialize_any(RecordVisitor)
137    }
138}
139
140struct RecordVisitor;
141
142impl<'de> Visitor<'de> for RecordVisitor {
143    type Value = PreemptionRecord;
144
145    fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
146        formatter.write_str("a legacy preemption record or an incomplete-preemption envelope")
147    }
148
149    fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<Self::Value, A::Error> {
150        let first = map.next_key::<String>()?;
151        if first.as_deref() == Some(ENVELOPE) {
152            let marked: Marked = map.next_value()?;
153            if map.next_key::<serde::de::IgnoredAny>()?.is_some() {
154                return Err(serde::de::Error::custom(
155                    "an incomplete-preemption envelope cannot contain other top-level fields",
156                ));
157            }
158            if marked.version != VERSION {
159                return Err(serde::de::Error::custom(format!(
160                    "unsupported incomplete_preemption_record version {}",
161                    marked.version
162                )));
163            }
164            let mut record = marked.record;
165            record.replay_departed_at = Some(marked.replay_departed_at);
166            Ok(record)
167        } else {
168            legacy_map(map, first, true)
169        }
170    }
171
172    fn visit_seq<A: SeqAccess<'de>>(self, sequence: A) -> Result<Self::Value, A::Error> {
173        // Preserve the old unmarked positional human-readable representation.
174        RawRecord::deserialize(SeqAccessDeserializer::new(sequence))
175    }
176}
177
178// An envelope's payload must be genuinely raw: neither a nested envelope nor
179// a flat marker (even null) may override or obscure its authoritative marker.
180fn deserialize_payload<'de, D: serde::Deserializer<'de>>(
181    deserializer: D,
182) -> Result<PreemptionRecord, D::Error> {
183    struct Payload;
184    impl<'de> Visitor<'de> for Payload {
185        type Value = PreemptionRecord;
186        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
187            formatter.write_str("unmarked legacy fields inside the envelope")
188        }
189        fn visit_map<A: MapAccess<'de>>(self, map: A) -> Result<Self::Value, A::Error> {
190            legacy_map(map, None, false)
191        }
192    }
193    deserializer.deserialize_map(Payload)
194}
195
196fn legacy_map<'de, A: MapAccess<'de>>(
197    map: A,
198    first: Option<String>,
199    allow_flat_marker: bool,
200) -> Result<PreemptionRecord, A::Error> {
201    let mut departure = None;
202    let mut seen = false;
203    let mut record = RawRecord::deserialize(MapAccessDeserializer::new(LegacyMap {
204        map,
205        first,
206        allow_flat_marker,
207        departure: &mut departure,
208        seen: &mut seen,
209    }))?;
210    record.replay_departed_at = departure;
211    Ok(record)
212}
213
214struct LegacyMap<'a, A> {
215    map: A,
216    first: Option<String>,
217    allow_flat_marker: bool,
218    departure: &'a mut Option<u64>,
219    seen: &'a mut bool,
220}
221
222impl<'de, A: MapAccess<'de>> MapAccess<'de> for LegacyMap<'_, A> {
223    type Error = A::Error;
224
225    fn next_key_seed<K: DeserializeSeed<'de>>(
226        &mut self,
227        seed: K,
228    ) -> Result<Option<K::Value>, Self::Error> {
229        loop {
230            let key = match self.first.take() {
231                Some(key) => Some(key),
232                None => self.map.next_key::<String>()?,
233            };
234            let Some(key) = key else { return Ok(None) };
235            if key == ENVELOPE {
236                return Err(serde::de::Error::custom(
237                    "an incomplete-preemption envelope cannot be mixed with legacy fields",
238                ));
239            }
240            if key != DEPARTURE {
241                return seed.deserialize(StringDeserializer::new(key)).map(Some);
242            }
243            if !self.allow_flat_marker {
244                return Err(serde::de::Error::custom(
245                    "an envelope payload cannot contain replay_departed_at",
246                ));
247            }
248            if *self.seen {
249                return Err(serde::de::Error::duplicate_field("replay_departed_at"));
250            }
251            *self.seen = true;
252            // Historical flat marked JSON is read, but its omission fact is
253            // retained and faithful admission rejects it. New writes never
254            // emit that old-reader-unsafe representation.
255            *self.departure = self.map.next_value()?;
256        }
257    }
258
259    fn next_value_seed<V: DeserializeSeed<'de>>(
260        &mut self,
261        seed: V,
262    ) -> Result<V::Value, Self::Error> {
263        self.map.next_value_seed(seed)
264    }
265}
266
267/// Internal metadata readers need both the legacy settings and the envelope's
268/// marker. This value never becomes a stored record or a public raw serializer.
269pub(super) fn settings_value(record: &PreemptionRecord) -> serde_json::Result<serde_json::Value> {
270    let mut value = serde_json::to_value(LegacyFields(record))?;
271    if let Some(event) = record.replay_departed_at {
272        value[DEPARTURE] = event.into();
273    }
274    Ok(value)
275}
276
277#[cfg(test)]
278mod tests {
279    use super::*;
280    use crate::preemptions::PreemptionWriter;
281    use crate::preemptions::boundary_recording::BoundaryRecording;
282    use crate::preemptions::boundary_recording::Recording;
283    use crate::preemptions::read_recorded_replay_departure;
284
285    #[test]
286    fn marked_envelopes_never_fall_back_to_legacy_fields() {
287        let raw = r#"{"per_thread":{},"global":[]}"#;
288        let valid = format!(r#"{{"version":1,"replay_departed_at":7,"record":{raw}}}"#);
289        let envelope = |body: &str| format!(r#"{{"incomplete_preemption_record":{body}}}"#);
290        let marked = envelope(&valid);
291        assert_eq!(
292            serde_json::from_str::<PreemptionRecord>(&marked)
293                .unwrap()
294                .replay_departed_at(),
295            Some(7)
296        );
297        let bad = [
298            envelope("null"),
299            envelope(&valid.replace("\"version\":1", "\"version\":2")),
300            envelope(&valid.replace("\"version\":1,", "")),
301            envelope(&valid.replace("\"replay_departed_at\":7,", "")),
302            envelope(&valid.replace("\"replay_departed_at\":7", "\"replay_departed_at\":null")),
303            envelope(&valid.replace("\"version\":1", "\"version\":1,\"version\":1")),
304            envelope(&valid.replace(
305                "\"replay_departed_at\":7",
306                "\"replay_departed_at\":7,\"replay_departed_at\":9",
307            )),
308            format!(
309                r#"{{"incomplete_preemption_record":{valid},"incomplete_preemption_record":{valid}}}"#
310            ),
311            format!(r#"{{"incomplete_preemption_record":{valid},"per_thread":{{}},"global":[]}}"#),
312            format!(r#"{{"per_thread":{{}},"global":[],"incomplete_preemption_record":{valid}}}"#),
313            r#"{"per_thread":{},"global":[],"incomplete_preemption_record":null}"#.to_owned(),
314            envelope(&valid.replace(raw, &marked)),
315            envelope(&valid.replace(
316                raw,
317                r#"{"per_thread":{},"global":[],"replay_departed_at":7}"#,
318            )),
319            envelope(&valid.replace(
320                raw,
321                r#"{"per_thread":{},"global":[],"replay_departed_at":null}"#,
322            )),
323            r#"{"per_thread":{},"global":[],"replay_departed_at":7,"replay_departed_at":9}"#
324                .to_owned(),
325        ];
326        let dir = tempfile::tempdir().unwrap();
327        let path = dir.path().join("bad.json");
328        for input in bad {
329            assert!(
330                serde_json::from_str::<PreemptionRecord>(&input).is_err(),
331                "{input}"
332            );
333            assert!(Recording::decode(input.as_bytes()).is_err(), "{input}");
334            // Every malformed wrapper also fails metadata unwrapping. Flat
335            // legacy metadata retains its historical individual-field contract.
336            if input.contains(ENVELOPE) {
337                std::fs::write(&path, &input).unwrap();
338                assert!(
339                    crate::preemptions::read_recorded_epoch(&path).is_err(),
340                    "{input}"
341                );
342                assert!(read_recorded_replay_departure(&path).is_err(), "{input}");
343            }
344        }
345    }
346
347    #[test]
348    fn historical_flat_departures_are_retained_without_restricting_ordinary_unknown_fields() {
349        let ordinary = r#"{"per_thread":{},"global":[],"future_setting":{"x":1}}"#;
350        let ordinary: PreemptionRecord = serde_json::from_str(ordinary).unwrap();
351        ordinary.require_faithful_replay().unwrap();
352        assert_eq!(
353            serde_json::to_string(&ordinary).unwrap(),
354            r#"{"per_thread":{},"global":[]}"#
355        );
356        for input in [
357            r#"{"replay_departed_at":0,"per_thread":{},"global":[]}"#,
358            r#"{"per_thread":{},"global":[],"replay_departed_at":0}"#,
359        ] {
360            let record: PreemptionRecord = serde_json::from_str(input).unwrap();
361            assert_eq!(record.replay_departed_at(), Some(0));
362            assert!(record.require_faithful_replay().is_err());
363            let rewritten = serde_json::to_string(&record).unwrap();
364            assert_eq!(
365                rewritten,
366                r#"{"incomplete_preemption_record":{"version":1,"replay_departed_at":0,"record":{"per_thread":{},"global":[]}}}"#
367            );
368            assert_eq!(
369                serde_json::from_str::<PreemptionRecord>(&rewritten).unwrap(),
370                record
371            );
372        }
373    }
374
375    #[test]
376    fn wrapped_settings_and_file_transforms_retain_the_authoritative_departure() {
377        let directory = tempfile::tempdir().unwrap();
378        for marker in [None, Some(7)] {
379            let mut writer = PreemptionWriter::new(None)
380                .with_epoch("2001-02-03T04:05:06Z".parse().unwrap())
381                .with_pipe_wake_latency(Some(0))
382                .with_signal_state_copy(false)
383                .with_rng_seed(31)
384                .with_expired_wait_runs_charged_once(true)
385                .with_cap_preemption_charge(true)
386                .with_chaos_timeslice(std::num::NonZeroU64::new(1000));
387            if let Some(event) = marker {
388                writer.mark_replay_departed(event);
389            }
390            let record: PreemptionRecord = serde_json::from_str(&writer.into_string()).unwrap();
391            for boundaries in [false, true] {
392                let recording = if boundaries {
393                    Recording::Boundaries(
394                        BoundaryRecording::new(
395                            record.clone(),
396                            detcore_model::schedule::boundary::Schedule::new(vec![]).unwrap(),
397                        )
398                        .unwrap(),
399                    )
400                } else {
401                    Recording::Legacy(record.clone())
402                };
403                let path = directory
404                    .path()
405                    .join(format!("{boundaries}-{marker:?}.json"));
406                recording.write(&path).unwrap();
407                assert_eq!(read_recorded_replay_departure(&path).unwrap(), marker);
408                assert_eq!(
409                    crate::preemptions::read_recorded_epoch(&path).unwrap(),
410                    record.epoch()
411                );
412                assert_eq!(
413                    crate::preemptions::read_recorded_pipe_wake_latency(&path).unwrap(),
414                    Some(0)
415                );
416                assert_eq!(
417                    crate::preemptions::read_recorded_signal_state_copy(&path).unwrap(),
418                    Some(false)
419                );
420                assert!(
421                    crate::preemptions::read_recorded_expired_wait_runs_charged_once(&path)
422                        .unwrap()
423                );
424                assert!(crate::preemptions::read_recorded_cap_preemption_charge(&path).unwrap());
425                assert_eq!(
426                    crate::preemptions::read_recorded_chaos_timeslice(&path).unwrap(),
427                    Some(1000)
428                );
429                let stripped =
430                    crate::preemptions::strip_times_from_events_file(&path, None).unwrap();
431                assert_eq!(Recording::read(&stripped).unwrap(), recording);
432                assert_eq!(read_recorded_replay_departure(&stripped).unwrap(), marker);
433                assert_eq!(
434                    Recording::read(&path)
435                        .unwrap()
436                        .require_faithful_replay()
437                        .is_err(),
438                    marker.is_some()
439                );
440            }
441            for copy in [
442                record.clone(),
443                record.clone_preemptions_only(),
444                record.schedule_prefix(0),
445                record.normalize(),
446                record.with_schedule(vec![]).unwrap(),
447            ] {
448                assert_eq!(copy.replay_departed_at(), marker);
449                assert_eq!(
450                    copy.retain_replay_departure(None).replay_departed_at(),
451                    marker
452                );
453            }
454        }
455    }
456}
detcore/preemptions/record_serde.rs

detcore/preemptions/
record_serde.rs
