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

icydb_core/db/
resumable_job.rs
