Bound the per-run inbound correlation maps
Every mid-run peer DM injected into a long-lived run allocated one entry in byMessageId and byPiTimestamp without ever being evicted — only run shutdown drops the registry. A CEO session that accumulates thousands of DMs between session rotations would grow the maps without bound. Evict the oldest entries symmetrically once a run's registry exceeds MAX_INBOUND_CORRELATIONS_PER_RUN, which is more than enough for the pending-to-drain window that message_end actually needs.
This commit is contained in:
1 parent
dd8485b7e2
commit
8bf480236f
1 file changed
+32
@@ -75,6 +75,21 @@ interface InboundCorrelationRegistry {
|
||||
lastAllocatedTimestamp: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Upper bound on inbound-correlation entries retained per active run.
|
||||
*
|
||||
* Every inbound peer message injected during a single pi run allocates
|
||||
* one entry in both the messageId and piTimestamp maps of the registry.
|
||||
* A long-lived CEO session that accumulates thousands of mid-run DMs
|
||||
* would otherwise grow the maps without bound until the session
|
||||
* compaction fires (which does not clean these structures). Keeping the
|
||||
* `MAX_INBOUND_CORRELATIONS_PER_RUN` most recent entries is sufficient:
|
||||
* `message_end` resolves author identity immediately after each turn,
|
||||
* so the in-memory window only needs to cover pending-to-drain inbound
|
||||
* messages, not the full run history.
|
||||
*/
|
||||
const MAX_INBOUND_CORRELATIONS_PER_RUN = 512;
|
||||
|
||||
export class RuntimeBridge {
|
||||
private readonly costWriter: CostWriter;
|
||||
private readonly activeRuns = new Map<string, ActiveRunContext>();
|
||||
@@ -171,11 +186,28 @@ export class RuntimeBridge {
|
||||
};
|
||||
registry.byMessageId.set(record.id, entry);
|
||||
registry.byPiTimestamp.set(allocated, record.id);
|
||||
this.evictOldestCorrelationsIfNeeded(registry);
|
||||
return projectInboxMessageToPi(record, {
|
||||
getAgentName: (id) => this.resolveAgentName(id),
|
||||
}, { timestampOverride: allocated });
|
||||
}
|
||||
|
||||
/**
|
||||
* Bound the per-run correlation maps. JavaScript `Map` preserves
|
||||
* insertion order, so dropping the head of each map retires the
|
||||
* oldest entries. Eviction operates symmetrically on both maps so
|
||||
* the reverse lookups remain consistent.
|
||||
*/
|
||||
private evictOldestCorrelationsIfNeeded(registry: InboundCorrelationRegistry): void {
|
||||
while (registry.byMessageId.size > MAX_INBOUND_CORRELATIONS_PER_RUN) {
|
||||
const oldestMessageId = registry.byMessageId.keys().next().value;
|
||||
if (!oldestMessageId) break;
|
||||
const entry = registry.byMessageId.get(oldestMessageId);
|
||||
registry.byMessageId.delete(oldestMessageId);
|
||||
if (entry) registry.byPiTimestamp.delete(entry.piTimestamp);
|
||||
}
|
||||
}
|
||||
|
||||
private getOrCreateInboundRegistry(runId: string): InboundCorrelationRegistry {
|
||||
const existing = this.inboundCorrelations.get(runId);
|
||||
if (existing) return existing;
|
||||
|
||||
Reference in new issue
Block a user