wangbo
132cda35d8
线上 P24 验收暴露出高频 kubectl exec 放大 K3s API 压力、门禁查询挤占关键连接池，以及 PostgreSQL 锁超时被误判为租约所有权丢失。 本次合并验收身份查询、在租约有效期内重试瞬态续期错误、修复人工审核残留 attempt，并增加滚动后 etcd 稳定窗口、节点直连指标和双站独立报告。 验证：Go 全量测试、go vet、聚焦 race、gofmt、迁移安全检查、bash -n、ShellCheck、manual release test。
48 lines
1.5 KiB
Go
48 lines
1.5 KiB
Go
package runner
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"time"
|
|
|
|
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
|
|
)
|
|
|
|
const (
|
|
taskExecutionLeaseTTL = 5 * time.Minute
|
|
taskExecutionRenewInterval = 30 * time.Second
|
|
leaseRenewalRetryInterval = 2 * time.Second
|
|
)
|
|
|
|
func (s *Service) renewTaskExecutionLease(ctx context.Context, cancel context.CancelFunc, taskID string, executionToken string) {
|
|
timer := time.NewTimer(taskExecutionRenewInterval)
|
|
defer timer.Stop()
|
|
leaseDeadline := time.Now().Add(taskExecutionLeaseTTL)
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-timer.C:
|
|
if err := s.coordinationStore.RenewTaskExecutionLease(ctx, taskID, executionToken, taskExecutionLeaseTTL); err != nil {
|
|
if errors.Is(err, store.ErrTaskExecutionFinished) {
|
|
return
|
|
}
|
|
if taskExecutionLeaseRenewalIsFatal(err, time.Now(), leaseDeadline) {
|
|
s.logger.Warn("task execution lease lost", "taskID", taskID, "error_category", "task_execution_lease_lost", "error", err)
|
|
cancel()
|
|
return
|
|
}
|
|
s.logger.Warn("task execution lease renewal failed; retrying before lease expiry", "taskID", taskID, "error_category", "task_execution_lease_renewal_transient", "error", err)
|
|
timer.Reset(leaseRenewalRetryInterval)
|
|
continue
|
|
}
|
|
leaseDeadline = time.Now().Add(taskExecutionLeaseTTL)
|
|
timer.Reset(taskExecutionRenewInterval)
|
|
}
|
|
}
|
|
}
|
|
|
|
func taskExecutionLeaseRenewalIsFatal(err error, now time.Time, leaseDeadline time.Time) bool {
|
|
return errors.Is(err, store.ErrTaskExecutionLeaseLost) || !now.Before(leaseDeadline)
|
|
}
|