From deb2398e27b553240792cd91636dcb67c14dfc95 Mon Sep 17 00:00:00 2001 From: Nixevol Date: Wed, 30 Sep 2026 15:05:42 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=A4=A7=E5=B8=A7=E5=90=8D=E9=A2=9D?= =?UTF-8?q?=E6=8C=89=20PacketID=20=E5=9C=A8=20PUBACK=20=E4=B8=8E=E6=96=AD?= =?UTF-8?q?=E7=BA=BF=E6=97=B6=E5=BD=92=E8=BF=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/DEVIATIONS.md | 9 ++ internal/broker/b02_large_test.go | 213 ++++++++++++++++++++++++++++++ internal/broker/broker.go | 100 ++++++++++---- internal/broker/hooks.go | 40 +++++- 4 files changed, 331 insertions(+), 31 deletions(-) create mode 100644 internal/broker/b02_large_test.go diff --git a/docs/DEVIATIONS.md b/docs/DEVIATIONS.md index c48d049..bdfd23f 100644 --- a/docs/DEVIATIONS.md +++ b/docs/DEVIATIONS.md @@ -1223,3 +1223,12 @@ issue #3 未关闭，`feat/fix-3-downlink-deadlock` 未合入 `main`。下面是 - 原因：mochi v2.7.9 在 `sendQuota>0` 时走 `NextImmediate` 递归读锁，并可因补发后删除 inflight 泄漏配额；已验证置 0 绕开整条路径。 - 备选方案：fork 修补 mochi（只修递归读锁仍观察到停滞）。 - 影响：服务端不再执行客户端 Receive Maximum；裸设备若带过小的 Receive Maximum，实际在途可能超过该值。 + +### 复审修复 B-02 + +- 日期：2026-09-30 +- 原条款：DEVELOPMENT 7.5 / DEVIATIONS N1/N2 第 4 条：大帧名额在 PUBACK、丢弃、断线时归还。Gitea #9。 +- 实际做法：`OnQosPublish` 按 PacketID 记下超过 64KiB 的出站包；`OnQosComplete`/`OnQosDropped`/断线按 ID 归还。获取名额最多等 5 秒，超时返回 `ErrLargeFrameTimeout`。Publish 未产生 inflight（无订阅者、队列丢弃）时立即归还。不采用「发布完成即归还」。 +- 原因：mochi 传给 `OnQosComplete` 的是 PUBACK，没有载荷，旧实现从未归还。 +- 备选方案：发布后立即归还（会把卡死点挪到消息包那份名额）。 +- 影响：只在 broker 保留一份全局 64 名额；确认超时仍由消息线踢线/清标记触发断线归还。 diff --git a/internal/broker/b02_large_test.go b/internal/broker/b02_large_test.go new file mode 100644 index 0000000..f353b73 --- /dev/null +++ b/internal/broker/b02_large_test.go @@ -0,0 +1,213 @@ +package broker + +import ( + "bytes" + "context" + "io" + "net" + "sync" + "testing" + "time" + + "git.asio.asia/nixevol/NixMsg/internal/app/port" + "github.com/mochi-mqtt/server/v2/packets" +) + +func TestLargeFrameQuotaReleasedOnPuback(t *testing.T) { + b, w, done := startTCPClient(t, "ep-large-ack") + defer func() { _ = b.Close() }() + defer func() { + _ = w.Close() + select { + case <-done: + case <-time.After(3 * time.Second): + } + }() + + writeConnect(t, w, "ep-large-ack", 30, 0) + readExactPacket(t, w, packets.Connack, 3*time.Second) + writeSubscribe(t, w, downTopic("ep-large-ack")) + readExactPacket(t, w, packets.Suback, 3*time.Second) + waitSession(t, b, "ep-large-ack") + + stop := make(chan struct{}) + defer close(stop) + var writeMu sync.Mutex + go autoPuback(w, stop, &writeMu) + + payload := bytes.Repeat([]byte("x"), 70*1024) + for i := 0; i < 65; i++ { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + err := b.PublishDown(ctx, "ep-large-ack", "", payload, port.PublishOpts{QoS: 1}) + cancel() + if err != nil { + t.Fatalf("publish %d: %v", i+1, err) + } + } + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if len(b.largeSem) == 0 { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("slots still held: %d", len(b.largeSem)) +} + +func TestLargeFrameQuotaReleasedOnDisconnect(t *testing.T) { + b, w, done := startTCPClient(t, "ep-large-disc") + defer func() { _ = b.Close() }() + + writeConnect(t, w, "ep-large-disc", 30, 0) + readExactPacket(t, w, packets.Connack, 3*time.Second) + writeSubscribe(t, w, downTopic("ep-large-disc")) + readExactPacket(t, w, packets.Suback, 3*time.Second) + waitSession(t, b, "ep-large-disc") + + go func() { + buf := make([]byte, 32*1024) + for { + _, err := w.Read(buf) + if err != nil { + return + } + } + }() + + payload := bytes.Repeat([]byte("y"), 70*1024) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + if err := b.PublishDown(ctx, "ep-large-disc", "", payload, port.PublishOpts{QoS: 1}); err != nil { + cancel() + t.Fatal(err) + } + cancel() + if len(b.largeSem) == 0 { + t.Fatal("expected a held large slot before disconnect") + } + _ = w.Close() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("client attach did not return") + } + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if len(b.largeSem) == 0 { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("slots after disconnect: %d", len(b.largeSem)) +} + +func TestLargeFrameQuotaReleasedWithoutSubscriber(t *testing.T) { + b, w, done := startTCPClient(t, "ep-large-nosub") + defer func() { _ = b.Close() }() + defer func() { + _ = w.Close() + select { + case <-done: + case <-time.After(3 * time.Second): + } + }() + + writeConnect(t, w, "ep-large-nosub", 30, 0) + readExactPacket(t, w, packets.Connack, 3*time.Second) + waitSession(t, b, "ep-large-nosub") + + payload := bytes.Repeat([]byte("z"), 70*1024) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + err := b.PublishDown(ctx, "ep-large-nosub", "", payload, port.PublishOpts{QoS: 1}) + cancel() + if err != nil { + t.Fatal(err) + } + if n := len(b.largeSem); n != 0 { + t.Fatalf("held slots without subscriber: %d", n) + } +} + +func startTCPClient(t *testing.T, _ string) (*Broker, net.Conn, chan struct{}) { + t.Helper() + b, err := New(Options{Authenticator: AllowAuthenticator{}}) + if err != nil { + t.Fatal(err) + } + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ln.Close() }) + done := make(chan struct{}) + go func() { + defer close(done) + 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) + } + return b, w, done +} + +func waitSession(t *testing.T, b *Broker, endpoint string) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if _, ok := b.ConnInfoOf(endpoint); ok { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("session not established") +} + +func autoPuback(w net.Conn, stop <-chan struct{}, writeMu *sync.Mutex) { + 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 + } + if hdr[0]>>4 != packets.Publish { + continue + } + qos := (hdr[0] >> 1) & 0x3 + 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() + } +} diff --git a/internal/broker/broker.go b/internal/broker/broker.go index 0ecfdd5..81e011b 100644 --- a/internal/broker/broker.go +++ b/internal/broker/broker.go @@ -34,6 +34,11 @@ var ErrPayloadTooLarge = errors.New("broker: payload exceeds client limit") // ErrNoConnection 目标端没有当前连接。 var ErrNoConnection = errors.New("broker: no active connection") +// ErrLargeFrameTimeout 全局大帧名额在有界等待内拿不到。 +var ErrLargeFrameTimeout = errors.New("broker: large frame quota timeout") + +const largeAcquireWait = 5 * time.Second + // AuthResult 是登录校验结论（N3 实现真实逻辑；N2 默认拒绝）。 type AuthResult struct { OK bool @@ -108,7 +113,8 @@ type connState struct { sessionToken string handshook bool subscribedDown bool - largeHeld int + largePIDs map[uint16]struct{} + largePending int metricsCounted bool mu sync.Mutex @@ -219,54 +225,98 @@ func (b *Broker) PublishDown(ctx context.Context, endpointID string, connID port } topic := downTopic(endpointID) large := len(payload) > largeFrameBytes - if large { - select { - case b.largeSem <- struct{}{}: - case <-ctx.Done(): - return ctx.Err() + if err := b.acquireLarge(ctx); err != nil { + return err } st.mu.Lock() - st.largeHeld++ + st.largePending++ st.mu.Unlock() } if err := b.server.Publish(topic, payload, false, qos); err != nil { if large { - b.releaseOneLarge(st) + b.finishLargePublish(st) } return err } - if large && qos == 0 { - b.releaseOneLarge(st) + if large { + b.finishLargePublish(st) } return nil } -func (b *Broker) releaseOneLarge(st *connState) { +func (b *Broker) acquireLarge(ctx context.Context) error { + timer := time.NewTimer(largeAcquireWait) + defer timer.Stop() + select { + case b.largeSem <- struct{}{}: + return nil + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return ErrLargeFrameTimeout + } +} + +func (b *Broker) releaseLargeSlot() { + select { + case <-b.largeSem: + default: + } +} + +func (b *Broker) finishLargePublish(st *connState) { + b.reconcileLargeInflight(st) st.mu.Lock() - if st.largeHeld > 0 { - st.largeHeld-- - st.mu.Unlock() - select { - case <-b.largeSem: - default: - } - return + n := st.largePending + st.largePending = 0 + st.mu.Unlock() + for i := 0; i < n; i++ { + b.releaseLargeSlot() + } +} + +func (b *Broker) releaseLargePID(st *connState, id uint16) { + st.mu.Lock() + _, ok := st.largePIDs[id] + if ok { + delete(st.largePIDs, id) } st.mu.Unlock() + if ok { + b.releaseLargeSlot() + } +} + +func (b *Broker) reconcileLargeInflight(st *connState) { + if st == nil { + return + } + st.mu.Lock() + ids := make([]uint16, 0, len(st.largePIDs)) + for id := range st.largePIDs { + ids = append(ids, id) + } + st.mu.Unlock() + for _, id := range ids { + if st.client != nil { + if _, ok := st.client.State.Inflight.Get(id); ok { + continue + } + } + b.releaseLargePID(st, id) + } } func (b *Broker) releaseAllLarge(st *connState) { st.mu.Lock() - n := st.largeHeld - st.largeHeld = 0 + n := len(st.largePIDs) + st.largePending + st.largePIDs = nil + st.largePending = 0 st.mu.Unlock() for i := 0; i < n; i++ { - select { - case <-b.largeSem: - default: - } + b.releaseLargeSlot() } } diff --git a/internal/broker/hooks.go b/internal/broker/hooks.go index d63f239..3638703 100644 --- a/internal/broker/hooks.go +++ b/internal/broker/hooks.go @@ -25,7 +25,9 @@ func (h *nixHook) Provides(b byte) bool { mqtt.OnPublishDropped, mqtt.OnSessionEstablished, mqtt.OnDisconnect, + mqtt.OnQosPublish, mqtt.OnQosComplete, + mqtt.OnQosDropped, mqtt.OnSubscribed, }, []byte{b}) } @@ -169,13 +171,13 @@ func (h *nixHook) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes [ func (h *nixHook) OnPublishDropped(cl *mqtt.Client, pk packets.Packet) { h.b.log.Debug("publish dropped", "client", cl.ID, "topic", pk.TopicName, "size", len(pk.Payload)) - if h.b.onDrop == nil { - return - } h.b.connsMu.RLock() st := h.b.byClient[cl] h.b.connsMu.RUnlock() - if st == nil { + if st != nil { + h.b.reconcileLargeInflight(st) + } + if h.b.onDrop == nil || st == nil { return } h.b.onDrop(context.Background(), st.endpointID, st.connID, append([]byte(nil), pk.Payload...)) @@ -263,7 +265,7 @@ func (h *nixHook) noteConnectionClose(st *connState) { st.metricsCounted = false } -func (h *nixHook) OnQosComplete(cl *mqtt.Client, pk packets.Packet) { +func (h *nixHook) OnQosPublish(cl *mqtt.Client, pk packets.Packet, _ int64, _ int) { if len(pk.Payload) <= largeFrameBytes { return } @@ -273,5 +275,31 @@ func (h *nixHook) OnQosComplete(cl *mqtt.Client, pk packets.Packet) { if st == nil { return } - h.b.releaseOneLarge(st) + st.mu.Lock() + if st.largePending > 0 { + st.largePending-- + } + if st.largePIDs == nil { + st.largePIDs = make(map[uint16]struct{}) + } + st.largePIDs[pk.PacketID] = struct{}{} + st.mu.Unlock() +} + +func (h *nixHook) OnQosComplete(cl *mqtt.Client, pk packets.Packet) { + h.releaseLargeByPacketID(cl, pk.PacketID) +} + +func (h *nixHook) OnQosDropped(cl *mqtt.Client, pk packets.Packet) { + h.releaseLargeByPacketID(cl, pk.PacketID) +} + +func (h *nixHook) releaseLargeByPacketID(cl *mqtt.Client, id uint16) { + h.b.connsMu.RLock() + st := h.b.byClient[cl] + h.b.connsMu.RUnlock() + if st == nil { + return + } + h.b.releaseLargePID(st, id) }