ferrum_interfaces/vnext/resource/
work.rs

1use super::{
2    invalid_resource, AdmissionFitPolicy, AdmissionPressureAction, Digest, DynamicResourceShape,
3    ExecutionFrameId, NodeId, RequestAuthorityId, ResourceWorkShape, SequenceAuthorityId,
4    Serialize, Sha256, TokenSpanWork, VNextError,
5};
6use crate::vnext::ReusableExecutionBucketId;
7use std::{ops::Range, sync::Arc};
8
9#[derive(Debug, Clone, PartialEq, Eq)]
10pub struct StepResourceAdmissionRequest {
11    pub(super) work_shape: Arc<BatchWorkShape>,
12    pub(super) fit_policy: AdmissionFitPolicy,
13    pub(super) pressure_action: AdmissionPressureAction,
14    pub(super) reusable_execution_bucket_id: Option<ReusableExecutionBucketId>,
15}
16
17impl StepResourceAdmissionRequest {
18    pub fn new(
19        work_shape: impl Into<Arc<BatchWorkShape>>,
20        fit_policy: AdmissionFitPolicy,
21        pressure_action: AdmissionPressureAction,
22    ) -> Result<Self, VNextError> {
23        Ok(Self {
24            work_shape: work_shape.into(),
25            fit_policy,
26            pressure_action,
27            reusable_execution_bucket_id: None,
28        })
29    }
30
31    pub fn with_reusable_execution_bucket(mut self, bucket_id: ReusableExecutionBucketId) -> Self {
32        self.reusable_execution_bucket_id = Some(bucket_id);
33        self
34    }
35
36    pub fn work_shape(&self) -> &BatchWorkShape {
37        &self.work_shape
38    }
39
40    pub(crate) fn immediate_shape(&self) -> DynamicResourceShape {
41        self.work_shape.immediate_shape()
42    }
43
44    pub(crate) fn fit_shape(&self) -> DynamicResourceShape {
45        match self.fit_policy {
46            AdmissionFitPolicy::ImmediateOnly => self.work_shape.immediate_shape(),
47            AdmissionFitPolicy::FullInputMustFit => self.work_shape.fit_shape(),
48        }
49    }
50
51    pub const fn fit_policy(&self) -> AdmissionFitPolicy {
52        self.fit_policy
53    }
54
55    pub const fn pressure_action(&self) -> AdmissionPressureAction {
56        self.pressure_action
57    }
58
59    pub fn reusable_execution_bucket_id(&self) -> Option<&ReusableExecutionBucketId> {
60        self.reusable_execution_bucket_id.as_ref()
61    }
62}
63
64#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
65pub struct BatchParticipantAuthority {
66    sequence_authority: SequenceAuthorityId,
67    request_authority: RequestAuthorityId,
68}
69
70impl BatchParticipantAuthority {
71    pub const fn new(
72        sequence_authority: SequenceAuthorityId,
73        request_authority: RequestAuthorityId,
74    ) -> Self {
75        Self {
76            sequence_authority,
77            request_authority,
78        }
79    }
80
81    pub const fn sequence_authority(self) -> SequenceAuthorityId {
82        self.sequence_authority
83    }
84
85    pub const fn request_authority(self) -> RequestAuthorityId {
86        self.request_authority
87    }
88
89    pub(super) const fn canonical_key(self) -> (u32, u64, u32, u64) {
90        (
91            self.sequence_authority.sparse_id(),
92            self.sequence_authority.generation(),
93            self.request_authority.sparse_id(),
94            self.request_authority.generation(),
95        )
96    }
97}
98
99/// One participant-local node topology key in the physical batch ledger.
100/// Attempt ids are deliberately absent so a fresh id cannot bypass overlap
101/// detection for the same sequence/frame/node work.
102#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize)]
103pub struct ParticipantNodeKey {
104    sequence_authority: SequenceAuthorityId,
105    request_authority: RequestAuthorityId,
106    frame_id: ExecutionFrameId,
107    node_id: NodeId,
108}
109
110impl ParticipantNodeKey {
111    pub(in crate::vnext) fn new(
112        participant: BatchParticipantAuthority,
113        frame_id: ExecutionFrameId,
114        node_id: NodeId,
115    ) -> Self {
116        Self {
117            sequence_authority: participant.sequence_authority(),
118            request_authority: participant.request_authority(),
119            frame_id,
120            node_id,
121        }
122    }
123
124    pub const fn sequence_authority(&self) -> SequenceAuthorityId {
125        self.sequence_authority
126    }
127
128    pub const fn request_authority(&self) -> RequestAuthorityId {
129        self.request_authority
130    }
131
132    pub const fn frame_id(&self) -> ExecutionFrameId {
133        self.frame_id
134    }
135
136    pub fn node_id(&self) -> &NodeId {
137        &self.node_id
138    }
139}
140
141/// Opaque association between one exact admitted participant and token work
142/// derived from that participant's actual token ids.
143#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
144pub struct BatchParticipantTokenSpan {
145    participant: BatchParticipantAuthority,
146    token_span: TokenSpanWork,
147}
148
149/// Exact packed-token projection for one participant in a scheduler step.
150/// The range addresses the shared batch transient arena; it is derived from
151/// the canonical participant work and cannot be supplied independently.
152#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
153pub struct BatchParticipantTokenRange {
154    participant: BatchParticipantAuthority,
155    immediate_start_token: u64,
156    immediate_end_token: u64,
157    source_start_token: u64,
158    source_end_token: u64,
159    full_input_tokens: u64,
160}
161
162impl BatchParticipantTokenRange {
163    fn new(
164        participant: BatchParticipantAuthority,
165        immediate_start_token: u64,
166        immediate_end_token: u64,
167        source_start_token: u64,
168        source_end_token: u64,
169        full_input_tokens: u64,
170    ) -> Result<Self, VNextError> {
171        if immediate_start_token >= immediate_end_token
172            || source_start_token >= source_end_token
173            || source_end_token > full_input_tokens
174            || immediate_end_token - immediate_start_token != source_end_token - source_start_token
175        {
176            return Err(invalid_resource(
177                "batch participant packed-token range is empty or exceeds its full input",
178            ));
179        }
180        Ok(Self {
181            participant,
182            immediate_start_token,
183            immediate_end_token,
184            source_start_token,
185            source_end_token,
186            full_input_tokens,
187        })
188    }
189
190    pub const fn participant(&self) -> BatchParticipantAuthority {
191        self.participant
192    }
193
194    pub fn immediate_token_range(&self) -> Range<u64> {
195        self.immediate_start_token..self.immediate_end_token
196    }
197
198    pub const fn immediate_tokens(&self) -> u64 {
199        self.immediate_end_token - self.immediate_start_token
200    }
201
202    pub fn source_token_range(&self) -> Range<u64> {
203        self.source_start_token..self.source_end_token
204    }
205
206    pub const fn full_input_tokens(&self) -> u64 {
207        self.full_input_tokens
208    }
209}
210
211impl BatchParticipantTokenSpan {
212    pub(super) fn new(participant: BatchParticipantAuthority, token_span: TokenSpanWork) -> Self {
213        Self {
214            participant,
215            token_span,
216        }
217    }
218
219    pub const fn participant(&self) -> BatchParticipantAuthority {
220        self.participant
221    }
222
223    pub fn token_span(&self) -> &TokenSpanWork {
224        &self.token_span
225    }
226}
227
228/// Immutable work authority for one exact non-empty participant set. The
229/// dimensions remain private so downstream claims and dispatch can only use
230/// the shape that core bound to this participant topology and fingerprint.
231#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
232pub struct BatchWorkShape {
233    participants: Vec<BatchParticipantAuthority>,
234    participant_work: Vec<BatchParticipantTokenSpan>,
235    participant_token_ranges: Vec<BatchParticipantTokenRange>,
236    resource_work: ResourceWorkShape,
237    fingerprint: String,
238}
239
240impl BatchWorkShape {
241    pub(super) fn new(
242        participant_work: Vec<BatchParticipantTokenSpan>,
243    ) -> Result<Self, VNextError> {
244        if participant_work.is_empty()
245            || participant_work.windows(2).any(|pair| {
246                pair[0].participant().canonical_key() >= pair[1].participant().canonical_key()
247            })
248        {
249            return Err(invalid_resource(
250                "batch work shape requires canonical non-empty unique participant work",
251            ));
252        }
253        let participants = participant_work
254            .iter()
255            .map(BatchParticipantTokenSpan::participant)
256            .collect::<Vec<_>>();
257        let mut next_token = 0_u64;
258        let participant_token_ranges = participant_work
259            .iter()
260            .map(|work| {
261                let start = next_token;
262                next_token = next_token
263                    .checked_add(work.token_span().immediate_tokens())
264                    .ok_or_else(|| invalid_resource("packed batch token range overflows u64"))?;
265                BatchParticipantTokenRange::new(
266                    work.participant(),
267                    start,
268                    next_token,
269                    work.token_span().immediate_token_range().start,
270                    work.token_span().immediate_token_range().end,
271                    work.token_span().full_input_tokens(),
272                )
273            })
274            .collect::<Result<Vec<_>, VNextError>>()?;
275        let resource_work = ResourceWorkShape::from_token_spans(
276            participant_work
277                .iter()
278                .map(|work| work.token_span().clone())
279                .collect(),
280        )?;
281        if resource_work.immediate_sequences()
282            != u32::try_from(participants.len())
283                .map_err(|_| invalid_resource("batch work participant count exceeds u32"))?
284            || next_token != resource_work.immediate_tokens()
285        {
286            return Err(invalid_resource(
287                "batch work shape aggregate differs from participant evidence",
288            ));
289        }
290        #[derive(Serialize)]
291        struct FingerprintInput<'a> {
292            domain: &'static str,
293            participant_work: &'a [BatchParticipantTokenSpan],
294            participant_token_ranges: &'a [BatchParticipantTokenRange],
295            resource_work_fingerprint: &'a str,
296        }
297        let input = FingerprintInput {
298            domain: "ferrum.runtime-vnext.batch-work-shape.v3",
299            participant_work: &participant_work,
300            participant_token_ranges: &participant_token_ranges,
301            resource_work_fingerprint: resource_work.fingerprint(),
302        };
303        let bytes = serde_json::to_vec(&input).map_err(|error| {
304            invalid_resource(format!("batch work shape encode failed: {error}"))
305        })?;
306        Ok(Self {
307            participants,
308            participant_work,
309            participant_token_ranges,
310            resource_work,
311            fingerprint: format!("{:x}", Sha256::digest(bytes)),
312        })
313    }
314
315    #[cfg(test)]
316    pub(crate) fn test_only(token_spans: Vec<TokenSpanWork>) -> Result<Self, VNextError> {
317        let participant_work = token_spans
318            .into_iter()
319            .enumerate()
320            .map(|(index, token_span)| {
321                let sparse_id =
322                    u32::try_from(index + 1).expect("bounded test participant index fits u32");
323                BatchParticipantTokenSpan::new(
324                    BatchParticipantAuthority::new(
325                        SequenceAuthorityId::test_only(sparse_id, 1),
326                        RequestAuthorityId::test_only(sparse_id, 1),
327                    ),
328                    token_span,
329                )
330            })
331            .collect();
332        Self::new(participant_work)
333    }
334
335    pub fn participants(&self) -> &[BatchParticipantAuthority] {
336        &self.participants
337    }
338
339    pub fn participant_work(&self) -> &[BatchParticipantTokenSpan] {
340        &self.participant_work
341    }
342
343    pub fn participant_token_ranges(&self) -> &[BatchParticipantTokenRange] {
344        &self.participant_token_ranges
345    }
346
347    pub fn resource_work(&self) -> &ResourceWorkShape {
348        &self.resource_work
349    }
350
351    pub const fn immediate_sequences(&self) -> u32 {
352        self.resource_work.immediate_sequences()
353    }
354
355    pub const fn immediate_tokens(&self) -> u64 {
356        self.resource_work.immediate_tokens()
357    }
358
359    pub const fn immediate_pages(&self) -> u64 {
360        self.resource_work.immediate_pages()
361    }
362
363    pub const fn fit_sequences(&self) -> u32 {
364        self.resource_work.fit_sequences()
365    }
366
367    pub const fn fit_tokens(&self) -> u64 {
368        self.resource_work.fit_tokens()
369    }
370
371    pub const fn fit_pages(&self) -> u64 {
372        self.resource_work.fit_pages()
373    }
374
375    pub fn fingerprint(&self) -> &str {
376        &self.fingerprint
377    }
378
379    /// Stable logical work identity for comparisons across fresh admissions.
380    /// Unlike [`Self::fingerprint`], this excludes request and sequence
381    /// authorities while retaining the exact token and committed-page shape.
382    pub fn logical_work_fingerprint(&self) -> &str {
383        self.resource_work.fingerprint()
384    }
385
386    pub(crate) const fn immediate_shape(&self) -> DynamicResourceShape {
387        self.resource_work.immediate_shape()
388    }
389
390    pub(crate) const fn fit_shape(&self) -> DynamicResourceShape {
391        self.resource_work.fit_shape()
392    }
393}
394
395#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
396pub struct StepParticipantFrameAssignment {
397    participant: BatchParticipantAuthority,
398    frame_id: ExecutionFrameId,
399}
400
401impl StepParticipantFrameAssignment {
402    pub(super) const fn new(
403        sequence_authority: SequenceAuthorityId,
404        request_authority: RequestAuthorityId,
405        frame_id: ExecutionFrameId,
406    ) -> Self {
407        Self {
408            participant: BatchParticipantAuthority::new(sequence_authority, request_authority),
409            frame_id,
410        }
411    }
412
413    pub const fn participant(self) -> BatchParticipantAuthority {
414        self.participant
415    }
416
417    pub const fn sequence_authority(self) -> SequenceAuthorityId {
418        self.participant.sequence_authority()
419    }
420
421    pub const fn request_authority(self) -> RequestAuthorityId {
422        self.participant.request_authority()
423    }
424
425    pub const fn frame_id(self) -> ExecutionFrameId {
426        self.frame_id
427    }
428
429    const fn canonical_key(self) -> (u32, u64, u32, u64) {
430        self.participant.canonical_key()
431    }
432}
433
434#[derive(Debug, Clone, PartialEq, Eq)]
435pub struct InvocationResourceAdmissionRequest {
436    pub(super) node_id: NodeId,
437    pub(super) work_shape: Arc<BatchWorkShape>,
438    pub(super) fit_policy: AdmissionFitPolicy,
439    pub(super) pressure_action: AdmissionPressureAction,
440}
441
442impl InvocationResourceAdmissionRequest {
443    pub fn new(
444        node_id: NodeId,
445        work_shape: impl Into<Arc<BatchWorkShape>>,
446        fit_policy: AdmissionFitPolicy,
447        pressure_action: AdmissionPressureAction,
448    ) -> Result<Self, VNextError> {
449        Ok(Self {
450            node_id,
451            work_shape: work_shape.into(),
452            fit_policy,
453            pressure_action,
454        })
455    }
456
457    pub fn for_all_step_participants(
458        node_id: NodeId,
459        work_shape: impl Into<Arc<BatchWorkShape>>,
460        fit_policy: AdmissionFitPolicy,
461        pressure_action: AdmissionPressureAction,
462    ) -> Result<Self, VNextError> {
463        Self::new(node_id, work_shape, fit_policy, pressure_action)
464    }
465
466    pub fn node_id(&self) -> &NodeId {
467        &self.node_id
468    }
469
470    pub fn work_shape(&self) -> &BatchWorkShape {
471        self.work_shape.as_ref()
472    }
473
474    pub(crate) fn immediate_shape(&self) -> DynamicResourceShape {
475        self.work_shape.immediate_shape()
476    }
477
478    pub(crate) fn fit_shape(&self) -> DynamicResourceShape {
479        match self.fit_policy {
480            AdmissionFitPolicy::ImmediateOnly => self.work_shape.immediate_shape(),
481            AdmissionFitPolicy::FullInputMustFit => self.work_shape.fit_shape(),
482        }
483    }
484
485    pub const fn fit_policy(&self) -> AdmissionFitPolicy {
486        self.fit_policy
487    }
488
489    pub const fn pressure_action(&self) -> AdmissionPressureAction {
490        self.pressure_action
491    }
492}
ferrum_interfaces/vnext/resource/work.rs

ferrum_interfaces/vnext/resource/
work.rs
