fix: object discovery was not using identities for listening addresses, per the latest changes

sbazerque · sbazerque · commit 52fb9f983d37 · 2022-06-02T21:43:34.000-03:00
diff --git a/src/data/collections/Types.ts b/src/data/collections/Types.ts
@@ -1,6 +1,8 @@
 import { HashedSet } from 'data/model';
 import { HashedObject } from '../model/immutable/HashedObject';
 
+// FIXME: the types thing should be a HashedSet, not a friggin array. What was I thinking?
+
 abstract class Types {
     
     static HashedObject = 'HashedObject';
diff --git a/src/mesh/agents/discovery/ObjectBroadcastAgent.ts b/src/mesh/agents/discovery/ObjectBroadcastAgent.ts
@@ -19,7 +19,7 @@ type ObjectBroadcastReply = {
 
 class ObjectBroadcastAgent implements Agent {
 
-    static log = new Logger(ObjectBroadcastAgent.name, LogLevel.INFO);
+    static log = new Logger(ObjectBroadcastAgent.name, LogLevel.DEBUG);
 
     static defaultBroadcastedSuffixBits = 36;
 
diff --git a/src/mesh/agents/discovery/ObjectDiscoveryAgent.ts b/src/mesh/agents/discovery/ObjectDiscoveryAgent.ts
@@ -20,7 +20,7 @@ type ObjectDiscoveryReplyParams = {maxAge?: number, linkupServers?: string[], lo
 
 class ObjectDiscoveryAgent implements Agent {
 
-    static log = new Logger(ObjectDiscoveryAgent.name, LogLevel.INFO);
+    static log = new Logger(ObjectDiscoveryAgent.name, LogLevel.DEBUG);
 
     static agentIdForHexHashSuffix(suffix: string) {
         return 'object-discovery-for-' + suffix;
@@ -70,7 +70,7 @@ class ObjectDiscoveryAgent implements Agent {
         this.pod = pod;
     }
 
-    query(linkupServers: string[], localEndpoint: Endpoint, count=1) {
+    query(linkupServers: string[], localAddress: LinkupAddress, count=1) {
 
         if (this.pod === undefined) {
             throw new Error('This ObjectDiscoveryAgent has not been registered to a mesh so it cannot accept queries yet.');
@@ -87,9 +87,12 @@ class ObjectDiscoveryAgent implements Agent {
             
         }
 
+        const localEndpoint = localAddress.url();
+        const localIdentity = localAddress.identity;
+
         if (!this.localEndpoints.has(localEndpoint)) {
             ObjectDiscoveryAgent.log.trace('listening on ' + localEndpoint);
-            this.getNetworkAgent().listenForLinkupMessages(localEndpoint);
+            this.getNetworkAgent().listenForLinkupMessages(localEndpoint, localIdentity);
             this.localEndpoints.add(localEndpoint);
         }
 
diff --git a/src/mesh/agents/network/NetworkAgent.ts b/src/mesh/agents/network/NetworkAgent.ts
@@ -512,8 +512,8 @@ class NetworkAgent implements Agent {
         this.linkupManager.listenForMessagesNewCall(address, this.newConnectionRequestCallback);
     }
 
-    listenForLinkupMessages(endpoint: Endpoint) {
-        let address = LinkupAddress.fromURL(endpoint);
+    listenForLinkupMessages(endpoint: Endpoint, identity?: Identity) {
+        let address = LinkupAddress.fromURL(endpoint, identity);
         this.linkupMessageListening.add(endpoint);
         this.linkupManager.listenForRawMessages(address, this.linkupMessageCallback);
     }
diff --git a/src/mesh/agents/peer/sources/ObjectDiscoveryPeerSource.ts b/src/mesh/agents/peer/sources/ObjectDiscoveryPeerSource.ts
@@ -2,6 +2,7 @@ import { Hash, HashedObject } from 'data/model';
 import { ObjectDiscoveryReply } from 'mesh/agents/discovery/ObjectDiscoveryAgent';
 import { Endpoint } from 'mesh/agents/network/NetworkAgent';
 import { Mesh } from 'mesh/service/Mesh';
+import { LinkupAddress } from 'net/linkup';
 import { AsyncStream } from 'util/streams';
 import { PeerInfo } from '../PeerGroupAgent';
 import { PeerSource } from '../PeerSource';
@@ -14,19 +15,19 @@ class ObjectDiscoveryPeerSource implements PeerSource {
     parseEndpoint: (ep: Endpoint) => Promise<PeerInfo | undefined>;
 
     linkupServers: string[];
-    replyEndpoint: Endpoint;
+    replyAddress: LinkupAddress;
     timeoutMillis: number;
     
     hash: Hash;
     replyStream?: AsyncStream<ObjectDiscoveryReply>;
 
-    constructor(mesh: Mesh, object: HashedObject, linkupServers: string[], replyEndpoint: Endpoint, parseEndpoint: (ep: Endpoint) => Promise<PeerInfo | undefined>, timeout=3) {
+    constructor(mesh: Mesh, object: HashedObject, linkupServers: string[], replyAddress: LinkupAddress, parseEndpoint: (ep: Endpoint) => Promise<PeerInfo | undefined>, timeout=3) {
         this.mesh = mesh;
         this.object = object;
         this.parseEndpoint = parseEndpoint;
 
         this.linkupServers = linkupServers;
-        this.replyEndpoint = replyEndpoint;
+        this.replyAddress = replyAddress;
         this.timeoutMillis = timeout * 1000;
 
         this.hash = object.hash();
@@ -94,11 +95,11 @@ class ObjectDiscoveryPeerSource implements PeerSource {
     }
 
     private tryObjectDiscovery(count: number) : AsyncStream<ObjectDiscoveryReply> {
-        return this.mesh.findObjectByHash(this.hash, this.linkupServers, this.replyEndpoint, count);
+        return this.mesh.findObjectByHash(this.hash, this.linkupServers, this.replyAddress, count);
     }
 
     private retryObjectDiscovery(count: number) {
-        this.mesh.findObjectByHashRetry(this.hash, this.linkupServers, this.replyEndpoint, count);
+        this.mesh.findObjectByHashRetry(this.hash, this.linkupServers, this.replyAddress, count);
     }
 }
 
diff --git a/src/mesh/service/Mesh.ts b/src/mesh/service/Mesh.ts
@@ -10,7 +10,7 @@ import { ObjectBroadcastAgent, ObjectDiscoveryAgent, ObjectDiscoveryReply, Objec
 
 import { AgentPod } from './AgentPod';
 import { AsyncStream } from 'util/streams';
-import { LinkupManager } from 'net/linkup';
+import { LinkupAddress, LinkupManager } from 'net/linkup';
 import { RNGImpl } from 'crypto/random';
 import { Logger, LogLevel } from 'util/logging';
 
@@ -347,39 +347,39 @@ class Mesh {
     //        are present in the received suffix!). To fix it, either change ObjectDiscoveryAgent to use all the
     //        received bits, or pass the number of bits explicitly when calling the constructor.
 
-    findObjectByHash(hash: Hash, linkupServers: string[], replyEndpoint: Endpoint, count=1, maxAge=30, strictEndpoints=false, includeErrors=false) : AsyncStream<ObjectDiscoveryReply> {
+    findObjectByHash(hash: Hash, linkupServers: string[], replyAddress: LinkupAddress, count=1, maxAge=30, strictEndpoints=false, includeErrors=false) : AsyncStream<ObjectDiscoveryReply> {
         const suffix = Hashing.toHex(hash);
-        return this.findObjectByHashSuffix(suffix, linkupServers, replyEndpoint, count, maxAge, strictEndpoints, includeErrors);
+        return this.findObjectByHashSuffix(suffix, linkupServers, replyAddress, count, maxAge, strictEndpoints, includeErrors);
     }
 
-    findObjectByHashSuffix(hashSuffix: string, linkupServers: string[], replyEndpoint: Endpoint, count=1, maxAge=30, strictEndpoints=false, includeErrors=false) : AsyncStream<ObjectDiscoveryReply> {
+    findObjectByHashSuffix(hashSuffix: string, linkupServers: string[], replyAddress: LinkupAddress, count=1, maxAge=30, strictEndpoints=false, includeErrors=false) : AsyncStream<ObjectDiscoveryReply> {
         
         const discoveryAgent = this.getDiscoveryAgentFor(hashSuffix);
 
-        discoveryAgent.query(linkupServers, replyEndpoint, count);
+        discoveryAgent.query(linkupServers, replyAddress, count);
 
         let params: ObjectDiscoveryReplyParams = {};
 
         params.maxAge = maxAge;
 
         if (strictEndpoints) {
             params.linkupServers = linkupServers;
-            params.localEndpoints = [replyEndpoint];
+            params.localEndpoints = [replyAddress.url()];
         }
 
         params.includeErrors = includeErrors;
 
         return discoveryAgent.getReplyStream(params);
     }
 
-    findObjectByHashRetry(hash: Hash, linkupServers: string[], replyEndpoint: Endpoint, count=1): void {
+    findObjectByHashRetry(hash: Hash, linkupServers: string[], replyAddress: LinkupAddress, count=1): void {
         const suffix = Hashing.toHex(hash);
-        this.findObjectByHashSuffixRetry(suffix, linkupServers, replyEndpoint, count);
+        this.findObjectByHashSuffixRetry(suffix, linkupServers, replyAddress, count);
     }
 
-    findObjectByHashSuffixRetry(hashSuffix: string, linkupServers: string[], replyEndpoint: Endpoint, count=1): void {
+    findObjectByHashSuffixRetry(hashSuffix: string, linkupServers: string[], replyAddress: LinkupAddress, count=1): void {
         const discoveryAgent = this.getDiscoveryAgentFor(hashSuffix);
-        discoveryAgent.query(linkupServers, replyEndpoint, count);
+        discoveryAgent.query(linkupServers, replyAddress, count);
     }
 
     getSyncAgentFor(peerGroupId: PeerGroupId, mutHash: Hash): StateSyncAgent|undefined {
diff --git a/src/mesh/service/PeerNode.ts b/src/mesh/service/PeerNode.ts
@@ -1,6 +1,7 @@
 
 import { Hash, HashedObject } from 'data/model';
 import { ObjectDiscoveryPeerSource, PeerInfo } from 'mesh/agents/peer';
+import { LinkupAddress } from 'net/linkup';
 import { Resources } from 'spaces/spaces';
 import { MultiMap } from 'util/multimap';
 import { PeerGroupInfo, SyncMode, UsageToken } from './Mesh';
@@ -109,7 +110,7 @@ class PeerNode {
 
     private async discoveryPeerGroupInfo(obj: HashedObject) : Promise<PeerGroupInfo> {
         let localPeer = this.resources.getPeersForDiscovery()[0];
-        let peerSource = new ObjectDiscoveryPeerSource(this.resources.mesh, obj, this.resources.config.linkupServers, localPeer.endpoint, this.resources.getEndointParserForDiscovery());
+        let peerSource = new ObjectDiscoveryPeerSource(this.resources.mesh, obj, this.resources.config.linkupServers, LinkupAddress.fromURL(localPeer.endpoint, localPeer.identity), this.resources.getEndointParserForDiscovery());
 
         return {
             id: PeerNode.discoveryPeerGroupInfoId(obj),
diff --git a/src/mesh/service/remoting/MeshHost.ts b/src/mesh/service/remoting/MeshHost.ts
@@ -8,6 +8,7 @@ import { ObjectDiscoveryReply } from 'mesh/agents/discovery';
 import { RNGImpl } from 'crypto/random';
 import { Identity, RSAKeyPair } from 'data/identity';
 import { Store } from 'storage/store';
+import { LinkupAddress } from 'net/linkup';
 
 type MeshCommand = JoinPeerGroup | LeavePeerGroup |
                    SyncObjectsWithPeerGroup | StopSyncObjectsWithPeerGroup |
@@ -21,8 +22,8 @@ type JoinPeerGroup = {
     peerGroupId: string,
     localPeerEndpoint: Endpoint;
     localPeerIdentityHash: Hash;
-    localPeerIdentity?: LiteralContext | undefined,
-    localPeerIdentityKeyPair?: LiteralContext | undefined,
+    localPeerIdentity?: LiteralContext,
+    localPeerIdentityKeyPair?: LiteralContext,
     //localPeer: PeerInfo,
     config?: PeerGroupAgentConfig;
     usageToken?: UsageToken
@@ -67,6 +68,7 @@ type FindObjectByHash = {
     hash: Hash,
     linkupServers: Array<string>,
     replyEndpoint: Endpoint,
+    replyIdentity?: LiteralContext,
     count?: number,
     maxAge?: number,
     strictEndpoints?: boolean,
@@ -80,6 +82,7 @@ type FindObjectByHashSuffix = {
     hashSuffix: string,
     linkupServers: Array<string>,
     replyEndpoint: Endpoint,
+    replyIdentity?: LiteralContext,
     count?: number,
     maxAge?: number,
     includeErrors?: boolean,
@@ -293,18 +296,21 @@ class MeshHost {
                    command.type === 'find-object-by-hash-suffix') {
             const find = command as FindObjectByHash | FindObjectByHashSuffix;
 
+            const id = find.replyIdentity === undefined? undefined : HashedObject.fromLiteralContext(find.replyIdentity) as Identity;
+            const replyAddress = LinkupAddress.fromURL(find.replyEndpoint, id);
+
             if (!find.retry) {
 
                 const streamId = command.streamId;
                 let replyStream: AsyncStream<ObjectDiscoveryReply>;
 
                 if (command.type === 'find-object-by-hash') {
                     replyStream = this.mesh.findObjectByHash(
-                        (find as FindObjectByHash).hash, find.linkupServers, find.replyEndpoint, find.count, find.maxAge, find.strictEndpoints, find.includeErrors
+                        (find as FindObjectByHash).hash, find.linkupServers, replyAddress, find.count, find.maxAge, find.strictEndpoints, find.includeErrors
                     );
                 } else {
                     replyStream = this.mesh.findObjectByHashSuffix(
-                        (find as FindObjectByHashSuffix).hashSuffix, find.linkupServers, find.replyEndpoint, find.count, find.maxAge, find.strictEndpoints, find.includeErrors
+                        (find as FindObjectByHashSuffix).hashSuffix, find.linkupServers, replyAddress, find.count, find.maxAge, find.strictEndpoints, find.includeErrors
                     );
                 }
                 
@@ -356,11 +362,11 @@ class MeshHost {
             } else {
                 if (command.type === 'find-object-by-hash') {
                     this.mesh.findObjectByHashRetry(
-                        (find as FindObjectByHash).hash, find.linkupServers, find.replyEndpoint, find.count
+                        (find as FindObjectByHash).hash, find.linkupServers, replyAddress, find.count
                     );
                 } else {
                     this.mesh.findObjectByHashSuffixRetry(
-                        (find as FindObjectByHashSuffix).hashSuffix, find.linkupServers, find.replyEndpoint, find.count
+                        (find as FindObjectByHashSuffix).hashSuffix, find.linkupServers, replyAddress, find.count
                     );
                 }
             }
diff --git a/src/mesh/service/remoting/MeshProxy.ts b/src/mesh/service/remoting/MeshProxy.ts
@@ -14,7 +14,7 @@ import { Context, Hash, HashedObject } from 'data/model';
 import { AsyncStream, BufferedAsyncStream, BufferingAsyncStreamSource } from 'util/streams';
 import { ObjectDiscoveryReply } from 'mesh/agents/discovery';
 import { Endpoint } from 'mesh/agents/network';
-import { LinkupManager, LinkupManagerCommand, LinkupManagerProxy } from 'net/linkup';
+import { LinkupAddress, LinkupManager, LinkupManagerCommand, LinkupManagerProxy } from 'net/linkup';
 import { WebRTCConnectionEvent, WebRTCConnectionsHost } from 'net/transport';
 
 class MeshProxy {
@@ -267,7 +267,7 @@ class MeshProxy {
         this.commandForwardingFn(cmd);
     }
 
-    findObjectByHash(hash: Hash, linkupServers: string[], replyEndpoint: Endpoint, count=1, maxAge=30, strictEndpoints=false) : AsyncStream<ObjectDiscoveryReply> {
+    findObjectByHash(hash: Hash, linkupServers: string[], replyAddress: LinkupAddress, count=1, maxAge=30, strictEndpoints=false) : AsyncStream<ObjectDiscoveryReply> {
         const streamId = new RNGImpl().randomHexString(64);
 
         const src = new BufferingAsyncStreamSource<ObjectDiscoveryReply>()    
@@ -278,7 +278,8 @@ class MeshProxy {
             type: 'find-object-by-hash',
             hash: hash,
             linkupServers: linkupServers,
-            replyEndpoint: replyEndpoint,
+            replyEndpoint: replyAddress.url(),
+            replyIdentity: replyAddress.identity === undefined? undefined: replyAddress.identity.toLiteralContext(),
             count: count,
             maxAge: maxAge,
             strictEndpoints: strictEndpoints,
@@ -291,7 +292,7 @@ class MeshProxy {
         return new BufferedAsyncStream<ObjectDiscoveryReply>(src);
     }
 
-    findObjectByHashSuffix(hashSuffix: string, linkupServers: string[], replyEndpoint: Endpoint, count=1, maxAge=30, strictEndpoints=false) : AsyncStream<ObjectDiscoveryReply> {
+    findObjectByHashSuffix(hashSuffix: string, linkupServers: string[], replyAddress: LinkupAddress, count=1, maxAge=30, strictEndpoints=false) : AsyncStream<ObjectDiscoveryReply> {
         const streamId = new RNGImpl().randomHexString(64);
 
         const src = new BufferingAsyncStreamSource<ObjectDiscoveryReply>()    
@@ -302,7 +303,8 @@ class MeshProxy {
             type: 'find-object-by-hash-suffix',
             hashSuffix: hashSuffix,
             linkupServers: linkupServers,
-            replyEndpoint: replyEndpoint,
+            replyEndpoint: replyAddress.url(),
+            replyIdentity: replyAddress.identity === undefined? undefined: replyAddress.identity.toLiteralContext(),
             count: count,
             maxAge: maxAge,
             strictEndpoints: strictEndpoints,
@@ -321,25 +323,27 @@ class MeshProxy {
         this.commandForwardingFn(cmd);
     }
 
-    findObjectByHashRetry(hash: Hash, linkupServers: string[], replyEndpoint: Endpoint, count=1): void {
+    findObjectByHashRetry(hash: Hash, linkupServers: string[], replyAddress: LinkupAddress, count=1): void {
         const cmd: FindObjectByHash = {
             type: 'find-object-by-hash',
             hash: hash,
             linkupServers: linkupServers,
-            replyEndpoint: replyEndpoint,
+            replyEndpoint: replyAddress.url(),
+            replyIdentity: replyAddress.identity === undefined? undefined: replyAddress.identity.toLiteralContext(),
             count: count,
             retry: true,
         }
 
         this.commandForwardingFn(cmd);
     }
 
-    findObjectByHashSuffixRetry(hashSuffix: string, linkupServers: string[], replyEndpoint: Endpoint, count=1): void {    
+    findObjectByHashSuffixRetry(hashSuffix: string, linkupServers: string[], replyAddress: LinkupAddress, count=1): void {    
         const cmd: FindObjectByHashSuffix = {
             type: 'find-object-by-hash-suffix',
             hashSuffix: hashSuffix,
             linkupServers: linkupServers,
-            replyEndpoint: replyEndpoint,
+            replyEndpoint: replyAddress.url(),
+            replyIdentity: replyAddress.identity === undefined? undefined: replyAddress.identity.toLiteralContext(),
             count: count,
             retry: true,
         }
diff --git a/src/net/transport/WebRTCConnection.ts b/src/net/transport/WebRTCConnection.ts
diff --git a/src/spaces/Space.ts b/src/spaces/Space.ts
diff --git a/test/mesh/mock/TestPeerGroupPods.ts b/test/mesh/mock/TestPeerGroupPods.ts

Original file line number	Diff line number	Diff line change
`@@ -512,8 +512,8 @@ class NetworkAgent implements Agent {`
`512`	`512`	`this.linkupManager.listenForMessagesNewCall(address, this.newConnectionRequestCallback);`
`513`	`513`	`}`
`514`	`514`
`515`		`- listenForLinkupMessages(endpoint: Endpoint) {`
`516`		`- let address = LinkupAddress.fromURL(endpoint);`
	`515`	`+ listenForLinkupMessages(endpoint: Endpoint, identity?: Identity) {`
	`516`	`+ let address = LinkupAddress.fromURL(endpoint, identity);`
`517`	`517`	`this.linkupMessageListening.add(endpoint);`
`518`	`518`	`this.linkupManager.listenForRawMessages(address, this.linkupMessageCallback);`
`519`	`519`	`}`