block /stream and /get requests on the stream manager starting

lbrynet/extras/daemon/Daemon.py

@@ -473,6 +473,8 @@ class Daemon(metaclass=JSONRPCServerType):
         else:
             name, claim_id = name_and_claim_id.split("/")
             uri = f"lbry://{name}#{claim_id}"
+        if not self.stream_manager.started.is_set():
+            await self.stream_manager.started.wait()
         stream = await self.jsonrpc_get(uri)
         if isinstance(stream, dict):
             raise web.HTTPServerError(text=stream['error'])
@@ -480,6 +482,8 @@ class Daemon(metaclass=JSONRPCServerType):
 
     async def handle_stream_range_request(self, request: web.Request):
         sd_hash = request.path.split("/stream/")[1]
+        if not self.stream_manager.started.is_set():
+            await self.stream_manager.started.wait()
         if sd_hash not in self.stream_manager.streams:
             return web.HTTPNotFound()
         return await self.stream_manager.stream_partial_content(request, sd_hash)

lbrynet/stream/stream_manager.py

@@ -74,6 +74,7 @@ class StreamManager:
         self.re_reflect_task: asyncio.Task = None
         self.update_stream_finished_futs: typing.List[asyncio.Future] = []
         self.running_reflector_uploads: typing.List[asyncio.Task] = []
+        self.started = asyncio.Event(loop=self.loop)
 
     async def _update_content_claim(self, stream: ManagedStream):
         claim_info = await self.storage.get_content_claim(stream.stream_hash)
@@ -191,6 +192,7 @@ class StreamManager:
         await self.load_streams_from_database()
         self.resume_downloading_task = self.loop.create_task(self.resume())
         self.re_reflect_task = self.loop.create_task(self.reflect_streams())
+        self.started.set()
 
     def stop(self):
         if self.resume_downloading_task and not self.resume_downloading_task.done():
@@ -204,6 +206,7 @@ class StreamManager:
             self.update_stream_finished_futs.pop().cancel()
         while self.running_reflector_uploads:
             self.running_reflector_uploads.pop().cancel()
+        self.started.clear()
 
     async def create_stream(self, file_path: str, key: typing.Optional[bytes] = None,
                             iv_generator: typing.Optional[typing.Generator[bytes, None, None]] = None) -> ManagedStream: