vrr/Pulse
2
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-10-03 04:38:48 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 17

fix(server): make local service-health probes dual-stack and bounded

Browse source 
Try both loopback families for IPv6 wildcard listeners without changing explicit or IPv4-only binds. Reserve each family a share of the API deadline, keep all later checks on the responding address, and retry a failed observation once with a fresh bounded budget in the telemetry background runner.

Keep genuine API, UI and asset failures visible and report only a closed timeout category, never transport details. Cover both real listener families, startup recovery, failure and privacy boundaries; synchronise the disclosure and subsystem contract.

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot] 2026-09-30 21:36:39 +01:00
parent 21a35494ae
commit fffaa64e57
10 changed files with 571 additions and 61 deletions
Download patch file Download diff file Expand all files Collapse all files

2
docs/PRIVACY.md
View file

@ -158,7 +158,7 @@ Every field is listed below with the reason it exists. Nothing else is included
| Update available | `true`/`false` | Report whether that last check offered a newer release, without sending which release |
| Service health observed | `true`/`false` | Distinguish a release that performed the bounded local UI/API self-check from an older release with no signal |
| Service health healthy | `true`/`false` | Report whether Pulse's locally bound listener served a healthy API response, UI document, and every referenced local frontend asset without sending an address, URL, response, or error text |
| Service health failure category | `listener`, `startup`, `runtime`, `api_connectivity`, `api_status`, `ui_status`, `frontend_assets`, or `unknown` | Classify a failed local self-check or startup path into one fixed category without sending the listener address, request URL, HTTP body, asset name, IP address, or raw error |
| Service health failure category | `listener`, `startup`, `runtime`, `api_connectivity`, `timeout`, `api_status`, `ui_status`, `frontend_assets`, or `unknown` | Classify a failed local self-check or startup path into one fixed category without sending the listener address, request URL, HTTP body, asset name, IP address, or raw error |
| Service health cohort | `first_observation`, `same_version`, or `version_change` | Mark whether this direct observation is the first schema-v13 observation, another observation of the same release, or the first observed state after a release-version change |
| Service health previous version | `6.4.0` | Retain only the normalized immediately previous Pulse release identity so aggregate reporting can compare a post-upgrade before/after cohort without treating rolling counters as new-release activity |
| Service health previous observed | `true`/`false` | State whether the immediately previous release recorded the bounded local service-health signal |

19
docs/release-control/v6/internal/subsystems/security-privacy.md
View file

@ -1492,9 +1492,9 @@ download URLs, command output, log lines, paths, hostnames, release asset URLs,
checksums, signatures, or operator-entered values.
Schema v13 adds a direct local service-health observation so a process-level
telemetry heartbeat is not mistaken for proof that the installed UI and API
are being served. The runtime probes its bound listener through loopback and
reports only observed/healthy booleans, one fixed failure class (`listener`,
`startup`, `runtime`, `api_connectivity`, `api_status`, `ui_status`,
are being served. The runtime probes the address bound by its TCP listener
(loopback for wildcard binds) and reports only observed/healthy booleans, one fixed failure class (`listener`,
`startup`, `runtime`, `api_connectivity`, `timeout`, `api_status`, `ui_status`,
`frontend_assets`, or `unknown`), a fixed observation cohort, and the
immediately previous normalized release's observed/healthy booleans. No probe
target, listener address, URL, IP address, asset path, response content, raw
@ -1502,6 +1502,19 @@ error, account, customer, or infrastructure identity may enter the payload or
persisted receiver row. The previous-release fields are direct adjacent-release
observations, not 30-day update counters, and are the only valid basis for a
before/after release-health cohort in the adoption report.
The service-health self-check must preserve the explicit bound TCP address and
port; wildcard IPv6/unspecified-family listeners try IPv4 and IPv6 loopback,
while an IPv4-only wildcard remains IPv4-only. No external address discovery,
proxy, redirect, or remote frontend asset fetch is permitted. One bounded
settling retry may follow a failed observation in the telemetry background
runner: at most two five-second attempts, separated by one second. Each attempt
shares its deadline across API/UI/assets and reserves time for an alternate
loopback family. Once an API responds, status/body/UI/asset failures cannot be
masked by switching families. Deadline and network timeout failures use only
the fixed `timeout` bucket; other failure classes remain visible. Tests must
cover unavailable IPv6 with working IPv4, IPv6-only serving, a stalled family,
startup recovery, final timeouts, explicit addresses, HTTPS, redirect refusal,
bounded bodies/assets, closed-category serialization and background execution.
That same outbound usage telemetry floor now also permits only content-free Pulse
Patrol control and governed Pulse Intelligence operations adoption flags and
counters inside the same rotating 30-day telemetry window:

2
frontend-modern/public/docs/PRIVACY.md
View file

@ -158,7 +158,7 @@ Every field is listed below with the reason it exists. Nothing else is included
| Update available | `true`/`false` | Report whether that last check offered a newer release, without sending which release |
| Service health observed | `true`/`false` | Distinguish a release that performed the bounded local UI/API self-check from an older release with no signal |
| Service health healthy | `true`/`false` | Report whether Pulse's locally bound listener served a healthy API response, UI document, and every referenced local frontend asset without sending an address, URL, response, or error text |
| Service health failure category | `listener`, `startup`, `runtime`, `api_connectivity`, `api_status`, `ui_status`, `frontend_assets`, or `unknown` | Classify a failed local self-check or startup path into one fixed category without sending the listener address, request URL, HTTP body, asset name, IP address, or raw error |
| Service health failure category | `listener`, `startup`, `runtime`, `api_connectivity`, `timeout`, `api_status`, `ui_status`, `frontend_assets`, or `unknown` | Classify a failed local self-check or startup path into one fixed category without sending the listener address, request URL, HTTP body, asset name, IP address, or raw error |
| Service health cohort | `first_observation`, `same_version`, or `version_change` | Mark whether this direct observation is the first schema-v13 observation, another observation of the same release, or the first observed state after a release-version change |
| Service health previous version | `6.4.0` | Retain only the normalized immediately previous Pulse release identity so aggregate reporting can compare a post-upgrade before/after cohort without treating rolling counters as new-release activity |
| Service health previous observed | `true`/`false` | State whether the immediately previous release recorded the bounded local service-health signal |

1
internal/telemetry/service_health.go
View file

@ -85,6 +85,7 @@ func canonicalServiceHealthFailureCategory(value string) string {
ServiceHealthFailureStartup,
ServiceHealthFailureRuntime,
ServiceHealthFailureAPIConnectivity,
ServiceHealthFailureTimeout,
ServiceHealthFailureAPIStatus,
ServiceHealthFailureUIStatus,
ServiceHealthFailureFrontendAssets:

81
internal/telemetry/service_health_test.go
View file

@ -165,3 +165,84 @@ func TestServiceHealthStateRejectsTamperedPreviousVersion(t *testing.T) {
t.Fatalf("tampered previous release escaped local boundary: %#v", ping)
}
}
// Closed categories survive both the outbound JSON payload and the persisted
// adjacent-version cohort. No transport error text is part of either shape.
func TestServiceHealthFailureCategoriesSerializeAndPersist(t *testing.T) {
categories := []string{
ServiceHealthFailureListener, ServiceHealthFailureStartup,
ServiceHealthFailureRuntime, ServiceHealthFailureAPIConnectivity,
ServiceHealthFailureTimeout, ServiceHealthFailureAPIStatus,
ServiceHealthFailureUIStatus, ServiceHealthFailureFrontendAssets,
ServiceHealthFailureUnknown,
}
for _, category := range categories {
t.Run(category, func(t *testing.T) {
dir := t.TempDir()
cfg := Config{Version: "6.4.6", DataDir: dir, GetServiceHealth: func() ServiceHealthObservation {
return ServiceHealthObservation{Observed: true, FailureCategory: " " + strings.ToUpper(category) + " "}
}}
ping, err := buildPingAt(cfg, "startup", time.Now().UTC())
if err != nil {
t.Fatal(err)
}
encoded, err := json.Marshal(ping)
if err != nil {
t.Fatal(err)
}
var fields map[string]interface{}
if err := json.Unmarshal(encoded, &fields); err != nil {
t.Fatal(err)
}
if fields["service_health_failure_category"] != category || fields["service_health_observed"] != true || fields["service_health_healthy"] != false {
t.Fatalf("serialized service-health fields = %#v", fields)
}
record := readServiceHealthRecord(dir)
if record.CurrentFailureCategory != category || !record.CurrentObserved || record.CurrentHealthy {
t.Fatalf("persisted observation = %#v", record)
}
cfg.Version = "6.4.7"
cfg.GetServiceHealth = func() ServiceHealthObservation { return ServiceHealthObservation{Observed: true, Healthy: true} }
recovered, err := buildPingAt(cfg, "startup", time.Now().UTC())
if err != nil {
t.Fatal(err)
}
record = readServiceHealthRecord(dir)
if recovered.ServiceHealthFailureCategory != "" || !recovered.ServiceHealthHealthy || record.PreviousFailureCategory != category || record.PreviousHealthy {
t.Fatalf("recovery lost the previous failure: %#v / %#v", recovered, record)
}
})
}
for _, category := range []string{"api_timeout: GET http://192.0.2.1/private", "timeout private.js", "timeout\nAuthorization"} {
if got := canonicalServiceHealthFailureCategory(category); got != ServiceHealthFailureUnknown {
t.Fatalf("free-form category accepted: %q -> %q", category, got)
}
}
}
func TestTelemetryStartDoesNotWaitForServiceHealth(t *testing.T) {
originalDelay := startupDelay
startupDelay = 0
defer func() { startupDelay = originalDelay }()
ctx, cancel := context.WithCancel(context.Background())
entered, release, started := make(chan struct{}), make(chan struct{}), make(chan struct{})
t.Cleanup(func() { close(release); cancel(); Stop() })
go func() {
Start(ctx, Config{Enabled: true, Version: "6.4.6", DataDir: t.TempDir(), GetServiceHealth: func() ServiceHealthObservation {
close(entered)
<-release
return ServiceHealthObservation{Observed: true, Healthy: true}
}})
close(started)
}()
select {
case <-entered:
case <-time.After(2 * time.Second):
t.Fatal("background probe did not begin")
}
select {
case <-started:
case <-time.After(2 * time.Second):
t.Fatal("telemetry startup blocked on its service-health probe")
}
}

1
internal/telemetry/telemetry.go
View file

@ -254,6 +254,7 @@ const (
ServiceHealthFailureStartup = "startup"
ServiceHealthFailureRuntime = "runtime"
ServiceHealthFailureAPIConnectivity = "api_connectivity"
ServiceHealthFailureTimeout = "timeout"
ServiceHealthFailureAPIStatus = "api_status"
ServiceHealthFailureUIStatus = "ui_status"
ServiceHealthFailureFrontendAssets = "frontend_assets"

183
pkg/server/service_health.go
View file

@ -4,6 +4,7 @@ import (
"context"
"crypto/tls"
"encoding/json"
"errors"
"fmt"
"io"
"net"
@ -18,6 +19,7 @@ import (
const (
serviceHealthProbeTimeout = 5 * time.Second
serviceHealthRetryDelay = time.Second
serviceHealthBodyLimit = 2 << 20
serviceHealthAssetLimit = 32
)
@ -25,21 +27,18 @@ const (
var frontendAssetReferencePattern = regexp.MustCompile(`(?i)(?:src|href)\s*=\s*["'](/assets/[^"'#?]+(?:\?[^"'#]*)?)["']`)
func newServiceHealthProbe(listener net.Listener, tlsEnabled bool) func() telemetry.ServiceHealthObservation {
baseURL, ok := localServiceHealthBaseURL(listener, tlsEnabled)
if !ok {
baseURLs := localServiceHealthBaseURLs(listener, tlsEnabled)
if len(baseURLs) == 0 {
return func() telemetry.ServiceHealthObservation {
return telemetry.ServiceHealthObservation{
Observed: true,
FailureCategory: telemetry.ServiceHealthFailureAPIConnectivity,
}
return unhealthyServiceObservation(telemetry.ServiceHealthFailureAPIConnectivity)
}
}
transport := &http.Transport{}
if tlsEnabled {
// The probe stays inside this process and connects only to the address
// already bound by listener. Certificate trust is a client-facing concern,
// while this probe verifies that Pulse can serve its own HTTPS handler.
// The probe connects only to the address already bound by listener
// (or loopback for a wildcard). Certificate trust is a client-facing
// concern; this verifies Pulse can serve its own HTTPS handler.
transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} // #nosec G402
}
client := &http.Client{
@ -50,66 +49,144 @@ func newServiceHealthProbe(listener net.Listener, tlsEnabled bool) func() teleme
},
}
return serviceHealthProbe(baseURLs, client, serviceHealthProbeTimeout, serviceHealthRetryDelay)
}
// Each observation gets at most two five-second attempts with one second for
// startup to settle between them. Telemetry invokes this in its background
// runner, never on the monitoring or HTTP serving path.
func serviceHealthProbe(baseURLs []string, client *http.Client, timeout, retryDelay time.Duration) func() telemetry.ServiceHealthObservation {
return func() telemetry.ServiceHealthObservation {
ctx, cancel := context.WithTimeout(context.Background(), serviceHealthProbeTimeout)
defer cancel()
apiBody, status, err := serviceHealthGET(ctx, client, baseURL+"/api/health")
if err != nil {
return unhealthyServiceObservation(telemetry.ServiceHealthFailureAPIConnectivity)
}
if status < http.StatusOK || status >= http.StatusMultipleChoices {
return unhealthyServiceObservation(telemetry.ServiceHealthFailureAPIStatus)
}
var health struct {
Status string `json:"status"`
}
if json.Unmarshal(apiBody, &health) != nil || health.Status != "healthy" {
return unhealthyServiceObservation(telemetry.ServiceHealthFailureAPIStatus)
}
indexBody, status, err := serviceHealthGET(ctx, client, baseURL+"/")
if err != nil || status < http.StatusOK || status >= http.StatusMultipleChoices ||
!strings.Contains(strings.ToLower(string(indexBody)), "<html") {
return unhealthyServiceObservation(telemetry.ServiceHealthFailureUIStatus)
}
assetPaths := frontendAssetPaths(indexBody)
if len(assetPaths) == 0 {
return unhealthyServiceObservation(telemetry.ServiceHealthFailureFrontendAssets)
}
for _, assetPath := range assetPaths {
body, assetStatus, assetErr := serviceHealthGET(ctx, client, baseURL+assetPath)
if assetErr != nil || assetStatus < http.StatusOK || assetStatus >= http.StatusMultipleChoices || len(body) == 0 {
return unhealthyServiceObservation(telemetry.ServiceHealthFailureFrontendAssets)
defer client.CloseIdleConnections()
var observation telemetry.ServiceHealthObservation
for attempt := 0; attempt < 2; attempt++ {
if attempt > 0 {
timer := time.NewTimer(retryDelay)
<-timer.C
}
ctx, cancel := context.WithTimeout(context.Background(), timeout)
observation = observeServiceHealth(ctx, client, baseURLs)
cancel()
if observation.Healthy {
return observation
}
}
return telemetry.ServiceHealthObservation{Observed: true, Healthy: true}
return observation
}
}
func localServiceHealthBaseURL(listener net.Listener, tlsEnabled bool) (string, bool) {
func observeServiceHealth(ctx context.Context, client *http.Client, baseURLs []string) telemetry.ServiceHealthObservation {
baseURL, apiBody, status, err := serviceHealthAPI(ctx, client, baseURLs)
if err != nil {
category := telemetry.ServiceHealthFailureAPIConnectivity
if status != 0 {
category = telemetry.ServiceHealthFailureAPIStatus
}
return unhealthyServiceObservation(serviceHealthErrorCategory(err, category))
}
if status < http.StatusOK || status >= http.StatusMultipleChoices {
return unhealthyServiceObservation(telemetry.ServiceHealthFailureAPIStatus)
}
var health struct {
Status string `json:"status"`
}
if json.Unmarshal(apiBody, &health) != nil || health.Status != "healthy" {
return unhealthyServiceObservation(telemetry.ServiceHealthFailureAPIStatus)
}
indexBody, status, err := serviceHealthGET(ctx, client, baseURL+"/")
if err != nil {
return unhealthyServiceObservation(serviceHealthErrorCategory(err, telemetry.ServiceHealthFailureUIStatus))
}
if status < http.StatusOK || status >= http.StatusMultipleChoices ||
!strings.Contains(strings.ToLower(string(indexBody)), "<html") {
return unhealthyServiceObservation(telemetry.ServiceHealthFailureUIStatus)
}
assetPaths := frontendAssetPaths(indexBody)
if len(assetPaths) == 0 {
return unhealthyServiceObservation(telemetry.ServiceHealthFailureFrontendAssets)
}
for _, assetPath := range assetPaths {
body, assetStatus, assetErr := serviceHealthGET(ctx, client, baseURL+assetPath)
if assetErr != nil {
return unhealthyServiceObservation(serviceHealthErrorCategory(assetErr, telemetry.ServiceHealthFailureFrontendAssets))
}
if assetStatus < http.StatusOK || assetStatus >= http.StatusMultipleChoices || len(body) == 0 {
return unhealthyServiceObservation(telemetry.ServiceHealthFailureFrontendAssets)
}
}
return telemetry.ServiceHealthObservation{Observed: true, Healthy: true}
}
// Only a transport failure tries the other loopback family. A response pins
// the entire API/UI/asset observation to that address, even when it is unhealthy.
// Divide the remaining API budget so a stalled family cannot starve the other.
func serviceHealthAPI(ctx context.Context, client *http.Client, baseURLs []string) (string, []byte, int, error) {
var lastErr, timeoutErr error
for i, baseURL := range baseURLs {
deadline, _ := ctx.Deadline()
candidateCtx, cancel := context.WithTimeout(ctx, time.Until(deadline)/time.Duration(len(baseURLs)-i))
body, status, err := serviceHealthGET(candidateCtx, client, baseURL+"/api/health")
cancel()
if err == nil {
return baseURL, body, status, nil
}
if status != 0 {
// A response whose body fails or exceeds the limit is still an
// observation of this server, not an unavailable address family.
return baseURL, nil, status, err
}
lastErr = err
if serviceHealthErrorCategory(err, "") == telemetry.ServiceHealthFailureTimeout {
timeoutErr = err
}
}
if timeoutErr != nil {
return "", nil, 0, timeoutErr
}
return "", nil, 0, lastErr
}
func serviceHealthErrorCategory(err error, fallback string) string {
var netErr net.Error
if errors.Is(err, context.DeadlineExceeded) || (errors.As(err, &netErr) && netErr.Timeout()) {
return telemetry.ServiceHealthFailureTimeout
}
return fallback
}
func localServiceHealthBaseURLs(listener net.Listener, tlsEnabled bool) []string {
if listener == nil {
return "", false
return nil
}
tcpAddr, ok := listener.Addr().(*net.TCPAddr)
if !ok || tcpAddr.Port <= 0 {
return "", false
if !ok || tcpAddr.Port <= 0 || tcpAddr.Port > 65535 {
return nil
}
ip := tcpAddr.IP
if ip == nil || ip.IsUnspecified() {
if ip != nil && ip.To4() == nil {
ip = net.IPv6loopback
} else {
ip = net.IPv4(127, 0, 0, 1)
hosts := []string{tcpAddr.IP.String()}
if tcpAddr.IP == nil || tcpAddr.IP.IsUnspecified() {
hosts = []string{"127.0.0.1"}
// An IPv6 wildcard can be dual-stack or IPv6-only. IPv4-first also
// works when IPv6 loopback is disabled. An IPv4-only wildcard must
// not inspect a different IPv6 listener that happens to share its port.
if tcpAddr.IP == nil || tcpAddr.IP.To4() == nil {
hosts = append(hosts, "::1")
}
} else if tcpAddr.Zone != "" && tcpAddr.IP.To4() == nil {
hosts[0] += "%" + tcpAddr.Zone
}
scheme := "http"
if tlsEnabled {
scheme = "https"
}
return fmt.Sprintf("%s://%s", scheme, net.JoinHostPort(ip.String(), fmt.Sprintf("%d", tcpAddr.Port))), true
baseURLs := make([]string, 0, len(hosts))
for _, host := range hosts {
baseURL := url.URL{Scheme: scheme, Host: net.JoinHostPort(host, fmt.Sprintf("%d", tcpAddr.Port))}
baseURLs = append(baseURLs, baseURL.String())
}
return baseURLs
}
func frontendAssetPaths(index []byte) []string {

310
pkg/server/service_health_regression_test.go Normal file
View file

@ -0,0 +1,310 @@
package server
import (
"context"
"errors"
"fmt"
"io"
"net"
"net/http"
"net/http/httptest"
"reflect"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/telemetry"
)
// Override only Addr, retaining the real listener for serving. This models an
// IPv6 wildcard reported on an install where IPv6 loopback is unavailable.
type serviceHealthAddrListener struct {
net.Listener
addr net.Addr
}
func (l serviceHealthAddrListener) Addr() net.Addr { return l.addr }
type serviceHealthRoundTripper func(*http.Request) (*http.Response, error)
func (f serviceHealthRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
func healthyServiceHealthResponse(r *http.Request) (*http.Response, error) {
body := "app()"
switch r.URL.Path {
case "/api/health":
body = `{"status":"healthy"}`
case "/":
body = `<html><script src="/assets/app.js"></script></html>`
}
return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(body)), Header: make(http.Header), Request: r}, nil
}
func TestServiceHealthWildcardFamiliesAndExplicitAddresses(t *testing.T) {
for _, tc := range []struct {
name string
addr net.Addr
tls bool
want []string
}{
{"IPv6 wildcard", &net.TCPAddr{IP: net.IPv6unspecified, Port: 7655}, false, []string{"http://127.0.0.1:7655", "http://[::1]:7655"}},
{"unspecified family", &net.TCPAddr{Port: 7655}, false, []string{"http://127.0.0.1:7655", "http://[::1]:7655"}},
{"IPv4 wildcard", &net.TCPAddr{IP: net.IPv4zero, Port: 7655}, false, []string{"http://127.0.0.1:7655"}},
{"explicit IPv4", &net.TCPAddr{IP: net.ParseIP("192.0.2.1"), Port: 7655}, false, []string{"http://192.0.2.1:7655"}},
{"explicit IPv6", &net.TCPAddr{IP: net.IPv6loopback, Port: 7655}, true, []string{"https://[::1]:7655"}},
{"explicit scoped IPv6", &net.TCPAddr{IP: net.ParseIP("fe80::1"), Zone: "eth0", Port: 7655}, true, []string{"https://[fe80::1%25eth0]:7655"}},
{"not TCP", &net.UnixAddr{Name: "/tmp/unused", Net: "unix"}, false, nil},
{"no port", &net.TCPAddr{IP: net.IPv4zero}, false, nil},
{"invalid port", &net.TCPAddr{IP: net.IPv4zero, Port: 65536}, false, nil},
} {
t.Run(tc.name, func(t *testing.T) {
got := localServiceHealthBaseURLs(serviceHealthAddrListener{addr: tc.addr}, tc.tls)
if !reflect.DeepEqual(got, tc.want) {
t.Fatalf("targets = %#v, want %#v", got, tc.want)
}
})
}
if got := localServiceHealthBaseURLs(nil, false); got != nil {
t.Fatalf("nil listener targets = %#v", got)
}
}
func TestServiceHealthIPv4SurvivesUnavailableIPv6(t *testing.T) {
for _, failure := range []error{errors.New("IPv6 disabled"), errors.New("IPv6 unreachable"), context.DeadlineExceeded} {
client := &http.Client{Transport: serviceHealthRoundTripper(func(r *http.Request) (*http.Response, error) {
if r.URL.Hostname() == "::1" {
return nil, failure
}
if r.URL.Hostname() != "127.0.0.1" {
t.Fatalf("non-loopback request: %s", r.URL)
}
return healthyServiceHealthResponse(r)
})}
// Red control: the former IPv6-only target cannot observe this healthy install.
old := serviceHealthProbe([]string{"http://[::1]:7655"}, client, time.Second, 0)()
if old.Healthy || !old.Observed {
t.Fatalf("IPv6-only control = %#v", old)
}
targets := localServiceHealthBaseURLs(serviceHealthAddrListener{addr: &net.TCPAddr{IP: net.IPv6unspecified, Port: 7655}}, false)
got := serviceHealthProbe(targets, client, time.Second, 0)()
if !got.Observed || !got.Healthy || got.FailureCategory != "" {
t.Fatalf("wildcard with available IPv4 = %#v", got)
}
}
}
func TestServiceHealthWildcardOnRealIPv4AndIPv6OnlyListeners(t *testing.T) {
for _, network := range []string{"tcp4", "tcp6"} {
t.Run(network, func(t *testing.T) {
address := "127.0.0.1:0"
if network == "tcp6" {
address = "[::1]:0"
}
listener, err := net.Listen(network, address)
if err != nil {
t.Fatalf("required %s loopback listen: %v", network, err)
}
server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
response, _ := healthyServiceHealthResponse(r)
defer response.Body.Close()
_, _ = io.Copy(w, response.Body)
}), ReadHeaderTimeout: time.Second}
done := make(chan struct{})
go func() { defer close(done); _ = server.Serve(listener) }()
t.Cleanup(func() { _ = server.Close(); <-done })
bound := listener.Addr().(*net.TCPAddr)
wildcard := serviceHealthAddrListener{Listener: listener, addr: &net.TCPAddr{IP: net.IPv6unspecified, Port: bound.Port}}
got := newServiceHealthProbe(wildcard, false)()
if !got.Observed || !got.Healthy || got.FailureCategory != "" {
t.Fatalf("real %s server through wildcard = %#v", network, got)
}
})
}
}
func TestServiceHealthStalledFamilyCannotStarveIPv6(t *testing.T) {
var hosts []string
client := &http.Client{Transport: serviceHealthRoundTripper(func(r *http.Request) (*http.Response, error) {
hosts = append(hosts, r.URL.Hostname())
if r.URL.Hostname() == "127.0.0.1" {
<-r.Context().Done()
return nil, r.Context().Err()
}
if r.Context().Err() != nil {
t.Fatalf("IPv6 has no reserved budget: %v", r.Context().Err())
}
return healthyServiceHealthResponse(r)
})}
got := serviceHealthProbe([]string{"http://127.0.0.1:7655", "http://[::1]:7655"}, client, 200*time.Millisecond, 0)()
if !got.Healthy || !reflect.DeepEqual(hosts, []string{"127.0.0.1", "::1", "::1", "::1"}) {
t.Fatalf("observation = %#v, requested hosts = %#v", got, hosts)
}
}
func TestServiceHealthSlowStartupRetriesOnceAndRecovers(t *testing.T) {
var apiCalls int
var firstDeadline time.Time
client := &http.Client{Transport: serviceHealthRoundTripper(func(r *http.Request) (*http.Response, error) {
if r.URL.Path == "/api/health" {
apiCalls++
deadline, ok := r.Context().Deadline()
if !ok {
t.Fatal("unbounded probe")
}
if apiCalls == 1 {
firstDeadline = deadline
<-r.Context().Done()
return nil, r.Context().Err()
}
if !deadline.After(firstDeadline) || r.Context().Err() != nil {
t.Fatal("retry reused exhausted startup budget")
}
}
return healthyServiceHealthResponse(r)
})}
got := serviceHealthProbe([]string{"http://127.0.0.1:7655"}, client, 50*time.Millisecond, time.Millisecond)()
if !got.Healthy || apiCalls != 2 {
t.Fatalf("observation = %#v, API calls = %d", got, apiCalls)
}
}
func TestServiceHealthFinalTimeoutAtEveryStageIsBounded(t *testing.T) {
for _, stage := range []string{"/api/health", "/", "/assets/app.js"} {
t.Run(stage, func(t *testing.T) {
apiCalls, timeouts := 0, 0
client := &http.Client{Transport: serviceHealthRoundTripper(func(r *http.Request) (*http.Response, error) {
if r.URL.Path == "/api/health" {
apiCalls++
}
if r.URL.Path == stage {
timeouts++
if _, ok := r.Context().Deadline(); !ok {
t.Fatal("unbounded request")
}
<-r.Context().Done()
return nil, r.Context().Err()
}
return healthyServiceHealthResponse(r)
})}
got := serviceHealthProbe([]string{"http://127.0.0.1:7655"}, client, 20*time.Millisecond, 0)()
if !got.Observed || got.Healthy || got.FailureCategory != telemetry.ServiceHealthFailureTimeout || apiCalls != 2 || timeouts != 2 {
t.Fatalf("observation = %#v, API calls/timeouts = %d/%d", got, apiCalls, timeouts)
}
})
}
}
func TestServiceHealthDoesNotHideGenuineFailuresOnAnotherFamily(t *testing.T) {
for _, tc := range []struct {
name, path, body, want string
status int
err error
}{
{name: "connectivity", path: "/api/health", err: errors.New("refused"), want: telemetry.ServiceHealthFailureAPIConnectivity},
{name: "API status", path: "/api/health", status: 503, body: `{"status":"unhealthy"}`, want: telemetry.ServiceHealthFailureAPIStatus},
{name: "API invalid JSON", path: "/api/health", status: 200, body: "invalid", want: telemetry.ServiceHealthFailureAPIStatus},
{name: "API unhealthy body", path: "/api/health", status: 200, body: `{"status":"unhealthy"}`, want: telemetry.ServiceHealthFailureAPIStatus},
{name: "API oversized body", path: "/api/health", status: 200, body: strings.Repeat("x", serviceHealthBodyLimit+1), want: telemetry.ServiceHealthFailureAPIStatus},
{name: "UI status", path: "/", status: 404, want: telemetry.ServiceHealthFailureUIStatus},
{name: "UI invalid HTML", path: "/", status: 200, body: "not HTML", want: telemetry.ServiceHealthFailureUIStatus},
{name: "UI oversized body", path: "/", status: 200, body: strings.Repeat("x", serviceHealthBodyLimit+1), want: telemetry.ServiceHealthFailureUIStatus},
{name: "missing asset references", path: "/", status: 200, body: "<html></html>", want: telemetry.ServiceHealthFailureFrontendAssets},
{name: "asset status", path: "/assets/app.js", status: 404, want: telemetry.ServiceHealthFailureFrontendAssets},
{name: "asset empty", path: "/assets/app.js", status: 200, want: telemetry.ServiceHealthFailureFrontendAssets},
{name: "asset oversized", path: "/assets/app.js", status: 200, body: strings.Repeat("x", serviceHealthBodyLimit+1), want: telemetry.ServiceHealthFailureFrontendAssets},
} {
t.Run(tc.name, func(t *testing.T) {
apiCalls, ipv6Calls := 0, 0
client := &http.Client{Transport: serviceHealthRoundTripper(func(r *http.Request) (*http.Response, error) {
if r.URL.Path == "/api/health" {
apiCalls++
}
if r.URL.Hostname() == "::1" {
ipv6Calls++
if tc.err == nil {
t.Fatal("response failure switched listener family")
}
}
if r.URL.Path == tc.path {
if tc.err != nil {
return nil, tc.err
}
return &http.Response{StatusCode: tc.status, Body: io.NopCloser(strings.NewReader(tc.body)), Header: make(http.Header)}, nil
}
return healthyServiceHealthResponse(r)
})}
got := serviceHealthProbe([]string{"http://127.0.0.1:7655", "http://[::1]:7655"}, client, time.Second, 0)()
wantAPICalls := 2
if tc.err != nil {
wantAPICalls = 4
}
if !got.Observed || got.Healthy || got.FailureCategory != tc.want || apiCalls != wantAPICalls {
t.Fatalf("observation = %#v, API calls = %d, IPv6 calls = %d", got, apiCalls, ipv6Calls)
}
})
}
}
func TestServiceHealthHTTPSAndRedirectRefusal(t *testing.T) {
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
response, _ := healthyServiceHealthResponse(r)
defer response.Body.Close()
_, _ = io.Copy(w, response.Body)
}))
t.Cleanup(server.Close)
if got := newServiceHealthProbe(server.Listener, true)(); !got.Healthy {
t.Fatalf("HTTPS service = %#v", got)
}
var redirected atomic.Int32
external := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { redirected.Add(1) }))
t.Cleanup(external.Close)
for _, stage := range []string{"/api/health", "/", "/assets/app.js"} {
t.Run(stage, func(t *testing.T) {
local := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == stage {
http.Redirect(w, r, external.URL, http.StatusFound)
return
}
response, _ := healthyServiceHealthResponse(r)
defer response.Body.Close()
_, _ = io.Copy(w, response.Body)
}))
t.Cleanup(local.Close)
got := newServiceHealthProbe(local.Listener, false)()
if got.Healthy || redirected.Load() != 0 {
t.Fatalf("redirect followed/accepted: %#v, external requests = %d", got, redirected.Load())
}
})
}
}
func TestServiceHealthConcurrentObservations(t *testing.T) {
probe := serviceHealthProbe([]string{"http://127.0.0.1:7655"}, &http.Client{Transport: serviceHealthRoundTripper(healthyServiceHealthResponse)}, time.Second, 0)
var wg sync.WaitGroup
for i := 0; i < 8; i++ {
wg.Add(1)
go func() {
defer wg.Done()
if got := probe(); !got.Healthy {
t.Errorf("concurrent observation = %#v", got)
}
}()
}
wg.Wait()
}
func TestFrontendAssetPathsEnforceExistingBounds(t *testing.T) {
var index strings.Builder
for i := 0; i < serviceHealthAssetLimit+10; i++ {
fmt.Fprintf(&index, `<script src="/assets/%d.js"></script>`, i)
}
if paths := frontendAssetPaths([]byte(index.String())); len(paths) != serviceHealthAssetLimit {
t.Fatalf("asset count = %d, want %d", len(paths), serviceHealthAssetLimit)
}
if paths := frontendAssetPaths([]byte(`<script src="/assets/a.js"></script><script src="/assets/a.js"></script><script src="https://outside.example/x.js"></script><script src="//outside.example/x.js"></script>`)); !reflect.DeepEqual(paths, []string{"/assets/a.js"}) {
t.Fatalf("duplicate/external paths escaped: %#v", paths)
}
}

12
pkg/server/service_health_test.go
View file

@ -105,9 +105,15 @@ func TestServiceHealthProbeCoversAPIUIAndFrontendAssets(t *testing.T) {
}
func TestServiceHealthProbeReportsConnectivityWithoutListener(t *testing.T) {
got := newServiceHealthProbe(nil, false)()
if !got.Observed || got.Healthy || got.FailureCategory != telemetry.ServiceHealthFailureAPIConnectivity {
t.Fatalf("service-health observation = %#v", got)
for _, listener := range []net.Listener{
nil,
serviceHealthAddrListener{addr: &net.UnixAddr{Name: "/tmp/unused", Net: "unix"}},
serviceHealthAddrListener{addr: &net.TCPAddr{IP: net.IPv4zero}},
} {
got := newServiceHealthProbe(listener, false)()
if !got.Observed || got.Healthy || got.FailureCategory != telemetry.ServiceHealthFailureAPIConnectivity {
t.Fatalf("invalid listener service-health observation = %#v", got)
}
}
}

21
scripts/tests/test_telemetry_adoption_report.py
View file

@ -595,6 +595,27 @@ class TelemetryAdoptionReportTest(unittest.TestCase):
{"healthy": 1, "frontend_assets": 1},
)
def test_target_release_service_health_keeps_timeouts_distinct_from_connectivity(self) -> None:
now = datetime(2026, 9, 30, 20, tzinfo=timezone.utc)
rows = {
category: {
"received_at": "2026-09-30 19:00:00",
"version": "6.4.6",
"service_health_observed": 1,
"service_health_healthy": 0,
"service_health_failure_category": category,
}
for category in ("timeout", "api_connectivity")
}
summary = report.summarize_target_release_service_health(
rows, {"6.4.6"}, "6.4.6", now=now
)
self.assertEqual(summary["unhealthy_installs"], 2)
self.assertEqual(
{entry["category"]: entry["installs"] for entry in summary["failure_categories"]},
{"timeout": 1, "api_connectivity": 1},
)
def test_target_release_followup_excludes_first_heartbeat_baselines_and_flags_rollbacks(self) -> None:
now = datetime(2026, 8, 19, 12, tzinfo=timezone.utc)