spatialrust_platform/
streaming.rs

1//! SpatialRust 1.2 bounded-streaming fail-closed release gate.
2
3use std::fmt::Write;
4
5use crate::{
6    BudgetKind, ConformanceReport, ConformanceStatus, LtsPolicy, PerformanceBudget,
7    PerformanceBudgetReport, ReleaseGate, ReleaseGateDecision, SecurityChecklist,
8    StabilityRegistry,
9};
10
11const REQUIRED_CASES: &[&str] = &[
12    "streaming-linux",
13    "streaming-windows",
14    "streaming-macos",
15    "streaming-memory-fail-closed",
16    "streaming-cancellation-cleanup",
17    "streaming-format-roundtrip",
18    "streaming-copc-range",
19    "streaming-deterministic-voxel",
20    "streaming-rust-cli",
21    "streaming-python-iterator",
22    "streaming-unsafe-audit",
23];
24
25const REQUIRED_RECEIPTS: &[&str] = &[
26    "epic121-streaming-contract",
27    "epic122-bounded-record-stream",
28    "epic123-streaming-io",
29    "epic124-chunk-ops",
30    "epic125-streaming-e2e",
31];
32
33const REQUIRED_EXAMPLES: &[&str] = &[
34    "streaming_receipt",
35    "bounded_record_stream",
36    "bounded_pcd_to_ply",
37    "bounded_voxel",
38    "spatialrust-stream",
39    "streaming_1_2_release_gate",
40];
41
42const MAX_MEMORY_BUDGET_BYTES: u64 = 256 * 1024 * 1024;
43const MAX_SPOOL_LIMIT_BYTES: u64 = 2 * 1024 * 1024 * 1024;
44const MAX_OPEN_SPILL_FILES: u64 = 1025;
45
46/// Typed resource and transfer measurements consumed by the 1.2 gate.
47#[derive(Clone, Copy, Debug, PartialEq, Eq)]
48pub struct Streaming12Measurements {
49    /// Configured hard tracked-memory ceiling.
50    pub memory_budget_bytes: u64,
51    /// Peak explicitly tracked resident bytes.
52    pub peak_tracked_bytes: u64,
53    /// Configured maximum temporary spool extent.
54    pub spool_limit_bytes: u64,
55    /// Temporary bytes written by the canonical external operation.
56    pub spilled_bytes: u64,
57    /// Tracked bytes still live after completion or cancellation cleanup.
58    pub current_bytes_after_finish: u64,
59    /// Host-copy bytes not represented by leased chunks or named IO.
60    pub hidden_host_copy_bytes: u64,
61    /// Explicit host-to-device bytes in the CPU-only 1.2 workflow.
62    pub host_to_device_bytes: u64,
63    /// Explicit device-to-host bytes in the CPU-only 1.2 workflow.
64    pub device_to_host_bytes: u64,
65    /// Cross-chunk/run-size deterministic output mismatches.
66    pub determinism_mismatches: u64,
67    /// Maximum simultaneously open spool/run files.
68    pub max_open_spill_files: u64,
69}
70
71/// Evidence gathered by CI and release tooling for SpatialRust 1.2.
72#[derive(Clone, Debug)]
73pub struct Streaming12ReleaseEvidence {
74    /// Required platform, correctness, audit, and workflow cases.
75    pub conformance: ConformanceReport,
76    /// Satisfied security audit evidence.
77    pub security: SecurityChecklist,
78    /// Typed memory, spill, cleanup, copy, and determinism measurements.
79    pub measurements: Streaming12Measurements,
80    /// Dated Epic receipt identifiers.
81    pub passed_receipts: Vec<String>,
82    /// Cargo/Python/CLI examples exercised by the candidate.
83    pub verified_examples: Vec<String>,
84    /// Migration policy identifier; must equal `bounded-streaming-1.2`.
85    pub migration_policy: String,
86}
87
88/// Mandatory SpatialRust 1.2 bounded-streaming release policy.
89#[derive(Clone, Copy, Debug, Default)]
90pub struct Streaming12ReleaseGate;
91
92impl Streaming12ReleaseGate {
93    /// Returns conformance ids that must be present exactly once with `Pass`.
94    pub const fn required_conformance_cases() -> &'static [&'static str] {
95        REQUIRED_CASES
96    }
97
98    /// Returns implementation receipt ids required by the candidate.
99    pub const fn required_receipts() -> &'static [&'static str] {
100        REQUIRED_RECEIPTS
101    }
102
103    /// Returns runnable workflow ids required by the candidate.
104    pub const fn required_examples() -> &'static [&'static str] {
105        REQUIRED_EXAMPLES
106    }
107
108    /// Evaluates every mandatory item and returns all denial reasons.
109    pub fn evaluate(evidence: &Streaming12ReleaseEvidence) -> ReleaseGateDecision {
110        let base = ReleaseGate {
111            stability: Some(StabilityRegistry::bounded_streaming_v1_2_surface()),
112            conformance: Some(evidence.conformance.clone()),
113            security: Some(evidence.security.clone()),
114            lts: Some(LtsPolicy::spatialrust_v1()),
115            budgets: Some(streaming_budgets(evidence.measurements)),
116            reject_experimental: true,
117        };
118        let mut decision = base.evaluate();
119        require_passing_cases(&mut decision.reasons, &evidence.conformance);
120        require_names(
121            &mut decision.reasons,
122            "receipt",
123            REQUIRED_RECEIPTS,
124            &evidence.passed_receipts,
125        );
126        require_names(
127            &mut decision.reasons,
128            "example",
129            REQUIRED_EXAMPLES,
130            &evidence.verified_examples,
131        );
132        if evidence.migration_policy != "bounded-streaming-1.2" {
133            decision
134                .reasons
135                .push("migration policy `bounded-streaming-1.2` was not acknowledged".into());
136        }
137        let values = evidence.measurements;
138        if values.peak_tracked_bytes > values.memory_budget_bytes {
139            decision.reasons.push(format!(
140                "streaming peak {} exceeds configured memory budget {}",
141                values.peak_tracked_bytes, values.memory_budget_bytes
142            ));
143        }
144        if values.spilled_bytes > values.spool_limit_bytes {
145            decision.reasons.push(format!(
146                "streaming spill {} exceeds configured spool limit {}",
147                values.spilled_bytes, values.spool_limit_bytes
148            ));
149        }
150        decision.allowed = decision.reasons.is_empty();
151        decision
152    }
153
154    /// Generates the auditable Markdown release receipt.
155    #[must_use]
156    pub fn render_markdown(evidence: &Streaming12ReleaseEvidence) -> String {
157        let decision = Self::evaluate(evidence);
158        let mut output = String::from("# SpatialRust 1.2 bounded-streaming release receipt\n\n");
159        let _ = writeln!(
160            output,
161            "Decision: **{}**\n",
162            if decision.allowed { "allowed" } else { "denied" }
163        );
164        output.push_str("| Measurement | Observed | Ceiling |\n");
165        output.push_str("| --- | ---: | ---: |\n");
166        for (label, observed, ceiling) in measurement_rows(evidence.measurements) {
167            let _ = writeln!(output, "| {label} | {observed} | {ceiling} |");
168        }
169        output.push_str("\nRequired receipts:\n\n");
170        for receipt in REQUIRED_RECEIPTS {
171            let present = evidence.passed_receipts.iter().any(|value| value == receipt);
172            let _ = writeln!(output, "- [{}] `{receipt}`", if present { "x" } else { " " });
173        }
174        if !decision.reasons.is_empty() {
175            output.push_str("\nDenial reasons:\n\n");
176            for reason in decision.reasons {
177                let _ = writeln!(output, "- {reason}");
178            }
179        }
180        output
181    }
182}
183
184fn require_passing_cases(reasons: &mut Vec<String>, conformance: &ConformanceReport) {
185    for required in REQUIRED_CASES {
186        let matching =
187            conformance.cases().iter().filter(|case| case.id == *required).collect::<Vec<_>>();
188        match matching.as_slice() {
189            [case] if case.status == ConformanceStatus::Pass => {}
190            [case] => {
191                reasons.push(format!("required conformance `{required}` is {:?}", case.status));
192            }
193            [] => reasons.push(format!("required conformance `{required}` is missing")),
194            _ => reasons.push(format!("required conformance `{required}` is duplicated")),
195        }
196    }
197}
198
199fn require_names(reasons: &mut Vec<String>, kind: &str, required: &[&str], actual: &[String]) {
200    for name in required {
201        let count = actual.iter().filter(|value| value.as_str() == *name).count();
202        match count {
203            1 => {}
204            0 => reasons.push(format!("required {kind} `{name}` is missing")),
205            _ => reasons.push(format!("required {kind} `{name}` is duplicated")),
206        }
207    }
208}
209
210fn measurement_rows(values: Streaming12Measurements) -> [(&'static str, u64, u64); 10] {
211    [
212        ("configured memory budget (bytes)", values.memory_budget_bytes, MAX_MEMORY_BUDGET_BYTES),
213        ("peak tracked memory (bytes)", values.peak_tracked_bytes, MAX_MEMORY_BUDGET_BYTES),
214        ("configured spool limit (bytes)", values.spool_limit_bytes, MAX_SPOOL_LIMIT_BYTES),
215        ("spilled bytes", values.spilled_bytes, MAX_SPOOL_LIMIT_BYTES),
216        ("live bytes after finish", values.current_bytes_after_finish, 0),
217        ("hidden host-copy bytes", values.hidden_host_copy_bytes, 0),
218        ("host-to-device bytes", values.host_to_device_bytes, 0),
219        ("device-to-host bytes", values.device_to_host_bytes, 0),
220        ("determinism mismatches", values.determinism_mismatches, 0),
221        ("open spill files", values.max_open_spill_files, MAX_OPEN_SPILL_FILES),
222    ]
223}
224
225fn streaming_budgets(values: Streaming12Measurements) -> PerformanceBudgetReport {
226    let kinds = [
227        BudgetKind::MemoryBytes,
228        BudgetKind::MemoryBytes,
229        BudgetKind::MemoryBytes,
230        BudgetKind::MemoryBytes,
231        BudgetKind::MemoryBytes,
232        BudgetKind::BytesCopied,
233        BudgetKind::BytesCopied,
234        BudgetKind::BytesCopied,
235        BudgetKind::AllocationCount,
236        BudgetKind::AllocationCount,
237    ];
238    let ids = [
239        "streaming-configured-memory-budget-bytes",
240        "streaming-peak-tracked-memory-bytes",
241        "streaming-configured-spool-limit-bytes",
242        "streaming-spilled-bytes",
243        "streaming-live-bytes-after-finish",
244        "streaming-hidden-host-copy-bytes",
245        "streaming-host-to-device-bytes",
246        "streaming-device-to-host-bytes",
247        "streaming-determinism-mismatches",
248        "streaming-open-spill-files",
249    ];
250    let mut report = PerformanceBudgetReport::new();
251    for (((_, observed, ceiling), kind), id) in
252        measurement_rows(values).into_iter().zip(kinds).zip(ids)
253    {
254        report.declare(PerformanceBudget { id: id.into(), kind, ceiling });
255        report.sample(id, observed);
256    }
257    report
258}
259
260#[cfg(test)]
261mod tests {
262    use super::{Streaming12Measurements, Streaming12ReleaseEvidence, Streaming12ReleaseGate};
263    use crate::{ConformanceReport, ConformanceStatus, SecurityChecklist};
264
265    fn passing() -> Streaming12ReleaseEvidence {
266        let mut conformance = ConformanceReport::new();
267        for &id in Streaming12ReleaseGate::required_conformance_cases() {
268            conformance.record(id, ConformanceStatus::Pass, Some("CI receipt".into()));
269        }
270        Streaming12ReleaseEvidence {
271            conformance,
272            security: SecurityChecklist::north_star_baseline_satisfied(),
273            measurements: Streaming12Measurements {
274                memory_budget_bytes: 1024 * 1024,
275                peak_tracked_bytes: 64 * 1024,
276                spool_limit_bytes: 1024 * 1024,
277                spilled_bytes: 4096,
278                current_bytes_after_finish: 0,
279                hidden_host_copy_bytes: 0,
280                host_to_device_bytes: 0,
281                device_to_host_bytes: 0,
282                determinism_mismatches: 0,
283                max_open_spill_files: 4,
284            },
285            passed_receipts: Streaming12ReleaseGate::required_receipts()
286                .iter()
287                .map(ToString::to_string)
288                .collect(),
289            verified_examples: Streaming12ReleaseGate::required_examples()
290                .iter()
291                .map(ToString::to_string)
292                .collect(),
293            migration_policy: "bounded-streaming-1.2".into(),
294        }
295    }
296
297    #[test]
298    fn streaming_complete_evidence_is_allowed_and_rendered() {
299        let evidence = passing();
300        assert!(Streaming12ReleaseGate::evaluate(&evidence).allowed);
301        let markdown = Streaming12ReleaseGate::render_markdown(&evidence);
302        assert!(markdown.contains("Decision: **allowed**"));
303        assert!(markdown.contains("epic125-streaming-e2e"));
304    }
305
306    #[test]
307    fn streaming_rejects_missing_skipped_and_duplicate_evidence() {
308        let mut evidence = passing();
309        let mut conformance = ConformanceReport::new();
310        for &id in Streaming12ReleaseGate::required_conformance_cases() {
311            if id == "streaming-macos" {
312                conformance.record(id, ConformanceStatus::Skip, None);
313            } else if id == "streaming-rust-cli" {
314                conformance.record(id, ConformanceStatus::Pass, None);
315                conformance.record(id, ConformanceStatus::Pass, None);
316            } else if id != "streaming-python-iterator" {
317                conformance.record(id, ConformanceStatus::Pass, None);
318            }
319        }
320        evidence.conformance = conformance;
321        evidence.passed_receipts.pop();
322        evidence.verified_examples.push("streaming_1_2_release_gate".into());
323        evidence.migration_policy = "vision-2".into();
324        let decision = Streaming12ReleaseGate::evaluate(&evidence);
325        assert!(!decision.allowed);
326        for needle in [
327            "streaming-python-iterator",
328            "streaming-macos",
329            "streaming-rust-cli",
330            "epic125-streaming-e2e",
331            "streaming_1_2_release_gate",
332            "migration policy",
333        ] {
334            assert!(decision.reasons.iter().any(|reason| reason.contains(needle)), "{needle}");
335        }
336    }
337
338    #[test]
339    fn streaming_rejects_each_resource_budget_overrun() {
340        let overruns: &[(&str, fn(&mut Streaming12Measurements))] = &[
341            ("configured-memory", |v| v.memory_budget_bytes = 256 * 1024 * 1024 + 1),
342            ("peak-tracked", |v| v.peak_tracked_bytes = 256 * 1024 * 1024 + 1),
343            ("configured-spool", |v| v.spool_limit_bytes = 2 * 1024 * 1024 * 1024 + 1),
344            ("spilled", |v| v.spilled_bytes = 2 * 1024 * 1024 * 1024 + 1),
345            ("live-bytes", |v| v.current_bytes_after_finish = 1),
346            ("hidden-host-copy", |v| v.hidden_host_copy_bytes = 1),
347            ("host-to-device", |v| v.host_to_device_bytes = 1),
348            ("device-to-host", |v| v.device_to_host_bytes = 1),
349            ("determinism", |v| v.determinism_mismatches = 1),
350            ("open-spill-files", |v| v.max_open_spill_files = 1026),
351        ];
352        for &(budget_id, mutate) in overruns {
353            let mut evidence = passing();
354            mutate(&mut evidence.measurements);
355            let decision = Streaming12ReleaseGate::evaluate(&evidence);
356            assert!(!decision.allowed, "{budget_id}");
357            assert!(
358                decision.reasons.iter().any(|reason| reason.contains(budget_id)),
359                "{budget_id}: {:?}",
360                decision.reasons
361            );
362        }
363    }
364
365    #[test]
366    fn streaming_rejects_peak_and_spill_above_configured_limits() {
367        let mut evidence = passing();
368        evidence.measurements.peak_tracked_bytes = evidence.measurements.memory_budget_bytes + 1;
369        evidence.measurements.spilled_bytes = evidence.measurements.spool_limit_bytes + 1;
370        let decision = Streaming12ReleaseGate::evaluate(&evidence);
371        assert!(!decision.allowed);
372        assert!(decision.reasons.iter().any(|reason| reason.contains("configured memory budget")));
373        assert!(decision.reasons.iter().any(|reason| reason.contains("configured spool limit")));
374    }
375}
spatialrust_platform/streaming.rs

spatialrust_platform/
streaming.rs
