1use std::{
10 collections::{BTreeMap, BTreeSet},
11 fs::{self, File, OpenOptions},
12 path::Path,
13 sync::atomic::{AtomicU8, AtomicU64, Ordering},
14};
15
16use memmap2::{Mmap, MmapMut, MmapOptions};
17use serde::{Deserialize, Serialize};
18
19use crate::rust_runtime::{RustProbeObservation, valid_probe_id};
20
21pub const RUST_TRANSPORT_ENV: &str = "SUPERCOV_RUST_TRANSPORT_FILE";
22pub const RUST_TRANSPORT_TOKEN_ENV: &str = "SUPERCOV_RUST_TRANSPORT_TOKEN";
23pub const RUST_CONTEXT_ENV: &str = "SUPERCOV_RUST_CONTEXT_ID";
24pub const DEFAULT_DESCRIPTOR_CAPACITY: u32 = 32_768;
25pub const DEFAULT_PAYLOAD_CAPACITY: u32 = 4 * 1024 * 1024;
26
27const MAGIC: &[u8; 8] = b"SCVRUST3";
28const VERSION: u32 = 3;
29const HEADER_SIZE: usize = 128;
30const DESCRIPTOR_SIZE: usize = 40;
31const ENDIAN_MARKER: u32 = 0x0102_0304;
32const NEXT_DESCRIPTOR_OFFSET: usize = 32;
33const NEXT_PAYLOAD_OFFSET: usize = 40;
34const DROPPED_OFFSET: usize = 48;
35const TOKEN_OFFSET: usize = 56;
36const TOKEN_SIZE: usize = 16;
37const ATTACHMENTS_OFFSET: usize = 72;
38const NEXT_PHASE_OFFSET: usize = 80;
39
40const COMMIT_OFFSET: usize = 0;
41const KIND_OFFSET: usize = 1;
42const OUTCOME_OFFSET: usize = 2;
43const PID_OFFSET: usize = 4;
44const CONTEXT_OFFSET: usize = 8;
45const PAYLOAD_OFFSET_OFFSET: usize = 16;
46const PAYLOAD_LENGTH_OFFSET: usize = 20;
47const ID_LENGTH_OFFSET: usize = 24;
48const VALUE_LENGTH_OFFSET: usize = 28;
49const CHECKSUM_OFFSET: usize = 32;
50
51const KIND_HIT: u8 = 1;
52const KIND_DECISION: u8 = 2;
53const KIND_ORDINAL_HIT: u8 = 3;
54const KIND_PHASE: u8 = 4;
55const KIND_THREAD_PHASE: u8 = 5;
56const KIND_THREAD_END: u8 = 6;
57const KIND_TEST_BOUNDARY: u8 = 7;
58
59const RUNTIME_TEMPLATE: &str = include_str!("../runtime-assets/rust-mmap-runtime.rs");
60
61#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
62#[serde(rename_all = "camelCase", deny_unknown_fields)]
63pub struct RustTransportRead {
64 pub observations: Vec<RustTransportObservation>,
65 pub ordinal_hits: Vec<RustOrdinalHit>,
66 pub phases: Vec<RustPhaseContext>,
67 pub thread_phases: Vec<RustThreadPhase>,
68 pub thread_ends: Vec<RustThreadEnd>,
69 pub test_boundaries: Vec<RustTestBoundary>,
70 pub committed: u64,
71 pub incomplete: u64,
72 pub dropped: u64,
73 pub attachments: u64,
74}
75
76impl RustTransportRead {
77 pub fn empty() -> Self {
78 Self {
79 observations: Vec::new(),
80 ordinal_hits: Vec::new(),
81 phases: Vec::new(),
82 thread_phases: Vec::new(),
83 thread_ends: Vec::new(),
84 test_boundaries: Vec::new(),
85 committed: 0,
86 incomplete: 0,
87 dropped: 0,
88 attachments: 0,
89 }
90 }
91}
92
93#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
94#[serde(rename_all = "camelCase", deny_unknown_fields)]
95pub struct RustTransportObservation {
96 pub process_id: u32,
97 pub context_id: u64,
98 pub observation: RustProbeObservation,
99}
100
101#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
102#[serde(rename_all = "camelCase", deny_unknown_fields)]
103pub struct RustOrdinalHit {
104 pub process_id: u32,
105 pub context_id: u64,
106 pub ordinal: u64,
107}
108
109#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
110#[serde(rename_all = "camelCase", deny_unknown_fields)]
111pub struct RustPhaseContext {
112 pub process_id: u32,
113 pub child_context_id: u64,
114 pub parent_context_id: u64,
115 pub invocation_nonce: u64,
116 pub decision_id: String,
117}
118
119#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
123#[serde(rename_all = "camelCase", deny_unknown_fields)]
124pub struct RustThreadPhase {
125 pub process_id: u32,
126 pub child_context_id: u64,
127 pub parent_context_id: u64,
128 pub invocation_nonce: u64,
129 pub commit_index: u64,
130}
131
132#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
134#[serde(rename_all = "camelCase", deny_unknown_fields)]
135pub struct RustThreadEnd {
136 pub process_id: u32,
137 pub context_id: u64,
138 pub commit_index: u64,
139}
140
141#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
143#[serde(rename_all = "camelCase", deny_unknown_fields)]
144pub struct RustTestBoundary {
145 pub process_id: u32,
146 pub context_id: u64,
147 pub commit_index: u64,
148}
149
150#[derive(Debug, Clone, PartialEq, Eq)]
151pub struct RustTransportPartition {
152 pub attributed: BTreeMap<u64, RustTransportRead>,
153 pub background: RustTransportRead,
154 pub thread_scope_limitations: BTreeSet<String>,
157}
158
159#[derive(Debug, Clone, PartialEq, Eq)]
160pub enum RustTransportError {
161 Io(String),
162 UnsafeFile(String),
163 InvalidHeader,
164 InvalidLength,
165 InvalidDescriptor(u64),
166 InvalidRecord(u64),
167 InvalidAssertionContext(String),
168 InvalidAttribution(String),
169}
170
171impl std::fmt::Display for RustTransportError {
172 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
173 match self {
174 Self::Io(error) => write!(formatter, "Rust transport I/O failed: {error}"),
175 Self::UnsafeFile(path) => write!(formatter, "unsafe Rust transport file: {path}"),
176 Self::InvalidHeader => write!(formatter, "invalid Rust transport header"),
177 Self::InvalidLength => write!(formatter, "invalid Rust transport length"),
178 Self::InvalidDescriptor(index) => {
179 write!(formatter, "invalid Rust transport descriptor {index}")
180 }
181 Self::InvalidRecord(index) => {
182 write!(formatter, "invalid Rust transport record {index}")
183 }
184 Self::InvalidAssertionContext(reason) => {
185 write!(formatter, "invalid Rust assertion context: {reason}")
186 }
187 Self::InvalidAttribution(reason) => {
188 write!(formatter, "invalid Rust transport attribution: {reason}")
189 }
190 }
191 }
192}
193
194impl std::error::Error for RustTransportError {}
195
196pub fn rust_assertion_context_id(
197 parent: u64,
198 decision_id: &str,
199 invocation_nonce: u64,
200) -> Result<u64, RustTransportError> {
201 if parent == 0 {
202 return Ok(0);
203 }
204 if parent == u64::MAX {
205 return Err(RustTransportError::InvalidAssertionContext(
206 "the reserved nesting sentinel cannot be a parent".into(),
207 ));
208 }
209 let digest = decision_id
210 .strip_prefix("rs:decision:")
211 .filter(|digest| digest.len() == 24 && digest.bytes().all(|byte| byte.is_ascii_hexdigit()))
212 .ok_or_else(|| {
213 RustTransportError::InvalidAssertionContext(format!(
214 "invalid decision ID {decision_id}"
215 ))
216 })?;
217 let id_high = u64::from_str_radix(&digest[..16], 16).map_err(|error| {
218 RustTransportError::InvalidAssertionContext(format!(
219 "invalid decision ID {decision_id}: {error}"
220 ))
221 })?;
222 let id_low = u32::from_str_radix(&digest[16..], 16).map_err(|error| {
223 RustTransportError::InvalidAssertionContext(format!(
224 "invalid decision ID {decision_id}: {error}"
225 ))
226 })?;
227 let mut value = 0xcbf2_9ce4_8422_2325_u64;
228 for byte in b"supercov-rust-assertion-phase-v2"
229 .iter()
230 .copied()
231 .chain(parent.to_le_bytes())
232 .chain(id_high.to_le_bytes())
233 .chain(id_low.to_le_bytes())
234 .chain(invocation_nonce.to_le_bytes())
235 {
236 value ^= u64::from(byte);
237 value = value.wrapping_mul(0x0000_0100_0000_01b3);
238 }
239 Ok(if matches!(value, 0 | u64::MAX) {
240 value ^ 0xa5a5_5a5a_d3c3_b4b4
241 } else {
242 value
243 })
244}
245
246pub fn rust_thread_context_id(parent: u64, invocation_nonce: u64) -> u64 {
250 let mut value = 0xcbf2_9ce4_8422_2325_u64;
251 for byte in b"supercov-rust-thread-phase-v1\0"
252 .iter()
253 .copied()
254 .chain(parent.to_le_bytes())
255 .chain(invocation_nonce.to_le_bytes())
256 {
257 value ^= u64::from(byte);
258 value = value.wrapping_mul(0x0000_0100_0000_01b3);
259 }
260 if matches!(value, 0 | u64::MAX) {
261 value ^ 0xa5a5_5a5a_d3c3_b4b4
262 } else {
263 value
264 }
265}
266
267fn put_u32(target: &mut [u8], offset: usize, value: u32) {
268 target[offset..offset + 4].copy_from_slice(&value.to_le_bytes());
269}
270
271fn get_u32(source: &[u8], offset: usize) -> Option<u32> {
272 Some(u32::from_le_bytes(
273 source.get(offset..offset + 4)?.try_into().ok()?,
274 ))
275}
276
277fn get_u64(source: &[u8], offset: usize) -> Option<u64> {
278 Some(u64::from_le_bytes(
279 source.get(offset..offset + 8)?.try_into().ok()?,
280 ))
281}
282
283fn total_size(descriptors: u32, payload: u32) -> Option<usize> {
284 HEADER_SIZE
285 .checked_add(
286 usize::try_from(descriptors)
287 .ok()?
288 .checked_mul(DESCRIPTOR_SIZE)?,
289 )?
290 .checked_add(usize::try_from(payload).ok()?)
291}
292
293fn map_mut(file: &File) -> Result<MmapMut, RustTransportError> {
294 unsafe { MmapOptions::new().map_mut(file) }
297 .map_err(|error| RustTransportError::Io(error.to_string()))
298}
299
300fn map(file: &File) -> Result<Mmap, RustTransportError> {
301 unsafe { MmapOptions::new().map(file) }
303 .map_err(|error| RustTransportError::Io(error.to_string()))
304}
305
306pub fn create_rust_transport(
307 path: &Path,
308 token: [u8; TOKEN_SIZE],
309 descriptor_capacity: u32,
310 payload_capacity: u32,
311) -> Result<(), RustTransportError> {
312 if descriptor_capacity == 0 || payload_capacity == 0 {
313 return Err(RustTransportError::InvalidLength);
314 }
315 let total = total_size(descriptor_capacity, payload_capacity)
316 .ok_or(RustTransportError::InvalidLength)?;
317 let mut options = OpenOptions::new();
318 options.read(true).write(true).create_new(true);
319 #[cfg(unix)]
320 {
321 use std::os::unix::fs::OpenOptionsExt as _;
322 options.mode(0o600);
323 }
324 let file = options
325 .open(path)
326 .map_err(|error| RustTransportError::Io(error.to_string()))?;
327 file.set_len(u64::try_from(total).map_err(|_| RustTransportError::InvalidLength)?)
328 .map_err(|error| RustTransportError::Io(error.to_string()))?;
329 let mut mapping = map_mut(&file)?;
330 mapping[..MAGIC.len()].copy_from_slice(MAGIC);
331 put_u32(&mut mapping, 8, VERSION);
332 put_u32(&mut mapping, 12, HEADER_SIZE as u32);
333 put_u32(&mut mapping, 16, DESCRIPTOR_SIZE as u32);
334 put_u32(&mut mapping, 20, descriptor_capacity);
335 put_u32(&mut mapping, 24, payload_capacity);
336 put_u32(&mut mapping, 28, ENDIAN_MARKER);
337 mapping[TOKEN_OFFSET..TOKEN_OFFSET + TOKEN_SIZE].copy_from_slice(&token);
338 mapping
339 .flush_range(0, HEADER_SIZE)
340 .map_err(|error| RustTransportError::Io(error.to_string()))
341}
342
343fn regular_file(path: &Path) -> Result<File, RustTransportError> {
344 let metadata =
345 fs::symlink_metadata(path).map_err(|error| RustTransportError::Io(error.to_string()))?;
346 if !metadata.file_type().is_file() {
347 return Err(RustTransportError::UnsafeFile(path.display().to_string()));
348 }
349 let mut options = OpenOptions::new();
350 options.read(true);
351 #[cfg(any(target_os = "linux", target_os = "macos"))]
352 {
353 use std::os::unix::fs::OpenOptionsExt as _;
354 #[cfg(target_os = "linux")]
355 const O_NOFOLLOW: i32 = 0x2_0000;
356 #[cfg(target_os = "macos")]
357 const O_NOFOLLOW: i32 = 0x100;
358 options.custom_flags(O_NOFOLLOW);
359 }
360 let file = options
361 .open(path)
362 .map_err(|error| RustTransportError::Io(error.to_string()))?;
363 if !file
364 .metadata()
365 .map_err(|error| RustTransportError::Io(error.to_string()))?
366 .file_type()
367 .is_file()
368 {
369 return Err(RustTransportError::UnsafeFile(path.display().to_string()));
370 }
371 Ok(file)
372}
373
374fn atomic_u64(mapping: &[u8], offset: usize) -> Result<&AtomicU64, RustTransportError> {
375 let pointer = mapping
376 .get(offset..offset + 8)
377 .ok_or(RustTransportError::InvalidHeader)?
378 .as_ptr();
379 if !(pointer as usize).is_multiple_of(std::mem::align_of::<AtomicU64>()) {
380 return Err(RustTransportError::InvalidHeader);
381 }
382 Ok(unsafe { &*pointer.cast::<AtomicU64>() })
385}
386
387fn atomic_u8(mapping: &[u8], offset: usize) -> Result<&AtomicU8, RustTransportError> {
388 let pointer = mapping
389 .get(offset)
390 .ok_or(RustTransportError::InvalidLength)? as *const u8;
391 Ok(unsafe { &*pointer.cast::<AtomicU8>() })
393}
394
395#[allow(clippy::too_many_arguments)]
396fn checksum(
397 kind: u8,
398 outcome: u8,
399 pid: u32,
400 context: u64,
401 payload_offset: u32,
402 payload_length: u32,
403 id_length: u32,
404 value_length: u32,
405 id: &[u8],
406 values: &[u8],
407) -> u64 {
408 let mut value = 0xcbf2_9ce4_8422_2325_u64;
409 for byte in [kind, outcome]
410 .into_iter()
411 .chain(pid.to_le_bytes())
412 .chain(context.to_le_bytes())
413 .chain(payload_offset.to_le_bytes())
414 .chain(payload_length.to_le_bytes())
415 .chain(id_length.to_le_bytes())
416 .chain(value_length.to_le_bytes())
417 .chain(id.iter().copied())
418 .chain(values.iter().copied())
419 {
420 value ^= u64::from(byte);
421 value = value.wrapping_mul(0x0000_0100_0000_01b3);
422 }
423 value
424}
425
426pub fn read_rust_transport(
427 path: &Path,
428 expected_token: &[u8; TOKEN_SIZE],
429) -> Result<RustTransportRead, RustTransportError> {
430 let file = regular_file(path)?;
431 let mapping = map(&file)?;
432 if mapping.get(..8) != Some(MAGIC.as_slice())
433 || get_u32(&mapping, 8) != Some(VERSION)
434 || get_u32(&mapping, 12) != Some(HEADER_SIZE as u32)
435 || get_u32(&mapping, 16) != Some(DESCRIPTOR_SIZE as u32)
436 || get_u32(&mapping, 28) != Some(ENDIAN_MARKER)
437 || mapping.get(TOKEN_OFFSET..TOKEN_OFFSET + TOKEN_SIZE) != Some(expected_token.as_slice())
438 || mapping.get(52..56).is_none_or(|bytes| bytes != [0; 4])
439 || mapping
440 .get(NEXT_PHASE_OFFSET + 8..HEADER_SIZE)
441 .is_none_or(|bytes| bytes != [0; 40])
442 {
443 return Err(RustTransportError::InvalidHeader);
444 }
445 let descriptors = get_u32(&mapping, 20).ok_or(RustTransportError::InvalidHeader)?;
446 let payload_capacity = get_u32(&mapping, 24).ok_or(RustTransportError::InvalidHeader)?;
447 if descriptors == 0 || payload_capacity == 0 {
448 return Err(RustTransportError::InvalidHeader);
449 }
450 if mapping.len()
451 != total_size(descriptors, payload_capacity).ok_or(RustTransportError::InvalidLength)?
452 {
453 return Err(RustTransportError::InvalidLength);
454 }
455 let next = atomic_u64(&mapping, NEXT_DESCRIPTOR_OFFSET)?.load(Ordering::Acquire);
456 let next_payload = atomic_u64(&mapping, NEXT_PAYLOAD_OFFSET)?.load(Ordering::Acquire);
457 let recorded_dropped = atomic_u64(&mapping, DROPPED_OFFSET)?.load(Ordering::Acquire);
458 let attachments = atomic_u64(&mapping, ATTACHMENTS_OFFSET)?.load(Ordering::Acquire);
459 let inspect = next.min(u64::from(descriptors));
460 let payload_base = HEADER_SIZE + descriptors as usize * DESCRIPTOR_SIZE;
461 let mut observations = Vec::new();
462 let mut ordinal_hits = Vec::new();
463 let mut phases = Vec::new();
464 let mut thread_phases = Vec::new();
465 let mut thread_ends = Vec::new();
466 let mut test_boundaries = Vec::new();
467 let mut phase_definitions = BTreeMap::<u64, (u8, u64, u64, String)>::new();
468 let mut thread_end_contexts = BTreeSet::<u64>::new();
469 let mut boundary_contexts = BTreeSet::<u64>::new();
470 let mut committed = 0_u64;
471 for index in 0..inspect {
472 let descriptor = HEADER_SIZE + index as usize * DESCRIPTOR_SIZE;
473 match atomic_u8(&mapping, descriptor + COMMIT_OFFSET)?.load(Ordering::Acquire) {
474 0 => continue,
475 1 => {}
476 _ => return Err(RustTransportError::InvalidDescriptor(index)),
477 }
478 committed += 1;
479 let kind = mapping[descriptor + KIND_OFFSET];
480 let outcome = mapping[descriptor + OUTCOME_OFFSET];
481 if mapping[descriptor + 3] != 0 {
482 return Err(RustTransportError::InvalidDescriptor(index));
483 }
484 let pid = get_u32(&mapping, descriptor + PID_OFFSET)
485 .ok_or(RustTransportError::InvalidDescriptor(index))?;
486 let context_id = get_u64(&mapping, descriptor + CONTEXT_OFFSET)
487 .ok_or(RustTransportError::InvalidDescriptor(index))?;
488 let payload_offset = get_u32(&mapping, descriptor + PAYLOAD_OFFSET_OFFSET)
489 .ok_or(RustTransportError::InvalidDescriptor(index))?
490 as usize;
491 let payload_length = get_u32(&mapping, descriptor + PAYLOAD_LENGTH_OFFSET)
492 .ok_or(RustTransportError::InvalidDescriptor(index))?
493 as usize;
494 let id_length = get_u32(&mapping, descriptor + ID_LENGTH_OFFSET)
495 .ok_or(RustTransportError::InvalidDescriptor(index))? as usize;
496 let value_length = get_u32(&mapping, descriptor + VALUE_LENGTH_OFFSET)
497 .ok_or(RustTransportError::InvalidDescriptor(index))?
498 as usize;
499 let expected_checksum = get_u64(&mapping, descriptor + CHECKSUM_OFFSET)
500 .ok_or(RustTransportError::InvalidDescriptor(index))?;
501 let end = payload_offset
502 .checked_add(payload_length)
503 .filter(|end| *end <= payload_capacity as usize)
504 .ok_or(RustTransportError::InvalidDescriptor(index))?;
505 if payload_length != id_length.saturating_add(value_length)
506 || u64::from(end as u32) > next_payload
507 {
508 return Err(RustTransportError::InvalidDescriptor(index));
509 }
510 let payload = mapping
511 .get(payload_base + payload_offset..payload_base + end)
512 .ok_or(RustTransportError::InvalidDescriptor(index))?;
513 let (id, values) = payload.split_at(id_length);
514 if checksum(
515 kind,
516 outcome,
517 pid,
518 context_id,
519 payload_offset as u32,
520 payload_length as u32,
521 id_length as u32,
522 value_length as u32,
523 id,
524 values,
525 ) != expected_checksum
526 {
527 return Err(RustTransportError::InvalidRecord(index));
528 }
529 let id = std::str::from_utf8(id).map_err(|_| RustTransportError::InvalidRecord(index))?;
530 if matches!(kind, KIND_HIT | KIND_DECISION | KIND_PHASE) && !valid_probe_id(id) {
531 return Err(RustTransportError::InvalidRecord(index));
532 }
533 match kind {
534 KIND_HIT if outcome == 0 && values.is_empty() => {
535 observations.push(RustTransportObservation {
536 process_id: pid,
537 context_id,
538 observation: RustProbeObservation::Hit { id: id.into() },
539 });
540 }
541 KIND_DECISION
542 if matches!(outcome, 0 | 1)
543 && id.starts_with("rs:decision:")
544 && !values.is_empty()
545 && values.iter().all(|value| matches!(*value, 0..=2)) =>
546 {
547 observations.push(RustTransportObservation {
548 process_id: pid,
549 context_id,
550 observation: RustProbeObservation::Decision {
551 id: id.into(),
552 values: values
553 .iter()
554 .map(|value| match value {
555 0 => None,
556 1 => Some(false),
557 2 => Some(true),
558 _ => unreachable!(),
559 })
560 .collect(),
561 outcome: outcome == 1,
562 },
563 });
564 }
565 KIND_ORDINAL_HIT if outcome == 0 && id.is_empty() && values.len() == 8 => {
566 ordinal_hits.push(RustOrdinalHit {
567 process_id: pid,
568 context_id,
569 ordinal: u64::from_le_bytes(
570 values
571 .try_into()
572 .map_err(|_| RustTransportError::InvalidRecord(index))?,
573 ),
574 });
575 }
576 KIND_PHASE
577 if outcome == 0
578 && id.starts_with("rs:decision:")
579 && values.len() == 16
580 && !matches!(context_id, 0 | u64::MAX) =>
581 {
582 let parent_context_id = u64::from_le_bytes(
583 values[..8]
584 .try_into()
585 .map_err(|_| RustTransportError::InvalidRecord(index))?,
586 );
587 let invocation_nonce = u64::from_le_bytes(
588 values[8..]
589 .try_into()
590 .map_err(|_| RustTransportError::InvalidRecord(index))?,
591 );
592 if matches!(parent_context_id, 0 | u64::MAX)
593 || rust_assertion_context_id(parent_context_id, id, invocation_nonce)?
594 != context_id
595 {
596 return Err(RustTransportError::InvalidAssertionContext(format!(
597 "phase record {index} does not derive child {context_id:016x} from parent {parent_context_id:016x} and {id}"
598 )));
599 }
600 let definition = (
601 KIND_PHASE,
602 parent_context_id,
603 invocation_nonce,
604 id.to_owned(),
605 );
606 if phase_definitions
607 .insert(context_id, definition.clone())
608 .is_some_and(|existing| existing != definition)
609 {
610 return Err(RustTransportError::InvalidAssertionContext(format!(
611 "child {context_id:016x} has conflicting phase definitions"
612 )));
613 }
614 phases.push(RustPhaseContext {
615 process_id: pid,
616 child_context_id: context_id,
617 parent_context_id,
618 invocation_nonce,
619 decision_id: id.into(),
620 });
621 }
622 KIND_THREAD_PHASE
623 if outcome == 0
624 && id.is_empty()
625 && values.len() == 16
626 && !matches!(context_id, 0 | u64::MAX) =>
627 {
628 let parent_context_id = u64::from_le_bytes(
629 values[..8]
630 .try_into()
631 .map_err(|_| RustTransportError::InvalidRecord(index))?,
632 );
633 let invocation_nonce = u64::from_le_bytes(
634 values[8..]
635 .try_into()
636 .map_err(|_| RustTransportError::InvalidRecord(index))?,
637 );
638 if matches!(parent_context_id, 0 | u64::MAX)
639 || rust_thread_context_id(parent_context_id, invocation_nonce) != context_id
640 {
641 return Err(RustTransportError::InvalidAssertionContext(format!(
642 "thread phase record {index} does not derive child {context_id:016x} from parent {parent_context_id:016x} and nonce {invocation_nonce}"
643 )));
644 }
645 let definition = (
646 KIND_THREAD_PHASE,
647 parent_context_id,
648 invocation_nonce,
649 String::new(),
650 );
651 if phase_definitions
652 .insert(context_id, definition.clone())
653 .is_some_and(|existing| existing != definition)
654 {
655 return Err(RustTransportError::InvalidAssertionContext(format!(
656 "child {context_id:016x} has conflicting phase definitions"
657 )));
658 }
659 thread_phases.push(RustThreadPhase {
660 process_id: pid,
661 child_context_id: context_id,
662 parent_context_id,
663 invocation_nonce,
664 commit_index: index,
665 });
666 }
667 KIND_THREAD_END
668 if outcome == 0
669 && id.is_empty()
670 && values.is_empty()
671 && !matches!(context_id, 0 | u64::MAX) =>
672 {
673 if !thread_end_contexts.insert(context_id) {
674 return Err(RustTransportError::InvalidRecord(index));
675 }
676 thread_ends.push(RustThreadEnd {
677 process_id: pid,
678 context_id,
679 commit_index: index,
680 });
681 }
682 KIND_TEST_BOUNDARY
683 if outcome == 0
684 && id.is_empty()
685 && values.is_empty()
686 && !matches!(context_id, 0 | u64::MAX) =>
687 {
688 if !boundary_contexts.insert(context_id) {
689 return Err(RustTransportError::InvalidRecord(index));
690 }
691 test_boundaries.push(RustTestBoundary {
692 process_id: pid,
693 context_id,
694 commit_index: index,
695 });
696 }
697 _ => return Err(RustTransportError::InvalidRecord(index)),
698 }
699 }
700 let overflow = next.saturating_sub(u64::from(descriptors));
701 Ok(RustTransportRead {
702 observations,
703 ordinal_hits,
704 phases,
705 thread_phases,
706 thread_ends,
707 test_boundaries,
708 committed,
709 incomplete: inspect.saturating_sub(committed),
710 dropped: recorded_dropped.max(overflow),
711 attachments,
712 })
713}
714
715pub fn validate_rust_phase_contexts(
716 base_context_id: u64,
717 read: &RustTransportRead,
718) -> Result<(), RustTransportError> {
719 if matches!(base_context_id, 0 | u64::MAX) {
720 return Err(RustTransportError::InvalidAssertionContext(
721 "the supervisor base context must be nonzero and not reserved".into(),
722 ));
723 }
724 let mut definitions = BTreeMap::<u64, (u8, u64, u64, &str)>::new();
725 for phase in &read.phases {
726 let definition = (
727 KIND_PHASE,
728 phase.parent_context_id,
729 phase.invocation_nonce,
730 phase.decision_id.as_str(),
731 );
732 if definitions
733 .insert(phase.child_context_id, definition)
734 .is_some_and(|existing| existing != definition)
735 {
736 return Err(RustTransportError::InvalidAssertionContext(format!(
737 "child {:016x} has conflicting phase definitions",
738 phase.child_context_id
739 )));
740 }
741 }
742 for phase in &read.thread_phases {
743 if rust_thread_context_id(phase.parent_context_id, phase.invocation_nonce)
744 != phase.child_context_id
745 {
746 return Err(RustTransportError::InvalidAssertionContext(format!(
747 "thread phase child {:016x} does not derive from parent {:016x} and nonce {}",
748 phase.child_context_id, phase.parent_context_id, phase.invocation_nonce
749 )));
750 }
751 let definition = (
752 KIND_THREAD_PHASE,
753 phase.parent_context_id,
754 phase.invocation_nonce,
755 "",
756 );
757 if definitions
758 .insert(phase.child_context_id, definition)
759 .is_some_and(|existing| existing != definition)
760 {
761 return Err(RustTransportError::InvalidAssertionContext(format!(
762 "child {:016x} has conflicting phase definitions",
763 phase.child_context_id
764 )));
765 }
766 }
767
768 let used_contexts = read
769 .observations
770 .iter()
771 .map(|observation| observation.context_id)
772 .chain(read.ordinal_hits.iter().map(|hit| hit.context_id))
773 .chain(read.thread_ends.iter().map(|end| end.context_id))
774 .chain(
775 read.test_boundaries
776 .iter()
777 .map(|boundary| boundary.context_id),
778 )
779 .filter(|context| *context != 0 && *context != base_context_id)
780 .collect::<BTreeSet<_>>();
781 for start in used_contexts {
782 let mut context = start;
783 let mut path = BTreeSet::new();
784 while context != base_context_id {
785 if !path.insert(context) {
786 return Err(RustTransportError::InvalidAssertionContext(format!(
787 "phase context cycle at {context:016x}"
788 )));
789 }
790 let (_, parent, _, _) = definitions.get(&context).ok_or_else(|| {
791 RustTransportError::InvalidAssertionContext(format!(
792 "context {context:016x} does not resolve to base {base_context_id:016x}"
793 ))
794 })?;
795 context = *parent;
796 if matches!(context, 0 | u64::MAX) {
797 return Err(RustTransportError::InvalidAssertionContext(format!(
798 "context {start:016x} crosses the supervisor attempt boundary"
799 )));
800 }
801 }
802 }
803 Ok(())
804}
805
806pub fn partition_rust_transport_by_test_contexts(
810 read: &RustTransportRead,
811 base_contexts: &BTreeSet<u64>,
812) -> Result<RustTransportPartition, RustTransportError> {
813 if base_contexts
814 .iter()
815 .any(|context| matches!(*context, 0 | u64::MAX))
816 {
817 return Err(RustTransportError::InvalidAttribution(
818 "known test roots must be nonzero and non-reserved".into(),
819 ));
820 }
821 if read.incomplete != 0 || read.dropped != 0 {
822 return Err(RustTransportError::InvalidAttribution(format!(
823 "shared transport is incomplete (incomplete={}, dropped={})",
824 read.incomplete, read.dropped
825 )));
826 }
827
828 let mut parents = BTreeMap::<u64, u64>::new();
829 let mut thread_children = BTreeSet::<u64>::new();
830 for (child, parent) in read
831 .phases
832 .iter()
833 .map(|phase| (phase.child_context_id, phase.parent_context_id))
834 .chain(
835 read.thread_phases
836 .iter()
837 .map(|phase| (phase.child_context_id, phase.parent_context_id)),
838 )
839 {
840 if matches!(child, 0 | u64::MAX)
841 || matches!(parent, 0 | u64::MAX)
842 || base_contexts.contains(&child)
843 || parents.insert(child, parent).is_some()
844 {
845 return Err(RustTransportError::InvalidAttribution(format!(
846 "phase {child:016x} has an invalid or repeated ownership definition"
847 )));
848 }
849 }
850 for phase in &read.thread_phases {
851 thread_children.insert(phase.child_context_id);
852 }
853 let mut thread_end_index = BTreeMap::<u64, u64>::new();
854 for end in &read.thread_ends {
855 if !thread_children.contains(&end.context_id)
856 || thread_end_index
857 .insert(end.context_id, end.commit_index)
858 .is_some()
859 {
860 return Err(RustTransportError::InvalidAttribution(format!(
861 "thread end {:016x} has no unique thread-phase definition",
862 end.context_id
863 )));
864 }
865 }
866 let mut boundary_index = BTreeMap::<u64, u64>::new();
867 for boundary in &read.test_boundaries {
868 if !base_contexts.contains(&boundary.context_id)
869 || boundary_index
870 .insert(boundary.context_id, boundary.commit_index)
871 .is_some()
872 {
873 return Err(RustTransportError::InvalidAttribution(format!(
874 "test boundary {:016x} does not identify one known test root",
875 boundary.context_id
876 )));
877 }
878 }
879
880 enum Resolution {
883 Attributed(u64),
884 Background,
885 Quarantined { root: u64, escaped: Vec<u64> },
886 }
887 let resolve = |start: u64| -> Result<Resolution, RustTransportError> {
888 if start == 0 {
889 return Ok(Resolution::Background);
890 }
891 if start == u64::MAX {
892 return Err(RustTransportError::InvalidAttribution(
893 "evidence used the reserved context sentinel".into(),
894 ));
895 }
896 let mut context = start;
897 let mut seen = BTreeSet::new();
898 let mut chain_threads = Vec::new();
899 loop {
900 if thread_children.contains(&context) {
901 chain_threads.push(context);
902 }
903 let Some(parent) = parents.get(&context) else {
904 break;
905 };
906 if !seen.insert(context) {
907 return Err(RustTransportError::InvalidAttribution(format!(
908 "phase context cycle at {context:016x}"
909 )));
910 }
911 context = *parent;
912 }
913 if !base_contexts.contains(&context) {
914 return Err(RustTransportError::InvalidAttribution(format!(
915 "context {start:016x} resolves to unknown test root {context:016x}"
916 )));
917 }
918 let boundary = boundary_index.get(&context).copied();
919 let escaped = chain_threads
920 .into_iter()
921 .filter(|thread| {
922 !matches!(
923 (thread_end_index.get(thread), boundary),
924 (Some(end), Some(boundary)) if *end < boundary
925 )
926 })
927 .collect::<Vec<_>>();
928 if escaped.is_empty() {
929 Ok(Resolution::Attributed(context))
930 } else {
931 Ok(Resolution::Quarantined {
932 root: context,
933 escaped,
934 })
935 }
936 };
937
938 let mut attributed = base_contexts
939 .iter()
940 .map(|context| (*context, RustTransportRead::empty()))
941 .collect::<BTreeMap<_, _>>();
942 let mut background = RustTransportRead::empty();
943 background.attachments = read.attachments;
944 let mut thread_scope_limitations = BTreeSet::new();
945 let mut note_escape = |root: u64, escaped: Vec<u64>| {
946 for thread in escaped {
947 thread_scope_limitations.insert(format!(
948 "RUST_THREAD_OUTLIVED_TEST: thread phase {thread:016x} escaped test {root:016x}"
949 ));
950 }
951 };
952
953 for observation in &read.observations {
954 match resolve(observation.context_id)? {
955 Resolution::Attributed(root) => attributed
956 .get_mut(&root)
957 .expect("resolved root was preallocated")
958 .observations
959 .push(observation.clone()),
960 Resolution::Background => background.observations.push(observation.clone()),
961 Resolution::Quarantined { root, escaped } => {
962 note_escape(root, escaped);
963 background.observations.push(observation.clone());
964 }
965 }
966 }
967 for hit in &read.ordinal_hits {
968 match resolve(hit.context_id)? {
969 Resolution::Attributed(root) => attributed
970 .get_mut(&root)
971 .expect("resolved root was preallocated")
972 .ordinal_hits
973 .push(*hit),
974 Resolution::Background => background.ordinal_hits.push(*hit),
975 Resolution::Quarantined { root, escaped } => {
976 note_escape(root, escaped);
977 background.ordinal_hits.push(*hit);
978 }
979 }
980 }
981 for phase in &read.phases {
982 match resolve(phase.child_context_id)? {
983 Resolution::Attributed(root) => attributed
984 .get_mut(&root)
985 .expect("resolved root was preallocated")
986 .phases
987 .push(phase.clone()),
988 Resolution::Background => {
989 return Err(RustTransportError::InvalidAttribution(format!(
990 "phase {:016x} resolved to background",
991 phase.child_context_id
992 )));
993 }
994 Resolution::Quarantined { root, escaped } => {
995 note_escape(root, escaped);
996 background.phases.push(phase.clone());
997 }
998 }
999 }
1000 for phase in &read.thread_phases {
1001 match resolve(phase.child_context_id)? {
1002 Resolution::Attributed(root) => attributed
1003 .get_mut(&root)
1004 .expect("resolved root was preallocated")
1005 .thread_phases
1006 .push(*phase),
1007 Resolution::Background => {
1008 return Err(RustTransportError::InvalidAttribution(format!(
1009 "thread phase {:016x} resolved to background",
1010 phase.child_context_id
1011 )));
1012 }
1013 Resolution::Quarantined { root, escaped } => {
1014 note_escape(root, escaped);
1015 background.thread_phases.push(*phase);
1016 }
1017 }
1018 }
1019 for end in &read.thread_ends {
1020 match resolve(end.context_id)? {
1021 Resolution::Attributed(root) => attributed
1022 .get_mut(&root)
1023 .expect("resolved root was preallocated")
1024 .thread_ends
1025 .push(*end),
1026 Resolution::Background => {
1027 return Err(RustTransportError::InvalidAttribution(format!(
1028 "thread end {:016x} resolved to background",
1029 end.context_id
1030 )));
1031 }
1032 Resolution::Quarantined { root, escaped } => {
1033 note_escape(root, escaped);
1034 background.thread_ends.push(*end);
1035 }
1036 }
1037 }
1038 for boundary in &read.test_boundaries {
1039 attributed
1040 .get_mut(&boundary.context_id)
1041 .expect("boundary contexts were checked against the known roots")
1042 .test_boundaries
1043 .push(*boundary);
1044 }
1045
1046 let set_committed = |transport: &mut RustTransportRead| -> Result<u64, RustTransportError> {
1047 transport.committed = u64::try_from(
1048 transport.observations.len()
1049 + transport.ordinal_hits.len()
1050 + transport.phases.len()
1051 + transport.thread_phases.len()
1052 + transport.thread_ends.len()
1053 + transport.test_boundaries.len(),
1054 )
1055 .map_err(|_| {
1056 RustTransportError::InvalidAttribution("partition record count exceeds u64".into())
1057 })?;
1058 Ok(transport.committed)
1059 };
1060 let mut assigned = set_committed(&mut background)?;
1061 for transport in attributed.values_mut() {
1062 assigned = assigned
1063 .checked_add(set_committed(transport)?)
1064 .ok_or_else(|| {
1065 RustTransportError::InvalidAttribution(
1066 "partition record count overflowed u64".into(),
1067 )
1068 })?;
1069 }
1070 if assigned != read.committed {
1071 return Err(RustTransportError::InvalidAttribution(format!(
1072 "partition assigned {assigned} of {} committed records",
1073 read.committed
1074 )));
1075 }
1076 for (base, transport) in &attributed {
1077 validate_rust_phase_contexts(*base, transport)?;
1078 }
1079 Ok(RustTransportPartition {
1080 attributed,
1081 background,
1082 thread_scope_limitations,
1083 })
1084}
1085
1086pub fn render_rust_mmap_runtime(module_name: &str) -> Result<String, String> {
1087 let valid_identifier = !module_name.is_empty()
1088 && module_name.bytes().enumerate().all(|(index, byte)| {
1089 byte == b'_' || byte.is_ascii_alphabetic() || (index > 0 && byte.is_ascii_digit())
1090 });
1091 if !valid_identifier {
1092 return Err("invalid Rust runtime module name".into());
1093 }
1094 Ok(RUNTIME_TEMPLATE.replace("__SUPERCOV_MODULE__", module_name))
1095}
1096
1097#[cfg(test)]
1098mod tests {
1099 use std::{
1100 fs,
1101 io::{BufRead as _, BufReader},
1102 process::{Command, Stdio},
1103 sync::atomic::{AtomicUsize, Ordering},
1104 time::{SystemTime, UNIX_EPOCH},
1105 };
1106
1107 use super::*;
1108
1109 const TOKEN: [u8; TOKEN_SIZE] = [0x42; TOKEN_SIZE];
1110 const CONTEXT: u64 = 42;
1111
1112 fn token_hex() -> String {
1113 TOKEN.iter().map(|byte| format!("{byte:02x}")).collect()
1114 }
1115
1116 fn temporary_directory(name: &str) -> std::path::PathBuf {
1117 let nonce = SystemTime::now()
1118 .duration_since(UNIX_EPOCH)
1119 .unwrap()
1120 .as_nanos();
1121 let path = std::env::temp_dir().join(format!(
1122 "supercov-rust-transport-{}-{nonce}-{name}",
1123 std::process::id()
1124 ));
1125 fs::create_dir(&path).unwrap();
1126 path
1127 }
1128
1129 fn compile_fixture(directory: &Path) -> std::path::PathBuf {
1130 let source = directory.join("main.rs");
1131 let binary = directory.join("program");
1132 let runtime = render_rust_mmap_runtime("__supercov_runtime_v1").unwrap();
1133 fs::write(
1134 &source,
1135 format!(
1136 r#"{runtime}
1137fn main() {{
1138 let mode = std::env::args().nth(1).unwrap_or_default();
1139 if mode == "threads" {{
1140 let mut threads = Vec::new();
1141 for _ in 0..8 {{
1142 threads.push(std::thread::spawn(|| {{
1143 for _ in 0..100 {{ __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567"); }}
1144 }}));
1145 }}
1146 for thread in threads {{ thread.join().unwrap(); }}
1147 let mut frame = __supercov_runtime_v1::DecisionFrame::new("rs:decision:0123456789abcdef01234567", 2);
1148 let first = __supercov_runtime_v1::condition(true, &mut frame, 0);
1149 let second = __supercov_runtime_v1::condition(false, &mut frame, 1);
1150 __supercov_runtime_v1::decision(first && second, &mut frame);
1151 __supercov_runtime_v1::ordinal_hit(7);
1152 }} else if mode == "contexts" {{
1153 __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1154 let outer = __supercov_runtime_v1::enter_context(100);
1155 __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1156 let inner = __supercov_runtime_v1::enter_context(200);
1157 __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1158 __supercov_runtime_v1::exit_context(inner);
1159 __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1160 __supercov_runtime_v1::exit_context(outer);
1161 __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1162 let assertion = __supercov_runtime_v1::enter_assertion_context(
1163 0x0123_4567_89ab_cdef,
1164 0x0123_4567,
1165 );
1166 __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1167 __supercov_runtime_v1::exit_context(assertion);
1168 __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1169 let repeated_assertion = __supercov_runtime_v1::enter_assertion_context(
1170 0x0123_4567_89ab_cdef,
1171 0x0123_4567,
1172 );
1173 __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1174 __supercov_runtime_v1::exit_context(repeated_assertion);
1175 __supercov_runtime_v1::hit("rs:statement:0123456789abcdef01234567");
1176 }} else if mode == "mir-decisions" {{
1177 let before_outer = __supercov_runtime_v1::enter_context(901);
1178 let outer = __supercov_runtime_v1::mir_decision_start(
1179 0x0123_4567_89ab_cdef,
1180 0x0123_4567,
1181 2,
1182 );
1183 __supercov_runtime_v1::mir_decision_condition(outer, 0, true);
1184 let before_inner = __supercov_runtime_v1::enter_context(902);
1185 let inner = __supercov_runtime_v1::mir_decision_start(
1186 0xfedc_ba98_7654_3210,
1187 0xfedc_ba98,
1188 1,
1189 );
1190 let migrated = std::thread::spawn(move || {{
1191 __supercov_runtime_v1::mir_decision_condition(inner, 0, false);
1192 __supercov_runtime_v1::mir_decision_finish(inner, false);
1193 }});
1194 __supercov_runtime_v1::exit_context(before_inner);
1195 __supercov_runtime_v1::mir_decision_condition(outer, 1, true);
1196 __supercov_runtime_v1::exit_context(before_outer);
1197 __supercov_runtime_v1::mir_decision_finish(outer, true);
1198 migrated.join().unwrap();
1199 let before_branch = __supercov_runtime_v1::enter_context(903);
1200 let branch = __supercov_runtime_v1::mir_branch_start();
1201 __supercov_runtime_v1::exit_context(before_branch);
1202 let migrated_branch = std::thread::spawn(move || {{
1203 __supercov_runtime_v1::mir_branch_hit(branch, 777);
1204 __supercov_runtime_v1::mir_branch_hit(branch, 888);
1205 }});
1206 migrated_branch.join().unwrap();
1207 }} else if mode == "kill" {{
1208 __supercov_runtime_v1::hit("rs:function:fedcba9876543210fedcba98");
1209 let interrupted = __supercov_runtime_v1::mir_decision_start(
1210 0x1111_1111_1111_1111,
1211 0x2222_2222,
1212 2,
1213 );
1214 __supercov_runtime_v1::mir_decision_condition(interrupted, 0, true);
1215 let _interrupted_branch = __supercov_runtime_v1::mir_branch_start();
1216 println!("ready");
1217 use std::io::Write as _;
1218 std::io::stdout().flush().unwrap();
1219 std::thread::sleep(std::time::Duration::from_secs(30));
1220 }} else {{
1221 for _ in 0..3 {{ __supercov_runtime_v1::hit("rs:branch:aaaaaaaaaaaaaaaaaaaaaaaa"); }}
1222 }}
1223}}
1224"#
1225 ),
1226 )
1227 .unwrap();
1228 let output = Command::new("rustc")
1229 .args(["--edition=2024"])
1230 .arg(&source)
1231 .arg("-o")
1232 .arg(&binary)
1233 .output()
1234 .unwrap();
1235 assert!(
1236 output.status.success(),
1237 "{}",
1238 String::from_utf8_lossy(&output.stderr)
1239 );
1240 binary
1241 }
1242
1243 #[test]
1244 fn implementation_matches_frozen_transport_contract() {
1245 let contract = supercov_contracts::rust_probe_transport_v3_contract().unwrap();
1246 assert_eq!(MAGIC.as_slice(), contract.magic.as_bytes());
1247 assert_eq!(VERSION, contract.protocol_version);
1248 assert_eq!(HEADER_SIZE, contract.header_size);
1249 assert_eq!(DESCRIPTOR_SIZE, contract.descriptor_size);
1250 assert_eq!(TOKEN_SIZE, contract.token_size);
1251 assert_eq!(ENDIAN_MARKER, contract.endian_marker);
1252 assert_eq!(
1253 NEXT_DESCRIPTOR_OFFSET,
1254 contract.header_offsets.next_descriptor
1255 );
1256 assert_eq!(NEXT_PAYLOAD_OFFSET, contract.header_offsets.next_payload);
1257 assert_eq!(DROPPED_OFFSET, contract.header_offsets.dropped);
1258 assert_eq!(TOKEN_OFFSET, contract.header_offsets.token);
1259 assert_eq!(ATTACHMENTS_OFFSET, contract.header_offsets.attachments);
1260 assert_eq!(Some(NEXT_PHASE_OFFSET), contract.header_offsets.next_phase);
1261 assert_eq!(COMMIT_OFFSET, contract.descriptor_offsets.commit);
1262 assert_eq!(PID_OFFSET, contract.descriptor_offsets.process_id);
1263 assert_eq!(CONTEXT_OFFSET, contract.descriptor_offsets.context_id);
1264 assert_eq!(
1265 PAYLOAD_OFFSET_OFFSET,
1266 contract.descriptor_offsets.payload_offset
1267 );
1268 assert_eq!(CHECKSUM_OFFSET, contract.descriptor_offsets.checksum);
1269 assert_eq!(KIND_HIT, contract.record_kinds.hit);
1270 assert_eq!(KIND_DECISION, contract.record_kinds.decision);
1271 assert_eq!(KIND_ORDINAL_HIT, contract.record_kinds.ordinal_hit);
1272 assert_eq!(Some(KIND_PHASE), contract.record_kinds.phase);
1273 assert_eq!(Some(KIND_THREAD_PHASE), contract.record_kinds.thread_phase);
1274 assert_eq!(Some(KIND_THREAD_END), contract.record_kinds.thread_end);
1275 assert_eq!(
1276 Some(KIND_TEST_BOUNDARY),
1277 contract.record_kinds.test_boundary
1278 );
1279 assert!(RUNTIME_TEMPLATE.contains("b\"SCVRUST3\""));
1280 assert!(RUNTIME_TEMPLATE.contains("const DESCRIPTOR_SIZE: usize = 40;"));
1281 }
1282
1283 #[test]
1284 fn assertion_context_derivation_is_exact_nested_and_never_promotes_background() {
1285 let first =
1286 rust_assertion_context_id(CONTEXT, "rs:decision:0123456789abcdef01234567", 0).unwrap();
1287 let nested =
1288 rust_assertion_context_id(first, "rs:decision:fedcba9876543210fedcba98", 1).unwrap();
1289 assert_ne!(first, CONTEXT);
1290 assert_ne!(nested, first);
1291 assert_eq!(
1292 rust_assertion_context_id(0, "rs:decision:0123456789abcdef01234567", 2).unwrap(),
1293 0
1294 );
1295 assert!(matches!(
1296 rust_assertion_context_id(CONTEXT, "not-a-decision", 2),
1297 Err(RustTransportError::InvalidAssertionContext(_))
1298 ));
1299 assert!(matches!(
1300 rust_assertion_context_id(u64::MAX, "rs:decision:0123456789abcdef01234567", 2),
1301 Err(RustTransportError::InvalidAssertionContext(_))
1302 ));
1303 }
1304
1305 #[test]
1306 fn phase_context_chains_resolve_exactly_to_the_supervisor_attempt() {
1307 let outer_id = "rs:decision:0123456789abcdef01234567";
1308 let inner_id = "rs:decision:fedcba9876543210fedcba98";
1309 let outer = rust_assertion_context_id(CONTEXT, outer_id, 10).unwrap();
1310 let inner = rust_assertion_context_id(outer, inner_id, 11).unwrap();
1311 let read = RustTransportRead {
1312 observations: vec![RustTransportObservation {
1313 process_id: 7,
1314 context_id: inner,
1315 observation: RustProbeObservation::Hit {
1316 id: "rs:statement:0123456789abcdef01234567".into(),
1317 },
1318 }],
1319 ordinal_hits: Vec::new(),
1320 phases: vec![
1321 RustPhaseContext {
1322 process_id: 7,
1323 child_context_id: outer,
1324 parent_context_id: CONTEXT,
1325 invocation_nonce: 10,
1326 decision_id: outer_id.into(),
1327 },
1328 RustPhaseContext {
1329 process_id: 7,
1330 child_context_id: inner,
1331 parent_context_id: outer,
1332 invocation_nonce: 11,
1333 decision_id: inner_id.into(),
1334 },
1335 ],
1336 thread_phases: Vec::new(),
1337 thread_ends: Vec::new(),
1338 test_boundaries: Vec::new(),
1339 committed: 3,
1340 incomplete: 0,
1341 dropped: 0,
1342 attachments: 1,
1343 };
1344 validate_rust_phase_contexts(CONTEXT, &read).unwrap();
1345
1346 let mut threaded = read.clone();
1347 let thread_child = rust_thread_context_id(inner, 7);
1348 threaded.thread_phases.push(RustThreadPhase {
1349 process_id: 7,
1350 child_context_id: thread_child,
1351 parent_context_id: inner,
1352 invocation_nonce: 7,
1353 commit_index: 3,
1354 });
1355 threaded.observations[0].context_id = thread_child;
1356 threaded.committed = 4;
1357 validate_rust_phase_contexts(CONTEXT, &threaded).unwrap();
1358
1359 let mut tampered_thread = threaded.clone();
1360 tampered_thread.thread_phases[0].invocation_nonce = 8;
1361 assert!(matches!(
1362 validate_rust_phase_contexts(CONTEXT, &tampered_thread),
1363 Err(RustTransportError::InvalidAssertionContext(_))
1364 ));
1365
1366 let mut invalid = read.clone();
1367 invalid.phases[0].parent_context_id = 99;
1368 assert!(matches!(
1369 validate_rust_phase_contexts(CONTEXT, &invalid),
1370 Err(RustTransportError::InvalidAssertionContext(_))
1371 ));
1372
1373 let mut unused = read.clone();
1374 let unused_child = rust_assertion_context_id(CONTEXT, inner_id, 12).unwrap();
1375 unused.phases.push(RustPhaseContext {
1376 process_id: 7,
1377 child_context_id: unused_child,
1378 parent_context_id: CONTEXT,
1379 invocation_nonce: 12,
1380 decision_id: inner_id.into(),
1381 });
1382 validate_rust_phase_contexts(CONTEXT, &unused).unwrap();
1383
1384 let mut cycle = read;
1385 cycle.phases[0].parent_context_id = inner;
1386 assert!(matches!(
1387 validate_rust_phase_contexts(CONTEXT, &cycle),
1388 Err(RustTransportError::InvalidAssertionContext(_))
1389 ));
1390 }
1391
1392 #[test]
1393 fn shared_transport_partitions_each_record_once_by_exact_test_root() {
1394 const SECOND: u64 = 84;
1395 let outer_id = "rs:decision:0123456789abcdef01234567";
1396 let inner_id = "rs:decision:fedcba9876543210fedcba98";
1397 let outer = rust_assertion_context_id(CONTEXT, outer_id, 10).unwrap();
1398 let inner = rust_assertion_context_id(outer, inner_id, 11).unwrap();
1399 let read = RustTransportRead {
1400 observations: vec![
1401 RustTransportObservation {
1402 process_id: 7,
1403 context_id: inner,
1404 observation: RustProbeObservation::Hit {
1405 id: "rs:statement:0123456789abcdef01234567".into(),
1406 },
1407 },
1408 RustTransportObservation {
1409 process_id: 7,
1410 context_id: SECOND,
1411 observation: RustProbeObservation::Hit {
1412 id: "rs:function:fedcba9876543210fedcba98".into(),
1413 },
1414 },
1415 RustTransportObservation {
1416 process_id: 7,
1417 context_id: 0,
1418 observation: RustProbeObservation::Hit {
1419 id: "rs:statement:aaaaaaaaaaaaaaaaaaaaaaaa".into(),
1420 },
1421 },
1422 ],
1423 ordinal_hits: vec![RustOrdinalHit {
1424 process_id: 7,
1425 context_id: CONTEXT,
1426 ordinal: 9,
1427 }],
1428 phases: vec![
1429 RustPhaseContext {
1430 process_id: 7,
1431 child_context_id: outer,
1432 parent_context_id: CONTEXT,
1433 invocation_nonce: 10,
1434 decision_id: outer_id.into(),
1435 },
1436 RustPhaseContext {
1437 process_id: 7,
1438 child_context_id: inner,
1439 parent_context_id: outer,
1440 invocation_nonce: 11,
1441 decision_id: inner_id.into(),
1442 },
1443 ],
1444 thread_phases: Vec::new(),
1445 thread_ends: Vec::new(),
1446 test_boundaries: Vec::new(),
1447 committed: 6,
1448 incomplete: 0,
1449 dropped: 0,
1450 attachments: 3,
1451 };
1452 let bases = BTreeSet::from([CONTEXT, SECOND]);
1453 let partition = partition_rust_transport_by_test_contexts(&read, &bases).unwrap();
1454 assert_eq!(partition.attributed[&CONTEXT].committed, 4);
1455 assert_eq!(partition.attributed[&SECOND].committed, 1);
1456 assert_eq!(partition.background.committed, 1);
1457 assert_eq!(partition.background.attachments, 3);
1458 assert!(partition.thread_scope_limitations.is_empty());
1459 assert_eq!(
1460 partition
1461 .attributed
1462 .values()
1463 .map(|transport| transport.committed)
1464 .sum::<u64>()
1465 + partition.background.committed,
1466 read.committed
1467 );
1468
1469 let mut foreign = read.clone();
1470 foreign.observations[1].context_id = 99;
1471 assert!(matches!(
1472 partition_rust_transport_by_test_contexts(&foreign, &bases),
1473 Err(RustTransportError::InvalidAttribution(_))
1474 ));
1475
1476 let mut incomplete = read.clone();
1477 incomplete.dropped = 1;
1478 assert!(matches!(
1479 partition_rust_transport_by_test_contexts(&incomplete, &bases),
1480 Err(RustTransportError::InvalidAttribution(_))
1481 ));
1482
1483 let mut cycle = read;
1484 cycle.phases[0].parent_context_id = inner;
1485 assert!(matches!(
1486 partition_rust_transport_by_test_contexts(&cycle, &bases),
1487 Err(RustTransportError::InvalidAttribution(_))
1488 ));
1489 }
1490
1491 #[test]
1492 fn thread_phases_are_join_bounded_by_the_test_boundary() {
1493 const SECOND: u64 = 84;
1494 let bases = BTreeSet::from([CONTEXT, SECOND]);
1495 let joined = rust_thread_context_id(CONTEXT, 0);
1496 let escaped = rust_thread_context_id(CONTEXT, 1);
1497 let hit = |context_id: u64| RustTransportObservation {
1498 process_id: 7,
1499 context_id,
1500 observation: RustProbeObservation::Hit {
1501 id: "rs:statement:0123456789abcdef01234567".into(),
1502 },
1503 };
1504 let read = RustTransportRead {
1505 observations: vec![hit(joined), hit(escaped), hit(CONTEXT)],
1506 thread_phases: vec![
1507 RustThreadPhase {
1508 process_id: 7,
1509 child_context_id: joined,
1510 parent_context_id: CONTEXT,
1511 invocation_nonce: 0,
1512 commit_index: 0,
1513 },
1514 RustThreadPhase {
1515 process_id: 7,
1516 child_context_id: escaped,
1517 parent_context_id: CONTEXT,
1518 invocation_nonce: 1,
1519 commit_index: 1,
1520 },
1521 ],
1522 thread_ends: vec![RustThreadEnd {
1523 process_id: 7,
1524 context_id: joined,
1525 commit_index: 5,
1526 }],
1527 test_boundaries: vec![
1528 RustTestBoundary {
1529 process_id: 7,
1530 context_id: CONTEXT,
1531 commit_index: 6,
1532 },
1533 RustTestBoundary {
1534 process_id: 7,
1535 context_id: SECOND,
1536 commit_index: 7,
1537 },
1538 ],
1539 committed: 8,
1540 attachments: 1,
1541 ..RustTransportRead::empty()
1542 };
1543 let partition = partition_rust_transport_by_test_contexts(&read, &bases).unwrap();
1544 assert_eq!(partition.attributed[&CONTEXT].committed, 5);
1547 assert_eq!(partition.attributed[&CONTEXT].thread_phases.len(), 1);
1548 assert_eq!(partition.attributed[&CONTEXT].thread_ends.len(), 1);
1549 assert_eq!(partition.attributed[&CONTEXT].test_boundaries.len(), 1);
1550 assert_eq!(partition.attributed[&SECOND].committed, 1);
1551 assert_eq!(partition.background.committed, 2);
1552 assert_eq!(partition.background.thread_phases.len(), 1);
1553 assert_eq!(
1554 partition.background.observations,
1555 vec![hit(escaped)],
1556 "the escaped thread's record must be background"
1557 );
1558 assert_eq!(
1559 partition.thread_scope_limitations,
1560 BTreeSet::from([format!(
1561 "RUST_THREAD_OUTLIVED_TEST: thread phase {escaped:016x} escaped test {CONTEXT:016x}"
1562 )])
1563 );
1564
1565 let mut late_end = read.clone();
1567 late_end.thread_ends.push(RustThreadEnd {
1568 process_id: 7,
1569 context_id: escaped,
1570 commit_index: 8,
1571 });
1572 late_end.committed = 9;
1573 let partition = partition_rust_transport_by_test_contexts(&late_end, &bases).unwrap();
1574 assert_eq!(partition.thread_scope_limitations.len(), 1);
1575 assert_eq!(partition.background.committed, 3);
1576
1577 let mut unbounded = read.clone();
1579 unbounded.test_boundaries.remove(0);
1580 unbounded.committed = 7;
1581 let partition = partition_rust_transport_by_test_contexts(&unbounded, &bases).unwrap();
1582 assert_eq!(partition.attributed[&CONTEXT].committed, 1);
1583 assert_eq!(partition.thread_scope_limitations.len(), 2);
1584
1585 let mut duplicate_end = read.clone();
1586 duplicate_end.thread_ends.push(RustThreadEnd {
1587 process_id: 7,
1588 context_id: joined,
1589 commit_index: 9,
1590 });
1591 duplicate_end.committed = 9;
1592 assert!(matches!(
1593 partition_rust_transport_by_test_contexts(&duplicate_end, &bases),
1594 Err(RustTransportError::InvalidAttribution(_))
1595 ));
1596
1597 let mut orphan_end = read.clone();
1598 orphan_end.thread_ends.push(RustThreadEnd {
1599 process_id: 7,
1600 context_id: 99,
1601 commit_index: 9,
1602 });
1603 orphan_end.committed = 9;
1604 assert!(matches!(
1605 partition_rust_transport_by_test_contexts(&orphan_end, &bases),
1606 Err(RustTransportError::InvalidAttribution(_))
1607 ));
1608
1609 let mut duplicate_boundary = read.clone();
1610 duplicate_boundary.test_boundaries.push(RustTestBoundary {
1611 process_id: 7,
1612 context_id: CONTEXT,
1613 commit_index: 9,
1614 });
1615 duplicate_boundary.committed = 9;
1616 assert!(matches!(
1617 partition_rust_transport_by_test_contexts(&duplicate_boundary, &bases),
1618 Err(RustTransportError::InvalidAttribution(_))
1619 ));
1620
1621 let mut unknown_boundary = read.clone();
1622 unknown_boundary.test_boundaries.push(RustTestBoundary {
1623 process_id: 7,
1624 context_id: 99,
1625 commit_index: 9,
1626 });
1627 unknown_boundary.committed = 9;
1628 assert!(matches!(
1629 partition_rust_transport_by_test_contexts(&unknown_boundary, &bases),
1630 Err(RustTransportError::InvalidAttribution(_))
1631 ));
1632
1633 let mut duplicate_definition = read;
1634 duplicate_definition.phases.push(RustPhaseContext {
1635 process_id: 7,
1636 child_context_id: joined,
1637 parent_context_id: CONTEXT,
1638 invocation_nonce: 0,
1639 decision_id: "rs:decision:0123456789abcdef01234567".into(),
1640 });
1641 duplicate_definition.committed = 9;
1642 assert!(matches!(
1643 partition_rust_transport_by_test_contexts(&duplicate_definition, &bases),
1644 Err(RustTransportError::InvalidAttribution(_))
1645 ));
1646 }
1647
1648 fn append_record(
1649 bytes: &mut [u8],
1650 kind: u8,
1651 outcome: u8,
1652 context_id: u64,
1653 id: &[u8],
1654 values: &[u8],
1655 ) {
1656 let descriptor_capacity = get_u32(bytes, 20).unwrap() as usize;
1657 let payload_base = HEADER_SIZE + descriptor_capacity * DESCRIPTOR_SIZE;
1658 let next = get_u64(bytes, NEXT_DESCRIPTOR_OFFSET).unwrap();
1659 let next_payload = get_u64(bytes, NEXT_PAYLOAD_OFFSET).unwrap();
1660 let descriptor = HEADER_SIZE + usize::try_from(next).unwrap() * DESCRIPTOR_SIZE;
1661 let payload_length = (id.len() + values.len()) as u32;
1662 bytes[descriptor + KIND_OFFSET] = kind;
1663 bytes[descriptor + OUTCOME_OFFSET] = outcome;
1664 bytes[descriptor + 3] = 0;
1665 put_u32(bytes, descriptor + PID_OFFSET, 7);
1666 bytes[descriptor + CONTEXT_OFFSET..descriptor + CONTEXT_OFFSET + 8]
1667 .copy_from_slice(&context_id.to_le_bytes());
1668 put_u32(
1669 bytes,
1670 descriptor + PAYLOAD_OFFSET_OFFSET,
1671 next_payload as u32,
1672 );
1673 put_u32(bytes, descriptor + PAYLOAD_LENGTH_OFFSET, payload_length);
1674 put_u32(bytes, descriptor + ID_LENGTH_OFFSET, id.len() as u32);
1675 put_u32(bytes, descriptor + VALUE_LENGTH_OFFSET, values.len() as u32);
1676 let record_checksum = checksum(
1677 kind,
1678 outcome,
1679 7,
1680 context_id,
1681 next_payload as u32,
1682 payload_length,
1683 id.len() as u32,
1684 values.len() as u32,
1685 id,
1686 values,
1687 );
1688 bytes[descriptor + CHECKSUM_OFFSET..descriptor + CHECKSUM_OFFSET + 8]
1689 .copy_from_slice(&record_checksum.to_le_bytes());
1690 let payload = payload_base + usize::try_from(next_payload).unwrap();
1691 bytes[payload..payload + id.len()].copy_from_slice(id);
1692 bytes[payload + id.len()..payload + id.len() + values.len()].copy_from_slice(values);
1693 bytes[descriptor + COMMIT_OFFSET] = 1;
1694 bytes[NEXT_DESCRIPTOR_OFFSET..NEXT_DESCRIPTOR_OFFSET + 8]
1695 .copy_from_slice(&(next + 1).to_le_bytes());
1696 bytes[NEXT_PAYLOAD_OFFSET..NEXT_PAYLOAD_OFFSET + 8]
1697 .copy_from_slice(&(next_payload + u64::from(payload_length)).to_le_bytes());
1698 }
1699
1700 #[test]
1701 fn thread_kind_records_are_authenticated_unique_and_strict() {
1702 let directory = temporary_directory("thread-kinds");
1703 let child = rust_thread_context_id(CONTEXT, 3);
1704 let mut definition = [0_u8; 16];
1705 definition[..8].copy_from_slice(&CONTEXT.to_le_bytes());
1706 definition[8..].copy_from_slice(&3_u64.to_le_bytes());
1707
1708 type SyntheticRecord<'bytes> = (u8, u8, u64, &'bytes [u8], &'bytes [u8]);
1709 let case = AtomicUsize::new(0);
1710 let build = |records: &[SyntheticRecord<'_>]| {
1711 let path = directory.join(format!(
1712 "case-{}.transport",
1713 case.fetch_add(1, Ordering::Relaxed)
1714 ));
1715 create_rust_transport(&path, TOKEN, 16, 4_096).unwrap();
1716 let mut bytes = fs::read(&path).unwrap();
1717 for (kind, outcome, context, id, values) in records {
1718 append_record(&mut bytes, *kind, *outcome, *context, id, values);
1719 }
1720 fs::write(&path, bytes).unwrap();
1721 path
1722 };
1723
1724 let valid = build(&[
1725 (KIND_THREAD_PHASE, 0, child, b"", &definition),
1726 (KIND_THREAD_END, 0, child, b"", b""),
1727 (KIND_TEST_BOUNDARY, 0, CONTEXT, b"", b""),
1728 ]);
1729 let read = read_rust_transport(&valid, &TOKEN).unwrap();
1730 assert_eq!(read.committed, 3);
1731 assert_eq!(
1732 read.thread_phases,
1733 vec![RustThreadPhase {
1734 process_id: 7,
1735 child_context_id: child,
1736 parent_context_id: CONTEXT,
1737 invocation_nonce: 3,
1738 commit_index: 0,
1739 }]
1740 );
1741 assert_eq!(
1742 read.thread_ends,
1743 vec![RustThreadEnd {
1744 process_id: 7,
1745 context_id: child,
1746 commit_index: 1,
1747 }]
1748 );
1749 assert_eq!(
1750 read.test_boundaries,
1751 vec![RustTestBoundary {
1752 process_id: 7,
1753 context_id: CONTEXT,
1754 commit_index: 2,
1755 }]
1756 );
1757
1758 let tampered = build(&[(KIND_THREAD_PHASE, 0, child ^ 1, b"", &definition)]);
1761 assert!(matches!(
1762 read_rust_transport(&tampered, &TOKEN),
1763 Err(RustTransportError::InvalidAssertionContext(_))
1764 ));
1765
1766 let conflicting_definition = {
1767 let mut other = [0_u8; 16];
1768 other[..8].copy_from_slice(&CONTEXT.to_le_bytes());
1769 other[8..].copy_from_slice(&4_u64.to_le_bytes());
1770 build(&[
1771 (KIND_THREAD_PHASE, 0, child, b"", &definition),
1772 (KIND_THREAD_PHASE, 0, child, b"", &other),
1773 ])
1774 };
1775 assert!(matches!(
1776 read_rust_transport(&conflicting_definition, &TOKEN),
1777 Err(RustTransportError::InvalidAssertionContext(_))
1778 ));
1779
1780 let duplicate_end = build(&[
1781 (KIND_THREAD_PHASE, 0, child, b"", &definition),
1782 (KIND_THREAD_END, 0, child, b"", b""),
1783 (KIND_THREAD_END, 0, child, b"", b""),
1784 ]);
1785 assert_eq!(
1786 read_rust_transport(&duplicate_end, &TOKEN),
1787 Err(RustTransportError::InvalidRecord(2))
1788 );
1789
1790 let duplicate_boundary = build(&[
1791 (KIND_TEST_BOUNDARY, 0, CONTEXT, b"", b""),
1792 (KIND_TEST_BOUNDARY, 0, CONTEXT, b"", b""),
1793 ]);
1794 assert_eq!(
1795 read_rust_transport(&duplicate_boundary, &TOKEN),
1796 Err(RustTransportError::InvalidRecord(1))
1797 );
1798
1799 for invalid in [
1800 build(&[(KIND_THREAD_END, 0, 0, b"", b"")]),
1802 build(&[(KIND_TEST_BOUNDARY, 0, u64::MAX, b"", b"")]),
1803 build(&[(KIND_THREAD_END, 0, CONTEXT, b"x", b"")]),
1805 build(&[(KIND_TEST_BOUNDARY, 0, CONTEXT, b"", b"y")]),
1806 build(&[(KIND_THREAD_PHASE, 0, child, b"", &definition[..8])]),
1807 build(&[(KIND_THREAD_END, 1, CONTEXT, b"", b"")]),
1809 ] {
1810 assert_eq!(
1811 read_rust_transport(&invalid, &TOKEN),
1812 Err(RustTransportError::InvalidRecord(0))
1813 );
1814 }
1815
1816 fs::remove_dir_all(directory).unwrap();
1817 }
1818
1819 #[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "windows")))]
1824 #[test]
1825 fn runtime_stub_compiles_and_records_nothing_elsewhere() {
1826 let directory = temporary_directory("stub");
1827 let binary = compile_fixture(&directory);
1828 let transport = directory.join("stub.transport");
1829 create_rust_transport(&transport, TOKEN, 1_024, 128 * 1024).unwrap();
1830 let output = Command::new(&binary)
1831 .arg("threads")
1832 .env(RUST_TRANSPORT_ENV, &transport)
1833 .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
1834 .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
1835 .output()
1836 .unwrap();
1837 assert!(output.status.success());
1838 let read = read_rust_transport(&transport, &TOKEN).unwrap();
1839 assert_eq!(read.committed, 0);
1840 assert_eq!(read.attachments, 0);
1841 fs::remove_dir_all(directory).unwrap();
1842 }
1843
1844 #[cfg(any(target_os = "macos", target_os = "linux", target_os = "windows"))]
1845 #[test]
1846 fn mmap_transport_is_concurrent_bounded_strict_and_kill_resilient() {
1847 let directory = temporary_directory("all");
1848 let binary = compile_fixture(&directory);
1849
1850 let concurrent = directory.join("concurrent.transport");
1851 create_rust_transport(&concurrent, TOKEN, 1_024, 128 * 1024).unwrap();
1852 let output = Command::new(&binary)
1853 .arg("threads")
1854 .env(RUST_TRANSPORT_ENV, &concurrent)
1855 .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
1856 .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
1857 .output()
1858 .unwrap();
1859 assert!(output.status.success());
1860 let read = read_rust_transport(&concurrent, &TOKEN).unwrap();
1861 assert_eq!(read.committed, 818);
1862 assert_eq!(read.incomplete, 0);
1863 assert_eq!(read.dropped, 0);
1864 assert_eq!(read.attachments, 1);
1865 assert_eq!(read.observations.len(), 801);
1866 assert!(
1867 matches!(read.observations.last(), Some(RustTransportObservation { context_id: CONTEXT, observation: RustProbeObservation::Decision { values, outcome: false, .. }, .. }) if values == &[Some(true), Some(false)])
1868 );
1869 assert_eq!(read.thread_phases.len(), 8);
1872 assert!(
1873 read.thread_phases
1874 .iter()
1875 .all(|phase| phase.parent_context_id == CONTEXT)
1876 );
1877 assert_eq!(read.thread_ends.len(), 8);
1878 let thread_contexts = read
1879 .thread_phases
1880 .iter()
1881 .map(|phase| phase.child_context_id)
1882 .collect::<BTreeSet<_>>();
1883 assert_eq!(thread_contexts.len(), 8);
1884 assert_eq!(
1885 read.thread_ends
1886 .iter()
1887 .map(|end| end.context_id)
1888 .collect::<BTreeSet<_>>(),
1889 thread_contexts
1890 );
1891 assert!(read.observations.iter().all(|item| {
1892 item.context_id == CONTEXT || thread_contexts.contains(&item.context_id)
1893 }));
1894 assert_eq!(
1895 read.observations
1896 .iter()
1897 .filter(|item| thread_contexts.contains(&item.context_id))
1898 .count(),
1899 800
1900 );
1901 validate_rust_phase_contexts(CONTEXT, &read).unwrap();
1902 assert_eq!(read.ordinal_hits.len(), 1);
1903 assert_eq!(read.ordinal_hits[0].ordinal, 7);
1904 assert_eq!(read.ordinal_hits[0].context_id, CONTEXT);
1905 assert_eq!(
1906 read_rust_transport(&concurrent, &[0x43; TOKEN_SIZE]),
1907 Err(RustTransportError::InvalidHeader),
1908 "a supervisor must never accept evidence from another task token"
1909 );
1910
1911 let mir_decisions = directory.join("mir-decisions.transport");
1912 create_rust_transport(&mir_decisions, TOKEN, 16, 4_096).unwrap();
1913 let output = Command::new(&binary)
1914 .arg("mir-decisions")
1915 .env(RUST_TRANSPORT_ENV, &mir_decisions)
1916 .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
1917 .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
1918 .output()
1919 .unwrap();
1920 assert!(output.status.success());
1921 let read = read_rust_transport(&mir_decisions, &TOKEN).unwrap();
1922 assert_eq!(
1923 read.committed, 7,
1924 "two migrated threads add one thread phase and one thread end each"
1925 );
1926 assert_eq!(read.thread_phases.len(), 2);
1927 assert_eq!(read.thread_ends.len(), 2);
1928 assert_eq!(read.dropped, 0);
1929 assert_eq!(read.incomplete, 0);
1930 assert!(read.observations.iter().any(|observation| matches!(
1931 observation,
1932 RustTransportObservation {
1933 context_id: 901,
1934 observation: RustProbeObservation::Decision { id, values, outcome: true },
1935 ..
1936 }
1937 if id == "rs:decision:0123456789abcdef01234567"
1938 && values == &[Some(true), Some(true)]
1939 )));
1940 assert!(
1941 read.ordinal_hits
1942 .iter()
1943 .any(|hit| hit.context_id == 903 && hit.ordinal == 777)
1944 );
1945 assert!(!read.ordinal_hits.iter().any(|hit| hit.ordinal == 888));
1946 assert!(read.observations.iter().any(|observation| matches!(
1947 observation,
1948 RustTransportObservation {
1949 context_id: 902,
1950 observation: RustProbeObservation::Decision { id, values, outcome: false },
1951 ..
1952 }
1953 if id == "rs:decision:fedcba9876543210fedcba98"
1954 && values == &[Some(false)]
1955 )));
1956
1957 let processes = directory.join("processes.transport");
1958 create_rust_transport(&processes, TOKEN, 64, 8 * 1024).unwrap();
1959 let mut children = (0..8)
1960 .map(|_| {
1961 Command::new(&binary)
1962 .env(RUST_TRANSPORT_ENV, &processes)
1963 .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
1964 .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
1965 .spawn()
1966 .unwrap()
1967 })
1968 .collect::<Vec<_>>();
1969 for child in &mut children {
1970 assert!(child.wait().unwrap().success());
1971 }
1972 let read = read_rust_transport(&processes, &TOKEN).unwrap();
1973 assert_eq!(read.attachments, 8);
1974 assert_eq!(read.committed, 24);
1975 assert_eq!(read.incomplete, 0);
1976 assert_eq!(read.dropped, 0);
1977 assert_eq!(
1978 read.observations
1979 .iter()
1980 .map(|item| item.process_id)
1981 .collect::<std::collections::BTreeSet<_>>()
1982 .len(),
1983 8
1984 );
1985
1986 let nested = directory.join("nested-context.transport");
1987 create_rust_transport(&nested, TOKEN, 16, 4_096).unwrap();
1988 let output = Command::new(&binary)
1989 .arg("contexts")
1990 .env(RUST_TRANSPORT_ENV, &nested)
1991 .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
1992 .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
1993 .output()
1994 .unwrap();
1995 assert!(output.status.success());
1996 let read = read_rust_transport(&nested, &TOKEN).unwrap();
1997 assert_eq!(
1998 read.observations
1999 .iter()
2000 .map(|item| item.context_id)
2001 .collect::<Vec<_>>(),
2002 [
2003 CONTEXT,
2004 100,
2005 200,
2006 100,
2007 CONTEXT,
2008 rust_assertion_context_id(CONTEXT, "rs:decision:0123456789abcdef01234567", 0)
2009 .unwrap(),
2010 CONTEXT,
2011 rust_assertion_context_id(CONTEXT, "rs:decision:0123456789abcdef01234567", 1)
2012 .unwrap(),
2013 CONTEXT,
2014 ]
2015 );
2016 let assertion =
2017 rust_assertion_context_id(CONTEXT, "rs:decision:0123456789abcdef01234567", 0).unwrap();
2018 assert_eq!(read.phases.len(), 2);
2019 assert_eq!(read.phases[0].child_context_id, assertion);
2020 assert_eq!(read.phases[0].parent_context_id, CONTEXT);
2021 assert_eq!(read.phases[0].invocation_nonce, 0);
2022 assert_eq!(
2023 read.phases[0].decision_id,
2024 "rs:decision:0123456789abcdef01234567"
2025 );
2026 assert_ne!(
2027 read.phases[0].child_context_id,
2028 read.phases[1].child_context_id
2029 );
2030 assert_eq!(read.phases[1].parent_context_id, CONTEXT);
2031 assert_eq!(read.phases[1].invocation_nonce, 1);
2032 assert_eq!(read.phases[1].decision_id, read.phases[0].decision_id);
2033 assert_eq!(read.committed, 11);
2034
2035 let invalid_phase = directory.join("invalid-phase.transport");
2036 fs::copy(&nested, &invalid_phase).unwrap();
2037 let mut bytes = fs::read(&invalid_phase).unwrap();
2038 let descriptor_capacity = get_u32(&bytes, 20).unwrap();
2039 let payload_base = HEADER_SIZE + descriptor_capacity as usize * DESCRIPTOR_SIZE;
2040 let descriptor = (0..read.committed as usize)
2041 .map(|index| HEADER_SIZE + index * DESCRIPTOR_SIZE)
2042 .find(|offset| bytes[*offset + KIND_OFFSET] == KIND_PHASE)
2043 .unwrap();
2044 let payload_offset = get_u32(&bytes, descriptor + PAYLOAD_OFFSET_OFFSET).unwrap();
2045 let payload_length = get_u32(&bytes, descriptor + PAYLOAD_LENGTH_OFFSET).unwrap();
2046 let id_length = get_u32(&bytes, descriptor + ID_LENGTH_OFFSET).unwrap();
2047 let value_length = get_u32(&bytes, descriptor + VALUE_LENGTH_OFFSET).unwrap();
2048 let payload = payload_base + payload_offset as usize;
2049 let id = bytes[payload..payload + id_length as usize].to_vec();
2050 let values =
2051 bytes[payload + id_length as usize..payload + payload_length as usize].to_vec();
2052 let invalid_child = 7_u64;
2053 bytes[descriptor + CONTEXT_OFFSET..descriptor + CONTEXT_OFFSET + 8]
2054 .copy_from_slice(&invalid_child.to_le_bytes());
2055 let replacement_checksum = checksum(
2056 KIND_PHASE,
2057 0,
2058 get_u32(&bytes, descriptor + PID_OFFSET).unwrap(),
2059 invalid_child,
2060 payload_offset,
2061 payload_length,
2062 id_length,
2063 value_length,
2064 &id,
2065 &values,
2066 );
2067 bytes[descriptor + CHECKSUM_OFFSET..descriptor + CHECKSUM_OFFSET + 8]
2068 .copy_from_slice(&replacement_checksum.to_le_bytes());
2069 fs::write(&invalid_phase, bytes).unwrap();
2070 assert!(matches!(
2071 read_rust_transport(&invalid_phase, &TOKEN),
2072 Err(RustTransportError::InvalidAssertionContext(_))
2073 ));
2074
2075 let rejected = directory.join("rejected-token.transport");
2076 create_rust_transport(&rejected, TOKEN, 16, 4_096).unwrap();
2077 let output = Command::new(&binary)
2078 .env(RUST_TRANSPORT_ENV, &rejected)
2079 .env(RUST_TRANSPORT_TOKEN_ENV, "00".repeat(TOKEN_SIZE))
2080 .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
2081 .output()
2082 .unwrap();
2083 assert!(output.status.success());
2084 let read = read_rust_transport(&rejected, &TOKEN).unwrap();
2085 assert_eq!(read.attachments, 0);
2086 assert_eq!(read.committed, 0);
2087 assert_eq!(read.dropped, 0);
2088
2089 let malformed_context = directory.join("malformed-context.transport");
2090 create_rust_transport(&malformed_context, TOKEN, 16, 4_096).unwrap();
2091 let output = Command::new(&binary)
2092 .env(RUST_TRANSPORT_ENV, &malformed_context)
2093 .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
2094 .env(RUST_CONTEXT_ENV, "not-a-context")
2095 .output()
2096 .unwrap();
2097 assert!(output.status.success());
2098 let read = read_rust_transport(&malformed_context, &TOKEN).unwrap();
2099 assert_eq!(read.attachments, 0);
2100 assert_eq!(read.committed, 0);
2101
2102 let bounded = directory.join("bounded.transport");
2103 create_rust_transport(&bounded, TOKEN, 2, 256).unwrap();
2104 let output = Command::new(&binary)
2105 .env(RUST_TRANSPORT_ENV, &bounded)
2106 .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
2107 .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
2108 .output()
2109 .unwrap();
2110 assert!(output.status.success());
2111 let read = read_rust_transport(&bounded, &TOKEN).unwrap();
2112 assert_eq!(read.committed, 2);
2113 assert_eq!(read.dropped, 1);
2114
2115 let payload_bounded = directory.join("payload-bounded.transport");
2116 create_rust_transport(&payload_bounded, TOKEN, 8, 16).unwrap();
2117 let output = Command::new(&binary)
2118 .env(RUST_TRANSPORT_ENV, &payload_bounded)
2119 .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
2120 .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
2121 .output()
2122 .unwrap();
2123 assert!(output.status.success());
2124 let read = read_rust_transport(&payload_bounded, &TOKEN).unwrap();
2125 assert_eq!(read.committed, 0);
2126 assert_eq!(read.incomplete, 3);
2127 assert_eq!(read.dropped, 3);
2128
2129 let killed = directory.join("killed.transport");
2130 create_rust_transport(&killed, TOKEN, 16, 4_096).unwrap();
2131 let mut child = Command::new(&binary)
2132 .arg("kill")
2133 .env(RUST_TRANSPORT_ENV, &killed)
2134 .env(RUST_TRANSPORT_TOKEN_ENV, token_hex())
2135 .env(RUST_CONTEXT_ENV, format!("{CONTEXT:016x}"))
2136 .stdout(Stdio::piped())
2137 .spawn()
2138 .unwrap();
2139 let mut ready = String::new();
2140 BufReader::new(child.stdout.take().unwrap())
2141 .read_line(&mut ready)
2142 .unwrap();
2143 assert_eq!(ready, "ready\n");
2144 child.kill().unwrap();
2145 child.wait().unwrap();
2146 let read = read_rust_transport(&killed, &TOKEN).unwrap();
2147 assert_eq!(read.committed, 1);
2148 assert_eq!(read.incomplete, 2);
2149 assert_eq!(read.dropped, 0);
2150 assert_eq!(
2151 read.observations,
2152 [RustTransportObservation {
2153 process_id: read.observations[0].process_id,
2154 context_id: CONTEXT,
2155 observation: RustProbeObservation::Hit {
2156 id: "rs:function:fedcba9876543210fedcba98".into()
2157 }
2158 }]
2159 );
2160
2161 let corrupt = directory.join("corrupt.transport");
2162 create_rust_transport(&corrupt, TOKEN, 2, 256).unwrap();
2163 let mut bytes = fs::read(&corrupt).unwrap();
2164 bytes[0] ^= 1;
2165 fs::write(&corrupt, bytes).unwrap();
2166 assert_eq!(
2167 read_rust_transport(&corrupt, &TOKEN),
2168 Err(RustTransportError::InvalidHeader)
2169 );
2170
2171 let invalid_commit = directory.join("invalid-commit.transport");
2172 fs::copy(&bounded, &invalid_commit).unwrap();
2173 let mut bytes = fs::read(&invalid_commit).unwrap();
2174 bytes[HEADER_SIZE + COMMIT_OFFSET] = 2;
2175 fs::write(&invalid_commit, bytes).unwrap();
2176 assert_eq!(
2177 read_rust_transport(&invalid_commit, &TOKEN),
2178 Err(RustTransportError::InvalidDescriptor(0))
2179 );
2180
2181 let invalid_flags = directory.join("invalid-flags.transport");
2182 fs::copy(&bounded, &invalid_flags).unwrap();
2183 let mut bytes = fs::read(&invalid_flags).unwrap();
2184 bytes[HEADER_SIZE + 3] = 1;
2185 fs::write(&invalid_flags, bytes).unwrap();
2186 assert_eq!(
2187 read_rust_transport(&invalid_flags, &TOKEN),
2188 Err(RustTransportError::InvalidDescriptor(0))
2189 );
2190
2191 let invalid_checksum = directory.join("invalid-checksum.transport");
2192 fs::copy(&bounded, &invalid_checksum).unwrap();
2193 let mut bytes = fs::read(&invalid_checksum).unwrap();
2194 bytes[HEADER_SIZE + CHECKSUM_OFFSET] ^= 1;
2195 fs::write(&invalid_checksum, bytes).unwrap();
2196 assert_eq!(
2197 read_rust_transport(&invalid_checksum, &TOKEN),
2198 Err(RustTransportError::InvalidRecord(0))
2199 );
2200
2201 #[cfg(unix)]
2202 {
2203 use std::os::unix::fs::symlink;
2204
2205 let linked = directory.join("linked.transport");
2206 symlink(&bounded, &linked).unwrap();
2207 assert_eq!(
2208 read_rust_transport(&linked, &TOKEN),
2209 Err(RustTransportError::UnsafeFile(linked.display().to_string()))
2210 );
2211 }
2212
2213 let truncated = directory.join("truncated.transport");
2214 create_rust_transport(&truncated, TOKEN, 2, 256).unwrap();
2215 OpenOptions::new()
2216 .write(true)
2217 .open(&truncated)
2218 .unwrap()
2219 .set_len(63)
2220 .unwrap();
2221 assert_eq!(
2222 read_rust_transport(&truncated, &TOKEN),
2223 Err(RustTransportError::InvalidHeader)
2224 );
2225
2226 let incomplete = directory.join("incomplete.transport");
2227 create_rust_transport(&incomplete, TOKEN, 2, 256).unwrap();
2228 let file = OpenOptions::new()
2229 .read(true)
2230 .write(true)
2231 .open(&incomplete)
2232 .unwrap();
2233 let mapping = map_mut(&file).unwrap();
2234 atomic_u64(&mapping, NEXT_DESCRIPTOR_OFFSET)
2235 .unwrap()
2236 .store(1, Ordering::Release);
2237 mapping.flush().unwrap();
2238 drop(mapping);
2239 let read = read_rust_transport(&incomplete, &TOKEN).unwrap();
2240 assert_eq!(read.committed, 0);
2241 assert_eq!(read.incomplete, 1);
2242 assert!(read.observations.is_empty());
2243 assert!(read.ordinal_hits.is_empty());
2244
2245 fs::remove_dir_all(directory).unwrap();
2246 }
2247}