From 004af5dea1b78770d73f7290c9169f4742dbaefc Mon Sep 17 00:00:00 2001 From: Mikhail Agapov Date: Wed, 16 Sep 2026 13:13:32 +0300 Subject: [PATCH 1/2] feat: restore presence on stacked iteration two protocol Signed-off-by: Mikhail Agapov --- proto/decentraland/pulse/pulse_presence.proto | 41 +++++++++++++++++++ public/comms.proto | 1 + 2 files changed, 42 insertions(+) create mode 100644 proto/decentraland/pulse/pulse_presence.proto 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 92e647f7..dc6862a3 100644 --- a/public/comms.proto +++ b/public/comms.proto @@ -4,3 +4,4 @@ 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"; import public "decentraland/pulse/pulse_clusters.proto"; +import public "decentraland/pulse/pulse_presence.proto"; From 09dbc6c471a985442269d7c7c5cd8c5151c1ad98 Mon Sep 17 00:00:00 2001 From: Mikhail Agapov Date: Tue, 15 Sep 2026 16:56:23 +0300 Subject: [PATCH 2/2] feat: add session takeover control messages and wire goldens Signed-off-by: Mikhail Agapov --- package.json | 3 +- .../kernel/comms/v3/archipelago.proto | 17 ++++ scripts/build-session-artifacts.ps1 | 40 +++++++++ scripts/test.sh | 2 + session-takeover-integration.md | 88 +++++++++++++++++++ session-takeover-report.md | 65 ++++++++++++++ test/fixtures/session-control.json | 13 +++ test/session-control.test.js | 25 ++++++ 8 files changed, 252 insertions(+), 1 deletion(-) create mode 100644 scripts/build-session-artifacts.ps1 create mode 100644 session-takeover-integration.md create mode 100644 session-takeover-report.md create mode 100644 test/fixtures/session-control.json create mode 100644 test/session-control.test.js diff --git a/package.json b/package.json index 9a137814..9d7c7294 100644 --- a/package.json +++ b/package.json @@ -14,7 +14,8 @@ "main": "index.js", "scripts": { "test": "./scripts/check-proto-compabitility.sh", - "gen:test": "node protoc-gen-bitwise/test/generator.test.js" + "gen:test": "node protoc-gen-bitwise/test/generator.test.js", + "test:session-control": "node test/session-control.test.js" }, "devDependencies": { "@protobuf-ts/protoc": "^2.8.1", diff --git a/proto/decentraland/kernel/comms/v3/archipelago.proto b/proto/decentraland/kernel/comms/v3/archipelago.proto index 69cac9f7..9383fe79 100644 --- a/proto/decentraland/kernel/comms/v3/archipelago.proto +++ b/proto/decentraland/kernel/comms/v3/archipelago.proto @@ -32,13 +32,29 @@ message JoinIslandMessage { } enum KickedReason { + // Authoritative logical-session supersession, not same-session socket replacement. KR_NEW_SESSION = 0; + KR_BANNED = 1; } message KickedMessage { KickedReason reason = 1; } +enum SessionStatus { + UNSPECIFIED = 0; + TAKEOVER_PENDING = 1; + TAKEOVER_FAILED = 2; +} + +// Session-addressed control. Success is still IslandChangedMessage with usable credentials. +message SessionStatusMessage { + SessionStatus state = 1; + // Minimum recovery suppression interval, not a countdown to a new handshake. + // Pending waits on the existing authenticated connection; failed requires explicit retry. + uint32 retry_after_ms = 2; +} + message ServerPacket { oneof message { ChallengeResponseMessage challenge_response = 1; @@ -47,6 +63,7 @@ message ServerPacket { LeftIslandMessage left_island = 4; JoinIslandMessage join_island = 5; KickedMessage kicked = 6; + SessionStatusMessage session_status = 7; } } diff --git a/scripts/build-session-artifacts.ps1 b/scripts/build-session-artifacts.ps1 new file mode 100644 index 00000000..824f1acf --- /dev/null +++ b/scripts/build-session-artifacts.ps1 @@ -0,0 +1,40 @@ +$ErrorActionPreference = 'Stop' +Set-Location (Split-Path $PSScriptRoot -Parent) +$mutex = [Threading.Mutex]::new($false, 'Local\DCL_It2_HeavyVerification') +$acquired = $false +try { + try { $acquired = $mutex.WaitOne(0) } catch [Threading.AbandonedMutexException] { $acquired = $true } + if (-not $acquired) { Write-Output 'Heavy verification occupied; retry later.'; exit 75 } + $freeGiB = (Get-CimInstance Win32_OperatingSystem).FreePhysicalMemory / 1MB + Write-Output "Free physical RAM: $freeGiB GiB" + if ($freeGiB -lt 7) { throw 'Need 5 GiB reserve plus 2 GiB Node heap headroom.' } + $env:NODE_OPTIONS = '--max-old-space-size=2048' + function Check-Exit { if ($LASTEXITCODE -ne 0) { throw "Command failed: $LASTEXITCODE" } } + New-Item -ItemType Directory -Force out-ts,out-js,out-session-takeover | Out-Null + Copy-Item node_modules/.bin/buf node_modules/.bin/buf.exe -Force + & node_modules/.bin/buf.exe lint proto/ + Check-Exit + & node_modules/.bin/buf.exe build proto/ + Check-Exit + & node_modules/.bin/buf.exe breaking proto/ --against 'https://github.com/decentraland/protocol.git#subdir=proto' + Check-Exit + foreach ($file in Get-ChildItem public/*.proto) { + $plugin = (Resolve-Path node_modules/.bin/protoc-gen-dcl_ts_proto.cmd).Path + & node_modules/.bin/protoc.cmd ("--plugin=protoc-gen-dcl_ts_proto=" + $plugin) --dcl_ts_proto_opt=esModuleInterop=true,returnObservable=false,outputServices=generic-definitions,fileSuffix=.gen,oneof=unions --dcl_ts_proto_out=out-ts -I=proto -I=public ("public/" + $file.Name) + Check-Exit + } + & node_modules/.bin/tsc.cmd -p tsconfig.json + Check-Exit + & node_modules/.bin/protoc.cmd --csharp_out=out-session-takeover --csharp_opt=file_extension=.gen.cs -I=proto proto/decentraland/kernel/comms/v3/archipelago.proto + Check-Exit + node test/session-control.test.js + Check-Exit + npm run gen:test + Check-Exit + npm pack --pack-destination out-session-takeover + Check-Exit + Get-FileHash out-session-takeover/Archipelago.gen.cs,out-session-takeover/dcl-protocol-1.0.0.tgz -Algorithm SHA256 +} finally { + if ($acquired) { $mutex.ReleaseMutex() } + $mutex.Dispose() +} diff --git a/scripts/test.sh b/scripts/test.sh index 284f89da..1f51126b 100755 --- a/scripts/test.sh +++ b/scripts/test.sh @@ -26,4 +26,6 @@ echo "> Compiling TS project" node_modules/.bin/tsc -p tsconfig.json +node test/session-control.test.js + rm -rf out-cs || true diff --git a/session-takeover-integration.md b/session-takeover-integration.md new file mode 100644 index 00000000..58f7c21b --- /dev/null +++ b/session-takeover-integration.md @@ -0,0 +1,88 @@ +# Session takeover protocol integration + +## Contract + +Package: `decentraland.kernel.comms.v3`, source `proto/decentraland/kernel/comms/v3/archipelago.proto`. + +| Declaration | Wire numbers | +| --- | --- | +| KickedReason | KR_NEW_SESSION=0 (unchanged), KR_BANNED=1 | +| SessionStatus | UNSPECIFIED=0, TAKEOVER_PENDING=1, TAKEOVER_FAILED=2 | +| SessionStatusMessage | state=1 (SessionStatus), retry_after_ms=2 (uint32) | +| ServerPacket.message | session_status=7; existing tags 1–6 unchanged | + +TypeScript imports: + +```ts +import { ServerPacket, SessionStatus, SessionStatusMessage, KickedMessage, KickedReason } + from '@dcl/protocol/out-js/decentraland/kernel/comms/v3/archipelago.gen' + +const status = { state: SessionStatus.TAKEOVER_PENDING, retryAfterMs: 1500 } +const natsBytes = SessionStatusMessage.encode(status).finish() +const socketBytes = ServerPacket.encode({ + message: { $case: 'sessionStatus', sessionStatus: status } +}).finish() +``` + +Gatekeeper publishes bare status to `engine.peer..session_status.` and bare +KickedMessage to `engine.peer..kicked.`. Connector authenticates and targets +exact wallet/session, supplies the envelope, and keeps pending sockets open. Success is the existing +island_changed. Control events bypass assignment deduplication. No wallet-only control fallback. + +Pending suppresses recovery for at least retryAfterMs; expiration is not permission to handshake. +Failure withholds credentials and requires user retry. KR_NEW_SESSION means authoritative logical +session supersession, not replacing another socket for the same session. Same-session socket +replacement must close only the old transport or await a separately approved reason. Unknown kick +reasons stop automatic retries but are not confirmed supersession. Legacy empty KickedMessage still +means KR_NEW_SESSION. Unknown binary enum numbers remain numbers in the generated JS decoder; +JSON conversions may map them to UNRECOGNIZED. Consumers must use an explicit switch/default. + +## Reproduce artifacts on Windows + +From this worktree, install dependencies with the shared heavy-verification mutex and RAM guard. +Use Node 24, protoc 22.2 (package config), @dcl/ts-proto 1.154.0, protobufjs 7.2.4, and Buf 1.8.0. +Provide the Windows Buf 1.8.0 binary at `node_modules/.bin/buf` (the existing local install has it). +Then run: + +```powershell +powershell -NoProfile -ExecutionPolicy Bypass -File scripts/build-session-artifacts.ps1 +``` + +The script acquires the mutex nonblocking (exit 75 means retry later), runs Buf checks, generates all +public TS entrypoints, compiles JS, generates Archipelago.gen.cs directly with protoc, runs the wire +and bitwise goldens, and packs a local archive. Outputs: + +- `out-session-takeover/dcl-protocol-1.0.0.tgz` +- `out-session-takeover/Archipelago.gen.cs` +- Reusable goldens: `test/fixtures/session-control.json` (hex bytes and canonical JSON). + +No registry version is reserved: the archive's 1.0.0 is the repository's development version, not a +new published stable release. Identify it by source commit and SHA256. Consumer owners may install +the absolute archive locally with `npm install --no-save --package-lock=false ` (under the +same mutex) in their isolated candidate. Keep temporary local paths out of committed manifests and +locks. Verify the installed generated exports before testing. Do not broaden unrelated dependency +pins or social-service's isolated alias. + +For Unity, copy the generated Archipelago.gen.cs to the existing generated protocol location after +checking local changes; keep its existing .meta file. It uses the existing Google.Protobuf runtime and +Vectors generated definitions. No handwritten parallel protocol classes are needed. + +## CI and release gates + +Linux reproduction after installing dependencies: `make buf-breaking`, `make test`, +`npm run gen:test`, `npm pack`; direct C# regeneration uses protoc 22.2 with +`--csharp_out= --csharp_opt=file_extension=.gen.cs -I=proto +proto/decentraland/kernel/comms/v3/archipelago.proto`. + +Before publication, CI needs either an approved immutable artifact upload keyed to the reviewed +protocol commit and SHA256, or a checkout/build of that exact protocol commit followed by a local +archive install. Local filesystem paths cannot serve as remote CI pins. This pass does not upload or +publish. Stable publication and coordinated consumer lock updates require separate approval. + +Required integration gates: all ban producers use KR_BANNED; same-session socket replacement does +not latch logical supersession; pending/failure reach only the winning session; authoritative kick +latches suppression before transport close; stale asynchronous callbacks cannot issue credentials or +late control events; old readers ignore status and retain parking compatibility; two-client tests +exercise lost status, reconnect replay, timeout/watchdog, and late assignment after supersession. +LiveKit Cloud revocation/removal and next-second cutoff behavior require actual Cloud evidence; +protocol goldens do not establish that gate. diff --git a/session-takeover-report.md b/session-takeover-report.md new file mode 100644 index 00000000..5b7a5ac6 --- /dev/null +++ b/session-takeover-report.md @@ -0,0 +1,65 @@ +# Session takeover amendment — Protocol report + +Base: `a1f2f3db4147b3689748af205c0923304ccc5c6a`, preserved on `codex/it2-sep15-protocol`. +Binding plan: `Claude outputs/session-takeover-client-plan.md`, September 15. + +## Implementation + +Added KR_BANNED=1, SessionStatus values UNSPECIFIED=0 / TAKEOVER_PENDING=1 / +TAKEOVER_FAILED=2, SessionStatusMessage (state=1, retry_after_ms=2), and +ServerPacket.session_status=7. Existing fields and enum values retain their numbers. +C0/C1 Pulse schemas and exports have no diff against the base. + +Committed regression goldens cover legacy welcome/assignment/supersession bytes, bare NATS and +wrapped WS pending/failure/ban, unspecified status, defaults, unknown binary enum values, and +old-reader tolerance of field 7. Normal scripts/test.sh now executes the generated-JS wire tests. +scripts/build-session-artifacts.ps1 supplies Windows reproduction and holds the shared mutex for +generation, compilation, tests and packing. Integration instructions are in session-takeover-integration.md. + +## RED/GREEN + +RED: `node test/session-control.test.js` against the pre-amendment generated module exited 1 at +`bare ban: decode`: numeric reason 1 decoded but the old named KR_BANNED mapped to -1. The four +preceding legacy cases passed. The test and fixtures were written before the schema amendment. + +Initial Windows build-script attempts exposed executable extension, protoc include-path, and npm +plugin path issues. These were corrected in the build helper; they are not schema regressions. +Mutex contention caused a nonblocking skip, followed by useful documentation work before retry. + +GREEN: `powershell -NoProfile -ExecutionPolicy Bypass -File scripts/build-session-artifacts.ps1` +exited 0: Buf lint/build/breaking against current main; all public TS codegen and tsc compilation; +direct protoc C# generation; 11 wire goldens/defaults/unknown values/old-reader checks; +two existing bitwise goldens; local npm pack (536 files). About 39 GiB RAM was free at start. + +Artifacts in `out-session-takeover/`: + +- `Archipelago.gen.cs`: SHA256 `9896bebec52112e53212b4368b08d97af80d16adc1b9777ea3b57589263e7b17` +- `dcl-protocol-1.0.0.tgz`: SHA256 `fa3f2c51c019d729d81a0c3993b384243ac615cc33def51479f797969c29b287` + +No runtime application or +LiveKit Cloud tests are claimed here. C# is generated by protoc; consumer Unity compilation remains +an integration gate. + +## Changed files + +- proto/decentraland/kernel/comms/v3/archipelago.proto +- package.json +- scripts/test.sh +- scripts/build-session-artifacts.ps1 +- test/session-control.test.js +- test/fixtures/session-control.json +- session-takeover-integration.md +- session-takeover-report.md + +## Gates and ownership + +No protocol collision was found. Connector identified existing KR_NEW_SESSION use for same-session +socket replacement: that must become close-only transport replacement or receive a separately +approved wire reason, since logical supersession is terminal. This decision was communicated to +connector and Unity owners; no extra enum was silently added. Protocol contains no ban producers. + +Consumer gate: audit bans, authoritative-only supersession, exact session subjects, replay and stale +operation fencing, client suppression before close, watchdog behavior, old-client compatibility, +and a real two-client LiveKit Cloud revocation scenario. The local archive is not a stable release. +Remote CI must consume an immutable reviewed commit/artifact, followed by approved stable +publication and coordinated pins. No push, publication, remote PR edit, or consumer re-pin occurred. diff --git a/test/fixtures/session-control.json b/test/fixtures/session-control.json new file mode 100644 index 00000000..c9bde6ff --- /dev/null +++ b/test/fixtures/session-control.json @@ -0,0 +1,13 @@ +[ + { "name": "legacy welcome", "type": "ServerPacket", "hex": "12030a0170", "json": { "welcome": { "peerId": "p" } } }, + { "name": "legacy assignment", "type": "ServerPacket", "hex": "1a060a0169120163", "json": { "islandChanged": { "islandId": "i", "connStr": "c" } } }, + { "name": "legacy supersession envelope", "type": "ServerPacket", "hex": "3200", "json": { "kicked": { "reason": "KR_NEW_SESSION" } } }, + { "name": "legacy bare supersession", "type": "KickedMessage", "hex": "", "json": { "reason": "KR_NEW_SESSION" } }, + { "name": "bare ban", "type": "KickedMessage", "hex": "0801", "json": { "reason": "KR_BANNED" } }, + { "name": "ban envelope", "type": "ServerPacket", "hex": "32020801", "json": { "kicked": { "reason": "KR_BANNED" } } }, + { "name": "bare pending 1500ms", "type": "SessionStatusMessage", "hex": "080110dc0b", "json": { "state": "TAKEOVER_PENDING", "retryAfterMs": 1500 } }, + { "name": "pending envelope", "type": "ServerPacket", "hex": "3a05080110dc0b", "json": { "sessionStatus": { "state": "TAKEOVER_PENDING", "retryAfterMs": 1500 } } }, + { "name": "bare failure", "type": "SessionStatusMessage", "hex": "0802", "json": { "state": "TAKEOVER_FAILED" } }, + { "name": "failure envelope", "type": "ServerPacket", "hex": "3a020802", "json": { "sessionStatus": { "state": "TAKEOVER_FAILED" } } }, + { "name": "unspecified envelope", "type": "ServerPacket", "hex": "3a00", "json": { "sessionStatus": {} } } +] diff --git a/test/session-control.test.js b/test/session-control.test.js new file mode 100644 index 00000000..e6a4f5b2 --- /dev/null +++ b/test/session-control.test.js @@ -0,0 +1,25 @@ +const assert = require('node:assert/strict') +const { ServerPacket, KickedMessage, SessionStatusMessage, SessionStatus, KickedReason } = require('../out-js/decentraland/kernel/comms/v3/archipelago.gen') +const fixtures = require('./fixtures/session-control.json') +for (const fixture of fixtures) { + const type = { ServerPacket, KickedMessage, SessionStatusMessage }[fixture.type] + assert.ok(type, fixture.type + ' must be generated') + const bytes = Buffer.from(fixture.hex, 'hex') + const expected = type.fromJSON(fixture.json) + assert.deepEqual(type.decode(bytes), expected, fixture.name + ': decode') + assert.deepEqual(Buffer.from(type.encode(expected).finish()), bytes, fixture.name + ': encode') +} +assert.equal(KickedReason.KR_NEW_SESSION, 0) +assert.equal(KickedReason.KR_BANNED, 1) +assert.equal(SessionStatus.UNSPECIFIED, 0) +assert.equal(SessionStatusMessage.decode(Buffer.alloc(0)).state, SessionStatus.UNSPECIFIED) +assert.equal(KickedMessage.decode(Buffer.alloc(0)).reason, KickedReason.KR_NEW_SESSION) +// Unknown binary enums must not be interpreted as confirmed supersession. +assert.equal(KickedMessage.decode(Buffer.from('0863', 'hex')).reason, 99) +assert.equal(SessionStatusMessage.decode(Buffer.from('0863', 'hex')).state, 99) +// A pre-amendment reader ignores the new oneof tag rather than interpreting it as an assignment. +const protobuf = require('protobufjs') +const oldPacket = protobuf.parse('syntax="proto3"; message Kicked { int32 reason=1; } message OldPacket { oneof message { Kicked kicked=6; } }').root.lookupType('OldPacket') +assert.equal(oldPacket.decode(Buffer.from('3a05080110dc0b', 'hex')).message, undefined) +assert.equal(oldPacket.decode(Buffer.from('3200', 'hex')).kicked.reason, 0) +console.log('PASS: ' + fixtures.length + ' wire goldens, defaults and unknown enums')