Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/Indroic/HexCore/llms.txt

Use this file to discover all available pages before exploring further.

When you run multiple replicas of your application — for example in a Kubernetes deployment — every instance of DynamicScheduler will tick at the same time and could enqueue the same cron job twice. HexCore’s ILockProvider interface solves this by letting you acquire an atomic, TTL-backed distributed lock before each job is submitted. This guide explains both built-in implementations — RedisLockProvider and PostgresLockProvider — and shows how to wire them into DynamicScheduler.

​
The ILockProvider Interface

Both providers implement the same two-method contract from hexcore.domain.cqrs.cron: 
class ILockProvider(abc.ABC):
    @abc.abstractmethod
    async def acquire_lock(self, lock_key: str, ttl_seconds: int) -> bool:
        """
        Try to acquire an exclusive lock.
        Returns True on success, False if the lock is already held.
        """

    @abc.abstractmethod
    async def release_lock(self, lock_key: str) -> None:
        """Release the lock explicitly before the TTL expires."""
​
acquire_lock
async (lock_key: str, ttl_seconds: int) -> bool
Atomically tries to take the lock. Returns True if this replica now owns it, False if another replica got there first. The ttl_seconds acts as a safety net — the lock expires automatically even if release_lock is never called (e.g. on a crash).
​
release_lock
async (lock_key: str) -> None
Deletes the lock immediately so the next scheduled tick can proceed without waiting for the TTL. Safe to call even if the lock no longer exists.

​
RedisLockProvider

RedisLockProvider uses a single Redis command — SET key "locked" NX EX ttl — which is atomic on the Redis server side. It requires a redis.asyncio.Redis client instance: 
from hexcore.infrastructure.cqrs.redis_lock import RedisLockProvider
import redis.asyncio as redis

redis_client = redis.from_url("redis://localhost:6379/0")
lock_provider = RedisLockProvider(redis_client=redis_client)

​
How it works internally

# SET key "locked" NX EX <ttl>
# Returns the string "OK" (truthy) if the key was set, None if it already existed.
result = await self.redis.set(lock_key, "locked", nx=True, ex=ttl_seconds)
return bool(result)
  • NX — only write if the key does not already exist (atomic compare-and-set)
  • EX ttl_seconds — auto-expire ensures no deadlock if a replica crashes before calling release_lock

​
Full example

src/infrastructure/scheduler.py
import redis.asyncio as redis
from hexcore.infrastructure.cqrs.redis_lock import RedisLockProvider
from hexcore.application.cqrs.scheduler import DynamicScheduler
from src.infrastructure.cron_repository import MyCronRepository
from src.infrastructure.task_queues import enqueuer   # your ITaskEnqueuer

redis_client = redis.from_url("redis://localhost:6379/0")
lock_provider = RedisLockProvider(redis_client=redis_client)

scheduler = DynamicScheduler(
    repository=MyCronRepository(),
    enqueuer=enqueuer,
    lock_provider=lock_provider,
    tick_interval_seconds=60,
)

# In FastAPI lifespan or a background task:
# asyncio.create_task(scheduler.start())

​
PostgresLockProvider

Use PostgresLockProvider when your stack already includes PostgreSQL and you want zero additional infrastructure. No Redis required — locks are stored in a dedicated table that HexCore creates for you.
PostgresLockProvider uses an asyncpg pool or connection and relies on a hexcore_cron_locks table with an ON CONFLICT DO UPDATE WHERE expires_at < NOW() upsert pattern — fully atomic in a single SQL statement: 
from hexcore.infrastructure.cqrs.postgres_lock import PostgresLockProvider
import asyncpg

pool = await asyncpg.create_pool("postgresql://user:pass@localhost/mydb")

lock_provider = PostgresLockProvider(pool=pool)

# Create the locks table (idempotent — safe to call on every startup):
await lock_provider.setup()

​
setup() — create the locks table

await lock_provider.setup()
This runs the following DDL (only once — it is idempotent): 
CREATE TABLE IF NOT EXISTS hexcore_cron_locks (
    lock_key TEXT PRIMARY KEY,
    expires_at TIMESTAMP WITH TIME ZONE NOT NULL
);
You must call await lock_provider.setup() at least once before the scheduler starts. If the table does not exist, acquire_lock will fail and log an error, causing every job to be skipped on that tick.

​
Custom table name

If you need the lock table in a specific schema or want a different name, pass table_name: 
lock_provider = PostgresLockProvider(pool=pool, table_name="myapp.cron_locks")
await lock_provider.setup()

​
How the upsert lock works

INSERT INTO hexcore_cron_locks (lock_key, expires_at)
VALUES ($1, NOW() + ($2 || ' seconds')::interval)
ON CONFLICT (lock_key) DO UPDATE
    SET expires_at = NOW() + ($2 || ' seconds')::interval
    WHERE hexcore_cron_locks.expires_at < NOW()
RETURNING lock_key;
  • If the row does not exist → insert and return the row → lock acquired
  • If the row exists and has expired → update and return the row → lock acquired
  • If the row exists and has not expired → WHERE clause blocks the update → no row returned → lock not acquired

​
Full example

src/infrastructure/scheduler.py
import asyncpg
from hexcore.infrastructure.cqrs.postgres_lock import PostgresLockProvider
from hexcore.application.cqrs.scheduler import DynamicScheduler
from src.infrastructure.cron_repository import MyCronRepository
from src.infrastructure.task_queues import enqueuer

async def start_scheduler():
    pool = await asyncpg.create_pool("postgresql://user:pass@localhost/mydb")

    lock_provider = PostgresLockProvider(pool=pool)
    await lock_provider.setup()   # create table if not exists

    scheduler = DynamicScheduler(
        repository=MyCronRepository(),
        enqueuer=enqueuer,
        lock_provider=lock_provider,
        tick_interval_seconds=60,
    )

    await scheduler.start()   # infinite loop

​
Choosing a Provider

ConcernRedisLockProviderPostgresLockProvider
InfrastructureRequires RedisPostgreSQL only
Atomicity mechanismSET NX EXSQL ON CONFLICT ... WHERE
TTL enforcementNative Redis TTLChecked on next acquire
Crash safetyAuto-expires via TTLAuto-expires via WHERE expires_at < NOW()
Setup requiredNoneawait lock_provider.setup()

​
Injecting a Lock Provider into DynamicScheduler

Both providers share the same injection point: 
from hexcore.application.cqrs.scheduler import DynamicScheduler

scheduler = DynamicScheduler(
    repository=repo,
    enqueuer=enqueuer,
    lock_provider=lock_provider,   # RedisLockProvider or PostgresLockProvider
    tick_interval_seconds=60,
)
When lock_provider is set, the scheduler calls acquire_lock before enqueuing each job. If the lock cannot be acquired, the job is silently skipped on that tick — another replica already claimed it. The lock is released automatically after the job is enqueued (or on error).
Set ttl_seconds to at least twice the tick_interval_seconds of your scheduler. This guarantees that a lock taken in one tick expires before the next tick begins, even if the release call fails.

Build docs developers (and LLMs) love

Get started for freeTalk to us