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

notedthat_write/
patch.rs
