fix: 修复清理ExecutingMarks的bug

lib/AppLoader.d.ts

@@ -19,6 +19,7 @@ export declare class AppLoader<ED extends EntityDict & BaseEntityDict, Cxt exten
     private watcherTimerId?;
     private scheduledJobs;
     private internalErrorHandlers;
+    private watcherExecutingData;
     regAllExceptionHandler(): void;
     /**
      * 注册一个内部错误处理器
@@ -32,9 +33,15 @@ export declare class AppLoader<ED extends EntityDict & BaseEntityDict, Cxt exten
     private requireSth;
     protected makeContext(cxtStr?: string, headers?: IncomingHttpHeaders): Promise<Cxt>;
     /**
-     * 后台启动的configuration，统一放在这里读取
+     *  获取数据库配置
+     * @returns  读取数据库配置
      */
-    private getConfiguration;
+    private getDbConfig;
+    /**
+     *  获取同步配置
+     * @returns  读取同步配置
+     */
+    private getSyncConfig;
     constructor(path: string, nsSubscribe?: Namespace, nsSocket?: Namespace, nsServer?: Namespace);
     protected registerTrigger(trigger: Trigger<ED, keyof ED, Cxt>): void;
     protected initTriggers(): void;
@@ -55,6 +62,30 @@ export declare class AppLoader<ED extends EntityDict & BaseEntityDict, Cxt exten
     }>][];
     protected operateInWatcher<T extends keyof ED>(entity: T, operation: ED[T]['Update'], context: Cxt, singleton?: true): Promise<OperationResult<ED>>;
     protected selectInWatcher<T extends keyof ED>(entity: T, selection: ED[T]['Selection'], context: Cxt, forUpdate?: true, singleton?: true): Promise<Partial<ED[T]["Schema"]>[]>;
+    /**
+     *  检查某个数据是否正在被watcher执行
+     * @param name  watcher名称
+     * @param dataId  数据ID
+     * @returns  如果没有正在执行则返回true，否则返回false
+     */
+    private checkDataExecuting;
+    /**
+     * 过滤出未在执行中的数据行，并标记为执行中
+     * @returns [过滤后的行, 是否有行被跳过]
+     */
+    private filterAndMarkExecutingRows;
+    /**
+     * 清理执行标记
+     */
+    private cleanupExecutingMarks;
+    /**
+     * 解析 filter 和 projection（支持函数或静态值）
+     */
+    private resolveFilterAndProjection;
+    /**
+     * 执行 WB 类型 watcher 的查询操作
+     */
+    private selectForWBWatcher;
     protected execWatcher(watcher: Watcher<ED, keyof ED, Cxt>): Promise<OperationResult<ED> | undefined>;
     protected getCheckpointTs(): number;
     protected checkpoint(): Promise<number>;

lib/AppLoader.js

@@ -30,6 +30,7 @@ class AppLoader extends types_1.AppLoader {
     watcherTimerId;
     scheduledJobs = {};
     internalErrorHandlers = new Array();
+    watcherExecutingData = new Map();
     regAllExceptionHandler() {
         const handlers = this.requireSth('lib/configuration/exception');
         if (Array.isArray(handlers)) {
@@ -92,20 +93,24 @@ class AppLoader extends types_1.AppLoader {
         return context;
     }
     /**
-     * 后台启动的configuration，统一放在这里读取
+     *  获取数据库配置
+     * @returns  读取数据库配置
      */
-    getConfiguration() {
+    getDbConfig() {
-        const dbConfig = (0, dbPriority_1.getDbConfig)(this.path);
+        return (0, dbPriority_1.getDbConfig)(this.path);
+    }
+    /**
+     *  获取同步配置
+     * @returns  读取同步配置
+     */
+    getSyncConfig() {
         const syncConfigFile = (0, path_1.join)(this.path, 'lib', 'configuration', 'sync.js');
         const syncConfigs = (0, fs_1.existsSync)(syncConfigFile) && require(syncConfigFile).default;
-        return {
+        return syncConfigs;
-            dbConfig: dbConfig,
-            syncConfig: syncConfigs,
-        };
     }
     constructor(path, nsSubscribe, nsSocket, nsServer) {
         super(path);
-        const { dbConfig } = this.getConfiguration();
+        const dbConfig = this.getDbConfig();
         const { storageSchema } = require(`${path}/lib/oak-app-domain/Storage`);
         const depGraph = (0, dependencyBuilder_1.analyzeDepedency)(process.cwd());
         this.externalDependencies = depGraph.ascOrder;
@@ -173,7 +178,7 @@ class AppLoader extends types_1.AppLoader {
     async mount(initialize) {
         const { path } = this;
         if (!initialize) {
-            const { syncConfig: syncConfig } = this.getConfiguration();
+            const syncConfig = this.getSyncConfig();
             if (syncConfig) {
                 this.synchronizer = new Synchronizer_1.default(syncConfig, this.dbStore.getSchema(), () => this.contextBuilder(this.dbStore));
             }
@@ -368,58 +373,154 @@ class AppLoader extends types_1.AppLoader {
             forUpdate,
         });
     }
+    /**
+     *  检查某个数据是否正在被watcher执行
+     * @param name  watcher名称
+     * @param dataId  数据ID
+     * @returns  如果没有正在执行则返回true，否则返回false
+     */
+    checkDataExecuting(name, dataId) {
+        let dataSet = this.watcherExecutingData.get(name);
+        if (!dataSet) {
+            dataSet = new Map();
+            this.watcherExecutingData.set(name, dataSet);
+        }
+        if (dataSet.has(dataId)) {
+            return false;
+        }
+        dataSet.set(dataId, true);
+        return true;
+    }
+    /**
+     * 过滤出未在执行中的数据行，并标记为执行中
+     * @returns [过滤后的行, 是否有行被跳过]
+     */
+    filterAndMarkExecutingRows(watcher, rows) {
+        if (watcher.exclusive !== true) {
+            // 不需要排他执行，直接返回所有行
+            return [rows, []];
+        }
+        const rowsWithoutExecuting = [];
+        const skipedRows = [];
+        const watcherName = watcher.name;
+        for (const row of rows) {
+            if (!row.id) {
+                console.error(`实例【${process.env.OAK_INSTANCE_ID || '单机'}】执行器【${watcherName}】获取的数据没有ID，跳过此数据的并发检查处理：`, row);
+                rowsWithoutExecuting.push(row);
+                continue;
+            }
+            if (this.checkDataExecuting(watcherName, row.id)) {
+                rowsWithoutExecuting.push(row);
+            }
+            else {
+                skipedRows.push(row);
+                console.warn(`实例【${process.env.OAK_INSTANCE_ID || '单机'}】执行器【${watcherName}】将跳过正在被执行的数据ID：【${row.id}】，请检查是否执行超时`);
+            }
+        }
+        return [rowsWithoutExecuting, skipedRows];
+    }
+    /**
+     * 清理执行标记
+     */
+    cleanupExecutingMarks(watcherName, rows) {
+        for (const row of rows) {
+            if (row.id) {
+                const dataSet = this.watcherExecutingData.get(watcherName);
+                if (dataSet) {
+                    dataSet.delete(row.id);
+                }
+            }
+        }
+    }
+    /**
+     * 解析 filter 和 projection（支持函数或静态值）
+     */
+    async resolveFilterAndProjection(filter, projection) {
+        const filter2 = typeof filter === 'function' ? await filter() : (0, lodash_1.cloneDeep)(filter);
+        const projection2 = typeof projection === 'function' ? await projection() : (0, lodash_1.cloneDeep)(projection);
+        return [filter2, projection2];
+    }
+    /**
+     * 执行 WB 类型 watcher 的查询操作
+     */
+    async selectForWBWatcher(watcher, context) {
+        const { entity, projection, filter, singleton, forUpdate } = watcher;
+        const [filter2, projection2] = await this.resolveFilterAndProjection(filter, projection);
+        return await this.selectInWatcher(entity, {
+            data: projection2,
+            filter: filter2,
+        }, context, forUpdate, singleton);
+    }
     async execWatcher(watcher) {
         let result;
-        if (watcher.hasOwnProperty('type') && watcher.type === 'free') {
+        // BBWatcher：直接操作，无需查询
-            const selectContext = await this.makeContext();
+        if (watcher.hasOwnProperty('actionData')) {
-            const { entity, projection, fn, filter, singleton, forUpdate } = watcher;
+            const context = await this.makeContext();
-            const filter2 = typeof filter === 'function' ? await filter() : (0, lodash_1.cloneDeep)(filter);
+            try {
-            const projection2 = typeof projection === 'function' ? await projection() : (0, lodash_1.cloneDeep)(projection);
-            const rows = await this.selectInWatcher(entity, {
-                data: projection2,
-                filter: filter2,
-            }, selectContext, forUpdate, singleton);
-            if (rows.length > 0) {
-                result = await fn(() => this.makeContext(), rows);
-            }
-            return result;
-        }
-        const context = await this.makeContext();
-        try {
-            if (watcher.hasOwnProperty('actionData')) {
                 const { entity, action, filter, actionData, singleton } = watcher;
                 const filter2 = typeof filter === 'function' ? await filter() : (0, lodash_1.cloneDeep)(filter);
-                const data = typeof actionData === 'function' ? await (actionData)() : (0, lodash_1.cloneDeep)(actionData);
+                const data = typeof actionData === 'function' ? await actionData() : (0, lodash_1.cloneDeep)(actionData);
                 result = await this.operateInWatcher(entity, {
                     id: await (0, uuid_1.generateNewIdAsync)(),
                     action: action,
                     data,
                     filter: filter2,
                 }, context, singleton);
+                await context.commit();
+                return result;
+            }
+            catch (err) {
+                if (err instanceof types_1.OakPartialSuccess) {
+                    await context.commit();
+                }
+                else {
+                    await context.rollback();
+                }
+                throw err;
+            }
+        }
+        // WBFreeWatcher 和 WBWatcher：查询后执行
+        const isFreeType = watcher.hasOwnProperty('type') &&
+            watcher.type === 'free';
+        // 1. 执行查询（WBFreeWatcher 使用独立 context）
+        const selectContext = isFreeType ? await this.makeContext() : await this.makeContext();
+        const rows = await this.selectForWBWatcher(watcher, selectContext);
+        if (isFreeType) {
+            await selectContext.commit();
+        }
+        // 2. 并发检查：过滤出未在执行中的数据
+        const [rowsWithoutExecuting, hasSkipped] = this.filterAndMarkExecutingRows(watcher, rows);
+        if (rowsWithoutExecuting.length === 0) {
+            if (!isFreeType) {
+                await selectContext.commit();
+            }
+            this.cleanupExecutingMarks(watcher.name, hasSkipped);
+            return result;
+        }
+        // 3. 执行业务逻辑
+        try {
+            if (isFreeType) {
+                const { fn } = watcher;
+                result = await fn(() => this.makeContext(), rowsWithoutExecuting);
             }
             else {
-                const { entity, projection, fn, filter, singleton, forUpdate } = watcher;
+                const { fn } = watcher;
-                const filter2 = typeof filter === 'function' ? await filter() : (0, lodash_1.cloneDeep)(filter);
+                result = await fn(selectContext, rowsWithoutExecuting);
-                const projection2 = typeof projection === 'function' ? await projection() : (0, lodash_1.cloneDeep)(projection);
+                await selectContext.commit();
-                const rows = await this.selectInWatcher(entity, {
-                    data: projection2,
-                    filter: filter2,
-                }, context, forUpdate, singleton);
-                if (rows.length > 0) {
-                    result = await fn(context, rows);
-                }
             }
-            await context.commit();
             return result;
         }
         catch (err) {
-            if (err instanceof types_1.OakPartialSuccess) {
+            // 清理执行标记
-                await context.commit();
+            this.cleanupExecutingMarks(watcher.name, rows);
+            if (!isFreeType) {
+                if (err instanceof types_1.OakPartialSuccess) {
+                    await selectContext.commit();
+                }
+                else {
+                    await selectContext.rollback();
+                }
             }
-            else {
-                await context.rollback();
-            }
-            // 不能在这里publish，因为这个方法可能是在timer中调用，也可能是在routine中调用
             throw err;
         }
     }

lib/DbStore.d.ts

@@ -2,6 +2,7 @@ import { DbConfiguration } from 'oak-db/src/types/configuration';
 import { EntityDict, StorageSchema, Trigger, Checker, SelectFreeEntities, UpdateFreeDict, AuthDeduceRelationMap, VolatileTrigger, OperateOption } from 'oak-domain/lib/types';
 import { EntityDict as BaseEntityDict } from 'oak-domain/lib/base-app-domain';
 import { BackendRuntimeContext } from 'oak-frontend-base/lib/context/BackendRuntimeContext';
+import { DbTypeSymbol } from './utils/dbPriority';
 import { CascadeStore } from 'oak-domain/lib/store/CascadeStore';
 import { DbStore } from 'oak-db/lib/types/dbStore';
 export type TriggerStore<ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>> = {
@@ -13,4 +14,6 @@ export type TriggerStore<ED extends EntityDict & BaseEntityDict, Cxt extends Bac
     independentCheckPoint(name: string, ts: number, instanceCount?: number, instanceId?: number): Promise<number>;
 };
 export type AppDbStore<ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>> = DbStore<ED, Cxt> & CascadeStore<ED> & TriggerStore<ED, Cxt>;
-export declare const createDbStore: <ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>>(storageSchema: StorageSchema<ED>, contextBuilder: () => Cxt, dbConfiguration: DbConfiguration, authDeduceRelationMap: AuthDeduceRelationMap<ED>, selectFreeEntities?: SelectFreeEntities<ED>, updateFreeDict?: UpdateFreeDict<ED>, onVolatileTrigger?: <T extends keyof ED>(entity: T, trigger: VolatileTrigger<ED, T, Cxt>, ids: string[], cxtStr: string, option: OperateOption) => Promise<void>) => AppDbStore<ED, Cxt>;
+export declare const createDbStore: <ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>>(storageSchema: StorageSchema<ED>, contextBuilder: () => Cxt, dbConfiguration: DbConfiguration & {
+    [DbTypeSymbol]?: string;
+}, authDeduceRelationMap: AuthDeduceRelationMap<ED>, selectFreeEntities?: SelectFreeEntities<ED>, updateFreeDict?: UpdateFreeDict<ED>, onVolatileTrigger?: <T extends keyof ED>(entity: T, trigger: VolatileTrigger<ED, T, Cxt>, ids: string[], cxtStr: string, option: OperateOption) => Promise<void>) => AppDbStore<ED, Cxt>;

lib/DbStore.js

@@ -5,8 +5,10 @@ const TriggerExecutor_1 = require("oak-domain/lib/store/TriggerExecutor");
 const RelationAuth_1 = require("oak-domain/lib/store/RelationAuth");
 const dbPriority_1 = require("./utils/dbPriority");
 const createDbStore = (storageSchema, contextBuilder, dbConfiguration, authDeduceRelationMap, selectFreeEntities = [], updateFreeDict = {}, onVolatileTrigger) => {
-    const BaseStoreClass = (0, dbPriority_1.getDbStoreClass)();
+    // TODO: 这里的类型检查会过不去，因为ts不知道上层已经实现这个抽象类。
+    const BaseStoreClass = (0, dbPriority_1.getDbStoreClass)(dbConfiguration);
     // 动态创建继承类
+    // @ts-ignore
     class DynamicDbStore extends BaseStoreClass {
         executor;
         relationAuth;

lib/utils/dbPriority.d.ts

@@ -1,10 +1,10 @@
 import { MysqlStore, PostgreSQLStore } from "oak-db";
 import { DbConfiguration } from "oak-db/src/types/configuration";
-import { AsyncRowStore } from "oak-domain/lib/store/AsyncRowStore";
 import { EntityDict, StorageSchema } from 'oak-domain/lib/types';
 import { EntityDict as BaseEntityDict } from 'oak-domain/lib/base-app-domain';
 import { BackendRuntimeContext } from 'oak-frontend-base/lib/context/BackendRuntimeContext';
 import { CascadeStore } from "oak-domain/lib/store/CascadeStore";
+import { DbStore } from "oak-db/lib/types/dbStore";
 /**
  * 数据库优先级列表，按顺序尝试获取配置文件
  */
@@ -12,5 +12,10 @@ export declare const dbList: {
     mysql: typeof MysqlStore;
     postgres: typeof PostgreSQLStore;
 };
-export declare const getDbConfig: (path: string) => DbConfiguration;
+export declare const DbTypeSymbol: unique symbol;
-export declare const getDbStoreClass: <ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>>() => new (schema: StorageSchema<ED>, config: DbConfiguration) => AsyncRowStore<ED, Cxt> & CascadeStore<ED>;
+export declare const getDbConfig: (path: string) => DbConfiguration & {
+    [DbTypeSymbol]: string;
+};
+export declare const getDbStoreClass: <ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>>(config: {
+    [DbTypeSymbol]?: string;
+}) => new (schema: StorageSchema<ED>, config: DbConfiguration) => DbStore<ED, Cxt> & CascadeStore<ED>;

lib/utils/dbPriority.js

@@ -1,6 +1,6 @@
 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-exports.getDbStoreClass = exports.getDbConfig = exports.dbList = void 0;
+exports.getDbStoreClass = exports.getDbConfig = exports.DbTypeSymbol = exports.dbList = void 0;
 const oak_db_1 = require("oak-db");
 const path_1 = require("path");
 const fs_1 = require("fs");
@@ -11,7 +11,7 @@ exports.dbList = {
     mysql: oak_db_1.MysqlStore,
     postgres: oak_db_1.PostgreSQLStore
 };
-let usedDbType = null;
+exports.DbTypeSymbol = Symbol.for('oak:backend:db:type');
 const getDbConfig = (path) => {
     for (const db of Object.keys(exports.dbList)) {
         try {
@@ -25,8 +25,14 @@ const getDbConfig = (path) => {
             }
             const config = require(dbConfigFile);
             console.log(`使用${db}作为数据库`);
-            usedDbType = db;
+            // 定义不可枚举的属性，避免被序列化
-            return Object.assign({}, config);
+            Object.defineProperty(config, exports.DbTypeSymbol, {
+                value: db,
+                enumerable: false,
+                writable: false,
+                configurable: false
+            });
+            return config;
         }
         catch (err) {
             // do nothing
@@ -35,10 +41,11 @@ const getDbConfig = (path) => {
     throw new Error(`没有找到数据库配置文件，请在configuration目录下添加任一配置文件：${Object.keys(exports.dbList).map(ele => `${ele}.json`).join('、')}`);
 };
 exports.getDbConfig = getDbConfig;
-const getDbStoreClass = () => {
+const getDbStoreClass = (config) => {
-    const dbType = usedDbType || (() => {
+    const dbType = Object.getOwnPropertyDescriptor(config, exports.DbTypeSymbol)?.value;
-        throw new Error('无法确定数据库类型');
+    if (!dbType) {
-    })();
+        throw new Error('无法获取数据库类型，请确认是否存在数据库配置文件');
+    }
     const DbStoreClass = exports.dbList[dbType.toLowerCase()];
     if (!DbStoreClass) {
         throw new Error(`不支持的数据库类型：${dbType}，请确认是否存在以下配置文件：${Object.keys(exports.dbList).map(ele => `${ele}.json`).join('、')}`);

package.json

@@ -1,6 +1,6 @@
 {
     "name": "oak-backend-base",
-    "version": "4.1.27",
+    "version": "4.1.28",
     "description": "oak-backend-base",
     "main": "lib/index",
     "author": {
@@ -21,10 +21,10 @@
         "mysql": "^2.18.1",
         "mysql2": "^2.3.3",
         "node-schedule": "^2.1.0",
-        "oak-common-aspect": "^3.0.5",
+        "oak-common-aspect": "file:../oak-common-aspect",
-        "oak-db": "^3.3.13",
+        "oak-db": "file:../oak-db",
-        "oak-domain": "^5.1.34",
+        "oak-domain": "file:../oak-domain",
-        "oak-frontend-base": "^5.3.45",
+        "oak-frontend-base": "file:../oak-frontend-base",
         "socket.io": "^4.8.1",
         "socket.io-client": "^4.7.2",
         "uuid": "^8.3.2"

src/AppLoader.ts

@@ -36,6 +36,8 @@ export class AppLoader<ED extends EntityDict & BaseEntityDict, Cxt extends Backe
     private scheduledJobs: Record<string, Job> = {};
     private internalErrorHandlers = new Array<InternalErrorHandler<ED, Cxt>>();
 
+    private watcherExecutingData: Map<string, Map<string, true>> = new Map();
+
     public regAllExceptionHandler() {
         const handlers = this.requireSth('lib/configuration/exception') as Array<InternalErrorHandler<ED, Cxt>> | InternalErrorHandler<ED, Cxt>;
         if (Array.isArray(handlers)) {
@@ -105,22 +107,26 @@ export class AppLoader<ED extends EntityDict & BaseEntityDict, Cxt extends Backe
     }
 
     /**
-     * 后台启动的configuration，统一放在这里读取
+     *  获取数据库配置
+     * @returns  读取数据库配置
      */
-    private getConfiguration() {
+    private getDbConfig() {
-        const dbConfig = getDbConfig(this.path);
+        return getDbConfig(this.path);
+    }
+
+    /**
+     *  获取同步配置
+     * @returns  读取同步配置
+     */
+    private getSyncConfig() {
         const syncConfigFile = join(this.path, 'lib', 'configuration', 'sync.js');
         const syncConfigs = existsSync(syncConfigFile) && require(syncConfigFile).default;
-
+        return syncConfigs as SyncConfig<ED, Cxt> | undefined;
-        return {
-            dbConfig: dbConfig,
-            syncConfig: syncConfigs as SyncConfig<ED, Cxt> | undefined,
-        };
     }
 
     constructor(path: string, nsSubscribe?: Namespace, nsSocket?: Namespace, nsServer?: Namespace) {
         super(path);
-        const { dbConfig } = this.getConfiguration();
+        const dbConfig = this.getDbConfig();
         const { storageSchema } = require(`${path}/lib/oak-app-domain/Storage`);
         const depGraph = analyzeDepedency(process.cwd());
         this.externalDependencies = depGraph.ascOrder;
@@ -224,7 +230,7 @@ export class AppLoader<ED extends EntityDict & BaseEntityDict, Cxt extends Backe
     async mount(initialize?: true) {
         const { path } = this;
         if (!initialize) {
-            const { syncConfig: syncConfig } = this.getConfiguration();
+            const syncConfig = this.getSyncConfig();
 
             if (syncConfig) {
                 this.synchronizer = new Synchronizer(syncConfig, this.dbStore.getSchema(), () => this.contextBuilder(this.dbStore));
@@ -447,60 +453,188 @@ export class AppLoader<ED extends EntityDict & BaseEntityDict, Cxt extends Backe
         })
     }
 
+    /**
+     *  检查某个数据是否正在被watcher执行
+     * @param name  watcher名称
+     * @param dataId  数据ID
+     * @returns  如果没有正在执行则返回true，否则返回false
+     */
+    private checkDataExecuting(name: string, dataId: string): boolean {
+        let dataSet = this.watcherExecutingData.get(name);
+        if (!dataSet) {
+            dataSet = new Map<string, true>();
+            this.watcherExecutingData.set(name, dataSet);
+        }
+        if (dataSet.has(dataId)) {
+            return false;
+        }
+        dataSet.set(dataId, true);
+        return true;
+    }
+
+    /**
+     * 过滤出未在执行中的数据行，并标记为执行中
+     * @returns [过滤后的行, 是否有行被跳过]
+     */
+    private filterAndMarkExecutingRows<T extends keyof ED>(
+        watcher: Watcher<ED, T, Cxt>,
+        rows: Partial<ED[T]['Schema']>[]
+    ): [Partial<ED[T]['Schema']>[], Partial<ED[T]['Schema']>[]] {
+        if (watcher.exclusive !== true) {
+            // 不需要排他执行，直接返回所有行
+            return [rows, []];
+        }
+
+        const rowsWithoutExecuting: Partial<ED[T]['Schema']>[] = [];
+        const skipedRows: Partial<ED[T]['Schema']>[] = [];
+        const watcherName = watcher.name;
+
+        for (const row of rows) {
+            if (!row.id) {
+                console.error(`实例【${process.env.OAK_INSTANCE_ID || '单机'}】执行器【${watcherName}】获取的数据没有ID，跳过此数据的并发检查处理：`, row);
+                rowsWithoutExecuting.push(row);
+                continue;
+            }
+
+            if (this.checkDataExecuting(watcherName, row.id)) {
+                rowsWithoutExecuting.push(row);
+            } else {
+                skipedRows.push(row);
+                console.warn(`实例【${process.env.OAK_INSTANCE_ID || '单机'}】执行器【${watcherName}】将跳过正在被执行的数据ID：【${row.id}】，请检查是否执行超时`);
+            }
+        }
+
+        return [rowsWithoutExecuting, skipedRows];
+    }
+
+    /**
+     * 清理执行标记
+     */
+    private cleanupExecutingMarks<T extends keyof ED>(
+        watcherName: string,
+        rows: Partial<ED[T]['Schema']>[]
+    ) {
+        for (const row of rows) {
+            if (row.id) {
+                const dataSet = this.watcherExecutingData.get(watcherName);
+                if (dataSet) {
+                    dataSet.delete(row.id);
+                }
+            }
+        }
+    }
+
+    /**
+     * 解析 filter 和 projection（支持函数或静态值）
+     */
+    private async resolveFilterAndProjection<T extends ED[keyof ED]['Filter']>(
+        filter: T | (() => Promise<T>),
+        projection: any | (() => Promise<any>)
+    ): Promise<[T, any]> {
+        const filter2 = typeof filter === 'function' ? await filter() : cloneDeep(filter);
+        const projection2 = typeof projection === 'function' ? await projection() : cloneDeep(projection);
+        return [filter2, projection2];
+    }
+
+    /**
+     * 执行 WB 类型 watcher 的查询操作
+     */
+    private async selectForWBWatcher<T extends keyof ED>(
+        watcher: WBWatcher<ED, T, Cxt> | WBFreeWatcher<ED, T, Cxt>,
+        context: Cxt
+    ) {
+        const { entity, projection, filter, singleton, forUpdate } = watcher;
+        const [filter2, projection2] = await this.resolveFilterAndProjection(filter, projection);
+
+        return await this.selectInWatcher(entity, {
+            data: projection2,
+            filter: filter2,
+        }, context, forUpdate, singleton);
+    }
+
     protected async execWatcher(watcher: Watcher<ED, keyof ED, Cxt>) {
         let result: OperationResult<ED> | undefined;
-        if (watcher.hasOwnProperty('type') && (watcher as WBFreeWatcher<ED, keyof ED, Cxt>).type === 'free') {
-            const selectContext = await this.makeContext();
-            const { entity, projection, fn, filter, singleton, forUpdate } = <WBFreeWatcher<ED, keyof ED, Cxt>>watcher;
-            const filter2 = typeof filter === 'function' ? await filter() : cloneDeep(filter);
-            const projection2 = typeof projection === 'function' ? await projection() : cloneDeep(projection);
-            const rows = await this.selectInWatcher(entity, {
-                data: projection2,
-                filter: filter2,
-            }, selectContext, forUpdate, singleton);
 
-            if (rows.length > 0) {
+        // BBWatcher：直接操作，无需查询
-                result = await fn(() => this.makeContext(), rows);
+        if (watcher.hasOwnProperty('actionData')) {
-            }
+            const context = await this.makeContext();
-            return result;
+            try {
-        }
-        const context = await this.makeContext();
-        try {
-            if (watcher.hasOwnProperty('actionData')) {
                 const { entity, action, filter, actionData, singleton } = <BBWatcher<ED, keyof ED>>watcher;
                 const filter2 = typeof filter === 'function' ? await filter() : cloneDeep(filter);
-                const data = typeof actionData === 'function' ? await (actionData)() : cloneDeep(actionData);
+                const data = typeof actionData === 'function' ? await actionData() : cloneDeep(actionData);
+
                 result = await this.operateInWatcher(entity, {
                     id: await generateNewIdAsync(),
                     action: action as string,
                     data,
                     filter: filter2,
                 }, context, singleton);
-            }
-            else {
-                const { entity, projection, fn, filter, singleton, forUpdate } = <WBWatcher<ED, keyof ED, Cxt>>watcher;
-                const filter2 = typeof filter === 'function' ? await filter() : cloneDeep(filter);
-                const projection2 = typeof projection === 'function' ? await projection() : cloneDeep(projection);
-                const rows = await this.selectInWatcher(entity, {
-                    data: projection2,
-                    filter: filter2,
-                }, context, forUpdate, singleton);
 
-                if (rows.length > 0) {
+                await context.commit();
-                    result = await fn(context, rows);
+                return result;
+            } catch (err) {
+                if (err instanceof OakPartialSuccess) {
+                    await context.commit();
+                } else {
+                    await context.rollback();
                 }
+                throw err;
             }
-            await context.commit();
+        }
+
+        // WBFreeWatcher 和 WBWatcher：查询后执行
+        const isFreeType = watcher.hasOwnProperty('type') &&
+            (watcher as WBFreeWatcher<ED, keyof ED, Cxt>).type === 'free';
+
+        // 1. 执行查询（WBFreeWatcher 使用独立 context）
+        const selectContext = isFreeType ? await this.makeContext() : await this.makeContext();
+        const rows = await this.selectForWBWatcher(
+            watcher as WBWatcher<ED, keyof ED, Cxt> | WBFreeWatcher<ED, keyof ED, Cxt>,
+            selectContext
+        );
+
+        if (isFreeType) {
+            await selectContext.commit();
+        }
+
+        // 2. 并发检查：过滤出未在执行中的数据
+        const [rowsWithoutExecuting, hasSkipped] = this.filterAndMarkExecutingRows(
+            watcher,
+            rows
+        );
+
+        if (rowsWithoutExecuting.length === 0) {
+            if (!isFreeType) {
+                await selectContext.commit();
+            }
+
+            this.cleanupExecutingMarks(watcher.name, hasSkipped);
             return result;
         }
-        catch (err) {
+
-            if (err instanceof OakPartialSuccess) {
+        // 3. 执行业务逻辑
-                await context.commit();
+        try {
+            if (isFreeType) {
+                const { fn } = watcher as WBFreeWatcher<ED, keyof ED, Cxt>;
+                result = await fn(() => this.makeContext(), rowsWithoutExecuting);
+            } else {
+                const { fn } = watcher as WBWatcher<ED, keyof ED, Cxt>;
+                result = await fn(selectContext, rowsWithoutExecuting);
+                await selectContext.commit();
             }
-            else {
+
-                await context.rollback();
+            return result;
+        } catch (err) {
+            // 清理执行标记
+            this.cleanupExecutingMarks(watcher.name, rows);
+
+            if (!isFreeType) {
+                if (err instanceof OakPartialSuccess) {
+                    await selectContext.commit();
+                } else {
+                    await selectContext.rollback();
+                }
             }
-            // 不能在这里publish，因为这个方法可能是在timer中调用，也可能是在routine中调用
             throw err;
         }
     }

src/DbStore.ts

@@ -6,7 +6,7 @@ import { TriggerExecutor } from 'oak-domain/lib/store/TriggerExecutor';
 import { BackendRuntimeContext } from 'oak-frontend-base/lib/context/BackendRuntimeContext';
 import { AsyncContext } from 'oak-domain/lib/store/AsyncRowStore';
 import { RelationAuth } from 'oak-domain/lib/store/RelationAuth';
-import { getDbStoreClass } from './utils/dbPriority';
+import { DbTypeSymbol, getDbStoreClass } from './utils/dbPriority';
 import { CascadeStore } from 'oak-domain/lib/store/CascadeStore';
 import { DbStore } from 'oak-db/lib/types/dbStore';
 
@@ -37,16 +37,20 @@ export type AppDbStore<ED extends EntityDict & BaseEntityDict, Cxt extends Backe
 export const createDbStore = <ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>>(
     storageSchema: StorageSchema<ED>,
     contextBuilder: () => Cxt,
-    dbConfiguration: DbConfiguration,
+    dbConfiguration: DbConfiguration & {
+        [DbTypeSymbol]?: string;
+    },
     authDeduceRelationMap: AuthDeduceRelationMap<ED>,
     selectFreeEntities: SelectFreeEntities<ED> = [],
     updateFreeDict: UpdateFreeDict<ED> = {},
     onVolatileTrigger?: <T extends keyof ED>(entity: T, trigger: VolatileTrigger<ED, T, Cxt>, ids: string[], cxtStr: string, option: OperateOption) => Promise<void>
 ): AppDbStore<ED, Cxt> => {
 
-    const BaseStoreClass = getDbStoreClass<ED, Cxt>() as any
+    // TODO: 这里的类型检查会过不去，因为ts不知道上层已经实现这个抽象类。
+    const BaseStoreClass = getDbStoreClass<ED, Cxt>(dbConfiguration)
     
     // 动态创建继承类
+    // @ts-ignore
     class DynamicDbStore extends BaseStoreClass implements TriggerStore<ED, Cxt> {
         private executor: TriggerExecutor<ED, Cxt>;
         private relationAuth: RelationAuth<ED>;
@@ -208,5 +212,5 @@ export const createDbStore = <ED extends EntityDict & BaseEntityDict, Cxt extend
         }
     }
 
-    return new DynamicDbStore() as any as AppDbStore<ED, Cxt>;
+    return new DynamicDbStore() as AppDbStore<ED, Cxt>;
 };

src/utils/dbPriority.ts

@@ -2,11 +2,12 @@ import { MysqlStore, PostgreSQLStore } from "oak-db";
 import { DbConfiguration } from "oak-db/src/types/configuration";
 import { AsyncContext, AsyncRowStore } from "oak-domain/lib/store/AsyncRowStore";
 import { join } from "path";
-import { EntityDict, StorageSchema, Trigger, Checker, SelectOption, SelectFreeEntities, UpdateFreeDict, AuthDeduceRelationMap, VolatileTrigger, OperateOption } from 'oak-domain/lib/types';
+import { EntityDict, StorageSchema, Trigger, Checker, SelectOption, SelectFreeEntities, UpdateFreeDict, AuthDeduceRelationMap, VolatileTrigger, OperateOption, RowStore } from 'oak-domain/lib/types';
 import { EntityDict as BaseEntityDict } from 'oak-domain/lib/base-app-domain';
 import { BackendRuntimeContext } from 'oak-frontend-base/lib/context/BackendRuntimeContext';
 import { CascadeStore } from "oak-domain/lib/store/CascadeStore";
 import { existsSync } from "fs";
+import { DbStore } from "oak-db/lib/types/dbStore";
 
 /**
  * 数据库优先级列表，按顺序尝试获取配置文件
@@ -16,9 +17,11 @@ export const dbList = {
     postgres: PostgreSQLStore
 }
 
-let usedDbType: string | null = null;
+export const DbTypeSymbol = Symbol.for('oak:backend:db:type')
 
-export const getDbConfig = (path: string): DbConfiguration => {
+export const getDbConfig = (path: string): DbConfiguration & {
+    [DbTypeSymbol]: string;
+} => {
     for (const db of Object.keys(dbList)) {
         try {
             // eslint-disable-next-line @typescript-eslint/no-var-requires
@@ -33,8 +36,14 @@ export const getDbConfig = (path: string): DbConfiguration => {
 
             const config = require(dbConfigFile);
             console.log(`使用${db}作为数据库`);
-            usedDbType = db;
+            // 定义不可枚举的属性，避免被序列化
-            return Object.assign({}, config);
+            Object.defineProperty(config, DbTypeSymbol, {
+                value: db,
+                enumerable: false,
+                writable: false,
+                configurable: false
+            });
+            return config;
         }
         catch (err) {
             // do nothing
@@ -45,13 +54,16 @@ export const getDbConfig = (path: string): DbConfiguration => {
 }
 
 export const getDbStoreClass = <ED extends EntityDict & BaseEntityDict, Cxt extends BackendRuntimeContext<ED>>
-    () => {
+    (config: {
-    const dbType = usedDbType || (() => {
+        [DbTypeSymbol]?: string;
-        throw new Error('无法确定数据库类型');
+    }) => {
-    })();
+    const dbType = Object.getOwnPropertyDescriptor(config, DbTypeSymbol)?.value;
+    if (!dbType) {
+        throw new Error('无法获取数据库类型，请确认是否存在数据库配置文件');
+    }
     const DbStoreClass = dbList[dbType.toLowerCase() as keyof typeof dbList];
     if (!DbStoreClass) {
         throw new Error(`不支持的数据库类型：${dbType}，请确认是否存在以下配置文件：${Object.keys(dbList).map(ele => `${ele}.json`).join('、')}`);
     }
-    return DbStoreClass as new (schema: StorageSchema<ED>, config: DbConfiguration) => AsyncRowStore<ED, Cxt> & CascadeStore<ED>;
+    return DbStoreClass as new (schema: StorageSchema<ED>, config: DbConfiguration) => DbStore<ED, Cxt> & CascadeStore<ED>;
 }