即时通讯后端【免费下载链接】WuKongIMMore than just IM 不只是即时通讯(IM)项目地址https://gitcode.com/gh_mirrors/wu/WuKongIM点击查看免费下载导读本文以 WuKongIM 仓库中docs/superpowers/plans/2026-05-13-send-cmd-conversation-intent-p2d-d.md这份实现计划为主体系统讲解 P2d-d 阶段如何用CMD 会话更新意图Conversation Intent UID 属主本地待处理缓冲Pending Buffer替代 P2d-c 的 post-commit 订阅者扫描投影从而在不引入持久化订阅者快照表的前提下恢复旧版兼容的 CMD 离线同步语义。读完本文你将掌握意图构建规则、待处理缓冲的合并/落盘/恢复机制、/message/sync待处理覆盖层与/message/syncack清理契约、UID-owner 路由与节点 RPC 适配以及当前仓库中该模型后续演进为UserCMDChannelMembership成员关系模型的实现现状。背景与目标为什么需要 P2d-dWuKongIM 的 CMD命令频道承载系统命令类消息。历史实现中ConversationManager并不会持久化每条消息的订阅者快照而是借助 distribute 路径已经解析好的 delivery tag把 tag 的本地 UID 分区转换成待处理会话更新pending conversation update周期性冲刷为ConversationTypeCMD行优雅停机会把未冲刷的待处理更新保存到本地 JSON 文件并在启动时恢复。这正是 P2d-d 要恢复的旧版兼容模型其完整链路可以概括为resolved CMD recipients - CMD conversation update intent - UID-owner pending buffer - periodic flush to CMDConversationState - /message/sync overlays pending persisted state在 P2d-c 阶段参见 P2d-c 实现计划仓库已经引入了专用的CMDConversationState表、UID 属主的/message/sync与/message/syncack接口以及一个cmdsync.Projector。但当时的投影模型仍然把 CMD 更新建模为提交后的订阅者扫描/投影MessageCommitted - cmdsync.Projector - page SubscriberResolver after commit - write CMDConversationState directly这带来两个遗留问题request-scoped durable 发送仍然依赖请求局部的MessageScopedUIDs进程重启后的 committed replay 无法从频道日志重建出当时的请求接收者范围未冲刷的 CMD 活跃状态没有以旧式待处理会话更新形态暴露给/message/sync。P2d-d 的目标对应 P2d-d 设计文档是使用已经解析好的接收者来源而非提交后二次扫描构建 CMD 会话更新新增 UID 属主本地的待处理 CMD 会话更新器支持优雅停机时文件恢复把待处理 CMD 会话合并进/message/sync候选选择在/message/syncack时清理待处理 UID/频道条目在写入待处理缓冲之前先把意图路由到 UID 属主用请求接收者 delivery UID 页的意图写入替换 P2d-c 的cmdsync.Projector订阅者扫描写入保持普通/conversation/sync与 CMD 待处理/状态完全隔离。明确不在 P2d-d 范围内订阅者快照持久化、频道日志格式变更、kill -9 意图持久化、/message/sendbatch、expire、AllowStranger、插件/Webhook/AI 行为以及任何绕过集群/UID-属主语义的本地捷径——单节点部署依然是一个单节点集群。核心数据结构从意图到待处理视图P2d-d 在internal/usecase/cmdsync中定义了三层数据结构计划文档中的完整定义// ConversationIntent describes one CMD conversation update before it is flushed. type ConversationIntent struct { CommandChannelID string ChannelType uint8 MessageSeq uint64 ActiveAt int64 SenderUID string UserReadSeqs map[string]uint64 } // PendingConversationUpdate is the owner-local buffered update for one CMD channel. type PendingConversationUpdate struct { CommandChannelID string json:command_channel_id ChannelType uint8 json:channel_type LastMsgSeq uint64 json:last_msg_seq ActiveAt int64 json:active_at UserReadSeqs map[string]uint64 json:user_read_seqs } // PendingConversationView is the pending overlay for one UID and CMD channel. type PendingConversationView struct { CommandChannelID string ChannelType uint8 LastMsgSeq uint64 ActiveAt int64 ReadSeq uint64 }各字段的语义与校验规则CommandChannelID必须是source____cmd形态非命令频道 ID 会在构建意图时被归一化追加____cmd后缀MessageSeq必须非零NoPersist消息与 seq 为零的实时消息永远不会产生意图UserReadSeqs只允许包含非空 UID重复 UID 在入队前折叠只有当发送者出现在解析后的接收者集合中时发送者 UID 才获得MessageSeq作为初始读进度接收者一律从0开始ActiveAt优先取消息时间戳否则取注入时钟测试应注入时钟而不是依赖墙钟时间。计划中还明确了CommandChannelKey的形状ChannelID stringChannelType uint8并强调所有频道级待处理操作都必须同时以CommandChannelID与ChannelType为键不能只按频道 ID 操作。值得指出的是当前仓库的internal/usecase/cmdsync/types.go中CommandChannelKey结构与计划完全一致但持久化模型已从计划中的CMDConversationState演进为UserCMDChannelMembership见下文仓库现状章节。意图构建器的规则BuildConversationIntent(message, recipientUIDs, now)是意图的唯一入口负责校验消息是否为 durable 的 CMD 投影消息把source归一化为source____cmd去除空 UID 与重复 UID仅当发送者在归一化 UID 集合中时为发送者分配MessageSeq读进度空接收者集合返回无意图。计划中的测试TestBuildConversationIntentUsesScopedUIDsAndSenderReadSeq展示了完整语义对一条ChannelID: g1____cmd、MessageSeq: 9、FromUID: u1、SyncOnce: true的组频道消息输入接收者[u2, u1, u2, ]后得到的意图UserReadSeqs应为{u1: 9, u2: 0}——即发送者自读、接收者零起点、重复与空 UID 被折叠。另有TestBuildConversationIntentNormalizesSyncOnceSourceChannel验证g1会被归一化为g1____cmd以及TestBuildConversationIntentRejectsNoPersistZeroSeqOrEmptyRecipients验证NoPersist、零 seq、空接收者均被拒绝。UID-owner 待处理更新器合并、冲刷与文件恢复接口与选项计划为待处理更新器定义了 owner-local 生命周期组件type ConversationUpdater interface { Start() error Stop() error PushIntent(ctx, intent) error ListPending(ctx, uid, limit) []PendingConversationView MarkSynced(ctx, uid, commandChannelKey, throughSeq) error Flush(ctx) error }其构造选项ConversationUpdaterOptions包括持久化Store、DataDir、FlushInterval、FlushBatchSize、ShardCount、注入时钟Now与日志器Logger。存储接口在types.go中扩展为三条契约type StateStore interface { ListCMDConversationActive(ctx context.Context, uid string, limit int) ([]metadb.CMDConversationState, error) AdvanceCMDConversationReadSeq(ctx context.Context, patches []metadb.CMDConversationReadPatch) error UpsertCMDConversationStates(ctx context.Context, states []metadb.CMDConversationState) error } type ConversationPendingStore interface { ListPending(ctx context.Context, uid string, limit int) []PendingConversationView MarkSynced(ctx context.Context, uid string, key CommandChannelKey, throughSeq uint64) error }新增UpsertCMDConversationStates的目的在于/message/syncack可以为仅存在于待处理中的 CMD 频道先创建 durable 读进度再移除待处理条目。分片与倒排索引实现细节上更新器按命令频道键分片fasthash.Hash(commandChannelID) % shardCount或本地哈希助手以降低锁竞争每个分片维护两个结构pendingByChannel[channelKey] - PendingConversationUpdate userIndex[uid] - set(channelKey)userIndex复刻了旧版缓存的形态让/message/sync可以按单个 UID 发现未冲刷的 CMD 频道而无需扫描整个缓冲。合并规则单调、幂等同一命令频道上的多条意图在缓冲内按单调最大值合并LastMsgSeq max(existing.LastMsgSeq, intent.MessageSeq)ActiveAt max(existing.ActiveAt, intent.ActiveAt)每个 UID 的UserReadSeqs[uid] max(existingReadSeq, newReadSeq)接收者的0更新绝不能回退发送者或已 ack 的读进度。幂等性保证了 request-scoped 直接意图路由与 delivery observer 重试可以重叠而不会产生重复的未读状态。计划测试TestPendingUpdaterCoalescesByCommandChannelAndUID验证了两条意图seq 5 与 seq 7合并后LastMsgSeq7、ActiveAt70、ReadSeq5的结果。冲刷与失败语义冲刷把待处理 UID/频道条目转换为CMDConversationStatestate : metadb.CMDConversationState{ UID: uid, ChannelID: update.CommandChannelID, ChannelType: int64(update.ChannelType), ReadSeq: readSeq, ActiveAt: update.ActiveAt, UpdatedAt: u.now().UnixNano(), }冲刷规则非常强调整批失败/整批成功的一致性以有界批次通过既有的全有或全无UpsertCMDConversationStates(ctx, []state) errorAPI 写入一批成功后才移除该批包含的所有 UID/频道条目同一命令频道上其他仍待处理的 UID 不得被一并移除某批失败时该批所有条目保持待处理并返回错误——因为存储 API 不暴露逐行结果绝不能假称逐行成功若前面的批次已成功而后续批次失败只移除已成功批次条目Stop()停止 ticker 后执行最后一次有界Flush(context.Background())再把剩余待处理更新保存到文件。计划中的两个测试TestPendingUpdaterKeepsWholeFailedFlushBatch与TestPendingUpdaterRemovesSuccessfulFlushBatch分别验证了失败保留与成功移除的对称语义。文件保存/加载与优雅恢复待处理文件路径为DataDir/conversationv2/cmd_conversation_updates.json保存把所有分片的更新序列化为[]PendingConversationUpdate先写入cmd_conversation_updates.json.tmpClose后再os.Rename原子替换避免半写文件覆盖上次完整文件加载在Start()进入冲刷循环之前加载成功后删除文件解码失败把文件重命名为cmd_conversation_updates.json.bad或带时间戳的.bad记录日志后返回 nil把坏文件留给运维排查绝不 panic。TestPendingUpdaterSaveLoadRoundTrip验证了 Push → Stop保存→ Start加载后ListPending结果一致的完整闭环。重要边界这仍不保证 kill -9 前新接收的内存意图被持久化——P2d-d 遵循旧版优雅停机持久化模型完整的崩溃安全恢复需要另行设计意图 WAL 或复制副作用日志见 P2d-d 设计文档 的 Rollout Notes。/message/sync待处理覆盖层与/message/syncack清理Sync合并持久化状态与待处理状态cmdsync.App.Sync在 P2d-c 的持久化CMDConversationState列表基础上合并 owner-local 待处理视图。合并算法为effectiveReadSeq : maxUint64(maxUint64(persisted.ReadSeq, persisted.DeletedToSeq), pending.ReadSeq) activeAt : maxInt64(persisted.ActiveAt, pending.ActiveAt)关键规则待处理接收者读进度0只负责激活频道不得覆盖持久化的读进度待处理发送者读进度等于MessageSeq时在没有持久化状态的情况下防止发送者自己的 CMD 消息被当作未读返回待处理表明频道活跃但按effectiveReadSeq 1拉取不到任何新消息时不为该频道创建 sync record持久化候选与待处理候选共享 P2d-c 的确定性排序activeAt降序、ChannelType、CommandChannelID与全局 limit 规则必须在拉取足够满足全局响应上限的消息之前决定候选顺序普通/conversation/sync绝不使用此待处理缓冲。计划中的TestAppSyncMergesPendingCMDConversationOverlay与TestAppSyncUsesMaxPersistedAndPendingReadSeq分别验证了合并后返回频道消息与读进度取持久化/待处理最大值5 vs 0 → fromSeq6的语义。SyncAck先落盘读进度再清理待处理/message/syncack的清理算法计划中的完整清理块强调一个核心契约只有在 durable 读进度已存在或刚成功 upsert 之后才允许清理待处理。步骤为从 sync records 构建有效读补丁validRecords对既有 durable 行调用AdvanceCMDConversationReadSeq(ctx, patches)若a.pending ! nil按 UID 列出待处理视图并构建(CommandChannelID, ChannelType)查找表对每个有匹配待处理视图的有效 record调用UpsertCMDConversationStates写入至少包含UID、ChannelID、ChannelType、ReadSeqLastReturnedMsgSeq、ActiveAtpending.ActiveAt、UpdatedAtnow的状态——当上一次 advance 因频道仅存在于待处理中而 no-op 时这一步创建 durable 行同时避免把 ack 时间伪装成消息活跃时间只有 upsert 成功后才调用MarkSynced清理该 UID/频道durable upsert 失败时不调用MarkSynced直接返回错误让待处理条目继续作为可重试的进度来源MarkSynced在 durable 进度已存在后失败仅记录告警并返回 nil——清理是 best effort因为读进度已经 durable。清理规则本身ackSeq pending.LastMsgSeq时从该待处理更新中移除该 UID该频道待处理更新不再有 UID 时移除频道与索引条目ackSeq低于LastMsgSeq时保留待处理条目供后续 sync。持久化读进度推进必须保持单调待处理清理绝不能成为唯一的读进度来源。相关测试包括TestAppSyncAckCleansPendingAfterPersistedAdvance、TestAppSyncAckTreatsPendingCleanupFailureAsBestEffort与TestAppSyncAckPersistsPendingOnlyReadProgressBeforeCleanup。UID-owner 路由与节点 RPC意图如何到达属主节点路由分层意图在写入待处理缓冲之前必须路由到 UID 属主节点因为待处理缓冲必须与为该 UID 提供/message/sync的节点保持一致。路由规则设计文档明确定义slotID : cluster.SlotForKey(uid) owner : cluster.LeaderOf(slotID)cmdConversationIntentRouter.PushIntent的实现要点路由前先经 cmdsync 辅助函数校验意图MessageSeq0、空/非命令CommandChannelID、ChannelType0、空 UID map 一律在任何本地或远端写入之前拒绝把UserReadSeqs按ownerNodeID(uid)SlotForKeyLeaderOf分组每个属主组克隆一份只含该组 UID 的意图本地组调用local.PushIntent远端组调用remote.PushCMDConversationIntent仅当所有组都成功时返回fullyAcceptedtrue若某组失败返回false与合并错误——重复重试是安全的若远端组返回cmdsync.ErrConversationIntentStaleOwner属主已迁移重新解析该组 UID 属主并重试一次仍失败则返回false与合并错误。owner 校验包装与 stale-owner 语义节点 RPC 接收侧使用ownerValidatingCMDIntentSink维护 owner-local 不变量对意图UserReadSeqs中的每个 UID 验证ownerNodeID(uid) localNodeID不满足则返回cmdsync.ErrConversationIntentStaleOwner且不本地存储部分非法意图。调用方路由捕获该哨兵错误后可以重新解析属主并重试。节点 RPC 层把该哨兵映射为可重试状态如rpcStatusStaleOwner stale_owner客户端把该状态转换回满足errors.Is(err, cmdsync.ErrConversationIntentStaleOwner)的错误从而保留重试语义。缺失 provider 或意图畸形MessageSeq0、空/非命令频道、ChannelType0、无任何非空 UID时返回rpcStatusRejected。二进制编解码的确定性RPC 请求中的意图通过internal/access/node/cmdsync_codec.go追加/读取。UserReadSeqsmap 的编码必须确定性先对 UID 键排序再追加以保证网络往返字节一致uids : make([]string, 0, len(intent.UserReadSeqs)) for uid : range intent.UserReadSeqs { uids append(uids, uid) } sort.Strings(uids) dst appendUvarint(dst, uint64(len(uids))) for _, uid : range uids { dst appendString(dst, uid) dst appendUvarint(dst, intent.UserReadSeqs[uid]) }相关 RPC 测试覆盖了编解码往返、缺失 provider 拒绝、畸形意图拒绝、stale-owner 可重试状态以及客户端推送到远端属主TestCMDConversationIntentBinaryCodecRoundTrip、TestCMDIntentAdapterMissingProviderReturnsRejectedStatus、TestCMDIntentAdapterStaleOwnerIsRetryable、TestCMDIntentClientPushCallsRemoteOwner等。两类意图来源request-scoped 直投与 delivery UID 页观察者来源一request-scoped durable 发送直投对 durable request-scoped CMD 发送SendCommand.RequestSubscribers本身就是精确的业务接收者集合因此最安全的旧版兼容路径是在消息通道内直接构建意图而无需等待异步 delivery 重新发现同一份请求局部接收者列表message.App.sendDurable - append to temp____cmd channel - build CMD conversation intent from RequestSubscribers - route intent to UID owners - submit committed delivery envelope with MessageScopedUIDs and intent-submitted flag发送结果不能仅仅因为append 成功后 best-effort 会话意图路由失败而变成虚假的 durable 失败意图路由失败只记录日志并返回非致命副作用结果durable append 成功仍是消息写入的事实来源。为此引入内存事件/信封元数据标志CMDConversationIntentSubmitted bool规则设计文档仅当 request-scoped 意图的所有 UID-owner 分区都被本地待处理更新器或远端属主 RPC 接受时才置位该标志任一 owner 分区失败则保持false让 delivery UID 观察者基于MessageScopedUIDs做第二次 best-effort 尝试部分成功可接受因为已接受分区的重复提交是幂等的该标志不写入频道日志不期望在 committed replay 后存活delivery 观察者仅在CMDConversationIntentSubmittedtrue时跳过 request-scoped 意图生成P2d-d 不放松 request-scoped committed delivery 提交的既有严格性append 后若 message-scoped delivery 提交失败发送路径仍返回既有分发错误只有 CMD 会话意图副作用是非致命的。internal/usecase/message只调用 app 提供的意图 sinkCMDConversationIntentSink接口不包含任何集群路由逻辑这符合分层边界。来源二delivery 解析器 UID 页观察者对于普通 durable CMD 频道接收者集合来自 delivery 解析器路径而非独立的提交后订阅者扫描。在 app delivery 解析器组装附近新增窄观察者接口type resolvedUIDPage struct { Envelope deliveryruntime.CommittedEnvelope UIDs []string } type resolvedUIDObserver interface { OnResolvedUIDPage(context.Context, resolvedUIDPage) }关键约束是在 presence 扩展之前通知——因为 CMD 离线同步也需要离线用户而在线RouteKey会丢弃离线用户subscriber/tag UID page - observer receives UIDs - presence authority expands online routes - delivery runtime pushes online packets观察者调用是副作用绝不能使 delivery 解析失败观察者实现可以入队、记录失败、上报指标但 delivery 代码不应根据观察者成功与否分支重试应实现于意图路由或待处理更新器队列内部而不是向 delivery actor 返回错误。notifyResolvedUIDPage辅助函数明确跳过 nil 观察者、空 UID 页以及已完全提交的 request-scoped 意图env.CMDConversationIntentSubmitted len(env.MessageScopedUIDs) 0三种情况。对 tag 式 delivery使用tagDeliveryResolver当前正在扩展的 tag 本地 UID 页follower 只收到本地分区时只提交该分区意图本地节点同时是该分区 UID 属主时由路由直接写本地待处理拓扑变化或分区包含属主已迁移的 UID 时路由重新解析属主并正确转发。对本地解析器回退路径使用SubscriberResolver.NextPage返回的 UIDpresence 扩展前按 UID 属主完全相同的方式路由。钩子对非 CMD 消息与NoPersist消息必须是 no-op。测试TestTagDeliveryResolverNotifiesUIDObserverBeforePresenceExpansion验证了观察者在 presence 扩展前收到包含离线/在线用户的 UID 页TestDeliveryUIDObserverSkipsWhenRequestScopedIntentAlreadySubmitted验证已提交意图不会二次通知。回退测试还要求tagDeliveryResolver在 tags/subscribers 为空时委托内联localDeliveryResolver且 UID 观察者仍被调用防止回退路径丢失 CMD 意图。事件元数据与信封传播CMDConversationIntentSubmitted标志需要在整条 committed 链路上传播Task 6 的修改清单internal/contracts/messageevents/events.go的MessageCommitted增加该布尔字段Clone()必须保留internal/runtime/delivery/types.go的CommittedEnvelope增加同名字段internal/usecase/delivery/submit.go在MessageCommitted与运行时信封之间拷贝internal/access/node/delivery_submit_codec.go引入deliverySubmitRequestMagicV3 [...]byte{W, K, D, C, 3}当len(MessageScopedUIDs) 0或标志为 true 时编码 V3V2 字段 一个尾部布尔V1 解码得到空 scoped UIDs 与 false 标志V2 解码 scoped UIDs 与 false 标志V3 两者都解码——保持既有 V1/V2 兼容覆盖。组合装配、生命周期与 committed replay 清理app 组合根与生命周期计划要求把internal/app/app.go中旧的cmdSyncProjector字段/生命周期标志替换为具体运行时字段cmdConversationUpdater *cmdsync.ConversationUpdater cmdConversationIntents cmdConversationIntentRouter cmdConversationUpdaterOn atomic.Bool生命周期组件名从appLifecycleCMDSyncProjector cmdsync_projector替换为appLifecycleCMDConversationUpdater cmd_conversation_updater停止顺序应让 delivery 先停止再执行最终待处理保存若生命周期按启动逆序停止则应在 delivery runtime 之前启动更新器以便它在 delivery 排空后停止。internal/app/build.go的组装要点app.cmdConversationUpdater cmdsync.NewConversationUpdater(cmdsync.ConversationUpdaterOptions{ Store: app.store, DataDir: cfg.Node.DataDir, FlushInterval: cfg.Conversation.ActiveHintFlushInterval, // or a local default FlushBatchSize: cfg.Conversation.ActiveHintFlushBatchSize, Logger: app.logger.Named(cmdsync.pending), }) cmdIntentRouter : cmdConversationIntentRouter{local: app.cmdConversationUpdater, remote: app.nodeClient, cluster: app.cluster, localNodeID: cfg.Node.ID, logger: app.logger.Named(cmdsync.intent)} app.cmdConversationIntents cmdIntentRouter随后把待处理更新器传入cmdsync.New的Pending选项把观察者注入tagDeliveryResolver与回退localDeliveryResolver构造器tagDeliveryResolver.BeginResolve内联创建回退解析器时必须拷贝uidObserver把路由传入message.New的CMDConversationIntents选项把ownerValidatingCMDIntentSink传入accessnode.New选项。旧app.cmdSyncProjector构造块与 committed fanout 订阅者被移除——fanout 只保留 delivery dispatcher。committed replay 只走 delivery计划明确删除独立的CMDSync.ProjectCommitted订阅者扫描 replaycommitted_replay.go移除committedReplayCMDSync接口与CMDSync配置字段不再调用ProjectCommitted完全依赖重放的 delivery 触发 delivery UID 观察者来为普通 CMD 消息重新产生意图。文档保留一个明确的局限request-scoped replay 没有MessageScopedUIDs时仍是 best effort本阶段不通过快照表修复这正是不引入订阅者快照持久化的设计取舍。仓库现状从 CMDConversationState 到 UserCMDChannelMembership 的演进值得注意的是当前仓库中 P2d-d 计划描述的部分组件已经历后续迭代。internal/usecase/cmdsync目录现在由 app.go、types.go、binding.go、records.go 等文件组成持久化模型从计划中的CMDConversationState按uid - readSeq每命令频道维护读进度演进为UserCMDChannelMembership成员关系模型持久化表定义在 pkg/db/meta/table_user_cmd_channel_membership.go主键为(uid, command_channel_id, channel_type)并携带StartSeq、AckSeq、Tombstone、TombstoneAt、UpdatedAt等字段把CMD 频道发现与ack 进度合并为一行成员关系internal/usecase/cmdsync/types.go中的StateStore相应提供ListUserCMDChannelMembershipPage、UpsertUserCMDChannelMemberships、AdvanceUserCMDChannelMembershipAcks、TombstoneUserCMDChannelMemberships方法同时新增 binding.go 的Bind/Unbind命令用于显式创建/墓碑化 CMD 目录行。计划中意图-待处理-落盘的核心思想用已解析接收者而非提交后扫描、先路由到 UID 属主、优雅停机文件恢复、/message/sync未冲刷活跃可见在这些后续迭代中以成员关系形态继续传承。HTTP 适配层保持稳定internal/access/api/message_sync.go 仍暴露POST /message/synchandleMessageSync与POST /message/syncackhandleMessageSyncAck请求体含uid、message_seq仅兼容接受、不选择状态、limit响应把SyncedMessage转换为 legacy 消息格式。app 组装根在 internal/app/wiring.go 的wireCMDSync()中通过clusterinfra.CMDSyncNode构建cmdsyncusecase.New(cmdsyncusecase.Options{CommandChannelSuffix: a.cfg.Message.CMDChannelSuffix, States: store, Messages: store})其中CommandChannelSuffix配置项默认空即 legacy 默认后缀决定命令频道 ID 编解码。端到端回归与文档同步Task 9/10 明确要求以下回归测试与文档更新普通/conversation/sync忽略 CMD 频道——即使 CMD 待处理更新器对该 UID 有条目/message/sync对 pending-only 频道只剥离恰好一层____cmd后缀/message/syncack使用最新 sync generation 并清理待处理绝不把last_message_seq当作频道选择器NoPersist命令风格消息不调用意图 sink、不创建待处理条目request-scoped durable 发送的部分 owner 路由失败通过 delivery observer 重试且已接受分区保持幂等聚焦包测试go test ./internal/usecase/cmdsync ./internal/usecase/message ./internal/usecase/delivery ./internal/access/node ./internal/app -count1存储/代理健全性go test ./pkg/slot/meta ./pkg/slot/fsm ./pkg/slot/proxy -count1导入边界go test ./internal -run TestInternalImportBoundaries -count1。文档层面docs/raw/send-path-business-logic-diff.md 的P2d-d 实施后状态2026-05-14章节确认了上述行为全部落地request-scoped durable send 在 append 成功后用RequestSubscribers构建意图并按 UID owner 路由普通 CMD delivery 在解析出 UID 页后、presence 扩展前构建意图ConversationUpdater按 command channel channel type 合并并维护 UID 索引/message/sync把 pending overlay 与已落盘CMDConversationState合并/message/syncack对 pending-only 频道先 upsert durable 读进度再清理优雅停机把未 flush 的 pending updates 保存到DataDir/conversationv2/cmd_conversation_updates.json并在启动时恢复启动加载后保留文件直到后续 durable flush/save 成功缩小恢复窗口内再次崩溃的丢失面但 kill -9 前新接收的内存意图仍不保证持久化committed replay 只重放 delivery/conversation不再运行独立 CMD subscriber scan/projector。仍未恢复的差异包括在线 cmd 专用systemcmdonline、普通临时频道投递、/message/sendbatch、expire、plugin/webhook/AI 钩子等继续延后。分层与边界总结P2d-d 的最终分层设计文档Layering是理解代码归属的钥匙internal/usecase/cmdsync意图模型、构建器、待处理更新器、sync 覆盖层、ack 清理internal/appUID-owner 路由、生命周期装配、delivery 观察者装配、集群/节点依赖internal/access/node远端属主意图提交的薄 RPC 适配器只做校验与传输映射不承载业务规则internal/usecase/message仅为 request-scoped durable 发送调用 app 提供的意图 sink不含集群路由逻辑internal/runtime/delivery不承载 CMD 业务规则如需触碰只通过 app 装配暴露泛化 UID 页观察者钩子。全阶段不引入新的全局服务对象。失败处理遵循旧版 best-effort 会话副作用语义缺失 UID 属主 leader 时仅在路由有重试队列时保留意图否则记录并作为副作用失败丢弃远端 RPC 失败在调用方已异步时有界重试待处理队列满则记录、上报指标并丢弃副作用而不是无限阻塞 delivery actor冲刷失败保留内存待处理并下个 tick 重试停止保存失败清晰记录、不 panic加载解码失败重命名.bad供运维检查。这套设计在不新增订阅者快照表的前提下把 CMD 会话更新恢复到了旧版 conversation intent 语义是理解 WuKongIM 发送链路从 P2d-c 投影模型演进到意图-待处理-成员关系模型的关键文档。赞分享即时通讯后端【免费下载链接】WuKongIMMore than just IM 不只是即时通讯(IM)项目地址https://gitcode.com/gh_mirrors/wu/WuKongIM点击查看免费下载相关推荐WuKongIM 进程级引导App Bootstrap实现剖析从 internal/app 到 cmd/wukongim 的集群优先组装路径WuKongIM 进程级引导App Bootstrap实现剖析从 internal/app 到 cmd/wukongim 的集群优先组装路径 导读 本文基即时通讯后端WuKongIM MQTT 本地 Owner 到期清扫Owner Sweeping机制解析WuKongIM MQTT 本地 Owner 到期清扫Owner Sweeping机制解析 本文以 WuKongIM 仓库中的设计契约文档 docs/spe即时通讯后端WuKongIM MQTT 订阅编排基于 Session 修订 CAS 的订阅意图协调机制解析WuKongIM MQTT 订阅编排基于 Session 修订 CAS 的订阅意图协调机制解析 导读 本文聚焦 WuKongIM 中 MQTT 订阅能力的关键即时通讯后端上一篇解密BERT学习率调度tohoku-nlp/bert-base-japanese中的warmup策略下一篇MobaXterm-Keygen源码深度解析自定义Base64与加密算法详解创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考