Method commit_at

src/txn-wal/src/txn_write.rs:119–340

(
        &mut self,
        handle: &mut TxnsHandle<K, V, T, D, C>,
        commit_ts: T,
    )

    /// Panics if any involved data shards were not registered before commit ts.
    #[instrument(level = "debug", fields(ts = ?commit_ts))]
    pub async fn commit_at<C>(
        &mut self,
        handle: &mut TxnsHandle<K, V, T, D, C>,
        commit_ts: T,
    ) -> Result<TxnApply<T>, T>
    where
        C: TxnsCodec,
    {
        let op = &Arc::clone(&handle.metrics).commit;
        op.run(async {
            let mut txns_upper = handle
                .txns_write
                .shared_upper()
                .into_option()
                .expect("txns shard should not be closed");

            loop {
                txns_upper = handle.txns_cache.update_ge(&txns_upper).await.clone();

                // txns_upper is the (inclusive) minimum timestamp at which we
                // could possibly write. If our requested commit timestamp is before
                // that, then it's no longer possible to write and the caller needs
                // to decide what to do.
                if commit_ts < txns_upper {
                    debug!(
                        "commit_at {:?} mismatch current={:?}",
                        commit_ts, txns_upper
                    );
                    return Err(txns_upper);
                }
                // Validate that the involved data shards are all registered.
                for (data_id, _) in self.writes.iter() {
                    assert!(
                        handle
                            .txns_cache
                            .registered_at_progress(data_id, &txns_upper),
                        "{} should be registered to commit at {:?}",
                        data_id,
                        txns_upper,
                    );
                }
                debug!(
                    "commit_at {:?}: [{:?}, {:?}) begin",
                    commit_ts,
                    txns_upper,
                    commit_ts.step_forward(),
                );

                let txn_batches_updates = FuturesUnordered::new();
                while let Some((data_id, updates)) = self.writes.pop_first() {
                    let data_write = handle.datas.take_write_for_commit(&data_id).unwrap_or_else(
                        || {
                            panic!(
                                "data shard {} must be registered with this Txn handle to commit",
                                data_id
                            )
                        },
                    );