diff --git a/docs/connectors.md b/docs/connectors.md index a900a7b3..1986544d 100644 --- a/docs/connectors.md +++ b/docs/connectors.md @@ -151,6 +151,21 @@ Gmail, Discord, Feishu) ship presets, so `[[objects]]` is optional there — use only to override the default fields. See [Search and browse](search-and-browse.md#locate-exact-hits) for how these become `content`, `locator`, and `metadata.fields` in a result. +### Processing order within a connector + +Add `priority` to any `[[objects]]` block to change which of *that connector's own* +objects get indexed first during a sync (lower runs earlier; unmatched objects keep the +connector's own default order): + +```toml +[[objects]] +match = "archive/**" +priority = 900 # push a low-value subtree to the back of the queue +``` + +This only reorders work inside one connector's sync job — it does not affect scheduling +across different connectors running at the same time. + ## After ingest Every scheme answers the same commands: diff --git a/docs/ingest-pipeline.md b/docs/ingest-pipeline.md index 56e561a6..3addf40f 100644 --- a/docs/ingest-pipeline.md +++ b/docs/ingest-pipeline.md @@ -74,7 +74,11 @@ Notes: - **Queue ① is durable.** `ObjectTask` rows sit in the metadata database, so a worker crash never loses pending work. Workers claim by `(priority ASC, - started_at ASC)`. + started_at ASC)`. `priority` defaults to the connector's own bucketing (e.g. + the file connector's entrypoint/src/tests/generated heuristic) but a user + `[[objects]] priority` match overrides it — see + [Connectors](connectors.md#processing-order-within-a-connector). This orders + work only within one connector's own sync job, not across connectors. - **The Producer pool is okind-dispatched.** `select_producer(okind, ctx)` maps the eight object kinds (`document`, `code`, `text_blob`, `image`, `message_stream`, `table_rows`, `record_collection`, `table_schema`) onto five diff --git a/server/python/src/mfs_server/connectors/base.py b/server/python/src/mfs_server/connectors/base.py index d29ef4e5..7caf7772 100644 --- a/server/python/src/mfs_server/connectors/base.py +++ b/server/python/src/mfs_server/connectors/base.py @@ -231,6 +231,11 @@ class ObjectConfig: indexable: bool = True chunk_max: int = 1_000_000 group_by: Optional[str] = None # message_stream: override the auto-detected thread key + # Task processing order within THIS connector's own sync job — lower runs earlier + # (matches the file connector's built-in task_priority() bucket scale, e.g. -350 for + # entrypoints). None (default) means "no user override, use the connector's own + # task_priority()". Does not affect ordering across different connectors/jobs. + priority: Optional[int] = None # Optional Python str.format template applied per record before chunking. # Lets message presets render `alice: 部署炸了, 503 spike` instead of the # default labeled `text: 部署炸了, 503 spike`, which embeds better and diff --git a/server/python/src/mfs_server/engine/engine.py b/server/python/src/mfs_server/engine/engine.py index 861d9c33..586962b3 100644 --- a/server/python/src/mfs_server/engine/engine.py +++ b/server/python/src/mfs_server/engine/engine.py @@ -705,6 +705,9 @@ async def _drain_job( # 'explicit_only' (yielded events, e.g. upload) honor them continue # never-delete connectors (slack/gmail) keep the index tid = uuid.uuid4().hex + # a user [[objects]] `priority =` match overrides the connector's own + # task_priority() default (e.g. file's README/tests/vendor buckets). + user_priority = plugin.ctx.object_config_for(ch.uri).priority await self.objects.insert_task( tid, job_id, @@ -712,7 +715,7 @@ async def _drain_job( ch.uri, ch.old_uri, ch.kind, - plugin.task_priority(ch), + user_priority if user_priority is not None else plugin.task_priority(ch), ) if ch.kind != "deleted": # Accumulate the dir tree (okind passed in — no extra DB hit, §6.4.1). diff --git a/server/python/tests/test_engine_object_priority_e2e.py b/server/python/tests/test_engine_object_priority_e2e.py new file mode 100644 index 00000000..d1dc5070 --- /dev/null +++ b/server/python/tests/test_engine_object_priority_e2e.py @@ -0,0 +1,99 @@ +"""End-to-end: a user `[[objects]] priority=` override changes the processing order of a +real `add()` on a real file connector — drives the actual `plugin.sync()` enumeration + +`_drain_job` task insertion, no fakes on that path. Verifies both that the override lands +on the right objects and that untouched objects still fall back to the connector's own +`task_priority()` buckets (§6.3), and that the DB claim order (priority ASC, existing +behavior) reflects it end to end.""" + +from __future__ import annotations + +from mfs_server.config import ServerConfig +from mfs_server.connectors.registry import load_builtin +from mfs_server.engine.engine import Engine + + +class _NoopMilvus: + def delete_by_connector(self, *args, **kwargs): + return None + + +class _NoopJobLane: + def register_job(self, *args, **kwargs): + return None + + def on_sync_done(self, *args, **kwargs): + return None + + def on_yield_object_change(self, *args, **kwargs): + return None + + def evict_job(self, *args, **kwargs): + return None + + +async def _build_engine(tmp_path) -> Engine: + load_builtin() + cfg = ServerConfig() + cfg.metadata.backend = "sqlite" + cfg.metadata.path = str(tmp_path / "metadata.db") + cfg.transformation_cache.backend = "sqlite" + cfg.transformation_cache.db_path = str(tmp_path / "tx.db") + cfg.artifact_cache.root = str(tmp_path / "artifacts") + cfg.milvus.uri = str(tmp_path / "milvus.db") + eng = Engine(cfg) + eng.milvus = _NoopMilvus() + eng._job_lane = _NoopJobLane() + await eng.meta.connect() + await eng.meta.init_schema() + return eng + + +async def test_objects_priority_override_wins_and_falls_back_end_to_end(tmp_path): + eng = await _build_engine(tmp_path) + try: + root = tmp_path / "repo" + (root / "src").mkdir(parents=True) + (root / "archive").mkdir() + (root / "README.md").write_text("entrypoint") + (root / "notes.md").write_text("scratch notes") + (root / "src" / "main.py").write_text("print(1)") + (root / "archive" / "old.txt").write_text("stale") + + root_uri = f"file://local{root}" + await eng.add( + str(root), + config={ + "root": str(root), + "client_id": "local", + "objects": [ + # user override: bump archive/* to the back of the line + {"match": "archive/*", "priority": 900}, + # user override: pull notes.md ahead of everything, including + # the file connector's own -350 entrypoint bucket + {"match": "notes.md", "priority": -500}, + ], + }, + process=False, # enumerate + insert_task only; no embedding/Milvus needed + ) + + cid = await eng.objects.get_connector_id_by_uri(root_uri) + assert cid is not None + rows = await eng.meta.fetchall( + "SELECT object_uri, priority FROM object_tasks WHERE connector_id=?", (cid,) + ) + by_uri = {r["object_uri"].lstrip("/"): r["priority"] for r in rows} + + assert by_uri["archive/old.txt"] == 900 # user override + assert by_uri["notes.md"] == -500 # user override, beats the built-in -350 bucket + assert by_uri["README.md"] == -350 # no override -> falls back to file's own bucket + assert by_uri["src/main.py"] == -220 # no override -> falls back to file's own bucket + + # end-to-end: the claim order (existing priority ASC, started_at ASC behavior) + # actually reflects the override, not just the raw column value. + claimed = await eng.objects.claim_tasks(cid, limit=10) + order = [r["object_uri"].lstrip("/") for r in claimed] + assert order.index("notes.md") < order.index("README.md") + assert order.index("README.md") < order.index("src/main.py") + assert order.index("src/main.py") < order.index("archive/old.txt") + finally: + await eng.meta.close() diff --git a/skills/mfs-ingest/SKILL.md b/skills/mfs-ingest/SKILL.md index 7b52bce7..b24e8b45 100644 --- a/skills/mfs-ingest/SKILL.md +++ b/skills/mfs-ingest/SKILL.md @@ -271,6 +271,7 @@ User wants to change a registered connector — new token, different | Cap on per-object chunks | `chunk_max` | | Which columns to embed (DB / SaaS) | `[[objects]] text_fields` | | Make a previously-indexed object stop indexing | `[[objects]] indexable = false` | + | Process some objects before others in this sync | `[[objects]] priority` (lower = earlier; doesn't affect ordering across different connectors) | See `reference/connectors/.md` for the full field list. @@ -293,6 +294,7 @@ User wants to change a registered connector — new token, different | `text_fields` (which columns become content) | yes, with `--force-index` (see below) | | `embedding.*` in server config | yes (and affects ALL connectors) | | `[[objects]] indexable = false` | drops that object from index | +| `[[objects]] priority` | no — only changes processing order for objects enumerated in a future sync, not already-succeeded ones | **Applying a `text_fields` change to existing objects needs `mfs add --force-index`.** A plain `mfs connector update` saves the new