Refactor redisReceiver to eliminate global state and prevent memory leaks (#1894)

## Key Improvements

### Architecture Changes
- **Eliminated global state**: Removed shared Redis clients and global promise
  maps that caused memory leaks and cross-request contamination
- **Lazy initialization**: Stream listening now starts only when getValue() is
  called, reducing unnecessary connections
- **Simplified promise management**: Replaced complex dual-lifecycle promise
  handling with a simple value map and promise cache
- **Single-phase stream reading**: Merged checkExistingMessages() and
  setupStreamListener() into one listenToStream() function

### Bug Fixes
- Fixed memory leaks by using Map.delete() instead of setting undefined
- Fixed race conditions in connection handling with proper promise guards
- Improved error handling and propagation throughout the module
- Fixed lastProcessedId tracking to prevent message loss
- Added proper cleanup in destroy() method

Author: AbanoubGhadban

.github/workflows/pro-integration-tests.yml

@@ -207,7 +207,7 @@ jobs:
       - name: Run Rails server in background
         run: |
           cd spec/dummy
-          RAILS_ENV=test rails server &
+          RAILS_ENV="test" rails server &
 
       - name: Wait for Rails server to start
         run: |
@@ -392,7 +392,7 @@ jobs:
       - name: Run Rails server in background
         run: |
           cd spec/dummy
-          RAILS_ENV=test rails server &
+          RAILS_ENV="test" rails server &
 
       - name: Wait for Rails server to start
         run: |

.github/workflows/pro-package-tests.yml

@@ -98,6 +98,17 @@ jobs:
   package-js-tests:
     needs: build-dummy-app-webpack-test-bundles
     runs-on: ubuntu-22.04
+    # Redis service container
+    services:
+      redis:
+        image: cimg/redis:6.2.6
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
     env:
       REACT_ON_RAILS_PRO_LICENSE: ${{ secrets.REACT_ON_RAILS_PRO_LICENSE }}
     steps:

react_on_rails_pro/packages/node-renderer/tests/redisClient.test.ts

@@ -0,0 +1,119 @@
+// This test in only for documenting Redis client usage
+
+import { createClient } from 'redis';
+
+const redisClient = createClient({ url: process.env.REDIS_URL || 'redis://localhost:6379' });
+
+interface RedisStreamMessage {
+  id: string;
+  message: Record<string, string>;
+}
+interface RedisStreamResult {
+  name: string;
+  messages: RedisStreamMessage[];
+}
+
+test('Redis client connects successfully', async () => {
+  await redisClient.connect();
+  expect(redisClient.isOpen).toBe(true);
+  await redisClient.quit();
+});
+
+test('calls connect after quit', async () => {
+  await redisClient.connect();
+  expect(redisClient.isOpen).toBe(true);
+  await redisClient.quit();
+
+  await redisClient.connect();
+  expect(redisClient.isOpen).toBe(true);
+  await redisClient.quit();
+});
+
+test('calls quit before connect is resolved', async () => {
+  const client = createClient({ url: process.env.REDIS_URL || 'redis://localhost:6379' });
+  const connectPromise = client.connect();
+  await client.quit();
+  await connectPromise;
+  expect(client.isOpen).toBe(false);
+});
+
+test('multiple connect calls', async () => {
+  const client = createClient({ url: process.env.REDIS_URL || 'redis://localhost:6379' });
+  const connectPromise1 = client.connect();
+  const connectPromise2 = client.connect();
+  await expect(connectPromise2).rejects.toThrow('Socket already opened');
+  await expect(connectPromise1).resolves.toMatchObject({});
+  expect(client.isOpen).toBe(true);
+  await client.quit();
+});
+
+test('write to stream and read back', async () => {
+  const client = createClient({ url: process.env.REDIS_URL || 'redis://localhost:6379' });
+  await client.connect();
+
+  const streamKey = 'test-stream';
+  await client.del(streamKey);
+  const messageId = await client.xAdd(streamKey, '*', { field1: 'value1' });
+
+  const result = (await client.xRead({ key: streamKey, id: '0-0' }, { COUNT: 1, BLOCK: 2000 })) as
+    | RedisStreamResult[]
+    | null;
+  expect(result).not.toBeNull();
+  expect(result).toBeDefined();
+
+  const [stream] = result!;
+  expect(stream).toBeDefined();
+  expect(stream?.messages.length).toBe(1);
+  const [message] = stream!.messages;
+  expect(message!.id).toBe(messageId);
+  expect(message!.message).toEqual({ field1: 'value1' });
+
+  await client.quit();
+});
+
+test('quit while reading from stream', async () => {
+  const client = createClient({ url: process.env.REDIS_URL || 'redis://localhost:6379' });
+  await client.connect();
+
+  const streamKey = 'test-stream-quit';
+
+  const readPromise = client.xRead({ key: streamKey, id: '$' }, { BLOCK: 0 });
+
+  // Wait a moment to ensure xRead is blocking
+  await new Promise((resolve) => {
+    setTimeout(resolve, 500);
+  });
+
+  client.destroy();
+
+  await expect(readPromise).rejects.toThrow();
+});
+
+it('expire sets TTL on stream', async () => {
+  const client = createClient({ url: process.env.REDIS_URL || 'redis://localhost:6379' });
+  await client.connect();
+
+  const streamKey = 'test-stream-expire';
+  await client.del(streamKey);
+  await client.xAdd(streamKey, '*', { field1: 'value1' });
+
+  const expireResult = await client.expire(streamKey, 1); // 1 second
+  expect(expireResult).toBe(1); // 1 means the key existed and TTL was set
+
+  const ttl1 = await client.ttl(streamKey);
+  expect(ttl1).toBeLessThanOrEqual(1);
+  expect(ttl1).toBeGreaterThan(0);
+
+  const existsBeforeTimeout = await client.exists(streamKey);
+  expect(existsBeforeTimeout).toBe(1); // Key should exist before timeout
+
+  // Wait for 1.1 seconds
+  await new Promise((resolve) => {
+    setTimeout(resolve, 1100);
+  });
+
+  const existsAfterTimeout = await client.exists(streamKey);
+  expect(existsAfterTimeout).toBe(0); // Key should have expired
+
+  await client.quit();
+});

react_on_rails_pro/spec/dummy/client/app/components/ErrorComponent.tsx

@@ -7,6 +7,7 @@ const ErrorComponent = ({ error }: { error: Error }) => {
     <div>
       <h1>Error happened while rendering RSC Page</h1>
       <p>{error.message}</p>
+      <p>{error.stack}</p>
     </div>
   );
 };

react_on_rails_pro/spec/dummy/client/app/ror-auto-load-components/RSCPostsPageOverRedis.jsx

@@ -3,14 +3,14 @@ import RSCPostsPage from '../components/RSCPostsPage/Main';
 import { listenToRequestData } from '../utils/redisReceiver';
 
 const RSCPostsPageOverRedis = ({ requestId, ...props }, railsContext) => {
-  const { getValue, close } = listenToRequestData(requestId);
+  const { getValue, destroy } = listenToRequestData(requestId);
 
   const fetchPosts = () => getValue('posts');
   const fetchComments = (postId) => getValue(`comments:${postId}`);
   const fetchUser = (userId) => getValue(`user:${userId}`);
 
   if ('addPostSSRHook' in railsContext) {
-    railsContext.addPostSSRHook(close);
+    railsContext.addPostSSRHook(destroy);
   }
 
   return () => (

react_on_rails_pro/spec/dummy/client/app/ror-auto-load-components/RedisReceiver.jsx

@@ -34,10 +34,10 @@ const AsyncToggleContainer = async ({ children, childrenTitle, getValue }) => {
 };
 
 const RedisReceiver = ({ requestId, asyncToggleContainer }, railsContext) => {
-  const { getValue, close } = listenToRequestData(requestId);
+  const { getValue, destroy } = listenToRequestData(requestId);
 
   if ('addPostSSRHook' in railsContext) {
-    railsContext.addPostSSRHook(close);
+    railsContext.addPostSSRHook(destroy);
   }
 
   const UsedToggleContainer = asyncToggleContainer ? AsyncToggleContainer : ToggleContainer;