fsqlite-core 0.4.2

#[allow(clippy::wildcard_imports)]
use super::*;

impl Connection {
    pub(super) async fn pragma_integrity_check_rows(
        &self,
        pragma: &fsqlite_ast::PragmaStatement,
    ) -> Vec<Row> {
        let quick = pragma.name.name.eq_ignore_ascii_case("quick_check");
        let max_errors = integrity_check_error_limit(pragma.value.as_ref());
        let mut failures = Vec::new();
        if let Err(error) = self.validate_database_integrity(quick).await {
            failures.push(error.to_string());
        }

        // Qualified attached PRAGMAs already delegate to their child Connection.
        // Unqualified whole-database checks must also visit every attachment;
        // a clean main database alone cannot establish an aggregate "ok" verdict.
        if pragma.name.schema.is_none() {
            let attached_schemas = self
                .attached_schemas
                .borrow()
                .all_schemas()
                .into_iter()
                .filter(|schema| !is_builtin_schema(schema))
                .map(str::to_owned)
                .collect::<Vec<_>>();
            for schema in attached_schemas {
                // The existing validator reports at most one failure per
                // database. N caps diagnostics across the entire traversal.
                if failures.len() >= max_errors {
                    break;
                }
                if let Err(error) = self
                    .with_attached_connection_async(&schema, async |child| {
                        child.validate_database_integrity(quick).await
                    })
                    .await
                {
                    failures.push(format!("*** in database {schema} ***\n{error}"));
                }
            }
        }

        if !fsqlite_observability::metrics::metrics_disabled() {
            let registry = fsqlite_observability::metrics::global();
            if failures.is_empty() {
                registry.integrity_check_ok_total.inc();
            } else {
                registry.integrity_check_fail_total.inc();
            }
        }
        if failures.is_empty() {
            failures.push("ok".to_owned());
        }
        let mut rows = failures
            .into_iter()
            .map(|outcome| Row {
                values: vec![SqliteValue::Text(outcome.into())],
            })
            .collect::<Vec<_>>();
        // bd-7o1vu (GH#370), complement option (1): surface a legacy orphaned
        // `%_content` shadow on a CONTENTLESS FTS5 table as an informational
        // NOTE. The shadow is a well-formed table, so it never fails the
        // ok/error verdict above (the database stays integrity-CLEAN); the note
        // only makes the condition discoverable so a user knows the one-time
        // first-open migration will reclaim it. Appended AFTER the verdict, so
        // an oracle that reads the first row still observes "ok".
        for shadow in self.orphaned_fts5_content_shadow_names() {
            rows.push(Row {
                values: vec![SqliteValue::Text(
                    format!(
                        "note: orphaned FTS5 contentless content shadow table {shadow} \
                         (reclaimable; the one-time first-open migration drops it)"
                    )
                    .into(),
                )],
            });
        }
        rows
    }

    pub(super) async fn pragma_wal_checkpoint_rows(
        &self,
        pragma: &fsqlite_ast::PragmaStatement,
    ) -> Result<Vec<Row>> {
        let mode = if let Some(ref val) = pragma.value {
            parse_checkpoint_mode(val)?
        } else {
            self.checkpoint_schedule_override_mode()
                .unwrap_or(CheckpointMode::Passive)
        };

        // TEMP objects are connection-local and not pager/WAL-backed in
        // FrankenSQLite. A qualified TEMP checkpoint therefore has SQLite's
        // standard non-WAL sentinel and must not checkpoint `main` by accident.
        if pragma
            .name
            .schema
            .as_deref()
            .is_some_and(|schema| schema.eq_ignore_ascii_case("temp"))
        {
            return Ok(vec![Row {
                values: [0, -1, -1].into_iter().map(SqliteValue::Integer).collect(),
            }]);
        }

        let mut primary = self.pragma_wal_checkpoint_database(mode).await?;

        // SQLite interprets an unqualified wal_checkpoint as "all schemas".
        // Result counts come from the first database (main), while SQLITE_BUSY
        // is aggregated across every checkpointed database. Attached databases
        // are separate child Connections here, so fan out in attach order and
        // retain main's log/backfill values.
        if pragma.name.schema.is_none() {
            let attached_schemas = self
                .attached_schemas
                .borrow()
                .all_schemas()
                .into_iter()
                .filter(|schema| !is_builtin_schema(schema))
                .map(str::to_owned)
                .collect::<Vec<_>>();
            for schema in attached_schemas {
                let attached = self
                    .with_attached_connection_async(&schema, async |child| {
                        child.pragma_wal_checkpoint_database(mode).await
                    })
                    .await?;
                primary[0] = primary[0].max(attached[0]);
            }
        }

        Ok(vec![Row {
            values: primary
                .into_iter()
                .map(SqliteValue::Integer)
                .collect::<Vec<_>>(),
        }])
    }

    async fn pragma_wal_checkpoint_database(&self, mode: CheckpointMode) -> Result<[i64; 3]> {
        // SQLite returns the sentinel tuple instead of erroring when the
        // database is not in WAL mode.
        if self.pager.journal_mode() != JournalMode::Wal {
            return Ok([0, -1, -1]);
        }
        let cx = self.op_cx()?;
        if self.wal_checkpoint_blocked_by_active_concurrent_txns() {
            let log_frames =
                i64::try_from(self.pager.wal_frame_count(&cx).await).unwrap_or(i64::MAX);
            return Ok([1, log_frames, 0]);
        }

        self.invalidate_cached_write_txn(&cx).await;
        self.invalidate_cached_read_snapshot(&cx).await;
        let checkpoint_metrics_before = fsqlite_wal::GLOBAL_WAL_METRICS.snapshot();
        let result = match self.pager.checkpoint(&cx, mode).await {
            Ok(result) => result,
            // GH#399: another process owns the checkpoint fence right now.
            // `sqlite3_wal_checkpoint_v2` reports that as SQLITE_BUSY without
            // consulting the busy handler, and the PRAGMA surfaces it as
            // `busy = 1` with nothing checkpointed rather than as an error, so
            // peers closing at the same moment do not fail each other's
            // close-time checkpoints.
            Err(FrankenError::Busy) => {
                let log_frames =
                    i64::try_from(self.pager.wal_frame_count(&cx).await).unwrap_or(i64::MAX);
                return Ok([1, log_frames, 0]);
            }
            Err(error) => return Err(error),
        };
        // GH #384: the pager refreshed its durable WAL horizon while holding
        // the checkpoint fence. Carry that horizon into the process-shared
        // MVCC clock before a later BEGIN is compared with CommitIndex.
        self.align_commit_clock_floor(self.pager.published_snapshot().visible_commit_seq);
        let checkpoint_metrics_after = fsqlite_wal::GLOBAL_WAL_METRICS.snapshot();
        let checkpoint_duration_us = checkpoint_metrics_after
            .checkpoint_duration_us_total
            .saturating_sub(checkpoint_metrics_before.checkpoint_duration_us_total);
        self.checkpoint_advisor_note_checkpoint(mode, &result, checkpoint_duration_us);

        // GH#399: SQLite reports `busy = 1` when readers kept the checkpoint
        // from finishing — frames beyond the oldest reader horizon stayed in
        // the WAL, or a RESTART/TRUNCATE could not replace the generation
        // because a peer process still pins it. Mirror that so callers can
        // retry instead of assuming the WAL was truncated.
        let reset_requested = matches!(mode, CheckpointMode::Restart | CheckpointMode::Truncate);
        let blocked_by_readers = !result.completed
            || (reset_requested && result.total_frames > 0 && !result.wal_was_reset);

        Ok([
            i64::from(blocked_by_readers),
            i64::from(result.total_frames),
            i64::from(result.frames_backfilled),
        ])
    }
}

fn integrity_check_error_limit(value: Option<&fsqlite_ast::PragmaValue>) -> usize {
    let Some(value) = value else {
        return 100;
    };
    let expr = match value {
        fsqlite_ast::PragmaValue::Assign(expr) | fsqlite_ast::PragmaValue::Call(expr) => expr,
    };
    // SQLite reads the integer token's magnitude even when it has a sign.
    let expr = match expr {
        Expr::UnaryOp {
            op: UnaryOp::Plus | UnaryOp::Negate,
            expr,
            ..
        } => expr.as_ref(),
        _ => expr,
    };
    match expr {
        Expr::Literal(Literal::Integer(limit), _) if *limit != 0 => {
            usize::try_from(limit.unsigned_abs()).unwrap_or(usize::MAX)
        }
        _ => 100,
    }
}

fn parse_checkpoint_mode(value: &fsqlite_ast::PragmaValue) -> Result<CheckpointMode> {
    let expr = match value {
        fsqlite_ast::PragmaValue::Assign(e) | fsqlite_ast::PragmaValue::Call(e) => e,
    };
    let text = match expr {
        Expr::Literal(Literal::String(s), _) => s.clone(),
        Expr::Column(col_ref, _) if col_ref.table.is_none() => col_ref.column.to_string(),
        _ => {
            return Err(FrankenError::Internal(
                "PRAGMA wal_checkpoint mode must be PASSIVE/FULL/RESTART/TRUNCATE".to_owned(),
            ));
        }
    };
    match text.to_uppercase().as_str() {
        "PASSIVE" => Ok(CheckpointMode::Passive),
        "FULL" => Ok(CheckpointMode::Full),
        "RESTART" => Ok(CheckpointMode::Restart),
        "TRUNCATE" => Ok(CheckpointMode::Truncate),
        _ => Err(FrankenError::Internal(format!(
            "PRAGMA wal_checkpoint mode must be PASSIVE/FULL/RESTART/TRUNCATE, got `{text}`"
        ))),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    async fn corrupt_integrity_test_root(conn: &Connection) -> Result<()> {
        let root_page = conn
            .schema
            .borrow()
            .iter()
            .find(|table| table.name.eq_ignore_ascii_case("t"))
            .map(|table| table.root_page)
            .expect("fixture table root");
        let cx = conn.op_cx()?;
        if conn.retained_autocommit_txn.borrow().is_some() {
            conn.flush_retained_autocommit_txn(&cx).await?;
        }
        conn.invalidate_cached_write_txn(&cx).await;
        conn.invalidate_cached_read_snapshot(&cx).await;
        let mut txn = conn.pager.begin(&cx, TransactionMode::Immediate).await?;
        let page_no = PageNumber::new(u32::try_from(root_page).unwrap()).unwrap();
        let mut page = txn.get_page(&cx, page_no).await?.into_vec();
        assert_eq!(page[0], 0x0D, "fixture must start as a table leaf");
        page[0] = 0xFF;
        txn.write_page(&cx, page_no, &page).await?;
        txn.commit(&cx).await
    }

    fn assert_integrity_ok(rows: &[Row]) {
        assert_eq!(rows.len(), 1, "clean databases return one verdict");
        assert_eq!(rows[0].values(), &[SqliteValue::Text("ok".into())]);
    }

    #[test]
    fn unqualified_integrity_checks_report_attached_corruption() {
        asupersync::test_utils::run_test(|| async {
            let conn = Connection::open(":memory:").await.unwrap();
            conn.execute("CREATE TABLE t(id INTEGER PRIMARY KEY, v TEXT);")
                .await
                .unwrap();
            for schema in ["good", "bad_first", "bad_second"] {
                conn.execute(&format!("ATTACH DATABASE ':memory:' AS {schema};"))
                    .await
                    .unwrap();
                conn.execute(&format!(
                    "CREATE TABLE {schema}.t(id INTEGER PRIMARY KEY, v TEXT); \
                     INSERT INTO {schema}.t VALUES (1, 'payload');"
                ))
                .await
                .unwrap();
            }
            for pragma in ["integrity_check", "quick_check"] {
                assert_integrity_ok(&conn.query(&format!("PRAGMA {pragma};")).await.unwrap());
                let prepared = conn.prepare(&format!("PRAGMA {pragma};")).await.unwrap();
                assert_integrity_ok(&prepared.query().await.unwrap());
            }
            for schema in ["bad_first", "bad_second"] {
                conn.with_attached_connection_async(schema, async |child| {
                    corrupt_integrity_test_root(child).await
                })
                .await
                .unwrap();
            }

            for pragma in ["integrity_check", "quick_check"] {
                for schema in ["main", "good"] {
                    assert_integrity_ok(
                        &conn
                            .query(&format!("PRAGMA {schema}.{pragma};"))
                            .await
                            .unwrap(),
                    );
                }
                // Establish actual corruption before testing aggregate dispatch.
                for schema in ["bad_first", "bad_second"] {
                    let rows = conn
                        .query(&format!("PRAGMA {schema}.{pragma};"))
                        .await
                        .unwrap();
                    assert_eq!(rows.len(), 1);
                    let SqliteValue::Text(message) = &rows[0].values()[0] else {
                        panic!("expected corruption diagnostic");
                    };
                    assert!(message.contains("invalid B-tree page type"), "{message}");
                }
                for prepared in [false, true] {
                    for (argument, expected_count) in [
                        ("", 2),
                        ("(1)", 1),
                        ("(2)", 2),
                        ("(0)", 2),
                        ("(-1)", 1),
                        ("(+1)", 1),
                    ] {
                        let sql = format!("PRAGMA {pragma}{argument};");
                        let rows = if prepared {
                            conn.prepare(&sql).await.unwrap().query().await.unwrap()
                        } else {
                            conn.query(&sql).await.unwrap()
                        };
                        assert_eq!(
                            rows.len(),
                            expected_count,
                            "{sql} prepared={prepared} must apply one shared error budget: {rows:?}"
                        );
                        // FrankenSQLite's existing validator returns one diagnostic
                        // per database, including for hard B-tree corruption. This
                        // guards traversal, not stock's hard-error sequencing.
                        for (row, schema) in rows.iter().zip(["bad_first", "bad_second"]) {
                            let SqliteValue::Text(message) = &row.values()[0] else {
                                panic!("expected corruption diagnostic");
                            };
                            assert!(
                                message.starts_with(&format!("*** in database {schema} ***\n")),
                                "{message}"
                            );
                            assert!(message.contains("invalid B-tree page type"), "{message}");
                        }
                    }
                }
            }
        });
    }

    #[test]
    fn qualified_integrity_checks_do_not_visit_corrupt_main() {
        asupersync::test_utils::run_test(|| async {
            let conn = Connection::open(":memory:").await.unwrap();
            conn.execute(
                "CREATE TABLE t(id INTEGER PRIMARY KEY); \
                 INSERT INTO t VALUES (1); \
                 ATTACH DATABASE ':memory:' AS aux; \
                 CREATE TABLE aux.t(id INTEGER PRIMARY KEY);",
            )
            .await
            .unwrap();
            corrupt_integrity_test_root(&conn).await.unwrap();
            for pragma in ["integrity_check", "quick_check"] {
                assert_integrity_ok(&conn.query(&format!("PRAGMA aux.{pragma};")).await.unwrap());
                for scope in ["", "main."] {
                    let rows = conn
                        .query(&format!("PRAGMA {scope}{pragma};"))
                        .await
                        .unwrap();
                    assert_eq!(rows.len(), 1);
                    let SqliteValue::Text(message) = &rows[0].values()[0] else {
                        panic!("expected corruption diagnostic");
                    };
                    assert!(message.contains("invalid B-tree page type"), "{message}");
                }
            }
        });
    }

    #[test]
    fn stock_integrity_checks_visit_attached_schemas() {
        let conn = rusqlite::Connection::open_in_memory().unwrap();
        conn.execute_batch(
            "ATTACH DATABASE ':memory:' AS aux; \
             ATTACH DATABASE ':memory:' AS aux_second; \
             CREATE TABLE aux.t(v INTEGER CHECK(v > 0)); \
             CREATE TABLE aux_second.t(v INTEGER CHECK(v > 0)); \
             PRAGMA ignore_check_constraints=ON; \
             INSERT INTO aux.t VALUES(-1); \
             INSERT INTO aux_second.t VALUES(-1); \
             PRAGMA ignore_check_constraints=OFF;",
        )
        .unwrap();
        for pragma in ["integrity_check", "quick_check"] {
            let main: String = conn
                .query_row(&format!("PRAGMA main.{pragma};"), [], |row| row.get(0))
                .unwrap();
            assert_eq!(main, "ok");
            for scope in ["", "aux."] {
                let report: String = conn
                    .query_row(&format!("PRAGMA {scope}{pragma};"), [], |row| row.get(0))
                    .unwrap();
                assert!(report.contains("CHECK constraint failed"), "{report}");
            }
            for (argument, expected_count) in [
                ("", 2),
                ("(1)", 1),
                ("(2)", 2),
                ("(0)", 2),
                ("(-1)", 1),
                ("(+1)", 1),
            ] {
                let sql = format!("PRAGMA {pragma}{argument};");
                let mut statement = conn.prepare(&sql).unwrap();
                let reports = statement
                    .query_map([], |row| row.get::<_, String>(0))
                    .unwrap()
                    .collect::<std::result::Result<Vec<_>, _>>()
                    .unwrap();
                assert_eq!(reports.len(), expected_count, "{sql}: {reports:?}");
                assert!(
                    reports
                        .iter()
                        .all(|report| report.contains("CHECK constraint failed"))
                );
            }
        }
    }

    #[test]
    fn unqualified_wal_checkpoint_truncates_attached_wal() {
        asupersync::test_utils::run_test(|| async {
            let dir = tempfile::tempdir().unwrap();
            let main_path = dir.path().join("main.db");
            let aux_path = dir.path().join("aux.db");
            let conn = Connection::open(main_path.to_str().unwrap()).await.unwrap();

            conn.execute("PRAGMA journal_mode=WAL;").await.unwrap();
            conn.execute(&format!(
                "ATTACH DATABASE '{}' AS aux;",
                aux_path.to_string_lossy().replace('\'', "''")
            ))
            .await
            .unwrap();
            conn.execute("PRAGMA aux.journal_mode=WAL;").await.unwrap();
            conn.execute("CREATE TABLE main_t(id INTEGER PRIMARY KEY, v TEXT);")
                .await
                .unwrap();
            conn.execute("CREATE TABLE aux.aux_t(id INTEGER PRIMARY KEY, v TEXT);")
                .await
                .unwrap();
            conn.execute("INSERT INTO main_t VALUES (1, 'main');")
                .await
                .unwrap();
            conn.execute("INSERT INTO aux.aux_t VALUES (1, 'aux');")
                .await
                .unwrap();

            let aux_frames_before = conn
                .with_attached_connection_async("aux", async |child| {
                    let cx = child.op_cx()?;
                    Ok(child.pager.wal_frame_count(&cx).await)
                })
                .await
                .unwrap();
            assert!(
                aux_frames_before > 0,
                "test requires a non-empty auxiliary WAL"
            );

            conn.query("PRAGMA wal_checkpoint(TRUNCATE);")
                .await
                .unwrap();

            let aux_frames_after = conn
                .with_attached_connection_async("aux", async |child| {
                    let cx = child.op_cx()?;
                    Ok(child.pager.wal_frame_count(&cx).await)
                })
                .await
                .unwrap();
            assert_eq!(aux_frames_after, 0, "unqualified checkpoint must visit aux");
        });
    }

    #[test]
    fn temp_wal_checkpoint_does_not_checkpoint_main() {
        asupersync::test_utils::run_test(|| async {
            let dir = tempfile::tempdir().unwrap();
            let main_path = dir.path().join("main.db");
            let conn = Connection::open(main_path.to_str().unwrap()).await.unwrap();

            conn.execute("PRAGMA journal_mode=WAL;").await.unwrap();
            conn.execute("CREATE TABLE main_t(id INTEGER PRIMARY KEY);")
                .await
                .unwrap();
            conn.execute("INSERT INTO main_t VALUES (1);")
                .await
                .unwrap();

            let cx = conn.op_cx().unwrap();
            let main_frames_before = conn.pager.wal_frame_count(&cx).await;
            assert!(main_frames_before > 0, "test requires a non-empty main WAL");

            let rows = conn
                .query("PRAGMA temp.wal_checkpoint(TRUNCATE);")
                .await
                .unwrap();
            let row = rows.first().expect("checkpoint returns one result row");
            assert_eq!(
                row.values,
                vec![
                    SqliteValue::Integer(0),
                    SqliteValue::Integer(-1),
                    SqliteValue::Integer(-1),
                ]
            );
            assert_eq!(
                conn.pager.wal_frame_count(&cx).await,
                main_frames_before,
                "TEMP checkpoint must not mutate main's WAL"
            );
        });
    }
}