Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Nixevol
|
8b7b78093b |
+8
-7
@@ -1719,11 +1719,12 @@ issue #3 未关闭，`feat/fix-3-downlink-deadlock` 未合入 `main`。下面是
|
||||
- 备选方案：为每个未测子项补验收用例（本波不做，避免为变绿放松断言）。
|
||||
- 影响：汇总改为通过 19、部分通过 4（F03/F08/F21/F22）、失败 0。F19 仍引用仓库内 SDK 清单、本波不重跑。
|
||||
|
||||
### 复审修复 R3-04
|
||||
### 复审修复 R3-08
|
||||
|
||||
- 日期：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 发布失败一次。
|
||||
1. **后台加人审计按实际插入数，全跳过不记 ok**
|
||||
- 日期：2026-09-30
|
||||
- 原条款：Gitea #72；`AdminAddMembers` 对已是成员与请求内重复编号静默跳过；审计用 `len(member_ids)-len(failed)` 当成功数。
|
||||
- 实际做法：`handleGroupAddMembers` 用加人前后 `group_members` 行数差得到 `added`；审计 `result` 用 `added` 与 `failed`（全跳过且无失败记 `noop`，不记 `ok`）；响应增加 `added`，`failed` 仍只含真正失败项。前端按 `added` 提示，已是成员不改成错误码。
|
||||
- 原因：全是已有成员时旧逻辑审计成 ok、界面提示已加人，实际插入 0 行。
|
||||
- 备选方案：把已是成员写入 `failed`（否决，会改变客户端错误语义）；扩展 `group.AddResult` 返回插入列表（可后续做，本波不改身份线接口）。
|
||||
- 影响：管理 API 加人成功体多 `added` 字段；契约文档示例仍写 `{"failed":[]}`，以本偏差为准。
|
||||
|
||||
@@ -316,6 +316,13 @@ func (h *Handler) handleGroupAddMembers(w http.ResponseWriter, r *http.Request)
|
||||
return
|
||||
}
|
||||
|
||||
before, err := h.groupMemberCount(r.Context(), id)
|
||||
if err != nil {
|
||||
h.auditP(p, "group_add_members", id, "error", ip)
|
||||
httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
|
||||
return
|
||||
}
|
||||
|
||||
res, err := h.groups.AdminAddMembers(r.Context(), id, req.MemberIDs)
|
||||
if err != nil {
|
||||
h.writeGroupErr(w, p, "group_add_members", id, ip, err)
|
||||
@@ -325,15 +332,29 @@ func (h *Handler) handleGroupAddMembers(w http.ResponseWriter, r *http.Request)
|
||||
if failed == nil {
|
||||
failed = []group.MemberFail{}
|
||||
}
|
||||
okN := len(req.MemberIDs) - len(failed)
|
||||
if okN < 0 {
|
||||
okN = 0
|
||||
|
||||
after, err := h.groupMemberCount(r.Context(), id)
|
||||
if err != nil {
|
||||
h.auditP(p, "group_add_members", id, "error", ip)
|
||||
httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误")
|
||||
return
|
||||
}
|
||||
h.auditPD(p, "group_add_members", id, batchAuditResult(okN, len(failed)), ip, map[string]any{
|
||||
added := after - before
|
||||
if added < 0 {
|
||||
added = 0
|
||||
}
|
||||
|
||||
// 已是成员/请求内重复编号会静默跳过，不算失败；审计用实际插入数，避免全跳过写成 ok。
|
||||
result := batchAuditResult(added, len(failed))
|
||||
if added == 0 && len(failed) == 0 {
|
||||
result = "noop"
|
||||
}
|
||||
h.auditPD(p, "group_add_members", id, result, ip, map[string]any{
|
||||
"members": req.MemberIDs,
|
||||
"failed": failed,
|
||||
"added": added,
|
||||
})
|
||||
httpx.WriteOK(w, map[string]any{"failed": failed})
|
||||
httpx.WriteOK(w, map[string]any{"failed": failed, "added": added})
|
||||
}
|
||||
|
||||
func (h *Handler) handleGroupRemoveMember(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -413,6 +434,13 @@ func (h *Handler) groupOwner(ctx context.Context, groupID string) (string, error
|
||||
return owner, err
|
||||
}
|
||||
|
||||
func (h *Handler) groupMemberCount(ctx context.Context, groupID string) (int, error) {
|
||||
var n int
|
||||
err := h.db.Read.QueryRowContext(ctx,
|
||||
`SELECT COUNT(*) FROM group_members WHERE group_id = ?`, groupID).Scan(&n)
|
||||
return n, err
|
||||
}
|
||||
|
||||
func (h *Handler) groupSummary(ctx context.Context, groupID string) (map[string]any, error) {
|
||||
var name, owner string
|
||||
var created int64
|
||||
|
||||
@@ -262,6 +262,7 @@ func TestH02BatchPartialImportAndGroups(t *testing.T) {
|
||||
insertEndpoint(t, db, "keep-1", "admin", true, false)
|
||||
insertEndpoint(t, db, "alice", "admin", true, false)
|
||||
insertEndpoint(t, db, "bob", "admin", true, false)
|
||||
insertEndpoint(t, db, "carol", "admin", true, false)
|
||||
|
||||
res := doReq(t, client, http.MethodPost, srv.URL+"/api/admin/endpoints/batch",
|
||||
`{"ids":["keep-1","missing-ep"],"action":"disable"}`,
|
||||
@@ -304,11 +305,24 @@ func TestH02BatchPartialImportAndGroups(t *testing.T) {
|
||||
}
|
||||
|
||||
res = doReq(t, client, http.MethodPost, srv.URL+"/api/admin/groups/"+created.ID+"/members",
|
||||
`{"member_ids":["keep-1","ghost-ep"]}`, csrf())
|
||||
`{"member_ids":["carol","ghost-ep"]}`, csrf())
|
||||
env = decodeEnv(t, res)
|
||||
if res.StatusCode != http.StatusOK || !env.OK {
|
||||
t.Fatalf("add members: %d %+v", res.StatusCode, env)
|
||||
}
|
||||
var addBody struct {
|
||||
Failed []any `json:"failed"`
|
||||
Added int `json:"added"`
|
||||
}
|
||||
if err := json.Unmarshal(env.Data, &addBody); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if addBody.Added != 1 {
|
||||
t.Fatalf("add response added=%d want 1", addBody.Added)
|
||||
}
|
||||
if len(addBody.Failed) != 1 {
|
||||
t.Fatalf("add response failed=%v want 1", addBody.Failed)
|
||||
}
|
||||
|
||||
res = doReq(t, client, http.MethodPost, srv.URL+"/api/admin/groups/"+created.ID+"/transfer",
|
||||
`{"endpoint_id":"bob"}`, csrf())
|
||||
@@ -346,8 +360,8 @@ func TestH02BatchPartialImportAndGroups(t *testing.T) {
|
||||
if ad == nil {
|
||||
t.Fatalf("add members missing detail: %v", add)
|
||||
}
|
||||
if add["result"] != "partial" && add["result"] != "failed" && add["result"] != "ok" {
|
||||
t.Fatalf("add members result=%v", add["result"])
|
||||
if add["result"] != "partial" {
|
||||
t.Fatalf("add members result=%v want partial", add["result"])
|
||||
}
|
||||
if _, ok := ad["members"]; !ok {
|
||||
t.Fatalf("add members missing members: %v", ad)
|
||||
@@ -355,6 +369,9 @@ func TestH02BatchPartialImportAndGroups(t *testing.T) {
|
||||
if _, ok := ad["failed"]; !ok {
|
||||
t.Fatalf("add members missing failed: %v", ad)
|
||||
}
|
||||
if ad["added"] != float64(1) {
|
||||
t.Fatalf("add members added=%v want 1", ad["added"])
|
||||
}
|
||||
|
||||
tr := lastAuditByAction(t, recs, "group_transfer")
|
||||
td, _ := tr["detail"].(map[string]any)
|
||||
@@ -364,3 +381,70 @@ func TestH02BatchPartialImportAndGroups(t *testing.T) {
|
||||
|
||||
assertNoSecrets(t, auditBuf.String(), "csv-pass-secret-1", "csv-pass-secret-2", testPassword)
|
||||
}
|
||||
|
||||
func TestH02AddMembersAllSkippedAuditNotOK(t *testing.T) {
|
||||
_, auditBuf, db, srv, client := setupH02(t)
|
||||
login(t, client, srv.URL)
|
||||
insertEndpoint(t, db, "alice", "admin", true, false)
|
||||
insertEndpoint(t, db, "bob", "admin", true, false)
|
||||
|
||||
res := doReq(t, client, http.MethodPost, srv.URL+"/api/admin/groups",
|
||||
`{"name":"一组","owner_id":"alice","member_ids":["bob"]}`, csrf())
|
||||
env := decodeEnv(t, res)
|
||||
if res.StatusCode != http.StatusOK || !env.OK {
|
||||
t.Fatalf("group create: %d %+v", res.StatusCode, env)
|
||||
}
|
||||
var created struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
if err := json.Unmarshal(env.Data, &created); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
auditBuf.Reset()
|
||||
var before int
|
||||
if err := db.Read.QueryRow(`SELECT COUNT(*) FROM group_members WHERE group_id = ?`, created.ID).Scan(&before); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
res = doReq(t, client, http.MethodPost, srv.URL+"/api/admin/groups/"+created.ID+"/members",
|
||||
`{"member_ids":["bob","bob","alice"]}`, csrf())
|
||||
env = decodeEnv(t, res)
|
||||
if res.StatusCode != http.StatusOK || !env.OK {
|
||||
t.Fatalf("add members: %d %+v", res.StatusCode, env)
|
||||
}
|
||||
var body struct {
|
||||
Failed []any `json:"failed"`
|
||||
Added int `json:"added"`
|
||||
}
|
||||
if err := json.Unmarshal(env.Data, &body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(body.Failed) != 0 {
|
||||
t.Fatalf("failed=%v want empty (already members are not errors)", body.Failed)
|
||||
}
|
||||
if body.Added != 0 {
|
||||
t.Fatalf("added=%d want 0", body.Added)
|
||||
}
|
||||
|
||||
var after int
|
||||
if err := db.Read.QueryRow(`SELECT COUNT(*) FROM group_members WHERE group_id = ?`, created.ID).Scan(&after); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if after != before {
|
||||
t.Fatalf("member rows before=%d after=%d want unchanged", before, after)
|
||||
}
|
||||
|
||||
recs := parseSlogJSON(t, auditBuf)
|
||||
add := lastAuditByAction(t, recs, "group_add_members")
|
||||
if add["result"] == "ok" {
|
||||
t.Fatalf("audit result must not be ok when all skipped: %v", add)
|
||||
}
|
||||
ad, _ := add["detail"].(map[string]any)
|
||||
if ad == nil {
|
||||
t.Fatalf("missing detail: %v", add)
|
||||
}
|
||||
if ad["added"] != float64(0) {
|
||||
t.Fatalf("detail.added=%v want 0", ad["added"])
|
||||
}
|
||||
}
|
||||
|
||||
@@ -156,66 +156,6 @@ 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)
|
||||
|
||||
@@ -3,7 +3,6 @@ package nixmsg
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
@@ -30,8 +29,6 @@ type FakeTransport struct {
|
||||
MaxFrameBytes int
|
||||
HelloDelay time.Duration
|
||||
ReceiveMaximumSet bool
|
||||
// failSendLeft 接下来若干次 type=send 的 PublishUp 返回错误（仍记入 up）。
|
||||
failSendLeft int
|
||||
}
|
||||
|
||||
type fakeConnect struct {
|
||||
@@ -68,13 +65,6 @@ 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...)
|
||||
@@ -90,17 +80,6 @@ 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
|
||||
}
|
||||
|
||||
|
||||
+2
-7
@@ -202,10 +202,7 @@ 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 {
|
||||
c.mu.Unlock()
|
||||
return
|
||||
}
|
||||
if item.epoch == epoch && item.inflight {
|
||||
item.inflight = false
|
||||
if c.inflight > 0 {
|
||||
c.inflight--
|
||||
@@ -216,10 +213,8 @@ func (c *Client) dispatchSend(tr transport, item *sendItem, rid string, payload
|
||||
c.finishSend(item, SendResult{}, errStop)
|
||||
return
|
||||
}
|
||||
// 保留 id/body/send_at_ms，换新 rid 后继续泵，避免 Send 永久挂起。
|
||||
c.regenerateSendLocked(item)
|
||||
}
|
||||
c.mu.Unlock()
|
||||
c.drainSendQueue()
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -296,7 +296,7 @@ export function deleteGroup(id: string) {
|
||||
|
||||
export function addGroupMembers(id: string, memberIds: string[]) {
|
||||
return run(() =>
|
||||
requestAdmin<{ failed: GroupFailed[] }>(
|
||||
requestAdmin<{ failed: GroupFailed[]; added: number }>(
|
||||
`/api/admin/groups/${encodeURIComponent(id)}/members`,
|
||||
{
|
||||
method: "POST",
|
||||
|
||||
+4
-2
@@ -691,13 +691,14 @@ export const mockApi = {
|
||||
return {};
|
||||
},
|
||||
|
||||
async addGroupMembers(id: string, memberIds: string[]): Promise<{ failed: GroupFailed[] }> {
|
||||
async addGroupMembers(id: string, memberIds: string[]): Promise<{ failed: GroupFailed[]; added: number }> {
|
||||
requireSession();
|
||||
const g = groups.find((x) => x.id === id);
|
||||
if (!g) {
|
||||
throw new ApiError("not_found", "群不存在", 404);
|
||||
}
|
||||
const failed: GroupFailed[] = [];
|
||||
let added = 0;
|
||||
for (const mid of memberIds) {
|
||||
if (!endpoints.some((e) => e.id === mid)) {
|
||||
failed.push({ id: mid, code: "not_found" });
|
||||
@@ -705,9 +706,10 @@ export const mockApi = {
|
||||
}
|
||||
if (!g.members.some((m) => m.id === mid)) {
|
||||
g.members.push({ id: mid, joined_at_ms: now() });
|
||||
added++;
|
||||
}
|
||||
}
|
||||
return { failed };
|
||||
return { failed, added };
|
||||
},
|
||||
|
||||
async removeGroupMember(id: string, endpointId: string): Promise<Record<string, never>> {
|
||||
|
||||
@@ -233,9 +233,15 @@ async function onAddMembers() {
|
||||
if (!ids.length) return;
|
||||
const res = await addGroupMembers(detail.value.id, ids);
|
||||
if (res.failed.length) {
|
||||
message.warning(`部分失败：${formatFailed(res.failed)}`);
|
||||
if (res.added > 0) {
|
||||
message.warning(`已加入 ${res.added} 人，部分失败：${formatFailed(res.failed)}`);
|
||||
} else {
|
||||
message.success("已加人");
|
||||
message.warning(`部分失败：${formatFailed(res.failed)}`);
|
||||
}
|
||||
} else if (res.added === 0) {
|
||||
message.info("没有新成员加入");
|
||||
} else {
|
||||
message.success(`已加人 ${res.added} 人`);
|
||||
}
|
||||
await loadDetail(detail.value.id);
|
||||
await load();
|
||||
|
||||
Reference in New Issue
Block a user