laminar_core/checkpoint/checkpoint_store/
artifact_intent.rs

1//! Durable sink artifact evidence written before an epoch can create files.
2
3use bytes::Bytes;
4use object_store::PutPayload;
5
6use super::validation::validate_abort_seal_request;
7use super::{CheckpointStoreError, ObjectStoreCheckpointStore};
8use crate::checkpoint::canonical_json_bytes;
9use crate::checkpoint::checkpoint_manifest::{
10    checkpoint_artifact_intent_sha256, CheckpointManifest, StateChunkId,
11    PREPARED_SINK_DESCRIPTOR_VERSION,
12};
13
14/// Maximum connector artifact-intent payload accepted for one sink.
15pub const MAX_CHECKPOINT_SINK_ARTIFACT_INTENT_BYTES: usize = 64 * 1024;
16const MAX_CHECKPOINT_SINK_ARTIFACT_INTENTS: usize = 1_024;
17/// Maximum aggregate connector intent payload retained by one participant.
18pub const MAX_CHECKPOINT_SINK_ARTIFACT_INTENT_AGGREGATE_BYTES: usize = 1024 * 1024;
19pub(super) const MAX_CHECKPOINT_ARTIFACT_INTENT_RECORD_BYTES: u64 = 5 * 1024 * 1024;
20const CHECKPOINT_ARTIFACT_INTENT_VERSION: u32 = 1;
21
22/// One connector's bounded cleanup evidence persisted before `begin_epoch`.
23#[derive(Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
24#[serde(deny_unknown_fields)]
25pub struct CheckpointSinkArtifactIntent {
26    /// Stable registered sink name.
27    pub sink_name: String,
28    /// Runtime intent-envelope version.
29    pub format_version: u16,
30    /// Connector-specific cleanup evidence, or `None` when cleanup needs no payload.
31    pub payload: Option<Vec<u8>>,
32    /// Presence-domain-separated digest of `payload`.
33    pub sha256: String,
34}
35
36impl std::fmt::Debug for CheckpointSinkArtifactIntent {
37    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
38        formatter
39            .debug_struct("CheckpointSinkArtifactIntent")
40            .field("sink_name", &self.sink_name)
41            .field("format_version", &self.format_version)
42            .field("payload_bytes", &self.payload.as_ref().map_or(0, Vec::len))
43            .field("sha256", &self.sha256)
44            .finish()
45    }
46}
47
48impl CheckpointSinkArtifactIntent {
49    /// Construct one canonical runtime envelope.
50    ///
51    /// # Errors
52    /// Returns an error for an empty name or an oversized payload.
53    pub fn try_new(
54        sink_name: String,
55        payload: Option<Vec<u8>>,
56    ) -> Result<Self, CheckpointStoreError> {
57        if sink_name.is_empty() {
58            return Err(CheckpointStoreError::Invalid(
59                "sink artifact intent name must not be empty".into(),
60            ));
61        }
62        if payload
63            .as_ref()
64            .is_some_and(|payload| payload.len() > MAX_CHECKPOINT_SINK_ARTIFACT_INTENT_BYTES)
65        {
66            return Err(CheckpointStoreError::Invalid(format!(
67                "sink artifact intent exceeds {MAX_CHECKPOINT_SINK_ARTIFACT_INTENT_BYTES} bytes"
68            )));
69        }
70        let sha256 = checkpoint_artifact_intent_sha256(payload.as_deref());
71        Ok(Self {
72            sink_name,
73            format_version: PREPARED_SINK_DESCRIPTOR_VERSION,
74            payload,
75            sha256,
76        })
77    }
78}
79
80#[derive(Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
81#[serde(deny_unknown_fields)]
82pub(super) struct CheckpointArtifactIntentRecord {
83    version: u32,
84    artifact_identity_sha256: String,
85    chunk: StateChunkId,
86    sink_intents: Vec<CheckpointSinkArtifactIntent>,
87}
88
89impl CheckpointArtifactIntentRecord {
90    pub(super) fn sink_intents(&self) -> &[CheckpointSinkArtifactIntent] {
91        &self.sink_intents
92    }
93}
94
95pub(super) fn validate_artifact_intent_record(
96    record: &CheckpointArtifactIntentRecord,
97    expected_chunk: StateChunkId,
98    expected_artifact_identity_sha256: &str,
99) -> Result<(), CheckpointStoreError> {
100    validate_abort_seal_request(expected_chunk, expected_artifact_identity_sha256)?;
101    if record.version != CHECKPOINT_ARTIFACT_INTENT_VERSION
102        || record.chunk != expected_chunk
103        || record.artifact_identity_sha256 != expected_artifact_identity_sha256
104    {
105        return Err(CheckpointStoreError::Invalid(format!(
106            "participant {} checkpoint {} has a different sink artifact intent",
107            expected_chunk.participant_id, expected_chunk.checkpoint_id
108        )));
109    }
110    validate_sink_intents(&record.sink_intents)
111}
112
113pub(super) fn validate_sink_intents(
114    intents: &[CheckpointSinkArtifactIntent],
115) -> Result<(), CheckpointStoreError> {
116    if intents.is_empty() || intents.len() > MAX_CHECKPOINT_SINK_ARTIFACT_INTENTS {
117        return Err(CheckpointStoreError::Invalid(format!(
118            "sink artifact intent count must be in 1..={MAX_CHECKPOINT_SINK_ARTIFACT_INTENTS}"
119        )));
120    }
121    if !intents
122        .windows(2)
123        .all(|pair| pair[0].sink_name < pair[1].sink_name)
124    {
125        return Err(CheckpointStoreError::Invalid(
126            "sink artifact intents must be strictly ordered and unique".into(),
127        ));
128    }
129    let mut aggregate = 0_usize;
130    for intent in intents {
131        if intent.sink_name.is_empty()
132            || intent.format_version != PREPARED_SINK_DESCRIPTOR_VERSION
133            || intent.sha256 != checkpoint_artifact_intent_sha256(intent.payload.as_deref())
134        {
135            return Err(CheckpointStoreError::Invalid(format!(
136                "sink artifact intent '{}' has an invalid envelope",
137                intent.sink_name
138            )));
139        }
140        let bytes = intent.payload.as_ref().map_or(0, Vec::len);
141        if bytes > MAX_CHECKPOINT_SINK_ARTIFACT_INTENT_BYTES {
142            return Err(CheckpointStoreError::Invalid(format!(
143                "sink artifact intent '{}' exceeds {MAX_CHECKPOINT_SINK_ARTIFACT_INTENT_BYTES} bytes",
144                intent.sink_name
145            )));
146        }
147        aggregate = aggregate
148            .checked_add(bytes)
149            .ok_or_else(|| CheckpointStoreError::Invalid("sink intent byte overflow".into()))?;
150        if aggregate > MAX_CHECKPOINT_SINK_ARTIFACT_INTENT_AGGREGATE_BYTES {
151            return Err(CheckpointStoreError::Invalid(format!(
152                "sink artifact intents exceed {MAX_CHECKPOINT_SINK_ARTIFACT_INTENT_AGGREGATE_BYTES} aggregate bytes"
153            )));
154        }
155    }
156    Ok(())
157}
158
159fn artifact_intent_bytes(
160    record: &CheckpointArtifactIntentRecord,
161) -> Result<Bytes, CheckpointStoreError> {
162    let encoded = Bytes::from(canonical_json_bytes(record)?);
163    if u64::try_from(encoded.len()).unwrap_or(u64::MAX)
164        > MAX_CHECKPOINT_ARTIFACT_INTENT_RECORD_BYTES
165    {
166        return Err(CheckpointStoreError::Invalid(format!(
167            "checkpoint sink artifact intent record exceeds {MAX_CHECKPOINT_ARTIFACT_INTENT_RECORD_BYTES} bytes"
168        )));
169    }
170    Ok(encoded)
171}
172
173impl ObjectStoreCheckpointStore {
174    pub(super) fn decode_pending_artifact_intent_record(
175        bytes: &Bytes,
176        chunk: StateChunkId,
177    ) -> Result<Option<CheckpointArtifactIntentRecord>, CheckpointStoreError> {
178        let Ok(record) = serde_json::from_slice::<CheckpointArtifactIntentRecord>(bytes) else {
179            return Ok(None);
180        };
181        validate_artifact_intent_record(&record, chunk, &record.artifact_identity_sha256)?;
182        if artifact_intent_bytes(&record)? != *bytes {
183            return Err(CheckpointStoreError::Invalid(format!(
184                "participant {} checkpoint {} sink artifact intent is not canonical",
185                chunk.participant_id, chunk.checkpoint_id
186            )));
187        }
188        Ok(Some(record))
189    }
190
191    pub(super) fn decode_artifact_intent_record(
192        bytes: &Bytes,
193        chunk: StateChunkId,
194        expected_artifact_identity_sha256: &str,
195    ) -> Result<Option<CheckpointArtifactIntentRecord>, CheckpointStoreError> {
196        let Some(record) = Self::decode_pending_artifact_intent_record(bytes, chunk)? else {
197            return Ok(None);
198        };
199        if record.artifact_identity_sha256 != expected_artifact_identity_sha256 {
200            return Err(CheckpointStoreError::Invalid(format!(
201                "participant {} checkpoint {} has a different sink artifact intent",
202                chunk.participant_id, chunk.checkpoint_id
203            )));
204        }
205        Ok(Some(record))
206    }
207
208    pub(super) async fn save_artifact_intent_record(
209        &self,
210        chunk: StateChunkId,
211        artifact_identity_sha256: &str,
212        sink_intents: Vec<CheckpointSinkArtifactIntent>,
213    ) -> Result<(), CheckpointStoreError> {
214        validate_abort_seal_request(chunk, artifact_identity_sha256)?;
215        validate_sink_intents(&sink_intents)?;
216        let record = CheckpointArtifactIntentRecord {
217            version: CHECKPOINT_ARTIFACT_INTENT_VERSION,
218            artifact_identity_sha256: artifact_identity_sha256.to_owned(),
219            chunk,
220            sink_intents,
221        };
222        let encoded = artifact_intent_bytes(&record)?;
223        let path = self.manifest_path(chunk);
224        if self
225            .create_immutable(&path, PutPayload::from_bytes(encoded.clone()))
226            .await?
227        {
228            return Ok(());
229        }
230        let existing = self
231            .load_manifest_or_abort_seal_bytes(chunk)
232            .await?
233            .ok_or_else(|| {
234                CheckpointStoreError::Invalid(format!(
235                    "participant {} checkpoint {} sink intent create conflicted but no object exists",
236                    chunk.participant_id, chunk.checkpoint_id
237                ))
238            })?;
239        if existing == encoded {
240            Ok(())
241        } else {
242            Err(CheckpointStoreError::Invalid(format!(
243                "participant {} checkpoint {} already has different durable artifact state",
244                chunk.participant_id, chunk.checkpoint_id
245            )))
246        }
247    }
248
249    pub(super) async fn promote_artifact_intent_to_manifest(
250        &self,
251        manifest: &CheckpointManifest,
252        encoded_manifest: Bytes,
253        current: &Bytes,
254    ) -> Result<bool, CheckpointStoreError> {
255        let chunk = manifest.node_data.chunk;
256        let inventory = crate::checkpoint_decision::CheckpointArtifactInventory {
257            deployment_id: manifest.deployment_id.clone(),
258            pipeline_identity: manifest.pipeline_identity.clone(),
259            attempt: crate::checkpoint::CheckpointAttempt::new(
260                manifest.epoch,
261                manifest.checkpoint_id,
262            ),
263            assignment_fence: manifest.assignment_fence.clone(),
264            sink_artifact_intent_protocol: !manifest.sink_artifact_intents.is_empty(),
265        };
266        let identity = super::checkpoint_artifact_identity_sha256(&inventory, chunk)?;
267        let Some(record) = Self::decode_artifact_intent_record(current, chunk, &identity)? else {
268            return Ok(false);
269        };
270        validate_manifest_intents(manifest, record.sink_intents())?;
271        self.replace_exact(&self.manifest_path(chunk), current, encoded_manifest)
272            .await?;
273        Ok(true)
274    }
275}
276
277fn validate_manifest_intents(
278    manifest: &CheckpointManifest,
279    intents: &[CheckpointSinkArtifactIntent],
280) -> Result<(), CheckpointStoreError> {
281    if manifest.sink_artifact_intents.len() != intents.len() {
282        return Err(CheckpointStoreError::Invalid(
283            "checkpoint manifest does not retain its admitted sink artifact intents".into(),
284        ));
285    }
286    for (descriptor, intent) in manifest.sink_artifact_intents.iter().zip(intents) {
287        if descriptor.sink_name != intent.sink_name
288            || descriptor.format_version != intent.format_version
289            || descriptor.payload.is_some() != intent.payload.is_some()
290            || descriptor.sha256 != intent.sha256
291        {
292            return Err(CheckpointStoreError::Invalid(format!(
293                "checkpoint manifest changed sink artifact intent '{}'",
294                intent.sink_name
295            )));
296        }
297    }
298    Ok(())
299}
laminar_core/checkpoint/checkpoint_store/artifact_intent.rs

laminar_core/checkpoint/checkpoint_store/
artifact_intent.rs
