From c36fc6250b8f2d8e0d6a6acda22d5a1cdfc81b4c Mon Sep 17 00:00:00 2001 From: shiv Date: Sun, 9 Aug 2026 22:01:31 +0530 Subject: [PATCH 1/4] =?UTF-8?q?=E2=9A=A1=20Bolt:=20Run=20datasource=20heal?= =?UTF-8?q?th=20checks=20concurrently=20in=20Registry.HealthReport?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .jules/bolt.md | 3 ++ pkg/proxy/registry.go | 74 ++++++++++++++++++++++---------------- pkg/proxy/registry_test.go | 32 +++++++++++++++++ 3 files changed, 79 insertions(+), 30 deletions(-) create mode 100644 .jules/bolt.md diff --git a/.jules/bolt.md b/.jules/bolt.md new file mode 100644 index 0000000..948f70b --- /dev/null +++ b/.jules/bolt.md @@ -0,0 +1,3 @@ +## 2026-08-09 - Parallel Datasource Health Checks +**Learning:** In proxy registries managing multiple datasources (DB, SSH, HTTP, Kafka, etc.), sequential health checks cause total latency to scale linearly as O(N * timeout), blocking reporting threads when target endpoints time out. +**Action:** Always perform multi-datasource health probes and metadata collection concurrently using goroutines with per-check context timeouts. diff --git a/pkg/proxy/registry.go b/pkg/proxy/registry.go index d5bf268..109f7a0 100644 --- a/pkg/proxy/registry.go +++ b/pkg/proxy/registry.go @@ -149,50 +149,64 @@ type DatasourceHealth struct { LastCheck string `json:"last_check"` // RFC3339 } -// HealthReport runs health checks on all registered datasources and returns a map +// HealthReport runs health checks on all registered datasources concurrently and returns a map // of datasource ID → health status. Used for periodic reporting to the cloud. func (r *Registry) HealthReport(ctx context.Context) map[string]DatasourceHealth { r.mu.RLock() - ids := make([]string, 0, len(r.proxies)) - for id := range r.proxies { - ids = append(ids, id) + type target struct { + id string + proxy Proxy + cfg DatasourceEntry + } + targets := make([]target, 0, len(r.proxies)) + for id, p := range r.proxies { + if cfg, ok := r.configs[id]; ok { + targets = append(targets, target{id: id, proxy: p, cfg: cfg}) + } } r.mu.RUnlock() - report := make(map[string]DatasourceHealth, len(ids)) - now := time.Now().UTC().Format(time.RFC3339) + if len(targets) == 0 { + return make(map[string]DatasourceHealth) + } - for _, id := range ids { - r.mu.RLock() - p, pOk := r.proxies[id] - cfg, cOk := r.configs[id] - r.mu.RUnlock() + report := make(map[string]DatasourceHealth, len(targets)) + now := time.Now().UTC().Format(time.RFC3339) + var mu sync.Mutex + var wg sync.WaitGroup - if !pOk || !cOk { - continue - } + // Run health checks concurrently to prevent a single slow or timing-out + // datasource from blocking the health status of other datasources. + for _, t := range targets { + wg.Add(1) + go func(t target) { + defer wg.Done() - health := DatasourceHealth{ - Type: cfg.Type, - ProxyType: cfg.ProxyType, - Name: cfg.Name, - LastCheck: now, - } + health := DatasourceHealth{ + Type: t.cfg.Type, + ProxyType: t.cfg.ProxyType, + Name: t.cfg.Name, + LastCheck: now, + } - checkCtx, cancel := context.WithTimeout(ctx, 10*time.Second) - err := p.HealthCheck(checkCtx) - cancel() + checkCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + err := t.proxy.HealthCheck(checkCtx) + cancel() - if err != nil { - health.Status = "error" - health.Error = err.Error() - } else { - health.Status = "healthy" - } + if err != nil { + health.Status = "error" + health.Error = err.Error() + } else { + health.Status = "healthy" + } - report[id] = health + mu.Lock() + report[t.id] = health + mu.Unlock() + }(t) } + wg.Wait() return report } diff --git a/pkg/proxy/registry_test.go b/pkg/proxy/registry_test.go index c5bf7fd..77b6890 100644 --- a/pkg/proxy/registry_test.go +++ b/pkg/proxy/registry_test.go @@ -2,6 +2,7 @@ package proxy import ( "context" + "fmt" "sort" "sync" "testing" @@ -157,3 +158,34 @@ func TestRegistry_ConcurrentAccess(t *testing.T) { wg.Wait() // No race detector failures = pass } + +func TestRegistry_HealthReportConcurrent(t *testing.T) { + r := NewRegistry() + + // Register 5 fake proxies with mock health checks + for i := 1; i <= 5; i++ { + id := fmt.Sprintf("ds-%d", i) + entry := DatasourceEntry{ + ID: id, + Type: "postgresql", + ProxyType: "db-proxy", + Name: fmt.Sprintf("DB %d", i), + } + r.Register(id, entry, &fakeProxy{proxyType: "db-proxy"}) + } + + report := r.HealthReport(context.Background()) + if len(report) != 5 { + t.Fatalf("expected 5 health report entries, got %d", len(report)) + } + + for id, h := range report { + if h.Status != "healthy" { + t.Errorf("expected status 'healthy' for %s, got %s", id, h.Status) + } + if h.ProxyType != "db-proxy" { + t.Errorf("expected proxy_type 'db-proxy' for %s, got %s", id, h.ProxyType) + } + } +} + From e517e0931ab9484335a0b34632689dd358171704 Mon Sep 17 00:00:00 2001 From: shiv Date: Sun, 9 Aug 2026 22:03:33 +0530 Subject: [PATCH 2/4] fix(registry): use defer cancel and check ctx.Err in HealthReport goroutines --- pkg/proxy/registry.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/pkg/proxy/registry.go b/pkg/proxy/registry.go index 109f7a0..0104f14 100644 --- a/pkg/proxy/registry.go +++ b/pkg/proxy/registry.go @@ -182,6 +182,10 @@ func (r *Registry) HealthReport(ctx context.Context) map[string]DatasourceHealth go func(t target) { defer wg.Done() + if ctx.Err() != nil { + return + } + health := DatasourceHealth{ Type: t.cfg.Type, ProxyType: t.cfg.ProxyType, @@ -190,8 +194,8 @@ func (r *Registry) HealthReport(ctx context.Context) map[string]DatasourceHealth } checkCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() err := t.proxy.HealthCheck(checkCtx) - cancel() if err != nil { health.Status = "error" From 7ef91b39422dfb3313eeab26439264075ccdf281 Mon Sep 17 00:00:00 2001 From: shiv Date: Sun, 9 Aug 2026 22:18:56 +0530 Subject: [PATCH 3/4] fix(registry): remove early return on ctx.Err in HealthReport to preserve full reporting --- pkg/proxy/registry.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/pkg/proxy/registry.go b/pkg/proxy/registry.go index 0104f14..aa2386d 100644 --- a/pkg/proxy/registry.go +++ b/pkg/proxy/registry.go @@ -182,10 +182,6 @@ func (r *Registry) HealthReport(ctx context.Context) map[string]DatasourceHealth go func(t target) { defer wg.Done() - if ctx.Err() != nil { - return - } - health := DatasourceHealth{ Type: t.cfg.Type, ProxyType: t.cfg.ProxyType, From 2a5d73a50479e65bf9edc4c751348ae99e9ed3f4 Mon Sep 17 00:00:00 2001 From: shiv Date: Sun, 9 Aug 2026 22:25:48 +0530 Subject: [PATCH 4/4] style: apply gofmt formatting to registry_test.go --- pkg/proxy/registry_test.go | 1 - 1 file changed, 1 deletion(-) diff --git a/pkg/proxy/registry_test.go b/pkg/proxy/registry_test.go index 77b6890..f32cfea 100644 --- a/pkg/proxy/registry_test.go +++ b/pkg/proxy/registry_test.go @@ -188,4 +188,3 @@ func TestRegistry_HealthReportConcurrent(t *testing.T) { } } } -