diff --git a/bundles/httpclient/bundle.go b/bundles/httpclient/bundle.go index ef55d21..0ebf0f2 100644 --- a/bundles/httpclient/bundle.go +++ b/bundles/httpclient/bundle.go @@ -379,7 +379,7 @@ func (b *Bundle) Initialize(app *framework.App) error { httpClient := &http.Client{ Transport: transport, Timeout: b.config.Timeout, - CheckRedirect: redirectHeaderStripper(b.config.APIKeyHeader), + CheckRedirect: redirectHeaderStripper(b.config.APIKeyHeader, b.config.AllowedHosts), } // Create enhanced client wrapper @@ -423,18 +423,27 @@ func (b *Bundle) Close() error { const maxRedirects = 10 // redirectHeaderStripper returns an http.Client CheckRedirect function that -// removes credential headers whenever a redirect crosses to a different host. -// Go's default redirect handling already strips "Authorization" on cross-host -// redirects but forwards custom headers like the configured API-key header -// unchanged, allowing a malicious or compromised upstream to harvest -// credentials via a 302 to an attacker-controlled host. Same-host redirects -// are unaffected and continue to carry auth headers as before. -func redirectHeaderStripper(apiKeyHeader string) func(*http.Request, []*http.Request) error { +// removes credential headers whenever a redirect crosses to a different host, +// and, when allowedHosts is non-empty, rejects redirects to a host outside +// that allowlist. Go's default redirect handling already strips +// "Authorization" on cross-host redirects but forwards custom headers like +// the configured API-key header unchanged, allowing a malicious or +// compromised upstream to harvest credentials via a 302 to an +// attacker-controlled host. Same-host redirects are unaffected and continue +// to carry auth headers as before. The allowlist check closes a related gap: +// checkHostAllowed only validates the initial request URL, so without this, +// a redirect could still be followed to a disallowed host (with credentials +// stripped) even when AllowedHosts is configured. +func redirectHeaderStripper(apiKeyHeader string, allowedHosts []string) func(*http.Request, []*http.Request) error { return func(req *http.Request, via []*http.Request) error { if len(via) >= maxRedirects { return fmt.Errorf("stopped after %d redirects", maxRedirects) } + if !hostAllowed(req.URL.Hostname(), allowedHosts) { + return fmt.Errorf("redirect to %w: %s", ErrHostNotAllowed, req.URL.Hostname()) + } + prev := via[len(via)-1] if prev.URL.Host != req.URL.Host { req.Header.Del("Authorization") @@ -794,6 +803,8 @@ func (c *Client) buildURL(path string) (string, error) { // to any other host are rejected with ErrHostNotAllowed. This covers both the // shared request path (buildURL results, which may be overridden by an // absolute path) and RawRequest, where the caller supplies the URL directly. +// Redirect targets are separately enforced by redirectHeaderStripper via the +// same hostAllowed helper, since this check only sees the initial URL. func (c *Client) checkHostAllowed(rawURL string) error { if len(c.config.AllowedHosts) == 0 { return nil @@ -805,13 +816,26 @@ func (c *Client) checkHostAllowed(rawURL string) error { } host := parsedURL.Hostname() - for _, allowed := range c.config.AllowedHosts { + if !hostAllowed(host, c.config.AllowedHosts) { + return fmt.Errorf("%w: %s", ErrHostNotAllowed, host) + } + + return nil +} + +// hostAllowed reports whether host is permitted under allowedHosts. An empty +// allowedHosts means unrestricted (everything is allowed), matching the +// opt-in nature of the AllowedHosts config field. +func hostAllowed(host string, allowedHosts []string) bool { + if len(allowedHosts) == 0 { + return true + } + for _, allowed := range allowedHosts { if host == allowed { - return nil + return true } } - - return fmt.Errorf("%w: %s", ErrHostNotAllowed, host) + return false } // addAuthHeaders adds authentication headers to the request. diff --git a/bundles/httpclient/bundle_test.go b/bundles/httpclient/bundle_test.go index 173e455..927b121 100644 --- a/bundles/httpclient/bundle_test.go +++ b/bundles/httpclient/bundle_test.go @@ -5,6 +5,7 @@ import ( "errors" "net/http" "net/http/httptest" + "net/url" "strings" "testing" "time" @@ -827,6 +828,130 @@ func TestClient_AllowedHosts_PermitsRequestToAllowedHost(t *testing.T) { } } +// --- AllowedHosts blocks redirects to disallowed hosts (residual SSRF gap) --- + +// localhostRedirectTarget rewrites rawURL (an httptest server URL, which uses +// the literal "127.0.0.1" host) to use the "localhost" hostname instead, +// keeping the same port. "localhost" resolves to the same loopback address, +// so the connection still succeeds, but the request's URL.Hostname() now +// differs textually from a "127.0.0.1"-hosted server, letting tests exercise +// AllowedHosts across two distinct-looking hosts without needing real DNS. +func localhostRedirectTarget(t *testing.T, rawURL, path string) string { + t.Helper() + parsed, err := url.Parse(rawURL) + if err != nil { + t.Fatalf("failed to parse URL %q: %v", rawURL, err) + } + return "http://localhost:" + parsed.Port() + path +} + +func TestClient_AllowedHosts_BlocksRedirectToDisallowedHost(t *testing.T) { + var targetReached bool + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + targetReached = true + w.WriteHeader(http.StatusOK) + })) + defer target.Close() + + redirector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, localhostRedirectTarget(t, target.URL, "/dest"), http.StatusFound) + })) + defer redirector.Close() + + redirectorURL, err := url.Parse(redirector.URL) + if err != nil { + t.Fatalf("failed to parse redirector URL: %v", err) + } + + cfg := DefaultConfig() + cfg.BaseURL = redirector.URL + cfg.RetryConfig.MaxRetries = 0 + cfg.CircuitBreakerConfig.Name = "test-allowedhosts-blocks-redirect" + // Only the initial host is allowed; the redirect target is a different host. + cfg.AllowedHosts = []string{redirectorURL.Hostname()} + + b := NewBundle(cfg) + if err := b.Initialize(nil); err != nil { + t.Fatalf("failed to initialize bundle: %v", err) + } + + err = b.Client().Get(context.Background(), "/start", nil) + if !errors.Is(err, ErrHostNotAllowed) { + t.Fatalf("expected errors.Is(err, ErrHostNotAllowed), got %v", err) + } + if targetReached { + t.Error("expected disallowed redirect target to never be reached") + } +} + +func TestClient_AllowedHosts_PermitsRedirectToAllowedHost(t *testing.T) { + var targetReached bool + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + targetReached = true + w.WriteHeader(http.StatusOK) + })) + defer target.Close() + + redirector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, localhostRedirectTarget(t, target.URL, "/dest"), http.StatusFound) + })) + defer redirector.Close() + + redirectorURL, err := url.Parse(redirector.URL) + if err != nil { + t.Fatalf("failed to parse redirector URL: %v", err) + } + + cfg := DefaultConfig() + cfg.BaseURL = redirector.URL + cfg.RetryConfig.MaxRetries = 0 + cfg.CircuitBreakerConfig.Name = "test-allowedhosts-permits-redirect" + cfg.AllowedHosts = []string{redirectorURL.Hostname(), "localhost"} + + b := NewBundle(cfg) + if err := b.Initialize(nil); err != nil { + t.Fatalf("failed to initialize bundle: %v", err) + } + + if err := b.Client().Get(context.Background(), "/start", nil); err != nil { + t.Fatalf("unexpected error following allowed redirect: %v", err) + } + if !targetReached { + t.Error("expected redirect target to be reached") + } +} + +func TestClient_AllowedHosts_NilPermitsRedirectToAnyHost(t *testing.T) { + var targetReached bool + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + targetReached = true + w.WriteHeader(http.StatusOK) + })) + defer target.Close() + + redirector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, localhostRedirectTarget(t, target.URL, "/dest"), http.StatusFound) + })) + defer redirector.Close() + + cfg := DefaultConfig() + cfg.BaseURL = redirector.URL + cfg.RetryConfig.MaxRetries = 0 + cfg.CircuitBreakerConfig.Name = "test-allowedhosts-nil-redirect" + + b := NewBundle(cfg) + if err := b.Initialize(nil); err != nil { + t.Fatalf("failed to initialize bundle: %v", err) + } + + if err := b.Client().Get(context.Background(), "/start", nil); err != nil { + t.Fatalf("unexpected error following redirect with nil AllowedHosts: %v", err) + } + if !targetReached { + t.Error("expected redirect target to be reached") + } +} + func TestRequest_CircuitBreakerOpen(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusInternalServerError) diff --git a/bundles/postgresql/bundle.go b/bundles/postgresql/bundle.go index c632b88..fd26ff8 100644 --- a/bundles/postgresql/bundle.go +++ b/bundles/postgresql/bundle.go @@ -64,7 +64,7 @@ import ( "fmt" "time" - _ "github.com/lib/pq" + _ "github.com/jackc/pgx/v5/stdlib" "github.com/datariot/forge/errors" "github.com/datariot/forge/framework" @@ -143,7 +143,7 @@ func (b *Bundle) Initialize(app *framework.App) error { } // Open database connection - db, err := sql.Open("postgres", b.config.DatabaseURL) + db, err := sql.Open("pgx", b.config.DatabaseURL) if err != nil { return errors.ErrRepositoryUnavailable.WithMessage("failed to open PostgreSQL connection").WithCause(err) } diff --git a/examples/postgresql-service/go.mod b/examples/postgresql-service/go.mod index 8ee0c2f..5517018 100644 --- a/examples/postgresql-service/go.mod +++ b/examples/postgresql-service/go.mod @@ -15,7 +15,10 @@ require ( github.com/google/uuid v1.6.0 // indirect github.com/grafana/regexp v0.0.0-20240518133315-a468a5bfb3bc // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect - github.com/lib/pq v1.10.9 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/pgx/v5 v5.10.0 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/mattn/go-colorable v0.1.13 // indirect github.com/mattn/go-isatty v0.0.19 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect @@ -38,6 +41,7 @@ require ( go.opentelemetry.io/proto/otlp v1.7.1 // indirect go.yaml.in/yaml/v2 v2.4.2 // indirect golang.org/x/net v0.48.0 // indirect + golang.org/x/sync v0.19.0 // indirect golang.org/x/sys v0.45.0 // indirect golang.org/x/text v0.32.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 // indirect diff --git a/examples/postgresql-service/go.sum b/examples/postgresql-service/go.sum index ae15668..3967939 100644 --- a/examples/postgresql-service/go.sum +++ b/examples/postgresql-service/go.sum @@ -5,6 +5,7 @@ github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F9 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= @@ -23,6 +24,14 @@ github.com/grafana/regexp v0.0.0-20240518133315-a468a5bfb3bc h1:GN2Lv3MGO7AS6PrR github.com/grafana/regexp v0.0.0-20240518133315-a468a5bfb3bc/go.mod h1:+JKpmjMGhpgPL+rXZ5nsZieVzvarn86asRlBg4uNGnk= github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 h1:8Tjv8EJ+pM1xP8mK6egEbD1OgnVTyacbefKhmbLhIhU= github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2/go.mod h1:pkJQ2tZHJ0aFOVEEot6oZmaVEZcRme73eIFmhiVuRWs= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= +github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= @@ -31,8 +40,6 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= -github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= -github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= @@ -58,6 +65,9 @@ github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7 github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0= github.com/rs/zerolog v1.34.0 h1:k43nTLIwcTVQAncfCw4KZ2VY6ukYoZaBPNOE8txlOeY= github.com/rs/zerolog v1.34.0/go.mod h1:bJsvje4Z08ROH4Nhs5iH600c3IkWhwp44iRc54W6wYQ= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= @@ -90,6 +100,8 @@ go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU= golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY= +golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= +golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= @@ -110,5 +122,6 @@ google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/framework/app.go b/framework/app.go index 89db954..c7581cd 100644 --- a/framework/app.go +++ b/framework/app.go @@ -253,9 +253,11 @@ type App struct { shutdownHooks []ShutdownHook // Lifecycle - mu sync.RWMutex - running bool - stopping bool + mu sync.RWMutex + running bool + starting bool + stopping bool + readinessStop chan struct{} } // AppOption configures the App during creation. @@ -325,6 +327,44 @@ func WithConfig(config *config.BaseConfig) AppOption { } } +// WithLogging injects a custom logging manager, overriding the framework's +// built-in default (a LoggingManager derived from the app config). +func WithLogging(logging *LoggingManager) AppOption { + return func(app *App) error { + if logging == nil { + return fmt.Errorf("logging manager cannot be nil") + } + app.logging = logging + return nil + } +} + +// WithObservability injects a custom observability manager, overriding the +// framework's built-in default (an ObservabilityManager derived from the app +// config and version). +func WithObservability(observability *ObservabilityManager) AppOption { + return func(app *App) error { + if observability == nil { + return fmt.Errorf("observability manager cannot be nil") + } + app.observability = observability + return nil + } +} + +// WithHealthRegistry injects a custom health registry, overriding the +// framework's built-in default (a forgeHealth.Registry logging through the +// app's logging manager). +func WithHealthRegistry(registry *forgeHealth.Registry) AppOption { + return func(app *App) error { + if registry == nil { + return fmt.Errorf("health registry cannot be nil") + } + app.healthRegistry = registry + return nil + } +} + // WithVersion sets the service version. func WithVersion(version string) AppOption { return func(app *App) error { @@ -460,6 +500,11 @@ func (a *App) HealthRegistry() *forgeHealth.Registry { return a.healthRegistry } +// Observability returns the observability manager. +func (a *App) Observability() *ObservabilityManager { + return a.observability +} + // IsRunning returns true if the application is running. func (a *App) IsRunning() bool { a.mu.RLock() @@ -519,13 +564,34 @@ func (a *App) Run(ctx context.Context) error { } // Start starts all application components. +// +// The app mutex is held only long enough to claim the starting state, not +// across the (I/O-bound) bundle, server, and component startup below: a +// component that calls IsRunning/IsStopping from its own Start would +// otherwise deadlock against this method holding the lock for the whole +// startup sequence. The lock is re-acquired only to record the outcome. func (a *App) Start(ctx context.Context) error { a.mu.Lock() - defer a.mu.Unlock() - - if a.running { + if a.running || a.starting { + a.mu.Unlock() return fmt.Errorf("application is already running") } + if a.stopping { + a.mu.Unlock() + return fmt.Errorf("application is stopping") + } + a.starting = true + a.mu.Unlock() + + started := false + defer func() { + a.mu.Lock() + a.starting = false + if started { + a.running = true + } + a.mu.Unlock() + }() logger := a.logging.WithService(a.config.ServiceName, "start") @@ -585,16 +651,26 @@ func (a *App) Start(ctx context.Context) error { } } - // Mark service as ready after initial delay + // Mark service as ready after initial delay. The delay is cancellable so + // that a Stop() during the delay window prevents the flip to ready: + // without this, a slow-starting service could be marked ready after it + // has already begun shutting down. if a.config.ReadinessInitialDelay > 0 { - go func() { - time.Sleep(a.config.ReadinessInitialDelay) - a.healthRegistry.SetReady(true) - a.updateGRPCHealthStatus() - logger.Info(). - Dur("delay", a.config.ReadinessInitialDelay). - Msg("Service marked as ready") - }() + delay := a.config.ReadinessInitialDelay + a.readinessStop = make(chan struct{}) + go func(stop <-chan struct{}) { + timer := time.NewTimer(delay) + defer timer.Stop() + + select { + case <-stop: + return + case <-timer.C: + a.healthRegistry.SetReady(true) + a.updateGRPCHealthStatus() + logger.Info().Dur("delay", delay).Msg("Service marked as ready") + } + }(a.readinessStop) } else { a.healthRegistry.SetReady(true) a.updateGRPCHealthStatus() @@ -608,7 +684,7 @@ func (a *App) Start(ctx context.Context) error { go a.syncGRPCHealthLoop(a.grpcHealthSyncStop) } - a.running = true + started = true logger.Info().Msg("Service started successfully") return nil @@ -633,6 +709,13 @@ func (a *App) Stop(ctx context.Context) error { a.healthRegistry.SetReady(false) a.updateGRPCHealthStatus() + // Cancel a pending readiness-delay flip, if any, so it can't race this + // SetReady(false) and mark the service ready again after shutdown began. + if a.readinessStop != nil { + close(a.readinessStop) + a.readinessStop = nil + } + if a.grpcHealthSyncStop != nil { close(a.grpcHealthSyncStop) a.grpcHealthSyncStop = nil diff --git a/framework/app_lifecycle_test.go b/framework/app_lifecycle_test.go index 1974538..e06656a 100644 --- a/framework/app_lifecycle_test.go +++ b/framework/app_lifecycle_test.go @@ -2,6 +2,9 @@ package framework import ( "context" + "runtime" + "strings" + "sync" "testing" "time" @@ -503,6 +506,202 @@ func TestApp_Lifecycle_HTTPOnlyMode(t *testing.T) { } } +// TestApp_Lifecycle_ComponentCallingIsRunningDuringStart verifies that a +// component calling IsRunning()/IsStopping() from its own Start() does not +// deadlock against App.Start() holding the app mutex across startup. +func TestApp_Lifecycle_ComponentCallingIsRunningDuringStart(t *testing.T) { + cfg := config.DefaultBaseConfig() + cfg.ServiceName = "is-running-during-start-test" + cfg.GRPCAddr = ":0" + cfg.HTTPAddr = ":0" + + comp := &isRunningCallingComponent{} + + app, err := New( + WithConfig(&cfg), + WithComponent(comp), + ) + if err != nil { + t.Fatalf("Failed to create app: %v", err) + } + comp.app = app + + startErr := make(chan error, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + startErr <- app.Start(ctx) + }() + + select { + case err := <-startErr: + if err != nil { + t.Fatalf("Start returned an error: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Start deadlocked when a component called IsRunning/IsStopping from its own Start") + } + + if !comp.called { + t.Fatal("expected component Start to have been called") + } + if comp.sawRunning { + t.Error("expected IsRunning() to report false while the app is still starting") + } + if comp.sawStopping { + t.Error("expected IsStopping() to report false during normal startup") + } + + stopCtx, stopCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer stopCancel() + if err := app.Stop(stopCtx); err != nil { + t.Errorf("Failed to stop app: %v", err) + } +} + +type isRunningCallingComponent struct { + app *App + called bool + sawRunning bool + sawStopping bool +} + +func (c *isRunningCallingComponent) Start(ctx context.Context) error { + c.called = true + c.sawRunning = c.app.IsRunning() + c.sawStopping = c.app.IsStopping() + return nil +} + +func (c *isRunningCallingComponent) Stop(ctx context.Context) error { + return nil +} + +// TestApp_Lifecycle_ConcurrentStartRejected verifies that a second concurrent +// Start() call is rejected while the first is still in flight, rather than +// both proceeding or racing on the running/starting state. +func TestApp_Lifecycle_ConcurrentStartRejected(t *testing.T) { + cfg := config.DefaultBaseConfig() + cfg.ServiceName = "concurrent-start-test" + cfg.GRPCAddr = ":0" + cfg.HTTPAddr = ":0" + + slow := &TestComponentWithCallbacks{ + startFn: func() error { + time.Sleep(200 * time.Millisecond) + return nil + }, + } + + app, err := New( + WithConfig(&cfg), + WithComponent(slow), + ) + if err != nil { + t.Fatalf("Failed to create app: %v", err) + } + + const attempts = 2 + errs := make([]error, attempts) + + var wg sync.WaitGroup + for i := 0; i < attempts; i++ { + wg.Add(1) + go func(idx int) { + defer wg.Done() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + errs[idx] = app.Start(ctx) + }(i) + } + wg.Wait() + + successes, failures := 0, 0 + for _, err := range errs { + switch { + case err == nil: + successes++ + case strings.Contains(err.Error(), "already running"): + failures++ + default: + t.Errorf("unexpected Start error: %v", err) + } + } + + if successes != 1 || failures != 1 { + t.Fatalf("expected exactly one Start to succeed and one to be rejected, got %d successes and %d failures", successes, failures) + } + + if !app.IsRunning() { + t.Error("expected app to be running after the concurrent Start calls settle") + } + + stopCtx, stopCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer stopCancel() + if err := app.Stop(stopCtx); err != nil { + t.Errorf("Failed to stop app: %v", err) + } +} + +// TestApp_Lifecycle_ReadinessDelay_StopCancelsReadyFlip verifies that calling +// Stop() while a ReadinessInitialDelay is pending prevents the delayed +// SetReady(true) from firing after shutdown has begun, and that the delay +// goroutine exits promptly instead of leaking. +func TestApp_Lifecycle_ReadinessDelay_StopCancelsReadyFlip(t *testing.T) { + cfg := config.DefaultBaseConfig() + cfg.ServiceName = "readiness-delay-stop-test" + cfg.GRPCAddr = ":0" + cfg.HTTPAddr = ":0" + cfg.ReadinessInitialDelay = 150 * time.Millisecond + + baseline := runtime.NumGoroutine() + + app, err := New(WithConfig(&cfg)) + if err != nil { + t.Fatalf("Failed to create app: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := app.Start(ctx); err != nil { + t.Fatalf("Failed to start app: %v", err) + } + + if app.HealthRegistry().IsReady() { + t.Fatal("expected service to not be ready immediately after Start with a readiness delay configured") + } + + stopCtx, stopCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer stopCancel() + if err := app.Stop(stopCtx); err != nil { + t.Fatalf("Failed to stop app: %v", err) + } + + if app.HealthRegistry().IsReady() { + t.Error("expected service to not be ready immediately after Stop") + } + + // Wait past the original delay to make sure the cancelled goroutine never + // fires the delayed SetReady(true). + time.Sleep(cfg.ReadinessInitialDelay * 3) + + if app.HealthRegistry().IsReady() { + t.Error("expected readiness delay goroutine to not mark the service ready after Stop") + } + + // Bounded wait for the goroutine population to settle back near baseline, + // confirming the readiness delay goroutine (and everything else spun up + // by Start/Stop) actually exited rather than leaking. + deadline := time.Now().Add(2 * time.Second) + for runtime.NumGoroutine() > baseline+2 { + if time.Now().After(deadline) { + t.Errorf("goroutine count did not settle after Stop: baseline=%d, current=%d", baseline, runtime.NumGoroutine()) + break + } + time.Sleep(10 * time.Millisecond) + } +} + // Helper test component with callback functions type TestComponentWithCallbacks struct { started bool diff --git a/framework/app_test.go b/framework/app_test.go index 5c1e596..b92f838 100644 --- a/framework/app_test.go +++ b/framework/app_test.go @@ -726,3 +726,87 @@ func TestApp_WithStreamInterceptor_Valid(t *testing.T) { t.Errorf("expected 1 stream interceptor, got %d", len(app.streamInterceptors)) } } + +func TestApp_WithLogging_Nil(t *testing.T) { + cfg := config.DefaultBaseConfig() + cfg.ServiceName = "test-service" + + _, err := New(WithConfig(&cfg), WithLogging(nil)) + if err == nil { + t.Fatal("expected error for nil logging manager") + } + if !strings.Contains(err.Error(), "logging manager cannot be nil") { + t.Errorf("expected 'logging manager cannot be nil', got %q", err.Error()) + } +} + +func TestApp_WithLogging_Valid(t *testing.T) { + cfg := config.DefaultBaseConfig() + cfg.ServiceName = "test-service" + + custom := NewLoggingManager(&cfg) + + app, err := New(WithConfig(&cfg), WithLogging(custom)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if app.Logger() != custom { + t.Error("expected app.Logger() to return the injected logging manager") + } +} + +func TestApp_WithObservability_Nil(t *testing.T) { + cfg := config.DefaultBaseConfig() + cfg.ServiceName = "test-service" + + _, err := New(WithConfig(&cfg), WithObservability(nil)) + if err == nil { + t.Fatal("expected error for nil observability manager") + } + if !strings.Contains(err.Error(), "observability manager cannot be nil") { + t.Errorf("expected 'observability manager cannot be nil', got %q", err.Error()) + } +} + +func TestApp_WithObservability_Valid(t *testing.T) { + cfg := config.DefaultBaseConfig() + cfg.ServiceName = "test-service" + + custom := NewObservabilityManager(NewObservabilityConfig(&cfg, "test-1.0.0")) + + app, err := New(WithConfig(&cfg), WithObservability(custom)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if app.Observability() != custom { + t.Error("expected app.Observability() to return the injected observability manager") + } +} + +func TestApp_WithHealthRegistry_Nil(t *testing.T) { + cfg := config.DefaultBaseConfig() + cfg.ServiceName = "test-service" + + _, err := New(WithConfig(&cfg), WithHealthRegistry(nil)) + if err == nil { + t.Fatal("expected error for nil health registry") + } + if !strings.Contains(err.Error(), "health registry cannot be nil") { + t.Errorf("expected 'health registry cannot be nil', got %q", err.Error()) + } +} + +func TestApp_WithHealthRegistry_Valid(t *testing.T) { + cfg := config.DefaultBaseConfig() + cfg.ServiceName = "test-service" + + custom := forgeHealth.NewRegistry(NewHealthLogger(NewLoggingManager(&cfg))) + + app, err := New(WithConfig(&cfg), WithHealthRegistry(custom)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if app.HealthRegistry() != custom { + t.Error("expected app.HealthRegistry() to return the injected health registry") + } +} diff --git a/go.mod b/go.mod index 227615b..577ba5b 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( github.com/cenkalti/backoff/v4 v4.3.0 github.com/fsnotify/fsnotify v1.9.0 github.com/golang-jwt/jwt/v5 v5.3.0 - github.com/lib/pq v1.10.9 + github.com/jackc/pgx/v5 v5.10.0 github.com/prometheus/client_golang v1.23.2 github.com/redis/go-redis/v9 v9.14.0 github.com/rs/zerolog v1.34.0 @@ -33,6 +33,9 @@ require ( github.com/google/uuid v1.6.0 // indirect github.com/grafana/regexp v0.0.0-20240518133315-a468a5bfb3bc // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/mattn/go-colorable v0.1.13 // indirect github.com/mattn/go-isatty v0.0.19 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect @@ -45,6 +48,7 @@ require ( go.opentelemetry.io/proto/otlp v1.7.1 // indirect go.yaml.in/yaml/v2 v2.4.2 // indirect golang.org/x/net v0.48.0 // indirect + golang.org/x/sync v0.19.0 // indirect golang.org/x/sys v0.45.0 // indirect golang.org/x/text v0.32.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 // indirect diff --git a/go.sum b/go.sum index d96ed42..c085a0d 100644 --- a/go.sum +++ b/go.sum @@ -36,6 +36,14 @@ github.com/grafana/regexp v0.0.0-20240518133315-a468a5bfb3bc h1:GN2Lv3MGO7AS6PrR github.com/grafana/regexp v0.0.0-20240518133315-a468a5bfb3bc/go.mod h1:+JKpmjMGhpgPL+rXZ5nsZieVzvarn86asRlBg4uNGnk= github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 h1:8Tjv8EJ+pM1xP8mK6egEbD1OgnVTyacbefKhmbLhIhU= github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2/go.mod h1:pkJQ2tZHJ0aFOVEEot6oZmaVEZcRme73eIFmhiVuRWs= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= +github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= @@ -44,8 +52,6 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= -github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= -github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= @@ -77,6 +83,7 @@ github.com/sony/gobreaker v1.0.0 h1:feX5fGGXSl3dYd4aHZItw+FpHLvvoaqkawKjVNiFMNQ= github.com/sony/gobreaker v1.0.0/go.mod h1:ZKptC7FHNvhBz7dN2LGjPVBz2sZJmc0/PkyDJOjmxWY= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= @@ -109,6 +116,8 @@ go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU= golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY= +golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= +golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= @@ -129,5 +138,6 @@ google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=