fix(dsm): sync context to currentStore to prevent leaking between concurrent handlers#7395
Open
robcarlan-datadog wants to merge 10 commits intomasterfrom
Open
fix(dsm): sync context to currentStore to prevent leaking between concurrent handlers#7395robcarlan-datadog wants to merge 10 commits intomasterfrom
robcarlan-datadog wants to merge 10 commits intomasterfrom
Conversation
Contributor
Overall package sizeSelf size: 4.5 MB Dependency sizes| name | version | self size | total size | |------|---------|-----------|------------| | import-in-the-middle | 2.0.3 | 76.87 kB | 808.03 kB | | dc-polyfill | 0.1.10 | 26.73 kB | 26.73 kB |🤖 This report was automatically generated by heaviest-objects-in-the-universe |
BenchmarksBenchmark execution time: 2026-02-02 20:13:23 Comparing candidate commit 2493ae6 in PR branch Found 0 performance improvements and 0 performance regressions! Performance is the same for 228 metrics, 32 unstable metrics. |
Collaborator
|
FYI: Due to changes in |
…king When processing concurrent Kafka messages, the DSM context was being set via enterWith() on AsyncLocalStorage but not synced to ctx.currentStore. Since ctx.currentStore is what gets returned from bindStart and bound to async continuations via runStores, this caused DSM context to leak between concurrent message handlers. The fix syncs the DSM context to ctx.currentStore after DSM operations complete, ensuring each handler's async continuations maintain the correct DSM context. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Add tests that verify DSM context is properly scoped to each handler's async continuations when using diagnostic channels with runStores. The tests verify that: 1. ctx.currentStore has dataStreamsContext after setDataStreamsContext 2. Concurrent handlers maintain isolated DSM contexts 3. DSM context persists through multiple async boundaries Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Add syncToStore helper in DSM context module that syncs DSM context from AsyncLocalStorage to ctx.currentStore after DSM operations. This fixes a race condition where DSM context was being set via enterWith() but not synced to ctx.currentStore, which is what gets bound to async continuations via store.run(). Without syncing, DSM context would leak between concurrent message handlers. Updated plugins: - kafkajs (consumer, producer) - amqplib (consumer, producer) - bullmq (consumer, producer) - rhea (consumer) - google-cloud-pubsub (consumer, producer) - aws-sdk (sqs, kinesis) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
robcarlan-datadog
force-pushed
the
rob.carlan/dsm-context-prop-race-conditions
branch
from
50700ad to
df9f01e
Compare
robcarlan-datadog
changed the title
robcarlan-datadog
changed the title
The unit test was useful for validating the fix during development but is contrived and doesn't add value as a permanent test. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
robcarlan-datadog
force-pushed
the
rob.carlan/dsm-context-prop-race-conditions
branch
from
08689a7 to
20b2e55
Compare
Add comprehensive tests for the new syncToStore helper: - Unit tests for syncToStore in context.spec.js covering normal operation, edge cases, and integration with setDataStreamsContext - Spy tests in 6 integration test files to verify syncToStore is called after produce and consume operations Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
|
✅ Tests 🎉 All green!❄️ No new flaky tests detected 🔗 Commit SHA: 2493ae6 | Docs | Datadog PR Page | Was this helpful? Give us feedback! |
Change all plugins to call DataStreamsContext.syncToStore(ctx) instead of destructuring syncToStore at import time. This allows sinon spies to intercept calls during testing. When functions are destructured at require-time, they bind to the original function reference. Spies set up later on the module object don't affect these bindings. Calling via the module object ensures spies work correctly. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
The syncToStore fix only applies to the consumer path where bindStart is used. AWS SDK plugins (SQS, Kinesis) use requestInject for producers, which doesn't need context synchronization since: 1. requestInject is called before the request is sent 2. DSM context is encoded directly into the message 3. There's no async continuation where context leaking would occur Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
tlhunter
added
datastreams
integration-kafkajs
integration-google-cloud-pubsub
integration-aws-sdk
and removed
bug
Member
|
Maybe we would be better off having some sort of bindStart static helper in the producer / consumer base classes? Something to signal that it should always happen at the end of bindStart? That would at least reduce the number of requires and hopefully ensure that the author of the next messaging plugin remembers to call the method. |
rochdev
requested changes
Feb 2, 2026
Member
rochdev
left a comment
rochdev
left a comment
There was a problem hiding this comment.
Such a helper should not be needed if we use Diagnostics Channel properly. There should either be a completely separate store or the syncing should happen implicitly.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Fixes DSM context leaking between concurrent message handlers by adding a centralized
syncToStorehelper that ensures DSM context is properly scoped to each handler's async continuations.Problem
When processing concurrent messages, DSM context was being set via
enterWith()on AsyncLocalStorage but not synced toctx.currentStore. Sincectx.currentStoreis what gets returned frombindStartand bound to async continuations viastore.run(), DSM context would leak between concurrent handlers.Root cause:
setDataStreamsContextusesenterWith()which modifies AsyncLocalStorage directly, butstore.run()(called by the diagnostic channel'srunStores) overwrites this with the value returned frombindStart. Without syncing, the DSM context set viaenterWithwas immediately lost.Solution
Added
syncToStore(ctx)helper inpackages/dd-trace/src/datastreams/context.js:All DSM-enabled plugins now call
syncToStore(ctx)after their DSM operations complete, ensuring the context is properly bound for async continuations.Changes
Core:
packages/dd-trace/src/datastreams/context.js- AddedsyncToStorehelperPlugins updated:
Tests
Unit tests (
packages/dd-trace/test/datastreams/context.spec.js):syncToStorehelper syncs DSM context toctx.currentStoresetDataStreamsContextIntegration spy tests (6 plugin test files):
syncToStoreis called after produce and consume operationsTest plan
syncToStorehelper (6 tests)Not included (different patterns)
rhea/producer.js- DSM inencodesubscription afterbindStartreturnskafkajs/batch-consumer.js- Usesstartsubscription, notbindStartaws-sdk/sns.js- Producer only, no consumer-side DSM🤖 Generated with Claude Code