Add support for Network Compression

sharadraju · sharadraju · commit 4543780c5c9f · 2025-02-11T12:57:19.000+05:30
diff --git a/doc/src/release_notes.rst b/doc/src/release_notes.rst
@@ -28,7 +28,12 @@ Common Changes
 Thin Mode Changes
 +++++++++++++++++
 
-#)  Added support for External Authentication using Transport Layer Security(TLS).
+#)  Added support for external authentication using Transport Layer
+    Security (TLS) protocol.
+
+#)  Added support for Advanced Network Compression which reduces the size
+    of the network data packets transmitted over a connection for improved
+    effective network throughput and bandwidth utilization.
 
 #)  Fixed issue that does not throw Authentication error for FastAuth
     when invalid token is used with external authentication.
diff --git a/lib/connection.js b/lib/connection.js
@@ -1216,6 +1216,10 @@ class Connection extends EventEmitter {
       this._impl.isHealthy());
   }
 
+  isCompressionEnabled() {
+    return (this._impl && this._impl.isCompressionEnabled());
+  }
+
   //---------------------------------------------------------------------------
   // maxIdentifierLength
   //
diff --git a/lib/errors.js b/lib/errors.js
@@ -206,6 +206,7 @@ const ERR_HOST_NOT_FOUND = 530;
 const ERR_ANO_PACKET = 531;
 const ERR_ANO_STATUS = 532;
 const ERR_ANO_NEGOTIATION = 533;
+const ERR_DATA_COMPRESSION = 534;
 
 // Oracle SUCCESS_WITH_INFO warning start from 700
 const WRN_COMPILATION_CREATE = 700;
@@ -562,7 +563,7 @@ messages.set(ERR_ANO_STATUS,                            // NJS-532
   '%s service recieved status failure');
 messages.set(ERR_ANO_NEGOTIATION,                       // NJS-533
   'Advanced Networking Option service negotiation failed. Native Network Encryption and DataIntegrity only supported in node-oracledb thick mode.\nCause: ORA-%s');
-
+messages.set(ERR_DATA_COMPRESSION, 'Error during data compression/decompression: %s\n'); //NJS-534
 // Oracle SUCCESS_WITH_INFO warning
 messages.set(WRN_COMPILATION_CREATE,                    // NJS-700
   'creation succeeded with compilation errors');
@@ -887,6 +888,7 @@ module.exports = {
   ERR_CONFIG_PROVIDER_FAILED_TO_RETRIEVE_CONFIG,
   ERR_CONFIG_PROVIDER_NOT_SUPPORTED,
   ERR_CONFIG_PROVIDER_LOAD_FAILED,
+  ERR_DATA_COMPRESSION,
   ERR_WALLET_TYPE_NOT_SUPPORTED,
   ERR_HOST_NOT_FOUND,
   ERR_ANO_PACKET,
diff --git a/lib/impl/connection.js b/lib/impl/connection.js
@@ -409,6 +409,10 @@ class ConnectionImpl {
     errors.throwNotImplemented("getting the health of the connection");
   }
 
+  isCompressionEnabled() {
+    errors.throwNotImplemented("getting the data compression status on the connection");
+  }
+
   //---------------------------------------------------------------------------
   // ping()
   //
diff --git a/lib/oracledb.js b/lib/oracledb.js
@@ -188,6 +188,18 @@ async function _verifyOptions(options, inCreatePool) {
       "walletLocation");
     outOptions.walletLocation = options.walletLocation;
   }
+  if (options.networkCompression !== undefined) {
+    errors.assertParamPropValue(typeof options.networkCompression === 'boolean', 1,
+      "networkCompression");
+    outOptions.networkCompression = options.networkCompression;
+    outOptions.networkCompressionLevels = [];
+    outOptions.networkCompressionLevels.push('high');
+  }
+
+  if (options.networkCompressionThreshold !== undefined) {
+    errors.assertParamPropValue(Number.isInteger(options.networkCompressionThreshold), 1, "networkCompressionThreshold");
+    outOptions.networkCompressionThreshold = options.networkCompressionThreshold;
+  }
 
   //wallet content must be a string
   if (options.walletContent !== undefined) {
@@ -1337,6 +1349,7 @@ module.exports = {
     return settings.configProviderCacheTimeout;
   },
 
+
   get thin() {
     return settings.thin;
   },
@@ -1517,6 +1530,7 @@ module.exports = {
     errors.assertPropValue(typeof value === 'string', "terminal");
     settings.terminal = value;
   },
+
   set configProviderCacheTimeout(value) {
     errors.assertPropValue(Number.isInteger(value) && value >= 0,
       "configProviderCacheTimeout");
diff --git a/lib/thin/connection.js b/lib/thin/connection.js
@@ -715,6 +715,9 @@ class ThinConnectionImpl extends ConnectionImpl {
     await this._protocol.breakMessage();
   }
 
+  isCompressionEnabled() {
+    return this.nscon.compressionEnabled;
+  }
   isHealthy() {
     try {
       if (this.nscon.recvInbandNotif() === 0)
diff --git a/lib/thin/protocol/packet.js b/lib/thin/protocol/packet.js
@@ -530,7 +530,7 @@ class WritePacket extends BaseBuffer {
     if (!this.nsi.ntAdapter) {
       errors.throwErr(errors.ERR_INVALID_CONNECTION);
     }
-    this.nsi.ntAdapter.send(buf);
+    this.nsi.sendPacket(buf);
   }
 
   /**
diff --git a/lib/thin/sqlnet/constants.js b/lib/thin/sqlnet/constants.js
@@ -1,4 +1,4 @@
-// Copyright (c) 2022, 2024, Oracle and/or its affiliates.
+// Copyright (c) 2022, 2025, Oracle and/or its affiliates.
 
 //-----------------------------------------------------------------------------
 //
@@ -155,6 +155,7 @@ module.exports =  {
   NSPDAFLG: 8,    // Data flags
   NSPDADAT: 10,   // Start of Data
   NSPDAFEOF: 0x40, // "end of file"
+  NSPDAFCMP: 0x400,  // "compressed data"
 
   /* Marker Packet */
   NSPMKTYP: 8,     // marker type (see below)
@@ -201,6 +202,9 @@ module.exports =  {
   PURITY: 8, // Purity
   SID: 9, // SID
 
+  /* Network Compression Algorithms */
+  NETWORK_COMPRESSION_ZLIB: 2,
+
   /* FLAGS */
   NSNOBLOCK: 0x0001, // Do not block
 };
diff --git a/lib/thin/sqlnet/ezConnectResolver.js b/lib/thin/sqlnet/ezConnectResolver.js
@@ -57,7 +57,7 @@ const EXT_PARAM_SEP = '&';
 const DESCRIPTION_PARAMS = ["ENABLE", "FAILOVER", "LOAD_BALANCE",
   "RECV_BUF_SIZE", "SEND_BUF_SIZE", "SDU",
   "SOURCE_ROUTE", "RETRY_COUNT", "RETRY_DELAY",
-  "CONNECT_TIMEOUT", "TRANSPORT_CONNECT_TIMEOUT", "RECV_TIMEOUT", "USE_SNI"];
+  "CONNECT_TIMEOUT", "TRANSPORT_CONNECT_TIMEOUT", "RECV_TIMEOUT", "USE_SNI", "COMPRESSION"];
 /*
    DESCRIPTION
     This class takes care resolving the EZConnect format to Long TNS URL format.
@@ -438,6 +438,7 @@ class EZConnectResolver {
   initializeUrlAlias() {
     const aliasMap = new Map();
     aliasMap.set("enable", "ENABLE");
+    aliasMap.set("compression", "COMPRESSION");
     aliasMap.set("failover", "FAILOVER");
     aliasMap.set("load_balance", "LOAD_BALANCE");
     aliasMap.set("recv_buf_size", "RECV_BUF_SIZE");
diff --git a/lib/thin/sqlnet/navNodes.js b/lib/thin/sqlnet/navNodes.js
@@ -294,6 +294,19 @@ class Description {
         }
       } else if (childnv.name.toUpperCase() == "SDU") {
         this.params.sdu = childnv.atom;
+      } else if (childnv.name.toUpperCase() == "COMPRESSION") {
+        this.params.networkCompression = (childnv.atom.toLowerCase() == "yes"
+                  || childnv.atom.toLowerCase() == "on"
+                  || childnv.atom.toLowerCase() == "true");
+        this.params.networkCompressionLevels = [];
+      } else if (childnv.name.toUpperCase() == "COMPRESSION_LEVELS") {
+        const listsize = childnv.getListSize();
+        for (let i = 0; i < listsize; i++) {
+          const tmpnv = childnv.getListElement(i);
+          if (tmpnv.name.toUpperCase() == "LEVEL") {
+            this.params.networkCompressionLevels.push(tmpnv.atom.toLowerCase());
+          }
+        }
       } else if (childnv.name.toUpperCase() == "EXPIRE_TIME") {
         if (childnv.atom > 0)
           this.params.expireTime = childnv.atom;
@@ -700,7 +713,6 @@ class NavDescription extends Description {
     }
 
     desc.params = this.params;
-
     if ('connectTimeout' in this.params) {
       cs.sBuf.push("(CONNECT_TIMEOUT=" + this.params.connectTimeout + ")");
     }
diff --git a/lib/thin/sqlnet/networkSession.js b/lib/thin/sqlnet/networkSession.js
@@ -25,7 +25,7 @@
 //-----------------------------------------------------------------------------
 
 'use strict';
-
+const zlib = require('zlib');
 const Packet = require("./packet.js");
 const NTTCP = require("./ntTcp.js");
 const SessionAtts = require("./sessionAtts.js");
@@ -139,6 +139,7 @@ class NetworkSession {
     this.isBreak = false;
     this.isReset = false;
     this.breakPosted = false;
+    this.compressionEnabled = false;
   }
 
   async getAddress(addressNode, userConfig) {
@@ -190,7 +191,7 @@ class NetworkSession {
    * @param {object} connectPkt Connect Packet
    */
   _sendConnect(connectPkt) {
-    this._sendPacket(connectPkt.buf);
+    this.sendPacket(connectPkt.buf);
     if (connectPkt.overflow) {
       this._send(connectPkt.connectData, 0, connectPkt.connectDataLen);
       this.flush();
@@ -365,9 +366,36 @@ class NetworkSession {
    */
   _processPacket(packet) {
     switch (packet.type) {
-      case constants.NSPTDA: /* DATA packet */
+      case constants.NSPTDA: { /* DATA packet */
+        const size = packet.buf.length;
+        const dataFlags = packet.buf.readUInt16BE(constants.NSPDAFLG);
+        if ((dataFlags & constants.NSPDAFCMP) != 0) { // compression enabled
+          const packetHeader = packet.buf.subarray(0, constants.NSPDADAT);
+          const buffertoDeCompress = packet.buf.subarray(constants.NSPDADAT, size);
+          let deCompressedDataBuffer;
+          try {
+            if (this.sAtts.firstCompressedPacket) {
+              deCompressedDataBuffer = zlib.inflateSync(buffertoDeCompress, {finishFlush: zlib.constants.Z_SYNC_FLUSH});
+              this.sAtts.firstCompressedPacket = 0;
+            } else {
+              deCompressedDataBuffer = zlib.inflateRawSync(buffertoDeCompress, {finishFlush: zlib.constants.Z_SYNC_FLUSH});
+            }
+          } catch (err) {
+            errors.throwErr(errors.ERR_DATA_COMPRESSION, err.message);
+          }
+          const resultLength = deCompressedDataBuffer.length;
+          //concatenate packet header with decompressed data
+          packet.buf = Buffer.concat([packetHeader, deCompressedDataBuffer]);
+          const length = resultLength + constants.NSPDADAT;
+          if (this.sAtts.largeSDU) {
+            packet.buf.writeUInt32BE(length, constants.NSPHDLEN);
+          } else {
+            packet.buf.writeUInt16BE(length, constants.NSPHDLEN);
+          }
+        }
         this.rcvDatapkt.fromPacket(packet);
         break;
+      }
       case constants.NSPTMK: /* MARKER packet */
         this.markerPkt.fromPacket(packet, this);
         break;
@@ -406,16 +434,48 @@ class NetworkSession {
       const packet = await this.ntAdapter.receive();
       if (!packet)
         break;
+
       this._processPacket(packet);
-      if (packet.type !== constants.NSPTCNL)
+      if (packet.type !== constants.NSPTCNL) {
         return packet;
+      }
     }
   }
 
   /**
    * Send data (Internal)
    */
-  _sendPacket(buf) {
+  sendPacket(buf) {
+    const packetType = buf.readUInt8(constants.NSPHDTYP);
+    // only data packets need to be compressed
+    if (packetType == constants.NSPTDA) {
+      const size = buf.length;
+      let dataFlags = buf.readUInt16BE(constants.NSPDAFLG);
+      if (this.sAtts.networkCompressionEnabled && size > this.sAtts.networkCompressionThreshold) {
+        this.compressionEnabled = true;
+        const buffertoCompress = buf.subarray(constants.NSPDADAT, size);
+        const bufferHeader = buf.subarray(0, constants.NSPDADAT);
+        let compressedDataBuffer;
+        try {
+          compressedDataBuffer = zlib.deflateSync(buffertoCompress, {finishFlush: zlib.constants.Z_SYNC_FLUSH});
+        } catch (err) {
+          errors.throwErr(errors.ERR_DATA_COMPRESSION, err.message);
+        }
+        const resultLength = compressedDataBuffer.length;
+        if (resultLength < size - constants.NSPDADAT) {
+          dataFlags |= constants.NSPDAFCMP;
+          // concatenate buffer header with the compressed data
+          buf = Buffer.concat([bufferHeader, compressedDataBuffer]);
+          buf.writeUInt16BE(dataFlags, constants.NSPDAFLG);
+          const pktOffset = resultLength + constants.NSPDADAT;
+          if (this.sAtts.largeSDU) {
+            buf.writeUInt32BE(pktOffset, constants.NSPHDLEN);
+          } else {
+            buf.writeUInt16BE(pktOffset, constants.NSPHDLEN);
+          }
+        }
+      }
+    }
     this.ntAdapter.send(buf);
   }
 
@@ -434,7 +494,7 @@ class NetworkSession {
 
     this.isBreak = true;
     this.markerPkt.prepare(constants.NSPMKTD1, constants.NIQIMARK);
-    this._sendPacket(this.markerPkt.buf);
+    this.sendPacket(this.markerPkt.buf);
   }
 
   /**
@@ -444,12 +504,12 @@ class NetworkSession {
     /* If posted send Break */
     if (this.breakPosted) {
       this.markerPkt.prepare(constants.NSPMKTD1, constants.NIQBMARK);
-      this._sendPacket(this.markerPkt.buf);
+      this.sendPacket(this.markerPkt.buf);
       this.breakPosted = false;
     }
     /* Send Reset */
     this.markerPkt.prepare(constants.NSPMKTD1, constants.NIQRMARK);
-    this._sendPacket(this.markerPkt.buf);
+    this.sendPacket(this.markerPkt.buf);
 
     /* Await Reset */
     while (!this.isReset) {
@@ -503,7 +563,7 @@ class NetworkSession {
     }
 
     while (len) {
-      this._sendPacket(this.sndDatapkt.dataBuf);
+      this.sendPacket(this.sndDatapkt.dataBuf);
 
       /* If break throw error now */
       if (this.isBreak) {
@@ -528,7 +588,7 @@ class NetworkSession {
     }
     this.sndDatapkt.dataLen = this.sndDatapkt.offset;
     this.sndDatapkt.prepare2Send();
-    this._sendPacket(Buffer.from(this.sndDatapkt.dataBuf));
+    this.sendPacket(Buffer.from(this.sndDatapkt.dataBuf));
     this.sndDatapkt.dataLen = this.sndDatapkt.dataPtr;
     this.sndDatapkt.offset = this.sndDatapkt.dataPtr;
   }
@@ -581,7 +641,7 @@ class NetworkSession {
       /* Send EOF packet */
       this.sndDatapkt.dataLen = this.sndDatapkt.offset;
       this.sndDatapkt.prepare2Send(constants.NSPDAFEOF);
-      this._sendPacket(this.sndDatapkt.dataBuf);
+      this.sendPacket(this.sndDatapkt.dataBuf);
     }
     this.ntAdapter.disconnect(type);
     this.ntAdapter = null;
diff --git a/lib/thin/sqlnet/packet.js b/lib/thin/sqlnet/packet.js
@@ -125,7 +125,21 @@ function ConnectPacket(connectData, sAtts, flags = 0) {
 
   this.buf.writeUInt32BE(sAtts.sdu, constants.NSPCNLSD); /* SDU */
   this.buf.writeUInt32BE(sAtts.tdu, constants.NSPCNLTD); /* TDU */
-  this.buf.writeUInt32BE(0, constants.NSPCNCFL); /* Compression not supported */
+  let compressionFieldBuilder = 0; // we will be using just the last 2 bytes
+  /*
+     * For the 2 byte field, the first bit represent ON/OFF , second is empty
+      * After that every 4 bits represent a value in the user preference of sch
+  eme
+     * scheme values are 1 = LZO, 2 = ZLIB, 3 = GZIP ( only ZLIB supported in node-oracledb) */
+
+  if (sAtts.networkCompression) {
+    compressionFieldBuilder = constants.NSPACCFON << 8 ; // set the ON bit
+    const schemeShiftCounter = 10; // shifting by 10 puts the 4 bit field in the right place
+    if (sAtts.networkCompressionLevels.includes('high'))
+      compressionFieldBuilder |= constants.NETWORK_COMPRESSION_ZLIB << schemeShiftCounter;
+  }
+
+  this.buf.writeUInt16BE(compressionFieldBuilder, constants.NSPCNCFL); /* Compression field added */
 
   this.buf.writeUInt32BE(
     0,
@@ -240,6 +254,14 @@ function AcceptPacket(packet, sAtts) {
     sAtts.sdu = packet.buf.readUInt32BE(constants.NSPACLSD);
     sAtts.tdu = packet.buf.readUInt32BE(constants.NSPACLTD);
     sAtts.largeSDU = true;
+    this.cflag = packet.buf.readUInt8(constants.NSPACCFL); // 00001000
+    if ((this.cflag & constants.NSPACCFON) != 0) { // compression ON
+      sAtts.negotiatedNetworkCompressionScheme = (this.cflag & 0x3c) >> 2; // 0x3c
+      sAtts.networkCompressionEnabled = true;
+      sAtts.firstCompressedPacket = true;
+    } else {
+      sAtts.networkCompressionEnabled = false;
+    }
   }
 
   /* Accept flags */
diff --git a/lib/thin/sqlnet/sessionAtts.js b/lib/thin/sqlnet/sessionAtts.js
diff --git a/test/ext/sqlnet-tests/networkCompressionTest.js b/test/ext/sqlnet-tests/networkCompressionTest.js

Original file line number	Diff line number	Diff line change
`@@ -715,6 +715,9 @@ class ThinConnectionImpl extends ConnectionImpl {`
`715`	`715`	`await this._protocol.breakMessage();`
`716`	`716`	`}`
`717`	`717`
	`718`	`+ isCompressionEnabled() {`
	`719`	`+ return this.nscon.compressionEnabled;`
	`720`	`+ }`
`718`	`721`	`isHealthy() {`
`719`	`722`	`try {`
`720`	`723`	`if (this.nscon.recvInbandNotif() === 0)`
Original file line number	Diff line number	Diff line change
`@@ -530,7 +530,7 @@ class WritePacket extends BaseBuffer {`
`530`	`530`	`if (!this.nsi.ntAdapter) {`
`531`	`531`	`errors.throwErr(errors.ERR_INVALID_CONNECTION);`
`532`	`532`	`}`
`533`		`- this.nsi.ntAdapter.send(buf);`
	`533`	`+ this.nsi.sendPacket(buf);`
`534`	`534`	`}`
`535`	`535`
`536`	`536`	`/**`