turbostats: the database section, for dbhealth

CHANGELOG.md

@@ -4,6 +4,13 @@
 
 ### Added
 
+- **TurboStats `database` section.** The section a `dbhealth` reporter
+  sends: the probe (serving, and how fast), resources (connections against
+  the maximum, size on disk, the oldest open transaction), per-table
+  freshness and row counts, and replication, for one database endpoint.
+  Facts only; the receiver judges. `wire.Database`, optional on the bundle
+  beside `Pipeline` and `Serve`. A failed probe marshals `probe` alone. The
+  spec is `docs/superpowers/specs/2026-10-05-turbostats-database-section-design.md`. (#441)
 - **Webhook source `ack: after_flush`.** A 200 means the body reached the
   sink: the answer waits for the batch holding it to flush and commit, and a
   failed flush answers 503. The default, `on_receive`, is unchanged. Refused

internal/turbostats/dimensional_test.go

@@ -403,6 +403,16 @@ func TestWire_NoFieldScalesWithCardinality(t *testing.T) {
 				if at == ".Bundle.Freshness.Tables" || at == ".Bundle.Rollup.Rollups" {
 					continue
 				}
+				// A database bundle's tables are the operator's list, capped
+				// by wire.MaxDatabaseTables; its replicas are the cluster's
+				// topology, capped by wire.MaxDatabaseReplicas; its errors
+				// are at most one per table and one per catalog query,
+				// wire.MaxDatabaseErrors. None grows with the data in the
+				// database. TestCollect_AFullDatabaseBundleStaysUnderTheCeiling
+				// prices all three at their caps.
+				if at == ".Bundle.Database.Tables" || at == ".Bundle.Database.Replication.Replicas" || at == ".Bundle.Database.Collection.Errors" {
+					continue
+				}
 				// A duration's buckets are a fixed-length array whose length
 				// the contract sets: len(wire.DurationBounds)+1, the same
 				// for every process forever. It is the one shape a receiver
@@ -565,13 +575,85 @@ func widestBundle(t *testing.T) wire.Bundle {
 			},
 			Duration: &wire.ServeDurations{Request: widestDuration()},
 		},
+		// One of each repeated field: enough for the schema to validate
+		// the shape. The full width is priced by
+		// TestCollect_AFullDatabaseBundleStaysUnderTheCeiling.
+		Database: widestDatabase(1, 1, 1),
 		Exit: &wire.Exit{
 			Reason: "system.internal.unexpected",
 			Code:   255,
 		},
 	}
 }
 
+// widestDatabase is a database section with every field at its widest
+// value: the longest identifiers Postgres allows, the longest host name DNS
+// allows, and 2^62 in every count, with the given number of tables,
+// replicas and errors.
+func widestDatabase(tables, replicas, errors int) *wire.Database {
+	big := int64(1) << 62
+	n := 1 << 30
+	at := time.Now().UTC()
+	secs := 1e9
+	ident := strings.Repeat("i", 63)
+	table := strings.Repeat("s", 63) + "." + strings.Repeat("t", 63)
+	host := strings.Repeat("h", 253)
+	d := &wire.Database{
+		Kind:          "postgres",
+		Target:        host + ":65535/" + ident,
+		Cluster:       strings.Repeat("c", 64),
+		ServerVersion: "PostgreSQL 18.1 (Debian 18.1-1.pgdg13+1) on aarch64-unknown-linux-gnu, compiled by gcc (Debian 14.2.0-19) 14.2.0, 64-bit",
+		Probe:         wire.DatabaseProbe{OK: true, LatencyMs: big, LastOKAt: &at, ConsecutiveFailures: n, Error: "timeout"},
+		Resources: &wire.DatabaseResources{
+			Connections:              &wire.DatabaseConnections{Used: n, Max: n, Waiting: n},
+			SizeBytes:                &big,
+			OldestTransactionSeconds: &big,
+			Memory:                   &wire.DatabaseMemory{SharedBuffersBytes: big},
+		},
+		Replication: &wire.DatabaseReplication{Role: "primary", LagSeconds: &secs, LastReplayedAt: &at, Upstream: host + ":65535"},
+		Collection:  wire.DatabaseCollection{Queries: n, DurationMs: big},
+	}
+	for i := 0; i < tables; i++ {
+		d.Tables = append(d.Tables, wire.DatabaseTable{
+			Name: table, FreshnessColumn: ident, NewestAt: &at, Rows: &big, RowsExact: true,
+			SizeBytes: big, LastVacuumAt: &at, CheckedAt: at,
+		})
+	}
+	for i := 0; i < replicas; i++ {
+		d.Replication.Replicas = append(d.Replication.Replicas, wire.DatabaseReplica{Name: ident, LagSeconds: &secs, State: "streaming"})
+	}
+	for i := 0; i < errors; i++ {
+		d.Collection.Errors = append(d.Collection.Errors, wire.DatabaseError{Table: table, Error: strings.Repeat("e", 200)})
+	}
+	return d
+}
+
+// The widest database bundle the contract allows stays under 48 KiB:
+// wire.MaxDatabaseTables tables, each with the longest schema and table
+// name Postgres allows, wire.MaxDatabaseReplicas replicas, an error for
+// every table and every catalog query, and every count at 2^62. It
+// measured 47725 bytes on 2026-10-05. The 16 KiB limit that holds every
+// pipeline bundle cannot hold it: a receiver that accepts database bundles
+// allows 64 KiB, so 48 leaves the alarm a margin without letting the shape
+// double quietly.
+func TestCollect_AFullDatabaseBundleStaysUnderTheCeiling(t *testing.T) {
+	coverage.Covers(t, "observability.turbostats")
+	at := time.Now().UTC()
+	b := wire.Bundle{
+		V: wire.Version, SentAt: at, IntervalSeconds: 86400, LastActivityAt: &at,
+		Instance: wire.Instance{
+			ID: strings.Repeat("i", 64), Name: strings.Repeat("n", 64), Version: "v2026.10.05", Commit: strings.Repeat("c", 40),
+			Arch: "linux/arm64", ConfigHash: "sha256:" + strings.Repeat("f", 64), Labels: widestLabels(t),
+		},
+		Process:  wire.Process{StartedAt: at, RSSBytes: 1 << 62, Goroutines: 1 << 30},
+		Database: widestDatabase(wire.MaxDatabaseTables, wire.MaxDatabaseReplicas, wire.MaxDatabaseErrors),
+	}
+	raw, err := json.Marshal(b)
+	assert.NoError(t, err)
+	t.Logf("the widest database bundle is %d bytes", len(raw))
+	assert.That(t, len(raw) < 48<<10)
+}
+
 // widestLabels is the largest label set the config accepts: the most keys,
 // each at its length limit, each with a value at its length limit.
 //

turbostats/wire/bundle.go

@@ -59,6 +59,9 @@ type Bundle struct {
 	Pipeline *Pipeline `json:"pipeline,omitempty"`
 	// Serve is present when the process answers dataset requests.
 	Serve *Serve `json:"serve,omitempty"`
+	// Database is present when the process watches a database endpoint, the
+	// dbhealth reporter. A bundle carries it and no Pipeline.
+	Database *Database `json:"database,omitempty"`
 	// Freshness is present when the process measures a store's tables: the
 	// rollup daemon's leader.
 	Freshness *Freshness `json:"freshness,omitempty"`

turbostats/wire/database.go

@@ -0,0 +1,128 @@
+package wire
+
+import "time"
+
+// MaxDatabaseTables bounds Database.Tables. A reporter watches at most this
+// many tables per endpoint; its config refuses more. It is the one number
+// that sets a database bundle's width, so it is in the contract: the widest
+// bundle of MaxDatabaseTables tables with the longest names Postgres allows
+// is priced by TestCollect_AFullDatabaseBundleStaysUnderTheCeiling, and the
+// receiver's body limit has to hold it.
+const MaxDatabaseTables = 50
+
+// MaxDatabaseReplicas bounds DatabaseReplication.Replicas. A primary with
+// more sends the first MaxDatabaseReplicas by name.
+const MaxDatabaseReplicas = 16
+
+// MaxDatabaseErrors bounds DatabaseCollection.Errors: one per watched table
+// and one per catalog query, which is fewer than ten.
+const MaxDatabaseErrors = MaxDatabaseTables + 10
+
+// Database is the section a dbhealth reporter sends: one database endpoint,
+// probed and measured. Facts only: timestamps, counts and limits. The
+// receiver judges them. Fields a kind cannot provide are absent, not zero.
+//
+// Spec: docs/superpowers/specs/2026-10-05-turbostats-database-section-design.md.
+type Database struct {
+	// Kind is postgres, mysql, mongo, snowflake or redshift.
+	Kind string `json:"kind"`
+	// Target is host:port/database. Never a DSN: it carries no credential.
+	Target string `json:"target"`
+	// Cluster groups a primary with its replicas. Absent when not configured.
+	Cluster       string `json:"cluster,omitempty"`
+	ServerVersion string `json:"server_version,omitempty"`
+
+	// Probe is always present. When it failed, nothing below it is.
+	Probe DatabaseProbe `json:"probe"`
+
+	Resources   *DatabaseResources   `json:"resources,omitempty"`
+	Tables      []DatabaseTable      `json:"tables,omitempty"`
+	Replication *DatabaseReplication `json:"replication,omitempty"`
+	Collection  DatabaseCollection   `json:"collection"`
+}
+
+// DatabaseProbe is one round trip per interval: is it serving, and how fast.
+type DatabaseProbe struct {
+	OK        bool       `json:"ok"`
+	LatencyMs int64      `json:"latency_ms"`
+	LastOKAt  *time.Time `json:"last_ok_at,omitempty"`
+	// ConsecutiveFailures counts probes failed in a row, 0 when OK.
+	ConsecutiveFailures int `json:"consecutive_failures"`
+	// Error is the class of the latest failure: refused, timeout, auth or
+	// other. Empty when OK.
+	Error string `json:"error,omitempty"`
+}
+
+// DatabaseResources is how close the database is to its limits. Each field
+// is absent when the kind, or the role, cannot read it.
+type DatabaseResources struct {
+	Connections *DatabaseConnections `json:"connections,omitempty"`
+	// SizeBytes is this database on disk.
+	SizeBytes *int64 `json:"size_bytes,omitempty"`
+	// OldestTransactionSeconds is the age of the oldest open transaction;
+	// one idle in transaction holds vacuum back.
+	OldestTransactionSeconds *int64          `json:"oldest_transaction_seconds,omitempty"`
+	Memory                   *DatabaseMemory `json:"memory,omitempty"`
+}
+
+// DatabaseConnections is connections in use against the configured maximum.
+type DatabaseConnections struct {
+	Used    int `json:"used"`
+	Max     int `json:"max"`
+	Waiting int `json:"waiting"`
+}
+
+// DatabaseMemory carries settings, not usage: a database does not know the
+// host's memory from SQL.
+type DatabaseMemory struct {
+	SharedBuffersBytes int64 `json:"shared_buffers_bytes"`
+}
+
+// DatabaseTable is one watched table's freshness, size and row count.
+type DatabaseTable struct {
+	Name            string `json:"name"`
+	FreshnessColumn string `json:"freshness_column,omitempty"`
+	// NewestAt is max(FreshnessColumn). Absent when the table has none.
+	NewestAt *time.Time `json:"newest_at,omitempty"`
+	// Rows is an estimate unless RowsExact, which says count(*) ran.
+	Rows         *int64     `json:"rows,omitempty"`
+	RowsExact    bool       `json:"rows_exact"`
+	SizeBytes    int64      `json:"size_bytes"`
+	LastVacuumAt *time.Time `json:"last_vacuum_at,omitempty"`
+	CheckedAt    time.Time  `json:"checked_at"`
+}
+
+// DatabaseReplication is measured from this endpoint's side. A replica
+// reports how far behind what it serves is; a primary reports what it knows
+// about each replica.
+type DatabaseReplication struct {
+	// Role is primary or replica.
+	Role           string     `json:"role"`
+	LagSeconds     *float64   `json:"lag_seconds,omitempty"`
+	LastReplayedAt *time.Time `json:"last_replayed_at,omitempty"`
+	Upstream       string     `json:"upstream,omitempty"`
+	// Replicas is a primary's view of each replica.
+	Replicas []DatabaseReplica `json:"replicas,omitempty"`
+}
+
+// DatabaseReplica is one replica as its primary sees it.
+type DatabaseReplica struct {
+	Name       string   `json:"name"`
+	LagSeconds *float64 `json:"lag_seconds,omitempty"`
+	State      string   `json:"state,omitempty"`
+}
+
+// DatabaseCollection is what the interval cost, so the receiver can show
+// the price of watching, and every per-table failure, so one table the
+// role cannot read does not hide the others.
+type DatabaseCollection struct {
+	Queries    int             `json:"queries"`
+	DurationMs int64           `json:"duration_ms"`
+	Errors     []DatabaseError `json:"errors"`
+}
+
+// DatabaseError is one query that failed this interval.
+type DatabaseError struct {
+	Table string `json:"table,omitempty"`
+	Error string `json:"error"`
+}

turbostats/wire/database_test.go

@@ -0,0 +1,72 @@
+package wire
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/turbolytics/sql-flow/internal/coverage"
+	"github.com/zeebo/assert"
+)
+
+// The spec's example, byte for byte after a round trip: every field the
+// section defines is here, and nothing a kind cannot provide is forced to
+// zero.
+func TestDatabase_SpecExampleRoundTrips(t *testing.T) {
+	coverage.Covers(t, "observability.turbostats")
+	at := time.Date(2026, 10, 5, 11, 59, 59, 0, time.UTC)
+	d := Database{
+		Kind: "postgres", Target: "pg-2.internal:5432/billing", Cluster: "billing", ServerVersion: "18.0",
+		Probe: DatabaseProbe{OK: true, LatencyMs: 3, LastOKAt: &at},
+		Resources: &DatabaseResources{
+			Connections:              &DatabaseConnections{Used: 18, Max: 100, Waiting: 0},
+			SizeBytes:                dbPtr(int64(14104255)),
+			OldestTransactionSeconds: dbPtr(int64(2)),
+			Memory:                   &DatabaseMemory{SharedBuffersBytes: 134217728},
+		},
+		Tables: []DatabaseTable{{
+			Name: "public.usage_per_minute", FreshnessColumn: "minute", NewestAt: &at,
+			Rows: dbPtr(int64(7832)), RowsExact: false, SizeBytes: 1826816, LastVacuumAt: &at, CheckedAt: at,
+		}},
+		Replication: &DatabaseReplication{Role: "replica", LagSeconds: dbPtr(0.4), LastReplayedAt: &at, Upstream: "pg-1.internal:5432"},
+		Collection:  DatabaseCollection{Queries: 9, DurationMs: 41, Errors: []DatabaseError{}},
+	}
+	b, err := json.Marshal(d)
+	assert.NoError(t, err)
+	var back Database
+	assert.NoError(t, json.Unmarshal(b, &back))
+	assert.Equal(t, d, back)
+	// Field names are the spec's.
+	for _, want := range []string{`"kind":"postgres"`, `"latency_ms":3`, `"rows_exact":false`, `"lag_seconds":0.4`, `"duration_ms":41`, `"size_bytes":14104255`} {
+		assert.That(t, strings.Contains(string(b), want))
+	}
+}
+
+// A failed probe sends probe alone: nothing else could run.
+func TestDatabase_FailedProbeCarriesProbeAlone(t *testing.T) {
+	coverage.Covers(t, "observability.turbostats")
+	d := Database{Kind: "postgres", Target: "pg:5432/x", Probe: DatabaseProbe{OK: false, ConsecutiveFailures: 3, Error: "refused"}}
+	b, err := json.Marshal(d)
+	assert.NoError(t, err)
+	for _, absent := range []string{`"resources"`, `"tables"`, `"replication"`} {
+		assert.That(t, !strings.Contains(string(b), absent))
+	}
+	assert.That(t, strings.Contains(string(b), `"error":"refused"`))
+	assert.That(t, strings.Contains(string(b), `"consecutive_failures":3`))
+}
+
+// A bundle with a database section and no pipeline is the shape dbhealth
+// sends; the field sits beside Pipeline and Serve, omitted when nil.
+func TestBundle_DatabaseSectionIsOptional(t *testing.T) {
+	coverage.Covers(t, "observability.turbostats")
+	b, err := json.Marshal(Bundle{V: 1, Database: &Database{Kind: "postgres", Target: "pg:5432/x", Probe: DatabaseProbe{OK: true}}})
+	assert.NoError(t, err)
+	assert.That(t, strings.Contains(string(b), `"database":{`))
+	assert.That(t, !strings.Contains(string(b), `"pipeline"`))
+	b, err = json.Marshal(Bundle{V: 1})
+	assert.NoError(t, err)
+	assert.That(t, !strings.Contains(string(b), `"database"`))
+}
+
+func dbPtr[T any](v T) *T { return &v }