Skip to content

Backport upstream fix for the JetStream fetch lock-order inversion #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alexey-milovidov merged 1 commit into from
Sep 13, 2026
Merged

Backport upstream fix for the JetStream fetch lock-order inversion #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 17 additions & 9 deletions src/js.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -2909,6 +2909,7 @@ natsStatus
js_maybeFetchMore(natsSubscription *sub, jsFetch *fetch)
{
jsFetchRequest req = {.Expires = 0};

if (fetch->opts.NextHandler == NULL)
return NATS_OK;

Expand All @@ -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);
}

Expand Down Expand Up @@ -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);
Expand Down
3 changes: 0 additions & 3 deletions src/natsp.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 2 additions & 0 deletions test/list_test.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,8 @@ _test(IPResolutionOrder)
_test(IsClosed)
_test(IsReconnectingAndStatus)
_test(IsValidSubscriber)
_test(JetStream_GH823)
_test(JetStream_GH823CustomInboxPrefix)
_test(JetStreamBackOffRedeliveries)
_test(JetStreamContext)
_test(JetStreamContextDomain)
Expand Down
189 changes: 189 additions & 0 deletions test/test.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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<numMsgs; i++)
{
char buf[64];
snprintf(buf, sizeof(buf), "hello-foo-%d", i);
s = js_Publish(NULL, js, "foo", buf, (int) strlen(buf), NULL, &jerr);
if (s != NATS_OK)
break;
snprintf(buf, sizeof(buf), "hello-bar-%d", i);
s = js_Publish(NULL, js, "bar", buf, (int) strlen(buf), NULL, &jerr);
if (s != NATS_OK)
break;
}
testCond(s == NATS_OK);

test("Make a separate connection for subscribers: ");
s = natsConnection_Connect(&ncSub, NULL);
if (s == NATS_OK)
s = natsConnection_JetStream(&jsSub, ncSub, NULL);
testCond(s == NATS_OK);

test("Create the first async pull subscriber and start receiving: ");
natsSubscription *sub1 = NULL;
jsOptions so;
jsOptions_Init(&so);
so.PullSubscribeAsync.NextHandler = _GH823_nextHandler;
s = js_PullSubscribeAsync(&sub1, jsSub, "foo", NULL, _recvPullAsyncCount, &args, &so, NULL, &jerr);
testCond(s == NATS_OK);

// The next handler asks for 5 messages per request, so a 6th delivery can only
// come from a second request, that is, from the dispatcher fetching again.
test("First subscriber received a second batch: ");
testCond(_GH823_waitForMsgs(&args, 6) >= 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<numMsgs); i++)
{
char buf[64];
snprintf(buf, sizeof(buf), "hello-foo-%d", i);
s = js_Publish(NULL, js, "foo", buf, (int) strlen(buf), NULL, &jerr);
}
testCond(s == NATS_OK);

// An inbox prefix has no length bound, so a long one makes the fetch reply
// subject longer than any fixed-size buffer sized from the default prefix.
memset(prefix, 'x', sizeof(prefix)-1);
prefix[sizeof(prefix)-1] = '\0';

test("Accept a 96 character inbox prefix: ");
s = natsOptions_Create(&subOpts);
IFOK(s, natsOptions_SetCustomInboxPrefix(subOpts, prefix));
testCond(s == NATS_OK);

test("Make a separate connection for the subscriber: ");
s = natsConnection_Connect(&ncSub, subOpts);
IFOK(s, natsConnection_JetStream(&jsSub, ncSub, NULL));
testCond(s == NATS_OK);

test("Create the async pull subscriber and start receiving: ");
jsOptions_Init(&so);
so.PullSubscribeAsync.NextHandler = _GH823_nextHandler;
s = js_PullSubscribeAsync(&sub, jsSub, "foo", NULL, _recvPullAsyncCount, &args, &so, NULL, &jerr);
testCond(s == NATS_OK);

// A truncated reply subject does not match the subscription, so nothing is
// delivered at all.
test("Receive on the long reply subject: ");
testCond(_GH823_waitForMsgs(&args, 1) >= 1);

natsSubscription_Destroy(sub);
jsCtx_Destroy(jsSub);
natsConnection_Destroy(ncSub);
natsOptions_Destroy(subOpts);

JS_TEARDOWN
_destroyDefaultThreadArgs(&args);
}



void test_JetStreamSubscribePullAsync(void)
{
natsStatus s;
Expand Down