23547724891

Committed 25 Mar 2026 02:55PM UTC coverage: 83.226% (+0.07%) from 83.161%

Build # 23547724891

Build Type

push

github

Committed by

cr15p1

Commit Message

improvement: raise upsert-one p99 bench budget to 20ms

Coverage Stats

880 of 1041 branches covered (84.53%)

Branch coverage included in aggregate %.

3918 of 4724 relevant lines covered (82.94%)

37.49 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.62

/packages/sync/src/utils/cacheWriteQueue.ts

import { type CacheStorage } from '../entity.js';
import { defaultRuntimeQueue, type RuntimeQueue } from '../runtimeQueue/index.js';

interface BuildCacheValue {
  (): string;
}

interface CacheSetOperation {
  type: 'set';
  value: string | BuildCacheValue;
}

interface CacheRemoveOperation {
  type: 'remove';
}

type CacheOperation =
  | CacheSetOperation
  | CacheRemoveOperation;

export interface CacheWriteQueue {
  enqueueSet: (key: string, value: string | BuildCacheValue) => void;
  enqueueRemove: (key: string) => void;
  flush: () => void;
}

export interface CreateCacheWriteQueueInput {
  storage: CacheStorage;
  batchSize?: number;
  runtimeQueue?: RuntimeQueue;
  queueKey?: string;
}

interface ReadOrCreateSharedCacheWriteQueueInput {
  storage: CacheStorage;
}

const DEFAULT_BATCH_SIZE = 50;
let nextCacheWriteQueueId = 0;
const sharedCacheWriteQueueByStorage = new WeakMap<CacheStorage, CacheWriteQueue>();

export const createCacheWriteQueue = ({
  storage,
  batchSize = DEFAULT_BATCH_SIZE,
  runtimeQueue = defaultRuntimeQueue,
  queueKey,
}: CreateCacheWriteQueueInput): CacheWriteQueue => {
  const operationsByKey = new Map<string, CacheOperation>();
  nextCacheWriteQueueId += 1;
  const flushQueueKey = queueKey ?? `cache-write:${nextCacheWriteQueueId}`;

  const runBatch = (): void => {
    const entries = Array.from(operationsByKey.entries());
    const currentBatch = entries.slice(0, batchSize);

    currentBatch.forEach(([key]) => {
      operationsByKey.delete(key);
    });

    currentBatch.forEach(([key, operation]) => {
      try {
        if (operation.type === 'set') {
          const value = typeof operation.value === 'function'
            ? operation.value()
            : operation.value;
          storage.setItem(key, value);
          return;
        }

        storage.removeItem(key);
      } catch {
        return;
      }
    });
  };

  const scheduleFlush = (): void => {
    runtimeQueue.enqueue({
      channel: 'storage',
      key: flushQueueKey,
      run: () => {
        runBatch();
        if (operationsByKey.size > 0) {
          scheduleFlush();
        }
      },
    });
  };

  const enqueueSet = (key: string, value: string | BuildCacheValue): void => {
    operationsByKey.set(key, {
      type: 'set',
      value,
    });
    scheduleFlush();
  };

  const enqueueRemove = (key: string): void => {
    operationsByKey.set(key, {
      type: 'remove',
    });
    scheduleFlush();
  };

  const flushRemaining = (): void => {
    if (operationsByKey.size === 0) {
      return;
    }

    runBatch();
    flushRemaining();
  };

  const flush = (): void => {
    flushRemaining();
  };

  return {
    enqueueSet,
    enqueueRemove,
    flush,
  };
};

export const readOrCreateSharedCacheWriteQueue = ({
  storage,
}: ReadOrCreateSharedCacheWriteQueueInput): CacheWriteQueue => {
  const existingQueue = sharedCacheWriteQueueByStorage.get(storage);
  if (existingQueue) {
    return existingQueue;
  }

  const nextQueue = createCacheWriteQueue({ storage });
  sharedCacheWriteQueueByStorage.set(storage, nextQueue);
  return nextQueue;
};

1	import { type CacheStorage } from '../entity.js';
2	import { defaultRuntimeQueue, type RuntimeQueue } from '../runtimeQueue/index.js';
3
4	interface BuildCacheValue {
5	(): string;
6	}
7
8	interface CacheSetOperation {
9	type: 'set';
10	value: string \| BuildCacheValue;
11	}
12
13	interface CacheRemoveOperation {
14	type: 'remove';
15	}
16
17	type CacheOperation =
18	\| CacheSetOperation
19	\| CacheRemoveOperation;
20
21	export interface CacheWriteQueue {
22	enqueueSet: (key: string, value: string \| BuildCacheValue) => void;
23	enqueueRemove: (key: string) => void;
24	flush: () => void;
25	}
26
27	export interface CreateCacheWriteQueueInput {
28	storage: CacheStorage;
29	batchSize?: number;
30	runtimeQueue?: RuntimeQueue;
31	queueKey?: string;
32	}
33
34	interface ReadOrCreateSharedCacheWriteQueueInput {
35	storage: CacheStorage;
36	}
37
38	const DEFAULT_BATCH_SIZE = 50;	42✔
39	let nextCacheWriteQueueId = 0;	42✔
40	const sharedCacheWriteQueueByStorage = new WeakMap<CacheStorage, CacheWriteQueue>();	42✔
41
42	export const createCacheWriteQueue = ({	42✔
43	storage,
44	batchSize = DEFAULT_BATCH_SIZE,
45	runtimeQueue = defaultRuntimeQueue,
46	queueKey,
47	}: CreateCacheWriteQueueInput): CacheWriteQueue => {
48	const operationsByKey = new Map<string, CacheOperation>();	22✔
49	nextCacheWriteQueueId += 1;	22✔
50	const flushQueueKey = queueKey ?? `cache-write:${nextCacheWriteQueueId}`;	22✔
51
52	const runBatch = (): void => {	22✔
53	const entries = Array.from(operationsByKey.entries());	12✔
54	const currentBatch = entries.slice(0, batchSize);	12✔
55
56	currentBatch.forEach(([key]) => {	12✔
57	operationsByKey.delete(key);	16✔
58	});
59
60	currentBatch.forEach(([key, operation]) => {	12✔
61	try {	16✔
62	if (operation.type === 'set') {	16✔
63	const value = typeof operation.value === 'function'	11✔
64	? operation.value()
65	: operation.value;
66	storage.setItem(key, value);	11✔
67	return;	11✔
68	}
69
70	storage.removeItem(key);	5✔
71	} catch {
72	return;	×
73	}
74	});
75	};
76
77	const scheduleFlush = (): void => {	22✔
78	runtimeQueue.enqueue({	17✔
79	channel: 'storage',
80	key: flushQueueKey,
81	run: () => {
82	runBatch();	12✔
83	if (operationsByKey.size > 0) {	12✔
84	scheduleFlush();	×
85	}
86	},
87	});
88	};
89
90	const enqueueSet = (key: string, value: string \| BuildCacheValue): void => {	22✔
91	operationsByKey.set(key, {	12✔
92	type: 'set',
93	value,
94	});
95	scheduleFlush();	12✔
96	};
97
98	const enqueueRemove = (key: string): void => {	22✔
99	operationsByKey.set(key, {	5✔
100	type: 'remove',
101	});
102	scheduleFlush();	5✔
103	};
104
105	const flushRemaining = (): void => {	22✔
106	if (operationsByKey.size === 0) {	×
107	return;	×
108	}
109
110	runBatch();	×
111	flushRemaining();	×
112	};
113
114	const flush = (): void => {	22✔
115	flushRemaining();	×
116	};
117
118	return {	22✔
119	enqueueSet,
120	enqueueRemove,
121	flush,
122	};
123	};
124
125	export const readOrCreateSharedCacheWriteQueue = ({	42✔
126	storage,
127	}: ReadOrCreateSharedCacheWriteQueueInput): CacheWriteQueue => {
128	const existingQueue = sharedCacheWriteQueueByStorage.get(storage);	45✔
129	if (existingQueue) {	45✔
130	return existingQueue;	23✔
131	}
132
133	const nextQueue = createCacheWriteQueue({ storage });	22✔
134	sharedCacheWriteQueueByStorage.set(storage, nextQueue);	22✔
135	return nextQueue;	22✔
136	};

live-input-vector-output-node / livon-ts / 23547724891

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous