From d398eda43222b8a64b3afb733f2e37646ec9300f Mon Sep 17 00:00:00 2001 From: kingwl Date: Sat, 25 Jul 2026 03:48:32 +0800 Subject: [PATCH] fix(telemetry): join overlapping flush hints; contain adoption replay per event MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second review round, both pinned red-first: - Overlapping turn-boundary flush hints now JOIN the outstanding flush promise (Promise.all) instead of displacing it: the SDK's concurrent-flush guard resolves an overlapping forceFlush() immediately, so retaining only the latest promise let shutdown() proceed while the first export was still in flight — the same silent drop the single-flush fix closed. - Adoption replay contains failures per event, matching the firehose: one rejected record is withheld fail-closed while the rest of the historical log still hands off. Wrapping the whole loop let a single failure silently skip the remainder on an already-adopted session. --- .../session-telemetry-otel/src/index.ts | 12 ++++--- .../session-telemetry-otel/tests/otel.spec.ts | 33 +++++++++++++++++++ .../session-telemetry/src/coordinator.ts | 18 ++++++---- .../session-telemetry/tests/telemetry.spec.ts | 26 +++++++++++++++ 4 files changed, 77 insertions(+), 12 deletions(-) diff --git a/packages/telemetry/session-telemetry-otel/src/index.ts b/packages/telemetry/session-telemetry-otel/src/index.ts index 8dde46828e..d415d1a7e8 100644 --- a/packages/telemetry/session-telemetry-otel/src/index.ts +++ b/packages/telemetry/session-telemetry-otel/src/index.ts @@ -147,7 +147,7 @@ export class TelemetryOtel extends Telemetry { }) } - /** The latest turn-boundary flush, retained so {@link shutdown} can order behind it. */ + /** Every not-yet-settled turn-boundary flush, retained so {@link shutdown} can order behind ALL of them. */ private inflightFlush: Promise = Promise.resolve() /** Forward the turn-boundary hint to the SDK's flush, fire-and-forget. */ @@ -157,15 +157,17 @@ export class TelemetryOtel extends Telemetry { // this once the fiber is disposed — a rejection would be SDK drift. The // settled promise is retained (not awaited): the SDK's concurrent-flush // guard makes a flush that overlaps another return WITHOUT draining, so - // shutdown must wait this one out before trusting its own flush. + // an overlapping hint resolves instantly and must JOIN the outstanding + // one, not displace it — shutdown orders behind the whole set. /* v8 ignore next -- unreachable guard: forceFlush does not reject while the provider is alive */ - this.inflightFlush = this.provider.forceFlush().catch(() => {}) + const flush = this.provider.forceFlush().catch(() => {}) + this.inflightFlush = Promise.all([this.inflightFlush, flush]).then(() => undefined) } /** * Delegate disposal to the SDK's shutdown contract: flush the queue and - * quiesce. Orders behind the last turn-boundary flush first — shutdown's - * internal flush is a no-op while one is in flight (the SDK's + * quiesce. Orders behind every outstanding turn-boundary flush first — + * shutdown's internal flush is a no-op while one is in flight (the SDK's * concurrent-flush guard), which would silently drop everything enqueued * after that flush snapshot, including the coordinator's dispose-time * `shutdown` markers. Awaited (and error-contained) by the coordinator's diff --git a/packages/telemetry/session-telemetry-otel/tests/otel.spec.ts b/packages/telemetry/session-telemetry-otel/tests/otel.spec.ts index b44b42ea04..85bfedeb1d 100644 --- a/packages/telemetry/session-telemetry-otel/tests/otel.spec.ts +++ b/packages/telemetry/session-telemetry-otel/tests/otel.spec.ts @@ -155,6 +155,39 @@ describe('TelemetryOtel wire', () => { expect(ops[0]!.record.attributes).toContainEqual({ key: 'telemetry.op', value: { stringValue: 'shutdown' } }) }) + it('orders shutdown behind the OLDEST in-flight flush when hints overlap', async () => { + // The SDK's concurrent-flush guard resolves an overlapping forceFlush() + // immediately; if the backend RETAINS only the latest flush promise, two + // back-to-back turn flushes leave shutdown awaiting the instantly-resolved + // second one while the first still exports — reopening the same silent + // drop the single-flush race test pins. + const gate = Promise.withResolvers() + const arrived = Promise.withResolvers() + const { url, captures } = await mockCollector(async (index) => { + if (index === 0) { + arrived.resolve(true) + await gate.promise + } + }) + const { ctx, fiber } = await boot(url) + const session = ctx.sessions.create(SessionId('race2'), { meta: {} }) + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + ctx.telemetry.flush!() + await arrived.promise + // Second hint while the first export is held open: resolves immediately + // under the SDK's guard and must not displace the outstanding one. + ctx.telemetry.flush!() + + const disposal = fiber.dispose() + await new Promise(resolve => setTimeout(resolve, 50)) + gate.resolve(true) + await disposal + + const ops = allRecords(captures).filter(r => r.scope === '@deepseek-ai/dsh-session-telemetry-otel/ops') + expect(ops).toHaveLength(1) + expect(ops[0]!.record.attributes).toContainEqual({ key: 'telemetry.op', value: { stringValue: 'shutdown' } }) + }) + it('passes exporter options beyond url and headers through to the SDK exporter', async () => { const { url, captures } = await mockCollector() const ctx = new Context() diff --git a/packages/telemetry/session-telemetry/src/coordinator.ts b/packages/telemetry/session-telemetry/src/coordinator.ts index e8213e588a..027c6af09e 100644 --- a/packages/telemetry/session-telemetry/src/coordinator.ts +++ b/packages/telemetry/session-telemetry/src/coordinator.ts @@ -113,15 +113,19 @@ export class TelemetryCoordinator { * @param session - the live session to adopt; a second adoption is a no-op. */ private adopt(session: Session): void { - this.contain(() => { - if (this.adopted.has(session)) return - this.adopted.add(session) - const cursor = handoffCursor.get(session) ?? -1 - for (const event of session.events) { + if (this.adopted.has(session)) return + this.adopted.add(session) + const cursor = handoffCursor.get(session) ?? -1 + // Containment is PER EVENT, matching the firehose: one rejected record + // is withheld fail-closed while the rest of the historical replay + // proceeds — wrapping the whole loop would let a single failure silently + // skip the remainder of the log on an already-adopted session. + for (const event of session.events) { + this.contain(() => { if (event.seq <= cursor) this.track(session, event) else this.capture(session, event) - } - }) + }) + } } /** Feed the chunk projection without handing off — the ≤cursor half of re-adoption. */ diff --git a/packages/telemetry/session-telemetry/tests/telemetry.spec.ts b/packages/telemetry/session-telemetry/tests/telemetry.spec.ts index 55b5ed694a..d52baa91f5 100644 --- a/packages/telemetry/session-telemetry/tests/telemetry.spec.ts +++ b/packages/telemetry/session-telemetry/tests/telemetry.spec.ts @@ -26,11 +26,15 @@ class FakeBackend implements TelemetryBackend { records: TelemetryRecord[] = [] calls: string[] = [] emitError: Error | undefined + rejectSeq: number | undefined shutdownError: Error | undefined shutdownResolved = false emit(record: TelemetryRecord): void { if (this.emitError) throw this.emitError + if (this.rejectSeq !== undefined && record.attributes['event.seq'] === this.rejectSeq) { + throw new Error(`backend rejected seq ${this.rejectSeq}`) + } this.records.push(record) this.calls.push(`emit:${String(record.attributes['event.seq'] ?? record.attributes['telemetry.op'])}`) } @@ -213,6 +217,28 @@ describe('TelemetryCoordinator adoption', () => { expect(second.ledger().map(r => r.attributes['event.type'])).toEqual(['turn/end']) }) + it('replays past a record the backend rejects: one event withheld, the rest adopted', async () => { + const backend = new FakeBackend() + const ctx = new Context() + await ctx.plugin(SessionStore) + const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {}) + const session = liveSession(ctx, 'partial') + appendTurn(session) + session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + // The backend rejects exactly the middle historical event: fail-closed + // must withhold THAT record only — an adoption replay that dies on the + // first contained failure would silently skip the rest of the log while + // the session stays marked adopted. + backend.rejectSeq = 1 + await ctx.plugin({ + name: 'fake-telemetry', + inject: ['sessions'], + apply: (inner: Context) => void new TelemetryCoordinator(inner, backend), + }) + expect(backend.ledger().map(r => r.attributes['event.seq'])).toEqual([0, 2]) + expect(warn).toHaveBeenCalled() + }) + it('re-hands the full log when no cursor survived (fresh session object)', async () => { const backend = new FakeBackend() const ctx = new Context()