command-queue.js
2.58 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
/**
* CommandQueue - 多 Lane 命令队列
*
* 参考 openclaw/src/process/command-queue.ts
*
* 核心概念：
* Lane = 逻辑隔离的命令通道 (如 "main", "subagent", "prewarm")
* Generation = 每次 clearAll 后递增，旧任务被 CommandLaneClearedError 拒绝
*
* 每个 lane 内的任务严格串行执行。
* 不同 lane 之间可以并行。
* clearAll() 会使尚在队列中的任务全部失败，防止 "脏" 回调污染新状态。
*/
class CommandLaneClearedError extends Error {
/**
* @param {string} [lane]
*/
constructor(lane) {
super(lane ? `Command lane "${lane}" cleared` : 'Command lane cleared');
this.name = 'CommandLaneClearedError';
this.lane = lane;
}
}
class CommandQueue {
constructor() {
/** @type {Map<string, Promise<void>>} lane → tail promise */
this._lanes = new Map();
this._generation = 0;
}
/**
* 向指定 lane 入队一个任务
*
* 任务会在该 lane 的前一个任务完成后执行。
* 如果在排队期间调用了 clearAll()，任务会被 CommandLaneClearedError 拒绝。
*
* @template T
* @param {string} lane - lane 名称
* @param {() => Promise<T>} task - 异步任务
* @returns {Promise<T>}
*/
enqueue(lane, task) {
const gen = this._generation;
const prior = this._lanes.get(lane) ?? Promise.resolve();
const wrapped = prior
.catch(() => undefined)
.then(() => {
// generation 变了 → 队列已被清空，拒绝执行
if (this._generation !== gen) {
throw new CommandLaneClearedError(lane);
}
return task();
});
// 追踪 lane 的尾部 promise（吞掉错误防止 unhandled rejection）
const tracked = wrapped.catch(() => undefined).finally(() => {
// 如果这是该 lane 最后一个任务，清理引用
if (this._lanes.get(lane) === tracked) {
this._lanes.delete(lane);
}
});
this._lanes.set(lane, tracked);
return wrapped;
}
/**
* 清除所有 lane，递增 generation
*
* 已排队但未执行的任务会收到 CommandLaneClearedError。
* 正在执行的任务不受影响（但其后续排队任务会被拒绝）。
*/
clearAll() {
this._generation++;
this._lanes.clear();
}
/**
* 清除特定 lane
* @param {string} lane
*/
clearLane(lane) {
this._lanes.delete(lane);
}
/** 当前 generation */
get generation() {
return this._generation;
}
/** lane 数量 */
get size() {
return this._lanes.size;
}
}
module.exports = { CommandQueue, CommandLaneClearedError };