graphforge_core/
hash_observation.rs

1//! SHA-256 with byte and elapsed-time observation during explicit captures.
2use sha2::digest::{FixedOutput, HashMarker, Output, OutputSizeUser, Reset, Update};
3use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
4static CAPTURES: AtomicUsize = AtomicUsize::new(0);
5
6/// Activates process-wide SHA-256 observation without resetting other captures.
7#[derive(Debug)]
8pub struct HashObservation(());
9impl HashObservation {
10    /// Begin optional observation on all process threads.
11    #[must_use]
12    pub fn start() -> Self {
13        CAPTURES.fetch_add(1, Ordering::Relaxed);
14        Self(())
15    }
16}
17impl Drop for HashObservation {
18    fn drop(&mut self) {
19        CAPTURES.fetch_sub(1, Ordering::Relaxed);
20    }
21}
22fn active() -> bool {
23    CAPTURES.load(Ordering::Relaxed) != 0
24}
25use std::time::Instant;
26
27static BYTES: AtomicU64 = AtomicU64::new(0);
28static NANOS: AtomicU64 = AtomicU64::new(0);
29
30/// SHA-256, preserving digest bytes while observing work on every process thread.
31#[derive(Clone)]
32pub struct ObservedSha256(
33    sha2::Sha256,
34    #[cfg(any(test, feature = "test-support"))] HashDomain,
35);
36
37/// Why cryptographic input bytes are consumed. Payload identity is never control metadata.
38#[derive(Clone, Copy, Debug, PartialEq, Eq)]
39pub enum HashDomain {
40    /// Immutable durable artifact bytes, including CAS identity and full verification.
41    ArtifactPayload,
42    /// Canonical identity required by a public contract.
43    ContractIdentity,
44    /// Authenticated control objects and receipts.
45    ControlAuthentication,
46    /// Portable archive, member and payload authentication at the transport boundary.
47    PortableAuthentication,
48    /// Optional diagnostic evidence.
49    OptionalEvidence,
50    /// A producer that has not supplied a domain.
51    Unclassified,
52}
53
54impl Default for ObservedSha256 {
55    fn default() -> Self {
56        Self::for_domain(HashDomain::Unclassified)
57    }
58}
59
60impl ObservedSha256 {
61    /// Construct a SHA-256 producer with its actual input domain.
62    #[must_use]
63    pub fn for_domain(domain: HashDomain) -> Self {
64        #[cfg(not(any(test, feature = "test-support")))]
65        let _ = domain;
66        Self(
67            sha2::Sha256::default(),
68            #[cfg(any(test, feature = "test-support"))]
69            domain,
70        )
71    }
72}
73
74/// Domain-specific cryptographic producers retain the standard `Digest` API.
75#[derive(Clone)]
76pub struct DomainSha256<const DOMAIN: u8>(ObservedSha256);
77/// Cryptographic durable artifact identity and full verification.
78pub type ArtifactSha256 = DomainSha256<0>;
79/// Canonical contract identities.
80pub type ContractSha256 = DomainSha256<1>;
81/// Control object and receipt authentication.
82pub type ControlSha256 = DomainSha256<2>;
83/// Optional evidence identity.
84pub type EvidenceSha256 = DomainSha256<3>;
85/// Portable archive, member and payload authentication.
86pub type PortableSha256 = DomainSha256<4>;
87impl<const DOMAIN: u8> Default for DomainSha256<DOMAIN> {
88    fn default() -> Self {
89        let domain = match DOMAIN {
90            0 => HashDomain::ArtifactPayload,
91            1 => HashDomain::ContractIdentity,
92            2 => HashDomain::ControlAuthentication,
93            3 => HashDomain::OptionalEvidence,
94            4 => HashDomain::PortableAuthentication,
95            _ => HashDomain::Unclassified,
96        };
97        Self(ObservedSha256::for_domain(domain))
98    }
99}
100impl<const DOMAIN: u8> OutputSizeUser for DomainSha256<DOMAIN> {
101    type OutputSize = <sha2::Sha256 as OutputSizeUser>::OutputSize;
102}
103impl<const DOMAIN: u8> HashMarker for DomainSha256<DOMAIN> {}
104impl<const DOMAIN: u8> Update for DomainSha256<DOMAIN> {
105    fn update(&mut self, data: &[u8]) {
106        Update::update(&mut self.0, data);
107    }
108}
109impl<const DOMAIN: u8> FixedOutput for DomainSha256<DOMAIN> {
110    fn finalize_into(self, out: &mut Output<Self>) {
111        FixedOutput::finalize_into(self.0, out);
112    }
113}
114impl<const DOMAIN: u8> Reset for DomainSha256<DOMAIN> {
115    fn reset(&mut self) {
116        Reset::reset(&mut self.0);
117    }
118}
119impl OutputSizeUser for ObservedSha256 {
120    type OutputSize = <sha2::Sha256 as OutputSizeUser>::OutputSize;
121}
122impl HashMarker for ObservedSha256 {}
123impl Update for ObservedSha256 {
124    fn update(&mut self, data: &[u8]) {
125        #[cfg(any(test, feature = "test-support"))]
126        operation::record_hash(self.1, data.len() as u64);
127        if !active() {
128            Update::update(&mut self.0, data);
129            return;
130        }
131        let started = Instant::now();
132        Update::update(&mut self.0, data);
133        NANOS.fetch_add(
134            u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX),
135            Ordering::Relaxed,
136        );
137        BYTES.fetch_add(data.len() as u64, Ordering::Relaxed);
138    }
139}
140impl FixedOutput for ObservedSha256 {
141    fn finalize_into(self, out: &mut Output<Self>) {
142        if !active() {
143            FixedOutput::finalize_into(self.0, out);
144            return;
145        }
146        let started = Instant::now();
147        FixedOutput::finalize_into(self.0, out);
148        NANOS.fetch_add(
149            u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX),
150            Ordering::Relaxed,
151        );
152    }
153}
154impl Reset for ObservedSha256 {
155    fn reset(&mut self) {
156        Reset::reset(&mut self.0);
157    }
158}
159/// Cumulative input bytes and update/finalize elapsed nanoseconds.
160#[must_use]
161pub fn totals() -> (u64, u64) {
162    (BYTES.load(Ordering::Relaxed), NANOS.load(Ordering::Relaxed))
163}
164
165/// Transfer the current operation to an owned worker. In production this is empty.
166#[derive(Clone, Default)]
167pub struct OperationContext(#[cfg(any(test, feature = "test-support"))] operation::Context);
168impl OperationContext {
169    #[must_use]
170    /// Capture the initiating operation without resetting counters.
171    pub fn capture() -> Self {
172        Self(
173            #[cfg(any(test, feature = "test-support"))]
174            operation::Context::capture(),
175        )
176    }
177    #[must_use]
178    /// Attach this operation until the returned guard drops.
179    pub fn attach(&self) -> OperationGuard {
180        OperationGuard {
181            #[cfg(any(test, feature = "test-support"))]
182            _guard: self.0.attach(),
183        }
184    }
185}
186/// Restores the initiating thread's previous operation after worker completion.
187pub struct OperationGuard {
188    #[cfg(any(test, feature = "test-support"))]
189    _guard: operation::Guard,
190}
191
192/// Account an actual admitted topology projection only when test support is compiled in.
193#[inline]
194pub fn record_topology_projection() {
195    #[cfg(any(test, feature = "test-support"))]
196    operation::record_topology_projection();
197}
198
199/// Account an actual composite request fingerprint producer in test support only.
200#[inline]
201pub fn record_composite_request_fingerprint() {
202    #[cfg(any(test, feature = "test-support"))]
203    operation::record_composite_request_fingerprint();
204}
205
206/// Isolated operation accounting used by Rust facade tests. Captures neither reset nor
207/// sample process-wide counters; worker handles attach only to their own operation.
208#[cfg(any(test, feature = "test-support"))]
209pub mod operation {
210    use super::HashDomain;
211    use std::cell::RefCell;
212    use std::sync::Arc;
213    use std::sync::atomic::{AtomicU64, Ordering};
214
215    #[derive(Default)]
216    struct Counters {
217        hash: [AtomicU64; 6],
218        checksum: AtomicU64,
219        topology_projections: AtomicU64,
220        composite_request_fingerprints: AtomicU64,
221    }
222    thread_local! {
223        static CURRENT: RefCell<Option<Arc<Counters>>> = const { RefCell::new(None) };
224    }
225    /// Exact application input bytes submitted to actual producers in one operation.
226    #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
227    pub struct Snapshot {
228        /// Durable artifact input bytes consumed by actual cryptographic producers.
229        pub artifact_payload_sha256_bytes: u64,
230        /// Canonical public-contract identity preimage bytes.
231        pub contract_identity_sha256_bytes: u64,
232        /// Required control object and receipt authentication input bytes.
233        pub control_authentication_sha256_bytes: u64,
234        /// Portable transport authentication input bytes, including archive members and payloads.
235        pub portable_authentication_sha256_bytes: u64,
236        /// Optional evidence identity input bytes.
237        pub optional_evidence_sha256_bytes: u64,
238        /// Unclassified cryptographic input bytes; admission tests require zero.
239        pub unclassified_sha256_bytes: u64,
240        /// Exact bytes submitted to actual corruption-checksum producers.
241        pub checksum_bytes: u64,
242        /// Actual admitted label topology projections started by this operation.
243        pub topology_projections: u64,
244        /// Successful whole composite request fingerprints, excluding participant subfingerprints.
245        pub composite_request_fingerprints: u64,
246    }
247    /// Captured operation context, transferable to synchronous worker threads.
248    #[derive(Clone, Default)]
249    pub struct Context(Option<Arc<Counters>>);
250    impl Context {
251        #[must_use]
252        /// Capture the initiating operation without resetting counters.
253        pub fn capture() -> Self {
254            CURRENT.with(|current| Self(current.borrow().clone()))
255        }
256        #[must_use]
257        /// Attach this operation until the returned guard drops.
258        pub fn attach(&self) -> Guard {
259            let previous = CURRENT.with(|current| current.replace(self.0.clone()));
260            Guard {
261                previous,
262                _thread: std::marker::PhantomData,
263            }
264        }
265    }
266    /// Restores the previous operation on drop, including nested and panicking workers.
267    pub struct Guard {
268        previous: Option<Arc<Counters>>,
269        _thread: std::marker::PhantomData<std::rc::Rc<()>>,
270    }
271    impl Drop for Guard {
272        fn drop(&mut self) {
273            CURRENT.with(|current| {
274                current.replace(self.previous.take());
275            });
276        }
277    }
278    /// Owns one operation's counters and attaches the initiating thread.
279    pub struct Capture {
280        counters: Arc<Counters>,
281        _guard: Guard,
282    }
283    impl Capture {
284        #[must_use]
285        /// Begin an isolated capture on the initiating thread.
286        pub fn start() -> Self {
287            let counters = Arc::new(Counters::default());
288            let context = Context(Some(Arc::clone(&counters)));
289            Self {
290                counters,
291                _guard: context.attach(),
292            }
293        }
294        #[must_use]
295        /// Transfer this capture to an owned worker.
296        pub fn context(&self) -> Context {
297            Context(Some(Arc::clone(&self.counters)))
298        }
299        #[must_use]
300        /// Read cumulative actual producer work for this operation.
301        pub fn snapshot(&self) -> Snapshot {
302            let load = |index: usize| self.counters.hash[index].load(Ordering::Relaxed);
303            Snapshot {
304                artifact_payload_sha256_bytes: load(0),
305                contract_identity_sha256_bytes: load(1),
306                control_authentication_sha256_bytes: load(2),
307                optional_evidence_sha256_bytes: load(3),
308                portable_authentication_sha256_bytes: load(4),
309                unclassified_sha256_bytes: load(5),
310                checksum_bytes: self.counters.checksum.load(Ordering::Relaxed),
311                topology_projections: self.counters.topology_projections.load(Ordering::Relaxed),
312                composite_request_fingerprints: self
313                    .counters
314                    .composite_request_fingerprints
315                    .load(Ordering::Relaxed),
316            }
317        }
318    }
319    pub(super) fn record_hash(domain: HashDomain, bytes: u64) {
320        let index = match domain {
321            HashDomain::ArtifactPayload => 0,
322            HashDomain::ContractIdentity => 1,
323            HashDomain::ControlAuthentication => 2,
324            HashDomain::OptionalEvidence => 3,
325            HashDomain::PortableAuthentication => 4,
326            HashDomain::Unclassified => 5,
327        };
328        CURRENT.with(|current| {
329            if let Some(counters) = current.borrow().as_ref() {
330                counters.hash[index].fetch_add(bytes, Ordering::Relaxed);
331            }
332        });
333    }
334    /// Called by the actual streaming corruption-checksum producer.
335    pub fn record_checksum(bytes: u64) {
336        CURRENT.with(|current| {
337            if let Some(counters) = current.borrow().as_ref() {
338                counters.checksum.fetch_add(bytes, Ordering::Relaxed);
339            }
340        });
341    }
342    pub(super) fn record_topology_projection() {
343        CURRENT.with(|current| {
344            if let Some(counters) = current.borrow().as_ref() {
345                counters
346                    .topology_projections
347                    .fetch_add(1, Ordering::Relaxed);
348            }
349        });
350    }
351
352    pub(super) fn record_composite_request_fingerprint() {
353        CURRENT.with(|current| {
354            if let Some(counters) = current.borrow().as_ref() {
355                counters
356                    .composite_request_fingerprints
357                    .fetch_add(1, Ordering::Relaxed);
358            }
359        });
360    }
361}
362
363#[cfg(test)]
364mod tests {
365    use super::{
366        HashDomain, ObservedSha256, OperationContext, PortableSha256,
367        operation::{Capture, Snapshot},
368    };
369    use crate::canonical::{CanonicalDomain, fingerprint};
370    use sha2::Digest;
371
372    #[test]
373    fn portable_authentication_keeps_all_operation_domains_disjoint() {
374        let capture = Capture::start();
375        for (domain, bytes) in [
376            (HashDomain::ArtifactPayload, 2),
377            (HashDomain::ContractIdentity, 3),
378            (HashDomain::ControlAuthentication, 5),
379            (HashDomain::OptionalEvidence, 7),
380            (HashDomain::PortableAuthentication, 11),
381            (HashDomain::Unclassified, 13),
382        ] {
383            let mut hash = ObservedSha256::for_domain(domain);
384            hash.update(vec![0_u8; bytes]);
385            hash.finalize();
386        }
387        assert_eq!(
388            capture.snapshot(),
389            Snapshot {
390                artifact_payload_sha256_bytes: 2,
391                contract_identity_sha256_bytes: 3,
392                control_authentication_sha256_bytes: 5,
393                portable_authentication_sha256_bytes: 11,
394                optional_evidence_sha256_bytes: 7,
395                unclassified_sha256_bytes: 13,
396                checksum_bytes: 0,
397                topology_projections: 0,
398                composite_request_fingerprints: 0,
399            }
400        );
401    }
402
403    #[test]
404    fn portable_sha256_preserves_digest_and_counts_clone_and_reset_inputs() {
405        let capture = Capture::start();
406        let mut hash = PortableSha256::new();
407        hash.update(b"abc");
408        let mut cloned = hash.clone();
409        cloned.update(b"!");
410        assert_eq!(cloned.finalize(), sha2::Sha256::digest(b"abc!"));
411        sha2::digest::Reset::reset(&mut hash);
412        hash.update(b"portable");
413        assert_eq!(hash.finalize(), sha2::Sha256::digest(b"portable"));
414        assert_eq!(
415            capture.snapshot(),
416            Snapshot {
417                portable_authentication_sha256_bytes: 12,
418                ..Snapshot::default()
419            }
420        );
421    }
422
423    #[test]
424    fn portable_captures_isolate_workers_and_restore_nested_and_detached_scopes() {
425        let outer = Capture::start();
426        let context = OperationContext::capture();
427        let independent = std::thread::spawn(|| {
428            let capture = Capture::start();
429            PortableSha256::digest([0_u8; 29]);
430            let context = OperationContext::capture();
431            std::thread::spawn(move || {
432                let _attached = context.attach();
433                PortableSha256::digest([0_u8; 31]);
434            })
435            .join()
436            .unwrap();
437            capture.snapshot()
438        });
439        std::thread::spawn(move || {
440            {
441                let _attached = context.attach();
442                PortableSha256::digest([0_u8; 17]);
443                {
444                    let nested = Capture::start();
445                    PortableSha256::digest([0_u8; 11]);
446                    assert_eq!(
447                        nested.snapshot(),
448                        Snapshot {
449                            portable_authentication_sha256_bytes: 11,
450                            ..Snapshot::default()
451                        }
452                    );
453                }
454                PortableSha256::digest([0_u8; 13]);
455            }
456            PortableSha256::digest([0_u8; 7]);
457        })
458        .join()
459        .unwrap();
460        assert_eq!(
461            outer.snapshot(),
462            Snapshot {
463                portable_authentication_sha256_bytes: 30,
464                ..Snapshot::default()
465            }
466        );
467        assert_eq!(
468            independent.join().unwrap(),
469            Snapshot {
470                portable_authentication_sha256_bytes: 60,
471                ..Snapshot::default()
472            }
473        );
474    }
475
476    #[test]
477    fn operation_captures_isolate_parallel_workers_and_restore_nested_scopes() {
478        let outer = Capture::start();
479        let first_context = outer.context();
480        let second = std::thread::spawn(|| {
481            let capture = Capture::start();
482            let context = capture.context();
483            std::thread::spawn(move || {
484                let _attached = context.attach();
485                let mut hash = ObservedSha256::for_domain(HashDomain::ArtifactPayload);
486                hash.update([0_u8; 29]);
487                hash.finalize();
488                fingerprint(CanonicalDomain::CompositeRequest, 1, b"second worker").unwrap();
489                super::record_composite_request_fingerprint();
490            })
491            .join()
492            .unwrap();
493            capture.snapshot()
494        });
495        std::thread::spawn(move || {
496            let _attached = first_context.attach();
497            let mut hash = ObservedSha256::for_domain(HashDomain::ArtifactPayload);
498            hash.update([0_u8; 17]);
499            hash.finalize();
500            fingerprint(CanonicalDomain::CompositeRequest, 1, b"first worker").unwrap();
501            super::record_composite_request_fingerprint();
502            {
503                let nested = Capture::start();
504                let mut hash = ObservedSha256::for_domain(HashDomain::ControlAuthentication);
505                hash.update([0_u8; 11]);
506                hash.finalize();
507                fingerprint(CanonicalDomain::CompositeRequest, 1, b"nested").unwrap();
508                assert_eq!(nested.snapshot().composite_request_fingerprints, 0);
509                super::record_composite_request_fingerprint();
510                fingerprint(
511                    CanonicalDomain::CompositeGraphMutationContent,
512                    1,
513                    b"another domain",
514                )
515                .unwrap();
516                assert_eq!(nested.snapshot().control_authentication_sha256_bytes, 11);
517                assert_eq!(nested.snapshot().composite_request_fingerprints, 1);
518            }
519            super::operation::record_checksum(7);
520        })
521        .join()
522        .unwrap();
523        assert_eq!(outer.snapshot().artifact_payload_sha256_bytes, 17);
524        assert_eq!(outer.snapshot().control_authentication_sha256_bytes, 0);
525        assert_eq!(outer.snapshot().checksum_bytes, 7);
526        assert_eq!(outer.snapshot().composite_request_fingerprints, 1);
527        let second = second.join().unwrap();
528        assert_eq!(second.artifact_payload_sha256_bytes, 29);
529        assert_eq!(second.composite_request_fingerprints, 1);
530    }
531}
graphforge_core/hash_observation.rs

graphforge_core/
hash_observation.rs
