distributed-pg-lock 0.1.2

A robust distributed locking system using PostgreSQL for coordinating resource access across distributed systems with automatic heartbeating, timeout management, and stale lock recovery.

Distributed PostgreSQL Lock

A robust distributed locking system using PostgreSQL for coordinating resource access across distributed systems with automatic heartbeating, timeout management, and stale lock recovery.

Features

• Distributed Locking: Coordinate resources across multiple processes/machines
• Heartbeat Monitoring: Built-in keep-alive mechanism prevents premature lock expiration for long operations
• Timeout Handling: Configurable lock duration with automatic release
• Stale Lock Recovery: Self-healing system cleans up abandoned locks
• Context Manager Support: Pythonic with syntax for automatic acquisition/release
• Lazy Initialization: Flexible database connection setup
• Robust Logging: Out-of-the-box diagnostics with custom configuration
• Cross-Platform: Works anywhere with PostgreSQL (Kubernetes, cloud, on-premise)
• High Availability: Database-backed persistence ensures lock integrity
• Customizable Timeouts: Fine-tune lock durations from seconds to hours
• Owner Identification: Track lock ownership across distributed systems
• Graceful Shutdown: Automatic lock release on process termination

Key Use Cases

1. Critical Section Protection: Safeguard shared resources in microservices architectures
2. Distributed Cron Jobs: Prevent duplicate execution across multiple nodes
3. Resource Pool Management: Coordinate access to limited resources (APIs, devices, files)
4. Leader Election: Implement master-node selection in clusters
5. Transaction Sequencing: Enforce ordered processing in queue systems
6. Database Migration Coordination: Safely execute schema changes across replicas

Real-World Use Cases

1. Distributed Script Execution

Ensure critical scripts run on only one pod in Kubernetes clusters:

with lock_manager.get_lock("db_maintenance_script"):
    if lock.is_acquired:
        run_database_maintenance()  # Executes on single pod only

2. Financial Transaction Processing

Prevent duplicate payments in microservices architectures:

def process_payment(txn_id):
    lock = lock_manager.get_lock(f"payment_{txn_id}")
    if lock.is_acquired:
        charge_credit_card(txn_id)  # Exclusive processing

3. Inventory Reservation Systems

Prevent overselling during flash sales:

def reserve_item(item_id):
    with lock_manager.get_lock(f"inventory_{item_id}"):
        if lock.is_acquired and check_stock(item_id) > 0:
            reduce_inventory(item_id)

4. Database Migration Coordination

Safely execute schema changes in clustered environments:

lock = lock_manager.get_lock("schema_migration_v2")
if lock.is_acquired:
    execute_migration()  # Single execution guarantee

5. IoT Command Sequencing

Serialize commands to smart devices:

def send_device_command(device_id, command):
    with lock_manager.get_lock(f"device_{device_id}"):
        if lock.is_acquired:
            device.send(command)  # No command collisions

6. Leader Election

Implement master node selection:

def elect_leader():
    lock = lock_manager.get_lock("cluster_leader")
    if lock.is_acquired:
        start_leader_processes()  # Only one node becomes leader

7. Batch Job Coordination

Prevent duplicate cron job execution:

def run_nightly_report():
    lock = lock_manager.get_lock("nightly_report_job")
    if lock.is_acquired:
        generate_reports()  # Runs once across entire cluster

8. Resource Pool Management

Coordinate limited API quotas:

def call_rate_limited_api():
    with lock_manager.get_lock("api_quota_slot"):
        if lock.is_acquired:
            make_api_call()  # Controlled concurrency

9. ETL Pipeline Synchronization

Ensure ordered data processing:

def process_data_chunk(chunk_id):
    lock = lock_manager.get_lock(f"chunk_{chunk_id}")
    if lock.is_acquired:
        transform_and_load(chunk_id)  # Sequential processing

10. CI/CD Deployment Locking

Prevent concurrent deployments:

def deploy_to_production():
    with lock_manager.get_lock("production_deployment"):
        if lock.is_acquired:
            run_deployment_script()  # Deployment safety

Ideal For

• Financial systems requiring transaction integrity
• Financial transaction processing systems
• E-commerce platforms handling flash sales
• IoT networks managing device communications
• Data pipelines processing critical datasets
• Kubernetes environments coordinating pods
• Microservices architectures needing resource synchronization
• Database maintenance operations
• Distributed cron job management
• Inventory management systems
• CI/CD pipeline coordination

Installation

pip install distributed-pg-lock

Basic Usage

import logging
from distributed_pg_lock import DistributedLockManager, db

# Centralized logging configuration
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Configure database (supports environment variable or direct passing)
# Option 1: Set DB_URL environment variable
export DB_URL='postgresql://user:pass@localhost/mydb'

# Option 2: Pass directly
db.initialize(db_url="postgresql://user:pass@localhost/mydb")

db.create_tables()  # Initialize required tables

lock_manager = DistributedLockManager()

# or

lock_manager = DistributedLockManager(
    lock_timeout_minutes=1,
    owner_id="app-server-1"
)
# Acquire lock using context manager
lock = lock_manager.get_lock("payment_processor")
with lock:
    if lock.is_acquired:
        logger.info("Lock acquired - processing payments")
        # Critical operations here
    else:
        logger.warning("Failed to acquire lock - skipping execution")

Advanced Usage

# High-concurrency processing example
import threading
from concurrent.futures import ThreadPoolExecutor
from distributed_pg_lock import DistributedLockManager, db


# Initialize database connection
db.initialize(db_url=os.getenv("DB_URL"))
db.create_tables()

lock_manager = DistributedLockManager(lock_timeout_minutes=0.5)

def process_task(task_id):
    lock = lock_manager.get_lock(f"task_{task_id}")
    with lock:
        if lock.is_acquired:
            logger.info(f"Processing task {task_id}")
            # Resource-intensive operations
        else:
            logger.debug(f"Lock contention for task {task_id}")

# Process 100 tasks concurrently
with ThreadPoolExecutor(max_workers=20) as executor:
    executor.map(process_task, range(100))
 
# Check system health
status = lock_manager.health_check()
print(f"Active locks: {status['active_locks']}")
print(f"Heartbeat status: {status['heartbeats']}")

## Documentation


### Database Configuration

### Option 1: Explicit Initialization (Recommended)
```python
from distributed_pg_lock import db

db.initialize(
    url="postgresql://user:pass@localhost/dbname",
    pool_size=15,
    max_overflow=25,
    pool_timeout=30,
    echo=True
)

Option 2: Environment Variable

# Set before running application
export DB_URL="postgresql://user:password@localhost:5432/dbname"

# In application
from distributed_pg_lock import db
db.initialize()  # Automatically uses DB_URL environment variable

Connection Error Handling

ValueError: Database URL not provided.
Please pass `db_url` explicitly or set the DB_URL environment variable.

Expected format examples:
  - PostgreSQL: postgresql://user:password@localhost:5432/dbname

Example usage:
  export DB_URL='postgresql://user:pass@localhost:5432/mydb'
  or
  db.initialize(db_url="postgresql://user:pass@localhost/mydb")

Lock Manager Options

# Custom lock configuration
lock_manager = DistributedLockManager(
    lock_timeout_minutes=1.5,        # Auto-release after 90s inactivity
    owner_id="worker-node-42",       # Optional custom owner ID (auto-generated if not provided)
)

Administration

# Force release a lock (admin function)
lock_manager.force_release_lock("stale_resource")


# Health monitoring
status = lock_manager.health_check()
print(f"Active locks: {status['active_locks']}")
print(f"Heartbeat status: {status['heartbeats']}")

Best Practices

1. Always use context managers

with lock_manager.get_lock("resource") as lock:
    if lock.is_acquired:
        # Safe operations

2. Initialize early

# At application startup
db.initialize(db_url=os.getenv("DB_URL"))
db.create_tables()

3. Configure timeouts appropriately

# Longer for resource-intensive operations
lock_manager = DistributedLockManager(lock_timeout_minutes=5)

4. Handle acquisition failures

lock = lock_manager.get_lock("busy_resource")
if not lock.acquire(timeout=10):
    print("Failed to acquire lock - try again later")

5. Use owner IDs for diagnostics

lock_manager = DistributedLockManager(owner_id=f"webapp-") # internally f"webapp-{os.getenv('HOSTNAME')}"

Testing

pip install -e .[test]
pytest tests/ --log-cli-level=INFO
# Run all tests
PYTHONPATH=src pytest tests/ -v --log-cli-level=INFO

# Run specific test file
PYTHONPATH=src pytest tests/test_distributed_lock.py -v

# Run with different DB URL
TEST_DB_URL="postgresql://user:pass@localhost:5432/other_db" pytest tests/
# or
DB_URL="postgresql://user:pass@localhost/testdb" pytest tests/ -v

Performance

See tests/load_test_performance.py for a load testing script that simulates:

• 200 concurrent pods/nodes
• High-contention resource access
• Mixed lock durations (3s–15s, including cases that exceed lock_timeout_minutes)
• Automatic stale lock recovery