feat: implement row-based JSON fetching for ciphers to handle large vaults and avoid SQLITE_TOOBIG errors
This commit is contained in:
parent
197b9c7dd0
commit
edd8fcbd76
4 changed files with 104 additions and 11 deletions
|
|
@ -232,12 +232,14 @@ pub async fn list_ciphers(
|
||||||
) -> Result<RawJson, AppError> {
|
) -> Result<RawJson, AppError> {
|
||||||
let db = db::get_db(&env)?;
|
let db = db::get_db(&env)?;
|
||||||
let include_attachments = attachments::attachments_enabled(env.as_ref());
|
let include_attachments = attachments::attachments_enabled(env.as_ref());
|
||||||
|
let force_row_query = super::ciphers_default_row_query(env.as_ref());
|
||||||
let ciphers_json = fetch_cipher_json_array_raw(
|
let ciphers_json = fetch_cipher_json_array_raw(
|
||||||
&db,
|
&db,
|
||||||
include_attachments,
|
include_attachments,
|
||||||
"WHERE c.user_id = ?1 AND c.deleted_at IS NULL",
|
"WHERE c.user_id = ?1 AND c.deleted_at IS NULL",
|
||||||
&[claims.sub.clone().into()],
|
&[claims.sub.clone().into()],
|
||||||
"ORDER BY c.updated_at DESC",
|
"ORDER BY c.updated_at DESC",
|
||||||
|
force_row_query,
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
|
|
@ -532,12 +534,14 @@ pub async fn restore_ciphers_bulk(
|
||||||
.map_err(db::map_d1_json_error)?;
|
.map_err(db::map_d1_json_error)?;
|
||||||
|
|
||||||
let include_attachments = attachments::attachments_enabled(env.as_ref());
|
let include_attachments = attachments::attachments_enabled(env.as_ref());
|
||||||
|
let force_row_query = super::ciphers_default_row_query(env.as_ref());
|
||||||
let ciphers_json = fetch_cipher_json_array_raw(
|
let ciphers_json = fetch_cipher_json_array_raw(
|
||||||
&db,
|
&db,
|
||||||
include_attachments,
|
include_attachments,
|
||||||
"WHERE c.user_id = ?1 AND c.id IN (SELECT value FROM json_each(?2, '$.ids'))",
|
"WHERE c.user_id = ?1 AND c.id IN (SELECT value FROM json_each(?2, '$.ids'))",
|
||||||
&[claims.sub.clone().into(), body.clone().into()],
|
&[claims.sub.clone().into(), body.clone().into()],
|
||||||
"",
|
"",
|
||||||
|
force_row_query,
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
|
|
@ -734,6 +738,11 @@ struct CipherJsonArrayRow {
|
||||||
ciphers_json: String,
|
ciphers_json: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Deserialize)]
|
||||||
|
struct CipherJsonRow {
|
||||||
|
cipher_json: String,
|
||||||
|
}
|
||||||
|
|
||||||
/// Build the SQL expression for a single cipher as JSON.
|
/// Build the SQL expression for a single cipher as JSON.
|
||||||
fn cipher_json_expr(attachments_enabled: bool) -> String {
|
fn cipher_json_expr(attachments_enabled: bool) -> String {
|
||||||
let attachments_expr = if attachments_enabled {
|
let attachments_expr = if attachments_enabled {
|
||||||
|
|
@ -820,6 +829,28 @@ fn cipher_json_array_sql(
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn cipher_json_rows_sql(
|
||||||
|
attachments_enabled: bool,
|
||||||
|
where_clause: &str,
|
||||||
|
order_clause: &str,
|
||||||
|
) -> String {
|
||||||
|
let cipher_expr = cipher_json_expr(attachments_enabled);
|
||||||
|
format!(
|
||||||
|
"SELECT {cipher_expr} AS cipher_json
|
||||||
|
FROM ciphers c
|
||||||
|
{where_clause}
|
||||||
|
{order_clause}",
|
||||||
|
cipher_expr = cipher_expr,
|
||||||
|
where_clause = where_clause,
|
||||||
|
order_clause = order_clause,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn is_sqlite_toobig(err: &worker::Error) -> bool {
|
||||||
|
let msg = err.to_string().to_ascii_lowercase();
|
||||||
|
msg.contains("sqlite_toobig") || msg.contains("string or blob too big")
|
||||||
|
}
|
||||||
|
|
||||||
/// Execute a cipher JSON projection query and return the raw JSON array string.
|
/// Execute a cipher JSON projection query and return the raw JSON array string.
|
||||||
/// This avoids JSON parsing in Rust, significantly reducing CPU time.
|
/// This avoids JSON parsing in Rust, significantly reducing CPU time.
|
||||||
pub(crate) async fn fetch_cipher_json_array_raw(
|
pub(crate) async fn fetch_cipher_json_array_raw(
|
||||||
|
|
@ -828,17 +859,59 @@ pub(crate) async fn fetch_cipher_json_array_raw(
|
||||||
where_clause: &str,
|
where_clause: &str,
|
||||||
params: &[JsValue],
|
params: &[JsValue],
|
||||||
order_clause: &str,
|
order_clause: &str,
|
||||||
|
force_row_query: bool,
|
||||||
) -> Result<String, AppError> {
|
) -> Result<String, AppError> {
|
||||||
|
async fn fetch_from_rows(
|
||||||
|
db: &worker::D1Database,
|
||||||
|
attachments_enabled: bool,
|
||||||
|
where_clause: &str,
|
||||||
|
params: &[JsValue],
|
||||||
|
order_clause: &str,
|
||||||
|
) -> Result<String, AppError> {
|
||||||
|
let sql = cipher_json_rows_sql(attachments_enabled, where_clause, order_clause);
|
||||||
|
let rows: Vec<CipherJsonRow> = db
|
||||||
|
.prepare(&sql)
|
||||||
|
.bind(params)?
|
||||||
|
.all()
|
||||||
|
.await
|
||||||
|
.map_err(db::map_d1_json_error)?
|
||||||
|
.results()
|
||||||
|
.map_err(|_| AppError::Database)?;
|
||||||
|
|
||||||
|
if rows.is_empty() {
|
||||||
|
return Ok("[]".to_string());
|
||||||
|
}
|
||||||
|
|
||||||
|
let total_len: usize = rows.iter().map(|r| r.cipher_json.len()).sum();
|
||||||
|
let separators_len = rows.len().saturating_sub(1);
|
||||||
|
let mut out = String::with_capacity(total_len + separators_len + 2);
|
||||||
|
out.push('[');
|
||||||
|
for (idx, row) in rows.into_iter().enumerate() {
|
||||||
|
if idx > 0 {
|
||||||
|
out.push(',');
|
||||||
|
}
|
||||||
|
out.push_str(&row.cipher_json);
|
||||||
|
}
|
||||||
|
out.push(']');
|
||||||
|
Ok(out)
|
||||||
|
}
|
||||||
|
|
||||||
|
if force_row_query {
|
||||||
|
return fetch_from_rows(db, attachments_enabled, where_clause, params, order_clause).await;
|
||||||
|
}
|
||||||
|
|
||||||
let sql = cipher_json_array_sql(attachments_enabled, where_clause, order_clause);
|
let sql = cipher_json_array_sql(attachments_enabled, where_clause, order_clause);
|
||||||
|
|
||||||
let row: Option<CipherJsonArrayRow> = db
|
let row: Result<Option<CipherJsonArrayRow>, worker::Error> =
|
||||||
.prepare(&sql)
|
db.prepare(&sql).bind(params)?.first(None).await;
|
||||||
.bind(params)?
|
|
||||||
.first(None)
|
|
||||||
.await
|
|
||||||
.map_err(db::map_d1_json_error)?;
|
|
||||||
|
|
||||||
Ok(row
|
match row {
|
||||||
.map(|r| r.ciphers_json)
|
Ok(row) => Ok(row
|
||||||
.unwrap_or_else(|| "[]".to_string()))
|
.map(|r| r.ciphers_json)
|
||||||
|
.unwrap_or_else(|| "[]".to_string())),
|
||||||
|
Err(err) if is_sqlite_toobig(&err) => {
|
||||||
|
fetch_from_rows(db, attachments_enabled, where_clause, params, order_clause).await
|
||||||
|
}
|
||||||
|
Err(err) => Err(db::map_d1_json_error(err)),
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -73,6 +73,18 @@ pub(crate) fn allow_totp_drift(env: &worker::Env) -> bool {
|
||||||
.unwrap_or(true)
|
.unwrap_or(true)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Whether to prefer fetching cipher JSON rows and building arrays in the Worker.
|
||||||
|
///
|
||||||
|
/// This avoids D1/SQLite `SQLITE_TOOBIG` errors when using `json_group_array` on large vaults.
|
||||||
|
/// Defaults to false (try the single-query aggregation, then fallback on `SQLITE_TOOBIG`).
|
||||||
|
pub(crate) fn ciphers_default_row_query(env: &worker::Env) -> bool {
|
||||||
|
env.var("CIPHERS_DEFAULT_ROW_QUERY")
|
||||||
|
.ok()
|
||||||
|
.map(|value| value.to_string().to_lowercase())
|
||||||
|
.map(|value| matches!(value.as_str(), "1" | "true" | "yes" | "on"))
|
||||||
|
.unwrap_or(false)
|
||||||
|
}
|
||||||
|
|
||||||
/// Whether the user has 2FA enabled.
|
/// Whether the user has 2FA enabled.
|
||||||
pub(crate) async fn two_factor_enabled(
|
pub(crate) async fn two_factor_enabled(
|
||||||
db: &worker::D1Database,
|
db: &worker::D1Database,
|
||||||
|
|
|
||||||
|
|
@ -6,7 +6,7 @@ use crate::{
|
||||||
auth::Claims,
|
auth::Claims,
|
||||||
db,
|
db,
|
||||||
error::AppError,
|
error::AppError,
|
||||||
handlers::{attachments, ciphers, domains, two_factor_enabled},
|
handlers::{attachments, ciphers, ciphers_default_row_query, domains, two_factor_enabled},
|
||||||
models::{
|
models::{
|
||||||
folder::{Folder, FolderResponse},
|
folder::{Folder, FolderResponse},
|
||||||
sync::Profile,
|
sync::Profile,
|
||||||
|
|
@ -79,12 +79,14 @@ pub async fn get_sync_data(
|
||||||
|
|
||||||
// Fetch ciphers as raw JSON array string (no parsing in Rust!)
|
// Fetch ciphers as raw JSON array string (no parsing in Rust!)
|
||||||
let include_attachments = attachments::attachments_enabled(env.as_ref());
|
let include_attachments = attachments::attachments_enabled(env.as_ref());
|
||||||
|
let force_row_query = ciphers_default_row_query(env.as_ref());
|
||||||
let ciphers_json = ciphers::fetch_cipher_json_array_raw(
|
let ciphers_json = ciphers::fetch_cipher_json_array_raw(
|
||||||
&db,
|
&db,
|
||||||
include_attachments,
|
include_attachments,
|
||||||
"WHERE c.user_id = ?1",
|
"WHERE c.user_id = ?1",
|
||||||
&[user_id.clone().into()],
|
&[user_id.clone().into()],
|
||||||
"",
|
"",
|
||||||
|
force_row_query,
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -52,6 +52,12 @@ run_worker_first = ["/api/*", "/identity/*"]
|
||||||
# Set to 0 means no batching (all records imported in a single batch).
|
# Set to 0 means no batching (all records imported in a single batch).
|
||||||
# IMPORT_BATCH_SIZE = "30"
|
# IMPORT_BATCH_SIZE = "30"
|
||||||
|
|
||||||
|
# Cipher sync/list JSON query mode.
|
||||||
|
# If enabled, fetch cipher JSON per-row and build the JSON array in the Worker
|
||||||
|
# to avoid D1/SQLite `SQLITE_TOOBIG` errors on very large vaults.
|
||||||
|
# Defaults to false (use `json_group_array` and fallback automatically on `SQLITE_TOOBIG`).
|
||||||
|
# CIPHERS_DEFAULT_ROW_QUERY = "true"
|
||||||
|
|
||||||
# Number of days to keep soft-deleted items before auto-purging.
|
# Number of days to keep soft-deleted items before auto-purging.
|
||||||
# Defaults to 30 days if not set. Set to 0 to disable auto-purge.
|
# Defaults to 30 days if not set. Set to 0 to disable auto-purge.
|
||||||
# TRASH_AUTO_DELETE_DAYS = "30"
|
# TRASH_AUTO_DELETE_DAYS = "30"
|
||||||
|
|
@ -132,4 +138,4 @@ persist = true
|
||||||
[env.dev.observability.traces]
|
[env.dev.observability.traces]
|
||||||
enabled = false
|
enabled = false
|
||||||
persist = true
|
persist = true
|
||||||
head_sampling_rate = 1
|
head_sampling_rate = 1
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue