Homework #1: Coordination Under Uncertainty

EE 547: Fall 2026

ImportantAssignment Details

Assigned: 25 August
Due: Monday, 7 September at 23:59

Gradescope: Homework 1 | How to Submit

WarningRequirements
  • Python 3.11+ required
  • Use only Python standard library modules unless explicitly permitted
  • pytest to run the provided test_interfaces.py (pip install pytest)

Overview

The three problems cover coordination, network uncertainty, and durability. None of them needs Docker, servers, or cloud services.

Getting Started

Download the starter code: hw1-starter.zip

unzip hw1-starter.zip
cd hw1-starter

Problem 1: Distributed Merge

Two workers each hold a portion of an unsorted list of integers and must merge them into a single sorted output, communicating only through file-based message passing.

Context

In distributed systems, processes coordinate through explicit messages rather than shared state. Even simple operations like merging sorted data require careful protocol design when participants can only communicate by passing messages.

Setup

Worker A holds an unsorted list of approximately 4000 integers. Worker B holds an unsorted list of approximately 3000 integers. The lists are disjoint (no duplicates across workers) but their ranges may overlap.

Both workers append to the same output file—the protocol must ensure values arrive in sorted order.

Workers are reconstructed each step. The dataset is provided only at first construction. Between steps, only the state file persists. The state file may hold at most 5000 integers (enforced by the monitor).

Workers communicate through two “dropbox” files:

  • A_to_B.msg: Worker A writes here, Worker B reads here
  • B_to_A.msg: Worker B writes here, Worker A reads here

Each worker can only write to its outbox and read from its inbox.

Message Format

Messages are JSON objects with the following structure:

{
    "msg_type": str,    # Maximum 5 characters (e.g., "DATA", "REQ", "DONE", "RANGE")
    "values": list[int] # Maximum 10 integers, each <= 2^31 - 1
}

You define the protocol—choose message types that enable coordination. Examples:

  • "RANGE" with [min, max, count] to communicate data bounds
  • "DATA" with actual values to transfer
  • "REQ" to request values in a specific range
  • "DONE" to signal completion

Worker Implementation

The starter’s merge_worker.py defines Message, WorkerStats, and a single MergeWorker class that works for both workers (distinguished by worker_id). Implement _initial_state() and step(). The constructor, the state file methods, and the message and output helpers are provided. _read_message() returns the inbox message, or None. _clear_inbox() removes the inbox message. The output file holds one integer per line.

Code: merge_worker.py 
"""Worker for the two-worker distributed merge."""

import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional


@dataclass
class Message:
    """Message format for worker communication."""
    msg_type: str  # Max 5 chars
    values: list[int] = field(default_factory=list)  # Max 10 integers


@dataclass
class WorkerStats:
    """Statistics about work performed."""
    comparisons: int = 0
    messages_sent: int = 0
    messages_received: int = 0
    values_output: int = 0


class MergeWorker:
    def __init__(self,
                 worker_id: str,             # "A" or "B"
                 data: Optional[list[int]],  # unsorted, and None after the first construction
                 inbox: Path,
                 outbox: Path,
                 output: Path,               # shared by both workers
                 state_file: Path):
        self.worker_id = worker_id
        self.data = data
        self.inbox = inbox
        self.outbox = outbox
        self.output = output
        self.state_file = state_file

        self.state: dict = self._load_state()
        self.stats = WorkerStats(**self.state["stats"])

    def _load_state(self) -> dict:
        """Load state from file, or initialize if first run."""
        if self.state_file.exists():
            with open(self.state_file) as f:
                return json.load(f)
        return self._initial_state()

    def _save_state(self) -> None:
        """Persist state to file. Syncs stats into state."""
        self.state["stats"] = vars(self.stats)
        with open(self.state_file, 'w') as f:
            json.dump(self.state, f)

    def _read_message(self) -> Optional[Message]:
        """Return the inbox message, or None if there is none."""
        if not self.inbox.exists():
            return None
        with open(self.inbox) as f:
            content = f.read().strip()
        if not content:
            return None
        data = json.loads(content)
        self.stats.messages_received += 1
        return Message(data["msg_type"], data.get("values", []))

    def _clear_inbox(self) -> None:
        """Remove the inbox message, if any."""
        self.inbox.unlink(missing_ok=True)

    def _write_message(self, msg: Message) -> None:
        """Write one message to the outbox, replacing any unread one."""
        with open(self.outbox, 'w') as f:
            json.dump({"msg_type": msg.msg_type, "values": msg.values}, f)
        self.stats.messages_sent += 1

    def _write_output(self, values: list[int]) -> None:
        """Append values to the shared output file, one integer per line."""
        if not values:
            return
        with open(self.output, 'a') as f:
            f.writelines(f"{v}\n" for v in values)
        self.stats.values_output += len(values)

    def _initial_state(self) -> dict:
        """Return initial state structure. Must include "stats"."""
        raise NotImplementedError

    def step(self) -> bool:
        """
        Read the inbox, update state, write at most one message and any
        finalized values, and save state. Returns False once done.
        """
        raise NotImplementedError

    def get_stats(self) -> WorkerStats:
        """Return statistics about work performed."""
        return self.stats

State Machine Guidance

Consider organizing your worker as a state machine with phases:

Phase 1: INIT
  - Exchange metadata (data range, count)
  - Determine merge strategy

Phase 2: MERGE
  - Exchange data in coordinated fashion
  - Write merged results to output

Phase 3: DONE
  - Signal completion to partner
  - Verify partner also done

A reasonable initial state might track:

def _initial_state(self) -> dict:
    return {
        "phase": "INIT",
        "remaining": sorted(self.data),
        "partner_min": None,
        "partner_max": None,
        "partner_count": None,
        "data_index": 0,
        "output_count": 0,
        "stats": {"comparisons": 0, "messages_sent": 0,
                  "messages_received": 0, "values_output": 0},
        # Add more state variables as needed
    }

Algorithm Approaches

Centralized approach (blocked by storage): One worker sends all its data to the other, which merges and writes everything. Holding both datasets at once exceeds the storage limit.

Range-based approach: Workers exchange range information first. If Worker A knows Worker B’s minimum value is 500, then A can immediately output all its values less than 500 without waiting. Workers coordinate to determine which values can be safely output and which require comparison with partner data.

Bubble-style exchange: Workers repeatedly exchange their current smallest unprocessed values. In each round, both workers send their smallest remaining value. Each worker outputs the global minimum (comparing received value against their own) and advances. This naturally balances work but requires many message rounds.

Coordinator and Runner (Provided)

We provide a coordinator that alternates worker execution. It constructs a fresh worker for every step—anything not saved to the state file is lost. A constraint monitor validates the message and state files after each step. run_merge.py generates the two datasets from a seed, runs the coordinator, checks the output, and writes the result to results/merge_result.json:

Code: coordinator.py 
"""
Coordinator for Distributed Merge - DO NOT MODIFY

Alternates workers until both done. Fresh worker per step -- only the
state file persists. Data is passed to the first construction only.
Monitor validates message + state after each step.
"""

from monitor import ConstraintMonitor, ConstraintViolation


class Coordinator:
    def __init__(self, make_worker_a, make_worker_b):
        # factories: callables taking first (bool), return a fresh worker:
        #   lambda first: MergeWorker("A", data_a if first else None, ...)
        self.factories = [make_worker_a, make_worker_b]
        self.monitor = ConstraintMonitor()

    def run(self, max_steps: int = 100000) -> dict:
        """Alternate workers until both done or max steps reached."""
        steps = 0
        active = [True, True]
        constructed = [False, False]
        stats = [None, None]
        violation = None

        while any(active) and steps < max_steps and violation is None:
            for i in range(2):
                if not active[i]:
                    continue
                worker = self.factories[i](not constructed[i])
                constructed[i] = True
                active[i] = worker.step()
                stats[i] = worker.get_stats()
                steps += 1
                try:
                    self.monitor.check_message(worker.outbox)
                    self.monitor.check_state(worker.state_file)
                except ConstraintViolation as e:
                    violation = str(e)
                    break

        return {
            "success": not any(active) and violation is None,
            "violation": violation,
            "total_steps": steps,
            "stats_a": stats[0],
            "stats_b": stats[1]
        }
Code: monitor.py 
"""
Constraint Monitor for Distributed Merge - DO NOT MODIFY

Validates messages and state files after each worker step.
"""

import json
from pathlib import Path

MAX_MESSAGE_VALUES = 10
MAX_MSG_TYPE_LEN = 5
MAX_VALUE = 2**31 - 1
MAX_STATE_INTEGERS = 5000


class ConstraintViolation(Exception):
    pass


class ConstraintMonitor:
    def __init__(self, state_limit: int = MAX_STATE_INTEGERS):
        self.state_limit = state_limit

    def check_message(self, msg_file: Path) -> None:
        """Validate current message, if any."""
        if not msg_file.exists():
            return
        try:
            with open(msg_file) as f:
                msg = json.load(f)
        except (json.JSONDecodeError, OSError):
            return  # mid-write; check next step
        if not isinstance(msg, dict):
            raise ConstraintViolation(f"{msg_file.name}: message must be a JSON object")
        msg_type = msg.get("msg_type", "")
        values = msg.get("values", [])
        if len(msg_type) > MAX_MSG_TYPE_LEN:
            raise ConstraintViolation(f"{msg_file.name}: msg_type exceeds {MAX_MSG_TYPE_LEN} characters")
        if len(values) > MAX_MESSAGE_VALUES:
            raise ConstraintViolation(f"{msg_file.name}: {len(values)} values in message (max {MAX_MESSAGE_VALUES})")
        for v in values:
            if not isinstance(v, int) or v > MAX_VALUE:
                raise ConstraintViolation(f"{msg_file.name}: invalid value {v!r}")

    def check_state(self, state_file: Path) -> None:
        """Enforce storage limit on persisted state."""
        if not state_file.exists():
            return
        try:
            with open(state_file) as f:
                state = json.load(f)
        except (json.JSONDecodeError, OSError):
            raise ConstraintViolation(f"{state_file.name}: state file is not valid JSON")
        count = self._count_integers(state)
        if count > self.state_limit:
            raise ConstraintViolation(
                f"{state_file.name}: state holds {count} integers (max {self.state_limit})")

    def _count_integers(self, obj) -> int:
        """Count ints in a JSON structure."""
        if isinstance(obj, bool):
            return 0
        if isinstance(obj, int):
            return 1
        if isinstance(obj, list):
            return sum(self._count_integers(x) for x in obj)
        if isinstance(obj, dict):
            return sum(self._count_integers(v) for v in obj.values())
        return 0
Code: run_merge.py 
"""
Merge runner - DO NOT MODIFY

Runs both workers on seeded data under the coordinator and writes the
result record to a JSON file.
"""

import argparse
import json
import random
import tempfile
from dataclasses import asdict
from pathlib import Path

from coordinator import Coordinator
from merge_worker import MergeWorker

VALUE_RANGE = 1_000_000


def make_data(seed: int, size_a: int, size_b: int) -> tuple[list[int], list[int]]:
    """Two disjoint unsorted lists with overlapping ranges."""
    rng = random.Random(seed)
    values = rng.sample(range(VALUE_RANGE), size_a + size_b)
    return values[:size_a], values[size_a:]


def run(data_a: list[int], data_b: list[int]) -> dict:
    """Merge data_a and data_b with two workers and return the result record."""
    with tempfile.TemporaryDirectory() as tmp:
        tmp = Path(tmp)
        a_to_b, b_to_a = tmp / "A_to_B.msg", tmp / "B_to_A.msg"
        output = tmp / "output.txt"
        state_a, state_b = tmp / "state_a.json", tmp / "state_b.json"

        coordinator = Coordinator(
            lambda first: MergeWorker("A", data_a if first else None,
                                      b_to_a, a_to_b, output, state_a),
            lambda first: MergeWorker("B", data_b if first else None,
                                      a_to_b, b_to_a, output, state_b),
        )
        result = coordinator.run()

        merged = []
        if output.exists():
            merged = [int(line) for line in output.read_text().split()]

    stats_a = asdict(result["stats_a"]) if result["stats_a"] else None
    stats_b = asdict(result["stats_b"]) if result["stats_b"] else None
    comps = [s["comparisons"] for s in (stats_a, stats_b) if s]
    balance = min(comps) / max(comps) if len(comps) == 2 and max(comps) > 0 else 0.0

    return {
        "success": result["success"],
        "violation": result["violation"],
        "correct": merged == sorted(data_a + data_b),
        "total_steps": result["total_steps"],
        "output_count": len(merged),
        "expected_count": len(data_a) + len(data_b),
        "work_balance": round(balance, 4),
        "total_messages": sum(s["messages_sent"] for s in (stats_a, stats_b) if s),
        "total_comparisons": sum(comps),
        "stats_a": stats_a,
        "stats_b": stats_b,
    }


def main():
    parser = argparse.ArgumentParser(description="Run the distributed merge")
    parser.add_argument("--seed", type=int, default=42)
    parser.add_argument("--size-a", type=int, default=4000)
    parser.add_argument("--size-b", type=int, default=3000)
    parser.add_argument("--result", type=Path, default=Path("results/merge_result.json"))
    args = parser.parse_args()

    data_a, data_b = make_data(args.seed, args.size_a, args.size_b)
    record = {"seed": args.seed, **run(data_a, data_b)}

    args.result.parent.mkdir(parents=True, exist_ok=True)
    args.result.write_text(json.dumps(record, indent=2) + "\n")

    for key in ("success", "violation", "correct", "total_steps",
                "work_balance", "total_messages", "total_comparisons"):
        print(f"{key}: {record[key]}")
    print(f"result written to {args.result}")


if __name__ == "__main__":
    main()

Scoring Criteria

Your solution is evaluated on three metrics:

Correctness (40%): The output file contains all values from both workers, sorted, with no duplicates or missing values.

Work Balance (30%): Measured as \(\min(W_A, W_B) / \max(W_A, W_B)\) where \(W\) is the comparison count. A score of 1.0 means perfect balance, and a score near 0 means one worker did everything.

Efficiency (30%): Total messages exchanged and total comparisons performed. Lower is better, but correctness and balance take priority.

Note: comparisons count toward both balance and efficiency—padding one degrades the other.

Constraints

  • Each message contains at most 10 integers
  • Each integer must be \(\leq 2^{31} - 1\)
  • Message type string maximum 5 characters
  • At most one message written per step() call
  • Workers must use only their designated inbox/outbox files
  • No shared memory or direct communication between workers
  • The state file holds at most 5000 integers (enforced by the monitor)
  • Workers write only to their outbox, output, and state files

Testing Your Solution

From q1/, run the interface tests, then the merge:

python -m pytest test_interfaces.py
python run_merge.py --seed 42
python run_merge.py --seed 7 --size-a 1000 --size-b 4000

run_merge.py reports correct, work_balance, total_messages, and total_comparisons, the quantities the scoring criteria use. Grading runs it with other seeds and sizes.

Deliverables

See Submission. Your README should explain:

  • What message types you defined and their purpose
  • Your merge strategy and why you chose it
  • How you balance work between workers
  • How your state stays within the storage limit

Problem 2: Robust HTTP Client

Implement an HTTP client that classifies failures, retries transient ones with exponential backoff, and reports each outcome through callbacks.

Context

An HTTP client must distinguish transient failures (retry) from permanent failures (do not retry). A 503 Service Unavailable indicates a temporarily overloaded server: retry after waiting. A 404 Not Found indicates a missing resource: retrying cannot help. A timeout is ambiguous: the server may be slow, overloaded, or down.

Setup

We provide a URLProvider class that generates URLs for testing. Each URL maps to a specific behavior (success, timeout, connection error, various error codes). ResponseValidator compares the callbacks your client fired for each URL against that URL’s expected behavior. test_interfaces.py runs against a local server with the same endpoints as httpbin.org and checks only that the client runs and writes the specified log and summary keys.

Code: url_provider.py 
"""
URL Provider for HTTP Client Testing

Generates URLs with known behaviors for testing retry logic, error
handling, and response classification. base_url replaces httpbin.org
with another server that has the same endpoints.

Do not modify.
"""

import random
from dataclasses import dataclass
from typing import Optional


# Default number of URLs to generate
DEFAULT_URL_COUNT = 50

DEFAULT_BASE_URL = "https://httpbin.org"
CONNECTION_ERROR_URL = "http://127.0.0.1:9/get"  # port 9 (discard): connection refused


@dataclass
class URLBehavior:
    """Describes expected behavior for a URL."""
    status_code: Optional[int]  # Expected status, None if timeout/connection error
    min_latency_ms: float       # Minimum expected response time
    should_retry: bool          # Whether client should retry on failure
    expected_callbacks: list[str]  # Callbacks that should fire
    body_keywords: list[str]    # Keywords expected in response body
    error_type: Optional[str]   # "timeout", "connection", or None


class URLProvider:
    """
    Generates test URLs with known behaviors.

    Uses httpbin.org endpoints to produce predictable responses:
    - /get: Fast 200 response
    - /delay/N: Slow response (N seconds). /delay/10 exceeds the 5 s timeout
    - /status/CODE: Returns specific status code
    - /html: Returns HTML content with predictable body
    """

    def __init__(self, seed: Optional[int] = None, count: int = DEFAULT_URL_COUNT,
                 base_url: str = DEFAULT_BASE_URL):
        """
        Initialize URL provider.

        Args:
            seed: Random seed for reproducibility
            count: Number of URLs to generate
            base_url: Server with httpbin.org's endpoints
        """
        self.rng = random.Random(seed)
        self.count = count
        self.base_url = base_url.rstrip("/")
        self._urls: list[str] = []
        self._behaviors: dict[str, URLBehavior] = {}
        self._index = 0
        self._generate_urls()

    def _generate_urls(self) -> None:
        """Generate URL set with various behaviors."""
        base = self.base_url

        # Distribution of URL types (roughly):
        # 40% success (fast), the remainder
        # 10% success (slow)
        # 15% client errors (4xx)
        # 15% server errors (5xx)
        # 5% timeouts
        # 5% connection errors
        # 5% with body keywords
        # 5% slow with body keywords

        url_specs = []

        # Slow successes (10%)
        for _ in range(int(self.count * 0.10)):
            delay = self.rng.choice([1, 2])  # 1-2 second delay
            url_specs.append(("success_slow", f"{base}/delay/{delay}"))

        # Client errors (15%)
        for _ in range(int(self.count * 0.15)):
            code = self.rng.choice([400, 401, 403, 404])
            url_specs.append(("client_error", f"{base}/status/{code}"))

        # Server errors (15%)
        for _ in range(int(self.count * 0.15)):
            code = self.rng.choice([500, 502, 503])
            url_specs.append(("server_error", f"{base}/status/{code}"))

        # Body keyword matches (5%)
        for _ in range(int(self.count * 0.05)):
            url_specs.append(("body_match", f"{base}/html"))

        # Slow with body match (5%)
        for _ in range(max(1, int(self.count * 0.05))):
            url_specs.append(("slow_body", f"{base}/delay/1"))

        # Timeouts (5%): longer than the client's 5 s timeout
        for _ in range(int(self.count * 0.05)):
            url_specs.append(("timeout", f"{base}/delay/10"))

        # Connection errors (5%)
        for _ in range(int(self.count * 0.05)):
            url_specs.append(("connection", CONNECTION_ERROR_URL))

        # Fast successes (the remainder)
        for _ in range(max(0, self.count - len(url_specs))):
            url_specs.append(("success_fast", f"{base}/get"))

        # Shuffle
        self.rng.shuffle(url_specs)

        # Assign behaviors
        for url_type, url in url_specs:
            # Make URLs unique by adding query param
            unique_url = f"{url}?_id={len(self._urls)}"
            self._urls.append(unique_url)
            self._behaviors[unique_url] = self._create_behavior(url_type, url)

    def _create_behavior(self, url_type: str, base_url: str) -> URLBehavior:
        """Create expected behavior for URL type."""
        if url_type == "success_fast":
            return URLBehavior(
                status_code=200,
                min_latency_ms=0,
                should_retry=False,
                expected_callbacks=["on_success"],
                body_keywords=[],
                error_type=None
            )
        elif url_type == "success_slow":
            return URLBehavior(
                status_code=200,
                min_latency_ms=1000,
                should_retry=False,
                expected_callbacks=["on_success", "on_slow_response"],
                body_keywords=[],
                error_type=None
            )
        elif url_type == "client_error":
            code = int(base_url.split("/")[-1])
            return URLBehavior(
                status_code=code,
                min_latency_ms=0,
                should_retry=False,
                expected_callbacks=["on_client_error"],
                body_keywords=[],
                error_type=None
            )
        elif url_type == "server_error":
            code = int(base_url.split("/")[-1])
            return URLBehavior(
                status_code=code,
                min_latency_ms=0,
                should_retry=True,
                expected_callbacks=["on_server_error", "on_retry"],
                body_keywords=[],
                error_type=None
            )
        elif url_type == "body_match":
            return URLBehavior(
                status_code=200,
                min_latency_ms=0,
                should_retry=False,
                expected_callbacks=["on_success", "on_body_match"],
                body_keywords=["Herman"],  # httpbin.org/html contains "Herman Melville"
                error_type=None
            )
        elif url_type == "timeout":
            return URLBehavior(
                status_code=None,
                min_latency_ms=5000,
                should_retry=True,
                expected_callbacks=["on_timeout", "on_retry", "on_max_retries"],
                body_keywords=[],
                error_type="timeout"
            )
        elif url_type == "connection":
            return URLBehavior(
                status_code=None,
                min_latency_ms=0,
                should_retry=True,
                expected_callbacks=["on_connection_error", "on_retry", "on_max_retries"],
                body_keywords=[],
                error_type="connection"
            )
        elif url_type == "slow_body":
            return URLBehavior(
                status_code=200,
                min_latency_ms=1000,
                should_retry=False,
                expected_callbacks=["on_success", "on_slow_response"],
                body_keywords=[],
                error_type=None
            )
        else:
            # Default: success
            return URLBehavior(
                status_code=200,
                min_latency_ms=0,
                should_retry=False,
                expected_callbacks=["on_success"],
                body_keywords=[],
                error_type=None
            )

    def next_url(self) -> Optional[str]:
        """
        Get next URL to fetch.

        Returns:
            URL string, or None if all URLs consumed.
        """
        if self._index >= len(self._urls):
            return None
        url = self._urls[self._index]
        self._index += 1
        return url

    def get_behavior(self, url: str) -> Optional[URLBehavior]:
        """
        Get expected behavior for URL.

        Args:
            url: URL to look up

        Returns:
            URLBehavior if known, None otherwise.
        """
        return self._behaviors.get(url)

    def remaining(self) -> int:
        """Number of URLs remaining."""
        return len(self._urls) - self._index

    def total(self) -> int:
        """Total number of URLs."""
        return len(self._urls)

    def reset(self) -> None:
        """Reset to beginning of URL list."""
        self._index = 0

    def get_all_urls(self) -> list[str]:
        """Get all URLs (for validation)."""
        return self._urls.copy()


class ResponseValidator:
    """
    Validates that client invoked correct callbacks for each URL.

    Usage:
        validator = ResponseValidator(provider)
        # ... client runs and logs callbacks ...
        validator.add_callback(url, "on_success", {...})
        results = validator.validate()
    """

    def __init__(self, provider: URLProvider):
        self.provider = provider
        self._recorded: dict[str, list[str]] = {}  # url -> list of callback names

    def add_callback(self, url: str, callback_name: str) -> None:
        """Record that a callback was invoked for URL."""
        if url not in self._recorded:
            self._recorded[url] = []
        self._recorded[url].append(callback_name)

    def validate(self) -> dict:
        """
        Validate recorded callbacks against expected behaviors.

        Returns:
            Validation results with pass/fail for each URL.
        """
        results = {
            "total": self.provider.total(),
            "passed": 0,
            "failed": 0,
            "failures": []
        }

        for url in self.provider.get_all_urls():
            behavior = self.provider.get_behavior(url)
            recorded = set(self._recorded.get(url, []))
            expected = set(behavior.expected_callbacks) if behavior else set()

            # Check required callbacks present
            # Note: on_retry may appear multiple times, on_max_retries if exhausted
            required = {cb for cb in expected if cb not in ("on_retry", "on_max_retries")}
            missing = required - recorded

            if missing:
                results["failed"] += 1
                results["failures"].append({
                    "url": url,
                    "expected": list(expected),
                    "recorded": list(recorded),
                    "missing": list(missing)
                })
            else:
                results["passed"] += 1

        return results


if __name__ == "__main__":
    # Demo usage
    provider = URLProvider(seed=42, count=20)

    print(f"Generated {provider.total()} URLs")
    print()

    while (url := provider.next_url()) is not None:
        behavior = provider.get_behavior(url)
        print(f"URL: {url[:60]}...")
        print(f"  Status: {behavior.status_code}")
        print(f"  Retry: {behavior.should_retry}")
        print(f"  Callbacks: {behavior.expected_callbacks}")
        print()

Client Implementation

The starter’s http_client.py defines the ResponseHandler callbacks, a LoggingResponseHandler that writes them to the log, the FetchStats summary, and RobustHTTPClient. Multiple callbacks may fire for a single request (a slow successful response triggers both on_success and on_slow_response). Implement:

  • The nine LoggingResponseHandler callbacks
  • RobustHTTPClient.calculate_backoff(), fetch(), and fetch_all()

main() is provided and is the entry point grading runs. SLOW_THRESHOLD_MS (500) and BASE_TIMEOUT_SEC (5) are fixed. The retry and backoff constants are yours to choose.

Code: http_client.py 
"""HTTP client with failure classification, retry, and backoff."""

import argparse
import json
import random
import socket
import time
import urllib.error
import urllib.request
from abc import ABC, abstractmethod
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path

from url_provider import DEFAULT_URL_COUNT, URLProvider


class ResponseHandler(ABC):
    """
    Callbacks for fetch outcomes. Multiple callbacks may fire for a
    single request (a slow successful response triggers both on_success
    and on_slow_response).
    """

    @abstractmethod
    def on_success(self, url: str, status: int, body: bytes, latency_ms: float) -> None:
        """Called for 2xx responses."""

    @abstractmethod
    def on_client_error(self, url: str, status: int, body: bytes) -> None:
        """Called for 4xx responses. These are not retried."""

    @abstractmethod
    def on_server_error(self, url: str, status: int, attempt: int) -> None:
        """Called for 5xx responses. These are retried with backoff."""

    @abstractmethod
    def on_timeout(self, url: str, attempt: int, timeout_sec: float) -> None:
        """Called when a request times out. Retried with backoff."""

    @abstractmethod
    def on_connection_error(self, url: str, attempt: int, error: str) -> None:
        """Called when the connection fails (DNS, refused). Retried with backoff."""

    @abstractmethod
    def on_slow_response(self, url: str, latency_ms: float) -> None:
        """Called when a 2xx response took more than SLOW_THRESHOLD_MS, in addition to on_success."""

    @abstractmethod
    def on_retry(self, url: str, attempt: int, wait_ms: float, reason: str) -> None:
        """Called before each retry. attempt is the 1-indexed attempt about to be made, and reason is the failure callback's name without "on_"."""

    @abstractmethod
    def on_body_match(self, url: str, keyword: str) -> None:
        """Called when a 2xx body contains a monitored keyword, in addition to on_success."""

    @abstractmethod
    def on_max_retries(self, url: str, attempts: int, last_error: str) -> None:
        """Called once when every attempt for a URL has failed with a retryable error. attempts counts all of them."""


class LoggingResponseHandler(ResponseHandler):
    """
    Writes one JSON object per line to log_path for every callback.

    Each line holds "timestamp" (UTC, ISO-8601, milliseconds, "Z"),
    "event" (the callback name without "on_"), and the callback's
    arguments except body, under their argument names.
    """

    def __init__(self, log_path: Path):
        self.log_path = log_path

    def _log(self, event: str, **fields) -> None:
        """Append one event line to the log."""
        timestamp = datetime.now(timezone.utc).isoformat(timespec="milliseconds")
        entry = {"timestamp": timestamp.replace("+00:00", "Z"), "event": event, **fields}
        with open(self.log_path, "a") as f:
            f.write(json.dumps(entry) + "\n")

    def on_success(self, url, status, body, latency_ms):
        raise NotImplementedError

    def on_client_error(self, url, status, body):
        raise NotImplementedError

    def on_server_error(self, url, status, attempt):
        raise NotImplementedError

    def on_timeout(self, url, attempt, timeout_sec):
        raise NotImplementedError

    def on_connection_error(self, url, attempt, error):
        raise NotImplementedError

    def on_slow_response(self, url, latency_ms):
        raise NotImplementedError

    def on_retry(self, url, attempt, wait_ms, reason):
        raise NotImplementedError

    def on_body_match(self, url, keyword):
        raise NotImplementedError

    def on_max_retries(self, url, attempts, last_error):
        raise NotImplementedError


@dataclass
class FetchStats:
    """Run totals, written as the summary file."""
    total_urls: int = 0
    successful: int = 0          # URLs ending in a 2xx
    failed: int = 0
    total_requests: int = 0      # attempts, retries included
    retries: int = 0
    avg_latency_ms: float = 0.0  # over 2xx responses
    slow_responses: int = 0
    by_status: dict = field(default_factory=dict)  # "503" -> responses, retries included
    by_error: dict = field(default_factory=dict)   # "timeout" | "connection" -> attempts


class RobustHTTPClient:
    SLOW_THRESHOLD_MS = 500.0
    BASE_TIMEOUT_SEC = 5.0  # below the provider's 10 s timeout delay
    MAX_RETRIES = 3         # after the first attempt
    INITIAL_BACKOFF_MS = 100.0
    MAX_BACKOFF_MS = 5000.0
    BACKOFF_MULTIPLIER = 2.0
    MONITORED_KEYWORDS = ["Herman"]  # URLProvider's body_keywords

    def __init__(self, handler: ResponseHandler):
        self.handler = handler
        self.stats = FetchStats()
        self.latencies: list[float] = []

    def calculate_backoff(self, attempt: int) -> float:
        """
        Delay in milliseconds before retry number attempt (0-indexed):
        min(base + uniform(0, 0.1 * base), MAX_BACKOFF_MS), where
        base = INITIAL_BACKOFF_MS * BACKOFF_MULTIPLIER ** attempt.
        """
        raise NotImplementedError

    def fetch(self, url: str) -> bool:
        """
        Fetch url, retrying 5xx, timeouts, and connection errors with
        backoff. 4xx is not retried. Fires the callbacks for each outcome
        and updates self.stats and self.latencies. Returns True on a 2xx.
        """
        raise NotImplementedError

    def fetch_all(self, provider: URLProvider) -> dict:
        """Fetch every URL from provider and return self.stats as a dict."""
        raise NotImplementedError

    def _count_status(self, status: int) -> None:
        key = str(status)
        self.stats.by_status[key] = self.stats.by_status.get(key, 0) + 1

    def _count_error(self, kind: str) -> None:
        self.stats.by_error[kind] = self.stats.by_error.get(kind, 0) + 1


def main():
    parser = argparse.ArgumentParser(description="Robust HTTP Client")
    parser.add_argument("--seed", type=int, default=42)
    parser.add_argument("--count", type=int, default=DEFAULT_URL_COUNT)
    parser.add_argument("--base-url", help="server with httpbin.org's endpoints")
    parser.add_argument("--log", type=Path, default=Path("results/output.log"))
    parser.add_argument("--summary", type=Path, default=Path("results/summary.json"))
    args = parser.parse_args()

    args.log.parent.mkdir(parents=True, exist_ok=True)
    args.summary.parent.mkdir(parents=True, exist_ok=True)
    args.log.unlink(missing_ok=True)

    provider_args = {"seed": args.seed, "count": args.count}
    if args.base_url:
        provider_args["base_url"] = args.base_url
    provider = URLProvider(**provider_args)
    client = RobustHTTPClient(LoggingResponseHandler(args.log))

    print(f"Fetching {provider.total()} URLs...")
    summary = client.fetch_all(provider)
    args.summary.write_text(json.dumps(summary, indent=2) + "\n")

    for key, value in summary.items():
        print(f"  {key}: {value}")
    print(f"log written to {args.log}, summary to {args.summary}")


if __name__ == "__main__":
    main()

Expected Behavior Matrix

Condition Callbacks Retry?
200 OK, fast on_success No
200 OK, slow (>500ms) on_success, on_slow_response No
200 OK, body contains keyword on_success, on_body_match("keyword") No
400 Bad Request on_client_error No
404 Not Found on_client_error No
500 Internal Server Error on_server_error, then on_retry, … Yes
503 Service Unavailable on_server_error, then on_retry, … Yes
Timeout on_timeout, then on_retry, … Yes
Connection refused on_connection_error, then on_retry, … Yes
Max retries hit on_max_retries N/A

Exponential Backoff

Retry delays grow exponentially with jitter, capped at MAX_BACKOFF_MS. attempt is the 0-indexed retry number:

\[\text{delay} = \min\left(b \cdot m^{\text{attempt}} + U(0,\ 0.1\, b \cdot m^{\text{attempt}}),\ \text{MAX\_BACKOFF\_MS}\right)\]

where \(b\) is INITIAL_BACKOFF_MS and \(m\) is BACKOFF_MULTIPLIER.

Output Format

Your client must produce a log file with one JSON object per line. event is the callback name without on_, and the remaining fields are the callback’s arguments except body:

{"timestamp": "2026-09-01T10:30:00.123Z", "event": "success", "url": "http://...", "status": 200, "latency_ms": 45.2}
{"timestamp": "2026-09-01T10:30:00.234Z", "event": "slow_response", "url": "http://...", "latency_ms": 623.5}
{"timestamp": "2026-09-01T10:30:00.456Z", "event": "server_error", "url": "http://...", "status": 503, "attempt": 1}
{"timestamp": "2026-09-01T10:30:00.567Z", "event": "retry", "url": "http://...", "attempt": 2, "wait_ms": 104.2, "reason": "server_error"}
{"timestamp": "2026-09-01T10:30:00.789Z", "event": "success", "url": "http://...", "status": 200, "latency_ms": 89.1}

main() writes the summary returned by fetch_all(). successful and failed count URLs, while total_requests, by_status, and by_error count attempts, retries included. avg_latency_ms is the mean over 2xx responses:

{
    "total_urls": 50,
    "successful": 32,
    "failed": 18,
    "total_requests": 83,
    "retries": 33,
    "avg_latency_ms": 517.5,
    "slow_responses": 10,
    "by_status": {
        "200": 32,
        "400": 5,
        "401": 1,
        "404": 1,
        "500": 16,
        "503": 12
    },
    "by_error": {
        "timeout": 8,
        "connection": 8
    }
}

Validation

We validate your implementation by:

  1. Running python http_client.py with fixed seeds
  2. Checking that correct callbacks were invoked for each URL
  3. Verifying that retries occur only for transient failures
  4. Checking backoff timing is within acceptable range
  5. Validating log format and completeness

Requirements

  • Use only urllib.request for HTTP (no requests library)
  • Use only standard library modules
  • Catch every exception, so a failed URL does not abort the run
  • All timestamps must be UTC in ISO-8601 format

Deliverables

See Submission. Your README should explain:

  • Your retry strategy
  • How you classify retryable vs. permanent failures
  • Backoff parameters you chose and why

Testing

From q2/, run the interface tests (local server, no network), then a full run against httpbin.org:

python -m pytest test_interfaces.py
python http_client.py --seed 42

The full run writes results/output.log and results/summary.json and takes one to two minutes, most of it on the timeout URLs. Run with several seeds.

Problem 3: Durable Event Log

Maintain a durable log of packets that arrive reordered, duplicated, and corrupted, from a process that may be terminated at any time, capturing as many packets as possible with as few ordering violations as possible.

Context

Packets arrive reordered, duplicated, corrupted, or not at all. The logger balances coverage (packets captured) against ordering (packets written in sequence).

The fundamental trade-off: write immediately and risk out-of-order entries, or buffer to reorder and risk losing buffered data on termination.

Setup

We provide the packet source, MessageSource:

Code: message_source.py 
"""
Message Source for Durable Event Log Testing

Generates packets with realistic network behavior:
- Reordering within a window
- Duplicate delivery
- Packet loss
- Checksum corruption

At random points, calls sys.exit(1) to simulate process termination.
State is persisted to allow resume on restart.

Do not modify.
"""

import hashlib
import json
import os
import random
import sys
import zlib
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Optional


# Default configuration
DEFAULT_TOTAL_PACKETS = 1000
DEFAULT_REORDER_WINDOW = 10
DEFAULT_DUPLICATE_PROB = 0.05
DEFAULT_LOSS_PROB = 0.02
DEFAULT_CORRUPTION_PROB = 0.03
DEFAULT_TERMINATION_PROB = 0.005  # Per-packet probability of termination


@dataclass
class Packet:
    """A single packet from the message source."""
    sequence: int       # Global sequence number (ground truth order)
    timestamp: float    # Generation timestamp
    payload: bytes      # Arbitrary payload data
    checksum: int       # CRC32 checksum of payload

    def to_dict(self) -> dict:
        return {
            "sequence": self.sequence,
            "timestamp": self.timestamp,
            "payload": self.payload.hex(),
            "checksum": self.checksum
        }

    @classmethod
    def from_dict(cls, d: dict) -> "Packet":
        return cls(
            sequence=d["sequence"],
            timestamp=d["timestamp"],
            payload=bytes.fromhex(d["payload"]),
            checksum=d["checksum"]
        )


class MessageSource:
    """
    Simulates unreliable message delivery with process termination.

    On first init with a seed, generates all packets and saves to file.
    On subsequent inits with same seed, loads packets and resumes.
    At random points, calls sys.exit(1) to simulate crash.
    """

    def __init__(self,
                 seed: int,
                 total_packets: int = DEFAULT_TOTAL_PACKETS,
                 reorder_window: int = DEFAULT_REORDER_WINDOW,
                 duplicate_prob: float = DEFAULT_DUPLICATE_PROB,
                 loss_prob: float = DEFAULT_LOSS_PROB,
                 corruption_prob: float = DEFAULT_CORRUPTION_PROB,
                 termination_prob: float = DEFAULT_TERMINATION_PROB,
                 state_dir: Optional[Path] = None):
        """
        Initialize message source.

        Args:
            seed: Random seed for reproducibility
            total_packets: Number of packets to generate
            reorder_window: Max positions a packet can be delayed
            duplicate_prob: Probability of duplicate delivery
            loss_prob: Probability of packet never arriving
            corruption_prob: Probability of corrupted checksum
            termination_prob: Per-packet probability of process termination
            state_dir: Directory for state files (default: /tmp)
        """
        self.seed = seed
        self.total_packets = total_packets
        self.reorder_window = reorder_window
        self.duplicate_prob = duplicate_prob
        self.loss_prob = loss_prob
        self.corruption_prob = corruption_prob
        self.termination_prob = termination_prob

        # State file paths
        if state_dir is None:
            state_dir = Path("/tmp")
        self._packets_file = state_dir / f".ms_{seed}_packets.json"
        self._position_file = state_dir / f".ms_{seed}_position.txt"
        self._retransmit_file = state_dir / f".ms_{seed}_retransmit.json"

        # Random generator for runtime decisions
        self.rng = random.Random(seed)

        # Load or generate packets
        self._delivery_queue: list[tuple[int, Packet, bool]] = []  # (priority, packet, corrupted)
        self._position = 0
        self._terminated = False
        self._delivered_here = False
        self._pending_retransmits: list[int] = []
        self._generated_sequences: set[int] = set()

        self._initialize()

    def _initialize(self) -> None:
        """Load existing state or generate new packets."""
        if self._packets_file.exists():
            self._load_state()
            # Reseed by position: a restarted process must not replay the same terminations
            self.rng = random.Random(f"{self.seed}:{self._position}")
        else:
            # Generate new packet sequence
            self._generate_packets()
            self._save_packets()
            self._save_position()

    def _generate_packets(self) -> None:
        """Generate all packets with network effects applied."""
        gen_rng = random.Random(self.seed)

        # Generate ground truth packets
        ground_truth = []
        for seq in range(self.total_packets):
            payload = f"packet_{seq:06d}_{gen_rng.randint(0, 999999)}".encode()
            checksum = zlib.crc32(payload)
            packet = Packet(
                sequence=seq,
                timestamp=seq * 0.001 + gen_rng.uniform(0, 0.0001),
                payload=payload,
                checksum=checksum
            )
            ground_truth.append(packet)
            self._generated_sequences.add(seq)

        # Apply network effects to create delivery queue
        delivery = []
        for packet in ground_truth:
            # Loss: packet never delivered
            if gen_rng.random() < self.loss_prob:
                continue

            # Reordering: add random delay
            delay = gen_rng.randint(0, self.reorder_window)
            priority = packet.sequence + delay

            # Corruption: bad checksum
            corrupted = gen_rng.random() < self.corruption_prob

            delivery.append((priority, packet, corrupted))

            # Duplication: add another copy
            if gen_rng.random() < self.duplicate_prob:
                dup_delay = gen_rng.randint(1, self.reorder_window * 2)
                delivery.append((priority + dup_delay, packet, corrupted))

        # Sort by priority (delivery order)
        delivery.sort(key=lambda x: x[0])
        self._delivery_queue = delivery

    def _save_packets(self) -> None:
        """Save generated packets to file."""
        data = {
            "seed": self.seed,
            "total_packets": self.total_packets,
            "generated_sequences": list(self._generated_sequences),
            "delivery_queue": [
                {"priority": p, "packet": pkt.to_dict(), "corrupted": c}
                for p, pkt, c in self._delivery_queue
            ]
        }
        with open(self._packets_file, 'w') as f:
            json.dump(data, f)

    def _load_state(self) -> None:
        """Load packets and position from files."""
        # Load packets
        with open(self._packets_file) as f:
            data = json.load(f)

        self._generated_sequences = set(data["generated_sequences"])
        self._delivery_queue = [
            (item["priority"], Packet.from_dict(item["packet"]), item["corrupted"])
            for item in data["delivery_queue"]
        ]

        # Load position
        if self._position_file.exists():
            with open(self._position_file) as f:
                self._position = int(f.read().strip())
        else:
            self._position = 0

        # Load pending retransmits
        if self._retransmit_file.exists():
            with open(self._retransmit_file) as f:
                self._pending_retransmits = json.load(f)
        else:
            self._pending_retransmits = []

    def _save_position(self) -> None:
        """Save current position."""
        with open(self._position_file, 'w') as f:
            f.write(str(self._position))

        with open(self._retransmit_file, 'w') as f:
            json.dump(self._pending_retransmits, f)

    def _cleanup(self) -> None:
        """Remove state files after completion."""
        for f in [self._packets_file, self._position_file, self._retransmit_file]:
            if f.exists():
                f.unlink()

    def receive(self) -> Optional[Packet]:
        """
        Receive next packet.

        Returns:
            Packet if available, None if all packets delivered.

        May call sys.exit(1) to simulate process termination.
        """
        if self._terminated:
            return None

        # Pending retransmits first, skipping any for a lost packet
        while self._pending_retransmits:
            seq = self._pending_retransmits.pop(0)
            self._save_position()
            for _, pkt, _ in self._delivery_queue:
                if pkt.sequence == seq:
                    if self.rng.random() < self.corruption_prob:
                        pkt = Packet(pkt.sequence, pkt.timestamp, pkt.payload,
                                    pkt.checksum ^ 0xDEADBEEF)
                    return pkt

        # Check if done
        if self._position >= len(self._delivery_queue):
            self._terminated = True
            self._cleanup()
            return None

        # Never before this process's first delivery, so every restart makes progress
        if self._delivered_here and self.rng.random() < self.termination_prob:
            self._save_position()
            sys.exit(1)

        # Get next packet from delivery queue
        priority, packet, corrupted = self._delivery_queue[self._position]
        self._position += 1
        self._delivered_here = True
        self._save_position()

        # Apply corruption if flagged
        if corrupted:
            packet = Packet(
                sequence=packet.sequence,
                timestamp=packet.timestamp,
                payload=packet.payload,
                checksum=packet.checksum ^ 0xDEADBEEF  # Corrupt checksum
            )

        return packet

    def request_retransmit(self, sequence: int) -> None:
        """
        Request retransmission of a packet.

        The packet will be re-queued for delivery.
        """
        if sequence not in self._pending_retransmits:
            self._pending_retransmits.append(sequence)
            self._save_position()

    def verify_checksum(self, packet: Packet) -> bool:
        """Check if packet checksum is valid."""
        return zlib.crc32(packet.payload) == packet.checksum

    def get_ground_truth(self) -> list[int]:
        """
        Get list of all sequence numbers that were generated.
        Only meaningful after completion.
        """
        return sorted(self._generated_sequences)

    def is_terminated(self) -> bool:
        """Check if source has terminated (all packets delivered)."""
        return self._terminated

    def get_position(self) -> int:
        """Get current position in delivery queue."""
        return self._position

    def get_total_deliveries(self) -> int:
        """Get total number of packets to deliver (including duplicates)."""
        return len(self._delivery_queue)


def run_demo():
    """Demo showing termination and restart behavior."""
    import tempfile

    with tempfile.TemporaryDirectory() as tmpdir:
        state_dir = Path(tmpdir)

        print("Creating message source (seed=42, 100 packets)...")
        source = MessageSource(
            seed=42,
            total_packets=100,
            termination_prob=0.0,  # Disable termination for demo
            state_dir=state_dir
        )

        print(f"Total deliveries: {source.get_total_deliveries()}")

        # Receive some packets
        received = []
        corrupted = 0
        while not source.is_terminated():
            packet = source.receive()
            if packet is None:
                break
            if not source.verify_checksum(packet):
                corrupted += 1
                source.request_retransmit(packet.sequence)
            else:
                received.append(packet.sequence)

        print(f"Received: {len(received)} packets")
        print(f"Corrupted: {corrupted}")
        print(f"Unique sequences: {len(set(received))}")

        # Check for duplicates
        from collections import Counter
        counts = Counter(received)
        dups = [seq for seq, count in counts.items() if count > 1]
        print(f"Duplicates: {len(dups)}")

        # Check ordering
        inversions = sum(1 for i in range(len(received)-1)
                        if received[i] > received[i+1])
        print(f"Inversions: {inversions}")


if __name__ == "__main__":
    run_demo()

Event Logger Implementation

The starter’s event_logger.py defines LoggerStats and EventLogger. Implement _recover_from_log(), run(), _handle_packet(), _should_flush(), _flush_buffer(), and _finalize(). _write_packet() writes one log line in the format below. main() is provided and is the entry point run_logger.py starts.

Code: event_logger.py 
"""Append-only event log for an unreliable, terminating packet source."""

import argparse
import sys
from dataclasses import asdict, dataclass
from pathlib import Path

from message_source import MessageSource, Packet


@dataclass
class LoggerStats:
    """Counts for one process."""
    packets_received: int = 0
    packets_written: int = 0
    duplicates_discarded: int = 0
    corrupted_packets: int = 0
    retransmit_requests: int = 0
    retransmits_received: int = 0
    inversions: int = 0             # LATE writes
    gaps: int = 0                   # sequences below the highest logged one, missing from the log
    buffer_flushes: int = 0
    final_buffer_size: int = 0      # at end of stream, before the final flush


class EventLogger:
    GAP_TIMEOUT = 20  # packets a gap persists before a retransmit request

    def __init__(self,
                 source: MessageSource,
                 log_file: Path,
                 buffer_size: int = 30):
        self.source = source
        self.log_file = log_file
        self.buffer_size = buffer_size  # forces a flush when reached

        self.buffer: list[Packet] = []
        self.seen_sequences: set[int] = set()
        self.last_written_seq: int = -1
        self.pending_retransmits: set[int] = set()
        self.retransmitted: set[int] = set()
        self.gap_age: dict[int, int] = {}

        self.stats = LoggerStats()
        self._recover_from_log()

    def _recover_from_log(self) -> None:
        """Rebuild state from the log."""
        raise NotImplementedError

    def _write_packet(self, packet: Packet, status: str) -> None:
        """Append one log line: sequence,timestamp,payload_hex,status."""
        line = f"{packet.sequence},{packet.timestamp:.6f},{packet.payload.hex()},{status}\n"
        with open(self.log_file, "a") as f:
            f.write(line)
        self.stats.packets_written += 1

    def run(self) -> LoggerStats:
        """Receive and log packets until the end of the stream, then return the stats."""
        raise NotImplementedError

    def _handle_packet(self, packet: Packet) -> None:
        """Process a single packet."""
        raise NotImplementedError

    def _should_flush(self) -> bool:
        """Determine if buffer should be flushed."""
        raise NotImplementedError

    def _flush_buffer(self) -> None:
        """Write buffered packets to log."""
        raise NotImplementedError

    def _finalize(self) -> None:
        """Flush the buffer at the end of the stream. Never reached on termination."""
        raise NotImplementedError


def main():
    parser = argparse.ArgumentParser(description="Durable Event Logger")
    parser.add_argument("--seed", type=int, default=42)
    parser.add_argument("--packets", type=int, default=1000)
    parser.add_argument("--buffer-size", type=int, default=30)
    parser.add_argument("--log", type=Path, default=Path("results/events.log"))
    parser.add_argument("--termination-prob", type=float, default=0.005)
    parser.add_argument("--state-dir", type=Path, default=Path("results"))
    args = parser.parse_args()

    args.log.parent.mkdir(parents=True, exist_ok=True)
    args.state_dir.mkdir(parents=True, exist_ok=True)

    source = MessageSource(seed=args.seed, total_packets=args.packets,
                           termination_prob=args.termination_prob,
                           state_dir=args.state_dir)
    logger = EventLogger(source, args.log, buffer_size=args.buffer_size)
    stats = logger.run()
    print(f"end of stream: {asdict(stats)}")


if __name__ == "__main__":
    main()

Restarts (Provided)

When the source terminates the process, run_logger.py starts event_logger.py again with the same arguments, until the logger reaches the end of the stream. Only the log file survives a termination: the new process rebuilds its state from the log in _recover_from_log(). A logger that does not recover re-writes every packet after each restart. Packets buffered at a termination are gone from the log but not from the source: a retransmit request recovers them. A lost packet is never delivered, and a retransmit request for it returns nothing. LoggerStats covers one process. run_logger.py scores the final log and writes results/metrics.json:

Code: run_logger.py 
"""
Logger runner - DO NOT MODIFY

Restarts event_logger.py after each termination until the end of the
stream, then scores the log and writes the metrics to a JSON file.
"""

import argparse
import json
import subprocess
import sys
from collections import Counter
from pathlib import Path

MAX_RUNS = 200


def read_log(log: Path) -> list[tuple[int, str]]:
    """(sequence, status) for each well-formed log line."""
    entries = []
    if log.exists():
        for line in log.read_text().splitlines():
            fields = line.split(",")
            if len(fields) == 4 and fields[0].isdigit():
                entries.append((int(fields[0]), fields[3]))
    return entries


def score(sequences: list[int], total_packets: int) -> dict:
    """Coverage and ordering as defined in the handout."""
    generated = set(range(total_packets))
    coverage = len(set(sequences) & generated) / total_packets if total_packets else 0.0
    n = len(sequences)
    inversions = sum(1 for i in range(n) for j in range(i + 1, n)
                     if sequences[i] > sequences[j])
    pairs = n * (n - 1) // 2
    ordering = 1 - inversions / pairs if pairs else 1.0
    return {"coverage": round(coverage, 4), "ordering": round(ordering, 6),
            "pair_inversions": inversions}


def main():
    parser = argparse.ArgumentParser(description="Run the event logger to the end of the stream")
    parser.add_argument("--seed", type=int, default=42)
    parser.add_argument("--packets", type=int, default=1000)
    parser.add_argument("--buffer-size", type=int, default=30)
    parser.add_argument("--termination-prob", type=float, default=0.005)
    parser.add_argument("--log", type=Path, default=Path("results/events.log"))
    parser.add_argument("--metrics", type=Path, default=Path("results/metrics.json"))
    args = parser.parse_args()

    state_dir = args.log.parent
    state_dir.mkdir(parents=True, exist_ok=True)
    args.log.unlink(missing_ok=True)
    for stale in state_dir.glob(f".ms_{args.seed}_*"):
        stale.unlink()

    command = [sys.executable, "event_logger.py",
               "--seed", str(args.seed), "--packets", str(args.packets),
               "--buffer-size", str(args.buffer_size),
               "--termination-prob", str(args.termination_prob),
               "--log", str(args.log), "--state-dir", str(state_dir)]

    runs = 0
    while True:
        runs += 1
        proc = subprocess.run(command, stderr=subprocess.PIPE, text=True)
        sys.stderr.write(proc.stderr)
        if proc.returncode == 0:
            break
        # An exception also exits 1, but only a termination leaves no traceback
        if proc.returncode != 1 or "Traceback" in proc.stderr:
            sys.exit(f"event_logger.py failed on run {runs} (status {proc.returncode})")
        if runs >= MAX_RUNS:
            sys.exit(f"no end of stream after {MAX_RUNS} runs")

    entries = read_log(args.log)
    sequences = [seq for seq, _ in entries]
    record = {
        "seed": args.seed,
        "packets": args.packets,
        "buffer_size": args.buffer_size,
        "runs": runs,
        "lines": len(entries),
        "unique_sequences": len(set(sequences)),
        "duplicate_lines": len(sequences) - len(set(sequences)),
        "status_counts": dict(Counter(status for _, status in entries)),
        **score(sequences, args.packets),
    }

    args.metrics.parent.mkdir(parents=True, exist_ok=True)
    args.metrics.write_text(json.dumps(record, indent=2) + "\n")
    for key in ("runs", "coverage", "ordering", "duplicate_lines", "status_counts"):
        print(f"{key}: {record[key]}")
    print(f"metrics written to {args.metrics}")


if __name__ == "__main__":
    main()

Log File Format

Write one line per packet to the log file:

sequence,timestamp,payload_hex,status

Where status is one of:

  • OK: Packet written in order
  • LATE: Packet arrived after later sequences were written (inversion)
  • RETRANSMIT: Packet recovered via retransmit request

Example:

42,1705123456.789,48656c6c6f,OK
43,1705123456.812,576f726c64,OK
45,1705123457.001,4d69737365,OK
44,1705123457.234,466f756e64,LATE
46,1705123457.456,52657472,RETRANSMIT

Handling Different Conditions

Condition Detection Recommended Response
Reordering packet.sequence < expected Buffer and sort, or write with LATE status
Duplicate sequence in seen_sequences Discard silently
Corruption verify_checksum() == False Call request_retransmit()
Gap Missing sequence after timeout Call request_retransmit() or accept loss
End of stream receive() returns None Flush remaining buffer
Abrupt termination Process exits mid-run No cleanup runs — only flushed data survives
Restart Log file exists at construction Rebuild state from the log in _recover_from_log()

The Trade-off

Aggressive writing (small buffer):

  • High coverage: data written before termination
  • High inversions: out-of-order packets written immediately
  • Low retransmit benefit: less time to recover packets

Conservative writing (large buffer):

  • Lower coverage: termination loses buffer contents
  • Low inversions: buffer allows reordering
  • High retransmit benefit: time for retransmits to arrive

Your buffer_size parameter and flush strategy determine this trade-off.

Gap Detection Strategy

When you notice missing sequence numbers (gaps), you have options:

  1. Wait: the packet may still arrive (reordering)
  2. Request retransmit: Ask for the packet again
  3. Accept loss: Write what you have, note the gap

One approach: if a gap persists for N packets or T time, request a retransmit. If the retransmit does not arrive within a bounded time, accept the loss.

Statistics

run() returns a LoggerStats for the process (see event_logger.py), and main() prints it at the end of the stream.

Evaluation

Your logger is evaluated on:

Coverage: What fraction of generated packets appear in your log? \[\text{coverage} = \frac{|\text{logged sequences} \cap \text{generated sequences}|}{|\text{generated sequences}|}\]

Ordering: What fraction of pairs are correctly ordered? \[\text{ordering} = 1 - \frac{\text{inversions}}{\binom{n}{2}}\]

where inversions counts pairs \((i, j)\) where \(i < j\) in ground truth but \(j\) appears before \(i\) in log.

Efficiency: Retransmit requests have a cost. Excessive requests indicate poor strategy.

These metrics have inherent tension—optimizing one often hurts another.

Testing Your Solution

From q3/, run the interface tests, then the logger to the end of the stream:

python -m pytest test_interfaces.py
python run_logger.py --seed 42 --buffer-size 30
python run_logger.py --seed 42 --buffer-size 5
python run_logger.py --seed 42 --buffer-size 100

run_logger.py reports the number of runs, coverage, ordering, and the count of each status. Run with several seeds and buffer sizes. Grading runs it with other seeds.

Requirements

  • Use only standard library modules
  • Log file must be append-only (never rewrite earlier content)
  • Do not catch SystemExit or use atexit — termination must behave as a crash
  • All timestamps must be from the packet’s original timestamp

Deliverables

See Submission. Your README should explain:

  • Your buffer size choice and reasoning
  • Your flush strategy (when do you write?)
  • Your gap handling approach
  • Trade-offs you observed during testing
TipSubmission 
README.md
q1/
├── merge_worker.py
└── README.md
q2/
├── http_client.py
└── README.md
q3/
├── event_logger.py
└── README.md