wowlab_analytics/analytics/merge/
mod.rs

1//! Incremental merge of `ChunkTelemetry` into `RunningAggregateStateV1`.
2
3mod emit;
4mod histogram;
5mod keys;
6mod sections;
7
8#[cfg(test)]
9mod tests;
10
11use emit::sort_state;
12use histogram::merge_histogram;
13use sections::{
14    merge_actions, merge_auras, merge_cooldowns, merge_damage_profile, merge_dictionary,
15    merge_dps_buckets, merge_execution, merge_instrumentation_coverage, merge_representative,
16    merge_resources,
17};
18use wowlab_types::{proto, stats};
19
20#[rustfmt::skip]
21pub use emit::{emit_result_snapshot, emit_timeline_snapshot};
22
23/// Error produced while merging a chunk into the running state.
24#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
25#[error("{kind}")]
26pub struct MergeError {
27    kind: MergeErrorKind,
28}
29
30#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
31enum MergeErrorKind {
32    #[error("chunk has zero iterations")]
33    ZeroIterations,
34    #[error("chunk index {index} out of range (total {total})")]
35    ChunkIndexOutOfRange { index: u32, total: u32 },
36    #[error("chunk bucket_ms does not match running state")]
37    BucketParamMismatch,
38    #[error("dictionary label conflict for target_id {target_id}")]
39    DictionaryLabelConflict { target_id: u32 },
40}
41
42impl MergeError {
43    const fn zero_iterations() -> Self {
44        Self {
45            kind: MergeErrorKind::ZeroIterations,
46        }
47    }
48
49    const fn chunk_index_out_of_range(index: u32, total: u32) -> Self {
50        Self {
51            kind: MergeErrorKind::ChunkIndexOutOfRange { index, total },
52        }
53    }
54
55    const fn bucket_param_mismatch() -> Self {
56        Self {
57            kind: MergeErrorKind::BucketParamMismatch,
58        }
59    }
60
61    const fn dictionary_label_conflict(target_id: u32) -> Self {
62        Self {
63            kind: MergeErrorKind::DictionaryLabelConflict { target_id },
64        }
65    }
66
67    /// Returns whether the chunk reported zero iterations.
68    #[must_use]
69    pub const fn is_zero_iterations(&self) -> bool {
70        matches!(self.kind, MergeErrorKind::ZeroIterations)
71    }
72
73    /// Returns the invalid chunk index and configured chunk total, when applicable.
74    #[must_use]
75    pub const fn chunk_index_range(&self) -> Option<(u32, u32)> {
76        match self.kind {
77            MergeErrorKind::ChunkIndexOutOfRange { index, total } => Some((index, total)),
78            _ => None,
79        }
80    }
81
82    /// Returns whether the chunk and running state use different bucket sizes.
83    #[must_use]
84    pub const fn is_bucket_param_mismatch(&self) -> bool {
85        matches!(self.kind, MergeErrorKind::BucketParamMismatch)
86    }
87
88    /// Returns the conflicting target identifier, when applicable.
89    #[must_use]
90    pub const fn conflicting_target_id(&self) -> Option<u32> {
91        match self.kind {
92            MergeErrorKind::DictionaryLabelConflict { target_id } => Some(target_id),
93            _ => None,
94        }
95    }
96}
97
98/// Build a zeroed `RunningAggregateStateV1` sized for `chunks_total` chunks.
99#[must_use]
100pub fn new_running_state(chunks_total: u32) -> proto::RunningAggregateStateV1 {
101    proto::RunningAggregateStateV1 {
102        chunks_total,
103        chunks_completed: 0,
104        iterations_total: 0,
105        total_fight_time_ms: 0,
106        weighted_mean_num_x10: 0,
107        m2_total_bits: 0_f64.to_bits(),
108        min_dps_x10: u32::MAX,
109        max_dps_x10: 0,
110        histogram: None,
111        actions: vec![],
112        auras: vec![],
113        resources: vec![],
114        cooldowns: vec![],
115        execution: None,
116        damage_profile: None,
117        dps_bucket_sums_x10: vec![],
118        dps_bucket_samples: vec![],
119        bucket_ms: 0,
120        representative: None,
121        representative_meta: None,
122        dictionary: None,
123        instrumentation_coverage: None,
124    }
125}
126
127/// Validate a chunk against the running state before merging it in.
128///
129/// # Errors
130///
131/// Returns an error for zero-iteration chunks, out-of-range indices, or incompatible buckets.
132pub fn validate_chunk(
133    chunk: &proto::ChunkTelemetry,
134    state: &proto::RunningAggregateStateV1,
135) -> Result<(), MergeError> {
136    if chunk.iterations == 0 {
137        return Err(MergeError::zero_iterations());
138    }
139
140    if chunk.chunk_index >= state.chunks_total {
141        return Err(MergeError::chunk_index_out_of_range(
142            chunk.chunk_index,
143            state.chunks_total,
144        ));
145    }
146
147    if state.chunks_completed > 0
148        && state.bucket_ms != 0
149        && chunk.bucket_ms != 0
150        && state.bucket_ms != chunk.bucket_ms
151    {
152        return Err(MergeError::bucket_param_mismatch());
153    }
154
155    Ok(())
156}
157
158/// Incrementally merge a `ChunkTelemetry` into `RunningAggregateStateV1`.
159///
160/// # Errors
161///
162/// Returns an error when validation fails or dictionary entries conflict.
163pub fn merge_chunk(
164    state: &mut proto::RunningAggregateStateV1,
165    chunk: &proto::ChunkTelemetry,
166) -> Result<(), MergeError> {
167    validate_chunk(chunk, state)?;
168
169    stats::merge_dps_stats(
170        &mut state.iterations_total,
171        &mut state.weighted_mean_num_x10,
172        &mut state.m2_total_bits,
173        &mut state.min_dps_x10,
174        &mut state.max_dps_x10,
175        chunk.iterations,
176        chunk.mean_dps_x10,
177        chunk.m2_dps_bits,
178        chunk.min_dps_x10,
179        chunk.max_dps_x10,
180    );
181
182    state.chunks_completed += 1;
183    state.total_fight_time_ms += chunk.total_fight_time_ms;
184
185    merge_histogram(state, chunk);
186    merge_actions(state, chunk);
187    merge_auras(state, chunk);
188    merge_resources(state, chunk);
189    merge_cooldowns(state, chunk);
190    merge_execution(state, chunk);
191    merge_damage_profile(state, chunk);
192    merge_dps_buckets(state, chunk);
193    merge_representative(state, chunk);
194    merge_dictionary(state, chunk)?;
195    merge_instrumentation_coverage(state, chunk);
196
197    sort_state(state);
198
199    Ok(())
200}
wowlab_analytics/analytics/merge/mod.rs

wowlab_analytics/analytics/merge/
mod.rs
