icechunk_storage/
readback.rs

1//! Lost-response recovery for conditional PUTs, shared across storage backends.
2//!
3//! A conditional PUT can land while its ack is lost; the client's retry then
4//! trips the condition against the object we just wrote, faking a conflict.
5//! Each conditional PUT stamps a unique [`WRITE_ID_METADATA_KEY`]; on conflict
6//! the backend HEADs the object, classifies the result with
7//! [`ReadbackOutcome::classify`], then a `resolve_*` fn maps the outcome to a
8//! [`VersionedUpdateResult`].
9
10use std::sync::Once;
11
12use tracing::{trace, warn};
13
14use crate::storage::{
15    Settings, StorageError, StorageResult, VersionInfo, VersionedUpdateResult,
16    other_error,
17};
18
19/// Per-PUT token stamped on conditional writes. Letters only: Azure needs a valid
20/// C# identifier (no `-`), and nginx-fronted S3 gateways drop names that contain `_`.
21pub const WRITE_ID_METADATA_KEY: &str = "icechunkwriteid";
22
23/// The key conditional writes stamped before the rename to alphanumeric names.
24pub const WRITE_ID_METADATA_KEY_DEPRECATED: &str = "icechunk_write_id";
25
26static CONDITIONAL_WITHOUT_METADATA_WARNED: Once = Once::new();
27
28/// Warn once: conditional writes on but metadata off → no lost-response recovery.
29fn warn_conditional_without_metadata_once() {
30    CONDITIONAL_WITHOUT_METADATA_WARNED.call_once(|| {
31        warn!(
32            "conditional PUT is enabled but `unsafe_use_metadata` is \
33             disabled — lost-response recovery for conditional writes \
34             requires user metadata to stamp write-ids; without it, \
35             transient PUT failures may surface as spurious conflicts \
36             even when the write actually landed. See \
37             icechunk_storage::Settings::unsafe_use_metadata."
38        );
39    });
40}
41
42/// Write-id for a conditional PUT: `None` for unconditional PUTs, `None`
43/// (with a one-time warning) when metadata is off.
44pub fn write_id_for(
45    settings: &Settings,
46    conditional: bool,
47    mk: impl FnOnce() -> String,
48) -> Option<String> {
49    if !conditional {
50        None
51    } else if settings.unsafe_use_metadata() {
52        Some(mk())
53    } else {
54        warn_conditional_without_metadata_once();
55        None
56    }
57}
58
59/// `OurWrite` is universally success; the rest are mapped per-caller.
60#[derive(Debug, PartialEq, Eq)]
61pub enum ReadbackOutcome {
62    OurWrite(VersionInfo),
63    /// The landed object isn't (provably) ours: a different write-id, no
64    /// write-id, or no object at all. All map identically per-caller.
65    NotOurs,
66    /// Our id matched but the object has no usable version identity.
67    MissingVersion,
68}
69
70impl ReadbackOutcome {
71    /// Classify a successful read-back of a found object. A failed read-back is
72    /// inconclusive and never reaches here — the backend returns the HEAD error
73    /// instead; an absent object is conclusively `NotOurs` at the backend.
74    pub fn classify(
75        our: &str,
76        stored_write_id: Option<&str>,
77        version: &VersionInfo,
78    ) -> ReadbackOutcome {
79        if stored_write_id != Some(our) {
80            ReadbackOutcome::NotOurs
81        } else if version.is_create() {
82            ReadbackOutcome::MissingVersion
83        } else {
84            ReadbackOutcome::OurWrite(version.clone())
85        }
86    }
87}
88
89/// Resolve a precondition (412/409) conflict. Absent → genuine race
90/// (`NotOnLatestVersion`); inconclusive read-back (`Err`) propagates — faking a
91/// conflict would reintroduce the spurious-conflict bug.
92pub fn resolve_precondition(
93    readback: StorageResult<ReadbackOutcome>,
94    key: &str,
95) -> StorageResult<VersionedUpdateResult> {
96    match readback? {
97        ReadbackOutcome::OurWrite(new_version) => {
98            trace!(
99                key,
100                "precondition failed but our write-id is stored; retried PUT, success"
101            );
102            Ok(VersionedUpdateResult::Updated { new_version })
103        }
104        ReadbackOutcome::NotOurs => Ok(VersionedUpdateResult::NotOnLatestVersion),
105        ReadbackOutcome::MissingVersion => {
106            Err(other_error("readback object is missing a version identity"))
107        }
108    }
109}
110
111/// Resolve a failure only our own landed write can rescue (S3 multipart
112/// `NoSuchUpload`/404, or an `object_store` generic transport error). Anything
113/// but `OurWrite` propagates `original`; a failed read-back propagates its error.
114pub fn resolve_lost_response(
115    readback: StorageResult<ReadbackOutcome>,
116    key: &str,
117    original: StorageError,
118) -> StorageResult<VersionedUpdateResult> {
119    match readback? {
120        ReadbackOutcome::OurWrite(new_version) => {
121            trace!(
122                key,
123                "PUT response lost but our write-id is stored; treating as success"
124            );
125            Ok(VersionedUpdateResult::Updated { new_version })
126        }
127        _ => Err(original),
128    }
129}
130
131#[cfg(test)]
132mod tests {
133    use super::*;
134    use crate::storage::{ETag, StorageErrorKind};
135
136    fn version(etag: Option<&str>) -> VersionInfo {
137        VersionInfo { etag: etag.map(|e| ETag(e.to_string())), generation: None }
138    }
139
140    #[test]
141    fn classify_match_returns_readback_version() {
142        // A match returns the read-back object's own version, not the previous one.
143        assert_eq!(
144            ReadbackOutcome::classify("W", Some("W"), &version(Some("E1"))),
145            ReadbackOutcome::OurWrite(VersionInfo::from_etag_only("E1".to_string()))
146        );
147    }
148
149    #[test]
150    fn classify_branches() {
151        // A different write-id or no write-id collapse to NotOurs.
152        assert_eq!(
153            ReadbackOutcome::classify("W", Some("OTHER"), &version(Some("E1"))),
154            ReadbackOutcome::NotOurs
155        );
156        assert_eq!(
157            ReadbackOutcome::classify("W", None, &version(Some("E1"))),
158            ReadbackOutcome::NotOurs
159        );
160        // Our id matched but no version identity.
161        assert_eq!(
162            ReadbackOutcome::classify("W", Some("W"), &version(None)),
163            ReadbackOutcome::MissingVersion
164        );
165    }
166
167    #[test]
168    fn precondition_absent_is_conflict_not_error() {
169        // Regression: absent is a genuine race → NotOnLatestVersion, not an error.
170        assert_eq!(
171            resolve_precondition(Ok(ReadbackOutcome::NotOurs), "k").unwrap(),
172            VersionedUpdateResult::NotOnLatestVersion
173        );
174    }
175
176    #[test]
177    fn precondition_inconclusive_readback_propagates() {
178        // Regression: inconclusive read-back must Err, never fake a conflict —
179        // else a lost-response write retries into a spurious conflict.
180        let err = resolve_precondition(Err(other_error("head failed")), "k")
181            .expect_err("inconclusive readback must propagate");
182        assert!(matches!(err.kind, StorageErrorKind::Other(_)));
183    }
184
185    #[test]
186    fn lost_response_inconclusive_propagates_head_error() {
187        let err = resolve_lost_response(
188            Err(other_error("head failed")),
189            "k",
190            other_error("original put error"),
191        )
192        .expect_err("inconclusive readback must propagate");
193        assert!(matches!(err.kind, StorageErrorKind::Other(s) if s == "head failed"));
194    }
195
196    #[test]
197    fn lost_response_non_ours_propagates_original() {
198        let err = resolve_lost_response(
199            Ok(ReadbackOutcome::NotOurs),
200            "k",
201            other_error("original put error"),
202        )
203        .expect_err("a write that did not land must propagate the original error");
204        assert!(
205            matches!(err.kind, StorageErrorKind::Other(s) if s == "original put error")
206        );
207    }
208
209    #[test]
210    fn precondition_missing_version_is_error() {
211        // Our write landed but has no version identity to hand back.
212        let err = resolve_precondition(Ok(ReadbackOutcome::MissingVersion), "k")
213            .expect_err("a landed write without a version identity cannot succeed");
214        assert!(matches!(err.kind, StorageErrorKind::Other(_)));
215    }
216
217    #[test]
218    fn lost_response_our_write_recovers_readback_version() {
219        assert_eq!(
220            resolve_lost_response(
221                Ok(ReadbackOutcome::OurWrite(version(Some("E1")))),
222                "k",
223                other_error("original put error"),
224            )
225            .unwrap(),
226            VersionedUpdateResult::Updated {
227                new_version: VersionInfo::from_etag_only("E1".to_string())
228            }
229        );
230    }
231
232    #[test]
233    fn write_id_metadata_key_is_portable() {
234        assert!(
235            WRITE_ID_METADATA_KEY.starts_with(|c: char| c.is_ascii_alphabetic())
236                && WRITE_ID_METADATA_KEY.chars().all(|c| c.is_ascii_alphanumeric()),
237            "Azure rejects `-` and nginx-fronted S3 gateways drop `_`"
238        );
239    }
240}
icechunk_storage/readback.rs

icechunk_storage/
readback.rs
