claude-force: Skill for Claude Code

High Availability Patterns

Production-ready HA patterns for cryptocurrency trading systems with 99.99% uptime.

Active-Passive Architecture with Redis Leader Election

import redis
import asyncio
import time
import uuid
from typing import Optional

class LeaderElection:
    """Redis-based leader election for active-passive HA"""

    def __init__(
        self,
        redis_client: redis.Redis,
        service_name: str,
        ttl_seconds: int = 10,
        instance_id: Optional[str] = None
    ):
        self.redis = redis_client
        self.service_name = service_name
        self.ttl = ttl_seconds
        self.instance_id = instance_id or str(uuid.uuid4())
        self.is_leader = False
        self.lock_key = f"leader:{service_name}"

    async def run_leader_election(self):
        """Continuous leader election loop"""
        while True:
            try:
                # Try to acquire leadership
                acquired = self.redis.set(
                    self.lock_key,
                    self.instance_id,
                    nx=True,  # Only set if not exists
                    ex=self.ttl  # Expire after TTL
                )

                if acquired:
                    if not self.is_leader:
                        logger.info(f"Instance {self.instance_id} became LEADER")
                        self.is_leader = True
                        await self._on_become_leader()

                    # Renew leadership
                    await self._renew_leadership()

                else:
                    # Check if we were leader before
                    if self.is_leader:
                        logger.warning(f"Instance {self.instance_id} lost leadership")
                        self.is_leader = False
                        await self._on_lose_leadership()

                # Sleep for half the TTL before renewing
                await asyncio.sleep(self.ttl / 2)

            except redis.RedisError as e:
                logger.error(f"Leader election error: {e}")
                self.is_leader = False
                await asyncio.sleep(1)

    async def _renew_leadership(self):
        """Renew leadership lock"""
        try:
            # Only renew if we still hold the lock
            current_leader = self.redis.get(self.lock_key)
            if current_leader and current_leader.decode() == self.instance_id:
                self.redis.expire(self.lock_key, self.ttl)
            else:
                self.is_leader = False
                logger.warning("Lost leadership during renewal")

        except redis.RedisError as e:
            logger.error(f"Leadership renewal failed: {e}")
            self.is_leader = False

    async def _on_become_leader(self):
        """Hook called when instance becomes leader"""
        # Perform state reconciliation
        await self._reconcile_state()

        # Start active trading
        await self._start_trading_engine()

        # Send notification
        logger.info("Transitioned to ACTIVE state")

    async def _on_lose_leadership(self):
        """Hook called when instance loses leadership"""
        # Stop trading immediately
        await self._stop_trading_engine()

        # Flush pending orders
        await self._flush_pending_orders()

        logger.info("Transitioned to PASSIVE state")