diff --git a/.env.example b/.env.example index 9bd6f18..059ac69 100644 --- a/.env.example +++ b/.env.example @@ -27,18 +27,36 @@ MACHINE_CPUS=4 # online (xfs_growfs). Shrinking is unsupported. Default shown. # PG_DATA_DISK_SIZE=140G -# Optional: host loopback ports the forwarder listens on — what local clients -# connect to. Offset from 5432 so a local PostgreSQL isn't shadowed. Defaults -# shown. PG_BACKEND_PORT is the port each backend is exposed on (machine eth0). +# Optional: the client-facing loopback ports — what local clients connect to +# (psql/.pgpass/status print these). Offset from 5432 so a local PostgreSQL isn't +# shadowed. The socat client proxy (`make proxy.install`) binds THESE. +# PG_BACKEND_PORT is the port each backend is exposed on (machine eth0). # PG_CLIENT_ACTIVE_PORT=5442 # PG_CLIENT_STAGING_PORT=5443 # PG_BACKEND_PORT=5432 -# Optional: address the host forwarder binds its client listeners to. Default -# 127.0.0.1 (loopback only). Set 0.0.0.0 ONLY if a sibling container/k3d can't -# reach the Mac's loopback — it exposes the dev-credentialed backend on every -# interface (LAN/Wi-Fi), so widen deliberately. -# PG_FORWARD_BIND=127.0.0.1 +# Optional: verbose (debug-level) slog output for the SQLite tracking DB and the +# socat proxy. ON by default (promote/reconcile are chatty on purpose); set 0 to +# quiet it to info-level. +# PG_PROXY_DEBUG=1 + +# Optional: ANSI color for that structured output (rendered by lmittmann/tint). +# "auto" (default) colors it only when stderr is a terminal — and honors +# NO_COLOR / TERM=dumb — so piping or redirecting a run yields clean, +# fully-dated lines. "always" forces color through a pipe (e.g. `|& less -R`), +# "never" disables it outright. +# PG_LOG_COLOR=auto + +# Optional: structured-log format. "text" (default) is the colored, human-first +# rendering above; "json" swaps in the stdlib slog JSON handler for a run that +# gets captured and machine-parsed rather than watched. +# PG_LOG_FORMAT=text + +# Optional: address the client proxy binds its listeners to. Default 127.0.0.1 +# (loopback only). Set 0.0.0.0 ONLY if a sibling container/k3d can't reach the +# Mac's loopback — it exposes the dev-credentialed backend on every interface +# (LAN/Wi-Fi), so widen deliberately. +# PG_CLIENT_BIND=127.0.0.1 # Optional: minimum free GiB on the macOS volume before `make pg.up` / restore / # import will run. The VM disks are sparse images that grow and never shrink, so diff --git a/.gitignore b/.gitignore index 9d73ff3..0697ad4 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,8 @@ /.zed .env +# Leftover local code-signing material from the removed forwarder/codesign flow. +# Nothing builds or reads it any more — kept ignored so a stray private key can +# never be committed; delete etc/keys yourself when you're done with it. /etc/keys/* # Everything under var/ is machine-local runtime or generated state — nothing diff --git a/Makefile b/Makefile index 48126e7..1d6280b 100644 --- a/Makefile +++ b/Makefile @@ -142,7 +142,10 @@ deploy: machine pgdevd # Cheap guard for targets that exec into already-running machines: fail fast # with a clear message instead of a raw Apple CLI 'notFound' error when a -# machine has never been created. +# machine has never been created. Deliberately NOT used by the read-only +# status targets: since `pg.staging.purge` a deleted machine is an expected +# steady state, so status must still report (it prints ABSENT for that slot) +# instead of refusing to run. .PHONY: machine.exists machine.exists: @for slot in a b; do \ @@ -227,11 +230,15 @@ start: deploy $(MAKE) status .PHONY: status/incus -status/incus: machine.exists +status/incus: @for slot in a b; do \ name="$(MACHINE_PREFIX)-$$slot"; \ echo "── $$name ──"; \ - container machine run --name "$$name" --root -- incus list 2>/dev/null || echo " (Incus not up)"; \ + if container machine inspect "$$name" >/dev/null 2>&1; then \ + container machine run --name "$$name" --root -- incus list 2>/dev/null || echo " (Incus not up)"; \ + else \ + echo " (machine does not exist)"; \ + fi; \ done # The one status command: active/staging machine roles, per-machine @@ -239,41 +246,45 @@ status/incus: machine.exists # (one per machine) over the HTTP API; the active/staging split is a host-side # pointer (var/active-machine), not part of the daemon contract. .PHONY: status -status: machine.exists pgdevd +status: pgdevd @$(PGDEV) status -# ----- stable macOS client endpoints -------------------------------------- -# Each Apple machine's IP drifts and cannot be pinned, so a host-side in-process -# Go forwarder (internal/forward, run by a per-user LaunchAgent) publishes +# ----- stable macOS client endpoints (doc/issues/0004) --------------------- +# Each Apple machine's IP drifts and cannot be pinned, so clients never talk to +# a machine IP: a socat-based client proxy under per-user LaunchAgents publishes # permanent 127.0.0.1:5442 (active) / :5443 (staging) endpoints and relays each -# to whichever machine currently holds that role (on its own eth0:5432). It owns -# the listeners for their whole lifetime and re-points itself from the pointer -# file — `pgdev promote` is just a pointer write. `start` (via `pgdev refresh`) -# validates the LaunchAgent every run and self-heals a missing, unloaded, stale -# (e.g. a plist left pointing at a deleted binary after the repo moved), or -# crashed agent (PG_ENDPOINT_AUTOINSTALL=0 opts out). - -.PHONY: endpoint.install -endpoint.install: machine.exists - @$(PGDEV) forward install - -# Restart the running forwarder so a macOS Local Network permission granted -# AFTER it started actually takes effect. TCC caches its allow/deny decision at -# process start, so ticking the Local Network box (System Settings → Privacy & -# Security → Local Network) does nothing for an already-running agent — it keeps -# failing to reach the VM subnet with EHOSTUNREACH ("server closed the -# connection unexpectedly") until it is restarted. Run this once after granting. -.PHONY: endpoint.restart -endpoint.restart: - @$(PGDEV) forward restart - -.PHONY: endpoint.uninstall -endpoint.uninstall: - @$(PGDEV) forward uninstall - -.PHONY: endpoint.status -endpoint.status: - @$(PGDEV) forward status +# to whichever machine currently holds that role (on its own eth0:5432). +# macOS Local Network Privacy still has to be granted ONCE (System Settings → +# Privacy & Security → Local Network, entry `socat`; restart the agents after +# granting — proxy.uninstall + proxy.install — since the decision is cached at +# process start). But only once: the grant is keyed to the signing identity, and +# socat is one stable Homebrew binary we never rebuild. The removed Go forwarder +# was re-signed on nearly every `make`, so it re-prompted per build. +# +# All machine tracking (active pointer, machine IPs, +# reconciled targets) lives in var/pgdev.db (SQLite) — each reconcile is a +# flock-guarded, DB-driven rewrite+reload of the socat LaunchAgents with an +# explicit port-free gate + post-verify. Install is explicit: nothing here runs, +# and promote/refresh stay hands-off, until `proxy.install`. + +# All four depend on `pgdevd` (the build target) so `make` rebuilds bin/pgdev +# before invoking it — without this a stale CLI predating the `proxy` command +# fails with `unknown command "proxy"`. +.PHONY: proxy.install +proxy.install: pgdevd machine.exists + @$(PGDEV) proxy install + +.PHONY: proxy.reconcile +proxy.reconcile: pgdevd + @$(PGDEV) proxy reconcile + +.PHONY: proxy.status +proxy.status: pgdevd + @$(PGDEV) proxy status + +.PHONY: proxy.uninstall +proxy.uninstall: pgdevd + @$(PGDEV) proxy uninstall .PHONY: stop stop: @@ -299,7 +310,7 @@ pg.down: deploy $(PGDEV) down .PHONY: pg.status -pg.status: machine.exists pgdevd +pg.status: pgdevd $(PGDEV) status @$(PGDEV) endpoint @@ -318,7 +329,7 @@ pg.shell: machine.exists @$(call PG_DEV_IN,$(ACTIVE_SLOT),shell) .PHONY: pg.ip -pg.ip: machine.exists pgdevd +pg.ip: pgdevd @$(PGDEV) ip .PHONY: pg.logs @@ -338,7 +349,7 @@ pg.restore-last: machine.exists pgdevd $(PGDEV) restore-last $(if $(force),--force,) .PHONY: pg.snapshots -pg.snapshots: machine.exists pgdevd +pg.snapshots: pgdevd $(PGDEV) snapshots # ----- staging backend ---------------------------------------------------- @@ -383,10 +394,32 @@ pg.staging.start: machine.exists pgdevd # fresh backend on it. The active machine — and its data — is never touched. # Unlike the soft resets above this is slow (machine boot + provision) but # actually returns space to macOS; see 'make disk'/§1 of issues/0002. +# +# Deliberately NO disk.check: this IS the reclaim command (it deletes the machine +# first, freeing the whole sparse image, then provisions a fresh EMPTY backend — +# no restore, so no headroom needed). Gating it on free space is self-defeating — +# you run it precisely when you're low, and it is what disk.check itself points +# you to. Do not re-add disk.check here. +# No machine.exists guard: rebuild is a delete+create path (applecli.Recreate's +# delete is a no-op when the machine is gone), and it is the documented way back +# from `pg.staging.purge` — which leaves vpg-b deleted. Guarding it would block +# exactly the recovery it advertises. .PHONY: pg.staging.rebuild -pg.staging.rebuild: disk.check machine.exists pgdevd +pg.staging.rebuild: pgdevd $(PGDEV) staging rebuild $(if $(force),--force,) +# Reclaim WITHOUT rebuild: delete the staging machine (freeing its whole sparse +# macOS disk) and LEAVE IT DOWN — no recreate/provision. Use this to reclaim +# space (or just shut staging down) when you don't want a fresh backend yet; +# bring it back later with pg.staging.rebuild or start. Like rebuild, deliberately +# NO disk.check (it frees space). The active machine is never touched. +# No machine.exists guard either: purge is idempotent (the delete no-ops when +# the machine is already gone) and re-running it still forgets a stale IP and +# drops the socat listener, which is the useful self-heal. +.PHONY: pg.staging.purge +pg.staging.purge: pgdevd + $(PGDEV) staging purge $(if $(force),--force,) + # ----- destructive outer-machine lifecycle ------------------------------- # Apple 1.1 frequently returns an XPC timeout when deleting a RUNNING machine diff --git a/README.md b/README.md index 5b97062..bece705 100644 --- a/README.md +++ b/README.md @@ -42,7 +42,7 @@ its own Incus + copy-on-write XFS snapshot store. One machine is **active** macOS client │ 127.0.0.1:5442 (active) / :5443 (staging) — stable, never changes ▼ -Go forwarder (host launchd agent, internal/forward) — re-points itself on promote +socat client proxy (host launchd agents, internal/socatproxy) — re-pointed on promote │ maps each role port to whichever machine holds that role ├───────────────────────────────┬───────────────────────────────┐ ▼ ▼ @@ -74,8 +74,8 @@ pgdev (macOS) ── HTTP/JSON ──▶ pgdevd (vpg-a | vpg-b) ──▶ Incus ``` Each daemon serves exactly its one backend (slot-implicit API); active/staging -is decided host-side. So `promote` is purely host: **flip `var/active-machine`, -re-point the forwarder** — no data moves, no daemon call. `up` provisions each +is decided host-side. So `promote` is purely host: **flip the active pointer, +re-point the client proxy** — no data moves, no daemon call. `up` provisions each machine's backend from a golden `pg-dev-base` image (PostgreSQL installed once, then `incus publish`ed) and adds its `eth0` proxy device; `status`/`ip`/ `refresh` fan out over both. Each daemon owns its own single-mutation lock, @@ -99,31 +99,36 @@ static-IP pinning. Do not use a backend's 10.x address from macOS. Each machine's `eth0` IP is an unpinnable `bootpd` DHCP lease that can change (the whole `/24` too) after a macOS reboot or machine recreation, and the two -leases drift independently. So the client endpoint is decoupled from them: a -per-user `launchd` agent runs an in-process Go forwarder (`pgdev forward serve`, -`internal/forward`) that owns `127.0.0.1:5442` (active) / `:5443` (staging) for -its whole lifetime and relays each to whichever machine currently holds that -role. It re-points **in place** by swapping the dial target — never rebinding — -so promote can't orphan a listener or leave a stale mapping. The ports are offset -from `5432` so a local PostgreSQL isn't shadowed. Clients always use -**`127.0.0.1:5442`** / **`:5443`** — permanent, identical on every Mac. - -Hands-off: **`make start` installs the forwarder**; after that `pgdev promote` is -just a pointer write — the resident forwarder notices within its poll interval, -re-points, and drops the sessions that were on the demoted machine (reconnect to -land on the new database). `make endpoint.status` / `endpoint.uninstall` manage -the agent (`PG_ENDPOINT_AUTOINSTALL=0` opts out of auto-install). +leases drift independently. So the client endpoint is decoupled from them: two +per-user `launchd` agents run `socat` (`internal/socatproxy`), owning +`127.0.0.1:5442` (active) / `:5443` (staging) and relaying each to whichever +machine currently holds that role. The ports are offset from `5432` so a local +PostgreSQL isn't shadowed. Clients always use **`127.0.0.1:5442`** / **`:5443`** +— permanent, identical on every Mac. + +socat cannot re-point itself, so every `promote` and IP change **reconciles** the +agents: rewrite the plist, `bootout`, wait for the port to actually free, load it +again, then verify the live process really dials the intended target. That +reconcile is driven from `var/pgdev.db` (SQLite) and serialized by a `flock`, so +two concurrent commands can't leave a stale mapping behind. Sessions on the +demoted machine are dropped by the reload — reconnect to land on the new +database. + +Install it once with **`make proxy.install`** (`make proxy.status` / +`proxy.uninstall` manage it); after that `promote`/`refresh` keep it in step +automatically and never touch launchd if it isn't installed. The first run also +needs a one-time macOS **Local Network** grant — see +[macOS Security](#macos-security--grant-local-network-once). `PG_PROXY_HOSTNAME` sets the hostname printed in psql/.pgpass lines (default `host.docker.internal`, so the endpoint also resolves from sibling -containers/k3d; use `127.0.0.1` for host-only). `PG_FORWARD_BIND` widens the +containers/k3d; use `127.0.0.1` for host-only). `PG_CLIENT_BIND` widens the listener bind (default `127.0.0.1`; set `0.0.0.0` only if a sibling container can't reach the Mac's loopback — it exposes the dev backend on every interface). No connection pooler: each port is a per-connection TCP passthrough, so `CREATE`/`DROP DATABASE`, `LISTEN`/`NOTIFY`, prepared statements, advisory locks -and parallel `pg_restore` behave like direct connections. Promoting re-points -the forwarder, which may drop existing sessions (reconnect); the role ports -don't change. +and parallel `pg_restore` behave like direct connections. Promoting reloads the +proxy, which drops existing sessions (reconnect); the role ports don't change. ### Snapshots on Apple's stock kernel @@ -176,8 +181,10 @@ make pg.status The first `make start` builds an Ubuntu 26.04 machine image (systemd, Incus, jq, XFS tools) and creates both machines; later starts reuse them. `make pg.up` installs PostgreSQL 17 in each machine's nested Ubuntu 24.04 container (several -minutes). `make start` also installs and re-points the host forwarder for the -stable `127.0.0.1` endpoint (see [Networking](#networking)). +minutes). Run `make proxy.install` once for the stable `127.0.0.1` endpoints +(see [Networking](#networking)); `make start` keeps them re-pointed afterwards. +The first connection also needs a one-time macOS **Local Network** grant — see +[macOS Security](#macos-security--grant-local-network-once). Status prints endpoints similar to: @@ -269,7 +276,7 @@ Snapshot names may contain letters, digits, dots, underscores, and hyphens. make status # endpoints, roles, states, IPs, timelines make status/incus # Incus versions/resources/list make pg.ip # both machines' IPs and endpoints -make pg.refresh # re-discover both machine IPs, re-point the forwarder +make pg.refresh # re-discover both machine IPs, re-point the client proxy make machine.status # Apple machine JSON (both) make machine.shell # shell into the active machine (slot=a|b to pick) make pg.shell # shell in the active backend; pg.staging.shell for staging @@ -298,58 +305,32 @@ its sparse image) and never touches active. `make recreate` is the full nuke. ## Constraints & gotchas -### macOS Security +### macOS Security — grant Local Network once On macOS 15 Sequoia and later (incl. 26 Tahoe), **Local Network Privacy** gates -the forwarder's connection to the machines' `192.168.64.0/24` subnet. Until it is -granted, the forwarder accepts the client on `:5442`/`:5443` then can't reach the -backend (`EHOSTUNREACH`), and `psql` reports _"server closed the connection -unexpectedly."_ Grant it under **System Settings → Privacy & Security → Local -Network** (enable `pgdev`), then apply the grant to the already-running agent: +any connection to the machines' `192.168.64.0/24` subnet, so the client proxy +needs that permission — **granted once, and only once**. -```sh -make endpoint.restart -``` - -(The grant is cached at process start, so a service that was already running when -you ticked the box stays blocked until restarted.) - -![Local Network permission](doc/img/LocalNetworkpermission.png) - -#### Making the grant survive rebuilds - -The permission is keyed to the binary's **code-signing identity**. By default -`make build` signs `bin/pgdev` **ad-hoc**, which has no stable identity — macOS -falls back to the binary's content hash (cdhash), so **every rebuild looks like a -new program**: it re-prompts and adds another duplicate row to the _Local -Network_ list. - -To fix that permanently, sign with a stable **self-signed code-signing -certificate** (local-dev only; no Apple account needed). One-time setup: - -```sh -openssl req -x509 -newkey rsa:2048 -days 3650 -keyout etc/keys/dev.key -out etc/keys/dev.crt -nodes \ - -subj "/CN=PGDev Signing" -addext "keyUsage=critical,digitalSignature" \ - -addext "extendedKeyUsage=codeSigning" -openssl pkcs12 -export -legacy -in etc/keys/dev.crt -inkey etc/keys/dev.key -out etc/keys/dev.p12 -password pass:dev -security import etc/keys/dev.p12 -k ~/Library/Keychains/login.keychain-db -P dev -T /usr/bin/codesign -# Keychain Access → "PGDev Signing" → Trust → Code Signing: Always Trust -``` - -Then build with that identity (put it in `.env` so it's automatic): +Until it is granted, the proxy accepts your client on `:5442`/`:5443` but cannot +reach the backend (`EHOSTUNREACH`), and `psql` reports _"server closed the +connection unexpectedly."_ Grant it under **System Settings → Privacy & Security +→ Local Network** (enable the `socat` entry), then restart the running agents so +they re-evaluate it — macOS caches the decision at process start, so a proxy that +was already running when you ticked the box stays blocked: ```sh -make build SIGN_CERT="PGDev Signing" # or: echo 'SIGN_CERT=PGDev Signing' >> .env +make proxy.uninstall && make proxy.install ``` -Grant _Local Network_ once; because the signing identity is now stable, every -later `make build` keeps the same grant — no re-prompt, no duplicate entries. +The grant then **sticks across rebuilds**. It is keyed to the binary's +code-signing identity, and the proxy runs Homebrew's `socat` — one stable, +already-signed binary at a fixed path that this repo never rebuilds. (The +removed Go forwarder was the opposite: `make` re-signed it on nearly every +build, so each rebuild looked like a new program — a fresh prompt and another +duplicate row in the _Local Network_ list.) -> Alternative — skip the permission entirely: Local Network Privacy does **not** -> apply to code running as **root**, so installing the forwarder as a -> _LaunchDaemon_ (`/Library/LaunchDaemons`, system domain) instead of a per-user -> _LaunchAgent_ sidesteps it — at the cost of running as root and needing `sudo` -> to install. Not the default here. +If a client hangs after the grant, check `make proxy.status` and the socat logs +in `var/-socat-.log`. ### Disk Space @@ -401,9 +382,14 @@ See `.env.example`. The main settings are: - `PG_DATA_DISK_SIZE` — first-creation XFS logical size (per machine); - `PG_BACKEND_PORT` — port each backend is exposed on (`5432`); - `PG_CLIENT_ACTIVE_PORT`, `PG_CLIENT_STAGING_PORT` — host loopback ports the - forwarder listens on (`5442`/`5443`); + client proxy listens on (`5442`/`5443`); - `PG_PROXY_HOSTNAME` — hostname printed in psql/.pgpass lines (default - `host.docker.internal`; `127.0.0.1` for host-only). + `host.docker.internal`; `127.0.0.1` for host-only); +- `PG_PROXY_DEBUG` — debug-level structured logging for the tracking DB and the + proxy reconcile (on by default); `PG_LOG_COLOR` (`auto`/`always`/`never`) for + its coloring — `auto` colors only a terminal and honors `NO_COLOR`, so + redirected runs stay clean and fully dated — and `PG_LOG_FORMAT` + (`text`/`json`) to swap the colored rendering for machine-readable JSON. ## Why a Makefile if there is a script? diff --git a/doc/img/LocalNetworkpermission.png b/doc/img/LocalNetworkpermission.png deleted file mode 100644 index 51d1703..0000000 Binary files a/doc/img/LocalNetworkpermission.png and /dev/null differ diff --git a/doc/issues/0003-internal-forward.md b/doc/issues/0003-internal-forward.md deleted file mode 100644 index 2cf7e75..0000000 --- a/doc/issues/0003-internal-forward.md +++ /dev/null @@ -1,128 +0,0 @@ -# Spec 0003 — `internal/forward`: replace the shell socat forwarder with Go - -Status: **proposed** (2026-07-23) · Owner: andi · Relates to: -`0001-pgdev-de-shelling-spec.md` (Slice 4 `internal/forward`, deferred), -`0002-two-machine-disk-reclaim.md` (two-machine model). Memory: -`pgdevd-token-home-mount-cold-cache`, `apple-container-cli-quirks`. - -This is a self-contained brief for a fresh session. Goal: retire -`scripts/host-endpoint` (+ the `socat` dependency) and move the host-side client -forwarder into a small in-process Go component. Everything else stays. - ---- - -## 1. Why - -The client forwarder is the **last shell holdover** and the source of every -recent host-side footgun — and they are all *process-lifecycle* bugs that only -exist because the relay is an external process (`socat`) supervised by another -process (`launchd`) over signals: - -- **Orphaned socats / rebind race.** `launchctl kickstart -k` restarts `_serve` - with **SIGKILL**, which bypasses the bash cleanup trap and orphans the socat - children. They keep holding `:5442`/`:5443`, so the re-pointed `_serve` fails - to bind (`Address already in use`) and the **stale mapping persists** — making - `promote` look like a no-op. (Worked around in host-endpoint by reaping stale - port holders on `_serve` start — a hack this spec removes.) -- **Swap-on-promote hazard.** When re-point silently fails, `:5442`/`:5443` keep - the pre-promote mapping, so a `pg_restore -p 5443` (expected: staging) can hit - the **active** DB. Client data path → correctness-critical. -- **launchd fragility** generally (EIO on bootstrap, sandbox plist-write limits). - -A Go forwarder holds the two listeners **in-process** and re-points by swapping -the dial target — nothing to orphan, no rebind, no SIGKILL trap-bypass. The -whole class disappears. - -## 2. What the forwarder does today (to replicate) - -`scripts/host-endpoint` (macOS host only; not in-machine): - -- Reads `MACHINE_PREFIX`, `PG_BACKEND_PORT` (5432), `PG_CLIENT_ACTIVE_PORT` - (5442), `PG_CLIENT_STAGING_PORT` (5443), and the state files - `var/active-machine` (`a`/`b`, default `a`), `var/machine-ip-a`, - `var/machine-ip-b`. -- Maps `127.0.0.1:5442` → `:5432` and `:5443` → - `:5432`, where active = the pointer and staging = the - other. A missing IP for one role must not block the other. -- Runs under a per-user launchd LaunchAgent (`me.pansen.-forward`, - `RunAtLoad`+`KeepAlive`) so it survives logins. Subcommands: `ip`, `install`, - `refresh`, `uninstall`, `status`, `_serve`. -- `refresh` re-reads IPs and `kickstart`s the agent (the fragile re-point path). -- `make start` auto-installs it; `pgdev promote`/`refresh` call `host-endpoint - refresh`; `PG_ENDPOINT_AUTOINSTALL=0` opts out. - -## 3. Design — `internal/forward` - -A long-running host process (macOS) that owns the listeners for their whole -lifetime and re-points targets in place: - -- On start, `net.Listen("tcp", ":5442")` and `:5443` **once**; never - rebind. -- Hold a mutex-guarded routing table: `{active → ":5432", staging → - ":5432"}`, computed from the pointer + IP files. -- **Watch** `var/active-machine` and `var/machine-ip-{a,b}` — poll (~1–2 s) is - simplest and robust on macOS; fsnotify optional. On change, recompute and swap - the two targets atomically. Listeners stay bound. -- Per accepted conn: read the current target for that listener's role, `net.Dial` - it, `io.Copy` both directions, close on either side. If the target IP is empty - (machine down), close the accepted conn with a logged reason. -- Bind address: default `127.0.0.1`. Add an optional wider bind (e.g. `0.0.0.0`) - behind a config knob so `PG_PROXY_HOSTNAME=host.docker.internal` is reachable - from sibling containers/k3d without relying on Docker Desktop forwarding to - loopback. (Today socat binds loopback only.) -- In-flight conns on a target change: leave established ones alone; new conns use - the new target. Matches the existing "promote may drop sessions; reconnect" - contract. (Optional: drop in-flight on promote to force reconnect — decide.) - -### CLI / integration - -- Add `pgdev forward serve | install | uninstall | status` (host CLI). `serve` - is what the LaunchAgent runs. `install`/`uninstall`/`status` manage the plist - (port the host-endpoint launchd logic to Go, or keep a thin installer). -- **`promote` collapses to a pointer write.** `cmd/pgdev` `refreshForwarder()` - becomes a no-op (or is removed): promote already does `active.Set(to)`, and the - running forwarder re-points itself within its poll interval — **no shell exec, - no kickstart, no launchd round-trip.** `refresh` keeps re-discovering IPs - (writing `var/machine-ip-{a,b}`); the forwarder picks those up too. -- `make start`/`endpoint.*` targets call the Go subcommands instead of the - script. - -### Reuse (already present) - -- `config.Config`: `ClientActivePort`, `ClientStagingPort`, `BackendPort`, - `ProxyHostname`, `MachineIPPath(slot)`, `ActiveMachinePath()`, - `MachineNameForSlot(slot)`. -- `internal/activeslot.Pointer` — read `var/active-machine`. -- `internal/applecli` — machine-IP discovery (keep in `pgdev ip`/`refresh`). - -### launchd (stays, but dumb) - -Persistence across login still wants a LaunchAgent — but it just runs `pgdev -forward serve` and **never needs restarting to re-point**. So promote no longer -touches launchd, and the SIGKILL/orphan/rebind problems can't recur. Plist-write -still needs `~/Library/LaunchAgents` (nono-sandbox caveat unchanged — orthogonal -to language). - -## 4. Retire on completion - -- `scripts/host-endpoint` (delete once `pgdev forward` covers install/status). -- `socat` from `make deps` and the README requirements. -- The host-endpoint reap-on-start hack and pointer-flip poll (obviated). - -## 5. Acceptance / tests - -- **Unit:** role→target computation from (pointer, IP files); a pointer flip - updates both targets; missing-IP handling; bind-address selection. -- **Live:** bring up `vpg-a`/`vpg-b`; connect via `:5442`/`:5443`; `pgdev - promote`; assert `:5442` follows to the new active **without any restart** - (verify by `SELECT system_identifier FROM pg_control_system()` vs each machine - IP — the check used during 0002 validation). Repeat promotes leave **no - orphaned processes** and never hit "Address already in use". -- `make check` green; `bash -n` no longer needed for host-endpoint once deleted. - -## 6. Open decisions - -1. Poll vs fsnotify for the watch (recommend: poll ~1–2 s). -2. Bind `127.0.0.1` only vs configurable wider bind for container access. -3. Drop in-flight conns on promote, or let them finish (recommend: let finish). -4. Fold the launchd installer into Go now, or keep a minimal shim temporarily. diff --git a/doc/issues/0004-sqlite-socat-proxy.md b/doc/issues/0004-sqlite-socat-proxy.md new file mode 100644 index 0000000..562fa3c --- /dev/null +++ b/doc/issues/0004-sqlite-socat-proxy.md @@ -0,0 +1,171 @@ +# Spec 0004 — SQLite machine tracking + socat client proxy (retire the Go forwarder) + +Status: **done** (2026-07-25) · Owner: andi · Relates to: +`0002-two-machine-disk-reclaim.md` (two-machine model) and spec 0003, the Go +forwarder this retired (its file was deleted with the code; see git history). +Memory: `pgdevd-token-home-mount-cold-cache`, `apple-apiserver-wedged-recovery`. + +This is a **living document**: it records the design as it was built and what +remains. The Go forwarder and its codesign ceremony are now **removed** (§5). + +--- + +## 1. Why (course correction from 0003) + +Spec 0003 replaced the shell `socat` relay with an in-process Go forwarder +(`internal/forward`, on 127.0.0.1:5442/:5443) to kill socat's process-lifecycle +bugs. It works — but its **own binary** trips macOS Local Network Privacy: it +needs a stable codesign identity plus a manual Local Network grant, and STILL +throws permission prompts on rebuilds (`make` re-signs the running agent's +executable). Homebrew's `socat` needs the Local Network grant too — TCC gates the +subnet, not the program — but it is ONE stable, already-signed binary at a fixed +path that we never rebuild, so the grant is asked **once** and then sticks. + +So we re-introduce a socat path **as an experiment**, but tame the lifecycle +races that got it retired, using: + +- **SQLite** (`var/pgdev.db`) as the single source of truth for machine + tracking, so a reconcile is one atomic, transactional decision instead of a + scatter of racing `os.WriteFile`s. +- an explicit **port-free gate + post-verify** around every socat reload — the + actual fix for the orphan-listener → EADDRINUSE → silent-stale-mapping bug + (SQLite alone does NOT fix that; it only serializes reconcilers). + +The socat proxy took over the **client ports** (5442 active / 5443 staging), so +existing external configs picked it up with no change. The Go forwarder ran +alongside on a second pair (5444 / 5445) for one integration phase and is now +deleted — socat is the only host-side client path. + +## 2. What shipped in this change + +### 2.1 `internal/track` — SQLite machine tracking (source of truth) + +`var/pgdev.db` (modernc.org/sqlite, **pure Go, no cgo**, WAL, busy_timeout). +Tables: + +- `schema_meta(version)` — schema version. +- `active(id=1, slot)` — which machine is active (replaces `var/active-machine`). +- `machine(slot, ip, updated_unix)` — cached drifting eth0 IPs (replaces + `var/machine-ip-{a,b}`). +- `proxy_target(role, port, target, updated_unix)` — the reconciled socat + upstreams, **observability only** (see §3). + +**Schema-change automation instead of migration tooling.** On `Open`, if the +stored version ≠ `schemaVersion`, `track` drops every table and recreates the +schema (all rows are reconstructible — IPs re-discover). The one decision that +is NOT a cache, the **active slot**, is carried across the reset (old table → +legacy `var/active-machine` file → `"a"`, and only the last is loud). `Open` +never touches launchd; it returns a `reset` bool and the caller reconciles. + +**Chokepoint + dual-write.** Every state write goes through `track`, which writes +the DB first and then **mirrors the flat file**. The mirrors were introduced to +keep the untouched Go forwarder correct; they outlived it because `pgdev` itself +(`internal/activeslot`, `machineIP`) and the Makefile's `ACTIVE_SLOT` still READ +those files. Retiring them means moving those readers onto the DB (§5.2). + +Structured logging throughout via stdlib **`log/slog`** (text handler → stderr, +`component=track`). `PG_PROXY_DEBUG=1` (the default) makes it chatty. + +### 2.2 `internal/socatproxy` — the socat LaunchAgents + +Two per-user LaunchAgents, `me.pansen.-socat-active` (:5442) and +`-socat-staging` (:5443) — the **client ports**. Each runs: + +``` +socat -d -d \ + TCP-LISTEN:5442,bind=127.0.0.1,reuseaddr,fork,keepalive,nodelay \ + TCP::5432,connect-timeout=5,keepalive,nodelay +``` + +`connect-timeout` is essential — without it a drifted/stale target hangs psql +~75s per connection. `-d -d` logs connection lifecycle to `var/-socat-.log`. + +socat cannot re-point itself, so each reconcile **rewrites and reloads** the +plist. The reload is the hardened sequence (this is the anti-orphan core): + +1. `launchctl bootout`, **poll until launchd reports it gone** (children exit). +2. **Port-free gate**: attempt `net.Listen` on the port; poll. If still held, + `lsof` the LISTEN pid and kill it **only if it is one of our own socat + children** (argv[0] is `socat` and it listens on exactly this port — an + ERROR, launchd leaked); a foreign holder is reported and the reconcile fails + rather than killing an unrelated process. A port that never frees is a **hard + failure** — refuse to bootstrap onto a held port (which is exactly how the + stale mapping used to persist). +3. Write plist (atomic), `launchctl bootstrap` (retry transient EIO). +4. **Post-verify**: poll until the port is LISTENing AND the live socat's argv + contains the intended target (`lsof` → pid → `ps`). Because socat's parent + never dials, this argv check is the ONLY stale-mapping detector, so it is + mandatory; a timeout fails the reconcile loudly. + +### 2.3 Reconcile flow — locking discipline + +The reconcile is guarded by an **flock** (`var/reconcile.flock`), NOT by holding a +SQLite transaction across `launchctl`. Rationale: a wedged `launchctl` (a known +macOS fragility on this host — see `apple-apiserver-wedged-recovery`) holding a +DB write lock would brick every other `pgdev` command. So: + +- **flock** = the reconcile mutex (dies with the process; no stale-lock recovery). +- **Tx 1 (ms)**: `DesiredTargets` reads active+IPs in one snapshot → can't mix a + pre-promote slot with post-promote IPs. +- Apply to launchd (outside any tx), comparing desired against the **on-disk + plist** (the durable truth of what launchd runs), not the DB — so a crash + between apply and record self-heals next pass. +- **Tx 2 (ms)**: `RecordApplied` stamps `proxy_target` for `proxy status`. + +Triggered from: `pgdev proxy reconcile`/`install`, and automatically (only if the +proxy is already installed) after `promote` and `refresh`. + +### 2.4 CLI + Make + +- `pgdev proxy install | reconcile | status | uninstall` +- `make proxy.install | proxy.reconcile | proxy.status | proxy.uninstall` + +`proxy install` is the single command that brings the client path up. Installing +is also what opts in: `promote`/`refresh` stay hands-off (never touch launchd) +until a plist exists (`Reconciler.AnyInstalled`). + +A port held by a foreign process hard-fails the gate loudly — we do not kill +arbitrary processes. + +## 3. Deliberate decisions / non-goals + +- **`proxy_target` is observability, not authority.** The durable record of what + socat runs is the plist; decisions compare against it. +- **`status` surfaces socat** — one line per role (launchd state + the DB's + last-applied target), which replaced the old forwarder section. +- **The Local Network grant is a one-time manual step**, not something the CLI + tries to automate: TCC cannot be granted programmatically. After granting, + the agents must be restarted (`proxy.uninstall` + `proxy.install`) because the + decision is cached at process start — a plain `proxy reconcile` is a no-op + when the mapping is already healthy, so it will NOT pick the grant up. +- **Data loss on schema change is accepted** (IPs re-discover; active carried). +- **Host-only DB.** Never open `var/pgdev.db` from inside a guest over virtiofs + — SQLite over virtiofs corrupts (cf. the token cold-cache burn in 0002). + +## 4. Open questions + +- [ ] Behaviour of an in-flight `pg_restore` when a reload kills the socat + children mid-transfer (expected: dropped; client reconnects). Acceptable + for dev; confirm. + +## 5. Removal of the Go forwarder + +1. **Done** — `internal/forward`, `pgdev forward *`, the `endpoint.*` make + targets, `PG_FORWARD_*`, and the `codesign` step in `pgdev/Makefile` (plus + `etc/keys`, spec 0003 and the Local-Network screenshot) are deleted. + `PG_FORWARD_BIND` became `PG_CLIENT_BIND` (it configures socat now). +2. **Open** — the **flat-file mirrors** in `internal/track` (`ActiveMirror`, + `IPMirror`) and the `activeslot` file reads. These are NOT dead: `pgdev` + resolves the active slot and machine IPs from `var/active-machine` / + `var/machine-ip-{a,b}`, and the Makefile reads `ACTIVE_SLOT` from the same + file at parse time. Retiring them means moving both readers onto the DB — + including a Makefile that can't shell out to a not-yet-built binary — after + which "no plain-text tracking" is fully realized. +3. **Done** — `pgdev status` shows the socat proxy where the forwarder section + used to be. +4. **Done** — README + `.env.example` updated. + +Stale runtime leftovers from the forwarder (`var/forward-state.json`, +`var/-forward.log`) and its LaunchAgent +(`~/Library/LaunchAgents/me.pansen.-forward.plist`) must be booted out +and deleted once on each machine that ran it. diff --git a/etc/keys/.gitkeep b/etc/keys/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/pgdev/Makefile b/pgdev/Makefile index 5cce612..99ddbc5 100644 --- a/pgdev/Makefile +++ b/pgdev/Makefile @@ -17,21 +17,6 @@ GOARCH ?= arm64 # (host pgdev ↔ in-machine pgdevd via GET /v1/version) matches. VERSION ?= $(shell git -C .. describe --tags --always --dirty 2>/dev/null || echo dev) LDFLAGS := -s -w -X main.version=$(VERSION) -# Code-signing for the host CLI. macOS Local Network Privacy (Tahoe+) keys the -# forwarder LaunchAgent's grant to the binary's SIGNING IDENTITY, and we re-sign -# on every host build so a plain `go build` can't drop it. -# -# SIGN_CERT — the `codesign --sign` identity. Default "-" is ad-hoc: it has no -# stable identity, so macOS falls back to the cdhash, which changes every -# build → each rebuild re-prompts for Local Network AND adds a duplicate row -# to the list. Set it to a stable keychain code-signing identity (a one-time -# self-signed cert is enough for local dev — see README "macOS Security") and -# the once-granted permission survives every rebuild, no re-prompt: -# make build SIGN_CERT="PGDev Signing" (or set it in .env) -# SIGN_ID — the stable --identifier, so the Local Network entry shows a fixed -# name (me.pansen.pgdev) regardless of which cert signs it. -SIGN_ID := me.pansen.pgdev -SIGN_CERT ?= - .DEFAULT_GOAL := build @@ -52,10 +37,6 @@ daemon: host: @mkdir -p $(BIN_DIR) @go build -trimpath -ldflags '$(LDFLAGS)' -o $(HOST_OUT) ./cmd/pgdev - @if command -v codesign >/dev/null 2>&1; then \ - codesign --force --sign "$(SIGN_CERT)" --identifier $(SIGN_ID) $(HOST_OUT) || exit 1; \ - echo "signed $(HOST_OUT) ($(SIGN_ID), cert=$(SIGN_CERT))"; \ - fi @echo "built $(HOST_OUT) (host, $(VERSION))" ## build-host: build for the host OS/arch (handy for local smoke runs) diff --git a/pgdev/cmd/pgdev/commands.go b/pgdev/cmd/pgdev/commands.go index 2c4e34f..5ed3312 100644 --- a/pgdev/cmd/pgdev/commands.go +++ b/pgdev/cmd/pgdev/commands.go @@ -32,8 +32,8 @@ func (a *app) upCmd() *cobra.Command { if err != nil { return err } - fmt.Printf("==> [%s] Provisioning (the first run builds the golden PostgreSQL image; this can take a few minutes)...\n", - a.cfg.MachineNameForSlot(slot)) + a.log.Info("provisioning backend (the first run builds the golden PostgreSQL image; this can take a few minutes)", + "machine", a.cfg.MachineNameForSlot(slot)) if _, err := cl.Up(ctx); err != nil { return fmt.Errorf("%s: %w", a.cfg.MachineNameForSlot(slot), err) } @@ -41,12 +41,12 @@ func (a *app) upCmd() *cobra.Command { // Default to "a" active the first time the pointer file has never // been written, so a fresh `pgdev up` has a well-defined role split. if _, err := os.Stat(a.cfg.ActiveMachinePath()); os.IsNotExist(err) { - if err := a.active.Set("a"); err != nil { + if err := a.setActive(ctx, "a"); err != nil { return err } } - a.ensureForwarder(ctx) - fmt.Println("==> pg-dev ready.") + a.reconcileProxyIfInstalled(ctx, "up") + a.log.Info("pg-dev ready") fmt.Println() a.renderStatus(ctx) return nil @@ -65,15 +65,15 @@ func (a *app) downCmd() *cobra.Command { machine := a.cfg.MachineNameForSlot(slot) cl, err := a.longClientFor(ctx, slot) if err != nil { - fmt.Fprintf(os.Stderr, "WARNING: %s: %v\n", machine, err) + a.log.Warn("skipping machine", "machine", machine, "err", err) continue } res, err := cl.Down(ctx) if err != nil { - fmt.Fprintf(os.Stderr, "WARNING: %s: %v\n", machine, err) + a.log.Warn("down failed", "machine", machine, "err", err) continue } - fmt.Printf("[%s] %s\n", machine, res.Message) + a.log.Info(res.Message, "machine", machine) } return nil }, @@ -100,6 +100,10 @@ func (a *app) statusCmd() *cobra.Command { type slotStatus struct { st agentapi.StatusResponse err error + // absent is set when the machine itself is gone (never created, or deleted + // by `pg.staging.purge`). That is an expected steady state, not a fault, so + // status reports it as ABSENT rather than UNREACHABLE. + absent bool } // fetchStatuses queries both machines' status, tolerating per-machine errors. @@ -107,12 +111,16 @@ func (a *app) fetchStatuses(ctx context.Context) map[string]slotStatus { out := make(map[string]slotStatus, len(slotsAB)) for _, slot := range slotsAB { cl, err := a.clientFor(ctx, slot) - if err != nil { - out[slot] = slotStatus{err: err} - continue + if err == nil { + var st agentapi.StatusResponse + if st, err = cl.Status(ctx); err == nil { + out[slot] = slotStatus{st: st} + continue + } } - st, err := cl.Status(ctx) - out[slot] = slotStatus{st: st, err: err} + // Only ask the (slow) Apple CLI whether the machine exists once + // something already went wrong — the happy path stays exec-free. + out[slot] = slotStatus{err: err, absent: !a.apple(slot).Exists(ctx)} } return out } @@ -142,15 +150,15 @@ func (a *app) renderStatus(ctx context.Context) { machine := a.cfg.MachineNameForSlot(slot) endpoint := fmt.Sprintf("%s:%d", a.cfg.ProxyHostname, a.cfg.ClientPort(role)) if ms.err != nil { - fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%s\t%s\n", role, machine, "-", "UNREACHABLE", endpoint, "-", "-") + fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%s\t%s\n", role, machine, "-", statusState(ms), endpoint, "-", "-") continue } fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%d\t%s\n", - role, machine, orDash(ms.st.Container), orAbsent(ms.st.State), endpoint, len(ms.st.Snapshots), orDash(strings.Join(ms.st.IPs, ","))) + role, machine, orDash(ms.st.Container), statusState(ms), endpoint, len(ms.st.Snapshots), orDash(strings.Join(ms.st.IPs, ","))) } tw.Flush() fmt.Println() - a.renderForwarder(ctx) + a.renderProxy(ctx) fmt.Println() a.renderSnapshots(statuses, true) } @@ -175,6 +183,10 @@ func (a *app) renderSnapshots(statuses map[string]slotStatus, withPsql bool) { ms := statuses[slot] machine := a.cfg.MachineNameForSlot(slot) fmt.Printf("─── %-7s (%s) ───\n", role, machine) + if ms.absent { + fmt.Printf("(machine %s does not exist — run '%s' to bring it back)\n\n", machine, rebuildHint(role)) + continue + } if ms.err != nil { fmt.Printf("(unreachable: %v)\n\n", ms.err) continue @@ -218,10 +230,8 @@ func (a *app) endpointCmd() *cobra.Command { fmt.Println("psql commands:") fmt.Printf(" active: %s\n", a.psqlCmd(a.cfg.ClientActivePort)) fmt.Printf(" staging: %s\n", a.psqlCmd(a.cfg.ClientStagingPort)) - if h == "127.0.0.1" { - fmt.Printf("\nnote: 127.0.0.1 needs the host forwarder ('make endpoint.install'); it\n") - fmt.Printf(" relays to each machine's IP, which may drift on reboot ('pgdev refresh' re-points it).\n") - } + fmt.Printf("\nnote: these ports are served by the socat client proxy ('make proxy.install'); it\n") + fmt.Printf(" relays to each machine's IP, which may drift on reboot ('pgdev refresh' re-points it).\n") return nil }, } @@ -239,7 +249,7 @@ func (a *app) ipCmd() *cobra.Command { for _, role := range []string{"active", "staging"} { slot := a.roleSlot(role) ip := a.machineIP(ctx, slot) - a.writeMachineIPFile(slot, ip) + a.writeMachineIPFile(ctx, slot, ip) endpoint := fmt.Sprintf("%s:%d", a.cfg.ProxyHostname, a.cfg.ClientPort(role)) fmt.Fprintf(tw, "%s\t%s\t%s\t%s\n", role, a.cfg.MachineNameForSlot(slot), orQ(ip), endpoint) } @@ -254,7 +264,7 @@ func (a *app) ipCmd() *cobra.Command { func (a *app) promoteCmd() *cobra.Command { return &cobra.Command{ Use: "promote", - Short: "Flip active↔staging (re-point the host forwarder, no data moves)", + Short: "Flip active↔staging (re-point the client proxy, no data moves)", Args: cobra.NoArgs, RunE: func(cmd *cobra.Command, _ []string) error { ctx := cmd.Context() @@ -285,22 +295,20 @@ func (a *app) promoteCmd() *cobra.Command { } } - // Promote collapses to a pointer write: the resident forwarder - // re-points itself within its poll interval and drops the sessions - // that were on the demoted machine (spec 0003 §3). No launchd - // round-trip, so nothing to fail-and-roll-back here. - if err := a.active.Set(to); err != nil { + // Promote is a pointer write plus a proxy reconcile: no data moves and + // no daemon call. + if err := a.setActive(ctx, to); err != nil { return err } - if !a.awaitForwarderRepoint(ctx, to) { - fmt.Fprintf(os.Stderr, - "WARNING: the forwarder did not confirm re-pointing to %s. The active pointer IS set; verify with\n"+ - " 'pgdev forward status' before using :%d — is the forwarder running ('pgdev forward install')?\n", - a.cfg.MachineNameForSlot(to), a.cfg.ClientActivePort) - } - - fmt.Printf("Promoted. active=%s staging=%s\n\n", - a.cfg.MachineNameForSlot(to), a.cfg.MachineNameForSlot(from)) + // socat serves the client ports (5442/5443) and cannot re-point itself, + // so re-point it synchronously (DB-driven, flock-guarded, verified) if + // installed. Sessions on the demoted machine are dropped; clients + // reconnect onto the new active. + a.reconcileProxyIfInstalled(ctx, "promote") + + a.log.Info("promoted", + "active", a.cfg.MachineNameForSlot(to), "staging", a.cfg.MachineNameForSlot(from)) + fmt.Println() a.renderStatus(ctx) return nil }, @@ -310,39 +318,39 @@ func (a *app) promoteCmd() *cobra.Command { func (a *app) refreshCmd() *cobra.Command { return &cobra.Command{ Use: "refresh", - Short: "Re-discover both machine IPs and re-point the host forwarder", + Short: "Re-discover both machine IPs and re-point the client proxy", Args: cobra.NoArgs, RunE: func(cmd *cobra.Command, _ []string) error { ctx := cmd.Context() for _, slot := range slotsAB { machine := a.cfg.MachineNameForSlot(slot) ip := a.machineIP(ctx, slot) - a.writeMachineIPFile(slot, ip) + a.writeMachineIPFile(ctx, slot, ip) if ip == "" { - fmt.Printf("[%s] no IP (machine down?) — skipping reconcile\n", machine) + a.log.Warn("no IP (machine down?) — skipping reconcile", "machine", machine) continue } cl, err := a.clientFor(ctx, slot) if err != nil { - fmt.Printf("[%s] %v\n", machine, err) + a.log.Warn("machine unreachable", "machine", machine, "err", err) continue } res, err := cl.Reconcile(ctx) if err != nil { - fmt.Printf("[%s] reconcile: %v\n", machine, err) + a.log.Error("backend reconcile failed", "machine", machine, "err", err) continue } - fmt.Printf("[%s] backend running=%v\n", machine, res.BackendRunning) + a.log.Info("backend reconciled", "machine", machine, "running", res.BackendRunning) for _, act := range res.Actions { - fmt.Printf(" %s\n", act) + a.log.Info("backend action", "machine", machine, "action", act) } } - // The forwarder re-reads the IP files we just rewrote on its next - // poll; refresh only needs to make sure the agent is installed. - a.ensureForwarder(ctx) - fmt.Printf("Endpoints: active %s:%d → %s, staging %s:%d → %s (forwarder tracks the IP files)\n", - a.cfg.ProxyHostname, a.cfg.ClientActivePort, a.cfg.MachineNameForSlot(a.active.Get()), - a.cfg.ProxyHostname, a.cfg.ClientStagingPort, a.cfg.MachineNameForSlot(a.active.Staging())) + // Re-point the socat proxy at the freshly-discovered IPs (only if it's + // installed; socat can't re-point itself). + a.reconcileProxyIfInstalled(ctx, "refresh") + a.log.Info("endpoints re-pointed", + "active", fmt.Sprintf("%s:%d → %s", a.cfg.ProxyHostname, a.cfg.ClientActivePort, a.cfg.MachineNameForSlot(a.active.Get())), + "staging", fmt.Sprintf("%s:%d → %s", a.cfg.ProxyHostname, a.cfg.ClientStagingPort, a.cfg.MachineNameForSlot(a.active.Staging()))) return nil }, } @@ -366,7 +374,7 @@ func (a *app) snapshotCmd(role string) *cobra.Command { if err != nil { return err } - fmt.Println(res.Message) + a.log.Info(res.Message, "machine", a.cfg.MachineNameForSlot(a.roleSlot(role)), "snapshot", args[0]) return nil }, } @@ -421,7 +429,7 @@ func (a *app) runRestore(ctx context.Context, role, name string, last, force boo return fmt.Errorf("no snapshots on %s", machine) } target = snaps.Snapshots[len(snaps.Snapshots)-1].Name - fmt.Printf("==> Restoring %s to most recent snapshot: %s\n", machine, target) + a.log.Info("restoring to most recent snapshot", "machine", machine, "snapshot", target) } after := snapshotsAfter(snaps.Snapshots, target) @@ -445,7 +453,7 @@ func (a *app) runRestore(ctx context.Context, role, name string, last, force boo if err != nil { return err } - fmt.Println(res.Message) + a.log.Info(res.Message, "machine", machine) return nil } @@ -470,6 +478,7 @@ func (a *app) stagingCmd() *cobra.Command { reset, a.stagingStartCmd(), a.stagingStopCmd(), + a.stagingPurgeCmd(), a.stagingRebuildCmd(), ) return c @@ -493,7 +502,7 @@ func (a *app) stagingStartCmd() *cobra.Command { if err != nil { return err } - fmt.Println(res.Message) + a.log.Info(res.Message, "machine", a.cfg.MachineNameForSlot(a.roleSlot("staging"))) return nil }, } @@ -514,12 +523,83 @@ func (a *app) stagingStopCmd() *cobra.Command { if err != nil { return err } - fmt.Println(res.Message) + a.log.Info(res.Message, "machine", a.cfg.MachineNameForSlot(a.roleSlot("staging"))) return nil }, } } +// stagingForDestruction resolves the staging slot and machine names and asserts +// staging is NOT the active machine — the load-bearing safety property shared by +// both destructive staging tiers (rebuild, purge). Structurally Staging() is the +// pointer's complement so they always differ, but this guards the one invariant +// that, if ever violated, would nuke live data — so it is asserted explicitly. +func (a *app) stagingForDestruction() (slot, machine, activeMachine string, err error) { + slot = a.active.Staging() + machine = a.cfg.MachineNameForSlot(slot) + activeMachine = a.cfg.MachineNameForSlot(a.active.Get()) + if machine == activeMachine { + return "", "", "", fmt.Errorf("refusing to operate on %s — it is the active machine", machine) + } + return slot, machine, activeMachine, nil +} + +// stagingPurgeCmd is the reclaim-WITHOUT-rebuild tier: delete ONLY the staging +// machine — reclaiming its grown sparse macOS disk — and LEAVE IT DOWN. Unlike +// rebuild it does not recreate/deploy/provision; staging stays gone until you +// `rebuild` (or `make start`) it back. The active machine is never touched. It +// also forgets staging's now-dead IP and drops its socat listener, so the +// endpoint honestly reads "down" instead of dialing a corpse. +func (a *app) stagingPurgeCmd() *cobra.Command { + var force bool + c := &cobra.Command{ + Use: "purge", + Short: "Delete the staging machine to reclaim its macOS disk and leave it DOWN (no rebuild)", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + ctx := cmd.Context() + slot, machine, activeMachine, err := a.stagingForDestruction() + if err != nil { + return err + } + + fmt.Printf("==> This DELETES %s (staging), discarding its data and snapshots, and reclaims its macOS disk.\n", machine) + fmt.Printf(" It will NOT be recreated — run 'make pg.staging.rebuild' or 'make start' to bring it back.\n") + fmt.Printf(" %s (active) is never touched.\n", activeMachine) + if !force { + ok, err := confirm() + if err != nil { + return err + } + if !ok { + return errors.New("aborted") + } + } + + cli := a.apple(slot) + a.log.Info("stopping and deleting the machine (this reclaims its macOS disk)", "machine", machine) + if err := cli.Delete(ctx); err != nil { + return err + } + + // Forget staging's now-invalid IP and drop its socat listener, so + // :5443 reads "down" rather than dialing the deleted machine. + if db, err := a.track(ctx); err == nil { + if err := db.ForgetMachine(ctx, slot); err != nil { + a.log.Warn("forgetting staging machine failed", "slot", slot, "err", err) + } + } + a.reconcileProxyIfInstalled(ctx, "staging purge") + + a.log.Info("purged — the machine is deleted and its macOS disk reclaimed; active was never touched", + "purged", machine, "active", activeMachine) + return nil + }, + } + c.Flags().BoolVar(&force, "force", false, "skip the confirmation prompt") + return c +} + // stagingRebuildCmd is the hard-reset reclaim tier (spec 0002 §0.1/§2): delete // and recreate ONLY the staging machine — reclaiming its grown sparse macOS // disk — then re-provision a fresh backend on it. The active machine is never @@ -532,15 +612,9 @@ func (a *app) stagingRebuildCmd() *cobra.Command { Args: cobra.NoArgs, RunE: func(cmd *cobra.Command, _ []string) error { ctx := cmd.Context() - staging := a.active.Staging() - active := a.active.Get() - machine := a.cfg.MachineNameForSlot(staging) - activeMachine := a.cfg.MachineNameForSlot(active) - // Structurally staging != active (Staging() is the pointer's - // complement), but this is the load-bearing safety property of the - // whole command, so assert it explicitly rather than trust that. - if machine == activeMachine { - return fmt.Errorf("refusing to rebuild %s — it is the active machine", machine) + staging, machine, activeMachine, err := a.stagingForDestruction() + if err != nil { + return err } fmt.Printf("==> This DELETES %s (staging), discarding its data and snapshots, and reclaims its macOS disk.\n", machine) @@ -556,18 +630,18 @@ func (a *app) stagingRebuildCmd() *cobra.Command { } cli := a.apple(staging) - fmt.Printf("==> [%s] Deleting and recreating (this reclaims its macOS disk)...\n", machine) + a.log.Info("deleting and recreating the machine (this reclaims its macOS disk)", "machine", machine) opts := applecli.CreateOpts{CPUs: a.cfg.MachineCPUs, Memory: a.cfg.MachineMemory, Image: a.cfg.MachineImage} if err := cli.Recreate(ctx, opts, 5*time.Minute); err != nil { return err } - fmt.Printf("==> [%s] Installing pgdevd...\n", machine) + a.log.Info("installing pgdevd", "machine", machine) if err := a.deploy(ctx, staging); err != nil { return err } - fmt.Printf("==> [%s] Provisioning fresh backend...\n", machine) + a.log.Info("provisioning a fresh backend", "machine", machine) cl, err := a.longClientFor(ctx, staging) if err != nil { return err @@ -576,9 +650,11 @@ func (a *app) stagingRebuildCmd() *cobra.Command { return err } - a.ensureForwarder(ctx) + // The recreated machine has a fresh DHCP lease, so point the proxy at it. + a.reconcileProxyIfInstalled(ctx, "staging rebuild") - fmt.Printf("==> Reclaim done. %s is fresh; %s (active) was never touched.\n", machine, activeMachine) + a.log.Info("reclaim done — the machine is fresh; active was never touched", + "rebuilt", machine, "active", activeMachine) return nil }, } @@ -644,9 +720,41 @@ func orAbsent(s string) string { } return s } + +// statusState is the STATE column for one slot. A deleted machine (never +// created, or purged) reads ABSENT — a legitimate steady state since +// `pg.staging.purge` — and is kept distinct from UNREACHABLE, which means the +// machine is there but its daemon did not answer. +func statusState(ms slotStatus) string { + switch { + case ms.absent: + return "ABSENT" + case ms.err != nil: + return "UNREACHABLE" + } + return orAbsent(ms.st.State) +} + +// rebuildHint names the command that recreates a deleted machine: staging has +// its own cheap rebuild (the everyday reclaim tier after `pg.staging.purge`), +// active only comes back through the full `make start` path. +func rebuildHint(role string) string { + if role == "staging" { + return "make pg.staging.rebuild" + } + return "make start" +} + func orQ(s string) string { if s == "" { return "?" } return s } + +func orNone(s string) string { + if s == "" { + return "(none)" + } + return s +} diff --git a/pgdev/cmd/pgdev/commands_test.go b/pgdev/cmd/pgdev/commands_test.go index 753d785..6ae8183 100644 --- a/pgdev/cmd/pgdev/commands_test.go +++ b/pgdev/cmd/pgdev/commands_test.go @@ -1,6 +1,7 @@ package main import ( + "errors" "reflect" "testing" @@ -76,3 +77,31 @@ func TestOrHelpers(t *testing.T) { t.Errorf("orQ(\"\") = %q, want \"?\"", got) } } + +// A purged machine must report ABSENT (and not abort status), distinctly from a +// machine that exists but whose daemon is silent. +func TestStatusState(t *testing.T) { + for _, tc := range []struct { + name string + ms slotStatus + want string + }{ + {"purged machine", slotStatus{err: errors.New("no IP"), absent: true}, "ABSENT"}, + {"exists but silent", slotStatus{err: errors.New("connection refused")}, "UNREACHABLE"}, + {"running", slotStatus{st: agentapi.StatusResponse{State: "RUNNING"}}, "RUNNING"}, + {"no state reported", slotStatus{}, "ABSENT"}, + } { + if got := statusState(tc.ms); got != tc.want { + t.Errorf("%s: statusState = %q, want %q", tc.name, got, tc.want) + } + } +} + +func TestRebuildHint(t *testing.T) { + if got := rebuildHint("staging"); got != "make pg.staging.rebuild" { + t.Errorf("rebuildHint(staging) = %q", got) + } + if got := rebuildHint("active"); got != "make start" { + t.Errorf("rebuildHint(active) = %q", got) + } +} diff --git a/pgdev/cmd/pgdev/deploy.go b/pgdev/cmd/pgdev/deploy.go index db01386..951b03a 100644 --- a/pgdev/cmd/pgdev/deploy.go +++ b/pgdev/cmd/pgdev/deploy.go @@ -70,15 +70,15 @@ func (a *app) agentVersionCmd() *cobra.Command { name := a.cfg.MachineNameForSlot(slot) cl, err := a.clientFor(ctx, slot) if err != nil { - fmt.Printf("[%s] %v\n", name, err) + a.log.Error("version handshake failed", "machine", name, "err", err) continue } v, err := cl.Version(ctx) if err != nil { - fmt.Printf("[%s] %v\n", name, err) + a.log.Error("version handshake failed", "machine", name, "err", err) continue } - fmt.Printf("[%s] pgdevd %s (api v%d)\n", name, v.Version, v.APIVersion) + a.log.Info("pgdevd version", "machine", name, "pgdevd", v.Version, "api", v.APIVersion) } return nil }, @@ -161,19 +161,19 @@ func (a *app) deploy(ctx context.Context, slot string) error { // A freshly created machine may still be finishing its first boot; wait until // systemd is up before installing/enabling, or Apple rejects the exec // ("Operation not supported by device") or systemctl can't reach the bus. - fmt.Printf("==> [%s] Waiting for the machine to be ready...\n", machine) + a.log.Info("waiting for the machine to be ready", "machine", machine) if err := cli.WaitReady(ctx, 120*time.Second); err != nil { return err } // A (re)created machine gets a fresh DHCP lease, so the cached // var/machine-ip- can be stale. Refresh it from the live address - // before the handshake dials the daemon (this also re-points the endpoint - // forwarder at the new IP once refreshForwarder next runs). + // before the handshake dials the daemon (the next proxy reconcile then picks + // the new IP up from the tracking DB). if ip, err := cli.MachineIP(ctx); err == nil && ip != "" { - a.writeMachineIPFile(slot, ip) + a.writeMachineIPFile(ctx, slot, ip) } - fmt.Printf("==> [%s] Installing pgdevd into the machine...\n", machine) + a.log.Info("installing pgdevd into the machine", "machine", machine) steps := [][]string{ // Atomic install to the machine-local run path: stage then rename, never // write-in-place → no ETXTBSY on the live binary. @@ -194,7 +194,7 @@ func (a *app) deploy(ctx context.Context, slot string) error { } } - fmt.Printf("==> [%s] Restarting pgdevd...\n", machine) + a.log.Info("restarting pgdevd", "machine", machine) if out, err := cli.Run(ctx, "systemctl", "restart", "pgdevd"); err != nil { return fmt.Errorf("restarting daemon: %w\n%s", err, out) } @@ -222,7 +222,7 @@ func (a *app) awaitVersion(ctx context.Context, slot string) error { if err == nil { last = v.Version if version == "dev" || v.Version == version { - fmt.Printf("==> [%s] Deployed. pgdevd %s (api v%d) is live.\n", machine, v.Version, v.APIVersion) + a.log.Info("deployed — pgdevd is live", "machine", machine, "pgdevd", v.Version, "api", v.APIVersion) return nil } } diff --git a/pgdev/cmd/pgdev/forward.go b/pgdev/cmd/pgdev/forward.go deleted file mode 100644 index a716610..0000000 --- a/pgdev/cmd/pgdev/forward.go +++ /dev/null @@ -1,233 +0,0 @@ -package main - -import ( - "context" - "fmt" - "os" - "path/filepath" - "strings" - "time" - - "github.com/spf13/cobra" - - "pansen.me/pgdev/internal/forward" -) - -// forwardCmd groups the host-side client forwarder (internal/forward), the Go -// replacement for scripts/host-endpoint (spec 0003). `serve` is the long-running -// relay a LaunchAgent runs; install/uninstall/status manage that agent. Promote -// no longer touches any of this — the running serve re-points itself from the -// pointer file. -func (a *app) forwardCmd() *cobra.Command { - c := &cobra.Command{ - Use: "forward", - Short: "Host client forwarder: stable 127.0.0.1:5442/:5443 → the active/staging machine", - } - c.AddCommand( - a.forwardServeCmd(), - a.forwardInstallCmd(), - a.forwardRestartCmd(), - a.forwardUninstallCmd(), - a.forwardStatusCmd(), - ) - return c -} - -// forwardOptions builds the Server config from the resolved cfg. -func (a *app) forwardOptions() forward.Options { - return forward.Options{ - Bind: a.cfg.ForwardBind, - ActivePort: a.cfg.ClientActivePort, - StagingPort: a.cfg.ClientStagingPort, - BackendPort: a.cfg.BackendPort, - ActiveMachinePath: a.cfg.ActiveMachinePath(), - MachineIPPath: a.cfg.MachineIPPath, - StatePath: a.cfg.ForwardStatePath(), - Log: logf, - } -} - -// launchd builds the LaunchAgent handle for the current binary. -func (a *app) launchd() (*forward.Launchd, error) { - exe, err := forwardExecutable() - if err != nil { - return nil, err - } - program := []string{exe, "forward", "serve"} - ld := forward.NewLaunchd(a.cfg.MachinePrefix, program, a.cfg.ForwardLogPath(), a.cfg.RepoRoot) - // Migration: on install, kill any orphaned socat still holding the client - // ports (done inside Install, AFTER bootout, so the retired KeepAlive agent - // can't respawn them). See Launchd.Install. - ld.ReapPorts = []int{a.cfg.ClientActivePort, a.cfg.ClientStagingPort} - return ld, nil -} - -func (a *app) forwardServeCmd() *cobra.Command { - return &cobra.Command{ - Use: "serve", - Short: "Run the forwarder in the foreground (what the LaunchAgent runs)", - Args: cobra.NoArgs, - RunE: func(cmd *cobra.Command, _ []string) error { - opts := a.forwardOptions() - if !isLoopback(opts.Bind) { - fmt.Fprintf(os.Stderr, - "WARNING: binding %s exposes the dev-credentialed PostgreSQL backend on every interface (LAN/Wi-Fi).\n", - opts.Bind) - } - return forward.New(opts).Serve(cmd.Context()) - }, - } -} - -func (a *app) forwardInstallCmd() *cobra.Command { - return &cobra.Command{ - Use: "install", - Short: "Install (and start) the per-user LaunchAgent that keeps the forwarder alive", - Args: cobra.NoArgs, - RunE: func(cmd *cobra.Command, _ []string) error { - ld, err := a.launchd() - if err != nil { - return err - } - if err := ld.Install(cmd.Context()); err != nil { - return err - } - fmt.Printf("==> Forwarder '%s' installed and started.\n", ld.Label) - fmt.Printf(" active 127.0.0.1:%d staging 127.0.0.1:%d (re-points itself on promote)\n", - a.cfg.ClientActivePort, a.cfg.ClientStagingPort) - return nil - }, - } -} - -// forwardRestartCmd restarts the running agent so a permission granted after it -// started (macOS Local Network Privacy caches the decision at process start) is -// re-evaluated — without this, ticking the Local Network box has no effect until -// the next reboot/reinstall and the forwarder keeps failing with EHOSTUNREACH. -func (a *app) forwardRestartCmd() *cobra.Command { - return &cobra.Command{ - Use: "restart", - Short: "Restart the running forwarder (apply a freshly-granted macOS Local Network permission)", - Args: cobra.NoArgs, - RunE: func(cmd *cobra.Command, _ []string) error { - ld, err := a.launchd() - if err != nil { - return err - } - if err := ld.Restart(cmd.Context()); err != nil { - return err - } - fmt.Printf("==> Forwarder '%s' restarted.\n", ld.Label) - return nil - }, - } -} - -func (a *app) forwardUninstallCmd() *cobra.Command { - return &cobra.Command{ - Use: "uninstall", - Short: "Stop and remove the forwarder LaunchAgent", - Args: cobra.NoArgs, - RunE: func(cmd *cobra.Command, _ []string) error { - ld, err := a.launchd() - if err != nil { - return err - } - if err := ld.Uninstall(cmd.Context()); err != nil { - return err - } - fmt.Printf("==> Forwarder '%s' removed.\n", ld.Label) - return nil - }, - } -} - -func (a *app) forwardStatusCmd() *cobra.Command { - return &cobra.Command{ - Use: "status", - Short: "Forwarder agent state, cached IPs, and the live effective mapping", - Args: cobra.NoArgs, - RunE: func(cmd *cobra.Command, _ []string) error { - ld, err := a.launchd() - if err != nil { - return err - } - fmt.Printf("plist: %s\n", ld.Plist) - fmt.Printf("installed: %v\n", ld.Installed()) - fmt.Printf("active machine: %s\n", a.cfg.MachineNameForSlot(a.active.Get())) - fmt.Printf("cached IP a: %s\n", orNone(readFileTrim(a.cfg.MachineIPPath("a")))) - fmt.Printf("cached IP b: %s\n", orNone(readFileTrim(a.cfg.MachineIPPath("b")))) - a.renderForwarder(cmd.Context()) - return nil - }, - } -} - -// renderForwarder prints the two live forwarder lines — the LaunchAgent's -// launchd state and the effective mapping/heartbeat from the state file. Shared -// by `forward status` and the main `status` command so both agree. -func (a *app) renderForwarder(ctx context.Context) { - ld, err := a.launchd() - if err != nil { - fmt.Printf("forwarder: %v\n", err) - return - } - statusLine := ld.Status(ctx) - st, ok := forward.ReadState(a.cfg.ForwardStatePath()) - if !ok { - fmt.Printf("%s\n", statusLine) - fmt.Printf("(no state file — forwarder has not run)\n") - return - } - live := "STALE" - if st.Fresh(10 * time.Second) { - live = "live" - } - age := time.Since(time.Unix(st.UpdatedUnix, 0)).Round(time.Second) - fmt.Printf("%s (updated %s ago, %s)\n", statusLine, age, live) - fmt.Printf("active -> %s\n", orNone(st.Active)) - fmt.Printf("staging -> %s\n", orNone(st.Staging)) -} - -// ----- helpers --------------------------------------------------------------- - -// logf is a logx.Func that streams the forwarder's progress lines to stderr. -func logf(format string, args ...any) { - fmt.Fprintf(os.Stderr, "==> "+format+"\n", args...) -} - -// forwardExecutable resolves the absolute path to bake into the plist, refusing -// a `go run` temp binary (which vanishes on exit, crash-looping the agent). -func forwardExecutable() (string, error) { - exe, err := os.Executable() - if err != nil { - return "", fmt.Errorf("resolve executable for LaunchAgent: %w", err) - } - exe, err = filepath.EvalSymlinks(exe) - if err != nil { - return "", err - } - if strings.HasPrefix(exe, os.TempDir()) { - return "", fmt.Errorf("refusing to install a LaunchAgent for a temporary binary (%s) — build pgdev first ('make -C pgdev build') and install from bin/pgdev", exe) - } - return exe, nil -} - -func isLoopback(bind string) bool { - return bind == "" || bind == "127.0.0.1" || bind == "::1" || bind == "localhost" -} - -func readFileTrim(path string) string { - b, err := os.ReadFile(path) - if err != nil { - return "" - } - return strings.TrimSpace(string(b)) -} - -func orNone(s string) string { - if s == "" { - return "(none)" - } - return s -} diff --git a/pgdev/cmd/pgdev/main.go b/pgdev/cmd/pgdev/main.go index 49a9de6..0714112 100644 --- a/pgdev/cmd/pgdev/main.go +++ b/pgdev/cmd/pgdev/main.go @@ -3,11 +3,11 @@ // pgdev talks HTTP/JSON to the resident pgdevd daemon (internal/agentapi) at each // machine's eth0. Since spec 0002 (two machines) there is one daemon PER machine // (vpg-a/vpg-b, one backend each) and active/staging is a HOST-side concept: a -// pointer file (internal/activeslot, now pointed at the ACTIVE MACHINE) picks -// which machine's client is "active" vs "staging", and the in-process host -// forwarder (internal/forward, run as `pgdev forward serve` under a LaunchAgent) -// maps the stable 127.0.0.1:5442/:5443 client ports onto whichever machine -// currently holds each role, re-pointing itself when the pointer flips. The only +// pointer (internal/activeslot, now pointed at the ACTIVE MACHINE) picks which +// machine's client is "active" vs "staging", and the socat client proxy +// (internal/socatproxy, per-user LaunchAgents driven by internal/track) maps the +// stable 127.0.0.1:5442/:5443 client ports onto whichever machine currently holds +// each role, re-pointed by a reconcile when the pointer flips. The only // `container` execs left // here are IP discovery (internal/applecli, a fallback) and `agent deploy`'s // one-shot install/restart, plus the hard-reset machine lifecycle used by @@ -17,9 +17,9 @@ package main import ( "context" "fmt" + "log/slog" "os" "os/signal" - "path/filepath" "strings" "syscall" "time" @@ -30,7 +30,7 @@ import ( "pansen.me/pgdev/internal/agentapi" "pansen.me/pgdev/internal/applecli" "pansen.me/pgdev/internal/config" - "pansen.me/pgdev/internal/forward" + "pansen.me/pgdev/internal/track" ) // version is stamped at build time (see Makefile), matched against each @@ -46,6 +46,12 @@ type app struct { // active is the host-side pointer to which MACHINE is active (behind // :5442); the other machine is staging (behind :5443). See spec 0002 §0.1. active activeslot.Pointer + // log is the structured (slog) logger threaded into the tracking DB and the + // socat proxy. Text handler to stderr; debug when cfg.ProxyVerbose. + log *slog.Logger + // trackDB is the SQLite tracking store (internal/track), opened lazily and + // cached for the process. nil until first use; see app.track. + trackDB *track.DB } func main() { @@ -62,6 +68,7 @@ func newApp() *app { return &app{ cfg: cfg, active: activeslot.Pointer{Path: cfg.ActiveMachinePath(), UID: cfg.HostUID, GID: cfg.HostGID}, + log: newLogger(cfg.ProxyVerbose, cfg.LogColor, cfg.LogFormat), } } @@ -84,7 +91,7 @@ func rootCmd() *cobra.Command { a.snapshotsCmd(), a.ipCmd(), a.endpointCmd(), - a.forwardCmd(), + a.proxyCmd(), a.snapshotCmd("active"), a.restoreCmd("active"), a.restoreLastCmd("active"), @@ -118,14 +125,44 @@ func (a *app) machineIP(ctx context.Context, slot string) string { } // writeMachineIPFile caches slot's discovered IP host-side so subsequent -// commands (and the host forwarder) don't need a live `container` exec. -func (a *app) writeMachineIPFile(slot, ip string) { +// commands don't need a live `container` exec. The cache lives in the SQLite +// tracking DB (the source of truth); track mirrors it back to the flat +// var/machine-ip- file this CLI still reads in machineIP (see +// doc/issues/0004 §5.2 — retiring the mirrors is the remaining step). +// There is deliberately NO file-only fallback when the DB is unavailable: +// a file-only write would diverge the mirror from the (stale) DB, and socat — +// the canonical client path — routes on the DB, so it would silently forward to +// the wrong machine. A DB that won't open is a real fault to surface, not paper +// over; IP caching is best-effort, so we log loudly and skip (next run retries). +func (a *app) writeMachineIPFile(ctx context.Context, slot, ip string) { if ip == "" { return } - path := a.cfg.MachineIPPath(slot) - _ = os.MkdirAll(filepath.Dir(path), 0o755) - _ = os.WriteFile(path, []byte(ip+"\n"), 0o644) + db, err := a.track(ctx) + if err != nil { + a.log.Error("cannot cache machine IP: tracking DB unavailable", "slot", slot, "ip", ip, "err", err) + return + } + if err := db.SetMachineIP(ctx, slot, ip); err != nil { + a.log.Warn("caching machine IP failed", "slot", slot, "err", err) + } +} + +// setActive is the CHOKEPOINT for the active pointer: it writes the SQLite +// tracking DB (source of truth), which mirrors the value back to the flat +// var/active-machine file that this CLI (activeslot) and the Makefile's +// ACTIVE_SLOT still read. If the DB can't be opened this is a HARD error — +// unlike IP caching, a promote must not half-succeed: writing only the flat file +// would leave the DB (and therefore socat, the client path) pointing at the OLD +// active machine, so a client would keep hitting the pre-promote database. Fail +// loudly so the operator knows the flip did not take, rather than silently +// splitting the two paths. +func (a *app) setActive(ctx context.Context, slot string) error { + db, err := a.track(ctx) + if err != nil { + return fmt.Errorf("cannot set active slot %q: tracking DB unavailable (%w)", slot, err) + } + return db.SetActive(ctx, slot) } // clientFor builds a typed daemon client against slot's machine. It ensures @@ -172,63 +209,3 @@ func (a *app) clientForRole(ctx context.Context, role string) (*agentapi.Client, func (a *app) longClientForRole(ctx context.Context, role string) (*agentapi.Client, error) { return a.longClientFor(ctx, a.roleSlot(role)) } - -// ensureForwarder makes the client forwarder hands-off: on `make start` it -// validates the LaunchAgent against the current build and self-heals a missing, -// unloaded, stale, or crashed one, so 127.0.0.1:5442/:5443 come up (and stay -// correct) automatically. This is deliberately stronger than the old bare -// "is a plist on disk?" check: after the repo moves or is renamed, a plist is -// still on disk but points at a deleted binary — it loads, exits EX_CONFIG, and -// leaves the ports dead. Ensure catches that program drift. On the healthy -// common path it is a single `launchctl print` + compare, no launchd writes, so -// the resident serve keeps re-pointing itself from the pointer file untouched -// (no kickstart on promote/refresh; spec 0003 §3). It NEVER fails the caller: -// PG_ENDPOINT_AUTOINSTALL=0 opts out, and a sandbox that can't write -// ~/Library/LaunchAgents just warns. -func (a *app) ensureForwarder(ctx context.Context) { - if os.Getenv("PG_ENDPOINT_AUTOINSTALL") == "0" { - return - } - ld, err := a.launchd() - if err != nil { - // e.g. running via `go run` (temp binary) — can't self-install; leave the - // forwarder to an explicit `pgdev forward install` from a built binary. - fmt.Fprintf(os.Stderr, "WARNING: skipping forwarder auto-install: %v\n", err) - return - } - res, err := ld.Ensure(ctx) - if err != nil { - fmt.Fprintf(os.Stderr, "WARNING: forwarder auto-install failed (run 'pgdev forward install'): %v\n", err) - return - } - switch res.Action { - case "installed": - fmt.Fprintf(os.Stderr, "==> Endpoint forwarder '%s' installed (%s).\n", ld.Label, res.Reason) - case "reinstalled": - fmt.Fprintf(os.Stderr, "==> Endpoint forwarder '%s' re-synced: %s.\n", ld.Label, res.Reason) - } -} - -// awaitForwarderRepoint gives the resident forwarder up to a couple of poll -// intervals to re-point onto the newly-active slot after a promote, then reports -// whether it confirmably did. It reads internal/forward's state file rather than -// touching launchd, so promote stays a pure pointer write. A false return is a -// warning, not a hard failure: the pointer IS the source of truth and the -// forwarder converges once it is running — but we surface an unconfirmed -// re-point so a following `pg_restore -p 5443` isn't silently misrouted. -func (a *app) awaitForwarderRepoint(ctx context.Context, slot string) bool { - deadline := time.Now().Add(4 * time.Second) - for { - if st, ok := forward.ReadState(a.cfg.ForwardStatePath()); ok && st.ActiveSlot == slot && st.Fresh(10*time.Second) { - return true - } - if time.Now().After(deadline) { - return false - } - select { - case <-ctx.Done(): - return false - case <-time.After(250 * time.Millisecond): - } - } -} diff --git a/pgdev/cmd/pgdev/proxy.go b/pgdev/cmd/pgdev/proxy.go new file mode 100644 index 0000000..90f48a3 --- /dev/null +++ b/pgdev/cmd/pgdev/proxy.go @@ -0,0 +1,357 @@ +package main + +import ( + "context" + "fmt" + "log/slog" + "os" + "os/exec" + "strings" + + "github.com/lmittmann/tint" + "github.com/spf13/cobra" + + "pansen.me/pgdev/internal/logx" + "pansen.me/pgdev/internal/socatproxy" + "pansen.me/pgdev/internal/track" +) + +// newLogger builds the process's structured logger. The default sink is +// lmittmann/tint — a drop-in slog.Handler that renders slog's own key=value +// shape with the level and keys colored. EVERY progress line goes through it +// (the old bare "==> " Printf lines included), so a run reads as one colored, +// timestamped stream on stderr, while stdout keeps only the reports you copy +// from: status tables, endpoints, .pgpass/psql lines, snapshot timelines — and +// the destructive confirm banners, which must render verbatim next to their +// prompt. verbose flips the level to Debug, which the tracking + socat paths use +// to be deliberately chatty (we want a fat forensic trail). +// +// Two knobs, resolved from .env like the rest of the config: +// +// - PG_LOG_COLOR=auto|always|never — auto (the default) colors only a real +// terminal and honors NO_COLOR/TERM (logx.UseColor; tint does no detection +// of its own). always forces escapes through a pipe (`… |& less -R`). +// - PG_LOG_FORMAT=text|json — json swaps in the stdlib JSON handler, for a run +// that is captured and machine-parsed rather than watched. +// +// On a terminal the stamp is a bare clock; anywhere else it carries the date +// too, so a redirected promote/reconcile trail stays self-dating. +func newLogger(verbose bool, colorMode, format string) *slog.Logger { + level := slog.LevelInfo + if verbose { + level = slog.LevelDebug + } + if strings.EqualFold(strings.TrimSpace(format), "json") { + return slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: level})) + } + color := logx.UseColor(colorMode, os.Stderr) + timeFormat := "2006-01-02 15:04:05.000" + if color { + timeFormat = "15:04:05.000" + } + // Only tint the level when we are actually coloring: with NoColor a tinted + // value takes a different render path in tint for no gain. + var replace func([]string, slog.Attr) slog.Attr + if color { + replace = greyDebugLevel + } + return slog.New(tint.NewTextHandler(os.Stderr, &tint.Options{ + Level: level, + TimeFormat: timeFormat, + NoColor: !color, + ReplaceAttr: replace, + })) +} + +// ansiGrey is tint's palette index for bright black (rendered as \033[90m) — +// the same grey it already uses for timestamps and keys. +const ansiGrey = 8 + +// greyDebugLevel paints the DBG label grey. tint colors INF green, WRN yellow +// and ERR red but leaves DBG in the default foreground, so with PG_PROXY_DEBUG +// on (the default) the chatty debug stream reads just as loud as the lines that +// matter. Grey lets it recede without losing it. +func greyDebugLevel(groups []string, a slog.Attr) slog.Attr { + if len(groups) != 0 || a.Key != slog.LevelKey { + return a + } + if lvl, ok := a.Value.Any().(slog.Level); ok && lvl < slog.LevelInfo { + return tint.Attr(ansiGrey, a) + } + return a +} + +// track lazily opens (and caches) the SQLite tracking DB. First open brings the +// schema to the current version — dropping+recreating on a version change while +// carrying the active slot across — and a reset is the cue to reconcile the +// socat proxy (only if the experiment is installed; otherwise a no-op). +func (a *app) track(ctx context.Context) (*track.DB, error) { + if a.trackDB != nil { + return a.trackDB, nil + } + db, reset, err := track.Open(ctx, track.Options{ + Path: a.cfg.TrackDBPath(), + Logger: a.log, + ActiveMirror: a.cfg.ActiveMachinePath(), + IPMirror: a.cfg.MachineIPPath, + MirrorUID: a.cfg.HostUID, + MirrorGID: a.cfg.HostGID, + }) + if err != nil { + return nil, err + } + a.trackDB = db + if reset { + a.log.Warn("tracking schema was reset — reconciling socat proxy if installed") + a.reconcileProxyIfInstalled(ctx, "schema reset") + } + return db, nil +} + +// reconciler builds the socat proxy reconciler from config. socat is resolved +// from PATH; an absolute path is baked into the plist so launchd (cwd "/") finds +// it. A missing socat is a clear, actionable error rather than a launchd EX_*. +func (a *app) reconciler() (*socatproxy.Reconciler, error) { + socat, err := exec.LookPath("socat") + if err != nil { + return nil, fmt.Errorf("socat not found on PATH (brew install socat): %w", err) + } + return &socatproxy.Reconciler{ + Prefix: a.cfg.MachinePrefix, + Bind: a.cfg.ClientBind, + Socat: socat, + LockPath: a.cfg.ReconcileLockPath(), + LogPath: a.cfg.SocatLogPath, + UID: a.cfg.HostUID, + Log: a.log, + }, nil +} + +// proxyPorts is the role->port map the reconciler/status need. socat binds the +// CANONICAL client ports (5442/5443) so existing external configs hit it. +func (a *app) proxyPorts() map[string]int { + return map[string]int{"active": a.cfg.ClientActivePort, "staging": a.cfg.ClientStagingPort} +} + +// desiredProxyTargets reads the current active pointer + machine IPs from the DB +// in one atomic snapshot and maps them onto the socat ports. +func (a *app) desiredProxyTargets(ctx context.Context, db *track.DB) ([]socatproxy.Target, error) { + ts, err := db.DesiredTargets(ctx, a.cfg.ClientActivePort, a.cfg.ClientStagingPort, a.cfg.BackendPort) + if err != nil { + return nil, err + } + out := make([]socatproxy.Target, 0, len(ts)) + for _, t := range ts { + out = append(out, socatproxy.Target{Role: t.Role, Port: t.Port, Target: t.Target}) + } + return out, nil +} + +// reconcileProxy is the full guarded reconcile: read desired targets from the DB +// (atomic snapshot), apply them to the socat LaunchAgents under the flock, and +// record what verified-applied back into the DB (observability). Returns the +// per-role outcome. +func (a *app) reconcileProxy(ctx context.Context, reason string) ([]socatproxy.Applied, error) { + a.log.Info("proxy reconcile requested", "reason", reason) + db, err := a.track(ctx) + if err != nil { + return nil, err + } + rec, err := a.reconciler() + if err != nil { + return nil, err + } + // Targets are fetched INSIDE Reconcile, under the flock, so a stale snapshot + // can never be applied last (see socatproxy.Reconcile). + return rec.Reconcile(ctx, + func(ctx context.Context) ([]socatproxy.Target, error) { + return a.desiredProxyTargets(ctx, db) + }, + func(t socatproxy.Target) error { + return db.RecordApplied(ctx, track.Target{Role: t.Role, Port: t.Port, Target: t.Target}) + }) +} + +// appliedErr turns per-role convergence failures into a command error so +// `proxy reconcile`/`install` (and the make targets chaining on them) exit +// non-zero instead of printing FAILED and returning success. +func appliedErr(applied []socatproxy.Applied) error { + var bad []string + for _, ap := range applied { + if ap.Err != nil { + bad = append(bad, ap.Role) + } + } + if len(bad) > 0 { + return fmt.Errorf("socat proxy: %s did not converge (see log above)", strings.Join(bad, ", ")) + } + return nil +} + +// reconcileProxyIfInstalled reconciles ONLY when the socat experiment is already +// installed (at least one plist on disk). This keeps promote/refresh completely +// hands-off for anyone not opted into the socat trial — they never touch +// launchd — while keeping an installed proxy in step with the active pointer and +// IP drift. Never fails the caller: a reconcile problem is logged, not fatal. +func (a *app) reconcileProxyIfInstalled(ctx context.Context, reason string) { + rec, err := a.reconciler() + if err != nil { + a.log.Debug("socat proxy not reconciled (no socat)", "err", err) + return + } + if !rec.AnyInstalled() { + a.log.Debug("socat proxy not installed — skipping reconcile", "reason", reason) + return + } + applied, err := a.reconcileProxy(ctx, reason) + if err != nil { + a.log.Error("socat proxy reconcile failed", "reason", reason, "err", err) + return + } + for _, ap := range applied { + if ap.Err != nil { + a.log.Error("socat proxy role did not converge", "role", ap.Role, "err", ap.Err) + } + } +} + +// renderProxy prints the client proxy's live state — one line per role: the +// LaunchAgent's launchd state and the target the DB last recorded as applied. +// Shared by `proxy status` and the main `status` command so both agree. Never +// fails: with no socat (or nothing installed) it says so and moves on, because +// status must keep rendering. +func (a *app) renderProxy(ctx context.Context) { + rec, err := a.reconciler() + if err != nil { + fmt.Printf("client proxy: %v\n", err) + return + } + live := rec.Status(ctx, a.proxyPorts()) + applied := map[string]track.Target{} + if db, err := a.track(ctx); err == nil { + if t, err := db.AppliedTargets(ctx); err == nil { + applied = t + } + } + for _, role := range []string{"active", "staging"} { + fmt.Printf("proxy %-8s :%d %s", role, a.cfg.ClientPort(role), live[role]) + if t, ok := applied[role]; ok { + fmt.Printf(" (db target: %s)", orNone(t.Target)) + } + fmt.Println() + } +} + +// ----- `pgdev proxy` command ------------------------------------------------- + +// proxyCmd groups the socat proxy (internal/socatproxy) that serves the client +// ports (5442/5443) — the only host-side client path since the Go forwarder was +// removed. Installing it is explicit (`proxy install`); promote/refresh only +// reconcile an ALREADY-installed proxy. See doc/issues/0004. +func (a *app) proxyCmd() *cobra.Command { + c := &cobra.Command{ + Use: "proxy", + Short: "socat client proxy on the client ports (5442/5443)", + } + c.AddCommand( + a.proxyReconcileCmd(), + a.proxyInstallCmd(), + a.proxyUninstallCmd(), + a.proxyStatusCmd(), + ) + return c +} + +func (a *app) proxyReconcileCmd() *cobra.Command { + return &cobra.Command{ + Use: "reconcile", + Short: "Re-point the socat LaunchAgents at the active/staging machine IPs (DB-driven, flock-guarded)", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + applied, err := a.reconcileProxy(cmd.Context(), "manual reconcile") + if err != nil { + return err + } + a.logApplied(applied) + return appliedErr(applied) + }, + } +} + +// proxyInstallCmd brings up the client path: the socat LaunchAgents on the +// client ports. Installing is what opts promote/refresh into reconciling, so +// this is the one command to run after a fresh checkout or a `proxy uninstall`. +func (a *app) proxyInstallCmd() *cobra.Command { + return &cobra.Command{ + Use: "install", + Short: "Install the socat client proxy on the client ports (5442/5443)", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + ctx := cmd.Context() + if !isLoopback(a.cfg.ClientBind) { + a.log.Warn("this bind exposes the dev-credentialed PostgreSQL backend on every interface (LAN/Wi-Fi)", + "bind", a.cfg.ClientBind) + } + applied, err := a.reconcileProxy(ctx, "install") + if err != nil { + return err + } + a.log.Info("socat proxy installed on the client ports — promote/refresh now keep it in step automatically", + "active", fmt.Sprintf("%s:%d", a.cfg.ClientBind, a.cfg.ClientActivePort), + "staging", fmt.Sprintf("%s:%d", a.cfg.ClientBind, a.cfg.ClientStagingPort)) + a.logApplied(applied) + return appliedErr(applied) + }, + } +} + +func (a *app) proxyUninstallCmd() *cobra.Command { + return &cobra.Command{ + Use: "uninstall", + Short: "Stop and remove both socat LaunchAgents", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + rec, err := a.reconciler() + if err != nil { + return err + } + if err := rec.Uninstall(cmd.Context(), a.proxyPorts()); err != nil { + return err + } + a.log.Info("socat proxy removed") + return nil + }, + } +} + +func (a *app) proxyStatusCmd() *cobra.Command { + return &cobra.Command{ + Use: "status", + Short: "socat LaunchAgent state and last-applied targets", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + a.renderProxy(cmd.Context()) + return nil + }, + } +} + +// logApplied reports the per-role outcome of a reconcile. A role that did not +// converge is an ERROR, not a printed line the eye slides over — the reconcile +// as a whole then fails via appliedErr. +func (a *app) logApplied(applied []socatproxy.Applied) { + for _, ap := range applied { + if ap.Err != nil { + a.log.Error("proxy role did not converge", "role", ap.Role, "port", ap.Port, "err", ap.Err) + continue + } + a.log.Info("proxy role applied", "role", ap.Role, "port", ap.Port, "action", ap.Action, "target", orNone(ap.Target)) + } +} + +// isLoopback reports whether the client listeners stay on the loopback — the +// safe default. Anything else publishes the dev-credentialed backend to the +// network, so `proxy install` says so out loud. +func isLoopback(bind string) bool { + return bind == "" || bind == "127.0.0.1" || bind == "::1" || bind == "localhost" +} diff --git a/pgdev/go.mod b/pgdev/go.mod index 2dbc36e..bcb4008 100644 --- a/pgdev/go.mod +++ b/pgdev/go.mod @@ -2,9 +2,26 @@ module pansen.me/pgdev go 1.25 -require github.com/spf13/cobra v1.10.2 +require ( + github.com/lmittmann/tint v1.2.0 + github.com/spf13/cobra v1.10.2 + modernc.org/sqlite v1.31.1 +) require ( + github.com/dustin/go-humanize v1.0.1 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/ncruces/go-strftime v0.1.9 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/spf13/pflag v1.0.9 // indirect + golang.org/x/sys v0.22.0 // indirect + modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 // indirect + modernc.org/libc v1.55.3 // indirect + modernc.org/mathutil v1.6.0 // indirect + modernc.org/memory v1.8.0 // indirect + modernc.org/strutil v1.2.0 // indirect + modernc.org/token v1.1.0 // indirect ) diff --git a/pgdev/go.sum b/pgdev/go.sum index a6ee3e0..009de2a 100644 --- a/pgdev/go.sum +++ b/pgdev/go.sum @@ -1,10 +1,61 @@ github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/google/pprof v0.0.0-20240409012703-83162a5b38cd h1:gbpYu9NMq8jhDVbvlGkMFWCjLFlqqEZjEmObmhUy6Vo= +github.com/google/pprof v0.0.0-20240409012703-83162a5b38cd/go.mod h1:kf6iHlnVGwgKolg33glAes7Yg/8iWP8ukqeldJSO7jw= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/lmittmann/tint v1.2.0 h1:AogHRHy8HUJUnNJBHJlYa+fR4YY8mko2cnCp67xn9JY= +github.com/lmittmann/tint v1.2.0/go.mod h1:HIS3gSy7qNwGCj+5oRjAutErFBl4BzdQP6cJZ0NfMwE= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4= +github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU= github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiTUUS4= github.com/spf13/pflag v1.0.9 h1:9exaQaMOCwffKiiiYk6/BndUBv+iRViNW+4lEMi0PvY= github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +golang.org/x/mod v0.16.0 h1:QX4fJ0Rr5cPQCF7O9lh9Se4pmwfwskqZfq5moyldzic= +golang.org/x/mod v0.16.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.22.0 h1:RI27ohtqKCnwULzJLqkv897zojh5/DwS/ENaMzUOaWI= +golang.org/x/sys v0.22.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/tools v0.19.0 h1:tfGCXNR1OsFG+sVdLAitlpjAvD/I6dHDKnYrpEZUHkw= +golang.org/x/tools v0.19.0/go.mod h1:qoJWxmGSIBmAeriMx19ogtrEPrGtDbPK634QFIcLAhc= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +modernc.org/cc/v4 v4.21.4 h1:3Be/Rdo1fpr8GrQ7IVw9OHtplU4gWbb+wNgeoBMmGLQ= +modernc.org/cc/v4 v4.21.4/go.mod h1:HM7VJTZbUCR3rV8EYBi9wxnJ0ZBRiGE5OeGXNA0IsLQ= +modernc.org/ccgo/v4 v4.19.2 h1:lwQZgvboKD0jBwdaeVCTouxhxAyN6iawF3STraAal8Y= +modernc.org/ccgo/v4 v4.19.2/go.mod h1:ysS3mxiMV38XGRTTcgo0DQTeTmAO4oCmJl1nX9VFI3s= +modernc.org/fileutil v1.3.0 h1:gQ5SIzK3H9kdfai/5x41oQiKValumqNTDXMvKo62HvE= +modernc.org/fileutil v1.3.0/go.mod h1:XatxS8fZi3pS8/hKG2GH/ArUogfxjpEKs3Ku3aK4JyQ= +modernc.org/gc/v2 v2.4.1 h1:9cNzOqPyMJBvrUipmynX0ZohMhcxPtMccYgGOJdOiBw= +modernc.org/gc/v2 v2.4.1/go.mod h1:wzN5dK1AzVGoH6XOzc3YZ+ey/jPgYHLuVckd62P0GYU= +modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 h1:5D53IMaUuA5InSeMu9eJtlQXS2NxAhyWQvkKEgXZhHI= +modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6/go.mod h1:Qz0X07sNOR1jWYCrJMEnbW/X55x206Q7Vt4mz6/wHp4= +modernc.org/libc v1.55.3 h1:AzcW1mhlPNrRtjS5sS+eW2ISCgSOLLNyFzRh/V3Qj/U= +modernc.org/libc v1.55.3/go.mod h1:qFXepLhz+JjFThQ4kzwzOjA/y/artDeg+pcYnY+Q83w= +modernc.org/mathutil v1.6.0 h1:fRe9+AmYlaej+64JsEEhoWuAYBkOtQiMEU7n/XgfYi4= +modernc.org/mathutil v1.6.0/go.mod h1:Ui5Q9q1TR2gFm0AQRqQUaBWFLAhQpCwNcuhBOSedWPo= +modernc.org/memory v1.8.0 h1:IqGTL6eFMaDZZhEWwcREgeMXYwmW83LYW8cROZYkg+E= +modernc.org/memory v1.8.0/go.mod h1:XPZ936zp5OMKGWPqbD3JShgd/ZoQ7899TUuQqxY+peU= +modernc.org/opt v0.1.3 h1:3XOZf2yznlhC+ibLltsDGzABUGVx8J6pnFMS3E4dcq4= +modernc.org/opt v0.1.3/go.mod h1:WdSiB5evDcignE70guQKxYUl14mgWtbClRi5wmkkTX0= +modernc.org/sortutil v1.2.0 h1:jQiD3PfS2REGJNzNCMMaLSp/wdMNieTbKX920Cqdgqc= +modernc.org/sortutil v1.2.0/go.mod h1:TKU2s7kJMf1AE84OoiGppNHJwvB753OYfNl2WRb++Ss= +modernc.org/sqlite v1.31.1 h1:XVU0VyzxrYHlBhIs1DiEgSl0ZtdnPtbLVy8hSkzxGrs= +modernc.org/sqlite v1.31.1/go.mod h1:UqoylwmTb9F+IqXERT8bW9zzOWN8qwAIcLdzeBZs4hA= +modernc.org/strutil v1.2.0 h1:agBi9dp1I+eOnxXeiZawM8F4LawKv4NzGWSaLfyeNZA= +modernc.org/strutil v1.2.0/go.mod h1:/mdcBmfOibveCTBxUl5B5l6W+TTH1FXPLHZE6bTosX0= +modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= +modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= diff --git a/pgdev/internal/config/config.go b/pgdev/internal/config/config.go index d115aa3..cdb897b 100644 --- a/pgdev/internal/config/config.go +++ b/pgdev/internal/config/config.go @@ -47,7 +47,7 @@ type Config struct { MachinePrefix string // vpg Slot string // "" host-side; "a"/"b" inside a machine's pgdevd // BackendPort is the single port each machine exposes its one backend on, - // over its own eth0 (no in-machine proxy). The host forwarder maps the + // over its own eth0 (no in-machine proxy). The host client proxy maps the // client ports to :BackendPort / :BackendPort. BackendPort int // 5432 // Per-machine resources for host-orchestrated create/hard-reset. Two @@ -59,18 +59,32 @@ type Config struct { // Machine-side proxy ports (the Incus proxy devices' listeners). LEGACY — // retired with the in-machine pg-proxy once routing moves host-side. ActivePort, StagingPort int // 5432 / 5433 - // Host loopback ports clients actually connect to. + // ClientActivePort/ClientStagingPort are the client-facing loopback ports — + // what psql/.pgpass/status print and what external configs point at. The + // socat proxy (internal/socatproxy) binds THESE. ClientActivePort, ClientStagingPort int // 5442 / 5443 + // ProxyVerbose makes the socat proxy + tracking DB log at debug level. On by + // default: this is an experiment we want to be chatty about (PG_PROXY_DEBUG=0 + // quiets it to info). + ProxyVerbose bool + // LogColor selects ANSI coloring for the structured (slog) output + // (PG_LOG_COLOR): "auto" (default — color only on a terminal, honoring + // NO_COLOR/TERM), "always" to force it through a pipe, "never" to disable. + LogColor string + // LogFormat selects the structured sink (PG_LOG_FORMAT): "text" (default — + // colored, human-first) or "json" (stdlib slog.JSONHandler) for a captured + // run that gets machine-parsed instead of read. + LogFormat string // ProxyHostname is the host printed in psql/.pgpass lines (PG_PROXY_HOSTNAME). // Defaults to host.docker.internal so the endpoint is reachable both from the // Mac and from sibling containers/k3d; 127.0.0.1 also works host-only. ProxyHostname string - // ForwardBind is the address the host forwarder (internal/forward) binds its - // client listeners to (PG_FORWARD_BIND). Default 127.0.0.1 (loopback only, as - // socat did); set to 0.0.0.0 to reach the endpoints from sibling - // containers/k3d that can't hit the Mac's loopback — this exposes the - // dev-credentialed backend to every interface, so widen deliberately. - ForwardBind string + // ClientBind is the address the socat proxy binds its client listeners to + // (PG_CLIENT_BIND). Default 127.0.0.1 (loopback only); set to 0.0.0.0 to reach + // the endpoints from sibling containers/k3d that can't hit the Mac's loopback + // — this exposes the dev-credentialed backend to every interface, so widen + // deliberately. + ClientBind string BackendPrefix string // pg-dev ProxyName string // pg-proxy — LEGACY (single-machine in-machine proxy) @@ -134,8 +148,11 @@ func Load() Config { StagingPort: atoi(get("PG_STAGING_PORT", "5433")), ClientActivePort: atoi(get("PG_CLIENT_ACTIVE_PORT", "5442")), ClientStagingPort: atoi(get("PG_CLIENT_STAGING_PORT", "5443")), + ProxyVerbose: get("PG_PROXY_DEBUG", "1") != "0", ProxyHostname: get("PG_PROXY_HOSTNAME", "host.docker.internal"), - ForwardBind: get("PG_FORWARD_BIND", "127.0.0.1"), + LogColor: get("PG_LOG_COLOR", "auto"), + LogFormat: get("PG_LOG_FORMAT", "text"), + ClientBind: get("PG_CLIENT_BIND", "127.0.0.1"), BackendPrefix: get("PG_BACKEND_PREFIX", DefaultBackendPrefix), ProxyName: get("PG_PROXY_NAME", DefaultProxyName), BackendAIP: get("PG_BACKEND_A_IP", ""), @@ -214,20 +231,28 @@ func (c Config) ActiveMachinePath() string { // MachineIPPath is the host-side cache of a machine's drifting eth0 IP, one file // per machine (var/machine-ip-a, var/machine-ip-b), since the two leases drift -// independently and the forwarder must track both. +// independently. Mirrored from the tracking DB, which is the source of truth. func (c Config) MachineIPPath(slot string) string { return filepath.Join(c.RepoRoot, "var", "machine-ip-"+slot) } -// ForwardStatePath is where a running `pgdev forward serve` writes its live -// mapping + heartbeat (internal/forward.State), read by promote/status. -func (c Config) ForwardStatePath() string { - return filepath.Join(c.RepoRoot, "var", "forward-state.json") +// ----- SQLite tracking + socat proxy (doc/issues/0004) --------------------- + +// TrackDBPath is the SQLite machine-tracking database (internal/track), the +// host-side source of truth for the active pointer, machine IPs, and the socat +// proxy's reconciled targets. Host-only: never open it from inside a guest over +// virtiofs (SQLite over virtiofs corrupts). +func (c Config) TrackDBPath() string { return filepath.Join(c.RepoRoot, "var", "pgdev.db") } + +// ReconcileLockPath is the flock the socat reconciler serializes on, held OUTSIDE +// any SQLite transaction so a wedged launchctl can't brick the CLI. +func (c Config) ReconcileLockPath() string { + return filepath.Join(c.RepoRoot, "var", "reconcile.flock") } -// ForwardLogPath is the LaunchAgent's combined stdout/stderr log. -func (c Config) ForwardLogPath() string { - return filepath.Join(c.RepoRoot, "var", c.MachinePrefix+"-forward.log") +// SocatLogPath is one socat LaunchAgent's log (socat -d -d writes lifecycle here). +func (c Config) SocatLogPath(role string) string { + return filepath.Join(c.RepoRoot, "var", c.MachinePrefix+"-socat-"+role+".log") } // ClientPort returns the host client port for a role ("active"/"staging"). diff --git a/pgdev/internal/forward/forward.go b/pgdev/internal/forward/forward.go deleted file mode 100644 index 878ad9f..0000000 --- a/pgdev/internal/forward/forward.go +++ /dev/null @@ -1,340 +0,0 @@ -package forward - -import ( - "context" - "fmt" - "io" - "net" - "os" - "strconv" - "strings" - "sync" - "time" - - "pansen.me/pgdev/internal/activeslot" - "pansen.me/pgdev/internal/logx" -) - -// Options configures a Server. The path/lookup fields are what let the forwarder -// track the drifting IPs and the active pointer without a live `container` exec. -type Options struct { - Bind string // listen address, default 127.0.0.1 - ActivePort int // host client port for the active role (5442) - StagingPort int // host client port for the staging role (5443) - BackendPort int // port each machine serves its backend on (5432) - - ActiveMachinePath string // var/active-machine - MachineIPPath func(slot string) string // slot ("a"/"b") -> var/machine-ip- - StatePath string // var/forward-state.json (heartbeat + live mapping); "" disables - - PollInterval time.Duration // re-read cadence for the pointer + IP files (default 1s) - DialTimeout time.Duration // per-connection upstream dial timeout (default 5s) - Log logx.Func // progress/diagnostic sink (nil = discard) -} - -// liveConn is one relayed client session, tagged with the role and target it was -// established against so a re-point can selectively tear down the now-stale ones. -type liveConn struct { - role string - target string - client net.Conn - up net.Conn -} - -// Server owns the two client listeners and a mutex-guarded routing table it -// re-points from the watched state files. Listeners are bound once and never -// rebound; only the dial targets swap. -type Server struct { - opts Options - log logx.Func - - mu sync.RWMutex - targets map[string]string // role -> "ip:port" ("" = unroutable) - live map[int64]liveConn // established sessions, for drop-on-repoint - nextID int64 - lastDialLog map[string]time.Time // per-role dial-failure log throttle -} - -// New builds a Server, filling defaults. -func New(o Options) *Server { - if o.Bind == "" { - o.Bind = "127.0.0.1" - } - if o.PollInterval <= 0 { - o.PollInterval = time.Second - } - if o.DialTimeout <= 0 { - o.DialTimeout = 5 * time.Second - } - return &Server{ - opts: o, - log: logx.Or(o.Log), - targets: map[string]string{}, - live: map[int64]liveConn{}, - lastDialLog: map[string]time.Time{}, - } -} - -// role pairs a listener with the routing key it serves. -type role struct { - name string - port int -} - -// Serve binds both listeners, starts the poll-watcher, and forwards until ctx is -// cancelled. It returns after the listeners are closed and their accept loops -// have stopped. A bind failure on either port is fatal (returned) — unlike the -// old socat path there is nothing to orphan, so a clean error is the whole point. -func (s *Server) Serve(ctx context.Context) error { - s.reload() // seed the table before accepting, so the first conn already routes - - roles := []role{{"active", s.opts.ActivePort}, {"staging", s.opts.StagingPort}} - lns := make([]net.Listener, 0, len(roles)) - lc := net.ListenConfig{} - for _, r := range roles { - addr := net.JoinHostPort(s.opts.Bind, strconv.Itoa(r.port)) - ln, err := lc.Listen(ctx, "tcp", addr) - if err != nil { - for _, done := range lns { - done.Close() - } - return fmt.Errorf("listen %s (%s): %w", addr, r.name, err) - } - s.log("listening on %s (%s)", addr, r.name) - lns = append(lns, ln) - } - - var wg sync.WaitGroup - for i, r := range roles { - wg.Add(1) - go func(ln net.Listener, r role) { - defer wg.Done() - s.acceptLoop(ctx, ln, r.name) - }(lns[i], r) - } - - wg.Add(1) - go func() { - defer wg.Done() - s.watch(ctx) - }() - - <-ctx.Done() - for _, ln := range lns { - ln.Close() // unblocks the accept loops - } - s.closeAllLive() // drop established sessions so shutdown is prompt (ExitTimeOut) - wg.Wait() - return nil -} - -// watch re-reads the pointer + IP files on a ticker and swaps the routing table. -// Poll (not fsnotify) is deliberately the simplest thing that is robust on macOS -// — the files change rarely and a ~1s lag on promote is within the "promote may -// drop sessions; reconnect" contract. -func (s *Server) watch(ctx context.Context) { - t := time.NewTicker(s.opts.PollInterval) - defer t.Stop() - for { - select { - case <-ctx.Done(): - return - case <-t.C: - s.reload() - } - } -} - -// reload recomputes targets from the current on-disk state, swaps them in, and -// tears down any established session whose role target changed (drop-on-repoint): -// a client left on the demoted machine after a promote would otherwise keep -// reading/writing the wrong database — the exact hazard this component exists to -// remove. New connections use the new target; dropped clients simply reconnect. -func (s *Server) reload() { - slot := activeslot.Pointer{Path: s.opts.ActiveMachinePath}.Get() - ipA := readIP(s.opts.MachineIPPath("a")) - ipB := readIP(s.opts.MachineIPPath("b")) - active, staging := Targets(slot, ipA, ipB, s.opts.BackendPort) - - s.mu.Lock() - changed := active != s.targets["active"] || staging != s.targets["staging"] - s.targets["active"], s.targets["staging"] = active, staging - var drop []liveConn - for id, lc := range s.live { - want := active - if lc.role == "staging" { - want = staging - } - if lc.target != want { - drop = append(drop, lc) - delete(s.live, id) - } - } - s.mu.Unlock() - - for _, lc := range drop { - lc.client.Close() - lc.up.Close() - } - if changed { - s.log("routing updated: active -> %s, staging -> %s (dropped %d stale session(s))", - orNone(active), orNone(staging), len(drop)) - } - s.writeState(slot, active, staging) -} - -func (s *Server) target(role string) string { - s.mu.RLock() - defer s.mu.RUnlock() - return s.targets[role] -} - -// acceptLoop accepts on ln until ctx is done or the listener is closed; each -// accepted conn is handled on its own goroutine so a slow dial never stalls the -// listener. A transient Accept error (e.g. EMFILE, a momentary resource pinch) -// must NOT abandon the role for the process's lifetime — that would defeat the -// whole point of a resident forwarder — so it backs off and retries (the -// net/http.Server pattern), returning only on context cancellation. -func (s *Server) acceptLoop(ctx context.Context, ln net.Listener, role string) { - var backoff time.Duration - for { - conn, err := ln.Accept() - if err != nil { - select { - case <-ctx.Done(): - return // shutdown: the listener was closed on purpose - default: - } - if backoff == 0 { - backoff = 5 * time.Millisecond - } else { - backoff *= 2 - } - if backoff > time.Second { - backoff = time.Second - } - s.log("accept on %s: %v; retrying in %s", role, err, backoff) - select { - case <-ctx.Done(): - return - case <-time.After(backoff): - } - continue - } - backoff = 0 - go s.handle(conn, role) - } -} - -// handle relays one client connection to the current target for its role. The -// target is snapshotted at accept time and the session registered so a later -// re-point can drop it (see reload). -func (s *Server) handle(client net.Conn, role string) { - defer client.Close() - target := s.target(role) - if target == "" { - s.log("%s: no target (machine down?); closing %s", role, client.RemoteAddr()) - return - } - upstream, err := net.DialTimeout("tcp", target, s.opts.DialTimeout) - if err != nil { - s.logDialFailure(role, target, err) - return - } - defer upstream.Close() - keepAlive(client) - keepAlive(upstream) - - id := s.register(liveConn{role: role, target: target, client: client, up: upstream}) - defer s.unregister(id) - pipe(client, upstream) -} - -func (s *Server) register(lc liveConn) int64 { - s.mu.Lock() - defer s.mu.Unlock() - s.nextID++ - id := s.nextID - s.live[id] = lc - return id -} - -func (s *Server) unregister(id int64) { - s.mu.Lock() - defer s.mu.Unlock() - delete(s.live, id) -} - -func (s *Server) closeAllLive() { - s.mu.Lock() - drop := make([]liveConn, 0, len(s.live)) - for id, lc := range s.live { - drop = append(drop, lc) - delete(s.live, id) - } - s.mu.Unlock() - for _, lc := range drop { - lc.client.Close() - lc.up.Close() - } -} - -// logDialFailure rate-limits per-role dial errors: a down machine leaves a stale -// IP file, so a retrying client would otherwise flood the log every reconnect. -func (s *Server) logDialFailure(role, target string, err error) { - s.mu.Lock() - now := time.Now() - quiet := now.Sub(s.lastDialLog[role]) < 30*time.Second - if !quiet { - s.lastDialLog[role] = now - } - s.mu.Unlock() - if !quiet { - s.log("%s: dial %s: %v", role, target, err) - } -} - -// pipe copies bytes both ways and returns only once BOTH directions have ended. -// Each half half-closes the peer's write side (CloseWrite) on EOF so a one-way -// shutdown propagates as a FIN; waiting for both (not the first) avoids -// truncating a reply still streaming the other way (e.g. COPY OUT). -func pipe(a, b net.Conn) { - var wg sync.WaitGroup - wg.Add(2) - cp := func(dst, src net.Conn) { - defer wg.Done() - io.Copy(dst, src) - if cw, ok := dst.(interface{ CloseWrite() error }); ok { - cw.CloseWrite() - } - } - go cp(a, b) - go cp(b, a) - wg.Wait() -} - -// keepAlive turns on TCP keepalive so a vanished machine (IP gone, no RST) -// eventually surfaces as a read error instead of leaking the goroutine pair. -func keepAlive(c net.Conn) { - if t, ok := c.(*net.TCPConn); ok { - _ = t.SetKeepAlive(true) - _ = t.SetKeepAlivePeriod(30 * time.Second) - } -} - -// readIP returns the trimmed contents of an IP cache file, or "" if it is -// missing/empty (the machine is down or hasn't reported yet). -func readIP(path string) string { - b, err := os.ReadFile(path) - if err != nil { - return "" - } - return strings.TrimSpace(string(b)) -} - -func orNone(s string) string { - if s == "" { - return "(none)" - } - return s -} diff --git a/pgdev/internal/forward/forward_test.go b/pgdev/internal/forward/forward_test.go deleted file mode 100644 index 4d59eac..0000000 --- a/pgdev/internal/forward/forward_test.go +++ /dev/null @@ -1,228 +0,0 @@ -package forward - -import ( - "context" - "io" - "net" - "os" - "path/filepath" - "strconv" - "testing" - "time" -) - -// writeFile is a tiny helper for the var/ state files the Server watches. -func writeFile(t *testing.T, path, content string) { - t.Helper() - if err := os.WriteFile(path, []byte(content), 0o644); err != nil { - t.Fatal(err) - } -} - -// newTestServer wires a Server over a temp dir's state files, returning it plus -// the dir so tests can rewrite the pointer/IP files under it. -func newTestServer(t *testing.T, o Options) (*Server, string) { - t.Helper() - dir := t.TempDir() - o.ActiveMachinePath = filepath.Join(dir, "active-machine") - o.MachineIPPath = func(slot string) string { return filepath.Join(dir, "machine-ip-"+slot) } - return New(o), dir -} - -// TestReloadSwapsOnPointerFlip drives the file-watch computation end to end: a -// rewrite of the pointer file must swap both role targets on the next reload, -// with the listeners never involved. -func TestReloadSwapsOnPointerFlip(t *testing.T) { - s, dir := newTestServer(t, Options{BackendPort: 5432}) - writeFile(t, filepath.Join(dir, "machine-ip-a"), "10.0.0.1\n") - writeFile(t, filepath.Join(dir, "machine-ip-b"), "10.0.0.2\n") - writeFile(t, s.opts.ActiveMachinePath, "a\n") - - s.reload() - if got := s.target("active"); got != "10.0.0.1:5432" { - t.Fatalf("active = %q, want 10.0.0.1:5432", got) - } - if got := s.target("staging"); got != "10.0.0.2:5432" { - t.Fatalf("staging = %q, want 10.0.0.2:5432", got) - } - - writeFile(t, s.opts.ActiveMachinePath, "b\n") - s.reload() - if got := s.target("active"); got != "10.0.0.2:5432" { - t.Fatalf("after flip active = %q, want 10.0.0.2:5432", got) - } - if got := s.target("staging"); got != "10.0.0.1:5432" { - t.Fatalf("after flip staging = %q, want 10.0.0.1:5432", got) - } -} - -// TestReloadDropsStaleSessionsOnRepoint verifies the drop-on-repoint safety -// property: an established session on a role whose target changes (a promote) -// is torn down so the client can't keep talking to the demoted database, while -// a session on an unchanged role is left alone. -func TestReloadDropsStaleSessionsOnRepoint(t *testing.T) { - s, dir := newTestServer(t, Options{BackendPort: 5432}) - writeFile(t, filepath.Join(dir, "machine-ip-a"), "10.0.0.1\n") - writeFile(t, filepath.Join(dir, "machine-ip-b"), "10.0.0.2\n") - writeFile(t, s.opts.ActiveMachinePath, "a\n") - s.reload() - - // Register a fake session on each role at the current target. - activeCli, activeUp := net.Pipe() - stagingCli, stagingUp := net.Pipe() - s.register(liveConn{role: "active", target: s.target("active"), client: activeCli, up: activeUp}) - s.register(liveConn{role: "staging", target: s.target("staging"), client: stagingCli, up: stagingUp}) - - // Promote: flip the pointer. Both role targets change (a<->b swap), so BOTH - // sessions are stale and must be dropped. - writeFile(t, s.opts.ActiveMachinePath, "b\n") - s.reload() - - if !isClosed(activeCli) { - t.Error("active session was not dropped after promote") - } - if !isClosed(stagingCli) { - t.Error("staging session was not dropped after promote") - } - if n := s.liveCount(); n != 0 { - t.Errorf("live sessions after repoint = %d, want 0", n) - } -} - -// TestReloadKeepsUnaffectedSession leaves a role's target unchanged (a staging -// IP refresh that doesn't alter the active mapping) and asserts the active -// session survives. -func TestReloadKeepsUnaffectedSession(t *testing.T) { - s, dir := newTestServer(t, Options{BackendPort: 5432}) - writeFile(t, filepath.Join(dir, "machine-ip-a"), "10.0.0.1\n") - writeFile(t, filepath.Join(dir, "machine-ip-b"), "10.0.0.2\n") - writeFile(t, s.opts.ActiveMachinePath, "a\n") - s.reload() - - cli, up := net.Pipe() - s.register(liveConn{role: "active", target: s.target("active"), client: cli, up: up}) - - // Only staging's IP changes; active target is untouched. - writeFile(t, filepath.Join(dir, "machine-ip-b"), "10.0.0.99\n") - s.reload() - - if isClosed(cli) { - t.Error("active session was dropped despite its target being unchanged") - } -} - -func (s *Server) liveCount() int { - s.mu.RLock() - defer s.mu.RUnlock() - return len(s.live) -} - -// isClosed reports whether a net.Pipe end has been closed, via a short read. -func isClosed(c net.Conn) bool { - c.SetReadDeadline(time.Now().Add(200 * time.Millisecond)) - _, err := c.Read(make([]byte, 1)) - return err != nil && err != os.ErrDeadlineExceeded && !isTimeout(err) -} - -func isTimeout(err error) bool { - te, ok := err.(interface{ Timeout() bool }) - return ok && te.Timeout() -} - -// freePort returns a currently-free localhost TCP port. -func freePort(t *testing.T) int { - t.Helper() - ln, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatal(err) - } - defer ln.Close() - return ln.Addr().(*net.TCPAddr).Port -} - -// echoServer accepts one-shot connections and echoes bytes back until EOF. -func echoServer(t *testing.T) (addrPort int) { - t.Helper() - ln, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { ln.Close() }) - go func() { - for { - c, err := ln.Accept() - if err != nil { - return - } - go func() { io.Copy(c, c); c.Close() }() - } - }() - return ln.Addr().(*net.TCPAddr).Port -} - -// TestServeProxiesActiveAndClosesUnroutableStaging brings the real Server up on -// loopback and checks: (1) the active port relays to the backend; (2) the -// staging port, whose machine has no cached IP, accepts then immediately closes. -func TestServeProxiesActiveAndClosesUnroutableStaging(t *testing.T) { - backend := echoServer(t) - activePort, stagingPort := freePort(t), freePort(t) - - s, dir := newTestServer(t, Options{ - Bind: "127.0.0.1", - ActivePort: activePort, - StagingPort: stagingPort, - BackendPort: backend, - PollInterval: 20 * time.Millisecond, - }) - writeFile(t, filepath.Join(dir, "machine-ip-a"), "127.0.0.1\n") // active reachable - // machine-ip-b intentionally absent -> staging target empty - writeFile(t, s.opts.ActiveMachinePath, "a\n") - - ctx, cancel := context.WithCancel(context.Background()) - done := make(chan error, 1) - go func() { done <- s.Serve(ctx) }() - t.Cleanup(func() { - cancel() - <-done - }) - - // The active port relays to the echo backend. - activeConn := dialWithRetry(t, activePort) - defer activeConn.Close() - if _, err := activeConn.Write([]byte("ping")); err != nil { - t.Fatal(err) - } - buf := make([]byte, 4) - activeConn.SetReadDeadline(time.Now().Add(2 * time.Second)) - if _, err := io.ReadFull(activeConn, buf); err != nil { - t.Fatalf("active read: %v", err) - } - if string(buf) != "ping" { - t.Fatalf("active echo = %q, want ping", buf) - } - - // The staging port has no target: the conn is accepted then closed (read - // returns EOF with no data). - stagingConn := dialWithRetry(t, stagingPort) - defer stagingConn.Close() - stagingConn.SetReadDeadline(time.Now().Add(2 * time.Second)) - if n, err := stagingConn.Read(make([]byte, 1)); err == nil && n > 0 { - t.Fatalf("staging returned data (%d bytes); want immediate close", n) - } -} - -func dialWithRetry(t *testing.T, port int) net.Conn { - t.Helper() - addr := net.JoinHostPort("127.0.0.1", strconv.Itoa(port)) - deadline := time.Now().Add(2 * time.Second) - for { - c, err := net.Dial("tcp", addr) - if err == nil { - return c - } - if time.Now().After(deadline) { - t.Fatalf("dial %s: %v", addr, err) - } - time.Sleep(10 * time.Millisecond) - } -} diff --git a/pgdev/internal/forward/launchd.go b/pgdev/internal/forward/launchd.go deleted file mode 100644 index 9c6b0a8..0000000 --- a/pgdev/internal/forward/launchd.go +++ /dev/null @@ -1,338 +0,0 @@ -package forward - -import ( - "bytes" - "context" - "encoding/xml" - "fmt" - "os" - "os/exec" - "path/filepath" - "slices" - "strconv" - "strings" - "text/template" - "time" -) - -// Launchd manages the per-user LaunchAgent that keeps `pgdev forward serve` -// running across logins. Unlike the retired socat agent, this one NEVER needs -// restarting to re-point — the resident process swaps targets itself — so -// promote/refresh no longer touch launchd at all. The agent exists purely for -// persistence (RunAtLoad) and crash-restart (KeepAlive). -type Launchd struct { - Label string // me.pansen.-forward - Plist string // ~/Library/LaunchAgents/