24607504909

Committed 18 Apr 2026 03:09PM UTC coverage: 91.218% (+0.09%) from 91.133%

Build # 24607504909

Build Type

push

github

Committed by

VaclavObornik

Commit Message

fix: address Copilot review round 11

1. waitUntilReactiveTasksIdle (whitelist mode) now always checks the
   planner buffer. Skipping it entirely let the helper return idle
   before an in-flight change event had a chance to be turned into
   the task record the DB check looks for. Only the worker count
   check remains global-scope-skipped so unrelated tests' worker
   pools cannot block a scoped wait.

2. docs/reactive-tasks/testing.md now notes the narrow race where
   triggering a write and immediately calling the scoped wait can
   return early if stabilityDurationMs is too short, with a
   recommended workaround (raise stabilityDurationMs or pre-poll
   for the task record).

3. cronTasksConcurrency 'no polls after stop' assertion tightened:
   stopCronTasks is fire-and-forget, so a worker already inside
   findOneAndUpdate at the time could still tick the spy. Capture
   the poll count after a 200ms grace period and assert it does
   not grow over a full 5s back-off window, instead of requiring
   spy.called === false outright.

Coverage Stats

1492 of 1748 branches covered (85.35%)

Branch coverage included in aggregate %.

3 of 3 new or added lines in 1 file covered. (100.0%)

54 existing lines in 9 files now uncovered.

2320 of 2431 relevant lines covered (95.43%)

373.23 hits per line

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

92.19

/src/createContinuousLock.ts

import { Collection, Filter, ObjectId, UpdateFilter } from 'mongodb';
import { onError } from './OnError';

type StopContinuousLock = () => Promise<void>;

export interface CreateContinuousLockOptions {
    /**
     * Initial value of the lock property written by whoever acquired the lock.
     * When provided, every renewal becomes a compare-and-swap: the update only
     * succeeds if the lock still carries the previously-written value. If another
     * actor has taken over the lock in the meantime, `onLockLost` is invoked and
     * no further renewals are attempted. This prevents a slow renewal from
     * accidentally extending a lock that has already been stolen.
     */
    expectedInitialValue?: unknown;
    /**
     * Invoked once when CAS detects we no longer own the lock. Only fires when
     * `expectedInitialValue` is set. Renewals stop after this callback.
     */
    onLockLost?: () => void;
}

export function createContinuousLock<DocumentType extends { _id: string | ObjectId }>(
    collection: Collection<DocumentType>,
    documentId: DocumentType['_id'],
    lockProperty: keyof DocumentType,
    lockTime: number,
    options?: CreateContinuousLockOptions,
): StopContinuousLock {
    let taskInProgress = true;
    let prolongLockTimeoutId: ReturnType<typeof setTimeout> | null = null;
    let lastProlongPromise: Promise<unknown> = Promise.resolve(); // all errors have to be suppressed

    const casEnabled = options?.expectedInitialValue !== undefined;
    let expectedValue: unknown = options?.expectedInitialValue;
    let lockLostNotified = false;

    function scheduleLockProlong() {
        prolongLockTimeoutId = setTimeout(() => {
            prolongLockTimeoutId = null;
            lastProlongPromise = (async () => {
                try {
                    if (!taskInProgress) return;
                    const newValue = new Date(Date.now() + lockTime);
                    const filter: Filter<DocumentType> = (
                        casEnabled ? { _id: documentId, [lockProperty]: expectedValue as never } : { _id: documentId }
                    ) as Filter<DocumentType>;

                    const result = await collection.updateOne(filter, {
                        $set: { [lockProperty]: newValue },
                    } as UpdateFilter<DocumentType>);

                    if (casEnabled && result.matchedCount === 0) {
                        taskInProgress = false;
                        if (!lockLostNotified) {
                            lockLostNotified = true;
                            try {
                                options?.onLockLost?.();
                            } catch (err) {
                                onError(err as Error);
                            }
                        }
                        return;
                    }
                    expectedValue = newValue;
                } catch (err) {
                    onError(err as Error);
                } finally {
                    if (taskInProgress) {
                        scheduleLockProlong();
                    }
                }
            })();
        }, lockTime / 5);
    }

    scheduleLockProlong();

    /** Should never throw! */
    return async () => {
        taskInProgress = false; // prevent next scheduling
        if (prolongLockTimeoutId) {
            clearTimeout(prolongLockTimeoutId);
        }
        await lastProlongPromise;
    };
}

1	import { Collection, Filter, ObjectId, UpdateFilter } from 'mongodb';
2	import { onError } from './OnError';	265✔
3
4	type StopContinuousLock = () => Promise<void>;
5
6	export interface CreateContinuousLockOptions {
7	/**
8	* Initial value of the lock property written by whoever acquired the lock.
9	* When provided, every renewal becomes a compare-and-swap: the update only
10	* succeeds if the lock still carries the previously-written value. If another
11	* actor has taken over the lock in the meantime, `onLockLost` is invoked and
12	* no further renewals are attempted. This prevents a slow renewal from
13	* accidentally extending a lock that has already been stolen.
14	*/
15	expectedInitialValue?: unknown;
16	/**
17	* Invoked once when CAS detects we no longer own the lock. Only fires when
18	* `expectedInitialValue` is set. Renewals stop after this callback.
19	*/
20	onLockLost?: () => void;
21	}
22
23	export function createContinuousLock<DocumentType extends { _id: string \| ObjectId }>(	265✔
24	collection: Collection<DocumentType>,
25	documentId: DocumentType['_id'],
26	lockProperty: keyof DocumentType,
27	lockTime: number,
28	options?: CreateContinuousLockOptions,
29	): StopContinuousLock {
30	let taskInProgress = true;	951✔
31	let prolongLockTimeoutId: ReturnType<typeof setTimeout> \| null = null;	951✔
32	let lastProlongPromise: Promise<unknown> = Promise.resolve(); // all errors have to be suppressed	951✔
33
34	const casEnabled = options?.expectedInitialValue !== undefined;	951✔
35	let expectedValue: unknown = options?.expectedInitialValue;	951✔
36	let lockLostNotified = false;	951✔
37
38	function scheduleLockProlong() {
39	prolongLockTimeoutId = setTimeout(() => {	995✔
40	prolongLockTimeoutId = null;	48✔
41	lastProlongPromise = (async () => {	48✔
42	try {	48✔
43	if (!taskInProgress) return;	48!
44	const newValue = new Date(Date.now() + lockTime);	48✔
45	const filter: Filter<DocumentType> = (
46	casEnabled ? { _id: documentId, [lockProperty]: expectedValue as never } : { _id: documentId }	48✔
47	) as Filter<DocumentType>;
48
49	const result = await collection.updateOne(filter, {	48✔
50	$set: { [lockProperty]: newValue },
51	} as UpdateFilter<DocumentType>);
52
53	if (casEnabled && result.matchedCount === 0) {	47✔
54	taskInProgress = false;	3✔
55	if (!lockLostNotified) {	3!
56	lockLostNotified = true;	3✔
57	try {	3✔
58	options?.onLockLost?.();	3!
59	} catch (err) {
UNCOV 60	onError(err as Error);	×
61	}
62	}
63	return;	3✔
64	}
65	expectedValue = newValue;	44✔
66	} catch (err) {
67	onError(err as Error);	1✔
68	} finally {
69	if (taskInProgress) {	48✔
70	scheduleLockProlong();	44✔
71	}
72	}
73	})();
74	}, lockTime / 5);
75	}
76
77	scheduleLockProlong();	951✔
78
79	/** Should never throw! */
80	return async () => {	951✔
81	taskInProgress = false; // prevent next scheduling	958✔
82	if (prolongLockTimeoutId) {	958✔
83	clearTimeout(prolongLockTimeoutId);	954✔
84	}
85	await lastProlongPromise;	958✔
86	};
87	}

VaclavObornik / mongodash / 24607504909

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