fix(core): fix race condition in resource() (#59851)

The refactoring of `resource()` to use `linkedSignal()` introduced the
potential for a race condition where resources would get stuck and not update
in response to a request change. This occurred under a specific condition:

1. The request changes while the resource is still in loading state
2. The resource resolves the previous load before its `effect()` reacts to the
   request change.

In practice, the window for this race is small, because the request change in
(1) will schedule the effect in (2) immediately. However, it's easier to
trigger this sequencing in tests, especially when one resource depends on the
output of another.

To fix the race condition, the resource impl is refactored to track the request
in its state, and ignore resolved values or streams for stale requests. This
refactoring actually makes the resource code simpler and easier to follow as
well.

Fixes #59842

PR Close #59851

goldens/public-api/core/index.api.md

@@ -1447,6 +1447,7 @@ export type Predicate<T> = (value: T) => boolean;
 // @public
 export interface PromiseResourceOptions<T, R> extends BaseResourceOptions<T, R> {
     loader: ResourceLoader<T, R>;
+    stream?: never;
 }
 
 // @public
@@ -1649,11 +1650,14 @@ export enum ResourceStatus {
 }
 
 // @public
-export type ResourceStreamingLoader<T, R> = (param: ResourceLoaderParams<R>) => PromiseLike<Signal<{
+export type ResourceStreamingLoader<T, R> = (param: ResourceLoaderParams<R>) => PromiseLike<Signal<ResourceStreamItem<T>>>;
+
+// @public (undocumented)
+export type ResourceStreamItem<T> = {
     value: T;
 } | {
     error: unknown;
-}>>;
+};
 
 // @public
 export const RESPONSE_INIT: InjectionToken<ResponseInit | null>;
@@ -1771,6 +1775,7 @@ export type StaticProvider = ValueProvider | ExistingProvider | StaticClassProvi
 
 // @public
 export interface StreamingResourceOptions<T, R> extends BaseResourceOptions<T, R> {
+    loader?: never;
     stream: ResourceStreamingLoader<T, R>;
 }
 

packages/core/rxjs-interop/src/rx_resource.ts

@@ -47,6 +47,7 @@ export function rxResource<T, R>(opts: RxResourceOptions<T, R>): ResourceRef<T |
   opts?.injector || assertInInjectionContext(rxResource);
   return resource<T, R>({
     ...opts,
+    loader: undefined,
     stream: (params) => {
       let sub: Subscription;
 

packages/core/src/resource/api.ts

@@ -169,7 +169,7 @@ export type ResourceLoader<T, R> = (param: ResourceLoaderParams<R>) => PromiseLi
  */
 export type ResourceStreamingLoader<T, R> = (
   param: ResourceLoaderParams<R>,
-) => PromiseLike<Signal<{value: T} | {error: unknown}>>;
+) => PromiseLike<Signal<ResourceStreamItem<T>>>;
 
 /**
  * Options to the `resource` function, for creating a resource.
@@ -212,6 +212,11 @@ export interface PromiseResourceOptions<T, R> extends BaseResourceOptions<T, R>
    * Loading function which returns a `Promise` of the resource's value for a given request.
    */
   loader: ResourceLoader<T, R>;
+
+  /**
+   * Cannot specify `stream` and `loader` at the same time.
+   */
+  stream?: never;
 }
 
 /**
@@ -225,9 +230,19 @@ export interface StreamingResourceOptions<T, R> extends BaseResourceOptions<T, R
    * request, which can change over time as new values are received from a stream.
    */
   stream: ResourceStreamingLoader<T, R>;
+
+  /**
+   * Cannot specify `stream` and `loader` at the same time.
+   */
+  loader?: never;
 }
 
 /**
  * @experimental
  */
 export type ResourceOptions<T, R> = PromiseResourceOptions<T, R> | StreamingResourceOptions<T, R>;
+
+/**
+ * @experimental
+ */
+export type ResourceStreamItem<T> = {value: T} | {error: unknown};