From c2661bb833e0f8ec46f78352f0b61a0cd1167d35 Mon Sep 17 00:00:00 2001 From: dev9-bb Date: Tue, 20 Oct 2026 17:32:47 +0800 Subject: [PATCH 1/3] fix(storage): reconcile native library routes transactionally --- cloudfile_ext/v2/__main__.py | 46 ++- cloudfile_ext/v2/native_discovery.py | 122 ++----- cloudfile_ext/v2/storage_reconcile.py | 291 ++++++++++++++++ cloudfile_ext/v2/tests/test_fresh.py | 10 +- .../v2/tests/test_storage_reconcile.py | 322 ++++++++++++++++++ docs/storage-reconcile.md | 83 +++++ 6 files changed, 771 insertions(+), 103 deletions(-) create mode 100644 cloudfile_ext/v2/storage_reconcile.py create mode 100644 cloudfile_ext/v2/tests/test_storage_reconcile.py create mode 100644 docs/storage-reconcile.md diff --git a/cloudfile_ext/v2/__main__.py b/cloudfile_ext/v2/__main__.py index 7cd52b8ec8..ec991c8a27 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,56 @@ 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 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.read(args.seafile_conf, args.seafile_data) if args.seafile_conf and args.seafile_data else None 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) + discovery.ensure_schema() + 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 +64,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) 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 18269f7528..96ef3997e5 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,33 @@ 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 + import os + from pathlib import Path + 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 + + data = os.environ.get('SEAFILE_DATA_DIR', '') + conf = os.environ.get('SEAFILE_CENTRAL_CONF_DIR') or data + if not data or not conf: + raise RuntimeError('SEAFILE_STORAGE_CONFIGURATION_UNAVAILABLE') + config = StorageConfig.read(Path(conf) / 'seafile.conf', data) + discovery = StorageReconciler(connection, config, schema=SeafileDB().db_name, + atomic=transaction.atomic) + discovery.ensure_schema() + 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'): + 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 0000000000..5644bc0b52 --- /dev/null +++ b/cloudfile_ext/v2/storage_reconcile.py @@ -0,0 +1,291 @@ +"""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() + + +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): + 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 not real_directory(config.root)): + raise ValueError('INVALID_STORAGE_ROOT') + return config + + def native_class(self, route): + item = self.classes.get(route, {}) + if not self.native_dir or not real_directory(self.native_dir): + return False + return all(item.get(kind, {}).get('backend') == 'fs' and + item[kind].get('dir') and + Path(item[kind]['dir']).is_absolute() and + real_directory(Path(item[kind]['dir'])) and + Path(item[kind]['dir']).resolve() == self.native_dir.resolve() + for kind in KINDS) + + def auto_error(self, key, repo_id): + if not valid_key(key): + return 'INVALID_KEY' + if not 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.prefix = identifier(schema) + '.' if schema else '' + self.atomic = atomic or self._atomic + + 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): + # Deliberately inspect only this reviewed table: unrelated old tables + # remain present and are never enumerated, queried, or altered. + with self.connection.cursor() as cursor: + cursor.execute(STORAGE_DDL.replace('CREATE TABLE cf_library_storage', + 'CREATE TABLE IF NOT EXISTS ' + self.table('cf_library_storage'), 1)) + expected = 'cfv2_expected_storage' + cursor.execute('DROP TEMPORARY TABLE IF EXISTS ' + expected) + try: + cursor.execute(STORAGE_DDL.replace('CREATE TABLE cf_library_storage', + 'CREATE TEMPORARY TABLE ' + expected, 1)) + cursor.execute('SHOW CREATE TABLE ' + self.table('cf_library_storage')) + actual = cursor.fetchone()[1] + cursor.execute('SHOW CREATE TABLE ' + expected) + wanted = cursor.fetchone()[1] + if actual[actual.index('('):] != wanted[wanted.index('('):]: + raise RuntimeError('STORAGE_SCHEMA_DRIFT') + finally: + cursor.execute('DROP TEMPORARY TABLE IF EXISTS ' + expected) + + 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')} + sql = ('SELECT r.repo_id,t.repo_id,g.repo_id,v.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, 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') + 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: + 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 state != target_state or category == 'NATIVE_CLASS_PINNED': + return blocked('STORAGE_BINDING_CONFLICT') + 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 storage_id=%s', + (repo_id, item['storage_id'])) + if cursor.rowcount != 1: + raise RuntimeError('NATIVE_ROUTE_CHANGED') + route = None if normalize_route else item['storage_id'] + 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 s.storage_id <=> %s FOR UPDATE') + with self.connection.cursor() as cursor: + cursor.execute(sql, (item['repo_key'], repo_id, 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): + 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) + return dict(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 4479ab00f9..825fca8807 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 0000000000..5401e6b038 --- /dev/null +++ b/cloudfile_ext/v2/tests/test_storage_reconcile.py @@ -0,0 +1,322 @@ +"""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) + 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_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) + 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)['reason'], 'STORAGE_BINDING_CONFLICT') + 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_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 0000000000..01eabe1b2f --- /dev/null +++ b/docs/storage-reconcile.md @@ -0,0 +1,83 @@ +# 原生存储登记预检 + +`origin/dev` 基线：`ee732e74ce02f184586e1313b489e2198dbc15fe`。 + +本命令只补齐既有 `cf_library_storage` 合同，不搬移对象、不修改库 UUID。 +扫描 `Repo ∪ RepoTrash ∪ GarbageRepos`，跳过 `VirtualRepo`；库路由只取 +`RepoStorageId`，不读取、修改或删除旧表。物理对象健康、真实读写与 fsck 需要另行验收。 + +## 命令 + +连接凭据沿用 `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`；`--apply` 显式登记。`--batch-size` 范围为 1–2000。 +`--repo-id UUID` 可定向检查。输出是单个 JSON，不附加成功提示文本： + +```json +{"mode":"dry-run","summary":{"AUTO_LOCAL":1,"NATIVE":1,"ORPHAN_DIRECTORY":0},"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"},{"repo_id":"00000000-0000-4000-8000-000000000002","status":"AUTO_LOCAL","storage_id":"auto-local:old-library","category":"AUTO_LOCAL","repo_key":"old-library","state":"active"}],"orphan_directories":[]} +``` + +- 无原生路由：`NATIVE`，登记 `native-`。 +- `auto-local:`：`AUTO_LOCAL`，检查合法 key、真实目录及至少一个对象类型的库目录。 +- 本地存储类三个后端均指向原生 `seafile-data`：`NATIVE_CLASS_PINNED`，仅报告。 +- 已登记且路由、key、生命周期一致：`REGISTERED`，不改动。 +- 异常或不支持的存储类：`BLOCKED`，逐条附 `reason`。目录未被任何路由引用时为 + `ORPHAN_DIRECTORY`，不登记、不删除。虚拟库为 `VIRTUAL_REPO`，不登记。 + +正常退出 0；存在 `BLOCKED` 时退出 2；数据库、配置文件访问等执行错误退出 1，仍输出 JSON。 +存在于 `Repo` 或 `RepoTrash` 的库登记为 `active`；仅在 `GarbageRepos` 的库为 `retired`。 + +预检与 apply 都允许幂等执行与 Server `cloudfile-v2.sql` 完全一致的 +`CREATE TABLE IF NOT EXISTS cf_library_storage`，并验证既有结构；不会创建其他表。 +因此 dry-run 不写登记记录，但不是完全只读的数据库操作，需要该表的建表与临时表权限。 + +## 显式解除原生默认类固定路由 + +先检查报告，只有运维明确选定的原生类才传 `--normalize-native-class ID`；可重复传入多个 ID。 +`--dry-run` 保持 `NATIVE_CLASS_PINNED` 分类。应用时删除该库的 `RepoStorageId` 行与登记 +`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 前必须处理仍然存在的 `NATIVE_CLASS_PINNED`；普通 `--apply` 和启动时 +reconcile 均不会自动解除固定路由。配置读取优先使用 `[cloudfile_v2] auto_local_root`； +没有该项时，从存储类 JSON 中唯一的 `auto-local` 本地模板读取旧存储根。 +相对 `storage_classes_file` 与原生 Server 一样相对于 `seafile.conf` 的目录解析； +`CF_STORAGE_CLASSES_JSON` 与原生 Server 一样优先于文件。 + +每库一个事务，写入用 `INSERT ... SELECT` 复核原生生命周期、路由和虚拟库事实。 +现有记录和唯一键冲突不会被覆盖。已成功的其他库保留登记，命令可幂等重试。 +启动／Hub 管理入口使用相同规则，只自动处理 `NATIVE` 和 `AUTO_LOCAL`，其余记录返回报告并记日志。 + +## 本地回归 + +使用独立的临时 MariaDB 11.4，容器只对本机开放端口，密码仅为测试夹具： + +```sh +docker run --detach --name cf-storage-reconcile-test-20261020 \ + -e MARIADB_ROOT_PASSWORD=storage-test-only -p 127.0.0.1:13319:3306 mariadb:11.4 +CF_STORAGE_TEST_MYSQL_PORT=13319 python3 -m unittest \ + cloudfile_ext.v2.tests.test_storage_reconcile \ + cloudfile_ext.v2.tests.test_native_discovery -v +docker rm --force cf-storage-reconcile-test-20261020 +``` + +实测结果：`Ran 25 tests ... OK`，其中 16 项使用独立 MariaDB 模式验证；包括逐类分类、 +CLI 默认预检/显式应用、幂等、并发同库登记、唯一键冲突、规范化失败回滚、Garbage 生命周期、 +Virtual 跳过、旧表不读取以及 Hub 运行时共用规则。 + +未覆盖：生产或 DEV 数据副本的 1,407 自动库 / 63 默认库计数核对；真实 Server 上传、下载、 +历史版本、恢复、GC、fsck；三仓镜像完整构建。未执行 DEV 或生产部署。 From 31038afecb53f7866d4591058524bff17accabe0 Mon Sep 17 00:00:00 2001 From: dev9-bb Date: Tue, 20 Oct 2026 17:39:48 +0800 Subject: [PATCH 2/3] fix(storage): harden reconciliation route and path boundaries --- cloudfile_ext/v2/__main__.py | 4 +- cloudfile_ext/v2/storage_reconcile.py | 32 +++-- .../v2/tests/test_storage_reconcile.py | 109 +++++++++++++++++- docs/storage-reconcile.md | 24 ++++ 4 files changed, 158 insertions(+), 11 deletions(-) diff --git a/cloudfile_ext/v2/__main__.py b/cloudfile_ext/v2/__main__.py index ec991c8a27..ffca363f72 100644 --- a/cloudfile_ext/v2/__main__.py +++ b/cloudfile_ext/v2/__main__.py @@ -27,6 +27,8 @@ def main(argv=None): 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: @@ -65,7 +67,7 @@ def main(argv=None): print(json.dumps(discovery.summary(args.repo_id), sort_keys=True)) except Exception as error: if storage: - reason = str(error) if isinstance(error, ValueError) else type(error).__name__ + 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)) diff --git a/cloudfile_ext/v2/storage_reconcile.py b/cloudfile_ext/v2/storage_reconcile.py index 5644bc0b52..70745b5ac7 100644 --- a/cloudfile_ext/v2/storage_reconcile.py +++ b/cloudfile_ext/v2/storage_reconcile.py @@ -28,6 +28,11 @@ 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 @@ -58,7 +63,7 @@ def read(cls, conf, native_dir): if not path.is_absolute(): path = conf.parent / path classes = json.loads(path.read_text()) - if not isinstance(classes, list): + 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: @@ -71,7 +76,7 @@ def read(cls, conf, native_dir): 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 not real_directory(config.root)): + 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 @@ -89,7 +94,7 @@ def native_class(self, route): def auto_error(self, key, repo_id): if not valid_key(key): return 'INVALID_KEY' - if not self.root or not real_directory(self.root): + 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' @@ -112,6 +117,7 @@ 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 @@ -132,6 +138,13 @@ def ensure_schema(self): # Deliberately inspect only this reviewed table: unrelated old tables # remain present and are never enumerated, queried, or altered. 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(STORAGE_DDL.replace('CREATE TABLE cf_library_storage', 'CREATE TABLE IF NOT EXISTS ' + self.table('cf_library_storage'), 1)) expected = 'cfv2_expected_storage' @@ -171,7 +184,7 @@ 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')} - sql = ('SELECT r.repo_id,t.repo_id,g.repo_id,v.repo_id,s.storage_id,c.repo_key,c.state ' + 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 ' @@ -181,7 +194,7 @@ def inspect(self, repo_id, *, lock=False): '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, key, state = cursor.fetchone() + 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) @@ -189,6 +202,8 @@ def blocked(reason): 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') @@ -232,11 +247,12 @@ def apply_one(self, repo_id, normalize=()): if normalize_route: with self.connection.cursor() as cursor: cursor.execute('DELETE FROM ' + self.table('RepoStorageId') + - ' WHERE repo_id=%s AND storage_id=%s', + ' 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 " @@ -246,9 +262,9 @@ def apply_one(self, repo_id, normalize=()): '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 s.storage_id <=> %s FOR UPDATE') + '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, route)) + 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) diff --git a/cloudfile_ext/v2/tests/test_storage_reconcile.py b/cloudfile_ext/v2/tests/test_storage_reconcile.py index 5401e6b038..815f446019 100644 --- a/cloudfile_ext/v2/tests/test_storage_reconcile.py +++ b/cloudfile_ext/v2/tests/test_storage_reconcile.py @@ -24,7 +24,8 @@ def test_key_matches_c_storage_contract(self): def test_reads_native_and_legacy_config_and_rejects_invalid_root(self): with TemporaryDirectory() as tmp: - root = Path(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) @@ -41,6 +42,24 @@ def test_reads_native_and_legacy_config_and_rejects_invalid_root(self): 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_cli_default_dry_run_and_machine_readable_blocked(self): from cloudfile_ext.v2.__main__ import main output = io.StringIO() @@ -75,7 +94,7 @@ def setUp(self): 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) + self.root = Path(self.temp.name).resolve() self.native = self.root / 'seafile-data' self.native.mkdir() self.auto = self.root / 'libraries' @@ -314,6 +333,92 @@ def test_hub_startup_uses_same_rules_and_never_normalizes_classes(self): 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_clean_registry_is_created_before_cli_dry_run(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) + self.assertEqual(json.loads(output.getvalue())['summary']['NATIVE'], 1) + self.assertEqual(self.sql('SELECT * FROM cf_library_storage'), ()) + self.assertEqual(self.sql('SELECT * FROM cf_library_storage_key'), ((42,),)) + + 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') diff --git a/docs/storage-reconcile.md b/docs/storage-reconcile.md index 01eabe1b2f..ba18ba3c54 100644 --- a/docs/storage-reconcile.md +++ b/docs/storage-reconcile.md @@ -81,3 +81,27 @@ Virtual 跳过、旧表不读取以及 Hub 运行时共用规则。 未覆盖：生产或 DEV 数据副本的 1,407 自动库 / 63 默认库计数核对；真实 Server 上传、下载、 历史版本、恢复、GC、fsck；三仓镜像完整构建。未执行 DEV 或生产部署。 + +## 安全边界复审（2026-10-20） + +复审修复三处边界：存储根任何祖先组件为符号链接时拒绝；存在但值为 NULL 的 +`RepoStorageId` 为 `BLOCKED: INVALID_STORAGE_ROUTE`，不当作无路由；写入复核使用真实 +路由行不存在或字节精确匹配，避免数据库默认排序规则将不同大小写视为同一路由。 +规范化删除路由也使用字节精确匹配。旧 `reconcile` 等命令拒绝 storage 专用模式参数， +防止误传 `--dry-run` 后仍执行变更。建表前必须验证五张原生来源表完整，错误数据库 +以 `NATIVE_SEAFILE_DATABASE_NOT_CONNECTED` 拒绝且不创建登记表。 + +新增 10 项针对性回归，并复验 4 项受影响的并发/规范化/Hub 入口用例；分三次执行： +`Ran 11 tests ... OK`、`Ran 3 tests ... OK`、`Ran 1 test ... OK`（其中干净登记表预检重复验证一次）。 +覆盖根祖先、storage/type/repo 目录链符号链接、NULL 路由、写时大小写路由变化、 +Trash 与 Garbage 重叠 active、批次外全局路由对 ORPHAN 的保护，以及预检前不存在登记表。 + +另外使用独立临时数据库运行 11 个合成库、1 个孤儿目录的 CLI 验收，batch_size=2： + +```json +{"dry_run":{"NATIVE":3,"AUTO_LOCAL":3,"NATIVE_CLASS_PINNED":1,"BLOCKED":3,"VIRTUAL_REPO":1,"ORPHAN_DIRECTORY":1},"apply":{"REGISTERED":7,"BLOCKED":3,"VIRTUAL_REPO":1,"ORPHAN_DIRECTORY":1},"replay":{"REGISTERED":7,"BLOCKED":3,"VIRTUAL_REPO":1,"ORPHAN_DIRECTORY":1},"result":"PASS"} +``` + +三个预置 BLOCKED 分别为 S3、非法 key、登记冲突。显式规范化后新增 7 条登记，重放无写入， +垃圾库 retired、Trash+Garbage 重叠库 active，旧表哨兵保持不变。数据完全合成，不作为 +1,407/63 个真实库数量核对证据；该真实数据核对按用户最新要求留至发布后验收。 From 2a546363723e3c4aa24ddaa675d2d6c4213aa3c5 Mon Sep 17 00:00:00 2001 From: dev9-bb Date: Tue, 20 Oct 2026 17:50:19 +0800 Subject: [PATCH 3/3] fix(storage): keep preflight read-only and report lifecycle drift --- cloudfile_ext/v2/__main__.py | 3 +- cloudfile_ext/v2/native_discovery.py | 11 +- cloudfile_ext/v2/storage_reconcile.py | 131 +++++++++++++---- .../v2/tests/test_storage_reconcile.py | 99 ++++++++++++- docs/storage-reconcile.md | 138 +++++++++--------- 5 files changed, 268 insertions(+), 114 deletions(-) diff --git a/cloudfile_ext/v2/__main__.py b/cloudfile_ext/v2/__main__.py index ffca363f72..50666ff5eb 100644 --- a/cloudfile_ext/v2/__main__.py +++ b/cloudfile_ext/v2/__main__.py @@ -35,7 +35,7 @@ def main(argv=None): parser.error('--batch-size must be between 1 and 2000') import pymysql try: - config = StorageConfig.read(args.seafile_conf, args.seafile_data) if args.seafile_conf and args.seafile_data else None + 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'], @@ -44,7 +44,6 @@ def main(argv=None): charset='utf8mb4', autocommit=True) as connection: if storage: discovery = StorageReconciler(connection, config) - discovery.ensure_schema() 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)) diff --git a/cloudfile_ext/v2/native_discovery.py b/cloudfile_ext/v2/native_discovery.py index 96ef3997e5..c91b98942b 100644 --- a/cloudfile_ext/v2/native_discovery.py +++ b/cloudfile_ext/v2/native_discovery.py @@ -169,24 +169,19 @@ def reconcile_from_hub(repo_id=None, *, batch_size=500): Startup/admin discovery handles native and auto-local bindings only. Pinned classes and invalid bindings are reported without conversion. """ - import os - from pathlib import Path from django.db import connection, transaction from seahub.utils.db_api import SeafileDB from .storage_reconcile import StorageConfig, StorageReconciler - data = os.environ.get('SEAFILE_DATA_DIR', '') - conf = os.environ.get('SEAFILE_CENTRAL_CONF_DIR') or data - if not data or not conf: + config = StorageConfig.detect() + if config is None: raise RuntimeError('SEAFILE_STORAGE_CONFIGURATION_UNAVAILABLE') - config = StorageConfig.read(Path(conf) / 'seafile.conf', data) discovery = StorageReconciler(connection, config, schema=SeafileDB().db_name, atomic=transaction.atomic) - discovery.ensure_schema() 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'): + 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'] diff --git a/cloudfile_ext/v2/storage_reconcile.py b/cloudfile_ext/v2/storage_reconcile.py index 70745b5ac7..796317f497 100644 --- a/cloudfile_ext/v2/storage_reconcile.py +++ b/cloudfile_ext/v2/storage_reconcile.py @@ -80,16 +80,40 @@ def read(cls, conf, native_dir): 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 or not real_directory(self.native_dir): + 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 - return all(item.get(kind, {}).get('backend') == 'fs' and - item[kind].get('dir') and - Path(item[kind]['dir']).is_absolute() and - real_directory(Path(item[kind]['dir'])) and - Path(item[kind]['dir']).resolve() == self.native_dir.resolve() - for kind in KINDS) def auto_error(self, key, repo_id): if not valid_key(key): @@ -120,6 +144,7 @@ def __init__(self, connection, config=None, *, schema=None, atomic=None): 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) @@ -134,32 +159,65 @@ def _atomic(self): self.connection.rollback() raise - def ensure_schema(self): - # Deliberately inspect only this reviewed table: unrelated old tables - # remain present and are never enumerated, queried, or altered. + 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,)) + "('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(STORAGE_DDL.replace('CREATE TABLE cf_library_storage', - 'CREATE TABLE IF NOT EXISTS ' + self.table('cf_library_storage'), 1)) - expected = 'cfv2_expected_storage' - cursor.execute('DROP TEMPORARY TABLE IF EXISTS ' + expected) - try: + 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 TEMPORARY TABLE ' + expected, 1)) - cursor.execute('SHOW CREATE TABLE ' + self.table('cf_library_storage')) - actual = cursor.fetchone()[1] - cursor.execute('SHOW CREATE TABLE ' + expected) - wanted = cursor.fetchone()[1] - if actual[actual.index('('):] != wanted[wanted.index('('):]: - raise RuntimeError('STORAGE_SCHEMA_DRIFT') - finally: - cursor.execute('DROP TEMPORARY TABLE IF EXISTS ' + expected) + '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) @@ -184,6 +242,8 @@ 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 ' @@ -221,18 +281,22 @@ def blocked(reason): return blocked('UNSUPPORTED_STORAGE_CLASS') item.update(category=category, repo_key=wanted, state=target_state) with self.connection.cursor() as cursor: - 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 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 state != target_state or category == 'NATIVE_CLASS_PINNED': + 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) @@ -294,6 +358,7 @@ def orphans(self): 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: @@ -303,5 +368,7 @@ def run(self, *, apply=False, batch_size=500, normalize=(), repo_id=None): orphans = self.orphans() counts = Counter(item['status'] for item in items) counts['ORPHAN_DIRECTORY'] += len(orphans) - return dict(mode='apply' if apply else 'dry-run', summary=dict(sorted(counts.items())), + 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_storage_reconcile.py b/cloudfile_ext/v2/tests/test_storage_reconcile.py index 815f446019..bf87a99120 100644 --- a/cloudfile_ext/v2/tests/test_storage_reconcile.py +++ b/cloudfile_ext/v2/tests/test_storage_reconcile.py @@ -60,6 +60,17 @@ def test_legacy_commands_cannot_silently_ignore_dry_run(self): 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() @@ -190,7 +201,7 @@ def test_existing_binding_consistency_state_and_occupied_key(self): 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)['reason'], 'STORAGE_BINDING_CONFLICT') + 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') @@ -368,7 +379,7 @@ def test_orphan_references_include_routes_outside_the_selected_batch(self): self.assertEqual(len(report['items']), 1) self.assertEqual(report['orphan_directories'], []) - def test_clean_registry_is_created_before_cli_dry_run(self): + 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') @@ -380,9 +391,89 @@ def test_clean_registry_is_created_before_cli_dry_run(self): 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) - self.assertEqual(json.loads(output.getvalue())['summary']['NATIVE'], 1) - self.assertEqual(self.sql('SELECT * FROM cf_library_storage'), ()) + 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') diff --git a/docs/storage-reconcile.md b/docs/storage-reconcile.md index ba18ba3c54..9f17955a68 100644 --- a/docs/storage-reconcile.md +++ b/docs/storage-reconcile.md @@ -1,15 +1,15 @@ # 原生存储登记预检 -`origin/dev` 基线：`ee732e74ce02f184586e1313b489e2198dbc15fe`。 +Hub 基线：`ee732e74ce02f184586e1313b489e2198dbc15fe`。 -本命令只补齐既有 `cf_library_storage` 合同，不搬移对象、不修改库 UUID。 -扫描 `Repo ∪ RepoTrash ∪ GarbageRepos`，跳过 `VirtualRepo`；库路由只取 -`RepoStorageId`，不读取、修改或删除旧表。物理对象健康、真实读写与 fsck 需要另行验收。 +命令只补齐既有 `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 数据库，演练使用副本。 +连接凭据使用 `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 \ @@ -18,32 +18,44 @@ python -m cloudfile_ext.v2 storage-reconcile \ --dry-run --batch-size 500 ``` -默认是 `--dry-run`；`--apply` 显式登记。`--batch-size` 范围为 1–2000。 -`--repo-id UUID` 可定向检查。输出是单个 JSON，不附加成功提示文本： +默认 `--dry-run` **严格只读**：数据库只执行 SELECT/SHOW，不创建普通表、临时表或登记行。 +结构校验仅读取 information_schema 和 SHOW CREATE TABLE，SELECT 权限即可运行。 +登记表不存在时独立报告 `SCHEMA_MISSING`，并把登记关系视为空集继续分类；这不是业务 BLOCKED。 +不会创建数据库或原生表。原生五张来源表不完整时在任何建表前拒绝。 ```json -{"mode":"dry-run","summary":{"AUTO_LOCAL":1,"NATIVE":1,"ORPHAN_DIRECTORY":0},"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"},{"repo_id":"00000000-0000-4000-8000-000000000002","status":"AUTO_LOCAL","storage_id":"auto-local:old-library","category":"AUTO_LOCAL","repo_key":"old-library","state":"active"}],"orphan_directories":[]} +{"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":[]} ``` -- 无原生路由：`NATIVE`，登记 `native-`。 -- `auto-local:`：`AUTO_LOCAL`，检查合法 key、真实目录及至少一个对象类型的库目录。 -- 本地存储类三个后端均指向原生 `seafile-data`：`NATIVE_CLASS_PINNED`，仅报告。 -- 已登记且路由、key、生命周期一致：`REGISTERED`，不改动。 -- 异常或不支持的存储类：`BLOCKED`，逐条附 `reason`。目录未被任何路由引用时为 - `ORPHAN_DIRECTORY`，不登记、不删除。虚拟库为 `VIRTUAL_REPO`，不登记。 - -正常退出 0；存在 `BLOCKED` 时退出 2；数据库、配置文件访问等执行错误退出 1，仍输出 JSON。 -存在于 `Repo` 或 `RepoTrash` 的库登记为 `active`；仅在 `GarbageRepos` 的库为 `retired`。 - -预检与 apply 都允许幂等执行与 Server `cloudfile-v2.sql` 完全一致的 -`CREATE TABLE IF NOT EXISTS cf_library_storage`，并验证既有结构；不会创建其他表。 -因此 dry-run 不写登记记录，但不是完全只读的数据库操作，需要该表的建表与临时表权限。 +只有 `--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`；可重复传入多个 ID。 -`--dry-run` 保持 `NATIVE_CLASS_PINNED` 分类。应用时删除该库的 `RepoStorageId` 行与登记 -`native-` 位于同一个事务，任一步失败均回滚；S3、其他本地路径及混合后端不能转换。 +只有运维明确选择的类才传 `--normalize-native-class ID`，可重复指定。dry-run 仍报告 pinned； +普通 apply 和运行时 reconcile 不解除固定路由。显式规范化在每库同一事务中删除原生路由、登记 +`native-`，失败全部回滚。S3、其他本地路径与混合后端不可转换。 ```sh python -m cloudfile_ext.v2 storage-reconcile \ @@ -52,56 +64,46 @@ python -m cloudfile_ext.v2 storage-reconcile \ --apply --normalize-native-class legacy-default ``` -切换为 filesystem 前必须处理仍然存在的 `NATIVE_CLASS_PINNED`；普通 `--apply` 和启动时 -reconcile 均不会自动解除固定路由。配置读取优先使用 `[cloudfile_v2] auto_local_root`； -没有该项时，从存储类 JSON 中唯一的 `auto-local` 本地模板读取旧存储根。 -相对 `storage_classes_file` 与原生 Server 一样相对于 `seafile.conf` 的目录解析； -`CF_STORAGE_CLASSES_JSON` 与原生 Server 一样优先于文件。 +切换 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，其他项仅报告。 -每库一个事务，写入用 `INSERT ... SELECT` 复核原生生命周期、路由和虚拟库事实。 -现有记录和唯一键冲突不会被覆盖。已成功的其他库保留登记，命令可幂等重试。 -启动／Hub 管理入口使用相同规则，只自动处理 `NATIVE` 和 `AUTO_LOCAL`，其余记录返回报告并记日志。 +## 版本证据与验证 -## 本地回归 +Docker `origin/dev a7e8c2bf1e4e7630dfa2e54427c1c03d98ad667e` 的 +`deploy/compose/docker-compose.yml:27` 明确使用 `mariadb:11.4`；这是部署配方证据。 +远端设备未连接，不能据此宣称远端当前实际运行版本已核实。 -使用独立的临时 MariaDB 11.4，容器只对本机开放端口，密码仅为测试夹具： +隔离测试服务器实际 `SELECT VERSION()`： + +- `11.4.13-MariaDB-ubu2404`：与部署配方相同 11.4 系列。 +- `10.11.19-MariaDB-ubu2204`：追加兼容验证，不称为 DEV 同版。 ```sh -docker run --detach --name cf-storage-reconcile-test-20261020 \ - -e MARIADB_ROOT_PASSWORD=storage-test-only -p 127.0.0.1:13319:3306 mariadb:11.4 CF_STORAGE_TEST_MYSQL_PORT=13319 python3 -m unittest \ - cloudfile_ext.v2.tests.test_storage_reconcile \ - cloudfile_ext.v2.tests.test_native_discovery -v -docker rm --force cf-storage-reconcile-test-20261020 + 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 25 tests ... OK`，其中 16 项使用独立 MariaDB 模式验证；包括逐类分类、 -CLI 默认预检/显式应用、幂等、并发同库登记、唯一键冲突、规范化失败回滚、Garbage 生命周期、 -Virtual 跳过、旧表不读取以及 Hub 运行时共用规则。 - -未覆盖：生产或 DEV 数据副本的 1,407 自动库 / 63 默认库计数核对；真实 Server 上传、下载、 -历史版本、恢复、GC、fsck；三仓镜像完整构建。未执行 DEV 或生产部署。 +两版本均 `Ran 39 tests ... OK`，后续新增结构变更只读拒绝用例在两版本各单独通过， +并对用户“native retired 但仍在 Trash”的精确场景在两版本追加定向通过。 +包含 SELECT-only 账户在 `START TRANSACTION READ ONLY` 下的有表/缺表双场景、缺表只读分类后 +apply 安装、reserved 兼容、STATE_DRIFT 保持、原生 symlink 布局、写时路由复核、并发唯一键、 +规范化回滚、Trash/Garbage、Virtual、全局 ORPHAN 与旧表哨兵不变。 -## 安全边界复审（2026-10-20） +实现阶段两轮元数据检查失败分别来自 index_name 默认排序顺序与 MariaDB 对 SELECT-only 用户隐藏 +CHECK_CONSTRAINTS；现分别用确定排序与 SHOW CREATE 纯解析修正，最终回归通过。 +原生表结构漂移（约束、唯一键、排序规则、列默认值）只读拒绝，不改表。 -复审修复三处边界：存储根任何祖先组件为符号链接时拒绝；存在但值为 NULL 的 -`RepoStorageId` 为 `BLOCKED: INVALID_STORAGE_ROUTE`，不当作无路由；写入复核使用真实 -路由行不存在或字节精确匹配，避免数据库默认排序规则将不同大小写视为同一路由。 -规范化删除路由也使用字节精确匹配。旧 `reconcile` 等命令拒绝 storage 专用模式参数， -防止误传 `--dry-run` 后仍执行变更。建表前必须验证五张原生来源表完整，错误数据库 -以 `NATIVE_SEAFILE_DATABASE_NOT_CONNECTED` 拒绝且不创建登记表。 - -新增 10 项针对性回归，并复验 4 项受影响的并发/规范化/Hub 入口用例；分三次执行： -`Ran 11 tests ... OK`、`Ran 3 tests ... OK`、`Ran 1 test ... OK`（其中干净登记表预检重复验证一次）。 -覆盖根祖先、storage/type/repo 目录链符号链接、NULL 路由、写时大小写路由变化、 -Trash 与 Garbage 重叠 active、批次外全局路由对 ORPHAN 的保护，以及预检前不存在登记表。 - -另外使用独立临时数据库运行 11 个合成库、1 个孤儿目录的 CLI 验收，batch_size=2： - -```json -{"dry_run":{"NATIVE":3,"AUTO_LOCAL":3,"NATIVE_CLASS_PINNED":1,"BLOCKED":3,"VIRTUAL_REPO":1,"ORPHAN_DIRECTORY":1},"apply":{"REGISTERED":7,"BLOCKED":3,"VIRTUAL_REPO":1,"ORPHAN_DIRECTORY":1},"replay":{"REGISTERED":7,"BLOCKED":3,"VIRTUAL_REPO":1,"ORPHAN_DIRECTORY":1},"result":"PASS"} -``` +未覆盖：远端运行版本实测、真实 Server 上传/下载/历史/恢复/GC/fsck、完整三仓镜像验收。 +1,407/63 个真实库数量核对按用户最新要求留至发布后；本轮不等待副本，未部署 DEV 或生产。 -三个预置 BLOCKED 分别为 S3、非法 key、登记冲突。显式规范化后新增 7 条登记，重放无写入， -垃圾库 retired、Trash+Garbage 重叠库 active，旧表哨兵保持不变。数据完全合成，不作为 -1,407/63 个真实库数量核对证据；该真实数据核对按用户最新要求留至发布后验收。 +独立的 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/既有登记冲突，旧表哨兵不变；全部数据为合成数据。