fix: 合入后修正迁移计数并唤醒回执推送

internal/app/message/push.go

@@ -226,6 +226,7 @@ LIMIT ?`, endpointID, room)
 			if rejErr := a.rejectTooLarge(ctx, it.seq, endpointID, it.senderID, nowMs); rejErr != nil {
 				return rejErr
 			}
+			a.WakePush(it.senderID)
 			continue
 		}
 		it.payload = payload
@@ -412,6 +413,7 @@ WHERE seq = ? AND endpoint_id = ? AND state = 'pending' AND pushed_conn = ?`,
 		if err != nil {
 			return err
 		}
+		a.WakePush(t.senderID)
 	}
 	return nil
 }
@@ -580,10 +582,10 @@ LIMIT ?`, endpointID, window+len(stale))
 		return err
 	}
 	type rcpt struct {
-		rid              int64
-		msgID, epID      string
-		state, reason    string
-		created          int64
+		rid           int64
+		msgID, epID   string
+		state, reason string
+		created       int64
 	}
 	var list []rcpt
 	for rows.Next() {

internal/app/message/recover.go

@@ -50,10 +50,13 @@ func (a *App) ExpireOnce(ctx context.Context, nowMs int64) error {
 		if time.Now().After(deadline) {
 			return nil
 		}
-		n, err := a.expireOnceBatch(ctx, nowMs)
+		n, senders, err := a.expireOnceBatch(ctx, nowMs)
 		if err != nil {
 			return err
 		}
+		for _, s := range senders {
+			a.WakePush(s)
+		}
 		if n == 0 {
 			break
 		}
@@ -62,8 +65,9 @@ func (a *App) ExpireOnce(ctx context.Context, nowMs int64) error {
 	return nil
 }
 
-func (a *App) expireOnceBatch(ctx context.Context, nowMs int64) (int, error) {
+func (a *App) expireOnceBatch(ctx context.Context, nowMs int64) (int, []string, error) {
 	var n int
+	var senders []string
 	err := a.db.Queue.Do(ctx, func(tx *sql.Tx) error {
 		if err := a.fillZombieExpireTx(tx, nowMs); err != nil {
 			return err
@@ -94,6 +98,7 @@ LIMIT ?`, nowMs, expireBatch)
 		}
 		_ = rows.Close()
 		n = len(list)
+		seen := map[string]struct{}{}
 		for _, it := range list {
 			state := DeliveryDropped
 			reason := ReasonOffline
@@ -104,10 +109,14 @@ LIMIT ?`, nowMs, expireBatch)
 			if err := a.finishDeliveryTx(tx, it.seq, it.endpointID, it.senderID, it.msgID, state, reason, false, nowMs); err != nil {
 				return err
 			}
+			if _, ok := seen[it.senderID]; !ok {
+				seen[it.senderID] = struct{}{}
+				senders = append(senders, it.senderID)
+			}
 		}
 		return finalizeStuckDispatchedTx(tx, nowMs, a.lim.RecordRetentionDays)
 	})
-	return n, err
+	return n, senders, err
 }
 
 func (a *App) fillZombieExpireTx(tx *sql.Tx, nowMs int64) error {

internal/app/message/review_c01_test.go

@@ -250,3 +250,43 @@ func itoa(i int) string {
 	}
 	return string(b[pos:])
 }
+
+func TestC01RejectTooLargeWakesSenderReceipt(t *testing.T) {
+	t.Parallel()
+	e := openDeliveryEnv(t, nil)
+	insertEndpoint(t, e.db, "alice", "", 1, 0)
+	insertEndpoint(t, e.db, "bob", "", 1, 0)
+	ctx := context.Background()
+	aliceConn := LiveConn{ConnID: "c-alice", Ready: true}
+	bobConn := LiveConn{ConnID: "c-bob", Ready: true, MaxReceiveBytes: 50}
+	e.conns.Set("alice", aliceConn)
+	e.conns.Set("bob", bobConn)
+	if err := e.app.OnHandshakeComplete(ctx, "alice", aliceConn); err != nil {
+		t.Fatal(err)
+	}
+	if err := e.app.OnHandshakeComplete(ctx, "bob", bobConn); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		e.app.stopPushWorker("alice", "")
+		e.app.stopPushWorker("bob", "")
+	})
+	req := baseSend("big-wake", "bob")
+	b := make([]byte, 80)
+	for i := range b {
+		b[i] = 'a'
+	}
+	req.Body.Data = string(b)
+	if _, err := e.app.Submit(ctx, "alice", port.ConnInfo{}, req); err != nil {
+		t.Fatal(err)
+	}
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if e.down.FilterType(protocol.TypeReceipt) > 0 {
+			return
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	t.Fatalf("sender got no receipt after too_large reject, receipts=%d msgs=%d",
+		e.down.FilterType(protocol.TypeReceipt), e.down.FilterType(protocol.TypeMsg))
+}

internal/app/message/submit.go

@@ -273,6 +273,7 @@ INSERT INTO messages(
 	if err != nil {
 		return SubmitResult{}, err
 	}
+	a.WakePush(senderID)
 	if result.State == StateDispatched {
 		a.wakeReceivers(ctx, result.ID, senderID)
 	}