summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-08-16 15:28:53 +0200
committerChristophe Besson <cbesson@gmail.com>2026-08-16 15:28:53 +0200
commitfd770dc6293be67298f582de5800f2ca6fe24a8b (patch)
treedf8a393741ccf391b5bb4fa490980c632695cbdd
parent0bd3f805ffbd04b40b5150474336a5e5d200e72b (diff)
downloadmeshbay-fd770dc6293be67298f582de5800f2ca6fe24a8b.tar.gz
feat(common): MNP 0.2 — liveness, and chat history read the way it is written
`get_messages` pages forward from the oldest message. That is the right shape for "what happened since I last looked" and the wrong one for opening a conversation, and the browser asked it for `since=0, limit=200` — so a group with more than two hundred messages showed its first two hundred and the exchange anyone came for was unreachable. Demonstrated on 300 messages: the newest was simply absent from the answer. `get_recent` and `get_before` page backwards, cursored on the row id rather than the timestamp. Nothing makes a `time.time()` float unique, and a cursor on a value two rows can share eventually skips a message or repeats it. PING/PONG covers liveness on an already-open channel: a DataChannel whose peer vanished without closing still reads as connected, and nothing noticed until a real request hung. It is not a discovery mechanism — opening a connection to ping costs a full ICE/DTLS handshake, measured at 0.6-7 s across two ISPs — so presence in the group list comes from the hub's registry instead. Both additions are backward compatible: an 0.1 peer sends no `before` and is answered with the newest page, which is what it wanted. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
-rw-r--r--packages/meshbay-common/src/meshbay_common/__init__.py7
-rw-r--r--packages/meshbay-common/src/meshbay_common/protocol.py10
-rw-r--r--packages/meshbay-node/src/meshbay_node/chat/store.py43
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/quic_server.py5
-rw-r--r--packages/meshbay-node/tests/test_chat_pagination.py150
5 files changed, 212 insertions, 3 deletions
diff --git a/packages/meshbay-common/src/meshbay_common/__init__.py b/packages/meshbay-common/src/meshbay_common/__init__.py
index ef0f1f6..60a9dc7 100644
--- a/packages/meshbay-common/src/meshbay_common/__init__.py
+++ b/packages/meshbay-common/src/meshbay_common/__init__.py
@@ -1,5 +1,8 @@
"""MeshBay common — shared crypto primitives and protocol types."""
-__version__ = "0.4.0"
-MNP_VERSION = "0.1"
+__version__ = "0.5.0"
+# 0.2: added PING/PONG, and `before`/`has_more` on chat history. Both are
+# additive — an 0.1 peer sends no `before` and gets the newest page, which is
+# what it wanted — so this is a MINOR bump, not a MAJOR one.
+MNP_VERSION = "0.2"
MHP_VERSION = "0.1"
diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py
index b5a54ff..1aadafe 100644
--- a/packages/meshbay-common/src/meshbay_common/protocol.py
+++ b/packages/meshbay-common/src/meshbay_common/protocol.py
@@ -27,8 +27,16 @@ class MNP:
STREAM_SEGMENT = "stream_seg" # HLS/DASH segment
CHAT_MESSAGE = "chat_msg" # Double Ratchet message
CHAT_ATTACHMENT = "chat_attach" # attachment metadata
- CHAT_HISTORY = "chat_hist" # request message history
+ CHAT_HISTORY = "chat_hist" # request message history (newest, or before a cursor)
CHAT_HISTORY_RESPONSE = "chat_hist_resp" # history response with messages
+ # Liveness on an *already open* channel. A peer that goes away without
+ # closing leaves a DataChannel that still reads as connected until the next
+ # real request hangs, and there was no way to ask. This is not a discovery
+ # mechanism: opening a connection in order to ping costs a full ICE/DTLS
+ # handshake (measured at 0.6-7 s across two ISPs), so presence in the group
+ # list comes from the hub's socket registry instead.
+ PING = "ping"
+ PONG = "pong"
# GEK_REQUEST / GEK_RESPONSE were removed (NS3, and finding L1): the node must
# never serve the GEK in plaintext. Members obtain it by unwrapping their own
# ECIES bundle. The constants lingered after the handlers were deleted, leaving
diff --git a/packages/meshbay-node/src/meshbay_node/chat/store.py b/packages/meshbay-node/src/meshbay_node/chat/store.py
index a5b2d7d..6f23905 100644
--- a/packages/meshbay-node/src/meshbay_node/chat/store.py
+++ b/packages/meshbay-node/src/meshbay_node/chat/store.py
@@ -45,6 +45,11 @@ class StoredMessage:
sender_name: str = ""
+def _row(r) -> StoredMessage:
+ return StoredMessage(id=r[0], sender_id=r[1], iteration=r[2], payload=r[3],
+ timestamp=r[4], thread_id=r[5], sender_name=r[6] or "")
+
+
class ChatStore:
"""Async SQLite chat store for one group."""
@@ -111,6 +116,44 @@ class ChatStore:
for r in rows
]
+ async def get_recent(self, limit: int = 100) -> list[StoredMessage]:
+ """The newest `limit` messages, oldest first so they render in order.
+
+ `get_messages(since=0, limit=n)` cannot answer this: it pages *forward*
+ from the oldest, so a client opening a group with 300 messages was shown
+ the first 200 and never the conversation it came for. Reaching the newest
+ rows means ordering DESC in SQL and reversing here.
+ """
+ cursor = await self._db.execute(
+ "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name "
+ "FROM messages ORDER BY id DESC LIMIT ?",
+ (limit,),
+ )
+ rows = await cursor.fetchall()
+ return [_row(r) for r in reversed(rows)]
+
+ async def get_before(self, before_id: int, limit: int = 50) -> list[StoredMessage]:
+ """The `limit` messages immediately older than `before_id`, oldest first.
+
+ The cursor is the row id, not the timestamp. Nothing makes `timestamp`
+ unique — it is a float from `time.time()` — and a cursor on a value two
+ rows can share is one that eventually skips a message or repeats it.
+ `id` is AUTOINCREMENT: unique, and ordered by insertion.
+ """
+ cursor = await self._db.execute(
+ "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name "
+ "FROM messages WHERE id < ? ORDER BY id DESC LIMIT ?",
+ (before_id, limit),
+ )
+ rows = await cursor.fetchall()
+ return [_row(r) for r in reversed(rows)]
+
+ async def has_before(self, message_id: int) -> bool:
+ """Whether anything older than `message_id` exists — drives "load older"."""
+ cursor = await self._db.execute(
+ "SELECT 1 FROM messages WHERE id < ? LIMIT 1", (message_id,))
+ return await cursor.fetchone() is not None
+
async def get_thread(self, thread_id: str, limit: int = 100) -> list[StoredMessage]:
"""Get messages in a thread."""
cursor = await self._db.execute(
diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
index ed3925d..73e668c 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
@@ -203,6 +203,11 @@ class _MNPServerProtocol(QuicConnectionProtocol):
self._do_stream_segment_sync(stream_id, msg)
elif mtype == MNP.CHAT_MESSAGE:
self._do_chat_message_sync(stream_id, msg)
+ elif mtype == MNP.PING:
+ # Liveness is transport-agnostic, and a native client over QUIC
+ # has the same half-open problem a DataChannel does.
+ self._send(stream_id, {"type": MNP.PONG, "v": MNP_VERSION,
+ "token": msg.get("token")})
else:
log.warning("Unknown MNP message type: %s", mtype)
except Exception as e:
diff --git a/packages/meshbay-node/tests/test_chat_pagination.py b/packages/meshbay-node/tests/test_chat_pagination.py
new file mode 100644
index 0000000..caf1f73
--- /dev/null
+++ b/packages/meshbay-node/tests/test_chat_pagination.py
@@ -0,0 +1,150 @@
+"""
+Reading a chat backwards.
+
+`get_messages(since, limit)` pages *forward* from the oldest message, which is
+the right shape for "what happened since I last looked" and the wrong shape for
+opening a conversation. The browser asked it for `since=0, limit=200`, so a
+group with more than 200 messages showed its first 200 and never the exchange
+the reader came for. These tests pin the direction, because the bug was not a
+crash — it was a plausible-looking screen full of the wrong messages.
+
+The cursor is the row id rather than the timestamp. `timestamp` is a float from
+`time.time()`, so two messages saved in the same tick can share one, and a
+timestamp cursor would then skip a message or return it twice.
+"""
+
+import asyncio
+from pathlib import Path
+
+import pytest
+import pytest_asyncio
+
+from meshbay_node.chat.store import ChatStore
+
+
+@pytest_asyncio.fixture
+async def store(tmp_path: Path):
+ async with ChatStore(tmp_path / "chat.db") as s:
+ yield s
+
+
+async def _fill(store, n: int) -> None:
+ for i in range(n):
+ await store.save_message(sender_id="u1", iteration=i,
+ payload=f"msg-{i:03d}".encode())
+
+
+def _texts(msgs) -> list[str]:
+ return [m.payload.decode() for m in msgs]
+
+
+@pytest.mark.asyncio
+async def test_recent_returns_the_newest_not_the_oldest(store):
+ await _fill(store, 300)
+ got = _texts(await store.get_recent(limit=100))
+
+ assert len(got) == 100
+ assert got[-1] == "msg-299", "the newest message must be in the answer"
+ assert got[0] == "msg-200"
+
+
+@pytest.mark.asyncio
+async def test_recent_is_ordered_oldest_first_for_rendering(store):
+ await _fill(store, 10)
+ got = _texts(await store.get_recent(limit=5))
+ assert got == ["msg-005", "msg-006", "msg-007", "msg-008", "msg-009"]
+
+
+@pytest.mark.asyncio
+async def test_recent_handles_fewer_messages_than_the_limit(store):
+ await _fill(store, 3)
+ assert _texts(await store.get_recent(limit=100)) == \
+ ["msg-000", "msg-001", "msg-002"]
+
+
+@pytest.mark.asyncio
+async def test_recent_on_an_empty_group(store):
+ assert await store.get_recent(limit=100) == []
+
+
+@pytest.mark.asyncio
+async def test_before_walks_backwards_without_gap_or_repeat(store):
+ """Paging to the start must show every message exactly once."""
+ await _fill(store, 250)
+
+ page = await store.get_recent(limit=100)
+ seen = _texts(page)
+ while await store.has_before(page[0].id):
+ page = await store.get_before(page[0].id, limit=50)
+ assert page, "has_before said there was more"
+ seen = _texts(page) + seen
+
+ assert seen == [f"msg-{i:03d}" for i in range(250)]
+ assert len(seen) == len(set(seen)), "no message returned twice"
+
+
+@pytest.mark.asyncio
+async def test_before_is_ordered_oldest_first(store):
+ await _fill(store, 20)
+ recent = await store.get_recent(limit=5)
+ older = _texts(await store.get_before(recent[0].id, limit=5))
+ assert older == ["msg-010", "msg-011", "msg-012", "msg-013", "msg-014"]
+
+
+@pytest.mark.asyncio
+async def test_before_excludes_the_cursor_message(store):
+ await _fill(store, 10)
+ recent = await store.get_recent(limit=3)
+ older = await store.get_before(recent[0].id, limit=10)
+ assert recent[0].id not in [m.id for m in older]
+
+
+@pytest.mark.asyncio
+async def test_has_before_is_false_at_the_start_of_history(store):
+ await _fill(store, 5)
+ page = await store.get_recent(limit=100)
+ assert await store.has_before(page[0].id) is False
+
+
+@pytest.mark.asyncio
+async def test_has_before_is_true_when_older_messages_exist(store):
+ await _fill(store, 200)
+ page = await store.get_recent(limit=100)
+ assert await store.has_before(page[0].id) is True
+
+
+@pytest.mark.asyncio
+async def test_messages_sharing_a_timestamp_are_still_paged_exactly_once(store):
+ """The reason the cursor is the row id rather than the timestamp.
+
+ Nothing makes `timestamp` unique — it is a float from `time.time()`. In
+ practice `save_message` commits per row and they rarely tie, so the tie is
+ forced here instead of hoped for: what matters is that a shared timestamp
+ cannot make a page skip a message or repeat one, not how often it happens.
+ """
+ for i in range(40):
+ await store._db.execute(
+ "INSERT INTO messages (sender_id, iteration, payload, timestamp, "
+ "thread_id, sender_name) VALUES (?, ?, ?, ?, ?, ?)",
+ ("u1", i, f"msg-{i:03d}".encode(), 1000.0, None, ""))
+ await store._db.commit()
+
+ rows = await store.get_recent(limit=40)
+ assert len({m.timestamp for m in rows}) == 1, "the tie should be in place"
+
+ page = await store.get_recent(limit=10)
+ seen = _texts(page)
+ while await store.has_before(page[0].id):
+ page = await store.get_before(page[0].id, limit=10)
+ seen = _texts(page) + seen
+ assert seen == [f"msg-{i:03d}" for i in range(40)]
+
+
+@pytest.mark.asyncio
+async def test_forward_paging_still_works_for_callers_that_want_it(store):
+ """`get_messages` keeps its meaning — the node UI reads "since" from it."""
+ await _fill(store, 10)
+ all_msgs = await store.get_messages(limit=100)
+ cutoff = all_msgs[4].timestamp
+ after = await store.get_messages(since=cutoff, limit=100)
+ assert all(m.timestamp > cutoff for m in after)