From e369f398d804f1bc9ed4098bce260abec262dd3c Mon Sep 17 00:00:00 2001 From: ryan Date: Sat, 15 Aug 2026 01:07:01 +0800 Subject: [PATCH] feat: enhance Safew error handling and polling mechanism - Added error_msg field to safewAPIResponse for better error descriptions. - Updated safewErrorDescription function to include error_msg in the output. - Introduced a new test to verify that error messages are included in Safew error descriptions. - Improved the polling mechanism in the Watcher to handle conflicts and ensure no overlapping polls occur. --- internal/adapter/safew.go | 24 ++++- internal/adapter/safew_chats_test.go | 7 ++ internal/safew/watcher.go | 147 +++++++++++++++++++++++---- internal/safew/watcher_test.go | 51 +++++++++- 4 files changed, 201 insertions(+), 28 deletions(-) diff --git a/internal/adapter/safew.go b/internal/adapter/safew.go index 4661b21..ac6390f 100644 --- a/internal/adapter/safew.go +++ b/internal/adapter/safew.go @@ -7,6 +7,7 @@ import ( "io" "net/http" "strings" + "time" ) const safewAPIBase = "https://api.safew.bot" @@ -29,6 +30,7 @@ type safewMessage struct { type safewAPIResponse struct { OK bool `json:"ok"` Description string `json:"description"` + ErrorMsg string `json:"error_msg"` } func (s *SafeWSender) Type() string { return "safew" } @@ -112,17 +114,22 @@ func (e *SafewAuthError) Error() string { return e.Description } +var safewLongPollClient = &http.Client{Timeout: 45 * time.Second} + func (s *SafeWSender) PollGroupChats(token string, offset int64, timeout int) ([]SafewChat, int64, error) { token = strings.TrimSpace(token) if token == "" { return nil, offset, fmt.Errorf("safew: token is required") } - reqBody, _ := json.Marshal(map[string]any{ + payload := map[string]any{ "timeout": timeout, - "offset": offset, "limit": 100, - }) - resp, err := http.Post(s.methodURL(token, "getUpdates"), "application/json", bytes.NewReader(reqBody)) + } + if offset > 0 { + payload["offset"] = offset + } + reqBody, _ := json.Marshal(payload) + resp, err := safewLongPollClient.Post(s.methodURL(token, "getUpdates"), "application/json", bytes.NewReader(reqBody)) if err != nil { return nil, offset, fmt.Errorf("safew getUpdates: %w", err) } @@ -183,7 +190,14 @@ func safewErrorDescription(body []byte) string { if err := json.Unmarshal(body, &api); err != nil { return strings.TrimSpace(string(body)) } - return api.Description + switch { + case api.Description != "" && api.ErrorMsg != "": + return api.Description + ": " + api.ErrorMsg + case api.ErrorMsg != "": + return api.ErrorMsg + default: + return api.Description + } } func escapeMarkdownV2(s string) string { diff --git a/internal/adapter/safew_chats_test.go b/internal/adapter/safew_chats_test.go index 84e60ad..c11ab05 100644 --- a/internal/adapter/safew_chats_test.go +++ b/internal/adapter/safew_chats_test.go @@ -99,6 +99,13 @@ func TestPollGroupChatsSuccess(t *testing.T) { } } +func TestSafewErrorDescriptionIncludesErrorMsg(t *testing.T) { + got := safewErrorDescription([]byte(`{"ok":false,"description":"BAD_REQUEST","error_msg":"Conflict: terminated by other getUpdates request"}`)) + if !strings.Contains(got, "Conflict") { + t.Fatalf("got %q, want error_msg", got) + } +} + func TestPollGroupChatsUnauthorized(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusUnauthorized) diff --git a/internal/safew/watcher.go b/internal/safew/watcher.go index 237ec04..ad11136 100644 --- a/internal/safew/watcher.go +++ b/internal/safew/watcher.go @@ -2,7 +2,9 @@ package safew import ( "context" + "errors" "log/slog" + "strings" "sync" "time" @@ -11,13 +13,44 @@ import ( type Poller func(token string, offset int64, timeout int) ([]adapter.SafewChat, int64, error) +const pollLockTTL = 60 * time.Second + +type runner struct { + cancel context.CancelFunc + firstDone chan struct{} + mu sync.Mutex + firstErr error +} + +func (r *runner) setFirst(err error) { + r.mu.Lock() + r.firstErr = err + r.mu.Unlock() + select { + case <-r.firstDone: + default: + close(r.firstDone) + } +} + +func (r *runner) waitFirst(ctx context.Context) error { + select { + case <-ctx.Done(): + return ctx.Err() + case <-r.firstDone: + r.mu.Lock() + defer r.mu.Unlock() + return r.firstErr + } +} + type Watcher struct { store ChatStore poll Poller bgIdle time.Duration mu sync.Mutex - running map[string]context.CancelFunc + running map[string]*runner stopped bool } @@ -26,39 +59,49 @@ func NewWatcher(store ChatStore, poll Poller) *Watcher { store: store, poll: poll, bgIdle: time.Second, - running: map[string]context.CancelFunc{}, + running: map[string]*runner{}, } } func (w *Watcher) Ensure(token string) { + w.ensure(token) +} + +func (w *Watcher) ensure(token string) *runner { if token == "" || w == nil { - return + return nil } w.mu.Lock() defer w.mu.Unlock() if w.stopped { - return + return nil } - if _, ok := w.running[token]; ok { - return + if r, ok := w.running[token]; ok { + return r } ctx, cancel := context.WithCancel(context.Background()) - w.running[token] = cancel - go w.loop(ctx, token) + r := &runner{cancel: cancel, firstDone: make(chan struct{})} + w.running[token] = r + go w.loop(ctx, token, r) + return r } func (w *Watcher) Stop() { w.mu.Lock() w.stopped = true - for _, cancel := range w.running { - cancel() + for _, r := range w.running { + r.cancel() } - w.running = map[string]context.CancelFunc{} + w.running = map[string]*runner{} w.mu.Unlock() } func (w *Watcher) Refresh(ctx context.Context, token string) error { - return w.pollOnce(ctx, token, 0) + r := w.ensure(token) + if r == nil { + return nil + } + return r.waitFirst(ctx) } func (w *Watcher) List(ctx context.Context, token, q string) ([]adapter.SafewChat, error) { @@ -69,28 +112,66 @@ func (w *Watcher) List(ctx context.Context, token, q string) ([]adapter.SafewCha return adapter.FilterSafewChats(chats, q), nil } -func (w *Watcher) loop(ctx context.Context, token string) { +func (w *Watcher) loop(ctx context.Context, token string, r *runner) { + for { + err := w.pollOnce(ctx, token, 0) + if ctx.Err() != nil { + r.setFirst(ctx.Err()) + return + } + if isSafewAuth(err) { + r.setFirst(err) + return + } + if isGetUpdatesConflict(err) { + if !w.sleep(ctx, conflictBackoff(w.bgIdle)) { + r.setFirst(ctx.Err()) + return + } + continue + } + r.setFirst(err) + break + } + for { if err := w.pollOnce(ctx, token, 30); err != nil { if ctx.Err() != nil { return } + if isSafewAuth(err) { + slog.Warn("safew watcher poll", "error", err) + return + } + if isGetUpdatesConflict(err) { + slog.Debug("safew watcher poll conflict", "error", err) + if !w.sleep(ctx, conflictBackoff(w.bgIdle)) { + return + } + continue + } slog.Warn("safew watcher poll", "error", err) } - idle := w.bgIdle - if idle <= 0 { - idle = time.Second - } - select { - case <-ctx.Done(): + if !w.sleep(ctx, w.bgIdle) { return - case <-time.After(idle): } } } +func (w *Watcher) sleep(ctx context.Context, d time.Duration) bool { + if d <= 0 { + d = time.Second + } + select { + case <-ctx.Done(): + return false + case <-time.After(d): + return true + } +} + func (w *Watcher) pollOnce(ctx context.Context, token string, timeout int) error { - ok, err := w.store.TrySafewPollLock(ctx, token, 35*time.Second) + ok, err := w.store.TrySafewPollLock(ctx, token, pollLockTTL) if err != nil { return err } @@ -115,3 +196,27 @@ func (w *Watcher) pollOnce(ctx context.Context, token string, timeout int) error } return nil } + +func isSafewAuth(err error) bool { + var e *adapter.SafewAuthError + return errors.As(err, &e) +} + +func isGetUpdatesConflict(err error) bool { + if err == nil { + return false + } + s := strings.ToLower(err.Error()) + return strings.Contains(s, "conflict") && strings.Contains(s, "getupdates") +} + +func conflictBackoff(idle time.Duration) time.Duration { + if idle <= 0 { + return 3 * time.Second + } + d := idle * 3 + if d < 3*time.Second { + return 3 * time.Second + } + return d +} diff --git a/internal/safew/watcher_test.go b/internal/safew/watcher_test.go index 37fc8c1..a4cd641 100644 --- a/internal/safew/watcher_test.go +++ b/internal/safew/watcher_test.go @@ -3,6 +3,7 @@ package safew import ( "context" "errors" + "sync/atomic" "testing" "time" @@ -13,7 +14,7 @@ func TestRefreshMergesAndAdvancesOffset(t *testing.T) { st := NewMemStore() poll := func(token string, offset int64, timeout int) ([]adapter.SafewChat, int64, error) { if timeout != 0 { - t.Fatalf("timeout=%d", timeout) + return nil, offset, nil } if offset != 0 { t.Fatalf("offset=%d", offset) @@ -36,6 +37,7 @@ func TestRefreshMergesAndAdvancesOffset(t *testing.T) { if off != 11 { t.Fatalf("offset=%d", off) } + w.Stop() } func TestRefreshAuthError(t *testing.T) { @@ -50,6 +52,44 @@ func TestRefreshAuthError(t *testing.T) { if _, ok := err.(*adapter.SafewAuthError); !ok { t.Fatalf("%T", err) } + w.Stop() +} + +type alwaysLockStore struct{ *MemStore } + +func (a *alwaysLockStore) TrySafewPollLock(context.Context, string, time.Duration) (bool, error) { + return true, nil +} + +func (a *alwaysLockStore) UnlockSafewPoll(context.Context, string) error { return nil } + +func TestRefreshAndEnsureDoNotOverlapPolls(t *testing.T) { + st := &alwaysLockStore{MemStore: NewMemStore()} + var inflight int32 + var overlap int32 + poll := func(token string, offset int64, timeout int) ([]adapter.SafewChat, int64, error) { + n := atomic.AddInt32(&inflight, 1) + if n > 1 { + atomic.AddInt32(&overlap, 1) + } + time.Sleep(40 * time.Millisecond) + atomic.AddInt32(&inflight, -1) + return []adapter.SafewChat{{ID: "1", Type: "group", Title: "g"}}, offset + 1, nil + } + w := NewWatcher(st, poll) + w.bgIdle = 20 * time.Millisecond + ctx := context.Background() + errCh := make(chan error, 1) + go func() { errCh <- w.Refresh(ctx, "tok") }() + w.Ensure("tok") + if err := <-errCh; err != nil { + t.Fatal(err) + } + time.Sleep(80 * time.Millisecond) + w.Stop() + if atomic.LoadInt32(&overlap) > 0 { + t.Fatalf("overlapping getUpdates: %d", overlap) + } } func TestEnsurePollsInBackground(t *testing.T) { @@ -57,7 +97,7 @@ func TestEnsurePollsInBackground(t *testing.T) { got := make(chan int, 1) w := NewWatcher(st, func(token string, offset int64, timeout int) ([]adapter.SafewChat, int64, error) { if timeout != 30 { - return nil, offset, errors.New("not background") + return nil, offset, nil } select { case got <- timeout: @@ -74,3 +114,10 @@ func TestEnsurePollsInBackground(t *testing.T) { } w.Stop() } + +func TestPollConflictIsRetryable(t *testing.T) { + err := errors.New("safew getUpdates status 400: BAD_REQUEST: Conflict: terminated by other getUpdates request; make sure that only one bot instance is running") + if !isGetUpdatesConflict(err) { + t.Fatal("expected conflict") + } +}