From d31d68bd5f59e904fc61f6dabdb9efa972487f44 Mon Sep 17 00:00:00 2001 From: luoye520ww <100058663+luoye520ww@users.noreply.github.com> Date: Tue, 14 Jul 2026 07:55:00 +0800 Subject: [PATCH] feat(sse): persist bounded resume cursors --- .../services/sse-resume-cursor-store.test.ts | 91 ++++++++++++ src/main/services/sse-resume-cursor-store.ts | 131 ++++++++++++++++++ src/shared/sse-cursor.ts | 21 +++ 3 files changed, 243 insertions(+) create mode 100644 src/main/services/sse-resume-cursor-store.test.ts create mode 100644 src/main/services/sse-resume-cursor-store.ts create mode 100644 src/shared/sse-cursor.ts diff --git a/src/main/services/sse-resume-cursor-store.test.ts b/src/main/services/sse-resume-cursor-store.test.ts new file mode 100644 index 000000000..e93863863 --- /dev/null +++ b/src/main/services/sse-resume-cursor-store.test.ts @@ -0,0 +1,91 @@ +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' +import { join } from 'node:path' +import { tmpdir } from 'node:os' +import { describe, expect, it } from 'vitest' +import { SseResumeCursorStore } from './sse-resume-cursor-store' + +const cursor = (scopeId: string, lastSequence: number, updatedAt: string, runtimeGeneration = 'g1') => ({ + scopeId, + streamId: 'stream-1', + lastSequence, + runtimeGeneration, + updatedAt +}) + +describe('SseResumeCursorStore', () => { + it('persists acknowledged cursors atomically and restores them', async () => { + const root = await mkdtemp(join(tmpdir(), 'kun-sse-cursor-')) + const path = join(root, 'nested', 'cursors.json') + try { + const store = new SseResumeCursorStore(path) + expect(await store.save(cursor('thread-1', 4, '2026-07-14T00:00:00.000Z'))).toBe(true) + expect(await store.get('thread-1')).toEqual(cursor('thread-1', 4, '2026-07-14T00:00:00.000Z')) + const restored = new SseResumeCursorStore(path) + expect(await restored.get('thread-1')).toEqual(cursor('thread-1', 4, '2026-07-14T00:00:00.000Z')) + expect(JSON.parse(await readFile(path, 'utf8')).version).toBe(1) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it('does not move a cursor backwards within the same runtime generation', async () => { + const root = await mkdtemp(join(tmpdir(), 'kun-sse-cursor-')) + try { + const store = new SseResumeCursorStore(join(root, 'cursors.json')) + expect(await store.save(cursor('thread-1', 10, '2026-07-14T00:00:01.000Z'))).toBe(true) + expect(await store.save(cursor('thread-1', 9, '2026-07-14T00:00:02.000Z'))).toBe(false) + expect(await store.get('thread-1')).toEqual(cursor('thread-1', 10, '2026-07-14T00:00:01.000Z')) + expect(await store.save(cursor('thread-1', 1, '2026-07-14T00:00:03.000Z', 'g2'))).toBe(true) + expect(await store.get('thread-1')).toEqual(cursor('thread-1', 1, '2026-07-14T00:00:03.000Z', 'g2')) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it('serializes concurrent writes and trims oldest scopes', async () => { + const root = await mkdtemp(join(tmpdir(), 'kun-sse-cursor-')) + try { + const store = new SseResumeCursorStore(join(root, 'cursors.json'), { maxEntries: 2 }) + await Promise.all([ + store.save(cursor('old', 1, '2026-07-14T00:00:00.000Z')), + store.save(cursor('newer', 2, '2026-07-14T00:00:02.000Z')), + store.save(cursor('newest', 3, '2026-07-14T00:00:03.000Z')) + ]) + expect(await store.get('old')).toBeNull() + expect(await store.list()).toHaveLength(2) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it('fails open on malformed data without overwriting the evidence', async () => { + const root = await mkdtemp(join(tmpdir(), 'kun-sse-cursor-')) + const path = join(root, 'cursors.json') + try { + await writeFile(path, '{not-json', 'utf8') + const store = new SseResumeCursorStore(path) + expect(await store.get('thread-1')).toBeNull() + expect(await readFile(path, 'utf8')).toBe('{not-json') + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + + it('ignores entries whose storage key does not match the cursor scope', async () => { + const root = await mkdtemp(join(tmpdir(), 'kun-sse-cursor-')) + const path = join(root, 'cursors.json') + try { + await writeFile(path, JSON.stringify({ + version: 1, + cursors: { + 'wrong-key': cursor('thread-1', 7, '2026-07-14T00:00:00.000Z') + } + }), 'utf8') + const store = new SseResumeCursorStore(path) + expect(await store.get('thread-1')).toBeNull() + expect(await store.list()).toEqual([]) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) +}) diff --git a/src/main/services/sse-resume-cursor-store.ts b/src/main/services/sse-resume-cursor-store.ts new file mode 100644 index 000000000..74163fbe9 --- /dev/null +++ b/src/main/services/sse-resume-cursor-store.ts @@ -0,0 +1,131 @@ +import { mkdir, readFile } from 'node:fs/promises' +import { dirname } from 'node:path' +import { atomicWriteFile } from '../../../kun/src/adapters/file/atomic-write.js' +import { + SSE_RESUME_CURSOR_SCHEMA_VERSION, + SseResumeCursorFileSchema, + SseResumeCursorSchema, + type SseResumeCursor +} from '../../shared/sse-cursor' + +const DEFAULT_MAX_ENTRIES = 256 + +function sameGeneration(left: SseResumeCursor, right: SseResumeCursor): boolean { + return !left.runtimeGeneration || !right.runtimeGeneration || left.runtimeGeneration === right.runtimeGeneration +} + +/** + * Small, bounded, single-writer store for renderer resume cursors. + * + * It deliberately does not delete a malformed file: a corrupt cursor file is + * evidence for diagnostics, while a fresh in-memory state lets the app recover + * without losing the authoritative thread history. + */ +export class SseResumeCursorStore { + private readonly maxEntries: number + private loaded = false + private entries = new Map() + private operationTail: Promise = Promise.resolve() + + constructor( + private readonly filePath: string, + options: { maxEntries?: number } = {} + ) { + const maxEntries = options.maxEntries ?? DEFAULT_MAX_ENTRIES + if (!Number.isSafeInteger(maxEntries) || maxEntries < 1 || maxEntries > 10_000) { + throw new Error('maxEntries must be a safe integer between 1 and 10000') + } + this.maxEntries = maxEntries + } + + async get(scopeId: string): Promise { + return this.enqueue(async () => { + await this.ensureLoaded() + const cursor = this.entries.get(scopeId) + return cursor ? { ...cursor } : null + }) + } + + async list(): Promise { + return this.enqueue(async () => { + await this.ensureLoaded() + return [...this.entries.values()] + .sort((left, right) => right.updatedAt.localeCompare(left.updatedAt)) + .map((cursor) => ({ ...cursor })) + }) + } + + async save(cursor: SseResumeCursor): Promise { + return this.enqueue(async () => { + await this.ensureLoaded() + const normalized = SseResumeCursorSchema.parse(cursor) + const previous = this.entries.get(normalized.scopeId) + if ( + previous && + sameGeneration(previous, normalized) && + normalized.lastSequence < previous.lastSequence + ) { + return false + } + this.entries.set(normalized.scopeId, normalized) + this.trim() + await this.persist() + return true + }) + } + + async clear(scopeId: string): Promise { + return this.enqueue(async () => { + await this.ensureLoaded() + const removed = this.entries.delete(scopeId) + if (removed) await this.persist() + return removed + }) + } + + private enqueue(task: () => Promise): Promise { + const next = this.operationTail.then(task, task) + this.operationTail = next.then(() => undefined, () => undefined) + return next + } + + private async ensureLoaded(): Promise { + if (this.loaded) return + this.loaded = true + let raw: string + try { + raw = await readFile(this.filePath, 'utf8') + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return + throw error + } + try { + const parsed = SseResumeCursorFileSchema.parse(JSON.parse(raw)) + this.entries = new Map( + Object.entries(parsed.cursors) + .filter(([scopeId, cursor]) => scopeId === cursor.scopeId) + ) + this.trim() + } catch { + // Keep the malformed file untouched and start from a recoverable cursor set. + this.entries = new Map() + } + } + + private trim(): void { + if (this.entries.size <= this.maxEntries) return + const sorted = [...this.entries.values()] + .sort((left, right) => right.updatedAt.localeCompare(left.updatedAt)) + .slice(0, this.maxEntries) + this.entries = new Map(sorted.map((cursor) => [cursor.scopeId, cursor])) + } + + private async persist(): Promise { + await mkdir(dirname(this.filePath), { recursive: true }) + const cursors = Object.fromEntries(this.entries) + await atomicWriteFile(this.filePath, JSON.stringify({ + version: SSE_RESUME_CURSOR_SCHEMA_VERSION, + cursors + }, null, 2)) + } +} diff --git a/src/shared/sse-cursor.ts b/src/shared/sse-cursor.ts new file mode 100644 index 000000000..43642636a --- /dev/null +++ b/src/shared/sse-cursor.ts @@ -0,0 +1,21 @@ +import { z } from 'zod' + +/** A renderer-owned acknowledgement cursor persisted across process restarts. */ +export const SseResumeCursorSchema = z.object({ + scopeId: z.string().trim().min(1).max(128), + streamId: z.string().trim().min(1).max(128), + lastSequence: z.number().int().nonnegative().safe(), + runtimeGeneration: z.string().trim().min(1).max(128).optional(), + updatedAt: z.string().datetime({ offset: true }) +}).strict() + +export type SseResumeCursor = z.infer + +export const SSE_RESUME_CURSOR_SCHEMA_VERSION = 1 as const + +export const SseResumeCursorFileSchema = z.object({ + version: z.literal(SSE_RESUME_CURSOR_SCHEMA_VERSION), + cursors: z.record(z.string().min(1).max(128), SseResumeCursorSchema) +}).strict() + +export type SseResumeCursorFile = z.infer