fix: external-only Directory catalog sync and bounded Meili indexing

.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: |

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,23 +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
-
-        if 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.
 

cloudfile_ext/management/commands/cf_worker.py

@@ -6,11 +6,11 @@
 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 logging

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.

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,22 @@ 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)))
     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=10)
+                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__':

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
 
@@ -136,22 +137,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_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=''):
-            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')