fix(process-list): cancel queries without closing sessions

Closes #3967
This commit is contained in:
t8y2 committed 2026-08-12 21:57:19 +00:00
1 parent 396ebf8393
commit 4106a8ce22
15 files changed
+189 -165

No files matched your search

@@ -1,6 +1,6 @@
import { describe, expect, it } from "vitest";
import type { QueryResult } from "@/types/database";
import { buildKillSql, clampInterval, createProcessListLoadCoordinator, mapProcessRows, processListExecutionError, processListSessionCount, supportsProcessList } from "@/lib/database/mysqlProcessList";
import { buildCancelQuerySql, clampInterval, createProcessListLoadCoordinator, mapProcessRows, processListExecutionError, processListSessionCount, supportsProcessList } from "@/lib/database/mysqlProcessList";
function result(columns: string[], rows: (string | number | boolean | null)[][]): QueryResult {
return { columns, rows, affected_rows: 0, execution_time_ms: 0 };
@@ -35,15 +35,16 @@ describe("mapProcessRows", () => {
});
});
describe("buildKillSql", () => {
it("builds KILL CONNECTION for a valid id", () => {
expect(buildKillSql(8213)).toBe("KILL CONNECTION 8213");
describe("buildCancelQuerySql", () => {
it("builds KILL QUERY for a valid id without disconnecting the session", () => {
expect(buildCancelQuerySql(8213)).toBe("KILL QUERY 8213");
});
it("rejects non-integer or negative ids", () => {
expect(() => buildKillSql(1.5)).toThrow();
expect(() => buildKillSql(-1)).toThrow();
expect(() => buildKillSql(Number.NaN)).toThrow();
it("rejects non-integer or nonpositive ids", () => {
expect(() => buildCancelQuerySql(1.5)).toThrow();
expect(() => buildCancelQuerySql(0)).toThrow();
expect(() => buildCancelQuerySql(-1)).toThrow();
expect(() => buildCancelQuerySql(Number.NaN)).toThrow();
});
});
@@ -1,12 +1,12 @@
import { describe, expect, it } from "vitest";
import type { QueryResult } from "@/types/database";
import {
buildKingbaseKillSql,
buildKingbasePgKillSql,
buildPgKillSql,
buildKingbaseCancelQuerySql,
buildKingbasePgCancelQuerySql,
buildPgCancelQuerySql,
isKingbaseOwnSessionCatalogCompatibilityError,
isKingbaseProcessListCatalogCompatibilityError,
isKingbaseTerminateCatalogCompatibilityError,
isKingbaseCancelCatalogCompatibilityError,
isPgProcessListCompatibilityError,
KINGBASE_OWN_SESSION_SQL,
KINGBASE_PG_OWN_SESSION_SQL,
@@ -15,7 +15,9 @@ import {
mapPgProcessRows,
OPENGAUSS_OWN_SESSION_SQL,
OPENGAUSS_PROCESS_LIST_SQL,
pgKillResultError,
kingbaseCancelQueryResultError,
kingbasePgCancelQueryResultError,
pgCancelQueryResultError,
PG_PROCESS_LIST_LEGACY_SQL,
PG_PROCESS_LIST_SQL,
} from "@/lib/database/postgresProcessList";
@@ -56,22 +58,22 @@ describe("mapPgProcessRows", () => {
});
});
describe("buildPgKillSql", () => {
it("builds pg_terminate_backend for a valid pid", () => {
expect(buildPgKillSql(4211)).toBe("SELECT pg_terminate_backend(4211)");
describe("buildPgCancelQuerySql", () => {
it("builds pg_cancel_backend for a valid pid without disconnecting the session", () => {
expect(buildPgCancelQuerySql(4211)).toBe("SELECT pg_cancel_backend(4211)");
});
it("rejects non-integer, zero, or negative pids", () => {
expect(() => buildPgKillSql(1.5)).toThrow();
expect(() => buildPgKillSql(0)).toThrow();
expect(() => buildPgKillSql(-1)).toThrow();
expect(() => buildPgKillSql(Number.NaN)).toThrow();
expect(() => buildPgCancelQuerySql(1.5)).toThrow();
expect(() => buildPgCancelQuerySql(0)).toThrow();
expect(() => buildPgCancelQuerySql(-1)).toThrow();
expect(() => buildPgCancelQuerySql(Number.NaN)).toThrow();
});
it("builds the KingbaseES sys_terminate_backend call with the same PID validation", () => {
expect(buildKingbaseKillSql(4211)).toBe("SELECT sys_terminate_backend(4211)");
expect(() => buildKingbaseKillSql(0)).toThrow();
expect(() => buildKingbaseKillSql(1.5)).toThrow();
it("builds the KingbaseES sys_cancel_backend call with the same PID validation", () => {
expect(buildKingbaseCancelQuerySql(4211)).toBe("SELECT sys_cancel_backend(4211)");
expect(() => buildKingbaseCancelQuerySql(0)).toThrow();
expect(() => buildKingbaseCancelQuerySql(1.5)).toThrow();
});
});
@@ -83,7 +85,7 @@ describe("PostgreSQL-family process SQL", () => {
expect(OPENGAUSS_PROCESS_LIST_SQL).toContain("FROM pg_catalog.pg_stat_activity");
expect(OPENGAUSS_PROCESS_LIST_SQL).toContain("CASE WHEN waiting");
expect(OPENGAUSS_PROCESS_LIST_SQL).not.toContain("wait_event_type");
expect(driver?.buildKillSql(7)).toBe("SELECT pg_terminate_backend(7)");
expect(driver?.buildCancelQuerySql(7)).toBe("SELECT pg_cancel_backend(7)");
});
it("uses KingbaseES's sys catalog and session functions", () => {
@@ -95,26 +97,26 @@ describe("PostgreSQL-family process SQL", () => {
expect(KINGBASE_PROCESS_LIST_SQL).toContain("extract(epoch FROM CAST(CURRENT_TIMESTAMP AS TIMESTAMP))");
expect(KINGBASE_PROCESS_LIST_SQL).toContain("extract(epoch FROM CAST(coalesce(query_start, xact_start, backend_start) AS TIMESTAMP))");
expect(KINGBASE_PROCESS_LIST_SQL).not.toContain("CURRENT_TIMESTAMP - coalesce(query_start");
expect(driver?.buildKillSql(7)).toBe("SELECT sys_terminate_backend(7)");
expect(driver?.buildCancelQuerySql(7)).toBe("SELECT sys_cancel_backend(7)");
});
it("provides pg_catalog fallbacks for pg-compatible KingbaseES servers", () => {
const driver = resolveProcessListDriver("kingbase");
expect(driver?.fallbackListSql).toBe(KINGBASE_PG_PROCESS_LIST_SQL);
expect(driver?.fallbackOwnSessionSql).toBe(KINGBASE_PG_OWN_SESSION_SQL);
expect(driver?.buildFallbackKillSql?.(7)).toBe("SELECT pg_terminate_backend(7)");
expect(driver?.buildFallbackCancelQuerySql?.(7)).toBe("SELECT pg_cancel_backend(7)");
expect(KINGBASE_PG_PROCESS_LIST_SQL).toContain("FROM pg_catalog.pg_stat_activity");
expect(KINGBASE_PG_PROCESS_LIST_SQL).not.toContain("sys_catalog");
expect(buildKingbasePgKillSql(4211)).toBe("SELECT pg_terminate_backend(4211)");
expect(buildKingbasePgCancelQuerySql(4211)).toBe("SELECT pg_cancel_backend(4211)");
});
it("retries only missing sys relation/function errors", () => {
expect(isKingbaseProcessListCatalogCompatibilityError(new Error('relation "sys_catalog.sys_stat_activity" does not exist (SQLSTATE 42P01)'))).toBe(true);
expect(isKingbaseOwnSessionCatalogCompatibilityError(Object.assign(new Error("function sys_backend_pid() does not exist"), { code: "42883" }))).toBe(true);
expect(isKingbaseTerminateCatalogCompatibilityError(new Error("function sys_terminate_backend(integer) does not exist (SQLSTATE 42883)"))).toBe(true);
expect(isKingbaseCancelCatalogCompatibilityError(new Error("function sys_cancel_backend(integer) does not exist (SQLSTATE 42883)"))).toBe(true);
expect(isKingbaseProcessListCatalogCompatibilityError(Object.assign(new Error("permission denied for relation sys_catalog.sys_stat_activity"), { code: "42501" }))).toBe(false);
expect(isKingbaseOwnSessionCatalogCompatibilityError(new Error("authentication failed for sys_backend_pid"))).toBe(false);
expect(isKingbaseTerminateCatalogCompatibilityError(new Error("connection refused"))).toBe(false);
expect(isKingbaseCancelCatalogCompatibilityError(new Error("connection refused"))).toBe(false);
});
});
@@ -128,11 +130,13 @@ describe("Postgres compatibility", () => {
expect(isPgProcessListCompatibilityError(new Error("permission denied for view pg_stat_activity"))).toBe(false);
});
it("requires pg_terminate_backend to confirm termination", () => {
expect(pgKillResultError([result(["pg_terminate_backend"], [[true]])])).toBeNull();
expect(pgKillResultError([result(["pg_terminate_backend"], [["t"]])])).toBeNull();
expect(pgKillResultError([result(["pg_terminate_backend"], [[false]])])).toContain("did not terminate");
expect(pgKillResultError([result(["pg_terminate_backend"], [])])).toContain("did not terminate");
it("requires cancellation functions to confirm that a running query was canceled", () => {
expect(pgCancelQueryResultError([result(["pg_cancel_backend"], [[true]])])).toBeNull();
expect(pgCancelQueryResultError([result(["pg_cancel_backend"], [["t"]])])).toBeNull();
expect(pgCancelQueryResultError([result(["pg_cancel_backend"], [[false]])])).toContain("did not cancel");
expect(pgCancelQueryResultError([result(["pg_cancel_backend"], [])])).toContain("did not cancel");
expect(kingbaseCancelQueryResultError([result(["sys_cancel_backend"], [[false]])])).toContain("did not cancel");
expect(kingbasePgCancelQueryResultError([result(["pg_cancel_backend"], [[true]])])).toBeNull();
});
});
@@ -140,11 +144,11 @@ describe("resolveProcessListDriver", () => {
it("routes MySQL and Postgres engines to distinct drivers", () => {
const mysql = resolveProcessListDriver("mysql");
const postgres = resolveProcessListDriver("postgres");
expect(mysql?.buildKillSql(7)).toBe("KILL CONNECTION 7");
expect(postgres?.buildKillSql(7)).toBe("SELECT pg_terminate_backend(7)");
expect(mysql?.buildCancelQuerySql(7)).toBe("KILL QUERY 7");
expect(postgres?.buildCancelQuerySql(7)).toBe("SELECT pg_cancel_backend(7)");
expect(postgres?.fallbackListSql).toBe(PG_PROCESS_LIST_LEGACY_SQL);
expect(postgres?.shouldUseFallbackListSql?.(new Error('column "wait_event" does not exist'))).toBe(true);
expect(postgres?.killResultError?.([result(["pg_terminate_backend"], [[false]])])).toContain("did not terminate");
expect(postgres?.cancelQueryResultError?.([result(["pg_cancel_backend"], [[false]])])).toContain("did not cancel");
expect(resolveProcessListDriver("sqlite")).toBeNull();
});
@@ -190,7 +194,7 @@ describe("connectionSupportsProcessList", () => {
describe("resolveProcessListDriverForConnection", () => {
it("returns the engine driver for a JDBC MySQL connection and null for lookalikes", () => {
expect(resolveProcessListDriverForConnection(conn({ db_type: "jdbc", connection_string: "jdbc:mysql://db/app" }))?.buildKillSql(9)).toBe("KILL CONNECTION 9");
expect(resolveProcessListDriverForConnection(conn({ db_type: "jdbc", connection_string: "jdbc:mysql://db/app" }))?.buildCancelQuerySql(9)).toBe("KILL QUERY 9");
expect(resolveProcessListDriverForConnection(conn({ db_type: "jdbc", connection_string: "jdbc:hive2://hs2/default" }))).toBeNull();
});
});
@@ -2,7 +2,7 @@ import type { DatabaseType, QueryResult } from "@/types/database";
/**
* Engines that speak the MySQL protocol and support `SHOW FULL PROCESSLIST` /
* `KILL CONNECTION`. MariaDB, TiDB, and OceanBase ride the `mysql` dbType via a
* `KILL QUERY`. MariaDB, TiDB, and OceanBase ride the `mysql` dbType via a
* driver profile, so they are covered by the `"mysql"` entry.
*/
const PROCESS_LIST_DB_TYPES = new Set<DatabaseType>(["mysql"]);
@@ -127,14 +127,14 @@ export function mapProcessRows(result: QueryResult | null | undefined): ProcessR
}
/**
* Build the `KILL CONNECTION <id>` statement. `id` must be a finite integer; it
* Build the `KILL QUERY <id>` statement. `id` must be a finite positive integer; it
* is validated (never interpolated as free text) so there is no injection path.
*/
export function buildKillSql(id: number): string {
if (!Number.isInteger(id) || id < 0) {
export function buildCancelQuerySql(id: number): string {
if (!Number.isInteger(id) || id <= 0) {
throw new Error(`Invalid session id: ${id}`);
}
return `KILL CONNECTION ${id}`;
return `KILL QUERY ${id}`;
}
/** Clamp a user-entered refresh interval to a safe integer range of seconds. */
@@ -149,7 +149,7 @@ export function mapPgProcessRows(result: QueryResult | null | undefined): PgProc
}
/**
* Build a `SELECT pg_terminate_backend(<pid>)` statement. `pid` is validated as a
* Build a `SELECT pg_cancel_backend(<pid>)` statement. `pid` is validated as a
* finite positive integer (never interpolated as free text) so there is no
* injection path.
*/
@@ -159,39 +159,39 @@ function validateBackendPid(pid: number): void {
}
}
export function buildPgKillSql(pid: number): string {
export function buildPgCancelQuerySql(pid: number): string {
validateBackendPid(pid);
return `SELECT pg_terminate_backend(${pid})`;
return `SELECT pg_cancel_backend(${pid})`;
}
export function buildKingbaseKillSql(pid: number): string {
export function buildKingbaseCancelQuerySql(pid: number): string {
validateBackendPid(pid);
return `SELECT sys_terminate_backend(${pid})`;
return `SELECT sys_cancel_backend(${pid})`;
}
export function buildKingbasePgKillSql(pid: number): string {
export function buildKingbasePgCancelQuerySql(pid: number): string {
validateBackendPid(pid);
return `SELECT pg_terminate_backend(${pid})`;
return `SELECT pg_cancel_backend(${pid})`;
}
/** Return an error when PostgreSQL declines to terminate the target backend. */
export function pgKillResultError(results: QueryResult[]): string | null {
return backendKillResultError(results, "pg_terminate_backend");
/** Return an error when PostgreSQL reports that no running query was canceled. */
export function pgCancelQueryResultError(results: QueryResult[]): string | null {
return backendCancelQueryResultError(results, "pg_cancel_backend");
}
export function kingbaseKillResultError(results: QueryResult[]): string | null {
return backendKillResultError(results, "sys_terminate_backend");
export function kingbaseCancelQueryResultError(results: QueryResult[]): string | null {
return backendCancelQueryResultError(results, "sys_cancel_backend");
}
export function kingbasePgKillResultError(results: QueryResult[]): string | null {
return backendKillResultError(results, "pg_terminate_backend");
export function kingbasePgCancelQueryResultError(results: QueryResult[]): string | null {
return backendCancelQueryResultError(results, "pg_cancel_backend");
}
function backendKillResultError(results: QueryResult[], functionName: string): string | null {
function backendCancelQueryResultError(results: QueryResult[], functionName: string): string | null {
const result = results.find((item) => item.execution_error !== true);
const value = result?.rows?.[0]?.[0];
if (value === true || value === 1 || String(value).toLowerCase() === "t" || String(value).toLowerCase() === "true") return null;
return `${functionName} did not terminate the backend`;
return `${functionName} did not cancel a running query`;
}
/** Detect the undefined-column failure produced by pre-9.6 pg_stat_activity. */
@@ -210,6 +210,6 @@ export function isKingbaseOwnSessionCatalogCompatibilityError(error: unknown): b
return isMissingKingbaseSysFunction(error, ["sys_backend_pid"]);
}
export function isKingbaseTerminateCatalogCompatibilityError(error: unknown): boolean {
return isMissingKingbaseSysFunction(error, ["sys_terminate_backend"]);
export function isKingbaseCancelCatalogCompatibilityError(error: unknown): boolean {
return isMissingKingbaseSysFunction(error, ["sys_cancel_backend"]);
}
@@ -1,16 +1,16 @@
import type { ConnectionConfig, DatabaseType, QueryResult } from "@/types/database";
import { effectiveDatabaseTypeForConnection } from "@/lib/database/jdbcDialect";
import { buildKillSql as buildMysqlKillSql, mapProcessRows as mapMysqlProcessRows, PROCESS_LIST_SQL as MYSQL_PROCESS_LIST_SQL, supportsProcessList as supportsMysqlProcessList } from "./mysqlProcessList";
import { buildCancelQuerySql as buildMysqlCancelQuerySql, mapProcessRows as mapMysqlProcessRows, PROCESS_LIST_SQL as MYSQL_PROCESS_LIST_SQL, supportsProcessList as supportsMysqlProcessList } from "./mysqlProcessList";
import {
buildKingbaseKillSql,
buildKingbasePgKillSql,
buildPgKillSql,
buildKingbaseCancelQuerySql,
buildKingbasePgCancelQuerySql,
buildPgCancelQuerySql,
isKingbaseOwnSessionCatalogCompatibilityError,
isKingbaseProcessListCatalogCompatibilityError,
isKingbaseTerminateCatalogCompatibilityError,
isKingbaseCancelCatalogCompatibilityError,
isPgProcessListCompatibilityError,
kingbaseKillResultError,
kingbasePgKillResultError,
kingbaseCancelQueryResultError,
kingbasePgCancelQueryResultError,
KINGBASE_OWN_SESSION_SQL,
KINGBASE_PG_OWN_SESSION_SQL,
KINGBASE_PG_PROCESS_LIST_SQL,
@@ -18,7 +18,7 @@ import {
mapPgProcessRows,
OPENGAUSS_OWN_SESSION_SQL,
OPENGAUSS_PROCESS_LIST_SQL,
pgKillResultError,
pgCancelQueryResultError,
PG_OWN_SESSION_SQL,
PG_PROCESS_LIST_LEGACY_SQL,
PG_PROCESS_LIST_SQL,
@@ -67,16 +67,16 @@ export interface ProcessListDriver {
maxRows: number;
/** Map a raw list result into typed rows. */
mapRows(result: QueryResult | null | undefined): ProcessRow[];
/** Build the validated statement that kills the given session id. */
buildKillSql(id: number): string;
/** Build the compatibility statement used when the primary kill function is unavailable. */
buildFallbackKillSql?(id: number): string;
/** Restrict kill fallback attempts to known compatibility failures. */
shouldUseFallbackKillSql?(error: unknown): boolean;
/** Validate any engine-specific success value returned by the kill statement. */
killResultError?(results: QueryResult[]): string | null;
/** Validate the success value returned by the compatibility kill statement. */
fallbackKillResultError?(results: QueryResult[]): string | null;
/** Build the validated statement that cancels the selected session's running query. */
buildCancelQuerySql(id: number): string;
/** Build the compatibility statement used when the primary cancellation function is unavailable. */
buildFallbackCancelQuerySql?(id: number): string;
/** Restrict cancellation fallback attempts to known compatibility failures. */
shouldUseFallbackCancelQuerySql?(error: unknown): boolean;
/** Validate any engine-specific success value returned by the cancellation statement. */
cancelQueryResultError?(results: QueryResult[]): string | null;
/** Validate the success value returned by the compatibility cancellation statement. */
fallbackCancelQueryResultError?(results: QueryResult[]): string | null;
}
const MYSQL_COLUMNS: ProcessColumn[] = [
@@ -110,7 +110,7 @@ const MYSQL_DRIVER: ProcessListDriver = {
maxRows: 5000,
// Typed structs carry no index signature; they are plain string-keyed objects at runtime.
mapRows: (result) => mapMysqlProcessRows(result) as unknown as ProcessRow[],
buildKillSql: buildMysqlKillSql,
buildCancelQuerySql: buildMysqlCancelQuerySql,
};
const POSTGRES_DRIVER: ProcessListDriver = {
@@ -122,8 +122,8 @@ const POSTGRES_DRIVER: ProcessListDriver = {
defaultSortKey: "time",
maxRows: 5000,
mapRows: (result) => mapPgProcessRows(result) as unknown as ProcessRow[],
buildKillSql: buildPgKillSql,
killResultError: pgKillResultError,
buildCancelQuerySql: buildPgCancelQuerySql,
cancelQueryResultError: pgCancelQueryResultError,
};
const OPENGAUSS_DRIVER: ProcessListDriver = {
@@ -133,8 +133,8 @@ const OPENGAUSS_DRIVER: ProcessListDriver = {
defaultSortKey: "time",
maxRows: 5000,
mapRows: (result) => mapPgProcessRows(result) as unknown as ProcessRow[],
buildKillSql: buildPgKillSql,
killResultError: pgKillResultError,
buildCancelQuerySql: buildPgCancelQuerySql,
cancelQueryResultError: pgCancelQueryResultError,
};
const KINGBASE_DRIVER: ProcessListDriver = {
@@ -148,11 +148,11 @@ const KINGBASE_DRIVER: ProcessListDriver = {
defaultSortKey: "time",
maxRows: 5000,
mapRows: (result) => mapPgProcessRows(result) as unknown as ProcessRow[],
buildKillSql: buildKingbaseKillSql,
buildFallbackKillSql: buildKingbasePgKillSql,
shouldUseFallbackKillSql: isKingbaseTerminateCatalogCompatibilityError,
killResultError: kingbaseKillResultError,
fallbackKillResultError: kingbasePgKillResultError,
buildCancelQuerySql: buildKingbaseCancelQuerySql,
buildFallbackCancelQuerySql: buildKingbasePgCancelQuerySql,
shouldUseFallbackCancelQuerySql: isKingbaseCancelCatalogCompatibilityError,
cancelQueryResultError: kingbaseCancelQueryResultError,
fallbackCancelQueryResultError: kingbasePgCancelQueryResultError,
};
/** Resolve the process-list driver for a connection, or null if unsupported. */