feat(bot-mode): Group Chats survive the authority gateway dying — log replication and fenced takeover
Every participant gateway can now keep a durable copy of a hosted room's ordered log and continue the room when its authority host is gone: - gateway/hosted_room_replicas.py: replica store in root state.db. ingest_page() persists authority-stamped groups.log pages idempotently, refusing sequence gaps and authority-epoch regressions. promote_replica() continues the room locally at epoch+1 with a lineage-proving authority.claimed event; the stale owner is fenced everywhere the claim replicates. demote_room() lets a returning stale authority fence itself (authority.lost) upon observing a newer epoch, killing split-brain writes. - tui_gateway/methods_groups.py: groups.replicate / groups.replica_state / groups.promote / groups.demote RPC surface. Promotion requires confirm=true — storage decides HOW takeover is atomic and provable, the caller (user action now, lease/quorum driver later) decides WHEN it is safe, matching the boundary blessed on #97681. Validation: 20 new tests incl. a full failover round-trip (A hosts, B replicates incrementally, A dies, B promotes with complete history, A returns demoted and fenced); 69 total across the hosted-rooms area; E2E with two real gateway stores and real install identities.
This commit is contained in:
Teknium
@@ -0,0 +1,570 @@
|
||||
"""Replica store and takeover primitives for hosted Group Chat rooms.
|
||||
|
||||
The authority gateway owns a room's ordered log in ``gateway/hosted_rooms.py``.
|
||||
This module gives every OTHER participant gateway a durable local copy of that
|
||||
log, and the fenced primitives to continue the room when the authority host
|
||||
dies:
|
||||
|
||||
- ``ingest_page()`` persists replay pages (``groups.log`` output, which carries
|
||||
the room's authority stamp) idempotently, refusing sequence gaps and
|
||||
authority-epoch regressions.
|
||||
- ``promote_replica()`` instantiates the replicated log as a locally-owned
|
||||
hosted room at ``epoch + 1`` with a lineage-proving ``authority.claimed``
|
||||
event, so a surviving participant can resume the room.
|
||||
- ``demote_room()`` fences a returning stale authority: presented with proof of
|
||||
a newer epoch, the local room records ``authority.lost`` and stops being
|
||||
authoritative.
|
||||
|
||||
Storage primitives only: none of these decide *when* takeover is safe. The
|
||||
caller (an explicit user action today; a lease/quorum driver later) must
|
||||
establish that the previous owner can no longer commit before promoting.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from gateway.hosted_rooms import (
|
||||
MAX_ACTOR_ID_CHARS,
|
||||
MAX_EVENT_JSON_BYTES,
|
||||
MAX_ROOM_ID_CHARS,
|
||||
HostedRoomError,
|
||||
RoomConflictError,
|
||||
_canonical_json,
|
||||
_connect,
|
||||
_transaction,
|
||||
_validate_identifier,
|
||||
_validate_members,
|
||||
_validate_room_name,
|
||||
local_authority_gateway_id,
|
||||
)
|
||||
|
||||
MAX_REPLICA_ROOMS = 256
|
||||
MAX_REPLICA_EVENT_BYTES = 256 * 1024 * 1024
|
||||
|
||||
|
||||
class ReplicaError(HostedRoomError):
|
||||
"""Base class for invalid or conflicting replica operations."""
|
||||
|
||||
|
||||
class ReplicaGapError(ReplicaError):
|
||||
"""A page does not start at the replica's next expected sequence."""
|
||||
|
||||
|
||||
class ReplicaEpochRegressionError(ReplicaError):
|
||||
"""A page or demotion carries an older authority epoch than stored."""
|
||||
|
||||
|
||||
def _initialize_replica_schema(conn: sqlite3.Connection) -> None:
|
||||
conn.execute(
|
||||
"""CREATE TABLE IF NOT EXISTS hosted_room_replicas (
|
||||
room_id TEXT PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
members_json TEXT NOT NULL,
|
||||
authority_gateway_id TEXT NOT NULL,
|
||||
authority_epoch INTEGER NOT NULL CHECK (authority_epoch >= 1),
|
||||
last_seq INTEGER NOT NULL DEFAULT 0 CHECK (last_seq >= 0),
|
||||
latest_seq INTEGER NOT NULL DEFAULT 0,
|
||||
event_bytes INTEGER NOT NULL DEFAULT 0,
|
||||
created_at REAL NOT NULL,
|
||||
updated_at REAL NOT NULL
|
||||
)"""
|
||||
)
|
||||
conn.execute(
|
||||
"""CREATE TABLE IF NOT EXISTS hosted_room_replica_events (
|
||||
room_id TEXT NOT NULL,
|
||||
seq INTEGER NOT NULL CHECK (seq >= 1),
|
||||
event_id TEXT NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
actor_json TEXT NOT NULL,
|
||||
authority_epoch INTEGER,
|
||||
payload_json TEXT NOT NULL,
|
||||
created_at REAL NOT NULL,
|
||||
PRIMARY KEY (room_id, seq)
|
||||
)"""
|
||||
)
|
||||
|
||||
|
||||
def _replica_transaction(db_path: Path | str):
|
||||
return _transaction(db_path, immediate=True)
|
||||
|
||||
|
||||
def _ensure_schema(db_path: Path | str) -> None:
|
||||
conn = _connect(db_path)
|
||||
try:
|
||||
with conn:
|
||||
_initialize_replica_schema(conn)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def _event_bytes(event: dict[str, Any]) -> int:
|
||||
return (
|
||||
len(str(event["event_id"]).encode("utf-8"))
|
||||
+ len(str(event["kind"]).encode("utf-8"))
|
||||
+ len(
|
||||
json.dumps(
|
||||
event["actor"], ensure_ascii=False, separators=(",", ":")
|
||||
).encode("utf-8")
|
||||
)
|
||||
+ len(
|
||||
json.dumps(
|
||||
event["payload"], ensure_ascii=False, separators=(",", ":")
|
||||
).encode("utf-8")
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _validate_page(page: Any) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
||||
if not isinstance(page, dict):
|
||||
raise ReplicaError("page must be an object")
|
||||
events = page.get("events")
|
||||
authority = page.get("authority")
|
||||
if not isinstance(events, list):
|
||||
raise ReplicaError("page.events must be a list")
|
||||
if not isinstance(authority, dict):
|
||||
raise ReplicaError("page.authority is required for replication")
|
||||
gateway_id = _validate_identifier(
|
||||
authority.get("gateway_id"),
|
||||
label="page.authority.gateway_id",
|
||||
max_chars=MAX_ACTOR_ID_CHARS,
|
||||
)
|
||||
epoch = authority.get("epoch")
|
||||
if isinstance(epoch, bool) or not isinstance(epoch, int) or epoch < 1:
|
||||
raise ReplicaError("page.authority.epoch must be a positive integer")
|
||||
previous_seq: int | None = None
|
||||
for event in events:
|
||||
if not isinstance(event, dict):
|
||||
raise ReplicaError("page events must be objects")
|
||||
seq = event.get("seq")
|
||||
if isinstance(seq, bool) or not isinstance(seq, int) or seq < 1:
|
||||
raise ReplicaError("event.seq must be a positive integer")
|
||||
if previous_seq is not None and seq != previous_seq + 1:
|
||||
raise ReplicaGapError("page events must be contiguous")
|
||||
previous_seq = seq
|
||||
for field in ("event_id", "kind"):
|
||||
if not isinstance(event.get(field), str) or not event[field]:
|
||||
raise ReplicaError(f"event.{field} must be a non-empty string")
|
||||
if not isinstance(event.get("actor"), dict):
|
||||
raise ReplicaError("event.actor must be an object")
|
||||
if "payload" not in event:
|
||||
raise ReplicaError("event.payload is required")
|
||||
return events, {"gateway_id": gateway_id, "epoch": epoch}
|
||||
|
||||
|
||||
def ingest_page(
|
||||
db_path: Path | str,
|
||||
*,
|
||||
room_id: Any,
|
||||
room_name: Any,
|
||||
members: Any,
|
||||
page: Any,
|
||||
now: float | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Persist one replay page for ``room_id``; idempotent, gap- and
|
||||
epoch-regression-safe.
|
||||
|
||||
``page`` is the verbatim result of the authority's ``groups.log`` call
|
||||
(``read_events()``), whose ``authority`` stamp proves lineage.
|
||||
"""
|
||||
room_id = _validate_identifier(
|
||||
room_id, label="room_id", max_chars=MAX_ROOM_ID_CHARS
|
||||
)
|
||||
room_name = _validate_room_name(room_name)
|
||||
_, members_json = _validate_members(members)
|
||||
events, authority = _validate_page(page)
|
||||
now = time.time() if now is None else float(now)
|
||||
_ensure_schema(db_path)
|
||||
|
||||
with _replica_transaction(db_path) as conn:
|
||||
_initialize_replica_schema(conn)
|
||||
row = conn.execute(
|
||||
"""SELECT authority_gateway_id, authority_epoch, last_seq,
|
||||
latest_seq, event_bytes
|
||||
FROM hosted_room_replicas WHERE room_id=?""",
|
||||
(room_id,),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
count = conn.execute(
|
||||
"SELECT COUNT(*) FROM hosted_room_replicas"
|
||||
).fetchone()[0]
|
||||
if int(count) >= MAX_REPLICA_ROOMS:
|
||||
raise ReplicaError("replica room capacity exhausted")
|
||||
stored_epoch = 0
|
||||
last_seq = 0
|
||||
stored_bytes = 0
|
||||
else:
|
||||
stored_epoch = int(row["authority_epoch"])
|
||||
last_seq = int(row["last_seq"])
|
||||
stored_bytes = int(row["event_bytes"])
|
||||
|
||||
if authority["epoch"] < stored_epoch:
|
||||
raise ReplicaEpochRegressionError(
|
||||
"page authority epoch is older than the stored replica epoch"
|
||||
)
|
||||
|
||||
new_events = [e for e in events if int(e["seq"]) > last_seq]
|
||||
if new_events and int(new_events[0]["seq"]) != last_seq + 1:
|
||||
raise ReplicaGapError(
|
||||
"page skips sequences the replica has not stored"
|
||||
)
|
||||
added_bytes = 0
|
||||
for event in new_events:
|
||||
size = _event_bytes(event)
|
||||
if stored_bytes + added_bytes + size > MAX_REPLICA_EVENT_BYTES:
|
||||
raise ReplicaError("replica event storage exhausted")
|
||||
actor_json = _canonical_json(
|
||||
event["actor"], label="actor", max_bytes=4 * 1024
|
||||
)
|
||||
payload_json = _canonical_json(
|
||||
event["payload"], label="payload", max_bytes=MAX_EVENT_JSON_BYTES
|
||||
)
|
||||
epoch_value = event.get("authority_epoch")
|
||||
conn.execute(
|
||||
"""INSERT INTO hosted_room_replica_events
|
||||
(room_id, seq, event_id, kind, actor_json, authority_epoch,
|
||||
payload_json, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
||||
(
|
||||
room_id,
|
||||
int(event["seq"]),
|
||||
event["event_id"],
|
||||
event["kind"],
|
||||
actor_json,
|
||||
epoch_value,
|
||||
payload_json,
|
||||
float(event.get("created_at") or now),
|
||||
),
|
||||
)
|
||||
added_bytes += size
|
||||
new_last = int(new_events[-1]["seq"]) if new_events else last_seq
|
||||
latest_seq = page.get("latest_seq")
|
||||
if isinstance(latest_seq, bool) or not isinstance(latest_seq, int):
|
||||
latest_seq = new_last
|
||||
if row is None:
|
||||
conn.execute(
|
||||
"""INSERT INTO hosted_room_replicas
|
||||
(room_id, name, members_json, authority_gateway_id,
|
||||
authority_epoch, last_seq, latest_seq, event_bytes,
|
||||
created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
||||
(
|
||||
room_id,
|
||||
room_name,
|
||||
members_json,
|
||||
authority["gateway_id"],
|
||||
authority["epoch"],
|
||||
new_last,
|
||||
max(latest_seq, new_last),
|
||||
added_bytes,
|
||||
now,
|
||||
now,
|
||||
),
|
||||
)
|
||||
else:
|
||||
conn.execute(
|
||||
"""UPDATE hosted_room_replicas
|
||||
SET name=?, members_json=?, authority_gateway_id=?,
|
||||
authority_epoch=?, last_seq=?, latest_seq=?,
|
||||
event_bytes=event_bytes+?, updated_at=?
|
||||
WHERE room_id=?""",
|
||||
(
|
||||
room_name,
|
||||
members_json,
|
||||
authority["gateway_id"],
|
||||
authority["epoch"],
|
||||
new_last,
|
||||
max(latest_seq, new_last),
|
||||
added_bytes,
|
||||
now,
|
||||
room_id,
|
||||
),
|
||||
)
|
||||
return {
|
||||
"room_id": room_id,
|
||||
"stored_seq": new_last,
|
||||
"ingested": len(new_events),
|
||||
"authority": authority,
|
||||
"caught_up": new_last >= max(latest_seq, new_last),
|
||||
}
|
||||
|
||||
|
||||
def replica_state(db_path: Path | str, *, room_id: Any) -> dict[str, Any]:
|
||||
"""Return the stored replica's coverage and authority lineage."""
|
||||
room_id = _validate_identifier(
|
||||
room_id, label="room_id", max_chars=MAX_ROOM_ID_CHARS
|
||||
)
|
||||
_ensure_schema(db_path)
|
||||
with _replica_transaction(db_path) as conn:
|
||||
_initialize_replica_schema(conn)
|
||||
row = conn.execute(
|
||||
"""SELECT room_id, name, members_json, authority_gateway_id,
|
||||
authority_epoch, last_seq, latest_seq, event_bytes,
|
||||
created_at, updated_at
|
||||
FROM hosted_room_replicas WHERE room_id=?""",
|
||||
(room_id,),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
raise ReplicaError("replica not found")
|
||||
return {
|
||||
"room_id": row["room_id"],
|
||||
"name": row["name"],
|
||||
"members": json.loads(row["members_json"]),
|
||||
"authority": {
|
||||
"gateway_id": row["authority_gateway_id"],
|
||||
"epoch": int(row["authority_epoch"]),
|
||||
},
|
||||
"last_seq": int(row["last_seq"]),
|
||||
"latest_seq": int(row["latest_seq"]),
|
||||
"event_bytes": int(row["event_bytes"]),
|
||||
"created_at": float(row["created_at"]),
|
||||
"updated_at": float(row["updated_at"]),
|
||||
}
|
||||
|
||||
|
||||
def promote_replica(
|
||||
db_path: Path | str,
|
||||
*,
|
||||
room_id: Any,
|
||||
reason: Any = "authority-unreachable",
|
||||
now: float | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Continue a replicated room on THIS gateway at ``epoch + 1``.
|
||||
|
||||
Copies the replica's log into the authoritative store, appends a lineage-
|
||||
proving ``authority.claimed`` event, and returns the new room state. The
|
||||
old authority is fenced everywhere the claim replicates: its epoch is now
|
||||
stale and every fenced primitive rejects it.
|
||||
|
||||
The caller decides that takeover is safe (the previous owner can no longer
|
||||
commit). This primitive only makes the takeover atomic and provable.
|
||||
"""
|
||||
room_id = _validate_identifier(
|
||||
room_id, label="room_id", max_chars=MAX_ROOM_ID_CHARS
|
||||
)
|
||||
if not isinstance(reason, str) or not reason or len(reason) > 200:
|
||||
raise ReplicaError("reason must be a non-empty string of at most 200 chars")
|
||||
now = time.time() if now is None else float(now)
|
||||
local_gateway = local_authority_gateway_id()
|
||||
_ensure_schema(db_path)
|
||||
|
||||
with _replica_transaction(db_path) as conn:
|
||||
_initialize_replica_schema(conn)
|
||||
replica = conn.execute(
|
||||
"""SELECT room_id, name, members_json, authority_gateway_id,
|
||||
authority_epoch, last_seq, event_bytes
|
||||
FROM hosted_room_replicas WHERE room_id=?""",
|
||||
(room_id,),
|
||||
).fetchone()
|
||||
if replica is None:
|
||||
raise ReplicaError("replica not found")
|
||||
if replica["authority_gateway_id"] == local_gateway:
|
||||
raise ReplicaError("this gateway already holds the room authority")
|
||||
if conn.execute(
|
||||
"SELECT 1 FROM hosted_rooms WHERE room_id=?", (room_id,)
|
||||
).fetchone():
|
||||
raise RoomConflictError(
|
||||
"room_id already exists in the local authoritative store"
|
||||
)
|
||||
if conn.execute(
|
||||
"SELECT 1 FROM hosted_room_retired_ids WHERE room_id=?",
|
||||
(room_id,),
|
||||
).fetchone():
|
||||
raise RoomConflictError("room_id belongs to a disbanded room")
|
||||
|
||||
previous_gateway = str(replica["authority_gateway_id"])
|
||||
previous_epoch = int(replica["authority_epoch"])
|
||||
target_epoch = previous_epoch + 1
|
||||
last_seq = int(replica["last_seq"])
|
||||
claim_seq = last_seq + 1
|
||||
claim_event_id = f"system:authority-claimed:{target_epoch}"
|
||||
claim_actor_json = _canonical_json(
|
||||
{"kind": "system", "id": "authority-control"},
|
||||
label="actor",
|
||||
max_bytes=4 * 1024,
|
||||
)
|
||||
claim_payload_json = _canonical_json(
|
||||
{
|
||||
"previous_gateway_id": previous_gateway,
|
||||
"authority_gateway_id": local_gateway,
|
||||
"authority_epoch": target_epoch,
|
||||
"promoted_from_replica": True,
|
||||
"reason": reason,
|
||||
},
|
||||
label="payload",
|
||||
max_bytes=MAX_EVENT_JSON_BYTES,
|
||||
)
|
||||
claim_bytes = (
|
||||
len(claim_event_id.encode("utf-8"))
|
||||
+ len(b"authority.claimed")
|
||||
+ len(claim_actor_json.encode("utf-8"))
|
||||
+ len(claim_payload_json.encode("utf-8"))
|
||||
)
|
||||
|
||||
conn.execute(
|
||||
"""INSERT INTO hosted_rooms
|
||||
(room_id, name, members_json, authority_gateway_id,
|
||||
authority_epoch, next_seq, event_bytes, revision,
|
||||
created_at, updated_at, disbanded_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, 1, ?, ?, NULL)""",
|
||||
(
|
||||
room_id,
|
||||
replica["name"],
|
||||
replica["members_json"],
|
||||
local_gateway,
|
||||
target_epoch,
|
||||
claim_seq + 1,
|
||||
int(replica["event_bytes"]) + claim_bytes,
|
||||
now,
|
||||
now,
|
||||
),
|
||||
)
|
||||
conn.execute(
|
||||
"""INSERT INTO hosted_room_events
|
||||
(room_id, seq, event_id, kind, actor_json, authority_epoch,
|
||||
payload_json, created_at)
|
||||
SELECT room_id, seq, event_id, kind, actor_json,
|
||||
authority_epoch, payload_json, created_at
|
||||
FROM hosted_room_replica_events WHERE room_id=?""",
|
||||
(room_id,),
|
||||
)
|
||||
conn.execute(
|
||||
"""INSERT INTO hosted_room_events
|
||||
(room_id, seq, event_id, kind, actor_json, authority_epoch,
|
||||
payload_json, created_at)
|
||||
VALUES (?, ?, ?, 'authority.claimed', ?, ?, ?, ?)""",
|
||||
(
|
||||
room_id,
|
||||
claim_seq,
|
||||
claim_event_id,
|
||||
claim_actor_json,
|
||||
target_epoch,
|
||||
claim_payload_json,
|
||||
now,
|
||||
),
|
||||
)
|
||||
conn.execute(
|
||||
"DELETE FROM hosted_room_replica_events WHERE room_id=?", (room_id,)
|
||||
)
|
||||
conn.execute(
|
||||
"DELETE FROM hosted_room_replicas WHERE room_id=?", (room_id,)
|
||||
)
|
||||
return {
|
||||
"room_id": room_id,
|
||||
"authority_gateway_id": local_gateway,
|
||||
"authority_epoch": target_epoch,
|
||||
"previous_gateway_id": previous_gateway,
|
||||
"previous_epoch": previous_epoch,
|
||||
"claim_seq": claim_seq,
|
||||
"latest_seq": claim_seq,
|
||||
}
|
||||
|
||||
|
||||
def demote_room(
|
||||
db_path: Path | str,
|
||||
*,
|
||||
room_id: Any,
|
||||
observed_gateway_id: Any,
|
||||
observed_epoch: Any,
|
||||
now: float | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Fence THIS gateway's stale room authority against a proven newer epoch.
|
||||
|
||||
Called when a returning gateway observes (via a replicated
|
||||
``authority.claimed`` event or a transport rejection) that another gateway
|
||||
now owns the room at a higher epoch. Appends ``authority.lost`` and adopts
|
||||
the observed lineage so no further local sends can be committed at the
|
||||
stale epoch. Idempotent for repeated observations of the same lineage.
|
||||
"""
|
||||
room_id = _validate_identifier(
|
||||
room_id, label="room_id", max_chars=MAX_ROOM_ID_CHARS
|
||||
)
|
||||
observed_gateway_id = _validate_identifier(
|
||||
observed_gateway_id,
|
||||
label="observed_gateway_id",
|
||||
max_chars=MAX_ACTOR_ID_CHARS,
|
||||
)
|
||||
if (
|
||||
isinstance(observed_epoch, bool)
|
||||
or not isinstance(observed_epoch, int)
|
||||
or observed_epoch < 1
|
||||
):
|
||||
raise ReplicaError("observed_epoch must be a positive integer")
|
||||
now = time.time() if now is None else float(now)
|
||||
local_gateway = local_authority_gateway_id()
|
||||
|
||||
with _replica_transaction(db_path) as conn:
|
||||
row = conn.execute(
|
||||
"""SELECT authority_gateway_id, authority_epoch, next_seq
|
||||
FROM hosted_rooms WHERE room_id=? AND disbanded_at IS NULL""",
|
||||
(room_id,),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
raise ReplicaError("room not found in the local authoritative store")
|
||||
current_gateway = str(row["authority_gateway_id"])
|
||||
current_epoch = int(row["authority_epoch"])
|
||||
if (
|
||||
current_gateway == observed_gateway_id
|
||||
and current_epoch == observed_epoch
|
||||
):
|
||||
return {
|
||||
"room_id": room_id,
|
||||
"authority_gateway_id": current_gateway,
|
||||
"authority_epoch": current_epoch,
|
||||
"idempotent": True,
|
||||
}
|
||||
if observed_epoch <= current_epoch:
|
||||
raise ReplicaEpochRegressionError(
|
||||
"observed epoch does not supersede the stored authority"
|
||||
)
|
||||
if current_gateway != local_gateway:
|
||||
raise ReplicaError(
|
||||
"room is not locally authoritative; nothing to demote"
|
||||
)
|
||||
seq = int(row["next_seq"])
|
||||
lost_actor_json = _canonical_json(
|
||||
{"kind": "system", "id": "authority-control"},
|
||||
label="actor",
|
||||
max_bytes=4 * 1024,
|
||||
)
|
||||
lost_payload_json = _canonical_json(
|
||||
{
|
||||
"previous_gateway_id": current_gateway,
|
||||
"authority_gateway_id": observed_gateway_id,
|
||||
"authority_epoch": observed_epoch,
|
||||
},
|
||||
label="payload",
|
||||
max_bytes=MAX_EVENT_JSON_BYTES,
|
||||
)
|
||||
conn.execute(
|
||||
"""INSERT INTO hosted_room_events
|
||||
(room_id, seq, event_id, kind, actor_json, authority_epoch,
|
||||
payload_json, created_at)
|
||||
VALUES (?, ?, ?, 'authority.lost', ?, ?, ?, ?)""",
|
||||
(
|
||||
room_id,
|
||||
seq,
|
||||
f"system:authority-lost:{observed_epoch}",
|
||||
lost_actor_json,
|
||||
observed_epoch,
|
||||
lost_payload_json,
|
||||
now,
|
||||
),
|
||||
)
|
||||
conn.execute(
|
||||
"""UPDATE hosted_rooms
|
||||
SET authority_gateway_id=?, authority_epoch=?,
|
||||
next_seq=next_seq+1, revision=revision+1, updated_at=?
|
||||
WHERE room_id=?""",
|
||||
(observed_gateway_id, observed_epoch, now, room_id),
|
||||
)
|
||||
return {
|
||||
"room_id": room_id,
|
||||
"authority_gateway_id": observed_gateway_id,
|
||||
"authority_epoch": observed_epoch,
|
||||
"idempotent": False,
|
||||
}
|
||||
@@ -0,0 +1,296 @@
|
||||
"""Tests for gateway/hosted_room_replicas.py — replica ingest, promotion, and
|
||||
stale-authority demotion for hosted Group Chat rooms."""
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
import gateway.hosted_room_replicas as replicas
|
||||
import gateway.hosted_rooms as rooms
|
||||
|
||||
USER = {"kind": "user", "id": "tek"}
|
||||
MEMBERS = [{"kind": "bot", "id": "planner"}, {"kind": "bot", "id": "coder"}]
|
||||
|
||||
AUTH_A = "install:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
||||
AUTH_B = "install:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
|
||||
|
||||
|
||||
def _authority_db(tmp_path, name="authority.db"):
|
||||
return tmp_path / name
|
||||
|
||||
|
||||
def _replica_db(tmp_path, name="replica.db"):
|
||||
return tmp_path / name
|
||||
|
||||
|
||||
def _seed_room(db, *, gateway_id=AUTH_A, n_events=3, room_id="room-1"):
|
||||
rooms.create_room(
|
||||
db,
|
||||
room_id=room_id,
|
||||
name="Field Room",
|
||||
members=MEMBERS,
|
||||
authority_gateway_id=gateway_id,
|
||||
)
|
||||
for index in range(n_events):
|
||||
rooms.append_event(
|
||||
db,
|
||||
room_id=room_id,
|
||||
event_id=f"e{index}",
|
||||
kind="message.user",
|
||||
actor=USER,
|
||||
payload={"text": f"msg {index} 😀"},
|
||||
authority_gateway_id=gateway_id,
|
||||
authority_epoch=1,
|
||||
)
|
||||
return rooms.read_events(db, room_id=room_id, since_seq=0, limit=100)
|
||||
|
||||
|
||||
def test_ingest_page_persists_events_and_lineage(tmp_path):
|
||||
page = _seed_room(_authority_db(tmp_path))
|
||||
rdb = _replica_db(tmp_path)
|
||||
result = replicas.ingest_page(
|
||||
rdb, room_id="room-1", room_name="Field Room", members=MEMBERS, page=page
|
||||
)
|
||||
assert result["ingested"] == 3
|
||||
assert result["stored_seq"] == 3
|
||||
assert result["caught_up"] is True
|
||||
state = replicas.replica_state(rdb, room_id="room-1")
|
||||
assert state["last_seq"] == 3
|
||||
assert state["authority"] == page["authority"]
|
||||
assert state["members"] == MEMBERS
|
||||
|
||||
|
||||
def test_ingest_page_is_idempotent(tmp_path):
|
||||
page = _seed_room(_authority_db(tmp_path))
|
||||
rdb = _replica_db(tmp_path)
|
||||
replicas.ingest_page(
|
||||
rdb, room_id="room-1", room_name="Field Room", members=MEMBERS, page=page
|
||||
)
|
||||
again = replicas.ingest_page(
|
||||
rdb, room_id="room-1", room_name="Field Room", members=MEMBERS, page=page
|
||||
)
|
||||
assert again["ingested"] == 0
|
||||
assert again["stored_seq"] == 3
|
||||
|
||||
|
||||
def test_ingest_rejects_sequence_gap(tmp_path):
|
||||
adb = _authority_db(tmp_path)
|
||||
_seed_room(adb, n_events=5)
|
||||
later = rooms.read_events(adb, room_id="room-1", since_seq=2, limit=100)
|
||||
rdb = _replica_db(tmp_path)
|
||||
with pytest.raises(replicas.ReplicaGapError):
|
||||
replicas.ingest_page(
|
||||
rdb,
|
||||
room_id="room-1",
|
||||
room_name="Field Room",
|
||||
members=MEMBERS,
|
||||
page=later,
|
||||
)
|
||||
|
||||
|
||||
def test_ingest_rejects_epoch_regression(tmp_path):
|
||||
page = _seed_room(_authority_db(tmp_path))
|
||||
rdb = _replica_db(tmp_path)
|
||||
newer = json.loads(json.dumps(page))
|
||||
newer["authority"]["epoch"] = 3
|
||||
replicas.ingest_page(
|
||||
rdb, room_id="room-1", room_name="Field Room", members=MEMBERS, page=newer
|
||||
)
|
||||
stale = json.loads(json.dumps(page))
|
||||
stale["authority"]["epoch"] = 2
|
||||
with pytest.raises(replicas.ReplicaEpochRegressionError):
|
||||
replicas.ingest_page(
|
||||
rdb,
|
||||
room_id="room-1",
|
||||
room_name="Field Room",
|
||||
members=MEMBERS,
|
||||
page=stale,
|
||||
)
|
||||
|
||||
|
||||
def test_ingest_requires_authority_stamp(tmp_path):
|
||||
page = _seed_room(_authority_db(tmp_path))
|
||||
page.pop("authority")
|
||||
with pytest.raises(replicas.ReplicaError):
|
||||
replicas.ingest_page(
|
||||
_replica_db(tmp_path),
|
||||
room_id="room-1",
|
||||
room_name="Field Room",
|
||||
members=MEMBERS,
|
||||
page=page,
|
||||
)
|
||||
|
||||
|
||||
def test_promote_replica_continues_room_at_next_epoch(tmp_path, monkeypatch):
|
||||
page = _seed_room(_authority_db(tmp_path))
|
||||
rdb = _replica_db(tmp_path)
|
||||
replicas.ingest_page(
|
||||
rdb, room_id="room-1", room_name="Field Room", members=MEMBERS, page=page
|
||||
)
|
||||
monkeypatch.setattr(replicas, "local_authority_gateway_id", lambda: AUTH_B)
|
||||
|
||||
promoted = replicas.promote_replica(rdb, room_id="room-1")
|
||||
assert promoted["authority_gateway_id"] == AUTH_B
|
||||
assert promoted["authority_epoch"] == 2
|
||||
assert promoted["previous_gateway_id"] == AUTH_A
|
||||
assert promoted["claim_seq"] == 4
|
||||
|
||||
# The room is now locally authoritative with the full history + claim.
|
||||
replay = rooms.read_events(rdb, room_id="room-1", since_seq=0, limit=100)
|
||||
assert [e["seq"] for e in replay["events"]] == [1, 2, 3, 4]
|
||||
claim = replay["events"][-1]
|
||||
assert claim["kind"] == "authority.claimed"
|
||||
assert claim["payload"]["previous_gateway_id"] == AUTH_A
|
||||
assert claim["payload"]["authority_epoch"] == 2
|
||||
assert replay["authority"] == {"gateway_id": AUTH_B, "epoch": 2}
|
||||
|
||||
# New work continues under the new epoch.
|
||||
rooms.append_event(
|
||||
rdb,
|
||||
room_id="room-1",
|
||||
event_id="post-takeover",
|
||||
kind="message.user",
|
||||
actor=USER,
|
||||
payload={"text": "continuing"},
|
||||
authority_gateway_id=AUTH_B,
|
||||
authority_epoch=2,
|
||||
)
|
||||
|
||||
# The old authority's identity/epoch is fenced out.
|
||||
with pytest.raises(rooms.HostedRoomError):
|
||||
rooms.append_event(
|
||||
rdb,
|
||||
room_id="room-1",
|
||||
event_id="stale-write",
|
||||
kind="message.user",
|
||||
actor=USER,
|
||||
payload={"text": "stale"},
|
||||
authority_gateway_id=AUTH_A,
|
||||
authority_epoch=1,
|
||||
)
|
||||
|
||||
# Replica bookkeeping is consumed by promotion.
|
||||
with pytest.raises(replicas.ReplicaError):
|
||||
replicas.replica_state(rdb, room_id="room-1")
|
||||
|
||||
|
||||
def test_promote_refuses_when_room_exists_locally(tmp_path, monkeypatch):
|
||||
db = _authority_db(tmp_path)
|
||||
page = _seed_room(db)
|
||||
# Same DB also holds a replica row for the same id — conflict must win.
|
||||
replicas.ingest_page(
|
||||
db, room_id="room-1", room_name="Field Room", members=MEMBERS, page=page
|
||||
)
|
||||
monkeypatch.setattr(replicas, "local_authority_gateway_id", lambda: AUTH_B)
|
||||
with pytest.raises(rooms.RoomConflictError):
|
||||
replicas.promote_replica(db, room_id="room-1")
|
||||
|
||||
|
||||
def test_promote_refuses_when_already_authority(tmp_path, monkeypatch):
|
||||
page = _seed_room(_authority_db(tmp_path))
|
||||
rdb = _replica_db(tmp_path)
|
||||
replicas.ingest_page(
|
||||
rdb, room_id="room-1", room_name="Field Room", members=MEMBERS, page=page
|
||||
)
|
||||
monkeypatch.setattr(replicas, "local_authority_gateway_id", lambda: AUTH_A)
|
||||
with pytest.raises(replicas.ReplicaError):
|
||||
replicas.promote_replica(rdb, room_id="room-1")
|
||||
|
||||
|
||||
def test_demote_fences_stale_local_authority(tmp_path, monkeypatch):
|
||||
adb = _authority_db(tmp_path)
|
||||
_seed_room(adb)
|
||||
monkeypatch.setattr(replicas, "local_authority_gateway_id", lambda: AUTH_A)
|
||||
|
||||
result = replicas.demote_room(
|
||||
adb, room_id="room-1", observed_gateway_id=AUTH_B, observed_epoch=2
|
||||
)
|
||||
assert result["idempotent"] is False
|
||||
assert result["authority_gateway_id"] == AUTH_B
|
||||
assert result["authority_epoch"] == 2
|
||||
|
||||
replay = rooms.read_events(adb, room_id="room-1", since_seq=0, limit=100)
|
||||
lost = replay["events"][-1]
|
||||
assert lost["kind"] == "authority.lost"
|
||||
assert lost["payload"]["authority_gateway_id"] == AUTH_B
|
||||
assert replay["authority"] == {"gateway_id": AUTH_B, "epoch": 2}
|
||||
|
||||
# Local sends at the stale identity/epoch are now rejected.
|
||||
with pytest.raises(rooms.HostedRoomError):
|
||||
rooms.append_event(
|
||||
adb,
|
||||
room_id="room-1",
|
||||
event_id="after-demote",
|
||||
kind="message.user",
|
||||
actor=USER,
|
||||
payload={"text": "stale"},
|
||||
authority_gateway_id=AUTH_A,
|
||||
authority_epoch=1,
|
||||
)
|
||||
|
||||
# Repeating the same observation is idempotent.
|
||||
again = replicas.demote_room(
|
||||
adb, room_id="room-1", observed_gateway_id=AUTH_B, observed_epoch=2
|
||||
)
|
||||
assert again["idempotent"] is True
|
||||
|
||||
|
||||
def test_demote_rejects_non_superseding_epoch(tmp_path, monkeypatch):
|
||||
adb = _authority_db(tmp_path)
|
||||
_seed_room(adb)
|
||||
monkeypatch.setattr(replicas, "local_authority_gateway_id", lambda: AUTH_A)
|
||||
with pytest.raises(replicas.ReplicaEpochRegressionError):
|
||||
replicas.demote_room(
|
||||
adb, room_id="room-1", observed_gateway_id=AUTH_B, observed_epoch=1
|
||||
)
|
||||
|
||||
|
||||
def test_full_failover_round_trip(tmp_path, monkeypatch):
|
||||
"""Authority A hosts, replica B follows, A dies, B promotes, A returns
|
||||
and is fenced + demoted; the room's history survives intact throughout."""
|
||||
adb = _authority_db(tmp_path)
|
||||
rdb = _replica_db(tmp_path)
|
||||
page = _seed_room(adb, n_events=4)
|
||||
replicas.ingest_page(
|
||||
rdb, room_id="room-1", room_name="Field Room", members=MEMBERS, page=page
|
||||
)
|
||||
|
||||
# A "dies"; B takes over.
|
||||
monkeypatch.setattr(replicas, "local_authority_gateway_id", lambda: AUTH_B)
|
||||
promoted = replicas.promote_replica(rdb, room_id="room-1")
|
||||
rooms.append_event(
|
||||
rdb,
|
||||
room_id="room-1",
|
||||
event_id="b-work",
|
||||
kind="message.user",
|
||||
actor=USER,
|
||||
payload={"text": "work continues on B"},
|
||||
authority_gateway_id=AUTH_B,
|
||||
authority_epoch=promoted["authority_epoch"],
|
||||
)
|
||||
|
||||
# A comes back, observes B's claim, and fences itself.
|
||||
monkeypatch.setattr(replicas, "local_authority_gateway_id", lambda: AUTH_A)
|
||||
replicas.demote_room(
|
||||
adb,
|
||||
room_id="room-1",
|
||||
observed_gateway_id=AUTH_B,
|
||||
observed_epoch=promoted["authority_epoch"],
|
||||
)
|
||||
with pytest.raises(rooms.HostedRoomError):
|
||||
rooms.append_event(
|
||||
adb,
|
||||
room_id="room-1",
|
||||
event_id="a-stale",
|
||||
kind="message.user",
|
||||
actor=USER,
|
||||
payload={"text": "split brain attempt"},
|
||||
authority_gateway_id=AUTH_A,
|
||||
authority_epoch=1,
|
||||
)
|
||||
|
||||
# B's room holds the complete history: 4 original + claim + new work.
|
||||
replay = rooms.read_events(rdb, room_id="room-1", since_seq=0, limit=100)
|
||||
kinds = [e["kind"] for e in replay["events"]]
|
||||
assert kinds == ["message.user"] * 4 + ["authority.claimed", "message.user"]
|
||||
assert replay["authority"]["gateway_id"] == AUTH_B
|
||||
@@ -0,0 +1,173 @@
|
||||
"""Tests for the ``groups.replicate`` / ``groups.promote`` / ``groups.demote``
|
||||
JSON-RPC surface — cross-gateway room durability."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
import tui_gateway.server as srv
|
||||
|
||||
MEMBERS = [{"kind": "bot", "id": "planner"}]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def home(tmp_path, monkeypatch):
|
||||
path = tmp_path / ".hermes"
|
||||
path.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(path))
|
||||
return path
|
||||
|
||||
|
||||
def _result(envelope):
|
||||
assert "error" not in envelope, envelope
|
||||
return envelope["result"]
|
||||
|
||||
|
||||
def _error(envelope):
|
||||
assert "error" in envelope, envelope
|
||||
return envelope["error"]
|
||||
|
||||
|
||||
def _authority_page(tmp_path, gateway_id="install:" + "a" * 32, n=3):
|
||||
"""Build a real room + log on a SEPARATE 'remote authority' DB and return
|
||||
its replay page, as a replicating client would fetch via groups.log."""
|
||||
from gateway import hosted_rooms as rooms
|
||||
|
||||
db = tmp_path / "remote-authority.db"
|
||||
rooms.create_room(
|
||||
db,
|
||||
room_id="room-1",
|
||||
name="Field Room",
|
||||
members=MEMBERS,
|
||||
authority_gateway_id=gateway_id,
|
||||
)
|
||||
for index in range(n):
|
||||
rooms.append_event(
|
||||
db,
|
||||
room_id="room-1",
|
||||
event_id=f"e{index}",
|
||||
kind="message.user",
|
||||
actor={"kind": "user", "id": "tek"},
|
||||
payload={"text": f"msg {index}"},
|
||||
authority_gateway_id=gateway_id,
|
||||
authority_epoch=1,
|
||||
)
|
||||
return rooms.read_events(db, room_id="room-1", since_seq=0, limit=100)
|
||||
|
||||
|
||||
def test_capabilities_advertise_replication(home):
|
||||
result = _result(srv._methods["groups.capabilities"](1, {}))
|
||||
assert "log_replication" in result["features"]
|
||||
assert "authority_takeover" in result["features"]
|
||||
for name in (
|
||||
"groups.replicate",
|
||||
"groups.replica_state",
|
||||
"groups.promote",
|
||||
"groups.demote",
|
||||
):
|
||||
assert name in result["methods"]
|
||||
assert name in srv._LONG_HANDLERS
|
||||
|
||||
|
||||
def test_replicate_then_state_roundtrip(home, tmp_path):
|
||||
page = _authority_page(tmp_path)
|
||||
result = _result(
|
||||
srv._methods["groups.replicate"](
|
||||
1,
|
||||
{
|
||||
"room_id": "room-1",
|
||||
"room_name": "Field Room",
|
||||
"members": MEMBERS,
|
||||
"page": page,
|
||||
},
|
||||
)
|
||||
)
|
||||
assert result["ingested"] == 3
|
||||
state = _result(srv._methods["groups.replica_state"](2, {"room_id": "room-1"}))
|
||||
assert state["last_seq"] == 3
|
||||
assert state["authority"] == page["authority"]
|
||||
|
||||
|
||||
def test_promote_requires_confirm_and_takes_over(home, tmp_path):
|
||||
page = _authority_page(tmp_path)
|
||||
_result(
|
||||
srv._methods["groups.replicate"](
|
||||
1,
|
||||
{
|
||||
"room_id": "room-1",
|
||||
"room_name": "Field Room",
|
||||
"members": MEMBERS,
|
||||
"page": page,
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
refused = _error(srv._methods["groups.promote"](2, {"room_id": "room-1"}))
|
||||
assert refused["code"] == 4118
|
||||
|
||||
promoted = _result(
|
||||
srv._methods["groups.promote"](3, {"room_id": "room-1", "confirm": True})
|
||||
)
|
||||
assert promoted["authority_epoch"] == 2
|
||||
assert promoted["previous_gateway_id"] == page["authority"]["gateway_id"]
|
||||
|
||||
# The room is now hosted locally with full history + claim event.
|
||||
log = _result(srv._methods["groups.log"](4, {"room_id": "room-1"}))
|
||||
kinds = [event["kind"] for event in log["events"]]
|
||||
assert kinds == ["message.user"] * 3 + ["authority.claimed"]
|
||||
assert log["authority"]["epoch"] == 2
|
||||
|
||||
|
||||
def test_demote_fences_local_room_against_newer_epoch(home):
|
||||
from gateway.hosted_rooms import local_authority_gateway_id
|
||||
|
||||
_result(
|
||||
srv._methods["groups.create"](
|
||||
1,
|
||||
{"room_id": "room-1", "name": "Local room", "members": MEMBERS},
|
||||
)
|
||||
)
|
||||
observed_gateway = "install:" + "b" * 32
|
||||
result = _result(
|
||||
srv._methods["groups.demote"](
|
||||
2,
|
||||
{
|
||||
"room_id": "room-1",
|
||||
"observed_gateway_id": observed_gateway,
|
||||
"observed_epoch": 2,
|
||||
},
|
||||
)
|
||||
)
|
||||
assert result["idempotent"] is False
|
||||
assert result["authority_gateway_id"] == observed_gateway
|
||||
|
||||
# Local sends at the stale authority now fail.
|
||||
envelope = srv._methods["groups.send"](
|
||||
3,
|
||||
{
|
||||
"room_id": "room-1",
|
||||
"event_id": "stale-send",
|
||||
"actor": {"kind": "user", "id": "tek"},
|
||||
"payload": {"text": "should fence"},
|
||||
},
|
||||
)
|
||||
assert "error" in envelope
|
||||
assert local_authority_gateway_id() != observed_gateway
|
||||
|
||||
|
||||
def test_replicate_rejects_gapped_page(home, tmp_path):
|
||||
from gateway import hosted_rooms as rooms
|
||||
|
||||
_authority_page(tmp_path, n=5)
|
||||
db = tmp_path / "remote-authority.db"
|
||||
gapped = rooms.read_events(db, room_id="room-1", since_seq=2, limit=100)
|
||||
envelope = srv._methods["groups.replicate"](
|
||||
1,
|
||||
{
|
||||
"room_id": "room-1",
|
||||
"room_name": "Field Room",
|
||||
"members": MEMBERS,
|
||||
"page": gapped,
|
||||
},
|
||||
)
|
||||
assert _error(envelope)["code"] == 4116
|
||||
@@ -19,6 +19,10 @@ LONG_HANDLERS = frozenset({
|
||||
"groups.send",
|
||||
"groups.log",
|
||||
"groups.disband",
|
||||
"groups.replicate",
|
||||
"groups.replica_state",
|
||||
"groups.promote",
|
||||
"groups.demote",
|
||||
})
|
||||
|
||||
|
||||
@@ -46,6 +50,8 @@ def _(rid, params: dict) -> dict:
|
||||
"replayable_disband",
|
||||
"typed_events",
|
||||
"actor_identity",
|
||||
"log_replication",
|
||||
"authority_takeover",
|
||||
],
|
||||
"methods": [
|
||||
"groups.capabilities",
|
||||
@@ -55,6 +61,10 @@ def _(rid, params: dict) -> dict:
|
||||
"groups.send",
|
||||
"groups.log",
|
||||
"groups.disband",
|
||||
"groups.replicate",
|
||||
"groups.replica_state",
|
||||
"groups.promote",
|
||||
"groups.demote",
|
||||
],
|
||||
"max_log_limit": MAX_LOG_LIMIT,
|
||||
},
|
||||
@@ -262,5 +272,97 @@ def _(rid, params: dict) -> dict:
|
||||
return _err(rid, 5113, str(exc))
|
||||
|
||||
|
||||
@method("groups.replicate")
|
||||
def _(rid, params: dict) -> dict:
|
||||
"""Persist one authority-stamped replay page into the local replica store.
|
||||
|
||||
``page`` is the verbatim ``groups.log`` result read from the room's
|
||||
authority gateway; ingest is idempotent and refuses sequence gaps and
|
||||
authority-epoch regressions.
|
||||
"""
|
||||
from gateway.hosted_room_replicas import ReplicaError, ingest_page
|
||||
from gateway.hosted_rooms import default_db_path
|
||||
|
||||
try:
|
||||
result = ingest_page(
|
||||
default_db_path(),
|
||||
room_id=params.get("room_id"),
|
||||
room_name=params.get("room_name"),
|
||||
members=params.get("members"),
|
||||
page=params.get("page"),
|
||||
)
|
||||
return _ok(rid, result)
|
||||
except ReplicaError as exc:
|
||||
return _err(rid, 4116, str(exc))
|
||||
except Exception as exc:
|
||||
return _err(rid, 5116, str(exc))
|
||||
|
||||
|
||||
@method("groups.replica_state")
|
||||
def _(rid, params: dict) -> dict:
|
||||
"""Report the local replica's coverage and authority lineage."""
|
||||
from gateway.hosted_room_replicas import ReplicaError, replica_state
|
||||
from gateway.hosted_rooms import default_db_path
|
||||
|
||||
try:
|
||||
return _ok(rid, replica_state(default_db_path(), room_id=params.get("room_id")))
|
||||
except ReplicaError as exc:
|
||||
return _err(rid, 4117, str(exc))
|
||||
except Exception as exc:
|
||||
return _err(rid, 5117, str(exc))
|
||||
|
||||
|
||||
@method("groups.promote")
|
||||
def _(rid, params: dict) -> dict:
|
||||
"""Continue a replicated room on THIS gateway at ``epoch + 1``.
|
||||
|
||||
Requires ``confirm: true`` — the caller asserts the previous authority can
|
||||
no longer commit (explicit user action; a lease/quorum driver later).
|
||||
"""
|
||||
from gateway.hosted_room_replicas import ReplicaError, promote_replica
|
||||
from gateway.hosted_rooms import HostedRoomError, default_db_path
|
||||
|
||||
if params.get("confirm") is not True:
|
||||
return _err(
|
||||
rid,
|
||||
4118,
|
||||
"promotion requires confirm=true acknowledging the previous "
|
||||
"authority can no longer commit",
|
||||
)
|
||||
try:
|
||||
result = promote_replica(
|
||||
default_db_path(),
|
||||
room_id=params.get("room_id"),
|
||||
reason=params.get("reason", "authority-unreachable"),
|
||||
)
|
||||
return _ok(rid, result)
|
||||
except ReplicaError as exc:
|
||||
return _err(rid, 4118, str(exc))
|
||||
except HostedRoomError as exc:
|
||||
return _err(rid, 4118, str(exc))
|
||||
except Exception as exc:
|
||||
return _err(rid, 5118, str(exc))
|
||||
|
||||
|
||||
@method("groups.demote")
|
||||
def _(rid, params: dict) -> dict:
|
||||
"""Fence this gateway's stale room authority against a proven newer epoch."""
|
||||
from gateway.hosted_room_replicas import ReplicaError, demote_room
|
||||
from gateway.hosted_rooms import default_db_path
|
||||
|
||||
try:
|
||||
result = demote_room(
|
||||
default_db_path(),
|
||||
room_id=params.get("room_id"),
|
||||
observed_gateway_id=params.get("observed_gateway_id"),
|
||||
observed_epoch=params.get("observed_epoch"),
|
||||
)
|
||||
return _ok(rid, result)
|
||||
except ReplicaError as exc:
|
||||
return _err(rid, 4119, str(exc))
|
||||
except Exception as exc:
|
||||
return _err(rid, 5119, str(exc))
|
||||
|
||||
|
||||
def register(server) -> None:
|
||||
_registry.install(server)
|
||||
|
||||
Reference in New Issue
Block a user