wowlab_sentinel/scheduler/
runtime_completion_log.rs

1//! Bounded idempotency log for the in-process runtime.
2
3use std::collections::VecDeque;
4
5use sha2::{Digest, Sha256};
6use wowlab_common::{ClaimToken, RuntimeChunkId, RuntimeWorkResult, WorkContextHash};
7use wowlab_types::sim::FastMap;
8
9const RECENT_COMPLETIONS_CAP: usize = 4_096;
10pub(crate) const HASH_LEN: usize = 32;
11const COMPLETION_LOG_TOKEN_COPIES: usize = 2;
12
13#[derive(Debug, Default)]
14pub(crate) struct CompletionLog {
15    hashes: FastMap<ClaimToken, [u8; HASH_LEN]>,
16    order: VecDeque<ClaimToken>,
17}
18
19impl CompletionLog {
20    pub(crate) fn new() -> Self {
21        Self::default()
22    }
23
24    pub(crate) fn record(&mut self, token: ClaimToken, hash: [u8; HASH_LEN]) {
25        if self.hashes.insert(token.clone(), hash).is_none() {
26            self.order.push_back(token);
27
28            if self.order.len() > RECENT_COMPLETIONS_CAP {
29                if let Some(evicted) = self.order.pop_front() {
30                    self.hashes.remove(&evicted);
31                }
32            }
33        }
34    }
35
36    pub(crate) fn get(&self, token: &ClaimToken) -> Option<&[u8; HASH_LEN]> {
37        self.hashes.get(token)
38    }
39
40    pub(crate) fn memory_bytes(&self) -> usize {
41        let entry = size_of::<ClaimToken>() + size_of::<[u8; HASH_LEN]>();
42        let token_str = self
43            .hashes
44            .keys()
45            .map(|t| t.as_str().len() * COMPLETION_LOG_TOKEN_COPIES)
46            .sum::<usize>();
47
48        self.hashes.capacity() * entry + self.order.capacity() * size_of::<ClaimToken>() + token_str
49    }
50}
51
52pub(crate) fn completion_hash(
53    chunk_id: RuntimeChunkId,
54    work_context_hash: &WorkContextHash,
55    node_public_key: &str,
56    results: &[RuntimeWorkResult],
57) -> [u8; HASH_LEN] {
58    let mut hasher = Sha256::new();
59
60    hasher.update(chunk_id.get().to_le_bytes());
61    hasher.update(work_context_hash.as_bytes());
62    hasher.update(node_public_key.len().to_le_bytes());
63    hasher.update(node_public_key.as_bytes());
64    hasher.update(results.len().to_le_bytes());
65
66    for r in results {
67        hasher.update(r.item_id.to_le_bytes());
68        hasher.update(r.tag.to_le_bytes());
69        hasher.update(r.iterations.to_le_bytes());
70        hasher.update(r.mean_dps_x10.to_le_bytes());
71        hasher.update(r.m2_dps_bits.to_le_bytes());
72
73        if let Some(telemetry) = &r.telemetry_pb {
74            hasher.update(telemetry.len().to_le_bytes());
75            hasher.update(telemetry);
76        } else {
77            hasher.update(0usize.to_le_bytes());
78        }
79    }
80
81    let mut out = [0u8; HASH_LEN];
82
83    out.copy_from_slice(&hasher.finalize());
84
85    out
86}
wowlab_sentinel/scheduler/runtime_completion_log.rs

wowlab_sentinel/scheduler/
runtime_completion_log.rs
