fix(telemetry): await SDK export slot cleanup between requests

This commit is contained in:
07akioni committed 2026-09-25 18:47:23 +08:00
1 parent 3dd52bfd9b
commit b97ef29e79
5 files changed
+8 -5

No files matched your search

@@ -27,8 +27,8 @@
en: 78d933a0f391cc58
zh: 0552999de92573b1
/deepseek-ai-dsh-session-telemetry-otel/use-this-package/failures-and-shutdown:
en: 1e5379dd884037db
zh: f9f53bcd044ecb17
en: a66831cb9de57e04
zh: 3ada28bb868e0932
/deepseek-ai-dsh-session-telemetry-otel/understand-the-implementation:
en: ee672a6a6f0fa02d
zh: c3bab8e6b60e768c
@@ -73,7 +73,7 @@ The backend measures each record once with the SDK OTLP JSON serializer, includi
### Failures and shutdown
Invalid byte limits, non-positive queue/timer values, `maxExportBatchSize > maxQueueSize`, invalid endpoints, and invalid shutdown deadlines fail at load. Processor defaults are 2,048 queued records, 512 records per request, a 1,000 ms scheduling delay, and a 30,000 ms request watchdog. Byte limits may split a count batch into multiple serial HTTP requests. `exportTimeoutMillis` warns per request but never frees an unsettled transport slot; the SDK transport owns network timeout/retry. Shutdown drains these requests until `shutdownTimeoutMillis`; expiry discards the remaining queue and prevents further sends, while an active request may still settle. Large prefixes can therefore remain partially unsent at CLI exit.
Invalid byte limits, non-positive queue/timer values, `maxExportBatchSize > maxQueueSize`, invalid endpoints, and invalid shutdown deadlines fail at load. Processor defaults are 2,048 queued records, 512 records per request, a 1,000 ms scheduling delay, and a 30,000 ms request watchdog. Byte limits may split a count batch into multiple serial HTTP requests. Each request waits for SDK export-queue cleanup after its callback before the next starts. `exportTimeoutMillis` warns per request but never frees an unsettled transport slot; the SDK transport owns network timeout/retry. Shutdown drains these requests until `shutdownTimeoutMillis`; expiry discards the remaining queue and prevents further sends, while an active request may still settle. Large prefixes can therefore remain partially unsent at CLI exit.
-----
@@ -73,7 +73,7 @@ kind: "package-reference"
### 失败与关闭
无效字节上限、非正队列/计时值、`maxExportBatchSize > maxQueueSize`、无效 endpoint 或关闭期限在加载时失败。processor 默认队列为 2,048 条、单请求最多 512 条、调度延迟 1,000 ms、请求监测期限 30,000 ms。字节上限可能将按条数划分的批次拆成多个串行 HTTP 请求。`exportTimeoutMillis` 针对每个请求告警,但不会释放未结束的传输槽位;SDK transport 负责网络超时和重试。关闭时发送这些请求直到 `shutdownTimeoutMillis`,到期后丢弃剩余队列并禁止后续发送,在途请求仍可能结束。因此较大的授权前缀在 CLI 退出时可能只发送了一部分。
无效字节上限、非正队列/计时值、`maxExportBatchSize > maxQueueSize`、无效 endpoint 或关闭期限在加载时失败。processor 默认队列为 2,048 条、单请求最多 512 条、调度延迟 1,000 ms、请求监测期限 30,000 ms。字节上限可能将按条数划分的批次拆成多个串行 HTTP 请求。每个请求在回调后等待 SDK 导出队列清理完成,才启动下一个请求。`exportTimeoutMillis` 针对每个请求告警,但不会释放未结束的传输槽位;SDK transport 负责网络超时和重试。关闭时发送这些请求直到 `shutdownTimeoutMillis`,到期后丢弃剩余队列并禁止后续发送,在途请求仍可能结束。因此较大的授权前缀在 CLI 退出时可能只发送了一部分。
-----
@@ -133,6 +133,8 @@ class SessionLogProcessor implements LogRecordProcessor {
const records = this.queue.splice(0, count).map(entry => entry.record)
this.bytes -= bytes
await this.send(records)
// The export callback precedes removal from the SDK concurrency queue.
await this.exporter.forceFlush()
}
} finally { this.active = undefined }
}
@@ -106,13 +106,14 @@ it('preserves full nested events without truncation or ambient headers and retai
it.each([CompressionAlgorithm.GZIP, CompressionAlgorithm.NONE])('bounds %s requests and preserves nested content', async (compression) => {
const target = await collector()
const { sender } = reporter(target.endpoint, { exporter: { url: target.endpoint, compression } })
const { sender, onFailure } = reporter(target.endpoint, { exporter: { url: target.endpoint, compression, concurrencyLimit: 1 } })
const records = [0,1,2].map(seq => sessionRecord('中"\\'.repeat(100000), seq))
for (const record of records) sender.reportSessionLog(record)
await sender.shutdown()
expect(target.captures.length).toBeGreaterThan(1)
expect(target.captures.every(c => c.bytes <= SESSION_LOG_MAX_REQUEST_BYTES)).toBe(true)
expect(contents(target.captures).map(parseContent)).toEqual(records.map(r => r.event))
expect(onFailure).not.toHaveBeenCalled()
})
it('admits an exact request and diagnoses an oversized record only once', async () => {