aiomysql returns wrong data when calling pool.acquire multiple times

[dhensen]

Describe the bug

I can reproduce letting aiomysql return wrong data when I'm using more than one Pool calling pool.acquire multiple times. In the documentation it nowhere warns me not to do this, is this a bug?

To Reproduce

import aiomysql
from sanic import Request, Sanic, response
from sanic.log import logger

app = Sanic("isolated_aiomysql_problem")
app.config.RESPONSE_TIMEOUT = 600

DATABASE_CONFIG = {
    "db": "repro",
    "user": "root",
    "password": "repro",
    "host": "127.0.0.1",
    "port": 13306,
    "echo": True,
}


async def ensure_database():
    create_db_config = DATABASE_CONFIG.copy()
    dbname = create_db_config.pop("db")

    conn = await aiomysql.connect(**create_db_config)
    async with conn.cursor() as cur:
        await cur.execute(f"DROP DATABASE IF EXISTS {dbname}")
        await cur.execute(f"CREATE DATABASE IF NOT EXISTS {dbname}")
        await cur.execute(
            f"""
            USE {dbname};
            CREATE TABLE IF NOT EXISTS users (
                user_id VARCHAR(255) PRIMARY KEY,
                username VARCHAR(255) NOT NULL,
                verification_token VARCHAR(255),
                UNIQUE(username)
            );
            INSERT INTO users (user_id, username, verification_token) VALUES (1, 'github', 'foobar1337');
        """
        )
        await conn.commit()
    conn.close()


@app.main_process_start
async def init(app, loop):
    await ensure_database()


@app.before_server_start
async def do_setup(app, loop):
    app.ctx.pool = await aiomysql.create_pool(loop=loop, **DATABASE_CONFIG)


@app.after_server_stop
async def do_cleanup(app, loop):
    app.ctx.pool.close()
    await app.ctx.pool.wait_closed()


@app.middleware("request")
async def add_db_connection(request: Request):
    request.ctx.conn = await app.ctx.pool.acquire()
    logger.debug(
        "acquired conn",
        extra={"verbosity": 0},
    )


@app.middleware("response")
async def remove_db_connection(request: Request, _):
    if hasattr(request.ctx, "conn"):
        await app.ctx.pool.release(request.ctx.conn)
        logger.debug(
            "release conn",
            extra={"verbosity": 0},
        )


# in the real app, this next endpoint is on a separate blueprint
@app.route("/verify/<verification_token>")
async def verify_user(request, verification_token):
    logger.info(f"verify_user for {verification_token=}")
    if verification_token:
        async with request.ctx.conn.cursor(aiomysql.DictCursor) as cur:
            logger.info("right before execute")
            await cur.execute(
                "SELECT * FROM users WHERE verification_token=%s",
                (verification_token,),
            )
            user = await cur.fetchone()
            logger.info(f"{user=}")
            logger.debug(f"{user=}")
            return response.json(user)


# Se this to False to "fix" the issue
ACQUIRE_NEW_CONN = True


@app.route("/update_token/<token>")
async def update_token(request, token):
    user_id = 1
    username = "github"
    # in the original code this is part of an upsert functionality
    if ACQUIRE_NEW_CONN:
        async with app.ctx.pool.acquire() as conn:
            async with conn.cursor(aiomysql.DictCursor) as cur:
                sql = """
                INSERT INTO users (user_id, username, verification_token)
                VALUES (%s, %s, %s)
                ON DUPLICATE KEY UPDATE verification_token = VALUES(verification_token);
                """
                await cur.execute(sql, (user_id, username, token))
                await conn.commit()
    else:
        conn = request.ctx.conn
        async with conn.cursor(aiomysql.DictCursor) as cur:
            sql = """
            INSERT INTO users (user_id, username, verification_token)
            VALUES (%s, %s, %s)
            ON DUPLICATE KEY UPDATE verification_token = VALUES(verification_token);
            """
            await cur.execute(sql, (user_id, username, token))
            await conn.commit()

    return response.json({"token": token})


if __name__ == "__main__":
    app.run(
        host="0.0.0.0",
        port=8004,
        access_log=True,
        debug=True,
        auto_reload=True,
        workers=1,
    )

Install: pip install sanic==23.6.0 aiomysql==0.2.0
Database: docker run -p "13306:3306" --env="MYSQL_ROOT_PASSWORD=repro" --env="MYSQL_DATABASE=repro" --env="MYSQL_USER=repro" --env="MYSQL_PASSWORD=repro" -d mysql:8

Run python3 repro.py.

One in one terminal run: watch -n1 -- curl http://localhost:8004/verify/foobar1337
In another terminal run once: curl http://localhost:8004/update_token/foobar1001

Now see the watch output toggle between null and the previous result.

Then stop the api. Edit code to
ACQUIRE_NEW_CONN = False
Then run again, seems like problem solved.

Or see https://youtu.be/q7BgBZQ1Jow (when I shot this I didnt have ACQUIRE_NEW_CONN yet, I fix it manually)

Expected behavior

I expect it to answer to queries with the truth, not some old value that belongs to previous queries connection.

Logs/tracebacks

After a few seconds of running below while loop I will run: `curl http://localhost:8004/update_token/foobar1001` to change the value (should be POST, but was lazy) after that I still get foobar1337.

$ while true; do curl "http://localhost:8004/verify/foobar1337"; echo""; sleep 4; done
{"user_id":"1","username":"github","verification_token":"foobar1337"}
{"user_id":"1","username":"github","verification_token":"foobar1337"}
null
{"user_id":"1","username":"github","verification_token":"foobar1337"}
null
{"user_id":"1","username":"github","verification_token":"foobar1337"}
null
{"user_id":"1","username":"github","verification_token":"foobar1337"}
....

Python Version

$ python --version
3.11.6

aiomysql Version

$ python -m pip show aiomysql
0.2.0

PyMySQL Version

$ python -m pip show PyMySQL
1.1.0

SQLAlchemy Version

$ python -m pip show sqlalchemy

OS

Arch Linux, but I get this inside Docker as well.

Database type and version

SELECT VERSION();
8.2.0

Additional context

Code of Conduct

• I agree to follow the aio-libs Code of Conduct

[dhensen]

I made a mistake in what I wrote above. I'm not using two pools, the difference is I'm calling acquire in two different places. The sanic request/response middleware already calls acquire before and after my endpoint implementation runs, then during my endpoint core I call acquire once more. I feel like that's something that should be fine.

[dhensen]

In the additional context, I posted a screenshot. Here the code immediately breaks in the first loop of the while True. So this gives me zero rows, when there obviously should be rows. This flips each time I call my endpoint, so one time there are rows, next time they are not there. This starts happening when I run the update statement after having done the select statement. It keeps happening when I update again and set the value back to what it was initially.

[dhensen]

This morning I woke up an realized this has got to do with transaction isolation levels...

When I commit after my select query, this problem disappears.
Or when I create my pool with autocommit=True, it of course also solves this.

But I'm still not sure if this is expected behaviour. If I do pool.acquire() I expect to get a connection based on the current truth of whatever is committed to the database when acquiring. Pools are re-using connections, so is there a state bug in here? of it my responsability to reset or commit? Also really don't get why the explicit acquire in my update statement shows this problem and not the conn I already acquired in the request middleware. The only thing I can think of.. is that acquire always gives the next available or new connection, which is always a different that the last connection it handed out, and two realities start existing. One where I run the update query and commit, and other connections, that are never committed and thus show the old state of the database.

@Nothing4You any idea?