《互联网产品的服务端架构·内容、社交与实时互动》13.1 发布→Feed→互动→通知

从「回声」一次发布因通知失败整条回滚的事故写起:拆解发布、Feed 分发、互动与通知四个阶段的同步异步边界和失败语义,给出 Flow、Step 与 Validate 的 Go 实现、表驱动测试与真实运行输出,并整理链路编排的常见错误与边界。

「回声」的一条动态,被一条通知拖回了草稿箱

「回声」的作者老周在晚上十点发了一条动态:一张骑行路线图,配了三行字。点击发布后,页面转了五秒,弹出“发布失败,请重试”。他又点了一次,还是失败。十分钟后,关注他的用户却在 Feed 里看到了这条动态,点赞和评论也开始进来。老周懵了:页面说没发出去,为什么别人能看到?

排查日志才发现,发布接口把四件事放在了一个同步调用链里:写内容表、写关注者 Feed、写作者的互动索引、调用推送网关发通知。推送网关当天响应很慢,超时是三秒,而发布接口的事务一直等到网关返回。网关超时后,事务回滚,内容表没有新行;但 Feed 写入走的是另一个已经提交的分支,因为开发同学为了让 Feed“快一点”,把它提前异步化了。于是数据库里没有内容,Feed 里却有卡片。

更糟的是重试。老周第二次点击时,客户端生成了新的幂等键,服务端又走了一遍流程;Feed 里出现了两张一模一样的卡片,通知也发了两遍。运营最后手动删掉了重复卡片,但作者已经对“发布成功没有”失去了信任。

这个事故不是“通知服务不稳定”这么简单。它暴露的是四个阶段的边界没有写清楚:哪些步骤必须在用户请求里完成,哪些可以异步,异步失败后谁来补偿,哪些失败必须让用户看到。把这些问题混在一个事务里,任何一次网络抖动都会变成一次内容事故。

四个阶段,四种不同的失败语义

一条内容从发布到被互动,至少要经过四个阶段。它们不是四套独立服务,而是四种对失败的承诺。把承诺写进代码之前,先写进表里。

阶段允许异步失败时用户看到什么失败后的补偿典型错误
publish必须同步明确的失败提示,内容不存在用幂等键安全重试把通知也塞进发布事务
feed可以异步作者主页可见,关注者暂不可见从事件日志重放异步失败后静默丢事件
interaction必须同步点赞/评论失败,计数不变按用户、目标、动作幂等重试先加计数后写关系
notify可以异步内容已发布,通知可能延迟独立队列重试与降级通知失败回滚内容

publish 和 interaction 必须同步,因为用户正在等待一个确定结果。用户点了发布,系统不能回答“也许发出去了”;用户点了点赞,界面不能先显示红心再偷偷撤回。这两个阶段失败时,最安全的做法是让请求失败,并保证重试不会产生重复副作用。

feed 和 notify 可以异步,因为它们服务的是“别人”和“稍后”。关注者晚几秒看到内容,通常比作者发布失败更容易接受;通知晚几秒到达,也比整条内容回滚更容易接受。但异步不等于可以丢:feed 写入要有事件日志和重放能力,通知要有独立的重试队列和降级策略。

顺序也不是装饰。publish 必须先于 feed,因为 Feed 需要内容 ID;interaction 必须晚于 feed,因为互动要指向一条已经可见的内容;notify 最后,因为它依赖互动或发布事件。顺序错了,系统会出现“通知先到、内容还查不到”的悬空状态。

为什么通知不能和发布共用事务

事务的边界应该包住“必须一起成功或一起失败”的写操作,而不是包住网络调用。推送网关是外部依赖,它的超时、限流和故障都不受发布服务控制。把网关调用放进事务,等于把数据库锁的持有时间交给一个不受控的服务。

正确做法是事务内只写内容、Feed 索引和一条 outbox 事件,事务提交后由独立消费者发通知。通知失败时,内容仍然存在,消费者按退避策略重试;重试多次仍失败,就进入死信队列并告警,而不是回滚已经提交的内容。

案例假设值:发布接口 P99 从 180ms 降到 60ms,通知延迟从“跟随发布”变成独立队列的 2 秒内;这些数字用于说明边界调整的方向,不是本机实测。

失败场景:通知超时,发布回滚,Feed 却已经写入

把事故按时间线拆开,能看到三个独立的问题叠在一起。

第一步,发布请求进入事务,写内容表和 outbox,然后同步调用通知网关。网关在第三秒超时,事务回滚,内容表没有新行,outbox 也没有留下事件。

第二步,Feed 写入在事务提交前就被异步投递了。消费者拿到的是内存里的事件对象,不是数据库里的 outbox 记录;它不知道发布事务已经回滚,仍然把卡片写进了关注者 Feed。

第三步,客户端重试时使用了新的幂等键。服务端没有把“同一次发布”认出来,于是又写了一遍 Feed,通知也重发了一次。用户看到的是重复卡片和重复通知,而不是一次干净的失败。

修复要同时做三件事。第一,把通知移出事务,改为 outbox 模式,事务提交后才允许投递。第二,Feed 消费者只消费已经落库的 outbox 事件,事务回滚时事件根本不存在。第三,发布接口要求客户端复用同一个幂等键,服务端以“作者 + 幂等键”去重,重试返回第一次的结果,而不是再执行一遍。

这三件事里,第二件最容易被忽略。很多团队把“异步”理解成“先发消息再提交事务”,结果消息比事务更快,消费者读到了还不存在的内容。异步的起点必须是已经提交的事实,而不是即将提交的意图。

用 Flow 和 Validate 把编排约束变成代码

光靠文档约定不够。四个阶段是否齐全、顺序是否单调、哪些阶段被错误地标成异步,这些都可以在启动时检查。下面的代码来自本书配套的 webapp-lab,把一条内容链路的步骤建模成 Flow,用 Validate 在配置加载阶段拒绝非法编排。

// file: internal/socialbase/flow.go
// tier: L1
// verify: go test ./internal/socialbase/
package socialbase

import (
	"errors"
	"fmt"
)

// Stage 标识内容链路中的一个处理阶段。
type Stage string

const (
	StagePublish     Stage = "publish"
	StageFeed        Stage = "feed"
	StageInteraction Stage = "interaction"
	StageNotify      Stage = "notify"
)

// Step 是链路中的一步。Async 表示这一步是否可以脱离用户请求异步执行。
type Step struct {
	Name  string
	Stage Stage
	Async bool
}

// Flow 描述一条内容从发布到通知的完整步骤。
type Flow struct {
	Steps []Step
}

var stageOrder = map[Stage]int{
	StagePublish:     0,
	StageFeed:        1,
	StageInteraction: 2,
	StageNotify:      3,
}

// Validate 检查四个阶段是否齐全、顺序是否正确,以及同步约束是否满足。
func Validate(f Flow) error {
	if len(f.Steps) == 0 {
		return errors.New("flow has no steps")
	}

	seenName := make(map[string]bool, len(f.Steps))
	seenStage := make(map[Stage]bool, 4)
	lastOrder := -1

	for i, st := range f.Steps {
		if st.Name == "" {
			return fmt.Errorf("step %d: empty name", i)
		}
		if seenName[st.Name] {
			return fmt.Errorf("step %d: duplicate name %q", i, st.Name)
		}
		seenName[st.Name] = true

		order, ok := stageOrder[st.Stage]
		if !ok {
			return fmt.Errorf("step %d %q: unknown stage %q", i, st.Name, st.Stage)
		}
		if order < lastOrder {
			return fmt.Errorf("step %d %q: stage %q is out of order", i, st.Name, st.Stage)
		}
		lastOrder = order
		seenStage[st.Stage] = true

		if (st.Stage == StagePublish || st.Stage == StageInteraction) && st.Async {
			return fmt.Errorf("step %d %q: stage %q must be synchronous", i, st.Name, st.Stage)
		}
	}

	for _, stage := range []Stage{StagePublish, StageFeed, StageInteraction, StageNotify} {
		if !seenStage[stage] {
			return fmt.Errorf("flow is missing stage %q", stage)
		}
	}
	return nil
}

Stage 是字符串类型,四个常量分别对应 publish、feed、interaction、notify。用字符串而不是整数,是为了在日志和错误信息里直接看到阶段名;顺序由 stageOrder 映射单独维护,避免常量值本身被当成顺序。

Validate 先拒绝空链路,再逐条检查步骤名和阶段。空名字和重复名字都会报错,因为后续的重试、监控和排障都依赖步骤名定位问题;两个步骤同名,告警里就分不清是哪一步失败。

顺序检查用“当前阶段序号不得小于上一个阶段序号”。这允许同一阶段出现多个步骤,比如 publish 阶段既写内容又写索引,feed 阶段既写收件箱又写发件箱;但不允许阶段倒退。四个阶段必须全部出现,缺一个就说明链路不完整。

同步约束只对 publish 和 interaction 生效。feed 和 notify 标成异步是合法的,标成同步也不会报错,因为同步执行异步阶段是性能问题,不是正确性问题;反过来,把必须同步的阶段标成异步,则是语义错误,必须拒绝。

测试:齐全、顺序、同步约束和命名

编排校验的测试不需要启动服务,重点是覆盖每一条拒绝路径,并确认合法的多步骤链路不会被误杀。

// file: internal/socialbase/flow_test.go
// tier: L1
// verify: go test ./internal/socialbase/
package socialbase

import (
	"strings"
	"testing"
)

func TestValidateAcceptsFullFlow(t *testing.T) {
	flow := Flow{Steps: []Step{
		{Name: "write", Stage: StagePublish},
		{Name: "fanout", Stage: StageFeed, Async: true},
		{Name: "like", Stage: StageInteraction},
		{Name: "push", Stage: StageNotify, Async: true},
	}}
	if err := Validate(flow); err != nil {
		t.Fatalf("Validate() error = %v", err)
	}
}

func TestValidateAcceptsMultipleStepsPerStage(t *testing.T) {
	flow := Flow{Steps: []Step{
		{Name: "write", Stage: StagePublish},
		{Name: "index", Stage: StagePublish},
		{Name: "fanout", Stage: StageFeed, Async: true},
		{Name: "like", Stage: StageInteraction},
		{Name: "comment", Stage: StageInteraction},
		{Name: "push", Stage: StageNotify, Async: true},
	}}
	if err := Validate(flow); err != nil {
		t.Fatalf("Validate() error = %v", err)
	}
}

func TestValidateRejectsBadFlow(t *testing.T) {
	base := []Step{
		{Name: "write", Stage: StagePublish},
		{Name: "fanout", Stage: StageFeed, Async: true},
		{Name: "like", Stage: StageInteraction},
		{Name: "push", Stage: StageNotify, Async: true},
	}
	cases := []struct {
		name string
		flow Flow
		want string
	}{
		{"empty", Flow{}, "no steps"},
		{"missing stage", Flow{Steps: base[:3]}, "missing stage"},
		{"out of order", Flow{Steps: []Step{
			{Name: "fanout", Stage: StageFeed, Async: true},
			{Name: "write", Stage: StagePublish},
			{Name: "like", Stage: StageInteraction},
			{Name: "push", Stage: StageNotify, Async: true},
		}}, "out of order"},
		{"publish async", Flow{Steps: []Step{
			{Name: "write", Stage: StagePublish, Async: true},
			{Name: "fanout", Stage: StageFeed, Async: true},
			{Name: "like", Stage: StageInteraction},
			{Name: "push", Stage: StageNotify, Async: true},
		}}, "must be synchronous"},
		{"interaction async", Flow{Steps: []Step{
			{Name: "write", Stage: StagePublish},
			{Name: "fanout", Stage: StageFeed, Async: true},
			{Name: "like", Stage: StageInteraction, Async: true},
			{Name: "push", Stage: StageNotify, Async: true},
		}}, "must be synchronous"},
		{"empty name", Flow{Steps: []Step{
			{Name: "", Stage: StagePublish},
			{Name: "fanout", Stage: StageFeed, Async: true},
			{Name: "like", Stage: StageInteraction},
			{Name: "push", Stage: StageNotify, Async: true},
		}}, "empty name"},
		{"duplicate name", Flow{Steps: []Step{
			{Name: "write", Stage: StagePublish},
			{Name: "write", Stage: StageFeed, Async: true},
			{Name: "like", Stage: StageInteraction},
			{Name: "push", Stage: StageNotify, Async: true},
		}}, "duplicate name"},
		{"unknown stage", Flow{Steps: []Step{
			{Name: "write", Stage: StagePublish},
			{Name: "fanout", Stage: Stage("timeline"), Async: true},
			{Name: "like", Stage: StageInteraction},
			{Name: "push", Stage: StageNotify, Async: true},
		}}, "unknown stage"},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			err := Validate(tc.flow)
			if err == nil {
				t.Fatalf("Validate() = nil, want error containing %q", tc.want)
			}
			if !strings.Contains(err.Error(), tc.want) {
				t.Fatalf("Validate() error = %q, want substring %q", err, tc.want)
			}
		})
	}
}

TestValidateAcceptsFullFlow 是最小合法链路:四个阶段各一步,feed 和 notify 异步。TestValidateAcceptsMultipleStepsPerStage 确认同一阶段可以有多步,避免实现把“四步”误写成“每阶段恰好一步”。

TestValidateRejectsBadFlow 用表驱动覆盖八种错误:空链路、缺阶段、顺序倒退、publish 异步、interaction 异步、空名字、重复名字和未知阶段。每个用例不只断言“有错”,还断言错误信息包含关键子串,这样错误信息被改坏时测试会失败。

运行命令与真实输出

在仓库根目录执行:

cd .local/checks/webapp-lab && go test ./internal/socialbase/ -run 'TestValidate' -count=1 -v

一次本机运行的真实输出如下,耗时随机器变化:

=== RUN   TestValidateAcceptsFullFlow
--- PASS: TestValidateAcceptsFullFlow (0.00s)
=== RUN   TestValidateAcceptsMultipleStepsPerStage
--- PASS: TestValidateAcceptsMultipleStepsPerStage (0.00s)
=== RUN   TestValidateRejectsBadFlow
=== RUN   TestValidateRejectsBadFlow/empty
=== RUN   TestValidateRejectsBadFlow/missing_stage
=== RUN   TestValidateRejectsBadFlow/out_of_order
=== RUN   TestValidateRejectsBadFlow/publish_async
=== RUN   TestValidateRejectsBadFlow/interaction_async
=== RUN   TestValidateRejectsBadFlow/empty_name
=== RUN   TestValidateRejectsBadFlow/duplicate_name
=== RUN   TestValidateRejectsBadFlow/unknown_stage
--- PASS: TestValidateRejectsBadFlow (0.00s)
    --- PASS: TestValidateRejectsBadFlow/empty (0.00s)
    --- PASS: TestValidateRejectsBadFlow/missing_stage (0.00s)
    --- PASS: TestValidateRejectsBadFlow/out_of_order (0.00s)
    --- PASS: TestValidateRejectsBadFlow/publish_async (0.00s)
    --- PASS: TestValidateRejectsBadFlow/interaction_async (0.00s)
    --- PASS: TestValidateRejectsBadFlow/empty_name (0.00s)
    --- PASS: TestValidateRejectsBadFlow/duplicate_name (0.00s)
    --- PASS: TestValidateRejectsBadFlow/unknown_stage (0.00s)
PASS
ok  	webapp/internal/socialbase	0.372s

输出里 TestValidateRejectsBadFlow 的八个子用例全部通过,说明每条拒绝路径都被覆盖。包级耗时包含测试二进制启动,单个用例显示 0.00s,因为校验本身只是几十次字符串比较和 map 查找。

小结

发布、Feed、互动、通知是四个不同承诺的阶段,不是四个可以随意拼接的函数。publish 和 interaction 必须同步,失败要让用户看见;feed 和 notify 可以异步,但必须有事件日志、重试和降级。通知不能和发布共用事务,异步的起点必须是已经提交的事实。

把阶段、顺序和同步约束写进 Validate,可以让配置错误在启动时暴露,而不是等到某次通知超时才发现内容被回滚。下一节把视角从单条链路拉到多端:一条消息发出后,手机、平板和服务端的未读数如何用端到端用例串起来验证。

阅读导航:上一节:目录与学习路径 · 下一节:13.2 IM 与未读的端到端验证 。

常见错误与边界

  • 把通知网关调用放进发布事务。外部依赖的超时会持有数据库锁,并让已经写入的 Feed 变成孤儿数据。
  • 先投递异步消息再提交事务。消费者可能读到尚未提交的内容,事务回滚后消息无法撤回。
  • 把 publish 或 interaction 标成异步。用户等待的是确定结果,异步化会把失败藏到用户看不见的地方。
  • 用步骤下标而不是阶段名做顺序判断。插入新步骤后下标会漂移,阶段约束应该按语义比较。
  • 只检查“四个阶段都存在”,不检查顺序。通知先于内容出现时,用户会点进一个查不到的链接。
  • 允许重复步骤名。重试和告警无法区分是哪一步,排障时只能靠猜。
  • 重试使用新的幂等键。发布失败后的重试会变成第二次发布,重复内容和重复通知同时出现。
  • 把“异步”当成“可以不送达”。feed 和 notify 可以延迟,但必须有重放和死信队列兜底。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「books」更多文章

  1. 《互联网产品的 AI 架构·平台化与治理》附录 C 平台化评审清单
  2. 《互联网产品的 AI 架构·平台化与治理》附录 B 能力目录与配额字段速查
  3. 《互联网产品的 AI 架构·平台化与治理》附录 A 治理角色与职责速查