ferrum_interfaces/vnext/operation/
compiled_identity.rs

1use serde::Serialize;
2use std::sync::Arc;
3
4use super::super::{
5    DeviceId, ExecutionIdentityEnvelope, ExecutionIdentityParts, ExecutionLaneId, NodeId,
6    NodeInvocationId, OperationId, PlanHash, PlanId, ProviderId, RequestIdentity, ResourcePoolId,
7    RunId, SpanId, StepParticipantFrameAssignment, TransactionId, EXECUTION_IDENTITY_VERSION,
8};
9use super::ProviderExecutionSemantics;
10
11#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
12pub(super) struct CompiledSubmissionWaveNodeIdentityTemplate {
13    node_index: u32,
14    node_id: NodeId,
15    operation_id: OperationId,
16    provider_id: ProviderId,
17    provider_implementation_fingerprint: String,
18    provider_execution_semantics: ProviderExecutionSemantics,
19}
20
21impl CompiledSubmissionWaveNodeIdentityTemplate {
22    pub(super) fn new(
23        node_index: u32,
24        node_id: NodeId,
25        operation_id: OperationId,
26        provider_id: ProviderId,
27        provider_implementation_fingerprint: String,
28        provider_execution_semantics: ProviderExecutionSemantics,
29    ) -> Self {
30        Self {
31            node_index,
32            node_id,
33            operation_id,
34            provider_id,
35            provider_implementation_fingerprint,
36            provider_execution_semantics,
37        }
38    }
39
40    pub(super) const fn node_index(&self) -> u32 {
41        self.node_index
42    }
43
44    pub(super) fn node_id(&self) -> &NodeId {
45        &self.node_id
46    }
47
48    pub(super) fn operation_id(&self) -> &OperationId {
49        &self.operation_id
50    }
51
52    pub(super) fn provider_id(&self) -> &ProviderId {
53        &self.provider_id
54    }
55
56    pub(super) fn provider_implementation_fingerprint(&self) -> &str {
57        &self.provider_implementation_fingerprint
58    }
59
60    pub(super) const fn provider_execution_semantics(&self) -> ProviderExecutionSemantics {
61        self.provider_execution_semantics
62    }
63}
64
65#[derive(Debug, PartialEq, Eq, Serialize)]
66struct CompiledSubmissionWaveIdentityData {
67    plan_id: PlanId,
68    plan_hash: PlanHash,
69    device_id: DeviceId,
70    runtime_implementation_fingerprint: String,
71    lane_id: ExecutionLaneId,
72    nodes: Vec<CompiledSubmissionWaveNodeIdentityTemplate>,
73    fingerprint: String,
74}
75
76/// Cold-path identity topology for one immutable plan on one execution lane.
77///
78/// The topology owns only plan-stable node/provider facts. A physical wave
79/// binds participant/frame seeds to it and materializes full operation
80/// identities only for nodes that are encoded, observed, or failed.
81#[derive(Debug, Clone, PartialEq, Eq)]
82pub struct CompiledSubmissionWaveIdentity {
83    data: Arc<CompiledSubmissionWaveIdentityData>,
84}
85
86impl CompiledSubmissionWaveIdentity {
87    #[allow(clippy::too_many_arguments)]
88    pub(super) fn from_validated(
89        plan_id: PlanId,
90        plan_hash: PlanHash,
91        device_id: DeviceId,
92        runtime_implementation_fingerprint: String,
93        lane_id: ExecutionLaneId,
94        nodes: Vec<CompiledSubmissionWaveNodeIdentityTemplate>,
95        fingerprint: String,
96    ) -> Self {
97        Self {
98            data: Arc::new(CompiledSubmissionWaveIdentityData {
99                plan_id,
100                plan_hash,
101                device_id,
102                runtime_implementation_fingerprint,
103                lane_id,
104                nodes,
105                fingerprint,
106            }),
107        }
108    }
109
110    pub fn plan_id(&self) -> &PlanId {
111        &self.data.plan_id
112    }
113
114    pub fn plan_hash(&self) -> &PlanHash {
115        &self.data.plan_hash
116    }
117
118    pub fn device_id(&self) -> &DeviceId {
119        &self.data.device_id
120    }
121
122    pub fn runtime_implementation_fingerprint(&self) -> &str {
123        &self.data.runtime_implementation_fingerprint
124    }
125
126    pub fn lane_id(&self) -> ExecutionLaneId {
127        self.data.lane_id
128    }
129
130    pub fn node_count(&self) -> usize {
131        self.data.nodes.len()
132    }
133
134    pub fn fingerprint(&self) -> &str {
135        &self.data.fingerprint
136    }
137
138    pub(super) fn node_id_at(&self, node_index: usize) -> Option<&NodeId> {
139        self.data.nodes.get(node_index).map(|node| &node.node_id)
140    }
141
142    pub(super) fn node_at(
143        &self,
144        node_index: usize,
145    ) -> Option<&CompiledSubmissionWaveNodeIdentityTemplate> {
146        self.data.nodes.get(node_index)
147    }
148
149    pub(super) fn operation_id_at(&self, node_index: usize) -> Option<&OperationId> {
150        self.data
151            .nodes
152            .get(node_index)
153            .map(|node| &node.operation_id)
154    }
155
156    pub(super) fn provider_id_at(&self, node_index: usize) -> Option<&ProviderId> {
157        self.data
158            .nodes
159            .get(node_index)
160            .map(|node| &node.provider_id)
161    }
162
163    pub(super) fn node_index(&self, node_id: &NodeId) -> Option<usize> {
164        self.data
165            .nodes
166            .iter()
167            .position(|node| &node.node_id == node_id)
168    }
169}
170
171#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
172pub(super) struct SubmissionWaveParticipantIdentitySeed {
173    frame: StepParticipantFrameAssignment,
174    run_id: RunId,
175    request_id: RequestIdentity,
176    resource_pool_id: Option<ResourcePoolId>,
177    resource_pool_identity_fingerprint: Option<String>,
178    provisioning_run_id: Option<RunId>,
179    provisioning_request_id: Option<RequestIdentity>,
180    transaction_id: Option<TransactionId>,
181    active_sequence_slot: u32,
182    admission_generation: u64,
183    activation_epoch: u64,
184    runtime_implementation_fingerprint: String,
185    active_sequence_fingerprint: String,
186    span_root: String,
187}
188
189impl SubmissionWaveParticipantIdentitySeed {
190    #[allow(clippy::too_many_arguments)]
191    pub(super) fn new(
192        frame: StepParticipantFrameAssignment,
193        run_id: RunId,
194        request_id: RequestIdentity,
195        resource_pool_id: Option<ResourcePoolId>,
196        resource_pool_identity_fingerprint: Option<String>,
197        provisioning_run_id: Option<RunId>,
198        provisioning_request_id: Option<RequestIdentity>,
199        transaction_id: Option<TransactionId>,
200        active_sequence_slot: u32,
201        admission_generation: u64,
202        activation_epoch: u64,
203        runtime_implementation_fingerprint: String,
204        active_sequence_fingerprint: String,
205        span_root: String,
206    ) -> Self {
207        Self {
208            frame,
209            run_id,
210            request_id,
211            resource_pool_id,
212            resource_pool_identity_fingerprint,
213            provisioning_run_id,
214            provisioning_request_id,
215            transaction_id,
216            active_sequence_slot,
217            admission_generation,
218            activation_epoch,
219            runtime_implementation_fingerprint,
220            active_sequence_fingerprint,
221            span_root,
222        }
223    }
224
225    pub(super) const fn frame(&self) -> StepParticipantFrameAssignment {
226        self.frame
227    }
228
229    pub(super) fn runtime_implementation_fingerprint(&self) -> &str {
230        &self.runtime_implementation_fingerprint
231    }
232
233    pub(super) fn operation_identity(
234        &self,
235        topology: &CompiledSubmissionWaveIdentity,
236        node_index: usize,
237    ) -> Option<ExecutionIdentityEnvelope> {
238        let node = topology.data.nodes.get(node_index)?;
239        let topology = topology.data.as_ref();
240        let node_count = u64::try_from(topology.nodes.len())
241            .expect("compiled submission-wave node count fits u64");
242        let node_index = u64::from(node.node_index);
243        let completed_frames = self.frame.frame_id().get() - 1;
244        let node_invocation = completed_frames
245            .checked_mul(node_count)
246            .and_then(|value| value.checked_add(node_index))
247            .and_then(|value| value.checked_add(1))
248            .expect("compiled submission-wave invocation range was validated");
249        let node_invocation_id = NodeInvocationId::try_from(node_invocation)
250            .expect("compiled submission-wave invocation id is non-zero");
251        let events_per_frame = node_count
252            .checked_mul(3)
253            .and_then(|value| value.checked_add(2))
254            .expect("compiled submission-wave event range was validated");
255        let sequence = completed_frames
256            .checked_mul(events_per_frame)
257            .and_then(|value| value.checked_add(node_index.checked_mul(3)?))
258            .and_then(|value| value.checked_add(5))
259            .expect("compiled submission-wave event sequence was validated");
260        let node_span = SpanId::new(format!(
261            "{}/frame/{}/node/{node_invocation}",
262            self.span_root,
263            self.frame.frame_id()
264        ))
265        .expect("compiled submission-wave node span is portable");
266        let operation_span = SpanId::new(format!("{node_span}/operation"))
267            .expect("compiled submission-wave operation span is portable");
268
269        Some(
270            ExecutionIdentityEnvelope::new(ExecutionIdentityParts {
271                version: EXECUTION_IDENTITY_VERSION,
272                run_id: self.run_id.clone(),
273                request_id: self.request_id.clone(),
274                sequence,
275                plan_id: Some(topology.plan_id.clone()),
276                plan_hash: Some(topology.plan_hash.clone()),
277                frame_id: Some(self.frame.frame_id()),
278                node_invocation_id: Some(node_invocation_id),
279                node_id: Some(node.node_id.clone()),
280                operation_id: Some(node.operation_id.clone()),
281                provider_id: Some(node.provider_id.clone()),
282                device_id: Some(topology.device_id.clone()),
283                resource_pool_id: self.resource_pool_id.clone(),
284                resource_pool_identity_fingerprint: self.resource_pool_identity_fingerprint.clone(),
285                provisioning_run_id: self.provisioning_run_id.clone(),
286                provisioning_request_id: self.provisioning_request_id.clone(),
287                transaction_id: self.transaction_id.clone(),
288                active_sequence_slot: Some(self.active_sequence_slot),
289                admission_generation: Some(self.admission_generation),
290                activation_epoch: Some(self.activation_epoch),
291                runtime_implementation_fingerprint: Some(
292                    self.runtime_implementation_fingerprint.clone(),
293                ),
294                active_sequence_fingerprint: Some(self.active_sequence_fingerprint.clone()),
295                completed_sequence_fingerprint: None,
296                aborted_sequence_fingerprint: None,
297                resource_id: None,
298                resource_generation: None,
299                resource_batch_fingerprint: None,
300                span_id: operation_span,
301                parent_span_id: Some(node_span),
302                async_links: Vec::new(),
303            })
304            .expect("compiled submission-wave participant seed is a valid operation identity"),
305        )
306    }
307}
ferrum_interfaces/vnext/operation/compiled_identity.rs

ferrum_interfaces/vnext/operation/
compiled_identity.rs
