mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-10 00:11:19 +00:00
refactor: consolidate exact keyed async queues (#99691)
* refactor: consolidate keyed async queues * chore: refresh merge-ref CI * refactor: encapsulate keyed queue ownership
This commit is contained in:
Dallin Romney
GitHub
parent
7608d38597
commit
c2fc7aa28a
30 changed files with 136 additions and 456 deletions
|
|
@ -12,6 +12,7 @@ import {
|
|||
type AuthProfileStore,
|
||||
} from "openclaw/plugin-sdk/agent-runtime";
|
||||
import { type FileLockOptions, withFileLock } from "openclaw/plugin-sdk/file-lock";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import {
|
||||
CODEX_PLUGINS_MARKETPLACE_NAME,
|
||||
normalizeCodexServiceTier,
|
||||
|
|
@ -40,7 +41,7 @@ const CODEX_APP_SERVER_BINDING_LOCK_OPTIONS: FileLockOptions = {
|
|||
},
|
||||
stale: CODEX_APP_SERVER_BINDING_GUARDED_REQUEST_TIMEOUT_MS * 2,
|
||||
};
|
||||
const bindingMutationQueues = new Map<string, Promise<void>>();
|
||||
const bindingMutationQueue = new KeyedAsyncQueue();
|
||||
const bindingMutationContext = new AsyncLocalStorage<Set<string>>();
|
||||
|
||||
type ProviderAuthAliasLookupParams = Parameters<typeof resolveProviderIdForAuth>[1];
|
||||
|
|
@ -118,30 +119,13 @@ export async function withCodexAppServerBindingLock<T>(
|
|||
// The SDK file lock is process-reentrant, so pair it with a local queue.
|
||||
// Nested writes from the same guarded mutation can proceed, but unrelated
|
||||
// same-process tasks cannot slip between compare/clear/start.
|
||||
const previous = bindingMutationQueues.get(bindingPath) ?? Promise.resolve();
|
||||
let releaseCurrent!: () => void;
|
||||
const current = new Promise<void>((resolve) => {
|
||||
releaseCurrent = resolve;
|
||||
});
|
||||
const queued = previous.then(
|
||||
() => current,
|
||||
() => current,
|
||||
);
|
||||
bindingMutationQueues.set(bindingPath, queued);
|
||||
await previous.catch(() => undefined);
|
||||
|
||||
const nestedOwnedBindings = new Set(ownedBindings);
|
||||
nestedOwnedBindings.add(bindingPath);
|
||||
try {
|
||||
return await bindingMutationContext.run(nestedOwnedBindings, () =>
|
||||
return await bindingMutationQueue.enqueue(bindingPath, () =>
|
||||
bindingMutationContext.run(nestedOwnedBindings, () =>
|
||||
withFileLock(bindingPath, CODEX_APP_SERVER_BINDING_LOCK_OPTIONS, run),
|
||||
);
|
||||
} finally {
|
||||
releaseCurrent();
|
||||
if (bindingMutationQueues.get(bindingPath) === queued) {
|
||||
bindingMutationQueues.delete(bindingPath);
|
||||
}
|
||||
}
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
/** Reads and normalizes a Codex app-server binding sidecar, returning undefined on stale data. */
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import {
|
|||
} from "openclaw/plugin-sdk/agent-harness-runtime";
|
||||
import { resolveSessionAgentIds } from "openclaw/plugin-sdk/agent-runtime";
|
||||
import { loadExecApprovals } from "openclaw/plugin-sdk/exec-approvals-runtime";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import type {
|
||||
PluginConversationBindingResolvedEvent,
|
||||
PluginHookInboundClaimContext,
|
||||
|
|
@ -120,7 +121,7 @@ type CodexConversationConfig = Parameters<
|
|||
type ResolvedCodexConversationConfig = NonNullable<CodexConversationConfig>;
|
||||
|
||||
type CodexConversationGlobalState = {
|
||||
queues: Map<string, Promise<void>>;
|
||||
queue: KeyedAsyncQueue;
|
||||
};
|
||||
|
||||
async function resolveConversationAppServerRuntime(params: {
|
||||
|
|
@ -169,7 +170,7 @@ function getGlobalState(): CodexConversationGlobalState {
|
|||
const globalState = globalThis as typeof globalThis & {
|
||||
[CODEX_CONVERSATION_GLOBAL_STATE]?: CodexConversationGlobalState;
|
||||
};
|
||||
globalState[CODEX_CONVERSATION_GLOBAL_STATE] ??= { queues: new Map() };
|
||||
globalState[CODEX_CONVERSATION_GLOBAL_STATE] ??= { queue: new KeyedAsyncQueue() };
|
||||
return globalState[CODEX_CONVERSATION_GLOBAL_STATE];
|
||||
}
|
||||
|
||||
|
|
@ -974,22 +975,7 @@ function isCodexThreadNotFoundError(error: unknown): boolean {
|
|||
}
|
||||
|
||||
function enqueueBoundTurn<T>(key: string, run: () => Promise<T>): Promise<T> {
|
||||
const state = getGlobalState();
|
||||
const previous = state.queues.get(key) ?? Promise.resolve();
|
||||
const next = previous.then(run, run);
|
||||
const queued = next.then(
|
||||
() => undefined,
|
||||
() => undefined,
|
||||
);
|
||||
state.queues.set(key, queued);
|
||||
void next
|
||||
.finally(() => {
|
||||
if (state.queues.get(key) === queued) {
|
||||
state.queues.delete(key);
|
||||
}
|
||||
})
|
||||
.catch(() => undefined);
|
||||
return next;
|
||||
return getGlobalState().queue.enqueue(key, run);
|
||||
}
|
||||
|
||||
function resolveThreadRequestModelProvider(params: {
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
import { createHash } from "node:crypto";
|
||||
import path from "node:path";
|
||||
import { ApplicationCommandType, type APIApplicationCommand } from "discord-api-types/v10";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import { privateFileStore } from "openclaw/plugin-sdk/security-runtime";
|
||||
import {
|
||||
createApplicationCommand,
|
||||
|
|
@ -33,25 +34,10 @@ type SerializedCommand = ReturnType<BaseCommand["serialize"]>;
|
|||
* file lock. Discord deployers only run inside the gateway process, so an
|
||||
* in-process mutex is sufficient for the documented concurrency surface.
|
||||
*/
|
||||
const cachePersistLocks = new Map<string, Promise<void>>();
|
||||
const cachePersistLocks = new KeyedAsyncQueue();
|
||||
|
||||
async function withCachePersistLock<T>(storePath: string, fn: () => Promise<T>): Promise<T> {
|
||||
const previous = cachePersistLocks.get(storePath) ?? Promise.resolve();
|
||||
let release: () => void = () => {};
|
||||
const next = new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
const chained = previous.then(() => next);
|
||||
cachePersistLocks.set(storePath, chained);
|
||||
try {
|
||||
await previous;
|
||||
return await fn();
|
||||
} finally {
|
||||
release();
|
||||
if (cachePersistLocks.get(storePath) === chained) {
|
||||
cachePersistLocks.delete(storePath);
|
||||
}
|
||||
}
|
||||
return await cachePersistLocks.enqueue(storePath, fn);
|
||||
}
|
||||
|
||||
export class DiscordCommandDeployer {
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
// Imessage plugin module implements catchup behavior.
|
||||
import { createHash } from "node:crypto";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import type { PluginStateSyncKeyedStore } from "openclaw/plugin-sdk/plugin-state-runtime";
|
||||
import { getIMessageRuntime } from "../runtime.js";
|
||||
|
||||
|
|
@ -29,7 +30,7 @@ const MAX_FAILURE_RETRY_MAP_JSON_BYTES = 48_000;
|
|||
const textEncoder = new TextEncoder();
|
||||
export const IMESSAGE_CATCHUP_CURSOR_NAMESPACE = "imessage.catchup-cursors";
|
||||
export const IMESSAGE_CATCHUP_CURSOR_MAX_ENTRIES = 256;
|
||||
const cursorWriteQueues = new Map<string, Promise<unknown>>();
|
||||
const cursorWriteQueue = new KeyedAsyncQueue();
|
||||
|
||||
export type IMessageCatchupConfig = {
|
||||
enabled?: boolean;
|
||||
|
|
@ -119,17 +120,7 @@ function updateCatchupCursorStore(
|
|||
|
||||
function enqueueCursorWrite<T>(accountId: string, fn: () => Promise<T>): Promise<T> {
|
||||
const key = resolveIMessageCatchupCursorKey(accountId);
|
||||
const prev = cursorWriteQueues.get(key) ?? Promise.resolve();
|
||||
const next = prev.then(fn, fn);
|
||||
cursorWriteQueues.set(key, next);
|
||||
next
|
||||
.finally(() => {
|
||||
if (cursorWriteQueues.get(key) === next) {
|
||||
cursorWriteQueues.delete(key);
|
||||
}
|
||||
})
|
||||
.catch(() => {});
|
||||
return next;
|
||||
return cursorWriteQueue.enqueue(key, fn);
|
||||
}
|
||||
|
||||
function sanitizeFailureRetriesInput(raw: unknown): Record<string, number> {
|
||||
|
|
|
|||
|
|
@ -27,6 +27,7 @@ import {
|
|||
resolveChannelContextVisibilityMode,
|
||||
} from "openclaw/plugin-sdk/context-visibility-runtime";
|
||||
import { isDangerousNameMatchingEnabled } from "openclaw/plugin-sdk/dangerous-name-runtime";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime";
|
||||
import {
|
||||
isFutureDateTimestampMs,
|
||||
|
|
@ -529,7 +530,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
|
|||
logVerboseMessage,
|
||||
});
|
||||
const roomHistoryTracker = createRoomHistoryTracker();
|
||||
const roomIngressTails = new Map<string, Promise<void>>();
|
||||
const roomIngressQueue = new KeyedAsyncQueue();
|
||||
const sharedDmContextNoticeRooms = new Set<string>();
|
||||
|
||||
const readStoreAllowFrom = async (): Promise<string[]> => {
|
||||
|
|
@ -576,22 +577,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam
|
|||
};
|
||||
|
||||
const runRoomIngress = async <T>(roomId: string, task: () => Promise<T>): Promise<T> => {
|
||||
const previous = roomIngressTails.get(roomId) ?? Promise.resolve();
|
||||
let releaseCurrent!: () => void;
|
||||
const current = new Promise<void>((resolve) => {
|
||||
releaseCurrent = resolve;
|
||||
});
|
||||
const chain = previous.catch(() => {}).then(() => current);
|
||||
roomIngressTails.set(roomId, chain);
|
||||
await previous.catch(() => {});
|
||||
try {
|
||||
return await task();
|
||||
} finally {
|
||||
releaseCurrent();
|
||||
if (roomIngressTails.get(roomId) === chain) {
|
||||
roomIngressTails.delete(roomId);
|
||||
}
|
||||
}
|
||||
return await roomIngressQueue.enqueue(roomId, task);
|
||||
};
|
||||
|
||||
return async (roomId: string, event: MatrixRawEvent) => {
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import type { MemorySearchResult } from "openclaw/plugin-sdk/memory-core-host-runtime-files";
|
||||
import {
|
||||
DEFAULT_MEMORY_DEEP_DREAMING_MAX_PROMOTED_SNIPPET_TOKENS,
|
||||
|
|
@ -88,7 +89,7 @@ const GENERIC_DAY_HEADING_RE =
|
|||
/^(?:(?:mon|monday|tue|tues|tuesday|wed|wednesday|thu|thur|thurs|thursday|fri|friday|sat|saturday|sun|sunday)(?:,\s+)?)?(?:(?:jan|january|feb|february|mar|march|apr|april|may|jun|june|jul|july|aug|august|sep|sept|september|oct|october|nov|november|dec|december)\s+\d{1,2}(?:st|nd|rd|th)?(?:,\s*\d{4})?|\d{1,2}[/-]\d{1,2}(?:[/-]\d{2,4})?|\d{4}[/-]\d{2}[/-]\d{2})$/i;
|
||||
const PROMOTION_LIST_MARKER_RE = /^(?:\d+\.\s+|[-*+]\s+)/;
|
||||
const MANAGED_DREAMING_HEADINGS = new Set(["light sleep", "rem sleep"]);
|
||||
const inProcessShortTermLocks = new Map<string, Promise<void>>();
|
||||
const inProcessShortTermLocks = new KeyedAsyncQueue();
|
||||
|
||||
type PromotionWeights = {
|
||||
frequency: number;
|
||||
|
|
@ -849,23 +850,7 @@ function isProcessLikelyAlive(pid: number): boolean {
|
|||
}
|
||||
|
||||
async function withInProcessShortTermLock<T>(lockPath: string, task: () => Promise<T>): Promise<T> {
|
||||
const previous = inProcessShortTermLocks.get(lockPath) ?? Promise.resolve();
|
||||
let releaseCurrent!: () => void;
|
||||
const current = new Promise<void>((resolve) => {
|
||||
releaseCurrent = resolve;
|
||||
});
|
||||
const queued = previous.catch(() => undefined).then(() => current);
|
||||
inProcessShortTermLocks.set(lockPath, queued);
|
||||
|
||||
await previous.catch(() => undefined);
|
||||
try {
|
||||
return await task();
|
||||
} finally {
|
||||
releaseCurrent();
|
||||
if (inProcessShortTermLocks.get(lockPath) === queued) {
|
||||
inProcessShortTermLocks.delete(lockPath);
|
||||
}
|
||||
}
|
||||
return await inProcessShortTermLocks.enqueue(lockPath, task);
|
||||
}
|
||||
|
||||
async function withShortTermLock<T>(workspaceDir: string, task: () => Promise<T>): Promise<T> {
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
// Msteams plugin module implements sqlite state behavior.
|
||||
import path from "node:path";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import { getMSTeamsRuntime } from "./runtime.js";
|
||||
import { withFileLock } from "./store-fs.js";
|
||||
|
||||
|
|
@ -55,28 +56,10 @@ export function resolveMSTeamsSqliteStateDir(
|
|||
);
|
||||
}
|
||||
|
||||
const sqliteMutationLocks = new Map<string, Promise<unknown>>();
|
||||
const sqliteMutationLocks = new KeyedAsyncQueue();
|
||||
|
||||
async function withProcessMutationLock<T>(lockPath: string, fn: () => Promise<T>): Promise<T> {
|
||||
const previous = sqliteMutationLocks.get(lockPath) ?? Promise.resolve();
|
||||
let release: () => void = () => {};
|
||||
const next = new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
const chained = previous.then(
|
||||
() => next,
|
||||
() => next,
|
||||
);
|
||||
sqliteMutationLocks.set(lockPath, chained);
|
||||
await previous.catch(() => undefined);
|
||||
try {
|
||||
return await fn();
|
||||
} finally {
|
||||
release();
|
||||
if (sqliteMutationLocks.get(lockPath) === chained) {
|
||||
sqliteMutationLocks.delete(lockPath);
|
||||
}
|
||||
}
|
||||
return await sqliteMutationLocks.enqueue(lockPath, fn);
|
||||
}
|
||||
|
||||
export async function withMSTeamsSqliteMutationLock<T>(
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@
|
|||
*/
|
||||
|
||||
import type { IncomingMessage, ServerResponse } from "node:http";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import {
|
||||
normalizeLowercaseStringOrEmpty,
|
||||
normalizeOptionalLowercaseString,
|
||||
|
|
@ -77,31 +78,10 @@ function checkRateLimit(accountId: string): boolean {
|
|||
// Mutex for Concurrent Publish Prevention
|
||||
// ============================================================================
|
||||
|
||||
const publishLocks = new Map<string, Promise<void>>();
|
||||
const publishLocks = new KeyedAsyncQueue();
|
||||
|
||||
async function withPublishLock<T>(accountId: string, fn: () => Promise<T>): Promise<T> {
|
||||
// Atomic mutex using promise chaining - prevents TOCTOU race condition
|
||||
const prev = publishLocks.get(accountId) ?? Promise.resolve();
|
||||
let resolve: () => void;
|
||||
const next = new Promise<void>((r) => {
|
||||
resolve = r;
|
||||
});
|
||||
// Atomically replace the lock before awaiting - any concurrent request
|
||||
// will now wait on our `next` promise
|
||||
publishLocks.set(accountId, next);
|
||||
|
||||
// Wait for previous operation to complete
|
||||
await prev.catch(() => {});
|
||||
|
||||
try {
|
||||
return await fn();
|
||||
} finally {
|
||||
resolve!();
|
||||
// Clean up if we're the last in chain
|
||||
if (publishLocks.get(accountId) === next) {
|
||||
publishLocks.delete(accountId);
|
||||
}
|
||||
}
|
||||
return await publishLocks.enqueue(accountId, fn);
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ import {
|
|||
} from "openclaw/plugin-sdk/channel-outbound";
|
||||
import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
|
||||
import { withTrustedEnvProxyGuardedFetchMode } from "openclaw/plugin-sdk/fetch-runtime";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import { resolveMarkdownTableMode } from "openclaw/plugin-sdk/markdown-table-runtime";
|
||||
import { requireRuntimeConfig } from "openclaw/plugin-sdk/plugin-config-runtime";
|
||||
import {
|
||||
|
|
@ -47,7 +48,7 @@ const SLACK_DNS_RETRY_CODES = new Set(["EAI_AGAIN", "ENOTFOUND", "UND_ERR_DNS_RE
|
|||
const SLACK_DNS_RETRY_ATTEMPTS = 2;
|
||||
const SLACK_DNS_RETRY_BASE_DELAY_MS = 250;
|
||||
const slackDmChannelCache = new Map<string, string>();
|
||||
const slackSendQueues = new Map<string, Promise<void>>();
|
||||
const slackSendQueue = new KeyedAsyncQueue();
|
||||
|
||||
type SlackRecipient =
|
||||
| {
|
||||
|
|
@ -488,22 +489,7 @@ function createSlackSendQueueKey(params: {
|
|||
}
|
||||
|
||||
async function runQueuedSlackSend<T>(key: string, task: () => Promise<T>): Promise<T> {
|
||||
const previous = slackSendQueues.get(key) ?? Promise.resolve();
|
||||
let releaseCurrent!: () => void;
|
||||
const current = new Promise<void>((resolve) => {
|
||||
releaseCurrent = resolve;
|
||||
});
|
||||
const queuedCurrent = previous.catch(() => undefined).then(() => current);
|
||||
slackSendQueues.set(key, queuedCurrent);
|
||||
await previous.catch(() => undefined);
|
||||
try {
|
||||
return await task();
|
||||
} finally {
|
||||
releaseCurrent();
|
||||
if (slackSendQueues.get(key) === queuedCurrent) {
|
||||
slackSendQueues.delete(key);
|
||||
}
|
||||
}
|
||||
return await slackSendQueue.enqueue(key, task);
|
||||
}
|
||||
|
||||
function createSlackDmCacheKey(params: {
|
||||
|
|
@ -602,10 +588,6 @@ export function clearSlackDmChannelCache(): void {
|
|||
slackDmChannelCache.clear();
|
||||
}
|
||||
|
||||
export function clearSlackSendQueuesForTest(): void {
|
||||
slackSendQueues.clear();
|
||||
}
|
||||
|
||||
export function clearSlackDefaultSendIdentitiesForTest(): void {
|
||||
slackDefaultSendIdentities.clear();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -54,8 +54,7 @@ vi.mock("./runtime-api.js", async () => {
|
|||
};
|
||||
});
|
||||
|
||||
const { sendMessageSlack, clearSlackDmChannelCache, clearSlackSendQueuesForTest } =
|
||||
await import("./send.js");
|
||||
const { sendMessageSlack, clearSlackDmChannelCache } = await import("./send.js");
|
||||
const SLACK_TEST_CFG = { channels: { slack: { botToken: "xoxb-test" } } };
|
||||
|
||||
type UploadTestClient = WebClient & {
|
||||
|
|
@ -167,7 +166,6 @@ describe("sendMessageSlack file upload with user IDs", () => {
|
|||
fetchWithSsrFGuard.mockClear();
|
||||
loadOutboundMediaFromUrlMock.mockClear();
|
||||
clearSlackDmChannelCache();
|
||||
clearSlackSendQueuesForTest();
|
||||
clearSlackThreadParticipationCache();
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ import {
|
|||
resolvePluginConversationBindingApproval,
|
||||
} from "openclaw/plugin-sdk/conversation-runtime";
|
||||
import { isApprovalNotFoundError } from "openclaw/plugin-sdk/error-runtime";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import { applyModelOverrideToSessionEntry } from "openclaw/plugin-sdk/model-session-runtime";
|
||||
import { formatModelsAvailableHeader } from "openclaw/plugin-sdk/models-provider-runtime";
|
||||
import { parseStrictPositiveInteger } from "openclaw/plugin-sdk/number-runtime";
|
||||
|
|
@ -261,7 +262,7 @@ export const registerTelegramHandlers = ({
|
|||
type PromptContextMessageSelection = ReadonlyMap<string, "include" | "exclude">;
|
||||
|
||||
const mediaGroupBuffer = new Map<string, BufferedMediaGroupEntry>();
|
||||
const mediaGroupProcessingByKey = new Map<string, Promise<void>>();
|
||||
const mediaGroupProcessingQueue = new KeyedAsyncQueue();
|
||||
const messageCache = createTelegramMessageCache({
|
||||
scope: resolveTelegramMessageCacheScope(telegramDeps.resolveStorePath(cfg.session?.store)),
|
||||
});
|
||||
|
|
@ -282,20 +283,16 @@ export const registerTelegramHandlers = ({
|
|||
timer: ReturnType<typeof setTimeout>;
|
||||
};
|
||||
const textFragmentBuffer = new Map<string, TextFragmentEntry>();
|
||||
const textFragmentProcessingByKey = new Map<string, Promise<void>>();
|
||||
const textFragmentProcessingQueue = new KeyedAsyncQueue();
|
||||
|
||||
const queueBufferedProcessing = async (
|
||||
processingByKey: Map<string, Promise<void>>,
|
||||
queue: KeyedAsyncQueue,
|
||||
key: string,
|
||||
task: () => Promise<void>,
|
||||
) => {
|
||||
const previous = processingByKey.get(key) ?? Promise.resolve();
|
||||
const current = previous.then(task).catch(() => undefined);
|
||||
processingByKey.set(key, current);
|
||||
await current;
|
||||
if (processingByKey.get(key) === current) {
|
||||
processingByKey.delete(key);
|
||||
}
|
||||
await queue.enqueue(key, async () => {
|
||||
await task().catch(() => undefined);
|
||||
});
|
||||
};
|
||||
|
||||
const debounceMs = resolveInboundDebounceMs({ cfg, channel: "telegram" });
|
||||
|
|
@ -1177,7 +1174,7 @@ export const registerTelegramHandlers = ({
|
|||
};
|
||||
|
||||
const queueTextFragmentFlush = async (entry: TextFragmentEntry) => {
|
||||
await queueBufferedProcessing(textFragmentProcessingByKey, entry.key, async () => {
|
||||
await queueBufferedProcessing(textFragmentProcessingQueue, entry.key, async () => {
|
||||
await flushTextFragments(entry);
|
||||
});
|
||||
};
|
||||
|
|
@ -2252,7 +2249,7 @@ export const registerTelegramHandlers = ({
|
|||
);
|
||||
existing.timer = setTimeout(() => {
|
||||
mediaGroupBuffer.delete(mediaGroupKey);
|
||||
void queueBufferedProcessing(mediaGroupProcessingByKey, mediaGroupKey, async () => {
|
||||
void queueBufferedProcessing(mediaGroupProcessingQueue, mediaGroupKey, async () => {
|
||||
await processMediaGroup(existing);
|
||||
});
|
||||
}, mediaGroupTimeoutMs);
|
||||
|
|
@ -2280,7 +2277,7 @@ export const registerTelegramHandlers = ({
|
|||
),
|
||||
timer: setTimeout(() => {
|
||||
mediaGroupBuffer.delete(mediaGroupKey);
|
||||
void queueBufferedProcessing(mediaGroupProcessingByKey, mediaGroupKey, async () => {
|
||||
void queueBufferedProcessing(mediaGroupProcessingQueue, mediaGroupKey, async () => {
|
||||
await processMediaGroup(entry);
|
||||
});
|
||||
}, mediaGroupTimeoutMs),
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ import {
|
|||
type SessionBindingRecord,
|
||||
} from "openclaw/plugin-sdk/conversation-runtime";
|
||||
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
|
||||
import { enqueueKeyedTask } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import type { PluginStateSyncKeyedStore } from "openclaw/plugin-sdk/plugin-state-runtime";
|
||||
import { normalizeAccountId, isAcpSessionKey } from "openclaw/plugin-sdk/routing";
|
||||
import { logVerbose } from "openclaw/plugin-sdk/runtime-env";
|
||||
|
|
@ -435,21 +436,14 @@ function enqueuePersistBindings(params: {
|
|||
if (!params.persist) {
|
||||
return Promise.resolve();
|
||||
}
|
||||
const previous =
|
||||
getThreadBindingsState().persistQueueByAccountId.get(params.accountId) ?? Promise.resolve();
|
||||
const next = previous
|
||||
.catch(() => undefined)
|
||||
.then(async () => {
|
||||
const state = getThreadBindingsState();
|
||||
return enqueueKeyedTask({
|
||||
tails: state.persistQueueByAccountId,
|
||||
key: params.accountId,
|
||||
task: async () => {
|
||||
await persistBindingsToStore(params);
|
||||
});
|
||||
getThreadBindingsState().persistQueueByAccountId.set(params.accountId, next);
|
||||
const cleanup = () => {
|
||||
if (getThreadBindingsState().persistQueueByAccountId.get(params.accountId) === next) {
|
||||
getThreadBindingsState().persistQueueByAccountId.delete(params.accountId);
|
||||
}
|
||||
};
|
||||
next.then(cleanup, cleanup);
|
||||
return next;
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function persistBindingsSafely(params: {
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
// Whatsapp plugin module implements creds persistence behavior.
|
||||
import { enqueueKeyedTask } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import { resolveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime";
|
||||
import { replaceFileAtomic } from "openclaw/plugin-sdk/security-runtime";
|
||||
import { assertWebCredsPathRegularFileOrMissing, resolveWebCredsPath } from "./creds-files.js";
|
||||
|
|
@ -48,18 +49,17 @@ export function enqueueCredsSave(
|
|||
saveCreds: () => Promise<void> | void,
|
||||
onError: (error: unknown) => void,
|
||||
): void {
|
||||
const previous = credsSaveQueues.get(authDir) ?? Promise.resolve();
|
||||
const next = previous
|
||||
.then(() => saveCreds())
|
||||
.catch((error: unknown) => {
|
||||
onError(error);
|
||||
})
|
||||
.finally(() => {
|
||||
if (credsSaveQueues.get(authDir) === next) {
|
||||
credsSaveQueues.delete(authDir);
|
||||
void enqueueKeyedTask({
|
||||
tails: credsSaveQueues,
|
||||
key: authDir,
|
||||
task: async () => {
|
||||
try {
|
||||
await saveCreds();
|
||||
} catch (error) {
|
||||
onError(error);
|
||||
}
|
||||
});
|
||||
credsSaveQueues.set(authDir, next);
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
export function waitForCredsSaveQueue(authDir?: string): Promise<void> {
|
||||
|
|
|
|||
|
|
@ -1442,49 +1442,6 @@ describe("AcpSessionManager", () => {
|
|||
expect(snapshot.errorsByCode.ACP_TURN_FAILED).toBe(1);
|
||||
});
|
||||
|
||||
it("cleans actor-tail bookkeeping after session turns complete", async () => {
|
||||
const runtimeState = createRuntime();
|
||||
hoisted.requireAcpRuntimeBackendMock.mockReturnValue({
|
||||
id: "acpx",
|
||||
runtime: runtimeState.runtime,
|
||||
});
|
||||
hoisted.readAcpSessionEntryMock.mockImplementation((paramsUnknown: unknown) => {
|
||||
const sessionKey = (paramsUnknown as { sessionKey?: string }).sessionKey ?? "";
|
||||
return {
|
||||
sessionKey,
|
||||
storeSessionKey: sessionKey,
|
||||
acp: {
|
||||
...readySessionMeta(),
|
||||
runtimeSessionName: `runtime:${sessionKey}`,
|
||||
},
|
||||
};
|
||||
});
|
||||
runtimeState.runTurn.mockImplementation(async function* () {
|
||||
yield { type: "done" as const };
|
||||
});
|
||||
|
||||
const manager = new AcpSessionManager();
|
||||
await manager.runTurn({
|
||||
cfg: baseCfg,
|
||||
sessionKey: "agent:codex:acp:session-a",
|
||||
text: "first",
|
||||
mode: "prompt",
|
||||
requestId: "r1",
|
||||
});
|
||||
await manager.runTurn({
|
||||
cfg: baseCfg,
|
||||
sessionKey: "agent:codex:acp:session-b",
|
||||
text: "second",
|
||||
mode: "prompt",
|
||||
requestId: "r2",
|
||||
});
|
||||
|
||||
const internals = manager as unknown as {
|
||||
actorQueue: { getTailMapForTesting(): Map<string, Promise<void>> };
|
||||
};
|
||||
expect(internals.actorQueue.getTailMapForTesting().size).toBe(0);
|
||||
});
|
||||
|
||||
it("surfaces backend failures raised after a done event", async () => {
|
||||
const runtimeState = createRuntime();
|
||||
hoisted.requireAcpRuntimeBackendMock.mockReturnValue({
|
||||
|
|
|
|||
|
|
@ -6,10 +6,6 @@ export class SessionActorQueue {
|
|||
private readonly queue = new KeyedAsyncQueue();
|
||||
private readonly pendingBySession = new Map<string, number>();
|
||||
|
||||
getTailMapForTesting(): Map<string, Promise<void>> {
|
||||
return this.queue.getTailMapForTesting();
|
||||
}
|
||||
|
||||
getTotalPendingCount(): number {
|
||||
let total = 0;
|
||||
for (const count of this.pendingBySession.values()) {
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
/**
|
||||
* OAuth credential manager.
|
||||
* Resolves usable access tokens, refreshes expired credentials under global
|
||||
|
|
@ -357,7 +358,7 @@ export function createOAuthManager(adapter: OAuthManagerAdapter) {
|
|||
return null;
|
||||
}
|
||||
|
||||
const refreshQueues = new Map<string, Promise<unknown>>();
|
||||
let refreshQueue = new KeyedAsyncQueue();
|
||||
|
||||
function refreshQueueKey(provider: string, profileId: string): string {
|
||||
return `${provider}\u0000${profileId}`;
|
||||
|
|
@ -653,21 +654,7 @@ export function createOAuthManager(adapter: OAuthManagerAdapter) {
|
|||
attemptedCredentials?: OAuthCredential[];
|
||||
}): Promise<ResolvedOAuthAccess | null> {
|
||||
const key = refreshQueueKey(params.provider, params.profileId);
|
||||
const prev = refreshQueues.get(key) ?? Promise.resolve();
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
refreshQueues.set(key, gate);
|
||||
try {
|
||||
await prev;
|
||||
return await doRefreshOAuthTokenWithLock(params);
|
||||
} finally {
|
||||
release();
|
||||
if (refreshQueues.get(key) === gate) {
|
||||
refreshQueues.delete(key);
|
||||
}
|
||||
}
|
||||
return await refreshQueue.enqueue(key, () => doRefreshOAuthTokenWithLock(params));
|
||||
}
|
||||
|
||||
async function resolveOAuthAccess(params: {
|
||||
|
|
@ -818,7 +805,7 @@ export function createOAuthManager(adapter: OAuthManagerAdapter) {
|
|||
}
|
||||
|
||||
function resetRefreshQueuesForTest(): void {
|
||||
refreshQueues.clear();
|
||||
refreshQueue = new KeyedAsyncQueue();
|
||||
}
|
||||
|
||||
return {
|
||||
|
|
|
|||
|
|
@ -115,21 +115,10 @@ describe("OAuth refresh in-process queue", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("resetOAuthRefreshQueuesForTest drains pending gates", () => {
|
||||
// We can't observe the internal map, but we can assert that calling the
|
||||
// reset is idempotent and safe from any state.
|
||||
expect(resetOAuthRefreshQueuesForTest()).toBeUndefined();
|
||||
expect(resetOAuthRefreshQueuesForTest()).toBeUndefined();
|
||||
});
|
||||
|
||||
it("serializes a 10-caller burst so later arrivals never pass an earlier caller", async () => {
|
||||
// Burst-arrival stress: 10 same-PID callers all fire concurrently.
|
||||
// The queue must chain them so each refresh completes fully before the
|
||||
// next one begins — i.e. no overlap between running refresh calls.
|
||||
// This pins the invariant that the map-overwrite pattern in the queue
|
||||
// wrapper does not let later arrivals skip ahead (see review P2: the
|
||||
// `refreshQueues.set(key, gate)` overwrites only the *map head*, while
|
||||
// FIFO ordering is enforced via the `await prev` chain).
|
||||
const profileId = "openai:default";
|
||||
const provider = "openai";
|
||||
saveAuthProfileStore(createExpiredOauthStore({ profileId, provider }), agentDir);
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
// Process-wide models.json coordination state. Dynamic imports can load this
|
||||
// module multiple times, so Symbol.for keeps write locks and ready-cache shared.
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
|
||||
const MODELS_JSON_STATE_KEY = Symbol.for("openclaw.modelsJsonState");
|
||||
|
||||
export type ModelsJsonReadyResult = {
|
||||
|
|
@ -13,7 +15,7 @@ export type ModelsJsonReadyState = {
|
|||
};
|
||||
|
||||
type ModelsJsonState = {
|
||||
writeLocks: Map<string, Promise<void>>;
|
||||
writeQueue: KeyedAsyncQueue;
|
||||
readyCache: Map<string, Promise<ModelsJsonReadyState>>;
|
||||
};
|
||||
|
||||
|
|
@ -23,7 +25,7 @@ export const MODELS_JSON_STATE = (() => {
|
|||
};
|
||||
if (!globalState[MODELS_JSON_STATE_KEY]) {
|
||||
globalState[MODELS_JSON_STATE_KEY] = {
|
||||
writeLocks: new Map<string, Promise<void>>(),
|
||||
writeQueue: new KeyedAsyncQueue(),
|
||||
readyCache: new Map<string, Promise<ModelsJsonReadyState>>(),
|
||||
};
|
||||
}
|
||||
|
|
@ -32,6 +34,6 @@ export const MODELS_JSON_STATE = (() => {
|
|||
|
||||
/** Clear models.json write/ready caches for tests. */
|
||||
export function resetModelsJsonReadyCacheForTest(): void {
|
||||
MODELS_JSON_STATE.writeLocks.clear();
|
||||
MODELS_JSON_STATE.writeQueue = new KeyedAsyncQueue();
|
||||
MODELS_JSON_STATE.readyCache.clear();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -340,22 +340,7 @@ export async function buildModelsJsonSourceFingerprint(
|
|||
}
|
||||
|
||||
async function withModelsJsonWriteLock<T>(targetPath: string, run: () => Promise<T>): Promise<T> {
|
||||
const prior = MODELS_JSON_STATE.writeLocks.get(targetPath) ?? Promise.resolve();
|
||||
let release: () => void = () => {};
|
||||
const gate = new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
const pending = prior.then(() => gate);
|
||||
MODELS_JSON_STATE.writeLocks.set(targetPath, pending);
|
||||
try {
|
||||
await prior;
|
||||
return await run();
|
||||
} finally {
|
||||
release();
|
||||
if (MODELS_JSON_STATE.writeLocks.get(targetPath) === pending) {
|
||||
MODELS_JSON_STATE.writeLocks.delete(targetPath);
|
||||
}
|
||||
}
|
||||
return await MODELS_JSON_STATE.writeQueue.enqueue(targetPath, run);
|
||||
}
|
||||
|
||||
/** Ensures models.json and plugin catalog sidecars are current for an agent. */
|
||||
|
|
|
|||
|
|
@ -5,8 +5,9 @@
|
|||
*/
|
||||
import { realpathSync } from "node:fs";
|
||||
import { resolve } from "node:path";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
|
||||
const fileMutationQueues = new Map<string, Promise<void>>();
|
||||
const fileMutationQueue = new KeyedAsyncQueue();
|
||||
|
||||
function getMutationQueueKey(filePath: string): string {
|
||||
const resolvedPath = resolve(filePath);
|
||||
|
|
@ -23,22 +24,5 @@ function getMutationQueueKey(filePath: string): string {
|
|||
*/
|
||||
export async function withFileMutationQueue<T>(filePath: string, fn: () => Promise<T>): Promise<T> {
|
||||
const key = getMutationQueueKey(filePath);
|
||||
const currentQueue = fileMutationQueues.get(key) ?? Promise.resolve();
|
||||
|
||||
let releaseNext!: () => void;
|
||||
const nextQueue = new Promise<void>((resolveQueue) => {
|
||||
releaseNext = resolveQueue;
|
||||
});
|
||||
const chainedQueue = currentQueue.then(() => nextQueue);
|
||||
fileMutationQueues.set(key, chainedQueue);
|
||||
|
||||
await currentQueue;
|
||||
try {
|
||||
return await fn();
|
||||
} finally {
|
||||
releaseNext();
|
||||
if (fileMutationQueues.get(key) === chainedQueue) {
|
||||
fileMutationQueues.delete(key);
|
||||
}
|
||||
}
|
||||
return await fileMutationQueue.enqueue(key, fn);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ import { AsyncLocalStorage } from "node:async_hooks";
|
|||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { isDeepStrictEqual } from "node:util";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import { formatErrorMessage } from "../infra/errors.js";
|
||||
import { withFileLock } from "../infra/file-lock.js";
|
||||
import { root as createFsRoot, type Root as FsSafeRoot } from "../infra/fs-safe.js";
|
||||
|
|
@ -68,7 +69,7 @@ const CONFIG_MUTATION_LOCK_OPTIONS = {
|
|||
|
||||
const DEFAULT_CONFIG_MUTATION_RETRY_ATTEMPTS = 5;
|
||||
const activeConfigMutationLocks = new AsyncLocalStorage<Set<string>>();
|
||||
const configMutationQueueTails = new Map<string, Promise<void>>();
|
||||
const configMutationQueue = new KeyedAsyncQueue();
|
||||
|
||||
export { ConfigMutationConflictError } from "./mutation-conflict.js";
|
||||
|
||||
|
|
@ -187,28 +188,14 @@ async function withConfigMutationLock<T>(
|
|||
assertConfigWriteAllowedInCurrentMode({ configPath });
|
||||
await fs.mkdir(path.dirname(configPath), { recursive: true, mode: 0o700 });
|
||||
|
||||
const previousTail = configMutationQueueTails.get(configPath) ?? Promise.resolve();
|
||||
let releaseQueueSlot!: () => void;
|
||||
const currentRun = new Promise<void>((resolve) => {
|
||||
releaseQueueSlot = resolve;
|
||||
});
|
||||
const currentTail = previousTail.catch(() => undefined).then(() => currentRun);
|
||||
configMutationQueueTails.set(configPath, currentTail);
|
||||
|
||||
await previousTail.catch(() => undefined);
|
||||
try {
|
||||
const nextActiveLocks = new Set(activeLocks ?? []);
|
||||
nextActiveLocks.add(configPath);
|
||||
return await activeConfigMutationLocks.run(
|
||||
const nextActiveLocks = new Set(activeLocks ?? []);
|
||||
nextActiveLocks.add(configPath);
|
||||
return await configMutationQueue.enqueue(configPath, () =>
|
||||
activeConfigMutationLocks.run(
|
||||
nextActiveLocks,
|
||||
async () => await withFileLock(configPath, CONFIG_MUTATION_LOCK_OPTIONS, fn),
|
||||
);
|
||||
} finally {
|
||||
releaseQueueSlot();
|
||||
if (configMutationQueueTails.get(configPath) === currentTail) {
|
||||
configMutationQueueTails.delete(configPath);
|
||||
}
|
||||
}
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
function markActiveConfigMutationPath(configPath: string): void {
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ import { randomUUID } from "node:crypto";
|
|||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { resolveTimestampMsToIsoString } from "@openclaw/normalization-core/number-coercion";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import type { AgentMessage } from "../../agents/runtime/index.js";
|
||||
import {
|
||||
acquireSessionWriteLock,
|
||||
|
|
@ -34,7 +35,7 @@ import { CURRENT_SESSION_VERSION } from "./version.js";
|
|||
|
||||
const SESSION_MANAGER_APPEND_MAX_BYTES = 8 * 1024 * 1024;
|
||||
|
||||
const transcriptAppendQueues = new Map<string, Promise<void>>();
|
||||
const transcriptAppendQueue = new KeyedAsyncQueue();
|
||||
|
||||
type TranscriptLeafInfo = {
|
||||
leafId?: string;
|
||||
|
|
@ -397,24 +398,9 @@ export async function withSessionTranscriptAppendQueue<T>(
|
|||
fn: () => Promise<T>,
|
||||
): Promise<T> {
|
||||
const queueKey = await resolveTranscriptAppendQueueKey(transcriptPath);
|
||||
const previous = transcriptAppendQueues.get(queueKey) ?? Promise.resolve();
|
||||
let releaseCurrent!: () => void;
|
||||
const current = new Promise<void>((resolve) => {
|
||||
releaseCurrent = resolve;
|
||||
});
|
||||
const tail = previous.catch(() => undefined).then(() => current);
|
||||
// Per-file queue is in-process only; the external session write lock still owns cross-process
|
||||
// ordering.
|
||||
transcriptAppendQueues.set(queueKey, tail);
|
||||
await previous.catch(() => undefined);
|
||||
try {
|
||||
return await fn();
|
||||
} finally {
|
||||
releaseCurrent();
|
||||
if (transcriptAppendQueues.get(queueKey) === tail) {
|
||||
transcriptAppendQueues.delete(queueKey);
|
||||
}
|
||||
}
|
||||
return await transcriptAppendQueue.enqueue(queueKey, fn);
|
||||
}
|
||||
|
||||
export type AppendSessionTranscriptMessageParams<TMessage = unknown> = {
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import {
|
|||
normalizeStringifiedOptionalString,
|
||||
} from "@openclaw/normalization-core/string-coerce";
|
||||
import { uniqueValues } from "@openclaw/normalization-core/string-normalization";
|
||||
import { enqueueKeyedTask } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import { parseByteSize } from "../cli/parse-bytes.js";
|
||||
import type { CronConfig } from "../config/types.cron.js";
|
||||
import {
|
||||
|
|
@ -150,11 +151,11 @@ export async function appendCronRunLog(params: {
|
|||
: { ...params.entry, jobId: normalizedJobId };
|
||||
const storeKey = cronStoreKey(params.storePath);
|
||||
const writeKey = cronRunLogWriteKey(params.storePath, entry.jobId);
|
||||
const prev = writesByTarget.get(writeKey) ?? Promise.resolve();
|
||||
// Keep writes for the same store/job ordered so prune-by-count cannot race a later insert.
|
||||
const next = prev
|
||||
.catch(() => undefined)
|
||||
.then(async () => {
|
||||
await enqueueKeyedTask({
|
||||
tails: writesByTarget,
|
||||
key: writeKey,
|
||||
task: async () => {
|
||||
runOpenClawStateWriteTransaction(({ db }) => {
|
||||
insertCronRunLogEntry(db, storeKey, entry);
|
||||
if (params.opts?.keepLines !== false) {
|
||||
|
|
@ -166,15 +167,8 @@ export async function appendCronRunLog(params: {
|
|||
);
|
||||
}
|
||||
});
|
||||
});
|
||||
writesByTarget.set(writeKey, next);
|
||||
try {
|
||||
await next;
|
||||
} finally {
|
||||
if (writesByTarget.get(writeKey) === next) {
|
||||
writesByTarget.delete(writeKey);
|
||||
}
|
||||
}
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/** Reads recent run-log entries synchronously for startup/task reconciliation paths. */
|
||||
|
|
|
|||
|
|
@ -1,8 +1,9 @@
|
|||
// Gateway auth rate-limit serialization.
|
||||
// Serializes limiter attempts per IP/scope so concurrent failures count correctly.
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import { AUTH_RATE_LIMIT_SCOPE_DEFAULT, normalizeRateLimitClientIp } from "./auth-rate-limit.js";
|
||||
|
||||
const pendingAttempts = new Map<string, Promise<void>>();
|
||||
const pendingAttempts = new KeyedAsyncQueue();
|
||||
|
||||
function normalizeScope(scope: string | undefined): string {
|
||||
return (scope ?? AUTH_RATE_LIMIT_SCOPE_DEFAULT).trim() || AUTH_RATE_LIMIT_SCOPE_DEFAULT;
|
||||
|
|
@ -17,24 +18,7 @@ export async function withSerializedKeyedAttempt<T>(params: {
|
|||
key: string;
|
||||
run: () => Promise<T>;
|
||||
}): Promise<T> {
|
||||
const key = params.key;
|
||||
const previous = pendingAttempts.get(key) ?? Promise.resolve();
|
||||
let releaseCurrent!: () => void;
|
||||
const current = new Promise<void>((resolve) => {
|
||||
releaseCurrent = resolve;
|
||||
});
|
||||
const tail = previous.catch(() => {}).then(() => current);
|
||||
pendingAttempts.set(key, tail);
|
||||
|
||||
await previous.catch(() => {});
|
||||
try {
|
||||
return await params.run();
|
||||
} finally {
|
||||
releaseCurrent();
|
||||
if (pendingAttempts.get(key) === tail) {
|
||||
pendingAttempts.delete(key);
|
||||
}
|
||||
}
|
||||
return await pendingAttempts.enqueue(params.key, params.run);
|
||||
}
|
||||
|
||||
/** Runs one rate-limit attempt after prior attempts for the same IP/scope finish. */
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import {
|
|||
normalizeOptionalString,
|
||||
readStringValue,
|
||||
} from "@openclaw/normalization-core/string-coerce";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import {
|
||||
ErrorCodes,
|
||||
errorShape,
|
||||
|
|
@ -410,7 +411,7 @@ async function mirrorDeliveredSourceReplyToTranscriptBestEffort(params: {
|
|||
}
|
||||
}
|
||||
|
||||
const sourceReplyTranscriptMirrorQueues = new Map<string, Promise<void>>();
|
||||
const sourceReplyTranscriptMirrorQueue = new KeyedAsyncQueue();
|
||||
|
||||
function resolveSourceReplyTranscriptMirrorQueueKey(
|
||||
mirror: Parameters<typeof mirrorDeliveredSourceReplyToTranscript>[0],
|
||||
|
|
@ -424,22 +425,11 @@ function scheduleDeliveredSourceReplyTranscriptMirror(params: {
|
|||
mirror: Parameters<typeof mirrorDeliveredSourceReplyToTranscript>[0];
|
||||
}): Promise<void> {
|
||||
const queueKey = resolveSourceReplyTranscriptMirrorQueueKey(params.mirror);
|
||||
const previous = sourceReplyTranscriptMirrorQueues.get(queueKey);
|
||||
// Queue per session so current-conversation source replies are visible before
|
||||
// a following turn can read the transcript.
|
||||
const queued = (async () => {
|
||||
await previous?.catch(() => undefined);
|
||||
await mirrorDeliveredSourceReplyToTranscriptBestEffort(params);
|
||||
})();
|
||||
sourceReplyTranscriptMirrorQueues.set(queueKey, queued);
|
||||
void queued
|
||||
.finally(() => {
|
||||
if (sourceReplyTranscriptMirrorQueues.get(queueKey) === queued) {
|
||||
sourceReplyTranscriptMirrorQueues.delete(queueKey);
|
||||
}
|
||||
})
|
||||
.catch(() => undefined);
|
||||
return queued;
|
||||
return sourceReplyTranscriptMirrorQueue.enqueue(queueKey, () =>
|
||||
mirrorDeliveredSourceReplyToTranscriptBestEffort(params),
|
||||
);
|
||||
}
|
||||
|
||||
export const sendHandlers: GatewayRequestHandlers = {
|
||||
|
|
|
|||
|
|
@ -116,17 +116,29 @@ describe("enqueueKeyedTask", () => {
|
|||
});
|
||||
|
||||
describe("KeyedAsyncQueue", () => {
|
||||
it("exposes tail map for observability", async () => {
|
||||
it("serializes tasks while preserving their return values", async () => {
|
||||
const queue = new KeyedAsyncQueue();
|
||||
const gate = createDeferred();
|
||||
const run = queue.enqueue("actor", async () => {
|
||||
const order: string[] = [];
|
||||
const first = queue.enqueue("actor", async () => {
|
||||
order.push("first:start");
|
||||
await gate.promise;
|
||||
return 1;
|
||||
});
|
||||
expect(queue.getTailMapForTesting().has("actor")).toBe(true);
|
||||
gate.resolve();
|
||||
await run;
|
||||
const second = queue.enqueue("actor", async () => {
|
||||
order.push("second:start");
|
||||
return 2;
|
||||
});
|
||||
await Promise.resolve();
|
||||
expect(queue.getTailMapForTesting().has("actor")).toBe(false);
|
||||
await Promise.resolve();
|
||||
expect(order).toEqual(["first:start"]);
|
||||
gate.resolve();
|
||||
await expect(Promise.all([first, second])).resolves.toEqual([1, 2]);
|
||||
expect(order).toEqual(["first:start", "second:start"]);
|
||||
});
|
||||
|
||||
it("retains the deprecated tail-map accessor for Plugin SDK compatibility", () => {
|
||||
const queue = new KeyedAsyncQueue();
|
||||
expect(queue.getTailMapForTesting()).toBeInstanceOf(Map);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -38,6 +38,10 @@ export function enqueueKeyedTask<T>(params: {
|
|||
export class KeyedAsyncQueue {
|
||||
private readonly tails = new Map<string, Promise<void>>();
|
||||
|
||||
/**
|
||||
* @deprecated Retained for shipped Plugin SDK compatibility. New callers must
|
||||
* not depend on queue storage; remove in a declared Plugin SDK breaking window.
|
||||
*/
|
||||
getTailMapForTesting(): Map<string, Promise<void>> {
|
||||
return this.tails;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,16 +1,9 @@
|
|||
// Skill serialization helpers compact skill metadata and coordinate sync queue updates.
|
||||
const SKILLS_SYNC_QUEUE = new Map<string, Promise<unknown>>();
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
|
||||
const skillsSyncQueue = new KeyedAsyncQueue();
|
||||
|
||||
/** Serializes async work by key so repeated skill loads do not race on shared files. */
|
||||
export async function serializeByKey<T>(key: string, task: () => Promise<T>) {
|
||||
const prev = SKILLS_SYNC_QUEUE.get(key) ?? Promise.resolve();
|
||||
const next = prev.then(task, task);
|
||||
SKILLS_SYNC_QUEUE.set(key, next);
|
||||
try {
|
||||
return await next;
|
||||
} finally {
|
||||
if (SKILLS_SYNC_QUEUE.get(key) === next) {
|
||||
SKILLS_SYNC_QUEUE.delete(key);
|
||||
}
|
||||
}
|
||||
return await skillsSyncQueue.enqueue(key, task);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ import crypto from "node:crypto";
|
|||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import { resolveStateDir } from "../../config/paths.js";
|
||||
import { sha256Hex } from "../../infra/crypto-digest.js";
|
||||
import { type FileLockOptions, withFileLock } from "../../infra/file-lock.js";
|
||||
|
|
@ -49,7 +50,7 @@ const SKILL_WORKSHOP_LOCK_OPTIONS: FileLockOptions = {
|
|||
},
|
||||
stale: 60_000,
|
||||
};
|
||||
const skillWorkshopProcessLocks = new Map<string, Promise<void>>();
|
||||
const skillWorkshopProcessLocks = new KeyedAsyncQueue();
|
||||
|
||||
type SkillWorkshopStoreOptions = {
|
||||
env?: NodeJS.ProcessEnv;
|
||||
|
|
@ -363,24 +364,10 @@ async function withSkillProposalManifestLock<T>(
|
|||
|
||||
async function withSkillWorkshopLock<T>(lockFile: string, fn: () => Promise<T>): Promise<T> {
|
||||
const lockKey = path.resolve(lockFile);
|
||||
const previous = skillWorkshopProcessLocks.get(lockKey) ?? Promise.resolve();
|
||||
let releaseQueued!: () => void;
|
||||
const current = new Promise<void>((resolve) => {
|
||||
releaseQueued = resolve;
|
||||
});
|
||||
const previousDone = previous.catch(() => undefined);
|
||||
const queued = previousDone.then(() => current);
|
||||
skillWorkshopProcessLocks.set(lockKey, queued);
|
||||
await previousDone;
|
||||
await fs.mkdir(path.dirname(lockFile), { recursive: true });
|
||||
try {
|
||||
return await skillWorkshopProcessLocks.enqueue(lockKey, async () => {
|
||||
await fs.mkdir(path.dirname(lockFile), { recursive: true });
|
||||
return await withFileLock(lockFile, SKILL_WORKSHOP_LOCK_OPTIONS, fn);
|
||||
} finally {
|
||||
releaseQueued();
|
||||
if (skillWorkshopProcessLocks.get(lockKey) === queued) {
|
||||
skillWorkshopProcessLocks.delete(lockKey);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
export async function readProposalSupportFiles(
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
// Trajectory runtime records runtime events into trajectory log files.
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
|
||||
import { sanitizeDiagnosticPayload } from "../agents/payload-redaction.js";
|
||||
import type {
|
||||
QueuedFileWriter,
|
||||
|
|
@ -46,7 +47,7 @@ type TrajectoryRuntimeRecorder = {
|
|||
};
|
||||
|
||||
const writers = new Map<string, TrajectoryRuntimeWriter>();
|
||||
const windowFlushes = new Map<string, Promise<void>>();
|
||||
const windowFlushes = new KeyedAsyncQueue();
|
||||
const MAX_TRAJECTORY_WRITERS = 100;
|
||||
const TRAJECTORY_RUNTIME_DATA_STRING_MAX_CHARS = 32_768;
|
||||
const TRAJECTORY_RUNTIME_DATA_ARRAY_MAX_ITEMS = 64;
|
||||
|
|
@ -389,19 +390,9 @@ async function queueTrajectoryWindowFlush(params: {
|
|||
maxFileBytes: number;
|
||||
appendedLines: string[];
|
||||
}): Promise<void> {
|
||||
const previous = windowFlushes.get(params.filePath) ?? Promise.resolve();
|
||||
const current = previous
|
||||
.catch(() => undefined)
|
||||
.then(async () => {
|
||||
await replaceTrajectoryWindow(params);
|
||||
})
|
||||
.finally(() => {
|
||||
if (windowFlushes.get(params.filePath) === current) {
|
||||
windowFlushes.delete(params.filePath);
|
||||
}
|
||||
});
|
||||
windowFlushes.set(params.filePath, current);
|
||||
await current;
|
||||
await windowFlushes.enqueue(params.filePath, async () => {
|
||||
await replaceTrajectoryWindow(params);
|
||||
});
|
||||
}
|
||||
|
||||
function createTrajectoryWindowWriter(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue