行的源碼級解析)
Langfuse 后臺遷移Background Migrations完全指南從 Prisma 注冊到 Env 門控執(zhí)行的源碼級解析【免費下載鏈接】langfuse Open source AI engineering platform: LLM evals, observability, metrics, prompt management, playground, datasets. Integrates with OpenTelemetry, LangChain, OpenAI SDK, LiteLLM, and more. YC W23項目地址: https://gitcode.com/GitHub_Trending/la/langfuse本文圍繞 Langfuse 倉庫中 worker/src/backgroundMigrations/README.md 的技術(shù)骨架展開系統(tǒng)講解后臺遷移機制的設(shè)計動機、注冊方式、IBackgroundMigration接口契約、管理器BackgroundMigrationManager的加鎖/心跳/斷點續(xù)跑執(zhí)行流程以及envGate休眠行機制的完整用法。讀完本文你將能夠判斷何種操作應(yīng)當作為后臺遷移而非普通 Prisma 遷移在 Langfuse 中正確新增一個可恢復(fù)、可排序、可門控的后臺遷移并理解 worker 如何保證遷移過程中事件處理不中斷、遷移失敗后如何被其他 worker 接管續(xù)跑。什么是后臺遷移何時必須使用它后臺遷移Background Migration是運行時間較長、必須在新版本應(yīng)用能正確服務(wù)之前完成的作業(yè)型任務(wù)。Langfuse 將其與常規(guī)數(shù)據(jù)庫遷移Prisma migration區(qū)分開當某個數(shù)據(jù)變更操作超過約 5 分鐘或不是原子操作時就不應(yīng)再放進標準遷移里而應(yīng)交給后臺遷移機制。典型的適用場景包括填充新增的可選列backfill 歷史數(shù)據(jù)在表之間或系統(tǒng)之間搬運數(shù)據(jù)如 PostgreSQL 與 ClickHouse 之間的數(shù)據(jù)同步對存量數(shù)據(jù)進行加密、重寫、清理等耗時操作。從倉庫現(xiàn)狀看這些場景都有真實對應(yīng)實現(xiàn)worker/src/backgroundMigrations 目錄下共 11 個遷移文件覆蓋了addGenerationsCostBackfill.ts——為歷史 generation 觀測回填成本字段README 中的 CLI 示例即此腳本backfillEventsFullFromObservations.ts、backfillEventsFullFromDatasetRunItems.ts——ClickHouse 事件表回填backfillSysIdForDatasetItems.ts、backfillValidToForDatasetItems.ts——dataset_items 系統(tǒng) ID 與valid_to時間戳回填createRootSpansFromTraces.ts、rewriteObservationsToPidTidSorting.ts、dropPidTidSortingTables.ts——V4 歷史回填鏈M1M5中的根 span 創(chuàng)建、表重寫與清理encryptBlobStorageSecrets.ts——對 blob 存儲集成中未加密的secretAccessKey進行加密補寫backfillBillingCycleAnchors.ts、patchLLMToolAndLLLMSchemaAuditLogs.ts——計費錨點與審計日志修補。從源碼結(jié)構(gòu)可以推斷該機制是 Langfuse 處理大規(guī)模數(shù)據(jù)遷移的統(tǒng)一通道幾乎每一次涉及存量數(shù)據(jù)改寫的能力發(fā)布都會經(jīng)由它執(zhí)行。新增后臺遷移的完整步驟新增一個后臺遷移需要同時做兩件事在數(shù)據(jù)庫background_migrations表中插入一行通常通過 Prisma migration SQL在當前目錄worker/src/backgroundMigrations新增一個遷移腳本文件。注冊行數(shù)據(jù)庫中的遷移狀態(tài)記錄后臺遷移的狀態(tài)持久化在background_migrations表中其 Prisma 模型定義位于 packages/shared/prisma/schema.prisma#L334-L347model BackgroundMigration { id String id default(cuid()) name String unique script String map(script) args Json map(args) state Json default({}) map(state) finishedAt DateTime? map(finished_at) failedAt DateTime? map(failed_at) failedReason String? map(failed_reason) workerId String? map(worker_id) lockedAt DateTime? map(locked_at) map(background_migrations) }各字段語義如下字段類型說明idString (PK)遷移行唯一 ID遷移腳本中硬編碼引用見下文nameString (unique)遷移名稱必須可排序建議以日期為前綴如20260701_v4_step_2_...scriptString指向遷移腳本文件的名稱管理器據(jù)此require(./script)加載類argsJson遷移運行參數(shù)可攜帶envGate門控聲明、projectId等stateJson遷移運行中間狀態(tài)如游標、分塊進度用于斷點續(xù)跑finishedAt/failedAt/failedReasonDateTime?/String?完成/失敗標記及原因workerId/lockedAtString?/DateTime?分布式鎖信息誰在跑、何時鎖定的實現(xiàn)類IBackgroundMigration 接口遷移腳本的默認導(dǎo)出必須實現(xiàn) IBackgroundMigration 接口完整定義如下export interface IBackgroundMigration { validate: ( args: Recordstring, unknown, ) Promise{ valid: boolean; invalidReason: string | undefined }; run: (args: Recordstring, unknown) Promisevoid; abort: () Promisevoid; }三個方法的職責validate(args)在執(zhí)行前校驗前置條件返回{ valid, invalidReason }。校驗失敗的遷移會被標記為failedAt不會執(zhí)行run。典型用法包括檢查注冊行是否存在、檢查目標列是否存在見 backfillValidToForDatasetItems.ts#L32-L67 中對information_schema.columns的查詢、檢查前置遷移是否已成功完成見下文遷移鏈依賴守衛(wèi)。run(args)執(zhí)行實際遷移邏輯必須可恢復(fù)、可中斷。abort()worker 關(guān)閉或鎖被接管時被調(diào)用遷移實現(xiàn)應(yīng)在此置位已中止標志并優(yōu)雅退出。以 backfillValidToForDatasetItems.ts 為范本一個標準的實現(xiàn)會在文件頂部用注釋硬編碼backgroundMigrationId該 ID 與 Prisma migration SQL 中插入的行 ID 一一對應(yīng)run()內(nèi)采用游標式分批處理每次處理一批默認batchSize 1000處理完將lastProcessedProjectId/lastProcessedId游標寫回background_migrations.state批次之間默認休眠delayBetweenBatchesMs 200ms以降低對數(shù)據(jù)庫的壓力每輪循環(huán)檢查this.isAborted被中止時立即跳出留下可續(xù)跑的狀態(tài)。其注釋明確描述了性能策略一次只處理一個 project以利用(project_id, id, valid_from)復(fù)合索引每個 project 內(nèi)最多批量取 100 個id對用LEAD()窗口函數(shù)計算相鄰版本的valid_to——窗口函數(shù)只作用于當前批次而不是整張表從而避免全表掃描式的內(nèi)存開銷。注冊行與腳本的對應(yīng)關(guān)系兩者的關(guān)聯(lián)通過硬編碼 UUID與script 文件名完成。以 V4 回填鏈為例Prisma 遷移 SQL 中插入INSERT INTO background_migrations (id, name, script, args) VALUES ( 9c2d5a4f-7b8e-4f6a-a91c-3e5d7f8a2b1c, 20260701_v4_step_2_rewrite_observations_to_pid_tid_sorting, rewriteObservationsToPidTidSorting, {envGate: LANGFUSE_BACKGROUND_MIGRATION_V4_ENABLE_HISTORIC_BACKFILL}::jsonb );而腳本 rewriteObservationsToPidTidSorting.ts 中同樣硬編碼了該 UUID。管理器正是通過script字段執(zhí)行new (require(\./${migration.script}).default)() 來加載遷移類見 backgroundMigrationManager.ts#L114。倉庫中已注冊的后臺遷移可見于 packages/shared/prisma/migrations 下多個以add_..._background_migration命名的遷移目錄例如20241024121500_add_generations_cost_backfill_background_migration、20260701115500_add_create_root_spans_background_migration等每個目錄的migration.sql都以INSERT INTO background_migrations (id, name, script, args)開頭。遷移的硬性要求RequirementsREADME 明確了后臺遷移必須遵守的七條要求這是設(shè)計評審的核心檢查清單任何時刻都必須可恢復(fù)recoverable遷移可以被中斷并且必須能在任意階段恢復(fù)。實現(xiàn)方式有兩種——跨系統(tǒng)遷移做成冪等idempotent單數(shù)據(jù)庫內(nèi)遷移保證每次變更原子。同一時刻只能有一個后臺遷移運行這并非技術(shù)限制而是為了讓推理遷移行為更容易。必須在 changelog 及其他頁面高亮提示如果代碼依賴某個后臺遷移已完成必須明確告知運維人員README 引用了 GitLab 的 upgrade stops 作為溝通范式。遷移名稱必須可排序因為遷移按name升序依次執(zhí)行最好以日期為前綴。必須假設(shè) worker 在遷移運行期間會繼續(xù)處理事件即遷移應(yīng)避免調(diào)用應(yīng)用代碼避免與應(yīng)用業(yè)務(wù)邏輯耦合。必須假設(shè)遷移運行期間有新事件持續(xù)寫入即遷移不能依賴數(shù)據(jù)庫狀態(tài)是靜態(tài)的要能容忍并發(fā)寫入。第 5、6 條尤其關(guān)鍵——它決定了后臺遷移與普通遷移的本質(zhì)區(qū)別后臺遷移運行在生產(chǎn)流量持續(xù)寫入的環(huán)境中因此逐條掃描、游標續(xù)跑、冪等重放是標配設(shè)計。執(zhí)行引擎BackgroundMigrationManager 的調(diào)度與鎖后臺遷移的執(zhí)行引擎是 backgroundMigrationManager.ts 中的BackgroundMigrationManager類。worker 在啟動時會檢查環(huán)境變量LANGFUSE_ENABLE_BACKGROUND_MIGRATIONS默認true定義于 worker/src/env.ts#L215-L217為真時異步啟動BackgroundMigrationManager.run()且不會阻塞隊列 worker見 worker/src/app.ts#L125-L130。主循環(huán)流程run()的執(zhí)行邏輯backgroundMigrationManager.ts#L41-L192可概括為掃描可運行遷移在數(shù)據(jù)庫事務(wù)中執(zhí)行findFirst篩選finishedAt null且failedAt null的行同時應(yīng)用 envGate 過濾見下一節(jié)按name升序取出最早一條。鎖檢查若取到遷移但lockedAt在最近60 秒內(nèi)被更新過則視為被其他 worker 持有直接結(jié)束本輪避免并發(fā)執(zhí)行。注意lockedAt不在數(shù)據(jù)庫查詢條件中因為findFirst可能返回其他未完成遷移若在查詢中過濾會導(dǎo)致漏鎖。加鎖通過update將workerId設(shè)為本進程的randomUUID()lockedAt設(shè)為當前時間。整個查詢加鎖過程包在isolationLevel: Serializable的事務(wù)中maxWait: 5000保證鎖獲取的原子性。心跳heartbeat調(diào)用heartBeat()后每15 秒將當前活躍遷移的lockedAt刷新為當前時間backgroundMigrationManager.ts#L20-L39。這既是我活著的信號也是鎖續(xù)約的機制。校驗并運行先調(diào)用migration.validate(args)。校驗失敗則寫入failedAt/failedReason并繼續(xù)下一條成功則調(diào)用migration.run(args)。標記完成run()正常返回且遷移仍處于活躍狀態(tài)時寫入finishedAt并清空lockedAt。循環(huán)繼續(xù)找下一條未完成遷移直到?jīng)]有可運行的遷移為止。崩潰接管與優(yōu)雅關(guān)閉worker 被 kill由于心跳停止lockedAt不再刷新60 秒后鎖自然過期另一個 worker會取到該遷移并從上次斷點繼續(xù)。這正是 README 描述的另一個 worker 會接手續(xù)跑機制的實現(xiàn)遷移進度依賴state字段中的游標/分塊狀態(tài)而非進程內(nèi)內(nèi)存。worker 優(yōu)雅關(guān)閉close()backgroundMigrationManager.ts#L194-L211會調(diào)用活躍遷移的abort()方法隨后清空lockedAt但不寫finishedAt因為遷移并未完成。該close()在 worker/src/utils/shutdown.ts#L105 的關(guān)閉流程中被調(diào)用。遷移鏈依賴守衛(wèi)管理器本身沒有依賴模型——它只按名稱順序執(zhí)行所有合格行并獨立標記每個遷移的完成/失敗。這意味著上游遷移失敗并不會自動阻止下游遷移在部分數(shù)據(jù)上運行。為此倉庫在 worker/src/backgroundMigrations/utils/backfillBase.ts#L122-L151 提供了checkPredecessorMigrationFinalized(predecessorId, predecessorName)工具前置遷移未注冊、failedAt非空、或finishedAt為空均返回校驗失敗下游遷移在validate()中調(diào)用它失敗信息會經(jīng)由管理器的校驗失敗路徑寫入failedAt由于鏈條上的每一步都守衛(wèi)自己的前置步驟單個失敗會傳遞性地中斷整條鏈。典型用例見 dropPidTidSortingTables.ts#L38-L45M5 清理遷移校驗時檢查 M420260701_v4_step_4_backfill_events_full_from_dataset_run_items是否已完成。本地與測試環(huán)境執(zhí)行命令行運行方式后臺遷移最好能通過命令行直接執(zhí)行以便本地開發(fā)或在 staging 環(huán)境測試。README 給出的標準命令是cd worker dotenv -e ../.env -- npx tsx src/backgroundMigrations/script-name.ts # 示例 dotenv -e ../.env -- npx tsx src/backgroundMigrations/addGenerationsCostBackfill.ts該命令依賴dotenv-cli加載倉庫根目錄的.env用tsx直接運行遷移腳本。腳本通過require.main module判斷自身是被直接執(zhí)行還是被導(dǎo)入被直接執(zhí)行時進入 CLI 入口例如 backfillValidToForDatasetItems.ts#L163-L176先validate再run失敗以非零退出碼結(jié)束進程。分塊回填類遷移還提供了一套標準 CLI 參數(shù)定義于 utils/backfillBase.ts#L834-L908 的runBackfillMigrationCli參數(shù)短選項默認值說明--concurrency-c1同時運行的 ClickHouse 查詢數(shù)--pollIntervalMs-p30000輪詢活躍查詢狀態(tài)的間隔毫秒--maxRetries-r3單個分塊的最大重試次數(shù)--retryFailed-ffalse將失敗分塊重置為 pending 重新嘗試--partitions可多值無僅處理指定的 yyyymm 分區(qū)深入分塊回填基類與 ClickHouse 斷點恢復(fù)V4 歷史回填鏈的遷移M2M4并非直接實現(xiàn)IBackgroundMigration而是繼承抽象基類ChunkedClickhouseBackfillMigrationutils/backfillBase.ts#L429-L823這是 README可恢復(fù)要求最完整的工程化體現(xiàn)分塊枚舉從 ClickHousesystem.parts發(fā)現(xiàn)活躍的 yyyymm 分區(qū)跳過patch-%與all元分區(qū)每個分區(qū)生成一個BaseChunkTodopending → in_progress → completed/failed見loadPartitionsFromClickhousebackfillBase.ts#L166-L208。fire-and-poll 模式fireQuerybackfillBase.ts#L247-L346發(fā)起一個長時運行的 ClickHouse 查詢確認其在system.processes中被跟蹤后主動斷開 HTTP 連接讓查詢在服務(wù)端繼續(xù)執(zhí)行再通過pollQueryStatus輪詢完成狀態(tài)并設(shè)置max_execution_time: 0避免服務(wù)端超時掐斷長查詢。崩潰恢復(fù)recoverInProgressTodosbackfillBase.ts#L360-L412在每次run()開始時重新關(guān)聯(lián)上一次 worker 遺留的 in-flight 查詢——已完成的標記完成失敗的按重試計數(shù)重置為 pending仍運行的繼續(xù)跟蹤。所有分塊狀態(tài)todos、activeQueries、phase、config都持久化在background_migrations.state中l(wèi)oadState/updateState。并發(fā)與重試調(diào)度循環(huán)按concurrency填充空閑槽位每個分塊失敗后遞增retryCount達到maxRetries才標記為永久失敗并拋出錯誤使管理器寫入failedAt--retry-failed可將其重置后重跑。遷移鏈守衛(wèi)與表存在性校驗validate()先檢查前置遷移是否finishedAt再對requiredTables逐表SHOW TABLES確認存在最多重試 5 次、間隔 10 秒隨后觸發(fā)afterTablesValidated鉤子執(zhí)行惰性 DDL如創(chuàng)建 scratch 表。這段實現(xiàn)解釋了 README 要求的底層落地所有進度都在數(shù)據(jù)庫中中斷只是原地暫?；謴?fù)只是重新 attach。環(huán)境變量門控遷移envGate / dormant 行動機與機制某些遷移需要在 release N 就隨包發(fā)布但只能在 release N1 才執(zhí)行或僅在運維人員顯式 opt-in 時執(zhí)行。為此引入envGate機制帶門控的遷移行處于dormant休眠狀態(tài)——它存在于background_migrations表中且finished_at NULL但管理器在查詢時跳過它直到對應(yīng)的環(huán)境變量被設(shè)為true。關(guān)鍵設(shè)計點門控檢查被推入findFirst的謂詞中backgroundMigrationManager.ts#L64-L76因此休眠行不會 head-of-line 阻塞其后按名稱排序的其他遷移——排在它后面的、未門控或門控已開啟的遷移照常運行。如何編寫一個門控遷移README 給出了三步流程結(jié)合倉庫證據(jù)展開如下第 1 步選擇門控名稱必須以LANGFUSE_BACKGROUND_MIGRATION_為前綴。管理器通過掃描已驗證的 env不是全部process.env中鍵以該前綴開頭且值為true的變量來發(fā)現(xiàn)活躍門控backgroundMigrationManager.ts#L51-L55。第 2 步在 Prisma 遷移 SQL 中通過行的args聲明門控INSERT INTO background_migrations (id, name, script, args) VALUES ( ..., 20260521120000_my_dormant_migration, myDormantMigration, {projectId: ..., envGate: LANGFUSE_BACKGROUND_MIGRATION_V4_ENABLE_MY_FEATURE}::jsonb );args.envGate的取值必須與第 1 步的門控名一致。倉庫中的真實樣例可見 20260701115504_add_drop_pid_tid_sorting_tables_background_migration/migration.sql門控LANGFUSE_BACKGROUND_MIGRATION_V4_DROP_PID_TID_SORTING_TABLES與 20260701115501_add_rewrite_observations_to_pid_tid_sorting_background_migration/migration.sql門控LANGFUSE_BACKGROUND_MIGRATION_V4_ENABLE_HISTORIC_BACKFILL。第 3 步在worker/src/env.ts的EnvSchema中注冊該環(huán)境變量使用z.enum([true, false]).default(false)使其類型化、啟動時校驗、且默認休眠LANGFUSE_BACKGROUND_MIGRATION_V4_ENABLE_HISTORIC_BACKFILL: z .enum([true, false]) .default(true), LANGFUSE_BACKGROUND_MIGRATION_V4_DROP_PID_TID_SORTING_TABLES: z .enum([true, false]) .default(false),以上兩行取自 worker/src/env.ts#L601-L606注釋還解釋了各自默認值的取舍歷史回填門控默認開啟對無歷史數(shù)據(jù)的新部署是 no-op而排序表清理門控默認關(guān)閉以便自托管用戶保留中間產(chǎn)物做排查直到確認新路徑健康后再開啟。門控的運行時行為環(huán)境變量為true該行對管理器可見按正常的名稱順序執(zhí)行環(huán)境變量為false或缺失該行不可見——不加鎖、不打印 skip 日志、不造成 head-of-line 阻塞。管理器執(zhí)行時會先收集所有值為true的LANGFUSE_BACKGROUND_MIGRATION_*變量構(gòu)成activeGates數(shù)組然后構(gòu)造OR條件要么args.envGate為空Prisma.AnyNull即未門控要么等于某個活躍門控名。這樣一條查詢就同時實現(xiàn)了跳過休眠行與保留普通遷移兩種語義見 backgroundMigrationManager.ts#L57-L76。自檢清單與運維注意事項綜合以上機制在 Langfuse 中落地一個新的后臺遷移建議按如下清單核查是否確實需要后臺遷移單次運行超過約 5 分鐘或非原子操作 → 是否則請走標準 Prisma migration。注冊完備性Prisma 遷移 SQL 已插入background_migrations行腳本文件已存在于 worker/src/backgroundMigrations默認導(dǎo)出實現(xiàn)了 IBackgroundMigration腳本內(nèi)硬編碼的 UUID 與 SQL 中的id一致。名稱可排序name以日期/序號前綴如20260701_v4_step_5_...確保執(zhí)行順序可預(yù)期?？苫謴?fù)性單庫內(nèi)逐步原子提交跨庫/跨系統(tǒng)PG ? ClickHouse冪等進度寫入state游標或分塊狀態(tài)。并發(fā)友好不依賴數(shù)據(jù)庫靜態(tài)狀態(tài)run()內(nèi)每輪檢查isAborted不調(diào)用應(yīng)用層業(yè)務(wù)代碼。門控如需休眠args.envGate以LANGFUSE_BACKGROUND_MIGRATION_前綴命名已在worker/src/env.ts的EnvSchema中注冊z.enum([true,false]).default(false)。依賴鏈如需下游遷移在validate()中調(diào)用checkPredecessorMigrationFinalized守衛(wèi)上游上游失敗會傳遞性阻斷整條鏈。變更溝通若應(yīng)用代碼依賴遷移完成需在 changelog 與部署文檔中顯式標注參考 GitLab 的 upgrade stops 溝通方式。運行時驗證本地用dotenv -e ../.env -- npx tsx src/backgroundMigrations/script-name.ts單跑驗證staging 可用--concurrency/--maxRetries/--retry-failed/--partitions控制執(zhí)行面生產(chǎn)環(huán)境確認LANGFUSE_ENABLE_BACKGROUND_MIGRATIONStrue且 worker 心跳15s 刷新lockedAt正常。故障排查遷移卡住時檢查background_migrations行的lockedAt60 秒內(nèi)視為鎖有效、failedReason、state中的游標失敗后清除failedAt可讓管理器重試該行分塊遷移可清failedAt后以--retry-failed重跑失敗分塊。通過這套機制Langfuse 得以在不中斷事件流處理的前提下安全地完成從填充可選列到整表跨系統(tǒng)重寫的各種重型數(shù)據(jù)遷移并將失敗恢復(fù)、多 worker 接管、分階段發(fā)布等運維復(fù)雜度收斂在 backgroundMigrationManager.ts 與 utils/backfillBase.ts 兩處基礎(chǔ)設(shè)施之中?！久赓M下載鏈接】langfuse Open source AI engineering platform: LLM evals, observability, metrics, prompt management, playground, datasets. Integrates with OpenTelemetry, LangChain, OpenAI SDK, LiteLLM, and more. YC W23項目地址: https://gitcode.com/GitHub_Trending/la/langfuse創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考 Langfuse 后臺遷移（Background Migrations）完全指南：從 Prisma 注冊到 Env 門控執(zhí)行的源碼級解析
Langfuse 后臺遷移Background Migrations完全指南從 Prisma 注冊到 Env 門控執(zhí)行的源碼級解析【免費下載鏈接】langfuse Open source AI engineering platform: LLM evals, observability, metrics, prompt management, playground, datasets. Integrates with OpenTelemetry, LangChain, OpenAI SDK, LiteLLM, and more. YC W23項目地址: https://gitcode.com/GitHub_Trending/la/langfuse本文圍繞 Langfuse 倉庫中 worker/src/backgroundMigrations/README.md 的技術(shù)骨架展開系統(tǒng)講解后臺遷移機制的設(shè)計動機、注冊方式、IBackgroundMigration接口契約、管理器BackgroundMigrationManager的加鎖/心跳/斷點續(xù)跑執(zhí)行流程以及envGate休眠行機制的完整用法。讀完本文你將能夠判斷何種操作應(yīng)當作為后臺遷移而非普通 Prisma 遷移在 Langfuse 中正確新增一個可恢復(fù)、可排序、可門控的后臺遷移并理解 worker 如何保證遷移過程中事件處理不中斷、遷移失敗后如何被其他 worker 接管續(xù)跑。什么是后臺遷移何時必須使用它后臺遷移Background Migration是運行時間較長、必須在新版本應(yīng)用能正確服務(wù)之前完成的作業(yè)型任務(wù)。Langfuse 將其與常規(guī)數(shù)據(jù)庫遷移Prisma migration區(qū)分開當某個數(shù)據(jù)變更操作超過約 5 分鐘或不是原子操作時就不應(yīng)再放進標準遷移里而應(yīng)交給后臺遷移機制。典型的適用場景包括填充新增的可選列backfill 歷史數(shù)據(jù)在表之間或系統(tǒng)之間搬運數(shù)據(jù)如 PostgreSQL 與 ClickHouse 之間的數(shù)據(jù)同步對存量數(shù)據(jù)進行加密、重寫、清理等耗時操作。從倉庫現(xiàn)狀看這些場景都有真實對應(yīng)實現(xiàn)worker/src/backgroundMigrations 目錄下共 11 個遷移文件覆蓋了addGenerationsCostBackfill.ts——為歷史 generation 觀測回填成本字段README 中的 CLI 示例即此腳本backfillEventsFullFromObservations.ts、backfillEventsFullFromDatasetRunItems.ts——ClickHouse 事件表回填backfillSysIdForDatasetItems.ts、backfillValidToForDatasetItems.ts——dataset_items 系統(tǒng) ID 與valid_to時間戳回填createRootSpansFromTraces.ts、rewriteObservationsToPidTidSorting.ts、dropPidTidSortingTables.ts——V4 歷史回填鏈M1M5中的根 span 創(chuàng)建、表重寫與清理encryptBlobStorageSecrets.ts——對 blob 存儲集成中未加密的secretAccessKey進行加密補寫backfillBillingCycleAnchors.ts、patchLLMToolAndLLLMSchemaAuditLogs.ts——計費錨點與審計日志修補。從源碼結(jié)構(gòu)可以推斷該機制是 Langfuse 處理大規(guī)模數(shù)據(jù)遷移的統(tǒng)一通道幾乎每一次涉及存量數(shù)據(jù)改寫的能力發(fā)布都會經(jīng)由它執(zhí)行。新增后臺遷移的完整步驟新增一個后臺遷移需要同時做兩件事在數(shù)據(jù)庫background_migrations表中插入一行通常通過 Prisma migration SQL在當前目錄worker/src/backgroundMigrations新增一個遷移腳本文件。注冊行數(shù)據(jù)庫中的遷移狀態(tài)記錄后臺遷移的狀態(tài)持久化在background_migrations表中其 Prisma 模型定義位于 packages/shared/prisma/schema.prisma#L334-L347model BackgroundMigration { id String id default(cuid()) name String unique script String map(script) args Json map(args) state Json default({}) map(state) finishedAt DateTime? map(finished_at) failedAt DateTime? map(failed_at) failedReason String? map(failed_reason) workerId String? map(worker_id) lockedAt DateTime? map(locked_at) map(background_migrations) }各字段語義如下字段類型說明idString (PK)遷移行唯一 ID遷移腳本中硬編碼引用見下文nameString (unique)遷移名稱必須可排序建議以日期為前綴如20260701_v4_step_2_...scriptString指向遷移腳本文件的名稱管理器據(jù)此require(./script)加載類argsJson遷移運行參數(shù)可攜帶envGate門控聲明、projectId等stateJson遷移運行中間狀態(tài)如游標、分塊進度用于斷點續(xù)跑finishedAt/failedAt/failedReasonDateTime?/String?完成/失敗標記及原因workerId/lockedAtString?/DateTime?分布式鎖信息誰在跑、何時鎖定的實現(xiàn)類IBackgroundMigration 接口遷移腳本的默認導(dǎo)出必須實現(xiàn) IBackgroundMigration 接口完整定義如下export interface IBackgroundMigration { validate: ( args: Recordstring, unknown, ) Promise{ valid: boolean; invalidReason: string | undefined }; run: (args: Recordstring, unknown) Promisevoid; abort: () Promisevoid; }三個方法的職責validate(args)在執(zhí)行前校驗前置條件返回{ valid, invalidReason }。校驗失敗的遷移會被標記為failedAt不會執(zhí)行run。典型用法包括檢查注冊行是否存在、檢查目標列是否存在見 backfillValidToForDatasetItems.ts#L32-L67 中對information_schema.columns的查詢、檢查前置遷移是否已成功完成見下文遷移鏈依賴守衛(wèi)。run(args)執(zhí)行實際遷移邏輯必須可恢復(fù)、可中斷。abort()worker 關(guān)閉或鎖被接管時被調(diào)用遷移實現(xiàn)應(yīng)在此置位已中止標志并優(yōu)雅退出。以 backfillValidToForDatasetItems.ts 為范本一個標準的實現(xiàn)會在文件頂部用注釋硬編碼backgroundMigrationId該 ID 與 Prisma migration SQL 中插入的行 ID 一一對應(yīng)run()內(nèi)采用游標式分批處理每次處理一批默認batchSize 1000處理完將lastProcessedProjectId/lastProcessedId游標寫回background_migrations.state批次之間默認休眠delayBetweenBatchesMs 200ms以降低對數(shù)據(jù)庫的壓力每輪循環(huán)檢查this.isAborted被中止時立即跳出留下可續(xù)跑的狀態(tài)。其注釋明確描述了性能策略一次只處理一個 project以利用(project_id, id, valid_from)復(fù)合索引每個 project 內(nèi)最多批量取 100 個id對用LEAD()窗口函數(shù)計算相鄰版本的valid_to——窗口函數(shù)只作用于當前批次而不是整張表從而避免全表掃描式的內(nèi)存開銷。注冊行與腳本的對應(yīng)關(guān)系兩者的關(guān)聯(lián)通過硬編碼 UUID與script 文件名完成。以 V4 回填鏈為例Prisma 遷移 SQL 中插入INSERT INTO background_migrations (id, name, script, args) VALUES ( 9c2d5a4f-7b8e-4f6a-a91c-3e5d7f8a2b1c, 20260701_v4_step_2_rewrite_observations_to_pid_tid_sorting, rewriteObservationsToPidTidSorting, {envGate: LANGFUSE_BACKGROUND_MIGRATION_V4_ENABLE_HISTORIC_BACKFILL}::jsonb );而腳本 rewriteObservationsToPidTidSorting.ts 中同樣硬編碼了該 UUID。管理器正是通過script字段執(zhí)行new (require(\./${migration.script}).default)() 來加載遷移類見 backgroundMigrationManager.ts#L114。倉庫中已注冊的后臺遷移可見于 packages/shared/prisma/migrations 下多個以add_..._background_migration命名的遷移目錄例如20241024121500_add_generations_cost_backfill_background_migration、20260701115500_add_create_root_spans_background_migration等每個目錄的migration.sql都以INSERT INTO background_migrations (id, name, script, args)開頭。遷移的硬性要求RequirementsREADME 明確了后臺遷移必須遵守的七條要求這是設(shè)計評審的核心檢查清單任何時刻都必須可恢復(fù)recoverable遷移可以被中斷并且必須能在任意階段恢復(fù)。實現(xiàn)方式有兩種——跨系統(tǒng)遷移做成冪等idempotent單數(shù)據(jù)庫內(nèi)遷移保證每次變更原子。同一時刻只能有一個后臺遷移運行這并非技術(shù)限制而是為了讓推理遷移行為更容易。必須在 changelog 及其他頁面高亮提示如果代碼依賴某個后臺遷移已完成必須明確告知運維人員README 引用了 GitLab 的 upgrade stops 作為溝通范式。遷移名稱必須可排序因為遷移按name升序依次執(zhí)行最好以日期為前綴。必須假設(shè) worker 在遷移運行期間會繼續(xù)處理事件即遷移應(yīng)避免調(diào)用應(yīng)用代碼避免與應(yīng)用業(yè)務(wù)邏輯耦合。必須假設(shè)遷移運行期間有新事件持續(xù)寫入即遷移不能依賴數(shù)據(jù)庫狀態(tài)是靜態(tài)的要能容忍并發(fā)寫入。第 5、6 條尤其關(guān)鍵——它決定了后臺遷移與普通遷移的本質(zhì)區(qū)別后臺遷移運行在生產(chǎn)流量持續(xù)寫入的環(huán)境中因此逐條掃描、游標續(xù)跑、冪等重放是標配設(shè)計。執(zhí)行引擎BackgroundMigrationManager 的調(diào)度與鎖后臺遷移的執(zhí)行引擎是 backgroundMigrationManager.ts 中的BackgroundMigrationManager類。worker 在啟動時會檢查環(huán)境變量LANGFUSE_ENABLE_BACKGROUND_MIGRATIONS默認true定義于 worker/src/env.ts#L215-L217為真時異步啟動BackgroundMigrationManager.run()且不會阻塞隊列 worker見 worker/src/app.ts#L125-L130。主循環(huán)流程run()的執(zhí)行邏輯backgroundMigrationManager.ts#L41-L192可概括為掃描可運行遷移在數(shù)據(jù)庫事務(wù)中執(zhí)行findFirst篩選finishedAt null且failedAt null的行同時應(yīng)用 envGate 過濾見下一節(jié)按name升序取出最早一條。鎖檢查若取到遷移但lockedAt在最近60 秒內(nèi)被更新過則視為被其他 worker 持有直接結(jié)束本輪避免并發(fā)執(zhí)行。注意lockedAt不在數(shù)據(jù)庫查詢條件中因為findFirst可能返回其他未完成遷移若在查詢中過濾會導(dǎo)致漏鎖。加鎖通過update將workerId設(shè)為本進程的randomUUID()lockedAt設(shè)為當前時間。整個查詢加鎖過程包在isolationLevel: Serializable的事務(wù)中maxWait: 5000保證鎖獲取的原子性。心跳heartbeat調(diào)用heartBeat()后每15 秒將當前活躍遷移的lockedAt刷新為當前時間backgroundMigrationManager.ts#L20-L39。這既是我活著的信號也是鎖續(xù)約的機制。校驗并運行先調(diào)用migration.validate(args)。校驗失敗則寫入failedAt/failedReason并繼續(xù)下一條成功則調(diào)用migration.run(args)。標記完成run()正常返回且遷移仍處于活躍狀態(tài)時寫入finishedAt并清空lockedAt。循環(huán)繼續(xù)找下一條未完成遷移直到?jīng)]有可運行的遷移為止。崩潰接管與優(yōu)雅關(guān)閉worker 被 kill由于心跳停止lockedAt不再刷新60 秒后鎖自然過期另一個 worker會取到該遷移并從上次斷點繼續(xù)。這正是 README 描述的另一個 worker 會接手續(xù)跑機制的實現(xiàn)遷移進度依賴state字段中的游標/分塊狀態(tài)而非進程內(nèi)內(nèi)存。worker 優(yōu)雅關(guān)閉close()backgroundMigrationManager.ts#L194-L211會調(diào)用活躍遷移的abort()方法隨后清空lockedAt但不寫finishedAt因為遷移并未完成。該close()在 worker/src/utils/shutdown.ts#L105 的關(guān)閉流程中被調(diào)用。遷移鏈依賴守衛(wèi)管理器本身沒有依賴模型——它只按名稱順序執(zhí)行所有合格行并獨立標記每個遷移的完成/失敗。這意味著上游遷移失敗并不會自動阻止下游遷移在部分數(shù)據(jù)上運行。為此倉庫在 worker/src/backgroundMigrations/utils/backfillBase.ts#L122-L151 提供了checkPredecessorMigrationFinalized(predecessorId, predecessorName)工具前置遷移未注冊、failedAt非空、或finishedAt為空均返回校驗失敗下游遷移在validate()中調(diào)用它失敗信息會經(jīng)由管理器的校驗失敗路徑寫入failedAt由于鏈條上的每一步都守衛(wèi)自己的前置步驟單個失敗會傳遞性地中斷整條鏈。典型用例見 dropPidTidSortingTables.ts#L38-L45M5 清理遷移校驗時檢查 M420260701_v4_step_4_backfill_events_full_from_dataset_run_items是否已完成。本地與測試環(huán)境執(zhí)行命令行運行方式后臺遷移最好能通過命令行直接執(zhí)行以便本地開發(fā)或在 staging 環(huán)境測試。README 給出的標準命令是cd worker dotenv -e ../.env -- npx tsx src/backgroundMigrations/script-name.ts # 示例 dotenv -e ../.env -- npx tsx src/backgroundMigrations/addGenerationsCostBackfill.ts該命令依賴dotenv-cli加載倉庫根目錄的.env用tsx直接運行遷移腳本。腳本通過require.main module判斷自身是被直接執(zhí)行還是被導(dǎo)入被直接執(zhí)行時進入 CLI 入口例如 backfillValidToForDatasetItems.ts#L163-L176先validate再run失敗以非零退出碼結(jié)束進程。分塊回填類遷移還提供了一套標準 CLI 參數(shù)定義于 utils/backfillBase.ts#L834-L908 的runBackfillMigrationCli參數(shù)短選項默認值說明--concurrency-c1同時運行的 ClickHouse 查詢數(shù)--pollIntervalMs-p30000輪詢活躍查詢狀態(tài)的間隔毫秒--maxRetries-r3單個分塊的最大重試次數(shù)--retryFailed-ffalse將失敗分塊重置為 pending 重新嘗試--partitions可多值無僅處理指定的 yyyymm 分區(qū)深入分塊回填基類與 ClickHouse 斷點恢復(fù)V4 歷史回填鏈的遷移M2M4并非直接實現(xiàn)IBackgroundMigration而是繼承抽象基類ChunkedClickhouseBackfillMigrationutils/backfillBase.ts#L429-L823這是 README可恢復(fù)要求最完整的工程化體現(xiàn)分塊枚舉從 ClickHousesystem.parts發(fā)現(xiàn)活躍的 yyyymm 分區(qū)跳過patch-%與all元分區(qū)每個分區(qū)生成一個BaseChunkTodopending → in_progress → completed/failed見loadPartitionsFromClickhousebackfillBase.ts#L166-L208。fire-and-poll 模式fireQuerybackfillBase.ts#L247-L346發(fā)起一個長時運行的 ClickHouse 查詢確認其在system.processes中被跟蹤后主動斷開 HTTP 連接讓查詢在服務(wù)端繼續(xù)執(zhí)行再通過pollQueryStatus輪詢完成狀態(tài)并設(shè)置max_execution_time: 0避免服務(wù)端超時掐斷長查詢。崩潰恢復(fù)recoverInProgressTodosbackfillBase.ts#L360-L412在每次run()開始時重新關(guān)聯(lián)上一次 worker 遺留的 in-flight 查詢——已完成的標記完成失敗的按重試計數(shù)重置為 pending仍運行的繼續(xù)跟蹤。所有分塊狀態(tài)todos、activeQueries、phase、config都持久化在background_migrations.state中l(wèi)oadState/updateState。并發(fā)與重試調(diào)度循環(huán)按concurrency填充空閑槽位每個分塊失敗后遞增retryCount達到maxRetries才標記為永久失敗并拋出錯誤使管理器寫入failedAt--retry-failed可將其重置后重跑。遷移鏈守衛(wèi)與表存在性校驗validate()先檢查前置遷移是否finishedAt再對requiredTables逐表SHOW TABLES確認存在最多重試 5 次、間隔 10 秒隨后觸發(fā)afterTablesValidated鉤子執(zhí)行惰性 DDL如創(chuàng)建 scratch 表。這段實現(xiàn)解釋了 README 要求的底層落地所有進度都在數(shù)據(jù)庫中中斷只是原地暫?；謴?fù)只是重新 attach。環(huán)境變量門控遷移envGate / dormant 行動機與機制某些遷移需要在 release N 就隨包發(fā)布但只能在 release N1 才執(zhí)行或僅在運維人員顯式 opt-in 時執(zhí)行。為此引入envGate機制帶門控的遷移行處于dormant休眠狀態(tài)——它存在于background_migrations表中且finished_at NULL但管理器在查詢時跳過它直到對應(yīng)的環(huán)境變量被設(shè)為true。關(guān)鍵設(shè)計點門控檢查被推入findFirst的謂詞中backgroundMigrationManager.ts#L64-L76因此休眠行不會 head-of-line 阻塞其后按名稱排序的其他遷移——排在它后面的、未門控或門控已開啟的遷移照常運行。如何編寫一個門控遷移README 給出了三步流程結(jié)合倉庫證據(jù)展開如下第 1 步選擇門控名稱必須以LANGFUSE_BACKGROUND_MIGRATION_為前綴。管理器通過掃描已驗證的 env不是全部process.env中鍵以該前綴開頭且值為true的變量來發(fā)現(xiàn)活躍門控backgroundMigrationManager.ts#L51-L55。第 2 步在 Prisma 遷移 SQL 中通過行的args聲明門控INSERT INTO background_migrations (id, name, script, args) VALUES ( ..., 20260521120000_my_dormant_migration, myDormantMigration, {projectId: ..., envGate: LANGFUSE_BACKGROUND_MIGRATION_V4_ENABLE_MY_FEATURE}::jsonb );args.envGate的取值必須與第 1 步的門控名一致。倉庫中的真實樣例可見 20260701115504_add_drop_pid_tid_sorting_tables_background_migration/migration.sql門控LANGFUSE_BACKGROUND_MIGRATION_V4_DROP_PID_TID_SORTING_TABLES與 20260701115501_add_rewrite_observations_to_pid_tid_sorting_background_migration/migration.sql門控LANGFUSE_BACKGROUND_MIGRATION_V4_ENABLE_HISTORIC_BACKFILL。第 3 步在worker/src/env.ts的EnvSchema中注冊該環(huán)境變量使用z.enum([true, false]).default(false)使其類型化、啟動時校驗、且默認休眠LANGFUSE_BACKGROUND_MIGRATION_V4_ENABLE_HISTORIC_BACKFILL: z .enum([true, false]) .default(true), LANGFUSE_BACKGROUND_MIGRATION_V4_DROP_PID_TID_SORTING_TABLES: z .enum([true, false]) .default(false),以上兩行取自 worker/src/env.ts#L601-L606注釋還解釋了各自默認值的取舍歷史回填門控默認開啟對無歷史數(shù)據(jù)的新部署是 no-op而排序表清理門控默認關(guān)閉以便自托管用戶保留中間產(chǎn)物做排查直到確認新路徑健康后再開啟。門控的運行時行為環(huán)境變量為true該行對管理器可見按正常的名稱順序執(zhí)行環(huán)境變量為false或缺失該行不可見——不加鎖、不打印 skip 日志、不造成 head-of-line 阻塞。管理器執(zhí)行時會先收集所有值為true的LANGFUSE_BACKGROUND_MIGRATION_*變量構(gòu)成activeGates數(shù)組然后構(gòu)造OR條件要么args.envGate為空Prisma.AnyNull即未門控要么等于某個活躍門控名。這樣一條查詢就同時實現(xiàn)了跳過休眠行與保留普通遷移兩種語義見 backgroundMigrationManager.ts#L57-L76。自檢清單與運維注意事項綜合以上機制在 Langfuse 中落地一個新的后臺遷移建議按如下清單核查是否確實需要后臺遷移單次運行超過約 5 分鐘或非原子操作 → 是否則請走標準 Prisma migration。注冊完備性Prisma 遷移 SQL 已插入background_migrations行腳本文件已存在于 worker/src/backgroundMigrations默認導(dǎo)出實現(xiàn)了 IBackgroundMigration腳本內(nèi)硬編碼的 UUID 與 SQL 中的id一致。名稱可排序name以日期/序號前綴如20260701_v4_step_5_...確保執(zhí)行順序可預(yù)期?？苫謴?fù)性單庫內(nèi)逐步原子提交跨庫/跨系統(tǒng)PG ? ClickHouse冪等進度寫入state游標或分塊狀態(tài)。并發(fā)友好不依賴數(shù)據(jù)庫靜態(tài)狀態(tài)run()內(nèi)每輪檢查isAborted不調(diào)用應(yīng)用層業(yè)務(wù)代碼。門控如需休眠args.envGate以LANGFUSE_BACKGROUND_MIGRATION_前綴命名已在worker/src/env.ts的EnvSchema中注冊z.enum([true,false]).default(false)。依賴鏈如需下游遷移在validate()中調(diào)用checkPredecessorMigrationFinalized守衛(wèi)上游上游失敗會傳遞性阻斷整條鏈。變更溝通若應(yīng)用代碼依賴遷移完成需在 changelog 與部署文檔中顯式標注參考 GitLab 的 upgrade stops 溝通方式。運行時驗證本地用dotenv -e ../.env -- npx tsx src/backgroundMigrations/script-name.ts單跑驗證staging 可用--concurrency/--maxRetries/--retry-failed/--partitions控制執(zhí)行面生產(chǎn)環(huán)境確認LANGFUSE_ENABLE_BACKGROUND_MIGRATIONStrue且 worker 心跳15s 刷新lockedAt正常。故障排查遷移卡住時檢查background_migrations行的lockedAt60 秒內(nèi)視為鎖有效、failedReason、state中的游標失敗后清除failedAt可讓管理器重試該行分塊遷移可清failedAt后以--retry-failed重跑失敗分塊。通過這套機制Langfuse 得以在不中斷事件流處理的前提下安全地完成從填充可選列到整表跨系統(tǒng)重寫的各種重型數(shù)據(jù)遷移并將失敗恢復(fù)、多 worker 接管、分階段發(fā)布等運維復(fù)雜度收斂在 backgroundMigrationManager.ts 與 utils/backfillBase.ts 兩處基礎(chǔ)設(shè)施之中?！久赓M下載鏈接】langfuse Open source AI engineering platform: LLM evals, observability, metrics, prompt management, playground, datasets. Integrates with OpenTelemetry, LangChain, OpenAI SDK, LiteLLM, and more. YC W23項目地址: https://gitcode.com/GitHub_Trending/la/langfuse創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考