هیچ پیام منتشرشده‌ای از دست نمی‌رود
Offset فقط پس از Commit موفق PostgreSQL جابه‌جا می‌شود. پیامی که خوانده شده اما نوشته نشده، هرگز به‌عنوان «پردازش‌شده» علامت نمی‌خورد.
مکانیزم:enable_auto_commit=False + ترتیب نوشتن پیش از Commit
این مستند معماری، فلوی اجرا، تضمین‌های داده‌ای و تصمیمات مهندسی زیرسیستم Storage Consumer را شرح می‌دهد. این زیرسیستم سه تاپیک app-stats، reviews و network-metrics را از Kafka مصرف می‌کند و آن‌ها را در PostgreSQL می‌نویسد.
نقش آن در پروژه یک نقش ویژه است و همین یک جمله، تفاوت سطح سخت‌گیری آن با بقیه‌ی زیرسیستم‌ها را توضیح می‌دهد: اگر Crawler وسط کار Crash کند، همان اپ در دور بعدی دوباره خزش می‌شود و چیزی از دست نمی‌رود. اگر Network Analyzer شکست بخورد، همان فایل pcap دوباره تحلیل می‌شود. اما اگر Storage Consumer یک پیام را از دست بدهد، آن داده برای همیشه رفته است. تمام پیچیدگی‌های این مستند — Commit دستی Offset، کلیدهای Idempotency، تراکنش واحد، و صف پیام‌های مرده — پاسخ مستقیم همین یک نیازمندی هستند.
۱ چهار تضمین اصلی زیرسیستم
پیش از هر جزئیات فنی، کل طراحی این زیرسیستم را می‌توان در چهار تضمین خلاصه کرد. هر تصمیم مهندسی در ادامه‌ی این مستند، در خدمت یکی از این چهار مورد است.
تحویل مجدد، داده را تکراری نمی‌کند
هر جدول یک قید UNIQUE روی کلید طبیعی پیام دارد و هر نوشتن یک INSERT ... ON CONFLICT است. کافکا At-least-once می‌دهد؛ قیدها آن را به Effectively-once تبدیل می‌کنند.
یک پیام خراب، کل تاپیک را متوقف نمی‌کند
پیام غیرقابل‌ذخیره در جدول dead_letter_events ثبت می‌شود و Offset از رویش عبور می‌کند — پس قفل‌شدن ابدی یک Partition روی یک پیام معیوب (Poison Pill) غیرممکن است.
یک Pipeline نیمه‌جان، یعنی فرآیند مرده
شکست هر Thread، تمام Threadهای دیگر را متوقف و فرآیند را با کد خروج غیرصفر خارج می‌کند. کانتینری با Uptime سالم و یک تاپیک بی‌صدا، از Crash بدتر است چون دیده نمی‌شود.
مکانیزم: Supervisor با قاعده‌ی همه‌یا‌هیچ۲ جایگاه زیرسیستم در معماری کل پروژه
این زیرسیستم انتهای زنجیره‌ی جمع‌آوری داده و ابتدای زنجیره‌ی تحلیل است. دو تولیدکننده‌ی مستقل (Crawler و Network Analyzer) روی سه تاپیک پیام می‌نویسند، و این زیرسیستم هر سه را به چهار جدول تحت مالکیت خودش تبدیل می‌کند.
flowchart LR
CR["Crawler
(هر ۱ ساعت)"] --> T1["Topic: app-stats
۳ Partition"] CR --> T2["Topic: reviews
۳ Partition"] NA["Network Analyzer
(بر اساس درخواست)"] --> T3["Topic: network-metrics
۳ Partition"] T1 --> SC T2 --> SC T3 --> SC subgraph SC["Storage Consumer - یک فرآیند، سه Thread"] direction TB P1["Pipeline: app-stats
Group: storage-consumer.app-stats"] P2["Pipeline: reviews
Group: storage-consumer.reviews"] P3["Pipeline: network-metrics
Group: storage-consumer.network-metrics"] end P1 --> DB1["جدول app_stats"] P2 --> DB2["جدول reviews"] P3 --> DB3["جدول network_metrics"] P1 --> DLQ["جدول dead_letter_events"] P2 --> DLQ P3 --> DLQ APP["جدول applications
(مالکیت: App API / Django)"] -.->|"فقط خواندن
package_name ← application_id"| SC style CR fill:#2563eb,color:#fff style NA fill:#7c3aed,color:#fff style SC fill:#0f172a,color:#fff style DLQ fill:#dc2626,color:#fff style APP fill:#f59e0b,color:#fff
(هر ۱ ساعت)"] --> T1["Topic: app-stats
۳ Partition"] CR --> T2["Topic: reviews
۳ Partition"] NA["Network Analyzer
(بر اساس درخواست)"] --> T3["Topic: network-metrics
۳ Partition"] T1 --> SC T2 --> SC T3 --> SC subgraph SC["Storage Consumer - یک فرآیند، سه Thread"] direction TB P1["Pipeline: app-stats
Group: storage-consumer.app-stats"] P2["Pipeline: reviews
Group: storage-consumer.reviews"] P3["Pipeline: network-metrics
Group: storage-consumer.network-metrics"] end P1 --> DB1["جدول app_stats"] P2 --> DB2["جدول reviews"] P3 --> DB3["جدول network_metrics"] P1 --> DLQ["جدول dead_letter_events"] P2 --> DLQ P3 --> DLQ APP["جدول applications
(مالکیت: App API / Django)"] -.->|"فقط خواندن
package_name ← application_id"| SC style CR fill:#2563eb,color:#fff style NA fill:#7c3aed,color:#fff style SC fill:#0f172a,color:#fff style DLQ fill:#dc2626,color:#fff style APP fill:#f59e0b,color:#fff
۳ تصمیم معماری کلیدی: چرا سه Thread و نه یک Consumer واحد؟
یک Consumer واحد می‌توانست به هر سه تاپیک Subscribe کند و کد ساده‌تری داشته باشد. این گزینه عمداً رد شد، و دلیل آن هیچ ارتباطی به سرعت ندارد — به «انصاف در تأخیر» مربوط است.
حجم بار این سه تاپیک به‌شدت نامتوازن است. تاپیک reviews یک مسئله‌ی توان عملیاتی است: در هر دور خزش تا ۲۰۰٫۰۰۰ پیام در بازه‌ی ۵ تا ۱۵ دقیقه منتشر می‌شود. در مقابل، network-metrics یک مسئله‌ی تأخیر است: روزی چند پیام، اما درست در لحظه‌ای که کاربر یک تست شبکه اجرا کرده و منتظر دیدن نتیجه است. با یک Consumer واحد، آن چند پیام حساس به تأخیر پشت صف ۲۰۰٫۰۰۰ پیامی نقدها می‌ماندند و دقیقه‌ها معطل می‌شدند.
flowchart TD
Q{"سه تاپیک با بار نامتوازن
چگونه مصرف شوند؟"} Q -->|"گزینه ۱"| ONE["یک Consumer واحد
Subscribe به هر سه تاپیک"] Q -->|"گزینه ۲ (انتخاب‌شده)"| MULTI["سه Thread، سه Consumer Group"] ONE --> BAD1["صف مشترک: ۲۰۰٫۰۰۰ نقد
جلوی یک پیام network-metrics"] ONE --> BAD2["Backoff پایگاه‌داده برای یک تاپیک
دو تاپیک دیگر را هم متوقف می‌کند"] ONE --> BAD3["Reset کردن Offset یک تاپیک
بدون تأثیر بر بقیه ممکن نیست"] MULTI --> GOOD1["تأخیر هر تاپیک مستقل
(اندازه‌گیری‌شده: زیر ۱ ثانیه هنگام ingest ۳٫۰۰۰ نقد)"] MULTI --> GOOD2["Share-Nothing: بدون Lock"] MULTI --> GOOD3["Lag و Offset هر تاپیک جدا و قابل مدیریت"] style MULTI fill:#16a34a,color:#fff style ONE fill:#dc2626,color:#fff style Q fill:#2563eb,color:#fff
چگونه مصرف شوند؟"} Q -->|"گزینه ۱"| ONE["یک Consumer واحد
Subscribe به هر سه تاپیک"] Q -->|"گزینه ۲ (انتخاب‌شده)"| MULTI["سه Thread، سه Consumer Group"] ONE --> BAD1["صف مشترک: ۲۰۰٫۰۰۰ نقد
جلوی یک پیام network-metrics"] ONE --> BAD2["Backoff پایگاه‌داده برای یک تاپیک
دو تاپیک دیگر را هم متوقف می‌کند"] ONE --> BAD3["Reset کردن Offset یک تاپیک
بدون تأثیر بر بقیه ممکن نیست"] MULTI --> GOOD1["تأخیر هر تاپیک مستقل
(اندازه‌گیری‌شده: زیر ۱ ثانیه هنگام ingest ۳٫۰۰۰ نقد)"] MULTI --> GOOD2["Share-Nothing: بدون Lock"] MULTI --> GOOD3["Lag و Offset هر تاپیک جدا و قابل مدیریت"] style MULTI fill:#16a34a,color:#fff style ONE fill:#dc2626,color:#fff style Q fill:#2563eb,color:#fff
نتیجه‌ی جانبی مهم: Thread در این طراحی ابزار جداسازی (Isolation) است، نه ابزار افزایش سرعت. هر Pipeline حلقه‌ی Poll، Consumer Group، اتصال پایگاه‌داده و کش مستقل خودش را دارد. چون هیچ‌چیز بین Threadها مشترک نیست (به‌جز تنظیمات Immutable و یک رخداد توقف)، در کل این زیرسیستم حتی یک Lock هم وجود ندارد.
- اندازه‌ی بچ
- ۲۰۰ رکورد
- ماهیت مسئله
- حجم متوسط و منظم؛ یک پیام به‌ازای هر اپ در هر دور
- اندازه‌ی بچ
- ۵۰۰ رکورد
- ماهیت مسئله
- توان عملیاتی؛ انبوه پیام در بازه‌ی کوتاه، بچ بزرگ‌تر بهینه‌تر است
- اندازه‌ی بچ
- ۵۰ رکورد
- ماهیت مسئله
- تأخیر؛ بهتر است هرچه دارد را بنویسد تا منتظر پر شدن بچ بماند
۴ فلوی کامل پردازش یک بچ
قلب این زیرسیستم یک حلقه‌ی کوتاه است: poll ← پردازش ← commit. آنچه این حلقه را کوتاه نگه می‌دارد یک قاعده است: هیچ حالتی (State) از یک Poll به Poll بعدی منتقل نمی‌شود. نتیجه‌ی این قاعده آن است که Rebalance شدن Consumer Group در میانه‌ی یک بچ به‌طور خودکار بی‌خطر است — بچ در جریان، نانوشته رها می‌شود و به هر عضوی که Partition را تحویل بگیرد دوباره تحویل داده می‌شود.
sequenceDiagram
participant K as Kafka
participant W as Worker
participant D as decoders
participant R as ApplicationResolver
participant B as batching
participant DLQ as DeadLetterRepository
participant Repo as Repository
participant PG as PostgreSQL
W->>K: poll(timeout=1000ms, max_records=N)
K-->>W: بچ رکوردها
Note over W: شروع تراکنش واحد برای کل بچ
W->>D: decode_batch(records)
D-->>W: (پیام‌های سالم، پیام‌های ردشده)
Note over D: JSON نامعتبر، فیلد الزامی غایب،
نوع اشتباه، مقدار خارج از بازه‌ی ستون W->>R: resolve(cursor, مجموعه‌ی package_nameها) R->>PG: SELECT ... WHERE package_name = ANY(%s) Note over R: یک کوئری برای کل بچ، نه یک کوئری به‌ازای هر پیام
+ کش TTL اختصاصی همین Thread PG-->>R: نگاشت package_name ← application_id R-->>W: application_idها W->>B: bind_application_ids(...) B-->>W: (رکوردهای متصل‌شده، اپ‌های ثبت‌نشده) W->>DLQ: record(cursor, تمام پیام‌های ردشده) Note over DLQ,PG: در همان تراکنش بچ W->>B: dedupe_by_key(rows) Note over B: الزامی است، نه بهینه‌سازی:
Postgres اجازه نمی‌دهد یک INSERT دو بار
همان ردیف را در ON CONFLICT DO UPDATE لمس کند W->>Repo: upsert(cursor, rows) Repo->>PG: INSERT ... VALUES %s ON CONFLICT ... RETURNING (xmax = 0) PG-->>Repo: ردیف‌های بازگشتی Repo-->>W: WriteOutcome(inserted, updated, skipped) Note over W,PG: COMMIT تراکنش — داده اکنون ماندگار است W->>K: commit() — و فقط اکنون Offset جابه‌جا می‌شود W->>W: یک خط لاگ INFO با تمام شمارنده‌ها
نوع اشتباه، مقدار خارج از بازه‌ی ستون W->>R: resolve(cursor, مجموعه‌ی package_nameها) R->>PG: SELECT ... WHERE package_name = ANY(%s) Note over R: یک کوئری برای کل بچ، نه یک کوئری به‌ازای هر پیام
+ کش TTL اختصاصی همین Thread PG-->>R: نگاشت package_name ← application_id R-->>W: application_idها W->>B: bind_application_ids(...) B-->>W: (رکوردهای متصل‌شده، اپ‌های ثبت‌نشده) W->>DLQ: record(cursor, تمام پیام‌های ردشده) Note over DLQ,PG: در همان تراکنش بچ W->>B: dedupe_by_key(rows) Note over B: الزامی است، نه بهینه‌سازی:
Postgres اجازه نمی‌دهد یک INSERT دو بار
همان ردیف را در ON CONFLICT DO UPDATE لمس کند W->>Repo: upsert(cursor, rows) Repo->>PG: INSERT ... VALUES %s ON CONFLICT ... RETURNING (xmax = 0) PG-->>Repo: ردیف‌های بازگشتی Repo-->>W: WriteOutcome(inserted, updated, skipped) Note over W,PG: COMMIT تراکنش — داده اکنون ماندگار است W->>K: commit() — و فقط اکنون Offset جابه‌جا می‌شود W->>W: یک خط لاگ INFO با تمام شمارنده‌ها
۵ ترتیب حیاتی: نوشتن پیش از Commit
مهم‌ترین خط این زیرسیستم، ترتیب دو عملیات است. enable_auto_commit عمداً خاموش است، چون Commit خودکار بر پایه‌ی زمان کار می‌کند نه بر پایه‌ی موفقیت نوشتن — یعنی می‌تواند Offset پیامی را جابه‌جا کند که هنوز در پایگاه‌داده ننشسته است. با Commit دستی و پس از نوشتن، تنها حالت خرابی ممکن، تحویل مجدد است، و تحویل مجدد به‌لطف قیدهای UNIQUE بی‌خطر است.
POLLخواندن بچ از Kafka
WRITEنوشتن داده و DLQ در یک تراکنش
COMMIT (DB)داده ماندگار شد
COMMIT (KAFKA)و فقط اکنون Offset جابه‌جا می‌شود
flowchart TD
S["بچ خوانده شد"] --> WR["نوشتن در PostgreSQL"]
WR -->|"موفق"| CM["Commit کردن Offset در Kafka"]
WR -->|"Crash پیش از نوشتن"| C1["Offset جابه‌جا نشده
← بچ دوباره تحویل داده می‌شود
← هیچ داده‌ای گم نشده"] CM -->|"موفق"| OK["چرخه‌ی بعدی"] CM -->|"Crash پس از نوشتن، پیش از Commit"| C2["داده نوشته شده ولی Offset قدیمی است
← بچ دوباره تحویل داده می‌شود
← ON CONFLICT آن را بی‌اثر می‌کند"] CM -->|"شکست Commit (Rebalance)"| C3["داده ماندگار است
← لاگ هشدار، ادامه‌ی کار
← پس از ۵ شکست پیاپی: خروج عامدانه"] C1 --> SAFE["نتیجه: At-least-once"] C2 --> SAFE C3 --> SAFE SAFE --> EO["به‌همراه Idempotency
= Effectively-once"] style WR fill:#2563eb,color:#fff style CM fill:#f59e0b,color:#fff style SAFE fill:#16a34a,color:#fff style EO fill:#0f172a,color:#fff
← بچ دوباره تحویل داده می‌شود
← هیچ داده‌ای گم نشده"] CM -->|"موفق"| OK["چرخه‌ی بعدی"] CM -->|"Crash پس از نوشتن، پیش از Commit"| C2["داده نوشته شده ولی Offset قدیمی است
← بچ دوباره تحویل داده می‌شود
← ON CONFLICT آن را بی‌اثر می‌کند"] CM -->|"شکست Commit (Rebalance)"| C3["داده ماندگار است
← لاگ هشدار، ادامه‌ی کار
← پس از ۵ شکست پیاپی: خروج عامدانه"] C1 --> SAFE["نتیجه: At-least-once"] C2 --> SAFE C3 --> SAFE SAFE --> EO["به‌همراه Idempotency
= Effectively-once"] style WR fill:#2563eb,color:#fff style CM fill:#f59e0b,color:#fff style SAFE fill:#16a34a,color:#fff style EO fill:#0f172a,color:#fff
نکته درباره‌ی شکست Commit: شکست Commit به‌خودی‌خود بی‌خطر است (داده از قبل ماندگار شده و تحویل مجدد بی‌اثر است)، پس لاگ می‌شود و کار ادامه پیدا می‌کند. اما اگر Commit هرگز موفق نشود، Pipeline تا ابد همان رکوردها را بازپردازش می‌کند در حالی که Lag هیچ‌وقت کاهش نمی‌یابد. این حالت
Livelock بسیار سخت‌تر از یک Crash تشخیص داده می‌شود، پس عمداً به Crash تبدیل شده است: پس از ۵ شکست پیاپی، فرآیند با خطا خارج می‌شود تا کانتینر Restart شود.۶ Idempotency: کلید طبیعی هر جدول
کافکا تحویل At-least-once می‌دهد؛ تبدیل آن به Effectively-once وظیفه‌ی این زیرسیستم است. مکانیزم، یک قید UNIQUE روی «کلید طبیعی» هر پیام است، به‌همراه یک INSERT ... ON CONFLICT متناسب با معنای همان جدول. انتخاب بین DO NOTHING و DO UPDATE در هر جدول یک تصمیم معنایی است، نه سلیقه‌ای.
| جدول | کلید Idempotency | رفتار در تعارض | استدلال |
|---|---|---|---|
app_stats | (application_id, crawled_at) | DO NOTHING | هر پیام با همان جفت کلید، طبق تعریف یک تحویل مجدد است — Crawler برای هر اپ در هر دور یک پیام با crawled_at دقیق (میکروثانیه) می‌فرستد، پس چیز تازه‌تری در پیام دوم نیست. خزش بعدی crawled_at جدید دارد و ردیف جدید می‌سازد؛ همین است که جدول را Append-Only نگه می‌دارد. |
reviews | (review_id) | DO UPDATE با نگهبان یکنوا | تنها جدولی که واقعاً به‌روزرسانی می‌شود، چون thumbs_up_count یک نقد در طول زمان تغییر می‌کند. اما به‌روزرسانی بی‌قید و شرط خطرناک است (بخش بعد). |
network_metrics | (analysis_id) | DO NOTHING | یک فایل pcap یک‌بار تحلیل می‌شود؛ همان analysis_id برای بار دوم یعنی تحویل مجدد، نه اصلاح. اجرای واقعی مجدد یک تست، UUID تازه و در نتیجه ردیف تازه تولید می‌کند. |
dead_letter_events | (topic, partition, kafka_offset) | DO NOTHING | مختصات یک رکورد در کافکا، هویت آن است. Replay یک Partition همان پیام‌های خراب را دوباره رد می‌کند و این نباید ردیف‌ها را چند برابر کند. |
۷ نگهبان یکنوا روی جدول reviews
جدول reviews تنها جدولی است که به‌روزرسانی می‌شود، و همین آن را در معرض دو خرابی واقعی قرار می‌دهد که هر دو با یک شرط WHERE حل شده‌اند.
ON CONFLICT (review_id) DO UPDATE SET
thumbs_up_count = EXCLUDED.thumbs_up_count,
score = EXCLUDED.score,
content = EXCLUDED.content,
last_synced_at = EXCLUDED.last_synced_at
WHERE EXCLUDED.last_synced_at > reviews.last_synced_at
RETURNING (xmax = 0) AS inserted
-- ستون‌های first_seen_at و sentiment عمداً در فهرست بالا نیستند
flowchart TD
IN["پیام نقد رسید
(کلید: review_id)"] --> EX{"این review_id
از قبل وجود دارد؟"} EX -->|"خیر"| INS["INSERT ردیف جدید
first_seen_at = now()"] EX -->|"بله"| GUARD{"last_synced_at پیام
تازه‌تر از ردیف موجود است؟"} GUARD -->|"بله — پیام تازه‌تر"| UPD["UPDATE مقادیر
(thumbs_up_count و ...)"] GUARD -->|"خیر — پیام کهنه"| SKIP["No-op
در شمارنده‌ی skipped ثبت می‌شود"] SKIP --> R1["سناریوی ۱ - Replay:
یک پیام قدیمی، ردیف تازه را بازنویسی نمی‌کند"] SKIP --> R2["سناریوی ۲ - ترتیب برهم‌خورده:
کافکا ترتیب را فقط داخل هر Partition تضمین می‌کند،
پس Retry کند یک خزش قدیمی می‌تواند بعد از خزش جدید برسد"] style INS fill:#16a34a,color:#fff style UPD fill:#2563eb,color:#fff style SKIP fill:#f59e0b,color:#fff style GUARD fill:#7c3aed,color:#fff
(کلید: review_id)"] --> EX{"این review_id
از قبل وجود دارد؟"} EX -->|"خیر"| INS["INSERT ردیف جدید
first_seen_at = now()"] EX -->|"بله"| GUARD{"last_synced_at پیام
تازه‌تر از ردیف موجود است؟"} GUARD -->|"بله — پیام تازه‌تر"| UPD["UPDATE مقادیر
(thumbs_up_count و ...)"] GUARD -->|"خیر — پیام کهنه"| SKIP["No-op
در شمارنده‌ی skipped ثبت می‌شود"] SKIP --> R1["سناریوی ۱ - Replay:
یک پیام قدیمی، ردیف تازه را بازنویسی نمی‌کند"] SKIP --> R2["سناریوی ۲ - ترتیب برهم‌خورده:
کافکا ترتیب را فقط داخل هر Partition تضمین می‌کند،
پس Retry کند یک خزش قدیمی می‌تواند بعد از خزش جدید برسد"] style INS fill:#16a34a,color:#fff style UPD fill:#2563eb,color:#fff style SKIP fill:#f59e0b,color:#fff style GUARD fill:#7c3aed,color:#fff
first_seen_at هرگز به‌روزرسانی نمی‌شودمقدار DEFAULT now() خودش را از لحظه‌ی Insert نگه می‌دارد، تا معنایش «اولین بار که این نقد را دیدیم» بماند، مهم نیست چند بار بعد از آن همگام‌سازی مجدد شود.sentiment هرگز لمس نمی‌شوداین ستون به زیرسیستم آینده‌ی sentiment/ تعلق دارد. دست زدن به آن در اینجا یعنی هر بار که Crawler یک نقد را دوباره می‌خواند، نتیجه‌ی تحلیل احساسات محاسبه‌شده پاک شود.۸ تفکیک سه‌گانه‌ی خطاها
اشتباه گرفتن این سه دسته با هم، دقیقاً همان جایی است که یک Consumer کافکا یا داده از دست می‌دهد یا تا ابد قفل می‌شود. به همین دلیل این تفکیک در کد صریح و آزمون‌شده است.
موقتی — Retry شود
OperationalError و InterfaceError. همان بچ با Backoff نمایی تکرار می‌شود (۱s → ۲s → ۴s → ۸s → ۱۶s) و هرگز Dead-letter نمی‌شود.
دائمی — Dead-letter شود
MessageDecodeError یا package_name ثبت‌نشده. در جدول ثبت می‌شود، Offset عبور می‌کند و هرگز Retry نمی‌شود.
مرگبار — فرآیند خارج شود
ConfigError، MigrationError و هر خطای پیش‌بینی‌نشده. خروج با کد غیرصفر و Restart کانتینر از آخرین Offset ثبت‌شده.
flowchart TD
E["خطایی رخ داد"] --> T{"دسته‌بندی خطا"}
T -->|"موقتی (Transient)"| TR["OperationalError, InterfaceError
(قطع اتصال، Deadlock، Restart پایگاه‌داده)"] T -->|"دائمی، مخصوص یک پیام"| PM["MessageDecodeError
یا package_name ثبت‌نشده"] T -->|"مرگبار (Fatal)"| FT["ConfigError, MigrationError
و هر خطای پیش‌بینی‌نشده"] TR --> TRA["Retry همان بچ با Backoff نمایی
۱s → ۲s → ۴s → ۸s → ۱۶s (۵ تلاش)
+ ساخت مجدد اتصال پیش از هر تلاش"] TRA --> TRB["هرگز Dead-letter نمی‌شود:
ریختن یک OperationalError به DLQ یعنی دور انداختن
داده‌ی کاملاً سالم فقط چون Postgres در حال Restart بود"] PM --> PMA["ثبت در dead_letter_events
+ عبور Offset از روی پیام"] PMA --> PMB["هرگز Retry نمی‌شود:
Retry کردن یک JSON نامعتبر، Partition را برای همیشه قفل می‌کند"] FT --> FTA["خروج فرآیند با کد غیرصفر
← Restart کانتینر
← ادامه از آخرین Offset ثبت‌شده"] style TR fill:#f59e0b,color:#fff style PM fill:#dc2626,color:#fff style FT fill:#0f172a,color:#fff style T fill:#2563eb,color:#fff
(قطع اتصال، Deadlock، Restart پایگاه‌داده)"] T -->|"دائمی، مخصوص یک پیام"| PM["MessageDecodeError
یا package_name ثبت‌نشده"] T -->|"مرگبار (Fatal)"| FT["ConfigError, MigrationError
و هر خطای پیش‌بینی‌نشده"] TR --> TRA["Retry همان بچ با Backoff نمایی
۱s → ۲s → ۴s → ۸s → ۱۶s (۵ تلاش)
+ ساخت مجدد اتصال پیش از هر تلاش"] TRA --> TRB["هرگز Dead-letter نمی‌شود:
ریختن یک OperationalError به DLQ یعنی دور انداختن
داده‌ی کاملاً سالم فقط چون Postgres در حال Restart بود"] PM --> PMA["ثبت در dead_letter_events
+ عبور Offset از روی پیام"] PMA --> PMB["هرگز Retry نمی‌شود:
Retry کردن یک JSON نامعتبر، Partition را برای همیشه قفل می‌کند"] FT --> FTA["خروج فرآیند با کد غیرصفر
← Restart کانتینر
← ادامه از آخرین Offset ثبت‌شده"] style TR fill:#f59e0b,color:#fff style PM fill:#dc2626,color:#fff style FT fill:#0f172a,color:#fff style T fill:#2563eb,color:#fff
یک نکته‌ی طراحی: خطاهای موقتی عمداً هیچ کلاس اختصاصی در این زیرسیستم ندارند. آن‌ها همان
psycopg2.OperationalError و kafka.errors.KafkaError اصلی باقی می‌مانند. بسته‌بندی‌کردن‌شان در یک Exception داخلی، فقط تشخیص «کدام مجموعه Retry می‌شود» را سخت‌تر می‌کرد.۹ صف پیام‌های مرده در همان تراکنش
پیامی که هرگز قابل ذخیره نیست (JSON نامعتبر، فیلد هویتی غایب، یا package_name ثبت‌نشده) در جدول dead_letter_events ثبت می‌شود. نکته‌ی کلیدی این طراحی، محل Commit است: این ثبت در همان تراکنشی انجام می‌شود که داده‌های سالم همان بچ در آن نوشته می‌شوند.
این همان خصوصیتی است که یک تاپیک DLQ در کافکا نمی‌توانست بدهد: با تراکنش واحد، غیرممکن است که Offset از روی یک پیام خراب عبور کند در حالی که رکورد آن پیام گم شده باشد. اگر تراکنش Rollback شود، هم داده و هم رکورد خطا با هم برمی‌گردند و بچ دوباره تحویل داده می‌شود.
flowchart LR
subgraph TX["یک تراکنش واحد PostgreSQL"]
direction TB
G["ردیف‌های سالم
INSERT ... ON CONFLICT"] BAD["ردیف‌های ردشده
INSERT INTO dead_letter_events"] end IN["بچ ۵۰۰ پیامی
(۴۹۷ سالم + ۳ خراب)"] --> TX TX -->|"COMMIT"| DONE["هر دو با هم ماندگار شدند
← اکنون Offset جابه‌جا می‌شود"] TX -->|"ROLLBACK"| BACK["هیچ‌کدام ماندگار نشد
← Offset دست‌نخورده
← تحویل مجدد کل بچ"] style TX fill:#eff6ff,color:#0f172a style G fill:#16a34a,color:#fff style BAD fill:#dc2626,color:#fff style DONE fill:#2563eb,color:#fff
INSERT ... ON CONFLICT"] BAD["ردیف‌های ردشده
INSERT INTO dead_letter_events"] end IN["بچ ۵۰۰ پیامی
(۴۹۷ سالم + ۳ خراب)"] --> TX TX -->|"COMMIT"| DONE["هر دو با هم ماندگار شدند
← اکنون Offset جابه‌جا می‌شود"] TX -->|"ROLLBACK"| BACK["هیچ‌کدام ماندگار نشد
← Offset دست‌نخورده
← تحویل مجدد کل بچ"] style TX fill:#eff6ff,color:#0f172a style G fill:#16a34a,color:#fff style BAD fill:#dc2626,color:#fff style DONE fill:#2563eb,color:#fff
Payload خام در ستون
BYTEAنه TEXT — دقیقاً به این دلیل که پیامی که به‌خاطر UTF-8 نامعتبر رد شده هم باید قابل ذخیره باشد. Payloadهای بزرگ‌تر از ۶۴ کیلوبایت بریده می‌شوند و این بریدگی در متن دلیل خطا ثبت می‌شود.کنترل حجم لاگدر سناریوی واقعی ممکن است تمام ۵۰۰ پیام یک بچ با هم رد شوند. پس فقط ۵ مورد اول با جزئیات لاگ می‌شوند و بقیه در یک خط خلاصه‌ی گروه‌بندی‌شده بر اساس دلیل جمع می‌شوند؛ جزئیات کامل در جدول موجود است.
۱۰ حل کلید خارجی: از package_name به application_id
پیام‌های هر سه تاپیک package_name حمل می‌کنند، هرگز application_id. این تصمیم عمدی است: تولیدکنندگان (یک Scraper و یک تحلیل‌گر بسته‌های شبکه) نباید هویت پایگاه‌داده را بشناسند. تبدیل این دو به هم، وظیفه‌ی این زیرسیستم است چون مالک اسکیما خودش است.
این خواندن مستقیماً از PostgreSQL انجام می‌شود، نه از طریق Endpoint مربوط به App API که دو زیرسیستم دیگر استفاده می‌کنند. استدلال: این زیرسیستم از قبل Credential پایگاه‌داده را دارد، این جست‌وجو یک کوئری روی یک Unique Index است، و گذاشتن یک فراخوانی HTTP روی مسیر داغِ نوشتن، فقط یک نقطه‌ی شکست جدید اضافه می‌کرد بدون هیچ فایده‌ای.
| تصمیم | مقدار / رفتار | دلیل |
|---|---|---|
| یک کوئری برای کل بچ | WHERE package_name = ANY(%s) | برای ۵۰۰ نقد یک اپ، تفاوت بین یک Round-trip و ۵۰۰ Round-trip است |
| کش اختصاصی هر Thread | TTL پیش‌فرض ۳۰۰ ثانیه | مشترک کردن یک کش بین سه Thread، نیازمند Lock روی هر پیام بود تا در ساعتی یک کوئری صرفه‌جویی شود |
| عدم کش کردن نتایج ناموفق | Miss در کش ذخیره نمی‌شود | Negative Caching باعث می‌شد اپی که اپراتور همین حالا ثبت کرده، تا انقضای TTL دیده نشود |
بدون فیلتر is_active | اپ غیرفعال هم application_id می‌گیرد | داده‌ای که از قبل تولید شده باید ذخیره شود؛ Soft Delete یک حکم درباره‌ی «چه چیزی را بعداً خزش کن» است، نه درباره‌ی داده‌ی تاریخی |
| اپ ثبت‌نشده = Dead Letter | بدون raise، فقط ردشدن همان پیام | قفل کردن Partition تا زمانی که کسی اپ را ثبت کند، ingestion تمام اپ‌های دیگر را هم متوقف می‌کرد |
۱۱ مالکیت اسکیما و مهاجرت‌های SQL
یک پایگاه‌داده، اما مالکیت جدول‌ها تفکیک‌شده است. جنگو (App API) مالک جدول applications است و این زیرسیستم مالک چهار جدول خودش. این چهار جدول هیچ Model جنگویی ندارند و با فایل‌های شماره‌گذاری‌شده‌ی SQL ساخته می‌شوند.
flowchart TD
subgraph DJ["مالکیت App API - جنگو"]
A["apps_registry_application
(جدول applications)"] end subgraph SC["مالکیت Storage Consumer - SQL خام"] S1["0001_app_stats.sql"] S2["0002_reviews.sql"] S3["0003_network_metrics.sql"] S4["0004_dead_letter_events.sql"] MT["storage_consumer_migrations
(جدول رهگیری وضعیت)"] end A -.->|"Foreign Key"| S1 A -.->|"Foreign Key"| S2 A -.->|"Foreign Key"| S3 PRE["بررسی پیش‌پرواز:
to_regclass روی جدول applications"] --> DEC{"جدول وجود دارد؟"} DEC -->|"خیر"| ERR["MigrationError با پیام صریح:
«ابتدا مهاجرت‌های app_api را اجرا کنید»
به‌جای خطای مبهم Foreign Key"] DEC -->|"بله"| LOCK["pg_advisory_xact_lock
(دو کانتینر همزمان، یک بار اجرا)"] LOCK --> APPLY["اعمال فایل‌های اعمال‌نشده به‌ترتیب شماره"] style DJ fill:#f59e0b,color:#fff style SC fill:#2563eb,color:#fff style ERR fill:#dc2626,color:#fff style APPLY fill:#16a34a,color:#fff
(جدول applications)"] end subgraph SC["مالکیت Storage Consumer - SQL خام"] S1["0001_app_stats.sql"] S2["0002_reviews.sql"] S3["0003_network_metrics.sql"] S4["0004_dead_letter_events.sql"] MT["storage_consumer_migrations
(جدول رهگیری وضعیت)"] end A -.->|"Foreign Key"| S1 A -.->|"Foreign Key"| S2 A -.->|"Foreign Key"| S3 PRE["بررسی پیش‌پرواز:
to_regclass روی جدول applications"] --> DEC{"جدول وجود دارد؟"} DEC -->|"خیر"| ERR["MigrationError با پیام صریح:
«ابتدا مهاجرت‌های app_api را اجرا کنید»
به‌جای خطای مبهم Foreign Key"] DEC -->|"بله"| LOCK["pg_advisory_xact_lock
(دو کانتینر همزمان، یک بار اجرا)"] LOCK --> APPLY["اعمال فایل‌های اعمال‌نشده به‌ترتیب شماره"] style DJ fill:#f59e0b,color:#fff style SC fill:#2563eb,color:#fff style ERR fill:#dc2626,color:#fff style APPLY fill:#16a34a,color:#fff
نکته‌ی فنی که جنگو آن را پنهان می‌کرد: PostgreSQL به‌ازای یک کلید خارجی به‌طور خودکار Index نمی‌سازد (این کار را جنگو در سطح ORM انجام می‌دهد). چون این چهار جدول Model جنگویی ندارند، Indexهای لازم روی کلیدهای خارجی صریحاً در همین فایل‌های SQL ساخته شده‌اند.
۱۲ لایه‌ی پایداری: یک بچ، یک دستور INSERT
هر جدول یک Repository دارد که فقط در سه چیز با بقیه تفاوت دارد: نام جدول، فهرست ستون‌ها، و متن ON CONFLICT. تمام منطق مشترک — تبدیل کل بچ به یک دستور INSERT چندردیفی و شمارش نتیجه — یک‌جا نوشته شده است.
Repositoryها هیچ اتصالی به پایگاه‌داده نگه نمی‌دارند؛ Cursor از بیرون (توسط Worker) به آن‌ها داده می‌شود. همین نکته است که اجازه می‌دهد نوشتن داده و ثبت پیام‌های مرده در یک تراکنش واحد قرار بگیرند.
| جزئیات پیاده‌سازی | توضیح |
|---|---|
execute_values با page_size=1000 | بزرگ‌تر از بزرگ‌ترین max_poll_records (۵۰۰ برای نقدها)، پس یک بچ در حالت معمول یک Round-trip است |
RETURNING (xmax = 0) AS inserted | Postgres مقدار xmax را برای ردیف تازه Insert شده صفر و برای ردیفی که DO UPDATE جایگزینش کرده غیرصفر می‌گذارد. همین است که به یک خط لاگ اجازه می‌دهد در یک نگاه بگوید «این Backfill است یا وضعیت پایدار» |
شمارنده‌ی skipped | ردیف‌هایی که پایگاه‌داده نپذیرفت: تعارض کلید در DO NOTHING، یا به‌روزرسانی‌ای که نگهبان یکنوا رد کرد. مقدار مداوماً غیرصفر یعنی چیزی در حال Replay است |
dedupe_by_key پیش از نوشتن | یک الزام است، نه بهینه‌سازی: Postgres یک ON CONFLICT DO UPDATE که دو بار همان ردیف را در یک دستور لمس کند رد می‌کند و کل تراکنش را Abort می‌کند |
۱۳ اعتبارسنجی در لحظه‌ی Decode
Decoderها فقط ساختار پیام را بررسی نمی‌کنند؛ آن‌ها محدودیت‌های ستون پایگاه‌داده را هم در همان لحظه اعمال می‌کنند: بازه‌ی SMALLINT و INTEGER و BIGINT، طول VARCHAR، معتبر بودن UUID، و مقادیر NaN و Inf.
دلیل سخت‌گیری در جای درست: چون کل بچ در یک تراکنش نوشته می‌شود، یک مقدار خارج از بازه که به SQL برسد، تراکنش را برای کل بچ Abort می‌کند — یعنی ۴۹۹ پیام سالم هم قربانی یک پیام معیوب می‌شوند. با گرفتن آن در لحظه‌ی Decode، همان یک پیام به DLQ می‌رود و بقیه سالم نوشته می‌شوند.
یک تصمیم مرتبط دیگر: کتابخانه‌ی google-play-scraper فیلد زمانی at را بدون Timezone (Naive) تولید می‌کند، در حالی که تمام ستون‌های زمانی این زیرسیستم TIMESTAMPTZ هستند. سیاست صریح این است که زمان‌های Naive در همین لایه به UTC نسبت داده شوند، تا هیچ ابهامی به پایگاه‌داده نرسد.
۱۴ پیکربندی: Fail-Fast و قابل تنظیم
تمام تنظیمات در یک کلاس Settings واحد و frozen جمع شده‌اند که در لحظه‌ی شروع Validate می‌شود. مقدار محیطی نامعتبر، بلافاصله برنامه را با پیام واضح متوقف می‌کند، نه ساعت‌ها بعد وسط یک بچ. این کلاس Immutable و Slotted است چون تنها شیئی است که بین سه Thread مشترک است (به‌جز رخداد توقف) و همین Immutable بودن است که مدل Share-Nothing را بدون Lock ایمن می‌کند.
| تنظیم | مقدار پیش‌فرض | دلیل انتخاب |
|---|---|---|
enable_auto_commit | False | Commit خودکار بر پایه‌ی زمان است نه موفقیت نوشتن؛ می‌توانست Offset پیامی را جابه‌جا کند که هنوز ننشسته است |
auto_offset_reset | earliest | یک Group تازه باید کل Backlog موجود را بخواند، نه اینکه داده‌ی منتشرشده را نادیده بگیرد |
poll_timeout_ms | 1000 | به‌قدر کافی کوتاه که SIGTERM حداکثر یک ثانیه معطل بماند |
max_poll_interval_ms | 300000 (۵ دقیقه) | سقف زمان پردازش یک بچ پیش از آنکه Broker عضو را مرده فرض کند؛ با احتساب بدترین حالت Retry پایگاه‌داده انتخاب شده |
session_timeout_ms | 45000 | تحمل نوسان شبکه بدون Rebalance بی‌مورد |
db_retry_max_attempts | 5 با Backoff پایه‌ی ۱ ثانیه | پوشش یک Restart کامل PostgreSQL (۱+۲+۴+۸+۱۶ ثانیه) بدون از دست دادن بچ |
application_cache_ttl_seconds | 300 | تعادل بین کم کردن کوئری و دیده شدن سریع اپ تازه ثبت‌شده |
shutdown_timeout_seconds | 30 | فرصت کافی برای اتمام و Commit بچ در جریان؛ در Compose با stop_grace_period: 45s هم‌راستا شده |
تنظیم اندازه‌ی بچ در سه سطح اولویت: متغیر مخصوص همان Pipeline (مثل
STORAGE_MAX_POLL_RECORDS_REVIEWS)، سپس متغیر عمومی STORAGE_MAX_POLL_RECORDS، و در نهایت مقدار پیش‌فرض تنظیم‌شده در کد — پس اپراتور می‌تواند بدون تغییر کد آن را تنظیم کند.۱۵ Supervisor، سلامت و خاموشی تمیز
Supervisor یک قاعده را اعمال می‌کند: فرآیند تا زمانی زنده است که تمام Pipelineهایش زنده باشند. کانتینری که با دو Pipeline از سه Pipeline در حال کار باشد، Uptime سالمی نشان می‌دهد در حالی که داده بی‌صدا قطع شده است — و این از Crash کردن بدتر است، چون دیده نمی‌شود.
stateDiagram-v2
[*] --> Migrating: اجرای مهاجرت‌ها
Migrating --> Running: اسکیما به‌روز است
Migrating --> Exit1: خطای MigrationError
Running --> Running: حلقه‌ی poll / write / commit
Running --> Stopping: دریافت SIGTERM یا SIGINT
Running --> Failing: شکست یکی از سه Pipeline
Stopping --> Joined: اتمام و Commit بچ در جریان، سپس LeaveGroup
Failing --> Joined: توقف Pipelineهای دیگر هم درخواست می‌شود
Joined --> Exit0: هیچ Pipelineای خطا نداشت
Joined --> Exit1: حداقل یک Pipeline خطا داشت
Exit0 --> [*]: کد خروج صفر
Exit1 --> [*]: کد خروج یک، سپس Restart کانتینر
سلامت به‌تفکیک Pipeline، بدون سرور HTTPهر Worker در هر دور حلقه فایل مخصوص خودش را
touch می‌کند و healthcheck.sh بررسی می‌کند تمام فایل‌های مورد انتظار تازه باشند. پس کانتینری که Thread نقدهایش گیر کرده، فقط به‌خاطر «در حال اجرا بودن» سالم گزارش نمی‌شود.خطای نوشتن Heartbeat، ingestion را متوقف نمی‌کندهر خطا تنها یک بار هشدار می‌دهد و نادیده گرفته می‌شود؛ پر بودن
/tmp یک مشکل مانیتورینگ است و نباید به قطعی ingestion تبدیل شود. عمل touch هم نرخ‌محدود است تا به یک System Call در هر Poll تبدیل نشود.exec در Entrypointفرآیند پایتون PID 1 می‌شود و SIGTERM را مستقیماً دریافت می‌کند، نه اینکه یک Shell آن را ببلعد.LeaveGroup در خاموشیسیگنال در مرز Poll دیده می‌شود، بچ در جریان کامل و Commit می‌شود، سپس Consumer از Group خارج می‌شود تا Broker بلافاصله Rebalance کند — در یک Rolling Restart این تفاوت میان چند ثانیه و نزدیک یک دقیقه ingestion متوقف است.۱۶ استراتژی تست
مجموعه‌ی تست‌ها به دو بخش تقسیم می‌شود، و این تقسیم مستقیماً از یک تصمیم معماری می‌آید: هسته‌ی زیرسیستم (config، events، decoders، batching، retry_policy) عمداً نه Kafka می‌شناسد و نه psycopg2. همین است که اجازه می‌دهد قرارداد داده‌ی هر سه تاپیک بدون بالا آوردن هیچ سرویسی آزمون شود.
flowchart TD
ALL["۲۷۸ تست"] --> U["۲۲۷ تست واحد
بدون Broker، بدون پایگاه‌داده
اجرا در حدود ۱.۵ ثانیه"] ALL --> I["۵۱ تست یکپارچگی
(نشان‌دار با marker: integration)
روی Kafka و PostgreSQL واقعی"] U --> U1["قرارداد Decode و بازه‌ی ستون‌ها"] U --> U2["اعتبارسنجی Settings"] U --> U3["ترتیب write-before-commit
با Consumer و Repository جعلی"] U --> U4["مجموعه‌ی خطاهای قابل Retry"] I --> I1["Idempotency واقعی:
Reset کردن Offset و Replay کامل تاپیک"] I --> I2["جداسازی همزمانی:
تأخیر network-metrics هنگام ingest انبوه نقدها"] I --> I3["اجرای واقعی مهاجرت‌ها روی پایگاه‌داده‌ی موقت"] I --> I4["استقلال Offset هر Pipeline"] style U fill:#16a34a,color:#fff style I fill:#f59e0b,color:#fff style ALL fill:#2563eb,color:#fff
بدون Broker، بدون پایگاه‌داده
اجرا در حدود ۱.۵ ثانیه"] ALL --> I["۵۱ تست یکپارچگی
(نشان‌دار با marker: integration)
روی Kafka و PostgreSQL واقعی"] U --> U1["قرارداد Decode و بازه‌ی ستون‌ها"] U --> U2["اعتبارسنجی Settings"] U --> U3["ترتیب write-before-commit
با Consumer و Repository جعلی"] U --> U4["مجموعه‌ی خطاهای قابل Retry"] I --> I1["Idempotency واقعی:
Reset کردن Offset و Replay کامل تاپیک"] I --> I2["جداسازی همزمانی:
تأخیر network-metrics هنگام ingest انبوه نقدها"] I --> I3["اجرای واقعی مهاجرت‌ها روی پایگاه‌داده‌ی موقت"] I --> I4["استقلال Offset هر Pipeline"] style U fill:#16a34a,color:#fff style I fill:#f59e0b,color:#fff style ALL fill:#2563eb,color:#fff
تست‌های یکپارچگی همه‌ی تاپیک‌ها و اپ‌های مورد نیاز خودشان را می‌سازند و پس از پایان پاک می‌کنند، پس اجرای آن‌ها روی یک Stack فعال بی‌خطر است. اگر Kafka یا PostgreSQL در دسترس نباشند، این تست‌ها به‌جای شکست، Skip می‌شوند تا مجموعه روی ماشینی بدون کانتینر هم سبز بماند.
۱۷ عملیات: تأیید، Replay و مقیاس‌دهی
سه عملیات روزمره‌ی این زیرسیستم به‌صورت صریح مستند و ابزارمند شده‌اند.
| عملیات | روش | نکته‌ی مهم |
|---|---|---|
| تأیید سلامت استقرار | اسکریپت scripts/verify/verify_storage_consumer.sh | کل مسیر را به‌ترتیب بررسی می‌کند: نسخه‌ی اسکیما، تعداد Partition تاپیک‌ها، پیام‌های تولیدشده، ردیف‌های نوشته‌شده، پیام‌های مرده، و Lag هر Group — پس شکست، مرحله‌ی خراب را نشان می‌دهد نه کل Pipeline را |
| Replay یک تاپیک | Reset کردن Offset همان Consumer Group به earliest | چون هر تاپیک Group مستقل دارد، می‌توان reviews را برای بازسازی کامل جدولش Replay کرد بدون اینکه app-stats هم دوباره پردازش شود. Idempotency باعث می‌شود این کار داده را تکراری نکند |
| مقیاس‌دهی یک Pipeline | اجرای نسخه‌ی دوم کانتینر با STORAGE_PIPELINES مخصوص همان Pipeline | چون هر تاپیک ۳ Partition دارد، تا ۳ عضو فعال در هر Group پشتیبانی می‌شود. همین متغیر اجازه می‌دهد همان Image به‌صورت سه کانتینر تکPipeline با Restart Policy مستقل هم اجرا شود، بدون تغییر کد |
یک نکته‌ی مهم برای اولین استقرار: تمام پیام‌هایی که همین حالا روی تاپیک‌های
app-stats و reviews نشسته‌اند، از قبل از تثبیت قرارداد فعلی فیلدها هستند (ساختار خام camelCase کتابخانه). این پیام‌ها طبق طراحی به dead_letter_events می‌روند. این یک باگ نیست؛ در اولین استقرار باید این Backlog قدیمی را رد کرد (Reset کردن Offset به latest) یا تاپیک‌ها را پاک کرد.۱۸ تصمیمات کلیدی دیگر
یک Consumer Group به‌ازای هر تاپیک، نه یک Group برای هر سه
Groupهای جدا همان چیزی هستند که اجازه می‌دهند اپراتور reviews را برای بازسازی جدولش Reset کند بدون Replay شدن app-stats، و همان چیزی که Lag هر تاپیک را در kafka-ui جداگانه قابل مشاهده می‌کند.
ساخت تاپیک با سه Partition از ابتدا
تاپیک ساخته‌شده به‌صورت خودکار فقط یک Partition دارد، و این سقف هر Consumer Group را برای همیشه روی یک عضو فعال قفل می‌کند. تعداد Partition یک تاپیک قابل کاهش نیست، پس فهمیدن این موضوع زیر بار یعنی بازسازی تاپیک؛ ساختن آن‌ها از ابتدا تنها لحظه‌ی ارزان برای درست انجام دادنش است.
Repositoryها بدون اتصال و بدون State
Cursor از بیرون تزریق می‌شود تا نوشتن به تراکنش بچ ملحق شود، نه اینکه تراکنش خودش را باز کند. همین یک تصمیم است که تراکنش واحد داده و DLQ را ممکن کرده است.
تفکیک .env هاست از مقادیر Docker
فایل storage_consumer/.env فقط برای اجرای مستقل روی هاست است (localhost:9092 و 127.0.0.1)، در حالی که سرویس در Compose به kafka:29092 و postgres نیاز دارد. این زیرسیستم روشن‌ترین نمونه‌ی این تفکیک در کل پروژه است، چون تنها زیرسیستمی است که هم به Kafka و هم به PostgreSQL وصل می‌شود و در نتیجه هر دو دسته آدرس را دارد.
تکرار عامدانه‌ی retry_policy
این فایل نسخه‌ی مستقلی از همتای خود در crawler/ و network_analyzer/ است، چون هر زیرسیستم یک واحد استقرار مستقل با Image و requirements.txt خودش است و سورس بقیه در Image آن حاضر نیست. آنچه واقعاً متفاوت است، مجموعه‌ی خطاهای قابل Retry است: دو زیرسیستم دیگر HTTP را Retry می‌کنند، این یکی پایگاه‌داده را.
۱۹ ابزارها و تکنولوژی‌های استفاده‌شده
| ابزار | نقش |
|---|---|
| kafka-python (KafkaConsumer) | مصرف پیام از سه تاپیک با Commit دستی Offset و مدیریت Consumer Group |
| psycopg2 | اتصال به PostgreSQL و اجرای INSERT چندردیفی با execute_values |
| threading (stdlib) | یک Thread به‌ازای هر Pipeline، به‌همراه رخداد توقف مشترک برای خاموشی هماهنگ |
| SQL خام + جدول رهگیری مهاجرت | مالکیت اسکیمای چهار جدول این زیرسیستم، مستقل از مهاجرت‌های جنگو |
| dataclasses (frozen, slots) | تنظیمات و رخدادهای Immutable که بین Threadها بی‌خطر مشترک می‌شوند |
| python-dotenv | بارگذاری پیکربندی محیطی در حالت اجرای مستقل (venv) |
| pytest (با marker جداگانه‌ی integration) | تفکیک تست‌های بدون سرویس از تست‌هایی که به Kafka و PostgreSQL واقعی نیاز دارند |
| Docker Healthcheck + فایل‌های Heartbeat | سنجش سلامت به‌تفکیک هر Pipeline، بدون نیاز به بالا آوردن یک سرور HTTP |