wip repair

Author: odesenfans

src/aleph/ccn_cli/commands/garbage_collector.py

@@ -5,57 +5,23 @@
 """
 import asyncio
 import datetime as dt
-from typing import Any, Dict, FrozenSet, List, Optional
+from typing import Dict, FrozenSet
 from typing import cast
 
 import pytz
 import typer
-from aleph_message.models import MessageType
 from configmanager import Config
 
 import aleph.model
 from aleph.ccn_cli.cli_config import CliConfig
 from aleph.config import get_defaults
 from aleph.model import init_db_globals
-from aleph.model.filepin import PermanentPin
 from aleph.model.hashes import delete_value as delete_gridfs_file
-from aleph.model.messages import Message
+from .toolkit.local_storage import list_expected_local_files
 
 gc_ns = typer.Typer()
 
 
-async def get_hashes(
-    item_type_field: str, item_hash_field: str, msg_type: Optional[MessageType] = None
-) -> FrozenSet[str]:
-    def rgetitem(dictionary: Any, fields: List[str]) -> Any:
-        value = dictionary[fields[0]]
-        if len(fields) > 1:
-            return rgetitem(value, fields[1:])
-        return value
-
-    filters = {
-        # Check if the hash field exists in case the message was forgotten
-        item_hash_field: {"$exists": 1},
-        item_type_field: {"$in": ["ipfs", "storage"]},
-    }
-    if msg_type:
-        filters["type"] = msg_type
-
-    hashes = [
-        rgetitem(msg, item_hash_field.split("."))
-        async for msg in Message.collection.find(
-            filters,
-            {item_hash_field: 1},
-            batch_size=1000,
-        )
-    ]
-
-    # Temporary fix for api2. A message has a list of dicts as item hash.
-    hashes = [h for h in hashes if isinstance(h, str)]
-
-    return frozenset(hashes)
-
-
 def print_files_to_preserve(files_to_preserve: Dict[str, FrozenSet[str]]) -> None:
     typer.echo("The following files will be preserved:")
     for file_type, files in files_to_preserve.items():
@@ -79,26 +45,8 @@ async def list_files_to_preserve(
         ]
     )
 
-    # Get all the messages that potentially store data in local storage:
-    # * any message with item_type in ["storage", "ipfs"]
-    # * STOREs with content.item_type in ["storage", "ipfs"]
-    files_to_preserve_dict["non-inline messages"] = await get_hashes(
-        item_type_field="item_type",
-        item_hash_field="item_hash",
-    )
-    files_to_preserve_dict["stores"] = await get_hashes(
-        item_type_field="content.item_type",
-        item_hash_field="content.item_hash",
-        msg_type=MessageType.store,
-    )
-
-    # We also keep permanent pins, even if they are also stored on IPFS
-    files_to_preserve_dict["file pins"] = frozenset(
-        [
-            pin["multihash"]
-            async for pin in PermanentPin.collection.find({}, {"multihash": 1})
-        ]
-    )
+    expected_local_files = await list_expected_local_files()
+    files_to_preserve_dict.update(expected_local_files)
 
     return files_to_preserve_dict
 

src/aleph/ccn_cli/commands/repair.py

@@ -0,0 +1,69 @@
+"""
+Repairs the local CCN by checking the following
+"""
+import asyncio
+import itertools
+from typing import Dict, FrozenSet
+from typing import cast
+
+import typer
+from configmanager import Config
+
+import aleph.model
+from aleph.ccn_cli.cli_config import CliConfig
+from aleph.config import get_defaults
+from aleph.model import init_db_globals
+from .toolkit.local_storage import list_expected_local_files
+
+repair_ns = typer.Typer()
+
+
+
+def print_files_to_preserve(files_to_preserve: Dict[str, FrozenSet[str]]) -> None:
+    typer.echo("The following files will be preserved:")
+    for file_type, files in files_to_preserve.items():
+        typer.echo(f"* {len(files)} {file_type}")
+
+
+async def list_missing_files() -> FrozenSet[str]:
+    # Get a set of all the files currently in GridFS
+    gridfs_files_dict = {
+        file["filename"]: file
+        async for file in aleph.model.db["fs.files"].find(
+            projection={"_id": 0, "filename": 1, "length": 1, "uploadDate": 1},
+            batch_size=1000,
+        )
+    }
+
+    gridfs_files = frozenset(gridfs_files_dict.keys())
+    typer.echo(f"Found {len(gridfs_files_dict)} files in local storage.")
+
+    expected_local_files_dict = await list_expected_local_files()
+    expected_local_files = frozenset(itertools.chain.from_iterable(expected_local_files_dict.values()))
+
+    missing_files = expected_local_files - gridfs_files
+    return missing_files
+
+
+async def fetch_missing_files():
+    missing_files = await list_missing_files()
+    typer.echo(f"Found {len(missing_files)} missing files.")
+
+
+async def run(ctx: typer.Context):
+    config = Config(schema=get_defaults())
+    cli_config = cast(CliConfig, ctx.obj)
+    config.yaml.load(str(cli_config.config_file_path))
+
+    init_db_globals(config=config)
+    if aleph.model.db is None:  # for mypy
+        raise ValueError("DB not initialized as expected.")
+
+    await fetch_missing_files()
+
+    typer.echo("Done.")
+
+
+@repair_ns.command(name="run")
+def run_repair(ctx: typer.Context):
+    asyncio.run(run(ctx))

src/aleph/ccn_cli/commands/toolkit/__init__.py

@@ -0,0 +1,68 @@
+from typing import Any, Dict, FrozenSet, List, Optional
+from aleph.model.messages import Message
+from aleph.model.filepin import PermanentPin
+from aleph_message.models import MessageType
+
+
+async def get_hashes(
+    item_type_field: str, item_hash_field: str, msg_type: Optional[MessageType] = None
+) -> FrozenSet[str]:
+    def rgetitem(dictionary: Any, fields: List[str]) -> Any:
+        value = dictionary[fields[0]]
+        if len(fields) > 1:
+            return rgetitem(value, fields[1:])
+        return value
+
+    filters = {
+        # Check if the hash field exists in case the message was forgotten
+        item_hash_field: {"$exists": 1},
+        item_type_field: {"$in": ["ipfs", "storage"]},
+    }
+    if msg_type:
+        filters["type"] = msg_type
+
+    hashes = [
+        rgetitem(msg, item_hash_field.split("."))
+        async for msg in Message.collection.find(
+            filters,
+            {item_hash_field: 1},
+            batch_size=1000,
+        )
+    ]
+
+    # Temporary fix for api2. A message has a list of dicts as item hash.
+    hashes = [h for h in hashes if isinstance(h, str)]
+
+    return frozenset(hashes)
+
+
+async def list_expected_local_files() -> Dict[str, FrozenSet[str]]:
+    """
+    Lists the files that are expected to be found on the local storage of the CCN.
+    This includes:
+    * the content of any message with item_type in ["storage", "ipfs"]
+    * the stored content of any STORE message with content.item_type in ["storage", "ipfs"]
+    * file pins
+    """
+
+    expected_files = {}
+
+    expected_files["non-inline messages"] = await get_hashes(
+        item_type_field="item_type",
+        item_hash_field="item_hash",
+    )
+    expected_files["stores"] = await get_hashes(
+        item_type_field="content.item_type",
+        item_hash_field="content.item_hash",
+        msg_type=MessageType.store,
+    )
+
+    # We also keep permanent pins, even if they are also stored on IPFS
+    expected_files["file pins"] = frozenset(
+        [
+            pin["multihash"]
+            async for pin in PermanentPin.collection.find({}, {"multihash": 1})
+        ]
+    )
+
+    return expected_files