Pin futures passed to run_op()

[divergentdave]

This changes Transaction::run_op() so that it takes a pinned mutable reference to a future, instead of taking the future by value. This is a workaround for rust-lang/rust#62958. Here's the impact of this change according to RUSTFLAGS=-Zprint-type-sizes.

2368 {async fn body of janus_aggregator_core::datastore::Transaction<'_, janus_core::time::RealClock>::run_op<tokio_postgres::statement::Statement, {async fn body of deadpool_postgres::Transaction<'_>::prepare_cached()}>()} align=8
      1 <discriminant>
   2367 variant Suspend0
         15 <padding>
       1160 op (upvar) align=8
          8 op_group
          8 self (upvar)
          2 <padding>
          1 coroutine_field9 align=1 type=bool
          5 <padding>
          8 coroutine_field2 align=8 type=&AtomicBool
       1160 __awaitee type={async fn body of deadpool_postgres::Transaction<'_>::prepare_cached()}
   1399 variant Suspend1
         16 rslt align=8 offset=0
       1160 op (upvar)
          8 op_group
          8 self (upvar)
          1 coroutine_field7 type=bool
          1 coroutine_field8 type=bool
          1 coroutine_field9 type=bool
          5 <padding>
          8 barrier align=8
          8 barrier
        184 __awaitee type={async fn body of tokio::sync::Barrier::wait()}
   1192 variant Unresumed, Returned, Panicked
         15 <padding>
       1160 op (upvar) align=8
          8 <padding>
          8 self (upvar) align=8

248 {async fn body of janus_aggregator_core::datastore::Transaction<'_, janus_core::time::RealClock>::run_op<tokio_postgres::statement::Statement, {async fn body of deadpool_postgres::Transaction<'_>::prepare_cached()}>()} align=8
      1 <discriminant>
    247 variant Suspend1
         16 rslt align=8 offset=0
          8 op_group
          8 self (upvar)
          8 op (upvar)
          1 coroutine_field7 type=bool
          1 coroutine_field8 type=bool
          6 <padding>
          8 barrier align=8
          8 barrier
        184 __awaitee type={async fn body of tokio::sync::barrier::Barrier::wait()}
     63 variant Suspend0
         15 <padding>
          8 op_group align=8
          8 self (upvar)
          8 op (upvar)
          8 <padding>
          8 coroutine_field2 align=8 type=&AtomicBool
          8 __awaitee type=Pin<&mut {async fn body of deadpool_postgres::Transaction<'_>::prepare_cached()}>
     40 variant Unresumed, Returned, Panicked
         23 <padding>
          8 self (upvar) align=8
          8 op (upvar)

Here's the impact on futures a few levels of abstraction up:

3192 {async fn body of janus_aggregator_core::datastore::Transaction<'_, janus_core::time::RealClock>::update_batch_aggregation<32, janus_messages::query_type::TimeInterval, prio::vdaf::prio3::Prio3<prio::flp::types::SumVec<prio::field::Field64, prio::flp::gadgets::ParallelSum<prio::field::Field64, prio::flp::gadgets::Mul<prio::field::Field64>>>, prio::vdaf::xof::XofHmacSha256Aes128, 32>>()} align=8
      1 <discriminant>
   3191 variant Suspend0
          8 batch_aggregation align=8 offset=0
         40 __tracing_attr_span
          8 self (upvar)
          8 batch_aggregation (upvar)
          1 coroutine_field4 type=bool
          1 coroutine_field5 type=bool
          6 <padding>
       3120 __awaitee align=8 type=tracing::instrument::Instrumented<{async block@janus_aggregator_core::datastore::Transaction<'_, RealClock>::update_batch_aggregation<32, janus_messages::query_type::TimeInterval, prio::vdaf::prio3::Prio3<prio::flp::types::SumVec<prio::field::Field64, prio::flp::gadgets::ParallelSum<prio::field::Field64, prio::flp::gadgets::Mul<prio::field::Field64>>>, prio::vdaf::xof::XofHmacSha256Aes128, 32>>::{closure#0}::{closure#0}}>
   3151 variant Suspend1
          8 batch_aggregation align=8 offset=0
         40 __tracing_attr_span
          8 self (upvar)
          8 batch_aggregation (upvar)
          1 coroutine_field4 type=bool
          1 coroutine_field5 type=bool
          6 <padding>
       3080 __awaitee align=8 type={async block@janus_aggregator_core::datastore::Transaction<'_, RealClock>::update_batch_aggregation<32, janus_messages::query_type::TimeInterval, prio::vdaf::prio3::Prio3<prio::flp::types::SumVec<prio::field::Field64, prio::flp::gadgets::ParallelSum<prio::field::Field64, prio::flp::gadgets::Mul<prio::field::Field64>>>, prio::vdaf::xof::XofHmacSha256Aes128, 32>>::{closure#0}::{closure#0}}
     64 variant Unresumed, Returned, Panicked
         47 <padding>
          8 self (upvar) align=8
          8 batch_aggregation (upvar)

2200 {async fn body of janus_aggregator_core::datastore::Transaction<'_, janus_core::time::RealClock>::update_batch_aggregation<32, janus_messages::query_type::TimeInterval, prio::vdaf::prio3::Prio3<prio::flp::types::SumVec<prio::field::Field64, prio::flp::gadgets::ParallelSum<prio::field::Field64, prio::flp::gadgets::Mul<prio::field::Field64>>>, prio::vdaf::xof::XofHmacSha256Aes128, 32>>()} align=8
      1 <discriminant>
   2199 variant Suspend0
          8 batch_aggregation align=8 offset=0
         40 __tracing_attr_span
          8 self (upvar)
          8 batch_aggregation (upvar)
          1 coroutine_field4 type=bool
          1 coroutine_field5 type=bool
          6 <padding>
       2128 __awaitee align=8 type=tracing::instrument::Instrumented<{async block@janus_aggregator_core::datastore::Transaction<'_, RealClock>::update_batch_aggregation<32, janus_messages::query_type::TimeInterval, prio::vdaf::prio3::Prio3<prio::flp::types::SumVec<prio::field::Field64, prio::flp::gadgets::ParallelSum<prio::field::Field64, prio::flp::gadgets::Mul<prio::field::Field64>>>, prio::vdaf::xof::XofHmacSha256Aes128, 32>>::{closure#0}::{closure#0}}>
   2159 variant Suspend1
          8 batch_aggregation align=8 offset=0
         40 __tracing_attr_span
          8 self (upvar)
          8 batch_aggregation (upvar)
          1 coroutine_field4 type=bool
          1 coroutine_field5 type=bool
          6 <padding>
       2088 __awaitee align=8 type={async block@janus_aggregator_core::datastore::Transaction<'_, RealClock>::update_batch_aggregation<32, janus_messages::query_type::TimeInterval, prio::vdaf::prio3::Prio3<prio::flp::types::SumVec<prio::field::Field64, prio::flp::gadgets::ParallelSum<prio::field::Field64, prio::flp::gadgets::Mul<prio::field::Field64>>>, prio::vdaf::xof::XofHmacSha256Aes128, 32>>::{closure#0}::{closure#0}}
     64 variant Unresumed, Returned, Panicked
         47 <padding>
          8 self (upvar) align=8
          8 batch_aggregation (upvar)