RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: address cache thrashing

This commit is contained in:
Rajat Ahuja 2025-11-21 17:50:27 -05:00
parent 414964dc4e
commit 8b270ed428
4 changed files with 119 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
src/crud/peer.py
View File

@ -11,6 +11,7 @@ from src.cache.client import cache, get_cache_namespace
from src.config import settings
from src.crud.workspace import get_or_create_workspace
from src.exceptions import ConflictException, ResourceNotFoundException
from src.models import Peer
from src.utils.filter import apply_filter
logger = getLogger(__name__)
@ -63,21 +64,37 @@ async def get_or_create_peers(
.where(models.Peer.name.in_(peer_names))
)
result = await db.execute(stmt)
existing_peers = list(result.scalars().all())
existing_peers: list[Peer] = list(result.scalars().all())
# Create a mapping of peer names to peer schemas for easy lookup
peer_schema_map = {p.name: p for p in peers}
# Track which peers actually changed
changed_peers: list[Peer] = []
# Update existing peers with metadata and configuration if provided
for existing_peer in existing_peers:
peer_schema = peer_schema_map[existing_peer.name]
changed = False
# Update with metadata and configuration if provided
if peer_schema.metadata is not None:
# Update with metadata if provided AND different
if (
peer_schema.metadata is not None
and existing_peer.h_metadata != peer_schema.metadata
):
existing_peer.h_metadata = peer_schema.metadata
changed = True
if peer_schema.configuration is not None:
# Update with configuration if provided AND different
if (
peer_schema.configuration is not None
and existing_peer.configuration != peer_schema.configuration
):
existing_peer.configuration = peer_schema.configuration
changed = True
if changed:
changed_peers.append(existing_peer)
# Find which peers need to be created
existing_names = {p.name for p in existing_peers}
@ -104,10 +121,15 @@ async def get_or_create_peers(
) from None
return await get_or_create_peers(db, workspace_name, peers, _retry=True)
# Invalidate cache for all updated/created peers - read-through pattern
for peer_obj in existing_peers + new_peers:
# Only invalidate cache for changed/new peers - read-through pattern
for peer_obj in changed_peers + new_peers:
cache_key = peer_cache_key(workspace_name, peer_obj.name)
await cache.delete(cache_key)
logger.debug(
"Peer %s cache invalidated in workspace %s (changed or new)",
peer_obj.name,
workspace_name,
)
# Return combined list of existing and new peers
return existing_peers + new_peers
@ -204,16 +226,31 @@ async def update_peer(
)
)[0]
if peer.metadata is not None:
honcho_peer.h_metadata = peer.metadata
needs_update = False
if peer.configuration is not None:
if peer.metadata is not None and honcho_peer.h_metadata != peer.metadata:
honcho_peer.h_metadata = peer.metadata
needs_update = True
if (
peer.configuration is not None
and honcho_peer.configuration != peer.configuration
):
honcho_peer.configuration = peer.configuration
needs_update = True
# Early exit if unchanged
if not needs_update:
logger.debug(
"Peer %s unchanged in workspace %s, skipping update",
peer_name,
workspace_name,
)
return honcho_peer
await db.commit()
await db.refresh(honcho_peer)
# Invalidate cache - read-through pattern
cache_key = peer_cache_key(workspace_name, honcho_peer.name)
await cache.delete(cache_key)

54
src/crud/session.py
View File

@ -126,6 +126,9 @@ async def get_or_create_session(
if honcho_session is not None:
honcho_session = await db.merge(honcho_session, load=False)
# Track if we need to update cache
needs_cache_update = False
# Check if session already exists
if honcho_session is None:
if session.peer_names:
@ -151,6 +154,7 @@ async def get_or_create_session(
db.add(honcho_session)
# Flush to ensure session exists in DB before adding peers
await db.flush()
except IntegrityError:
await db.rollback()
logger.debug(
@ -163,10 +167,18 @@ async def get_or_create_session(
return await get_or_create_session(db, session, workspace_name, _retry=True)
else:
# Update existing session with metadata and feature flags if provided
if session.metadata is not None:
if (
session.metadata is not None
and honcho_session.h_metadata != session.metadata
):
honcho_session.h_metadata = session.metadata
if session.configuration is not None:
needs_cache_update = True
if (
session.configuration is not None
and honcho_session.configuration != session.configuration
):
honcho_session.configuration = session.configuration
needs_cache_update = True
# Add all peers to session
if session.peer_names:
@ -187,10 +199,16 @@ async def get_or_create_session(
await db.commit()
await db.refresh(honcho_session)
cache_key = session_cache_key(workspace_name, session.name)
await cache.set(
cache_key, honcho_session, expire=settings.CACHE.DEFAULT_TTL_SECONDS
)
# Only update cache if session data changed or was newly created
if needs_cache_update:
cache_key = session_cache_key(workspace_name, session.name)
await cache.set(
cache_key, honcho_session, expire=settings.CACHE.DEFAULT_TTL_SECONDS
)
logger.debug(
"Session %s cache updated in workspace %s", session.name, workspace_name
)
return honcho_session
@ -251,16 +269,32 @@ async def update_session(
db, schemas.SessionCreate(name=session_name), workspace_name=workspace_name
)
if session.metadata is not None:
honcho_session.h_metadata = session.metadata
# Track if anything changed
needs_update = False
if session.configuration is not None:
if session.metadata is not None and honcho_session.h_metadata != session.metadata:
honcho_session.h_metadata = session.metadata
needs_update = True
if (
session.configuration is not None
and honcho_session.configuration != session.configuration
):
honcho_session.configuration = session.configuration
needs_update = True
if not needs_update:
logger.debug(
"Session %s unchanged in workspace %s, skipping update",
session_name,
workspace_name,
)
return honcho_session
await db.commit()
await db.refresh(honcho_session)
# Invalidate cache - read-through pattern
# Only invalidate if we actually updated
cache_key = session_cache_key(workspace_name, session_name)
await cache.delete(cache_key)

24
src/crud/workspace.py
View File

@ -172,16 +172,32 @@ async def update_workspace(
),
)
if workspace.metadata is not None:
honcho_workspace.h_metadata = workspace.metadata
# Track if anything changed
needs_update = False
if workspace.configuration is not None:
if (
workspace.metadata is not None
and honcho_workspace.h_metadata != workspace.metadata
):
honcho_workspace.h_metadata = workspace.metadata
needs_update = True
if (
workspace.configuration is not None
and honcho_workspace.configuration != workspace.configuration
):
honcho_workspace.configuration = workspace.configuration
needs_update = True
# Early exit if unchanged
if not needs_update:
logger.debug("Workspace %s unchanged, skipping update", workspace_name)
return honcho_workspace
await db.commit()
await db.refresh(honcho_workspace)
# Invalidate cache
# Only invalidate if we actually updated
cache_key = workspace_cache_key(workspace_name)
await cache.delete(cache_key)

15
src/utils/summarizer.py
View File

@ -373,7 +373,7 @@ async def _create_and_save_summary(
message_public_id=message_public_id,
)
# Only track tokens if this was a real LLM call
# Only track tokens and save summary if this was a real LLM call
if not is_fallback:
# Get base prompt tokens based on summary type
if summary_type == SummaryType.SHORT:
@ -397,12 +397,13 @@ async def _create_and_save_summary(
component="total",
).inc(new_summary["token_count"])
await _save_summary(
db,
new_summary,
workspace_name,
session_name,
)
# Save summary to database
await _save_summary(
db,
new_summary,
workspace_name,
session_name,
)
accumulate_metric(
f"summary_{workspace_name}_{message_id}",