From eb059707f0182bbb9e9a3a922a367b9463093b1f Mon Sep 17 00:00:00 2001 From: cybermaggedon Date: Thu, 23 Jul 2026 12:13:54 +0100 Subject: [PATCH] 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/async_processor.py | 25 ++++++++++++++----- .../trustgraph/schema/services/config.py | 5 ++-- .../trustgraph/config/service/config.py | 24 ++++++++++++++++++ trustgraph-flow/trustgraph/tables/config.py | 21 ++++++++++++++++ 4 files changed, 67 insertions(+), 8 deletions(-) diff --git a/trustgraph-base/trustgraph/base/async_processor.py b/trustgraph-base/trustgraph/base/async_processor.py index dd582078..b65074bb 100644 --- a/trustgraph-base/trustgraph/base/async_processor.py +++ b/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. diff --git a/trustgraph-base/trustgraph/schema/services/config.py b/trustgraph-base/trustgraph/schema/services/config.py index 0bdae3a4..5321d008 100644 --- a/trustgraph-base/trustgraph/schema/services/config.py +++ b/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 diff --git a/trustgraph-flow/trustgraph/config/service/config.py b/trustgraph-flow/trustgraph/config/service/config.py index a9314772..97ee8baa 100644 --- a/trustgraph-flow/trustgraph/config/service/config.py +++ b/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"): diff --git a/trustgraph-flow/trustgraph/tables/config.py b/trustgraph-flow/trustgraph/tables/config.py index c87cb3b5..b7d54bf3 100644 --- a/trustgraph-flow/trustgraph/tables/config.py +++ b/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(