From 61cb0cb12e7c7e57a483c161f8384f98c5358229 Mon Sep 17 00:00:00 2001 From: andi Date: Thu, 23 Jul 2026 19:48:12 +0200 Subject: [PATCH 1/6] logging --- .env.example | 7 + pgdev/cmd/pgdev/forward.go | 1 + pgdev/internal/config/config.go | 6 + pgdev/internal/forward/forward.go | 306 +++++++++++++++++++++---- pgdev/internal/forward/forward_test.go | 6 +- pgdev/internal/forward/logging_test.go | 70 ++++++ 6 files changed, 354 insertions(+), 42 deletions(-) create mode 100644 pgdev/internal/forward/logging_test.go diff --git a/.env.example b/.env.example index 9bd6f18..accdf47 100644 --- a/.env.example +++ b/.env.example @@ -40,6 +40,13 @@ MACHINE_CPUS=4 # interface (LAN/Wi-Fi), so widen deliberately. # PG_FORWARD_BIND=127.0.0.1 +# Optional: set to 1 for verbose forwarder logging — adds a per-poll DBG trace +# line (~1/s) on top of the always-on boot banner, per-connection lifecycle, +# classified dial failures, and 15s heartbeat in var/-forward.log. The +# forwarder reads this at process start, so `make endpoint.restart` after +# setting it. Leave off for normal use. +# PG_FORWARD_DEBUG=0 + # 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 # a large restore can fill macOS and shut a guest PostgreSQL store down mid-write. diff --git a/pgdev/cmd/pgdev/forward.go b/pgdev/cmd/pgdev/forward.go index a716610..401ea5e 100644 --- a/pgdev/cmd/pgdev/forward.go +++ b/pgdev/cmd/pgdev/forward.go @@ -44,6 +44,7 @@ func (a *app) forwardOptions() forward.Options { MachineIPPath: a.cfg.MachineIPPath, StatePath: a.cfg.ForwardStatePath(), Log: logf, + Verbose: a.cfg.ForwardVerbose, } } diff --git a/pgdev/internal/config/config.go b/pgdev/internal/config/config.go index d115aa3..1832659 100644 --- a/pgdev/internal/config/config.go +++ b/pgdev/internal/config/config.go @@ -71,6 +71,11 @@ type Config struct { // containers/k3d that can't hit the Mac's loopback — this exposes the // dev-credentialed backend to every interface, so widen deliberately. ForwardBind string + // ForwardVerbose enables the forwarder's per-poll DBG trace (PG_FORWARD_DEBUG=1) + // on top of its always-on lifecycle logging. Read from .env like the rest so + // the LaunchAgent (which loads config via PG_REPO_ROOT/.env, not the process + // env) actually honors it. + ForwardVerbose bool BackendPrefix string // pg-dev ProxyName string // pg-proxy — LEGACY (single-machine in-machine proxy) @@ -136,6 +141,7 @@ func Load() Config { ClientStagingPort: atoi(get("PG_CLIENT_STAGING_PORT", "5443")), ProxyHostname: get("PG_PROXY_HOSTNAME", "host.docker.internal"), ForwardBind: get("PG_FORWARD_BIND", "127.0.0.1"), + ForwardVerbose: get("PG_FORWARD_DEBUG", "") == "1", BackendPrefix: get("PG_BACKEND_PREFIX", DefaultBackendPrefix), ProxyName: get("PG_PROXY_NAME", DefaultProxyName), BackendAIP: get("PG_BACKEND_A_IP", ""), diff --git a/pgdev/internal/forward/forward.go b/pgdev/internal/forward/forward.go index 878ad9f..1fb8edf 100644 --- a/pgdev/internal/forward/forward.go +++ b/pgdev/internal/forward/forward.go @@ -2,6 +2,7 @@ package forward import ( "context" + "errors" "fmt" "io" "net" @@ -9,6 +10,7 @@ import ( "strconv" "strings" "sync" + "syscall" "time" "pansen.me/pgdev/internal/activeslot" @@ -30,6 +32,7 @@ type Options struct { 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) + Verbose bool // extra per-poll DBG tracing (PG_FORWARD_DEBUG=1) } // liveConn is one relayed client session, tagged with the role and target it was @@ -48,11 +51,24 @@ 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 + mu sync.RWMutex + targets map[string]string // role -> "ip:port" ("" = unroutable) + live map[int64]liveConn // established sessions, for drop-on-repoint + nextID int64 // monotonic connection id (logging + live-map key) + dial map[string]*dialHealth // per-role dial outcome (health summary + log throttle) + + started time.Time // process start, for uptime in the heartbeat +} + +// dialHealth tracks the outcome of upstream dials for one role, so the heartbeat +// can report "active: ok, last 3s ago" vs "staging: FAILING 12x EHOSTUNREACH" +// and so repeated identical failures are throttled instead of flooding the log. +type dialHealth struct { + lastOK time.Time // last successful dial + lastErr time.Time // last failed dial + lastClass string // classifyDial() bucket of the last failure + fails int // consecutive failures since the last success + loggedAt time.Time // last failure line we actually emitted (throttle anchor) } // New builds a Server, filling defaults. @@ -67,11 +83,36 @@ func New(o Options) *Server { 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{}, + opts: o, + log: logx.Or(o.Log), + targets: map[string]string{}, + live: map[int64]liveConn{}, + dial: map[string]*dialHealth{}, + } +} + +// ----- logging --------------------------------------------------------------- +// +// Every forwarder line carries a millisecond timestamp and a level tag, because +// this log is the primary forensic record when the endpoint mysteriously stops +// working (see the macOS Local Network Privacy trap). "was it ever working, and +// exactly when did it flip?" must be answerable from this file alone. Format +// strings are compile-time literals; values ride in args. + +func (s *Server) logl(level, format string, args ...any) { + stamp := time.Now().Format("2006-01-02 15:04:05.000") + s.log("%s %-4s "+format, append([]any{stamp, level}, args...)...) +} + +func (s *Server) info(format string, args ...any) { s.logl("INFO", format, args...) } +func (s *Server) warn(format string, args ...any) { s.logl("WARN", format, args...) } + +// debug lines (per-poll target recomputation) are silent unless Verbose — they +// are the "rather too much" tier, one line per second, opt-in via +// PG_FORWARD_DEBUG=1. +func (s *Server) debug(format string, args ...any) { + if s.opts.Verbose { + s.logl("DBG", format, args...) } } @@ -86,6 +127,8 @@ type role struct { // 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.started = time.Now() + s.logBoot() s.reload() // seed the table before accepting, so the first conn already routes roles := []role{{"active", s.opts.ActivePort}, {"staging", s.opts.StagingPort}} @@ -95,12 +138,13 @@ func (s *Server) Serve(ctx context.Context) error { addr := net.JoinHostPort(s.opts.Bind, strconv.Itoa(r.port)) ln, err := lc.Listen(ctx, "tcp", addr) if err != nil { + s.warn("FATAL: cannot bind %s (%s): %v — is another forwarder/socat holding the port?", addr, r.name, err) 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) + s.info("listening on %s (%s)", addr, r.name) lns = append(lns, ln) } @@ -119,7 +163,15 @@ func (s *Server) Serve(ctx context.Context) error { s.watch(ctx) }() + wg.Add(1) + go func() { + defer wg.Done() + s.heartbeat(ctx) + }() + <-ctx.Done() + s.info("shutdown: signal received, closing %d listener(s) and %d live session(s) (up %s)", + len(lns), s.liveN(), time.Since(s.started).Round(time.Second)) for _, ln := range lns { ln.Close() // unblocks the accept loops } @@ -128,6 +180,61 @@ func (s *Server) Serve(ctx context.Context) error { return nil } +// logBoot dumps the full effective configuration and the initial resolved +// mapping (with the IP-file provenance) at startup — the "what did this process +// see when it came up" record every incident starts from. +func (s *Server) logBoot() { + s.info("boot: pgdev forward serve pid=%d bind=%s active=:%d staging=:%d backend=:%d poll=%s dial-timeout=%s verbose=%t", + os.Getpid(), s.opts.Bind, s.opts.ActivePort, s.opts.StagingPort, s.opts.BackendPort, + s.opts.PollInterval, s.opts.DialTimeout, s.opts.Verbose) + s.info("boot: watching pointer=%s ip-a=%s ip-b=%s state=%s", + s.opts.ActiveMachinePath, s.opts.MachineIPPath("a"), s.opts.MachineIPPath("b"), orNone(s.opts.StatePath)) + slot := activeslot.Pointer{Path: s.opts.ActiveMachinePath}.Get() + ipA, ipB := readIP(s.opts.MachineIPPath("a")), readIP(s.opts.MachineIPPath("b")) + active, staging := Targets(slot, ipA, ipB, s.opts.BackendPort) + s.info("boot: initial active-slot=%s ip-a=%s ip-b=%s -> active=%s staging=%s", + orNone(slot), orNone(ipA), orNone(ipB), orNone(active), orNone(staging)) + if !isLoopbackBind(s.opts.Bind) { + s.warn("boot: bind=%s is NOT loopback — the dev-credentialed backend is exposed on every interface", s.opts.Bind) + } +} + +// heartbeat emits one summary line on a slow ticker: uptime, live session count, +// and each role's current target plus its dial health. This is the line that +// answers "was it working, and when did it stop" long after the event — so it +// runs even when idle (deliberately chatty; that is the point). +func (s *Server) heartbeat(ctx context.Context) { + t := time.NewTicker(15 * time.Second) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + s.mu.RLock() + active, staging := s.targets["active"], s.targets["staging"] + live := len(s.live) + as, ss := s.dialSummaryLocked("active"), s.dialSummaryLocked("staging") + s.mu.RUnlock() + s.info("beat: uptime=%s live=%d | active=%s [%s] | staging=%s [%s]", + time.Since(s.started).Round(time.Second), live, orNone(active), as, orNone(staging), ss) + } + } +} + +// dialSummaryLocked renders one role's dial health for the heartbeat. Caller +// holds s.mu (read is fine; the recorders take the write lock). +func (s *Server) dialSummaryLocked(role string) string { + h := s.dial[role] + if h == nil || (h.lastOK.IsZero() && h.fails == 0) { + return "no dials yet" + } + if h.fails > 0 { + return fmt.Sprintf("FAILING %dx %s, last %s ago", h.fails, h.lastClass, time.Since(h.lastErr).Round(time.Second)) + } + return fmt.Sprintf("ok, last %s ago", time.Since(h.lastOK).Round(time.Second)) +} + // 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 @@ -177,12 +284,30 @@ func (s *Server) reload() { 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.info("route: active -> %s, staging -> %s (slot=%s ip-a=%s ip-b=%s; dropped %d stale session(s))", + orNone(active), orNone(staging), orNone(slot), orNone(ipA), orNone(ipB), len(drop)) + } else { + s.debug("poll: active=%s staging=%s (slot=%s ip-a=%s ip-b=%s)", + orNone(active), orNone(staging), orNone(slot), orNone(ipA), orNone(ipB)) } s.writeState(slot, active, staging) } +// liveN returns the current live-session count (own lock). +func (s *Server) liveN() int { + s.mu.RLock() + defer s.mu.RUnlock() + return len(s.live) +} + +func isLoopbackBind(bind string) bool { + switch bind { + case "", "127.0.0.1", "::1", "localhost": + return true + } + return false +} + func (s *Server) target(role string) string { s.mu.RLock() defer s.mu.RUnlock() @@ -213,7 +338,7 @@ func (s *Server) acceptLoop(ctx context.Context, ln net.Listener, role string) { if backoff > time.Second { backoff = time.Second } - s.log("accept on %s: %v; retrying in %s", role, err, backoff) + s.warn("accept on %s: %v; retrying in %s", role, err, backoff) select { case <-ctx.Done(): return @@ -228,35 +353,52 @@ func (s *Server) acceptLoop(ctx context.Context, ln net.Listener, role string) { // 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). +// re-point can drop it (see reload). Every phase (accept, dial, close) is logged +// with the connection id so a single session can be followed end to end. func (s *Server) handle(client net.Conn, role string) { + id := s.nextConnID() + remote := client.RemoteAddr() defer client.Close() + target := s.target(role) if target == "" { - s.log("%s: no target (machine down?); closing %s", role, client.RemoteAddr()) + s.warn("conn#%d %s: no target — machine down or IP file empty; closing client %s", id, role, remote) return } + s.debug("conn#%d %s: accept from %s, dialing %s", id, role, remote, target) + + t0 := time.Now() upstream, err := net.DialTimeout("tcp", target, s.opts.DialTimeout) if err != nil { - s.logDialFailure(role, target, err) + s.recordDialFail(id, role, target, err, time.Since(t0)) return } + s.recordDialOK(id, role, target, time.Since(t0)) defer upstream.Close() keepAlive(client) keepAlive(upstream) - id := s.register(liveConn{role: role, target: target, client: client, up: upstream}) + s.register(id, liveConn{role: role, target: target, client: client, up: upstream}) defer s.unregister(id) - pipe(client, upstream) + + up, down := pipe(client, upstream) + s.info("conn#%d %s: closed after %s (client→up %s, up→client %s) target=%s", + id, role, time.Since(t0).Round(time.Millisecond), humanBytes(up), humanBytes(down), target) } -func (s *Server) register(lc liveConn) int64 { +// nextConnID hands out a monotonic id used both for log correlation and as the +// live-session map key. +func (s *Server) nextConnID() int64 { s.mu.Lock() defer s.mu.Unlock() s.nextID++ - id := s.nextID + return s.nextID +} + +func (s *Server) register(id int64, lc liveConn) { + s.mu.Lock() + defer s.mu.Unlock() s.live[id] = lc - return id } func (s *Server) unregister(id int64) { @@ -279,38 +421,124 @@ func (s *Server) closeAllLive() { } } -// 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) { +// recordDialOK marks a successful dial and, if the role had been failing, logs a +// recovery line naming how many attempts it took — the "it started working +// again at HH:MM:SS" bookend to a failure streak. +func (s *Server) recordDialOK(id int64, role, target string, took time.Duration) { + s.mu.Lock() + h := s.dialHealthLocked(role) + recovered := h.fails + h.lastOK, h.fails, h.lastClass = time.Now(), 0, "" + s.mu.Unlock() + + if recovered > 0 { + s.info("conn#%d %s: dial %s OK in %s — RECOVERED after %d consecutive failure(s)", + id, role, target, took.Round(time.Millisecond), recovered) + return + } + s.debug("conn#%d %s: dial %s OK in %s", id, role, target, took.Round(time.Millisecond)) +} + +// recordDialFail classifies the error, updates the role's health, and logs it. +// The first failure of a streak (or a change of error class) is always logged +// loudly WITH a remediation hint; identical repeats are throttled to one line +// per 10s (with a running count) so a hammering client can't flood the file but +// the failure is never invisible. +func (s *Server) recordDialFail(id int64, role, target string, err error, took time.Duration) { + class, hint := classifyDial(err) + s.mu.Lock() - now := time.Now() - quiet := now.Sub(s.lastDialLog[role]) < 30*time.Second + h := s.dialHealthLocked(role) + h.lastErr = time.Now() + h.fails++ + firstOfStreak := h.fails == 1 || h.lastClass != class + h.lastClass = class + quiet := !firstOfStreak && time.Since(h.loggedAt) < 10*time.Second if !quiet { - s.lastDialLog[role] = now + h.loggedAt = time.Now() } + fails := h.fails s.mu.Unlock() - if !quiet { - s.log("%s: dial %s: %v", role, target, err) + + if quiet { + return + } + s.warn("conn#%d %s: dial %s FAILED after %s: %v [%s] (%d in a row)", + id, role, target, took.Round(time.Millisecond), err, class, fails) + if hint != "" && firstOfStreak { + s.warn("conn#%d %s: hint: %s", id, role, hint) + } +} + +func (s *Server) dialHealthLocked(role string) *dialHealth { + h := s.dial[role] + if h == nil { + h = &dialHealth{} + s.dial[role] = h } + return h } -// 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) { +// classifyDial buckets an upstream dial error and returns a one-line remediation +// hint. The EHOSTUNREACH case is the star: it is the macOS Local Network Privacy +// signature (this process is denied the VM subnet even though a terminal dial to +// the same IP works, because CLI tools inherit Terminal's grant). +func classifyDial(err error) (class, hint string) { + switch { + case errors.Is(err, syscall.EHOSTUNREACH): + return "EHOSTUNREACH", "macOS Local Network Privacy is almost certainly blocking THIS process from the VM subnet " + + "(a plain `nc`/psql from a terminal can still reach it — CLI tools inherit Terminal's grant). " + + "Grant Local Network to pgdev in System Settings → Privacy & Security, then `make endpoint.restart`." + case errors.Is(err, syscall.ECONNREFUSED): + return "ECONNREFUSED", "the VM is reachable but nothing is listening on the backend port — Postgres/the container is likely down on that machine (`make status`)." + case errors.Is(err, syscall.ENETUNREACH): + return "ENETUNREACH", "the host has no route to the VM subnet — bridge100/vmnet is down or the machine never booted (`make start`)." + case isDialTimeout(err): + return "timeout", "the target IP did not answer within the dial timeout — a stale IP after recreate, or a wedged VM (`make status` to re-check the IP)." + default: + return "other", "" + } +} + +func isDialTimeout(err error) bool { + var ne net.Error + return errors.As(err, &ne) && ne.Timeout() +} + +// pipe copies bytes both ways and returns only once BOTH directions have ended, +// reporting the bytes moved each way (client→upstream, upstream→client) for the +// close log. 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(client, upstream net.Conn) (clientToUp, upToClient int64) { var wg sync.WaitGroup wg.Add(2) - cp := func(dst, src net.Conn) { + cp := func(dst, src net.Conn, n *int64) { defer wg.Done() - io.Copy(dst, src) + c, _ := io.Copy(dst, src) + *n = c if cw, ok := dst.(interface{ CloseWrite() error }); ok { cw.CloseWrite() } } - go cp(a, b) - go cp(b, a) + go cp(upstream, client, &clientToUp) + go cp(client, upstream, &upToClient) wg.Wait() + return clientToUp, upToClient +} + +// humanBytes renders a byte count compactly for the per-connection close line. +func humanBytes(n int64) string { + const unit = 1024 + if n < unit { + return fmt.Sprintf("%dB", n) + } + div, exp := int64(unit), 0 + for x := n / unit; x >= unit; x /= unit { + div *= unit + exp++ + } + return fmt.Sprintf("%.1f%cB", float64(n)/float64(div), "KMGTPE"[exp]) } // keepAlive turns on TCP keepalive so a vanished machine (IP gone, no RST) diff --git a/pgdev/internal/forward/forward_test.go b/pgdev/internal/forward/forward_test.go index 4d59eac..22526a0 100644 --- a/pgdev/internal/forward/forward_test.go +++ b/pgdev/internal/forward/forward_test.go @@ -70,8 +70,8 @@ func TestReloadDropsStaleSessionsOnRepoint(t *testing.T) { // 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}) + s.register(s.nextConnID(), liveConn{role: "active", target: s.target("active"), client: activeCli, up: activeUp}) + s.register(s.nextConnID(), 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. @@ -100,7 +100,7 @@ func TestReloadKeepsUnaffectedSession(t *testing.T) { s.reload() cli, up := net.Pipe() - s.register(liveConn{role: "active", target: s.target("active"), client: cli, up: up}) + s.register(s.nextConnID(), 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") diff --git a/pgdev/internal/forward/logging_test.go b/pgdev/internal/forward/logging_test.go new file mode 100644 index 0000000..d0ad7d3 --- /dev/null +++ b/pgdev/internal/forward/logging_test.go @@ -0,0 +1,70 @@ +package forward + +import ( + "fmt" + "net" + "strings" + "syscall" + "testing" +) + +// TestClassifyDial locks the error buckets and, crucially, that the EHOSTUNREACH +// case surfaces the macOS Local Network Privacy remediation — the whole reason +// this classification exists (that failure otherwise reads as an inscrutable +// "no route to host" while a terminal dial to the same IP works). +func TestClassifyDial(t *testing.T) { + cases := []struct { + name string + err error + wantClass string + hintSubstr string // "" = expect no hint + }{ + {"ehostunreach", wrapDial(syscall.EHOSTUNREACH), "EHOSTUNREACH", "Local Network"}, + {"econnrefused", wrapDial(syscall.ECONNREFUSED), "ECONNREFUSED", "nothing is listening"}, + {"enetunreach", wrapDial(syscall.ENETUNREACH), "ENETUNREACH", "no route to the VM subnet"}, + {"timeout", timeoutErr{}, "timeout", "dial timeout"}, + {"other", fmt.Errorf("boom"), "other", ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + class, hint := classifyDial(tc.err) + if class != tc.wantClass { + t.Fatalf("class = %q, want %q", class, tc.wantClass) + } + if tc.hintSubstr == "" { + if hint != "" { + t.Fatalf("expected no hint, got %q", hint) + } + return + } + if !strings.Contains(hint, tc.hintSubstr) { + t.Fatalf("hint %q does not mention %q", hint, tc.hintSubstr) + } + }) + } +} + +// wrapDial mirrors how the kernel error reaches us — inside a *net.OpError, the +// way net.DialTimeout returns it — to prove errors.Is unwraps through it. +func wrapDial(sys syscall.Errno) error { + return &net.OpError{Op: "dial", Net: "tcp", Err: &net.OpError{Op: "connect", Err: sys}} +} + +type timeoutErr struct{} + +func (timeoutErr) Error() string { return "i/o timeout" } +func (timeoutErr) Timeout() bool { return true } +func (timeoutErr) Temporary() bool { return true } + +func TestHumanBytes(t *testing.T) { + for _, tc := range []struct { + n int64 + want string + }{ + {0, "0B"}, {512, "512B"}, {1024, "1.0KB"}, {1536, "1.5KB"}, {1048576, "1.0MB"}, + } { + if got := humanBytes(tc.n); got != tc.want { + t.Errorf("humanBytes(%d) = %q, want %q", tc.n, got, tc.want) + } + } +} From 3934e5418bf81219850c52973a1c32bec545c2ee Mon Sep 17 00:00:00 2001 From: andi Date: Sat, 25 Jul 2026 10:32:36 +0200 Subject: [PATCH 2/6] Add SQLite-based proxy management - Replaces loose state files with `var/pgdev.db` (SQLite) for atomic, transactional machine tracking. - Implements `internal/socatproxy` using `launchd` and `socat` to serve canonical client ports (5442/5443). - Moves the Go forwarder to ports 5444/5445 to run alongside the socat proxy during the integration phase. - Adds `pgdev proxy` commands for managing the new infrastructure. - Hardens socat reloads with port-free gates and argv verification to prevent stale-mapping bugs. --- .env.example | 20 +- Makefile | 59 ++- doc/issues/0004-sqlite-socat-proxy.md | 175 ++++++++ pgdev/cmd/pgdev/commands.go | 104 ++++- pgdev/cmd/pgdev/deploy.go | 2 +- pgdev/cmd/pgdev/forward.go | 10 +- pgdev/cmd/pgdev/main.go | 73 +++- pgdev/cmd/pgdev/proxy.go | 299 ++++++++++++++ pgdev/go.mod | 14 +- pgdev/go.sum | 21 + pgdev/internal/config/config.go | 110 +++-- pgdev/internal/forward/forward.go | 4 +- pgdev/internal/forward/router.go | 8 +- pgdev/internal/socatproxy/launchd.go | 413 +++++++++++++++++++ pgdev/internal/socatproxy/reconcile.go | 257 ++++++++++++ pgdev/internal/socatproxy/socatproxy_test.go | 181 ++++++++ pgdev/internal/track/state.go | 329 +++++++++++++++ pgdev/internal/track/track.go | 293 +++++++++++++ pgdev/internal/track/track_test.go | 245 +++++++++++ 19 files changed, 2545 insertions(+), 72 deletions(-) create mode 100644 doc/issues/0004-sqlite-socat-proxy.md create mode 100644 pgdev/cmd/pgdev/proxy.go create mode 100644 pgdev/internal/socatproxy/launchd.go create mode 100644 pgdev/internal/socatproxy/reconcile.go create mode 100644 pgdev/internal/socatproxy/socatproxy_test.go create mode 100644 pgdev/internal/track/state.go create mode 100644 pgdev/internal/track/track.go create mode 100644 pgdev/internal/track/track_test.go diff --git a/.env.example b/.env.example index accdf47..3e243b1 100644 --- a/.env.example +++ b/.env.example @@ -27,13 +27,27 @@ 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 CANONICAL client-facing loopback ports — what local clients +# connect to (psql/.pgpass/status print these). Offset from 5432 so a local +# PostgreSQL isn't shadowed. Since spec 0004 the socat proxy binds THESE, so +# existing configs pick socat up transparently. 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 (doc/issues/0004): where the Go forwarder (internal/forward) now +# listens — MOVED off the canonical pair onto this second pair so it runs +# alongside socat during the integration phase. `make endpoint.install` manages +# it; `make proxy.install` brings up socat on the canonical ports above. +# PG_FORWARD_ACTIVE_PORT=5444 +# PG_FORWARD_STAGING_PORT=5445 + +# Optional: verbose (debug-level) slog output for the SQLite tracking DB and the +# socat proxy. ON by default (the experiment is chatty on purpose); set 0 to quiet +# it to info-level. +# PG_PROXY_DEBUG=1 + # 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 diff --git a/Makefile b/Makefile index 48126e7..177cffa 100644 --- a/Makefile +++ b/Makefile @@ -245,8 +245,10 @@ status: machine.exists pgdevd # ----- 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 -# 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 +# permanent 127.0.0.1:5444 (active) / :5445 (staging) endpoints and relays each +# to whichever machine currently holds that role (on its own eth0:5432). Since +# spec 0004 it MOVED off the canonical 5442/5443 (now served by the socat proxy, +# see proxy.* below) onto this second pair, and runs alongside it. 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 @@ -275,6 +277,42 @@ endpoint.uninstall: endpoint.status: @$(PGDEV) forward status +# ----- socat client proxy on the CANONICAL ports (doc/issues/0004) -------- +# A socat-based client proxy on 127.0.0.1:5442 (active) / :5443 (staging) — the +# CANONICAL client ports, so existing external configs pick it up with no +# change. It runs ALONGSIDE the Go forwarder (now moved to 5444/5445) during an +# integration phase. Homebrew's socat does not trip macOS Local Network Privacy +# the way the Go forwarder's own binary does (no codesign ceremony, no recurring +# prompts), so this routes clients through it. 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. Opt-in: nothing +# here runs, and promote/refresh stay hands-off, until `proxy.install`. +# +# `proxy.install` is the ONE reference command for "install anything proxy": it +# brings up BOTH the socat proxy (canonical 5442/5443) and the Go forwarder +# (5444/5445), so `endpoint.install` above is no longer a separate step to +# remember (it still works for forwarder-only). + +# 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: @for slot in a b; do \ @@ -383,10 +421,25 @@ 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. .PHONY: pg.staging.rebuild -pg.staging.rebuild: disk.check machine.exists pgdevd +pg.staging.rebuild: machine.exists 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. +.PHONY: pg.staging.purge +pg.staging.purge: machine.exists 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/doc/issues/0004-sqlite-socat-proxy.md b/doc/issues/0004-sqlite-socat-proxy.md new file mode 100644 index 0000000..3c3d3ee --- /dev/null +++ b/doc/issues/0004-sqlite-socat-proxy.md @@ -0,0 +1,175 @@ +# Spec 0004 — SQLite machine tracking + socat client proxy (retire the Go forwarder) + +Status: **in progress** (2026-07-24) · Owner: andi · Relates to: +`0002-two-machine-disk-reclaim.md` (two-machine model), +`0003-internal-forward.md` (the Go forwarder this proposes to retire). Memory: +`forwarder-local-network-privacy`, `pgdevd-token-home-mount-cold-cache`, +`apple-apiserver-wedged-recovery`. + +This is a **living document**: it records the design as it is built and tracks +the eventual removal of the Go forwarder. Keep it current as the trial proceeds. + +--- + +## 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 occasional permission prompts (`make build` strips the identifier; see +memory `forwarder-local-network-privacy`). Homebrew's `socat` — an already-known, +already-trusted binary — does **not** hit this. + +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 takes over the **canonical client ports** (5442 active / 5443 +staging), so existing external configs pick it up with no change; the Go +forwarder is **moved to a second pair** (5444 / 5445) and runs alongside it, so +the two can be compared before the forwarder is removed. + +## 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 legacy flat file** the still-running Go +forwarder polls. This is deliberate: rebuilding the Go forwarder to read SQLite +would re-trigger the exact codesign/TCC ceremony we are trying to escape, so we +keep its binary byte-identical during the trial. The mirrors die with the +forwarder (§5). + +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 **canonical 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 (an ERROR — launchd leaked), re-probe. + 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 reference command for "install anything proxy"**: +it brings up BOTH the socat proxy (canonical 5442/5443) and the Go forwarder +(5444/5445, best-effort, honoring `PG_ENDPOINT_AUTOINSTALL=0`), so there is no +separate `endpoint.install` to remember. + +It also **frees the canonical ports as part of install**: step 1 does a FULL +(re)install of the Go forwarder (bootout + bootstrap), which boots any pre-swap +forwarder off 5442/5443 (where a process that started before the swap is still +bound, and which a gentle `Ensure` would leave alone since its argv is unchanged) +and brings it back on 5444/5445 — so socat can bind the canonical ports in step +2. Killing that forwarder's PID would not suffice: it has `KeepAlive`, so launchd +resurrects it; only booting out the *job* frees the port. (A genuinely foreign +holder still hard-fails the gate loudly — we do not kill arbitrary processes.) + +Opt-in: `promote`/`refresh` stay hands-off (never touch launchd) until +`proxy install` has created a plist (`Reconciler.AnyInstalled`). + +## 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` does not surface socat** (per the brief — still an experiment). +- **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 / to validate during the trial + +- [ ] Does socat under a LaunchAgent avoid the Local Network prompt in practice, + or does TCC attribute the connection to the launchd job's binary anyway? + **This is the whole hypothesis — validate first.** Because socat now holds + the canonical 5442/5443, `make proxy.install` puts it on the client path + directly — running any existing client against 5442/5443 exercises it. +- [ ] Since socat serves the canonical ports, `make endpoint.install` (the Go + forwarder, now 5444/5445) and `make proxy.install` (socat, 5442/5443) can + both run; make sure they never both try to bind the same port. +- [ ] 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 plan for the Go forwarder (the end state) + +Socat already holds the canonical client ports, so removal is now purely a +deletion — no client repoint needed: + +1. Delete `internal/forward` and `pgdev forward *` + the `endpoint.*` make + targets that wrap it (and free `PG_FORWARD_ACTIVE_PORT`/`STAGING_PORT`). +2. Delete the **legacy flat-file mirrors** in `internal/track` (`ActiveMirror`, + `IPMirror`) and `activeslot` file reads — the DB becomes the sole store, so + "no plain text/JSON tracking anymore" is fully realized (`var/active-machine`, + `var/machine-ip-{a,b}`, `var/forward-state.json` all deleted). +3. Move `pgdev status`'s forwarder section onto the socat proxy. +4. Update README + `.env.example` (drop `PG_FORWARD_*`, keep `PG_PROXY_*`). + +Until then, both paths coexist and the mirrors keep the Go forwarder correct. diff --git a/pgdev/cmd/pgdev/commands.go b/pgdev/cmd/pgdev/commands.go index 2c4e34f..e8533ee 100644 --- a/pgdev/cmd/pgdev/commands.go +++ b/pgdev/cmd/pgdev/commands.go @@ -41,7 +41,7 @@ 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 } } @@ -239,7 +239,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) } @@ -289,14 +289,19 @@ func (a *app) promoteCmd() *cobra.Command { // 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 { + if err := a.setActive(ctx, to); err != nil { return err } + // socat now serves the CANONICAL client ports (5442/5443), so re-point + // it synchronously (DB-driven, flock-guarded, verified) if installed — + // this is the client-facing path now. The Go forwarder re-points itself + // from the mirrored pointer file on its own ports (5444/5445). + a.reconcileProxyIfInstalled(ctx, "promote") 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"+ + "WARNING: the Go 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) + a.cfg.MachineNameForSlot(to), a.cfg.ForwardActivePort) } fmt.Printf("Promoted. active=%s staging=%s\n\n", @@ -317,7 +322,7 @@ func (a *app) refreshCmd() *cobra.Command { 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) continue @@ -340,6 +345,9 @@ func (a *app) refreshCmd() *cobra.Command { // 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) + // Re-point the socat experiment at the freshly-discovered IPs too + // (only if it's installed; socat can't re-point itself). + a.reconcileProxyIfInstalled(ctx, "refresh") 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())) @@ -470,6 +478,7 @@ func (a *app) stagingCmd() *cobra.Command { reset, a.stagingStartCmd(), a.stagingStopCmd(), + a.stagingPurgeCmd(), a.stagingRebuildCmd(), ) return c @@ -520,6 +529,77 @@ func (a *app) stagingStopCmd() *cobra.Command { } } +// 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) + fmt.Printf("==> [%s] Stopping and deleting (this reclaims its macOS disk)...\n", 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. The Go + // forwarder likewise sees the mirror gone and marks staging unroutable. + 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") + + fmt.Printf("==> Purged. %s is deleted and its macOS disk reclaimed; %s (active) was never touched.\n", machine, 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) diff --git a/pgdev/cmd/pgdev/deploy.go b/pgdev/cmd/pgdev/deploy.go index db01386..a9e3358 100644 --- a/pgdev/cmd/pgdev/deploy.go +++ b/pgdev/cmd/pgdev/deploy.go @@ -170,7 +170,7 @@ func (a *app) deploy(ctx context.Context, slot string) error { // before the handshake dials the daemon (this also re-points the endpoint // forwarder at the new IP once refreshForwarder next runs). 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) diff --git a/pgdev/cmd/pgdev/forward.go b/pgdev/cmd/pgdev/forward.go index 401ea5e..3ce5f22 100644 --- a/pgdev/cmd/pgdev/forward.go +++ b/pgdev/cmd/pgdev/forward.go @@ -37,8 +37,8 @@ func (a *app) forwardCmd() *cobra.Command { func (a *app) forwardOptions() forward.Options { return forward.Options{ Bind: a.cfg.ForwardBind, - ActivePort: a.cfg.ClientActivePort, - StagingPort: a.cfg.ClientStagingPort, + ActivePort: a.cfg.ForwardActivePort, + StagingPort: a.cfg.ForwardStagingPort, BackendPort: a.cfg.BackendPort, ActiveMachinePath: a.cfg.ActiveMachinePath(), MachineIPPath: a.cfg.MachineIPPath, @@ -59,7 +59,7 @@ func (a *app) launchd() (*forward.Launchd, error) { // 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} + ld.ReapPorts = []int{a.cfg.ForwardActivePort, a.cfg.ForwardStagingPort} return ld, nil } @@ -94,8 +94,8 @@ func (a *app) forwardInstallCmd() *cobra.Command { 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) + fmt.Printf(" active 127.0.0.1:%d staging 127.0.0.1:%d (re-points itself on promote; socat now serves the canonical %d/%d)\n", + a.cfg.ForwardActivePort, a.cfg.ForwardStagingPort, a.cfg.ClientActivePort, a.cfg.ClientStagingPort) return nil }, } diff --git a/pgdev/cmd/pgdev/main.go b/pgdev/cmd/pgdev/main.go index 49a9de6..b835d60 100644 --- a/pgdev/cmd/pgdev/main.go +++ b/pgdev/cmd/pgdev/main.go @@ -17,9 +17,9 @@ package main import ( "context" "fmt" + "log/slog" "os" "os/signal" - "path/filepath" "strings" "syscall" "time" @@ -31,6 +31,7 @@ import ( "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 +47,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 +69,7 @@ func newApp() *app { return &app{ cfg: cfg, active: activeslot.Pointer{Path: cfg.ActiveMachinePath(), UID: cfg.HostUID, GID: cfg.HostGID}, + log: newLogger(cfg.ProxyVerbose), } } @@ -85,6 +93,7 @@ func rootCmd() *cobra.Command { a.ipCmd(), a.endpointCmd(), a.forwardCmd(), + a.proxyCmd(), a.snapshotCmd("active"), a.restoreCmd("active"), a.restoreLastCmd("active"), @@ -118,14 +127,42 @@ 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 (and both forwarders) 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 legacy var/machine-ip- file the resident Go forwarder still +// polls. 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 legacy +// var/active-machine file the resident Go forwarder polls. If the DB can't be +// opened this is a HARD error — unlike IP caching, a promote must not +// half-succeed: writing only the legacy file would leave the DB (and therefore +// socat, the canonical client path) pointing at the OLD active machine, so a +// client would keep hitting the pre-promote DB. 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 @@ -209,6 +246,30 @@ func (a *app) ensureForwarder(ctx context.Context) { } } +// reinstallForwarder force-(re)installs the Go forwarder LaunchAgent: a full +// bootout + bootstrap, NOT the gentle Ensure. `proxy install` uses this (not +// ensureForwarder) because of the port swap: a forwarder process that started +// BEFORE socat took over 5442/5443 is still bound there, and Ensure would leave +// it alone (its program args are unchanged, so it looks "healthy") — so socat +// could never bind the canonical ports. Killing that forwarder's PID can't win +// either: it has KeepAlive, so launchd resurrects it. Only booting out its JOB +// frees the port, which bootout+bootstrap does — the bootout releases 5442/5443, +// the bootstrap brings the forwarder back on its new 5444/5445. Best-effort: +// honors PG_ENDPOINT_AUTOINSTALL=0 and warns rather than failing. +func (a *app) reinstallForwarder(ctx context.Context) { + if os.Getenv("PG_ENDPOINT_AUTOINSTALL") == "0" { + return + } + ld, err := a.launchd() + if err != nil { + fmt.Fprintf(os.Stderr, "WARNING: skipping Go forwarder install: %v\n", err) + return + } + if err := ld.Install(ctx); err != nil { + fmt.Fprintf(os.Stderr, "WARNING: Go forwarder install failed (run 'pgdev forward install'): %v\n", err) + } +} + // 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 diff --git a/pgdev/cmd/pgdev/proxy.go b/pgdev/cmd/pgdev/proxy.go new file mode 100644 index 0000000..3a4185f --- /dev/null +++ b/pgdev/cmd/pgdev/proxy.go @@ -0,0 +1,299 @@ +package main + +import ( + "context" + "fmt" + "log/slog" + "os" + "os/exec" + "strings" + + "github.com/spf13/cobra" + + "pansen.me/pgdev/internal/socatproxy" + "pansen.me/pgdev/internal/track" +) + +// newLogger builds the process's structured logger. Text handler to stderr so +// the lines interleave readably with the existing "==> " progress output; +// verbose flips the level to Debug, which the tracking + socat paths use to be +// deliberately chatty (the experiment wants a fat forensic trail). Format is +// key=value slog text, timestamped, so a promote/reconcile can be reconstructed +// entirely from stderr. +func newLogger(verbose bool) *slog.Logger { + level := slog.LevelInfo + if verbose { + level = slog.LevelDebug + } + h := slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: level}) + return slog.New(h) +} + +// 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.ForwardBind, + 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) + } + } +} + +// ----- `pgdev proxy` command ------------------------------------------------- + +// proxyCmd groups the socat proxy (internal/socatproxy) that serves the +// canonical client ports (5442/5443), alongside the Go forwarder now on +// 5444/5445. It is opt-in: nothing here runs unless you `proxy +// install`/`reconcile`. See doc/issues/0004. +func (a *app) proxyCmd() *cobra.Command { + c := &cobra.Command{ + Use: "proxy", + Short: "socat client proxy on the canonical 5442/5443 (Go forwarder moved to 5444/5445)", + } + 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 + } + printApplied(applied) + return appliedErr(applied) + }, + } +} + +// proxyInstallCmd is the ONE reference command for "install anything proxy": it +// brings up BOTH client paths — the client-facing socat proxy on the canonical +// 5442/5443 AND the Go forwarder on 5444/5445 — so there is no separate +// endpoint.install step to remember. The Go forwarder install is best-effort +// (it self-heals and warns rather than failing, and honors +// PG_ENDPOINT_AUTOINSTALL=0), because socat is the canonical path whose outcome +// must be surfaced; the forwarder is the secondary integration path. +func (a *app) proxyInstallCmd() *cobra.Command { + return &cobra.Command{ + Use: "install", + Short: "Install everything: the socat proxy (canonical 5442/5443) AND the Go forwarder (5444/5445)", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + ctx := cmd.Context() + + // 1. Go forwarder on 5444/5445. A FULL (re)install, not a gentle + // ensure: this boots any pre-swap forwarder off the canonical + // 5442/5443 (which it would otherwise still hold, blocking socat) + // and brings it back on 5444/5445 — freeing the ports BEFORE socat + // binds them in step 2. best-effort; honors PG_ENDPOINT_AUTOINSTALL=0. + a.reinstallForwarder(ctx) + fmt.Printf("==> Go forwarder: active 127.0.0.1:%d staging 127.0.0.1:%d\n", + a.cfg.ForwardActivePort, a.cfg.ForwardStagingPort) + + // 2. socat proxy on the canonical client ports (the client-facing path). + applied, err := a.reconcileProxy(ctx, "install") + if err != nil { + return err + } + fmt.Printf("==> socat proxy: active 127.0.0.1:%d staging 127.0.0.1:%d (the canonical client ports)\n", + a.cfg.ClientActivePort, a.cfg.ClientStagingPort) + printApplied(applied) + fmt.Println(" (promote/refresh now keep both in step automatically)") + 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 + } + fmt.Println("==> 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 { + ctx := cmd.Context() + rec, err := a.reconciler() + if err != nil { + return err + } + live := rec.Status(ctx, a.proxyPorts()) + db, err := a.track(ctx) + if err != nil { + return err + } + applied, err := db.AppliedTargets(ctx) + if err != nil { + return err + } + for _, role := range []string{"active", "staging"} { + fmt.Printf("%-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() + } + return nil + }, + } +} + +func printApplied(applied []socatproxy.Applied) { + for _, ap := range applied { + if ap.Err != nil { + fmt.Printf(" %-8s :%d FAILED: %v\n", ap.Role, ap.Port, ap.Err) + continue + } + fmt.Printf(" %-8s :%d %s -> %s\n", ap.Role, ap.Port, ap.Action, orNone(ap.Target)) + } +} diff --git a/pgdev/go.mod b/pgdev/go.mod index 2dbc36e..4b10651 100644 --- a/pgdev/go.mod +++ b/pgdev/go.mod @@ -2,9 +2,21 @@ module pansen.me/pgdev go 1.25 -require github.com/spf13/cobra v1.10.2 +require ( + 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/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/libc v1.55.3 // indirect + modernc.org/mathutil v1.6.0 // indirect + modernc.org/memory v1.8.0 // indirect ) diff --git a/pgdev/go.sum b/pgdev/go.sum index a6ee3e0..a5b3eff 100644 --- a/pgdev/go.sum +++ b/pgdev/go.sum @@ -1,10 +1,31 @@ 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/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +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/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/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= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +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/sqlite v1.31.1 h1:XVU0VyzxrYHlBhIs1DiEgSl0ZtdnPtbLVy8hSkzxGrs= +modernc.org/sqlite v1.31.1/go.mod h1:UqoylwmTb9F+IqXERT8bW9zzOWN8qwAIcLdzeBZs4hA= diff --git a/pgdev/internal/config/config.go b/pgdev/internal/config/config.go index 1832659..91763da 100644 --- a/pgdev/internal/config/config.go +++ b/pgdev/internal/config/config.go @@ -59,8 +59,20 @@ 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 CANONICAL client-facing loopback + // ports — what psql/.pgpass/status print and what external configs point at. + // The socat proxy (internal/socatproxy) now binds THESE, so existing clients + // pick socat up transparently without reconfiguration. ClientActivePort, ClientStagingPort int // 5442 / 5443 + // ForwardActivePort/ForwardStagingPort are where the Go forwarder + // (internal/forward) now listens — MOVED off the canonical pair onto a second + // pair so it runs alongside socat during the integration phase, no longer on + // the ports clients use (doc/issues/0004). + ForwardActivePort, ForwardStagingPort int // 5444 / 5445 + // 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 // 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. @@ -125,38 +137,41 @@ func Load() Config { } c := Config{ - PGUser: get("PG_USER", ""), - PGDB: get("PG_DB", ""), - PGPassword: get("PG_PASSWORD", ""), - MachineName: get("MACHINE_NAME", "vpg"), - MachinePrefix: get("MACHINE_PREFIX", get("MACHINE_NAME", "vpg")), - Slot: get("PG_SLOT", ""), - BackendPort: atoi(get("PG_BACKEND_PORT", "5432")), - MachineCPUs: atoi(get("MACHINE_CPUS", "4")), - MachineMemory: get("MACHINE_MEMORY", "8G"), - MachineImage: get("MACHINE_IMAGE", "local/pg-incus-machine:26.04"), - ActivePort: atoi(get("PG_ACTIVE_PORT", "5432")), - StagingPort: atoi(get("PG_STAGING_PORT", "5433")), - ClientActivePort: atoi(get("PG_CLIENT_ACTIVE_PORT", "5442")), - ClientStagingPort: atoi(get("PG_CLIENT_STAGING_PORT", "5443")), - ProxyHostname: get("PG_PROXY_HOSTNAME", "host.docker.internal"), - ForwardBind: get("PG_FORWARD_BIND", "127.0.0.1"), - ForwardVerbose: get("PG_FORWARD_DEBUG", "") == "1", - BackendPrefix: get("PG_BACKEND_PREFIX", DefaultBackendPrefix), - ProxyName: get("PG_PROXY_NAME", DefaultProxyName), - BackendAIP: get("PG_BACKEND_A_IP", ""), - BackendBIP: get("PG_BACKEND_B_IP", ""), - DataRoot: get("PG_DATA_ROOT", DefaultDataRoot), - DataDiskSize: get("PG_DATA_DISK_SIZE", "140G"), - DataImage: get("PG_DATA_IMAGE", DefaultDataRoot+".xfs"), - BaseImage: get("PG_BASE_IMAGE", "images:ubuntu/24.04/cloud"), - GoldenImage: get("PG_GOLDEN_IMAGE", "pg-dev-base"), - AgentPort: atoi(get("PG_AGENT_PORT", "5440")), - AgentToken: get("PG_AGENT_TOKEN", ""), - MachineIP: get("PG_MACHINE_IP", ""), - RepoRoot: repo, - HostUID: get("HOST_UID", ""), - HostGID: get("HOST_GID", ""), + PGUser: get("PG_USER", ""), + PGDB: get("PG_DB", ""), + PGPassword: get("PG_PASSWORD", ""), + MachineName: get("MACHINE_NAME", "vpg"), + MachinePrefix: get("MACHINE_PREFIX", get("MACHINE_NAME", "vpg")), + Slot: get("PG_SLOT", ""), + BackendPort: atoi(get("PG_BACKEND_PORT", "5432")), + MachineCPUs: atoi(get("MACHINE_CPUS", "4")), + MachineMemory: get("MACHINE_MEMORY", "8G"), + MachineImage: get("MACHINE_IMAGE", "local/pg-incus-machine:26.04"), + ActivePort: atoi(get("PG_ACTIVE_PORT", "5432")), + StagingPort: atoi(get("PG_STAGING_PORT", "5433")), + ClientActivePort: atoi(get("PG_CLIENT_ACTIVE_PORT", "5442")), + ClientStagingPort: atoi(get("PG_CLIENT_STAGING_PORT", "5443")), + ForwardActivePort: atoi(get("PG_FORWARD_ACTIVE_PORT", "5444")), + ForwardStagingPort: atoi(get("PG_FORWARD_STAGING_PORT", "5445")), + ProxyVerbose: get("PG_PROXY_DEBUG", "1") != "0", + ProxyHostname: get("PG_PROXY_HOSTNAME", "host.docker.internal"), + ForwardBind: get("PG_FORWARD_BIND", "127.0.0.1"), + ForwardVerbose: get("PG_FORWARD_DEBUG", "") == "1", + BackendPrefix: get("PG_BACKEND_PREFIX", DefaultBackendPrefix), + ProxyName: get("PG_PROXY_NAME", DefaultProxyName), + BackendAIP: get("PG_BACKEND_A_IP", ""), + BackendBIP: get("PG_BACKEND_B_IP", ""), + DataRoot: get("PG_DATA_ROOT", DefaultDataRoot), + DataDiskSize: get("PG_DATA_DISK_SIZE", "140G"), + DataImage: get("PG_DATA_IMAGE", DefaultDataRoot+".xfs"), + BaseImage: get("PG_BASE_IMAGE", "images:ubuntu/24.04/cloud"), + GoldenImage: get("PG_GOLDEN_IMAGE", "pg-dev-base"), + AgentPort: atoi(get("PG_AGENT_PORT", "5440")), + AgentToken: get("PG_AGENT_TOKEN", ""), + MachineIP: get("PG_MACHINE_IP", ""), + RepoRoot: repo, + HostUID: get("HOST_UID", ""), + HostGID: get("HOST_GID", ""), } c.AgentTokenPath = get("PG_AGENT_TOKEN_PATH", filepath.Join(repo, "var", "agent-token")) return c @@ -236,6 +251,35 @@ func (c Config) ForwardLogPath() string { return filepath.Join(c.RepoRoot, "var", c.MachinePrefix+"-forward.log") } +// ----- 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") +} + +// 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") +} + +// ForwardPort returns the Go forwarder's loopback port for a role +// ("active"/"staging") — the second pair (5444/5445) it moved to when socat took +// over the canonical client ports. +func (c Config) ForwardPort(role string) int { + if role == "staging" { + return c.ForwardStagingPort + } + return c.ForwardActivePort +} + // ClientPort returns the host client port for a role ("active"/"staging"). func (c Config) ClientPort(role string) int { if role == "staging" { diff --git a/pgdev/internal/forward/forward.go b/pgdev/internal/forward/forward.go index 1fb8edf..a53bb5d 100644 --- a/pgdev/internal/forward/forward.go +++ b/pgdev/internal/forward/forward.go @@ -21,8 +21,8 @@ import ( // 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) + ActivePort int // host port for the active role (5444; socat holds the canonical 5442) + StagingPort int // host port for the staging role (5445; socat holds the canonical 5443) BackendPort int // port each machine serves its backend on (5432) ActiveMachinePath string // var/active-machine diff --git a/pgdev/internal/forward/router.go b/pgdev/internal/forward/router.go index 1407707..eec74b6 100644 --- a/pgdev/internal/forward/router.go +++ b/pgdev/internal/forward/router.go @@ -1,7 +1,9 @@ // Package forward is the host-side (macOS) client forwarder that replaces the -// shell socat relay (scripts/host-endpoint, retired in spec 0003). It owns the -// two stable loopback listeners — 127.0.0.1:5442 (active) and :5443 (staging) — -// for their whole lifetime and re-points by swapping the dial target IN PLACE, +// shell socat relay (scripts/host-endpoint, retired in spec 0003). It owns two +// stable loopback listeners for their whole lifetime and re-points by swapping +// the dial target IN PLACE. Since spec 0004 it listens on 127.0.0.1:5444 +// (active) / :5445 (staging) — MOVED off the canonical client ports, which the +// socat proxy (internal/socatproxy) now serves; the forwarder runs alongside it, // never rebinding. That kills the socat/launchd process-lifecycle bugs: nothing // external to orphan, no "Address already in use" on re-point, no SIGKILL // trap-bypass. `promote` then collapses to a pointer write — the running diff --git a/pgdev/internal/socatproxy/launchd.go b/pgdev/internal/socatproxy/launchd.go new file mode 100644 index 0000000..6c93902 --- /dev/null +++ b/pgdev/internal/socatproxy/launchd.go @@ -0,0 +1,413 @@ +// Package socatproxy is the EXPERIMENTAL socat-based host client proxy that runs +// ALONGSIDE the resident Go forwarder (internal/forward) during an integration +// phase (see doc/issues/0004). It exists because the Go forwarder's own binary +// trips macOS Local Network Privacy — needing a codesign ceremony and still +// throwing occasional prompts — whereas Homebrew's socat does not. So we trial +// routing through socat under launchd on a SECOND pair of ports (5444 active / +// 5445 staging), leaving 5442/5443 untouched. +// +// socat cannot re-point itself the way the Go forwarder does: each promote / IP +// drift must rewrite and reload a launchd job. That reload is exactly the +// process-lifecycle race that got socat retired in spec 0003 — an orphaned +// listener still holding the port makes the reloaded socat fail to bind, and the +// stale mapping silently persists (pg_restore → wrong DB). This package tames it +// NOT with the SQLite transaction (that only serializes reconcilers) but with an +// explicit port-free GATE before bind and a post-bind VERIFY that the live socat +// argv actually carries the intended target. Reconcile (reconcile.go) drives it. +package socatproxy + +import ( + "bytes" + "context" + "encoding/xml" + "fmt" + "log/slog" + "net" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "text/template" + "time" +) + +// job is one role's socat LaunchAgent: label, plist path, and the parameters +// baked into the socat argv. One job per role because a socat process forwards +// exactly one listener. +type job struct { + label string // me.pansen.-socat- + plist string // ~/Library/LaunchAgents/