diff --git a/proto/decentraland/pulse/pulse_clusters.proto b/proto/decentraland/pulse/pulse_clusters.proto new file mode 100644 index 00000000..3c32b5e0 --- /dev/null +++ b/proto/decentraland/pulse/pulse_clusters.proto @@ -0,0 +1,22 @@ +syntax = "proto3"; + +package decentraland.pulse; + +// Published by Pulse to NATS on `peer.{address}.cluster_change` whenever a peer's post-debounce +// cluster assignment changes. Pulse is the sole author of cluster membership. +// +// A "cluster" is Pulse's own grouping of co-located peers, derived by union-find over occupied +// area-of-interest grid cells. It is deliberately a different concept from archipelago's "island": +// clusters have no size cap, are realm-partitioned, and carry no transport connection details. +// Comms Gatekeeper subscribes here, mints the LiveKit connection string (the ban check is +// authority-local to gatekeeper), and re-emits the legacy +// decentraland.kernel.comms.v3.IslandChangedMessage on `engine.peer.{address}.island_changed`, +// so the WebSocket Connector and clients stay untouched. +message PeerClusterChange { + // Cluster the peer now belongs to. Stable across passes for an unchanged crowd. + string cluster_id = 1; + + // The peer's realm. Clusters never span realms, so this qualifies cluster_id and keeps the + // event self-contained — consumers never have to join against the topology feed. + string realm = 2; +} diff --git a/proto/decentraland/pulse/pulse_presence.proto b/proto/decentraland/pulse/pulse_presence.proto new file mode 100644 index 00000000..3b166e62 --- /dev/null +++ b/proto/decentraland/pulse/pulse_presence.proto @@ -0,0 +1,41 @@ +syntax = "proto3"; + +package decentraland.pulse; + +// Published by Pulse to NATS on `engine.parcel_changes`: the platform's single presence feed. +// One entry per (peer, realm) state change: the first non-null placement in a realm is the +// connect signal, an entry with `parcel` absent is the exit signal for that realm/instance, +// and every entry carries the realm and the parcel the peer now stands on. +// +// Guarantees the producer (Pulse) holds — see archipelago-workers/docs/contracts/iteration-2/README.md: +// 1. A peer's first placement in a realm emits one non-null entry, even if it never moves afterwards. +// 2. Every exit path emits exactly one parcel-absent entry (clean disconnect, PENDING_AUTH / idle +// timeout, duplicate-session kick, BanEnforcer eviction, PeerDefense kick). A realm change +// emits a non-null entry for the new realm only; the old realm is implied. +// 3. Within one batch a wallet appears at most once, with its latest state. +// 4. Cadence: every Presence:BatchIntervalMs (default 2000) when non-empty; a snapshot=true batch +// on publisher start, every Presence:SnapshotIntervalMs (default 60000), and immediately after +// any outbox eviction, so consumers never run on a known-lossy delta stream. +// 5. `realm` and `address` are lowercase; `server_name` is stable for the process lifetime. +// +// Consumer rule: keep lastSeq[server_name]; on a gap keep serving stale state and wait for the next +// snapshot (<= 60 s) — never drop the map. On snapshot=true replace every entry whose server_name matches. + +message Parcel { + int32 x = 1; + int32 y = 2; +} + +message ParcelChange { + string address = 1; // lowercase 0x wallet + string realm = 2; // canonical lowercase + optional Parcel parcel = 3; // absent => peer left this realm/instance +} + +message ParcelChangesBatch { + string server_name = 1; // NatsOptions.ServerName; consumers key seq per server_name + uint64 seq = 2; // monotonic per server_name, resets on restart (snapshot=true follows) + bool snapshot = 3; // true => full state of this server, replaces everything known for server_name + uint64 server_time = 4; // unix ms + repeated ParcelChange changes = 5; +} diff --git a/public/comms.proto b/public/comms.proto index b81b7467..06345884 100644 --- a/public/comms.proto +++ b/public/comms.proto @@ -2,4 +2,6 @@ syntax = "proto3"; import public "decentraland/kernel/comms/v3/archipelago.proto"; import public "decentraland/kernel/comms/rfc4/comms.proto"; -import public "decentraland/kernel/comms/rfc5/ws_comms.proto"; \ No newline at end of file +import public "decentraland/kernel/comms/rfc5/ws_comms.proto"; +import public "decentraland/pulse/pulse_clusters.proto"; +import public "decentraland/pulse/pulse_presence.proto"; \ No newline at end of file