perf(workspace): index event polls and share cached stream heads

workspace/backend/alembic/versions/083_event_poll_indexes.py

@@ -0,0 +1,31 @@
+"""Selective ordered indexes for hot event-poll prefixes."""
+from alembic import op
+import sqlalchemy as sa
+
+revision = '083'
+down_revision = '082'
+branch_labels = None
+depends_on = None
+
+INDEXES = (
+    ('idx_events_poll_message', 'workspace.message.posted%'),
+    ('idx_events_poll_control', 'workspace.agent.control%'),
+    ('idx_events_poll_tool_result', 'workspace.tool_result%'),
+)
+
+
+def upgrade():
+    with op.get_context().autocommit_block():
+        for name, pattern in INDEXES:
+            op.execute(f"CREATE INDEX CONCURRENTLY IF NOT EXISTS {name} "
+                       f"ON events (network_id, timestamp DESC, id DESC) WHERE type LIKE '{pattern}'")
+            valid = op.get_bind().execute(sa.text(
+                'SELECT indisvalid FROM pg_index WHERE indexrelid = to_regclass(:name)'), {'name': name}).scalar_one()
+            if not valid:
+                raise RuntimeError(f'Invalid {name}: drop concurrently and retry')
+
+
+def downgrade():
+    with op.get_context().autocommit_block():
+        for name, _ in reversed(INDEXES):
+            op.execute(f'DROP INDEX CONCURRENTLY IF EXISTS {name}')

workspace/backend/app/cache.py

@@ -118,6 +118,58 @@ def json_read_through(
     return value
 
 
+def json_singleflight(key, ttl_seconds, compute, cache_if=None):
+    """Share identical cache fills across workers, with a bounded wait.
+
+    Redis/lock failures fall through to the database. Never swallow compute
+    errors or cache failed computations. The lock's ownership token prevents
+    an expired leader from releasing a newer leader's lock.
+    """
+    if key is None:
+        return compute()
+    missing = object()
+
+    def read():
+        raw = get_bytes(key)
+        if raw is not None:
+            try:
+                return json.loads(raw)
+            except (TypeError, ValueError):
+                pass
+        return missing
+
+    value = read()
+    if value is not missing:
+        return value
+    client = _lazy_client()
+    lock = None
+    acquired = False
+    if client is not None:
+        try:
+            lock = client.lock(key + ':fill', timeout=5, blocking_timeout=0.1, sleep=0.01)
+            acquired = lock.acquire()
+        except Exception:
+            pass
+    try:
+        # A preceding leader may have filled the cache while we waited.
+        value = read() if client is not None else missing
+        if value is not missing:
+            return value
+        value = compute()
+        if cache_if is None or cache_if(value):
+            try:
+                set_bytes(key, json.dumps(value, default=str, separators=(',', ':')).encode(), ttl_seconds)
+            except (TypeError, ValueError):
+                pass
+        return value
+    finally:
+        if acquired:
+            try:
+                lock.release()
+            except Exception:
+                pass
+
+
 # ---------------------------------------------------------------------------
 # Pub/Sub — used by SSE streaming
 # ---------------------------------------------------------------------------

workspace/backend/app/models.py

@@ -70,6 +70,12 @@ class EventRecord(Base):
         Index("idx_events_network_type_target_ts", "network_id", "type", "target", "timestamp"),
         Index("idx_events_channel_latest", "network_id", "target", timestamp.desc(), id.desc(),
               postgresql_where=text("target LIKE 'channel/%'")),
+        Index("idx_events_poll_message", "network_id", timestamp.desc(), id.desc(),
+              postgresql_where=text("type LIKE 'workspace.message.posted%'")),
+        Index("idx_events_poll_control", "network_id", timestamp.desc(), id.desc(),
+              postgresql_where=text("type LIKE 'workspace.agent.control%'")),
+        Index("idx_events_poll_tool_result", "network_id", timestamp.desc(), id.desc(),
+              postgresql_where=text("type LIKE 'workspace.tool_result%'")),
     )