ensure that storm httpapis tear down runtime tasks if the remote disconnects

synapse/lib/httpapi.py

@@ -300,11 +300,16 @@ async def data_received(self, chunk):
 
 class StormNodesV1(Handler):
 
+    def on_connection_close(self):
+        if self.task:
+            self.task.schedCoroSafe(self.task.fini())
+
     async def post(self):
         return await self.get()
 
     async def get(self):
 
+        self.task = None
         user, body = await self.getUserBody()
         if body is s_common.novalu:
             return
@@ -313,7 +318,7 @@ async def get(self):
         opts = body.get('opts')
         query = body.get('query')
 
-        await self.cell.boss.promote('storm', user=user, info={'query': query})
+        self.task = await self.cell.boss.promote('storm', user=user, info={'query': query})
 
         opts = await self._reqValidOpts(opts)
 
@@ -324,11 +329,15 @@ async def get(self):
 
 class StormV1(Handler):
 
+    def on_connection_close(self):
+        if self.task:
+            self.task.schedCoroSafe(self.task.fini())
+
     async def post(self):
         return await self.get()
 
     async def get(self):
-
+        self.task = None
         user, body = await self.getUserBody()
         if body is s_common.novalu:
             return
@@ -341,7 +350,7 @@ async def get(self):
         opts = await self._reqValidOpts(opts)
         opts['editformat'] = 'splices'
 
-        await self.cell.boss.promote('storm', user=user, info={'query': query})
+        self.task = await self.cell.boss.promote('storm', user=user, info={'query': query})
 
         async for mesg in self.cell.storm(query, opts=opts):
             self.write(json.dumps(mesg))

synapse/tests/test_lib_httpapi.py

@@ -773,6 +773,42 @@ async def test_http_storm(self):
 
                     self.eq(0x01020304, node[0][1])
 
+                # Task cancellation during long running storm queries works as intended
+                body = {'query': '.created | sleep 10'}
+                task = None
+                async with sess.get(f'https://localhost:{port}/api/v1/storm', json=body) as resp:
+
+                    async for byts, x in resp.content.iter_chunks():
+
+                        if not byts:
+                            break
+
+                        mesg = json.loads(byts)
+                        if mesg[0] == 'node':
+                            task = core.boss.tasks.get(list(core.boss.tasks.keys())[0])
+                            break
+
+                self.nn(task)
+                self.true(await task.waitfini(10))
+                self.len(0, core.boss.tasks)
+
+                task = None
+                async with sess.get(f'https://localhost:{port}/api/v1/storm/nodes', json=body) as resp:
+
+                    async for byts, x in resp.content.iter_chunks():
+
+                        if not byts:
+                            break
+
+                        mesg = json.loads(byts)
+                        self.len(2, mesg) # Is if roughly shaped like a node?
+                        task = core.boss.tasks.get(list(core.boss.tasks.keys())[0])
+                        break
+
+                self.nn(task)
+                self.true(await task.waitfini(10))
+                self.len(0, core.boss.tasks)
+
                 # check reqvalidstorm with various queries
                 tvs = (
                     ('test:str=test', {}, 'ok'),