Function logs

crates/client-api/src/routes/database.rs:620–683

(
    State(worker_ctx): State<S>,
    Path(LogsParams { name_or_identity }): Path<LogsParams>,
    Query(LogsQuery { num_lines, follow }): Query<LogsQuery>,
    Extension(auth): Extension<SpacetimeAu

}

pub async fn logs<S>(
    State(worker_ctx): State<S>,
    Path(LogsParams { name_or_identity }): Path<LogsParams>,
    Query(LogsQuery { num_lines, follow }): Query<LogsQuery>,
    Extension(auth): Extension<SpacetimeAuth>,
) -> axum::response::Result<impl IntoResponse>
where
    S: ControlStateDelegate + NodeDelegate + Authorization,
{
    // You should not be able to read the logs from a database that you do not own
    // so, unless you are the owner, this will fail.

    let database_identity: Identity = name_or_identity.resolve(&worker_ctx).await?;
    let database = worker_ctx_find_database(&worker_ctx, &database_identity)
        .await?
        .ok_or(NO_SUCH_DATABASE)?;

    worker_ctx
        .authorize_action(auth.claims.identity, database.database_identity, Action::ViewModuleLogs)
        .await?;

    fn log_err(database: Identity) -> impl Fn(&io::Error) {
        move |e| warn!("error serving module logs for database {database}: {e:#}")
    }

    let body = match worker_ctx.leader(database.id).await {
        Ok(host) => {
            let module = host.module().await.map_err(log_and_500)?;
            let logs = module.database_logger().tail(num_lines, follow).await.map_err(|e| {
                warn!("database={database_identity} unable to tail logs: {e:#}");
                (StatusCode::SERVICE_UNAVAILABLE, "Logs are temporarily not available")
            })?;
            Body::from_stream(logs.inspect_err(log_err(database_identity)))
        }
        Err(e) if e.is_misdirected() => return Err(MISDIRECTED.into()),
        // If this is the right node for the current or last-known leader,
        // we may still be able to serve logs from disk,
        // even if we can't get hold of a running [ModuleHost].
        Err(e) => {
            warn!("could not obtain leader host for module logs: {e:#}");
            let Some(replica) = worker_ctx.get_leader_replica_by_database(database.id).await else {
                return Err(MISDIRECTED.into());
            };
            let logs_dir = worker_ctx.module_logs_dir(replica.id);
            if !logs_dir.0.try_exists().map_err(log_and_500)? {
                // Probably an in-memory database.
                // Logs may become available at a later time.
                return Err((
                    StatusCode::SERVICE_UNAVAILABLE,
                    "Database is not running and doesn't have persistent logs",
                )
                    .into());
            }
            let logs = DatabaseLogger::read_latest_on_disk(logs_dir, num_lines);
            Body::from_stream(logs.inspect_err(log_err(database_identity)))
        }
    };