aljaz/turso
1
0
Fork 0
mirror of https://github.com/aljazceru/turso.git synced 2026-02-23 17:05:36 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge 'Simulator Multiple Connections' from Pedro Muniz

Browse Source 
This is the start of the work to add transactional properties to the
Simulator. Before this PR, we only executed the queries on one
connection, now the default is to run with 10 connections. Each
connection is selected at random to run the database queries.
A lot of the code was hardcoded to work with 1 connection only, so most
of the refactor came from me adjusting the `Interaction` and
`Interactions` structs to hold a connection index. As I had to refactor
a lot of code, I took the opportunity to deduplicate code that across
`run_simulation_default`, `watch::run_simulation`,
`differential::run_simulation`, `doublecheck::run_simulation`.
In my next PR, I intend to add transaction isolation checking which will
involve some more refactoring to ensure the simulator correctly tracks
transaction state and table visibility.
And in a future PR, we can extend this further with concurrent
transactions as well

Closes #3163
This commit is contained in:
Jussi Saurio
2025-09-18 09:10:01 +03:00
committed by GitHub
parent d2376a239a 27a8f019c6
commit 26f56d794e
11 changed files with 1122 additions and 1257 deletions
Download Patch File Download Diff File Expand all files Collapse all files

401
simulator/generation/plan.rs
View File

@@ -1,6 +1,8 @@
use std::{
fmt::{Debug, Display},
ops::{Deref, DerefMut},
path::Path,
rc::Rc,
sync::Arc,
vec,
};
@@ -40,7 +42,7 @@ impl InteractionPlan {
/// delete interactions from the human readable file, and this function uses the JSON file as
/// a baseline to detect with interactions were deleted and constructs the plan from the
/// remaining interactions.
pub(crate) fn compute_via_diff(plan_path: &Path) -> Vec<Vec<Interaction>> {
pub(crate) fn compute_via_diff(plan_path: &Path) -> Vec<Interaction> {
let interactions = std::fs::read_to_string(plan_path).unwrap();
let interactions = interactions.lines().collect::<Vec<_>>();
@@ -95,18 +97,73 @@ impl InteractionPlan {
}
}
let _ = plan.split_off(j);
plan
plan.into_iter().flatten().collect()
}
pub fn interactions_list(&self) -> Vec<Interaction> {
self.plan
.clone()
.into_iter()
.flat_map(|interactions| interactions.interactions().into_iter())
.collect()
}
pub fn interactions_list_with_secondary_index(&self) -> Vec<(usize, Interaction)> {
self.plan
.clone()
.into_iter()
.enumerate()
.flat_map(|(idx, interactions)| {
interactions
.interactions()
.into_iter()
.map(move |interaction| (idx, interaction))
})
.collect()
}
}
pub(crate) struct InteractionPlanState {
pub(crate) stack: Vec<ResultSet>,
pub(crate) interaction_pointer: usize,
pub(crate) secondary_pointer: usize,
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub struct InteractionPlanState {
pub interaction_pointer: usize,
}
#[derive(Debug, Default, Clone)]
pub struct ConnectionState {
pub stack: Vec<ResultSet>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) enum Interactions {
pub struct Interactions {
pub connection_index: usize,
pub interactions: InteractionsType,
}
impl Interactions {
pub fn new(connection_index: usize, interactions: InteractionsType) -> Self {
Self {
connection_index,
interactions,
}
}
}
impl Deref for Interactions {
type Target = InteractionsType;
fn deref(&self) -> &Self::Target {
&self.interactions
}
}
impl DerefMut for Interactions {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.interactions
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum InteractionsType {
Property(Property),
Query(Query),
Fault(Fault),
@@ -116,130 +173,101 @@ impl Shadow for Interactions {
type Result = ();
fn shadow(&self, tables: &mut SimulatorTables) {
match self {
Interactions::Property(property) => {
match &self.interactions {
InteractionsType::Property(property) => {
let initial_tables = tables.clone();
let mut is_error = false;
for interaction in property.interactions() {
match interaction {
Interaction::Query(query) | Interaction::FsyncQuery(query) => {
if query.shadow(tables).is_err() {
is_error = true;
}
}
Interaction::FaultyQuery(..) => {}
Interaction::Assertion(_) => {}
Interaction::Assumption(_) => {}
Interaction::Fault(_) => {}
}
if is_error {
for interaction in property.interactions(self.connection_index) {
let res = interaction.shadow(tables);
if res.is_err() {
// If any interaction fails, we reset the tables to the initial state
*tables = initial_tables.clone();
break;
}
}
}
Interactions::Query(query) => {
InteractionsType::Query(query) => {
let _ = query.shadow(tables);
}
Interactions::Fault(_) => {}
InteractionsType::Fault(_) => {}
}
}
}
impl Interactions {
pub(crate) fn name(&self) -> Option<&str> {
match self {
Interactions::Property(property) => Some(property.name()),
Interactions::Query(_) => None,
Interactions::Fault(_) => None,
match &self.interactions {
InteractionsType::Property(property) => Some(property.name()),
InteractionsType::Query(_) => None,
InteractionsType::Fault(_) => None,
}
}
pub(crate) fn interactions(&self) -> Vec<Interaction> {
match self {
Interactions::Property(property) => property.interactions(),
Interactions::Query(query) => vec![Interaction::Query(query.clone())],
Interactions::Fault(fault) => vec![Interaction::Fault(fault.clone())],
match &self.interactions {
InteractionsType::Property(property) => property.interactions(self.connection_index),
InteractionsType::Query(query) => vec![Interaction::new(
self.connection_index,
InteractionType::Query(query.clone()),
)],
InteractionsType::Fault(fault) => vec![Interaction::new(
self.connection_index,
InteractionType::Fault(*fault),
)],
}
}
}
impl Interactions {
pub(crate) fn dependencies(&self) -> IndexSet<String> {
match self {
Interactions::Property(property) => {
property
.interactions()
.iter()
.fold(IndexSet::new(), |mut acc, i| match i {
Interaction::Query(q) => {
acc.extend(q.dependencies());
acc
}
_ => acc,
})
}
Interactions::Query(query) => query.dependencies(),
Interactions::Fault(_) => IndexSet::new(),
match &self.interactions {
InteractionsType::Property(property) => property
.interactions(self.connection_index)
.iter()
.fold(IndexSet::new(), |mut acc, i| match &i.interaction {
InteractionType::Query(q) => {
acc.extend(q.dependencies());
acc
}
_ => acc,
}),
InteractionsType::Query(query) => query.dependencies(),
InteractionsType::Fault(_) => IndexSet::new(),
}
}
pub(crate) fn uses(&self) -> Vec<String> {
match self {
Interactions::Property(property) => {
property
.interactions()
.iter()
.fold(vec![], |mut acc, i| match i {
Interaction::Query(q) => {
acc.extend(q.uses());
acc
}
_ => acc,
})
}
Interactions::Query(query) => query.uses(),
Interactions::Fault(_) => vec![],
match &self.interactions {
InteractionsType::Property(property) => property
.interactions(self.connection_index)
.iter()
.fold(vec![], |mut acc, i| match &i.interaction {
InteractionType::Query(q) => {
acc.extend(q.uses());
acc
}
_ => acc,
}),
InteractionsType::Query(query) => query.uses(),
InteractionsType::Fault(_) => vec![],
}
}
}
// FIXME: for the sql display come back and add connection index as a comment
impl Display for InteractionPlan {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
for interactions in &self.plan {
match interactions {
Interactions::Property(property) => {
match &interactions.interactions {
InteractionsType::Property(property) => {
let name = property.name();
writeln!(f, "-- begin testing '{name}'")?;
for interaction in property.interactions() {
write!(f, "\t")?;
match interaction {
Interaction::Query(query) => writeln!(f, "{query};")?,
Interaction::Assumption(assumption) => {
writeln!(f, "-- ASSUME {};", assumption.name)?
}
Interaction::Assertion(assertion) => {
writeln!(f, "-- ASSERT {};", assertion.name)?
}
Interaction::Fault(fault) => writeln!(f, "-- FAULT '{fault}';")?,
Interaction::FsyncQuery(query) => {
writeln!(f, "-- FSYNC QUERY;")?;
writeln!(f, "{query};")?;
writeln!(f, "{query};")?
}
Interaction::FaultyQuery(query) => {
writeln!(f, "{query}; -- FAULTY QUERY")?
}
}
for interaction in property.interactions(interactions.connection_index) {
writeln!(f, "\t{interaction}")?;
}
writeln!(f, "-- end testing '{name}'")?;
}
Interactions::Fault(fault) => {
InteractionsType::Fault(fault) => {
writeln!(f, "-- FAULT '{fault}'")?;
}
Interactions::Query(query) => {
InteractionsType::Query(query) => {
writeln!(f, "{query};")?;
}
}
@@ -293,40 +321,16 @@ impl Display for InteractionStats {
}
}
#[derive(Debug)]
pub(crate) enum Interaction {
Query(Query),
Assumption(Assertion),
Assertion(Assertion),
Fault(Fault),
/// Will attempt to run any random query. However, when the connection tries to sync it will
/// close all connections and reopen the database and assert that no data was lost
FsyncQuery(Query),
FaultyQuery(Query),
}
impl Display for Interaction {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Query(query) => write!(f, "{query}"),
Self::Assumption(assumption) => write!(f, "ASSUME {}", assumption.name),
Self::Assertion(assertion) => write!(f, "ASSERT {}", assertion.name),
Self::Fault(fault) => write!(f, "FAULT '{fault}'"),
Self::FsyncQuery(query) => write!(f, "{query}"),
Self::FaultyQuery(query) => write!(f, "{query}; -- FAULTY QUERY"),
}
}
}
type AssertionFunc = dyn Fn(&Vec<ResultSet>, &mut SimulatorEnv) -> Result<Result<(), String>>;
enum AssertionAST {
Pick(),
}
pub(crate) struct Assertion {
pub(crate) func: Box<AssertionFunc>,
pub(crate) name: String, // For display purposes in the plan
#[derive(Clone)]
pub struct Assertion {
pub func: Rc<AssertionFunc>,
pub name: String, // For display purposes in the plan
}
impl Debug for Assertion {
@@ -337,7 +341,19 @@ impl Debug for Assertion {
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
impl Assertion {
pub fn new<F>(name: String, func: F) -> Self
where
F: Fn(&Vec<ResultSet>, &mut SimulatorEnv) -> Result<Result<(), String>> + 'static,
{
Self {
func: Rc::new(func),
name,
}
}
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
pub(crate) enum Fault {
Disconnect,
ReopenDatabase,
@@ -386,18 +402,18 @@ impl InteractionPlan {
}
}
for interactions in &self.plan {
match interactions {
Interactions::Property(property) => {
for interaction in &property.interactions() {
if let Interaction::Query(query) = interaction {
match &interactions.interactions {
InteractionsType::Property(property) => {
for interaction in &property.interactions(interactions.connection_index) {
if let InteractionType::Query(query) = &interaction.interaction {
query_stat(query, &mut stats);
}
}
}
Interactions::Query(query) => {
InteractionsType::Query(query) => {
query_stat(query, &mut stats);
}
Interactions::Fault(_) => {}
InteractionsType::Fault(_) => {}
}
}
@@ -413,8 +429,11 @@ impl InteractionPlan {
let create_query = Create::arbitrary(rng, env);
env.tables.push(create_query.table.clone());
plan.plan
.push(Interactions::Query(Query::Create(create_query)));
// initial query starts at 0th connection
plan.plan.push(Interactions::new(
0,
InteractionsType::Query(Query::Create(create_query)),
));
while plan.plan.len() < num_interactions {
tracing::debug!(
@@ -432,7 +451,74 @@ impl InteractionPlan {
}
}
impl Shadow for Interaction {
#[derive(Debug, Clone)]
pub struct Interaction {
pub connection_index: usize,
pub interaction: InteractionType,
}
impl Deref for Interaction {
type Target = InteractionType;
fn deref(&self) -> &Self::Target {
&self.interaction
}
}
impl DerefMut for Interaction {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.interaction
}
}
impl Interaction {
pub fn new(connection_index: usize, interaction: InteractionType) -> Self {
Self {
connection_index,
interaction,
}
}
}
#[derive(Debug, Clone)]
pub enum InteractionType {
Query(Query),
Assumption(Assertion),
Assertion(Assertion),
Fault(Fault),
/// Will attempt to run any random query. However, when the connection tries to sync it will
/// close all connections and reopen the database and assert that no data was lost
FsyncQuery(Query),
FaultyQuery(Query),
}
// FIXME: add the connection index here later
impl Display for Interaction {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.interaction)
}
}
impl Display for InteractionType {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Query(query) => write!(f, "{query}"),
Self::Assumption(assumption) => write!(f, "-- ASSUME {};", assumption.name),
Self::Assertion(assertion) => {
write!(f, "-- ASSERT {};", assertion.name)
}
Self::Fault(fault) => write!(f, "-- FAULT '{fault}';"),
Self::FsyncQuery(query) => {
writeln!(f, "-- FSYNC QUERY;")?;
writeln!(f, "{query};")?;
write!(f, "{query};")
}
Self::FaultyQuery(query) => write!(f, "{query}; -- FAULTY QUERY"),
}
}
}
impl Shadow for InteractionType {
type Result = anyhow::Result<Vec<Vec<SimValue>>>;
fn shadow(&self, env: &mut SimulatorTables) -> Self::Result {
match self {
@@ -448,7 +534,7 @@ impl Shadow for Interaction {
}
}
}
impl Interaction {
impl InteractionType {
pub(crate) fn execute_query(&self, conn: &mut Arc<Connection>) -> ResultSet {
if let Self::Query(query) = self {
let query_str = query.to_string();
@@ -460,9 +546,7 @@ impl Interaction {
&query_str[0..query_str.len().min(4096)],
err
);
if let Some(turso_core::LimboError::ParseError(e)) = err {
panic!("Unexpected parse error: {e}");
}
// Do not panic on parse error, because DoubleCreateFailure relies on it
return Err(err.unwrap());
}
let rows = rows?;
@@ -755,31 +839,52 @@ fn random_create<R: rand::Rng>(rng: &mut R, env: &SimulatorEnv) -> Interactions
while env.tables.iter().any(|t| t.name == create.table.name) {
create = Create::arbitrary(rng, env);
}
Interactions::Query(Query::Create(create))
Interactions::new(
env.choose_conn(rng),
InteractionsType::Query(Query::Create(create)),
)
}
fn random_read<R: rand::Rng>(rng: &mut R, env: &SimulatorEnv) -> Interactions {
Interactions::Query(Query::Select(Select::arbitrary(rng, env)))
Interactions::new(
env.choose_conn(rng),
InteractionsType::Query(Query::Select(Select::arbitrary(rng, env))),
)
}
fn random_expr<R: rand::Rng>(rng: &mut R, env: &SimulatorEnv) -> Interactions {
Interactions::Query(Query::Select(SelectFree::arbitrary(rng, env).0))
Interactions::new(
env.choose_conn(rng),
InteractionsType::Query(Query::Select(SelectFree::arbitrary(rng, env).0)),
)
}
fn random_write<R: rand::Rng>(rng: &mut R, env: &SimulatorEnv) -> Interactions {
Interactions::Query(Query::Insert(Insert::arbitrary(rng, env)))
Interactions::new(
env.choose_conn(rng),
InteractionsType::Query(Query::Insert(Insert::arbitrary(rng, env))),
)
}
fn random_delete<R: rand::Rng>(rng: &mut R, env: &SimulatorEnv) -> Interactions {
Interactions::Query(Query::Delete(Delete::arbitrary(rng, env)))
Interactions::new(
env.choose_conn(rng),
InteractionsType::Query(Query::Delete(Delete::arbitrary(rng, env))),
)
}
fn random_update<R: rand::Rng>(rng: &mut R, env: &SimulatorEnv) -> Interactions {
Interactions::Query(Query::Update(Update::arbitrary(rng, env)))
Interactions::new(
env.choose_conn(rng),
InteractionsType::Query(Query::Update(Update::arbitrary(rng, env))),
)
}
fn random_drop<R: rand::Rng>(rng: &mut R, env: &SimulatorEnv) -> Interactions {
Interactions::Query(Query::Drop(Drop::arbitrary(rng, env)))
Interactions::new(
env.choose_conn(rng),
InteractionsType::Query(Query::Drop(Drop::arbitrary(rng, env))),
)
}
fn random_create_index<R: rand::Rng>(rng: &mut R, env: &SimulatorEnv) -> Option<Interactions> {
@@ -799,7 +904,10 @@ fn random_create_index<R: rand::Rng>(rng: &mut R, env: &SimulatorEnv) -> Option<
create_index = CreateIndex::arbitrary(rng, env);
}
Some(Interactions::Query(Query::CreateIndex(create_index)))
Some(Interactions::new(
env.choose_conn(rng),
InteractionsType::Query(Query::CreateIndex(create_index)),
))
}
fn random_fault<R: rand::Rng>(rng: &mut R, env: &SimulatorEnv) -> Interactions {
@@ -808,8 +916,8 @@ fn random_fault<R: rand::Rng>(rng: &mut R, env: &SimulatorEnv) -> Interactions {
} else {
vec![Fault::Disconnect, Fault::ReopenDatabase]
};
let fault = faults[rng.random_range(0..faults.len())].clone();
Interactions::Fault(fault)
let fault = faults[rng.random_range(0..faults.len())];
Interactions::new(env.choose_conn(rng), InteractionsType::Fault(fault))
}
impl ArbitraryFrom<(&SimulatorEnv, InteractionStats)> for Interactions {
@@ -824,7 +932,14 @@ impl ArbitraryFrom<(&SimulatorEnv, InteractionStats)> for Interactions {
(
u32::min(remaining_.select, remaining_.insert) + remaining_.create,
Box::new(|rng: &mut R| {
Interactions::Property(Property::arbitrary_from(rng, env, (env, &stats)))
Interactions::new(
env.choose_conn(rng),
InteractionsType::Property(Property::arbitrary_from(
rng,
env,
(env, &stats),
)),
)
}),
),
(

393
simulator/generation/property.rs
View File

@@ -16,7 +16,10 @@ use turso_core::{LimboError, types};
use turso_parser::ast::{self, Distinctness};
use crate::{
generation::Shadow as _, model::Query, profiles::query::QueryProfile, runner::env::SimulatorEnv,
generation::{Shadow as _, plan::InteractionType},
model::Query,
profiles::query::QueryProfile,
runner::env::SimulatorEnv,
};
use super::plan::{Assertion, Interaction, InteractionStats, ResultSet};
@@ -215,30 +218,30 @@ impl Property {
/// interactions construct a list of interactions, which is an executable representation of the property.
/// the requirement of property -> vec<interaction> conversion emerges from the need to serialize the property,
/// and `interaction` cannot be serialized directly.
pub(crate) fn interactions(&self) -> Vec<Interaction> {
pub(crate) fn interactions(&self, connection_index: usize) -> Vec<Interaction> {
match self {
Property::TableHasExpectedContent { table } => {
let table = table.to_string();
let table_name = table.clone();
let assumption = Interaction::Assumption(Assertion {
name: format!("table {} exists", table.clone()),
func: Box::new(move |_: &Vec<ResultSet>, env: &mut SimulatorEnv| {
let assumption = InteractionType::Assumption(Assertion::new(
format!("table {} exists", table.clone()),
move |_: &Vec<ResultSet>, env: &mut SimulatorEnv| {
if env.tables.iter().any(|t| t.name == table_name) {
Ok(Ok(()))
} else {
Ok(Err(format!("table {table_name} does not exist")))
}
}),
});
},
));
let select_interaction = Interaction::Query(Query::Select(Select::simple(
let select_interaction = InteractionType::Query(Query::Select(Select::simple(
table.clone(),
Predicate::true_(),
)));
let assertion = Interaction::Assertion(Assertion {
name: format!("table {} should have the expected content", table.clone()),
func: Box::new(move |stack: &Vec<ResultSet>, env| {
let assertion = InteractionType::Assertion(Assertion::new(
format!("table {} should have the expected content", table.clone()),
move |stack: &Vec<ResultSet>, env| {
let rows = stack.last().unwrap();
let Ok(rows) = rows else {
return Ok(Err(format!("expected rows but got error: {rows:?}")));
@@ -266,37 +269,41 @@ impl Property {
}
}
Ok(Ok(()))
}),
});
},
));
vec![assumption, select_interaction, assertion]
vec![
Interaction::new(connection_index, assumption),
Interaction::new(connection_index, select_interaction),
Interaction::new(connection_index, assertion),
]
}
Property::ReadYourUpdatesBack { update, select } => {
let table = update.table().to_string();
let assumption = Interaction::Assumption(Assertion {
name: format!("table {} exists", table.clone()),
func: Box::new(move |_: &Vec<ResultSet>, env: &mut SimulatorEnv| {
let assumption = InteractionType::Assumption(Assertion::new(
format!("table {} exists", table.clone()),
move |_: &Vec<ResultSet>, env: &mut SimulatorEnv| {
if env.tables.iter().any(|t| t.name == table.clone()) {
Ok(Ok(()))
} else {
Ok(Err(format!("table {} does not exist", table.clone())))
}
}),
});
},
));
let update_interaction = Interaction::Query(Query::Update(update.clone()));
let select_interaction = Interaction::Query(Query::Select(select.clone()));
let update_interaction = InteractionType::Query(Query::Update(update.clone()));
let select_interaction = InteractionType::Query(Query::Select(select.clone()));
let update = update.clone();
let table = update.table().to_string();
let assertion = Interaction::Assertion(Assertion {
name: format!(
let assertion = InteractionType::Assertion(Assertion::new(
format!(
"updated rows should be found and have the updated values for table {}",
table.clone()
),
func: Box::new(move |stack: &Vec<ResultSet>, _| {
move |stack: &Vec<ResultSet>, _| {
let rows = stack.last().unwrap();
match rows {
Ok(rows) => {
@@ -314,14 +321,14 @@ impl Property {
}
Err(err) => Err(LimboError::InternalError(err.to_string())),
}
}),
});
},
));
vec![
assumption,
update_interaction,
select_interaction,
assertion,
Interaction::new(connection_index, assumption),
Interaction::new(connection_index, update_interaction),
Interaction::new(connection_index, select_interaction),
Interaction::new(connection_index, assertion),
]
}
Property::InsertValuesSelect {
@@ -348,9 +355,9 @@ impl Property {
let row = values[*row_index].clone();
// Assume that the table exists
let assumption = Interaction::Assumption(Assertion {
name: format!("table {} exists", insert.table()),
func: Box::new({
let assumption = InteractionType::Assumption(Assertion::new(
format!("table {} exists", insert.table()),
{
let table_name = table.clone();
move |_: &Vec<ResultSet>, env: &mut SimulatorEnv| {
if env.tables.iter().any(|t| t.name == table_name) {
@@ -359,11 +366,11 @@ impl Property {
Ok(Err(format!("table {table_name} does not exist")))
}
}
}),
});
},
));
let assertion = Interaction::Assertion(Assertion {
name: format!(
let assertion = InteractionType::Assertion(Assertion::new(
format!(
"row [{:?}] should be found in table {}, interactive={} commit={}, rollback={}",
row.iter().map(|v| v.to_string()).collect::<Vec<String>>(),
insert.table(),
@@ -377,7 +384,7 @@ impl Property {
.map(|i| !i.end_with_commit)
.unwrap_or(false),
),
func: Box::new(move |stack: &Vec<ResultSet>, _| {
move |stack: &Vec<ResultSet>, _| {
let rows = stack.last().unwrap();
match rows {
Ok(rows) => {
@@ -393,43 +400,50 @@ impl Property {
}
Err(err) => Err(LimboError::InternalError(err.to_string())),
}
}),
});
},
));
let mut interactions = Vec::new();
interactions.push(assumption);
interactions.push(Interaction::Query(Query::Insert(insert.clone())));
interactions.extend(queries.clone().into_iter().map(Interaction::Query));
interactions.push(Interaction::Query(Query::Select(select.clone())));
interactions.push(assertion);
interactions.push(Interaction::new(connection_index, assumption));
interactions.push(Interaction::new(
connection_index,
InteractionType::Query(Query::Insert(insert.clone())),
));
interactions.extend(
queries
.clone()
.into_iter()
.map(|q| Interaction::new(connection_index, InteractionType::Query(q))),
);
interactions.push(Interaction::new(
connection_index,
InteractionType::Query(Query::Select(select.clone())),
));
interactions.push(Interaction::new(connection_index, assertion));
interactions
}
Property::DoubleCreateFailure { create, queries } => {
let table_name = create.table.name.clone();
let assumption = Interaction::Assumption(Assertion {
name: "Double-Create-Failure should not be called on an existing table"
.to_string(),
func: Box::new(move |_: &Vec<ResultSet>, env: &mut SimulatorEnv| {
let assumption = InteractionType::Assumption(Assertion::new(
"Double-Create-Failure should not be called on an existing table".to_string(),
move |_: &Vec<ResultSet>, env: &mut SimulatorEnv| {
if !env.tables.iter().any(|t| t.name == table_name) {
Ok(Ok(()))
} else {
Ok(Err(format!("table {table_name} already exists")))
}
}),
});
},
));
let cq1 = Interaction::Query(Query::Create(create.clone()));
let cq2 = Interaction::Query(Query::Create(create.clone()));
let cq1 = InteractionType::Query(Query::Create(create.clone()));
let cq2 = InteractionType::Query(Query::Create(create.clone()));
let table_name = create.table.name.clone();
let assertion = Interaction::Assertion(Assertion {
name:
"creating two tables with the name should result in a failure for the second query"
.to_string(),
func: Box::new(move |stack: &Vec<ResultSet>, _| {
let assertion = InteractionType::Assertion(Assertion::new("creating two tables with the name should result in a failure for the second query"
.to_string(), move |stack: &Vec<ResultSet>, _| {
let last = stack.last().unwrap();
match last {
Ok(success) => Ok(Err(format!("expected table creation to fail but it succeeded: {success:?}"))),
@@ -441,21 +455,25 @@ impl Property {
}
}
}
}),
});
}) );
let mut interactions = Vec::new();
interactions.push(assumption);
interactions.push(cq1);
interactions.extend(queries.clone().into_iter().map(Interaction::Query));
interactions.push(cq2);
interactions.push(assertion);
interactions.push(Interaction::new(connection_index, assumption));
interactions.push(Interaction::new(connection_index, cq1));
interactions.extend(
queries
.clone()
.into_iter()
.map(|q| Interaction::new(connection_index, InteractionType::Query(q))),
);
interactions.push(Interaction::new(connection_index, cq2));
interactions.push(Interaction::new(connection_index, assertion));
interactions
}
Property::SelectLimit { select } => {
let assumption = Interaction::Assumption(Assertion {
name: format!(
let assumption = InteractionType::Assumption(Assertion::new(
format!(
"table ({}) exists",
select
.dependencies()
@@ -463,7 +481,7 @@ impl Property {
.collect::<Vec<_>>()
.join(", ")
),
func: Box::new({
{
let table_name = select.dependencies();
move |_: &Vec<ResultSet>, env: &mut SimulatorEnv| {
if table_name
@@ -479,16 +497,16 @@ impl Property {
Ok(Err(format!("missing tables: {missing_tables:?}")))
}
}
}),
});
},
));
let limit = select
.limit
.expect("Property::SelectLimit without a LIMIT clause");
let assertion = Interaction::Assertion(Assertion {
name: "select query should respect the limit clause".to_string(),
func: Box::new(move |stack: &Vec<ResultSet>, _| {
let assertion = InteractionType::Assertion(Assertion::new(
"select query should respect the limit clause".to_string(),
move |stack: &Vec<ResultSet>, _| {
let last = stack.last().unwrap();
match last {
Ok(rows) => {
@@ -504,13 +522,16 @@ impl Property {
}
Err(_) => Ok(Ok(())),
}
}),
});
},
));
vec![
assumption,
Interaction::Query(Query::Select(select.clone())),
assertion,
Interaction::new(connection_index, assumption),
Interaction::new(
connection_index,
InteractionType::Query(Query::Select(select.clone())),
),
Interaction::new(connection_index, assertion),
]
}
Property::DeleteSelect {
@@ -518,9 +539,9 @@ impl Property {
predicate,
queries,
} => {
let assumption = Interaction::Assumption(Assertion {
name: format!("table {table} exists"),
func: Box::new({
let assumption = InteractionType::Assumption(Assertion::new(
format!("table {table} exists"),
{
let table = table.clone();
move |_: &Vec<ResultSet>, env: &mut SimulatorEnv| {
if env.tables.iter().any(|t| t.name == table) {
@@ -535,22 +556,22 @@ impl Property {
}
}
}
}),
});
},
));
let delete = Interaction::Query(Query::Delete(Delete {
let delete = InteractionType::Query(Query::Delete(Delete {
table: table.clone(),
predicate: predicate.clone(),
}));
let select = Interaction::Query(Query::Select(Select::simple(
let select = InteractionType::Query(Query::Select(Select::simple(
table.clone(),
predicate.clone(),
)));
let assertion = Interaction::Assertion(Assertion {
name: format!("`{select}` should return no values for table `{table}`",),
func: Box::new(move |stack: &Vec<ResultSet>, _| {
let assertion = InteractionType::Assertion(Assertion::new(
format!("`{select}` should return no values for table `{table}`",),
move |stack: &Vec<ResultSet>, _| {
let rows = stack.last().unwrap();
match rows {
Ok(rows) => {
@@ -569,15 +590,20 @@ impl Property {
}
Err(err) => Err(LimboError::InternalError(err.to_string())),
}
}),
});
},
));
let mut interactions = Vec::new();
interactions.push(assumption);
interactions.push(delete);
interactions.extend(queries.clone().into_iter().map(Interaction::Query));
interactions.push(select);
interactions.push(assertion);
interactions.push(Interaction::new(connection_index, assumption));
interactions.push(Interaction::new(connection_index, delete));
interactions.extend(
queries
.clone()
.into_iter()
.map(|q| Interaction::new(connection_index, InteractionType::Query(q))),
);
interactions.push(Interaction::new(connection_index, select));
interactions.push(Interaction::new(connection_index, assertion));
interactions
}
@@ -586,9 +612,9 @@ impl Property {
queries,
select,
} => {
let assumption = Interaction::Assumption(Assertion {
name: format!("table {table} exists"),
func: Box::new({
let assumption = InteractionType::Assumption(Assertion::new(
format!("table {table} exists"),
{
let table = table.clone();
move |_, env: &mut SimulatorEnv| {
if env.tables.iter().any(|t| t.name == table) {
@@ -603,14 +629,14 @@ impl Property {
}
}
}
}),
});
},
));
let table_name = table.clone();
let assertion = Interaction::Assertion(Assertion {
name: format!("select query should result in an error for table '{table}'"),
func: Box::new(move |stack: &Vec<ResultSet>, _| {
let assertion = InteractionType::Assertion(Assertion::new(
format!("select query should result in an error for table '{table}'"),
move |stack: &Vec<ResultSet>, _| {
let last = stack.last().unwrap();
match last {
Ok(success) => Ok(Err(format!(
@@ -628,29 +654,34 @@ impl Property {
}
}
}
}),
});
},
));
let drop = Interaction::Query(Query::Drop(Drop {
let drop = InteractionType::Query(Query::Drop(Drop {
table: table.clone(),
}));
let select = Interaction::Query(Query::Select(select.clone()));
let select = InteractionType::Query(Query::Select(select.clone()));
let mut interactions = Vec::new();
interactions.push(assumption);
interactions.push(drop);
interactions.extend(queries.clone().into_iter().map(Interaction::Query));
interactions.push(select);
interactions.push(assertion);
interactions.push(Interaction::new(connection_index, assumption));
interactions.push(Interaction::new(connection_index, drop));
interactions.extend(
queries
.clone()
.into_iter()
.map(|q| Interaction::new(connection_index, InteractionType::Query(q))),
);
interactions.push(Interaction::new(connection_index, select));
interactions.push(Interaction::new(connection_index, assertion));
interactions
}
Property::SelectSelectOptimizer { table, predicate } => {
let assumption = Interaction::Assumption(Assertion {
name: format!("table {table} exists"),
func: Box::new({
let assumption = InteractionType::Assumption(Assertion::new(
format!("table {table} exists"),
{
let table = table.clone();
move |_: &Vec<ResultSet>, env: &mut SimulatorEnv| {
if env.tables.iter().any(|t| t.name == table) {
@@ -665,10 +696,10 @@ impl Property {
}
}
}
}),
});
},
));
let select1 = Interaction::Query(Query::Select(Select::single(
let select1 = InteractionType::Query(Query::Select(Select::single(
table.clone(),
vec![ResultColumn::Expr(predicate.clone())],
Predicate::true_(),
@@ -678,11 +709,11 @@ impl Property {
let select2_query = Query::Select(Select::simple(table.clone(), predicate.clone()));
let select2 = Interaction::Query(select2_query);
let select2 = InteractionType::Query(select2_query);
let assertion = Interaction::Assertion(Assertion {
name: "select queries should return the same amount of results".to_string(),
func: Box::new(move |stack: &Vec<ResultSet>, _| {
let assertion = InteractionType::Assertion(Assertion::new(
"select queries should return the same amount of results".to_string(),
move |stack: &Vec<ResultSet>, _| {
let select_star = stack.last().unwrap();
let select_predicate = stack.get(stack.len() - 2).unwrap();
match (select_predicate, select_star) {
@@ -725,26 +756,35 @@ impl Property {
Err(LimboError::InternalError(e.to_string()))
}
}
}),
});
},
));
vec![assumption, select1, select2, assertion]
vec![
Interaction::new(connection_index, assumption),
Interaction::new(connection_index, select1),
Interaction::new(connection_index, select2),
Interaction::new(connection_index, assertion),
]
}
Property::FsyncNoWait { query, tables } => {
let checks = assert_all_table_values(tables);
let checks = assert_all_table_values(tables, connection_index);
Vec::from_iter(
std::iter::once(Interaction::FsyncQuery(query.clone())).chain(checks),
std::iter::once(Interaction::new(
connection_index,
InteractionType::FsyncQuery(query.clone()),
))
.chain(checks),
)
}
Property::FaultyQuery { query, tables } => {
let checks = assert_all_table_values(tables);
let checks = assert_all_table_values(tables, connection_index);
let query_clone = query.clone();
let assert = Assertion {
// A fault may not occur as we first signal we want a fault injected,
// then when IO is called the fault triggers. It may happen that a fault is injected
// but no IO happens right after it
name: "fault occured".to_string(),
func: Box::new(move |stack, env: &mut SimulatorEnv| {
// A fault may not occur as we first signal we want a fault injected,
// then when IO is called the fault triggers. It may happen that a fault is injected
// but no IO happens right after it
let assert = Assertion::new(
"fault occured".to_string(),
move |stack, env: &mut SimulatorEnv| {
let last = stack.last().unwrap();
match last {
Ok(_) => {
@@ -758,18 +798,19 @@ impl Property {
Ok(Ok(()))
}
}
}),
};
},
);
let first = [
Interaction::FaultyQuery(query.clone()),
Interaction::Assertion(assert),
InteractionType::FaultyQuery(query.clone()),
InteractionType::Assertion(assert),
]
.into_iter();
.into_iter()
.map(|i| Interaction::new(connection_index, i));
Vec::from_iter(first.chain(checks))
}
Property::WhereTrueFalseNull { select, predicate } => {
let assumption = Interaction::Assumption(Assertion {
name: format!(
let assumption = InteractionType::Assumption(Assertion::new(
format!(
"tables ({}) exists",
select
.dependencies()
@@ -777,7 +818,7 @@ impl Property {
.collect::<Vec<_>>()
.join(", ")
),
func: Box::new({
{
let tables = select.dependencies();
move |_: &Vec<ResultSet>, env: &mut SimulatorEnv| {
if tables
@@ -793,8 +834,8 @@ impl Property {
Ok(Err(format!("missing tables: {missing_tables:?}")))
}
}
}),
});
},
));
let old_predicate = select.body.select.where_clause.clone();
@@ -843,13 +884,13 @@ impl Property {
limit: None,
};
let select = Interaction::Query(Query::Select(select.clone()));
let select_tlp = Interaction::Query(Query::Select(select_tlp));
let select = InteractionType::Query(Query::Select(select.clone()));
let select_tlp = InteractionType::Query(Query::Select(select_tlp));
// select and select_tlp should return the same rows
let assertion = Interaction::Assertion(Assertion {
name: "select and select_tlp should return the same rows".to_string(),
func: Box::new(move |stack: &Vec<ResultSet>, _: &mut SimulatorEnv| {
let assertion = InteractionType::Assertion(Assertion::new(
"select and select_tlp should return the same rows".to_string(),
move |stack: &Vec<ResultSet>, _: &mut SimulatorEnv| {
if stack.len() < 2 {
return Err(LimboError::InternalError(
"Not enough result sets on the stack".to_string(),
@@ -914,10 +955,15 @@ impl Property {
Err(LimboError::InternalError(e.to_string()))
}
}
}),
});
},
));
vec![assumption, select, select_tlp, assertion]
vec![
Interaction::new(connection_index, assumption),
Interaction::new(connection_index, select),
Interaction::new(connection_index, select_tlp),
Interaction::new(connection_index, assertion),
]
}
Property::UNIONAllPreservesCardinality {
select,
@@ -929,12 +975,12 @@ impl Property {
let s3 = Select::compound(s1.clone(), s2.clone(), CompoundOperator::UnionAll);
vec![
Interaction::Query(Query::Select(s1.clone())),
Interaction::Query(Query::Select(s2.clone())),
Interaction::Query(Query::Select(s3.clone())),
Interaction::Assertion(Assertion {
name: "UNION ALL should preserve cardinality".to_string(),
func: Box::new(move |stack: &Vec<ResultSet>, _: &mut SimulatorEnv| {
InteractionType::Query(Query::Select(s1.clone())),
InteractionType::Query(Query::Select(s2.clone())),
InteractionType::Query(Query::Select(s3.clone())),
InteractionType::Assertion(Assertion::new(
"UNION ALL should preserve cardinality".to_string(),
move |stack: &Vec<ResultSet>, _: &mut SimulatorEnv| {
if stack.len() < 3 {
return Err(LimboError::InternalError(
"Not enough result sets on the stack".to_string(),
@@ -963,24 +1009,25 @@ impl Property {
Err(LimboError::InternalError(e.to_string()))
}
}
}),
}),
]
},
)),
].into_iter().map(|i| Interaction::new(connection_index, i)).collect()
}
}
}
}
fn assert_all_table_values(tables: &[String]) -> impl Iterator<Item = Interaction> + use<'_> {
tables.iter().flat_map(|table| {
let select = Interaction::Query(Query::Select(Select::simple(
fn assert_all_table_values(
tables: &[String],
connection_index: usize,
) -> impl Iterator<Item = Interaction> + use<'_> {
tables.iter().flat_map(move |table| {
let select = InteractionType::Query(Query::Select(Select::simple(
table.clone(),
Predicate::true_(),
)));
let assertion = Interaction::Assertion(Assertion {
name: format!("table {table} should contain all of its expected values"),
func: Box::new({
let assertion = InteractionType::Assertion(Assertion::new(format!("table {table} should contain all of its expected values"), {
let table = table.clone();
move |stack: &Vec<ResultSet>, env: &mut SimulatorEnv| {
let table = env.tables.iter().find(|t| t.name == table).ok_or_else(|| {
@@ -1023,9 +1070,8 @@ fn assert_all_table_values(tables: &[String]) -> impl Iterator<Item = Interactio
Err(err) => Err(LimboError::InternalError(format!("{err}"))),
}
}
}),
});
[select, assertion].into_iter()
}));
[select, assertion].into_iter().map(move |i| Interaction::new(connection_index, i))
})
}
@@ -1236,12 +1282,9 @@ fn property_double_create_failure<R: rand::Rng>(
env: &SimulatorEnv,
remaining: &Remaining,
) -> Property {
// Get a random table
let table = pick(&env.tables, rng);
// Create the table
let create_query = Create {
table: table.clone(),
};
let create_query = Create::arbitrary(rng, env);
let table = &create_query.table;
// Create random queries respecting the constraints
let mut queries = Vec::new();

142
simulator/main.rs
View File

@@ -7,9 +7,9 @@ use notify::{EventKind, RecursiveMode, Watcher};
use rand::prelude::*;
use runner::bugbase::{Bug, BugBase, LoadedBug};
use runner::cli::{SimulatorCLI, SimulatorCommand};
use runner::differential;
use runner::env::SimulatorEnv;
use runner::execution::{Execution, ExecutionHistory, ExecutionResult, execute_plans};
use runner::{differential, watch};
use runner::execution::{Execution, ExecutionHistory, ExecutionResult, execute_interactions};
use std::any::Any;
use std::backtrace::Backtrace;
use std::fs::OpenOptions;
@@ -21,6 +21,7 @@ use tracing_subscriber::field::MakeExt;
use tracing_subscriber::fmt::format;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
use crate::generation::plan::ConnectionState;
use crate::profiles::Profile;
use crate::runner::doublecheck;
use crate::runner::env::{Paths, SimulationPhase, SimulationType};
@@ -167,7 +168,7 @@ fn watch_mode(env: SimulatorEnv) -> notify::Result<()> {
// below will be monitored for changes.
watcher.watch(&env.get_plan_path(), RecursiveMode::NonRecursive)?;
// Block forever, printing out events as they come in
let last_execution = Arc::new(Mutex::new(Execution::new(0, 0, 0)));
let last_execution = Arc::new(Mutex::new(Execution::new(0, 0)));
for res in rx {
match res {
Ok(event) => {
@@ -178,8 +179,7 @@ fn watch_mode(env: SimulatorEnv) -> notify::Result<()> {
let result = SandboxedResult::from(
std::panic::catch_unwind(move || {
let mut env = env;
let plan: Vec<Vec<Interaction>> =
InteractionPlan::compute_via_diff(&env.get_plan_path());
let plan = InteractionPlan::compute_via_diff(&env.get_plan_path());
tracing::error!("plan_len: {}", plan.len());
env.clear();
@@ -190,7 +190,7 @@ fn watch_mode(env: SimulatorEnv) -> notify::Result<()> {
// });
let env = Arc::new(Mutex::new(env.clone_without_connections()));
watch::run_simulation(env, &mut [plan], last_execution_.clone())
run_simulation_default(env, plan, last_execution_.clone())
}),
last_execution.clone(),
);
@@ -217,7 +217,7 @@ fn run_simulator(
mut bugbase: Option<&mut BugBase>,
cli_opts: &SimulatorCLI,
env: SimulatorEnv,
plans: Vec<InteractionPlan>,
plan: InteractionPlan,
) -> anyhow::Result<()> {
std::panic::set_hook(Box::new(move |info| {
tracing::error!("panic occurred");
@@ -235,11 +235,12 @@ fn run_simulator(
tracing::error!("captured backtrace:\n{}", bt);
}));
let last_execution = Arc::new(Mutex::new(Execution::new(0, 0, 0)));
let last_execution = Arc::new(Mutex::new(Execution::new(0, 0)));
let env = Arc::new(Mutex::new(env));
let result = SandboxedResult::from(
std::panic::catch_unwind(|| {
run_simulation(env.clone(), &mut plans.clone(), last_execution.clone())
let interactions = plan.interactions_list();
run_simulation(env.clone(), interactions, last_execution.clone())
}),
last_execution.clone(),
);
@@ -269,10 +270,8 @@ fn run_simulator(
for execution in history.history.iter() {
writeln!(
f,
"{} {} {}",
execution.connection_index,
execution.interaction_index,
execution.secondary_index
"{} {}",
execution.connection_index, execution.interaction_index,
)
.unwrap();
}
@@ -284,51 +283,38 @@ fn run_simulator(
tracing::info!("shrinking is disabled, skipping shrinking");
if let Some(bugbase) = bugbase.as_deref_mut() {
bugbase
.add_bug(
env.opts.seed,
plans[0].clone(),
Some(error.clone()),
cli_opts,
)
.add_bug(env.opts.seed, plan.clone(), Some(error.clone()), cli_opts)
.unwrap();
}
return Err(anyhow!("failed with error: '{}'", error));
}
tracing::info!("Starting to shrink");
let (shrunk_plans, shrunk) = if !cli_opts.disable_heuristic_shrinking {
let shrunk_plans = plans
.iter()
.map(|plan| {
let shrunk = plan.shrink_interaction_plan(last_execution);
tracing::info!("{}", shrunk.stats());
shrunk
})
.collect::<Vec<_>>();
let (shrunk_plan, shrunk) = if !cli_opts.disable_heuristic_shrinking {
let shrunk_plan = plan.shrink_interaction_plan(last_execution);
tracing::info!("{}", shrunk_plan.stats());
// Write the shrunk plan to a file
let shrunk_plan_path = env
.paths
.plan(&SimulationType::Default, &SimulationPhase::Shrink);
let mut f = std::fs::File::create(&shrunk_plan_path).unwrap();
tracing::trace!("writing shrunk plan to {}", shrunk_plan_path.display());
f.write_all(shrunk_plans[0].to_string().as_bytes()).unwrap();
f.write_all(shrunk_plan.to_string().as_bytes()).unwrap();
let last_execution = Arc::new(Mutex::new(*last_execution));
let env = env.clone_at_phase(SimulationPhase::Shrink);
let env = Arc::new(Mutex::new(env));
let shrunk = SandboxedResult::from(
std::panic::catch_unwind(|| {
run_simulation(
env.clone(),
&mut shrunk_plans.clone(),
last_execution.clone(),
)
let interactions = shrunk_plan.interactions_list();
run_simulation(env.clone(), interactions, last_execution.clone())
}),
last_execution,
);
(shrunk_plans, shrunk)
(shrunk_plan, shrunk)
} else {
(plans.to_vec(), result.clone())
(plan.clone(), result.clone())
};
match (&shrunk, &result) {
@@ -344,16 +330,11 @@ fn run_simulator(
tracing::trace!(
"adding bug to bugbase, seed: {}, plan: {}, error: {}",
env.opts.seed,
plans[0].plan.len(),
plan.plan.len(),
error
);
bugbase
.add_bug(
env.opts.seed,
plans[0].clone(),
Some(error.clone()),
cli_opts,
)
.add_bug(env.opts.seed, plan.clone(), Some(error.clone()), cli_opts)
.unwrap();
}
@@ -369,30 +350,26 @@ fn run_simulator(
let env = env.clone_at_phase(SimulationPhase::Shrink);
let env = Arc::new(Mutex::new(env));
let final_plans = if cli_opts.enable_brute_force_shrinking {
let brute_shrunk_plans = shrunk_plans
.iter()
.map(|plan| {
plan.brute_shrink_interaction_plan(&shrunk, env.clone())
})
.collect::<Vec<_>>();
let final_plan = if cli_opts.enable_brute_force_shrinking {
let brute_shrunk_plan =
shrunk_plan.brute_shrink_interaction_plan(&shrunk, env.clone());
tracing::info!("Brute force shrinking completed");
brute_shrunk_plans
brute_shrunk_plan
} else {
shrunk_plans
shrunk_plan
};
tracing::info!(
"shrinking succeeded, reduced the plan from {} to {}",
plans[0].plan.len(),
final_plans[0].plan.len()
plan.plan.len(),
final_plan.plan.len()
);
// Save the shrunk database
if let Some(bugbase) = bugbase.as_deref_mut() {
bugbase.make_shrunk(
seed,
cli_opts,
final_plans[0].clone(),
final_plan.clone(),
Some(e1.clone()),
)?;
}
@@ -410,12 +387,7 @@ fn run_simulator(
);
if let Some(bugbase) = bugbase {
bugbase
.add_bug(
env.opts.seed,
plans[0].clone(),
Some(error.clone()),
cli_opts,
)
.add_bug(env.opts.seed, plan.clone(), Some(error.clone()), cli_opts)
.unwrap();
}
Err(anyhow!("failed with error: '{}'", error))
@@ -482,7 +454,7 @@ fn setup_simulation(
bugbase: Option<&mut BugBase>,
cli_opts: &SimulatorCLI,
profile: &Profile,
) -> (u64, SimulatorEnv, Vec<InteractionPlan>) {
) -> (u64, SimulatorEnv, InteractionPlan) {
if let Some(seed) = &cli_opts.load {
let seed = seed.parse::<u64>().expect("seed should be a number");
let bugbase = bugbase.expect("BugBase must be enabled to load a bug");
@@ -521,8 +493,7 @@ fn setup_simulation(
serde_json::to_string_pretty(&plan).unwrap(),
)
.unwrap();
let plans = vec![plan];
(seed, env, plans)
(seed, env, plan)
} else {
let seed = cli_opts.seed.unwrap_or_else(|| {
let mut rng = rand::rng();
@@ -549,12 +520,9 @@ fn setup_simulation(
tracing::info!("Generating database interaction plan...");
let plans = (1..=env.opts.max_connections)
.map(|_| InteractionPlan::generate_plan(&mut env.rng.clone(), &mut env))
.collect::<Vec<_>>();
let plan = InteractionPlan::generate_plan(&mut env.rng.clone(), &mut env);
// todo: for now, we only use 1 connection, so it's safe to use the first plan.
let plan = &plans[0];
tracing::info!("{}", plan.stats());
std::fs::write(env.get_plan_path(), plan.to_string()).unwrap();
std::fs::write(
@@ -563,13 +531,13 @@ fn setup_simulation(
)
.unwrap();
(seed, env, plans)
(seed, env, plan)
}
}
fn run_simulation(
env: Arc<Mutex<SimulatorEnv>>,
plans: &mut [InteractionPlan],
plan: Vec<Interaction>,
last_execution: Arc<Mutex<Execution>>,
) -> ExecutionResult {
let simulation_type = {
@@ -580,14 +548,14 @@ fn run_simulation(
};
match simulation_type {
SimulationType::Default => run_simulation_default(env, plans, last_execution),
SimulationType::Default => run_simulation_default(env, plan, last_execution),
SimulationType::Differential => {
let limbo_env = {
let env = env.lock().unwrap();
env.clone_as(SimulationType::Default)
};
let limbo_env = Arc::new(Mutex::new(limbo_env));
differential::run_simulation(limbo_env, env, plans, last_execution)
differential::run_simulation(limbo_env, env, plan, last_execution)
}
SimulationType::Doublecheck => {
let limbo_env = {
@@ -595,28 +563,38 @@ fn run_simulation(
env.clone_as(SimulationType::Default)
};
let limbo_env = Arc::new(Mutex::new(limbo_env));
doublecheck::run_simulation(limbo_env, env, plans, last_execution)
doublecheck::run_simulation(limbo_env, env, plan, last_execution)
}
}
}
fn run_simulation_default(
env: Arc<Mutex<SimulatorEnv>>,
plans: &mut [InteractionPlan],
plan: Vec<Interaction>,
last_execution: Arc<Mutex<Execution>>,
) -> ExecutionResult {
tracing::info!("Executing database interaction plan...");
let mut states = plans
.iter()
.map(|_| InteractionPlanState {
stack: vec![],
interaction_pointer: 0,
secondary_pointer: 0,
})
let num_conns = {
let env = env.lock().unwrap();
env.connections.len()
};
let mut conn_states = (0..num_conns)
.map(|_| ConnectionState::default())
.collect::<Vec<_>>();
let mut result = execute_plans(env.clone(), plans, &mut states, last_execution);
let mut state = InteractionPlanState {
interaction_pointer: 0,
};
let mut result = execute_interactions(
env.clone(),
plan,
&mut state,
&mut conn_states,
last_execution,
);
let env = env.lock().unwrap();
env.io.print_stats();

4
simulator/profiles/mod.rs
View File

@@ -28,17 +28,19 @@ pub struct Profile {
#[garde(skip)]
/// Experimental MVCC feature
pub experimental_mvcc: bool,
#[garde(range(min = 1))]
pub max_connections: usize,
#[garde(dive)]
pub io: IOProfile,
#[garde(dive)]
pub query: QueryProfile,
}
#[allow(clippy::derivable_impls)]
impl Default for Profile {
fn default() -> Self {
Self {
experimental_mvcc: false,
max_connections: 10,
io: Default::default(),
query: Default::default(),
}

537
simulator/runner/differential.rs
View File

@@ -1,54 +1,41 @@
use std::sync::{Arc, Mutex};
use sql_generation::{generation::pick_index, model::table::SimValue};
use turso_core::Value;
use crate::{
InteractionPlan,
generation::{
Shadow as _,
plan::{Interaction, InteractionPlanState, ResultSet},
},
model::Query,
runner::execution::ExecutionContinuation,
};
use crate::generation::plan::{ConnectionState, Interaction, InteractionPlanState};
use super::{
env::{SimConnection, SimulatorEnv},
execution::{Execution, ExecutionHistory, ExecutionResult, execute_interaction},
env::SimulatorEnv,
execution::{Execution, ExecutionHistory, ExecutionResult},
};
pub(crate) fn run_simulation(
pub fn run_simulation(
env: Arc<Mutex<SimulatorEnv>>,
rusqlite_env: Arc<Mutex<SimulatorEnv>>,
plans: &mut [InteractionPlan],
plan: Vec<Interaction>,
last_execution: Arc<Mutex<Execution>>,
) -> ExecutionResult {
tracing::info!("Executing database interaction plan...");
let mut states = plans
.iter()
.map(|_| InteractionPlanState {
stack: vec![],
interaction_pointer: 0,
secondary_pointer: 0,
})
let num_conns = {
let env = env.lock().unwrap();
env.connections.len()
};
let mut conn_states = (0..num_conns)
.map(|_| ConnectionState::default())
.collect::<Vec<_>>();
let mut rusqlite_states = plans
.iter()
.map(|_| InteractionPlanState {
stack: vec![],
interaction_pointer: 0,
secondary_pointer: 0,
})
.collect::<Vec<_>>();
let mut rusqlite_states = conn_states.clone();
let result = execute_plans(
let mut state = InteractionPlanState {
interaction_pointer: 0,
};
let result = execute_interactions(
env,
rusqlite_env,
plans,
&mut states,
plan,
&mut state,
&mut conn_states,
&mut rusqlite_states,
last_execution,
);
@@ -58,116 +45,75 @@ pub(crate) fn run_simulation(
result
}
fn execute_query_rusqlite(
connection: &rusqlite::Connection,
query: &Query,
) -> rusqlite::Result<Vec<Vec<SimValue>>> {
match query {
Query::Create(create) => {
connection.execute(create.to_string().as_str(), ())?;
Ok(vec![])
}
Query::Select(select) => {
let mut stmt = connection.prepare(select.to_string().as_str())?;
let columns = stmt.column_count();
let rows = stmt.query_map([], |row| {
let mut values = vec![];
for i in 0..columns {
let value = row.get_unwrap(i);
let value = match value {
rusqlite::types::Value::Null => Value::Null,
rusqlite::types::Value::Integer(i) => Value::Integer(i),
rusqlite::types::Value::Real(f) => Value::Float(f),
rusqlite::types::Value::Text(s) => Value::build_text(s),
rusqlite::types::Value::Blob(b) => Value::Blob(b),
};
values.push(SimValue(value));
}
Ok(values)
})?;
let mut result = vec![];
for row in rows {
result.push(row?);
}
Ok(result)
}
Query::Insert(insert) => {
connection.execute(insert.to_string().as_str(), ())?;
Ok(vec![])
}
Query::Delete(delete) => {
connection.execute(delete.to_string().as_str(), ())?;
Ok(vec![])
}
Query::Drop(drop) => {
connection.execute(drop.to_string().as_str(), ())?;
Ok(vec![])
}
Query::Update(update) => {
connection.execute(update.to_string().as_str(), ())?;
Ok(vec![])
}
Query::CreateIndex(create_index) => {
connection.execute(create_index.to_string().as_str(), ())?;
Ok(vec![])
}
Query::Begin(begin) => {
connection.execute(begin.to_string().as_str(), ())?;
Ok(vec![])
}
Query::Commit(commit) => {
connection.execute(commit.to_string().as_str(), ())?;
Ok(vec![])
}
Query::Rollback(rollback) => {
connection.execute(rollback.to_string().as_str(), ())?;
Ok(vec![])
}
}
}
pub(crate) fn execute_plans(
pub(crate) fn execute_interactions(
env: Arc<Mutex<SimulatorEnv>>,
rusqlite_env: Arc<Mutex<SimulatorEnv>>,
plans: &mut [InteractionPlan],
states: &mut [InteractionPlanState],
rusqlite_states: &mut [InteractionPlanState],
interactions: Vec<Interaction>,
state: &mut InteractionPlanState,
conn_states: &mut [ConnectionState],
rusqlite_states: &mut [ConnectionState],
last_execution: Arc<Mutex<Execution>>,
) -> ExecutionResult {
let mut history = ExecutionHistory::new();
let now = std::time::Instant::now();
let mut env = env.lock().unwrap();
let mut rusqlite_env = rusqlite_env.lock().unwrap();
for _tick in 0..env.opts.ticks {
// Pick the connection to interact with
let connection_index = pick_index(env.connections.len(), &mut env.rng);
let state = &mut states[connection_index];
env.tables.clear();
rusqlite_env.tables.clear();
history.history.push(Execution::new(
connection_index,
state.interaction_pointer,
state.secondary_pointer,
));
let now = std::time::Instant::now();
for _tick in 0..env.opts.ticks {
if state.interaction_pointer >= interactions.len() {
break;
}
let interaction = &interactions[state.interaction_pointer];
let connection_index = interaction.connection_index;
let turso_conn_state = &mut conn_states[connection_index];
let rusqlite_conn_state = &mut rusqlite_states[connection_index];
history
.history
.push(Execution::new(connection_index, state.interaction_pointer));
let mut last_execution = last_execution.lock().unwrap();
last_execution.connection_index = connection_index;
last_execution.interaction_index = state.interaction_pointer;
last_execution.secondary_index = state.secondary_pointer;
// Execute the interaction for the selected connection
match execute_plan(
let mut turso_state = state.clone();
// first execute turso
let turso_res = super::execution::execute_plan(
&mut env,
interaction,
turso_conn_state,
&mut turso_state,
);
let mut rusqlite_state = state.clone();
// second execute rusqlite
let rusqlite_res = super::execution::execute_plan(
&mut rusqlite_env,
connection_index,
plans,
states,
rusqlite_states,
interaction,
rusqlite_conn_state,
&mut rusqlite_state,
);
// Compare results
if let Err(err) = compare_results(
turso_res,
turso_conn_state,
rusqlite_res,
rusqlite_conn_state,
) {
Ok(_) => {}
Err(err) => {
return ExecutionResult::new(history, Some(err));
}
return ExecutionResult::new(history, Some(err));
}
state.interaction_pointer += 1;
// Check if the maximum time for the simulation has been reached
if now.elapsed().as_secs() >= env.opts.max_time_simulation as u64 {
return ExecutionResult::new(
@@ -182,257 +128,122 @@ pub(crate) fn execute_plans(
ExecutionResult::new(history, None)
}
fn execute_plan(
env: &mut SimulatorEnv,
rusqlite_env: &mut SimulatorEnv,
connection_index: usize,
plans: &mut [InteractionPlan],
states: &mut [InteractionPlanState],
rusqlite_states: &mut [InteractionPlanState],
fn compare_results(
turso_res: turso_core::Result<()>,
turso_conn_state: &mut ConnectionState,
rusqlite_res: turso_core::Result<()>,
rusqlite_conn_state: &mut ConnectionState,
) -> turso_core::Result<()> {
let connection = &env.connections[connection_index];
let rusqlite_connection = &rusqlite_env.connections[connection_index];
let plan = &mut plans[connection_index];
let state = &mut states[connection_index];
let rusqlite_state = &mut rusqlite_states[connection_index];
if state.interaction_pointer >= plan.plan.len() {
return Ok(());
}
let interaction = &plan.plan[state.interaction_pointer].interactions()[state.secondary_pointer];
tracing::debug!(
"execute_plan(connection_index={}, interaction={})",
connection_index,
interaction
);
tracing::debug!(
"connection: {}, rusqlite_connection: {}",
connection,
rusqlite_connection
);
match (connection, rusqlite_connection) {
(SimConnection::Disconnected, SimConnection::Disconnected) => {
tracing::debug!("connecting {}", connection_index);
env.connect(connection_index);
rusqlite_env.connect(connection_index);
}
(SimConnection::LimboConnection(_), SimConnection::SQLiteConnection(_)) => {
let limbo_result =
execute_interaction(env, connection_index, interaction, &mut state.stack);
let ruqlite_result = execute_interaction_rusqlite(
rusqlite_env,
connection_index,
interaction,
&mut rusqlite_state.stack,
);
match (limbo_result, ruqlite_result) {
(Ok(next_execution), Ok(next_execution_rusqlite)) => {
if next_execution != next_execution_rusqlite {
tracing::error!(
"expected next executions of limbo and rusqlite do not match"
);
tracing::debug!(
"limbo result: {:?}, rusqlite result: {:?}",
next_execution,
next_execution_rusqlite
);
return Err(turso_core::LimboError::InternalError(
"expected next executions of limbo and rusqlite do not match".into(),
));
}
let limbo_values = state.stack.last();
let rusqlite_values = rusqlite_state.stack.last();
match (turso_res, rusqlite_res) {
(Ok(..), Ok(..)) => {
let limbo_values = turso_conn_state.stack.last();
let rusqlite_values = rusqlite_conn_state.stack.last();
match (limbo_values, rusqlite_values) {
(Some(limbo_values), Some(rusqlite_values)) => {
match (limbo_values, rusqlite_values) {
(Some(limbo_values), Some(rusqlite_values)) => {
match (limbo_values, rusqlite_values) {
(Ok(limbo_values), Ok(rusqlite_values)) => {
if limbo_values != rusqlite_values {
tracing::error!(
"returned values from limbo and rusqlite results do not match"
);
let diff = limbo_values
.iter()
.zip(rusqlite_values.iter())
.enumerate()
.filter(|(_, (l, r))| l != r)
.collect::<Vec<_>>();
(Ok(limbo_values), Ok(rusqlite_values)) => {
if limbo_values != rusqlite_values {
tracing::error!(
"returned values from limbo and rusqlite results do not match"
);
let diff = limbo_values
.iter()
.zip(rusqlite_values.iter())
.enumerate()
.filter(|(_, (l, r))| l != r)
.collect::<Vec<_>>();
let diff = diff
.iter()
.flat_map(|(i, (l, r))| {
let mut diffs = vec![];
for (j, (l, r)) in
l.iter().zip(r.iter()).enumerate()
{
if l != r {
tracing::debug!(
"difference at index {}, {}: {} != {}",
i,
j,
l.to_string(),
r.to_string()
);
diffs
.push(((i, j), (l.clone(), r.clone())));
}
}
diffs
})
.collect::<Vec<_>>();
tracing::debug!("limbo values {:?}", limbo_values);
tracing::debug!("rusqlite values {:?}", rusqlite_values);
tracing::debug!(
"differences: {}",
diff.iter()
.map(|((i, j), (l, r))| format!(
"\t({i}, {j}): ({l}) != ({r})"
))
.collect::<Vec<_>>()
.join("\n")
);
return Err(turso_core::LimboError::InternalError(
"returned values from limbo and rusqlite results do not match".into(),
));
}
}
(Err(limbo_err), Err(rusqlite_err)) => {
tracing::warn!(
"limbo and rusqlite both fail, requires manual check"
);
tracing::warn!("limbo error {}", limbo_err);
tracing::warn!("rusqlite error {}", rusqlite_err);
}
(Ok(limbo_result), Err(rusqlite_err)) => {
tracing::error!(
"limbo and rusqlite results do not match, limbo returned values but rusqlite failed"
);
tracing::error!("limbo values {:?}", limbo_result);
tracing::error!("rusqlite error {}", rusqlite_err);
return Err(turso_core::LimboError::InternalError(
"limbo and rusqlite results do not match".into(),
));
}
(Err(limbo_err), Ok(_)) => {
tracing::error!(
"limbo and rusqlite results do not match, limbo failed but rusqlite returned values"
);
tracing::error!("limbo error {}", limbo_err);
return Err(turso_core::LimboError::InternalError(
"limbo and rusqlite results do not match".into(),
));
}
let diff = diff
.iter()
.flat_map(|(i, (l, r))| {
let mut diffs = vec![];
for (j, (l, r)) in l.iter().zip(r.iter()).enumerate() {
if l != r {
tracing::debug!(
"difference at index {}, {}: {} != {}",
i,
j,
l.to_string(),
r.to_string()
);
diffs.push(((i, j), (l.clone(), r.clone())));
}
}
diffs
})
.collect::<Vec<_>>();
tracing::debug!("limbo values {:?}", limbo_values);
tracing::debug!("rusqlite values {:?}", rusqlite_values);
tracing::debug!(
"differences: {}",
diff.iter()
.map(|((i, j), (l, r))| format!(
"\t({i}, {j}): ({l}) != ({r})"
))
.collect::<Vec<_>>()
.join("\n")
);
return Err(turso_core::LimboError::InternalError(
"returned values from limbo and rusqlite results do not match"
.into(),
));
}
}
(None, None) => {}
_ => {
tracing::error!("limbo and rusqlite results do not match");
(Err(limbo_err), Err(rusqlite_err)) => {
tracing::warn!("limbo and rusqlite both fail, requires manual check");
tracing::warn!("limbo error {}", limbo_err);
tracing::warn!("rusqlite error {}", rusqlite_err);
}
(Ok(limbo_result), Err(rusqlite_err)) => {
tracing::error!(
"limbo and rusqlite results do not match, limbo returned values but rusqlite failed"
);
tracing::error!("limbo values {:?}", limbo_result);
tracing::error!("rusqlite error {}", rusqlite_err);
return Err(turso_core::LimboError::InternalError(
"limbo and rusqlite results do not match".into(),
));
}
(Err(limbo_err), Ok(_)) => {
tracing::error!(
"limbo and rusqlite results do not match, limbo failed but rusqlite returned values"
);
tracing::error!("limbo error {}", limbo_err);
return Err(turso_core::LimboError::InternalError(
"limbo and rusqlite results do not match".into(),
));
}
}
// Move to the next interaction or property
match next_execution {
ExecutionContinuation::NextInteraction => {
if state.secondary_pointer + 1
>= plan.plan[state.interaction_pointer].interactions().len()
{
// If we have reached the end of the interactions for this property, move to the next property
state.interaction_pointer += 1;
state.secondary_pointer = 0;
} else {
// Otherwise, move to the next interaction
state.secondary_pointer += 1;
}
}
ExecutionContinuation::NextProperty => {
// Skip to the next property
state.interaction_pointer += 1;
state.secondary_pointer = 0;
}
}
}
(Err(err), Ok(_)) => {
(None, None) => {}
_ => {
tracing::error!("limbo and rusqlite results do not match");
tracing::error!("limbo error {}", err);
return Err(err);
}
(Ok(val), Err(err)) => {
tracing::error!("limbo and rusqlite results do not match");
tracing::error!("limbo {:?}", val);
tracing::error!("rusqlite error {}", err);
return Err(err);
}
(Err(err), Err(err_rusqlite)) => {
tracing::error!("limbo and rusqlite both fail, requires manual check");
tracing::error!("limbo error {}", err);
tracing::error!("rusqlite error {}", err_rusqlite);
// todo: Previously, we returned an error here, but now we just log it.
// The problem is that the errors might be different, and we cannot
// just assume both of them being errors has the same semantics.
// return Err(err);
return Err(turso_core::LimboError::InternalError(
"limbo and rusqlite results do not match".into(),
));
}
}
}
_ => unreachable!("{} vs {}", connection, rusqlite_connection),
(Err(err), Ok(_)) => {
tracing::error!("limbo and rusqlite results do not match");
tracing::error!("limbo error {}", err);
return Err(err);
}
(Ok(val), Err(err)) => {
tracing::error!("limbo and rusqlite results do not match");
tracing::error!("limbo {:?}", val);
tracing::error!("rusqlite error {}", err);
return Err(err);
}
(Err(err), Err(err_rusqlite)) => {
tracing::error!("limbo and rusqlite both fail, requires manual check");
tracing::error!("limbo error {}", err);
tracing::error!("rusqlite error {}", err_rusqlite);
// todo: Previously, we returned an error here, but now we just log it.
// The problem is that the errors might be different, and we cannot
// just assume both of them being errors has the same semantics.
// return Err(err);
}
}
Ok(())
}
fn execute_interaction_rusqlite(
env: &mut SimulatorEnv,
connection_index: usize,
interaction: &Interaction,
stack: &mut Vec<ResultSet>,
) -> turso_core::Result<ExecutionContinuation> {
tracing::trace!(
"execute_interaction_rusqlite(connection_index={}, interaction={})",
connection_index,
interaction
);
match interaction {
Interaction::Query(query) => {
let conn = match &mut env.connections[connection_index] {
SimConnection::SQLiteConnection(conn) => conn,
SimConnection::LimboConnection(_) => unreachable!(),
SimConnection::Disconnected => unreachable!(),
};
tracing::debug!("{}", interaction);
let results = execute_query_rusqlite(conn, query).map_err(|e| {
turso_core::LimboError::InternalError(format!("error executing query: {e}"))
});
tracing::debug!("{:?}", results);
stack.push(results);
}
Interaction::FsyncQuery(..) => {
unimplemented!("cannot implement fsync query in rusqlite, as we do not control IO");
}
Interaction::Assertion(_) => {
interaction.execute_assertion(stack, env)?;
stack.clear();
}
Interaction::Assumption(_) => {
let assumption_result = interaction.execute_assumption(stack, env);
stack.clear();
if assumption_result.is_err() {
tracing::warn!("assumption failed: {:?}", assumption_result);
return Ok(ExecutionContinuation::NextProperty);
}
}
Interaction::Fault(_) => {
interaction.execute_fault(env, connection_index)?;
}
Interaction::FaultyQuery(_) => {
unimplemented!("cannot implement faulty query in rusqlite, as we do not control IO");
}
}
let _ = interaction.shadow(&mut env.tables);
Ok(ExecutionContinuation::NextInteraction)
}

346
simulator/runner/doublecheck.rs
View File

@@ -3,49 +3,42 @@ use std::{
sync::{Arc, Mutex},
};
use sql_generation::generation::pick_index;
use crate::{
InteractionPlan, generation::plan::InteractionPlanState,
runner::execution::ExecutionContinuation,
};
use crate::generation::plan::{ConnectionState, Interaction, InteractionPlanState};
use super::{
env::{SimConnection, SimulatorEnv},
execution::{Execution, ExecutionHistory, ExecutionResult, execute_interaction},
env::SimulatorEnv,
execution::{Execution, ExecutionHistory, ExecutionResult},
};
pub(crate) fn run_simulation(
pub fn run_simulation(
env: Arc<Mutex<SimulatorEnv>>,
doublecheck_env: Arc<Mutex<SimulatorEnv>>,
plans: &mut [InteractionPlan],
plan: Vec<Interaction>,
last_execution: Arc<Mutex<Execution>>,
) -> ExecutionResult {
tracing::info!("Executing database interaction plan...");
let mut states = plans
.iter()
.map(|_| InteractionPlanState {
stack: vec![],
interaction_pointer: 0,
secondary_pointer: 0,
})
let num_conns = {
let env = env.lock().unwrap();
env.connections.len()
};
let mut conn_states = (0..num_conns)
.map(|_| ConnectionState::default())
.collect::<Vec<_>>();
let mut doublecheck_states = plans
.iter()
.map(|_| InteractionPlanState {
stack: vec![],
interaction_pointer: 0,
secondary_pointer: 0,
})
.collect::<Vec<_>>();
let mut doublecheck_states = conn_states.clone();
let mut state = InteractionPlanState {
interaction_pointer: 0,
};
let mut result = execute_plans(
env.clone(),
doublecheck_env.clone(),
plans,
&mut states,
plan,
&mut state,
&mut conn_states,
&mut doublecheck_states,
last_execution,
);
@@ -85,45 +78,72 @@ pub(crate) fn run_simulation(
pub(crate) fn execute_plans(
env: Arc<Mutex<SimulatorEnv>>,
doublecheck_env: Arc<Mutex<SimulatorEnv>>,
plans: &mut [InteractionPlan],
states: &mut [InteractionPlanState],
doublecheck_states: &mut [InteractionPlanState],
interactions: Vec<Interaction>,
state: &mut InteractionPlanState,
conn_states: &mut [ConnectionState],
doublecheck_states: &mut [ConnectionState],
last_execution: Arc<Mutex<Execution>>,
) -> ExecutionResult {
let mut history = ExecutionHistory::new();
let now = std::time::Instant::now();
let mut env = env.lock().unwrap();
let mut doublecheck_env = doublecheck_env.lock().unwrap();
for _tick in 0..env.opts.ticks {
// Pick the connection to interact with
let connection_index = pick_index(env.connections.len(), &mut env.rng);
let state = &mut states[connection_index];
env.tables.clear();
doublecheck_env.tables.clear();
history.history.push(Execution::new(
connection_index,
state.interaction_pointer,
state.secondary_pointer,
));
let now = std::time::Instant::now();
for _tick in 0..env.opts.ticks {
if state.interaction_pointer >= interactions.len() {
break;
}
let interaction = &interactions[state.interaction_pointer];
let connection_index = interaction.connection_index;
let turso_conn_state = &mut conn_states[connection_index];
let doublecheck_conn_state = &mut doublecheck_states[connection_index];
history
.history
.push(Execution::new(connection_index, state.interaction_pointer));
let mut last_execution = last_execution.lock().unwrap();
last_execution.connection_index = connection_index;
last_execution.interaction_index = state.interaction_pointer;
last_execution.secondary_index = state.secondary_pointer;
// Execute the interaction for the selected connection
match execute_plan(
let mut turso_state = state.clone();
// first execute turso
let turso_res = super::execution::execute_plan(
&mut env,
interaction,
turso_conn_state,
&mut turso_state,
);
let mut doublecheck_state = state.clone();
// second execute doublecheck
let doublecheck_res = super::execution::execute_plan(
&mut doublecheck_env,
connection_index,
plans,
states,
doublecheck_states,
interaction,
doublecheck_conn_state,
&mut doublecheck_state,
);
// Compare results
if let Err(err) = compare_results(
turso_res,
turso_conn_state,
doublecheck_res,
doublecheck_conn_state,
) {
Ok(_) => {}
Err(err) => {
return ExecutionResult::new(history, Some(err));
}
return ExecutionResult::new(history, Some(err));
}
state.interaction_pointer += 1;
// Check if the maximum time for the simulation has been reached
if now.elapsed().as_secs() >= env.opts.max_time_simulation as u64 {
return ExecutionResult::new(
@@ -138,176 +158,92 @@ pub(crate) fn execute_plans(
ExecutionResult::new(history, None)
}
fn execute_plan(
env: &mut SimulatorEnv,
doublecheck_env: &mut SimulatorEnv,
connection_index: usize,
plans: &mut [InteractionPlan],
states: &mut [InteractionPlanState],
doublecheck_states: &mut [InteractionPlanState],
fn compare_results(
turso_res: turso_core::Result<()>,
turso_state: &mut ConnectionState,
doublecheck_res: turso_core::Result<()>,
doublecheck_state: &mut ConnectionState,
) -> turso_core::Result<()> {
let connection = &env.connections[connection_index];
let doublecheck_connection = &doublecheck_env.connections[connection_index];
let plan = &mut plans[connection_index];
let state = &mut states[connection_index];
let doublecheck_state = &mut doublecheck_states[connection_index];
if state.interaction_pointer >= plan.plan.len() {
return Ok(());
}
let interaction = &plan.plan[state.interaction_pointer].interactions()[state.secondary_pointer];
tracing::debug!(
"execute_plan(connection_index={}, interaction={})",
connection_index,
interaction
);
tracing::debug!(
"connection: {}, doublecheck_connection: {}",
connection,
doublecheck_connection
);
match (connection, doublecheck_connection) {
(SimConnection::Disconnected, SimConnection::Disconnected) => {
tracing::debug!("connecting {}", connection_index);
env.connect(connection_index);
doublecheck_env.connect(connection_index);
}
(SimConnection::LimboConnection(_), SimConnection::LimboConnection(_)) => {
let limbo_result =
execute_interaction(env, connection_index, interaction, &mut state.stack);
let doublecheck_result = execute_interaction(
doublecheck_env,
connection_index,
interaction,
&mut doublecheck_state.stack,
);
match (limbo_result, doublecheck_result) {
(Ok(next_execution), Ok(next_execution_doublecheck)) => {
if next_execution != next_execution_doublecheck {
tracing::error!(
"expected next executions of limbo and doublecheck do not match"
);
tracing::debug!(
"limbo result: {:?}, doublecheck result: {:?}",
next_execution,
next_execution_doublecheck
);
return Err(turso_core::LimboError::InternalError(
"expected next executions of limbo and doublecheck do not match".into(),
));
}
let limbo_values = state.stack.last();
let doublecheck_values = doublecheck_state.stack.last();
match (turso_res, doublecheck_res) {
(Ok(..), Ok(..)) => {
let turso_values = turso_state.stack.last();
let doublecheck_values = doublecheck_state.stack.last();
match (turso_values, doublecheck_values) {
(Some(limbo_values), Some(doublecheck_values)) => {
match (limbo_values, doublecheck_values) {
(Some(limbo_values), Some(doublecheck_values)) => {
match (limbo_values, doublecheck_values) {
(Ok(limbo_values), Ok(doublecheck_values)) => {
if limbo_values != doublecheck_values {
tracing::error!(
"returned values from limbo and doublecheck results do not match"
);
tracing::debug!("limbo values {:?}", limbo_values);
tracing::debug!(
"doublecheck values {:?}",
doublecheck_values
);
return Err(turso_core::LimboError::InternalError(
(Ok(limbo_values), Ok(doublecheck_values)) => {
if limbo_values != doublecheck_values {
tracing::error!(
"returned values from limbo and doublecheck results do not match"
);
tracing::debug!("limbo values {:?}", limbo_values);
tracing::debug!("doublecheck values {:?}", doublecheck_values);
return Err(turso_core::LimboError::InternalError(
"returned values from limbo and doublecheck results do not match".into(),
));
}
}
(Err(limbo_err), Err(doublecheck_err)) => {
if limbo_err.to_string() != doublecheck_err.to_string() {
tracing::error!(
"limbo and doublecheck errors do not match"
);
tracing::error!("limbo error {}", limbo_err);
tracing::error!("doublecheck error {}", doublecheck_err);
return Err(turso_core::LimboError::InternalError(
"limbo and doublecheck errors do not match".into(),
));
}
}
(Ok(limbo_result), Err(doublecheck_err)) => {
tracing::error!(
"limbo and doublecheck results do not match, limbo returned values but doublecheck failed"
);
tracing::error!("limbo values {:?}", limbo_result);
tracing::error!("doublecheck error {}", doublecheck_err);
return Err(turso_core::LimboError::InternalError(
"limbo and doublecheck results do not match".into(),
));
}
(Err(limbo_err), Ok(_)) => {
tracing::error!(
"limbo and doublecheck results do not match, limbo failed but doublecheck returned values"
);
tracing::error!("limbo error {}", limbo_err);
return Err(turso_core::LimboError::InternalError(
"limbo and doublecheck results do not match".into(),
));
}
}
}
(None, None) => {}
_ => {
tracing::error!("limbo and doublecheck results do not match");
(Err(limbo_err), Err(doublecheck_err)) => {
if limbo_err.to_string() != doublecheck_err.to_string() {
tracing::error!("limbo and doublecheck errors do not match");
tracing::error!("limbo error {}", limbo_err);
tracing::error!("doublecheck error {}", doublecheck_err);
return Err(turso_core::LimboError::InternalError(
"limbo and doublecheck errors do not match".into(),
));
}
}
(Ok(limbo_result), Err(doublecheck_err)) => {
tracing::error!(
"limbo and doublecheck results do not match, limbo returned values but doublecheck failed"
);
tracing::error!("limbo values {:?}", limbo_result);
tracing::error!("doublecheck error {}", doublecheck_err);
return Err(turso_core::LimboError::InternalError(
"limbo and doublecheck results do not match".into(),
));
}
(Err(limbo_err), Ok(_)) => {
tracing::error!(
"limbo and doublecheck results do not match, limbo failed but doublecheck returned values"
);
tracing::error!("limbo error {}", limbo_err);
return Err(turso_core::LimboError::InternalError(
"limbo and doublecheck results do not match".into(),
));
}
}
// Move to the next interaction or property
match next_execution {
ExecutionContinuation::NextInteraction => {
if state.secondary_pointer + 1
>= plan.plan[state.interaction_pointer].interactions().len()
{
// If we have reached the end of the interactions for this property, move to the next property
state.interaction_pointer += 1;
state.secondary_pointer = 0;
} else {
// Otherwise, move to the next interaction
state.secondary_pointer += 1;
}
}
ExecutionContinuation::NextProperty => {
// Skip to the next property
state.interaction_pointer += 1;
state.secondary_pointer = 0;
}
}
}
(Err(err), Ok(_)) => {
(None, None) => {}
_ => {
tracing::error!("limbo and doublecheck results do not match");
tracing::error!("limbo error {}", err);
return Err(err);
}
(Ok(val), Err(err)) => {
tracing::error!("limbo and doublecheck results do not match");
tracing::error!("limbo {:?}", val);
tracing::error!("doublecheck error {}", err);
return Err(err);
}
(Err(err), Err(err_doublecheck)) => {
if err.to_string() != err_doublecheck.to_string() {
tracing::error!("limbo and doublecheck errors do not match");
tracing::error!("limbo error {}", err);
tracing::error!("doublecheck error {}", err_doublecheck);
return Err(turso_core::LimboError::InternalError(
"limbo and doublecheck errors do not match".into(),
));
}
return Err(turso_core::LimboError::InternalError(
"limbo and doublecheck results do not match".into(),
));
}
}
}
_ => unreachable!("{} vs {}", connection, doublecheck_connection),
(Err(err), Ok(_)) => {
tracing::error!("limbo and doublecheck results do not match");
tracing::error!("limbo error {}", err);
return Err(err);
}
(Ok(val), Err(err)) => {
tracing::error!("limbo and doublecheck results do not match");
tracing::error!("limbo {:?}", val);
tracing::error!("doublecheck error {}", err);
return Err(err);
}
(Err(err), Err(err_doublecheck)) => {
if err.to_string() != err_doublecheck.to_string() {
tracing::error!("limbo and doublecheck errors do not match");
tracing::error!("limbo error {}", err);
tracing::error!("doublecheck error {}", err_doublecheck);
return Err(turso_core::LimboError::InternalError(
"limbo and doublecheck errors do not match".into(),
));
}
}
}
Ok(())
}

9
simulator/runner/env.rs
View File

@@ -173,6 +173,10 @@ impl SimulatorEnv {
env.clear();
env
}
pub fn choose_conn(&self, rng: &mut impl Rng) -> usize {
rng.random_range(0..self.connections.len())
}
}
impl SimulatorEnv {
@@ -188,8 +192,6 @@ impl SimulatorEnv {
let opts = SimulatorOpts {
seed,
ticks: rng.random_range(cli_opts.minimum_tests..=cli_opts.maximum_tests),
max_connections: 1, // TODO: for now let's use one connection as we didn't implement
// correct transactions processing
max_tables: rng.random_range(0..128),
disable_select_optimizer: cli_opts.disable_select_optimizer,
disable_insert_values_select: cli_opts.disable_insert_values_select,
@@ -276,7 +278,7 @@ impl SimulatorEnv {
}
};
let connections = (0..opts.max_connections)
let connections = (0..profile.max_connections)
.map(|_| SimConnection::Disconnected)
.collect::<Vec<_>>();
@@ -383,7 +385,6 @@ impl Display for SimConnection {
pub(crate) struct SimulatorOpts {
pub(crate) seed: u64,
pub(crate) ticks: usize,
pub(crate) max_connections: usize,
pub(crate) max_tables: usize,
pub(crate) disable_select_optimizer: bool,

305
simulator/runner/execution.rs
View File

@@ -1,65 +1,63 @@
use std::sync::{Arc, Mutex};
use sql_generation::generation::pick_index;
use sql_generation::model::table::SimValue;
use tracing::instrument;
use turso_core::{Connection, LimboError, Result, StepResult};
use turso_core::{Connection, LimboError, Result, StepResult, Value};
use crate::generation::{
Shadow as _,
plan::{Interaction, InteractionPlan, InteractionPlanState, ResultSet},
use crate::{
generation::{
Shadow as _,
plan::{ConnectionState, Interaction, InteractionPlanState, InteractionType, ResultSet},
},
model::Query,
};
use super::env::{SimConnection, SimulatorEnv};
#[derive(Debug, Clone, Copy)]
pub(crate) struct Execution {
pub(crate) connection_index: usize,
pub(crate) interaction_index: usize,
pub(crate) secondary_index: usize,
pub struct Execution {
pub connection_index: usize,
pub interaction_index: usize,
}
impl Execution {
pub(crate) fn new(
connection_index: usize,
interaction_index: usize,
secondary_index: usize,
) -> Self {
pub fn new(connection_index: usize, interaction_index: usize) -> Self {
Self {
connection_index,
interaction_index,
secondary_index,
}
}
}
#[derive(Debug, Clone)]
pub(crate) struct ExecutionHistory {
pub(crate) history: Vec<Execution>,
pub struct ExecutionHistory {
pub history: Vec<Execution>,
}
impl ExecutionHistory {
pub(crate) fn new() -> Self {
pub fn new() -> Self {
Self {
history: Vec::new(),
}
}
}
pub(crate) struct ExecutionResult {
pub(crate) history: ExecutionHistory,
pub(crate) error: Option<LimboError>,
pub struct ExecutionResult {
pub history: ExecutionHistory,
pub error: Option<LimboError>,
}
impl ExecutionResult {
pub(crate) fn new(history: ExecutionHistory, error: Option<LimboError>) -> Self {
pub fn new(history: ExecutionHistory, error: Option<LimboError>) -> Self {
Self { history, error }
}
}
pub(crate) fn execute_plans(
pub(crate) fn execute_interactions(
env: Arc<Mutex<SimulatorEnv>>,
plans: &mut [InteractionPlan],
states: &mut [InteractionPlanState],
interactions: Vec<Interaction>,
state: &mut InteractionPlanState,
conn_states: &mut [ConnectionState],
last_execution: Arc<Mutex<Execution>>,
) -> ExecutionResult {
let mut history = ExecutionHistory::new();
@@ -71,21 +69,24 @@ pub(crate) fn execute_plans(
for _tick in 0..env.opts.ticks {
tracing::trace!("Executing tick {}", _tick);
// Pick the connection to interact with
let connection_index = pick_index(env.connections.len(), &mut env.rng);
let state = &mut states[connection_index];
history.history.push(Execution::new(
connection_index,
state.interaction_pointer,
state.secondary_pointer,
));
if state.interaction_pointer >= interactions.len() {
break;
}
let interaction = &interactions[state.interaction_pointer];
let connection_index = interaction.connection_index;
let conn_state = &mut conn_states[connection_index];
history
.history
.push(Execution::new(connection_index, state.interaction_pointer));
let mut last_execution = last_execution.lock().unwrap();
last_execution.connection_index = connection_index;
last_execution.interaction_index = state.interaction_pointer;
last_execution.secondary_index = state.secondary_pointer;
// Execute the interaction for the selected connection
match execute_plan(&mut env, connection_index, plans, states) {
match execute_plan(&mut env, interaction, conn_state, state) {
Ok(_) => {}
Err(err) => {
return ExecutionResult::new(history, Some(err));
@@ -105,50 +106,26 @@ pub(crate) fn execute_plans(
ExecutionResult::new(history, None)
}
fn execute_plan(
pub fn execute_plan(
env: &mut SimulatorEnv,
connection_index: usize,
plans: &mut [InteractionPlan],
states: &mut [InteractionPlanState],
interaction: &Interaction,
conn_state: &mut ConnectionState,
state: &mut InteractionPlanState,
) -> Result<()> {
let connection = &env.connections[connection_index];
let plan = &mut plans[connection_index];
let state = &mut states[connection_index];
if state.interaction_pointer >= plan.plan.len() {
return Ok(());
}
let interaction = &plan.plan[state.interaction_pointer].interactions()[state.secondary_pointer];
let connection_index = interaction.connection_index;
let connection = &mut env.connections[connection_index];
if let SimConnection::Disconnected = connection {
tracing::debug!("connecting {}", connection_index);
env.connections[connection_index] = SimConnection::LimboConnection(
env.db.as_ref().expect("db to be Some").connect().unwrap(),
);
env.connect(connection_index);
} else {
tracing::debug!("connection {} already connected", connection_index);
match execute_interaction(env, connection_index, interaction, &mut state.stack) {
match execute_interaction(env, interaction, &mut conn_state.stack) {
Ok(next_execution) => {
tracing::debug!("connection {} processed", connection_index);
// Move to the next interaction or property
match next_execution {
ExecutionContinuation::NextInteraction => {
if state.secondary_pointer + 1
>= plan.plan[state.interaction_pointer].interactions().len()
{
// If we have reached the end of the interactions for this property, move to the next property
state.interaction_pointer += 1;
state.secondary_pointer = 0;
} else {
// Otherwise, move to the next interaction
state.secondary_pointer += 1;
}
}
ExecutionContinuation::NextProperty => {
// Skip to the next property
state.interaction_pointer += 1;
state.secondary_pointer = 0;
}
}
}
@@ -170,26 +147,39 @@ fn execute_plan(
pub(crate) enum ExecutionContinuation {
/// Default continuation, execute the next interaction.
NextInteraction,
/// Typically used in the case of preconditions failures, skip to the next property.
NextProperty,
// /// Typically used in the case of preconditions failures, skip to the next property.
// NextProperty,
}
#[instrument(skip(env, interaction, stack), fields(seed = %env.opts.seed, interaction = %interaction))]
pub(crate) fn execute_interaction(
pub fn execute_interaction(
env: &mut SimulatorEnv,
connection_index: usize,
interaction: &Interaction,
stack: &mut Vec<ResultSet>,
) -> Result<ExecutionContinuation> {
let connection = &mut env.connections[interaction.connection_index];
match connection {
SimConnection::LimboConnection(..) => execute_interaction_turso(env, interaction, stack),
SimConnection::SQLiteConnection(..) => {
execute_interaction_rusqlite(env, interaction, stack)
}
SimConnection::Disconnected => unreachable!(),
}
}
#[instrument(skip(env, interaction, stack), fields(seed = %env.opts.seed, interaction = %interaction))]
pub fn execute_interaction_turso(
env: &mut SimulatorEnv,
interaction: &Interaction,
stack: &mut Vec<ResultSet>,
) -> Result<ExecutionContinuation> {
let SimConnection::LimboConnection(conn) = &mut env.connections[interaction.connection_index]
else {
unreachable!()
};
// Leave this empty info! here to print the span of the execution
tracing::info!("");
match interaction {
Interaction::Query(_) => {
let conn = match &mut env.connections[connection_index] {
SimConnection::LimboConnection(conn) => conn,
SimConnection::SQLiteConnection(_) => unreachable!(),
SimConnection::Disconnected => unreachable!(),
};
match &interaction.interaction {
InteractionType::Query(_) => {
tracing::debug!(?interaction);
let results = interaction.execute_query(conn);
if results.is_err() {
@@ -198,46 +188,38 @@ pub(crate) fn execute_interaction(
stack.push(results);
limbo_integrity_check(conn)?;
}
Interaction::FsyncQuery(query) => {
let conn = match &env.connections[connection_index] {
SimConnection::LimboConnection(conn) => conn.clone(),
SimConnection::SQLiteConnection(_) => unreachable!(),
SimConnection::Disconnected => unreachable!(),
};
InteractionType::FsyncQuery(query) => {
let results = interaction.execute_fsync_query(conn.clone(), env);
if results.is_err() {
tracing::error!(?results);
}
stack.push(results);
let query_interaction = Interaction::Query(query.clone());
let query_interaction = Interaction::new(
interaction.connection_index,
InteractionType::Query(query.clone()),
);
execute_interaction(env, connection_index, &query_interaction, stack)?;
execute_interaction(env, &query_interaction, stack)?;
}
Interaction::Assertion(_) => {
InteractionType::Assertion(_) => {
interaction.execute_assertion(stack, env)?;
stack.clear();
}
Interaction::Assumption(_) => {
InteractionType::Assumption(_) => {
let assumption_result = interaction.execute_assumption(stack, env);
stack.clear();
if assumption_result.is_err() {
tracing::warn!("assumption failed: {:?}", assumption_result);
return Ok(ExecutionContinuation::NextProperty);
if let Err(err) = assumption_result {
tracing::warn!("assumption failed: {:?}", err);
return Err(err);
}
}
Interaction::Fault(_) => {
interaction.execute_fault(env, connection_index)?;
InteractionType::Fault(_) => {
interaction.execute_fault(env, interaction.connection_index)?;
}
Interaction::FaultyQuery(_) => {
let conn = match &env.connections[connection_index] {
SimConnection::LimboConnection(conn) => conn.clone(),
SimConnection::SQLiteConnection(_) => unreachable!(),
SimConnection::Disconnected => unreachable!(),
};
InteractionType::FaultyQuery(_) => {
let conn = conn.clone();
let results = interaction.execute_faulty_query(&conn, env);
if results.is_err() {
tracing::error!(?results);
@@ -293,3 +275,122 @@ fn limbo_integrity_check(conn: &Arc<Connection>) -> Result<()> {
}
Ok(())
}
fn execute_interaction_rusqlite(
env: &mut SimulatorEnv,
interaction: &Interaction,
stack: &mut Vec<ResultSet>,
) -> turso_core::Result<ExecutionContinuation> {
tracing::trace!(
"execute_interaction_rusqlite(connection_index={}, interaction={})",
interaction.connection_index,
interaction
);
let SimConnection::SQLiteConnection(conn) = &mut env.connections[interaction.connection_index]
else {
unreachable!()
};
match &interaction.interaction {
InteractionType::Query(query) => {
tracing::debug!("{}", interaction);
let results = execute_query_rusqlite(conn, query).map_err(|e| {
turso_core::LimboError::InternalError(format!("error executing query: {e}"))
});
tracing::debug!("{:?}", results);
stack.push(results);
}
InteractionType::FsyncQuery(..) => {
unimplemented!("cannot implement fsync query in rusqlite, as we do not control IO");
}
InteractionType::Assertion(_) => {
interaction.execute_assertion(stack, env)?;
stack.clear();
}
InteractionType::Assumption(_) => {
let assumption_result = interaction.execute_assumption(stack, env);
stack.clear();
if let Err(err) = assumption_result {
tracing::warn!("assumption failed: {:?}", err);
return Err(err);
}
}
InteractionType::Fault(_) => {
interaction.execute_fault(env, interaction.connection_index)?;
}
InteractionType::FaultyQuery(_) => {
unimplemented!("cannot implement faulty query in rusqlite, as we do not control IO");
}
}
let _ = interaction.shadow(&mut env.tables);
Ok(ExecutionContinuation::NextInteraction)
}
fn execute_query_rusqlite(
connection: &rusqlite::Connection,
query: &Query,
) -> rusqlite::Result<Vec<Vec<SimValue>>> {
match query {
Query::Create(create) => {
connection.execute(create.to_string().as_str(), ())?;
Ok(vec![])
}
Query::Select(select) => {
let mut stmt = connection.prepare(select.to_string().as_str())?;
let columns = stmt.column_count();
let rows = stmt.query_map([], |row| {
let mut values = vec![];
for i in 0..columns {
let value = row.get_unwrap(i);
let value = match value {
rusqlite::types::Value::Null => Value::Null,
rusqlite::types::Value::Integer(i) => Value::Integer(i),
rusqlite::types::Value::Real(f) => Value::Float(f),
rusqlite::types::Value::Text(s) => Value::build_text(s),
rusqlite::types::Value::Blob(b) => Value::Blob(b),
};
values.push(SimValue(value));
}
Ok(values)
})?;
let mut result = vec![];
for row in rows {
result.push(row?);
}
Ok(result)
}
Query::Insert(insert) => {
connection.execute(insert.to_string().as_str(), ())?;
Ok(vec![])
}
Query::Delete(delete) => {
connection.execute(delete.to_string().as_str(), ())?;
Ok(vec![])
}
Query::Drop(drop) => {
connection.execute(drop.to_string().as_str(), ())?;
Ok(vec![])
}
Query::Update(update) => {
connection.execute(update.to_string().as_str(), ())?;
Ok(vec![])
}
Query::CreateIndex(create_index) => {
connection.execute(create_index.to_string().as_str(), ())?;
Ok(vec![])
}
Query::Begin(begin) => {
connection.execute(begin.to_string().as_str(), ())?;
Ok(vec![])
}
Query::Commit(commit) => {
connection.execute(commit.to_string().as_str(), ())?;
Ok(vec![])
}
Query::Rollback(rollback) => {
connection.execute(rollback.to_string().as_str(), ())?;
Ok(vec![])
}
}
}

1
simulator/runner/mod.rs
View File

@@ -9,7 +9,6 @@ pub mod execution;
pub mod file;
pub mod io;
pub mod memory;
pub mod watch;
pub const FAULT_ERROR_MSG: &str = "Injected Fault";

145
simulator/runner/watch.rs
View File

@@ -1,145 +0,0 @@
use std::sync::{Arc, Mutex};
use sql_generation::generation::pick_index;
use crate::{
generation::plan::{Interaction, InteractionPlanState},
integrity_check,
runner::execution::ExecutionContinuation,
};
use super::{
env::{SimConnection, SimulatorEnv},
execution::{Execution, ExecutionHistory, ExecutionResult, execute_interaction},
};
pub(crate) fn run_simulation(
env: Arc<Mutex<SimulatorEnv>>,
plans: &mut [Vec<Vec<Interaction>>],
last_execution: Arc<Mutex<Execution>>,
) -> ExecutionResult {
let mut states = plans
.iter()
.map(|_| InteractionPlanState {
stack: vec![],
interaction_pointer: 0,
secondary_pointer: 0,
})
.collect::<Vec<_>>();
let mut result = execute_plans(env.clone(), plans, &mut states, last_execution);
let env = env.lock().unwrap();
env.io.print_stats();
tracing::info!("Simulation completed");
if result.error.is_none() {
let ic = integrity_check(&env.get_db_path());
if let Err(err) = ic {
tracing::error!("integrity check failed: {}", err);
result.error = Some(turso_core::LimboError::InternalError(err.to_string()));
} else {
tracing::info!("integrity check passed");
}
}
result
}
pub(crate) fn execute_plans(
env: Arc<Mutex<SimulatorEnv>>,
plans: &mut [Vec<Vec<Interaction>>],
states: &mut [InteractionPlanState],
last_execution: Arc<Mutex<Execution>>,
) -> ExecutionResult {
let mut history = ExecutionHistory::new();
let now = std::time::Instant::now();
let mut env = env.lock().unwrap();
for _tick in 0..env.opts.ticks {
// Pick the connection to interact with
let connection_index = pick_index(env.connections.len(), &mut env.rng);
let state = &mut states[connection_index];
history.history.push(Execution::new(
connection_index,
state.interaction_pointer,
state.secondary_pointer,
));
let mut last_execution = last_execution.lock().unwrap();
last_execution.connection_index = connection_index;
last_execution.interaction_index = state.interaction_pointer;
last_execution.secondary_index = state.secondary_pointer;
// Execute the interaction for the selected connection
match execute_plan(&mut env, connection_index, plans, states) {
Ok(_) => {}
Err(err) => {
return ExecutionResult::new(history, Some(err));
}
}
// Check if the maximum time for the simulation has been reached
if now.elapsed().as_secs() >= env.opts.max_time_simulation as u64 {
return ExecutionResult::new(
history,
Some(turso_core::LimboError::InternalError(
"maximum time for simulation reached".into(),
)),
);
}
}
ExecutionResult::new(history, None)
}
fn execute_plan(
env: &mut SimulatorEnv,
connection_index: usize,
plans: &mut [Vec<Vec<Interaction>>],
states: &mut [InteractionPlanState],
) -> turso_core::Result<()> {
let connection = &env.connections[connection_index];
let plan = &mut plans[connection_index];
let state = &mut states[connection_index];
if state.interaction_pointer >= plan.len() {
return Ok(());
}
let interaction = &plan[state.interaction_pointer][state.secondary_pointer];
if let SimConnection::Disconnected = connection {
tracing::debug!("connecting {}", connection_index);
env.connections[connection_index] = SimConnection::LimboConnection(
env.db.as_ref().expect("db to be Some").connect().unwrap(),
);
} else {
match execute_interaction(env, connection_index, interaction, &mut state.stack) {
Ok(next_execution) => {
tracing::debug!("connection {} processed", connection_index);
// Move to the next interaction or property
match next_execution {
ExecutionContinuation::NextInteraction => {
if state.secondary_pointer + 1 >= plan[state.interaction_pointer].len() {
// If we have reached the end of the interactions for this property, move to the next property
state.interaction_pointer += 1;
state.secondary_pointer = 0;
} else {
// Otherwise, move to the next interaction
state.secondary_pointer += 1;
}
}
ExecutionContinuation::NextProperty => {
// Skip to the next property
state.interaction_pointer += 1;
state.secondary_pointer = 0;
}
}
}
Err(err) => {
tracing::error!("error {}", err);
return Err(err);
}
}
}
Ok(())
}

96
simulator/shrink/plan.rs
View File

@@ -1,7 +1,9 @@
use indexmap::IndexSet;
use crate::{
SandboxedResult, SimulatorEnv,
generation::{
plan::{Interaction, InteractionPlan, Interactions},
plan::{InteractionPlan, InteractionType, Interactions, InteractionsType},
property::Property,
},
model::Query,
@@ -17,21 +19,27 @@ impl InteractionPlan {
// - Shrink to multiple values by removing random interactions
// - Shrink properties by removing their extensions, or shrinking their values
let mut plan = self.clone();
let failing_property = &self.plan[failing_execution.interaction_index];
let mut depending_tables = failing_property.dependencies();
// let failing_property = &self.plan[failing_execution.interaction_index];
let mut depending_tables = IndexSet::new();
let interactions = failing_property.interactions();
let all_interactions = self.interactions_list_with_secondary_index();
// Index of the parent property where the interaction originated from
let secondary_interactions_index = all_interactions[failing_execution.interaction_index].0;
{
let mut idx = failing_execution.secondary_index;
let mut idx = failing_execution.interaction_index;
loop {
match &interactions[idx] {
Interaction::Query(query) => {
if all_interactions[idx].0 != secondary_interactions_index {
// Stop when we reach a different property
break;
}
match &all_interactions[idx].1.interaction {
InteractionType::Query(query) => {
depending_tables = query.dependencies();
break;
}
// Fault does not depend on
Interaction::Fault(..) => break,
InteractionType::Fault(..) => break,
_ => {
// In principle we should never fail this checked_sub.
// But if there is a bug in how we count the secondary index
@@ -50,15 +58,15 @@ impl InteractionPlan {
let before = self.plan.len();
// Remove all properties after the failing one
plan.plan.truncate(failing_execution.interaction_index + 1);
plan.plan.truncate(secondary_interactions_index + 1);
let mut idx = 0;
// Remove all properties that do not use the failing tables
plan.plan.retain_mut(|interactions| {
let retain = if idx == failing_execution.interaction_index {
if let Interactions::Property(
let retain = if idx == secondary_interactions_index {
if let InteractionsType::Property(
Property::FsyncNoWait { tables, .. } | Property::FaultyQuery { tables, .. },
) = interactions
) = &mut interactions.interactions
{
tables.retain(|table| depending_tables.contains(table));
}
@@ -71,7 +79,7 @@ impl InteractionPlan {
if has_table {
// Remove the extensional parts of the properties
if let Interactions::Property(p) = interactions {
if let InteractionsType::Property(p) = &mut interactions.interactions {
match p {
Property::InsertValuesSelect { queries, .. }
| Property::DoubleCreateFailure { queries, .. }
@@ -101,14 +109,16 @@ impl InteractionPlan {
.iter()
.any(|t| depending_tables.contains(t));
}
let is_fault = matches!(interactions, Interactions::Fault(..));
let is_fault = matches!(interactions.interactions, InteractionsType::Fault(..));
is_fault
|| (has_table
&& !matches!(
interactions,
Interactions::Query(Query::Select(_))
| Interactions::Property(Property::SelectLimit { .. })
| Interactions::Property(Property::SelectSelectOptimizer { .. })
interactions.interactions,
InteractionsType::Query(Query::Select(_))
| InteractionsType::Property(Property::SelectLimit { .. })
| InteractionsType::Property(
Property::SelectSelectOptimizer { .. }
)
))
};
idx += 1;
@@ -147,16 +157,19 @@ impl InteractionPlan {
};
let mut plan = self.clone();
let failing_property = &self.plan[failing_execution.interaction_index];
let interactions = failing_property.interactions();
let all_interactions = self.interactions_list_with_secondary_index();
let secondary_interactions_index = all_interactions[failing_execution.interaction_index].0;
{
let mut idx = failing_execution.secondary_index;
let mut idx = failing_execution.interaction_index;
loop {
match &interactions[idx] {
if all_interactions[idx].0 != secondary_interactions_index {
// Stop when we reach a different property
break;
}
match &all_interactions[idx].1.interaction {
// Fault does not depend on
Interaction::Fault(..) => break,
InteractionType::Fault(..) => break,
_ => {
// In principle we should never fail this checked_sub.
// But if there is a bug in how we count the secondary index
@@ -174,11 +187,11 @@ impl InteractionPlan {
let before = self.plan.len();
plan.plan.truncate(failing_execution.interaction_index + 1);
plan.plan.truncate(secondary_interactions_index + 1);
// phase 1: shrink extensions
for interaction in &mut plan.plan {
if let Interactions::Property(property) = interaction {
if let InteractionsType::Property(property) = &mut interaction.interactions {
match property {
Property::InsertValuesSelect { queries, .. }
| Property::DoubleCreateFailure { queries, .. }
@@ -187,7 +200,12 @@ impl InteractionPlan {
let mut temp_plan = InteractionPlan {
plan: queries
.iter()
.map(|q| Interactions::Query(q.clone()))
.map(|q| {
Interactions::new(
interaction.connection_index,
InteractionsType::Query(q.clone()),
)
})
.collect(),
};
@@ -196,14 +214,15 @@ impl InteractionPlan {
failing_execution,
result,
env.clone(),
secondary_interactions_index,
);
//temp_plan = Self::shrink_queries(temp_plan, failing_execution, result, env);
*queries = temp_plan
.plan
.into_iter()
.filter_map(|i| match i {
Interactions::Query(q) => Some(q),
.filter_map(|i| match i.interactions {
InteractionsType::Query(q) => Some(q),
_ => None,
})
.collect();
@@ -221,7 +240,13 @@ impl InteractionPlan {
}
// phase 2: shrink the entire plan
plan = Self::iterative_shrink(plan, failing_execution, result, env);
plan = Self::iterative_shrink(
plan,
failing_execution,
result,
env,
secondary_interactions_index,
);
let after = plan.plan.len();
@@ -240,9 +265,10 @@ impl InteractionPlan {
failing_execution: &Execution,
old_result: &SandboxedResult,
env: Arc<Mutex<SimulatorEnv>>,
secondary_interaction_index: usize,
) -> InteractionPlan {
for i in (0..plan.plan.len()).rev() {
if i == failing_execution.interaction_index {
if i == secondary_interaction_index {
continue;
}
let mut test_plan = plan.clone();
@@ -265,11 +291,9 @@ impl InteractionPlan {
let last_execution = Arc::new(Mutex::new(*failing_execution));
let result = SandboxedResult::from(
std::panic::catch_unwind(|| {
run_simulation(
env.clone(),
&mut [test_plan.clone()],
last_execution.clone(),
)
let interactions = test_plan.interactions_list();
run_simulation(env.clone(), interactions, last_execution.clone())
}),
last_execution,
);