krabka_object_store/
fault.rs

1//! A fault-injecting [`ObjectStore`] wrapper, for drills against a store that
2//! misbehaves.
3//!
4//! Every production bound in this crate -- the retry budget, the retry
5//! deadline, the request and connect timeouts -- exists because a real object
6//! store throttles, stalls, and fails. None of that is reachable from a test
7//! that only ever meets a store which answers. [`FaultInjectingStore`] wraps
8//! any other store and, per operation, delays the call and then fails it with
9//! the error the backend would have produced.
10//!
11//! The failure schedule is deterministic, not random: a rate of 500 per
12//! thousand fails every second call to that operation, always the same ones,
13//! so a suite that asserts on which call failed asserts on a fact rather than
14//! on a seed.
15//!
16//! This is a testing seam. Nothing in the broker's serving path constructs it.
17//! Tests can wrap an existing store directly or use [`build_faulty_object_store`]
18//! to hand a fault policy to code that takes an `Arc<dyn ObjectStore>`.
19
20use {
21    crate::{build::build_object_store, config::ObjectStoreConfig, error::ObjectStoreError},
22    bytes::Bytes,
23    futures_util::{
24        stream,
25        stream::{BoxStream, StreamExt as _},
26    },
27    object_store::{
28        CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore,
29        PutMultipartOptions, PutOptions, PutPayload, PutResult, memory::InMemory, path::Path,
30    },
31    std::{
32        collections::BTreeMap,
33        fmt,
34        sync::{
35            Arc,
36            atomic::{AtomicU64, Ordering},
37        },
38        time::Duration,
39    },
40    tokio::time,
41};
42
43/// The operation classes a [`FaultPolicy`] can single out.
44///
45/// The granularity is the one an operator's incident is described in -- "puts
46/// are being throttled", "listing is slow" -- rather than one variant per
47/// trait method, so a policy written for `Put` covers the multipart path too.
48///
49/// `ALL` lists every variant, so a policy can be applied across the board and
50/// a table test can walk them. `label` is the `store` field of the errors an
51/// op's faults produce.
52#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, krabka_macros::EnumStr)]
53#[enum_str(as_str = label, all)]
54pub enum StoreOp {
55    /// `put_opts` and `put_multipart_opts`.
56    #[enum_str(name = "FaultInjectingStore::put")]
57    Put,
58    /// `get_opts` and everything layered on it, `head` included.
59    #[enum_str(name = "FaultInjectingStore::get")]
60    Get,
61    /// `delete_stream`.
62    #[enum_str(name = "FaultInjectingStore::delete")]
63    Delete,
64    /// `list` and `list_with_delimiter`.
65    #[enum_str(name = "FaultInjectingStore::list")]
66    List,
67    /// `copy_opts`.
68    #[enum_str(name = "FaultInjectingStore::copy")]
69    Copy,
70}
71
72/// What a failing call fails with.
73#[derive(Debug, Clone, Copy, PartialEq, Eq)]
74pub enum FaultKind {
75    /// `503 SlowDown`, the answer S3 gives a client it is rate-limiting. This
76    /// is the one that matters: it is retryable, so it is the shape that turns
77    /// into a long hang rather than a fast failure.
78    Throttled,
79    /// `500 Internal Error`.
80    ServerError,
81    /// The object is not there. Unlike the other two this is not a transport
82    /// fault, and a caller is entitled to treat it as an answer.
83    NotFound,
84}
85
86impl FaultKind {
87    /// Render this fault as the `object_store` error a backend would return.
88    fn error(self, store: &'static str, location: &Path) -> object_store::Error {
89        match self {
90            Self::Throttled => object_store::Error::Generic {
91                store,
92                source: "503 SlowDown: Please reduce your request rate.".into(),
93            },
94            Self::ServerError => object_store::Error::Generic {
95                store,
96                source: "500 InternalError: We encountered an internal error.".into(),
97            },
98            Self::NotFound => object_store::Error::NotFound {
99                path: location.to_string(),
100                source: "injected".into(),
101            },
102        }
103    }
104}
105
106/// What one operation class does under fault injection.
107#[derive(Debug, Clone, Copy, PartialEq, Eq)]
108pub struct OpFault {
109    /// The error a failing call produces.
110    pub kind: FaultKind,
111    /// How many calls out of every thousand fail. `0` never fails, `1000`
112    /// always does, and the calls that fail are the same ones on every run:
113    /// call `n` fails when `(n + 1) * rate / 1000` exceeds `n * rate / 1000`.
114    pub fail_per_thousand: u32,
115    /// How long every call to this operation sleeps before it is answered or
116    /// failed. This is the stall: the call still completes, eventually, which
117    /// is exactly what makes an unbounded caller hang on it.
118    pub latency: Duration,
119}
120
121impl OpFault {
122    /// Fail every call with `kind`, immediately.
123    #[must_use]
124    pub const fn always(kind: FaultKind) -> Self {
125        Self {
126            kind,
127            fail_per_thousand: 1_000,
128            latency: Duration::ZERO,
129        }
130    }
131
132    /// Fail `per_thousand` calls out of every thousand with `kind`.
133    #[must_use]
134    pub const fn rate(kind: FaultKind, per_thousand: u32) -> Self {
135        Self {
136            kind,
137            fail_per_thousand: per_thousand,
138            latency: Duration::ZERO,
139        }
140    }
141
142    /// Answer every call correctly, `latency` late. A store that is merely
143    /// slow, which is the fault no error counter ever sees.
144    #[must_use]
145    pub const fn stall(latency: Duration) -> Self {
146        Self {
147            kind: FaultKind::ServerError,
148            fail_per_thousand: 0,
149            latency,
150        }
151    }
152
153    /// The same fault, `latency` late.
154    #[must_use]
155    pub const fn with_latency(mut self, latency: Duration) -> Self {
156        self.latency = latency;
157        self
158    }
159
160    /// Whether the zero-based call number `n` to this operation fails.
161    const fn fails_call(self, n: u64) -> bool {
162        let rate = self.fail_per_thousand as u64;
163        ((n + 1) * rate) / 1_000 > (n * rate) / 1_000
164    }
165}
166
167/// Which operations misbehave, and how.
168///
169/// An operation with no entry is passed straight through to the inner store.
170#[derive(Debug, Clone, Default, PartialEq, Eq)]
171pub struct FaultPolicy {
172    faults: BTreeMap<StoreOp, OpFault>,
173}
174
175impl FaultPolicy {
176    /// A policy that injects nothing.
177    #[must_use]
178    pub fn none() -> Self { Self::default() }
179
180    /// Apply `fault` to `op`, replacing any previous entry for it.
181    #[must_use]
182    pub fn with(mut self, op: StoreOp, fault: OpFault) -> Self {
183        self.faults.insert(op, fault);
184        self
185    }
186
187    /// Apply `fault` to every operation class.
188    #[must_use]
189    pub fn with_all(mut self, fault: OpFault) -> Self {
190        for op in StoreOp::ALL {
191            self.faults.insert(op, fault);
192        }
193        self
194    }
195
196    /// The fault configured for `op`, if any.
197    #[must_use]
198    pub fn fault(&self, op: StoreOp) -> Option<OpFault> { self.faults.get(&op).copied() }
199}
200
201/// Per-operation call counters, so a test can assert how many times the store
202/// was reached rather than inferring it.
203#[derive(Debug, Default)]
204struct Counters {
205    put: AtomicU64,
206    get: AtomicU64,
207    delete: AtomicU64,
208    list: AtomicU64,
209    copy: AtomicU64,
210}
211
212impl Counters {
213    fn slot(&self, op: StoreOp) -> &AtomicU64 {
214        match op {
215            StoreOp::Put => &self.put,
216            StoreOp::Get => &self.get,
217            StoreOp::Delete => &self.delete,
218            StoreOp::List => &self.list,
219            StoreOp::Copy => &self.copy,
220        }
221    }
222}
223
224/// An [`ObjectStore`] that delays and fails calls according to a
225/// [`FaultPolicy`], delegating everything it does not fail to an inner store.
226#[derive(Debug)]
227pub struct FaultInjectingStore {
228    inner: Arc<dyn ObjectStore>,
229    policy: FaultPolicy,
230    counters: Counters,
231    get_replacement: Option<Bytes>,
232}
233
234fn failed_stream<T: Send + 'static>(
235    kind: FaultKind,
236    operation: StoreOp,
237) -> BoxStream<'static, object_store::Result<T>> {
238    let error = kind.error(operation.label(), &listing_path());
239    stream::once(async move { Err(error) }).boxed()
240}
241
242impl FaultInjectingStore {
243    /// A fault-free in-memory store for consistency tests that inject one
244    /// targeted GET replacement.
245    #[must_use]
246    pub fn memory() -> Self { Self::new(Arc::new(InMemory::new()), FaultPolicy::none()) }
247
248    /// Wrap `inner` with `policy`.
249    #[must_use]
250    pub fn new(inner: Arc<dyn ObjectStore>, policy: FaultPolicy) -> Self {
251        Self {
252            inner,
253            policy,
254            counters: Counters::default(),
255            get_replacement: None,
256        }
257    }
258
259    /// Overwrite the requested object immediately before each GET.
260    ///
261    /// HEAD calls still report the existing object. This injects a deterministic
262    /// overwrite between a metadata check and its read, for consistency drills.
263    /// The write bypasses put faults and uses the inner store's normal overwrite.
264    #[must_use]
265    pub fn with_get_overwrite(mut self, replacement: Bytes) -> Self {
266        self.get_replacement = Some(replacement);
267        self
268    }
269
270    /// How many calls of class `op` have reached this store, failed ones
271    /// included.
272    #[must_use]
273    pub fn attempts(&self, op: StoreOp) -> u64 { self.counters.slot(op).load(Ordering::SeqCst) }
274
275    /// Count the call, sleep out any injected latency, and report whether the
276    /// caller should be failed.
277    async fn enter(&self, op: StoreOp, location: &Path) -> Result<(), object_store::Error> {
278        let n = self.counters.slot(op).fetch_add(1, Ordering::SeqCst);
279        let Some(fault) = self.policy.fault(op) else {
280            return Ok(());
281        };
282        if !fault.latency.is_zero() {
283            time::sleep(fault.latency).await;
284        }
285        if fault.fails_call(n) {
286            return Err(fault.kind.error(op.label(), location));
287        }
288        Ok(())
289    }
290}
291
292impl fmt::Display for FaultInjectingStore {
293    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
294        write!(f, "FaultInjectingStore({})", self.inner)
295    }
296}
297
298/// The path a listing fault reports. Listing has no single location, and the
299/// error kinds that matter carry the store name rather than the path.
300fn listing_path() -> Path { Path::from("") }
301
302#[async_trait::async_trait]
303impl ObjectStore for FaultInjectingStore {
304    async fn put_opts(
305        &self,
306        location: &Path,
307        payload: PutPayload,
308        opts: PutOptions,
309    ) -> object_store::Result<PutResult> {
310        self.enter(StoreOp::Put, location).await?;
311        self.inner.put_opts(location, payload, opts).await
312    }
313
314    async fn put_multipart_opts(
315        &self,
316        location: &Path,
317        opts: PutMultipartOptions,
318    ) -> object_store::Result<Box<dyn MultipartUpload>> {
319        self.enter(StoreOp::Put, location).await?;
320        self.inner.put_multipart_opts(location, opts).await
321    }
322
323    async fn get_opts(
324        &self,
325        location: &Path,
326        options: GetOptions,
327    ) -> object_store::Result<GetResult> {
328        self.enter(StoreOp::Get, location).await?;
329        if !options.head
330            && let Some(replacement) = &self.get_replacement
331        {
332            self.inner
333                .put_opts(
334                    location,
335                    PutPayload::from_bytes(replacement.clone()),
336                    PutOptions::default(),
337                )
338                .await?;
339        }
340        self.inner.get_opts(location, options).await
341    }
342
343    fn delete_stream(
344        &self,
345        locations: BoxStream<'static, object_store::Result<Path>>,
346    ) -> BoxStream<'static, object_store::Result<Path>> {
347        let n = self
348            .counters
349            .slot(StoreOp::Delete)
350            .fetch_add(1, Ordering::SeqCst);
351        match self.policy.fault(StoreOp::Delete) {
352            // The delete stream is built synchronously, so an injected
353            // latency here is applied per yielded path rather than once.
354            Some(fault) if fault.fails_call(n) => failed_stream(fault.kind, StoreOp::Delete),
355            Some(fault) if !fault.latency.is_zero() => {
356                let stream = self.inner.delete_stream(locations);
357                stream
358                    .then(move |item| async move {
359                        time::sleep(fault.latency).await;
360                        item
361                    })
362                    .boxed()
363            }
364            _ => self.inner.delete_stream(locations),
365        }
366    }
367
368    fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, object_store::Result<ObjectMeta>> {
369        let n = self
370            .counters
371            .slot(StoreOp::List)
372            .fetch_add(1, Ordering::SeqCst);
373        match self.policy.fault(StoreOp::List) {
374            Some(fault) if fault.fails_call(n) => failed_stream(fault.kind, StoreOp::List),
375            Some(fault) if !fault.latency.is_zero() => {
376                let stream = self.inner.list(prefix);
377                let latency = fault.latency;
378                stream::once(async move { time::sleep(latency).await })
379                    .flat_map(move |()| stream::empty())
380                    .chain(stream)
381                    .boxed()
382            }
383            _ => self.inner.list(prefix),
384        }
385    }
386
387    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> object_store::Result<ListResult> {
388        self.enter(StoreOp::List, &listing_path()).await?;
389        self.inner.list_with_delimiter(prefix).await
390    }
391
392    async fn copy_opts(
393        &self,
394        from: &Path,
395        to: &Path,
396        options: CopyOptions,
397    ) -> object_store::Result<()> {
398        self.enter(StoreOp::Copy, from).await?;
399        self.inner.copy_opts(from, to, options).await
400    }
401}
402
403/// Build the store `cfg` selects and wrap it in `policy`.
404///
405/// **Testing seam.** This exists so a fault drill can hand a misbehaving store
406/// to code whose only entry point takes an `Arc<dyn ObjectStore>`. The broker's
407/// serving path never calls it; [`crate::build_object_store`] is the builder
408/// that does.
409///
410/// # Errors
411///
412/// Whatever [`crate::build_object_store`] returns for `cfg`.
413pub fn build_faulty_object_store(
414    cfg: &ObjectStoreConfig,
415    policy: FaultPolicy,
416) -> Result<Arc<FaultInjectingStore>, ObjectStoreError> {
417    Ok(Arc::new(FaultInjectingStore::new(
418        build_object_store(cfg)?,
419        policy,
420    )))
421}
422
423#[cfg(test)]
424mod tests {
425    use {
426        super::*,
427        crate::{config::ObjectStoreConfig, test_support::round_trip},
428        assert2::{assert, check},
429        object_store::{ObjectStoreExt as _, memory::InMemory},
430        std::array,
431        tokio::time::Instant,
432    };
433
434    fn memory(policy: FaultPolicy) -> Arc<FaultInjectingStore> {
435        build_faulty_object_store(&ObjectStoreConfig::InMemory, policy).unwrap()
436    }
437
438    #[tokio::test]
439    async fn get_overwrite_changes_the_body_after_head_without_changing_head_itself() {
440        let inner: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
441        let key = Path::from("mutable");
442        inner
443            .put(&key, PutPayload::from_static(b"old"))
444            .await
445            .unwrap();
446        let store = FaultInjectingStore::new(inner, FaultPolicy::none())
447            .with_get_overwrite(Bytes::from_static(b"replacement"));
448        check!(store.head(&key).await.unwrap().size == 3);
449        check!(
450            store
451                .get(&key)
452                .await
453                .unwrap()
454                .bytes()
455                .await
456                .unwrap()
457                .as_ref()
458                == b"replacement"
459        );
460        check!(store.head(&key).await.unwrap().size == 11);
461        check!(store.attempts(StoreOp::Get) == 3);
462    }
463
464    /// A rate is a schedule, not a coin flip: the same rate must fail the same
465    /// call numbers every run, or a suite that asserts on which call failed is
466    /// asserting on luck.
467    #[yare::parameterized(
468        never = { 0, [false, false, false, false] },
469        one_in_four = { 250, [false, false, false, true] },
470        every_other = { 500, [false, true, false, true] },
471        always = { 1_000, [true, true, true, true] },
472    )]
473    fn fail_rates_pick_the_same_calls_every_run(rate: u32, want: [bool; 4]) {
474        let fault = OpFault::rate(FaultKind::Throttled, rate);
475        let got: [bool; 4] = array::from_fn(|n| fault.fails_call(n as u64));
476        check!(got == want, "rate {rate}");
477    }
478
479    /// A `503 SlowDown` on `Put` must reach the caller as an error, and the
480    /// object must not land: a fault that quietly succeeded would let a copy
481    /// path pass a drill it never survived.
482    #[tokio::test]
483    async fn a_throttled_put_fails_and_writes_nothing() {
484        let store =
485            memory(FaultPolicy::none().with(StoreOp::Put, OpFault::always(FaultKind::Throttled)));
486        let path = Path::from("seg.log");
487
488        let err = store
489            .put(&path, PutPayload::from(b"bytes".to_vec()))
490            .await
491            .unwrap_err();
492
493        check!(err.to_string().contains("503 SlowDown"));
494        assert!(store.get(&path).await.is_err());
495        check!(store.attempts(StoreOp::Put) == 1);
496    }
497
498    /// Only the named operation misbehaves. A policy aimed at `Put` must leave
499    /// `Get` alone, or every drill would be indistinguishable from a store
500    /// that is entirely down.
501    #[tokio::test]
502    async fn faults_are_scoped_to_the_named_operation() {
503        let store =
504            memory(FaultPolicy::none().with(StoreOp::Get, OpFault::always(FaultKind::NotFound)));
505        let path = Path::from("seg.log");
506
507        store
508            .put(&path, PutPayload::from(b"bytes".to_vec()))
509            .await
510            .unwrap();
511        let err = store.get(&path).await.unwrap_err();
512
513        check!(matches!(err, object_store::Error::NotFound { .. }));
514        check!(store.attempts(StoreOp::Put) == 1);
515        check!(store.attempts(StoreOp::Get) == 1);
516    }
517
518    /// A stall answers correctly, late. That is the fault an error counter
519    /// never sees, and the reason a caller needs a deadline of its own.
520    #[tokio::test(start_paused = true)]
521    async fn a_stall_succeeds_but_only_after_the_injected_latency() {
522        let store =
523            memory(FaultPolicy::none().with(StoreOp::Put, OpFault::stall(Duration::from_secs(30))));
524        let path = Path::from("seg.log");
525
526        let started = Instant::now();
527        store
528            .put(&path, PutPayload::from(b"bytes".to_vec()))
529            .await
530            .unwrap();
531
532        check!(started.elapsed() >= Duration::from_secs(30));
533        assert!(store.get(&path).await.is_ok());
534    }
535
536    /// The policy is a wrapper, so a store with no faults must behave exactly
537    /// like the one it wraps.
538    #[tokio::test]
539    async fn an_empty_policy_passes_everything_through() {
540        let store = memory(FaultPolicy::none());
541        let path = Path::from("seg.log");
542
543        let got = round_trip(&store, &path, b"bytes").await;
544
545        check!(&got[..] == b"bytes");
546        check!(store.attempts(StoreOp::Put) == 1);
547    }
548
549    /// `with_all` reaches every class, including the two whose faults are
550    /// injected into a stream rather than an `async fn`.
551    #[tokio::test]
552    async fn with_all_fails_listing_and_deleting_too() {
553        let store = memory(FaultPolicy::none().with_all(OpFault::always(FaultKind::ServerError)));
554
555        let listed = store.list(None).collect::<Vec<_>>().await;
556        let deleted = store.delete(&Path::from("seg.log")).await;
557
558        check!(listed.len() == 1);
559        check!(listed[0].is_err());
560        check!(deleted.is_err());
561        check!(store.attempts(StoreOp::List) == 1);
562        check!(store.attempts(StoreOp::Delete) == 1);
563    }
564}
krabka_object_store/fault.rs

krabka_object_store/
fault.rs
