diff --git a/docs/DEVIATIONS.md b/docs/DEVIATIONS.md index 7deb4c5..3a66731 100644 --- a/docs/DEVIATIONS.md +++ b/docs/DEVIATIONS.md @@ -1718,3 +1718,12 @@ issue #3 未关闭,`feat/fix-3-downlink-deadlock` 未合入 `main`。下面是 - 原因:原先 12 项备注写着未穷尽仍标通过,交付说明写成「通过 23」。 - 备选方案:为每个未测子项补验收用例(本波不做,避免为变绿放松断言)。 - 影响:汇总改为通过 19、部分通过 4(F03/F08/F21/F22)、失败 0。F19 仍引用仓库内 SDK 清单、本波不重跑。 + +### 复审修复 R3-04 + +- 日期:2026-09-30 +- 原条款:Gitea #68;`dispatchSend` 在 `PublishUp` 失败且未停止重连时清 inflight 后静默 return,`Send` 永久挂起。 +- 实际做法:失败且未 `stopReconnect` 时保留 id/body/send_at_ms,经 `regenerateSendLocked` 换新 rid,清本次 inflight 并 `drainSendQueue` 继续泵;已停止重连时仍 `finishSend` 返回错误。 +- 原因:条目已不在途,重连时的 `requeueInflightLocked` 不会捡回,调用方一直等 `result`。 +- 备选方案:失败即 `finishSend` 报错(否决,与断线/限速重交语义不一致);只换 rid 不唤醒队列(否决,等同 JS #69)。 +- 影响:仅 `sdk/go`;假传输可模拟 send 发布失败一次。 diff --git a/sdk/go/client_test.go b/sdk/go/client_test.go index 1b68b2c..7e4c7cb 100644 --- a/sdk/go/client_test.go +++ b/sdk/go/client_test.go @@ -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) diff --git a/sdk/go/fake_transport_test.go b/sdk/go/fake_transport_test.go index 0d43e0e..d312fbd 100644 --- a/sdk/go/fake_transport_test.go +++ b/sdk/go/fake_transport_test.go @@ -3,6 +3,7 @@ package nixmsg import ( "context" "encoding/json" + "errors" "sync" "sync/atomic" "time" @@ -29,6 +30,8 @@ type FakeTransport struct { MaxFrameBytes int HelloDelay time.Duration ReceiveMaximumSet bool + // failSendLeft 接下来若干次 type=send 的 PublishUp 返回错误(仍记入 up)。 + failSendLeft int } type fakeConnect struct { @@ -65,6 +68,13 @@ func (f *FakeTransport) Start(ctx context.Context, cfg transportConfig) error { return nil } +// FailSendPublishN 让接下来 n 次 send 帧 PublishUp 失败(hello 等其它类型不受影响)。 +func (f *FakeTransport) FailSendPublishN(n int) { + f.mu.Lock() + f.failSendLeft = n + f.mu.Unlock() +} + func (f *FakeTransport) PublishUp(payload []byte) error { f.mu.Lock() cp := append([]byte(nil), payload...) @@ -80,6 +90,17 @@ func (f *FakeTransport) PublishUp(payload []byte) error { if auto && head.Type == "hello" && head.RID != "" { f.replyHello(head.RID) } + if head.Type == "send" { + f.mu.Lock() + fail := f.failSendLeft > 0 + if fail { + f.failSendLeft-- + } + f.mu.Unlock() + if fail { + return errors.New("publish failed") + } + } return nil } diff --git a/sdk/go/send.go b/sdk/go/send.go index ca2d255..6c353b5 100644 --- a/sdk/go/send.go +++ b/sdk/go/send.go @@ -202,19 +202,24 @@ func (c *Client) dispatchSend(tr transport, item *sendItem, rid string, payload if err := tr.PublishUp(payload); err != nil { c.mu.Lock() delete(c.pending, rid) - if item.epoch == epoch && item.inflight { - item.inflight = false - if c.inflight > 0 { - c.inflight-- - } - if c.stopReconnect { - errStop := c.stopErrLocked() - c.mu.Unlock() - c.finishSend(item, SendResult{}, errStop) - return - } + if item.epoch != epoch || !item.inflight { + c.mu.Unlock() + return } + item.inflight = false + if c.inflight > 0 { + c.inflight-- + } + if c.stopReconnect { + errStop := c.stopErrLocked() + c.mu.Unlock() + c.finishSend(item, SendResult{}, errStop) + return + } + // 保留 id/body/send_at_ms,换新 rid 后继续泵,避免 Send 永久挂起。 + c.regenerateSendLocked(item) c.mu.Unlock() + c.drainSendQueue() return }