Celery in Plane's Architecture: How the Open-Source Project Management Platform Handles Asynchronous Workflows

Celery decouples time-consuming operations from Plane's Django request-response cycle, enabling background processing for email notifications, metrics, and scheduled maintenance tasks.

Plane, an open-source project management platform, uses Celery as its distributed task queue to handle operations that would otherwise block HTTP requests. The integration resides in the apps/api/plane package and powers everything from real-time email delivery to daily data cleanup.

Where Celery Lives in Plane's Codebase

The Celery integration spans several carefully organized files that establish the task infrastructure, expose it to workers, and define domain-specific background jobs.

Core Celery App Configuration

The file apps/api/plane/celery.py instantiates the Celery application and wires it into Django's settings system.

# apps/api/plane/celery.py (conceptual structure)

from celery import Celery
from celery.beat import crontab

app = Celery("plane")

# Load configuration from Django settings

app.config_from_object("django.conf:settings", namespace="CELERY")

# Configure database-backed scheduler for persistent cron jobs

app.conf.beat_scheduler = "django_celery_beat.schedulers.DatabaseScheduler"

# Register task modules

app.autodiscover_tasks()

This configuration uses django-celery-beat with a DatabaseScheduler, allowing Plane to store periodic task schedules in the database rather than static configuration files. This enables runtime schedule modifications without worker restarts.

Celery App Exposure

The apps/api/plane/__init__.py file makes the Celery app discoverable by importing it as celery_app:

# apps/api/plane/__init__.py

from .celery import app as celery_app

__all__ = ("celery_app",)

This pattern follows Django's recommended convention and ensures external Celery workers can locate the application when started with --app=plane.

Celery Beat: Scheduled Tasks in Plane

Plane leverages Celery Beat for recurring operations. The beat_schedule configuration in celery.py maps task names to execution intervals using crontab or schedule objects.

# Example beat schedule configuration from plane/celery.py

app.conf.beat_schedule = {
    "check-every-five-minutes-to-send-email-notifications": {
        "task": "plane.bgtasks.email_notification_task.send_email_notification",
        "schedule": 300.0,  # 5 minutes in seconds

    },
    "push-instance-metrics": {
        "task": "plane.bgtasks.silo_task.push_instance_metrics",
        "schedule": crontab(minute=0),  # hourly

    },
    "clear-old-sessions": {
        "task": "plane.bgtasks.cleanup_task.delete_api_logs",
        "schedule": crontab(hour=2, minute=30),  # daily at 02:30 UTC

    },
}

These scheduled tasks run without manual intervention, handling:

• Email notification batching — aggregated delivery every five minutes
• Instance metrics — hourly telemetry pushes for monitoring
• Data retention — nightly cleanup of expired API logs and session data

Background Task Implementation with @shared_task

Plane defines individual tasks in the bgtasks module using Celery's @shared_task decorator. This makes tasks discoverable without coupling them to a specific Celery app instance.

Email Notification Pipeline

The file apps/api/plane/bgtasks/email_notification_task.py demonstrates a complex task chain:

# apps/api/plane/bgtasks/email_notification_task.py (simplified)

from celery import shared_task, chain
from plane.settings.redis import redis_instance

@shared_task(bind=True, max_retries=3)
def send_email_notification(self):
    """Fetch pending notifications and enqueue email delivery."""
    # Acquire Redis lock to prevent duplicate processing

    lock_key = "send_email_notification_lock"
    if not redis_instance.set(lock_key, "1", nx=True, ex=300):
        return "Lock already held, skipping"
    
    try:
        # Task logic: fetch notifications, build payloads, send emails

        pending = get_pending_notifications()
        for notification in pending:
            build_and_send_email.delay(notification.id)
    finally:
        redis_instance.delete(lock_key)

This task uses Redis locking (via plane.settings.redis.redis_instance) to guarantee idempotency when multiple workers compete for the same notification batch. The max_retries=3 parameter ensures transient failures trigger automatic retry with exponential backoff.

Triggering Tasks from Django Views

Plane's views enqueue background work without waiting for completion:

# Example view triggering asynchronous task

from django.http import JsonResponse
from plane.bgtasks.example_task import add_numbers

def some_view(request):
    """Enqueue computation and return immediately with task ID."""
    result = add_numbers.delay(3, 5)  # Non-blocking

    return JsonResponse({
        "task_id": result.id,
        "status": "queued"
    })

The .delay() method serializes arguments to the message broker and returns an AsyncResult instantly, keeping HTTP response times predictable regardless of task complexity.

Redis as Locking Backend

Many Plane tasks require distributed coordination. The plane.settings.redis module provides a Redis client that tasks import for mutual exclusion:

Usage Pattern	Purpose	Example in Plane
set(key, value, nx=True, ex=seconds)	Acquire time-bounded lock	send_email_notification prevents duplicate email sends
delete(key)	Release lock after completion	Cleanup in task finally blocks
get(key)	Check lock status	Conditional task skipping

This Redis integration complements Celery's own broker functionality (typically Redis or RabbitMQ) by adding application-level locking where Celery's built-in mechanisms don't suffice.

Common Celery Tasks in Plane

The bgtasks directory contains domain-specific background operations:

• email_notification_task.py — Notification aggregation and delivery chains
• workspace_seed_task.py — Template project generation for new workspaces
• issue_automation_task.py — Recurring issue status transitions and SLA checks
• cleanup_task.py — Data retention, expired token removal, log rotation
• silo_task.py — Instance telemetry and metrics pushing

Each module follows the @shared_task pattern and may combine simple functions with Celery's chain, group, or chord primitives for multi-stage workflows.

Summary

• Celery's role in Plane — Decouples blocking operations (email, metrics, cleanup) from Django's synchronous request handling
• Entry points — celery.py for configuration, __init__.py for app exposure, bgtasks/ for task definitions
• Scheduling — DatabaseScheduler enables persistent, database-backed periodic tasks without worker restarts
• Reliability patterns — Redis locking for idempotency, max_retries for automatic failure recovery, task chains for multi-stage workflows
• Scalability — Worker processes scale independently from web servers; beat scheduler can run as separate service

Frequently Asked Questions

What message broker does Plane use with Celery?

Plane typically uses Redis as both message broker and result backend, though the architecture supports RabbitMQ. The CELERY_BROKER_URL setting in Django configuration determines the broker, with Redis providing the additional benefit of distributed locking via plane.settings.redis.

How does Plane prevent duplicate task execution?

Tasks like send_email_notification acquire a Redis lock using nx=True (only set if not exists) with an expiration time. If the lock exists, the task exits immediately. This pattern protects against race conditions when multiple workers process the same scheduled job or retry scenario.

What's the difference between .delay() and .apply_async() in Plane's codebase?

Plane primarily uses .delay() for simple fire-and-for queueing, which wraps .apply_async() with default options. For tasks requiring specific routing, countdown delays, or custom retry policies, contributors can use .apply_async(countdown=60, queue='priority') directly.

Can Plane's Celery beat schedule be modified at runtime?

Yes. Because Plane configures django_celery_beat.schedulers.DatabaseScheduler, administrators can add, modify, or disable periodic tasks through Django's admin interface or ORM without restarting Celery workers. Changes persist in the database and take effect on the next scheduler tick.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}