diff --git a/README.md b/README.md index 29f9df4..a4f94cb 100644 --- a/README.md +++ b/README.md @@ -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` 直接放弃整条命令。 + ## 示例 ### 完整转码工作流 diff --git a/cmd/server/main.go b/cmd/server/main.go index 549ce83..1d6cc42 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -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" @@ -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) diff --git a/pkg/cli/client/client.go b/pkg/cli/client/client.go index 4afdd3d..eb70dcb 100644 --- a/pkg/cli/client/client.go +++ b/pkg/cli/client/client.go @@ -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 @@ -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 @@ -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. diff --git a/pkg/cli/client/health_probe_test.go b/pkg/cli/client/health_probe_test.go new file mode 100644 index 0000000..9623228 --- /dev/null +++ b/pkg/cli/client/health_probe_test.go @@ -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) + } +} diff --git a/pkg/server/accesslog/accesslog.go b/pkg/server/accesslog/accesslog.go new file mode 100644 index 0000000..5146bb1 --- /dev/null +++ b/pkg/server/accesslog/accesslog.go @@ -0,0 +1,164 @@ +// Package accesslog writes the server's HTTP access log without coupling +// response latency to the log destination. +// +// chi's Logger writes the access line from inside the handler chain, before the +// response is flushed: a blocked write there (full pipe, filesystem under +// memory or I/O pressure) delays every in-flight response it logs, health +// probes included. Sink queues lines and drops them once the destination +// stalls, so no request ever waits for the log. +package accesslog + +import ( + "fmt" + "io" + "log" + "net/http" + "runtime" + "sync/atomic" + "time" + + "github.com/go-chi/chi/v5/middleware" +) + +// DefaultQueueDepth is how many access log lines a Sink buffers before it +// starts dropping. It is deep enough to absorb a burst (every concurrent +// request logs one line) while staying bounded, so a permanently stalled +// destination cannot grow memory without limit. +const DefaultQueueDepth = 4096 + +// closeDrainTimeout bounds how long Close waits for queued lines to reach a +// stalled destination. A var so tests can shorten it. +var closeDrainTimeout = 2 * time.Second + +// Sink is an access log destination that never blocks its caller: lines are +// queued and flushed by a background goroutine, and lines the destination loses +// — dropped from a full queue, or rejected by a failing write — are counted and +// reported. +type Sink struct { + out io.Writer + lines chan []byte + stop chan struct{} + writerDone chan struct{} + stopping atomic.Bool + dropped atomic.Uint64 + writeFailures atomic.Uint64 +} + +// NewSink starts a Sink writing to out, buffering up to queueDepth lines +// (DefaultQueueDepth when queueDepth <= 0). +func NewSink(out io.Writer, queueDepth int) *Sink { + if queueDepth <= 0 { + queueDepth = DefaultQueueDepth + } + s := &Sink{ + out: out, + lines: make(chan []byte, queueDepth), + stop: make(chan struct{}), + writerDone: make(chan struct{}), + } + go s.drain() + return s +} + +// Write queues p and returns immediately — it never blocks and never fails, so +// the response it logs is never gated by the destination. Lines arriving once +// the queue is full are dropped and counted. +func (s *Sink) Write(p []byte) (int, error) { + if !s.stopping.Load() { + line := make([]byte, len(p)) + copy(line, p) + select { + case s.lines <- line: + default: + s.dropped.Add(1) + } + } + return len(p), nil +} + +// Dropped reports how many lines have been discarded because the queue was +// full. +func (s *Sink) Dropped() uint64 { + return s.dropped.Load() +} + +// Close stops accepting lines, flushes what is still queued and returns. A +// destination that is still stalled must not hold shutdown hostage, so Close +// gives up after closeDrainTimeout — leaving the drain goroutine parked in its +// write until the process exits, which is the price of never blocking a caller +// on the destination. +func (s *Sink) Close() { + if s.stopping.Swap(true) { + return + } + close(s.stop) + select { + case <-s.writerDone: + case <-time.After(closeDrainTimeout): + } +} + +// drain flushes queued lines in arrival order until Close, then flushes the +// remainder. +func (s *Sink) drain() { + defer close(s.writerDone) + for { + select { + case line := <-s.lines: + s.emit(line) + case <-s.stop: + for { + select { + case line := <-s.lines: + s.emit(line) + default: + s.reportLosses() + return + } + } + } + } +} + +// emit writes one line, preceded by a notice when earlier lines were lost: the +// notice travels the same path the lost lines would have, so it appears exactly +// when the destination starts accepting writes again. +func (s *Sink) emit(line []byte) { + s.reportLosses() + if _, err := s.out.Write(line); err != nil { + // A failing destination (stdout on a full disk) loses the line the same + // way a full queue does; counting it keeps the loss visible instead of + // silently shrinking the log. + s.writeFailures.Add(1) + } +} + +// reportLosses writes the accumulated loss notice and clears what the notice +// accounted for. Counters are cleared only after the notice reached the +// destination, so a destination that is still failing keeps the count for a +// later notice instead of swallowing it; a destination that never accepts +// another write cannot report anything, by construction. +func (s *Sink) reportLosses() { + dropped, failures := s.dropped.Load(), s.writeFailures.Load() + if dropped == 0 && failures == 0 { + return + } + notice := fmt.Sprintf("%s [accesslog] lost %d access log line(s) (%d dropped from a full queue, %d write errors)\n", + time.Now().Format("2006/01/02 15:04:05"), dropped+failures, dropped, failures) + if _, err := s.out.Write([]byte(notice)); err != nil { + return + } + s.dropped.CompareAndSwap(dropped, 0) + s.writeFailures.CompareAndSwap(failures, 0) +} + +// Middleware returns the chi access log middleware writing through sink. It is +// middleware.Logger with the destination replaced: the line is queued instead of +// written on the response path. NoColor mirrors chi's own default, so the line +// format stays byte-identical to the middleware it replaces. +func Middleware(sink *Sink) func(http.Handler) http.Handler { + return middleware.RequestLogger(&middleware.DefaultLogFormatter{ + Logger: log.New(sink, "", log.LstdFlags), + NoColor: runtime.GOOS == "windows", + }) +} diff --git a/pkg/server/accesslog/accesslog_test.go b/pkg/server/accesslog/accesslog_test.go new file mode 100644 index 0000000..5d6886f --- /dev/null +++ b/pkg/server/accesslog/accesslog_test.go @@ -0,0 +1,199 @@ +package accesslog + +import ( + "bytes" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// writerFunc adapts a function to io.Writer. +type writerFunc func(p []byte) (int, error) + +func (f writerFunc) Write(p []byte) (int, error) { return f(p) } + +// TestMiddlewareKeepsResponsesOffTheStalledSink pins the contract this package +// exists for: a log destination that stops accepting writes must not delay the +// responses it logs. chi's own middleware.Logger writes the line on the response +// path, so with it the first write into a stalled destination hangs the request +// that produced it and every request behind it — health probes included. +func TestMiddlewareKeepsResponsesOffTheStalledSink(t *testing.T) { + release := make(chan struct{}) + destination := writerFunc(func(p []byte) (int, error) { + <-release + return len(p), nil + }) + + const queueDepth = 4 + const requests = 25 + sink := NewSink(destination, queueDepth) + defer sink.Close() + + srv := httptest.NewServer(Middleware(sink)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }))) + defer srv.Close() + + client := &http.Client{Timeout: 5 * time.Second} + for i := 1; i <= requests; i++ { + start := time.Now() + resp, err := client.Get(srv.URL) + if err != nil { + t.Fatalf("request %d waited on the stalled log destination: %v", i, err) + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if elapsed := time.Since(start); elapsed > time.Second { + t.Fatalf("request %d took %v with a stalled log destination; responses must not wait for it", i, elapsed) + } + } + + if got := sink.Dropped(); got == 0 { + t.Errorf("Dropped() = 0 after %d requests through a %d-line queue into a stalled destination, want > 0", + requests, queueDepth) + } + close(release) +} + +// TestSinkFlushesQueuedLinesAndReportsDroppedOnes pins the lossy-but-bounded +// contract: lines that fit the queue reach the destination in order, the rest +// are dropped, and the drop is reported once the destination accepts writes +// again. +func TestSinkFlushesQueuedLinesAndReportsDroppedOnes(t *testing.T) { + var out bytes.Buffer + entered := make(chan struct{}) + release := make(chan struct{}) + var once sync.Once + destination := writerFunc(func(p []byte) (int, error) { + once.Do(func() { close(entered) }) + <-release + return out.Write(p) + }) + + sink := NewSink(destination, 2) + defer sink.Close() + + sink.Write([]byte("line-0\n")) + <-entered // the drain goroutine is now parked inside the stalled destination + for i := 1; i <= 6; i++ { + sink.Write([]byte(fmt.Sprintf("line-%d\n", i))) + } + + if got := sink.Dropped(); got != 4 { + t.Fatalf("Dropped() = %d, want 4 (6 lines offered into a 2-line queue)", got) + } + close(release) + sink.Close() + + got := out.String() + for _, want := range []string{"line-0\n", "line-1\n", "line-2\n"} { + if !strings.Contains(got, want) { + t.Errorf("queued line %q missing from destination output %q", want, got) + } + } + for _, dropped := range []string{"line-4", "line-5", "line-6"} { + if strings.Contains(got, dropped) { + t.Errorf("dropped line %q reached the destination: %q", dropped, got) + } + } + if !strings.Contains(got, "lost 4 access log line(s)") || !strings.Contains(got, "4 dropped from a full queue") { + t.Errorf("destination output %q does not report the 4 dropped lines", got) + } +} + +// TestSinkCountsFailingWritesAndReportsThem pins the other loss path: a +// destination that accepts the write call but returns an error (stdout +// redirected to a full disk) must be counted and reported, not swallowed. +func TestSinkCountsFailingWritesAndReportsThem(t *testing.T) { + var out bytes.Buffer + failing := atomic.Bool{} + failing.Store(true) + var failed atomic.Int64 + twoFailed := make(chan struct{}) + destination := writerFunc(func(p []byte) (int, error) { + if failing.Load() { + if failed.Add(1) == 2 { + close(twoFailed) + } + return 0, errors.New("destination write failed") + } + return out.Write(p) + }) + + sink := NewSink(destination, 8) + defer sink.Close() + + sink.Write([]byte("line-0\n")) + sink.Write([]byte("line-1\n")) + <-twoFailed // both lines have been handed to the failing destination + + failing.Store(false) + sink.Write([]byte("line-2\n")) + sink.Close() + + got := out.String() + if !strings.Contains(got, "lost 2 access log line(s)") || !strings.Contains(got, "2 write errors") { + t.Errorf("destination output %q does not report the 2 failed writes", got) + } + if !strings.Contains(got, "line-2") { + t.Errorf("destination output %q does not contain the line that followed the failures", got) + } +} + +// TestSinkCloseGivesUpOnStalledDestination pins that shutdown is bounded too: a +// destination that never accepts another write must not hold Close forever. +func TestSinkCloseGivesUpOnStalledDestination(t *testing.T) { + original := closeDrainTimeout + closeDrainTimeout = 50 * time.Millisecond + t.Cleanup(func() { closeDrainTimeout = original }) + + entered := make(chan struct{}) + release := make(chan struct{}) + var once sync.Once + sink := NewSink(writerFunc(func(p []byte) (int, error) { + once.Do(func() { close(entered) }) + <-release + return len(p), nil + }), 1) + + sink.Write([]byte("line\n")) + <-entered + + start := time.Now() + sink.Close() + if elapsed := time.Since(start); elapsed > time.Second { + t.Errorf("Close() took %v with a stalled destination, want it to give up after closeDrainTimeout", elapsed) + } + close(release) +} + +// TestMiddlewareLogsRequests pins that the middleware still produces the access +// line — the response path is decoupled from the destination, not the log. +func TestMiddlewareLogsRequests(t *testing.T) { + var out bytes.Buffer + sink := NewSink(&out, 8) + + srv := httptest.NewServer(Middleware(sink)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusTeapot) + }))) + resp, err := http.Get(srv.URL + "/api/v1/health") + if err != nil { + t.Fatalf("GET failed: %v", err) + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + srv.Close() + sink.Close() + + line := out.String() + if !strings.Contains(line, "GET") || !strings.Contains(line, "/api/v1/health") || !strings.Contains(line, "418") { + t.Errorf("access log line %q does not describe the request", line) + } +}