fix: 大帧名额按 PacketID 在 PUBACK 与断线时归还

docs/DEVIATIONS.md

@@ -1180,3 +1180,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 名额；确认超时仍由消息线踢线/清标记触发断线归还。

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()
+	}
+}

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 {
+		if err := b.acquireLarge(ctx); err != nil {
-		case b.largeSem <- struct{}{}:
+			return err
-		case <-ctx.Done():
-			return ctx.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 {
+	if large {
-		b.releaseOneLarge(st)
+		b.finishLargePublish(st)
 	}
 	return nil
 }
 
-func (b *Broker) releaseOneLarge(st *connState) {
+func (b *Broker) acquireLarge(ctx context.Context) error {
-	st.mu.Lock()
+	timer := time.NewTimer(largeAcquireWait)
-	if st.largeHeld > 0 {
+	defer timer.Stop()
-		st.largeHeld--
+	select {
-		st.mu.Unlock()
+	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:
 	}
-		return
+}
+
+func (b *Broker) finishLargePublish(st *connState) {
+	b.reconcileLargeInflight(st)
+	st.mu.Lock()
+	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
+	n := len(st.largePIDs) + st.largePending
-	st.largeHeld = 0
+	st.largePIDs = nil
+	st.largePending = 0
 	st.mu.Unlock()
 	for i := 0; i < n; i++ {
-		select {
+		b.releaseLargeSlot()
-		case <-b.largeSem:
-		default:
-		}
 	}
 }
 

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)
 }