fix: 后台加人人数改为事务内实际插入数

docs/DEVIATIONS.md

@@ -1786,7 +1786,7 @@ issue #3 未关闭，`feat/fix-3-downlink-deadlock` 未合入 `main`。下面是
    - 实际做法：`handleGroupAddMembers` 用加人前后 `group_members` 行数差得到 `added`；审计 `result` 用 `added` 与 `failed`（全跳过且无失败记 `noop`，不记 `ok`）；响应增加 `added`，`failed` 仍只含真正失败项。前端按 `added` 提示，已是成员不改成错误码。  
    - 原因：全是已有成员时旧逻辑审计成 ok、界面提示已加人，实际插入 0 行。  
    - 备选方案：把已是成员写入 `failed`（否决，会改变客户端错误语义）；扩展 `group.AddResult` 返回插入列表（可后续做，本波不改身份线接口）。  
-   - 影响：管理 API 加人成功体多 `added` 字段；契约文档示例仍写 `{"failed":[]}`，以本偏差为准。
+   - 影响：管理 API 加人成功体多 `added` 字段；契约文档示例仍写 `{"failed":[]}`，以本偏差为准。`added` 取写事务内实际插入数（`AddResult.Added`，不进端协议 JSON），不用事务外两次 COUNT 的差，避免并发加人/踢人把别人的变更算进本请求。
 
 ### 复审修复 R3-02
 
@@ -1796,4 +1796,5 @@ issue #3 未关闭，`feat/fix-3-downlink-deadlock` 未合入 `main`。下面是
    - 实际做法：带断开的下行帧用本帧 `OnPacketSent` 完成信号（优先 packet id，否则按载荷匹配），不再用连接级 `sentPub` 总数。`Shutdown` 在 ctx 未取消时先等下行队列与 `wirePending` 排空，再 `DisconnectClient` 发 `0x8B` 并在截止前等连接拆掉；ctx 已取消则发完即 `Close`。`serve` 仍给 5 秒预算并记录非超时错误。未合 `feat/fix-3-downlink-deadlock`。  
    - 原因：前面 PUBLISH 的 `OnPacketSent` 会让总数等待提前返回；`Shutdown` 对 ctx 非阻塞 select 使 5 秒预算用不上，有 outbound 积压时 `0x8B` 只进 outbuf 随 `Stop` 丢掉。  
    - 备选方案：恢复固定 `Sleep`（否决）；改 `PublishDown` 签名（否决）。  
-   - 影响：队列/outbound 有积压时 fatal、logout 先到客户端再断开；停机在预算内尽量发出 `0x8B`，超时返回 ctx 错误而非空等。
+   - 影响：队列/outbound 有积压时 fatal、logout 先到客户端再断开；停机在预算内尽量发出 `0x8B`，超时返回 ctx 错误而非空等。  
+   - 补强：`sendOne` 从取出帧到返回前记 `inSend`，排空等待把它算上（含等大帧名额、尚未 `wirePending++` 的窗口）。有截止时间时排空最多用到截止前 1 秒，剩下的时间留给 `0x8B` 写出；排空没完成也不会因此跳过这段等待。

internal/admin/groups.go

@@ -316,13 +316,6 @@ func (h *Handler) handleGroupAddMembers(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
-	before, err := h.groupMemberCount(r.Context(), id)
-	if err != nil {
-		h.auditP(p, "group_add_members", id, "error", ip)
-		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
-		return
-	}
-
 	res, err := h.groups.AdminAddMembers(r.Context(), id, req.MemberIDs)
 	if err != nil {
 		h.writeGroupErr(w, p, "group_add_members", id, ip, err)
@@ -332,14 +325,7 @@ func (h *Handler) handleGroupAddMembers(w http.ResponseWriter, r *http.Request)
 	if failed == nil {
 		failed = []group.MemberFail{}
 	}
-
-	after, err := h.groupMemberCount(r.Context(), id)
-	if err != nil {
-		h.auditP(p, "group_add_members", id, "error", ip)
-		httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
-		return
-	}
-	added := after - before
+	added := res.Added
 	if added < 0 {
 		added = 0
 	}
@@ -434,13 +420,6 @@ func (h *Handler) groupOwner(ctx context.Context, groupID string) (string, error
 	return owner, err
 }
 
-func (h *Handler) groupMemberCount(ctx context.Context, groupID string) (int, error) {
-	var n int
-	err := h.db.Read.QueryRowContext(ctx,
-		`SELECT COUNT(*) FROM group_members WHERE group_id = ?`, groupID).Scan(&n)
-	return n, err
-}
-
 func (h *Handler) groupSummary(ctx context.Context, groupID string) (map[string]any, error) {
 	var name, owner string
 	var created int64

internal/app/group/app.go

@@ -620,7 +620,7 @@ func (a *App) AdminAddMembers(ctx context.Context, groupID string, memberIDs []s
 		failed = append(failed, MemberFail{ID: id, Code: protocol.CodeGroupFull})
 	}
 	if len(toAdd) == 0 {
-		return AddResult{Failed: failed}, nil
+		return AddResult{Failed: failed, Added: 0}, nil
 	}
 	now := a.nowMs()
 	var inserted []string
@@ -664,7 +664,7 @@ func (a *App) AdminAddMembers(ctx context.Context, groupID string, memberIDs []s
 	for _, id := range inserted {
 		a.emit(ctx, notify, groupID, eventMemberAdded, id, now)
 	}
-	return AddResult{Failed: failed}, nil
+	return AddResult{Failed: failed, Added: len(inserted)}, nil
 }
 
 func (a *App) createAdmin(ctx context.Context, ownerID, name, gid string, members []protocol.GroupMemberIn) (CreateResult, error) {

internal/app/group/service.go

@@ -28,6 +28,8 @@ type CreateResult struct {
 // AddResult 是加人结果。
 type AddResult struct {
 	Failed []MemberFail `json:"failed,omitempty"`
+	// Added 是本请求在写事务内实际插入的人数。不进协议 JSON，供后台审计使用。
+	Added int `json:"-"`
 }
 
 // ListItem 是 group.list 一项。

internal/broker/b04_b06_b08_test.go

@@ -274,6 +274,27 @@ func TestShutdownCancelledContextReturnsQuickly(t *testing.T) {
 	}
 }
 
+func TestWaitConnsQuietSeesInSend(t *testing.T) {
+	b, err := New(Options{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = b.Close() }()
+	st := &connState{downCh: make(chan downItem, 1)}
+	st.inSend.Store(1)
+	ctx, cancel := context.WithTimeout(context.Background(), 40*time.Millisecond)
+	defer cancel()
+	if b.waitConnsQuiet(ctx, []*connState{st}) {
+		t.Fatal("inSend should keep shutdown from treating the conn as quiet")
+	}
+	st.inSend.Store(0)
+	ctx2, cancel2 := context.WithTimeout(context.Background(), 200*time.Millisecond)
+	defer cancel2()
+	if !b.waitConnsQuiet(ctx2, []*connState{st}) {
+		t.Fatal("quiet when inSend is 0 and queues are empty")
+	}
+}
+
 func TestEffectivePayloadLimitSubtractsOverhead(t *testing.T) {
 	got := EffectivePayloadLimit(200, 0)
 	if got != 200-packetOverheadBudget {

internal/broker/broker.go

@@ -142,6 +142,7 @@ type connState struct {
 	downDone       chan struct{}
 	downBytes      atomic.Int64
 	wirePending    atomic.Int64 // Publish 入 mochi outbound 后、OnPacketSent 前
+	inSend         atomic.Int32 // downLoop 已取出帧、尚未从 sendOne 返回
 	mu             sync.Mutex
 
 	// 带断开的下行帧：只等本帧 OnPacketSent，不用连接级计数。
@@ -237,8 +238,8 @@ func (b *Broker) Close() error {
 }
 
 // Shutdown 向所有连接发 MQTT 5 0x8B 后关闭。完整 HTTP 停机顺序见 L-03。
-// ctx 未取消时先等下行队列与 wirePending 排空，再 DisconnectClient（此时 outbound 空，
-// 0x8B 直写套接字），并在截止前等连接拆掉；ctx 已取消则发完即 Close，不等待。
+// ctx 未取消时先等下行队列、正在 sendOne 的帧与 wirePending 排空（若有截止时间则预留约 1 秒），
+// 再 DisconnectClient，并在截止前等连接拆掉；ctx 已取消则发完即 Close，不等待。
 func (b *Broker) Shutdown(ctx context.Context) error {
 	if b.closed.Load() {
 		return nil
@@ -275,17 +276,26 @@ func (b *Broker) Shutdown(ctx context.Context) error {
 
 	var waitErr error
 	if !alreadyCancelled {
-		if !b.waitConnsQuiet(ctx, states) {
+		// 留出约 1 秒给 0x8B 写出，避免排空把整个截止时间用完后立刻 Close。
+		quietCtx := ctx
+		cancelQuiet := func() {}
+		if dl, ok := ctx.Deadline(); ok && time.Until(dl) > time.Second {
+			quietCtx, cancelQuiet = context.WithDeadline(ctx, dl.Add(-time.Second))
+		}
+		if !b.waitConnsQuiet(quietCtx, states) && ctx.Err() != nil {
 			waitErr = ctx.Err()
 		}
+		cancelQuiet()
 	}
 
 	for _, cl := range clients {
 		_ = b.server.DisconnectClient(cl, packets.ErrServerShuttingDown)
 	}
 
-	if !alreadyCancelled && waitErr == nil {
-		waitErr = b.waitConnsGone(ctx)
+	if !alreadyCancelled && ctx.Err() == nil {
+		if err := b.waitConnsGone(ctx); err != nil {
+			waitErr = err
+		}
 	}
 
 	closeErr := b.Close()
@@ -299,7 +309,7 @@ func (b *Broker) waitConnsQuiet(ctx context.Context, states []*connState) bool {
 	for {
 		quiet := true
 		for _, st := range states {
-			if st.wirePending.Load() > 0 {
+			if st.inSend.Load() > 0 || st.wirePending.Load() > 0 {
 				quiet = false
 				break
 			}

internal/broker/downlink.go

@@ -118,6 +118,8 @@ func (st *connState) sendOne(b *Broker, item downItem) {
 		st.signalSent(item)
 		return
 	}
+	st.inSend.Add(1)
+	defer st.inSend.Add(-1)
 	large := len(item.payload) > largeFrameBytes
 	if large {
 		if err := b.acquireLarge(context.Background()); err != nil {