aedb 0.3.1

//! Same-primary-key rewrite benchmark: N sequential single-mutation `Upsert`
//! commits to ONE table PK, followed by the same shape with `KvSet` on one KV
//! key, printing per-100-commit mean latency for both.
//!
//! Written to chase a report (observed through Arcana) of table upserts to a
//! hot PK degrading from ~250us to ~800us per commit within 500 rewrites while
//! same-key KvSet commits stayed flat. AEDB stores exactly one inline MVCC
//! version per row (`StoredRow::Versioned`) and the commit path never walks
//! per-key history, so per-commit latency here is FLAT in the rewrite count;
//! this bench pins that property. Upsert commits do cost a small constant
//! multiple of KvSet commits (validation, index maintenance, larger WAL
//! frames, no inline KV epoch fast path). Latency-vs-rewrite-count growth
//! observed on a live system therefore points at the deployment environment
//! (e.g. saturated-disk WAL fsync/write-back stalls, which also inflate and
//! wobble the absolute numbers printed here) or at layers above AEDB, not at
//! per-key accumulation inside the storage engine.
//!
//! Run with:
//! ```bash
//! cargo test --release --test same_pk_upsert_bench -- --ignored --nocapture
//! ```
//!
//! Knobs (env vars):
//! - `BENCH_ITERS`     number of commits per phase (default 1000)
//! - `BENCH_READS=1`   interleave an AtLatest point read after every upsert
//! - `BENCH_INDEX=1`   add a BTree index on a column that changes every upsert
//! - `BENCH_PARALLEL=1` enable the parallel apply executor
//! - `BENCH_METRICS=1` print executor telemetry (WAL append/sync breakdown)
//! - `AEDB_ENFORCE_BENCH_GATES=1` assert the last upsert window is <= 3x the
//!   first (off by default: on saturated disks fsync stalls dominate the
//!   means and can produce spurious drift unrelated to the engine)

use aedb::AedbInstance;
use aedb::catalog::DdlOperation;
use aedb::catalog::schema::ColumnDef;
use aedb::catalog::types::{ColumnType, Row, Value};
use aedb::commit::validation::Mutation;
use aedb::config::{AedbConfig, DurabilityMode, RecoveryMode};
use std::time::Instant;
use tempfile::tempdir;

const PROJECT_ID: &str = "bench";
const SCOPE_ID: &str = "app";
const TABLE_NAME: &str = "singleton";
const WINDOW: usize = 100;

fn iters() -> usize {
    std::env::var("BENCH_ITERS")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(1_000)
}

fn env_flag(name: &str) -> bool {
    std::env::var(name).ok().as_deref() == Some("1")
}

/// Mirrors the config shape Arcana runs AEDB with (serial apply, batch
/// durability with a long coalescing interval), where the slowdown was
/// reported.
fn bench_config() -> AedbConfig {
    AedbConfig {
        durability_mode: DurabilityMode::Batch,
        batch_interval_ms: 100,
        batch_max_bytes: usize::MAX,
        parallel_apply_enabled: env_flag("BENCH_PARALLEL"),
        recovery_mode: RecoveryMode::Permissive,
        hash_chain_required: false,
        ..AedbConfig::default()
    }
}

async fn setup() -> (tempfile::TempDir, AedbInstance) {
    let dir = tempdir().expect("temp dir");
    let db = AedbInstance::open_anonymous(bench_config(), dir.path()).expect("open");
    db.create_project(PROJECT_ID).await.expect("project");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: None,
        if_not_exists: false,
        project_id: PROJECT_ID.into(),
        scope_id: SCOPE_ID.into(),
        table_name: TABLE_NAME.into(),
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
            ColumnDef {
                name: "counter".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("table");
    if env_flag("BENCH_INDEX") {
        db.commit(Mutation::Ddl(DdlOperation::CreateIndex {
            if_not_exists: false,
            project_id: PROJECT_ID.into(),
            scope_id: SCOPE_ID.into(),
            table_name: TABLE_NAME.into(),
            index_name: "by_counter".into(),
            columns: vec!["counter".into()],
            index_type: aedb::catalog::schema::IndexType::BTree,
            partial_filter: None,
        }))
        .await
        .expect("index");
    }
    (dir, db)
}

fn report(label: &str, latencies_ns: &[u128]) -> Vec<u64> {
    let mut per_window_us = Vec::new();
    for (window_index, window) in latencies_ns.chunks(WINDOW).enumerate() {
        let mean_us = (window.iter().sum::<u128>() / window.len() as u128 / 1_000) as u64;
        per_window_us.push(mean_us);
        println!(
            "{label}: commits {:>4}-{:<4} mean {:>6}us",
            window_index * WINDOW + 1,
            window_index * WINDOW + window.len(),
            mean_us
        );
    }
    per_window_us
}

fn dump_metrics_delta(
    label: &str,
    prev: &aedb::commit::executor::ExecutorMetrics,
    cur: &aedb::commit::executor::ExecutorMetrics,
) {
    if !env_flag("BENCH_METRICS") {
        return;
    }
    println!(
        "{label} metrics: commits +{} epochs +{} wal_sync_ops +{} (avg {}us) wal_append avg {}us \
         prestage_validate avg {}us epoch_process avg {}us coordinator_apply avg {}us",
        cur.commits_total - prev.commits_total,
        cur.epochs_total - prev.epochs_total,
        cur.wal_sync_ops - prev.wal_sync_ops,
        cur.avg_wal_sync_micros,
        cur.avg_wal_append_micros,
        cur.avg_prestage_validate_micros,
        cur.avg_epoch_process_micros,
        cur.avg_coordinator_apply_micros,
    );
}

async fn point_read_singleton(db: &AedbInstance) {
    let mut query = aedb::query::plan::Query::select(&["id", "name", "counter"]).from(TABLE_NAME);
    query.predicate = Some(aedb::query::plan::col("id").eq(aedb::query::plan::lit(1)));
    query.limit = Some(1);
    let result = db
        .query(PROJECT_ID, SCOPE_ID, query)
        .await
        .expect("point read");
    assert_eq!(result.rows.len(), 1);
}

#[tokio::test]
#[ignore = "benchmark; run explicitly with --release --nocapture"]
async fn same_pk_upsert_vs_kvset_latency() {
    let (_dir, db) = setup().await;
    let interleave_reads = env_flag("BENCH_READS");

    // Phase 1: repeated single-mutation Upsert commits to ONE primary key.
    let metrics_start = db.metrics();
    let mut upsert_lat_ns = Vec::with_capacity(iters());
    for i in 0..iters() {
        let mutation = Mutation::Upsert {
            project_id: PROJECT_ID.into(),
            scope_id: SCOPE_ID.into(),
            table_name: TABLE_NAME.into(),
            primary_key: vec![Value::Integer(1)],
            row: Row {
                values: vec![
                    Value::Integer(1),
                    Value::Text(format!("state-{i}").into()),
                    Value::Integer(i as i64),
                ],
            },
        };
        let t0 = Instant::now();
        db.commit(mutation).await.expect("upsert");
        upsert_lat_ns.push(t0.elapsed().as_nanos());
        if interleave_reads {
            point_read_singleton(&db).await;
        }
    }
    let upsert_windows = report("upsert same-pk", &upsert_lat_ns);
    let metrics_after_upsert = db.metrics();
    dump_metrics_delta("upsert", &metrics_start, &metrics_after_upsert);

    // Phase 2: the same commit shape with KvSet on one key for comparison.
    let mut kv_lat_ns = Vec::with_capacity(iters());
    for i in 0..iters() {
        let mutation = Mutation::KvSet {
            project_id: PROJECT_ID.into(),
            scope_id: SCOPE_ID.into(),
            key: b"singleton:state".to_vec(),
            value: format!("state-{i}").into_bytes(),
        };
        let t0 = Instant::now();
        db.commit(mutation).await.expect("kvset");
        kv_lat_ns.push(t0.elapsed().as_nanos());
    }
    let kv_windows = report("kvset same-key", &kv_lat_ns);
    dump_metrics_delta("kvset", &metrics_after_upsert, &db.metrics());

    let first = upsert_windows.first().copied().unwrap_or(0).max(1);
    let last = upsert_windows.last().copied().unwrap_or(0);
    println!(
        "upsert first-window {first}us, last-window {last}us, kv last-window {}us",
        kv_windows.last().copied().unwrap_or(0)
    );
    // Regression gate (opt-in, matching benchmark_gate.rs): same-PK upsert
    // latency must not grow with the number of rewrites to that PK.
    if env_flag("AEDB_ENFORCE_BENCH_GATES") {
        assert!(
            last <= first.saturating_mul(3),
            "same-PK upsert commit latency grew {first}us -> {last}us over {} commits",
            iters()
        );
    }
}