Projektowanie ogranicznika przepustowości i kanału Twittera

Dlaczego ograniczanie liczby żądań jest niezbędne

# Rate limiting scenarios
use_cases = [
    ('API authentication endpoint', '5 attempts per 15 min per IP', 'Brute-force protection'),
    ('Public search API',           '100 requests per minute per key', 'Fair use enforcement'),
    ('Email sending',               '50 emails per hour per user', 'Spam prevention'),
    ('Payment processing',          '10 transactions per second per account', 'Fraud prevention'),
    ('File upload',                 '5 uploads per minute per user', 'Resource quota'),
    ('Notification service',        '1000 pushes per second globally', 'Cost control'),
]
print(f'{'Endpoint/Feature':35s} {'Limit':40s} {'Reason'}')
print('-'*95)
for endpoint, limit, reason in use_cases:
    print(f'{endpoint:35s} {limit:40s} {reason}')

Algorytm ograniczania liczby żądań 1: Token Bucket

import time

class TokenBucket:
    def __init__(self, capacity, refill_rate):
        self.capacity = capacity       # max tokens (burst size)
        self.refill_rate = refill_rate # tokens added per second
        self.tokens = capacity         # start full
        self.last_refill = time.time()

    def allow(self):
        now = time.time()
        elapsed = now - self.last_refill
        # Refill tokens based on elapsed time
        self.tokens = min(self.capacity,
                          self.tokens + elapsed * self.refill_rate)
        self.last_refill = now
        if self.tokens >= 1:
            self.tokens -= 1
            return True    # request allowed
        return False       # rate limited

bucket = TokenBucket(capacity=5, refill_rate=2)  # 2 tokens/sec, burst=5
for i in range(8):
    allowed = bucket.allow()
    print(f'Request {i+1}: {"ALLOWED" if allowed else "REJECTED"} (tokens={bucket.tokens:.1f})')
    time.sleep(0.1)  # 0.1s between requests

Algorytm ograniczania liczby żądań 2: log przesuwnego okna

import time
from collections import deque

class SlidingWindowLog:
    def __init__(self, limit, window_seconds):
        self.limit = limit
        self.window = window_seconds
        self.logs = {}     # user_id -> deque of timestamps

    def allow(self, user_id):
        now = time.time()
        if user_id not in self.logs:
            self.logs[user_id] = deque()
        log = self.logs[user_id]
        window_start = now - self.window
        # Remove expired timestamps
        while log and log[0] <= window_start:
            log.popleft()
        # Check limit
        if len(log) < self.limit:
            log.append(now)
            return True
        return False

limiter = SlidingWindowLog(limit=3, window_seconds=10)
for i in range(5):
    allowed = limiter.allow('user123')
    print(f'Request {i+1}: {"ALLOWED" if allowed else "REJECTED"}')
    time.sleep(0.5)

Algorytm ograniczania liczby żądań 3: licznik przesuwnego okna

import time
import math

class SlidingWindowCounter:
    def __init__(self, limit, window_seconds=60):
        self.limit = limit
        self.window = window_seconds
        self.buckets = {}   # user_id -> {prev_count, curr_count, curr_window_start}

    def allow(self, user_id):
        now = time.time()
        window_start = int(now // self.window) * self.window

        if user_id not in self.buckets or self.buckets[user_id]['window'] < window_start - self.window:
            self.buckets[user_id] = {'prev': 0, 'curr': 0, 'window': window_start}
        elif self.buckets[user_id]['window'] < window_start:
            self.buckets[user_id] = {'prev': self.buckets[user_id]['curr'], 'curr': 0, 'window': window_start}

        b = self.buckets[user_id]
        fraction = (now - window_start) / self.window
        estimated = b['prev'] * (1 - fraction) + b['curr']

        if estimated < self.limit:
            b['curr'] += 1
            return True
        return False

limiter = SlidingWindowCounter(limit=5, window_seconds=10)
for i in range(7):
    print(f'Request {i+1}: {"OK" if limiter.allow("user1") else "RATE LIMITED"}')
    time.sleep(0.3)

Rozproszone ograniczanie liczby żądań za pomocą Redis

# Distributed rate limiting with Redis (pseudocode / simulation)

# Fixed window counter using Redis INCR + EXPIRE
def redis_fixed_window(redis_client, user_id, limit, window_sec):
    key = f'rl:{user_id}:{int(time.time() // window_sec)}'
    count = redis_client.incr(key)          # atomic increment
    if count == 1:
        redis_client.expire(key, window_sec)  # set TTL on first request
    return count <= limit

# Sliding window with sorted set
def redis_sliding_window(redis_client, user_id, limit, window_sec):
    now = time.time()
    key = f'rl:{user_id}'
    # Remove old entries, count recent, add current
    # Atomic with Lua: multi-step operation
    lua_script = '''
    local key = KEYS[1]
    local now = ARGV[1]
    local window = ARGV[2]
    local limit = ARGV[3]
    redis.call('ZREMRANGEBYSCORE', key, '-inf', now - window)
    local count = redis.call('ZCARD', key)
    if count < tonumber(limit) then
        redis.call('ZADD', key, now, now)
        redis.call('EXPIRE', key, window)
        return 1  -- allowed
    end
    return 0  -- rejected
    '''
    print('Redis Lua script ensures atomicity across ZREM + ZCARD + ZADD')

Projektowanie kanału Twittera: wymagania

# Twitter feed requirements and estimates
reqs = {
    'Functional': [
        'Post tweet (text, image, video)',
        'Follow/unfollow users',
        'View home feed (tweets from followees, newest first)',
        'View user timeline (all tweets by one user)',
        'Like and retweet',
        'Search tweets (basic keyword)',
    ],
    'Non-functional': [
        '300M DAU, 500M tweets/day => 5800 writes/sec',
        '100:1 read:write => 580K feed reads/sec',
        'Feed load < 2 seconds (p95)',
        '99.99% availability',
        'Tweets retained indefinitely (tweets never deleted by default)',
    ],
    'Estimates': [
        'Storage: 500M tweets * 300B = 150 GB/day, 54 TB/year',
        'Media: separate object store (S3), CDN-served',
        'Feed cache: 300M users * top-100-tweets * 100B = 3 TB (hot feeds in Redis)',
    ],
}
for category, items in reqs.items():
    print(f'{category}:')
    for item in items: print(f'  - {item}')
    print()

Fan-out podczas zapisu: wstępnie obliczone kanały

# Fan-out on write (push model)
fan_out_steps = [
    '1. User posts tweet => write to tweets table (source of truth)',
    '2. Publish event to message queue (Kafka topic: tweet-created)',
    '3. Fan-out workers consume from queue:',
    '   a. Fetch list of followers from follows table',
    '   b. For each follower: LPUSH feed:{follower_id} tweet_id',
    '   c. Trim feed to last 1000 tweets: LTRIM feed:{follower_id} 0 999',
    '4. Feed read: LRANGE feed:{user_id} 0 99 => hydrate tweet_ids => response',
]
for step in fan_out_steps:
    print(step)

print('\nPros:')
print('  - Feed reads are O(1): just read from Redis list')
print('  - Feed is always sorted by recency automatically')
print('\nCons:')
print('  - Celebrities with 100M followers => 100M Redis writes per tweet')
print('  - Fan-out lag: followers may see tweet 10-30 seconds late at peak')
print('  - Inactive users waste Redis storage for precomputed feeds')

Hybrydowy fan-out: rozwiązanie problemu celebrytów

# Hybrid fan-out implementation sketch
CELEBRITY_THRESHOLD = 1_000_000   # followers > 1M => celebrity

def on_post_tweet(user_id, tweet_id, follower_count):
    if follower_count <= CELEBRITY_THRESHOLD:
        # Fan-out to all followers (async via Kafka)
        print(f'User {user_id}: fan-out tweet {tweet_id} to {follower_count} followers')
        # => queue to fan-out workers
    else:
        print(f'Celebrity {user_id}: tweet {tweet_id} stored in timeline only')
        # => only write to tweets table + user timeline
        # => followers get it on demand when reading feed

def get_home_feed(user_id, followees):
    # 1. Get precomputed feed (fan-out on write tweets)
    precomputed = f'LRANGE feed:{user_id} 0 499'   # up to 500 tweets

    # 2. Find celebrity followees
    celebrity_followees = [u for u in followees if is_celebrity(u)]

    # 3. Fetch recent tweets from celebrities (fan-out on read)
    celebrity_tweets = []
    for celeb in celebrity_followees:
        tweets = f'GET tweets WHERE user_id={celeb} ORDER BY created_at DESC LIMIT 20'
        celebrity_tweets.extend(tweets)

    # 4. Merge and sort by recency
    combined = merge_and_sort(precomputed, celebrity_tweets)
    return combined[:100]

print('on_post_tweet for regular user:')
on_post_tweet('user123', 'tweet_abc', 500)
print('on_post_tweet for celebrity:')
on_post_tweet('celebrity456', 'tweet_xyz', 50_000_000)

Kompletna architektura kanału Twittera

# Twitter architecture summary
architecture = '''
[User] --> [API Gateway + Load Balancer]
                |
    +-----------+-----------+
    |           |           |
[Tweet Svc] [Feed Svc]  [Follow Svc]
    |           |           |
[Cassandra] [Redis Feeds] [Graph DB]
    |           |
[Kafka] <-- [Fan-out
    |        Workers]
[S3 + CDN]  (tweet_ids
(media)      => follower
             feed lists)

Key design choices:
- Tweets stored in Cassandra (PRIMARY KEY (user_id, created_at))
- Feed stored in Redis as list of tweet_ids per user (LPUSH/LTRIM/LRANGE)
- Fan-out via Kafka + workers (decoupled, retryable, scalable)
- Hybrid: regular users = push; celebrities = pull-on-read
- Hydration: tweet_ids -> full tweet objects via Cassandra read
'''
print(architecture)

Nagłówki i odpowiedzi błędów ogranicznika liczby żądań

# Rate limit response headers
def build_rate_limit_headers(limit, remaining, reset_timestamp, retry_after=None):
    headers = {
        'X-RateLimit-Limit': str(limit),
        'X-RateLimit-Remaining': str(max(0, remaining)),
        'X-RateLimit-Reset': str(int(reset_timestamp)),
    }
    if retry_after is not None:
        headers['Retry-After'] = str(retry_after)
    return headers

import time

# Simulated response for allowed request
headers = build_rate_limit_headers(
    limit=100,
    remaining=73,
    reset_timestamp=time.time() + 45
)
print('Allowed request headers:')
for k, v in headers.items():
    print(f'  {k}: {v}')

# Rate limited response
headers_429 = build_rate_limit_headers(
    limit=100,
    remaining=0,
    reset_timestamp=time.time() + 30,
    retry_after=30
)
print('\n429 Too Many Requests headers:')
for k, v in headers_429.items():
    print(f'  {k}: {v}')

Porównanie algorytmów ograniczania częstotliwości żądań

# Algorithm comparison matrix
comparison = [
    ('Token Bucket',          'Allows bursts',     'O(1)',         'Most APIs, default choice'),
    ('Leaky Bucket',          'Smooth output rate', 'O(1)',         'Traffic shaping, message queues'),
    ('Fixed Window Counter',  'Very simple',        'O(1)',         'Low-traffic, approximate OK'),
    ('Sliding Window Log',    'Most accurate',      'O(requests)', 'High-accuracy, low traffic'),
    ('Sliding Window Counter','Approximate+fast',   'O(1)',         'High-traffic, Cloudflare-style'),
]
print(f'{'Algorithm':30s} {'Burst Handling':20s} {'Memory':15s} {'Use Case'}')
print('-'*85)
for name, burst, mem, use in comparison:
    print(f'{name:30s} {burst:20s} {mem:15s} {use}')

Szybki test

Podsumowanie lekcji