Backend Live Coding

# ======================================================================
# Q: Every screen sends a heartbeat every 60 seconds. How do you store it?
#
# 250,000 screens / 60 s is about 4,000 writes per second for heartbeats alone.
# One UPDATE per heartbeat in PostgreSQL creates dead tuples, autovacuum load
# and locks. I write the heartbeat to Redis and a periodic task moves the data
# to the database with a bulk update every few minutes (write coalescing).
# "Offline" then becomes a fast query in Redis. If a screen stays offline, I
# notify the customer (email/Slack).
# ======================================================================

# in the heartbeat endpoint: O(1), no database
redis.zadd('screens:last_seen', {screen_id: time.time()})

# Celery beat every 2 minutes
@shared_task
def flush_heartbeats():
    cutoff = time.time() - 120
    rows = redis.zrangebyscore('screens:last_seen', cutoff, '+inf', withscores=True)
    for chunk in batched(rows, 1000):
        Screen.objects.bulk_update(
            [Screen(pk=int(pk), last_seen=datetime.fromtimestamp(ts, UTC)) for pk, ts in chunk],
            ['last_seen'])

# Which screens are offline? ZRANGEBYSCORE with a score less than now - 5 minutes

# ======================================================================
# Q: How do you make sure a periodic task does not run twice at the same time?
#
# The timeout on the lock is important: if the worker dies, the lock expires by
# itself and the system doesn't get stuck.
# ======================================================================

@shared_task
def sync_all_calendars():
    # cache.add = SET NX: succeeds only if the key does not exist
    if not cache.add('lock:sync_all_calendars', 1, timeout=600):
        logger.info('previous run still in progress, skipping')
        return
    try:
        for ids in batched(active_calendar_ids(), 500):
            sync_calendar_batch.delay(list(ids))
    finally:
        cache.delete('lock:sync_all_calendars')

# ======================================================================
# Q: Which Celery settings do you always change in production?
#
# Separate queues and workers for heavy tasks (video), for integrations and for
# fast tasks (emails), so one doesn't block the others. Celery beat runs on a
# single instance, otherwise periodic tasks run twice.
# ======================================================================

CELERY_TASK_ACKS_LATE = True               # ack after execution
CELERY_TASK_REJECT_ON_WORKER_LOST = True
CELERY_WORKER_PREFETCH_MULTIPLIER = 1       # fair distribution for large tasks
CELERY_TASK_TIME_LIMIT = 300                # hard kill
CELERY_TASK_SOFT_TIME_LIMIT = 270           # SoftTimeLimitExceeded for cleanup
CELERY_WORKER_MAX_TASKS_PER_CHILD = 1000    # safety net for memory leaks
CELERY_TASK_ROUTES = {
    'media.tasks.transcode': {'queue': 'heavy'},
    'integrations.tasks.*': {'queue': 'integrations'},
}
# Redis broker: visibility_timeout > the slowest task, otherwise double execution
CELERY_BROKER_TRANSPORT_OPTIONS = {'visibility_timeout': 3600}

# ======================================================================
# Q: What are chain, group and chord? Give an example.
#
# A chord needs a result backend and is sensitive if one part fails. For very
# large fan-outs (e.g. 250,000 screens) I don't create one task per screen but
# batches (e.g. 500 ids per task) so the broker doesn't fill up.
# ======================================================================

from celery import chain, group, chord

# chain: one after the other, the result is passed to the next
chain(download_media.s(media_id), transcode.s(), make_thumbnail.s())()

# group: in parallel
group(push_update.s(screen_id) for screen_id in screen_ids)()

# chord: in parallel, then a callback when all are done
chord(render_report_part.s(i) for i in range(10))(merge_report.s(report_id))

# ======================================================================
# Q: A provider is down for hours. How do you protect the system?
#
# Circuit breaker: after N consecutive failures I stop calling it for a while
# and serve the last good value from cache (graceful degradation: the screen
# shows yesterday's weather instead of nothing). A separate Celery queue per
# integration, so one slow API doesn't fill everyone's workers. Alerts on the
# error rate per provider.
# ======================================================================

def call_with_breaker(name, func, *args, threshold=5, cooldown=120):
    if cache.get(f'breaker:{name}:open'):
        raise ServiceUnavailable(name)
    try:
        result = func(*args)
    except ExternalAPIError:
        cache.get_or_set(f'breaker:{name}:fails', 0, 300)   # initialization with a TTL
        if cache.incr(f'breaker:{name}:fails') >= threshold:
            cache.set(f'breaker:{name}:open', 1, cooldown)
        raise
    cache.delete(f'breaker:{name}:fails')
    return result