fix(duckdbservice): wipe user-attached catalogs at session create on shared-warm workers

CLAUDE.md

@@ -740,6 +740,18 @@ Invariants for anyone touching this path:
   + the `__default_*`/`duckgres_*` prefixes, which activation re-creates). It
   MUST run before replay on every CreateSession in shared-warm mode, and a
   wipe failure MUST fail the session.
+- **The same boundary covers user ATTACHes** (`wipeUserCatalogs`,
+  `duckdbservice/user_catalogs.go`, runs immediately before the secret wipe):
+  DuckDB catalogs are instance-global too, so an external catalog attached by
+  one session (e.g. a postgres_scanner source holding a live, authenticated
+  upstream connection pool) would otherwise be inherited by the next session
+  on a hot-idle worker. Every non-internal catalog is detached except the
+  system-managed reserved set (`ducklake`, `delta`, `memory` — activation /
+  the pg_catalog compat layer own those). A wipe failure fails the session.
+  Note the inherited pool is frozen at its creation-time configuration:
+  postgres_scanner's `pg_pool_max_connections` SET has no set-callback and
+  only applies at pool creation, so a session inheriting a stale attach cannot
+  reconfigure its pool — one more reason the attach must not survive.
 - **Execute-then-persist ordering.** Persist only statements DuckDB accepted;
   a store failure after a successful exec is an ERROR telling the user the
   secret will NOT survive the session. Replay failures at session create are
@@ -764,7 +776,8 @@ Invariants for anyone touching this path:
   the original (un-redacted) query so it can classify.
 - Touching the interception, wipe/replay, or payload shape → update
   `server/conn_user_secrets_test.go`, `duckdbservice/user_secrets_test.go`,
-  and the `persistent_user_secret`(+`_isolation`) assertions in
+  `duckdbservice/user_catalogs_test.go` (the catalog wipe), and the
+  `persistent_user_secret`(+`_isolation`) / `user_catalog_wipe` assertions in
   `tests/mw-dev/e2e/harness.sh`.
 
 ## Admin Console (VPC-private web UI, `kubernetes` tag)

duckdbservice/service.go

@@ -1224,11 +1224,14 @@ func (p *SessionPool) CreateSession(username, memoryLimit string, threads int, s
 	// secrets are instance-global, so a hot-idle worker reused by a different
 	// user of the same org would otherwise see the previous user's secrets —
 	// both persistent ones and non-persistent (plain/TEMPORARY CREATE SECRET)
-	// ones that pass through to the worker. Wipe ALL user secrets first
-	// (mandatory — this is the cross-user isolation step), then replay this
-	// user's secrets from the control plane. Replay failures degrade to
-	// warnings; a wipe failure fails the session because handing user A's
-	// secrets to user B is not acceptable.
+	// ones that pass through to the worker. Attached catalogs are likewise
+	// instance-global, so user ATTACHes are wiped on the same boundary (a
+	// catalog holds a live upstream connection pool). Detach ALL user catalogs
+	// and wipe ALL user secrets first (mandatory — this is the cross-user
+	// isolation step), then replay this user's secrets from the control plane.
+	// Replay failures degrade to warnings; a wipe failure fails the session
+	// because handing user A's secrets or attached sources to user B is not
+	// acceptable.
 	var secretWarnings []string
 	switch {
 	case p.sharedWarmMode && p.maxSessions != 1:
@@ -1245,6 +1248,24 @@ func (p *SessionPool) CreateSession(username, memoryLimit string, threads int, s
 		}
 	case p.sharedWarmMode:
 		secretCtx, secretCancel := context.WithTimeout(context.Background(), userSecretOpTimeout)
+		// Catalog wipe first: an attached catalog (e.g. postgres_scanner) holds
+		// a live, authenticated upstream connection pool, so it is the same
+		// cross-user isolation boundary as secrets, one level up. Detaching
+		// before the secret wipe also closes pools that reference the secrets
+		// about to be dropped. A failure fails the session, exactly like the
+		// secret wipe below.
+		wipedCatalogs, catalogWipeErr := wipeUserCatalogs(secretCtx, conn)
+		if catalogWipeErr != nil {
+			secretCancel()
+			_ = conn.Close()
+			p.mu.Lock()
+			p.reserved--
+			p.mu.Unlock()
+			return nil, nil, fmt.Errorf("wipe user catalogs before session start: %w", catalogWipeErr)
+		}
+		if len(wipedCatalogs) > 0 {
+			slog.Info("Detached user catalogs left by previous session.", "user", username, "catalogs", wipedCatalogs)
+		}
 		wiped, wipeErr := wipeUserSecrets(secretCtx, conn)
 		if wipeErr != nil {
 			secretCancel()

duckdbservice/user_catalogs.go

@@ -0,0 +1,68 @@
+package duckdbservice
+
+import (
+	"context"
+	"fmt"
+	"strings"
+)
+
+// reservedCatalogNames are the attached catalogs that must survive the
+// session-create wipe: the org's DuckLake/Delta catalogs (activation owns
+// them — detaching would break the session's metadata init) and `memory`,
+// which carries the pg_catalog compatibility layer (memory.main.*).
+// `system`/`temp` are excluded structurally instead: duckdb_databases()
+// marks them internal, and DuckDB refuses to detach them.
+var reservedCatalogNames = map[string]bool{
+	"ducklake": true,
+	"delta":    true,
+	"memory":   true,
+}
+
+// wipeUserCatalogs detaches every user-attached catalog on the shared per-org
+// DuckDB instance. DuckDB ATTACH is instance-global and a hot-idle worker is
+// reused across sessions of an org, so without this a catalog attached by one
+// session (say an external Postgres source with a live, authenticated
+// connection pool inside postgres_scanner) is silently inherited by the next
+// session — the same cross-user isolation boundary as wipeUserSecrets, one
+// level up. (A stale inherited attach also carries its creation-time
+// connection-pool configuration: postgres_scanner's pg_pool_max_connections
+// SET only applies at pool creation, so the inheriting session cannot
+// reconfigure the pool it just inherited.)
+//
+// A wipe failure fails the session: handing user A's attached sources (and
+// their authenticated upstream connections) to user B is not acceptable.
+// Returns the names that were detached.
+func wipeUserCatalogs(ctx context.Context, h secretDBHandle) ([]string, error) {
+	rows, err := h.QueryContext(ctx, "SELECT database_name, internal FROM duckdb_databases()")
+	if err != nil {
+		return nil, fmt.Errorf("list databases: %w", err)
+	}
+	var names []string
+	for rows.Next() {
+		var name string
+		var internal bool
+		if err := rows.Scan(&name, &internal); err != nil {
+			_ = rows.Close()
+			return nil, err
+		}
+		if internal || reservedCatalogNames[name] {
+			continue
+		}
+		names = append(names, name)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	_ = rows.Close()
+
+	var wiped []string
+	for _, name := range names {
+		// Catalog names are user-controlled; quote defensively.
+		quoted := `"` + strings.ReplaceAll(name, `"`, `""`) + `"`
+		if _, err := h.ExecContext(ctx, "DETACH DATABASE IF EXISTS "+quoted); err != nil {
+			return wiped, fmt.Errorf("detach catalog %q: %w", name, err)
+		}
+		wiped = append(wiped, name)
+	}
+	return wiped, nil
+}

duckdbservice/user_catalogs_test.go

@@ -0,0 +1,131 @@
+package duckdbservice
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+	"time"
+
+	_ "github.com/duckdb/duckdb-go/v2"
+	"github.com/posthog/duckgres/server"
+)
+
+func databaseNames(t *testing.T, db *sql.DB) map[string]bool {
+	t.Helper()
+	rows, err := db.Query("SELECT database_name FROM duckdb_databases()")
+	if err != nil {
+		t.Fatalf("duckdb_databases: %v", err)
+	}
+	defer func() { _ = rows.Close() }()
+	names := map[string]bool{}
+	for rows.Next() {
+		var name string
+		if err := rows.Scan(&name); err != nil {
+			t.Fatalf("scan: %v", err)
+		}
+		names[name] = true
+	}
+	return names
+}
+
+// A hot-idle worker is reused across sessions of an org, and DuckDB attached
+// catalogs are instance-global: a user-attached catalog (e.g. an external
+// Postgres source) would otherwise linger into the next session — leaking the
+// previous session's data sources (and, for postgres_scanner, its live pooled
+// connections) to whoever gets the worker next. The session-create wipe must
+// detach every user catalog while preserving the system-managed ones
+// (ducklake, delta, memory, and the internal system/temp databases).
+func TestWipeUserCatalogs(t *testing.T) {
+	db := openSecretTestDB(t)
+	mustExec := func(q string) {
+		t.Helper()
+		if _, err := db.Exec(q); err != nil {
+			t.Fatalf("exec %q: %v", q, err)
+		}
+	}
+	// User catalogs left behind by a "previous session". One name uses
+	// identifier-quoting edge cases to prove the wipe quotes safely.
+	mustExec("ATTACH ':memory:' AS userdb")
+	mustExec(`ATTACH ':memory:' AS "weird ""quoted"" name"`)
+	// Name-simulate the reserved system-managed catalogs (the real ones are
+	// attached by activation with these exact names).
+	mustExec("ATTACH ':memory:' AS ducklake")
+	mustExec("ATTACH ':memory:' AS delta")
+
+	wiped, err := wipeUserCatalogs(context.Background(), db)
+	if err != nil {
+		t.Fatalf("wipeUserCatalogs: %v", err)
+	}
+	if len(wiped) != 2 {
+		t.Fatalf("expected 2 wiped catalogs, got %v", wiped)
+	}
+	names := databaseNames(t, db)
+	for _, gone := range []string{"userdb", `weird "quoted" name`} {
+		if names[gone] {
+			t.Errorf("catalog %q survived the wipe", gone)
+		}
+	}
+	for _, kept := range []string{"ducklake", "delta", "memory", "system", "temp"} {
+		if !names[kept] {
+			t.Errorf("reserved/internal catalog %q was wiped", kept)
+		}
+	}
+}
+
+// CreateSession on a shared-warm (k8s) worker must detach catalogs attached by
+// the previous session of the same worker: DuckDB ATTACH is instance-global
+// and the worker is reused, so without the wipe the next session inherits the
+// previous session's external sources (and their authenticated pools).
+func TestCreateSessionWipesPreviousSessionCatalogs(t *testing.T) {
+	pool := &SessionPool{
+		sessions:       make(map[string]*Session),
+		stopRefresh:    make(map[string]func()),
+		duckLakeSem:    make(chan struct{}, 1),
+		cfg:            server.Config{Users: map[string]string{"postgres": "postgres"}},
+		startTime:      time.Now(),
+		warmupDone:     make(chan struct{}),
+		sharedWarmMode: true,
+		maxSessions:    1, // k8s workers run with DUCKGRES_DUCKDB_MAX_SESSIONS=1
+	}
+	close(pool.warmupDone)
+	pool.createDBPair = func(server.Config, chan struct{}, string, time.Time, string) (*DuckDBPair, error) {
+		db, err := sql.Open("duckdb", "")
+		if err != nil {
+			return nil, err
+		}
+		return PairFromMain(db), nil
+	}
+	pool.activateDBConnection = func(*sql.DB, server.Config, chan struct{}, string) error { return nil }
+	if err := pool.activateTenant(ActivationPayload{
+		WorkerControlMetadata: server.WorkerControlMetadata{
+			OwnerEpoch:   1,
+			CPInstanceID: "cp-test",
+			WorkerID:     1,
+		},
+		OrgID: "test-org",
+	}); err != nil {
+		t.Fatalf("activateTenant: %v", err)
+	}
+
+	first, _, err := pool.CreateSession("alice", "", 0, nil)
+	if err != nil {
+		t.Fatalf("first CreateSession: %v", err)
+	}
+	// Alice's session attaches an external source catalog.
+	if _, err := pool.warmupDB.Exec("ATTACH ':memory:' AS userdb"); err != nil {
+		t.Fatalf("attach user catalog: %v", err)
+	}
+	if err := pool.DestroySession(first.ID); err != nil {
+		t.Fatalf("DestroySession: %v", err)
+	}
+
+	if _, _, err := pool.CreateSession("bob", "", 0, nil); err != nil {
+		t.Fatalf("second CreateSession: %v", err)
+	}
+	if databaseNames(t, pool.warmupDB)["userdb"] {
+		t.Fatal("user-attached catalog from the previous session survived CreateSession")
+	}
+	if !databaseNames(t, pool.warmupDB)["memory"] {
+		t.Fatal("reserved memory catalog was wiped")
+	}
+}

tests/mw-dev/e2e/harness.sh

@@ -2555,6 +2555,38 @@ persistent_user_secret_isolation() { # org rootpw
   log "user secret isolation OK on $org ($u2 blind to root's persistent AND temporary secrets; root's stored copy intact)"
 }
 
+# Cross-session catalog hygiene: DuckDB ATTACH is instance-global, so a catalog
+# attached by one session (e.g. an external postgres_scanner source holding a
+# live authenticated connection pool) would linger on a hot-idle worker and be
+# inherited by the next session unless the session-create wipe detaches it.
+# The second connection typically reuses the first's hot-idle worker — the leak
+# path — but the invariant holds either way: a fresh session must never see
+# another session's attached catalogs, while the system catalogs (ducklake,
+# memory) must be preserved.
+user_catalog_wipe() { # org password
+  org="$1"; pw="$2"; cname="e2e_probe_catalog_$$"
+  log "user catalog wipe on $org"
+
+  # Same-session check: BOTH statements must ride ONE connection, so feed psql
+  # via stdin (each statement is its own Q message on the same session). Two
+  # separate `pg` calls would be two sessions — and the second one's
+  # CreateSession wipe would already have detached the probe.
+  out="$(printf "ATTACH IF NOT EXISTS ':memory:' AS %s;\nSELECT count(*) FROM duckdb_databases() WHERE database_name = '%s' AND NOT internal;\n" "$cname" "$cname" | \
+    PGPASSWORD="$pw" psql \
+      "sslmode=require host=$org$SNI_SUFFIX hostaddr=$CP_IP port=5432 user=root dbname=ducklake" \
+      -v ON_ERROR_STOP=1 -tA 2>&1)" || fail "user catalog wipe: attach session failed: $out"
+  [ "$(printf '%s' "$out" | tail -1)" = "1" ] || fail "user catalog wipe: probe catalog not attached in its own session ($out)"
+
+  # ...and gone from the NEXT fresh session (same hot-idle worker or not).
+  n="$(pg "$org" "$pw" ducklake "SELECT count(*) FROM duckdb_databases() WHERE database_name = '$cname' AND NOT internal")"
+  [ "$n" = "0" ] || fail "user catalog wipe: probe catalog leaked into a fresh session (count=$n)"
+
+  # The system-managed catalogs must survive the wipe on every session.
+  n="$(pg "$org" "$pw" ducklake "SELECT count(*) FROM duckdb_databases() WHERE database_name IN ('ducklake','memory')")"
+  [ "$n" = "2" ] || fail "user catalog wipe: system catalogs missing on a fresh session (count=$n)"
+  log "user catalog wipe OK on $org (probe attached in-session, detached by next session, ducklake+memory preserved)"
+}
+
 # ---- resilience -----------------------------------------------------------
 # Worker pod killed mid-life → CP refills and a fresh query succeeds.
 # Ported from TestK8sWorkerCrashRecovery.
@@ -4362,6 +4394,7 @@ lane_cnpg() { # full wire/catalog/concurrency/sizing coverage on the cnpg org
   httpfs_retry_budget    "$CNPG" "$cnpg_pw"   # S3-503 retry budget raised per worker (applyHTTPFSRetryBudget)
   persistent_user_secret "$CNPG" "$cnpg_pw"   # after rw_ducklake (org worker hot)
   persistent_user_secret_isolation "$CNPG" "$cnpg_pw"
+  user_catalog_wipe "$CNPG" "$cnpg_pw"
   pipeline_error_recovery "$CNPG" "$cnpg_pw"  # after rw_ducklake (table writes proven)
   server_side_cursors    "$CNPG" "$cnpg_pw"
   cancel_then_reuse_same_session "$CNPG" "$cnpg_pw"