Race condition on CREATE EXTENSION vector in PGEngine (missing advisory lock)

[yukiharada1228]

Summary

PGEngine.ainit_vectorstore_table (v2) is missing the advisory lock that the legacy PGVector.create_vector_extension uses, so the two code paths can race on CREATE EXTENSION IF NOT EXISTS vector and hit a duplicate-key error.

Where

File	Behavior
langchain_postgres/vectorstores.py (_create_vector_extension)	Takes pg_advisory_xact_lock(1573678846307946496) before CREATE EXTENSION IF NOT EXISTS vector
langchain_postgres/v2/engine.py (ainit_vectorstore_table, ~line 215)	Runs the same CREATE EXTENSION IF NOT EXISTS vector without any lock

Why it matters

The advisory lock only serializes callers that take the same lock. Since the v2 path takes no lock at all, it provides zero protection against Postgres's known race where concurrent sessions can both pass the "not exists" check and then collide on insert:

duplicate key value violates unique constraint "pg_extension_name_index"

This can be hit whenever an application initializes a PGEngine-based store concurrently with another PGEngine/PGVector initialization, e.g.:

• Two asyncio.gather'd ainit_vectorstore_table calls
• A PGVectorStore init racing a legacy PGVector init

How we found it

While maintaining a TypeScript port of this package, our test runner executes test files in parallel by default, which reliably reproduces the race between the unlocked PGEngine path and the locked legacy PGVector path. The same asymmetry exists in this Python source, so concurrent use in a real application (not just tests) could hit it too.

Suggested fix

Wrap the CREATE EXTENSION IF NOT EXISTS vector call in v2/engine.py with the same pg_advisory_xact_lock (reusing the existing lock key or helper) so both code paths are mutually exclusive.

Happy to send a PR if that's useful.

[yukiharada1228]

Thanks for taking a look! Yes, that's the fix I have in mind: take the same pg_advisory_xact_lock(1573678846307946496) that the legacy _create_vector_extension uses before running CREATE EXTENSION IF NOT EXISTS vector in PGEngine.ainit_vectorstore_table. Since advisory locks only serialize callers that take the same key, reusing the existing key keeps the v2 path and the legacy PGVector path mutually exclusive.

One small clarification on the symptom: because the extension creation runs and commits before any table DDL, the losing session fails loudly with duplicate key value violates unique constraint "pg_extension_name_index". It doesn't leave a partially initialized store behind. So it's a hard init failure under concurrency rather than silent corruption.

I'm holding off on a PR for now until a maintainer confirms this is the direction they'd like. Happy to put one together (with a concurrency test, e.g. asyncio.gather on multiple ainit_vectorstore_table calls) once that's settled.

[yukiharada1228]

Follow-up with a reproduction on the Python package (PostgreSQL 17.11 + pgvector 0.8.6, fresh database per trial, 4 concurrent ainit_vectorstore_table calls via asyncio.gather, distinct table names):

• Current code: 20/20 trials failed, all with duplicate key value violates unique constraint "pg_extension_name_index". The extension ended up correctly installed in every trial, so this is a hard init failure, not a half-initialized state.
• With SELECT pg_advisory_xact_lock(1573678846307946496); prepended to the CREATE EXTENSION statement in PGEngine.ainit_vectorstore_table: 0/20 trials failed.

Reproduction script

Point ADMIN_URL / DB_URL at a Postgres with pgvector available.

import asyncio

import psycopg

from langchain_postgres import PGEngine

ADMIN_URL = "postgresql://postgres@127.0.0.1:55432/postgres"
DB_URL = "postgresql+psycopg://postgres@127.0.0.1:55432/race_repro"


def recreate_db() -> None:
    with psycopg.connect(ADMIN_URL, autocommit=True) as conn:
        conn.execute("DROP DATABASE IF EXISTS race_repro")
        conn.execute("CREATE DATABASE race_repro")


async def trial() -> list:
    recreate_db()  # fresh database, so the vector extension does not exist yet
    engine = PGEngine.from_connection_string(DB_URL)
    try:
        return await asyncio.gather(
            *(
                engine.ainit_vectorstore_table(table_name=f"t{i}", vector_size=8)
                for i in range(4)
            ),
            return_exceptions=True,
        )
    finally:
        await engine.close()


async def main() -> None:
    failed = 0
    for _ in range(20):
        errors = [r for r in await trial() if isinstance(r, Exception)]
        if errors:
            failed += 1
            print(type(errors[0]).__name__, str(errors[0]).splitlines()[0])
    print(f"{failed}/20 trials failed")


asyncio.run(main())

Output on the current code:

IntegrityError (psycopg.errors.UniqueViolation) duplicate key value violates unique constraint "pg_extension_name_index"
...
20/20 trials failed