Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -1110,6 +1110,10 @@ Response:
}
```

**并发下的探活隔离**:该端点只读进程内状态(版本、鉴权是否启用),不查数据库、不碰存储,因此不与业务端点争用连接池——13 个客户端并发上传/提交不会通过共享资源把它挤掉。响应的写出路径同样不耦合日志目的地:access log 写入有界队列、由后台 goroutine 落盘,目的地阻塞(stdout 管道写满、文件系统压力)时丢弃并计数(目的地恢复后输出 `[accesslog] lost N access log line(s)`,分别报告队列满丢弃数与写入失败数),而不是让响应等待日志写完。这一点必须有:chi 默认的 access log 写在响应 flush 之前,一次阻塞的日志写会让**所有**在途响应一起等待,表现为「服务端日志里 200/µs、客户端却等了几十秒」的全端点延迟尖刺。

**CLI 启动探针的重试预算**:CLI 在每次命令开始时探测该端点,使用独立于业务请求的客户端与预算——单次尝试 20s、最多 3 次、退避 1s/2s,最坏 **63s** 后放弃(3×20s + 1s + 2s)。这比它所取代的单次 30s 尝试更长:最坏等待翻倍是有意的取舍——探针存在的意义是熬过短暂卡顿而不是立刻报错,且该最坏值只在每次尝试都挂起时才会付出;连接被拒与非 200 状态仍立即报错(重试不能让不存在的服务端应答,也不改变已给出的否定答复)。因此服务端因宿主机内存/IO 压力短暂卡顿时,后续尝试可以接管,而不是以 `server health check failed: context deadline exceeded` 直接放弃整条命令。

## 示例

### 完整转码工作流
Expand Down
11 changes: 9 additions & 2 deletions cmd/server/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"gopkg.in/natefinch/lumberjack.v2"

"github.com/tsic404/rffmpeg/pkg/config"
"github.com/tsic404/rffmpeg/pkg/server/accesslog"
"github.com/tsic404/rffmpeg/pkg/server/auth"
"github.com/tsic404/rffmpeg/pkg/server/db"
"github.com/tsic404/rffmpeg/pkg/server/handlers"
Expand Down Expand Up @@ -264,9 +265,15 @@ func main() {
jobScheduler.SetRateLimiter(h.GetRateLimiter())
jobScheduler.SetJobNotifier(h.GetWSHub())

// Setup router
// Setup router. Access logging goes through a non-blocking sink: chi's
// Logger writes the access line from inside the handler chain, before the
// response is flushed, so a stalled stdout would stall every response it
// logs — health probes and business endpoints alike (see pkg/server/accesslog).
accessLog := accesslog.NewSink(os.Stdout, accesslog.DefaultQueueDepth)
defer accessLog.Close()

r := chi.NewRouter()
r.Use(middleware.Logger)
r.Use(accesslog.Middleware(accessLog))
r.Use(middleware.Recoverer)
r.Use(middleware.RequestID)
r.Use(middleware.RealIP)
Expand Down
76 changes: 70 additions & 6 deletions pkg/cli/client/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,23 @@ const (
DefaultTimeout = 30 * time.Second
UploadTimeout = 10 * time.Minute
DownloadTimeout = 10 * time.Minute
// HealthProbeTimeout bounds a single health probe attempt. The probe is a
// readiness gate, not a data transfer: a server that accepts the connection
// but cannot answer within this budget is stalled, and a fresh attempt is
// cheaper than waiting longer on one. Kept below DefaultTimeout so the
// probe never inherits the business timeout it does not need.
HealthProbeTimeout = 20 * time.Second
// HealthProbeAttempts bounds the probe's retry loop: 3 attempts of
// HealthProbeTimeout plus 1s/2s backoff wait up to 63s in total before a
// hung server is reported. That is deliberately longer than the single 30s
// attempt it replaces — a server busy enough to miss one probe (a
// concurrent-client burst, a host under memory or I/O pressure) is reached
// by a later one instead of failing the command outright — and the worst
// case is only paid when every attempt hangs; a refused connection still
// fails immediately.
HealthProbeAttempts = 3
// HealthProbeBackoff is the first retry interval; it doubles per attempt.
HealthProbeBackoff = time.Second
// ProbeTimeout bounds a single probe request. The probe endpoint is
// synchronous: it dispatches a job and waits for a worker to run ffprobe,
// which on a cold worker (first ffmpeg/GPU initialization) can exceed the
Expand Down Expand Up @@ -121,6 +138,14 @@ var pollInterval = PollInterval
// delay.
var probeTimeout = ProbeTimeout

// healthProbeTimeout and healthProbeBackoff are the effective health-probe
// budget; vars (rather than const references) so tests can shrink the retry
// loop instead of waiting out real probe timeouts.
var (
healthProbeTimeout = HealthProbeTimeout
healthProbeBackoff = HealthProbeBackoff
)

// Overridable sleep for rate-limit resubmission backoff; production value
// waits d or until ctx is done (whichever comes first). Internal tests stub
// it so the 429 retry loop runs instantly instead of waiting out real
Expand Down Expand Up @@ -1140,26 +1165,65 @@ func (c *Client) DownloadOutput(fileID, outputPath string) error {
return nil
}

// HealthCheck checks server health
// HealthCheck reports whether the server answers a health probe.
//
// The probe runs on its own client and attempt budget instead of the shared
// business client (DefaultTimeout): a readiness gate must not queue behind
// business traffic, and one stalled answer must not abort the command — a
// server busy enough to miss a probe (a concurrent-client burst) can still
// serve the job that follows. Only attempts with no answer at all are retried;
// a refused connection or a non-200 status is reported at once.
func (c *Client) HealthCheck() error {
probe := &http.Client{Timeout: healthProbeTimeout}
var lastErr error
for attempt := 1; attempt <= HealthProbeAttempts; attempt++ {
retryable, err := c.probeHealth(probe)
if err == nil {
return nil
}
if !retryable {
return err
}
lastErr = err
if attempt < HealthProbeAttempts {
time.Sleep(healthProbeBackoff << (attempt - 1))
}
}
return fmt.Errorf("%w (after %d attempts)", lastErr, HealthProbeAttempts)
}

// probeHealth performs one health probe attempt and reports whether a further
// attempt could succeed.
func (c *Client) probeHealth(probe *http.Client) (bool, error) {
req, err := http.NewRequest("GET", c.serverURL+HealthEndpoint, nil)
if err != nil {
return fmt.Errorf("failed to create request: %w", err)
return false, fmt.Errorf("failed to create request: %w", err)
}

c.setAuthHeader(req)

resp, err := c.http.Do(req)
resp, err := probe.Do(req)
if err != nil {
return fmt.Errorf("health check failed: %w", err)
return isProbeTimeout(err), fmt.Errorf("health check failed: %w", err)
}
defer DrainAndClose(resp.Body)

if resp.StatusCode != http.StatusOK {
return fmt.Errorf("server unhealthy: status %d", resp.StatusCode)
return false, fmt.Errorf("server unhealthy: status %d", resp.StatusCode)
}

return nil
return false, nil
}

// isProbeTimeout reports whether err is a probe that got no answer within its
// time budget — the failure a retry can clear. A refused connection is not
// retried: it means nothing is listening, and no retry changes that.
func isProbeTimeout(err error) bool {
if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, os.ErrDeadlineExceeded) {
return true
}
var netErr net.Error
return errors.As(err, &netErr) && netErr.Timeout()
}

// Probe sends a probe request to the server and returns media information.
Expand Down
122 changes: 122 additions & 0 deletions pkg/cli/client/health_probe_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,122 @@
package client

import (
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
)

// shortenHealthProbeBudget shrinks the probe's attempt timeout and backoff so
// retry-path tests run in milliseconds instead of the production ~2 minutes.
func shortenHealthProbeBudget(t *testing.T) {
t.Helper()
timeout, backoff := healthProbeTimeout, healthProbeBackoff
healthProbeTimeout, healthProbeBackoff = 50*time.Millisecond, 10*time.Millisecond
t.Cleanup(func() { healthProbeTimeout, healthProbeBackoff = timeout, backoff })
}

// TestHealthCheckRetriesStalledProbe pins the fix for the concurrent-load
// regression: a server that misses one probe because it is momentarily stalled
// must not fail the command, because a later attempt succeeds.
func TestHealthCheckRetriesStalledProbe(t *testing.T) {
shortenHealthProbeBudget(t)

var attempts atomic.Int64
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if attempts.Add(1) == 1 {
// Longer than the probe's per-attempt timeout: the client must give
// up on this attempt, not on the server.
time.Sleep(10 * healthProbeTimeout)
}
w.WriteHeader(http.StatusOK)
}))
defer srv.Close()

c := New(srv.URL, "test-token")
if err := c.HealthCheck(); err != nil {
t.Fatalf("HealthCheck() = %v, want nil after the stalled first attempt", err)
}
if got := attempts.Load(); got != 2 {
t.Errorf("server saw %d probe(s), want 2 (stalled attempt + successful retry)", got)
}
}

// TestHealthCheckGivesUpAfterRetryBudget pins the other side of the retry loop:
// a server that never answers ends in the probe error, after exactly the
// configured number of attempts.
func TestHealthCheckGivesUpAfterRetryBudget(t *testing.T) {
shortenHealthProbeBudget(t)

var attempts atomic.Int64
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
attempts.Add(1)
time.Sleep(10 * healthProbeTimeout)
}))
defer srv.Close()

c := New(srv.URL, "test-token")
err := c.HealthCheck()
if err == nil {
t.Fatal("HealthCheck() = nil, want a timeout error for a server that never answers")
}
if !strings.Contains(err.Error(), "health check failed") || !strings.Contains(err.Error(), "context deadline exceeded") {
t.Errorf("HealthCheck() = %v, want a health check timeout error", err)
}
if got := attempts.Load(); got != HealthProbeAttempts {
t.Errorf("server saw %d probe(s), want %d (the full attempt budget)", got, HealthProbeAttempts)
}
}

// TestHealthCheckDoesNotRetryRefusedConnection pins that an absent server is
// reported immediately: no retry makes a closed port answer, and the operator
// should not wait out a retry budget to learn the address is wrong. The retried
// path is recognizable by its "(after N attempts)" marker, so no wall-clock
// assertion is needed — those are jitter-prone under -race or a loaded CI host.
func TestHealthCheckDoesNotRetryRefusedConnection(t *testing.T) {
shortenHealthProbeBudget(t)

srv := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}))
unreachable := srv.URL
srv.Close()

c := New(unreachable, "test-token")
err := c.HealthCheck()
if err == nil {
t.Fatal("HealthCheck() = nil, want an error for a refused connection")
}
if !strings.Contains(err.Error(), "connection refused") {
t.Errorf("HealthCheck() = %v, want a connection-refused error", err)
}
if strings.Contains(err.Error(), "(after ") {
t.Errorf("HealthCheck() = %v, want a refused connection reported without a retry", err)
}
}

// TestHealthCheckDoesNotRetryUnhealthyStatus pins that a server which answers
// with a non-200 status is reported as unhealthy at once: it is a definite
// answer, not a transient transport failure.
func TestHealthCheckDoesNotRetryUnhealthyStatus(t *testing.T) {
shortenHealthProbeBudget(t)

var attempts atomic.Int64
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
attempts.Add(1)
w.WriteHeader(http.StatusServiceUnavailable)
}))
defer srv.Close()

c := New(srv.URL, "test-token")
err := c.HealthCheck()
if err == nil {
t.Fatal("HealthCheck() = nil, want an error for an unhealthy status")
}
if !strings.Contains(err.Error(), "status 503") {
t.Errorf("HealthCheck() = %v, want the status reported", err)
}
if got := attempts.Load(); got != 1 {
t.Errorf("server saw %d probe(s), want 1 (no retry for a definite answer)", got)
}
}
Loading
Loading