#![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, mysqldump_invocation, mysqldump_tables_only_invocation,
pg_dump_invocation, pg_restore_invocation, pg_restore_shadow_invocation, 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;
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 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 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 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 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>,
}
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(),
}
}
#[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_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 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 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 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(" ")
}
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,
},
Mysql {
conn: MysqlConnection,
image: String,
data_only: bool,
},
Libsql { qsite: String },
Blob { key: String },
}
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)
}
}
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,
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,
},
));
} 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, .. }) => {
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 = pg_dump_invocation(&conn, false);
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,
} => {
let inv = pg_dump_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!("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)?;
}
}
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 devnull = temp.with_extension("restore.out");
let code = self
.run_sql_client(&image, &inv.argv, &inv.env, Some(temp), &devnull)
.await?;
let _ = tokio::fs::remove_file(&devnull).await;
if code != 0 {
return Err(Perr::Other(format!("pg_restore for {db:?} exited {code}")));
}
}
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 devnull = temp.with_extension("myrestore.out");
let code = self
.run_sql_client(&image, &inv.argv, &inv.env, Some(temp), &devnull)
.await?;
let _ = tokio::fs::remove_file(&devnull).await;
if code != 0 {
return Err(Perr::Other(format!(
"mysql restore for {db:?} exited {code}"
)));
}
}
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;
}
}
}
Ok(())
}
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);
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 = shadow_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 up = line.to_ascii_uppercase();
let dangerous_global = up.contains("ON *.*") && !up.contains("USAGE ON *.*");
if up.contains("SUPER") || up.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_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?;
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;
}
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;
}
}
}
}
} 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\\_%';";
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
}
}
fn mysql_lit(s: &str) -> String {
format!("'{}'", s.replace('\'', "''"))
}
#[cfg(test)]
mod tests {
use super::*;
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 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()),
},
)
.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()),
},
)
.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,
},
)
.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()),
},
)
.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()),
},
)
.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"
);
}
#[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);
}
#[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));
}
}