Distributed Systems Principles and Tradeoffs Questions

import time
import random
import uuid
import functools
from typing import Iterable, Callable, Optional

def retry(
    max_attempts: int = 3,
    base_delay_seconds: float = 0.5,
    backoff_factor: float = 2.0,
    full_jitter: bool = True,
    retry_on_exceptions: Iterable[Exception] = (Exception,),
    propagate_idempotency_key: Optional[str] = None,  # header name to inject, e.g. "Idempotency-Key"
    max_delay_seconds: float = 30.0,
):
    def decorator(fn: Callable):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            attempts = 0
            last_exc = None

            # helper to inject idempotency key into kwargs['headers'] if requested
            def maybe_inject_idempotency():
                if not propagate_idempotency_key:
                    return
                headers = kwargs.get("headers")
                if headers is None:
                    headers = {}
                # do not overwrite existing key
                if propagate_idempotency_key not in headers:
                    headers = dict(headers)  # copy to avoid mutating caller's dict
                    headers[propagate_idempotency_key] = str(uuid.uuid4())
                    kwargs["headers"] = headers

            while attempts < max_attempts:
                maybe_inject_idempotency()
                try:
                    return fn(*args, **kwargs)
                except tuple(retry_on_exceptions) as e:
                    last_exc = e
                    attempts += 1
                    if attempts >= max_attempts:
                        break
                    # compute exponential backoff cap
                    cap = min(base_delay_seconds * (backoff_factor ** (attempts - 1)), max_delay_seconds)
                    if full_jitter:
                        sleep_for = random.uniform(0, cap)
                    else:
                        sleep_for = cap
                    # avoid busy-wait by sleeping
                    time.sleep(sleep_for)
                except Exception:
                    # immediately raise non-retryable exceptions
                    raise
            # exhausted retries -> re-raise last retryable exception
            if last_exc is not None:
                raise last_exc
            # unlikely: no exception captured
            return fn(*args, **kwargs)
        return wrapper
    return decorator

import requests

@retry(max_attempts=5, base_delay_seconds=0.2, backoff_factor=2, full_jitter=True,
       retry_on_exceptions=(requests.exceptions.Timeout, requests.exceptions.ConnectionError),
       propagate_idempotency_key="Idempotency-Key")
def fetch(url, headers=None):
    return requests.post(url, headers=headers, timeout=2)

package ratelimit

import (
	"context"
	"sync"
	"time"
)

// TokenBucket is a thread-safe token bucket rate limiter.
type TokenBucket struct {
	rate     float64       // tokens per second
	capacity float64       // max tokens
	mu       sync.Mutex
	tokens   float64
	last     time.Time
}

// NewTokenBucket creates a new bucket with refill rate r tokens/sec and burst capacity.
func NewTokenBucket(r float64, burst int) *TokenBucket {
	if r <= 0 {
		panic("rate must be > 0")
	}
	return &TokenBucket{
		rate:     r,
		capacity: float64(burst),
		tokens:   float64(burst),
		last:     time.Now(),
	}
}

// refill updates the token count based on elapsed time.
func (b *TokenBucket) refill(now time.Time) {
	elapsed := now.Sub(b.last).Seconds()
	if elapsed <= 0 {
		return
	}
	b.tokens += elapsed * b.rate
	if b.tokens > b.capacity {
		b.tokens = b.capacity
	}
	b.last = now
}

// Allow tries to take n tokens, returns true if successful.
func (b *TokenBucket) Allow(n int) bool {
	if n <= 0 {
		return true
	}
	want := float64(n)

	b.mu.Lock()
	defer b.mu.Unlock()

	now := time.Now()
	b.refill(now)

	if b.tokens >= want {
		b.tokens -= want
		return true
	}
	return false
}

// Wait blocks until n tokens are available or ctx is canceled.
func (b *TokenBucket) Wait(ctx context.Context, n int) error {
	if n <= 0 {
		return nil
	}
	want := float64(n)

	for {
		b.mu.Lock()
		now := time.Now()
		b.refill(now)

		if b.tokens >= want {
			b.tokens -= want
			b.mu.Unlock()
			return nil
		}

		// compute time until enough tokens accumulate
		needed := want - b.tokens
		seconds := needed / b.rate
		if seconds < 0 {
			seconds = 0
		}
		deadline := now.Add(time.Duration(seconds * float64(time.Second)))
		b.mu.Unlock()

		// Wait either for timer or ctx cancellation
		timer := time.NewTimer(time.Until(deadline))
		select {
		case <-timer.C:
			// loop and try again
		case <-ctx.Done():
			if !timer.Stop() {
				<-timer.C
			}
			return ctx.Err()
		}
	}
}

b := NewTokenBucket(5, 10) // 5 tokens/sec, burst 10
if b.Allow(3) {
    // proceed
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
if err := b.Wait(ctx, 8); err != nil {
    // timed out or canceled
}