Projektowanie systemów

import bisect
import hashlib

class ConsistentHashRing:
    def __init__(self, nodes: list[str] | None = None, virtual_nodes: int = 150):
        self.virtual_nodes = virtual_nodes
        self.ring: dict[int, str] = {}
        self.sorted_keys: list[int] = []
        for node in nodes or []:
            self.add_node(node)

    def _hash(self, key: str) -> int:
        return int(hashlib.sha256(key.encode()).hexdigest(), 16)

    def add_node(self, node: str) -> None:
        for i in range(self.virtual_nodes):
            h = self._hash(f"{node}#{i}")
            self.ring[h] = node
            bisect.insort(self.sorted_keys, h)

    def remove_node(self, node: str) -> None:
        for i in range(self.virtual_nodes):
            h = self._hash(f"{node}#{i}")
            del self.ring[h]
            self.sorted_keys.remove(h)

    def get_node(self, key: str) -> str:
        if not self.ring:
            raise RuntimeError("empty ring")
        h = self._hash(key)
        idx = bisect.bisect(self.sorted_keys, h) % len(self.sorted_keys)
        return self.ring[self.sorted_keys[idx]]

def get_user(user_id: int) -> dict:
    cached = redis_client.get(f"user:{user_id}")
    if cached is not None:
        return json.loads(cached)

    user = db.query_user(user_id)
    redis_client.setex(f"user:{user_id}", 300, json.dumps(user))
    return user

def update_user(user_id: int, data: dict) -> None:
    db.update_user(user_id, data)
    redis_client.delete(f"user:{user_id}")  # invalidate, nie update

def quorum_write(replicas: list["Node"], key: str, value: "VersionedValue", w: int) -> bool:
    acks = 0
    for node in replicas:
        try:
            node.write(key, value)
            acks += 1
        except NodeUnavailable:
            continue
    return acks >= w

def quorum_read(replicas: list["Node"], key: str, r: int) -> "VersionedValue":
    responses = []
    for node in replicas:
        try:
            responses.append(node.read(key))
        except NodeUnavailable:
            continue
    if len(responses) < r:
        raise QuorumNotReached(f"got {len(responses)}, need {r}")
    return max(responses, key=lambda v: v.version)

# W + R > N gwarantuje, że każdy odczyt widzi co najmniej jedną replikę z najnowszym zapisem

import hashlib

def shard_for_user(user_id: int, shard_count: int) -> int:
    # hash-based: hashlib zamiast wbudowanego hash() (Python losuje hash() dla str/bytes per proces)
    digest = hashlib.sha256(str(user_id).encode()).digest()
    return int.from_bytes(digest[:8], "big") % shard_count

def shard_for_order(order_date: str, boundaries: list[str]) -> int:
    # range-based: zakresy dat jako granice shardów, ułatwia range scan po dacie
    for shard_index, boundary in enumerate(boundaries):
        if order_date < boundary:
            return shard_index
    return len(boundaries)

import time
from enum import Enum, auto

class State(Enum):
    CLOSED = auto()
    OPEN = auto()
    HALF_OPEN = auto()

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, reset_timeout: float = 30.0):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.state = State.CLOSED
        self.opened_at: float | None = None

    def call(self, fn, *args, **kwargs):
        if self.state is State.OPEN:
            if time.monotonic() - self.opened_at < self.reset_timeout:
                raise RuntimeError("circuit open, failing fast")
            self.state = State.HALF_OPEN

        try:
            result = fn(*args, **kwargs)
        except Exception:
            self.failures += 1
            if self.failures >= self.failure_threshold:
                self.state = State.OPEN
                self.opened_at = time.monotonic()
            raise
        else:
            self.failures = 0
            self.state = State.CLOSED
            return result

import random
import time

def retry_with_backoff(fn, max_attempts: int = 5, base_delay: float = 0.5, max_delay: float = 30.0):
    for attempt in range(max_attempts):
        try:
            return fn()
        except TransientError:
            if attempt == max_attempts - 1:
                raise
            capped = min(max_delay, base_delay * 2 ** attempt)
            sleep_time = random.uniform(0, capped)  # full jitter
            time.sleep(sleep_time)

def charge_payment(idempotency_key: str, amount: int, currency: str) -> dict:
    reserved = db.reserve_idempotency_key(idempotency_key)  # atomowy INSERT ... ON CONFLICT (idempotency_key) DO NOTHING RETURNING id
    if not reserved:
        return db.find_payment_by_idempotency_key(idempotency_key)  # replay: klucz już zarezerwowany przez oryginalny request

    payment = payment_gateway.charge(amount, currency)
    db.save_payment(idempotency_key, payment)
    return payment