From b41317aede9e2206d11d72bb5d0c221f4866b48e Mon Sep 17 00:00:00 2001 From: Groene AI <270696204+groeneai@users.noreply.github.com> Date: Tue, 1 Sep 2026 21:42:07 +0000 Subject: [PATCH] Backport upstream fix for the JetStream fetch lock-order inversion Cherry-picked from nats-io/nats.c `7220e9f6e081` (nats-io/nats.c#834), which fixes nats-io/nats.c#823. Released upstream in v3.10.0; this fork's branch is cut from v3.9.2 and never inherited it. `src/` is upstream's verbatim, and upstream's `test/list_test.txt` hunk 2 is dropped because it reorders `_test(SSLVerificationCallback)`, a test this fork does not have. Upstream's new test is extended here with delivery counts and a custom inbox prefix arm, so that reverting either half of the fix reddens it. The library's lock order is `nc->mu`, then `nc->subsMu`, then `sub->mu` (taken with the subscription's dispatcher lock). `js_maybeFetchMore` inverted it by publishing the JetStream pull request while still holding the subscription and dispatcher locks, so `sub->mu` was held while `natsConn_publish` acquired `nc->mu`. That closes a cycle with the two ordinary edges: `natsConn_subscribeImpl` takes `nc->mu` then the subscription lock, and `natsConn_processMsg` takes `nc->subsMu` then the subscription lock. `js_PullSubscribeAsync` produced the same inverted edge a second way, because it called `js_maybeFetchMore` while already holding both locks and `natsMutex` is `PTHREAD_MUTEX_RECURSIVE`, so the violation of that function's own stated contract was silent rather than a self-deadlock. The fix publishes outside the locks and releases them before the first fetch, so both call paths honour the contract. ClickHouse reaches the cycle whenever a NATS JetStream table subscribes a second consumer, or a second subject, on one shared connection while the first subscription's dispatcher is already fetching. ThreadSanitizer reports it as `lock-order-inversion (potential deadlock)`, and it also materializes as a real deadlock, after which unrelated DDL on the server times out with `Timeout exceeded while reading from socket` for the life of the process. Validated with the extended `test_JetStream_GH823` under ThreadSanitizer with spinning disabled, three arms x five runs: unpatched reports the inversion on 5 of 5 runs, the partial fix without the `js_PullSubscribeAsync` change still reports it on 5 of 5, and the full fix reports none on 5 of 5. On the unpatched tree the extended test also fails its second-batch assertion on 5 of 5 runs, and keeps failing it when the wait is widened to 30 seconds, so the refetch there does not merely run late. With spinning left on, the unpatched tree deadlocks on 4 of 5 runs and the fixed tree on 0 of 5. The thirteen other JetStream tests, covering the rest of the pull and pull-async family, pass unchanged on both trees, and the two backported tests pass 70 of 70 runs on the fixed tree across the three build configurations. The backport also drops the fixed-size `jsFetch::replySubject` field in favour of a buffer sized from the subscription's subject. That field was `NATS_DEFAULT_INBOX_PRE_LEN + NUID_BUFFER_LEN + 32` bytes, which assumes the default inbox prefix, but `natsOptions_SetCustomInboxPrefix` accepts any valid prefix with no length bound, so a long enough custom prefix made `snprintf` truncate the reply subject silently. `test_JetStream_GH823CustomInboxPrefix` pins that: it fails on a tree that differs from this one only by restoring the field, while `test_JetStream_GH823` passes there. Only the truncation is fixed: `_fetchIDFromSubject` still parses the fetch id at a fixed default-prefix offset, so fetch status messages under a custom prefix stay unrecognized. That is pre-existing, unchanged here, and left to upstream. Two concurrency windows come with the patch, and both are unchanged in upstream `main` today, so this adopts upstream's current state rather than diverging from it. First, `js_maybeFetchMore` now publishes the request before it adds the batch to `fetch->requestedMsgs`, so a reply that lands in between lets the default `_autoNextFetchRequest` size the next request from a stale counter and exceed `jsFetchOptions.MaxMessages`; the surplus is then destroyed undelivered, and under `js_AckNone`, which a pull consumer on a limits-retention stream accepts, the server does not redeliver it. Second, the first fetch now runs unlocked, so a `NextHandler` or a first-message callback that destroys the not-yet-returned subscription races the dispatcher's own release. Neither is reachable from ClickHouse: its JetStream consumer leaves `MaxMessages` and `MaxBytes` at zero, sets `ManualAck`, and destroys no subscription from a callback. Reported on: https://github.com/ClickHouse/ClickHouse/pull/96487#issuecomment-5499960316 CI report: https://s3.amazonaws.com/clickhouse-test-reports/json.html?PR=96487&sha=3d1580985ee545bb52169f392baa8a3bc643d4fe&name_0=PR&name_1=Integration%20tests%20%28amd_tsan%2C%205%2F6%29 Related: https://github.com/ClickHouse/ClickHouse/pull/96487 Related: https://github.com/nats-io/nats.c/issues/823 Related: https://github.com/nats-io/nats.c/pull/834 --- src/js.c | 26 ++++--- src/natsp.h | 3 - test/list_test.txt | 2 + test/test.c | 189 +++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 208 insertions(+), 12 deletions(-) diff --git a/src/js.c b/src/js.c index b84945c7e..65b57440a 100644 --- a/src/js.c +++ b/src/js.c @@ -1815,7 +1815,7 @@ js_checkFetchedMsg(natsSubscription *sub, natsMsg *msg, uint64_t fetchID, bool c } static natsStatus -_sendPullRequest(natsConnection *nc, const char *subj, const char *rply, +_publishPullRequest(natsConnection *nc, const char *subj, const char *rply, natsBuffer *buf, jsFetchRequest *req) { natsStatus s; @@ -1991,7 +1991,7 @@ _fetch(natsMsgList *list, natsSubscription *sub, jsFetchRequest *req, bool simpl req->Batch = req->Batch - count; req->Expires = NATS_MILLIS_TO_NANOS(timeout); req->NoWait = noWait; - s = _sendPullRequest(nc, subj, rply, &buf, req); + s = _publishPullRequest(nc, subj, rply, &buf, req); } IFOK(s, natsSub_nextMsg(&msg, sub, timeout, true)); if (s == NATS_OK) @@ -2909,6 +2909,7 @@ natsStatus js_maybeFetchMore(natsSubscription *sub, jsFetch *fetch) { jsFetchRequest req = {.Expires = 0}; + if (fetch->opts.NextHandler == NULL) return NATS_OK; @@ -2923,27 +2924,34 @@ js_maybeFetchMore(natsSubscription *sub, jsFetch *fetch) req.NoWait = fetch->opts.NoWait; req.Heartbeat = fetch->opts.Heartbeat * 1000 * 1000; // ns, go time.Duration + size_t replySubjectSize = 1 + strlen(sub->subject) + 20; + char *replySubject = NATS_MALLOC(replySubjectSize); + if (replySubject == NULL) + return nats_setDefaultError(NATS_NO_MEMORY); + char buffer[128]; natsBuffer buf; natsBuf_InitWithBackend(&buf, buffer, 0, sizeof(buffer)); nats_lockSubAndDispatcher(sub); - jsSub *jsi = sub->jsi; jsi->inFetch = true; jsi->fetchID++; - snprintf(fetch->replySubject, sizeof(fetch->replySubject), "%.*s%" PRIu64, + snprintf(replySubject, replySubjectSize, "%.*s%" PRIu64, (int)strlen(sub->subject) - 1, sub->subject, // exclude the last '*' jsi->fetchID); - natsStatus s = _sendPullRequest(sub->conn, jsi->nxtMsgSubj, fetch->replySubject, &buf, &req); + nats_unlockSubAndDispatcher(sub); + + natsStatus s = _publishPullRequest(sub->conn, jsi->nxtMsgSubj, replySubject, &buf, &req); if (s == NATS_OK) { + nats_lockSubAndDispatcher(sub); fetch->requestedMsgs += req.Batch; + nats_unlockSubAndDispatcher(sub); } - nats_unlockSubAndDispatcher(sub); - natsBuf_Destroy(&buf); + NATS_FREE(replySubject); return NATS_UPDATE_ERR_STACK(s); } @@ -3086,14 +3094,14 @@ js_PullSubscribeAsync(natsSubscription **newsub, jsCtx *js, const char *subject, natsTimer_Reset(jsi->hbTimer, dur); } + nats_unlockSubAndDispatcher(sub); + if (s == NATS_OK) { // Send the first fetch request. s = js_maybeFetchMore(sub, fetch); } - nats_unlockSubAndDispatcher(sub); - if (s != NATS_OK) { natsSubscription_Destroy(sub); diff --git a/src/natsp.h b/src/natsp.h index 80164238b..1a6cc1410 100644 --- a/src/natsp.h +++ b/src/natsp.h @@ -389,9 +389,6 @@ typedef struct __jsFetch // Timer for the fetch expiration. We leverage the existing jsi->hbTimer for // checking missed heartbeats. natsTimer *expiresTimer; - - // Matches jsi->fetchID - char replySubject[NATS_DEFAULT_INBOX_PRE_LEN + NUID_BUFFER_LEN + 32]; // big enough for {INBOX}.number } jsFetch; typedef struct __jsSub diff --git a/test/list_test.txt b/test/list_test.txt index a7a2218a1..8a2045581 100644 --- a/test/list_test.txt +++ b/test/list_test.txt @@ -77,6 +77,8 @@ _test(IPResolutionOrder) _test(IsClosed) _test(IsReconnectingAndStatus) _test(IsValidSubscriber) +_test(JetStream_GH823) +_test(JetStream_GH823CustomInboxPrefix) _test(JetStreamBackOffRedeliveries) _test(JetStreamContext) _test(JetStreamContextDomain) diff --git a/test/test.c b/test/test.c index 039971bd9..d55d33ec6 100644 --- a/test/test.c +++ b/test/test.c @@ -28932,6 +28932,20 @@ _recvPullAsync(natsConnection *nc, natsSubscription *sub, natsMsg *msg, natsMutex_Unlock(arg->m); } +static void +_recvPullAsyncCount(natsConnection *nc, natsSubscription *sub, natsMsg *msg, + void *closure) +{ + struct threadArg *arg = (struct threadArg *)closure; + + natsMutex_Lock(arg->m); + arg->sum++; + natsCondition_Signal(arg->c); + natsMutex_Unlock(arg->m); + + natsMsg_Destroy(msg); +} + static void _completePullAsync(natsConnection *nc, natsSubscription *sub, natsStatus exitStatus, void *closure) @@ -28994,6 +29008,181 @@ _testBatchCompleted(struct threadArg *args, natsSubscription *sub, natsStatus ex return result; } + +static bool _GH823_nextHandler(int *messages, int64_t *maxBytes, natsSubscription *sub, void *closure) +{ + *messages = 5; + return true; +} + +static int +_GH823_waitForMsgs(struct threadArg *arg, int count) +{ + natsStatus s = NATS_OK; + int sum; + + natsMutex_Lock(arg->m); + while ((s != NATS_TIMEOUT) && (arg->sum < count)) + s = natsCondition_TimedWait(arg->c, arg->m, 2000); + sum = arg->sum; + natsMutex_Unlock(arg->m); + + return sum; +} + +void test_JetStream_GH823(void) +{ + natsStatus s = NATS_OK; + jsErrCode jerr = 0; + jsStreamConfig sc; + struct threadArg args; + struct threadArg args2; + const int numMsgs = 5000; + natsConnection *ncSub = NULL; + jsCtx *jsSub = NULL; + + JS_SETUP(2, 9, 2); + + s = _createDefaultThreadArgsForCbTests(&args); + if (s == NATS_OK) + s = _createDefaultThreadArgsForCbTests(&args2); + if (s != NATS_OK) + FAIL("Unable to setup test"); + + test("Create stream for foo, bar: "); + jsStreamConfig_Init(&sc); + sc.Name = "TEST"; + sc.Subjects = (const char *[2]){"foo","bar"}; + sc.SubjectsLen = 2; + s = js_AddStream(NULL, js, &sc, NULL, &jerr); + testCond((s == NATS_OK) && (jerr == 0)); + + test("Publish thousands of test messages: "); + for (int i=0; i= 6); + + test("Create the second async pull subscriber and start receiving: "); + natsSubscription *sub2 = NULL; + s = js_PullSubscribeAsync(&sub2, jsSub, "bar", NULL, _recvPullAsyncCount, &args2, &so, NULL, &jerr); + testCond(s == NATS_OK); + + test("Second subscriber received a second batch: "); + testCond(_GH823_waitForMsgs(&args2, 6) >= 6); + + natsSubscription_Destroy(sub1); + natsSubscription_Destroy(sub2); + jsCtx_Destroy(jsSub); + natsConnection_Destroy(ncSub); + + JS_TEARDOWN + _destroyDefaultThreadArgs(&args); + _destroyDefaultThreadArgs(&args2); +} + +void test_JetStream_GH823CustomInboxPrefix(void) +{ + natsStatus s = NATS_OK; + jsErrCode jerr = 0; + jsStreamConfig sc; + struct threadArg args; + const int numMsgs = 10; + char prefix[97]; + natsOptions *subOpts = NULL; + natsConnection *ncSub = NULL; + jsCtx *jsSub = NULL; + natsSubscription *sub = NULL; + jsOptions so; + + JS_SETUP(2, 9, 2); + + s = _createDefaultThreadArgsForCbTests(&args); + if (s != NATS_OK) + FAIL("Unable to setup test"); + + test("Create stream for foo: "); + jsStreamConfig_Init(&sc); + sc.Name = "TEST"; + sc.Subjects = (const char *[1]){"foo"}; + sc.SubjectsLen = 1; + s = js_AddStream(NULL, js, &sc, NULL, &jerr); + testCond((s == NATS_OK) && (jerr == 0)); + + test("Publish test messages: "); + for (int i=0; (s == NATS_OK) && (i= 1); + + natsSubscription_Destroy(sub); + jsCtx_Destroy(jsSub); + natsConnection_Destroy(ncSub); + natsOptions_Destroy(subOpts); + + JS_TEARDOWN + _destroyDefaultThreadArgs(&args); +} + + + void test_JetStreamSubscribePullAsync(void) { natsStatus s;