use crate::common::{assert_reclaimed, assert_retirement, plan};
use cartel_sqlite::runtime::{self, Runtime};
use cartel_sqlite::{Action, Error, OpenOptions, Progress, RequestPhase, open};
use common::catalog::*;
use dope_core::driver::life::recovery::Policy;
use dope_core::driver::settings::demands::slots::Slots;
use dope_manifold::timing::Throughput;
use dope_runtime::{process::lifecycle, topology};
use dope_test::fibers;
use dope_test::harness::support::file::Directory;
use std::num::NonZero;
use std::path::Path;
use std::time;
pub mod common;
#[test]
fn work_limit_releases() -> Result<(), plan::Error> {
let runtime = Runtime::install(unsafe { runtime::raw::Permit::exclusive() })
.map_err(plan::Error::Runtime)?;
{
let base = Path::new(env!("CARGO_TARGET_TMPDIR"));
let dir = Directory::try_with(base, "cartel-sqlite-bounded-command", |removed| {
removed.expect("temporary directory cleanup")
})
.expect("temporary path");
let path = dir.path().join("catalog.db");
let target = open(runtime, &path)?
.commands(EXPENSIVE_COMMANDS)
.map_err(plan::Error::Client)?;
let result = plan::Plan::<Commands>::try_new(target, try_limits()?, try_memory()?)?
.try_install(Policy::new(assert_reclaimed, assert_retirement), |result| {
result.expect("SQLite scope")
});
assert!(matches!(result,
Err(plan::Failure::Install(
topology::InstallError::Install(open::Error::Step(Error::WorkLimit {
action: Action::Step, code
}))
)) if code.primary() == libsqlite3_sys::SQLITE_INTERRUPT
));
assert_eq!(
plan::Plan::try_new(open(runtime, &path)?, try_limits()?, try_memory()?)?
.try_count(
Bounded::COUNT,
Policy::new(assert_reclaimed, assert_retirement),
|result| result.expect("SQLite scope"),
)
.expect("SQLite count scope")?,
0
);
}
open_work_is_aggregate(runtime)?;
short_commands_exhaust_open_work(runtime)?;
use_connection_limits(runtime)?;
startup_timeout_releases(runtime)?;
readonly_reuses_schema(runtime)
}
fn open_work_is_aggregate(runtime: Runtime) -> Result<(), plan::Error> {
let base = Path::new(env!("CARGO_TARGET_TMPDIR"));
let mut bounded = try_limits()?;
bounded.work.open.progress = Progress::new(NonZero::new(70).ok_or(plan::Error::Progress)?);
let one = Directory::try_with(base, "cartel-sqlite-single-open-work", |removed| {
removed.expect("temporary directory cleanup")
})
.expect("temporary path");
let target = open(runtime, &one.path().join("catalog.db"))?
.commands(SINGLE_WORK_COMMANDS)
.map_err(plan::Error::Client)?;
plan::Plan::<Commands>::try_new(target, bounded, try_memory()?)?
.try_install(Policy::new(assert_reclaimed, assert_retirement), |result| {
result.expect("SQLite scope")
})
.expect("one command fits the open work budget");
let sum = Directory::try_with(base, "cartel-sqlite-aggregate-open-work", |removed| {
removed.expect("temporary directory cleanup")
})
.expect("temporary path");
let path = sum.path().join("catalog.db");
let target = open(runtime, &path)?
.commands(AGGREGATE_WORK_COMMANDS)
.map_err(plan::Error::Client)?;
let result = plan::Plan::<Commands>::try_new(target, bounded, try_memory()?)?
.try_install(Policy::new(assert_reclaimed, assert_retirement), |result| {
result.expect("SQLite scope")
});
assert!(matches!(result,
Err(plan::Failure::Install(
topology::InstallError::Install(open::Error::Step(Error::WorkLimit {
action: Action::Step, code
}))
)) if code.primary() == libsqlite3_sys::SQLITE_INTERRUPT
));
assert_eq!(
plan::Plan::try_new(open(runtime, &path)?, try_limits()?, try_memory()?)?
.try_count(
Aggregate::COUNT,
Policy::new(assert_reclaimed, assert_retirement),
|result| result.expect("SQLite scope"),
)
.expect("SQLite count scope")?,
1001
);
Ok(())
}
fn short_commands_exhaust_open_work(runtime: Runtime) -> Result<(), plan::Error> {
for (commands, fits) in [
(&SHORT_WORK_COMMANDS[..2], true),
(SHORT_WORK_COMMANDS, false),
] {
let target = OpenOptions::new()
.open_memory(runtime)
.map_err(plan::Error::Client)?
.commands(commands)
.map_err(plan::Error::Client)?;
let result = plan::Plan::<Commands>::try_new(target, try_limits()?, try_memory()?)?
.try_install(Policy::new(assert_reclaimed, assert_retirement), |result| {
result.expect("SQLite scope")
});
if fits {
result.expect("short command prefix fits the open work budget");
} else {
assert!(matches!(
result,
Err(plan::Failure::Install(topology::InstallError::Install(
open::Error::Step(Error::WorkExhausted)
)))
));
}
}
Ok(())
}
fn use_connection_limits(runtime: Runtime) -> Result<(), plan::Error> {
let base = Path::new(env!("CARGO_TARGET_TMPDIR"));
let dir = Directory::try_with(base, "cartel-sqlite-local-command", |removed| {
removed.expect("temporary directory cleanup")
})
.expect("temporary path");
let path = dir.path().join("catalog.db");
let target = open(runtime, &path)?
.commands(OVERSIZED_COMMANDS)
.map_err(plan::Error::Client)?;
let result = plan::Plan::<Commands>::try_new(target, try_limits()?, try_memory()?)?
.try_install(Policy::new(assert_reclaimed, assert_retirement), |result| {
result.expect("SQLite scope")
});
assert!(matches!(result,
Err(plan::Failure::Install(
topology::InstallError::Install(open::Error::Step(Error::Sqlite {
action: Action::Step, code
}))
)) if code.primary() == libsqlite3_sys::SQLITE_TOOBIG
));
assert_eq!(
plan::Plan::try_new(open(runtime, &path)?, try_limits()?, try_memory()?)?
.try_count(
Compact::COUNT,
Policy::new(assert_reclaimed, assert_retirement),
|result| result.expect("SQLite scope"),
)
.expect("SQLite count scope")?,
0
);
Ok(())
}
fn startup_timeout_releases(runtime: Runtime) -> Result<(), plan::Error> {
let base = Path::new(env!("CARGO_TARGET_TMPDIR"));
let dir = Directory::try_with(base, "cartel-sqlite-startup-contention", |removed| {
removed.expect("temporary directory cleanup")
})
.expect("temporary path");
let path = dir.path().join("catalog.db");
assert_eq!(
plan::Plan::try_new(
open(runtime, &path)?
.commands(STABLE_COMMANDS)
.map_err(plan::Error::Client)?,
try_limits()?,
try_memory()?
)
.and_then(|plan| plan
.try_count(
Stable::COUNT,
Policy::new(assert_reclaimed, assert_retirement),
|result| result.expect("SQLite scope"),
)
.expect("SQLite count scope"))?,
1
);
let held = open(runtime, &path)?;
let rival = open(runtime, &path)?
.commands(CONTENDED_COMMANDS)
.map_err(plan::Error::Client)?;
let yielding = &path;
let result = plan::Plan::<Stable>::try_new(held, try_limits()?, try_memory()?)?.try_run(
|driver, mut conn| {
let mut tx = fibers::TEST
.try_run(driver, conn.transaction())
.map_err(plan::Error::Run)?
.map_err(plan::Error::Call)?;
fibers::TEST
.try_run(driver, tx.query().run_no_rows(Stable::INSERT, (2_i64, ())))
.map_err(plan::Error::Run)?
.map_err(plan::Error::Call)?;
use lifecycle::Task as _;
lifecycle::Entry(move || {
for (retries, expected) in [
(NonZero::<u32>::MIN, 2..=2),
(NonZero::<u32>::MIN.saturating_add(4_095), 2..=100),
] {
let mut limits = try_limits().expect("SQLite limits");
limits.work.open.busy.window = time::Duration::MAX;
limits.work.open.busy.interval = time::Duration::ZERO;
limits.work.open.busy.retries = retries;
open(runtime, yielding)
.expect("yielding database")
.commands(CONTENDED_COMMANDS)
.expect("yielding startup commands")
.plan::<Commands>(limits, try_memory().expect("SQLite memory"))
.expect("yielding contention plan")
.for_profile::<Throughput>()
.expect("yielding contention resources")
.try_create(
(),
|worker| {
let mut cleanup = open::Completion::new();
let mut polls = 0;
assert!(matches!(
worker.install_with(|open, stage| plan::Polls {
root: open.install(stage, &mut cleanup),
count: &mut polls,
}),
Err(topology::InstallError::Install(open::Error::Step(
Error::Busy {
action: Action::Step,
code,
}
))) if code.primary() == libsqlite3_sys::SQLITE_BUSY
));
assert!(expected.contains(&polls), "startup polls: {polls}");
assert!(cleanup.take().is_ok(), "yielded startup cleanup");
},
Policy::new(assert_reclaimed, assert_retirement),
|outcome| {
outcome
.as_ref()
.map_err(|error| error.cause())
.expect("yielding SQLite scope");
},
);
}
let mut limits = try_limits().expect("SQLite limits");
limits.work.open.busy.window = time::Duration::from_millis(20);
let plan = plan::Plan::<Commands>::try_with(
rival,
limits,
try_memory().expect("SQLite memory"),
Slots::ZERO,
);
assert!(plan.is_ok(), "contention plan construction failed");
let plan = plan.expect("contention plan construction failed");
let result = plan
.try_install(Policy::new(assert_reclaimed, assert_retirement), |result| {
result.expect("SQLite scope")
});
assert!(
matches!(result, Err(plan::Failure::Install(
topology::InstallError::Install(open::Error::Step(Error::RequestTimeout {
elapsed,
action: Action::Step,
code,
}))
)) if elapsed.phase() == RequestPhase::Contention && code.primary() == 5),
"contention must exhaust the startup deadline"
);
})
.with(|| ())
.expect("contention worker failed");
fibers::TEST
.try_run(driver, tx.rollback())
.map_err(plan::Error::Run)?
.map_err(plan::Error::Call)?;
Ok(())
},
Policy::new(assert_reclaimed, assert_retirement),
|result| result.expect("SQLite scope"),
);
result.expect("command contention scope failed")?;
assert!(
plan::Plan::<Commands>::try_new(
open(runtime, &path)?
.commands(CONTENDED_COMMANDS)
.map_err(plan::Error::Client)?,
try_limits()?,
try_memory()?
)
.expect("SQLite runtime succeeds")
.try_install(Policy::new(assert_reclaimed, assert_retirement), |result| {
result.expect("SQLite scope")
},)
.is_ok(),
"SQLite installation succeeds"
);
Ok(())
}
fn readonly_reuses_schema(runtime: Runtime) -> Result<(), plan::Error> {
let base = Path::new(env!("CARGO_TARGET_TMPDIR"));
let dir = Directory::try_with(base, "cartel-sqlite-read-only", |removed| {
removed.expect("temporary directory cleanup")
})
.expect("temporary path");
let path = dir.path().join("catalog.db");
let target = open(runtime, &path)?
.commands(STABLE_COMMANDS)
.map_err(plan::Error::Client)?;
plan::Plan::<Stable>::try_new(target, try_limits()?, try_memory()?)?
.try_install(Policy::new(assert_reclaimed, assert_retirement), |result| {
result.expect("SQLite scope")
})
.expect("fresh schema installation succeeds");
assert_eq!(
plan::Plan::try_new(readonly(runtime, &path)?, try_limits()?, try_memory()?)?
.try_count(
Stable::COUNT,
Policy::new(assert_reclaimed, assert_retirement),
|result| result.expect("SQLite scope"),
)
.expect("SQLite count scope")?,
1
);
Ok(())
}