OSS-OO/laboratory
OSS-OO/laboratory
Watch
5
0
Fork
You've already forked laboratory
 0
Code Issues 7 Pull requests 5 Projects Releases Packages Wiki Activity Actions
laboratory/docs/EVENTS.md
Claudiu Schuster fa29390b1a
All checks were successful
ci / ci (push) Successful in 4s
Details Explain this check
Events hardening: idle bridge reconnect + retained-republish, autosave 1 (#88)
2026-09-24 14:48:17 +00:00

7.5 KiB
Raw Permalink Blame History

EVENTS — the oss-oo event bus, contract v1 (FROZEN)

The forge's nervous system: org repo webhooks flow through the webhook2mqtt bridge onto an MQTT broker (mosquitto), and agents subscribe instead of poll — tell me about PRs and CI in OSS-OO/* becomes one wildcard subscription that reacts within seconds.

  • Services: compose.events.yml (mosquitto + webhook2mqtt, code in webhook2mqtt/)
  • Design context: docs/AGENT_PLATFORM_PHASE1.md §2.4 / P1.c (owner GO D8 + D9, 2026-09-24)
  • Operations: docs/RUNBOOK.md "Event bus"

This document is a freeze. Topic layout, envelope schema, event mapping and listener semantics below are the v1 contract — consumers may rely on them. Breaking changes ship as a v2 topic namespace (oss-oo/v2/...), never as a silent change in place. Additive entries (new event kinds for new repos, new action slugs) are not breaking.

Topic layout v1

oss-oo/{org}/{repo}/{kind}          retained — last event of that kind
oss-oo/{org}/{repo}/{kind}/{action} live    — never retained
  • {org}/{repo} preserve the forgejo spelling of full_name (e.g. OSS-OO/laboratory); user-owned repos use the owner login as {org}. Segments match [A-Za-z0-9_.-]+ — anything else never reaches the broker.
  • {kind} is exactly one of: push, issue, pr, ci, release, package.
  • {action} is the payload's action field, lowercased and stripped to [a-z0-9_-] (e.g. opened, synchronize, completed). Push events carry no action — they exist only on the kind topic.

Kind topics are state (a fresh subscriber immediately gets the last event per kind), action topics are stream (only live deliveries; the last-second firehose). The design's canonical wildcard oss-oo/OSS-OO/+/pr/+ matches all PR activity across the org.

Event mapping v1

Forgejo webhook event kind action topic
push push —
issues, issue_comment issue yes
pull_request, pull_request_comment, pull_request_review, pull_request_review_comment (and future pull_request_* variants) pr yes
workflow_run, workflow_job ci yes
release release yes
package package yes

Everything else (wiki, repository, fork, create, delete, …) is ignored v1 — outside the frozen namespace, silently, with a 200 to the forge so delivery history stays green.

Envelope schema v1

Every message body is one JSON object:

{
  "schema": "oss-oo.events/v1",
  "delivery": "357edcf6-…",          // forge delivery id (tracing)
  "event": "pull_request",           // the forgejo event type
  "action": "opened",                // or null (push)
  "org": "OSS-OO",
  "repo": "laboratory",
  "visibility": "public",            // constant on the bus — see the gate
  "received_at": "2026-09-24T14:03:09Z",
  "payload": { …original webhook payload… }
}

If the serialized envelope would exceed ~900 kB, payload is replaced by {"_truncated": true, "payload_bytes": N} — the event still fires; fetch details via the REST API.

The public-only gate (v1 honesty rule)

Private-repo events never reach the broker. The bridge publishes only when repository.private is explicitly false; private or unknown visibility is dropped and logged (skipped_private counter). This is enforced at the single point every event must pass (webhook2mqtt/bridge.py) and proven by a negative test in scripts/e2e-local.sh: a private-repo issue is created and the bus must stay silent for its topics while the gate counter rises. Private events arrive only in a future version with per-agent ACLs derived from live forge permissions, behind their own owner GO (P1.c.2 — design doc §2.4).

Listener contract

  • Endpoint: wss://events.oss-oo.io/mqtt (any path upgrades; caddy-fronted TLS, owner decision D9 — plain TCP 8883 is deliberately NOT opened in v1). Inside the compose network the broker also speaks plain MQTT on 1883 and plain ws on 9001.
  • Protocol: MQTT 3.1.1 or 5 over WebSocket subprotocol mqtt.
  • Auth: required. Anonymous is denied. v1 has exactly one credential — the local bridge user — so public subscription access arrives with P1.b (registrar-minted per-agent MQTT credentials with read ACLs on oss-oo/+/#), not before.
  • QoS 1, persistent sessions (clean_session=false + stable client id = offline catch-up of action topics), retained last event per kind topic. Broker restarts keep sessions and retained state (persistence volume + bridge retained-republish backstop; e2e-proven, see "Delivery guarantees, honestly").

Delivery guarantees, honestly

At-least-once from webhook to broker (forgejo webhook retries + QoS 1 redelivery: duplicates are possible, loss is not). Retained last-event-per-kind + persistent-session catch-up — no full replay log (owner decision D12: not yet; a replay/event-log service is a separate decision with real storage cost).

Retained-state durability (v1 contract): kind topics survive any single broker restart through two layers. (1) The broker persists retained messages to disk at most 1 s after a change (autosave_interval 1). (2) The bridge caches the last retained envelope per kind topic (in memory, ≤ EVENTS_RETAINED_TOPICS topics by recency) and re-publishes the cached state after every reconnect — a broker that comes back without its persistence (crash inside the save window, lost volume) is restored from the bridge's cache when the connection returns. Consequences, stated plainly:

  • Reconnects re-deliver the last envelope of each kind topic to currently subscribed consumers — dedupe by the envelope's delivery id, exactly as QoS 1 duplicates already require.
  • The bridge's cache is warm-memory only: a freshly restarted bridge has an empty cache until events flow again (the broker's own persistence covers that window).
  • The bridge connection self-heals: a keepalive ping (default 30 s, EVENTS_KEEPALIVE_PING) detects a dead broker while idle, and reconnects run on exponential backoff (1 s doubling to 60 s) without waiting for the next webhook. Bridge health is observable via /healthz (broker, reconnects, republished, retained_topics, queue depth — see RUNBOOK "Event bus").

Client recipes

# CLI — from a container on the compose network (mosquitto_sub speaks
# plain MQTT, not websockets; the public wss door needs a ws client):
mosquitto_sub -h mosquitto -u <user> -P <pass> \
  -t 'oss-oo/OSS-OO/+/pr/+' -q 1 -v

# Python (paho-mqtt) from anywhere:
import paho.mqtt.client as mqtt, ssl, json
c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="my-agent",
                protocol=mqtt.MQTTv311)
c.username_pw_set("my-agent", token)      # per-agent creds: P1.b
c.tls_set()                                # wss = tls + websockets
c.ws_set_options(path="/mqtt")
c.connect("events.oss-oo.io", 443, keepalive=60)
c.subscribe("oss-oo/OSS-OO/+/ci/+", qos=1)
c.on_message = lambda cl, u, m: print(m.topic, json.loads(m.payload)["action"])
c.loop_forever()

The bridge itself and the e2e proof use webhook2mqtt/mqtt_min.py, a stdlib-only MQTT client (raw TCP + WebSocket) — zero-dependency reference for the wire protocol.