diff --git a/CHANGELOG.md b/CHANGELOG.md index b7f7f74..14f98b7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,4 +11,12 @@ ### Release compatibility -Runtime minimum becomes Node 6 because Buffer.alloc is now used; test development requires modern Node (CI: 22/24/26). Dropping previously advertised Node 0.6 support requires a major release. No version has been bumped yet. Top-k tie ordering is unspecified and may change. The existing binary format is retained; it does not record the original top-k capacity when a sketch is serialized before filling, so that capacity cannot be fully recovered. Malformed-input/deserialization validation remains a follow-up before a release is considered complete. +Runtime minimum becomes Node 6 because Buffer.alloc is now used; test development requires modern Node (CI: 22/24/26). Dropping previously advertised Node 0.6 support requires a major release. No version has been bumped yet. Top-k tie ordering is unspecified and may change. The follow-up below adds versioned serialization and validates malformed inputs. See SERIALIZATION.md for legacy import/export and the major-release migration. + +## Release validation follow-up + +- Validate construction probabilities, allocation bounds, keys and exact serialized byte windows; reject malformed data before unbounded allocation. +- Preserve CountMinSketch capacity in CMS2 output, read legacy bytes, and offer explicit legacy capacity/import and legacy export options. See SERIALIZATION.md. +- Reject overflowing counters atomically and validate HLL merges before mutation. +- Fix the signed-minimum hash bucket edge case and retain the existing mapping for all other hashes. +- Add legacy golden fixtures and tests for truncation, malformed metadata, invalid Unicode, capacity preservation and overflow. diff --git a/SERIALIZATION.md b/SERIALIZATION.md new file mode 100644 index 0000000..16dea23 --- /dev/null +++ b/SERIALIZATION.md @@ -0,0 +1,24 @@ +# Serialization and migration + +CountMinSketch writes the versioned **CMS2** format by default. Readers automatically accept the old untagged layout. Integers are unsigned 32-bit little-endian, except legacy key lengths (one byte). + +CMS2: magic bytes `CMS2`, original `maxEntries`, log2(width), depth, width, `depth * width` row-major counters, hash count (equal to depth), that many odd hash multipliers, entry count, then each entry's count, UTF-8 byte length and key bytes. Each key length is uint32. Legacy omits the first eight bytes and uses uint8 key lengths. + +To write bytes for an old reader, use `sketch.serialize({ legacy: true })`. This rejects keys longer than 255 UTF-8 bytes and cannot preserve capacity. New files cannot be read by old package versions. + +Old files do not contain capacity. Import them with `CountMinSketch.deserialize(buffer, start, length, { maxEntries: originalCapacity })` if known. If omitted, legacy capacity defaults to the stored entry count, or one for an empty sketch. A supplied capacity must be at least the stored entry count. Versioned files preserve capacity automatically and reject conflicting overrides. + +HyperLogLog retains its layout: uint32 `32 - log2(registerCount)`, float64 scale factor, then uint32 registers. Legacy files with 2/4/8 registers remain readable, although new counters allocate at least 16. Saturation at the limit of the 32-bit hash universe reports Infinity rather than NaN. + +Readers honor the exact `(start, length)` byte window; omitted length means the rest of the buffer after start. Trailing or truncated bytes, invalid dimensions, invalid UTF-8, duplicate keys, inconsistent entry counts and impossible registers reject synchronously. Buffers from concatenated streams must be passed with their exact length. + +## Bounds + +- Probabilities/error rates must be finite numbers strictly between zero and one. Factory defaults apply only when an argument is omitted. +- CMS: 1–1,048,576 capacity; 1–64 rows; power-of-two width from 4–1,048,576; no more than 4,194,304 cells in total. +- HLL: newly constructed register arrays contain 16–1,048,576 registers. +- Keys must be valid Unicode strings, at most 1,048,576 UTF-8 bytes. Empty strings are allowed. +- Serialized buffers are limited to 64 MiB. Limits are checked before data-dependent allocation. +- Counts cannot exceed uint32. An increment that would overflow rejects before changing state. + +These validation rules and the default CMS2 output are breaking changes and belong in the next major release. The legacy hash mapping is retained, with the signed-minimum bucket calculation corrected so it cannot produce a negative array index. diff --git a/index.js b/index.js index 1eb944d..773ae13 100644 --- a/index.js +++ b/index.js @@ -1,4 +1,5 @@ var HyperLogLog = require('./lib/hyperLogLog'); +var valid = require('./lib/validation'); var CountMinSketch = require('./lib/countMinSketch'); exports.createUniquesCounter = createUniquesCounter; @@ -20,7 +21,7 @@ exports.PRNG = require('./lib/prng'); * tradeoff. 0.01 is the default. */ function createUniquesCounter(stdError) { - return new HyperLogLog(stdError || 0.01); + return new HyperLogLog(stdError === undefined ? 0.01 : stdError); } /** @@ -40,7 +41,7 @@ function createUniquesCounter(stdError) { * tradeoff. 0.0001 is the default. */ function createViewsCounter(topEntryCount, errFactor, failRate) { - return new CountMinSketch(topEntryCount, errFactor || 0.002, failRate || 0.0001); + return new CountMinSketch(topEntryCount, errFactor === undefined ? 0.002 : errFactor, failRate === undefined ? 0.0001 : failRate); } /** @@ -50,21 +51,19 @@ function createViewsCounter(topEntryCount, errFactor, failRate) { * numbers. */ function getUniquesObjSize(stdError) { - var acc = 1.04 / stdError; - var k = Math.ceil(Math.log(acc * acc) / Math.LN2); - return 12 + Math.pow(2, k) * 4; + return 12 + valid.hllLayout(stdError === undefined ? 0.01 : stdError) * 4; } /** * Returns the serialized size of a views counter (CountMinSketch) object in * bytes given an errFactor and failRate. NOTE: This does not include the size * of the serialized MinHeap which includes the size of each unique ID (up to a - * max of topEntryCount) plus 5 bytes overhead per entry. NOTE2: The memory + * max of topEntryCount) plus 8 bytes overhead per entry. NOTE2: The memory * usage will be higher than this number since we serialize 32-bit integers but * JavaScript uses 64-bit numbers. */ function getViewsObjSize(errFactor, failRate) { - var depth = Math.max(Math.ceil(Math.log(1.0 / failRate)), 1); - var width = Math.pow(2, Math.ceil(Math.log(Math.ceil(Math.E / errFactor)) / Math.LN2)); - return 4 + 8 + depth * width * 4 + 4 + depth * 4 + 4; + var layout = valid.cmsLayout(errFactor === undefined ? 0.002 : errFactor, + failRate === undefined ? 0.0001 : failRate); + return 28 + layout.depth * layout.width * 4 + layout.depth * 4; } diff --git a/lib/countMinSketch.js b/lib/countMinSketch.js index f196a6b..7ae2e88 100644 --- a/lib/countMinSketch.js +++ b/lib/countMinSketch.js @@ -1,303 +1,140 @@ +'use strict'; var MinHeap = require('./minHeap'); var hashing = require('./hashing'); var PRNG = require('./prng'); - -module.exports = CountMinSketch; - -var INT_SIZE = 32; +var valid = require('./validation'); var MAX_INT = 0xFFFFFFFF; +var MAGIC = 0x32534D43; // ASCII CMS2, never a valid legacy width exponent. +module.exports = CountMinSketch; -/** - * Count-Min Sketch is an algorithm for estimating frequency counts for large - * amounts of data. It is useful in real-time web analytics, such as counting - * the number of times every video on a site has been watched. - * - * See and - * - * - * @param {Number} maxEntries Maximum number of members to track frequency - * counts for. Only this many of the most frequently occurring members - * will be returned by getTopK(). - * @param {Number} epsilon The maximum error when answering a query will be - * within a factor of epsilon. - * @param {Number} delta The probability of getting the answer for a query - * completely wrong. From (0-1). - * @param {Number} lgWidth Internal use. - * @param {Array} counts Internal use. - * @param {Array} hashFunctions Internal use. - * @param {Array} heap Internal use. - */ -function CountMinSketch(maxEntries, epsilon, delta, lgWidth, counts, hashFunctions, heap) { - var i; - var mapLen = 0; - var map = Object.create(null); - - if (maxEntries) { - // Depth of the 2D storage array. Equal to the number of hash functions - var depth = Math.max(Math.ceil(Math.log(1.0 / delta)), 1); - // Width of the 2D storage array. Equal to the number of buckets for each - // hash function - var width = Math.ceil(Math.E / epsilon); - - // Round width up to a power of 2. This implementation uses the multiply- - // shift family of hashing functions, requiring the number of hashing - // buckets to be a power of two and the hash function constants to be - // positive odd integers - lgWidth = Math.ceil(log2(width)); - width = Math.pow(2, lgWidth); - - // Initialize the columns of the 2D storage array - counts = new Array(depth); - - // Initialize the array of random integers that define each hash function - hashFunctions = new Array(depth); - - // Initialize the backing store for the priority queue. Later it will store - // tuples of the form [count, key] - heap = []; - - // Create a random number generator with a constant seed to get - // reproducible results - var prng = new PRNG(1); - for (i = 0; i < depth; i++) { - // Generate a random odd positive integer defining each hash function - hashFunctions[i] = randomOddInt(prng); - - // Initialize each row of the 2D storage array, filling it with zeros - counts[i] = new Array(width); - for (var j = 0; j < width; j++) - counts[i][j] = 0; - } - } else { - // Since maxEntries was not specified, assume we are constructing from - // deserialized data - mapLen = heap.length; - for (i = 0; i < heap.length; i++) - map[heap[i][1]] = heap[i]; +function CountMinSketch(maxEntries, epsilon, delta) { + valid.integer(maxEntries, 1, valid.MAX_ENTRIES, 'maxEntries'); + var layout = valid.cmsLayout(epsilon, delta); + var counts = [], hashes = [], prng = new PRNG(1); + for (var i = 0; i < layout.depth; i++) { + hashes.push((Math.floor(prng.random() * 30) << 1) | 1); + counts.push(new Array(layout.width).fill(0)); } + return create(maxEntries, layout.lgWidth, counts, hashes, []); +} - var heapq = new MinHeap(heap, sortAsc); +function create(maxEntries, lgWidth, counts, hashes, heap) { + var map = Object.create(null); + heap.forEach(function(entry) { map[entry[1]] = entry; }); + var heapq = new MinHeap(heap, function(a, b) { return a[0] - b[0]; }); - /** - * Record an observation of the given key. - * @param {String} key Key to increment the observation count for. - */ function increment(key) { - // Map the key to an integer value - var ix = hashing.fnv1a(key); - var est = MAX_INT; + valid.key(key); + var ix = hashing.fnv1a(key), est = MAX_INT; var i, j; - - // Find the lowest stored value in the corresponding buckets for this key - for (i = 0; i < hashFunctions.length; i++) { - j = multiplyShift(lgWidth, hashFunctions[i], ix); + for (i = 0; i < hashes.length; i++) { + j = bucket(lgWidth, hashes[i], ix); est = Math.min(est, counts[i][j]); } - - // Conservative update. Only update the corresponding buckets with the - // lowest observed value, with the intuition that buckets containing higher - // values are due to a collision - for (i = 0; i < hashFunctions.length; i++) { - j = multiplyShift(lgWidth, hashFunctions[i], ix); - if (counts[i][j] === est) - counts[i][j] = est + 1; + if (est === MAX_INT) throw new RangeError('counter would overflow uint32'); + for (i = 0; i < hashes.length; i++) { + j = bucket(lgWidth, hashes[i], ix); + if (counts[i][j] === est) counts[i][j] = est + 1; } - - // Update the priority queue with the updated [count, key] tuple - updateHeap(key, est + 1); - } - - function updateHeap(key, est) { - // If we have already hit maxEntries and this count is lower than the - // smallest value in the priority queue, do nothing - if (heap[0] && heap[0][0] >= est && mapLen >= maxEntries) - return; - - // Attempt to retrieve the existing tuple for this key + est++; var probe = map[key]; - if (probe === undefined) { - // Create a new [count, key] tuple - var entry = [est, key]; - - if (mapLen < maxEntries) { - // Still growing... - heapq.push(entry); - if (map[key] === undefined) - ++mapLen; - map[key] = entry; - } else { - // Push this guy out - heapq.push(entry); - var oldEntry = heapq.pop(); - delete map[oldEntry[1]]; - --mapLen; - if (map[key] === undefined) - ++mapLen; - map[key] = entry; - } - } else { - // Update the existing tuple and re-sort the priority queue + if (probe !== undefined) { probe[0] = est; - heapq.heapify(heapq.heap.indexOf(probe)); + heapq.heapify(heap.indexOf(probe)); + } else if (heap.length < maxEntries || heap[0][0] < est) { + var entry = [est, key]; + heapq.push(entry); + map[key] = entry; + if (heap.length > maxEntries) delete map[heapq.pop()[1]]; } } - /** - * Returns a sorted list of tuples containing the estimated frequency count - * and key for the maxEntries top observed members. - * @returns {Array} An array of length maxEntries, containing arrays where - * the first value is the estimated frequency count and the second - * value is the given key. - */ - function getTopK() { - // Create a copy of the heap backing store - var vals = heapq.heap.map(function(entry) { return entry.slice(); }); - // Sort in descending order since the priority queue is sorted in ascending - // order and only maintains partial ordering - vals.sort(sortDesc); - return vals; - } - - /** - * Serializes this data structure to a binary buffer. - * @returns {Buffer} Binary buffer holding the serialized form of this - * structure. - */ - function serialize() { - var i; - - var heapLen = 0; - for (i = 0; i < heapq.heap.length; i++) - heapLen += 4 + 1 + Buffer.byteLength(heapq.heap[i][1]); - - var depth = counts.length; - var width = counts[0].length; - - var buffer = Buffer.alloc( - 4 + - 8 + counts.length * width * 4 + - 4 + hashFunctions.length * 4 + - 4 + heapLen); - - // lgWidth - var pos = 0; - buffer.writeUInt32LE(lgWidth, pos, true); - pos += 4; - // depth and width - buffer.writeUInt32LE(depth, pos, true); - pos += 4; - buffer.writeUInt32LE(width, pos, true); - pos += 4; - // counts - for (i = 0; i < depth; i++) { - var countsRow = counts[i]; - for (var j = 0; j < width; j++) - buffer.writeUInt32LE(countsRow[j], pos + j * 4, true); - pos += width * 4; - } - // hashFunctions - buffer.writeUInt32LE(hashFunctions.length, pos, true); - pos += 4; - for (i = 0; i < hashFunctions.length; i++) - buffer.writeUInt32LE(hashFunctions[i], pos + i * 4, true); - pos += hashFunctions.length * 4; - // heap - buffer.writeUInt32LE(heapq.heap.length, pos, true); - pos += 4; - for (i = 0; i < heapq.heap.length; i++) { - // Estimated count - buffer.writeUInt32LE(heapq.heap[i][0], pos, true); - pos += 4; - // Key - var keyLen = Buffer.byteLength(heapq.heap[i][1]); - buffer.writeUInt8(keyLen, pos, true); - pos++; - buffer.write(heapq.heap[i][1], pos, keyLen, 'utf8'); - pos += keyLen; - } - + function serialize(options) { + options = options || {}; + if (typeof options !== 'object') throw new TypeError('options must be an object'); + var legacy = options.legacy === true; + var bytes = (legacy ? 0 : 8) + 20 + counts.length * counts[0].length * 4 + hashes.length * 4; + heap.forEach(function(entry) { + var length = Buffer.byteLength(entry[1]); + if (legacy && length > 255) throw new RangeError('legacy keys cannot exceed 255 UTF-8 bytes'); + bytes += (legacy ? 5 : 8) + length; + }); + valid.integer(bytes, 0, valid.MAX_BYTES, 'serialized byte length'); + var buffer = Buffer.alloc(bytes), pos = 0; + function u32(x) { buffer.writeUInt32LE(x, pos); pos += 4; } + if (!legacy) { u32(MAGIC); u32(maxEntries); } + u32(lgWidth); u32(counts.length); u32(counts[0].length); + counts.forEach(function(row) { row.forEach(u32); }); + u32(hashes.length); hashes.forEach(u32); u32(heap.length); + heap.forEach(function(entry) { + u32(entry[0]); + var length = Buffer.byteLength(entry[1]); + if (legacy) buffer[pos++] = length; else u32(length); + buffer.write(entry[1], pos, length, 'utf8'); pos += length; + }); return buffer; } - - return { increment: increment, getTopK: getTopK, serialize: serialize }; + return { + increment: increment, + getTopK: function() { + return heap.map(function(entry) { return entry.slice(); }).sort(function(a, b) { return b[0] - a[0]; }); + }, + serialize: serialize + }; } -/** - * Deserialize a binary buffer into a reconstituted CountMinSketch structure. - * @param {Buffer} buffer Binary buffer holding the serialized structure. - * @param {Number} start Starting offset of the structure in the buffer. - * @param {Number} length Length of the serialized structure in the buffer. - * @returns {CountMinSketch} A CountMinSketch object. - */ -CountMinSketch.deserialize = function(buffer, start, length) { - start = start || 0; - length = length || buffer.length; - if (start + length > buffer.length) - throw new Error('start and buffer cannot go past the end of buffer'); - - var i; - - // lgWidth - var pos = start; - var lgWidth = buffer.readUInt32LE(pos, true); - pos += 4; - // depth and width - var depth = buffer.readUInt32LE(pos, true); - pos += 4; - var width = buffer.readUInt32LE(pos, true); - pos += 4; - // counts - var counts = new Array(depth); +CountMinSketch.deserialize = function(buffer, start, length, options) { + options = options || {}; + if (typeof options !== 'object') throw new TypeError('options must be an object'); + var r = valid.reader(buffer, start, length); + var first = r.u32(), version2 = first === MAGIC; + var capacity = version2 ? r.u32() : undefined; + var lgWidth = version2 ? r.u32() : first; + valid.integer(lgWidth, 2, 20, 'width exponent'); + var depth = r.u32(), width = r.u32(); + valid.integer(depth, 1, 64, 'depth'); + if (width !== Math.pow(2, lgWidth)) throw new RangeError('inconsistent sketch width'); + valid.integer(depth * width, 1, valid.MAX_CELLS, 'cell count'); + // Validate all fixed-width data availability before allocating any row. + r.need(depth * width * 4 + 4 + depth * 4 + 4); + var counts = [], hashes = [], heap = [], i, j; for (i = 0; i < depth; i++) { - counts[i] = new Array(width); - for (var j = 0; j < width; j++) - counts[i][j] = buffer.readUInt32LE(pos + j * 4, true); - pos += width * 4; + var row = new Array(width); + for (j = 0; j < width; j++) row[j] = r.u32(); + counts.push(row); } - // hashFunctions - var hashFunctionsLen = buffer.readUInt32LE(pos, true); - pos += 4; - var hashFunctions = new Array(hashFunctionsLen); - for (i = 0; i < hashFunctionsLen; i++) - hashFunctions[i] = buffer.readUInt32LE(pos + i * 4, true); - pos += hashFunctionsLen * 4; - // heap - var heapLen = buffer.readUInt32LE(pos, true); - pos += 4; - var heap = new Array(heapLen); - for (i = 0; i < heapLen; i++) { - // Estimated count - var est = buffer.readUInt32LE(pos, true); - pos += 4; - // Key - var keyLen = buffer.readUInt8(pos, true); - pos++; - var key = buffer.toString('utf8', pos, pos + keyLen, true); - pos += keyLen; - - heap[i] = [est, key]; + if (r.u32() !== depth) throw new RangeError('hash count must equal depth'); + for (i = 0; i < depth; i++) { + var hash = r.u32(); + if (!(hash & 1)) throw new RangeError('hash multipliers must be odd'); + hashes.push(hash); } - - return new CountMinSketch(null, null, null, lgWidth, counts, hashFunctions, heap); + var entries = r.u32(); + valid.integer(entries, 0, valid.MAX_ENTRIES, 'entry count'); + r.need(entries * (version2 ? 8 : 5)); + if (!version2) capacity = options.maxEntries === undefined ? Math.max(1, entries) : options.maxEntries; + valid.integer(capacity, Math.max(1, entries), valid.MAX_ENTRIES, 'maxEntries'); + if (version2 && options.maxEntries !== undefined && options.maxEntries !== capacity) + throw new RangeError('maxEntries cannot override a versioned sketch capacity'); + var keys = Object.create(null); + for (i = 0; i < entries; i++) { + var count = r.u32(); + valid.integer(count, 1, MAX_INT, 'entry count'); + var key = r.string(version2 ? r.u32() : r.u8()); + if (keys[key]) throw new RangeError('duplicate sketch key'); + keys[key] = true; + var ix = hashing.fnv1a(key); + for (j = 0; j < depth; j++) { + if (counts[j][bucket(lgWidth, hashes[j], ix)] < count) + throw new RangeError('entry exceeds its sketch counters'); + } + heap.push([count, key]); + } + r.end(); + return create(capacity, lgWidth, counts, hashes, heap); }; -function sortAsc(a, b) { - return a[0] - b[0]; -} - -function sortDesc(a, b) { - return b[0] - a[0]; -} - -function log2(x) { - return Math.log(x) / Math.LN2; -} - -function multiplyShift(m, a, x) { - return Math.abs((a * x) & MAX_INT) >> (INT_SIZE - m); -} - -function randomOddInt(prng) { - var n = Math.floor(prng.random() * (INT_SIZE - 2)); - return (n << 1) | 1; +function bucket(m, a, x) { + // Preserve the legacy hash mapping so existing serialized counters remain usable. + return Math.abs((a * x) & MAX_INT) >>> (32 - m); } diff --git a/lib/hyperLogLog.js b/lib/hyperLogLog.js index c3fb77d..d71c654 100644 --- a/lib/hyperLogLog.js +++ b/lib/hyperLogLog.js @@ -1,169 +1,75 @@ +'use strict'; var hashing = require('./hashing'); - +var valid = require('./validation'); +var POW_2_32 = 0x100000000; module.exports = HyperLogLog; -var POW_2_32 = 0xFFFFFFFF + 1; - -/** - * HyperLogLog is an algorithm for estimating the cardinality of a set. It is - * useful for real-time web analytics, such as counting the number of unique - * visitors. - * - * See - * - * @param {Number} stdError A value from (0-1) indicating the acceptable error - * rate. This controls the accuracy / memory usage tradeoff. 0.065 is a - * reasonable starting point. - * @param {Array} M Internal use. - * @param {Number} k_comp Internal use. - * @param {Number} alpha_m Internal use. - */ -function HyperLogLog(stdError, M, k_comp, alpha_m) { - var m; - - if (!M) { - // Compute the number of bits to use for register indexing - // From the original paper, stdError = 1.04/sqrt(m). - var acc = 1.04 / stdError; - var k = Math.ceil(log2(acc * acc)); - - // 32 minus register indexing bits leaves the number of bits used to count - // consecutive zeros in - k_comp = 32 - k; - // Compute the size of the register array as 2^register_indexing_bits - m = Math.pow(2, k); - - // Determine the value of the scale factor alpha_m by using hardcoded - // values from the paper for m in [16...64], otherwise a formula - alpha_m = (m == 16) ? 0.673 - : m == 32 ? 0.697 - : m == 64 ? 0.709 - : 0.7213 / (1 + 1.079 / m); - - // Initialize the register array to all zeros - this.M = new Array(m); - for (var i = 0; i < m; ++i) - this.M[i] = 0; - } else { - // Since M was not specified, assume we are constructing from deserialized - // data - m = M.length; - this.M = M; - } - - /** - * Add a member to the set. - * @param {String} key Key to add to the set. - */ - function add(key) { - // Map the key to an integer value - var hash = hashing.fnv1a(key); - // Use the k left-most bits as a register index `j` - var j = hash >>> k_comp; - // Compare the rank (position of the right-most 1-bit in the hash) with the - // existing register value, keeping the larger of the two - this.M[j] = Math.max(this.M[j], rank(hash, k_comp)); - } +function HyperLogLog(stdError) { + var m = valid.hllLayout(stdError); + return create(new Array(m).fill(0), 32 - Math.round(Math.log(m) / Math.LN2), valid.alpha(m)); +} - /** - * Count the number of unique members in the set. - * @returns {Number} Estimated cardinality of the set. - */ - function count() { - // Initial estimate based on the harmonic mean of all register values - // multiplied by scale factor alpha_m - // E = alpha_m * m^2 * sum(2^-M[j])^-1 - var i, c = 0.0; - for (i = 0; i < m; ++i) - c += 1 / Math.pow(2, this.M[i]); - var E = alpha_m * m * m / c; +function checkRegisters(M, k_comp) { + if (!Array.isArray(M) || M.length !== Math.pow(2, 32 - k_comp)) + throw new RangeError('invalid register array length'); + for (var i = 0; i < M.length; i++) valid.integer(M[i], 0, k_comp + 1, 'register rank'); +} - // Make corrections - if (E <= 5 / 2 * m) { - // Small range correction. E = m * log(m / empty_register_count) - var V = 0; - for (i = 0; i < m; ++i) { - if (this.M[i] === 0) - ++V; +function create(M, k_comp, alpha_m) { + var m = M.length; + return { + M: M, + add: function(key) { + valid.key(key); + var hash = hashing.fnv1a(key), j = hash >>> k_comp; + this.M[j] = Math.max(this.M[j], rank(hash, k_comp)); + }, + count: function() { + var sum = 0, empty = 0; + for (var i = 0; i < m; i++) { + sum += 1 / Math.pow(2, this.M[i]); + if (this.M[i] === 0) empty++; } - if (V > 0) - E = m * Math.log(m / V); - } else if (E > 1 / 30 * POW_2_32) { - // Large range correction, uses the alternative formula below - E = -POW_2_32 * Math.log(1 - E / POW_2_32); - } - - return E; - } - - /** - * Serializes this data structure to a binary buffer. - * @returns {Buffer} Binary buffer holding the serialized form of this - * structure. - */ - function serialize() { - var buffer = Buffer.alloc(12 + this.M.length * 4); - buffer.writeUInt32LE(k_comp, 0, true); - buffer.writeDoubleLE(alpha_m, 4, true); - for (var i = 0; i < this.M.length; i++) - buffer.writeUInt32LE(this.M[i], 12 + i * 4, true); - return buffer; - } - - /** - * Merge another HyperLogLog structure of the same size into this one. - * @param {HyperLogLog} hyperLogLog The structure to merge in. - */ - function merge(hyperLogLog) { - if (hyperLogLog.M.length != this.M.length) - throw new Error('cannot merge HyperLogLog structures of different size'); - - for (var i = 0; i < hyperLogLog.M.length; i++) { - if (hyperLogLog.M[i] > this.M[i]) - this.M[i] = hyperLogLog.M[i]; + var estimate = alpha_m * m * m / sum; + if (estimate <= 2.5 * m && empty > 0) return m * Math.log(m / empty); + if (estimate > POW_2_32 / 30) + return estimate >= POW_2_32 ? Infinity : -POW_2_32 * Math.log(1 - estimate / POW_2_32); + return estimate; + }, + serialize: function() { + checkRegisters(this.M, k_comp); + var buffer = Buffer.alloc(12 + m * 4); + buffer.writeUInt32LE(k_comp, 0); buffer.writeDoubleLE(alpha_m, 4); + for (var i = 0; i < m; i++) buffer.writeUInt32LE(this.M[i], 12 + i * 4); + return buffer; + }, + merge: function(other) { + if (!other || !Array.isArray(other.M) || other.M.length !== m) + throw new RangeError('cannot merge HyperLogLog structures of different size'); + // Validate the complete source before mutating the destination. + checkRegisters(other.M, k_comp); + for (var i = 0; i < m; i++) this.M[i] = Math.max(this.M[i], other.M[i]); } - } - - return { add: add, count: count, serialize: serialize, merge: merge, M: this.M }; + }; } -/** - * Deserialize a binary buffer into a reconstituted HyperLogLog structure. - * @param {Buffer} buffer Binary buffer holding the serialized structure. - * @param {Number} start Starting offset of the structure in the buffer. - * @param {Number} length Length of the serialized structure in the buffer. - * @returns {HyperLogLog} A HyperLogLog object. - */ HyperLogLog.deserialize = function(buffer, start, length) { - start = start || 0; - length = length || buffer.length; - if (start + length > buffer.length) - throw new Error('start and buffer cannot go past the end of buffer'); - if (length * 0.25 !== Math.floor(length * 0.25)) - throw new Error('length must be a multiple of 4'); - - var k_comp = buffer.readUInt32LE(start + 0, true); - var alpha_m = buffer.readDoubleLE(start + 4, true); - var m = (length - 12) * 0.25; - + var r = valid.reader(buffer, start, length); + var k_comp = r.u32(); + // Legacy files may contain fewer than 16 registers; keep them readable. + valid.integer(k_comp, 12, 31, 'register index bits'); + var m = Math.pow(2, 32 - k_comp), alpha = r.f64(); + if (!Number.isFinite(alpha) || Math.abs(alpha - valid.alpha(m)) > 1e-15) + throw new RangeError('invalid HyperLogLog scale factor'); + if (r.length !== 12 + m * 4) throw new RangeError('inconsistent register byte length'); var M = new Array(m); - for (var i = 0; i < m; i++) - M[i] = buffer.readUInt32LE(start + 12 + i * 4, true); - - return new HyperLogLog(null, M, k_comp, alpha_m); + for (var i = 0; i < m; i++) M[i] = valid.integer(r.u32(), 0, k_comp + 1, 'register rank'); + r.end(); + return create(M, k_comp, alpha); }; -function log2(x) { - return Math.log(x) / Math.LN2; -} - function rank(hash, max) { - // Returns the position of the right-most 1-bit of the binary string of hash, - // considering up to `max` bits var r = 1; - while ((hash & 1) === 0 && r <= max) { - ++r; - hash >>>= 1; // Zero-fill right shift by one - } + while ((hash & 1) === 0 && r <= max) { ++r; hash >>>= 1; } return r; } diff --git a/lib/validation.js b/lib/validation.js new file mode 100644 index 0000000..741a8cb --- /dev/null +++ b/lib/validation.js @@ -0,0 +1,81 @@ +'use strict'; + +exports.MAX_CELLS = 4194304; +exports.MAX_ENTRIES = 1048576; +exports.MAX_BYTES = 67108864; +exports.MAX_KEY_BYTES = 1048576; + +function integer(value, min, max, name) { + if (!Number.isSafeInteger(value) || value < min || value > max) + throw new RangeError(name + ' must be an integer in [' + min + ', ' + max + ']'); + return value; +} +exports.integer = integer; + +function probability(value, name) { + if (typeof value !== 'number' || !Number.isFinite(value) || value <= 0 || value >= 1) + throw new RangeError(name + ' must be finite and strictly between 0 and 1'); +} + +exports.cmsLayout = function(epsilon, delta) { + probability(epsilon, 'epsilon'); + probability(delta, 'delta'); + var depth = Math.ceil(-Math.log(delta)); + var lgWidth = Math.ceil(Math.log(Math.ceil(Math.E / epsilon)) / Math.LN2); + integer(depth, 1, 64, 'depth'); + integer(lgWidth, 2, 20, 'width exponent'); + var width = Math.pow(2, lgWidth); + integer(depth * width, 1, exports.MAX_CELLS, 'cell count'); + return { depth: depth, width: width, lgWidth: lgWidth }; +}; + +exports.hllLayout = function(stdError) { + probability(stdError, 'stdError'); + var k = Math.max(4, Math.ceil(2 * Math.log(1.04 / stdError) / Math.LN2)); + integer(k, 4, 20, 'register exponent'); + return Math.pow(2, k); +}; + +exports.alpha = function(m) { + return m === 16 ? 0.673 : m === 32 ? 0.697 : m === 64 ? 0.709 : 0.7213 / (1 + 1.079 / m); +}; + +exports.key = function(key) { + if (typeof key !== 'string') throw new TypeError('key must be a string'); + // ASCII is the usual hot path. UTF-8 roundtrips reject unpaired surrogates. + if (key.length > exports.MAX_KEY_BYTES) throw new RangeError('key is too long'); + if (/[^\x00-\x7f]/.test(key)) { + var encoded = Buffer.from(key, 'utf8'); + if (encoded.length > exports.MAX_KEY_BYTES) throw new RangeError('key is too long'); + if (encoded.toString('utf8') !== key) throw new TypeError('key must be valid Unicode'); + } +}; + +exports.reader = function(buffer, start, length) { + if (!Buffer.isBuffer(buffer)) throw new TypeError('buffer must be a Buffer'); + start = start === undefined ? 0 : start; + integer(start, 0, buffer.length, 'start'); + length = length === undefined ? buffer.length - start : length; + integer(length, 0, Math.min(buffer.length - start, exports.MAX_BYTES), 'length'); + var data = buffer.slice(start, start + length); + var pos = 0; + function need(bytes) { + if (!Number.isSafeInteger(bytes) || bytes < 0 || bytes > data.length - pos) + throw new RangeError('truncated serialized sketch'); + } + return { + need: need, + u32: function() { need(4); var x = data.readUInt32LE(pos); pos += 4; return x; }, + f64: function() { need(8); var x = data.readDoubleLE(pos); pos += 8; return x; }, + u8: function() { need(1); return data[pos++]; }, + string: function(size) { + integer(size, 0, exports.MAX_KEY_BYTES, 'key byte length'); need(size); + var raw = data.slice(pos, pos + size); pos += size; + var str = raw.toString('utf8'); + if (!Buffer.from(str, 'utf8').equals(raw)) throw new TypeError('invalid UTF-8 key'); + return str; + }, + end: function() { if (pos !== data.length) throw new RangeError('trailing serialized data'); }, + length: data.length + }; +}; diff --git a/test/countminsketch-test.js b/test/countminsketch-test.js index 1547832..ed7c0cf 100644 --- a/test/countminsketch-test.js +++ b/test/countminsketch-test.js @@ -59,7 +59,7 @@ runBatch('CountMinSketch', { var top = cms.getTopK(); var packed = cms.serialize(); - assert.equal(packed.length, 328320); + assert.equal(packed.length, 328388); var cms2 = CountMinSketch.deserialize(packed); var top2 = cms2.getTopK(); diff --git a/test/fixtures/legacy-cms.bin b/test/fixtures/legacy-cms.bin new file mode 100644 index 0000000..423c974 Binary files /dev/null and b/test/fixtures/legacy-cms.bin differ diff --git a/test/fixtures/legacy-hll.bin b/test/fixtures/legacy-hll.bin new file mode 100644 index 0000000..a05573e Binary files /dev/null and b/test/fixtures/legacy-hll.bin differ diff --git a/test/validation-test.js b/test/validation-test.js new file mode 100644 index 0000000..50911d4 --- /dev/null +++ b/test/validation-test.js @@ -0,0 +1,123 @@ +const { test } = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const path = require('node:path'); +const api = require('..'); +const { CountMinSketch: CMS, HyperLogLog: HLL } = api; +const invalid = [0, -1, NaN, Infinity, -Infinity, 1, '0.1', null, undefined]; + +test('constructors reject invalid probabilities and impractical allocations', () => { + for (const value of invalid) { + assert.throws(() => new CMS(10, value, .1), RangeError); + assert.throws(() => new CMS(10, .1, value), RangeError); + assert.throws(() => new HLL(value), RangeError); + } + for (const value of [0, -1, 1.5, Infinity, NaN, '10', 1048577]) assert.throws(() => new CMS(value, .1, .1), RangeError); + for (const value of [Number.MIN_VALUE, 1e-100]) { + assert.throws(() => new CMS(10, value, .1), RangeError); + assert.throws(() => new CMS(10, .1, value), RangeError); + assert.throws(() => new HLL(value), RangeError); + } +}); +test('factory defaults apply only to omitted values and size helpers agree', () => { + for (const value of [0, null, NaN, '0.1']) { + assert.throws(() => api.createUniquesCounter(value)); + assert.throws(() => api.createViewsCounter(10, value)); + assert.throws(() => api.getUniquesObjSize(value)); + assert.throws(() => api.getViewsObjSize(value)); + } + assert.equal(api.createUniquesCounter().serialize().length, api.getUniquesObjSize()); + assert.equal(api.createViewsCounter(10).serialize().length, api.getViewsObjSize()); + assert.equal(new HLL(.99).M.length, 16); +}); +test('versioned empty and partially full sketches preserve their capacity', () => { + for (const initial of [[], ['a'], ['a','b']]) { + const original = new CMS(5, .05, .01); + initial.forEach(key => original.increment(key)); + const restored = CMS.deserialize(original.serialize()); + for (const key of ['c','d','e','f','g']) restored.increment(key); + assert.equal(restored.getTopK().length, 5); + } +}); +test('legacy fixtures import, retain bytes on legacy export, and accept explicit capacity', () => { + const bytes = fs.readFileSync(path.join(__dirname, 'fixtures/legacy-cms.bin')); + const restored = CMS.deserialize(bytes, 0, bytes.length, {maxEntries: 5}); + assert.deepEqual(restored.serialize({legacy: true}), bytes); + for (const key of ['c','d','e']) restored.increment(key); + assert.equal(restored.getTopK().length, 5); + assert.equal(CMS.deserialize(bytes).getTopK().length, 2); + const hll = fs.readFileSync(path.join(__dirname, 'fixtures/legacy-hll.bin')); + assert.deepEqual(HLL.deserialize(hll).serialize(), hll); +}); +test('legacy empty sketches can recover capacity when supplied', () => { + const bytes = new CMS(10, .2, .1).serialize({legacy:true}); + const restored = CMS.deserialize(bytes, undefined, undefined, {maxEntries:10}); + ['a','b','c'].forEach(key => restored.increment(key)); + assert.equal(restored.getTopK().length, 3); + assert.throws(() => CMS.deserialize(bytes, undefined, undefined, {maxEntries:0})); + assert.throws(() => CMS.deserialize(new CMS(10,.2,.1).serialize(), undefined, undefined, {maxEntries:5})); +}); +test('Unicode and long keys roundtrip without truncation', () => { + const sketch = new CMS(10,.1,.1); + const keys = ['', '__proto__', 'é🙂'.repeat(100)]; + keys.forEach(key => sketch.increment(key)); + assert.deepEqual(CMS.deserialize(sketch.serialize()).getTopK(), sketch.getTopK()); + assert.throws(() => sketch.serialize({legacy:true}), /255/); + for (const key of [null, 123, {}, '\ud800', 'a'.repeat(1048577)]) { + const before = sketch.serialize(); + assert.throws(() => sketch.increment(key)); + assert.deepEqual(sketch.serialize(), before); + assert.throws(() => new HLL(.1).add(key)); + } +}); +test('every truncated payload fails, and readers honor exact byte windows', () => { + for (const [Type, bytes] of [[CMS,new CMS(2,.2,.1).serialize()], [CMS,new CMS(2,.2,.1).serialize({legacy:true})], [HLL,new HLL(.2).serialize()]]) { + for (let size=0;size Type.deserialize(bytes.subarray(0,size))); + const wrapped=Buffer.concat([Buffer.alloc(7),bytes,Buffer.alloc(3)]); + assert.deepEqual(Type.deserialize(wrapped,7,bytes.length).serialize(), Type.deserialize(bytes).serialize()); + assert.deepEqual(Type.deserialize(Buffer.concat([Buffer.alloc(7),bytes]),7).serialize(),Type.deserialize(bytes).serialize()); + assert.throws(() => Type.deserialize(wrapped,7)); + for (const value of [-1,.5,NaN,Infinity,'0',null]) { + assert.throws(() => Type.deserialize(bytes,value)); + assert.throws(() => Type.deserialize(bytes,0,value)); + } + assert.throws(() => Type.deserialize(bytes,bytes.length+1)); + assert.throws(() => Type.deserialize(bytes,0,0)); + assert.throws(() => Type.deserialize(new Uint8Array(bytes))); + } +}); +test('malformed dimensions, ranks and scale factors fail before allocation', () => { + const cms=new CMS(2,.2,.1).serialize(); + for(const [offset,value] of [[4,0],[8,32],[12,0],[12,0xffffffff],[16,1]]) { + const mutated=Buffer.from(cms);mutated.writeUInt32LE(value,offset);assert.throws(()=>CMS.deserialize(mutated)); + } + const hll=new HLL(.2).serialize(); + for(const [offset,value] of [[0,0],[0,32],[12,100]]) { + const mutated=Buffer.from(hll);mutated.writeUInt32LE(value,offset);assert.throws(()=>HLL.deserialize(mutated)); + } + for(const value of [NaN, Infinity, -1, .1]) { + const mutated=Buffer.from(hll);mutated.writeDoubleLE(value,4);assert.throws(()=>HLL.deserialize(mutated)); + } +}); +test('malformed heap keys, duplicate entries, hash counts and UTF-8 reject', () => { + const s=new CMS(2,.2,.1);s.increment('a');s.increment('b');const bytes=s.serialize(); + const depth=bytes.readUInt32LE(12), width=bytes.readUInt32LE(16); + const hashes=20+depth*width*4, heap=hashes+4+depth*4, entries=heap+4; + for(const [offset,value] of [[hashes,0],[hashes+4,2],[heap,3],[entries,0],[entries+4,0xffffffff]]) { + const b=Buffer.from(bytes);b.writeUInt32LE(value,offset);assert.throws(()=>CMS.deserialize(b)); + } + const duplicate=Buffer.from(bytes);duplicate[entries+9+8]=duplicate[entries+8];assert.throws(()=>CMS.deserialize(duplicate),/duplicate/); + const invalidUtf8=Buffer.from(bytes);invalidUtf8[entries+8]=0xff;assert.throws(()=>CMS.deserialize(invalidUtf8),/UTF-8/); +}); +test('overflowing counters reject atomically', () => { + const bytes=new CMS(2,.2,.1).serialize(); + const depth=bytes.readUInt32LE(12),width=bytes.readUInt32LE(16); + for(let offset=20;offset<20+depth*width*4;offset+=4)bytes.writeUInt32LE(0xffffffff,offset); + const sketch=CMS.deserialize(bytes);const before=sketch.serialize(); + assert.throws(()=>sketch.increment('a'),/overflow/);assert.deepEqual(sketch.serialize(),before); +}); +test('invalid merge sources reject before modifying the destination', () => { + const target=new HLL(.2);target.add('a');const before=target.serialize(); + const source=new HLL(.2);source.M[0]=2;source.M[source.M.length-1]=100; + assert.throws(()=>target.merge(source));assert.deepEqual(target.serialize(),before); +});