Subcluster emissions seem to always be published and never streamed

[jcmoore]

What OS are you using (uname -a, or Windows version)?

Darwin Kernel Version 22.4.0: Mon Mar 6 21:01:02 PST 2023; root:xnu-8796.101.5~3/RELEASE_ARM64_T8112 arm64
Mac OS Ventura 13.3.1 (22E261)

What version Socket Runtime are you using?

Reading an unreleased commit after v0.5.4 -- https://github.com/socketsupply/socket/tree/0c235fb4fa97f552c759551c9daab815a96eeaa9

What programming language are you using (C/C++/Go/Rust)?

N/A (but usually Javascript/Typescript)

What did you expect to see and what you saw instead?

When investigating how/if socket apps can disconnect from the p2p-network (looks like the answer might be (await network({})).peer.close() as opposed to (await network({})).disconnect()) I spotted what appears to be a network optimization that is never executed when invoking subcluster.emit().

socket/api/stream-relay/sugar.js

Line 256 in 0c235fb

if (sub.peers.values().length) {

On the surface, the "fix" might be as simple as checking for sub.peers.size rather than sub.peers.values().length (because Map iterators have no such property by default).

Looking deeper, that change would exercise code which was likely never executing prior, and may have the following side effects compared to the local Peer.publish() fallback (which was presumably always safe):

1. no "#packet" events are emitted to the socket/bus as otherwise would occur during a Peer.publish()

   socket/api/stream-relay/index.js

   Lines 974 to 984 in 0c235fb

   	for (const packet of packets) {
   	const p = Packet.from(packet)
   	this.cache.insert(packet.packetId.toString('hex'), p)
   	if (this.onPacket) this.onPacket(p, this.port, this.address, true)
   	this.unpublished[packet.packetId.toString('hex')] = Date.now()
   	if (globalThis.navigator && !globalThis.navigator.onLine) continue
   	this.mcast(packet)
   	}

   socket/api/stream-relay/sugar.js

   Lines 264 to 272 in 0c235fb

   	for (const packet of packets) {
   	const p = Packet.from(packet)
   	_peer.cache.insert(packet.packetId.toString('hex'), p)
   	_peer.unpublished[packet.packetId.toString('hex')] = Date.now()
   	if (globalThis.navigator && !globalThis.navigator.onLine) continue
   	_peer.mcast(packet)
   	}

   socket/api/stream-relay/sugar.js

   Lines 60 to 63 in 0c235fb

   	_peer.onPacket = (packet, ...args) => {
   	if (!packet.clusterId.equals(clusterId)) return
   	bus._emit('#packet', packet, ...args)
   	}

2. the packets ultimately handed to _peer.mcast() will have usr4 and usr3 set to the peerId of the local peer and the first remote peer respectively -- and will be of type PacketStream rather than PacketPublish

   socket/api/stream-relay/sugar.js

   Lines 260 to 261 in 0c235fb

   	const r = await p._peer.write(sub.sharedKey, args)
   	if (packets.length === 0) packets = r

   socket/api/stream-relay/index.js

   Lines 200 to 204 in 0c235fb

   	args.usr3 = Buffer.from(this.peerId, 'hex')
   	args.usr4 = Buffer.from(this.localPeer.peerId, 'hex')
   	args.message = this.localPeer.encryption.seal(args.message, keys)
   	const packets = await this.localPeer._message2packets(PacketStream, args.message, args)

   socket/api/stream-relay/index.js

   Line 972 in 0c235fb

   const packets = await this._message2packets(PacketPublish, message, args)

3. it's not entirely obvious to me that RemotePeer.write() was getting exercised anywhere, and though I doubt it's the case, it might not work as expected -- one place it's potentially getting exercised is below, which seems to use a possibly stale reference to the closure's peer.write instead of ee._peer.write

   socket/api/stream-relay/sugar.js

   Lines 317 to 325 in 0c235fb

   	ee.emit = async (eventName, value, opts = {}) => {
   	if (!ee._peer.write) return
   	opts.clusterId = opts.clusterId || clusterId
   	opts.subclusterId = opts.subclusterId || sub.subclusterId
   	const args = await pack(eventName, value, opts)
   	return peer.write(sub.sharedKey, args)
   	}

If any of these seem like genuine issues, I'd be happy to submit a PR, but I'd want some guidance on how to appropriately test proposed changes. Do you have a solution for mocking a network of peers?