fixes to sync state event system

sbazerque · sbazerque · commit 5dea7f9ac1ce · 2022-11-01T01:03:59.000-03:00
diff --git a/src/mesh/agents/state/HeaderBasedSyncAgent.ts b/src/mesh/agents/state/HeaderBasedSyncAgent.ts
@@ -57,7 +57,7 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {
     synchronizer : HistorySynchronizer;
     provider     : HistoryProvider;
 
-    syncEventSource?: EventRelay<HashedObject>;
+    syncEventSource?: EventRelay<HashedObject, SyncState>;
 
     terminated = false;
 
@@ -104,8 +104,11 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {
                 }
             }
         );
+
         this.watchStoreForOps();
 
+        this.emitSyncState();
+
         this.controlLog.debug('Started agent for ' + this.mutableObjHash);
     }
 
@@ -114,7 +117,7 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {
         const ev: SyncStateUpdateEvent = {
             emitter: this.mutableObj,
             action: SyncObserverEventTypes.SyncStateUpdate,
-            data: {inSync: false, synchronizing: false, opsToFetch: 0, remoteStateHashes: {}}
+            data: {allPeersInSync: false, synchronizing: false, opsToFetch: 0, remoteStateHashes: {}}
         };
 
         this.syncEventSource?.emit(ev);
@@ -157,6 +160,9 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {
                     this.synchronizer.onNewHistory(sender, unknown);
                 }
 
+                this.emitSyncState();
+
+
             }
 
         }
@@ -214,15 +220,7 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {
             await this.updateStateFromStore();  
         }
 
-        if (this.syncEventSource !== undefined) {
-            const ev: SyncStateUpdateEvent = {
-                emitter: this.mutableObj,
-                action: SyncObserverEventTypes.SyncStateUpdate,
-                data: this.getSyncState()
-            };
-
-            this.syncEventSource.emit(ev);
-        }
+        this.emitSyncState();
     };
 
     literalIsValidOp(literal?: Literal, log=false): boolean { 
@@ -354,9 +352,27 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {
         return new EventRelay(this.mutableObj as HashedObject, new Map());
     }
 
+    // by default use our own syncEventSource, but allow the caller to specify another event relay
+    // (used by the SyncObserverAgent to send the initial state)
+    emitSyncState() {
+
+        if (this.syncEventSource !== undefined) {
+            const ev: SyncStateUpdateEvent = {
+                emitter: this.mutableObj,
+                action: SyncObserverEventTypes.SyncStateUpdate,
+                data: this.getSyncState()
+            };
+
+            console.log('new sync state event:', ev);
+            this.syncEventSource.emit(ev);
+        } else {
+            console.log('not emitting new sync state event: no relay')
+        }
+    }
+
     getSyncState(): SyncState {
 
-        const gossipAgent = this.pod?.getAgent(StateGossipAgent.agentIdForGossipId(this.getAgentId())) as StateGossipAgent;
+        const gossipAgent = this.pod?.getAgent(StateGossipAgent.agentIdForGossipId(this.peerGroupAgent.peerGroupId)) as StateGossipAgent;
 
         const remoteStateHashes = gossipAgent.getAllRemoteStateForAgent(this.getAgentId());
         const localStateHash    = gossipAgent.getLocalStateForAgent(this.getAgentId());
@@ -371,7 +387,7 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {
 
         const opsToFetch = this.synchronizer.discoveredHistory.contents.size;
 
-        return {remoteStateHashes: remoteStateHashes, localStateHash: localStateHash, inSync: inSync, opsToFetch: opsToFetch, synchronizing: true}
+        return {remoteStateHashes: remoteStateHashes, localStateHash: localStateHash, allPeersInSync: inSync, opsToFetch: opsToFetch, synchronizing: true}
     }
 
     getMutableObject(): MutableObject {
diff --git a/src/mesh/agents/state/StateGossipAgent.ts b/src/mesh/agents/state/StateGossipAgent.ts
@@ -696,9 +696,11 @@ class StateGossipAgent extends PeeringAgentBase {
             
             if (stateHash !== undefined) {
                 state[endpoint] = stateHash;
+                console.log('ep=', endpoint, ', hash=', stateHash);
             }
         }
 
+        console.log(state);
         return state;
     }
 
diff --git a/src/mesh/agents/state/SyncObserverAgent.ts b/src/mesh/agents/state/SyncObserverAgent.ts
@@ -24,10 +24,12 @@ import { StateSyncAgent } from './StateSyncAgent';
  * - addSyncObserver() is called -> ... to be continued
  */
 
-type SyncObserver = Observer<HashedObject>;
+type SyncState = { remoteStateHashes: {[key: Endpoint]: Hash}, localStateHash?: Hash, allPeersInSync: boolean, opsToFetch: number, synchronizing: boolean };
+
+type SyncObserver = Observer<HashedObject, SyncState>;
 type SyncEvent    = Event<HashedObject, SyncState>;
 
-type SyncState = { remoteStateHashes: {[key: Endpoint]: Hash}, localStateHash?: Hash, inSync: boolean, opsToFetch: number, synchronizing: boolean }
+
 
 enum SyncObserverEventTypes {
     SyncStateUpdate = 'sync-state-update'
@@ -43,7 +45,7 @@ class SyncObserverAgent implements Agent {
 
     pod?: AgentPod;
 
-    relays: Map<AgentId, EventRelay<HashedObject>>;
+    relays: Map<AgentId, [EventRelay<HashedObject>, SyncObserver]>;
 
     constructor() {
         this.relays = new Map();
@@ -59,27 +61,44 @@ class SyncObserverAgent implements Agent {
 
     addSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId: PeerGroupId) {
 
+        console.log('adding sync observer ', obs, ' for ', mut.getLastHash(), ' in peer group ', peerGroupId)
+
         if (this.pod === undefined) {
             throw new Error('Trying to add a sync observer, but the SyncObserverAgent is not ready.');
         }
 
         const syncAgentId = mut.getSyncAgentId(peerGroupId);
+        const syncAgent   = this.pod.getAgent(syncAgentId) as StateSyncAgent|undefined;
 
-        let relay = this.relays.get(syncAgentId);
+        let pair = this.relays.get(syncAgentId);
+        
 
-        if (relay === undefined) {
-            const upstream = new Map<string, EventRelay<HashedObject>>();
+        if (pair === undefined) {
 
-            const syncAgent = this.pod.getAgent(syncAgentId) as StateSyncAgent;
-            if (syncAgent !== undefined) {
-                upstream.set('agent', syncAgent.getSyncEventSource());
+            const relay = new EventRelay<HashedObject>(mut);
+            const syncAgentObs = (ev: SyncEvent) => {
+                relay.emit(ev);
             }
 
-            relay = new EventRelay<HashedObject>(mut, upstream);
-            this.relays.set(syncAgentId, relay);
+            pair = [relay, syncAgentObs];
+    
+            this.relays.set(syncAgentId, [relay, syncAgentObs]);
+    
+            
+            if (syncAgent !== undefined) {
+                syncAgent.getSyncEventSource().addObserver(syncAgentObs);
+            }
         }
 
-        relay.addObserver(obs);
+        pair[0].addObserver(obs);
+
+        if (syncAgent !== undefined) {
+            obs({
+                emitter: mut,
+                action: SyncObserverEventTypes.SyncStateUpdate,
+                data: syncAgent.getSyncState()
+            });
+        }
     }
 
     removeSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId: PeerGroupId) {
@@ -88,13 +107,20 @@ class SyncObserverAgent implements Agent {
         }
 
         const syncAgentId = mut.getSyncAgentId(peerGroupId);
-        let   relay = this.relays.get(syncAgentId);
+        const pair        = this.relays.get(syncAgentId);
 
-        if (relay !== undefined) {
+        if (pair !== undefined) {
+            const [relay, syncAgentObs] = pair;
             relay.removeObserver(obs);
 
             if (relay.observers.size === 0) {
-                relay.removeAllUpstreamRelays();
+
+                const syncAgent = this.pod.getAgent(syncAgentId) as StateSyncAgent|undefined;
+
+                if (syncAgent !== undefined) {
+                    syncAgent.getSyncEventSource().removeObserver(syncAgentObs);
+                }
+                
                 this.relays.delete(syncAgentId);
             }
         }
@@ -106,14 +132,19 @@ class SyncObserverAgent implements Agent {
             const agentEv = ev as AgentSetChangeEvent;
 
             const syncAgentId = agentEv.content.agentId;
-            const relay = this.relays.get(syncAgentId);
+            const pair = this.relays.get(syncAgentId);
 
-            if (relay !== undefined) {
+            if (pair !== undefined) {
                 if (agentEv.content.change === AgentSetChange.Addition) {
                     const syncAgent = this.pod?.getAgent(syncAgentId) as StateSyncAgent;
-                    relay.addUpstreamRelay('agent', syncAgent.getSyncEventSource());
+                    const [_relay, syncAgentObserver] = pair;
+                    syncAgent.getSyncEventSource().addObserver(syncAgentObserver);
+                    console.log('+++ adding agent relay ', syncAgent.getSyncEventSource(), ' to ', _relay)
                 } else if (agentEv.content.change === AgentSetChange.Removal) {
-                    relay.removeUpstreamRelay('agent');
+                    const syncAgent = this.pod?.getAgent(syncAgentId) as StateSyncAgent|undefined;
+                    const [_relay, syncAgentObserver] = pair;
+                    syncAgent?.getSyncEventSource().addObserver(syncAgentObserver);
+                    console.log('+++ removing agent relay from ', _relay)
                 }
     
             }
diff --git a/src/mesh/service/Mesh.ts b/src/mesh/service/Mesh.ts
@@ -4,7 +4,7 @@ import { Store } from 'storage/store';
 import { MultiMap } from 'util/multimap';
 
 import { Endpoint, NetworkAgent, NetworkAgentProxyConfig, SecureNetworkAgent } from '../agents/network';
-import { PeerInfo, PeerSource, PeerGroupAgent, PeerGroupAgentConfig } from '../agents/peer';
+import { PeerInfo, PeerSource, PeerGroupAgent, PeerGroupAgentConfig, ObjectDiscoveryPeerSource } from '../agents/peer';
 import { StateGossipAgent, StateSyncAgent, SyncObserver, SyncObserverAgent, SyncState } from 'mesh/agents/state';
 import { ObjectBroadcastAgent, ObjectDiscoveryAgent, ObjectDiscoveryReply, ObjectDiscoveryReplyParams } from '../agents/discovery';
 
@@ -17,6 +17,7 @@ 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';
+import { Resources } from 'spaces/Resources';
 
 
 /* Connect to the Hyper Hyper Space service mesh.
@@ -101,7 +102,7 @@ enum SyncMode {
 
 class CannotInferPeerGroup extends Error {
     constructor(mutHash: Hash) {
-        super('The sync state for ' + mutHash + ' was requested, but no peerGroupId was specified, and there are more than one peer groups synchronizing this object!');
+        super('The sync state for ' + mutHash + ' was requested, but no peerGroupId was specified, and there is not exactly one peer group synchronizing this object, so we cannot infer which one was intended!');
     }
 }
 
@@ -332,6 +333,8 @@ class Mesh {
             peerGroupId = this.inferPeerGroupId(mut);
         }
 
+        console.log('mesh: observer to ', mut.getLastHash(), ' peerGroupId=', peerGroupId);
+
         this.syncObserver.addSyncObserver(obs, mut, peerGroupId);
     }
 
@@ -843,14 +846,38 @@ class Mesh {
 
     }
 
+    async getDiscoveryPeerGroup(obj: HashedObject, resources?: Resources) : Promise<PeerGroupInfo> {
+
+        resources = resources || obj.getResources();
+
+        if (resources === undefined) {
+            throw new Error('Could not find a valid resources object to use for the discovery peer group.');
+        }
+
+        let localPeer = resources.getPeersForDiscovery()[0];
+        let peerSource = new ObjectDiscoveryPeerSource(this, obj, resources.config.linkupServers, LinkupAddress.fromURL(localPeer.endpoint, localPeer.identity), resources.getEndointParserForDiscovery());
+
+        return {
+            id: Mesh.discoveryPeerGroupId(obj),
+            localPeer: localPeer,
+            peerSource: peerSource
+        };
+
+    }
+
+    static discoveryPeerGroupId(obj: HashedObject) {
+        return  'sync-for-' + obj.getLastHash();
+    }
+
     private inferPeerGroupId(mut: MutableObject) {
         const mutHash = mut.getLastHash();
 
         const agents = this.gossipIdsPerObject.get(mutHash);
 
-        if (agents.size > 0) {
+        if (agents.size !== 1) {
             throw new CannotInferPeerGroup(mutHash);
         } else {
+            console.log('agents ', agents)
             return agents.values().next().value as string;
         }
     }
diff --git a/src/mesh/service/MeshNode.ts b/src/mesh/service/MeshNode.ts
@@ -2,12 +2,12 @@
 import { Identity } from 'data/identity';
 import { Hash, HashedObject } from 'data/model';
 import { Endpoint } from 'mesh/agents/network';
-import { ObjectDiscoveryPeerSource, PeerGroupState, PeerInfo } from 'mesh/agents/peer';
+import { PeerGroupState, PeerInfo } from 'mesh/agents/peer';
 import { ObjectSpawnAgent, SpawnCallback } from 'mesh/agents/spawn';
 import { LinkupAddress, LinkupManager } from 'net/linkup';
 import { Resources } from 'spaces/spaces';
 import { MultiMap } from 'util/multimap';
-import { PeerGroupInfo, SyncMode, UsageToken } from './Mesh';
+import { Mesh, PeerGroupInfo, SyncMode, UsageToken } from './Mesh';
 
 type Key = string;
 
@@ -64,15 +64,15 @@ class MeshNode {
     }
 
     async getPeerGroupStateForSyncObj(obj: HashedObject) {
-        const peerGroup = await this.discoveryPeerGroupInfo(obj);
+        const peerGroup = await this.getDiscoveryPeerGroup(obj);
 
         return this.getPeerGroupState(peerGroup.id);
     }
 
     async sync(obj: HashedObject, mode :SyncMode = SyncMode.full, peerGroup?: PeerGroupInfo): Promise<void> {
 
         if (peerGroup === undefined) {
-            peerGroup = await this.discoveryPeerGroupInfo(obj);
+            peerGroup = await this.getDiscoveryPeerGroup(obj);
         }
 
         const peerGroupKey = MeshNode.generateKey([peerGroup.id]);
@@ -97,7 +97,7 @@ class MeshNode {
 
     async stopSync(obj: HashedObject, peerGroupId?: string, gossipId?: string) : Promise<void> {
         if (peerGroupId === undefined) {
-            peerGroupId = MeshNode.discoveryPeerGroupInfoId(obj);
+            peerGroupId = Mesh.discoveryPeerGroupId(obj);
         }
 
         const syncKey   = MeshNode.generateKey([obj.hash(), peerGroupId, gossipId]);
@@ -129,20 +129,8 @@ class MeshNode {
         this.resources.mesh.sendObjectSpawnRequest(object, sender, receiver, senderEndpoint, receiverLinkupServers, spawnId)
     }
 
-    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, LinkupAddress.fromURL(localPeer.endpoint, localPeer.identity), this.resources.getEndointParserForDiscovery());
-
-        return {
-            id: MeshNode.discoveryPeerGroupInfoId(obj),
-            localPeer: localPeer,
-            peerSource: peerSource
-        };
-
-    }
-
-    private static discoveryPeerGroupInfoId(obj: HashedObject) {
-        return  'sync-for-' + obj.hash();
+    async getDiscoveryPeerGroup(obj: HashedObject) : Promise<PeerGroupInfo> {
+        return this.resources.mesh.getDiscoveryPeerGroup(obj, this.resources);
     }
 
     private static generateKey(parts: (string|undefined)[]): string {
@@ -164,7 +152,7 @@ class MeshNode {
     async expectingMoreOps(obj: HashedObject, receivedOps?: Set<Hash>, peerGroupId?: string, rootObject?: HashedObject): Promise<boolean> {
 
         if (peerGroupId === undefined) {
-            const peerGroup = await this.discoveryPeerGroupInfo(rootObject !== undefined? rootObject : obj);
+            const peerGroup = await this.getDiscoveryPeerGroup(rootObject !== undefined? rootObject : obj);
             peerGroupId = peerGroup.id;
         }
 
diff --git a/src/util/events.ts b/src/util/events.ts

Original file line number	Diff line number	Diff line change
`@@ -696,9 +696,11 @@ class StateGossipAgent extends PeeringAgentBase {`
`696`	`696`
`697`	`697`	`if (stateHash !== undefined) {`
`698`	`698`	`state[endpoint] = stateHash;`
	`699`	`+ console.log('ep=', endpoint, ', hash=', stateHash);`
`699`	`700`	`}`
`700`	`701`	`}`
`701`	`702`
	`703`	`+ console.log(state);`
`702`	`704`	`return state;`
`703`	`705`	`}`
`704`	`706`