[DO NOT MERGE] feat(sdk): clients target the authoritative server; registry collisions; players dedupe

docs/REFERENCES.md

@@ -23,6 +23,7 @@ Curated map of every doc in this repo. Use this as the entry point when looking
 | [on-demand-composite-loading.md](on-demand-composite-loading.md) | Runtime composite instantiation via `engine.addEntityFromComposite(src, options)`. The provider abstraction and pre-registration flow. |
 | [material-getflat-api.md](material-getflat-api.md) | Material `getFlat` API: reading a fully-resolved material descriptor instead of the discriminated union. |
 | [network-timestamp-trust.md](network-timestamp-trust.md) | Why the authoritative server trusts client-supplied CRDT timestamps, the griefing that enables, and what server-side re-stamping would cost. |
+| [network-peer-visibility.md](network-peer-visibility.md) | Why the server can't exclude a sender from its broadcast and why a same-frame player join and leave goes unreported. Two defects that need a layer below the network code. |
 
 ## Guides (how-to)
 

docs/network-peer-visibility.md

@@ -0,0 +1,183 @@
+# Peer visibility limits in the network layer
+
+Two known defects in the authoritative network layer cannot be fixed inside
+that layer. Both come down to something the network code is never in a position
+to observe: who is connected to the room, and which components changed between
+two frames. This note records what was measured, why the obvious fixes are worse
+than the defects, and what a real fix would need.
+
+Both have a red test pinned as `it.failing` in
+`test/sdk/network/defects-red.spec.ts` (#11 and #13). Leave them failing until
+the underlying layer changes.
+
+## The server does not know who is connected
+
+`broadcastBatchedMessages` in `packages/@dcl/sdk/src/network/server/index.ts`
+takes an `excludeSender` argument and does nothing with it beyond logging. Every
+accepted client write is re-broadcast to the whole room, including back to the
+client that sent it. That client already applied the change locally, so the echo
+is a wasted round trip for every write in the scene.
+
+Removing the echo means addressing the broadcast to everyone except the sender.
+The comms API has no primitive for that: `PeerMessageData.address` in
+`packages/@dcl/js-runtime/apis.d.ts` is a list of recipients, and an empty list
+means broadcast. To exclude one peer, you must name all the others — which
+requires a roster the server does not have.
+
+### What the server actually holds
+
+Instrumenting the red #11 scenario (an authoritative server and two clients,
+where only the first client writes) shows everything available server-side after
+the write propagates:
+
+```
+players= []  avatars= 0  createdBy= ["clientA"]
+clientB-sent-ever= []
+```
+
+Three candidate rosters, and why each one fails:
+
+- **The players helper is empty.** `definePlayerHelper` reads
+  `PlayerIdentityData` and `AvatarBase`, which the renderer writes into a
+  scene's CRDT stream. An authoritative server has no renderer, and nothing in
+  this repository writes those components server-side. The query returns
+  nothing.
+- **Observed senders are incomplete by construction.** The server knows
+  `clientA` because `clientA` wrote something, recorded in `CreatedBy`. A peer
+  that only listens never appears. Excluding the sender from a roster built this
+  way would leave the room with nobody to address.
+- **`~system/Players.getConnectedPlayers` is unproven here.** The runtime API
+  exists in `packages/@dcl/js-runtime/apis.d.ts`, but nothing in the SDK calls
+  it, it's marked for deprecation, and the only headless host in this repository
+  (`packages/@dcl/sdk-commands/src/commands/code-to-composite/scene-executor.ts`)
+  doesn't mock `~system/Players` at all. There's no evidence the server host
+  implements it.
+
+### Why a partial roster is worse than the echo
+
+The failure modes aren't symmetric. If a roster comes back empty, the code falls
+back to broadcasting and the fix does nothing in production while passing in
+tests. If a roster comes back *partial*, the server stops addressing the peers
+it omits, and a client that never writes silently stops receiving world updates.
+
+Trading a measurable bandwidth cost for possible silent divergence is the wrong
+direction. Delivery-level exclusion is only safe on a roster known to be
+complete.
+
+### What a real fix needs
+
+Either of these makes the exclusion sound:
+
+- **A wire-level exclude.** A "broadcast except these addresses" option on
+  `sendBinary` is the correct primitive. It costs nothing per peer, needs no
+  roster in the scene, and can't go stale.
+- **A roster the server host commits to.** If the host guarantees
+  `getConnectedPlayers` reflects the comms room on a server runtime, pass it into
+  `addSyncTransport` and target the roster minus the sender, falling back to
+  broadcast whenever the roster is empty.
+
+## A player that joins and leaves in one frame is invisible
+
+The per-frame diff in `packages/@dcl/sdk/src/players/index.ts` compares the
+current `PlayerIdentityData` and `AvatarBase` entities against a cached map. A
+player whose entity is created and destroyed between two system runs never
+appears in either snapshot, so neither `onEnterScene` nor `onLeaveScene` fires.
+
+This isn't a matter of polling more carefully. Instrumenting the red #13
+scenario shows what survives to the next system run:
+
+```
+system-saw= ["rows=0 dirty=[512] has(512)=false val=null"]
+onChange-fired-on-local-writes= ["null"]
+```
+
+The entity id is still in the dirty set. The value is gone: `has()` is false and
+the component reads as null, so the player's address is unrecoverable.
+
+The second line rules out the event-driven alternative for local writes. That
+probe subscribed `onChange` up front on the exact entity id — something the real
+helper can never do, since the entity doesn't exist when the helper starts — and
+the callback still fired once, with `undefined`. `create`, `createOrReplace`,
+and `deleteFrom` in
+`packages/@dcl/ecs/src/engine/lww-element-set-component-definition.ts` never
+invoke `__onChangeCallbacks`. Those callbacks run from the flush in
+`packages/@dcl/ecs/src/engine/index.ts`, which reports the settled state, and a
+create and a delete in the same frame collapse to a single notification. The
+outgoing wire collapses the same way: `getCrdtUpdates` emits only a
+`DELETE_COMPONENT`, so the value never reaches another peer either.
+
+### The wire path is different, and it defines the requirement
+
+A player never arrives through local writes in production. The renderer delivers
+`PlayerIdentityData` over CRDT, so the case that matters is a `PUT_COMPONENT`
+and a `DELETE_COMPONENT` for the same entity arriving in one batch. Feeding
+exactly that through a transport, with `onChange` again pre-subscribed on the
+entity id, gives a different result:
+
+```
+onChange-fired= ["{\"address\":\"0xplayer\",\"isGuest\":false}", "null"]
+system-saw= ["rows=0"]
+```
+
+The receive path applies each message in turn and notifies for each, so the
+callback fires twice and **the first one carries the address**. The value is
+genuinely observable on the wire path; it's the poll that misses it, because by
+the time a system runs the entity is gone again.
+
+That difference is the whole follow-up. Nothing prevents the players helper from
+seeing a renderer-delivered join and leave in the same frame except the shape of
+the subscription: `onChange` is registered per entity, and an entity that
+doesn't exist yet can't be subscribed to. The requirement on `@dcl/ecs` is
+therefore a **component-wide change subscription** — "call me for every change to
+`PlayerIdentityData`, whatever the entity" — not a change to when notifications
+fire. With that in place, every player that arrives the way real players arrive
+is reported correctly, and only the purely local same-frame case remains
+unreachable.
+
+Note that red #13's fixture uses local writes, so it exercises the one variant
+that stays unfixable even after the subscription lands. Rewrite the fixture to
+drive the component over a transport when picking this up.
+
+### What was fixed, and what wasn't
+
+The diff loop is now a mark-and-sweep poll over both directions, which fixes two
+real bugs the old shortcut hid:
+
+- A frame in which one player joins and another leaves keeps the entity count
+  equal, and the old `players.length === playerEntities.size` check skipped the
+  whole frame, missing the join.
+- Departures used to be reported by a per-entity `AvatarBase.onChange`
+  registered when the player was first seen, which never fired for a player
+  whose entity was removed whole. The sweep uses the cached address instead.
+
+Every player visible for at least one frame is now reported correctly. A player
+who comes and goes inside a single frame is still missed, and stays missed until
+the component-wide subscription described above exists.
+
+## What not to do now
+
+Do not make either test pass by narrowing the fix to the test:
+
+- **Do not build the roster from observed senders.** It reaches only peers that
+  have already written, which is precisely the set that doesn't need the echo.
+- **Do not patch `PlayerIdentityData.create` from inside the players helper.**
+  Local writes are the only calls it would intercept, and renderer-driven
+  players arrive through `updateFromCrdt`. The test goes green and production is
+  unchanged.
+- **Do not fire `onChange` synchronously on local writes** as a targeted fix. It
+  redefines `onChange` for every scene, in a package held to 100% coverage, and
+  it addresses the wrong half of the problem: on the path players actually
+  arrive by, the notification already carries the value. Add the component-wide
+  subscription instead.
+
+## Related
+
+- `packages/@dcl/sdk/src/network/server/index.ts` — `broadcastBatchedMessages`
+  and the validator.
+- `packages/@dcl/sdk/src/players/index.ts` — the player-diff system.
+- `packages/@dcl/ecs/src/engine/lww-element-set-component-definition.ts` —
+  component writes, dirty tracking, and change callbacks.
+- `packages/@dcl/ecs/src/engine/index.ts` — where `__onChangeCallbacks` runs, and
+  why the receive path notifies per message while the flush reports net state.
+- [network-timestamp-trust.md](network-timestamp-trust.md) — a second trust
+  boundary in the same layer.

packages/@dcl/sdk/src/network/events/implementation.ts

@@ -122,7 +122,7 @@ export class Room<T extends EventSchemaRegistry = EventSchemaRegistry> {
         this.binaryMessageBus.emit(CommsMessage.CUSTOM_EVENT, buffer, options?.to)
       } else {
         // Client always sends to authoritative server
-        this.binaryMessageBus.emit(CommsMessage.CUSTOM_EVENT, buffer)
+        this.binaryMessageBus.emit(CommsMessage.CUSTOM_EVENT, buffer, [AUTH_SERVER_PEER_ID])
       }
     } catch (error) {
       console.error(`[EventBus] Failed to send event '${String(eventType)}':`, error)
@@ -234,14 +234,34 @@ export function setGlobalRoom(roomInstance: Room): void {
   globalRoom = roomInstance
 }
 
+/** message keys the SDK keeps for itself, so its own traffic can never clash with a scene's */
+export const RESERVED_MESSAGE_PREFIX = '~sdk/'
+
 /**
  * Register message schemas for use with the room
  * Call this before main() to define your custom messages
+ *
+ * The registry is flat and global: a key names the same message for every peer in
+ * the room. Registering one twice is therefore reported — the last schema wins,
+ * and any peer still holding the earlier one decodes the payload as garbage.
+ * Keys under `~sdk/` are reserved for SDK-internal messages and are not registered.
+ *
  * @param messages - Object containing your message schemas
  * @returns Typed room instance for your registered messages
  */
 export function registerMessages<T extends EventSchemaRegistry>(messages: T): Room<T> {
-  Object.assign(globalEventRegistry, messages)
+  for (const [key, schema] of Object.entries(messages)) {
+    // reported through `error` and not `warn`: the scene runtime's console has
+    // only `log` and `error`
+    if (key.startsWith(RESERVED_MESSAGE_PREFIX)) {
+      console.error(`[Room] ignoring '${key}': the '${RESERVED_MESSAGE_PREFIX}' prefix is reserved for the SDK`)
+      continue
+    }
+    if (key in globalEventRegistry) {
+      console.error(`[Room] message '${key}' is already registered; the schema registered last is the one that decodes`)
+    }
+    globalEventRegistry[key] = schema
+  }
   if (!globalRoom) {
     throw new Error('Room not initialized. Make sure the SDK network transport is initialized.')
   }

packages/@dcl/sdk/src/network/message-bus-sync.ts

@@ -8,7 +8,7 @@ import { fetchProfile } from './utils'
 import { entityUtils } from './entities'
 import { createServerValidator } from './server'
 import { GetUserDataRequest, GetUserDataResponse } from '~system/UserIdentity'
-import { definePlayerHelper } from '../players'
+import { getPlayerHelper } from '../players'
 import { serializeCrdtMessages } from '../internal/transports/logger'
 import { IsServerRequest, IsServerResponse } from '~system/EngineApi'
 import { AUTH_SERVER_PEER_ID, DEBUG_NETWORK_MESSAGES, IProfile } from './constants'
@@ -69,7 +69,7 @@ export function addSyncTransport(
     pendingMessageBusMessagesToSend.length = 0
     return messages
   }
-  const players = definePlayerHelper(engine)
+  const players = getPlayerHelper(engine)
 
   const RealmInfo = components.RealmInfo(engine)
   const NetworkEntity = components.NetworkEntity(engine)
@@ -85,6 +85,20 @@ export function addSyncTransport(
    */
   let tick = 0
   const TRANSPORT_INITIALIZED_NUMBER = isTestEnvironment() ? 0 : 2
+
+  /**
+   * Who this peer's own CRDT is addressed to. A client only ever talks to the
+   * authoritative server; the server's fan-out to the room is legitimate, so it
+   * keeps broadcasting (`undefined` — an empty address list means broadcast).
+   *
+   * Comms is buffered until the role resolves, so nothing *received* is handled
+   * before it is known. `transport.send` is not: it runs on the engine clock and
+   * can reach here first. Broadcasting in that window is the pre-role fallback,
+   * not a leftover of the peer-to-peer topology.
+   */
+  function crdtAudience(): string[] | undefined {
+    return isServerAtom.getOrNull() === false ? [AUTH_SERVER_PEER_ID] : undefined
+  }
   // Add Sync Transport
   const transport: Transport = {
     filter: syncFilter(engine),
@@ -97,7 +111,7 @@ export function addSyncTransport(
 
           // Convert regular messages to network messages for broadcasting with chunking
           for (const chunk of serverValidator.convertRegularToNetworkMessage(message)) {
-            binaryMessageBus.emit(CommsMessage.CRDT, chunk)
+            binaryMessageBus.emit(CommsMessage.CRDT, chunk, crdtAudience())
           }
         }
       }
@@ -159,7 +173,8 @@ export function addSyncTransport(
     for (const effect of effects) {
       if (effect === 'requestState') {
         DEBUG_NETWORK_MESSAGES() && console.log('Requesting state...')
-        binaryMessageBus.emit(CommsMessage.REQ_CRDT_STATE, new Uint8Array())
+        // unconditionally addressed: the FSM only ever asks for state as a client
+        binaryMessageBus.emit(CommsMessage.REQ_CRDT_STATE, new Uint8Array(), [AUTH_SERVER_PEER_ID])
       } else if (effect === 'markSynced') {
         // the room only counts as ready once comms has answered at least once
         if (RealmInfo.getOrNull(engine.RootEntity)) {

packages/@dcl/sdk/src/players/index.ts

@@ -32,27 +32,22 @@ export function definePlayerHelper(engine: IEngine) {
   const onEnterSceneCb: ((player: GetPlayerDataRes) => void)[] = []
   const onLeaveSceneCb: ((userId: string) => void)[] = []
 
+  // Both directions are polled. The `players.length === playerEntities.size`
+  // shortcut this replaces went blind on a frame where one player joined and
+  // another left, and the per-entity `AvatarBase.onChange` that used to report a
+  // departure never fired for a player whose entity was removed whole.
   engine.addSystem(() => {
-    const players = Array.from(engine.getEntitiesWith(PlayerIdentityData, AvatarBase))
-    if (players.length === playerEntities.size) return
-
-    for (const [entity, identity] of players) {
-      if (!playerEntities.has(entity)) {
-        playerEntities.set(entity, identity.address)
-
-        // Call onEnter callback
-        if (onEnterSceneCb.length) {
-          onEnterSceneCb.forEach((cb) => cb(getPlayer({ userId: identity.address })!))
-        }
-
-        // Check for changes/remove callbacks
-        AvatarBase.onChange(entity, (value) => {
-          if (!value && playerEntities.get(entity)) {
-            onLeaveSceneCb.forEach((cb) => cb(playerEntities.get(entity)!))
-            playerEntities.delete(entity)
-          }
-        })
-      }
+    const present = new Set<Entity>()
+    for (const [entity, identity] of engine.getEntitiesWith(PlayerIdentityData, AvatarBase)) {
+      present.add(entity)
+      if (playerEntities.has(entity)) continue
+      playerEntities.set(entity, identity.address)
+      for (const cb of onEnterSceneCb) cb(getPlayer({ userId: identity.address })!)
+    }
+    for (const [entity, address] of playerEntities) {
+      if (present.has(entity)) continue
+      playerEntities.delete(entity)
+      for (const cb of onLeaveSceneCb) cb(address)
     }
   })
 
@@ -100,7 +95,23 @@ export function definePlayerHelper(engine: IEngine) {
   }
 }
 
-const players = definePlayerHelper(engine)
+type PlayerHelper = ReturnType<typeof definePlayerHelper>
+const helpers = new WeakMap<IEngine, PlayerHelper>()
+
+/**
+ * One helper per engine. Both the public API below and `addSyncTransport` need
+ * one, and two of them on the same engine means two identical per-frame diffs
+ * and every `onEnterScene` callback firing twice.
+ */
+export function getPlayerHelper(engine: IEngine): PlayerHelper {
+  const existing = helpers.get(engine)
+  if (existing) return existing
+  const helper = definePlayerHelper(engine)
+  helpers.set(engine, helper)
+  return helper
+}
+
+const players = getPlayerHelper(engine)
 const { getPlayer, onEnterScene, onLeaveScene } = players
 
 export { getPlayer, onEnterScene, onLeaveScene }

test/sdk/network/characterization-flow.spec.ts

@@ -40,9 +40,7 @@ describe('network flow characterization', () => {
 
     const requests = harness.sentBy(CLIENT_A, CommsMessage.REQ_CRDT_STATE)
     expect(requests.length).toBeGreaterThanOrEqual(1)
-    // QUIRK(pinned): the state request is broadcast to every peer instead of
-    // being addressed to the authoritative server — see defect #1/#10.
-    expect(requests[0].to).toEqual([])
+    expect(requests[0].to).toEqual([SERVER])
 
     const responses = harness.sentBy(SERVER, CommsMessage.RES_CRDT_STATE)
     expect(responses.map((response) => response.to)).toEqual(expect.arrayContaining([[CLIENT_A], [CLIENT_B]]))
@@ -76,9 +74,6 @@ describe('network flow characterization', () => {
     )[0]
     expect(clientB.components.Transform.get(clientBEntity).position).toMatchObject({ x: 1, y: 2, z: 3 })
 
-    // QUIRK(pinned): a client emits CRDT with an empty address list (broadcast)
-    // rather than addressing the server — see defect #10.
-    expect(harness.sentBy(CLIENT_A, CommsMessage.CRDT)[0].to).toEqual([])
     // QUIRK(pinned): the server re-broadcasts to everybody, so the originating
     // client receives an echo of its own write — see defect #11.
     expect(harness.sentBy(SERVER, CommsMessage.CRDT)[0].to).toEqual([])