env-07-pubsub

# Publish/subscribe broker

Four nodes are provisioned on a private, no-internet network. Build a topic-based pub/sub system.

Nodes and the program each one runs:

- **broker** — create **`broker.py`** (launched as `python broker.py`). Listens on `$PORT` and keeps,
  per topic, the **ordered list** of messages published to it. Endpoints:
  - `POST /publish` with a JSON body `{"topic": "<name>", "message": "<text>"}` — append the message
    to that topic's list. Respond `200`.
  - `GET /messages?topic=<name>` — respond `200` with that topic's messages so far, **one per line**
    in publish order (empty body if the topic has none).

- **publisher** — create **`publisher.py`** (launched as `python publisher.py`). The broker is at
  `PEER_BROKER_HOST`/`PEER_BROKER_PORT`. Publish **exactly this sequence, in this order**:
  | topic | message |
  |-------|---------|
  | `alpha` | `one` |
  | `alpha` | `two` |
  | `alpha` | `three` |
  | `beta`  | `red` |
  | `beta`  | `green` |
  Then exit.

- **subscriber_a**, **subscriber_b** — identical subscribers. Create **`subscriber.py`** (launched as
  `python subscriber.py`) on each. A subscriber reads the topic it follows from a file `topic.txt`
  in its working directory (already present), then keeps its file `received.txt` up to date with that
  topic's messages from the broker — **one per line, in order**. (Poll the broker periodically; the
  publisher may not have sent everything yet.)

The task is solved when each subscriber's `received.txt` contains exactly its topic's messages in
publish order: `subscriber_a` (topic `alpha`) → `one`,`two`,`three`; `subscriber_b` (topic `beta`)
→ `red`,`green`.

Only the Python standard library is available (no internet, no pip). Retry connections that are
briefly refused while a peer is starting.

[turn 1] list_nodes({}) -> {"nodes": [{"name": "broker", "provides": "python broker.py", "ports": [9000], "needs": []}, {"name": "subscriber_a", "provides": "python subscriber.py", "ports": [], "needs": ["broker"]}, {"name": "s
[turn 2] run({"node": "broker", "cmd": "env | grep -E 'PORT|PEER'"}) -> {"rc": 0, "output": "PORT=48757\nPORTS=48757\n"}
[turn 3] run({"node": "publisher", "cmd": "env | grep -E 'PORT|PEER'"}) -> {"rc": 0, "output": "PEER_BROKER_HOST=127.0.0.1\nPEER_BROKER_PORT=48757\n"}
[turn 4] run({"node": "subscriber_a", "cmd": "env | grep -E 'PORT|PEER' && cat topic.txt 2>/dev/null || echo \"topic.txt not found\""}) -> {"rc": 0, "output": "PEER_BROKER_HOST=127.0.0.1\nPEER_BROKER_PORT=48757\nalpha"}
[turn 5] run({"node": "subscriber_b", "cmd": "env | grep -E 'PORT|PEER' && cat topic.txt 2>/dev/null || echo \"topic.txt not found\""}) -> {"rc": 0, "output": "PEER_BROKER_HOST=127.0.0.1\nPEER_BROKER_PORT=48757\nbeta"}
[turn 6] write_file({"node": "broker", "path": "broker.py", "content": "#!/usr/bin/env python3\nimport json\nimport os\nfrom http.server import HTTPServer, BaseHTTPRequestHandler\n\n# Store messages per topic in order\nt) -> {"ok": true, "node": "broker", "path": "broker.py"}
[turn 7] write_file({"node": "publisher", "path": "publisher.py", "content": "#!/usr/bin/env python3\nimport json\nimport os\nimport time\nimport urllib.request\nimport urllib.error\n\ndef publish(topic, message):\n    b) -> {"ok": true, "node": "publisher", "path": "publisher.py"}
[turn 8] write_file({"node": "subscriber_a", "path": "subscriber.py", "content": "#!/usr/bin/env python3\nimport os\nimport time\nimport urllib.request\n\ndef get_messages(topic):\n    broker_host = os.environ['PEER_BROK) -> {"ok": true, "node": "subscriber_a", "path": "subscriber.py"}
[turn 9] write_file({"node": "subscriber_b", "path": "subscriber.py", "content": "#!/usr/bin/env python3\nimport os\nimport time\nimport urllib.request\n\ndef get_messages(topic):\n    broker_host = os.environ['PEER_BROK) -> {"ok": true, "node": "subscriber_b", "path": "subscriber.py"}
[turn 10] verify({}) -> {"passed": false, "checks": [{"name": "subscriber_a received its topic's messages in order", "ok": false, "detail": "got [] want ['one', 'two', 'three']"}, {"name": "subscriber_b received its topic's 
[turn 11] run({"node": "broker", "cmd": "cat /tmp/node_logs/broker 2>/dev/null || echo \"No log file\""}) -> {"rc": 0, "output": "No log file\n"}
[turn 12] run({"node": "subscriber_a", "cmd": "ls -la && cat topic.txt && cat received.txt 2>/dev/null || echo \"received.txt not found\""}) -> {"rc": 0, "output": "total 20\ndrwxrwxr-x 2 psc psc 4096 Jul  2 14:51 .\ndrwx------ 6 psc psc 4096 Jul  2 14:50 ..\n-rw-rw-r-- 1 psc psc    0 Jul  2 14:51 .log-0\n-rw-rw-r-- 1 psc psc    1 Jul  2 14:5
[turn 13] run({"node": "broker", "cmd": "python broker.py &\nsleep 2\ncurl -X POST http://localhost:$PORT/publish -d '{\"topic\":\"alpha\",\"message\":\"test\"}' 2>/dev/null || echo \"curl failed\"\nps aux | grep p) -> {"rc": 0, "output": "      633449  0.0  0.0   2816  1912 ?        Ss   14:51   0:00 /bin/sh -c python subscriber.py\npsc       633450  0.1  0.0  35340 23112 ?        S    14:51   0:00 python subscribe
[turn 14] run({"node": "broker", "cmd": "pkill -f \"python broker.py\" 2>/dev/null; sleep 1; echo \"killed\""}) -> {"rc": -15, "output": ""}