@@ -4,59 +4,93 @@ import { Worker } from 'node:worker_threads';
44import { createHourlyBudget , hourlyBudgetLength } from '../src/util/hourlyBudget.js' ;
55
66/**
7- * The hourly budget (util/hourlyBudget.js) the entity registry's adoptions spend: one count per lane per hour ,
8- * shared by every worker thread through a shared buffer.
7+ * The hourly budget (util/hourlyBudget.js) the entity registry's adoptions spend: one 64-bit cell per lane ,
8+ * holding the hour and the count, shared by every worker thread through a shared buffer.
99 */
1010
1111const MODULE = new URL ( '../src/util/hourlyBudget.js' , import . meta. url ) . href ;
12+ const HOUR = 3_600_000 ;
1213const NOW = Date . UTC ( 2026 , 9 , 2 , 10 , 30 , 0 ) ;
14+ const cells = ( lanes = 2 ) =>
15+ new BigInt64Array ( new SharedArrayBuffer ( hourlyBudgetLength ( lanes ) * BigInt64Array . BYTES_PER_ELEMENT ) ) ;
1316
14- test ( 'across real worker threads the limit holds exactly: no thread multiplies it' , async ( ) => {
15- const shared = new Int32Array ( new SharedArrayBuffer ( hourlyBudgetLength ( 2 ) * Int32Array . BYTES_PER_ELEMENT ) ) ;
16- const workers = 4 ;
17- const attempts = 200 ;
18- const limit = 150 ;
19- const granted = await Promise . all (
20- Array . from (
21- { length : workers } ,
22- ( ) =>
23- new Promise ( ( resolve , reject ) => {
24- const worker = new Worker (
25- `const { parentPort, workerData } = require('node:worker_threads');
26- import(workerData.module).then(({ createHourlyBudget }) => {
27- const budget = createHourlyBudget(() => workerData.cell);
28- let granted = 0;
29- for (let i = 0; i < workerData.attempts; i++) if (budget.reserve(workerData.limit, workerData.now, 0)) granted++;
30- parentPort.postMessage(granted);
31- });` ,
32- { eval : true , workerData : { module : MODULE , cell : shared , attempts, limit, now : NOW } }
33- ) ;
34- worker . once ( 'message' , resolve ) ;
35- worker . once ( 'error' , reject ) ;
36- } )
37- )
17+ // Four real threads, released together from a gate, each try `attempts` reserves against `limit`. With `rollAt`, a
18+ // thread's attempts from that index on are made an hour later.
19+ const race = async ( shared , { attempts, limit, rollAt = null } ) => {
20+ const gate = new Int32Array ( new SharedArrayBuffer ( 8 ) ) ; // [ready, go]
21+ const runs = Array . from (
22+ { length : 4 } ,
23+ ( ) =>
24+ new Promise ( ( resolve , reject ) => {
25+ const worker = new Worker (
26+ `const { parentPort, workerData } = require('node:worker_threads');
27+ import(workerData.module).then(({ createHourlyBudget }) => {
28+ const budget = createHourlyBudget(() => workerData.cells);
29+ Atomics.add(workerData.gate, 0, 1);
30+ Atomics.wait(workerData.gate, 1, 0);
31+ const granted = [0, 0];
32+ for (let i = 0; i < workerData.attempts; i++) {
33+ const late = workerData.rollAt !== null && i >= workerData.rollAt;
34+ if (budget.reserve(workerData.limit, workerData.now + (late ? workerData.hour : 0), 0)) granted[late ? 1 : 0]++;
35+ }
36+ parentPort.postMessage(granted);
37+ });` ,
38+ {
39+ eval : true ,
40+ workerData : { module : MODULE , cells : shared , gate, attempts, limit, rollAt, now : NOW , hour : HOUR } ,
41+ }
42+ ) ;
43+ worker . once ( 'message' , resolve ) ;
44+ worker . once ( 'error' , reject ) ;
45+ } )
3846 ) ;
47+ while ( Atomics . load ( gate , 0 ) < 4 ) await new Promise ( ( r ) => setTimeout ( r , 1 ) ) ;
48+ Atomics . store ( gate , 1 , 1 ) ;
49+ Atomics . notify ( gate , 1 ) ;
50+ return Promise . all ( runs ) ;
51+ } ;
52+
53+ test ( 'across real worker threads the limit holds exactly: no thread multiplies it' , async ( ) => {
54+ const shared = cells ( ) ;
55+ // 200,000 attempts for a limit of 150,000: headroom for the reserves contention refuses.
56+ const granted = await race ( shared , { attempts : 50_000 , limit : 150_000 } ) ;
3957 assert . equal (
40- granted . reduce ( ( a , b ) => a + b , 0 ) ,
41- limit ,
42- `granted ${ granted . join ( ' + ' ) } ${ workers }
58+ granted . reduce ( ( a , [ now ] ) => a + now , 0 ) ,
59+ 150_000 ,
60+ `granted ${ granted . map ( ( [ n ] ) => n ) . join ( ' + ' ) } 4 threads`
4361 ) ;
44- assert . equal ( shared [ 2 ] , 0 , 'the other lane is untouched' ) ;
62+ assert . equal ( shared [ 1 ] , 0n , 'the other lane is untouched' ) ;
4563} ) ;
4664
47- test ( 'lanes are independent, a release returns a slot, and the next hour starts afresh' , ( ) => {
48- const cell = new Int32Array ( new SharedArrayBuffer ( hourlyBudgetLength ( 2 ) * Int32Array . BYTES_PER_ELEMENT ) ) ;
49- const budget = createHourlyBudget ( ( ) => cell ) ;
65+ test ( 'the hour rolls under contention without handing out more than one hour’s budget per hour' , async ( ) => {
66+ // Every thread crosses into the next hour mid-run, at its own pace. The first hour grants at most its limit (a
67+ // thread still in it once another has rolled the cell is a straggler, and refused); the next hour grants exactly
68+ // its limit: no stale count from the first hour refuses it (the race a separate hour word and count word had).
69+ // Each hour has well over its limit in attempts: a reserve that loses eight races in a row is refused (the safe
70+ // direction), so "in full" needs headroom.
71+ const granted = await race ( cells ( ) , { attempts : 50_000 , limit : 40_000 , rollAt : 20_000 } ) ;
72+ const first = granted . reduce ( ( a , [ n ] ) => a + n , 0 ) ;
73+ const second = granted . reduce ( ( a , [ , n ] ) => a + n , 0 ) ;
74+ assert . ok ( first <= 40_000 , `the first hour granted ${ first } ) ;
75+ assert . equal ( second , 40_000 , 'the next hour, in full' ) ;
76+ } ) ;
77+
78+ test ( 'lanes are independent, a release returns a slot this hour only, and the hour never moves back' , ( ) => {
79+ const shared = cells ( ) ;
80+ const budget = createHourlyBudget ( ( ) => shared ) ;
5081 assert . equal ( budget . reserve ( 1 , NOW , 0 ) , true ) ;
5182 assert . equal ( budget . reserve ( 1 , NOW , 0 ) , false ) ;
5283 assert . equal ( budget . reserve ( 1 , NOW , 1 ) , true , 'lane 1 has its own count' ) ;
53- budget . release ( 0 ) ;
84+ budget . release ( 0 , NOW ) ;
5485 assert . equal ( budget . reserve ( 1 , NOW , 0 ) , true , 'a released slot is spent again' ) ;
55- assert . equal ( budget . reserve ( 1 , NOW + 3_600_000 , 0 ) , true , 'a new hour' ) ;
56- assert . equal ( budget . reserve ( 1 , NOW + 3_600_000 , 1 ) , true , 'every lane rolls' ) ;
57- // A straggler carrying the old hour does not zero what the new hour has spent.
58- assert . equal ( budget . reserve ( 1 , NOW , 0 ) , false ) ;
59- budget . release ( 0 ) ;
60- budget . release ( 0 ) ;
61- assert . equal ( cell [ 1 ] , 0 , 'a release never goes below zero' ) ;
86+ assert . equal ( budget . reserve ( 1 , NOW + HOUR , 0 ) , true , 'a new hour' ) ;
87+ assert . equal ( budget . reserve ( 1 , NOW + HOUR , 1 ) , true , 'every lane rolls on its own first use' ) ;
88+ assert . equal ( budget . reserve ( 1 , NOW , 0 ) , false , 'a straggler carrying the old hour is refused, not rolled back' ) ;
89+ budget . release ( 0 , NOW ) ;
90+ assert . equal ( budget . reserve ( 1 , NOW + HOUR , 0 ) , false , 'a release for an hour already gone returns nothing' ) ;
91+ budget . release ( 0 , NOW + HOUR ) ;
92+ budget . release ( 0 , NOW + HOUR ) ;
93+ assert . equal ( budget . reserve ( 2 , NOW + HOUR , 0 ) , true , 'a release never goes below zero' ) ;
94+ assert . equal ( budget . reserve ( 2 , NOW + HOUR , 0 ) , true ) ;
95+ assert . equal ( budget . reserve ( 2 , NOW + HOUR , 0 ) , false ) ;
6296} ) ;
0 commit comments