LBRYCommunity/lbry-sdk
1
0
Fork 1
Code Issues Pull requests Projects Releases 282 Packages Wiki Activity

fix reflector loop propagating cancelled error

Browse source
This commit is contained in:
Jack Robison 2021-01-08 09:53:12 -05:00
parent 652773d2cf
commit 7952fc8324
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
2 changed files with 5 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
lbry/stream/managed_stream.py
View file

@ -354,20 +354,19 @@ class ManagedStream(ManagedDownloadSource):
self.reflector_progress = int((i + 1) / len(we_have) * 100) self.reflector_progress = int((i + 1) / len(we_have) * 100)
except (asyncio.TimeoutError, ValueError): except (asyncio.TimeoutError, ValueError):
return sent return sent
except ConnectionRefusedError: except ConnectionError:
return sent return sent
except (OSError, Exception) as err: except (OSError, Exception) as err:
if isinstance(err, asyncio.CancelledError): if isinstance(err, asyncio.CancelledError):
log.warning("stopped uploading %s#%s to reflector", self.claim_name, self.claim_id) log.warning("stopped uploading %s#%s to reflector", self.claim_name, self.claim_id)
raise err elif isinstance(err, OSError):
if isinstance(err, OSError):
log.warning( log.warning(
"stopped uploading %s#%s to reflector because blobs were deleted or moved", self.claim_name, "stopped uploading %s#%s to reflector because blobs were deleted or moved", self.claim_name,
self.claim_id self.claim_id
) )
else: else:
log.exception("unexpected error reflecting %s#%s", self.claim_name, self.claim_id) log.exception("unexpected error reflecting %s#%s", self.claim_name, self.claim_id)
raise asyncio.CancelledError() return sent
finally: finally:
if protocol.transport: if protocol.transport:
protocol.transport.close() protocol.transport.close()

4
lbry/stream/stream_manager.py
View file

@ -160,13 +160,13 @@ class StreamManager(SourceManager):
async def reflect_streams(self): async def reflect_streams(self):
try: try:
return await self._reflact_streams() return await self._reflect_streams()
except asyncio.CancelledError: except asyncio.CancelledError:
raise raise
except Exception: except Exception:
log.exception("reflector task encountered an unexpected error!") log.exception("reflector task encountered an unexpected error!")
async def _reflact_streams(self): async def _reflect_streams(self):
# todo: those debug statements are temporary for #2987 - remove them if its closed # todo: those debug statements are temporary for #2987 - remove them if its closed
while True: while True:
if self.config.reflect_streams and self.config.reflector_servers: if self.config.reflect_streams and self.config.reflector_servers: