Compare commits

...

1 Commits

Author SHA1 Message Date
unkinben ad6dfbdc5b feat: wire the circuit breaker into the proxy fetch path
ci/woodpecker/pr/test Pipeline failed
ci/woodpecker/pr/pre-commit Pipeline failed
ci/woodpecker/pr/build Pipeline failed
The circuit breaker was fully implemented and tested but never called, so
a failing upstream was hit on every request. Engine.Fetch now short-
circuits when the breaker is open (serving stale if available, else 503),
records a failure on each UpstreamError, and resets on success.

Refs #74
2026-07-02 22:33:38 +10:00
+19
View File
@@ -31,6 +31,7 @@ type Engine struct {
cache *cache.Redis cache *cache.Redis
store *storage.S3 store *storage.S3
cas *storage.CAS cas *storage.CAS
circuit *CircuitBreaker
accessLog chan database.AccessLogEntry accessLog chan database.AccessLogEntry
} }
@@ -40,6 +41,7 @@ func NewEngine(db *database.DB, c *cache.Redis, s *storage.S3) *Engine {
cache: c, cache: c,
store: s, store: s,
cas: storage.NewCAS(s), cas: storage.NewCAS(s),
circuit: NewCircuitBreaker(c),
accessLog: make(chan database.AccessLogEntry, accessLogBufferSize), accessLog: make(chan database.AccessLogEntry, accessLogBufferSize),
} }
go e.runAccessLogWriter() go e.runAccessLogWriter()
@@ -154,10 +156,26 @@ func (e *Engine) Fetch(ctx context.Context, remote models.Remote, path string, p
fwdHeaders = clientHeaders[0] fwdHeaders = clientHeaders[0]
} }
// Short-circuit upstream calls when the remote's breaker is open: serve
// stale from the store if we have it, otherwise fail fast rather than
// hammering a known-bad upstream.
if e.circuit.IsOpen(ctx, remote.Name) {
if stale, serr := e.serveFromStore(ctx, remote, path); serr == nil {
slog.Warn("circuit open, serving stale", "remote", remote.Name, "path", path)
stale.Source = "cache"
e.logAccess(remote.Name, path, true, stale.Size, 0)
return stale, nil
}
return nil, &ProxyError{Status: http.StatusServiceUnavailable, Message: "upstream circuit open"}
}
start := time.Now() start := time.Now()
result, err := e.fetchFromUpstream(ctx, remote, path, prov, class, ttl, fwdHeaders) result, err := e.fetchFromUpstream(ctx, remote, path, prov, class, ttl, fwdHeaders)
upstreamMS := int(time.Since(start).Milliseconds()) upstreamMS := int(time.Since(start).Milliseconds())
if err != nil { if err != nil {
if isNetworkError(err) {
e.circuit.RecordFailure(ctx, remote.Name)
}
if remote.StaleOnError && isNetworkError(err) { if remote.StaleOnError && isNetworkError(err) {
_ = e.cache.SetTTL(ctx, remote.Name, path, ttl) _ = e.cache.SetTTL(ctx, remote.Name, path, ttl)
stale, serr := e.serveFromStore(ctx, remote, path) stale, serr := e.serveFromStore(ctx, remote, path)
@@ -171,6 +189,7 @@ func (e *Engine) Fetch(ctx context.Context, remote models.Remote, path string, p
return nil, err return nil, err
} }
e.circuit.RecordSuccess(ctx, remote.Name)
e.logAccess(remote.Name, path, false, result.Size, upstreamMS) e.logAccess(remote.Name, path, false, result.Size, upstreamMS)
return result, nil return result, nil
} }