observability for peer group & sync state - WIP

sbazerque · sbazerque · commit 9697571284fe · 2022-10-12T11:54:05.000-03:00
diff --git a/src/mesh/agents/peer/PeerGroupAgent.ts b/src/mesh/agents/peer/PeerGroupAgent.ts
@@ -17,6 +17,7 @@ import { Identity } from 'data/identity';
 import { Logger, LogLevel } from 'util/logging';
 import { Lock } from 'util/concurrency';
 import { LRUCache } from 'util/caching';
+import { PeerGroupState } from './PeerGroupState';
 
 type PeerInfo = { endpoint: Endpoint, identityHash: Hash, identity?: Identity };
 
@@ -549,6 +550,24 @@ class PeerGroupAgent implements Agent {
 
     }
 
+    getState(): PeerGroupState {
+
+        const remote = new Map<Endpoint, PeerInfo>();
+
+        for (const ep of this.connectionsPerEndpoint.keys()) {
+            const connId = this.findWorkingConnectionId(ep);
+            const conn   = connId === undefined? undefined : this.connections.get(connId);
+            if (conn !== undefined) {
+                remote.set(ep, conn.peer);
+            }
+        }
+
+        return {
+            local: this.localPeer,
+            remote: remote
+        };
+    }
+
     shutdown() {
         if (this.tickTimerRef !== undefined) {
             clearInterval(this.tickTimerRef);
diff --git a/src/mesh/agents/peer/PeerGroupState.ts b/src/mesh/agents/peer/PeerGroupState.ts
@@ -0,0 +1,10 @@
+import { Endpoint } from '../network';
+import { PeerInfo } from './PeerGroupAgent';
+
+
+type PeerGroupState = {
+    local: PeerInfo,
+    remote: Map<Endpoint, PeerInfo>
+};
+
+export type { PeerGroupState  };
diff --git a/src/mesh/agents/state/SyncState.ts b/src/mesh/agents/state/SyncState.ts
@@ -0,0 +1,13 @@
+import { Hash } from 'data/model';
+
+type SyncState = {
+    objectHash: Hash,
+
+    fetchedEverything: boolean, // all remote known ops have been fetched
+    sentEverything: boolean,    // all local ops have been sent to all current peers
+    synchronzing: boolean,      // ops are being exchanged at the moment
+
+    opsToReceive: number        // how many ops we know we need to fetch
+};
+
+export type { SyncState };
diff --git a/src/mesh/service/Mesh.ts b/src/mesh/service/Mesh.ts
@@ -16,6 +16,7 @@ import { Logger, LogLevel } from 'util/logging';
 import { Identity } from 'data/identity';
 import { ObjectSpawnAgent, SpawnCallback } from 'mesh/agents/spawn/ObjectSpawnAgent';
 import { ObjectInvokeAgent } from 'mesh/agents/spawn/ObjectInvokeAgent';
+import { PeerGroupState } from 'mesh/agents/peer/PeerGroupState';
 
 
 
@@ -207,6 +208,19 @@ class Mesh {
 
     }
 
+    async getPeerGroupState(peerGroupId: string): Promise<PeerGroupState|undefined> {
+
+        const agentId = PeerGroupAgent.agentIdForPeerGroup(peerGroupId);
+
+        const agent = this.pod.getAgent(agentId) as PeerGroupAgent;
+
+        if (agent === undefined) {
+            return undefined;
+        } else {
+            return agent.getState();
+        }
+    }
+
     /*getConnectedPeers(peerGroupId: string): Array<PeerInfo> {
         let agent = this.pod.getAgent(PeerGroupAgent.agentIdForPeerGroup(peerGroupId)) as PeerGroupAgent;
 
diff --git a/src/mesh/service/remoting/MeshHost.ts b/src/mesh/service/remoting/MeshHost.ts
@@ -1,15 +1,20 @@
 import { PeerGroupAgentConfig, PeerInfo, PeerSource } from '../../agents/peer';
 import { Mesh, SyncMode, UsageToken } from '../../service/Mesh';
+import { SpawnCallback } from '../../agents/spawn';
+import { PeerGroupState } from '../../agents/peer/PeerGroupState';
+import { ObjectDiscoveryReply } from '../../agents/discovery';
+import { Endpoint } from '../../agents/network';
+
+import { Identity, RSAKeyPair } from 'data/identity';
 import { Context, HashedObject, LiteralContext } from 'data/model';
 import { Hash } from 'data/model';
-import { Endpoint } from 'mesh/agents/network';
+
 import { AsyncStream } from 'util/streams';
-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';
-import { SpawnCallback } from 'mesh/agents/spawn';
 
 /* Run a mesh remotely, and access it through a MeshProxy */
 
@@ -25,7 +30,7 @@ import { SpawnCallback } from 'mesh/agents/spawn';
  * 
  * Ah, the things we do for you, Hyper Hyper Space. */
 
-type MeshCommand = JoinPeerGroup | LeavePeerGroup |
+type MeshCommand = JoinPeerGroup | LeavePeerGroup | ForwardPeerGroupState |
                    SyncObjectsWithPeerGroup | StopSyncObjectsWithPeerGroup |
                    StartObjectBroadcast | StopObjectBroadcast |
                    FindObjectByHash | FindObjectByHashSuffix |  Shutdown |
@@ -50,6 +55,12 @@ type LeavePeerGroup = {
     usageToken: UsageToken
 }
 
+type ForwardPeerGroupState = {
+    type: 'forward-peer-group-state',
+    requestId: string,
+    peerGroupId: string
+}
+
 type SyncObjectsWithPeerGroup = {
     type: 'sync-objects-with-peer-group',
     peerGroupId: string,
@@ -147,7 +158,7 @@ type ForwardGetPeerForEndpointReply = {
 
 type PeerInfoContext = { endpoint: Endpoint, identityHash: Hash, identity?: LiteralContext };
 
-type CommandStreamedReply = LiteralObjectDiscoveryReply | DiscoveryEndReply | ObjectSpawnCallback;
+type CommandStreamedReply = LiteralObjectDiscoveryReply | DiscoveryEndReply | ObjectSpawnCallback | PeerGroupStateReply;
 
 type LiteralObjectDiscoveryReply = {
     type: 'object-discovery-reply'
@@ -173,6 +184,15 @@ type ObjectSpawnCallback = {
     senderEndpoint: string
 }
 
+type LiteralPeerInfo =  { endpoint: Endpoint, identityHash: Hash, identity?: LiteralContext };
+
+type PeerGroupStateReply = {
+    type: 'peer-group-state-reply',
+    requestId: string,
+    local?: LiteralPeerInfo,
+    remote?: Array<LiteralPeerInfo>
+}
+
 
 type PeerSourceRequest = GetPeersRequest | GetPeerForEndpointRequest;
 
@@ -199,6 +219,7 @@ class MeshHost {
         return (type === 'join-peer-group' ||
             type === 'check-peer-group-usage' ||
             type === 'leave-peer-group' ||
+            type === 'forward-peer-group-state' ||
             type === 'sync-objects-with-peer-group' ||
             type === 'stop-sync-objects-with-peer-group' ||
             type === 'start-object-broadcast' ||
@@ -220,7 +241,8 @@ class MeshHost {
 
         return (type === 'object-discovery-reply' ||
                 type === 'object-discovery-end'   || 
-                type === 'object-spawn-callback');
+                type === 'object-spawn-callback'  ||
+                type === 'peer-group-state-reply');
     }
 
     static isPeerSourceRequest(msg: any): boolean {
@@ -275,6 +297,42 @@ class MeshHost {
         } else if (command.type === 'leave-peer-group') {
             const leave = command as LeavePeerGroup;
             this.mesh.leavePeerGroup(leave.usageToken);
+
+        } else if (command.type === 'forward-peer-group-state') {
+
+            this.mesh.getPeerGroupState(command.peerGroupId).then((state: PeerGroupState|undefined) => {
+
+                const reply: PeerGroupStateReply = {
+                    type: 'peer-group-state-reply',
+                    requestId: command.requestId,
+                }
+
+                if (state !== undefined) {
+                    const local: LiteralPeerInfo = {endpoint: state.local.endpoint, identityHash: state.local.identityHash};
+
+                    if (state.local.identity !== undefined) {
+                        local.identity = state.local.identity.toLiteralContext();
+                    }
+
+                    reply.local = local;
+
+                    reply.remote = [];
+
+                    for (const peerInfo of state.remote.values()) {
+                        const lit: LiteralPeerInfo = {endpoint: peerInfo.endpoint, identityHash: peerInfo.identityHash};
+
+                        if (peerInfo.identity !== undefined) {
+                            lit.identity = peerInfo.identity.toLiteralContext();
+                        }
+
+                        reply.remote.push(lit);
+                    }
+                }
+
+                this.streamedReplyCb(reply);
+
+            })
+
         } else if (command.type === 'sync-objects-with-peer-group') {
             const syncObjs = command as SyncObjectsWithPeerGroup;
 
@@ -570,11 +628,11 @@ class PeerSourceProxy implements PeerSource {
 }
 
 export { MeshHost, MeshCommand,
-         JoinPeerGroup, LeavePeerGroup,
+         JoinPeerGroup, LeavePeerGroup, ForwardPeerGroupState, 
          SyncObjectsWithPeerGroup, StopSyncObjectsWithPeerGroup,
          StartObjectBroadcast, StopObjectBroadcast,
          FindObjectByHash, FindObjectByHashSuffix, AddObjectSpawnCallback, SendObjectSpawnRequest,
          Shutdown,
-         CommandStreamedReply, LiteralObjectDiscoveryReply, DiscoveryEndReply, ObjectSpawnCallback,
+         CommandStreamedReply, LiteralObjectDiscoveryReply, DiscoveryEndReply, ObjectSpawnCallback, PeerGroupStateReply, LiteralPeerInfo, 
          ForwardGetPeersReply, ForwardGetPeerForEndpointReply,
          PeerSourceRequest, GetPeersRequest, GetPeerForEndpointRequest, PeerInfoContext };
diff --git a/src/mesh/service/remoting/MeshProxy.ts b/src/mesh/service/remoting/MeshProxy.ts
@@ -10,7 +10,8 @@ import { MeshCommand,
     Shutdown,
     PeerInfoContext,
     AddObjectSpawnCallback,
-    SendObjectSpawnRequest} from './MeshHost';
+    SendObjectSpawnRequest,
+    ForwardPeerGroupState} from './MeshHost';
 
 import { RNGImpl } from 'crypto/random';
 import { Context, Hash, HashedObject } from 'data/model';
@@ -21,9 +22,12 @@ import { LinkupAddress, LinkupManager, LinkupManagerCommand, LinkupManagerProxy
 import { WebRTCConnectionEvent, WebRTCConnectionsHost } from 'net/transport';
 import { Identity } from 'data/identity';
 import { ObjectSpawnAgent, SpawnCallback } from 'mesh/agents/spawn';
+import { PeerGroupState } from 'mesh/agents/peer/PeerGroupState';
 
 /* Access a mesh remotely, see the MeshHost class. */
 
+type RequestId = string;
+
 class MeshProxy {
 
     commandForwardingFn: (cmd: MeshCommand) => void;
@@ -37,6 +41,8 @@ class MeshProxy {
     peerSources: Map<string, PeerSource>;
     peerSourceRequestIngestFn: (req: PeerSourceRequest) => void;
 
+    pendingPeerGroupStates: Map<RequestId, {resolve: (result: PeerGroupState|undefined) => void, reject: (reason: any) => void, timeout: any}>;
+
     
 
     constructor(meshCommandFwdFn: (cmd: MeshCommand) => void, linkupCommandFwdFn?: (cmd: LinkupManagerCommand) => void, webRTCConnEventIngestFn?: (ev: WebRTCConnectionEvent) => void) {
@@ -78,6 +84,47 @@ class MeshProxy {
                     const sender = HashedObject.fromLiteralContext(reply.sender) as Identity;
                     cb(object, sender, reply.senderEndpoint);
                 }
+            } else if (reply.type === 'peer-group-state-reply') {
+                const cb = this.pendingPeerGroupStates.get(reply.requestId)?.resolve;
+                const to = this.pendingPeerGroupStates.get(reply.requestId)?.timeout;
+
+                this.pendingPeerGroupStates.delete(reply.requestId);
+
+                if (to !== undefined) {
+                    window.clearTimeout(to);
+                }
+
+                if (cb !== undefined) {
+
+                    if (reply.local !== undefined && reply.remote !== undefined) {
+                        const remote = new Map<Endpoint, PeerInfo>();
+
+                        const state: PeerGroupState = {
+                            local: {endpoint: reply.local.endpoint, identityHash: reply.local.identityHash},
+                            remote: remote
+                        }
+    
+                        if (reply.local.identity !== undefined) {
+                            state.local.identity = HashedObject.fromLiteralContext(reply.local.identity) as Identity;
+                        }
+    
+                        for (const litPeerInfo of reply.remote) {
+
+                            const peerInfo: PeerInfo = {endpoint: litPeerInfo.endpoint, identityHash: litPeerInfo.identityHash};
+
+                            if (litPeerInfo.identity !== undefined) {
+                                peerInfo.identity = HashedObject.fromLiteralContext(litPeerInfo.identity) as Identity;
+                            }
+
+                            remote.set(peerInfo.endpoint, peerInfo);
+                        }
+
+                        cb(state);    
+                    } else {
+                        cb(undefined);
+                    }
+                    
+                }
             }
         }
 
@@ -143,6 +190,8 @@ class MeshProxy {
                 }
             }
         };
+
+        this.pendingPeerGroupStates = new Map();
     }
 
     getCommandStreamedReplyIngestFn() {
@@ -182,6 +231,35 @@ class MeshProxy {
         this.commandForwardingFn(cmd);
     }
 
+    async getPeerGroupState(peerGroupId: string, timeout=10000): Promise<PeerGroupState|undefined> {
+
+        const p = new Promise<PeerGroupState|undefined>((resolve: (result: PeerGroupState|undefined) => void, reject: (reason: any) => void) => {
+            const requestId = new RNGImpl().randomHexString(128);
+
+            const cmd: ForwardPeerGroupState = {
+                type: 'forward-peer-group-state',
+                requestId: requestId,
+                peerGroupId: peerGroupId
+            };
+    
+            
+
+            const to = window.setTimeout(() => {
+                if (this.pendingPeerGroupStates.has(requestId)) {
+                    this.pendingPeerGroupStates.delete(requestId)
+                    reject('timeout');
+                }
+            }, timeout);
+
+            this.pendingPeerGroupStates.set(requestId, {resolve: resolve, reject: reject, timeout: to});
+
+            this.commandForwardingFn(cmd);
+    
+        });
+
+        return p;
+    }
+
     syncObjectWithPeerGroup(peerGroupId: string, obj: HashedObject, mode:SyncMode=SyncMode.full, gossipId?: string, usageToken?: UsageToken): UsageToken {
         
         const ctx = obj.toContext();
