fix: PublishUp 失败时换 rid 并继续泵发送队列

2026-09-30 19:18:47 +08:00
parent aa38cfc76b
commit 241531479e
4 changed files with 106 additions and 11 deletions
@@ -156,6 +156,66 @@ func TestFrameTooLargeLocal(t *testing.T) {
 	}
 }

+func TestPublishUpFailRetriesWithNewRID(t *testing.T) {
+	fake := NewFakeTransport()
+	fake.FailSendPublishN(1)
+	c := connectFake(t, fake)
+	defer c.Close()
+
+	at := time.UnixMilli(1_700_000_000_000)
+	fixedID := "msg-publish-retry"
+	done := make(chan struct{})
+	var firstRID, secondRID string
+
+	go func() {
+		defer close(done)
+		deadline := time.Now().Add(3 * time.Second)
+		for time.Now().Before(deadline) {
+			sends := fake.FindUp("send")
+			if len(sends) < 2 {
+				time.Sleep(5 * time.Millisecond)
+				continue
+			}
+			first, second := sends[0], sends[1]
+			firstRID, _ = first["rid"].(string)
+			secondRID, _ = second["rid"].(string)
+			if first["id"] != fixedID || second["id"] != fixedID {
+				t.Errorf("id changed: %v -> %v", first["id"], second["id"])
+			}
+			if first["send_at_ms"] != second["send_at_ms"] {
+				t.Errorf("send_at_ms changed: %v -> %v", first["send_at_ms"], second["send_at_ms"])
+			}
+			body1, _ := json.Marshal(first["body"])
+			body2, _ := json.Marshal(second["body"])
+			if string(body1) != string(body2) {
+				t.Errorf("body changed: %s -> %s", body1, body2)
+			}
+			if firstRID == "" || firstRID == secondRID {
+				t.Errorf("rid not regenerated: %q -> %q", firstRID, secondRID)
+			}
+			fake.ReplyOK(secondRID, map[string]any{
+				"id": fixedID, "send_at_ms": second["send_at_ms"], "state": "scheduled",
+			})
+			return
+		}
+		t.Error("timed out waiting for send retry")
+	}()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	res, err := c.Send(ctx, Target{Kind: "endpoint", ID: "b"}, Body{Enc: "utf8", Data: "hi"}, SendOptions{ID: fixedID, SendAt: &at})
+	<-done
+	if err != nil {
+		t.Fatalf("Send hung or failed: %v", err)
+	}
+	if res.ID != fixedID {
+		t.Fatalf("result id=%q want %q", res.ID, fixedID)
+	}
+	if firstRID == "" || secondRID == "" || firstRID == secondRID {
+		t.Fatalf("rids=%q/%q", firstRID, secondRID)
+	}
+}
+
 func TestResendKeepsIDAndSendAt(t *testing.T) {
 	fake := NewFakeTransport()
 	c := connectFake(t, fake)