From 6d8e827db4422da46581df18820c277bb6e2fb35 Mon Sep 17 00:00:00 2001 From: Antoine Toussaint <312536060+antoinetoussaint-byte@users.noreply.github.com> Date: Thu, 8 Oct 2026 17:47:00 -0400 Subject: [PATCH 1/5] fix: preserve dev publishing when repinning release workflow --- .github/workflows/releaser.yml | 90 +++++++++++++++- AGENTS.md | 2 + docs/releases.md | 75 ++++++++++++++ release_dev_test.go | 181 +++++++++++++++++++++++++++++++++ workflow_test.go | 2 + 5 files changed, 348 insertions(+), 2 deletions(-) create mode 100644 docs/releases.md create mode 100644 release_dev_test.go diff --git a/.github/workflows/releaser.yml b/.github/workflows/releaser.yml index 9e5552b..99fbee5 100644 --- a/.github/workflows/releaser.yml +++ b/.github/workflows/releaser.yml @@ -71,11 +71,11 @@ jobs: .github/scripts/tag-runtime-image.sh "$name:$RELEASE_TAG" "$name:$tag" release: - if: github.event_name == 'push' + if: github.event_name == 'push' && !contains(github.ref_name, '-dev.') needs: image permissions: contents: read - uses: codefly-dev/core/.github/workflows/go-service-release.yml@08e37db5962cd98ddf77f8fb7dffe810a7ca1fc9 + uses: codefly-dev/core/.github/workflows/go-service-release.yml@e54bc852f3d40cd1aeafc0ced2369a8c13a51bb8 # core v0.16.0 with: # The bootstrap-image tests build linux/arm64 and refuse to run without # emulation registered ("CI must register emulation"). ci.yml gets it from @@ -85,6 +85,92 @@ jobs: setup-run: docker run --privileged --rm tonistiigi/binfmt --install arm64 secrets: inherit + # Iteration tags deliberately point off main. Build without the release PAT, + # then upload data from a fresh runner; core's release gate stays intact. + dev-build: + if: github.event_name == 'push' && contains(github.ref_name, '-dev.') + needs: image + permissions: + contents: read + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 + with: + fetch-depth: 0 + persist-credentials: false + - name: Validate dev tag + env: + RELEASE_TAG: ${{ github.ref_name }} + run: | + set -euo pipefail + [[ "$RELEASE_TAG" =~ ^v[0-9]+\.[0-9]+\.[0-9]+-dev\.[0-9a-f]+$ ]] + .github/scripts/release-backfill.sh validate-tag . + - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff + with: + go-version-file: go.mod + cache: false + - uses: docker/setup-qemu-action@c7c53464625b32c7a7e944ae62b3e17d2b600130 + - name: Test + run: go test -v ./... + - name: Install Syft for archive SBOMs + uses: anchore/sbom-action/download-syft@66cbf4bc1f1c0d2edc94016e65bc221b6bb0ad6c + with: + syft-version: v1.44.0 + - name: Build dev release + uses: goreleaser/goreleaser-action@f06c13b6b1a9625abc9e6e439d9c05a8f2190e94 + with: + version: '~> v2' + args: release --clean --skip=publish,announce + - name: Upload dev release + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 + with: + name: dev-release + if-no-files-found: error + retention-days: 7 + path: | + dist/*.tar.gz + dist/*.tar.gz.sbom.json + dist/*_checksums.txt + dist/CHANGELOG.md + + dev-publish: + if: github.event_name == 'push' && contains(github.ref_name, '-dev.') + needs: dev-build + permissions: + contents: write + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + # No source checkout, hooks, or executables from the build job. The token + # is repository-scoped and only passed to the upload step, never GH_PAT. + - uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 + with: + name: dev-release + path: dist + - name: Publish dev prerelease + env: + GH_TOKEN: ${{ github.token }} + GH_REPO: ${{ github.repository }} + RELEASE_TAG: ${{ github.ref_name }} + run: | + set -euo pipefail + [[ "$RELEASE_TAG" =~ ^v[0-9]+\.[0-9]+\.[0-9]+-dev\.[0-9a-f]+$ ]] + prefix="dist/service-postgres_${RELEASE_TAG#v}" + assets=("${prefix}_checksums.txt") + for platform in darwin_amd64 darwin_arm64 linux_amd64; do + assets+=("${prefix}_${platform}.tar.gz" "${prefix}_${platform}.tar.gz.sbom.json") + done + for asset in "${assets[@]}" dist/CHANGELOG.md; do + if [[ ! -s "$asset" ]]; then + echo "::error::missing or empty dev release file: $asset" >&2 + exit 1 + fi + done + gh release create "$RELEASE_TAG" "${assets[@]}" \ + --verify-tag --prerelease --latest=false \ + --title "$RELEASE_TAG" --notes-file dist/CHANGELOG.md + backfill: if: github.event_name == 'workflow_dispatch' permissions: diff --git a/AGENTS.md b/AGENTS.md index 6433ba5..ff43d0f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -149,6 +149,8 @@ Loaded on demand rather than carried here — `.claude/skills/`: ## Workflow - Branch and PR; never commit to `main`. Conventional Commits for the title. +- Release flow: `docs/releases.md`. Keep dev tags publishing when repinning + core; only normal releases use its default-branch reachability gate. - The PR body carries the fix-or-hack classification and what you did not verify. Both are rules above; answer them there rather than omitting them. - `agentcontext_test.go` holds this file's length budget and each skill's diff --git a/docs/releases.md b/docs/releases.md new file mode 100644 index 0000000..b0648e5 --- /dev/null +++ b/docs/releases.md @@ -0,0 +1,75 @@ +# Publishing releases and dev builds + +`agent.codefly.yaml` declares `.github/workflows/releaser.yml` as the sole +publisher. Its `v*` tag trigger includes `codefly publish dev` tags of the form +`v..-dev.`. Do not exclude those tags: the workflow is +what turns an iteration tag into a downloadable prerelease. + +Both tag paths qualify `libs/go` and retag the digest in `runtime-image.json` +with the Git tag and the lock's runtime tag before publishing binaries. Runs +share a concurrency group and queue without cancellation so they cannot race +on that runtime tag. + +## Normal releases + +Tags without `-dev.` call core's `go-service-release.yml`, pinned to the commit +for core v0.16.0 (`e54bc852f3d40cd1aeafc0ced2369a8c13a51bb8`). It runs the full Go +test suite, then checks that the tag's commit is reachable from the repository's +default branch before running GoReleaser with `GH_PAT`. The caller registers +arm64 emulation for the bootstrap-image tests. Stable tags produce ordinary +GitHub releases; `.goreleaser.yaml` continues to control their assets and release +metadata. + +## Dev prereleases + +Iteration commits need not be on the default branch. They therefore use local +jobs instead of core's release job: + +1. `dev-build` checks out the tag with full history and no persisted Git + credential. It validates the dev-tag format and tag/HEAD identity, exports + `GORELEASER_CURRENT_TAG` (including when multiple tags share a commit), sets up + arm64 emulation, and runs the full Go test suite. GoReleaser uses the same + `.goreleaser.yaml` as normal releases, with Syft installed and + `--skip=publish,announce`. This job has read-only repository permission and + receives no publishing token or PAT. +2. `dev-publish` downloads that run's build artifact on a fresh runner, with no + source checkout or build hooks. It checks that all seven expected assets and + the generated changelog are nonempty, then calls `gh release create` with + `--verify-tag --prerelease --latest=false`. Only this upload step receives the + repository-scoped `GITHUB_TOKEN` with `contents: write`; it never uses + `GH_PAT`. + +The seven assets match normal releases: `darwin_amd64`, `darwin_arm64`, and +`linux_amd64` tarballs, one `.tar.gz.sbom.json` alongside each, and a +`service-postgres__checksums.txt`. The changelog becomes release notes, +not an eighth asset. No GoReleaser snapshot mode is used, so archive names keep +the exact dev version. Dev builds cannot become GitHub's Latest release. + +This job split avoids running off-main build hooks with a release credential. +It does not authorize arbitrary tag pushes: repository tag/workflow write +access remains the trust boundary. Releases created with `GITHUB_TOKEN` do not +trigger downstream `release` workflows; this repository has no such consumer. + +## Backfills and verification + +The manual `workflow_dispatch` path remains the historical `backfill` job. It +checks out the requested source, repairs dependency metadata in a temporary +modfile, and uploads missing archives/checksums without replacing existing +assets. It is not the dev publishing path and does not supply the seven-asset +prerelease contract described above. + +When repinning core, retain the mutually exclusive tag routing and run: + +```sh +actionlint .github/workflows/releaser.yml +go test . -run 'Test(Release|Dev)' -count=1 +``` + +The tests pin the routing, build/publish separation, and complete seven-asset +upload. They execute the publisher shell with a fake GitHub client, including +missing-asset and non-dev-tag failures. They do not prove a live GitHub publish. +For end-to-end acceptance, observe a normal release tag on the default branch +and a dev tag on an off-main commit containing this workflow: both runs must +succeed, with seven assets each, and only the dev release marked prerelease +(and never Latest). Existing historical tags still contain their old workflow; +rerunning those runs does not exercise this change. diff --git a/release_dev_test.go b/release_dev_test.go new file mode 100644 index 0000000..46085bc --- /dev/null +++ b/release_dev_test.go @@ -0,0 +1,181 @@ +package main + +import ( + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + + "github.com/stretchr/testify/require" + "gopkg.in/yaml.v3" +) + +func TestReleaseWorkflowRoutesDevTagsWithoutTheSharedGate(t *testing.T) { + payload, err := os.ReadFile(".github/workflows/releaser.yml") + require.NoError(t, err) + var trigger struct { + On struct { + Push struct{ Tags []string } + } + } + require.NoError(t, yaml.Unmarshal(payload, &trigger)) + require.Equal(t, []string{"v*"}, trigger.On.Push.Tags, "excluding dev tags removes their publisher") + + workflow := readWorkflow(t, ".github/workflows/releaser.yml") + release := workflow.Jobs["release"] + require.Equal(t, "github.event_name == 'push' && !contains(github.ref_name, '-dev.')", release.If) + require.Equal(t, "image", release.Needs) + require.Equal(t, "codefly-dev/core/.github/workflows/go-service-release.yml@e54bc852f3d40cd1aeafc0ced2369a8c13a51bb8", release.Uses, + "core v0.16.0 enforces default-branch reachability for real releases") + for _, name := range []string{"dev-build", "dev-publish"} { + require.Equal(t, "github.event_name == 'push' && contains(github.ref_name, '-dev.')", workflow.Jobs[name].If) + require.Empty(t, workflow.Jobs[name].Uses, "dev tags must not enter core's default-branch gate") + } + require.Equal(t, "github.event_name == 'workflow_dispatch'", workflow.Jobs["backfill"].If) +} + +func TestDevReleaseBuildAndPublishAreSeparate(t *testing.T) { + workflow := readWorkflow(t, ".github/workflows/releaser.yml") + build := workflow.Jobs["dev-build"] + publish := workflow.Jobs["dev-publish"] + require.Equal(t, "image", build.Needs) + require.Equal(t, "dev-build", publish.Needs) + require.Equal(t, map[string]string{"contents": "read"}, build.Permissions) + require.Equal(t, map[string]string{"contents": "write"}, publish.Permissions) + checkout := findWorkflowAction(t, build, "actions/checkout") + require.Equal(t, 0, checkout.With["fetch-depth"]) + require.Equal(t, false, checkout.With["persist-credentials"]) + qemu := -1 + for i, step := range build.Steps { + if strings.HasPrefix(step.Uses, "docker/setup-qemu-action@") { + qemu = i + } + for _, value := range step.Env { + require.NotContains(t, value, "secrets.") + require.NotContains(t, value, "github.token") + } + } + testIndex, test := findWorkflowStepAt(t, build, "Test") + require.GreaterOrEqual(t, qemu, 0) + require.Less(t, qemu, testIndex) + require.Equal(t, "go test -v ./...", test.Run) + syftIndex, syft := findWorkflowStepAt(t, build, "Install Syft for archive SBOMs") + buildIndex, goreleaser := findWorkflowStepAt(t, build, "Build dev release") + uploadIndex, upload := findWorkflowStepAt(t, build, "Upload dev release") + require.Less(t, testIndex, buildIndex) + require.Less(t, syftIndex, buildIndex) + require.Less(t, buildIndex, uploadIndex) + require.Equal(t, "v1.44.0", syft.With["syft-version"]) + require.Equal(t, "release --clean --skip=publish,announce", goreleaser.With["args"]) + require.Equal(t, "error", upload.With["if-no-files-found"]) + require.Equal(t, "dist/*.tar.gz\ndist/*.tar.gz.sbom.json\ndist/*_checksums.txt\ndist/CHANGELOG.md\n", upload.With["path"]) + require.Len(t, publish.Steps, 2, "the publishing runner only downloads and uploads data") + download := findWorkflowAction(t, publish, "actions/download-artifact") + require.Equal(t, upload.With["name"], download.With["name"]) + require.Equal(t, "dist", download.With["path"]) + step := findWorkflowStep(t, publish, "Publish dev prerelease") + require.Equal(t, "${{ github.token }}", step.Env["GH_TOKEN"]) + require.Equal(t, "${{ github.repository }}", step.Env["GH_REPO"]) + require.Equal(t, "${{ github.ref_name }}", step.Env["RELEASE_TAG"]) +} + +// Run the actual upload shell against a fake gh. Extra files must never become +// assets, and a missing SBOM must fail before creating a partial prerelease. +func TestDevPublisherRequiresAllSevenAssetsAndNeverPublishesStable(t *testing.T) { + const devTag = "v0.0.139-dev.93280f407cc4" + for _, tc := range []struct { + name, tag, missing string + wantSuccess bool + }{ + {name: "complete dev release", tag: devTag, wantSuccess: true}, + {name: "missing SBOM", tag: devTag, missing: "darwin_arm64.tar.gz.sbom.json"}, + {name: "missing archive", tag: devTag, missing: "linux_amd64.tar.gz"}, + {name: "missing checksums", tag: devTag, missing: "checksums.txt"}, + {name: "missing notes", tag: devTag, missing: "CHANGELOG.md"}, + {name: "stable", tag: "v0.0.141"}, + {name: "other prerelease", tag: "v0.0.141-rc.1"}, + {name: "malformed dev", tag: "v0.0.141-dev.not-a-sha"}, + } { + t.Run(tc.name, func(t *testing.T) { + dir := t.TempDir() + require.NoError(t, os.Mkdir(filepath.Join(dir, "dist"), 0o755)) + assets := []string{"service-postgres_" + strings.TrimPrefix(tc.tag, "v") + "_checksums.txt"} + for _, platform := range []string{"darwin_amd64", "darwin_arm64", "linux_amd64"} { + archive := fmt.Sprintf("service-postgres_%s_%s.tar.gz", strings.TrimPrefix(tc.tag, "v"), platform) + assets = append(assets, archive, archive+".sbom.json") + } + for _, name := range append(append([]string{}, assets...), "CHANGELOG.md", "unrelated.tar.gz", "hook.sh") { + if tc.missing != "" && strings.HasSuffix(name, tc.missing) { + continue + } + require.NoError(t, os.WriteFile(filepath.Join(dir, "dist", name), []byte("fixture"), 0o644)) + } + log := filepath.Join(dir, "gh.log") + step := findWorkflowStep(t, readWorkflow(t, ".github/workflows/releaser.yml").Jobs["dev-publish"], "Publish dev prerelease") + command := exec.Command("bash", "-c", step.Run) + command.Dir = dir + command.Env = append(os.Environ(), + "PATH="+fakeGitHubCLI(t)+string(os.PathListSeparator)+os.Getenv("PATH"), + "GH_LOG="+log, "GH_TOKEN=test-token", "GH_REPO=codefly-dev/service-postgres", "RELEASE_TAG="+tc.tag) + output, err := command.CombinedOutput() + if !tc.wantSuccess { + require.Error(t, err, string(output)) + _, err = os.Stat(log) + require.True(t, os.IsNotExist(err), "no GitHub write may happen on invalid input") + return + } + require.NoError(t, err, string(output)) + calls := readCalls(t, log) + require.Len(t, calls, 1) + want := "release create " + tc.tag + for _, asset := range assets { + want += " dist/" + asset + } + want += " --verify-tag --prerelease --latest=false --title " + tc.tag + " --notes-file dist/CHANGELOG.md" + require.Equal(t, want, calls[0]) + }) + } +} + +func TestDevTagValidationAcceptsOffMainAndUsesExactTag(t *testing.T) { + dir := t.TempDir() + runTestCommand(t, dir, nil, "git", "init", "--quiet", "--initial-branch=main") + runTestCommand(t, dir, nil, "git", "config", "user.name", "Test") + runTestCommand(t, dir, nil, "git", "config", "user.email", "test@example.com") + runTestCommand(t, dir, nil, "git", "config", "commit.gpgsign", "false") + runTestCommand(t, dir, nil, "git", "config", "tag.gpgsign", "false") + runTestCommand(t, dir, nil, "git", "commit", "--quiet", "--allow-empty", "-m", "main") + runTestCommand(t, dir, nil, "git", "checkout", "--quiet", "--detach") + runTestCommand(t, dir, nil, "git", "commit", "--quiet", "--allow-empty", "-m", "off-main iteration") + sha := strings.TrimSpace(runTestCommand(t, dir, nil, "git", "rev-parse", "HEAD")) + tag := "v0.0.139-dev." + sha[:12] + runTestCommand(t, dir, nil, "git", "tag", tag) + // An annotated tag wins git describe over the lightweight dev tag. + runTestCommand(t, dir, nil, "git", "tag", "-a", "v0.0.140", "-m", "another tag") + require.Equal(t, "v0.0.140", strings.TrimSpace(runTestCommand(t, dir, nil, "git", "describe", "--tags", "--exact-match"))) + gate := exec.Command("git", "merge-base", "--is-ancestor", "HEAD", "main") + gate.Dir = dir + require.Error(t, gate.Run(), "fixture must be rejected by a default-branch reachability gate") + + script, err := os.ReadFile(".github/scripts/release-backfill.sh") + require.NoError(t, err) + require.NoError(t, os.MkdirAll(filepath.Join(dir, ".github", "scripts"), 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(dir, ".github", "scripts", "release-backfill.sh"), script, 0o755)) + envFile := filepath.Join(t.TempDir(), "github-env") + step := findWorkflowStep(t, readWorkflow(t, ".github/workflows/releaser.yml").Jobs["dev-build"], "Validate dev tag") + runTestCommand(t, dir, []string{"RELEASE_TAG=" + tag, "GITHUB_ENV=" + envFile}, "bash", "-c", step.Run) + environment, err := os.ReadFile(envFile) + require.NoError(t, err) + require.Equal(t, "GORELEASER_CURRENT_TAG="+tag+"\n", string(environment)) + + // A valid dev tag pointing at a different commit must still be rejected. + runTestCommand(t, dir, nil, "git", "checkout", "--quiet", "main") + command := exec.Command("bash", "-c", step.Run) + command.Dir = dir + command.Env = append(os.Environ(), "RELEASE_TAG="+tag, "GITHUB_ENV="+envFile) + output, err := command.CombinedOutput() + require.Error(t, err) + require.Contains(t, string(output), "does not point at the checked-out commit") +} diff --git a/workflow_test.go b/workflow_test.go index 2284d41..b89f3e3 100644 --- a/workflow_test.go +++ b/workflow_test.go @@ -26,6 +26,8 @@ type workflowConcurrency struct { } type workflowJob struct { + If string `yaml:"if"` + Needs string `yaml:"needs"` Permissions map[string]string `yaml:"permissions"` TimeoutMinutes int `yaml:"timeout-minutes"` Steps []workflowStep `yaml:"steps"` From be6d0b58a21e3ed38ca9e145a57b67a46d760a15 Mon Sep 17 00:00:00 2001 From: Antoine Toussaint <312536060+antoinetoussaint-byte@users.noreply.github.com> Date: Tue, 27 Oct 2026 09:30:32 -0400 Subject: [PATCH 2/5] feat(runtime): one native Postgres server per run, a database and an owner per service Co-Authored-By: Claude Opus 5.5 --- forward.go | 117 +++++ forward_test.go | 65 +++ go.mod | 2 +- hostipc_lifecycle_test.go | 7 +- lifecycle_test.go | 18 +- main.go | 10 +- main_test.go | 14 +- nixpg.go | 992 +++++++++++++++++++++++++------------- nixpg_auth_test.go | 44 +- nixpg_run_test.go | 105 ++++ nixpg_scope_test.go | 38 +- nixpg_shared_test.go | 170 +++++++ recovery_scope_test.go | 15 + runtime.go | 87 ++-- runtime_access.go | 31 +- runtime_native.go | 78 +++ 16 files changed, 1351 insertions(+), 442 deletions(-) create mode 100644 forward.go create mode 100644 forward_test.go create mode 100644 nixpg_run_test.go create mode 100644 nixpg_shared_test.go create mode 100644 runtime_native.go diff --git a/forward.go b/forward.go new file mode 100644 index 0000000..34f1304 --- /dev/null +++ b/forward.go @@ -0,0 +1,117 @@ +package main + +// forward.go — a service's own endpoint on its run's shared server. +// +// The run's server listens on a private Unix socket only (nixpg.go). Each +// service keeps the TCP endpoint Codefly allocated it: its agent listens there, +// on loopback as the per-service server did, and carries every connection to +// the socket byte for byte. Authentication (SCRAM) is between the client and +// the server, so the agent sees no credential, and the address in every +// connection string a consumer receives is the one it always was. + +import ( + "errors" + "fmt" + "io" + "net" + "sync" + "time" +) + +type socketForwarder struct { + listener net.Listener + socket string + + mu sync.Mutex + conns map[net.Conn]struct{} + closed bool + wg sync.WaitGroup +} + +// forwardToSocket listens on address and forwards each connection to socket. +func forwardToSocket(address, socket string) (*socketForwarder, error) { + listener, err := net.Listen("tcp", address) + if err != nil { + return nil, fmt.Errorf("listen on this service's endpoint %s: %w", address, err) + } + f := &socketForwarder{listener: listener, socket: socket, conns: map[net.Conn]struct{}{}} + f.wg.Add(1) + go f.serve() + return f, nil +} + +func (f *socketForwarder) serve() { + defer f.wg.Done() + for { + client, err := f.listener.Accept() + if err != nil { + return + } + if !f.track(client) { + _ = client.Close() + return + } + f.wg.Add(1) + go f.pipe(client) + } +} + +func (f *socketForwarder) pipe(client net.Conn) { + defer f.wg.Done() + defer f.untrack(client) + server, err := net.DialTimeout("unix", f.socket, 5*time.Second) + if err != nil { + return + } + if !f.track(server) { + _ = server.Close() + return + } + defer f.untrack(server) + done := make(chan struct{}, 2) + carry := func(to, from net.Conn) { + _, _ = io.Copy(to, from) + done <- struct{}{} + } + go carry(server, client) + go carry(client, server) + // Either side ending ends the session: closing both unblocks the other copy. + <-done + _ = client.Close() + _ = server.Close() + <-done +} + +func (f *socketForwarder) track(conn net.Conn) bool { + f.mu.Lock() + defer f.mu.Unlock() + if f.closed { + return false + } + f.conns[conn] = struct{}{} + return true +} + +func (f *socketForwarder) untrack(conn net.Conn) { + f.mu.Lock() + defer f.mu.Unlock() + delete(f.conns, conn) + _ = conn.Close() +} + +// Close stops listening and ends every forwarded session, so the endpoint is +// released the moment the service is. +func (f *socketForwarder) Close() error { + f.mu.Lock() + f.closed = true + err := f.listener.Close() + for conn := range f.conns { + _ = conn.Close() + } + f.mu.Unlock() + f.wg.Wait() + if errors.Is(err, net.ErrClosed) { + return nil + } + return err +} diff --git a/forward_test.go b/forward_test.go new file mode 100644 index 0000000..bad9d6c --- /dev/null +++ b/forward_test.go @@ -0,0 +1,65 @@ +package main + +import ( + "bufio" + "net" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// TestForwarderCarriesSessionsToTheSocketAndReleasesThePort is the forwarder's +// whole contract: bytes both ways, unchanged, and nothing left listening or +// connected once the service is released. +func TestForwarderCarriesSessionsToTheSocketAndReleasesThePort(t *testing.T) { + socket := filepath.Join(shortTempDir(t), "s") + upstream, err := net.Listen("unix", socket) + require.NoError(t, err) + defer upstream.Close() + go func() { + for { + conn, err := upstream.Accept() + if err != nil { + return + } + go func(conn net.Conn) { + defer conn.Close() + line, err := bufio.NewReader(conn).ReadString('\n') + if err == nil { + _, _ = conn.Write([]byte("echo " + line)) + } + // Hold the session open, as a database connection does. + _, _ = conn.Read(make([]byte, 1)) + }(conn) + } + }() + + forwarder, err := forwardToSocket("127.0.0.1:0", socket) + require.NoError(t, err) + address := forwarder.listener.Addr().String() + + client, err := net.DialTimeout("tcp", address, time.Second) + require.NoError(t, err) + _, err = client.Write([]byte("hello\n")) + require.NoError(t, err) + reply, err := bufio.NewReader(client).ReadString('\n') + require.NoError(t, err) + require.Equal(t, "echo hello\n", reply) + + require.NoError(t, forwarder.Close()) + require.NoError(t, client.SetReadDeadline(time.Now().Add(2*time.Second))) + _, err = client.Read(make([]byte, 1)) + require.Error(t, err, "a released endpoint must end the sessions it carried") + _, err = net.DialTimeout("tcp", address, time.Second) + require.Error(t, err, "a released endpoint must stop listening") +} + +func TestForwarderRefusesAnEndpointAlreadyInUse(t *testing.T) { + taken, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer taken.Close() + _, err = forwardToSocket(taken.Addr().String(), filepath.Join(t.TempDir(), "s")) + require.ErrorContains(t, err, "listen on this service's endpoint") +} diff --git a/go.mod b/go.mod index db12e80..4d8cde4 100644 --- a/go.mod +++ b/go.mod @@ -6,6 +6,7 @@ require ( github.com/codefly-dev/core v0.19.4 github.com/codefly-dev/gortk v0.2.0 github.com/codefly-dev/service-postgres/libs/go v0.0.0-20260915032422-272301033aed + github.com/gofrs/flock v0.13.1 github.com/golang-migrate/migrate/v4 v4.19.1 github.com/jackc/pgx/v5 v5.10.0 github.com/lib/pq v1.12.3 @@ -68,7 +69,6 @@ require ( github.com/go-openapi/swag/typeutils v0.29.2 // indirect github.com/go-openapi/swag/yamlutils v0.29.2 // indirect github.com/go-viper/mapstructure/v2 v2.5.0 // indirect - github.com/gofrs/flock v0.13.1 // indirect github.com/golang-jwt/jwt/v5 v5.3.1 // indirect github.com/google/gnostic-models v0.7.1 // indirect github.com/google/go-github/v89 v89.0.0 // indirect diff --git a/hostipc_lifecycle_test.go b/hostipc_lifecycle_test.go index b94ef85..0d57298 100644 --- a/hostipc_lifecycle_test.go +++ b/hostipc_lifecycle_test.go @@ -64,7 +64,12 @@ func TestNativeLifecycleRecoversHostResources(t *testing.T) { // Stop and Destroy release a postmaster, which is the moment its System V // resources become collectable. A bare handle stands in for a started - // cluster so the teardown wiring is covered on hosts without nix too. + // cluster so the teardown wiring is covered on hosts without nix too; on a + // host with nix, the server Init really started is released first. + if runtime.nixRuntime != nil { + _, err = runtime.releaseNative(ctx) + require.NoError(t, err) + } runtime.nixRuntime = &nixPostgres{} _, err = runtime.Stop(ctx, &runtimev0.StopRequest{}) require.NoError(t, err) diff --git a/lifecycle_test.go b/lifecycle_test.go index 37c9265..1a483cc 100644 --- a/lifecycle_test.go +++ b/lifecycle_test.go @@ -36,19 +36,25 @@ type postgresFixture struct { func newPostgresFixture(t *testing.T, ctx context.Context, runtimeContext *basev0.RuntimeContext) *postgresFixture { t.Helper() + return newPostgresFixtureAt(t, ctx, t.TempDir(), "mod", fmt.Sprintf("svc-%v", time.Now().UnixMilli()), runtimeContext) +} + +// newPostgresFixtureAt scaffolds module/serviceName in workspaceDir, so +// several fixtures can be services of one run. The scaffold's database is +// named after the module. +func newPostgresFixtureAt(t *testing.T, ctx context.Context, workspaceDir, module, serviceName string, runtimeContext *basev0.RuntimeContext) *postgresFixture { + t.Helper() workspace := &resources.Workspace{Name: "test"} - workspaceDir := t.TempDir() - serviceName := fmt.Sprintf("svc-%v", time.Now().UnixMilli()) service := resources.Service{Name: serviceName, Version: "test-me"} - require.NoError(t, service.SaveAtDir(ctx, path.Join(workspaceDir, "mod", service.Name))) + require.NoError(t, service.SaveAtDir(ctx, path.Join(workspaceDir, module, service.Name))) identity := &basev0.ServiceIdentity{ Name: service.Name, - Module: "mod", + Module: module, Workspace: workspace.Name, WorkspacePath: workspaceDir, - RelativeToWorkspace: fmt.Sprintf("mod/%s", service.Name), + RelativeToWorkspace: fmt.Sprintf("%s/%s", module, service.Name), } builder := NewBuilder() @@ -105,7 +111,7 @@ func newPostgresFixture(t *testing.T, ctx context.Context, runtimeContext *basev identity: identity, environment: environment, configuration: &basev0.Configuration{ - Origin: fmt.Sprintf("mod/%s", service.Name), + Origin: fmt.Sprintf("%s/%s", module, service.Name), RuntimeContext: resources.NewRuntimeContextFree(), Infos: []*basev0.ConfigurationInformation{ {Name: "postgres", diff --git a/main.go b/main.go index 285a45f..dd81d18 100644 --- a/main.go +++ b/main.go @@ -363,8 +363,11 @@ type Service struct { // Settings *Settings - postgresUser string - postgresPassword string + postgresUser string + postgresPassword string + // nativeOwner, when set, is the migration owner's role on the run's shared + // native server, and replaces the configured POSTGRES_USER there. + nativeOwner string readOnlyPassword string readWritePassword string // loginPasswords holds each declared runtime login's password by login name. @@ -434,6 +437,9 @@ func (s *Service) LoadConfiguration(ctx context.Context, conf *basev0.Configurat if err != nil { return s.Wool.Wrapf(err, "cannot get user") } + if s.nativeOwner != "" { + s.postgresUser = s.nativeOwner + } s.postgresPassword, err = resources.GetConfigurationValue(ctx, conf, "postgres", "POSTGRES_PASSWORD") if err != nil { return s.Wool.Wrapf(err, "cannot get password") diff --git a/main_test.go b/main_test.go index c2026f9..8fdb03b 100644 --- a/main_test.go +++ b/main_test.go @@ -80,7 +80,7 @@ func testCreateToRun(t *testing.T, runtimeContext *basev0.RuntimeContext) { require.Error(t, reader.CreateRelation(ctx, "reader_escape"), "reader must not create schema objects") require.Error(t, writer.CreateRelation(ctx, "writer_escape"), "writer must not create schema objects") require.Error(t, writer.CreateLoginRole(ctx, "writer_escape_role"), "writer must not create roles") - require.Error(t, writer.AssumeRole(ctx, "postgres"), "writer must not assume the migration owner") + require.Error(t, writer.AssumeRole(ctx, runtime.postgresUser), "writer must not assume the migration owner") owner, err := openPostgresCapabilityProbe(ctx, runtime.connection) require.NoError(t, err) @@ -289,7 +289,13 @@ func testCreateToRun(t *testing.T, runtimeContext *basev0.RuntimeContext) { } require.Error(t, err, "maintenance must not accept the migration owner") - assertExternalIdentityReconciliation(t, ctx, migrationControl) + // The external-identity reconciler is a deployment path; the local server + // stands in for the managed one, and its administrator for the provider's, + // which creates the cloud's login principals out-of-band. + providerControl, err := migrationtest.OpenControlPlane(ctx, runtime.extensionConnection()) + require.NoError(t, err) + defer providerControl.Close() + assertExternalIdentityReconciliation(t, ctx, providerControl) assertExternalIdentityTokenAuthentication(t, ctx, readOnlyConnection, readWriteConnection) assertSchemaPrerequisitesFailClosed(t, ctx, migrationControl, runtime.connection, readWriteConnection) @@ -464,10 +470,12 @@ func assertExternalIdentityReconciliation(t *testing.T, ctx context.Context, con readOnlyGroup, readWriteGroup := runtimeRoleNames(isolate.Name) readerPrincipal := readOnlyGroup + "_p" writerPrincipal := readWriteGroup + "_p" + var owner string + require.NoError(t, isolate.DB.QueryRowContext(ctx, `SELECT current_user`).Scan(&owner)) base := pgcontrol.RuntimeAccess{ Database: isolate.Name, - OwnerRole: "postgres", + OwnerRole: owner, ReadOnlyRole: readOnlyGroup, ReadWriteRole: readWriteGroup, Schemas: []string{"public"}, diff --git a/nixpg.go b/nixpg.go index 7393df1..a12a47e 100644 --- a/nixpg.go +++ b/nixpg.go @@ -1,24 +1,51 @@ package main -// nixpg.go — Docker-free postgres runtime. +// nixpg.go — the Docker-free postgres runtime: ONE server per run. // // The postgres service agent runs the database in a container by default -// (NewDockerHeadlessEnvironment). On hosts without Docker, the same agent can -// run postgres NATIVELY using a nix-provisioned binary: the codefly -// NixEnvironment materializes `postgresql_17` from the embedded flake (so no -// system install is required), and this file drives the native postgres -// lifecycle the Docker image's entrypoint would otherwise handle — initdb on -// first boot, launch `postgres`, and create the configured database. +// (NewDockerHeadlessEnvironment). On the native runtime it runs postgres from a +// nix-provisioned binary instead: the codefly NixEnvironment materializes +// `postgresql_17` (with pgvector) from the embedded flake, so no system install +// is required. // -// Both runtimes serve the same postgres 17 + the same connection string, so the -// rest of the agent (migrations, readiness, configuration) is unchanged. +// A composed run declares one Postgres service per database. A postmaster per +// service costs one System V shared-memory segment each (macOS allows 32 on the +// whole machine) and a cluster directory each, so a platform run exhausted the +// host. Every Postgres service of a run therefore shares ONE server, the way a +// deployed environment binds its databases to one instance: each service keeps +// its own database, its own migration owner and its own runtime logins, and +// none of them can connect to another's database. +// +// The run is the workspace and its naming scope, the identity Codefly already +// isolates concurrent runs by. Its server +// +// - is started by whichever of the run's agents initializes first, detached +// through pg_ctl so that no single agent's lifetime is the server's; +// - listens on no TCP port at all: it is reached through a private Unix +// socket, and each service's agent publishes the service's own allocated +// port by forwarding it to that socket (forward.go), so every address and +// connection string a consumer receives keeps its shape; +// - is administered by a bootstrap superuser whose password lives only in an +// owner-only file under the run's root; no service ever receives it; +// - records each agent that uses it (holders/) and is stopped by the last one +// to release it, which is what Stop, Destroy and an interrupted run's +// orderly shutdown all do; and +// - when every recorded holder is gone without releasing (the run was killed), +// is stopped by the next agent that initializes on this machine. +// +// Stopping the server releases its shared memory and semaphores; the cluster +// directory is retained, exactly as a per-service cluster was (no operation of +// this agent deletes data, codefly-dev/core#424). import ( + "bufio" "context" + "crypto/rand" "crypto/sha256" "database/sql" _ "embed" "encoding/hex" + "encoding/json" "errors" "fmt" "io" @@ -31,7 +58,7 @@ import ( "time" runners "github.com/codefly-dev/core/runners/base" - "github.com/codefly-dev/core/shared" + "github.com/gofrs/flock" "github.com/lib/pq" ) @@ -41,166 +68,350 @@ var pgFlakeNix string //go:embed nix/flake.lock var pgFlakeLock string -// nixPostgres runs a native postgres server off a nix-provisioned binary. +const ( + // nativeMajor is the PostgreSQL major the embedded flake provisions. It is + // part of a run server's identity: a cluster directory only opens under the + // major that initialized it. + nativeMajor = "17" + + // serverAdminRole is the run server's bootstrap superuser. Only the agents + // of the run hold its password; it provisions each service's database and + // owner, creates extensions and reconciles runtime grants. + serverAdminRole = "codefly_admin" + + // serverSocketPort names the socket file (.s.PGSQL.). The server + // binds no TCP port, so nothing is allocated for it. + serverSocketPort = 5432 + + // serverMaxConnections is the run server's connection budget. Each service + // used to have a server with PostgreSQL's default of 100 to itself; one + // server now carries every service of the run. + serverMaxConnections = 500 + + // serverStartSeconds bounds pg_ctl's wait for a started server to accept + // connections, and serverStopTimeout a fast shutdown. + serverStartSeconds = 60 + serverStopTimeout = 60 * time.Second + + // serverWatchInterval paces the supervision of the shared postmaster. + serverWatchInterval = 2 * time.Second +) + +// runServersDir is the machine directory holding every run server's root. +// Tests point it elsewhere. +var runServersDir = func() (string, error) { + cache, err := os.UserCacheDir() + if err != nil { + cache = os.TempDir() + } + return filepath.Join(cache, "codefly", "postgres", "run"), nil +} + +// runServerKey identifies a run's server: its workspace, its naming scope and +// the PostgreSQL major. The separators are unambiguous and the key is hashed +// before it names a directory. +func runServerKey(workspacePath, namingScope string) string { + return workspacePath + "\x00naming-scope\x00" + namingScope + "\x00postgres\x00" + nativeMajor +} + +// serviceHash is the hex sha256 of a key. +func serviceHash(key string) string { + sum := sha256.Sum256([]byte(key)) + return hex.EncodeToString(sum[:]) +} + +// nixPostgres is one agent's attachment to its run's server: the server's +// location and administration, this agent's holder record, and the forwarder +// that publishes this service's endpoint on it. type nixPostgres struct { - env *runners.NixEnvironment - flakeDir string - dataDir string - socketDir string - port uint16 - user string - password string - dbName string - logLevel string - out io.Writer - proc runners.Proc - // serverCancel cancels serverCtx — the context the postgres process runs - // under. It MUST outlive Init: starting postgres under the Init RPC's ctx - // kills it ("smart shutdown") the instant Init returns and that ctx is - // cancelled. Cancelled only by Stop. + root string + dataDir string + socketDir string + adminPassword string + holder string + forwarder *socketForwarder + + // serverCancel cancels serverCtx, which supervision watches: a postmaster + // that disappears while this agent still holds the server is reported, one + // that goes away after this agent released it is not. serverCtx context.Context serverCancel context.CancelFunc - // serverExit reports the process's real terminal result. Readiness observes - // it so a bind failure cannot be hidden by an unrelated database already - // listening on the proposed TCP port. + // serverExit reports the shared postmaster's disappearance. serverExit <-chan error - // binDir is the absolute nix store bin dir holding initdb + postgres. - // Resolved once after materialization and used for ALL postgres invocations - // so PATH contamination (e.g. a system Homebrew postgres) can never make - // initdb pick a different-version postgres than itself. - binDir string } -// newNixPostgres materializes the embedded flake and prepares a native postgres -// whose runtime state (data dir, nix cache, flake, unix socket) lives ENTIRELY -// OUTSIDE the user's source tree. -// -// stateKey is derived from the agent's service location and the authoritative -// Codefly naming scope. The source repo is later consumed as a nix flake `path:` -// input, so runtime state cannot live below it. The scoped key also prevents -// independent Codefly flows from sharing credentials or database contents, -// while an unscoped restart retains the historical per-location cluster. -func newNixPostgres(ctx context.Context, stateKey string, port uint16, user, password, dbName, logLevel string, out io.Writer) (*nixPostgres, error) { - if strings.TrimSpace(password) == "" { - return nil, fmt.Errorf("postgres password is required for the native runtime") - } - runtimeRoot, err := nixRuntimeRoot(stateKey) +// processMark is a process's identity as far as liveness goes: the PID plus +// the start facts that distinguish it from a later process reusing the number. +type processMark struct { + PID int `json:"pid"` + StartID uint64 `json:"start"` + BootID string `json:"boot"` +} + +func markProcess(pid int) (processMark, error) { + identity, err := runners.InspectProcess(pid) if err != nil { - return nil, err + return processMark{}, err } - flakeDir := filepath.Join(runtimeRoot, "nix") - if err := os.MkdirAll(flakeDir, 0o755); err != nil { - return nil, fmt.Errorf("create nix flake dir: %w", err) + return processMark{PID: pid, StartID: identity.StartID, BootID: identity.BootID}, nil +} + +// alive reports whether the marked process still runs. An inspection that +// fails for any reason other than the process being gone counts as alive: a +// wrong "dead" stops a server a live run is using. +func (m processMark) alive() bool { + identity, err := runners.InspectProcess(m.PID) + if err != nil { + return !errors.Is(err, runners.ErrProcessNotFound) } - if err := os.WriteFile(filepath.Join(flakeDir, "flake.nix"), []byte(pgFlakeNix), 0o644); err != nil { - return nil, fmt.Errorf("write flake.nix: %w", err) + return identity.StartID == m.StartID && identity.BootID == m.BootID +} + +// serverHolder records one agent using a run server: the agent and the +// codefly run that launched it. An agent whose run is gone is orphaned, and +// its run's server is no longer anybody's. +type serverHolder struct { + Service string `json:"service"` + Agent processMark `json:"agent"` + Run *processMark `json:"run,omitempty"` +} + +func (h serverHolder) alive() bool { + if !h.Agent.alive() { + return false } - if err := os.WriteFile(filepath.Join(flakeDir, "flake.lock"), []byte(pgFlakeLock), 0o644); err != nil { - return nil, fmt.Errorf("write flake.lock: %w", err) + return h.Run == nil || h.Run.alive() +} + +// acquireRunServer attaches this agent to its run's server, starting the +// server when no live one exists, and records this agent as a holder. +func acquireRunServer(ctx context.Context, key, service, logLevel string, out io.Writer) (_ *nixPostgres, acquireErr error) { + servers, err := runServersDir() + if err != nil { + return nil, err } - env, err := runners.NewNixEnvironment(ctx, flakeDir) + root := filepath.Join(servers, serviceHash(key)[:16]) + if err := os.MkdirAll(filepath.Join(root, "holders"), 0o700); err != nil { + return nil, fmt.Errorf("create run server directory: %w", err) + } + collectAbandonedServers(ctx, servers, root) + + unlock, err := lockRunServer(ctx, root) if err != nil { - return nil, fmt.Errorf("nix environment (is nix installed?): %w", err) + return nil, err } - env.WithCacheDir(filepath.Join(runtimeRoot, ".nix-cache")) + defer unlock() - // The unix socket path has a ~104-char limit, so it cannot live under a deep - // cache path. Use an unpredictable, owner-only temporary directory: a stable - // /tmp name can be pre-created or symlinked by another local user. - socketDir, err := os.MkdirTemp("", "cfpg-"+serviceHash(stateKey)[:8]+"-") + n := &nixPostgres{root: root, dataDir: filepath.Join(root, "pgdata")} + if _, err := pruneHolders(root); err != nil { + return nil, err + } + pid, socketDir, running := runningPostmaster(n.dataDir) + if !running { + if err := n.start(ctx, logLevel, out); err != nil { + return nil, err + } + if pid, socketDir, running = runningPostmaster(n.dataDir); !running { + return nil, fmt.Errorf("postgres reported started but %s names no live postmaster; see %s", + filepath.Join(n.dataDir, "postmaster.pid"), n.logPath()) + } + } + n.socketDir = socketDir + if n.adminPassword, err = n.readAdminPassword(); err != nil { + return nil, err + } + postmaster, err := markProcess(pid) if err != nil { - return nil, fmt.Errorf("create postgres socket dir: %w", err) + return nil, fmt.Errorf("inspect the run's postmaster (pid %d): %w", pid, err) } - if err := os.Chmod(socketDir, 0o700); err != nil { - return nil, fmt.Errorf("secure postgres socket dir: %w", err) + if err := n.ping(ctx); err != nil { + return nil, fmt.Errorf("the run's postgres server (pid %d) does not answer on %s: %w; see %s", + pid, n.socketDir, err, n.logPath()) } - - return &nixPostgres{ - env: env, - flakeDir: flakeDir, - dataDir: filepath.Join(runtimeRoot, "pgdata"), - socketDir: socketDir, - port: port, - user: user, - password: password, - dbName: dbName, - logLevel: logLevel, - out: out, - }, nil + if err := n.register(service); err != nil { + return nil, err + } + n.watch(postmaster) + return n, nil } -// nixPostgresStateKey preserves the existing unscoped data location while -// making Codefly's advanced naming-scope isolation apply equally to the Nix -// and Docker Postgres runtimes. The separator is unambiguous and the resulting -// value is hashed before it is used as a filesystem component. -func nixPostgresStateKey(serviceLocation, namingScope string) string { - if namingScope == "" { - return serviceLocation +// lockRunServer serializes every start, attach, release and collection of one +// run server across the run's agent processes. +func lockRunServer(ctx context.Context, root string) (func(), error) { + lock := flock.New(filepath.Join(root, "lock")) + locked, err := lock.TryLockContext(ctx, 100*time.Millisecond) + if err != nil { + return nil, fmt.Errorf("lock the run's postgres server: %w", err) + } + if !locked { + return nil, fmt.Errorf("lock the run's postgres server: not acquired") } - return serviceLocation + "\x00naming-scope\x00" + namingScope + return func() { _ = lock.Unlock() }, nil } -// serviceHash is the hex sha256 of a service's scoped state key. -func serviceHash(stateKey string) string { - sum := sha256.Sum256([]byte(stateKey)) - return hex.EncodeToString(sum[:]) -} +func (n *nixPostgres) logPath() string { return filepath.Join(n.root, "server.log") } -// nixRuntimeRoot is the stable, out-of-source runtime root for a postgres -// service (data dir, nix cache, flake), keyed by its scoped state identity so -// same-scope restarts reuse the cluster without crossing isolation scopes. -// Falls back to the OS temp dir if no user cache dir is available. -func nixRuntimeRoot(stateKey string) (string, error) { - cache, err := os.UserCacheDir() +// start initializes the cluster on first use and launches it detached. The +// caller holds the run server's lock. +func (n *nixPostgres) start(ctx context.Context, logLevel string, out io.Writer) error { + flakeDir := filepath.Join(n.root, "nix") + if err := os.MkdirAll(flakeDir, 0o700); err != nil { + return fmt.Errorf("create nix flake dir: %w", err) + } + if err := os.WriteFile(filepath.Join(flakeDir, "flake.nix"), []byte(pgFlakeNix), 0o644); err != nil { + return fmt.Errorf("write flake.nix: %w", err) + } + if err := os.WriteFile(filepath.Join(flakeDir, "flake.lock"), []byte(pgFlakeLock), 0o644); err != nil { + return fmt.Errorf("write flake.lock: %w", err) + } + env, err := runners.NewNixEnvironment(ctx, flakeDir) if err != nil { - cache = os.TempDir() + return fmt.Errorf("nix environment (is nix installed?): %w", err) } - return filepath.Join(cache, "codefly", "postgres", serviceHash(stateKey)[:16]), nil -} - -// Init materializes the nix env, initdb's the cluster on first boot, launches -// postgres in the background, waits for readiness, and creates the database. -func (n *nixPostgres) Init(ctx context.Context) (initErr error) { - defer func() { - if initErr == nil { - return - } - cleanupCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - initErr = errors.Join(initErr, n.Stop(cleanupCtx)) - }() - if err := n.env.Init(ctx); err != nil { + env.WithCacheDir(filepath.Join(n.root, ".nix-cache")) + if err := env.Init(ctx); err != nil { return fmt.Errorf("materialize nix postgres env: %w", err) } - if err := n.resolveBinDir(); err != nil { + binDir, err := resolveBinDir() + if err != nil { return err } - if err := n.initdbIfNeeded(ctx); err != nil { + if err := n.initdbIfNeeded(ctx, env, binDir, out); err != nil { return err } if err := n.clearStalePostmasterPid(); err != nil { return err } - if err := n.startServer(ctx); err != nil { + // The socket path is limited to ~104 bytes, so it cannot live under a deep + // cache path. An unpredictable, owner-only temporary directory: a stable + // name can be pre-created or symlinked by another local user. + socketDir, err := os.MkdirTemp("", "cfpg-"+filepath.Base(n.root)[:8]+"-") + if err != nil { + return fmt.Errorf("create postgres socket dir: %w", err) + } + if err := os.Chmod(socketDir, 0o700); err != nil { + _ = os.RemoveAll(socketDir) + return fmt.Errorf("secure postgres socket dir: %w", err) + } + options, err := serverOptions(socketDir, logLevel) + if err != nil { + _ = os.RemoveAll(socketDir) + return err + } + // pg_ctl detaches the postmaster into its own session and waits until it + // accepts connections; the server outlives the agent that started it. + proc, err := env.NewProcess(filepath.Join(binDir, "pg_ctl"), + "start", "-D", n.dataDir, "-w", "-t", strconv.Itoa(serverStartSeconds), "-l", n.logPath(), "-o", options) + if err != nil { + _ = os.RemoveAll(socketDir) return err } - if err := n.waitReady(ctx); err != nil { + if out != nil { + proc.WithOutput(out) + } + if err := proc.Run(ctx); err != nil { + _ = os.RemoveAll(socketDir) + return fmt.Errorf("start the run's postgres server: %w%s", err, n.logTail()) + } + return n.restrictMaintenanceDatabases(ctx, socketDir) +} + +// serverOptions are the postmaster's options, passed through pg_ctl's shell. +// No TCP listener; the private socket directory; the run-wide connection +// budget; and the starting service's quietening when it configured one. +func serverOptions(socketDir, logLevel string) (string, error) { + if strings.ContainsAny(socketDir, " \t\n'\"\\$`") { + return "", fmt.Errorf("postgres socket dir %q cannot be passed to pg_ctl", socketDir) + } + options := []string{ + "-c listen_addresses=''", + "-k " + socketDir, + "-p " + strconv.Itoa(serverSocketPort), + "-c max_connections=" + strconv.Itoa(serverMaxConnections), + } + if lvl := strings.ToLower(strings.TrimSpace(logLevel)); lvl != "" { + for _, r := range lvl { + if (r < 'a' || r > 'z') && (r < '0' || r > '9') { + return "", fmt.Errorf("log-level %q is not a postgres log level", logLevel) + } + } + options = append(options, + "-c log_min_messages="+lvl, + "-c log_statement=none", + "-c log_connections=off", + "-c log_disconnections=off", + ) + } + return strings.Join(options, " "), nil +} + +// logTail is the end of the server log, for a refusal that has to say why. +func (n *nixPostgres) logTail() string { + data, err := os.ReadFile(n.logPath()) + if err != nil || len(data) == 0 { + return "" + } + lines := strings.Split(strings.TrimRight(string(data), "\n"), "\n") + if len(lines) > 20 { + lines = lines[len(lines)-20:] + } + return "\nserver log (" + n.logPath() + "):\n" + strings.Join(lines, "\n") +} + +// restrictMaintenanceDatabases closes the maintenance and template databases +// to every role but the administrator, so a service's logins reach only the +// database granted to them. +func (n *nixPostgres) restrictMaintenanceDatabases(ctx context.Context, socketDir string) error { + n.socketDir = socketDir + password, err := n.readAdminPassword() + if err != nil { return err } - if err := n.ensureAuthentication(ctx); err != nil { + n.adminPassword = password + db, err := sql.Open("postgres", n.adminDSN("postgres")) + if err != nil { return err } - return n.ensureDatabase(ctx) + defer db.Close() + for _, statement := range []string{ + `REVOKE ALL ON DATABASE postgres FROM PUBLIC`, + `REVOKE ALL ON DATABASE template1 FROM PUBLIC`, + } { + if _, err := db.ExecContext(ctx, statement); err != nil { + return fmt.Errorf("restrict the run server's maintenance databases: %w", err) + } + } + return nil +} + +// runningPostmaster reads the cluster's postmaster.pid: the live postmaster's +// PID and the socket directory it listens in. +func runningPostmaster(dataDir string) (int, string, bool) { + file, err := os.Open(filepath.Join(dataDir, "postmaster.pid")) + if err != nil { + return 0, "", false + } + defer file.Close() + var lines []string + scanner := bufio.NewScanner(file) + for scanner.Scan() && len(lines) < 5 { + lines = append(lines, strings.TrimSpace(scanner.Text())) + } + if len(lines) < 5 { + return 0, "", false + } + pid, err := strconv.Atoi(lines[0]) + if err != nil || pid <= 0 || !processAlive(pid) { + return 0, "", false + } + return pid, lines[4], lines[4] != "" } // clearStalePostmasterPid removes a leftover postmaster.pid when no live -// postmaster owns it. The nix runner launches `postgres` directly (not -// pg_ctl), so it must replicate pg_ctl's stale-lock cleanup — otherwise a -// crashed or SIGKILL'd prior run (codefly currently reaps orphaned nix deps -// only on the NEXT startup, after a fresh run already tried to start) leaves a -// pid file that aborts boot: `FATAL: lock file "postmaster.pid" is empty` / -// "is the remnant of a previous server startup crash". A pid file owned by a -// LIVE process is left untouched so we never stomp a concurrent postmaster. +// postmaster owns it. A crashed or SIGKILL'd server leaves a pid file that +// aborts the next boot (`FATAL: lock file "postmaster.pid" is empty` / "is the +// remnant of a previous server startup crash"). A pid file owned by a LIVE +// process is left untouched so a concurrent postmaster is never stomped. func (n *nixPostgres) clearStalePostmasterPid() error { pidPath := filepath.Join(n.dataDir, "postmaster.pid") data, err := os.ReadFile(pidPath) @@ -231,27 +442,32 @@ func processAlive(pid int) bool { } // initdbIfNeeded creates the cluster the first time (PG_VERSION marks an -// initialized data dir). Both local and TCP connections require SCRAM; the -// bootstrap password is passed through an owner-only temporary file, never -// argv or the environment. -func (n *nixPostgres) initdbIfNeeded(ctx context.Context) error { +// initialized data dir), with a freshly generated administrator password. +// Both local and TCP connections require SCRAM; the password reaches initdb +// through an owner-only file, never argv or the environment. +func (n *nixPostgres) initdbIfNeeded(ctx context.Context, env *runners.NixEnvironment, binDir string, out io.Writer) error { if _, err := os.Stat(filepath.Join(n.dataDir, "PG_VERSION")); err == nil { - return nil // already initialized + return nil + } + if err := os.RemoveAll(n.dataDir); err != nil { + return fmt.Errorf("clear an uninitialized data dir: %w", err) } if err := os.MkdirAll(n.dataDir, 0o700); err != nil { return fmt.Errorf("create data dir: %w", err) } - passwordFile, err := n.writeInitPasswordFile() - if err != nil { - return err + secret := make([]byte, 32) + if _, err := rand.Read(secret); err != nil { + return fmt.Errorf("generate the run server's administrator password: %w", err) } - defer os.Remove(passwordFile) - proc, err := n.env.NewProcess(filepath.Join(n.binDir, "initdb"), n.initdbArgs(passwordFile)...) + if err := writeOwnerOnly(n.adminPasswordPath(), hex.EncodeToString(secret)+"\n"); err != nil { + return fmt.Errorf("write the run server's administrator password: %w", err) + } + proc, err := env.NewProcess(filepath.Join(binDir, "initdb"), n.initdbArgs(n.adminPasswordPath())...) if err != nil { return err } - if n.out != nil { - proc.WithOutput(n.out) + if out != nil { + proc.WithOutput(out) } if err := proc.Run(ctx); err != nil { return fmt.Errorf("initdb: %w", err) @@ -262,7 +478,7 @@ func (n *nixPostgres) initdbIfNeeded(ctx context.Context) error { func (n *nixPostgres) initdbArgs(passwordFile string) []string { return []string{ "-D", n.dataDir, - "-U", n.user, + "-U", serverAdminRole, "--auth-local=scram-sha-256", "--auth-host=scram-sha-256", "--pwfile=" + passwordFile, @@ -271,88 +487,221 @@ func (n *nixPostgres) initdbArgs(passwordFile string) []string { } } -func (n *nixPostgres) writeInitPasswordFile() (string, error) { - file, err := os.CreateTemp(filepath.Dir(n.dataDir), ".initdb-password-*") +func (n *nixPostgres) adminPasswordPath() string { return filepath.Join(n.root, "admin.password") } + +func (n *nixPostgres) readAdminPassword() (string, error) { + data, err := os.ReadFile(n.adminPasswordPath()) if err != nil { - return "", fmt.Errorf("create initdb password file: %w", err) + return "", fmt.Errorf("read the run server's administrator password (remove %s to start this run's databases over): %w", n.root, err) } - path := file.Name() - if err := file.Chmod(0o600); err != nil { - _ = file.Close() - _ = os.Remove(path) - return "", fmt.Errorf("secure initdb password file: %w", err) + password := strings.TrimSpace(string(data)) + if password == "" { + return "", fmt.Errorf("the run server's administrator password at %s is empty", n.adminPasswordPath()) } - if _, err := io.WriteString(file, n.password+"\n"); err != nil { - _ = file.Close() - _ = os.Remove(path) - return "", fmt.Errorf("write initdb password file: %w", err) + return password, nil +} + +func writeOwnerOnly(path, content string) error { + temporary, err := os.CreateTemp(filepath.Dir(path), "."+filepath.Base(path)+"-*") + if err != nil { + return err } - if err := file.Close(); err != nil { - _ = os.Remove(path) - return "", fmt.Errorf("close initdb password file: %w", err) + name := temporary.Name() + defer os.Remove(name) + if err := temporary.Chmod(0o600); err != nil { + _ = temporary.Close() + return err + } + if _, err := io.WriteString(temporary, content); err != nil { + _ = temporary.Close() + return err + } + if err := temporary.Close(); err != nil { + return err } - return path, nil + return os.Rename(name, path) } -// startServer launches postgres listening on 127.0.0.1:port. Quietened to the -// configured log level when set (mirrors the Docker -c flags). -func (n *nixPostgres) startServer(ctx context.Context) error { - args := []string{ - "-D", n.dataDir, - "-p", fmt.Sprintf("%d", n.port), - "-k", n.socketDir, - "-c", "listen_addresses=127.0.0.1", - } - if lvl := strings.ToLower(strings.TrimSpace(n.logLevel)); lvl != "" { - args = append(args, - "-c", "log_min_messages="+lvl, - "-c", "log_statement=none", - "-c", "log_connections=off", - "-c", "log_disconnections=off", - ) +// adminDSN reaches one database of the run's server as its administrator, +// through the private socket. Bootstrap work never uses a public TCP mapping: +// another process can own that port, and accepting its successful ping would +// authenticate, migrate and report readiness for the wrong database. +func (n *nixPostgres) adminDSN(database string) string { + u := &url.URL{ + Scheme: "postgresql", + User: url.UserPassword(serverAdminRole, n.adminPassword), + Path: "/" + database, } - proc, err := n.env.NewProcess(filepath.Join(n.binDir, "postgres"), args...) + query := u.Query() + query.Set("host", n.socketDir) + query.Set("port", strconv.Itoa(serverSocketPort)) + query.Set("sslmode", "disable") + u.RawQuery = query.Encode() + return u.String() +} + +func (n *nixPostgres) socketPath() string { + return filepath.Join(n.socketDir, ".s.PGSQL."+strconv.Itoa(serverSocketPort)) +} + +func (n *nixPostgres) ping(ctx context.Context) error { + db, err := sql.Open("postgres", n.adminDSN("postgres")) if err != nil { return err } - if n.out != nil { - proc.WithOutput(n.out) + defer db.Close() + pingCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + return db.PingContext(pingCtx) +} + +// provisionDatabase gives one service its database on the run's server: the +// database, owned by the service's own migration owner, closed to every other +// role. The owner may create roles (its migrations create the application's +// NOLOGIN roles) and databases (the migration test control plane creates +// isolated ones it owns), but it is not a superuser: it cannot connect to +// another service's database, copy it, or alter another service's roles. A +// database another service of the run already claimed is refused by name. +func (n *nixPostgres) provisionDatabase(ctx context.Context, service, database, owner, password string) error { + db, err := sql.Open("postgres", n.adminDSN("postgres")) + if err != nil { + return err } - // Run postgres under a context that outlives Init — NOT the Init RPC ctx, - // which is cancelled the moment Init returns and would SIGTERM the server. - n.serverCtx, n.serverCancel = context.WithCancel(context.Background()) - if err := proc.Start(n.serverCtx); err != nil { - n.serverCancel() - return fmt.Errorf("start postgres: %w", err) + defer db.Close() + + var exists bool + if err := db.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = $1)`, owner).Scan(&exists); err != nil { + return fmt.Errorf("look up the migration owner: %w", err) + } + quotedOwner := pq.QuoteIdentifier(owner) + if !exists { + if _, err := db.ExecContext(ctx, `CREATE ROLE `+quotedOwner); err != nil { + return fmt.Errorf("create the migration owner: %w", err) + } + } + if _, err := db.ExecContext(ctx, `ALTER ROLE `+quotedOwner+ + ` WITH LOGIN NOSUPERUSER CREATEROLE CREATEDB NOREPLICATION NOBYPASSRLS PASSWORD `+pq.QuoteLiteral(password)); err != nil { + return fmt.Errorf("configure the migration owner: %w", err) + } + + marker := databaseClaim(service) + var claim sql.NullString + err = db.QueryRowContext(ctx, + `SELECT shobj_description(oid, 'pg_database') FROM pg_database WHERE datname = $1`, database).Scan(&claim) + quotedDatabase := pq.QuoteIdentifier(database) + switch { + case errors.Is(err, sql.ErrNoRows): + if _, err := db.ExecContext(ctx, `CREATE DATABASE `+quotedDatabase+` OWNER `+quotedOwner); err != nil { + return fmt.Errorf("create database %q: %w", database, err) + } + if _, err := db.ExecContext(ctx, `COMMENT ON DATABASE `+quotedDatabase+` IS `+pq.QuoteLiteral(marker)); err != nil { + return fmt.Errorf("record database %q's service: %w", database, err) + } + case err != nil: + return fmt.Errorf("look up database %q: %w", database, err) + case claim.String != marker: + return fmt.Errorf("database %q on this run's postgres server belongs to %s, not %s: give each service of a run its own database-name", + database, strings.TrimPrefix(claim.String, databaseClaimPrefix), service) + } + for _, statement := range []string{ + `ALTER DATABASE ` + quotedDatabase + ` OWNER TO ` + quotedOwner, + `REVOKE ALL ON DATABASE ` + quotedDatabase + ` FROM PUBLIC`, + } { + if _, err := db.ExecContext(ctx, statement); err != nil { + return fmt.Errorf("restrict database %q to its service: %w", database, err) + } + } + return nil +} + +const databaseClaimPrefix = "codefly service " + +func databaseClaim(service string) string { return databaseClaimPrefix + service } + +// register records this agent, and the run that launched it, as a holder. +func (n *nixPostgres) register(service string) error { + agent, err := markProcess(os.Getpid()) + if err != nil { + return fmt.Errorf("inspect this agent: %w", err) + } + holder := serverHolder{Service: service, Agent: agent} + if parent := os.Getppid(); parent > 1 { + if run, err := markProcess(parent); err == nil { + holder.Run = &run + } + } + data, err := json.Marshal(holder) + if err != nil { + return err + } + n.holder = filepath.Join(n.root, "holders", serviceHash(service)[:16]+".json") + return writeOwnerOnly(n.holder, string(data)) +} + +// pruneHolders removes the records of agents that are gone and returns how +// many live ones remain. The caller holds the run server's lock. +func pruneHolders(root string) (int, error) { + dir := filepath.Join(root, "holders") + entries, err := os.ReadDir(dir) + if errors.Is(err, os.ErrNotExist) { + return 0, nil + } + if err != nil { + return 0, fmt.Errorf("read the run server's holders: %w", err) + } + live := 0 + for _, entry := range entries { + if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".json") { + continue + } + path := filepath.Join(dir, entry.Name()) + data, err := os.ReadFile(path) + var holder serverHolder + if err == nil && json.Unmarshal(data, &holder) == nil && holder.alive() { + live++ + continue + } + _ = os.Remove(path) } - n.proc = proc + return live, nil +} + +// watch supervises the shared postmaster for as long as this agent holds it. +func (n *nixPostgres) watch(postmaster processMark) { + n.serverCtx, n.serverCancel = context.WithCancel(context.Background()) exit := make(chan error, 1) - go func() { - exit <- proc.Wait(context.Background()) - close(exit) - }() n.serverExit = exit - return nil + go func(ctx context.Context) { + ticker := time.NewTicker(serverWatchInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if !postmaster.alive() { + exit <- fmt.Errorf("the run's postgres server (pid %d) is gone; see %s", postmaster.PID, n.logPath()) + close(exit) + return + } + } + } + }(n.serverCtx) } -// Supervise watches the running postgres process after startup and invokes -// onExit exactly once if the postmaster terminates for any reason OTHER than a -// deliberate Stop. The Nix runtime is a host process the agent owns directly: -// without this the postmaster can exit mid-run (crash, an external SIGTERM, an -// invalidated data-directory lock) while the agent keeps reporting STARTED, +// Supervise invokes onExit exactly once if the run's postmaster terminates +// while this agent still holds it. Without it the server can disappear mid-run +// (a crash, an external signal) while the agent keeps reporting STARTED, // leaving codefly's Follow loop blind and every dependent wedged on connection -// refused (codefly-dev/cli#380). This mirrors the user-binary supervision -// service-go performs by feeding the exit into MarkRunnerExited. +// refused (codefly-dev/cli#380). // -// Call it once, after a successful Init/Start: startServer has populated -// serverExit and serverCtx, and waitReady has finished observing serverExit, so -// this goroutine is the sole remaining reader of the channel. +// Call it once, after a successful Init/Start. func (n *nixPostgres) Supervise(onExit func(error)) { exit := n.serverExit if exit == nil { return } - // Stop cancels serverCtx before terminating the process, so a cancelled + // release cancels serverCtx before the server can stop, so a cancelled // context is the authoritative "this shutdown was orderly" signal. var stopped <-chan struct{} if n.serverCtx != nil { @@ -374,174 +723,131 @@ func (n *nixPostgres) Supervise(onExit func(error)) { }() } -// adminDSN reaches this runtime's private Unix socket. Bootstrap work must not -// use the public TCP mapping: another backend or stale process can still own -// that port, and accepting its successful ping would authenticate, migrate, -// and report readiness for the wrong database. -func (n *nixPostgres) adminDSN() string { - u := &url.URL{ - Scheme: "postgresql", - User: url.UserPassword(n.user, n.password), - Path: "/postgres", - } - query := u.Query() - query.Set("host", n.socketDir) - query.Set("port", strconv.Itoa(int(n.port))) - query.Set("sslmode", "disable") - u.RawQuery = query.Encode() - return u.String() -} - -// ensureAuthentication upgrades legacy trust-initialized clusters in place -// and pins pg_hba.conf to password authentication. Updating the role before -// reloading the HBA keeps existing developer data while closing the old -// unauthenticated local/TCP access. -func (n *nixPostgres) ensureAuthentication(ctx context.Context) error { - db, err := sql.Open("postgres", n.adminDSN()) - if err != nil { - return fmt.Errorf("open postgres for auth hardening: %w", err) +// release closes this service's endpoint and removes this agent's holder +// record; the last holder stops the server. It reports whether it did. +func (n *nixPostgres) release(ctx context.Context) (bool, error) { + if n.serverCancel != nil { + n.serverCancel() } - defer db.Close() - statement := fmt.Sprintf("ALTER ROLE %s WITH PASSWORD %s", pq.QuoteIdentifier(n.user), pq.QuoteLiteral(n.password)) - if _, err := db.ExecContext(ctx, statement); err != nil { - return fmt.Errorf("set postgres role password: %w", err) + var forwardErr error + if n.forwarder != nil { + forwardErr = n.forwarder.Close() + n.forwarder = nil } - hba := []byte("local all all scram-sha-256\n" + - "host all all 127.0.0.1/32 scram-sha-256\n" + - "host all all ::1/128 scram-sha-256\n") - hbaPath := filepath.Join(n.dataDir, "pg_hba.conf") - if err := shared.WriteFileAtomic(ctx, hbaPath, hba, 0o600); err != nil { - return fmt.Errorf("write secure pg_hba.conf: %w", err) + if n.root == "" { + return false, forwardErr } - if _, err := db.ExecContext(ctx, "SELECT pg_reload_conf()"); err != nil { - return fmt.Errorf("reload secure pg_hba.conf: %w", err) + unlock, err := lockRunServer(ctx, n.root) + if err != nil { + return false, errors.Join(forwardErr, err) } - return nil -} - -// waitReady polls this process's private socket and also observes its exit. A -// public TCP ping is insufficient because a stale Docker proxy can own the -// proposed port while the newly launched Nix postgres dies during bind. -func (n *nixPostgres) waitReady(ctx context.Context) error { - deadline := time.NewTimer(30 * time.Second) - defer deadline.Stop() - retry := time.NewTicker(500 * time.Millisecond) - defer retry.Stop() - var lastErr error - for { - select { - case exitErr := <-n.serverExit: - return postgresExitedBeforeReady(exitErr) - default: - } - db, err := sql.Open("postgres", n.adminDSN()) - if err == nil { - pingCtx, cancel := context.WithTimeout(ctx, 2*time.Second) - lastErr = db.PingContext(pingCtx) - cancel() - _ = db.Close() - if lastErr == nil { - select { - case exitErr := <-n.serverExit: - return postgresExitedBeforeReady(exitErr) - default: - return nil - } - } - } else { - lastErr = err - } - select { - case exitErr := <-n.serverExit: - return postgresExitedBeforeReady(exitErr) - case <-ctx.Done(): - return fmt.Errorf("wait for postgres readiness: %w", ctx.Err()) - case <-deadline.C: - if lastErr == nil { - lastErr = errors.New("no readiness probe completed") - } - return fmt.Errorf("postgres did not become ready: %w", lastErr) - case <-retry.C: + defer unlock() + if n.holder != "" { + if err := os.Remove(n.holder); err != nil && !errors.Is(err, os.ErrNotExist) { + return false, errors.Join(forwardErr, err) } + n.holder = "" } -} - -func postgresExitedBeforeReady(err error) error { - if err == nil { - return errors.New("postgres exited before becoming ready") + live, err := pruneHolders(n.root) + if err != nil || live > 0 { + return false, errors.Join(forwardErr, err) } - return fmt.Errorf("postgres exited before becoming ready: %w", err) + stopped, err := stopServer(n.dataDir) + return stopped, errors.Join(forwardErr, err) } -// ensureDatabase creates the configured database if it does not exist (the -// Docker image does this via POSTGRES_DB; natively we do it ourselves). -func (n *nixPostgres) ensureDatabase(ctx context.Context) error { - db, err := sql.Open("postgres", n.adminDSN()) +// stopServer shuts the cluster's postmaster down (fast shutdown: sessions are +// disconnected, the server exits cleanly and removes its shared memory) and +// removes its socket directory. It reports whether a server was running. +func stopServer(dataDir string) (bool, error) { + pid, socketDir, running := runningPostmaster(dataDir) + if !running { + return false, nil + } + process, err := os.FindProcess(pid) if err != nil { - return err + return false, err } - defer func() { _ = db.Close() }() - - var exists bool - if err := db.QueryRowContext(ctx, "SELECT EXISTS (SELECT 1 FROM pg_database WHERE datname=$1)", n.dbName).Scan(&exists); err != nil { - return fmt.Errorf("check database exists: %w", err) + if err := process.Signal(syscall.SIGINT); err != nil && processAlive(pid) { + return false, fmt.Errorf("stop the run's postgres server (pid %d): %w", pid, err) } - if exists { - return nil + if !waitGone(pid, serverStopTimeout) { + // Immediate shutdown: the server still releases its shared memory. + _ = process.Signal(syscall.SIGQUIT) + if !waitGone(pid, 10*time.Second) { + return false, fmt.Errorf("the run's postgres server (pid %d) did not stop", pid) + } } - // Database names cannot be parameterized; quote to avoid injection. - if _, err := db.ExecContext(ctx, fmt.Sprintf("CREATE DATABASE %s", pgQuoteIdent(n.dbName))); err != nil { - return fmt.Errorf("create database %q: %w", n.dbName, err) + if strings.HasPrefix(filepath.Base(socketDir), "cfpg-") { + _ = os.RemoveAll(socketDir) } - return nil + return true, nil } -// Stop terminates the postgres server. -func (n *nixPostgres) Stop(ctx context.Context) error { - if n.serverCancel != nil { - n.serverCancel() +func waitGone(pid int, budget time.Duration) bool { + deadline := time.Now().Add(budget) + for time.Now().Before(deadline) { + if !processAlive(pid) { + return true + } + time.Sleep(100 * time.Millisecond) } - var stopErr error - if n.proc != nil { - stopErr = n.proc.Stop(ctx) + return !processAlive(pid) +} + +// collectAbandonedServers stops every other run server on this machine whose +// holders are all gone: a run that was killed rather than stopped. A server +// whose lock is taken is being started, attached or released right now, and is +// left alone. +func collectAbandonedServers(ctx context.Context, servers, self string) { + entries, err := os.ReadDir(servers) + if err != nil { + return + } + for _, entry := range entries { + root := filepath.Join(servers, entry.Name()) + if !entry.IsDir() || root == self || ctx.Err() != nil { + continue + } + lock := flock.New(filepath.Join(root, "lock")) + locked, err := lock.TryLock() + if err != nil || !locked { + continue + } + if live, err := pruneHolders(root); err == nil && live == 0 { + _, _ = stopServer(filepath.Join(root, "pgdata")) + } + _ = lock.Unlock() } - removeErr := os.RemoveAll(n.socketDir) - return errors.Join(stopErr, removeErr) } // resolveBinDir locates the nix store bin dir that holds postgres 17's initdb, -// so every invocation uses that exact build. env.Init has already materialized -// (downloaded) the package, so it is present in the store. Using the absolute -// dir — rather than the bare command on PATH — guarantees initdb and postgres -// are the same version even when a different system postgres shadows PATH. -func (n *nixPostgres) resolveBinDir() error { +// pg_ctl and postgres, so every invocation uses that exact build. The package +// is materialized by then. Using the absolute dir — rather than the bare +// command on PATH — guarantees one version even when a different system +// postgres shadows PATH. +func resolveBinDir() (string, error) { // Broad glob: match BOTH the plain `postgresql-17*` build AND the - // `withPackages` wrapper (`postgresql-and-plugins-17*`) that bundles - // pgvector. The old pattern `*-postgresql-17*` did NOT match the wrapper - // (its name is "postgresql-and-plugins-17…", which has no "postgresql-17" - // substring), so it always resolved to the PLAIN build — whose extension - // dir has no vector.control. Mind's KG migration does `CREATE EXTENSION - // vector`, which then aborts mid-migration (golang-migrate goes dirty) and - // kg_nodes is never created. - matches, err := filepath.Glob("/nix/store/*postgresql*17*/bin/initdb") + // `withPackages` wrapper (`postgresql-and-plugins-17…`) that bundles + // pgvector, whose name has no "postgresql-17" substring. + matches, err := filepath.Glob("/nix/store/*postgresql*" + nativeMajor + "*/bin/initdb") if err != nil { - return fmt.Errorf("glob nix postgres: %w", err) + return "", fmt.Errorf("glob nix postgres: %w", err) } var fallback string for _, m := range matches { binDir := filepath.Dir(m) - // Must have the postgres server binary alongside initdb (skip + // Must have the server and its control binary alongside initdb (skip // lib/man/doc-only outputs). - if _, err := os.Stat(filepath.Join(binDir, "postgres")); err != nil { + if !hasFiles(binDir, "postgres", "pg_ctl") { continue } - // PREFER a build whose extension dir ships pgvector — that is the - // withPackages wrapper the flake builds. /bin/initdb → - // /share/postgresql/extension/vector.control. + // PREFER a build whose extension dir ships pgvector — the withPackages + // wrapper the flake builds. A plain build would make CREATE EXTENSION + // vector fail mid-migration. vectorControl := filepath.Join(filepath.Dir(binDir), "share", "postgresql", "extension", "vector.control") if _, err := os.Stat(vectorControl); err == nil { - n.binDir = binDir - return nil + return binDir, nil } if fallback == "" { fallback = binDir @@ -549,15 +855,17 @@ func (n *nixPostgres) resolveBinDir() error { } if fallback != "" { // No pgvector-enabled build found — use a plain one so non-vector - // schemas still work; CREATE EXTENSION vector will then fail loudly - // rather than us silently skipping a vector build that does exist. - n.binDir = fallback - return nil + // schemas still work; CREATE EXTENSION vector will then fail loudly. + return fallback, nil } - return fmt.Errorf("no nix postgresql-17 with both initdb+postgres found in /nix/store (materialization may have failed)") + return "", fmt.Errorf("no nix postgresql-%s with initdb, pg_ctl and postgres found in /nix/store (materialization may have failed)", nativeMajor) } -// pgQuoteIdent double-quotes a postgres identifier, escaping embedded quotes. -func pgQuoteIdent(name string) string { - return `"` + strings.ReplaceAll(name, `"`, `""`) + `"` +func hasFiles(dir string, names ...string) bool { + for _, name := range names { + if _, err := os.Stat(filepath.Join(dir, name)); err != nil { + return false + } + } + return true } diff --git a/nixpg_auth_test.go b/nixpg_auth_test.go index 31b17b5..546d479 100644 --- a/nixpg_auth_test.go +++ b/nixpg_auth_test.go @@ -1,7 +1,6 @@ package main import ( - "context" "net/url" "os" "path/filepath" @@ -12,23 +11,14 @@ import ( "github.com/lib/pq" ) -func TestNewNixPostgresRejectsEmptyPasswordBeforeProvisioning(t *testing.T) { - if _, err := newNixPostgres(context.Background(), t.TempDir(), 5432, "postgres", "", "postgres", "", nil); err == nil { - t.Fatal("native postgres accepted an empty password") - } -} - func TestInitdbUsesPrivatePasswordFileAndSCRAM(t *testing.T) { root := t.TempDir() - n := &nixPostgres{dataDir: filepath.Join(root, "data"), user: "postgres", password: `s p#ss"word`} - if err := os.MkdirAll(n.dataDir, 0o700); err != nil { + n := &nixPostgres{root: root, dataDir: filepath.Join(root, "pgdata")} + const password = `s p#ss"word` + if err := writeOwnerOnly(n.adminPasswordPath(), password+"\n"); err != nil { t.Fatal(err) } - passwordFile, err := n.writeInitPasswordFile() - if err != nil { - t.Fatal(err) - } - defer os.Remove(passwordFile) + passwordFile := n.adminPasswordPath() info, err := os.Stat(passwordFile) if err != nil { t.Fatal(err) @@ -41,32 +31,42 @@ func TestInitdbUsesPrivatePasswordFileAndSCRAM(t *testing.T) { t.Fatalf("initdb args do not require SCRAM: %q", args) } joined := strings.Join(args, " ") - if strings.Contains(joined, "trust") || strings.Contains(joined, n.password) { + if strings.Contains(joined, "trust") || strings.Contains(joined, password) { t.Fatalf("unsafe initdb args: %q", args) } + if !slices.Contains(args, serverAdminRole) { + t.Fatalf("initdb args do not name the run server's administrator: %q", args) + } + read, err := n.readAdminPassword() + if err != nil || read != password { + t.Fatalf("administrator password did not round trip: %q, %v", read, err) + } } func TestAdminDSNEncodesCredentialsAndTargetsOwnedSocket(t *testing.T) { n := &nixPostgres{ - user: "db user", password: `p@ss:/?#[]`, port: 5432, - socketDir: filepath.Join(t.TempDir(), "private socket"), + adminPassword: `p@ss:/?#[]`, + socketDir: filepath.Join(t.TempDir(), "private socket"), } - parsed, err := url.Parse(n.adminDSN()) + parsed, err := url.Parse(n.adminDSN("orders db")) if err != nil { t.Fatal(err) } - if parsed.User.Username() != n.user { - t.Fatalf("username = %q, want %q", parsed.User.Username(), n.user) + if parsed.User.Username() != serverAdminRole { + t.Fatalf("username = %q, want %q", parsed.User.Username(), serverAdminRole) + } + if parsed.Path != "/orders db" { + t.Fatalf("database = %q, want the service's", parsed.Path) } password, ok := parsed.User.Password() - if !ok || password != n.password { + if !ok || password != n.adminPassword { t.Fatalf("password did not round trip: %q", password) } if parsed.Host != "" || parsed.Query().Get("host") != n.socketDir || parsed.Query().Get("port") != "5432" || parsed.Query().Get("sslmode") != "disable" { t.Fatalf("unexpected DSN %q", parsed.String()) } - pqDSN, err := pq.ParseURL(n.adminDSN()) + pqDSN, err := pq.ParseURL(n.adminDSN("orders db")) if err != nil { t.Fatal(err) } diff --git a/nixpg_run_test.go b/nixpg_run_test.go new file mode 100644 index 0000000..fbf2013 --- /dev/null +++ b/nixpg_run_test.go @@ -0,0 +1,105 @@ +package main + +import ( + "encoding/json" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +// shortTempDir is a temporary directory whose path fits a Unix socket. +func shortTempDir(t *testing.T) string { + t.Helper() + dir, err := os.MkdirTemp("/tmp", "cfpg-test-") + require.NoError(t, err) + t.Cleanup(func() { _ = os.RemoveAll(dir) }) + return dir +} + +func TestServerOptionsBindNoTCPPort(t *testing.T) { + options, err := serverOptions("/tmp/cfpg-abc", "") + require.NoError(t, err) + require.Contains(t, options, "-c listen_addresses=''") + require.Contains(t, options, "-k /tmp/cfpg-abc") + require.Contains(t, options, "-c max_connections=500") + require.NotContains(t, options, "log_min_messages") + + quiet, err := serverOptions("/tmp/cfpg-abc", "WARNING") + require.NoError(t, err) + require.Contains(t, quiet, "-c log_min_messages=warning") +} + +// TestServerOptionsRefuseWhatTheShellWouldInterpret: pg_ctl hands the options +// to /bin/sh, so nothing that the shell would read as syntax may reach them. +func TestServerOptionsRefuseWhatTheShellWouldInterpret(t *testing.T) { + _, err := serverOptions("/tmp/cfpg abc", "") + require.Error(t, err) + _, err = serverOptions("/tmp/cfpg-abc", "warning; rm -rf /") + require.Error(t, err) +} + +func TestNativeOwnerRoleIsTheDatabasesOwn(t *testing.T) { + owner := nativeOwnerRole("robin_sessions") + readOnly, readWrite := runtimeRoleNames("robin_sessions") + require.LessOrEqual(t, len(owner), 63) + require.NotEqual(t, readOnly, owner) + require.NotEqual(t, readWrite, owner) + require.Equal(t, owner, nativeOwnerRole("robin_sessions")) + require.NotEqual(t, owner, nativeOwnerRole("documents")) + require.True(t, strings.HasSuffix(owner, "_owner")) +} + +func writeHolder(t *testing.T, root, name string, holder serverHolder) string { + t.Helper() + data, err := json.Marshal(holder) + require.NoError(t, err) + path := filepath.Join(root, "holders", name+".json") + require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o700)) + require.NoError(t, os.WriteFile(path, data, 0o600)) + return path +} + +// TestPruneHoldersKeepsOnlyLiveRuns is what decides when a run's server stops: +// an agent that is gone, or whose codefly run is gone, holds nothing. +func TestPruneHoldersKeepsOnlyLiveRuns(t *testing.T) { + root := t.TempDir() + self, err := markProcess(os.Getpid()) + require.NoError(t, err) + parent, err := markProcess(os.Getppid()) + require.NoError(t, err) + + exited := exec.Command("sleep", "60") + require.NoError(t, exited.Start()) + gone, err := markProcess(exited.Process.Pid) + require.NoError(t, err) + require.NoError(t, exited.Process.Kill()) + _ = exited.Wait() + + reused := self + reused.StartID++ // the same PID, a different process + + live := writeHolder(t, root, "live", serverHolder{Service: "mod/a", Agent: self, Run: &parent}) + deadAgent := writeHolder(t, root, "dead-agent", serverHolder{Service: "mod/b", Agent: gone}) + deadRun := writeHolder(t, root, "dead-run", serverHolder{Service: "mod/c", Agent: self, Run: &gone}) + reusedPID := writeHolder(t, root, "reused", serverHolder{Service: "mod/d", Agent: reused}) + garbage := filepath.Join(root, "holders", "garbage.json") + require.NoError(t, os.WriteFile(garbage, []byte("{"), 0o600)) + + count, err := pruneHolders(root) + require.NoError(t, err) + require.Equal(t, 1, count) + require.FileExists(t, live) + for _, path := range []string{deadAgent, deadRun, reusedPID, garbage} { + require.NoFileExists(t, path) + } +} + +func TestReleaseOfAnUnattachedHandleStopsNothing(t *testing.T) { + stopped, err := (&nixPostgres{}).release(t.Context()) + require.NoError(t, err) + require.False(t, stopped) +} diff --git a/nixpg_scope_test.go b/nixpg_scope_test.go index 8e59a62..dab5403 100644 --- a/nixpg_scope_test.go +++ b/nixpg_scope_test.go @@ -2,26 +2,30 @@ package main import "testing" -func TestNixPostgresStateKeyPreservesUnscopedIdentity(t *testing.T) { - const location = "/workspace/modules/infra/services/postgres" - if got := nixPostgresStateKey(location, ""); got != location { - t.Fatalf("unscoped state key = %q, want existing location %q", got, location) +func TestRunServerKeyIsStablePerRun(t *testing.T) { + const workspace = "/workspace/robin/playground" + if runServerKey(workspace, "qa") != runServerKey(workspace, "qa") { + t.Fatal("one run produced two server keys") } } -func TestNixPostgresStateKeySeparatesNamingScopes(t *testing.T) { - const location = "/workspace/modules/infra/services/postgres" - first := nixPostgresStateKey(location, "qualification-a") - repeated := nixPostgresStateKey(location, "qualification-a") - second := nixPostgresStateKey(location, "qualification-b") - - if first != repeated { - t.Fatalf("same naming scope produced unstable keys %q and %q", first, repeated) - } - if first == second { - t.Fatalf("different naming scopes shared state key %q", first) +// TestRunServerKeySeparatesRuns pins the isolation boundary: a naming scope or +// another workspace is another run, with its own server, data and credentials. +func TestRunServerKeySeparatesRuns(t *testing.T) { + const workspace = "/workspace/robin/playground" + keys := map[string]string{ + "unscoped": runServerKey(workspace, ""), + "scope a": runServerKey(workspace, "qualification-a"), + "scope b": runServerKey(workspace, "qualification-b"), + "other workspace": runServerKey("/workspace/other", "qualification-a"), + "ambiguous concat": runServerKey(workspace+"\x00naming-scope\x00qualification", "-a"), } - if serviceHash(first) == serviceHash(second) { - t.Fatal("different naming scopes shared the same runtime-root hash") + seen := map[string]string{} + for name, key := range keys { + hash := serviceHash(key)[:16] + if other, ok := seen[hash]; ok { + t.Fatalf("%s and %s share a run server", name, other) + } + seen[hash] = name } } diff --git a/nixpg_shared_test.go b/nixpg_shared_test.go new file mode 100644 index 0000000..7ad31c7 --- /dev/null +++ b/nixpg_shared_test.go @@ -0,0 +1,170 @@ +package main + +import ( + "context" + "database/sql" + "net/url" + "os" + "testing" + + runtimev0 "github.com/codefly-dev/core/generated/go/codefly/services/runtime/v0" + "github.com/codefly-dev/core/resources" + runners "github.com/codefly-dev/core/runners/base" + "github.com/codefly-dev/core/shared" + "github.com/stretchr/testify/require" +) + +func requireNix(t *testing.T) { + t.Helper() + if !runners.CheckNixInstalled() || !runners.IsNixSupported() { + t.Skip("nix not installed/supported on this host") + } +} + +// withDatabase is connection pointed at another database of the same server. +func withDatabase(t *testing.T, connection, database string) string { + t.Helper() + parsed, err := url.Parse(connection) + require.NoError(t, err) + parsed.Path = "/" + database + return parsed.String() +} + +func withUser(t *testing.T, connection, user, password string) string { + t.Helper() + parsed, err := url.Parse(connection) + require.NoError(t, err) + parsed.User = url.UserPassword(user, password) + return parsed.String() +} + +func connects(ctx context.Context, connection string) error { + db, err := sql.Open("postgres", connection) + if err != nil { + return err + } + defer db.Close() + return db.PingContext(ctx) +} + +// TestRunServerIsSharedAndIsolated is the run server's contract end to end: +// two services of one run get one postmaster, each its own database at its own +// endpoint; neither service's logins reach the other's database; releasing one +// leaves the other serving; the last release stops the server; and the data +// is there for the next run. +func TestRunServerIsSharedAndIsolated(t *testing.T) { + requireNix(t) + ctx := context.Background() + workspace := t.TempDir() + alpha := newPostgresFixtureAt(t, ctx, workspace, "alpha", "store", resources.NewRuntimeContextNix()) + beta := newPostgresFixtureAt(t, ctx, workspace, "beta", "store", resources.NewRuntimeContextNix()) + + runAlpha, _ := alpha.start(t, ctx) + runBeta, _ := beta.start(t, ctx) + + // One server for the run. + require.Equal(t, runAlpha.nixRuntime.dataDir, runBeta.nixRuntime.dataDir) + postmaster, _, running := runningPostmaster(runAlpha.nixRuntime.dataDir) + require.True(t, running) + + // Each service at its own endpoint, with its own data. + requireListener(t, alpha.address) + requireListener(t, beta.address) + writeSentinel(t, ctx, runAlpha.connection, alpha.serviceName, "00000000-0000-0000-0000-0000000000a1") + writeSentinel(t, ctx, runBeta.connection, beta.serviceName, "00000000-0000-0000-0000-0000000000b1") + + // The owner is the database's own, and not a superuser. + require.Equal(t, nativeOwnerRole("alpha"), runAlpha.postgresUser) + db, err := sql.Open("postgres", runAlpha.connection) + require.NoError(t, err) + var superuser bool + require.NoError(t, db.QueryRowContext(ctx, `SELECT rolsuper FROM pg_roles WHERE rolname = current_user`).Scan(&superuser)) + require.NoError(t, db.Close()) + require.False(t, superuser, "a service's migration owner must not be the run server's superuser") + + // Isolation: no login of one service reaches the other's database, or the + // server's maintenance database. + readOnly, _ := runtimeRoleNames(runAlpha.DatabaseName) + alphaReader := withUser(t, runAlpha.connection, readOnly, runAlpha.readOnlyPassword) + require.NoError(t, connects(ctx, alphaReader), "the reader reaches its own database") + for name, connection := range map[string]string{ + "owner into the other database": withDatabase(t, runAlpha.connection, runBeta.DatabaseName), + "reader into the other database": withDatabase(t, alphaReader, runBeta.DatabaseName), + "owner into postgres": withDatabase(t, runAlpha.connection, "postgres"), + "owner into template1": withDatabase(t, runAlpha.connection, "template1"), + } { + require.ErrorContains(t, connects(ctx, connection), "permission denied", name) + } + + // Releasing one service leaves the run's server serving the other. + stopped, err := runAlpha.Stop(ctx, &runtimev0.StopRequest{}) + require.NoError(t, err) + require.Contains(t, stopped.GetStatus().GetMessage(), "keeps serving its other services") + requireNoListener(t, alpha.address) + require.True(t, hasSentinel(t, ctx, runBeta.connection, beta.serviceName, "00000000-0000-0000-0000-0000000000b1")) + + // The last release stops it. + stopped, err = runBeta.Stop(ctx, &runtimev0.StopRequest{}) + require.NoError(t, err) + require.Contains(t, stopped.GetStatus().GetMessage(), "stopped native postgres") + requireNoListener(t, beta.address) + require.False(t, processAlive(postmaster), "the run's postmaster must be gone once no service holds it") + + // The next run finds its data. + revived, _ := alpha.start(t, ctx) + require.True(t, hasSentinel(t, ctx, revived.connection, alpha.serviceName, "00000000-0000-0000-0000-0000000000a1")) +} + +// TestRunServerRefusesTwoServicesClaimingOneDatabase: a server's databases are +// its services', one each, and a second claim is refused naming both. +func TestRunServerRefusesTwoServicesClaimingOneDatabase(t *testing.T) { + requireNix(t) + ctx := context.Background() + workspace := t.TempDir() + first := newPostgresFixtureAt(t, ctx, workspace, "shared", "first", resources.NewRuntimeContextNix()) + second := newPostgresFixtureAt(t, ctx, workspace, "shared", "second", resources.NewRuntimeContextNix()) + + run, _ := first.start(t, ctx) + + refused := NewRuntime() + _, err := refused.Load(ctx, &runtimev0.LoadRequest{Identity: second.identity, Environment: shared.Must(second.environment.Proto()), DisableCatch: true}) + require.NoError(t, err) + init, err := refused.Init(ctx, &runtimev0.InitRequest{ + RuntimeContext: second.runtimeContext, + Configuration: second.configuration, + ProposedNetworkMappings: second.networkMappings, + }) + require.NoError(t, err) + require.Equal(t, runtimev0.InitStatus_ERROR, init.GetStatus().GetState()) + require.Contains(t, init.GetStatus().GetMessage(), + `database "shared" on this run's postgres server belongs to shared/first, not shared/second`) + require.Nil(t, refused.nixRuntime, "a refused service holds no server") + requireNoListener(t, second.address) + + // The refusal took nothing from the service that holds the database. + require.NoError(t, connects(ctx, run.connection)) +} + +// TestAbandonedRunServerIsCollected: a run killed without releasing leaves a +// server nobody holds; the next agent to initialize anywhere stops it. +func TestAbandonedRunServerIsCollected(t *testing.T) { + requireNix(t) + ctx := context.Background() + abandoned := newPostgresFixtureAt(t, ctx, t.TempDir(), "gone", "store", resources.NewRuntimeContextNix()) + run, _ := abandoned.start(t, ctx) + dataDir := run.nixRuntime.dataDir + postmaster, _, running := runningPostmaster(dataDir) + require.True(t, running) + + // The run dies: its agent's record names a process that no longer exists, + // and its endpoint went with it. + require.NoError(t, run.nixRuntime.forwarder.Close()) + run.nixRuntime.serverCancel() + writeHolder(t, run.nixRuntime.root, "store", serverHolder{Service: "gone/store", Agent: processMark{PID: 2147483646}}) + require.NoError(t, os.Remove(run.nixRuntime.holder)) + run.nixRuntime = nil + + next := newPostgresFixtureAt(t, ctx, t.TempDir(), "next", "store", resources.NewRuntimeContextNix()) + next.start(t, ctx) + require.False(t, processAlive(postmaster), "an abandoned run's server must be stopped by the next run") +} diff --git a/recovery_scope_test.go b/recovery_scope_test.go index 41255c9..d8e2ef0 100644 --- a/recovery_scope_test.go +++ b/recovery_scope_test.go @@ -3,6 +3,7 @@ package main import ( "fmt" "os" + "path/filepath" "testing" "github.com/codefly-dev/core/runners/dockerrun" @@ -27,7 +28,21 @@ func TestMain(m *testing.M) { fmt.Fprintln(os.Stderr, err) return 1 } + // Native run servers started by this suite live under the suite's own + // directory, never beside a developer's real runs, and none outlives it. + servers := filepath.Join(dir, "run-servers") + runServersDir = func() (string, error) { return servers, nil } + defer stopEveryRunServer(servers) return m.Run() }() os.Exit(code) } + +func stopEveryRunServer(servers string) { + entries, _ := os.ReadDir(servers) + for _, entry := range entries { + if stopped, err := stopServer(filepath.Join(servers, entry.Name(), "pgdata")); err != nil || stopped { + fmt.Fprintf(os.Stderr, "run server %s outlived its test (stopped=%v err=%v)\n", entry.Name(), stopped, err) + } + } +} diff --git a/runtime.go b/runtime.go index b8efc1a..589a723 100644 --- a/runtime.go +++ b/runtime.go @@ -50,10 +50,18 @@ type Runtime struct { runnerEnvironment *dockerrun.DockerEnvironment // nixRuntime is set instead of runnerEnvironment when the caller requests - // RuntimeContextNix — postgres then runs natively from a nix-provisioned - // binary (no Docker), serving the same connection string + database. + // RuntimeContextNix: this service's attachment to its run's one native + // postgres server (nixpg.go), serving the same connection string and + // database. nixRuntime *nixPostgres + // extensionAdmin is the connection extensions are created with. On the + // run's shared server that is the server's administrator in this service's + // database — the migration owner is not a superuser there, and pgvector is + // not a trusted extension — the way a managed server's provider creates + // them. Empty elsewhere: the owner connection creates them. + extensionAdmin string + postgresPort uint16 // lifecycleMu serializes Stop and Destroy. Both act on runnerEnvironment, @@ -170,6 +178,16 @@ func (s *Runtime) Init(ctx context.Context, req *runtimev0.InitRequest) (*runtim s.NetworkMappings = req.ProposedNetworkMappings + // On the native runtime the service's database lives on its run's shared + // server, where roles are server-wide: the migration owner is this + // database's own role there, not the configured POSTGRES_USER every local + // service ships with. + native := req.GetRuntimeContext().GetKind() == resources.RuntimeContextNix + s.nativeOwner = "" + if native { + s.nativeOwner = nativeOwnerRole(s.DatabaseName) + } + configuration := req.GetConfiguration() net, err := resources.FindNetworkMapping(ctx, s.NetworkMappings, s.TcpEndpoint) @@ -235,31 +253,19 @@ func (s *Runtime) Init(ctx context.Context, req *runtimev0.InitRequest) (*runtim return s.Runtime.InitError(err) } - // Nix runtime: run postgres natively from a nix-provisioned binary instead - // of a Docker container — selected when the caller requests - // RuntimeContextNix (e.g. a host without Docker). Same connection string + - // database as the Docker path, so the rest of the agent is unchanged. - if rc := req.GetRuntimeContext(); rc != nil && rc.Kind == resources.RuntimeContextNix { - w.Debug("using nix runtime for postgres", wool.Field("port", instance.Port)) + // Native runtime: this service's database on its run's one postgres + // server (nixpg.go), published at this service's own endpoint + // (forward.go). Same connection strings and database as the Docker path, + // so the rest of the agent is unchanged. + if native { + w.Debug("using the run's native postgres server", wool.Field("port", instance.Port)) // An interrupted native run leaves System V resources the postmaster // never got to remove, and a leaked semaphore run poisons every later - // start on this host. Collect them before the cluster is initialized, - // while nothing this invocation owns is attached to them yet. + // start on this host. Collect them before a server is started, while + // nothing this invocation owns is attached to them yet. s.reapHostResources(ctx) - nixpg, errNix := newNixPostgres(ctx, nixPostgresStateKey(s.Location, s.Environment.NamingScope), uint16(instance.Port), - s.postgresUser, s.postgresPassword, s.DatabaseName, s.LogLevel, newPGLogWriter(s.Wool)) - if errNix != nil { - return s.Runtime.InitError(errNix) - } - if errNix = nixpg.Init(ctx); errNix != nil { - return s.Runtime.InitError(errNix) - } - s.nixRuntime = nixpg - s.retainedDataPath = nixpg.dataDir - s.released = false - s.Wool.Debug("nix postgres init successful") - if errNix = s.migrateOnInit(ctx); errNix != nil { - return s.Runtime.InitError(errNix) + if errNative := s.initNative(ctx, instance); errNative != nil { + return s.Runtime.InitError(errNative) } return s.Runtime.InitResponse() } @@ -385,7 +391,7 @@ func (s *Runtime) applySchema(ctx context.Context, prerequisites *schemaPrerequi // use. The convenience defaults and declarations marked optional record a // structured skip instead. func (s *Runtime) ensureExtensions(ctx context.Context, prerequisites *schemaPrerequisites) error { - db, err := sql.Open("postgres", s.connection) + db, err := sql.Open("postgres", s.extensionConnection()) if err != nil { return s.Wool.Wrapf(err, "cannot open database to create extensions") } @@ -679,32 +685,26 @@ func (s *Runtime) Stop(ctx context.Context, req *runtimev0.StopRequest) (*runtim s.StopWatcher() // keep-running asks for a server the NEXT invocation reattaches to, and only - // the container backend can reattach: GetContainer adopts an existing - // container by name. nixPostgres has no such path — Init always launches a - // new postmaster, and clearStalePostmasterPid deliberately leaves a live - // owner's lock file alone, so a retained nix postmaster makes the next Init - // start a second postmaster against the same data directory and port, which - // postgres refuses. Honouring the setting there would trade a cold start for - // a broken one, so nix stops and keeps the half of the contract it can: the - // data. + // the container backend has one per service: GetContainer adopts an + // existing container by name. The native server belongs to the run, not to + // this service, and stops when the run's last service releases it, so one + // service cannot keep it up for itself. Native releases and keeps the half + // of the contract it can: the data. if s.Settings.KeepRunning { if s.nixRuntime == nil && !s.Runtime.IsNixRuntime() { s.Wool.Debug("keep-running is set: leaving postgres up for reuse") return s.stopResponse("kept postgres running for reuse") } - s.Wool.Warn("keep-running is not available on the nix runtime: stopping postgres, its data is retained") + s.Wool.Warn("keep-running is not available on the nix runtime: releasing postgres, its data is retained") } if s.nixRuntime != nil { - if err := s.nixRuntime.Stop(ctx); err != nil { + action, err := s.releaseNative(ctx) + if err != nil { return s.Runtime.StopError(err) } - // Stop removes the handle's socket directory, so it can never serve a - // second run — drop it instead of leaving a dead handle behind. - s.nixRuntime = nil - s.released = true s.reapHostResources(ctx) - return s.stopResponse("stopped native postgres") + return s.stopResponse(action) } if s.runnerEnvironment != nil { @@ -735,13 +735,12 @@ func (s *Runtime) Destroy(ctx context.Context, req *runtimev0.DestroyRequest) (* s.StopWatcher() if s.nixRuntime != nil { - if err := s.nixRuntime.Stop(ctx); err != nil { + action, err := s.releaseNative(ctx) + if err != nil { return s.Runtime.DestroyError(err) } - s.nixRuntime = nil - s.released = true s.reapHostResources(ctx) - return s.destroyResponse("stopped native postgres") + return s.destroyResponse(action) } // A nix invocation has no container to remove, and reaching for one would // demand a docker daemon from a host that was very likely chosen for not diff --git a/runtime_access.go b/runtime_access.go index db5f518..5ef2d05 100644 --- a/runtime_access.go +++ b/runtime_access.go @@ -233,6 +233,12 @@ func (s *Service) runtimeAccess() (readOnlyRole, readWriteRole string, err error } func runtimeRoleNames(database string) (readOnly, readWrite string) { + prefix := runtimeRolePrefix(database) + return prefix + "_ro", prefix + "_rw" +} + +// runtimeRolePrefix is the server-wide name stem of a database's roles. +func runtimeRolePrefix(database string) string { var slug strings.Builder for _, r := range strings.ToLower(database) { if unicode.IsLetter(r) || unicode.IsDigit(r) { @@ -254,8 +260,7 @@ func runtimeRoleNames(database string) (readOnly, readWrite string) { } sum := sha256.Sum256([]byte(database)) digest := hex.EncodeToString(sum[:4]) - prefix := "codefly_" + base + "_" + digest - return prefix + "_ro", prefix + "_rw" + return "codefly_" + base + "_" + digest } func normalizedRuntimeSchemas(configured []string) ([]string, error) { @@ -480,8 +485,26 @@ func ensureLoginRole(ctx context.Context, tx *sql.Tx, role, password string, rea return err } } - attributes := ` WITH LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT NOREPLICATION NOBYPASSRLS PASSWORD ` + pq.QuoteLiteral(password) - if _, err := tx.ExecContext(ctx, `ALTER ROLE `+quotedRole+attributes); err != nil { + // Only a superuser may even restate SUPERUSER, REPLICATION or BYPASSRLS, so + // a migration owner that is not one — the run's shared native server, a + // managed server — clears only the elevated attributes the role actually + // holds. A role holding one it cannot clear still fails, as it must. + var superuser, createDB, createRole, replication, bypassRLS bool + if err := tx.QueryRowContext(ctx, ` + SELECT rolsuper, rolcreatedb, rolcreaterole, rolreplication, rolbypassrls + FROM pg_roles WHERE rolname = $1`, role).Scan(&superuser, &createDB, &createRole, &replication, &bypassRLS); err != nil { + return err + } + attributes := []string{"LOGIN", "NOINHERIT"} + for _, attribute := range []struct { + set bool + clear string + }{{superuser, "NOSUPERUSER"}, {createDB, "NOCREATEDB"}, {createRole, "NOCREATEROLE"}, {replication, "NOREPLICATION"}, {bypassRLS, "NOBYPASSRLS"}} { + if attribute.set { + attributes = append(attributes, attribute.clear) + } + } + if _, err := tx.ExecContext(ctx, `ALTER ROLE `+quotedRole+` WITH `+strings.Join(attributes, " ")+` PASSWORD `+pq.QuoteLiteral(password)); err != nil { return err } if readOnly { diff --git a/runtime_native.go b/runtime_native.go new file mode 100644 index 0000000..d8ecf21 --- /dev/null +++ b/runtime_native.go @@ -0,0 +1,78 @@ +package main + +import ( + "context" + "errors" + "fmt" + "net" + "strconv" + + basev0 "github.com/codefly-dev/core/generated/go/codefly/base/v0" +) + +// nativeOwnerRole is a database's migration owner on its run's shared server. +// The run's services all ship the same local POSTGRES_USER, and a server's +// roles are server-wide, so the owner is named after the database the way its +// runtime roles already are. Consumers receive it inside the owner connection +// string, as they received POSTGRES_USER. +func nativeOwnerRole(database string) string { + return runtimeRolePrefix(database) + "_owner" +} + +// initNative attaches this service to its run's server, gives it its database +// there, publishes its endpoint, and migrates — the native half of Init. A +// failure releases whatever it took, so a refused service holds no server. +func (s *Runtime) initNative(ctx context.Context, instance *basev0.NetworkInstance) (initErr error) { + server, err := acquireRunServer(ctx, + runServerKey(s.Identity.WorkspacePath, s.Environment.NamingScope), + s.Unique(), s.LogLevel, newPGLogWriter(s.Wool)) + if err != nil { + return err + } + s.nixRuntime = server + defer func() { + if initErr != nil { + _, releaseErr := s.releaseNative(context.WithoutCancel(ctx)) + initErr = errors.Join(initErr, releaseErr) + } + }() + if err := server.provisionDatabase(ctx, s.Unique(), s.DatabaseName, s.postgresUser, s.postgresPassword); err != nil { + return err + } + address := net.JoinHostPort("127.0.0.1", strconv.Itoa(int(instance.Port))) + if server.forwarder, err = forwardToSocket(address, server.socketPath()); err != nil { + return err + } + if s.extensionAdmin, err = withConnectTimeout(server.adminDSN(s.DatabaseName), s.Settings.Timeouts.connect()); err != nil { + return err + } + s.retainedDataPath = server.dataDir + s.released = false + s.Wool.Debug("database ready on the run's native postgres server") + return s.migrateOnInit(ctx) +} + +// releaseNative closes this service's endpoint and releases its hold on the +// run's server, which stops when this was its last service. It returns what +// happened, for the Stop or Destroy result. +func (s *Runtime) releaseNative(ctx context.Context) (string, error) { + stopped, err := s.nixRuntime.release(ctx) + if err != nil { + return "", fmt.Errorf("release the run's native postgres server: %w", err) + } + s.nixRuntime = nil + s.extensionAdmin = "" + s.released = true + if stopped { + return "stopped native postgres", nil + } + return "released this service's database; the run's native postgres server keeps serving its other services", nil +} + +// extensionConnection is the connection extensions are created with. +func (s *Runtime) extensionConnection() string { + if s.extensionAdmin != "" { + return s.extensionAdmin + } + return s.connection +} From 6e619bf1afad647e2e97552531e39f8858ff767f Mon Sep 17 00:00:00 2001 From: Antoine Toussaint <312536060+antoinetoussaint-byte@users.noreply.github.com> Date: Tue, 27 Oct 2026 09:32:56 -0400 Subject: [PATCH 3/5] docs: one native Postgres server per run Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 3 ++- templates/agent/README.md.tmpl | 18 +++++++++++++++--- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 2fd815c..6a5a6b2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -122,7 +122,8 @@ variables only the qualify script supplies. | `migrations.go` | one golang-migrate lineage per source, budgets, fail-closed | | `schema.go`, `schemaplan.go` | prerequisite resolution into `bootstrap/plan.json` | | `runtime_access.go` | roles, grants, RLS, password vs `external-identity` | -| `nixpg.go` | the docker-free backend, from the embedded `nix/flake.*` | +| `nixpg.go` | the docker-free backend, from the embedded `nix/flake.*`: one server per run | +| `forward.go`, `runtime_native.go` | a service's endpoint and database on its run's native server | | `hotreload.go` | forward-only migration reload | | `templates/` | everything embedded into the binary; `factory/` is what `Create` scaffolds | | `libs/go` | **separate module**: the Go client library, no core, no parent | diff --git a/templates/agent/README.md.tmpl b/templates/agent/README.md.tmpl index efd8d73..89bbde0 100644 --- a/templates/agent/README.md.tmpl +++ b/templates/agent/README.md.tmpl @@ -22,7 +22,7 @@ The bootstrap job applies migrations and reconciles runtime roles and grants on ## Externally provisioned deployments -Add an entry to `external-instances` when a deployment environment uses a managed Postgres server instead of a server owned by this service. Bindings are keyed by environment so staging and production cannot silently share an address. The deployment emits the bootstrap Job but no Service, StatefulSet, or persistent volume. Local Docker and Nix runs still provision their own Postgres server. +Add an entry to `external-instances` when a deployment environment uses a managed Postgres server instead of a server owned by this service. Bindings are keyed by environment so staging and production cannot silently share an address. The deployment emits the bootstrap Job but no Service, StatefulSet, or persistent volume. Local runs are described under "One native server per run" below. ```yaml database-name: accounts @@ -41,13 +41,25 @@ External instances currently require password authentication. `auth-mode: extern Local database state survives ordinary stop/start and container recreation. In Docker mode, the agent mounts the official image's data directory from a Codefly-owned persistent runtime directory scoped by workspace, service, and naming scope. Stable and development scopes therefore keep independent databases even when they run from the same checkout. +## One native server per run + +On the Nix runtime every Postgres service of a run shares **one** Postgres server, the way a deployed environment binds its databases to one instance. The run is the workspace and its naming scope; another scope or another workspace is another run, with its own server, data and credentials. Nothing changes for a service: it keeps its endpoint, its connection strings, its migrations and its runtime logins. + +- **One database and one owner per service.** The migration owner on the shared server is the database's own role (`codefly___owner`), not the configured `POSTGRES_USER` — every local service ships the same one, and roles are server-wide. Its password is still `POSTGRES_PASSWORD`, and the owner connection string carries the role. The owner may create roles and databases but is not a superuser. +- **Isolation.** Each database is closed to `PUBLIC`; only its own owner and runtime logins connect to it. No login of one service reaches another's database or the server's maintenance databases. Two services of one run declaring the same `database-name` are refused, naming both. +- **Administration.** The server's bootstrap superuser exists only for the agent: it creates each database, its owner, and the declared extensions (pgvector is not a trusted extension). Its password is generated when the server is first initialized and is kept in an owner-only file beside the cluster; no service receives it. +- **Endpoints.** The server listens on no TCP port, only on a private Unix socket. Each service's agent listens on the service's own allocated port, on loopback, and forwards every connection to that socket; authentication stays between the client and the server. +- **Lifecycle.** The first service to start starts the server; each stop or destroy releases that service, and the last one stops the server, which frees its shared memory. A run killed without stopping leaves a server no live agent holds; the next service to start on the machine stops it. The cluster directory is retained per run, as each service's was. + +Existing per-service native clusters are not migrated: a run's first start on this version creates its databases empty and migrates them, and the old directories are left where they are. + ## Lifecycle: stop, keep-running, destroy Stop and destroy have one state-ownership contract, and it is the same on Docker and Nix: | operation | execution resources | database state | | --- | --- | --- | -| stop | released — the postmaster is terminated (Nix) or the container is stopped (Docker), freeing the published port | retained | +| stop | released — the service's endpoint closes and, for the run's last service, the postmaster is terminated (Nix), or the container is stopped (Docker), freeing the published port | retained | | stop with `keep-running: true` (Docker only) | retained — the container stays up for the next invocation to reuse | retained | | destroy | released; the Docker container is also removed | retained | @@ -64,7 +76,7 @@ spec: keep-running: true ``` -`keep-running` applies to the Docker runtime only. Reuse means the next invocation reattaches to a live server, and only Docker can do that — the Nix runtime always launches a new postmaster, so a retained one would make the next start fail on a data directory another process still holds. On Nix the setting is reported and ignored: postgres stops and its data is retained. +`keep-running` applies to the Docker runtime only. Reuse means the next invocation reattaches to a live server of this service's own, and only Docker has one — the Nix server belongs to the run and stops when the run's last service releases it. On Nix the setting is reported and ignored: the service is released and its data is retained. Test fixtures should own disposable state instead of depending on a warm server: give the fixture its own naming scope so its database is separate, and let stop retain it or discard the scope wholesale. Reading a sentinel row back after stop and re-init is the supported way to prove retention. ## Schema prerequisites From bb070331f773d8f2d43fcca590366e38e5771e85 Mon Sep 17 00:00:00 2001 From: Antoine Toussaint <312536060+antoinetoussaint-byte@users.noreply.github.com> Date: Tue, 27 Oct 2026 09:39:17 -0400 Subject: [PATCH 4/5] fix(runtime): stop a run server this agent started but could not hand out Co-Authored-By: Claude Opus 5.5 --- nixpg.go | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/nixpg.go b/nixpg.go index a12a47e..47c2ac1 100644 --- a/nixpg.go +++ b/nixpg.go @@ -203,11 +203,19 @@ func acquireRunServer(ctx context.Context, key, service, logLevel string, out io defer unlock() n := &nixPostgres{root: root, dataDir: filepath.Join(root, "pgdata")} - if _, err := pruneHolders(root); err != nil { + live, err := pruneHolders(root) + if err != nil { return nil, err } pid, socketDir, running := runningPostmaster(n.dataDir) if !running { + // A server this call starts and then cannot hand out is nobody's: stop + // it rather than leave it for the next run's collection. + defer func() { + if acquireErr != nil && live == 0 { + _, _ = stopServer(n.dataDir) + } + }() if err := n.start(ctx, logLevel, out); err != nil { return nil, err } From ff1b2d23150cab6a98f0082a84c80d1b333f575f Mon Sep 17 00:00:00 2001 From: Antoine Toussaint <312536060+antoinetoussaint-byte@users.noreply.github.com> Date: Tue, 27 Oct 2026 11:31:50 -0400 Subject: [PATCH 5/5] fix(runtime): signal only the postmaster this run recorded starting A postmaster.pid outlives the process that wrote it, PIDs are reused, and collection sweeps every other run root on the machine at every native start, so trusting the number in that file made an unrelated process of this user the target of a SIGINT. The postmaster's birth is now recorded when the server starts and re-confirmed before any signal; a lock file naming a live process that is provably not a postgres server is cleared instead of waited on, and one a real postmaster holds is refused by name rather than removed. Also refuse an owner role another service created: PostgreSQL 16 grants a created role back to its creator WITH ADMIN OPTION, so adopting one would hand that creator the database. Co-Authored-By: Claude Opus 5.5 --- nixpg.go | 365 +++++++++++++++++++++++++++++++---------- nixpg_pidclean_test.go | 37 ++--- nixpg_shared_test.go | 89 +++++++++- nixpg_signal_test.go | 120 ++++++++++++++ recovery_scope_test.go | 3 +- 5 files changed, 500 insertions(+), 114 deletions(-) create mode 100644 nixpg_signal_test.go diff --git a/nixpg.go b/nixpg.go index 47c2ac1..eea154d 100644 --- a/nixpg.go +++ b/nixpg.go @@ -38,7 +38,6 @@ package main // this agent deletes data, codefly-dev/core#424). import ( - "bufio" "context" "crypto/rand" "crypto/sha256" @@ -167,6 +166,16 @@ func (m processMark) alive() bool { return identity.StartID == m.StartID && identity.BootID == m.BootID } +// confirmed is alive's opposite default, and it is the only question a signal +// may be sent on. A PID says nothing on its own: the number is reused, so a +// process that merely answers to it can be anything this user happens to be +// running. An inspection that does not positively re-establish the recorded +// birth therefore means "do not signal", where alive means "keep holding". +func (m processMark) confirmed() bool { + identity, err := runners.InspectProcess(m.PID) + return err == nil && identity.StartID == m.StartID && identity.BootID == m.BootID +} + // serverHolder records one agent using a run server: the agent and the // codefly run that launched it. An agent whose run is gone is orphaned, and // its run's server is no longer anybody's. @@ -207,39 +216,34 @@ func acquireRunServer(ctx context.Context, key, service, logLevel string, out io if err != nil { return nil, err } - pid, socketDir, running := runningPostmaster(n.dataDir) + record, running, err := attachedPostmaster(n.root, n.dataDir) + if err != nil { + return nil, err + } if !running { // A server this call starts and then cannot hand out is nobody's: stop // it rather than leave it for the next run's collection. defer func() { if acquireErr != nil && live == 0 { - _, _ = stopServer(n.dataDir) + _, _ = stopServer(n.root, n.dataDir) } }() - if err := n.start(ctx, logLevel, out); err != nil { + if record, err = n.start(ctx, logLevel, out); err != nil { return nil, err } - if pid, socketDir, running = runningPostmaster(n.dataDir); !running { - return nil, fmt.Errorf("postgres reported started but %s names no live postmaster; see %s", - filepath.Join(n.dataDir, "postmaster.pid"), n.logPath()) - } } - n.socketDir = socketDir + n.socketDir = record.SocketDir if n.adminPassword, err = n.readAdminPassword(); err != nil { return nil, err } - postmaster, err := markProcess(pid) - if err != nil { - return nil, fmt.Errorf("inspect the run's postmaster (pid %d): %w", pid, err) - } if err := n.ping(ctx); err != nil { return nil, fmt.Errorf("the run's postgres server (pid %d) does not answer on %s: %w; see %s", - pid, n.socketDir, err, n.logPath()) + record.Postmaster.PID, n.socketDir, err, n.logPath()) } if err := n.register(service); err != nil { return nil, err } - n.watch(postmaster) + n.watch(record.Postmaster) return n, nil } @@ -259,52 +263,49 @@ func lockRunServer(ctx context.Context, root string) (func(), error) { func (n *nixPostgres) logPath() string { return filepath.Join(n.root, "server.log") } -// start initializes the cluster on first use and launches it detached. The -// caller holds the run server's lock. -func (n *nixPostgres) start(ctx context.Context, logLevel string, out io.Writer) error { +// start initializes the cluster on first use, launches it detached, and +// records the postmaster it started. The caller holds the run server's lock. +func (n *nixPostgres) start(ctx context.Context, logLevel string, out io.Writer) (postmasterRecord, error) { flakeDir := filepath.Join(n.root, "nix") if err := os.MkdirAll(flakeDir, 0o700); err != nil { - return fmt.Errorf("create nix flake dir: %w", err) + return postmasterRecord{}, fmt.Errorf("create nix flake dir: %w", err) } if err := os.WriteFile(filepath.Join(flakeDir, "flake.nix"), []byte(pgFlakeNix), 0o644); err != nil { - return fmt.Errorf("write flake.nix: %w", err) + return postmasterRecord{}, fmt.Errorf("write flake.nix: %w", err) } if err := os.WriteFile(filepath.Join(flakeDir, "flake.lock"), []byte(pgFlakeLock), 0o644); err != nil { - return fmt.Errorf("write flake.lock: %w", err) + return postmasterRecord{}, fmt.Errorf("write flake.lock: %w", err) } env, err := runners.NewNixEnvironment(ctx, flakeDir) if err != nil { - return fmt.Errorf("nix environment (is nix installed?): %w", err) + return postmasterRecord{}, fmt.Errorf("nix environment (is nix installed?): %w", err) } env.WithCacheDir(filepath.Join(n.root, ".nix-cache")) if err := env.Init(ctx); err != nil { - return fmt.Errorf("materialize nix postgres env: %w", err) + return postmasterRecord{}, fmt.Errorf("materialize nix postgres env: %w", err) } binDir, err := resolveBinDir() if err != nil { - return err + return postmasterRecord{}, err } if err := n.initdbIfNeeded(ctx, env, binDir, out); err != nil { - return err - } - if err := n.clearStalePostmasterPid(); err != nil { - return err + return postmasterRecord{}, err } // The socket path is limited to ~104 bytes, so it cannot live under a deep // cache path. An unpredictable, owner-only temporary directory: a stable // name can be pre-created or symlinked by another local user. socketDir, err := os.MkdirTemp("", "cfpg-"+filepath.Base(n.root)[:8]+"-") if err != nil { - return fmt.Errorf("create postgres socket dir: %w", err) + return postmasterRecord{}, fmt.Errorf("create postgres socket dir: %w", err) } if err := os.Chmod(socketDir, 0o700); err != nil { _ = os.RemoveAll(socketDir) - return fmt.Errorf("secure postgres socket dir: %w", err) + return postmasterRecord{}, fmt.Errorf("secure postgres socket dir: %w", err) } options, err := serverOptions(socketDir, logLevel) if err != nil { _ = os.RemoveAll(socketDir) - return err + return postmasterRecord{}, err } // pg_ctl detaches the postmaster into its own session and waits until it // accepts connections; the server outlives the agent that started it. @@ -312,16 +313,38 @@ func (n *nixPostgres) start(ctx context.Context, logLevel string, out io.Writer) "start", "-D", n.dataDir, "-w", "-t", strconv.Itoa(serverStartSeconds), "-l", n.logPath(), "-o", options) if err != nil { _ = os.RemoveAll(socketDir) - return err + return postmasterRecord{}, err } if out != nil { proc.WithOutput(out) } if err := proc.Run(ctx); err != nil { _ = os.RemoveAll(socketDir) - return fmt.Errorf("start the run's postgres server: %w%s", err, n.logTail()) + return postmasterRecord{}, fmt.Errorf("start the run's postgres server: %w%s", err, n.logTail()) + } + // The one moment postmaster.pid is authoritative: pg_ctl -w has just + // reported the server it started is accepting connections, this call holds + // the run lock, and any pid file that was not this postmaster's was + // already cleared. Its birth is recorded here so that every later decision + // — attach, release, collection — rests on the identity rather than on the + // number in a file that outlives the process. + pid, ok := postmasterPidFileOwner(filepath.Join(n.dataDir, "postmaster.pid")) + if !ok { + return postmasterRecord{}, fmt.Errorf("postgres reported started but %s names no live postmaster; see %s", + filepath.Join(n.dataDir, "postmaster.pid"), n.logPath()) } - return n.restrictMaintenanceDatabases(ctx, socketDir) + postmaster, err := markProcess(pid) + if err != nil { + return postmasterRecord{}, fmt.Errorf("inspect the run's postmaster (pid %d): %w", pid, err) + } + record := postmasterRecord{Postmaster: postmaster, SocketDir: socketDir} + if err := writePostmasterRecord(n.root, record); err != nil { + return postmasterRecord{}, fmt.Errorf("record the run's postmaster: %w", err) + } + if err := n.restrictMaintenanceDatabases(ctx, socketDir); err != nil { + return postmasterRecord{}, err + } + return record, nil } // serverOptions are the postmaster's options, passed through pg_ctl's shell. @@ -392,54 +415,133 @@ func (n *nixPostgres) restrictMaintenanceDatabases(ctx context.Context, socketDi return nil } -// runningPostmaster reads the cluster's postmaster.pid: the live postmaster's -// PID and the socket directory it listens in. -func runningPostmaster(dataDir string) (int, string, bool) { - file, err := os.Open(filepath.Join(dataDir, "postmaster.pid")) +// postmasterRecord is what the run knows about its own server: the birth of +// the postmaster it started, and where that postmaster listens. It is written +// when the server starts, under the run's own 0700 root. +// +// The record — never postmaster.pid — is what authorizes a signal. A pid file +// outlives the process that wrote it (SIGKILL, OOM, a machine that lost +// power), PIDs are reused, and this agent collects the servers of OTHER runs +// on every native start. Trusting the number in that file would make an +// unrelated process of this user, which merely inherited the number, the +// target of the collection. +type postmasterRecord struct { + Postmaster processMark `json:"postmaster"` + SocketDir string `json:"socket"` +} + +func postmasterRecordPath(root string) string { return filepath.Join(root, "postmaster.json") } + +// readPostmasterRecord returns the recorded postmaster, if one was recorded. +// An unreadable or unparseable record is no record: nothing may be signalled +// on it, and the cluster is treated as having no server of ours. +func readPostmasterRecord(root string) (postmasterRecord, bool) { + data, err := os.ReadFile(postmasterRecordPath(root)) if err != nil { - return 0, "", false + return postmasterRecord{}, false } - defer file.Close() - var lines []string - scanner := bufio.NewScanner(file) - for scanner.Scan() && len(lines) < 5 { - lines = append(lines, strings.TrimSpace(scanner.Text())) + var record postmasterRecord + if err := json.Unmarshal(data, &record); err != nil || record.Postmaster.PID <= 0 { + return postmasterRecord{}, false } - if len(lines) < 5 { - return 0, "", false - } - pid, err := strconv.Atoi(lines[0]) - if err != nil || pid <= 0 || !processAlive(pid) { - return 0, "", false + return record, true +} + +func writePostmasterRecord(root string, record postmasterRecord) error { + data, err := json.Marshal(record) + if err != nil { + return err } - return pid, lines[4], lines[4] != "" + return writeOwnerOnly(postmasterRecordPath(root), string(data)) } -// clearStalePostmasterPid removes a leftover postmaster.pid when no live -// postmaster owns it. A crashed or SIGKILL'd server leaves a pid file that -// aborts the next boot (`FATAL: lock file "postmaster.pid" is empty` / "is the -// remnant of a previous server startup crash"). A pid file owned by a LIVE -// process is left untouched so a concurrent postmaster is never stomped. -func (n *nixPostgres) clearStalePostmasterPid() error { - pidPath := filepath.Join(n.dataDir, "postmaster.pid") - data, err := os.ReadFile(pidPath) - if errors.Is(err, os.ErrNotExist) { +// attachedPostmaster returns the run's live server, when there is one this +// run recorded and that is still the same process. Otherwise it reports no +// server, having first made sure the cluster can actually be started: see +// clearDisownedPostmasterPid. +func attachedPostmaster(root, dataDir string) (postmasterRecord, bool, error) { + record, recorded := readPostmasterRecord(root) + if recorded && record.Postmaster.alive() && record.SocketDir != "" { + return record, true, nil + } + if err := clearDisownedPostmasterPid(dataDir, record, recorded); err != nil { + return postmasterRecord{}, false, err + } + return postmasterRecord{}, false, nil +} + +// clearDisownedPostmasterPid removes a postmaster.pid this run's server no +// longer owns, and refuses when it cannot prove that. +// +// A crashed or SIGKILL'd server leaves a pid file that aborts the next boot +// (`FATAL: lock file "postmaster.pid" is empty` / "is the remnant of a +// previous server startup crash"), so a stale one has to go. But a pid file +// whose PID is alive is two completely different situations: +// +// - the number was reused by an unrelated process of this user. The file is +// stale, and leaving it would wedge every later start of this run behind a +// process that has nothing to do with postgres. +// - a postmaster really does hold this cluster. Removing its lock file would +// let a second postmaster start on the same data directory, which is how a +// cluster is corrupted. It is refused, by name, rather than worked around. +// +// The two are told apart by inspecting the process, never by signalling it. +func clearDisownedPostmasterPid(dataDir string, record postmasterRecord, recorded bool) error { + pidPath := filepath.Join(dataDir, "postmaster.pid") + pid, ok := postmasterPidFileOwner(pidPath) + if !ok { + // Absent, unreadable, corrupt, or naming a process that is gone. + if err := os.Remove(pidPath); err != nil && !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("remove the stale postgres lock file %s: %w", pidPath, err) + } return nil } + identity, err := runners.InspectProcess(pid) if err != nil { - return nil // best-effort; let postgres surface any real error + if errors.Is(err, runners.ErrProcessNotFound) { + return os.Remove(pidPath) + } + return fmt.Errorf("inspect the process holding %s (pid %d): %w", pidPath, pid, err) } - // Line 1 of postmaster.pid is the postmaster PID. An empty/corrupt file is - // stale by definition. - fields := strings.Fields(string(data)) - if len(fields) > 0 { - if pid, perr := strconv.Atoi(fields[0]); perr == nil && pid > 0 && processAlive(pid) { - return nil // a live process owns the lock — do NOT remove it + if isPostmasterExecutable(identity.Executable) { + if recorded { + return fmt.Errorf("the postgres server at %s is held by pid %d, which is not the postmaster this run started (pid %d): stop it before running again", + dataDir, pid, record.Postmaster.PID) } + return fmt.Errorf("the postgres server at %s is held by pid %d, which this run did not start: stop it before running again", + dataDir, pid) } + // Provably not a postmaster, so the number was reused and the file is a + // leftover. Nothing is signalled: the lock file is simply dropped. return os.Remove(pidPath) } +// postmasterPidFileOwner reads line 1 of a postmaster.pid and reports the PID +// when the file names a live process. Everything else — absent, unreadable, +// empty, corrupt, a process that is gone — reads as no owner. +func postmasterPidFileOwner(pidPath string) (int, bool) { + data, err := os.ReadFile(pidPath) + if err != nil { + return 0, false + } + fields := strings.Fields(string(data)) + if len(fields) == 0 { + return 0, false + } + pid, err := strconv.Atoi(fields[0]) + if err != nil || pid <= 0 || !processAlive(pid) { + return 0, false + } + return pid, true +} + +// isPostmasterExecutable reports whether an inspected process is a postgres +// server. core reports the executable's base name, and both the nix build and +// the historical name are accepted. +func isPostmasterExecutable(executable string) bool { + return executable == "postgres" || executable == "postmaster" +} + // processAlive reports whether a PID names a live process (signal 0 probe). func processAlive(pid int) bool { p, err := os.FindProcess(pid) @@ -577,10 +679,26 @@ func (n *nixPostgres) provisionDatabase(ctx context.Context, service, database, } defer db.Close() - var exists bool - if err := db.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = $1)`, owner).Scan(&exists); err != nil { + // A migration owner may create roles, because its migrations create the + // application's own NOLOGIN roles. On PostgreSQL 16 and later a role + // created that way is granted back to its creator WITH ADMIN OPTION, so a + // service that creates a role named after ANOTHER service's database — + // before that service first runs — would be adopted here as that + // database's owner and keep the admin grant, which is enough to take the + // role and read the data. An owner role is this server's to create: one + // that already carries a membership was not created by this agent. + var exists, granted bool + if err := db.QueryRowContext(ctx, ` + SELECT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = $1), + EXISTS (SELECT 1 FROM pg_auth_members m + JOIN pg_roles r ON r.oid = m.roleid + WHERE r.rolname = $1)`, owner).Scan(&exists, &granted); err != nil { return fmt.Errorf("look up the migration owner: %w", err) } + if granted { + return fmt.Errorf("migration owner %q on this run's postgres server is granted to another role, so it was not created for %s: remove it, or give this service its own database-name", + owner, service) + } quotedOwner := pq.QuoteIdentifier(owner) if !exists { if _, err := db.ExecContext(ctx, `CREATE ROLE `+quotedOwner); err != nil { @@ -592,10 +710,20 @@ func (n *nixPostgres) provisionDatabase(ctx context.Context, service, database, return fmt.Errorf("configure the migration owner: %w", err) } + // The database records the service it belongs to, and that record is a + // comment any role with CREATEDB could have written on a database of its + // own. Its current owner is the half that cannot be forged: a service + // creating a database owns it, and it cannot hand ownership to an owner + // role it is not a member of — which the membership check above is what + // keeps true. A database owned by this service's own owner with no comment + // yet is the one legitimate gap, between CREATE DATABASE and COMMENT, and + // the comment is simply written. marker := databaseClaim(service) var claim sql.NullString + var currentOwner string err = db.QueryRowContext(ctx, - `SELECT shobj_description(oid, 'pg_database') FROM pg_database WHERE datname = $1`, database).Scan(&claim) + `SELECT shobj_description(oid, 'pg_database'), pg_get_userbyid(datdba) FROM pg_database WHERE datname = $1`, + database).Scan(&claim, ¤tOwner) quotedDatabase := pq.QuoteIdentifier(database) switch { case errors.Is(err, sql.ErrNoRows): @@ -607,9 +735,16 @@ func (n *nixPostgres) provisionDatabase(ctx context.Context, service, database, } case err != nil: return fmt.Errorf("look up database %q: %w", database, err) - case claim.String != marker: + case currentOwner != owner: + return fmt.Errorf("database %q on this run's postgres server is owned by %q, not by %s's migration owner: give each service of a run its own database-name", + database, currentOwner, service) + case claim.Valid && claim.String != marker: return fmt.Errorf("database %q on this run's postgres server belongs to %s, not %s: give each service of a run its own database-name", database, strings.TrimPrefix(claim.String, databaseClaimPrefix), service) + case !claim.Valid: + if _, err := db.ExecContext(ctx, `COMMENT ON DATABASE `+quotedDatabase+` IS `+pq.QuoteLiteral(marker)); err != nil { + return fmt.Errorf("record database %q's service: %w", database, err) + } } for _, statement := range []string{ `ALTER DATABASE ` + quotedDatabase + ` OWNER TO ` + quotedOwner, @@ -760,47 +895,95 @@ func (n *nixPostgres) release(ctx context.Context) (bool, error) { if err != nil || live > 0 { return false, errors.Join(forwardErr, err) } - stopped, err := stopServer(n.dataDir) + stopped, err := stopServer(n.root, n.dataDir) return stopped, errors.Join(forwardErr, err) } -// stopServer shuts the cluster's postmaster down (fast shutdown: sessions are +// stopServer shuts a run's own postmaster down (fast shutdown: sessions are // disconnected, the server exits cleanly and removes its shared memory) and // removes its socket directory. It reports whether a server was running. -func stopServer(dataDir string) (bool, error) { - pid, socketDir, running := runningPostmaster(dataDir) - if !running { +// +// It signals the recorded birth and nothing else. With no record, or a record +// whose process cannot be re-confirmed, there is no server of this run's left +// to stop — whatever else may hold that PID by now is not this function's to +// touch, least of all on the machine-wide collection path. +func stopServer(root, dataDir string) (bool, error) { + record, recorded := readPostmasterRecord(root) + if !recorded || !record.Postmaster.confirmed() { + forgetPostmasterRecord(root, recorded) return false, nil } - process, err := os.FindProcess(pid) - if err != nil { - return false, err - } - if err := process.Signal(syscall.SIGINT); err != nil && processAlive(pid) { + pid := record.Postmaster.PID + if err := signalPostmaster(record.Postmaster, syscall.SIGINT); err != nil { + if !record.Postmaster.confirmed() { + forgetPostmasterRecord(root, true) + return false, nil + } return false, fmt.Errorf("stop the run's postgres server (pid %d): %w", pid, err) } - if !waitGone(pid, serverStopTimeout) { - // Immediate shutdown: the server still releases its shared memory. - _ = process.Signal(syscall.SIGQUIT) - if !waitGone(pid, 10*time.Second) { + if !waitGone(record.Postmaster, serverStopTimeout) { + // Immediate shutdown: the server still releases its shared memory. The + // birth is re-confirmed inside signalPostmaster, because a minute is + // long enough for the postmaster to have exited and its number to have + // been handed to somebody else. + _ = signalPostmaster(record.Postmaster, syscall.SIGQUIT) + if !waitGone(record.Postmaster, 10*time.Second) { return false, fmt.Errorf("the run's postgres server (pid %d) did not stop", pid) } } - if strings.HasPrefix(filepath.Base(socketDir), "cfpg-") { - _ = os.RemoveAll(socketDir) + if strings.HasPrefix(filepath.Base(record.SocketDir), "cfpg-") { + _ = os.RemoveAll(record.SocketDir) } + forgetPostmasterRecord(root, true) + // The postmaster removes its own lock file on an orderly exit; a SIGQUIT + // leaves it, and the next start would then have to tell that leftover + // apart from a live server all over again. + _ = os.Remove(filepath.Join(dataDir, "postmaster.pid")) return true, nil } -func waitGone(pid int, budget time.Duration) bool { +// signalPostmaster delivers a signal to a recorded postmaster, and only while +// that exact birth is still the process behind the number. +// +// The confirmation and the delivery are two system calls, so a process that +// exits between them cannot be signalled by this call at all (the signal is +// then rejected, as the PID no longer resolves) unless the number is reused in +// that same window. Narrowing it to those microseconds is the most a caller +// can do for a process that is not its child; what this removes is the real +// exposure, which is a file on disk naming a PID whose process died minutes, +// hours or a reboot ago. +func signalPostmaster(postmaster processMark, signal syscall.Signal) error { + if !postmaster.confirmed() { + return fmt.Errorf("postgres server %d is no longer the process this run started", postmaster.PID) + } + process, err := os.FindProcess(postmaster.PID) + if err != nil { + return err + } + return process.Signal(signal) +} + +// forgetPostmasterRecord drops a record no server answers to any more, so a +// later start is not told about a postmaster that is gone. +func forgetPostmasterRecord(root string, recorded bool) { + if !recorded { + return + } + _ = os.Remove(postmasterRecordPath(root)) +} + +// waitGone waits for a recorded birth to end. A PID that is answered by some +// other process already counts as gone: the postmaster this run started is +// what has to stop, not whatever inherits its number. +func waitGone(postmaster processMark, budget time.Duration) bool { deadline := time.Now().Add(budget) for time.Now().Before(deadline) { - if !processAlive(pid) { + if !postmaster.confirmed() { return true } time.Sleep(100 * time.Millisecond) } - return !processAlive(pid) + return !postmaster.confirmed() } // collectAbandonedServers stops every other run server on this machine whose @@ -823,7 +1006,7 @@ func collectAbandonedServers(ctx context.Context, servers, self string) { continue } if live, err := pruneHolders(root); err == nil && live == 0 { - _, _ = stopServer(filepath.Join(root, "pgdata")) + _, _ = stopServer(root, filepath.Join(root, "pgdata")) } _ = lock.Unlock() } diff --git a/nixpg_pidclean_test.go b/nixpg_pidclean_test.go index 70ce0e7..f580bda 100644 --- a/nixpg_pidclean_test.go +++ b/nixpg_pidclean_test.go @@ -7,10 +7,13 @@ import ( "testing" ) -// TestClearStalePostmasterPid verifies the pg_ctl-style stale-lock cleanup: -// a postmaster.pid owned by a dead/empty/corrupt entry is removed, but one -// owned by a live process is preserved (never stomp a concurrent postmaster). -func TestClearStalePostmasterPid(t *testing.T) { +// TestClearDisownedPostmasterPid verifies the pg_ctl-style stale-lock +// cleanup: a postmaster.pid that is empty, corrupt, or names a process that +// is gone is removed, and so is one whose live PID is provably not a postgres +// server, because the number was reused and the run would otherwise wait on a +// stranger. A lock file a real postmaster holds is refused instead, which +// TestLockFileHeldByARealPostmasterIsRefused covers. +func TestClearDisownedPostmasterPid(t *testing.T) { write := func(dir, content string) string { t.Helper() if err := os.MkdirAll(dir, 0o700); err != nil { @@ -28,8 +31,7 @@ func TestClearStalePostmasterPid(t *testing.T) { } t.Run("missing file is a no-op", func(t *testing.T) { - n := &nixPostgres{dataDir: t.TempDir()} - if err := n.clearStalePostmasterPid(); err != nil { + if err := clearDisownedPostmasterPid(t.TempDir(), postmasterRecord{}, false); err != nil { t.Fatalf("clear: %v", err) } }) @@ -37,8 +39,7 @@ func TestClearStalePostmasterPid(t *testing.T) { t.Run("empty file is removed", func(t *testing.T) { dir := t.TempDir() p := write(dir, "") - n := &nixPostgres{dataDir: dir} - if err := n.clearStalePostmasterPid(); err != nil { + if err := clearDisownedPostmasterPid(dir, postmasterRecord{}, false); err != nil { t.Fatalf("clear: %v", err) } if exists(p) { @@ -50,8 +51,7 @@ func TestClearStalePostmasterPid(t *testing.T) { dir := t.TempDir() // PID 2^31-1 is not a live process on any sane host. p := write(dir, "2147483646\n/data\n1700000000\n5432\n") - n := &nixPostgres{dataDir: dir} - if err := n.clearStalePostmasterPid(); err != nil { + if err := clearDisownedPostmasterPid(dir, postmasterRecord{}, false); err != nil { t.Fatalf("clear: %v", err) } if exists(p) { @@ -59,19 +59,16 @@ func TestClearStalePostmasterPid(t *testing.T) { } }) - t.Run("live pid is preserved", func(t *testing.T) { + t.Run("live pid that is not a postmaster is removed", func(t *testing.T) { dir := t.TempDir() - // Our own PID is definitely alive — must NOT be removed. - p := write(dir, "") - if err := os.WriteFile(p, []byte(strconv.Itoa(os.Getpid())+"\n/data\n"), 0o600); err != nil { - t.Fatal(err) - } - n := &nixPostgres{dataDir: dir} - if err := n.clearStalePostmasterPid(); err != nil { + // This test binary is alive and is certainly not a postgres server, + // so the lock file naming it is a leftover with a reused number. + p := write(dir, strconv.Itoa(os.Getpid())+"\n/data\n") + if err := clearDisownedPostmasterPid(dir, postmasterRecord{}, false); err != nil { t.Fatalf("clear: %v", err) } - if !exists(p) { - t.Fatal("postmaster.pid owned by a LIVE process must be preserved") + if exists(p) { + t.Fatal("postmaster.pid naming a live non-postgres process should have been removed") } }) } diff --git a/nixpg_shared_test.go b/nixpg_shared_test.go index 7ad31c7..0ba217d 100644 --- a/nixpg_shared_test.go +++ b/nixpg_shared_test.go @@ -5,12 +5,14 @@ import ( "database/sql" "net/url" "os" + "path/filepath" "testing" runtimev0 "github.com/codefly-dev/core/generated/go/codefly/services/runtime/v0" "github.com/codefly-dev/core/resources" runners "github.com/codefly-dev/core/runners/base" "github.com/codefly-dev/core/shared" + "github.com/lib/pq" "github.com/stretchr/testify/require" ) @@ -64,8 +66,10 @@ func TestRunServerIsSharedAndIsolated(t *testing.T) { // One server for the run. require.Equal(t, runAlpha.nixRuntime.dataDir, runBeta.nixRuntime.dataDir) - postmaster, _, running := runningPostmaster(runAlpha.nixRuntime.dataDir) + attached, running, err := attachedPostmaster(runAlpha.nixRuntime.root, runAlpha.nixRuntime.dataDir) + require.NoError(t, err) require.True(t, running) + postmaster := attached.Postmaster.PID // Each service at its own endpoint, with its own data. requireListener(t, alpha.address) @@ -153,8 +157,10 @@ func TestAbandonedRunServerIsCollected(t *testing.T) { abandoned := newPostgresFixtureAt(t, ctx, t.TempDir(), "gone", "store", resources.NewRuntimeContextNix()) run, _ := abandoned.start(t, ctx) dataDir := run.nixRuntime.dataDir - postmaster, _, running := runningPostmaster(dataDir) + attached, running, err := attachedPostmaster(run.nixRuntime.root, dataDir) + require.NoError(t, err) require.True(t, running) + postmaster := attached.Postmaster.PID // The run dies: its agent's record names a process that no longer exists, // and its endpoint went with it. @@ -168,3 +174,82 @@ func TestAbandonedRunServerIsCollected(t *testing.T) { next.start(t, ctx) require.False(t, processAlive(postmaster), "an abandoned run's server must be stopped by the next run") } + +// TestLockFileHeldByARealPostmasterIsRefused is the case where the lock file +// must be left exactly as it is. A record this agent cannot read — lost, or +// written by a version that did not keep one — does not make the postmaster +// holding the cluster disposable: removing its lock file would let a second +// postmaster start on the same data directory, which is how a cluster is +// corrupted. The run refuses and names the process instead, and signals +// nothing. +func TestLockFileHeldByARealPostmasterIsRefused(t *testing.T) { + requireNix(t) + ctx := context.Background() + fixture := newPostgresFixtureAt(t, ctx, t.TempDir(), "held", "store", resources.NewRuntimeContextNix()) + run, _ := fixture.start(t, ctx) + root := run.nixRuntime.root + pidPath := filepath.Join(run.nixRuntime.dataDir, "postmaster.pid") + postmaster, running := postmasterPidFileOwner(pidPath) + require.True(t, running) + + // The record goes missing while its postmaster keeps serving. + recorded, err := os.ReadFile(postmasterRecordPath(root)) + require.NoError(t, err) + require.NoError(t, os.Remove(postmasterRecordPath(root))) + t.Cleanup(func() { _ = os.WriteFile(postmasterRecordPath(root), recorded, 0o600) }) + + _, attached, err := attachedPostmaster(root, run.nixRuntime.dataDir) + require.False(t, attached) + require.ErrorContains(t, err, "stop it before running again") + require.FileExists(t, pidPath, "a live postmaster's lock file is never removed") + require.True(t, processAlive(postmaster), "and its postmaster is never signalled") + + // Unreadable record, same answer: nothing of this run's may be stopped on + // a PID alone, so the server is still there for the restored record. + require.NoError(t, os.WriteFile(postmasterRecordPath(root), []byte("{"), 0o600)) + _, attached, err = attachedPostmaster(root, run.nixRuntime.dataDir) + require.False(t, attached) + require.ErrorContains(t, err, "stop it before running again") + require.True(t, processAlive(postmaster)) +} + +// TestRunServerRefusesAnOwnerRoleAnotherServiceCreated closes the one way a +// service could reach another's data on the shared server. A migration owner +// may create roles, because migrations create the application's own roles, +// and PostgreSQL 16 and later grant a role created that way back to its +// creator WITH ADMIN OPTION. A service that creates a role named after +// ANOTHER service's database, before that service has ever run, would have +// been adopted as that database's owner while keeping the admin grant — which +// is enough to take the role and read the data. +func TestRunServerRefusesAnOwnerRoleAnotherServiceCreated(t *testing.T) { + requireNix(t) + ctx := context.Background() + workspace := t.TempDir() + squatter := newPostgresFixtureAt(t, ctx, workspace, "squatter", "store", resources.NewRuntimeContextNix()) + victim := newPostgresFixtureAt(t, ctx, workspace, "victim", "store", resources.NewRuntimeContextNix()) + + run, _ := squatter.start(t, ctx) + // The squatter's migrations create the victim's owner role before the + // victim first runs. This is exactly what a migration is allowed to do. + db, err := sql.Open("postgres", run.connection) + require.NoError(t, err) + defer db.Close() + stolen := nativeOwnerRole("victim") + _, err = db.ExecContext(ctx, `CREATE ROLE `+pq.QuoteIdentifier(stolen)+` LOGIN`) + require.NoError(t, err, "a migration owner may create roles") + + refused := NewRuntime() + _, err = refused.Load(ctx, &runtimev0.LoadRequest{ + Identity: victim.identity, Environment: shared.Must(victim.environment.Proto()), DisableCatch: true, + }) + require.NoError(t, err) + init, err := refused.Init(ctx, &runtimev0.InitRequest{ + RuntimeContext: victim.runtimeContext, + Configuration: victim.configuration, + ProposedNetworkMappings: victim.networkMappings, + }) + require.NoError(t, err) + require.Equal(t, runtimev0.InitStatus_ERROR, init.GetStatus().GetState()) + require.Contains(t, init.GetStatus().GetMessage(), "was not created for victim/store") + require.Nil(t, refused.nixRuntime, "a refused service holds no server") +} diff --git a/nixpg_signal_test.go b/nixpg_signal_test.go new file mode 100644 index 0000000..537198a --- /dev/null +++ b/nixpg_signal_test.go @@ -0,0 +1,120 @@ +package main + +import ( + "context" + "encoding/json" + "os" + "os/exec" + "path/filepath" + "strconv" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// liveStranger starts a process that is not a postgres server, dies on SIGINT, +// and outlives the test unless something kills it. Its PID stands in for the +// number a postmaster had before it died and the kernel handed on. +func liveStranger(t *testing.T) int { + t.Helper() + stranger := exec.Command("sleep", "300") + require.NoError(t, stranger.Start()) + t.Cleanup(func() { + _ = stranger.Process.Kill() + _ = stranger.Wait() + }) + return stranger.Process.Pid +} + +// abandonedRoot is a run server root left behind by a run that died: holders, +// a cluster directory, and a postmaster.pid naming pid. Nothing records a +// postmaster, because nothing orderly happened here. +func abandonedRoot(t *testing.T, servers, name string, pid int) string { + t.Helper() + root := filepath.Join(servers, name) + require.NoError(t, os.MkdirAll(filepath.Join(root, "holders"), 0o700)) + require.NoError(t, os.MkdirAll(filepath.Join(root, "pgdata"), 0o700)) + require.NoError(t, os.WriteFile(filepath.Join(root, "pgdata", "postmaster.pid"), + []byte(strconv.Itoa(pid)+"\n"+filepath.Join(root, "pgdata")+"\n1700000000\n5432\n/tmp/cfpg-gone\n"), 0o600)) + return root +} + +func stillAlive(t *testing.T, pid int) bool { + t.Helper() + // A SIGINT'd `sleep` dies promptly; give it room to, so this cannot pass + // by being quicker than the signal. + time.Sleep(500 * time.Millisecond) + return processAlive(pid) +} + +// TestStaleLockFileIsNeverSignalled is the regression for the one way this +// agent could reach outside its own run: a postmaster.pid outlives the +// postmaster that wrote it (SIGKILL, OOM, a machine that lost power), PIDs are +// reused, and collection runs over every other run root on the machine at +// every native start. Signalling the number in that file makes an unrelated +// process of this user the target. +func TestStaleLockFileIsNeverSignalled(t *testing.T) { + ctx := context.Background() + servers := t.TempDir() + stranger := liveStranger(t) + root := abandonedRoot(t, servers, "abandoned", stranger) + + stopped, err := stopServer(root, filepath.Join(root, "pgdata")) + require.NoError(t, err) + require.False(t, stopped, "no postmaster was recorded, so none was stopped") + + collectAbandonedServers(ctx, servers, filepath.Join(servers, "other")) + + require.True(t, stillAlive(t, stranger), + "a process that merely inherited the PID in a stale postmaster.pid must never be signalled") +} + +// TestRecordedPostmasterIsSignalledOnlyAsTheSameBirth: the record is not a +// PID either. A recorded birth whose process is gone — its number now +// answered by something else — is nobody to signal. +func TestRecordedPostmasterIsSignalledOnlyAsTheSameBirth(t *testing.T) { + servers := t.TempDir() + stranger := liveStranger(t) + root := abandonedRoot(t, servers, "reused", stranger) + + mark, err := markProcess(stranger) + require.NoError(t, err) + mark.StartID++ // the same number, a different process + record, err := json.Marshal(postmasterRecord{Postmaster: mark, SocketDir: "/tmp/cfpg-gone"}) + require.NoError(t, err) + require.NoError(t, os.WriteFile(postmasterRecordPath(root), record, 0o600)) + + stopped, err := stopServer(root, filepath.Join(root, "pgdata")) + require.NoError(t, err) + require.False(t, stopped) + require.True(t, stillAlive(t, stranger), + "a recorded birth that no longer owns its PID must not be signalled") + require.NoFileExists(t, postmasterRecordPath(root), + "a record no server answers to is dropped, so a later start is not told about it") +} + +// TestStaleLockFileDoesNotWedgeTheRun is the other half of the same stale +// file: it must not be read as "a server is already running" either, or the +// run waits on a stranger's process instead of starting its own server. +func TestStaleLockFileDoesNotWedgeTheRun(t *testing.T) { + servers := t.TempDir() + stranger := liveStranger(t) + root := abandonedRoot(t, servers, "wedged", stranger) + pidPath := filepath.Join(root, "pgdata", "postmaster.pid") + + record, running, err := attachedPostmaster(root, filepath.Join(root, "pgdata")) + require.NoError(t, err) + require.False(t, running, "a stale lock file is not a running server") + require.Zero(t, record.Postmaster.PID) + require.NoFileExists(t, pidPath, "the stale lock file is cleared so the server can start") + require.True(t, stillAlive(t, stranger), "clearing it signals nothing") +} + +func TestIsPostmasterExecutable(t *testing.T) { + require.True(t, isPostmasterExecutable("postgres")) + require.True(t, isPostmasterExecutable("postmaster")) + for _, stranger := range []string{"", "sleep", "postgres-wrapper", "pg_ctl", "node"} { + require.False(t, isPostmasterExecutable(stranger), stranger) + } +} diff --git a/recovery_scope_test.go b/recovery_scope_test.go index d8e2ef0..05f2a56 100644 --- a/recovery_scope_test.go +++ b/recovery_scope_test.go @@ -41,7 +41,8 @@ func TestMain(m *testing.M) { func stopEveryRunServer(servers string) { entries, _ := os.ReadDir(servers) for _, entry := range entries { - if stopped, err := stopServer(filepath.Join(servers, entry.Name(), "pgdata")); err != nil || stopped { + root := filepath.Join(servers, entry.Name()) + if stopped, err := stopServer(root, filepath.Join(root, "pgdata")); err != nil || stopped { fmt.Fprintf(os.Stderr, "run server %s outlived its test (stopped=%v err=%v)\n", entry.Name(), stopped, err) } }