log_store/object_store_wal/
io.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Object store access for WAL objects: a deterministic key layout and
16//! conditional creates that make a retry of the same object a no-op.
17
18use bytes::Bytes;
19use object_store::{ErrorKind, ObjectStore};
20use snafu::{OptionExt, ResultExt, ensure};
21
22use crate::error::{
23    CorruptedWalObjectSnafu, InvalidWalObjectStoreSnafu, Result, WalObjectConflictSnafu,
24    WalObjectStoreSnafu,
25};
26
27/// Width of the zero-padded object sequence in an object key, wide enough for
28/// [`u64::MAX`] so that lexicographic and numeric order agree.
29const OBJECT_SEQ_WIDTH: usize = 20;
30const OBJECT_SUFFIX: &str = ".wal";
31
32/// Outcome of a conditional create.
33#[derive(Debug, Clone, Copy, PartialEq, Eq)]
34pub(crate) enum PutResult {
35    /// This call created the object.
36    Created,
37    /// The object already held the same content, so the call was a retry.
38    AlreadyPresent,
39}
40
41/// An object discovered by a LIST.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub(crate) struct ListedObject {
44    pub(crate) object_seq: u64,
45    pub(crate) path: String,
46    /// Length of the object in bytes, as reported by the listing.
47    pub(crate) size: u64,
48}
49
50/// Reads and writes the WAL objects under one prefix.
51pub(crate) struct ObjectStoreIo {
52    store: ObjectStore,
53    object_prefix: String,
54}
55
56impl ObjectStoreIo {
57    /// Binds `store` to the objects under `prefix`. The store must support
58    /// conditional creates, which is what keeps a retried write from
59    /// overwriting a durable object.
60    pub(crate) fn new(store: ObjectStore, prefix: impl AsRef<str>) -> Result<Self> {
61        let object_prefix = normalize_prefix(prefix.as_ref())?;
62        ensure!(
63            store.info().capability().write_with_if_not_exists,
64            InvalidWalObjectStoreSnafu {
65                reason: "object store does not support conditional create",
66            }
67        );
68        Ok(Self {
69            store,
70            object_prefix,
71        })
72    }
73
74    /// Creates the object `object_seq` unless it already exists. A retry with
75    /// the same content is accepted, different content is rejected.
76    ///
77    /// When the store reports that the object exists but the read that
78    /// compares its content fails, the read failure is returned, so that its
79    /// retry hint says whether comparing again may settle the outcome. Any
80    /// other create failure is returned as the write failure it is.
81    pub(crate) async fn put_if_absent(&self, object_seq: u64, content: Bytes) -> Result<PutResult> {
82        let path = self.object_path(object_seq);
83        let write_result = self
84            .store
85            .write_with(&path, content.clone())
86            .if_not_exists(true)
87            .await;
88
89        match write_result {
90            Ok(_) => Ok(PutResult::Created),
91            Err(error) => match self.store.read(&path).await {
92                Ok(existing) if existing.to_bytes() == content => Ok(PutResult::AlreadyPresent),
93                Ok(_) => WalObjectConflictSnafu { path }.fail(),
94                Err(read_error) if reports_existing_object(&error) => {
95                    Err(read_error).context(WalObjectStoreSnafu {
96                        operation: "read",
97                        path,
98                    })
99                }
100                Err(_) => Err(error).context(WalObjectStoreSnafu {
101                    operation: "write",
102                    path,
103                }),
104            },
105        }
106    }
107
108    /// Reads the object `object_seq`.
109    pub(crate) async fn get(&self, object_seq: u64) -> Result<Bytes> {
110        let path = self.object_path(object_seq);
111        self.store
112            .read(&path)
113            .await
114            .map(|content| content.to_bytes())
115            .context(WalObjectStoreSnafu {
116                operation: "read",
117                path,
118            })
119    }
120
121    /// Reads `len` bytes of the object `object_seq` starting at `offset`. The
122    /// range must lie inside the object; one that reaches past its end fails.
123    pub(crate) async fn get_range(&self, object_seq: u64, offset: u64, len: u64) -> Result<Bytes> {
124        let path = self.object_path(object_seq);
125        let end = offset
126            .checked_add(len)
127            .with_context(|| CorruptedWalObjectSnafu {
128                reason: format!("byte range {offset}..{len} overflows the object"),
129            })?;
130        self.store
131            .read_with(&path)
132            .range(offset..end)
133            .await
134            .map(|content| content.to_bytes())
135            .context(WalObjectStoreSnafu {
136                operation: "read",
137                path,
138            })
139    }
140
141    /// Lists the objects under the prefix, ordered by object sequence. Keys
142    /// that do not follow the object layout are ignored.
143    pub(crate) async fn list(&self) -> Result<Vec<ListedObject>> {
144        let entries = self
145            .store
146            .list(&self.object_prefix)
147            .await
148            .with_context(|_| WalObjectStoreSnafu {
149                operation: "list",
150                path: self.object_prefix.clone(),
151            })?;
152        let mut objects = entries
153            .into_iter()
154            .filter_map(|entry| {
155                self.parse_object_seq(entry.path())
156                    .map(|object_seq| ListedObject {
157                        object_seq,
158                        path: entry.path().to_string(),
159                        size: entry.metadata().content_length(),
160                    })
161            })
162            .collect::<Vec<_>>();
163        objects.sort_unstable_by_key(|object| object.object_seq);
164        Ok(objects)
165    }
166
167    pub(crate) fn object_path(&self, object_seq: u64) -> String {
168        format!(
169            "{}{object_seq:0OBJECT_SEQ_WIDTH$}{OBJECT_SUFFIX}",
170            self.object_prefix
171        )
172    }
173
174    fn parse_object_seq(&self, path: &str) -> Option<u64> {
175        let value = path
176            .strip_prefix(&self.object_prefix)?
177            .strip_suffix(OBJECT_SUFFIX)?;
178        if value.len() != OBJECT_SEQ_WIDTH || !value.bytes().all(|byte| byte.is_ascii_digit()) {
179            return None;
180        }
181        value.parse().ok()
182    }
183}
184
185/// Returns true when a conditional create failed because the object exists.
186/// Only the precondition failure says so: a store may report `AlreadyExists`
187/// for an unrelated path, such as a parent that is a file.
188fn reports_existing_object(error: &object_store::Error) -> bool {
189    error.kind() == ErrorKind::ConditionNotMatch
190}
191
192fn normalize_prefix(prefix: &str) -> Result<String> {
193    let prefix = prefix.trim();
194    ensure!(
195        !prefix.is_empty() && !prefix.starts_with('/'),
196        InvalidWalObjectStoreSnafu {
197            reason: format!("object prefix {prefix:?} is empty or absolute"),
198        }
199    );
200    let prefix = prefix.strip_suffix('/').unwrap_or(prefix);
201    ensure!(
202        !prefix
203            .split('/')
204            .any(|segment| segment.is_empty() || segment == "." || segment == ".."),
205        InvalidWalObjectStoreSnafu {
206            reason: format!("object prefix {prefix:?} has an empty or relative component"),
207        }
208    );
209    Ok(format!("{prefix}/objects/"))
210}
211
212#[cfg(test)]
213mod tests {
214    use std::sync::Arc;
215
216    use common_error::ext::{ErrorExt, RetryHint};
217    use common_test_util::temp_dir::create_temp_dir;
218    use object_store::layers::mock::{self, MockLayerBuilder, oio};
219    use object_store::secure_fs::SecureFsRoot;
220    use object_store::services::Memory;
221
222    use super::*;
223    use crate::error::Error;
224
225    fn memory_store() -> ObjectStore {
226        ObjectStore::new(Memory::default()).unwrap()
227    }
228
229    /// A reader whose every read fails with a temporary error.
230    struct FailingReader;
231
232    impl oio::Read for FailingReader {
233        async fn open(
234            &self,
235            _range: mock::BytesRange,
236        ) -> mock::Result<(mock::RpRead, Box<dyn oio::ReadStreamDyn>)> {
237            Err(injected_failure("read").set_temporary())
238        }
239
240        async fn read(
241            &self,
242            _range: mock::BytesRange,
243        ) -> mock::Result<(mock::RpRead, mock::Buffer)> {
244            Err(injected_failure("read").set_temporary())
245        }
246    }
247
248    /// A writer that fails to close, so nothing is written.
249    struct FailingWriter;
250
251    impl oio::Write for FailingWriter {
252        async fn write(&mut self, _buffer: mock::Buffer) -> mock::Result<()> {
253            Ok(())
254        }
255
256        async fn close(&mut self) -> mock::Result<mock::Metadata> {
257            Err(injected_failure("write"))
258        }
259
260        async fn abort(&mut self) -> mock::Result<()> {
261            Ok(())
262        }
263    }
264
265    fn injected_failure(operation: &str) -> mock::Error {
266        mock::Error::new(
267            mock::ErrorKind::Unexpected,
268            format!("injected {operation} failure"),
269        )
270    }
271
272    fn with_failing_reads(store: ObjectStore) -> ObjectStore {
273        store.layer(
274            MockLayerBuilder::default()
275                .reader_factory(Arc::new(|_, _, _| Box::new(FailingReader)))
276                .build()
277                .unwrap(),
278        )
279    }
280
281    fn with_failing_writes(store: ObjectStore) -> ObjectStore {
282        store.layer(
283            MockLayerBuilder::default()
284                .writer_factory(Arc::new(|_, _, _| Box::new(FailingWriter)))
285                .build()
286                .unwrap(),
287        )
288    }
289
290    fn memory_io() -> ObjectStoreIo {
291        ObjectStoreIo::new(memory_store(), "datanodes/1/epochs/2/").unwrap()
292    }
293
294    #[test]
295    fn test_io_builds_and_parses_deterministic_object_paths() {
296        let io = ObjectStoreIo::new(memory_store(), " datanodes/1/epochs/2/ ").unwrap();
297        assert_eq!(
298            "datanodes/1/epochs/2/objects/00000000000000000000.wal",
299            io.object_path(0)
300        );
301        assert_eq!(
302            "datanodes/1/epochs/2/objects/00000000000000000042.wal",
303            io.object_path(42)
304        );
305        assert_eq!(
306            "datanodes/1/epochs/2/objects/18446744073709551615.wal",
307            io.object_path(u64::MAX)
308        );
309        assert_eq!(Some(42), io.parse_object_seq(&io.object_path(42)));
310        assert_eq!(
311            None,
312            io.parse_object_seq("datanodes/1/epochs/2/objects/42.wal")
313        );
314    }
315
316    #[test]
317    fn test_io_rejects_invalid_prefixes() {
318        for prefix in [
319            "",
320            " ",
321            "/absolute",
322            ".",
323            "./node",
324            "node/./epoch",
325            "..",
326            "node/../epoch",
327            "node//epoch",
328            "node//",
329            "node///",
330        ] {
331            match ObjectStoreIo::new(memory_store(), prefix) {
332                Err(Error::InvalidWalObjectStore { .. }) => {}
333                Err(error) => panic!("unexpected error for prefix {prefix:?}: {error:?}"),
334                Ok(_) => panic!("expected prefix {prefix:?} to be rejected"),
335            }
336        }
337    }
338
339    #[tokio::test]
340    async fn test_io_puts_gets_and_lists_objects_by_sequence() {
341        let io = memory_io();
342
343        for object_seq in [10, 2, 1] {
344            assert_eq!(
345                PutResult::Created,
346                io.put_if_absent(object_seq, Bytes::from(object_seq.to_string()))
347                    .await
348                    .unwrap()
349            );
350        }
351
352        assert_eq!(Bytes::from_static(b"2"), io.get(2).await.unwrap());
353        let objects = io.list().await.unwrap();
354        assert_eq!(vec![1, 2, 10], object_seqs(objects.clone()));
355        assert_eq!(io.object_path(1), objects[0].path);
356        assert_eq!(
357            vec![1, 1, 2],
358            objects.iter().map(|object| object.size).collect::<Vec<_>>()
359        );
360    }
361
362    #[tokio::test]
363    async fn test_io_get_range_reads_a_slice_inside_the_object() {
364        let io = memory_io();
365        io.put_if_absent(3, Bytes::from_static(b"0123456789"))
366            .await
367            .unwrap();
368
369        assert_eq!(
370            Bytes::from_static(b"234"),
371            io.get_range(3, 2, 3).await.unwrap()
372        );
373        assert_eq!(
374            Bytes::from_static(b"89"),
375            io.get_range(3, 8, 2).await.unwrap()
376        );
377        for (object_seq, offset, len) in [(3, 8, 5), (3, 10, 1), (7, 0, 1)] {
378            let error = io.get_range(object_seq, offset, len).await.unwrap_err();
379            assert!(
380                matches!(error, Error::WalObjectStore { operation: "read", ref path, .. } if path == &io.object_path(object_seq)),
381                "unexpected error for range {offset}..{len} of object {object_seq}: {error:?}"
382            );
383        }
384        let error = io.get_range(3, u64::MAX, 1).await.unwrap_err();
385        assert!(
386            matches!(error, Error::CorruptedWalObject { .. }),
387            "unexpected error: {error:?}"
388        );
389    }
390
391    #[tokio::test]
392    async fn test_io_same_content_is_an_idempotent_retry() {
393        let io = memory_io();
394        let content = Bytes::from_static(b"immutable");
395
396        assert_eq!(
397            PutResult::Created,
398            io.put_if_absent(7, content.clone()).await.unwrap()
399        );
400        assert_eq!(
401            PutResult::AlreadyPresent,
402            io.put_if_absent(7, content.clone()).await.unwrap()
403        );
404        assert_eq!(content, io.get(7).await.unwrap());
405    }
406
407    #[tokio::test]
408    async fn test_io_rejects_different_content_without_overwriting() {
409        let io = memory_io();
410        let original = Bytes::from_static(b"original");
411        io.put_if_absent(7, original.clone()).await.unwrap();
412
413        let error = io
414            .put_if_absent(7, Bytes::from_static(b"replacement"))
415            .await
416            .unwrap_err();
417
418        assert!(
419            matches!(error, Error::WalObjectConflict { ref path, .. } if path == &io.object_path(7)),
420            "unexpected error: {error:?}"
421        );
422        assert_eq!(original, io.get(7).await.unwrap());
423    }
424
425    #[tokio::test]
426    async fn test_io_retry_of_an_existing_object_reports_the_failed_read() {
427        let store = memory_store();
428        let prefix = "datanodes/1/epochs/2/";
429        let content = Bytes::from_static(b"immutable");
430        ObjectStoreIo::new(store.clone(), prefix)
431            .unwrap()
432            .put_if_absent(7, content.clone())
433            .await
434            .unwrap();
435
436        // The object exists, so the create is refused, and the read that would
437        // settle the retry fails: that failure and its retry hint come back.
438        let io = ObjectStoreIo::new(with_failing_reads(store.clone()), prefix).unwrap();
439        let error = io.put_if_absent(7, content.clone()).await.unwrap_err();
440        assert!(
441            matches!(error, Error::WalObjectStore { operation: "read", ref path, .. } if path == &io.object_path(7)),
442            "unexpected error: {error:?}"
443        );
444        assert_eq!(RetryHint::Retryable, error.retry_hint());
445
446        // A create that fails for another reason is reported as the write
447        // failure it is, although the read of the missing object fails too.
448        let io = ObjectStoreIo::new(with_failing_writes(store), prefix).unwrap();
449        let error = io.put_if_absent(8, content).await.unwrap_err();
450        assert!(
451            matches!(error, Error::WalObjectStore { operation: "write", ref path, .. } if path == &io.object_path(8)),
452            "unexpected error: {error:?}"
453        );
454        assert_eq!(RetryHint::NonRetryable, error.retry_hint());
455    }
456
457    #[tokio::test]
458    async fn test_io_create_under_a_file_parent_is_reported_as_the_write_failure() {
459        let temp_dir = create_temp_dir("object_store_wal_io_file_parent");
460        std::fs::write(temp_dir.path().join("datanodes"), []).unwrap();
461        let store = SecureFsRoot::open(temp_dir.path())
462            .unwrap()
463            .build_operator();
464        let io = ObjectStoreIo::new(store, "datanodes/1/epochs/2/").unwrap();
465
466        // The store reports `AlreadyExists` for the parent, not for the
467        // object, so the create failure itself comes back.
468        let error = io
469            .put_if_absent(7, Bytes::from_static(b"wal"))
470            .await
471            .unwrap_err();
472        assert!(
473            matches!(error, Error::WalObjectStore { operation: "write", ref path, .. } if path == &io.object_path(7)),
474            "unexpected error: {error:?}"
475        );
476    }
477
478    #[tokio::test]
479    async fn test_io_get_of_a_missing_object_fails() {
480        let io = memory_io();
481
482        let error = io.get(7).await.unwrap_err();
483        assert!(
484            matches!(error, Error::WalObjectStore { operation: "read", ref path, .. } if path == &io.object_path(7)),
485            "unexpected error: {error:?}"
486        );
487    }
488
489    #[tokio::test]
490    async fn test_io_list_ignores_objects_outside_the_wal_layout() {
491        let io = memory_io();
492        io.store.write("unrelated", "outside").await.unwrap();
493        io.store
494            .write(
495                "datanodes/1/epochs/2/objects/not-a-sequence.wal",
496                "malformed",
497            )
498            .await
499            .unwrap();
500        io.put_if_absent(3, Bytes::from_static(b"wal"))
501            .await
502            .unwrap();
503
504        assert_eq!(vec![3], object_seqs(io.list().await.unwrap()));
505    }
506
507    #[tokio::test]
508    async fn test_io_list_is_isolated_by_prefix() {
509        let store = memory_store();
510        let first = ObjectStoreIo::new(store.clone(), "datanodes/1/epochs/1").unwrap();
511        let second = ObjectStoreIo::new(store, "datanodes/1/epochs/2").unwrap();
512        first
513            .put_if_absent(1, Bytes::from_static(b"first"))
514            .await
515            .unwrap();
516        second
517            .put_if_absent(2, Bytes::from_static(b"second"))
518            .await
519            .unwrap();
520
521        assert_eq!(vec![1], object_seqs(first.list().await.unwrap()));
522        assert_eq!(vec![2], object_seqs(second.list().await.unwrap()));
523    }
524
525    #[tokio::test]
526    async fn test_io_equal_sequences_in_different_prefixes_do_not_conflict() {
527        let store = memory_store();
528        let first = ObjectStoreIo::new(store.clone(), "datanodes/1/epochs/1").unwrap();
529        let second = ObjectStoreIo::new(store, "datanodes/1/epochs/2").unwrap();
530
531        assert_eq!(
532            PutResult::Created,
533            first
534                .put_if_absent(7, Bytes::from_static(b"first"))
535                .await
536                .unwrap()
537        );
538        assert_eq!(
539            PutResult::Created,
540            second
541                .put_if_absent(7, Bytes::from_static(b"different"))
542                .await
543                .unwrap()
544        );
545        assert_eq!(Bytes::from_static(b"first"), first.get(7).await.unwrap());
546        assert_eq!(
547            Bytes::from_static(b"different"),
548            second.get(7).await.unwrap()
549        );
550    }
551
552    fn object_seqs(objects: Vec<ListedObject>) -> Vec<u64> {
553        objects
554            .into_iter()
555            .map(|object| object.object_seq)
556            .collect()
557    }
558}
log_store/object_store_wal/io.rs

log_store/object_store_wal/
io.rs
