fix(api-server): SSE stream pulls incremental events via seq cursor
Replace per-poll 5000-row scan + JS filter with indexed WHERE seq > ? query (uses idx_run_events_run). Batch up to 500 per call, drain loop on reconnect. Poll interval raised 500ms -> 1000ms.
This commit is contained in:
1 parent
c5c3919c78
commit
c81d72fb9c
3 files changed
+33
-7
No files matched your search
@@ -3,7 +3,8 @@ import { requireParam, type AgentApiServer } from "./server-core";
|
||||
|
||||
const DEFAULT_EVENT_LIMIT = 500;
|
||||
const MAX_EVENT_LIMIT = 5_000;
|
||||
const EVENT_POLL_INTERVAL_MS = 500;
|
||||
const SSE_BATCH_LIMIT = 500;
|
||||
const EVENT_POLL_INTERVAL_MS = 1_000;
|
||||
const SSE_HEARTBEAT_MS = 15_000;
|
||||
|
||||
function parsePositiveInt(value: string | null, fallback: number) {
|
||||
@@ -64,12 +65,14 @@ export function registerRunRoutes(server: AgentApiServer) {
|
||||
writeSse(res, "connected", { runId, after: cursor });
|
||||
|
||||
const flushPendingEvents = () => {
|
||||
const events = server.db
|
||||
.listRunEvents(runId, token.companyId, { offset: 0, limit: MAX_EVENT_LIMIT })
|
||||
.filter((event) => event.seq > cursor);
|
||||
for (const event of events) {
|
||||
writeSse(res, "run-event", event);
|
||||
cursor = event.seq;
|
||||
while (true) {
|
||||
const events = server.db.listRunEventsAfter(runId, token.companyId, cursor, SSE_BATCH_LIMIT);
|
||||
if (events.length === 0) return;
|
||||
for (const event of events) {
|
||||
writeSse(res, "run-event", event);
|
||||
cursor = event.seq;
|
||||
}
|
||||
if (events.length < SSE_BATCH_LIMIT) return;
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
@@ -276,6 +276,7 @@ export class AppDatabaseBase {
|
||||
getInterruptedRunsForRecovery() { return this.runs.getInterruptedRunsForRecovery(); }
|
||||
appendRunEvent(input: { companyId: string; runId: string; agentId: string; eventType: string; stream?: string | null; level?: string | null; message?: string | null; payload?: Record<string, unknown> | null }) { return this.runs.appendRunEvent(input); }
|
||||
listRunEvents(runId: string, companyId: string, opts?: { offset?: number; limit?: number }) { return this.runs.listRunEvents(runId, companyId, opts); }
|
||||
listRunEventsAfter(runId: string, companyId: string, afterSeq: number, limit: number) { return this.runs.listRunEventsAfter(runId, companyId, afterSeq, limit); }
|
||||
updateRunActor(runId: string, actorType: string, actorId: string | null) { return this.runs.updateRunActor(runId, actorType, actorId); }
|
||||
|
||||
// ─── Approvals ─────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -340,6 +340,28 @@ export function buildRunMethods(
|
||||
}));
|
||||
},
|
||||
|
||||
listRunEventsAfter(runId: string, companyId: string, afterSeq: number, limit: number): Array<{
|
||||
id: number; runId: string; agentId: string; seq: number; eventType: string;
|
||||
stream: string | null; level: string | null; message: string | null;
|
||||
payload: Record<string, unknown> | null; createdAt: string;
|
||||
}> {
|
||||
return readAll<SqliteRow>(
|
||||
"SELECT id, run_id, agent_id, seq, event_type, stream, level, message, payload, created_at FROM run_events WHERE run_id = ? AND company_id = ? AND seq > ? ORDER BY seq ASC LIMIT ?",
|
||||
runId, companyId, afterSeq, limit
|
||||
).map((row: SqliteRow) => ({
|
||||
id: row.id,
|
||||
runId: row.run_id,
|
||||
agentId: row.agent_id,
|
||||
seq: row.seq,
|
||||
eventType: row.event_type,
|
||||
stream: row.stream,
|
||||
level: row.level,
|
||||
message: row.message,
|
||||
payload: row.payload ? JSON.parse(row.payload) : null,
|
||||
createdAt: row.created_at,
|
||||
}));
|
||||
},
|
||||
|
||||
updateRunActor(runId: string, actorType: string, actorId: string | null): void {
|
||||
getDb().prepare(
|
||||
"UPDATE runs SET actor_type = ?, actor_id = ?, updated_at = ? WHERE id = ?"
|
||||
|
||||
Reference in new issue
Block a user