fix: 在 OnConnect 把 mochi 发送配额置 0 规避死锁

docs/DEVIATIONS.md

@@ -1214,3 +1214,12 @@ issue #3 未关闭，`feat/fix-3-downlink-deadlock` 未合入 `main`。下面是
 5. **建议的正确方向**  
    - 在 broker 把对本连接的下行 `InjectPacket` 与上行 worker 解耦：上行读循环先写完 PUBACK，处理 `HandleUplink` 期间不要同步向本连接注入；handler 返回后再发 `resp` 和 `group_event`。不要靠固定 `Sleep`。`InlineClient: true` 保持，`OnPublish` 对 InlineClient 继续放行。  
    - 覆盖 presence 等其他同步 `PublishDown`，而不只包一层 `emit`。
+
+### 复审修复 B-01
+
+- 日期：2026-09-30
+- 原条款：DEVELOPMENT 第 5 节装配 mochi；未写客户端 Receive Maximum。Gitea #8。
+- 实际做法：`OnConnect` 在心跳校正后调用 `cl.State.Inflight.ResetSendQuota(0)`，不 fork mochi。CONNECT 声明的 Receive Maximum 小于 256 时打 warn，连接仍接受。应用层窗口（推送 32、回执 64、在途 resp 等）约束未确认的 QoS 1。
+- 原因：mochi v2.7.9 在 `sendQuota>0` 时走 `NextImmediate` 递归读锁，并可因补发后删除 inflight 泄漏配额；已验证置 0 绕开整条路径。
+- 备选方案：fork 修补 mochi（只修递归读锁仍观察到停滞）。
+- 影响：服务端不再执行客户端 Receive Maximum；裸设备若带过小的 Receive Maximum，实际在途可能超过该值。

internal/broker/b01_quota_test.go

@@ -0,0 +1,225 @@
+package broker
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"net"
+	"runtime"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"git.asio.asia/nixevol/NixMsg/internal/app/port"
+	"github.com/mochi-mqtt/server/v2/packets"
+)
+
+func TestReceiveMaximumDoesNotDeadlockPublish(t *testing.T) {
+	b, err := New(Options{Authenticator: AllowAuthenticator{}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = b.Close() }()
+
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = ln.Close() }()
+
+	clientDone := make(chan struct{})
+	go func() {
+		defer close(clientDone)
+		c, accErr := ln.Accept()
+		if accErr != nil {
+			return
+		}
+		_ = b.AttachTCP(c)
+	}()
+
+	w, err := net.Dial("tcp", ln.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = w.Close() }()
+
+	endpoint := "ep-rm-quota"
+	writeConnectFull(t, w, endpoint, 30, 0, 20)
+	readExactPacket(t, w, packets.Connack, 3*time.Second)
+	writeSubscribe(t, w, downTopic(endpoint))
+	readExactPacket(t, w, packets.Suback, 3*time.Second)
+
+	deadline := time.Now().Add(3 * time.Second)
+	for {
+		if _, ok := b.ConnInfoOf(endpoint); ok {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("session not established")
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+
+	var received atomic.Int64
+	var writeMu sync.Mutex
+	stop := make(chan struct{})
+	var stopOnce sync.Once
+	halt := func() { stopOnce.Do(func() { close(stop) }) }
+	defer halt()
+
+	go func() {
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			_ = w.SetReadDeadline(time.Now().Add(200 * time.Millisecond))
+			hdr := make([]byte, 1)
+			if _, err := io.ReadFull(w, hdr); err != nil {
+				continue
+			}
+			rem, err := readRemainingLengthConn(w)
+			if err != nil {
+				continue
+			}
+			body := make([]byte, rem)
+			if _, err := io.ReadFull(w, body); err != nil {
+				continue
+			}
+			typ := hdr[0] >> 4
+			if typ != packets.Publish {
+				continue
+			}
+			qos := (hdr[0] >> 1) & 0x3
+			received.Add(1)
+			if qos == 0 {
+				continue
+			}
+			pk := new(packets.Packet)
+			pk.ProtocolVersion = 5
+			pk.FixedHeader = packets.FixedHeader{Type: packets.Publish, Remaining: rem, Qos: qos}
+			if decErr := pk.PublishDecode(body); decErr != nil {
+				continue
+			}
+			ack := packets.Packet{
+				FixedHeader:     packets.FixedHeader{Type: packets.Puback},
+				ProtocolVersion: 5,
+				PacketID:        pk.PacketID,
+			}
+			var ab bytes.Buffer
+			_ = ack.PubackEncode(&ab)
+			writeMu.Lock()
+			_, _ = w.Write(ab.Bytes())
+			writeMu.Unlock()
+		}
+	}()
+
+	go func() {
+		tick := time.NewTicker(2 * time.Millisecond)
+		defer tick.Stop()
+		pk := packets.Packet{
+			FixedHeader:     packets.FixedHeader{Type: packets.Pingreq},
+			ProtocolVersion: 5,
+		}
+		var buf bytes.Buffer
+		_ = pk.PingreqEncode(&buf)
+		ping := append([]byte(nil), buf.Bytes()...)
+		for {
+			select {
+			case <-stop:
+				return
+			case <-tick.C:
+				writeMu.Lock()
+				_, _ = w.Write(ping)
+				writeMu.Unlock()
+			}
+		}
+	}()
+
+	payload := []byte(`{"v":1,"type":"resp","rid":"x"}`)
+	pubCtx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var wg sync.WaitGroup
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for pubCtx.Err() == nil {
+				_ = b.PublishDown(pubCtx, endpoint, "", payload, port.PublishOpts{QoS: 1})
+			}
+		}()
+	}
+
+	runFor := 5 * time.Second
+	watch := 2 * time.Second
+	start := time.Now()
+	last := received.Load()
+	lastChange := time.Now()
+	for time.Since(start) < runFor {
+		time.Sleep(50 * time.Millisecond)
+		n := received.Load()
+		if n > last {
+			last = n
+			lastChange = time.Now()
+		}
+		if time.Since(lastChange) > watch {
+			buf := make([]byte, 1<<20)
+			nstack := runtime.Stack(buf, true)
+			halt()
+			cancel()
+			_ = w.Close()
+			t.Fatalf("progress stalled at %d after %s\n%s", last, time.Since(lastChange), buf[:nstack])
+		}
+	}
+	cancel()
+	wg.Wait()
+	halt()
+	if last < 100 {
+		t.Fatalf("too few publishes delivered: %d", last)
+	}
+
+	_ = w.Close()
+	select {
+	case <-clientDone:
+	case <-time.After(3 * time.Second):
+	}
+}
+
+func readExactPacket(t *testing.T, conn net.Conn, wantType byte, timeout time.Duration) {
+	t.Helper()
+	_ = conn.SetReadDeadline(time.Now().Add(timeout))
+	hdr := make([]byte, 1)
+	if _, err := io.ReadFull(conn, hdr); err != nil {
+		t.Fatal(err)
+	}
+	if hdr[0]>>4 != wantType {
+		t.Fatalf("want packet type %d got %d", wantType, hdr[0]>>4)
+	}
+	rem, err := readRemainingLengthConn(conn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	body := make([]byte, rem)
+	if _, err := io.ReadFull(conn, body); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func readRemainingLengthConn(r io.Reader) (int, error) {
+	var mul uint32 = 1
+	var value uint32
+	for i := 0; i < 4; i++ {
+		var b [1]byte
+		if _, err := io.ReadFull(r, b[:]); err != nil {
+			return 0, err
+		}
+		value += uint32(b[0]&127) * mul
+		if b[0]&128 == 0 {
+			return int(value), nil
+		}
+		mul *= 128
+	}
+	return 0, io.ErrUnexpectedEOF
+}

internal/broker/broker_test.go

@@ -212,6 +212,11 @@ func connectAndSubscribe(t *testing.T, w net.Conn, endpoint string, maxPacket ui
 }
 
 func writeConnect(t *testing.T, w net.Conn, endpoint string, keepalive uint16, maxPacket uint32) {
+	t.Helper()
+	writeConnectFull(t, w, endpoint, keepalive, maxPacket, 0)
+}
+
+func writeConnectFull(t *testing.T, w net.Conn, endpoint string, keepalive uint16, maxPacket uint32, receiveMax uint16) {
 	t.Helper()
 	pk := packets.Packet{
 		FixedHeader:     packets.FixedHeader{Type: packets.Connect},
@@ -228,6 +233,7 @@ func writeConnect(t *testing.T, w net.Conn, endpoint string, keepalive uint16, m
 		},
 		Properties: packets.Properties{
 			MaximumPacketSize: maxPacket,
+			ReceiveMaximum:    receiveMax,
 		},
 	}
 	var buf bytes.Buffer

internal/broker/hooks.go

@@ -67,6 +67,16 @@ func (h *nixHook) OnConnect(cl *mqtt.Client, pk packets.Packet) error {
 		cl.State.ServerKeepalive = true
 	}
 
+	// B-01：绕开 mochi 发送配额路径（NextImmediate 递归读锁 + PUBACK 配额泄漏）。
+	// ParseConnect 已按客户端 Receive Maximum 设过 sendQuota；此处一律置 0。
+	if cl.State.Inflight != nil {
+		cl.State.Inflight.ResetSendQuota(0)
+	}
+	if rm := pk.Properties.ReceiveMaximum; rm > 0 && rm < 256 {
+		h.b.log.Warn("client receive maximum below 256; server ignores MQTT send quota",
+			"endpoint", endpointID, "receive_maximum", rm)
+	}
+
 	res, err := h.b.auth.Authenticate(context.Background(), endpointID, pk.Connect.Password, remoteIP)
 	if err != nil {
 		st.authErr = err