Fix application error desyncing auto-commit result

Merged
RO
2mo ago
+146-46

Raising an application error inside a with driver.session() as s: block that has an on-going auto-commit transaction, i.e., res = s.run(...) has been called but res has not yet been consumed will lead to the session returning its backing connection to the pool (where it can be picked up by other sessions) without properly finalizing the auto-commit transaction. res then holds a reference to (and potentially uses) a connection that is no longer bound to res's parent session. This can lead to all sorts of misbehavior (from data races casing protocol errors all the way to reading records of another query).

Reproducer:

from __future__ import annotations

from neo4j import GraphDatabase, Result
from neo4j.debug import watch

watch("neo4j")

URL = "bolt://localhost:7687"
AUTH = ("neo4j", "pass")
DB = "neo4j"

class MyError(RuntimeError):
    pass

def inspect_records(result: neo4j.Result, n: int, stream_name: str):
    print(f"\ninspecting stream {stream_name}: fetching {n} records")
    records = [record["res"] for record in result.fetch(n)]
    print(f"got {len(records)} records")
    for record in records:
        print(repr(record))
        assert isinstance(record, str) and record.startswith(f"{stream_name}-")

def main():
    range_top = 10
    query = (
        'UNWIND range(1, $upper) AS i '
        'RETURN "res" + $n + "-" + apoc.text.lpad(toString(i), $pad, "0") '
        'AS res'
    )
    params = {"upper": range_top, "pad": len(str(range_top))}

    with GraphDatabase.driver(URL, auth=AUTH) as driver:
        fetch_size = 2
        try:
            with (
                driver.session(database=DB, fetch_size=fetch_size) as session
            ):
                res1 = session.run(query, parameters={**params, "n": 1})
                inspect_records(res1, 3, "res1")
                raise MyError("test error")
        except MyError as e:
            print(f"Caught MyError: {e}")

        with (
            driver.session(database=DB, fetch_size=fetch_size) as session
        ):
            res2 = session.run(query, parameters={**params, "n": 2})
            inspect_records(res2, 3, "res2")
            # !!! res1 now reading records from res2
            inspect_records(res1, 3, "res1")

if __name__ == "__main__":
    main()

Fixes: DRIVERS-529