fix(operator): terminate other client backends before drop_schemas_on

Before each DROP SCHEMA ... CASCADE run during schema-migration prep,
ask postgres to terminate every other client backend connected to
the target database. The operator owns the restore database fully
until handover, so any other session is necessarily a stray client
whose lock-holding would queue our DDL indefinitely.

Real scenario that triggered this: a long-running external psql
session got stuck mid-CREATE TABLE in the target restore while it
was still in the operator's Switching phase, holding namespace
locks that all subsequent DROP SCHEMA attempts queued behind for
20+ hours. With this change, the operator clears the deck before
attempting the DROP.

Best-effort by design: backends stuck in uninterruptible kernel
waits won't exit on SIGTERM (the case above eventually needed a
pod restart). But those are rare; the 99% case is a normal live
client and pg_terminate_backend takes care of it.

Author: passcod

src/controllers/postgres.rs

@@ -309,9 +309,40 @@ pub async fn existing_schemas_on(
 		.collect())
 }
 
+/// Terminate every other client backend connected to the current database.
+/// The operator owns each restore's database fully until handover, so any
+/// other session is a transient or stray client whose lock-holding would
+/// block our DDL. Best-effort: rows where `pg_terminate_backend` returns
+/// false (already gone, raced with another terminator) are ignored.
+///
+/// Note: this only dislodges sessions PG can interrupt. A backend stuck in
+/// an uninterruptible kernel wait (rare but it happens — e.g. a long-dead
+/// client whose TCP socket the kernel hasn't reaped) will not exit. For
+/// those, the restore pod must be restarted.
+pub async fn terminate_other_backends(pg: &tokio_postgres::Client) -> Result<u64> {
+	let row = pg
+		.query_one(
+			"SELECT count(*) FILTER (WHERE pg_terminate_backend(pid)) \
+			 FROM pg_stat_activity \
+			 WHERE datname = current_database() \
+			   AND backend_type = 'client backend' \
+			   AND pid <> pg_backend_pid()",
+			&[],
+		)
+		.await?;
+	let killed: i64 = row.get(0);
+	if killed > 0 {
+		info!(count = killed, "terminated other client backends");
+	}
+	Ok(killed as u64)
+}
+
 /// Drop the given schemas (and all contents) on an already-open connection.
+/// Terminates other client backends on the database first so stray sessions
+/// holding locks on the target schemas don't queue our DDL behind them.
 /// Idempotent: missing schemas are skipped via `DROP SCHEMA IF EXISTS`.
 pub async fn drop_schemas_on(pg: &tokio_postgres::Client, schemas: &[String]) -> Result<()> {
+	terminate_other_backends(pg).await?;
 	for schema in schemas {
 		let stmt = format!("DROP SCHEMA IF EXISTS {} CASCADE", quote_ident(schema));
 		debug!(schema = schema, "dropping schema");