skorotkiewicz/rDrama
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix: Issue where cron would end up in nested transaction blocks and unable to escape. (#586)

Browse source 
* Use sessionmaker instead of context block

Essentially there were two bugs:
1) The logging lines before explicit `Session.begin` caused the root
   "A transaction is already begun on this Session.", which is
   resolved by moving the logging within the explicit session block,
   as intended.
2) We were reusing a Session across cron runs. The architecture was
   intended to not do so, but we didn't have an actual sessionmaker
   available to cron. We probably ought to get rid of scoped_session
   elsewhere, but baby steps to resolve the immediate issue.

In testing, this works correctly with the fix to `db.begin`, and
tests with deliberately reintroducing the logging bug, the changes
to session creation prevent the every-15sec stack trace spam.

---------

Almost-entirely-authored-by: TLSM <duolsm@outlook.com>
This commit is contained in:
Ben Rog-Wilhelm 2023-05-01 04:25:05 -05:00 • committed by GitHub
parent 94309ed257
commit c2eeb8b3e7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 9 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

5
files/__main__.py
View file

@ -194,11 +194,12 @@ limiter = flask_limiter.Limiter(
# ...and then after that we can load the database. # ...and then after that we can load the database.
engine: Engine = create_engine(DATABASE_URL) engine: Engine = create_engine(DATABASE_URL)
db_session: scoped_session = scoped_session(sessionmaker( db_session_factory: sessionmaker = sessionmaker(
bind=engine, bind=engine,
autoflush=False, autoflush=False,
future=True, future=True,
)) )
db_session: scoped_session = scoped_session(db_session_factory)
# now that we've that, let's add the cache, compression, and mail extensions to our app... # now that we've that, let's add the cache, compression, and mail extensions to our app...

11
files/commands/cron.py
View file

@ -4,9 +4,9 @@ import time
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import Final from typing import Final
from sqlalchemy.orm import scoped_session, Session from sqlalchemy.orm import sessionmaker, Session
from files.__main__ import app, db_session from files.__main__ import app, db_session_factory
from files.classes.cron.tasks import (DayOfWeek, RepeatableTask, from files.classes.cron.tasks import (DayOfWeek, RepeatableTask,
RepeatableTaskRun, ScheduledTaskState) RepeatableTaskRun, ScheduledTaskState)
@ -41,7 +41,7 @@ def cron_app_worker():
logging.info("Starting scheduler worker process") logging.info("Starting scheduler worker process")
while True: while True:
try: try:
_run_tasks(db_session) _run_tasks(db_session_factory)
except Exception as e: except Exception as e:
logging.exception( logging.exception(
"An unhandled exception occurred while running tasks", "An unhandled exception occurred while running tasks",
@ -77,7 +77,7 @@ def _acquire_lock_exclusive(db: Session, table: str):
raise raise
def _run_tasks(db_session_factory: scoped_session): def _run_tasks(db_session_factory: sessionmaker):
''' '''
Runs tasks, attempting to guarantee that a task is ran once and only once. Runs tasks, attempting to guarantee that a task is ran once and only once.
This uses postgres to lock the table containing our tasks at key points in This uses postgres to lock the table containing our tasks at key points in
@ -116,10 +116,11 @@ def _run_tasks(db_session_factory: scoped_session):
task.run_time_last = now task.run_time_last = now
task.run_state_enum = ScheduledTaskState.RUNNING task.run_state_enum = ScheduledTaskState.RUNNING
# This *must* happen before we start doing db queries, including sqlalchemy db queries
db.begin()
task_debug_identifier = f"(ID {task.id}:{task.label})" task_debug_identifier = f"(ID {task.id}:{task.label})"
logging.info(f"Running task {task_debug_identifier}") logging.info(f"Running task {task_debug_identifier}")
db.begin()
run: RepeatableTaskRun = task.run(db, task.run_time_last_or_created_utc) run: RepeatableTaskRun = task.run(db, task.run_time_last_or_created_utc)
if run.exception: if run.exception: