diff --git a/internal/app/app.go b/internal/app/app.go index ae167f3..03be50e 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -38,10 +38,11 @@ import ( // App 持有运行期所有需要被关停的资源。字段保持非导出,避免外部直接操作。 type App struct { - cfg *config.Config - log *slog.Logger - pool *db.Pool - server *http.Server + cfg *config.Config + log *slog.Logger + pool *db.Pool + server *http.Server + dispatcher *notify.Dispatcher } // New 把所有模块按启动顺序拼装好。任何阶段失败都会返回错误,不留半成品资源。 @@ -68,9 +69,6 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) { notifyRepo := notify.NewPostgresRepository(pool) notifyDispatcher := notify.NewDispatcher(notifyRepo, log) notifyDispatcher.Register(notify.NewDummyNotifier(log)) - // 当前 dispatcher 仅暴露给本包:未来引入定时任务/后台 worker 才需把它 - // 通过 App 字段透出去,目前保留局部变量以避免暴露未稳定 API。 - _ = notifyDispatcher spaSub, err := fs.Sub(dist, "web/dist") if err != nil { @@ -113,7 +111,7 @@ func New(ctx context.Context, cfg *config.Config, dist embed.FS) (*App, error) { WriteTimeout: 60 * time.Second, IdleTimeout: 120 * time.Second, } - return &App{cfg: cfg, log: log, pool: pool, server: srv}, nil + return &App{cfg: cfg, log: log, pool: pool, server: srv, dispatcher: notifyDispatcher}, nil } // Run 启动 HTTP server 并阻塞直到 ctx 取消(Shutdown)或服务器报错。 @@ -137,9 +135,16 @@ func (a *App) Run(ctx context.Context) error { if shutErr != nil { a.log.Error("server shutdown", "err", shutErr.Error()) } + // server 已停止接收新请求;等异步通知投递结束再关 pool + if err := a.dispatcher.Shutdown(shutCtx); err != nil { + a.log.Warn("notify dispatcher drain", "err", err.Error()) + } a.pool.Close() return shutErr case err := <-errCh: + drainCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = a.dispatcher.Shutdown(drainCtx) a.pool.Close() return err } diff --git a/internal/infra/notify/dispatcher.go b/internal/infra/notify/dispatcher.go index 74b03d9..592f2a6 100644 --- a/internal/infra/notify/dispatcher.go +++ b/internal/infra/notify/dispatcher.go @@ -6,6 +6,7 @@ package notify import ( "context" "log/slog" + "sync" "time" "github.com/google/uuid" @@ -13,12 +14,13 @@ import ( // Dispatcher 是 notify 模块对外的发送门面。Repository 由调用方注入(生产 // 用 pgRepo,测试用 fake),Notifier 通过 Register 动态追加,now 钩子留给 -// 测试覆盖时间相关行为。 +// 测试覆盖时间相关行为。wg 用于 Shutdown 时等待所有异步投递结束。 type Dispatcher struct { repo Repository notifiers []Notifier log *slog.Logger now func() time.Time + wg sync.WaitGroup } // NewDispatcher 用 Repository 与 Logger 构造 Dispatcher。注册 Notifier 通过 @@ -36,6 +38,10 @@ func (d *Dispatcher) Register(n Notifier) { // Dispatch 同步落库;落库失败返回错误,不再异步投递。落库成功后, // fire-and-forget 派发给所有外部 notifier,单个 notifier 的失败仅记日志, // 不阻塞其它通道;调用方无法感知异步投递结果。 +// +// 落库使用 context.WithoutCancel + 5s 硬超时,避免请求 ctx 取消导致通知 +// 漏写;外部投递使用独立 background ctx + 30s 硬超时,lifecycle 由 Shutdown +// 通过 wg 管控。 func (d *Dispatcher) Dispatch(ctx context.Context, msg Message) error { if msg.ID == uuid.Nil { msg.ID = uuid.New() @@ -43,14 +49,18 @@ func (d *Dispatcher) Dispatch(ctx context.Context, msg Message) error { if msg.CreatedAt.IsZero() { msg.CreatedAt = d.now() } - if err := d.repo.Insert(ctx, msg); err != nil { + persistCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + defer cancel() + if err := d.repo.Insert(persistCtx, msg); err != nil { return err } for _, n := range d.notifiers { + d.wg.Add(1) go func(n Notifier) { - // 异步投递使用独立 ctx,避免请求 ctx 取消导致投递中断; - // 真实 notifier 应在内部加超时限制。 - if err := n.Send(context.Background(), msg); err != nil { + defer d.wg.Done() + sendCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := n.Send(sendCtx, msg); err != nil { d.log.Error("notify channel failed", "channel", n.Name(), "topic", msg.Topic, @@ -61,3 +71,20 @@ func (d *Dispatcher) Dispatch(ctx context.Context, msg Message) error { } return nil } + +// Shutdown 阻塞等待所有异步投递完成。调用方应保证此后不再有新的 Dispatch +// 调用(通常上游 HTTP server 已经 Shutdown)。ctx 用作硬截止:超时会返回 +// ctx.Err(),但泄漏的 goroutine 会继续跑到 Send 的 30s timeout 自然结束。 +func (d *Dispatcher) Shutdown(ctx context.Context) error { + done := make(chan struct{}) + go func() { + d.wg.Wait() + close(done) + }() + select { + case <-done: + return nil + case <-ctx.Done(): + return ctx.Err() + } +}