notedthat_write/
patch.rs

1//! Shared PATCH splice primitive for object writes.
2
3use bytes::{Bytes, BytesMut};
4use notedthat_core::{
5    ByteRange, ConditionalHeaders, KbSlug, LineIndex, LineRange, ObjectPath, PutOutcome, Storage,
6    StorageError,
7};
8
9use crate::WriteError;
10use crate::sinks::WriteSinks;
11
12/// Specifies how an object's bytes should be spliced.
13#[derive(Debug, Clone)]
14pub enum PatchMode {
15    /// Replace a byte range.
16    Bytes {
17        /// Byte range to replace.
18        range: ByteRange,
19        /// Replacement bytes.
20        body: Bytes,
21    },
22    /// Replace a line range.
23    Lines {
24        /// Line range to replace or insertion point.
25        range: LineRange,
26        /// Replacement bytes.
27        body: Bytes,
28    },
29    /// Append to the end.
30    Append {
31        /// Bytes to append.
32        body: Bytes,
33    },
34}
35
36/// Request data for one optimistic PATCH operation.
37pub struct PatchRequest<'a> {
38    /// Knowledge base containing the object.
39    pub kb: &'a KbSlug,
40    /// Object path to patch.
41    pub path: &'a ObjectPath,
42    /// Patch splice mode.
43    pub patch_mode: PatchMode,
44    /// Caller-supplied conditional headers.
45    pub caller_conditionals: ConditionalHeaders,
46    /// Maximum patchable object size in bytes.
47    pub max_patchable_size: u64,
48    /// Caller-supplied content type, if any.
49    pub caller_content_type: Option<&'a str>,
50}
51
52/// Apply one optimistic PATCH attempt using the current HEAD `ETag` as the internal CAS anchor.
53///
54/// # Errors
55/// Returns [`WriteError`] when caller preconditions fail, the requested splice is invalid, the
56/// object is too large to patch, storage rejects the internal CAS, or the indexer queue is full.
57pub async fn patch(
58    storage: &dyn Storage,
59    sinks: &WriteSinks<'_>,
60    request: PatchRequest<'_>,
61) -> Result<PutOutcome, WriteError> {
62    const MAX_ATTEMPTS: u32 = 3;
63    let PatchRequest {
64        kb,
65        path,
66        patch_mode,
67        caller_conditionals,
68        max_patchable_size,
69        caller_content_type,
70    } = request;
71
72    validate_caller_if_match(&patch_mode, &caller_conditionals)?;
73
74    let mut attempt = 0u32;
75    loop {
76        attempt += 1;
77
78        let meta = storage
79            .head_object(kb, path, ConditionalHeaders::default())
80            .await?;
81        check_caller_precondition(&patch_mode, &caller_conditionals, meta.etag.as_deref())?;
82
83        if meta.size > max_patchable_size {
84            return Err(WriteError::PatchTooLarge {
85                size: meta.size,
86                limit: max_patchable_size,
87            });
88        }
89
90        let head_etag = meta
91            .etag
92            .clone()
93            .ok_or_else(|| WriteError::PatchInvalidRange {
94                message: "backend did not return ETag on HEAD".into(),
95            })?;
96
97        let get_conditionals = ConditionalHeaders {
98            if_match: Some(head_etag.clone()),
99            ..ConditionalHeaders::default()
100        };
101        let read = match storage.get_object(kb, path, None, get_conditionals).await {
102            Ok(read) => read,
103            Err(StorageError::PreconditionFailed) if attempt < MAX_ATTEMPTS => {
104                tracing::debug!(target: "notedthat::patch", kb = %kb, path = %path, attempt, stage = "get", "PATCH_RETRY_PRECONDITION");
105                continue;
106            }
107            Err(error) => return Err(WriteError::Storage(error)),
108        };
109        let read_len = bytes_len_u64(read.bytes.len())?;
110        let (byte_range, new_len) = splice_plan(&patch_mode, &read.bytes, read_len)?;
111
112        if new_len > max_patchable_size {
113            return Err(WriteError::PatchTooLarge {
114                size: new_len,
115                limit: max_patchable_size,
116            });
117        }
118
119        let new_bytes = match (&patch_mode, byte_range) {
120            (PatchMode::Bytes { body, .. } | PatchMode::Lines { body, .. }, Some(br)) => {
121                let start =
122                    usize::try_from(br.start).map_err(|_| WriteError::PatchInvalidRange {
123                        message: "splice start does not fit usize".into(),
124                    })?;
125                let end = usize::try_from(br.end).map_err(|_| WriteError::PatchInvalidRange {
126                    message: "splice end does not fit usize".into(),
127                })?;
128                splice_bytes(&read.bytes, start..end, body)
129            }
130            (PatchMode::Append { body }, None) => {
131                let mut buf = BytesMut::with_capacity(capacity_from_u64(new_len)?);
132                buf.extend_from_slice(&read.bytes);
133                buf.extend_from_slice(body);
134                buf.freeze()
135            }
136            (PatchMode::Bytes { .. } | PatchMode::Lines { .. }, None)
137            | (PatchMode::Append { .. }, Some(_)) => {
138                return Err(WriteError::PatchInvalidRange {
139                    message: "internal patch mode/range contradiction".into(),
140                });
141            }
142        };
143
144        crate::manifest::check_manifest_bytes(kb, path, &new_bytes)?;
145        let put_conditionals = ConditionalHeaders {
146            if_match: Some(head_etag),
147            ..ConditionalHeaders::default()
148        };
149        let content_type = caller_content_type
150            .or(read.meta.content_type.as_deref())
151            .unwrap_or("application/octet-stream");
152        let new_size = bytes_len_u64(new_bytes.len())?;
153        let outcome = match storage
154            .put_object(kb, path, new_bytes, Some(content_type), put_conditionals)
155            .await
156        {
157            Ok(outcome) => outcome,
158            Err(StorageError::PreconditionFailed) if attempt < MAX_ATTEMPTS => {
159                tracing::debug!(target: "notedthat::patch", kb = %kb, path = %path, attempt, stage = "put", "PATCH_RETRY_PRECONDITION");
160                continue;
161            }
162            Err(error) => return Err(WriteError::Storage(error)),
163        };
164
165        crate::commit::after_write(sinks, kb, path, &outcome, new_size, content_type).await?;
166
167        return Ok(outcome);
168    }
169}
170
171pub(crate) fn splice_bytes(src: &Bytes, byte_range: std::ops::Range<usize>, body: &Bytes) -> Bytes {
172    let mut buf =
173        BytesMut::with_capacity(byte_range.start + body.len() + (src.len() - byte_range.end));
174    buf.extend_from_slice(&src[..byte_range.start]);
175    buf.extend_from_slice(body);
176    buf.extend_from_slice(&src[byte_range.end..]);
177    buf.freeze()
178}
179
180fn validate_caller_if_match(
181    patch_mode: &PatchMode,
182    caller_conditionals: &ConditionalHeaders,
183) -> Result<(), WriteError> {
184    match patch_mode {
185        PatchMode::Bytes { .. } | PatchMode::Lines { .. } => {
186            require_strong_if_match(caller_conditionals)?;
187        }
188        PatchMode::Append { .. } => {
189            if let Some(etag) = &caller_conditionals.if_match
190                && (etag == "*" || etag.contains(','))
191            {
192                return Err(WriteError::PatchInvalidRange {
193                    message: "If-Match: * and multi-value If-Match not supported on PATCH in v1"
194                        .into(),
195                });
196            }
197        }
198    }
199    Ok(())
200}
201
202/// Shared If-Match presence + shape validation for PATCH and POST /replace.
203pub(crate) fn require_strong_if_match(
204    caller_conditionals: &ConditionalHeaders,
205) -> Result<(), WriteError> {
206    let Some(etag) = &caller_conditionals.if_match else {
207        return Err(WriteError::PatchInvalidRange {
208            message: "If-Match required on PATCH (bytes/lines mode)".into(),
209        });
210    };
211    if etag == "*" || etag.contains(',') {
212        return Err(WriteError::PatchInvalidRange {
213            message: "If-Match: * and multi-value If-Match not supported on PATCH in v1".into(),
214        });
215    }
216    Ok(())
217}
218
219fn check_caller_precondition(
220    patch_mode: &PatchMode,
221    caller_conditionals: &ConditionalHeaders,
222    current_etag: Option<&str>,
223) -> Result<(), WriteError> {
224    let should_check_caller =
225        !matches!(patch_mode, PatchMode::Append { .. }) || caller_conditionals.if_match.is_some();
226    if should_check_caller
227        && let Some(caller_etag) = &caller_conditionals.if_match
228        && current_etag != Some(caller_etag.as_str())
229    {
230        return Err(WriteError::Storage(StorageError::PreconditionFailed));
231    }
232    Ok(())
233}
234
235fn splice_plan(
236    patch_mode: &PatchMode,
237    bytes: &Bytes,
238    read_len: u64,
239) -> Result<(Option<std::ops::Range<u64>>, u64), WriteError> {
240    match patch_mode {
241        PatchMode::Bytes { range, body } => {
242            let br = range.to_exclusive_range(read_len).ok_or_else(|| {
243                WriteError::PatchInvalidRange {
244                    message: format!("byte range unsatisfiable at size {}", bytes.len()),
245                }
246            })?;
247            let new_len = spliced_len(read_len, br.end - br.start, body.len(), "byte-range")?;
248            Ok((Some(br), new_len))
249        }
250        PatchMode::Lines { range, body } => {
251            let idx = LineIndex::from_bytes(bytes);
252            let br = idx.byte_range(range).ok_or_else(|| {
253                let (first, last) = line_range_bounds(range);
254                WriteError::PatchLineOutOfRange {
255                    first,
256                    last,
257                    total_lines: idx.total_lines,
258                    total_bytes: idx.total_bytes,
259                }
260            })?;
261            let new_len = spliced_len(read_len, br.end - br.start, body.len(), "line-range")?;
262            Ok((Some(br), new_len))
263        }
264        PatchMode::Append { body } => {
265            let body_len = bytes_len_u64(body.len())?;
266            let new_len =
267                read_len
268                    .checked_add(body_len)
269                    .ok_or_else(|| WriteError::PatchInvalidRange {
270                        message: "append length overflows u64".into(),
271                    })?;
272            Ok((None, new_len))
273        }
274    }
275}
276
277fn spliced_len(
278    read_len: u64,
279    replaced: u64,
280    body_len: usize,
281    mode: &str,
282) -> Result<u64, WriteError> {
283    read_len
284        .checked_sub(replaced)
285        .and_then(|n| n.checked_add(bytes_len_u64(body_len).ok()?))
286        .ok_or_else(|| WriteError::PatchInvalidRange {
287            message: format!("{mode} splice length overflows u64"),
288        })
289}
290
291fn bytes_len_u64(len: usize) -> Result<u64, WriteError> {
292    u64::try_from(len).map_err(|_| WriteError::PatchInvalidRange {
293        message: "buffer length does not fit u64".into(),
294    })
295}
296
297fn capacity_from_u64(len: u64) -> Result<usize, WriteError> {
298    usize::try_from(len).map_err(|_| WriteError::PatchInvalidRange {
299        message: "patched length does not fit usize".into(),
300    })
301}
302
303fn line_range_bounds(range: &LineRange) -> (u64, u64) {
304    match range {
305        LineRange::FromStart { first, last } => (*first, *last),
306        LineRange::FromStartOpen { first } => (*first, u64::MAX),
307        LineRange::Suffix { length } => (0, *length),
308        LineRange::Insert { before } => (*before, before.saturating_sub(1)),
309    }
310}
311
312#[cfg(test)]
313mod tests {
314    mod retry;
315    pub mod skeleton;
316    mod support;
317}
notedthat_write/patch.rs

notedthat_write/
patch.rs
