Oak-Team
/
oak-domain
13
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 33 Wiki Activity

合并了checkpoint逻辑，统一使用按commitTrigger的名称逐个处理

This commit is contained in:
Xu Chang 2024-12-25 17:19:16 +08:00
parent f4af94a98a
commit 82b6f95318
2 changed files with 4 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
lib/store/TriggerExecutor.js
View File

@ -485,68 +485,8 @@ class TriggerExecutor {
}
async checkpoint(timestamp) {
let result = 0;
for (const entity of this.volatileEntities) {
const filter = {
[Entity_1.TriggerUuidAttribute]: {
$exists: true,
},
[Entity_1.UpdateAtAttribute]: {
$lt: timestamp,
}
};
const context = this.contextBuilder();
await context.begin();
try {
const rows = await context.select(entity, {
data: {
id: 1,
},
filter,
}, {
includedDeleted: true,
dontCollect: true,
});
if (rows.length > 0) {
// 要用id来再锁一次，不然会锁住filter的范围，影响并发性
// by Xc 20240314，在haina-busi和haina-cn数据sync过程中发现这个问题
const rows2 = await context.select(entity, {
data: {
id: 1,
[Entity_1.TriggerDataAttribute]: 1,
[Entity_1.TriggerUuidAttribute]: 1,
},
filter: {
id: {
$in: rows.map(ele => ele.id),
},
},
}, {
includedDeleted: true,
dontCollect: true,
forUpdate: 'skip locked', // 如果加不上锁就下次再处理，或者有可能应用自己在处理
});
const grouped = (0, lodash_1.groupBy)(rows2, Entity_1.TriggerUuidAttribute);
for (const uuid in grouped) {
const rs = grouped[uuid];
const { [Entity_1.TriggerDataAttribute]: triggerData } = rs[0];
const { name, cxtStr, option } = triggerData;
await context.initialize(JSON.parse(cxtStr), true);
await this.execVolatileTrigger(entity, name, rs.map(ele => ele.id), context, option);
}
}
await context.commit();
result += rows.length;
}
catch (err) {
if (!(err instanceof types_1.OakPartialSuccess)) {
await context.rollback();
this.logger.error(`error in checkpoint on entity 「${entity}」`, err);
}
else {
await context.commit();
this.logger.error(`error in checkpoint on entity 「${entity}」`, err);
}
}
for (const name in this.triggerNameMap) {
result += await this.independentCheckPoint(name, timestamp);
}
return result;
}

68
src/store/TriggerExecutor.ts
View File

@ -597,72 +597,8 @@ export class TriggerExecutor<ED extends EntityDict & BaseEntityDict, Cxt extends
async checkpoint(timestamp: number): Promise<number> {
let result = 0;
for (const entity of this.volatileEntities) {
const filter: ED[keyof ED]['Selection']['filter'] = {
[TriggerUuidAttribute]: {
$exists: true,
},
[UpdateAtAttribute]: {
$lt: timestamp,
}
};
const context = this.contextBuilder();
await context.begin();
try {
const rows = await context.select(entity, {
data: {
id: 1,
},
filter,
}, {
includedDeleted: true,
dontCollect: true,
});
if (rows.length > 0) {
// 要用id来再锁一次，不然会锁住filter的范围，影响并发性
// by Xc 20240314，在haina-busi和haina-cn数据sync过程中发现这个问题
const rows2 = await context.select(entity, {
data: {
id: 1,
[TriggerDataAttribute]: 1,
[TriggerUuidAttribute]: 1,
},
filter: {
id: {
$in: rows.map(ele => ele.id!),
},
},
}, {
includedDeleted: true,
dontCollect: true,
forUpdate: 'skip locked', // 如果加不上锁就下次再处理，或者有可能应用自己在处理
});
const grouped = groupBy(rows2, TriggerUuidAttribute);
for (const uuid in grouped) {
const rs = grouped[uuid];
const { [TriggerDataAttribute]: triggerData } = rs[0];
const { name, cxtStr, option } = triggerData!;
await context.initialize(JSON.parse(cxtStr), true);
await this.execVolatileTrigger(entity, name, rs.map(ele => ele.id!), context, option);
}
}
await context.commit();
result += rows.length;
}
catch (err) {
if (!(err instanceof OakPartialSuccess)) {
await context.rollback();
this.logger.error(`error in checkpoint on entity 「${entity as string}」`, err);
}
else {
await context.commit();
this.logger.error(`error in checkpoint on entity 「${entity as string}」`, err);
}
}
for (const name in this.triggerNameMap) {
result += await this.independentCheckPoint(name,timestamp);
}
return result;
}