feat(migrate): expose context-aware in-process runner

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

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
 )

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,14 +286,17 @@ 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,
 	drv atlmigrate.Driver,
 	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)

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{}

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