Transaction retry: Remove unnecessary Arc and callback boxing (#6922)

We do still need the `Mutex` around `self.inner` here, even though there
is never any contention. To remove the mutex, we'd need
`retry_callback()` to take `&mut self`, not `&self`, so that it could
directly mutate `self.inner`. However, `transaction_async_with_retry`
(defined in `async-bb8-diesel`) has this requirement on the retry
closure:

```rust
        RetryFut: FutureExt<Output = bool> + Send,
        RetryFunc: Fn() -> RetryFut + Send + Sync,
```

We'd need to change this `Fn` to `FnMut`. That's doable, but then the
closure we attempt to construct (`|| self.retry_callback()`) wouldn't
pass the borrow checker: the future returned by `self.retry_callback()`
would hold `&mut self`, but the generic bounds above would allow
`transaction_async_with_retry` to call the closure multiple times and
get multiple `RetryFut`s, therefore constructing multiple futures that
all hold `&mut self`s (which is obviously not okay).
`transaction_async_with_retry` doesn't actually _do_ that - it only
constructs one at a time and always awaits it before creating another
one - but I don't think there's any way to express that in the type
system.

We could fix this by moving `self` into the closure, but that defeats
the point: we need to access the details recorded in `self.inner`
_after_ `transaction_async_with_retry`, so we can't move `self` (or even
`self.inner`) into the closure. All this to say: I think the mutex this
has is the most straightforward way to make this all okay, so this PR
only changes some of the somewhat unnecessary surrounding code.

Author: jgallagher

nexus/db-queries/src/transaction_retry.rs

@@ -106,15 +106,14 @@ impl RetryHelper {
             + Send
             + Sync,
     {
-        let slef = Arc::new(self);
         let result = conn
-            .transaction_async_with_retry(f, slef.clone().as_callback())
+            .transaction_async_with_retry(f, || self.retry_callback())
             .await;
 
-        let retry_info = slef.inner.lock().unwrap();
+        let retry_info = self.inner.lock().unwrap();
         if retry_info.has_retried() {
             info!(
-                slef.log,
+                self.log,
                 "transaction completed";
                 "attempts" => retry_info.attempts,
             );
@@ -169,17 +168,6 @@ impl RetryHelper {
         let inner = self.inner.lock().unwrap().tick();
         return inner.attempts < MAX_RETRY_ATTEMPTS;
     }
-
-    // Converts this function to a retryable callback that can be used from
-    // "transaction_async_with_retry".
-    fn as_callback(
-        self: Arc<Self>,
-    ) -> impl Fn() -> futures::future::BoxFuture<'static, bool> {
-        move || {
-            let r = self.clone();
-            Box::pin(async move { r.retry_callback().await })
-        }
-    }
 }
 
 impl oximeter::Producer for Producer {