From 3ea10b41ab38e5a851c1df4536d3792059e2ea85 Mon Sep 17 00:00:00 2001 From: Alex Rickabaugh Date: Tue, 4 Feb 2025 09:32:09 -0800 Subject: [PATCH] fix(core): fix race condition in resource() 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 --- goldens/public-api/core/index.api.md | 9 +- packages/core/rxjs-interop/src/rx_resource.ts | 1 + packages/core/src/resource/api.ts | 17 +- packages/core/src/resource/resource.ts | 169 +++++++++--------- packages/core/test/resource/resource_spec.ts | 44 +++++ 5 files changed, 157 insertions(+), 83 deletions(-) diff --git a/goldens/public-api/core/index.api.md b/goldens/public-api/core/index.api.md index 28197e6d9f02..4e8fde3b2ea0 100644 --- a/goldens/public-api/core/index.api.md +++ b/goldens/public-api/core/index.api.md @@ -1447,6 +1447,7 @@ export type Predicate = (value: T) => boolean; // @public export interface PromiseResourceOptions extends BaseResourceOptions { loader: ResourceLoader; + stream?: never; } // @public @@ -1649,11 +1650,14 @@ export enum ResourceStatus { } // @public -export type ResourceStreamingLoader = (param: ResourceLoaderParams) => PromiseLike = (param: ResourceLoaderParams) => PromiseLike>>; + +// @public (undocumented) +export type ResourceStreamItem = { value: T; } | { error: unknown; -}>>; +}; // @public export const RESPONSE_INIT: InjectionToken; @@ -1771,6 +1775,7 @@ export type StaticProvider = ValueProvider | ExistingProvider | StaticClassProvi // @public export interface StreamingResourceOptions extends BaseResourceOptions { + loader?: never; stream: ResourceStreamingLoader; } diff --git a/packages/core/rxjs-interop/src/rx_resource.ts b/packages/core/rxjs-interop/src/rx_resource.ts index 805eeaa46304..92976f577369 100644 --- a/packages/core/rxjs-interop/src/rx_resource.ts +++ b/packages/core/rxjs-interop/src/rx_resource.ts @@ -47,6 +47,7 @@ export function rxResource(opts: RxResourceOptions): ResourceRef({ ...opts, + loader: undefined, stream: (params) => { let sub: Subscription; diff --git a/packages/core/src/resource/api.ts b/packages/core/src/resource/api.ts index ddb1f95acf20..a6cbab70cb6d 100644 --- a/packages/core/src/resource/api.ts +++ b/packages/core/src/resource/api.ts @@ -169,7 +169,7 @@ export type ResourceLoader = (param: ResourceLoaderParams) => PromiseLi */ export type ResourceStreamingLoader = ( param: ResourceLoaderParams, -) => PromiseLike>; +) => PromiseLike>>; /** * Options to the `resource` function, for creating a resource. @@ -212,6 +212,11 @@ export interface PromiseResourceOptions extends BaseResourceOptions * Loading function which returns a `Promise` of the resource's value for a given request. */ loader: ResourceLoader; + + /** + * Cannot specify `stream` and `loader` at the same time. + */ + stream?: never; } /** @@ -225,9 +230,19 @@ export interface StreamingResourceOptions extends BaseResourceOptions; + + /** + * Cannot specify `stream` and `loader` at the same time. + */ + loader?: never; } /** * @experimental */ export type ResourceOptions = PromiseResourceOptions | StreamingResourceOptions; + +/** + * @experimental + */ +export type ResourceStreamItem = {value: T} | {error: unknown}; diff --git a/packages/core/src/resource/resource.ts b/packages/core/src/resource/resource.ts index 18ba9d0f7d2d..79d889c07f2f 100644 --- a/packages/core/src/resource/resource.ts +++ b/packages/core/src/resource/resource.ts @@ -15,14 +15,15 @@ import { ResourceOptions, ResourceStatus, WritableResource, - ResourceLoader, Resource, ResourceRef, ResourceStreamingLoader, - PromiseResourceOptions, StreamingResourceOptions, + ResourceStreamItem, } from './api'; + import {ValueEqualityFn} from '@angular/core/primitives/signals'; + import {Injector} from '../di/injector'; import {assertInInjectionContext} from '../di/contextual'; import {inject} from '../di/injector_compatibility'; @@ -67,16 +68,30 @@ export function resource(options: ResourceOptions): ResourceRef { - // Error state is defined as Resolved && state.error. - status: Exclude; +interface ResourceProtoState { + extRequest: WrappedRequest; + + // For simplicity, status is internally tracked as a subset of the public status enum. + // Reloading and Error statuses are projected from Loading and Resolved based on other state. + status: ResourceInternalStatus; +} + +interface ResourceState extends ResourceProtoState { previousStatus: ResourceStatus; - stream: Signal<{value: T} | {error: unknown}> | undefined; + stream: Signal> | undefined; } +type WrappedRequest = {request: unknown; reload: number}; + /** * Base class which implements `.value` as a `WritableSignal` by delegating `.set` and `.update`. */ @@ -116,18 +131,18 @@ abstract class BaseWritableResource implements WritableResource { * Implementation for `resource()` which uses a `linkedSignal` to manage the resource's state. */ class ResourceImpl extends BaseWritableResource implements ResourceRef { + private readonly pendingTasks: PendingTasks; + /** * The current state of the resource. Status, value, and error are derived from this. */ private readonly state: WritableSignal>; /** - * Signal of both the request value `R` and a writable `reload` signal that's linked/associated - * to the given request. Changing the value of the `reload` signal causes the resource to reload. + * Combines the current request with a reload counter which allows the resource to be reloaded on + * imperative command. */ - private readonly extendedRequest: Signal<{request: R; reload: WritableSignal}>; - - private readonly pendingTasks: PendingTasks; + private readonly extRequest: WritableSignal; private readonly effectRef: EffectRef; private pendingController: AbortController | undefined; @@ -146,49 +161,48 @@ class ResourceImpl extends BaseWritableResource implements ResourceRef< // `WritableSignal` that delegates to `ResourceImpl.set`. computed( () => { - const stream = this.state()?.stream?.(); - return stream && isResolved(stream) ? stream.value : this.defaultValue; + const streamValue = this.state().stream?.(); + return streamValue && isResolved(streamValue) ? streamValue.value : this.defaultValue; }, {equal}, ), ); - this.pendingTasks = injector.get(PendingTasks); // Extend `request()` to include a writable reload signal. - this.extendedRequest = computed(() => ({ - request: request(), - reload: signal(0), - })); + this.extRequest = linkedSignal({ + source: request, + computation: (request) => ({request, reload: 0}), + }); // The main resource state is managed in a `linkedSignal`, which allows the resource to change // state instantaneously when the request signal changes. - this.state = linkedSignal< - ResourceStatus.Idle | ResourceStatus.Loading | ResourceStatus.Reloading, - ResourceState - >({ - // We use the request (as well as its reload signal) to derive the initial status of the - // resource (Idle, Loading, or Reloading) in response to request changes. From this initial - // status, the resource's effect will then trigger the loader and update to a Resolved or - // Error state as appropriate. - source: () => { - const {request, reload} = this.extendedRequest(); - if (request === undefined || this.destroyed) { - return ResourceStatus.Idle; + this.state = linkedSignal>({ + // Whenever the request changes, + source: this.extRequest, + // Compute the state of the resource given a change in status. + computation: (extRequest, previous) => { + const status = + extRequest.request === undefined ? ResourceStatus.Idle : ResourceStatus.Loading; + if (!previous) { + return { + extRequest, + status, + previousStatus: ResourceStatus.Idle, + stream: undefined, + }; + } else { + return { + extRequest, + status, + previousStatus: projectStatusOfState(previous.value), + // If the request hasn't changed, keep the previous stream. + stream: + previous.value.extRequest.request === extRequest.request + ? previous.value.stream + : undefined, + }; } - return reload() === 0 ? ResourceStatus.Loading : ResourceStatus.Reloading; }, - // Compute the state of the resource given a change in status. - computation: (status, previous) => - ({ - status, - // When the state of the resource changes due to the request, remember the previous status - // for the loader to consider. - previousStatus: computeStatusOfState(previous?.value), - // In `Reloading` state, we keep the previous value if there is one, since the identity of - // the request hasn't changed. Otherwise, we switch back to the default value. - stream: - previous && status === ResourceStatus.Reloading ? previous.value.stream : undefined, - }) satisfies ResourceState, }); this.effectRef = effect(this.loadEffect.bind(this), { @@ -196,16 +210,13 @@ class ResourceImpl extends BaseWritableResource implements ResourceRef< manualCleanup: true, }); + this.pendingTasks = injector.get(PendingTasks); + // Cancel any pending request when the resource itself is destroyed. injector.get(DestroyRef).onDestroy(() => this.destroy()); } - override readonly status = computed(() => { - if (this.state().status !== ResourceStatus.Resolved) { - return this.state().status; - } - return isResolved(this.state().stream!()) ? ResourceStatus.Resolved : ResourceStatus.Error; - }); + override readonly status = computed(() => projectStatusOfState(this.state())); override readonly error = computed(() => { const stream = this.state().stream?.(); @@ -221,9 +232,10 @@ class ResourceImpl extends BaseWritableResource implements ResourceRef< } const current = untracked(this.value); + const state = untracked(this.state); if ( - untracked(this.status) === ResourceStatus.Local && + state.status === ResourceStatus.Local && (this.equal ? this.equal(current, value) : current === value) ) { return; @@ -231,6 +243,7 @@ class ResourceImpl extends BaseWritableResource implements ResourceRef< // Enter Local state with the user-defined value. this.state.set({ + extRequest: state.extRequest, status: ResourceStatus.Local, previousStatus: ResourceStatus.Local, stream: signal({value}), @@ -243,17 +256,13 @@ class ResourceImpl extends BaseWritableResource implements ResourceRef< override reload(): boolean { // We don't want to restart in-progress loads. - const status = untracked(this.status); - if ( - status === ResourceStatus.Idle || - status === ResourceStatus.Loading || - status === ResourceStatus.Reloading - ) { + const {status} = untracked(this.state); + if (status === ResourceStatus.Idle || status === ResourceStatus.Loading) { return false; } - // Increment the reload signal to trigger the `state` linked signal to switch us to `Reload` - untracked(this.extendedRequest).reload.update((v) => v + 1); + // Increment the request reload to trigger the `state` linked signal to switch us to `Reload` + this.extRequest.update(({request, reload}) => ({request, reload: reload + 1})); return true; } @@ -264,6 +273,7 @@ class ResourceImpl extends BaseWritableResource implements ResourceRef< // Destroyed resources enter Idle state. this.state.set({ + extRequest: {request: undefined, reload: 0}, status: ResourceStatus.Idle, previousStatus: ResourceStatus.Idle, stream: undefined, @@ -271,25 +281,17 @@ class ResourceImpl extends BaseWritableResource implements ResourceRef< } private async loadEffect(): Promise { + const extRequest = this.extRequest(); + // Capture the previous status before any state transitions. Note that this is `untracked` since // we do not want the effect to depend on the state of the resource, only on the request. const {status: currentStatus, previousStatus} = untracked(this.state); - const {request, reload: reloadCounter} = this.extendedRequest(); - // Subscribe side-effectfully to `reloadCounter`, although we don't actually care about its - // value. This is used to rerun the effect when `reload()` is triggered. - reloadCounter(); - - if (request === undefined) { + if (extRequest.request === undefined) { // Nothing to load (and we should already be in a non-loading state). return; - } else if ( - currentStatus !== ResourceStatus.Loading && - currentStatus !== ResourceStatus.Reloading - ) { - // We might've transitioned into a loading state, but has since been overwritten (likely via - // `.set`). - // In this case, the resource has nothing to do. + } else if (currentStatus !== ResourceStatus.Loading) { + // We're not in a loading or reloading state, so this loading request is stale. return; } @@ -316,7 +318,7 @@ class ResourceImpl extends BaseWritableResource implements ResourceRef< // which side of the `await` they are. const stream = await untracked(() => { return this.loaderFn({ - request: request as Exclude, + request: extRequest.request as Exclude, abortSignal, previous: { status: previousStatus, @@ -324,21 +326,25 @@ class ResourceImpl extends BaseWritableResource implements ResourceRef< }); }); - if (abortSignal.aborted) { + // If this request has been aborted, or the current request no longer + // matches this load, then we should ignore this resolution. + if (abortSignal.aborted || untracked(this.extRequest) !== extRequest) { return; } this.state.set({ + extRequest, status: ResourceStatus.Resolved, previousStatus: ResourceStatus.Resolved, stream, }); } catch (err) { - if (abortSignal.aborted) { + if (abortSignal.aborted || untracked(this.extRequest) !== extRequest) { return; } this.state.set({ + extRequest, status: ResourceStatus.Resolved, previousStatus: ResourceStatus.Error, stream: signal({error: err}), @@ -387,17 +393,20 @@ function isStreamingResourceOptions( return !!(options as StreamingResourceOptions).stream; } -function computeStatusOfState(state: ResourceState | undefined): ResourceStatus { - switch (state?.status) { - case undefined: - return ResourceStatus.Idle; +/** + * Project from a state with `ResourceInternalStatus` to the user-facing `ResourceStatus` + */ +function projectStatusOfState(state: ResourceState): ResourceStatus { + switch (state.status) { + case ResourceStatus.Loading: + return state.extRequest.reload === 0 ? ResourceStatus.Loading : ResourceStatus.Reloading; case ResourceStatus.Resolved: return isResolved(untracked(state.stream!)) ? ResourceStatus.Resolved : ResourceStatus.Error; default: - return state!.status; + return state.status; } } -function isResolved(state: {value: T} | {error: unknown}): state is {value: T} { +function isResolved(state: ResourceStreamItem): state is {value: T} { return (state as {error: unknown}).error === undefined; } diff --git a/packages/core/test/resource/resource_spec.ts b/packages/core/test/resource/resource_spec.ts index ef04b4d92385..8feecbefd9d2 100644 --- a/packages/core/test/resource/resource_spec.ts +++ b/packages/core/test/resource/resource_spec.ts @@ -180,6 +180,50 @@ describe('resource', () => { expect(echoResource.error()).toEqual(Error('KO')); }); + it('should respond to a request that changes while loading', async () => { + const appRef = TestBed.inject(ApplicationRef); + + const request = signal(0); + let resolve: Array<() => void> = []; + const res = resource({ + request, + loader: async ({request}) => { + const p = Promise.withResolvers(); + resolve.push(() => p.resolve(request)); + return p.promise; + }, + injector: TestBed.inject(Injector), + }); + + // Start the load by running the effect inside the resource. + appRef.tick(); + + // We should have a pending load. + expect(resolve.length).toBe(1); + + // Change the request. + request.set(1); + + // Resolve the first load. + resolve[0](); + await flushMicrotasks(); + + // The resource should still be loading. Ticking (triggering the 2nd effect) + // should not change the loading status. + expect(res.status()).toBe(ResourceStatus.Loading); + appRef.tick(); + expect(res.status()).toBe(ResourceStatus.Loading); + expect(resolve.length).toBe(2); + + // Resolve the second load. + resolve[1]?.(); + await flushMicrotasks(); + + // We should see the resolved value. + expect(res.status()).toBe(ResourceStatus.Resolved); + expect(res.value()).toBe(1); + }); + it('should return a default value if provided', async () => { const DEFAULT: string[] = []; const request = signal(0);