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        let put_conditionals = ConditionalHeaders {
145            if_match: Some(head_etag),
146            ..ConditionalHeaders::default()
147        };
148        let content_type = caller_content_type
149            .or(read.meta.content_type.as_deref())
150            .unwrap_or("application/octet-stream");
151        let new_size = bytes_len_u64(new_bytes.len())?;
152        let outcome = match storage
153            .put_object(kb, path, new_bytes, Some(content_type), put_conditionals)
154            .await
155        {
156            Ok(outcome) => outcome,
157            Err(StorageError::PreconditionFailed) if attempt < MAX_ATTEMPTS => {
158                tracing::debug!(target: "notedthat::patch", kb = %kb, path = %path, attempt, stage = "put", "PATCH_RETRY_PRECONDITION");
159                continue;
160            }
161            Err(error) => return Err(WriteError::Storage(error)),
162        };
163
164        crate::commit::after_write(sinks, kb, path, &outcome, new_size, content_type).await?;
165
166        return Ok(outcome);
167    }
168}
169
170pub(crate) fn splice_bytes(src: &Bytes, byte_range: std::ops::Range<usize>, body: &Bytes) -> Bytes {
171    let mut buf =
172        BytesMut::with_capacity(byte_range.start + body.len() + (src.len() - byte_range.end));
173    buf.extend_from_slice(&src[..byte_range.start]);
174    buf.extend_from_slice(body);
175    buf.extend_from_slice(&src[byte_range.end..]);
176    buf.freeze()
177}
178
179fn validate_caller_if_match(
180    patch_mode: &PatchMode,
181    caller_conditionals: &ConditionalHeaders,
182) -> Result<(), WriteError> {
183    match patch_mode {
184        PatchMode::Bytes { .. } | PatchMode::Lines { .. } => {
185            require_strong_if_match(caller_conditionals)?;
186        }
187        PatchMode::Append { .. } => {
188            if let Some(etag) = &caller_conditionals.if_match
189                && (etag == "*" || etag.contains(','))
190            {
191                return Err(WriteError::PatchInvalidRange {
192                    message: "If-Match: * and multi-value If-Match not supported on PATCH in v1"
193                        .into(),
194                });
195            }
196        }
197    }
198    Ok(())
199}
200
201/// Shared If-Match presence + shape validation for PATCH and POST /replace.
202pub(crate) fn require_strong_if_match(
203    caller_conditionals: &ConditionalHeaders,
204) -> Result<(), WriteError> {
205    let Some(etag) = &caller_conditionals.if_match else {
206        return Err(WriteError::PatchInvalidRange {
207            message: "If-Match required on PATCH (bytes/lines mode)".into(),
208        });
209    };
210    if etag == "*" || etag.contains(',') {
211        return Err(WriteError::PatchInvalidRange {
212            message: "If-Match: * and multi-value If-Match not supported on PATCH in v1".into(),
213        });
214    }
215    Ok(())
216}
217
218fn check_caller_precondition(
219    patch_mode: &PatchMode,
220    caller_conditionals: &ConditionalHeaders,
221    current_etag: Option<&str>,
222) -> Result<(), WriteError> {
223    let should_check_caller =
224        !matches!(patch_mode, PatchMode::Append { .. }) || caller_conditionals.if_match.is_some();
225    if should_check_caller
226        && let Some(caller_etag) = &caller_conditionals.if_match
227        && current_etag != Some(caller_etag.as_str())
228    {
229        return Err(WriteError::Storage(StorageError::PreconditionFailed));
230    }
231    Ok(())
232}
233
234fn splice_plan(
235    patch_mode: &PatchMode,
236    bytes: &Bytes,
237    read_len: u64,
238) -> Result<(Option<std::ops::Range<u64>>, u64), WriteError> {
239    match patch_mode {
240        PatchMode::Bytes { range, body } => {
241            let br = range.to_exclusive_range(read_len).ok_or_else(|| {
242                WriteError::PatchInvalidRange {
243                    message: format!("byte range unsatisfiable at size {}", bytes.len()),
244                }
245            })?;
246            let new_len = spliced_len(read_len, br.end - br.start, body.len(), "byte-range")?;
247            Ok((Some(br), new_len))
248        }
249        PatchMode::Lines { range, body } => {
250            let idx = LineIndex::from_bytes(bytes);
251            let br = idx.byte_range(range).ok_or_else(|| {
252                let (first, last) = line_range_bounds(range);
253                WriteError::PatchLineOutOfRange {
254                    first,
255                    last,
256                    total_lines: idx.total_lines,
257                    total_bytes: idx.total_bytes,
258                }
259            })?;
260            let new_len = spliced_len(read_len, br.end - br.start, body.len(), "line-range")?;
261            Ok((Some(br), new_len))
262        }
263        PatchMode::Append { body } => {
264            let body_len = bytes_len_u64(body.len())?;
265            let new_len =
266                read_len
267                    .checked_add(body_len)
268                    .ok_or_else(|| WriteError::PatchInvalidRange {
269                        message: "append length overflows u64".into(),
270                    })?;
271            Ok((None, new_len))
272        }
273    }
274}
275
276fn spliced_len(
277    read_len: u64,
278    replaced: u64,
279    body_len: usize,
280    mode: &str,
281) -> Result<u64, WriteError> {
282    read_len
283        .checked_sub(replaced)
284        .and_then(|n| n.checked_add(bytes_len_u64(body_len).ok()?))
285        .ok_or_else(|| WriteError::PatchInvalidRange {
286            message: format!("{mode} splice length overflows u64"),
287        })
288}
289
290fn bytes_len_u64(len: usize) -> Result<u64, WriteError> {
291    u64::try_from(len).map_err(|_| WriteError::PatchInvalidRange {
292        message: "buffer length does not fit u64".into(),
293    })
294}
295
296fn capacity_from_u64(len: u64) -> Result<usize, WriteError> {
297    usize::try_from(len).map_err(|_| WriteError::PatchInvalidRange {
298        message: "patched length does not fit usize".into(),
299    })
300}
301
302fn line_range_bounds(range: &LineRange) -> (u64, u64) {
303    match range {
304        LineRange::FromStart { first, last } => (*first, *last),
305        LineRange::FromStartOpen { first } => (*first, u64::MAX),
306        LineRange::Suffix { length } => (0, *length),
307        LineRange::Insert { before } => (*before, before.saturating_sub(1)),
308    }
309}
310
311#[cfg(test)]
312mod tests {
313    mod retry;
314    pub mod skeleton;
315    mod support;
316}
notedthat_write/patch.rs

notedthat_write/
patch.rs
