# Scheduling & Delayed Delivery (/integrations/celery/how-to/scheduling)



## Overview [#overview]

Celery's `countdown` and `eta` arguments map directly onto KubeMQ's native `delay_in_seconds` on the [Queues](/learn/queues) messages that carry every task. The broker holds the message server-side and delivers it when the delay elapses — there is **zero client-side polling** and **no broker plugin** to install.

This is a real difference from other brokers. With Redis, Celery implements `countdown`/`eta` using a client-side polling loop. With RabbitMQ, you need the `rabbitmq_delayed_message_exchange` plugin. With KubeMQ, delayed delivery is a built-in property of the queue message:

| Broker   | Delayed delivery mechanism                        |
| -------- | ------------------------------------------------- |
| Redis    | Client-side polling loop                          |
| RabbitMQ | `rabbitmq_delayed_message_exchange` plugin        |
| KubeMQ   | Native `delay_in_seconds` — no polling, no plugin |

From the API perspective nothing changes — `apply_async(countdown=...)` and `apply_async(eta=...)` work exactly as they do on any Celery broker:

```python
task.apply_async(countdown=60)          # delivers after 60 seconds
task.apply_async(eta=future_datetime)   # delivers at a specific time
```

<Callout type="info">
  **One-time scheduling vs. recurring schedules.** `countdown` and `eta` schedule a task to run **once** at a future time. For tasks that repeat on a fixed cadence (every minute, daily, at sunset), use **Celery Beat** — covered later on this page.
</Callout>

### Run a broker [#run-a-broker]

All examples on this page connect to a KubeMQ broker over the native gRPC port `50000`. The Celery transport is a Kombu virtual transport — it talks gRPC directly, so no HTTP connector flag is required (port `9090` is the shared HTTP server for the CloudEvents, REST, MCP, and A2A connectors, which Celery does not use).

<RunKubeMQ ports="[50000, 9090]" />

```bash
pip install kubemq-celery
```

## Countdown [#countdown]

`apply_async(countdown=N)` tells KubeMQ to hold the message for `N` seconds before delivering it to a worker. Tasks with a shorter countdown are delivered first, regardless of dispatch order.

```python title="countdown_delay.py"
import os
import time

from celery import Celery

import kubemq_celery  # noqa: F401

app = Celery("countdown_delay")
app.config_from_object(
    {
        "broker_url": os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
        "result_backend": os.environ.get("CELERY_RESULT_BACKEND", "kubemq://localhost:50000"),
        "result_expires": 3600,
    }
)


@app.task
def delayed_greeting(name: str) -> dict:
    """A greeting that arrives after a countdown delay."""
    return {"message": f"Hello, {name}!", "delivered_at": time.time()}


@app.task
def scheduled_cleanup(resource_id: str) -> dict:
    """Clean up a resource after a delay."""
    return {"resource_id": resource_id, "action": "cleaned_up", "at": time.time()}
```

Dispatch tasks with different countdowns. `countdown=N` maps to KubeMQ's `delay_in_seconds` — no client-side polling is involved:

```python
# Delivered 5s after dispatch.
r1 = delayed_greeting.apply_async(args=("Alice",), countdown=5)

# Dispatched second, but delivered first (shorter delay).
r2 = delayed_greeting.apply_async(args=("Bob",), countdown=3)

# Delivered 10s after dispatch.
r3 = scheduled_cleanup.apply_async(args=("res-001",), countdown=10)
```

Run a worker against the broker to observe the delays:

```bash
celery -A countdown_delay worker --loglevel=info
```

## ETA [#eta]

`apply_async(eta=datetime)` schedules a task for an **exact** wall-clock time. The transport computes `delay_in_seconds = (eta - now).total_seconds()` and hands that to KubeMQ. Always pass a timezone-aware datetime — UTC is recommended, and Celery's `enable_utc` keeps ETA math unambiguous.

```python title="eta_scheduling.py"
import os
import time
from datetime import datetime, timedelta, timezone

from celery import Celery

import kubemq_celery  # noqa: F401

app = Celery("eta_scheduling")
app.config_from_object(
    {
        "broker_url": os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
        "result_backend": os.environ.get("CELERY_RESULT_BACKEND", "kubemq://localhost:50000"),
        "result_expires": 3600,
        "enable_utc": True,
    }
)


@app.task
def run_at_time(label: str) -> dict:
    """Task scheduled to run at a specific ETA."""
    return {
        "label": label,
        "executed_at": datetime.now(timezone.utc).isoformat(),
        "timestamp": time.time(),
    }
```

Schedule tasks for specific future times. As with `countdown`, the task with the earlier ETA is delivered first:

```python
now = datetime.now(timezone.utc)

# Runs 10 seconds from now.
run_at_time.apply_async(args=("10s-from-now",), eta=now + timedelta(seconds=10))

# Scheduled after the first call, but executes earlier (earlier ETA).
run_at_time.apply_async(args=("5s-from-now",), eta=now + timedelta(seconds=5))
```

<Callout type="info">
  **`eta` vs. `countdown`** — `eta=datetime` runs at a specific timezone-aware time; `countdown=seconds` runs after N seconds from now. Both compile down to the same KubeMQ `delay_in_seconds` primitive. Use timezone-aware datetimes (UTC recommended) to avoid drift.
</Callout>

## The 24-hour delay cap [#the-24-hour-delay-cap]

KubeMQ's `delay_in_seconds` caps at &#x2A;*86400 seconds (24 hours)**. If a `countdown` or `eta` exceeds 24 hours, the transport **caps the effective delay at 24h and emits a warning log** rather than rejecting the task. For longer horizons, use Celery Beat (below).

<Callout type="warn">
  **Delays over 24 hours are capped.** A `countdown=172800` (48h) is delivered after 24h, not 48h, with a warning. The same cap applies to ETA values more than 24 hours in the future, and to `result_expires` (queue message expiration, also 86400s max). For schedules beyond a day, use **Celery Beat**; for one-shot tasks beyond a day, store the target time in a database and let Beat poll it.
</Callout>

The behavior is illustrated below — delays at or under the limit pass through, while anything over is clamped to `86400`:

```python title="beat_max_delay_warning.py"
MAX_DELAY = 86400  # 24 hours in seconds

delay_tests = [
    ("within-limit-1h", 3600),    # OK — delivered after 1 hour
    ("within-limit-12h", 43200),  # OK — delivered after 12 hours
    ("at-limit-24h", 86400),      # OK — delivered after exactly 24 hours
    ("over-limit-48h", 172800),   # CAPPED — 48h requested, delivered after 24h
    ("over-limit-7d", 604800),    # CAPPED — 7 days requested, delivered after 24h
]

for label, delay in delay_tests:
    effective = min(delay, MAX_DELAY)
    if delay > MAX_DELAY:
        print(f"WARNING: delay {delay}s capped to {effective}s")
```

## Celery Beat — crontab schedules [#celery-beat--crontab-schedules]

For **recurring** tasks, run the **Celery Beat** scheduler. Beat owns the schedule and publishes each task to KubeMQ when it is due; workers then pick it up like any other message. Use `crontab()` for clock-based cadences (specific minutes, hours, days of week, days of month).

```python title="beat_crontab.py"
import os
from datetime import datetime, timezone

from celery import Celery
from celery.schedules import crontab

import kubemq_celery  # noqa: F401

app = Celery("beat_crontab")
app.config_from_object(
    {
        "broker_url": os.environ.get("CELERY_BROKER_URL", "kubemq://localhost:50000"),
        "result_backend": os.environ.get("CELERY_RESULT_BACKEND", "kubemq://localhost:50000"),
        "result_expires": 3600,
        "enable_utc": True,
        "beat_schedule": {
            # Every minute (crontab() defaults to every minute).
            "every-minute-health": {
                "task": "beat_crontab.health_check",
                "schedule": crontab(),
            },
            # Every 5 minutes.
            "every-5min-metrics": {
                "task": "beat_crontab.collect_metrics",
                "schedule": crontab(minute="*/5"),
            },
            # Daily at midnight UTC.
            "daily-report": {
                "task": "beat_crontab.generate_daily_report",
                "schedule": crontab(minute=0, hour=0),
            },
            # Every Monday at 9:00 AM UTC.
            "weekly-digest": {
                "task": "beat_crontab.send_weekly_digest",
                "schedule": crontab(minute=0, hour=9, day_of_week="monday"),
            },
            # 1st of every month at 6:00 AM UTC.
            "monthly-billing": {
                "task": "beat_crontab.run_monthly_billing",
                "schedule": crontab(minute=0, hour=6, day_of_month=1),
            },
        },
    }
)


@app.task
def health_check() -> dict:
    return {"status": "healthy", "at": datetime.now(timezone.utc).isoformat()}
```

Start a worker and a Beat scheduler in separate processes:

```bash
# Terminal 1: worker
celery -A beat_crontab worker --loglevel=info

# Terminal 2: beat scheduler
celery -A beat_crontab beat --loglevel=info
```

<Callout type="warn">
  **Run exactly one Beat instance.** Beat state is **local** — it lives in a `celerybeat-schedule` file, not in KubeMQ. Running multiple Beat processes produces duplicate task dispatches. The combined `worker -B` form is for development only; in production, run a single dedicated Beat process.
</Callout>

## Celery Beat — intervals and solar schedules [#celery-beat--intervals-and-solar-schedules]

Beyond crontab, Beat supports fixed-interval schedules with `timedelta` and astronomical schedules with `solar`.

<Tabs groupId="celery-beat" items="['Timedelta intervals', 'Solar schedules']">
  <Tab value="Timedelta intervals">
    A `timedelta` schedule runs a task at a fixed interval measured from its previous run. Use it for heartbeats, polling, and periodic syncs. Entries can carry positional `args` and keyword `kwargs`.

    ```python title="beat_timedelta.py"
    from datetime import timedelta

    # inside app.config_from_object({... "beat_schedule": { ... }})
    "beat_schedule": {
        # Every 10 seconds.
        "heartbeat-10s": {
            "task": "beat_timedelta.heartbeat",
            "schedule": timedelta(seconds=10),
        },
        # Every 30 seconds, with positional args.
        "sensor-reading-30s": {
            "task": "beat_timedelta.read_sensor",
            "schedule": timedelta(seconds=30),
            "args": ("temperature",),
        },
        # Every 2 minutes, with keyword args.
        "cache-refresh-2m": {
            "task": "beat_timedelta.refresh_cache",
            "schedule": timedelta(minutes=2),
            "kwargs": {"cache_name": "user_profiles", "ttl": 300},
        },
        # Every hour.
        "hourly-sync": {
            "task": "beat_timedelta.sync_data",
            "schedule": timedelta(hours=1),
        },
    }
    ```

    `timedelta(seconds=30)` fires at a fixed 30-second interval from the last run, whereas `crontab(minute="*/1")` fires at specific clock times. Choose `timedelta` for "every N seconds/minutes" cadences and `crontab` for "at these clock times" cadences.
  </Tab>

  <Tab value="Solar schedules">
    A `solar` schedule fires relative to a solar event (sunrise, sunset, dawn, dusk) at a given latitude/longitude. Solar schedules recompute the next event time dynamically, accounting for daylight saving time and seasonal change. They require the `ephem` package (`pip install ephem`).

    ```python title="beat_solar.py"
    from celery.schedules import solar

    NYC_LAT = 40.7128
    NYC_LON = -74.0060

    # inside the beat_schedule mapping
    "beat_schedule": {
        "sunset-lights-on": {
            "task": "beat_solar.control_lights",
            "schedule": solar("sunset", NYC_LAT, NYC_LON),
            "kwargs": {"action": "on", "zone": "outdoor"},
        },
        "sunrise-lights-off": {
            "task": "beat_solar.control_lights",
            "schedule": solar("sunrise", NYC_LAT, NYC_LON),
            "kwargs": {"action": "off", "zone": "outdoor"},
        },
        "dawn-data-collection": {
            "task": "beat_solar.start_data_collection",
            "schedule": solar("dawn_astronomical", NYC_LAT, NYC_LON),
        },
        "dusk-data-stop": {
            "task": "beat_solar.stop_data_collection",
            "schedule": solar("dusk_astronomical", NYC_LAT, NYC_LON),
        },
    }
    ```

    Available solar events include `dawn_astronomical`, `dawn_nautical`, `dawn_civil`, `sunrise`, `solar_noon`, `sunset`, `dusk_civil`, `dusk_nautical`, and `dusk_astronomical`.
  </Tab>
</Tabs>

## Dynamic and DB-backed schedules [#dynamic-and-db-backed-schedules]

`app.conf.beat_schedule` is a plain dictionary, so you can add, modify, and remove entries at runtime. This works when Beat runs in the same process as the change (for example, a worker started with the `-B` flag).

```python title="dynamic_periodic_tasks.py"
from datetime import timedelta


def add_periodic_task(name, task, schedule, args=None, kwargs=None):
    """Add a periodic task to the beat schedule at runtime."""
    entry = {"task": task, "schedule": schedule}
    if args:
        entry["args"] = args
    if kwargs:
        entry["kwargs"] = kwargs
    app.conf.beat_schedule[name] = entry


def remove_periodic_task(name):
    """Remove a periodic task from the beat schedule."""
    app.conf.beat_schedule.pop(name, None)


# Add a task that runs every 15 seconds.
add_periodic_task(
    "monitor-api",
    "dynamic_periodic_tasks.monitor_service",
    timedelta(seconds=15),
    args=("api-gateway",),
)

# Modify an existing entry's interval.
app.conf.beat_schedule["initial-heartbeat"]["schedule"] = timedelta(seconds=60)

# Remove an entry.
remove_periodic_task("monitor-api")
```

<Callout type="info">
  **In-memory changes are lost on restart.** The default scheduler keeps state in a local file, so runtime edits to `beat_schedule&#x60; do not survive a Beat restart. For schedules that must persist and be editable without code changes, install &#x2A;*`django-celery-beat`** (a DB-backed scheduler) — it ships in this project's `examples` extra:

  ```bash
  pip install "kubemq-celery[examples]"   # includes django-celery-beat >= 2.5
  ```
</Callout>

## One-shot scheduled tasks [#one-shot-scheduled-tasks]

A common pattern is scheduling a single task to fire once at a future moment — sending a confirmation email, expiring an offer, releasing a held reservation. Use `eta` for an absolute time or `countdown` for a relative delay. These are one-time dispatches, distinct from Beat's recurring entries.

```python title="one_shot_scheduled.py"
from datetime import datetime, timedelta, timezone

now = datetime.now(timezone.utc)

# Fire once, 10 seconds from now, at an absolute time.
send_email.apply_async(
    kwargs={
        "to": "user@example.com",
        "subject": "Your order confirmation",
        "body": "Thank you for your purchase!",
    },
    eta=now + timedelta(seconds=10),
)

# Fire once, after a relative delay.
expire_offer.apply_async(args=("PROMO-2024-SPRING",), countdown=15)

# Fire once, at an absolute release time.
release_reservation.apply_async(args=("RES-12345",), eta=now + timedelta(seconds=20))
```

Both forms use KubeMQ's native `delay_in_seconds` and are subject to the 24-hour cap. For a one-shot task more than 24 hours out, persist the target time in a database and let a Beat task poll for due items.

## Beat in production [#beat-in-production]

In production, run Beat as a **dedicated scheduler container** — separate from your workers and with exactly one replica. The project ships a Docker Compose file that wires up a KubeMQ broker, two workers on different queues, and a single Beat scheduler:

```yaml title="docker-compose.yaml"
services:
  kubemq:
    image: europe-docker.pkg.dev/kubemq/images/kubemq:next
    ports:
      - "50000:50000"  # gRPC
      - "9090:9090"    # Shared HTTP server (REST/health)
      - "8080:8080"    # Management API dashboard
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:9090/health"]
      interval: 10s
      timeout: 5s
      retries: 3
      start_period: 10s

  worker-default:
    build:
      context: ../..
      dockerfile: examples/kubernetes/Dockerfile
    command: celery -A basic_task worker --loglevel=info -Q celery --concurrency=2
    environment:
      - CELERY_BROKER_URL=kubemq://kubemq:50000
      - CELERY_RESULT_BACKEND=kubemq://kubemq:50000
    depends_on:
      kubemq:
        condition: service_healthy
    restart: unless-stopped

  # Dedicated Beat scheduler — exactly one instance.
  beat:
    build:
      context: ../..
      dockerfile: examples/kubernetes/Dockerfile
    command: celery -A basic_task beat --loglevel=info
    environment:
      - CELERY_BROKER_URL=kubemq://kubemq:50000
      - CELERY_RESULT_BACKEND=kubemq://kubemq:50000
    depends_on:
      kubemq:
        condition: service_healthy
    restart: unless-stopped
```

```bash
cd examples/kubernetes
docker compose up -d
```

Workers and Beat share the same broker but run as independent services, so you can scale workers freely while keeping a single Beat process — the rule that avoids duplicate dispatches. For a full Kubernetes deployment with service discovery and KEDA autoscaling, see the Kubernetes deployment guide.

## Next steps [#next-steps]

<Cards>
  <Card title="Canvas Workflows" href="/integrations/celery/how-to/canvas-workflows" description="Compose scheduled tasks into chains, groups, and chords on KubeMQ." />

  <Card title="Configuration" href="/integrations/celery/how-to/configuration" description="message_expiration vs. eta, result_expires, and the full transport option reference." />

  <Card title="Kubernetes Deployment" href="/integrations/celery/how-to/kubernetes" description="Run workers and a dedicated Beat scheduler in-cluster with KEDA autoscaling." />
</Cards>