Sakurasan
979973ddde
流式请求在首个内容事件到达前由网关缓冲（借鉴 magpie 的 holdWriter）：上游的错误 事件、空流、断连、协议形态不符（要流给了整包）都在放行前换渠道消化，客户端无感知。 扣住期间照常发 SSE 注释心跳；缓冲超过 1MB 直接放行退化为普通转发，判定失效不吞正文。 - 新增 proxy/hold.go：streamHold（缓冲/放行/心跳/首 token 计时与放行互斥）、 sseEventHasContent（chat/messages/responses 三种协议的内容判定）、 sseUpstreamError（200+流式形式的错误事件识别）、classifyUpstreamError + quotaSignals（余额/配额特征识别） - 非流式改为先读完整响应再写头：读失败（网络错误/超时）发生在写出之前，可换渠道 重试；此前先写 200 头再读，超时只能给客户端留一个 200 空响应 - 协议形态校验：要流却给了整包（didNotStream）按失败换渠道，反复如此的渠道会被 健康计数剔除 - 记账：每次尝试独立 trySink，只有成功走到 EOF 才并入请求级 sink——失败渠道的 message_start 用量不计入账单 - 冷却按原因分类（channel.FailKind）：余额/配额类第一次失败即冷却 30 分钟（429 不 一定是限流，智谱 429「余额不足」也在此列）；其余按连续进冷却次数指数退避 30s→60s→120s，封顶 10 分钟；短于退避值的 Retry-After 不再缩短冷却 - 新增 OT_PROXY_FIRST_TOKEN_WAIT（默认 0=不限）：放行前等内容的最长时间，超时即 取消本次尝试换下一个渠道 - 渠道列表接口返回 cooldown_reason，后台可区分「限流」与「余额耗尽」 - 测试：hold 缓冲/放行/心跳、内容与错误事件判定、失败分类与指数退避、容量热生效 - docs/magpie-plugin-mechanism.md：参考实现的插件机制调研
71 lines
1.6 KiB
Go
71 lines
1.6 KiB
Go
// Package cli 命令行入口。
|
|
package cli
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"log"
|
|
"net/http"
|
|
"os"
|
|
"os/signal"
|
|
"strconv"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/openteam/server/internal/api"
|
|
"github.com/openteam/server/internal/app"
|
|
"github.com/openteam/server/internal/config"
|
|
"github.com/openteam/server/internal/proxy"
|
|
)
|
|
|
|
// serve 启动 HTTP 服务（默认子命令）。
|
|
func serve() error {
|
|
cfg, err := config.Load()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
a, err := app.New(cfg)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer a.Shutdown(context.Background())
|
|
|
|
gw := proxy.NewGateway(a.DB, a.Enc, a.Usage, a.Limit, proxy.Options{
|
|
UserRPS: cfg.RateLimit.UserRPS,
|
|
LogRaw: cfg.Proxy.LogRaw,
|
|
GWTimeout: cfg.Proxy.Timeout,
|
|
MaxBodyMB: cfg.Proxy.MaxBodyMB,
|
|
FailoverBudget: cfg.Proxy.FailoverBudget,
|
|
StreamHeartbeat: cfg.Proxy.StreamHeartbeat,
|
|
FirstTokenWait: cfg.Proxy.FirstTokenWait,
|
|
})
|
|
router := api.NewRouter(a, gw)
|
|
|
|
srv := &http.Server{
|
|
Addr: ":" + strconv.Itoa(cfg.Port),
|
|
Handler: router,
|
|
ReadHeaderTimeout: 10 * time.Second,
|
|
}
|
|
|
|
go func() {
|
|
log.Printf("openteam listening on %s (env=%s)", srv.Addr, cfg.Env)
|
|
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
|
log.Fatalf("server: %v", err)
|
|
}
|
|
}()
|
|
|
|
quit := make(chan os.Signal, 1)
|
|
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
|
|
<-quit
|
|
|
|
log.Println("shutting down...")
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
if err := srv.Shutdown(ctx); err != nil {
|
|
log.Printf("server shutdown: %v", err)
|
|
}
|
|
a.Usage.Close()
|
|
return nil
|
|
}
|