Skip to content

symfonic.diagnostics.checks.check_postgres_dsn

check_postgres_dsn

POSTGRES_DSN reachability probe (v7.16, item 5).

Network-sensitive check that opens a TCP connection to the host:port in POSTGRES_DSN and (when possible) issues a startup-handshake to verify credentials.

Budget (architect verdict):

  • TCP connect: 2 seconds
  • Auth handshake: 1 second

The check must auto-skip when CI=true (or when the runner is invoked with --no-network). CI runners frequently lack outbound network to adopter Postgres clusters, and a 3-second probe per CI job is paper-cut latency we don't want to charge. The skip path is handled by the runner (which emits an INFO directly), so this module runs only when network checks are enabled.

Output severity is WARN (not ERROR) because the adopter may be running a development workflow with no Postgres bound -- we want to surface the issue without breaking symfonic doctor in environments where the DSN is intentionally unset.

check_postgres_dsn async 

check_postgres_dsn(
    agent: SymfonicAgent,
) -> list[CheckResult]

Probe POSTGRES_DSN for TCP reachability and basic auth.

Source code in src/symfonic/diagnostics/checks/check_postgres_dsn.py 
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
async def check_postgres_dsn(
    agent: SymfonicAgent,  # noqa: ARG001 -- uniform signature
) -> list[CheckResult]:
    """Probe ``POSTGRES_DSN`` for TCP reachability and basic auth."""
    raw = (os.environ.get("POSTGRES_DSN") or "").strip()
    if not raw:
        # No DSN configured -- nothing to probe.  Stay silent so the
        # adopter sees the embedder / layer checks rather than DSN
        # noise.
        return []

    host, port = _parse_host_port(raw)
    if host is None or port is None:
        return [
            CheckResult(
                name="postgres_dsn.malformed",
                severity=Severity.WARN,
                message=(
                    f"POSTGRES_DSN={raw!r} could not be parsed as a "
                    "host:port pair."
                ),
                fix_hint=(
                    "Expected shape: postgres://user:password@host:port/dbname"
                ),
            ),
        ]

    print(
        f"  probing POSTGRES_DSN={host}:{port} ...",
        file=sys.stderr,
        flush=True,
    )
    tcp_ok, tcp_err = await _probe_tcp(host, port)
    if not tcp_ok:
        return [
            CheckResult(
                name="postgres_dsn.unreachable",
                severity=Severity.WARN,
                message=(
                    f"POSTGRES_DSN host {host}:{port} is unreachable "
                    f"within {_CONNECT_TIMEOUT_SECONDS}s ({tcp_err})."
                ),
                fix_hint=(
                    "Verify the host is running and reachable from this "
                    "process.  Set --no-network (or CI=true) to skip this "
                    "probe in environments without outbound DB access."
                ),
            ),
        ]

    # TCP succeeded.  Auth-probe is best-effort: if asyncpg is not
    # installed we report TCP-only success at INFO; if asyncpg is
    # available, we try a connect-and-close within _AUTH_TIMEOUT_SECONDS.
    auth_result = await _probe_auth(raw)
    if auth_result is not None:
        return [auth_result]

    return []