Teable 计算活动Computed Activity架构解析字段/表计算元数据的运行时投影、实时同步与客户端集成【免费下载链接】teable✨ AI Spreadsheet for Business项目地址: https://gitcode.com/GitHub_Trending/te/teable本指南以 packages/v2/adapter-table-repository-postgres/src/record/computed/activity/ARCHITECTURE.md 为核心骨架系统讲解 Teablev2 架构中 Computed Activity计算活动子系统的设计目标、领域模型、投影存储、生命周期钩子、实时同步管线、查询 API 与前端集成方式。读完本文你将掌握计算活动如何在不污染字段 schemameta、不重载is_pending的前提下为异步公式/查找/汇总formula/lookup/rollup计算提供字段级 calculating 与表级 N formulas calculating / just completed duration 的可观测体验以及该项目是如何把该机制落到 PostgreSQL 投影表与 ShareDB 实时文档上的。一、什么是 Computed Activity解决什么问题在 Teable 的 v2 计算链路中公式formula、查找lookup、汇总rollup等字段的计算由异步的 computed outbox 队列驱动见 ComputedUpdateOutbox.ts。计算任务在执行过程中存在明确的生命周期入队enqueue→ 领取claim→ 执行 → 完成/失败/重试。用户在界面上需要感知这些状态才能获得类似飞书Feishu的交互体验字段正在计算field calculating表级提示N 个公式正在计算 / 刚刚完成 耗时可扩展的复杂度、规模与诊断信息complexity / scale / diagnostics。该文档明确给出了两个不这样做的约束这是理解整个子系统设计动机的关键不把状态塞进字段 schema 的meta中schemameta属于字段定义的一部分混入高频变化的运行时状态会污染持久化模型、引发同步风暴并耦合 schema 演进不重载is_pending标志is_pending语义过于单一无法表达 queued/running/failed 等多种状态与耗时、复杂度等附加信息。因此项目选择了一条独立的 runtime compute metadata 路线计算状态是运行时投影runtime projection与字段 schema 解耦由专门的投影存储与领域聚合来维护。二、领域模型核心包中的纯聚合Computed Activity 的领域逻辑位于核心包teable/v2-core目录 packages/v2/core/src/domain/computed包含三类核心构件领域构件职责FieldComputeMeta/TableComputeMeta/ComputeStatus字段级、表级计算元数据的纯值聚合与状态机ComputedActivity面向状态迁移的纯聚合工作区in-memory aggregateComputedActivityBatchChanged领域事件触发实时realtime下游投影2.1 ComputeStatus两级状态机文件 ComputeStatus.ts 定义了两种正交的状态集合字段级FIELD_COMPUTE_STATUSES [idle, queued, running, failed]。状态由活跃任务计数与处理中任务计数推导FieldComputeStatus.fromActive在activeTaskCount 0时返回failed若标记失败或idle在processingTaskCount 0时返回running否则返回queued。表级TABLE_COMPUTE_STATUSES [idle, calculating]。TableComputeStatus.fromActiveFieldCount在活跃字段数大于 0 时返回calculating。这种用计数推导状态的设计非常关键它天然支持多任务并发归属同一个字段的场景例如同一字段被多个批次任务同时计算只要维护好activeTaskCount/processingTaskCount两个计数器状态机就不会出现脏状态。2.2 FieldComputeMeta字段级投影FieldComputeMeta.ts 以 zod schemafieldComputeMetaSchema约束其 DTO字段包括身份与归属fieldId、tableId、baseId状态与计数status、activeTaskCount、processingTaskCount、generation预估信息estimatedComplexity预估复杂度、estimatedDirtyRecords预估脏记录数、hasAllTargetRecords是否全表目标记录时间戳queuedAt、startedAt、updatedAt、lastCompletedAt、lastDurationMs最近一次耗时毫秒错误与扩展lastError{ code?, message }、extensions任意扩展对象例如批次进度batchProgress。其核心迁移方法对应任务生命周期attachTask入队递增activeTaskCount记录queuedAt更新复杂度/脏记录为最大值通过batchProgress{ groupId, total, completed }跟踪批次进度然后recomputeStatusmarkProcessing开始处理仅当processingTaskCount activeTaskCount时递增首次记录startedAtreconcileProcessing重试/领取时对账直接用持久化的processingTaskCount覆盖内存计数支持注入lastError保证与持久化真相一致releaseTask完成/失败释放递减计数记录lastDurationMs与lastCompletedAt任务全部释放后清空预估信息与时间戳并根据是否传入了 error 决定最终状态是否为failed。源码细节每个迁移方法末尾都会调用recomputeStatus(now)该方法基于计数调用FieldComputeStatus.fromActive重算状态并每次generation 1。generation 是投影版本号直接用于下游实时文档版本推导见第五节。2.3 TableComputeMeta表级汇总TableComputeMeta.ts 维护表级摘要statusidle/calculating、calculatingFieldCount、queuedFieldCountestimatedComplexity取活跃字段中最大复杂度recentCompletions有界的最近完成列表默认上限DEFAULT_RECENT_LIMIT 20pushCompletion采用头插 slice(0, limit)的方式保留最近完成记录每条记录含fieldId、durationMs、completedAt这正是表级just completed duration提示的数据来源generation、updatedAt以及固定值computeMode: server当前计算模式为服务端计算。表级状态不单独维护计数而是由recomputeFromFields(fields, now)从字段级投影重算遍历该表所有字段的 DTO统计running与queued数量取最大复杂度并推进自己的generation。这种字段驱动表的派生方式保证了单一事实来源字段级与汇总一致。2.4 ComputedActivity纯聚合工作区ComputedActivity.ts 是上述两个聚合的内存工作区持久化适配器负责把快照加载进来、把迁移结果保存回去领域逻辑状态迁移、表级重算全部集中在内存中。关键方法fromSnapshot/snapshot()与 DTO 数组双向转换attachTask/markProcessing/reconcileProcessing/releaseTask批量迁移字段级状态随后调用私有recomputeTables对受影响表执行table.recomputeFromFieldsreleaseTask在成功完成durationMs ! null durationMs 0 !error时还会table.pushCompletion写入最近完成记录。三、投影存储三张 PostgreSQL 表文档给出的投影存储Projection store角色如下表角色computed_field_activity字段级状态、引用计数refcount、复杂度、最近耗时computed_table_activity表级汇总 最近完成记录computed_task_field_ref任务→字段集合用于幂等引用计数这三个表的表名常量定义于 ComputedActivityProjector.ts 顶部FIELD_ACTIVITY_TABLE、TABLE_ACTIVITY_TABLE、TASK_FIELD_REF_TABLE。从ensureActivityRows的插入语句可见两个投影表的初始形态字段表初始为statusidle、active_task_count0、processing_task_count0、generation0表级初始为statusidle、calculating_field_count0、queued_field_count0、recent_completionsJSON.stringify([])。onConflict doNothing保证了行只被初始化一次。computed_task_field_ref表是幂等性的基石每条记录以(task_id, field_id)为唯一约束插入时onConflict(...).doNothing()并带was_processing标志是否已进入处理阶段。只要任务与字段的引用关系持久化在案无论任务是被领取、重试还是重新投递引用计数都能从持久化真相中重建从而让 claim/retry 等过渡状态做到幂等。四、生命周期钩子与 ComputedUpdateOutbox 的事务集成ComputedActivityProjector由 ComputedUpdateOutbox.ts 在与 outbox 变更相同的数据库事务内调用源码中可见this.activityProjector.onTaskEnqueued(...)、onTasksClaimed(...)、onTaskDone(...)、onTaskFailed(...)等调用点且ComputedUpdateOutbox构造时默认注入noopComputedActivityProjector以保持可测试性与向后兼容。文档给出的 outbox → activity 钩子映射表Outbox 阶段Activity 行为enqueuecreate/mergeonTaskEnqueued→ 附加引用refs状态queuedclaimonTasksClaimed→ 状态runningseed plan 之后claim 后附加新发现的 refs状态保持runningmarkDoneonTaskDone→ 释放 refs记录lastDurationMs状态idlemarkFailed终态onTaskFailed(terminal)→ 释放 状态failedmarkFailed重试/ releaseForRetryonTaskFailed(!terminal)→ 清除处理中标记一个值得注意的细节claimed 的 seed 任务此时还不知道自己的计算目标。任务被领取后worker 需要先做规划planning再把发现的目标字段注册到投影中——文档原文为the worker registers those targets in a short projection transaction after planning and before execution即规划与执行之间有一个独立的短事务用于注册目标。4.1 并发控制advisory lock 行锁从ComputedActivityProjector源码可以看到两层串行化表级 advisory locklockTouchedTables对每个受影响表执行buildAdvisoryLockQuery(trx, v2:computed-activity:table:{tableId})来自 ComputedUpdateLock.ts按 tableId 排序加锁避免不同任务对同一表的读-改-写交错投影行锁loadActivity对computed_field_activity与computed_table_activity使用SELECT ... FOR UPDATE配合 advisory lock 保证快照串行化。此外onTaskEnqueued还通过serializeEnqueueProjection在进程内按 tableId 维护一条 promise 尾链enqueueProjectionTails将并发入队投影串行化——因为入队阶段的任务 id 尚未落库进程内串行可以避免同表并发入队导致的引用计数竞态。4.2 状态迁移细节onTaskEnqueued先向computed_task_field_ref插入(task_id, field_id, table_id, base_id, was_processingfalse)onConflict doNothing保证同一任务对同一字段只计一次若全部冲突无新目标则直接返回null否则ensureActivityRows初始化行attachTask迁移状态后persistSnapshot写回。onTasksClaimed按 taskId 查 refs过滤was_processing ! true的挂起引用将对应的 refs 置为was_processingtrue调用activity.reconcileProcessing用countProcessingRefs统计该字段was_processingtrue的 ref 数覆盖内存计数实现从持久化真相的对账。onTaskDone/onTaskFailed(terminal)走私有releaseTask——按 taskId 查 refs先按字段聚合was_processing标志支持同一任务跨字段的混合状态删除 refs再逐字段activity.releaseTask终态失败会携带lastError。onTaskFailed(!terminal)重试路径将 refs 的was_processing清回falsereconcileProcessing注入lastError并保留 queued 状态等待重新领取。五、领域事件与实时投影ShareDBcmp_{tableId}集合5.1 事务提交后才发布事件投影在事务内更新三张表后ComputedActivityBatchChanged领域事件ComputedActivityBatchChanged.ts携带baseId、变更的fields与tablesDTO 数组只在包裹事务提交之后才发布。这是关键的一致性保障事件发布与数据库提交之间的窗口被消除实时订阅方不会看到未提交的中间状态。5.2 RealtimeProjection 的写入逻辑ComputedActivityRealtimeProjection.ts 是ComputedActivityBatchChanged的投影处理器ProjectionHandler(ComputedActivityBatchChanged)。它将表与字段文档写入cmp_{tableId}这个 ShareDB 集合文档 idtable表级 computeMeta 摘要与{fieldId}字段级 computeMeta每个变更过的文档只被写一次按 tableId 去重后逐表处理版本语义Activity generation 就是 ShareDB 文档版本——generation 1时通过ensure创建文档generation 1时通过applyChange提交一个{ type: set, path: [], value }的根替换操作并指定{ version: generation - 1 }即文档版本 generation - 1。这一设计使得客户端的每次状态推进都对应一个单调递增的文档版本。写入前字段与表 DTO 会被映射为公共实时视图表级暴露status / calculatingFieldCount / queuedFieldCount / estimatedComplexity / recentCompletions / generation / computeMode / updatedAt字段级暴露status / estimatedComplexity / estimatedDirtyRecords / generation / startedAt / lastDurationMs / lastError / updatedAt / activeTaskCount / processingTaskCount / batchProgressbatchProgress由getFieldComputeBatchProgress从extensions中解析仅当字段非 idle 且有进度时返回。5.3 快照加载器的鉴权文档说明后端快照加载器对普通客户端通过字段读权限field-read permissions授权并校验 share-view 客户端请求的是其共享的表。配合断开重连恢复机制客户端重连时只合成缺失的[from, generation)操作区间且以最新快照为基底避免重放全部历史操作。六、查询 APIGET /tables/getComputeActivity文档指定的 API 为GET /tables/getComputeActivity。路由注册位于 contract-http-implementation/src/router.tsos.tables.getComputeActivity.handler(...)并通过getComputeActivity: tablesGetComputeActivity挂载HTTP 处理器见 handlers/tables/getComputeActivity.ts。该查询首先验证 base/table 关联并执行常规的表读取操作守卫table-read operation guard然后才读取诊断数据。核心处理逻辑在 GetComputeActivityHandler.ts构造Table.specs(baseId).byId(tableId)规格并tableRepository.findOne查找表未找到返回table.not_found通过TableOperationPluginRunner.prepare({ kind: TableOperationKind.read, ... })执行读操作插件链并guard()校验权限调用activityReader.getByTableId(context, tableId)读取投影快照当活动行为空时回填baseId。6.1 读取适配器与诊断信息PostgreSQL 读取实现 PostgresComputedActivityReader.ts 按table_id查询computed_field_activity与computed_table_activity并用buildDiagnostics构建诊断摘要包括activeFieldCountqueued running、queuedFieldCount、calculatingFieldCount、failedFieldCounthighComplexityFieldCountestimatedComplexity HIGH_COMPLEXITY_THRESHOLD阈值常量从teable/v2-core导入anomalies列表包含failed携带lastError.message、high_complexity携带预估复杂度、all_target_records全表重算进行中或刚投影三类告警固定computeMode: server。另外文档提到 Table DTO 加载器也可能 join 活动行把可选的字段/表computeMeta暴露到 Table DTO 中getTableById处理器可见ComputedActivity相关引用。七、客户端集成ComputeActivityProvider 与实时合并SDK 侧teable/sdk每个挂载的表对应一个ComputeActivityProviderComputeActivityProvider.tsx订阅逻辑在 use-compute-activity.ts。其行为要点合并 HTTP 快照与 ShareDB 更新初次渲染以 HTTP 快照兜底之后以 ShareDB 实时推送为准实时字段状态优先于陈旧 HTTP 状态避免界面短暂回退到旧状态驱动琥珀色计算中表头amber calculating headersrunning/queued字段在表格视图中以醒目颜色标识计算中失败诊断保持可见即使活跃工作已停止如任务终态失败后failed诊断信息仍保留在界面上供排查。ComputeActivityProvider每表仅运行一次订阅并把 revision/fieldMeta 通过 Context 共享给useGridColumns与字段面板使列主题column themes能随计算状态变化即时重算。八、测试与验证仓库为整个子系统提供了完整测试覆盖可作为深入研读与验证的入口投影器单测ComputedActivityProjector.spec.ts、目标解析单测 resolveFieldTargets.spec.ts领域聚合单测packages/v2/core/src/domain/computed/ComputedActivity.spec.ts、实时投影单测 ComputedActivityRealtimeProjection.spec.ts查询处理器单测GetComputeActivityHandler.spec.ts端到端测试computed-activity.e2e.spec.ts、outbox 集成单测 ComputedUpdateOutbox.spec.ts。九、非目标Non-goals与边界文档明确划定了该子系统的边界值得实现者注意不负责公式中间结果的大小限制formula intermediate size limits 由计算引擎层处理不属于活动元数据范畴不维护持久化的完整任务历史只保留有界默认 20 条的最近完成摘要recent_completions过期记录随pushCompletion的滑动窗口自然淘汰。这两点说明 Computed Activity 定位是运行时可观测元数据而非审计日志或任务编排存储任何扩展需求都应在该边界内设计。十、总结一条贯穿领域 → 存储 → 实时 → UI 的完整链路Computed Activity 的设计可以用一条链路概括异步 outbox 任务的生命周期事件enqueue/claim/done/failed→ 同一事务内更新三张投影表field/table/task-field-ref→ 提交后发布ComputedActivityBatchChanged→ RealtimeProjection 以 generation 驱动写入 ShareDBcmp_{tableId}文档 → 客户端ComputeActivityProvider合并快照与实时流 → 网格表头展示 calculating/完成耗时/失败诊断。全链路的核心不变量是字段 schema 零污染、引用计数幂等可重建、generation 与文档版本一一对应、事件只在事务提交后发布。对于任何需要为异步计算任务构建运行时进度可观测能力的项目而言这份架构文档与其仓库实现核心聚合、PostgreSQL 投影适配器、ShareDB 实时投影、HTTP 查询与 SDK 订阅构成了一套可复用的参考范式。【免费下载链接】teable✨ AI Spreadsheet for Business项目地址: https://gitcode.com/GitHub_Trending/te/teable创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考