Mirror_Repos/anomalyco_opencode
2
0
Fork 0
Code Issues Pull Requests Actions 95 Packages Projects Releases Wiki Activity

refactor(server): narrow event feed scope

Browse Source
This commit is contained in:
Kit Langton
2026-07-11 22:24:57 -04:00
parent 2d90816ede
commit 57a222de2b
4 changed files with 23 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/src/event-logger.ts
+12 -11
View File
@@ -1,22 +1,23 @@
export * as EventLogger from "./event-logger"
import { Effect, Layer, Stream } from "effect"
import { Agent } from "@opencode-ai/schema/agent"
import { Catalog } from "@opencode-ai/schema/catalog"
import { Command } from "@opencode-ai/schema/command"
import { Config } from "@opencode-ai/schema/config"
import { Effect, Layer } from "effect"
import { makeGlobalNode } from "./effect/app-node"
import { EventV2 } from "./event"
const Types = new Set([
"agent.updated",
"catalog.updated",
"command.updated",
"config.updated",
])
export const layer = Layer.effectDiscard(
Effect.gen(function* () {
const events = yield* EventV2.Service
yield* events
.subscribe([Agent.Event.Updated, Catalog.Event.Updated, Command.Event.Updated, Config.Event.Updated])
.pipe(
Stream.runForEach((event) => Effect.logInfo("event", { event })),
Effect.forkScoped({ startImmediately: true }),
)
const unsubscribe = yield* events.listen((event) =>
Types.has(event.type) ? Effect.logInfo("event", { event }) : Effect.void,
)
yield* Effect.addFinalizer(() => unsubscribe)
}),
)
packages/server/script/bench-event-feed.ts
-59
View File
@@ -1,59 +0,0 @@
import { OpenCodeEvent } from "@opencode-ai/protocol/groups/event"
import { EventV2 } from "@opencode-ai/core/event"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { DateTime } from "effect"
import { EventFeed } from "../src/event-feed"
const clients = Number(process.argv[2] ?? 10)
const events = Number(process.argv[3] ?? 1_000)
const mode = process.argv[4] ?? "current"
const runs = 9
const event: OpenCodeEvent = {
id: EventV2.ID.make("evt_benchmark"),
created: DateTime.makeUnsafe(Date.now()),
type: "mcp.status.changed",
location: { directory: AbsolutePath.make("/tmp/opencode-benchmark") },
metadata: {
output: "x".repeat(8_192),
nested: Array.from({ length: 32 }, (_, index) => ({ index, value: `value-${index}` })),
},
data: { server: "benchmark" },
}
function current() {
let bytes = 0
for (let index = 0; index < events; index++) {
for (let client = 0; client < clients; client++) bytes += EventFeed.frame(event).length
}
return bytes
}
function shared() {
let bytes = 0
for (let index = 0; index < events; index++) {
const encoded = EventFeed.frame(event)
for (let client = 0; client < clients; client++) bytes += encoded.length
}
return bytes
}
const benchmark = mode === "shared" ? shared : current
benchmark()
const samples = Array.from({ length: runs }, () => {
Bun.gc(true)
const start = performance.now()
const bytes = benchmark()
return { duration: performance.now() - start, bytes }
})
const durations = samples.map((sample) => sample.duration).toSorted((a, b) => a - b)
const median = durations[Math.floor(durations.length / 2)]
const absoluteDeviations = durations.map((duration) => Math.abs(duration - median)).toSorted((a, b) => a - b)
const mad = absoluteDeviations[Math.floor(absoluteDeviations.length / 2)]
console.log(`mode=${mode} clients=${clients} events=${events} runs=${runs}`)
console.log(
`median=${median.toFixed(3)}ms mad=${mad.toFixed(3)}ms best=${durations[0].toFixed(3)}ms worst=${durations[durations.length - 1].toFixed(3)}ms`,
)
console.log(`METRIC event_feed_${mode}_ms=${median.toFixed(3)}`)
console.log(`METRIC event_feed_${mode}_mad_ms=${mad.toFixed(3)}`)
packages/server/test/event-feed.test.ts
+8
View File
@@ -1,6 +1,7 @@
import { describe, expect, test } from "bun:test"
import { AgentV2 } from "@opencode-ai/core/agent"
import { EventV2 } from "@opencode-ai/core/event"
import { OpenCodeEvent } from "@opencode-ai/protocol/groups/event"
import { DateTime, Deferred, Effect, Exit, Fiber, Option, Schema, Stream } from "effect"
import { EventFeed } from "../src/event-feed"
@@ -35,6 +36,13 @@ function makeSource() {
}
describe("EventFeed", () => {
test("preserves the public SSE frame encoding", () => {
const payload = event("wire")
expect(EventFeed.frame(payload)).toBe(
`data: ${JSON.stringify(Schema.encodeUnknownSync(OpenCodeEvent)(payload))}\n\n`,
)
})
test("encodes once and delivers the same frame to every subscriber", async () => {
let encodes = 0
const result = await Effect.runPromise(
specs/v2/event-stream-architecture.md
+3 -17
View File
@@ -5,7 +5,7 @@
The public HTTP event stream uses one Server-scoped encoded feed with one independently bounded queue per connection.
```text
Core EventV2.subscribe()
Core EventV2.listen()
|
| one global subscription
v
@@ -165,10 +165,6 @@ If one accepted public event cannot be encoded:
Keeping current clients connected would create a silent gap. Permanently terminating the feed would poison future connections.
### Source termination
Unexpected Core source termination logs the cause and closes current subscribers. Normal Server-scope interruption is not logged as an error.
## HTTP And Code Generation
Protocol remains unchanged:
@@ -196,23 +192,13 @@ Because method, path, schema, and wire representation do not change:
## Core Cleanup
The Server no longer uses `EventV2.liveBounded`, so Core removes that helper and its transport-specific overflow error.
`EventLogger` moves from callback `listen` to a scoped `events.subscribe()` consumer.
The deprecated `listen` interface remains temporarily for V1 compatibility and tests. Removing the callback fan-out completely requires migrating those callers separately and must not be conflated with this Server transport optimization.
The Server no longer uses `EventV2.liveBounded`, so Core removes that dead helper and its transport-specific overflow error. The feed registers one observer through the existing `listen` interface; other listeners are unchanged.
Transactional projector registration is unrelated and remains unchanged.
## Benchmark
The disposable benchmark reproduces the current per-connection schema/JSON/SSE encoding path with a representative 8 KiB public event. It uses one warmup and nine measured runs; median is the primary metric and median absolute deviation is reported.
Command from `packages/server`:
```sh
bun run script/bench-event-feed.ts <clients> 1000 <current|shared>
```
The disposable benchmark reproduced the previous per-connection schema/JSON/SSE encoding path with a representative 8 KiB public event. It used one warmup and nine measured runs; median was the primary metric and median absolute deviation was reported. The benchmark was intentionally not committed because it isolated the removed encoding boundary rather than exercising the complete HTTP stack.
Results on Apple Silicon with Bun 1.3.14: