pipecircuit 0.0.22

Correct. Robust. Fast. Portable. Pick four.
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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

use super::super::observed::ObservedTable;
use super::super::{
    WarpTable, WarpTableGetOutcome, WarpTableInsertWithIdOutcome, WarpTableRemoveOutcome,
};
use super::{Change, Error, PersistentRow};
use alloc::vec::Vec;

/// Apply an already validated group to a published materialization. Removal
/// and reinsertion permit unique-key swaps; replacements retain incarnation.
/// Any unexpected failure poisons the owning authority before it can be read.
pub(super) fn apply_observed<Row: PersistentRow>(
    source: &mut ObservedTable<Row>,
    changes: &[Change<Row, Row::Id>],
    incarnations: &mut Vec<(Row::Id, u64)>,
) -> Result<(), Error> {
    source
        .sequence
        .checked_add(u64::try_from(changes.len()).map_err(|_| Error::Exhausted)?)
        .ok_or(Error::Exhausted)?;
    incarnations.clear();
    for change in changes {
        match source
            .table
            .primary_index
            .find(change.id(), &source.table.ids)
            .map_err(|_| Error::Corrupt)?
        {
            super::super::dense_key_index::FindOutcome::Present { location, .. } => {
                let row = location.row;
                let incarnation = *source.incarnations.get(row).ok_or(Error::Corrupt)?;
                incarnations.push((change.id(), incarnation));
                remove_observed_row(source, change.id(), row)?;
            }
            super::super::dense_key_index::FindOutcome::Absent => {}
        }
    }
    for change in changes {
        let sequence = source.sequence.checked_add(1).ok_or(Error::Exhausted)?;
        match change {
            Change::Insert { id, row } => {
                insert(&mut source.table, *id, row.clone())?;
                source.incarnations.push(sequence);
            }
            Change::Replace { id, row } => {
                let (_, incarnation) = incarnations
                    .iter()
                    .find(|(stored, _)| stored == id)
                    .ok_or(Error::Corrupt)?;
                insert(&mut source.table, *id, row.clone())?;
                source.incarnations.push(*incarnation);
            }
            Change::Remove { .. } => {}
        }
        source.record(change.id(), sequence);
    }
    incarnations.clear();
    Ok(())
}

pub(super) fn remove_observed_row<Row: PersistentRow>(
    source: &mut ObservedTable<Row>,
    id: Row::Id,
    row: usize,
) -> Result<(), Error> {
    let (_id, _removed, _relocation) =
        super::super::remove::require_removal(source.table.remove(id), Error::Corrupt)?;
    source.incarnations.swap_remove(row);
    Ok(())
}

pub(super) fn insert<Row: PersistentRow>(
    table: &mut WarpTable<Row>,
    id: Row::Id,
    row: Row,
) -> Result<(), Error> {
    match table.insert_with_id(id, row) {
        WarpTableInsertWithIdOutcome::Inserted { .. } => Ok(()),
        WarpTableInsertWithIdOutcome::AlreadyPresent { .. } => Err(Error::Present),
        WarpTableInsertWithIdOutcome::Full { .. } => Err(Error::Full),
        WarpTableInsertWithIdOutcome::UniqueConflict { .. } => Err(Error::UniqueConflict),
    }
}

/// The caller supplies private staging storage. A refusal restores logical rows
/// and indexes, including swaps of two unique indexed values in one operation.
pub(super) fn apply<Row: PersistentRow>(
    table: &mut WarpTable<Row>,
    changes: &[Change<Row, Row::Id>],
    undo: &mut Vec<(Row::Id, Row)>,
) -> Result<(), Error> {
    undo.clear();
    for (index, change) in changes.iter().enumerate() {
        if changes
            .iter()
            .take(index)
            .any(|other| other.id() == change.id())
        {
            return Err(Error::DuplicateKey);
        }
        let present = matches!(table.get(change.id()), WarpTableGetOutcome::Present { .. });
        match change {
            Change::Insert { .. } if present => return Err(Error::Present),
            Change::Replace { .. } | Change::Remove { .. } if !present => {
                return Err(Error::Absent);
            }
            Change::Insert { .. } | Change::Replace { .. } | Change::Remove { .. } => {}
        }
    }
    for change in changes {
        match table.remove(change.id()) {
            WarpTableRemoveOutcome::Removed { removed, .. } => undo.push((change.id(), removed)),
            WarpTableRemoveOutcome::Absent { .. } => {}
        }
    }
    let mut failure = None;
    for change in changes {
        let result = match change {
            Change::Insert { id, row } | Change::Replace { id, row } => {
                insert(table, *id, row.clone())
            }
            Change::Remove { .. } => Ok(()),
        };
        if let Err(error) = result {
            failure = Some(error);
            break;
        }
    }
    if let Some(error) = failure {
        for change in changes {
            match table.remove(change.id()) {
                WarpTableRemoveOutcome::Removed { .. } | WarpTableRemoveOutcome::Absent { .. } => {}
            }
        }
        for (id, row) in undo.drain(..) {
            insert(table, id, row).map_err(|_| Error::Corrupt)?;
        }
        return Err(error);
    }
    undo.clear();
    Ok(())
}