# Copyright (C) 2026 Intel Corporation
# SPDX-License-Identifier: PostgreSQL

# 58_residency_rollback_reconcile.pl — a rolled-back INSERT's growth is
# credited back as reclaimable residency rather than left permanently
# stranded against the database's residency budget.
#
# Claim under test: SvsMemoryReanchorInsert records each insert's own growth
# and the graph generation it landed in; on abort, SvsMemoryCreditAbortedInserts
# moves that growth from residency_bytes_committed into
# residency_bytes_reclaimable, bounded by svs.compact_threshold_pct of the
# database's budget, with a COMPACT reclaiming the debt once the cap is
# crossed and a checkpoint's own compaction reclaiming it too. Eleven cases:
#
#   T1  single-session rollback credits exactly its own growth back
#   T2  a rollback-then-commit ends up within the credited gap of a plain
#       commit of the same data
#   T3  an INSERT-only, non-owner role cannot exhaust a tightly pinned
#       budget by looping BEGIN/INSERT/ROLLBACK
#   T4  ROLLBACK TO SAVEPOINT credits that subtransaction's growth; the
#       outer COMMIT does not change it again
#   T5  nested savepoints plus RELEASE credit their total growth exactly
#       once at the top-level ROLLBACK
#   T6  a worker restart plus a forced reload advances the graph
#       generation, so a stale ROLLBACK afterward credits nothing
#   T7  pg_terminate_backend on the inserting session still runs the abort
#       path and credits the growth
#   T8  a rollback larger than the reclaim cap forces a COMPACT; pinning
#       svs.compact_threshold_pct to 100 documents the no-credit opt-out
#   T9  a rollback below the cap is still reclaimed once a checkpoint
#       compacts the index, with no later insert needed
#   T10 two interleaved sessions: one commits, the other aborts; the sum of
#       committed and reclaimable bytes is unmoved by the abort
#   T11 a checkpoint that fails after SVSSaveIndex already compacted the
#       live graph reconciles immediately in its own PG_CATCH, rather than
#       leaving accounting stale until an unrelated later write self-corrects
#       it (requires --enable-injection-points; skipped otherwise)
#
# Execution order below groups cases by which database/GUC state they need,
# not the T1..T11 reading order: T3 and T8 get their own tightly-pinned
# database so a tiny budget doesn't interfere with the main database's
# cases, and T9/T11 are last because they change svs.checkpoint_operations
# for the rest of the file.
#
# Row-counting trap: an unqualified count(*) on a vamana-indexed table can
# be planned as a key-less Index Only Scan and silently return 0 regardless
# of actual contents. row_count() below forces a seqscan before counting;
# every other assertion reads residency_bytes_committed /
# residency_bytes_reclaimable from pg_stat_vamana_worker instead, which does
# not go through the index at all.

use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
use Time::HiRes qw(usleep time);

use FindBin qw($Bin);
use lib "$Bin/../perl";
use VamanaTestUtils qw(:all);

my $node = PostgreSQL::Test::Cluster->new('residency_rollback_reconcile');
$node->init;
$node->append_conf('postgresql.conf', "shared_preload_libraries = 'svs'");
$node->append_conf('postgresql.conf', "wal_level = logical");
$node->append_conf('postgresql.conf', "max_replication_slots = 20");
$node->append_conf('postgresql.conf', "max_wal_senders = 20");
$node->append_conf('postgresql.conf', "max_worker_processes = 32");
$node->append_conf('postgresql.conf', "svs.launcher_database = 'postgres'");
# Five databases get enrolled across this file (postgres, dba, dbb,
# attack_db, tightdb); search_work_mem defaults to
# svs.default_search_work_mem (100MB) per database regardless of any
# residency_memory override, so the search_work_mem ceiling needs covering
# 5 x 100MB on its own. Both cluster-wide ceilings are raised generously,
# same reasoning as 50_undo_growth_bound.pl.
$node->append_conf('postgresql.conf', "svs.max_residency_memory = '600MB'");
$node->append_conf('postgresql.conf', "svs.max_search_work_mem = '600MB'");
# No unplanned compaction (VACUUM-triggered or otherwise) should land in the
# middle of a measurement.
$node->append_conf('postgresql.conf', "autovacuum = off");
# T9 waits on the DEBUG1 "checkpoint complete" log line (PerformCheckpoint,
# vamana_checkpoint.c), which is otherwise never emitted at the default
# log level.
$node->append_conf('postgresql.conf', "log_min_messages = debug1");
# T8's over-the-cap case rolls back a 12,000-row single-transaction INSERT,
# whose undo has to delete all 12,000 entries from the worker synchronously;
# the default 5s svs.worker_timeout_ms is too tight for that (the same
# "vamana worker timed out after 5000 ms" characteristic test/t/50_undo_
# growth_bound.pl documents for large single-transaction rollbacks).
$node->append_conf('postgresql.conf', "svs.worker_timeout_ms = 60000");
$node->start;

$node->safe_psql('postgres', "CREATE EXTENSION vector;");
$node->safe_psql('postgres', "CREATE EXTENSION svs;");
$node->safe_psql('postgres',
    "INSERT INTO vamana_databases (datname, enabled) VALUES ('postgres', true);");

my $worker_pid = wait_for_worker($node, 40);
ok($worker_pid =~ /^\d+$/, "main worker running (pid=$worker_pid)");

$node->safe_psql('postgres', qq(
    CREATE TABLE t (val vector($dim));
    INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 200);
    CREATE INDEX t_idx ON t USING vamana (val vector_l2_ops);
));
wait_for_worker($node, 10);

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

# committed_reclaimable: ($residency_bytes_committed, $residency_bytes_reclaimable)
# for $dbname, read from pg_stat_vamana_worker. Always queried from the
# 'postgres' connection, like worker_committed_totals in VamanaTestUtils.pm,
# since the view's rows span every enrolled database and a superuser sees
# them all regardless of which database it is connected to.
sub committed_reclaimable
{
    my ($dbname) = @_;
    $dbname //= 'postgres';
    my $row = $node->safe_psql('postgres', qq(
        SELECT residency_bytes_committed, residency_bytes_reclaimable
        FROM pg_stat_vamana_worker
        WHERE db_oid = (SELECT oid FROM pg_database WHERE datname = '$dbname');
    ));
    chomp $row;
    my ($c, $r) = split(/\|/, $row);
    return ($c + 0, $r + 0);
}

sub get_worker_pid
{
    my ($db) = @_;
    my $pid = $node->safe_psql('postgres',
        "SELECT worker_pid FROM pg_stat_vamana_worker w "
      . "JOIN pg_database d ON d.oid = w.db_oid WHERE d.datname = '$db';");
    chomp $pid;
    return $pid;
}

sub wait_for_new_worker_pid
{
    my ($db, $old, $attempts) = @_;
    $attempts //= 60;
    for my $i (1 .. $attempts)
    {
        usleep(500_000);
        my $pid = get_worker_pid($db);
        return $pid if $pid =~ /^\d+$/ && $pid ne $old;
    }
    return '';
}

# row_count: count(*) on $table in $dbname with the planner forced off the
# index, per the T-SCAN-1 trap noted at the top of this file.
sub row_count
{
    my ($dbname, $table) = @_;
    $table //= 't';
    my $n = $node->safe_psql($dbname,
        "SET enable_indexscan = off; SET enable_indexonlyscan = off; "
      . "SELECT count(*) FROM $table;");
    chomp $n;
    return $n + 0;
}

# enroll_fresh_db: a new database, both extensions, enrolled from
# svs.launcher_database, with a $seed_rows-row seed table/index and an
# optional residency_memory override (MB). Adapted from the same-named
# helper in test/t/50_undo_growth_bound.pl.
sub enroll_fresh_db
{
    my ($dbname, $seed_rows, $residency_mb_override) = @_;

    $node->safe_psql('postgres', "DROP DATABASE IF EXISTS $dbname;");
    $node->safe_psql('postgres', "CREATE DATABASE $dbname;");
    $node->safe_psql($dbname, "CREATE EXTENSION vector;");
    $node->safe_psql($dbname, "CREATE EXTENSION svs;");

    my $residency_sql = defined($residency_mb_override) ? "$residency_mb_override" : "NULL";
    $node->safe_psql('postgres',
        "INSERT INTO vamana_databases (datname, enabled, residency_memory) "
      . "VALUES ('$dbname', true, $residency_sql);");
    my $pid = wait_for_worker_db($node, $dbname, 40);

    $node->safe_psql($dbname, "CREATE TABLE t (val vector($dim));");
    $node->safe_psql($dbname,
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, $seed_rows);");
    $node->safe_psql($dbname,
        "CREATE INDEX t_idx ON t USING vamana (val vector_l2_ops);");
    wait_for_worker_db($node, $dbname, 10);

    return $pid;
}

sub log_since
{
    my ($pos) = @_;
    return substr($node->log_content(), $pos);
}

# ---------------------------------------------------------------------------
# T1: a single session's BEGIN/INSERT/ROLLBACK credits exactly its own
# growth back; committed returns to its exact pre-transaction value.
# ---------------------------------------------------------------------------
diag("=== T1: single-session rollback credits its own growth exactly ===");
{
    my ($c0, $r0) = committed_reclaimable();

    my $a = $node->background_psql('postgres');
    $a->query_safe("BEGIN;");
    $a->query_safe(
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 100);");

    my ($c_mid, $r_mid) = committed_reclaimable();
    my $growth = $c_mid - $c0;

    $a->query_safe("ROLLBACK;");
    $a->quit;

    my ($c1, $r1) = committed_reclaimable();

    diag(sprintf("T1: c0=%d c_mid=%d growth=%d c1=%d r0=%d r1=%d",
        $c0, $c_mid, $growth, $c1, $r0, $r1));

    is($c1, $c0, "T1: committed returns exactly to its pre-transaction value");
    is($r1, $r0 + $growth, "T1: reclaimable rises by exactly the growth the rollback credited back");
}

# ---------------------------------------------------------------------------
# T2: rollback 100 then commit 100 (dba), vs. a plain commit of 100 with the
# same seed (dbb). dba's committed figure must not exceed dbb's by more than
# dba's own reclaimable covers; a forced COMPACT in both brings them equal
# (modulo the block-quantization gap this file does not silently paper over).
# ---------------------------------------------------------------------------
diag("=== T2: rollback-then-commit vs. a twin database's plain commit ===");
{
    enroll_fresh_db('dba', 200, undef);
    enroll_fresh_db('dbb', 200, undef);

    $node->safe_psql('dba', "SELECT setseed(0.5);");
    $node->safe_psql('dba',
        "BEGIN; INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 100); ROLLBACK;");
    $node->safe_psql('dba', "SELECT setseed(0.5);");
    $node->safe_psql('dba',
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 100);");

    $node->safe_psql('dbb', "SELECT setseed(0.5);");
    $node->safe_psql('dbb',
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 100);");

    my ($c_dba, $r_dba) = committed_reclaimable('dba');
    my ($c_dbb, $r_dbb) = committed_reclaimable('dbb');

    diag(sprintf("T2 before compact: dba committed=%d reclaimable=%d, dbb committed=%d reclaimable=%d",
        $c_dba, $r_dba, $c_dbb, $r_dbb));

    cmp_ok($c_dba, '<=', $c_dbb,
        "T2: a churned-then-settled dba never shows more committed than a plain commit");
    cmp_ok($c_dba + $r_dba, '>=', $c_dbb,
        "T2: dba's committed+reclaimable covers dbb's committed (nothing is actually lost)");

    is(row_count('dba'), 300, "T2: dba ends with 300 live rows (200 seed + 100 surviving commit)");
    is(row_count('dbb'), 300, "T2: dbb ends with 300 live rows (200 seed + 100 committed)");

    # Force a COMPACT in both the same way test/t/90_is194_reuse_probe.pl's
    # scratch probe did: lower the threshold and VACUUM.
    for my $db (qw(dba dbb))
    {
        $node->safe_psql($db, "SET svs.compact_threshold_pct = 1;");
        $node->safe_psql($db, "VACUUM t;");
    }
    wait_for_worker_db($node, 'dba', 10);
    wait_for_worker_db($node, 'dbb', 10);

    my ($c_dba2, $r_dba2) = committed_reclaimable('dba');
    my ($c_dbb2, $r_dbb2) = committed_reclaimable('dbb');

    diag(sprintf("T2 after forced compact: dba committed=%d reclaimable=%d, dbb committed=%d reclaimable=%d",
        $c_dba2, $r_dba2, $c_dbb2, $r_dbb2));

    is($r_dba2, 0, "T2: dba's reclaimable is 0 after a forced compact");
    if ($c_dba2 == $c_dbb2)
    {
        is($c_dba2, $c_dbb2, "T2: after compacting both, dba and dbb agree exactly");
    }
    else
    {
        diag(sprintf(
            "T2: dba and dbb differ by %d bytes after compacting both -- attributed to SVS's "
          . "block-quantized growth (F3), not re-litigated here",
            $c_dba2 - $c_dbb2));
        cmp_ok(abs($c_dba2 - $c_dbb2), '<', 65536,
            "T2: any post-compact gap between dba and dbb is small (well under one growth block)");
    }
}

# ---------------------------------------------------------------------------
# T4: SAVEPOINT; INSERT; ROLLBACK TO; INSERT; COMMIT -- the subtransaction's
# growth is credited at ROLLBACK TO, and the later COMMIT does not move it.
# ---------------------------------------------------------------------------
diag("=== T4: ROLLBACK TO SAVEPOINT credits that subtransaction's growth ===");
{
    # Note on the assertions below: committed and reclaimable are sampled from
    # a *separate* session than the one holding the open transaction, so a
    # "growth" figure computed as (committed just before ROLLBACK TO) minus
    # (committed just before the INSERT) is not reliable once the index has
    # any prior undo/consolidate history (as it does here, after T1): SVS's
    # own reported memory usage can dip slightly between two inserts even
    # with no delete in between (observed directly against this build via a
    # manual repro), and committed only reflects the fresh raw measurement,
    # not the sum of each row's own non-negative growth contribution the
    # undo log actually credits. The robust invariant -- the one this fix's
    # correctness actually depends on -- is conservation: an abort can only
    # move bytes from committed into reclaimable, never change their sum.
    my ($c0, $r0) = committed_reclaimable();

    my $a = $node->background_psql('postgres');
    $a->query_safe("BEGIN;");
    $a->query_safe("SAVEPOINT s;");
    $a->query_safe(
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 100);");

    my ($c_mid, $r_mid) = committed_reclaimable();

    $a->query_safe("ROLLBACK TO s;");
    my ($c_rb, $r_rb) = committed_reclaimable();

    $a->query_safe("INSERT INTO t VALUES ('[$query_sql]');");
    $a->query_safe("COMMIT;");
    $a->quit;

    my ($c1, $r1) = committed_reclaimable();

    diag(sprintf("T4: c0=%d r0=%d c_mid=%d r_mid=%d c_rb=%d r_rb=%d c1=%d r1=%d",
        $c0, $r0, $c_mid, $r_mid, $c_rb, $r_rb, $c1, $r1));

    is($c_rb + $r_rb, $c_mid + $r_mid,
        "T4: ROLLBACK TO moves bytes from committed to reclaimable without changing their sum");
    cmp_ok($r_rb, '>', $r_mid, "T4: reclaimable rises at ROLLBACK TO (the subtransaction's growth is credited)");
    is($r1, $r_rb, "T4: the later COMMIT does not change reclaimable again");
}

# ---------------------------------------------------------------------------
# T5: nested savepoints plus RELEASE, then a top-level ROLLBACK -- total
# credit equals total growth, credited exactly once, no underflow WARNING.
# ---------------------------------------------------------------------------
diag("=== T5: nested savepoints + RELEASE, credited once at top-level ROLLBACK ===");
{
    my ($c0, $r0) = committed_reclaimable();
    my $log_pos = length($node->log_content());

    my $a = $node->background_psql('postgres');
    $a->query_safe("BEGIN;");
    $a->query_safe("SAVEPOINT s1;");
    $a->query_safe(
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 50);");
    $a->query_safe("SAVEPOINT s2;");
    $a->query_safe(
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 50);");

    my ($c_mid, $r_mid) = committed_reclaimable();

    $a->query_safe("RELEASE s2;");
    $a->query_safe("ROLLBACK;");
    $a->quit;

    my ($c1, $r1) = committed_reclaimable();
    my $log = log_since($log_pos);

    diag(sprintf("T5: c0=%d r0=%d c_mid=%d r_mid=%d c1=%d r1=%d", $c0, $r0, $c_mid, $r_mid, $c1, $r1));

    # See T4's comment: conservation, not an exact baseline/growth match, is
    # the robust invariant once an index has prior undo history.
    is($c1 + $r1, $c_mid + $r_mid,
        "T5: the top-level ROLLBACK moves bytes without changing their sum");
    cmp_ok($r1, '>', $r_mid,
        "T5: the combined growth from both savepoints is credited exactly once (reclaimable rises)");
    unlike($log, qr/accounting underflow/, "T5: no accounting underflow WARNING");
}

# ---------------------------------------------------------------------------
# T6: a worker restart plus a forced reload advances the resident
# generation; a ROLLBACK sampled against the old generation credits
# nothing, and the counter is left at (at least) the fresh reload figure.
# ---------------------------------------------------------------------------
diag("=== T6: stale generation after a worker restart credits nothing ===");
{
    my $old_pid = get_worker_pid('postgres');

    my $a = $node->background_psql('postgres');
    $a->query_safe("BEGIN;");
    $a->query_safe(
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 100);");

    $node->safe_psql('postgres', "SELECT svs_restart_worker('postgres');");
    my $new_pid = wait_for_new_worker_pid('postgres', $old_pid, 60);
    ok($new_pid =~ /^\d+$/ && $new_pid ne $old_pid,
        "T6: worker respawned with a new pid ($old_pid -> $new_pid)");

    # Force the new worker process to actually (re)load the index, so
    # SvsMemoryReconcileLoad advances residentGeneration before the
    # background session's ROLLBACK runs.
    $node->safe_psql('postgres', "SELECT svs_warmup_index('t_idx'::regclass);");
    my ($c_reload, $r_reload) = committed_reclaimable();

    my $log_pos = length($node->log_content());
    $a->query_safe("ROLLBACK;");
    $a->quit;

    my ($c1, $r1) = committed_reclaimable();
    my $log = log_since($log_pos);

    diag(sprintf("T6: c_reload=%d r_reload=%d c1=%d r1=%d", $c_reload, $r_reload, $c1, $r1));

    is($r1, 0, "T6: reclaimable is 0 -- the stale-generation ROLLBACK credited nothing");
    cmp_ok($c1, '>=', $c_reload, "T6: committed is at least the fresh reload measurement");
    unlike($log, qr/accounting underflow/, "T6: no accounting underflow WARNING");
}

# ---------------------------------------------------------------------------
# T7: pg_terminate_backend on the inserting session still runs the FATAL
# abort path and credits the growth, same as an ordinary ROLLBACK.
# ---------------------------------------------------------------------------
diag("=== T7: pg_terminate_backend still credits the growth via the FATAL path ===");
{
    my ($c0, $r0) = committed_reclaimable();

    my $a = $node->background_psql('postgres');
    my $a_pid = $a->query_safe("SELECT pg_backend_pid();");
    chomp $a_pid;

    $a->query_safe("BEGIN;");
    $a->query_safe(
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 100);");

    my ($c_mid, $r_mid) = committed_reclaimable();

    $node->safe_psql('postgres', "SELECT pg_terminate_backend($a_pid);");

    # pg_terminate_backend is asynchronous; poll until reclaimable moves,
    # then take one more sample to let it settle.
    my ($c1, $r1) = ($c_mid, $r_mid);
    for (1 .. 100)
    {
        usleep(200_000);
        ($c1, $r1) = committed_reclaimable();
        last if $r1 > $r_mid;
    }
    usleep(300_000);
    ($c1, $r1) = committed_reclaimable();
    eval { $a->quit };

    diag(sprintf("T7: c0=%d r0=%d c_mid=%d r_mid=%d c1=%d r1=%d", $c0, $r0, $c_mid, $r_mid, $c1, $r1));

    # See T4's comment: conservation, not an exact baseline/growth match, is
    # the robust invariant once an index has prior undo history.
    is($c1 + $r1, $c_mid + $r_mid,
        "T7: the FATAL abort path moves bytes without changing their sum, same as ROLLBACK");
    cmp_ok($r1, '>', $r_mid, "T7: the FATAL abort path credits the growth (reclaimable rises)");
}

# ---------------------------------------------------------------------------
# T10: two interleaved sessions. A inserts, B inserts and commits, A aborts.
# committed+reclaimable is unmoved by A's ROLLBACK (it only moves bytes
# between the two counters); committed never dips below the pre-test
# baseline.
# ---------------------------------------------------------------------------
diag("=== T10: interleaved commit and abort move bytes, not totals ===");
{
    my ($c0, $r0) = committed_reclaimable();

    my $a = $node->background_psql('postgres');
    $a->query_safe("BEGIN;");
    $a->query_safe(
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 50);");

    my $b = $node->background_psql('postgres');
    $b->query_safe("BEGIN;");
    $b->query_safe(
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 50);");
    $b->query_safe("COMMIT;");
    $b->quit;

    my ($c_mid, $r_mid) = committed_reclaimable();

    $a->query_safe("ROLLBACK;");
    $a->quit;

    my ($c1, $r1) = committed_reclaimable();

    diag(sprintf("T10: c0=%d r0=%d c_mid=%d r_mid=%d c1=%d r1=%d", $c0, $r0, $c_mid, $r_mid, $c1, $r1));

    is($c1 + $r1, $c_mid + $r_mid,
        "T10: committed+reclaimable is unchanged by A's abort (bytes move, totals do not)");
    cmp_ok($r1, '>=', $r_mid, "T10: A's abort does not decrease reclaimable");
    # Not a strict committed >= baseline check: SVS's own reported memory can
    # dip by a small amount even with no delete involved once an index has
    # prior undo/consolidate history (see T4's comment), so a tiny dip below
    # c0 here is expected measurement behavior, not a regression. Conservation
    # above is the invariant this fix actually guarantees.
    cmp_ok($c1, '>=', $c0 - 65536,
        "T10: committed stays within one growth block of the pre-test baseline");
}

# ---------------------------------------------------------------------------
# T3: an INSERT-only, non-owner role loops BEGIN/INSERT/ROLLBACK against a
# tightly pinned budget (1MB, matching the original issue's reproduction).
# No loop is refused; reclaimable never exceeds the cap for long; the
# owner's own committed INSERT still succeeds afterward.
# ---------------------------------------------------------------------------
diag("=== T3: INSERT-only attacker cannot exhaust a tightly pinned budget ===");
{
    enroll_fresh_db('attack_db', 50, 1);
    $node->safe_psql('attack_db', "CREATE ROLE attacker LOGIN;");
    $node->safe_psql('attack_db', "GRANT INSERT ON t TO attacker;");

    my $budget_row = $node->safe_psql('postgres',
        "SELECT residency_memory_limit FROM pg_stat_vamana_worker w "
      . "JOIN pg_database d ON d.oid = w.db_oid WHERE d.datname = 'attack_db';");
    chomp $budget_row;
    my $cap = int($budget_row * 0.10);    # svs.compact_threshold_pct default is 10

    my $n_loops = 2000;
    my $batches = 20;
    my $per_batch = $n_loops / $batches;
    my $refused = 0;
    my $max_reclaimable_seen = 0;

    for my $b (1 .. $batches)
    {
        my $sql = '';
        $sql .= "BEGIN; INSERT INTO t (val) VALUES ('[$query_sql]'); ROLLBACK;\n"
          for 1 .. $per_batch;

        my ($rc, $out, $err) = $node->psql('attack_db', $sql,
            extra_params => [ '-U', 'attacker' ], on_error_stop => 0);
        if ($rc != 0 || ($err // '') =~ /residency budget/i)
        {
            $refused = 1;
            diag("T3: batch $b refused: $err");
            last;
        }

        my (undef, $r) = committed_reclaimable('attack_db');
        $max_reclaimable_seen = $r if $r > $max_reclaimable_seen;
    }

    diag(sprintf("T3: cap=%d max_reclaimable_seen=%d refused=%d",
        $cap, $max_reclaimable_seen, $refused));

    ok(!$refused, "T3: no batch of the attack loop was refused for exceeding the residency budget");
    # A generous slack factor over the strict cap: the credit that pushes a
    # database over the cap is itself allowed to land before the COMPACT it
    # triggers finishes (section 4.5's documented bound), and this samples
    # only once per batch, not after every single loop.
    cmp_ok($max_reclaimable_seen, '<', $cap * 3,
        "T3: reclaimable stays within a small bounded multiple of the cap throughout");

    my $owner_ok = eval {
        $node->safe_psql('attack_db', "INSERT INTO t (val) VALUES ('[$query_sql]');");
        1;
    };
    ok($owner_ok, "T3: the owner's own committed INSERT still succeeds afterward");

    # T3b: svs.compact_threshold_pct is PGC_SUSET, so the same INSERT-only
    # attacker cannot widen or disable the database-wide credit cap itself.
    my ($rc_set, $out_set, $err_set) = $node->psql('attack_db',
        "SET svs.compact_threshold_pct = 0;",
        extra_params => [ '-U', 'attacker' ], on_error_stop => 1);
    isnt($rc_set, 0, "T3b: an unprivileged role cannot SET svs.compact_threshold_pct");
    like($err_set, qr/permission denied to set parameter/,
        "T3b: the attempt fails with permission denied to set parameter");

    my $superuser_set_ok = eval {
        $node->safe_psql('attack_db', "SET svs.compact_threshold_pct = 0;");
        1;
    };
    ok($superuser_set_ok, "T3b: the superuser can still SET svs.compact_threshold_pct");
}

# ---------------------------------------------------------------------------
# T8: a rollback larger than the reclaim cap forces a COMPACT. Pinning
# svs.compact_threshold_pct to 100 for one session documents the opt-out:
# no credit is given at all, same as today's (pre-fix) behavior.
# ---------------------------------------------------------------------------
diag("=== T8: a rollback over the cap forces a COMPACT; pct=100 opts out ===");
{
    # 20MB budget (cap = 10% = ~2MB). Empirically against this build (see the
    # session that wrote this test), growing a 50-row seed index by 12,000
    # rows in one transaction reaches ~2.4MB of committed growth -- comfortably
    # over the ~2MB cap, but nowhere near exhausting the 20MB budget itself
    # (which would turn this into a residency-budget ERROR instead of the
    # COMPACT this case is checking for).
    enroll_fresh_db('tightdb', 50, 20);

    my ($c0, $r0) = committed_reclaimable('tightdb');

    my $b1 = $node->background_psql('tightdb');
    $b1->query_safe("BEGIN;");
    $b1->query_safe(
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 12000);");
    my ($c_mid, $r_mid) = committed_reclaimable('tightdb');
    $b1->query_safe("ROLLBACK;");
    $b1->quit;

    my ($c1, $r1) = committed_reclaimable('tightdb');
    diag(sprintf("T8: c0=%d r0=%d c_mid=%d r_mid=%d c1=%d r1=%d", $c0, $r0, $c_mid, $r_mid, $c1, $r1));

    is($r1, 0, "T8: a rollback over the cap triggers its own COMPACT, reclaimable back to 0");
    cmp_ok($c1, '<=', $c0 + 65536,
        "T8: post-compact committed is close to the pre-rollback baseline (within one growth block)");

    # Sub-case: svs.compact_threshold_pct = 100 for this session only, which
    # already means "never compact" for VACUUM; this fix treats it as "never
    # credit either," documenting the opt-out. No need to cross any cap here,
    # so a much smaller batch keeps this sub-case fast.
    my ($c2, $r2) = committed_reclaimable('tightdb');

    my $b2 = $node->background_psql('tightdb');
    $b2->query_safe("SET svs.compact_threshold_pct = 100;");
    $b2->query_safe("BEGIN;");
    $b2->query_safe(
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 3000);");
    my ($c_mid2, undef) = committed_reclaimable('tightdb');
    $b2->query_safe("ROLLBACK;");
    $b2->quit;

    my ($c3, $r3) = committed_reclaimable('tightdb');
    diag(sprintf("T8b: c2=%d c_mid2=%d c3=%d r2=%d r3=%d", $c2, $c_mid2, $c3, $r2, $r3));

    is($r3, $r2, "T8b: with compact_threshold_pct=100, reclaimable is untouched (no credit at all)");
    is($c3, $c_mid2, "T8b: with compact_threshold_pct=100, committed stays at the grown figure");
}

# ---------------------------------------------------------------------------
# T9: a rollback below the cap is reclaimed once a checkpoint compacts the
# index, with no later insert needed -- the rollback's own undo-driven
# DELETE/CONSOLIDATE operations trip the checkpoint themselves.
# ---------------------------------------------------------------------------
diag("=== T9: a below-cap rollback is reclaimed by the next checkpoint ===");
{
    $node->safe_psql('postgres', "ALTER SYSTEM SET svs.checkpoint_operations = 1;");
    $node->safe_psql('postgres', "SELECT pg_reload_conf();");

    my ($c0, $r0) = committed_reclaimable();

    my $a = $node->background_psql('postgres');
    $a->query_safe("BEGIN;");
    $a->query_safe(
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 100);");

    my $log_pos = length($node->log_content());
    $a->query_safe("ROLLBACK;");
    $a->quit;

    $node->wait_for_log(qr/checkpoint complete/, $log_pos);

    my ($c1, $r1) = committed_reclaimable();
    diag(sprintf("T9: c0=%d c1=%d r0=%d r1=%d", $c0, $c1, $r0, $r1));

    is($r1, 0, "T9: the checkpoint's own compaction reconciled reclaimable to 0, no later insert needed");
    cmp_ok($c1, '<=', $c0 + 65536,
        "T9: post-checkpoint committed is close to the pre-rollback baseline (within one growth block)");

    $node->safe_psql('postgres', "ALTER SYSTEM RESET svs.checkpoint_operations;");
    $node->safe_psql('postgres', "SELECT pg_reload_conf();");
}

# ---------------------------------------------------------------------------
# T11: a checkpoint that fails *after* SVSSaveIndex has already compacted
# the live graph (here: VamanaMarkIndexSaved fails, right after SVSSaveIndex
# and the TID-map write both succeeded) must not leave numDeleted/residency
# stale until some unrelated later write happens to self-correct it.
# PerformCheckpoint's PG_CATCH only restores the pre-attempt baseline when
# the compaction itself never ran; otherwise it reconciles with a fresh
# measurement before re-throwing, so the failed checkpoint's own log line is
# enough -- no further insert should be needed to see correct figures.
# ---------------------------------------------------------------------------
diag("=== T11: a checkpoint failing after SVS has already compacted reconciles immediately ===");
SKIP: {
    skip "server not built with --enable-injection-points", 3
        unless ($ENV{enable_injection_points} // 'no') eq 'yes';

    $node->safe_psql('postgres', "CREATE EXTENSION IF NOT EXISTS injection_points;");

    my $a = $node->background_psql('postgres');
    $a->query_safe("BEGIN;");
    $a->query_safe(
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 100);");
    $a->query_safe("ROLLBACK;");
    $a->quit;

    my ($c0, $r0) = committed_reclaimable();
    ok($r0 > 0, "T11: the rollback is credited as reclaimable before the failed checkpoint");

    $node->safe_psql('postgres',
        "SELECT injection_points_attach('vamana-mark-index-saved-error', 'error');");
    $node->safe_psql('postgres', "ALTER SYSTEM SET svs.checkpoint_operations = 1;");
    $node->safe_psql('postgres', "SELECT pg_reload_conf();");

    my $log_pos = length($node->log_content());
    # One more write to trip the now-tiny checkpoint debounce; the checkpoint
    # attempt this triggers is what hits the injection point.
    $node->safe_psql('postgres',
        "INSERT INTO t SELECT ARRAY[$array_sql]::vector FROM generate_series(1, 1);");
    $node->wait_for_log(qr/not checkpointed this cycle, will retry/, $log_pos);

    $node->safe_psql('postgres',
        "SELECT injection_points_detach('vamana-mark-index-saved-error');");
    $node->safe_psql('postgres', "ALTER SYSTEM RESET svs.checkpoint_operations;");
    $node->safe_psql('postgres', "SELECT pg_reload_conf();");

    my ($c1, $r1) = committed_reclaimable();
    diag(sprintf("T11: c0=%d r0=%d c1=%d r1=%d", $c0, $r0, $c1, $r1));

    is($r1, 0,
        "T11: reclaimable is already reconciled to 0 right after the failed checkpoint, no later insert needed");
    cmp_ok($c1, '<=', $c0 + 65536,
        "T11: committed is already close to the compacted baseline right after the failed checkpoint");
}

$node->stop;
done_testing();