zksync_consensus_network 0.13.0

ZKsync consensus network component
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

use rand::Rng;
use tracing::instrument::Instrument as _;
use zksync_concurrency::{ctx, io, scope, testonly::abort_on_panic};

use crate::noise;

#[tokio::test]
async fn transmit_ok() {
    abort_on_panic();
    let ctx = &ctx::test_root(&ctx::RealClock);
    let (mut s1, mut s2) = noise::testonly::pipe(ctx).await;
    let msg = "hello";
    let n = 10;
    scope::run!(ctx, |ctx, s| async {
        s.spawn(
            async {
                let mut got = vec![0; msg.len()];
                for _ in 0..n {
                    io::read_exact(ctx, &mut s1, &mut got).await??;
                    assert_eq!(&got, msg.as_bytes());
                }
                io::shutdown(ctx, &mut s1).await??;
                Ok(())
            }
            .instrument(tracing::trace_span!("server")),
        );
        s.spawn(
            async {
                for i in 0..n {
                    io::write_all(ctx, &mut s2, msg.as_bytes()).await??;
                    if i % 3 == 0 {
                        io::flush(ctx, &mut s2).await??;
                    }
                }
                io::shutdown(ctx, &mut s2).await??;
                Ok(())
            }
            .instrument(tracing::trace_span!("client")),
        );
        anyhow::Ok(())
    })
    .await
    .unwrap();
}

#[tokio::test]
async fn transmit_sender_dies() {
    abort_on_panic();
    let ctx = &ctx::test_root(&ctx::RealClock);
    let (mut s1, mut s2) = noise::testonly::pipe(ctx).await;
    let msg = "hello";
    scope::run!(ctx, |ctx, s| async {
        s.spawn(
            async {
                let mut got = vec![0; msg.len() + 1];
                assert!(io::read_exact(ctx, &mut s1, &mut got).await?.is_err());
                io::shutdown(ctx, &mut s1).await??;
                Ok(())
            }
            .instrument(tracing::trace_span!("server")),
        );
        s.spawn(
            async {
                io::write_all(ctx, &mut s2, msg.as_bytes()).await??;
                io::shutdown(ctx, &mut s2).await??;
                Ok(())
            }
            .instrument(tracing::trace_span!("client")),
        );
        anyhow::Ok(())
    })
    .await
    .unwrap();
}

#[tokio::test]
async fn transmit_receiver_dies() {
    abort_on_panic();
    let ctx = &ctx::test_root(&ctx::RealClock);
    let rng = &mut ctx.rng();
    let (s1, s2) = noise::testonly::pipe(ctx).await;
    let mut msg = [0u8; 1024];
    rng.fill(&mut msg);

    scope::run!(ctx, |ctx, s| async {
        s.spawn(
            async {
                let mut s1 = s1;
                let mut got = vec![0; msg.len()];
                io::read_exact(ctx, &mut s1, &mut got).await??;
                assert_eq!(&got, &msg);
                Ok(())
            }
            .instrument(tracing::trace_span!("server")),
        );
        s.spawn(
            async {
                let mut s2 = s2;
                while io::write_all(ctx, &mut s2, &msg).await?.is_ok() {}
                Ok(())
            }
            .instrument(tracing::trace_span!("client")),
        );
        anyhow::Ok(())
    })
    .await
    .unwrap();
}