Verify Nexus handler exists before distributing events to mirrors (SY…

…N-7251)

changes/e46aae21fa34bf1a661eae5b0912d248.yaml

@@ -0,0 +1,6 @@
+---
+desc: 'Fixed an issue where Nexus events from updated mirrors pushed to a leader on
+  an older version which did not yet support those events were not handled correctly.'
+prs: []
+type: bug
+...

synapse/lib/nexus.py

@@ -379,6 +379,13 @@ async def setindex(self, indx):
     async def _eat(self, item, indx=None):
 
         if self.donexslog:
+            nexsiden, event = item[:2]
+            if (nexus := self._nexskids.get(nexsiden)) is None:
+                raise s_exc.NoSuchName(mesg=f'No Nexus Pusher with iden {nexsiden}.', name=nexsiden)
+
+            if event not in nexus._nexshands:
+                raise s_exc.NoSuchName(mesg=f'No Nexus handler for event {event}.', name=event)
+
             saveindx = await self.nexslog.add(item, indx=indx)
             [dist.update() for dist in tuple(self._mirrors)]
 

synapse/tests/test_lib_nexus.py

@@ -310,6 +310,32 @@ def wrapAddWriteHold(reason):
                         await cell01.sync()
                     self.isin(s_nexus.leaderversion, cell01.nexsroot.writeholds)
 
+                cell00.getCellInfo = getCellInfo
+
+                # test case where a mirror which is updated first may push events
+                # the leader does not yet have handlers for
+                async with await s_cell.Cell.anit(dirn=path) as cell01:
+                    cell01.nexsiden = 'newp'
+                    with self.raises(s_exc.NoSuchName) as cm:
+                        await cell01.sync()
+                    self.eq(cm.exception.get('mesg'), 'No Nexus Pusher with iden newp.')
+
+                    self.none(await cell00.nexsroot.nexslog.last())
+                    self.none(await cell01.nexsroot.nexslog.last())
+
+                    cell01.nexsiden = cell00.nexsiden
+                    await cell01.sync()
+
+                    self.eq(0, (await cell00.nexsroot.nexslog.last())[0])
+                    self.eq(0, (await cell01.nexsroot.nexslog.last())[0])
+
+                    with self.raises(s_exc.NoSuchName) as cm:
+                        await cell01._push('newp')
+                    self.eq(cm.exception.get('mesg'), 'No Nexus handler for event newp.')
+
+                    self.eq(0, (await cell00.nexsroot.nexslog.last())[0])
+                    self.eq(0, (await cell01.nexsroot.nexslog.last())[0])
+
     async def test_mirror_nexus_loop_failure(self):
         with self.getTestDir() as dirn: