mirror of
https://github.com/PerpetualSoftware/pad.git
synced 2026-09-23 02:53:31 +00:00
xarmian
264f5b0041
* feat(collab): add OpBus interface + in-process MemoryOpBus (TASK-1253)
New internal/collab package for the dumb-relay collab server in
PLAN-1248. Defines the OpBus pub/sub interface and ships
MemoryOpBus, the in-process implementation used by every shipping
target today (single-binary self-host, single-replica pad-cloud).
OpBus shape mirrors internal/events.MemoryBus so a future RedisOpBus
is a drop-in for multi-replica deployments — that's filed as a
separate IDEA at PLAN-1248 close, since the dumb-relay design
intentionally keeps Redis off the self-host dependency surface.
OpEvent carries:
- ItemID fan-out filter
- ClientID Yjs client id, used by designated-applier election
(TASK-1257); the bus itself does not interpret it
- Type "sync" (Y.Doc binary update — persisted by the room
manager) or "awareness" (cursor/presence — broadcast
only, never persisted)
- Data raw y-protocol message; opaque to the server
- Timestamp UnixMilli, auto-stamped on Publish
MemoryOpBus semantics:
- 64-event buffered subscriber channels (matches internal/events
default — sized against keystroke-rate workload).
- Non-blocking Publish: a slow subscriber whose channel is full has
events DROPPED with a warn log rather than back-pressuring the
broadcast loop. The room manager (TASK-1255) is responsible for
closing genuinely unhealthy peers; the bus only protects itself.
- Idempotent Unsubscribe (no panic on double-unsubscribe).
- Close clears the subscriber map and closes every channel under
the same write lock that gates Publish, so a final inflight
Publish can't race a Close into delivering on a closed channel.
Tests cover: subscribe/publish fan-out per item filter, unsubscribe
closes the channel, slow-consumer drop without blocking, accurate
SubscriberCount across subscribe/unsubscribe, Close cleans up every
channel regardless of itemID, and concurrent-publishers race-clean
under -race (small drop count tolerated — that's the slow-consumer
contract; an exact-delivery test would defeat its own purpose).
No external dependencies beyond stdlib + slog. RedisOpBus stub is
intentionally NOT included — separate IDEA per the Plan body.
Parent: PLAN-1248. Phase 1 — Backend foundation.
* fix(collab): clone OpEvent.Data + document recovery contract per Codex review (round 1)
P1 — Sync-op drops were undocumented as recoverable. Sync drops ARE
recoverable in the dumb-relay design: the room manager (TASK-1255)
appends to the op-log BEFORE Publish, so any peer that misses a sync
op via channel-full drop can replay since their last cursor on
reconnect (TASK-1252's LoadYjsUpdatesSince + Yjs state-vector
negotiation). The room manager is responsible for detecting slow
channels and force-closing the owning WebSocket, which kicks the
peer into a fresh reconnect + replay. The bus does not take that
action itself because it has no concept of which peer owns which
channel — that mapping is the room manager's domain. Doc comment now
spells this out explicitly.
P2 — OpEvent.Data is a []byte; the same slice header was queued to
every subscriber, so a publisher's later buffer reuse OR any
subscriber's mutation could corrupt the bytes other subscribers
observe. gorilla/websocket's ReadMessage is allowed to reuse its
read buffer between messages, so this hazard is real for the
production publisher (the WS handler in TASK-1254). Clone Data once
at the publish boundary; document the per-receiver immutability
expectation in the same comment block.
No behavioral test change — the existing slow-consumer-drop test
still passes; the clone path adds one allocation per Publish but
nothing observable to callers beyond the immutability guarantee.