wal multiplexedblob: implement is_present

Summary:
Is_present on multiplexed blobstore:
- return `Present` as soon as any of the underlying `is_present` successfully returns `Present`
- return `Absent` when at least read-quorum blobstores returned `Absent` (we don't wait on the other `is_present`s)
- return `ProbablyNotPresent` if some blobstores returned `Absent` (but not enough for quorum), the rest failed
- fail if all of the blobstores failed.

It is possible for multiplexed `is_present` to return inconsistent results if the previous put failed. For example, the put operation successfully logged itself to the WAL, but 2 out of 3 underlying puts failed. It means only 1 blobstore has the blob.

Now multiplexed `is_present` tries to read from the blobstores: first 2 underlying `is_present` returned `Absent` -> quorum achieved, multiplexed returns `Absent`. But if first returns the blobstore that actually has the blob, multiplexed `is_present` will return `Present`.

This is fine because the only guarantee we have: if the put succeeded, the blob will be readable from the storage. But in case like above, the blob will be either eventually replicated to all the blobstores by healer or will be rewritten by another put (because the first put failed).

Reviewed By: farnz

Differential Revision: D37342528

fbshipit-source-id: 21c62a9ecd2c2d117ee0b73d073f6ecd03f383c1

Author: kris1319

eden/mononoke/blobstore/multiplexedblob_wal/src/multiplex.rs

@@ -44,6 +44,8 @@ pub enum ErrorKind {
     SomePutsFailed(Arc<BlobstoresReturnedError>),
     #[error("Failures on get in underlying single blobstores: {0:?}")]
     SomeGetsFailed(Arc<BlobstoresReturnedError>),
+    #[error("Failures on is_present in underlying single blobstores: {0:?}")]
+    SomeIsPresentsFailed(Arc<BlobstoresReturnedError>),
 }
 
 #[derive(Clone, Debug)]
@@ -238,7 +240,7 @@ impl WalMultiplexedBlobstore {
                 Ok(None) => {
                     quorum = quorum.saturating_sub(1);
                     if quorum == 0 {
-                        // quorum blobstores couldn't find the given key in the blobstoers
+                        // quorum blobstores couldn't find the given key in the blobstores
                         // let's trust them
                         return Ok(None);
                     }
@@ -266,6 +268,61 @@ impl WalMultiplexedBlobstore {
 
         Err(result_err.into())
     }
+
+    // TODO(aida): comprehensive lookup (D30839608)
+    async fn is_present_impl<'a>(
+        &'a self,
+        ctx: &'a CoreContext,
+        key: &'a str,
+    ) -> Result<BlobstoreIsPresent> {
+        let mut futs = inner_multi_is_present(ctx, self.blobstores.clone(), key);
+
+        // Wait for the quorum successful "Not Found" reads before
+        // returning Ok(None).
+        let mut quorum: usize = self.quorum.read.get();
+        let mut errors = HashMap::with_capacity(futs.len());
+        while let Some(result) = futs.next().await {
+            match result {
+                (_, Ok(BlobstoreIsPresent::Present)) => {
+                    return Ok(BlobstoreIsPresent::Present);
+                }
+                (_, Ok(BlobstoreIsPresent::Absent)) => {
+                    quorum = quorum.saturating_sub(1);
+                    if quorum == 0 {
+                        // quorum blobstores couldn't find the given key in the blobstores
+                        // let's trust them
+                        return Ok(BlobstoreIsPresent::Absent);
+                    }
+                }
+                (bs_id, Ok(BlobstoreIsPresent::ProbablyNotPresent(err))) => {
+                    // Treat this like an error from the underlying blobstore.
+                    // In reality, this won't happen as multiplexed operates over sinle
+                    // standard blobstores, which always can answer if the blob is present.
+                    errors.insert(bs_id, err);
+                }
+                (bs_id, Err(err)) => {
+                    errors.insert(bs_id, err);
+                }
+            }
+        }
+
+        // TODO(aida): read from write-mostly blobstores once in a while, but don't use
+        // those in the quorum.
+
+        // At this point the multiplexed is_present either failed or cannot say for sure
+        // if the blob is present:
+        // - no blob was found, but some of the blobstore `is_present` calls failed
+        // - there was no read quorum on "not found" result
+        let errors = Arc::new(errors);
+        if errors.len() == self.blobstores.len() {
+            // all main reads failed -> is_present failed
+            return Err(ErrorKind::AllFailed(errors).into());
+        }
+
+        Ok(BlobstoreIsPresent::ProbablyNotPresent(
+            ErrorKind::SomeIsPresentsFailed(errors).into(),
+        ))
+    }
 }
 
 #[async_trait]
@@ -280,10 +337,10 @@ impl Blobstore for WalMultiplexedBlobstore {
 
     async fn is_present<'a>(
         &'a self,
-        _ctx: &'a CoreContext,
-        _key: &'a str,
+        ctx: &'a CoreContext,
+        key: &'a str,
     ) -> Result<BlobstoreIsPresent> {
-        unimplemented!();
+        self.is_present_impl(ctx, key).await
     }
 
     async fn put<'a>(
@@ -374,3 +431,18 @@ fn inner_multi_get<'a>(
         .collect();
     get_futs
 }
+
+fn inner_multi_is_present<'a>(
+    ctx: &'a CoreContext,
+    blobstores: Arc<[(BlobstoreId, Arc<dyn BlobstorePutOps>)]>,
+    key: &'a str,
+) -> FuturesUnordered<impl Future<Output = (BlobstoreId, Result<BlobstoreIsPresent, Error>)> + 'a> {
+    let futs: FuturesUnordered<_> = blobstores
+        .iter()
+        .map(|(bs_id, bs)| {
+            cloned!(bs_id, bs, ctx);
+            async move { (bs_id, bs.is_present(&ctx, key).await) }
+        })
+        .collect();
+    futs
+}

eden/mononoke/blobstore/multiplexedblob_wal/src/test.rs

@@ -5,9 +5,11 @@
  * GNU General Public License version 2.
  */
 
+use anyhow::anyhow;
 use anyhow::Result;
 use blobstore::Blobstore;
 use blobstore::BlobstoreGetData;
+use blobstore::BlobstoreIsPresent;
 use blobstore::BlobstorePutOps;
 use blobstore_sync_queue::BlobstoreWal;
 use blobstore_sync_queue::OperationKey;
@@ -558,8 +560,214 @@ async fn test_get_on_existing(fb: FacebookInit) -> Result<()> {
     Ok(())
 }
 
-async fn assert_pending<T: PartialEq + Debug>(fut: &mut (impl Future<Output = T> + Unpin)) {
-    assert_eq!(PollOnce::new(fut).await, Poll::Pending);
+#[fbinit::test]
+async fn test_is_present_missing(fb: FacebookInit) -> Result<()> {
+    let ctx = CoreContext::test_mock(fb);
+
+    let (_tickable_queue, wal_queue) = setup_queue();
+    let (tickable_blobstores, blobstores) = setup_blobstores(3);
+
+    let quorum = 2;
+    let multiplex =
+        WalMultiplexedBlobstore::new(MultiplexId::new(1), wal_queue, blobstores, vec![], quorum)?;
+
+    // No blobstores have the key
+    let k = "k1";
+
+    // all `is_present` succeed, multiplexed returns `Absent`
+    {
+        let mut fut = multiplex.is_present(&ctx, k).map_err(|_| ()).boxed();
+        assert_pending(&mut fut).await;
+
+        tickable_blobstores[0].1.tick(None);
+        assert_pending(&mut fut).await;
+        tickable_blobstores[1].1.tick(None);
+
+        // the read-quorum on `None` achieved, multiplexed returns `Absent`
+        assert_is_present_ok(fut.await, BlobstoreIsPresent::Absent);
+        tickable_blobstores[2].1.drain(1);
+    }
+
+    // two `is_present`s succeed, multiplexed returns `Absent`
+    {
+        let mut fut = multiplex.is_present(&ctx, k).map_err(|_| ()).boxed();
+        assert_pending(&mut fut).await;
+
+        tickable_blobstores[0].1.tick(None);
+        tickable_blobstores[1].1.tick(Some("bs1 failed"));
+        // muliplexed is_present waits on the third
+        assert_pending(&mut fut).await;
+        tickable_blobstores[2].1.tick(None);
+
+        // the read-quorum achieved, multiplexed returns `Absent`
+        assert_is_present_ok(fut.await, BlobstoreIsPresent::Absent);
+    }
+
+    // two `is_present`s fail, multiplexed returns `ProbablyNotPresent`
+    {
+        let mut fut = multiplex.is_present(&ctx, k).map_err(|_| ()).boxed();
+        assert_pending(&mut fut).await;
+
+        tickable_blobstores[0].1.tick(Some("bs0 failed"));
+        tickable_blobstores[1].1.tick(None);
+        // muliplexed is_present waits on the third
+        assert_pending(&mut fut).await;
+        tickable_blobstores[2].1.tick(Some("bs2 failed"));
+
+        assert_is_present_ok(
+            fut.await,
+            BlobstoreIsPresent::ProbablyNotPresent(anyhow!("some failed!")),
+        );
+    }
+
+    // all `is_present`s fail, multiplexed fails
+    {
+        let mut fut = multiplex.is_present(&ctx, k).map_err(|_| ()).boxed();
+        assert_pending(&mut fut).await;
+
+        for (id, store) in tickable_blobstores {
+            store.tick(Some(format!("bs{} failed!", id).as_str()));
+        }
+        assert!(fut.await.is_err());
+    }
+
+    Ok(())
+}
+
+#[fbinit::test]
+async fn test_is_present_existing(fb: FacebookInit) -> Result<()> {
+    let ctx = CoreContext::test_mock(fb);
+
+    let (tickable_queue, wal_queue) = setup_queue();
+    let (tickable_blobstores, blobstores) = setup_blobstores(3);
+
+    let quorum = 2;
+    let multiplex =
+        WalMultiplexedBlobstore::new(MultiplexId::new(1), wal_queue, blobstores, vec![], quorum)?;
+
+    // Two blobstores have the key, one failed to write: [ ] [x] [ ]
+    {
+        let v = make_value("v1");
+        let k = "k1";
+
+        let mut put_fut = multiplex
+            .put(&ctx, k.to_owned(), v.clone())
+            .map_err(|_| ())
+            .boxed();
+        assert_pending(&mut put_fut).await;
+
+        // wal queue write succeeds
+        tickable_queue.tick(None);
+        assert_pending(&mut put_fut).await;
+
+        tickable_blobstores[0].1.tick(None);
+        tickable_blobstores[1].1.tick(Some("bs1 failed"));
+        tickable_blobstores[2].1.tick(None);
+
+        // multiplexed put succeeds: write quorum achieved
+        assert!(put_fut.await.is_ok());
+
+        // first `is_present` succeed with `Present`, multiplexed returns `Present`
+        {
+            let mut fut = multiplex.is_present(&ctx, k).map_err(|_| ()).boxed();
+            assert_pending(&mut fut).await;
+
+            tickable_blobstores[0].1.tick(None);
+            assert_is_present_ok(fut.await, BlobstoreIsPresent::Present);
+
+            for (_id, store) in &tickable_blobstores[1..] {
+                store.drain(1);
+            }
+        }
+
+        // first `is_present` fails, second succeed with `Absent`, third returns `Present`
+        // multiplexed returns `Present`
+        {
+            let mut fut = multiplex.is_present(&ctx, k).map_err(|_| ()).boxed();
+            assert_pending(&mut fut).await;
+
+            tickable_blobstores[0].1.tick(Some("bs0 failed"));
+            tickable_blobstores[1].1.tick(None);
+            assert_pending(&mut fut).await;
+
+            tickable_blobstores[2].1.tick(None);
+            assert_is_present_ok(fut.await, BlobstoreIsPresent::Present);
+        }
+    }
+
+    // Two blobstores failed to write, one succeeded: [x] [ ] [x]
+    {
+        let v = make_value("v2");
+        let k = "k2";
+
+        let mut put_fut = multiplex
+            .put(&ctx, k.to_owned(), v.clone())
+            .map_err(|_| ())
+            .boxed();
+        assert_pending(&mut put_fut).await;
+
+        // wal queue write succeeds
+        tickable_queue.tick(None);
+        assert_pending(&mut put_fut).await;
+
+        tickable_blobstores[0].1.tick(Some("bs0 failed"));
+        tickable_blobstores[1].1.tick(None);
+        tickable_blobstores[2].1.tick(Some("bs2 failed"));
+
+        // multiplexed put failed: no write quorum
+        assert!(put_fut.await.is_err());
+
+        // the first `is_present` to succeed returns `Present`, multiplexed returns `Present`
+        {
+            let mut fut = multiplex.is_present(&ctx, k).map_err(|_| ()).boxed();
+            assert_pending(&mut fut).await;
+
+            tickable_blobstores[1].1.tick(None);
+            assert_is_present_ok(fut.await, BlobstoreIsPresent::Present);
+
+            tickable_blobstores[0].1.drain(1);
+            tickable_blobstores[2].1.drain(1);
+        }
+
+        // if the first two `is_present` calls return `Absent`, multiplexed returns `Absent`
+        {
+            let mut fut = multiplex.is_present(&ctx, k).map_err(|_| ()).boxed();
+            assert_pending(&mut fut).await;
+
+            tickable_blobstores[0].1.tick(None);
+            tickable_blobstores[2].1.tick(None);
+
+            assert_is_present_ok(fut.await, BlobstoreIsPresent::Absent);
+            tickable_blobstores[1].1.drain(1);
+        }
+
+        // if one `is_present` returns `Absent`, another 2 fail, multiplexed is unsure
+        {
+            let mut fut = multiplex.is_present(&ctx, k).map_err(|_| ()).boxed();
+            assert_pending(&mut fut).await;
+
+            tickable_blobstores[0].1.tick(None);
+            for (id, store) in &tickable_blobstores[1..] {
+                store.tick(Some(format!("bs{} failed", id).as_str()));
+            }
+
+            assert_is_present_ok(
+                fut.await,
+                BlobstoreIsPresent::ProbablyNotPresent(anyhow!("some failed!")),
+            );
+        }
+    }
+
+    Ok(())
+}
+
+async fn assert_pending<T: Debug>(fut: &mut (impl Future<Output = T> + Unpin)) {
+    match PollOnce::new(fut).await {
+        Poll::Pending => {}
+        state => {
+            panic!("future must be pending, received: {:?}", state);
+        }
+    }
 }
 
 type TickableBytes = Tickable<(BlobstoreBytes, u64)>;
@@ -601,3 +809,19 @@ fn validate_blob(
         assert_eq!(get_data.as_ref(), expected);
     }
 }
+
+fn assert_is_present_ok(result: Result<BlobstoreIsPresent, ()>, expected: BlobstoreIsPresent) {
+    assert!(result.is_ok());
+    match (result.unwrap(), expected) {
+        (BlobstoreIsPresent::Absent, BlobstoreIsPresent::Absent)
+        | (BlobstoreIsPresent::Present, BlobstoreIsPresent::Present)
+        | (BlobstoreIsPresent::ProbablyNotPresent(_), BlobstoreIsPresent::ProbablyNotPresent(_)) => {
+        }
+        (res, exp) => {
+            panic!(
+                "`is_present` call must return {:?}, received: {:?}",
+                exp, res
+            );
+        }
+    }
+}