Oak-Team
/
oak-backend-base
13
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 18 Wiki Activity

clusterApploader原来处理commitTriggers的逻辑是错误的，漏掉了非cs类型的

This commit is contained in:
Xu Chang 2025-04-18 16:14:11 +08:00
parent 51c93a9402
commit b60bddd66e
3 changed files with 29 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
lib/ClusterAppLoader.d.ts vendored
View File

@ -5,7 +5,7 @@ import { AppLoader } from './AppLoader';
import { Namespace } from 'socket.io'; import { Namespace } from 'socket.io';
import { Socket } from 'socket.io-client'; import { Socket } from 'socket.io-client';
export declare class ClusterAppLoader<ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>> extends AppLoader<ED, Cxt> { export declare class ClusterAppLoader<ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>> extends AppLoader<ED, Cxt> {
private csTriggers; private commitTriggers;
static VolatileTriggerEvent: string; static VolatileTriggerEvent: string;
protected socket: Socket; protected socket: Socket;
private connectServerSocket; private connectServerSocket;

25
lib/ClusterAppLoader.js
View File

@ -10,13 +10,13 @@ const AppLoader_1 = require("./AppLoader");
const assert_1 = tslib_1.__importDefault(require("assert")); const assert_1 = tslib_1.__importDefault(require("assert"));
const socket_io_client_1 = require("socket.io-client"); const socket_io_client_1 = require("socket.io-client");
class ClusterAppLoader extends AppLoader_1.AppLoader { class ClusterAppLoader extends AppLoader_1.AppLoader {
csTriggers; commitTriggers;
static VolatileTriggerEvent = 'vtEvent'; static VolatileTriggerEvent = 'vtEvent';
socket; socket;
connectServerSocket() { connectServerSocket() {
const { instanceId } = (0, env_1.getClusterInfo)(); const { instanceId } = (0, env_1.getClusterInfo)();
this.socket.on('connect', () => { this.socket.on('connect', () => {
const csTriggerNames = Object.keys(this.csTriggers).map(ele => `${ele}-${instanceId}`); const csTriggerNames = Object.keys(this.commitTriggers).filter(ele => this.commitTriggers[ele]).map(ele => `${ele}-${instanceId}`);
if (csTriggerNames.length > 0) { if (csTriggerNames.length > 0) {
this.socket.emit('sub', csTriggerNames); this.socket.emit('sub', csTriggerNames);
} }
@ -37,10 +37,8 @@ class ClusterAppLoader extends AppLoader_1.AppLoader {
} }
}); });
} }
sub(name, clusterSensative) { sub(name) {
const { instanceId } = (0, env_1.getClusterInfo)(); const { instanceId } = (0, env_1.getClusterInfo)();
(0, assert_1.default)(!this.csTriggers[name], `命名为${name}的trigger出现了多次，请检查`);
this.csTriggers[name] = !!clusterSensative;
if (this.socket.connected) { if (this.socket.connected) {
this.socket.emit('sub', [`${name}-${instanceId}`]); this.socket.emit('sub', [`${name}-${instanceId}`]);
} }
@ -112,7 +110,7 @@ class ClusterAppLoader extends AppLoader_1.AppLoader {
await execLocal(ids); await execLocal(ids);
} }
}); });
this.csTriggers = {}; this.commitTriggers = {};
const { name } = nsServer; const { name } = nsServer;
// 本机pm2的socketio连接，在cli中连接到adaptor之后，会被自然推到redis，这边继续保持pm2的socketio连接即可 // 本机pm2的socketio连接，在cli中连接到adaptor之后，会被自然推到redis，这边继续保持pm2的socketio连接即可
const socketUrl = `http://localhost:${process.env.PM2_PORT || 8080}${name}`; const socketUrl = `http://localhost:${process.env.PM2_PORT || 8080}${name}`;
@ -124,11 +122,14 @@ class ClusterAppLoader extends AppLoader_1.AppLoader {
registerTrigger(trigger) { registerTrigger(trigger) {
// 如果是cluster sensative的trigger，注册到socket事件上 // 如果是cluster sensative的trigger，注册到socket事件上
// 如果是singletone的trigger，只有0号实例注册 // 如果是singletone的trigger，只有0号实例注册
const { when, cs, singleton } = trigger; const { when, cs, singleton, name } = trigger;
(0, assert_1.default)(!(cs && singleton)); (0, assert_1.default)(!(cs && singleton));
if (when === 'commit' && (cs || singleton && (0, env_1.getClusterInfo)().instanceId === 0)) { if (when === 'commit') {
const { name } = trigger; (0, assert_1.default)(!this.commitTriggers[name], `命名为${name}的trigger出现了多次，请检查`);
this.sub(name, cs); this.commitTriggers[name] = !!cs;
if (cs || singleton && (0, env_1.getClusterInfo)().instanceId === 0) {
this.sub(name);
}
} }
this.dbStore.registerTrigger(trigger); this.dbStore.registerTrigger(trigger);
} }
@ -175,8 +176,8 @@ class ClusterAppLoader extends AppLoader_1.AppLoader {
async checkpoint() { async checkpoint() {
const { instanceCount, instanceId } = (0, env_1.getClusterInfo)(); const { instanceCount, instanceId } = (0, env_1.getClusterInfo)();
let count = 0; let count = 0;
for (const name in this.csTriggers) { for (const name in this.commitTriggers) {
if (this.csTriggers[name]) { if (this.commitTriggers[name]) {
count += await this.dbStore.independentCheckPoint(name, this.getCheckpointTs(), instanceCount, instanceId); count += await this.dbStore.independentCheckPoint(name, this.getCheckpointTs(), instanceCount, instanceId);
} }
else { else {

27
src/ClusterAppLoader.ts
View File

@ -11,14 +11,16 @@ import { Namespace } from 'socket.io';
import { io, Socket } from 'socket.io-client'; import { io, Socket } from 'socket.io-client';
export class ClusterAppLoader<ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>> extends AppLoader<ED, Cxt> { export class ClusterAppLoader<ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>> extends AppLoader<ED, Cxt> {
private csTriggers: Record<string, boolean>; private commitTriggers: Record<string, boolean>;
static VolatileTriggerEvent = 'vtEvent'; static VolatileTriggerEvent = 'vtEvent';
protected socket: Socket; protected socket: Socket;
private connectServerSocket() { private connectServerSocket() {
const { instanceId } = getClusterInfo()!; const { instanceId } = getClusterInfo()!;
this.socket.on('connect', () => { this.socket.on('connect', () => {
const csTriggerNames = Object.keys(this.csTriggers).map( const csTriggerNames = Object.keys(this.commitTriggers).filter(
ele => this.commitTriggers[ele]
).map(
ele => `${ele}-${instanceId}` ele => `${ele}-${instanceId}`
); );
if (csTriggerNames.length > 0) { if (csTriggerNames.length > 0) {
@ -42,10 +44,8 @@ export class ClusterAppLoader<ED extends EntityDict & BaseEntityDict, Cxt extend
}); });
} }
private sub(name: string, clusterSensative?: true) { private sub(name: string) {
const { instanceId } = getClusterInfo()!; const { instanceId } = getClusterInfo()!;
assert(!this.csTriggers[name], `命名为${name}的trigger出现了多次，请检查`);
this.csTriggers[name] = !!clusterSensative;
if (this.socket!.connected) { if (this.socket!.connected) {
this.socket!.emit('sub', [`${name}-${instanceId}`]); this.socket!.emit('sub', [`${name}-${instanceId}`]);
} }
@ -121,7 +121,7 @@ export class ClusterAppLoader<ED extends EntityDict & BaseEntityDict, Cxt extend
} }
} }
); );
this.csTriggers = {}; this.commitTriggers = {};
const { name } = nsServer; const { name } = nsServer;
// 本机pm2的socketio连接，在cli中连接到adaptor之后，会被自然推到redis，这边继续保持pm2的socketio连接即可 // 本机pm2的socketio连接，在cli中连接到adaptor之后，会被自然推到redis，这边继续保持pm2的socketio连接即可
const socketUrl = `http://localhost:${process.env.PM2_PORT || 8080}${name}`; const socketUrl = `http://localhost:${process.env.PM2_PORT || 8080}${name}`;
@ -134,11 +134,14 @@ export class ClusterAppLoader<ED extends EntityDict & BaseEntityDict, Cxt extend
protected registerTrigger(trigger: Trigger<ED, keyof ED, Cxt>): void { protected registerTrigger(trigger: Trigger<ED, keyof ED, Cxt>): void {
// 如果是cluster sensative的trigger，注册到socket事件上 // 如果是cluster sensative的trigger，注册到socket事件上
// 如果是singletone的trigger，只有0号实例注册 // 如果是singletone的trigger，只有0号实例注册
const { when, cs, singleton } = trigger as VolatileTrigger<ED, keyof ED, Cxt>; const { when, cs, singleton, name } = trigger as VolatileTrigger<ED, keyof ED, Cxt>;
assert(!(cs && singleton)); assert(!(cs && singleton));
if (when === 'commit' && (cs || singleton && getClusterInfo().instanceId === 0)) { if (when === 'commit') {
const { name } = trigger; assert(!this.commitTriggers[name], `命名为${name}的trigger出现了多次，请检查`);
this.sub(name, cs); this.commitTriggers[name] = !!cs;
if (cs || singleton && getClusterInfo().instanceId === 0) {
this.sub(name);
}
} }
this.dbStore.registerTrigger(trigger); this.dbStore.registerTrigger(trigger);
} }
@ -190,8 +193,8 @@ export class ClusterAppLoader<ED extends EntityDict & BaseEntityDict, Cxt extend
protected async checkpoint() { protected async checkpoint() {
const { instanceCount, instanceId } = getClusterInfo(); const { instanceCount, instanceId } = getClusterInfo();
let count = 0; let count = 0;
for (const name in this.csTriggers) { for (const name in this.commitTriggers) {
if (this.csTriggers[name]) { if (this.commitTriggers[name]) {
count += await this.dbStore.independentCheckPoint(name, this.getCheckpointTs(), instanceCount, instanceId); count += await this.dbStore.independentCheckPoint(name, this.getCheckpointTs(), instanceCount, instanceId);
} }
else { else {