diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 4105973f6..2a8103924 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -31,8 +31,8 @@ jobs: matrix: os: - ubuntu-latest - #- macos-latest - #- windows-latest + - macos-latest + - windows-latest runs-on: ${{ matrix.os }} steps: - uses: actions/checkout@v2 @@ -49,10 +49,18 @@ jobs: key: ${{ runner.os }}-pip-${{ hashFiles('setup.py') }} restore-keys: ${{ runner.os }}-pip- - run: pip install --user --upgrade pip wheel - - run: pip install -e .[torrent,test] - - env: + - if: startsWith(runner.os, 'linux') + run: pip install -e .[torrent,test] + - if: startsWith(runner.os, 'linux') + env: HOME: /tmp run: make test-unit-coverage + - if: startsWith(runner.os, 'linux') != true + run: pip install -e .[test] + - if: startsWith(runner.os, 'linux') != true + env: + HOME: /tmp + run: python -m unittest tests/unit/test_conf.py tests-integration: name: "tests / integration" diff --git a/lbry/blob/disk_space_manager.py b/lbry/blob/disk_space_manager.py new file mode 100644 index 000000000..58e8b19bc --- /dev/null +++ b/lbry/blob/disk_space_manager.py @@ -0,0 +1,64 @@ +import os +import asyncio +import logging + +log = logging.getLogger(__name__) + + +class DiskSpaceManager: + + def __init__(self, config, cleaning_interval=30 * 60): + self.config = config + self.cleaning_interval = cleaning_interval + self.running = False + self.task = None + + @property + def space_used_bytes(self): + used = 0 + data_dir = os.path.join(self.config.data_dir, 'blobfiles') + for item in os.scandir(data_dir): + if item.is_file: + used += item.stat().st_size + return used + + @property + def space_used_mb(self): + return int(self.space_used_bytes/1024.0/1024.0) + + def clean(self): + if not self.config.blob_storage_limit: + return 0 + used = 0 + files = [] + data_dir = os.path.join(self.config.data_dir, 'blobfiles') + for file in os.scandir(data_dir): + if file.is_file: + file_stats = file.stat() + used += file_stats.st_size + files.append((file_stats.st_mtime, file_stats.st_size, file.path)) + files.sort() + available = (self.config.blob_storage_limit*1024*1024) - used + cleaned = 0 + for _, file_size, file in files: + available += file_size + if available > 0: + break + os.remove(file) + cleaned += 1 + return cleaned + + async def cleaning_loop(self): + while self.running: + await asyncio.get_event_loop().run_in_executor(None, self.clean) + await asyncio.sleep(self.cleaning_interval) + + async def start(self): + self.running = True + self.task = asyncio.create_task(self.cleaning_loop()) + self.task.add_done_callback(lambda _: log.info("Stopping blob cleanup service.")) + + async def stop(self): + if self.running: + self.running = False + self.task.cancel() diff --git a/lbry/conf.py b/lbry/conf.py index 4250fe579..480be87ab 100644 --- a/lbry/conf.py +++ b/lbry/conf.py @@ -634,6 +634,7 @@ class Config(CLIConfig): # blob announcement and download save_blobs = Toggle("Save encrypted blob files for hosting, otherwise download blobs to memory only.", True) + blob_storage_limit = Integer("Disk space in MB to be allocated for blob storage. 0 = no limit", 0) blob_lru_cache_size = Integer( "LRU cache size for decrypted downloaded blobs used to minimize re-downloading the same blobs when " "replying to a range request. Set to 0 to disable.", 32 diff --git a/lbry/extras/daemon/components.py b/lbry/extras/daemon/components.py index 822022f32..a383b9ea5 100644 --- a/lbry/extras/daemon/components.py +++ b/lbry/extras/daemon/components.py @@ -15,6 +15,7 @@ from lbry.dht.node import Node from lbry.dht.peer import is_valid_public_ipv4 from lbry.dht.blob_announcer import BlobAnnouncer from lbry.blob.blob_manager import BlobManager +from lbry.blob.disk_space_manager import DiskSpaceManager from lbry.blob_exchange.server import BlobServer from lbry.stream.stream_manager import StreamManager from lbry.file.file_manager import FileManager @@ -40,6 +41,7 @@ WALLET_SERVER_PAYMENTS_COMPONENT = "wallet_server_payments" DHT_COMPONENT = "dht" HASH_ANNOUNCER_COMPONENT = "hash_announcer" FILE_MANAGER_COMPONENT = "file_manager" +DISK_SPACE_COMPONENT = "disk_space" PEER_PROTOCOL_SERVER_COMPONENT = "peer_protocol_server" UPNP_COMPONENT = "upnp" EXCHANGE_RATE_MANAGER_COMPONENT = "exchange_rate_manager" @@ -375,6 +377,30 @@ class FileManagerComponent(Component): self.file_manager.stop() +class DiskSpaceComponent(Component): + component_name = DISK_SPACE_COMPONENT + + def __init__(self, component_manager): + super().__init__(component_manager) + self.disk_space_manager = DiskSpaceManager(self.conf) + + @property + def component(self) -> typing.Optional[DiskSpaceManager]: + return self.disk_space_manager + + async def get_status(self): + return { + 'space_used': str(self.disk_space_manager.space_used_mb), + 'running': self.disk_space_manager.running, + } + + async def start(self): + await self.disk_space_manager.start() + + async def stop(self): + await self.disk_space_manager.stop() + + class TorrentComponent(Component): component_name = LIBTORRENT_COMPONENT diff --git a/lbry/extras/daemon/daemon.py b/lbry/extras/daemon/daemon.py index 4443c209f..814f4e50e 100644 --- a/lbry/extras/daemon/daemon.py +++ b/lbry/extras/daemon/daemon.py @@ -41,7 +41,7 @@ from lbry.error import ( from lbry.extras import system_info from lbry.extras.daemon import analytics from lbry.extras.daemon.components import WALLET_COMPONENT, DATABASE_COMPONENT, DHT_COMPONENT, BLOB_COMPONENT -from lbry.extras.daemon.components import FILE_MANAGER_COMPONENT +from lbry.extras.daemon.components import FILE_MANAGER_COMPONENT, DISK_SPACE_COMPONENT from lbry.extras.daemon.components import EXCHANGE_RATE_MANAGER_COMPONENT, UPNP_COMPONENT from lbry.extras.daemon.componentmanager import RequiredCondition from lbry.extras.daemon.componentmanager import ComponentManager @@ -56,7 +56,7 @@ from lbry.schema.url import URL if typing.TYPE_CHECKING: from lbry.blob.blob_manager import BlobManager from lbry.dht.node import Node - from lbry.extras.daemon.components import UPnPComponent + from lbry.extras.daemon.components import UPnPComponent, DiskSpaceManager from lbry.extras.daemon.exchange_rate_manager import ExchangeRateManager from lbry.extras.daemon.storage import SQLiteStorage from lbry.wallet import WalletManager, Ledger @@ -375,6 +375,10 @@ class Daemon(metaclass=JSONRPCServerType): def blob_manager(self) -> typing.Optional['BlobManager']: return self.component_manager.get_component(BLOB_COMPONENT) + @property + def disk_space_manager(self) -> typing.Optional['DiskSpaceManager']: + return self.component_manager.get_component(DISK_SPACE_COMPONENT) + @property def upnp(self) -> typing.Optional['UPnPComponent']: return self.component_manager.get_component(UPNP_COMPONENT) @@ -4916,6 +4920,22 @@ class Daemon(metaclass=JSONRPCServerType): raise NotImplementedError() + @requires(DISK_SPACE_COMPONENT) + async def jsonrpc_blob_clean(self): + """ + Deletes blobs to cleanup disk space + + Usage: + blob_clean + + Options: + None + + Returns: + (bool) true if successful + """ + return self.disk_space_manager.clean() + @requires(FILE_MANAGER_COMPONENT) async def jsonrpc_file_reflect(self, **kwargs): """ diff --git a/lbry/testcase.py b/lbry/testcase.py index 9b65e2fe8..210b99b3c 100644 --- a/lbry/testcase.py +++ b/lbry/testcase.py @@ -607,6 +607,12 @@ class CommandTestCase(IntegrationTestCase): await asyncio.wait([self.ledger.wait(tx, self.blockchain.block_expected) for tx in txs]) return self.sout(txs) + async def blob_clean(self): + return await self.out(self.daemon.jsonrpc_blob_clean()) + + async def status(self): + return await self.out(self.daemon.jsonrpc_status()) + async def resolve(self, uri, **kwargs): return (await self.out(self.daemon.jsonrpc_resolve(uri, **kwargs)))[uri] diff --git a/tests/integration/datanetwork/test_file_commands.py b/tests/integration/datanetwork/test_file_commands.py index ee65c755b..dccaca324 100644 --- a/tests/integration/datanetwork/test_file_commands.py +++ b/tests/integration/datanetwork/test_file_commands.py @@ -511,3 +511,20 @@ class FileCommands(CommandTestCase): await tx.sign([self.account]) await self.broadcast(tx) await self.confirm_tx(tx.id) + + +class DiskSpaceManagement(CommandTestCase): + + async def test_file_management(self): + status = await self.status() + self.assertIn('disk_space', status) + self.assertEqual('0', status['disk_space']['space_used']) + self.assertEqual(True, status['disk_space']['running']) + await self.stream_create('foo1', '0.01', data=('0' * 3 * 1024 * 1024).encode()) + await self.stream_create('foo2', '0.01', data=('0' * 2 * 1024 * 1024).encode()) + self.assertEqual('5', (await self.status())['disk_space']['space_used']) + await self.blob_clean() + self.assertEqual('5', (await self.status())['disk_space']['space_used']) + self.daemon.conf.blob_storage_limit = 3 + await self.blob_clean() + self.assertEqual('3', (await self.status())['disk_space']['space_used']) diff --git a/tests/unit/blob/test_disk_space_manager.py b/tests/unit/blob/test_disk_space_manager.py new file mode 100644 index 000000000..54871f092 --- /dev/null +++ b/tests/unit/blob/test_disk_space_manager.py @@ -0,0 +1,30 @@ +import os +import unittest +import tempfile + +import lbry.wallet +from lbry.conf import Config +from lbry.blob.disk_space_manager import DiskSpaceManager + + +class ConfigurationTests(unittest.TestCase): + + def test_space_management(self): + with tempfile.TemporaryDirectory() as temp_dir: + os.mkdir(os.path.join(temp_dir, 'blobfiles')) + config = Config( + blob_storage_limit=5, + data_dir=temp_dir, + wallet_dir=temp_dir, + config=os.path.join(temp_dir, 'settings.yml'), + ) + dsm = DiskSpaceManager(config) + self.assertEqual(0, dsm.space_used_mb) + for file_no in range(10): + with open(os.path.join(config.data_dir, 'blobfiles', f'3mb-{file_no}'), 'w') as blob: + blob.write('0' * 1 * 1024 * 1024) + self.assertEqual(10, dsm.space_used_mb) + self.assertTrue(dsm.clean()) + self.assertEqual(5, dsm.space_used_mb) + self.assertFalse(dsm.clean()) + diff --git a/tests/unit/components/test_component_manager.py b/tests/unit/components/test_component_manager.py index 48d844168..f539f033e 100644 --- a/tests/unit/components/test_component_manager.py +++ b/tests/unit/components/test_component_manager.py @@ -15,6 +15,7 @@ class TestComponentManager(AsyncioTestCase): self.default_components_sort = [ [ components.DatabaseComponent, + components.DiskSpaceComponent, components.ExchangeRateManagerComponent, components.TorrentComponent, components.UPnPComponent diff --git a/tests/unit/test_conf.py b/tests/unit/test_conf.py index e241dea5f..645406462 100644 --- a/tests/unit/test_conf.py +++ b/tests/unit/test_conf.py @@ -21,6 +21,27 @@ class TestConfig(BaseConfig): class ConfigurationTests(unittest.TestCase): + @unittest.skipIf('darwin' not in sys.platform, 'skipping mac only test') + def test_mac_defaults(self): + c = Config() + self.assertEqual(c.data_dir, os.path.expanduser("~/Library/Application Support/LBRY")) + self.assertEqual(c.wallet_dir, os.path.expanduser('~/.lbryum')) + self.assertEqual(c.download_dir, os.path.expanduser('~/Downloads')) + self.assertEqual(c.config, os.path.join(c.data_dir, 'daemon_settings.yml')) + self.assertEqual(c.api_connection_url, 'http://localhost:5279/lbryapi') + self.assertEqual(c.log_file_path, os.path.join(c.data_dir, 'lbrynet.log')) + + @unittest.skipIf('win32' not in sys.platform, 'skipping windows only test') + def test_windows_defaults(self): + c = Config() + prefix = os.path.join(r"C:\Users", os.getlogin(), r"AppData\Local\lbry") + self.assertEqual(c.data_dir, os.path.join(prefix, 'lbrynet')) + self.assertEqual(c.wallet_dir, os.path.join(prefix, 'lbryum')) + self.assertEqual(c.download_dir, os.path.join(r"C:\Users", os.getlogin(), "Downloads")) + self.assertEqual(c.config, os.path.join(c.data_dir, 'daemon_settings.yml')) + self.assertEqual(c.api_connection_url, 'http://localhost:5279/lbryapi') + self.assertEqual(c.log_file_path, os.path.join(c.data_dir, 'lbrynet.log')) + @unittest.skipIf('linux' not in sys.platform, 'skipping linux only test') def test_linux_defaults(self): c = Config()