diff --git a/.changeset/cloud-appender-durable-shutdown.md b/.changeset/cloud-appender-durable-shutdown.md new file mode 100644 index 0000000000..0c6afd6823 --- /dev/null +++ b/.changeset/cloud-appender-durable-shutdown.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/agent-core-v2": patch +--- + +Make CloudAppender shutdown durable: serialize flush() to prevent concurrent buffer races, add deadline-bounded shutdown() with AbortController that hands unsent events to durable storage and replays v2 spool data before completing. diff --git a/packages/agent-core-v2/src/app/telemetry/cloudAppender.ts b/packages/agent-core-v2/src/app/telemetry/cloudAppender.ts index 860065e1ec..97ceb2f38e 100644 --- a/packages/agent-core-v2/src/app/telemetry/cloudAppender.ts +++ b/packages/agent-core-v2/src/app/telemetry/cloudAppender.ts @@ -1,11 +1,13 @@ /** - * `telemetry` domain — `CloudAppender`, an `ITelemetryAppender` that + * `telemetry` domain (L1) — `CloudAppender`, an `ITelemetryAppender` that * batches events, drops non-primitive properties, redacts PII from string * values, enriches events with common context, and posts them to the * telemetry endpoint through `CloudTransport`, which persists failed events - * through the `storage` byte layer. Reads host facts (`clientIdentity`, env, - * platform/arch) from `IBootstrapService`; `createCloudAppender` assembles - * one from a `ServicesAccessor` so hosts only supply identity facts. + * through the `storage` byte layer (`IFileSystemStorageService`). Reads host + * facts (env, platform/arch) from `IBootstrapService`; + * `createCloudAppender` assembles one from a `ServicesAccessor` so hosts only + * supply identity facts. + * * App-scoped; independent of `@moonshot-ai/kimi-telemetry`. */ @@ -74,6 +76,7 @@ export function createCloudAppender( const DEFAULT_FLUSH_THRESHOLD = 50; const DEFAULT_FLUSH_INTERVAL_MS = 30_000; +const DEFAULT_SHUTDOWN_TIMEOUT_MS = 5_000; export class CloudAppender implements ITelemetryAppender { private readonly transport: CloudTransport; @@ -85,6 +88,10 @@ export class CloudAppender implements ITelemetryAppender { private buffer: EnrichedCloudEvent[] = []; private flushTimer: ReturnType | null = null; + private flushInFlight: Promise | null = null; + private shutdownController: AbortController | null = null; + private shutDown = false; + constructor(options: CloudAppenderOptions) { this.deviceId = options.deviceId; this.sessionId = options.sessionId ?? null; @@ -105,6 +112,7 @@ export class CloudAppender implements ITelemetryAppender { } track(event: string, properties?: TelemetryProperties): void { + if (this.shutDown) return; const eventSessionId = properties?.['sessionId']; const enriched: EnrichedCloudEvent = { event_id: randomUUID().replaceAll('-', ''), @@ -137,15 +145,70 @@ export class CloudAppender implements ITelemetryAppender { } async flush(): Promise { + const prev = this.flushInFlight; + const flushPromise = (async () => { + if (prev !== null) { + await prev.catch(() => {}); + } + if (this.buffer.length === 0) return; + await this.doFlush(); + })(); + this.flushInFlight = flushPromise; + await flushPromise; + } + + private async doFlush(): Promise { if (this.buffer.length === 0) return; const events = this.buffer; this.buffer = []; - await this.transport.send(events); + const signal = this.shutdownController?.signal; + await this.transport.send(events, signal); } - async shutdown(): Promise { + async shutdown(deadlineMs?: number): Promise { + if (this.shutDown) return; + this.shutDown = true; + this.stopPeriodicFlush(); - await this.flush(); + + const deadline = deadlineMs ?? Date.now() + DEFAULT_SHUTDOWN_TIMEOUT_MS; + this.shutdownController = new AbortController(); + const signal = this.shutdownController.signal; + const remainingMs = Math.max(0, deadline - Date.now()); + + let deadlineTimer: ReturnType | undefined; + if (remainingMs > 0) { + deadlineTimer = setTimeout(() => { + this.shutdownController?.abort(new Error('shutdown deadline expired')); + }, remainingMs); + deadlineTimer.unref?.(); + } else { + this.shutdownController.abort(new Error('shutdown deadline already expired')); + } + + try { + // Bound shutdown even when a pre-existing flush is already running: + // a send started before shutdownController existed has no abort signal, + // so a hung request could block the server close past the deadline. + await Promise.race([ + this.flush().catch(() => {}), + new Promise((resolve) => { + const t = setTimeout(() => resolve(), remainingMs); + t.unref?.(); + }), + ]); + + if (this.buffer.length > 0) { + await this.transport.saveToDisk(this.buffer).catch(() => {}); + this.buffer = []; + } + + await this.transport.retryDiskEvents(signal).catch(() => {}); + } finally { + if (deadlineTimer !== undefined) { + clearTimeout(deadlineTimer); + } + } } startPeriodicFlush(): void { diff --git a/packages/agent-core-v2/src/app/telemetry/cloudTransport.ts b/packages/agent-core-v2/src/app/telemetry/cloudTransport.ts index 1c740e90d9..f7803bfe6f 100644 --- a/packages/agent-core-v2/src/app/telemetry/cloudTransport.ts +++ b/packages/agent-core-v2/src/app/telemetry/cloudTransport.ts @@ -1,6 +1,6 @@ /** - * `telemetry` domain — `CloudTransport`, the HTTP transport for cloud - * telemetry. Posts enriched events to the telemetry endpoint with Bearer + * `telemetry` domain (L1) — `CloudTransport`, the HTTP transport behind + * `CloudAppender`. Posts enriched events to the telemetry endpoint with Bearer * auth, retry, and a byte-store fallback for failed events, persisted through * the `storage` byte layer (`IFileSystemStorageService`) under the `telemetry` scope. * App-scoped; independent of `@moonshot-ai/kimi-telemetry`. @@ -139,10 +139,12 @@ export class CloudTransport { await this.storage.write(TELEMETRY_SCOPE, key, textEncoder.encode(text)); } - async retryDiskEvents(): Promise { + async retryDiskEvents(signal?: AbortSignal): Promise { + if (signal?.aborted === true) return; const keys = await this.storage.list(TELEMETRY_SCOPE, FAILED_PREFIX); const now = this.now(); for (const key of keys) { + if (signal?.aborted === true) return; if (!key.startsWith(FAILED_PREFIX) || !key.endsWith(JSONL_SUFFIX)) continue; const createdAt = parseFailedTimestamp(key); if (createdAt === undefined || now - createdAt > DISK_EVENT_MAX_AGE_MS) { @@ -163,10 +165,11 @@ export class CloudTransport { } try { - await this.sendHttp(payload); + await this.sendHttp(payload, signal); await this.storage.delete(TELEMETRY_SCOPE, key); } catch (error) { if (error instanceof TransientCloudError) continue; + if (signal?.aborted === true) return; } } } diff --git a/packages/agent-core-v2/test/app/telemetry/cloudAppender.test.ts b/packages/agent-core-v2/test/app/telemetry/cloudAppender.test.ts index 7d70c3d4f1..a76c36df7f 100644 --- a/packages/agent-core-v2/test/app/telemetry/cloudAppender.test.ts +++ b/packages/agent-core-v2/test/app/telemetry/cloudAppender.test.ts @@ -11,7 +11,7 @@ import { import { FileStorageService } from '#/persistence/backends/node-fs/fileStorageService'; import { CloudAppender, type CloudAppenderOptions } from '#/app/telemetry/cloudAppender'; -import { stubBootstrap, stubClientIdentity } from '../bootstrap/stubs'; +import { stubBootstrap } from '../bootstrap/stubs'; interface CapturedRequest { readonly url: string; @@ -50,7 +50,7 @@ function baseOptions( const { homeDir: dir = '', storage, ...rest } = overrides; return { storage: storage ?? new FileStorageService(dir), - bootstrap: { ...stubBootstrap(), clientIdentity: { ...stubClientIdentity, version: '1.0.0' } }, + bootstrap: { ...stubBootstrap(), clientVersion: '1.0.0' }, deviceId: 'dev', appName: 'test-app', sleep: async () => {}, @@ -292,4 +292,127 @@ describe('CloudAppender', () => { resetUnexpectedErrorHandler(); } }); + + describe('shutdown durability', () => { + it('serializes concurrent flush calls without losing events', async () => { + const requests: CapturedRequest[] = []; + const appender = new CloudAppender( + baseOptions({ + homeDir, + fetchImpl: makeFetch((req) => { + requests.push(req); + return okResponse(); + }), + }), + ); + + // Track multiple events + appender.track('e1'); + appender.track('e2'); + appender.track('e3'); + + // Fire multiple concurrent flushes — they must serialize, not race + await Promise.all([appender.flush(), appender.flush(), appender.flush()]); + + // All events must have been sent (possibly in multiple batches, but + // the total event count must be 3) + const totalEvents = requests.reduce( + (sum, req) => sum + req.body.events.length, + 0, + ); + expect(totalEvents).toBe(3); + }); + + it('shutdown is idempotent — calling it twice only flushes once', async () => { + let sends = 0; + const appender = new CloudAppender( + baseOptions({ + homeDir, + fetchImpl: makeFetch(() => { + sends += 1; + return okResponse(); + }), + }), + ); + + appender.track('e1'); + await appender.shutdown(); + await appender.shutdown(); // Second call should be a no-op + expect(sends).toBe(1); + }); + + it('shutdown respects the deadline and hands unsent events to disk', async () => { + const appender = new CloudAppender( + baseOptions({ + homeDir, + // Fetch that never resolves within the deadline + fetchImpl: makeFetch( + () => new Promise((resolve) => setTimeout(() => resolve(okResponse()), 10_000)), + ), + }), + ); + + appender.track('e1'); + appender.track('e2'); + + // Shutdown with a very short deadline (already expired) + await appender.shutdown(Date.now()); + + // Events should have been saved to disk + const files = readdirSync(join(homeDir, 'telemetry')).filter((f) => + f.startsWith('failed_'), + ); + expect(files.length).toBeGreaterThanOrEqual(1); + }); + + it('shutdown replays spool data from disk', async () => { + let sends = 0; + let shouldFail = true; + const appender = new CloudAppender( + baseOptions({ + homeDir, + fetchImpl: makeFetch(() => { + if (shouldFail) return statusResponse(500); + sends += 1; + return okResponse(); + }), + }), + ); + + // First flush fails → events go to disk + appender.track('disk_event'); + await appender.flush(); + expect( + readdirSync(join(homeDir, 'telemetry')).filter((f) => f.startsWith('failed_')), + ).toHaveLength(1); + + // Now make fetch succeed and call shutdown — it should replay disk events + shouldFail = false; + await appender.shutdown(); + + // Disk file should be cleaned up after successful replay + expect( + readdirSync(join(homeDir, 'telemetry')).filter((f) => f.startsWith('failed_')), + ).toHaveLength(0); + expect(sends).toBe(1); // The replayed event was sent + }); + + it('track() after shutdown is silently ignored', async () => { + let sends = 0; + const appender = new CloudAppender( + baseOptions({ + homeDir, + fetchImpl: makeFetch(() => { + sends += 1; + return okResponse(); + }), + }), + ); + + appender.track('before_shutdown'); + await appender.shutdown(); + appender.track('after_shutdown'); // Should be ignored + expect(sends).toBe(1); + }); + }); }); diff --git a/packages/kap-server/src/services/telemetry.ts b/packages/kap-server/src/services/telemetry.ts index 25efe53505..6009d49180 100644 --- a/packages/kap-server/src/services/telemetry.ts +++ b/packages/kap-server/src/services/telemetry.ts @@ -79,15 +79,5 @@ export async function shutdownServerTelemetry( ): Promise { telemetry.registration?.dispose(); if (telemetry.appender === undefined) return; - let timer: ReturnType | undefined; - try { - await Promise.race([ - telemetry.appender.shutdown(), - new Promise((resolve) => { - timer = setTimeout(resolve, Math.max(0, deadlineMs - Date.now())); - }), - ]); - } finally { - if (timer !== undefined) clearTimeout(timer); - } + await telemetry.appender.shutdown(deadlineMs); }