lunanting/ai-chatbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add image batch dispatch pacing
Some checks failed
Lint / build (20) (push) Has been cancelled
Details
Playwright Tests / test (push) Has been cancelled
Details

Browse Source
This commit is contained in:
Your Name 2026-05-01 01:14:07 +08:00
parent e7cf10bb28
commit 94a22f0cc6
4 changed files with 317 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
app/(admin)/settings/admin-settings-client.tsx
View File

@ -78,8 +78,11 @@ import {
} from "@/lib/model3d/model-alias-config";
import {
CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY,
DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS,
DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY,
MAX_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS,
MIN_CANVAS_IMAGE_TASK_CONCURRENCY,
MIN_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS,
MAX_CANVAS_IMAGE_TASK_CONCURRENCY,
normalizeCanvasImageTaskQueueConfig,
type CanvasImageTaskQueueConfig,
@ -2087,6 +2090,7 @@ export function AdminSettingsClient() {
useState<CanvasImageTaskQueueConfig>(() =>
normalizeCanvasImageTaskQueueConfig({
concurrency: DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY,
batchDispatchIntervalMs: DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS,
})
);
@ -5110,10 +5114,10 @@ export function AdminSettingsClient() {
normalizeCanvasImageTaskQueueConfig(payload?.setting?.value || normalized)
);
await mutateCanvasImageTaskQueueConfig();
toast.success("全站生图单模型并发配置保存成功");
toast.success("全站生图并发与批次间隔配置保存成功");
} catch (error) {
toast.error(
error instanceof Error ? error.message : "保存全站生图单模型并发配置失败"
error instanceof Error ? error.message : "保存全站生图并发与批次间隔配置失败"
);
}
};
@ -20155,10 +20159,10 @@ export function AdminSettingsClient() {
<div className="space-y-1.5">
<CardTitle className="flex items-center gap-2 text-xl">
<ImageIcon className="h-5 w-5" />
全站生图单模型并发
全站生图单模型并发与批次间隔
</CardTitle>
<CardDescription>
控制全站生图链路中，同一个模型允许同时运行的任务数。配置保存到数据库，异步队列和直调生成都会自动读取，无需重启服务。
控制全站生图链路中，同一个模型允许同时运行的任务数，以及排队后下一批任务的发起间隔。配置保存到数据库，异步队列和直调生成都会自动读取，无需重启服务。
</CardDescription>
</div>
<LocalAiSectionToggleButton
@ -20179,40 +20183,78 @@ export function AdminSettingsClient() {
<p>• 影响画布异步生图、智能体直调生图、视频分镜高清预览等全站生图入口。</p>
<p>• 多用户共享同一套并发限制，但会按模型分别计数，不同模型之间不再互相阻塞。</p>
<p>• 当前配置的是单模型并发，不是全站总并发，也不是单用户并发。</p>
<p>• 批次间隔只控制排队补位的“下一批”发起节奏，不影响首批任务即时启动。</p>
<p>• 数值过高可能触发上游模型 429、超时或 504，建议配合模型 Key 池继续细分。</p>
</div>
</div>
<div className="space-y-2">
<Label htmlFor="canvasImageTaskQueueConcurrency">
单模型最大同时运行任务数
</Label>
<Input
id="canvasImageTaskQueueConcurrency"
type="number"
min={MIN_CANVAS_IMAGE_TASK_CONCURRENCY}
max={MAX_CANVAS_IMAGE_TASK_CONCURRENCY}
value={canvasImageTaskQueueConfig.concurrency}
onChange={(e) =>
setCanvasImageTaskQueueConfig(
normalizeCanvasImageTaskQueueConfig({
concurrency: e.target.value,
})
)
}
/>
<div className="text-xs text-muted-foreground">
可配置范围 {MIN_CANVAS_IMAGE_TASK_CONCURRENCY}-
{MAX_CANVAS_IMAGE_TASK_CONCURRENCY}，当前为{" "}
{canvasImageTaskQueueConfig.concurrency} 并发，默认值为{" "}
{DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY}。
<div className="space-y-4">
<div className="space-y-2">
<Label htmlFor="canvasImageTaskQueueConcurrency">
单模型最大同时运行任务数
</Label>
<Input
id="canvasImageTaskQueueConcurrency"
type="number"
min={MIN_CANVAS_IMAGE_TASK_CONCURRENCY}
max={MAX_CANVAS_IMAGE_TASK_CONCURRENCY}
value={canvasImageTaskQueueConfig.concurrency}
onChange={(e) =>
setCanvasImageTaskQueueConfig(
normalizeCanvasImageTaskQueueConfig({
...canvasImageTaskQueueConfig,
concurrency: e.target.value,
})
)
}
/>
<div className="text-xs text-muted-foreground">
可配置范围 {MIN_CANVAS_IMAGE_TASK_CONCURRENCY}-
{MAX_CANVAS_IMAGE_TASK_CONCURRENCY}，当前为{" "}
{canvasImageTaskQueueConfig.concurrency} 并发，默认值为{" "}
{DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY}。
</div>
</div>
<div className="space-y-2">
<Label htmlFor="canvasImageTaskBatchDispatchIntervalSeconds">
下一批排队任务发起间隔（秒）
</Label>
<Input
id="canvasImageTaskBatchDispatchIntervalSeconds"
type="number"
min={MIN_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS / 1000}
max={MAX_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS / 1000}
step="0.5"
value={String(
canvasImageTaskQueueConfig.batchDispatchIntervalMs / 1000
)}
onChange={(e) =>
setCanvasImageTaskQueueConfig(
normalizeCanvasImageTaskQueueConfig({
...canvasImageTaskQueueConfig,
batchDispatchIntervalMs: Math.round(
Number.parseFloat(e.target.value || "0") * 1000
),
})
)
}
/>
<div className="text-xs text-muted-foreground">
可配置范围 {MIN_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS / 1000}-
{MAX_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS / 1000} 秒，当前为{" "}
{(canvasImageTaskQueueConfig.batchDispatchIntervalMs / 1000).toFixed(1)}{" "}
秒，默认值为{" "}
{(DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS / 1000).toFixed(1)}{" "}
秒。
</div>
</div>
</div>
</div>
<div className="flex justify-end">
<Button onClick={handleSaveCanvasImageTaskQueueConfig}>
<Save className="mr-2 h-4 w-4" /> 保存单模型并发
<Save className="mr-2 h-4 w-4" /> 保存并发与间隔
</Button>
</div>
</CardContent>

109
lib/canvas/image-concurrency.ts
View File

@ -3,7 +3,9 @@ import "server-only";
import { getSystemSetting } from "@/lib/db/queries";
import {
CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY,
DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS,
DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY,
type CanvasImageTaskQueueConfig,
normalizeCanvasImageTaskQueueConfig,
} from "./image-task-config";
@ -18,16 +20,18 @@ interface ModelWaiter {
interface ModelConcurrencyState {
running: Set<string>;
waiters: ModelWaiter[];
dispatchTimer: ReturnType<typeof setTimeout> | null;
delayedUntil: number | null;
}
interface CanvasImageConcurrencyState {
models: Map<string, ModelConcurrencyState>;
}
let cachedConcurrency:
let cachedConfig:
| {
expiresAt: number;
value: number;
value: CanvasImageTaskQueueConfig;
}
| null = null;
@ -55,11 +59,22 @@ function getOrCreateModelState(modelKey: string) {
const created: ModelConcurrencyState = {
running: new Set<string>(),
waiters: [],
dispatchTimer: null,
delayedUntil: null,
};
state.models.set(modelKey, created);
return created;
}
function clearModelDispatchTimer(modelState: ModelConcurrencyState) {
if (modelState.dispatchTimer) {
clearTimeout(modelState.dispatchTimer);
modelState.dispatchTimer = null;
}
modelState.delayedUntil = null;
}
function cleanupModelState(modelKey: string) {
const state = getGlobalConcurrencyState();
const modelState = state.models.get(modelKey);
@ -68,6 +83,7 @@ function cleanupModelState(modelKey: string) {
}
if (modelState.running.size === 0 && modelState.waiters.length === 0) {
clearModelDispatchTimer(modelState);
state.models.delete(modelKey);
}
}
@ -81,25 +97,28 @@ export function normalizeCanvasImageModelKey(value: unknown) {
return normalized || "__default__";
}
export async function getCanvasImageSingleModelConcurrency(): Promise<number> {
export async function getCanvasImageTaskQueueConfig(): Promise<CanvasImageTaskQueueConfig> {
const now = Date.now();
if (cachedConcurrency && cachedConcurrency.expiresAt > now) {
return cachedConcurrency.value;
if (cachedConfig && cachedConfig.expiresAt > now) {
return cachedConfig.value;
}
const envConcurrency = process.env.CANVAS_IMAGE_TASK_CONCURRENCY;
let value = normalizeCanvasImageTaskQueueConfig(undefined, {
const envBatchDispatchIntervalMs = process.env.CANVAS_IMAGE_TASK_BATCH_DISPATCH_INTERVAL_MS;
let config = normalizeCanvasImageTaskQueueConfig(undefined, {
envConcurrency,
}).concurrency;
envBatchDispatchIntervalMs,
});
try {
const setting = await getSystemSetting({
key: CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY,
});
if (setting?.value !== undefined && setting.value !== null) {
value = normalizeCanvasImageTaskQueueConfig(setting.value, {
config = normalizeCanvasImageTaskQueueConfig(setting.value, {
envConcurrency,
}).concurrency;
envBatchDispatchIntervalMs,
});
}
} catch (error) {
console.warn("[Canvas Image Concurrency] Failed to load DB config, using fallback", {
@ -107,12 +126,25 @@ export async function getCanvasImageSingleModelConcurrency(): Promise<number> {
});
}
cachedConcurrency = {
cachedConfig = {
expiresAt: now + CONCURRENCY_CONFIG_CACHE_MS,
value: value || DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY,
value: {
concurrency: config.concurrency || DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY,
batchDispatchIntervalMs: Number.isFinite(config.batchDispatchIntervalMs)
? config.batchDispatchIntervalMs
: DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS,
},
};
return cachedConcurrency.value;
return cachedConfig.value;
}
export async function getCanvasImageSingleModelConcurrency(): Promise<number> {
return (await getCanvasImageTaskQueueConfig()).concurrency;
}
export async function getCanvasImageBatchDispatchIntervalMs(): Promise<number> {
return (await getCanvasImageTaskQueueConfig()).batchDispatchIntervalMs;
}
export function getCanvasImageRunningCountForModel(modelKey: string) {
@ -120,7 +152,10 @@ export function getCanvasImageRunningCountForModel(modelKey: string) {
return getGlobalConcurrencyState().models.get(normalizedModelKey)?.running.size || 0;
}
async function drainModelWaiters(modelKey: string) {
async function drainModelWaiters(
modelKey: string,
trigger: "arrival" | "slot-release" | "timer"
) {
const normalizedModelKey = normalizeCanvasImageModelKey(modelKey);
const modelState = getGlobalConcurrencyState().models.get(normalizedModelKey);
if (!modelState || modelState.waiters.length === 0) {
@ -128,7 +163,31 @@ async function drainModelWaiters(modelKey: string) {
return;
}
const maxConcurrent = await getCanvasImageSingleModelConcurrency();
const { concurrency: maxConcurrent, batchDispatchIntervalMs } =
await getCanvasImageTaskQueueConfig();
if (modelState.running.size >= maxConcurrent) {
return;
}
const now = Date.now();
if (modelState.delayedUntil && modelState.delayedUntil > now) {
return;
}
if (trigger === "slot-release" && batchDispatchIntervalMs > 0) {
if (!modelState.dispatchTimer) {
modelState.delayedUntil = now + batchDispatchIntervalMs;
modelState.dispatchTimer = setTimeout(() => {
const state = getGlobalConcurrencyState().models.get(normalizedModelKey);
if (!state) return;
clearModelDispatchTimer(state);
void drainModelWaiters(normalizedModelKey, "timer");
}, batchDispatchIntervalMs);
}
return;
}
clearModelDispatchTimer(modelState);
while (modelState.running.size < maxConcurrent && modelState.waiters.length > 0) {
const waiter = modelState.waiters.shift();
if (!waiter) {
@ -147,22 +206,32 @@ export async function acquireCanvasImageModelSlot(params: {
debugMeta?: Record<string, unknown>;
}) {
const normalizedModelKey = normalizeCanvasImageModelKey(params.modelKey);
const maxConcurrent = await getCanvasImageSingleModelConcurrency();
const {
concurrency: maxConcurrent,
batchDispatchIntervalMs,
} = await getCanvasImageTaskQueueConfig();
const modelState = getOrCreateModelState(normalizedModelKey);
const ticketId = `${normalizedModelKey}:${Date.now()}:${Math.random().toString(36).slice(2, 10)}`;
const waitingSince = Date.now();
const hasActiveDelay =
modelState.delayedUntil !== null && modelState.delayedUntil > waitingSince;
if (modelState.waiters.length === 0 && modelState.running.size < maxConcurrent) {
if (
modelState.waiters.length === 0 &&
!hasActiveDelay &&
modelState.running.size < maxConcurrent
) {
modelState.running.add(ticketId);
return {
release: async () => {
const state = getGlobalConcurrencyState().models.get(normalizedModelKey);
if (!state) return;
state.running.delete(ticketId);
await drainModelWaiters(normalizedModelKey);
await drainModelWaiters(normalizedModelKey, "slot-release");
},
waitMs: 0,
maxConcurrent,
batchDispatchIntervalMs,
};
}
@ -171,7 +240,7 @@ export async function acquireCanvasImageModelSlot(params: {
ticketId,
resolve,
});
void drainModelWaiters(normalizedModelKey);
void drainModelWaiters(normalizedModelKey, "arrival");
});
const waitMs = Date.now() - waitingSince;
@ -180,6 +249,7 @@ export async function acquireCanvasImageModelSlot(params: {
debugLabel: params.debugLabel || "canvas-image",
modelKey: normalizedModelKey,
waitMs,
batchDispatchIntervalMs,
...params.debugMeta,
});
}
@ -189,9 +259,10 @@ export async function acquireCanvasImageModelSlot(params: {
const state = getGlobalConcurrencyState().models.get(normalizedModelKey);
if (!state) return;
state.running.delete(ticketId);
await drainModelWaiters(normalizedModelKey);
await drainModelWaiters(normalizedModelKey, "slot-release");
},
waitMs,
maxConcurrent,
batchDispatchIntervalMs,
};
}

84
lib/canvas/image-task-config.ts
View File

@ -3,9 +3,13 @@ export const CANVAS_IMAGE_TASK_QUEUE_CONFIG_KEY = "canvas_image_task_queue_confi
export const MIN_CANVAS_IMAGE_TASK_CONCURRENCY = 1;
export const DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY = 8;
export const MAX_CANVAS_IMAGE_TASK_CONCURRENCY = 100;
export const MIN_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS = 0;
export const DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS = 0;
export const MAX_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS = 60 * 1000;
export interface CanvasImageTaskQueueConfig {
concurrency: number;
batchDispatchIntervalMs: number;
}
function clampConcurrency(value: number) {
@ -15,6 +19,13 @@ function clampConcurrency(value: number) {
);
}
function clampBatchDispatchIntervalMs(value: number) {
return Math.min(
MAX_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS,
Math.max(MIN_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS, Math.floor(value))
);
}
function parseConcurrencyCandidate(value: unknown): number | null {
if (typeof value === "number" && Number.isFinite(value)) {
return clampConcurrency(value);
@ -30,14 +41,38 @@ function parseConcurrencyCandidate(value: unknown): number | null {
return null;
}
function parseBatchDispatchIntervalCandidate(
value: unknown,
options?: { treatAsSeconds?: boolean }
): number | null {
const normalize = (candidate: number) =>
clampBatchDispatchIntervalMs(
options?.treatAsSeconds ? candidate * 1000 : candidate
);
if (typeof value === "number" && Number.isFinite(value)) {
return normalize(value);
}
if (typeof value === "string" && value.trim()) {
const parsed = Number.parseFloat(value.trim());
if (Number.isFinite(parsed)) {
return normalize(parsed);
}
}
return null;
}
export function normalizeCanvasImageTaskQueueConfig(
value: unknown,
options?: { envConcurrency?: unknown }
options?: { envConcurrency?: unknown; envBatchDispatchIntervalMs?: unknown }
): CanvasImageTaskQueueConfig {
const raw = value && typeof value === "object" && !Array.isArray(value)
? (value as Record<string, unknown>)
: {};
const candidates = [
const concurrencyCandidates = [
raw.concurrency,
raw.maxConcurrency,
raw.maxConcurrent,
@ -45,12 +80,51 @@ export function normalizeCanvasImageTaskQueueConfig(
options?.envConcurrency,
];
for (const candidate of candidates) {
let concurrency = DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY;
for (const candidate of concurrencyCandidates) {
const parsed = parseConcurrencyCandidate(candidate);
if (parsed !== null) {
return { concurrency: parsed };
concurrency = parsed;
break;
}
}
return { concurrency: DEFAULT_CANVAS_IMAGE_TASK_CONCURRENCY };
const intervalCandidates = [
raw.batchDispatchIntervalMs,
raw.dispatchIntervalMs,
raw.batchDispatchDelayMs,
raw.batchDispatchInterval,
raw.dispatchInterval,
options?.envBatchDispatchIntervalMs,
];
const intervalSecondsCandidates = [
raw.batchDispatchIntervalSeconds,
raw.dispatchIntervalSeconds,
];
let batchDispatchIntervalMs = DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS;
for (const candidate of intervalCandidates) {
const parsed = parseBatchDispatchIntervalCandidate(candidate);
if (parsed !== null) {
batchDispatchIntervalMs = parsed;
break;
}
}
if (batchDispatchIntervalMs === DEFAULT_CANVAS_IMAGE_BATCH_DISPATCH_INTERVAL_MS) {
for (const candidate of intervalSecondsCandidates) {
const parsed = parseBatchDispatchIntervalCandidate(candidate, {
treatAsSeconds: true,
});
if (parsed !== null) {
batchDispatchIntervalMs = parsed;
break;
}
}
}
return {
concurrency,
batchDispatchIntervalMs,
};
}

81
lib/canvas/image-task-queue.ts
View File

@ -4,6 +4,7 @@ import { and, asc, eq, lt, notInArray, sql } from "drizzle-orm";
import { db } from "@/lib/db";
import { canvasImageTask } from "@/lib/db/schema";
import {
getCanvasImageBatchDispatchIntervalMs,
getCanvasImageSingleModelConcurrency,
normalizeCanvasImageModelKey,
} from "./image-concurrency";
@ -25,6 +26,9 @@ interface QueueState {
pendingSet: Set<string>;
running: Map<string, string>;
runningByModel: Map<string, Set<string>>;
nextBatchEligibleAt: Map<string, number>;
nextBatchTimers: Map<string, ReturnType<typeof setTimeout>>;
awaitingNextBatch: Set<string>;
draining: boolean;
}
@ -42,6 +46,9 @@ function getQueueState(): QueueState {
pendingSet: new Set<string>(),
running: new Map<string, string>(),
runningByModel: new Map<string, Set<string>>(),
nextBatchEligibleAt: new Map<string, number>(),
nextBatchTimers: new Map<string, ReturnType<typeof setTimeout>>(),
awaitingNextBatch: new Set<string>(),
draining: false,
};
}
@ -84,7 +91,7 @@ function releaseRunningTask(taskId: string) {
const state = getQueueState();
const modelKey = state.running.get(taskId);
if (!modelKey) {
return;
return null;
}
state.running.delete(taskId);
@ -97,6 +104,55 @@ function releaseRunningTask(taskId: string) {
if (tasks.size === 0) {
state.runningByModel.delete(modelKey);
}
return modelKey;
}
function isModelBatchDispatchDelayed(modelKey: string) {
const state = getQueueState();
const delayedUntil = state.nextBatchEligibleAt.get(modelKey);
if (!delayedUntil) {
return false;
}
if (delayedUntil <= Date.now()) {
state.nextBatchEligibleAt.delete(modelKey);
const timer = state.nextBatchTimers.get(modelKey);
if (timer) {
clearTimeout(timer);
state.nextBatchTimers.delete(modelKey);
}
return false;
}
return true;
}
function markModelAwaitingNextBatch(modelKey: string) {
getQueueState().awaitingNextBatch.add(modelKey);
}
function scheduleModelNextBatch(modelKey: string, delayMs: number) {
const state = getQueueState();
if (delayMs <= 0) {
state.awaitingNextBatch.delete(modelKey);
return false;
}
if (state.nextBatchTimers.has(modelKey)) {
return true;
}
state.awaitingNextBatch.delete(modelKey);
state.nextBatchEligibleAt.set(modelKey, Date.now() + delayMs);
const timer = setTimeout(() => {
const latestState = getQueueState();
latestState.nextBatchTimers.delete(modelKey);
latestState.nextBatchEligibleAt.delete(modelKey);
void drainCanvasImageTaskQueue();
}, delayMs);
state.nextBatchTimers.set(modelKey, timer);
return true;
}
async function enqueueLocalTask(taskId: string, modelKey?: string) {
@ -111,6 +167,11 @@ async function enqueueLocalTask(taskId: string, modelKey?: string) {
modelKey: resolvedModelKey,
});
state.pendingSet.add(taskId);
const maxConcurrentPerModel = await getCanvasImageSingleModelConcurrency();
if (getRunningCountForModel(resolvedModelKey) >= maxConcurrentPerModel) {
markModelAwaitingNextBatch(resolvedModelKey);
}
}
function hasModelCapacity(modelKey: string, maxConcurrentPerModel: number) {
@ -126,7 +187,11 @@ async function claimNextPendingTask(
const entry = state.pending[index];
if (!entry) continue;
if (state.running.has(entry.taskId)) continue;
if (!hasModelCapacity(entry.modelKey, maxConcurrentPerModel)) continue;
if (isModelBatchDispatchDelayed(entry.modelKey)) continue;
if (!hasModelCapacity(entry.modelKey, maxConcurrentPerModel)) {
markModelAwaitingNextBatch(entry.modelKey);
continue;
}
state.pending.splice(index, 1);
state.pendingSet.delete(entry.taskId);
@ -153,7 +218,11 @@ async function claimNextPendingTask(
if (state.pendingSet.has(row.id) || state.running.has(row.id)) continue;
const modelKey = normalizeTaskModelKey(row.request);
if (isModelBatchDispatchDelayed(modelKey)) {
continue;
}
if (!hasModelCapacity(modelKey, maxConcurrentPerModel)) {
markModelAwaitingNextBatch(modelKey);
continue;
}
@ -210,7 +279,13 @@ async function runSingleTask(entry: PendingTaskEntry) {
})
.where(eq(canvasImageTask.id, entry.taskId));
} finally {
releaseRunningTask(entry.taskId);
const modelKey = releaseRunningTask(entry.taskId);
if (modelKey) {
const batchDispatchIntervalMs = await getCanvasImageBatchDispatchIntervalMs();
if (getQueueState().awaitingNextBatch.has(modelKey)) {
scheduleModelNextBatch(modelKey, batchDispatchIntervalMs);
}
}
void drainCanvasImageTaskQueue();
}
}