Merge pull request #583 from nats-io/msg_hdr_data_len

[CHANGED] Count message headers toward subscription limits
nats-io · Sep 19, 2022 · 5b8f3aa · 5b8f3aa
2 parents 75166b4 + 79a13a8
commit 5b8f3aa
Show file tree

Hide file tree

Showing 7 changed files with 16 additions and 15 deletions.
diff --git a/src/conn.c b/src/conn.c
@@ -2592,7 +2592,6 @@ natsConn_processMsg(natsConnection *nc, char *buf, int bufLen)
     natsMsgDlvWorker *ldw = NULL;
     bool             sc   = false;
     bool             sm   = false;
-    int              dl   = 0;
     nats_MsgList     *list = NULL;
     natsMutex        *mu   = NULL;
     natsCondition    *cond = NULL;
@@ -2626,10 +2625,9 @@ natsConn_processMsg(natsConnection *nc, char *buf, int bufLen)
     s = _createMsg(&msg, nc, buf, bufLen, nc->ps->ma.hdr);
     if (s != NATS_OK)
         return s;
-    // bufLen is the overall buffer len. In presence of headers, we need
-    // to capture the real message payload data length, which has been
-    // computed as the bufLen - header size.
-    dl = msg->dataLen;
+    // bufLen is the total length of headers + data. Since headers become
+    // more and more prevalent, it makes sense to count them both toward
+    // the subscription's pending limit. So use bufLen for accounting.
 
     if (mf != NULL)
     {
@@ -2692,7 +2690,7 @@ natsConn_processMsg(natsConnection *nc, char *buf, int bufLen)
     if (!ctrlMsg)
     {
         sub->msgList.msgs++;
-        sub->msgList.bytes += dl;
+        sub->msgList.bytes += bufLen;
 
         if (((sub->msgsLimit > 0) && (sub->msgList.msgs > sub->msgsLimit))
             || ((sub->bytesLimit > 0) && (sub->msgList.bytes > sub->bytesLimit)))
@@ -2706,7 +2704,7 @@ natsConn_processMsg(natsConnection *nc, char *buf, int bufLen)
 
             // Undo stats from above.
             sub->msgList.msgs--;
-            sub->msgList.bytes -= dl;
+            sub->msgList.bytes -= bufLen;
         }
         else
         {

diff --git a/src/js.c b/src/js.c
@@ -848,7 +848,7 @@ _timeoutPubAsync(natsTimer *t, void *closure)
         }
         js->rsub->msgList.tail = m;
         js->rsub->msgList.msgs++;
-        js->rsub->msgList.bytes += m->dataLen;
+        js->rsub->msgList.bytes += natsMsg_dataAndHdrLen(m);
         natsSub_Unlock(js->rsub);
 
         js->pmHead = pm->next;

diff --git a/src/msg.h b/src/msg.h
@@ -45,6 +45,8 @@
 #define natsMsg_isTimeout(m)        (((m)->flags &   (1 << 3)) != 0)
 #define natsMsg_clearTimeout(m)     ((m)->flags  &= ~(1 << 3))
 
+#define natsMsg_dataAndHdrLen(m)    ((m)->dataLen + (m)->hdrLen)
+
 struct __natsMsg
 {
     natsGCItem          gc;

diff --git a/src/nats.c b/src/nats.c
@@ -1750,7 +1750,7 @@ _deliverMsgs(void *arg)
 
         // Update before checking closed state.
         sub->msgList.msgs--;
-        sub->msgList.bytes -= msg->dataLen;
+        sub->msgList.bytes -= natsMsg_dataAndHdrLen(msg);
 
         // Need to check for closed subscription again here.
         // The subscription could have been unsubscribed from a callback

diff --git a/src/pub.c b/src/pub.c
@@ -120,14 +120,16 @@ natsConn_publish(natsConnection *nc, natsMsg *msg, const char *reply, bool direc
             totalLen = hdrl;
         }
     }
+    // This will represent headers + data
+    totalLen += msg->dataLen;
 
-    if (!nc->initc && ((int64_t) msg->dataLen > nc->info.maxPayload))
+    if (!nc->initc && ((int64_t) totalLen > nc->info.maxPayload))
     {
         natsConn_Unlock(nc);
 
         return nats_setError(NATS_MAX_PAYLOAD,
                              "Payload %d greater than maximum allowed: %" PRId64,
-                             msg->dataLen, nc->info.maxPayload);
+                             totalLen, nc->info.maxPayload);
     }
 
     // Check if we are reconnecting, and if so check if
@@ -142,7 +144,6 @@ natsConn_publish(natsConnection *nc, natsMsg *msg, const char *reply, bool direc
         }
     }
 
-    totalLen += msg->dataLen;
     GETBYTES_SIZE(totalLen, dlb, dli)
     dlSize = (BYTES_SIZE_MAX - dli);
 

diff --git a/src/sub.c b/src/sub.c
@@ -239,7 +239,7 @@ natsSub_deliverMsgs(void *arg)
             sub->msgList.tail = NULL;
 
         sub->msgList.msgs--;
-        sub->msgList.bytes -= msg->dataLen;
+        sub->msgList.bytes -= natsMsg_dataAndHdrLen(msg);
 
         msg->next = NULL;
 
@@ -731,7 +731,7 @@ natsSub_nextMsg(natsMsg **nextMsg, natsSubscription *sub, int64_t timeout, bool
                 sub->msgList.tail = NULL;
 
             sub->msgList.msgs--;
-            sub->msgList.bytes -= msg->dataLen;
+            sub->msgList.bytes -= natsMsg_dataAndHdrLen(msg);
 
             msg->next = NULL;
 

diff --git a/test/test.c b/test/test.c
@@ -24142,7 +24142,7 @@ test_JetStreamPublishAsync(void)
         rsub->msgList.head = msg;
         rsub->msgList.tail = msg;
         rsub->msgList.msgs = 1;
-        rsub->msgList.bytes = msg->dataLen;
+        rsub->msgList.bytes = natsMsg_dataAndHdrLen(msg);
         natsCondition_Signal(rsub->cond);
         natsSub_Unlock(rsub);