Add reconnect method to Client interface (#322)

Context:

Occasionally a client can experience a `stream not found` error when
turso-server is restarted with an existing ongoing stateful `stream`,
that has a baton referencing a different `ConnectionManager` instance.

There is currently no way of resetting the connection on the client
side, without calling `close`. After calling `close` a new Client
instance must be created because there is no way of re-opening the
connection. This adds a `reconnect` method to the Client interface to
allow for this.

Author: penberg

packages/libsql-client-wasm/src/wasm.ts

@@ -219,10 +219,22 @@ export class Sqlite3Client implements Client {
         );
     }
 
+    async reconnect(): Promise<void> {
+        try {
+            if (!this.closed && this.#db !== null) {
+                this.#db.close();
+            }
+        } finally {
+            this.#db = new this.#sqlite3.oo1.DB(this.#path, "c");
+            this.closed = false;
+        }
+    }
+
     close(): void {
         this.closed = true;
         if (this.#db !== null) {
             this.#db.close();
+            this.#db = null;
         }
     }
 

packages/libsql-client/src/http.ts

@@ -74,6 +74,10 @@ const sqlCacheCapacity = 30;
 export class HttpClient implements Client {
     #client: hrana.HttpClient;
     protocol: "http";
+    #url: URL;
+    #intMode: IntMode;
+    #customFetch: Function | undefined;
+    #concurrency: number;
     #authToken: string | undefined;
     #promiseLimitFunction: ReturnType<typeof promiseLimit<any>>;
 
@@ -85,11 +89,20 @@ export class HttpClient implements Client {
         customFetch: Function | undefined,
         concurrency: number,
     ) {
-        this.#client = hrana.openHttp(url, authToken, customFetch);
-        this.#client.intMode = intMode;
-        this.protocol = "http";
+        this.#url = url;
         this.#authToken = authToken;
-        this.#promiseLimitFunction = promiseLimit<any>(concurrency);
+        this.#intMode = intMode;
+        this.#customFetch = customFetch;
+        this.#concurrency = concurrency;
+
+        this.#client = hrana.openHttp(
+            this.#url,
+            this.#authToken,
+            this.#customFetch,
+        );
+        this.#client.intMode = this.#intMode;
+        this.protocol = "http";
+        this.#promiseLimitFunction = promiseLimit<any>(this.#concurrency);
     }
 
     private async limit<T>(fn: () => Promise<T>): Promise<T> {
@@ -267,6 +280,23 @@ export class HttpClient implements Client {
         this.#client.close();
     }
 
+    async reconnect(): Promise<void> {
+        try {
+            if (!this.closed) {
+                // Abort in-flight ops and free resources
+                this.#client.close();
+            }
+        } finally {
+            // Recreate the underlying hrana client
+            this.#client = hrana.openHttp(
+                this.#url,
+                this.#authToken,
+                this.#customFetch,
+            );
+            this.#client.intMode = this.#intMode;
+        }
+    }
+
     get closed(): boolean {
         return this.#client.closed;
     }

packages/libsql-client/src/sqlite3.ts

@@ -222,10 +222,22 @@ export class Sqlite3Client implements Client {
         } as Replicated;
     }
 
+    async reconnect(): Promise<void> {
+        try {
+            if (!this.closed && this.#db !== null) {
+                this.#db.close();
+            }
+        } finally {
+            this.#db = new Database(this.#path, this.#options);
+            this.closed = false;
+        }
+    }
+
     close(): void {
         this.closed = true;
         if (this.#db !== null) {
             this.#db.close();
+            this.#db = null;
         }
     }
 

packages/libsql-client/src/ws.ts

@@ -403,6 +403,35 @@ export class WsClient implements Client {
         }
     }
 
+    async reconnect(): Promise<void> {
+        try {
+            for (const st of Array.from(this.#connState.streamStates)) {
+                try {
+                    st.stream.close();
+                } catch {}
+            }
+            this.#connState.client.close();
+        } catch {}
+
+        if (this.#futureConnState) {
+            try {
+                this.#futureConnState.client.close();
+            } catch {}
+            this.#futureConnState = undefined;
+        }
+
+        const next = this.#openConn();
+        const version = await next.client.getVersion();
+
+        next.useSqlCache = version >= 2;
+        if (next.useSqlCache) {
+            next.sqlCache.capacity = sqlCacheCapacity;
+        }
+
+        this.#connState = next;
+        this.closed = false;
+    }
+
     _closeStream(streamState: StreamState): void {
         streamState.stream.close();
 
@@ -421,6 +450,13 @@ export class WsClient implements Client {
     close(): void {
         this.#connState.client.close();
         this.closed = true;
+        if (this.#futureConnState) {
+            try {
+                this.#futureConnState.client.close();
+            } catch {}
+            this.#futureConnState = undefined;
+        }
+        this.closed = true;
     }
 }
 

packages/libsql-core/src/api.ts

@@ -240,6 +240,10 @@ export interface Client {
      */
     close(): void;
 
+    /** Reconnect after the client has been closed.
+     */
+    reconnect(): void;
+
     /** Is the client closed?
      *
      * This is set to `true` after a call to {@link close} or if the client encounters an unrecoverable