From e7f27b2a6f1f9f1818c2c4e7512fc80b80633f1d Mon Sep 17 00:00:00 2001 From: Erzambayu Date: Thu, 9 Jul 2026 12:43:36 +0800 Subject: [PATCH] fix: sync presence timeout, url path escape, gateway client timeout - core/sync/presence.go: use context.WithTimeout(10s) for all Presence() calls to prevent goroutine leaks on network hang - core/sync/sync.go: url.PathEscape user input in UserByName to prevent path corruption with special characters - core/mitm/server.go: replace http.DefaultClient with gatewayClient (5min timeout) in intercept path, matching the passthroughClient pattern - core/sync/items.go + sync.go: O(1) dedup lookup maps for full-sync pull to avoid O(n^2) store scans at scale - server/handlers/api_debug.go: serialize proc access with mutex to prevent data race in concurrent Percent() calls --- core/mitm/server.go | 4 +- core/sync/items.go | 78 ++++++++++++++++++++++++++---------- core/sync/presence.go | 14 +++++-- core/sync/sync.go | 43 +++++++++++++++++++- server/handlers/api_debug.go | 4 ++ 5 files changed, 116 insertions(+), 27 deletions(-) diff --git a/core/mitm/server.go b/core/mitm/server.go index 90a3d4f..057f19f 100644 --- a/core/mitm/server.go +++ b/core/mitm/server.go @@ -143,7 +143,7 @@ func (s *Server) intercept(w http.ResponseWriter, r *http.Request, tool Tool) { } req.Header.Set("Content-Type", "application/json") req.Header.Set("Authorization", "Bearer "+s.apiKey) - resp, err := http.DefaultClient.Do(req) + resp, err := gatewayClient.Do(req) if err != nil { s.passthroughBody(w, r, body) return @@ -201,6 +201,8 @@ var passthroughClient = &http.Client{ }, } +var gatewayClient = &http.Client{Timeout: 5 * time.Minute} + func copyHeaders(dst, src http.Header) { for k, vs := range src { if strings.EqualFold(k, "Connection") || strings.EqualFold(k, "Transfer-Encoding") { diff --git a/core/sync/items.go b/core/sync/items.go index bf126ff..5e7618b 100644 --- a/core/sync/items.go +++ b/core/sync/items.go @@ -223,17 +223,25 @@ type syncCombo struct { // --- apply: pulled items → local rows --- +type fullSyncLookup struct { + accounts map[string]bool // shortHash(secret+creds) → true + keys map[string]bool // shortHash(secret) → true + proxies map[string]bool // shortHash(scheme+host+port+username) → true + prefixes map[string]bool // custom provider prefix → true +} + // applyFullItem applies one non-playlist pulled item. Returns true if handled. -func (m *Manager) applyFullItem(ctx context.Context, ri item) bool { +// lookup maps, when non-nil, provide O(1) dedup instead of scanning the store. +func (m *Manager) applyFullItem(ctx context.Context, ri item, lk *fullSyncLookup) bool { switch ri.Type { case typeCustomProvider: - return m.applyCustomProvider(ctx, ri) + return m.applyCustomProvider(ctx, ri, lk) case typeAccount: - return m.applyAccount(ctx, ri) + return m.applyAccount(ctx, ri, lk) case typeAPIKey: - return m.applyAPIKey(ctx, ri) + return m.applyAPIKey(ctx, ri, lk) case typeProxy: - return m.applyProxy(ctx, ri) + return m.applyProxy(ctx, ri, lk) case typeAlias: return m.applyAlias(ctx, ri) case typeCombo: @@ -242,7 +250,7 @@ func (m *Manager) applyFullItem(ctx context.Context, ri item) bool { return false } -func (m *Manager) applyCustomProvider(ctx context.Context, ri item) bool { +func (m *Manager) applyCustomProvider(ctx context.Context, ri item, lk *fullSyncLookup) bool { if m.custom == nil { return false } @@ -270,10 +278,16 @@ func (m *Manager) applyCustomProvider(ctx context.Context, ri item) bool { if json.Unmarshal([]byte(ri.Payload), &cp) != nil { return false } - // Upsert by prefix: skip if we already have this prefix, else create + register. - for _, e := range existing { - if e.Prefix == cp.Prefix { - return true // already present (LWW: keep local) + if lk != nil && lk.prefixes != nil { + if lk.prefixes[cp.Prefix] { + return true + } + } else { + // Upsert by prefix: skip if we already have this prefix, else create + register. + for _, e := range existing { + if e.Prefix == cp.Prefix { + return true // already present (LWW: keep local) + } } } id, err := m.custom.Create(ctx, cp) @@ -287,7 +301,7 @@ func (m *Manager) applyCustomProvider(ctx context.Context, ri item) bool { return true } -func (m *Manager) applyAccount(ctx context.Context, ri item) bool { +func (m *Manager) applyAccount(ctx context.Context, ri item, lk *fullSyncLookup) bool { if m.accounts == nil { return false } @@ -309,6 +323,14 @@ func (m *Manager) applyAccount(ctx context.Context, ri item) bool { if !ri.Encrypted { return false } + // Fast-path: skip decryption for accounts we already have locally. + // The item id embeds shortHash(secret+creds) which matches the dedup + // map key, so we check before any crypto or SQLite work. + if lk != nil && lk.accounts != nil { + if _, hash, ok := parseAccountID(ri.ItemID); ok && lk.accounts[hash] { + return true + } + } key := m.credKey(ctx) if key == nil { return false @@ -321,19 +343,21 @@ func (m *Manager) applyAccount(ctx context.Context, ri item) bool { if json.Unmarshal(raw, &sa) != nil { return false } - // Dedup: skip if an account with the same secret+creds already exists. - existing, _ := m.accounts.List(ctx, sa.Provider) - target := shortHash(sa.Secret + fmt.Sprint(sa.Creds)) - for _, e := range existing { - if shortHash(e.Secret+fmt.Sprint(e.Creds)) == target { - return true + // Fallback dedup when no lookup map was provided (legacy path). + if lk == nil || lk.accounts == nil { + target := shortHash(sa.Secret + fmt.Sprint(sa.Creds)) + existing, _ := m.accounts.List(ctx, sa.Provider) + for _, e := range existing { + if shortHash(e.Secret+fmt.Sprint(e.Creds)) == target { + return true + } } } _, _ = m.accounts.Add(ctx, store.Account{Provider: sa.Provider, Label: sa.Label, Secret: sa.Secret, Creds: sa.Creds, Status: sa.Status, Disabled: sa.Disabled}) return true } -func (m *Manager) applyProxy(ctx context.Context, ri item) bool { +func (m *Manager) applyProxy(ctx context.Context, ri item, lk *fullSyncLookup) bool { if m.proxies == nil { return false } @@ -366,6 +390,12 @@ func (m *Manager) applyProxy(ctx context.Context, ri item) bool { if json.Unmarshal(raw, &sp) != nil { return false } + target := shortHash(sp.Scheme + sp.Host + fmt.Sprint(sp.Port) + sp.Username) + if lk != nil && lk.proxies != nil { + if lk.proxies[target] { + return true + } + } // Add upserts on identity, so this is idempotent whether or not it exists. _, _ = m.proxies.Add(ctx, store.Proxy{ Label: sp.Label, Scheme: sp.Scheme, Host: sp.Host, Port: sp.Port, @@ -387,7 +417,7 @@ func parseAccountID(id string) (provider, hash string, ok bool) { return rest[:i], rest[i+1:], true } -func (m *Manager) applyAPIKey(ctx context.Context, ri item) bool { +func (m *Manager) applyAPIKey(ctx context.Context, ri item, lk *fullSyncLookup) bool { if m.keys == nil { return false } @@ -421,8 +451,14 @@ func (m *Manager) applyAPIKey(ctx context.Context, ri item) bool { if json.Unmarshal(raw, &sk) != nil { return false } - if existing, _ := m.keys.BySecret(ctx, sk.Secret); existing != nil { - return true // already have it + if lk != nil && lk.keys != nil { + if lk.keys[shortHash(sk.Secret)] { + return true + } + } else { + if existing, _ := m.keys.BySecret(ctx, sk.Secret); existing != nil { + return true // already have it + } } _, _ = m.keys.Add(ctx, store.APIKey{Label: sk.Label, Secret: sk.Secret, TokenLimit: sk.TokenLimit, MaxConcurrent: sk.MaxConcurrent, Enabled: sk.Enabled}) return true diff --git a/core/sync/presence.go b/core/sync/presence.go index 18ca35d..9992f6d 100644 --- a/core/sync/presence.go +++ b/core/sync/presence.go @@ -44,12 +44,18 @@ func (m *Manager) BrowserDisconnected() { close(presenceStop) presenceStop = nil // Best-effort "going offline" so the server doesn't wait for the TTL. - go m.Presence(context.Background(), false) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + go func() { + defer cancel() + m.Presence(ctx, false) + }() } } func (m *Manager) presenceLoop(stop <-chan struct{}) { - m.Presence(context.Background(), true) // immediate first beat + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + m.Presence(ctx, true) // immediate first beat + cancel() t := time.NewTicker(presenceInterval) defer t.Stop() for { @@ -57,7 +63,9 @@ func (m *Manager) presenceLoop(stop <-chan struct{}) { case <-stop: return case <-t.C: - m.Presence(context.Background(), true) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + m.Presence(ctx, true) + cancel() } } } diff --git a/core/sync/sync.go b/core/sync/sync.go index 9096778..3589b42 100644 --- a/core/sync/sync.go +++ b/core/sync/sync.go @@ -278,7 +278,7 @@ func (m *Manager) PublicProfile(ctx context.Context, id string) (string, error) // UserByName resolves a username to a user id (for @mention profile links). func (m *Manager) UserByName(ctx context.Context, name string) (string, error) { var raw json.RawMessage - if err := m.call(ctx, http.MethodGet, "/users/by-name/"+name, nil, &raw); err != nil { + if err := m.call(ctx, http.MethodGet, "/users/by-name/"+url.PathEscape(name), nil, &raw); err != nil { return "", err } return string(raw), nil @@ -1529,10 +1529,49 @@ func (m *Manager) Sync(ctx context.Context) (pushed, pulled int, err error) { if err := m.call(ctx, http.MethodGet, "/sync/pull?since="+fmt.Sprint(cursor), nil, &pull); err != nil { return pushed, 0, err } + + // Pre-build dedup lookup maps so each pulled item does O(1) checks + // instead of scanning the entire store per item (O(n^2) at 1300+ accounts). + var lk fullSyncLookup + if m.hasFullSync(ctx) { + if m.accounts != nil { + if all, err := m.accounts.List(ctx, ""); err == nil { + lk.accounts = make(map[string]bool, len(all)) + for _, a := range all { + lk.accounts[shortHash(a.Secret+fmt.Sprint(a.Creds))] = true + } + } + } + if m.keys != nil { + if all, err := m.keys.List(ctx); err == nil { + lk.keys = make(map[string]bool, len(all)) + for _, k := range all { + lk.keys[shortHash(k.Secret)] = true + } + } + } + if m.proxies != nil { + if all, err := m.proxies.List(ctx); err == nil { + lk.proxies = make(map[string]bool, len(all)) + for _, p := range all { + lk.proxies[shortHash(p.Scheme+p.Host+fmt.Sprint(p.Port)+p.Username)] = true + } + } + } + if m.custom != nil { + if all, err := m.custom.List(ctx); err == nil { + lk.prefixes = make(map[string]bool, len(all)) + for _, cp := range all { + lk.prefixes[cp.Prefix] = true + } + } + } + } + for _, ri := range pull.Items { if ri.Type != typePlaylist { // Full-sync types (accounts/keys/aliases/custom providers). - if m.applyFullItem(ctx, ri) { + if m.applyFullItem(ctx, ri, &lk) { pulled++ } continue diff --git a/server/handlers/api_debug.go b/server/handlers/api_debug.go index fc385c4..5c7ae36 100644 --- a/server/handlers/api_debug.go +++ b/server/handlers/api_debug.go @@ -5,6 +5,7 @@ import ( "os" "runtime" "runtime/debug" + "sync" "time" "github.com/shirou/gopsutil/v4/process" @@ -15,6 +16,7 @@ type Debug struct { version string started time.Time proc *process.Process + mu sync.Mutex } func NewDebug(version string, started time.Time) *Debug { @@ -35,12 +37,14 @@ func (h *Debug) Get(w http.ResponseWriter, _ *http.Request) { cpuPct := 0.0 var rss uint64 if h.proc != nil { + h.mu.Lock() if v, err := h.proc.Percent(0); err == nil { cpuPct = v } if mi, err := h.proc.MemoryInfo(); err == nil && mi != nil { rss = mi.RSS } + h.mu.Unlock() } lastGC := ""