diff --git a/cloudfile_ext/v2/__main__.py b/cloudfile_ext/v2/__main__.py index 7cd52b8ec8f..50666ff5ebb 100644 --- a/cloudfile_ext/v2/__main__.py +++ b/cloudfile_ext/v2/__main__.py @@ -1,4 +1,4 @@ -"""Explicit fresh install/check entry point; secrets are never CLI arguments.""" +"""Explicit install/check and storage preflight; secrets are never CLI arguments.""" import argparse import json import os @@ -7,32 +7,57 @@ from .schema import FreshSchema from .native_discovery import NativeLibraryDiscovery +from .storage_reconcile import StorageConfig, StorageReconciler -def main(): +def main(argv=None): parser = argparse.ArgumentParser() - parser.add_argument('command', choices=('install', 'check', 'discover', 'reconcile', 'inspect')) + parser.add_argument('command', choices=('install', 'check', 'discover', 'reconcile', 'inspect', + 'storage-reconcile')) parser.add_argument('--repo-id', help='original native Seafile repository UUID') - parser.add_argument('--ddl', type=Path, required=True) - args = parser.parse_args() + parser.add_argument('--ddl', type=Path) + parser.add_argument('--seafile-conf', type=Path) + parser.add_argument('--seafile-data', type=Path) + parser.add_argument('--batch-size', type=int, default=500) + modes = parser.add_mutually_exclusive_group() + modes.add_argument('--dry-run', action='store_true') + modes.add_argument('--apply', action='store_true') + parser.add_argument('--normalize-native-class', action='append', default=[], metavar='ID') + args = parser.parse_args(argv) + storage = args.command == 'storage-reconcile' + if storage and (not args.seafile_conf or not args.seafile_data): + parser.error('storage-reconcile requires --seafile-conf and --seafile-data') + if not storage and (args.dry_run or args.apply or args.normalize_native_class): + parser.error('storage mode flags require storage-reconcile') + if not storage and not args.ddl: + parser.error('--ddl is required for schema commands') + if not 1 <= args.batch_size <= 2000: + parser.error('--batch-size must be between 1 and 2000') import pymysql try: + config = StorageConfig.detect(args.seafile_conf, args.seafile_data) with pymysql.connect(host=os.environ['CF_V2_DB_HOST'], port=int(os.environ.get('CF_V2_DB_PORT', '3306')), user=os.environ['CF_V2_DB_USER'], password=os.environ['CF_V2_DB_PASSWORD'], database=os.environ['CF_V2_DB_NAME'], charset='utf8mb4', autocommit=True) as connection: + if storage: + discovery = StorageReconciler(connection, config) + report = discovery.run(apply=args.apply, batch_size=args.batch_size, + normalize=args.normalize_native_class, repo_id=args.repo_id) + print(json.dumps(report, sort_keys=True)) + return 2 if report['summary'].get('BLOCKED', 0) else 0 NativeLibraryDiscovery.check_native_schema(connection) schema = FreshSchema(connection, args.ddl.read_text()) if args.command == 'install': schema.install() else: schema.verify() - discovery = NativeLibraryDiscovery(connection) + discovery = NativeLibraryDiscovery(connection, config) if args.command in ('install', 'reconcile'): - print(json.dumps({"statuses": discovery.reconcile(args.repo_id), - "unresolved_registry": discovery.unresolved_registry_count()}, sort_keys=True)) + print(json.dumps({'statuses': discovery.reconcile(args.repo_id), + 'unresolved_registry': discovery.unresolved_registry_count()}, sort_keys=True)) elif args.command == 'inspect': if not args.repo_id: parser.error('inspect requires --repo-id') @@ -40,6 +65,12 @@ def main(): elif args.command == 'discover': print(json.dumps(discovery.summary(args.repo_id), sort_keys=True)) except Exception as error: + if storage: + reason = str(error) if isinstance(error, (ValueError, RuntimeError)) else type(error).__name__ + report = dict(mode='apply' if args.apply else 'dry-run', summary={'BLOCKED': 1}, + items=[{'status': 'BLOCKED', 'reason': reason}], orphan_directories=[]) + print(json.dumps(report, sort_keys=True)) + return 2 if isinstance(error, ValueError) else 1 print('V2 schema unavailable: ' + type(error).__name__, file=sys.stderr) return 1 print('V2 schema ready: 3 reviewed extension tables, native discovery checked') diff --git a/cloudfile_ext/v2/native_discovery.py b/cloudfile_ext/v2/native_discovery.py index 18269f75282..c91b98942bf 100644 --- a/cloudfile_ext/v2/native_discovery.py +++ b/cloudfile_ext/v2/native_discovery.py @@ -45,8 +45,9 @@ class NativeLibraryDiscovery: database: the actual Seafile Repo/Commit/FS/Block facts must first exist. """ - def __init__(self, connection): + def __init__(self, connection, config=None): self.connection = connection + self.config = config @staticmethod def check_native_schema(connection): @@ -101,35 +102,14 @@ def scan(self, batch_size=500): last = rows[-1][0] def register_native(self, repo_id): - # This INSERT SELECT explicitly rechecks the native Repo and route in - # the same statement, and a concurrent worker cannot overwrite rows. - before = self.inspect(repo_id) - if before["status"] != "NATIVE_UNREGISTERED": - return before - try: - with self.connection.cursor() as cursor: - cursor.execute( - "INSERT INTO cf_library_storage(repo_key,repo_id,state) " - "SELECT %s,r.repo_id,'active' FROM Repo r " - "LEFT JOIN RepoStorageId s ON s.repo_id=r.repo_id " - "LEFT JOIN VirtualRepo v ON v.repo_id=r.repo_id " - "WHERE r.repo_id=%s AND s.repo_id IS NULL AND v.repo_id IS NULL", - (native_key(repo_id), repo_id)) - except Exception as error: - # Concurrent reconciliations can hit the unique key; another - # worker's existing row is never changed or silently overwritten. - if not getattr(error, "args", ()) or error.args[0] != 1062: - raise - return self.inspect(repo_id) + # Preserve the public helper while using the same guarded lifecycle, + # native routing and per-library transaction as storage-reconcile. + from .storage_reconcile import StorageReconciler + return StorageReconciler(self.connection, self.config).apply_one(repo_id) def reconcile(self, repo_id=None): - counts = Counter() - iterator = (self.inspect(repo_id),) if repo_id else self.scan() - for item in iterator: - if item["status"] == "NATIVE_UNREGISTERED": - item = self.register_native(item["repo_id"]) - counts[item["status"]] += 1 - return dict(sorted(counts.items())) + from .storage_reconcile import StorageReconciler + return StorageReconciler(self.connection, self.config).run(apply=True, repo_id=repo_id)['summary'] def unresolved_registry_count(self): """Orphan diagnostics only: never retire or delete a record automatically. @@ -184,71 +164,28 @@ def storage_catalog_status(repo_id, repo_present, key, state, storage_id, readab return "READY" if readable else "RESOURCE_UNREADABLE" def reconcile_from_hub(repo_id=None, *, batch_size=500): - """Idempotent native Repo -> cf_library_storage reconciliation from Seahub. + """Use native reconciliation rules from the current Seahub connection. - This uses the current Seafile schema through Django's existing connection, - not a second set of DB credentials. Etech and the native create endpoint - share the same guarded insertion, and existing V2 auto-local mappings win. + Startup/admin discovery handles native and auto-local bindings only. + Pinned classes and invalid bindings are reported without conversion. """ - from django.db import connection, IntegrityError + from django.db import connection, transaction from seahub.utils.db_api import SeafileDB - from .schema import identifier - - schema = identifier(SeafileDB().db_name) - repo = schema + '.Repo' - routes = schema + '.RepoStorageId' - virtual = schema + '.VirtualRepo' - registry = schema + '.cf_library_storage' - if repo_id is not None and native_key(repo_id) is None: - raise ValueError('Invalid canonical Repo UUID') - added = 0 - already = 0 - after = '' - while True: - with connection.cursor() as cursor: - if repo_id: - cursor.execute( - 'SELECT r.repo_id FROM ' + repo + ' r ' - 'LEFT JOIN ' + virtual + ' v ON v.repo_id=r.repo_id ' - 'WHERE r.repo_id=%s AND v.repo_id IS NULL', [repo_id]) - else: - cursor.execute( - 'SELECT r.repo_id FROM ' + repo + ' r ' - 'LEFT JOIN ' + virtual + ' v ON v.repo_id=r.repo_id ' - 'WHERE r.repo_id>%s AND v.repo_id IS NULL ' - 'ORDER BY r.repo_id LIMIT %s', [after, batch_size]) - ids = [row[0] for row in cursor.fetchall()] - if not ids: - break - for native_id in ids: - if native_key(native_id) is None: - continue - # The query decides the route at write time. New routed repos, - # tombstones, and existing business bindings are NEVER changed. - with connection.cursor() as cursor: - try: - cursor.execute( - 'INSERT INTO ' + registry + ' (repo_key,repo_id,state) ' - "SELECT %s,r.repo_id,'active' FROM " + repo + ' r ' - 'LEFT JOIN ' + routes + ' s ON s.repo_id=r.repo_id ' - 'LEFT JOIN ' + virtual + ' v ON v.repo_id=r.repo_id ' - 'LEFT JOIN ' + registry + ' c ON c.repo_id=r.repo_id ' - 'WHERE r.repo_id=%s AND s.repo_id IS NULL ' - 'AND v.repo_id IS NULL AND c.repo_id IS NULL', - ['native-' + native_id, native_id]) - added += cursor.rowcount - except IntegrityError as exc: - if getattr(exc, 'args', (None,))[0] != 1062: - raise - # Another instance inserted the same UUID concurrently. - cursor.execute('SELECT 1 FROM ' + registry + ' WHERE repo_id=%s', [native_id]) - if cursor.fetchone() is None: - raise - already += 1 - after = ids[-1] - if repo_id or len(ids) < batch_size: - break - return {'registered': added, 'already': already} + from .storage_reconcile import StorageConfig, StorageReconciler + + config = StorageConfig.detect() + if config is None: + raise RuntimeError('SEAFILE_STORAGE_CONFIGURATION_UNAVAILABLE') + discovery = StorageReconciler(connection, config, schema=SeafileDB().db_name, + atomic=transaction.atomic) + report = discovery.run(apply=True, repo_id=repo_id, batch_size=batch_size) + import logging + for item in report['items'] + report['orphan_directories']: + if item['status'] in ('BLOCKED', 'NATIVE_CLASS_PINNED', 'ORPHAN_DIRECTORY', 'STATE_DRIFT'): + logging.getLogger(__name__).warning('Storage reconciliation: %s', item) + report['registered'] = sum(item.get('action') == 'REGISTERED' for item in report['items']) + report['already'] = report['summary'].get('REGISTERED', 0) - report['registered'] + return report def inspect_from_hub(repo_ids): diff --git a/cloudfile_ext/v2/storage_reconcile.py b/cloudfile_ext/v2/storage_reconcile.py new file mode 100644 index 00000000000..796317f4970 --- /dev/null +++ b/cloudfile_ext/v2/storage_reconcile.py @@ -0,0 +1,374 @@ +"""Reconcile native storage facts without moving bytes or reading legacy metadata.""" +from collections import Counter +from configparser import ConfigParser +from contextlib import contextmanager +import json +import os +from pathlib import Path +import re + +from .schema import identifier + +STORAGE_DDL = """CREATE TABLE cf_library_storage ( + repo_key VARCHAR(80) CHARACTER SET ascii COLLATE ascii_bin NOT NULL, + repo_id CHAR(36) CHARACTER SET ascii COLLATE ascii_bin NOT NULL, + state VARCHAR(8) CHARACTER SET ascii COLLATE ascii_bin NOT NULL, + PRIMARY KEY (repo_key), + UNIQUE KEY library_storage_repo (repo_id), + CONSTRAINT library_storage_state CHECK (state IN ('reserved','active','retired')) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin""" +KINDS = ('commits', 'fs', 'blocks') + + +def valid_key(key): + return isinstance(key, str) and bool(re.fullmatch(r'[a-z0-9](?:[a-z0-9-]{0,78}[a-z0-9])?', key)) + + +def real_directory(path): + return path.is_dir() and not path.is_symlink() + + +def symlink_in_path(path): + """Reject managed-root aliases, including any ancestor component.""" + return any(component.is_symlink() for component in (path, *path.parents)) + + +class StorageConfig: + def __init__(self, root=None, native_dir=None, classes=()): + self.root = Path(root) if root else None + self.native_dir = Path(native_dir) if native_dir else None + self.classes = {} + for item in classes: + if not isinstance(item, dict): + raise ValueError('INVALID_STORAGE_CLASSES') + sid = item.get('storage_id') + if not isinstance(sid, str) or sid in self.classes: + raise ValueError('INVALID_STORAGE_CLASSES') + self.classes[sid] = item + + @classmethod + def read(cls, conf, native_dir): + conf = Path(conf) + parser = ConfigParser(interpolation=None) + with conf.open() as stream: + parser.read_file(stream) + classes = [] + # Read configured JSON even during an explicit transition from multi. + filename = parser.get('storage', 'storage_classes_file', fallback='') + inline = os.environ.get('CF_STORAGE_CLASSES_JSON', '') + if inline: + classes = json.loads(inline) + elif filename: + path = Path(filename) + if not path.is_absolute(): + path = conf.parent / path + classes = json.loads(path.read_text()) + if not isinstance(classes, list) or any(not isinstance(item, dict) for item in classes): + raise ValueError('INVALID_STORAGE_CLASSES') + root = parser.get('cloudfile_v2', 'auto_local_root', fallback='') + if not root: + templates = [item for item in classes if item.get('storage_id') == 'auto-local'] + if templates: + template = templates[0] + roots = {template.get(kind, {}).get('dir') for kind in KINDS} + if (len(roots) != 1 or None in roots or + any(template.get(kind, {}).get('backend') != 'fs' for kind in KINDS)): + raise ValueError('INVALID_AUTO_LOCAL_TEMPLATE') + root = roots.pop() + config = cls(root, native_dir, classes) + if config.root and (not config.root.is_absolute() or symlink_in_path(config.root) or not real_directory(config.root)): + raise ValueError('INVALID_STORAGE_ROOT') + return config + + @classmethod + def detect(cls, conf=None, native_dir=None): + """Use existing Seafile layout/environment for legacy entry points.""" + native_dir = native_dir or os.environ.get('SEAFILE_DATA_DIR') + if not native_dir: + native_dir = next((path for path in ('/shared/seafile/seafile-data', '/opt/seafile/seafile-data') + if Path(path).is_dir()), None) + if not native_dir: + return None + if conf: + return cls.read(conf, native_dir) + central = os.environ.get('SEAFILE_CENTRAL_CONF_DIR') + candidates = ([Path(central) / 'seafile.conf'] if central else []) + [ + Path(native_dir).parent / 'conf' / 'seafile.conf', + Path(native_dir) / 'seafile.conf', + Path('/shared/seafile/conf/seafile.conf'), Path('/opt/seafile/conf/seafile.conf')] + return next((cls.read(path, native_dir) for path in candidates if path.is_file()), None) + + def native_class(self, route): + item = self.classes.get(route, {}) + if not self.native_dir: + return False + try: + native = self.native_dir.resolve(strict=True) + if not native.is_dir(): + return False + # CE's /opt/seafile/seafile-data alias is normal. Only native + # aliases are resolved; managed automatic roots reject symlinks. + return all(item.get(kind, {}).get('backend') == 'fs' and + item[kind].get('dir') and Path(item[kind]['dir']).is_absolute() and + Path(item[kind]['dir']).resolve(strict=True).is_dir() and + Path(item[kind]['dir']).resolve(strict=True) == native for kind in KINDS) + except (OSError, RuntimeError): + return False + + def auto_error(self, key, repo_id): + if not valid_key(key): + return 'INVALID_KEY' + if not self.root or symlink_in_path(self.root) or not real_directory(self.root): + return 'INVALID_STORAGE_ROOT' + base = self.root / key + storage = base / 'storage' + if base.is_symlink() or storage.is_symlink(): + return 'SYMLINK_DIRECTORY' + if not real_directory(base) or not real_directory(storage): + return 'MISSING_STORAGE_DIRECTORY' + found = False + for kind in KINDS: + directory = storage / kind + objects = directory / repo_id + if directory.is_symlink() or objects.is_symlink(): + return 'SYMLINK_DIRECTORY' + if real_directory(directory) and real_directory(objects): + found = True + return None if found else 'MISSING_REPOSITORY_DIRECTORY' + + +class StorageReconciler: + def __init__(self, connection, config=None, *, schema=None, atomic=None): + self.connection = connection + self.config = config or StorageConfig() + self.schema = schema + self.prefix = identifier(schema) + '.' if schema else '' + self.atomic = atomic or self._atomic + self.registry_present = True + + def table(self, name): + return self.prefix + identifier(name) + + @contextmanager + def _atomic(self): + self.connection.begin() + try: + yield + self.connection.commit() + except Exception: + self.connection.rollback() + raise + + def ensure_schema(self, *, apply=True): + """Read the exact contract; only an explicit apply may create it. + + Both modes inspect existing metadata without disposable tables or any + writes. Missing registry metadata is an independent dry-run result. + """ + with self.connection.cursor() as cursor: + required = {'Repo', 'RepoTrash', 'GarbageRepos', 'VirtualRepo', 'RepoStorageId'} + cursor.execute("SELECT table_name FROM information_schema.tables " + "WHERE table_schema=COALESCE(%s,DATABASE()) AND table_name IN " + "('Repo','RepoTrash','GarbageRepos','VirtualRepo','RepoStorageId')", (self.schema,)) + if {row[0] for row in cursor.fetchall()} != required: + raise RuntimeError('NATIVE_SEAFILE_DATABASE_NOT_CONNECTED') + cursor.execute("SELECT engine,table_collation FROM information_schema.tables " + "WHERE table_schema=COALESCE(%s,DATABASE()) AND table_name='cf_library_storage'", + (self.schema,)) + table = cursor.fetchone() + if table is None and not apply: + self.registry_present = False + return 'SCHEMA_MISSING' + if table is None: + cursor.execute(STORAGE_DDL.replace('CREATE TABLE cf_library_storage', + 'CREATE TABLE IF NOT EXISTS ' + self.table('cf_library_storage'), 1)) + cursor.execute("SELECT engine,table_collation FROM information_schema.tables " + "WHERE table_schema=COALESCE(%s,DATABASE()) AND table_name='cf_library_storage'", + (self.schema,)) + table = cursor.fetchone() + cursor.execute("SELECT column_name,column_type,is_nullable,column_default,character_set_name," + "collation_name,extra FROM information_schema.columns " + "WHERE table_schema=COALESCE(%s,DATABASE()) AND table_name='cf_library_storage' " + "ORDER BY ordinal_position", (self.schema,)) + columns = list(cursor.fetchall()) + expected_columns = [(name, kind, 'NO', None, 'ascii', 'ascii_bin', '') for name, kind in + (('repo_key', 'varchar(80)'), ('repo_id', 'char(36)'), ('state', 'varchar(8)'))] + cursor.execute("SELECT index_name,non_unique,seq_in_index,column_name,sub_part,index_type " + "FROM information_schema.statistics WHERE table_schema=COALESCE(%s,DATABASE()) " + "AND table_name='cf_library_storage' ORDER BY index_name,seq_in_index", (self.schema,)) + indexes = sorted(cursor.fetchall(), key=lambda row: (row[0], row[2])) + expected_indexes = [('PRIMARY', 0, 1, 'repo_key', None, 'BTREE'), + ('library_storage_repo', 0, 1, 'repo_id', None, 'BTREE')] + # MariaDB hides CHECK_CONSTRAINTS from SELECT-only users even + # when SHOW CREATE TABLE exposes their table's complete contract. + cursor.execute('SHOW CREATE TABLE ' + self.table('cf_library_storage')) + definition = cursor.fetchone()[1] + checks = re.findall(r'CONSTRAINT\s+`?([A-Za-z0-9_]+)`?\s+CHECK\s+([^\n]+)', + definition, flags=re.IGNORECASE) + def clause(value): + # MySQL may add _ascii literal introducers and parentheses; + # these spell the same reviewed expression on MariaDB. + value = re.sub(r"_[a-zA-Z0-9]+(?=')", '', value) + return re.sub(r'[\s`()]+', '', value).lower() + valid_check = (len(checks) == 1 and checks[0][0] == 'library_storage_state' and + clause(checks[0][1]) == clause("state IN ('reserved','active','retired')")) + if (table != ('InnoDB', 'utf8mb4_bin') or columns != expected_columns or + indexes != expected_indexes or not valid_check or 'NOT ENFORCED' in definition.upper() or + 'FOREIGN KEY' in definition.upper()): + raise RuntimeError('STORAGE_SCHEMA_DRIFT') + self.registry_present = True + return 'SCHEMA_READY' + + def sources(self): + return ' UNION '.join('SELECT repo_id FROM ' + self.table(name) + for name in ('Repo', 'RepoTrash', 'GarbageRepos')) + + def scan(self, batch_size=500): + if not 1 <= batch_size <= 2000: + raise ValueError('batch_size out of range') + after = '' + while True: + with self.connection.cursor() as cursor: + cursor.execute('SELECT repo_id FROM (' + self.sources() + ') sources ' + 'WHERE repo_id>%s ORDER BY repo_id LIMIT %s', (after, batch_size)) + ids = [row[0] for row in cursor.fetchall()] + if not ids: + return + for repo_id in ids: + yield self.inspect(repo_id) + after = ids[-1] + + def inspect(self, repo_id, *, lock=False): + from .native_discovery import native_key + tables = {name: self.table(name) for name in + ('Repo', 'RepoTrash', 'GarbageRepos', 'VirtualRepo', 'RepoStorageId', 'cf_library_storage')} + if not self.registry_present: + tables['cf_library_storage'] = '(SELECT NULL AS repo_id,NULL AS repo_key,NULL AS state WHERE 1=0)' + sql = ('SELECT r.repo_id,t.repo_id,g.repo_id,v.repo_id,s.repo_id,s.storage_id,c.repo_key,c.state ' + 'FROM (SELECT %s AS repo_id) i ' + 'LEFT JOIN {Repo} r ON r.repo_id=i.repo_id ' + 'LEFT JOIN {RepoTrash} t ON t.repo_id=i.repo_id ' + 'LEFT JOIN {GarbageRepos} g ON g.repo_id=i.repo_id ' + 'LEFT JOIN {VirtualRepo} v ON v.repo_id=i.repo_id ' + 'LEFT JOIN {RepoStorageId} s ON s.repo_id=i.repo_id ' + 'LEFT JOIN {cf_library_storage} c ON c.repo_id=i.repo_id').format(**tables) + with self.connection.cursor() as cursor: + cursor.execute(sql + (' FOR UPDATE' if lock else ''), (repo_id,)) + present, trash, garbage, virtual, route_owner, route, key, state = cursor.fetchone() + item = dict(repo_id=repo_id, storage_id=route, status='BLOCKED') + def blocked(reason): + return dict(item, reason=reason) + if virtual: + return dict(item, status='VIRTUAL_REPO') + if not (present or trash or garbage): + return blocked('NATIVE_RESOURCE_MISSING') + if route_owner is not None and route is None: + return blocked('INVALID_STORAGE_ROUTE') + marker = native_key(repo_id) + if marker is None: + return blocked('INVALID_NATIVE_REPO_ID') + target_state = 'active' if present or trash else 'retired' + if route is None: + category, wanted = 'NATIVE', marker + elif route.startswith('auto-local:'): + category, wanted = 'AUTO_LOCAL', route[len('auto-local:'):] + error = self.config.auto_error(wanted, repo_id) + if error: + return blocked(error) + elif self.config.native_class(route): + category, wanted = 'NATIVE_CLASS_PINNED', marker + else: + return blocked('UNSUPPORTED_STORAGE_CLASS') + item.update(category=category, repo_key=wanted, state=target_state) + with self.connection.cursor() as cursor: + if self.registry_present: + cursor.execute('SELECT repo_id FROM ' + self.table('cf_library_storage') + + ' WHERE repo_key=%s AND repo_id<>%s', (wanted, repo_id)) + if cursor.fetchone(): + return blocked('KEY_OCCUPIED') + if category == 'AUTO_LOCAL': + cursor.execute('SELECT repo_id FROM ' + self.table('RepoStorageId') + + ' WHERE storage_id=%s AND repo_id<>%s', (route, repo_id)) + if cursor.fetchone(): + return blocked('KEY_OCCUPIED') + if key is not None: + if key != wanted or category == 'NATIVE_CLASS_PINNED': + return blocked('STORAGE_BINDING_CONFLICT') + item.update(state=state, expected_state=target_state) + if state != target_state and not (state == 'reserved' and target_state == 'active'): + return dict(item, status='STATE_DRIFT', reason='LIFECYCLE_STATE_MISMATCH') + return dict(item, status='REGISTERED') + return dict(item, status=category) + + def apply_one(self, repo_id, normalize=()): + try: + with self.atomic(): + item = self.inspect(repo_id, lock=True) + category = item['status'] + normalize_route = (category == 'NATIVE_CLASS_PINNED' and item['storage_id'] in normalize) + if category not in ('NATIVE', 'AUTO_LOCAL') and not normalize_route: + return item + if normalize_route: + with self.connection.cursor() as cursor: + cursor.execute('DELETE FROM ' + self.table('RepoStorageId') + + ' WHERE repo_id=%s AND BINARY storage_id = BINARY %s', + (repo_id, item['storage_id'])) + if cursor.rowcount != 1: + raise RuntimeError('NATIVE_ROUTE_CHANGED') + route = None if normalize_route else item['storage_id'] + route_check = 's.repo_id IS NULL' if route is None else 'BINARY s.storage_id = BINARY %s' + sql = ('INSERT INTO ' + self.table('cf_library_storage') + '(repo_key,repo_id,state) ' + 'SELECT %s,i.repo_id,CASE WHEN r.repo_id IS NOT NULL OR t.repo_id IS NOT NULL ' + "THEN 'active' ELSE 'retired' END FROM (SELECT %s AS repo_id) i " + 'LEFT JOIN ' + self.table('Repo') + ' r ON r.repo_id=i.repo_id ' + 'LEFT JOIN ' + self.table('RepoTrash') + ' t ON t.repo_id=i.repo_id ' + 'LEFT JOIN ' + self.table('GarbageRepos') + ' g ON g.repo_id=i.repo_id ' + 'LEFT JOIN ' + self.table('VirtualRepo') + ' v ON v.repo_id=i.repo_id ' + 'LEFT JOIN ' + self.table('RepoStorageId') + ' s ON s.repo_id=i.repo_id ' + 'WHERE (r.repo_id IS NOT NULL OR t.repo_id IS NOT NULL OR g.repo_id IS NOT NULL) ' + 'AND v.repo_id IS NULL AND ' + route_check + ' FOR UPDATE') + with self.connection.cursor() as cursor: + cursor.execute(sql, (item['repo_key'], repo_id) + (() if route is None else (route,))) + if cursor.rowcount != 1: + raise RuntimeError('NATIVE_ROUTE_CHANGED') + after = self.inspect(repo_id) + if after['status'] != 'REGISTERED': + raise RuntimeError('STORAGE_FACTS_CHANGED') + return dict(after, action='REGISTERED', normalized=normalize_route) + except Exception as error: + if getattr(error, 'args', (None,))[0] == 1062: + # A competing writer wins only if its complete binding agrees. + after = self.inspect(repo_id) + if after['status'] == 'REGISTERED': + return after + return dict(after, status='BLOCKED', reason='UNIQUE_KEY_CONFLICT') + if isinstance(error, RuntimeError) and str(error) in ('NATIVE_ROUTE_CHANGED', 'STORAGE_FACTS_CHANGED'): + return dict(repo_id=repo_id, status='BLOCKED', reason=str(error)) + raise + + def orphans(self): + if not self.config.root or not real_directory(self.config.root): + return [] + with self.connection.cursor() as cursor: + cursor.execute('SELECT storage_id FROM ' + self.table('RepoStorageId') + + " WHERE storage_id LIKE 'auto-local:%'") + referenced = {row[0][len('auto-local:'):] for row in cursor.fetchall()} + return [dict(path=str(path), status='ORPHAN_DIRECTORY') + for path in sorted(self.config.root.iterdir()) + if (path.is_dir() or path.is_symlink()) and path.name not in referenced] + + def run(self, *, apply=False, batch_size=500, normalize=(), repo_id=None): + schema_status = self.ensure_schema(apply=apply) + items = [] + iterator = (self.inspect(repo_id),) if repo_id else self.scan(batch_size) + for item in iterator: + if apply: + item = self.apply_one(item['repo_id'], normalize) + items.append(item) + orphans = self.orphans() + counts = Counter(item['status'] for item in items) + counts['ORPHAN_DIRECTORY'] += len(orphans) + if schema_status == 'SCHEMA_MISSING': + counts['SCHEMA_MISSING'] = 1 + return dict(schema={'status': schema_status}, mode='apply' if apply else 'dry-run', summary=dict(sorted(counts.items())), + items=items, orphan_directories=orphans) diff --git a/cloudfile_ext/v2/tests/test_fresh.py b/cloudfile_ext/v2/tests/test_fresh.py index 4479ab00f9e..825fca8807c 100644 --- a/cloudfile_ext/v2/tests/test_fresh.py +++ b/cloudfile_ext/v2/tests/test_fresh.py @@ -281,8 +281,8 @@ def test_historical_native_repo_registration_is_idempotent_and_preserves_facts(s for table in ("Repo", "RepoOwner", "Branch", "RepoStorageId")} discovery = NativeLibraryDiscovery(self.connection) self.assertEqual(discovery.inspect(repo)["status"], "NATIVE_UNREGISTERED") - self.assertEqual(discovery.reconcile(repo), {"NATIVE_REGISTERED": 1}) - self.assertEqual(discovery.reconcile(repo), {"NATIVE_REGISTERED": 1}) + self.assertEqual(discovery.reconcile(repo), {"REGISTERED": 1, "ORPHAN_DIRECTORY": 0}) + self.assertEqual(discovery.reconcile(repo), {"REGISTERED": 1, "ORPHAN_DIRECTORY": 0}) self.schema.install() # fresh schema recheck must not erase an adopted row self.assertEqual(discovery.inspect(repo)["status"], "NATIVE_REGISTERED") self.assertEqual(self.query("SELECT repo_key,state FROM cf_library_storage WHERE repo_id=%s", (repo,)), @@ -292,19 +292,19 @@ def test_historical_native_repo_registration_is_idempotent_and_preserves_facts(s def test_historical_registration_refuses_missing_native_and_existing_metadata(self): discovery = NativeLibraryDiscovery(self.connection) missing = str(uuid4()) - self.assertEqual(discovery.reconcile(missing), {"NATIVE_RESOURCE_MISSING": 1}) + self.assertEqual(discovery.reconcile(missing), {"BLOCKED": 1, "ORPHAN_DIRECTORY": 0}) self.assertEqual(self.query("SELECT * FROM cf_library_storage"), ()) external = str(uuid4()) self.query("INSERT INTO Repo(repo_id) VALUES(%s)", (external,)) self.query("INSERT INTO RepoStorageId(repo_id,storage_id) VALUES(%s,%s)", (external, "external-legacy")) self.assertEqual(discovery.reconcile(external), - {"EXTERNAL_STORAGE_REQUIRES_REVIEW": 1}) + {"BLOCKED": 1, "ORPHAN_DIRECTORY": 0}) conflict = str(uuid4()) self.query("INSERT INTO Repo(repo_id) VALUES(%s)", (conflict,)) self.query("INSERT INTO cf_library_storage(repo_key,repo_id,state) VALUES(%s,%s,%s)", ("kept-business-key", conflict, "active")) self.assertEqual(discovery.reconcile(conflict), - {"STORAGE_BINDING_CONFLICT": 1}) + {"BLOCKED": 1, "ORPHAN_DIRECTORY": 0}) self.assertEqual(self.query("SELECT repo_key FROM cf_library_storage WHERE repo_id=%s", (conflict,)), (("kept-business-key",),)) diff --git a/cloudfile_ext/v2/tests/test_storage_reconcile.py b/cloudfile_ext/v2/tests/test_storage_reconcile.py new file mode 100644 index 00000000000..bf87a991200 --- /dev/null +++ b/cloudfile_ext/v2/tests/test_storage_reconcile.py @@ -0,0 +1,518 @@ +"""Storage classification and real disposable MariaDB transaction regressions.""" +from concurrent.futures import ThreadPoolExecutor +from contextlib import redirect_stdout +import io +import json +import os +from pathlib import Path +from tempfile import TemporaryDirectory +from threading import Barrier +import unittest +from types import SimpleNamespace +from unittest.mock import patch +from uuid import uuid4 + +from cloudfile_ext.v2.storage_reconcile import StorageConfig, StorageReconciler, STORAGE_DDL, valid_key + + +class ConfigTests(unittest.TestCase): + def test_key_matches_c_storage_contract(self): + for key in ('a', 'a-1', 'a' * 80): + self.assertTrue(valid_key(key)) + for key in ('', '-a', 'a-', '../a', 'A', 'a_b', 'a' * 81): + self.assertFalse(valid_key(key)) + + def test_reads_native_and_legacy_config_and_rejects_invalid_root(self): + with TemporaryDirectory() as tmp: + root = Path(tmp).resolve() + tmp = str(root) + conf = root / 'seafile.conf' + conf.write_text('[cloudfile_v2]\nauto_local_root=' + tmp + '\n[storage]\nbackend=filesystem\n') + self.assertEqual(StorageConfig.read(conf, root).root, root) + classes = [{'storage_id': 'auto-local', **{kind: {'backend': 'fs', 'dir': tmp} + for kind in ('commits', 'fs', 'blocks')}}] + (root / 'classes.json').write_text(json.dumps(classes)) + conf.write_text('[storage]\nenable_storage_classes=true\nstorage_classes_file=classes.json\n') + self.assertEqual(StorageConfig.read(conf, root).root, root) + conf.write_text('[cloudfile_v2]\nauto_local_root=' + tmp + '/missing\n') + with self.assertRaisesRegex(ValueError, 'INVALID_STORAGE_ROOT'): + StorageConfig.read(conf, root) + (root / 'link').symlink_to(root, target_is_directory=True) + conf.write_text('[cloudfile_v2]\nauto_local_root=' + tmp + '/link\n') + with self.assertRaisesRegex(ValueError, 'INVALID_STORAGE_ROOT'): + StorageConfig.read(conf, root) + + def test_managed_root_rejects_symlink_ancestor(self): + with TemporaryDirectory() as tmp: + root = Path(tmp).resolve() + (root / 'real' / 'libraries').mkdir(parents=True) + (root / 'alias').symlink_to(root / 'real', target_is_directory=True) + conf = root / 'seafile.conf' + conf.write_text('[cloudfile_v2]\nauto_local_root=' + str(root / 'alias' / 'libraries') + '\n') + with self.assertRaisesRegex(ValueError, 'INVALID_STORAGE_ROOT'): + StorageConfig.read(conf, root) + config = StorageConfig(root / 'alias' / 'libraries', root) + self.assertEqual(config.auto_error('key', str(uuid4())), 'INVALID_STORAGE_ROOT') + + def test_legacy_commands_cannot_silently_ignore_dry_run(self): + from cloudfile_ext.v2.__main__ import main + with patch('sys.stderr', io.StringIO()), self.assertRaises(SystemExit) as error: + main(['reconcile', '--ddl', '/unused.sql', '--dry-run']) + self.assertEqual(error.exception.code, 2) + + def test_legacy_entry_points_detect_existing_seafile_configuration(self): + with TemporaryDirectory() as tmp: + root = Path(tmp).resolve() + data = root / 'seafile-data' + data.mkdir() + conf = root / 'conf' + conf.mkdir() + (conf / 'seafile.conf').write_text('[cloudfile_v2]\nauto_local_root=' + str(root) + '\n') + with patch.dict(os.environ, {'SEAFILE_DATA_DIR': str(data)}, clear=True): + self.assertEqual(StorageConfig.detect().root, root) + + def test_cli_default_dry_run_and_machine_readable_blocked(self): + from cloudfile_ext.v2.__main__ import main + output = io.StringIO() + with TemporaryDirectory() as tmp: + conf = Path(tmp) / 'seafile.conf' + conf.write_text('[cloudfile_v2]\nauto_local_root=' + tmp + '/missing\n') + with redirect_stdout(output): + code = main(['storage-reconcile', '--seafile-conf', str(conf), '--seafile-data', tmp]) + report = json.loads(output.getvalue()) + self.assertEqual((code, report['mode'], report['summary']), (2, 'dry-run', {'BLOCKED': 1})) + self.assertEqual(report['items'][0]['reason'], 'INVALID_STORAGE_ROOT') + + +@unittest.skipUnless(os.environ.get('CF_STORAGE_TEST_MYSQL_PORT'), 'disposable MariaDB required') +class StorageSQLTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + import pymysql + cls.options = dict(host='127.0.0.1', port=int(os.environ['CF_STORAGE_TEST_MYSQL_PORT']), + user='root', password='storage-test-only', autocommit=True, charset='utf8mb4') + cls.admin = pymysql.connect(**cls.options) + + @classmethod + def tearDownClass(cls): + cls.admin.close() + + def setUp(self): + import pymysql + self.schema = 'storage_' + uuid4().hex + with self.admin.cursor() as cursor: + cursor.execute('CREATE DATABASE ' + self.schema) + self.db_options = dict(self.options, database=self.schema) + self.connection = pymysql.connect(**self.db_options) + self.temp = TemporaryDirectory() + self.root = Path(self.temp.name).resolve() + self.native = self.root / 'seafile-data' + self.native.mkdir() + self.auto = self.root / 'libraries' + self.auto.mkdir() + self.config = StorageConfig(self.auto, self.native) + with self.connection.cursor() as cursor: + for table in ('Repo', 'RepoTrash', 'GarbageRepos', 'VirtualRepo'): + cursor.execute('CREATE TABLE ' + table + '(repo_id CHAR(36) PRIMARY KEY) ENGINE=InnoDB') + cursor.execute('CREATE TABLE RepoStorageId(repo_id CHAR(36) PRIMARY KEY,storage_id VARCHAR(255)) ENGINE=InnoDB') + # The legacy table exists with no usable fields: touching it breaks tests. + cursor.execute('CREATE TABLE cf_library_storage_key(untouched INT) ENGINE=InnoDB') + cursor.execute('INSERT INTO cf_library_storage_key VALUES(42)') + self.service = StorageReconciler(self.connection, self.config) + self.service.ensure_schema() + + def tearDown(self): + self.connection.close() + with self.admin.cursor() as cursor: + cursor.execute('DROP DATABASE ' + self.schema) + self.temp.cleanup() + + def sql(self, sql, args=()): + with self.connection.cursor() as cursor: + cursor.execute(sql, args) + return cursor.fetchall() + + def repo(self, *, route=None, source='Repo', key=None, state='active', physical=True): + rid = str(uuid4()) + self.sql('INSERT INTO ' + source + '(repo_id) VALUES(%s)', (rid,)) + if route is not None: + self.sql('INSERT INTO RepoStorageId VALUES(%s,%s)', (rid, route)) + if key is not None: + self.sql('INSERT INTO cf_library_storage VALUES(%s,%s,%s)', (key, rid, state)) + if route and route.startswith('auto-local:') and physical and valid_key(route[11:]): + (self.auto / route[11:] / 'storage' / 'commits' / rid).mkdir(parents=True, exist_ok=True) + return rid + + def test_schema_exact_idempotent_and_legacy_untouched(self): + self.service.ensure_schema() + self.service.ensure_schema() + self.assertEqual(self.sql('SELECT * FROM cf_library_storage_key'), ((42,),)) + self.sql('ALTER TABLE cf_library_storage ADD bad INT') + with self.assertRaisesRegex(RuntimeError, 'STORAGE_SCHEMA_DRIFT'): + self.service.ensure_schema() + + def test_native_auto_trash_and_garbage_lifecycle_and_idempotency(self): + native = self.repo() + auto = self.repo(route='auto-local:old-key') + trash = self.repo(source='RepoTrash', route='auto-local:trash-key') + garbage = self.repo(source='GarbageRepos') + self.sql('INSERT INTO GarbageRepos VALUES(%s)', (native,)) + report = self.service.run(batch_size=1) + self.assertEqual(report['summary'], {'AUTO_LOCAL': 2, 'NATIVE': 2, 'ORPHAN_DIRECTORY': 0}) + self.assertEqual(self.sql('SELECT * FROM cf_library_storage'), ()) + applied = self.service.run(apply=True, batch_size=1) + self.assertEqual(applied['summary']['REGISTERED'], 4) + expected = {native: ('native-' + native, 'active'), auto: ('old-key', 'active'), + trash: ('trash-key', 'active'), garbage: ('native-' + garbage, 'retired')} + self.assertEqual({rid: (key, state) for key, rid, state in self.sql('SELECT * FROM cf_library_storage')}, expected) + self.assertFalse(any('action' in item for item in self.service.run(apply=True)['items'])) + self.assertEqual(self.sql('SELECT * FROM cf_library_storage_key'), ((42,),)) + + def test_virtual_skipped_in_all_native_lifecycle_locations(self): + for source in ('Repo', 'RepoTrash', 'GarbageRepos'): + rid = self.repo(source=source) + self.sql('INSERT INTO VirtualRepo VALUES(%s)', (rid,)) + report = self.service.run(apply=True, batch_size=1) + self.assertEqual(report['summary']['VIRTUAL_REPO'], 3) + self.assertEqual(self.sql('SELECT * FROM cf_library_storage'), ()) + + def test_invalid_key_missing_directory_symlink_and_empty_objects_are_blocked(self): + cases = [('auto-local:../escape', 'INVALID_KEY', False), + ('auto-local:missing', 'MISSING_STORAGE_DIRECTORY', False), + ('auto-local:empty', 'MISSING_REPOSITORY_DIRECTORY', False)] + (self.auto / 'empty' / 'storage').mkdir(parents=True) + for route, reason, physical in cases: + rid = self.repo(route=route, physical=physical) + self.assertEqual(self.service.inspect(rid)['reason'], reason) + rid = self.repo(route='auto-local:linked', physical=False) + (self.auto / 'linked').symlink_to(self.native, target_is_directory=True) + self.assertEqual(self.service.inspect(rid)['reason'], 'SYMLINK_DIRECTORY') + rid = self.repo(route='auto-local:object-link') + objects = self.auto / 'object-link' / 'storage' / 'commits' / rid + objects.rmdir() + objects.symlink_to(self.native, target_is_directory=True) + self.assertEqual(self.service.inspect(rid)['reason'], 'SYMLINK_DIRECTORY') + self.assertEqual(self.service.run(apply=True)['summary']['BLOCKED'], 5) + + def test_existing_binding_consistency_state_and_occupied_key(self): + rid = self.repo() + self.sql('INSERT INTO cf_library_storage VALUES(%s,%s,%s)', ('native-' + rid, rid, 'active')) + self.assertEqual(self.service.inspect(rid)['status'], 'REGISTERED') + conflict = self.repo(key='wrong') + self.assertEqual(self.service.inspect(conflict)['reason'], 'STORAGE_BINDING_CONFLICT') + wrong_state = self.repo(route='auto-local:wrong-state', key='wrong-state', state='retired') + self.assertEqual(self.service.inspect(wrong_state)['status'], 'STATE_DRIFT') + occupied = self.repo(route='auto-local:occupied') + self.repo(key='occupied') + self.assertEqual(self.service.inspect(occupied)['reason'], 'KEY_OCCUPIED') + self.service.run(apply=True) + self.assertEqual(self.sql('SELECT repo_key FROM cf_library_storage WHERE repo_id=%s', (conflict,)), (('wrong',),)) + + def test_two_native_routes_cannot_claim_same_key(self): + first = self.repo(route='auto-local:duplicate') + second = self.repo(route='auto-local:duplicate') + for rid in (first, second): + self.assertEqual(self.service.apply_one(rid)['reason'], 'KEY_OCCUPIED') + self.assertEqual(self.sql('SELECT * FROM cf_library_storage'), ()) + + def test_pinned_native_class_requires_explicit_normalization(self): + self.config.classes['legacy-default'] = {kind: {'backend': 'fs', 'dir': str(self.native)} + for kind in ('commits', 'fs', 'blocks')} + rid = self.repo(route='legacy-default') + self.assertEqual(self.service.run(apply=True)['summary']['NATIVE_CLASS_PINNED'], 1) + self.assertEqual(self.sql('SELECT storage_id FROM RepoStorageId WHERE repo_id=%s', (rid,)), (('legacy-default',),)) + item = self.service.apply_one(rid, normalize=['legacy-default']) + self.assertEqual((item['status'], item['normalized']), ('REGISTERED', True)) + self.assertEqual(self.sql('SELECT * FROM RepoStorageId WHERE repo_id=%s', (rid,)), ()) + self.assertEqual(self.sql('SELECT repo_key FROM cf_library_storage WHERE repo_id=%s', (rid,)), (('native-' + rid,),)) + + def test_normalize_rolls_back_route_deletion_when_insert_fails(self): + self.config.classes['legacy-default'] = {kind: {'backend': 'fs', 'dir': str(self.native)} + for kind in ('commits', 'fs', 'blocks')} + rid = self.repo(route='legacy-default') + self.sql("CREATE TRIGGER reject_storage BEFORE INSERT ON cf_library_storage FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='test refusal'") + with self.assertRaises(Exception): + self.service.apply_one(rid, normalize=['legacy-default']) + self.assertEqual(self.sql('SELECT storage_id FROM RepoStorageId WHERE repo_id=%s', (rid,)), (('legacy-default',),)) + self.assertEqual(self.sql('SELECT * FROM cf_library_storage'), ()) + + def test_s3_mixed_backend_external_local_and_unknown_class_blocked(self): + for name, kinds in [('s3', {kind: {'backend': 's3', 'dir': str(self.native)} for kind in ('commits', 'fs', 'blocks')}), + ('other-local', {kind: {'backend': 'fs', 'dir': str(self.auto)} for kind in ('commits', 'fs', 'blocks')}), + ('mixed', {kind: {'backend': 'fs' if kind != 'blocks' else 's3', 'dir': str(self.native)} for kind in ('commits', 'fs', 'blocks')})]: + self.config.classes[name] = kinds + self.assertEqual(self.service.inspect(self.repo(route=name))['reason'], 'UNSUPPORTED_STORAGE_CLASS') + self.assertEqual(self.service.inspect(self.repo(route='unknown'))['reason'], 'UNSUPPORTED_STORAGE_CLASS') + + def test_orphans_report_only_and_all_routes_count_as_references(self): + (self.auto / 'orphan').mkdir() + rid = self.repo(route='auto-local:referenced', source='GarbageRepos') + orphan = self.service.run(apply=True)['orphan_directories'] + self.assertEqual(orphan, [{'path': str(self.auto / 'orphan'), 'status': 'ORPHAN_DIRECTORY'}]) + self.assertTrue((self.auto / 'orphan').exists()) + self.assertEqual(self.service.inspect(rid)['state'], 'retired') + + def test_concurrent_same_library_is_idempotent(self): + import pymysql + rid = self.repo(route='auto-local:race') + barrier = Barrier(2) + def worker(): + conn = pymysql.connect(**self.db_options) + try: + barrier.wait(timeout=5) + return StorageReconciler(conn, self.config).apply_one(rid) + finally: + conn.close() + with ThreadPoolExecutor(max_workers=2) as pool: + results = list(pool.map(lambda _: worker(), range(2))) + self.assertEqual([item['status'] for item in results], ['REGISTERED', 'REGISTERED']) + self.assertEqual(self.sql('SELECT COUNT(*) FROM cf_library_storage'), ((1,),)) + self.assertEqual(sum(item.get('action') == 'REGISTERED' for item in results), 1) + + def test_unique_key_collision_never_overwrites_and_normalize_rolls_back(self): + rid = self.repo() + original = self.service.inspect + def inspected(*args, **kwargs): + item = original(*args, **kwargs) + if kwargs.get('lock'): + # A conflicting committed row is already present by INSERT time. + self.sql('INSERT INTO cf_library_storage VALUES(%s,%s,%s)', + ('native-' + rid, str(uuid4()), 'active')) + return item + with patch.object(self.service, 'inspect', side_effect=inspected): + result = self.service.apply_one(rid) + # Conflict insertion was part of this transaction and rolls back too. + self.assertEqual((result['status'], result['reason']), ('BLOCKED', 'UNIQUE_KEY_CONFLICT')) + self.assertEqual(self.sql('SELECT * FROM cf_library_storage'), ()) + + def test_cli_dry_run_apply_and_blocked_exit_codes_on_real_schema(self): + from cloudfile_ext.v2.__main__ import main + rid = self.repo(route='auto-local:cli-library') + conf = self.root / 'seafile.conf' + conf.write_text('[cloudfile_v2]\nauto_local_root=' + str(self.auto) + '\n') + args = ['storage-reconcile', '--seafile-conf', str(conf), '--seafile-data', str(self.native)] + env = dict(CF_V2_DB_HOST='127.0.0.1', CF_V2_DB_PORT=str(self.options['port']), + CF_V2_DB_USER='root', CF_V2_DB_PASSWORD='storage-test-only', CF_V2_DB_NAME=self.schema) + def invoke(extra): + output = io.StringIO() + with patch.dict(os.environ, env), redirect_stdout(output): + code = main(args + extra) + return code, json.loads(output.getvalue()) + code, report = invoke([]) + self.assertEqual((code, report['mode'], report['summary']['AUTO_LOCAL']), (0, 'dry-run', 1)) + self.assertEqual(self.sql('SELECT * FROM cf_library_storage'), ()) + code, report = invoke(['--apply', '--batch-size', '1']) + self.assertEqual((code, report['mode'], report['summary']['REGISTERED']), (0, 'apply', 1)) + self.repo(route='unknown') + code, report = invoke(['--dry-run']) + self.assertEqual((code, report['summary']['BLOCKED']), (2, 1)) + + def test_unique_key_collision_during_normalize_rolls_back_delete(self): + self.config.classes['legacy-default'] = {kind: {'backend': 'fs', 'dir': str(self.native)} + for kind in ('commits', 'fs', 'blocks')} + rid = self.repo(route='legacy-default') + original = self.service.inspect + def inspected(*args, **kwargs): + item = original(*args, **kwargs) + if kwargs.get('lock'): + self.sql('INSERT INTO cf_library_storage VALUES(%s,%s,%s)', + ('native-' + rid, str(uuid4()), 'active')) + return item + with patch.object(self.service, 'inspect', side_effect=inspected): + result = self.service.apply_one(rid, normalize=['legacy-default']) + self.assertEqual((result['status'], result['reason']), ('BLOCKED', 'UNIQUE_KEY_CONFLICT')) + self.assertEqual(self.sql('SELECT storage_id FROM RepoStorageId WHERE repo_id=%s', (rid,)), (('legacy-default',),)) + self.assertEqual(self.sql('SELECT * FROM cf_library_storage'), ()) + + def test_hub_startup_uses_same_rules_and_never_normalizes_classes(self): + from cloudfile_ext.v2.native_discovery import reconcile_from_hub + native = self.repo() + auto = self.repo(route='auto-local:startup') + pinned = self.repo(route='legacy-default') + classes = [{'storage_id': 'legacy-default', **{kind: {'backend': 'fs', 'dir': str(self.native)} + for kind in ('commits', 'fs', 'blocks')}}] + conf = self.root / 'seafile.conf' + conf.write_text('[cloudfile_v2]\nauto_local_root=' + str(self.auto) + '\n[storage]\nstorage_classes_file=classes.json\n') + (self.root / 'classes.json').write_text(json.dumps(classes)) + modules = {'django.db': SimpleNamespace(connection=self.connection, + transaction=SimpleNamespace(atomic=self.service._atomic)), + 'seahub.utils.db_api': SimpleNamespace(SeafileDB=lambda: SimpleNamespace(db_name=self.schema))} + env = dict(SEAFILE_DATA_DIR=str(self.native), SEAFILE_CENTRAL_CONF_DIR=str(self.root)) + with patch.dict('sys.modules', modules), patch.dict(os.environ, env), self.assertLogs('cloudfile_ext.v2.native_discovery', level='WARNING'): + report = reconcile_from_hub(batch_size=1) + self.assertEqual((report['registered'], report['summary']['NATIVE_CLASS_PINNED']), (2, 1)) + self.assertEqual(self.sql('SELECT storage_id FROM RepoStorageId WHERE repo_id=%s', (pinned,)), (('legacy-default',),)) + self.assertEqual({row[0] for row in self.sql('SELECT repo_id FROM cf_library_storage')}, {native, auto}) + + def test_null_native_route_is_blocked_instead_of_treated_as_absent(self): + rid = self.repo() + self.sql('INSERT INTO RepoStorageId VALUES(%s,NULL)', (rid,)) + self.assertEqual(self.service.apply_one(rid)['reason'], 'INVALID_STORAGE_ROUTE') + self.assertEqual(self.sql('SELECT * FROM cf_library_storage'), ()) + + def test_internal_storage_kind_and_repo_symlinks_are_blocked(self): + for part in ('storage', 'commits', 'repo'): + key = 'linked-' + part + rid = self.repo(route='auto-local:' + key) + directory = self.auto / key / 'storage' + if part == 'commits': + directory = directory / 'commits' + elif part == 'repo': + directory = directory / 'commits' / rid + moved = self.root / ('outside-' + part) + directory.rename(moved) + directory.symlink_to(moved, target_is_directory=True) + self.assertEqual(self.service.apply_one(rid)['reason'], 'SYMLINK_DIRECTORY') + self.assertEqual(self.sql('SELECT * FROM cf_library_storage'), ()) + + def test_trash_and_garbage_overlap_stays_active(self): + rid = self.repo(source='RepoTrash', route='auto-local:trash-overlap') + self.sql('INSERT INTO GarbageRepos VALUES(%s)', (rid,)) + self.assertEqual(self.service.apply_one(rid)['state'], 'active') + self.assertEqual(self.sql('SELECT state FROM cf_library_storage WHERE repo_id=%s', (rid,)), (('active',),)) + + def test_orphan_references_include_routes_outside_the_selected_batch(self): + selected = self.repo() + outside = self.repo(route='auto-local:outside') + self.sql('DELETE FROM Repo WHERE repo_id=%s', (outside,)) + report = self.service.run(repo_id=selected, batch_size=1) + self.assertEqual(len(report['items']), 1) + self.assertEqual(report['orphan_directories'], []) + + def test_missing_registry_dry_run_classifies_without_creating_then_apply_installs(self): + from cloudfile_ext.v2.__main__ import main + rid = self.repo() + self.sql('DROP TABLE cf_library_storage') + conf = self.root / 'seafile.conf' + conf.write_text('[cloudfile_v2]\nauto_local_root=' + str(self.auto) + '\n') + env = dict(CF_V2_DB_HOST='127.0.0.1', CF_V2_DB_PORT=str(self.options['port']), + CF_V2_DB_USER='root', CF_V2_DB_PASSWORD='storage-test-only', CF_V2_DB_NAME=self.schema) + output = io.StringIO() + with patch.dict(os.environ, env), redirect_stdout(output): + code = main(['storage-reconcile', '--seafile-conf', str(conf), '--seafile-data', str(self.native)]) + self.assertEqual(code, 0) + report = json.loads(output.getvalue()) + self.assertEqual(report['summary']['NATIVE'], 1) + self.assertEqual(report['summary']['SCHEMA_MISSING'], 1) + self.assertEqual(report['schema']['status'], 'SCHEMA_MISSING') + self.assertEqual(self.sql("SELECT table_name FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name='cf_library_storage'"), ()) + self.assertEqual(self.sql('SELECT * FROM cf_library_storage_key'), ((42,),)) + with patch.dict(os.environ, env), redirect_stdout(io.StringIO()): + self.assertEqual(main(['storage-reconcile', '--seafile-conf', str(conf), '--seafile-data', str(self.native), '--apply']), 0) + self.assertEqual(self.sql('SELECT state FROM cf_library_storage WHERE repo_id=%s', (rid,)), (('active',),)) + + def test_dry_run_works_with_select_only_user_and_read_only_transaction(self): + import pymysql + rid = self.repo() + username = 'storage_ro_' + uuid4().hex[:16] + with self.admin.cursor() as cursor: + cursor.execute('CREATE USER %s IDENTIFIED BY %s', (username, 'select-only-test')) + cursor.execute('GRANT SELECT ON ' + self.schema + '.* TO %s', (username,)) + try: + for missing in (False, True): + if missing: + self.sql('DROP TABLE cf_library_storage') + readonly = pymysql.connect(**dict(self.db_options, user=username, password='select-only-test')) + try: + with readonly.cursor() as cursor: + cursor.execute('START TRANSACTION READ ONLY') + report = StorageReconciler(readonly, self.config).run() + self.assertEqual(report['summary']['NATIVE'], 1) + self.assertEqual(report['schema']['status'], 'SCHEMA_MISSING' if missing else 'SCHEMA_READY') + finally: + readonly.rollback() + readonly.close() + self.assertEqual(self.sql("SELECT table_name FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name='cf_library_storage'"), ()) + finally: + with self.admin.cursor() as cursor: + cursor.execute('DROP USER %s', (username,)) + + def test_read_only_schema_verifier_rejects_changed_contract_without_altering_it(self): + variants = [STORAGE_DDL.replace("'reserved','active','retired'", "'reserved','active'"), + STORAGE_DDL.replace(' UNIQUE KEY library_storage_repo (repo_id),\n', ''), + STORAGE_DDL.replace('COLLATE=utf8mb4_bin', 'COLLATE=utf8mb4_general_ci'), + STORAGE_DDL.replace('state VARCHAR(8) CHARACTER SET ascii COLLATE ascii_bin NOT NULL', + "state VARCHAR(8) CHARACTER SET ascii COLLATE ascii_bin NOT NULL DEFAULT 'active'")] + for ddl in variants: + with self.subTest(ddl=ddl): + self.sql('DROP TABLE cf_library_storage') + self.sql(ddl) + before = self.sql('SHOW CREATE TABLE cf_library_storage') + with self.assertRaisesRegex(RuntimeError, 'STORAGE_SCHEMA_DRIFT'): + self.service.ensure_schema(apply=False) + self.assertEqual(before, self.sql('SHOW CREATE TABLE cf_library_storage')) + + def test_reserved_active_compatibility_and_state_drift_do_not_write_or_block(self): + reserved = self.repo(route='auto-local:reserved-library', key='reserved-library', state='reserved') + retired = self.repo(source='RepoTrash', route='auto-local:retired-trash', key='retired-trash', state='retired') + active_gc = self.repo(source='GarbageRepos', route='auto-local:active-garbage', key='active-garbage', state='active') + native_retired = self.repo(source='RepoTrash') + self.sql('INSERT INTO cf_library_storage VALUES(%s,%s,%s)', ('native-' + native_retired, native_retired, 'retired')) + before = self.sql('SELECT * FROM cf_library_storage ORDER BY repo_id') + report = self.service.run(apply=True) + self.assertNotIn('BLOCKED', report['summary']) + self.assertEqual(report['summary']['REGISTERED'], 1) + self.assertEqual(report['summary']['STATE_DRIFT'], 3) + by_id = {item['repo_id']: item for item in report['items']} + self.assertEqual((by_id[reserved]['state'], by_id[reserved]['expected_state']), ('reserved', 'active')) + self.assertEqual((by_id[retired]['state'], by_id[retired]['expected_state']), ('retired', 'active')) + self.assertEqual((by_id[active_gc]['state'], by_id[active_gc]['expected_state']), ('active', 'retired')) + self.assertEqual((by_id[native_retired]['state'], by_id[native_retired]['expected_state']), ('retired', 'active')) + self.assertEqual(before, self.sql('SELECT * FROM cf_library_storage ORDER BY repo_id')) + self.assertFalse(any('action' in item for item in report['items'])) + + def test_native_default_class_resolves_ce_symlink_layout_only(self): + alias = self.root / 'seafile-data-alias' + alias.symlink_to(self.native, target_is_directory=True) + self.config.native_dir = alias + self.config.classes['legacy-default'] = {kind: {'backend': 'fs', 'dir': str(alias)} + for kind in ('commits', 'fs', 'blocks')} + rid = self.repo(route='legacy-default') + self.assertEqual(self.service.inspect(rid)['status'], 'NATIVE_CLASS_PINNED') + self.assertEqual(self.service.apply_one(rid, normalize=['legacy-default'])['status'], 'REGISTERED') + managed_alias = self.root / 'managed-alias' + managed_alias.symlink_to(self.auto, target_is_directory=True) + self.config.root = managed_alias + self.assertEqual(self.config.auto_error('some-key', str(uuid4())), 'INVALID_STORAGE_ROOT') + + def test_wrong_database_fails_before_creating_registry(self): + self.sql('DROP TABLE cf_library_storage') + self.sql('RENAME TABLE Repo TO NotNativeRepo') + with self.assertRaisesRegex(RuntimeError, 'NATIVE_SEAFILE_DATABASE_NOT_CONNECTED'): + self.service.ensure_schema() + self.assertEqual(self.sql("SELECT table_name FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name='cf_library_storage'"), ()) + + def test_write_statement_rechecks_absent_route_not_null_storage_value(self): + rid = self.repo() + original = self.service.inspect + def changed(*args, **kwargs): + item = original(*args, **kwargs) + if kwargs.get('lock'): + self.sql('INSERT INTO RepoStorageId VALUES(%s,NULL)', (rid,)) + return item + with patch.object(self.service, 'inspect', side_effect=changed): + item = self.service.apply_one(rid) + self.assertEqual(item['reason'], 'NATIVE_ROUTE_CHANGED') + self.assertEqual(self.sql('SELECT * FROM cf_library_storage'), ()) + self.assertEqual(self.sql('SELECT * FROM RepoStorageId'), ()) + + def test_write_statement_rechecks_case_sensitive_route(self): + rid = self.repo(route='auto-local:exact') + original = self.service.inspect + def changed(*args, **kwargs): + item = original(*args, **kwargs) + if kwargs.get('lock'): + self.sql('UPDATE RepoStorageId SET storage_id=%s WHERE repo_id=%s', ('auto-local:EXACT', rid)) + return item + with patch.object(self.service, 'inspect', side_effect=changed): + item = self.service.apply_one(rid) + self.assertEqual(item['reason'], 'NATIVE_ROUTE_CHANGED') + self.assertEqual(self.sql('SELECT * FROM cf_library_storage'), ()) + self.assertEqual(self.sql('SELECT storage_id FROM RepoStorageId WHERE repo_id=%s', (rid,)), (('auto-local:exact',),)) + + def test_route_change_between_scan_and_apply_is_reclassified(self): + rid = self.repo() + self.assertEqual(self.service.inspect(rid)['status'], 'NATIVE') + self.sql('INSERT INTO RepoStorageId VALUES(%s,%s)', (rid, 's3')) + self.assertEqual(self.service.apply_one(rid)['reason'], 'UNSUPPORTED_STORAGE_CLASS') + self.assertEqual(self.sql('SELECT * FROM cf_library_storage'), ()) diff --git a/docs/storage-reconcile.md b/docs/storage-reconcile.md new file mode 100644 index 00000000000..9f17955a68d --- /dev/null +++ b/docs/storage-reconcile.md @@ -0,0 +1,109 @@ +# 原生存储登记预检 + +Hub 基线：`ee732e74ce02f184586e1313b489e2198dbc15fe`。 + +命令只补齐既有 `cf_library_storage` 合同，不搬移对象、不修改库 UUID。数据来自 +`Repo ∪ RepoTrash ∪ GarbageRepos`、`VirtualRepo`、`RepoStorageId`、Seafile 配置与物理目录； +旧存储 key 表不读取、不修改、不删除。对象健康与实际读写需要单独验收。 + +## 命令与只读合同 + +连接凭据使用 `CF_V2_DB_HOST`、`CF_V2_DB_PORT`（默认 3306）、`CF_V2_DB_USER`、 +`CF_V2_DB_PASSWORD`、`CF_V2_DB_NAME`，目标为当前 Seafile 数据库。 + +```sh +python -m cloudfile_ext.v2 storage-reconcile \ + --seafile-conf /shared/seafile/conf/seafile.conf \ + --seafile-data /shared/seafile/seafile-data \ + --dry-run --batch-size 500 +``` + +默认 `--dry-run` **严格只读**：数据库只执行 SELECT/SHOW，不创建普通表、临时表或登记行。 +结构校验仅读取 information_schema 和 SHOW CREATE TABLE，SELECT 权限即可运行。 +登记表不存在时独立报告 `SCHEMA_MISSING`，并把登记关系视为空集继续分类；这不是业务 BLOCKED。 +不会创建数据库或原生表。原生五张来源表不完整时在任何建表前拒绝。 + +```json +{"mode":"dry-run","schema":{"status":"SCHEMA_MISSING"},"summary":{"NATIVE":1,"ORPHAN_DIRECTORY":0,"SCHEMA_MISSING":1},"items":[{"repo_id":"00000000-0000-4000-8000-000000000001","status":"NATIVE","storage_id":null,"category":"NATIVE","repo_key":"native-00000000-0000-4000-8000-000000000001","state":"active"}],"orphan_directories":[]} +``` + +只有 `--apply` 可幂等创建与 Server `cloudfile-v2.sql` 完全一致的登记表并写登记行。 +`--batch-size` 范围 1–2000；`--repo-id UUID` 可定向检查。正常退出 0；有业务 `BLOCKED` 时退出 2； +数据库/文件访问、结构漂移等执行错误退出 1，均输出单个 JSON。`SCHEMA_MISSING`、`STATE_DRIFT` +单独存在时不导致非零退出。旧 install/reconcile 命令拒绝 storage 专用模式参数，避免 +误传 `--dry-run` 后仍执行变更。 + +## 分类与生命周期 + +- 无原生路由：`NATIVE`，新登记 key 为 `native-`。存在但 storage_id 为 NULL 的路由行为 + `BLOCKED: INVALID_STORAGE_ROUTE`，不会误认作没有路由。 +- `auto-local:`：`AUTO_LOCAL`，要求 key 合法，存储根及祖先、key/storage/type/repo + 目录链没有符号链接，且至少一个对象类型存在真实库目录。 +- 三个 fs 后端均指向真实原生 `seafile-data` 的存储类：`NATIVE_CLASS_PINNED`，默认只报告。 + CE 标准原生路径符号链接可解析到相同真实原生目录；managed auto root 仍禁止符号链接。 +- 已登记且 key/路由/owner 一致：`REGISTERED`。既有 reserved 与推算 active 兼容，不阻断也不改状态。 +- 仅生命周期 active/retired 不一致：`STATE_DRIFT`，返回实际 `state` 和推算 `expected_state`， + 只报告，不 BLOCKED，也不自动覆盖。例：已 retired 但库仍在 RepoTrash 中，保持 retired，报告预期 active。 +- key、路由或 owner 冲突、非法/缺失/符号链接目录、不支持的后端：`BLOCKED`，逐条附 `reason`。 + 虚拟库为 `VIRTUAL_REPO`，不登记。未被任何原生路由引用的目录为 `ORPHAN_DIRECTORY`，不登记、不删除； + 引用检查读取全部 RepoStorageId，不限于扫描批次。 + +仅新登记记录按原生事实推算状态：Repo 或 RepoTrash 中为 active，仅 GarbageRepos 中为 retired。 +RepoTrash 与 GarbageRepos 重叠时优先 active。既有登记的状态保持不变。 + +## 显式解除原生默认类固定路由 + +只有运维明确选择的类才传 `--normalize-native-class ID`，可重复指定。dry-run 仍报告 pinned； +普通 apply 和运行时 reconcile 不解除固定路由。显式规范化在每库同一事务中删除原生路由、登记 +`native-`，失败全部回滚。S3、其他本地路径与混合后端不可转换。 + +```sh +python -m cloudfile_ext.v2 storage-reconcile \ + --seafile-conf /shared/seafile/conf/seafile.conf \ + --seafile-data /shared/seafile/seafile-data \ + --apply --normalize-native-class legacy-default +``` + +切换 filesystem 前处理业务 BLOCKED 与仍然存在的 NATIVE_CLASS_PINNED；缺登记表的只读预检可先通过， +然后显式 apply 建表、登记。配置优先 `[cloudfile_v2] auto_local_root`，否则读取存储类 JSON 的 +唯一 auto-local/fs 模板。相对 storage_classes_file 与原生 Server 一样相对于 conf 目录； +CF_STORAGE_CLASSES_JSON 优先于文件。legacy install/reconcile 从现有 Seafile 环境变量和标准目录检测配置。 + +每库一个事务；INSERT SELECT 在写时复核原生事实与字节精确路由，唯一键冲突不会覆盖其他登记。 +已成功的其他库保留，可幂等重试。Hub runtime 共用规则，只自动处理 NATIVE/AUTO_LOCAL，其他项仅报告。 + +## 版本证据与验证 + +Docker `origin/dev a7e8c2bf1e4e7630dfa2e54427c1c03d98ad667e` 的 +`deploy/compose/docker-compose.yml:27` 明确使用 `mariadb:11.4`；这是部署配方证据。 +远端设备未连接，不能据此宣称远端当前实际运行版本已核实。 + +隔离测试服务器实际 `SELECT VERSION()`： + +- `11.4.13-MariaDB-ubu2404`：与部署配方相同 11.4 系列。 +- `10.11.19-MariaDB-ubu2204`：追加兼容验证，不称为 DEV 同版。 + +```sh +CF_STORAGE_TEST_MYSQL_PORT=13319 python3 -m unittest \ + cloudfile_ext.v2.tests.test_storage_reconcile cloudfile_ext.v2.tests.test_native_discovery -v +CF_STORAGE_TEST_MYSQL_PORT=13320 python3 -m unittest \ + cloudfile_ext.v2.tests.test_storage_reconcile cloudfile_ext.v2.tests.test_native_discovery -v +``` + +两版本均 `Ran 39 tests ... OK`，后续新增结构变更只读拒绝用例在两版本各单独通过， +并对用户“native retired 但仍在 Trash”的精确场景在两版本追加定向通过。 +包含 SELECT-only 账户在 `START TRANSACTION READ ONLY` 下的有表/缺表双场景、缺表只读分类后 +apply 安装、reserved 兼容、STATE_DRIFT 保持、原生 symlink 布局、写时路由复核、并发唯一键、 +规范化回滚、Trash/Garbage、Virtual、全局 ORPHAN 与旧表哨兵不变。 + +实现阶段两轮元数据检查失败分别来自 index_name 默认排序顺序与 MariaDB 对 SELECT-only 用户隐藏 +CHECK_CONSTRAINTS；现分别用确定排序与 SHOW CREATE 纯解析修正，最终回归通过。 +原生表结构漂移（约束、唯一键、排序规则、列默认值）只读拒绝，不改表。 + +未覆盖：远端运行版本实测、真实 Server 上传/下载/历史/恢复/GC/fsck、完整三仓镜像验收。 +1,407/63 个真实库数量核对按用户最新要求留至发布后；本轮不等待副本，未部署 DEV 或生产。 + +独立的 11 库 + 1 孤儿目录、batch_size=2 CLI 合成验收也在两版本通过：缺登记表预检报告 +SCHEMA_MISSING 且确实未建表；有表预检 NATIVE=3/AUTO_LOCAL=3/PINNED=1/BLOCKED=3/Virtual=1/Orphan=1； +显式 apply/normalize 与重放 REGISTERED=7/BLOCKED=3/Virtual=1/Orphan=1，重放无写入。 +预置 BLOCKED 为 S3/非法 key/既有登记冲突，旧表哨兵不变；全部数据为合成数据。