agronholm
commented
6 months ago I need you to minimize this. That is, to remove everything not needed to reproduce the issue.
Closed vadympop closed 2 months ago
agronholm
commented
6 months ago I need you to minimize this. That is, to remove everything not needed to reproduce the issue.
vadympop
commented
6 months ago engine = create_async_engine("postgresql+asyncpg://...")
Session = async_sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=AsyncSession)
Base = declarative_base()
class Item(Base):
__tablename__ = "items"
id = Column(BigInteger, autoincrement=True, primary_key=True)
code = Column(String, unique=True, nullable=False)
async def job_1():
print("JOB_1 EXECUTE START")
await asyncio.sleep(5)
print("JOB_1 EXECUTE END")
async def job_2():
print("JOB_2 EXECUTE START")
async with Session() as db:
new_obj = Item(code="1a1a1a")
db.add(new_obj)
await db.commit()
print("JOB_2 EXECUTE END")
class SchedulerMiddleware:
def __init__(self, app: ASGIApp, scheduler: AsyncScheduler) -> None:
self.app = app
self.scheduler = scheduler
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] == "lifespan":
async with self.scheduler:
await self.scheduler.add_schedule(job_1, IntervalTrigger(minutes=1), id="job_1")
await self.scheduler.add_schedule(job_2, IntervalTrigger(minutes=1), id="job_2")
await self.scheduler.start_in_background()
await self.app(scope, receive, send)
else:
await self.app(scope, receive, send)
@asynccontextmanager
async def lifespan(_: FastAPI):
print("LIFESPAN EVENT")
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
data_store = SQLAlchemyDataStore(engine)
event_broker = AsyncpgEventBroker.from_async_sqla_engine(engine)
scheduler = AsyncScheduler(data_store, event_broker)
app = FastAPI(lifespan=lifespan, middleware=[Middleware(SchedulerMiddleware, scheduler=scheduler)])In this code we're simulating sqalchemy error(ex. sqlalchemy.exc.IntegrityError: (sqlalchemy.dialects.postgresql.asyncpg.IntegrityError) <class 'asyncpg.exceptions.UniqueViolationError'>), after throwing this error the scheduler stops working.
agronholm
commented
6 months ago Will it not trigger the error if job 2 doesn't use the database?
vadympop
commented
6 months ago If, for example, job 2 throws a ValueError, the scheduler will continue work
vadympop
commented
6 months ago But if job 2 will throw error like this raise ValueError("aaasfsdafsghw45hj6tr5wj6rthewh"*10000) will be the same error
agronholm
commented
6 months ago So, in other words, the database stuff isn't required, just raising a ValueError will trigger the problem?
vadympop
commented
6 months ago I think it's because the thrown error text is too long
apscheduler.SerializationError: Serialized event object exceeds 7999 bytes in size
| During handling of the above exception, another exception occurred:
|
| Traceback (most recent call last):
| File "D:\PyProg\PizzaBotAPI\venv\lib\site-packages\apscheduler\_schedulers\async_.py", line 947, in _run_job
| await self.event_broker.publish(
| File "D:\PyProg\PizzaBotAPI\venv\lib\site-packages\apscheduler\eventbrokers\asyncpg.py", line 176, in publish
| raise SerializationError(
| apscheduler.SerializationError: Serialized event object exceeds 7999 bytes in size
+------------------------------------
2023-12-31 09:43:21,305 | ERROR | uvicorn.error:134 - Traceback (most recent call last):
File "D:\PyProg\PizzaBotAPI\venv\lib\site-packages\starlette\routing.py", line 714, in lifespan
await receive()
File "D:\PyProg\PizzaBotAPI\venv\lib\site-packages\uvicorn\lifespan\on.py", line 137, in receive
return await self.receive_queue.get()
File "C:\Python310\lib\asyncio\queues.py", line 159, in get
await getter
asyncio.exceptions.CancelledError: Cancelled by cancel scope 1dd93bea0e0So if we can reproduce it without db, we can minimize code to this:
async def job_1():
print("JOB_1 EXECUTE START")
await asyncio.sleep(5)
print("JOB_1 EXECUTE END")
async def job_2():
print("JOB_2 EXECUTE START")
raise ValueError("aaasfsdafsghw45hj6tr5wj6rthewh"*10000)
class SchedulerMiddleware:
def __init__(self, app: ASGIApp, scheduler: AsyncScheduler) -> None:
self.app = app
self.scheduler = scheduler
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] == "lifespan":
async with self.scheduler:
await self.scheduler.add_schedule(job_1, IntervalTrigger(minutes=1), id="job_1")
await self.scheduler.add_schedule(job_2, IntervalTrigger(minutes=1), id="job_2")
await self.scheduler.start_in_background()
await self.app(scope, receive, send)
else:
await self.app(scope, receive, send)
@asynccontextmanager
async def lifespan(_: FastAPI):
print("LIFESPAN EVENT")
yield
engine = create_async_engine("postgresql+asyncpg://...")
data_store = SQLAlchemyDataStore(engine)
event_broker = AsyncpgEventBroker.from_async_sqla_engine(engine)
scheduler = AsyncScheduler(data_store, event_broker)
app = FastAPI(lifespan=lifespan, middleware=[Middleware(SchedulerMiddleware, scheduler=scheduler)])Right – so in fact, the web framework shouldn't be required to reproduce this.
vadympop
commented
6 months ago Yes, but if it's an error such as sqlalchemy error, then it stops the entire scheduler
agronholm
commented
6 months ago I think I need to remove exception tracebacks and messages from the event structure.
agronholm
commented
6 months ago Ellipsizing them might be a less intrusive change. Could you test with the limit-jobreleased-size branch to see if the problem is gone? I'm done for today.
vadympop
commented
6 months ago limit-jobreleased-size branch is working correctly with this type of error. Thank you!
Things to check first
[X] I have checked that my issue does not already have a solution in the FAQ
[X] I have searched the existing issues and didn't find my bug already reported there
[X] I have checked that my bug is still present in the latest release
Version
4.0.0a4
What happened?
After an exception the AsyncScheduler crashes. I'm tried reproduce this with throwing basic exception, but it works only with Sqlalchemy exception. Also I'm using Sqlalchemy datastore. After exception apscheduler is throwing this:
So after this all schedules stop executing
How can we reproduce the bug?
Here is a minimal code example:
apscheduler_test.py
logging.json