Function test_concurrency

src/storage/src/source/reclock.rs:476–546

error: unsupported operation: can't call foreign function `decNumberFromInt32` on OS `linux`

()

    #[mz_ore::test(tokio::test)]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `decNumberFromInt32` on OS `linux`
    async fn test_concurrency() {
        // Create two operators pointing to the same shard
        let shared_shard = ShardId::new();
        let (mut op_a, _) = make_test_operator(shared_shard, Antichain::from_elem(0.into())).await;
        let (mut op_b, _) = make_test_operator(shared_shard, Antichain::from_elem(0.into())).await;

        // Mint some bindings through operator A
        let source_upper = partitioned_frontier([(0, MzOffset::from(3))]);
        let mut batch = op_a
            .mint(
                1000.into(),
                Antichain::from_elem(1001.into()),
                source_upper.borrow(),
            )
            .await;
        let mut expected_batch: ReclockBatch<_, Timestamp> = ReclockBatch {
            updates: vec![
                (
                    Partitioned::new_range(RB::NegInfinity, RB::before(0), MzOffset::from(0)),
                    1000.into(),
                    Diff::ONE,
                ),
                (
                    Partitioned::new_range(RB::after(0), RB::PosInfinity, MzOffset::from(0)),
                    1000.into(),
                    Diff::ONE,
                ),
                (
                    Partitioned::new_range(RB::NegInfinity, RB::PosInfinity, MzOffset::from(0)),
                    1000.into(),
                    Diff::MINUS_ONE,
                ),
                (
                    Partitioned::new_singleton(RB::exact(0), MzOffset::from(3)),
                    1000.into(),
                    Diff::ONE,
                ),
            ],
            upper: Antichain::from_elem(Timestamp::from(1001)),
        };
        batch.updates.sort();
        expected_batch.updates.sort();
        assert_eq!(batch, expected_batch);

        // Operator B should attempt to mint in one go, fail, re-sync, and retry only for the
        // bindings that still need minting
        let source_upper = partitioned_frontier([(0, MzOffset::from(5))]);
        let mut batch = op_b
            .mint(
                11000.into(),
                Antichain::from_elem(11001.into()),
                source_upper.borrow(),
            )
            .await;
        expected_batch.updates.extend([
            (
                Partitioned::new_singleton(RB::exact(0), MzOffset::from(3)),
                11000.into(),