4599645360

Build Type

push

github

Committed by GitHub

Commit Message

Bump cryptography from 2.5 to 39.0.1

Run Details

2807 of 6557 branches covered (42.81%)

Branch coverage included in aggregate %.

12289 of 19915 relevant lines covered (61.71%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.12

/lbry/stream/stream_manager.py

import os
import asyncio
import binascii
import logging
import random
import typing
from typing import Optional
from aiohttp.web import Request
from lbry.error import InvalidStreamDescriptorError
from lbry.file.source_manager import SourceManager
from lbry.stream.descriptor import StreamDescriptor
from lbry.stream.managed_stream import ManagedStream
from lbry.file.source import ManagedDownloadSource
if typing.TYPE_CHECKING:
    from lbry.conf import Config
    from lbry.blob.blob_manager import BlobManager
    from lbry.dht.node import Node
    from lbry.wallet.wallet import WalletManager
    from lbry.wallet.transaction import Transaction
    from lbry.extras.daemon.analytics import AnalyticsManager
    from lbry.extras.daemon.storage import SQLiteStorage, StoredContentClaim

log = logging.getLogger(__name__)


def path_or_none(encoded_path) -> Optional[str]:
    if not encoded_path:
        return
    return binascii.unhexlify(encoded_path).decode()


class StreamManager(SourceManager):
    _sources: typing.Dict[str, ManagedStream]

    filter_fields = SourceManager.filter_fields
    filter_fields.update({
        'sd_hash',
        'stream_hash',
        'full_status',  # TODO: remove
        'blobs_remaining',
        'blobs_in_stream',
        'uploading_to_reflector',
        'is_fully_reflected'
    })

    def __init__(self, loop: asyncio.AbstractEventLoop, config: 'Config', blob_manager: 'BlobManager',
                 wallet_manager: 'WalletManager', storage: 'SQLiteStorage', node: Optional['Node'],
                 analytics_manager: Optional['AnalyticsManager'] = None):
        super().__init__(loop, config, storage, analytics_manager)
        self.blob_manager = blob_manager
        self.wallet_manager = wallet_manager
        self.node = node
        self.resume_saving_task: Optional[asyncio.Task] = None
        self.re_reflect_task: Optional[asyncio.Task] = None
        self.update_stream_finished_futs: typing.List[asyncio.Future] = []
        self.running_reflector_uploads: typing.Dict[str, asyncio.Task] = {}
        self.started = asyncio.Event()

    @property
    def streams(self):
        return self._sources

    def add(self, source: ManagedStream):
        super().add(source)
        self.storage.content_claim_callbacks[source.stream_hash] = lambda: self._update_content_claim(source)

    async def _update_content_claim(self, stream: ManagedStream):
        claim_info = await self.storage.get_content_claim(stream.stream_hash)
        self._sources.setdefault(stream.sd_hash, stream).set_claim(claim_info, claim_info['value'])

    async def recover_streams(self, file_infos: typing.List[typing.Dict]):
        to_restore = []
        to_check = []

        async def recover_stream(sd_hash: str, stream_hash: str, stream_name: str,
                                 suggested_file_name: str, key: str,
                                 content_fee: Optional['Transaction']) -> Optional[StreamDescriptor]:
            sd_blob = self.blob_manager.get_blob(sd_hash)
            blobs = await self.storage.get_blobs_for_stream(stream_hash)
            descriptor = await StreamDescriptor.recover(
                self.blob_manager.blob_dir, sd_blob, stream_hash, stream_name, suggested_file_name, key, blobs
            )
            if not descriptor:
                return
            to_restore.append((descriptor, sd_blob, content_fee))
            to_check.extend([sd_blob.blob_hash] + [blob.blob_hash for blob in descriptor.blobs[:-1]])

        await asyncio.gather(*[
            recover_stream(
                file_info['sd_hash'], file_info['stream_hash'], binascii.unhexlify(file_info['stream_name']).decode(),
                binascii.unhexlify(file_info['suggested_file_name']).decode(), file_info['key'],
                file_info['content_fee']
            ) for file_info in file_infos
        ])

        if to_restore:
            await self.storage.recover_streams(to_restore, self.config.download_dir)
        if to_check:
            await self.blob_manager.ensure_completed_blobs_status(to_check)

        # if self.blob_manager._save_blobs:
        #     log.info("Recovered %i/%i attempted streams", len(to_restore), len(file_infos))

    async def _load_stream(self, rowid: int, sd_hash: str, file_name: Optional[str],
                           download_directory: Optional[str], status: str,
                           claim: Optional['StoredContentClaim'], content_fee: Optional['Transaction'],
                           added_on: Optional[int], fully_reflected: Optional[bool]):
        try:
            descriptor = await self.blob_manager.get_stream_descriptor(sd_hash)
        except InvalidStreamDescriptorError as err:
            log.warning("Failed to start stream for sd %s - %s", sd_hash, str(err))
            return
        stream = ManagedStream(
            self.loop, self.config, self.blob_manager, descriptor.sd_hash, download_directory, file_name, status,
            claim, content_fee=content_fee, rowid=rowid, descriptor=descriptor,
            analytics_manager=self.analytics_manager, added_on=added_on
        )
        if fully_reflected:
            stream.fully_reflected.set()
        self.add(stream)

    async def initialize_from_database(self):
        to_recover = []
        to_start = []

        await self.storage.update_manually_removed_files_since_last_run()

        for file_info in await self.storage.get_all_lbry_files():
            # if the sd blob is not verified, try to reconstruct it from the database
            # this could either be because the blob files were deleted manually or save_blobs was not true when
            # the stream was downloaded
            if not self.blob_manager.is_blob_verified(file_info['sd_hash']):
                to_recover.append(file_info)
            to_start.append(file_info)
        if to_recover:
            await self.recover_streams(to_recover)

        log.info("Initializing %i files", len(to_start))
        to_resume_saving = []
        add_stream_tasks = []
        for file_info in to_start:
            file_name = path_or_none(file_info['file_name'])
            download_directory = path_or_none(file_info['download_directory'])
            if file_name and download_directory and not file_info['saved_file'] and file_info['status'] == 'running':
                to_resume_saving.append((file_name, download_directory, file_info['sd_hash']))
            add_stream_tasks.append(self.loop.create_task(self._load_stream(
                file_info['rowid'], file_info['sd_hash'], file_name,
                download_directory, file_info['status'],
                file_info['claim'], file_info['content_fee'],
                file_info['added_on'], file_info['fully_reflected']
            )))
        if add_stream_tasks:
            await asyncio.gather(*add_stream_tasks)
        log.info("Started stream manager with %i files", len(self._sources))
        if not self.node:
            log.info("no DHT node given, resuming downloads trusting that we can contact reflector")
        if to_resume_saving:
            log.info("Resuming saving %i files", len(to_resume_saving))
            self.resume_saving_task = asyncio.ensure_future(asyncio.gather(
                *(self._sources[sd_hash].save_file(file_name, download_directory)
                  for (file_name, download_directory, sd_hash) in to_resume_saving),
            ))

    async def reflect_streams(self):
        try:
            return await self._reflect_streams()
        except Exception:
            log.exception("reflector task encountered an unexpected error!")

    async def _reflect_streams(self):
        # todo: those debug statements are temporary for #2987 - remove them if its closed
        while True:
            if self.config.reflect_streams and self.config.reflector_servers:
                log.debug("collecting streams to reflect")
                sd_hashes = await self.storage.get_streams_to_re_reflect()
                sd_hashes = [sd for sd in sd_hashes if sd in self._sources]
                batch = []
                while sd_hashes:
                    stream = self.streams[sd_hashes.pop()]
                    if self.blob_manager.is_blob_verified(stream.sd_hash) and stream.blobs_completed and \
                            stream.sd_hash not in self.running_reflector_uploads and not \
                            stream.fully_reflected.is_set():
                        batch.append(self.reflect_stream(stream))
                    if len(batch) >= self.config.concurrent_reflector_uploads:
                        log.debug("waiting for batch of %s reflecting streams", len(batch))
                        await asyncio.gather(*batch)
                        log.debug("done processing %s streams", len(batch))
                        batch = []
                if batch:
                    log.debug("waiting for batch of %s reflecting streams", len(batch))
                    await asyncio.gather(*batch)
                    log.debug("done processing %s streams", len(batch))
            await asyncio.sleep(300)

    async def start(self):
        await super().start()
        self.re_reflect_task = self.loop.create_task(self.reflect_streams())

    async def stop(self):
        await super().stop()
        if self.resume_saving_task and not self.resume_saving_task.done():
            self.resume_saving_task.cancel()
        if self.re_reflect_task and not self.re_reflect_task.done():
            self.re_reflect_task.cancel()
        while self.update_stream_finished_futs:
            self.update_stream_finished_futs.pop().cancel()
        while self.running_reflector_uploads:
            _, t = self.running_reflector_uploads.popitem()
            t.cancel()
        self.started.clear()
        log.info("finished stopping the stream manager")

    def reflect_stream(self, stream: ManagedStream, server: Optional[str] = None,
                       port: Optional[int] = None) -> asyncio.Task:
        if not server or not port:
            server, port = random.choice(self.config.reflector_servers)
        if stream.sd_hash in self.running_reflector_uploads:
            return self.running_reflector_uploads[stream.sd_hash]
        task = self.loop.create_task(self._retriable_reflect_stream(stream, server, port))
        self.running_reflector_uploads[stream.sd_hash] = task
        task.add_done_callback(
            lambda _: None if stream.sd_hash not in self.running_reflector_uploads else
            self.running_reflector_uploads.pop(stream.sd_hash)
        )
        return task

    @staticmethod
    async def _retriable_reflect_stream(stream, host, port):
        sent = await stream.upload_to_reflector(host, port)
        while not stream.is_fully_reflected and stream.reflector_progress > 0 and len(sent) > 0:
            stream.reflector_progress = 0
            sent = await stream.upload_to_reflector(host, port)
        return sent

    async def create(self, file_path: str, key: Optional[bytes] = None,
                     iv_generator: Optional[typing.Generator[bytes, None, None]] = None) -> ManagedStream:
        descriptor = await StreamDescriptor.create_stream(
            self.loop, self.blob_manager.blob_dir, file_path, key=key, iv_generator=iv_generator,
            blob_completed_callback=self.blob_manager.blob_completed
        )
        await self.storage.store_stream(
            self.blob_manager.get_blob(descriptor.sd_hash, is_mine=True), descriptor
        )
        row_id = await self.storage.save_published_file(
            descriptor.stream_hash, os.path.basename(file_path), os.path.dirname(file_path), 0
        )
        stream = ManagedStream(
            self.loop, self.config, self.blob_manager, descriptor.sd_hash, os.path.dirname(file_path),
            os.path.basename(file_path), status=ManagedDownloadSource.STATUS_FINISHED,
            rowid=row_id, descriptor=descriptor
        )
        self.streams[stream.sd_hash] = stream
        self.storage.content_claim_callbacks[stream.stream_hash] = lambda: self._update_content_claim(stream)
        if self.config.reflect_streams and self.config.reflector_servers:
            self.reflect_stream(stream)
        return stream

    async def delete(self, source: ManagedDownloadSource, delete_file: Optional[bool] = False):
        if not isinstance(source, ManagedStream):
            return
        if source.identifier in self.running_reflector_uploads:
            self.running_reflector_uploads[source.identifier].cancel()
        await source.stop_tasks()
        if source.identifier in self.streams:
            del self.streams[source.identifier]
        blob_hashes = [source.identifier] + [b.blob_hash for b in source.descriptor.blobs[:-1]]
        await self.blob_manager.delete_blobs(blob_hashes, delete_from_db=False)
        await self.storage.delete_stream(source.descriptor)
        if delete_file and source.output_file_exists:
            os.remove(source.full_path)

    async def stream_partial_content(self, request: Request, sd_hash: str):
        stream = self._sources[sd_hash]
        if not stream.downloader.node:
            stream.downloader.node = self.node
        return await stream.stream_file(request)

1	import os	1✔
2	import asyncio	1✔
3	import binascii	1✔
4	import logging	1✔
5	import random	1✔
6	import typing	1✔
7	from typing import Optional	1✔
8	from aiohttp.web import Request	1✔
9	from lbry.error import InvalidStreamDescriptorError	1✔
10	from lbry.file.source_manager import SourceManager	1✔
11	from lbry.stream.descriptor import StreamDescriptor	1✔
12	from lbry.stream.managed_stream import ManagedStream	1✔
13	from lbry.file.source import ManagedDownloadSource	1✔
14	if typing.TYPE_CHECKING:	1!
15	from lbry.conf import Config	×
16	from lbry.blob.blob_manager import BlobManager	×
17	from lbry.dht.node import Node	×
18	from lbry.wallet.wallet import WalletManager	×
19	from lbry.wallet.transaction import Transaction	×
20	from lbry.extras.daemon.analytics import AnalyticsManager	×
21	from lbry.extras.daemon.storage import SQLiteStorage, StoredContentClaim	×
22
23	log = logging.getLogger(__name__)	1✔
24
25
26	def path_or_none(encoded_path) -> Optional[str]:	1✔
27	if not encoded_path:	1✔
28	return	1✔
29	return binascii.unhexlify(encoded_path).decode()	1✔
30
31
32	class StreamManager(SourceManager):	1✔
33	_sources: typing.Dict[str, ManagedStream]	1✔
34
35	filter_fields = SourceManager.filter_fields	1✔
36	filter_fields.update({	1✔
37	'sd_hash',
38	'stream_hash',
39	'full_status', # TODO: remove
40	'blobs_remaining',
41	'blobs_in_stream',
42	'uploading_to_reflector',
43	'is_fully_reflected'
44	})
45
46	def __init__(self, loop: asyncio.AbstractEventLoop, config: 'Config', blob_manager: 'BlobManager',	1✔
47	wallet_manager: 'WalletManager', storage: 'SQLiteStorage', node: Optional['Node'],
48	analytics_manager: Optional['AnalyticsManager'] = None):
49	super().__init__(loop, config, storage, analytics_manager)	1✔
50	self.blob_manager = blob_manager	1✔
51	self.wallet_manager = wallet_manager	1✔
52	self.node = node	1✔
53	self.resume_saving_task: Optional[asyncio.Task] = None	1✔
54	self.re_reflect_task: Optional[asyncio.Task] = None	1✔
55	self.update_stream_finished_futs: typing.List[asyncio.Future] = []	1✔
56	self.running_reflector_uploads: typing.Dict[str, asyncio.Task] = {}	1✔
57	self.started = asyncio.Event()	1✔
58
59	@property	1✔
60	def streams(self):	1✔
61	return self._sources	1✔
62
63	def add(self, source: ManagedStream):	1✔
64	super().add(source)	1✔
65	self.storage.content_claim_callbacks[source.stream_hash] = lambda: self._update_content_claim(source)	1✔
66
67	async def _update_content_claim(self, stream: ManagedStream):	1✔
68	claim_info = await self.storage.get_content_claim(stream.stream_hash)	1✔
69	self._sources.setdefault(stream.sd_hash, stream).set_claim(claim_info, claim_info['value'])	1✔
70
71	async def recover_streams(self, file_infos: typing.List[typing.Dict]):	1✔
72	to_restore = []	1✔
73	to_check = []	1✔
74
75	async def recover_stream(sd_hash: str, stream_hash: str, stream_name: str,	1✔
76	suggested_file_name: str, key: str,
77	content_fee: Optional['Transaction']) -> Optional[StreamDescriptor]:
78	sd_blob = self.blob_manager.get_blob(sd_hash)	1✔
79	blobs = await self.storage.get_blobs_for_stream(stream_hash)	1✔
80	descriptor = await StreamDescriptor.recover(	1✔
81	self.blob_manager.blob_dir, sd_blob, stream_hash, stream_name, suggested_file_name, key, blobs
82	)
83	if not descriptor:	1!
84	return	×
85	to_restore.append((descriptor, sd_blob, content_fee))	1✔
86	to_check.extend([sd_blob.blob_hash] + [blob.blob_hash for blob in descriptor.blobs[:-1]])	1✔
87
88	await asyncio.gather(*[	1✔
89	recover_stream(
90	file_info['sd_hash'], file_info['stream_hash'], binascii.unhexlify(file_info['stream_name']).decode(),
91	binascii.unhexlify(file_info['suggested_file_name']).decode(), file_info['key'],
92	file_info['content_fee']
93	) for file_info in file_infos
94	])
95
96	if to_restore:	1!
97	await self.storage.recover_streams(to_restore, self.config.download_dir)	1✔
98	if to_check:	1!
99	await self.blob_manager.ensure_completed_blobs_status(to_check)	1✔
100
101	# if self.blob_manager._save_blobs:
102	# log.info("Recovered %i/%i attempted streams", len(to_restore), len(file_infos))
103
104	async def _load_stream(self, rowid: int, sd_hash: str, file_name: Optional[str],	1✔
105	download_directory: Optional[str], status: str,
106	claim: Optional['StoredContentClaim'], content_fee: Optional['Transaction'],
107	added_on: Optional[int], fully_reflected: Optional[bool]):
108	try:	1✔
109	descriptor = await self.blob_manager.get_stream_descriptor(sd_hash)	1✔
110	except InvalidStreamDescriptorError as err:	×
111	log.warning("Failed to start stream for sd %s - %s", sd_hash, str(err))	×
112	return	×
113	stream = ManagedStream(	1✔
114	self.loop, self.config, self.blob_manager, descriptor.sd_hash, download_directory, file_name, status,
115	claim, content_fee=content_fee, rowid=rowid, descriptor=descriptor,
116	analytics_manager=self.analytics_manager, added_on=added_on
117	)
118	if fully_reflected:	1!
119	stream.fully_reflected.set()	×
120	self.add(stream)	1✔
121
122	async def initialize_from_database(self):	1✔
123	to_recover = []	1✔
124	to_start = []	1✔
125
126	await self.storage.update_manually_removed_files_since_last_run()	1✔
127
128	for file_info in await self.storage.get_all_lbry_files():	1✔
129	# if the sd blob is not verified, try to reconstruct it from the database
130	# this could either be because the blob files were deleted manually or save_blobs was not true when
131	# the stream was downloaded
132	if not self.blob_manager.is_blob_verified(file_info['sd_hash']):	1✔
133	to_recover.append(file_info)	1✔
134	to_start.append(file_info)	1✔
135	if to_recover:	1✔
136	await self.recover_streams(to_recover)	1✔
137
138	log.info("Initializing %i files", len(to_start))	1✔
139	to_resume_saving = []	1✔
140	add_stream_tasks = []	1✔
141	for file_info in to_start:	1✔
142	file_name = path_or_none(file_info['file_name'])	1✔
143	download_directory = path_or_none(file_info['download_directory'])	1✔
144	if file_name and download_directory and not file_info['saved_file'] and file_info['status'] == 'running':	1!
145	to_resume_saving.append((file_name, download_directory, file_info['sd_hash']))	×
146	add_stream_tasks.append(self.loop.create_task(self._load_stream(	1✔
147	file_info['rowid'], file_info['sd_hash'], file_name,
148	download_directory, file_info['status'],
149	file_info['claim'], file_info['content_fee'],
150	file_info['added_on'], file_info['fully_reflected']
151	)))
152	if add_stream_tasks:	1!
153	await asyncio.gather(*add_stream_tasks)	1✔
154	log.info("Started stream manager with %i files", len(self._sources))	1✔
155	if not self.node:	1!
156	log.info("no DHT node given, resuming downloads trusting that we can contact reflector")	×
157	if to_resume_saving:	1!
158	log.info("Resuming saving %i files", len(to_resume_saving))	×
159	self.resume_saving_task = asyncio.ensure_future(asyncio.gather(	×
160	*(self._sources[sd_hash].save_file(file_name, download_directory)
161	for (file_name, download_directory, sd_hash) in to_resume_saving),
162	))
163
164	async def reflect_streams(self):	1✔
165	try:	1✔
166	return await self._reflect_streams()	1✔
167	except Exception:	1✔
168	log.exception("reflector task encountered an unexpected error!")	×
169
170	async def _reflect_streams(self):	1✔
171	# todo: those debug statements are temporary for #2987 - remove them if its closed
172	while True:
173	if self.config.reflect_streams and self.config.reflector_servers:	1!
174	log.debug("collecting streams to reflect")	1✔
175	sd_hashes = await self.storage.get_streams_to_re_reflect()	1✔
176	sd_hashes = [sd for sd in sd_hashes if sd in self._sources]	1✔
177	batch = []	1✔
178	while sd_hashes:	1✔
179	stream = self.streams[sd_hashes.pop()]	1✔
180	if self.blob_manager.is_blob_verified(stream.sd_hash) and stream.blobs_completed and \	1✔
181	stream.sd_hash not in self.running_reflector_uploads and not \
182	stream.fully_reflected.is_set():
183	batch.append(self.reflect_stream(stream))	1✔
184	if len(batch) >= self.config.concurrent_reflector_uploads:	1!
185	log.debug("waiting for batch of %s reflecting streams", len(batch))	×
186	await asyncio.gather(*batch)	×
187	log.debug("done processing %s streams", len(batch))	×
188	batch = []	×
189	if batch:	1✔
190	log.debug("waiting for batch of %s reflecting streams", len(batch))	1✔
191	await asyncio.gather(*batch)	1✔
192	log.debug("done processing %s streams", len(batch))	1✔
193	await asyncio.sleep(300)	1✔
194
195	async def start(self):	1✔
196	await super().start()	1✔
197	self.re_reflect_task = self.loop.create_task(self.reflect_streams())	1✔
198
199	async def stop(self):	1✔
200	await super().stop()	1✔
201	if self.resume_saving_task and not self.resume_saving_task.done():	1!
202	self.resume_saving_task.cancel()	×
203	if self.re_reflect_task and not self.re_reflect_task.done():	1!
204	self.re_reflect_task.cancel()	×
205	while self.update_stream_finished_futs:	1!
206	self.update_stream_finished_futs.pop().cancel()	×
207	while self.running_reflector_uploads:	1!
208	_, t = self.running_reflector_uploads.popitem()	×
209	t.cancel()	×
210	self.started.clear()	1✔
211	log.info("finished stopping the stream manager")	1✔
212
213	def reflect_stream(self, stream: ManagedStream, server: Optional[str] = None,	1✔
214	port: Optional[int] = None) -> asyncio.Task:
215	if not server or not port:	1!
216	server, port = random.choice(self.config.reflector_servers)	1✔
217	if stream.sd_hash in self.running_reflector_uploads:	1!
218	return self.running_reflector_uploads[stream.sd_hash]	×
219	task = self.loop.create_task(self._retriable_reflect_stream(stream, server, port))	1✔
220	self.running_reflector_uploads[stream.sd_hash] = task	1✔
221	task.add_done_callback(	1✔
222	lambda _: None if stream.sd_hash not in self.running_reflector_uploads else
223	self.running_reflector_uploads.pop(stream.sd_hash)
224	)
225	return task	1✔
226
227	@staticmethod	1✔
228	async def _retriable_reflect_stream(stream, host, port):	1✔
229	sent = await stream.upload_to_reflector(host, port)	1✔
230	while not stream.is_fully_reflected and stream.reflector_progress > 0 and len(sent) > 0:	1!
231	stream.reflector_progress = 0	×
232	sent = await stream.upload_to_reflector(host, port)	×
233	return sent	1✔
234
235	async def create(self, file_path: str, key: Optional[bytes] = None,	1✔
236	iv_generator: Optional[typing.Generator[bytes, None, None]] = None) -> ManagedStream:
237	descriptor = await StreamDescriptor.create_stream(	1✔
238	self.loop, self.blob_manager.blob_dir, file_path, key=key, iv_generator=iv_generator,
239	blob_completed_callback=self.blob_manager.blob_completed
240	)
241	await self.storage.store_stream(	1✔
242	self.blob_manager.get_blob(descriptor.sd_hash, is_mine=True), descriptor
243	)
244	row_id = await self.storage.save_published_file(	1✔
245	descriptor.stream_hash, os.path.basename(file_path), os.path.dirname(file_path), 0
246	)
247	stream = ManagedStream(	1✔
248	self.loop, self.config, self.blob_manager, descriptor.sd_hash, os.path.dirname(file_path),
249	os.path.basename(file_path), status=ManagedDownloadSource.STATUS_FINISHED,
250	rowid=row_id, descriptor=descriptor
251	)
252	self.streams[stream.sd_hash] = stream	1✔
253	self.storage.content_claim_callbacks[stream.stream_hash] = lambda: self._update_content_claim(stream)	1!
254	if self.config.reflect_streams and self.config.reflector_servers:	1!
255	self.reflect_stream(stream)	×
256	return stream	1✔
257
258	async def delete(self, source: ManagedDownloadSource, delete_file: Optional[bool] = False):	1✔
259	if not isinstance(source, ManagedStream):	1!
260	return	×
261	if source.identifier in self.running_reflector_uploads:	1!
262	self.running_reflector_uploads[source.identifier].cancel()	×
263	await source.stop_tasks()	1✔
264	if source.identifier in self.streams:	1!
265	del self.streams[source.identifier]	1✔
266	blob_hashes = [source.identifier] + [b.blob_hash for b in source.descriptor.blobs[:-1]]	1✔
267	await self.blob_manager.delete_blobs(blob_hashes, delete_from_db=False)	1✔
268	await self.storage.delete_stream(source.descriptor)	1✔
269	if delete_file and source.output_file_exists:	1!
270	os.remove(source.full_path)	1✔
271
272	async def stream_partial_content(self, request: Request, sd_hash: str):	1✔
273	stream = self._sources[sd_hash]	×
274	if not stream.downloader.node:	×
275	stream.downloader.node = self.node	×
276	return await stream.stream_file(request)	×

lbryio / lbry-sdk / 4599645360

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous