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?