[embeddable] fix race condition in useBatchedPublishingSubjects (#216399)

Closes https://github.com/elastic/kibana/issues/214176 and
https://github.com/elastic/kibana/issues/214853

[upgrade from chrome 134 to 135 in functional test
runner](https://github.com/elastic/kibana/issues/213919) revealed a race
condition in `useBatchedPublishingSubjects` where batched observables
could emit new values before `useEffect` sets up the subscription. This
PR resolves this issue by setting up subscription in useRef, which has
no timing delays.

In chrome 134, `useBatchedPublishingSubjects` `useEffect` gets called
(setting up subscription) before lens embeddable emits any changes to
batched observables.
<img width="300" alt="chrome134"
src="https://github.com/user-attachments/assets/b0356f74-e0c7-4d93-a23a-ace519194d5d"
/>

In chrome 135, `useBatchedPublishingSubjects` `useEffect` gets called
after lens embeddable emits changes to batched observables. This causes
the lens embeddable to not render since the `LensEmbeddableComponent`
has a stale value for `expressionParams`.
<img width="300" alt="chrome135"
src="https://github.com/user-attachments/assets/320bfb7e-8b3f-4b48-a138-1c47c5ff9961"
/>

---------

Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>
Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com>

src/platform/packages/shared/presentation/presentation_publishing/publishing_subject/publishing_batcher.ts

@@ -7,7 +7,7 @@
  * License v3.0 only", or the "Server Side Public License, v 1".
  */
 
-import { useEffect, useRef, useState } from 'react';
+import { useEffect, useMemo, useRef, useState } from 'react';
 import { combineLatest, debounceTime, skip } from 'rxjs';
 import { AnyPublishingSubject, PublishingSubject, UnwrapPublishingSubjectTuple } from './types';
 
@@ -120,22 +120,37 @@ export const useBatchedPublishingSubjects = <
 
   /**
    * Subscribe to all subjects and update the latest values when any of them change.
+   *
+   * Can not set up subscription in useEffect.
+   * useEffect introduces a race condition where subscriptions may emit after values are set with useState
+   * but before subscription is setup in useEffect.
+   *
+   * Can not set up subscription in useRef.
+   * useRef executes initialization logic every render.
    */
-  useEffect(() => {
-    const subscription = combineLatest(subjects)
-      .pipe(
-        // When a new observer subscribes to a BehaviorSubject, it immediately receives the current value. Skip this emit.
-        skip(1),
-        debounceTime(0)
-      )
-      .subscribe((values) => {
-        setLatestPublishedValues(values as UnwrapPublishingSubjectTuple<SubjectsType>);
-      });
-    return () => subscription.unsubscribe();
+  const subscription = useMemo(
+    () =>
+      combineLatest(subjects)
+        .pipe(
+          // When a new observer subscribes to a BehaviorSubject, it immediately receives the current value. Skip this emit.
+          skip(1),
+          debounceTime(0)
+        )
+        .subscribe((values) => {
+          setLatestPublishedValues(values as UnwrapPublishingSubjectTuple<SubjectsType>);
+        }),
     // 'subjects' gets a new reference on each call because of spread
     // Use 'useBatchedOptionalPublishingSubjects' when 'subjects' are expected to change.
     // eslint-disable-next-line react-hooks/exhaustive-deps
-  }, []);
+    []
+  );
+
+  /**
+   * Clean up subscription on unmount.
+   */
+  useEffect(() => {
+    return () => subscription.unsubscribe();
+  }, [subscription]);
 
   return latestPublishedValues;
 };

src/platform/packages/shared/presentation/presentation_publishing/publishing_subject/publishing_subject.test.tsx

@@ -118,6 +118,30 @@ describe('publishing subject', () => {
       expect(renderCount).toBe(2);
     });
 
+    test('useBatchedPublishingSubjects should synchronously subscribe to observables to avoid race conditions', async () => {
+      function Component() {
+        const [value1] = useBatchedPublishingSubjects(subject1);
+
+        // synchronously emit new values for observables
+        // this will cause test to fail if subscriptions are not setup synchronously
+        incrementAll();
+
+        return (
+          <>
+            <span>{`value1: ${value1}`}</span>
+          </>
+        );
+      }
+      render(<Component />);
+      await waitFor(() => {
+        expect(
+          // If there is a race condition, then 'value1: 0' will be rendered
+          // because value1 will have the original value '0' instead of latest value
+          screen.getByText('value1: 1')
+        ).toBeInTheDocument();
+      });
+    });
+
     test('should batch state updates when using useBatchedOptionalPublishingSubjects', async () => {
       let renderCount = 0;
       function Component() {