fix(review): 处理 PR #5666 四个阻断项(B1-B4)

- B1: 移除根目录 docker-compose.yml(个人镜像测试产物混入)
- B2: 新增迁移 224 放宽 user_platform_quotas CHECK 至 8 平台,补
  BulkInsertInitial 国产平台集成测试与迁移内容断言测试
- B3: CC/Responses×anthropic 四个上游读循环接入间隔超时泵
  (gateway.stream_data_interval_timeout,默认 180s):上游挂住 SSE
  不发数据也不断连时结束排水/组装、关闭 resp.Body 归还连接池位,
  排水超时仍按已累计 usage 返回(与 messages 主路径同语义)
- B4: 配额/余额探测发起前过 cnValidateProbeURL(与网关转发同一套
  security.url_allowlist 策略),被拒时零出站、API key 不离开本机
This commit is contained in:
Randark
2026-08-17 04:07:06 +00:00
parent 901a0439f1
commit 4b667ccd45
12 changed files with 795 additions and 354 deletions
@@ -98,6 +98,35 @@ func TestUserPlatformQuotaRepository_BulkInsertInitial_GrokAllowed(t *testing.T)
require.InDelta(t, 9.0, *rec.DailyLimitUSD, 1e-9)
}
// TestUserPlatformQuotaRepository_BulkInsertInitial_CNProvidersAllowed 回归迁移 224:
// kimi/zhipu/deepseek 平台必须能写入 user_platform_quotas(CHECK 约束已含国产供应商)。
// 历史 bug:三个平台不在约束内 → 注册预填充 8 平台默认配额时整条多行 INSERT 中止 →
// fail-open 吞错 → 新用户拿到零条配额记录(缺失配额行 = 无限额)。
func TestUserPlatformQuotaRepository_BulkInsertInitial_CNProvidersAllowed(t *testing.T) {
ctx := context.Background()
tx := testEntTx(t)
txCtx := dbent.NewTxContext(ctx, tx)
client := tx.Client()
userID := mustCreateUserForQuota(t, client)
repo := NewUserPlatformQuotaRepository(client)
daily := 12.0
records := []UserPlatformQuotaRecord{
{UserID: userID, Platform: "kimi", DailyLimitUSD: &daily},
{UserID: userID, Platform: "zhipu"},
{UserID: userID, Platform: "deepseek"},
}
require.NoError(t, repo.BulkInsertInitial(txCtx, records),
"kimi/zhipu/deepseek 平台应可写入(迁移 224 后 CHECK 约束已含国产供应商)")
for _, platform := range []string{"kimi", "zhipu", "deepseek"} {
rec, err := repo.GetByUserPlatform(txCtx, userID, platform)
require.NoError(t, err)
require.NotNil(t, rec, "%s 配额行应已写入", platform)
}
}
func TestUserPlatformQuotaRepository_GetByUserPlatform(t *testing.T) {
ctx := context.Background()
tx := testEntTx(t)
@@ -125,6 +125,13 @@ func (s *CNProviderBalanceService) queryBalance(ctx context.Context, accountID i
}
targetURL := cnBalanceURL(account)
// 探测发起前过出站 URL 安全策略(与网关转发/Grok 探测同一套校验):
// DeepSeek 端点由账号 base_url 衍生,不得把 API key 发往策略外主机。
validatedURL, err := cnValidateProbeURL(s.cfg, targetURL)
if err != nil {
return nil, infraerrors.New(http.StatusForbidden, "CN_BALANCE_URL_REJECTED", err.Error())
}
targetURL = validatedURL
proxyURL := s.resolveProxyURL(ctx, account)
callCtx, cancel := context.WithTimeout(ctx, cnBalanceUpstreamTimeout)
defer cancel()
@@ -0,0 +1,48 @@
package service
// CN 供应商探测端点的出站 URL 安全策略校验(配额/余额探测共用)。
//
// 背景(review B4):这两条探测路径会把账号 API key 发往 base_url 衍生端点,
// 此前完全绕过 security.url_allowlist——在加固部署里构成任意外发与内网探测
// 面(本项目此前发生过账号测试 SSRF 生产事件)。与网关转发
// (validateUpstreamBaseURL)、Grok 探测(grokOperatorPolicyValidator)一致,
// 探测发起前必须过同一套运营者策略。
import (
"errors"
"fmt"
"strings"
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/Wei-Shaw/sub2api/internal/util/urlvalidator"
)
// cnValidateProbeURL 按全局出站 URL 安全策略校验探测端点,返回规范化 URL。
// 白名单开启时强制 UpstreamHosts(阻断私网与未列名主机);关闭时仅做格式
// 校验(HTTP 允许与否跟随配置);cfg 为 nil 时退化为纯格式校验。
func cnValidateProbeURL(cfg *config.Config, raw string) (string, error) {
trimmed := strings.TrimSpace(raw)
if trimmed == "" {
return "", errors.New("probe url is required")
}
if cfg != nil && cfg.Security.URLAllowlist.Enabled {
normalized, err := urlvalidator.ValidateHTTPSURL(trimmed, urlvalidator.ValidationOptions{
AllowedHosts: cfg.Security.URLAllowlist.UpstreamHosts,
RequireAllowlist: true,
AllowPrivate: cfg.Security.URLAllowlist.AllowPrivateHosts,
})
if err != nil {
return "", fmt.Errorf("probe target rejected by URL security policy: %w", err)
}
return normalized, nil
}
var allowInsecureHTTP bool
if cfg != nil {
allowInsecureHTTP = cfg.Security.URLAllowlist.AllowInsecureHTTP
}
normalized, err := urlvalidator.ValidateURLFormat(trimmed, allowInsecureHTTP)
if err != nil {
return "", fmt.Errorf("probe target rejected by URL security policy: %w", err)
}
return normalized, nil
}
@@ -0,0 +1,129 @@
package service
// CN 供应商探测端点 URL 安全策略回归测试(review B4):
// 配额/余额探测不得绕过 security.url_allowlist——base_url 衍生的探测端点
// 必须先过运营者策略,被拒绝时不得发起任何上游请求(API key 不出站)。
import (
"context"
"net/http"
"testing"
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint"
"github.com/stretchr/testify/require"
)
func cnProbeAllowlistConfig(hosts ...string) *config.Config {
return &config.Config{
Security: config.SecurityConfig{
URLAllowlist: config.URLAllowlistConfig{
Enabled: true,
UpstreamHosts: hosts,
},
},
}
}
func TestCNValidateProbeURL_AllowlistPolicy(t *testing.T) {
cfg := cnProbeAllowlistConfig("api.moonshot.cn", "api.deepseek.com")
// 白名单内主机放行(保留完整路径)。
ok, err := cnValidateProbeURL(cfg, "https://api.moonshot.cn/v1/users/me/balance")
require.NoError(t, err)
require.Equal(t, "https://api.moonshot.cn/v1/users/me/balance", ok)
// 白名单外主机拒绝。
_, err = cnValidateProbeURL(cfg, "https://relay.attacker.example/v1/usages")
require.Error(t, err)
require.Contains(t, err.Error(), "rejected by URL security policy")
// 私网主机拒绝(内网探测面)。
_, err = cnValidateProbeURL(cfg, "http://169.254.169.254/latest/meta-data")
require.Error(t, err)
// 白名单关闭:仅格式校验,任意 https 主机放行。
formatOnly, err := cnValidateProbeURL(&config.Config{
Security: config.SecurityConfig{URLAllowlist: config.URLAllowlistConfig{Enabled: false}},
}, "https://relay.attacker.example/v1/usages")
require.NoError(t, err)
require.Equal(t, "https://relay.attacker.example/v1/usages", formatOnly)
}
// recordingHTTPUpstream 断言探测被策略拒绝时没有任何上游请求发出。
type recordingHTTPUpstream struct{ calls int }
func (u *recordingHTTPUpstream) Do(req *http.Request, proxyURL string, accountID int64, accountConcurrency int) (*http.Response, error) {
u.calls++
return nil, context.DeadlineExceeded
}
func (u *recordingHTTPUpstream) DoWithTLS(req *http.Request, proxyURL string, accountID int64, accountConcurrency int, profile *tlsfingerprint.Profile) (*http.Response, error) {
u.calls++
return nil, context.DeadlineExceeded
}
type fakeCNProbeAccountRepo struct {
AccountRepository
account *Account
}
func (r *fakeCNProbeAccountRepo) GetByID(ctx context.Context, id int64) (*Account, error) {
return r.account, nil
}
// kimi coding 账号的 base_url 指向中转(含 api.kimi.com/coding 路径段即可被识别
// 为 kimi coding plan)→ 衍生额度端点落在中转主机上,白名单未列名必须拒绝。
func TestCNProviderQuotaService_RejectsURLBlockedByPolicy(t *testing.T) {
repo := &fakeCNProbeAccountRepo{account: &Account{
ID: 1, Platform: PlatformKimi, Type: AccountTypeAPIKey, Status: StatusActive,
Credentials: map[string]any{
"account_mode": "coding",
"api_key": "sk-test",
"base_url": "https://relay.attacker.example/api.kimi.com/coding",
},
}}
upstream := &recordingHTTPUpstream{}
svc := NewCNProviderQuotaService(repo, nil, upstream, cnProbeAllowlistConfig("api.kimi.com"))
_, err := svc.QueryUsage(context.Background(), 1)
require.Error(t, err)
require.Contains(t, err.Error(), "CN_QUOTA_URL_REJECTED")
require.Zero(t, upstream.calls, "probe must not issue any upstream request when URL policy rejects the target")
}
// deepseek payg 账号自定义 base_url → 余额端点落在中转主机上,必须先过策略。
func TestCNProviderBalanceService_RejectsURLBlockedByPolicy(t *testing.T) {
repo := &fakeCNProbeAccountRepo{account: &Account{
ID: 2, Platform: PlatformDeepseek, Type: AccountTypeAPIKey, Status: StatusActive,
Credentials: map[string]any{
"account_mode": "payg",
"api_key": "sk-test",
"base_url": "https://relay.attacker.example",
},
}}
upstream := &recordingHTTPUpstream{}
svc := NewCNProviderBalanceService(repo, nil, upstream, cnProbeAllowlistConfig("api.deepseek.com"))
_, err := svc.QueryBalance(context.Background(), 2)
require.Error(t, err)
require.Contains(t, err.Error(), "CN_BALANCE_URL_REJECTED")
require.Zero(t, upstream.calls, "probe must not issue any upstream request when URL policy rejects the target")
}
// 白名单包含官方主机的正常路径:URL 校验通过后才发出上游请求(此处允许到达
// httpUpstream 层即视为通过校验,不发真实网络)。
func TestCNProviderBalanceService_OfficialHostPassesValidation(t *testing.T) {
repo := &fakeCNProbeAccountRepo{account: &Account{
ID: 3, Platform: PlatformDeepseek, Type: AccountTypeAPIKey, Status: StatusActive,
Credentials: map[string]any{
"account_mode": "payg",
"api_key": "sk-test",
},
}}
upstream := &recordingHTTPUpstream{}
svc := NewCNProviderBalanceService(repo, nil, upstream, cnProbeAllowlistConfig("api.deepseek.com"))
_, _ = svc.QueryBalance(context.Background(), 3)
require.Equal(t, 1, upstream.calls, "official host must pass URL policy and reach the upstream layer")
}
@@ -144,6 +144,14 @@ func (s *CNProviderQuotaService) queryUsage(ctx context.Context, accountID int64
authHeader = apiKey // 智谱额度端点鉴权不加 Bearer 前缀
}
// 探测发起前过出站 URL 安全策略(与网关转发/Grok 探测同一套校验):
// 端点多由账号 base_url 衍生,不得把 API key 发往策略外主机。
validatedURL, err := cnValidateProbeURL(s.cfg, targetURL)
if err != nil {
return nil, infraerrors.New(http.StatusForbidden, "CN_QUOTA_URL_REJECTED", err.Error())
}
targetURL = validatedURL
proxyURL := s.resolveProxyURL(ctx, account)
callCtx, cancel := context.WithTimeout(ctx, cnQuotaUpstreamTimeout)
defer cancel()
@@ -0,0 +1,126 @@
package service
// 国产供应商 Anthropic 协议转换路径的上游 SSE 行泵。
//
// 这两条转换链(CC×anthropic / Responses×anthropic)的上游 ctx 是
// WithoutCancel(detachStreamUpstreamContext)、http.Client 无整体 Timeout,
// 客户端断开后的排水阶段若上游挂住 SSE(不发数据也不断连),
// scanner.Scan() 将永久阻塞:goroutine 钉死、resp.Body 不归还、连接池位
// 被占用、usage 永不落库。
//
// 与 handleAnthropicStreamingResponse / readOpenAICompatBufferedTerminal 的
// 同类排水一致,本泵用 gateway.stream_data_interval_timeout(默认 180s)作为
// 逐行读间隔上限,超时即向调用方返回 errAnthropicNativeStreamIdle,由调用方
// 关闭 resp.Body 解除阻塞的读并结束排水。
import (
"bufio"
"errors"
"io"
"time"
)
// errAnthropicNativeStreamIdle 表示上游流读间隔超时(见上方文件注释)。
var errAnthropicNativeStreamIdle = errors.New("stream data interval timeout")
// anthropicNativeLineEvent 是行泵交付的单次读取结果:line 为一行 SSE 文本,
// err 为 scanner 读错误(流自然结束时 next 返回 io.EOF,不经过本字段)。
type anthropicNativeLineEvent struct {
line string
err error
}
// anthropicNativeLinePump 以独立 goroutine 泵送 scanner 的行,并对逐行到达
// 间隔施加 interval 上限(<=0 表示禁用,保持无界读的旧行为)。
type anthropicNativeLinePump struct {
events chan anthropicNativeLineEvent
done chan struct{}
timer *time.Timer
interval time.Duration
}
// newAnthropicNativeLinePump 启动泵 goroutine;调用方 defer pump.stop()。
func newAnthropicNativeLinePump(scanner *bufio.Scanner, interval time.Duration) *anthropicNativeLinePump {
p := &anthropicNativeLinePump{
events: make(chan anthropicNativeLineEvent, 16),
done: make(chan struct{}),
interval: interval,
}
if interval > 0 {
p.timer = time.NewTimer(interval)
}
go func() {
defer close(p.events)
for scanner.Scan() {
select {
case p.events <- anthropicNativeLineEvent{line: scanner.Text()}:
case <-p.done:
return
}
}
if err := scanner.Err(); err != nil {
select {
case p.events <- anthropicNativeLineEvent{err: err}:
case <-p.done:
}
}
}()
return p
}
// next 阻塞返回下一行。返回 io.EOF 表示上游正常收流;errAnthropicNativeStreamIdle
// 表示 interval 内无任何数据到达(计时从收到上一行时起算,事件处理耗时不算入,
// 与 readOpenAICompatBufferedTerminal 的 resetTimeout 语义一致)。
func (p *anthropicNativeLinePump) next() (string, error) {
var timeoutCh <-chan time.Time
if p.timer != nil {
timeoutCh = p.timer.C
}
select {
case ev, ok := <-p.events:
if !ok {
return "", io.EOF
}
p.resetTimer()
return ev.line, ev.err
case <-timeoutCh:
return "", errAnthropicNativeStreamIdle
}
}
// resetTimer 在收到一行后重启间隔计时器。
func (p *anthropicNativeLinePump) resetTimer() {
if p.timer == nil {
return
}
if !p.timer.Stop() {
select {
case <-p.timer.C:
default:
}
}
p.timer.Reset(p.interval)
}
// stop 终止泵 goroutine。注意:goroutine 若正阻塞在 scanner.Read 上,需由
// 调用方关闭 resp.Body(间隔超时分支已做)才能真正退出。
func (p *anthropicNativeLinePump) stop() {
close(p.done)
if p.timer != nil {
if !p.timer.Stop() {
select {
case <-p.timer.C:
default:
}
}
}
}
// anthropicNativeStreamInterval 返回本组转换路径适用的读间隔上限;
// gateway.stream_data_interval_timeout <= 0 时视为禁用。
func (s *OpenAIGatewayService) anthropicNativeStreamInterval() time.Duration {
if s.cfg != nil && s.cfg.Gateway.StreamDataIntervalTimeout > 0 {
return time.Duration(s.cfg.Gateway.StreamDataIntervalTimeout) * time.Second
}
return 0
}
@@ -0,0 +1,244 @@
package service
// 国产供应商 Anthropic 协议转换路径的上游读间隔超时回归测试(B3):
// 上游挂住 SSE(不发数据也不断连)时,CC×anthropic / Responses×anthropic
// 的读循环必须按 gateway.stream_data_interval_timeout 结束,而不是永久阻塞。
import (
"bufio"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/Wei-Shaw/sub2api/internal/pkg/apicompat"
"github.com/gin-gonic/gin"
)
func newNativeAnthropicHangTestService(intervalSec int) *OpenAIGatewayService {
return &OpenAIGatewayService{
cfg: &config.Config{
Gateway: config.GatewayConfig{
StreamDataIntervalTimeout: intervalSec,
MaxLineSize: defaultMaxLineSize,
},
},
}
}
func newHangingUpstreamResponse() (*http.Response, *io.PipeReader, *io.PipeWriter) {
pr, pw := io.Pipe()
return &http.Response{StatusCode: http.StatusOK, Body: pr, Header: http.Header{}}, pr, pw
}
// miniAnthropicSSEStream 是一段最小可转换的 Anthropic 事件流。
func miniAnthropicSSEStream() string {
return strings.Join([]string{
"event: message_start",
`data: {"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"glm-4.7","usage":{"input_tokens":10,"output_tokens":1}}}`,
"",
"event: content_block_start",
`data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`,
"",
"event: content_block_delta",
`data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}`,
"",
"event: content_block_stop",
`data: {"type":"content_block_stop","index":0}`,
"",
"event: message_delta",
`data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}`,
"",
"event: message_stop",
`data: {"type":"message_stop"}`,
"",
"",
}, "\n")
}
func TestAnthropicNativeLinePump_TimesOutWithoutData(t *testing.T) {
pr, _ := io.Pipe()
scanner := bufio.NewScanner(pr)
defer func() { _ = pr.Close() }()
pump := newAnthropicNativeLinePump(scanner, 50*time.Millisecond)
defer pump.stop()
start := time.Now()
_, err := pump.next()
if err == nil || !strings.Contains(err.Error(), "stream data interval timeout") {
t.Fatalf("expected interval timeout, got %v", err)
}
if elapsed := time.Since(start); elapsed > 5*time.Second {
t.Fatalf("timeout not respected: %v", elapsed)
}
}
func TestAnthropicNativeLinePump_DataResetsTimer(t *testing.T) {
pr, pw := io.Pipe()
scanner := bufio.NewScanner(pr)
pump := newAnthropicNativeLinePump(scanner, 1*time.Second)
defer pump.stop()
go func() {
_, _ = pw.Write([]byte("event: ping\n"))
// 保持流打开且不再发数据:第二次 next 必须超时。
time.Sleep(3 * time.Second)
_ = pw.Close()
}()
defer func() { _ = pr.Close() }()
line, err := pump.next()
if err != nil || line != "event: ping" {
t.Fatalf("expected first line, got %q err=%v", line, err)
}
start := time.Now()
_, err = pump.next()
if err == nil || !strings.Contains(err.Error(), "stream data interval timeout") {
t.Fatalf("expected interval timeout after data stops, got %v", err)
}
if elapsed := time.Since(start); elapsed > 5*time.Second {
t.Fatalf("timeout not respected: %v", elapsed)
}
}
func TestCCStreamingFromNativeAnthropic_HangTimesOut(t *testing.T) {
gin.SetMode(gin.TestMode)
svc := newNativeAnthropicHangTestService(1)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/", nil)
resp, pr, pw := newHangingUpstreamResponse()
start := time.Now()
res, err := svc.handleCCStreamingFromNativeAnthropic(resp, c, "glm-4.7", "glm-4.7", "glm-4.7", nil, start, true)
_ = pw.Close()
_ = pr.Close()
if err == nil || !strings.Contains(err.Error(), "stream data interval timeout") {
t.Fatalf("expected stream timeout error, got %v", err)
}
if res == nil {
t.Fatalf("expected result carrying accumulated usage")
}
if elapsed := time.Since(start); elapsed > 5*time.Second {
t.Fatalf("handler did not respect interval bound: %v", elapsed)
}
}
func TestCCBufferedFromNativeAnthropic_HangTimesOut(t *testing.T) {
gin.SetMode(gin.TestMode)
svc := newNativeAnthropicHangTestService(1)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/", nil)
resp, pr, pw := newHangingUpstreamResponse()
start := time.Now()
_, err := svc.handleCCBufferedFromNativeAnthropic(resp, c, "glm-4.7", "glm-4.7", "glm-4.7", nil, start)
_ = pw.Close()
_ = pr.Close()
if err == nil || !strings.Contains(err.Error(), "stream data interval timeout") {
t.Fatalf("expected stream timeout error, got %v", err)
}
if elapsed := time.Since(start); elapsed > 5*time.Second {
t.Fatalf("handler did not respect interval bound: %v", elapsed)
}
if !strings.Contains(rec.Body.String(), "Upstream stream data interval timeout") {
t.Fatalf("expected 502 error body, got %q", rec.Body.String())
}
}
func TestResponsesStreamingFromNativeAnthropic_HangTimesOut(t *testing.T) {
gin.SetMode(gin.TestMode)
svc := newNativeAnthropicHangTestService(1)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/", nil)
resp, pr, pw := newHangingUpstreamResponse()
start := time.Now()
res, err := svc.handleResponsesStreamingFromNativeAnthropic(resp, c, "glm-4.7", "glm-4.7", "glm-4.7", nil, start, apicompat.ResponsesClientToolMapping{})
_ = pw.Close()
_ = pr.Close()
if err == nil || !strings.Contains(err.Error(), "stream data interval timeout") {
t.Fatalf("expected stream timeout error, got %v", err)
}
if res == nil {
t.Fatalf("expected result carrying accumulated usage")
}
if elapsed := time.Since(start); elapsed > 5*time.Second {
t.Fatalf("handler did not respect interval bound: %v", elapsed)
}
}
func TestCCStreamingFromNativeAnthropic_HappyPathStillConverts(t *testing.T) {
gin.SetMode(gin.TestMode)
svc := newNativeAnthropicHangTestService(5)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/", nil)
resp, pr, pw := newHangingUpstreamResponse()
go func() {
_, _ = pw.Write([]byte(miniAnthropicSSEStream()))
_ = pw.Close()
}()
defer func() { _ = pr.Close() }()
res, err := svc.handleCCStreamingFromNativeAnthropic(resp, c, "glm-4.7", "glm-4.7", "glm-4.7", nil, time.Now(), true)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if res == nil {
t.Fatalf("expected result")
}
body := rec.Body.String()
if !strings.Contains(body, "Hello") {
t.Fatalf("expected converted text chunk, got %q", body)
}
if !strings.Contains(body, "data: [DONE]") {
t.Fatalf("expected [DONE] terminator, got %q", body)
}
}
func TestCCBufferedFromNativeAnthropic_HappyPathStillConverts(t *testing.T) {
gin.SetMode(gin.TestMode)
svc := newNativeAnthropicHangTestService(5)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/", nil)
resp, pr, pw := newHangingUpstreamResponse()
go func() {
_, _ = pw.Write([]byte(miniAnthropicSSEStream()))
_ = pw.Close()
}()
defer func() { _ = pr.Close() }()
res, err := svc.handleCCBufferedFromNativeAnthropic(resp, c, "glm-4.7", "glm-4.7", "glm-4.7", nil, time.Now())
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if res == nil {
t.Fatalf("expected result")
}
body := rec.Body.String()
if !strings.Contains(body, "Hello") {
t.Fatalf("expected converted text in buffered response, got %q", body)
}
if res.Usage.InputTokens != 10 || res.Usage.OutputTokens != 5 {
t.Fatalf("expected usage 10/5, got %+v", res.Usage)
}
}
@@ -14,6 +14,7 @@ import (
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"strings"
"time"
@@ -163,18 +164,53 @@ func (s *OpenAIGatewayService) handleCCBufferedFromNativeAnthropic(
var finalResp *apicompat.AnthropicResponse
var usage ClaudeUsage
for scanner.Scan() {
line := scanner.Text()
// 读间隔上限:上游挂住 SSE 时中止组装(缓冲路径尚未提交响应头,可回 502)。
streamInterval := s.anthropicNativeStreamInterval()
pump := newAnthropicNativeLinePump(scanner, streamInterval)
defer pump.stop()
logReadErr := func(err error) {
if !errors.Is(err, io.EOF) && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
logger.L().Warn("openai cc via native anthropic buffered: read error",
zap.Error(err),
zap.String("request_id", requestID),
)
}
}
onIdle := func() (*OpenAIForwardResult, error) {
_ = resp.Body.Close()
logger.L().Warn("openai cc via native anthropic buffered: data interval timeout",
zap.String("request_id", requestID),
zap.Duration("interval", streamInterval),
)
writeChatCompletionsError(c, http.StatusBadGateway, "server_error", "Upstream stream data interval timeout")
return nil, fmt.Errorf("stream data interval timeout")
}
for {
line, rerr := pump.next()
if rerr != nil {
if errors.Is(rerr, errAnthropicNativeStreamIdle) {
return onIdle()
}
logReadErr(rerr)
break
}
// SSE 规范允许 `event:xxx`(冒号后无空格):Kimi 等 Anthropic 兼容上游
// 返回紧凑格式,严格匹配 "event: " 会丢弃全部事件(#4653 同根因)。
if _, ok := extractOpenAISSEEventLine(line); !ok {
continue
}
if !scanner.Scan() {
dataLine, rerr := pump.next()
if rerr != nil {
if errors.Is(rerr, errAnthropicNativeStreamIdle) {
return onIdle()
}
logReadErr(rerr)
break
}
payload, ok := extractOpenAISSEDataLine(scanner.Text())
payload, ok := extractOpenAISSEDataLine(dataLine)
if !ok {
continue
}
@@ -214,15 +250,6 @@ func (s *OpenAIGatewayService) handleCCBufferedFromNativeAnthropic(
}
}
if err := scanner.Err(); err != nil {
if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
logger.L().Warn("openai cc via native anthropic buffered: read error",
zap.Error(err),
zap.String("request_id", requestID),
)
}
}
if finalResp == nil {
writeChatCompletionsError(c, http.StatusBadGateway, "server_error", "Upstream stream ended without a response")
return nil, fmt.Errorf("upstream stream ended without response")
@@ -322,6 +349,34 @@ func (s *OpenAIGatewayService) handleCCStreamingFromNativeAnthropic(
}
}
// 读间隔上限:上游挂住 SSE(不发数据也不断连)时结束排水。上游 ctx 为
// WithoutCancel 且 http.Client 无整体 Timeout,无此界限则客户端断开后
// scanner.Scan() 永久阻塞(见 anthropic native pump 文件注释)。
streamInterval := s.anthropicNativeStreamInterval()
pump := newAnthropicNativeLinePump(scanner, streamInterval)
defer pump.stop()
logReadErr := func(err error) {
if !errors.Is(err, io.EOF) && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
logger.L().Warn("openai cc via native anthropic stream: read error",
zap.Error(err),
zap.String("request_id", requestID),
)
}
}
// onIdle 关闭上游连接(解除阻塞的读、归还连接池位),并按已累计 usage
// 返回——与 messages 主路径 "stream usage incomplete after timeout" 同语义。
onIdle := func() (*OpenAIForwardResult, error) {
_ = resp.Body.Close()
if !clientDisconnected {
logger.L().Warn("openai cc via native anthropic stream: data interval timeout",
zap.String("request_id", requestID),
zap.Duration("interval", streamInterval),
)
}
return resultWithUsage(), fmt.Errorf("stream data interval timeout")
}
writeChunk := func(chunk apicompat.ChatCompletionsChunk) bool {
if clientDisconnected {
return false // 已断开:不再写客户端,只排水上游累计 usage
@@ -372,16 +427,29 @@ func (s *OpenAIGatewayService) handleCCStreamingFromNativeAnthropic(
return false
}
for scanner.Scan() {
line := scanner.Text()
for {
line, rerr := pump.next()
if rerr != nil {
if errors.Is(rerr, errAnthropicNativeStreamIdle) {
return onIdle()
}
logReadErr(rerr)
break
}
if _, ok := extractOpenAISSEEventLine(line); !ok {
continue
}
if !scanner.Scan() {
dataLine, rerr := pump.next()
if rerr != nil {
if errors.Is(rerr, errAnthropicNativeStreamIdle) {
return onIdle()
}
// EOF / 读错误:事件行后流终止,进入 finalize。
logReadErr(rerr)
break
}
payload, ok := extractOpenAISSEDataLine(scanner.Text())
payload, ok := extractOpenAISSEDataLine(dataLine)
if !ok {
continue
}
@@ -396,15 +464,6 @@ func (s *OpenAIGatewayService) handleCCStreamingFromNativeAnthropic(
}
}
if err := scanner.Err(); err != nil {
if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
logger.L().Warn("openai cc via native anthropic stream: read error",
zap.Error(err),
zap.String("request_id", requestID),
)
}
}
// Finalize both state machines(客户端已断开时仍执行,保证 usage 汇总完整)。
finalResEvents := apicompat.FinalizeAnthropicResponsesStream(anthState)
for _, resEvt := range finalResEvents {
@@ -14,6 +14,7 @@ import (
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"strings"
"time"
@@ -166,17 +167,52 @@ func (s *OpenAIGatewayService) handleResponsesBufferedFromNativeAnthropic(
var finalResp *apicompat.AnthropicResponse
var usage ClaudeUsage
for scanner.Scan() {
line := scanner.Text()
// 读间隔上限:上游挂住 SSE 时中止组装(缓冲路径尚未提交响应头,可回 502)。
streamInterval := s.anthropicNativeStreamInterval()
pump := newAnthropicNativeLinePump(scanner, streamInterval)
defer pump.stop()
logReadErr := func(err error) {
if !errors.Is(err, io.EOF) && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
logger.L().Warn("openai responses via native anthropic buffered: read error",
zap.Error(err),
zap.String("request_id", requestID),
)
}
}
onIdle := func() (*OpenAIForwardResult, error) {
_ = resp.Body.Close()
logger.L().Warn("openai responses via native anthropic buffered: data interval timeout",
zap.String("request_id", requestID),
zap.Duration("interval", streamInterval),
)
writeResponsesError(c, http.StatusBadGateway, "server_error", "Upstream stream data interval timeout")
return nil, fmt.Errorf("stream data interval timeout")
}
for {
line, rerr := pump.next()
if rerr != nil {
if errors.Is(rerr, errAnthropicNativeStreamIdle) {
return onIdle()
}
logReadErr(rerr)
break
}
// SSE 规范允许 `event:xxx`(冒号后无空格):Kimi 等上游返回紧凑格式。
if _, ok := extractOpenAISSEEventLine(line); !ok {
continue
}
if !scanner.Scan() {
dataLine, rerr := pump.next()
if rerr != nil {
if errors.Is(rerr, errAnthropicNativeStreamIdle) {
return onIdle()
}
logReadErr(rerr)
break
}
payload, ok := extractOpenAISSEDataLine(scanner.Text())
payload, ok := extractOpenAISSEDataLine(dataLine)
if !ok {
continue
}
@@ -216,15 +252,6 @@ func (s *OpenAIGatewayService) handleResponsesBufferedFromNativeAnthropic(
}
}
if err := scanner.Err(); err != nil {
if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
logger.L().Warn("openai responses via native anthropic buffered: read error",
zap.Error(err),
zap.String("request_id", requestID),
)
}
}
if finalResp == nil {
writeResponsesError(c, http.StatusBadGateway, "server_error", "Upstream stream ended without a response")
return nil, fmt.Errorf("upstream stream ended without response")
@@ -326,6 +353,30 @@ func (s *OpenAIGatewayService) handleResponsesStreamingFromNativeAnthropic(
}
}
// 读间隔上限:上游挂住 SSE(不发数据也不断连)时结束转换循环。上游 ctx 为
// WithoutCancel 且 http.Client 无整体 Timeout,无此界限则 scanner.Scan()
// 永久阻塞(见 anthropic native pump 文件注释)。
streamInterval := s.anthropicNativeStreamInterval()
pump := newAnthropicNativeLinePump(scanner, streamInterval)
defer pump.stop()
logReadErr := func(err error) {
if !errors.Is(err, io.EOF) && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
logger.L().Warn("openai responses via native anthropic stream: read error",
zap.Error(err),
zap.String("request_id", requestID),
)
}
}
onIdle := func() (*OpenAIForwardResult, error) {
_ = resp.Body.Close()
logger.L().Warn("openai responses via native anthropic stream: data interval timeout",
zap.String("request_id", requestID),
zap.Duration("interval", streamInterval),
)
return resultWithUsage(), fmt.Errorf("stream data interval timeout")
}
processAnthropicEvent := func(event *apicompat.AnthropicStreamEvent) bool {
if firstChunk {
firstChunk = false
@@ -365,16 +416,28 @@ func (s *OpenAIGatewayService) handleResponsesStreamingFromNativeAnthropic(
return false
}
for scanner.Scan() {
line := scanner.Text()
for {
line, rerr := pump.next()
if rerr != nil {
if errors.Is(rerr, errAnthropicNativeStreamIdle) {
return onIdle()
}
logReadErr(rerr)
break
}
if _, ok := extractOpenAISSEEventLine(line); !ok {
continue
}
if !scanner.Scan() {
dataLine, rerr := pump.next()
if rerr != nil {
if errors.Is(rerr, errAnthropicNativeStreamIdle) {
return onIdle()
}
logReadErr(rerr)
break
}
payload, ok := extractOpenAISSEDataLine(scanner.Text())
payload, ok := extractOpenAISSEDataLine(dataLine)
if !ok {
continue
}
@@ -389,15 +452,6 @@ func (s *OpenAIGatewayService) handleResponsesStreamingFromNativeAnthropic(
}
}
if err := scanner.Err(); err != nil {
if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
logger.L().Warn("openai responses via native anthropic stream: read error",
zap.Error(err),
zap.String("request_id", requestID),
)
}
}
// Finalize state machine(客户端已断开时仍执行,保证 usage 汇总完整)。
if finalEvents := apicompat.FinalizeAnthropicResponsesStream(state); len(finalEvents) > 0 {
for _, evt := range finalEvents {
@@ -0,0 +1,17 @@
-- 把 kimi/zhipu/deepseek 平台加入 user_platform_quotas.platform 的 CHECK 约束。
--
-- 背景:国产供应商进入 AllowedQuotaPlatforms(internal/service/domain_constants.go),
-- 注册时 GetDefaultPlatformQuotas 会为全部 8 平台预填充默认配额行,但 157 号迁移的
-- CHECK 仍只允许 5 平台。BulkInsertInitial 是单条多行 INSERT,任一违约行会中止整条
-- 语句 → 注册路径 fail-open 吞错 → 新用户拿到零条配额记录(含原有 5 平台,缺失配额
-- 行 = 无限额)。与 157 头注释记载的 grok 同型事故一致。
--
-- 修复:把约束与代码平台列表(PlatformKimi/PlatformZhipu/PlatformDeepseek)对齐。
-- DROP ... IF EXISTS 保证可重入;新约束是旧约束的超集,存量行(仅 5 平台)瞬时校验通过。
ALTER TABLE user_platform_quotas
DROP CONSTRAINT IF EXISTS user_platform_quotas_platform_check;
ALTER TABLE user_platform_quotas
ADD CONSTRAINT user_platform_quotas_platform_check
CHECK (platform IN ('anthropic', 'openai', 'gemini', 'antigravity', 'grok',
'kimi', 'zhipu', 'deepseek'));
@@ -0,0 +1,22 @@
package migrations
import (
"strings"
"testing"
"github.com/stretchr/testify/require"
)
// TestUserPlatformQuotasCNProvidersMigration 校验 224 号迁移把 kimi/zhipu/deepseek
// 加入 user_platform_quotas.platform 的 CHECK 约束(对照 157 号 grok 迁移)。
// 约束未放宽时,注册预填充 8 平台默认配额会整条 INSERT 中止 → 新用户零配额行
//(缺失配额行 = 无限额),管理端设置国产平台配额直接 500。
func TestUserPlatformQuotasCNProvidersMigration(t *testing.T) {
content, err := FS.ReadFile("224_user_platform_quotas_add_cn_providers.sql")
require.NoError(t, err)
sql := strings.Join(strings.Fields(string(content)), " ")
require.Contains(t, sql, "DROP CONSTRAINT IF EXISTS user_platform_quotas_platform_check")
require.Contains(t, sql,
"CHECK (platform IN ('anthropic', 'openai', 'gemini', 'antigravity', 'grok', 'kimi', 'zhipu', 'deepseek'))")
}
-302
View File
@@ -1,302 +0,0 @@
# =============================================================================
# Sub2API Docker Compose Configuration
# =============================================================================
# Quick Start:
# 1. Copy .env.example to .env and configure
# 2. docker-compose up -d
# 3. Check logs: docker-compose logs -f sub2api
# 4. Access: http://localhost:8080
#
# All configuration is done via environment variables.
# No Setup Wizard needed - the system auto-initializes on first run.
# =============================================================================
services:
# ===========================================================================
# Sub2API Application
# ===========================================================================
sub2api:
image: randark/sub2api:latest
container_name: sub2api
restart: unless-stopped
security_opt:
- no-new-privileges:true
ulimits:
nofile:
soft: 100000
hard: 100000
ports:
- "${BIND_HOST:-0.0.0.0}:${SERVER_PORT:-8080}:8080"
volumes:
# Data persistence (config.yaml will be auto-generated here)
- sub2api_data:/app/data
# Optional: Mount custom config.yaml (uncomment and create the file first)
# Copy config.example.yaml to config.yaml, modify it, then uncomment:
# - ./config.yaml:/app/data/config.yaml
# Optional: Mount a custom Codex instructions template file, then point
# gateway.forced_codex_instructions_template_file at /app/data/codex-instructions.md.tmpl
# in config.yaml.
# - ./codex-instructions.md.tmpl:/app/data/codex-instructions.md.tmpl:ro
environment:
# =======================================================================
# Auto Setup (REQUIRED for Docker deployment)
# =======================================================================
- AUTO_SETUP=true
# =======================================================================
# Server Configuration
# =======================================================================
- SERVER_HOST=0.0.0.0
- SERVER_PORT=8080
- SERVER_MODE=${SERVER_MODE:-release}
- ENABLE_SERVER_TIMING=${ENABLE_SERVER_TIMING:-false}
- RUN_MODE=${RUN_MODE:-standard}
- UPDATE_GITHUB_TOKEN=${UPDATE_GITHUB_TOKEN:-}
- ALIPAY_MOBILE_PRECREATE_DEEP_LINK=${ALIPAY_MOBILE_PRECREATE_DEEP_LINK:-}
# =======================================================================
# Database Configuration (PostgreSQL)
# =======================================================================
- DATABASE_HOST=postgres
- DATABASE_PORT=5432
- DATABASE_USER=${POSTGRES_USER:-sub2api}
- DATABASE_PASSWORD=${POSTGRES_PASSWORD:?POSTGRES_PASSWORD is required}
- DATABASE_DBNAME=${POSTGRES_DB:-sub2api}
- DATABASE_SSLMODE=disable
- DATABASE_MAX_OPEN_CONNS=${DATABASE_MAX_OPEN_CONNS:-50}
- DATABASE_MAX_IDLE_CONNS=${DATABASE_MAX_IDLE_CONNS:-10}
- DATABASE_CONN_MAX_LIFETIME_MINUTES=${DATABASE_CONN_MAX_LIFETIME_MINUTES:-30}
- DATABASE_CONN_MAX_IDLE_TIME_MINUTES=${DATABASE_CONN_MAX_IDLE_TIME_MINUTES:-5}
# =======================================================================
# Redis Configuration
# =======================================================================
- REDIS_HOST=redis
- REDIS_PORT=6379
- REDIS_USERNAME=${REDIS_USERNAME:-}
- REDIS_PASSWORD=${REDIS_PASSWORD:-}
- REDIS_DB=${REDIS_DB:-0}
- REDIS_POOL_SIZE=${REDIS_POOL_SIZE:-1024}
- REDIS_MIN_IDLE_CONNS=${REDIS_MIN_IDLE_CONNS:-10}
- REDIS_ENABLE_TLS=${REDIS_ENABLE_TLS:-false}
# =======================================================================
# Admin Account (auto-created on first run)
# =======================================================================
- ADMIN_EMAIL=${ADMIN_EMAIL:-admin@sub2api.local}
- ADMIN_PASSWORD=${ADMIN_PASSWORD:-}
# =======================================================================
# JWT Configuration
# =======================================================================
# IMPORTANT: Set a fixed JWT_SECRET to prevent login sessions from being
# invalidated after container restarts. If left empty, a random secret
# will be generated on each startup.
# Generate a secure secret: openssl rand -hex 32
- JWT_SECRET=${JWT_SECRET:-}
- JWT_EXPIRE_HOUR=${JWT_EXPIRE_HOUR:-24}
# =======================================================================
# Setup Configuration
# =======================================================================
- SETUP_MIGRATION_TIMEOUT_SECONDS=${SETUP_MIGRATION_TIMEOUT_SECONDS:-0}
# =======================================================================
# TOTP (2FA) Configuration
# =======================================================================
# IMPORTANT: Set a fixed encryption key for TOTP secrets. If left empty,
# a random key will be generated on each startup, causing all existing
# TOTP configurations to become invalid (users won't be able to login
# with 2FA).
# Generate a secure key: openssl rand -hex 32
- TOTP_ENCRYPTION_KEY=${TOTP_ENCRYPTION_KEY:-}
# =======================================================================
# Timezone Configuration
# This affects ALL time operations in the application:
# - Database timestamps
# - Usage statistics "today" boundary
# - Subscription expiry times
# - Log timestamps
# Common values: Asia/Shanghai, America/New_York, Europe/London, UTC
# =======================================================================
- TZ=${TZ:-Asia/Shanghai}
# =======================================================================
# Gemini OAuth Configuration (for Gemini accounts)
# =======================================================================
- GEMINI_OAUTH_CLIENT_ID=${GEMINI_OAUTH_CLIENT_ID:-}
- GEMINI_OAUTH_CLIENT_SECRET=${GEMINI_OAUTH_CLIENT_SECRET:-}
- GEMINI_OAUTH_SCOPES=${GEMINI_OAUTH_SCOPES:-}
- GEMINI_QUOTA_POLICY=${GEMINI_QUOTA_POLICY:-}
# Built-in OAuth client secrets (optional)
# SECURITY: This repo does not embed third-party client_secret.
- GEMINI_CLI_OAUTH_CLIENT_SECRET=${GEMINI_CLI_OAUTH_CLIENT_SECRET:-}
- ANTIGRAVITY_OAUTH_CLIENT_SECRET=${ANTIGRAVITY_OAUTH_CLIENT_SECRET:-}
- ANTIGRAVITY_USER_AGENT_VERSION=${ANTIGRAVITY_USER_AGENT_VERSION:-}
# =======================================================================
# Security Configuration (URL Allowlist)
# =======================================================================
# Enable URL allowlist validation (false to skip allowlist checks)
- SECURITY_URL_ALLOWLIST_ENABLED=${SECURITY_URL_ALLOWLIST_ENABLED:-false}
# Allow insecure HTTP URLs when allowlist is disabled (default: true; set to false to require https)
- SECURITY_URL_ALLOWLIST_ALLOW_INSECURE_HTTP=${SECURITY_URL_ALLOWLIST_ALLOW_INSECURE_HTTP:-true}
# Allow private IP addresses for upstream/pricing/CRS (default: true; set to false to block private hosts)
- SECURITY_URL_ALLOWLIST_ALLOW_PRIVATE_HOSTS=${SECURITY_URL_ALLOWLIST_ALLOW_PRIVATE_HOSTS:-true}
# Upstream hosts whitelist (comma-separated, only used when enabled=true)
- SECURITY_URL_ALLOWLIST_UPSTREAM_HOSTS=${SECURITY_URL_ALLOWLIST_UPSTREAM_HOSTS:-}
# =======================================================================
# Update Configuration (在线更新配置)
# =======================================================================
# Proxy for accessing GitHub (online updates + pricing data)
# Examples: http://host:port, socks5://host:port
- UPDATE_PROXY_URL=${UPDATE_PROXY_URL:-}
# =======================================================================
# Image Generation Stream & Concurrency
# =======================================================================
# OpenAI HTTP upstream protocol/timeout
- GATEWAY_OPENAI_RESPONSE_HEADER_TIMEOUT=${GATEWAY_OPENAI_RESPONSE_HEADER_TIMEOUT:-0}
- GATEWAY_OPENAI_HTTP2_ENABLED=${GATEWAY_OPENAI_HTTP2_ENABLED:-true}
- GATEWAY_OPENAI_HTTP2_ALLOW_PROXY_FALLBACK_TO_HTTP1=${GATEWAY_OPENAI_HTTP2_ALLOW_PROXY_FALLBACK_TO_HTTP1:-true}
- GATEWAY_OPENAI_HTTP2_FALLBACK_ERROR_THRESHOLD=${GATEWAY_OPENAI_HTTP2_FALLBACK_ERROR_THRESHOLD:-2}
- GATEWAY_OPENAI_HTTP2_FALLBACK_WINDOW_SECONDS=${GATEWAY_OPENAI_HTTP2_FALLBACK_WINDOW_SECONDS:-60}
- GATEWAY_OPENAI_HTTP2_FALLBACK_TTL_SECONDS=${GATEWAY_OPENAI_HTTP2_FALLBACK_TTL_SECONDS:-600}
- GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_FAILURE_THRESHOLD:-2}
- GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_WINDOW_SECONDS:-60}
- GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS=${GATEWAY_OPENAI_PROXY_STREAM_CIRCUIT_TTL_SECONDS:-600}
- GATEWAY_IMAGE_STREAM_DATA_INTERVAL_TIMEOUT=${GATEWAY_IMAGE_STREAM_DATA_INTERVAL_TIMEOUT:-900}
- GATEWAY_IMAGE_STREAM_KEEPALIVE_INTERVAL=${GATEWAY_IMAGE_STREAM_KEEPALIVE_INTERVAL:-10}
- GATEWAY_IMAGE_CONCURRENCY_ENABLED=${GATEWAY_IMAGE_CONCURRENCY_ENABLED:-false}
- GATEWAY_IMAGE_CONCURRENCY_MAX_CONCURRENT_REQUESTS=${GATEWAY_IMAGE_CONCURRENCY_MAX_CONCURRENT_REQUESTS:-0}
- GATEWAY_IMAGE_CONCURRENCY_OVERFLOW_MODE=${GATEWAY_IMAGE_CONCURRENCY_OVERFLOW_MODE:-reject}
- GATEWAY_IMAGE_CONCURRENCY_WAIT_TIMEOUT_SECONDS=${GATEWAY_IMAGE_CONCURRENCY_WAIT_TIMEOUT_SECONDS:-30}
- GATEWAY_IMAGE_CONCURRENCY_MAX_WAITING_REQUESTS=${GATEWAY_IMAGE_CONCURRENCY_MAX_WAITING_REQUESTS:-100}
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_healthy
networks:
- sub2api-network
healthcheck:
test:
[
"CMD",
"wget",
"-q",
"-T",
"5",
"-O",
"/dev/null",
"http://localhost:8080/health",
]
interval: 30s
timeout: 10s
retries: 3
start_period: 30s
# ===========================================================================
# PostgreSQL Database
# ===========================================================================
postgres:
image: postgres:18-alpine
container_name: sub2api-postgres
restart: unless-stopped
ulimits:
nofile:
soft: 100000
hard: 100000
# Pass the tuning values documented in .env.example to postgres. The
# fallbacks are the postgres:18 stock defaults, so nothing changes when
# the variables are not set.
command: >
postgres
-c max_connections=${POSTGRES_MAX_CONNECTIONS:-100}
-c shared_buffers=${POSTGRES_SHARED_BUFFERS:-128MB}
-c effective_cache_size=${POSTGRES_EFFECTIVE_CACHE_SIZE:-4GB}
-c maintenance_work_mem=${POSTGRES_MAINTENANCE_WORK_MEM:-64MB}
volumes:
- postgres_data:/var/lib/postgresql/data
environment:
# postgres:18-alpine 默认 PGDATA=/var/lib/postgresql/18/docker(位于镜像声明的匿名卷 /var/lib/postgresql 内)。
# 若不显式设置 PGDATA,则即使挂载了 postgres_data 到 /var/lib/postgresql/data,数据也不会落盘到该命名卷,
# docker compose down/up 后会触发 initdb 重新初始化,导致用户/密码等数据丢失。
- PGDATA=/var/lib/postgresql/data
- POSTGRES_USER=${POSTGRES_USER:-sub2api}
- POSTGRES_PASSWORD=${POSTGRES_PASSWORD:?POSTGRES_PASSWORD is required}
- POSTGRES_DB=${POSTGRES_DB:-sub2api}
- TZ=${TZ:-Asia/Shanghai}
networks:
- sub2api-network
healthcheck:
test:
[
"CMD-SHELL",
"pg_isready -U ${POSTGRES_USER:-sub2api} -d ${POSTGRES_DB:-sub2api}",
]
interval: 10s
timeout: 5s
retries: 5
start_period: 10s
# 注意:不暴露端口到宿主机,应用通过内部网络连接
# 如需调试,可临时添加:ports: ["127.0.0.1:5433:5432"]
# ===========================================================================
# Redis Cache
# ===========================================================================
redis:
image: redis:8-alpine
container_name: sub2api-redis
restart: unless-stopped
ulimits:
nofile:
soft: 100000
hard: 100000
volumes:
- redis_data:/data
# The command is one quoted script for the inner `sh -c`. Compose keeps
# the newlines inside the quoted string, so every line needs a trailing
# `\` — without it, `redis-server` on the first line runs with no flags
# at all, and the --save/--appendonly/--appendfsync lines are never read.
command: >
sh -c '
redis-server \
--save 60 1 \
--appendonly yes \
--appendfsync everysec \
${REDIS_PASSWORD:+--requirepass "$REDIS_PASSWORD"}'
environment:
- TZ=${TZ:-Asia/Shanghai}
# REDISCLI_AUTH is used by redis-cli for authentication (safer than -a flag)
- REDISCLI_AUTH=${REDIS_PASSWORD:-}
networks:
- sub2api-network
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 10s
timeout: 5s
retries: 5
start_period: 5s
# 注意:不暴露端口到宿主机,应用通过内部网络连接
# 如需调试,可临时添加:ports: ["127.0.0.1:6379:6379"]
# =============================================================================
# Volumes
# =============================================================================
volumes:
sub2api_data:
driver: local
postgres_data:
driver: local
redis_data:
driver: local
# =============================================================================
# Networks
# =============================================================================
networks:
sub2api-network:
driver: bridge