From 5b12f90ca23d5eab8c39a85e4740886196deaad1 Mon Sep 17 00:00:00 2001 From: Ivan Kozlovic Date: Fri, 4 Jul 2025 11:25:02 -0600 Subject: [PATCH] [IMPROVED] JetStream: handling of publish async timeouts When publishing a JetStream message asynchronously but with a publish max wait, we would schedule an internal timeout message to our internal subscription to handle ack timeout, even if the message was actually already ack'ed. The handler for the ack would disregard that timeout message because it would not be found in the hash table but we optimize here by not scheduling if the message has already been handled in `_handleAsyncReply()`. This was reported in #880 Signed-off-by: Ivan Kozlovic --- src/conn.c | 2 +- src/js.c | 26 +++++++++++++++----------- 2 files changed, 16 insertions(+), 12 deletions(-) diff --git a/src/conn.c b/src/conn.c index 376b643d7..390b14e85 100644 --- a/src/conn.c +++ b/src/conn.c @@ -3360,7 +3360,7 @@ _processUrlString(natsOptions *opts, const char *urls) serverUrls = (char**) NATS_CALLOC(count + 1, sizeof(char*)); if (serverUrls == NULL) return NATS_NO_MEMORY; - + if (s == NATS_OK) { urlsCopy = NATS_STRDUP(urls); diff --git a/src/js.c b/src/js.c index 7b3d39880..949d1f3b6 100644 --- a/src/js.c +++ b/src/js.c @@ -842,19 +842,23 @@ _timeoutPubAsync(natsTimer *t, void *closure) while (((pm = js->pmHead) != NULL) && (pm->deadline <= now)) { - natsMsg *m = NULL; - - if (natsMsg_Create(&m, pm->subject, NULL, NULL, 0) != NATS_OK) - break; - - natsMsg_setTimeout(m); + // Check if the corresponding message is still in the hashtable. + char *id = (pm->subject+js->rpreLen); + if (natsStrHash_Get(js->pm, id) != NULL) + { + natsMsg *m = NULL; - // Best attempt, ignore NATS_SLOW_CONSUMER errors which may be returned - // here. - nats_lockSubAndDispatcher(js->rsub); - natsSub_enqueueUserMessage(js->rsub, m); - nats_unlockSubAndDispatcher(js->rsub); + if (natsMsg_Create(&m, pm->subject, NULL, NULL, 0) == NATS_OK) + { + natsMsg_setTimeout(m); + // Best attempt, ignore NATS_SLOW_CONSUMER errors which may be returned here. + nats_lockSubAndDispatcher(js->rsub); + natsSub_enqueueUserMessage(js->rsub, m); + nats_unlockSubAndDispatcher(js->rsub); + } + } + // Remove from the list. js->pmHead = pm->next; _destroyPMInfo(pm); }