From e6fcb45c0efc0be250289443cd5c1abf887c5935 Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:04:54 +0900 Subject: [PATCH 01/16] fix(worker): schedule enterprise catalog only when SSO directory is active --- cloudfile_ext/apps.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/cloudfile_ext/apps.py b/cloudfile_ext/apps.py index 0ee6525f0f..8ba9b2783b 100644 --- a/cloudfile_ext/apps.py +++ b/cloudfile_ext/apps.py @@ -59,7 +59,10 @@ def _register_directory_catalog(registry): """ from django.conf import settings - if not getattr(settings, 'CF_DIRECTORY_URL', ''): + # A Directory URL can also be present for startup-only admin roster + # synchronization. That configuration must not create a recurring job. + if (not getattr(settings, 'CF_ENABLE_SSO', False) or + not getattr(settings, 'CF_DIRECTORY_URL', '')): return interval = getattr(settings, 'CF_DIRECTORY_SYNC_INTERVAL_SECONDS', 600) if type(interval) is not int or not 60 <= interval <= 86400: From 263eba952527599e5b0c1a1fd47fdbe5cf4fecfe Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:04:59 +0900 Subject: [PATCH 02/16] fix(worker): schedule enterprise catalog only when SSO directory is active --- cloudfile_ext/v2/tests/test_directory_adapter.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/cloudfile_ext/v2/tests/test_directory_adapter.py b/cloudfile_ext/v2/tests/test_directory_adapter.py index 1b2624fa53..20ee1bd287 100644 --- a/cloudfile_ext/v2/tests/test_directory_adapter.py +++ b/cloudfile_ext/v2/tests/test_directory_adapter.py @@ -140,14 +140,19 @@ def test_directory_catalog_is_registered_at_ten_minutes_without_requests(self): from cloudfile_ext.apps import CloudFileConfig scheduled = Mock() with override_settings(CF_DIRECTORY_URL='https://directory.test', - CF_DIRECTORY_SYNC_INTERVAL_SECONDS=600): + CF_ENABLE_SSO=True, CF_DIRECTORY_SYNC_INTERVAL_SECONDS=600): CloudFileConfig._register_directory_catalog(scheduled) scheduled.register_periodic_task.assert_called_once() name, interval, fn = scheduled.register_periodic_task.call_args.args self.assertEqual((name, interval), ('directory-entities', 600)) self.assertEqual(fn.__name__, 'sync_entity_catalog') scheduled.reset_mock() - with override_settings(CF_DIRECTORY_URL=''): + # Startup-only enterprise admin integration may expose Directory credentials + # with SSO capability disabled. No periodic entity polling in that case. + with override_settings(CF_DIRECTORY_URL='https://directory.test', CF_ENABLE_SSO=False): + CloudFileConfig._register_directory_catalog(scheduled) + scheduled.register_periodic_task.assert_not_called() + with override_settings(CF_DIRECTORY_URL='', CF_ENABLE_SSO=True): CloudFileConfig._register_directory_catalog(scheduled) scheduled.register_periodic_task.assert_not_called() From 6936384d0416b27b6634cefaf24ba1f492cf5b25 Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:05:37 +0900 Subject: [PATCH 03/16] feat(worker): one scheduler lock and aggregate task outcomes, no extra service --- .../management/commands/cf_worker.py | 53 ++++++++++++++----- 1 file changed, 40 insertions(+), 13 deletions(-) diff --git a/cloudfile_ext/management/commands/cf_worker.py b/cloudfile_ext/management/commands/cf_worker.py index 8f4a14ee75..5c05641f77 100644 --- a/cloudfile_ext/management/commands/cf_worker.py +++ b/cloudfile_ext/management/commands/cf_worker.py @@ -13,11 +13,13 @@ enabled periodic task this command exits 0 immediately. """ +import fcntl import logging +import os import signal import time -from django.core.management.base import BaseCommand +from django.core.management.base import BaseCommand, CommandError from cloudfile_ext.registry import registry @@ -44,23 +46,48 @@ def handle(self, *args, **options): 'Enable a capability that needs one before running cf_worker.') return - self.stdout.write('Running %d CloudFile task(s): %s' % ( - len(tasks), ', '.join(t['name'] for t in tasks))) - - if options['once']: - for task in tasks: - self._run(task) - return - - self._loop(tasks) + # One scheduler for a mounted Seafile instance, not one per web + # process or compose replica. Docker passes a persistent /shared path. + # Local test environments can use the /tmp default without a volume. + path = os.environ.get('CF_WORKER_LOCK_FILE', '/tmp/cloudfile-cf-worker.lock') + try: + with open(path, 'a') as lock: + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + self.stdout.write('CF_WORKER_ALREADY_RUNNING: skip duplicate scheduler') + return + self.stdout.write('Running %d CloudFile task(s): %s' % ( + len(tasks), ', '.join(t['name'] for t in tasks))) + if options['once']: + for task in tasks: + self._run(task) + return + self._loop(tasks) + except OSError as error: + raise CommandError('CF_WORKER_LOCK_UNAVAILABLE: %s' % type(error).__name__) from error def _run(self, task): + started = time.monotonic() try: - task['func']() + result = task['func']() except Exception: - # One failing task must not take the worker down with it; the next - # tick retries. + # One failing task does not kill the process or trigger a storm + # of immediate retries: it runs again at its configured cadence. logger.exception('CloudFile task %s failed', task['name']) + self.stderr.write('CF_TASK name=%s status=FAILED duration_ms=%d' % ( + task['name'], int((time.monotonic() - started) * 1000))) + return + # Only expose aggregate counts. Never log provider UIDs, credentials + # or organization/role records returned by arbitrary task callbacks. + counters = [] + if isinstance(result, dict): + for key in ('entities', 'created', 'renamed', 'unchanged', 'membership_writes'): + if type(result.get(key)) is int and result[key] >= 0: + counters.append('%s=%d' % (key, result[key])) + self.stdout.write('CF_TASK name=%s status=OK duration_ms=%d%s' % ( + task['name'], int((time.monotonic() - started) * 1000), + (' ' + ' '.join(counters)) if counters else '')) def _loop(self, tasks): stopping = {'now': False} From 910367e40cd70b97180b8af9958b8ca00ae16f9b Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:07:23 +0900 Subject: [PATCH 04/16] perf(search): bounded resumable first index per library and configured V2 interval --- cloudfile_ext/v2/search/worker.py | 125 +++++++++++++++++++++++++++--- 1 file changed, 113 insertions(+), 12 deletions(-) diff --git a/cloudfile_ext/v2/search/worker.py b/cloudfile_ext/v2/search/worker.py index b128538a00..c625cd6ede 100644 --- a/cloudfile_ext/v2/search/worker.py +++ b/cloudfile_ext/v2/search/worker.py @@ -111,6 +111,64 @@ def scan(self, repo, root): break self.publish(documents) + def scan_slice(self, repo, progress, max_pages=10): + """Index at most max_pages native directory windows from one fixed HEAD. + + The caller checkpoints the resulting stack only after Meili confirms + every upsert. Replaying a batch after a crash is safe: doc IDs are + deterministic. No whole-library traversal in one background tick. + """ + head, pending = progress.get('head'), progress.get('pending') + if (not isinstance(head, str) or len(head) != 40 or + any(c not in '0123456789abcdef' for c in head) or + not isinstance(pending, list) or len(pending) > 10000 or + type(max_pages) is not int or not 1 <= max_pages <= 100): + raise SearchUnavailable('invalid initial index checkpoint') + scanned = 0 + while pending and scanned < max_pages: + frame = pending[-1] + if not isinstance(frame, dict) or set(frame) != {'path', 'offset'}: + raise SearchUnavailable('invalid initial index frame') + parent, offset = normalize_path(frame['path']), frame['offset'] + if type(offset) is not int or not 0 <= offset <= 2147483447: + raise SearchUnavailable('invalid initial index offset') + entries = self.native.list_dir_by_commit_and_path(repo.id, head, parent, offset, 200) + if entries is None or len(entries) > 200: + raise SearchUnavailable('native directory snapshot unavailable') + documents, subdirs, seen = [], [], set() + refs = [] + for entry in entries: + name = getattr(entry, 'obj_name', None) + if (not isinstance(name, str) or not name or name in ('.', '..') or + '/' in name or '\\' in name): + raise SearchUnavailable('invalid native directory name') + if name in seen: + raise SearchUnavailable('duplicate native entry name') + seen.add(name) + child = normalize_path(posixpath.join(parent, name)) + if stat.S_ISDIR(entry.mode): + subdirs.append(child) + kind = 'dir' + elif stat.S_ISREG(entry.mode): + kind = 'file' + else: + raise SearchUnavailable('unsupported native object') + refs.append((child, kind)) + tags = tag_labels(repo.id, refs) + for entry, (child, _) in zip(entries, refs): + documents.append(self.upsert(repo, child, entry, {child: [None, tags[child]]})) + self.publish(documents) + # Mutate the stack only after the async Meilisearch task is + # confirmed. On failure the next run can safely repeat the page. + pending.pop() + if len(entries) == 200: + pending.append(dict(path=parent, offset=offset + 200)) + pending.extend(dict(path=child, offset=0) for child in reversed(subdirs)) + if len(pending) > 10000: + raise SearchUnavailable('initial index traversal too wide') + scanned += 1 + return not pending + def refresh(self, repo, path, subtree=True): entry = self.native.get_dirent_by_path(repo.id, path) if entry is None: @@ -206,12 +264,28 @@ def reconcile(self, repo, saved, bootstrap_before): return dict(head=repo.head_cmmt_id, tag_cursor=cursor) -def tick(client, native, checkpoint, bootstrap_before=0): +def tick(client, native, checkpoint, bootstrap_before=0, pages_per_repo=10): + """One bounded pass across repos, with a durable initial-scan cursor. + + Existing libraries retain the commit-diff/tag-sweep mechanism. New large + libraries get at most pages_per_repo directory windows per round, resuming + at the exact pinned commit even after a worker restart. + """ from django.db import close_old_connections close_old_connections() + if type(pages_per_repo) is not int or not 1 <= pages_per_repo <= 100: + raise SearchUnavailable('invalid indexing page budget') path = Path(checkpoint) state = json.loads(path.read_text()) if path.exists() else {} + if not isinstance(state, dict): + raise SearchUnavailable('invalid search checkpoint') indexer, offset, present = Indexer(client, native), 0, set() + + def persist(): + temporary = path.with_suffix('.tmp') + temporary.write_text(json.dumps(state, ensure_ascii=False)) + temporary.replace(path) + while True: repos = native.get_repo_list(offset, 100) if repos is None: @@ -221,22 +295,40 @@ def tick(client, native, checkpoint, bootstrap_before=0): if repo.is_virtual or repo.encrypted: continue try: - state[repo.id] = indexer.reconcile(repo, state.get(repo.id), bootstrap_before) - temporary = path.with_suffix('.tmp') - temporary.write_text(json.dumps(state, ensure_ascii=False)) - temporary.replace(path) + saved = state.get(repo.id) + if saved is not None and not isinstance(saved, dict): + raise SearchUnavailable('invalid per-library index checkpoint') + if not saved or (not saved.get('head') and 'initial' not in saved): + old_head = indexer.baseline(repo, bootstrap_before) + if old_head is None: + saved = {'initial': {'head': repo.head_cmmt_id, + 'pending': [{'path': '/', 'offset': 0}]}} + state[repo.id] = saved + persist() # Must survive a crash before the first page. + else: + saved = {'head': old_head, 'tag_cursor': ''} + if 'initial' in saved: + initial = saved['initial'] + complete = indexer.scan_slice(repo, initial, pages_per_repo) + if complete: + # The scan used this fixed historic commit. If HEAD + # advanced during it, next tick catches up by diff. + state[repo.id] = dict(head=initial['head'], tag_cursor='') + persist() + continue + state[repo.id] = indexer.reconcile(repo, saved, bootstrap_before) + persist() except Exception: logger.exception('Search index catch-up failed for library %s; checkpoint retained', repo.id) offset += len(repos) if len(repos) < 100: break - # Removal only follows a complete successful native library enumeration. + # Only a complete native library enumeration can certify deletion. for repo_id in state.keys() - present: indexer.remove(repo_id, '/') del state[repo_id] - temporary = path.with_suffix('.tmp') - temporary.write_text(json.dumps(state, ensure_ascii=False)) - temporary.replace(path) + persist() + def main(): @@ -251,16 +343,25 @@ def main(): args = parser.parse_args() logging.basicConfig(level=logging.INFO) os.umask(0o077) + from django.conf import settings + interval = max(15, int(getattr(settings, 'CF_SEARCH_INDEX_INTERVAL', 60))) + pages = int(os.environ.get('CF_SEARCH_BOOTSTRAP_PAGES_PER_TICK', '10')) + if not 1 <= pages <= 100: + raise ValueError('CF_SEARCH_BOOTSTRAP_PAGES_PER_TICK must be 1..100') with open(args.checkpoint + '.lock', 'w') as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) while True: + started = time.monotonic() try: - tick(configured_client(), seafile_api, args.checkpoint, args.bootstrap_before) + tick(configured_client(), seafile_api, args.checkpoint, + args.bootstrap_before, pages_per_repo=pages) + logger.info('CF_SEARCH_INDEX status=OK duration_ms=%d', + int((time.monotonic() - started) * 1000)) except Exception: - logger.exception('Search index worker unavailable; file services remain independent') + logger.exception('CF_SEARCH_INDEX status=FAILED; file services remain independent') if args.once: break - time.sleep(60) + time.sleep(interval) if __name__ == '__main__': From 1272652fdac5e29d5411c86ea40e87c285fb4af5 Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:10:39 +0900 Subject: [PATCH 05/16] test(worker): bounded pinned-HEAD recovery, retry, scheduler registration --- .../tests/test_search_worker_schedule_pure.py | 111 ++++++++++++++++++ 1 file changed, 111 insertions(+) create mode 100644 cloudfile_ext/v2/tests/test_search_worker_schedule_pure.py diff --git a/cloudfile_ext/v2/tests/test_search_worker_schedule_pure.py b/cloudfile_ext/v2/tests/test_search_worker_schedule_pure.py new file mode 100644 index 0000000000..71b81cdd8c --- /dev/null +++ b/cloudfile_ext/v2/tests/test_search_worker_schedule_pure.py @@ -0,0 +1,111 @@ +"""Cheap real indexing-cursor regressions, no Seafile/Redis/Meilisearch service.""" +import json +import os +import stat +import sys +import tempfile +import types +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from cloudfile_ext.v2.search.worker import Indexer, SearchUnavailable, tick + +RID = "a6fbc97c-1000-4000-8000-010101010101" +HEAD_A = "a" * 40 +HEAD_B = "b" * 40 + + +class InitialIndexTests(unittest.TestCase): + def setUp(self): + self.repo = SimpleNamespace(id=RID, head_cmmt_id=HEAD_A, + is_virtual=False, encrypted=False) + self.calls = [] + self.first = [SimpleNamespace(obj_name="file-%03d.txt" % n, + mode=stat.S_IFREG | 0o644, size=100, mtime=1) + for n in range(200)] + self.last = [SimpleNamespace(obj_name="final.txt", + mode=stat.S_IFREG | 0o644, size=100, mtime=1)] + + def page(repo, head, parent, offset, limit): + self.calls.append((repo, head, parent, offset, limit)) + self.assertEqual((repo, parent, limit), (RID, "/", 200)) + return self.first if offset == 0 else self.last if offset == 200 else [] + + self.native = Mock() + self.native.list_dir_by_commit_and_path.side_effect = page + self.native.get_repo_list.side_effect = lambda offset, limit: [self.repo] if offset == 0 else [] + self.native.get_repo.side_effect = lambda repo: self.repo + self.native.diff_commits.return_value = [] + self.client = Mock(index_path="/indexes/cloudfile_files") + self.client.call.return_value = {"results": []} + self.tags = patch('cloudfile_ext.v2.search.worker.tag_labels', + side_effect=lambda rid, refs: {name: [] for name, kind in refs}) + self.publish = patch.object(Indexer, 'publish', return_value=None) + self.tags.start() + self.publish_mock = self.publish.start() + self.addCleanup(self.tags.stop) + self.addCleanup(self.publish.stop) + + def test_slice_is_bounded_and_mutates_cursor_only_after_publish(self): + worker = Indexer(self.client, self.native) + initial = {"head": HEAD_A, "pending": [{"path": "/", "offset": 0}]} + self.assertFalse(worker.scan_slice(self.repo, initial, max_pages=1)) + self.assertEqual(initial['pending'], [{"path": "/", "offset": 200}]) + self.assertEqual(len(self.calls), 1) + self.assertTrue(worker.scan_slice(self.repo, initial, max_pages=1)) + self.assertEqual(initial['pending'], []) + self.assertEqual(len(self.calls), 2) + self.assertEqual(self.publish_mock.call_count, 2) + + def test_meilisearch_failure_does_not_advance_the_cursor(self): + self.publish_mock.side_effect = SearchUnavailable('Meili failed') + progress = {"head": HEAD_A, "pending": [{"path": "/", "offset": 0}]} + with self.assertRaises(SearchUnavailable): + Indexer(self.client, self.native).scan_slice(self.repo, progress, max_pages=1) + self.assertEqual(progress["pending"], [{"path": "/", "offset": 0}]) + + def test_restart_resumes_pinned_head_then_catches_up(self): + django = types.ModuleType("django") + db = types.ModuleType("django.db") + db.close_old_connections = lambda: None + django.db = db + with tempfile.TemporaryDirectory() as directory, patch.dict( + sys.modules, {'django': django, 'django.db': db}): + checkpoint = str(Path(directory) / "index.json") + tick(self.client, self.native, checkpoint, pages_per_repo=1) + saved = json.loads(Path(checkpoint).read_text()) + self.assertEqual(saved[RID]['initial']['pending'], + [{"path": "/", "offset": 200}]) + + self.repo.head_cmmt_id = HEAD_B + tick(self.client, self.native, checkpoint, pages_per_repo=1) + saved = json.loads(Path(checkpoint).read_text()) + self.assertEqual(saved[RID]['head'], HEAD_A) + self.assertNotIn('initial', saved[RID]) + self.assertEqual({head for _, head, *_ in self.calls}, {HEAD_A}) + + with patch.object(Indexer, "sweep", return_value=""): + tick(self.client, self.native, checkpoint, pages_per_repo=1) + saved = json.loads(Path(checkpoint).read_text()) + self.assertEqual(saved[RID]['head'], HEAD_B) + self.native.diff_commits.assert_called_with(RID, HEAD_A, HEAD_B, 1) + + +class WorkerRegistrationSourceTests(unittest.TestCase): + def test_directory_is_not_registered_for_admin_only_startup(self): + source = (Path(__file__).resolve().parents[2] / 'apps.py').read_text() + self.assertIn("getattr(settings, 'CF_ENABLE_SSO', False)", source) + self.assertIn("getattr(settings, 'CF_DIRECTORY_URL', '')", source) + + def test_scheduler_has_one_lock_and_aggregate_logs(self): + source = (Path(__file__).resolve().parents[2] / + 'management/commands/cf_worker.py').read_text() + self.assertIn('CF_WORKER_LOCK_FILE', source) + self.assertIn('fcntl.LOCK_EX | fcntl.LOCK_NB', source) + self.assertIn('CF_TASK name=%s status=OK duration_ms=%d', source) + + +if __name__ == '__main__': + unittest.main() From 7711243e5c4d4ad673145ecef2fc51ce440a6342 Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:10:59 +0900 Subject: [PATCH 06/16] test(ci): run cheap cursor-resume and worker singleton regressions --- .github/workflows/quick.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/quick.yml b/.github/workflows/quick.yml index 1c184839ed..f87b7e3ccb 100644 --- a/.github/workflows/quick.yml +++ b/.github/workflows/quick.yml @@ -65,6 +65,9 @@ jobs: - name: Check CE and UUID V2 route diagnostics (pure Python) run: python3 -m unittest cloudfile_ext.v2.tests.test_native_discovery -v + - name: Verify bounded search indexing and single-worker source contract + run: python3 -m unittest cloudfile_ext.v2.tests.test_search_worker_schedule_pure -v + - name: Validate only changed source shell: bash run: | From 21053a0b1350ab617cb479cdf387fb1e96629f3b Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:12:05 +0900 Subject: [PATCH 07/16] docs: directory catalog scheduler requires active SSO and trusted directory --- docs/ENTERPRISE-DIRECTORY-CONTRACT.md | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/docs/ENTERPRISE-DIRECTORY-CONTRACT.md b/docs/ENTERPRISE-DIRECTORY-CONTRACT.md index 6065b39ac6..b8954a4bf3 100644 --- a/docs/ENTERPRISE-DIRECTORY-CONTRACT.md +++ b/docs/ENTERPRISE-DIRECTORY-CONTRACT.md @@ -67,9 +67,11 @@ adapter, without inventing a framework or adding cf_* tables. ## Runtime organization/role synchronization (2026-10-25) -- Configured enterprise Directory: existing `cf_worker` registers - `directory-entities` every `CF_DIRECTORY_SYNC_INTERVAL_SECONDS` seconds, - default 600 (10 minutes, supported 60–86400). The worker must be running. +- With `CF_ENABLE_SSO=true` **and** a trusted `CF_DIRECTORY_URL`, the + existing `cf_worker` registers `directory-entities` every + `CF_DIRECTORY_SYNC_INTERVAL_SECONDS` seconds, default 600 (10 minutes, + supported 60–86400). A URL configured solely for the startup administrator + roster does **not** register periodic polling. The worker must be running. - Periodic sync reconciles organization/role mappings and names, not users or individual GroupUser memberships. Unchanged groups are read in one snapshot rather than each acquiring a lock and native RPC. From 715eff2a1f4d0df47888288ec2352972217990b8 Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:42:38 +0900 Subject: [PATCH 08/16] fix(directory): remove CloudFile-initiated org/role polling; preserve external API --- cloudfile_ext/apps.py | 23 ++--------------------- 1 file changed, 2 insertions(+), 21 deletions(-) diff --git a/cloudfile_ext/apps.py b/cloudfile_ext/apps.py index 8ba9b2783b..49590f4d96 100644 --- a/cloudfile_ext/apps.py +++ b/cloudfile_ext/apps.py @@ -28,7 +28,8 @@ def ready(self): module.register(registry) self._register_native_acl_gate(registry) - self._register_directory_catalog(registry) + # Org/role synchronization is triggered by the owning external app + # through the explicit admin API; never register periodic polling here. registry.seal() @@ -50,26 +51,6 @@ def _register_native_acl_gate(registry): from cloudfile_ext.v2 import native_gate registry.register_permission_check(native_gate.check_permission) - @staticmethod - def _register_directory_catalog(registry): - """Sync enterprise org/role catalog on the existing worker, never on ACL reads. - - The login callback projects an authenticated existing user's memberships; - a missing group or user in an ordinary file request does not create it. - """ - from django.conf import settings - - # A Directory URL can also be present for startup-only admin roster - # synchronization. That configuration must not create a recurring job. - if (not getattr(settings, 'CF_ENABLE_SSO', False) or - not getattr(settings, 'CF_DIRECTORY_URL', '')): - return - interval = getattr(settings, 'CF_DIRECTORY_SYNC_INTERVAL_SECONDS', 600) - if type(interval) is not int or not 60 <= interval <= 86400: - raise ValueError('CF_DIRECTORY_SYNC_INTERVAL_SECONDS must be 60..86400') - from cloudfile_ext.v2.directory_adapter import sync_entity_catalog - registry.register_periodic_task('directory-entities', interval, sync_entity_catalog) - def _configure_database(self): """Add the seafile-db connection and the cf_* router. From d980c4a95188bfd5111e30cff7d1acedd52d7301 Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:42:59 +0900 Subject: [PATCH 09/16] test(directory): external-only catalog synchronization with no scheduled registration --- .../v2/tests/test_directory_adapter.py | 28 ++++++------------- 1 file changed, 8 insertions(+), 20 deletions(-) diff --git a/cloudfile_ext/v2/tests/test_directory_adapter.py b/cloudfile_ext/v2/tests/test_directory_adapter.py index 20ee1bd287..3c37d538c9 100644 --- a/cloudfile_ext/v2/tests/test_directory_adapter.py +++ b/cloudfile_ext/v2/tests/test_directory_adapter.py @@ -136,27 +136,15 @@ def test_directory_request_is_short_lived_scoped_and_does_not_follow_redirect(se with self.assertRaises(IdentityUnavailable):Directory().get('/groups?view=entities') - def test_directory_catalog_is_registered_at_ten_minutes_without_requests(self): - from cloudfile_ext.apps import CloudFileConfig - scheduled = Mock() - with override_settings(CF_DIRECTORY_URL='https://directory.test', - CF_ENABLE_SSO=True, CF_DIRECTORY_SYNC_INTERVAL_SECONDS=600): - CloudFileConfig._register_directory_catalog(scheduled) - scheduled.register_periodic_task.assert_called_once() - name, interval, fn = scheduled.register_periodic_task.call_args.args - self.assertEqual((name, interval), ('directory-entities', 600)) - self.assertEqual(fn.__name__, 'sync_entity_catalog') - scheduled.reset_mock() - # Startup-only enterprise admin integration may expose Directory credentials - # with SSO capability disabled. No periodic entity polling in that case. - with override_settings(CF_DIRECTORY_URL='https://directory.test', CF_ENABLE_SSO=False): - CloudFileConfig._register_directory_catalog(scheduled) - scheduled.register_periodic_task.assert_not_called() - with override_settings(CF_DIRECTORY_URL='', CF_ENABLE_SSO=True): - CloudFileConfig._register_directory_catalog(scheduled) - scheduled.register_periodic_task.assert_not_called() + def test_external_catalog_sync_remains_available_without_scheduler(self): + # Nothing in CloudFile's Django AppConfig creates an org/role polling job. + # Etech PowerJob and other integrations call CatalogSync explicitly. + from pathlib import Path + source = (Path(__file__).resolve().parents[2] / 'apps.py').read_text() + self.assertNotIn('register_periodic_task', source) + self.assertNotIn('_register_directory_catalog', source) - def test_periodic_entity_sync_updates_only_native_catalog(self): + def test_externally_triggered_entity_sync_updates_only_native_catalog(self): from cloudfile_ext.v2.directory_adapter import sync_entity_catalog def fake_native(group_id): return SimpleNamespace(group_name='Department') From a56ac67624d01880b6498c2c1f11025671673a78 Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:43:18 +0900 Subject: [PATCH 10/16] test(worker): assert external-only catalog and independent search indexing --- .../tests/test_search_worker_schedule_pure.py | 24 +++++++++++-------- 1 file changed, 14 insertions(+), 10 deletions(-) diff --git a/cloudfile_ext/v2/tests/test_search_worker_schedule_pure.py b/cloudfile_ext/v2/tests/test_search_worker_schedule_pure.py index 71b81cdd8c..deafb15ac1 100644 --- a/cloudfile_ext/v2/tests/test_search_worker_schedule_pure.py +++ b/cloudfile_ext/v2/tests/test_search_worker_schedule_pure.py @@ -94,17 +94,21 @@ def test_restart_resumes_pinned_head_then_catches_up(self): class WorkerRegistrationSourceTests(unittest.TestCase): - def test_directory_is_not_registered_for_admin_only_startup(self): + def test_directory_catalog_has_no_cloudfile_scheduler(self): source = (Path(__file__).resolve().parents[2] / 'apps.py').read_text() - self.assertIn("getattr(settings, 'CF_ENABLE_SSO', False)", source) - self.assertIn("getattr(settings, 'CF_DIRECTORY_URL', '')", source) - - def test_scheduler_has_one_lock_and_aggregate_logs(self): - source = (Path(__file__).resolve().parents[2] / - 'management/commands/cf_worker.py').read_text() - self.assertIn('CF_WORKER_LOCK_FILE', source) - self.assertIn('fcntl.LOCK_EX | fcntl.LOCK_NB', source) - self.assertIn('CF_TASK name=%s status=OK duration_ms=%d', source) + self.assertNotIn('directory-entities', source) + self.assertNotIn('register_periodic_task', source) + self.assertNotIn('_register_directory_catalog', source) + + def test_external_admin_api_keeps_entity_sync(self): + source = (Path(__file__).resolve().parents[2] / 'http.py').read_text() + self.assertIn('class CatalogSync(', source) + self.assertIn('result = sync_entity_catalog()', source) + + def test_v2_search_stays_independent(self): + source = (Path(__file__).resolve().parents[1] / 'search/worker.py').read_text() + self.assertIn('def scan_slice(', source) + self.assertIn('time.sleep(interval)', source) if __name__ == '__main__': From 0007913a2f3fff425fe72385d54deb59bc27ce85 Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:44:04 +0900 Subject: [PATCH 11/16] docs(directory): stop internal org polling and retain external-triggered catalog API --- .../management/commands/cf_worker.py | 63 ++++++------------- 1 file changed, 18 insertions(+), 45 deletions(-) diff --git a/cloudfile_ext/management/commands/cf_worker.py b/cloudfile_ext/management/commands/cf_worker.py index 5c05641f77..5192b1c98f 100644 --- a/cloudfile_ext/management/commands/cf_worker.py +++ b/cloudfile_ext/management/commands/cf_worker.py @@ -6,20 +6,18 @@ the cf-worker container runs this command from the same image as the application. -Registration happens in ``CloudFileConfig.ready()``. Enterprise Directory -org/role catalog synchronization uses this worker at a default 600-second -interval. It never provisions users or GroupUser during permission checks. -Login remains the existing user membership synchronization point. With no -enabled periodic task this command exits 0 immediately. +Registration happens in ``CloudFileConfig.ready()`` for optional features. +Enterprise organization/role synchronization is NOT scheduled here: the external +source application explicitly invokes the existing administrator catalog API. +With no registered periodic tasks this command exits 0 immediately and Docker +does not start it as part of the default stack. """ -import fcntl import logging -import os import signal import time -from django.core.management.base import BaseCommand, CommandError +from django.core.management.base import BaseCommand from cloudfile_ext.registry import registry @@ -46,48 +44,23 @@ def handle(self, *args, **options): 'Enable a capability that needs one before running cf_worker.') return - # One scheduler for a mounted Seafile instance, not one per web - # process or compose replica. Docker passes a persistent /shared path. - # Local test environments can use the /tmp default without a volume. - path = os.environ.get('CF_WORKER_LOCK_FILE', '/tmp/cloudfile-cf-worker.lock') - try: - with open(path, 'a') as lock: - try: - fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) - except BlockingIOError: - self.stdout.write('CF_WORKER_ALREADY_RUNNING: skip duplicate scheduler') - return - self.stdout.write('Running %d CloudFile task(s): %s' % ( - len(tasks), ', '.join(t['name'] for t in tasks))) - if options['once']: - for task in tasks: - self._run(task) - return - self._loop(tasks) - except OSError as error: - raise CommandError('CF_WORKER_LOCK_UNAVAILABLE: %s' % type(error).__name__) from error + self.stdout.write('Running %d CloudFile task(s): %s' % ( + len(tasks), ', '.join(t['name'] for t in tasks))) + + if options['once']: + for task in tasks: + self._run(task) + return + + self._loop(tasks) def _run(self, task): - started = time.monotonic() try: - result = task['func']() + task['func']() except Exception: - # One failing task does not kill the process or trigger a storm - # of immediate retries: it runs again at its configured cadence. + # One failing task must not take the worker down with it; the next + # tick retries. logger.exception('CloudFile task %s failed', task['name']) - self.stderr.write('CF_TASK name=%s status=FAILED duration_ms=%d' % ( - task['name'], int((time.monotonic() - started) * 1000))) - return - # Only expose aggregate counts. Never log provider UIDs, credentials - # or organization/role records returned by arbitrary task callbacks. - counters = [] - if isinstance(result, dict): - for key in ('entities', 'created', 'renamed', 'unchanged', 'membership_writes'): - if type(result.get(key)) is int and result[key] >= 0: - counters.append('%s=%d' % (key, result[key])) - self.stdout.write('CF_TASK name=%s status=OK duration_ms=%d%s' % ( - task['name'], int((time.monotonic() - started) * 1000), - (' ' + ' '.join(counters)) if counters else '')) def _loop(self, tasks): stopping = {'now': False} From 3721421fda9a4b7786aac9788fd971c7b647c10d Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:44:09 +0900 Subject: [PATCH 12/16] docs(directory): stop internal org polling and retain external-triggered catalog API --- cloudfile_ext/v2/directory_adapter.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cloudfile_ext/v2/directory_adapter.py b/cloudfile_ext/v2/directory_adapter.py index c2f7bbecd7..7f2fefec9b 100644 --- a/cloudfile_ext/v2/directory_adapter.py +++ b/cloudfile_ext/v2/directory_adapter.py @@ -232,7 +232,7 @@ def ensure_subject(uid, username): """Login-only refresh for an already authenticated/created native account. Never create missing groups as a side effect of user login or ACL lookup. - The periodic catalog synchronizer owns organization/role creation; a + External admin-triggered catalog synchronization owns organization/role creation; a missing mapped entity refuses membership projection until the next sync. """ if subject_ready(uid, username): @@ -242,7 +242,7 @@ def ensure_subject(uid, username): def sync_entity_catalog(): - """Scheduled/manual org+role reconciliation; no user or GroupUser writes. + """Externally triggered org+role reconciliation; no user or GroupUser writes. Read current native mappings once. Unchanged thousands of groups must not each acquire a provider lock, query CE and call ccnet RPC every ten minutes. From 025601a25412f8f35392df2551d48f05df3cd430 Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:44:14 +0900 Subject: [PATCH 13/16] docs(directory): stop internal org polling and retain external-triggered catalog API --- docs/ENTERPRISE-DIRECTORY-CONTRACT.md | 56 ++++++++++++++------------- 1 file changed, 30 insertions(+), 26 deletions(-) diff --git a/docs/ENTERPRISE-DIRECTORY-CONTRACT.md b/docs/ENTERPRISE-DIRECTORY-CONTRACT.md index b8954a4bf3..49a134f48c 100644 --- a/docs/ENTERPRISE-DIRECTORY-CONTRACT.md +++ b/docs/ENTERPRISE-DIRECTORY-CONTRACT.md @@ -65,33 +65,37 @@ described in #61. This PR does **not** create generic network routing. Aim: a provider-neutral, tiny contract compatible with the existing enterprise Directory adapter, without inventing a framework or adding cf_* tables. -## Runtime organization/role synchronization (2026-10-25) +## External organization/role synchronization (2026-10-25) -- With `CF_ENABLE_SSO=true` **and** a trusted `CF_DIRECTORY_URL`, the - existing `cf_worker` registers `directory-entities` every - `CF_DIRECTORY_SYNC_INTERVAL_SECONDS` seconds, default 600 (10 minutes, - supported 60–86400). A URL configured solely for the startup administrator - roster does **not** register periodic polling. The worker must be running. -- Periodic sync reconciles organization/role mappings and names, not users - or individual GroupUser memberships. Unchanged groups are read in one - snapshot rather than each acquiring a lock and native RPC. -- OAuth login refreshes the authenticated existing user's memberships. +- **Etech owns the business Directory and schedules catalog updates.** Its existing + PowerJob `cloudDirectorySync` calls the CloudFile administrator endpoint + `POST /api/v2.1/admin/cloudfile/directory/entities/sync/`. Other enterprise + integrations may use their own scheduler. CloudFile never schedules recurring + org/role fetches by itself; there is no interval/mode parameter. +- The Hub endpoint fetches `GET /groups?view=entities` from the trusted + enterprise Directory using the existing machine JWT, validates the snapshot + and reconciles only organization/role mappings and native Group names. + It does **not** bulk-create users or write GroupUser memberships. +- A standalone CE/CloudFile installation without an external Directory needs + no organizational synchronization or CloudFile background scheduler. +- OAuth login refreshes only the authenticated existing user's memberships. Missing role/org mapping refuses the refresh. Neither login nor an - ACL check automatically creates a missing group; scheduled or explicit - directory synchronization owns catalog creation. -- On file permission checks the built-in principal reads only existing - CE User/Profile/GroupUser/cf_sso_group_map facts. No external HTTP - Directory fetch, JIT user provision, group creation or membership mutation. + ACL check automatically creates a missing group; external application + catalog synchronization owns creation and retry. +- File permission checks use only existing CE User/Profile/GroupUser and + `cf_sso_group_map` facts. No external HTTP Directory fetch, automatic + user/group provisioning or membership mutation occurs on these checks. A missing native user is denied. -- A user's changed membership is refreshed at the next authenticated login - or explicit administrative user-sync event, **not** on each ACL request. - The ten-minute org/role catalog cadence is not a guarantee of ten-minute - active-user membership revocation. -- Internal native **write** permission checks load current ACL rows without - cross-request TTL caching while reusing results within the request; - read/list permission checks keep bounded shared library caching. -- Custom `CF_ACL_SUBJECT_ENSURE` adapters must be read-only. The old - built-in value `cloudfile_ext.v2.directory_adapter.ensure_subject` - remains accepted but no longer triggers synchronization in ACL reads. +- A user's changed membership is refreshed at their next authenticated login + or explicit administrative user-sync event, **not** at every ACL request. + The external catalog schedule does not guarantee prompt revocation of + active-user memberships. +- Internal native **write** permission checks load current ACL rules without + cross-request TTL caching while reusing results within one request; reads + retain bounded shared caching. +- Custom `CF_ACL_SUBJECT_ENSURE` adapters must be read-only. The built-in + `cloudfile_ext.v2.directory_adapter.ensure_subject` value remains + recognized but does not trigger synchronization in ACL reads. -No new CloudFile tables, queue, permission database or scheduler service. +CloudFile owns native mapping and authorization; Etech owns the source facts +and the trigger. No new task tables, schedule modes, or polling services. From 1134e5eac9cb0fb78c84bc197d8bedd3c1e5f139 Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:44:48 +0900 Subject: [PATCH 14/16] perf(search): fixed bounded bootstrap budget; no additional operator setting --- cloudfile_ext/v2/search/worker.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/cloudfile_ext/v2/search/worker.py b/cloudfile_ext/v2/search/worker.py index c625cd6ede..32e0fc011a 100644 --- a/cloudfile_ext/v2/search/worker.py +++ b/cloudfile_ext/v2/search/worker.py @@ -345,16 +345,13 @@ def main(): os.umask(0o077) from django.conf import settings interval = max(15, int(getattr(settings, 'CF_SEARCH_INDEX_INTERVAL', 60))) - pages = int(os.environ.get('CF_SEARCH_BOOTSTRAP_PAGES_PER_TICK', '10')) - if not 1 <= pages <= 100: - raise ValueError('CF_SEARCH_BOOTSTRAP_PAGES_PER_TICK must be 1..100') with open(args.checkpoint + '.lock', 'w') as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) while True: started = time.monotonic() try: tick(configured_client(), seafile_api, args.checkpoint, - args.bootstrap_before, pages_per_repo=pages) + args.bootstrap_before, pages_per_repo=10) logger.info('CF_SEARCH_INDEX status=OK duration_ms=%d', int((time.monotonic() - started) * 1000)) except Exception: From 8f60cac5d4579af7f8fc528da7be854ce7cac3af Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:48:16 +0900 Subject: [PATCH 15/16] test(directory): fix source checks for external-only catalog endpoint --- cloudfile_ext/v2/tests/test_directory_adapter.py | 1 + 1 file changed, 1 insertion(+) diff --git a/cloudfile_ext/v2/tests/test_directory_adapter.py b/cloudfile_ext/v2/tests/test_directory_adapter.py index 3c37d538c9..0b76ef0a7e 100644 --- a/cloudfile_ext/v2/tests/test_directory_adapter.py +++ b/cloudfile_ext/v2/tests/test_directory_adapter.py @@ -2,6 +2,7 @@ from contextlib import contextmanager, nullcontext from datetime import datetime, timezone from types import SimpleNamespace +from pathlib import Path from unittest import TestCase from unittest.mock import Mock, patch From 3e38cfc704c5c612526b7f127589d30dd3ebd9f8 Mon Sep 17 00:00:00 2001 From: servicememeai Date: Sun, 25 Oct 2026 16:48:22 +0900 Subject: [PATCH 16/16] test(directory): fix source checks for external-only catalog endpoint --- cloudfile_ext/v2/tests/test_search_worker_schedule_pure.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cloudfile_ext/v2/tests/test_search_worker_schedule_pure.py b/cloudfile_ext/v2/tests/test_search_worker_schedule_pure.py index deafb15ac1..2cc4f197d8 100644 --- a/cloudfile_ext/v2/tests/test_search_worker_schedule_pure.py +++ b/cloudfile_ext/v2/tests/test_search_worker_schedule_pure.py @@ -101,7 +101,7 @@ def test_directory_catalog_has_no_cloudfile_scheduler(self): self.assertNotIn('_register_directory_catalog', source) def test_external_admin_api_keeps_entity_sync(self): - source = (Path(__file__).resolve().parents[2] / 'http.py').read_text() + source = (Path(__file__).resolve().parents[1] / 'http.py').read_text() self.assertIn('class CatalogSync(', source) self.assertIn('result = sync_entity_catalog()', source)