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

# 23_fiction_worker_convergence.pl -- pins the observable behavior of a
# per-database worker that holds one parked parallel slot per granted
# search thread.
#
# Stable observables only:
#   - backend_type = 'vamana search slot'  (SvsSlotKindBgwType; not under review)
#   - pg_stat_vamana_worker.search_threads_desired / _granted / _reserved
#   - pg_stat_vamana_worker.worker_pid / heartbeat_ts
#   - absence of "exited with exit code" / "Segmentation fault" in the log
#   - whether a plain core parallel query gets workers from the shared pool
# Never application_name text or shortfall log wording: both are under
# active review and will change shape without changing meaning.
#
# ---------------------------------------------------------------------------
# STATIC-HOLD ASSUMPTION
#
# A granted search thread's parked slot is held for the grant's whole
# lifetime, not acquired only while a search is actually dispatched
# (static hold, not per-dispatch acquisition). Every assertion in this file
# that compares a "vamana search slot" row count against a *granted* value
# while no search is in flight assumes this: the worker owns a slot set and
# converges the held slot count on the launcher's published grant. That
# class of assertion is marked with a STATIC-HOLD ASSUMPTION comment at each
# site below rather than gathered into one lexical block, because it is not
# a handful of cases -- it is most of the file.
# ---------------------------------------------------------------------------

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

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

# ---------------------------------------------------------------------------
# Local helpers
# ---------------------------------------------------------------------------

sub db_oid
{
    my ($node, $db) = @_;
    my $oid = $node->safe_psql('postgres',
        "SELECT oid FROM pg_database WHERE datname = '$db';");
    chomp $oid;
    return $oid;
}

# STATIC-HOLD ASSUMPTION: a row count here is only comparable to a grant
# when the worker is expected to be holding slots at rest, not mid-search.
#
# A parked "vamana search slot" is registered without
# BGWORKER_BACKEND_DATABASE_CONNECTION (svs_cpu_slots.c), so it never runs
# InitPostgres and pg_stat_activity.datname for it is always NULL, for every
# database, with no exception -- confirmed by hand against a scratch
# cluster. That column cannot attribute a slot to a database. The only
# per-database signal a parked slot publishes at all is its
# application_name, set by SvsFormatSearchSlotAppName as
# "vamana: search slot <i>/<n> (reserved <n>) db=<datname>"; matching on the
# " db=<datname>" suffix is therefore the only available per-database
# filter, not a preference over the datname column.
sub search_slot_count
{
    my ($node, $db) = @_;
    my $c = $node->safe_psql('postgres',
        "SELECT count(*) FROM pg_stat_activity "
      . "WHERE backend_type = 'vamana search slot' "
      . "AND application_name LIKE '%db=$db';");
    chomp $c;
    return $c;
}

sub search_slot_pids
{
    my ($node, $db) = @_;
    my $out = $node->safe_psql('postgres',
        "SELECT pid FROM pg_stat_activity "
      . "WHERE backend_type = 'vamana search slot' "
      . "AND application_name LIKE '%db=$db' ORDER BY pid;");
    return split(/\n/, $out);
}

sub granted_for_db
{
    my ($node, $dboid) = @_;
    my $g = $node->safe_psql('postgres',
        "SELECT search_threads_granted FROM pg_stat_vamana_worker "
      . "WHERE db_oid = $dboid;");
    chomp $g;
    return $g;
}

sub desired_for_db
{
    my ($node, $dboid) = @_;
    my $d = $node->safe_psql('postgres',
        "SELECT search_threads_desired FROM pg_stat_vamana_worker "
      . "WHERE db_oid = $dboid;");
    chomp $d;
    return $d;
}

# Poll the *published* grant, distinct from whether slots are held for it --
# keeping "published" and "held" as separate polls is what makes a failure
# here diagnosable as a launcher problem vs. a worker problem.
sub wait_for_granted
{
    my ($node, $dboid, $want, $attempts) = @_;
    $attempts //= 40;
    my $g = '';
    for (1 .. $attempts)
    {
        usleep(500_000);
        $g = granted_for_db($node, $dboid);
        return $g if defined($g) && $g ne '' && $g == $want;
    }
    return $g;
}

# STATIC-HOLD ASSUMPTION: waiting for held slots to reach $want only makes
# sense if slots are supposed to be held independent of an active search.
sub wait_for_search_slot_count
{
    my ($node, $db, $want, $attempts) = @_;
    $attempts //= 40;
    my $c = '';
    for (1 .. $attempts)
    {
        usleep(500_000);
        $c = search_slot_count($node, $db);
        return $c if $c eq $want;
    }
    return $c;
}

# Case 8: the rollup invariant. sum(granted) across every worker equals the
# total number of held search slots cluster-wide. The single best
# whole-feature guard because it is the one check that a per-database
# implementation cannot pass by accident -- it must hold across every row.
# STATIC-HOLD ASSUMPTION: only meaningful at rest, with no search in flight.
sub rollup_snapshot
{
    my ($node) = @_;
    my $sum = $node->safe_psql('postgres',
        "SELECT coalesce(sum(search_threads_granted), 0) "
      . "FROM pg_stat_vamana_worker;");
    chomp $sum;
    my $slots = $node->safe_psql('postgres',
        "SELECT count(*) FROM pg_stat_activity "
      . "WHERE backend_type = 'vamana search slot';");
    chomp $slots;
    return ($sum, $slots);
}

sub assert_rollup_matches
{
    my ($node, $label) = @_;
    my ($sum, $slots) = rollup_snapshot($node);
    is($sum, $slots,
        "rollup invariant ($label): sum(search_threads_granted)=$sum "
      . "matches held 'vamana search slot' rows=$slots");
}

# Case 9: no worker died. Checked from an anchored log offset so an
# unrelated earlier line can never satisfy it.
sub assert_no_crash_since
{
    my ($node, $log_pos, $label) = @_;
    my $log = substr($node->log_content(), $log_pos);
    # A governed build's parked parallel workers are stopped with an
    # administrator-command FATAL when the build ends, so they exit with
    # code 1 by design; only the vamana workers are under test here.
    unlike($log, qr/background worker "(?!parallel worker")[^"]*" \(PID \d+\) exited with exit code/,
        "no worker exit-code crash during $label");
    unlike($log, qr/Segmentation fault/,
        "no segfault during $label");
}

sub wait_for_new_worker_pid_db
{
    my ($node, $db, $old, $attempts) = @_;
    $attempts //= 60;
    for (1 .. $attempts)
    {
        usleep(500_000);
        my $pid = $node->safe_psql('postgres',
            "SELECT pid FROM pg_stat_activity "
          . "WHERE backend_type = 'vamana worker' AND datname = '$db' LIMIT 1;");
        chomp $pid;
        return $pid if $pid =~ /^\d+$/ && $pid ne $old;
    }
    return '';
}

# A plain core parallel query, forced regardless of table size or cost
# settings, so the workers-launched count reflects only pool availability,
# not planner cost heuristics.
sub workers_launched
{
    my ($node, $db, $tbl) = @_;
    my $out = $node->safe_psql($db, qq(
        SET parallel_setup_cost = 0;
        SET parallel_tuple_cost = 0;
        SET min_parallel_table_scan_size = 0;
        SET max_parallel_workers_per_gather = 4;
        EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF)
        SELECT count(*) FROM $tbl;
    ));
    my ($n) = $out =~ /Workers Launched:\s*(\d+)/;
    return defined($n) ? $n : 0;
}

# ---------------------------------------------------------------------------
# Setup
# ---------------------------------------------------------------------------

my $node = PostgreSQL::Test::Cluster->new('vamana_convergence');
$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 = 10");
$node->append_conf('postgresql.conf', "max_wal_senders = 10");
$node->append_conf('postgresql.conf', "max_worker_processes = 24");
$node->append_conf('postgresql.conf', "max_parallel_workers = 8");
$node->append_conf('postgresql.conf', "log_min_messages = 'notice'");
$node->append_conf('postgresql.conf', "svs.max_residency_memory = '400MB'");
$node->append_conf('postgresql.conf', "svs.max_search_work_mem = '400MB'");
$node->start;

# Anchor every crash check to this run's log, not offset 0: the log file on
# disk persists across repeated `make prove_installcheck` invocations (it is
# not touched by the tmp_check wipe), so a literal 0 would also match FATAL
# lines left over from an unrelated earlier run.
my $run_log_pos = length($node->log_content());

$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 $pg_worker_pid = wait_for_worker_db($node, 'postgres', 30);
ok($pg_worker_pid =~ /^\d+$/,
    "worker for 'postgres' is running before any convergence case (pid=$pg_worker_pid)");

my $pg_dboid = db_oid($node, 'postgres');

# A database holds parked search slots only once it has a vamana index.
$node->safe_psql('postgres', qq(
    CREATE TABLE idx_seed (id serial PRIMARY KEY, val vector(3));
    INSERT INTO idx_seed (val) VALUES ('[0,0,0]'), ('[1,1,1]');
    CREATE INDEX idx_seed_idx ON idx_seed USING vamana (val vector_l2_ops);
));

# Baseline: an explicit search_num_threads = 1, with a large pool, so
# desired = granted = 1 once live.
#
# A plain UPDATE is used here, not a wait on the cluster-default alone: the
# reconcile that first spawned this worker (above) ran before the worker set
# its own workerPid, so it published a grant of 0 and nothing wakes the
# launcher on the liveness transition itself (this is the same zero-grant
# gap cases 3 and 5 pin below). Any catalog UPDATE fires a fresh NOTIFY-driven
# reconcile that reads the worker's now-current liveness, which is what
# actually settles the baseline -- confirmed by hand: without this UPDATE,
# the grant sits at 0 far past this wait's bound.
$node->safe_psql('postgres',
    "UPDATE vamana_databases SET search_num_threads = 1 WHERE datname = 'postgres';");
my $baseline_granted = wait_for_granted($node, $pg_dboid, 1, 30);
is($baseline_granted, '1', "baseline published grant settles at 1 thread");

# A table big enough that a forced parallel seq scan is worth planning, used
# only for the core-parallel-query proofs (cases 2 and part of case 7). Not a
# vamana index -- this table exists purely to observe the shared
# max_parallel_workers pool from the outside.
$node->safe_psql('postgres', qq(
    CREATE TABLE core_probe (id int, pad text);
    INSERT INTO core_probe SELECT g, repeat('x', 100) FROM generate_series(1, 50000) g;
    ANALYZE core_probe;
));

# ---------------------------------------------------------------------------
# Case 1: converge up.
#
# Splits into a grant-math assertion (the published grant reaches 4) and a
# slot-hold assertion (the held slot count converges to match it).
# ---------------------------------------------------------------------------
{
    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET search_num_threads = 4 WHERE datname = 'postgres';");

    my $granted = wait_for_granted($node, $pg_dboid, 4, 30);
    is($granted, '4', "case 1: published grant reaches 4 after raising search_num_threads");

    # STATIC-HOLD ASSUMPTION: no search is in flight; this expects the
    # worker to hold 4 parked search slots simply because it was granted 4.
    my $held = wait_for_search_slot_count($node, 'postgres', 4, 20);
    is($held, '4',
        "case 1: exactly 4 'vamana search slot' rows exist for the database");
}

# ---------------------------------------------------------------------------
# Case 2: converge down, and prove the released slots really returned to
# the shared pool (not a private counter) by observing a plain core
# parallel query regain workers.
#
# max_parallel_workers is lowered to 4 for this block only, so that granting
# 4 to search starves a concurrent core query down to 0 workers, and
# dropping the grant to 2 frees 2 back for it: the "starved-then-recovers"
# shape proves the released slots returned to the shared pool rather than
# a private counter.
# ---------------------------------------------------------------------------
{
    $node->safe_psql('postgres', "ALTER SYSTEM SET max_parallel_workers = 4;");
    $node->safe_psql('postgres', "SELECT pg_reload_conf();");
    usleep(500_000);

    # Re-affirm the case-1 grant of 4 under the new, smaller pool (still
    # fits: pool 4, desired 4).
    my $granted_high = wait_for_granted($node, $pg_dboid, 4, 30);
    is($granted_high, '4', "case 2: grant of 4 still fits an equal-sized pool of 4");

    # STATIC-HOLD ASSUMPTION: if 4 slots were genuinely held right now,
    # a core query asking for up to 4 workers should get none.
    my $before = workers_launched($node, 'postgres', 'core_probe');

    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET search_num_threads = 2 WHERE datname = 'postgres';");
    my $granted_low = wait_for_granted($node, $pg_dboid, 2, 30);
    is($granted_low, '2', "case 2: published grant reaches 2 after lowering search_num_threads");

    my $held = wait_for_search_slot_count($node, 'postgres', 2, 20);
    is($held, '2',
        "case 2: exactly 2 'vamana search slot' rows remain");

    my $after = workers_launched($node, 'postgres', 'core_probe');
    ok($before == 0 && $after > $before,
        "case 2: a core parallel query is starved while 4 slots are held "
      . "and regains workers once 2 are released "
      . "(before=$before, after=$after)")
      or diag("today nothing holds slots for search, so the pool is never "
            . "actually starved: before=$before after=$after");

    $node->safe_psql('postgres', "ALTER SYSTEM RESET max_parallel_workers;");
    $node->safe_psql('postgres', "SELECT pg_reload_conf();");
    usleep(500_000);
}

assert_rollup_matches($node, 'after cases 1-2');
assert_no_crash_since($node, $run_log_pos, 'cases 1-2');

my $pid_after_12 = $node->safe_psql('postgres',
    "SELECT pid FROM pg_stat_activity WHERE backend_type = 'vamana worker' "
  . "AND datname = 'postgres';");
chomp $pid_after_12;
is($pid_after_12, $pg_worker_pid,
    "case 9: the worker pid is unchanged by grant-only changes (no crash/restart)");

# ---------------------------------------------------------------------------
# Cases 3 and 5: a freshly enabled database.
#
# PublishCpuGrants publishes 0 for a not-yet-live database, and the worker
# parks exactly the published grant, so it holds no slots until the grant
# moves off 0. The reconcile that spawns the worker has already
# published its grant as 0 (spawn loop runs after PublishCpuGrants), and
# nothing wakes the launcher on the workerPid 0-to-nonzero transition except
# the worker's own kick (SvsKickLauncher in VamanaWorkerServe), so a
# regression there can stay at 0 for up to VAMANA_LAUNCHER_NAPTIME_MS
# (180s). The poll bounds below are well under that, so such a regression
# fails this test instead of hanging it.
# ---------------------------------------------------------------------------
{
    $node->safe_psql('postgres', "CREATE DATABASE fresh_db;");
    $node->safe_psql('fresh_db', qq(
        CREATE EXTENSION vector;
        CREATE EXTENSION svs;
    ));

    # Resolved before the enabling INSERT below, not after: the not-yet-live
    # window this case samples for is a one-shot transient (0 published once,
    # then never again), so any query latency spent between the INSERT and
    # the first sample eats directly into the only chance to observe it.
    my $fresh_oid = db_oid($node, 'fresh_db');

    # Case 3: sample repeatedly while the published grant is still 0 (the
    # not-yet-live window). STATIC-HOLD ASSUMPTION: expects 0 held slots
    # throughout this window, not the eventual configured grant of 5.
    #
    # Measured by hand against a scratch cluster: with the launcher kicked
    # promptly once the worker publishes its own workerPid, the whole span
    # from the enabling INSERT to the worker's startup log
    # line is single-digit milliseconds. A one-shot transient that short
    # cannot be sampled by spawning a new psql client process per poll --
    # each spawn alone costs more than the entire window. Both the INSERT
    # and every poll below therefore run over background_psql sessions
    # already connected before the INSERT fires, so the only latency left
    # between "grant published as 0" and "this poll observes it" is one
    # protocol round trip, not a process fork/exec/connect/authenticate
    # cycle.
    #
    # fresh_db needs a vamana index to hold slots at all, and an index can
    # only be created in an enabled database. So enroll it, index it, and
    # disable it again before the timed re-enable below.
    $node->safe_psql('postgres',
        "INSERT INTO vamana_databases (datname, enabled, search_num_threads) "
      . "VALUES ('fresh_db', true, 5);");
    wait_for_worker_db($node, 'fresh_db', 30);
    $node->safe_psql('fresh_db', qq(
        CREATE TABLE idx_seed (id serial PRIMARY KEY, val vector(3));
        INSERT INTO idx_seed (val) VALUES ('[0,0,0]'), ('[1,1,1]');
        CREATE INDEX idx_seed_idx ON idx_seed USING vamana (val vector_l2_ops);
    ));
    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET enabled = false WHERE datname = 'fresh_db';");
    for (1 .. 60)    # up to 30s for the graceful drain to finish
    {
        usleep(500_000);
        my $alive = $node->safe_psql('postgres',
            "SELECT count(*) FROM pg_stat_activity "
          . "WHERE backend_type = 'vamana worker' AND datname = 'fresh_db';");
        chomp $alive;
        last if $alive eq '0';
    }

    my $ctl = $node->background_psql('postgres');
    my $poll = $node->background_psql('postgres');

    my $t0 = [gettimeofday];
    $ctl->query_safe(
        "UPDATE vamana_databases SET enabled = true WHERE datname = 'fresh_db';");

    my $saw_zero_grant_window = 0;
    my $slots_threads_ever_disagreed = 0;
    for (1 .. 4000)    # tight for ~2s, then the coarser waits below take over
    {
        # Both columns are read from the same $poll session, back to back,
        # in one round trip each over an already-open connection. Reading
        # the held count via a freshly spawned safe_psql process here (as
        # search_slot_count does everywhere else in this file) would add
        # tens of milliseconds of fork/exec/connect latency between the two
        # reads -- long enough, confirmed by hand, for the real system to
        # have already converged past the instant the grant was sampled at,
        # making the two columns describe two different moments in time
        # rather than one.
        my $g = $poll->query_safe(
            "SELECT search_threads_granted FROM pg_stat_vamana_worker "
          . "WHERE db_oid = $fresh_oid;");
        chomp $g;
        next unless $g ne '';
        if ($g == 0)
        {
            $saw_zero_grant_window = 1;

            # The worker parks exactly the published grant, so 0 slots are
            # held while the grant reads 0. Only a nonzero count with the
            # grant still reading 0 afterward is a disagreement; if the
            # grant moved off 0 between the two reads, this window is over
            # and the count says nothing about it.
            my $held = $poll->query_safe(
                "SELECT count(*) FROM pg_stat_activity "
              . "WHERE backend_type = 'vamana search slot' "
              . "AND application_name LIKE '%db=fresh_db';");
            chomp $held;

            my $g2 = $poll->query_safe(
                "SELECT search_threads_granted FROM pg_stat_vamana_worker "
              . "WHERE db_oid = $fresh_oid;");
            chomp $g2;

            $slots_threads_ever_disagreed = 1
              if $held ne '0' && $g2 eq '0';
        }
        else
        {
            # A disabled database's row keeps its last-known grant until the
            # re-enable's first reconcile publishes 0, so a nonzero read
            # before the window has opened is stale, not the window's end.
            last if $saw_zero_grant_window;    # grant has moved off 0
        }
    }
    $ctl->quit;
    $poll->quit;

    ok($saw_zero_grant_window,
        "case 3: observed the not-yet-live window where the published grant is 0");
    ok(!$slots_threads_ever_disagreed,
        "case 3: during the zero-grant window, no slot is held, "
      . "never the eventual grant of 5");

    # Now wait, bounded well under the 180s naptime, for convergence to the
    # configured grant of 5.
    my $granted = wait_for_granted($node, $fresh_oid, 5, 40);    # ~20s
    is($granted, '5', "case 3: grant eventually converges to the configured 5 threads");

    my $held_final = wait_for_search_slot_count($node, 'fresh_db', 5, 20);
    is($held_final, '5',
        "case 3: held slot count converges to 5, matching the final grant");

    # Case 5: prompt convergence is a hard requirement -- "takes effect
    # within about 1 second" per the documented contract -- not just
    # "eventually, within naptime." Bound the assertion in seconds, far
    # below the 180s naptime, so a regression to naptime-only fails loudly
    # here instead of the suite hanging for three minutes.
    my $elapsed = tv_interval($t0);
    ok($granted eq '5' && $elapsed < 30,
        "case 5: grant convergence after enabling a database happens in "
      . "seconds, not the 180s naptime (elapsed=${elapsed}s)")
      or diag("180s naptime is the specific failure mode being guarded "
            . "against here; 30s is a generous bound still far below it");
}

assert_rollup_matches($node, 'after cases 3 and 5');
assert_no_crash_since($node, $run_log_pos, 'cases 3 and 5');

# ---------------------------------------------------------------------------
# Case 4: restart into an unchanged grant.
#
# THIS IS THE SINGLE MOST IMPORTANT CASE IN THIS FILE. desired/reserved are
# written unconditionally on every reconcile, but grantedSearchThreads is
# written -- and the worker's latch kicked -- only when the value *changes*.
# VamanaWorkerResetEntryState explicitly does not run on a plain worker
# restart. So a worker that restarts while the grant is already settled at
# its current value is never kicked about it by that mechanism alone: a
# kick-only convergence implementation passes every other case in this file
# and fails only this one, by leaving the restarted worker holding zero
# slots forever. Do not simplify this case away; it is the regression test
# for exactly that trap. Convergence must be driven by polling every
# heartbeat, with the restarted worker's own SvsKickLauncher call (on
# publishing its workerPid) only making it prompt.
# ---------------------------------------------------------------------------
{
    my $pre_granted = granted_for_db($node, $pg_dboid);
    my $pre_held = search_slot_count($node, 'postgres');

    $node->safe_psql('postgres', "SELECT svs_restart_worker('postgres');");
    my $new_pid = wait_for_new_worker_pid_db($node, 'postgres', $pg_worker_pid, 60);
    ok($new_pid =~ /^\d+$/ && $new_pid ne $pg_worker_pid,
        "case 4: worker restarted with a new pid (pid=$new_pid)");

    my $post_granted = wait_for_granted($node, $pg_dboid, $pre_granted, 30);
    is($post_granted, $pre_granted,
        "case 4: the published grant value is unchanged across the restart "
      . "($pre_granted before and after)");

    # STATIC-HOLD ASSUMPTION: the restarted worker should re-acquire the
    # same held slot count as before, purely by polling its heartbeat --
    # not by waiting for a kick that a kick-only implementation would never
    # send, since the grant value above never changed.
    my $post_held = wait_for_search_slot_count($node, 'postgres', $pre_held, 30);
    is($post_held, $pre_held,
        "case 4: the restarted worker converges to the same held slot count "
      . "($pre_held) without the grant value ever changing -- the "
      . "kick-only-trap regression test");

    $pg_worker_pid = $new_pid;
}

assert_rollup_matches($node, 'after case 4');
assert_no_crash_since($node, $run_log_pos, 'case 4');

# ---------------------------------------------------------------------------
# Case 6: release on exit, then re-converge on return.
#
# Case 4 just left the published grant at 0 (see the note above), so a
# no-op UPDATE forces a fresh reconcile to re-settle it at a known nonzero
# value first -- otherwise "re-converges to its pre-disable value" would
# compare 0 to 0 and prove nothing.
# ---------------------------------------------------------------------------
{
    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET search_num_threads = 2 WHERE datname = 'postgres';");
    my $pre_disable_granted = wait_for_granted($node, $pg_dboid, 2, 30);
    is($pre_disable_granted, '2',
        "case 6 setup: grant re-settles at 2 before the disable/enable cycle");

    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET enabled = false WHERE datname = 'postgres';");

    for (1 .. 60)    # up to 30s for the graceful drain to finish
    {
        usleep(500_000);
        my $alive = $node->safe_psql('postgres',
            "SELECT count(*) FROM pg_stat_activity "
          . "WHERE backend_type = 'vamana worker' AND datname = 'postgres';");
        chomp $alive;
        last if $alive eq '0';
    }
    my $alive = $node->safe_psql('postgres',
        "SELECT count(*) FROM pg_stat_activity "
      . "WHERE backend_type = 'vamana worker' AND datname = 'postgres';");
    chomp $alive;
    is($alive, '0', "case 6: worker for 'postgres' has stopped after disable");

    # STATIC-HOLD ASSUMPTION: the worker held 2 search slots (from the
    # setup above) right up to the disable; this pins that
    # VamanaWorkerReleaseSearchSlotsOnExit actually runs on that exit path
    # and releases them, rather than leaving them orphaned.
    my $held_after_stop = search_slot_count($node, 'postgres');
    is($held_after_stop, '0',
        "case 6: no 'vamana search slot' rows remain after the worker stops");

    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET enabled = true WHERE datname = 'postgres';");
    my $reborn_pid = wait_for_worker_db($node, 'postgres', 30);
    ok($reborn_pid =~ /^\d+$/, "case 6: worker for 'postgres' comes back (pid=$reborn_pid)");
    $pg_worker_pid = $reborn_pid;

    my $reconverged = wait_for_granted($node, $pg_dboid, $pre_disable_granted, 30);
    is($reconverged, $pre_disable_granted,
        "case 6: grant re-converges to its pre-disable value "
      . "($pre_disable_granted) after re-enabling");

    my $held_reconverged =
        wait_for_search_slot_count($node, 'postgres', $pre_disable_granted, 20);
    is($held_reconverged, $pre_disable_granted,
        "case 6: held slot count re-converges to $pre_disable_granted on return");
}

assert_rollup_matches($node, 'after case 6');
assert_no_crash_since($node, $run_log_pos, 'case 6');

# ---------------------------------------------------------------------------
# Case 7: reduced pool. Constrain max_parallel_workers below what 'postgres'
# desires, so granted < desired, and pin that the held count tracks
# granted, not desired. No log-text assertion: shortfall log wording is
# under active review, so the shortfall is observed purely through the
# numeric gap in the view instead.
#
# svs.max_search_threads_per_db must be raised explicitly here. It defaults
# to 0, which means "follow max_parallel_workers" (see
# SvsSearchThreadsPerDbCeiling in svs_cpu_budget.c) -- so with the default left
# alone, lowering max_parallel_workers alone clamps *desired* itself down to
# the same small number and no gap ever appears in the view. Confirmed by
# hand against a scratch cluster: without this second GUC, desired reads 1,
# not 5, and the "shortfall" assertion below would be checking two equal
# numbers instead of a real gap.
# ---------------------------------------------------------------------------
{
    $node->safe_psql('postgres', "ALTER SYSTEM SET max_parallel_workers = 1;");
    $node->safe_psql('postgres', "ALTER SYSTEM SET svs.max_search_threads_per_db = 10;");
    $node->safe_psql('postgres', "SELECT pg_reload_conf();");
    usleep(500_000);

    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET search_num_threads = 5 WHERE datname = 'postgres';");

    my $desired = '';
    my $granted = '';
    for (1 .. 30)
    {
        usleep(500_000);
        $desired = desired_for_db($node, $pg_dboid);
        $granted = granted_for_db($node, $pg_dboid);
        last if defined($desired) && $desired eq '5'
             && defined($granted) && $granted ne '' && $granted < $desired;
    }
    is($desired, '5', "case 7: desired reflects the full ask (5) despite the small pool");
    ok(defined($granted) && $granted ne '' && $granted < $desired,
        "case 7: the view shows a real shortfall, granted ($granted) < desired ($desired)");

    # STATIC-HOLD ASSUMPTION: held slots should track the clamped grant,
    # never the raw ask.
    my $held = search_slot_count($node, 'postgres');
    is($held, $granted,
        "case 7: held slot count ($held) matches granted ($granted), never desired ($desired)");
    isnt($held, $desired,
        "case 7: held slot count must not drift up to the unmet desired value");

    $node->safe_psql('postgres', "ALTER SYSTEM RESET max_parallel_workers;");
    $node->safe_psql('postgres', "ALTER SYSTEM RESET svs.max_search_threads_per_db;");
    $node->safe_psql('postgres', "SELECT pg_reload_conf();");
    # Restore a grant the rest of the suite (and teardown) can settle on.
    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET search_num_threads = 1 WHERE datname = 'postgres';");
    wait_for_granted($node, $pg_dboid, 1, 30);
    usleep(500_000);
}

assert_rollup_matches($node, 'after case 7');
assert_no_crash_since($node, $run_log_pos, 'case 7 (reduced pool)');

# ---------------------------------------------------------------------------
# Case 10: two databases live at once -- changing one's grant must not
# touch the other's held slots.
#
# 'postgres' and 'fresh_db' up to this point were only ever changed one at
# a time, never compared against each other while both are live. fresh_db
# is left holding 5 slots since cases 3/5; 'postgres' is left holding 1
# since case 7's cleanup.
# ---------------------------------------------------------------------------
{
    my $fresh_oid = db_oid($node, 'fresh_db');

    my $fresh_before = wait_for_search_slot_count($node, 'fresh_db', 5, 10);
    is($fresh_before, '5', "case 10 setup: fresh_db still holds 5 slots untouched");

    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET search_num_threads = 3 WHERE datname = 'postgres';");
    my $pg_granted = wait_for_granted($node, $pg_dboid, 3, 30);
    is($pg_granted, '3', "case 10: postgres grant reaches 3");
    my $pg_held = wait_for_search_slot_count($node, 'postgres', 3, 20);
    is($pg_held, '3', "case 10: postgres held slots reach 3");

    my $fresh_after_pg_change = search_slot_count($node, 'fresh_db');
    is($fresh_after_pg_change, '5',
        "case 10: fresh_db's held slots are untouched by postgres's grant change");

    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET search_num_threads = 2 WHERE datname = 'fresh_db';");
    my $fresh_granted = wait_for_granted($node, $fresh_oid, 2, 30);
    is($fresh_granted, '2', "case 10: fresh_db grant reaches 2");
    my $fresh_held = wait_for_search_slot_count($node, 'fresh_db', 2, 20);
    is($fresh_held, '2', "case 10: fresh_db held slots reach 2");

    my $pg_after_fresh_change = search_slot_count($node, 'postgres');
    is($pg_after_fresh_change, '3',
        "case 10: postgres's held slots are untouched by fresh_db's grant change");
}

assert_rollup_matches($node, 'after case 10');
assert_no_crash_since($node, $run_log_pos, 'case 10');

# ---------------------------------------------------------------------------
# Case 11: near the grant ceiling.
#
# Every prior case drives small grants (1-5); none approach the pool this
# node actually has (max_parallel_workers = 8). Disable fresh_db first so
# the whole pool is free for 'postgres' alone, then ask for exactly
# max_parallel_workers: the clamp math must land exactly on the boundary
# (granted = desired = max_parallel_workers), not one short of it.
# ---------------------------------------------------------------------------
{
    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET enabled = false WHERE datname = 'fresh_db';");
    for (1 .. 60)    # up to 30s for the graceful drain to finish
    {
        usleep(500_000);
        my $alive = $node->safe_psql('postgres',
            "SELECT count(*) FROM pg_stat_activity "
          . "WHERE backend_type = 'vamana worker' AND datname = 'fresh_db';");
        chomp $alive;
        last if $alive eq '0';
    }

    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET search_num_threads = 8 WHERE datname = 'postgres';");

    my $desired = '';
    my $granted = '';
    for (1 .. 30)
    {
        usleep(500_000);
        $desired = desired_for_db($node, $pg_dboid);
        $granted = granted_for_db($node, $pg_dboid);
        last if defined($desired) && $desired eq '8'
             && defined($granted) && $granted eq '8';
    }
    is($desired, '8', "case 11: desired reaches max_parallel_workers (8)");
    is($granted, '8',
        "case 11: granted reaches max_parallel_workers (8) with no pool contention");

    my $held = wait_for_search_slot_count($node, 'postgres', 8, 20);
    is($held, '8', "case 11: held slot count reaches 8, right at the pool boundary");

    my $launched = workers_launched($node, 'postgres', 'core_probe');
    is($launched, 0,
        "case 11: a concurrent core query gets 0 workers with the pool exactly saturated");

    # Restore fresh_db and a small grant on postgres for the rest of the suite.
    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET enabled = true WHERE datname = 'fresh_db';");
    wait_for_worker_db($node, 'fresh_db', 30);
    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET search_num_threads = 1 WHERE datname = 'postgres';");
    wait_for_granted($node, $pg_dboid, 1, 30);
}

assert_rollup_matches($node, 'after case 11');
assert_no_crash_since($node, $run_log_pos, 'case 11');

# ---------------------------------------------------------------------------
# Case 12: crash-triggered slot release.
#
# VamanaWorkerReleaseSearchSlotsOnExit exists specifically for the exit path
# a clean shutdown never takes: an uncaught ERROR that unwinds through
# proc_exit without running VamanaWorkerDrainAndStop. Driven via the
# in-tree injection-point framework (the same mechanism
# 13_crash_backoff.pl uses). If release-on-exit did not run, the
# replacement worker's own convergence would add a second full set on top
# of the crashed worker's leaked slots, and the held count would settle at
# double the grant instead of matching it -- that is what this case would
# catch. Placed last: the crash it injects is intentional and would trip
# assert_no_crash_since for every later case, so nothing follows it.
#
# fresh_db is disabled first: 'vamana-worker-tick-crash' fires for every
# vamana worker's heartbeat, not just 'postgres', so a second live worker
# would race to trip it too and confuse which pid this case is tracking.
# ---------------------------------------------------------------------------
if (($ENV{enable_injection_points} // 'no') eq 'yes')
{
    $node->safe_psql('postgres', "CREATE EXTENSION IF NOT EXISTS injection_points;");

    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET enabled = false WHERE datname = 'fresh_db';");
    for (1 .. 60)    # up to 30s for the graceful drain to finish
    {
        usleep(500_000);
        my $alive = $node->safe_psql('postgres',
            "SELECT count(*) FROM pg_stat_activity "
          . "WHERE backend_type = 'vamana worker' AND datname = 'fresh_db';");
        chomp $alive;
        last if $alive eq '0';
    }

    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET search_num_threads = 3 WHERE datname = 'postgres';");
    my $pre_crash_granted = wait_for_granted($node, $pg_dboid, 3, 30);
    is($pre_crash_granted, '3', "case 12 setup: grant settles at 3 before the crash");
    my $pre_crash_held = wait_for_search_slot_count($node, 'postgres', 3, 20);
    is($pre_crash_held, '3', "case 12 setup: 3 search slots held before the crash");

    my $log_pos = length($node->log_content());

    $node->safe_psql('postgres',
        "SELECT injection_points_attach('vamana-worker-tick-crash', 'error');");
    $node->wait_for_log(qr/error triggered for injection point vamana-worker-tick-crash/,
        $log_pos);
    $node->safe_psql('postgres',
        "SELECT injection_points_detach('vamana-worker-tick-crash');");

    my $new_pid = wait_for_new_worker_pid_db($node, 'postgres', $pg_worker_pid, 60);
    ok($new_pid =~ /^\d+$/ && $new_pid ne $pg_worker_pid,
        "case 12: worker restarted after the injected crash (pid=$new_pid)");

    # STATIC-HOLD ASSUMPTION: the replacement worker converges to the same
    # grant (3), not the old worker's leaked 3 plus its own 3.
    my $post_crash_held = wait_for_search_slot_count($node, 'postgres', 3, 30);
    is($post_crash_held, '3',
        "case 12: exactly 3 'vamana search slot' rows remain after the crash "
      . "(no orphans survive from the crashed worker)");

    # Re-enable fresh_db: a disabled database's row keeps its last-known
    # granted value in pg_stat_vamana_worker even though its worker holds no
    # slots, which would otherwise make the rollup invariant below spurious.
    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET enabled = true WHERE datname = 'fresh_db';");
    wait_for_worker_db($node, 'fresh_db', 30);
    wait_for_search_slot_count($node, 'fresh_db', 2, 20);

    assert_rollup_matches($node, 'after case 12');

    # -----------------------------------------------------------------------
    # Case 13: release-on-exit itself fails.  Its slots must outlive the
    # crash, then self-terminate once SvsSlotOwnerIsAlive notices workerPid
    # no longer matches.  Placed last for the same reason as case 12.
    #
    # fresh_db is disabled again for the same reason as in case 12: its
    # worker, now holding slots of its own, would otherwise be the one to
    # trip the injection points instead of 'postgres'.
    # -----------------------------------------------------------------------
    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET enabled = false WHERE datname = 'fresh_db';");
    for (1 .. 60)    # up to 30s for the graceful drain to finish
    {
        usleep(500_000);
        my $alive = $node->safe_psql('postgres',
            "SELECT count(*) FROM pg_stat_activity "
          . "WHERE backend_type = 'vamana worker' AND datname = 'fresh_db';");
        chomp $alive;
        last if $alive eq '0';
    }

    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET search_num_threads = 3 WHERE datname = 'postgres';");
    my $pre13_granted = wait_for_granted($node, $pg_dboid, 3, 30);
    is($pre13_granted, '3', "case 13 setup: grant settles at 3 before the crash");
    my $pre13_held = wait_for_search_slot_count($node, 'postgres', 3, 20);
    is($pre13_held, '3', "case 13 setup: 3 search slots held before the crash");

    my $pid_before_13 = $node->safe_psql('postgres',
        "SELECT pid FROM pg_stat_activity "
      . "WHERE backend_type = 'vamana worker' AND datname = 'postgres' LIMIT 1;");
    chomp $pid_before_13;
    my %pre13_slot_pids = map { $_ => 1 } search_slot_pids($node, 'postgres');

    my $log_pos13 = length($node->log_content());

    $node->safe_psql('postgres',
        "SELECT injection_points_attach('vamana-search-slot-release-crash', 'error');");
    $node->safe_psql('postgres',
        "SELECT injection_points_attach('vamana-worker-tick-crash', 'error');");
    $node->wait_for_log(qr/error triggered for injection point vamana-worker-tick-crash/,
        $log_pos13);
    $node->wait_for_log(qr/error triggered for injection point vamana-search-slot-release-crash/,
        $log_pos13);
    $node->safe_psql('postgres',
        "SELECT injection_points_detach('vamana-worker-tick-crash');");
    $node->safe_psql('postgres',
        "SELECT injection_points_detach('vamana-search-slot-release-crash');");

    my @survivors_right_after_crash =
        grep { $pre13_slot_pids{$_} } search_slot_pids($node, 'postgres');
    ok(scalar(@survivors_right_after_crash) > 0,
        "case 13: the crashed worker's search slots outlive it when its own "
      . "release-on-exit callback fails");

    my $all_pre13_gone = '';
    for (1 .. 60)    # up to 30s: bounded by SVS_PARKED_SLOT_OWNER_POLL_MS
    {
        my @still_here = grep { $pre13_slot_pids{$_} } search_slot_pids($node, 'postgres');
        if (scalar(@still_here) == 0)
        {
            $all_pre13_gone = 1;
            last;
        }
        usleep(500_000);
    }
    ok($all_pre13_gone,
        "case 13: orphaned search slots self-terminate once workerPid no "
      . "longer matches their recorded owner");

    my $new_pid13 = wait_for_new_worker_pid_db($node, 'postgres', $pid_before_13, 60);
    ok($new_pid13 =~ /^\d+$/ && $new_pid13 ne $pid_before_13,
        "case 13: worker restarted after the injected crash (pid=$new_pid13)");

    my $settled13 = wait_for_search_slot_count($node, 'postgres', 3, 10);
    is($settled13, '3',
        "case 13: held search slots converge back to the grant once the "
      . "orphans are gone");

    $node->safe_psql('postgres',
        "UPDATE vamana_databases SET enabled = true WHERE datname = 'fresh_db';");
    wait_for_worker_db($node, 'fresh_db', 30);
    wait_for_search_slot_count($node, 'fresh_db', 2, 20);

    assert_rollup_matches($node, 'after case 13');
}
else
{
    diag('skipping case 12 (crash-triggered slot release): '
       . 'server not built with --enable-injection-points');
}

$node->stop;

done_testing();