Browse Source

fix(tests): give each concurrent-dispatch session its own connection

    The scheduler's concurrent-dispatch tests failed on CI and passed
    locally, on a commit that touched nothing but a text file. Six tests
    in tests/unit/test_scheduler_concurrent_dispatch.py went red with
    every queue item logged as "Status set to 'printing'" and four of
    them read back as "pending", alongside "cannot commit transaction -
    SQL statements in progress" and a PrintArchive that could not be
    refreshed.

    The fixtures built their farm on sqlite+aiosqlite:///:memory:, and
    SQLAlchemy backs an in-memory SQLite with a StaticPool: one DBAPI
    connection handed to every session, with nothing keeping them apart.
    That was harmless while check_queue awaited its uploads inline,
    because only one session was ever live at a time. Under the
    refillable upload pool the uploads run as concurrent background
    tasks with a session each, so their transactions interleave on that
    single connection - a sibling session's close() rolls back another's
    flushed-but-uncommitted UPDATE, and a commit() landing while another
    session still holds a cursor raises the commit error above. Whether
    the interleaving lands badly comes down to core count and
    interpreter version, which is why a 30-core box on 3.13 stayed green
    and a 4-vCPU runner on 3.11 did not.

    The four fixtures now put the database in the test's own tmp_path,
    which gets an AsyncAdaptedQueuePool and a connection per session -
    what the application itself runs with (_resolve_pool_kwargs in
    backend/app/core/database.py: pool_size 20, max_overflow 200). So
    the harness is being brought in line with production rather than
    having its assertions relaxed; nothing in the scheduler changes,
    and the concurrency under test was correct throughout.

    Checked against the failure mode rather than against a green run: an
    isolated repro of the same shape - six writer sessions and one
    reader session closing mid-transaction - yields all-pending on the
    in-memory engine and all-printing on a file. The new risk is real
    SQLite write contention on one file, so the file was run twelve
    times pinned to two cores with no failures and no lock errors.

    tests/unit/test_scheduler_busy_reasons_3018.py carries the same
    harness on an in-memory engine, but dispatches to a single printer,
    so it has no second session to race; left as is.
maziggy 4 days ago
parent
commit
1a80430c45
1 changed files with 27 additions and 4 deletions
  1. 27 4
      backend/tests/unit/test_scheduler_concurrent_dispatch.py

+ 27 - 4
backend/tests/unit/test_scheduler_concurrent_dispatch.py

@@ -57,10 +57,33 @@ from backend.app.services.print_scheduler import PrintScheduler
 UPLOAD_SECONDS = 0.15
 UPLOAD_SECONDS = 0.15
 
 
 
 
+def _test_engine(tmp_path):
+    """An engine that gives every session its own connection.
+
+    ``sqlite+aiosqlite:///:memory:`` is the obvious choice here and it is wrong:
+    SQLAlchemy backs an in-memory SQLite with a ``StaticPool`` -- one DBAPI
+    connection handed to every session, with nothing keeping them apart. That was
+    harmless while ``check_queue`` awaited its uploads inline, because only one
+    session was ever live. Under the pool model (#2602) the uploads run as
+    concurrent tasks with a session each, so their transactions interleave on that
+    single connection: a sibling session's ``close()`` rolls back another's
+    flushed-but-uncommitted UPDATE -- rows read back ``pending`` although the log
+    says ``Status set to 'printing'`` -- and a ``commit()`` landing while another
+    session still holds a cursor raises "cannot commit transaction - SQL
+    statements in progress". It failed on CI and passed locally purely on core
+    count and interpreter version.
+
+    A file gets ``AsyncAdaptedQueuePool`` and a connection per session, which is
+    what the app runs with in production (``_resolve_pool_kwargs`` in
+    ``backend/app/core/database.py``: pool_size 20, max_overflow 200).
+    """
+    return create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'queue.db'}", echo=False)
+
+
 @pytest.fixture
 @pytest.fixture
 async def farm(tmp_path):
 async def farm(tmp_path):
     """Build a farm of N printers, each with one pending queue item."""
     """Build a farm of N printers, each with one pending queue item."""
-    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+    engine = _test_engine(tmp_path)
     async with engine.begin() as conn:
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
         await conn.run_sync(Base.metadata.create_all)
     session_maker = async_sessionmaker(engine, expire_on_commit=False)
     session_maker = async_sessionmaker(engine, expire_on_commit=False)
@@ -529,7 +552,7 @@ class TestSharedLibraryRow:
 
 
         Nothing here mutates the library row, so all four must upload at once.
         Nothing here mutates the library row, so all four must upload at once.
         """
         """
-        engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+        engine = _test_engine(tmp_path)
         async with engine.begin() as conn:
         async with engine.begin() as conn:
             await conn.run_sync(Base.metadata.create_all)
             await conn.run_sync(Base.metadata.create_all)
         session_maker = async_sessionmaker(engine, expire_on_commit=False)
         session_maker = async_sessionmaker(engine, expire_on_commit=False)
@@ -551,7 +574,7 @@ class TestSharedLibraryRow:
         Each of these deletes the library row and unlinks the 3MF when done.
         Each of these deletes the library row and unlinks the 3MF when done.
         Exactly one may go per pass; the rest stay pending for a later one.
         Exactly one may go per pass; the rest stay pending for a later one.
         """
         """
-        engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+        engine = _test_engine(tmp_path)
         async with engine.begin() as conn:
         async with engine.begin() as conn:
             await conn.run_sync(Base.metadata.create_all)
             await conn.run_sync(Base.metadata.create_all)
         session_maker = async_sessionmaker(engine, expire_on_commit=False)
         session_maker = async_sessionmaker(engine, expire_on_commit=False)
@@ -583,7 +606,7 @@ async def test_library_print_without_a_parseable_print_time_does_not_crash(tmp_p
     dispatch. Two printers here: if the first one's dispatch blows up, the second
     dispatch. Two printers here: if the first one's dispatch blows up, the second
     must still go out.
     must still go out.
     """
     """
-    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+    engine = _test_engine(tmp_path)
     async with engine.begin() as conn:
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
         await conn.run_sync(Base.metadata.create_all)
     session_maker = async_sessionmaker(engine, expire_on_commit=False)
     session_maker = async_sessionmaker(engine, expire_on_commit=False)