fix: index slot recovery lookups by LSN

[psteinroe]

add a partitioned BRIN index on pgstream.events(lsn) so invalidated-slot recovery is fast.

chose brin because writes are cheap, lsns closely follow physical row order and in most recovery scenarios the filtered set is small enough for a cheap top-n sort by (created_at, id).

Production rollout

run the following script before deploying the new version. do not wrap it in a transaction.

\set ON_ERROR_STOP on

-- A valid parent means this rollout already completed.
SELECT coalesce((
    SELECT index_metadata.indisvalid
    FROM pg_index AS index_metadata
    WHERE index_metadata.indexrelid = to_regclass('pgstream.events_lsn_brin_idx')
), false) AS rollout_complete
\gset

\if :rollout_complete
    \echo 'pgstream.events_lsn_brin_idx is already valid; nothing to do.'
\else
    SELECT pg_advisory_lock(hashtext('pgstream.events_lsn_brin_idx'));

    -- Remove invalid standalone indexes left by an interrupted build.
    SELECT format('DROP INDEX CONCURRENTLY %I.%I;', namespace.nspname, child_index.relname)
    FROM pg_partition_tree('pgstream.events'::regclass) AS partition_tree
    JOIN pg_class AS partition ON partition.oid = partition_tree.relid
    JOIN pg_namespace AS namespace ON namespace.oid = partition.relnamespace
    JOIN pg_class AS child_index
      ON child_index.relnamespace = namespace.oid
     AND child_index.relname = partition.relname || '_lsn_brin_idx'
    JOIN pg_index AS child_metadata ON child_metadata.indexrelid = child_index.oid
    WHERE partition_tree.isleaf
      AND child_metadata.indrelid = partition.oid
      AND NOT child_metadata.indisvalid
      AND NOT EXISTS (
          SELECT 1 FROM pg_inherits WHERE inhrelid = child_index.oid
      )
    ORDER BY partition.relname
    \gexec

    -- Build one leaf index at a time without blocking writes.
    SELECT format(
        'CREATE INDEX CONCURRENTLY %I ON %I.%I USING brin (lsn) WITH (pages_per_range = 64, autosummarize = on);',
        partition.relname || '_lsn_brin_idx', namespace.nspname, partition.relname
    )
    FROM pg_partition_tree('pgstream.events'::regclass) AS partition_tree
    JOIN pg_class AS partition ON partition.oid = partition_tree.relid
    JOIN pg_namespace AS namespace ON namespace.oid = partition.relnamespace
    LEFT JOIN pg_class AS expected_index
      ON expected_index.relnamespace = namespace.oid
     AND expected_index.relname = partition.relname || '_lsn_brin_idx'
    WHERE partition_tree.isleaf
      AND expected_index.oid IS NULL
      AND NOT EXISTS (
          SELECT 1
          FROM pg_inherits AS attachment
          JOIN pg_index AS inherited_index ON inherited_index.indexrelid = attachment.inhrelid
          WHERE attachment.inhparent = to_regclass('pgstream.events_lsn_brin_idx')
            AND inherited_index.indrelid = partition.oid
      )
    ORDER BY partition.relname
    \gexec

    -- Create only the parent metadata, then attach the concurrently-built children.
    SET lock_timeout = '5s';
    CREATE INDEX IF NOT EXISTS events_lsn_brin_idx
        ON ONLY pgstream.events
        USING brin (lsn)
        WITH (pages_per_range = 64, autosummarize = on);

    SELECT format(
        'ALTER INDEX pgstream.events_lsn_brin_idx ATTACH PARTITION %I.%I;',
        namespace.nspname, child_index.relname
    )
    FROM pg_partition_tree('pgstream.events'::regclass) AS partition_tree
    JOIN pg_class AS partition ON partition.oid = partition_tree.relid
    JOIN pg_namespace AS namespace ON namespace.oid = partition.relnamespace
    JOIN pg_class AS child_index
      ON child_index.relnamespace = namespace.oid
     AND child_index.relname = partition.relname || '_lsn_brin_idx'
    JOIN pg_index AS child_metadata ON child_metadata.indexrelid = child_index.oid
    WHERE partition_tree.isleaf
      AND child_metadata.indrelid = partition.oid
      AND child_metadata.indisready
      AND child_metadata.indisvalid
      AND NOT EXISTS (
          SELECT 1
          FROM pg_inherits AS attachment
          JOIN pg_index AS inherited_index ON inherited_index.indexrelid = attachment.inhrelid
          WHERE attachment.inhparent = 'pgstream.events_lsn_brin_idx'::regclass
            AND inherited_index.indrelid = partition.oid
      )
    ORDER BY partition.relname
    \gexec
    RESET lock_timeout;

    DO $$
    BEGIN
        IF NOT EXISTS (
            SELECT 1
            FROM pg_index
            WHERE indexrelid = 'pgstream.events_lsn_brin_idx'::regclass
              AND indisready
              AND indisvalid
        ) THEN
            RAISE EXCEPTION 'pgstream.events_lsn_brin_idx is not valid; rerun the rollout before deploying';
        END IF;
    END
    $$;

    SELECT pg_advisory_unlock(hashtext('pgstream.events_lsn_brin_idx'));
\endif