chain pipeline commands

docs/transactions.md

@@ -27,3 +27,27 @@ await multi.execTyped(); // [string]
 You can also [watch](https://redis.io/docs/interact/transactions/#optimistic-locking-using-check-and-set) keys by calling `.watch()`. Your transaction will abort if any of the watched keys change.
 
 The `WATCH` state is stored on the connection (by the server). In case you need to run multiple `WATCH` & `MULTI` in parallel you'll need to use a [pool](./pool.md).
+
+## `execAsPipeline`
+
+`execAsPipeline` will execute the commands without "wrapping" it with `MULTI` & `EXEC` (and lose the transactional semantics).
+
+```javascript
+await client.multi()
+  .get('a')
+  .get('b')
+  .execAsPipeline();
+```
+
+the diffrence between the above pipeline and `Promise.all`:
+
+```javascript
+await Promise.all([
+  client.get('a'),
+  client.get('b')
+]);
+```
+
+is that if the socket disconnects during the pipeline, any unwritten commands will be discarded. i.e. if the socket disconnects after `GET a` is written to the socket, but before `GET b` is:
+- using `Promise.all` - the client will try to execute `GET b` when the socket reconnects
+- using `execAsPipeline` - `GET b` promise will be rejected as well

docs/v4-to-v5.md

@@ -125,6 +125,19 @@ await cluster.multi()
   .exec();
 ```
 
+## `MULTI.execAsPipeline()`
+
+```javascript
+await client.multi()
+  .set('a', 'a')
+  .set('b', 'b')
+  .execAsPipeline();
+```
+
+In older versions, if the socket disconnects during the pipeline execution, i.e. after writing `SET a a` and before `SET b b`, the returned promise is rejected, but `SET b b` will still be executed on the server.
+
+In v5, any unwritten commands (in the same pipeline) will be discarded.
+
 ## Commands
 
 ### Redis

packages/client/lib/client/index.ts

@@ -749,11 +749,13 @@ export default class RedisClient<
       return Promise.reject(new ClientClosedError());
     }
 
-    const promise = Promise.all(
+    const chainId = Symbol('Pipeline Chain'),
-      commands.map(({ args }) => this._self.#queue.addCommand(args, {
+      promise = Promise.all(
-        typeMapping: this._commandOptions?.typeMapping
+        commands.map(({ args }) => this._self.#queue.addCommand(args, {
-      }))
+          chainId,
-    );
+          typeMapping: this._commandOptions?.typeMapping
+        }))
+      );
     this._self.#scheduleWrite();
     const result = await promise;