add transact limits test and make the code easier to read

Author: Vincent Rouillé

foundationdb/src/database.rs

@@ -133,8 +133,8 @@ impl Database {
         let mut trx = self.create_trx()?;
         let mut can_retry = move || {
             tries += 1;
-            retry_limit.filter(|&limit| tries > limit).is_none()
-                && time_out.filter(|&t| Instant::now() > t).is_none()
+            retry_limit.map(|limit| tries < limit).unwrap_or(true)
+                && time_out.map(|t| Instant::now() < t).unwrap_or(true)
         };
         loop {
             let r = f.transact(trx).await;

foundationdb/tests/get.rs

@@ -19,6 +19,8 @@ fn test_get() {
         futures::executor::block_on(test_set_conflict_async()).expect("failed to run");
         futures::executor::block_on(test_set_conflict_snapshot_async()).expect("failed to run");
         futures::executor::block_on(test_transact_async()).expect("failed to run");
+        futures::executor::block_on(test_transact_limit()).expect("failed to run");
+        futures::executor::block_on(test_transact_timeout()).expect("failed to run");
         futures::executor::block_on(test_versionstamp_async()).expect("failed to run");
         futures::executor::block_on(test_read_version_async()).expect("failed to run");
         futures::executor::block_on(test_set_read_version_async()).expect("failed to run");
@@ -167,6 +169,86 @@ async fn test_transact_async() -> FdbResult<()> {
     Ok(())
 }
 
+async fn test_transact_limit() -> FdbResult<()> {
+    const KEY: &[u8] = b"test_transact_limit";
+    async fn async_body(
+        db: &Database,
+        trx: &Transaction,
+        try_count0: Arc<AtomicUsize>,
+    ) -> FdbResult<()> {
+        // increment try counter
+        try_count0.fetch_add(1, Ordering::SeqCst);
+
+        // update conflict range
+        trx.get(KEY, false).await?;
+
+        // make current transaction invalid by making conflict
+        make_dirty(&db, KEY).await?;
+
+        trx.set(KEY, common::random_str(10).as_bytes());
+
+        // `Database::transact` will handle commit by itself, so returns without commit
+        Ok(())
+    }
+
+    let try_count = Arc::new(AtomicUsize::new(0));
+    let db = common::database().await?;
+    let res = db
+        .transact_boxed(
+            &db,
+            |trx, db| async_body(db, trx, try_count.clone()).boxed(),
+            TransactOption {
+                retry_limit: Some(5),
+                ..TransactOption::default()
+            },
+        )
+        .await;
+    assert!(res.is_err(), "should not be able to commit");
+
+    assert_eq!(try_count.load(Ordering::SeqCst), 5);
+
+    Ok(())
+}
+
+async fn test_transact_timeout() -> FdbResult<()> {
+    const KEY: &[u8] = b"test_transact_timeout";
+    async fn async_body(
+        db: &Database,
+        trx: &Transaction,
+        try_count0: Arc<AtomicUsize>,
+    ) -> FdbResult<()> {
+        // increment try counter
+        try_count0.fetch_add(1, Ordering::SeqCst);
+
+        // update conflict range
+        trx.get(KEY, false).await?;
+
+        // make current transaction invalid by making conflict
+        make_dirty(&db, KEY).await?;
+
+        trx.set(KEY, common::random_str(10).as_bytes());
+
+        // `Database::transact` will handle commit by itself, so returns without commit
+        Ok(())
+    }
+
+    let try_count = Arc::new(AtomicUsize::new(0));
+    let db = common::database().await?;
+    let res = db
+        .transact_boxed(
+            &db,
+            |trx, db| async_body(db, trx, try_count.clone()).boxed(),
+            TransactOption {
+                time_out: Some(std::time::Duration::from_millis(250)),
+                ..TransactOption::default()
+            },
+        )
+        .await;
+    assert!(res.is_err(), "should not be able to commit");
+
+    Ok(())
+}
+
 async fn test_versionstamp_async() -> FdbResult<()> {
     const KEY: &[u8] = b"test_versionstamp";
     let db = common::database().await?;