wip

Author: nkaradzhov

packages/client/lib/client/enterprise-maintenance-manager.ts

@@ -8,9 +8,14 @@ import { setTimeout } from "node:timers/promises";
 import { RedisTcpSocketOptions } from "./socket";
 import diagnostics_channel from "node:diagnostics_channel";
 
-type RedisType = RedisClient<any, any, any, any, any>
+type RedisType = RedisClient<any, any, any, any, any>;
 
 export const SMIGRATED_EVENT = "__SMIGRATED";
+export interface SMigratedEvent {
+  source: { host: string, port: number };
+  destination: { host: string, port: number };
+  ranges: (number | [number, number])[]
+}
 
 export const MAINTENANCE_EVENTS = {
   PAUSE_WRITING: "pause-writing",
@@ -87,7 +92,7 @@ export default class EnterpriseMaintenanceManager {
 
     if (!host) return;
 
-    const tls = options.socket?.tls ?? false
+    const tls = options.socket?.tls ?? false;
 
     const movingEndpointType = await determineEndpoint(tls, host, options);
     return {
@@ -129,12 +134,12 @@ export default class EnterpriseMaintenanceManager {
     const type = String(push[0]);
 
     emitDiagnostics({
-          type,
-          timestamp: Date.now(),
-          data: {
-            push: push.map(String),
-          },
-        });
+      type,
+      timestamp: Date.now(),
+      data: {
+        push: push.map(String),
+      },
+    });
     switch (type) {
       case PN.MOVING: {
         // [ 'MOVING', '17', '15', '54.78.247.156:12075' ]
@@ -223,7 +228,7 @@ export default class EnterpriseMaintenanceManager {
 
     // If the URL is provided, it takes precedense
     // the options object could just be mutated
-    if(this.#options.url) {
+    if (this.#options.url) {
       const u = new URL(this.#options.url);
       u.hostname = host;
       u.port = String(port);
@@ -232,15 +237,17 @@ export default class EnterpriseMaintenanceManager {
       this.#options.socket = {
         ...this.#options.socket,
         host,
-        port
-      }
+        port,
+      };
     }
     const tmpClient = this.#client.duplicate();
-    tmpClient.on('error', (error: unknown) => {
+    tmpClient.on("error", (error: unknown) => {
       //We dont know how to handle tmp client errors
-      dbgMaintenance(`[ERR]`, error)
+      dbgMaintenance(`[ERR]`, error);
     });
-    dbgMaintenance(`Tmp client created in ${( performance.now() - start ).toFixed(2)}ms`);
+    dbgMaintenance(
+      `Tmp client created in ${(performance.now() - start).toFixed(2)}ms`,
+    );
     dbgMaintenance(
       `Set timeout for tmp client to ${this.#options.maintRelaxedSocketTimeout}`,
     );
@@ -251,7 +258,9 @@ export default class EnterpriseMaintenanceManager {
     dbgMaintenance(`Connecting tmp client: ${host}:${port}`);
     start = performance.now();
     await tmpClient.connect();
-    dbgMaintenance(`Connected to tmp client in ${(performance.now() - start).toFixed(2)}ms`);
+    dbgMaintenance(
+      `Connected to tmp client in ${(performance.now() - start).toFixed(2)}ms`,
+    );
     // 3 [EVENT] New socket connected
 
     dbgMaintenance(`Wait for all in-flight commands to complete`);
@@ -297,12 +306,11 @@ export default class EnterpriseMaintenanceManager {
 
     const update: MaintenanceUpdate = {
       relaxedCommandTimeout: undefined,
-      relaxedSocketTimeout: undefined
+      relaxedSocketTimeout: undefined,
     };
 
     this.#client._maintenanceUpdate(update);
   };
-
 }
 
 export type MovingEndpointType =
@@ -341,9 +349,7 @@ async function determineEndpoint(
 ): Promise<MovingEndpointType> {
   assert(options.maintEndpointType !== undefined);
   if (options.maintEndpointType !== "auto") {
-    dbgMaintenance(
-      `Determine endpoint type: ${options.maintEndpointType}`,
-    );
+    dbgMaintenance(`Determine endpoint type: ${options.maintEndpointType}`);
     return options.maintEndpointType;
   }
 

packages/client/lib/cluster/cluster-slots.ts

@@ -7,7 +7,7 @@ import { RedisArgument, RedisFunctions, RedisModules, RedisScripts, RespVersions
 import calculateSlot from 'cluster-key-slot';
 import { RedisSocketOptions } from '../client/socket';
 import { BasicPooledClientSideCache, PooledClientSideCacheProvider } from '../client/cache';
-import { SMIGRATED_EVENT, dbgMaintenance } from '../client/enterprise-maintenance-manager';
+import { SMIGRATED_EVENT, SMigratedEvent, dbgMaintenance } from '../client/enterprise-maintenance-manager';
 
 interface NodeAddress {
   host: string;
@@ -114,7 +114,6 @@ export default class RedisClusterSlots<
   readonly nodeByAddress = new Map<string, MasterNode<M, F, S, RESP, TYPE_MAPPING> | ShardNode<M, F, S, RESP, TYPE_MAPPING>>();
   pubSubNode?: PubSubNode<M, F, S, RESP, TYPE_MAPPING>;
   clientSideCache?: PooledClientSideCacheProvider;
-  smigratedSequenceIdsSeen = new Set<number>();
 
   #isOpen = false;
 
@@ -254,15 +253,74 @@ export default class RedisClusterSlots<
     }
   }
 
-  #handleSmigrated = async (sequenceId: number) => {
-    dbgMaintenance(`[CLUSTER_SLOTS]: handle smigrated`)
-    if(this.smigratedSequenceIdsSeen.has(sequenceId)) {
-      dbgMaintenance(`[CLUSTER_SLOTS]: smigrated sequence id ${sequenceId} already seen, skipping rediscovery`);
+  #handleSmigrated = async (event: SMigratedEvent) => {
+    dbgMaintenance(`[CSlots]: handle smigrated`, event);
+
+    // slots = new Array<Shard<M, F, S, RESP, TYPE_MAPPING>>(RedisClusterSlots.#SLOTS);
+    // masters = new Array<MasterNode<M, F, S, RESP, TYPE_MAPPING>>();
+    // replicas = new Array<ShardNode<M, F, S, RESP, TYPE_MAPPING>>();
+    // readonly nodeByAddress = new Map<string, MasterNode<M, F, S, RESP, TYPE_MAPPING> | ShardNode<M, F, S, RESP, TYPE_MAPPING>>();
+    // pubSubNode?: PubSubNode<M, F, S, RESP, TYPE_MAPPING>;
+
+    const sourceAddress = `${event.source.host}:${event.source.port}`;
+    const sourceNode = this.nodeByAddress.get(sourceAddress);
+    if(!sourceNode) {
+      dbgMaintenance(`[CSlots]: address ${sourceAddress} not in 'nodeByAddress', abort SMIGRATED handling`);
+      return;
+    }
+
+    // 1. Pausing
+    //TODO - check the single pubsubnode
+    sourceNode.client?._pause();
+    if('pubSub' in sourceNode) {
+      sourceNode.pubSub?.client._pause();
+    }
+
+    const destinationAddress = `${event.destination.host}:${event.destination.port}`;
+    let destinationNode = this.nodeByAddress.get(destinationAddress);
+    let destinationShard: Shard<M, F, S, RESP, TYPE_MAPPING>;
+
+    // 2. Create new Master
+    if(!destinationNode) {
+      const promises: Promise<unknown>[] = [];
+      destinationNode = this.#initiateSlotNode({ host: event.destination.host, port: event.destination.port, id: 'asdff' }, false, true, new Set(), promises);
+      await Promise.all(promises);
+      // 2.1 Pause
+      destinationNode.client?._pause();
+      // In case destination node didnt exist, this means Shard didnt exist as well, so creating a new Shard is completely fine
+      destinationShard = {
+        master: destinationNode
+      };
     } else {
-      dbgMaintenance(`[CLUSTER_SLOTS]: smigrated sequence id ${sequenceId} is new, triggering rediscovery`);
-      this.smigratedSequenceIdsSeen.add(sequenceId);
-      await this.#discoverWithRootNodes()
+      // In case destination node existed, this means there was a Shard already, so its best if we can find it.
+      const existingShard = this.slots.find(shard => shard.master.host === event.destination.host && shard.master.port === event.destination.port);
+      if(!existingShard) {
+        dbgMaintenance("Could not find shard");
+        throw new Error('Could not find shard');
+      }
+      destinationShard = existingShard;
     }
+
+    // 3. Soft update shards
+    for(const range of event.ranges) {
+      if(typeof range === 'number') {
+        this.slots[range] = destinationShard;
+      } else {
+        for (let slot = range[0]; slot <= range[1]; slot++) {
+          this.slots[slot] = destinationShard;
+        }
+      }
+    }
+
+    // 4. For all affected clients (normal, pubsub, spubsub):
+    //   4.1 Wait for inflight commands to complete
+    //   4.2 Extract commands, channels, sharded channels
+    //   4.3 Kill if no slots are pointing to it
+    //
+
+    // 5. Prepend extracted commands, chans
+    // 5.1 Unpause
+
   }
 
   async #getShards(rootNode: RedisClusterClientOptions) {