ozzy-replication 0.1.1

Deterministic replication state for Ozzy.
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

use super::*;
use crate::flow::{ReceiveEpoch, Receiver};
use crate::{Digest, OpNumber, Scope};
use ozzy_proto::GroupId;

#[test]
fn sender_metadata_capacity_survives_partial_receipts_and_ring_wraparound() {
    let limits = PipelineLimits {
        max_operations: 4,
        max_body_bytes: 100,
    };
    let channel = Channel {
        scope: Scope {
            group_id: GroupId::from_bytes([1; 16]),
            configuration_epoch: 1,
            configuration_digest: Digest::from_bytes([2; 32]),
            view: 0,
        },
        epoch: ReceiveEpoch::new(1).unwrap(),
    };
    let remote = PipelineLimits {
        max_operations: 64,
        max_body_bytes: 8192,
    };
    let mut receiver = Receiver::new(channel, Prefix::GENESIS, remote).unwrap();
    let mut sender = Sender::open(receiver.report(), limits, Prefix::GENESIS).unwrap();
    let capacity = sender.pending.capacity();
    for _ in 0..2_500 {
        let mut previous = sender.sent();
        let operations: [_; 4] = std::array::from_fn(|_| {
            let operation = Operation {
                prefix: Prefix {
                    op: OpNumber(previous.op.0 + 1),
                    digest: Digest::from_bytes([1; 32]),
                },
                previous_digest: previous.digest,
                body_bytes: 1,
            };
            previous = operation.prefix;
            operation
        });
        sender.record_send(channel, &operations).unwrap();
        assert_eq!(sender.available().max_operations, 0);
        receiver.retain(channel, &operations[..1]).unwrap();
        receiver.release(operations[0].prefix).unwrap();
        sender.observe(receiver.report()).unwrap();
        assert_eq!(sender.pending.len(), 3);
        assert_eq!(sender.pending.capacity(), capacity);
        receiver.retain(channel, &operations[1..]).unwrap();
        receiver.release(operations[3].prefix).unwrap();
        sender.observe(receiver.report()).unwrap();
        assert!(sender.pending.is_empty());
        assert_eq!(sender.pending.capacity(), capacity);
    }
}

#[test]
fn reopening_after_lost_data_reuses_the_send_ledger() {
    let limits = PipelineLimits {
        max_operations: 4,
        max_body_bytes: 100,
    };
    let channel = Channel {
        scope: Scope {
            group_id: GroupId::from_bytes([1; 16]),
            configuration_epoch: 1,
            configuration_digest: Digest::from_bytes([2; 32]),
            view: 0,
        },
        epoch: ReceiveEpoch::new(1).unwrap(),
    };
    let mut receiver = Receiver::new(channel, Prefix::GENESIS, limits).unwrap();
    let mut sender = Sender::open(receiver.report(), limits, Prefix::GENESIS).unwrap();
    let capacity = sender.pending.capacity();
    for epoch in 2..10_002 {
        let previous = sender.sent();
        let op = Operation {
            prefix: Prefix {
                op: OpNumber(previous.op.0 + 1),
                digest: Digest::from_bytes([1; 32]),
            },
            previous_digest: previous.digest,
            body_bytes: 1,
        };
        sender.record_send(sender.channel(), &[op]).unwrap(); // Lost packet.
        receiver
            .retract(ReceiveEpoch::new(epoch).unwrap(), previous)
            .unwrap();
        sender.reopen(receiver.report(), previous).unwrap();
        assert!(sender.pending.is_empty());
        assert_eq!(sender.sent(), previous);
        sender.record_send(sender.channel(), &[op]).unwrap();
        receiver.retain(sender.channel(), &[op]).unwrap();
        receiver.release(op.prefix).unwrap();
        sender.observe(receiver.report()).unwrap();
        assert_eq!(sender.pending.capacity(), capacity);
    }
}