Stale Lock Cleanup Process

Relevant source files

The following files were used as context for generating this wiki page:

• README.md
• src/meltano_state_backend_clickhouse/backend.py
• tests/test_backend.py
• pyproject.toml
• src/meltano_state_backend_clickhouse/init.py

The Stale Lock Cleanup Process is a critical mechanism within the ClickHouse state backend for Meltano designed to ensure system liveness in an environment that lacks native transactional row-level locking. Since ClickHouse does not support standard primary key violation constraints for locking, the backend implements a "best-effort" advisory lock using an insert-then-verify pattern.

The cleanup process addresses scenarios where a lock holder might crash or fail to release a lock, preventing other processes from accessing the state for a specific state_id. It employs a dual-layered approach: active synchronous cleanup during lock acquisition attempts and passive background cleanup via ClickHouse's Time-To-Live (TTL) mechanism.

Sources: src/meltano_state_backend_clickhouse/backend.py:11-20, README.md:33-40

Lock Architecture and Storage

Locks are managed in a dedicated table, distinct from the main state table. This table is structured to handle high-concurrency environments using the ReplacingMergeTree engine.

Lock Table Schema

The lock table is defined with specific fields to track ownership and duration:

Field	Type	Description
state_id	String	The unique identifier for the Meltano pipeline state.
lock_id	String	A unique UUID generated by the process attempting to acquire the lock.
locked_at	DateTime	The timestamp when the lock record was inserted (defaults to now()).

Sources: src/meltano_state_backend_clickhouse/backend.py:246-252

Storage Engine and TTL

The lock table uses the ReplacingMergeTree(locked_at) engine, which collapses rows with the same state_id during background merges. Crucially, it implements a native ClickHouse TTL to provide a "backstop" for stale locks.

Cleanup Mechanisms

The system utilizes two primary methods to remove stale locks, ensuring that a dead lock-holder does not permanently block a state_id.

Active Synchronous Cleanup

When a process attempts to call acquire_lock, it first checks if a lock is currently held. If a holder is found, the backend invokes _cleanup_stale_locks() to proactively delete any rows where locked_at is older than the defined STALE_LOCK_SECONDS. This ensures that reclamation happens immediately when contention occurs, rather than waiting for ClickHouse's asynchronous background tasks.

Sources: src/meltano_state_backend_clickhouse/backend.py:363-368, src/meltano_state_backend_clickhouse/backend.py:388-393

Passive TTL Backstop

The table is created with a TTL clause: TTL locked_at + INTERVAL 300 SECOND. This serves as a secondary cleanup layer where ClickHouse itself will eventually reap old rows during background merges even if no active cleanup is triggered by a new process.

Sources: src/meltano_state_backend_clickhouse/backend.py:255, src/meltano_state_backend_clickhouse/backend.py:53

Lock Acquisition Flow

The cleanup process is tightly integrated into the lock acquisition lifecycle. To prevent race conditions inherent in ClickHouse's eventual consistency, the backend uses a "settle-then-decide" strategy.

Configuration and Constants

The cleanup behavior is governed by several internal constants that define the "stale" threshold and the settling window.

Constant	Value	Description
STALE_LOCK_SECONDS	300 (5 mins)	Threshold after which a lock is considered dead and eligible for cleanup.
LOCK_SETTLE_SECONDS	1.0	Duration to wait after insertion to ensure all concurrent candidates are visible.
LOCK_TIMEOUT_SECONDS	30	Total time a process will attempt to acquire and clean up locks before failing.

Sources: src/meltano_state_backend_clickhouse/backend.py:52-59

Implementation Details

The core cleanup logic is implemented in the _cleanup_stale_locks method of the ClickhouseStateStoreManager class.

def _cleanup_stale_locks(self) -> None:
    """Remove locks older than STALE_LOCK_SECONDS."""
    self.client.command(
        f"DELETE FROM {self.lock_table} "
        f"WHERE locked_at < now() - {STALE_LOCK_SECONDS}",
    )

Sources: src/meltano_state_backend_clickhouse/backend.py:363-368

This active cleanup is validated in tests by manually inserting "dead" rows with timestamps in the past and verifying that subsequent acquire_lock calls succeed in reclaiming the slot.

Sources: tests/test_backend.py:114-126

Summary

The Stale Lock Cleanup Process provides a robust workaround for ClickHouse's lack of traditional locking primitives. By combining active DELETE operations during lock acquisition with native ClickHouse TTL policies, the system ensures that pipeline state remains accessible even in the event of process failures. The integration of a settling period further guarantees that cleanup and re-acquisition remain safe under high concurrency.