Architecture
Asynchronous & Background Task Pipeline
Celery workers, Redis message broker, distributed locks, and concurrency management.
Asynchronous & Background Task Pipeline
IMPLEMENTED
Long-running I/O operations, network polling, payment ingestion, and recurring billing runs are decoupled from synchronous HTTP cycles via Celery and Redis.
1. Asynchronous Invariants
- Explicit Tenant Scope: Every Celery task must accept an explicit
tenant_idparameter. Celery workers do not have an HTTP request context and must never assumerequest.tenant. - Distributed Locking: All recurring mutations (recharge, auto-lock, invoice generation, router telemetry sync) must acquire a Redis distributed lock (
apps.core.lock.distributed_lock) to prevent race conditions across parallel workers. - Idempotent Tasks: Tasks are written to be safely retried without creating duplicate invoices or double-crediting balances.
2. Core Background Tasks (backend/apps/core/tasks.py)
| Task Name | Schedule / Trigger | Purpose | Lock Key |
|---|---|---|---|
expire_customers | Daily midnight cron | Locks subscribers whose expiry date has passed (respecting grace periods), terminates sessions | lock:expiry:{tenant_id} |
generate_monthly_invoices | 1st of every month / API | Generates monthly recurring subscription invoices, itemized lines, and auto-settles advance | lock:invoice_gen:{tenant_id} |
process_network_sync_job | Post-commit dispatch | Asynchronously executes MikroTik/OLT actions (ENABLE_USER, DISABLE_USER, etc.) after DB commit | lock:network_sync:{tenant_id}:{job_id} |
process_payment_event | Inbound SMS webhook | Matches unlinked MFS transactions against subscribers, extends expiry, posts ledger entries | lock:payment_event:{tenant_id}:{event_id} |
process_recharge | Staff manual recharge | Applies subscription renewals, updates expiry dates, updates queues | lock:recharge:{customer_id} |
sync_router_telemetry | Every 5 minutes | Polls MikroTik routers for CPU, memory, uptime, and active PPPoE count | lock:router_sync:{router_id} |
execute_database_backup | Scheduled / On-demand | Dumps PostgreSQL database and uploads to backup storage | lock:backup:{tenant_id} |
3. Distributed Lock Pattern
from apps.core.lock import distributed_lock, LockAcquisitionError
@shared_task(bind=True, max_retries=3)
def process_recharge(self, tenant_id, customer_id, amount):
lock_key = f"lock:recharge:{customer_id}"
try:
with distributed_lock(lock_key, timeout=60, blocking=False):
with transaction.atomic():
customer = Customer.objects.select_for_update().get(id=customer_id, tenant_id=tenant_id)
# ... perform recharge, ledger entry, and MikroTik queue profile update ...
except LockAcquisitionError:
logger.warning("Recharge task for customer %s skipped: lock active", customer_id)