feat: add getkeys-all-ws config operation to avoid oversized responses (#1058)

With many workspaces, the getvalues-all-ws response for types with
large values (e.g. prompt templates) exceeds Pulsar's max message size.

Add a getkeys-all-ws operation that returns workspace/key pairs without
values. Processors now use this to discover which workspaces have config
of a given type, then fetch values per-workspace. Each individual
response stays within message size limits.

trustgraph-base/trustgraph/base/async_processor.py

@@ -147,10 +147,13 @@ class AsyncProcessor:
 
     async def _fetch_type_all_workspaces(self, client, config_type):
         """Fetch config values of a single type across all workspaces.
-        Returns dict of {workspace: {key: value}}."""
+        Uses getkeys-all-ws to discover workspaces, then fetches values
+        per-workspace to avoid oversized responses."""
+
+        # Step 1: get workspace/key pairs (small response)
         resp = await client.request(
             ConfigRequest(
-                operation="getvalues-all-ws",
+                operation="getkeys-all-ws",
                 type=config_type,
             ),
             timeout=10,
@@ -158,11 +161,21 @@ class AsyncProcessor:
         if resp.error:
             raise RuntimeError(f"Config error: {resp.error.message}")
 
-        grouped = {}
+        version = resp.version
+
+        # Group keys by workspace
+        workspaces = set()
         for v in resp.values:
-            ws = grouped.setdefault(v.workspace, {})
-            ws[v.key] = v.value
-        return grouped, resp.version
+            workspaces.add(v.workspace)
+
+        # Step 2: fetch values per workspace
+        grouped = {}
+        for ws in workspaces:
+            kv = await self._fetch_type_workspace(client, ws, config_type)
+            if kv:
+                grouped[ws] = kv
+
+        return grouped, version
 
     # This is called to start dynamic behaviour.
     # Implements the subscribe-then-fetch pattern to avoid race conditions.

trustgraph-base/trustgraph/schema/services/config.py

@@ -11,6 +11,7 @@ from ..core.primitives import Error
 #   list(workspace, type) -> (version, directory)
 #   getvalues(workspace, type) -> (version, values)
 #   getvalues-all-ws(type) -> (version, values with workspace field)
+#   getkeys-all-ws(type) -> (version, values with workspace+key, no value)
 #   put(workspace, values) -> ()
 #   delete(workspace, keys) -> ()
 #   config(workspace) -> (version, config)
@@ -36,8 +37,8 @@ class ConfigValue:
 
 @dataclass
 class ConfigRequest:
-    # Operations: get, list, getvalues, getvalues-all-ws, delete, put,
-    # config
+    # Operations: get, list, getvalues, getvalues-all-ws,
+    # getkeys-all-ws, delete, put, config
     operation: str = ""
 
     # Workspace scope — required on all operations except

trustgraph-flow/trustgraph/config/service/config.py

@@ -96,6 +96,27 @@ class Configuration:
             values = values,
         )
 
+    async def handle_getkeys_all_ws(self, v):
+        """Fetch all workspace/key pairs of a given type across all
+        workspaces. Returns keys without values so the response stays
+        small regardless of value size."""
+
+        keys = await self.table_store.get_keys_all_ws(v.type)
+
+        values = [
+            ConfigValue(
+                workspace = row[0],
+                type = v.type,
+                key = row[1],
+            )
+            for row in keys
+        ]
+
+        return ConfigResponse(
+            version = await self.get_version(),
+            values = values,
+        )
+
     async def handle_delete(self, v, workspace):
 
         types = list(set(k.type for k in v.keys))
@@ -244,6 +265,9 @@ class Configuration:
         if msg.operation == "getvalues-all-ws":
             resp = await self.handle_getvalues_all_ws(msg)
 
+        elif msg.operation == "getkeys-all-ws":
+            resp = await self.handle_getkeys_all_ws(msg)
+
         elif msg.operation in ("get", "list", "getvalues", "delete",
                                "put", "config"):
 

trustgraph-flow/trustgraph/tables/config.py

@@ -166,6 +166,12 @@ class ConfigTableStore:
             ALLOW FILTERING;
         """)
 
+        self.get_keys_all_ws_stmt = self.cassandra.prepare("""
+            SELECT workspace, key FROM config
+            WHERE class = ?
+            ALLOW FILTERING;
+        """)
+
     async def put_config(self, workspace, cls, key, value):
         try:
             await async_execute(
@@ -220,6 +226,21 @@ class ConfigTableStore:
 
         return [(row[0], row[1], row[2]) for row in rows]
 
+    async def get_keys_all_ws(self, cls):
+        """Return (workspace, key) tuples for all workspaces
+        with entries of the given class."""
+        try:
+            rows = await async_execute(
+                self.cassandra,
+                self.get_keys_all_ws_stmt,
+                (cls,),
+            )
+        except Exception:
+            logger.error("Exception occurred", exc_info=True)
+            raise
+
+        return [(row[0], row[1]) for row in rows]
+
     async def get_all(self):
         try:
             rows = await async_execute(