BUG
This commit is contained in:
parent
3c9625add6
commit
99fb1876f5
1
.gitignore
vendored
1
.gitignore
vendored
@ -7,3 +7,4 @@ package-lock.json
|
|||||||
ecosystem.config.*.json
|
ecosystem.config.*.json
|
||||||
docker-compose.*.yml
|
docker-compose.*.yml
|
||||||
data_nedb/
|
data_nedb/
|
||||||
|
yarn.lock
|
||||||
|
|||||||
@ -190,14 +190,14 @@ export class GenerationTask {
|
|||||||
static async getTimeoutTasks(): Promise<IGenerationTask[]> {
|
static async getTimeoutTasks(): Promise<IGenerationTask[]> {
|
||||||
const currentTime = Math.floor(Date.now() / 1000);
|
const currentTime = Math.floor(Date.now() / 1000);
|
||||||
|
|
||||||
return await this.find({
|
// NeDB 不支持 $expr,先查询所有活跃任务,然后在内存中过滤超时任务
|
||||||
status: { $in: ['processing', 'polling'] },
|
const activeTasks = await this.find({
|
||||||
$expr: {
|
status: { $in: ['processing', 'polling'] }
|
||||||
$lt: [
|
});
|
||||||
{ $add: ['$created_at', '$task_timeout'] },
|
|
||||||
currentTime
|
return activeTasks.filter(task => {
|
||||||
]
|
const timeoutAt = task.created_at + task.task_timeout;
|
||||||
}
|
return timeoutAt < currentTime;
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -273,10 +273,11 @@ export class DatabaseGenerationService {
|
|||||||
*/
|
*/
|
||||||
async getServerLoad(serverId: string): Promise<number> {
|
async getServerLoad(serverId: string): Promise<number> {
|
||||||
try {
|
try {
|
||||||
return await GenerationTask.countDocuments({
|
const tasks = await GenerationTask.find({
|
||||||
server_id: serverId,
|
server_id: serverId,
|
||||||
status: { $in: ['processing', 'polling'] }
|
status: { $in: ['processing', 'polling'] }
|
||||||
});
|
});
|
||||||
|
return tasks.length;
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
logger.error(`Failed to get server load for ${serverId}:`, error);
|
logger.error(`Failed to get server load for ${serverId}:`, error);
|
||||||
return 0;
|
return 0;
|
||||||
@ -301,17 +302,19 @@ export class DatabaseGenerationService {
|
|||||||
async cancelTask(taskId: string): Promise<boolean> {
|
async cancelTask(taskId: string): Promise<boolean> {
|
||||||
try {
|
try {
|
||||||
const result = await GenerationTask.updateOne(
|
const result = await GenerationTask.updateOne(
|
||||||
{
|
{
|
||||||
task_id: taskId,
|
task_id: taskId,
|
||||||
status: { $in: ['pending', 'processing', 'polling'] }
|
status: { $in: ['pending', 'processing', 'polling'] }
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
|
$set: {
|
||||||
status: 'failed',
|
status: 'failed',
|
||||||
error_message: 'Task cancelled by user',
|
error_message: 'Task cancelled by user',
|
||||||
completed_at: Math.floor(Date.now() / 1000),
|
completed_at: Math.floor(Date.now() / 1000),
|
||||||
updated_at: Math.floor(Date.now() / 1000)
|
updated_at: Math.floor(Date.now() / 1000)
|
||||||
}
|
}
|
||||||
);
|
}
|
||||||
|
);
|
||||||
|
|
||||||
const cancelled = result > 0;
|
const cancelled = result > 0;
|
||||||
|
|
||||||
|
|||||||
@ -211,9 +211,11 @@ export class TaskPollingService {
|
|||||||
const updateResult = await GenerationTask.updateOne(
|
const updateResult = await GenerationTask.updateOne(
|
||||||
{ task_id: task.task_id, status: 'pending' },
|
{ task_id: task.task_id, status: 'pending' },
|
||||||
{
|
{
|
||||||
status: 'processing',
|
$set: {
|
||||||
started_at: currentTime,
|
status: 'processing',
|
||||||
updated_at: currentTime
|
started_at: currentTime,
|
||||||
|
updated_at: currentTime
|
||||||
|
}
|
||||||
}
|
}
|
||||||
);
|
);
|
||||||
|
|
||||||
@ -236,10 +238,12 @@ export class TaskPollingService {
|
|||||||
await GenerationTask.updateOne(
|
await GenerationTask.updateOne(
|
||||||
{ task_id: task.task_id },
|
{ task_id: task.task_id },
|
||||||
{
|
{
|
||||||
status: 'polling',
|
$set: {
|
||||||
'internal_params.history_id': historyId,
|
status: 'polling',
|
||||||
next_poll_at: currentTime + task.poll_interval,
|
'internal_params.history_id': historyId,
|
||||||
updated_at: currentTime
|
next_poll_at: currentTime + task.poll_interval,
|
||||||
|
updated_at: currentTime
|
||||||
|
}
|
||||||
}
|
}
|
||||||
);
|
);
|
||||||
|
|
||||||
@ -279,8 +283,10 @@ export class TaskPollingService {
|
|||||||
await GenerationTask.updateOne(
|
await GenerationTask.updateOne(
|
||||||
{ task_id: task.task_id },
|
{ task_id: task.task_id },
|
||||||
{
|
{
|
||||||
next_poll_at: currentTime + task.poll_interval,
|
$set: {
|
||||||
updated_at: currentTime
|
next_poll_at: currentTime + task.poll_interval,
|
||||||
|
updated_at: currentTime
|
||||||
|
}
|
||||||
}
|
}
|
||||||
);
|
);
|
||||||
return;
|
return;
|
||||||
@ -293,8 +299,10 @@ export class TaskPollingService {
|
|||||||
await GenerationTask.updateOne(
|
await GenerationTask.updateOne(
|
||||||
{ task_id: task.task_id },
|
{ task_id: task.task_id },
|
||||||
{
|
{
|
||||||
next_poll_at: currentTime + task.poll_interval,
|
$set: {
|
||||||
updated_at: currentTime
|
next_poll_at: currentTime + task.poll_interval,
|
||||||
|
updated_at: currentTime
|
||||||
|
}
|
||||||
}
|
}
|
||||||
);
|
);
|
||||||
} else if (status === 10 || (status !== 30 && item_list && item_list.length > 0)) {
|
} else if (status === 10 || (status !== 30 && item_list && item_list.length > 0)) {
|
||||||
@ -368,9 +376,11 @@ export class TaskPollingService {
|
|||||||
await GenerationTask.updateOne(
|
await GenerationTask.updateOne(
|
||||||
{ task_id: task.task_id },
|
{ task_id: task.task_id },
|
||||||
{
|
{
|
||||||
status: 'completed',
|
$set: {
|
||||||
completed_at: currentTime,
|
status: 'completed',
|
||||||
updated_at: currentTime
|
completed_at: currentTime,
|
||||||
|
updated_at: currentTime
|
||||||
|
}
|
||||||
}
|
}
|
||||||
);
|
);
|
||||||
|
|
||||||
@ -414,11 +424,13 @@ export class TaskPollingService {
|
|||||||
await GenerationTask.updateOne(
|
await GenerationTask.updateOne(
|
||||||
{ task_id: task.task_id },
|
{ task_id: task.task_id },
|
||||||
{
|
{
|
||||||
status: 'failed',
|
$set: {
|
||||||
error_message: errorMessage,
|
status: 'failed',
|
||||||
fail_code: failCode,
|
error_message: errorMessage,
|
||||||
completed_at: now,
|
fail_code: failCode,
|
||||||
updated_at: now
|
completed_at: now,
|
||||||
|
updated_at: now
|
||||||
|
}
|
||||||
}
|
}
|
||||||
);
|
);
|
||||||
|
|
||||||
@ -444,9 +456,11 @@ export class TaskPollingService {
|
|||||||
await GenerationTask.updateOne(
|
await GenerationTask.updateOne(
|
||||||
{ task_id: task.task_id },
|
{ task_id: task.task_id },
|
||||||
{
|
{
|
||||||
retry_count: newRetryCount,
|
$set: {
|
||||||
next_poll_at: currentTime + nextPollDelay,
|
retry_count: newRetryCount,
|
||||||
updated_at: currentTime
|
next_poll_at: currentTime + nextPollDelay,
|
||||||
|
updated_at: currentTime
|
||||||
|
}
|
||||||
}
|
}
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
@ -883,15 +897,15 @@ export class DatabaseCleanupService {
|
|||||||
static async cleanupTimeoutTasks(): Promise<number> {
|
static async cleanupTimeoutTasks(): Promise<number> {
|
||||||
const currentTime = Math.floor(Date.now() / 1000);
|
const currentTime = Math.floor(Date.now() / 1000);
|
||||||
|
|
||||||
// 查找超时的任务并标记为失败
|
// 查找所有正在处理或轮询的任务
|
||||||
const timeoutTasks = await GenerationTask.find({
|
const activeTasks = await GenerationTask.find({
|
||||||
status: { $in: ['processing', 'polling'] },
|
status: { $in: ['processing', 'polling'] }
|
||||||
$expr: {
|
});
|
||||||
$gt: [
|
|
||||||
{ $subtract: [currentTime, '$started_at'] },
|
// 在内存中过滤超时任务(NeDB 不支持 $expr)
|
||||||
'$task_timeout'
|
const timeoutTasks = activeTasks.filter(task => {
|
||||||
]
|
const elapsedTime = currentTime - (task.started_at || task.created_at);
|
||||||
}
|
return elapsedTime > (task.task_timeout || 3600); // 默认1小时超时
|
||||||
});
|
});
|
||||||
|
|
||||||
// 批量更新超时任务为失败状态
|
// 批量更新超时任务为失败状态
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user