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
142
143
144
145
146
147
148
149
//! Integral operator that panics when any record's accumulated weight
//! becomes negative.
use crate::{
ZWeight,
circuit::{Circuit, Stream, checkpointer::Checkpoint},
trace::{BatchReader as TraceBatchReader, Cursor as TraceCursor},
typed_batch::{BatchReader as TypedBatchReader, Spine, ZSet},
};
impl<C, B> Stream<C, B>
where
C: Circuit,
B: Checkpoint + Clone + ZSet + 'static,
B::InnerBatch: Send,
{
/// Integrates the stream, panicking if any record's weight in the integral
/// becomes negative.
///
/// Maintains a running trace of the input stream.
/// After every clock tick, every key that appears in the
/// current delta is checked: if the key's weight in the trace is
/// negative, the operator panics, displaying the offending key, weight,
/// and a custom message.
///
/// This operator is intended for debugging: it verifies that a stream
/// behaves like a valid set or multiset where every element has positive
/// multiplicity.
///
/// The operator passes the input delta through unchanged as its output.
///
/// # Panics
///
/// Panics when any weight in the integral is < 0.
#[track_caller]
pub fn integral_weight_validator(&self, message: &str) -> Self {
let message = message.to_owned();
self.circuit().region("integral_weight_validator", || {
let sharded = self.shard();
let trace = sharded.integrate_trace();
sharded.apply2(&trace, move |delta: &B, integrated: &Spine<B>| {
let mut delta_cursor = TraceBatchReader::cursor(delta.inner());
let mut trace_cursor = TraceBatchReader::cursor(integrated.inner());
while delta_cursor.key_valid() {
// Only examine keys that appear in the current delta with a negative weight
let delta_weight: ZWeight = **delta_cursor.weight();
if delta_weight < 0 {
let key = delta_cursor.key();
if trace_cursor.seek_key_exact(key, None) {
debug_assert!(trace_cursor.val_valid());
let weight: ZWeight = **trace_cursor.weight();
if weight < 0 {
panic!("{message}: negative weight found {weight} for key {key:?}");
}
}
}
delta_cursor.step_key();
}
});
self.clone()
})
}
}
#[cfg(test)]
mod tests {
use crate::{
Circuit, Runtime, ZWeight, operator::Generator, typed_batch::OrdZSet, utils::Tup2, zset,
};
/// A stream of purely positive deltas should pass the check at every step.
#[test]
fn integral_weight_validator_positive_weights() {
let (mut circuit, ()) = Runtime::init_circuit(1, move |circuit| {
let mut step = 0u64;
let source = circuit.add_source(Generator::new(move || {
let batch = zset! { step => 1 };
step += 1;
batch
}));
let out = source.integral_weight_validator("test_table");
let mut expected_step = 0u64;
out.inspect(move |delta| {
assert_eq!(delta, &zset! { expected_step => 1 });
expected_step += 1;
});
Ok(())
})
.unwrap();
for _ in 0..10 {
circuit.transaction().unwrap();
}
circuit.kill().unwrap();
}
/// Inserting an element once and then deleting it twice causes the trace
/// weight to go negative, which must trigger a panic.
#[test]
fn integral_weight_validator_panics_on_negative_integral() {
let (mut circuit, ()) = Runtime::init_circuit(1, move |circuit| {
let mut step = 0i64;
let source = circuit.add_source(Generator::new(move || {
// Step 0: insert key 42 with weight +1 → weight 1.
// Step 1: delete key 42 with weight -2 → weight -1 (panic).
let batch: OrdZSet<i64> = if step == 0 {
OrdZSet::from_tuples((), vec![Tup2(Tup2(42i64, ()), 1 as ZWeight)])
} else {
OrdZSet::from_tuples((), vec![Tup2(Tup2(42i64, ()), -2 as ZWeight)])
};
step += 1;
batch
}));
source.integral_weight_validator("table 'test'");
Ok(())
})
.unwrap();
circuit.transaction().unwrap(); // step 0: weight 1 (ok)
let err = circuit.transaction().expect_err("expected a panic");
assert!(
err.to_string()
.contains("table 'test': negative weight found"),
"unexpected error: {err}"
);
circuit.kill().unwrap();
}
/// A weight that goes negative (e.g., deleting something never inserted)
/// must also panic.
#[test]
fn integral_weight_validator_panics_on_negative_weight() {
let (mut circuit, ()) = Runtime::init_circuit(1, move |circuit| {
// Emit weight -1 immediately
let source = circuit.add_source(Generator::new(|| zset! { 42i64 => -1 }));
source.integral_weight_validator("table 'test'");
Ok(())
})
.unwrap();
let err = circuit.transaction().expect_err("expected a panic");
assert!(
err.to_string()
.contains("table 'test': negative weight found"),
"unexpected error: {err}"
);
circuit.kill().unwrap();
}
}