diff --git a/api/latica/api.js b/api/latica/api.js index 26728f2d5..22108cbe5 100644 --- a/api/latica/api.js +++ b/api/latica/api.js @@ -356,7 +356,8 @@ async function api (options = {}, events, dgram) { ee._peer = peer sub.peers.set(peer.peerId, ee) - if (!oldPeer || change) sub._emit('#join', ee, packet) + const isStateChange = !oldPeer || change + sub._emit('#join', ee, packet, isStateChange) }) const handlePacket = async (packet, peer, port, address) => { @@ -378,7 +379,7 @@ async function api (options = {}, events, dgram) { bus._on('#packet', handlePacket) bus._on('#disconnection', peer => { - for (const sub of bus.subclusters) { + for (const sub of [...bus.subclusters.values()]) { sub._emit('#leave', peer) sub.peers.delete(peer.peerId) } diff --git a/api/latica/index.js b/api/latica/index.js index cdad3fb70..eeba6c849 100644 --- a/api/latica/index.js +++ b/api/latica/index.js @@ -1106,7 +1106,8 @@ export class Peer { // if we are out of sync send our cache summary const data = await Packet.encode(new PacketSync({ - message: Cache.encodeSummary(summary) + message: Cache.encodeSummary(summary), + usr4: Buffer.from(String(Date.now())) })) this.send(data, rinfo.port, rinfo.address, peer.socket) @@ -1157,7 +1158,12 @@ export class Peer { * */ cachePredicate (packet) { - return packet.version === VERSION && packet.timestamp > Date.now() - Packet.ttl + if (packet.usr4.byteLength < 8 || packet.usr4.byteLength > 16) return + + const timestamp = parseInt(Buffer.from(packet.usr4).toString(), 10) + const ts = Math.min(Packet.ttl, timestamp) + + return packet.version === VERSION && ts > Date.now() - Packet.ttl } /** @@ -1285,7 +1291,8 @@ export class Peer { // const nextLevel = await this.cache.summarize(local.prefix + i.toString(16), this.cachePredicate) const data = await Packet.encode(new PacketSync({ - message: Cache.encodeSummary(nextLevel) + message: Cache.encodeSummary(nextLevel), + usr4: Buffer.from(String(Date.now())) })) this.send(data, port, address) }