یه اتفاق خیلی عجیب برای ما توی پروداکشن افتاد که ساعت یک شب برای بخش خرید ارور 500 گرفتیم
حالا موضوع چی بود؟
فروشگاه ما به این شکل بود که محصولات از طریق یه provider اپدیت موجودی و قیمت میشدن تقریبا هر اپدیت 5 ثانیه طول میکشید
ما توی فروشگاهمون یه سری محصولات خیلی ارزون هم داشتیم که کمتر از چند سنت بودن و هفته ایی یک یا دوبار شارژ میشدن
یه فیچری زده بودیم که مشتری هامون میتونستن بگن هرموقع این محصول موجود شد برامون خرید رو بزن(چون خیلی سریع هم تموم میشد)
ما هم میددیم اگه موجودی ولتش اندازه بود میخریدیم براش و فعال میکردیم روی اکانتش
حالا این وسط یه مشتری ما با موجودی بر فرض 5$ اومده بود یه خرید بزنه و اشتراک بخره
در لحظه که میاد اشتراک رو بخره
یه تسک دیگه میاد از حساب ولتش چندسنت خرید رو کم میکنه و اون سرویس هفتگی رو براش فعال میکنه(تقریبا 11 سنت)
حالا این وسط چه اتفاقی می افته
این دوتا Request یجورایی پشت هم اتفاق می افتن
و ولت کاربر توی دیتابیس شده 4.89$ ولی خرید اون محصول قیمتش بوده 4.99$
از اون جالبتر اینکه query ما هم داشته میگفته 4.99
در جایی که اصلا نمیتونسته اگه موجودی کافی نبوده به این قسمت برسه!
حالا بریم بررسی کنیم داستان چی بوده
یه موضوعی هست توی SQlalchemy
در رابطه با session ها
برنامه نویس این سرویس بنده خدا فکر میکرده وقتی داره مینویسه
result = await session.execute(
select(Wallet).where(Wallet.id == 1)
)
time.sleep(10) # PROCESS
result = await session.execute(
select(Wallet)
.where(Wallet.id == 1)
)
این یعنی SQlalchemy دوباره داره کوئری میزنه به دیتابیس و مقدار جدید رو میگیره
اما امان از دل غافل
شما وقتی توی یه session یه ابجکتی رو دارین
و قبلا گرفتین
اگه دوباره Query بزنین . Alchemy میبینه اینو توی session اش داره و دوباره درخواست نمیزنه به دیتابیس و مقدار واقعی دیتابیس رو بگیره
میاد همون مقدار توی session رو برمیگردونه
شما باید از
result = await session.execute(
select(Wallet)
.where(Wallet.id == 1).execution_options(populate_existing=True)
)
استفاده کنین
بماند که این پروژه رو ما چندین ماه داشتیم ریفکتور میکردیم و این اتفاق به دلیل وحشتناک بودن و مزخرف بودن سورس کد افتاده بود
در کل مهمه که توی Sqlalchemy حواستون باشه که دیتا داره از ابجکتی که توی session هست خونده میشه یا کوئری واقعا زده شده به سمت دیتابیس
یه موردی:
شاید در لحظه بگین session.refresh(obj) هم میتونستیم استفاده کنیم
که درست هم هست
ولی این دوتا یه فرق اساسی دارن
تو وقتی میای رفرش میزنی میگی که عزیز من برو همین obj مشخص رو دوباره از دیتابیس بخون و اپدیتش کن
ولی وقتی از execution_options(populate_existing=True) استفاده میکنی داری میگی نتایج این کوئری رو حتی اگه توی session وجود دارن برو بگیر و دوباره اپدیتش کن
اگه خوشتون اومده باشه بعدا میتونیم بیشتر راجع بهش صحبت کنیم
4 · 162 · Threadیه اتفاق خیلی عجیب برای ما توی پروداکشن افتاد که ساعت یک شب برای بخش خرید ارور 500 گرفتیم
20 messages · –X S رفتاری که درباره session های sqlalchemy گفتن درسته ولی این مشکل نباید همون ابتدا به ساکن رخ میداد. یکی از راه هاش همون with_for_update. وقتی یه چیزی رو میگیریم میخونیم تا بعدا آپدیت کنیم لاک میخوایم دیگه. این رفتار های تو مخ orm ها هست. یکی دیگه اینه که شما یه آبجکت رو میگیرید، بعد توی همون transaction یه update روی یکی از فیلد های اون آبجکت انجام میدی، sqla میاد اون اتریبیوت از اون آبجکت رو آپدیت میکنه خودسر. و اگه شما اون مقدار رو قبلا نگرفته باشی جدا گانه باید فکر کنی ببینی باگ از کجا بود. اتفاقا چند وقت پیش صحبت میکردیم با مانی سر همین موضوع. این sync کردنای implicit دردسر داره. چیزی که نوشته naked_sqla سعی داشته همین مشکلات رو فیکس کنه.𝐀 یه موضوعی من الان یه مدل کوچیک اووردم بالا یه مدل product با Categpry مدل Product و کتگوری رو گرفتم(کتگوری رو جوین زدم) بعد فرض بر این بزار یه اتفاق ۵ ثانیه ایی افتاد وسطش یه ترنزکشن دیگه اتفاق افتاد status کتگوری رو false شد و بعد از ۵ثانیه دوباره Query زدم مطمئنی بعدش status درست رو نشون میده؟ فک کنم فقط باید از populate_existing استفاده کردS 𝐀 S من میگم اصلا دو تا request جدا که همزمان رسیدن به بک اند درسته؟ که session جدا دارن چه برسه به transaction جدا. وقتی توی A شما میزنید with_for_update، توی دومی چه بزنید with_for_update چه از خود update(table).values(...) استفاده کنید هردوش بلاک میشه تا اولی آزاد کنه. لاک توی دیتابیس هست کل row لاک میشه. اون نمیتونه اصلا آپدیت رو انجام بده. اگر هم که B زودتر برسه که هیچی آپدیتش رو انجام میده و with_for_update تو A بعدش اتفاق میفته𝐀 S """Demonstrate PostgreSQL row locking and stale AsyncSession ORM state. Run with: uv run python temp_sqlalchemy_identity_map_demo.py Docker must be running. The PostgreSQL container is removed automatically. """ import asyncio from rich.console import Console from sqlalchemy import Integer, select, update from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from testcontainers.postgres import PostgresContainer console = Console() class Base(DeclarativeBase): pass class Wallet(Base): __tablename__ = "wallet" id: Mapped[int] = mapped_column(primary_key=True) balance_cents: Mapped[int] = mapped_column(Integer) async def apply_weekly_charge( session_factory: async_sessionmaker[AsyncSession], ) -> None: await asyncio.sleep(0.1) async with session_factory() as session: async with session.begin(): console.print("weekly charge is attempting UPDATE...", style="green") await session.execute(update(Wallet).where(Wallet.id == 1).values(balance_cents=489)) console.print("weekly charge committed: database balance = $4.89", style="green") async def main() -> None: with PostgresContainer("postgres:16-alpine", driver="asyncpg") as postgres: engine = create_async_engine(postgres.get_connection_url(), echo=True) session_factory = async_sessionmaker(engine, expire_on_commit=False) async with engine.begin() as connection: await connection.run_sync(Base.metadata.create_all) async with session_factory() as session: session.add(Wallet(id=1, balance_cents=500)) await session.commit() async with session_factory.begin() as checkout_session: wallet = ( await checkout_session.execute( select(Wallet).where(Wallet.id == 1).with_for_update() # here1 ) ).scalar_one() console.print( f"checkout locked wallet: ${wallet.balance_cents / 100:.2f}", style="green", ) charge_task = asyncio.create_task(apply_weekly_charge(session_factory)) console.print("checkout committed; row lock released", style="green") await asyncio.sleep(0.5) queried_again = ( await checkout_session.execute(select(Wallet).where(Wallet.id == 1)) ).scalar_one() console.print( f"ordinary SELECT in same session: ${queried_again.balance_cents / 100:.2f}", style="green", ) console.print( f"same Python object: {queried_again is wallet}", style="green", ) assert queried_again is wallet assert queried_again.balance_cents == 500 refreshed = ( await checkout_session.execute( select(Wallet).where(Wallet.id == 1).execution_options(populate_existing=True) ) ).scalar_one() console.print( f"with populate_existing: ${refreshed.balance_cents / 100:.2f}", style="green", ) assert refreshed is wallet await asyncio.sleep(0.5) async with session_factory.begin() as final_session: wallet = ( await final_session.execute(select(Wallet).where(Wallet.id == 1)) ).scalar_one() console.print( f"after everything: ${wallet.balance_cents / 100:.2f}", style="green", ) await engine.dispose() if __name__ == "__main__": asyncio.run(main()) اینو llm نوشت. به ترتیب log ها نگاه کن. لاگ آخری که توی apply_weekly_change هست. آخرش از باقی اومده چون لاک شده.
𝐀 فکر کنم فهمیدم موضوع این نیست که دوتا request اتفاق افتاد موضوع اینه یه سشن باز شد یه get انجام شد روی تیبل پروداکت و جوین زدیم و گرفتیم کتگوری رو هم جای دیگه ایی از سرویس ترنزکشن اتفاق افتاد(روی تیبل کتگوری) و بعد روی همون سشن قبلی اومدیم دوباره همون Query رو زدیم روی Product جوابی که میگیری جواب اپدیت شده نیست حتی اگه ازاد شده باشه
𝐀