fhEVM Relayer 请求状态机与最新版 SQL 查询设计全解

fhEVM Relayer 请求状态机与最新版 SQL 查询设计全解【免费下载链接】fhevmFHEVM, a full-stack framework for integrating Fully Homomorphic Encryption (FHE) with blockchain applications项目地址: https://gitcode.com/GitHub_Trending/fh/fhevm本文以 relayer/docs/queries-latest-version.md 为骨架结合 fhEVM 仓库中 Relayer 的真实源码与数据库迁移脚本逐条还原“用户解密User Decrypt、公开解密Public Decrypt、输入证明Input Proof”三类请求从 POST 进入、排队、上链、监听回执、收齐份额/结果到最终 GET 返回的完整状态机与查询设计并给出可直接套用的建表 SQL、关键查询 SQL 与时间策略。阅读前提Relayer 在 fhEVM 中的角色fhEVM Relayer 是链接“用户/应用 HTTP 接口”与“网关链Gateway”之间的桥接进程其核心职责是接收解密/输入证明请求 → 做就绪检查与 ACL 校验 → 封装成链上交易发送给 Gateway → 监听链上事件回执 → 聚合份额或结果 → 供用户通过 GET 接口查询。所有中间状态都落库在 PostgreSQL 中因此“查询设计”本质上就是这套请求状态机在 SQL 层的完整映射。相关代码位于 relayer/src 与 relayer/relayer-migrate/migrations。一、三张核心表与统一状态枚举1. 请求状态枚举req_status建表脚本 relayer/relayer-migrate/migrations/20251109145104_create_tables.sql 定义了全项目统一的状态枚举CREATE TYPE req_status AS ENUM (queued, processing, tx_in_flight, receipt_received, completed, timed_out, failure);对应 Rust 侧枚举见 relayer/src/store/sql/models/req_status_enum_model.rs各状态语义如下状态语义queued请求刚入库等待就绪检查通过处于“就绪队列”processing就绪检查通过等待构造并发送 Gateway 交易处于“发送队列”tx_in_flight交易已提交到网关链等待回执receipt_received已收到链上交易回执gw_req_tx_hash、gw_reference_id已回填等待结果事件completed已收齐结果用户解密达到阈值份额 / 公开解密收到结果 / 输入证明被接受或拒绝可被 GET 查询timed_out超过超时窗口未完成带err_reasonfailure发送失败或链上失败带err_reason设计文档特别强调两条“最终态”约束relayer/docs/queries-latest-version.mdtimed_out与failure是终态之后的任何事件如迟到的份额都不会再改变行状态表结构层面用“部分唯一索引”保证同一int_job_id只能有一条活跃非终态记录这是幂等去重的第一道闸门。2. 三张请求表来自建表迁移-- 用户解密请求表 CREATE TABLE user_decrypt_req( id SERIAL PRIMARY KEY, ext_job_id UUID NOT NULL UNIQUE, -- 对外暴露的请求 IDuuid v4/v7 int_job_id BYTEA NOT NULL UNIQUE, -- 内部作业 ID32 字节哈希友好 gw_reference_id BYTEA, -- 网关侧引用 ID收到回执后回填 req JSONB NOT NULL, -- 请求载荷EIP-712 签名等 req_status req_status NOT NULL DEFAULT queued, gw_req_tx_hash TEXT, gw_consensus_tx_hash TEXT, -- 用户解密特有共识交易哈希 err_reason TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); -- 用户解密份额表每份额一行 CREATE TABLE user_decrypt_share ( id SERIAL PRIMARY KEY, gw_reference_id BYTEA NOT NULL, tx_hash TEXT, share_index INTEGER NOT NULL, share TEXT NOT NULL, kms_signature TEXT NOT NULL, extra_data TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); -- 公开解密请求表 CREATE TABLE public_decrypt_req( id SERIAL PRIMARY KEY, ext_job_id UUID NOT NULL UNIQUE, int_job_id BYTEA NOT NULL UNIQUE, gw_reference_id BYTEA, req JSONB NOT NULL, res JSONB, -- 公开解密结果JSON req_status req_status NOT NULL DEFAULT queued, gw_req_tx_hash TEXT, gw_response_tx_hash TEXT, err_reason TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); -- 输入证明请求表 CREATE TABLE input_proof_req( id SERIAL PRIMARY KEY, ext_job_id UUID NOT NULL UNIQUE, int_request_id UUID NOT NULL UNIQUE, -- uuid v7每个内部请求都对应一次网关请求 gw_reference_id BYTEA, accepted BOOLEAN DEFAULT null, -- 输入证明特有的接受/拒绝标记 req JSONB NOT NULL, res JSONB, req_status req_status NOT NULL DEFAULT queued, gw_req_tx_hash TEXT, gw_response_tx_hash TEXT, err_reason TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() );关键差异源自文档注释解密类请求使用int_job_idBYTEA输入证明使用int_request_idUUID v7——因为“每个内部请求都会触发一次网关请求”UUID v7 本身时间有序天然适合 B-Tree 索引。建表脚本同时通过trigger_set_timestamp()触发器让所有updated_at在 UPDATE 时自动刷新。3. 索引设计与设计文档中的 TODO建表脚本中为三类请求统一建立了idx_*_ext_job_idext_job_id的 HASH 索引等值查询支持 ON CONFLICT 幂等部分唯一索引idx_*_unique_int_job_id_partialWHERE req_status NOT IN (failure,timed_out)保证同一作业 ID 只有一条活跃记录idx_*_gw_reference_id网关引用 ID 索引供监听器按gw_reference_id更新行三个超时检查索引idx_*_timeout_checkWHERE req_status receipt_received的部分索引专门服务“回执后超时”扫描用户份额唯一索引idx_user_decrypt_share_unique_comp_gw_reference_id_share_index ON user_decrypt_share (gw_reference_id, share_index)用于份额幂等同一share_index只入一次。设计文档保留的 TODOrelayer/docs/queries-latest-version.md也值得注意早期草案曾考虑把internal_decryption_id改为统一命名并以 BYTEA 存储以启用哈希索引、将tx_sent更名为receipt_received以表意更清晰——这些在最终实现里均已落地字段名为int_job_id、状态为receipt_received并在代码注释中留下了“按最小字段返回”“统一internal_req_id/external_req_id结构”等设计纪律。二、启动恢复接管上次留下的状态设计文档第一节“NEEDED DATA STRUCT”规定了 Relayer 启动时的两张“续接”查询relayer/docs/queries-latest-version.md按updated_at排序查询所有处于queued的请求跨三张表混合返回int_indexer_id与req交由就绪检查器继续按updated_at排序查询所有处于processing的请求交由交易发送辅助流程直接继续就绪检查已通过。实现上该逻辑位于 relayer/src/startup_recovery.rs 与 relayer/src/sweep.rs。启动时先执行 init_status_counts_from_db 把三张表的各状态计数灌入指标Gauge随后由 sweep 用claim_incomplete_requests按owner_epoch与FOR UPDATE SKIP LOCKED原子认领未完成行queued行重新派发ReqRcvdFromUser事件重新走就绪检查processing行重新派发ReadinessCheckPassed事件跳过就绪检查直接发送tx_in_flight在认领的同一 UPDATE 中被改写为processing后派发receipt_received不恢复——交给网关监听器自然接管。对应认领 SQLrelayer/src/store/sql/repositories/user_decrypt_repo.rsWITH claimed AS ( SELECT id FROM user_decrypt_req WHERE req_status IN (queued::req_status, processing::req_status, tx_in_flight::req_status) AND owner_epoch $1 ORDER BY owner_epoch, id LIMIT $2 FOR UPDATE SKIP LOCKED ) UPDATE user_decrypt_req SET owner_epoch $1, req_status CASE WHEN req_status tx_in_flight::req_status THEN processing::req_status ELSE req_status END FROM claimed WHERE user_decrypt_req.id claimed.id RETURNING user_decrypt_req.int_job_id, user_decrypt_req.req, user_decrypt_req.req_type, user_decrypt_req.req_status;三、用户解密User Decrypt完整流程1. POST 请求幂等插入与就绪检查文档规定的 v2 POST 流程relayer/docs/queries-latest-version.md从载荷计算int_job_id查user_decrypt_req已存在 → 直接返回ext_job_id202 OK不存在 → 异步调用主机 ACL 就绪检查器v2 流程中当前为“DUMMY 恒通过”后续将由 host listener/substreams/poller 实现真实检查检查失败 → 400Not ready For Decryption成功 → 用ON CONFLICT插入req、ext_job_id、int_job_id冲突时取回既有ext_job_idv2 路由返回ext_req_id202 Created。核心插入 SQL 见 insert_data_on_conflict_and_get_ext_job_id其中用(xmax 0)判断本次 INSERT 是真插入还是冲突WITH ins AS ( INSERT INTO user_decrypt_req (ext_job_id, int_job_id, req, req_type, owner_epoch) VALUES ($1, $2, $3, $4::user_decrypt_req_type, $5) ON CONFLICT (int_job_id) WHERE req_status NOT IN (failure::req_status, timed_out::req_status) DO UPDATE SET updated_at user_decrypt_req.updated_at RETURNING ext_job_id, (xmax 0) AS is_inserted, req_status, gw_reference_id ) SELECT ins.ext_job_id, ins.is_inserted, ins.req_status, ins.gw_reference_id, (SELECT COUNT(*) FROM (SELECT 1 FROM user_decrypt_req q WHERE q.req_status queued::req_status LIMIT $6) depth) AS readiness_queue_size, (SELECT COUNT(*) FROM (SELECT 1 FROM user_decrypt_req q WHERE q.req_status processing::req_status LIMIT $6) depth) AS tx_queue_size FROM ins;同一语句内还顺带统计了两条队列的深度readiness_queue_size、tx_queue_size用于 POST 响应中的排队位置预估返回的UserDecryptInsertResult区分Inserted/DuplicateCompleted重复但已完成直接组装已聚合的份额响应/DuplicateProcessing重复仍在处理中。2. 就绪通过 → 发送状态推进到processing网关就绪检查器以大超时窗口运行约 30 分钟覆盖就绪后按int_job_id把状态推进到processing并触发Transaction::Send30 分钟仍未就绪则置为timed_out并填err_reason同时发出UserDecrypt::Failed事件relayer/docs/queries-latest-version.md。状态推进使用“CTE 捕获旧状态 条件 UPDATE”的写法update_status_to_processingWITH old AS ( SELECT req_status, updated_at FROM user_decrypt_req WHERE int_job_id $1 AND req_status NOT IN (failure::req_status, timed_out::req_status) ), upd AS ( UPDATE user_decrypt_req SET req_status processing::req_status, owner_epoch $2 WHERE int_job_id $1 AND req_status queued::req_status AND owner_epoch $2 RETURNING req_status, updated_at ) SELECT old.req_status, old.updated_at, upd.updated_at FROM old, upd;此写法贯穿所有状态迁移一次查询同时返回新旧状态与新旧updated_at便于记录状态迁移指标metrics::record_status_transition。owner_epoch条件实现 epoch 围栏仅当前调度 epoch 拥有该行时才允许推进防止多副本竞争写同一行。3. 交易发送与回执Transaction::Send发出后交易处理器TxHandler分两路更新relayer/docs/queries-latest-version.mdTransaction::Success拿到回执更新为receipt_received同时回填gw_req_tx_hash与gw_reference_idTransaction::Failed置为failure并填err_reason向编排器派发错误事件。对应 SQLupdate_status_to_receipt_received_on_tx_success中的 UPDATE 子句UPDATE user_decrypt_req SET req_status receipt_received::req_status, gw_req_tx_hash $1, gw_reference_id $2, owner_epoch $4 WHERE int_job_id $3 AND req_status tx_in_flight::req_status AND owner_epoch $4即只有处于tx_in_flight的行才允许收到回执从状态机角度保证了“先有交易在途才可能收到回执”。4. 监听器共识哈希与份额聚合共识交易哈希relayer/docs/queries-latest-version.md收到consensus reached事件时按gw_reference_id更新gw_consensus_tx_hash但只在gw_consensus_tx_hash IS NULL且状态为receipt_received/completed时写无论是否写成功都返回(status, updated_at, err_reason, int_job_id)——因为即使该行已timed_out调用方仍需要拿到真实状态。实现为update_consensus_hash_and_return_staterelayer/src/store/sql/repositories/user_decrypt_repo.rsWITH target AS ( SELECT id FROM user_decrypt_req WHERE gw_reference_id $2 ), updated_row AS ( UPDATE user_decrypt_req SET gw_consensus_tx_hash $1 WHERE id (SELECT id FROM target) AND gw_consensus_tx_hash IS NULL AND req_status IN (receipt_received::req_status, completed::req_status) RETURNING req_status, updated_at, err_reason, int_job_id ) SELECT req_status, updated_at, err_reason, int_job_id FROM updated_row UNION ALL SELECT req_status, updated_at, err_reason, int_job_id FROM user_decrypt_req WHERE id (SELECT id FROM target) AND NOT EXISTS (SELECT 1 FROM updated_row);份额插入与阈值完成relayer/docs/queries-latest-version.md每个份额事件对应“两次调用/两个事务”INSERT INTO user_decrypt_share (gw_reference_id, share_index, share, kms_signature, extra_data)返回该gw_reference_id的份额总数若count threshold则同一事务内把user_decrypt_req状态置为completed仅当status ! timed_out等非终态并返回全部份额 int_job_idstatusupdated_aterr_reason。实现insert_share_and_complete_if_threshold_reachedrelayer/src/store/sql/repositories/user_decrypt_repo.rs的核心手法INSERT INTO user_decrypt_share (gw_reference_id, tx_hash, share_index, share, kms_signature, extra_data) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (gw_reference_id, share_index) DO NOTHING; -- 份额幂等 SELECT COUNT(*) FROM user_decrypt_share WHERE gw_reference_id $1; -- 同事务内计数 -- 若达到阈值同事务完成 WITH old AS ( SELECT req_status, updated_at FROM user_decrypt_req WHERE gw_reference_id $1 ), upd AS ( UPDATE user_decrypt_req SET req_status completed::req_status, resolved_threshold $2 WHERE gw_reference_id $1 AND req_status receipt_received::req_status RETURNING int_job_id, req_status, updated_at, err_reason ) SELECT old.req_status, old.updated_at, upd.int_job_id, upd.new_status, upd.updated_at, upd.err_reason FROM old LEFT JOIN upd ON true;同时整个事务先执行SELECT pg_advisory_xact_lock(...)按gw_reference_id计算锁 ID串行化同一引用的份额写入避免并发 INSERT 撞唯一索引浪费序列号也避免计数失真与“份额插入与完成之间被 pg_cron 超时任务插队”的竞态。返回值ShareCompletionOutcome明确区分ThresholdNotReached/Completed/AlreadyCompleted重复份额/AlreadyInFinalState终态不可再完成。文档特别标注这里存在“非相关份额non-relevant shares”的可能且即使请求已超时监听到的迟到份额也仍会登记入表只影响计数不再改变终态此行为应在代码中显式注释说明。5. GET 请求按状态构建响应GET 以ext_req_id查询relayer/docs/queries-latest-version.md按ext_job_id查user_decrypt_req并 JOINuser_decrypt_share取回该gw_reference_id的全部份额1 条查询完成需返回状态字段 份额 updated_at。响应按状态分支completed→ 200构造含份额的完整响应processing→ 202返回updated_atext_request_id 状态queued/receipt_received→ 返回ext_job_id 状态 updated_attimed_out→ 504返回ext_req_id 状态failure→ 400返回ext_req_id 状态 err_reason。实现见 find_req_and_shares_by_ext_job_id份额用jsonb_agg聚合为 JSON 数组并按COALESCE(resolved_threshold, $2)决定拉取多少条份额完成时动态阈值优先缺失时回退静态阈值同时计算queue_position同状态队列中排在自己前面的数量与tx_queue_sizeSELECT r.ext_job_id, r.req_status, r.updated_at, r.err_reason, r.gw_req_tx_hash, r.gw_consensus_tx_hash, r.resolved_threshold, COALESCE(jsonb_agg(jsonb_build_object( share, s.share, kms_signature, s.kms_signature, extra_data, s.extra_data ) ORDER BY s.share_index) FILTER (WHERE s.id IS NOT NULL), []::jsonb) AS shares, (SELECT COUNT(*) FROM ( SELECT 1 FROM user_decrypt_req q WHERE q.req_status r.req_status AND q.req_status IN (queued::req_status, processing::req_status, tx_in_flight::req_status) AND q.id r.id LIMIT $3 ) ahead) AS queue_position, (SELECT COUNT(*) FROM ( SELECT 1 FROM user_decrypt_req q WHERE q.req_status processing::req_status LIMIT $3 ) depth) AS tx_queue_size FROM user_decrypt_req r LEFT JOIN (SELECT * FROM user_decrypt_share WHERE gw_reference_id IN (SELECT gw_reference_id FROM user_decrypt_req WHERE ext_job_id $1) ORDER BY created_at ASC, share_index ASC LIMIT COALESCE((SELECT resolved_threshold FROM user_decrypt_req WHERE ext_job_id $1), $2) ) s ON r.gw_reference_id s.gw_reference_id WHERE r.ext_job_id $1 GROUP BY r.id;四、公开解密Public Decrypt流程公开解密与用户解密高度同构relayer/docs/queries-latest-version.mdPOST计算int_job_id→ 查public_decrypt_req已存在返回ext_req_id202否则调用主机 ACL 就绪检查同样当前为 DUMMY 恒通过成功则ON CONFLICT插入v2 返回 202 Created新 API 下若已存在可直接返回结果与状态网关就绪 发送就绪后置processing触发Transaction::Send未就绪超时置timed_out并发PublicDecrypt::Failed交易回执Success →receipt_received 回填gw_req_tx_hash/gw_reference_idFailed →failureerr_reason监听器完成relayer/docs/queries-latest-version.md收到公开解密结果事件后按gw_reference_id更新res 收到的结果值、状态置completed仅当status ! timed_out返回(int_job_id, status, updated_at, err_reason)。公开解密不需要份额表——结果是一个整体res JSONB因此不存在阈值聚合完成逻辑为单条 UPDATERETURNING见 relayer/src/gateway/public_decrypt_handler.rs 与仓库中的 public_decrypt_repo.rs。GET按ext_reference_id查询public_decrypt_req需返回status、res、err_reason、updated_at、ext_request_id分支同上completed→ 200processing→ 202queued/receipt_received→ 返回状态与updated_attimed_out→ 504failure→ 400 err_reason。公开解密与用户解密的仓库均实现为“epoch 围栏写、监听器不围栏”模式凡是“发送决策”类状态写都带owner_epoch条件仅当前调度 epoch 可写而“记录链上事实”的监听器路径如complete_req_with_res、update_consensus_hash_and_return_state刻意不加围栏——它们不是发送决策链监听器不受 epoch 门控且状态守卫已保证幂等。详见 public_decrypt_repo.rs 模块注释。五、输入证明Input Proof流程输入证明没有就绪检查relayer/docs/queries-latest-version.md流程最简POSTrelayer/docs/queries-latest-version.md创建 uuidV7 的int_request_id插入ext_reference_id、int_request_id、request到input_proof_reqv2 路由返回ext_reference_id202 Created发送与回执触发Transaction::SendSuccess →receipt_received 回填gw_req_tx_hash/gw_reference_idFailed →failureerr_reason监听器完成relayer/docs/queries-latest-version.md证明被接受更新res 网关事件中的接收值req_status completed、accepted true并回填gw_response_tx_hash返回int_request_id证明被拒绝accepted false、req_status completed、gw_response_tx_hash 事件交易哈希同时写入resGETrelayer/docs/queries-latest-version.md按ext_reference_id查input_proof_req返回status、response、err_reason、updated_at、accepted、req_statuscompleted→ 200返回响应与acceptedqueued/receipt_received→ 返回ext_req_id 状态 updated_ataccepted nulltimed_out→ 504failure→ 400 err_reason。实现位于 relayer/src/store/sql/repositories/input_proof_repo.rs 与 relayer/src/gateway/input_handlers.rs。由于输入证明每个内部请求都对应一次网关请求重放恢复时统一重新派发ReqRcvdFromUser事件relayer/src/startup_recovery.rs不做状态分支。六、超时策略与 pg_cron 后台任务设计文档中反复出现同一句超时规则relayer/docs/queries-latest-version.md若status receipt_received且now - updated_at 30 min按int_indexer_id更新req_status timed_out、err_reason response timed out与 POST 模式中的查询相同。以 pg_cron 每分钟执行后期落地。最终实现把“pg_cron”落地为 Relayer 内置的两个后台 workerrelayer/src/store/sql/repositories/cron_task.rs超时 workercreate_timeout_worker_future按timeout_cron_interval周期调用 time_out_stale_requests对三张表分别执行同构的 CTE 超时 UPDATEWITH stale_rows AS ( SELECT id, updated_at FROM user_decrypt_req WHERE req_status receipt_received::req_status AND updated_at NOW() - make_interval(secs $1) FOR UPDATE SKIP LOCKED ), updated_rows AS ( UPDATE user_decrypt_req SET req_status timed_out::req_status, err_reason $2, updated_at NOW() FROM stale_rows WHERE user_decrypt_req.id stale_rows.id RETURNING user_decrypt_req.updated_at, stale_rows.updated_at ) SELECT * FROM updated_rows;FOR UPDATE SKIP LOCKED保证多副本下同一行只被一个 worker 处理错误原因统一为Gateway chain did not respond within the expected timeframe三张表的超时阈值分别由cron_config.user_decrypt_timeout/public_decrypt_timeout/input_proof_timeout控制默认约 30 分钟量级可配置仅确认的调度者dispatcher执行超时写避免两个副本互相冲突。过期清理 workercreate_expiry_worker_future调用purge_stale_data定期清理超过保留期的历史行两个 worker 都挂在 DispatchGate 之后且崩溃后 5 秒超时/ 30 秒过期自动重启。七、设计要点回顾与落地对照设计文档要点最终实现落点启动时按状态续接queued/processingstartup_recovery.rs claim_incomplete_requestssweep 认领POST 幂等ON CONFLICT 冲突取回ext_job_idinsert_data_on_conflict_and_get_ext_job_id三类仓库均有就绪通过置processing、未通过超时置timed_outupdate_status_to_processing/update_status_to_timed_out回执后置receipt_received并回填哈希与引用 IDupdate_status_to_receipt_received_on_tx_success用户解密份额插入 阈值完成两事务insert_share_and_complete_if_threshold_reachedadvisory lock 同事务完成共识哈希仅在receipt_received且为空时更新但始终返回真实状态update_consensus_hash_and_return_stateGET 按状态分支返回200/202/504/400find_req_and_shares_by_ext_job_id/ 对应公开解密、输入证明查询pg_cron 每分钟超时扫描TimeoutRepository::time_out_stale_requestsRelayer 内置 worker 周期执行命名统一BYTEAint_job_id、receipt_received状态20251109145104_create_tables.sql 及后续迁移延伸阅读完整的表结构与索引见 relayer/relayer-migrate/migrations/20251109145104_create_tables.sql状态迁移与 SQL 指标可见 relayer/src/metrics/docs_and_dashboards/sql_metrics.md端到端行为可由 relayer/tests/user_decrypt_v2_test.rs、relayer/tests/public_decrypt_v2_test.rs、relayer/tests/input_proof_v2_test.rs 等集成测试验证幂等性设计见 relayer/docs/idempotency-audit.md。【免费下载链接】fhevmFHEVM, a full-stack framework for integrating Fully Homomorphic Encryption (FHE) with blockchain applications项目地址: https://gitcode.com/GitHub_Trending/fh/fhevm创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考