mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-11-03 11:45:04 +03:00
fix: keep panel restarts from hanging on the CPU job and Telegram (#7029)
Restarting the panel left it unreachable for up to a minute. CheckCpuJob sampled cpu.Percent over its whole one-minute period, so cron.Stop nearly always waited on it; it now diffs cpu.Times against its previous run (iowait counted as idle, as before). With Telegram unreachable, bus.Stop waited on an in-flight send and tgbot.Start ran getMe/setMyCommands inline; each bot call now returns once StopBot closes the bot's stop channel, the bot stops before the bus, and the artwork/commands warm-up runs in the background.
This commit is contained in:
@@ -0,0 +1,43 @@
|
||||
package tgbot
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
ta "github.com/mymmrac/telego/telegoapi"
|
||||
)
|
||||
|
||||
var errBotStopped = errors.New("telegram bot stopped")
|
||||
|
||||
// stoppableCaller returns as soon as its bot is stopped or the call's context
|
||||
// ends; fasthttp honours only deadlines, so a blocked route held panel restarts.
|
||||
type stoppableCaller struct {
|
||||
inner ta.Caller
|
||||
stopped <-chan struct{}
|
||||
}
|
||||
|
||||
type callResult struct {
|
||||
resp *ta.Response
|
||||
err error
|
||||
}
|
||||
|
||||
func (c stoppableCaller) Call(ctx context.Context, url string, data *ta.RequestData) (*ta.Response, error) {
|
||||
select {
|
||||
case <-c.stopped:
|
||||
return nil, errBotStopped
|
||||
default:
|
||||
}
|
||||
done := make(chan callResult, 1)
|
||||
go func() {
|
||||
resp, err := c.inner.Call(ctx, url, data)
|
||||
done <- callResult{resp, err}
|
||||
}()
|
||||
select {
|
||||
case r := <-done:
|
||||
return r.resp, r.err
|
||||
case <-c.stopped:
|
||||
return nil, errBotStopped
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
package tgbot
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/mymmrac/telego"
|
||||
ta "github.com/mymmrac/telego/telegoapi"
|
||||
"github.com/valyala/fasthttp"
|
||||
)
|
||||
|
||||
// With Telegram unreachable a send hangs for its whole fasthttp timeout, and
|
||||
// the event bus, so the panel restart behind it, waited that long.
|
||||
func TestStopBotEndsInFlightTelegramCall(t *testing.T) {
|
||||
arrived := make(chan struct{}, 1)
|
||||
release := make(chan struct{})
|
||||
hang := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {
|
||||
arrived <- struct{}{}
|
||||
<-release
|
||||
}))
|
||||
t.Cleanup(func() { close(release); hang.Close() })
|
||||
|
||||
stop := make(chan struct{})
|
||||
b, err := telego.NewBot(testBotToken, telego.WithAPIServer(hang.URL),
|
||||
telego.WithAPICaller(stoppableCaller{inner: ta.FastHTTPCaller{Client: &fasthttp.Client{}}, stopped: stop}))
|
||||
if err != nil {
|
||||
t.Fatalf("NewBot: %v", err)
|
||||
}
|
||||
origBot := liveBot()
|
||||
botMu.Lock()
|
||||
bot, botStop = b, stop
|
||||
botMu.Unlock()
|
||||
t.Cleanup(func() { setTestBot(origBot) })
|
||||
|
||||
errc := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := b.GetMe(context.Background())
|
||||
errc <- err
|
||||
}()
|
||||
select {
|
||||
case <-arrived:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("getMe never reached the fake Telegram API")
|
||||
}
|
||||
|
||||
StopBot()
|
||||
|
||||
select {
|
||||
case err := <-errc:
|
||||
if !errors.Is(err, errBotStopped) {
|
||||
t.Fatalf("getMe returned %v, want errBotStopped", err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("getMe still blocked 2s after StopBot")
|
||||
}
|
||||
if _, err := b.GetMe(context.Background()); !errors.Is(err, errBotStopped) {
|
||||
t.Fatalf("call after StopBot returned %v, want errBotStopped", err)
|
||||
}
|
||||
select {
|
||||
case <-arrived:
|
||||
t.Fatal("a call after StopBot still went out to Telegram")
|
||||
default:
|
||||
}
|
||||
}
|
||||
@@ -25,6 +25,7 @@ import (
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/web/service"
|
||||
|
||||
"github.com/mymmrac/telego"
|
||||
ta "github.com/mymmrac/telego/telegoapi"
|
||||
th "github.com/mymmrac/telego/telegohandler"
|
||||
"github.com/valyala/fasthttp"
|
||||
"github.com/valyala/fasthttp/fasthttpproxy"
|
||||
@@ -35,6 +36,8 @@ var (
|
||||
// botMu guards bot: EnsureBot may replace it while a background goroutine
|
||||
// (the screen artwork warm-up) is still using the previous liveBot().
|
||||
botMu sync.RWMutex
|
||||
// botStop is closed by StopBot to end every in-flight call of the live bot.
|
||||
botStop chan struct{}
|
||||
|
||||
// botCancel stores the function to cancel the context, stopping Long Polling gracefully.
|
||||
botCancel context.CancelFunc
|
||||
@@ -367,16 +370,24 @@ func (t *Tgbot) Start(i18nFS embed.FS) error {
|
||||
}
|
||||
|
||||
// Create new Telegram bot instance
|
||||
stop := make(chan struct{})
|
||||
botMu.Lock()
|
||||
bot, err = t.NewBot(tgBotToken, tgBotProxy, tgBotAPIServer)
|
||||
bot, err = t.NewBot(tgBotToken, tgBotProxy, tgBotAPIServer, stop)
|
||||
if err == nil {
|
||||
botStop = stop
|
||||
}
|
||||
botMu.Unlock()
|
||||
if err != nil {
|
||||
logger.Error("Failed to initialize Telegram bot API:", err)
|
||||
return err
|
||||
}
|
||||
|
||||
t.resolveScreenArt()
|
||||
t.trySetBotCommands(liveBot())
|
||||
// Off the start path: with Telegram unreachable these block for a minute,
|
||||
// holding the panel's signal loop and the sub server behind them.
|
||||
go func(b *telego.Bot) {
|
||||
t.resolveScreenArt()
|
||||
t.trySetBotCommands(b)
|
||||
}(liveBot())
|
||||
|
||||
// Start receiving Telegram bot messages
|
||||
tgBotMutex.Lock()
|
||||
@@ -462,7 +473,7 @@ func panelEgressDial(resolve func() string) fasthttp.DialFunc {
|
||||
}
|
||||
|
||||
// NewBot creates a new Telegram bot instance with optional proxy and API server settings.
|
||||
func (t *Tgbot) NewBot(token string, proxyUrl string, apiServerUrl string) (*telego.Bot, error) {
|
||||
func (t *Tgbot) NewBot(token string, proxyUrl string, apiServerUrl string, stopped <-chan struct{}) (*telego.Bot, error) {
|
||||
// Validate proxy URL if provided
|
||||
if proxyUrl != "" {
|
||||
if !isSupportedBotProxyScheme(proxyUrl) {
|
||||
@@ -490,7 +501,10 @@ func (t *Tgbot) NewBot(token string, proxyUrl string, apiServerUrl string) (*tel
|
||||
|
||||
// Build bot options
|
||||
var options []telego.BotOption
|
||||
options = append(options, telego.WithFastHTTPClient(client))
|
||||
options = append(options, telego.WithAPICaller(stoppableCaller{
|
||||
inner: ta.FastHTTPCaller{Client: client},
|
||||
stopped: stopped,
|
||||
}))
|
||||
|
||||
if apiServerUrl != "" {
|
||||
options = append(options, telego.WithAPIServer(apiServerUrl))
|
||||
@@ -603,6 +617,13 @@ func StopBot() {
|
||||
isRunning = false
|
||||
tgBotMutex.Unlock()
|
||||
|
||||
botMu.Lock()
|
||||
if botStop != nil {
|
||||
close(botStop)
|
||||
botStop = nil
|
||||
}
|
||||
botMu.Unlock()
|
||||
|
||||
userStateMgr.reset()
|
||||
addClientDrafts.resetAll()
|
||||
broadcastResetAll()
|
||||
|
||||
Reference in New Issue
Block a user