12116550809

Committed 02 Dec 2024 09:37AM UTC coverage: 85.0% (-0.6%) from 85.625%

Build # 12116550809

Build Type

push

github

Committed by

Cweili

Commit Message

test: fix test process does not end

Run Details

29 of 47 branches covered (61.7%)

Branch coverage included in aggregate %.

11 of 12 new or added lines in 1 file covered. (91.67%)

3 existing lines in 1 file now uncovered.

107 of 113 relevant lines covered (94.69%)

12.73 hits per line

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

84.07

/src/storage.ts

/* eslint-disable no-restricted-syntax */
import v8 from 'v8'
import { resolve as pathResolve } from 'path'
import { Worker } from 'worker_threads'
import type { Options as StoreOptions } from 'conf'
import { throttle } from 'throttle-debounce'

const SYNC_STORE_SIZE = 64 * 1024 * 1024

type Resolver<T = unknown> = ((value: T | PromiseLike<T>) => void) | ((reason?: Error) => void)

interface Options<T> extends Omit<StoreOptions<T>, 'accessPropertiesByDotNotation' | 'serialize'> {
  saveThrottle?: number
}

export class Store<T extends Record<string, any> = Record<string, any>> {
  private options: Options<T>

  private storeMap?: Map<keyof T, T[keyof T]>

  private storeObj = {} as T

  private version = 0

  private storeObjVersion = -1

  private calls = new Map<number, Resolver[]>()

  private worker?: Worker

  private id = 0

  private terminated = false

  private saveThrottle: throttle<() => void>

  constructor(options: Options<T>) {
    this.options = {
      saveThrottle: 10000,
      ...options
    }
    this.saveThrottle = throttle(this.options.saveThrottle!, () => {
      this.save()
    })
    this.createWorker()
  }

  private restartWorker = (handleError?: boolean) => (err) => {
    if (handleError) console.error(err)
    if (this.terminated) return
    this.createWorker()
  }

  private createWorker() {
    this.worker = (new Worker(
      pathResolve(__dirname, process.env.JEST_WORKER_ID ? '../dist' : '', 'worker.js'),
      { workerData: this.options },
    ))
      .on('error', this.restartWorker(true))
      .on('exit', this.restartWorker())

    this.worker!.on('message', (resBuffer) => {
      const {
        id,
        res,
        didThrow,
      } = v8.deserialize(Buffer.from(resBuffer))
      const [
        resolveCall,
        rejectCall,
      ] = this.calls.get(id) || [];
      (didThrow ? rejectCall : resolveCall)?.(res)
      this.calls.delete(id)
    })
  }

  private exec(method: string, ...args: unknown[]) {
    return new Promise((resolve, reject) => {
      const id = ++this.id
      const buf = v8.serialize({
        id,
        method,
        args,
      })
      const reqBuf = buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength)
      this.calls.set(id, [resolve, reject])
      this.worker!.postMessage(reqBuf, [reqBuf])
    })
  }

  private execSync(method: string, ...args: unknown[]) {
    const sharedBuffer = new SharedArrayBuffer(SYNC_STORE_SIZE)
    const semaphore = new Int32Array(sharedBuffer)
    this.worker!.postMessage({
      method,
      args,
      res: sharedBuffer,
    })
    Atomics.wait(semaphore, 0, 0)
    const resBuf = Buffer.from(sharedBuffer)
    let len = resBuf.readInt32LE()
    let didThrow
    if (len < 0) {
      len = 0 - len
      didThrow = true
    }
    const res = v8.deserialize(
      resBuf.subarray(Int32Array.BYTES_PER_ELEMENT, Int32Array.BYTES_PER_ELEMENT + len),
    )
    if (didThrow) {
      throw res
    }
    return res
  }

  async init() {
    if (this.storeMap) return
    this.storeMap = new Map<keyof T, T[keyof T]>()
    this.storeMap = await this.exec('storeMap') as Map<keyof T, T[keyof T]>
  }

  initSync() {
    if (this.storeMap) return
    this.storeMap = this.execSync('storeMap') as Map<keyof T, T[keyof T]>
  }

  get store() {
    if (this.storeObjVersion !== this.version) {
      this.storeObj = Object.fromEntries(this.storeMap!) as T
      this.storeObjVersion = this.version
    }
    return this.storeObj
  }

  get keys() {
    return Array.from(this.storeMap?.keys() || [])
  }

  get values() {
    return Array.from(this.storeMap?.values() || [])
  }

  /**
    Get an item.

    @param key - The key of the item to get.
    @param defaultValue - The default value if the item does not exist.
    */
  get<Key extends keyof T>(key: Key): T[Key];

  get<Key extends keyof T>(key: Key, defaultValue: Required<T>[Key]): Required<T>[Key];

  // This overload is used for dot-notation access.
  // We exclude `keyof T` as an incorrect type for the default value
  // should not fall through to this overload.
  get<Key extends string, Value = unknown>(key: Exclude<Key, keyof T>, defaultValue?: Value): Value;

  get(key: string, defaultValue?: unknown): unknown {
    const { storeMap: store } = this
    return store!.has(key) ? store!.get(key) : defaultValue
  }

  /**
    Set an item or multiple items at once.

    @param {key|object} - You can use a hashmap of items to set at once.
    @param value - Must be JSON serializable. Trying to set the type `undefined`,
                  `function`, or `symbol` will result in a `TypeError`.
    */
  set<Key extends keyof T>(key: Key, value?: T[Key]): void

  set(key: string, value: unknown): void

  set(object: Partial<T>): void

  set<Key extends keyof T>(key: Partial<T> | Key | string, value?: T[Key] | unknown): void {
    ++this.version

    if (typeof key === 'object') {
      const object = key
      for (const [k, v] of Object.entries(object)) {
        this.storeMap!.set(k, v!)
      }
    } else {
      this.storeMap!.set(key, value as T[keyof T])
    }

    this.saveThrottle()
  }

  save() {
    if (!this.storeMap) return Promise.resolve()
    return this.exec('setMap', this.storeMap)
  }

  /**
    Check if an item exists.

    @param key - The key of the item to check.
    */
  has<Key extends keyof T>(key: Key | string): boolean {
    return this.storeMap!.has(key as string)
  }

  /**
    Delete an item.

    @param key - The key of the item to delete.
    */
  delete<Key extends keyof T>(key: Key): void;

  delete(key: string): void {
    if (this.storeMap!.delete(key)) {
      ++this.version
      this.saveThrottle()
    }
  }

  /**
    Delete all items.

    This resets known items to their default values,
    if defined by the `defaults` or `schema` option.
    */
  clear(): void {
    ++this.version
    this.storeMap!.clear()
    this.saveThrottle()
  }

  /**
   * Destroy the store instance.
   */
  async destroy() {
    this.saveThrottle.cancel()
    await this.save()
    this.terminated = true
    return this.worker!.terminate()
  }
}

1	/* eslint-disable no-restricted-syntax */
2	import v8 from 'v8'	1✔
3	import { resolve as pathResolve } from 'path'	1✔
4	import { Worker } from 'worker_threads'	1✔
5	import type { Options as StoreOptions } from 'conf'
6	import { throttle } from 'throttle-debounce'	1✔
7
8	const SYNC_STORE_SIZE = 64 * 1024 * 1024	1✔
9
10	type Resolver<T = unknown> = ((value: T \| PromiseLike<T>) => void) \| ((reason?: Error) => void)
11
12	interface Options<T> extends Omit<StoreOptions<T>, 'accessPropertiesByDotNotation' \| 'serialize'> {
13	saveThrottle?: number
14	}
15
16	export class Store<T extends Record<string, any> = Record<string, any>> {	1✔
17	private options: Options<T>
18
19	private storeMap?: Map<keyof T, T[keyof T]>
20
21	private storeObj = {} as T	16✔
22
23	private version = 0	16✔
24
25	private storeObjVersion = -1	16✔
26
27	private calls = new Map<number, Resolver[]>()	16✔
28
29	private worker?: Worker
30
31	private id = 0	16✔
32
33	private terminated = false	16✔
34
35	private saveThrottle: throttle<() => void>
36
37	constructor(options: Options<T>) {
38	this.options = {	16✔
39	saveThrottle: 10000,
40	...options
41	}
42	this.saveThrottle = throttle(this.options.saveThrottle!, () => {	16✔
43	this.save()	16✔
44	})
45	this.createWorker()	16✔
46	}
47
48	private restartWorker = (handleError?: boolean) => (err) => {	32✔
49	if (handleError) console.error(err)	16!
50	if (this.terminated) return	16✔
NEW 51	this.createWorker()	×
52	}
53
54	private createWorker() {
55	this.worker = (new Worker(	16✔
56	pathResolve(__dirname, process.env.JEST_WORKER_ID ? '../dist' : '', 'worker.js'),	16!
57	{ workerData: this.options },
58	))
59	.on('error', this.restartWorker(true))
60	.on('exit', this.restartWorker())
61
62	this.worker!.on('message', (resBuffer) => {	16✔
63	const {
64	id,
65	res,
66	didThrow,
67	} = v8.deserialize(Buffer.from(resBuffer))	40✔
68	const [
69	resolveCall,
70	rejectCall,
71	] = this.calls.get(id) \|\| [];	40!
72	(didThrow ? rejectCall : resolveCall)?.(res)	40!
73	this.calls.delete(id)	40✔
74	})
75	}
76
77	private exec(method: string, ...args: unknown[]) {
78	return new Promise((resolve, reject) => {	40✔
79	const id = ++this.id	40✔
80	const buf = v8.serialize({	40✔
81	id,
82	method,
83	args,
84	})
85	const reqBuf = buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength)	40✔
86	this.calls.set(id, [resolve, reject])	40✔
87	this.worker!.postMessage(reqBuf, [reqBuf])	40✔
88	})
89	}
90
91	private execSync(method: string, ...args: unknown[]) {
92	const sharedBuffer = new SharedArrayBuffer(SYNC_STORE_SIZE)	8✔
93	const semaphore = new Int32Array(sharedBuffer)	8✔
94	this.worker!.postMessage({	8✔
95	method,
96	args,
97	res: sharedBuffer,
98	})
99	Atomics.wait(semaphore, 0, 0)	8✔
100	const resBuf = Buffer.from(sharedBuffer)	8✔
101	let len = resBuf.readInt32LE()	8✔
102	let didThrow
103	if (len < 0) {	8!
UNCOV 104	len = 0 - len	×
UNCOV 105	didThrow = true	×
106	}
107	const res = v8.deserialize(	8✔
108	resBuf.subarray(Int32Array.BYTES_PER_ELEMENT, Int32Array.BYTES_PER_ELEMENT + len),
109	)
110	if (didThrow) {	8!
UNCOV 111	throw res	×
112	}
113	return res	8✔
114	}
115
116	async init() {
117	if (this.storeMap) return	8!
118	this.storeMap = new Map<keyof T, T[keyof T]>()	8✔
119	this.storeMap = await this.exec('storeMap') as Map<keyof T, T[keyof T]>	8✔
120	}
121
122	initSync() {
123	if (this.storeMap) return	8!
124	this.storeMap = this.execSync('storeMap') as Map<keyof T, T[keyof T]>	8✔
125	}
126
127	get store() {
128	if (this.storeObjVersion !== this.version) {	18✔
129	this.storeObj = Object.fromEntries(this.storeMap!) as T	18✔
130	this.storeObjVersion = this.version	18✔
131	}
132	return this.storeObj	18✔
133	}
134
135	get keys() {
136	return Array.from(this.storeMap?.keys() \|\| [])	2!
137	}
138
139	get values() {
140	return Array.from(this.storeMap?.values() \|\| [])	2!
141	}
142
143	/**
144	Get an item.
145
146	@param key - The key of the item to get.
147	@param defaultValue - The default value if the item does not exist.
148	*/
149	get<Key extends keyof T>(key: Key): T[Key];
150
151	get<Key extends keyof T>(key: Key, defaultValue: Required<T>[Key]): Required<T>[Key];
152
153	// This overload is used for dot-notation access.
154	// We exclude `keyof T` as an incorrect type for the default value
155	// should not fall through to this overload.
156	get<Key extends string, Value = unknown>(key: Exclude<Key, keyof T>, defaultValue?: Value): Value;
157
158	get(key: string, defaultValue?: unknown): unknown {
159	const { storeMap: store } = this	12✔
160	return store!.has(key) ? store!.get(key) : defaultValue	12✔
161	}
162
163	/**
164	Set an item or multiple items at once.
165
166	@param {key\|object} - You can use a hashmap of items to set at once.
167	@param value - Must be JSON serializable. Trying to set the type `undefined`,
168	`function`, or `symbol` will result in a `TypeError`.
169	*/
170	set<Key extends keyof T>(key: Key, value?: T[Key]): void
171
172	set(key: string, value: unknown): void
173
174	set(object: Partial<T>): void
175
176	set<Key extends keyof T>(key: Partial<T> \| Key \| string, value?: T[Key] \| unknown): void {
177	++this.version	20✔
178
179	if (typeof key === 'object') {	20✔
180	const object = key	2✔
181	for (const [k, v] of Object.entries(object)) {	2✔
182	this.storeMap!.set(k, v!)	4✔
183	}
184	} else {
185	this.storeMap!.set(key, value as T[keyof T])	18✔
186	}
187
188	this.saveThrottle()	20✔
189	}
190
191	save() {
192	if (!this.storeMap) return Promise.resolve()	32!
193	return this.exec('setMap', this.storeMap)	32✔
194	}
195
196	/**
197	Check if an item exists.
198
199	@param key - The key of the item to check.
200	*/
201	has<Key extends keyof T>(key: Key \| string): boolean {
202	return this.storeMap!.has(key as string)	4✔
203	}
204
205	/**
206	Delete an item.
207
208	@param key - The key of the item to delete.
209	*/
210	delete<Key extends keyof T>(key: Key): void;
211
212	delete(key: string): void {
213	if (this.storeMap!.delete(key)) {	4✔
214	++this.version	4✔
215	this.saveThrottle()	4✔
216	}
217	}
218
219	/**
220	Delete all items.
221
222	This resets known items to their default values,
223	if defined by the `defaults` or `schema` option.
224	*/
225	clear(): void {
226	++this.version	2✔
227	this.storeMap!.clear()	2✔
228	this.saveThrottle()	2✔
229	}
230
231	/**
232	* Destroy the store instance.
233	*/
234	async destroy() {
235	this.saveThrottle.cancel()	16✔
236	await this.save()	16✔
237	this.terminated = true	16✔
238	return this.worker!.terminate()	16✔
239	}
240	}

Cweili / store-worker / 12116550809

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