Add Versionstamp generator (#3716)
Co-authored-by: Rushmore Mushambi <rushmore@webenchanter.com>
This commit is contained in:
Przemyslaw Hugh Kaznowski
GitHub
parent
ce8e2d4578
commit
2b4fb84511
3 changed files with 245 additions and 345 deletions
|
|
@ -44,6 +44,12 @@ pub mod rpc;
|
|||
#[doc(hidden)]
|
||||
pub mod syn;
|
||||
|
||||
#[doc(hidden)]
|
||||
pub mod test_helpers {
|
||||
pub use crate::vs::conv::to_u128_be;
|
||||
pub use crate::vs::generate_versionstamp_sequences;
|
||||
}
|
||||
|
||||
#[doc(hidden)]
|
||||
/// Channels for receiving a SurrealQL database export
|
||||
pub mod channel {
|
||||
|
|
|
|||
|
|
@ -21,3 +21,91 @@ pub(crate) mod oracle;
|
|||
|
||||
pub use self::conv::*;
|
||||
pub use self::oracle::*;
|
||||
|
||||
/// Generate S-tuples of valid, sequenced versionstamps within range.
|
||||
/// The limit is used, because these are combinatorics - without an upper bound, combinations aren't possible.
|
||||
#[doc(hidden)]
|
||||
pub fn generate_versionstamp_sequences(start: Versionstamp) -> VersionstampSequence {
|
||||
VersionstampSequence {
|
||||
next_state: Some(start),
|
||||
}
|
||||
}
|
||||
|
||||
#[doc(hidden)]
|
||||
pub struct VersionstampSequence {
|
||||
next_state: Option<Versionstamp>,
|
||||
}
|
||||
|
||||
#[doc(hidden)]
|
||||
impl Iterator for VersionstampSequence {
|
||||
type Item = Versionstamp;
|
||||
|
||||
fn next(&mut self) -> Option<Self::Item> {
|
||||
self.next_state?;
|
||||
let returned_state = self.next_state.unwrap();
|
||||
// Now calculate next
|
||||
let mut next_state = self.next_state.unwrap();
|
||||
let index_to_increase =
|
||||
next_state.iter().enumerate().rev().skip(2).find(|(_, &x)| x < 255u8).take();
|
||||
if index_to_increase.is_none() {
|
||||
self.next_state = None;
|
||||
return Some(returned_state);
|
||||
}
|
||||
let (index_to_increase, _) = index_to_increase.unwrap();
|
||||
next_state[index_to_increase] += 1;
|
||||
for next_state_byte in
|
||||
next_state.iter_mut().take(returned_state.len() - 2).skip(index_to_increase + 1)
|
||||
{
|
||||
*next_state_byte = 0;
|
||||
}
|
||||
self.next_state = Some(next_state);
|
||||
Some(returned_state)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod test {
|
||||
use crate::vs::{to_u128_be, Versionstamp};
|
||||
|
||||
#[test]
|
||||
pub fn generate_one_vs() {
|
||||
let vs = super::generate_versionstamp_sequences([0; 10]).take(1).collect::<Vec<_>>();
|
||||
assert_eq!(vs.len(), 1, "Should be 1, but was {:?}", vs);
|
||||
assert_eq!(vs[0], [0; 10]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
pub fn generate_two_vs_in_sequence() {
|
||||
let vs =
|
||||
super::generate_versionstamp_sequences([0, 0, 0, 0, 0, 0, 0, 1, 0, 0]).flat_map(|vs| {
|
||||
let skip_because_first_is_equal = 1;
|
||||
super::generate_versionstamp_sequences(vs)
|
||||
.skip(skip_because_first_is_equal)
|
||||
.map(move |vs2| (vs, vs2))
|
||||
});
|
||||
let versionstamps = vs.take(4).collect::<Vec<(Versionstamp, Versionstamp)>>();
|
||||
|
||||
assert_eq!(
|
||||
versionstamps.len(),
|
||||
4,
|
||||
"We expect the combinations to be 2x2 matrix, but was {:?}",
|
||||
versionstamps
|
||||
);
|
||||
|
||||
let acceptable_values = [65536u128, 131072, 196608, 262144, 327680, 393216];
|
||||
for (first, second) in versionstamps {
|
||||
assert!(first < second, "First: {:?}, Second: {:?}", first, second);
|
||||
let first = to_u128_be(first);
|
||||
let second = to_u128_be(second);
|
||||
assert!(acceptable_values.contains(&first));
|
||||
assert!(acceptable_values.contains(&second));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
pub fn iteration_stops_past_end() {
|
||||
let mut iter = super::generate_versionstamp_sequences([255; 10]);
|
||||
assert!(iter.next().is_some());
|
||||
assert!(iter.next().is_none());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ use surrealdb::kvs::Datastore;
|
|||
use surrealdb::kvs::LockType::Optimistic;
|
||||
use surrealdb::kvs::TransactionType::Write;
|
||||
use surrealdb::sql::Value;
|
||||
use surrealdb_core2::test_helpers::{generate_versionstamp_sequences, to_u128_be};
|
||||
|
||||
mod helpers;
|
||||
|
||||
|
|
@ -64,212 +65,47 @@ async fn database_change_feeds() -> Result<(), Error> {
|
|||
let tmp = res.remove(0).result;
|
||||
assert!(tmp.is_ok());
|
||||
|
||||
// Two timestamps
|
||||
let variance = 4;
|
||||
let first_timestamp = generate_versionstamp_sequences([0; 10]).take(variance);
|
||||
let second_timestamp = first_timestamp.flat_map(|vs1| {
|
||||
generate_versionstamp_sequences(vs1).skip(1).take(variance).map(move |vs2| (vs1, vs2))
|
||||
});
|
||||
|
||||
let potential_show_changes_values: Vec<Value> = match FFLAGS.change_feed_live_queries.enabled()
|
||||
{
|
||||
true => vec![
|
||||
true => second_timestamp
|
||||
.map(|(vs1, vs2)| {
|
||||
let vs1 = to_u128_be(vs1);
|
||||
let vs2 = to_u128_be(vs2);
|
||||
Value::parse(
|
||||
"[
|
||||
{
|
||||
versionstamp: 65536,
|
||||
changes: [
|
||||
{
|
||||
create: {
|
||||
id: person:test,
|
||||
name: 'Name: Tobie'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 131072,
|
||||
changes: [
|
||||
{
|
||||
delete: {
|
||||
id: person:test
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]",
|
||||
),
|
||||
format!(
|
||||
r#"[
|
||||
{{ versionstamp: {}, changes: [ {{ create: {{ id: person:test, name: 'Name: Tobie' }} }} ] }},
|
||||
{{ versionstamp: {}, changes: [ {{ delete: {{ id: person:test }} }} ] }}
|
||||
]"#,
|
||||
vs1, vs2
|
||||
)
|
||||
.as_str(),
|
||||
)
|
||||
})
|
||||
.collect(),
|
||||
false => second_timestamp
|
||||
.map(|(vs1, vs2)| {
|
||||
let vs1 = to_u128_be(vs1);
|
||||
let vs2 = to_u128_be(vs2);
|
||||
Value::parse(
|
||||
"[
|
||||
{
|
||||
versionstamp: 65536,
|
||||
changes: [
|
||||
{
|
||||
create: {
|
||||
id: person:test,
|
||||
name: 'Name: Tobie'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 196608,
|
||||
changes: [
|
||||
{
|
||||
delete: {
|
||||
id: person:test
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]",
|
||||
),
|
||||
Value::parse(
|
||||
"[
|
||||
{
|
||||
versionstamp: 131072,
|
||||
changes: [
|
||||
{
|
||||
create: {
|
||||
id: person:test,
|
||||
name: 'Name: Tobie'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 196608,
|
||||
changes: [
|
||||
{
|
||||
delete: {
|
||||
id: person:test
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]",
|
||||
),
|
||||
Value::parse(
|
||||
"[
|
||||
{
|
||||
versionstamp: 131072,
|
||||
changes: [
|
||||
{
|
||||
create: {
|
||||
id: person:test,
|
||||
name: 'Name: Tobie'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 262144,
|
||||
changes: [
|
||||
{
|
||||
delete: {
|
||||
id: person:test
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]",
|
||||
),
|
||||
],
|
||||
false => vec![
|
||||
Value::parse(
|
||||
"[
|
||||
{
|
||||
versionstamp: 65536,
|
||||
changes: [
|
||||
{
|
||||
update: {
|
||||
id: person:test,
|
||||
name: 'Name: Tobie'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 131072,
|
||||
changes: [
|
||||
{
|
||||
delete: {
|
||||
id: person:test
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]",
|
||||
),
|
||||
Value::parse(
|
||||
"[
|
||||
{
|
||||
versionstamp: 65536,
|
||||
changes: [
|
||||
{
|
||||
update: {
|
||||
id: person:test,
|
||||
name: 'Name: Tobie'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 196608,
|
||||
changes: [
|
||||
{
|
||||
delete: {
|
||||
id: person:test
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]",
|
||||
),
|
||||
Value::parse(
|
||||
"[
|
||||
{
|
||||
versionstamp: 131072,
|
||||
changes: [
|
||||
{
|
||||
update: {
|
||||
id: person:test,
|
||||
name: 'Name: Tobie'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 196608,
|
||||
changes: [
|
||||
{
|
||||
delete: {
|
||||
id: person:test
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]",
|
||||
),
|
||||
Value::parse(
|
||||
"[
|
||||
{
|
||||
versionstamp: 131072,
|
||||
changes: [
|
||||
{
|
||||
update: {
|
||||
id: person:test,
|
||||
name: 'Name: Tobie'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 262144,
|
||||
changes: [
|
||||
{
|
||||
delete: {
|
||||
id: person:test
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]",
|
||||
),
|
||||
],
|
||||
format!(
|
||||
r#"[
|
||||
{{ versionstamp: {}, changes: [ {{ update: {{ id: person:test, name: 'Name: Tobie' }} }} ] }},
|
||||
{{ versionstamp: {}, changes: [ {{ delete: {{ id: person:test }} }} ] }}
|
||||
]"#,
|
||||
vs1, vs2
|
||||
)
|
||||
.as_str(),
|
||||
)
|
||||
})
|
||||
.collect(),
|
||||
};
|
||||
|
||||
// Declare check that is repeatable
|
||||
|
|
@ -309,7 +145,15 @@ async fn database_change_feeds() -> Result<(), Error> {
|
|||
.find(|x| *x == &tmp)
|
||||
// We actually dont want to capture if its found
|
||||
.map(|_v| ())
|
||||
.ok_or(format!("Expected SHOW CHANGES value not found:\n{}", tmp))?;
|
||||
.ok_or(format!(
|
||||
"Expected SHOW CHANGES value not found:\n{}\nin:\n{}",
|
||||
tmp,
|
||||
cf_val_arr
|
||||
.iter()
|
||||
.map(|vs| vs.to_string())
|
||||
.reduce(|left, right| format!("{}\n{}", left, right))
|
||||
.unwrap()
|
||||
))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
@ -351,7 +195,7 @@ async fn database_change_feeds() -> Result<(), Error> {
|
|||
let res = &mut dbs.execute(sql, &ses, None).await?;
|
||||
let tmp = res.remove(0).result?;
|
||||
let val = Value::parse("[]");
|
||||
assert_eq!(tmp, val);
|
||||
assert_eq!(val, tmp);
|
||||
//
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -450,145 +294,98 @@ async fn table_change_feeds() -> Result<(), Error> {
|
|||
let _tmp = res.remove(0).result?;
|
||||
// SHOW CHANGES
|
||||
let tmp = res.remove(0).result?;
|
||||
let val = match FFLAGS.change_feed_live_queries.enabled() {
|
||||
true => Value::parse(
|
||||
"[
|
||||
{
|
||||
versionstamp: 65536,
|
||||
changes: [
|
||||
{
|
||||
define_table: {
|
||||
name: 'person'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 131072,
|
||||
changes: [
|
||||
{
|
||||
create: {
|
||||
id: person:test,
|
||||
name: 'Name: Tobie'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 196608,
|
||||
changes: [
|
||||
{
|
||||
update: {
|
||||
id: person:test,
|
||||
name: 'Name: Jaime'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 262144,
|
||||
changes: [
|
||||
{
|
||||
update: {
|
||||
id: person:test,
|
||||
name: 'Name: Tobie'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 327680,
|
||||
changes: [
|
||||
{
|
||||
delete: {
|
||||
id: person:test
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 393216,
|
||||
changes: [
|
||||
{
|
||||
create: {
|
||||
id: person:1000,
|
||||
name: 'Name: Yusuke'
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]",
|
||||
),
|
||||
false => Value::parse(
|
||||
"[
|
||||
{
|
||||
versionstamp: 65536,
|
||||
changes: [
|
||||
{
|
||||
define_table: {
|
||||
name: 'person'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 131072,
|
||||
changes: [
|
||||
{
|
||||
update: {
|
||||
id: person:test,
|
||||
name: 'Name: Tobie'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 196608,
|
||||
changes: [
|
||||
{
|
||||
update: {
|
||||
id: person:test,
|
||||
name: 'Name: Jaime'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 262144,
|
||||
changes: [
|
||||
{
|
||||
update: {
|
||||
id: person:test,
|
||||
name: 'Name: Tobie'
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 327680,
|
||||
changes: [
|
||||
{
|
||||
delete: {
|
||||
id: person:test
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
versionstamp: 393216,
|
||||
changes: [
|
||||
{
|
||||
update: {
|
||||
id: person:1000,
|
||||
name: 'Name: Yusuke'
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]",
|
||||
),
|
||||
// If you want to write a macro, you are welcome to
|
||||
let limit_variance = 3;
|
||||
let first = generate_versionstamp_sequences([0; 10]).take(limit_variance);
|
||||
let second = first.flat_map(|vs1| {
|
||||
generate_versionstamp_sequences(vs1).take(limit_variance).skip(1).map(move |vs2| (vs1, vs2))
|
||||
});
|
||||
let third = second.flat_map(|(vs1, vs2)| {
|
||||
generate_versionstamp_sequences(vs2)
|
||||
.take(limit_variance)
|
||||
.skip(1)
|
||||
.map(move |vs3| (vs1, vs2, vs3))
|
||||
});
|
||||
let fourth = third.flat_map(|(vs1, vs2, vs3)| {
|
||||
generate_versionstamp_sequences(vs3)
|
||||
.take(limit_variance)
|
||||
.skip(1)
|
||||
.map(move |vs4| (vs1, vs2, vs3, vs4))
|
||||
});
|
||||
let fifth = fourth.flat_map(|(vs1, vs2, vs3, vs4)| {
|
||||
generate_versionstamp_sequences(vs4)
|
||||
.take(limit_variance)
|
||||
.skip(1)
|
||||
.map(move |vs5| (vs1, vs2, vs3, vs4, vs5))
|
||||
});
|
||||
let sixth = fifth.flat_map(|(vs1, vs2, vs3, vs4, vs5)| {
|
||||
generate_versionstamp_sequences(vs5)
|
||||
.take(limit_variance)
|
||||
.skip(1)
|
||||
.map(move |vs6| (vs1, vs2, vs3, vs4, vs5, vs6))
|
||||
});
|
||||
let allowed_values: Vec<Value> = match FFLAGS.change_feed_live_queries.enabled() {
|
||||
true => sixth
|
||||
.map(|(vs1, vs2, vs3, vs4, vs5, vs6)| {
|
||||
let (vs1, vs2, vs3, vs4, vs5, vs6) = (
|
||||
to_u128_be(vs1),
|
||||
to_u128_be(vs2),
|
||||
to_u128_be(vs3),
|
||||
to_u128_be(vs4),
|
||||
to_u128_be(vs5),
|
||||
to_u128_be(vs6),
|
||||
);
|
||||
Value::parse(
|
||||
format!(
|
||||
r#"[
|
||||
{{ versionstamp: {vs1}, changes: [ {{ define_table: {{ name: 'person' }} }} ] }},
|
||||
{{ versionstamp: {vs2}, changes: [ {{ create: {{ id: person:test, name: 'Name: Tobie' }} }} ] }},
|
||||
{{ versionstamp: {vs3}, changes: [ {{ update: {{ id: person:test, name: 'Name: Jaime' }} }} ] }},
|
||||
{{ versionstamp: {vs4}, changes: [ {{ update: {{ id: person:test, name: 'Name: Tobie' }} }} ] }},
|
||||
{{ versionstamp: {vs5}, changes: [ {{ delete: {{ id: person:test }} }} ] }},
|
||||
{{ versionstamp: {vs6}, changes: [ {{ create: {{ id: person:1000, name: 'Name: Yusuke' }} }} ] }}
|
||||
]"#,
|
||||
)
|
||||
.as_str(),
|
||||
)
|
||||
})
|
||||
.collect(),
|
||||
false => sixth
|
||||
.map(|(vs1, vs2, vs3, vs4, vs5, vs6)| {
|
||||
let (vs1, vs2, vs3, vs4, vs5, vs6) = (
|
||||
to_u128_be(vs1),
|
||||
to_u128_be(vs2),
|
||||
to_u128_be(vs3),
|
||||
to_u128_be(vs4),
|
||||
to_u128_be(vs5),
|
||||
to_u128_be(vs6),
|
||||
);
|
||||
Value::parse(
|
||||
format!(
|
||||
r#"[
|
||||
{{ versionstamp: {vs1}, changes: [ {{ define_table: {{ name: 'person' }} }} ] }},
|
||||
{{ versionstamp: {vs2}, changes: [ {{ update: {{ id: person:test, name: 'Name: Tobie' }} }} ] }},
|
||||
{{ versionstamp: {vs3}, changes: [ {{ update: {{ id: person:test, name: 'Name: Jaime' }} }} ] }},
|
||||
{{ versionstamp: {vs4}, changes: [ {{ update: {{ id: person:test, name: 'Name: Tobie' }} }} ] }},
|
||||
{{ versionstamp: {vs5}, changes: [ {{ delete: {{ id: person:test }} }} ] }},
|
||||
{{ versionstamp: {vs6}, changes: [ {{ update: {{ id: person:1000, name: 'Name: Yusuke' }} }} ] }}
|
||||
]"#
|
||||
)
|
||||
.as_str(),
|
||||
)
|
||||
})
|
||||
.collect(),
|
||||
};
|
||||
assert_eq!(tmp, val);
|
||||
assert!(
|
||||
allowed_values.contains(&tmp),
|
||||
"tmp:\n{}\nchecked:\n{}",
|
||||
tmp,
|
||||
allowed_values
|
||||
.iter()
|
||||
.map(|v| v.to_string())
|
||||
.reduce(|a, b| format!("{}\n{}", a, b))
|
||||
.unwrap()
|
||||
);
|
||||
// Retain for 1h
|
||||
let sql = "
|
||||
SHOW CHANGES FOR TABLE person SINCE 0;
|
||||
|
|
@ -596,7 +393,16 @@ async fn table_change_feeds() -> Result<(), Error> {
|
|||
dbs.tick_at(end_ts + 3599).await?;
|
||||
let res = &mut dbs.execute(sql, &ses, None).await?;
|
||||
let tmp = res.remove(0).result?;
|
||||
assert_eq!(tmp, val);
|
||||
assert!(
|
||||
allowed_values.contains(&tmp),
|
||||
"tmp:\n{}\nchecked:\n{}",
|
||||
tmp,
|
||||
allowed_values
|
||||
.iter()
|
||||
.map(|v| v.to_string())
|
||||
.reduce(|a, b| format!("{}\n{}", a, b))
|
||||
.unwrap()
|
||||
);
|
||||
// GC after 1hs
|
||||
dbs.tick_at(end_ts + 3600).await?;
|
||||
let res = &mut dbs.execute(sql, &ses, None).await?;
|
||||
|
|
|
|||
Loading…
Reference in a new issue