Add safe weighted increments, superseding PR 1

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## 2.0.0 (release candidate)
 
+- Add `increment(key, incrementBy = 1)` for nonnegative uint32 weights, with atomic overflow rejection and collision-correct conservative updates. Supersedes the weighted-increment proposal in #1; credit to Ruslan Dzhumakaliev.
+
 - Count keys such as `__proto__`, `constructor` and `toString` without interacting with Object.prototype.
 - Return independent top-k tuples so callers cannot modify internal sketch state.
 - Match serialized-size estimates to the actual power-of-two width.

README.md

@@ -243,3 +243,17 @@ Version 2 writes capacity-preserving CMS2 sketches by default. Old files remain
 TypeScript declarations cover the package root and existing class/helper deep imports. Node TypeScript projects need `@types/node`. Runtime compatibility starts at Node 6; development and the complete test suite use Node 22 or newer. CI runs the full suite on 22/24/26 and a separate Node 6 runtime smoke test.
 
 Before releasing: `npm ci`, `npm test`, then review the archive produced by `npm pack`. The test suite itself installs the archive into an independent consumer, checks CommonJS/ESM and compiles NodeNext/Node16 TypeScript fixtures. `npm publish` runs the tests through `prepublishOnly`; publishing still requires the maintainer's authenticated release action. The version bump in this branch is preparation, not evidence of a published release.
+
+### Weighted observations
+
+`counter.increment(key, incrementBy = 1)` accepts a nonnegative integer weight up to 4,294,967,295. Zero is a no-op after key validation. Negative/fractional/nonfinite values, other types and an update exceeding the uint32 counter limit throw before mutation. The implementation raises every selected sketch bucket below `minimum + incrementBy`, matching repeated unit conservative updates for that key without looping over the weight.
+
+```js
+const views = streamcount.createViewsCounter(10);
+views.increment('page-a', 250);
+views.increment('page-a'); // 251
+```
+
+Weighted observations can aggregate complete per-key counts. Replaying only each worker's top-k list loses omitted keys and is not a general merge of CountMinSketch states. Conservative sketches also cannot be assumed to equal a sketch of the concatenated stream by simply adding their cells.
+
+The weighted-increment API was proposed by Ruslan Dzhumakaliev in [PR #1](https://github.com/jhurliman/node-streamcount/pull/1). This implementation retains that use case while adding conservative-update correctness, validation, overflow handling and type coverage.

index.d.ts

@@ -4,7 +4,8 @@ export interface DeserializeOptions { maxEntries?: number; }
 /** Constructors return structural counter objects, not instanceof-compatible instances. */
 export class CountMinSketch {
   constructor(maxEntries: number, epsilon: number, delta: number);
-  increment(key: string): void;
+  /** Add a nonnegative uint32 integer weight (default 1); zero is a no-op. */
+  increment(key: string, incrementBy?: number): void;
   getTopK(): Array<[count: number, key: string]>;
   serialize(options?: SerializeOptions): Buffer;
   static deserialize(buffer: Buffer, start?: number, length?: number, options?: DeserializeOptions): CountMinSketch;

lib/countMinSketch.js

@@ -23,20 +23,26 @@ function create(maxEntries, lgWidth, counts, hashes, heap) {
   heap.forEach(function(entry) { map[entry[1]] = entry; });
   var heapq = new MinHeap(heap, function(a, b) { return a[0] - b[0]; });
 
-  function increment(key) {
+  function increment(key, incrementBy) {
     valid.key(key);
+    incrementBy = incrementBy === undefined ? 1 : incrementBy;
+    valid.integer(incrementBy, 0, MAX_INT, 'incrementBy');
+    if (incrementBy === 0) return;
     var ix = hashing.fnv1a(key), est = MAX_INT;
     var i, j;
     for (i = 0; i < hashes.length; i++) {
       j = bucket(lgWidth, hashes[i], ix);
       est = Math.min(est, counts[i][j]);
     }
-    if (est === MAX_INT) throw new RangeError('counter would overflow uint32');
+    if (incrementBy > MAX_INT - est) throw new RangeError('counter would overflow uint32');
+    var target = est + incrementBy;
     for (i = 0; i < hashes.length; i++) {
       j = bucket(lgWidth, hashes[i], ix);
-      if (counts[i][j] === est) counts[i][j] = est + 1;
+      // Equivalent to repeated unit conservative updates: raise every
+      // selected counter below the new minimum, preserving higher collisions.
+      if (counts[i][j] < target) counts[i][j] = target;
     }
-    est++;
+    est = target;
     var probe = map[key];
     if (probe !== undefined) {
       probe[0] = est;

test/package-test.js

@@ -23,7 +23,9 @@ import HLL = require('streamcount/lib/hyperLogLog');
 import Heap = require('streamcount/lib/minHeap');
 import Random = require('streamcount/lib/prng');
 const views: api.CountMinSketch = new CMS(10, .1, .1);
-views.increment('key');
+views.increment('key'); views.increment('key', 5);
+// @ts-expect-error weights must be numbers
+views.increment('key', '5');
 const rows: Array<[number,string]> = views.getTopK();
 const restored = CMS.deserialize(views.serialize({legacy:true}), undefined, undefined, {maxEntries:10});
 const uniques: api.HyperLogLog = new HLL(.1); uniques.merge(api.createUniquesCounter());

test/runtime-smoke.js

@@ -8,3 +8,7 @@ assert.strictEqual(restored.getTopK().length, 2);
 var hll = api.createUniquesCounter(); hll.add('key');
 assert.ok(api.HyperLogLog.deserialize(hll.serialize()).count() > 0);
 assert.throws(function() { api.createViewsCounter(0); });
+
+var weighted = require('..').createViewsCounter(3);
+weighted.increment('weighted', 7);
+if (weighted.getTopK()[0][0] !== 7) throw new Error('weighted increment');

test/weighted-test.js

@@ -0,0 +1,62 @@
+'use strict';
+const {test}=require('node:test');
+const assert=require('node:assert/strict');
+const CMS=require('../lib/countMinSketch');
+function sketch(capacity=10){return new CMS(capacity,.5,.2);}
+function counters(buffer){const depth=buffer.readUInt32LE(12),width=buffer.readUInt32LE(16);return buffer.subarray(20,20+depth*width*4);}
+function seededRows(first,second){
+  const bytes=sketch().serialize(),width=bytes.readUInt32LE(16);
+  for(let row=0;row<2;row++)for(let col=0;col<width;col++)bytes.writeUInt32LE(row?second:first,20+(row*width+col)*4);
+  return CMS.deserialize(bytes);
+}
+test('weighted updates default to one and zero leaves all state unchanged',()=>{
+ const cms=sketch();const empty=cms.serialize();cms.increment('absent',0);assert.deepEqual(cms.serialize(),empty);
+ cms.increment('a');cms.increment('a',undefined);cms.increment('a',5);
+ assert.deepEqual(cms.getTopK(),[[7,'a']]);const before=cms.serialize();cms.increment('a',0);assert.deepEqual(cms.serialize(),before);
+});
+test('weights must be nonnegative uint32 integers and invalid updates are atomic',()=>{
+ const cms=sketch();cms.increment('a',3);const before=cms.serialize();
+ for(const weight of [-1,.5,NaN,Infinity,-Infinity,4294967296,Number.MAX_SAFE_INTEGER,null,'2',true,{},[]]){
+  assert.throws(()=>cms.increment('a',weight),RangeError);assert.deepEqual(cms.serialize(),before);
+ }
+ assert.throws(()=>cms.increment(123,0),TypeError);
+});
+test('weighted conservative updates raise intermediate collision buckets',()=>{
+ const cms=seededRows(5,7);cms.increment('a',10);
+ assert.deepEqual(cms.getTopK(),[[15,'a']]);
+ assert.deepEqual(CMS.deserialize(cms.serialize()).getTopK(),[[15,'a']]);
+ const unit=seededRows(5,7);for(let i=0;i<10;i++)unit.increment('a');
+ assert.deepEqual(cms.serialize(),unit.serialize());
+});
+test('weighted updates preserve collision counters above the target',()=>{
+ const cms=seededRows(5,100);cms.increment('a',10);
+ const width=cms.serialize().readUInt32LE(16),row=counters(cms.serialize()).subarray(width*4);
+ for(let i=0;i<width;i++)assert.equal(row.readUInt32LE(i*4),100);
+ assert.deepEqual(cms.getTopK(),[[15,'a']]);
+});
+test('uint32 limit and overflow reject without changing counters or top-k',()=>{
+ const cms=sketch();cms.increment('max',4294967295);const before=cms.serialize();
+ cms.increment('max',0);assert.deepEqual(cms.serialize(),before);
+ assert.throws(()=>cms.increment('max'),RangeError);assert.deepEqual(cms.serialize(),before);
+ const near=seededRows(4294967290,4294967292),snapshot=near.serialize();
+ assert.throws(()=>near.increment('a',6),RangeError);assert.deepEqual(near.serialize(),snapshot);
+ near.increment('a',5);assert.deepEqual(CMS.deserialize(near.serialize()).getTopK(),[[4294967295,'a']]);
+});
+test('collision-heavy weighted streams match repeated unit updates',()=>{
+ const bulk=sketch(3),units=sketch(3);
+ let seed=123;
+ for(let i=0;i<400;i++){
+  seed=(Math.imul(seed,1664525)+1013904223)>>>0;
+  const key=['__proto__','constructor','λ','a','b','c','d'][seed%7],weight=(seed>>>8)%19;
+  bulk.increment(key,weight);for(let j=0;j<weight;j++)units.increment(key);
+  assert.deepEqual(counters(bulk.serialize()),counters(units.serialize()));
+  const order=(a,b)=>a[1].localeCompare(b[1]);assert.deepEqual(bulk.getTopK().sort(order),units.getTopK().sort(order));
+ }
+});
+test('weighted counts survive legacy and CMS2 serialization and further updates',()=>{
+ const cms=sketch();cms.increment('weighted',20);
+ for(const options of [{},{legacy:true}]){
+  const restored=CMS.deserialize(cms.serialize(options),undefined,undefined,options.legacy?{maxEntries:10}:undefined);
+  restored.increment('weighted',5);assert.deepEqual(restored.getTopK(),[[25,'weighted']]);
+ }
+});