Using Celery for Asynchronous Binlog Upload Processing #

Cron-driven binary log archiving introduces a critical operational vulnerability during high-throughput transactional windows. When a synchronous shell script or systemd timer triggers FLUSH BINARY LOGS, the calling process blocks until the underlying filesystem rotation completes and the subsequent network egress to object storage finishes. Under sustained write pressure the symptom is unmistakable: Waiting for binlog lock events pile up in SHOW PROCESSLIST, Threads_running climbs, replication lag inflates, and — worst of all — a delayed upload lets binlog_expire_logs_seconds purge a segment that was never copied, punching an unrecoverable hole in the GTID chain. Handing the egress work to a Celery worker pool removes that synchronous coupling so local rotation and remote upload proceed on independent clocks, and the primary’s commit latency stays flat regardless of storage-side backpressure.

Visual Overview #

Context & Prerequisites #

This page is the worker-implementation detail beneath Async Processing & Queue Management, which defines why detection must be split from transport and how per-instance ordering is preserved; read that first for the queue-topology rationale. The procedure below assumes MySQL 8.0.22+ (for stable binlog_expire_logs_seconds semantics), Python 3.10+, Celery 5.3+, and a Redis or RabbitMQ broker reachable from the archiving host but not co-located with the MySQL data volume. The worker owns the Compression & Encryption Workflows transform in-process, and its retry behaviour is a direct application of Error Handling & Retry Logic. A producer daemon — driven only by Rotation Scheduling & Cron Automation, never invoking an upload itself — is expected to publish one task per closed segment.

Step-by-Step Implementation #

1. Publish one task per rotated segment #

A systemd.path unit or inotify watcher observes the MySQL datadir and, on each newly closed binlog file, publishes a task carrying the absolute path, the server_uuid, the numeric sequence suffix, and the segment’s GTID set. Detection and transport stay in separate processes so egress bandwidth never stalls FLUSH BINARY LOGS. PITR relevance: the GTID set travels with the task, so a later recovery can map an archived object straight to a transaction boundary without re-reading the file.

# producer.py — Python 3.10+
from dataclasses import dataclass, asdict
from celery import Celery

app = Celery("binlog_worker", broker="redis://redis-broker:6379/0")


@dataclass(slots=True, frozen=True)
class BinlogTask:
    path: str
    server_uuid: str
    sequence: int
    gtid_set: str


def publish(segment: BinlogTask) -> None:
    # Route by server_uuid so per-instance ordering is preserved (see step 3).
    app.send_task(
        "upload_binlog",
        kwargs=asdict(segment),
        queue=f"binlog.{segment.server_uuid}",
    )

2. Configure consumers for at-least-once delivery #

Celery defaults optimize for throughput, not durability. For archiving you want the opposite: never acknowledge a task until the object is durably stored. PITR relevance: task_acks_late guarantees a crashed worker’s segment is re-delivered rather than silently lost, which is the difference between a complete and a severed recovery window.

# celeryconfig.py
broker_connection_retry_on_startup = True
worker_prefetch_multiplier = 1        # no task hoarding during bursty rotation
task_acks_late = True                 # ack only after a verified upload
task_reject_on_worker_lost = True     # requeue if the worker dies mid-flight
result_backend = None                 # don't persist per-segment results

Disable the result backend. Persisting metadata for thousands of daily rotations adds broker I/O and memory pressure for no recovery value; track queue depth and latency in Prometheus instead.

3. Enforce sequence-aware routing #

Binary logs must be stored in strict chronological order per instance or timestamp-targeted restores land on the wrong transaction. Bind each server’s tasks to a dedicated queue and run exactly one worker (concurrency 1) per queue, or shard with sequence % N while keeping each shard single-threaded. PITR relevance: ordering within a server_uuid is what lets mysqlbinlog --stop-position trust that segment N-1 is already archived before N is acknowledged.

# celeryconfig.py — one ordered lane per instance
task_routes = {
    "upload_binlog": {"queue": "binlog.default"},
}
task_queue_max_priority = None  # priority queues reorder work; never enable here

4. Guard staging disk before compressing #

Compression and encryption run in a staging directory. If that volume saturates mid-stream you get OSError: [Errno 28] No space left on device and a truncated payload that fails checksum on restore. A task_prerun handler that fails fast is cheaper than a corrupt object. PITR relevance: a fast abort requeues the segment while the source binlog still exists locally, preserving a second chance at the archive.

# watermark.py — Python 3.10+
import os
from celery.signals import task_prerun

STAGING_PATH = "/var/lib/mysql/staging"
MIN_FREE_BYTES = 2 * 1024**3  # keep 2x the largest expected compressed payload


@task_prerun.connect
def validate_staging_space(sender=None, task_id=None, **kwargs) -> None:
    stat = os.statvfs(STAGING_PATH)
    if (free := stat.f_bavail * stat.f_frsize) < MIN_FREE_BYTES:
        raise RuntimeError(f"Staging watermark breached: {free} bytes free")

5. Compress, encrypt, and upload inside the worker #

Stream zstd at level 3 (a 60–80% ratio on typical binlogs at low CPU cost) then encrypt with AES-256-GCM using a KMS-issued data key — see Implementing AES-256 Encryption for Archived Binlogs for the streaming key-management detail. Verify the multipart ETag before returning. PITR relevance: encrypting a still-open file yields InvalidTag on restore, so the worker must only ever touch a closed, rotated segment.

6. Retry transient failures without duplicate uploads #

Network partitions, DNS blips, and IAM credential rotation are routine. Classify errors explicitly: retry the transient ones with exponential backoff plus jitter, and let non-retryable ones (missing file, checksum mismatch) fail loudly. PITR relevance: blind retries on a checksum mismatch would archive corrupt data as if it were valid, so classification is a recovery-integrity gate, not just resilience.

# tasks.py — Python 3.10+
import random
import botocore.exceptions
from celery import Celery

app = Celery("binlog_worker", broker="redis://redis-broker:6379/0")


@app.task(bind=True, name="upload_binlog", max_retries=5, acks_late=True)
def upload_binlog(self, path: str, server_uuid: str, sequence: int, gtid_set: str):
    try:
        perform_multipart_upload(path, sequence)
    except Exception as exc:
        match exc:
            case ConnectionError() | botocore.exceptions.HTTPClientError():
                countdown = 2 ** self.request.retries + random.uniform(0, 1)
                raise self.retry(exc=exc, countdown=countdown)
            case FileNotFoundError() | ValueError():
                raise  # non-retryable: surface immediately, do not archive
            case _:
                raise


def perform_multipart_upload(path: str, sequence: int) -> None:
    """Placeholder: implement the actual S3/GCS multipart upload here."""
    raise NotImplementedError

Refresh IAM tokens proactively with STS AssumeRole or Workload Identity Federation on a background thread; never hardcode credentials in celeryconfig.py. Sustained 503 SlowDown / 429 responses are a storage-side signal, not a network fault — handle them per Handling S3 Throttling During High-Throughput Binlog Archiving.

7. Coordinate with the base backup and expose metrics #

Tag every uploaded segment with the base backup’s GTID set and append it to an object-storage manifest so a restore can target a precise timestamp — the coordination contract is defined in Base Backup Integration for PITR. Export celery_queue_length, task success/failure counts, and upload p95 latency to Prometheus. PITR relevance: a growing queue is early warning that egress is falling behind rotation, i.e. that your recovery point is drifting.

Configuration Reference #

Minimal, copy-pasteable settings that make Celery safe for ordered, durable archiving:

-- MySQL 8.0.22+ : local retention must outlast the worst-case queue drain,
-- never the reverse. Size this to comfortably exceed peak archiving lag.
SET GLOBAL binlog_expire_logs_seconds = 259200;  -- 72h safety margin

Verification Checklist #

• Producer publishes exactly one task per closed binlog segment (no open-file uploads).
• worker_prefetch_multiplier = 1 and one worker per instance queue confirmed via celery inspect active_queues.
• task_acks_late and task_reject_on_worker_lost verified by killing a worker mid-upload and observing re-delivery.
• Broker visibility_timeout exceeds the slowest observed multipart upload.
• Staging-disk task_prerun guard aborts cleanly when free space drops below the watermark.
• Downloaded objects pass mysqlbinlog --verify-binlog-checksum and decrypt without InvalidTag.
• Manifest rows carry the correct gtid_set and map one-to-one onto stored objects.
• Prometheus alerts fire on celery_queue_length growth before binlog_expire_logs_seconds would purge unarchived logs.
• A dry-run restore to a staging instance replays the archived sequence with no GTID gaps.

Gotchas & Version-Specific Caveats #

Result-backend memory creep. Leaving result_backend pointed at Redis silently stores a key per task. At thousands of rotations a day this evicts hot broker data; set it to None.

Redis visibility-timeout duplicates. If a multipart upload runs longer than the Redis visibility_timeout, the broker re-delivers the task and a second worker races the first — duplicate objects with mismatched ETags. Always size the timeout above your slowest upload, or use RabbitMQ where late acks are native.

SHOW MASTER STATUS is gone in 8.4. Producer or verification scripts that scrape the current coordinate must use SHOW BINARY LOG STATUS on MySQL 8.4+; SHOW MASTER STATUS is removed, and SHOW BINARY LOGS replaces SHOW MASTER LOGS. Guard for the running version rather than assuming 8.0 syntax.

-- MySQL 8.4+ : replacement for the removed SHOW MASTER STATUS
SHOW BINARY LOG STATUS;

WRITESET dependency tracking. binlog_transaction_dependency_tracking was deprecated and the WRITESET_SESSION value behaviour tightened in the 8.x line; do not rely on parallel-apply hints surviving an archive round-trip — the archived file preserves events, not the applier’s scheduling metadata.

prefetch > 1 breaks ordering, not just fairness. Any prefetch above 1 on a shared queue lets a worker acknowledge segment N+1 before N is durable. For archiving this is a correctness bug, not a tuning knob.

Never treat binlog_expire_logs_seconds as an upload backstop. It is a wall-clock timer independent of Celery. If the queue stalls, the timer will still purge unarchived segments; the retention window must always be longer than your worst-case drain time, and purge should be gated on verified archival — see Binlog Retention Boundaries.

Related #

• Async Processing & Queue Management — the queue topology and backpressure model this worker plugs into.
• Handling S3 Throttling During High-Throughput Binlog Archiving — the adaptive-retry and prefix-sharding companion to step 6.
• GTID Tracking & Enforcement — verifying the archived sequence is gap-free before a recovery trusts it.

Back to Async Processing & Queue Management.