perf(startup): fetch distinct credential sources concurrently

CHANGELOG.md

@@ -4,6 +4,12 @@ Gatekeeper is a standalone credential-injecting TLS-intercepting proxy. It trans
 
 Gatekeeper is pre-1.0. The configuration schema and credential source interface may change between minor versions.
 
+## v0.25.2 — 2026-10-12
+
+### Changed
+
+- **Startup fetches distinct credential sources concurrently** (`gatekeeper.go`) — `loadCredentials` fetched each distinct source one after another before the listener opened. A production config holds about a dozen network sources (`gcp-secretmanager`, `gcp-service-account`, `github-app`), so a container needed about 1.5 s from start to listening. Sources are now built in config order, fetched at once (at most 16 in flight, each bounded by the same 10 s), and applied in config order. Behavior is unchanged: every credential is loaded before the listener accepts a connection, identical `source` blocks still share one fetch and one refresh goroutine, a build error (a bad source block, a missing key file) is reported before any fetch error, the first credential in the file whose fetch fails is the fetch error that stops startup, and no error carries a credential value. Host commands of `process` sources now run at the same time as other sources at startup. A failed startup closes every source it built. Refresh goroutines now start in config order instead of map order.
+
 ## v0.25.1 — 2026-10-05
 
 ### Fixed

gatekeeper.go

@@ -506,14 +506,45 @@ func New(ctx context.Context, cfg *Config, version string) (*Server, error) {
 // source (e.g., two hosts using the same github-app), a single Fetch is made at
 // startup and a single background refresh goroutine is registered. The token is
 // applied to all hosts that share the source.
-func (s *Server) loadCredentials(ctx context.Context, cfg *Config) error {
+func (s *Server) loadCredentials(ctx context.Context, cfg *Config) (err error) {
 	// Dedup sources: same SourceConfig → one Fetch, one refresh goroutine.
-	type fetchedSource struct {
+	// Sources are built in config order, fetched concurrently (each fetch is
+	// a network round trip, and a production config has about a dozen), and
+	// applied in config order. Build errors come before fetch errors; among
+	// fetch errors the first credential in the file is the one reported.
+	type pendingSource struct {
 		src credentialsource.CredentialSource
 		val string
+		err error
 	}
-	fetched := make(map[SourceConfig]*fetchedSource)
-	refreshMap := make(map[SourceConfig]*pendingRefresh)
+	type credEntry struct {
+		cred   CredentialConfig
+		header string
+		source *pendingSource
+		first  bool
+	}
+	sources := make(map[SourceConfig]*pendingSource)
+	var unique []*pendingSource
+	var entries []credEntry
+	// New returns no Server on failure, so Stop never runs: release every
+	// source and Postgres resolver built here instead. Header sources join
+	// s.closers only once all of them loaded.
+	closersBase := len(s.closers)
+	defer func() {
+		if err == nil {
+			return
+		}
+		for _, ps := range unique {
+			if c, ok := ps.src.(io.Closer); ok {
+				_ = c.Close()
+			}
+		}
+		for _, c := range s.closers[closersBase:] {
+			_ = c.Close()
+		}
+		s.closers = s.closers[:closersBase]
+	}()
+	var sourceClosers []io.Closer
 
 	for _, cred := range cfg.Credentials {
 		if cred.Host == "" {
@@ -542,17 +573,10 @@ func (s *Server) loadCredentials(ctx context.Context, cfg *Config) error {
 			return fmt.Errorf("credential for %s: format %q is only supported with the Authorization header", cred.Host, cred.Format)
 		}
 
-		// Reuse a previously fetched source with the same config, avoiding
+		// Reuse a previously built source with the same config, avoiding
 		// redundant source construction (e.g., os.ReadFile for github-app keys).
-		if fs, ok := fetched[cred.Source]; ok {
-			val := fs.val
-			if strings.EqualFold(header, "Authorization") {
-				val = ensureAuthScheme(val, cred.Prefix, cred.Format)
-			}
-			s.proxy.SetCredentialWithGrant(cred.Host, header, val, cred.Grant)
-			if _, ok := fs.src.(credentialsource.RefreshingSource); ok {
-				refreshMap[cred.Source].creds = append(refreshMap[cred.Source].creds, cred)
-			}
+		if ps, ok := sources[cred.Source]; ok {
+			entries = append(entries, credEntry{cred: cred, header: header, source: ps})
 			continue
 		}
 
@@ -579,34 +603,64 @@ func (s *Server) loadCredentials(ctx context.Context, cfg *Config) error {
 			}
 		}
 
-		fetchCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
-		val, fetchErr := src.Fetch(fetchCtx)
-		cancel()
-		if fetchErr != nil {
-			return fmt.Errorf("credential for %s: fetch failed: %w", cred.Host, fetchErr)
+		ps := &pendingSource{src: src}
+		sources[cred.Source] = ps
+		unique = append(unique, ps)
+		entries = append(entries, credEntry{cred: cred, header: header, source: ps, first: true})
+	}
+
+	const maxConcurrentFetches = 16
+	var wg sync.WaitGroup
+	slots := make(chan struct{}, maxConcurrentFetches)
+	for _, ps := range unique {
+		wg.Add(1)
+		slots <- struct{}{}
+		go func() {
+			defer wg.Done()
+			defer func() { <-slots }()
+			fetchCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+			defer cancel()
+			ps.val, ps.err = ps.src.Fetch(fetchCtx)
+		}()
+	}
+	wg.Wait()
+
+	refreshMap := make(map[*pendingSource]*pendingRefresh)
+	var refreshOrder []*pendingSource
+	for _, e := range entries {
+		if e.source.err != nil {
+			// Never include the credential value in the error.
+			return fmt.Errorf("credential for %s: fetch failed: %w", e.cred.Host, e.source.err)
 		}
-		fetched[cred.Source] = &fetchedSource{src: src, val: val}
-		if c, ok := src.(io.Closer); ok {
-			s.closers = append(s.closers, c)
+		if e.first {
+			if c, ok := e.source.src.(io.Closer); ok {
+				sourceClosers = append(sourceClosers, c)
+			}
 		}
 
 		// For Authorization headers, ensure the value includes an auth
 		// scheme prefix. In the CLI flow, providers handle this (e.g.,
 		// GitHub provider prepends "Bearer "). The gatekeeper bypasses
 		// providers, so we auto-detect the scheme from the token format.
-		if strings.EqualFold(header, "Authorization") {
-			val = ensureAuthScheme(val, cred.Prefix, cred.Format)
+		val := e.source.val
+		if strings.EqualFold(e.header, "Authorization") {
+			val = ensureAuthScheme(val, e.cred.Prefix, e.cred.Format)
 		}
+		s.proxy.SetCredentialWithGrant(e.cred.Host, e.header, val, e.cred.Grant)
 
-		s.proxy.SetCredentialWithGrant(cred.Host, header, val, cred.Grant)
-
-		if rs, ok := src.(credentialsource.RefreshingSource); ok {
-			refreshMap[cred.Source] = &pendingRefresh{src: rs, creds: []CredentialConfig{cred}}
+		if rs, ok := e.source.src.(credentialsource.RefreshingSource); ok {
+			if pr, ok := refreshMap[e.source]; ok {
+				pr.creds = append(pr.creds, e.cred)
+			} else {
+				refreshMap[e.source] = &pendingRefresh{src: rs, creds: []CredentialConfig{e.cred}}
+				refreshOrder = append(refreshOrder, e.source)
+			}
 		}
 	}
 
-	for _, pr := range refreshMap {
-		s.pendingRefreshes = append(s.pendingRefreshes, *pr)
+	s.closers = append(s.closers, sourceClosers...)
+	for _, ps := range refreshOrder {
+		s.pendingRefreshes = append(s.pendingRefreshes, *refreshMap[ps])
 	}
 	return nil
 }

gatekeeper_test.go

@@ -3560,3 +3560,198 @@ func TestServerStopClosesIdleHTTP2Upstream(t *testing.T) {
 		t.Fatal("standalone stop left an idle upstream HTTP/2 connection open")
 	}
 }
+
+// barrierSource blocks each Fetch until want Fetches are in flight at once.
+type barrierSource struct {
+	inflight *atomic.Int32
+	want     int32
+	name     string
+}
+
+func (b barrierSource) Fetch(ctx context.Context) (string, error) {
+	b.inflight.Add(1)
+	for b.inflight.Load() < b.want {
+		select {
+		case <-ctx.Done():
+			return "", fmt.Errorf("%s: only %d fetches in flight, want %d: %w", b.name, b.inflight.Load(), b.want, ctx.Err())
+		case <-time.After(time.Millisecond):
+		}
+	}
+	return "tok_" + b.name, nil
+}
+
+func (b barrierSource) Type() string { return "barrier" }
+
+// A production config holds about a dozen network credential sources
+// (gcp-secretmanager, gcp-service-account, github-app). Fetching them one by
+// one put 1.5 s between container start and the listener.
+func TestLoadCredentials_FetchesDistinctSourcesConcurrently(t *testing.T) {
+	srv, err := New(context.Background(), &Config{Proxy: ProxyConfig{Port: 0, Host: "127.0.0.1"}}, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var inflight atomic.Int32
+	const n = 12
+	srv.resolveSource = func(sc SourceConfig) (credentialsource.CredentialSource, error) {
+		return barrierSource{inflight: &inflight, want: n, name: sc.Secret}, nil
+	}
+	cfg := &Config{}
+	for i := 0; i < n; i++ {
+		name := fmt.Sprintf("secret-%d", i)
+		cfg.Credentials = append(cfg.Credentials, CredentialConfig{
+			Host: fmt.Sprintf("api%d.example.com", i), Grant: name,
+			Source: SourceConfig{Type: "gcp-secretmanager", Secret: name},
+		})
+	}
+	if err := srv.loadCredentials(context.Background(), cfg); err != nil {
+		t.Fatalf("loadCredentials: %v", err)
+	}
+	if got := inflight.Load(); got != n {
+		t.Fatalf("fetches started = %d, want %d", got, n)
+	}
+}
+
+// failAfterSource fails once release closes; failNow fails at once.
+type failAfterSource struct {
+	release <-chan struct{}
+	msg     string
+	done    chan<- struct{}
+}
+
+func (f failAfterSource) Fetch(ctx context.Context) (string, error) {
+	if f.release != nil {
+		select {
+		case <-f.release:
+		case <-ctx.Done():
+			return "", ctx.Err()
+		}
+	}
+	if f.done != nil {
+		close(f.done)
+	}
+	return "", errors.New(f.msg)
+}
+
+func (f failAfterSource) Type() string { return "fail" }
+
+func TestLoadCredentials_FetchFailureNamesTheFirstFailingCredentialInConfigOrder(t *testing.T) {
+	srv, err := New(context.Background(), &Config{Proxy: ProxyConfig{Port: 0, Host: "127.0.0.1"}}, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	bFailed := make(chan struct{})
+	srv.resolveSource = func(sc SourceConfig) (credentialsource.CredentialSource, error) {
+		if sc.Secret == "a" {
+			return failAfterSource{release: bFailed, msg: "boom a"}, nil
+		}
+		return failAfterSource{msg: "boom b", done: bFailed}, nil
+	}
+	cfg := &Config{Credentials: []CredentialConfig{
+		{Host: "a.example.com", Grant: "a", Source: SourceConfig{Type: "mock", Secret: "a"}},
+		{Host: "b.example.com", Grant: "b", Source: SourceConfig{Type: "mock", Secret: "b"}},
+	}}
+	err = srv.loadCredentials(context.Background(), cfg)
+	if err == nil || !strings.Contains(err.Error(), "a.example.com") || !strings.Contains(err.Error(), "boom a") {
+		t.Fatalf("err = %v, want the failure of a.example.com though b.example.com failed first", err)
+	}
+}
+
+type closeTrackingSource struct {
+	closed *atomic.Int32
+	fail   bool
+}
+
+func (c closeTrackingSource) Fetch(context.Context) (string, error) {
+	if c.fail {
+		return "", errors.New("boom")
+	}
+	return "tok", nil
+}
+func (c closeTrackingSource) Type() string { return "close-tracking" }
+func (c closeTrackingSource) Close() error { c.closed.Add(1); return nil }
+
+func TestLoadCredentials_ClosesEveryBuiltSourceWhenAFetchFails(t *testing.T) {
+	srv, err := New(context.Background(), &Config{Proxy: ProxyConfig{Port: 0, Host: "127.0.0.1"}}, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var closed atomic.Int32
+	srv.resolveSource = func(sc SourceConfig) (credentialsource.CredentialSource, error) {
+		return closeTrackingSource{closed: &closed, fail: sc.Secret == "s0"}, nil
+	}
+	cfg := &Config{}
+	for i := 0; i < 5; i++ {
+		name := fmt.Sprintf("s%d", i)
+		cfg.Credentials = append(cfg.Credentials, CredentialConfig{Host: name + ".example.com", Grant: name, Source: SourceConfig{Type: "mock", Secret: name}})
+	}
+	if err := srv.loadCredentials(context.Background(), cfg); err == nil {
+		t.Fatal("loadCredentials succeeded, want the fetch failure")
+	}
+	if got := closed.Load(); got != 5 {
+		t.Fatalf("closed sources = %d, want 5", got)
+	}
+}
+
+type peakSource struct {
+	cur, peak *atomic.Int32
+}
+
+func (p peakSource) Fetch(context.Context) (string, error) {
+	n := p.cur.Add(1)
+	for {
+		old := p.peak.Load()
+		if n <= old || p.peak.CompareAndSwap(old, n) {
+			break
+		}
+	}
+	time.Sleep(5 * time.Millisecond)
+	p.cur.Add(-1)
+	return "tok", nil
+}
+func (p peakSource) Type() string { return "peak" }
+
+func TestLoadCredentials_BoundsConcurrentFetches(t *testing.T) {
+	srv, err := New(context.Background(), &Config{Proxy: ProxyConfig{Port: 0, Host: "127.0.0.1"}}, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var cur, peak atomic.Int32
+	srv.resolveSource = func(SourceConfig) (credentialsource.CredentialSource, error) {
+		return peakSource{cur: &cur, peak: &peak}, nil
+	}
+	cfg := &Config{}
+	for i := 0; i < 40; i++ {
+		name := fmt.Sprintf("s%d", i)
+		cfg.Credentials = append(cfg.Credentials, CredentialConfig{Host: name + ".example.com", Grant: name, Source: SourceConfig{Type: "mock", Secret: name}})
+	}
+	if err := srv.loadCredentials(context.Background(), cfg); err != nil {
+		t.Fatalf("loadCredentials: %v", err)
+	}
+	if got := peak.Load(); got > 16 || got < 2 {
+		t.Fatalf("peak concurrent fetches = %d, want between 2 and 16", got)
+	}
+}
+
+func TestLoadCredentials_ClosesPostgresResolversWhenALaterFetchFails(t *testing.T) {
+	srv, err := New(context.Background(), &Config{Proxy: ProxyConfig{Port: 0, Host: "127.0.0.1"}}, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var closed atomic.Int32
+	srv.resolveSource = func(sc SourceConfig) (credentialsource.CredentialSource, error) {
+		return closeTrackingSource{closed: &closed, fail: sc.Secret == "bad"}, nil
+	}
+	cfg := &Config{Credentials: []CredentialConfig{
+		{Host: "db.example.com", Grant: "db", Source: SourceConfig{Type: "mock", Secret: "neon"}, Postgres: &PostgresCredentialConfig{Resolver: "neon", Project: "p"}},
+		{Host: "bad.example.com", Grant: "bad", Source: SourceConfig{Type: "mock", Secret: "bad"}},
+	}}
+	if err := srv.loadCredentials(context.Background(), cfg); err == nil {
+		t.Fatal("loadCredentials succeeded, want the fetch failure")
+	}
+	if closed.Load() < 2 {
+		t.Fatalf("closed = %d, want the Postgres source and the failed source closed", closed.Load())
+	}
+	if len(srv.closers) != 0 {
+		t.Fatalf("closers left after a failed load = %d, want 0", len(srv.closers))
+	}
+}