#![cfg(feature = "project-export")]
use std::collections::{BTreeMap, HashMap};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use async_trait::async_trait;
use futures::StreamExt;
use futures::io::{AsyncReadExt, AsyncWriteExt};
use sha2::{Digest, Sha256};
use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
use boatramp_core::Storage;
use boatramp_core::compute::{ComputeBackend, ReplicaPhase};
use boatramp_core::deploy::DeployStore;
use boatramp_core::env::EnvSource;
use boatramp_core::envelope::KeyEnvelope;
use boatramp_core::kv::KvStore;
use boatramp_core::project::ProjectRef;
use boatramp_core::project_export::{
ExportOptions, IdentityChoice, ImportOptions, ImportReport, ProjectExport, ProjectExportError,
};
use boatramp_core::sql::SqlValue;
use boatramp_storage::sql_libsql::LibsqlSqlBackends;
use crate::config::ExternalDatabaseConfig;
use crate::managed_sql::{ManagedSqlCredentials, operator_target};
use crate::pg_client::{
MysqlConnection, PgConnection, mysql_admin_invocation, mysql_restore_invocation,
mysql_shadow_invocation, mysqlbinlog_capture_invocation, mysqldump_capture_base_invocation,
mysqldump_invocation, mysqldump_tables_only_invocation, pg_dump_invocation,
pg_restore_invocation, pg_restore_shadow_invocation, pg_server_major, psql_invocation,
};
use crate::project_bundle::{
self, IDENTITY_MEMBER_PATH, KV_MEMBER_PATH, MemberKind, MemberSpec, PgMemberInfo, SourceInfo,
};
use boatramp_storage::ExternalSqlKind;
use boatramp_storage::tenant_provision::quote_ident;
type Perr = ProjectExportError;
fn other<E: std::fmt::Display>(e: E) -> Perr {
Perr::Other(e.to_string())
}
fn now_secs() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
const STREAM_BYTE_CHANNEL: usize = 16;
pub const DEFAULT_CHUNK_SIZE: usize = 64 * 1024 * 1024;
const DEFAULT_IMPORT_MEMBER_CAP: u64 = 256 * 1024 * 1024 * 1024;
const DEFAULT_IMPORT_TOTAL_CAP: u64 = 1024 * 1024 * 1024 * 1024;
const MYSQL_DELTA_MEMBER_MAX: u64 = 2 * 1024 * 1024 * 1024;
fn kind_is_postgres(kind: &str) -> bool {
matches!(
kind.to_ascii_lowercase().as_str(),
"postgres" | "postgresql" | "pg"
)
}
#[allow(dead_code)]
fn kind_is_libsql(kind: &str) -> bool {
matches!(
kind.to_ascii_lowercase().as_str(),
"libsql" | "sqlite" | "sqlite3"
)
}
fn kind_is_mysql(kind: &str) -> bool {
matches!(kind.to_ascii_lowercase().as_str(), "mysql" | "mariadb")
}
fn is_managed_binding(cfg: &ExternalDatabaseConfig) -> bool {
cfg.compute.as_deref().is_some_and(|c| !c.is_empty())
}
fn parse_sql_url(url: &str) -> Option<(String, String, String, u16, String)> {
let (scheme, rest) = url.split_once("://")?;
let default_port = match scheme {
"postgres" | "postgresql" => 5432,
"mysql" | "mariadb" => 3306,
_ => return None,
};
let (userinfo, hostpart) = rest.rsplit_once('@')?;
let (user, password) = match userinfo.split_once(':') {
Some((u, p)) => (u.to_string(), p.to_string()),
None => (userinfo.to_string(), String::new()),
};
let (hostport, db) = hostpart.split_once('/')?;
let db = db.split(['?', '&']).next().unwrap_or(db).to_string();
let (host, port) = match hostport.rsplit_once(':') {
Some((h, p)) => (h.to_string(), p.parse().ok()?),
None => (hostport.to_string(), default_port),
};
Some((user, password, host, port, db))
}
fn pg_member_path(db: &str) -> String {
format!("sql/{db}.pgc")
}
fn db_from_pg_member(path: &str) -> Option<&str> {
path.strip_prefix("sql/")
.and_then(|s| s.strip_suffix(".pgc"))
}
fn pgwal_member_path(db: &str) -> String {
format!("sql/{db}.pgwal")
}
fn db_from_pgwal_member(path: &str) -> Option<&str> {
path.strip_prefix("sql/")
.and_then(|s| s.strip_suffix(".pgwal"))
}
fn mysql_member_path(db: &str) -> String {
format!("sql/{db}.sql")
}
fn db_from_mysql_member(path: &str) -> Option<&str> {
path.strip_prefix("sql/")
.and_then(|s| s.strip_suffix(".sql"))
}
fn mysqlbinlog_member_path(db: &str) -> String {
format!("sql/{db}.mysqlbinlog")
}
fn db_from_mysqlbinlog_member(path: &str) -> Option<&str> {
path.strip_prefix("sql/")
.and_then(|s| s.strip_suffix(".mysqlbinlog"))
}
fn libsql_member_path(project: &str, site: &str) -> String {
format!("libsql/{project}/{site}/default.sqlite")
}
fn site_from_libsql_member(path: &str) -> Option<&str> {
let body = path
.strip_prefix("libsql/")?
.strip_suffix("/default.sqlite")?;
Some(body.rsplit_once('/').map(|(_, s)| s).unwrap_or(body))
}
fn libsql_changeset_member_path(project: &str, site: &str) -> String {
format!("libsql/{project}/{site}/default.changeset")
}
fn site_from_libsql_changeset_member(path: &str) -> Option<&str> {
let body = path
.strip_prefix("libsql/")?
.strip_suffix("/default.changeset")?;
Some(body.rsplit_once('/').map(|(_, s)| s).unwrap_or(body))
}
fn check_project_slug(project: &str) -> Result<(), Perr> {
boatramp_core::project::validate_resource_name("project", project)
.map_err(|e| Perr::Other(e.to_string()))
}
fn data_only_skew_blocks(src_fp: Option<&str>, tgt_fp: Option<&str>) -> bool {
!matches!((src_fp, tgt_fp), (Some(s), Some(t)) if s == t)
}
fn target_blob_key(member_path: &str, source: &str, target: &str) -> Result<String, Perr> {
let src_key = member_path.strip_prefix("blobs/").ok_or_else(|| {
Perr::Other(format!(
"blob member path {member_path:?} is missing its 'blobs/' prefix"
))
})?;
let src_pref = format!("hblob/{source}/");
let rest = src_key.strip_prefix(&src_pref).ok_or_else(|| {
Perr::Other(format!(
"blob member {src_key:?} is not under the source project scope {src_pref:?}; \
refusing to import a wrong-scope blob (cross-tenant isolation)"
))
})?;
Ok(format!("hblob/{target}/{rest}"))
}
fn blob_project_prefix(project: &str) -> String {
format!("hblob/{project}/")
}
fn choose_host(endpoint_host: &str, direct_host: Option<&str>, published: bool) -> String {
match (published, direct_host) {
(true, Some(h)) if !h.is_empty() => h.to_string(),
_ => endpoint_host.to_string(),
}
}
fn scalar_text(rows: &boatramp_core::sql::SqlRows) -> Option<String> {
match rows.rows.first().and_then(|r| r.first()) {
Some(SqlValue::Text(s)) => Some(s.clone()),
Some(SqlValue::Integer(n)) => Some(n.to_string()),
_ => None,
}
}
fn delta_capture_not_restarted_message(db: &str) -> String {
format!(
"database {db:?}: delta export refused — `delta_capture` is enabled but the server has \
not been restarted to adopt it yet (true-delta needs wal_level=logical / binary \
logging, which only take effect at server start); ensure `delta_capture: true` is set \
and run `boatramp db restart --db {db}` (or apply with `restart_policy: auto`) before \
a delta export of this database"
)
}
fn fingerprint_rows(rows: &boatramp_core::sql::SqlRows) -> String {
let mut hasher = Sha256::new();
for row in &rows.rows {
for cell in row {
let s = match cell {
SqlValue::Text(s) => s.clone(),
SqlValue::Integer(n) => n.to_string(),
other => format!("{other:?}"),
};
hasher.update((s.len() as u64).to_le_bytes());
hasher.update(s.as_bytes());
}
hasher.update(b"\x00");
}
hex::encode(hasher.finalize())
}
#[derive(Debug, Clone)]
pub enum IdentityEnvelope {
Local { kek_file: PathBuf },
Vault,
None,
}
const IDENTITY_PAYLOAD_MAGIC: &[u8; 6] = b"BRIDP1";
fn identity_payload_bytes(kek: &[u8]) -> Result<Vec<u8>, Perr> {
if kek.len() > u8::MAX as usize {
return Err(Perr::Other("KEK too large for the identity payload".into()));
}
let mut v = Vec::with_capacity(IDENTITY_PAYLOAD_MAGIC.len() + 2 + kek.len());
v.extend_from_slice(IDENTITY_PAYLOAD_MAGIC);
v.push(1u8);
v.push(kek.len() as u8);
v.extend_from_slice(kek);
Ok(v)
}
fn parse_identity_payload(bytes: &[u8]) -> Result<Vec<u8>, Perr> {
let rest = bytes
.strip_prefix(IDENTITY_PAYLOAD_MAGIC.as_slice())
.ok_or_else(|| Perr::Other("identity payload: bad magic".into()))?;
if rest.len() < 2 {
return Err(Perr::Other("identity payload: truncated header".into()));
}
if rest[0] != 1 {
return Err(Perr::Other(format!(
"identity payload: unsupported version {}",
rest[0]
)));
}
let kek_len = rest[1] as usize;
let kek = &rest[2..];
if kek.len() != kek_len {
return Err(Perr::Other("identity payload: KEK length mismatch".into()));
}
Ok(kek.to_vec())
}
async fn write_kek_file(path: &std::path::Path, kek: &[u8]) -> Result<(), Perr> {
if let Some(parent) = path.parent() {
tokio::fs::create_dir_all(parent).await.map_err(other)?;
}
let path = path.to_path_buf();
let kek = kek.to_vec();
tokio::task::spawn_blocking(move || -> std::io::Result<()> {
#[cfg(unix)]
{
use std::io::Write;
use std::os::unix::fs::OpenOptionsExt;
let mut f = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
.open(&path)?;
f.write_all(&kek)
}
#[cfg(not(unix))]
{
std::fs::write(&path, &kek)
}
})
.await
.map_err(other)?
.map_err(other)
}
pub struct NodeProjectExport {
databases: BTreeMap<String, ExternalDatabaseConfig>,
deploy: DeployStore,
kv: Arc<dyn KvStore>,
blobs: Arc<dyn Storage>,
libsql: Option<Arc<LibsqlSqlBackends>>,
backend: Option<Arc<dyn ComputeBackend>>,
envelope: Option<Arc<dyn KeyEnvelope>>,
operator_sql: Option<Arc<dyn boatramp_core::sql::OperatorSql>>,
env_source: Arc<dyn EnvSource>,
published_endpoint: bool,
identity_source: Option<IdentityEnvelope>,
work_dir: PathBuf,
connect_timeout_secs: u32,
pg_client_image: Option<String>,
mysql_client_image: Option<String>,
chunk_size: usize,
import_member_cap: u64,
import_total_cap: u64,
trusted_extensions: Vec<String>,
resumable_base: Option<PathBuf>,
}
impl NodeProjectExport {
pub fn new(
databases: BTreeMap<String, ExternalDatabaseConfig>,
deploy: DeployStore,
kv: Arc<dyn KvStore>,
blobs: Arc<dyn Storage>,
work_dir: PathBuf,
) -> Self {
Self {
databases,
deploy,
kv,
blobs,
libsql: None,
backend: None,
envelope: None,
operator_sql: None,
env_source: Arc::new(boatramp_core::env::SystemEnv),
published_endpoint: true,
identity_source: None,
work_dir,
connect_timeout_secs: 10,
pg_client_image: None,
mysql_client_image: None,
chunk_size: DEFAULT_CHUNK_SIZE,
import_member_cap: DEFAULT_IMPORT_MEMBER_CAP,
import_total_cap: DEFAULT_IMPORT_TOTAL_CAP,
trusted_extensions: Vec::new(),
resumable_base: None,
}
}
#[must_use]
pub fn with_trusted_extensions(mut self, extensions: Vec<String>) -> Self {
self.trusted_extensions = extensions;
self
}
#[must_use]
pub fn with_chunk_size(mut self, chunk_size: usize) -> Self {
self.chunk_size = chunk_size.max(1);
self
}
#[must_use]
pub fn with_import_caps(mut self, member_cap: u64, total_cap: u64) -> Self {
self.import_member_cap = member_cap;
self.import_total_cap = total_cap;
self
}
#[must_use]
pub fn with_client_images(
mut self,
pg_client_image: Option<String>,
mysql_client_image: Option<String>,
) -> Self {
self.pg_client_image = pg_client_image;
self.mysql_client_image = mysql_client_image;
self
}
fn pg_image(
&self,
server_version_num: &str,
) -> Result<String, boatramp_core::compute::VersionMapError> {
match &self.pg_client_image {
Some(pinned) => Ok(pinned.clone()),
None => boatramp_core::compute::pg_client_image_for_server_version(server_version_num),
}
}
fn mysql_image(
&self,
version: &str,
) -> Result<String, boatramp_core::compute::VersionMapError> {
match &self.mysql_client_image {
Some(pinned) => Ok(pinned.clone()),
None => boatramp_core::compute::mysql_client_image_for_server_version(version),
}
}
#[must_use]
pub fn with_env_source(mut self, env_source: Arc<dyn EnvSource>) -> Self {
self.env_source = env_source;
self
}
#[must_use]
pub fn with_managed_pg(
mut self,
backend: Arc<dyn ComputeBackend>,
envelope: Arc<dyn KeyEnvelope>,
operator_sql: Arc<dyn boatramp_core::sql::OperatorSql>,
published_endpoint: bool,
) -> Self {
self.backend = Some(backend);
self.envelope = Some(envelope);
self.operator_sql = Some(operator_sql);
self.published_endpoint = published_endpoint;
self
}
#[must_use]
pub fn with_libsql(mut self, libsql: Arc<LibsqlSqlBackends>) -> Self {
self.libsql = Some(libsql);
self
}
#[must_use]
pub fn with_resumable_base(mut self, base: Option<PathBuf>) -> Self {
self.resumable_base = base;
self
}
pub(crate) async fn resolve_resumable_dir(
&self,
dir_name: &str,
create: bool,
) -> Result<PathBuf, Perr> {
use std::path::Component;
let base = self.resumable_base.as_ref().ok_or_else(|| {
Perr::Other(
"resumable export/import is not enabled on this node; set \
`[serve] resumable_export_dir` to a base directory"
.into(),
)
})?;
if dir_name.is_empty() {
return Err(Perr::Other("resumable directory name is required".into()));
}
let rel = Path::new(dir_name);
if rel.is_absolute() {
return Err(Perr::Other(
"resumable directory must be a relative name under the configured base, \
not an absolute path"
.into(),
));
}
for comp in rel.components() {
if !matches!(comp, Component::Normal(_)) {
return Err(Perr::Other(
"resumable directory name must not contain `..`, `.`, or a root/prefix \
component"
.into(),
));
}
}
if create {
tokio::fs::create_dir_all(base).await.map_err(other)?;
}
let canon_base = tokio::fs::canonicalize(base).await.map_err(|e| {
Perr::Other(format!(
"resumable base {} is not accessible: {e}",
base.display()
))
})?;
let joined = canon_base.join(rel);
if create {
tokio::fs::create_dir_all(&joined).await.map_err(other)?;
} else if !tokio::fs::try_exists(&joined).await.map_err(other)? {
return Err(Perr::Other(format!(
"resumable directory {dir_name:?} does not exist under the configured base"
)));
}
let canon = tokio::fs::canonicalize(&joined).await.map_err(other)?;
if !canon.starts_with(&canon_base) {
return Err(Perr::Other(
"resumable directory escapes the configured base (symlink?)".into(),
));
}
Ok(canon)
}
#[must_use]
pub fn with_identity_source(mut self, source: IdentityEnvelope) -> Self {
self.identity_source = Some(source);
self
}
fn require_backend(&self) -> Result<&Arc<dyn ComputeBackend>, Perr> {
self.backend.as_ref().ok_or_else(|| {
Perr::Unavailable(
"pg/mysql export/import needs a compute backend to run the client container — \
none on this node"
.into(),
)
})
}
fn require_op(&self) -> Result<&Arc<dyn boatramp_core::sql::OperatorSql>, Perr> {
self.operator_sql.as_ref().ok_or_else(|| {
Perr::Unavailable(
"pg/mysql export/import needs the operator-SQL seam (a managed SQL binding) — \
not configured on this node"
.into(),
)
})
}
async fn assert_pg_delta_capture_live(&self, project: &str, db: &str) -> Result<(), Perr> {
let op = self.require_op()?;
if let Ok(rows) = op.query(project, db, "SHOW wal_level").await
&& let Some(level) = scalar_text(&rows)
&& !level.trim().eq_ignore_ascii_case("logical")
{
return Err(Perr::Other(delta_capture_not_restarted_message(db)));
}
Ok(())
}
async fn assert_mysql_delta_capture_live(&self, project: &str, db: &str) -> Result<(), Perr> {
let op = self.require_op()?;
if let Ok(rows) = op.query(project, db, "SELECT @@GLOBAL.log_bin").await
&& let Some(v) = scalar_text(&rows)
&& v.trim() != "1"
{
return Err(Perr::Other(delta_capture_not_restarted_message(db)));
}
Ok(())
}
async fn resolve_endpoint(
&self,
project: &str,
db: &str,
cfg: &ExternalDatabaseConfig,
) -> Result<SqlEndpoint, Perr> {
if cfg.compute.as_deref().is_some_and(|c| !c.is_empty()) {
self.resolve_managed_endpoint(project, db, cfg).await
} else if !cfg.url_env.is_empty() {
let url = self.env_source.get(&cfg.url_env).ok_or_else(|| {
Perr::Other(format!(
"database {db:?}: external URL env var {} is unset",
cfg.url_env
))
})?;
let (user, password, host, port, database) = parse_sql_url(&url).ok_or_else(|| {
Perr::Other(format!(
"database {db:?}: external URL in {} is not a parseable \
postgres://|mysql:// DSN",
cfg.url_env
))
})?;
Ok(SqlEndpoint {
host,
port,
user,
password,
database,
})
} else {
Err(Perr::Other(format!(
"database {db:?}: neither compute-backed (managed) nor `url_env` (external) — \
nothing to dump"
)))
}
}
async fn resolve_managed_endpoint(
&self,
project: &str,
db: &str,
cfg: &ExternalDatabaseConfig,
) -> Result<SqlEndpoint, Perr> {
let backend = self.require_backend()?;
let envelope = self.envelope.as_ref().ok_or_else(|| {
Perr::Unavailable(format!(
"managed database {db:?} needs a [secrets] envelope to unseal its credential — \
none on this node"
))
})?;
let target = operator_target(cfg, project, db).map_err(other)?;
let states = self
.deploy
.list_replica_states(ProjectRef::new(&target.endpoint_project), &target.workload)
.await
.map_err(other)?;
let replica = states
.into_iter()
.find(|s| s.phase == ReplicaPhase::Running && s.healthy)
.ok_or_else(|| {
Perr::Other(format!(
"database {db:?}: no healthy running replica of workload {:?} to dump/restore",
target.workload
))
})?;
let direct = if self.published_endpoint {
backend.replica_direct_host(&replica.handle).await
} else {
None
};
let host = choose_host(
&replica.endpoint.host,
direct.as_deref(),
self.published_endpoint,
);
let password = ManagedSqlCredentials::new(self.kv.clone(), envelope.clone())
.password(&target.cred_project, &target.cred_workload)
.await
.map_err(Perr::Other)?;
Ok(SqlEndpoint {
host,
port: replica.endpoint.port,
user: target.user,
password,
database: target.database,
})
}
async fn resolve_pg(
&self,
project: &str,
db: &str,
cfg: &ExternalDatabaseConfig,
) -> Result<(PgConnection, String, String), Perr> {
let op = self.require_op()?;
let ep = self.resolve_endpoint(project, db, cfg).await?;
let banner = op
.query(project, db, "SHOW server_version_num")
.await
.map_err(other)?;
let server_version_num = scalar_text(&banner).ok_or_else(|| {
Perr::Other(format!(
"database {db:?}: empty SHOW server_version_num result"
))
})?;
let image = self
.pg_image(&server_version_num)
.map_err(|e| Perr::Other(format!("database {db:?}: {e}")))?;
Ok((
ep.into_pg(self.connect_timeout_secs),
image,
server_version_num,
))
}
async fn resolve_mysql(
&self,
project: &str,
db: &str,
cfg: &ExternalDatabaseConfig,
) -> Result<(MysqlConnection, String, String), Perr> {
let op = self.require_op()?;
let ep = self.resolve_endpoint(project, db, cfg).await?;
let banner = op
.query(project, db, "SELECT VERSION()")
.await
.map_err(other)?;
let version = scalar_text(&banner).ok_or_else(|| {
Perr::Other(format!("database {db:?}: empty SELECT VERSION() result"))
})?;
let image = self
.mysql_image(&version)
.map_err(|e| Perr::Other(format!("database {db:?}: {e}")))?;
Ok((ep.into_mysql(self.connect_timeout_secs), image, version))
}
async fn schema_fingerprint(&self, project: &str, db: &str) -> Option<String> {
let op = self.operator_sql.as_ref()?;
let sql = "SELECT id, ordinal, content_hash FROM boatramp_migrations.schema_migrations \
ORDER BY ordinal, id";
match op.query(project, db, sql).await {
Ok(rows) => Some(fingerprint_rows(&rows)),
Err(_) => None,
}
}
async fn run_sql_client(
&self,
image: &str,
argv: &[String],
env: &[(String, String)],
stdin_path: Option<&Path>,
stdout_path: &Path,
) -> Result<i32, Perr> {
let backend = self.require_backend()?;
let out = tokio::fs::File::create(stdout_path).await.map_err(other)?;
let mut sink = out.compat_write();
let code = if let Some(inp) = stdin_path {
let f = tokio::fs::File::open(inp).await.map_err(other)?;
let mut rdr = f.compat();
backend
.run_ephemeral(image, argv, env, Some(&mut rdr), &mut sink)
.await
.map_err(other)?
} else {
backend
.run_ephemeral(image, argv, env, None, &mut sink)
.await
.map_err(other)?
};
Ok(code)
}
async fn run_sql_capture(
&self,
image: &str,
argv: &[String],
env: &[(String, String)],
stdin_path: Option<&Path>,
stage_dir: &Path,
) -> Result<(i32, String), Perr> {
let out = stage_dir.join(format!("cap-{}.out", shadow_suffix()));
let merged = merge_stderr_argv(argv);
let code = self
.run_sql_client(image, &merged, env, stdin_path, &out)
.await?;
let s = tokio::fs::read_to_string(&out).await.unwrap_or_default();
let _ = tokio::fs::remove_file(&out).await;
Ok((code, s.trim().to_string()))
}
async fn run_sql_capture_nomerge(
&self,
image: &str,
argv: &[String],
env: &[(String, String)],
stdin_path: Option<&Path>,
stage_dir: &Path,
) -> Result<(i32, String), Perr> {
let out = stage_dir.join(format!("cap-{}.out", shadow_suffix()));
let code = self
.run_sql_client(image, argv, env, stdin_path, &out)
.await?;
let s = tokio::fs::read_to_string(&out).await.unwrap_or_default();
let _ = tokio::fs::remove_file(&out).await;
Ok((code, s.trim().to_string()))
}
}
fn mysql_eph_suffix() -> String {
let mut b = [0u8; 8];
getrandom::getrandom(&mut b).expect("os rng for mysql ephemeral suffix");
b.iter().map(|x| format!("{x:02x}")).collect()
}
fn shadow_suffix() -> String {
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
format!("{nanos}_{}", std::process::id())
}
fn pg_lit(s: &str) -> String {
format!("'{}'", s.replace('\'', "''"))
}
fn sh_join(argv: &[String]) -> String {
argv.iter()
.map(|a| format!("'{}'", a.replace('\'', "'\\''")))
.collect::<Vec<_>>()
.join(" ")
}
fn merge_stderr_argv(argv: &[String]) -> Vec<String> {
vec![
"sh".to_string(),
"-c".to_string(),
format!("exec {} 2>&1", sh_join(argv)),
]
}
struct SqlEndpoint {
host: String,
port: u16,
user: String,
password: String,
database: String,
}
impl SqlEndpoint {
fn into_pg(self, connect_timeout_secs: u32) -> PgConnection {
PgConnection {
host: self.host,
port: self.port,
user: self.user,
password: self.password,
database: self.database,
connect_timeout_secs,
}
}
fn into_mysql(self, connect_timeout_secs: u32) -> MysqlConnection {
MysqlConnection {
host: self.host,
port: self.port,
user: self.user,
password: self.password,
database: self.database,
connect_timeout_secs,
}
}
}
struct ChannelWriter {
tx: tokio::sync::mpsc::Sender<Vec<u8>>,
}
impl Write for ChannelWriter {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.tx.blocking_send(buf.to_vec()).map_err(|_| {
std::io::Error::new(std::io::ErrorKind::BrokenPipe, "export sink closed")
})?;
Ok(buf.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
struct ChannelReader {
rx: tokio::sync::mpsc::Receiver<Vec<u8>>,
buf: Vec<u8>,
pos: usize,
}
impl ChannelReader {
fn new(rx: tokio::sync::mpsc::Receiver<Vec<u8>>) -> Self {
Self {
rx,
buf: Vec::new(),
pos: 0,
}
}
}
impl Read for ChannelReader {
fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
while self.pos >= self.buf.len() {
match self.rx.blocking_recv() {
Some(chunk) if !chunk.is_empty() => {
self.buf = chunk;
self.pos = 0;
}
Some(_) => continue, None => return Ok(0), }
}
let n = std::cmp::min(out.len(), self.buf.len() - self.pos);
out[..n].copy_from_slice(&self.buf[self.pos..self.pos + n]);
self.pos += n;
Ok(n)
}
}
struct AsyncChannelWriter {
sender: tokio_util::sync::PollSender<Vec<u8>>,
}
impl AsyncChannelWriter {
fn new(tx: tokio::sync::mpsc::Sender<Vec<u8>>) -> Self {
Self {
sender: tokio_util::sync::PollSender::new(tx),
}
}
}
impl futures::io::AsyncWrite for AsyncChannelWriter {
fn poll_write(
self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<std::io::Result<usize>> {
use std::task::Poll;
let me = self.get_mut();
match me.sender.poll_reserve(cx) {
Poll::Ready(Ok(())) => match me.sender.send_item(buf.to_vec()) {
Ok(()) => Poll::Ready(Ok(buf.len())),
Err(_) => Poll::Ready(Err(std::io::Error::new(
std::io::ErrorKind::BrokenPipe,
"export dump sink closed",
))),
},
Poll::Ready(Err(_)) => Poll::Ready(Err(std::io::Error::new(
std::io::ErrorKind::BrokenPipe,
"export dump sink closed",
))),
Poll::Pending => Poll::Pending,
}
}
fn poll_flush(
self: std::pin::Pin<&mut Self>,
_cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
std::task::Poll::Ready(Ok(()))
}
fn poll_close(
self: std::pin::Pin<&mut Self>,
_cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
self.get_mut().sender.close();
std::task::Poll::Ready(Ok(()))
}
}
struct AsyncChannelReader {
rx: tokio::sync::mpsc::Receiver<Vec<u8>>,
buf: Vec<u8>,
pos: usize,
}
impl AsyncChannelReader {
fn new(rx: tokio::sync::mpsc::Receiver<Vec<u8>>) -> Self {
Self {
rx,
buf: Vec::new(),
pos: 0,
}
}
}
impl futures::io::AsyncRead for AsyncChannelReader {
fn poll_read(
self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
out: &mut [u8],
) -> std::task::Poll<std::io::Result<usize>> {
use std::task::Poll;
let me = self.get_mut();
while me.pos >= me.buf.len() {
match me.rx.poll_recv(cx) {
Poll::Ready(Some(chunk)) if !chunk.is_empty() => {
me.buf = chunk;
me.pos = 0;
}
Poll::Ready(Some(_)) => continue, Poll::Ready(None) => return Poll::Ready(Ok(0)), Poll::Pending => return Poll::Pending,
}
}
let n = std::cmp::min(out.len(), me.buf.len() - me.pos);
out[..n].copy_from_slice(&me.buf[me.pos..me.pos + n]);
me.pos += n;
Poll::Ready(Ok(n))
}
}
enum StageJob {
Identity { container: Vec<u8> },
Kv { created_at: u64 },
Pg {
conn: PgConnection,
image: String,
data_only: bool,
server_version_num: String,
no_publications: bool,
},
Mysql {
conn: MysqlConnection,
image: String,
data_only: bool,
},
Libsql { qsite: String },
Blob { key: String },
KvDelta { bytes: Vec<u8> },
BlobTombstone { bytes: Vec<u8> },
PgWalDelta { bytes: Vec<u8> },
MysqlBinlogDelta { bytes: Vec<u8> },
LibsqlChangeset { bytes: Vec<u8> },
}
struct StageReq {
path: String,
chunked: bool,
reply: tokio::sync::oneshot::Sender<StageReply>,
}
enum StageReply {
Sidecar(PathBuf),
Chunked {
rx: tokio::sync::mpsc::Receiver<Vec<u8>>,
poison: Arc<std::sync::atomic::AtomicBool>,
},
Err,
}
type ChunkedProducer = (
tokio::sync::mpsc::Receiver<Vec<u8>>,
Arc<std::sync::atomic::AtomicBool>,
tokio::task::JoinHandle<Result<(), Perr>>,
);
struct ChunkedBodyReader {
rx: tokio::sync::mpsc::Receiver<Vec<u8>>,
poison: Arc<std::sync::atomic::AtomicBool>,
buf: Vec<u8>,
pos: usize,
}
impl Read for ChunkedBodyReader {
fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
while self.pos >= self.buf.len() {
match self.rx.blocking_recv() {
Some(chunk) if !chunk.is_empty() => {
self.buf = chunk;
self.pos = 0;
}
Some(_) => continue,
None => {
if self.poison.load(std::sync::atomic::Ordering::SeqCst) {
return Err(std::io::Error::other("export member producer failed"));
}
return Ok(0); }
}
}
let n = std::cmp::min(out.len(), self.buf.len() - self.pos);
out[..n].copy_from_slice(&self.buf[self.pos..self.pos + n]);
self.pos += n;
Ok(n)
}
}
enum ImportMsg {
Head {
data_only: bool,
source_project: String,
},
Member {
spec: MemberSpec,
temp: PathBuf,
reply: tokio::sync::oneshot::Sender<Result<(), ()>>,
},
ChunkBegin {
spec: MemberSpec,
body_rx: tokio::sync::mpsc::Receiver<Vec<u8>>,
reply: tokio::sync::oneshot::Sender<Result<(), ()>>,
},
ChunkFinish {
verified: bool,
reply: tokio::sync::oneshot::Sender<Result<(), ()>>,
},
}
struct ImportBridge {
req_tx: tokio::sync::mpsc::Sender<ImportMsg>,
body_tx: Option<tokio::sync::mpsc::Sender<Vec<u8>>>,
chunked_bytes: u64,
member_cap: u64,
}
impl ImportBridge {
fn gone() -> project_bundle::BundleError {
project_bundle::BundleError::MalformedStream("import servicer gone".into())
}
fn over_cap(cap: u64) -> project_bundle::BundleError {
project_bundle::BundleError::MalformedStream(format!(
"import refused: a member exceeds the per-member byte cap of {cap} bytes"
))
}
}
impl project_bundle::BundleRestorer for ImportBridge {
fn restore_sidecar(
&mut self,
spec: &MemberSpec,
temp: &std::path::Path,
) -> Result<(), project_bundle::BundleError> {
if let Ok(m) = std::fs::metadata(temp)
&& m.len() > self.member_cap
{
return Err(Self::over_cap(self.member_cap));
}
let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
self.req_tx
.blocking_send(ImportMsg::Member {
spec: spec.clone(),
temp: temp.to_path_buf(),
reply: reply_tx,
})
.map_err(|_| Self::gone())?;
match reply_rx.blocking_recv() {
Ok(Ok(())) => Ok(()),
_ => Err(project_bundle::BundleError::MalformedStream(
"import restore aborted".into(),
)),
}
}
fn begin_chunked(&mut self, spec: &MemberSpec) -> Result<(), project_bundle::BundleError> {
let (body_tx, body_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(STREAM_BYTE_CHANNEL);
let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
self.req_tx
.blocking_send(ImportMsg::ChunkBegin {
spec: spec.clone(),
body_rx,
reply: reply_tx,
})
.map_err(|_| Self::gone())?;
self.body_tx = Some(body_tx);
self.chunked_bytes = 0;
match reply_rx.blocking_recv() {
Ok(Ok(())) => Ok(()),
_ => Err(project_bundle::BundleError::MalformedStream(
"import chunk begin aborted".into(),
)),
}
}
fn write_chunk(
&mut self,
_spec: &MemberSpec,
buf: &[u8],
) -> Result<(), project_bundle::BundleError> {
self.chunked_bytes = self.chunked_bytes.saturating_add(buf.len() as u64);
if self.chunked_bytes > self.member_cap {
return Err(Self::over_cap(self.member_cap));
}
match self.body_tx.as_ref() {
Some(tx) => tx.blocking_send(buf.to_vec()).map_err(|_| {
project_bundle::BundleError::MalformedStream("chunk sink closed".into())
}),
None => Err(project_bundle::BundleError::MalformedStream(
"write_chunk without begin".into(),
)),
}
}
fn finish_chunked(
&mut self,
_spec: &MemberSpec,
verified: bool,
) -> Result<(), project_bundle::BundleError> {
self.body_tx = None;
let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
self.req_tx
.blocking_send(ImportMsg::ChunkFinish {
verified,
reply: reply_tx,
})
.map_err(|_| Self::gone())?;
match reply_rx.blocking_recv() {
Ok(Ok(())) => Ok(()),
_ => Err(project_bundle::BundleError::MalformedStream(
"import chunk finish aborted".into(),
)),
}
}
}
enum ChunkState {
Libsql {
shadow: boatramp_storage::sql_libsql::DumpShadow,
feed_err: Option<Perr>,
},
Staged {
spec: MemberSpec,
temp: PathBuf,
write_err: Option<Perr>,
},
Pg {
plan: Box<PgShadowPlan>,
restore: tokio::task::JoinHandle<Result<i32, Perr>>,
out_path: PathBuf,
},
Mysql {
plan: Box<MysqlShadowPlan>,
restore: tokio::task::JoinHandle<Result<i32, Perr>>,
out_path: PathBuf,
},
}
#[derive(Clone)]
struct PgShadowPlan {
live_db: String,
shadow_db: String,
suffix: String,
maint: PgConnection,
eph_role: String,
eph_password: String,
owner_user: String,
image: String,
trusted_extensions: Vec<String>,
}
impl PgShadowPlan {
fn maint_on_shadow(&self) -> PgConnection {
let mut c = self.maint.clone();
c.database = self.shadow_db.clone();
c
}
fn apply_conn(&self) -> PgConnection {
let mut c = self.maint.clone();
c.user = self.eph_role.clone();
c.password = self.eph_password.clone();
c.database = self.shadow_db.clone();
c
}
}
#[derive(Clone)]
struct MysqlShadowPlan {
live_db: String,
shadow_db: String,
suffix: String,
maint: MysqlConnection,
eph_user: String,
eph_password: String,
image: String,
}
impl MysqlShadowPlan {
fn apply_conn(&self) -> MysqlConnection {
let mut c = self.maint.clone();
c.user = self.eph_user.clone();
c.password = self.eph_password.clone();
c.database = self.shadow_db.clone();
c
}
}
fn random_token() -> String {
let mut bytes = [0u8; 16];
getrandom::getrandom(&mut bytes).expect("os rng for ephemeral role password");
bytes.iter().map(|b| format!("{b:02x}")).collect()
}
#[async_trait]
impl ProjectExport for NodeProjectExport {
async fn export(
&self,
project: &str,
opts: &ExportOptions,
sink: &mut (dyn futures::io::AsyncWrite + Send + Unpin),
) -> Result<(), Perr> {
check_project_slug(project)?;
if opts.data_only && opts.include_identity {
return Err(Perr::Other(
"--data-only and --include-identity are mutually exclusive: a data-only \
bundle carries application data only (no control-plane KV, no identity)"
.into(),
));
}
let data_only = opts.data_only;
let created_at = now_secs();
let session = self.work_dir.join(format!("export-{project}-{created_at}"));
let stage_dir = session.join("stage");
tokio::fs::create_dir_all(&stage_dir).await.map_err(other)?;
let (specs, jobs, source) = self.plan_export(project, opts, created_at).await?;
let chunk_size = self.chunk_size;
let (byte_tx, mut byte_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(STREAM_BYTE_CHANNEL);
let (stage_tx, mut stage_rx) = tokio::sync::mpsc::channel::<StageReq>(1);
let writer_out = ChannelWriter { tx: byte_tx };
let stage_tx_task = stage_tx.clone();
let specs_for_task = specs.clone();
let source_for_task = source.clone();
let write_handle =
tokio::task::spawn_blocking(move || -> Result<(), project_bundle::BundleError> {
let request = |path: &str,
chunked: bool|
-> Result<StageReply, project_bundle::BundleError> {
let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
stage_tx_task
.blocking_send(StageReq {
path: path.to_string(),
chunked,
reply: reply_tx,
})
.map_err(|_| {
project_bundle::BundleError::MalformedStream(
"export stager gone".into(),
)
})?;
reply_rx.blocking_recv().map_err(|_| {
project_bundle::BundleError::MalformedStream("export stager dropped".into())
})
};
project_bundle::write_bundle_streaming_v3(
writer_out,
created_at,
data_only,
source_for_task,
specs_for_task,
chunk_size,
|spec| match request(&spec.path, false)? {
StageReply::Sidecar(tmp) => Ok(tmp),
_ => Err(project_bundle::BundleError::MalformedStream(
"export staging failed".into(),
)),
},
|spec| match request(&spec.path, true)? {
StageReply::Chunked { rx, poison } => Ok(Box::new(ChunkedBodyReader {
rx,
poison,
buf: Vec::new(),
pos: 0,
})
as Box<dyn Read>),
_ => Err(project_bundle::BundleError::MalformedStream(
"export chunked member open failed".into(),
)),
},
)
});
drop(stage_tx);
let stager = async {
let mut idx = 0usize;
let mut producers: Vec<tokio::task::JoinHandle<Result<(), Perr>>> = Vec::new();
while let Some(req) = stage_rx.recv().await {
let job = jobs.get(&req.path);
if req.chunked {
match self.open_chunked_producer(job) {
Ok((rx, poison, handle)) => {
producers.push(handle);
let _ = req.reply.send(StageReply::Chunked { rx, poison });
}
Err(e) => {
let _ = req.reply.send(StageReply::Err);
return Err::<(), Perr>(e);
}
}
} else {
let res = match job {
Some(job) => self.stage_job(project, job, &stage_dir, idx).await,
None => Err(Perr::Other(format!(
"export: internal error — no stage job for member {:?}",
req.path
))),
};
idx += 1;
match res {
Ok(tmp) => {
let _ = req.reply.send(StageReply::Sidecar(tmp));
}
Err(e) => {
let _ = req.reply.send(StageReply::Err);
return Err::<(), Perr>(e);
}
}
}
}
for h in producers {
match h.await {
Ok(Ok(())) => {}
Ok(Err(e)) => return Err(e),
Err(je) => return Err(other(je)),
}
}
Ok(())
};
let drain = async {
while let Some(chunk) = byte_rx.recv().await {
sink.write_all(&chunk)
.await
.map_err(|e| Perr::Other(format!("export sink write failed: {e}")))?;
}
sink.flush()
.await
.map_err(|e| Perr::Other(format!("export sink flush failed: {e}")))?;
Ok::<(), Perr>(())
};
let (stage_res, drain_res, write_res) = tokio::join!(stager, drain, write_handle);
let _ = tokio::fs::remove_dir_all(&session).await;
stage_res?;
match write_res {
Ok(Ok(())) => {}
Ok(Err(be)) => return Err(other(be)),
Err(je) => return Err(other(je)),
}
drain_res?;
Ok(())
}
async fn import(
&self,
project: &str,
src: &mut (dyn futures::io::AsyncRead + Send + Unpin),
opts: ImportOptions,
) -> Result<ImportReport, Perr> {
check_project_slug(project)?;
if !opts.overwrite {
for prefix in
boatramp_core::project_export::project_owned_prefixes(ProjectRef::new(project))
{
if !self
.kv
.list_prefix(&prefix)
.await
.map_err(other)?
.is_empty()
{
return Err(Perr::Other(format!(
"import refused: target project {project:?} already has control-plane \
data under {prefix:?}; pass overwrite=true to replace an existing \
project (a clone into a fresh target needs no overwrite)"
)));
}
}
if !self
.blobs
.list(&blob_project_prefix(project))
.await
.map_err(other)?
.is_empty()
{
return Err(Perr::Other(format!(
"import refused: target project {project:?} already has blobs; pass \
overwrite=true to replace an existing project"
)));
}
}
let session = self
.work_dir
.join(format!("import-{project}-{}", now_secs()));
let stage_dir = session.join("stage");
tokio::fs::create_dir_all(&stage_dir).await.map_err(other)?;
let member_cap = self.import_member_cap;
let (byte_tx, byte_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(STREAM_BYTE_CHANNEL);
let (req_tx, mut req_rx) = tokio::sync::mpsc::channel::<ImportMsg>(1);
let reader = ChannelReader::new(byte_rx);
let stage_dir_task = stage_dir.clone();
let req_tx_task = req_tx.clone();
let read_handle = tokio::task::spawn_blocking(
move || -> Result<Vec<MemberSpec>, project_bundle::BundleError> {
let head_tx = req_tx_task.clone();
let on_head = move |data_only: bool, source: &SourceInfo| {
let _ = head_tx.blocking_send(ImportMsg::Head {
data_only,
source_project: source.project.clone(),
});
};
let mut bridge = ImportBridge {
req_tx: req_tx_task,
body_tx: None,
chunked_bytes: 0,
member_cap,
};
project_bundle::read_bundle_streaming_atomic(
reader,
&stage_dir_task,
on_head,
&mut bridge,
)
},
);
drop(req_tx);
let total_cap = self.import_total_cap;
let feeder = async {
let mut buf = vec![0u8; 64 * 1024];
let mut total: u64 = 0;
loop {
let n = src
.read(&mut buf)
.await
.map_err(|e| Perr::Other(format!("import body read failed: {e}")))?;
if n == 0 {
break;
}
total = total.saturating_add(n as u64);
if total > total_cap {
return Err(Perr::Other(format!(
"import refused: bundle exceeds the total byte cap of {total_cap} bytes"
)));
}
if byte_tx.send(buf[..n].to_vec()).await.is_err() {
break; }
}
drop(byte_tx);
Ok::<(), Perr>(())
};
let servicer = async {
let mut report = ImportReport {
dry_run: !opts.apply,
..Default::default()
};
let mut data_only = false;
let mut source_project = String::new();
let mut saw_identity = false;
let mut pending_chunk: Option<ChunkState> = None;
let mut chunk_idx = 0usize;
while let Some(msg) = req_rx.recv().await {
match msg {
ImportMsg::Head {
data_only: d,
source_project: s,
} => {
data_only = d;
source_project = s;
}
ImportMsg::Member { spec, temp, reply } => {
let r = self
.restore_member(
project,
data_only,
&source_project,
&spec,
&temp,
&opts,
&mut report,
&mut saw_identity,
)
.await;
match r {
Ok(()) => {
let _ = reply.send(Ok(()));
}
Err(e) => {
let _ = reply.send(Err(()));
return Err::<_, Perr>(e);
}
}
}
ImportMsg::ChunkBegin {
spec,
mut body_rx,
reply,
} => {
match spec.kind {
MemberKind::Libsql => {
let site = match site_from_libsql_member(&spec.path) {
Some(s) => s,
None => {
let _ = reply.send(Err(()));
return Err(Perr::Other(format!(
"bad libsql member path {:?}",
spec.path
)));
}
};
let Some(libsql) = self.libsql.as_ref() else {
let _ = reply.send(Err(()));
return Err(Perr::Unavailable(
"bundle carries a libsql member but this node has no local \
libsql backend to restore it into"
.into(),
));
};
let qsite = ProjectRef::new(project).qualified(site);
if opts.apply {
match libsql.begin_dump_shadow(&qsite, "", opts.overwrite).await
{
Ok(mut shadow) => {
let _ = reply.send(Ok(()));
let mut feed_err = None;
while let Some(chunk) = body_rx.recv().await {
if feed_err.is_none()
&& let Err(e) = shadow.feed(&chunk).await
{
feed_err = Some(other(e));
}
}
pending_chunk =
Some(ChunkState::Libsql { shadow, feed_err });
}
Err(e) => {
let _ = reply.send(Err(()));
return Err(other(e));
}
}
} else {
let _ = reply.send(Ok(()));
while body_rx.recv().await.is_some() {}
report.libsql_databases += 1;
pending_chunk = None;
}
}
MemberKind::Blob | MemberKind::Kv => {
let temp = stage_dir.join(format!("chunk-{chunk_idx}"));
chunk_idx += 1;
match tokio::fs::File::create(&temp).await {
Ok(mut f) => {
let _ = reply.send(Ok(()));
use tokio::io::AsyncWriteExt as _;
let mut write_err = None;
while let Some(chunk) = body_rx.recv().await {
if write_err.is_none()
&& let Err(e) = f.write_all(&chunk).await
{
write_err = Some(other(e));
}
}
if write_err.is_none()
&& let Err(e) = f.flush().await
{
write_err = Some(other(e));
}
drop(f);
pending_chunk = Some(ChunkState::Staged {
spec,
temp,
write_err,
});
}
Err(e) => {
let _ = reply.send(Err(()));
return Err(other(e));
}
}
}
MemberKind::PgDump => {
let db = match db_from_pg_member(&spec.path) {
Some(d) => d.to_string(),
None => {
let _ = reply.send(Err(()));
return Err(Perr::Other(format!(
"bad pg member path {:?}",
spec.path
)));
}
};
let cfg = match self.databases.get(&db) {
Some(c) => c.clone(),
None => {
let _ = reply.send(Err(()));
return Err(Perr::Other(format!(
"import: no database named {db:?} on this node"
)));
}
};
if !opts.apply {
let _ = reply.send(Ok(()));
while body_rx.recv().await.is_some() {}
report.pg_databases += 1;
pending_chunk = None;
} else {
let plan = match self.pg_shadow_plan(project, &db, &cfg).await {
Ok(p) => p,
Err(e) => {
let _ = reply.send(Err(()));
return Err(e);
}
};
if let Err(e) = self.pg_create_shadow(&plan, &stage_dir).await {
self.pg_drop_shadow(&plan, &stage_dir).await;
let _ = reply.send(Err(()));
return Err(e);
}
if let Err(e) = self
.pg_assert_not_superuser(
&plan.apply_conn(),
&plan.image,
&stage_dir,
)
.await
{
self.pg_drop_shadow(&plan, &stage_dir).await;
let _ = reply.send(Err(()));
return Err(e);
}
let out_path =
stage_dir.join(format!("pgrestore-{chunk_idx}.out"));
chunk_idx += 1;
let restore = match self.spawn_pg_restore(
&plan,
body_rx,
out_path.clone(),
) {
Ok(h) => h,
Err(e) => {
self.pg_drop_shadow(&plan, &stage_dir).await;
let _ = reply.send(Err(()));
return Err(e);
}
};
let _ = reply.send(Ok(()));
pending_chunk = Some(ChunkState::Pg {
plan: Box::new(plan),
restore,
out_path,
});
}
}
MemberKind::MysqlDump => {
let db = match db_from_mysql_member(&spec.path) {
Some(d) => d.to_string(),
None => {
let _ = reply.send(Err(()));
return Err(Perr::Other(format!(
"bad mysql member path {:?}",
spec.path
)));
}
};
let cfg = match self.databases.get(&db) {
Some(c) => c.clone(),
None => {
let _ = reply.send(Err(()));
return Err(Perr::Other(format!(
"import: no database named {db:?} on this node"
)));
}
};
if !opts.apply {
let _ = reply.send(Ok(()));
while body_rx.recv().await.is_some() {}
report.pg_databases += 1;
pending_chunk = None;
} else {
let plan =
match self.mysql_shadow_plan(project, &db, &cfg).await {
Ok(p) => p,
Err(e) => {
let _ = reply.send(Err(()));
return Err(e);
}
};
if let Err(e) =
self.mysql_create_shadow(&plan, &stage_dir).await
{
self.mysql_drop_shadow(&plan, &stage_dir).await;
let _ = reply.send(Err(()));
return Err(e);
}
if let Err(e) =
self.mysql_assert_not_superuser(&plan, &stage_dir).await
{
self.mysql_drop_shadow(&plan, &stage_dir).await;
let _ = reply.send(Err(()));
return Err(e);
}
let out_path =
stage_dir.join(format!("myrestore-{chunk_idx}.out"));
chunk_idx += 1;
let restore = match self.spawn_mysql_restore(
&plan,
body_rx,
out_path.clone(),
) {
Ok(h) => h,
Err(e) => {
self.mysql_drop_shadow(&plan, &stage_dir).await;
let _ = reply.send(Err(()));
return Err(e);
}
};
let _ = reply.send(Ok(()));
pending_chunk = Some(ChunkState::Mysql {
plan: Box::new(plan),
restore,
out_path,
});
}
}
_ => {
let _ = reply.send(Err(()));
return Err(Perr::Unavailable(format!(
"chunked framing is not wired for member kind {:?}",
spec.kind
)));
}
}
}
ImportMsg::ChunkFinish { verified, reply } => {
let state = pending_chunk.take();
let res: Result<(), Perr> = match state {
Some(ChunkState::Libsql { shadow, feed_err }) => {
if let Some(e) = feed_err {
drop(shadow); Err(e)
} else if !verified {
drop(shadow); Ok(())
} else {
match self
.libsql
.as_ref()
.expect("libsql shadow implies a libsql backend")
.commit_dump_shadow(shadow)
.await
{
Ok(()) => {
report.libsql_databases += 1;
Ok(())
}
Err(e) => Err(other(e)),
}
}
}
Some(ChunkState::Staged {
spec,
temp,
write_err,
}) => {
let r = if let Some(e) = write_err {
Err(e)
} else if !verified {
Ok(()) } else {
self.restore_member(
project,
data_only,
&source_project,
&spec,
&temp,
&opts,
&mut report,
&mut saw_identity,
)
.await
};
let _ = tokio::fs::remove_file(&temp).await;
r
}
Some(ChunkState::Pg {
plan,
restore,
out_path,
}) => {
let code = restore.await;
let out = tokio::fs::read_to_string(&out_path)
.await
.unwrap_or_default();
let _ = tokio::fs::remove_file(&out_path).await;
match code {
Ok(Ok(0)) if verified => {
match self.pg_swap_shadow(&plan, &stage_dir).await {
Ok(()) => {
report.pg_databases += 1;
Ok(())
}
Err(e) => {
self.pg_drop_shadow(&plan, &stage_dir).await;
Err(e)
}
}
}
Ok(Ok(0)) => {
self.pg_drop_shadow(&plan, &stage_dir).await;
Ok(())
}
Ok(Ok(c)) => {
self.pg_drop_shadow(&plan, &stage_dir).await;
Err(Perr::Other(format!(
"pg_restore into shadow {:?} exited {c}; live untouched: {out}",
plan.shadow_db
)))
}
Ok(Err(e)) => {
self.pg_drop_shadow(&plan, &stage_dir).await;
Err(e)
}
Err(je) => {
self.pg_drop_shadow(&plan, &stage_dir).await;
Err(other(je))
}
}
}
Some(ChunkState::Mysql {
plan,
restore,
out_path,
}) => {
let code = restore.await;
let out = tokio::fs::read_to_string(&out_path)
.await
.unwrap_or_default();
let _ = tokio::fs::remove_file(&out_path).await;
match code {
Ok(Ok(0)) if verified => {
match self.mysql_swap_shadow(&plan, &stage_dir).await {
Ok(()) => {
report.pg_databases += 1;
Ok(())
}
Err(e) => {
self.mysql_drop_shadow(&plan, &stage_dir).await;
Err(e)
}
}
}
Ok(Ok(0)) => {
self.mysql_drop_shadow(&plan, &stage_dir).await;
Ok(())
}
Ok(Ok(c)) => {
self.mysql_drop_shadow(&plan, &stage_dir).await;
Err(Perr::Other(format!(
"mysql restore into shadow {:?} exited {c}; live untouched: {out}",
plan.shadow_db
)))
}
Ok(Err(e)) => {
self.mysql_drop_shadow(&plan, &stage_dir).await;
Err(e)
}
Err(je) => {
self.mysql_drop_shadow(&plan, &stage_dir).await;
Err(other(je))
}
}
}
None => {
Ok(())
}
};
match res {
Ok(()) => {
let _ = reply.send(Ok(()));
}
Err(e) => {
let _ = reply.send(Err(()));
return Err(e);
}
}
}
}
}
match pending_chunk.take() {
Some(ChunkState::Pg { plan, restore, .. }) => {
restore.abort();
self.pg_drop_shadow(&plan, &stage_dir).await;
}
Some(ChunkState::Mysql { plan, restore, .. }) => {
restore.abort();
self.mysql_drop_shadow(&plan, &stage_dir).await;
}
_ => {}
}
Ok((report, data_only, saw_identity))
};
let (feed_res, serv_res, read_res) = tokio::join!(feeder, servicer, read_handle);
let (mut report, data_only, saw_identity) = match serv_res {
Ok(v) => v,
Err(e) => {
let _ = tokio::fs::remove_dir_all(&session).await;
return Err(e);
}
};
match read_res {
Ok(Ok(_specs)) => {}
Ok(Err(be)) => {
let _ = tokio::fs::remove_dir_all(&session).await;
return Err(other(be));
}
Err(je) => {
let _ = tokio::fs::remove_dir_all(&session).await;
return Err(other(je));
}
}
if let Err(e) = feed_res {
let _ = tokio::fs::remove_dir_all(&session).await;
return Err(e);
}
if data_only {
report
.notes
.push("control-plane KV (identity/RBAC/routing) NOT imported: --data-only".into());
}
if matches!(opts.identity, IdentityChoice::Adopt { .. }) && !saw_identity {
let _ = tokio::fs::remove_dir_all(&session).await;
return Err(Perr::Other(
"--identity adopt requires a bundle exported with --include-identity \
(no identity member present); use --identity fresh to import a bundle \
without identity"
.into(),
));
}
if matches!(opts.identity, IdentityChoice::Fresh) && !data_only && opts.apply {
let prefixes =
boatramp_core::project_export::sealed_secret_prefixes(ProjectRef::new(project));
let mut dropped = 0usize;
for prefix in &prefixes {
for key in self.kv.list_prefix(prefix).await.map_err(other)? {
self.kv.delete_checkpointed(&key).await.map_err(other)?;
dropped += 1;
}
}
report.notes.push(format!(
"identity: --fresh dropped {dropped} source-sealed secret key(s); re-provision \
secrets + the managed-DB credential on the target"
));
}
if !data_only {
report.notes.push(
"post-import: re-deploy function/component code (deploy artifacts are not \
bundled), re-verify + re-attach the project's domains, and let ACME re-issue TLS \
certs (routing + certs are intentionally not carried in a portable bundle)"
.into(),
);
}
let _ = tokio::fs::remove_dir_all(&session).await;
Ok(report)
}
async fn export_resumable(
&self,
project: &str,
opts: &ExportOptions,
dir_name: &str,
) -> Result<(), Perr> {
let dir = self.resolve_resumable_dir(dir_name, true).await?;
self.export_resumable_dir(project, opts, &dir).await
}
async fn import_resumable(
&self,
project: &str,
dir_name: &str,
opts: ImportOptions,
) -> Result<ImportReport, Perr> {
let dir = self.resolve_resumable_dir(dir_name, false).await?;
self.import_resumable_dir(project, &dir, opts).await
}
}
impl NodeProjectExport {
async fn plan_export(
&self,
project: &str,
opts: &ExportOptions,
created_at: u64,
) -> Result<(Vec<MemberSpec>, HashMap<String, StageJob>, SourceInfo), Perr> {
let data_only = opts.data_only;
let mut ordered: Vec<(MemberSpec, StageJob)> = Vec::new();
if opts.include_identity {
let transport_key = opts
.transport_key
.as_deref()
.ok_or_else(|| Perr::Other("--include-identity requires a transport key".into()))?;
match self.identity_source.as_ref() {
Some(IdentityEnvelope::Local { kek_file }) => {
let kek = tokio::fs::read(kek_file).await.map_err(|e| {
Perr::Other(format!("reading KEK {}: {e}", kek_file.display()))
})?;
let payload = identity_payload_bytes(&kek)?;
let container =
boatramp_server::envelope::transport_wrap(&payload, transport_key)
.map_err(|e| {
Perr::Other(format!("wrapping the identity container: {e}"))
})?;
ordered.push((
MemberSpec::sidecar(
IDENTITY_MEMBER_PATH.to_string(),
MemberKind::Identity,
None,
),
StageJob::Identity { container },
));
}
Some(IdentityEnvelope::Vault) => {
return Err(Perr::Other(
"--include-identity is not supported for a Vault KEK: the key never \
leaves the KMS and cannot be bundled. For a faithful recovery, point \
the target node's [secrets.vault] at the same key; otherwise use \
--identity fresh."
.into(),
));
}
Some(IdentityEnvelope::None) | None => {
return Err(Perr::Other(
"--include-identity requires a local KEK envelope \
([secrets].envelope = \"local\"); none is configured on this node"
.into(),
));
}
}
}
if !data_only {
self.deploy.checkpoint().await.map_err(other)?;
}
let blob_metas = self
.blobs
.list(&blob_project_prefix(project))
.await
.map_err(other)?;
for meta in blob_metas {
let member_path = format!("blobs/{}", meta.key);
let spec = if meta.size.is_some_and(|s| s > self.chunk_size as u64) {
MemberSpec::chunked(member_path, MemberKind::Blob, None)
} else {
MemberSpec::sidecar(member_path, MemberKind::Blob, None)
};
ordered.push((spec, StageJob::Blob { key: meta.key }));
}
for (db, cfg) in &self.databases {
let is_managed = cfg.compute.as_deref().is_some_and(|c| !c.is_empty());
let dumpable = is_managed || !cfg.url_env.is_empty();
let is_shared = matches!(cfg.tenant, crate::config::TenantIsolation::Shared);
let chunked = is_managed && !data_only && !is_shared;
if kind_is_postgres(&cfg.kind) && dumpable {
let (conn, image, server_version_num) = self.resolve_pg(project, db, cfg).await?;
let schema_fingerprint = self.schema_fingerprint(project, db).await;
let pg = Some(PgMemberInfo {
server_version_num: server_version_num.clone(),
client_image: image.clone(),
schema_fingerprint,
});
let spec = if chunked {
MemberSpec::chunked(pg_member_path(db), MemberKind::PgDump, pg)
} else {
MemberSpec::sidecar(pg_member_path(db), MemberKind::PgDump, pg)
};
ordered.push((
spec,
StageJob::Pg {
conn,
image,
data_only,
server_version_num,
no_publications: cfg.delta_capture,
},
));
} else if kind_is_mysql(&cfg.kind) && dumpable {
let (conn, image, version) = self.resolve_mysql(project, db, cfg).await?;
let schema_fingerprint = self.schema_fingerprint(project, db).await;
let pg = Some(PgMemberInfo {
server_version_num: version,
client_image: image.clone(),
schema_fingerprint,
});
let spec = if chunked {
MemberSpec::chunked(mysql_member_path(db), MemberKind::MysqlDump, pg)
} else {
MemberSpec::sidecar(mysql_member_path(db), MemberKind::MysqlDump, pg)
};
ordered.push((
spec,
StageJob::Mysql {
conn,
image,
data_only,
},
));
}
}
if let Some(libsql) = &self.libsql {
let sites = self
.deploy
.list_sites(ProjectRef::new(project))
.await
.map_err(other)?;
for site in sites {
let qsite = ProjectRef::new(project).qualified(&site);
if !libsql.local_database_exists(&qsite, "") {
continue; }
ordered.push((
MemberSpec::chunked(
libsql_member_path(project, &site),
MemberKind::Libsql,
None,
),
StageJob::Libsql { qsite },
));
}
}
if !data_only {
ordered.push((
MemberSpec::sidecar(KV_MEMBER_PATH.to_string(), MemberKind::Kv, None),
StageJob::Kv { created_at },
));
}
let specs: Vec<MemberSpec> = ordered.iter().map(|(s, _)| s.clone()).collect();
let jobs: HashMap<String, StageJob> =
ordered.into_iter().map(|(s, j)| (s.path, j)).collect();
Ok((
specs,
jobs,
SourceInfo {
project: project.to_string(),
site: None,
},
))
}
fn open_chunked_producer(&self, job: Option<&StageJob>) -> Result<ChunkedProducer, Perr> {
use std::sync::atomic::Ordering::SeqCst;
let (body_tx, body_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(STREAM_BYTE_CHANNEL);
let poison = Arc::new(std::sync::atomic::AtomicBool::new(false));
let poison2 = poison.clone();
let handle = match job {
Some(StageJob::Libsql { qsite }) => {
let libsql = self.libsql.clone().ok_or_else(|| {
Perr::Other("internal: libsql chunked job with no libsql backend".into())
})?;
let qsite = qsite.clone();
tokio::spawn(async move {
let r = libsql
.dump_database(&qsite, "", &body_tx)
.await
.map_err(other);
if r.is_err() {
poison2.store(true, SeqCst);
}
drop(body_tx); r
})
}
Some(StageJob::Blob { key }) => {
let blobs = self.blobs.clone();
let key = key.clone();
tokio::spawn(async move {
let r: Result<(), Perr> = async {
let obj = blobs.get(&key).await.map_err(other)?;
let mut body = obj.body;
while let Some(chunk) = body.next().await {
let chunk = chunk.map_err(other)?;
body_tx
.send(chunk.as_ref().to_vec())
.await
.map_err(|_| Perr::Other("export blob body sink closed".into()))?;
}
Ok(())
}
.await;
if r.is_err() {
poison2.store(true, SeqCst);
}
drop(body_tx);
r
})
}
Some(StageJob::Pg {
conn,
image,
server_version_num,
no_publications,
..
}) => {
let backend = self.require_backend()?.clone();
let conn = conn.clone();
let image = image.clone();
let server_version_num = server_version_num.clone();
let no_publications = *no_publications;
let exclude_extensions = self.trusted_extensions.clone();
tokio::spawn(async move {
let mut sink = AsyncChannelWriter::new(body_tx);
let r: Result<(), Perr> = async {
let inv = pg_dump_invocation(
&conn,
false,
&server_version_num,
&exclude_extensions,
no_publications,
);
let code = backend
.run_ephemeral(&image, &inv.argv, &inv.env, None, &mut sink)
.await
.map_err(other)?;
if code != 0 {
return Err(Perr::Other(format!("pg_dump exited {code}")));
}
Ok(())
}
.await;
if r.is_err() {
poison2.store(true, SeqCst);
}
drop(sink); r
})
}
Some(StageJob::Mysql { conn, image, .. }) => {
let backend = self.require_backend()?.clone();
let conn = conn.clone();
let image = image.clone();
tokio::spawn(async move {
let mut sink = AsyncChannelWriter::new(body_tx);
let r: Result<(), Perr> = async {
let inv = mysqldump_tables_only_invocation(&conn);
let code = backend
.run_ephemeral(&image, &inv.argv, &inv.env, None, &mut sink)
.await
.map_err(other)?;
if code != 0 {
return Err(Perr::Other(format!("mysqldump exited {code}")));
}
Ok(())
}
.await;
if r.is_err() {
poison2.store(true, SeqCst);
}
drop(sink);
r
})
}
_ => {
return Err(Perr::Other(
"export: chunked framing is wired only for pg/mysql, libsql + large blobs"
.into(),
));
}
};
Ok((body_rx, poison, handle))
}
async fn stage_job(
&self,
project: &str,
job: &StageJob,
stage_dir: &Path,
idx: usize,
) -> Result<PathBuf, Perr> {
let tmp = stage_dir.join(format!("m{idx}"));
match job {
StageJob::Identity { container } => {
tokio::fs::write(&tmp, container).await.map_err(other)?;
}
StageJob::Kv { created_at } => {
let all = boatramp_core::kv_dump::scan_dumpable(self.kv.as_ref())
.await
.map_err(other)?;
let scoped = boatramp_core::project_export::scope_entries_to_project(
all,
ProjectRef::new(project),
);
let kv_bytes = boatramp_core::kv_dump::encode_dump(&scoped, *created_at);
tokio::fs::write(&tmp, &kv_bytes).await.map_err(other)?;
}
StageJob::Pg {
conn,
image,
data_only,
server_version_num,
no_publications,
} => {
let inv =
pg_dump_invocation(conn, *data_only, server_version_num, &[], *no_publications);
let code = self
.run_sql_client(image, &inv.argv, &inv.env, None, &tmp)
.await?;
if code != 0 {
return Err(Perr::Other(format!("pg_dump exited {code}")));
}
}
StageJob::Mysql {
conn,
image,
data_only,
} => {
let inv = mysqldump_invocation(conn, *data_only);
let code = self
.run_sql_client(image, &inv.argv, &inv.env, None, &tmp)
.await?;
if code != 0 {
return Err(Perr::Other(format!("mysqldump exited {code}")));
}
}
StageJob::Libsql { qsite } => {
let libsql = self.libsql.as_ref().ok_or_else(|| {
Perr::Other("internal error: libsql stage job with no libsql backend".into())
})?;
libsql.snapshot_into(qsite, "", &tmp).await.map_err(other)?;
}
StageJob::Blob { key } => {
let obj = self.blobs.get(key).await.map_err(other)?;
let mut body = obj.body;
let mut f = tokio::fs::File::create(&tmp).await.map_err(other)?;
use tokio::io::AsyncWriteExt as _;
while let Some(chunk) = body.next().await {
let chunk = chunk.map_err(other)?;
f.write_all(chunk.as_ref()).await.map_err(other)?;
}
f.flush().await.map_err(other)?;
}
StageJob::KvDelta { bytes }
| StageJob::BlobTombstone { bytes }
| StageJob::PgWalDelta { bytes }
| StageJob::MysqlBinlogDelta { bytes }
| StageJob::LibsqlChangeset { bytes } => {
tokio::fs::write(&tmp, bytes).await.map_err(other)?;
}
}
Ok(tmp)
}
#[allow(clippy::too_many_arguments)]
async fn restore_member(
&self,
project: &str,
data_only: bool,
source_project: &str,
spec: &MemberSpec,
temp: &Path,
opts: &ImportOptions,
report: &mut ImportReport,
saw_identity: &mut bool,
) -> Result<(), Perr> {
match spec.kind {
MemberKind::Identity => {
*saw_identity = true;
match &opts.identity {
IdentityChoice::Adopt { transport_key } => {
match self.identity_source.as_ref() {
Some(IdentityEnvelope::Local { kek_file }) => {
let container = tokio::fs::read(temp).await.map_err(other)?;
let payload = boatramp_server::envelope::transport_unwrap(
&container,
transport_key,
)
.map_err(|e| {
Perr::Other(format!(
"opening the identity container (wrong transport key?): {e}"
))
})?;
let kek = parse_identity_payload(&payload)?;
match tokio::fs::read(kek_file).await {
Ok(existing) if existing != kek => {
return Err(Perr::Other(format!(
"refusing to adopt: this node already holds a DIFFERENT \
KEK at {} — adopt is for recovery onto a fresh node; use \
--identity fresh to import into a different environment",
kek_file.display()
)));
}
Ok(_) => { }
Err(_) => {
if opts.apply {
write_kek_file(kek_file, &kek).await?;
}
}
}
report
.notes
.push("identity: adopted the source KEK (recovery)".into());
}
Some(IdentityEnvelope::Vault) | Some(IdentityEnvelope::None) | None => {
return Err(Perr::Other(
"--identity adopt installs a local KEK from the bundle, but this \
node's secrets envelope is not local; for a Vault recovery point \
[secrets.vault] at the same key, or use --identity fresh"
.into(),
));
}
}
}
IdentityChoice::Fresh => {
report.notes.push(
"identity: bundle carries an identity member, but --identity fresh \
was selected; the source KEK was NOT installed"
.into(),
);
}
}
}
MemberKind::Blob => {
let target_key = target_blob_key(&spec.path, source_project, project)?;
report.blobs += 1;
if opts.apply {
let staged_dir = temp.parent().ok_or_else(|| {
Perr::Other("blob staging temp has no parent directory".into())
})?;
let staged_name = temp
.file_name()
.and_then(|n| n.to_str())
.ok_or_else(|| Perr::Other("blob staging temp has no file name".into()))?;
let staged = boatramp_storage::FsStorage::new(staged_dir);
let obj = staged.get(staged_name).await.map_err(other)?;
self.blobs
.put(
&target_key,
obj.body,
boatramp_core::PutMeta { content_type: None },
)
.await
.map_err(other)?;
}
}
MemberKind::PgDump => {
let db = db_from_pg_member(&spec.path)
.ok_or_else(|| Perr::Other(format!("bad pg member path {:?}", spec.path)))?;
let cfg = self.databases.get(db).ok_or_else(|| {
Perr::Other(format!("import: no database named {db:?} on this node"))
})?;
let (conn, image, _ver) = self.resolve_pg(project, db, cfg).await?;
if data_only {
let src_fp = spec.pg.as_ref().and_then(|p| p.schema_fingerprint.clone());
let tgt_fp = self.schema_fingerprint(project, db).await;
if data_only_skew_blocks(src_fp.as_deref(), tgt_fp.as_deref()) {
report.schema_skew.push(db.to_string());
if !opts.allow_schema_skew {
return Err(Perr::SchemaSkew(vec![db.to_string()]));
}
}
}
if opts.apply {
if is_managed_binding(cfg) {
return Err(Perr::Unavailable(format!(
"import refused: a MANAGED pg database ({db:?}) can only be restored via \
the atomic chunked shadow path (a FULL, dedicated/Single bundle). A \
data-only or Shared sidecar restore would apply the untrusted dump as \
the bootstrap superuser (RCE risk); a least-privilege sidecar apply is a \
pending follow-up. Re-export as a FULL dedicated bundle, or import into \
an external (bring-your-own-URL) binding."
)));
}
let inv = pg_restore_invocation(&conn, data_only);
let merged = merge_stderr_argv(&inv.argv);
let outfile = temp.with_extension("restore.out");
let code = self
.run_sql_client(&image, &merged, &inv.env, Some(temp), &outfile)
.await?;
if code != 0 {
let diag = tokio::fs::read_to_string(&outfile)
.await
.unwrap_or_default();
let _ = tokio::fs::remove_file(&outfile).await;
return Err(Perr::Other(format!(
"pg_restore for {db:?} exited {code}: {}",
diag.trim()
)));
}
let _ = tokio::fs::remove_file(&outfile).await;
}
report.pg_databases += 1;
}
MemberKind::MysqlDump => {
let db = db_from_mysql_member(&spec.path)
.ok_or_else(|| Perr::Other(format!("bad mysql member path {:?}", spec.path)))?;
let cfg = self.databases.get(db).ok_or_else(|| {
Perr::Other(format!("import: no database named {db:?} on this node"))
})?;
let (conn, image, _ver) = self.resolve_mysql(project, db, cfg).await?;
if data_only {
let src_fp = spec.pg.as_ref().and_then(|p| p.schema_fingerprint.clone());
let tgt_fp = self.schema_fingerprint(project, db).await;
if data_only_skew_blocks(src_fp.as_deref(), tgt_fp.as_deref()) {
report.schema_skew.push(db.to_string());
if !opts.allow_schema_skew {
return Err(Perr::SchemaSkew(vec![db.to_string()]));
}
}
}
if opts.apply {
if is_managed_binding(cfg) {
return Err(Perr::Unavailable(format!(
"import refused: a MANAGED mysql database ({db:?}) can only be restored \
via the atomic chunked shadow path (a FULL, dedicated/Single bundle); a \
data-only/Shared sidecar restore would apply the untrusted dump as a \
superuser (RCE risk). Re-export as a FULL dedicated bundle, or import \
into an external binding."
)));
}
let inv = mysql_restore_invocation(&conn, data_only);
let merged = merge_stderr_argv(&inv.argv);
let outfile = temp.with_extension("myrestore.out");
let code = self
.run_sql_client(&image, &merged, &inv.env, Some(temp), &outfile)
.await?;
if code != 0 {
let diag = tokio::fs::read_to_string(&outfile)
.await
.unwrap_or_default();
let _ = tokio::fs::remove_file(&outfile).await;
return Err(Perr::Other(format!(
"mysql restore for {db:?} exited {code}: {}",
diag.trim()
)));
}
let _ = tokio::fs::remove_file(&outfile).await;
}
report.pg_databases += 1;
}
MemberKind::Libsql => {
let Some(libsql) = &self.libsql else {
return Err(Perr::Unavailable(
"bundle carries libsql members but this node has no local libsql backend \
to restore them into"
.into(),
));
};
let site = site_from_libsql_member(&spec.path).ok_or_else(|| {
Perr::Other(format!("bad libsql member path {:?}", spec.path))
})?;
let qsite = ProjectRef::new(project).qualified(site);
if opts.apply {
libsql
.restore_snapshot(&qsite, "", temp, opts.overwrite)
.await
.map_err(other)?;
}
report.libsql_databases += 1;
}
MemberKind::Kv => {
if data_only {
return Ok(());
}
if matches!(opts.identity, IdentityChoice::Adopt { .. }) && !*saw_identity {
return Err(Perr::Other(
"--identity adopt requires a bundle exported with --include-identity \
(no identity member present); use --identity fresh to import a bundle \
without identity"
.into(),
));
}
let bytes = tokio::fs::read(temp).await.map_err(other)?;
let mut decoded = boatramp_core::kv_dump::decode_dump(&bytes).map_err(other)?;
boatramp_core::project_export::rekey_entries_to_target(
&mut decoded.entries,
source_project,
project,
);
boatramp_core::project_export::enforce_target_scoped_kv(
&decoded.entries,
ProjectRef::new(project),
)
.map_err(Perr::Other)?;
if opts.apply {
match boatramp_core::kv_dump::apply_import(
self.kv.as_ref(),
&decoded.entries,
opts.force,
)
.await
{
Ok(rep) => {
report.kv_records = rep.written;
report.kv_verified = rep.verified;
}
Err(boatramp_core::kv_dump::ImportError::Commingle(_)) => {
return Err(Perr::Commingle(
"control-plane KV import refused a non-empty destination; \
re-run with --force to overlay it"
.into(),
));
}
Err(e) => return Err(other(e)),
}
} else {
let plan = boatramp_core::kv_dump::plan_import(
self.kv.as_ref(),
&decoded.entries,
opts.force,
)
.await
.map_err(other)?;
report.kv_records = plan.would_write;
}
}
MemberKind::KvDelta => {
if data_only {
return Ok(());
}
if matches!(opts.identity, IdentityChoice::Adopt { .. }) && !*saw_identity {
return Err(Perr::Other(
"--identity adopt requires a bundle exported with --include-identity \
(no identity member present); use --identity fresh"
.into(),
));
}
let bytes = tokio::fs::read(temp).await.map_err(other)?;
let delta = boatramp_core::project_delta::decode_kv_delta(&bytes).map_err(other)?;
let mut upserts = delta.upserts;
boatramp_core::project_export::rekey_entries_to_target(
&mut upserts,
source_project,
project,
);
boatramp_core::project_export::enforce_target_scoped_kv(
&upserts,
ProjectRef::new(project),
)
.map_err(Perr::Other)?;
let mut delete_keys = Vec::with_capacity(delta.tombstones.len());
for k in &delta.tombstones {
let rk =
boatramp_core::project_export::rekey_owned_key(k, source_project, project);
if !boatramp_core::project_export::project_owns_key(
&rk,
ProjectRef::new(project),
) {
return Err(Perr::Other(format!(
"control-plane KV delta import refused: tombstone {k:?} is outside the \
target project {project}'s owned scope after re-key — a delta must not \
delete another tenant's or a global (routing / TLS / daemon) key"
)));
}
delete_keys.push(rk);
}
if opts.apply {
match boatramp_core::kv_dump::apply_import(self.kv.as_ref(), &upserts, true)
.await
{
Ok(rep) => {
report.kv_records += rep.written;
report.kv_verified += rep.verified;
}
Err(e) => return Err(other(e)),
}
for k in &delete_keys {
self.kv.delete_checkpointed(k).await.map_err(other)?;
}
} else {
report.kv_records += upserts.len();
}
report.notes.push(format!(
"kv delta: {} upsert(s), {} tombstone(s)",
upserts.len(),
delete_keys.len()
));
}
MemberKind::BlobTombstone => {
let bytes = tokio::fs::read(temp).await.map_err(other)?;
let keys =
boatramp_core::project_delta::decode_blob_tombstones(&bytes).map_err(other)?;
let mut deleted = 0u64;
for storage_key in &keys {
let target_key =
target_blob_key(&format!("blobs/{storage_key}"), source_project, project)?;
if opts.apply {
self.blobs.delete(&target_key).await.map_err(other)?;
}
deleted += 1;
}
report
.notes
.push(format!("blob delta: {deleted} tombstone(s)"));
}
MemberKind::PgWalDelta | MemberKind::MysqlBinlogDelta | MemberKind::LibsqlChangeset => {
return Err(Perr::Other(format!(
"this build cannot restore a SQL true-delta member ({:?}) here — a pg WAL delta \
is replayed only via the managed-pg atomic-shadow chain restore; the {} replay \
engine is not available on this path (SQL true-delta program)",
spec.kind,
match spec.kind {
MemberKind::PgWalDelta => "Postgres WAL (non-managed/non-chain)",
MemberKind::MysqlBinlogDelta => "MySQL binlog",
_ => "libsql changeset",
}
)));
}
}
Ok(())
}
pub async fn export_resumable_dir(
&self,
project: &str,
opts: &ExportOptions,
dir: &Path,
) -> Result<(), Perr> {
check_project_slug(project)?;
if opts.data_only && opts.include_identity {
return Err(Perr::Other(
"--data-only and --include-identity are mutually exclusive".into(),
));
}
tokio::fs::create_dir_all(dir).await.map_err(other)?;
let created_at = now_secs();
let session = self
.work_dir
.join(format!("export-res-{project}-{created_at}"));
let stage_dir = session.join("stage");
tokio::fs::create_dir_all(&stage_dir).await.map_err(other)?;
let (specs, jobs, source, delta_info) = match &opts.delta {
None => {
let (s, j, src) = self.plan_export(project, opts, created_at).await?;
(s, j, src, None)
}
Some(req) => {
let (s, j, src, di) = self
.plan_delta_export(project, opts, req, created_at)
.await?;
(s, j, src, Some(di))
}
};
let manifest_path = dir.join(project_bundle::MANIFEST_PATH);
let specs = if tokio::fs::try_exists(&manifest_path).await.map_err(other)? {
let bytes = tokio::fs::read(&manifest_path).await.map_err(other)?;
let m: project_bundle::StreamManifest = serde_json::from_slice(&bytes)
.map_err(|e| Perr::Other(format!("read resumable MANIFEST: {e}")))?;
if m.source.project != source.project {
return Err(Perr::Other(format!(
"resumable export dir holds a MANIFEST for project {:?}, not {:?} — \
refusing to mix projects in one dir",
m.source.project, source.project
)));
}
m.members
} else {
let m = project_bundle::StreamManifest {
magic: project_bundle::BUNDLE_MAGIC.to_string(),
format_version: project_bundle::BUNDLE_FORMAT_VERSION_V3,
created_at,
data_only: opts.data_only,
source: source.clone(),
members: specs.clone(),
delta: delta_info,
};
let bytes = serde_json::to_vec_pretty(&m)
.map_err(|e| Perr::Other(format!("encode MANIFEST: {e}")))?;
write_atomic(&manifest_path, &bytes).await?;
specs
};
if opts.delta.is_none() {
self.pg_capture_base_slots(project, dir, &stage_dir).await;
}
for (idx, spec) in specs.iter().enumerate() {
let obj = dir.join(member_object_name(idx));
let meta = dir.join(member_meta_name(idx));
if member_object_valid(&obj, &meta).await? {
continue; }
let _ = tokio::fs::remove_file(&obj).await;
let _ = tokio::fs::remove_file(dir.join(member_part_name(idx))).await;
let capture_base_mysql = opts.delta.is_none()
&& spec.kind == MemberKind::MysqlDump
&& db_from_mysql_member(&spec.path)
.and_then(|d| self.databases.get(d))
.is_some_and(Self::mysql_capture_eligible);
let capture_base_pg = opts.delta.is_none()
&& spec.kind == MemberKind::PgDump
&& db_from_pg_member(&spec.path)
.and_then(|d| self.databases.get(d))
.is_some_and(Self::pg_capture_eligible);
let digest = if capture_base_mysql {
let db = db_from_mysql_member(&spec.path).expect("mysql member path");
let cfg = self.databases.get(db).expect("capture-eligible db").clone();
self.produce_mysql_capture_base(project, dir, idx, db, &cfg, &stage_dir)
.await?
} else if capture_base_pg {
let db = db_from_pg_member(&spec.path).expect("pg member path");
let cfg = self.databases.get(db).expect("capture-eligible db").clone();
self.produce_pg_capture_base(project, dir, idx, db, &cfg, &stage_dir)
.await?
} else {
let job = jobs.get(&spec.path);
match spec.framing {
project_bundle::MemberFraming::Sidecar => {
let job = job.ok_or_else(|| {
Perr::Other(format!("resumable export: no stage job for {}", spec.path))
})?;
let tmp = self.stage_job(project, job, &stage_dir, idx).await?;
let d = write_member_from_file(dir, idx, &tmp).await?;
let _ = tokio::fs::remove_file(&tmp).await;
d
}
project_bundle::MemberFraming::ChunkedTrailer => {
let producer = self.open_chunked_producer(job)?;
write_member_from_producer(dir, idx, producer).await?
}
}
};
let meta_bytes = serde_json::to_vec(&digest)
.map_err(|e| Perr::Other(format!("encode member meta: {e}")))?;
write_atomic(&meta, &meta_bytes).await?;
}
if opts.delta.is_none() {
self.mysql_capture_base_gtids(project, dir, &specs).await;
}
#[cfg(feature = "sql-delta-libsql")]
self.libsql_capture_base_snapshots(project, dir).await;
let _ = tokio::fs::remove_dir_all(&session).await;
Ok(())
}
async fn plan_delta_export(
&self,
project: &str,
opts: &ExportOptions,
req: &boatramp_core::project_delta::DeltaRequest,
created_at: u64,
) -> Result<
(
Vec<MemberSpec>,
HashMap<String, StageJob>,
SourceInfo,
boatramp_core::project_delta::DeltaInfo,
),
Perr,
> {
use boatramp_core::project_delta as pd;
if opts.data_only {
return Err(Perr::Other(
"delta export is not supported for a --data-only bundle (a delta carries the \
control-plane KV delta, which a data-only bundle excludes)"
.into(),
));
}
if opts.include_identity {
return Err(Perr::Other(
"delta export does not carry the identity transport container; include identity \
only in the FULL base bundle (import it with --identity adopt), then chain deltas"
.into(),
));
}
let base_dir = self
.resolve_resumable_dir(&req.base_reference, false)
.await?;
let base_manifest = self.read_resumable_manifest(&base_dir).await?;
if matches!(req.mode, pd::DeltaMode::Differential) && base_manifest.delta.is_some() {
return Err(Perr::Other(format!(
"differential delta requires a FULL base bundle, but {:?} is itself a delta — use \
--mode incremental to chain onto a delta",
req.base_reference
)));
}
let chain = self.resolve_chain(&base_dir, &base_manifest).await?;
if chain[0].1.data_only {
return Err(Perr::Other(
"delta export refused: the root base bundle is --data-only; a delta needs a FULL \
base (with the control-plane KV slice) to diff against"
.into(),
));
}
if chain[0].1.source.project != project {
return Err(Perr::Other(format!(
"delta export refused: the base chain's root is for project {:?}, not {project:?}",
chain[0].1.source.project
)));
}
for (db, cfg) in &self.databases {
if Self::pg_capture_eligible(cfg) {
self.assert_pg_delta_capture_live(project, db).await?;
} else if Self::mysql_capture_eligible(cfg) {
self.assert_mysql_delta_capture_live(project, db).await?;
}
}
let (kv_frontier, blob_frontier) = self.compute_chain_frontier(&chain).await?;
let full_opts = ExportOptions {
delta: None,
..opts.clone()
};
let (full_specs, mut full_jobs, source) =
self.plan_export(project, &full_opts, created_at).await?;
let all_kv = boatramp_core::kv_dump::scan_dumpable(self.kv.as_ref())
.await
.map_err(other)?;
let current_kv = boatramp_core::project_export::scope_entries_to_project(
all_kv,
ProjectRef::new(project),
);
let base_kv_entries: Vec<boatramp_core::kv::KvDumpEntry> = kv_frontier
.iter()
.map(|(k, v)| boatramp_core::kv::KvDumpEntry {
key: k.clone(),
value: v.clone(),
version: 1,
})
.collect();
let kv_delta = pd::compute_kv_delta(&base_kv_entries, ¤t_kv);
let kv_delta_bytes = pd::encode_kv_delta(&kv_delta, created_at);
let current_blob_metas = self
.blobs
.list(&blob_project_prefix(project))
.await
.map_err(other)?;
let mut current_sizes: HashMap<String, Option<u64>> = HashMap::new();
let mut current_keys: std::collections::BTreeSet<String> =
std::collections::BTreeSet::new();
for m in ¤t_blob_metas {
current_sizes.insert(m.key.clone(), m.size);
current_keys.insert(m.key.clone());
}
let parent_capture = if matches!(req.mode, pd::DeltaMode::Incremental) {
read_pg_capture_sidecar(&base_dir).await
} else {
None
};
let parent_mysql_capture = if matches!(req.mode, pd::DeltaMode::Incremental) {
read_mysql_capture_sidecar(&base_dir).await
} else {
None
};
let parent_libsql_capture = read_libsql_capture_sidecar(&base_dir).await;
let cap_dir = self
.work_dir
.join(format!("delta-capture-{project}-{created_at}"));
if parent_capture.is_some()
|| parent_mysql_capture.is_some()
|| parent_libsql_capture.is_some()
{
tokio::fs::create_dir_all(&cap_dir).await.map_err(other)?;
}
let mut any_pg_true_delta = false;
let mut any_mysql_true_delta = false;
let mut any_libsql_true_delta = false;
let mut ordered: Vec<(MemberSpec, StageJob)> = Vec::new();
for spec in &full_specs {
match spec.kind {
MemberKind::Identity => { }
MemberKind::Kv => {
ordered.push((
MemberSpec::sidecar(
project_bundle::KV_DELTA_MEMBER_PATH.to_string(),
MemberKind::KvDelta,
None,
),
StageJob::KvDelta {
bytes: kv_delta_bytes.clone(),
},
));
}
MemberKind::Blob => {
let storage_key = spec
.path
.strip_prefix("blobs/")
.unwrap_or(&spec.path)
.to_string();
let base_sig = blob_frontier.get(&storage_key);
let cur_size = current_sizes.get(&storage_key).copied().flatten();
let is_add = match pd::blob_decision_by_size(base_sig, cur_size) {
pd::BlobDecision::Add => true,
pd::BlobDecision::NeedsContentCheck => match base_sig {
Some(b) => self.blob_sha256(&storage_key).await? != b.sha256,
None => true,
},
};
if is_add {
let job = full_jobs.remove(&spec.path).ok_or_else(|| {
Perr::Other(format!(
"delta export: no stage job for blob {}",
spec.path
))
})?;
ordered.push((spec.clone(), job));
}
}
MemberKind::PgDump => {
let db = db_from_pg_member(&spec.path);
let wal_member = match (db, parent_capture.as_ref()) {
(Some(db), Some(sidecar)) => {
self.try_pg_wal_delta(project, db, sidecar, &cap_dir).await
}
_ => None,
};
if let Some((bytes, cursor)) = wal_member {
any_pg_true_delta = true;
let db = db.expect("wal_member implies a parsed db");
ordered.push((
MemberSpec::sidecar(
pgwal_member_path(db),
MemberKind::PgWalDelta,
None,
)
.with_sql_cursor(cursor),
StageJob::PgWalDelta { bytes },
));
let _ = full_jobs.remove(&spec.path);
} else {
let job = full_jobs.remove(&spec.path).ok_or_else(|| {
Perr::Other(format!("delta export: no stage job for {}", spec.path))
})?;
ordered.push((spec.clone(), job));
}
}
MemberKind::MysqlDump => {
let db = db_from_mysql_member(&spec.path);
let binlog_member = match (db, parent_mysql_capture.as_ref()) {
(Some(db), Some(sidecar)) => {
self.try_mysql_binlog_delta(project, db, sidecar, &cap_dir)
.await
}
_ => None,
};
if let Some((bytes, cursor)) = binlog_member {
any_mysql_true_delta = true;
let db = db.expect("binlog_member implies a parsed db");
ordered.push((
MemberSpec::sidecar(
mysqlbinlog_member_path(db),
MemberKind::MysqlBinlogDelta,
None,
)
.with_sql_cursor(cursor),
StageJob::MysqlBinlogDelta { bytes },
));
let _ = full_jobs.remove(&spec.path);
} else {
let job = full_jobs.remove(&spec.path).ok_or_else(|| {
Perr::Other(format!("delta export: no stage job for {}", spec.path))
})?;
ordered.push((spec.clone(), job));
}
}
MemberKind::Libsql => {
let site = site_from_libsql_member(&spec.path);
let cs_member = match (site, parent_libsql_capture.as_ref()) {
(Some(site), Some(sidecar)) => match sidecar.sites.get(site) {
Some(entry) => {
self.try_libsql_changeset_delta(
project, site, entry, &base_dir, &cap_dir,
)
.await
}
None => None,
},
_ => None,
};
if let Some((bytes, cursor)) = cs_member {
any_libsql_true_delta = true;
let site = site.expect("cs_member implies a parsed site");
ordered.push((
MemberSpec::sidecar(
libsql_changeset_member_path(project, site),
MemberKind::LibsqlChangeset,
None,
)
.with_sql_cursor(cursor),
StageJob::LibsqlChangeset { bytes },
));
let _ = full_jobs.remove(&spec.path);
} else {
let job = full_jobs.remove(&spec.path).ok_or_else(|| {
Perr::Other(format!("delta export: no stage job for {}", spec.path))
})?;
ordered.push((spec.clone(), job));
}
}
MemberKind::KvDelta
| MemberKind::BlobTombstone
| MemberKind::PgWalDelta
| MemberKind::MysqlBinlogDelta
| MemberKind::LibsqlChangeset => {
}
}
}
let base_blob_keys: std::collections::BTreeSet<String> =
blob_frontier.keys().cloned().collect();
let tombstones = pd::compute_blob_tombstones(&base_blob_keys, ¤t_keys);
if !tombstones.is_empty() {
ordered.push((
MemberSpec::sidecar(
project_bundle::BLOB_TOMBSTONE_MEMBER_PATH.to_string(),
MemberKind::BlobTombstone,
None,
),
StageJob::BlobTombstone {
bytes: pd::encode_blob_tombstones(&tombstones),
},
));
}
let _ = tokio::fs::remove_dir_all(&cap_dir).await;
let specs: Vec<MemberSpec> = ordered.iter().map(|(s, _)| s.clone()).collect();
let jobs: HashMap<String, StageJob> =
ordered.into_iter().map(|(s, j)| (s.path, j)).collect();
let delta_info = pd::DeltaInfo {
mode: req.mode,
base_reference: req.base_reference.clone(),
base_created_at: base_manifest.created_at,
sql_strategy: if any_pg_true_delta {
pd::SqlDeltaKind::PgWalReplay
} else if any_mysql_true_delta {
pd::SqlDeltaKind::MysqlBinlogReplay
} else if any_libsql_true_delta {
pd::SqlDeltaKind::LibsqlChangeset
} else {
pd::SqlDeltaKind::FullDumpFallback
},
};
Ok((specs, jobs, source, delta_info))
}
async fn read_resumable_manifest(
&self,
dir: &Path,
) -> Result<project_bundle::StreamManifest, Perr> {
let manifest_path = dir.join(project_bundle::MANIFEST_PATH);
let bytes = tokio::fs::read(&manifest_path).await.map_err(|e| {
Perr::Other(format!(
"read base/parent MANIFEST at {}: {e}",
manifest_path.display()
))
})?;
serde_json::from_slice(&bytes)
.map_err(|e| Perr::Other(format!("parse base/parent MANIFEST: {e}")))
}
async fn resolve_chain(
&self,
start_dir: &Path,
start_manifest: &project_bundle::StreamManifest,
) -> Result<Vec<(PathBuf, project_bundle::StreamManifest)>, Perr> {
let mut chain = vec![(start_dir.to_path_buf(), start_manifest.clone())];
let mut guard = 0usize;
while let Some(di) = chain.last().unwrap().1.delta.clone() {
guard += 1;
if guard > 256 {
return Err(Perr::Other(
"delta chain exceeds 256 links (cycle or runaway) — refusing".into(),
));
}
let parent_dir = self
.resolve_resumable_dir(&di.base_reference, false)
.await?;
let parent_manifest = self.read_resumable_manifest(&parent_dir).await?;
if parent_manifest.created_at != di.base_created_at {
return Err(Perr::Other(format!(
"delta base mismatch: {:?} records base created_at {} but the resolved base has \
{} — refusing a stale or mismatched base",
di.base_reference, di.base_created_at, parent_manifest.created_at
)));
}
chain.push((parent_dir, parent_manifest));
}
chain.reverse();
Ok(chain)
}
async fn compute_chain_frontier(
&self,
chain: &[(PathBuf, project_bundle::StreamManifest)],
) -> Result<
(
BTreeMap<String, Vec<u8>>,
BTreeMap<String, boatramp_core::project_delta::BlobSig>,
),
Perr,
> {
use boatramp_core::project_delta as pd;
let mut kv: BTreeMap<String, Vec<u8>> = BTreeMap::new();
let mut blobs: BTreeMap<String, pd::BlobSig> = BTreeMap::new();
for (dir, m) in chain {
for (idx, spec) in m.members.iter().enumerate() {
match spec.kind {
MemberKind::Kv => {
let bytes = self.read_member_object(dir, idx).await?;
let decoded = boatramp_core::kv_dump::decode_dump(&bytes).map_err(other)?;
for e in decoded.entries {
kv.insert(e.key, e.value);
}
}
MemberKind::KvDelta => {
let bytes = self.read_member_object(dir, idx).await?;
let delta = pd::decode_kv_delta(&bytes).map_err(other)?;
pd::apply_kv_delta_to_frontier(&mut kv, &delta);
}
MemberKind::Blob => {
let storage_key = spec
.path
.strip_prefix("blobs/")
.unwrap_or(&spec.path)
.to_string();
let meta = self.read_member_meta(dir, idx).await?;
blobs.insert(
storage_key,
pd::BlobSig {
size: meta.size,
sha256: meta.sha256,
},
);
}
MemberKind::BlobTombstone => {
let bytes = self.read_member_object(dir, idx).await?;
for k in pd::decode_blob_tombstones(&bytes).map_err(other)? {
blobs.remove(&k);
}
}
_ => {}
}
}
}
Ok((kv, blobs))
}
async fn read_member_object(&self, dir: &Path, idx: usize) -> Result<Vec<u8>, Perr> {
tokio::fs::read(dir.join(member_object_name(idx)))
.await
.map_err(|e| Perr::Other(format!("read member object m{idx:05}: {e}")))
}
async fn read_member_meta(
&self,
dir: &Path,
idx: usize,
) -> Result<project_bundle::MemberMeta, Perr> {
let bytes = tokio::fs::read(dir.join(member_meta_name(idx)))
.await
.map_err(|e| Perr::Other(format!("read member meta m{idx:05}: {e}")))?;
serde_json::from_slice(&bytes)
.map_err(|e| Perr::Other(format!("parse member meta m{idx:05}: {e}")))
}
async fn blob_sha256(&self, storage_key: &str) -> Result<String, Perr> {
let obj = self.blobs.get(storage_key).await.map_err(other)?;
let mut body = obj.body;
let mut hasher = Sha256::new();
while let Some(chunk) = body.next().await {
let chunk = chunk.map_err(other)?;
hasher.update(chunk.as_ref());
}
Ok(hex::encode(hasher.finalize()))
}
async fn import_delta_chain(
&self,
project: &str,
this_dir: &Path,
this_manifest: &project_bundle::StreamManifest,
opts: &ImportOptions,
) -> Result<ImportReport, Perr> {
let di = this_manifest
.delta
.as_ref()
.ok_or_else(|| Perr::Other("import_delta_chain called on a non-delta bundle".into()))?;
let base_dir = self
.resolve_resumable_dir(&di.base_reference, false)
.await?;
let base_manifest = self.read_resumable_manifest(&base_dir).await?;
let mut chain = self.resolve_chain(&base_dir, &base_manifest).await?; chain.push((this_dir.to_path_buf(), this_manifest.clone())); if chain[0].1.delta.is_some() {
return Err(Perr::Other(
"delta chain does not terminate in a FULL base bundle — refusing".into(),
));
}
let root_src = &chain[0].1.source;
if let Some((bad_dir, bad)) = chain
.iter()
.find(|(_, m)| m.source.project != root_src.project || m.source.site != root_src.site)
{
return Err(Perr::Other(format!(
"delta import refused: chain link {:?} is for source {:?}/{:?}, but the chain root \
is {:?}/{:?} — a delta chain must share one source",
bad_dir, bad.source.project, bad.source.site, root_src.project, root_src.site
)));
}
if !opts.overwrite {
for prefix in
boatramp_core::project_export::project_owned_prefixes(ProjectRef::new(project))
{
if !self
.kv
.list_prefix(&prefix)
.await
.map_err(other)?
.is_empty()
{
return Err(Perr::Other(format!(
"import refused: target project {project:?} already has control-plane data \
under {prefix:?}; pass overwrite=true"
)));
}
}
if !self
.blobs
.list(&blob_project_prefix(project))
.await
.map_err(other)?
.is_empty()
{
return Err(Perr::Other(format!(
"import refused: target project {project:?} already has blobs; pass \
overwrite=true"
)));
}
}
let session = self
.work_dir
.join(format!("import-delta-{project}-{}", now_secs()));
tokio::fs::create_dir_all(&session).await.map_err(other)?;
let mut report = ImportReport::default();
let mut saw_identity = false;
let last = chain.len() - 1;
let libsql_changeset_sites: std::collections::BTreeSet<String> = chain
.iter()
.flat_map(|(_, m)| m.members.iter())
.filter(|s| s.kind == MemberKind::LibsqlChangeset)
.filter_map(|s| site_from_libsql_changeset_member(&s.path).map(str::to_string))
.collect();
for (pos, (dir, m)) in chain.iter().enumerate() {
let is_newest = pos == last;
for (idx, spec) in m.members.iter().enumerate() {
let obj = dir.join(member_object_name(idx));
let meta = dir.join(member_meta_name(idx));
if !member_object_valid(&obj, &meta).await? {
let _ = tokio::fs::remove_dir_all(&session).await;
return Err(Perr::Other(format!(
"delta restore: chain member {idx} ({}) in {} is missing or fails its SHA",
spec.path,
dir.display()
)));
}
let apply_here = match spec.kind {
MemberKind::PgDump => {
!db_from_pg_member(&spec.path)
.map(|db| self.is_managed_pg(db))
.unwrap_or(false)
&& is_newest
}
MemberKind::PgWalDelta => false,
MemberKind::MysqlDump => {
!db_from_mysql_member(&spec.path)
.map(|db| self.is_managed_mysql(db))
.unwrap_or(false)
&& is_newest
}
MemberKind::MysqlBinlogDelta => false,
MemberKind::Libsql => {
is_newest
&& !site_from_libsql_member(&spec.path)
.map(|s| libsql_changeset_sites.contains(s))
.unwrap_or(false)
}
MemberKind::LibsqlChangeset => false,
MemberKind::Identity => false,
_ => true,
};
if !apply_here {
continue;
}
if let Err(e) = self
.restore_member(
project,
m.data_only,
&m.source.project,
spec,
&obj,
opts,
&mut report,
&mut saw_identity,
)
.await
{
let _ = tokio::fs::remove_dir_all(&session).await;
return Err(e);
}
}
}
if let Err(e) = self
.restore_pg_true_delta_in_chain(project, &chain, opts, &mut report, &session)
.await
{
let _ = tokio::fs::remove_dir_all(&session).await;
return Err(e);
}
if let Err(e) = self
.restore_mysql_true_delta_in_chain(project, &chain, opts, &mut report, &session)
.await
{
let _ = tokio::fs::remove_dir_all(&session).await;
return Err(e);
}
if let Err(e) = self
.restore_libsql_true_delta_in_chain(
project,
&chain,
&libsql_changeset_sites,
opts,
&mut report,
)
.await
{
let _ = tokio::fs::remove_dir_all(&session).await;
return Err(e);
}
let _ = tokio::fs::remove_dir_all(&session).await;
report.notes.push(format!(
"delta restore: applied a {}-link chain ({} base + {} delta(s)) in {} mode",
chain.len(),
1,
chain.len() - 1,
di.mode.as_flag()
));
Ok(report)
}
pub async fn import_resumable_dir(
&self,
project: &str,
dir: &Path,
opts: ImportOptions,
) -> Result<ImportReport, Perr> {
check_project_slug(project)?;
let manifest_path = dir.join(project_bundle::MANIFEST_PATH);
let bytes = tokio::fs::read(&manifest_path)
.await
.map_err(|e| Perr::Other(format!("read resumable MANIFEST: {e}")))?;
let manifest: project_bundle::StreamManifest = serde_json::from_slice(&bytes)
.map_err(|e| Perr::Other(format!("parse resumable MANIFEST: {e}")))?;
if manifest.delta.is_some() {
return self
.import_delta_chain(project, dir, &manifest, &opts)
.await;
}
if !opts.overwrite {
for prefix in
boatramp_core::project_export::project_owned_prefixes(ProjectRef::new(project))
{
if !self
.kv
.list_prefix(&prefix)
.await
.map_err(other)?
.is_empty()
{
return Err(Perr::Other(format!(
"import refused: target project {project:?} already has control-plane \
data under {prefix:?}; pass overwrite=true"
)));
}
}
if !self
.blobs
.list(&blob_project_prefix(project))
.await
.map_err(other)?
.is_empty()
{
return Err(Perr::Other(format!(
"import refused: target project {project:?} already has blobs; pass \
overwrite=true"
)));
}
}
let session = self
.work_dir
.join(format!("import-res-{project}-{}", now_secs()));
tokio::fs::create_dir_all(&session).await.map_err(other)?;
let mut report = ImportReport::default();
let mut saw_identity = false;
for (idx, spec) in manifest.members.iter().enumerate() {
let obj = dir.join(member_object_name(idx));
let meta = dir.join(member_meta_name(idx));
if !member_object_valid(&obj, &meta).await? {
return Err(Perr::Other(format!(
"resumable import: member {idx} ({}) is missing or fails its SHA — the \
export directory is incomplete or corrupt",
spec.path
)));
}
self.restore_member(
project,
manifest.data_only,
&manifest.source.project,
spec,
&obj,
&opts,
&mut report,
&mut saw_identity,
)
.await?;
}
let _ = tokio::fs::remove_dir_all(&session).await;
Ok(report)
}
async fn pg_shadow_plan(
&self,
project: &str,
db: &str,
cfg: &ExternalDatabaseConfig,
) -> Result<PgShadowPlan, Perr> {
if cfg.compute.as_deref().is_none_or(|c| c.is_empty()) {
return Err(Perr::Unavailable(format!(
"database {db:?}: the atomic chunked pg shadow-restore is for MANAGED \
(compute-backed) databases only; an external bring-your-own-URL binding uses the \
sidecar path"
)));
}
if matches!(cfg.tenant, crate::config::TenantIsolation::Shared) {
return Err(Perr::Unavailable(format!(
"database {db:?}: the atomic chunked pg shadow-restore is for DEDICATED/Single \
managed servers only; a Shared multi-tenant server's per-import physical shadow db \
is a cross-tenant surface deferred to the temp-schema follow-up"
)));
}
let (operator_conn, image, _ver) = self.resolve_pg(project, db, cfg).await?;
let live_db = operator_conn.database.clone();
let owner_user = operator_conn.user.clone();
let mut maint = operator_conn;
maint.database = "postgres".to_string();
let suffix = shadow_suffix();
let shadow_db = format!("{live_db}__shadow_{suffix}");
let eph_role = format!("brexp_apply_{suffix}");
Ok(PgShadowPlan {
live_db,
shadow_db,
suffix,
maint,
eph_role,
eph_password: random_token(),
owner_user,
image,
trusted_extensions: self.trusted_extensions.clone(),
})
}
async fn pg_assert_not_superuser(
&self,
conn: &PgConnection,
image: &str,
stage_dir: &Path,
) -> Result<(), Perr> {
let inv = psql_invocation(
conn,
"SELECT rolsuper FROM pg_roles WHERE rolname = current_user",
);
let (code, out) = self
.run_sql_capture(image, &inv.argv, &inv.env, None, stage_dir)
.await?;
if code != 0 {
return Err(Perr::Other(format!(
"pg shadow: could not probe the apply role's superuser attribute (exit {code}): {out}"
)));
}
if out.trim() != "f" {
return Err(Perr::Other(format!(
"pg shadow REFUSED: the resolved apply role is a SUPERUSER (rolsuper={:?}); an \
untrusted bundle dump must only be applied by a non-superuser role",
out.trim()
)));
}
Ok(())
}
async fn pg_run_ddl(
&self,
plan: &PgShadowPlan,
sql: &str,
stage_dir: &Path,
) -> Result<(), Perr> {
let inv = psql_invocation(&plan.maint, sql);
let (code, out) = self
.run_sql_capture(&plan.image, &inv.argv, &inv.env, None, stage_dir)
.await?;
if code != 0 {
return Err(Perr::Other(format!(
"pg shadow DDL exited {code} ({sql}): {out}"
)));
}
Ok(())
}
async fn pg_db_exists(
&self,
plan: &PgShadowPlan,
name: &str,
stage_dir: &Path,
) -> Result<bool, Perr> {
let sql = format!("SELECT 1 FROM pg_database WHERE datname = {}", pg_lit(name));
let inv = psql_invocation(&plan.maint, &sql);
let (code, out) = self
.run_sql_capture(&plan.image, &inv.argv, &inv.env, None, stage_dir)
.await?;
if code != 0 {
return Err(Perr::Other(format!(
"pg shadow: could not probe database {name:?} (exit {code}): {out}"
)));
}
Ok(out.trim() == "1")
}
async fn pg_create_shadow(&self, plan: &PgShadowPlan, stage_dir: &Path) -> Result<(), Perr> {
let role_q = quote_ident(ExternalSqlKind::Postgres, &plan.eph_role);
let shadow_q = quote_ident(ExternalSqlKind::Postgres, &plan.shadow_db);
self.pg_run_ddl(
plan,
&format!(
"CREATE ROLE {role_q} LOGIN PASSWORD {} NOSUPERUSER NOCREATEDB NOCREATEROLE \
NOREPLICATION NOBYPASSRLS",
pg_lit(&plan.eph_password)
),
stage_dir,
)
.await?;
self.pg_run_ddl(
plan,
&format!("CREATE DATABASE {shadow_q} OWNER {role_q}"),
stage_dir,
)
.await?;
if !plan.trusted_extensions.is_empty() {
let maint_shadow = PgShadowPlan {
maint: plan.maint_on_shadow(),
..plan.clone()
};
for ext in &plan.trusted_extensions {
let ext_q = quote_ident(ExternalSqlKind::Postgres, ext);
self.pg_run_ddl(
&maint_shadow,
&format!("CREATE EXTENSION IF NOT EXISTS {ext_q}"),
stage_dir,
)
.await?;
}
}
Ok(())
}
async fn pg_drop_shadow(&self, plan: &PgShadowPlan, stage_dir: &Path) {
let shadow_q = quote_ident(ExternalSqlKind::Postgres, &plan.shadow_db);
let role_q = quote_ident(ExternalSqlKind::Postgres, &plan.eph_role);
let _ = self
.pg_run_ddl(
plan,
&format!(
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity \
WHERE datname = {} AND pid <> pg_backend_pid()",
pg_lit(&plan.shadow_db)
),
stage_dir,
)
.await;
if let Err(e) = self
.pg_run_ddl(
plan,
&format!("DROP DATABASE IF EXISTS {shadow_q}"),
stage_dir,
)
.await
{
tracing::warn!(error = %e, shadow = %plan.shadow_db, "pg shadow drop-on-abort failed; the startup sweep will reclaim it");
}
if let Err(e) = self
.pg_run_ddl(plan, &format!("DROP ROLE IF EXISTS {role_q}"), stage_dir)
.await
{
tracing::warn!(error = %e, role = %plan.eph_role, "pg ephemeral apply-role drop-on-abort failed; the startup sweep will reclaim it");
}
}
async fn pg_swap_shadow(&self, plan: &PgShadowPlan, stage_dir: &Path) -> Result<(), Perr> {
let live_q = quote_ident(ExternalSqlKind::Postgres, &plan.live_db);
let shadow_q = quote_ident(ExternalSqlKind::Postgres, &plan.shadow_db);
let role_q = quote_ident(ExternalSqlKind::Postgres, &plan.eph_role);
let owner_q = quote_ident(ExternalSqlKind::Postgres, &plan.owner_user);
let old_db = format!("{}__old_{}", plan.live_db, plan.suffix);
let old_q = quote_ident(ExternalSqlKind::Postgres, &old_db);
if !self.pg_db_exists(plan, &plan.shadow_db, stage_dir).await? {
return Err(Perr::Other(format!(
"pg shadow swap REFUSED: the shadow database {:?} is absent — refusing to rename \
live {:?} aside without a replacement. Live left intact.",
plan.shadow_db, plan.live_db
)));
}
let maint_shadow = PgShadowPlan {
maint: plan.maint_on_shadow(),
..plan.clone()
};
self.pg_run_ddl(
&maint_shadow,
&format!("REASSIGN OWNED BY {role_q} TO {owner_q}"),
stage_dir,
)
.await?;
self.pg_run_ddl(
plan,
&format!("ALTER DATABASE {shadow_q} OWNER TO {owner_q}"),
stage_dir,
)
.await?;
let live_exists = self.pg_db_exists(plan, &plan.live_db, stage_dir).await?;
if live_exists {
self.pg_run_ddl(
plan,
&format!(
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity \
WHERE datname = {} AND pid <> pg_backend_pid()",
pg_lit(&plan.live_db)
),
stage_dir,
)
.await?;
self.pg_run_ddl(
plan,
&format!("ALTER DATABASE {live_q} RENAME TO {old_q}"),
stage_dir,
)
.await?;
}
self.pg_run_ddl(
plan,
&format!(
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity \
WHERE datname = {} AND pid <> pg_backend_pid()",
pg_lit(&plan.shadow_db)
),
stage_dir,
)
.await?;
if let Err(e) = self
.pg_run_ddl(
plan,
&format!("ALTER DATABASE {shadow_q} RENAME TO {live_q}"),
stage_dir,
)
.await
{
if live_exists {
let _ = self
.pg_run_ddl(
plan,
&format!("ALTER DATABASE {old_q} RENAME TO {live_q}"),
stage_dir,
)
.await;
}
return Err(e);
}
if live_exists {
let _ = self
.pg_run_ddl(
plan,
&format!(
"SELECT pg_terminate_backend(pid) FROM pg_stat_activity \
WHERE datname = {} AND pid <> pg_backend_pid()",
pg_lit(&old_db)
),
stage_dir,
)
.await;
let _ = self
.pg_run_ddl(plan, &format!("DROP DATABASE IF EXISTS {old_q}"), stage_dir)
.await;
}
let _ = self
.pg_run_ddl(plan, &format!("DROP ROLE IF EXISTS {role_q}"), stage_dir)
.await;
Ok(())
}
fn spawn_pg_restore(
&self,
plan: &PgShadowPlan,
body_rx: tokio::sync::mpsc::Receiver<Vec<u8>>,
out_path: PathBuf,
) -> Result<tokio::task::JoinHandle<Result<i32, Perr>>, Perr> {
let backend = self.require_backend()?.clone();
let image = plan.image.clone();
let inv = pg_restore_shadow_invocation(&plan.apply_conn(), &plan.shadow_db);
let cmd = format!("exec {} 2>&1", sh_join(&inv.argv));
let argv = vec!["sh".to_string(), "-c".to_string(), cmd];
let env = inv.env;
Ok(tokio::spawn(async move {
let mut reader = AsyncChannelReader::new(body_rx);
let out = tokio::fs::File::create(&out_path).await.map_err(other)?;
let mut sink = out.compat_write();
backend
.run_ephemeral(&image, &argv, &env, Some(&mut reader), &mut sink)
.await
.map_err(other)
}))
}
async fn mysql_shadow_plan(
&self,
project: &str,
db: &str,
cfg: &ExternalDatabaseConfig,
) -> Result<MysqlShadowPlan, Perr> {
if cfg.compute.as_deref().is_none_or(|c| c.is_empty()) {
return Err(Perr::Unavailable(format!(
"database {db:?}: the atomic chunked mysql shadow-restore is for MANAGED \
(compute-backed) databases only; an external binding uses the sidecar path"
)));
}
if matches!(cfg.tenant, crate::config::TenantIsolation::Shared) {
return Err(Perr::Unavailable(format!(
"database {db:?}: the atomic chunked mysql shadow-restore is for DEDICATED/Single \
managed servers only; a Shared multi-tenant server is deferred"
)));
}
let (operator_conn, image, _ver) = self.resolve_mysql(project, db, cfg).await?;
let live_db = operator_conn.database.clone();
let maint = operator_conn; let suffix = mysql_eph_suffix();
let shadow_db = format!("{live_db}__shadow_{suffix}");
let eph_user = format!("brexp_apply_{suffix}");
Ok(MysqlShadowPlan {
live_db,
shadow_db,
suffix,
maint,
eph_user,
eph_password: random_token(),
image,
})
}
async fn mysql_assert_not_superuser(
&self,
plan: &MysqlShadowPlan,
stage_dir: &Path,
) -> Result<(), Perr> {
let apply = plan.apply_conn();
let inv = mysql_admin_invocation(&apply);
let in_path = stage_dir.join(format!("mysql-probe-{}.sql", shadow_suffix()));
tokio::fs::write(&in_path, "SHOW GRANTS FOR CURRENT_USER();")
.await
.map_err(other)?;
let r = self
.run_sql_capture(&plan.image, &inv.argv, &inv.env, Some(&in_path), stage_dir)
.await;
let _ = tokio::fs::remove_file(&in_path).await;
let (code, out) = r?;
if code != 0 {
return Err(Perr::Other(format!(
"mysql shadow: could not probe the apply user's grants (exit {code}): {out}"
)));
}
for line in out.lines() {
let privs = mysql_grant_privileges(line).to_ascii_uppercase();
let up = line.to_ascii_uppercase();
let dangerous_global = up.contains(" ON *.*") && !up.contains("USAGE ON *.*");
if privs.contains("SUPER") || privs.contains("FILE") || dangerous_global {
return Err(Perr::Other(format!(
"mysql shadow REFUSED: the resolved apply user holds SUPER/FILE/*.* privileges; \
an untrusted bundle dump must only be applied by a least-privilege user. \
grants: {out}"
)));
}
}
Ok(())
}
async fn mysql_run_admin(
&self,
plan: &MysqlShadowPlan,
sql: &str,
stage_dir: &Path,
) -> Result<(i32, String), Perr> {
let inv = mysql_admin_invocation(&plan.maint);
let in_path = stage_dir.join(format!("mysql-admin-{}.sql", shadow_suffix()));
tokio::fs::write(&in_path, sql).await.map_err(other)?;
let r = self
.run_sql_capture(&plan.image, &inv.argv, &inv.env, Some(&in_path), stage_dir)
.await;
let _ = tokio::fs::remove_file(&in_path).await;
r
}
async fn mysql_list_tables(
&self,
plan: &MysqlShadowPlan,
db: &str,
stage_dir: &Path,
) -> Result<Vec<String>, Perr> {
let sql = format!(
"SELECT table_name FROM information_schema.tables \
WHERE table_schema = {} AND table_type = 'BASE TABLE';",
mysql_lit(db)
);
let (code, out) = self.mysql_run_admin(plan, &sql, stage_dir).await?;
if code != 0 {
return Err(Perr::Other(format!(
"mysql shadow: could not list tables of {db:?} (exit {code}): {out}"
)));
}
Ok(out
.lines()
.map(|l| l.trim().to_string())
.filter(|l| !l.is_empty())
.collect())
}
async fn mysql_db_exists(
&self,
plan: &MysqlShadowPlan,
name: &str,
stage_dir: &Path,
) -> Result<bool, Perr> {
let sql = format!(
"SELECT 1 FROM information_schema.schemata WHERE schema_name = {};",
mysql_lit(name)
);
let (code, out) = self.mysql_run_admin(plan, &sql, stage_dir).await?;
if code != 0 {
return Err(Perr::Other(format!(
"mysql shadow: could not probe database {name:?} (exit {code}): {out}"
)));
}
Ok(out.trim() == "1")
}
async fn mysql_create_shadow(
&self,
plan: &MysqlShadowPlan,
stage_dir: &Path,
) -> Result<(), Perr> {
let shadow_q = quote_ident(ExternalSqlKind::Mysql, &plan.shadow_db);
let user_lit = mysql_lit(&plan.eph_user);
let pw_lit = mysql_lit(&plan.eph_password);
let (code, out) = self
.mysql_run_admin(
plan,
&format!(
"CREATE DATABASE {shadow_q}; \
CREATE USER {user_lit}@'%' IDENTIFIED BY {pw_lit}; \
GRANT ALL PRIVILEGES ON {shadow_q}.* TO {user_lit}@'%';"
),
stage_dir,
)
.await?;
if code != 0 {
return Err(Perr::Other(format!(
"mysql shadow create (db + ephemeral user) exited {code}: {out}"
)));
}
Ok(())
}
async fn mysql_drop_shadow(&self, plan: &MysqlShadowPlan, stage_dir: &Path) {
let shadow_q = quote_ident(ExternalSqlKind::Mysql, &plan.shadow_db);
let user_lit = mysql_lit(&plan.eph_user);
if let Err(e) = self
.mysql_run_admin(
plan,
&format!("DROP DATABASE IF EXISTS {shadow_q}; DROP USER IF EXISTS {user_lit}@'%';"),
stage_dir,
)
.await
{
tracing::warn!(error = %e, shadow = %plan.shadow_db, "mysql shadow drop-on-abort failed; the startup sweep will reclaim it");
}
}
async fn mysql_swap_shadow(
&self,
plan: &MysqlShadowPlan,
stage_dir: &Path,
) -> Result<(), Perr> {
let live_tables = self
.mysql_list_tables(plan, &plan.live_db, stage_dir)
.await?;
let shadow_tables = self
.mysql_list_tables(plan, &plan.shadow_db, stage_dir)
.await?;
if swap_would_destroy_live(&live_tables, &shadow_tables) {
return Err(Perr::Other(format!(
"mysql shadow swap REFUSED: the shadow {:?} is EMPTY while live {:?} has \
{} table(s) — swapping would destroy live data (a concurrent same-db import may \
have dropped this import's shadow). Live left intact.",
plan.shadow_db,
plan.live_db,
live_tables.len()
)));
}
let liveq = |t: &str| {
format!(
"{}.{}",
quote_ident(ExternalSqlKind::Mysql, &plan.live_db),
quote_ident(ExternalSqlKind::Mysql, t)
)
};
let shadowq = |t: &str| {
format!(
"{}.{}",
quote_ident(ExternalSqlKind::Mysql, &plan.shadow_db),
quote_ident(ExternalSqlKind::Mysql, t)
)
};
let aside_name = |t: &str| format!("__old_{}_{t}", plan.suffix);
let mut moves: Vec<String> = Vec::new();
for t in &live_tables {
moves.push(format!("{} TO {}", liveq(t), liveq(&aside_name(t))));
}
for t in &shadow_tables {
moves.push(format!("{} TO {}", shadowq(t), liveq(t)));
}
if !moves.is_empty() {
let (code, out) = self
.mysql_run_admin(
plan,
&format!("RENAME TABLE {};", moves.join(", ")),
stage_dir,
)
.await?;
if code != 0 {
return Err(Perr::Other(format!(
"mysql shadow swap (RENAME TABLE) exited {code}: {out}"
)));
}
}
for t in &live_tables {
let _ = self
.mysql_run_admin(
plan,
&format!("DROP TABLE IF EXISTS {};", liveq(&aside_name(t))),
stage_dir,
)
.await;
}
self.mysql_drop_shadow(plan, stage_dir).await;
Ok(())
}
fn spawn_mysql_restore(
&self,
plan: &MysqlShadowPlan,
body_rx: tokio::sync::mpsc::Receiver<Vec<u8>>,
out_path: PathBuf,
) -> Result<tokio::task::JoinHandle<Result<i32, Perr>>, Perr> {
let backend = self.require_backend()?.clone();
let image = plan.image.clone();
let inv = mysql_shadow_invocation(&plan.apply_conn(), &plan.shadow_db);
Ok(tokio::spawn(async move {
let mut reader = AsyncChannelReader::new(body_rx);
let out = tokio::fs::File::create(&out_path).await.map_err(other)?;
let mut sink = out.compat_write();
backend
.run_ephemeral(&image, &inv.argv, &inv.env, Some(&mut reader), &mut sink)
.await
.map_err(other)
}))
}
pub async fn sweep_stale_sql_shadows(&self, project: &str) -> usize {
let mut dropped = 0usize;
let session = self.work_dir.join(format!("shadow-sweep-{}", now_secs()));
if tokio::fs::create_dir_all(&session).await.is_err() {
return 0;
}
let wanted_capture = self.collect_pending_capture_names().await;
for (db, cfg) in &self.databases {
if cfg.compute.as_deref().is_none_or(|c| c.is_empty()) {
continue; }
if kind_is_postgres(&cfg.kind) {
if let Ok(plan) = self.pg_shadow_plan(project, db, cfg).await {
let sql = "SELECT datname FROM pg_database \
WHERE datname LIKE '%\\_\\_shadow\\_%' OR datname LIKE '%\\_\\_old\\_%'";
let inv = psql_invocation(&plan.maint, sql);
if let Ok((0, out)) = self
.run_sql_capture(&plan.image, &inv.argv, &inv.env, None, &session)
.await
{
for name in out.lines().map(str::trim).filter(|l| !l.is_empty()) {
let q = quote_ident(ExternalSqlKind::Postgres, name);
if self
.pg_run_ddl(
&plan,
&format!("DROP DATABASE IF EXISTS {q}"),
&session,
)
.await
.is_ok()
{
dropped += 1;
}
}
}
let role_sql =
"SELECT rolname FROM pg_roles WHERE rolname LIKE 'brexp\\_apply\\_%'";
let role_inv = psql_invocation(&plan.maint, role_sql);
if let Ok((0, out)) = self
.run_sql_capture(&plan.image, &role_inv.argv, &role_inv.env, None, &session)
.await
{
for role in out.lines().map(str::trim).filter(|l| !l.is_empty()) {
let q = quote_ident(ExternalSqlKind::Postgres, role);
if self
.pg_run_ddl(&plan, &format!("DROP ROLE IF EXISTS {q}"), &session)
.await
.is_ok()
{
dropped += 1;
}
}
}
let slot_sql = "SELECT slot_name FROM pg_replication_slots \
WHERE slot_name LIKE 'brexp\\_slot\\_%' AND NOT active";
let slot_inv = psql_invocation(&plan.maint, slot_sql);
if let Ok((0, out)) = self
.run_sql_capture(&plan.image, &slot_inv.argv, &slot_inv.env, None, &session)
.await
{
for slot in out.lines().map(str::trim).filter(|l| !l.is_empty()) {
if wanted_capture.contains(slot) {
continue;
}
if self
.pg_run_ddl(
&plan,
&format!("SELECT pg_drop_replication_slot({})", pg_lit(slot)),
&session,
)
.await
.is_ok()
{
dropped += 1;
}
}
}
let crole_sql =
"SELECT rolname FROM pg_roles WHERE rolname LIKE 'brexp\\_capture\\_%'";
let crole_inv = psql_invocation(&plan.maint, crole_sql);
if let Ok((0, out)) = self
.run_sql_capture(
&plan.image,
&crole_inv.argv,
&crole_inv.env,
None,
&session,
)
.await
{
for role in out.lines().map(str::trim).filter(|l| !l.is_empty()) {
if wanted_capture.contains(role) {
continue;
}
let q = quote_ident(ExternalSqlKind::Postgres, role);
if self
.pg_run_ddl(&plan, &format!("DROP ROLE IF EXISTS {q}"), &session)
.await
.is_ok()
{
dropped += 1;
}
}
}
let maint_live = {
let mut c = plan.maint.clone();
c.database = plan.live_db.clone();
c
};
let pub_sql =
"SELECT pubname FROM pg_publication WHERE pubname LIKE 'brexp\\_pub\\_%'";
let pub_inv = psql_invocation(&maint_live, pub_sql);
if let Ok((0, out)) = self
.run_sql_capture(&plan.image, &pub_inv.argv, &pub_inv.env, None, &session)
.await
{
for p in out.lines().map(str::trim).filter(|l| !l.is_empty()) {
if wanted_capture.contains(p) {
continue;
}
let q = quote_ident(ExternalSqlKind::Postgres, p);
let pdrop = psql_invocation(
&maint_live,
&format!("DROP PUBLICATION IF EXISTS {q}"),
);
if let Ok((0, _)) = self
.run_sql_capture(
&plan.image,
&pdrop.argv,
&pdrop.env,
None,
&session,
)
.await
{
dropped += 1;
}
}
}
}
} else if kind_is_mysql(&cfg.kind)
&& let Ok(plan) = self.mysql_shadow_plan(project, db, cfg).await
{
let sql = "SELECT schema_name FROM information_schema.schemata \
WHERE schema_name LIKE '%\\_\\_shadow\\_%';";
if let Ok((0, out)) = self.mysql_run_admin(&plan, sql, &session).await {
for name in out.lines().map(str::trim).filter(|l| !l.is_empty()) {
let q = quote_ident(ExternalSqlKind::Mysql, name);
if let Ok((0, _)) = self
.mysql_run_admin(
&plan,
&format!("DROP DATABASE IF EXISTS {q};"),
&session,
)
.await
{
dropped += 1;
}
}
}
let user_sql = "SELECT user FROM mysql.user \
WHERE user LIKE 'brexp\\_apply\\_%' OR user LIKE 'brexp\\_rd\\_%';";
if let Ok((0, out)) = self.mysql_run_admin(&plan, user_sql, &session).await {
for u in out.lines().map(str::trim).filter(|l| !l.is_empty()) {
let user_lit = mysql_lit(u);
if let Ok((0, _)) = self
.mysql_run_admin(
&plan,
&format!("DROP USER IF EXISTS {user_lit}@'%';"),
&session,
)
.await
{
dropped += 1;
}
}
}
}
}
let _ = tokio::fs::remove_dir_all(&session).await;
dropped
}
}
const PG_CAPTURE_SIDECAR: &str = "pg_capture.json";
const STALE_RESUMABLE_CAPTURE_TTL_SECS: u64 = 7 * 24 * 60 * 60;
fn create_replication_slot_export_sql(slot: &str, server_version_num: &str) -> String {
let slot_q = format!("\"{}\"", slot.replace('"', "\"\""));
let opts = match pg_server_major(server_version_num) {
Some(m) if m >= 15 => "(SNAPSHOT 'export')",
_ => "EXPORT_SNAPSHOT",
};
format!("CREATE_REPLICATION_SLOT {slot_q} LOGICAL pgoutput {opts};")
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct PgCaptureEntry {
slot: String,
publication: String,
role: String,
base_lsn: String,
#[serde(default)]
schema_fingerprint: Option<String>,
#[serde(default)]
created_at: u64,
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
struct PgCaptureSidecar {
dbs: BTreeMap<String, PgCaptureEntry>,
}
async fn read_pg_capture_sidecar(dir: &Path) -> Option<PgCaptureSidecar> {
let bytes = tokio::fs::read(dir.join(PG_CAPTURE_SIDECAR)).await.ok()?;
serde_json::from_slice(&bytes).ok()
}
async fn pg_capture_sidecar_age_secs(dir: &Path, now: u64) -> u64 {
let Ok(meta) = tokio::fs::metadata(dir.join(PG_CAPTURE_SIDECAR)).await else {
return 0;
};
let Ok(modified) = meta.modified() else {
return 0;
};
let mtime_secs = modified
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
now.saturating_sub(mtime_secs)
}
fn capture_entry_shielded(created_at: u64, file_age_secs: u64, now: u64, ttl: u64) -> bool {
let age = if created_at > 0 {
now.saturating_sub(created_at)
} else {
file_age_secs
};
age <= ttl
}
fn capture_role_conn(su: &PgConnection, role: &str, password: &str) -> PgConnection {
let mut c = su.clone();
c.user = role.to_string();
c.password = password.to_string();
c
}
impl NodeProjectExport {
fn pg_capture_eligible(cfg: &ExternalDatabaseConfig) -> bool {
is_managed_binding(cfg)
&& kind_is_postgres(&cfg.kind)
&& cfg.delta_capture
&& !matches!(cfg.tenant, crate::config::TenantIsolation::Shared)
}
fn is_managed_pg(&self, db: &str) -> bool {
self.databases
.get(db)
.map(|c| is_managed_binding(c) && kind_is_postgres(&c.kind))
.unwrap_or(false)
}
async fn pg_capture_sql(
&self,
conn: &PgConnection,
image: &str,
sql: &str,
stage_dir: &Path,
) -> Result<(i32, String), Perr> {
let inv = psql_invocation(conn, sql);
self.run_sql_capture(image, &inv.argv, &inv.env, None, stage_dir)
.await
}
async fn pg_begin_capture(
&self,
project: &str,
db: &str,
cfg: &ExternalDatabaseConfig,
stage_dir: &Path,
) -> Result<PgCaptureEntry, Perr> {
let (su, image, _ver) = self.resolve_pg(project, db, cfg).await?;
let suffix = shadow_suffix();
let slot = format!("brexp_slot_{suffix}");
let publication = format!("brexp_pub_{suffix}");
let role = format!("brexp_capture_{suffix}");
let role_pw = random_token();
let role_q = quote_ident(ExternalSqlKind::Postgres, &role);
let pub_q = quote_ident(ExternalSqlKind::Postgres, &publication);
let create_role = format!(
"CREATE ROLE {role_q} LOGIN PASSWORD {} REPLICATION NOSUPERUSER NOCREATEDB \
NOCREATEROLE NOBYPASSRLS",
pg_lit(&role_pw)
);
let (code, out) = self
.pg_capture_sql(&su, &image, &create_role, stage_dir)
.await?;
if code != 0 {
return Err(Perr::Other(format!(
"pg capture: CREATE capture role for {db:?} exited {code}: {out}"
)));
}
let (code, out) = self
.pg_capture_sql(
&su,
&image,
&format!("CREATE PUBLICATION {pub_q} FOR ALL TABLES"),
stage_dir,
)
.await?;
if code != 0 {
let _ = self
.pg_capture_sql(
&su,
&image,
&format!("DROP ROLE IF EXISTS {role_q}"),
stage_dir,
)
.await;
return Err(Perr::Other(format!(
"pg capture: CREATE PUBLICATION for {db:?} exited {code}: {out}"
)));
}
let schema_fingerprint = self.schema_fingerprint(project, db).await;
Ok(PgCaptureEntry {
slot,
publication,
role,
base_lsn: String::new(), schema_fingerprint,
created_at: now_secs(),
})
}
async fn produce_pg_capture_base(
&self,
project: &str,
dir: &Path,
idx: usize,
db: &str,
cfg: &ExternalDatabaseConfig,
stage_dir: &Path,
) -> Result<project_bundle::MemberMeta, Perr> {
let (conn, image, server_version_num) = self.resolve_pg(project, db, cfg).await?;
let entry = read_pg_capture_sidecar(dir)
.await
.and_then(|s| s.dbs.get(db).cloned());
let staged = stage_dir.join(format!("pgcap-base-{idx}"));
if let Some(entry) = entry {
match self
.pg_snapshot_pinned_dump(
&conn,
&image,
&server_version_num,
&entry,
&staged,
stage_dir,
)
.await
{
Ok(base_lsn) => {
self.pg_sidecar_set_base_lsn(dir, db, &base_lsn).await;
let meta = write_member_from_file(dir, idx, &staged).await?;
let _ = tokio::fs::remove_file(&staged).await;
return Ok(meta);
}
Err(e) => {
tracing::warn!(db = %db, error = %e, "pg delta capture: could not snapshot-pin the base; this db's delta will fall back to a full dump");
self.pg_drop_capture(project, db, cfg, &entry, stage_dir)
.await;
self.pg_sidecar_remove_db(dir, db).await;
}
}
}
let _ = tokio::fs::remove_file(&staged).await;
let exclude_extensions = self.trusted_extensions.clone();
let inv = pg_dump_invocation(
&conn,
false,
&server_version_num,
&exclude_extensions,
cfg.delta_capture,
);
let code = self
.run_sql_client(&image, &inv.argv, &inv.env, None, &staged)
.await?;
if code != 0 {
let _ = tokio::fs::remove_file(&staged).await;
return Err(Perr::Other(format!(
"pg capture base fallback dump for {db:?} exited {code}"
)));
}
let meta = write_member_from_file(dir, idx, &staged).await?;
let _ = tokio::fs::remove_file(&staged).await;
Ok(meta)
}
async fn pg_snapshot_pinned_dump(
&self,
conn: &PgConnection,
image: &str,
server_version_num: &str,
entry: &PgCaptureEntry,
staged: &Path,
stage_dir: &Path,
) -> Result<String, Perr> {
let role_q = quote_ident(ExternalSqlKind::Postgres, &entry.role);
let slot_lit = pg_lit(&entry.slot);
let (code, out) = self
.pg_capture_sql(
conn,
image,
&format!(
"SELECT pg_drop_replication_slot({slot_lit}) WHERE EXISTS (SELECT 1 FROM \
pg_replication_slots WHERE slot_name = {slot_lit})"
),
stage_dir,
)
.await?;
if code != 0 {
return Err(Perr::Other(format!(
"pg capture: drop stale slot {:?} before re-pin exited {code}: {out}",
entry.slot
)));
}
let role_pw = random_token();
let (code, out) = self
.pg_capture_sql(
conn,
image,
&format!("ALTER ROLE {role_q} PASSWORD {}", pg_lit(&role_pw)),
stage_dir,
)
.await?;
if code != 0 {
return Err(Perr::Other(format!(
"pg capture: reset capture-role password exited {code}: {out}"
)));
}
let exclude_extensions = self.trusted_extensions.clone();
let dump = pg_dump_invocation(conn, false, server_version_num, &exclude_extensions, true);
let dump_flags = sh_join(&dump.argv[1..]);
let slot_cmd = create_replication_slot_export_sql(&entry.slot, server_version_num);
let script = format!(
r#"set -u
COORD=$(mktemp) || exit 91
ERRF=$(mktemp) || exit 91
FIFO=$(mktemp -u) || exit 91
mkfifo "$FIFO" || exit 91
# The held replication session's query result → $COORD; its own stdout → /dev/null so NOTHING but
# the foreground pg_dump's `-Fc` archive ever reaches the container stdout (the member sink).
PGPASSWORD="$BREXP_CAP_PW" psql "host=$PGHOST port=$PGPORT user=$BREXP_CAP_ROLE dbname=$PGDATABASE replication=database" -At -F'|' -o "$COORD" <"$FIFO" >/dev/null 2>"$ERRF" &
REPL=$!
exec 9>"$FIFO"
printf '%s\n' {slot_cmd_q} >&9
SNAP=""
i=0
while [ $i -lt 600 ]; do
SNAP=$(awk -F'|' 'NR==1{{print $3; exit}}' "$COORD" 2>/dev/null)
[ -n "$SNAP" ] && break
i=$((i+1)); sleep 0.1
done
if [ -z "$SNAP" ]; then
echo "brexp: snapshot export failed: $(cat "$ERRF" 2>/dev/null)" >&2
exec 9>&-; wait "$REPL" 2>/dev/null; exit 92
fi
pg_dump {dump_flags} --snapshot="$SNAP"
RC=$?
exec 9>&-
wait "$REPL" 2>/dev/null
exit $RC
"#,
slot_cmd_q = sh_join(&[slot_cmd]),
dump_flags = dump_flags,
);
let mut env = conn.env();
env.push(("BREXP_CAP_ROLE".to_string(), entry.role.clone()));
env.push(("BREXP_CAP_PW".to_string(), role_pw));
let argv = vec!["sh".to_string(), "-c".to_string(), script];
let code = self
.run_sql_client(image, &argv, &env, None, staged)
.await?;
if code != 0 {
return Err(Perr::Other(format!(
"pg capture: snapshot-pinned base dump for slot {:?} exited {code} (snapshot export \
or pg_dump failed)",
entry.slot
)));
}
let (code, lsn) = self
.pg_capture_sql(
conn,
image,
&format!(
"SELECT confirmed_flush_lsn FROM pg_replication_slots WHERE slot_name = {slot_lit}"
),
stage_dir,
)
.await?;
let lsn = lsn.trim().lines().next().unwrap_or("").trim().to_string();
if code != 0 || lsn.is_empty() {
return Err(Perr::Other(format!(
"pg capture: snapshot-pinned slot {:?} has no confirmed_flush_lsn after creation \
(exit {code})",
entry.slot
)));
}
Ok(lsn)
}
async fn pg_sidecar_set_base_lsn(&self, dir: &Path, db: &str, base_lsn: &str) {
let Some(mut sidecar) = read_pg_capture_sidecar(dir).await else {
return;
};
if let Some(e) = sidecar.dbs.get_mut(db) {
e.base_lsn = base_lsn.to_string();
if let Ok(bytes) = serde_json::to_vec_pretty(&sidecar) {
let _ = write_atomic(&dir.join(PG_CAPTURE_SIDECAR), &bytes).await;
}
}
}
async fn pg_sidecar_remove_db(&self, dir: &Path, db: &str) {
let Some(mut sidecar) = read_pg_capture_sidecar(dir).await else {
return;
};
if sidecar.dbs.remove(db).is_some() {
if sidecar.dbs.is_empty() {
let _ = tokio::fs::remove_file(dir.join(PG_CAPTURE_SIDECAR)).await;
} else if let Ok(bytes) = serde_json::to_vec_pretty(&sidecar) {
let _ = write_atomic(&dir.join(PG_CAPTURE_SIDECAR), &bytes).await;
}
}
}
async fn pg_drain_capture(
&self,
project: &str,
db: &str,
cfg: &ExternalDatabaseConfig,
entry: &PgCaptureEntry,
stage_dir: &Path,
) -> Result<Option<Vec<u8>>, Perr> {
let (su, image, _ver) = self.resolve_pg(project, db, cfg).await?;
let slot_lit = pg_lit(&entry.slot);
let role_q = quote_ident(ExternalSqlKind::Postgres, &entry.role);
let probe = format!(
"SELECT coalesce((SELECT wal_status FROM pg_replication_slots WHERE slot_name = \
{slot_lit}), 'missing')"
);
let (code, status) = self.pg_capture_sql(&su, &image, &probe, stage_dir).await?;
if code != 0 {
return Err(Perr::Other(format!(
"pg capture: probe slot {:?} exited {code}: {status}",
entry.slot
)));
}
let status = status.trim();
if status == "missing" || status == "lost" {
self.pg_drop_capture(project, db, cfg, entry, stage_dir)
.await;
return Ok(None);
}
let role_pw = random_token();
let (code, out) = self
.pg_capture_sql(
&su,
&image,
&format!("ALTER ROLE {role_q} PASSWORD {}", pg_lit(&role_pw)),
stage_dir,
)
.await?;
if code != 0 {
self.pg_drop_capture(project, db, cfg, entry, stage_dir)
.await;
return Err(Perr::Other(format!(
"pg capture: reset capture-role password for {db:?} exited {code}: {out}"
)));
}
let role_conn = capture_role_conn(&su, &entry.role, &role_pw);
let drain = format!(
"SELECT encode(data, 'hex') FROM pg_logical_slot_get_binary_changes({slot_lit}, NULL, \
NULL, 'proto_version', '1', 'publication_names', {})",
pg_lit(&entry.publication)
);
let drain_inv = psql_invocation(&role_conn, &drain);
let (code, out) = self
.run_sql_capture_nomerge(&image, &drain_inv.argv, &drain_inv.env, None, stage_dir)
.await?;
if code != 0 {
self.pg_drop_capture(project, db, cfg, entry, stage_dir)
.await;
return Err(Perr::Other(format!(
"pg capture: drain slot {:?} for {db:?} exited {code}",
entry.slot
)));
}
let mut framed = Vec::new();
for line in out.lines() {
let h = line.trim();
if h.is_empty() {
continue;
}
let raw = match hex::decode(h) {
Ok(r) => r,
Err(e) => {
self.pg_drop_capture(project, db, cfg, entry, stage_dir)
.await;
return Err(Perr::Other(format!(
"pg capture: a drained row is not valid hex for {db:?}: {e}"
)));
}
};
framed.extend_from_slice(&boatramp_core::pgoutput::frame_message(&raw));
}
self.pg_drop_capture(project, db, cfg, entry, stage_dir)
.await;
Ok(Some(framed))
}
async fn pg_drop_capture(
&self,
project: &str,
db: &str,
cfg: &ExternalDatabaseConfig,
entry: &PgCaptureEntry,
stage_dir: &Path,
) {
let Ok((su, image, _ver)) = self.resolve_pg(project, db, cfg).await else {
return;
};
let role_q = quote_ident(ExternalSqlKind::Postgres, &entry.role);
let pub_q = quote_ident(ExternalSqlKind::Postgres, &entry.publication);
let slot_lit = pg_lit(&entry.slot);
let _ = self
.pg_capture_sql(
&su,
&image,
&format!(
"SELECT pg_drop_replication_slot({slot_lit}) WHERE EXISTS (SELECT 1 FROM \
pg_replication_slots WHERE slot_name = {slot_lit})"
),
stage_dir,
)
.await;
let _ = self
.pg_capture_sql(
&su,
&image,
&format!("DROP PUBLICATION IF EXISTS {pub_q}"),
stage_dir,
)
.await;
let _ = self
.pg_capture_sql(
&su,
&image,
&format!("DROP ROLE IF EXISTS {role_q}"),
stage_dir,
)
.await;
}
async fn try_pg_wal_delta(
&self,
project: &str,
db: &str,
sidecar: &PgCaptureSidecar,
cap_dir: &Path,
) -> Option<(Vec<u8>, boatramp_core::project_delta::SqlMemberCursor)> {
let cfg = self.databases.get(db)?;
if !Self::pg_capture_eligible(cfg) {
return None;
}
let entry = sidecar.dbs.get(db)?;
if entry.base_lsn.is_empty() {
tracing::info!(db = %db, "pg delta: base capture never completed (no pinned slot) — falling back to a full dump");
return None;
}
let current_fp = self.schema_fingerprint(project, db).await;
if current_fp != entry.schema_fingerprint {
tracing::info!(db = %db, "pg delta: schema fingerprint changed since the base — falling back to a full dump");
return None;
}
match self
.pg_drain_capture(project, db, cfg, entry, cap_dir)
.await
{
Ok(Some(bytes)) => Some((
bytes,
boatramp_core::project_delta::SqlMemberCursor::PgLsn {
lsn: entry.base_lsn.clone(),
},
)),
Ok(None) => None, Err(e) => {
tracing::warn!(error = %e, db = %db, "pg delta: WAL drain failed — falling back to a full dump");
None
}
}
}
async fn collect_pending_capture_names(&self) -> std::collections::HashSet<String> {
let mut wanted = std::collections::HashSet::new();
let Some(base) = self.resumable_base.as_ref() else {
return wanted;
};
let now = now_secs();
let mut stack = vec![base.clone()];
let mut budget = 4096usize; while let Some(dir) = stack.pop() {
if budget == 0 {
break;
}
budget -= 1;
if let Some(sidecar) = read_pg_capture_sidecar(&dir).await {
let file_age_secs = pg_capture_sidecar_age_secs(&dir, now).await;
for e in sidecar.dbs.values() {
if !capture_entry_shielded(
e.created_at,
file_age_secs,
now,
STALE_RESUMABLE_CAPTURE_TTL_SECS,
) {
continue; }
wanted.insert(e.slot.clone());
wanted.insert(e.role.clone());
wanted.insert(e.publication.clone());
}
}
if let Ok(mut rd) = tokio::fs::read_dir(&dir).await {
while let Ok(Some(ent)) = rd.next_entry().await {
if ent.file_type().await.map(|t| t.is_dir()).unwrap_or(false) {
stack.push(ent.path());
}
}
}
}
wanted
}
async fn pg_capture_base_slots(&self, project: &str, dir: &Path, stage_dir: &Path) {
if read_pg_capture_sidecar(dir).await.is_some() {
return; }
let mut sidecar = PgCaptureSidecar::default();
for (db, cfg) in &self.databases {
if !Self::pg_capture_eligible(cfg) {
continue;
}
match self.pg_begin_capture(project, db, cfg, stage_dir).await {
Ok(entry) => {
sidecar.dbs.insert(db.clone(), entry);
}
Err(e) => {
tracing::warn!(error = %e, db = %db, "pg delta capture: could not create a base slot; this db's delta will fall back to a full dump");
}
}
}
if !sidecar.dbs.is_empty()
&& let Ok(bytes) = serde_json::to_vec_pretty(&sidecar)
{
let _ = write_atomic(&dir.join(PG_CAPTURE_SIDECAR), &bytes).await;
}
}
async fn restore_pg_true_delta_in_chain(
&self,
project: &str,
chain: &[(PathBuf, project_bundle::StreamManifest)],
opts: &ImportOptions,
report: &mut ImportReport,
stage_dir: &Path,
) -> Result<(), Perr> {
struct DbPlan {
base: Option<(PathBuf, usize)>,
wal: Vec<(PathBuf, usize)>,
}
let mut per_db: BTreeMap<String, DbPlan> = BTreeMap::new();
for (dir, m) in chain {
for (idx, spec) in m.members.iter().enumerate() {
match spec.kind {
MemberKind::PgDump => {
if let Some(db) =
db_from_pg_member(&spec.path).filter(|d| self.is_managed_pg(d))
{
let e = per_db.entry(db.to_string()).or_insert(DbPlan {
base: None,
wal: Vec::new(),
});
e.base = Some((dir.clone(), idx));
e.wal.clear();
}
}
MemberKind::PgWalDelta => {
if let Some(db) = db_from_pgwal_member(&spec.path) {
if !self.is_managed_pg(db) {
return Err(Perr::Other(format!(
"pg true-delta restore: a PgWalDelta member {:?} names db {db:?} \
which is not a managed pg binding on this node — malformed bundle",
spec.path
)));
}
per_db
.entry(db.to_string())
.or_insert(DbPlan {
base: None,
wal: Vec::new(),
})
.wal
.push((dir.clone(), idx));
}
}
_ => {}
}
}
}
if per_db.is_empty() {
return Ok(());
}
if !opts.apply {
report.pg_databases += per_db.len();
return Ok(());
}
for (db, plan_db) in per_db {
let Some((base_dir, base_idx)) = plan_db.base else {
return Err(Perr::Other(format!(
"pg true-delta restore: db {db:?} has WAL delta member(s) but no full base dump \
in the chain — cannot replay onto nothing"
)));
};
let cfg = self
.databases
.get(&db)
.ok_or_else(|| {
Perr::Other(format!("import: no database named {db:?} on this node"))
})?
.clone();
let plan = self.pg_shadow_plan(project, &db, &cfg).await?;
if let Err(e) = self.pg_create_shadow(&plan, stage_dir).await {
self.pg_drop_shadow(&plan, stage_dir).await;
return Err(e);
}
if let Err(e) = self
.pg_assert_not_superuser(&plan.apply_conn(), &plan.image, stage_dir)
.await
{
self.pg_drop_shadow(&plan, stage_dir).await;
return Err(e);
}
let base_obj = base_dir.join(member_object_name(base_idx));
if let Err(e) = self
.pg_restore_base_into_shadow(&plan, &base_obj, stage_dir)
.await
{
self.pg_drop_shadow(&plan, stage_dir).await;
return Err(e);
}
for (dir, idx) in &plan_db.wal {
let framed = tokio::fs::read(dir.join(member_object_name(*idx)))
.await
.map_err(other)?;
if let Err(e) = self.pg_replay_wal(&plan, &framed, stage_dir).await {
self.pg_drop_shadow(&plan, stage_dir).await;
return Err(e);
}
}
if let Err(e) = self.pg_swap_shadow(&plan, stage_dir).await {
self.pg_drop_shadow(&plan, stage_dir).await;
return Err(e);
}
report.pg_databases += 1;
}
Ok(())
}
async fn pg_restore_base_into_shadow(
&self,
plan: &PgShadowPlan,
base_obj: &Path,
stage_dir: &Path,
) -> Result<(), Perr> {
let (tx, rx) = tokio::sync::mpsc::channel::<Vec<u8>>(16);
let out_path = stage_dir.join(format!("pgwal-base-restore-{}.out", shadow_suffix()));
let restore = self.spawn_pg_restore(plan, rx, out_path.clone())?;
let src = base_obj.to_path_buf();
let feeder = tokio::spawn(async move {
use tokio::io::AsyncReadExt;
let mut f = tokio::fs::File::open(&src).await?;
let mut buf = vec![0u8; 256 * 1024];
loop {
let n = f.read(&mut buf).await?;
if n == 0 {
break;
}
if tx.send(buf[..n].to_vec()).await.is_err() {
break;
}
}
Ok::<(), std::io::Error>(())
});
let code = restore.await;
let _ = feeder.await;
let out = tokio::fs::read_to_string(&out_path)
.await
.unwrap_or_default();
let _ = tokio::fs::remove_file(&out_path).await;
match code {
Ok(Ok(0)) => Ok(()),
Ok(Ok(c)) => Err(Perr::Other(format!(
"pg true-delta base restore into shadow {:?} exited {c}: {out}",
plan.shadow_db
))),
Ok(Err(e)) => Err(e),
Err(je) => Err(other(je)),
}
}
async fn pg_replay_wal(
&self,
plan: &PgShadowPlan,
framed: &[u8],
stage_dir: &Path,
) -> Result<(), Perr> {
let changes = boatramp_core::pgoutput::decode_framed(framed)
.map_err(|e| Perr::Other(format!("pg WAL delta: decode failed: {e}")))?;
let stmts = boatramp_core::pgoutput::render_changes(&changes).map_err(|e| {
Perr::Other(format!(
"pg WAL delta: cannot faithfully replay ({e}); such a delta must fall back to a \
full dump — refusing a partial restore"
))
})?;
if stmts.is_empty() {
return Ok(());
}
let mut script = String::from("SET standard_conforming_strings = on;\n");
for s in &stmts {
script.push_str(s);
script.push_str(";\n");
}
let script_path = stage_dir.join(format!("pgwal-replay-{}.sql", shadow_suffix()));
tokio::fs::write(&script_path, script.as_bytes())
.await
.map_err(other)?;
let apply = plan.apply_conn();
let argv = vec![
"psql".to_string(),
"-v".to_string(),
"ON_ERROR_STOP=1".to_string(),
];
let r = self
.run_sql_capture(
&plan.image,
&argv,
&apply.env(),
Some(&script_path),
stage_dir,
)
.await;
let _ = tokio::fs::remove_file(&script_path).await;
let (code, out) = r?;
if code != 0 {
return Err(Perr::Other(format!(
"pg WAL delta replay into shadow {:?} exited {code}: {out}",
plan.shadow_db
)));
}
Ok(())
}
}
fn mysql_lit(s: &str) -> String {
format!("'{}'", s.replace('\'', "''"))
}
const MYSQL_CAPTURE_SIDECAR: &str = "mysql_capture.json";
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct MysqlCaptureEntry {
base_gtid: String,
start_log: String,
#[serde(default)]
schema_fingerprint: Option<String>,
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
struct MysqlCaptureSidecar {
dbs: BTreeMap<String, MysqlCaptureEntry>,
}
async fn read_mysql_capture_sidecar(dir: &Path) -> Option<MysqlCaptureSidecar> {
let bytes = tokio::fs::read(dir.join(MYSQL_CAPTURE_SIDECAR))
.await
.ok()?;
serde_json::from_slice(&bytes).ok()
}
fn mysql_binlog_image(client_image: &str) -> String {
if client_image.ends_with("-debian") {
client_image.to_string()
} else {
format!("{client_image}-debian")
}
}
fn mysql_delta_shadow_db(live_db: &str) -> String {
format!("{live_db}__shadow_brdelta")
}
fn swap_would_destroy_live(live_tables: &[String], shadow_tables: &[String]) -> bool {
!live_tables.is_empty() && shadow_tables.is_empty()
}
fn mysql_line_is_sql_log_bin_toggle(line: &str) -> bool {
let t = line.trim_start();
t.starts_with("SET @MYSQLDUMP_TEMP_LOG_BIN") || t.starts_with("SET @@SESSION.SQL_LOG_BIN")
}
fn mysql_grant_privileges(line: &str) -> &str {
line.split(" ON ").next().unwrap_or(line)
}
fn parse_mysql_base_coord(header: &str) -> Option<(String, String)> {
let between = |hay: &str, open: &str| -> Option<String> {
let start = hay.find(open)? + open.len();
let rest = &hay[start..];
let end = rest.find('\'')?;
Some(rest[..end].to_string())
};
let gtid = between(header, "GTID_PURGED='")?;
let gtid = gtid.trim_start_matches('+').to_string();
let start_log = between(header, "MASTER_LOG_FILE='")?;
if gtid.is_empty() || start_log.is_empty() {
return None;
}
Some((gtid, start_log))
}
fn mysql_binlog_decoded_has_ddl(decoded: &str) -> bool {
for raw in decoded.lines() {
let line = raw.trim();
if line.is_empty()
|| line.starts_with('#') || line.starts_with("/*") || line.starts_with("DELIMITER")
|| line.starts_with("SET ") || line.starts_with("BINLOG ") || line == "'" || line.starts_with("'/*")
{
continue;
}
let up = line.to_ascii_uppercase();
if up.starts_with("BEGIN")
|| up.starts_with("COMMIT")
|| up.starts_with("ROLLBACK")
|| up.starts_with("START TRANSACTION")
|| up.starts_with("SAVEPOINT")
|| up.starts_with("RELEASE SAVEPOINT")
{
continue;
}
if line
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '+' || c == '/' || c == '=')
{
continue;
}
return true;
}
false
}
fn mysql_reader_conn(op: &MysqlConnection, user: &str, password: &str) -> MysqlConnection {
let mut c = op.clone();
c.user = user.to_string();
c.password = password.to_string();
c
}
impl NodeProjectExport {
fn mysql_capture_eligible(cfg: &ExternalDatabaseConfig) -> bool {
is_managed_binding(cfg)
&& kind_is_mysql(&cfg.kind)
&& cfg.delta_capture
&& !matches!(cfg.tenant, crate::config::TenantIsolation::Shared)
}
fn is_managed_mysql(&self, db: &str) -> bool {
self.databases
.get(db)
.map(|c| is_managed_binding(c) && kind_is_mysql(&c.kind))
.unwrap_or(false)
}
async fn mysql_admin_sql(
&self,
conn: &MysqlConnection,
image: &str,
sql: &str,
stage_dir: &Path,
) -> Result<(i32, String), Perr> {
let inv = mysql_admin_invocation(conn);
let in_path = stage_dir.join(format!("mysql-adm-{}.sql", shadow_suffix()));
tokio::fs::write(&in_path, sql).await.map_err(other)?;
let r = self
.run_sql_capture(image, &inv.argv, &inv.env, Some(&in_path), stage_dir)
.await;
let _ = tokio::fs::remove_file(&in_path).await;
r
}
async fn produce_mysql_capture_base(
&self,
project: &str,
dir: &Path,
idx: usize,
db: &str,
cfg: &ExternalDatabaseConfig,
stage_dir: &Path,
) -> Result<project_bundle::MemberMeta, Perr> {
let (conn, image, _ver) = self.resolve_mysql(project, db, cfg).await?;
let inv = mysqldump_capture_base_invocation(&conn);
let raw = stage_dir.join(format!("mysqlcap-raw-{idx}"));
let code = self
.run_sql_client(&image, &inv.argv, &inv.env, None, &raw)
.await?;
if code != 0 {
let _ = tokio::fs::remove_file(&raw).await;
return Err(Perr::Other(format!(
"mysql capture base dump for {db:?} exited {code}"
)));
}
let stripped = stage_dir.join(format!("mysqlcap-strip-{idx}"));
{
use tokio::io::{AsyncBufReadExt, AsyncWriteExt};
let rd = tokio::fs::File::open(&raw).await.map_err(other)?;
let mut lines = tokio::io::BufReader::new(rd).lines();
let mut wr = tokio::fs::File::create(&stripped).await.map_err(other)?;
while let Some(line) = lines.next_line().await.map_err(other)? {
if mysql_line_is_sql_log_bin_toggle(&line) {
continue;
}
wr.write_all(line.as_bytes()).await.map_err(other)?;
wr.write_all(b"\n").await.map_err(other)?;
}
wr.flush().await.map_err(other)?;
}
let _ = tokio::fs::remove_file(&raw).await;
let meta = write_member_from_file(dir, idx, &stripped).await?;
let _ = tokio::fs::remove_file(&stripped).await;
Ok(meta)
}
async fn mysql_capture_base_gtids(&self, project: &str, dir: &Path, specs: &[MemberSpec]) {
if read_mysql_capture_sidecar(dir).await.is_some() {
return; }
let mut sidecar = MysqlCaptureSidecar::default();
for (idx, spec) in specs.iter().enumerate() {
if spec.kind != MemberKind::MysqlDump {
continue;
}
let Some(db) = db_from_mysql_member(&spec.path) else {
continue;
};
let Some(cfg) = self.databases.get(db) else {
continue;
};
if !Self::mysql_capture_eligible(cfg) {
continue;
}
let obj = dir.join(member_object_name(idx));
let header = match read_file_prefix(&obj, 16 * 1024).await {
Ok(h) => h,
Err(e) => {
tracing::warn!(error = %e, db = %db, "mysql delta: could not read base member header; this db's delta will fall back to a full dump");
continue;
}
};
let Some((base_gtid, start_log)) = parse_mysql_base_coord(&header) else {
tracing::warn!(db = %db, "mysql delta: base dump carries no GTID/binlog coordinate (binlog/gtid not enabled?); this db's delta will fall back to a full dump");
continue;
};
let schema_fingerprint = self.schema_fingerprint(project, db).await;
sidecar.dbs.insert(
db.to_string(),
MysqlCaptureEntry {
base_gtid,
start_log,
schema_fingerprint,
},
);
}
if !sidecar.dbs.is_empty()
&& let Ok(bytes) = serde_json::to_vec_pretty(&sidecar)
{
let _ = write_atomic(&dir.join(MYSQL_CAPTURE_SIDECAR), &bytes).await;
}
}
async fn try_mysql_binlog_delta(
&self,
project: &str,
db: &str,
sidecar: &MysqlCaptureSidecar,
cap_dir: &Path,
) -> Option<(Vec<u8>, boatramp_core::project_delta::SqlMemberCursor)> {
let cfg = self.databases.get(db)?;
if !Self::mysql_capture_eligible(cfg) {
return None;
}
let entry = sidecar.dbs.get(db)?;
let current_fp = self.schema_fingerprint(project, db).await;
if current_fp != entry.schema_fingerprint {
tracing::info!(db = %db, "mysql delta: schema fingerprint changed since the base — falling back to a full dump");
return None;
}
match self
.mysql_drain_binlog(project, db, cfg, entry, cap_dir)
.await
{
Ok(Some(bytes)) => Some((
bytes,
boatramp_core::project_delta::SqlMemberCursor::MysqlGtid {
gtid: entry.base_gtid.clone(),
},
)),
Ok(None) => None, Err(e) => {
tracing::warn!(error = %e, db = %db, "mysql delta: binlog drain failed — falling back to a full dump");
None
}
}
}
async fn mysql_drain_binlog(
&self,
project: &str,
db: &str,
cfg: &ExternalDatabaseConfig,
entry: &MysqlCaptureEntry,
stage_dir: &Path,
) -> Result<Option<Vec<u8>>, Perr> {
let (op, image, _ver) = self.resolve_mysql(project, db, cfg).await?;
let binlog_image = mysql_binlog_image(&image);
let live_db = cfg.database.clone().unwrap_or_else(|| db.to_string());
let shadow_db = mysql_delta_shadow_db(&live_db);
let suffix = mysql_eph_suffix();
let reader = format!("brexp_rd_{suffix}");
let reader_pw = random_token();
let reader_lit = mysql_lit(&reader);
let create = format!(
"CREATE USER {reader_lit}@'%' IDENTIFIED BY {}; \
GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO {reader_lit}@'%';",
mysql_lit(&reader_pw)
);
let (code, out) = self
.mysql_admin_sql(&op, &image, &create, stage_dir)
.await?;
if code != 0 {
return Err(Perr::Other(format!(
"mysql capture: CREATE reader user for {db:?} exited {code}: {out}"
)));
}
let reader_conn = mysql_reader_conn(&op, &reader, &reader_pw);
let cap = mysqlbinlog_capture_invocation(
&reader_conn,
&live_db,
&shadow_db,
&entry.base_gtid,
&entry.start_log,
);
let cmd = format!("exec {} 2>/tmp/caperr", sh_join(&cap.argv));
let argv = vec!["sh".to_string(), "-c".to_string(), cmd];
let member_path = stage_dir.join(format!("mysqlbinlog-{suffix}.sql"));
let code = self
.run_sql_client(&binlog_image, &argv, &cap.env, None, &member_path)
.await;
let drop_sql = format!("DROP USER IF EXISTS {reader_lit}@'%';");
let _ = self
.mysql_admin_sql(&op, &image, &drop_sql, stage_dir)
.await;
let code = code?;
if code != 0 {
let _ = tokio::fs::remove_file(&member_path).await;
tracing::info!(db = %db, "mysql delta: binlog capture exited non-zero (start log purged, or server unreachable) — falling back to a full dump");
return Ok(None);
}
let member_len = tokio::fs::metadata(&member_path)
.await
.map_err(other)?
.len();
if member_len > MYSQL_DELTA_MEMBER_MAX {
let _ = tokio::fs::remove_file(&member_path).await;
tracing::info!(db = %db, member_len, "mysql delta: binlog delta exceeds the member cap — falling back to a full dump");
return Ok(None);
}
let bytes = tokio::fs::read(&member_path).await.map_err(other)?;
let _ = tokio::fs::remove_file(&member_path).await;
let decoded = String::from_utf8_lossy(&bytes);
if mysql_binlog_decoded_has_ddl(&decoded) {
tracing::info!(db = %db, "mysql delta: a DDL statement is present in the binlog range — falling back to a full dump");
return Ok(None);
}
Ok(Some(bytes))
}
}
impl NodeProjectExport {
async fn restore_mysql_true_delta_in_chain(
&self,
project: &str,
chain: &[(PathBuf, project_bundle::StreamManifest)],
opts: &ImportOptions,
report: &mut ImportReport,
stage_dir: &Path,
) -> Result<(), Perr> {
struct DbPlan {
base: Option<(PathBuf, usize)>,
binlog: Vec<(PathBuf, usize)>,
}
let mut per_db: BTreeMap<String, DbPlan> = BTreeMap::new();
for (dir, m) in chain {
for (idx, spec) in m.members.iter().enumerate() {
match spec.kind {
MemberKind::MysqlDump => {
if let Some(db) =
db_from_mysql_member(&spec.path).filter(|d| self.is_managed_mysql(d))
{
let e = per_db.entry(db.to_string()).or_insert(DbPlan {
base: None,
binlog: Vec::new(),
});
e.base = Some((dir.clone(), idx));
e.binlog.clear();
}
}
MemberKind::MysqlBinlogDelta => {
if let Some(db) = db_from_mysqlbinlog_member(&spec.path) {
if !self.is_managed_mysql(db) {
return Err(Perr::Other(format!(
"mysql true-delta restore: a MysqlBinlogDelta member {:?} names \
db {db:?} which is not a managed mysql binding on this node — \
malformed bundle",
spec.path
)));
}
per_db
.entry(db.to_string())
.or_insert(DbPlan {
base: None,
binlog: Vec::new(),
})
.binlog
.push((dir.clone(), idx));
}
}
_ => {}
}
}
}
if per_db.is_empty() {
return Ok(());
}
if !opts.apply {
report.pg_databases += per_db.len();
return Ok(());
}
for (db, plan_db) in per_db {
let Some((base_dir, base_idx)) = plan_db.base else {
return Err(Perr::Other(format!(
"mysql true-delta restore: db {db:?} has binlog delta member(s) but no full base \
dump in the chain — cannot replay onto nothing"
)));
};
let cfg = self
.databases
.get(&db)
.ok_or_else(|| {
Perr::Other(format!("import: no database named {db:?} on this node"))
})?
.clone();
let mut plan = self.mysql_shadow_plan(project, &db, &cfg).await?;
plan.shadow_db = mysql_delta_shadow_db(&plan.live_db);
if self
.mysql_db_exists(&plan, &plan.shadow_db, stage_dir)
.await?
{
return Err(Perr::Unavailable(format!(
"mysql true-delta restore: the delta shadow {:?} already exists — a concurrent \
import of db {db:?} is in flight (the binlog-delta shadow name is \
deterministic). Retry after it completes.",
plan.shadow_db
)));
}
self.mysql_create_shadow(&plan, stage_dir).await?;
if let Err(e) = self.mysql_grant_replay(&plan, stage_dir).await {
self.mysql_drop_shadow(&plan, stage_dir).await;
return Err(e);
}
if let Err(e) = self
.mysql_assert_replay_user_confined(&plan, stage_dir)
.await
{
self.mysql_drop_shadow(&plan, stage_dir).await;
return Err(e);
}
if let Err(e) = self
.mysql_restore_base_into_shadow(
&plan,
&base_dir.join(member_object_name(base_idx)),
stage_dir,
)
.await
{
self.mysql_drop_shadow(&plan, stage_dir).await;
return Err(e);
}
for (dir, idx) in &plan_db.binlog {
let member_path = dir.join(member_object_name(*idx));
let member_len = tokio::fs::metadata(&member_path)
.await
.map_err(other)?
.len();
if member_len > MYSQL_DELTA_MEMBER_MAX {
self.mysql_drop_shadow(&plan, stage_dir).await;
return Err(Perr::Other(format!(
"mysql true-delta restore REFUSED: binlog delta member {} is {member_len} \
bytes, over the {MYSQL_DELTA_MEMBER_MAX}-byte cap — refusing to read an \
oversized untrusted member into memory",
member_path.display()
)));
}
let member = tokio::fs::read(&member_path).await.map_err(other)?;
if let Err(e) = self.mysql_replay_binlog(&plan, &member, stage_dir).await {
self.mysql_drop_shadow(&plan, stage_dir).await;
return Err(e);
}
}
if let Err(e) = self.mysql_swap_shadow(&plan, stage_dir).await {
self.mysql_drop_shadow(&plan, stage_dir).await;
return Err(e);
}
report.pg_databases += 1;
}
Ok(())
}
async fn mysql_restore_base_into_shadow(
&self,
plan: &MysqlShadowPlan,
base_obj: &Path,
stage_dir: &Path,
) -> Result<(), Perr> {
let inv = mysql_shadow_invocation(&plan.apply_conn(), &plan.shadow_db);
let merged = merge_stderr_argv(&inv.argv);
let out = stage_dir.join(format!("mysqlbase-{}.out", shadow_suffix()));
let code = self
.run_sql_client(&plan.image, &merged, &inv.env, Some(base_obj), &out)
.await?;
let diag = tokio::fs::read_to_string(&out).await.unwrap_or_default();
let _ = tokio::fs::remove_file(&out).await;
if code != 0 {
return Err(Perr::Other(format!(
"mysql true-delta base restore into shadow {:?} exited {code}: {}",
plan.shadow_db,
diag.trim()
)));
}
Ok(())
}
async fn mysql_grant_replay(
&self,
plan: &MysqlShadowPlan,
stage_dir: &Path,
) -> Result<(), Perr> {
let user_lit = mysql_lit(&plan.eph_user);
let sql =
format!("GRANT REPLICATION_APPLIER, SESSION_VARIABLES_ADMIN ON *.* TO {user_lit}@'%';");
let (code, out) = self.mysql_run_admin(plan, &sql, stage_dir).await?;
if code != 0 {
return Err(Perr::Other(format!(
"mysql true-delta: GRANT replay privileges to the apply user exited {code}: {out}"
)));
}
Ok(())
}
async fn mysql_assert_replay_user_confined(
&self,
plan: &MysqlShadowPlan,
stage_dir: &Path,
) -> Result<(), Perr> {
let apply = plan.apply_conn();
let inv = mysql_admin_invocation(&apply);
let in_path = stage_dir.join(format!("mysql-replay-probe-{}.sql", shadow_suffix()));
tokio::fs::write(&in_path, "SHOW GRANTS FOR CURRENT_USER();")
.await
.map_err(other)?;
let r = self
.run_sql_capture(&plan.image, &inv.argv, &inv.env, Some(&in_path), stage_dir)
.await;
let _ = tokio::fs::remove_file(&in_path).await;
let (code, out) = r?;
if code != 0 {
return Err(Perr::Other(format!(
"mysql replay: could not probe the apply user's grants (exit {code}): {out}"
)));
}
for line in out.lines() {
let privs = mysql_grant_privileges(line).to_ascii_uppercase();
let up = line.to_ascii_uppercase();
if privs.contains("SUPER") || privs.contains("FILE") || privs.contains("BINLOG_ADMIN") {
return Err(Perr::Other(format!(
"mysql replay REFUSED: the apply user holds SUPER/FILE/BINLOG_ADMIN; an untrusted \
binlog must only be replayed by a least-privilege user. grants: {out}"
)));
}
if up.contains(" ON *.*") {
let ok = privs.contains("USAGE")
|| privs.contains("REPLICATION_APPLIER")
|| privs.contains("SESSION_VARIABLES_ADMIN");
let has_broad_dml = privs.contains("ALL PRIVILEGES")
|| privs.contains("INSERT")
|| privs.contains("UPDATE")
|| privs.contains("DELETE")
|| privs.contains("CREATE")
|| privs.contains("DROP");
if !ok || has_broad_dml {
return Err(Perr::Other(format!(
"mysql replay REFUSED: the apply user holds a global `*.*` privilege beyond \
the replay dynamics (REPLICATION_APPLIER / SESSION_VARIABLES_ADMIN); an \
untrusted binlog must stay confined to the shadow. grants: {out}"
)));
}
}
}
Ok(())
}
async fn mysql_replay_binlog(
&self,
plan: &MysqlShadowPlan,
member: &[u8],
stage_dir: &Path,
) -> Result<(), Perr> {
let apply = plan.apply_conn();
let inv = mysql_shadow_invocation(&apply, &plan.shadow_db);
let merged = merge_stderr_argv(&inv.argv);
let member_path = stage_dir.join(format!("mysqlreplay-{}.sql", shadow_suffix()));
tokio::fs::write(&member_path, member)
.await
.map_err(other)?;
let out = stage_dir.join(format!("mysqlreplay-{}.out", shadow_suffix()));
let code = self
.run_sql_client(&plan.image, &merged, &inv.env, Some(&member_path), &out)
.await;
let _ = tokio::fs::remove_file(&member_path).await;
let diag = tokio::fs::read_to_string(&out).await.unwrap_or_default();
let _ = tokio::fs::remove_file(&out).await;
let code = code?;
if code != 0 {
return Err(Perr::Other(format!(
"mysql binlog delta replay into shadow {:?} exited {code}: {}",
plan.shadow_db,
diag.trim()
)));
}
Ok(())
}
}
const LIBSQL_CAPTURE_SIDECAR: &str = "libsql_capture.json";
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct LibsqlCaptureEntry {
snapshot_file: String,
base_sha256: String,
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
struct LibsqlCaptureSidecar {
sites: BTreeMap<String, LibsqlCaptureEntry>,
}
async fn read_libsql_capture_sidecar(dir: &Path) -> Option<LibsqlCaptureSidecar> {
let bytes = tokio::fs::read(dir.join(LIBSQL_CAPTURE_SIDECAR))
.await
.ok()?;
serde_json::from_slice(&bytes).ok()
}
#[cfg(any(feature = "sql-delta-libsql", test))]
fn confine_bundle_snapshot(base_dir: &Path, snapshot_file: &str) -> Result<PathBuf, Perr> {
use std::path::Component;
let rel = Path::new(snapshot_file);
if rel.as_os_str().is_empty()
|| rel.is_absolute()
|| rel.components().any(|c| !matches!(c, Component::Normal(_)))
{
return Err(Perr::Other(format!(
"libsql delta: refusing out-of-tree base snapshot path {snapshot_file:?}"
)));
}
let joined = base_dir.join(rel);
let canon_base = std::fs::canonicalize(base_dir).map_err(|e| {
Perr::Other(format!(
"libsql delta: canonicalize base dir {}: {e}",
base_dir.display()
))
})?;
let canon = std::fs::canonicalize(&joined).map_err(|e| {
Perr::Other(format!(
"libsql delta: base snapshot {snapshot_file:?} not found/canonicalize: {e}"
))
})?;
if !canon.starts_with(&canon_base) {
return Err(Perr::Other(format!(
"libsql delta: base snapshot path {snapshot_file:?} escapes the bundle dir (symlink?)"
)));
}
Ok(canon)
}
impl NodeProjectExport {
#[cfg(feature = "sql-delta-libsql")]
async fn libsql_capture_base_snapshots(&self, project: &str, dir: &Path) {
let Some(libsql) = &self.libsql else {
return;
};
let sites = match self.deploy.list_sites(ProjectRef::new(project)).await {
Ok(s) => s,
Err(e) => {
tracing::warn!(%project, error = %e, "libsql delta capture: list_sites failed; skipping base snapshots");
return;
}
};
let mut sidecar = LibsqlCaptureSidecar::default();
for (i, site) in sites.into_iter().enumerate() {
let qsite = ProjectRef::new(project).qualified(&site);
if !libsql.local_database_exists(&qsite, "") {
continue; }
let rel = format!("libsql_capture/{i}.db");
let dest = dir.join(&rel);
match libsql.snapshot_delta_base(&qsite, "", &dest).await {
Ok(sha) => {
sidecar.sites.insert(
site,
LibsqlCaptureEntry {
snapshot_file: rel,
base_sha256: sha,
},
);
}
Err(e) => {
tracing::warn!(%project, %site, error = %e, "libsql delta capture: base snapshot failed; this site will full-dump at the next delta");
}
}
}
if !sidecar.sites.is_empty()
&& let Ok(bytes) = serde_json::to_vec_pretty(&sidecar)
{
let _ = write_atomic(&dir.join(LIBSQL_CAPTURE_SIDECAR), &bytes).await;
}
}
async fn try_libsql_changeset_delta(
&self,
project: &str,
site: &str,
entry: &LibsqlCaptureEntry,
base_dir: &Path,
cap_dir: &Path,
) -> Option<(Vec<u8>, boatramp_core::project_delta::SqlMemberCursor)> {
let _ = (project, site, entry, base_dir, cap_dir);
#[cfg(feature = "sql-delta-libsql")]
{
use boatramp_storage::sql_libsql_changeset::ChangesetCapture;
let libsql = self.libsql.as_ref()?;
let qsite = ProjectRef::new(project).qualified(site);
if !libsql.local_database_exists(&qsite, "") {
return None; }
let base_snapshot = match confine_bundle_snapshot(base_dir, &entry.snapshot_file) {
Ok(p) => p,
Err(e) => {
tracing::warn!(%project, %site, error = %e, "libsql delta: base snapshot path rejected; full-dump fallback");
return None;
}
};
match libsql
.capture_delta_changeset(&qsite, "", &base_snapshot, &entry.base_sha256, cap_dir)
.await
{
Ok(ChangesetCapture::Changeset(cs)) => {
let framed = boatramp_core::project_delta::encode_libsql_changeset(&cs);
let cursor =
boatramp_core::project_delta::SqlMemberCursor::LibsqlChangesetBase {
base_sha256: entry.base_sha256.clone(),
};
Some((framed, cursor))
}
Ok(ChangesetCapture::Fallback(reason)) => {
tracing::info!(%project, %site, %reason, "libsql delta: full-dump fallback");
None
}
Err(e) => {
tracing::warn!(%project, %site, error = %e, "libsql delta: changeset capture failed; full-dump fallback");
None
}
}
}
#[cfg(not(feature = "sql-delta-libsql"))]
{
None
}
}
async fn restore_libsql_true_delta_in_chain(
&self,
project: &str,
chain: &[(PathBuf, project_bundle::StreamManifest)],
changeset_sites: &std::collections::BTreeSet<String>,
opts: &ImportOptions,
report: &mut ImportReport,
) -> Result<(), Perr> {
if changeset_sites.is_empty() {
return Ok(());
}
struct SitePlan {
base_dir: Option<PathBuf>,
changesets: Vec<(PathBuf, usize)>,
}
let mut per_site: BTreeMap<String, SitePlan> = BTreeMap::new();
for (dir, m) in chain {
for (idx, spec) in m.members.iter().enumerate() {
match spec.kind {
MemberKind::Libsql => {
if let Some(site) = site_from_libsql_member(&spec.path)
.filter(|s| changeset_sites.contains(*s))
{
let e = per_site.entry(site.to_string()).or_insert(SitePlan {
base_dir: None,
changesets: Vec::new(),
});
e.base_dir = Some(dir.clone());
e.changesets.clear();
}
}
MemberKind::LibsqlChangeset => {
if let Some(site) = site_from_libsql_changeset_member(&spec.path) {
per_site
.entry(site.to_string())
.or_insert(SitePlan {
base_dir: None,
changesets: Vec::new(),
})
.changesets
.push((dir.clone(), idx));
}
}
_ => {}
}
}
}
per_site.retain(|_, p| !p.changesets.is_empty());
if per_site.is_empty() {
return Ok(());
}
#[cfg(not(feature = "sql-delta-libsql"))]
{
let _ = (project, opts, report);
Err(Perr::Unavailable(
"bundle carries libsql changeset true-delta members but this build lacks the libsql \
changeset engine (feature `sql-delta-libsql`) — refusing to restore (never a silent \
stale-base full-dump)"
.into(),
))
}
#[cfg(feature = "sql-delta-libsql")]
{
let Some(libsql) = self.libsql.clone() else {
return Err(Perr::Unavailable(
"bundle carries libsql changeset members but this node has no local libsql \
backend to restore them into"
.into(),
));
};
if !opts.apply {
report.libsql_databases += per_site.len();
return Ok(());
}
for (site, plan) in per_site {
let Some(base_dir) = plan.base_dir else {
return Err(Perr::Other(format!(
"libsql true-delta restore: site {site:?} has changeset member(s) but no full \
base libsql member in the chain — cannot replay onto nothing"
)));
};
let sidecar = read_libsql_capture_sidecar(&base_dir).await.ok_or_else(|| {
Perr::Other(format!(
"libsql true-delta restore: base dir {} has no {LIBSQL_CAPTURE_SIDECAR} — \
cannot resolve the base snapshot for site {site:?}",
base_dir.display()
))
})?;
let entry = sidecar.sites.get(&site).ok_or_else(|| {
Perr::Other(format!(
"libsql true-delta restore: no base snapshot recorded for site {site:?} in \
{LIBSQL_CAPTURE_SIDECAR}"
))
})?;
let base_snapshot = confine_bundle_snapshot(&base_dir, &entry.snapshot_file)?;
let mut changesets: Vec<Vec<u8>> = Vec::with_capacity(plan.changesets.len());
for (dir, idx) in &plan.changesets {
let obj = dir.join(member_object_name(*idx));
let bytes = tokio::fs::read(&obj).await.map_err(other)?;
let cs = boatramp_core::project_delta::decode_libsql_changeset(&bytes)
.map_err(|e| {
Perr::Other(format!(
"libsql true-delta restore: decode changeset member {}: {e}",
obj.display()
))
})?;
changesets.push(cs);
}
let qsite = ProjectRef::new(project).qualified(&site);
libsql
.restore_delta_changeset_chain(
&qsite,
"",
&base_snapshot,
&entry.base_sha256,
changesets,
opts.overwrite,
)
.await
.map_err(other)?;
report.libsql_databases += 1;
}
Ok(())
}
}
}
async fn read_file_prefix(path: &Path, max: usize) -> Result<String, Perr> {
use tokio::io::AsyncReadExt;
let mut f = tokio::fs::File::open(path).await.map_err(other)?;
let mut buf = vec![0u8; max];
let n = f.read(&mut buf).await.map_err(other)?;
buf.truncate(n);
Ok(String::from_utf8_lossy(&buf).into_owned())
}
fn member_object_name(idx: usize) -> String {
format!("m{idx:05}")
}
fn member_meta_name(idx: usize) -> String {
format!("m{idx:05}.meta")
}
fn member_part_name(idx: usize) -> String {
format!("m{idx:05}.part")
}
async fn write_atomic(path: &Path, bytes: &[u8]) -> Result<(), Perr> {
let tmp = path.with_extension("writing.tmp");
tokio::fs::write(&tmp, bytes).await.map_err(other)?;
tokio::fs::rename(&tmp, path).await.map_err(other)?;
Ok(())
}
async fn sha256_and_size_of_file(path: &Path) -> Result<(u64, String), Perr> {
use tokio::io::AsyncReadExt;
let mut f = tokio::fs::File::open(path).await.map_err(other)?;
let mut hasher = Sha256::new();
let mut size = 0u64;
let mut buf = vec![0u8; 64 * 1024];
loop {
let n = f.read(&mut buf).await.map_err(other)?;
if n == 0 {
break;
}
hasher.update(&buf[..n]);
size += n as u64;
}
Ok((size, hex::encode(hasher.finalize())))
}
async fn member_object_valid(obj: &Path, meta: &Path) -> Result<bool, Perr> {
if !tokio::fs::try_exists(obj).await.map_err(other)?
|| !tokio::fs::try_exists(meta).await.map_err(other)?
{
return Ok(false);
}
let meta_bytes = tokio::fs::read(meta).await.map_err(other)?;
let recorded: project_bundle::MemberMeta = match serde_json::from_slice(&meta_bytes) {
Ok(m) => m,
Err(_) => return Ok(false),
};
let (size, sha) = sha256_and_size_of_file(obj).await?;
Ok(size == recorded.size && sha == recorded.sha256)
}
async fn write_member_from_file(
dir: &Path,
idx: usize,
src: &Path,
) -> Result<project_bundle::MemberMeta, Perr> {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let part = dir.join(member_part_name(idx));
let mut rd = tokio::fs::File::open(src).await.map_err(other)?;
let mut wr = tokio::fs::File::create(&part).await.map_err(other)?;
let mut hasher = Sha256::new();
let mut size = 0u64;
let mut buf = vec![0u8; 64 * 1024];
loop {
let n = rd.read(&mut buf).await.map_err(other)?;
if n == 0 {
break;
}
wr.write_all(&buf[..n]).await.map_err(other)?;
hasher.update(&buf[..n]);
size += n as u64;
}
wr.flush().await.map_err(other)?;
drop(wr);
tokio::fs::rename(&part, dir.join(member_object_name(idx)))
.await
.map_err(other)?;
Ok(project_bundle::MemberMeta {
size,
sha256: hex::encode(hasher.finalize()),
})
}
async fn write_member_from_producer(
dir: &Path,
idx: usize,
producer: ChunkedProducer,
) -> Result<project_bundle::MemberMeta, Perr> {
use std::sync::atomic::Ordering::SeqCst;
use tokio::io::AsyncWriteExt;
let (mut rx, poison, handle) = producer;
let part = dir.join(member_part_name(idx));
let mut wr = tokio::fs::File::create(&part).await.map_err(other)?;
let mut hasher = Sha256::new();
let mut size = 0u64;
while let Some(chunk) = rx.recv().await {
wr.write_all(&chunk).await.map_err(other)?;
hasher.update(&chunk);
size += chunk.len() as u64;
}
wr.flush().await.map_err(other)?;
drop(wr);
match handle.await {
Ok(Ok(())) => {}
Ok(Err(e)) => return Err(e),
Err(e) => return Err(Perr::Other(format!("export member producer panicked: {e}"))),
}
if poison.load(SeqCst) {
return Err(Perr::Other(format!(
"export member {idx} producer poisoned (incomplete body) — not committing"
)));
}
tokio::fs::rename(&part, dir.join(member_object_name(idx)))
.await
.map_err(other)?;
Ok(project_bundle::MemberMeta {
size,
sha256: hex::encode(hasher.finalize()),
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn create_replication_slot_export_sql_is_version_aware_and_quoted() {
let pg16 = create_replication_slot_export_sql("brexp_slot_x", "160004");
assert_eq!(
pg16,
"CREATE_REPLICATION_SLOT \"brexp_slot_x\" LOGICAL pgoutput (SNAPSHOT 'export');"
);
let pg17 = create_replication_slot_export_sql("brexp_slot_x", "170000");
assert!(pg17.contains("(SNAPSHOT 'export')"));
let pg14 = create_replication_slot_export_sql("brexp_slot_x", "140010");
assert!(
pg14.ends_with("LOGICAL pgoutput EXPORT_SNAPSHOT;"),
"pre-15 must use the positional keyword: {pg14}"
);
let bad = create_replication_slot_export_sql("brexp_slot_x", "garbage");
assert!(bad.ends_with("EXPORT_SNAPSHOT;"));
let weird = create_replication_slot_export_sql("s\"lot", "160004");
assert!(
weird.contains("\"s\"\"lot\""),
"slot name must be quoted: {weird}"
);
}
#[test]
fn capture_entry_shielded_applies_the_stale_resumable_ttl() {
let ttl = STALE_RESUMABLE_CAPTURE_TTL_SECS;
let now = 1_000_000u64;
assert!(capture_entry_shielded(now - 10, 0, now, ttl));
assert!(capture_entry_shielded(now - ttl, 0, now, ttl)); assert!(!capture_entry_shielded(now - ttl - 1, 0, now, ttl)); assert!(capture_entry_shielded(0, ttl - 1, now, ttl)); assert!(!capture_entry_shielded(0, ttl + 1, now, ttl)); assert!(capture_entry_shielded(now + 5, 0, now, ttl));
}
async fn export_to_buf(
drv: &NodeProjectExport,
project: &str,
opts: &ExportOptions,
) -> Result<Vec<u8>, Perr> {
let mut sink = futures::io::Cursor::new(Vec::<u8>::new());
drv.export(project, opts, &mut sink).await?;
Ok(sink.into_inner())
}
fn read_members(bytes: &[u8]) -> (bool, Vec<(MemberSpec, Vec<u8>)>) {
let mut data_only = false;
let mut got: Vec<(MemberSpec, Vec<u8>)> = Vec::new();
project_bundle::read_bundle_streaming_with_head(
bytes,
|d, _s| data_only = d,
|spec, _sz, body| {
let mut v = Vec::new();
body.read_to_end(&mut v)
.map_err(project_bundle::BundleError::Io)?;
got.push((spec.clone(), v));
Ok(())
},
)
.expect("read back the exported v2 bundle");
(data_only, got)
}
async fn import_buf(
drv: &NodeProjectExport,
project: &str,
bytes: &[u8],
opts: ImportOptions,
) -> Result<ImportReport, Perr> {
let mut src = futures::io::Cursor::new(bytes.to_vec());
drv.import(project, &mut src, opts).await
}
#[test]
fn kind_detection() {
assert!(kind_is_postgres("postgres"));
assert!(kind_is_postgres("PostgreSQL"));
assert!(kind_is_postgres("pg"));
assert!(!kind_is_postgres("libsql"));
assert!(kind_is_libsql("libsql"));
assert!(kind_is_libsql("sqlite3"));
assert!(!kind_is_libsql("mysql"));
assert!(kind_is_mysql("mysql"));
assert!(kind_is_mysql("MariaDB"));
assert!(!kind_is_mysql("postgres"));
}
#[test]
fn client_image_override_is_used_verbatim_else_computed() {
use boatramp_core::Storage;
use boatramp_core::deploy::DeployStore;
use boatramp_core::kv::{KvStore, MemoryKv};
use boatramp_storage::fs::FsStorage;
use std::collections::BTreeMap;
use std::sync::Arc;
let tmp = tempfile::tempdir().unwrap();
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let blobs: Arc<dyn Storage> = Arc::new(FsStorage::new(tmp.path().join("blobs")));
let deploy = DeployStore::new(blobs.clone(), kv.clone());
let drv =
NodeProjectExport::new(BTreeMap::new(), deploy, kv, blobs, tmp.path().join("work"));
assert_eq!(
drv.pg_image("160004").unwrap(),
boatramp_core::compute::pg_client_image_for_server_version("160004").unwrap()
);
assert!(drv.pg_image("not-a-number").is_err());
let pinned = drv.with_client_images(
Some("pgvector/pgvector:pg16@sha256:dead".into()),
Some("mysql:8.4@sha256:beef".into()),
);
assert_eq!(
pinned.pg_image("not-a-number").unwrap(),
"pgvector/pgvector:pg16@sha256:dead"
);
assert_eq!(
pinned.mysql_image("not-a-version").unwrap(),
"mysql:8.4@sha256:beef"
);
}
#[test]
fn mysql_member_path_round_trips() {
assert_eq!(mysql_member_path("shop"), "sql/shop.sql");
assert_eq!(db_from_mysql_member("sql/shop.sql"), Some("shop"));
assert_eq!(db_from_mysql_member("sql/app.pgc"), None);
}
#[test]
fn parse_sql_url_handles_pg_and_mysql_dsns() {
assert_eq!(
parse_sql_url("postgres://u:p@h:5432/db"),
Some(("u".into(), "p".into(), "h".into(), 5432, "db".into()))
);
assert_eq!(
parse_sql_url("mysql://svc:pw@10.0.0.6:3306/shop?ssl=true"),
Some((
"svc".into(),
"pw".into(),
"10.0.0.6".into(),
3306,
"shop".into()
))
);
assert_eq!(
parse_sql_url("postgres://u:p@h/db").map(|t| t.3),
Some(5432)
);
assert_eq!(parse_sql_url("mysql://u:p@h/db").map(|t| t.3), Some(3306));
assert_eq!(
parse_sql_url("mysql://u:a:b@h:3306/db").map(|t| t.1),
Some("a:b".into())
);
assert_eq!(parse_sql_url("redis://h/0"), None);
assert_eq!(parse_sql_url("not-a-url"), None);
}
#[test]
fn pg_member_path_round_trips() {
assert_eq!(pg_member_path("app"), "sql/app.pgc");
assert_eq!(db_from_pg_member("sql/app.pgc"), Some("app"));
assert_eq!(db_from_pg_member("libsql/x.sqlite"), None);
}
#[test]
fn mysqlbinlog_member_path_round_trips() {
assert_eq!(mysqlbinlog_member_path("app"), "sql/app.mysqlbinlog");
assert_eq!(
db_from_mysqlbinlog_member("sql/app.mysqlbinlog"),
Some("app")
);
assert_eq!(db_from_mysqlbinlog_member("sql/app.sql"), None);
assert_eq!(db_from_mysql_member("sql/app.mysqlbinlog"), None);
}
#[test]
fn mysql_binlog_image_appends_debian_once() {
assert_eq!(mysql_binlog_image("mysql:8.0"), "mysql:8.0-debian");
assert_eq!(mysql_binlog_image("mysql:8.4"), "mysql:8.4-debian");
assert_eq!(mysql_binlog_image("mysql:8.0-debian"), "mysql:8.0-debian");
}
#[test]
fn mysql_sql_log_bin_toggle_detection() {
assert!(mysql_line_is_sql_log_bin_toggle(
"SET @MYSQLDUMP_TEMP_LOG_BIN = @@SESSION.SQL_LOG_BIN;"
));
assert!(mysql_line_is_sql_log_bin_toggle(
"SET @@SESSION.SQL_LOG_BIN= 0;"
));
assert!(mysql_line_is_sql_log_bin_toggle(
"SET @@SESSION.SQL_LOG_BIN = @MYSQLDUMP_TEMP_LOG_BIN;"
));
assert!(!mysql_line_is_sql_log_bin_toggle(
"INSERT INTO `t` VALUES (1,'a');"
));
assert!(!mysql_line_is_sql_log_bin_toggle(
"/* SET @@GLOBAL.GTID_PURGED='+uuid:1-9';*/"
));
}
#[test]
fn parse_mysql_base_coord_extracts_gtid_and_log_file() {
let header = "-- Server version\tn8.0.46\n\
SET @@SESSION.SQL_LOG_BIN= 0;\n\
/* SET @@GLOBAL.GTID_PURGED='+142694f1-d0bc-11f1:1-9';*/\n\
-- CHANGE MASTER TO MASTER_LOG_FILE='node-bin.000003', MASTER_LOG_POS=1068;\n\
DROP TABLE IF EXISTS `t`;\n";
let (gtid, log) = parse_mysql_base_coord(header).expect("coord parses");
assert_eq!(gtid, "142694f1-d0bc-11f1:1-9"); assert_eq!(log, "node-bin.000003");
assert_eq!(parse_mysql_base_coord("DROP TABLE IF EXISTS `t`;"), None);
}
#[test]
fn mysql_grant_privileges_splits_before_the_target_db() {
let g = |l: &str| mysql_grant_privileges(l).to_ascii_uppercase();
assert_eq!(
g("GRANT ALL PRIVILEGES ON `supervisors__shadow_ab`.* TO 'x'@'%'"),
"GRANT ALL PRIVILEGES"
);
assert!(!g("GRANT ALL PRIVILEGES ON `supervisors`.* TO 'x'@'%'").contains("SUPER"));
assert!(!g("GRANT ALL PRIVILEGES ON `filestore`.* TO 'x'@'%'").contains("FILE"));
assert!(!g("GRANT ALL PRIVILEGES ON `profiles`.* TO 'x'@'%'").contains("FILE"));
assert!(g("GRANT SUPER ON *.* TO 'x'@'%'").contains("SUPER"));
assert!(g("GRANT FILE ON *.* TO 'x'@'%'").contains("FILE"));
assert!(
g("GRANT REPLICATION_APPLIER, SESSION_VARIABLES_ADMIN ON *.* TO 'x'@'%'")
.contains("REPLICATION_APPLIER")
);
let dyn_privs = g("GRANT REPLICATION_APPLIER, SESSION_VARIABLES_ADMIN ON *.* TO 'x'@'%'");
assert!(
!dyn_privs.contains("SUPER")
&& !dyn_privs.contains("FILE")
&& !dyn_privs.contains("BINLOG_ADMIN")
);
}
#[test]
fn mysql_delta_shadow_db_is_deterministic_and_sweepable() {
assert_eq!(mysql_delta_shadow_db("app"), "app__shadow_brdelta");
assert!(mysql_delta_shadow_db("tenant_x").contains("__shadow_"));
}
#[test]
fn swap_safety_refuses_empty_shadow_over_nonempty_live() {
let s = |xs: &[&str]| xs.iter().map(|x| x.to_string()).collect::<Vec<_>>();
assert!(swap_would_destroy_live(&s(&["t", "victim"]), &s(&[])));
assert!(!swap_would_destroy_live(&s(&["t"]), &s(&["t"])));
assert!(!swap_would_destroy_live(&s(&[]), &s(&["t"])));
assert!(!swap_would_destroy_live(&s(&[]), &s(&[])));
assert!(!swap_would_destroy_live(&s(&["t", "victim"]), &s(&["t"])));
}
#[test]
fn mysql_binlog_ddl_scan_distinguishes_row_dml_from_ddl() {
let row_only = "/*!50530 SET @@SESSION.PSEUDO_SLAVE_MODE=1*/;\n\
# at 197\n\
SET @@session.sql_mode=1168113696/*!*/;\n\
BEGIN\n\
/*!*/;\n\
BINLOG '\n\
jnneag8BAAAAegAAAH4AAAA=\n\
'/*!*/;\n\
COMMIT\n\
/*!*/;\n";
assert!(
!mysql_binlog_decoded_has_ddl(row_only),
"a pure ROW/DML delta carries no DDL"
);
let with_ddl = format!("{row_only}ALTER TABLE `t` ADD COLUMN `c` int\n/*!*/;\n");
assert!(
mysql_binlog_decoded_has_ddl(&with_ddl),
"an ALTER in the range must be detected as DDL"
);
let with_create = format!("{row_only}CREATE TABLE `u` (id int)\n/*!*/;\n");
assert!(mysql_binlog_decoded_has_ddl(&with_create));
}
#[test]
fn check_project_slug_rejects_traversal() {
assert!(check_project_slug("acme").is_ok());
assert!(check_project_slug("preview-b").is_ok());
assert!(check_project_slug("../etc").is_err());
assert!(check_project_slug("a/b").is_err());
assert!(check_project_slug("").is_err());
}
#[test]
fn data_only_skew_fails_closed_on_a_missing_fingerprint() {
assert!(!data_only_skew_blocks(Some("abc"), Some("abc")));
assert!(data_only_skew_blocks(Some("abc"), Some("xyz")));
assert!(data_only_skew_blocks(None, Some("abc")));
assert!(data_only_skew_blocks(Some("abc"), None));
assert!(data_only_skew_blocks(None, None));
}
#[tokio::test]
async fn data_only_export_omits_the_control_plane_kv_slice() {
use boatramp_core::Storage;
use boatramp_core::deploy::DeployStore;
use boatramp_core::kv::{KvStore, MemoryKv};
use boatramp_storage::fs::FsStorage;
use std::collections::BTreeMap;
use std::sync::Arc;
let tmp = tempfile::tempdir().unwrap();
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
kv.put("project/acme/config/tenancy", b"{}".to_vec())
.await
.unwrap();
let blobs: Arc<dyn Storage> = Arc::new(FsStorage::new(tmp.path().join("blobs")));
let deploy = DeployStore::new(blobs.clone(), kv.clone());
let work = tmp.path().join("work");
std::fs::create_dir_all(&work).unwrap();
let drv = NodeProjectExport::new(BTreeMap::new(), deploy, kv.clone(), blobs.clone(), work);
let full = export_to_buf(
&drv,
"acme",
&ExportOptions {
data_only: false,
..Default::default()
},
)
.await
.expect("full export");
let (full_data_only, full_members) = read_members(&full);
assert!(!full_data_only, "full manifest is not data-only");
assert!(
full_members.iter().any(|(m, _)| m.kind == MemberKind::Kv),
"a FULL bundle must carry the control-plane KV slice"
);
let donly = export_to_buf(
&drv,
"acme",
&ExportOptions {
data_only: true,
..Default::default()
},
)
.await
.expect("data-only export");
let (donly_data_only, donly_members) = read_members(&donly);
assert!(donly_data_only, "data-only manifest flags data_only");
assert!(
!donly_members.iter().any(|(m, _)| m.kind == MemberKind::Kv),
"a data-only bundle must NOT carry the control-plane KV slice (Ask-4a leak)"
);
}
#[tokio::test]
async fn include_identity_wraps_the_kek_and_refusals_hold() {
use boatramp_core::Storage;
use boatramp_core::deploy::DeployStore;
use boatramp_core::kv::{KvStore, MemoryKv};
use boatramp_storage::fs::FsStorage;
use std::collections::BTreeMap;
use std::sync::Arc;
let tmp = tempfile::tempdir().unwrap();
let kek = [0x5Au8; 32];
let kek_file = tmp.path().join("kek");
std::fs::write(&kek_file, kek).unwrap();
let mk = |suffix: &str, src: Option<IdentityEnvelope>| {
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let blobs: Arc<dyn Storage> =
Arc::new(FsStorage::new(tmp.path().join(format!("b{suffix}"))));
let deploy = DeployStore::new(blobs.clone(), kv.clone());
let work = tmp.path().join(format!("w{suffix}"));
std::fs::create_dir_all(&work).unwrap();
let mut d = NodeProjectExport::new(BTreeMap::new(), deploy, kv, blobs, work);
if let Some(s) = src {
d = d.with_identity_source(s);
}
d
};
let pass = b"operator-transport-passphrase".to_vec();
let drv = mk(
"local",
Some(IdentityEnvelope::Local {
kek_file: kek_file.clone(),
}),
);
let bundle = export_to_buf(
&drv,
"acme",
&ExportOptions {
data_only: false,
include_identity: true,
transport_key: Some(pass.clone()),
delta: None,
},
)
.await
.expect("include-identity export");
let (_do, members) = read_members(&bundle);
let (idspec, container) = members
.iter()
.find(|(m, _)| m.kind == MemberKind::Identity)
.expect("identity member present");
assert_eq!(
idspec.path, IDENTITY_MEMBER_PATH,
"identity member at the declared path"
);
let payload =
boatramp_server::envelope::transport_unwrap(container, &pass).expect("unwrap");
assert_eq!(
parse_identity_payload(&payload).unwrap(),
kek,
"the wrapped KEK round-trips"
);
assert!(
boatramp_server::envelope::transport_unwrap(container, b"wrong").is_err(),
"wrong transport key cannot open the container"
);
assert_eq!(
members.first().map(|(m, _)| m.kind),
Some(MemberKind::Identity),
"identity is the first member in restore order"
);
let plain = export_to_buf(&drv, "acme", &ExportOptions::default())
.await
.expect("default export");
let (_do2, m2) = read_members(&plain);
assert!(
!m2.iter().any(|(m, _)| m.kind == MemberKind::Identity),
"default bundle carries no identity member"
);
assert!(
export_to_buf(
&drv,
"acme",
&ExportOptions {
data_only: true,
include_identity: true,
transport_key: Some(pass.clone()),
delta: None,
},
)
.await
.is_err(),
"data-only + include-identity is refused"
);
assert!(
export_to_buf(
&drv,
"acme",
&ExportOptions {
data_only: false,
include_identity: true,
transport_key: None,
delta: None,
},
)
.await
.is_err(),
"include-identity without a transport key is refused"
);
let vault = mk("vault", Some(IdentityEnvelope::Vault));
assert!(
export_to_buf(
&vault,
"acme",
&ExportOptions {
data_only: false,
include_identity: true,
transport_key: Some(pass.clone()),
delta: None,
},
)
.await
.is_err(),
"include-identity on a Vault envelope is refused"
);
}
#[tokio::test]
async fn adopt_installs_kek_and_fresh_drops_source_secrets() {
use boatramp_core::Storage;
use boatramp_core::deploy::DeployStore;
use boatramp_core::kv::{KvStore, MemoryKv};
use boatramp_storage::fs::FsStorage;
use std::collections::BTreeMap;
use std::sync::Arc;
let tmp = tempfile::tempdir().unwrap();
let kek = [0x3Cu8; 32];
let src_kek_file = tmp.path().join("src-kek");
std::fs::write(&src_kek_file, kek).unwrap();
let src_kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
for (k, v) in [
("project/acme/secret/api-key", &b"sealed"[..]),
("project/acme/tenant-secret/t1/token", &b"sealed-t"[..]),
("project/acme/email/smtp", &b"sealed-pw"[..]), ("managed-sql-cred/acme/pg", &b"sealed-cred"[..]), ("_cp/leader", &b"node-1"[..]),
("project/acme/config/tenancy", &b"{}"[..]),
] {
src_kv.put(k, v.to_vec()).await.unwrap();
}
let src_blobs: Arc<dyn Storage> = Arc::new(FsStorage::new(tmp.path().join("src-blobs")));
let src = NodeProjectExport::new(
BTreeMap::new(),
DeployStore::new(src_blobs.clone(), src_kv.clone()),
src_kv.clone(),
src_blobs,
{
let w = tmp.path().join("src-work");
std::fs::create_dir_all(&w).unwrap();
w
},
)
.with_identity_source(IdentityEnvelope::Local {
kek_file: src_kek_file.clone(),
});
let pass = b"transport-pass".to_vec();
let bundle = export_to_buf(
&src,
"acme",
&ExportOptions {
data_only: false,
include_identity: true,
transport_key: Some(pass.clone()),
delta: None,
},
)
.await
.expect("export with identity");
let target = |tag: &str| {
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let blobs: Arc<dyn Storage> =
Arc::new(FsStorage::new(tmp.path().join(format!("{tag}-blobs"))));
let work = tmp.path().join(format!("{tag}-work"));
std::fs::create_dir_all(&work).unwrap();
let kek_file = tmp.path().join(format!("{tag}-kek"));
let drv = NodeProjectExport::new(
BTreeMap::new(),
DeployStore::new(blobs.clone(), kv.clone()),
kv.clone(),
blobs,
work,
)
.with_identity_source(IdentityEnvelope::Local {
kek_file: kek_file.clone(),
});
(drv, kv, kek_file)
};
let (adopt_drv, a_kv, a_kek) = target("a");
import_buf(
&adopt_drv,
"acme",
&bundle,
ImportOptions {
apply: true,
identity: IdentityChoice::Adopt {
transport_key: pass.clone(),
},
..Default::default()
},
)
.await
.expect("adopt import");
assert_eq!(
std::fs::read(&a_kek).unwrap(),
kek,
"adopt installed the source KEK"
);
assert!(
a_kv.get("project/acme/secret/api-key")
.await
.unwrap()
.is_some(),
"adopt keeps the sealed secret (it unseals under the adopted KEK)"
);
assert!(
a_kv.get("project/acme/email/smtp").await.unwrap().is_some()
&& a_kv
.get("managed-sql-cred/acme/pg")
.await
.unwrap()
.is_some(),
"adopt keeps the sealed email profile + managed-DB credential"
);
assert!(
a_kv.get("_cp/leader").await.unwrap().is_none(),
"import never writes a reserved _cp/* key (Ask-4b)"
);
let (fresh_drv, f_kv, f_kek) = target("f");
import_buf(
&fresh_drv,
"acme",
&bundle,
ImportOptions {
apply: true,
identity: IdentityChoice::Fresh,
..Default::default()
},
)
.await
.expect("fresh import");
assert!(!f_kek.exists(), "fresh installs no KEK");
assert!(
f_kv.get("project/acme/secret/api-key")
.await
.unwrap()
.is_none(),
"fresh dropped the source sealed secret (Ask-4c)"
);
assert!(
f_kv.get("project/acme/tenant-secret/t1/token")
.await
.unwrap()
.is_none(),
"fresh dropped the source per-tenant secret (Ask-4c)"
);
assert!(
f_kv.get("project/acme/email/smtp").await.unwrap().is_none(),
"fresh dropped the source SMTP profile's sealed password (H1)"
);
assert!(
f_kv.get("managed-sql-cred/acme/pg")
.await
.unwrap()
.is_none(),
"fresh dropped the source managed-DB credential (H1)"
);
assert!(
f_kv.get("project/acme/config/tenancy")
.await
.unwrap()
.is_some(),
"fresh keeps non-secret, environment-neutral control-plane config (tenancy map)"
);
assert!(
f_kv.get("_cp/leader").await.unwrap().is_none(),
"import never writes a reserved _cp/* key (Ask-4b)"
);
let plain = export_to_buf(&src, "acme", &ExportOptions::default())
.await
.expect("plain export");
let (no_id_drv, no_id_kv, _f) = target("n");
assert!(
import_buf(
&no_id_drv,
"acme",
&plain,
ImportOptions {
apply: true,
identity: IdentityChoice::Adopt {
transport_key: pass.clone(),
},
..Default::default()
},
)
.await
.is_err(),
"adopt on a bundle with no identity member is refused"
);
assert!(
no_id_kv
.get("project/acme/secret/api-key")
.await
.unwrap()
.is_none(),
"a refused adopt must not have written the source-sealed KV slice"
);
}
#[tokio::test]
async fn large_blob_round_trips_through_the_chunked_streaming_path() {
use boatramp_core::Storage;
use boatramp_core::deploy::DeployStore;
use boatramp_core::kv::{KvStore, MemoryKv};
use boatramp_storage::fs::FsStorage;
use std::collections::BTreeMap;
use std::sync::Arc;
let tmp = tempfile::tempdir().unwrap();
let mk = |tag: &str| {
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let blobs: Arc<dyn Storage> =
Arc::new(FsStorage::new(tmp.path().join(format!("{tag}-blobs"))));
let deploy = DeployStore::new(blobs.clone(), kv.clone());
let work = tmp.path().join(format!("{tag}-work"));
std::fs::create_dir_all(&work).unwrap();
let drv = NodeProjectExport::new(BTreeMap::new(), deploy, kv, blobs.clone(), work)
.with_chunk_size(16);
(drv, blobs)
};
let (src, src_blobs) = mk("src");
let big: Vec<u8> = (0u32..5000).map(|i| (i % 251) as u8).collect();
let small = b"tiny".to_vec();
let put = |blobs: Arc<dyn Storage>, key: &'static str, bytes: Vec<u8>| async move {
let body: boatramp_core::ByteStream =
futures::stream::once(async move { Ok(bytes::Bytes::from(bytes)) }).boxed();
blobs
.put(key, body, boatramp_core::PutMeta { content_type: None })
.await
.unwrap();
};
put(
src_blobs.clone(),
"hblob/acme/web/assets/big.bin",
big.clone(),
)
.await;
put(
src_blobs.clone(),
"hblob/acme/web/assets/small.txt",
small.clone(),
)
.await;
let bundle = export_to_buf(
&src,
"acme",
&ExportOptions {
data_only: true,
..Default::default()
},
)
.await
.expect("export");
let manifest: project_bundle::StreamManifest = {
use std::io::Read as _;
let mut ar = tar::Archive::new(std::io::Cursor::new(&bundle));
let mut first = ar
.entries()
.unwrap()
.next()
.expect("manifest entry")
.unwrap();
let mut json = String::new();
first.read_to_string(&mut json).unwrap();
serde_json::from_str(&json).expect("parse the v3 StreamManifest")
};
assert_eq!(
manifest.format_version,
project_bundle::BUNDLE_FORMAT_VERSION_V3,
"a bundle carrying a chunked member is v3"
);
let framing = |path: &str| {
manifest
.members
.iter()
.find(|m| m.path == path)
.map(|m| m.framing)
};
assert_eq!(
framing("blobs/hblob/acme/web/assets/big.bin"),
Some(project_bundle::MemberFraming::ChunkedTrailer),
"a blob larger than the chunk size must take the chunked streaming framing"
);
assert_eq!(
framing("blobs/hblob/acme/web/assets/small.txt"),
Some(project_bundle::MemberFraming::Sidecar),
"a small blob stays sidecar"
);
let (tgt, tgt_blobs) = mk("tgt");
let report = import_buf(
&tgt,
"acme2",
&bundle,
ImportOptions {
apply: true,
..Default::default()
},
)
.await
.expect("import");
assert!(
report.blobs >= 2,
"both blobs imported (got {})",
report.blobs
);
let got = tgt_blobs
.get("hblob/acme2/web/assets/big.bin")
.await
.expect("the large blob landed under the re-keyed target scope");
let mut body = got.body;
let mut buf = Vec::new();
while let Some(chunk) = body.next().await {
buf.extend_from_slice(chunk.unwrap().as_ref());
}
assert_eq!(
buf, big,
"the large blob is byte-identical after a chunked export+import round-trip"
);
}
#[test]
fn blob_rekey_lands_in_target_scope_and_fails_closed() {
assert_eq!(
target_blob_key("blobs/hblob/acme/web/c/obj", "acme", "acme").unwrap(),
"hblob/acme/web/c/obj"
);
let t = target_blob_key("blobs/hblob/preview-a/web/c/obj", "preview-a", "preview-b")
.expect("rename re-key");
assert_eq!(t, "hblob/preview-b/web/c/obj");
assert!(
!t.contains("preview-a"),
"NO source-scope segment survives a rename re-key"
);
assert!(target_blob_key("blobs/hblob/globex/secret/obj", "acme", "beta").is_err());
assert!(target_blob_key("blobs/hblob/acmecorp/x", "acme", "beta").is_err());
assert!(target_blob_key("hblob/acme/x", "acme", "beta").is_err());
assert!(target_blob_key("blobs/not-hblob/x", "acme", "beta").is_err());
}
#[test]
fn libsql_member_path_round_trips_the_site() {
assert_eq!(
libsql_member_path("acme", "blog"),
"libsql/acme/blog/default.sqlite"
);
assert_eq!(
site_from_libsql_member("libsql/acme/blog/default.sqlite"),
Some("blog")
);
assert_eq!(
site_from_libsql_member("libsql/default/www/default.sqlite"),
Some("www")
);
assert_eq!(site_from_libsql_member("sql/app.pgc"), None);
assert_eq!(site_from_libsql_member("libsql/bad"), None);
assert_eq!(
libsql_changeset_member_path("acme", "blog"),
"libsql/acme/blog/default.changeset"
);
assert_eq!(
site_from_libsql_changeset_member("libsql/acme/blog/default.changeset"),
Some("blog")
);
assert_eq!(
site_from_libsql_changeset_member("libsql/acme/blog/default.sqlite"),
None
);
}
#[test]
fn confine_bundle_snapshot_refuses_out_of_tree_paths() {
let base = tempfile::tempdir().unwrap();
let cap = base.path().join("libsql_capture");
std::fs::create_dir_all(&cap).unwrap();
std::fs::write(cap.join("0.db"), b"x").unwrap();
let ok = confine_bundle_snapshot(base.path(), "libsql_capture/0.db").expect("in-tree ok");
assert!(ok.starts_with(std::fs::canonicalize(base.path()).unwrap()));
for bad in [
"/dev/zero",
"/etc/passwd",
"../../../etc/passwd",
"libsql_capture/../../escape",
"./libsql_capture/0.db", "",
] {
assert!(
confine_bundle_snapshot(base.path(), bad).is_err(),
"an out-of-tree snapshot path must be refused: {bad:?}"
);
}
}
#[test]
fn blob_prefix_is_two_segment_for_every_project() {
assert_eq!(blob_project_prefix("acme"), "hblob/acme/");
assert_eq!(blob_project_prefix("default"), "hblob/default/");
assert!(!blob_project_prefix("foobar").starts_with(&blob_project_prefix("foo")));
}
#[test]
fn choose_host_prefers_bridge_only_under_published() {
assert_eq!(
choose_host("127.0.0.1", Some("172.17.0.3"), true),
"172.17.0.3"
);
assert_eq!(choose_host("127.0.0.1", None, true), "127.0.0.1");
assert_eq!(
choose_host("db.example.com", Some("172.17.0.3"), false),
"db.example.com"
);
}
#[test]
fn fingerprint_is_stable_and_order_sensitive() {
use boatramp_core::sql::SqlRows;
let a = SqlRows {
columns: vec!["id".into()],
rows: vec![
vec![SqlValue::Text("m1".into()), SqlValue::Integer(1)],
vec![SqlValue::Text("m2".into()), SqlValue::Integer(2)],
],
};
let same = SqlRows {
columns: vec!["id".into()],
rows: vec![
vec![SqlValue::Text("m1".into()), SqlValue::Integer(1)],
vec![SqlValue::Text("m2".into()), SqlValue::Integer(2)],
],
};
let different = SqlRows {
columns: vec!["id".into()],
rows: vec![vec![SqlValue::Text("m1".into()), SqlValue::Integer(1)]],
};
assert_eq!(fingerprint_rows(&a), fingerprint_rows(&same));
assert_ne!(fingerprint_rows(&a), fingerprint_rows(&different));
}
#[tokio::test]
async fn resumable_member_object_write_and_resume_decision() {
let tmp = tempfile::tempdir().unwrap();
let dir = tmp.path();
let src = dir.join("staged-src");
tokio::fs::write(&src, b"hello-member-zero").await.unwrap();
let meta = write_member_from_file(dir, 0, &src).await.unwrap();
assert_eq!(meta.size, b"hello-member-zero".len() as u64);
write_atomic(
&dir.join(member_meta_name(0)),
&serde_json::to_vec(&meta).unwrap(),
)
.await
.unwrap();
let obj0 = dir.join(member_object_name(0));
let meta0 = dir.join(member_meta_name(0));
assert!(member_object_valid(&obj0, &meta0).await.unwrap());
assert!(!dir.join(member_part_name(0)).exists());
assert!(
!member_object_valid(
&dir.join(member_object_name(1)),
&dir.join(member_meta_name(1))
)
.await
.unwrap()
);
tokio::fs::write(&obj0, b"trunc").await.unwrap();
assert!(
!member_object_valid(&obj0, &meta0).await.unwrap(),
"a truncated object must fail its recorded SHA and be re-produced on resume"
);
}
#[tokio::test]
async fn resumable_chunked_member_commits_and_poison_fails_loud() {
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering::SeqCst};
let tmp = tempfile::tempdir().unwrap();
let dir = tmp.path();
let (tx, rx) = tokio::sync::mpsc::channel::<Vec<u8>>(4);
let poison = Arc::new(AtomicBool::new(false));
let h = tokio::spawn(async move {
tx.send(b"ab".to_vec()).await.unwrap();
tx.send(b"cd".to_vec()).await.unwrap();
Ok(())
});
let meta = write_member_from_producer(dir, 2, (rx, poison, h))
.await
.unwrap();
assert_eq!(meta.size, 4);
assert!(dir.join(member_object_name(2)).exists());
let (tx, rx) = tokio::sync::mpsc::channel::<Vec<u8>>(4);
let poison = Arc::new(AtomicBool::new(false));
let p2 = poison.clone();
let h = tokio::spawn(async move {
tx.send(b"partial".to_vec()).await.unwrap();
p2.store(true, SeqCst);
Ok(())
});
let r = write_member_from_producer(dir, 3, (rx, poison, h)).await;
assert!(
r.is_err(),
"a poisoned (incomplete) producer must fail loud"
);
assert!(
!dir.join(member_object_name(3)).exists(),
"no member object may be committed when the producer poisoned"
);
}
#[tokio::test]
async fn resumable_dir_is_confined_under_the_configured_base() {
use boatramp_core::Storage;
use boatramp_core::deploy::DeployStore;
use boatramp_core::kv::{KvStore, MemoryKv};
use boatramp_storage::fs::FsStorage;
use std::collections::BTreeMap;
use std::sync::Arc;
let tmp = tempfile::tempdir().unwrap();
let mk = |base: Option<PathBuf>| {
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let blobs: Arc<dyn Storage> = Arc::new(FsStorage::new(tmp.path().join("blobs")));
let deploy = DeployStore::new(blobs.clone(), kv.clone());
NodeProjectExport::new(BTreeMap::new(), deploy, kv, blobs, tmp.path().join("work"))
.with_resumable_base(base)
};
assert!(
mk(None).resolve_resumable_dir("ok", true).await.is_err(),
"no configured base must refuse the resumable surface"
);
let base = tmp.path().join("exports");
let drv = mk(Some(base.clone()));
let resolved = drv
.resolve_resumable_dir("snap1", true)
.await
.expect("a plain name is confined + created");
let canon_base = tokio::fs::canonicalize(&base).await.unwrap();
assert!(
resolved.starts_with(&canon_base),
"resolved {resolved:?} must be under the base {canon_base:?}"
);
for bad in ["../escape", "a/../../etc", "/etc/passwd", "", "./x"] {
assert!(
drv.resolve_resumable_dir(bad, true).await.is_err(),
"confinement must refuse {bad:?}"
);
}
let outside = tmp.path().join("outside");
tokio::fs::create_dir_all(&outside).await.unwrap();
tokio::fs::create_dir_all(&canon_base).await.unwrap();
#[cfg(unix)]
{
let link = base.join("sneaky");
let _ = std::os::unix::fs::symlink(&outside, &link);
assert!(
drv.resolve_resumable_dir("sneaky", false).await.is_err(),
"a symlink escaping the base must be refused"
);
}
}
#[cfg(test)]
fn mk_delta_driver(
tmp: &std::path::Path,
shared_base: &std::path::Path,
tag: &str,
) -> (
NodeProjectExport,
std::sync::Arc<dyn boatramp_core::kv::KvStore>,
std::sync::Arc<dyn boatramp_core::Storage>,
) {
use boatramp_core::deploy::DeployStore;
use boatramp_core::kv::{KvStore, MemoryKv};
use boatramp_storage::fs::FsStorage;
use std::collections::BTreeMap;
use std::sync::Arc;
let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
let blobs: Arc<dyn boatramp_core::Storage> =
Arc::new(FsStorage::new(tmp.join(format!("{tag}-blobs"))));
let deploy = DeployStore::new(blobs.clone(), kv.clone());
let work = tmp.join(format!("{tag}-work"));
std::fs::create_dir_all(&work).unwrap();
let drv = NodeProjectExport::new(BTreeMap::new(), deploy, kv.clone(), blobs.clone(), work)
.with_resumable_base(Some(shared_base.to_path_buf()));
(drv, kv, blobs)
}
#[cfg(test)]
async fn seed_blob(store: &dyn boatramp_core::Storage, key: &str, data: &[u8]) {
use boatramp_core::{ByteStream, PutMeta, StorageError};
use bytes::Bytes;
let data = data.to_vec();
let stream: ByteStream =
futures::stream::once(async move { Ok::<Bytes, StorageError>(Bytes::from(data)) })
.boxed();
store
.put(key, stream, PutMeta { content_type: None })
.await
.expect("seed blob");
}
#[cfg(test)]
async fn read_blob(store: &dyn boatramp_core::Storage, key: &str) -> Option<Vec<u8>> {
match store.get(key).await {
Ok(obj) => {
let mut body = obj.body;
let mut buf = Vec::new();
while let Some(chunk) = body.next().await {
buf.extend_from_slice(chunk.ok()?.as_ref());
}
Some(buf)
}
Err(_) => None,
}
}
#[tokio::test]
async fn delta_differential_round_trip_kv_and_blobs() {
use boatramp_core::project_delta::{DeltaMode, DeltaRequest};
use boatramp_core::project_export::ProjectExport;
let tmp = tempfile::tempdir().unwrap();
let shared_base = tmp.path().join("exports");
let (src, kv_s, blobs_s) = mk_delta_driver(tmp.path(), &shared_base, "src");
kv_s.put("project/acme/config/a", b"one".to_vec())
.await
.unwrap();
kv_s.put("project/acme/config/b", b"two".to_vec())
.await
.unwrap();
seed_blob(blobs_s.as_ref(), "hblob/acme/web/x", b"X-v1").await;
seed_blob(blobs_s.as_ref(), "hblob/acme/web/y", b"Y-v1").await;
src.export_resumable("acme", &ExportOptions::default(), "base")
.await
.expect("full base export");
kv_s.put("project/acme/config/b", b"TWO".to_vec())
.await
.unwrap();
kv_s.put("project/acme/config/c", b"new".to_vec())
.await
.unwrap();
kv_s.delete("project/acme/config/a").await.unwrap();
seed_blob(blobs_s.as_ref(), "hblob/acme/web/x", b"X-v2-longer").await;
seed_blob(blobs_s.as_ref(), "hblob/acme/web/z", b"Z-v1").await;
blobs_s.delete("hblob/acme/web/y").await.unwrap();
let delta_opts = ExportOptions {
delta: Some(DeltaRequest {
mode: DeltaMode::Differential,
base_reference: "base".to_string(),
}),
..Default::default()
};
src.export_resumable("acme", &delta_opts, "d1")
.await
.expect("differential delta export");
let d1_dir = src.resolve_resumable_dir("d1", false).await.unwrap();
let m = src.read_resumable_manifest(&d1_dir).await.unwrap();
let di = m.delta.as_ref().expect("delta manifest carries DeltaInfo");
assert_eq!(di.mode, DeltaMode::Differential);
assert_eq!(di.base_reference, "base");
assert_eq!(
di.sql_strategy,
boatramp_core::project_delta::SqlDeltaKind::FullDumpFallback
);
let kinds: Vec<MemberKind> = m.members.iter().map(|s| s.kind).collect();
assert!(
kinds.contains(&MemberKind::KvDelta),
"a delta carries a KV-delta member"
);
assert!(
kinds.contains(&MemberKind::BlobTombstone),
"a delta with a deleted blob carries a blob-tombstone member"
);
assert!(
!kinds.contains(&MemberKind::Kv),
"a delta carries NO full KV member"
);
let blob_adds: Vec<&str> = m
.members
.iter()
.filter(|s| s.kind == MemberKind::Blob)
.map(|s| s.path.as_str())
.collect();
assert_eq!(
blob_adds.len(),
2,
"exactly the changed (x) + new (z) blobs ship, got {blob_adds:?}"
);
assert!(
blob_adds
.iter()
.all(|p| p.contains("/x") || p.contains("/z"))
);
let (tgt, kv_t, blobs_t) = mk_delta_driver(tmp.path(), &shared_base, "tgt");
let report = tgt
.import_resumable(
"acme",
"d1",
ImportOptions {
apply: true,
..Default::default()
},
)
.await
.expect("delta chain restore");
assert!(
report.notes.iter().any(|n| n.contains("delta restore")),
"{:?}",
report.notes
);
assert_eq!(
kv_t.get("project/acme/config/a").await.unwrap(),
None,
"a was tombstoned"
);
assert_eq!(
kv_t.get("project/acme/config/b").await.unwrap().as_deref(),
Some(b"TWO".as_slice())
);
assert_eq!(
kv_t.get("project/acme/config/c").await.unwrap().as_deref(),
Some(b"new".as_slice())
);
assert_eq!(
read_blob(blobs_t.as_ref(), "hblob/acme/web/x")
.await
.as_deref(),
Some(b"X-v2-longer".as_slice())
);
assert_eq!(
read_blob(blobs_t.as_ref(), "hblob/acme/web/z")
.await
.as_deref(),
Some(b"Z-v1".as_slice())
);
assert_eq!(
read_blob(blobs_t.as_ref(), "hblob/acme/web/y").await,
None,
"y was tombstoned"
);
}
#[tokio::test]
async fn delta_incremental_chain_round_trip() {
use boatramp_core::project_delta::{DeltaMode, DeltaRequest};
use boatramp_core::project_export::ProjectExport;
let tmp = tempfile::tempdir().unwrap();
let shared_base = tmp.path().join("exports");
let (src, kv_s, blobs_s) = mk_delta_driver(tmp.path(), &shared_base, "src");
kv_s.put("project/acme/config/k1", b"1".to_vec())
.await
.unwrap();
kv_s.put("project/acme/config/k2", b"2".to_vec())
.await
.unwrap();
seed_blob(blobs_s.as_ref(), "hblob/acme/web/p", b"P1").await;
src.export_resumable("acme", &ExportOptions::default(), "base")
.await
.expect("base");
kv_s.put("project/acme/config/k1", b"1b".to_vec())
.await
.unwrap();
seed_blob(blobs_s.as_ref(), "hblob/acme/web/q", b"Q1").await;
let inc = |base: &str| ExportOptions {
delta: Some(DeltaRequest {
mode: DeltaMode::Incremental,
base_reference: base.to_string(),
}),
..Default::default()
};
src.export_resumable("acme", &inc("base"), "d1")
.await
.expect("d1");
kv_s.delete("project/acme/config/k2").await.unwrap();
kv_s.put("project/acme/config/k3", b"3".to_vec())
.await
.unwrap();
blobs_s.delete("hblob/acme/web/p").await.unwrap();
src.export_resumable("acme", &inc("d1"), "d2")
.await
.expect("d2");
let d2_dir = src.resolve_resumable_dir("d2", false).await.unwrap();
let m2 = src.read_resumable_manifest(&d2_dir).await.unwrap();
assert_eq!(m2.delta.as_ref().unwrap().base_reference, "d1");
assert_eq!(m2.delta.as_ref().unwrap().mode, DeltaMode::Incremental);
let (tgt, kv_t, blobs_t) = mk_delta_driver(tmp.path(), &shared_base, "tgt");
tgt.import_resumable(
"acme",
"d2",
ImportOptions {
apply: true,
..Default::default()
},
)
.await
.expect("incremental chain restore");
assert_eq!(
kv_t.get("project/acme/config/k1").await.unwrap().as_deref(),
Some(b"1b".as_slice())
);
assert_eq!(
kv_t.get("project/acme/config/k2").await.unwrap(),
None,
"k2 tombstoned in d2"
);
assert_eq!(
kv_t.get("project/acme/config/k3").await.unwrap().as_deref(),
Some(b"3".as_slice())
);
assert_eq!(
read_blob(blobs_t.as_ref(), "hblob/acme/web/p").await,
None,
"p tombstoned in d2"
);
assert_eq!(
read_blob(blobs_t.as_ref(), "hblob/acme/web/q")
.await
.as_deref(),
Some(b"Q1".as_slice())
);
}
#[tokio::test]
async fn delta_import_refuses_a_chain_with_a_foreign_base_source() {
use boatramp_core::project_delta::{DeltaMode, DeltaRequest};
use boatramp_core::project_export::ProjectExport;
let tmp = tempfile::tempdir().unwrap();
let shared_base = tmp.path().join("exports");
let (src, kv_s, blobs_s) = mk_delta_driver(tmp.path(), &shared_base, "src");
kv_s.put("project/acme/config/k1", b"1".to_vec())
.await
.unwrap();
seed_blob(blobs_s.as_ref(), "hblob/acme/web/p", b"P1").await;
src.export_resumable("acme", &ExportOptions::default(), "base")
.await
.expect("base");
kv_s.put("project/acme/config/k1", b"1b".to_vec())
.await
.unwrap();
src.export_resumable(
"acme",
&ExportOptions {
delta: Some(DeltaRequest {
mode: DeltaMode::Incremental,
base_reference: "base".into(),
}),
..Default::default()
},
"d1",
)
.await
.expect("d1");
let base_dir = src.resolve_resumable_dir("base", false).await.unwrap();
let mpath = base_dir.join(project_bundle::MANIFEST_PATH);
let mut m: project_bundle::StreamManifest =
serde_json::from_slice(&tokio::fs::read(&mpath).await.unwrap()).unwrap();
m.source.project = "evil".into();
tokio::fs::write(&mpath, serde_json::to_vec_pretty(&m).unwrap())
.await
.unwrap();
let (tgt, _kv_t, _blobs_t) = mk_delta_driver(tmp.path(), &shared_base, "tgt");
let err = tgt
.import_resumable(
"acme",
"d1",
ImportOptions {
apply: true,
..Default::default()
},
)
.await
.expect_err("a foreign-base chain must be refused");
let msg = format!("{err}");
assert!(
msg.contains("must share one source"),
"expected a one-source coherence refusal, got: {msg}"
);
}
#[tokio::test]
async fn delta_restore_refuses_a_cross_tenant_kv_tombstone() {
use boatramp_core::kv::KvDumpEntry;
use boatramp_core::project_delta::{self as pd, KvDelta};
let tmp = tempfile::tempdir().unwrap();
let shared_base = tmp.path().join("exports");
let (drv, kv, _blobs) = mk_delta_driver(tmp.path(), &shared_base, "t");
kv.put("project/globex/secret/x", b"victim".to_vec())
.await
.unwrap();
let hostile = KvDelta {
upserts: vec![KvDumpEntry {
key: "project/acme/config/ok".to_string(),
value: b"v".to_vec(),
version: 1,
}],
tombstones: vec!["project/globex/secret/x".to_string()],
};
let temp = tmp.path().join("hostile.brkvdelta");
tokio::fs::write(&temp, pd::encode_kv_delta(&hostile, 0))
.await
.unwrap();
let spec = project_bundle::MemberSpec::sidecar(
project_bundle::KV_DELTA_MEMBER_PATH.to_string(),
MemberKind::KvDelta,
None,
);
let mut report = ImportReport::default();
let mut saw = false;
let r = drv
.restore_member(
"acme",
false,
"acme",
&spec,
&temp,
&ImportOptions {
apply: true,
..Default::default()
},
&mut report,
&mut saw,
)
.await;
assert!(r.is_err(), "a cross-tenant tombstone must be refused");
assert!(
format!("{}", r.unwrap_err()).contains("outside the target project"),
"the refusal names the cross-tenant scope violation"
);
assert_eq!(
kv.get("project/globex/secret/x").await.unwrap().as_deref(),
Some(b"victim".as_slice()),
"a refused delta deletes NOTHING"
);
}
}