diff --git a/docs/30-unit-tdd/client-runtime-and-delegation.md b/docs/30-unit-tdd/client-runtime-and-delegation.md index ac3db54d..67e52138 100644 --- a/docs/30-unit-tdd/client-runtime-and-delegation.md +++ b/docs/30-unit-tdd/client-runtime-and-delegation.md @@ -200,9 +200,11 @@ exporter 为 10 秒,经 relay 为 35 秒,以覆盖服务端最多 32 秒的 取消已开始的 SDK 导出,也不证明排空成功;页面退出只能 best effort。Job 页面提供可选 `diagnostics_url` 链接,用户以 Job ID 在诊断端查询,不把 Grafana 查询语法写入业务层。 -`JobManager.create` 在同一个 PostgREST INSERT 中写入 SDK 注入的提交 carrier。超出 -512 UTF-8 bytes 的可选 tracestate 会被省略并计数。执行成功 claim 后创建独立 trace, -通过 SDK Span Link 关联提交;claim、取消和 close 不重写提交列。Job ID 是诊断关联字段, +`JobManager.create` 只在有效 trace context 存在时,在同一个 PostgREST INSERT 中写入 +SDK 注入的提交 carrier。遥测关闭或没有有效 traceparent 时不发送提交追踪列;缺失或 +空 tracestate 也不发送对应列,超出 512 UTF-8 bytes 的可选 tracestate 会被省略并计数。 +带上下文的提交要求部署数据库已包含这些可选列;不在提交失败后删字段重试。 +执行成功 claim 后创建独立 trace,通过 SDK Span Link 关联提交;claim、取消和 close 不重写提交列。Job ID 是诊断关联字段, 不是 trace ID。时延指标 `inkcre.operation.duration` 使用与 Core 一致的 `operation` / `outcome` 标签和 5ms 至 300s 的显式秒分桶,便于跨 Peer 聚合;日志保留 `inkcre.*` 属性。 `job.submitted`、`job.started`、`job.closed` 只从明确的业务边界发出结构化 diff --git a/packages/core/src/obsrv/telemetry.ts b/packages/core/src/obsrv/telemetry.ts index 9a0a61c1..ee319585 100644 --- a/packages/core/src/obsrv/telemetry.ts +++ b/packages/core/src/obsrv/telemetry.ts @@ -281,17 +281,18 @@ export async function observeOperation( } export function captureSubmission(parent: Context): { - submission_traceparent: string | null - submission_tracestate: string | null + submission_traceparent?: string + submission_tracestate?: string } { const carrier: Record = {} if (tracer) propagator.inject(parent, carrier, setter) + if (!carrier.traceparent) return {} const state = carrier.tracestate const oversized = state !== undefined && new TextEncoder().encode(state).length > 512 if (oversized) droppedCarrier?.add(1, { reason: 'tracestate_too_large' }) return { - submission_traceparent: carrier.traceparent ?? null, - submission_tracestate: oversized ? null : (state ?? null), + submission_traceparent: carrier.traceparent, + ...(state && !oversized ? { submission_tracestate: state } : {}), } } export function submissionLinks(job: { diff --git a/tests/e2e/peer-database.spec.ts b/tests/e2e/peer-database.spec.ts index 1014933f..5bdb112d 100644 --- a/tests/e2e/peer-database.spec.ts +++ b/tests/e2e/peer-database.spec.ts @@ -655,7 +655,7 @@ test('disabling while registration is pending prevents its later lease or worker } }) -test('disabling participation drains a dispatched claim before saving configuration', async ({ +test('untraced Job creation omits carrier columns and disabling drains its claim', async ({ page, }) => { const authorization = `Bearer ${await token()}` @@ -677,6 +677,7 @@ test('disabling participation drains a dispatched claim before saving configurat release = resolve }) let claimStored = false + let carrierlessInsertObserved = false try { await page.goto('/sources') await expect(page.getByRole('main', { name: 'Sources', exact: true })).toBeVisible() @@ -685,14 +686,16 @@ test('disabling participation drains a dispatched claim before saving configurat 'Connecting this Web Peer; database pages are available, local jobs are not running yet.' ) ).toHaveCount(0) - const created = await fetch(`${postgrestUrl}jobs`, { - method: 'POST', - headers: { ...headers, Prefer: 'return=representation' }, - body: JSON.stringify({ type, parameters: {}, timeout_seconds: 3600 }), - }) - expect(created.status).toBe(201) - id = (await created.json())[0].id await page.route('**/jobs?**', async (route) => { + if (route.request().method() === 'POST') { + const submitted = route.request().postDataJSON() + expect(submitted.type).toBe(type) + expect(submitted).not.toHaveProperty('submission_traceparent') + expect(submitted).not.toHaveProperty('submission_tracestate') + carrierlessInsertObserved = true + await route.continue() + return + } if ( route.request().method() !== 'PATCH' || route.request().postDataJSON().status !== 'running' || @@ -708,38 +711,49 @@ test('disabling participation drains a dispatched claim before saving configurat await route.fulfill({ response }) }) // Native extensions use this same shared SDK; the fixture adds no production test hook. - await page.evaluate( - async ({ id, type }) => { - type Sdk = typeof import('../../packages/core/src/index') - const globals = globalThis as typeof globalThis & { - __FEDERATION__: { - __INSTANCES__: Array<{ - options: { name: string } - loadShare(name: string): Promise<() => Sdk> - }> - } - __claimDrain: { stop(): Promise; handlerRuns: number; stopCompleted: boolean } + id = await page.evaluate(async (type) => { + type Sdk = typeof import('../../packages/core/src/index') + const globals = globalThis as typeof globalThis & { + __FEDERATION__: { + __INSTANCES__: Array<{ + options: { name: string } + loadShare(name: string): Promise<() => Sdk> + }> } - const host = globals.__FEDERATION__.__INSTANCES__.find( - (instance) => instance.options.name === 'host' - )! - const sdk = (await host.loadShare('@inkcre/core'))() - globals.__claimDrain = { - stop: () => sdk.JobManager.stopWorker(), - handlerRuns: 0, - stopCompleted: false, + __claimDrain: { + start(id: number): void + stop(): Promise + handlerRuns: number + stopCompleted: boolean } - sdk.JobManager.registerHandler(type, { - parameters: sdk.MetaConfigSchema, - canHandle: () => true, - handle: async () => { - globals.__claimDrain.handlerRuns++ - }, - }) - void sdk.JobManager.run(id!) - }, - { id, type } - ) + } + const host = globals.__FEDERATION__.__INSTANCES__.find( + (instance) => instance.options.name === 'host' + )! + const sdk = (await host.loadShare('@inkcre/core'))() + globals.__claimDrain = { + start: (id) => { + sdk.JobManager.registerHandler(type, { + parameters: sdk.MetaConfigSchema, + canHandle: () => true, + handle: async () => { + globals.__claimDrain.handlerRuns++ + }, + }) + void sdk.JobManager.run(id) + }, + stop: () => sdk.JobManager.stopWorker(), + handlerRuns: 0, + stopCompleted: false, + } + return (await sdk.JobManager.create(type, {})).id + }, type) + expect(carrierlessInsertObserved).toBe(true) + await page.evaluate((id) => { + ;( + globalThis as typeof globalThis & { __claimDrain: { start(id: number): void } } + ).__claimDrain.start(id) + }, id) await expect.poll(() => claimStored).toBe(true) const stoppedBeforeClaimReturned = await page.evaluate(async () => { const fixture = (