Job locks and rate limits (FastAPI + workers)

API rate limit ──► enqueue (maybe) ──► RabbitMQ
 │
 job rate limit + lock ──► side effect

# PSEUDOCODE : SET key value NX EX ttl
async def acquire_lock(redis, key: str, token: str, ttl_sec: int) -> bool:
 return await redis.set(key, token, nx=True, ex=ttl_sec)

async def release_lock(redis, key: str, token: str) -> None:
 # Lua: delete only if value == token (avoid deleting someone else's lock)
 await redis.eval(RELEASE_LUA, keys=[key], args=[token])

# PSEUDOCODE : worker
@celery_app.task(bind=True, max_retries=10)
def update_score(self, job_id: str, user_id: str):
 token = new_uuid()
 key = f"lock:score:{user_id}"
 if not acquire_lock(redis, key, token, ttl_sec=120):
 # another worker holds it : retry later (job rate / overlap)
 raise self.retry(countdown=15)
 try:
 mark_running(job_id)
 do_update(user_id)
 mark_succeeded(job_id)
 finally:
 release_lock(redis, key, token)

-- PSEUDOCODE : Postgres advisory lock
SELECT pg_try_advisory_lock(hashtext(:user_id));

-- PSEUDOCODE
UPDATE jobs
SET status = 'running', locked_by = :worker, locked_at = now()
WHERE id = :id AND status = 'pending'
RETURNING *;

# PSEUDOCODE : unique enqueue
key = f"reindex:{product_id}"
job = await db.insert_job_if_absent(idempotency_key=key, ...)
if job.already_existed:
 return job # 202 with existing id
await bus.publish(...)

# PSEUDOCODE : lock while running (even if two slipped through)
lock_key = f"lock:reindex:{product_id}"

# PSEUDOCODE : Redis fixed window (simple)
async def allow(redis, key: str, limit: int, window_sec: int) -> bool:
 n = await redis.incr(key)
 if n == 1:
 await redis.expire(key, window_sec)
 return n <= limit

# key examples:
# ratelimit:ip:{ip}
# ratelimit:user:{user_id}:enqueue

# PSEUDOCODE : FastAPI dependency
async def rate_limit_enqueue(user=Depends(auth), redis=Depends(get_redis)):
 ok = await allow(redis, f"rl:enqueue:{user.id}", limit=30, window_sec=60)
 if not ok:
 raise HTTPException(429, "Too many job submissions")

# PSEUDOCODE : Celery built-in (process-local-ish; prefer Redis for multi-worker fairness)
@celery_app.task(rate_limit="30/m")
def send_email(job_id: str):
 ...

# PSEUDOCODE : Redis token bucket shared across workers
async def take_token(redis, key: str, rate_per_sec: float, burst: int) -> bool:
 # classic token bucket Lua or redis-cell / redis-py bucket
 ...

@celery_app.task(bind=True, max_retries=20)
def call_vendor(self, job_id: str, tenant_id: str):
 if not take_token(redis, f"rl:vendor:{tenant_id}", rate_per_sec=5, burst=10):
 raise self.retry(countdown=2)
 do_http_call(...)

# PSEUDOCODE
@celery_app.task(bind=True, max_retries=25)
def sync_account(self, job_id: str, account_id: str):
 if not take_token(redis, "rl:sync-global", rate_per_sec=20, burst=40):
 raise self.retry(countdown=1)

 token = new_uuid()
 lock_key = f"lock:account:{account_id}"
 if not acquire_lock(redis, lock_key, token, ttl_sec=300):
 raise self.retry(countdown=10)

 try:
 if job_already_succeeded(job_id):
 return
 mark_running(job_id)
 run_sync(account_id)
 mark_succeeded(job_id)
 except TransientError as e:
 mark_retry(job_id, str(e))
 raise self.retry(countdown=backoff(self.request.retries))
 except PermanentError as e:
 mark_failed(job_id, str(e))
 finally:
 release_lock(redis, lock_key, token)

-- RELEASE_LUA
if redis.call("get", KEYS[1]) == ARGV[1] then
 return redis.call("del", KEYS[1])
else
 return 0
end

# PSEUDOCODE
async def with_lock(key, ttl=60):
 token = new_uuid()
 if not acquire(key, token, ttl):
 raise LockBusy()
 renew = asyncio.create_task(renew_loop(key, token, ttl))
 try:
 yield
 finally:
 renew.cancel()
 release(key, token)