Skip to content

perf(server): bound the provider runtime-event bus #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ronak-guliani merged 1 commit into from
Jun 10, 2026
Merged

perf(server): bound the provider runtime-event bus #5

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion apps/server/src/provider/Layers/ProviderService.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,17 @@ const dieOnMissingBindingInstanceId = (
);
};

// Capacity of the unified runtime-event bus. A bounded PubSub caps the
// per-subscriber backlog so a lagging consumer (e.g. ProviderRuntimeIngestion
// or CheckpointReactor falling behind during a burst) applies backpressure all
// the way up to the adapter stdio reader instead of letting the buffer grow
// without bound and eventually OOM the process. The window is generous enough
// to absorb normal event bursts (streaming deltas, tool calls) while keeping a
// hard ceiling on retained events. Runtime events must never be dropped — they
// feed event-sourced orchestration ingestion — so we use the suspending
// (backpressure) strategy rather than sliding/dropping.
const RUNTIME_EVENT_BUS_CAPACITY = 4096;

const correlateRuntimeEventWithInstance = (
source: {
readonly instanceId: ProviderInstanceId;
Expand Down Expand Up @@ -199,7 +210,9 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (

const registry = yield* ProviderAdapterRegistry;
const directory = yield* ProviderSessionDirectory;
const runtimeEventPubSub = yield* PubSub.unbounded<ProviderRuntimeEvent>();
const runtimeEventPubSub = yield* PubSub.bounded<ProviderRuntimeEvent>(
RUNTIME_EVENT_BUS_CAPACITY,
);

const publishRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
Effect.succeed(event).pipe(
Expand Down
Loading