سؤال‌های مصاحبه

سؤال: یک سرویس هر شب سفارش‌های پرداخت‌نشده را پیدا می‌کند. در محیط production حدود ۱۰۰٬۰۰۰ سفارش داریم و حدود ۱۰۰٬۰۰۰ شناسه پرداخت. کد این است:

def find_unpaid(order_ids: list[int], paid_ids: list[int]) -> list[int]:
    unpaid = []
    for order_id in order_ids:
        if order_id not in paid_ids:
            unpaid.append(order_id)
    return unpaid

تست‌ها با ۲۰ سفارش سبز هستند. این کد را در code review می‌بینی. نظرت چیست؟ آیا مشکلی دارد؟

جواب کوتاه: کد درست کار می‌کند، ولی کند است. عملگر in روی list یعنی جستجوی خطی: پایتون از اول لیست یکی یکی مقایسه می‌کند. پس کل کار تقریباً n × m مقایسه است، یعنی O(n·m). اگر paid_ids را به یک set تبدیل کنیم، هر جستجو به طور میانگین O(1) می‌شود و کل کار O(n + m).

سرنخ (اگر کاندید گفت مشکلی نیست): در تست‌ها کار در چند میلی‌ثانیه تمام می‌شود. ولی در production همین job چند دقیقه طول می‌کشد و CPU تمام این مدت روی ۱۰۰٪ است. حجم داده فقط ۵۰۰۰ برابر شده، ولی زمان خیلی بیشتر از ۵۰۰۰ برابر شده. چرا؟

مفهوم Big O به زبان ساده:

• نوشتن Big O می‌گوید وقتی ورودی بزرگ می‌شود، تعداد کارها چطور رشد می‌کند. زمان دقیق را نمی‌گوید.
• نماد O(n) یعنی اگر ورودی ۲ برابر شود، کار هم حدود ۲ برابر می‌شود.
• نماد O(n²) یعنی اگر ورودی ۲ برابر شود، کار حدود ۴ برابر می‌شود.
• نماد O(1) یعنی کار به اندازه ورودی بستگی ندارد.

چرا این اتفاق می‌افتد؟ (قدم به قدم)

1. حلقه for روی ۱۰۰٬۰۰۰ سفارش می‌چرخد.
2. برای هر سفارش، عملگر in روی لیست paid_ids اجرا می‌شود. این خودش یک حلقه پنهان است.
3. برای سفارش پرداخت‌نشده، این حلقه پنهان تا آخر لیست می‌رود: ۱۰۰٬۰۰۰ مقایسه.
4. پس در بدترین حالت ۱۰۰٬۰۰۰ × ۱۰۰٬۰۰۰ یعنی ۱۰ میلیارد مقایسه داریم.
5. در تست با ۲۰ × ۲۰ فقط ۴۰۰ مقایسه بود. برای همین مشکل دیده نشد.

راه حل:

1. لیست را یک بار به set تبدیل کنیم. این کار O(m) است.
2. جستجو در set با hash انجام می‌شود و به طور میانگین O(1) است.

def find_unpaid(order_ids: list[int], paid_ids: list[int]) -> list[int]:
    paid = set(paid_ids)  # build once: O(m)
    return [order_id for order_id in order_ids if order_id not in paid]

3. حالا کل کار حدود ۲۰۰٬۰۰۰ عملیات است، نه ۱۰ میلیارد.
4. هزینه: set حافظه بیشتری از list می‌گیرد. برای ۱۰۰٬۰۰۰ عدد این هزینه معمولاً کم است.
5. نکته: ترتیب خروجی همان ترتیب order_ids می‌ماند، چون روی order_ids حلقه می‌زنیم.

یک نکته مهم دیگر: اگر داده از دیتابیس می‌آید، شاید بهتر باشد اصلاً این کار در خود دیتابیس انجام شود، مثلاً با یک کوئری LEFT JOIN یا NOT EXISTS. این طور ۱۰۰٬۰۰۰ ردیف به برنامه منتقل نمی‌شود.

سؤال پیگیری: آیا set همیشه O(1) است؟ چه وقت تبدیل به set ارزش ندارد؟

جواب پیگیری:

• جستجو در set به طور میانگین O(1) است. در بدترین حالت، مثلاً با تابع hash بد و برخورد زیاد، می‌تواند کندتر شود. با int و str در پایتون این حالت در عمل نادر است.
• ساختن set خودش O(m) هزینه دارد. اگر فقط یک بار جستجو کنیم، ساختن set سودی ندارد.
• اگر لیست خیلی کوچک باشد (مثلاً ۵ عضو)، تفاوت دیده نمی‌شود. خوانایی مهم‌تر است.
• عضوهای set باید hashable باشند. مثلاً یک list را نمی‌شود داخل set گذاشت.

نشانه خطر: نمی‌داند عملگر in روی list خودش یک حلقه است، یا فقط می‌گوید «سرور قوی‌تر بگیریم» بدون اینکه پیچیدگی الگوریتم را ببیند.

ویرایش در GitHub

سؤال: یک فایل با ۱ میلیون ایمیل کاربر داریم. باید ایمیل‌هایی را پیدا کنیم که بیش از یک بار آمده‌اند. یک کاندید اول این کد را می‌نویسد:

def find_duplicates(emails: list[str]) -> list[str]:
    duplicates = []
    for i in range(len(emails)):
        for j in range(i + 1, len(emails)):
            if emails[i] == emails[j] and emails[i] not in duplicates:
                duplicates.append(emails[i])
    return duplicates

نظرت درباره این کد چیست؟ چه راه‌های دیگری داریم و هر کدام چه هزینه‌ای دارد؟

جواب کوتاه: کد درست است، ولی O(n²) است. برای ۱ میلیون ایمیل یعنی حدود ۵۰۰ میلیارد مقایسه، که عملاً تمام نمی‌شود. سه راه داریم:

1. حلقه دوتایی: O(n²) زمان، حافظه اضافه تقریباً صفر.
2. مرتب کردن و مقایسه همسایه‌ها: O(n log n) زمان.
3. استفاده از set یا dict (hash): O(n) زمان، ولی حافظه اضافه O(n).

سرنخ (اگر کاندید گفت مشکلی نیست): با ۱۰۰ ایمیل فوری جواب می‌دهد. با ۱۰٬۰۰۰ ایمیل چند ثانیه طول می‌کشد. با ۱ میلیون ایمیل بعد از یک ساعت هنوز تمام نشده است.

چرا حلقه دوتایی کند است؟ (قدم به قدم)

1. هر ایمیل با همه ایمیل‌های بعد از خودش مقایسه می‌شود.
2. تعداد جفت‌ها حدود n × n تقسیم بر ۲ است. برای ۱ میلیون، حدود ۵۰۰ میلیارد.
3. علاوه بر این، عملگر in روی لیست duplicates هم یک حلقه پنهان است و کار را بیشتر می‌کند.
4. اگر ورودی ۱۰ برابر شود، زمان حدود ۱۰۰ برابر می‌شود.

راه دوم: مرتب کردن

1. بعد از مرتب کردن، ایمیل‌های یکسان کنار هم قرار می‌گیرند.
2. فقط کافی است هر عضو را با عضو قبلی مقایسه کنیم.
3. مرتب کردن O(n log n) است و مقایسه همسایه‌ها O(n).

def find_duplicates_sorted(emails: list[str]) -> list[str]:
    s = sorted(emails)
    result = []
    for i in range(1, len(s)):
        if s[i] == s[i - 1] and (not result or result[-1] != s[i]):
            result.append(s[i])
    return result

4. مزیت: اگر داده از قبل مرتب باشد یا خیلی بزرگ باشد و در حافظه جا نشود، روش مرتب کردن خارجی (external sort) روی دیسک ممکن است.
5. عیب: ترتیب اصلی ورودی از بین می‌رود.

راه سوم: hash (بهترین انتخاب معمول)

from collections import Counter

def find_duplicates_hash(emails: list[str]) -> list[str]:
    counts = Counter(emails)
    return [email for email, count in counts.items() if count > 1]

1. هر ایمیل یک بار خوانده می‌شود و شمارنده‌اش در dict زیاد می‌شود. پس O(n) است.
2. هزینه: باید همه ایمیل‌های یکتا را در حافظه نگه داریم.
3. برای ۱ میلیون ایمیل این حافظه در یک سرور معمولی به راحتی جا می‌شود.

یک نکته مهم دیگر: «تکراری» یعنی چه؟ آیا Ali@Mail.com و ali@mail.com یکی هستند؟ آیا فاصله اول و آخر مهم است؟ کاندید خوب این را می‌پرسد و قبل از مقایسه، ایمیل را نرمال می‌کند (مثلاً حروف کوچک و حذف فاصله).

سؤال پیگیری: اگر ۱ میلیارد ایمیل داشته باشیم و در حافظه یک ماشین جا نشود، چه می‌کنی؟

جواب پیگیری:

• یک راه: داده را بر اساس hash ایمیل به چند فایل کوچک‌تر تقسیم کنیم. ایمیل‌های یکسان همیشه در یک فایل می‌روند. بعد هر فایل را جدا با روش hash بررسی کنیم.
• راه دیگر: مرتب کردن خارجی روی دیسک و بعد مقایسه همسایه‌ها.
• اگر داده در دیتابیس است، ساده‌ترین راه یک کوئری GROUP BY با HAVING COUNT بزرگ‌تر از ۱ است.
• اگر فقط جواب تقریبی کافی است، ساختارهایی مثل Bloom filter حافظه کمی می‌گیرند، ولی ممکن است اشتباه مثبت داشته باشند.

نشانه خطر: فقط یک راه را می‌شناسد و نمی‌تواند هزینه زمان و حافظه راه‌ها را با هم مقایسه کند.

ویرایش در GitHub

سؤال: یک تیم ۶ نفره روی شاخه مشترک develop کار می‌کند. صبح امروز یکی از هم‌تیمی‌ها می‌خواست تاریخچه را تمیز کند. این دستورها را روی develop اجرا کرد:

git checkout develop
git rebase -i HEAD~15   # squash and reorder old commits
git push --force

بعد از این، بقیه اعضای تیم وقتی git pull می‌زنند conflict های عجیب می‌گیرند. یک نفر هم می‌گوید commit دیروزش در develop نیست. چه اتفاقی افتاده؟ قانون تیم درباره rebase باید چه باشد؟

جواب کوتاه: دستور rebase تاریخچه را بازنویسی می‌کند: commit های قدیمی با commit های جدید با شناسه (hash) جدید عوض می‌شوند. وقتی این کار روی یک شاخه مشترک انجام شود و با force push فرستاده شود، تاریخچه سرور با تاریخچه کامپیوتر بقیه فرق می‌کند. قانون ساده: تاریخچه شاخه مشترک را هرگز بازنویسی نکن. روی شاخه شخصی خودت rebase آزاد است.

چرا این اتفاق می‌افتد؟ (قدم به قدم)

1. هر commit یک شناسه دارد که از محتوا و از commit پدرش ساخته می‌شود.
2. دستور rebase commit ها را از نو می‌سازد. حتی اگر محتوا یکی باشد، پدر یا محتوا عوض شده، پس شناسه جدید است.
3. هم‌تیمی‌ها هنوز commit های قدیمی را دارند و کارشان را روی آن‌ها ساخته‌اند.
4. وقتی pull می‌زنند، گیت دو تاریخچه متفاوت می‌بیند و سعی می‌کند آن‌ها را merge کند. نتیجه: conflict و commit های تکراری.
5. دستور push با force یعنی «نسخه سرور را با نسخه من عوض کن». اگر کسی بعد از آخرین pull این همکار و قبل از push او، commit جدیدی فرستاده بود، آن commit از سرور پاک می‌شود. این همان commit گم‌شده است.

راه نجات:

1. هیچ commit ای واقعاً فوراً پاک نمی‌شود. کسی که commit گم‌شده را نوشته، آن را هنوز روی کامپیوتر خودش دارد.
2. با دستور reflog می‌شود جای قبلی شاخه را پیدا کرد:

git reflog show develop        # find the old position, e.g. develop@{1}
git branch rescue develop@{1}  # keep it safe on a new branch

3. بعد تیم با هم تصمیم می‌گیرد کدام نسخه درست است و commit های گم‌شده را با cherry-pick یا merge برمی‌گرداند.
4. نکته: reflog محلی است. یعنی فقط روی همان کامپیوتری است که commit را دیده.

قانون‌های تیم:

• روی شاخه‌های مشترک (develop، main) هرگز rebase یا force push نکن. برای گرفتن تغییرات از merge استفاده کن.
• روی شاخه شخصی feature، قبل از Pull Request، rebase برای تمیز کردن commit ها خوب است.
• به جای force ساده، از force-with-lease استفاده کن. این گزینه فقط وقتی push می‌کند که سرور هنوز همان چیزی باشد که تو آخرین بار دیدی. پس کار دیگران را بی‌خبر پاک نمی‌کند.
• در سرور گیت (مثلاً GitHub یا GitLab) برای develop و main «branch protection» روشن کن تا force push ممنوع باشد.

تفاوت merge و rebase:

• با merge، تاریخچه همان‌طور که اتفاق افتاده می‌ماند و یک commit ادغام اضافه می‌شود. امن است، ولی تاریخچه شلوغ‌تر است.
• با rebase، تاریخچه خطی و تمیز می‌شود، ولی commit ها از نو ساخته می‌شوند.

سؤال پیگیری: شاخه feature خودت را rebase کرده‌ای و push معمولی خطا می‌دهد. چه می‌کنی؟

جواب پیگیری:

• اول مطمئن می‌شود هیچ‌کس دیگری روی این شاخه کار نمی‌کند.
• بعد با force-with-lease push می‌کند، نه با force ساده.
• اگر شخص دیگری هم روی شاخه commit زده، اول با او هماهنگ می‌کند یا به جای rebase از merge استفاده می‌کند.

نشانه خطر: استفاده از force push را راه حل عادی هر خطای push می‌داند، یا نمی‌داند rebase شناسه commit ها را عوض می‌کند.

ویرایش در GitHub

سؤال: یک API عمومی برای اپ موبایل و چند شرکت همکار داریم. در code review می‌بینی که همه endpoint ها همیشه کد HTTP 200 برمی‌گردانند. نتیجه در بدنه جواب است:

GET /api/orders/9999 HTTP/1.1

HTTP/1.1 200 OK
Content-Type: application/json

{"success": false, "error": "not found"}

برای خطای دیتابیس هم همین است: کد 200 و بدنه‌ای با success برابر false. نظرت چیست؟

جواب کوتاه: این طراحی خوب نیست. کد وضعیت HTTP یک قرارداد استاندارد است. خیلی چیزها فقط به همین کد نگاه می‌کنند و بدنه را نمی‌خوانند: کتابخانه‌های کلاینت، cache ها، load balancer ها، ابزارهای مانیتورینگ و منطق retry. وقتی همیشه 200 برگردانیم، همه این‌ها فکر می‌کنند همه چیز موفق بوده است.

سرنخ (اگر کاندید گفت مشکلی نیست): داشبورد مانیتورینگ نشان می‌دهد نرخ خطا ۰٪ است. ولی کاربران از خطا شکایت می‌کنند. یک شرکت همکار هم می‌گوید کدش وقتی دیتابیس ما مشکل داشت، دوباره تلاش نکرد. چرا؟

چرا مهم است؟ (قدم به قدم)

1. ابزار مانیتورینگ نرخ خطا را از کدهای 5xx حساب می‌کند. اگر خطای سرور هم 200 باشد، هشدار هیچ وقت فعال نمی‌شود.
2. منطق retry در کلاینت معمولاً روی 5xx یا 503 دوباره تلاش می‌کند و روی 4xx نه. با 200 نمی‌داند باید چه کند.
3. یک cache یا CDN ممکن است جواب 200 را ذخیره کند. پس جواب خطا ممکن است برای کاربران دیگر هم برگردد.
4. هر کلاینت باید بدنه را بخواند و فیلد success را چک کند. اگر یک نفر فراموش کند، خطا را داده درست حساب می‌کند.

گروه‌های اصلی:

• گروه 2xx یعنی موفق. مثلاً 200 برای OK، 201 برای «ساخته شد»، 204 برای «موفق، بدون بدنه».
• گروه 4xx یعنی کلاینت اشتباه کرده. تکرار همان request کمکی نمی‌کند.
• گروه 5xx یعنی سرور مشکل دارد. شاید بعداً دوباره تلاش جواب بدهد.

کدهای مهم 4xx و 5xx:

راه حل:

1. کد درست را برگردانیم و بدنه خطا را هم نگه داریم، چون پیام خطا برای انسان و برای debug مفید است.

HTTP/1.1 404 Not Found
Content-Type: application/problem+json

{"type": "about:blank", "title": "Not Found", "status": 404,
 "detail": "Order 9999 does not exist"}

2. برای شکل بدنه خطا می‌شود از استاندارد Problem Details استفاده کرد (RFC 9457). این طور همه endpoint ها یک شکل خطا دارند.
3. چون API عمومی است، تغییر را ناگهانی انجام ندهیم. کلاینت‌های فعلی به 200 عادت دارند. بهتر است در نسخه جدید API تغییر دهیم و به همکاران خبر بدهیم.

سؤال پیگیری: تفاوت 401 و 403 چیست؟ و چرا بعضی API ها به جای 403، کد 404 برمی‌گردانند؟

جواب پیگیری:

• کد 401 یعنی «نمی‌دانم تو کی هستی». کلاینت باید login کند یا توکن جدید بگیرد.
• کد 403 یعنی «می‌دانم کی هستی، ولی اجازه نداری». login دوباره کمکی نمی‌کند.
• گاهی برای منبع خصوصی کد 404 برمی‌گردانند تا حتی وجود آن منبع را لو ندهند. مثلاً کاربر نباید بفهمد سفارش ۹۹۹۹ متعلق به کس دیگری وجود دارد.

نشانه خطر: کد وضعیت را فقط «تزئینی» می‌داند، یا 401 و 403 را از هم تشخیص نمی‌دهد.

ویرایش در GitHub

سؤال: یک فروشگاه آنلاین یک جعبه جستجو دارد. کاربر نام محصول را می‌نویسد. کد سمت سرور این است (دیتابیس PostgreSQL و کتابخانه psycopg):

def search_products(conn, term: str):
    sql = f"SELECT id, name, price FROM products WHERE name LIKE '%{term}%'"
    with conn.cursor() as cur:
        cur.execute(sql)
        return cur.fetchall()

برنامه با کاربر admin دیتابیس وصل می‌شود. این کد را در code review می‌بینی. نظرت چیست؟

جواب کوتاه: این کد SQL Injection دارد. متن کاربر مستقیم داخل متن SQL چسبانده می‌شود. پس کاربر می‌تواند به جای یک کلمه، کد SQL بفرستد و دیتابیس آن را اجرا می‌کند. راه حل: کوئری پارامتری. یعنی متن SQL و مقدار کاربر جدا به دیتابیس فرستاده شوند. همچنین برنامه نباید با کاربر admin وصل شود.

سرنخ (اگر کاندید گفت مشکلی نیست): اگر کاربر در جعبه جستجو یک تک‌کوتیشن بنویسد (مثلاً نام «O’Reilly»)، صفحه خطای ۵۰۰ می‌دهد و در لاگ یک خطای syntax از SQL دیده می‌شود. چرا یک حرف ساده کوئری را خراب می‌کند؟

چرا این خطرناک است؟ (قدم به قدم)

1. فرض کن کاربر این متن را می‌نویسد:

x' UNION SELECT id, email, 0 FROM users --

2. متن نهایی SQL این می‌شود:

SELECT id, name, price FROM products WHERE name LIKE '%x'
UNION SELECT id, email, 0 FROM users --%'

3. تک‌کوتیشن کاربر رشته را می‌بندد. بقیه متن کاربر حالا دستور SQL است، نه داده.
4. دو خط تیره بقیه کوئری را به comment تبدیل می‌کند.
5. نتیجه: ایمیل همه کاربران در نتیجه جستجو نمایش داده می‌شود. با همین روش می‌شود ستون‌های دیگر را هم خواند.
6. چون برنامه با admin وصل است، مهاجم شاید بتواند جدول‌ها را حذف کند یا داده را عوض کند.

راه حل:

1. از کوئری پارامتری استفاده کنیم. مقدار کاربر از متن SQL جداست و دیتابیس آن را همیشه داده می‌بیند، نه کد.

def search_products(conn, term: str):
    sql = "SELECT id, name, price FROM products WHERE name LIKE %s"
    with conn.cursor() as cur:
        cur.execute(sql, (f"%{term}%",))  # value is sent as a parameter
        return cur.fetchall()

2. دقت کن: علامت درصد اینجا داخل مقدار است، نه داخل متن SQL. این امن است.
3. فرار دادن دستی کاراکترها (escape) یا فیلتر کردن کلمه‌هایی مثل UNION راه درستی نیست. همیشه راهی برای دور زدنش پیدا می‌شود.
4. اصل کمترین دسترسی: برنامه با یک کاربر دیتابیس وصل شود که فقط اجازه‌های لازم را دارد. مثلاً فقط SELECT، INSERT و UPDATE روی جدول‌های خودش. این طور اگر یک باگ injection باقی بماند، خسارت کمتر است.
5. این‌ها هم کمک می‌کنند: یک ORM که خودش پارامتر می‌سازد، و ابزار تحلیل کد (static analysis) که الگوی f-string داخل execute را پیدا می‌کند.

یک نکته مهم دیگر: پارامتر فقط برای مقدار کار می‌کند. نام جدول یا نام ستون (مثلاً برای مرتب کردن) را نمی‌شود پارامتر کرد. برای آن‌ها باید از یک لیست سفید (allowlist) از نام‌های مجاز استفاده کنیم.

سؤال پیگیری: کاربر می‌خواهد نتیجه را بر اساس ستونی که خودش انتخاب می‌کند مرتب کند (price یا name). چطور امن پیاده‌سازی می‌کنی؟

جواب پیگیری:

• ورودی کاربر را مستقیم در ORDER BY نمی‌گذارد.
• یک نگاشت ثابت از گزینه‌های مجاز به نام ستون می‌سازد. هر چیز دیگری را رد می‌کند یا مقدار پیش‌فرض می‌گذارد.

SORT_COLUMNS = {"price": "price", "name": "name"}

column = SORT_COLUMNS.get(user_sort, "name")  # unknown value -> default
sql = f"SELECT id, name, price FROM products ORDER BY {column}"

• این f-string امن است، چون column فقط می‌تواند یکی از مقدارهای ثابت خود ما باشد، نه متن کاربر.

نشانه خطر: راه حلش «حذف تک‌کوتیشن از ورودی» است، یا فکر می‌کند چون جعبه جستجو در فرانت‌اند اعتبارسنجی می‌شود، سرور امن است.

ویرایش در GitHub

سؤال: در کد یک فروشگاه، این تابع حدود ۱۵۰ خط است. از ۸ جای مختلف صدا زده می‌شود:

def process(order, validate=True, send_email=True, discount_code=None):
    if validate:
        ...  # 40 lines of validation
    if discount_code is not None:
        ...  # 30 lines of discount logic
    ...      # 30 lines: save to database
    if send_email:
        ...  # 40 lines: build and send email

# somewhere else in the code:
process(order, True, False, None)
process(order, False, True, "SUMMER")

این کد را در code review می‌بینی. نظرت چیست؟

جواب کوتاه: این تابع چند کار جدا انجام می‌کند: اعتبارسنجی، تخفیف، ذخیره و ایمیل. پس اصل Single Responsibility را رعایت نمی‌کند. پارامترهای بولی (به آن‌ها flag argument می‌گویند) نشان می‌دهند که تابع در واقع چند تابع است که در هم رفته‌اند. همچنین خط صدا زدنی که فقط True، False و None پشت هم دارد خوانا نیست. راه حل: تابع را به چند تابع کوچک با نام روشن تقسیم کنیم.

سرنخ (اگر کاندید گفت مشکلی نیست): یک برنامه‌نویس جدید خط اول صدا زدن (با True، False و None) را می‌خواند. نمی‌داند True و False یعنی چه. یک باگ هم پیدا شده: یک جا، جای دو بولی اشتباهی عوض شده و سفارش بدون اعتبارسنجی ذخیره شده است. تست این تابع هم سخت است.

چرا مشکل است؟ (قدم به قدم)

1. با ۲ پارامتر بولی و یک پارامتر اختیاری، حداقل ۸ حالت مختلف داریم. هر حالت باید تست شود.
2. وقتی کسی خط صدا زدن را می‌خواند، باید تعریف تابع را باز کند تا معنی True و False را بفهمد.
3. اگر جای دو بولی عوض شود، پایتون خطا نمی‌دهد. باگ بی‌صدا وارد می‌شود.
4. هر تغییر در بخش ایمیل، فایلی را عوض می‌کند که منطق ذخیره هم در آن است. پس خطر خراب کردن بخش دیگر بیشتر است.
5. هر پرچم جدید، تعداد حالت‌ها را دو برابر می‌کند.

راه حل:

1. هر کار را به یک تابع کوچک با نام روشن تبدیل کنیم.
2. تابع اصلی فقط ترتیب کارها را نشان دهد.

def validate_order(order) -> None: ...
def apply_discount(order, code: str) -> None: ...
def save_order(order) -> None: ...
def send_confirmation_email(order) -> None: ...

def place_order(order, discount_code: str | None = None) -> None:
    validate_order(order)
    if discount_code:
        apply_discount(order, discount_code)
    save_order(order)
    send_confirmation_email(order)

3. جایی که کار خاصی لازم است، همان تابع‌های کوچک را مستقیم صدا بزنیم. مثلاً import داده قدیمی فقط save_order را صدا می‌زند.
4. اگر واقعاً یک گزینه لازم است، حداقل آن را keyword-only کنیم تا خط صدا زدن خوانا باشد:

def place_order(order, *, send_email: bool = True) -> None: ...

place_order(order, send_email=False)  # clear at the call site

5. تغییر را قدم به قدم انجام دهیم. اول برای رفتار فعلی تست بنویسیم، بعد تقسیم کنیم.

یک نکته مهم دیگر: همیشه پرچم بد نیست. یک گزینه ساده مثل reverse در تابع sorted پایتون مشکلی ندارد، چون فقط یک جزء کوچک از رفتار را عوض می‌کند. مشکل وقتی است که پرچم یک کار کاملاً جدا را روشن یا خاموش می‌کند.

سؤال پیگیری: در یک جا validate برابر False است، چون سفارش از سیستم قدیمی می‌آید و اعتبارسنجی آن را رد می‌کند. با این مورد چه می‌کنی؟

جواب پیگیری:

• اول می‌پرسد چرا رد می‌شود. شاید قانون اعتبارسنجی یا داده قدیمی مشکل دارد.
• اگر این مسیر واقعاً متفاوت است، یک تابع جدا با نام روشن می‌سازد، مثلاً import_legacy_order، که قانون‌های خودش را دارد.
• این طور تصمیم «بدون اعتبارسنجی» در نام تابع دیده می‌شود، نه پشت یک False پنهان.

نشانه خطر: راه حلش اضافه کردن یک پرچم دیگر است، یا تابع ۱۵۰ خطی با چند مسئولیت را عادی می‌داند.

ویرایش در GitHub

سؤال: یک فروشگاه آنلاین قیمت‌ها را به دلار نگه می‌دارد. کد جمع سبد خرید این است:

class CartItem:
    def __init__(self, price: float, quantity: int):
        self.price = price
        self.quantity = quantity

def cart_total(items: list[CartItem]) -> float:
    total = 0.0
    for item in items:
        total += item.price * item.quantity
    return total

در دیتابیس هم ستون قیمت از نوع FLOAT است. این کد را در code review می‌بینی. نظرت چیست؟

جواب کوتاه: نوع float برای پول مناسب نیست. float عدد را در مبنای ۲ نگه می‌دارد و خیلی از کسرهای ده‌دهی (مثل 0.1) را دقیق نمی‌تواند نشان دهد. خطاهای خیلی کوچک جمع می‌شوند و گاهی جمع نهایی یک سنت اختلاف پیدا می‌کند. راه حل: نوع Decimal یا نگه داشتن پول به صورت عدد صحیح سنت. در دیتابیس هم نوع NUMERIC یا DECIMAL.

سرنخ (اگر کاندید گفت مشکلی نیست): تیم مالی می‌گوید جمع فاکتورهای ماه با جمع پرداخت‌ها یک یا دو سنت فرق دارد. در کنسول پایتون این را امتحان می‌کنیم:

>>> 0.1 + 0.2
0.30000000000000004
>>> 0.1 + 0.2 == 0.3
False

چرا این اتفاق می‌افتد؟ (قدم به قدم)

1. کامپیوتر float را با بیت‌ها، یعنی در مبنای ۲، نگه می‌دارد.
2. در مبنای ۱۰، کسر یک‌سوم دقیق نوشته نمی‌شود: ۰٫۳۳۳۳ و تا بی‌نهایت ادامه دارد.
3. به همین شکل، در مبنای ۲ عدد 0.1 دقیق نوشته نمی‌شود. پس نزدیک‌ترین عدد ممکن ذخیره می‌شود.
4. هر جمع و ضرب یک خطای خیلی کوچک اضافه می‌کند.
5. بعد از هزاران عملیات، یا موقع گرد کردن، این خطا می‌تواند به یک سنت برسد.
6. مقایسه با علامت مساوی هم قابل اعتماد نیست، چون 0.1 + 0.2 دقیقاً 0.3 نیست.

راه حل:

1. در پایتون از Decimal استفاده کنیم. مقدار را از رشته بسازیم، نه از float، وگرنه خطای float همراهش می‌آید.

from decimal import Decimal, ROUND_HALF_UP

price = Decimal("19.99")
total = price * 3
print(total)  # 59.97

tax = (total * Decimal("0.09")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)

2. راه دیگر: پول را به صورت عدد صحیح در کوچک‌ترین واحد نگه داریم، مثلاً ۱۹۹۹ سنت به جای ۱۹٫۹۹ دلار. جمع اعداد صحیح همیشه دقیق است.
3. در دیتابیس ستون را از نوع NUMERIC با دقت مشخص تعریف کنیم، مثلاً NUMERIC با ۱۲ رقم که ۲ رقم آن اعشار است.
4. قانون گرد کردن را مشخص کنیم: کجا گرد می‌کنیم (هر ردیف یا فقط جمع نهایی) و با چه روشی (مثلاً half-up یا گرد کردن بانکی). این یک تصمیم کسب‌وکار است و باید با تیم مالی هماهنگ شود.
5. پول را همیشه همراه با واحد پول نگه داریم. ۱۰ دلار و ۱۰ یورو را نباید با هم جمع کرد.

یک نکته مهم دیگر: همه پول‌ها دو رقم اعشار ندارند. مثلاً ین ژاپن واحد کوچک‌تر رایج ندارد. پس تعداد رقم اعشار را برای هر واحد پول ثابت فرض نکنیم.

سؤال پیگیری: باید ۱۰۰ دلار را بین ۳ نفر تقسیم کنیم. چطور این کار را درست انجام می‌دهی؟

جواب پیگیری:

1. اگر هر نفر ۳۳٫۳۳ بگیرد، جمع ۹۹٫۹۹ می‌شود و یک سنت گم می‌شود.
2. راه درست: با سنت کار کنیم. ۱۰٬۰۰۰ سنت تقسیم بر ۳ می‌شود ۳۳۳۳ و باقیمانده ۱.
3. باقیمانده را به یکی از سهم‌ها اضافه کنیم: ۳۳۳۴، ۳۳۳۳ و ۳۳۳۳.
4. حالا جمع سهم‌ها دقیقاً با مبلغ اصلی برابر است. قانون اینکه سنت اضافه به چه کسی برسد را کسب‌وکار تعیین می‌کند.

نشانه خطر: می‌گوید «فقط آخر کار round می‌کنیم» و نمی‌داند چرا 0.1 در float دقیق نیست.

ویرایش در GitHub

سؤال: یک هم‌تیمی یک Pull Request فرستاده و از تو خواسته امروز آن را review کنی. آمار PR این است:

60 files changed, 3,041 insertions(+), 1,877 deletions(-)

Commits:
- add loyalty points feature
- refactor OrderService into smaller classes
- run new formatter on the whole project
- fix tests

توضیح PR فقط یک خط است: «امتیاز وفاداری اضافه شد». چطور این PR را review می‌کنی؟ به نویسنده چه می‌گویی؟

جواب کوتاه: این PR سه کار جدا را با هم دارد: یک feature جدید، یک refactor و تغییر فرمت. با این اندازه، review واقعی تقریباً ممکن نیست. تغییرات مهم بین هزاران خط فرمت گم می‌شوند. بهتر است با مهربانی از نویسنده بخواهیم PR را به چند PR کوچک تقسیم کند: اول فرمت، بعد refactor بدون تغییر رفتار، و در آخر feature.

سرنخ (اگر کاندید گفت مشکلی نیست): دو هفته بعد یک باگ در محاسبه قیمت سفارش پیدا می‌شود. معلوم می‌شود خط باگ‌دار در همین PR بوده، ولی بین صدها خط تغییر فرمت در OrderService بوده و هیچ‌کس آن را ندیده است.

چرا PR بزرگ مشکل است؟ (قدم به قدم)

1. توجه انسان محدود است. بعد از چند صد خط، reviewer خسته می‌شود و فقط نگاه سطحی می‌کند.
2. تغییر فرمت هزاران خط diff می‌سازد که هیچ معنایی ندارند. ولی reviewer باید از بین آن‌ها خط‌های مهم را پیدا کند.
3. وقتی refactor و feature با هم هستند، نمی‌شود فهمید یک تغییر رفتار عمدی است یا اشتباه.
4. اگر بعداً مشکلی پیدا شود، revert کردن کل PR همه کارها را با هم برمی‌گرداند.
5. یک PR بزرگ مدت طولانی باز می‌ماند. پس conflict با کار بقیه بیشتر می‌شود.

چه کار می‌کنم؟

1. اول با نویسنده حرف می‌زنم، نه فقط comment می‌گذارم. شاید فشار زمانی دارد.
2. پیشنهاد می‌دهم این طور تقسیم کند:
   • یک PR فقط برای فرمت. این را می‌شود سریع با یک نگاه و تست‌های سبز تأیید کرد.
   • یک PR برای refactor که رفتار را عوض نمی‌کند. تست‌های فعلی باید بدون تغییر سبز بمانند.
   • یک PR برای feature، با توضیح کامل و تست‌های جدید.
3. اگر تقسیم واقعاً ممکن نیست، حداقل commit ها را جدا review می‌کنم و diff را با گزینه «نادیده گرفتن فاصله‌ها» نگاه می‌کنم.
4. در review روی چیزهای مهم تمرکز می‌کنم: درستی منطق، حالت‌های مرزی، امنیت، تست‌ها. سلیقه شخصی و فرمت را به linter و formatter می‌سپارم.
5. از نویسنده یک توضیح بهتر می‌خواهم: چه چیزی، چرا، و چطور تست شده است.

مهربانی در comment ها:

• به جای «این اشتباه است»، می‌گویم «فکر می‌کنم اینجا وقتی لیست خالی باشد خطا می‌دهد. نظرت چیست؟»
• درباره کد حرف می‌زنم، نه درباره آدم.
• مشخص می‌کنم کدام comment ضروری است و کدام فقط پیشنهاد است (مثلاً با پیشوند nit برای نکته کوچک).
• نکته‌های خوب PR را هم می‌گویم.

سؤال پیگیری: برای اینکه در آینده PR های این‌قدر بزرگ ساخته نشوند، در تیم چه کار می‌کنی؟

جواب پیگیری:

• تغییر فرمت سراسری را یک بار جدا انجام می‌دهیم و formatter را در CI یا pre-commit می‌گذاریم تا دیگر در PR ها دیده نشود.
• برای feature های بزرگ از feature flag استفاده می‌کنیم تا بتوانیم کد ناتمام را در PR های کوچک merge کنیم بدون اینکه کاربر آن را ببیند.
• یک قالب (template) برای توضیح PR می‌گذاریم.
• در تیم توافق می‌کنیم که PR کوچک، سریع‌تر review می‌شود. این خودش انگیزه می‌دهد.

نشانه خطر: کاندید PR را بدون خواندن تأیید می‌کند چون «تست‌ها سبز است»، یا comment های تند و شخصی می‌نویسد.

ویرایش در GitHub

سؤال: یک کوپن تخفیف تاریخ انقضا دارد. این تابع و تست آن را در code review می‌بینی:

from datetime import datetime, timedelta

def is_expired(coupon) -> bool:
    return datetime.now() > coupon.expires_at

def test_coupon_expiring_tomorrow_is_valid():
    now = datetime.now()
    tomorrow = now.replace(day=now.day + 1, hour=0, minute=0)
    coupon = Coupon(expires_at=tomorrow)
    assert is_expired(coupon) is False

def test_coupon_from_one_second_ago_is_expired():
    coupon = Coupon(expires_at=datetime.now() - timedelta(seconds=1))
    assert is_expired(coupon) is True

تست‌ها روی کامپیوتر نویسنده سبز هستند. نظرت چیست؟

جواب کوتاه: تابع و تست هر دو ساعت واقعی سیستم را می‌خوانند. پس نتیجه تست به زمان اجرا بستگی دارد. یعنی تست قطعی (deterministic) نیست. راه حل: زمان را به تابع تزریق کنیم، یا به صورت پارامتر یا به صورت یک clock که در تست قابل تعویض است. در تست هم یک زمان ثابت بدهیم.

سرنخ (اگر کاندید گفت مشکلی نیست): این تست بیشتر روزها سبز است. ولی روز آخر هر ماه در CI قرمز می‌شود و خطای «day is out of range for month» می‌دهد. فردای آن روز، بدون هیچ تغییری در کد، دوباره سبز می‌شود.

چرا این اتفاق می‌افتد؟ (قدم به قدم)

1. در تست اول، «فردا» با اضافه کردن ۱ به روز ماه ساخته می‌شود.
2. اگر امروز ۳۱ام باشد، روز ۳۲ وجود ندارد. پس متد replace خطا می‌دهد. این یک باگ در خود تست است.
3. مشکل بزرگ‌تر: تست یک بار تابع now از datetime را صدا می‌زند و تابع is_expired یک بار دیگر. بین این دو، زمان جلو می‌رود. نزدیک مرزها، این فاصله کوچک می‌تواند نتیجه را عوض کند.
4. نتیجه تست به ساعت، روز، منطقه زمانی (time zone) و سرعت ماشین CI بستگی دارد. تستی که گاهی سبز و گاهی قرمز است، flaky است.
5. تست flaky به تیم یاد می‌دهد که قرمز شدن را نادیده بگیرد. این از نداشتن تست هم بدتر است.
6. یک مشکل دیگر: نمی‌شود حالت‌های مرزی را تست کرد. مثلاً «دقیقاً در لحظه انقضا چه می‌شود؟»

راه حل:

1. ساده‌ترین راه: زمان فعلی را به عنوان پارامتر بگیریم.

from datetime import datetime, timezone

def is_expired(coupon, now: datetime) -> bool:
    return now >= coupon.expires_at

def test_coupon_is_expired_exactly_at_expiry_time():
    expiry = datetime(2026, 1, 31, 23, 59, 59, tzinfo=timezone.utc)
    coupon = Coupon(expires_at=expiry)
    assert is_expired(coupon, now=expiry) is True

2. حالا تست هر روز و هر ساعت همان نتیجه را می‌دهد. می‌شود مرزها را هم دقیق تست کرد.
3. اگر زمان در لایه‌های زیادی لازم است، یک clock تزریق کنیم. در production ساعت واقعی و در تست یک ساعت ثابت:

class SystemClock:
    def now(self) -> datetime:
        return datetime.now(timezone.utc)

class FixedClock:
    def __init__(self, at: datetime):
        self._at = at
    def now(self) -> datetime:
        return self._at

4. کتابخانه‌هایی هم هستند که datetime را در تست ثابت می‌کنند (مثلاً freezegun). کار می‌کنند، ولی وابستگی پنهان به زمان را در کد باقی می‌گذارند.
5. در همین تغییر، تصمیم لحظه مرز را روشن کنیم: در لحظه انقضا کوپن معتبر است یا نه؟ این را با تیم محصول مشخص کنیم و برایش تست بنویسیم.
6. زمان را با منطقه زمانی (مثلاً UTC) نگه داریم تا مقایسه دو زمان از منطقه‌های مختلف اشتباه نشود.

سؤال پیگیری: به جز ساعت، چه چیزهای دیگری تست را غیرقطعی می‌کنند؟

جواب پیگیری:

• عدد تصادفی بدون seed ثابت. راه حل: تزریق random یا ثابت کردن seed در تست.
• شبکه و سرویس‌های بیرونی. راه حل: جایگزین (fake یا stub) در تست واحد.
• ترتیب اجرای تست‌ها و داده مشترک بین آن‌ها. راه حل: هر تست داده خودش را بسازد.
• ترتیب در ساختارهایی مثل set، و کار هم‌زمان (thread). راه حل: به ترتیب تکیه نکنیم و از sleep برای هماهنگی استفاده نکنیم.

نشانه خطر: می‌گوید «اگر قرمز شد، دوباره اجرا می‌کنیم»، یا مشکل را فقط باگ روز ۳۲ می‌بیند و وابستگی به ساعت را نمی‌بیند.

ویرایش در GitHub

سؤال: یک سرویس پرداخت از چند روش پرداخت پشتیبانی می‌کند. هر چند ماه یک روش جدید اضافه می‌شود. این تابع در کد است:

def charge(payment_type: str, amount, details):
    if payment_type == "card":
        ...  # 30 lines: call card gateway
    elif payment_type == "paypal":
        ...  # 25 lines: call PayPal API
    elif payment_type == "crypto":
        ...  # 40 lines: create invoice, wait for confirmation
    else:
        raise ValueError(f"unknown payment type: {payment_type}")

چهار تابع دیگر هم همین زنجیره if/elif را روی payment_type دارند: refund، fee، validate_details و display_name. حالا یک PR آمده که روش جدید «bank_transfer» را اضافه می‌کند. این PR به هر پنج تابع یک شاخه elif اضافه کرده است. نظرت چیست؟

جواب کوتاه: کد کار می‌کند، ولی اصل باز-بسته (Open/Closed) از SOLID را رعایت نمی‌کند: برای اضافه کردن یک روش جدید باید کد قدیمیِ پنج جای مختلف را تغییر دهیم. دانش هر روش پرداخت هم در پنج جا پخش شده است. راه حل: هر روش پرداخت یک کلاس جدا بشود که همه رفتارهای آن روش را دارد (Strategy یا polymorphism). یک registry هم نوع را به کلاس وصل می‌کند. ولی اگر فقط دو یا سه حالت ساده داریم که کم عوض می‌شوند، همان if ساده کافی است (YAGNI).

سرنخ (اگر کاندید گفت مشکلی نیست): ماه پیش وقتی crypto اضافه شد، برنامه‌نویس یادش رفت شاخه آن را در تابع refund اضافه کند. اولین مشتری که خواست پول crypto را برگرداند، خطای «unknown payment type» گرفت. تست‌ها این را نگرفتند، چون فقط charge را تست می‌کردند.

چرا این مشکل است؟ (قدم به قدم)

1. منطق یک روش پرداخت در پنج تابع پخش شده است. برای فهمیدن کامل crypto، باید پنج جا را بخوانیم.
2. هر روش جدید یعنی تغییر در پنج تابع قدیمی. هر تغییر خطر خراب کردن روش‌های قبلی را دارد.
3. اگر یک جا فراموش شود، خطا فقط در زمان اجرا دیده می‌شود، آن هم فقط در همان مسیر کم‌استفاده.
4. هر تابع بلندتر می‌شود. چند نفر هم‌زمان در یک تابع تغییر می‌دهند و conflict زیاد می‌شود.

راه حل:

1. یک قرارداد مشترک برای همه روش‌ها تعریف کنیم.
2. هر روش یک کلاس جدا است که همه رفتارهایش را کنار هم دارد.
3. یک registry نام روش را به کلاس وصل می‌کند.

from typing import Protocol

class PaymentMethod(Protocol):
    def charge(self, amount, details) -> None: ...
    def refund(self, amount, details) -> None: ...
    def fee(self, amount): ...
    def validate_details(self, details) -> None: ...
    def display_name(self) -> str: ...

class CardPayment:
    def charge(self, amount, details) -> None: ...
    def refund(self, amount, details) -> None: ...
    def fee(self, amount): ...
    def validate_details(self, details) -> None: ...
    def display_name(self) -> str: ...

METHODS: dict[str, PaymentMethod] = {
    "card": CardPayment(),
    # "paypal": PayPalPayment(), ...
}

def get_method(payment_type: str) -> PaymentMethod:
    try:
        return METHODS[payment_type]
    except KeyError:
        raise ValueError(f"unknown payment type: {payment_type}")

4. حالا اضافه کردن bank_transfer یعنی یک کلاس جدید و یک خط در registry. کد روش‌های قبلی دست نمی‌خورد.
5. اگر کلاس جدید یک متد را نداشته باشد، type checker (مثل mypy) موقع ثبت در registry هشدار می‌دهد. پس فراموش کردن refund سخت‌تر می‌شود.
6. یک تست ساده هم می‌شود نوشت که برای همه روش‌های registry، همه متدها را صدا بزند.

کی همان if ساده بهتر است؟

• وقتی فقط دو یا سه حالت داریم و فقط در یک جا از آن‌ها استفاده می‌شود.
• وقتی هیچ نشانه‌ای نیست که حالت جدید اضافه شود.
• وقتی هر شاخه فقط یکی دو خط است. ساختن چند کلاس برای آن، پیچیدگی بی‌دلیل است.
• قانون عملی: وقتی همان if/elif در چند جا تکرار شد، یا بار دوم و سوم مجبور شدیم حالت جدید اضافه کنیم، وقت تغییر ساختار است.

سؤال پیگیری: این تغییر را چطور در یک سیستم پرداخت در حال کار انجام می‌دهی، بدون اینکه چیزی خراب شود؟

جواب پیگیری:

1. اول برای رفتار فعلی هر پنج تابع و هر سه روش تست می‌نویسد.
2. بعد روش‌ها را یکی یکی به کلاس منتقل می‌کند. تابع‌های قدیمی در ابتدا فقط به کلاس جدید ارجاع می‌دهند.
3. بعد از هر قدم تست‌ها را اجرا می‌کند و تغییر را در PR های کوچک می‌فرستد.
4. روش جدید bank_transfer را فقط بعد از تمام شدن این تغییر، با ساختار جدید اضافه می‌کند.

نشانه خطر: یا هیچ مشکلی در پنج زنجیره تکراری نمی‌بیند، یا برای هر if کوچکی یک الگوی طراحی و چند کلاس پیشنهاد می‌دهد.

ویرایش در GitHub

سؤال: در یک سرویس فروش، این کلاس سفارش را ثبت می‌کند و به مشتری ایمیل می‌فرستد. تیم می‌خواهد برایش تست بنویسد:

class OrderService:
    def __init__(self):
        self.db = PostgresConnection("postgres://prod-db:5432/shop")
        self.mailer = SmtpClient("smtp.company.com", 587)

    def place_order(self, customer_email, items):
        order_id = self.db.insert_order(customer_email, items)
        self.mailer.send(customer_email, f"Order {order_id} received")
        return order_id

این کد را در code review می‌بینی. نظرت چیست؟ آیا مشکلی دارد؟

جواب کوتاه: کلاس وابستگی‌هایش را خودش می‌سازد. پس این وابستگی‌ها پنهان‌اند و نمی‌شود آن‌ها را عوض کرد. راه حل تزریق وابستگی (Dependency Injection) است: کلاس دیتابیس و فرستنده ایمیل را از بیرون، در سازنده، می‌گیرد. حالا در تست می‌شود یک نسخه ساختگی (fake) داد.

سرنخ (اگر کاندید گفت مشکلی نیست): تیم برای این کلاس یک تست نوشت. تست فقط وقتی اجرا می‌شود که یک دیتابیس واقعی در دسترس باشد. هر بار که تست اجرا شد، یک ایمیل واقعی به آدرس تستی رفت. یک بار هم کسی تست را با تنظیمات اشتباه اجرا کرد و یک سفارش آزمایشی در دیتابیس اصلی ثبت شد. چرا نمی‌شود این کلاس را جدا تست کرد؟

چرا این یک مشکل است؟ (قدم به قدم)

1. سازنده کلاس خودش اتصال دیتابیس و کلاینت ایمیل را می‌سازد.
2. کسی که از بیرون کلاس را می‌سازد، هیچ راهی ندارد که بگوید «این بار یک دیتابیس دیگر بده».
3. پس هر تست مجبور است دیتابیس واقعی و سرور ایمیل واقعی داشته باشد.
4. تست کند می‌شود، به شبکه وابسته است، و گاهی بی‌دلیل شکست می‌خورد.
5. از امضای سازنده هم معلوم نیست که کلاس به دیتابیس و ایمیل نیاز دارد. این وابستگی‌ها پنهان هستند.
6. آدرس سرورها هم داخل کد ثابت شده است. برای محیط تست یا staging باید کد را عوض کرد.

راه حل:

1. وابستگی‌ها را از بیرون بگیر:

from typing import Protocol

class OrderRepository(Protocol):
    def insert_order(self, email: str, items: list) -> int: ...

class Mailer(Protocol):
    def send(self, to: str, body: str) -> None: ...

class OrderService:
    def __init__(self, repo: OrderRepository, mailer: Mailer):
        self.repo = repo
        self.mailer = mailer

2. کلاس حالا به یک قرارداد (interface) وابسته است، نه به یک کلاس مشخص مثل Postgres یا SMTP.
3. در تست، نسخه‌های ساختگی ساده می‌دهیم:

class FakeMailer:
    def __init__(self):
        self.sent = []
    def send(self, to, body):
        self.sent.append((to, body))

def test_place_order_sends_email():
    mailer = FakeMailer()
    service = OrderService(InMemoryOrderRepository(), mailer)
    service.place_order("a@example.com", ["book"])
    assert len(mailer.sent) == 1

4. همه چیز در یک جا سرهم می‌شود. به این جا Composition Root می‌گویند. معمولاً همان نقطه شروع برنامه است. فقط این‌جا اشیای واقعی ساخته می‌شوند و تنظیمات (مثل آدرس سرور) از متغیرهای محیطی خوانده می‌شود.
5. برای این کار لزوماً به یک کتابخانه DI نیاز نیست. پاس دادن ساده در سازنده کافی است.

یک هشدار: زیاده‌روی نکن.

• همه چیز نیاز به interface ندارد. فقط چیزهایی که با دنیای بیرون حرف می‌زنند (دیتابیس، شبکه، ایمیل، ساعت، فایل) باید قابل تعویض باشند.
• یک تابع ساده که فقط حساب می‌کند، نیازی به تزریق ندارد.
• اگر هر کلاس یک interface با فقط یک پیاده‌سازی داشته باشد که هیچ وقت عوض نمی‌شود، کد فقط شلوغ‌تر شده است.

سؤال پیگیری: آیا با fake ها دیگر نیازی به تست با دیتابیس واقعی نداریم؟

جواب پیگیری:

• نه. تست واحد با fake منطق کلاس را چک می‌کند، سریع و جدا.
• ولی fake نمی‌تواند ثابت کند که کوئری SQL واقعی درست است.
• پس چند تست یکپارچه (integration) هم لازم است که خود کلاس دیتابیس را با یک دیتابیس واقعی تست کند. مثلاً یک دیتابیس موقت در کانتینر، نه دیتابیس اصلی.
• ایمیل واقعی در تست‌ها هرگز نباید فرستاده شود.

نشانه خطر: فکر می‌کند تست کردن یعنی اجرا کردن روی دیتابیس اصلی. یا برعکس، برای هر کلاس کوچک یک interface و یک کتابخانه DI سنگین پیشنهاد می‌دهد، بدون این‌که دلیلی بیاورد.

ویرایش در GitHub

سؤال: یک برنامه وب پایتون داریم که با چند thread درخواست‌ها را جواب می‌دهد. می‌خواهیم تعداد بازدید صفحه اصلی را بشماریم. کد این است:

visit_count = 0

def home_page(request):
    global visit_count
    visit_count += 1
    return render("home.html")

این کد را در code review می‌بینی. نظرت چیست؟ آیا مشکلی دارد؟

جواب کوتاه: این یک race condition است. عمل «اضافه کردن یک واحد» در واقع سه قدم است: خواندن، اضافه کردن، نوشتن. این سه قدم با هم و یکجا (atomic) انجام نمی‌شوند. اگر دو thread هم‌زمان این کار را بکنند، یکی از افزایش‌ها گم می‌شود. راه حل یک قفل (lock) است. ولی اگر برنامه چند process یا چند سرور دارد، قفل داخل حافظه کمکی نمی‌کند و باید شمارش را به دیتابیس یا Redis سپرد.

سرنخ (اگر کاندید گفت مشکلی نیست): با ابزار تست بار، دقیقاً ۱۰۰٬۰۰۰ درخواست به صفحه اصلی فرستادیم. همه با موفقیت جواب گرفتند. ولی شمارنده در پایان عددی کمتر از ۱۰۰٬۰۰۰ نشان می‌دهد. با بار کم، عدد درست است. چرا؟

چرا این اتفاق می‌افتد؟ (قدم به قدم)

1. دستور اضافه کردن یک واحد، در عمل این سه کار است: مقدار فعلی را بخوان، یکی اضافه کن، نتیجه را بنویس.
2. فرض کن مقدار ۱۰ است.
3. ترد A مقدار را می‌خواند: ۱۰.
4. قبل از این‌که A بنویسد، سیستم‌عامل یا مفسر پایتون اجرا را به ترد B می‌دهد. ترد B هم ۱۰ را می‌خواند.
5. ترد B عدد ۱۱ را می‌نویسد. بعد ترد A هم ۱۱ را می‌نویسد.
6. دو بازدید داشتیم، ولی شمارنده فقط یکی بالا رفت.
7. با بار کم، احتمال این هم‌زمانی کم است. برای همین مشکل فقط زیر بار دیده می‌شود.

درباره GIL در پایتون: بعضی‌ها فکر می‌کنند GIL این مشکل را حل می‌کند. ولی پایتون تضمین نمی‌کند که این دستور atomic باشد. پس نباید به آن تکیه کرد.

راه حل در یک process:

import threading

visit_count = 0
count_lock = threading.Lock()

def home_page(request):
    global visit_count
    with count_lock:
        visit_count += 1
    return render("home.html")

حالا در هر لحظه فقط یک thread می‌تواند خواندن و نوشتن را انجام دهد. کار داخل قفل را کوتاه نگه دار تا threadها زیاد منتظر نمانند.

ولی در محیط واقعی:

1. معمولاً برنامه چند process دارد (مثلاً چند worker در Gunicorn) یا روی چند سرور اجرا می‌شود.
2. هر process حافظه و شمارنده خودش را دارد. قفل یک process، process دیگر را نمی‌بیند.
3. وقتی process دوباره راه‌اندازی شود، عدد داخل حافظه هم صفر می‌شود.
4. پس شمارنده باید در یک جای مشترک باشد و افزایش باید atomic همان‌جا انجام شود.

UPDATE page_stats SET visits = visits + 1 WHERE page = 'home';

5. یا در Redis از دستور INCR استفاده کن. این دستور atomic است.
6. نکته مهم: «اول بخوان، بعد در کد اضافه کن، بعد بنویس» در دیتابیس هم همان race condition را دارد. افزایش باید در خود یک دستور باشد.

سؤال پیگیری: اگر میلیون‌ها بازدید در دقیقه داشته باشیم، یک ردیف در دیتابیس که همه آن را آپدیت می‌کنند مشکلی ایجاد نمی‌کند؟

جواب پیگیری:

• چرا، آن ردیف گلوگاه می‌شود. همه درخواست‌ها برای قفل همان یک ردیف صف می‌کشند.
• یک راه: هر process در حافظه خودش (با قفل) بشمارد و هر چند ثانیه یک بار مجموع را به دیتابیس اضافه کند. کمی دقت لحظه‌ای را فدا می‌کنیم.
• راه دیگر: چند ردیف شمارنده داشته باشیم و هر درخواست یکی را تصادفی آپدیت کند. موقع خواندن، جمعشان را حساب می‌کنیم.
• اول باید پرسید: آیا عدد دقیق لحظه‌ای واقعاً لازم است؟

نشانه خطر: می‌گوید «پایتون GIL دارد، پس مشکلی نیست». یا قفل داخل حافظه را برای برنامه‌ای که روی چند سرور اجرا می‌شود کافی می‌داند.

ویرایش در GitHub

سؤال: جدول users حدود ۵ میلیون ردیف دارد. روی ستون email یک ایندکس هست. برای ورود کاربر، چون بعضی کاربران ایمیل را با حروف بزرگ وارد می‌کنند، یک همکار این کوئری را نوشته است (دیتابیس PostgreSQL است):

CREATE INDEX idx_users_email ON users (email);

SELECT id, password_hash
FROM users
WHERE LOWER(email) = LOWER('Sara@Example.com');

این کد را در code review می‌بینی. نظرت چیست؟ آیا مشکلی دارد؟

جواب کوتاه: ایندکس روی خود مقدار email ساخته شده است، نه روی حروف کوچک آن. وقتی تابع LOWER روی ستون اجرا می‌شود، دیتابیس دیگر نمی‌تواند از آن ایندکس استفاده کند. پس کل جدول را می‌خواند. راه حل: یک ایندکس روی عبارت (expression index) روی حروف کوچک email بسازیم، یا ایمیل را از اول با حروف کوچک ذخیره کنیم. برای مطمئن شدن، نقشه اجرا را با EXPLAIN نگاه می‌کنیم.

سرنخ (اگر کاندید گفت مشکلی نیست): صفحه ورود حدود ۲ ثانیه طول می‌کشد. همکار می‌گوید «ولی روی email ایندکس داریم». وقتی همین کوئری را بدون LOWER اجرا می‌کنیم، جواب در چند میلی‌ثانیه می‌آید. فرق کجاست؟

چرا این اتفاق می‌افتد؟ (قدم به قدم)

1. ایندکس B-tree مثل یک لیست مرتب از مقدارهای ستون email است. مثلاً «Sara@Example.com» همان‌طور که هست در آن ذخیره شده.
2. کوئری دنبال مقدار ستون نیست. دنبال «نتیجه تابع LOWER روی ستون» است.
3. این نتیجه در ایندکس ذخیره نشده است.
4. پس دیتابیس مجبور است برای هر ۵ میلیون ردیف، تابع LOWER را اجرا کند و مقایسه کند. به این Seq Scan (خواندن کل جدول) می‌گویند.
5. همین قاعده برای هر تابع یا محاسبه روی ستون صدق می‌کند. مثلاً گرفتن سال از یک ستون تاریخ، یا جمع کردن ستون با یک عدد.

چطور مطمئن شویم؟

EXPLAIN ANALYZE
SELECT id FROM users WHERE LOWER(email) = 'sara@example.com';

در خروجی، اگر Seq Scan روی users ببینیم، یعنی ایندکس استفاده نشده. بعد از اصلاح، باید Index Scan یا Bitmap Index Scan ببینیم.

راه حل:

1. راه اول، ایندکس روی عبارت. حالا خود نتیجه LOWER در ایندکس ذخیره می‌شود:

CREATE INDEX idx_users_email_lower ON users (LOWER(email));

2. کوئری باید دقیقاً همان عبارت را داشته باشد تا ایندکس استفاده شود.
3. راه دوم، داده را نرمال ذخیره کن. موقع ثبت‌نام و ورود، ایمیل را در کد به حروف کوچک تبدیل کن. حالا کوئری ساده می‌شود و همان ایندکس معمولی کار می‌کند.
4. اگر ایمیل باید یکتا باشد، ایندکس یکتا را هم روی همان شکل نرمال بساز. وگرنه «Sara@Example.com» و «sara@example.com» دو حساب جدا می‌شوند.
5. در PostgreSQL نوع داده citext هم هست که مقایسه را بدون حساسیت به بزرگی و کوچکی حروف انجام می‌دهد. انتخاب بین این راه‌ها به تیم بستگی دارد.

سؤال پیگیری: پس بهتر نیست روی هر ستونی که در WHERE می‌آید یک ایندکس بسازیم؟

جواب پیگیری:

• نه. هر ایندکس هزینه دارد.
• هر INSERT و UPDATE و DELETE باید همه ایندکس‌ها را هم به‌روز کند. پس نوشتن کندتر می‌شود.
• هر ایندکس فضای دیسک و حافظه می‌گیرد.
• ایندکس روی ستونی که مقدارهای کمی دارد (مثل یک ستون true/false) معمولاً کمکی نمی‌کند.
• پس فقط برای کوئری‌های واقعی و پرتکرار ایندکس بساز، و اثرش را با EXPLAIN اندازه بگیر.

نشانه خطر: فکر می‌کند «ایندکس داریم، پس کوئری سریع است». یا هیچ وقت نقشه اجرای یک کوئری را نگاه نکرده است.

ویرایش در GitHub

سؤال: یک فروشگاه آنلاین حدود ۱ میلیون سفارش دارد و هر روز چند هزار سفارش اضافه می‌شود. پنل مدیریت و چند سرویس دیگر از این endpoint استفاده می‌کنند:

@app.get("/orders")
def list_orders():
    rows = db.execute("SELECT * FROM orders ORDER BY created_at DESC").fetchall()
    return [order_to_dict(r) for r in rows]

این کد را در code review می‌بینی. نظرت چیست؟ این API را چطور طراحی می‌کنی؟

جواب کوتاه: این endpoint همه سفارش‌ها را یک‌جا می‌خواند و در یک JSON برمی‌گرداند. با رشد داده، حافظه، زمان و شبکه همه با هم رشد می‌کنند. باید صفحه‌بندی (pagination) داشته باشد، با یک حداکثر اندازه صفحه، و امکان فیلتر. برای جدول‌های بزرگ، صفحه‌بندی با cursor (keyset) بهتر از OFFSET است.

سرنخ (اگر کاندید گفت مشکلی نیست): وقتی سفارش‌ها کم بودند همه چیز خوب بود. حالا جواب این endpoint ۴۰ ثانیه طول می‌کشد، حجمش چند صد مگابایت است، و گاهی سرور با خطای کمبود حافظه از کار می‌افتد. پنل مدیریت هم فقط ۲۰ سفارش آخر را نشان می‌دهد.

چرا این مشکل است؟ (قدم به قدم)

1. دیتابیس باید ۱ میلیون ردیف را بخواند و بفرستد.
2. برنامه همه را در حافظه نگه می‌دارد، به دیکشنری تبدیل می‌کند، و بعد یک رشته JSON بزرگ می‌سازد. چند نسخه از همان داده هم‌زمان در حافظه است.
3. این حجم از شبکه می‌رود و کلاینت هم باید کل آن را parse کند.
4. اگر چند درخواست هم‌زمان بیاید، حافظه سرور تمام می‌شود.
5. هر روز داده بیشتر می‌شود. پس این endpoint هر روز کندتر می‌شود. هزینه با کل داده رشد می‌کند، نه با چیزی که کاربر لازم دارد.

راه حل اول: OFFSET و LIMIT

SELECT * FROM orders ORDER BY created_at DESC, id DESC
LIMIT 50 OFFSET 100000;

• ساده است و می‌شود مستقیم به صفحه ۲۰۰۰ رفت.
• ولی برای صفحه‌های عمیق کند است. دیتابیس باید ۱۰۰٬۰۰۰ ردیف اول را بخواند و دور بریزد، بعد ۵۰ ردیف بدهد.
• اگر وسط ورق زدن سفارش جدید اضافه شود، ردیف‌ها جابه‌جا می‌شوند. کاربر یک سفارش را دو بار می‌بیند یا یکی را اصلاً نمی‌بیند.

راه حل دوم: cursor یا keyset

SELECT * FROM orders
WHERE (created_at, id) < (:last_created_at, :last_id)
ORDER BY created_at DESC, id DESC
LIMIT 50;

• کلاینت مقدار آخرین ردیف صفحه قبل را می‌فرستد (معمولاً به شکل یک رشته مبهم به نام cursor).
• با یک ایندکس روی همین دو ستون، دیتابیس مستقیم به همان نقطه می‌رود. صفحه ۱ و صفحه ۲۰۰۰ تقریباً یک سرعت دارند.
• ستون id برای این است که ترتیب یکتا باشد، چون چند سفارش ممکن است زمان یکسان داشته باشند.
• عیب: نمی‌شود مستقیم به «صفحه ۲۰۰۰» پرید. فقط «بعدی» داریم.

چیزهای دیگر که API باید داشته باشد:

• یک اندازه صفحه پیش‌فرض (مثلاً ۵۰) و یک سقف (مثلاً ۲۰۰). اگر کلاینت عدد بزرگ‌تری خواست، سرور همان سقف را می‌دهد.
• فیلترها، مثل وضعیت سفارش یا بازه تاریخ. اغلب کلاینت اصلاً همه سفارش‌ها را لازم ندارد.
• فقط ستون‌های لازم، نه SELECT همه ستون‌ها.
• لینک یا cursor صفحه بعد در جواب، تا کلاینت خودش آن را نسازد.

سؤال پیگیری: تیم مالی واقعاً همه سفارش‌های سال را برای گزارش لازم دارد. چه می‌کنی؟

جواب پیگیری:

• این یک نیاز دیگر است و نباید از همان API صفحه‌بندی‌شده با هزاران درخواست انجام شود.
• یک راه: یک کار پس‌زمینه (export) که فایل CSV می‌سازد و لینک دانلود می‌دهد.
• راه دیگر: داده را به یک انبار داده یا دیتابیس گزارش‌گیری بفرستیم تا کوئری‌های سنگین روی دیتابیس اصلی اجرا نشوند.
• اگر واقعاً باید از API بیاید، جواب را stream کنیم و همه را یک‌جا در حافظه نسازیم.

نشانه خطر: فکر می‌کند «الان فقط ۱ میلیون است، مشکلی نیست». یا نمی‌داند چرا OFFSET برای صفحه‌های عمیق کند می‌شود.

ویرایش در GitHub

سؤال: اپلیکیشن موبایل یک فروشگاه، برای ثبت سفارش درخواست POST به آدرس orders می‌فرستد. بسیاری از کاربران روی اینترنت موبایل ضعیف هستند. کد سمت اپ این است:

def submit_order(cart):
    for attempt in range(3):
        try:
            return http.post("/orders", json=cart, timeout=5)
        except TimeoutError:
            continue
    raise OrderFailed()

و سمت سرور:

@app.post("/orders")
def create_order(cart):
    order = db.insert_order(cart)
    payments.charge(cart.customer_id, cart.total)
    return order

این کد را در code review می‌بینی. نظرت چیست؟ آیا مشکلی دارد؟

جواب کوتاه: تکرار درخواست بعد از timeout خوب است، ولی درخواست POST idempotent نیست. یعنی اگر دو بار اجرا شود، دو نتیجه می‌سازد. timeout یعنی «جواب را نگرفتم»، نه «سرور کار را انجام نداد». راه حل: کلاینت برای هر سفارش یک کلید idempotency می‌سازد و در همه تکرارها همان را می‌فرستد. سرور کلید را با یک محدودیت یکتا (unique constraint) ذخیره می‌کند و اگر دوباره دید، همان نتیجه اول را برمی‌گرداند.

سرنخ (اگر کاندید گفت مشکلی نیست): پشتیبانی گزارش می‌دهد که بعضی مشتری‌ها دو سفارش یکسان دارند و دو بار پول پرداخته‌اند. تقریباً همه آن‌ها روی شبکه موبایل بوده‌اند. در لاگ سرور، هر دو درخواست با موفقیت تمام شده‌اند.

چرا این اتفاق می‌افتد؟ (قدم به قدم)

1. اپ درخواست را می‌فرستد. سرور آن را می‌گیرد، سفارش را ثبت می‌کند و پول را کم می‌کند.
2. جواب سرور در راه برگشت، به خاطر شبکه ضعیف، دیر می‌رسد یا گم می‌شود.
3. بعد از ۵ ثانیه، اپ فکر می‌کند درخواست شکست خورده است. پس دوباره می‌فرستد.
4. سرور نمی‌داند این همان سفارش قبلی است. پس یک سفارش جدید می‌سازد و دوباره پول می‌گیرد.
5. اپ نمی‌تواند فرق «درخواست نرسید» و «جواب نرسید» را بفهمد. پس تکرار لازم است. ولی تکرار فقط وقتی امن است که عملیات idempotent باشد.

راه حل:

1. اپ وقتی کاربر دکمه «ثبت سفارش» را می‌زند، یک شناسه تصادفی (مثلاً UUID) می‌سازد. این شناسه را در همه تکرارها، در یک هدر می‌فرستد:

POST /orders
Idempotency-Key: 7f3c9a2e-5b1d-4e8a-9c0f-2d6b1a4e8f30

2. سرور یک جدول دارد که کلید را با یک محدودیت یکتا نگه می‌دارد:

CREATE TABLE idempotency_keys (
    key          TEXT PRIMARY KEY,
    customer_id  BIGINT NOT NULL,
    response     JSONB,
    created_at   TIMESTAMPTZ NOT NULL DEFAULT now()
);

3. وقتی درخواست می‌رسد، سرور اول سعی می‌کند کلید را ثبت کند.
4. اگر ثبت شد، یعنی درخواست جدید است. سفارش را می‌سازد و جواب را کنار کلید ذخیره می‌کند. بهتر است ثبت سفارش و ذخیره جواب در یک تراکنش باشند.
5. اگر کلید از قبل بود و جواب دارد، همان جواب ذخیره‌شده را برمی‌گرداند. هیچ کار جدیدی انجام نمی‌دهد.
6. اگر کلید از قبل بود ولی هنوز جواب ندارد (درخواست اول هنوز در حال اجراست)، یک خطای مشخص مثل 409 می‌دهد تا کلاینت کمی بعد دوباره امتحان کند.
7. کلید را به مشتری گره بزن، تا کلید یک کاربر روی کاربر دیگر اثر نگذارد.
8. کلیدهای قدیمی را بعد از مدتی (مثلاً یک روز) پاک کن.

یک نکته دیگر: سرویس پرداخت هم باید همین کار را بکند. اگر درگاه پرداخت کلید idempotency می‌پذیرد (بعضی درگاه‌ها می‌پذیرند)، شناسه سفارش را به عنوان کلید به آن بفرست.

سؤال پیگیری: چرا فقط دکمه را بعد از اولین کلیک غیرفعال نکنیم؟

جواب پیگیری:

• غیرفعال کردن دکمه خوب است و کلیک دوباره کاربر را کم می‌کند.
• ولی مشکل این‌جا کلیک کاربر نیست. تکرار را خود کد اپ انجام می‌دهد.
• کلاینت‌های دیگر، proxy ها و load balancer ها هم ممکن است درخواست را تکرار کنند.
• پس سرور باید خودش از تکرار محافظت کند. سمت کلاینت هیچ وقت کافی نیست.

نشانه خطر: پیشنهاد می‌دهد تکرار را کلاً حذف کنیم. یا می‌خواهد سفارش تکراری را با مقایسه «همان مشتری و همان مبلغ در یک دقیقه اخیر» پیدا کند، که یک خرید واقعی دوم را هم رد می‌کند.

ویرایش در GitHub

سؤال: یک سایت حدود ۲ میلیون کاربر دارد. رمز عبور کاربران این‌طور ذخیره و چک می‌شود:

import hashlib

def hash_password(password: str) -> str:
    return hashlib.sha256(password.encode()).hexdigest()

def register(email, password):
    db.insert_user(email, hash_password(password))

def check_login(email, password):
    user = db.find_user(email)
    return user.password_hash == hash_password(password)

همکار می‌گوید: «رمزها را به شکل متن ساده ذخیره نمی‌کنیم. هش می‌کنیم، پس امن است.» این کد را در code review می‌بینی. نظرت چیست؟

جواب کوتاه: هش کردن بهتر از متن ساده است، ولی این روش برای رمز عبور ضعیف است. دو مشکل دارد: SHA-256 خیلی سریع است، و salt ندارد. باید از یک الگوریتم مخصوص رمز عبور استفاده کرد که عمداً کند است و salt را خودش می‌سازد: Argon2id (توصیه OWASP)، یا scrypt یا bcrypt. برای کاربران فعلی، رمز را هنگام ورود بعدی با الگوریتم جدید دوباره هش می‌کنیم.

سرنخ (اگر کاندید گفت مشکلی نیست): فرض کن یک روز نسخه پشتیبان دیتابیس لو برود. مهاجم حالا همه هش‌ها را دارد. در دیتابیس می‌بینیم چند هزار کاربر دقیقاً هش یکسان دارند. مهاجم چقدر طول می‌کشد تا رمزهای رایج را پیدا کند؟

چرا این مشکل است؟ (قدم به قدم)

1. هش یک‌طرفه است. مهاجم نمی‌تواند آن را «باز» کند. ولی می‌تواند حدس بزند: یک رمز احتمالی را هش کند و با هش ذخیره‌شده مقایسه کند.
2. توابعی مثل SHA-256 برای سرعت ساخته شده‌اند. یک کارت گرافیک معمولی می‌تواند در هر ثانیه تعداد بسیار زیادی (میلیاردها) هش SHA-256 حساب کند.
3. پس امتحان کردن یک لیست بزرگ از رمزهای رایج و ترکیب‌هایشان خیلی سریع است.
4. بدون salt، دو کاربر با رمز یکسان هش یکسان دارند. پس مهاجم با یک حدس، همه آن‌ها را با هم پیدا می‌کند.
5. بدون salt، مهاجم می‌تواند از جدول‌های از پیش حساب‌شده (rainbow table) هم استفاده کند.
6. یک نکته کوچک‌تر: مقایسه با علامت مساوی معمولی در زمان ثابت انجام نمی‌شود. بهتر است از مقایسه زمان-ثابت استفاده کرد. کتابخانه‌های رمز عبور این را خودشان انجام می‌دهند.

تعریف salt: یک مقدار تصادفی و یکتا برای هر کاربر که قبل از هش به رمز اضافه می‌شود و کنار هش ذخیره می‌شود. مخفی نیست. فقط باعث می‌شود رمزهای یکسان هش متفاوت داشته باشند و هر کاربر جدا حمله شود.

راه حل:

1. از یک کتابخانه آماده استفاده کن، خودت الگوریتم نساز. مثلاً با کتابخانه argon2-cffi:

from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError

ph = PasswordHasher()  # Argon2id with library defaults

def hash_password(password: str) -> str:
    return ph.hash(password)  # salt and parameters are stored inside the result

def verify_password(stored_hash: str, password: str) -> bool:
    try:
        return ph.verify(stored_hash, password)
    except VerifyMismatchError:
        return False

2. این الگوریتم‌ها عمداً کند هستند و حافظه زیادی می‌خواهند. برای یک ورود، چند ده میلی‌ثانیه مهم نیست. برای مهاجمی که میلیاردها حدس می‌زند، همین کندی خیلی مهم است.
3. پارامترها (زمان و حافظه) را طوری تنظیم کن که روی سرور خودت قابل قبول باشد. با بهتر شدن سخت‌افزار، آن‌ها را بالا ببر.

مهاجرت کاربران فعلی:

1. ما رمز اصلی کاربران را نداریم. پس نمی‌توانیم همه را یک‌جا دوباره هش کنیم.
2. وقتی کاربر وارد می‌شود، رمز را با روش قدیمی چک می‌کنیم. اگر درست بود، همان لحظه رمز را با Argon2id هش می‌کنیم و جایگزین می‌کنیم.
3. یک ستون یا پیشوند نشان می‌دهد هر هش با کدام الگوریتم ساخته شده است.
4. برای کاربرانی که مدت‌ها وارد نمی‌شوند، یک راه این است که هش قدیمی را داخل Argon2id بپیچیم (هشِ هش). راه دیگر این است که رمزشان را باطل کنیم و از آن‌ها بخواهیم رمز را بازنشانی کنند.

سؤال پیگیری: چرا رمزها را رمزنگاری (encrypt) نکنیم تا اگر لازم شد بتوانیم آن‌ها را برگردانیم؟

جواب پیگیری:

• سیستم هیچ وقت نباید رمز اصلی کاربر را لازم داشته باشد. فقط باید بتواند آن را چک کند.
• رمزنگاری برگشت‌پذیر است. اگر کلید هم لو برود، همه رمزها یک‌جا لو می‌روند.
• کاربران اغلب یک رمز را در چند سایت استفاده می‌کنند. پس لو رفتن رمز ما به حساب‌های دیگر آن‌ها هم آسیب می‌زند.

نشانه خطر: فکر می‌کند «هش شده، پس امن است». یا پیشنهاد می‌دهد یک الگوریتم هش جدید را خودش بنویسد، یا SHA-256 را چند بار پشت سر هم اجرا کند به جای استفاده از یک کتابخانه استاندارد.

ویرایش در GitHub

سؤال: صفحه پروفایل کاربر خیلی پربازدید است. برای کم کردن بار دیتابیس، تیم یک کش Redis جلوی آن گذاشته است. برنامه روی چند سرور اجرا می‌شود و همه از همان Redis استفاده می‌کنند:

CACHE_TTL = 600  # seconds

def get_profile(user_id):
    key = f"profile:{user_id}"
    cached = redis.get(key)
    if cached:
        return json.loads(cached)
    profile = db.load_profile(user_id)
    redis.set(key, json.dumps(profile), ex=CACHE_TTL)
    return profile

def update_name(user_id, new_name):
    db.execute("UPDATE users SET name = %s WHERE id = %s", (new_name, user_id))

این کد را در code review می‌بینی. نظرت چیست؟ آیا مشکلی دارد؟

جواب کوتاه: تابع تغییر نام فقط دیتابیس را عوض می‌کند و به کش دست نمی‌زند. پس تا ۱۰ دقیقه، کش نسخه قدیمی را نشان می‌دهد. راه حل: بعد از نوشتن در دیتابیس، کلید کش را پاک کن (invalidate). ترتیب کار مهم است: اول دیتابیس، بعد پاک کردن کش. حتی در این حالت یک race کوچک باقی می‌ماند، پس TTL را به عنوان تور ایمنی نگه دار.

سرنخ (اگر کاندید گفت مشکلی نیست): کاربران شکایت می‌کنند: «نامم را عوض کردم، پیام موفقیت دیدم، ولی صفحه پروفایل هنوز نام قبلی را نشان می‌دهد.» بعد از چند دقیقه، خودبه‌خود درست می‌شود. در دیتابیس، نام جدید درست ذخیره شده است.

چرا این اتفاق می‌افتد؟ (قدم به قدم)

1. کاربر صفحه پروفایل را باز می‌کند. داده از دیتابیس خوانده و برای ۶۰۰ ثانیه در کش گذاشته می‌شود.
2. کاربر نامش را عوض می‌کند. دیتابیس به‌روز می‌شود.
3. کش هیچ خبری از این تغییر ندارد.
4. کاربر صفحه را دوباره باز می‌کند. کش هنوز داده دارد، پس دیتابیس اصلاً خوانده نمی‌شود.
5. تا وقتی TTL تمام نشود، نام قدیمی نمایش داده می‌شود.

راه حل:

def update_name(user_id, new_name):
    db.execute("UPDATE users SET name = %s WHERE id = %s", (new_name, user_id))
    redis.delete(f"profile:{user_id}")

1. چرا پاک کردن، نه نوشتن مقدار جدید در کش؟ چون پاک کردن ساده‌تر است. خواندن بعدی خودش داده تازه را از دیتابیس می‌آورد. اگر دو update هم‌زمان مقدار کش را بنویسند، ممکن است مقدار قدیمی‌تر آخر از همه نوشته شود.
2. چرا اول دیتابیس، بعد کش؟ اگر اول کش را پاک کنیم، یک خواندن هم‌زمان ممکن است قبل از تغییر دیتابیس، مقدار قدیمی را دوباره در کش بگذارد.
3. اگر دیتابیس در یک تراکنش است، کش را بعد از commit پاک کن، نه قبل از آن.

همان race کوچک که باقی می‌ماند:

1. درخواست خواندن A کش را خالی می‌بیند و نام قدیمی را از دیتابیس می‌خواند.
2. قبل از این‌که A در کش بنویسد، update انجام می‌شود و کش را پاک می‌کند.
3. حالا A نام قدیمی را در کش می‌نویسد.
4. نتیجه: کش دوباره کهنه است، تا TTL تمام شود.

این حالت نادر است، چون باید زمان‌بندی دقیقاً این‌طور باشد. برای همین TTL را حذف نمی‌کنیم. TTL سقف زمانی کهنگی است. اگر لازم است، می‌شود کلید را با یک تأخیر کوتاه دوباره پاک کرد (delayed double delete)، یا یک شماره نسخه در داده نگه داشت.

یک نکته دیگر: اگر Redis در لحظه پاک کردن در دسترس نباشد، چه می‌شود؟ نباید update کاربر شکست بخورد. خطا را لاگ کن و به TTL تکیه کن.

سؤال پیگیری: برای این صفحه، TTL مناسب چقدر است؟ و آیا همه چیز را باید با این روش کش کرد؟

جواب پیگیری:

• جواب یک عدد ثابت نیست. باید پرسید: داده چقدر کهنه می‌تواند باشد بدون این‌که کاربر یا کسب‌وکار ضرر کند؟
• برای نام و عکس پروفایل، چند دقیقه کهنگی برای دیگران معمولاً قابل قبول است. ولی خود کاربر باید تغییرش را فوراً ببیند.
• برای چیزهایی مثل موجودی حساب یا دسترسی‌ها، کهنگی خطرناک است. یا کش نکن، یا خیلی با احتیاط کش کن.
• قبل از کش کردن، اندازه بگیر. شاید یک ایندکس درست مشکل کندی را حل کند و کش اصلاً لازم نباشد.

نشانه خطر: راه حلش فقط «TTL را کم کنیم» است. یا فکر می‌کند پاک کردن کش هیچ race condition ای ندارد.

ویرایش در GitHub

سؤال: سرویس ورود کاربران روی ۶ سرور اجرا می‌شود و روزانه حدود ۲ میلیون درخواست دارد. لاگ‌ها در یک سیستم مرکزی جمع می‌شوند. کد لاگ این‌طور است:

def handle_login(request):
    print("user data: " + str(request.json))
    try:
        user = auth.login(request.json["email"], request.json["password"])
        print("login ok")
        return user
    except Exception as e:
        print("Error happened")
        return error_response(500)

و نمونه خروجی:

user data: {'email': 'sara@example.com', 'password': 'Summer2024!'}
login ok
Error happened
Error happened

این کد را در code review می‌بینی. نظرت چیست؟

جواب کوتاه: دو مشکل جدی دارد. اول، رمز عبور و اطلاعات شخصی در لاگ نوشته می‌شود. این یک مشکل امنیتی است. دوم، لاگ‌ها برای پیدا کردن مشکل بی‌فایده‌اند: نه سطح (level) دارند، نه زمینه (کدام درخواست، کدام کاربر)، نه خود خطا و stack trace. راه حل: لاگ ساختاریافته (structured logging)، با فیلدهای زمینه مثل request id، سطح درست، ثبت کامل exception، و هرگز ثبت رمز و توکن.

سرنخ (اگر کاندید گفت مشکلی نیست): دیشب ۳٪ ورودها شکست خوردند. مهندس کشیک در سیستم لاگ فقط هزاران خط «Error happened» دید. نمی‌دانست خطا چه بود، برای کدام کاربر بود، و روی کدام سرور. یک هفته بعد، تیم امنیت هم متوجه شد که هر کسی به سیستم لاگ دسترسی دارد، رمز کاربران را می‌بیند.

چرا این مشکل است؟ (قدم به قدم)

1. هر کسی که به لاگ دسترسی دارد (توسعه‌دهنده، پشتیبانی، سرویس‌های بیرونی لاگ) حالا رمز کاربران را دارد.
2. لاگ‌ها معمولاً کپی می‌شوند، مدت زیادی نگه داشته می‌شوند و به اندازه دیتابیس محافظت نمی‌شوند.
3. پیام «Error happened» نوع خطا، پیام خطا و محل آن در کد را نمی‌گوید.
4. بدون request id نمی‌شود خط‌های مربوط به یک درخواست را از میان میلیون‌ها خط از ۶ سرور جدا کرد.
5. دستور print سطح ندارد. پس نمی‌شود فقط خطاها را دید، یا در production لاگ‌های جزئی را خاموش کرد.
6. متن آزاد را سخت می‌شود جستجو و شمارش کرد. مثلاً «چند خطا برای این مشتری در یک ساعت گذشته؟»

راه حل:

import logging

logger = logging.getLogger("auth")

def handle_login(request):
    email = request.json.get("email")
    try:
        user = auth.login(email, request.json.get("password"))
        logger.info("login succeeded", extra={"user_id": user.id})
        return user
    except InvalidCredentials:
        logger.warning("login failed: invalid credentials")
        return error_response(401)
    except Exception:
        logger.exception("login crashed")  # logs the stack trace at ERROR level
        return error_response(500)

1. یک formatter خروجی را JSON می‌کند. هر خط فیلدهای جدا دارد: زمان، سطح، پیام، نام سرویس، سرور.
2. یک middleware برای هر درخواست یک request id می‌سازد (یا از هدر ورودی می‌خواند) و آن را به همه لاگ‌های همان درخواست اضافه می‌کند. همین id را در جواب خطا هم به کاربر بده تا پشتیبانی بتواند پیدایش کند.
3. شناسه کاربر را لاگ کن، نه ایمیل و نام او.
4. سطح درست را انتخاب کن:
   • سطح DEBUG برای جزئیات توسعه، معمولاً در production خاموش.
   • سطح INFO برای اتفاق‌های عادی و مهم.
   • سطح WARNING برای چیزی غیرعادی که سیستم از پسش برآمد، مثل رمز اشتباه.
   • سطح ERROR برای خطایی که یک درخواست را خراب کرد، همراه با stack trace.
5. هرگز این‌ها را لاگ نکن: رمز عبور، توکن، کلید API، شماره کامل کارت بانکی. اطلاعات شخصی را فقط وقتی لازم است و با پوشاندن (masking) ثبت کن.
6. یک لایه محافظ هم بگذار: فیلتری که فیلدهایی با نام‌هایی مثل password یا token را قبل از نوشتن حذف می‌کند.

سؤال پیگیری: رمزها تا امروز در لاگ نوشته شده‌اند. حالا چه می‌کنی؟

جواب پیگیری:

• اول کد را درست می‌کند تا دیگر رمزی نوشته نشود.
• لاگ‌های قدیمی که رمز دارند را از سیستم لاگ و نسخه‌های پشتیبان پاک می‌کند.
• این را یک رخداد امنیتی می‌داند و به تیم امنیت خبر می‌دهد. شاید لازم باشد کاربران آسیب‌دیده رمزشان را عوض کنند.
• بعد بررسی می‌کند چه کسانی به آن لاگ‌ها دسترسی داشته‌اند.

نشانه خطر: می‌گوید «لاگ‌ها داخلی‌اند، پس ثبت رمز اشکالی ندارد». یا برای حل مشکل، بیشتر print با متن آزاد اضافه می‌کند.

ویرایش در GitHub

سؤال: یک تیم ۸ نفره روی یک اپ وب فروشگاهی کار می‌کند. این خلاصه وضعیت تست‌های پروژه است:

End-to-end browser tests (Selenium/Playwright):  900
Integration tests (API + database):               40
Unit tests:                                       20
CI pipeline duration:                             ~70 minutes

بیشتر تست‌های مرورگر قانون‌های تجاری را چک می‌کنند. مثلاً «اگر کد تخفیف منقضی شده باشد، پیام خطا نشان داده شود» یا «هزینه ارسال برای سبد بالای یک مبلغ رایگان است». مدیر می‌گوید: «پوشش تست ما عالی است.» نظرت چیست؟ این وضعیت را چطور می‌بینی؟

جواب کوتاه: این ساختار برعکس هرم تست است. بیشتر تست‌ها در گران‌ترین و کندترین لایه‌اند. تست‌های مرورگر کندند و به خاطر شبکه و زمان‌بندی، گاهی بی‌دلیل شکست می‌خورند. راه حل: منطق تجاری را با تست واحد چک کنیم، ارتباط با دیتابیس و API را با تست یکپارچه، و فقط چند تست مرورگر برای مسیرهای حیاتی (مثل ثبت‌نام، خرید، پرداخت) نگه داریم.

سرنخ (اگر کاندید گفت مشکلی نیست): هر pull request بیش از یک ساعت منتظر CI می‌ماند. حدود یک بار از هر چند اجرا، چند تست تصادفی شکست می‌خورند و با اجرای دوباره درست می‌شوند. برنامه‌نویس‌ها حالا عادت کرده‌اند که بدون نگاه کردن، دکمه «اجرای دوباره» را بزنند. هفته پیش یک باگ واقعی هم به همین شکل نادیده گرفته شد.

چرا این مشکل است؟ (قدم به قدم)

1. هر تست مرورگر باید برنامه را بالا بیاورد، مرورگر را باز کند، صفحه‌ها را بارگذاری کند و منتظر بماند. هر تست چند ثانیه طول می‌کشد. تست واحد چند میلی‌ثانیه.
2. تست مرورگر قسمت‌های متحرک زیادی دارد: شبکه، زمان‌بندی صفحه، داده مشترک، سرویس‌های بیرونی. هر کدام می‌تواند باعث شکست تصادفی (flaky) شود.
3. وقتی تست‌ها تصادفی شکست بخورند، تیم به آن‌ها اعتماد نمی‌کند. شکست واقعی هم نادیده گرفته می‌شود.
4. وقتی تست مرورگر شکست بخورد، معلوم نیست مشکل کجاست: در UI، API، دیتابیس، یا خود قانون تجاری.
5. بازخورد ۷۰ دقیقه‌ای یعنی برنامه‌نویس سراغ کار دیگری می‌رود. تغییرهای کوچک‌تر و کمتر merge می‌شوند.
6. قانون «تخفیف منقضی» یک تابع ساده است. برای چک کردنش لازم نیست مرورگر باز شود.

هرم تست:

• پایه پهن: تست‌های واحد زیاد، سریع و جدا. منطق تجاری این‌جا تست می‌شود.
• وسط: تست‌های یکپارچه کمتر. چک می‌کنند که کد با دیتابیس، صف یا API واقعی درست کار می‌کند.
• نوک باریک: چند تست end-to-end. فقط چک می‌کنند که مسیرهای اصلی از اول تا آخر کار می‌کنند.

راه حل:

1. تست‌های مرورگر را دسته‌بندی کن. برای هر کدام بپرس: «این تست واقعاً چه چیزی را چک می‌کند؟»
2. منطق تجاری را به تست واحد منتقل کن. یک تست مرورگر درباره هزینه ارسال می‌تواند به ده تست واحد تبدیل شود که همه حالت‌های مرزی را در کسری از ثانیه چک می‌کنند:

def test_free_shipping_above_threshold():
    assert shipping_cost(cart_total=Decimal("120.00")) == Decimal("0")

def test_shipping_charged_below_threshold():
    assert shipping_cost(cart_total=Decimal("99.99")) == Decimal("7.50")

3. برای این کار، شاید لازم باشد منطق را از کد UI و کنترلر جدا کنی تا قابل تست باشد.
4. فقط چند ده تست مرورگر برای مسیرهای حیاتی نگه دار.
5. تست‌های flaky را پیدا کن. یا درستشان کن، یا موقتاً قرنطینه کن. هرگز به «اجرای دوباره تا سبز شود» عادت نکن.
6. تست‌ها را موازی اجرا کن، و تست‌های سریع را اول اجرا کن تا بازخورد زودتر برسد.
7. این کار را یک‌جا انجام نده. هر بار که به یک بخش دست می‌زنی، تست‌های همان بخش را جابه‌جا کن.

سؤال پیگیری: مدیر نگران است: «اگر ۸۰۰ تست مرورگر را پاک کنیم، پوشش ما کم نمی‌شود؟»

جواب پیگیری:

• پوشش از بین نمی‌رود. همان رفتارها در لایه پایین‌تر چک می‌شوند، حتی با حالت‌های بیشتر.
• هیچ تستی را قبل از این‌که جایگزینش نوشته شده باشد پاک نکن.
• معیار موفقیت فقط تعداد تست نیست. زمان CI، تعداد شکست‌های تصادفی، و باگ‌هایی که به production می‌رسند را اندازه بگیر.
• یک مجموعه تست که کسی به آن اعتماد ندارد، در عمل پوششی نمی‌دهد.

نشانه خطر: راه حلش فقط «سرور CI قوی‌تر بخریم» یا «تست‌های شکست‌خورده را خودکار دوباره اجرا کنیم» است. یا فکر می‌کند تست واحد ارزشی ندارد چون «کاربر واقعی را شبیه‌سازی نمی‌کند».

ویرایش در GitHub

سؤال: یک اپ رزرو نوبت برای کلینیک‌ها داریم. اول فقط در یک شهر کار می‌کرد. حالا کاربرانی در چند کشور دارد. شرکت هم قرار است سرورها را به یک دیتاسنتر در منطقه دیگری منتقل کند. زمان نوبت‌ها این‌طور ذخیره می‌شود و یک کار زمان‌بندی‌شده هر دقیقه یادآوری‌ها را می‌فرستد:

from datetime import datetime, timedelta

def book(user_id, date_str, time_str):
    # e.g. "2026-03-29", "09:30"
    starts_at = datetime.strptime(f"{date_str} {time_str}", "%Y-%m-%d %H:%M")
    db.insert_appointment(user_id, starts_at)  # column type: TIMESTAMP (no time zone)

def send_reminders():
    now = datetime.now()  # server local time
    for appt in db.appointments_between(now + timedelta(hours=1),
                                        now + timedelta(hours=1, minutes=1)):
        notify(appt.user_id, "Your appointment is in 1 hour")

این کد را در code review می‌بینی. نظرت چیست؟ آیا مشکلی دارد؟

جواب کوتاه: زمان بدون منطقه زمانی ذخیره می‌شود. پس «۹:۳۰» معلوم نیست ۹:۳۰ کجاست. سرور هم آن را با ساعت محلی خودش مقایسه می‌کند. با جابه‌جایی سرور، کاربران در کشورهای دیگر، و تغییر ساعت تابستانی (DST)، یادآوری‌ها در ساعت اشتباه یا دو بار فرستاده می‌شوند. راه حل: برای لحظه‌های زمانی UTC ذخیره کن. برای رویدادهای محلی آینده، منطقه زمانی کاربر را هم با نام IANA (مثل Europe/Berlin) نگه دار. تبدیل را فقط در لبه‌ها انجام بده: موقع ورودی گرفتن و موقع نمایش.

سرنخ (اگر کاندید گفت مشکلی نیست): کاربران در یک کشور دیگر می‌گویند یادآوری چند ساعت زودتر یا دیرتر می‌رسد. بعد از انتقال آزمایشی به سرور جدید، یادآوری همه کاربران جابه‌جا شد. و در شبی که ساعت‌ها جلو یا عقب کشیده شدند، بعضی‌ها یادآوری دریافت نکردند و بعضی‌ها دو یادآوری گرفتند.

چرا این اتفاق می‌افتد؟ (قدم به قدم)

1. مقدار «۲۹ مارس، ۹:۳۰» بدون منطقه زمانی یک لحظه مشخص نیست. در تهران، برلین و تورنتو، لحظه‌های متفاوتی است.
2. کد فرض می‌کند منطقه زمانی کاربر و سرور یکی است. برای کاربر کشور دیگر این غلط است.
3. وقتی سرور به منطقه دیگری برود، تابعی که ساعت فعلی سرور را می‌خواند مقدار دیگری برمی‌گرداند. ولی داده‌های ذخیره‌شده عوض نمی‌شوند. پس همه مقایسه‌ها به هم می‌ریزد.
4. در کشورهایی که ساعت تابستانی دارند، یک بار در سال یک ساعت از روز وجود ندارد (ساعت جلو می‌رود)، و یک بار یک ساعت دو بار تکرار می‌شود (ساعت عقب می‌رود).
5. در شب عقب کشیدن ساعت، ساعت محلی سرور یک ساعت را دو بار طی می‌کند. پس یک کار که بر اساس ساعت محلی اجرا می‌شود، ممکن است همان نوبت‌ها را دو بار پیدا کند. در شب جلو کشیدن ساعت، یک ساعت جا می‌افتد و یادآوری‌های آن ساعت فرستاده نمی‌شوند.

دو نوع زمان داریم:

• لحظه‌ای که گذشته یا لحظه دقیق یک اتفاق (زمان ساخت سفارش، زمان ورود). این را همیشه به UTC ذخیره کن.
• رویداد محلی در آینده (نوبت ساعت ۹:۳۰ صبح در کلینیک برلین). کاربر «۹:۳۰ به وقت محلی» را می‌خواهد. اگر قانون ساعت تابستانی آن کشور تا آن روز عوض شود، باید هنوز ۹:۳۰ محلی باشد. پس زمان محلی و نام منطقه زمانی را نگه دار. UTC را از روی آن‌ها حساب کن.

راه حل:

from datetime import datetime, timezone
from zoneinfo import ZoneInfo

def book(user_id, local_str, tz_name):
    # local_str like "2026-03-29 09:30", tz_name like "Europe/Berlin"
    local = datetime.strptime(local_str, "%Y-%m-%d %H:%M").replace(tzinfo=ZoneInfo(tz_name))
    starts_at_utc = local.astimezone(timezone.utc)
    db.insert_appointment(user_id, local_str, tz_name, starts_at_utc)

def send_reminders():
    now = datetime.now(timezone.utc)  # never depends on the server's zone
    for appt in db.due_reminders(now):
        notify(appt.user_id, "Your appointment is in 1 hour")
        db.mark_reminder_sent(appt.id)

1. ستون زمان در دیتابیس از نوعی باشد که منطقه زمانی را می‌فهمد (مثلاً TIMESTAMPTZ در PostgreSQL)، یا صریحاً UTC باشد.
2. منطقه زمانی کاربر یا کلینیک را با نام IANA ذخیره کن، نه با یک اختلاف ثابت مثل «+۲». اختلاف ثابت با ساعت تابستانی عوض می‌شود.
3. منطقه زمانی سرورها را هم UTC بگذار. ولی کد نباید به آن وابسته باشد.
4. یادآوری را با یک پرچم «فرستاده شد» علامت بزن. پس حتی اگر کار دو بار اجرا شود، یادآوری دو بار نمی‌رود. این از بازه‌های دقیقه‌ای شکننده هم امن‌تر است.
5. ساعت‌های وجود نداشته یا تکراری را در ورودی بررسی کن. مثلاً اگر کاربر ساعتی را انتخاب کند که به خاطر ساعت تابستانی وجود ندارد، به او هشدار بده.

سؤال پیگیری: یک نوبت هفتگی «هر دوشنبه ساعت ۱۸» داریم. آن را چطور ذخیره می‌کنی؟

جواب پیگیری:

• قانون تکرار را ذخیره می‌کند، نه یک لیست از لحظه‌های UTC: «دوشنبه‌ها، ۱۸:۰۰، منطقه Europe/Berlin».
• هر نوبت را از روی قانون و منطقه زمانی حساب می‌کند.
• اگر فقط یک زمان UTC ثابت ذخیره کند و هر هفته ۷ روز اضافه کند، بعد از تغییر ساعت تابستانی نوبت یک ساعت جابه‌جا می‌شود.

نشانه خطر: می‌گوید «همه چیز را به ساعت سرور ذخیره کنیم، سرورها همه در یک منطقه‌اند». یا فکر می‌کند ذخیره همه چیز به UTC به تنهایی هر مشکلی را حل می‌کند، حتی برای رویدادهای محلی آینده.

ویرایش در GitHub

سؤال: تو مهندس ارشد در یک تیم ۵ نفره هستی. مدیر محصول پیش تو می‌آید و می‌گوید: «مشتری‌ها می‌خواهند سفارش‌هایشان را به Excel خروجی بگیرند. این قابلیت خروجی Excel چقدر طول می‌کشد؟ همین الان یک عدد به من بده.» هنوز هیچ نیازمندی نوشته‌شده‌ای ندیده‌ای. چه جوابی می‌دهی؟

جواب کوتاه: جواب خوب نه «نمی‌دانم» است و نه یک حدس مثل «۳ روز». جواب خوب این است:

1. چند سؤال سریع می‌پرسد تا دامنه کار روشن شود.
2. کار را به بخش‌های کوچک می‌شکند.
3. چیزهای نامعلوم را نام می‌برد.
4. یک بازه همراه با میزان اطمینان می‌دهد. مثلاً «به احتمال زیاد ۴ تا ۶ روز، و تقریباً مطمئنم کمتر از ۸ روز».
5. اگر یک نامعلوم بزرگ هست، اول یک spike پیشنهاد می‌دهد. یعنی یک آزمایش کوچک با زمان محدود.
6. قول می‌دهد هر وقت چیز تازه‌ای فهمید، تخمین را به‌روز کند.

سرنخ (اگر کاندید گفت مشکلی نیست): مدیر عدد تو را در برنامه می‌گذارد و به مشتری هم می‌گوید. بعد معلوم می‌شود بعضی مشتری‌ها ۲ میلیون سفارش دارند، و خروجی باید اقلام هر سفارش را هم داشته باشد و با فیلترها کار کند. حالا عدد تو چه می‌شود؟

اول چه سؤال‌هایی بپرسیم:

• چه داده‌ای؟ فقط سفارش‌ها، یا سفارش همراه با اقلام و اطلاعات مشتری؟
• چقدر بزرگ؟ ۵۰۰ ردیف یا ۲ میلیون ردیف؟ بهتر است فایل بزرگ را داخل یک درخواست HTTP نسازیم. یک کار پس‌زمینه (background job) لازم دارد.
• چه فرمتی؟ فایل واقعی xlsx، یا یک CSV که در Excel باز می‌شود کافی است؟ ساختن CSV خیلی ساده‌تر است.
• آیا فیلترها، تاریخ، منطقه زمانی و فرمت اعداد مهم است؟
• چه کسی اجازه خروجی گرفتن دارد؟ آیا بررسی دسترسی یا لاگ ممیزی (audit log) لازم است؟

شکستن کار (مثال برای دامنه متوسط):

1. ساختن endpoint در بک‌اند و کوئری با فیلترها.
2. ساختن فایل با یک کتابخانه.
3. کار پس‌زمینه و لینک دانلود، اگر فایل‌ها بزرگ‌اند.
4. دکمه در UI و نشان دادن وضعیت پیشرفت.
5. تست، code review و deploy.

تخمین هر بخش کوچک از تخمین کل قابلیت آسان‌تر است. بخش‌های کوچک کارهایی را هم نشان می‌دهند که معمولاً فراموش می‌شوند، مثل تست و deploy.

چرا بازه و نه یک عدد؟ (قدم به قدم)

1. تخمین یک حدس درباره آینده است. پس عدم قطعیت دارد.
2. یک عدد تنها این عدم قطعیت را پنهان می‌کند. آدم‌ها آن را یک قول می‌شنوند.
3. بازه، ریسک را نشان می‌دهد. بازه پهن یعنی «کم می‌دانیم». بازه باریک یعنی «زیاد می‌دانیم».
4. حالا مدیر می‌تواند تصمیم بگیرد: ریسک را قبول کند، دامنه را کم کند، یا منتظر نتیجه spike بماند.

استفاده از spike برای ریسک: اگر بزرگ‌ترین نامعلوم این است که «آیا می‌توانیم ۲ میلیون ردیف را بدون پر شدن حافظه خروجی بگیریم؟»، نصف روز یا یک روز وقت می‌گذاریم و امتحانش می‌کنیم. بعد از spike، بازه خیلی باریک‌تر می‌شود.

یک نمونه جواب: «اگر فقط CSV از سفارش‌ها با همین فیلترهای فعلی باشد، حدود ۲ تا ۳ روز. اگر Excel واقعی با اقلام و فایل‌های بزرگ لازم است، نزدیک ۶ تا ۱۰ روز. بگذار نصف روز اندازه داده و کتابخانه را بررسی کنم، بعد عدد دقیق‌تری می‌دهم.»

سؤال پیگیری: مدیر می‌گوید: «برای نقشه راه فقط یک عدد لازم دارم، بازه نمی‌خواهم.» چه می‌کنی؟

جواب پیگیری:

• یک عدد می‌دهد، ولی فرض‌هایش را روشن می‌گوید. مثلاً: «۸ روز، با این فرض که دامنه شبیه CSV است و خروجی در پس‌زمینه ساخته می‌شود.»
• عددی نزدیک سقف بازه انتخاب می‌کند، نه خوش‌بینانه‌ترین عدد.
• یک نسخه اول کوچک‌تر پیشنهاد می‌دهد (مثلاً فقط CSV) که زودتر منتشر شود.
• فرض‌ها را می‌نویسد. اگر یکی عوض شود، تخمین هم عوض می‌شود و همه می‌دانند چرا.
• وقتی چیز تازه‌ای می‌فهمد، زود تخمین را به‌روز می‌کند. خبر بد زودهنگام خیلی بهتر از خبر بد در روز آخر است.

نشانه خطر: بدون هیچ سؤالی یک عدد قطعی می‌دهد، یا اصلاً حاضر نیست هیچ تخمینی بدهد.

ویرایش در GitHub

سؤال: یکی از هم‌تیمی‌ها تابعی نوشته که شناسه یک محصول را در یک لیست مرتب پیدا می‌کند. لیست حدود ۵۰٬۰۰۰ شناسه دارد. کد این است:

def binary_search(a: list[int], target: int) -> int:
    low = 0
    high = len(a) - 1
    while low < high:
        mid = (low + high) // 2
        if a[mid] == target:
            return mid
        if a[mid] < target:
            low = mid + 1
        else:
            high = mid - 1
    return -1

تست‌ها چند شناسه را در وسط یک لیست ۱۰تایی جستجو می‌کنند و پاس می‌شوند. این کد را در code review می‌بینی. نظرت چیست؟ چه تست‌هایی اضافه می‌کنی؟

جواب کوتاه: یک باگ off-by-one (اشتباه یکی کم یا زیاد) دارد. اینجا high شامل است، یعنی به یک عنصر واقعی اشاره می‌کند. پس بازه جستجو از low تا high است و هر دو سر آن حساب می‌شوند. وقتی low با high برابر می‌شود، هنوز یک عنصر مانده که باید بررسی شود. ولی شرط حلقه، یعنی low < high، حلقه را تمام می‌کند و آن عنصر آخر هیچ وقت مقایسه نمی‌شود. راه حل: شرط حلقه باید low <= high باشد.

سرنخ (اگر کاندید گفت مشکلی نیست): یک لیست با یک عضو، یعنی ۵، را امتحان کن و دنبال ۵ بگرد. بعد یک لیست با دو عضو، یعنی ۱ و ۳، را امتحان کن و دنبال ۳ بگرد. تابع چه برمی‌گرداند؟

چرا این اتفاق می‌افتد؟ (قدم به قدم)

1. لیست شامل ۱ و ۳ است و هدف ۳ است. در شروع، low برابر ۰ و high برابر ۱ است.
2. حلقه اجرا می‌شود. مقدار mid برابر ۰ است. عنصر خانه ۰ یعنی ۱، از ۳ کوچک‌تر است. پس low برابر ۱ می‌شود.
3. حالا low و high هر دو ۱ هستند. بازه هنوز یک عنصر دارد: عنصر خانه ۱ که همان ۳ است.
4. ولی شرط «۱ کوچک‌تر از ۱» نادرست است. حلقه تمام می‌شود و منفی ۱ برمی‌گردد.
5. با لیست یک‌عضوی بدتر است: low و high هر دو ۰ هستند و حلقه اصلاً اجرا نمی‌شود.

با invariant فکر کن: یک invariant قانونی است که در هر قدم حلقه درست است. اینجا قانون این است: «اگر هدف در لیست باشد، جایی بین low و high است و هر دو سر هم حساب می‌شوند». با این قانون:

• بازه فقط وقتی خالی است که low از high بزرگ‌تر شود. پس حلقه باید تا وقتی low <= high است اجرا شود.
• بعد از بررسی mid، باید mid را از بازه بیرون کنیم. برای همین mid + 1 و mid - 1 را می‌نویسیم.

یک باگ رایج دیگر: بعضی نسخه‌ها از بازه نیمه‌باز استفاده می‌کنند: high از طول لیست شروع می‌شود (نه طول منهای یک)، حلقه تا وقتی low < high است اجرا می‌شود، و در شاخه else مقدار high برابر mid می‌شود. این هم درست است. ولی اگر کسی low را برابر mid بگذارد و نه mid + 1، حلقه ممکن است هیچ وقت تمام نشود. وقتی high دقیقاً یکی بیشتر از low است، mid با low برابر می‌شود و low هیچ وقت جلو نمی‌رود.

راه حل:

def binary_search(a: list[int], target: int) -> int:
    low, high = 0, len(a) - 1
    while low <= high:
        mid = low + (high - low) // 2
        if a[mid] == target:
            return mid
        if a[mid] < target:
            low = mid + 1
        else:
            high = mid - 1
    return -1

تست‌هایی که باید اضافه شوند (حالت‌های مرزی):

• لیست خالی.
• لیست یک‌عضوی: هم وقتی هدف پیدا می‌شود، هم وقتی پیدا نمی‌شود.
• هدف اولین عنصر باشد، و آخرین عنصر.
• هدف از همه کوچک‌تر باشد، از همه بزرگ‌تر باشد، و بین دو عنصر باشد.
• لیست دوعضوی، و جستجوی هر کدام از دو عنصر.

چرا low + (high - low) // 2؟ در پایتون، int محدودیت اندازه ندارد. پس low + high سرریز (overflow) نمی‌کند. ولی در زبان‌هایی که int اندازه ثابت دارد، مثل Java یا C#، جمع low + high ممکن است از بزرگ‌ترین int بیشتر شود و منفی شود. این یک باگ معروف است که سال‌ها در جستجوی دودویی کتابخانه استاندارد Java وجود داشت. شکل low + (high - low) / 2 این مشکل را ندارد.

سؤال پیگیری: در یک پروژه واقعی، جستجوی دودویی را خودت می‌نویسی؟

جواب پیگیری:

• معمولاً نه. پایتون ماژول bisect را دارد که تست‌شده و سریع است. زبان‌های دیگر هم تابع‌های مشابه دارند.
• همان‌طور که این کد نشان می‌دهد، اشتباه کردن در جستجوی دودویی آسان است. کتابخانه این ریسک را حذف می‌کند.
• اگر لیست زیاد عوض می‌شود، شاید لیست مرتب ساختار داده درستی نباشد. یک set یا dict به طور میانگین جستجوی O(1) دارد.
• نوشتنش با دست هنوز در مصاحبه منطقی است، یا وقتی جستجو روی چیز خاصی است، مثل «اولین روزی که فروش از یک حد بیشتر شد».

نشانه خطر: فقط حالت عادی وسط لیست را تست می‌کند و هیچ وقت به ورودی خالی، لیست یک‌عضوی، یا دو سر لیست فکر نمی‌کند.

ویرایش در GitHub

سؤال: یک سرویس بانکی کوچک، حساب‌ها را در حافظه نگه می‌دارد. چندین thread همزمان درخواست‌های انتقال پول را انجام می‌دهند. هر حساب قفل (lock) خودش را دارد تا انتقال بین حساب‌های مختلف بتواند موازی اجرا شود. کد این است:

import threading

class Account:
    def __init__(self, account_id: int, balance: int):
        self.id = account_id
        self.balance = balance
        self.lock = threading.Lock()

def transfer(src: Account, dst: Account, amount: int) -> None:
    with src.lock:
        with dst.lock:
            src.balance -= amount
            dst.balance += amount

این کد را در code review می‌بینی. نظرت چیست؟ آیا مشکلی دارد؟

جواب کوتاه: این کد ممکن است دچار deadlock (بن‌بست) شود. اگر یک thread انتقال از a به b را اجرا کند و thread دیگر همزمان انتقال از b به a را، اولی قفل a را می‌گیرد و منتظر b می‌ماند. دومی قفل b را می‌گیرد و منتظر a می‌ماند. هر دو برای همیشه منتظر می‌مانند. راه حل معمول یک ترتیب ثابت برای قفل‌ها است: همیشه اول قفل حسابی را بگیر که شناسه کوچک‌تری دارد.

سرنخ (اگر کاندید گفت مشکلی نیست): در تست‌ها همه چیز درست کار می‌کند. در production، چند بار در هفته، سرویس دیگر جواب نمی‌دهد. مصرف CPU نزدیک صفر است. یک thread dump نشان می‌دهد دو thread هر کدام منتظر گرفتن یک قفل‌اند. ری‌استارت سرویس مشکل را برای مدتی حل می‌کند.

چرا این اتفاق می‌افتد؟ (قدم به قدم)

1. اولین thread انتقال از a به b را شروع می‌کند و قفل a را می‌گیرد.
2. در همان لحظه، thread دوم انتقال از b به a را شروع می‌کند و قفل b را می‌گیرد.
3. حالا thread اول قفل b را می‌خواهد. ولی دست thread دوم است، پس منتظر می‌ماند.
4. حالا thread دوم قفل a را می‌خواهد. ولی دست thread اول است، پس منتظر می‌ماند.
5. هیچ‌کدام قفلش را رها نمی‌کند. هر دو برای همیشه گیر کرده‌اند، و هر درخواست دیگری هم که به a یا b نیاز دارد گیر می‌کند.

این اتفاق کم پیش می‌آید، چون زمان‌بندی باید دقیقاً جور شود. برای همین تست‌ها آن را نمی‌گیرند.

چهار شرط deadlock: بن‌بست فقط وقتی ممکن است که هر چهار شرط همزمان درست باشند:

1. انحصار متقابل (mutual exclusion): فقط یک thread می‌تواند یک قفل را داشته باشد.
2. نگه‌داشتن و منتظر ماندن (hold and wait): یک thread یک قفل را نگه می‌دارد و منتظر قفل دیگری است.
3. نبود پس‌گرفتن (no preemption): هیچ کس نمی‌تواند قفل را به زور از یک thread بگیرد.
4. انتظار چرخشی (circular wait): اولی منتظر دومی است و دومی منتظر اولی.

اگر فقط یکی از این‌ها را بشکنیم، بن‌بست ممکن نیست. معمولاً شکستن انتظار چرخشی از همه آسان‌تر است.

راه حل: ترتیب ثابت برای قفل‌ها

def transfer(src: Account, dst: Account, amount: int) -> None:
    if src.id == dst.id:
        return
    first, second = (src, dst) if src.id < dst.id else (dst, src)
    with first.lock:
        with second.lock:
            src.balance -= amount
            dst.balance += amount

1. حالا هر دو انتقال، چه از a به b و چه از b به a، اول قفل شناسه کوچک‌تر را می‌گیرند.
2. پس هر دو thread سر همان قفل اول رقابت می‌کنند. یکی برنده می‌شود و دیگری بدون نگه داشتن هیچ قفلی منتظر می‌ماند.
3. چرخه‌ای وجود ندارد، پس بن‌بست هم نداریم.
4. انتقال به همان حساب را هم جدا بررسی کرده‌ایم. بدون این بررسی، thread سعی می‌کند یک قفل را دو بار بگیرد و خودش را قفل می‌کند، چون Lock معمولی reentrant نیست.

راه دیگر: timeout. یک thread می‌تواند قفل دوم را با یک timeout امتحان کند. اگر نگرفت، قفل اول را رها می‌کند، کمی صبر تصادفی می‌کند و دوباره تلاش می‌کند. این کار شرط «نگه‌داشتن و منتظر ماندن» را می‌شکند. ولی پیچیده‌تر است و زیر بار زیاد باعث تلاش‌های تکراری می‌شود. معمولاً ترتیب روشن قفل‌ها ساده‌تر است.

همین ایده در دیتابیس: دو تراکنش ممکن است دو ردیف یکسان را با ترتیب برعکس آپدیت کنند. دیتابیس‌هایی مثل PostgreSQL، MySQL و SQL Server این را تشخیص می‌دهند، یکی از تراکنش‌ها را با خطای deadlock لغو می‌کنند و اجازه می‌دهند دیگری ادامه دهد. پس برنامه باید:

• ردیف‌ها را با یک ترتیب ثابت آپدیت کند، مثلاً بر اساس شناسه حساب.
• وقتی خطای deadlock گرفت، تراکنش را دوباره اجرا کند.

سؤال پیگیری: چرا برای همه انتقال‌ها فقط یک قفل بزرگ سراسری نگذاریم؟ آن که بن‌بست ندارد.

جواب پیگیری:

• درست است. یک قفل تنها بن‌بست نمی‌سازد و کد ساده‌تر است.
• ولی در هر لحظه فقط یک انتقال می‌تواند اجرا شود. با thread های زیاد، همه در صف می‌مانند و توان پردازش (throughput) پایین می‌آید.
• اگر ترافیک کم است، یک قفل شاید کافی باشد. ساده و درست بهتر از سریع و خراب است.
• اگر انتقال موازی لازم داریم، قفل هر حساب را نگه می‌داریم و ترتیب ثابت را رعایت می‌کنیم.

نشانه خطر: می‌گوید «یک sleep اضافه کنیم» یا «هر وقت گیر کرد ری‌استارت کنیم»، یا نمی‌تواند توضیح دهد که ترتیب برعکس قفل‌ها علت مشکل است.

ویرایش در GitHub

سؤال: یک اپ کیف پول به کاربرها اجازه برداشت پول می‌دهد. سرویس روی ۴ نمونه (instance) پشت یک load balancer اجرا می‌شود و همه از یک دیتابیس PostgreSQL استفاده می‌کنند. کد برداشت این است:

def withdraw(conn, user_id: int, amount: int) -> None:
    with conn.transaction():
        row = conn.execute(
            "SELECT balance FROM wallets WHERE user_id = %s", (user_id,)
        ).fetchone()
        if row.balance < amount:
            raise InsufficientFunds()
        conn.execute(
            "UPDATE wallets SET balance = %s WHERE user_id = %s",
            (row.balance - amount, user_id),
        )

کد داخل یک تراکنش اجرا می‌شود و تست‌ها پاس می‌شوند. این کد را در code review می‌بینی. نظرت چیست؟ آیا مشکلی دارد؟

جواب کوتاه: این یک race از نوع check-then-act است که به آن lost update (آپدیت گم‌شده) هم می‌گویند. بررسی (خواندن موجودی) و عمل (نوشتن موجودی جدید) دو قدم جدا هستند. دو برداشت همزمان می‌توانند هر دو همان موجودی قدیمی را بخوانند، هر دو از بررسی رد شوند، و هر دو بنویسند. تراکنش در سطح ایزولاسیون پیش‌فرض (در PostgreSQL یعنی READ COMMITTED) جلوی این را نمی‌گیرد. بهترین راه حل یک UPDATE اتمیک است که بررسی و تغییر را در یک دستور انجام دهد.

سرنخ (اگر کاندید گفت مشکلی نیست): کاربری ۱۰۰ در کیف پول دارد. دو بار خیلی سریع روی «برداشت ۸۰» می‌زند، یا اپ موبایل درخواست را دوباره می‌فرستد. گاهی هر دو درخواست موفق می‌شوند. حالا موجودی یا منفی ۶۰ است یا ۲۰، ولی کاربر ۱۶۰ گرفته است. کدام یکی، به زمان‌بندی بستگی دارد.

چرا این اتفاق می‌افتد؟ (قدم به قدم)

1. درخواست A موجودی را می‌خواند: ۱۰۰.
2. درخواست B روی یک نمونه دیگر، موجودی را می‌خواند: باز هم ۱۰۰. چون A هنوز چیزی ننوشته است.
3. درخواست A بررسی می‌کند که ۱۰۰ از ۸۰ کمتر نیست. درست است. درخواست B هم همین را بررسی می‌کند. باز هم درست است.
4. درخواست A مقدار ۲۰ را می‌نویسد. درخواست B هم ۲۰ را می‌نویسد.
5. دو برداشت ۸۰تایی انجام شد، ولی موجودی ۲۰ است. یک آپدیت گم شد. اگر کد به شکل «موجودی منهای ۸۰» در SQL می‌نوشت، موجودی منفی ۶۰ می‌شد.

خود تراکنش کمکی نمی‌کند. در سطح READ COMMITTED، هر دستور آخرین داده commit شده را می‌بیند، ولی هیچ چیز ردیف را بین SELECT و UPDATE قفل نمی‌کند.

راه حل ۱ (بهترین): یک UPDATE اتمیک

UPDATE wallets
SET balance = balance - %(amount)s
WHERE user_id = %(user_id)s AND balance >= %(amount)s;

1. دیتابیس بررسی و تغییر را با هم، روی یک ردیف قفل‌شده، انجام می‌دهد.
2. اگر درخواست دوم همزمان برسد، منتظر اولی می‌ماند و بعد موجودی جدید یعنی ۲۰ را می‌بیند.
3. حالا شرط «۲۰ دست‌کم ۸۰ است» نادرست است، پس هیچ ردیفی آپدیت نمی‌شود.
4. در کد، اگر تعداد ردیف‌های آپدیت‌شده صفر بود، خطای InsufficientFunds می‌دهیم.

راه حل ۲: قفل کردن ردیف با SELECT FOR UPDATE

SELECT balance FROM wallets WHERE user_id = %s FOR UPDATE;

ردیف تا آخر تراکنش قفل می‌ماند. درخواست دوم روی همین خط منتظر می‌ماند. این روش وقتی به درد می‌خورد که منطق بین خواندن و نوشتن برای یک UPDATE تنها خیلی پیچیده است. تراکنش را کوتاه نگه دار، چون بقیه منتظر می‌مانند.

راه حل ۳: همزمانی خوش‌بینانه با ستون version

UPDATE wallets SET balance = %s, version = version + 1
WHERE user_id = %s AND version = %s;

موجودی و version را می‌خوانیم. موقع نوشتن، شرط می‌گذاریم که version هنوز همان باشد. اگر کسی در این فاصله ردیف را عوض کرده باشد، هیچ ردیفی آپدیت نمی‌شود. پس دوباره می‌خوانیم و تلاش می‌کنیم. این روش وقتی خوب است که تداخل کم پیش می‌آید.

یک تور ایمنی: یک CHECK constraint اضافه کن که موجودی هیچ وقت زیر صفر نرود. این طوری حتی یک باگ در آینده هم نمی‌تواند موجودی را منفی کند.

سؤال پیگیری: آیا فقط با بالا بردن سطح ایزولاسیون به SERIALIZABLE مشکل حل می‌شود؟

جواب پیگیری:

• بله. در SERIALIZABLE دیتابیس طوری رفتار می‌کند که انگار تراکنش‌ها یکی‌یکی اجرا شده‌اند. یکی از دو برداشت با خطای serialization شکست می‌خورد.
• ولی برنامه باید این خطا را بگیرد و تراکنش را دوباره اجرا کند. خیلی از تیم‌ها این را فراموش می‌کنند.
• زیر بار سنگین، ممکن است تعداد زیادی تراکنش شکست بخورد و تکرار شود.
• برای این مورد ساده، UPDATE اتمیک روشن‌تر و ارزان‌تر است و در سطح پیش‌فرض هم امن است.

نشانه خطر: فکر می‌کند «داخل تراکنش است، پس امن است»، یا یک قفل در کد پایتون پیشنهاد می‌دهد که با ۴ نمونه سرویس هیچ کمکی نمی‌کند.

ویرایش در GitHub

سؤال: یک سرویس API پروفایل کاربران را نشان می‌دهد. ساختن هر پروفایل ۳ کوئری دیتابیس لازم دارد، پس یکی از هم‌تیمی‌ها «برای سرعت» یک کش اضافه کرده است. سرویس حدود ۲ میلیون کاربر دارد و بین دو deploy چند هفته روشن می‌ماند. کد این است:

_profile_cache: dict[int, dict] = {}

def get_profile(user_id: int) -> dict:
    if user_id in _profile_cache:
        return _profile_cache[user_id]
    profile = build_profile(user_id)  # 3 database queries
    _profile_cache[user_id] = profile
    return profile

زمان پاسخ خیلی بهتر شده است. این کد را در code review می‌بینی. نظرت چیست؟ آیا مشکلی دارد؟

جواب کوتاه: این کش هیچ محدودیت اندازه‌ای ندارد و هیچ زمان انقضایی هم ندارد. هر شناسه کاربر جدید یک ورودی اضافه می‌کند و هیچ چیزی ورودی‌ها را پاک نمی‌کند. پس حافظه تا وقتی پروسس زنده است بزرگ‌تر می‌شود. کش بی‌حد یعنی نشت حافظه (memory leak). باید آن را محدود کنیم: یک حداکثر اندازه با قانون حذف مثل LRU، یا یک زمان انقضا (TTL)، یا یک کش بیرونی مثل Redis.

سرنخ (اگر کاندید گفت مشکلی نیست): بعد از deploy، حافظه ۳۰۰ مگابایت است. هر روز کمی بیشتر می‌شود. بعد از حدود ۴ روز، container به سقف حافظه‌اش می‌رسد و kill می‌شود (OOM). دوباره بالا می‌آید و همین چرخه تکرار می‌شود. بعضی کاربرها هم می‌گویند اسمشان را عوض کرده‌اند ولی هنوز اسم قدیمی نشان داده می‌شود.

چرا این اتفاق می‌افتد؟ (قدم به قدم)

1. هر صدا زدن با یک شناسه کاربر جدید، یک پروفایل به dict اضافه می‌کند.
2. این dict در سطح ماژول است، پس به اندازه عمر پروسس زنده می‌ماند.
3. هیچ چیزی ورودی‌ها را پاک نمی‌کند. در طول چند روز، کاربرهای متفاوت بیشتری سر می‌زنند.
4. با ۲ میلیون کاربر، این dict ممکن است در نهایت میلیون‌ها پروفایل نگه دارد.
5. حافظه تا سقف container بالا می‌رود و پروسس kill می‌شود.

دو مشکل دیگر:

• داده کهنه (stale): وقتی کاربر پروفایلش را عوض می‌کند، کش هنوز نسخه قدیمی را برمی‌گرداند. نه انقضایی هست و نه پاک‌سازی.
• چند پروسس: اگر سرویس ۸ پروسس worker داشته باشد، هر کدام نسخه خودش از کش را دارد. پس مصرف حافظه ۸ برابر می‌شود و worker های مختلف ممکن است داده‌های متفاوت برگردانند.

راه حل:

1. برای یک کش ساده داخل پروسس با محدودیت اندازه، از دکوریتور lru_cache در ماژول functools با یک maxsize استفاده کن. وقتی پر شود، ورودی‌ای را حذف می‌کند که از همه دیرتر استفاده شده است.

from functools import lru_cache

@lru_cache(maxsize=10_000)
def get_profile(user_id: int) -> dict:
    return build_profile(user_id)

2. دقت کن که lru_cache زمان انقضا ندارد. اگر داده عوض می‌شود، TTL اضافه کن. کتابخانه بیرونی cachetools یک کلاس TTLCache دارد که هم حداکثر اندازه دارد و هم زمان انقضا.
3. یک تله دیگر: lru_cache هر بار همان شیء dict را برمی‌گرداند. اگر کسی آن را تغییر دهد، مقدار داخل کش برای همه عوض می‌شود. یک کپی یا یک شیء تغییرناپذیر برگردان.
4. اگر چند پروسس یا چند سرور به همان داده نیاز دارند، از یک کش بیرونی مثل Redis با TTL روی هر کلید استفاده کن. این طوری فقط یک نسخه داریم و حافظه بیرون از پروسس برنامه است.

چطور اندازه بگیریم و ثابت کنیم:

• حافظه پروسس را در طول زمان روی یک داشبورد ببین. خطی که فقط بالا می‌رود یک نشانه هشدار است.
• تعداد ورودی‌های کش را بشمار. برای lru_cache، متد cache_info تعداد hit، miss و اندازه فعلی را نشان می‌دهد.
• در پایتون، ماژول tracemalloc نشان می‌دهد کدام خط‌های کد بیشترین حافظه را می‌گیرند.

سؤال پیگیری: حداکثر اندازه و TTL را چطور انتخاب می‌کنی؟

جواب پیگیری:

• برای حداکثر اندازه، اندازه یک ورودی را تخمین می‌زنیم و تصمیم می‌گیریم کش چقدر حافظه می‌تواند بگیرد. مثلاً اگر هر پروفایل حدود ۲ کیلوبایت باشد و ۵۰ مگابایت اجازه بدهیم، حدود ۲۵٬۰۰۰ ورودی می‌شود.
• بعد نرخ hit را در production نگاه می‌کنیم. اگر از قبل بالاست، کش بزرگ‌تر سود کمی دارد.
• برای TTL، از تیم کسب‌وکار می‌پرسیم داده چقدر می‌تواند قدیمی باشد. نشان دادن اسمی که ۵ دقیقه قدیمی است شاید مشکلی نباشد. ولی موجودی حسابی که ۵ دقیقه قدیمی است مشکل است.
• با هر آپدیت پروفایل، کلید کش را هم پاک می‌کنیم تا کاربر تغییر خودش را فوراً ببیند.

نشانه خطر: کش را بدون محدودیت اندازه یا انقضا اضافه می‌کند، یا مشکل را با «حافظه بیشتر به container بدهیم» حل می‌کند.

ویرایش در GitHub

سؤال: یک فروشگاه آنلاین یک درگاه پرداخت بیرونی را از طریق HTTP صدا می‌زند. در ساعت شلوغی، فروشگاه حدود ۲۰۰ درخواست پرداخت در ثانیه می‌فرستد. درگاه گاهی خطا می‌دهد، پس یکی از هم‌تیمی‌ها retry اضافه کرده است:

def charge(order_id: str, amount: int) -> dict:
    for attempt in range(10):
        try:
            resp = http.post(PROVIDER_URL, json={"order": order_id, "amount": amount}, timeout=5)
            resp.raise_for_status()
            return resp.json()
        except Exception:
            continue
    raise PaymentFailed(order_id)

این کد را در code review می‌بینی. نظرت چیست؟ آیا مشکلی دارد؟

جواب کوتاه: چند مشکل دارد:

1. تلاش دوباره فوراً و بدون هیچ صبری انجام می‌شود. وقتی درگاه قطع است، این کار ترافیک را تا ۱۰ برابر می‌کند و نمی‌گذارد درگاه دوباره سرپا شود. به exponential backoff همراه با jitter نیاز داریم.
2. برای همه خطاها دوباره تلاش می‌کند، حتی خطاهایی که هیچ وقت موفق نمی‌شوند، مثل 400 Bad Request.
3. پرداخت به طور پیش‌فرض قابل تکرار امن نیست. اگر درخواست اول کارت را شارژ کرده باشد ولی جواب گم شده باشد (timeout)، تلاش دوباره ممکن است دو بار از مشتری پول بگیرد. به یک idempotency key نیاز داریم.
4. هیچ retry budget یا circuit breaker ندارد تا وقتی درگاه آشکارا از کار افتاده، دیگر تلاش نکند.

سرنخ (اگر کاندید گفت مشکلی نیست): یک روز درگاه ۲ دقیقه مشکل کوتاهی دارد. ترافیک ما به آن‌ها از ۲۰۰ به نزدیک ۲۰۰۰ درخواست در ثانیه می‌رسد. مشکل آن‌ها به جای ۲ دقیقه، ۴۰ دقیقه طول می‌کشد. بعداً پشتیبانی ایمیل‌هایی از مشتری‌هایی می‌گیرد که دو بار از آن‌ها پول کم شده است.

چرا این اتفاق می‌افتد؟ (قدم به قدم)

1. درگاه کند می‌شود و شروع به خطا دادن می‌کند.
2. هر درخواست ناموفق، فوراً و تا ۱۰ بار دوباره فرستاده می‌شود. پس هر پرداخت واقعی تا ۱۰ درخواست می‌شود.
3. حالا درگاه حدود ۱۰ برابر بار بیشتر می‌گیرد، درست وقتی که از همیشه ضعیف‌تر است.
4. بار بیشتر یعنی خطای بیشتر، و خطای بیشتر یعنی retry بیشتر. به این retry storm (طوفان تلاش دوباره) می‌گویند.
5. اگر لایه‌های دیگر هم retry کنند (اپ موبایل، API gateway)، عددها در هم ضرب می‌شوند. سه لایه که هر کدام ۳ بار تلاش کنند، یک کلیک را به ۲۷ درخواست تبدیل می‌کنند.

راه حل:

1. صبر با jitter: بعد از هر شکست بیشتر صبر کن، مثلاً حدود ۰٫۲، ۰٫۴ و ۰٫۸ ثانیه، همراه با یک بخش تصادفی. بخش تصادفی نمی‌گذارد همه کلاینت‌ها در یک لحظه دوباره تلاش کنند.
2. فقط خطاهای قابل تکرار: timeout، خطای اتصال، 503 و 429 Too Many Requests. اگر درگاه هدر Retry-After فرستاد، به آن احترام بگذار. برای 400، 401 یا جواب «کارت رد شد» دوباره تلاش نکن.
3. تلاش کمتر: معمولاً ۳ بار کافی است. ۱۰ بار بیشتر فقط بار اضافه می‌سازد.
4. کلید idempotency: اگر درگاه پشتیبانی می‌کند، با هر تلاش همان کلید یکتا را بفرست (مثلاً ساخته‌شده از شناسه سفارش). این طوری درگاه فقط یک بار پول کم می‌کند، حتی اگر درخواست را دو بار بگیرد.

import random, time

RETRYABLE = {429, 502, 503, 504}

def charge(order_id: str, amount: int) -> dict:
    headers = {"Idempotency-Key": f"charge-{order_id}"}
    for attempt in range(3):
        try:
            resp = http.post(PROVIDER_URL, json={"order": order_id, "amount": amount},
                             headers=headers, timeout=5)
            if resp.status_code not in RETRYABLE:
                resp.raise_for_status()  # 4xx: fail fast, no retry
                return resp.json()
        except (TimeoutError, ConnectionError):
            pass
        time.sleep(random.uniform(0, 0.2 * 2 ** attempt))  # backoff with full jitter
    raise PaymentFailed(order_id)

بودجه retry و circuit breaker:

• بودجه retry: تلاش‌های دوباره حداکثر می‌توانند سهم کوچکی از همه درخواست‌ها باشند، مثلاً ۱۰ درصد. اگر بودجه تمام شد، دیگر تلاش نمی‌کنیم و سریع خطا می‌دهیم.
• مدارشکن (circuit breaker): اگر در زمان کوتاهی درخواست‌های زیادی شکست بخورند، مدارشکن «باز» می‌شود. برای مدتی اصلاً درگاه را صدا نمی‌زنیم و سریع خطا می‌دهیم. بعد از یک مکث، چند درخواست آزمایشی را رد می‌کنیم (حالت half-open). اگر موفق شدند، مدارشکن را دوباره می‌بندیم.
• هر دو هم از درگاه محافظت می‌کنند و هم از thread های خود ما، تا منتظر یک سرویس مرده نمانند.

سؤال پیگیری: مدارشکن باز است. مشتری در صفحه پرداخت چه می‌بیند؟

جواب پیگیری:

• نه یک صفحه خطای خام. یک پیام روشن نشان می‌دهیم: «پرداخت موقتاً در دسترس نیست، لطفاً چند دقیقه دیگر دوباره امتحان کنید.»
• اگر کسب‌وکار اجازه بدهد، سفارش را با وضعیت «در انتظار پرداخت» ذخیره می‌کنیم و بعداً از یک صف کامل می‌کنیم.
• اگر یک درگاه پرداخت دوم داریم، شاید بشود به آن رفت. ولی این یک تصمیم کسب‌وکاری و قراردادی است.
• به مهندس on-call هشدار می‌دهیم، چون مدارشکن باز یعنی پول واقعی وارد نمی‌شود.

نشانه خطر: فکر می‌کند «retry بیشتر یعنی سیستم مطمئن‌تر»، یا پرداخت را بدون idempotency key دوباره تلاش می‌کند و به پرداخت دوباره فکر نمی‌کند.

ویرایش در GitHub

سؤال: تیم شما حدود ۶۰۰ تست دارد. یک تست integration برای ساختن سفارش، تقریباً از هر ۱۰ اجرا یک بار در CI شکست می‌خورد. هیچ کس تا حالا شکستش را روی لپ‌تاپ ندیده است. وقتی شکست می‌خورد، تیم روی «re-run» می‌زند و معمولاً پاس می‌شود. CI تست‌ها را در چند worker موازی، روی یک دیتابیس تست مشترک اجرا می‌کند. تو به عنوان مهندس ارشد وارد تیم شده‌ای. نظرت چیست؟ چه کار می‌کنی؟

جواب کوتاه: تست flaky (ناپایدار) یک مشکل واقعی است، نه بدشانسی. یا خود تست خراب است، یا کد یک باگ واقعی دارد (مثلاً race condition) که فقط گاهی دیده می‌شود. اجرای دوباره آن را پنهان می‌کند و به تیم یاد می‌دهد build قرمز را نادیده بگیرد. راه درست این است: تکرارش کن، علتش را پیدا کن و درستش کن. اگر امروز درست نمی‌شود، آن را قرنطینه کن، ولی فقط با یک مسئول و یک مهلت.

سرنخ (اگر کاندید گفت مشکلی نیست): ماه پیش یک باگ واقعی به production رسید. CI برای آن تغییر build قرمز نشان داده بود، ولی همه فکر کردند «باز همان تست flaky است» و دوباره اجرایش کردند. هر اجرای دوباره هم حدود ۱۵ دقیقه تیم را منتظر نگه می‌دارد.

علت‌های رایج تست flaky:

1. وضعیت مشترک بین تست‌ها: یک تست داده‌ای در دیتابیس یا یک متغیر سراسری جا می‌گذارد و تست دیگری به آن وابسته است.
2. وابستگی به ترتیب: یک تست فقط وقتی پاس می‌شود که تست دیگری قبلش اجرا شده باشد (یا نشده باشد).
3. اجرای موازی روی یک دیتابیس: دو worker همزمان سفارشی با همان ایمیل یا شناسه می‌سازند، یا یک worker ردیف‌هایی را پاک می‌کند که worker دیگر لازم دارد.
4. زمان‌بندی و sleep: تست ۱ ثانیه صبر می‌کند و امیدوار است کار پس‌زمینه تمام شده باشد. روی ماشین کند CI، تمام نشده است.
5. زمان واقعی: تست از تاریخ فعلی استفاده می‌کند و نزدیک نیمه‌شب، آخر ماه، یا در یک منطقه زمانی دیگر شکست می‌خورد. سرورهای CI اغلب با UTC کار می‌کنند.
6. شبکه: تست یک سرویس بیرونی واقعی را صدا می‌زند که گاهی کند یا قطع است.
7. داده تصادفی یا نتیجه بی‌ترتیب: مثلاً کوئری‌ای بدون ORDER BY که معمولاً، ولی نه همیشه، ردیف‌ها را با همان ترتیب برمی‌گرداند.

چرا روی لپ‌تاپ پاس می‌شود؟ (قدم به قدم)

1. روی لپ‌تاپ، تست‌ها معمولاً یکی‌یکی، با ترتیب ثابت، روی یک ماشین سریع اجرا می‌شوند.
2. در CI، موازی اجرا می‌شوند، شاید با ترتیب دیگر، روی یک ماشین مشترک و شلوغ.
3. پس مشکل‌هایی که از اشتراک، ترتیب یا زمان‌بندی می‌آیند، فقط در CI دیده می‌شوند.

چطور علت را پیدا کنیم:

• تست را در یک حلقه بارها اجرا کن، هم روی لپ‌تاپ و هم در CI، تا شکست بخورد. هدف این است که عمداً شکستش بدهیم.

for i in $(seq 1 200); do pytest tests/test_orders.py -x -q || break; done

• ترتیب تست‌ها را تصادفی کن (مثلاً با پلاگینی مثل pytest-randomly) و مثل CI با چند worker موازی اجرا کن.
• خواندن شکست را آسان کن: داده تست، زمان، شناسه worker و خطای کامل را لاگ کن.
• تاریخچه شکست‌ها را نگاه کن. آیا با یک worker خاص یا در ساعت خاصی از روز شکست می‌خورد؟

چطور درستش کنیم:

• به هر تست داده خودش را بده: شناسه و ایمیل یکتا، یا یک تراکنش که بعد از هر تست rollback می‌شود، یا یک دیتابیس (schema) جدا برای هر worker.
• به جای sleep، منتظر یک شرط روشن بمان، همراه با timeout.
• ساعت را تزریق کن تا تست‌ها زمان را کنترل کنند.
• به جای صدا زدن سرویس بیرونی واقعی، از یک fake یا یک سرور stub استفاده کن.
• اگر علت یک race condition واقعی در کد است، کد را درست کن. تست flaky یک باگ واقعی پیدا کرده است.

سؤال پیگیری: درست کردنش یک هفته طول می‌کشد. انتشار فرداست. چه می‌کنی؟

جواب پیگیری:

• تست را قرنطینه می‌کنیم: از مجموعه تست‌های مسدودکننده بیرونش می‌بریم، پس هنوز اجرا می‌شود ولی جلوی merge را نمی‌گیرد.
• یک تیکت با مسئول مشخص و مهلت می‌سازیم، مثلاً دو هفته.
• مطمئن می‌شویم رفتاری که این تست بررسی می‌کرد هنوز به نوعی پوشش داده می‌شود، حتی به صورت دستی برای همین انتشار.
• لیست قرنطینه را دنبال می‌کنیم. اگر فقط بزرگ‌تر می‌شود، تیم فقط با چند قدم بیشتر تست‌ها را نادیده می‌گیرد.

نشانه خطر: می‌گوید «فقط flaky است، دوباره اجرا کن» یا برای همه تست‌ها retry خودکار می‌گذارد، بدون این که هیچ وقت دنبال علت بگردد.

ویرایش در GitHub

سؤال: تو مهندس on-call یک فروشگاه آنلاین هستی. ساعت ۲ بامداد است. گوشی‌ات با یک هشدار زنگ می‌زند: نرخ خطای API پرداخت (checkout) ۳۰ درصد است. در حالت عادی زیر ۱ درصد است. ۲۰ دقیقه پیش یک deploy از سرویس checkout انجام شده است. مشتری‌ها در منطقه‌های زمانی دیگر همین الان در حال خرید هستند. قدم به قدم بگو چه کار می‌کنی.

جواب کوتاه: اول جلوی آسیب را بگیر، بعد دنبال علت بگرد. محتمل‌ترین علت همان deploy بیست دقیقه پیش است. پس اولین حرکت معمولاً rollback است (یا خاموش کردن feature flag جدید)، حتی قبل از این که باگ را کامل بفهمیم. همزمان اطلاع‌رسانی کن: به بقیه بگو یک incident باز است. وقتی نرخ خطا به حالت عادی برگشت، با لاگ‌ها، متریک‌ها و trace ها علت اصلی را پیدا کن. بعداً یک post-mortem بدون سرزنش (blameless) با کارهای مشخص بنویس.

قدم ۱: تأیید و قبول مسئولیت (چند دقیقه اول)

1. هشدار را acknowledge کن تا بقیه بدانند کسی پیگیرش است.
2. سریع داشبورد را ببین: واقعاً ۳۰ درصد است؟ همه درخواست‌های checkout است، یا فقط یک منطقه، یک روش پرداخت، یا یک نسخه اپ؟
3. یک کانال incident باز کن (یا روند incident تیم را دنبال کن) و یک خط بنویس: «در حال بررسی خطای ۳۰ درصدی checkout، شروع بعد از deploy ساعت ۰۱:۴۰.»

قدم ۲: اول کاهش آسیب (چرا؟)

1. هر دقیقه، حدود ۳ نفر از هر ۱۰ مشتری نمی‌توانند پرداخت کنند. یعنی پول و اعتماد از دست می‌رود.
2. زمان شروع مشکل به شدت به deploy اشاره می‌کند. یک rollback معمولاً سریع، آشنا و امن است.
3. پیدا کردن باگ دقیق در ساعت ۲ بامداد ممکن است یک ساعت طول بکشد. ولی rollback چند دقیقه طول می‌کشد.
4. پس اول rollback می‌کنیم و بعداً با آرامش دیباگ می‌کنیم.

قبل از rollback یک چیز را بررسی کن: آیا این deploy یک migration دیتابیس داشت؟ اگر کد جدید schema را عوض کرده باشد، شاید کد قدیمی با آن کار نکند. در این صورت با احتیاط rollback کن، یا به جای آن از feature flag استفاده کن.

قدم ۳: بررسی نتیجه

• بعد از rollback نرخ خطا را نگاه کن. آیا زیر ۱ درصد برگشت؟
• اگر نه، شاید deploy علت نبوده است. تغییرهای دیگر را ببین: یک وابستگی (درگاه پرداخت، دیتابیس)، ترافیک، تنظیمات، گواهی‌ها یا زیرساخت.

قدم ۴: اطلاع‌رسانی

• در کانال incident به‌روزرسانی‌های کوتاه و منظم بفرست، مثلاً هر ۱۵ تا ۳۰ دقیقه، حتی اگر فقط «هنوز در حال بررسی» باشد.
• اگر آسیب بزرگ است، نفر دوم یا مسئول incident را بیدار کن. کمک خواستن عادی است، نه نشانه ضعف.
• اگر صفحه وضعیت (status page) یا تیم پشتیبانی داریم، به آن‌ها خبر بده تا بتوانند به مشتری‌ها جواب بدهند.

قدم ۵: پیدا کردن علت اصلی (بعد از کاهش آسیب)

• لاگ‌ها: بر اساس سرویس checkout و خطا فیلتر کن. پیام خطا چیست؟ کدام خط کد؟
• ردگیری (trace): یک درخواست شکست‌خورده را دنبال کن. کجا شکست می‌خورد: کد ما، دیتابیس، یا درگاه پرداخت؟
• متریک‌ها: قبل و بعد از deploy را مقایسه کن: latency، اتصال‌های دیتابیس، حافظه.
• تغییرات کد: تغییر کد در آن deploy را بخوان. اغلب علت همان‌جا دیده می‌شود.

قدم ۶: post-mortem بدون سرزنش

• یک خط زمانی بنویس: کی شروع شد، کی فهمیدیم، کی آسیب را کم کردیم.
• روی سیستم تمرکز کن، نه روی آدم‌ها: «چرا تست‌ها و canary ما این را نگرفتند؟» و نه «چه کسی خرابش کرد؟»
• وقتی آدم‌ها سرزنش می‌شوند، مشکل‌ها را پنهان می‌کنند. بازبینی بدون سرزنش باعث می‌شود داستان واقعی را بگویند.
• کارهای بعدی را با مسئول و تاریخ مشخص بساز. مثلاً: یک تست اضافه کنیم، یک مرحله canary اضافه کنیم، یک هشدار برای خطای checkout به تفکیک روش پرداخت بگذاریم.

سؤال پیگیری: rollback مشکل را حل کرد. آیا برای آن ۲۰ دقیقه پرداخت ناموفق هم باید کاری بکنیم؟

جواب پیگیری:

• بله. باید بررسی کنیم آیا پرداختی انجام شده ولی سفارش ذخیره نشده، یا برعکس. این داده باید درست شود.
• مشتری‌های آسیب‌دیده را پیدا می‌کنیم. شاید کسب‌وکار تصمیم بگیرد با آن‌ها تماس بگیرد یا چیزی جبران کند.
• بررسی می‌کنیم آیا بعضی درخواست‌ها دوباره فرستاده شده‌اند و تکراری ساخته‌اند.
• این بررسی‌ها را به post-mortem اضافه می‌کنیم تا incident بعدی یک لیست پاک‌سازی روشن داشته باشد.

نشانه خطر: یک ساعت کد را دیباگ می‌کند در حالی که مشتری‌ها همچنان شکست می‌خورند، تنها کار می‌کند و به کسی خبر نمی‌دهد، یا دنبال مقصر می‌گردد.

ویرایش در GitHub

سؤال: یک کلاس کش با ظرفیت ثابت N طراحی کن. دو عملیات دارد:

• عملیات get با یک کلید: اگر کلید در کش هست مقدارش را برگردان، وگرنه هیچ چیز برنگردان.
• عملیات put با یک کلید و یک مقدار: مقدار را اضافه یا به‌روز کن. اگر کش پر است، اول آیتمی را حذف کن که از همه دیرتر استفاده شده (least recently used).

هر دو عملیات باید در زمان O(1) اجرا شوند. «استفاده» یعنی get یا put روی آن کلید. ساختار داده‌هایت را توضیح بده، بعد کد را با پایتون بنویس.

جواب کوتاه: دو ساختار را با هم استفاده می‌کنیم:

1. یک hash map از کلید به گره (node). هر آیتم را در O(1) پیدا می‌کند.
2. یک لیست پیوندی دوطرفه (doubly linked list) از گره‌ها، به ترتیب آخرین استفاده. تازه‌ترین در جلو است و قدیمی‌ترین در عقب. جابه‌جا کردن یا حذف یک گره O(1) است، چون هر گره گره قبلی و بعدی‌اش را می‌شناسد.

در get، گره را در map پیدا می‌کنیم و به جلوی لیست می‌بریم. در put، گره را در جلو اضافه یا به‌روز می‌کنیم. اگر از ظرفیت بیشتر شدیم، گره ته لیست را حذف می‌کنیم و کلیدش را هم از map پاک می‌کنیم.

چرا چیزی ساده‌تر نه؟ (قدم به قدم)

1. یک hash map تنها، جستجوی O(1) دارد، ولی نمی‌داند کدام آیتم قدیمی‌ترین است. پیدا کردنش یعنی گشتن همه آیتم‌ها: O(N).
2. یک لیست یا آرایه تنها، ترتیب را نگه می‌دارد، ولی پیدا کردن کلید در آن O(N) است و حذف از وسط هم O(N).
3. یک لیست پیوندی یک‌طرفه نمی‌تواند گره را در O(1) حذف کند، چون به گره قبلی نیاز داریم.
4. لیست پیوندی دوطرفه همراه با map هر دو را می‌دهد: جستجوی سریع و جابه‌جایی سریع.

مدل داده:

• هر گره این‌ها را دارد: key، value، prev و next.
• گره کلید را هم نگه می‌دارد. وقتی گره آخر را حذف می‌کنیم، کلیدش را لازم داریم تا از map هم پاکش کنیم.
• دو گره ساختگی (dummy) به نام head و tail در دو سر لیست هستند. این طوری هیچ وقت حالت خاص برای لیست خالی لازم نداریم.

پیاده‌سازی:

class Node:
    __slots__ = ("key", "value", "prev", "next")

    def __init__(self, key=None, value=None):
        self.key, self.value = key, value
        self.prev = self.next = None

class LRUCache:
    def __init__(self, capacity: int):
        if capacity <= 0:
            raise ValueError("capacity must be positive")
        self.capacity = capacity
        self.map: dict = {}
        self.head, self.tail = Node(), Node()  # dummy nodes
        self.head.next, self.tail.prev = self.tail, self.head

    def _remove(self, node: Node) -> None:
        node.prev.next, node.next.prev = node.next, node.prev

    def _add_front(self, node: Node) -> None:
        node.prev, node.next = self.head, self.head.next
        self.head.next.prev = node
        self.head.next = node

    def get(self, key):
        node = self.map.get(key)
        if node is None:
            return None
        self._remove(node)
        self._add_front(node)  # mark as most recently used
        return node.value

    def put(self, key, value) -> None:
        node = self.map.get(key)
        if node is not None:
            node.value = value
            self._remove(node)
        else:
            if len(self.map) == self.capacity:
                lru = self.tail.prev  # least recently used
                self._remove(lru)
                del self.map[lru.key]
            node = Node(key, value)
            self.map[key] = node
        self._add_front(node)

راه کوتاه در پایتون: کلاس OrderedDict در ماژول collections کلیدها را به ترتیب اضافه شدن نگه می‌دارد. در داخل هم یک hash map و یک لیست پیوندی دوطرفه دارد، یعنی همان ایده بالا. متد move_to_end یک کلید را تازه‌ترین می‌کند، و متد popitem با پارامتر last برابر False قدیمی‌ترین کلید را حذف می‌کند. هر دو O(1) هستند. در پروژه واقعی، دکوریتور lru_cache در functools همین کار را برای نتیجه تابع‌ها انجام می‌دهد. در مصاحبه، حتی اگر از OrderedDict استفاده می‌کنی، ساختار داخلی را توضیح بده.

حالت‌های مرزی برای تست: ظرفیت ۱، گرفتن کلیدی که وجود ندارد، put روی کلیدی که از قبل هست (باید به‌روز شود و به جلو برود، نه این که اندازه بزرگ شود)، و یک get که آیتمی را از حذف شدن نجات می‌دهد.

سؤال پیگیری: چندین thread همزمان از این کش استفاده می‌کنند. امن است؟ چطور درستش می‌کنی؟

جواب پیگیری:

• امن نیست. عملیات put چند اشاره‌گر و خود map را در چند قدم تغییر می‌دهد. اگر دو thread همزمان این کار را بکنند، لیست ممکن است خراب شود. مثلاً گره‌ای در map باشد ولی در لیست نباشد.
• دقت کن که حتی get هم لیست را تغییر می‌دهد، چون گره را به جلو می‌برد. پس خواندن هم محافظت لازم دارد و قفل خواندن‌نوشتن (read-write lock) اینجا کمک زیادی نمی‌کند.
• راه ساده: یک قفل دور بدنه get و put. هر صدا زدن کوتاه است، پس این روش اغلب به اندازه کافی سریع است.
• اگر قفل گلوگاه شد، کش را بر اساس hash کلید به چند shard تقسیم کن. هر shard قفل خودش و ظرفیت خودش را دارد. هزینه‌اش این است که حذف دیگر «LRU در هر shard» است، نه LRU دقیق سراسری.

نشانه خطر: برای حذف از یک لیست و گشتن همه آیتم‌ها استفاده می‌کند و اسمش را O(1) می‌گذارد، یا فراموش می‌کند در get آیتم را به جلو ببرد.

ویرایش در GitHub

سؤال: سرویسی طراحی کن که کاربرها در آن فایل آپلود می‌کنند: ویدیو و سند، هر کدام تا ۵ گیگابایت. روزانه حدود ۱۰۰٬۰۰۰ آپلود داریم. خیلی از کاربرها با موبایل و روی شبکه ناپایدار هستند، پس اتصال زیاد قطع می‌شود. بعد از آپلود، فایل‌ها باید از نظر ویروس بررسی شوند و ویدیوها یک تصویر پیش‌نمایش لازم دارند. کاربرها بعداً می‌توانند فایل‌های خودشان را دانلود کنند. طراحی‌ات را قدم به قدم توضیح بده.

جواب کوتاه: بایت‌های فایل را از سرورهای API خودمان رد نکن. کلاینت از API اجازه می‌گیرد و چند pre-signed URL (آدرس امضاشده موقت) دریافت می‌کند. بعد فایل را مستقیم به object storage (مثل Amazon S3) و تکه‌تکه آپلود می‌کند (multipart upload). اگر شبکه قطع شود، فقط تکه‌ای که شکست خورده دوباره فرستاده می‌شود. یک دیتابیس metadata رکورد فایل و وضعیتش را نگه می‌دارد. وقتی آپلود تمام شد، یک پیام به یک صف می‌رود و worker های پس‌زمینه بررسی ویروس و پردازش را انجام می‌دهند.

نیازمندی‌ها و عددها:

1. صد هزار آپلود در روز یعنی به طور میانگین حدود ۱٫۲ آپلود در ثانیه. اوج ترافیک ممکن است چند برابر باشد.
2. فرض می‌کنیم (باید بپرسیم) میانگین فایل ۵۰ مگابایت است. یعنی حدود ۵ ترابایت داده جدید در روز و نزدیک ۲ پتابایت در سال. پس هزینه ذخیره‌سازی و قانون نگه‌داری داده مهم است.
3. پنج ترابایت در روز یعنی به طور میانگین حدود ۵۸ مگابایت در ثانیه، در تمام شبانه‌روز. اگر این از سرورهای API ما رد شود، وقتشان صرف کپی کردن بایت‌ها می‌شود.
4. آپلود یک فایل ۵ گیگابایتی روی شبکه ضعیف موبایل ممکن است خیلی طول بکشد. باید از قطع‌های زیاد جان سالم به در ببرد.

بخش‌های اصلی:

1. سرویس API آپلود: کاربر را بررسی می‌کند، محدودیت‌ها را چک می‌کند (اندازه، نوع، سهمیه)، یک رکورد فایل می‌سازد و pre-signed URL ها را برمی‌گرداند.
2. ذخیره‌ساز object storage: بایت‌ها را نگه می‌دارد. هزینه هر گیگابایت خیلی کم است، داده خیلی امن می‌ماند و بدون کار ما scale می‌شود.
3. دیتابیس metadata: برای هر فایل یک ردیف.
4. صف و worker ها: بررسی ویروس، تصویر پیش‌نمایش، و شاید تبدیل فرمت ویدیو.
5. دانلود: API دسترسی را بررسی می‌کند و یک pre-signed URL کوتاه‌مدت برای دانلود، یا یک آدرس CDN، برمی‌گرداند.

روند آپلود (قدم به قدم):

1. کلاینت نام فایل، اندازه و نوع را می‌فرستد. API یک رکورد با وضعیت «uploading» می‌سازد و یک multipart upload را شروع می‌کند.
2. سرور API برای تکه‌ها pre-signed URL برمی‌گرداند (مثلاً تکه‌های ۱۰ مگابایتی). هر URL فقط برای مدت کوتاهی و فقط برای همان یک تکه کار می‌کند.
3. کلاینت تکه‌ها را آپلود می‌کند، چند تا را موازی. یادداشت می‌کند کدام تکه‌ها تمام شده‌اند. بعد از قطع شدن، از API می‌پرسد کدام تکه‌ها مانده‌اند و فقط همان‌ها را می‌فرستد.
4. وقتی همه تکه‌ها تمام شد، کلاینت «complete» را صدا می‌زند. API به storage می‌گوید تکه‌ها را به هم بچسباند، وضعیت را «scanning» می‌کند و یک پیام در صف می‌گذارد.
5. یک worker فایل را بررسی می‌کند. اگر سالم بود، وضعیت «ready» می‌شود. اگر نه، فایل پاک یا قرنطینه می‌شود و وضعیت «rejected» می‌شود.
6. کاربرها فقط فایل‌هایی را می‌توانند دانلود کنند که وضعیتشان «ready» است.

مدل داده (یک جدول):

CREATE TABLE files (
    id           UUID PRIMARY KEY,
    owner_id     BIGINT NOT NULL,
    storage_key  TEXT NOT NULL,      -- random key, not the user's file name
    file_name    TEXT NOT NULL,
    size_bytes   BIGINT NOT NULL,
    content_type TEXT NOT NULL,
    status       TEXT NOT NULL,      -- uploading, scanning, ready, rejected
    created_at   TIMESTAMPTZ NOT NULL
);

بخش‌های سخت و بده‌بستان‌ها:

• چرا multipart اینجا ضروری است: مثلاً در S3، یک درخواست آپلود تنها حداکثر ۵ گیگابایت می‌تواند باشد و یک شکست یعنی شروع دوباره از صفر. با multipart upload تا ۱۰٬۰۰۰ تکه داریم و هر تکه جداگانه دوباره تلاش می‌شود. یک پروتکل باز مثل tus هم گزینه دیگری برای آپلود قابل ادامه است.
• آپلودهای رهاشده: خیلی از آپلودها هیچ وقت تمام نمی‌شوند. ولی تکه‌هایشان هنوز هزینه دارد. با یک قانون lifecycle در storage، آپلودهای ناتمام را بعد از چند روز لغو کن. یک کار پاک‌سازی هم برای رکوردهای «uploading» خیلی قدیمی بگذار.
• امنیت: هیچ وقت به نوع فایلی که کلاینت می‌گوید اعتماد نکن. محتوای واقعی را بررسی کن (magic bytes). کلیدهای ذخیره‌سازی تصادفی بساز تا کسی نتواند URL را حدس بزند. bucket را خصوصی نگه دار. pre-signed URL ها را کوتاه‌مدت نگه دار. هیچ فایلی را قبل از تأیید بررسی ویروس به کسی نده.
• محدودیت‌ها: حداکثر اندازه فایل، سهمیه هر کاربر، و rate limit روی شروع آپلود. اندازه را هم موقع شروع و هم موقع پایان آپلود بررسی کن، چون کلاینت ممکن است دروغ بگوید.
• پردازش async: بررسی یک فایل ۵ گیگابایتی کند است. با صف، worker ها مستقل scale می‌شوند و اگر بررسی شکست خورد، بدون آپلود دوباره کاربر تکرار می‌شود.

چطور scale می‌شود: سرور API فقط درخواست‌های کوچک JSON را جواب می‌دهد، پس چند نمونه کافی است. ذخیره‌ساز object storage خودش scale می‌شود. تعداد worker ها با طول صف بالا و پایین می‌رود. جدول metadata سالی حدود ۳۶ میلیون ردیف بزرگ می‌شود که یک دیتابیس با index درست (روی owner_id و created_at) راحت از پسش برمی‌آید.

سؤال پیگیری: آپلود کاربرهایی که خیلی دور هستند، مثلاً در یک قاره دیگر، خیلی کند است. چه کار می‌توانی بکنی؟

جواب پیگیری:

• از storage در بیش از یک منطقه (region) استفاده کن و به هر کاربر pre-signed URL نزدیک‌ترین منطقه را بده.
• بعضی سرویس‌دهنده‌ها قابلیت آپلود شتاب‌داده‌شده دارند که ترافیک را از نقطه‌های لبه (edge) رد می‌کند. هزینه اضافه دارد، پس اول اندازه بگیر.
• اندازه تکه‌ها را تنظیم کن: تکه کوچک‌تر روی شبکه ناپایدار بهتر است، تکه بزرگ‌تر یعنی درخواست کمتر.
• برای دانلود، CDN خیلی کمک می‌کند، ولی فقط برای فایل‌هایی که زیاد دانلود می‌شوند.

نشانه خطر: فایل‌های ۵ گیگابایتی را در یک درخواست HTTP از سرورهای API رد می‌کند، یا فایل‌ها را عمومی می‌کند و قبل از بررسی ویروس در اختیار دیگران می‌گذارد.

ویرایش در GitHub