mirror of
https://github.com/nimbold/Firelink.git
synced 2026-09-01 21:48:00 +00:00
fix(tools): harden scheduler and diagnostics
- preserve same-queue scheduler starts across superseded Run Now generations while retaining Stop cancellation - pause saved and tracked scheduler ownership, fence engine cache writes, and normalize log merge/redaction boundaries - add postcondition regressions for scheduler handoff, engine request generations, compact JSON redaction, and live log overlap
This commit is contained in:
+18
-7
@@ -11364,7 +11364,7 @@ pub(crate) fn redact_sensitive_text(line: &str) -> String {
|
|||||||
.expect("valid sensitive header redaction regex")
|
.expect("valid sensitive header redaction regex")
|
||||||
});
|
});
|
||||||
let query = QUERY.get_or_init(|| {
|
let query = QUERY.get_or_init(|| {
|
||||||
regex::Regex::new(r"([A-Za-z][A-Za-z0-9+.-]*://[^\s?]+)\?[^\s]+")
|
regex::Regex::new(r#"([A-Za-z][A-Za-z0-9+.-]*://[^\s?\"'<>},\]]+)\?[^\s\"'<>},\]]+"#)
|
||||||
.expect("valid URL query redaction regex")
|
.expect("valid URL query redaction regex")
|
||||||
});
|
});
|
||||||
let userinfo = USERINFO.get_or_init(|| {
|
let userinfo = USERINFO.get_or_init(|| {
|
||||||
@@ -11375,13 +11375,13 @@ pub(crate) fn redact_sensitive_text(line: &str) -> String {
|
|||||||
regex::Regex::new(r"([A-Za-z][A-Za-z0-9+.-]*://[^\s?#]+)#\S+")
|
regex::Regex::new(r"([A-Za-z][A-Za-z0-9+.-]*://[^\s?#]+)#\S+")
|
||||||
.expect("valid URL fragment redaction regex")
|
.expect("valid URL fragment redaction regex")
|
||||||
});
|
});
|
||||||
let redacted = header.replace_all(line, "$1: [redacted]");
|
let redacted = query.replace_all(line, "$1?[redacted]");
|
||||||
let redacted = quoted_secret.replace_all(&redacted, "$1$2$3$4[redacted]");
|
let redacted = fragment.replace_all(&redacted, "$1#[redacted]");
|
||||||
let redacted = secret.replace_all(&redacted, "$1=[redacted]");
|
|
||||||
let redacted = userinfo.replace_all(&redacted, "$1[redacted]@");
|
let redacted = userinfo.replace_all(&redacted, "$1[redacted]@");
|
||||||
let redacted = query.replace_all(&redacted, "$1?[redacted]");
|
let redacted = header.replace_all(&redacted, "$1: [redacted]");
|
||||||
fragment
|
let redacted = quoted_secret.replace_all(&redacted, "$1$2$3$4[redacted]");
|
||||||
.replace_all(&redacted, "$1#[redacted]")
|
secret
|
||||||
|
.replace_all(&redacted, "$1=[redacted]")
|
||||||
.into_owned()
|
.into_owned()
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -14122,6 +14122,17 @@ mod tests {
|
|||||||
assert!(redacted.contains("[redacted]"));
|
assert!(redacted.contains("[redacted]"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn preserves_compact_json_delimiters_while_redacting_url_queries() {
|
||||||
|
let redacted = redact_log_line(
|
||||||
|
r#"{"url":"https://example.com/file?token=secret","next":1}"#,
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
redacted,
|
||||||
|
r#"{"url":"https://example.com/file?[redacted]","next":1}"#
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn collects_primary_url_and_unique_mirrors_in_order() {
|
fn collects_primary_url_and_unique_mirrors_in_order() {
|
||||||
let uris = collect_download_uris(
|
let uris = collect_download_uris(
|
||||||
|
|||||||
+16
-4
@@ -43,7 +43,12 @@ import { synchronizeDocumentAppearance } from './utils/documentAppearance';
|
|||||||
import { createMainWindowSizePersistence } from './utils/mainWindowState';
|
import { createMainWindowSizePersistence } from './utils/mainWindowState';
|
||||||
import { createSidebarResizeSession } from './utils/sidebarResize';
|
import { createSidebarResizeSession } from './utils/sidebarResize';
|
||||||
import type { MainWindowSize } from './bindings/MainWindowSize';
|
import type { MainWindowSize } from './bindings/MainWindowSize';
|
||||||
import { beginSchedulerControl, isSchedulerControlCurrent } from './utils/schedulerControl';
|
import {
|
||||||
|
beginSchedulerControl,
|
||||||
|
consumeSchedulerHandoffIds,
|
||||||
|
handoffSupersededSchedulerIds,
|
||||||
|
isSchedulerControlCurrent
|
||||||
|
} from './utils/schedulerControl';
|
||||||
import { createSerialTaskQueue } from './utils/serialTaskQueue';
|
import { createSerialTaskQueue } from './utils/serialTaskQueue';
|
||||||
|
|
||||||
const loadSettingsView = () => import('./components/SettingsView');
|
const loadSettingsView = () => import('./components/SettingsView');
|
||||||
@@ -921,9 +926,9 @@ function App() {
|
|||||||
processingScheduleKeys.add(payload.key);
|
processingScheduleKeys.add(payload.key);
|
||||||
try {
|
try {
|
||||||
if (payload.action === 'start') {
|
if (payload.action === 'start') {
|
||||||
const generation = beginSchedulerControl();
|
|
||||||
clearPendingPostActionTimer();
|
clearPendingPostActionTimer();
|
||||||
const scheduledQueueIds = getScheduledQueueIds();
|
const scheduledQueueIds = getScheduledQueueIds();
|
||||||
|
const generation = beginSchedulerControl(scheduledQueueIds);
|
||||||
if (scheduledQueueIds.length === 0) {
|
if (scheduledQueueIds.length === 0) {
|
||||||
state.setSchedulerActiveDownloadIds([]);
|
state.setSchedulerActiveDownloadIds([]);
|
||||||
state.setSchedulerRunning(false);
|
state.setSchedulerRunning(false);
|
||||||
@@ -941,16 +946,23 @@ function App() {
|
|||||||
);
|
);
|
||||||
const acceptedIds = startedResults.flat();
|
const acceptedIds = startedResults.flat();
|
||||||
if (!isSchedulerControlCurrent(generation)) {
|
if (!isSchedulerControlCurrent(generation)) {
|
||||||
|
const handoffIds = handoffSupersededSchedulerIds(
|
||||||
|
acceptedIds,
|
||||||
|
id => useDownloadStore.getState().downloads.find(download => download.id === id)?.queueId || MAIN_QUEUE_ID
|
||||||
|
);
|
||||||
await Promise.allSettled(
|
await Promise.allSettled(
|
||||||
acceptedIds.map(id => useDownloadStore.getState().pauseDownload(id))
|
acceptedIds
|
||||||
|
.filter(id => !handoffIds.has(id))
|
||||||
|
.map(id => useDownloadStore.getState().pauseDownload(id))
|
||||||
);
|
);
|
||||||
await invoke('ack_schedule_trigger', { action: 'start', key: payload.key });
|
await invoke('ack_schedule_trigger', { action: 'start', key: payload.key });
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const scheduledQueueSet = new Set(scheduledQueueIds);
|
const scheduledQueueSet = new Set(scheduledQueueIds);
|
||||||
|
const handoffIds = consumeSchedulerHandoffIds(generation);
|
||||||
const trackedIds = useDownloadStore.getState().downloads
|
const trackedIds = useDownloadStore.getState().downloads
|
||||||
.filter(download =>
|
.filter(download =>
|
||||||
previouslyTrackedIds.has(download.id) &&
|
(previouslyTrackedIds.has(download.id) || handoffIds.has(download.id)) &&
|
||||||
scheduledQueueSet.has(download.queueId || MAIN_QUEUE_ID) &&
|
scheduledQueueSet.has(download.queueId || MAIN_QUEUE_ID) &&
|
||||||
isActiveDownloadStatus(download.status)
|
isActiveDownloadStatus(download.status)
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -12,7 +12,12 @@ import { useToast } from '../contexts/ToastContext';
|
|||||||
import { usePlatformInfo } from '../utils/platform';
|
import { usePlatformInfo } from '../utils/platform';
|
||||||
import { useTranslation } from 'react-i18next';
|
import { useTranslation } from 'react-i18next';
|
||||||
import { formatDateTime } from '../utils/dateTime';
|
import { formatDateTime } from '../utils/dateTime';
|
||||||
import { beginSchedulerControl, isSchedulerControlCurrent } from '../utils/schedulerControl';
|
import {
|
||||||
|
beginSchedulerControl,
|
||||||
|
consumeSchedulerHandoffIds,
|
||||||
|
handoffSupersededSchedulerIds,
|
||||||
|
isSchedulerControlCurrent
|
||||||
|
} from '../utils/schedulerControl';
|
||||||
|
|
||||||
const days = [
|
const days = [
|
||||||
{ value: 0, key: 'su' },
|
{ value: 0, key: 'su' },
|
||||||
@@ -162,22 +167,29 @@ export default function SchedulerView() {
|
|||||||
};
|
};
|
||||||
|
|
||||||
const runNow = async () => {
|
const runNow = async () => {
|
||||||
const generation = beginSchedulerControl();
|
const generation = beginSchedulerControl(effectiveSelectedQueueIds);
|
||||||
const previouslyTrackedIds = new Set(useSettingsStore.getState().schedulerActiveDownloadIds);
|
const previouslyTrackedIds = new Set(useSettingsStore.getState().schedulerActiveDownloadIds);
|
||||||
const results = await Promise.all(
|
const results = await Promise.all(
|
||||||
effectiveSelectedQueueIds.map(queueId => useDownloadStore.getState().startQueue(queueId))
|
effectiveSelectedQueueIds.map(queueId => useDownloadStore.getState().startQueue(queueId))
|
||||||
);
|
);
|
||||||
const acceptedIds = results.flat();
|
const acceptedIds = results.flat();
|
||||||
if (!isSchedulerControlCurrent(generation)) {
|
if (!isSchedulerControlCurrent(generation)) {
|
||||||
|
const handoffIds = handoffSupersededSchedulerIds(
|
||||||
|
acceptedIds,
|
||||||
|
id => useDownloadStore.getState().downloads.find(download => download.id === id)?.queueId || MAIN_QUEUE_ID
|
||||||
|
);
|
||||||
await Promise.allSettled(
|
await Promise.allSettled(
|
||||||
acceptedIds.map(id => useDownloadStore.getState().pauseDownload(id))
|
acceptedIds
|
||||||
|
.filter(id => !handoffIds.has(id))
|
||||||
|
.map(id => useDownloadStore.getState().pauseDownload(id))
|
||||||
);
|
);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const selectedQueueSet = new Set(effectiveSelectedQueueIds);
|
const selectedQueueSet = new Set(effectiveSelectedQueueIds);
|
||||||
|
const handoffIds = consumeSchedulerHandoffIds(generation);
|
||||||
const trackedIds = useDownloadStore.getState().downloads
|
const trackedIds = useDownloadStore.getState().downloads
|
||||||
.filter(download =>
|
.filter(download =>
|
||||||
previouslyTrackedIds.has(download.id) &&
|
(previouslyTrackedIds.has(download.id) || handoffIds.has(download.id)) &&
|
||||||
selectedQueueSet.has(download.queueId || MAIN_QUEUE_ID) &&
|
selectedQueueSet.has(download.queueId || MAIN_QUEUE_ID) &&
|
||||||
isActiveDownloadStatus(download.status)
|
isActiveDownloadStatus(download.status)
|
||||||
)
|
)
|
||||||
@@ -199,11 +211,23 @@ export default function SchedulerView() {
|
|||||||
|
|
||||||
const pauseNow = async () => {
|
const pauseNow = async () => {
|
||||||
const generation = beginSchedulerControl();
|
const generation = beginSchedulerControl();
|
||||||
|
const savedQueueIds = savedSettings.selectedQueueIds
|
||||||
|
.filter(queueId => availableQueueIds.has(queueId));
|
||||||
|
const savedQueueSet = new Set(savedQueueIds);
|
||||||
|
const trackedIdsOutsideSavedQueues = useSettingsStore.getState().schedulerActiveDownloadIds
|
||||||
|
.filter(id => {
|
||||||
|
const queueId = useDownloadStore.getState().downloads.find(download => download.id === id)?.queueId || MAIN_QUEUE_ID;
|
||||||
|
return !savedQueueSet.has(queueId);
|
||||||
|
});
|
||||||
const counts = await Promise.all(
|
const counts = await Promise.all(
|
||||||
effectiveSelectedQueueIds.map(queueId => useDownloadStore.getState().pauseQueue(queueId))
|
savedQueueIds.map(queueId => useDownloadStore.getState().pauseQueue(queueId))
|
||||||
|
);
|
||||||
|
const directPauseResults = await Promise.allSettled(
|
||||||
|
trackedIdsOutsideSavedQueues.map(id => useDownloadStore.getState().pauseDownload(id))
|
||||||
);
|
);
|
||||||
if (!isSchedulerControlCurrent(generation)) return;
|
if (!isSchedulerControlCurrent(generation)) return;
|
||||||
const count = counts.reduce((total, queueCount) => total + queueCount, 0);
|
const count = counts.reduce((total, queueCount) => total + queueCount, 0)
|
||||||
|
+ directPauseResults.filter(result => result.status === 'fulfilled').length;
|
||||||
useSettingsStore.getState().setSchedulerRunning(false);
|
useSettingsStore.getState().setSchedulerRunning(false);
|
||||||
useSettingsStore.getState().setSchedulerActiveDownloadIds([]);
|
useSettingsStore.getState().setSchedulerActiveDownloadIds([]);
|
||||||
addToast({
|
addToast({
|
||||||
|
|||||||
@@ -51,6 +51,7 @@ import {
|
|||||||
} from '../utils/downloads';
|
} from '../utils/downloads';
|
||||||
import { useTranslation } from 'react-i18next';
|
import { useTranslation } from 'react-i18next';
|
||||||
import { localeDirection, resolveAppLocale } from '../i18n';
|
import { localeDirection, resolveAppLocale } from '../i18n';
|
||||||
|
import { createEngineStatusRequestTracker } from '../utils/engineStatusRequests';
|
||||||
|
|
||||||
const settingsTabs: { type: SettingsTab; icon: typeof Download }[] = [
|
const settingsTabs: { type: SettingsTab; icon: typeof Download }[] = [
|
||||||
{ type: 'downloads', icon: Download },
|
{ type: 'downloads', icon: Download },
|
||||||
@@ -222,6 +223,7 @@ const networkSettingsSectionFromStorage = (): NetworkSettingsSection => {
|
|||||||
|
|
||||||
const engineStatusCache = new Map<string, EngineStatusItem>();
|
const engineStatusCache = new Map<string, EngineStatusItem>();
|
||||||
const engineStatusInFlight = new Map<string, Promise<EngineStatusItem>>();
|
const engineStatusInFlight = new Map<string, Promise<EngineStatusItem>>();
|
||||||
|
const engineStatusRequests = createEngineStatusRequestTracker();
|
||||||
|
|
||||||
const upsertEngineStatus = (items: EngineStatusItem[], item: EngineStatusItem) => {
|
const upsertEngineStatus = (items: EngineStatusItem[], item: EngineStatusItem) => {
|
||||||
const next = items.filter(existing => existing.kind !== item.kind);
|
const next = items.filter(existing => existing.kind !== item.kind);
|
||||||
@@ -309,10 +311,13 @@ const runEngineStatusCheck = (check: EngineCheck, force: boolean) => {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (force) engineStatusCache.delete(check.kind);
|
if (force) engineStatusCache.delete(check.kind);
|
||||||
|
const requestId = engineStatusRequests.begin(check.kind);
|
||||||
|
|
||||||
const promise = invoke(check.command)
|
const promise = invoke(check.command)
|
||||||
.then(item => {
|
.then(item => {
|
||||||
if (item.ready) engineStatusCache.set(item.kind, item);
|
if (item.ready && engineStatusRequests.isCurrent(check.kind, requestId)) {
|
||||||
|
engineStatusCache.set(item.kind, item);
|
||||||
|
}
|
||||||
return item;
|
return item;
|
||||||
})
|
})
|
||||||
.catch(error => buildEngineStatusError(check, error))
|
.catch(error => buildEngineStatusError(check, error))
|
||||||
|
|||||||
@@ -0,0 +1,22 @@
|
|||||||
|
import { describe, expect, it } from 'vitest';
|
||||||
|
import { createEngineStatusRequestTracker } from './engineStatusRequests';
|
||||||
|
|
||||||
|
describe('engine status request tracker', () => {
|
||||||
|
it('invalidates an older result for the same engine after a forced recheck', () => {
|
||||||
|
const tracker = createEngineStatusRequestTracker();
|
||||||
|
const first = tracker.begin('aria2');
|
||||||
|
const second = tracker.begin('aria2');
|
||||||
|
|
||||||
|
expect(tracker.isCurrent('aria2', first)).toBe(false);
|
||||||
|
expect(tracker.isCurrent('aria2', second)).toBe(true);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('keeps independent engine checks current independently', () => {
|
||||||
|
const tracker = createEngineStatusRequestTracker();
|
||||||
|
const aria2 = tracker.begin('aria2');
|
||||||
|
const ytdlp = tracker.begin('ytdlp');
|
||||||
|
|
||||||
|
expect(tracker.isCurrent('aria2', aria2)).toBe(true);
|
||||||
|
expect(tracker.isCurrent('ytdlp', ytdlp)).toBe(true);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -0,0 +1,15 @@
|
|||||||
|
export const createEngineStatusRequestTracker = () => {
|
||||||
|
let nextRequestId = 0;
|
||||||
|
const latestRequestByKind = new Map<string, number>();
|
||||||
|
|
||||||
|
return {
|
||||||
|
begin(kind: string): number {
|
||||||
|
const requestId = ++nextRequestId;
|
||||||
|
latestRequestByKind.set(kind, requestId);
|
||||||
|
return requestId;
|
||||||
|
},
|
||||||
|
isCurrent(kind: string, requestId: number): boolean {
|
||||||
|
return latestRequestByKind.get(kind) === requestId;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
};
|
||||||
@@ -61,6 +61,19 @@ describe('log entry streaming', () => {
|
|||||||
).map(item => item.message)).toEqual(['one', 'repeat', 'three', 'repeat', 'four']);
|
).map(item => item.message)).toEqual(['one', 'repeat', 'three', 'repeat', 'four']);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('deduplicates a persisted line and its differently formatted live event', () => {
|
||||||
|
const snapshot = [persistedLogEntry('[2026-07-10][18:00:00][INFO][firelink] repeat')];
|
||||||
|
const live = [liveLogEntry(3, 'repeat', new Date('2026-07-10T14:30:00Z'))];
|
||||||
|
|
||||||
|
expect(mergeLogSnapshotAndLiveEntries(snapshot, live)).toEqual(snapshot);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('preserves compact JSON delimiters while redacting URL queries', () => {
|
||||||
|
const redacted = redactLogText('{"url":"https://example.com/file?token=secret","next":1}');
|
||||||
|
|
||||||
|
expect(redacted).toBe('{"url":"https://example.com/file?[redacted]","next":1}');
|
||||||
|
});
|
||||||
|
|
||||||
it('bounds burst updates to the newest entries', () => {
|
it('bounds burst updates to the newest entries', () => {
|
||||||
expect(appendBoundedLogEntries(
|
expect(appendBoundedLogEntries(
|
||||||
[entry('old')],
|
[entry('old')],
|
||||||
|
|||||||
+17
-10
@@ -37,27 +37,34 @@ export const redactLogText = (message: string, homePath = ''): string => {
|
|||||||
}
|
}
|
||||||
|
|
||||||
redacted = redacted.replace(
|
redacted = redacted.replace(
|
||||||
/(["'])(authorization|proxy-authorization|cookie|set-cookie|password|token|secret|credential|pairing[-_ ]?token|api[-_ ]?key)(["'])(\s*[:=]\s*)["'][^"\r\n,;]*["']/gi,
|
/([A-Za-z][A-Za-z0-9+.-]*:\/\/[^\s?"'<>},\]]+)\?[^\s"'<>},\]]+/g,
|
||||||
'$1$2$3$4[redacted]'
|
|
||||||
);
|
|
||||||
redacted = redacted.replace(
|
|
||||||
/([A-Za-z][A-Za-z0-9+.-]*:\/\/)[^@\s/?#]+@/g,
|
|
||||||
'$1[redacted]@'
|
|
||||||
);
|
|
||||||
redacted = redacted.replace(
|
|
||||||
/([A-Za-z][A-Za-z0-9+.-]*:\/\/[^\s?]+)\?[^\s]+/g,
|
|
||||||
'$1?[redacted]'
|
'$1?[redacted]'
|
||||||
);
|
);
|
||||||
redacted = redacted.replace(
|
redacted = redacted.replace(
|
||||||
/([A-Za-z][A-Za-z0-9+.-]*:\/\/[^\s?#]+)#\S+/g,
|
/([A-Za-z][A-Za-z0-9+.-]*:\/\/[^\s?#]+)#\S+/g,
|
||||||
'$1#[redacted]'
|
'$1#[redacted]'
|
||||||
);
|
);
|
||||||
|
redacted = redacted.replace(
|
||||||
|
/([A-Za-z][A-Za-z0-9+.-]*:\/\/)[^@\s/?#]+@/g,
|
||||||
|
'$1[redacted]@'
|
||||||
|
);
|
||||||
|
redacted = redacted.replace(
|
||||||
|
/(["'])(authorization|proxy-authorization|cookie|set-cookie|password|token|secret|credential|pairing[-_ ]?token|api[-_ ]?key)(["'])(\s*[:=]\s*)["'][^"\r\n,;]*["']/gi,
|
||||||
|
'$1$2$3$4[redacted]'
|
||||||
|
);
|
||||||
return redacted.replace(
|
return redacted.replace(
|
||||||
/(authorization|proxy-authorization|cookie|set-cookie|password|token|secret|credential|pairing[-_ ]?token|api[-_ ]?key)(\s*)([:=])(\s*)([^\r\n,;]+)/gi,
|
/(authorization|proxy-authorization|cookie|set-cookie|password|token|secret|credential|pairing[-_ ]?token|api[-_ ]?key)(\s*)([:=])(\s*)([^\r\n,;]+)/gi,
|
||||||
'$1$2$3$4[redacted]'
|
'$1$2$3$4[redacted]'
|
||||||
);
|
);
|
||||||
};
|
};
|
||||||
|
|
||||||
|
const mergeKey = (entry: LogEntry): string => {
|
||||||
|
const message = entry.message
|
||||||
|
.replace(/^\[\d{4}-\d{2}-\d{2}\]\[\d{2}:\d{2}:\d{2}\]\[[A-Z]+\](?:\[[^\]]+\])?\s*/, '')
|
||||||
|
.replace(/^\[\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\]\s*\[[A-Z]+\]\s*/, '');
|
||||||
|
return `${entry.level}:${message}`;
|
||||||
|
};
|
||||||
|
|
||||||
export const liveLogEntry = (
|
export const liveLogEntry = (
|
||||||
numericLevel: number,
|
numericLevel: number,
|
||||||
message: string,
|
message: string,
|
||||||
@@ -113,7 +120,7 @@ export const mergeLogSnapshotAndLiveEntries = (
|
|||||||
const snapshotStart = snapshot.length - candidate;
|
const snapshotStart = snapshot.length - candidate;
|
||||||
let matches = true;
|
let matches = true;
|
||||||
for (let index = 0; index < candidate; index += 1) {
|
for (let index = 0; index < candidate; index += 1) {
|
||||||
if (snapshot[snapshotStart + index].message !== liveEntries[index].message) {
|
if (mergeKey(snapshot[snapshotStart + index]) !== mergeKey(liveEntries[index])) {
|
||||||
matches = false;
|
matches = false;
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,5 +1,10 @@
|
|||||||
import { describe, expect, it } from 'vitest';
|
import { describe, expect, it } from 'vitest';
|
||||||
import { beginSchedulerControl, isSchedulerControlCurrent } from './schedulerControl';
|
import {
|
||||||
|
beginSchedulerControl,
|
||||||
|
consumeSchedulerHandoffIds,
|
||||||
|
handoffSupersededSchedulerIds,
|
||||||
|
isSchedulerControlCurrent
|
||||||
|
} from './schedulerControl';
|
||||||
|
|
||||||
describe('scheduler control generation', () => {
|
describe('scheduler control generation', () => {
|
||||||
it('invalidates an older asynchronous scheduler operation', () => {
|
it('invalidates an older asynchronous scheduler operation', () => {
|
||||||
@@ -10,4 +15,23 @@ describe('scheduler control generation', () => {
|
|||||||
expect(isSchedulerControlCurrent(first)).toBe(false);
|
expect(isSchedulerControlCurrent(first)).toBe(false);
|
||||||
expect(isSchedulerControlCurrent(second)).toBe(true);
|
expect(isSchedulerControlCurrent(second)).toBe(true);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('hands superseded starts to a newer run for the same queue', () => {
|
||||||
|
const first = beginSchedulerControl(['queue-a']);
|
||||||
|
const second = beginSchedulerControl(['queue-a']);
|
||||||
|
|
||||||
|
expect(handoffSupersededSchedulerIds(['download-a', 'download-b'], id => (
|
||||||
|
id === 'download-a' ? 'queue-a' : 'queue-b'
|
||||||
|
))).toEqual(new Set(['download-a']));
|
||||||
|
expect(consumeSchedulerHandoffIds(second)).toEqual(new Set(['download-a']));
|
||||||
|
expect(consumeSchedulerHandoffIds(first)).toEqual(new Set());
|
||||||
|
});
|
||||||
|
|
||||||
|
it('does not hand work to a superseding pause control', () => {
|
||||||
|
beginSchedulerControl(['queue-a']);
|
||||||
|
const pause = beginSchedulerControl();
|
||||||
|
|
||||||
|
expect(handoffSupersededSchedulerIds(['download-a'], () => 'queue-a')).toEqual(new Set());
|
||||||
|
expect(consumeSchedulerHandoffIds(pause)).toEqual(new Set());
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -1,14 +1,46 @@
|
|||||||
let schedulerControlGeneration = 0;
|
let schedulerControlGeneration = 0;
|
||||||
|
let latestRunQueueIds: ReadonlySet<string> | null = null;
|
||||||
|
const schedulerHandoffs = new Map<number, Set<string>>();
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Start a new scheduler control lifecycle. A later manual pause or scheduler
|
* Start a new scheduler control lifecycle. A later manual pause or scheduler
|
||||||
* event invalidates earlier asynchronous queue operations before they can
|
* event invalidates earlier asynchronous queue operations before they can
|
||||||
* publish stale running state.
|
* publish stale running state.
|
||||||
*/
|
*/
|
||||||
export const beginSchedulerControl = (): number => {
|
export const beginSchedulerControl = (runQueueIds?: readonly string[]): number => {
|
||||||
schedulerControlGeneration += 1;
|
schedulerControlGeneration += 1;
|
||||||
|
latestRunQueueIds = runQueueIds ? new Set(runQueueIds) : null;
|
||||||
|
schedulerHandoffs.clear();
|
||||||
|
if (runQueueIds) schedulerHandoffs.set(schedulerControlGeneration, new Set());
|
||||||
return schedulerControlGeneration;
|
return schedulerControlGeneration;
|
||||||
};
|
};
|
||||||
|
|
||||||
export const isSchedulerControlCurrent = (generation: number): boolean =>
|
export const isSchedulerControlCurrent = (generation: number): boolean =>
|
||||||
schedulerControlGeneration === generation;
|
schedulerControlGeneration === generation;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A superseded start may have admitted work before a newer start reached the
|
||||||
|
* same queue. Hand the IDs to that newer start instead of pausing its work.
|
||||||
|
* A stop/manual control has no run intent and therefore receives no handoff.
|
||||||
|
*/
|
||||||
|
export const handoffSupersededSchedulerIds = (
|
||||||
|
ids: readonly string[],
|
||||||
|
queueIdForId: (id: string) => string | undefined,
|
||||||
|
): ReadonlySet<string> => {
|
||||||
|
if (!latestRunQueueIds || schedulerControlGeneration === 0) return new Set();
|
||||||
|
const handoff = schedulerHandoffs.get(schedulerControlGeneration);
|
||||||
|
if (!handoff) return new Set();
|
||||||
|
|
||||||
|
for (const id of ids) {
|
||||||
|
const queueId = queueIdForId(id);
|
||||||
|
if (queueId && latestRunQueueIds.has(queueId)) handoff.add(id);
|
||||||
|
}
|
||||||
|
return new Set(handoff);
|
||||||
|
};
|
||||||
|
|
||||||
|
export const consumeSchedulerHandoffIds = (generation: number): ReadonlySet<string> => {
|
||||||
|
if (!isSchedulerControlCurrent(generation)) return new Set();
|
||||||
|
const handoff = schedulerHandoffs.get(generation) ?? new Set<string>();
|
||||||
|
schedulerHandoffs.delete(generation);
|
||||||
|
return new Set(handoff);
|
||||||
|
};
|
||||||
|
|||||||
Reference in New Issue
Block a user