From d94fb254982c3b0ef767182c11d2506bb02e51bc Mon Sep 17 00:00:00 2001 From: Nixevol Date: Wed, 30 Sep 2026 14:58:45 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=9C=A8=20OnConnect=20=E6=8A=8A=20moch?= =?UTF-8?q?i=20=E5=8F=91=E9=80=81=E9=85=8D=E9=A2=9D=E7=BD=AE=200=20?= =?UTF-8?q?=E8=A7=84=E9=81=BF=E6=AD=BB=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/DEVIATIONS.md | 9 ++ internal/broker/b01_quota_test.go | 225 ++++++++++++++++++++++++++++++ internal/broker/broker_test.go | 6 + internal/broker/hooks.go | 10 ++ 4 files changed, 250 insertions(+) create mode 100644 internal/broker/b01_quota_test.go diff --git a/docs/DEVIATIONS.md b/docs/DEVIATIONS.md index 43fdeb9..c48d049 100644 --- a/docs/DEVIATIONS.md +++ b/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，实际在途可能超过该值。 diff --git a/internal/broker/b01_quota_test.go b/internal/broker/b01_quota_test.go new file mode 100644 index 0000000..3c1f0f4 --- /dev/null +++ b/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 +} diff --git a/internal/broker/broker_test.go b/internal/broker/broker_test.go index 69a66c2..bdabdb3 100644 --- a/internal/broker/broker_test.go +++ b/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 diff --git a/internal/broker/hooks.go b/internal/broker/hooks.go index 758da07..d63f239 100644 --- a/internal/broker/hooks.go +++ b/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