icydb_core/db/
resumable_job.rs

1//! Module: db::resumable_job
2//! Responsibility: bounded application-owned resumable job state and receipts.
3//! Does not own: application authorization, accumulator meaning, or page planning.
4//! Boundary: compare-proof-and-advance session API -> excluded progress storage.
5
6use crate::db::{
7    ReadSetRevisionError, ReadSetRevisionProof, ReadSetStoreIdentity, ReadSetStoreRevision,
8};
9use candid::CandidType;
10use serde::Deserialize;
11use std::{error::Error as StdError, fmt};
12
13/// Maximum retained application accumulator/state bytes per job.
14pub const MAX_RESUMABLE_JOB_STATE_BYTES: usize = 256 * 1024;
15/// Maximum retained application receipt bytes per committed request.
16pub const MAX_RESUMABLE_JOB_RECEIPT_BYTES: usize = 64 * 1024;
17/// Maximum UTF-8 bytes in one application idempotency key.
18pub const MAX_RESUMABLE_JOB_IDEMPOTENCY_KEY_BYTES: usize = 256;
19/// Maximum UTF-8 bytes in one retained opaque continuation.
20pub const MAX_RESUMABLE_JOB_CONTINUATION_BYTES: usize = 16 * 1024;
21
22/// Nonzero application-owned identity for one durable resumable job.
23#[derive(CandidType, Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd)]
24pub struct ResumableJobId([u8; 32]);
25
26impl ResumableJobId {
27    /// Admit one nonzero application-owned job identity.
28    pub fn try_from_bytes(bytes: [u8; 32]) -> Result<Self, ResumableJobError> {
29        if bytes == [0; 32] {
30            return Err(ResumableJobError::InvalidJobId);
31        }
32        Ok(Self(bytes))
33    }
34
35    /// Return the application-owned identity bytes.
36    #[must_use]
37    pub const fn to_bytes(self) -> [u8; 32] {
38        self.0
39    }
40}
41
42/// Bounded application request identity used for lost-response replay.
43#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
44pub struct ResumableJobIdempotencyKey(String);
45
46impl ResumableJobIdempotencyKey {
47    /// Admit one nonempty bounded UTF-8 idempotency key.
48    pub fn new(value: impl Into<String>) -> Result<Self, ResumableJobError> {
49        let value = value.into();
50        if value.is_empty() || value.len() > MAX_RESUMABLE_JOB_IDEMPOTENCY_KEY_BYTES {
51            return Err(ResumableJobError::InvalidIdempotencyKey);
52        }
53        Ok(Self(value))
54    }
55
56    /// Borrow the application key.
57    #[must_use]
58    pub const fn as_str(&self) -> &str {
59        self.0.as_str()
60    }
61
62    pub(in crate::db) const fn validate(&self) -> Result<(), ResumableJobError> {
63        if self.0.is_empty() || self.0.len() > MAX_RESUMABLE_JOB_IDEMPOTENCY_KEY_BYTES {
64            return Err(ResumableJobError::InvalidIdempotencyKey);
65        }
66        Ok(())
67    }
68}
69
70/// Durable lifecycle of one generic application job.
71#[derive(CandidType, Clone, Copy, Debug, Deserialize, Eq, PartialEq)]
72pub enum ResumableJobStatus {
73    /// The next expected sequence may advance.
74    Active,
75    /// Exhaustion committed and only replay or acknowledgement remains.
76    Completed,
77    /// Protected source authority changed and the job must restart.
78    Invalidated,
79}
80
81/// Current bounded durable state of one application-owned job.
82#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
83pub struct ResumableJobState {
84    /// Application-owned job identity.
85    pub job_id: ResumableJobId,
86    /// Next sequence expected by compare-proof-and-advance.
87    pub sequence: u64,
88    /// Current job lifecycle.
89    pub status: ResumableJobStatus,
90    /// Complete immutable protected-source proof.
91    pub proof: ReadSetRevisionProof,
92    /// Opaque page continuation retained after the last successful advance.
93    pub continuation: Option<String>,
94    /// Application-defined bounded accumulator or phase state.
95    pub application_state: Vec<u8>,
96}
97
98/// Identity and expected sequence for one idempotent advance request.
99#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
100pub struct ResumableJobAdvanceRequest {
101    /// Target application job.
102    pub job_id: ResumableJobId,
103    /// Exact sequence observed before issuing the request.
104    pub expected_sequence: u64,
105    /// Stable application identity reused after a lost reply.
106    pub idempotency_key: ResumableJobIdempotencyKey,
107}
108
109impl ResumableJobAdvanceRequest {
110    /// Construct one advance request from already admitted identities.
111    #[must_use]
112    pub const fn new(
113        job_id: ResumableJobId,
114        expected_sequence: u64,
115        idempotency_key: ResumableJobIdempotencyKey,
116    ) -> Self {
117        Self {
118            job_id,
119            expected_sequence,
120            idempotency_key,
121        }
122    }
123}
124
125/// Bounded temporary next state produced by one synchronous page operation.
126#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
127pub struct ResumableJobAdvance {
128    /// Opaque continuation for the next page, or `None` after exhaustion.
129    pub continuation: Option<String>,
130    /// Complete next application accumulator or phase state.
131    pub application_state: Vec<u8>,
132    /// Bounded application receipt returned and retained for replay.
133    pub application_receipt: Vec<u8>,
134}
135
136impl ResumableJobAdvance {
137    /// Admit one bounded candidate state and receipt.
138    pub fn new(
139        continuation: Option<String>,
140        application_state: Vec<u8>,
141        application_receipt: Vec<u8>,
142    ) -> Result<Self, ResumableJobError> {
143        let advance = Self {
144            continuation,
145            application_state,
146            application_receipt,
147        };
148        advance.validate()?;
149        Ok(advance)
150    }
151
152    pub(in crate::db) fn validate(&self) -> Result<(), ResumableJobError> {
153        validate_continuation(self.continuation.as_deref())?;
154        if self.application_state.len() > MAX_RESUMABLE_JOB_STATE_BYTES
155            || self.application_receipt.len() > MAX_RESUMABLE_JOB_RECEIPT_BYTES
156        {
157            return Err(ResumableJobError::PayloadTooLarge);
158        }
159        Ok(())
160    }
161}
162
163/// Outcome committed for one advance request.
164#[derive(CandidType, Clone, Copy, Debug, Deserialize, Eq, PartialEq)]
165pub enum ResumableJobAdvanceStatus {
166    /// Candidate continuation and application state committed.
167    Advanced,
168    /// Source drift discarded the candidate and invalidated the job.
169    Invalidated,
170}
171
172/// Replayable receipt for one committed advance request.
173#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
174pub struct ResumableJobAdvanceReceipt {
175    /// Sequence named by the request.
176    pub request_sequence: u64,
177    /// Durable job sequence after this receipt committed.
178    pub committed_sequence: u64,
179    /// Whether next state or invalidation committed.
180    pub status: ResumableJobAdvanceStatus,
181    /// Committed next continuation, when advanced.
182    pub continuation: Option<String>,
183    /// Application-defined replay payload.
184    pub application_receipt: Vec<u8>,
185    idempotency_key: ResumableJobIdempotencyKey,
186}
187
188impl ResumableJobAdvanceReceipt {
189    /// Borrow the application request identity retained for replay.
190    #[must_use]
191    pub const fn idempotency_key(&self) -> &ResumableJobIdempotencyKey {
192        &self.idempotency_key
193    }
194}
195
196/// Typed protocol or persistence failure for generic resumable jobs.
197#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
198pub enum ResumableJobError {
199    /// Job identity was all zeroes.
200    InvalidJobId,
201    /// Idempotency key was empty or exceeded its byte bound.
202    InvalidIdempotencyKey,
203    /// State, continuation, or receipt exceeded its bound.
204    PayloadTooLarge,
205    /// A job with the same application identity already exists.
206    AlreadyExists,
207    /// The requested job does not exist.
208    NotFound,
209    /// The request did not name the job's current sequence.
210    StaleSequence { expected: u64, actual: u64 },
211    /// A non-replay request targeted an invalidated job.
212    Invalidated,
213    /// A non-replay request targeted a completed job.
214    Completed,
215    /// Acknowledgement targeted an active job with remaining traversal work.
216    NotTerminal,
217    /// Protected source authority was invalid or unsupported.
218    SourceProof(ReadSetRevisionError),
219    /// The shared excluded progress store reached a hard capacity.
220    CapacityExceeded,
221    /// Retained progress bytes or state closure were corrupt.
222    CorruptProgressStore,
223    /// Retained progress bytes use an unsupported current format.
224    IncompatibleProgressFormat,
225    /// An internal database invariant prevented the operation.
226    Internal,
227    /// The enclosing request exhausted aggregate IcyDB work allowance.
228    ExecutionBudgetExceeded {
229        resource: u64,
230        limit: u64,
231        observed: u64,
232        scope: u64,
233        lane: u64,
234        normalized_shape_fingerprint_prefix: u64,
235    },
236}
237
238impl fmt::Display for ResumableJobError {
239    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
240        formatter.write_str("resumable job operation failed")
241    }
242}
243
244impl StdError for ResumableJobError {}
245
246impl From<ReadSetRevisionError> for ResumableJobError {
247    fn from(error: ReadSetRevisionError) -> Self {
248        Self::SourceProof(error)
249    }
250}
251
252/// Failure from protocol handling or the application page closure.
253#[derive(Debug)]
254pub enum CompareProofAndAdvanceError<E> {
255    /// IcyDB rejected proof, sequence, bounds, or progress persistence.
256    Protocol(ResumableJobError),
257    /// The application page closure returned its own failure.
258    Operation(E),
259}
260
261impl<E: fmt::Display> fmt::Display for CompareProofAndAdvanceError<E> {
262    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
263        match self {
264            Self::Protocol(error) => error.fmt(formatter),
265            Self::Operation(error) => error.fmt(formatter),
266        }
267    }
268}
269
270impl<E: StdError + 'static> StdError for CompareProofAndAdvanceError<E> {}
271
272impl<E> From<ResumableJobError> for CompareProofAndAdvanceError<E> {
273    fn from(error: ResumableJobError) -> Self {
274        Self::Protocol(error)
275    }
276}
277
278#[derive(Clone, Debug, Eq, PartialEq)]
279pub(in crate::db) struct ResumableJobRecord {
280    state: ResumableJobState,
281    last_receipt: Option<ResumableJobAdvanceReceipt>,
282}
283
284impl ResumableJobRecord {
285    pub(in crate::db) fn new(
286        job_id: ResumableJobId,
287        proof: ReadSetRevisionProof,
288        application_state: Vec<u8>,
289    ) -> Result<Self, ResumableJobError> {
290        let state = ResumableJobState {
291            job_id,
292            sequence: 0,
293            status: ResumableJobStatus::Active,
294            proof,
295            continuation: None,
296            application_state,
297        };
298        let record = Self {
299            state,
300            last_receipt: None,
301        };
302        record.validate()?;
303        Ok(record)
304    }
305
306    pub(in crate::db) const fn state(&self) -> &ResumableJobState {
307        &self.state
308    }
309
310    pub(in crate::db) const fn last_receipt(&self) -> Option<&ResumableJobAdvanceReceipt> {
311        self.last_receipt.as_ref()
312    }
313
314    pub(in crate::db) fn apply_advance(
315        &self,
316        request: &ResumableJobAdvanceRequest,
317        advance: ResumableJobAdvance,
318    ) -> Result<(Self, ResumableJobAdvanceReceipt), ResumableJobError> {
319        advance.validate()?;
320        let committed_sequence = self
321            .state
322            .sequence
323            .checked_add(1)
324            .ok_or(ResumableJobError::CapacityExceeded)?;
325        let receipt = ResumableJobAdvanceReceipt {
326            request_sequence: request.expected_sequence,
327            committed_sequence,
328            status: ResumableJobAdvanceStatus::Advanced,
329            continuation: advance.continuation.clone(),
330            application_receipt: advance.application_receipt,
331            idempotency_key: request.idempotency_key.clone(),
332        };
333        let record = Self {
334            state: ResumableJobState {
335                job_id: self.state.job_id,
336                sequence: committed_sequence,
337                status: if advance.continuation.is_some() {
338                    ResumableJobStatus::Active
339                } else {
340                    ResumableJobStatus::Completed
341                },
342                proof: self.state.proof.clone(),
343                continuation: advance.continuation,
344                application_state: advance.application_state,
345            },
346            last_receipt: Some(receipt.clone()),
347        };
348        record.validate()?;
349        Ok((record, receipt))
350    }
351
352    pub(in crate::db) fn invalidate(
353        &self,
354        request: &ResumableJobAdvanceRequest,
355    ) -> Result<(Self, ResumableJobAdvanceReceipt), ResumableJobError> {
356        let committed_sequence = self
357            .state
358            .sequence
359            .checked_add(1)
360            .ok_or(ResumableJobError::CapacityExceeded)?;
361        let receipt = ResumableJobAdvanceReceipt {
362            request_sequence: request.expected_sequence,
363            committed_sequence,
364            status: ResumableJobAdvanceStatus::Invalidated,
365            continuation: None,
366            application_receipt: Vec::new(),
367            idempotency_key: request.idempotency_key.clone(),
368        };
369        let record = Self {
370            state: ResumableJobState {
371                sequence: committed_sequence,
372                status: ResumableJobStatus::Invalidated,
373                continuation: None,
374                ..self.state.clone()
375            },
376            last_receipt: Some(receipt.clone()),
377        };
378        record.validate()?;
379        Ok((record, receipt))
380    }
381
382    pub(in crate::db) fn validate(&self) -> Result<(), ResumableJobError> {
383        if self.state.job_id.to_bytes() == [0; 32] {
384            return Err(ResumableJobError::InvalidJobId);
385        }
386        self.state.proof.validate()?;
387        validate_continuation(self.state.continuation.as_deref())?;
388        if self.state.application_state.len() > MAX_RESUMABLE_JOB_STATE_BYTES {
389            return Err(ResumableJobError::PayloadTooLarge);
390        }
391        if let Some(receipt) = &self.last_receipt {
392            receipt.idempotency_key.validate()?;
393            validate_continuation(receipt.continuation.as_deref())?;
394            let state_matches_receipt = match (self.state.status, receipt.status) {
395                (
396                    ResumableJobStatus::Active | ResumableJobStatus::Completed,
397                    ResumableJobAdvanceStatus::Advanced,
398                ) => self.state.continuation == receipt.continuation,
399                (ResumableJobStatus::Invalidated, ResumableJobAdvanceStatus::Invalidated) => {
400                    self.state.continuation.is_none() && receipt.continuation.is_none()
401                }
402                _ => false,
403            };
404            if receipt.application_receipt.len() > MAX_RESUMABLE_JOB_RECEIPT_BYTES
405                || receipt.committed_sequence != self.state.sequence
406                || receipt.request_sequence.checked_add(1) != Some(receipt.committed_sequence)
407                || !state_matches_receipt
408            {
409                return Err(ResumableJobError::CorruptProgressStore);
410            }
411        } else if self.state.sequence != 0
412            || self.state.status != ResumableJobStatus::Active
413            || self.state.continuation.is_some()
414        {
415            return Err(ResumableJobError::CorruptProgressStore);
416        }
417        Ok(())
418    }
419}
420
421fn validate_continuation(continuation: Option<&str>) -> Result<(), ResumableJobError> {
422    if continuation.is_some_and(|value| value.len() > MAX_RESUMABLE_JOB_CONTINUATION_BYTES) {
423        return Err(ResumableJobError::PayloadTooLarge);
424    }
425    Ok(())
426}
427
428pub(in crate::db) fn encode_resumable_job_payload(
429    record: &ResumableJobRecord,
430) -> Result<Vec<u8>, ResumableJobError> {
431    record.validate()?;
432    let mut bytes = Vec::new();
433    bytes.extend_from_slice(&record.state.job_id.to_bytes());
434    bytes.extend_from_slice(&record.state.sequence.to_be_bytes());
435    bytes.push(match record.state.status {
436        ResumableJobStatus::Active => 0,
437        ResumableJobStatus::Invalidated => 1,
438        ResumableJobStatus::Completed => 2,
439    });
440    write_proof(&mut bytes, &record.state.proof)?;
441    write_optional_string(&mut bytes, record.state.continuation.as_deref())?;
442    write_bytes(&mut bytes, &record.state.application_state)?;
443    match &record.last_receipt {
444        None => bytes.push(0),
445        Some(receipt) => {
446            bytes.push(1);
447            bytes.extend_from_slice(&receipt.request_sequence.to_be_bytes());
448            bytes.extend_from_slice(&receipt.committed_sequence.to_be_bytes());
449            bytes.push(match receipt.status {
450                ResumableJobAdvanceStatus::Advanced => 0,
451                ResumableJobAdvanceStatus::Invalidated => 1,
452            });
453            write_string(&mut bytes, receipt.idempotency_key.as_str())?;
454            write_optional_string(&mut bytes, receipt.continuation.as_deref())?;
455            write_bytes(&mut bytes, &receipt.application_receipt)?;
456        }
457    }
458    Ok(bytes)
459}
460
461pub(in crate::db) fn decode_resumable_job_payload(
462    bytes: &[u8],
463) -> Result<ResumableJobRecord, ResumableJobError> {
464    let mut reader = Reader::new(bytes);
465    let job_id = ResumableJobId::try_from_bytes(reader.array()?)?;
466    let sequence = reader.u64()?;
467    let status = match reader.u8()? {
468        0 => ResumableJobStatus::Active,
469        1 => ResumableJobStatus::Invalidated,
470        2 => ResumableJobStatus::Completed,
471        _ => return Err(ResumableJobError::CorruptProgressStore),
472    };
473    let proof = read_proof(&mut reader)?;
474    let continuation = read_optional_string(&mut reader, MAX_RESUMABLE_JOB_CONTINUATION_BYTES)?;
475    let application_state = reader.bytes(MAX_RESUMABLE_JOB_STATE_BYTES)?.to_vec();
476    let last_receipt = match reader.u8()? {
477        0 => None,
478        1 => {
479            let request_sequence = reader.u64()?;
480            let committed_sequence = reader.u64()?;
481            let receipt_status = match reader.u8()? {
482                0 => ResumableJobAdvanceStatus::Advanced,
483                1 => ResumableJobAdvanceStatus::Invalidated,
484                _ => return Err(ResumableJobError::CorruptProgressStore),
485            };
486            let idempotency_key = ResumableJobIdempotencyKey::new(
487                reader.string(MAX_RESUMABLE_JOB_IDEMPOTENCY_KEY_BYTES)?,
488            )?;
489            let receipt_continuation =
490                read_optional_string(&mut reader, MAX_RESUMABLE_JOB_CONTINUATION_BYTES)?;
491            let application_receipt = reader.bytes(MAX_RESUMABLE_JOB_RECEIPT_BYTES)?.to_vec();
492            Some(ResumableJobAdvanceReceipt {
493                request_sequence,
494                committed_sequence,
495                status: receipt_status,
496                continuation: receipt_continuation,
497                application_receipt,
498                idempotency_key,
499            })
500        }
501        _ => return Err(ResumableJobError::CorruptProgressStore),
502    };
503    if !reader.is_empty() {
504        return Err(ResumableJobError::CorruptProgressStore);
505    }
506    let record = ResumableJobRecord {
507        state: ResumableJobState {
508            job_id,
509            sequence,
510            status,
511            proof,
512            continuation,
513            application_state,
514        },
515        last_receipt,
516    };
517    record.validate()?;
518    Ok(record)
519}
520
521fn write_proof(bytes: &mut Vec<u8>, proof: &ReadSetRevisionProof) -> Result<(), ResumableJobError> {
522    proof.validate()?;
523    bytes.extend_from_slice(&proof.database_incarnation());
524    bytes.extend_from_slice(&proof.accepted_root_revision().to_be_bytes());
525    bytes.push(proof.accepted_root_fingerprint_method());
526    bytes.extend_from_slice(&proof.accepted_root_fingerprint());
527    let count =
528        u32::try_from(proof.stores().len()).map_err(|_| ResumableJobError::PayloadTooLarge)?;
529    bytes.extend_from_slice(&count.to_be_bytes());
530    for store in proof.stores() {
531        bytes.extend_from_slice(&store.store().to_bytes());
532        bytes.extend_from_slice(&store.data_revision().to_be_bytes());
533        bytes.extend_from_slice(&store.access_state_revision().to_be_bytes());
534    }
535    Ok(())
536}
537
538fn read_proof(reader: &mut Reader<'_>) -> Result<ReadSetRevisionProof, ResumableJobError> {
539    let database_incarnation = reader.array()?;
540    let accepted_root_revision = reader.u64()?;
541    let accepted_root_fingerprint_method = reader.u8()?;
542    let accepted_root_fingerprint = reader.array()?;
543    let count = reader.u32()? as usize;
544    if count == 0 || count > crate::db::MAX_READ_SET_PROOF_STORES {
545        return Err(ResumableJobError::CorruptProgressStore);
546    }
547    let mut stores = Vec::with_capacity(count);
548    for _ in 0..count {
549        stores.push(ReadSetStoreRevision::new(
550            ReadSetStoreIdentity::from_bytes(reader.array()?),
551            reader.u64()?,
552            reader.u64()?,
553        ));
554    }
555    ReadSetRevisionProof::from_parts(
556        database_incarnation,
557        accepted_root_revision,
558        accepted_root_fingerprint_method,
559        accepted_root_fingerprint,
560        stores,
561    )
562    .map_err(Into::into)
563}
564
565fn write_string(bytes: &mut Vec<u8>, value: &str) -> Result<(), ResumableJobError> {
566    write_bytes(bytes, value.as_bytes())
567}
568
569fn write_optional_string(
570    bytes: &mut Vec<u8>,
571    value: Option<&str>,
572) -> Result<(), ResumableJobError> {
573    match value {
574        None => bytes.push(0),
575        Some(value) => {
576            bytes.push(1);
577            write_string(bytes, value)?;
578        }
579    }
580    Ok(())
581}
582
583fn write_bytes(bytes: &mut Vec<u8>, value: &[u8]) -> Result<(), ResumableJobError> {
584    let len = u32::try_from(value.len()).map_err(|_| ResumableJobError::PayloadTooLarge)?;
585    bytes.extend_from_slice(&len.to_be_bytes());
586    bytes.extend_from_slice(value);
587    Ok(())
588}
589
590fn read_optional_string(
591    reader: &mut Reader<'_>,
592    max: usize,
593) -> Result<Option<String>, ResumableJobError> {
594    match reader.u8()? {
595        0 => Ok(None),
596        1 => reader.string(max).map(Some),
597        _ => Err(ResumableJobError::CorruptProgressStore),
598    }
599}
600
601struct Reader<'a> {
602    bytes: &'a [u8],
603    offset: usize,
604}
605
606impl<'a> Reader<'a> {
607    const fn new(bytes: &'a [u8]) -> Self {
608        Self { bytes, offset: 0 }
609    }
610
611    fn u8(&mut self) -> Result<u8, ResumableJobError> {
612        let value = *self
613            .bytes
614            .get(self.offset)
615            .ok_or(ResumableJobError::CorruptProgressStore)?;
616        self.offset += 1;
617        Ok(value)
618    }
619
620    fn u32(&mut self) -> Result<u32, ResumableJobError> {
621        Ok(u32::from_be_bytes(self.array()?))
622    }
623
624    fn u64(&mut self) -> Result<u64, ResumableJobError> {
625        Ok(u64::from_be_bytes(self.array()?))
626    }
627
628    fn array<const N: usize>(&mut self) -> Result<[u8; N], ResumableJobError> {
629        self.take(N)?
630            .try_into()
631            .map_err(|_| ResumableJobError::CorruptProgressStore)
632    }
633
634    fn bytes(&mut self, max: usize) -> Result<&'a [u8], ResumableJobError> {
635        let len = self.u32()? as usize;
636        if len > max {
637            return Err(ResumableJobError::CorruptProgressStore);
638        }
639        self.take(len)
640    }
641
642    fn string(&mut self, max: usize) -> Result<String, ResumableJobError> {
643        let bytes = self.bytes(max)?;
644        std::str::from_utf8(bytes)
645            .map(str::to_string)
646            .map_err(|_| ResumableJobError::CorruptProgressStore)
647    }
648
649    fn take(&mut self, len: usize) -> Result<&'a [u8], ResumableJobError> {
650        let end = self
651            .offset
652            .checked_add(len)
653            .ok_or(ResumableJobError::CorruptProgressStore)?;
654        let bytes = self
655            .bytes
656            .get(self.offset..end)
657            .ok_or(ResumableJobError::CorruptProgressStore)?;
658        self.offset = end;
659        Ok(bytes)
660    }
661
662    const fn is_empty(&self) -> bool {
663        self.offset == self.bytes.len()
664    }
665}
666
667#[cfg(test)]
668mod tests {
669    use super::*;
670
671    fn proof() -> ReadSetRevisionProof {
672        ReadSetRevisionProof::from_parts(
673            [1; 16],
674            7,
675            1,
676            [2; 32],
677            vec![ReadSetStoreRevision::new(
678                ReadSetStoreIdentity::from_bytes([3; 32]),
679                11,
680                13,
681            )],
682        )
683        .expect("bounded canonical proof should admit")
684    }
685
686    fn job_id() -> ResumableJobId {
687        ResumableJobId::try_from_bytes([4; 32]).expect("nonzero job identity should admit")
688    }
689
690    #[test]
691    fn current_resumable_job_payload_round_trips_state_and_replay_receipt() {
692        let record = ResumableJobRecord::new(job_id(), proof(), vec![1, 2, 3])
693            .expect("initial resumable record should admit");
694        let request = ResumableJobAdvanceRequest::new(
695            job_id(),
696            0,
697            ResumableJobIdempotencyKey::new("page-0")
698                .expect("bounded idempotency key should admit"),
699        );
700        let advance = ResumableJobAdvance::new(
701            Some("opaque-continuation".to_string()),
702            vec![4, 5],
703            vec![6, 7],
704        )
705        .expect("bounded advance should admit");
706        let (advanced, _) = record
707            .apply_advance(&request, advance)
708            .expect("current request should advance");
709
710        let bytes = encode_resumable_job_payload(&advanced)
711            .expect("current resumable payload should encode");
712        assert!(!bytes.starts_with(b"DIDL"));
713        assert_eq!(
714            decode_resumable_job_payload(&bytes).expect("current resumable payload should decode"),
715            advanced,
716        );
717    }
718
719    #[test]
720    fn resumable_job_payload_rejects_truncation_and_trailing_bytes() {
721        let record = ResumableJobRecord::new(job_id(), proof(), Vec::new())
722            .expect("initial resumable record should admit");
723        let bytes =
724            encode_resumable_job_payload(&record).expect("current resumable payload should encode");
725
726        assert_eq!(
727            decode_resumable_job_payload(&bytes[..bytes.len() - 1]),
728            Err(ResumableJobError::CorruptProgressStore),
729        );
730        let mut trailing = bytes;
731        trailing.push(0);
732        assert_eq!(
733            decode_resumable_job_payload(&trailing),
734            Err(ResumableJobError::CorruptProgressStore),
735        );
736    }
737}
icydb_core/db/resumable_job.rs

icydb_core/db/
resumable_job.rs
