fix for mirror window consumer

synapse/cortex.py

@@ -788,12 +788,11 @@ async def _initCoreMirror(self, url):
                         await self.fini()
                         return
 
-                    logger.warning(f'mirror loop connected ({url}')
-
                     # assume only the main layer for now...
                     layr = self.getLayer()
 
                     offs = await layr.getOffset(layr.iden)
+                    logger.warning(f'mirror loop connected ({url} offset={offs})')
 
                     if offs == 0:
                         stat = await layr.stat()
@@ -842,7 +841,7 @@ async def consume(x):
                             await layr.setOffset(layr.iden, items[-1][0])
 
             except asyncio.CancelledError: # pragma: no cover
-                raise
+                return
 
             except Exception:
                 logger.exception('error in initCoreMirror loop')

synapse/lib/layer.py

@@ -195,16 +195,26 @@ async def stor(self, sops, splices=None):
             await func(oper)
 
         if splices:
+            await self._storFireSplices(splices)
 
-            await self._storSplices(splices)
+    async def _storFireSplices(self, splices):
+        '''
+        Fire events, windows, etc for splices.
+        '''
+        indx = await self._storSplices(splices)
 
-            self.spliced.set()
-            self.spliced.clear()
+        self.spliced.set()
+        self.spliced.clear()
 
-            # go fast and protect against edit-while-iter issues
-            [(await wind.puts(splices)) for wind in tuple(self.windows)]
+        items = [(indx + i, s) for (i, s) in enumerate(splices)]
+        # go fast and protect against edit-while-iter issues
+        [(await wind.puts(items)) for wind in tuple(self.windows)]
 
     async def _storSplices(self, splices):  # pragma: no cover
+        '''
+        Store the splices into a sequentially accessible storage structure.
+        Returns the indx of the first splice stored.
+        '''
         raise NotImplementedError
 
     async def _liftByFormRe(self, oper):

synapse/lib/lmdblayer.py

@@ -99,9 +99,7 @@ async def stor(self, sops, splices=None):
             func(oper)
 
         if splices:
-            self._storSplicesSync(splices)
-            self.spliced.set()
-            self.spliced.clear()
+            await self._storFireSplices(splices)
 
     def _migrate_db_pre010(self, dbname, newslab):
         '''
@@ -397,8 +395,9 @@ def _storPropDel(self, oper):
             if univ:
                 self.layrslab.delete(penc + oldi, pvvalu, db=self.byuniv)
 
-    def _storSplicesSync(self, splices):
-        self.splicelog.save(splices)
+    async def _storSplices(self, splices):
+        info = self.splicelog.save(splices)
+        return info.get('orig')
 
     async def _liftByIndx(self, oper):
         # ('indx', (<dbname>, <prefix>, (<indxopers>...))

synapse/lib/node.py

@@ -229,7 +229,6 @@ async def _setops(self, name, valu, editatom, init=False):
                     raise s_exc.ReadOnlyProp(name=prop.full)
 
                 # not setting a set-once prop unless we are init...
-                await self.snap.warn(f'ReadOnlyProp: name={prop.full}')
                 return False
 
             # check for type specific merging...

synapse/lib/slabseqn.py

@@ -74,9 +74,8 @@ def save(self, items):
 
         origindx = self.indx
         self.indx = indx
-        return {'indx': indx, 'size': size, 'count': len(items), 'time': tick, 'took': took}
 
-        return origindx
+        return {'indx': indx, 'size': size, 'count': len(items), 'time': tick, 'took': took, 'orig': origindx}
 
     def append(self, item):
 

synapse/tests/test_servers_cortex.py

@@ -55,12 +55,22 @@ async def test_server_mirror(self):
                         '--mirror', core00.getLocalUrl(),
                         path01]
 
+                # add a node for core01 to sync before window
+                await core00.nodes('[ inet:ipv4=5.5.5.5 ]')
+
                 async with await s_s_cortex.main(argv, outp=outp) as core01:
-                    evnt = await core01._getWaitFor('inet:ipv4', '5.5.5.5')
-                    await core00.nodes('[ inet:ipv4=5.5.5.5 ]')
-                    await evnt.wait()
+
+                    evnt00 = await core01._getWaitFor('inet:ipv4', '5.5.5.5')
+                    await evnt00.wait()
                     self.len(1, await core01.nodes('inet:ipv4=5.5.5.5'))
 
+                    # add a node for core01 to sync via window
+                    evnt01 = await core01._getWaitFor('inet:ipv4', '6.6.6.6')
+                    self.len(1, await core00.nodes('[ inet:ipv4=6.6.6.6 ]'))
+
+                    await evnt01.wait()
+                    self.len(1, await core01.nodes('inet:ipv4=6.6.6.6'))
+
     async def test_server_mirror_badiden(self):
 
         with self.getTestDir() as dirn: