diff --git a/.changeset/quiet-steps-claim-telemetry.md b/.changeset/quiet-steps-claim-telemetry.md new file mode 100644 index 0000000000..2d057609af --- /dev/null +++ b/.changeset/quiet-steps-claim-telemetry.md @@ -0,0 +1,6 @@ +--- +'@workflow/core': patch +'@workflow/world-vercel': patch +--- + +Add bounded trace attributes identifying lazy step-start claim strategies and ownership stamps. diff --git a/packages/core/src/runtime/step-executor.ts b/packages/core/src/runtime/step-executor.ts index a4136268b8..814ec69025 100644 --- a/packages/core/src/runtime/step-executor.ts +++ b/packages/core/src/runtime/step-executor.ts @@ -697,6 +697,17 @@ export async function executeStep( (params.forceOptimisticStart === true && !isOptimisticInlineStartExplicitlyDisabled())); + // Keep this on the enclosing step span from the beginning so a losing + // claim retains the strategy after the 409 is reconciled as `skipped`. + // Bare background starts and owned recovery deliberately have no value. + if (params.preclaimedStart) { + span?.setAttributes(Attribute.StepStartStrategy('batch_preclaimed')); + } else if (params.lazyStepInput !== undefined) { + span?.setAttributes( + Attribute.StepStartStrategy(optimisticStart ? 'optimistic' : 'awaited') + ); + } + let step: StartedStep; // Params for the `step_started` create on either path below. The slot // snapshot is not spread here: `createEvent` attaches it to every write, diff --git a/packages/core/src/telemetry/semantic-conventions.ts b/packages/core/src/telemetry/semantic-conventions.ts index d7d4d8f394..dbd0f7b5ae 100644 --- a/packages/core/src/telemetry/semantic-conventions.ts +++ b/packages/core/src/telemetry/semantic-conventions.ts @@ -342,6 +342,16 @@ export const StepLatencyOptimizations = SemanticConvention( 'step.latency_optimizations' ); +/** + * How the step's initial `step_started` claim was made. Only present for + * inline create claims; ordinary background starts and owned recovery remain + * unlabeled. + */ +export type StepStartStrategy = 'awaited' | 'optimistic' | 'batch_preclaimed'; +export const StepStartStrategy = SemanticConvention( + 'workflow.step_start.strategy' +); + /** Whether the step was skipped during execution */ export const StepSkipped = SemanticConvention('step.skipped'); diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index 576d77d427..9bea49d3c5 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -74,6 +74,8 @@ import { WorkflowClientVersion, WorkflowEventsTransport, WorkflowEventType, + WorkflowStepStartMode, + WorkflowStepStartOwnerStamped, WorkflowWsRequestId, WorkflowWsUrl, } from './telemetry.js'; @@ -113,7 +115,7 @@ async function fetchV4( init: { method: string; headers: Headers; body?: Uint8Array }, config: APIConfig | undefined, opName: string, - attributes?: Record + attributes?: Record ): Promise { const dispatcher = getEventsDispatcher(config); const response = await instrumentedFetch({ @@ -786,6 +788,20 @@ async function postWorkflowRunEventV4( ...WorkflowEventsTransport('http'), ...WorkflowEventType(input.eventType), ...WorkflowClientVersion(`@workflow/world-vercel/${version}`), + ...(input.eventType === 'step_started' + ? { + ...WorkflowStepStartMode( + input.payload === undefined + ? input.ownerMessageId !== undefined + ? 'single_owned_recovery' + : 'single_bare' + : 'single_lazy_create_claim' + ), + ...WorkflowStepStartOwnerStamped( + input.ownerMessageId !== undefined + ), + } + : {}), ...(input.stso !== undefined ? StepStsoMs(input.stso) : {}), ...(input.optimizations !== undefined ? StepLatencyOptimizations(input.optimizations) @@ -1041,6 +1057,18 @@ export async function createWorkflowRunEventsBatchV4( { ...WorkflowEventsTransport('http'), 'workflow.batch.bytes': body.byteLength, + ...(input.events.some((event) => event.eventType === 'step_started') + ? { + ...WorkflowStepStartMode( + input.events.some((event) => event.eventType === 'step_created') + ? 'batch_create_claim' + : 'batch_bare' + ), + ...WorkflowStepStartOwnerStamped( + input.events.some((event) => event.ownerMessageId !== undefined) + ), + } + : {}), } ); @@ -1253,6 +1281,20 @@ async function postEventFrameOverWs( ...(input.optimizations !== undefined ? StepLatencyOptimizations(input.optimizations) : {}), + ...(input.eventType === 'step_started' + ? { + ...WorkflowStepStartMode( + input.payload === undefined + ? input.ownerMessageId !== undefined + ? 'single_owned_recovery' + : 'single_bare' + : 'single_lazy_create_claim' + ), + ...WorkflowStepStartOwnerStamped( + input.ownerMessageId !== undefined + ), + } + : {}), ...NetworkProtocolName('websocket'), ...WorkflowWsUrl(wsUrl), }, diff --git a/packages/world-vercel/src/telemetry.ts b/packages/world-vercel/src/telemetry.ts index 95b4490d5e..815d95a5a9 100644 --- a/packages/world-vercel/src/telemetry.ts +++ b/packages/world-vercel/src/telemetry.ts @@ -307,6 +307,22 @@ export const WorkflowEventType = SemanticConvention( 'workflow.event.type' ); +/** Server-side classification of a step_started write. */ +export type WorkflowStepStartMode = + | 'single_lazy_create_claim' + | 'single_owned_recovery' + | 'single_bare' + | 'batch_create_claim' + | 'batch_bare'; +export const WorkflowStepStartMode = SemanticConvention( + 'workflow.step_start.mode' +); + +/** Whether a step_started write carries an inline ownership stamp. */ +export const WorkflowStepStartOwnerStamped = SemanticConvention( + 'workflow.step_start.owner_stamped' +); + /** Version of the Workflow client package issuing the request. */ export const WorkflowClientVersion = SemanticConvention( 'workflow.client.version' diff --git a/packages/world-vercel/src/ws-transport-spans.test.ts b/packages/world-vercel/src/ws-transport-spans.test.ts index 39fb1adc3c..85431ee8d5 100644 --- a/packages/world-vercel/src/ws-transport-spans.test.ts +++ b/packages/world-vercel/src/ws-transport-spans.test.ts @@ -242,7 +242,56 @@ const writeSpan = (): ReadableSpan => { return spans[0]; }; +const startedBody = (eventId = 'evnt_1') => + new Uint8Array( + encode({ + event: { + eventId, + runId: 'wrun_1', + createdAt: CREATED_AT, + eventType: 'step_started', + specVersion: 2, + correlationId: 'step_1', + eventData: { stepName: 'step' }, + }, + step: { + runId: 'wrun_1', + stepId: 'step_1', + stepName: 'step', + status: 'running', + attempt: 1, + createdAt: CREATED_AT, + updatedAt: CREATED_AT, + startedAt: CREATED_AT, + }, + }) + ); + describe('per-write client span', () => { + it('attributes a stamped lazy step claim', async () => { + await withOpenChannel(() => ({ + status: 201, + body: startedBody(), + })); + + await createWorkflowRunEventV4( + { + ...input, + eventType: 'step_started', + payload: new Uint8Array([1]), + ownerMessageId: 'msg_1', + stepName: 'step', + }, + { token: 'test-token' } + ); + + const span = writeSpan(); + expect(span.attributes['workflow.step_start.mode']).toBe( + 'single_lazy_create_claim' + ); + expect(span.attributes['workflow.step_start.owner_stamped']).toBe(true); + }); + it('emits one `http POST` CLIENT span per event write', async () => { await withOpenChannel();