From a9ec8ccfe176ae2ed6dc5b0a8a9194dcba441d33 Mon Sep 17 00:00:00 2001 From: Jon Langevin Date: Sat, 3 Oct 2026 03:54:45 -0400 Subject: [PATCH] feat(migrate): expose in-process runner --- CHANGELOG.md | 18 + go.mod | 27 +- go.sum | 104 ++-- internal/atlas/driver.go | 42 +- internal/atlas/driver_recover_test.go | 16 +- internal/atlas/revisions.go | 4 +- internal/golangmigrate/driver.go | 239 +++++++- internal/golangmigrate/driver_context_test.go | 178 ++++++ pkg/cli/root.go | 152 ++--- pkg/cli/root_test.go | 126 +++- pkg/driver/registry.go | 13 + pkg/runner/runner.go | 164 ++++++ pkg/runner/runner_test.go | 542 ++++++++++++++++++ 13 files changed, 1432 insertions(+), 193 deletions(-) create mode 100644 internal/golangmigrate/driver_context_test.go create mode 100644 pkg/runner/runner.go create mode 100644 pkg/runner/runner_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index f03f898..8f0fc8c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Public `pkg/runner.New` and `driver.NewDefaultRegistry` for ordered in-process + migrations with explicit requests, caller context, operation timeouts, typed + results, and injected output. Runner diagnostics redact connection credentials + and are limited to 4 KiB. +- Ordinary migration CLI commands use the same runner with command context and + output instead of process-global test hooks; `--timeout` bounds operations. + - `up --up-if-clean` flag: makes the `up` subcommand idempotent in deploy CMDs. Same effect as plain `up` when no migrations are pending (exits 0 with an informative message); the difference is that the flag is now accepted by cobra. Resolves @@ -20,6 +27,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Native golang-migrate operations propagate caller cancellation through + connection establishment, SQL, and advisory-lock waits. PostgreSQL receives a + cancel request; failed initialization closes its reserved connection. +- Atlas revision-table and library initialization use the operation context; + the timeout flag is limited to the four runner-backed CLI commands. +- Status retains the dirty nil-version marker after failed rollback of the first + migration instead of reporting a clean database. + +- External SDK CLI dispatch accepts the leading `db-migrate` command supplied by + wfctl instead of interpreting it as an unknown nested command. + - Atlas Executor panic recovery: `Up()` and `Status()` in the atlas driver now wrap calls into `ariga.io/atlas/sql/migrate.(*Executor).*` with `defer recover()` so an upstream-library panic (`runtime error: index out of range [0] with length 0` observed diff --git a/go.mod b/go.mod index 773612f..253fd1c 100644 --- a/go.mod +++ b/go.mod @@ -1,13 +1,13 @@ module github.com/GoCodeAlone/workflow-plugin-migrations -go 1.26.4 +go 1.26.5 require ( ariga.io/atlas v1.2.0 - github.com/GoCodeAlone/workflow v0.80.24 + github.com/GoCodeAlone/workflow v0.86.0 github.com/fergusstrange/embedded-postgres v1.34.0 github.com/golang-migrate/migrate/v4 v4.19.1 - github.com/jackc/pgx/v5 v5.9.2 + github.com/jackc/pgx/v5 v5.10.0 github.com/pressly/goose/v3 v3.27.0 github.com/spf13/cobra v1.10.2 gopkg.in/yaml.v3 v3.0.1 @@ -16,7 +16,7 @@ require ( require ( github.com/BurntSushi/toml v1.6.0 // indirect github.com/GoCodeAlone/go-plugin v1.7.0 // indirect - github.com/GoCodeAlone/modular v1.13.4 // indirect + github.com/GoCodeAlone/modular v1.13.6 // indirect github.com/GoCodeAlone/yaegi v0.17.2 // indirect github.com/agext/levenshtein v1.2.1 // indirect github.com/apparentlymart/go-textseg/v13 v13.0.0 // indirect @@ -42,7 +42,7 @@ require ( github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/json-iterator/go v1.1.13-0.20220915233716-71ac16282d12 // indirect - github.com/lib/pq v1.10.9 // indirect + github.com/lib/pq v1.12.3 // indirect github.com/mattn/go-colorable v0.1.15 // indirect github.com/mattn/go-isatty v0.0.22 // indirect github.com/mfridman/interpolate v0.0.2 // indirect @@ -59,18 +59,17 @@ require ( github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect github.com/zclconf/go-cty v1.14.4 // indirect github.com/zclconf/go-cty-yaml v1.1.0 // indirect - go.opentelemetry.io/otel/sdk/metric v1.44.0 // indirect go.uber.org/multierr v1.11.0 // indirect go.uber.org/zap v1.28.0 // indirect - golang.org/x/net v0.56.0 // indirect - golang.org/x/sync v0.21.0 // indirect - golang.org/x/sys v0.46.0 // indirect - golang.org/x/text v0.38.0 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20260608224507-4308a22a1bab // indirect - google.golang.org/grpc v1.81.1 // indirect + golang.org/x/net v0.58.0 // indirect + golang.org/x/sync v0.23.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.org/x/text v0.42.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260630182238-925bb5da69e7 // indirect + google.golang.org/grpc v1.83.2 // indirect google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect - modernc.org/libc v1.72.3 // indirect + modernc.org/libc v1.73.4 // indirect modernc.org/mathutil v1.7.1 // indirect modernc.org/memory v1.11.0 // indirect - modernc.org/sqlite v1.51.0 // indirect + modernc.org/sqlite v1.53.0 // indirect ) diff --git a/go.sum b/go.sum index 978b8c9..d3c54ec 100644 --- a/go.sum +++ b/go.sum @@ -10,18 +10,18 @@ github.com/DataDog/datadog-go/v5 v5.8.3 h1:s58CUJ9s8lezjhTNJO/SxkPBv2qZjS3ktpRSq github.com/DataDog/datadog-go/v5 v5.8.3/go.mod h1:K9kcYBlxkcPP8tvvjZZKs/m1edNAUFzBbdpTUKfCsuw= github.com/GoCodeAlone/go-plugin v1.7.0 h1:EwnhqPlXiNmp85S+MXnKKvm3YlfA6O4NzBb4+GSlEVY= github.com/GoCodeAlone/go-plugin v1.7.0/go.mod h1:HbGQRZUIa+jbDfjsaZIMJYvrz+LnxL0mJpggfynSTMk= -github.com/GoCodeAlone/modular v1.13.4 h1:De4p2qyJSVmstRGno/PM+fPdUCMu/7a9WgU5FUVGDa8= -github.com/GoCodeAlone/modular v1.13.4/go.mod h1:+JEPUYOxGaD332EMZ5PbJCz5rxwvFu4Tm6MrnZT0vxM= -github.com/GoCodeAlone/modular/modules/auth v1.17.0 h1:GbKG6s/2qe6N9YZ8vtvYsNon56MLWECncPxWvAsazSc= -github.com/GoCodeAlone/modular/modules/auth v1.17.0/go.mod h1:E9dDIxiAxIrXK8gn/rEhaqI5OYe6Aw/uGpRyI7iyxj8= +github.com/GoCodeAlone/modular v1.13.6 h1:AIdfdR/KTPHX8T5KJBXUHMokliutUHjl6jRfkqdh/7Q= +github.com/GoCodeAlone/modular v1.13.6/go.mod h1:Vs92YltRNqA/v3peoHdqHTCjD10ww7rY3iURf1Ju450= +github.com/GoCodeAlone/modular/modules/auth v1.19.0 h1:ughD47BqEAhg7+3trucwcs5KzEpq9gIS/6jY4nkeGtA= +github.com/GoCodeAlone/modular/modules/auth v1.19.0/go.mod h1:pGPazN7Irlza8kUxE7WsXri44PhURKYBGym1tlBUl+k= github.com/GoCodeAlone/modular/modules/eventbus/v2 v2.10.0 h1:2ljVafd/1LYchF47WrnA1+ji8mcmVXMJ4F5qDrhZZi4= github.com/GoCodeAlone/modular/modules/eventbus/v2 v2.10.0/go.mod h1:AKLcRGsw5gp2Q1zhuK0TBnMJOsaRe3aJ2OKnLFE2O5w= -github.com/GoCodeAlone/workflow v0.80.24 h1:feY7H0rhs+FQYjcDsgK/1O1e4ZhangBvP2V7URhHLxA= -github.com/GoCodeAlone/workflow v0.80.24/go.mod h1:lGLd3hhGEvz52ovdlmS2d5DAFvpkVMlbJPTaSyGYSIE= +github.com/GoCodeAlone/workflow v0.86.0 h1:9NatmWg2120RHDFesltawOAS7jJvzyJ4k9Q+5WueySg= +github.com/GoCodeAlone/workflow v0.86.0/go.mod h1:WaH+HhX8lrmoGu/vQjkyloZTydUWZVVRo9KFJsZAUtI= github.com/GoCodeAlone/yaegi v0.17.2 h1:WK6Y6e0t1a6U7r+S2dN3CGWW1PizYD3zO0zneToZPxM= github.com/GoCodeAlone/yaegi v0.17.2/go.mod h1:z5Pr6Wse6QJcQvpgxTxzMAevFarH0N37TG88Y9dprx0= -github.com/IBM/sarama v1.47.0 h1:GcQFEd12+KzfPYeLgN69Fh7vLCtYRhVIx0rO4TZO318= -github.com/IBM/sarama v1.47.0/go.mod h1:7gLLIU97nznOmA6TX++Qds+DRxH89P2XICY2KAQUzAY= +github.com/IBM/sarama v1.50.3 h1:zpY2iZYmt+z+0Bo3aYF+cD48OBt2hIgiDPZUuZKTXcc= +github.com/IBM/sarama v1.50.3/go.mod h1:Jo4MSfdDT3ycmQj7/ab8eLZwnvwCKZm/8H7SCbtyo8U= github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= github.com/agext/levenshtein v1.2.1 h1:QmvMAjj2aEICytGiWzmxoE0x2KZvE0fvmqMOfy2tjT8= @@ -108,15 +108,13 @@ github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkp github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/eapache/go-resiliency v1.7.0 h1:n3NRTnBn5N0Cbi/IeOHuQn9s2UwVUH7Ga0ZWcP+9JTA= github.com/eapache/go-resiliency v1.7.0/go.mod h1:5yPzW0MIvSe0JDsv0v+DvcjEv2FyD6iZYSs1ZI+iQho= -github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc= -github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I= github.com/expr-lang/expr v1.17.8 h1:W1loDTT+0PQf5YteHSTpju2qfUfNoBt4yw9+wOEU9VM= github.com/expr-lang/expr v1.17.8/go.mod h1:8/vRC7+7HBzESEqt5kKpYXxrxkr31SaO8r40VO/1IT4= github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk= github.com/fatih/color v1.19.0 h1:Zp3PiM21/9Ld6FzSKyL5c/BULoe/ONr9KlbYVOfG8+w= github.com/fatih/color v1.19.0/go.mod h1:zNk67I0ZUT1bEGsSGyCZYZNrHuTkJJB+r6Q9VuMi0LE= -github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= -github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= +github.com/felixge/httpsnoop v1.1.0 h1:3YtUj32ZZkqZtt3sZZsClsymw/QDuVfpNhoA31zeORc= +github.com/felixge/httpsnoop v1.1.0/go.mod h1:Zqxgdd+1Rkcz8euOqdr7lqgCRJztwr5hp9vDSi5UZCE= github.com/fergusstrange/embedded-postgres v1.34.0 h1:c6RKhPKFsLVU+Tdxsx8q0UxCHsvZZ/iShAnljRBXs6s= github.com/fergusstrange/embedded-postgres v1.34.0/go.mod h1:w0YvnCgf19o6tskInrOOACtnqfVlOvluz3hlNLY7tRk= github.com/fsnotify/fsnotify v1.10.1 h1:b0/UzAf9yR5rhf3RPm9gf3ehBPpf0oZKIjtpKrx59Ho= @@ -129,6 +127,8 @@ github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/go-openapi/inflect v0.19.0 h1:9jCH9scKIbHeV9m12SmPilScz6krDxKRasNNSNPXu/4= github.com/go-openapi/inflect v0.19.0/go.mod h1:lHpZVlpIQqLyKwJ4N+YSc9hchQy/i12fJykb83CRBH4= +github.com/go-openapi/jsonpointer v0.24.0 h1:AA6mCjHYHmZ+1RU2Js089EaOK/iwXXNwQsTgnsTha2M= +github.com/go-openapi/jsonpointer v0.24.0/go.mod h1:Z3rw7dWu1p9IgitXCFamSlA5lmDiklEB6vkaxcNZW5Y= github.com/go-test/deep v1.1.1 h1:0r/53hagsehfO4bzD2Pgr/+RgHqhmf+k1Bpse2cTu1U= github.com/go-test/deep v1.1.1/go.mod h1:5C2ZWiW0ErCdrYzpqxLbTX7MG14M9iiw8DgHncVwcsE= github.com/godbus/dbus v0.0.0-20190726142602-4481cbc300e2 h1:ZpnhV/YsD2/4cESfV5+Hoeu/iUR3ruzNvZ+yQfO03a0= @@ -201,8 +201,8 @@ github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsI github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= -github.com/jackc/pgx/v5 v5.9.2 h1:3ZhOzMWnR4yJ+RW1XImIPsD1aNSz4T4fyP7zlQb56hw= -github.com/jackc/pgx/v5 v5.9.2/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= +github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/jcmturner/aescts/v2 v2.0.0 h1:9YKLH6ey7H4eDBXW8khjYslgyqG2xZikXP0EQFKrle8= @@ -219,8 +219,8 @@ github.com/jhump/protoreflect v1.16.0 h1:54fZg+49widqXYQ0b+usAFHbMkBGR4PpXrsHc8+ github.com/jhump/protoreflect v1.16.0/go.mod h1:oYPd7nPvcBw/5wlDfm/AVmU9zH9BgqGCI469pGxfj/8= github.com/json-iterator/go v1.1.13-0.20220915233716-71ac16282d12 h1:9Nu54bhS/H/Kgo2/7xNSUuC5G28VR8ljfrLKU2G4IjU= github.com/json-iterator/go v1.1.13-0.20220915233716-71ac16282d12/go.mod h1:TBzl5BIHNXfS9+C35ZyJaklL7mLDbgUkcgXzSLa8Tk0= -github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= -github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.19.0 h1:sXLILfc9jV2QYWkzFOPWStmcUVH2RHEB1JCdY2oVvCQ= +github.com/klauspost/compress v1.19.0/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= @@ -230,8 +230,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= -github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= -github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= +github.com/lib/pq v1.12.3 h1:tTWxr2YLKwIvK90ZXEw8GP7UFHtcbTtty8zsI+YjrfQ= +github.com/lib/pq v1.12.3/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA= github.com/mattn/go-colorable v0.1.9/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc= github.com/mattn/go-colorable v0.1.12/go.mod h1:u5H1YNBxpqRaxsYJYSkiCWKzEfiAb1Gb520KVy5xxl4= github.com/mattn/go-colorable v0.1.15 h1:+u9SLTRGnXv73cEsnsmoZBom+dMU88B2M0aDcWy0/jY= @@ -278,8 +278,8 @@ github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M= -github.com/pierrec/lz4/v4 v4.1.26 h1:GrpZw1gZttORinvzBdXPUXATeqlJjqUG/D87TKMnhjY= -github.com/pierrec/lz4/v4 v4.1.26/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= +github.com/pierrec/lz4/v4 v4.1.27 h1:+PhzhWDrjRj89TH2sw43nE3+4+W8lSxIuQadEHZyjUk= +github.com/pierrec/lz4/v4 v4.1.27/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= @@ -298,8 +298,8 @@ github.com/prometheus/procfs v0.20.1 h1:XwbrGOIplXW/AU3YhIhLODXMJYyC1isLFfYCsTEy github.com/prometheus/procfs v0.20.1/go.mod h1:o9EMBZGRyvDrSPH1RqdxhojkuXstoe4UlK79eF5TGGo= github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 h1:bsUq1dX0N8AOIL7EB/X911+m4EHsnWEHeJ0c+3TTBrg= github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= -github.com/redis/go-redis/v9 v9.20.0 h1:WnQYxLkgO2xiXTCJY0ldIiI8dNqCDlQAG+AtaH7a2a0= -github.com/redis/go-redis/v9 v9.20.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= +github.com/redis/go-redis/v9 v9.21.0 h1:FPBE4hhbAke+TLmcY3WkpbDffJEomdqPn3HYiqAtL9E= +github.com/redis/go-redis/v9 v9.21.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= @@ -380,38 +380,38 @@ go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= -golang.org/x/crypto v0.53.0 h1:QZ4Muo8THX6CizN2vPPd5fBGHyogrdK9fG4wLPFUsto= -golang.org/x/crypto v0.53.0/go.mod h1:DNLU434OwVakk9PzuwV8w62mAJpRJL3vsgcfp4Qnsio= -golang.org/x/mod v0.37.0 h1:vF1DjpVEshcIqoEaauuHebaLk1O1forxjxBaVn884JQ= -golang.org/x/mod v0.37.0/go.mod h1:m8S8VeM9r4dzDwjrKO0a1sZP3YjeMamRRlD+fmR2Q/0= -golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o= -golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/mod v0.41.0 h1:qJmnOUb4YB+FsEuM3HcWucdZASCPGhsX6uljO6pog0c= +golang.org/x/mod v0.41.0/go.mod h1:Ek9pY8RKWXwsWvd3rQiHYtMqkjSUV+s1Rj7j4H5Ur6o= +golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= +golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= -golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= -golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210927094055-39ccf1dd6fa6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= -golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -golang.org/x/text v0.38.0 h1:sXmwo9DwP3OK9EZ7PqAdaooSGozfl/3a6/xJcbzPRhE= -golang.org/x/text v0.38.0/go.mod h1:YXZt3QhHUKYT53r2lLKFIVi6Ao1jdzrTR/KQ09qyxF4= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= -golang.org/x/tools v0.45.0 h1:18qN3FAooORvApf5XjCXgsuayZOEtXf6JK18I3+ONa8= -golang.org/x/tools v0.45.0/go.mod h1:LuUGqqaXcXMEFEruIVJVm5mgDD8vww/z/SR1gQ4uE/0= +golang.org/x/tools v0.49.0 h1:3NI7VXzL9+1WZD52Dx2ttoPwD5DWrFGpl9mFZDlmisI= +golang.org/x/tools v0.49.0/go.mod h1:SJNXV9DBKT0UbdttsQjbfJlAE/q+y36++zo3uL3N0Oo= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= google.golang.org/genproto v0.0.0-20250603155806-513f23925822 h1:rHWScKit0gvAPuOnu87KpaYtjK5zBMLcULh7gxkCXu4= -google.golang.org/genproto/googleapis/api v0.0.0-20260608224507-4308a22a1bab h1:Foefixyu0l973HSYkX8Etw/fPxAmKRhyMGwuqXFiVI0= -google.golang.org/genproto/googleapis/api v0.0.0-20260608224507-4308a22a1bab/go.mod h1:KdNqO+rCIWgFumrNBSEDlDNrkrQnpkax7Tv1WxNY8V4= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260608224507-4308a22a1bab h1:cY0oV1VnAqvaim8VsR8ZyEKAudzbRJMRGwD3W/L7yOw= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260608224507-4308a22a1bab/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= -google.golang.org/grpc v1.81.1 h1:VnnIIZ88UzOOKLukQi+ImGz8O1Wdp8nAGGnvOfEIWQQ= -google.golang.org/grpc v1.81.1/go.mod h1:xGH9GfzOyMTGIOXBJmXt+BX/V0kcdQbdcuwQ/zNw42I= +google.golang.org/genproto/googleapis/api v0.0.0-20260630182238-925bb5da69e7 h1:jQ9p21COKWjP3VwuFrNRiiOTMh3mPpN45R7SLrH/HUU= +google.golang.org/genproto/googleapis/api v0.0.0-20260630182238-925bb5da69e7/go.mod h1:KqHwBx2upmfa1XSi1WuRvC+2VGCLtooKkfmyvRbUmqA= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260630182238-925bb5da69e7 h1:eM/YSd5bBFagF51o1E745Ta7RwzpW0h+z+QDNZOgmQ8= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260630182238-925bb5da69e7/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU= +google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsokw1S8= google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af h1:+5/Sw3GsDNlEmu7TfklWKPdQ0Ykja5VEmq2i817+jbI= google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= @@ -420,20 +420,20 @@ gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EV gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= -modernc.org/cc/v4 v4.28.2 h1:3tQ0lf2ADtoby2EtSP+J7IE2SHwEJdP8ioR59wx7XpY= -modernc.org/cc/v4 v4.28.2/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI= -modernc.org/ccgo/v4 v4.34.0 h1:yRLPFZieg532OT4rp4JFNIVcquwalMX26G95WQDqwCQ= -modernc.org/ccgo/v4 v4.34.0/go.mod h1:AS5WYMyBakQ+fhsHhtP8mWB82KTGPkNNJDGfGQCe0/A= +modernc.org/cc/v4 v4.28.4 h1:Hd/4Es+MBj+/7hSdZaisNyu6bv3V0Dp2MdllyfqaH+c= +modernc.org/cc/v4 v4.28.4/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI= +modernc.org/ccgo/v4 v4.34.4 h1:OVnSOWQjVKOYkFxoHYB+qQmSHK5gqMqARM+K9DpR/Ws= +modernc.org/ccgo/v4 v4.34.4/go.mod h1:qdKqE8FNIYyysougB1RX9MxCzp5oJOcQXSobANJ4TuE= modernc.org/fileutil v1.4.0 h1:j6ZzNTftVS054gi281TyLjHPp6CPHr2KCxEXjEbD6SM= modernc.org/fileutil v1.4.0/go.mod h1:EqdKFDxiByqxLk8ozOxObDSfcVOv/54xDs/DUHdvCUU= modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI= modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito= -modernc.org/gc/v3 v3.1.2 h1:ZtDCnhonXSZexk/AYsegNRV1lJGgaNZJuKjJSWKyEqo= -modernc.org/gc/v3 v3.1.2/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY= +modernc.org/gc/v3 v3.1.3 h1:6QAplYyVO+KdPW3pGnqmJDUxtkec8ooEWvks/hhU3lc= +modernc.org/gc/v3 v3.1.3/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY= modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks= modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI= -modernc.org/libc v1.72.3 h1:ZnDF4tXn4NBXFutMMQC4vtbTFSXhhKzR73fv0beZEAU= -modernc.org/libc v1.72.3/go.mod h1:dn0dZNnnn1clLyvRxLxYExxiKRZIRENOfqQ8XEeg4Qs= +modernc.org/libc v1.73.4 h1:+ra4Ui8ngyt8HDcO1FTDPWlkAh6yOdaO2yAoh8MddQA= +modernc.org/libc v1.73.4/go.mod h1:DXZ3eO8qMCNn2SnmTNCiC71nJ9Rcq3PsnpU6Vc4rWK8= modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI= @@ -442,8 +442,8 @@ modernc.org/opt v0.2.0 h1:tGyef5ApycA7FSEOMraay9SaTk5zmbx7Tu+cJs4QKZg= modernc.org/opt v0.2.0/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns= modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w= modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE= -modernc.org/sqlite v1.51.0 h1:aH/MMSoayAIhozZ7uJbVTT9QO/VhzBf0J9tymmmuC/U= -modernc.org/sqlite v1.51.0/go.mod h1:tcNzv5p84E0skkmJn038y+hWJbLQXQqEnQfeh5r2JLM= +modernc.org/sqlite v1.53.0 h1:20WG8N9q4ji/dEqGk4uiI0c6OPjSeLTNYGFCc3+7c1M= +modernc.org/sqlite v1.53.0/go.mod h1:xoEpOIpGrgT48H5iiyt/YXPCZPEzlfmfFwtk8Lklw8s= modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0= modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A= modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= diff --git a/internal/atlas/driver.go b/internal/atlas/driver.go index 1c28b39..a315133 100644 --- a/internal/atlas/driver.go +++ b/internal/atlas/driver.go @@ -30,10 +30,10 @@ type atlasExecutor interface { // openForTest is the seam tests use to bypass real DB setup. Production // calls go straight through to open(). -var openForTest = func(req interfaces.MigrationRequest) ( +var openForTest = func(ctx context.Context, req interfaces.MigrationRequest) ( *sql.DB, *atlmigrate.LocalDir, *sqlRevisionRW, atlmigrate.Driver, func(), error, ) { - return open(req) + return open(ctx, req) } // newAtlasExecutorForTest is the seam tests use to inject a fake executor. @@ -77,7 +77,7 @@ func (d *Driver) Up(ctx context.Context, req interfaces.MigrationRequest) (inter } start := time.Now() - db, dir, rrw, drv, cleanup, err := openForTest(req) + db, dir, rrw, drv, cleanup, err := openForTest(ctx, req) if err != nil { return interfaces.MigrationResult{}, err } @@ -125,7 +125,7 @@ func (d *Driver) Down(ctx context.Context, req interfaces.MigrationRequest) (int } start := time.Now() - db, _, rrw, _, cleanup, err := open(req) + db, _, rrw, _, cleanup, err := open(ctx, req) if err != nil { return interfaces.MigrationResult{}, err } @@ -183,7 +183,7 @@ func (d *Driver) Status(ctx context.Context, req interfaces.MigrationRequest) (i return interfaces.MigrationStatus{}, err } - db, dir, rrw, drv, cleanup, err := openForTest(req) + db, dir, rrw, drv, cleanup, err := openForTest(ctx, req) if err != nil { return interfaces.MigrationStatus{}, err } @@ -237,7 +237,7 @@ func (d *Driver) Goto(ctx context.Context, req interfaces.MigrationRequest, targ } start := time.Now() - db, dir, rrw, drv, cleanup, err := open(req) + db, dir, rrw, drv, cleanup, err := open(ctx, req) if err != nil { return interfaces.MigrationResult{}, err } @@ -286,7 +286,7 @@ func (d *Driver) Goto(ctx context.Context, req interfaces.MigrationRequest, targ } // open builds all the objects needed to talk to the database and the migration dir. -func open(req interfaces.MigrationRequest) ( +func open(ctx context.Context, req interfaces.MigrationRequest) ( db *sql.DB, dir *atlmigrate.LocalDir, rrw *sqlRevisionRW, @@ -294,6 +294,9 @@ func open(req interfaces.MigrationRequest) ( cleanup func(), err error, ) { + if err = ctx.Err(); err != nil { + return + } db, err = sql.Open("pgx", req.DSN) if err != nil { err = fmt.Errorf("atlas: open db: %w", err) @@ -307,14 +310,14 @@ func open(req interfaces.MigrationRequest) ( return } - rrw, err = newSQLRevisionRW(db, "") + rrw, err = newSQLRevisionRW(ctx, db, "") if err != nil { _ = db.Close() err = fmt.Errorf("atlas: revisions table: %w", err) return } - drv, err = atlpg.Open(db) + drv, err = atlpg.Open(operationQuerier{DB: db, ctx: ctx}) if err != nil { _ = db.Close() err = fmt.Errorf("atlas: postgres driver: %w", err) @@ -325,6 +328,27 @@ func open(req interfaces.MigrationRequest) ( return } +// Atlas initializes with Background; bind unbounded library calls to this operation. +type operationQuerier struct { + *sql.DB + ctx context.Context +} + +func (q operationQuerier) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) { + return q.DB.QueryContext(q.boundedContext(ctx), query, args...) +} + +func (q operationQuerier) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) { + return q.DB.ExecContext(q.boundedContext(ctx), query, args...) +} + +func (q operationQuerier) boundedContext(ctx context.Context) context.Context { + if ctx.Done() == nil { + return q.ctx + } + return ctx +} + // openDir opens a LocalDir, auto-generating atlas.sum if it is absent. func openDir(path string) (*atlmigrate.LocalDir, error) { dir, err := atlmigrate.NewLocalDir(path) diff --git a/internal/atlas/driver_recover_test.go b/internal/atlas/driver_recover_test.go index 21b6f8f..7f3ce8a 100644 --- a/internal/atlas/driver_recover_test.go +++ b/internal/atlas/driver_recover_test.go @@ -65,7 +65,7 @@ func TestRunWithRecover_PropagatesNonPanicErrors(t *testing.T) { func TestUp_RecoversAtlasExecutorPanic(t *testing.T) { // Seam out open() to bypass real DB setup. oldOpen := openForTest - openForTest = func(_ interfaces.MigrationRequest) (*sql.DB, *atlmigrate.LocalDir, *sqlRevisionRW, atlmigrate.Driver, func(), error) { + openForTest = func(_ context.Context, _ interfaces.MigrationRequest) (*sql.DB, *atlmigrate.LocalDir, *sqlRevisionRW, atlmigrate.Driver, func(), error) { return nil, nil, &sqlRevisionRW{}, nil, func() {}, nil } defer func() { openForTest = oldOpen }() @@ -93,6 +93,20 @@ func TestUp_RecoversAtlasExecutorPanic(t *testing.T) { } } +func TestAtlasLibraryContextBound(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + q := operationQuerier{ctx: ctx} + if got := q.boundedContext(context.Background()); got != ctx { + t.Fatal("unbounded Atlas initialization does not use operation context") + } + child, stop := context.WithCancel(ctx) + defer stop() + if got := q.boundedContext(child); got != child { + t.Fatal("Atlas child-operation context was replaced") + } +} + // panickingExecutor implements atlasExecutor and intentionally panics on // ExecuteN to simulate the upstream atlas panic seen in workflow#513. type panickingExecutor struct{} diff --git a/internal/atlas/revisions.go b/internal/atlas/revisions.go index b8a272f..c7649f4 100644 --- a/internal/atlas/revisions.go +++ b/internal/atlas/revisions.go @@ -20,12 +20,12 @@ type sqlRevisionRW struct { } // newSQLRevisionRW returns a sqlRevisionRW backed by the given *sql.DB. -func newSQLRevisionRW(db *sql.DB, table string) (*sqlRevisionRW, error) { +func newSQLRevisionRW(ctx context.Context, db *sql.DB, table string) (*sqlRevisionRW, error) { if table == "" { table = defaultRevisionsTable } rw := &sqlRevisionRW{db: db, table: table} - if err := rw.init(context.Background()); err != nil { + if err := rw.init(ctx); err != nil { return nil, err } return rw, nil diff --git a/internal/golangmigrate/driver.go b/internal/golangmigrate/driver.go index af6590d..6918f1f 100644 --- a/internal/golangmigrate/driver.go +++ b/internal/golangmigrate/driver.go @@ -3,17 +3,26 @@ package golangmigrate import ( "context" + "database/sql" + sqldriver "database/sql/driver" "errors" "fmt" "log" + "net/url" "os" "strconv" "strings" + "sync" "time" "github.com/golang-migrate/migrate/v4" - _ "github.com/golang-migrate/migrate/v4/database/pgx/v5" + "github.com/golang-migrate/migrate/v4/database" + pgxmigrate "github.com/golang-migrate/migrate/v4/database/pgx/v5" migratefile "github.com/golang-migrate/migrate/v4/source/file" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgconn/ctxwatch" + "github.com/jackc/pgx/v5/stdlib" "github.com/GoCodeAlone/workflow/interfaces" ) @@ -28,12 +37,13 @@ func New() *Driver { return &Driver{} } func (d *Driver) Name() string { return "golang-migrate" } // Up applies all pending migrations. -func (d *Driver) Up(ctx context.Context, req interfaces.MigrationRequest) (interfaces.MigrationResult, error) { +func (d *Driver) Up(ctx context.Context, req interfaces.MigrationRequest) (result interfaces.MigrationResult, err error) { + defer preserveContextError(ctx, &err) if err := req.Validate(); err != nil { return interfaces.MigrationResult{}, err } start := time.Now() - m, err := newMigrate(req) + m, err := newMigrate(ctx, req) if err != nil { return interfaces.MigrationResult{}, fmt.Errorf("golang-migrate: %w", err) } @@ -72,12 +82,13 @@ func (d *Driver) Up(ctx context.Context, req interfaces.MigrationRequest) (inter } // Down rolls back N migrations (Options.Steps, default 1). -func (d *Driver) Down(ctx context.Context, req interfaces.MigrationRequest) (interfaces.MigrationResult, error) { +func (d *Driver) Down(ctx context.Context, req interfaces.MigrationRequest) (result interfaces.MigrationResult, err error) { + defer preserveContextError(ctx, &err) if err := req.Validate(); err != nil { return interfaces.MigrationResult{}, err } start := time.Now() - m, err := newMigrate(req) + m, err := newMigrate(ctx, req) if err != nil { return interfaces.MigrationResult{}, fmt.Errorf("golang-migrate: %w", err) } @@ -127,11 +138,12 @@ func (d *Driver) Down(ctx context.Context, req interfaces.MigrationRequest) (int } // Status returns the current migration version and pending migrations. -func (d *Driver) Status(_ context.Context, req interfaces.MigrationRequest) (interfaces.MigrationStatus, error) { +func (d *Driver) Status(ctx context.Context, req interfaces.MigrationRequest) (status interfaces.MigrationStatus, err error) { + defer preserveContextError(ctx, &err) if err := req.Validate(); err != nil { return interfaces.MigrationStatus{}, err } - m, err := newMigrate(req) + m, err := newMigrate(ctx, req) if err != nil { return interfaces.MigrationStatus{}, fmt.Errorf("golang-migrate: %w", err) } @@ -159,12 +171,13 @@ func (d *Driver) Status(_ context.Context, req interfaces.MigrationRequest) (int } // Goto migrates to the specified version (up or down). -func (d *Driver) Goto(_ context.Context, req interfaces.MigrationRequest, target string) (interfaces.MigrationResult, error) { +func (d *Driver) Goto(ctx context.Context, req interfaces.MigrationRequest, target string) (result interfaces.MigrationResult, err error) { + defer preserveContextError(ctx, &err) if err := req.Validate(); err != nil { return interfaces.MigrationResult{}, err } start := time.Now() - m, err := newMigrate(req) + m, err := newMigrate(ctx, req) if err != nil { return interfaces.MigrationResult{}, fmt.Errorf("golang-migrate: %w", err) } @@ -201,7 +214,8 @@ type RepairDirtyOptions struct { } // Force sets the recorded migration version without applying migration files. -func (d *Driver) Force(_ context.Context, req interfaces.MigrationRequest, target string, opts ForceOptions) (interfaces.MigrationResult, error) { +func (d *Driver) Force(ctx context.Context, req interfaces.MigrationRequest, target string, opts ForceOptions) (result interfaces.MigrationResult, err error) { + defer preserveContextError(ctx, &err) if err := req.Validate(); err != nil { return interfaces.MigrationResult{}, err } @@ -221,7 +235,7 @@ func (d *Driver) Force(_ context.Context, req interfaces.MigrationRequest, targe } } - m, err := newMigrate(req) + m, err := newMigrate(ctx, req) if err != nil { return interfaces.MigrationResult{}, fmt.Errorf("golang-migrate: %w", err) } @@ -247,7 +261,8 @@ func (d *Driver) Force(_ context.Context, req interfaces.MigrationRequest, targe // RepairDirty verifies a dirty database is at the exact expected version before // forcing metadata. With UpIfClean, a clean database runs normal up instead. -func (d *Driver) RepairDirty(ctx context.Context, req interfaces.MigrationRequest, opts RepairDirtyOptions) (interfaces.MigrationResult, error) { +func (d *Driver) RepairDirty(ctx context.Context, req interfaces.MigrationRequest, opts RepairDirtyOptions) (result interfaces.MigrationResult, err error) { + defer preserveContextError(ctx, &err) if err := req.Validate(); err != nil { return interfaces.MigrationResult{}, err } @@ -282,7 +297,7 @@ func (d *Driver) RepairDirty(ctx context.Context, req interfaces.MigrationReques } } - m, err := newMigrate(req) + m, err := newMigrate(ctx, req) if err != nil { return interfaces.MigrationResult{}, fmt.Errorf("golang-migrate: %w", err) } @@ -373,21 +388,189 @@ func versionExists(dir string, target uint) (bool, error) { } } -// newMigrate creates a migrate.Migrate from a MigrationRequest. -// The DSN is expected to be a postgres:// URL; we rewrite it to pgx5:// for -// the pgx/v5 driver. -func newMigrate(req interfaces.MigrationRequest) (*migrate.Migrate, error) { - dsn := req.DSN - // golang-migrate pgx/v5 driver registers as "pgx5" and expects pgx5:// scheme. - switch { - case strings.HasPrefix(dsn, "postgres://"): - dsn = "pgx5://" + dsn[len("postgres://"):] - case strings.HasPrefix(dsn, "postgresql://"): - dsn = "pgx5://" + dsn[len("postgresql://"):] - } - - sourceURL := "file://" + req.Source.Dir - return migrate.New(sourceURL, dsn) +// golang-migrate's pgx backend uses background contexts. Bind its SQL connection +// to this operation while retaining the backend's migration and locking logic. +func newMigrate(ctx context.Context, req interfaces.MigrationRequest) (*migrationSession, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + u, err := url.Parse(req.DSN) + if err != nil { + return nil, err + } + switch u.Scheme { + case "postgres", "postgresql", "pgx5": + u.Scheme = "postgres" + default: + return nil, fmt.Errorf("golang-migrate: unsupported database scheme %q", u.Scheme) + } + backendConfig, err := postgresConfig(u) + if err != nil { + return nil, err + } + config, err := pgx.ParseConfig(migrate.FilterCustomQuery(u).String()) + if err != nil { + return nil, err + } + config.BuildContextWatcherHandler = func(conn *pgconn.PgConn) ctxwatch.Handler { + return &pgconn.CancelRequestContextWatcherHandler{Conn: conn, DeadlineDelay: time.Second} + } + connector := &contextConnector{Connector: stdlib.GetConnector(*config), ctx: ctx} + db := sql.OpenDB(connector) + db.SetMaxOpenConns(1) + backend, err := pgxmigrate.WithInstance(db, backendConfig) + if err != nil { + _ = db.Close() + // WithInstance can fail after reserving its sql.Conn without releasing it. + connector.closeConnection() + return nil, err + } + m, err := migrate.NewWithDatabaseInstance("file://"+req.Source.Dir, "pgx5", backend) + if err != nil { + _ = backend.Close() + return nil, err + } + return &migrationSession{migrationEngine: m, backend: backend}, nil +} + +type migrationEngine = migrate.Migrate + +type migrationSession struct { + *migrationEngine + backend database.Driver +} + +// Migrate.Version discards dirty=true for a failed down to the nil version. +// Read the same backend metadata while retaining that fail-closed status. +func (m *migrationSession) Version() (uint, bool, error) { + version, dirty, err := m.backend.Version() + if err != nil { + return 0, false, err + } + if version == database.NilVersion { + return 0, dirty, migrate.ErrNilVersion + } + return uint(version), dirty, nil +} + +// Preserve the pgx backend's URL options, including the existing advisory-lock +// identity (URL path, schema, table) so old and new clients serialize together. +func postgresConfig(u *url.URL) (*pgxmigrate.Config, error) { + q := u.Query() + config := &pgxmigrate.Config{DatabaseName: u.Path, MigrationsTable: q.Get("x-migrations-table"), MultiStatementMaxSize: pgxmigrate.DefaultMultiStatementMaxSize} + for key, target := range map[string]*bool{"x-migrations-table-quoted": &config.MigrationsTableQuoted, "x-multi-statement": &config.MultiStatementEnabled} { + if value := q.Get(key); value != "" { + parsed, err := strconv.ParseBool(value) + if err != nil { + return nil, fmt.Errorf("golang-migrate: %s: %w", key, err) + } + *target = parsed + } + } + if config.MigrationsTableQuoted && config.MigrationsTable != "" && (!strings.HasPrefix(config.MigrationsTable, "\"") || !strings.HasSuffix(config.MigrationsTable, "\"")) { + return nil, fmt.Errorf("golang-migrate: x-migrations-table must be quoted") + } + if value := q.Get("x-statement-timeout"); value != "" { + ms, err := strconv.Atoi(value) + if err != nil { + return nil, err + } + config.StatementTimeout = time.Duration(ms) * time.Millisecond + } + if value := q.Get("x-multi-statement-max-size"); value != "" { + size, err := strconv.Atoi(value) + if err != nil { + return nil, err + } + if size > 0 { + config.MultiStatementMaxSize = size + } + } + return config, nil +} + +func preserveContextError(ctx context.Context, err *error) { + if *err != nil && ctx.Err() != nil { + *err = errors.Join(*err, ctx.Err()) + } +} + +type contextConnector struct { + sqldriver.Connector + ctx context.Context + mu sync.Mutex + connection *stdlib.Conn +} + +func (c *contextConnector) Connect(context.Context) (sqldriver.Conn, error) { + conn, err := c.Connector.Connect(c.ctx) + if err != nil { + return nil, err + } + native := conn.(*stdlib.Conn) + c.mu.Lock() + c.connection = native + c.mu.Unlock() + return &contextConnection{Conn: native, ctx: c.ctx}, nil +} + +func (c *contextConnector) closeConnection() { + c.mu.Lock() + defer c.mu.Unlock() + if c.connection != nil { + _ = c.connection.Close() + } +} + +type contextConnection struct { + *stdlib.Conn + ctx context.Context +} + +func (c *contextConnection) ExecContext(ctx context.Context, query string, args []sqldriver.NamedValue) (sqldriver.Result, error) { + operation := c.ctx + // Keep a backend x-statement-timeout if it precedes the caller's deadline. + if deadline, ok := ctx.Deadline(); ok { + var cancel context.CancelFunc + operation, cancel = context.WithDeadline(operation, deadline) + defer cancel() + } + return c.Conn.ExecContext(operation, query, args) +} + +func (c *contextConnection) QueryContext(_ context.Context, query string, args []sqldriver.NamedValue) (sqldriver.Rows, error) { + return c.Conn.QueryContext(c.ctx, query, args) +} + +func (c *contextConnection) Ping(context.Context) error { return c.Conn.Ping(c.ctx) } +func (c *contextConnection) ResetSession(context.Context) error { return c.Conn.ResetSession(c.ctx) } +func (c *contextConnection) Begin() (sqldriver.Tx, error) { + return c.BeginTx(c.ctx, sqldriver.TxOptions{}) +} +func (c *contextConnection) BeginTx(_ context.Context, opts sqldriver.TxOptions) (sqldriver.Tx, error) { + return c.Conn.BeginTx(c.ctx, opts) +} +func (c *contextConnection) Prepare(query string) (sqldriver.Stmt, error) { + return c.PrepareContext(c.ctx, query) +} +func (c *contextConnection) PrepareContext(_ context.Context, query string) (sqldriver.Stmt, error) { + stmt, err := c.Conn.PrepareContext(c.ctx, query) + if err != nil { + return nil, err + } + return &contextStatement{Stmt: stmt, ctx: c.ctx}, nil +} + +type contextStatement struct { + sqldriver.Stmt + ctx context.Context +} + +func (s *contextStatement) ExecContext(_ context.Context, args []sqldriver.NamedValue) (sqldriver.Result, error) { + return s.Stmt.(sqldriver.StmtExecContext).ExecContext(s.ctx, args) +} +func (s *contextStatement) QueryContext(_ context.Context, args []sqldriver.NamedValue) (sqldriver.Rows, error) { + return s.Stmt.(sqldriver.StmtQueryContext).QueryContext(s.ctx, args) } // versionsInRange opens the file source and returns version strings v where diff --git a/internal/golangmigrate/driver_context_test.go b/internal/golangmigrate/driver_context_test.go new file mode 100644 index 0000000..b18f91f --- /dev/null +++ b/internal/golangmigrate/driver_context_test.go @@ -0,0 +1,178 @@ +package golangmigrate_test + +import ( + "context" + "database/sql" + "errors" + "fmt" + "net/url" + "testing" + "time" + + "github.com/GoCodeAlone/workflow-plugin-migrations/internal/golangmigrate" + "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/testharness" + "github.com/GoCodeAlone/workflow/interfaces" +) + +func contextFixture(t *testing.T) (interfaces.MigrationRequest, *sql.DB) { + t.Helper() + if testing.Short() { + t.Skip("PostgreSQL integration") + } + h, err := testharness.New() + if err != nil { + t.Fatalf("PostgreSQL unavailable: %v", err) + } + t.Cleanup(func() { h.Close(t) }) + db, err := sql.Open("pgx", h.DSN()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + dir := t.TempDir() + writeSQL(t, dir, "1_context.up.sql", "CREATE TABLE context_work(id int);") + writeSQL(t, dir, "1_context.down.sql", "DROP TABLE context_work;") + return interfaces.MigrationRequest{DSN: h.DSN(), Source: interfaces.MigrationSource{Dir: dir}}, db +} + +func TestDriverContextPreCancelledMethodsDoNotConnect(t *testing.T) { + for _, operation := range []string{"up", "down", "status", "goto", "force", "repair-dirty"} { + t.Run(operation, func(t *testing.T) { + req, db := contextFixture(t) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + d := golangmigrate.New() + var err error + switch operation { + case "up": + _, err = d.Up(ctx, req) + case "down": + _, err = d.Down(ctx, req) + case "status": + _, err = d.Status(ctx, req) + case "goto": + _, err = d.Goto(ctx, req, "1") + case "force": + _, err = d.Force(ctx, req, "1", golangmigrate.ForceOptions{AllowClean: true}) + case "repair-dirty": + _, err = d.RepairDirty(ctx, req, golangmigrate.RepairDirtyOptions{ExpectedDirtyVersion: "1", ForceVersion: "1", UpIfClean: true}) + } + if !errors.Is(err, context.Canceled) { + t.Errorf("%s error=%v; want context.Canceled", operation, err) + } + var mutated bool + if err = db.QueryRow("SELECT to_regclass('schema_migrations') IS NOT NULL OR to_regclass('context_work') IS NOT NULL").Scan(&mutated); err != nil || mutated { + t.Fatalf("pre-cancel database mutation=%t, %v", mutated, err) + } + }) + } +} + +func TestDriverContextActiveDownAndGotoStopSQL(t *testing.T) { + for _, operation := range []string{"down", "goto"} { + t.Run(operation, func(t *testing.T) { + req, db := contextFixture(t) + u, err := url.Parse(req.DSN) + if err != nil { + t.Fatal(err) + } + application := fmt.Sprintf("native_task15_%d", time.Now().UnixNano()) + q := u.Query() + q.Set("application_name", application) + u.RawQuery = q.Encode() + req.DSN = u.String() + d := golangmigrate.New() + if operation == "down" { + if _, err = d.Up(context.Background(), req); err != nil { + t.Fatal(err) + } + } + query := "SELECT pg_sleep(10); CREATE TABLE native_late(id int);" + suffix := "up" + if operation == "down" { + suffix = "down" + } + writeSQL(t, req.Source.Dir, "1_context."+suffix+".sql", query) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + var e error + if operation == "down" { + _, e = d.Down(ctx, req) + } else { + _, e = d.Goto(ctx, req, "1") + } + done <- e + }() + waitCtx, stop := context.WithTimeout(context.Background(), 5*time.Second) + defer stop() + for { + var active bool + if err = db.QueryRowContext(waitCtx, "SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE application_name=$1 AND query=$2 AND state='active')", application, query).Scan(&active); err != nil { + t.Fatal(err) + } + if active { + break + } + time.Sleep(10 * time.Millisecond) + } + cancel() + select { + case err = <-done: + if !errors.Is(err, context.Canceled) { + t.Errorf("native active cancellation=%v", err) + } + case <-time.After(2 * time.Second): + t.Error("native cancelled SQL still running") + <-done + } + var late, active bool + if err = db.QueryRow("SELECT to_regclass('native_late') IS NOT NULL, EXISTS(SELECT 1 FROM pg_stat_activity WHERE application_name=$1 AND state='active')", application).Scan(&late, &active); err != nil || late || active { + t.Fatalf("native late=%t active=%t, %v", late, active, err) + } + st, e := d.Status(context.Background(), req) + current := "1" + if operation == "down" { + current = "" + } + if e != nil || !st.Dirty || st.Current != current { + t.Fatalf("cancelled version not dirty: %+v, %v", st, e) + } + }) + } +} + +func TestDriverContextPreservesNativeOptions(t *testing.T) { + req, db := contextFixture(t) + u, err := url.Parse(req.DSN) + if err != nil { + t.Fatal(err) + } + var schema string + if err = db.QueryRow("SELECT current_schema()").Scan(&schema); err != nil { + t.Fatal(err) + } + q := u.Query() + q.Set("x-migrations-table", fmt.Sprintf("%q.%q", schema, "context_versions")) + q.Set("x-migrations-table-quoted", "true") + q.Set("x-multi-statement", "true") + q.Set("x-multi-statement-max-size", "2048") + q.Set("x-statement-timeout", "3000") + u.RawQuery = q.Encode() + req.DSN = u.String() + writeSQL(t, req.Source.Dir, "1_context.up.sql", "CREATE TABLE context_work(id int); INSERT INTO context_work VALUES (1);") + d := golangmigrate.New() + result, err := d.Up(context.Background(), req) + if err != nil || len(result.Applied) != 1 { + t.Fatalf("native options up=%+v, %v", result, err) + } + st, err := d.Status(context.Background(), req) + if err != nil || st.Current != "1" || st.Dirty { + t.Fatalf("custom version table status=%+v, %v", st, err) + } + var count int + if err = db.QueryRow("SELECT count(*) FROM context_work").Scan(&count); err != nil || count != 1 { + t.Fatalf("multi-statement count=%d, %v", count, err) + } +} diff --git a/pkg/cli/root.go b/pkg/cli/root.go index 2541401..9e25b1f 100644 --- a/pkg/cli/root.go +++ b/pkg/cli/root.go @@ -14,9 +14,9 @@ import ( "github.com/GoCodeAlone/workflow/interfaces" - atlasdriver "github.com/GoCodeAlone/workflow-plugin-migrations/internal/atlas" "github.com/GoCodeAlone/workflow-plugin-migrations/internal/golangmigrate" - "github.com/GoCodeAlone/workflow-plugin-migrations/internal/goose" + "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/driver" + "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/runner" ) // cliProvider implements sdk.CLIProvider by dispatching to the Cobra root. @@ -28,6 +28,10 @@ func NewCLIProvider() *cliProvider { return &cliProvider{} } // RunCLI implements sdk.CLIProvider. func (c *cliProvider) RunCLI(args []string) int { root := NewRoot() + // The SDK includes the dynamic command name; Cobra expects only its args. + if len(args) > 0 && args[0] == root.Name() { + args = args[1:] + } root.SetArgs(args) if err := root.Execute(); err != nil { fmt.Fprintln(os.Stderr, err) @@ -38,16 +42,21 @@ func (c *cliProvider) RunCLI(args []string) int { // NewRoot builds the Cobra root command for the db-migrate CLI. func NewRoot() *cobra.Command { + return newRoot(driver.NewDefaultRegistry()) +} + +func newRoot(registry *driver.Registry) *cobra.Command { root := &cobra.Command{ - Use: "db-migrate", - Short: "Database migration commands", - Long: "Run, inspect, and test database migrations via golang-migrate, goose, or atlas.", + Use: "db-migrate", + SilenceUsage: true, + Short: "Database migration commands", + Long: "Run, inspect, and test database migrations via golang-migrate, goose, or atlas.", } root.AddCommand( - newUpCmd(), - newDownCmd(), - newStatusCmd(), - newGotoCmd(), + newUpCmd(registry), + newDownCmd(registry), + newStatusCmd(registry), + newGotoCmd(registry), newForceCmd(), newRepairDirtyCmd(), newValidateUpgradeCmd(), @@ -66,43 +75,56 @@ func sharedFlags(cmd *cobra.Command) { _ = cmd.MarkFlagRequired("source-dir") } -// buildDriverAndRequestForTest is the package-level seam that lets tests stub -// out driver construction. Production calls go straight through to -// buildDriverAndRequest. -var buildDriverAndRequestForTest = buildDriverAndRequest +func runnerFlags(cmd *cobra.Command) { + sharedFlags(cmd) + cmd.Flags().Duration("timeout", 0, "Operation timeout (0 uses the caller deadline)") +} // buildDriverAndRequest resolves the driver and constructs a MigrationRequest from flags. func buildDriverAndRequest(cmd *cobra.Command) (interfaces.MigrationDriver, interfaces.MigrationRequest, error) { + req, err := requestFromFlags(cmd) + if err != nil { + return nil, req, err + } driverName, _ := cmd.Flags().GetString("driver") + if driverName == "" { + driverName = "golang-migrate" + } + d, err := driver.NewDefaultRegistry().Get(driverName) + return d, req, err +} + +func buildRunnerAndRequest(cmd *cobra.Command, registry *driver.Registry) (driver.Driver, driver.Request, error) { + req, err := requestFromFlags(cmd) + if err != nil { + return nil, req, err + } + name, _ := cmd.Flags().GetString("driver") + // Cobra owns presentation and error reporting; the runner still supplies + // validation, cancellation, and redacted errors for every ordinary operation. + r, err := runner.New(registry, name, io.Discard, io.Discard) + return r, req, err +} + +func requestFromFlags(cmd *cobra.Command) (driver.Request, error) { sourceDir, _ := cmd.Flags().GetString("source-dir") dsn, _ := cmd.Flags().GetString("dsn") if dsn == "" { dsn = os.Getenv("DATABASE_URL") } if dsn == "" { - return nil, interfaces.MigrationRequest{}, fmt.Errorf("no DSN: set --dsn or DATABASE_URL env var") - } - - var d interfaces.MigrationDriver - switch driverName { - case "golang-migrate", "": - d = golangmigrate.New() - case "goose": - d = goose.New() - case "atlas": - d = atlasdriver.New() - default: - return nil, interfaces.MigrationRequest{}, fmt.Errorf("unknown driver %q (supported: golang-migrate, goose, atlas)", driverName) + return driver.Request{}, fmt.Errorf("no DSN: set --dsn or DATABASE_URL env var") } - + timeout, _ := cmd.Flags().GetDuration("timeout") req := interfaces.MigrationRequest{ - DSN: dsn, - Source: interfaces.MigrationSource{Dir: sourceDir}, + DSN: dsn, + Source: interfaces.MigrationSource{Dir: sourceDir}, + Options: interfaces.MigrationOptions{Timeout: timeout}, } - return d, req, nil + return req, nil } -func newUpCmd() *cobra.Command { +func newUpCmd(registry *driver.Registry) *cobra.Command { cmd := &cobra.Command{ Use: "up", Short: "Apply all pending migrations", @@ -116,113 +138,115 @@ intent explicit and, crucially, must be accepted by cobra so deploy CMDs that pass it succeed.`, RunE: func(cmd *cobra.Command, _ []string) error { upIfClean, _ := cmd.Flags().GetBool("up-if-clean") - d, req, err := buildDriverAndRequestForTest(cmd) + d, req, err := buildRunnerAndRequest(cmd, registry) if err != nil { return err } - result, err := d.Up(context.Background(), req) + result, err := d.Up(cmd.Context(), req) if err != nil { return fmt.Errorf("migrate up: %w", err) } if len(result.Applied) == 0 { if upIfClean { - fmt.Println("up-if-clean: no pending migrations; database is clean.") + _, err = fmt.Fprintln(cmd.OutOrStdout(), "up-if-clean: no pending migrations; database is clean.") } else { - fmt.Println("No pending migrations.") + _, err = fmt.Fprintln(cmd.OutOrStdout(), "No pending migrations.") } - return nil + return err } - fmt.Printf("Applied %d migration(s): %v\n", len(result.Applied), result.Applied) - return nil + _, err = fmt.Fprintf(cmd.OutOrStdout(), "Applied %d migration(s): %v\n", len(result.Applied), result.Applied) + return err }, } - sharedFlags(cmd) + runnerFlags(cmd) cmd.Flags().Bool("up-if-clean", false, "Idempotent up: exit 0 when no migrations are pending. Required for deploy CMDs that may re-run against an already-current database.") return cmd } -func newDownCmd() *cobra.Command { +func newDownCmd(registry *driver.Registry) *cobra.Command { cmd := &cobra.Command{ Use: "down", Short: "Roll back N migrations (default: 1)", RunE: func(cmd *cobra.Command, _ []string) error { - d, req, err := buildDriverAndRequest(cmd) + d, req, err := buildRunnerAndRequest(cmd, registry) if err != nil { return err } steps, _ := cmd.Flags().GetInt("steps") req.Options.Steps = steps - result, err := d.Down(context.Background(), req) + result, err := d.Down(cmd.Context(), req) if err != nil { return fmt.Errorf("migrate down: %w", err) } - fmt.Printf("Rolled back %d migration(s): %v\n", len(result.Applied), result.Applied) - return nil + _, err = fmt.Fprintf(cmd.OutOrStdout(), "Rolled back %d migration(s): %v\n", len(result.Applied), result.Applied) + return err }, } - sharedFlags(cmd) + runnerFlags(cmd) cmd.Flags().Int("steps", 1, "Number of migrations to roll back") return cmd } -func newStatusCmd() *cobra.Command { +func newStatusCmd(registry *driver.Registry) *cobra.Command { cmd := &cobra.Command{ Use: "status", Short: "Show current migration status", RunE: func(cmd *cobra.Command, _ []string) error { - d, req, err := buildDriverAndRequest(cmd) + d, req, err := buildRunnerAndRequest(cmd, registry) if err != nil { return err } - st, err := d.Status(context.Background(), req) + st, err := d.Status(cmd.Context(), req) if err != nil { return fmt.Errorf("migrate status: %w", err) } - writeStatus(cmd.OutOrStdout(), st) - return nil + return writeStatus(cmd.OutOrStdout(), st) }, } - sharedFlags(cmd) + runnerFlags(cmd) return cmd } -func writeStatus(out io.Writer, st interfaces.MigrationStatus) { +func writeStatus(out io.Writer, st interfaces.MigrationStatus) error { + var text string if st.Current == "" { - fmt.Fprintln(out, "No migrations applied.") + text = "No migrations applied.\n" } else { - fmt.Fprintf(out, "Current: %s\n", st.Current) + text = fmt.Sprintf("Current: %s\n", st.Current) } if len(st.Pending) > 0 { - fmt.Fprintf(out, "Pending: %v\n", st.Pending) + text += fmt.Sprintf("Pending: %v\n", st.Pending) } else { - fmt.Fprintln(out, "No pending migrations.") + text += "No pending migrations.\n" } - fmt.Fprintf(out, "Dirty: %t\n", st.Dirty) + text += fmt.Sprintf("Dirty: %t\n", st.Dirty) if st.Dirty { - fmt.Fprintln(out, "WARNING: database is in dirty state!") + text += "WARNING: database is in dirty state!\n" } + _, err := io.WriteString(out, text) + return err } -func newGotoCmd() *cobra.Command { +func newGotoCmd(registry *driver.Registry) *cobra.Command { cmd := &cobra.Command{ Use: "goto ", Short: "Migrate to a specific version (up or down)", Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { - d, req, err := buildDriverAndRequest(cmd) + d, req, err := buildRunnerAndRequest(cmd, registry) if err != nil { return err } target := args[0] - result, err := d.Goto(context.Background(), req, target) + result, err := d.Goto(cmd.Context(), req, target) if err != nil { return fmt.Errorf("migrate goto %s: %w", target, err) } - fmt.Printf("Migrated to %s (%d steps): %v\n", target, len(result.Applied), result.Applied) - return nil + _, err = fmt.Fprintf(cmd.OutOrStdout(), "Migrated to %s (%d steps): %v\n", target, len(result.Applied), result.Applied) + return err }, } - sharedFlags(cmd) + runnerFlags(cmd) return cmd } diff --git a/pkg/cli/root_test.go b/pkg/cli/root_test.go index 79a0b60..233ed8c 100644 --- a/pkg/cli/root_test.go +++ b/pkg/cli/root_test.go @@ -4,14 +4,15 @@ import ( "bytes" "context" "database/sql" + "errors" "fmt" "os" "path/filepath" + "reflect" "strings" "testing" - "github.com/spf13/cobra" - + "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/driver" "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/testharness" "github.com/GoCodeAlone/workflow/interfaces" _ "github.com/jackc/pgx/v5/stdlib" @@ -29,6 +30,22 @@ func TestRootIncludesForceCommand(t *testing.T) { } } +func TestTimeoutFlagOnlyOnRunnerCommands(t *testing.T) { + root := NewRoot() + for _, name := range []string{"up", "down", "status", "goto", "force", "repair-dirty", "validate-upgrade", "test"} { + t.Run(name, func(t *testing.T) { + cmd, _, err := root.Find([]string{name}) + if err != nil { + t.Fatal(err) + } + want := name == "up" || name == "down" || name == "status" || name == "goto" + if got := cmd.Flags().Lookup("timeout") != nil; got != want { + t.Fatalf("%s advertises timeout=%t; runner-backed=%t", name, got, want) + } + }) + } +} + func TestRootIncludesRepairDirtyCommand(t *testing.T) { root := NewRoot() @@ -48,7 +65,7 @@ func TestRootIncludesRepairDirtyCommand(t *testing.T) { func TestWriteStatusIncludesExplicitDirtyFlag(t *testing.T) { var out bytes.Buffer - writeStatus(&out, interfaces.MigrationStatus{ + _ = writeStatus(&out, interfaces.MigrationStatus{ Current: "202604270001", Dirty: false, }) @@ -59,7 +76,7 @@ func TestWriteStatusIncludesExplicitDirtyFlag(t *testing.T) { } out.Reset() - writeStatus(&out, interfaces.MigrationStatus{ + _ = writeStatus(&out, interfaces.MigrationStatus{ Current: "202604270001", Dirty: true, }) @@ -389,7 +406,7 @@ func (f *fakeUpDriver) Goto(_ context.Context, _ interfaces.MigrationRequest, _ } func TestUpCmd_AcceptsUpIfCleanFlag(t *testing.T) { - cmd := newUpCmd() + cmd := newUpCmd(driver.NewDefaultRegistry()) flag := cmd.Flags().Lookup("up-if-clean") if flag == nil { t.Fatal("up command missing --up-if-clean flag") @@ -400,33 +417,96 @@ func TestUpCmd_AcceptsUpIfCleanFlag(t *testing.T) { } func TestUpCmd_UpIfCleanIsNoopWhenAlreadyClean(t *testing.T) { - // Override the driver-construction seam so the test doesn't need a real DB. - old := buildDriverAndRequestForTest - buildDriverAndRequestForTest = func(cmd *cobra.Command) (interfaces.MigrationDriver, interfaces.MigrationRequest, error) { - return &fakeUpDriver{appliedCount: 0}, interfaces.MigrationRequest{}, nil - } - defer func() { buildDriverAndRequestForTest = old }() - - cmd := newUpCmd() - cmd.SetArgs([]string{"--up-if-clean", "--source-dir", t.TempDir()}) + reg := driver.NewRegistry() + reg.MustRegister(&fakeUpDriver{appliedCount: 0}) + cmd := newUpCmd(reg) + var out bytes.Buffer + cmd.SetOut(&out) + cmd.SetArgs([]string{"--driver", "fake", "--dsn", "postgres://local/db", "--up-if-clean", "--source-dir", t.TempDir()}) if err := cmd.Execute(); err != nil { t.Fatalf("up --up-if-clean against clean DB: got error %v, want nil", err) } + if !strings.Contains(out.String(), "up-if-clean: no pending migrations") { + t.Fatalf("injected output=%q", out.String()) + } } func TestUpCmd_UpIfCleanAppliesWhenPendingMigrationsExist(t *testing.T) { - // When migrations ARE pending, --up-if-clean should still apply them. - old := buildDriverAndRequestForTest - buildDriverAndRequestForTest = func(cmd *cobra.Command) (interfaces.MigrationDriver, interfaces.MigrationRequest, error) { - return &fakeUpDriver{appliedCount: 2}, interfaces.MigrationRequest{}, nil - } - defer func() { buildDriverAndRequestForTest = old }() - - cmd := newUpCmd() - cmd.SetArgs([]string{"--up-if-clean", "--source-dir", t.TempDir()}) + reg := driver.NewRegistry() + reg.MustRegister(&fakeUpDriver{appliedCount: 2}) + cmd := newUpCmd(reg) + var out bytes.Buffer + cmd.SetOut(&out) + cmd.SetArgs([]string{"--driver", "fake", "--dsn", "postgres://local/db", "--up-if-clean", "--source-dir", t.TempDir()}) if err := cmd.Execute(); err != nil { t.Fatalf("up --up-if-clean with pending migrations: got error %v, want nil", err) } + if !strings.Contains(out.String(), "Applied 2 migration(s)") { + t.Fatalf("injected output=%q", out.String()) + } +} + +func TestRootRunnerHonorsExecuteContext(t *testing.T) { + for _, args := range [][]string{{"up"}, {"down"}, {"status"}, {"goto", "1"}} { + t.Run(args[0], func(t *testing.T) { + reg := driver.NewRegistry() + reg.MustRegister(&fakeUpDriver{appliedCount: 1}) + root := newRoot(reg) + var out, diagnostics bytes.Buffer + root.SetOut(&out) + root.SetErr(&diagnostics) + root.SetArgs(append(args, "--driver", "fake", "--dsn", "postgres://local/db", "--source-dir", t.TempDir())) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := root.ExecuteContext(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("ExecuteContext=%v", err) + } + if out.Len() != 0 { + t.Fatalf("cancelled command output=%q", out.String()) + } + }) + } +} + +func TestRootRunnerRedactsErrorsAndUsesCommandOutput(t *testing.T) { + for _, args := range [][]string{{"up"}, {"down"}, {"status"}, {"goto", "1"}} { + t.Run(args[0], func(t *testing.T) { + reg := driver.NewRegistry() + reg.MustRegister(&fakeUpDriver{appliedCount: 1}) + root := newRoot(reg) + var out bytes.Buffer + root.SetOut(&out) + root.SetArgs(append(args, "--driver", "fake", "--dsn", "postgres://local/db", "--source-dir", t.TempDir())) + if err := root.Execute(); err != nil { + t.Fatal(err) + } + if out.Len() == 0 { + t.Fatal("command bypassed injected output") + } + }) + } + root := NewRoot() + var diagnostics bytes.Buffer + root.SetErr(&diagnostics) + root.SetArgs([]string{"up", "--dsn", "postgres://user:task15-private@localhost/db?connect_timeout=invalid", "--source-dir", t.TempDir()}) + err := root.Execute() + if err == nil { + t.Fatal("invalid connection accepted") + } + if strings.Contains(err.Error(), "task15-private") || strings.Contains(diagnostics.String(), "task15-private") { + t.Fatal("CLI leaked DSN password") + } +} + +func TestCLIProviderAcceptsSDKCommandPrefix(t *testing.T) { + args := []string{"db-migrate", "--help"} + original := append([]string(nil), args...) + if code := NewCLIProvider().RunCLI(args); code != 0 { + t.Fatalf("SDK CLI command returned %d; want help success", code) + } + if !reflect.DeepEqual(args, original) { + t.Fatal("CLI provider mutated caller arguments") + } } func writeCLISQL(t *testing.T, dir, name, sql string) { diff --git a/pkg/driver/registry.go b/pkg/driver/registry.go index 4d89e15..ff90614 100644 --- a/pkg/driver/registry.go +++ b/pkg/driver/registry.go @@ -4,6 +4,10 @@ import ( "fmt" "sort" "sync" + + "github.com/GoCodeAlone/workflow-plugin-migrations/internal/atlas" + "github.com/GoCodeAlone/workflow-plugin-migrations/internal/golangmigrate" + "github.com/GoCodeAlone/workflow-plugin-migrations/internal/goose" ) // Registry is a thread-safe in-process registry of named MigrationDrivers. @@ -17,6 +21,15 @@ func NewRegistry() *Registry { return &Registry{drivers: make(map[string]Driver)} } +// NewDefaultRegistry returns a fresh registry of the built-in native drivers. +func NewDefaultRegistry() *Registry { + r := NewRegistry() + r.MustRegister(golangmigrate.New()) + r.MustRegister(goose.New()) + r.MustRegister(atlas.New()) + return r +} + // Register adds a driver to the registry. Returns an error if a driver with // the same name is already registered. func (r *Registry) Register(d Driver) error { diff --git a/pkg/runner/runner.go b/pkg/runner/runner.go new file mode 100644 index 0000000..b1b86ec --- /dev/null +++ b/pkg/runner/runner.go @@ -0,0 +1,164 @@ +// Package runner exposes in-process migrations without CLI or process-global state. +package runner + +import ( + "context" + "errors" + "fmt" + "io" + "net/url" + "strings" + + "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/driver" + "github.com/GoCodeAlone/workflow/interfaces" + "github.com/jackc/pgx/v5" +) + +// New binds a registered driver and per-instance output. Empty name selects +// golang-migrate; nil writers discard output. Requests supply the DSN explicitly. +// Operations preserve caller cancellation and apply Options.Timeout when positive. +func New(registry *driver.Registry, name string, stdout, stderr io.Writer) (driver.Driver, error) { + if registry == nil { + return nil, fmt.Errorf("migration runner: registry is required") + } + if name == "" { + name = "golang-migrate" + } + d, err := registry.Get(name) + if err != nil { + return nil, err + } + if stdout == nil { + stdout = io.Discard + } + if stderr == nil { + stderr = io.Discard + } + return &runner{inner: d, out: stdout, diagnostics: stderr}, nil +} + +type runner struct { + inner driver.Driver + out, diagnostics io.Writer +} + +func (r *runner) Name() string { return r.inner.Name() } + +func operationContext(ctx context.Context, req driver.Request) (context.Context, context.CancelFunc, error) { + if err := ctx.Err(); err != nil { + return ctx, func() {}, err + } + if err := req.Validate(); err != nil { + return ctx, func() {}, err + } + if req.Options.Timeout < 0 { + return ctx, func() {}, fmt.Errorf("%w: negative migration timeout", interfaces.ErrValidation) + } + if req.Options.Timeout > 0 { + ctx, cancel := context.WithTimeout(ctx, req.Options.Timeout) + return ctx, cancel, nil + } + return ctx, func() {}, nil +} + +func (r *runner) Up(ctx context.Context, req driver.Request) (driver.Result, error) { + return r.run(ctx, req, "up", func(ctx context.Context) (driver.Result, error) { return r.inner.Up(ctx, req) }) +} +func (r *runner) Down(ctx context.Context, req driver.Request) (driver.Result, error) { + return r.run(ctx, req, "down", func(ctx context.Context) (driver.Result, error) { return r.inner.Down(ctx, req) }) +} +func (r *runner) Goto(ctx context.Context, req driver.Request, target string) (driver.Result, error) { + return r.run(ctx, req, "goto", func(ctx context.Context) (driver.Result, error) { return r.inner.Goto(ctx, req, target) }) +} + +func (r *runner) run(ctx context.Context, req driver.Request, op string, fn func(context.Context) (driver.Result, error)) (driver.Result, error) { + ctx, cancel, err := operationContext(ctx, req) + defer cancel() + if err != nil { + return driver.Result{}, r.failure(req, err) + } + result, err := fn(ctx) + if ctx.Err() != nil { + err = errors.Join(err, ctx.Err()) + } + if err != nil { + return result, r.failure(req, err) + } + _, err = io.WriteString(r.out, sanitize(req, fmt.Sprintf("%s: %d migration(s): %v\n", op, len(result.Applied), result.Applied))) + if err != nil { + return result, r.failure(req, err) + } + return result, nil +} + +func (r *runner) Status(ctx context.Context, req driver.Request) (driver.Status, error) { + ctx, cancel, err := operationContext(ctx, req) + defer cancel() + if err != nil { + return driver.Status{}, r.failure(req, err) + } + st, err := r.inner.Status(ctx, req) + if ctx.Err() != nil { + err = errors.Join(err, ctx.Err()) + } + if err != nil { + return st, r.failure(req, err) + } + _, err = io.WriteString(r.out, sanitize(req, fmt.Sprintf("Current: %s\nPending: %v\nDirty: %t\n", st.Current, st.Pending, st.Dirty))) + if err != nil { + return st, r.failure(req, err) + } + return st, nil +} + +type diagnosticError struct { + message string + cause error +} + +func (e *diagnosticError) Error() string { return e.message } +func (e *diagnosticError) Unwrap() error { return e.cause } + +func (r *runner) failure(req driver.Request, cause error) error { + message := sanitize(req, "migration runner: "+cause.Error()) + if _, err := io.WriteString(r.diagnostics, message+"\n"); err != nil { + cause = errors.Join(cause, err) + } + return &diagnosticError{message: message, cause: cause} +} + +// Redact before truncating so a credential crossing the limit cannot leak. +// Diagnostics, including their newline, have a fixed 4 KiB budget. +func sanitize(req driver.Request, message string) string { + if req.DSN != "" { + message = strings.ReplaceAll(message, req.DSN, "[redacted DSN]") + for _, scheme := range []string{"postgres://", "postgresql://", "pgx5://"} { + if i := strings.Index(req.DSN, "://"); i >= 0 { + message = strings.ReplaceAll(message, scheme+req.DSN[i+3:], "[redacted DSN]") + } + } + var passwords []string + if u, err := url.Parse(req.DSN); err == nil { + if u.User != nil { + if p, ok := u.User.Password(); ok { + passwords = append(passwords, p) + } + } + passwords = append(passwords, u.Query()["password"]...) + } + if config, err := pgx.ParseConfig(req.DSN); err == nil { + passwords = append(passwords, config.Password) + } + for _, password := range passwords { + if password != "" { + for _, form := range []string{password, url.QueryEscape(password), url.PathEscape(password)} { + message = strings.ReplaceAll(message, form, "[redacted]") + } + } + } + } + if len(message) > 4095 { + message = message[:4092] + "..." + } + return message +} diff --git a/pkg/runner/runner_test.go b/pkg/runner/runner_test.go new file mode 100644 index 0000000..c9891e4 --- /dev/null +++ b/pkg/runner/runner_test.go @@ -0,0 +1,542 @@ +package runner_test + +import ( + "bytes" + "context" + "database/sql" + "errors" + "fmt" + "io" + "net" + "net/url" + "os" + "os/exec" + "path/filepath" + "reflect" + "strings" + "testing" + "time" + + "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/driver" + "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/runner" + "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/testharness" + "github.com/GoCodeAlone/workflow/interfaces" + "github.com/golang-migrate/migrate/v4/database" +) + +type recordingDriver struct { + ctx context.Context + req driver.Request + target string + op string + err error + wait bool +} + +func (d *recordingDriver) Name() string { return "recording" } +func (d *recordingDriver) record(ctx context.Context, req driver.Request, op, target string) error { + d.ctx, d.req, d.op, d.target = ctx, req, op, target + if d.wait { + <-ctx.Done() + return ctx.Err() + } + return d.err +} +func (d *recordingDriver) Up(ctx context.Context, req driver.Request) (driver.Result, error) { + return driver.Result{Applied: []string{"1", "2"}}, d.record(ctx, req, "up", "") +} +func (d *recordingDriver) Down(ctx context.Context, req driver.Request) (driver.Result, error) { + return driver.Result{Applied: []string{"2"}}, d.record(ctx, req, "down", "") +} +func (d *recordingDriver) Goto(ctx context.Context, req driver.Request, target string) (driver.Result, error) { + return driver.Result{Applied: []string{target}}, d.record(ctx, req, "goto", target) +} +func (d *recordingDriver) Status(ctx context.Context, req driver.Request) (driver.Status, error) { + return driver.Status{Current: "2", Pending: []string{"3"}, Dirty: true}, d.record(ctx, req, "status", "") +} + +func TestPublicRunnerRegistryAndForwarding(t *testing.T) { + reg := driver.NewDefaultRegistry() + if got := reg.Names(); !reflect.DeepEqual(got, []string{"atlas", "golang-migrate", "goose"}) { + t.Fatalf("builtins = %v", got) + } + if _, err := runner.New(reg, "unsupported", nil, nil); err == nil { + t.Fatal("unsupported driver accepted") + } + if _, err := runner.New(nil, "recording", nil, nil); err == nil { + t.Fatal("nil registry accepted") + } + if d, err := runner.New(reg, "", nil, nil); err != nil || d.Name() != "golang-migrate" { + t.Fatalf("default driver = %v, %v", d, err) + } + args, env := append([]string(nil), os.Args...), os.Environ() + for _, op := range []string{"up", "down", "status", "goto"} { + t.Run(op, func(t *testing.T) { + native := &recordingDriver{} + local := driver.NewRegistry() + local.MustRegister(native) + var out, diagnostics bytes.Buffer + r, err := runner.New(local, "recording", &out, &diagnostics) + if err != nil { + t.Fatal(err) + } + ctx := context.WithValue(context.Background(), contextKey{}, "consumer") + req := driver.Request{DSN: "postgres://user:private@example.invalid/db", Source: driver.Source{Dir: t.TempDir(), SchemaName: "tenant"}, Options: driver.Options{Steps: 2, Version: "2"}} + switch op { + case "up": + _, err = r.Up(ctx, req) + case "down": + _, err = r.Down(ctx, req) + case "status": + var st driver.Status + st, err = r.Status(ctx, req) + if !st.Dirty || st.Current != "2" { + t.Fatalf("status = %+v", st) + } + case "goto": + _, err = r.Goto(ctx, req, "2") + if native.target != "2" { + t.Fatal("target lost") + } + } + if err != nil { + t.Fatal(err) + } + if native.op != op || !reflect.DeepEqual(native.req, req) || native.ctx.Value(contextKey{}) != "consumer" { + t.Fatalf("forwarded %s: %+v", native.op, native.req) + } + if out.Len() == 0 || diagnostics.Len() != 0 { + t.Fatalf("injected IO: out=%q err=%q", out.String(), diagnostics.String()) + } + }) + } + if !reflect.DeepEqual(os.Args, args) || !reflect.DeepEqual(os.Environ(), env) { + t.Fatal("runner mutated process arguments/environment") + } +} + +type contextKey struct{} + +func TestRunnerCancellationTimeoutAndValidation(t *testing.T) { + for _, op := range []string{"up", "down", "status", "goto"} { + t.Run(op, func(t *testing.T) { + native := &recordingDriver{} + reg := driver.NewRegistry() + reg.MustRegister(native) + r, _ := runner.New(reg, native.Name(), nil, nil) + req := driver.Request{DSN: "postgres://local/db", Source: driver.Source{Dir: t.TempDir()}} + ctx, cancel := context.WithCancel(context.Background()) + cancel() + var err error + switch op { + case "up": + _, err = r.Up(ctx, req) + case "down": + _, err = r.Down(ctx, req) + case "status": + _, err = r.Status(ctx, req) + case "goto": + _, err = r.Goto(ctx, req, "1") + } + if !errors.Is(err, context.Canceled) || native.ctx != nil { + t.Fatalf("pre-cancel = %v; driver called=%t", err, native.ctx != nil) + } + if _, err := r.Up(context.Background(), driver.Request{}); !errors.Is(err, interfaces.ErrValidation) { + t.Fatalf("validation = %v", err) + } + native.wait = true + req.Options.Timeout = 20 * time.Millisecond + start := time.Now() + _, err = r.Up(context.Background(), req) + if !errors.Is(err, context.DeadlineExceeded) || time.Since(start) > time.Second { + t.Fatalf("timeout = %v", err) + } + parent, stop := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer stop() + req.Options.Timeout = time.Hour + _, err = r.Up(parent, req) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("parent deadline = %v", err) + } + }) + } +} + +func TestRunnerAtlasInitializationTimeout(t *testing.T) { + for _, operation := range []string{"up", "down", "status", "goto"} { + t.Run(operation, func(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + release, stopped := make(chan struct{}), make(chan struct{}) + go func() { + defer close(stopped) + conn, err := listener.Accept() + if err != nil { + return + } + defer func() { _ = conn.Close() }() + <-release + }() + t.Cleanup(func() { + close(release) + _ = listener.Close() + <-stopped + }) + r, err := runner.New(driver.NewDefaultRegistry(), "atlas", nil, nil) + if err != nil { + t.Fatal(err) + } + req := driver.Request{ + DSN: "postgres://test:test@" + listener.Addr().String() + "/fixture?sslmode=disable&connect_timeout=3", + Source: driver.Source{Dir: t.TempDir()}, + Options: driver.Options{Timeout: 100 * time.Millisecond}, + } + start := time.Now() + switch operation { + case "up": + _, err = r.Up(context.Background(), req) + case "down": + _, err = r.Down(context.Background(), req) + case "status": + _, err = r.Status(context.Background(), req) + case "goto": + _, err = r.Goto(context.Background(), req, "1") + } + if elapsed := time.Since(start); !errors.Is(err, context.DeadlineExceeded) || elapsed > time.Second { + t.Fatalf("%s stalled initialization: elapsed=%s error=%v", operation, elapsed, err) + } + }) + } +} + +func TestRunnerErrorsAreBoundedRedactedAndPreserveIdentity(t *testing.T) { + secret := "runner-p@ss/word" + dsn := "postgres://runner:" + url.QueryEscape(secret) + "@localhost/db?password=" + url.QueryEscape(secret) + sentinel := errors.New("driver denied") + native := &recordingDriver{err: fmt.Errorf("%w: %s %s %s %s", sentinel, dsn, strings.Replace(dsn, "postgres://", "pgx5://", 1), secret, strings.Repeat("x", 12000))} + reg := driver.NewRegistry() + reg.MustRegister(native) + var out, diagnostics bytes.Buffer + r, _ := runner.New(reg, native.Name(), &out, &diagnostics) + _, err := r.Up(context.Background(), driver.Request{DSN: dsn, Source: driver.Source{Dir: t.TempDir()}}) + if !errors.Is(err, sentinel) { + t.Fatalf("lost identity: %v", err) + } + for _, text := range []string{err.Error(), diagnostics.String()} { + if len(text) > 4096 { + t.Fatalf("unbounded diagnostic: %d bytes", len(text)) + } + for _, forbidden := range []string{dsn, secret, url.QueryEscape(secret), "pgx5://runner:"} { + if strings.Contains(text, forbidden) { + t.Fatal("diagnostic leaked credentials") + } + } + } + if out.Len() != 0 || diagnostics.Len() == 0 { + t.Fatal("error IO was not isolated") + } +} + +func postgres(t *testing.T) (string, *sql.DB) { + t.Helper() + if testing.Short() { + t.Skip("PostgreSQL integration") + } + h, err := testharness.New() + if err != nil { + t.Fatalf("PostgreSQL fixture unavailable: %v", err) + } + t.Cleanup(func() { h.Close(t) }) + db, err := sql.Open("pgx", h.DSN()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + return h.DSN(), db +} + +func migration(t *testing.T, dir, version, up, down string) { + t.Helper() + for suffix, body := range map[string]string{"up": up, "down": down} { + if err := os.WriteFile(filepath.Join(dir, version+"_fixture."+suffix+".sql"), []byte(body), 0600); err != nil { + t.Fatal(err) + } + } +} + +func nativeRunner(t *testing.T) driver.Driver { + t.Helper() + r, err := runner.New(driver.NewDefaultRegistry(), "golang-migrate", io.Discard, io.Discard) + if err != nil { + t.Fatal(err) + } + return r +} + +func TestPostgresOrderedRepeatTargetDown(t *testing.T) { + dsn, db := postgres(t) + dir := t.TempDir() + versions := []string{"20240101000001", "20240201000001", "20240301000001"} + for i, v := range versions { + migration(t, dir, v, fmt.Sprintf("CREATE TABLE ordered_%d(id int);", i), fmt.Sprintf("DROP TABLE ordered_%d;", i)) + } + r := nativeRunner(t) + req := driver.Request{DSN: dsn, Source: driver.Source{Dir: dir}} + st, err := r.Status(context.Background(), req) + if err != nil || !reflect.DeepEqual(st.Pending, versions) { + t.Fatalf("initial status = %+v, %v", st, err) + } + got, err := r.Up(context.Background(), req) + if err != nil || !reflect.DeepEqual(got.Applied, versions) { + t.Fatalf("ordered up = %+v, %v", got, err) + } + st, err = r.Status(context.Background(), req) + if err != nil || st.Dirty || st.Current != versions[2] || len(st.Pending) != 0 { + t.Fatalf("clean status = %+v, %v", st, err) + } + got, err = r.Up(context.Background(), req) + if err != nil || len(got.Applied) != 0 { + t.Fatalf("repeat = %+v, %v", got, err) + } + if _, err = r.Goto(context.Background(), req, versions[1]); err != nil { + t.Fatal(err) + } + req.Options.Steps = 1 + got, err = r.Down(context.Background(), req) + if err != nil || !reflect.DeepEqual(got.Applied, []string{versions[1]}) { + t.Fatalf("down = %+v, %v", got, err) + } + var count int + if err = db.QueryRow("SELECT count(*) FROM pg_tables WHERE schemaname=current_schema() AND tablename LIKE 'ordered_%'").Scan(&count); err != nil || count != 1 { + t.Fatalf("catalog count=%d, %v", count, err) + } +} + +func TestPostgresPartialFailureStaysDirty(t *testing.T) { + dsn, db := postgres(t) + dir := t.TempDir() + migration(t, dir, "1", "CREATE TABLE committed(id int);", "DROP TABLE committed;") + migration(t, dir, "2", "THIS IS INVALID SQL;", "SELECT 1;") + migration(t, dir, "3", "CREATE TABLE never_applied(id int);", "DROP TABLE never_applied;") + r := nativeRunner(t) + req := driver.Request{DSN: dsn, Source: driver.Source{Dir: dir}} + if _, err := r.Up(context.Background(), req); err == nil { + t.Fatal("partial failure accepted") + } + st, err := r.Status(context.Background(), req) + if err != nil || st.Current != "2" || !st.Dirty { + t.Fatalf("dirty status=%+v, %v", st, err) + } + if _, err = r.Up(context.Background(), req); err == nil { + t.Fatal("dirty state automatically repaired") + } + var committed, later bool + if err = db.QueryRow("SELECT to_regclass('committed') IS NOT NULL, to_regclass('never_applied') IS NOT NULL").Scan(&committed, &later); err != nil || !committed || later { + t.Fatalf("partial catalog committed=%t later=%t, %v", committed, later, err) + } +} + +func waitQuery(t *testing.T, db *sql.DB, query, application string) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + for ctx.Err() == nil { + var running bool + if err := db.QueryRowContext(ctx, "SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE pid<>pg_backend_pid() AND datname=current_database() AND state='active' AND query=$1 AND application_name=$2)", query, application).Scan(&running); err != nil { + t.Fatal(err) + } + if running { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("migration did not reach expected database query") +} + +func applicationDSN(t *testing.T, dsn string) (string, string) { + t.Helper() + u, err := url.Parse(dsn) + if err != nil { + t.Fatal(err) + } + name := fmt.Sprintf("task15_%d", time.Now().UnixNano()) + query := u.Query() + query.Set("application_name", name) + u.RawQuery = query.Encode() + return u.String(), name +} + +func TestPostgresActiveCancellationStopsWork(t *testing.T) { + dsn, db := postgres(t) + dsn, application := applicationDSN(t, dsn) + dir := t.TempDir() + query := "SELECT pg_sleep(10); CREATE TABLE late_cancel(id int);" + migration(t, dir, "1", query, "DROP TABLE IF EXISTS late_cancel;") + r := nativeRunner(t) + req := driver.Request{DSN: dsn, Source: driver.Source{Dir: dir}} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { _, err := r.Up(ctx, req); done <- err }() + waitQuery(t, db, query, application) + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("active cancel=%v", err) + } + case <-time.After(2 * time.Second): + t.Error("cancelled SQL still running") + if err := <-done; err == nil { + t.Error("cancelled migration reported success") + } + } + var late, active bool + if err := db.QueryRow("SELECT to_regclass('late_cancel') IS NOT NULL, EXISTS(SELECT 1 FROM pg_stat_activity WHERE pid<>pg_backend_pid() AND state='active' AND query=$1 AND application_name=$2)", query, application).Scan(&late, &active); err != nil || late || active { + t.Fatalf("cancelled work: late=%t active=%t, %v", late, active, err) + } +} + +func TestPostgresCancelledLockWaitHasNoLateWork(t *testing.T) { + dsn, db := postgres(t) + dsn, application := applicationDSN(t, dsn) + dir := t.TempDir() + migration(t, dir, "1", "CREATE TABLE after_lock(id int);", "DROP TABLE after_lock;") + r := nativeRunner(t) + req := driver.Request{DSN: dsn, Source: driver.Source{Dir: dir}} + if _, err := r.Status(context.Background(), req); err != nil { + t.Fatal(err) + } + var schema string + if err := db.QueryRow("SELECT current_schema()").Scan(&schema); err != nil { + t.Fatal(err) + } + u, err := url.Parse(dsn) + if err != nil { + t.Fatal(err) + } + key, err := database.GenerateAdvisoryLockId(u.Path, schema, "schema_migrations") + if err != nil { + t.Fatal(err) + } + holder, err := db.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + defer func() { _ = holder.Close() }() + if _, err = holder.ExecContext(context.Background(), "SELECT pg_advisory_lock($1)", key); err != nil { + t.Fatal(err) + } + defer func() { _, _ = holder.ExecContext(context.Background(), "SELECT pg_advisory_unlock($1)", key) }() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { _, e := r.Up(ctx, req); done <- e }() + waitQuery(t, db, "SELECT pg_advisory_lock($1)", application) + cancel() + select { + case err = <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("lock cancellation=%v", err) + } + case <-time.After(2 * time.Second): + t.Error("cancelled advisory-lock waiter still running") + if _, err = holder.ExecContext(context.Background(), "SELECT pg_advisory_unlock($1)", key); err != nil { + t.Fatal(err) + } + <-done + } + var late bool + if err = db.QueryRow("SELECT to_regclass('after_lock') IS NOT NULL").Scan(&late); err != nil || late { + t.Fatalf("late lock work=%t, %v", late, err) + } + if _, err = holder.ExecContext(context.Background(), "SELECT pg_advisory_unlock($1)", key); err != nil { + t.Fatal(err) + } + if _, err = r.Up(context.Background(), req); err != nil { + t.Fatalf("fresh-context lock recovery=%v", err) + } +} + +func TestPostgresConcurrentUpUsesEngineLock(t *testing.T) { + dsn, db := postgres(t) + dir := t.TempDir() + migration(t, dir, "1", "CREATE TABLE once_only(id int); INSERT INTO once_only VALUES (1); SELECT pg_sleep(0.1);", "DROP TABLE once_only;") + r := nativeRunner(t) + req := driver.Request{DSN: dsn, Source: driver.Source{Dir: dir}} + done := make(chan error, 2) + for range 2 { + go func() { _, err := r.Up(context.Background(), req); done <- err }() + } + for range 2 { + if err := <-done; err != nil { + t.Fatalf("concurrent migration=%v", err) + } + } + var count int + if err := db.QueryRow("SELECT count(*) FROM once_only").Scan(&count); err != nil || count != 1 { + t.Fatalf("duplicate migration work: count=%d, %v", count, err) + } +} + +// Compile a separate main package so acceptance crosses the actual public API +// and process boundary instead of substituting either runner or native driver. +func TestExternalConsumerPostgres(t *testing.T) { + dsn, _ := postgres(t) + dir := t.TempDir() + migration(t, dir, "1", "CREATE TABLE external_applied(id int);", "DROP TABLE external_applied;") + consumer := t.TempDir() + source := `package main +import ( + "context" + "database/sql" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "time" + "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/driver" + "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/runner" + _ "github.com/jackc/pgx/v5/stdlib" +) +func main() { + r,err:=runner.New(driver.NewDefaultRegistry(),"golang-migrate",io.Discard,io.Discard) + if err!=nil { panic(err) } + req:=driver.Request{DSN:os.Args[1],Source:driver.Source{Dir:os.Args[2]}} + first,err:=r.Up(context.Background(),req);if err!=nil{panic(err)} + repeat,err:=r.Up(context.Background(),req);if err!=nil{panic(err)} + st,err:=r.Status(context.Background(),req);if err!=nil{panic(err)} + if len(first.Applied)!=1||len(repeat.Applied)!=0||st.Current!="1"||st.Dirty{panic("incorrect persisted migration state")} + for name,sql:=range map[string]string{"2_cancel.up.sql":"SELECT pg_sleep(5); CREATE TABLE external_late(id int);","2_cancel.down.sql":"DROP TABLE IF EXISTS external_late;"}{ + if err=os.WriteFile(filepath.Join(req.Source.Dir,name),[]byte(sql),0600);err!=nil{panic(err)} + } + req.Options.Timeout=100*time.Millisecond + start:=time.Now();_,err=r.Up(context.Background(),req);elapsed:=time.Since(start) + db,e:=sql.Open("pgx",req.DSN);if e!=nil{panic(e)};defer db.Close() + var late bool;if e=db.QueryRow("SELECT to_regclass('external_late') IS NOT NULL").Scan(&late);e!=nil{panic(e)} + cancelled:=errors.Is(err,context.DeadlineExceeded) + fmt.Printf("applied=%v repeat=%v current=%s cancelled=%t late=%t elapsed=%s\n",first.Applied,repeat.Applied,st.Current,cancelled,late,elapsed) + if !cancelled||late||elapsed>2*time.Second{os.Exit(1)} +} +` + path := filepath.Join(consumer, "main.go") + if err := os.WriteFile(path, []byte(source), 0600); err != nil { + t.Fatal(err) + } + binary := filepath.Join(consumer, "consumer") + build := exec.Command("go", "build", "-o", binary, path) + if output, err := build.CombinedOutput(); err != nil { + t.Fatalf("external consumer build: %v\n%s", err, output) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + launch := exec.CommandContext(ctx, binary, dsn, dir) + output, err := launch.CombinedOutput() + t.Logf("external consumer result: %s", output) + if err != nil { + t.Fatalf("external consumer launch: %v", err) + } +}