feat(mysql): terminate sessions from the process list
This commit is contained in:
21 files changed
+312
-18
No files matched your search
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type { QueryResult } from "@/types/database";
|
||||
import { buildCancelQuerySql, clampInterval, createProcessListLoadCoordinator, mapProcessRows, processListExecutionError, processListSessionCount, supportsProcessList } from "@/lib/database/mysqlProcessList";
|
||||
import { buildCancelQuerySql, buildTerminateSessionSql, 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 };
|
||||
@@ -48,6 +48,19 @@ describe("buildCancelQuerySql", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("buildTerminateSessionSql", () => {
|
||||
it("builds KILL for a valid id so an idle session is really disconnected", () => {
|
||||
expect(buildTerminateSessionSql(8213)).toBe("KILL 8213");
|
||||
});
|
||||
|
||||
it("rejects non-integer or nonpositive ids", () => {
|
||||
expect(() => buildTerminateSessionSql(1.5)).toThrow();
|
||||
expect(() => buildTerminateSessionSql(0)).toThrow();
|
||||
expect(() => buildTerminateSessionSql(-1)).toThrow();
|
||||
expect(() => buildTerminateSessionSql(Number.NaN)).toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
describe("clampInterval", () => {
|
||||
it("clamps below the minimum to 1 second", () => {
|
||||
expect(clampInterval(0)).toBe(1);
|
||||
|
||||
@@ -3,10 +3,14 @@ import type { QueryResult } from "@/types/database";
|
||||
import {
|
||||
buildKingbaseCancelQuerySql,
|
||||
buildKingbasePgCancelQuerySql,
|
||||
buildKingbasePgTerminateSessionSql,
|
||||
buildKingbaseTerminateSessionSql,
|
||||
buildPgCancelQuerySql,
|
||||
buildPgTerminateSessionSql,
|
||||
isKingbaseOwnSessionCatalogCompatibilityError,
|
||||
isKingbaseProcessListCatalogCompatibilityError,
|
||||
isKingbaseCancelCatalogCompatibilityError,
|
||||
isKingbaseTerminateCatalogCompatibilityError,
|
||||
isPgProcessListCompatibilityError,
|
||||
KINGBASE_OWN_SESSION_SQL,
|
||||
KINGBASE_PG_OWN_SESSION_SQL,
|
||||
@@ -18,6 +22,7 @@ import {
|
||||
kingbaseCancelQueryResultError,
|
||||
kingbasePgCancelQueryResultError,
|
||||
pgCancelQueryResultError,
|
||||
pgTerminateSessionResultError,
|
||||
PG_PROCESS_LIST_LEGACY_SQL,
|
||||
PG_PROCESS_LIST_SQL,
|
||||
} from "@/lib/database/postgresProcessList";
|
||||
@@ -120,6 +125,32 @@ describe("PostgreSQL-family process SQL", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("session termination statements", () => {
|
||||
it("builds pg_terminate_backend so an idle backend is really disconnected", () => {
|
||||
expect(buildPgTerminateSessionSql(4211)).toBe("SELECT pg_terminate_backend(4211)");
|
||||
expect(buildKingbaseTerminateSessionSql(4211)).toBe("SELECT sys_terminate_backend(4211)");
|
||||
expect(buildKingbasePgTerminateSessionSql(4211)).toBe("SELECT pg_terminate_backend(4211)");
|
||||
});
|
||||
|
||||
it("rejects non-integer or nonpositive pids", () => {
|
||||
expect(() => buildPgTerminateSessionSql(1.5)).toThrow();
|
||||
expect(() => buildPgTerminateSessionSql(0)).toThrow();
|
||||
expect(() => buildPgTerminateSessionSql(Number.NaN)).toThrow();
|
||||
});
|
||||
|
||||
it("requires the server to confirm that a backend was terminated", () => {
|
||||
expect(pgTerminateSessionResultError([result(["pg_terminate_backend"], [[true]])])).toBeNull();
|
||||
expect(pgTerminateSessionResultError([result(["pg_terminate_backend"], [["t"]])])).toBeNull();
|
||||
expect(pgTerminateSessionResultError([result(["pg_terminate_backend"], [[false]])])).toContain("did not terminate");
|
||||
expect(pgTerminateSessionResultError([result(["pg_terminate_backend"], [])])).toContain("did not terminate");
|
||||
});
|
||||
|
||||
it("detects the Kingbase fallback trigger for a missing sys_terminate_backend", () => {
|
||||
expect(isKingbaseTerminateCatalogCompatibilityError(new Error("function sys_terminate_backend(integer) does not exist"))).toBe(true);
|
||||
expect(isKingbaseTerminateCatalogCompatibilityError(new Error("permission denied"))).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("Postgres compatibility", () => {
|
||||
it("provides a pre-9.6 query and only falls back for missing wait-event columns", () => {
|
||||
expect(PG_PROCESS_LIST_SQL).toContain("wait_event_type");
|
||||
@@ -145,7 +176,10 @@ describe("resolveProcessListDriver", () => {
|
||||
const mysql = resolveProcessListDriver("mysql");
|
||||
const postgres = resolveProcessListDriver("postgres");
|
||||
expect(mysql?.buildCancelQuerySql(7)).toBe("KILL QUERY 7");
|
||||
expect(mysql?.buildTerminateSessionSql?.(7)).toBe("KILL 7");
|
||||
expect(postgres?.buildCancelQuerySql(7)).toBe("SELECT pg_cancel_backend(7)");
|
||||
expect(postgres?.buildTerminateSessionSql?.(7)).toBe("SELECT pg_terminate_backend(7)");
|
||||
expect(postgres?.terminateSessionResultError?.([result(["pg_terminate_backend"], [[false]])])).toContain("did not terminate");
|
||||
expect(postgres?.fallbackListSql).toBe(PG_PROCESS_LIST_LEGACY_SQL);
|
||||
expect(postgres?.shouldUseFallbackListSql?.(new Error('column "wait_event" does not exist'))).toBe(true);
|
||||
expect(postgres?.cancelQueryResultError?.([result(["pg_cancel_backend"], [[false]])])).toContain("did not cancel");
|
||||
@@ -177,6 +211,8 @@ describe("connectionSupportsProcessList", () => {
|
||||
expect(connectionSupportsProcessList(conn({ db_type: "gaussdb", driver_profile: "opengauss" }))).toBe(true);
|
||||
expect(connectionSupportsProcessList(conn({ db_type: "gaussdb", driver_profile: "gaussdb" }))).toBe(false);
|
||||
expect(connectionSupportsProcessList(conn({ db_type: "kingbase" }))).toBe(true);
|
||||
expect(resolveProcessListDriver("kingbase")?.buildTerminateSessionSql?.(7)).toBe("SELECT sys_terminate_backend(7)");
|
||||
expect(resolveProcessListDriver("opengauss")?.buildTerminateSessionSql?.(7)).toBe("SELECT pg_terminate_backend(7)");
|
||||
expect(connectionSupportsProcessList(conn({ db_type: "sqlite" }))).toBe(false);
|
||||
expect(connectionSupportsProcessList(undefined)).toBe(false);
|
||||
});
|
||||
|
||||
@@ -137,6 +137,21 @@ export function buildCancelQuerySql(id: number): string {
|
||||
return `KILL QUERY ${id}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the `KILL <id>` statement that disconnects the session itself.
|
||||
*
|
||||
* `KILL QUERY` only interrupts the statement running at that instant, so an idle
|
||||
* (`Sleep`) session survives it and keeps showing up in the process list. When the
|
||||
* user asks to close a session, only this statement actually removes it (and rolls
|
||||
* back the session's uncommitted transaction, the same as the server-side `KILL`).
|
||||
*/
|
||||
export function buildTerminateSessionSql(id: number): string {
|
||||
if (!Number.isInteger(id) || id <= 0) {
|
||||
throw new Error(`Invalid session id: ${id}`);
|
||||
}
|
||||
return `KILL ${id}`;
|
||||
}
|
||||
|
||||
/** Clamp a user-entered refresh interval to a safe integer range of seconds. */
|
||||
export function clampInterval(seconds: number): number {
|
||||
if (!Number.isFinite(seconds)) return DEFAULT_REFRESH_SECONDS;
|
||||
|
||||
@@ -174,6 +174,27 @@ export function buildKingbasePgCancelQuerySql(pid: number): string {
|
||||
return `SELECT pg_cancel_backend(${pid})`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the statement that disconnects the backend instead of only interrupting the
|
||||
* statement it is running right now. `pg_cancel_backend` cancels the current query and
|
||||
* leaves an idle backend alive, so the process list keeps listing it; terminating the
|
||||
* backend is what actually removes the session (rolling back its open transaction).
|
||||
*/
|
||||
export function buildPgTerminateSessionSql(pid: number): string {
|
||||
validateBackendPid(pid);
|
||||
return `SELECT pg_terminate_backend(${pid})`;
|
||||
}
|
||||
|
||||
export function buildKingbaseTerminateSessionSql(pid: number): string {
|
||||
validateBackendPid(pid);
|
||||
return `SELECT sys_terminate_backend(${pid})`;
|
||||
}
|
||||
|
||||
export function buildKingbasePgTerminateSessionSql(pid: number): string {
|
||||
validateBackendPid(pid);
|
||||
return `SELECT pg_terminate_backend(${pid})`;
|
||||
}
|
||||
|
||||
/** 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");
|
||||
@@ -187,11 +208,24 @@ export function kingbasePgCancelQueryResultError(results: QueryResult[]): string
|
||||
return backendCancelQueryResultError(results, "pg_cancel_backend");
|
||||
}
|
||||
|
||||
function backendCancelQueryResultError(results: QueryResult[], functionName: string): string | null {
|
||||
/** Return an error when the server reports that no backend was terminated. */
|
||||
export function pgTerminateSessionResultError(results: QueryResult[]): string | null {
|
||||
return backendCancelQueryResultError(results, "pg_terminate_backend", "did not terminate the session");
|
||||
}
|
||||
|
||||
export function kingbaseTerminateSessionResultError(results: QueryResult[]): string | null {
|
||||
return backendCancelQueryResultError(results, "sys_terminate_backend", "did not terminate the session");
|
||||
}
|
||||
|
||||
export function kingbasePgTerminateSessionResultError(results: QueryResult[]): string | null {
|
||||
return backendCancelQueryResultError(results, "pg_terminate_backend", "did not terminate the session");
|
||||
}
|
||||
|
||||
function backendCancelQueryResultError(results: QueryResult[], functionName: string, failure = "did not cancel a running query"): 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 cancel a running query`;
|
||||
return `${functionName} ${failure}`;
|
||||
}
|
||||
|
||||
/** Detect the undefined-column failure produced by pre-9.6 pg_stat_activity. */
|
||||
@@ -213,3 +247,7 @@ export function isKingbaseOwnSessionCatalogCompatibilityError(error: unknown): b
|
||||
export function isKingbaseCancelCatalogCompatibilityError(error: unknown): boolean {
|
||||
return isMissingKingbaseSysFunction(error, ["sys_cancel_backend"]);
|
||||
}
|
||||
|
||||
export function isKingbaseTerminateCatalogCompatibilityError(error: unknown): boolean {
|
||||
return isMissingKingbaseSysFunction(error, ["sys_terminate_backend"]);
|
||||
}
|
||||
@@ -1,16 +1,22 @@
|
||||
import type { ConnectionConfig, DatabaseType, QueryResult } from "@/types/database";
|
||||
import { effectiveDatabaseTypeForConnection } from "@/lib/database/jdbcDialect";
|
||||
import { buildCancelQuerySql as buildMysqlCancelQuerySql, mapProcessRows as mapMysqlProcessRows, PROCESS_LIST_SQL as MYSQL_PROCESS_LIST_SQL, supportsProcessList as supportsMysqlProcessList } from "./mysqlProcessList";
|
||||
import { buildCancelQuerySql as buildMysqlCancelQuerySql, buildTerminateSessionSql as buildMysqlTerminateSessionSql, mapProcessRows as mapMysqlProcessRows, PROCESS_LIST_SQL as MYSQL_PROCESS_LIST_SQL, supportsProcessList as supportsMysqlProcessList } from "./mysqlProcessList";
|
||||
import {
|
||||
buildKingbaseCancelQuerySql,
|
||||
buildKingbasePgCancelQuerySql,
|
||||
buildKingbasePgTerminateSessionSql,
|
||||
buildKingbaseTerminateSessionSql,
|
||||
buildPgCancelQuerySql,
|
||||
buildPgTerminateSessionSql,
|
||||
isKingbaseOwnSessionCatalogCompatibilityError,
|
||||
isKingbaseProcessListCatalogCompatibilityError,
|
||||
isKingbaseCancelCatalogCompatibilityError,
|
||||
isKingbaseTerminateCatalogCompatibilityError,
|
||||
isPgProcessListCompatibilityError,
|
||||
kingbaseCancelQueryResultError,
|
||||
kingbasePgCancelQueryResultError,
|
||||
kingbasePgTerminateSessionResultError,
|
||||
kingbaseTerminateSessionResultError,
|
||||
KINGBASE_OWN_SESSION_SQL,
|
||||
KINGBASE_PG_OWN_SESSION_SQL,
|
||||
KINGBASE_PG_PROCESS_LIST_SQL,
|
||||
@@ -19,6 +25,7 @@ import {
|
||||
OPENGAUSS_OWN_SESSION_SQL,
|
||||
OPENGAUSS_PROCESS_LIST_SQL,
|
||||
pgCancelQueryResultError,
|
||||
pgTerminateSessionResultError,
|
||||
PG_OWN_SESSION_SQL,
|
||||
PG_PROCESS_LIST_LEGACY_SQL,
|
||||
PG_PROCESS_LIST_SQL,
|
||||
@@ -79,6 +86,19 @@ export interface ProcessListDriver {
|
||||
cancelQueryResultError?(results: QueryResult[]): string | null;
|
||||
/** Validate the success value returned by the compatibility cancellation statement. */
|
||||
fallbackCancelQueryResultError?(results: QueryResult[]): string | null;
|
||||
/**
|
||||
* Build the validated statement that disconnects the session. Engines that omit it
|
||||
* only offer query cancellation, because they cannot close a session over the wire.
|
||||
*/
|
||||
buildTerminateSessionSql?(id: number): string;
|
||||
/** Build the compatibility statement used when the primary terminate function is unavailable. */
|
||||
buildFallbackTerminateSessionSql?(id: number): string;
|
||||
/** Restrict terminate fallback attempts to known compatibility failures. */
|
||||
shouldUseFallbackTerminateSessionSql?(error: unknown): boolean;
|
||||
/** Validate any engine-specific success value returned by the terminate statement. */
|
||||
terminateSessionResultError?(results: QueryResult[]): string | null;
|
||||
/** Validate the success value returned by the compatibility terminate statement. */
|
||||
fallbackTerminateSessionResultError?(results: QueryResult[]): string | null;
|
||||
}
|
||||
|
||||
const MYSQL_COLUMNS: ProcessColumn[] = [
|
||||
@@ -114,6 +134,7 @@ const MYSQL_DRIVER: ProcessListDriver = {
|
||||
// Typed structs carry no index signature; they are plain string-keyed objects at runtime.
|
||||
mapRows: (result) => mapMysqlProcessRows(result) as unknown as ProcessRow[],
|
||||
buildCancelQuerySql: buildMysqlCancelQuerySql,
|
||||
buildTerminateSessionSql: buildMysqlTerminateSessionSql,
|
||||
};
|
||||
|
||||
const POSTGRES_DRIVER: ProcessListDriver = {
|
||||
@@ -127,6 +148,8 @@ const POSTGRES_DRIVER: ProcessListDriver = {
|
||||
mapRows: (result) => mapPgProcessRows(result) as unknown as ProcessRow[],
|
||||
buildCancelQuerySql: buildPgCancelQuerySql,
|
||||
cancelQueryResultError: pgCancelQueryResultError,
|
||||
buildTerminateSessionSql: buildPgTerminateSessionSql,
|
||||
terminateSessionResultError: pgTerminateSessionResultError,
|
||||
};
|
||||
|
||||
const OPENGAUSS_DRIVER: ProcessListDriver = {
|
||||
@@ -138,6 +161,8 @@ const OPENGAUSS_DRIVER: ProcessListDriver = {
|
||||
mapRows: (result) => mapPgProcessRows(result) as unknown as ProcessRow[],
|
||||
buildCancelQuerySql: buildPgCancelQuerySql,
|
||||
cancelQueryResultError: pgCancelQueryResultError,
|
||||
buildTerminateSessionSql: buildPgTerminateSessionSql,
|
||||
terminateSessionResultError: pgTerminateSessionResultError,
|
||||
};
|
||||
|
||||
const KINGBASE_DRIVER: ProcessListDriver = {
|
||||
@@ -156,6 +181,11 @@ const KINGBASE_DRIVER: ProcessListDriver = {
|
||||
shouldUseFallbackCancelQuerySql: isKingbaseCancelCatalogCompatibilityError,
|
||||
cancelQueryResultError: kingbaseCancelQueryResultError,
|
||||
fallbackCancelQueryResultError: kingbasePgCancelQueryResultError,
|
||||
buildTerminateSessionSql: buildKingbaseTerminateSessionSql,
|
||||
buildFallbackTerminateSessionSql: buildKingbasePgTerminateSessionSql,
|
||||
shouldUseFallbackTerminateSessionSql: isKingbaseTerminateCatalogCompatibilityError,
|
||||
terminateSessionResultError: kingbaseTerminateSessionResultError,
|
||||
fallbackTerminateSessionResultError: kingbasePgTerminateSessionResultError,
|
||||
};
|
||||
|
||||
/** Resolve the process-list driver for a connection, or null if unsupported. */
|
||||
|
||||
Reference in new issue
Block a user