ozzy-replication 0.5.0

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

//! Fixed-size election evidence. Full-WAL transfer is a separate exchange.

use super::{
    ReadFields, Reader, WireError, WriteFields, Writer, validate_election_view, validate_prefix,
};
use crate::{DoViewChange, FrozenLog, JournalGeneration, Prefix, Scope, StartView};

pub(super) fn validate_report(message: DoViewChange) -> Result<(), WireError> {
    validate_election_view(message.scope)?;
    validate_history(
        message.generation,
        message.log.accepted,
        message.log.committed,
    )?;
    if message.log.last_normal_view >= message.scope.view {
        return Err(WireError::History);
    }
    Ok(())
}

fn validate_history(
    generation: JournalGeneration,
    accepted: Prefix,
    committed: Prefix,
) -> Result<(), WireError> {
    validate_prefix(accepted)?;
    validate_prefix(committed)?;
    if generation.0 == 0
        || committed.op > accepted.op
        || (committed.op == accepted.op && committed != accepted)
    {
        return Err(WireError::History);
    }
    Ok(())
}

pub(super) fn write_report(
    writer: &mut Writer<'_>,
    message: DoViewChange,
) -> Result<(), WireError> {
    writer.bytes(&message.generation.0.to_be_bytes())?;
    writer.u64(message.log.last_normal_view)?;
    writer.prefix(message.log.accepted)?;
    writer.prefix(message.log.committed)?;
    Ok(())
}

pub(super) fn read_report(
    scope: Scope,
    reader: &mut Reader<'_>,
) -> Result<DoViewChange, WireError> {
    let message = DoViewChange {
        scope,
        generation: JournalGeneration(u128::from_be_bytes(reader.array()?)),
        log: FrozenLog {
            last_normal_view: reader.u64()?,
            accepted: reader.prefix()?,
            committed: reader.prefix()?,
        },
    };
    validate_report(message)?;
    Ok(message)
}

pub(super) fn validate_start(message: StartView) -> Result<(), WireError> {
    validate_election_view(message.scope)?;
    validate_history(message.generation, message.accepted, message.committed)
}

pub(super) fn write_start(writer: &mut Writer<'_>, message: StartView) -> Result<(), WireError> {
    writer.bytes(&message.generation.0.to_be_bytes())?;
    writer.prefix(message.accepted)?;
    writer.prefix(message.committed)?;
    Ok(())
}

pub(super) fn read_start(scope: Scope, reader: &mut Reader<'_>) -> Result<StartView, WireError> {
    let message = StartView {
        scope,
        generation: JournalGeneration(u128::from_be_bytes(reader.array()?)),
        accepted: reader.prefix()?,
        committed: reader.prefix()?,
    };
    validate_start(message)?;
    Ok(message)
}