fix(connect): retry audit-sheet throttles; flush reports completed items

Two polish items surfaced during the digital-twin smoke test:

- audit-sheet append now retries transient throttles (THREADPOOL_BUSY /
  timeout / 429) up to 3x with backoff, instead of silently dropping the
  audit row on the first busy response. isTransientSheetErr keeps permanent
  errors (bad node id, permission) non-retried.
- flushDeferred reports WHAT it completed (a numbered per-item list), not just
  a count. When the owner is also the requester — the common self-request
  case — this list is their only completion receipt, since notifyRequester
  skips owner==requester to avoid double-messaging.

Tests: isTransientSheetErr classification, flush per-item completion report.
This commit is contained in:
修雨 committed 2026-06-15 16:32:41 +08:00
1 parent a10169344f
commit 4d831e5054
2 files changed
+95 -7

No files matched your search

+49 -7
View File
@@ -127,9 +127,38 @@ func (s *sheetAuditSink) record(ctx context.Context, req *ApprovalRequest) {
"sheetId": s.sheetID,
"values": [][]any{row},
})
if _, err := s.runner.Run(ctx, inv); err != nil {
fmt.Fprintf(os.Stderr, "[connect][approval][audit] 写审计表格失败 approvalId=%s: %v\n", req.ID, err)
// The online-sheet API throttles (e.g. THREADPOOL_BUSY) under bursts, dropping
// an audit row on a transient error. Retry a few times with backoff so the
// trail stays complete; still best-effort — give up (log) after the last try.
var err error
for attempt := 1; attempt <= 3; attempt++ {
if _, err = s.runner.Run(ctx, inv); err == nil {
return
}
if attempt < 3 && isTransientSheetErr(err) {
time.Sleep(time.Duration(attempt) * 600 * time.Millisecond)
continue
}
break
}
if err != nil {
fmt.Fprintf(os.Stderr, "[connect][approval][audit] 写审计表格失败(已重试)approvalId=%s: %v\n", req.ID, err)
}
}
// isTransientSheetErr reports whether a sheet-append error is a transient
// throttle/timeout worth retrying (vs. a permanent error like a bad node id).
func isTransientSheetErr(err error) bool {
if err == nil {
return false
}
s := strings.ToUpper(err.Error())
for _, sig := range []string{"BUSY", "THREADPOOL", "TIMEOUT", "RATE", "LIMIT", "TOO MANY", "503", "429"} {
if strings.Contains(s, sig) {
return true
}
}
return false
}
// approvalOrchestrator is the connector-side controller for the gate. It is
@@ -422,7 +451,8 @@ func (o *approvalOrchestrator) flushDeferred(ctx context.Context) bool {
_ = o.notifier.sendOTOText(ctx, []string{o.ownerUserID}, "当前没有积压的请求。")
return true
}
done, stuck := 0, 0
doneSummaries := make([]string, 0, len(pending))
stuck := 0
for _, req := range pending {
out, execErr := o.execute(ctx, req)
if execErr != nil {
@@ -434,13 +464,25 @@ func (o *approvalOrchestrator) flushDeferred(ctx context.Context) bool {
o.gate.markExecuted(req.ID)
o.recordAudit(ctx, req.ID)
o.notifyRequester(ctx, req, "(已恢复)已为你完成:"+req.Summary+"\n"+out)
done++
doneSummaries = append(doneSummaries, req.Summary)
}
// Report to the owner WHAT was done (not just a count) — when the owner is
// also the requester, this per-item list is their only completion receipt
// (notifyRequester skips owner==requester to avoid double-messaging).
var b strings.Builder
fmt.Fprintf(&b, "已补做 %d 个积压请求", len(doneSummaries))
if len(doneSummaries) > 0 {
b.WriteString(":\n")
for i, s := range doneSummaries {
fmt.Fprintf(&b, "%d. %s\n", i+1, s)
}
} else {
b.WriteString("。")
}
summary := fmt.Sprintf("已补做 %d 个积压请求。", done)
if stuck > 0 {
summary += fmt.Sprintf("仍有 %d 个未成功(身份可能还没对上,确认 dws 登录后再回复「重试」)。", stuck)
fmt.Fprintf(&b, "仍有 %d 个未成功(身份可能还没对上,确认 dws 登录后再回复「重试」)。", stuck)
}
_ = o.notifier.sendOTOText(ctx, []string{o.ownerUserID}, summary)
_ = o.notifier.sendOTOText(ctx, []string{o.ownerUserID}, strings.TrimRight(b.String(), "\n"))
return true
}
@@ -15,6 +15,7 @@ package helpers
import (
"context"
"fmt"
"strings"
"sync"
"testing"
@@ -559,6 +560,51 @@ func TestDeferredExecution_HoldsAndRetries(t *testing.T) {
}
}
// TestIsTransientSheetErr distinguishes retryable throttles from permanent errors.
func TestIsTransientSheetErr(t *testing.T) {
for _, e := range []string{"system error: THREADPOOL_BUSY", "server busy", "request timeout", "HTTP 429 too many requests"} {
if !isTransientSheetErr(fmt.Errorf("%s", e)) {
t.Fatalf("%q should be transient", e)
}
}
for _, e := range []string{"invalid nodeId", "permission denied", "not found"} {
if isTransientSheetErr(fmt.Errorf("%s", e)) {
t.Fatalf("%q should NOT be transient", e)
}
}
if isTransientSheetErr(nil) {
t.Fatal("nil is not transient")
}
}
// TestFlushDeferred_ReportsCompletedItems verifies the owner gets a per-item
// completion list (not just a count) when the backlog is flushed.
func TestFlushDeferred_ReportsCompletedItems(t *testing.T) {
t.Setenv("DWS_CONFIG_DIR", t.TempDir())
gate := newApprovalGate("ctflush")
runner := &fakeRunner{fail: true}
notifier := &fakeNotifier{}
o := newTextApprovalOrchestrator(gate, runner, "owner", notifier)
o.handleReply(context.Background(), "owner", "conv-o", `[[ACTION:todo.create title="我的待办"]]`, func(_ context.Context, _, _ string) error { return nil })
if gate.list()[0].State != approvalDeferred {
t.Fatalf("want deferred, got %s", gate.list()[0].State)
}
runner.setFail(false)
o.handleOwnerDecision(context.Background(), "owner", "重试")
notifier.mu.Lock()
var last string
if len(notifier.sent) > 0 {
last = notifier.sent[len(notifier.sent)-1].text
}
notifier.mu.Unlock()
if !strings.Contains(last, "已补做 1 个") || !strings.Contains(last, "创建待办:我的待办") {
t.Fatalf("flush summary should list the completed item, got %q", last)
}
}
// TestIsRetryWord checks whole-message retry-keyword matching.
func TestIsRetryWord(t *testing.T) {
for _, w := range []string{"重试", " 恢复 ", "retry", "继续"} {