Node: Add FUNCTION KILL command (valkey-io#2114)

* Add `FUNCTION KILL` command.

Signed-off-by: Yury-Fridlyand <yury.fridlyand@improving.com>
Co-authored-by: Guian Gumpac <guian.gumpac@improving.com>
Signed-off-by: lior sventitzky <liorsve@amazon.com>

CHANGELOG.md

@@ -1,4 +1,5 @@
 #### Changes
+* Node: Added FUNCTION KILL command ([#2114](https://github.com/valkey-io/valkey-glide/pull/2114))
 * Node: Added XAUTOCLAIM command ([#2108](https://github.com/valkey-io/valkey-glide/pull/2108))
 * Node: Added XPENDING commands ([#2085](https://github.com/valkey-io/valkey-glide/pull/2085))
 * Node: Added HSCAN command ([#2098](https://github.com/valkey-io/valkey-glide/pull/2098/))

java/integTest/src/test/java/glide/TestUtilities.java

@@ -323,8 +323,8 @@ public static GlideString generateLuaLibCodeBinary(
     }
 
     /**
-     * Create a lua lib with a RO function which runs an endless loop up to timeout sec.<br>
-     * Execution takes at least 5 sec regardless of the timeout configured.<br>
+     * Create a lua lib with a function which runs an endless loop up to timeout sec.<br>
+     * Execution takes at least 5 sec regardless of the timeout configured.
      */
     public static String createLuaLibWithLongRunningFunction(
             String libName, String funcName, int timeout, boolean readOnly) {

node/src/Commands.ts

@@ -2240,6 +2240,11 @@ export function createFunctionStats(): command_request.Command {
     return createCommand(RequestType.FunctionStats, []);
 }
 
+/** @internal */
+export function createFunctionKill(): command_request.Command {
+    return createCommand(RequestType.FunctionKill, []);
+}
+
 /**
  * Represents offsets specifying a string interval to analyze in the {@link BaseClient.bitcount|bitcount} command. The offsets are
  * zero-based indexes, with `0` being the first index of the string, `1` being the next index and so on.

node/src/GlideClient.ts

@@ -34,6 +34,7 @@ import {
     createFlushDB,
     createFunctionDelete,
     createFunctionFlush,
+    createFunctionKill,
     createFunctionList,
     createFunctionLoad,
     createFunctionStats,
@@ -634,6 +635,24 @@ export class GlideClient extends BaseClient {
         return this.createWritePromise(createFunctionStats());
     }
 
+    /**
+     * Kills a function that is currently executing.
+     * `FUNCTION KILL` terminates read-only functions only.
+     *
+     * See https://valkey.io/commands/function-kill/ for details.
+     *
+     * since Valkey version 7.0.0.
+     *
+     * @returns `OK` if function is terminated. Otherwise, throws an error.
+     * @example
+     * ```typescript
+     * await client.functionKill();
+     * ```
+     */
+    public async functionKill(): Promise<"OK"> {
+        return this.createWritePromise(createFunctionKill());
+    }
+
     /**
      * Deletes all the keys of all the existing databases. This command never fails.
      *

node/src/GlideClusterClient.ts

@@ -36,6 +36,7 @@ import {
     createFlushDB,
     createFunctionDelete,
     createFunctionFlush,
+    createFunctionKill,
     createFunctionList,
     createFunctionLoad,
     createFunctionStats,
@@ -960,6 +961,28 @@ export class GlideClusterClient extends BaseClient {
         });
     }
 
+    /**
+     * Kills a function that is currently executing.
+     * `FUNCTION KILL` terminates read-only functions only.
+     *
+     * See https://valkey.io/commands/function-kill/ for details.
+     *
+     * since Valkey version 7.0.0.
+     *
+     * @param route - (Optional) The client will route the command to the nodes defined by `route`.
+     *     If not defined, the command will be routed to all primary nodes.
+     * @returns `OK` if function is terminated. Otherwise, throws an error.
+     * @example
+     * ```typescript
+     * await client.functionKill();
+     * ```
+     */
+    public async functionKill(route?: Routes): Promise<"OK"> {
+        return this.createWritePromise(createFunctionKill(), {
+            route: toProtobufRoute(route),
+        });
+    }
+
     /**
      * Deletes all the keys of all the existing databases. This command never fails.
      *

node/tests/GlideClient.test.ts

@@ -28,6 +28,7 @@ import {
     checkFunctionListResponse,
     checkFunctionStatsResponse,
     convertStringArrayToBuffer,
+    createLuaLibWithLongRunningFunction,
     encodableTransactionTest,
     encodedTransactionTest,
     flushAndCloseClient,
@@ -37,6 +38,7 @@ import {
     parseEndpoints,
     transactionTest,
     validateTransactionResponse,
+    waitForNotBusy,
 } from "./TestUtilities";
 
 /* eslint-disable @typescript-eslint/no-var-requires */
@@ -830,6 +832,155 @@ describe("GlideClient", () => {
         },
     );
 
+    it.each([ProtocolVersion.RESP2, ProtocolVersion.RESP3])(
+        "function kill RO func %p",
+        async (protocol) => {
+            if (cluster.checkIfServerVersionLessThan("7.0.0")) return;
+
+            const config = getClientConfigurationOption(
+                cluster.getAddresses(),
+                protocol,
+                10000,
+            );
+            const client = await GlideClient.createClient(config);
+            const testClient = await GlideClient.createClient(config);
+
+            try {
+                const libName = "function_kill_no_write";
+                const funcName = "deadlock_no_write";
+                const code = createLuaLibWithLongRunningFunction(
+                    libName,
+                    funcName,
+                    6,
+                    true,
+                );
+                expect(await client.functionFlush()).toEqual("OK");
+                // nothing to kill
+                await expect(client.functionKill()).rejects.toThrow(/notbusy/i);
+
+                // load the lib
+                expect(await client.functionLoad(code, true)).toEqual(libName);
+
+                try {
+                    // call the function without await
+                    const promise = testClient
+                        .fcall(funcName, [], [])
+                        .catch((e) =>
+                            expect((e as Error).message).toContain(
+                                "Script killed",
+                            ),
+                        );
+
+                    let killed = false;
+                    let timeout = 4000;
+                    await new Promise((resolve) => setTimeout(resolve, 1000));
+
+                    while (timeout >= 0) {
+                        try {
+                            expect(await client.functionKill()).toEqual("OK");
+                            killed = true;
+                            break;
+                        } catch {
+                            // do nothing
+                        }
+
+                        await new Promise((resolve) =>
+                            setTimeout(resolve, 500),
+                        );
+                        timeout -= 500;
+                    }
+
+                    expect(killed).toBeTruthy();
+                    await promise;
+                } finally {
+                    await waitForNotBusy(client);
+                }
+            } finally {
+                expect(await client.functionFlush()).toEqual("OK");
+                testClient.close();
+                client.close();
+            }
+        },
+    );
+
+    it.each([ProtocolVersion.RESP2, ProtocolVersion.RESP3])(
+        "function kill RW func %p",
+        async (protocol) => {
+            if (cluster.checkIfServerVersionLessThan("7.0.0")) return;
+
+            const config = getClientConfigurationOption(
+                cluster.getAddresses(),
+                protocol,
+                10000,
+            );
+            const client = await GlideClient.createClient(config);
+            const testClient = await GlideClient.createClient(config);
+
+            try {
+                const libName = "function_kill_write";
+                const key = libName;
+                const funcName = "deadlock_write";
+                const code = createLuaLibWithLongRunningFunction(
+                    libName,
+                    funcName,
+                    6,
+                    false,
+                );
+                expect(await client.functionFlush()).toEqual("OK");
+                // nothing to kill
+                await expect(client.functionKill()).rejects.toThrow(/notbusy/i);
+
+                // load the lib
+                expect(await client.functionLoad(code, true)).toEqual(libName);
+
+                let promise = null;
+
+                try {
+                    // call the function without await
+                    promise = testClient.fcall(funcName, [key], []);
+
+                    let foundUnkillable = false;
+                    let timeout = 4000;
+                    await new Promise((resolve) => setTimeout(resolve, 1000));
+
+                    while (timeout >= 0) {
+                        try {
+                            // valkey kills a function with 5 sec delay
+                            // but this will always throw an error in the test
+                            await client.functionKill();
+                        } catch (err) {
+                            // looking for an error with "unkillable" in the message
+                            // at that point we can break the loop
+                            if (
+                                (err as Error).message
+                                    .toLowerCase()
+                                    .includes("unkillable")
+                            ) {
+                                foundUnkillable = true;
+                                break;
+                            }
+                        }
+
+                        await new Promise((resolve) =>
+                            setTimeout(resolve, 500),
+                        );
+                        timeout -= 500;
+                    }
+
+                    expect(foundUnkillable).toBeTruthy();
+                } finally {
+                    // If function wasn't killed, and it didn't time out - it blocks the server and cause rest
+                    // test to fail. Wait for the function to complete (we cannot kill it)
+                    expect(await promise).toContain("Timed out");
+                }
+            } finally {
+                expect(await client.functionFlush()).toEqual("OK");
+                testClient.close();
+                client.close();
+            }
+        },
+    );
+
     it.each([ProtocolVersion.RESP2, ProtocolVersion.RESP3])(
         "sort sortstore sort_store sortro sort_ro sortreadonly test_%p",
         async (protocol) => {