Skip to content

[bfops/debug-sql]: debug logs #5523

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
bfops wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

[bfops/debug-sql]: debug logs #5523

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
22636ee
[bfops/debug-sql]: debug logs
bfops Jul 10, 2026
045b998
[bfops/debug-sql]: spacetime_with_timeout
bfops Jul 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
92 changes: 89 additions & 3 deletions crates/client-api/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,14 +137,28 @@ impl Host {
pub async fn exec_sql(
&self,
auth: AuthCtx,
_database: Database,
database: Database,
confirmed_read: bool,
body: String,
) -> axum::response::Result<Vec<SqlStmtResult<ProductValue>>> {
let request_start = std::time::Instant::now();
let sql_len = body.len();
log::debug!(
"sql exec: start database={} caller={} confirmed={confirmed_read} sql_len={sql_len}",
database.database_identity,
auth.caller()
);
let start = std::time::Instant::now();
log::debug!("sql exec: module lookup start database={}", database.database_identity);
let module_host = self
.module()
.await
.map_err(|_| (StatusCode::NOT_FOUND, "module not found".to_string()))?;
log::debug!(
"sql exec: module lookup finished database={} elapsed={:?}",
database.database_identity,
start.elapsed()
);

tracing::info!(sql = body);
let mut header = vec![];
Expand All @@ -154,6 +168,11 @@ impl Host {
let db = module_host.relational_db().clone();
let durable_offset = db.durable_tx_offset();

log::debug!(
"sql exec: sql::execute::run start database={} caller={}",
database.database_identity,
auth.caller()
);
let result = sql::execute::run(
db,
body,
Expand All @@ -164,9 +183,20 @@ impl Host {
)
.await
.map_err(|e| {
log::debug!(
"sql exec: sql::execute::run failed database={} elapsed={:?} err={e}",
database.database_identity,
sql_start.elapsed()
);
log::warn!("{e}");
(StatusCode::BAD_REQUEST, e.to_string())
})?;
log::debug!(
"sql exec: sql::execute::run finished database={} elapsed={:?} rows={}",
database.database_identity,
sql_start.elapsed(),
result.rows.len()
);

let total_duration = sql_start.elapsed();
drop(_guard);
Expand All @@ -184,12 +214,68 @@ impl Host {
total_duration_micros: total_duration.as_micros() as u64,
stats: SqlStmtStats::from_metrics(&result.metrics),
}];
log::debug!(
"sql exec: response prepared database={} stmt_results={} rows={} confirmed={confirmed_read}",
database.database_identity,
json.len(),
json.iter().map(|result| result.rows.len()).sum::<usize>()
);

if confirmed_read && let Some(mut durable_offset) = durable_offset {
let tx_offset = tx_offset.await.map_err(|_| log_and_500("transaction aborted"))?;
durable_offset.wait_for(tx_offset).await.map_err(log_and_500)?;
let start = std::time::Instant::now();
log::debug!(
"sql exec: awaiting transaction offset database={}",
database.database_identity
);
let tx_offset = tx_offset.await.map_err(|_| {
log::debug!(
"sql exec: transaction offset await failed database={} elapsed={:?}",
database.database_identity,
start.elapsed()
);
log_and_500("transaction aborted")
})?;
log::debug!(
"sql exec: transaction offset received database={} tx_offset={} elapsed={:?}",
database.database_identity,
tx_offset,
start.elapsed()
);

let start = std::time::Instant::now();
log::debug!(
"sql exec: waiting for durable offset database={} tx_offset={}",
database.database_identity,
tx_offset
);
durable_offset.wait_for(tx_offset).await.map_err(|err| {
log::debug!(
"sql exec: durable offset wait failed database={} tx_offset={} elapsed={:?} err={err}",
database.database_identity,
tx_offset,
start.elapsed()
);
log_and_500(err)
})?;
log::debug!(
"sql exec: durable offset reached database={} tx_offset={} elapsed={:?}",
database.database_identity,
tx_offset,
start.elapsed()
);
} else {
log::debug!(
"sql exec: skipping durable offset wait database={} confirmed={confirmed_read} has_durable_offset={}",
database.database_identity,
durable_offset.is_some()
);
}

log::debug!(
"sql exec: finished database={} elapsed={:?}",
database.database_identity,
request_start.elapsed()
);
Ok(json)
}

Expand Down
104 changes: 95 additions & 9 deletions crates/client-api/src/routes/database.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -717,22 +717,48 @@ where
{
tokio::spawn(async move {
let connection_id = generate_random_connection_id();
let request_start = std::time::Instant::now();
log::debug!("sql request {connection_id}: spawned connection lifecycle caller={caller_identity}");

// Run the module's client_connected reducer, if any.
// If it rejects the connection, bail before executing `fut`
let start = std::time::Instant::now();
log::debug!("sql request {connection_id}: client_connected start caller={caller_identity}");
module
.call_identity_connected(caller_auth, connection_id)
.await
.map_err(client_connected_error_to_response)?;
log::debug!(
"sql request {connection_id}: client_connected finished elapsed={:?}",
start.elapsed()
);

let start = std::time::Instant::now();
log::debug!("sql request {connection_id}: handler future start caller={caller_identity}");
let result = fut(module.clone(), caller_identity, connection_id).await;
log::debug!(
"sql request {connection_id}: handler future finished elapsed={:?} ok={}",
start.elapsed(),
result.is_ok()
);

// Always disconnect, even if authorization or execution failed.
let start = std::time::Instant::now();
log::debug!("sql request {connection_id}: client_disconnected start caller={caller_identity}");
module
.call_identity_disconnected(caller_identity, connection_id)
.await
.map_err(client_disconnected_error_to_response)?;
log::debug!(
"sql request {connection_id}: client_disconnected finished elapsed={:?}",
start.elapsed()
);

log::debug!(
"sql request {connection_id}: connection lifecycle finished elapsed={:?} ok={}",
request_start.elapsed(),
result.is_ok()
);
result
})
.await
Expand All @@ -750,21 +776,65 @@ pub async fn sql_direct<S>(
where
S: NodeDelegate + ControlStateDelegate + Authorization + 'static,
{
let confirmed_read = confirmed.unwrap_or(crate::DEFAULT_CONFIRMED_READS);
let sql_len = sql.len();
log::debug!(
"sql request: resolving database name_or_identity={name_or_identity:?} caller={caller_identity} confirmed={confirmed_read} sql_len={sql_len}"
);
let start = std::time::Instant::now();
let (host, database) = find_leader_and_database(&worker_ctx, name_or_identity).await?;

log::debug!(
"sql request: resolved database identity={} id={} elapsed={:?}",
database.database_identity,
database.id,
start.elapsed()
);

let start = std::time::Instant::now();
log::debug!(
"sql request: loading module host database={}",
database.database_identity
);
let module = host.module().await.map_err(log_and_500)?;
let fut = async move |_module: ModuleHost, caller_identity: Identity, _connection_id: ConnectionId| {
log::debug!(
"sql request: loaded module host database={} elapsed={:?}",
database.database_identity,
start.elapsed()
);
let fut = async move |_module: ModuleHost, caller_identity: Identity, connection_id: ConnectionId| {
let start = std::time::Instant::now();
log::debug!(
"sql request {connection_id}: authorize_sql start caller={caller_identity} database={}",
database.database_identity
);
let sql_auth = worker_ctx
.authorize_sql(caller_identity, database.database_identity)
.await?;
log::debug!(
"sql request {connection_id}: authorize_sql finished elapsed={:?}",
start.elapsed()
);

host.exec_sql(
sql_auth,
database,
confirmed.unwrap_or(crate::DEFAULT_CONFIRMED_READS),
sql,
)
.await
let start = std::time::Instant::now();
log::debug!(
"sql request {connection_id}: host.exec_sql start database={} confirmed={confirmed_read} sql_len={sql_len}",
database.database_identity
);
host.exec_sql(sql_auth, database, confirmed_read, sql)
.await
.inspect(|rows| {
log::debug!(
"sql request {connection_id}: host.exec_sql finished elapsed={:?} stmt_results={}",
start.elapsed(),
rows.len()
);
})
.inspect_err(|err| {
log::debug!(
"sql request {connection_id}: host.exec_sql failed elapsed={:?} err={err:?}",
start.elapsed(),
);
})
};

with_connection(module, caller_auth, caller_identity, fut).await
Expand All @@ -780,11 +850,27 @@ pub async fn sql<S>(
where
S: NodeDelegate + ControlStateDelegate + Authorization + 'static,
{
log::debug!(
"sql route: received request caller={} confirmed={:?} sql_len={}",
auth.claims.identity,
params.confirmed,
body.len()
);
let caller_identity = auth.claims.identity;
let caller_auth: ConnectionAuthCtx = auth.into();
let start = std::time::Instant::now();
let json = sql_direct(worker_ctx, name_or_identity, params, caller_identity, caller_auth, body).await?;
log::debug!(
"sql route: sql_direct finished caller={caller_identity} elapsed={:?} stmt_results={}",
start.elapsed(),
json.len()
);

let total_duration = json.iter().fold(0, |acc, x| acc + x.total_duration_micros);
log::debug!(
"sql route: returning response caller={caller_identity} execution_duration_micros={}",
total_duration
);

Ok((
TypedHeader(SpacetimeExecutionDurationMicros(Duration::from_micros(total_duration))),
Expand Down
60 changes: 53 additions & 7 deletions crates/core/src/host/module_host.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1821,6 +1821,10 @@ impl ModuleHost {
{
self.guard_closed()?;
let timer_guard = self.start_call_timer(label);
let start = std::time::Instant::now();
if label == "call_sql" {
log::debug!("module host call_sql: dispatch start");
}

scopeguard::defer_on_unwind!({
log::warn!("module operation {label} panicked");
Expand All @@ -1829,21 +1833,42 @@ impl ModuleHost {

Ok(match &*self.inner {
ModuleHostInner::Wasm(host) => {
if label == "call_sql" {
log::debug!("module host call_sql: dispatching to wasm main worker");
}
let executor = host.executor.clone();
executor
let result = executor
.run_sync_job(move |state| {
state.with_instance(move |inst| {
drop(timer_guard);
wasm(arg, inst)
})
})
.await
.await;
if label == "call_sql" {
log::debug!(
"module host call_sql: wasm main worker finished elapsed={:?}",
start.elapsed()
);
}
result
}
ModuleHostInner::Js(host) => {
if label == "call_sql" {
log::debug!("module host call_sql: dispatching to js main worker");
}
drop(timer_guard);
host.main_instance
let result = host
.main_instance
.with_instance(|inst| async move { js(arg, &inst).await })
.await
.await;
if label == "call_sql" {
log::debug!(
"module host call_sql: js main worker finished elapsed={:?}",
start.elapsed()
);
}
result
}
})
}
Expand Down Expand Up @@ -1986,10 +2011,18 @@ impl ModuleHost {
}

async fn call_sql_command(&self, cmd: SqlCommand) -> Result<SqlCommandResult, DBError> {
call_instance!(self, "call_sql", cmd, |cmd, inst| inst.call_sql(cmd), |cmd, inst| inst
let start = std::time::Instant::now();
log::debug!("module host call_sql: command dispatch start");
let result = call_instance!(self, "call_sql", cmd, |cmd, inst| inst.call_sql(cmd), |cmd, inst| inst
.call_sql(cmd)
.await,)
.map_err(|_| DBError::Other(anyhow::anyhow!("no such module")))
.map_err(|_| DBError::Other(anyhow::anyhow!("no such module")));
log::debug!(
"module host call_sql: command dispatch finished elapsed={:?} ok={}",
start.elapsed(),
result.is_ok()
);
result
}

pub async fn disconnect_client(&self, client_id: ClientActorId) {
Expand Down Expand Up @@ -2510,6 +2543,13 @@ impl ModuleHost {
subs: Option<ModuleSubscriptions>,
head: &mut Vec<(RawIdentifier, AlgebraicType)>,
) -> Result<SqlResult, DBError> {
let start = std::time::Instant::now();
log::debug!(
"module host call_sql: start caller={} sql_len={} subscriptions={}",
auth.caller(),
sql_text.len(),
subs.is_some()
);
let cmd = SqlCommand {
db,
sql_text,
Expand All @@ -2520,7 +2560,13 @@ impl ModuleHost {
let res = self.call_sql_command(cmd).await?;

*head = res.head;
res.result
let result = res.result;
log::debug!(
"module host call_sql: finished elapsed={:?} ok={}",
start.elapsed(),
result.is_ok()
);
result
}

pub async fn call_procedure(
Expand Down
Loading
Loading