From e8e08799146abf17c94e965964bdfd7e0ebf7288 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sun, 11 Oct 2026 08:50:12 +0300 Subject: [PATCH] =?UTF-8?q?SQS:=20ValidateQueueName=20=D0=B2=20createTenan?= =?UTF-8?q?tQueue,=20JWT=20=D1=82=D1=80=D0=B5=D0=B1=D1=83=D0=B5=D1=82=20ex?= =?UTF-8?q?p,=20leader=20=D1=81=D0=B1=D1=80=D0=B0=D1=81=D1=8B=D0=B2=D0=B0?= =?UTF-8?q?=D0=B5=D1=82=20=D0=BB=D0=B8=D0=B4=D0=B5=D1=80=D1=81=D1=82=D0=B2?= =?UTF-8?q?=D0=BE=20=D0=BF=D1=80=D0=B8=20=D0=B4=D0=BB=D0=B8=D1=82=D0=B5?= =?UTF-8?q?=D0=BB=D1=8C=D0=BD=D0=BE=D0=BC=20=D1=81=D0=B1=D0=BE=D0=B5=20ref?= =?UTF-8?q?resh?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/admin/admin.go | 6 ++++++ app/auth/jwt.go | 5 ++++- app/leader/leader.go | 35 ++++++++++++++++++++++++++++++----- 3 files changed, 40 insertions(+), 6 deletions(-) diff --git a/app/admin/admin.go b/app/admin/admin.go index 90cbcb8..8977ee4 100644 --- a/app/admin/admin.go +++ b/app/admin/admin.go @@ -510,6 +510,12 @@ func (h *Handler) createTenantQueue(w http.ResponseWriter, r *http.Request) { jsonErr(w, http.StatusBadRequest, "name is required") return } + // Та же валидация имени, что и в SQS CreateQueue (иначе имена с ':'/'/' + // ломали бы ключ "{accessKey}:{name}"). + if err := gosqs.ValidateQueueName(req.Name); err != nil { + jsonErr(w, http.StatusBadRequest, "invalid queue name") + return + } key := t.AccessKey + ":" + req.Name models.SyncQueues.Lock() if _, exists := models.SyncQueues.Queues[key]; exists { diff --git a/app/auth/jwt.go b/app/auth/jwt.go index c154572..c032829 100644 --- a/app/auth/jwt.go +++ b/app/auth/jwt.go @@ -59,7 +59,10 @@ func ParseJWTClaims(token string) (*JWTClaims, error) { if claims.Sub == "" { return nil, fmt.Errorf("JWT missing 'sub' claim") } - if claims.Exp > 0 && claims.Exp < time.Now().Unix() { + if claims.Exp == 0 { + return nil, fmt.Errorf("JWT missing 'exp' claim") + } + if claims.Exp < time.Now().Unix() { return nil, fmt.Errorf("JWT token expired") } return &claims, nil diff --git a/app/leader/leader.go b/app/leader/leader.go index cb2666c..6fd9634 100644 --- a/app/leader/leader.go +++ b/app/leader/leader.go @@ -33,16 +33,21 @@ type Elector struct { isLeader bool stopCh chan struct{} stopped bool + + // lastRefresh — время последнего успешного refresh/acquire лока. + // Нужен для сброса лидерства при длительном сбое refresh (split-brain). + lastRefresh time.Time } // New — создаёт Elector. key — ключ лока, id — уникальный id инстанса. func New(key, id string, ttl, interval time.Duration) *Elector { return &Elector{ - key: key, - id: id, - ttl: ttl, - interval: interval, - stopCh: make(chan struct{}), + key: key, + id: id, + ttl: ttl, + interval: interval, + stopCh: make(chan struct{}), + lastRefresh: time.Now(), } } @@ -108,8 +113,15 @@ func (e *Elector) step() { ok, err := persistence.RefreshLock(e.key, e.id, e.ttl) if err != nil { log.Errorf("leader: refresh: %v", err) + // Refresh не удаётся: если с последнего успеха прошло > TTL — лок + // гарантированно истёк, и другой инстанс мог стать лидером (split-brain). + // Сбрасываем лидерство (fail-safe), а не остаёмся «лидером» без лока. + if time.Since(e.lastRefreshAt()) > e.ttl { + e.setLeader(false) + } return } + e.markRefreshed() if !ok { e.setLeader(false) } @@ -122,5 +134,18 @@ func (e *Elector) step() { } if ok { e.setLeader(true) + e.markRefreshed() } } + +func (e *Elector) lastRefreshAt() time.Time { + e.mu.RLock() + defer e.mu.RUnlock() + return e.lastRefresh +} + +func (e *Elector) markRefreshed() { + e.mu.Lock() + e.lastRefresh = time.Now() + e.mu.Unlock() +}