laminar_core/state/
in_process.rs

1//! [`InProcessBackend`] — non-durable checkpoint-artifact storage backed by
2//! an in-memory hashmap. Used for tests and embedded single-process runs.
3
4use async_trait::async_trait;
5use bytes::Bytes;
6use parking_lot::RwLock;
7use rustc_hash::FxHashMap;
8
9use crate::checkpoint::{CheckpointAssignmentFence, LeaderProof, PipelineIdentity};
10
11use super::backend::{
12    digest_hex, sha256, CheckpointAttempt, CheckpointSeal, CheckpointSealInventory,
13    SealedCommitDescriptor, SealedCommitDescriptorWriter, SealedVnodePartial, SealedVnodeWriter,
14    StateBackend, StateBackendError, StateNamespaceBinding, STATE_NAMESPACE_RESOURCE,
15};
16
17#[derive(Debug, Clone, PartialEq, Eq)]
18struct StoredPartial {
19    bytes: Bytes,
20    attestation: SealedVnodePartial,
21}
22
23#[derive(Debug, Clone, PartialEq, Eq)]
24struct StoredCommitDescriptor {
25    bytes: Bytes,
26    attestation: SealedCommitDescriptor,
27}
28
29/// In-process, non-durable checkpoint-artifact backend.
30#[derive(Debug)]
31pub struct InProcessBackend {
32    namespace: RwLock<Option<StateNamespaceBinding>>,
33    partials: RwLock<FxHashMap<(CheckpointAttempt, u32), StoredPartial>>,
34    /// `attempt -> key -> descriptor`, mirroring the durable attempt namespace.
35    descriptors: RwLock<FxHashMap<CheckpointAttempt, FxHashMap<String, StoredCommitDescriptor>>>,
36    /// Exact attempts sealed by [`StateBackend::seal_checkpoint`].
37    sealed: RwLock<FxHashMap<CheckpointAttempt, CheckpointSeal>>,
38    execution_id: uuid::Uuid,
39    vnode_capacity: u32,
40}
41
42impl InProcessBackend {
43    /// Create a new backend sized for `vnode_capacity` vnodes.
44    #[must_use]
45    pub fn new(vnode_capacity: u32) -> Self {
46        Self {
47            namespace: RwLock::new(None),
48            partials: RwLock::new(FxHashMap::default()),
49            descriptors: RwLock::new(FxHashMap::default()),
50            sealed: RwLock::new(FxHashMap::default()),
51            execution_id: uuid::Uuid::new_v4(),
52            vnode_capacity,
53        }
54    }
55
56    fn check_vnode(&self, v: u32) -> Result<(), StateBackendError> {
57        if v >= self.vnode_capacity {
58            Err(StateBackendError::Io(format!(
59                "vnode {v} out of range (capacity {})",
60                self.vnode_capacity
61            )))
62        } else {
63            Ok(())
64        }
65    }
66
67    fn ensure_canonical_attempt(attempt: CheckpointAttempt) -> Result<(), StateBackendError> {
68        if attempt.is_canonical() {
69            Ok(())
70        } else {
71            Err(StateBackendError::Conflict {
72                resource: format!(
73                    "state-v2/epoch={}/checkpoint={}",
74                    attempt.epoch, attempt.checkpoint_id
75                ),
76                message: "state attempt must use one nonzero canonical checkpoint ID".into(),
77            })
78        }
79    }
80
81    fn store_partial(
82        &self,
83        attempt: CheckpointAttempt,
84        vnode: u32,
85        assignment_version: u64,
86        writer: Option<SealedVnodeWriter>,
87        bytes: Bytes,
88    ) -> Result<(), StateBackendError> {
89        Self::ensure_canonical_attempt(attempt)?;
90        self.check_vnode(vnode)?;
91        let stored = StoredPartial {
92            attestation: SealedVnodePartial::new(vnode, assignment_version, writer, &bytes),
93            bytes,
94        };
95        let mut partials = self.partials.write();
96        match partials.get(&(attempt, vnode)) {
97            Some(existing) if existing == &stored => Ok(()),
98            Some(_) => Err(StateBackendError::Conflict {
99                resource: format!(
100                    "state-v2/epoch={}/checkpoint={}/vnode={vnode}/partial.bin",
101                    attempt.epoch, attempt.checkpoint_id
102                ),
103                message: "partial already exists with different bytes or writer certificate".into(),
104            }),
105            None => {
106                partials.insert((attempt, vnode), stored);
107                Ok(())
108            }
109        }
110    }
111
112    fn store_commit_descriptor(
113        &self,
114        attempt: CheckpointAttempt,
115        key: &str,
116        assignment_version: u64,
117        writer: Option<SealedCommitDescriptorWriter>,
118        bytes: Bytes,
119    ) -> Result<(), StateBackendError> {
120        Self::ensure_canonical_attempt(attempt)?;
121        if key.is_empty() {
122            return Err(StateBackendError::Conflict {
123                resource: format!(
124                    "state-v2/epoch={}/checkpoint={}/commit/",
125                    attempt.epoch, attempt.checkpoint_id
126                ),
127                message: "commit descriptor key is empty".into(),
128            });
129        }
130        let stored = StoredCommitDescriptor {
131            attestation: SealedCommitDescriptor::new(
132                key.to_owned(),
133                assignment_version,
134                writer,
135                &bytes,
136            ),
137            bytes,
138        };
139        let mut descriptors = self.descriptors.write();
140        let attempt_descriptors = descriptors.entry(attempt).or_default();
141        match attempt_descriptors.get(key) {
142            Some(existing) if existing == &stored => Ok(()),
143            Some(_) => Err(StateBackendError::Conflict {
144                resource: format!(
145                    "state-v2/epoch={}/checkpoint={}/commit/{key}",
146                    attempt.epoch, attempt.checkpoint_id
147                ),
148                message:
149                    "commit descriptor already exists with different bytes or writer certificate"
150                        .into(),
151            }),
152            None => {
153                attempt_descriptors.insert(key.to_owned(), stored);
154                Ok(())
155            }
156        }
157    }
158}
159
160#[async_trait]
161impl StateBackend for InProcessBackend {
162    fn key_group_capacity(&self) -> u32 {
163        self.vnode_capacity
164    }
165
166    async fn bind_state_namespace(
167        &self,
168        deployment_id: &str,
169        pipeline_identity: &PipelineIdentity,
170    ) -> Result<(), StateBackendError> {
171        let requested = StateNamespaceBinding::try_new(deployment_id, pipeline_identity)?;
172        let mut namespace = self.namespace.write();
173        match namespace.as_ref() {
174            Some(existing) if existing == &requested => Ok(()),
175            Some(_) => Err(StateBackendError::Conflict {
176                resource: STATE_NAMESPACE_RESOURCE.into(),
177                message: "in-process state backend is already bound to a different namespace"
178                    .into(),
179            }),
180            None => {
181                *namespace = Some(requested);
182                Ok(())
183            }
184        }
185    }
186
187    /// In-process backend opts out of the assignment fence — there's
188    /// only one process so the scenario is moot. `assignment_version`
189    /// is accepted and ignored.
190    async fn write_partial(
191        &self,
192        attempt: CheckpointAttempt,
193        vnode: u32,
194        assignment_version: u64,
195        bytes: Bytes,
196    ) -> Result<(), StateBackendError> {
197        self.store_partial(attempt, vnode, assignment_version, None, bytes)
198    }
199
200    async fn write_certified_partial(
201        &self,
202        attempt: CheckpointAttempt,
203        vnode: u32,
204        assignment_fence: &CheckpointAssignmentFence,
205        writer_node_id: u64,
206        bytes: Bytes,
207    ) -> Result<(), StateBackendError> {
208        let writer =
209            SealedVnodeWriter::from_fence(assignment_fence, writer_node_id).ok_or_else(|| {
210                StateBackendError::Conflict {
211                    resource: format!(
212                        "state-v2/epoch={}/checkpoint={}/vnode={vnode}/partial.bin",
213                        attempt.epoch, attempt.checkpoint_id
214                    ),
215                    message: "partial writer is absent from the canonical assignment certificate"
216                        .into(),
217                }
218            })?;
219        if !assignment_fence.is_canonical() {
220            return Err(StateBackendError::Conflict {
221                resource: format!(
222                    "state-v2/epoch={}/checkpoint={}/vnode={vnode}/partial.bin",
223                    attempt.epoch, attempt.checkpoint_id
224                ),
225                message: "assignment certificate is not canonical".into(),
226            });
227        }
228        self.store_partial(
229            attempt,
230            vnode,
231            assignment_fence.assignment_version,
232            Some(writer),
233            bytes,
234        )
235    }
236
237    async fn read_partial(
238        &self,
239        attempt: CheckpointAttempt,
240        vnode: u32,
241    ) -> Result<Option<Bytes>, StateBackendError> {
242        Self::ensure_canonical_attempt(attempt)?;
243        self.check_vnode(vnode)?;
244        Ok(self
245            .partials
246            .read()
247            .get(&(attempt, vnode))
248            .map(|partial| partial.bytes.clone()))
249    }
250
251    async fn write_commit_descriptor(
252        &self,
253        attempt: CheckpointAttempt,
254        key: &str,
255        bytes: Bytes,
256    ) -> Result<(), StateBackendError> {
257        self.store_commit_descriptor(attempt, key, 0, None, bytes)
258    }
259
260    async fn write_certified_commit_descriptor(
261        &self,
262        attempt: CheckpointAttempt,
263        key: &str,
264        assignment_fence: &CheckpointAssignmentFence,
265        writer_node_id: u64,
266        leader_proof: &LeaderProof,
267        bytes: Bytes,
268    ) -> Result<(), StateBackendError> {
269        let writer = SealedCommitDescriptorWriter::from_fence(
270            assignment_fence,
271            writer_node_id,
272            leader_proof,
273        )
274        .ok_or_else(|| StateBackendError::Conflict {
275            resource: format!(
276                "state-v2/epoch={}/checkpoint={}/commit/{key}",
277                attempt.epoch, attempt.checkpoint_id
278            ),
279            message: "commit descriptor writer or leader is absent from the canonical assignment certificate"
280                .into(),
281        })?;
282        self.store_commit_descriptor(
283            attempt,
284            key,
285            assignment_fence.assignment_version,
286            Some(writer),
287            bytes,
288        )
289    }
290
291    async fn read_commit_descriptor(
292        &self,
293        attempt: CheckpointAttempt,
294        key: &str,
295    ) -> Result<Option<Bytes>, StateBackendError> {
296        Self::ensure_canonical_attempt(attempt)?;
297        Ok(self
298            .descriptors
299            .read()
300            .get(&attempt)
301            .and_then(|descriptors| descriptors.get(key))
302            .map(|descriptor| descriptor.bytes.clone()))
303    }
304
305    async fn read_sealed_commit_descriptor_bounded(
306        &self,
307        attempt: CheckpointAttempt,
308        sealed: &SealedCommitDescriptor,
309        max_bytes: u64,
310    ) -> Result<Option<Bytes>, StateBackendError> {
311        Self::ensure_canonical_attempt(attempt)?;
312        let resource = format!(
313            "state-v2/epoch={}/checkpoint={}/commit/{}",
314            attempt.epoch, attempt.checkpoint_id, sealed.key
315        );
316        if sealed.payload_len > max_bytes {
317            return Err(StateBackendError::Conflict {
318                resource,
319                message: format!(
320                    "sealed commit descriptor declares {} bytes; read bound is {max_bytes}",
321                    sealed.payload_len
322                ),
323            });
324        }
325        let descriptors = self.descriptors.read();
326        let Some(stored) = descriptors
327            .get(&attempt)
328            .and_then(|descriptors| descriptors.get(&sealed.key))
329        else {
330            return Ok(None);
331        };
332        if &stored.attestation != sealed {
333            return Err(StateBackendError::Conflict {
334                resource,
335                message: "stored commit descriptor attestation does not match the checkpoint seal"
336                    .into(),
337            });
338        }
339        if stored.bytes.len() as u64 != sealed.payload_len
340            || digest_hex(&sha256(&stored.bytes)) != sealed.payload_sha256
341        {
342            return Err(StateBackendError::Serialization(
343                "stored commit descriptor payload does not match its sealed length and digest"
344                    .into(),
345            ));
346        }
347        Ok(Some(stored.bytes.clone()))
348    }
349
350    async fn seal_checkpoint(
351        &self,
352        attempt: CheckpointAttempt,
353        assignment_fence: Option<&CheckpointAssignmentFence>,
354        vnodes: &[u32],
355        required_descriptors: &[String],
356    ) -> Result<bool, StateBackendError> {
357        Self::ensure_canonical_attempt(attempt)?;
358        if assignment_fence.is_some_and(|fence| !fence.is_canonical()) {
359            return Err(StateBackendError::Conflict {
360                resource: format!(
361                    "state-v2/epoch={}/checkpoint={}/_SEAL",
362                    attempt.epoch, attempt.checkpoint_id
363                ),
364                message: "assignment certificate is not canonical".into(),
365            });
366        }
367        let assignment_version = assignment_fence.map_or(0, |fence| fence.assignment_version);
368        let sealed_partials = {
369            let map = self.partials.read();
370            let mut sealed_partials = Vec::with_capacity(vnodes.len());
371            for &v in vnodes {
372                self.check_vnode(v)?;
373                let Some(partial) = map.get(&(attempt, v)) else {
374                    return Ok(false);
375                };
376                if partial.attestation.assignment_version != assignment_version {
377                    return Err(StateBackendError::Conflict {
378                        resource: format!(
379                            "state-v2/epoch={}/checkpoint={}/vnode={v}/partial.bin",
380                            attempt.epoch, attempt.checkpoint_id
381                        ),
382                        message: format!(
383                            "partial assignment version {} cannot satisfy seal version {assignment_version}",
384                            partial.attestation.assignment_version
385                        ),
386                    });
387                }
388                sealed_partials.push(partial.attestation.clone());
389            }
390            sealed_partials
391        };
392        let sealed_descriptors = {
393            let descs = self.descriptors.read();
394            let attempt_descs = descs.get(&attempt);
395            let mut sealed_descriptors = Vec::with_capacity(required_descriptors.len());
396            for key in required_descriptors {
397                let Some(descriptor) = attempt_descs.and_then(|items| items.get(key)) else {
398                    return Ok(false);
399                };
400                sealed_descriptors.push(descriptor.attestation.clone());
401            }
402            sealed_descriptors
403        };
404        let seal = CheckpointSeal::new(
405            "in-process".to_string(),
406            self.execution_id,
407            CheckpointSealInventory {
408                attempt,
409                assignment_fence: assignment_fence.cloned(),
410                assignment_version,
411                required_vnodes: vnodes.to_vec(),
412                sealed_partials,
413                required_descriptors: required_descriptors.to_vec(),
414                sealed_descriptors,
415            },
416        );
417        seal.validate()
418            .map_err(|message| StateBackendError::Conflict {
419                resource: format!(
420                    "state-v2/epoch={}/checkpoint={}/_SEAL",
421                    attempt.epoch, attempt.checkpoint_id
422                ),
423                message,
424            })?;
425        let mut sealed = self.sealed.write();
426        match sealed.get(&attempt) {
427            Some(existing) if existing == &seal => Ok(true),
428            Some(existing) => Err(StateBackendError::Conflict {
429                resource: format!(
430                    "state-v2/epoch={}/checkpoint={}/_SEAL",
431                    attempt.epoch, attempt.checkpoint_id
432                ),
433                message: format!(
434                    "existing seal belongs to execution {}",
435                    existing.execution_id
436                ),
437            }),
438            None => {
439                sealed.insert(attempt, seal);
440                Ok(true)
441            }
442        }
443    }
444
445    async fn checkpoint_seal_inventory(
446        &self,
447        attempt: CheckpointAttempt,
448    ) -> Result<Option<CheckpointSealInventory>, StateBackendError> {
449        Self::ensure_canonical_attempt(attempt)?;
450        Ok(self
451            .sealed
452            .read()
453            .get(&attempt)
454            .map(CheckpointSeal::inventory))
455    }
456
457    async fn verify_checkpoint_artifact_metadata(
458        &self,
459        inventory: &CheckpointSealInventory,
460    ) -> Result<(), StateBackendError> {
461        let attempt = inventory.attempt;
462        Self::ensure_canonical_attempt(attempt)?;
463        let resource = format!(
464            "state-v2/epoch={}/checkpoint={}",
465            attempt.epoch, attempt.checkpoint_id
466        );
467        let sealed = self.sealed.read();
468        let Some(seal) = sealed.get(&attempt) else {
469            return Err(StateBackendError::Conflict {
470                resource,
471                message: "checkpoint seal is missing during metadata verification".into(),
472            });
473        };
474        if seal.inventory() != *inventory {
475            return Err(StateBackendError::Conflict {
476                resource,
477                message: "checkpoint seal changed during metadata verification".into(),
478            });
479        }
480
481        let partials = self.partials.read();
482        for expected in &inventory.sealed_partials {
483            let path = format!(
484                "state-v2/epoch={}/checkpoint={}/vnode={}/partial.bin",
485                attempt.epoch, attempt.checkpoint_id, expected.vnode
486            );
487            let Some(stored) = partials.get(&(attempt, expected.vnode)) else {
488                return Err(StateBackendError::Conflict {
489                    resource: path,
490                    message: "sealed vnode partial is missing".into(),
491                });
492            };
493            let stored_len =
494                u64::try_from(stored.bytes.len()).map_err(|_| StateBackendError::Conflict {
495                    resource: path.clone(),
496                    message: "sealed vnode partial length is not representable".into(),
497                })?;
498            if stored.attestation != *expected || stored_len != expected.payload_len {
499                return Err(StateBackendError::Conflict {
500                    resource: path,
501                    message: "sealed vnode partial metadata does not match the seal".into(),
502                });
503            }
504        }
505
506        let descriptors = self.descriptors.read();
507        let attempt_descriptors = descriptors.get(&attempt);
508        for expected in &inventory.sealed_descriptors {
509            let path = format!(
510                "state-v2/epoch={}/checkpoint={}/commit/{}",
511                attempt.epoch, attempt.checkpoint_id, expected.key
512            );
513            let Some(stored) = attempt_descriptors.and_then(|entries| entries.get(&expected.key))
514            else {
515                return Err(StateBackendError::Conflict {
516                    resource: path,
517                    message: "sealed commit descriptor is missing".into(),
518                });
519            };
520            let stored_len =
521                u64::try_from(stored.bytes.len()).map_err(|_| StateBackendError::Conflict {
522                    resource: path.clone(),
523                    message: "sealed commit descriptor length is not representable".into(),
524                })?;
525            if stored.attestation != *expected || stored_len != expected.payload_len {
526                return Err(StateBackendError::Conflict {
527                    resource: path,
528                    message: "sealed commit descriptor metadata does not match the seal".into(),
529                });
530            }
531        }
532        Ok(())
533    }
534
535    async fn prune_before(&self, before: u64) -> Result<(), StateBackendError> {
536        // Without this, every checkpoint leaks one Bytes per vnode
537        // forever.
538        self.sealed
539            .write()
540            .retain(|attempt, _| attempt.epoch >= before);
541        self.partials
542            .write()
543            .retain(|&(attempt, _), _| attempt.epoch >= before);
544        self.descriptors
545            .write()
546            .retain(|attempt, _| attempt.epoch >= before);
547        Ok(())
548    }
549}
550
551#[cfg(test)]
552mod tests {
553    use super::*;
554    use crate::state::StateBackendDurability;
555
556    fn attempt(checkpoint_id: u64) -> CheckpointAttempt {
557        CheckpointAttempt::canonical(checkpoint_id)
558    }
559
560    #[tokio::test]
561    async fn namespace_binding_is_create_once_and_idempotent() {
562        let backend = InProcessBackend::new(1);
563        let deployment = uuid::Uuid::from_u128(1).to_string();
564        let identity = crate::checkpoint::PipelineIdentity::empty();
565
566        backend
567            .bind_state_namespace(&deployment, &identity)
568            .await
569            .unwrap();
570        backend
571            .bind_state_namespace(&deployment, &identity)
572            .await
573            .unwrap();
574
575        let other_deployment = uuid::Uuid::from_u128(2).to_string();
576        assert!(matches!(
577            backend
578                .bind_state_namespace(&other_deployment, &identity)
579                .await,
580            Err(StateBackendError::Conflict { .. })
581        ));
582        let mut other_identity = identity;
583        other_identity.sha256 = "11".repeat(32);
584        assert!(matches!(
585            backend
586                .bind_state_namespace(&deployment, &other_identity)
587                .await,
588            Err(StateBackendError::Conflict { .. })
589        ));
590    }
591
592    #[tokio::test]
593    async fn write_read_roundtrip() {
594        let b = InProcessBackend::new(4);
595        let checkpoint = attempt(7);
596        let payload = Bytes::from_static(b"hello");
597        b.write_partial(checkpoint, 2, 0, payload.clone())
598            .await
599            .unwrap();
600        let got = b.read_partial(checkpoint, 2).await.unwrap().unwrap();
601        assert_eq!(got, payload);
602        assert!(b.read_partial(attempt(8), 2).await.unwrap().is_none());
603    }
604
605    #[tokio::test]
606    async fn noncanonical_attempt_is_rejected_before_state_mutation() {
607        let backend = InProcessBackend::new(1);
608        let invalid = CheckpointAttempt::new(1, 2);
609
610        let error = backend
611            .write_partial(invalid, 0, 0, Bytes::from_static(b"invalid"))
612            .await
613            .unwrap_err();
614        assert!(error.to_string().contains("canonical checkpoint ID"));
615        assert!(backend.partials.read().is_empty());
616        assert!(backend.read_partial(invalid, 0).await.is_err());
617    }
618
619    #[tokio::test]
620    async fn immutable_partial_accepts_identical_retry_and_rejects_conflict() {
621        let b = InProcessBackend::new(4);
622        let checkpoint = attempt(1);
623        b.write_partial(checkpoint, 0, 0, Bytes::from_static(b"first"))
624            .await
625            .unwrap();
626        b.write_partial(checkpoint, 0, 0, Bytes::from_static(b"first"))
627            .await
628            .unwrap();
629        assert!(matches!(
630            b.write_partial(checkpoint, 0, 0, Bytes::from_static(b"different"))
631                .await,
632            Err(StateBackendError::Conflict { .. })
633        ));
634        assert_eq!(
635            b.read_partial(checkpoint, 0).await.unwrap().unwrap(),
636            Bytes::from_static(b"first")
637        );
638    }
639
640    #[tokio::test]
641    async fn descriptors_are_immutable_per_attempt() {
642        let b = InProcessBackend::new(2);
643        let checkpoint = attempt(2);
644        b.write_commit_descriptor(checkpoint, "sink", Bytes::from_static(b"d"))
645            .await
646            .unwrap();
647        b.write_commit_descriptor(checkpoint, "sink", Bytes::from_static(b"d"))
648            .await
649            .unwrap();
650        assert!(matches!(
651            b.write_commit_descriptor(checkpoint, "sink", Bytes::from_static(b"other"))
652                .await,
653            Err(StateBackendError::Conflict { .. })
654        ));
655        assert_eq!(
656            b.read_commit_descriptor(checkpoint, "sink").await.unwrap(),
657            Some(Bytes::from_static(b"d"))
658        );
659        assert_eq!(
660            b.read_commit_descriptor(checkpoint, "missing")
661                .await
662                .unwrap(),
663            None
664        );
665    }
666
667    #[tokio::test]
668    async fn cluster_seal_requires_exact_descriptor_process_and_leader() {
669        use crate::checkpoint::{CheckpointParticipant, LeaderProofOwner};
670
671        let boot = uuid::Uuid::from_u128(1);
672        let fence = CheckpointAssignmentFence::from_owner_map(
673            1,
674            &[1],
675            vec![CheckpointParticipant {
676                node_id: 1,
677                boot_incarnation: boot,
678            }],
679        )
680        .unwrap();
681        let proof = LeaderProof {
682            owner: LeaderProofOwner {
683                node_id: 1,
684                boot_id: boot,
685                process_term: 1,
686            },
687            fencing_token: 1,
688        };
689        let key = "participant=1/ready";
690
691        let poisoned = InProcessBackend::new(1);
692        poisoned
693            .write_commit_descriptor(attempt(3), key, Bytes::from_static(b"stale"))
694            .await
695            .unwrap();
696        assert!(matches!(
697            poisoned
698                .seal_checkpoint(attempt(3), Some(&fence), &[], &[key.to_owned()])
699                .await,
700            Err(StateBackendError::Conflict { .. })
701        ));
702
703        let valid = InProcessBackend::new(1);
704        valid
705            .write_certified_commit_descriptor(
706                attempt(4),
707                key,
708                &fence,
709                1,
710                &proof,
711                Bytes::from_static(b"ready"),
712            )
713            .await
714            .unwrap();
715        assert!(valid
716            .seal_checkpoint(attempt(4), Some(&fence), &[], &[key.to_owned()])
717            .await
718            .unwrap());
719        let inventory = valid
720            .checkpoint_seal_inventory(attempt(4))
721            .await
722            .unwrap()
723            .unwrap();
724        assert_eq!(inventory.descriptor_leader_proof().unwrap(), Some(&proof));
725        let sealed = inventory.sealed_descriptor(key).unwrap();
726        assert_eq!(
727            valid
728                .read_sealed_commit_descriptor_bounded(attempt(4), sealed, 5)
729                .await
730                .unwrap(),
731            Some(Bytes::from_static(b"ready"))
732        );
733
734        let mut wrong_attestation = sealed.clone();
735        wrong_attestation.payload_sha256 = "00".repeat(32);
736        let error = valid
737            .read_sealed_commit_descriptor_bounded(attempt(4), &wrong_attestation, 5)
738            .await
739            .unwrap_err();
740        assert!(matches!(error, StateBackendError::Conflict { .. }));
741        assert!(error
742            .to_string()
743            .contains("does not match the checkpoint seal"));
744
745        let error = valid
746            .read_sealed_commit_descriptor_bounded(attempt(4), sealed, 4)
747            .await
748            .unwrap_err();
749        assert!(matches!(error, StateBackendError::Conflict { .. }));
750        assert!(error.to_string().contains("read bound is 4"));
751    }
752
753    #[tokio::test]
754    async fn checkpoint_attempts_are_isolated() {
755        let b = InProcessBackend::new(2);
756        let old = CheckpointAttempt::canonical(5);
757        let new = CheckpointAttempt::canonical(99);
758        b.write_partial(old, 0, 0, Bytes::from_static(b"old"))
759            .await
760            .unwrap();
761        b.write_partial(new, 0, 0, Bytes::from_static(b"new"))
762            .await
763            .unwrap();
764        assert_eq!(
765            b.read_partial(old, 0).await.unwrap().unwrap(),
766            Bytes::from_static(b"old")
767        );
768        assert_eq!(
769            b.read_partial(new, 0).await.unwrap().unwrap(),
770            Bytes::from_static(b"new")
771        );
772    }
773
774    #[test]
775    fn in_process_backend_is_volatile() {
776        assert_eq!(
777            InProcessBackend::new(4).durability_scope(),
778            StateBackendDurability::Volatile
779        );
780    }
781
782    #[tokio::test]
783    async fn seal_checkpoint_requires_every_vnode() {
784        let b = InProcessBackend::new(4);
785        let checkpoint = attempt(1);
786        let vnodes = [0u32, 1, 2];
787        assert!(!b
788            .seal_checkpoint(checkpoint, None, &vnodes, &[])
789            .await
790            .unwrap());
791        b.write_partial(checkpoint, 0, 0, Bytes::from_static(b"a"))
792            .await
793            .unwrap();
794        b.write_partial(checkpoint, 1, 0, Bytes::from_static(b"b"))
795            .await
796            .unwrap();
797        assert!(!b
798            .seal_checkpoint(checkpoint, None, &vnodes, &[])
799            .await
800            .unwrap());
801        b.write_partial(checkpoint, 2, 0, Bytes::from_static(b"c"))
802            .await
803            .unwrap();
804        assert!(b
805            .seal_checkpoint(checkpoint, None, &vnodes, &[])
806            .await
807            .unwrap());
808        let inventory = b
809            .checkpoint_seal_inventory(checkpoint)
810            .await
811            .unwrap()
812            .unwrap();
813        assert_eq!(inventory.attempt, checkpoint);
814        assert_eq!(inventory.required_vnodes, vnodes);
815        assert_eq!(inventory.required_descriptors, Vec::<String>::new());
816        assert!(inventory.sealed_descriptors.is_empty());
817        assert_eq!(inventory.sealed_partials.len(), vnodes.len());
818        assert!(inventory.sealed_partials.iter().all(|partial| {
819            partial.assignment_version == 0
820                && partial.payload_len == 1
821                && partial.payload_sha256.len() == 64
822        }));
823        let error = b
824            .seal_checkpoint(checkpoint, None, &[0, 1], &[])
825            .await
826            .unwrap_err();
827        assert!(matches!(error, StateBackendError::Conflict { .. }));
828        assert!(!b
829            .seal_checkpoint(attempt(2), None, &vnodes, &[])
830            .await
831            .unwrap());
832    }
833
834    #[tokio::test]
835    async fn out_of_range_vnode_errors() {
836        let b = InProcessBackend::new(2);
837        let r = b
838            .write_partial(attempt(1), 5, 0, Bytes::from_static(b"x"))
839            .await
840            .unwrap_err();
841        assert!(matches!(r, StateBackendError::Io(_)));
842    }
843
844    #[test]
845    fn state_backend_is_object_safe() {
846        let _: std::sync::Arc<dyn StateBackend> = std::sync::Arc::new(InProcessBackend::new(2));
847    }
848
849    #[tokio::test]
850    async fn prune_before_drops_old_epochs() {
851        let b = InProcessBackend::new(4);
852        for epoch in 1..=5 {
853            b.write_partial(attempt(epoch), 0, 0, Bytes::from_static(b"x"))
854                .await
855                .unwrap();
856            b.write_partial(attempt(epoch), 1, 0, Bytes::from_static(b"y"))
857                .await
858                .unwrap();
859        }
860        // Retain epochs >= 4. Entries for 1,2,3 must go away.
861        b.prune_before(4).await.unwrap();
862        for epoch in 1..=3 {
863            assert!(b.read_partial(attempt(epoch), 0).await.unwrap().is_none());
864            assert!(b.read_partial(attempt(epoch), 1).await.unwrap().is_none());
865        }
866        for epoch in 4..=5 {
867            assert!(b.read_partial(attempt(epoch), 0).await.unwrap().is_some());
868            assert!(b.read_partial(attempt(epoch), 1).await.unwrap().is_some());
869        }
870    }
871}
laminar_core/state/in_process.rs

laminar_core/state/
in_process.rs
