How to prevent multiple schedules sending on docker compose restart

[rylofu]

I have fastapi application + taskiq with nats jetstream.

result_backend = NATSObjectStoreResultBackend(servers=settings.NATS_URL)

broker = PushBasedJetStreamBroker(
    servers=settings.NATS_URL,
    consumer_config=ConsumerConfig(
        durable_name=f"consumer-{socket.gethostname()}",
        deliver_group="my_group",
    ), ).with_result_backend(result_backend=result_backend)

if settings.MODE == "test":
    broker = InMemoryBroker()


scheduler = TaskiqScheduler(
    broker=broker,
    sources=[LabelScheduleSource(broker)])


def create_lifespan():
    @asynccontextmanager
    async def lifespan(app: FastAPI):
        if not broker.is_worker_process:
            await broker.startup()
        yield
        if not broker.is_worker_process:
            await broker.shutdown()

    return lifespan

running with docker compose

services:
  db_prod:
    container_name: "db_prod"
    image: postgres:16-alpine
    restart: always
    cap_add:
      - SYS_PTRACE
    environment:
      - POSTGRES_USER=admin
      - POSTGRES_PASSWORD=admin
      - POSTGRES_DB=postgres
      - POSTGRES_INITDB_ARGS='--locale=ru_RU.UTF8'
    ports:
      - "5434:5432"
    networks:
      - my_network
    volumes:
      - ./data_prod:/var/lib/postgresql/data

  app:
    stop_signal: SIGINT
    command: pdm run python app/main.py
    container_name: "application"
    build: .
    cap_add:
      - SYS_PTRACE
    ports:
      - "8182:8181"
    networks:
      - my_network
    volumes:
      - ./logs-docker:/shop/logs
      - ./profiles:/shop/profiles


  nats:
    container_name: "nats"
    image: nats:2.11
    command: -js -m 8222
    ports:
      - "4222:4222"
      - "8222:8222"
    volumes:
      - ./nats-data:/data
    networks:
      - my_network


  worker:
    build: .
    command: >
      pdm run taskiq worker 
      app.config_data.tasks_queue.tasks_queue:broker 
      app.user_context.application.use_cases.user_use_cases.user_use_cases
      app.user_context.application.use_cases.order_use_cases.tasks
    networks:
      - my_network
    environment:
      NATS_URL: nats://nats:4222
    depends_on:
      - app
      - nats
    volumes:
      - ./worker-logs:/shop/logs

  scheduler:
    build: .
    command: >
      pdm run taskiq scheduler 
      app.config_data.tasks_queue.tasks_queue:scheduler 
      app.user_context.application.use_cases.order_use_cases.tasks 
      

    networks:
      - my_network
    environment:
      NATS_URL: nats://nats:4222
    depends_on:
      - app
      - worker
      - nats




networks:
  my_network:
    driver: bridge

To make the application run correctly in the broker settings I specify durable_name=f “consumer-{socket.gethostname()}” otherwise I get the error consumer name already in use. I guess that because of this at some point schedules start to run multiple times. During the first run of docker compose everything works as expected, but the next times the number of schedules starts to increase. The docker container prune solves this for one run, but then it happens again. What can I do in this situation?

[kokoc30]

Don't use the container hostname as the durable consumer identity. A new durable_name creates a new persistent reader with its own position in the stream; it doesn't resume the previous container's position. With retained messages, that new reader can receive older tasks again. This can look like schedules were submitted multiple times even when there was only one scheduler publish.

For a shared Taskiq worker pool, one documented alternative is a pull-based broker with a stable durable name:

from taskiq_nats import PullBasedJetStreamBroker

broker = PullBasedJetStreamBroker(
    servers=settings.NATS_URL,
    durable="my-task-workers",  # Stable across restarts and worker replicas.
).with_result_backend(result_backend=result_backend)

Use the same stream/subject configuration and durable name for workers that are meant to share the queue. The taskiq-nats README documents durable for the pull broker, and NATS's worker-pool guide shows multiple workers sharing one consumer.

In your push configuration, setting ConsumerConfig.deliver_group is also not equivalent to supplying the broker's queue option: the implementation passes queue=self.queue or "" to js.subscribe. Push consumers need a consistently configured shared consumer/queue relationship; creating a fresh durable per process sidesteps the naming error by changing the delivery semantics.

Existing hostname-based consumers and stream data remain server-side after changing the Python configuration. Inspect their pending/acknowledged state before retiring obsolete consumers, and decide which backlog the stable consumer should process. Don't use repeated container pruning as the queue-state fix. NATS explains how durable consumers retain their reading position.

Finally, keep only one scheduler publishing a given label schedule unless you explicitly coordinate multiple schedulers. Check whether one scheduled tick produces multiple messages or one message is being delivered repeatedly; those require different fixes. Stable consumers solve the identity/replay issue, but JetStream can still redeliver unacknowledged work, so task side effects should remain idempotent.