diff --git a/packages/@dcl/sdk/src/network/clock-sync.ts b/packages/@dcl/sdk/src/network/clock-sync.ts new file mode 100644 index 000000000..a8060977d --- /dev/null +++ b/packages/@dcl/sdk/src/network/clock-sync.ts @@ -0,0 +1,223 @@ +import { Schemas } from '@dcl/ecs' +import { Room } from './events/implementation' + +/** + * Clock synchronization between clients and the authoritative server. + * + * Every client periodically sends a ping stamped with its own clock; the server answers with a pong that echoes + * that stamp and adds its own clock; the client acknowledges with the server stamp so the server can measure the + * round trip too. From each pong the client derives the round-trip time and a clock offset (server minus local at + * the midpoint of the trip). The offset with the lowest round trip in a rolling window wins, because a fast trip + * is the least distorted by queueing. + * + * Scenes read `getServerTime()` for a clock every participant agrees on, and the server reads `getPlayerLatency()` + * to bound how early a client can claim an input happened. + * + * @public + */ +export type ClockSyncStats = { + /** Estimated server clock minus local clock, in milliseconds. */ + offset: number + /** Latest measured round trip, in milliseconds. */ + rtt: number + /** Lowest round trip among the samples the offset was picked from. */ + bestRtt: number + /** Number of samples in the window. */ + samples: number + /** True while the offset is held fixed (see `freeze`). */ + frozen: boolean +} + +/** Reserved message names; scenes cannot collide with them because they start with a double underscore. */ +export const CLOCK_SYNC_MESSAGES = { + __sdkClockPing: Schemas.Map({ clientAt: Schemas.Int64 }), + __sdkClockPong: Schemas.Map({ clientAt: Schemas.Int64, serverAt: Schemas.Int64 }), + __sdkClockAck: Schemas.Map({ serverAt: Schemas.Int64 }) +} +type ClockRoom = Room + +export const CLOCK_SYNC_DEFAULTS = { + /** How often a client pings, in milliseconds. */ + pingIntervalMs: 1500, + /** Samples kept per client; the best of these decides the offset. */ + windowSize: 10, + /** Round trips above this are dropped as outliers, in milliseconds. */ + maxRttMs: 2000, + /** The server answers at most one ping per client in this many milliseconds. */ + serverMinPingIntervalMs: 800, + /** Server-side round trips are clamped to this, in milliseconds. */ + serverMaxRttMs: 500 +} + +/** + * Pure estimator: feeds on pong samples, keeps the best of a rolling window. Testable without a room. + * @internal + */ +export class ClockEstimator { + private samples: { rtt: number; offset: number }[] = [] + private lastRtt = 0 + private frozenOffset: number | undefined + private liveOffset: number | undefined + + constructor( + private readonly windowSize = CLOCK_SYNC_DEFAULTS.windowSize, + private readonly maxRttMs = CLOCK_SYNC_DEFAULTS.maxRttMs + ) {} + + /** Returns true when the sample was accepted. */ + addSample(clientAt: number, serverAt: number, receivedAt: number): boolean { + const rtt = receivedAt - clientAt + if (rtt < 0 || rtt > this.maxRttMs) return false + this.lastRtt = rtt + this.samples.push({ rtt, offset: serverAt - (clientAt + rtt / 2) }) + if (this.samples.length > this.windowSize) this.samples.shift() + this.liveOffset = this.best().offset + return true + } + + private best() { + return this.samples.reduce((best, sample) => (sample.rtt < best.rtt ? sample : best)) + } + + get ready() { + return this.liveOffset !== undefined + } + + /** The offset in force: the frozen one while frozen, otherwise the live estimate. */ + get offset(): number | undefined { + return this.frozenOffset ?? this.liveOffset + } + + /** + * Hold the current offset fixed so the scene clock cannot jump while something time-critical runs, + * for example a synchronized performance. Samples keep flowing so the estimate is fresh when unfrozen. + */ + freeze(frozen: boolean) { + if (frozen && this.liveOffset !== undefined) this.frozenOffset = this.liveOffset + if (!frozen) this.frozenOffset = undefined + } + + reset() { + this.samples = [] + this.lastRtt = 0 + this.frozenOffset = undefined + this.liveOffset = undefined + } + + stats(): ClockSyncStats | undefined { + if (this.liveOffset === undefined || this.offset === undefined) return undefined + return { + offset: this.offset, + rtt: this.lastRtt, + bestRtt: this.best().rtt, + samples: this.samples.length, + frozen: this.frozenOffset !== undefined + } + } +} + +/** + * Server-side bookkeeping: rate limits pings and keeps each player's measured round trip. + * @internal + */ +export class ServerClockRegistry { + private probes = new Map() + private rtts = new Map() + + constructor( + private readonly minPingIntervalMs = CLOCK_SYNC_DEFAULTS.serverMinPingIntervalMs, + private readonly maxRttMs = CLOCK_SYNC_DEFAULTS.serverMaxRttMs + ) {} + + /** Returns the server stamp to answer with, or undefined when the ping is rate limited. */ + onPing(from: string, now: number): number | undefined { + const key = from.toLowerCase() + if (now - (this.probes.get(key) ?? -Infinity) < this.minPingIntervalMs) return undefined + this.probes.set(key, now) + return now + } + + /** Records the round trip when the ack matches the last stamp sent to that player. */ + onAck(from: string, serverAt: number, now: number) { + const key = from.toLowerCase() + if (this.probes.get(key) !== serverAt) return + this.rtts.set(key, Math.max(0, Math.min(this.maxRttMs, now - serverAt))) + } + + latency(from: string): number | undefined { + return this.rtts.get(from.toLowerCase()) + } + + forget(from: string) { + const key = from.toLowerCase() + this.probes.delete(key) + this.rtts.delete(key) + } +} + +/** + * Wires the exchange onto a room. Call once from the sync transport; the returned object is the scene-facing API. + * @internal + */ +export function createClockSync( + room: ClockRoom, + isServer: () => boolean | null, + now: () => number = () => Date.now(), + options: Partial = {} +) { + const settings = { ...CLOCK_SYNC_DEFAULTS, ...options } + const estimator = new ClockEstimator(settings.windowSize, settings.maxRttMs) + const registry = new ServerClockRegistry(settings.serverMinPingIntervalMs, settings.serverMaxRttMs) + let lastPingAt = -Infinity + let pendingClientAt: number | undefined + + room.onMessage('__sdkClockPing', (data, context) => { + if (!isServer() || !context) return + const serverAt = registry.onPing(context.from, now()) + if (serverAt === undefined) return + void room.send('__sdkClockPong', { clientAt: data.clientAt, serverAt }, { to: [context.from] }) + }) + room.onMessage('__sdkClockAck', (data, context) => { + if (!isServer() || !context) return + registry.onAck(context.from, data.serverAt, now()) + }) + room.onMessage('__sdkClockPong', (data) => { + if (isServer()) return + // Only the reply to the most recent ping is trusted; late replies would carry stale trips. + if (data.clientAt !== pendingClientAt) return + if (estimator.addSample(data.clientAt, data.serverAt, now())) + void room.send('__sdkClockAck', { serverAt: data.serverAt }) + }) + + return { + /** Drive from a system every frame: sends a ping when one is due. Clients only. */ + update() { + if (isServer() !== false || !room.isReady()) return + const t = now() + if (t - lastPingAt < settings.pingIntervalMs) return + lastPingAt = t + pendingClientAt = t + void room.send('__sdkClockPing', { clientAt: t }) + }, + /** Server clock estimate in milliseconds since the epoch, or undefined before the first sample. */ + getServerTime(): number | undefined { + const offset = estimator.offset + return offset === undefined ? undefined : now() + offset + }, + getStats: () => estimator.stats(), + freeze: (frozen: boolean) => estimator.freeze(frozen), + reset: () => { + estimator.reset() + pendingClientAt = undefined + lastPingAt = -Infinity + }, + /** Server only: a player's last measured round trip in milliseconds, or undefined if never acknowledged. */ + getPlayerLatency: (userId: string) => registry.latency(userId), + forgetPlayer: (userId: string) => registry.forget(userId), + /** @internal exposed for tests */ + estimator, + registry + } +} + +export type ClockSync = ReturnType diff --git a/packages/@dcl/sdk/src/network/index.ts b/packages/@dcl/sdk/src/network/index.ts index 6ba6e2e6b..0d4daa6ec 100644 --- a/packages/@dcl/sdk/src/network/index.ts +++ b/packages/@dcl/sdk/src/network/index.ts @@ -27,9 +27,48 @@ const { getFirstChild, isStateSyncronized, binaryMessageBus, - eventBus + eventBus, + clockSync } = addSyncTransport(engine, sendBinary, getUserData, isServerApi, 'network') +/** + * Estimated authoritative-server clock, in milliseconds since the epoch, or undefined until the first + * ping has been answered. Every client in the room converges on this clock, so it is the one to schedule + * shared events against (countdowns, song positions, race starts). The server returns its own clock. + * @public + */ +export function getServerTime(): number | undefined { + return isServer() ? Date.now() : clockSync.getServerTime() +} + +/** + * Quality of the current server-clock estimate: offset, round trip, sample count, frozen flag. + * Undefined until the first ping has been answered. Clients only. + * @public + */ +export function getClockSyncStats() { + return clockSync.getStats() +} + +/** + * Hold the server-clock estimate fixed so `getServerTime()` cannot jump while something time-critical runs, + * for example a synchronized performance. Pass false to resume following the live estimate. Clients only. + * @public + */ +export function freezeServerClock(frozen: boolean) { + clockSync.freeze(frozen) +} + +/** + * Server only: the last measured round trip to a connected player in milliseconds, or undefined if that + * player has not acknowledged a clock reply yet. Use half of it to bound how early a client may claim an + * input happened relative to the server's own arrival time. + * @public + */ +export function getPlayerLatency(userId: string): number | undefined { + return clockSync.getPlayerLatency(userId) +} + // Re-export the room messaging system export { registerMessages, getRoom } from './events' diff --git a/packages/@dcl/sdk/src/network/message-bus-sync.ts b/packages/@dcl/sdk/src/network/message-bus-sync.ts index ef32a7817..97bd5ada1 100644 --- a/packages/@dcl/sdk/src/network/message-bus-sync.ts +++ b/packages/@dcl/sdk/src/network/message-bus-sync.ts @@ -12,7 +12,8 @@ import { definePlayerHelper } from '../players' import { serializeCrdtMessages } from '../internal/transports/logger' import { IsServerRequest, IsServerResponse } from '~system/EngineApi' import { Atom } from '../atom' -import { setGlobalRoom, Room } from './events/implementation' +import { setGlobalRoom, Room, getEventRegistry } from './events/implementation' +import { CLOCK_SYNC_MESSAGES, createClockSync } from './clock-sync' export type IProfile = { networkId: number; userId: string } // user that we asked for the inital crdt state @@ -110,6 +111,12 @@ export function addSyncTransport( // Set global eventBus instance setGlobalRoom(eventBus) + // Built-in clock synchronization rides on the room with reserved message names + Object.assign(getEventRegistry(), CLOCK_SYNC_MESSAGES) + const clockSync = createClockSync(eventBus as unknown as Room, () => + isServerAtom.getOrNull() + ) + engine.addTransport(transport) // End add sync transport @@ -242,6 +249,7 @@ export function addSyncTransport( // System to retry state request if no response is received within the retry interval engine.addSystem((dt: number) => { + clockSync.update() if (requestingState && !stateIsSyncronized) { elapsedTimeSinceRequest += dt if (elapsedTimeSinceRequest >= STATE_REQUEST_RETRY_INTERVAL) { @@ -255,6 +263,7 @@ export function addSyncTransport( players.onLeaveScene((userId) => { DEBUG_NETWORK_MESSAGES() && console.log('[onLeaveScene]', userId) + clockSync.forgetPlayer(userId) }) function isStateSyncronized() { @@ -267,6 +276,7 @@ export function addSyncTransport( isStateSyncronized, binaryMessageBus, eventBus, - isRoomReadyAtom + isRoomReadyAtom, + clockSync } } diff --git a/test/sdk/network/clock-sync.spec.ts b/test/sdk/network/clock-sync.spec.ts new file mode 100644 index 000000000..8e047d927 --- /dev/null +++ b/test/sdk/network/clock-sync.spec.ts @@ -0,0 +1,325 @@ +import { + ClockEstimator, + ServerClockRegistry, + createClockSync, + CLOCK_SYNC_MESSAGES +} from '../../../packages/@dcl/sdk/src/network/clock-sync' +import type { Room, EventContext } from '../../../packages/@dcl/sdk/src/network/events/implementation' + +type ClockRoom = Room +type Listener = (data: any, context?: EventContext) => void + +/** A room double that records sends and lets the test deliver messages to listeners. */ +function fakeRoom(initiallyReady = true) { + let ready = initiallyReady + const listeners = new Map>() + const sent: { eventType: string; data: any; options?: { to?: string[] } }[] = [] + const room = { + isReady: () => ready, + setReady: (value: boolean) => { + ready = value + }, + send: jest.fn(async (eventType: string, data: any, options?: { to?: string[] }) => { + sent.push({ eventType, data, options }) + }), + onMessage: jest.fn((eventType: string, cb: Listener) => { + if (!listeners.has(eventType)) listeners.set(eventType, new Set()) + listeners.get(eventType)!.add(cb) + return () => listeners.get(eventType)!.delete(cb) + }), + deliver(eventType: string, data: any, context?: EventContext) { + listeners.get(eventType)?.forEach((cb) => cb(data, context)) + }, + sent + } + return room +} + +describe('ClockEstimator', () => { + let estimator: ClockEstimator + + beforeEach(() => { + estimator = new ClockEstimator(3, 2000) + }) + + describe('when a pong arrives', () => { + beforeEach(() => { + // ping at 1000, server answered at 5100, reply received at 1200: rtt 200, offset 5100 - 1100 = 4000 + estimator.addSample(1000, 5100, 1200) + }) + + it('should estimate the offset from the midpoint of the round trip', () => { + expect(estimator.offset).toBe(4000) + }) + + it('should report the sample count and round trip', () => { + expect(estimator.stats()).toEqual({ offset: 4000, rtt: 200, bestRtt: 200, samples: 1, frozen: false }) + }) + }) + + describe('when several pongs arrive with different round trips', () => { + beforeEach(() => { + estimator.addSample(1000, 5100, 1200) // rtt 200, offset 4000 + estimator.addSample(2000, 6250, 2500) // rtt 500, offset 4000 - but noisier: 6250 - 2250 = 4000 + estimator.addSample(3000, 7040, 3080) // rtt 80, offset 7040 - 3040 = 4000 + }) + + it('should pick the offset from the fastest round trip', () => { + expect(estimator.stats()?.bestRtt).toBe(80) + }) + + it('should drop the oldest sample once the window is full', () => { + estimator.addSample(4000, 8150, 4300) // rtt 300, offset 4000 + expect(estimator.stats()?.samples).toBe(3) + }) + }) + + describe('when a reply is implausible', () => { + it('should reject a negative round trip', () => { + expect(estimator.addSample(1000, 5000, 900)).toBe(false) + }) + + it('should reject a round trip above the maximum', () => { + expect(estimator.addSample(1000, 5000, 3500)).toBe(false) + }) + + it('should stay not ready without an accepted sample', () => { + estimator.addSample(1000, 5000, 900) + expect(estimator.ready).toBe(false) + }) + }) + + describe('when the offset is frozen', () => { + beforeEach(() => { + estimator.addSample(1000, 5100, 1200) // rtt 200, offset 4000 + estimator.freeze(true) + estimator.addSample(3000, 7120, 3040) // rtt 40, offset 7120 - 3020 = 4100: the fastest, so the live estimate moves + }) + + it('should keep reporting the offset it had when frozen', () => { + expect(estimator.offset).toBe(4000) + }) + + it('should flag the stats as frozen', () => { + expect(estimator.stats()?.frozen).toBe(true) + }) + + it('should switch to the fresh estimate once unfrozen', () => { + estimator.freeze(false) + expect(estimator.offset).toBe(4100) + }) + }) +}) + +describe('ServerClockRegistry', () => { + let registry: ServerClockRegistry + + beforeEach(() => { + registry = new ServerClockRegistry(800, 500) + }) + + describe('when a client pings twice quickly', () => { + let first: number | undefined + let second: number | undefined + + beforeEach(() => { + first = registry.onPing('0xabc', 10_000) + second = registry.onPing('0xabc', 10_300) + }) + + it('should answer the first ping with the server stamp', () => { + expect(first).toBe(10_000) + }) + + it('should rate limit the second ping', () => { + expect(second).toBeUndefined() + }) + }) + + describe('when the client acknowledges the stamp it was sent', () => { + beforeEach(() => { + registry.onPing('0xabc', 10_000) + registry.onAck('0xabc', 10_000, 10_180) + }) + + it('should record that player round trip', () => { + expect(registry.latency('0xabc')).toBe(180) + }) + + it('should ignore an ack for a stamp that was never sent', () => { + registry.onAck('0xabc', 9_999, 10_400) + expect(registry.latency('0xabc')).toBe(180) + }) + + it('should clamp an absurd round trip', () => { + registry.onPing('0xabc', 20_000) + registry.onAck('0xabc', 20_000, 21_900) + expect(registry.latency('0xabc')).toBe(500) + }) + + it('should forget the player on request', () => { + registry.forget('0xabc') + expect(registry.latency('0xabc')).toBeUndefined() + }) + }) +}) + +describe('createClockSync', () => { + let now: number + const clock = () => now + + describe('when running on a client', () => { + let room: ReturnType + let sync: ReturnType + + beforeEach(() => { + now = 100_000 + room = fakeRoom(true) + sync = createClockSync(room as unknown as ClockRoom, () => false, clock, { pingIntervalMs: 1500 }) + }) + + afterEach(() => { + jest.restoreAllMocks() + }) + + it('should not know the server time before any reply', () => { + expect(sync.getServerTime()).toBeUndefined() + }) + + it('should send a ping stamped with the local clock on the first update', () => { + sync.update() + expect(room.sent).toEqual([{ eventType: '__sdkClockPing', data: { clientAt: 100_000 }, options: undefined }]) + }) + + it('should not ping again before the interval elapses', () => { + sync.update() + now += 1000 + sync.update() + expect(room.sent).toHaveLength(1) + }) + + it('should ping again once the interval elapsed', () => { + sync.update() + now += 1500 + sync.update() + expect(room.sent.map((m) => m.data.clientAt)).toEqual([100_000, 101_500]) + }) + + it('should not ping while the room is not ready', () => { + room.setReady(false) + sync.update() + expect(room.sent).toHaveLength(0) + }) + + describe('and the server replies to the latest ping', () => { + beforeEach(() => { + sync.update() // ping at 100_000 + now = 100_200 + room.deliver('__sdkClockPong', { clientAt: 100_000, serverAt: 500_100 }) + }) + + it('should estimate the server time from the reply', () => { + // offset = 500_100 - (100_000 + 100) = 400_000 + expect(sync.getServerTime()).toBe(500_200) + }) + + it('should acknowledge the server stamp so the server can measure its round trip', () => { + expect(room.sent[1]).toEqual({ eventType: '__sdkClockAck', data: { serverAt: 500_100 }, options: undefined }) + }) + + it('should expose the measured round trip', () => { + expect(sync.getStats()?.rtt).toBe(200) + }) + }) + + describe('and a reply to an older ping arrives', () => { + beforeEach(() => { + sync.update() // ping at 100_000 + now = 101_500 + sync.update() // ping at 101_500 (the pending one) + now = 101_600 + room.deliver('__sdkClockPong', { clientAt: 100_000, serverAt: 500_100 }) + }) + + it('should ignore the stale reply', () => { + expect(sync.getServerTime()).toBeUndefined() + }) + }) + + describe('and the offset is frozen', () => { + beforeEach(() => { + sync.update() + now = 100_200 + room.deliver('__sdkClockPong', { clientAt: 100_000, serverAt: 500_100 }) // offset 400_000 + sync.freeze(true) + now = 101_500 + sync.update() + now = 101_540 + room.deliver('__sdkClockPong', { clientAt: 101_500, serverAt: 501_570 }) // rtt 40, offset 400_050 + }) + + it('should keep the frozen offset', () => { + expect(sync.getServerTime()).toBe(101_540 + 400_000) + }) + + it('should adopt the newer estimate after unfreezing', () => { + sync.freeze(false) + expect(sync.getServerTime()).toBe(101_540 + 400_050) + }) + }) + + it('should never answer pings itself', () => { + room.deliver('__sdkClockPing', { clientAt: 1 }, { from: '0xother' }) + expect(room.sent).toHaveLength(0) + }) + }) + + describe('when running on the server', () => { + let room: ReturnType + let sync: ReturnType + + beforeEach(() => { + now = 500_000 + room = fakeRoom(true) + sync = createClockSync(room as unknown as ClockRoom, () => true, clock) + }) + + it('should never ping', () => { + sync.update() + expect(room.sent).toHaveLength(0) + }) + + describe('and a client pings', () => { + beforeEach(() => { + room.deliver('__sdkClockPing', { clientAt: 100_000 }, { from: '0xAbC' }) + }) + + it('should answer only that client with both stamps', () => { + expect(room.sent).toEqual([ + { eventType: '__sdkClockPong', data: { clientAt: 100_000, serverAt: 500_000 }, options: { to: ['0xAbC'] } } + ]) + }) + + it('should record the round trip when the client acknowledges', () => { + now = 500_150 + room.deliver('__sdkClockAck', { serverAt: 500_000 }, { from: '0xAbC' }) + expect(sync.getPlayerLatency('0xabc')).toBe(150) + }) + + it('should rate limit a second ping from the same client', () => { + now = 500_300 + room.deliver('__sdkClockPing', { clientAt: 100_300 }, { from: '0xAbC' }) + expect(room.sent).toHaveLength(1) + }) + }) + + it('should ignore a ping without sender context', () => { + room.deliver('__sdkClockPing', { clientAt: 100_000 }) + expect(room.sent).toHaveLength(0) + }) + + it('should not know a latency for a player that never acknowledged', () => { + expect(sync.getPlayerLatency('0xnobody')).toBeUndefined() + }) + }) +})