From 78ff8914fdc48485bf58e11c772715adf7116a9b Mon Sep 17 00:00:00 2001 From: Michael Zabka Date: Thu, 16 Jul 2026 14:00:16 +0200 Subject: [PATCH 1/2] Improve saga dispatch concurrency Index combined sagas by handled action type, run matching leaf sagas independently, and track action completion without mutating frozen actions. Clean up finished iterators and observable subscriptions while preserving legacy action promise compatibility. Ported from signageOS Box MR 3682 commit a304d1d005f846f00261b402aff0a912653ba128. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 7 + README.md | 4 + src/ActionPromise.ts | 57 ++++ src/ISaga.ts | 11 + src/Profiler/profiler.ts | 37 ++- src/SagaGroup.ts | 34 +++ src/combineSagas.ts | 289 ++++++++++++++------ src/createModelSaga.ts | 206 ++++++++++---- src/index.ts | 2 + tests/integration/Profiler/profiler.spec.ts | 18 +- tests/integration/index.spec.ts | 10 + tests/unit/ActionPromise.spec.ts | 55 ++++ tests/unit/combineSagas.spec.ts | 82 +++++- tests/unit/createModelSaga.spec.ts | 196 +++++++++++++ tests/unit/sumModelMock.ts | 2 + 15 files changed, 853 insertions(+), 157 deletions(-) create mode 100644 src/ActionPromise.ts create mode 100644 src/SagaGroup.ts create mode 100644 tests/unit/ActionPromise.spec.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 43e9c4c..e7b7db7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,13 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html). +## [Unreleased] +### Changed +- Require sagas to declare handled action types and run matching sagas concurrently + +### Fixed +- Clean up completed iterators and observable subscriptions + ## [2.2.0] ### Added - Simple profiler options for saga time usage diff --git a/README.md b/README.md index 727eec8..e31c317 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ const wait = (timeout) => new Promise((resolve) => setTimeout(resolve, timeout)) const { createModelSaga, put } = require('aperol'); const appReducer = require('./appReducer'); const appSaga = { + actionTypes: ['SET_GREETING', 'GREET_WITH_DELAY'], reducer(model, action) { switch (action.type) { case 'SET_GREETING': @@ -52,6 +53,7 @@ function repeat(interval) { }); } const appSaga = { + actionTypes: ['GREET', ObservableSubscribed, 'GREET_REPEATABLE', 'STOP_GREETING'], reducer(model, action) { switch (action.type) { case 'GREET': @@ -89,6 +91,8 @@ const appSaga = { // ... app code ``` +Every saga must declare the action types it handles. Use `ANY_ACTION` in `actionTypes` when a saga needs to process every action. + ### Combining more sagas diff --git a/src/ActionPromise.ts b/src/ActionPromise.ts new file mode 100644 index 0000000..9c33cc1 --- /dev/null +++ b/src/ActionPromise.ts @@ -0,0 +1,57 @@ +import { Action } from 'redux'; + +interface ILegacyPromiseAction extends Action { + __promise?: Promise; +} + +/** + * Side-channel holding the "this action's sagas have finished" promise for each dispatched action. + * + * The promise used to be written onto the action object itself as a non-configurable, non-writable + * `__promise`. That meant re-dispatching the same action object threw "Cannot redefine property", + * and dispatching a frozen action (any freezing middleware, common in development) threw + * "object is not extensible". A WeakMap keyed on the action has neither problem and does not mutate + * user-owned objects; entries disappear with the action itself. + */ +const promisesByAction = new WeakMap>(); + +/** @deprecated Prefer `getActionPromise(action)`. Kept so existing `action.__promise` code works. */ +function attachLegacyProperty(action: object, promise: Promise) { + try { + Object.defineProperty(action, '__promise', { + enumerable: false, + configurable: true, + writable: true, + value: promise + }); + } catch { + // Frozen or sealed action. The WeakMap entry still works, so this is not fatal. + } +} + +export function setActionPromise(action: unknown, promise: Promise) { + if (typeof action !== 'object' || action === null) { + return; + } + promisesByAction.set(action, promise); + attachLegacyProperty(action, promise); +} + +/** + * The promise that settles once every saga reacting to this action has finished, or `undefined` if + * the action did not pass through a model saga (e.g. another middleware swallowed it). + */ +export function getActionPromise(action: unknown): Promise | undefined { + if (typeof action !== 'object' || action === null) { + return undefined; + } + const promise = promisesByAction.get(action); + if (promise) { + return promise; + } + // An action that crossed a store boundary may only carry the legacy property. + const legacy = (action as ILegacyPromiseAction).__promise; + return legacy instanceof Promise ? legacy : undefined; +} + +export type { Action }; diff --git a/src/ISaga.ts b/src/ISaga.ts index 0792bed..3982e60 100644 --- a/src/ISaga.ts +++ b/src/ISaga.ts @@ -5,5 +5,16 @@ import IUpdaterYield from './IUpdaterYield'; interface ISaga { reducer(model: TModel, action: Action): TModel; updater(model: TModel, action: Action): Iterator | AsyncIterator; + /** + * Every action type this saga reacts to, in EITHER its reducer or its updater. + * + * `combineSagas` will not run this saga at all for any other action type, which is what keeps + * dispatch cost proportional to the sagas that actually react rather than to the total number of + * sagas. The declaration must therefore cover the reducer as well: if a saga reduces on a type it + * did not declare, that reduction will not happen. + * + * Declare `[ANY_ACTION]` for a saga that genuinely reacts to every action. + */ + actionTypes: readonly string[]; } export default ISaga; diff --git a/src/Profiler/profiler.ts b/src/Profiler/profiler.ts index ed7ff0c..473ffec 100644 --- a/src/Profiler/profiler.ts +++ b/src/Profiler/profiler.ts @@ -1,7 +1,7 @@ import { Action } from "redux"; import AnyIterator from "../AnyIterator"; -type OnWarning = (message: string, action: Action) => void; +type OnWarning = (message: string, action: Action, sagaKey: string) => void; export interface IProfilerOptions { thresholdMs?: number; @@ -12,6 +12,7 @@ export interface IProfiler { track( iterator: AnyIterator, sourceAction: Action, + sagaKey: string, ): ITracking; } @@ -19,21 +20,45 @@ export interface ITracking { stop(): void; } +/** Shared no-op, so a profiler configured without a threshold costs nothing per tracked update. */ +const NO_TRACKING: ITracking = { + stop() { + // Nothing to measure. + } +}; + export function startProfiler(options: IProfilerOptions): IProfiler { - const onWarning = options.onWarning ?? console.warn.bind(console); + const onWarning = options.onWarning ?? ((message: string, action: Action) => console.warn(message, action)); + const thresholdMs = options.thresholdMs; + + if (!thresholdMs) { + return { + track() { + return NO_TRACKING; + } + }; + } + return { track( _iterator: AnyIterator, sourceAction: Action, + sagaKey: string, ) { - const startTime = new Date().valueOf(); + // Date.now() rather than new Date().valueOf(), which allocated a Date object on both the + // start and the stop of every tracked update. + const startTime = Date.now(); return { stop() { - const time = new Date().valueOf() - startTime; - if (options.thresholdMs && time >= options.thresholdMs) { + const time = Date.now() - startTime; + if (time >= thresholdMs) { + // Naming the saga is the point: at a few hundred sagas, knowing only that + // *an* action was slow does not tell you which saga to go and look at. + const saga = sagaKey ? ` saga=${sagaKey}` : ''; onWarning( - `The threshold of profiler has been reached: time=${time}ms`, + `The threshold of profiler has been reached:${saga} time=${time}ms`, sourceAction, + sagaKey, ); } }, diff --git a/src/SagaGroup.ts b/src/SagaGroup.ts new file mode 100644 index 0000000..3ee1c73 --- /dev/null +++ b/src/SagaGroup.ts @@ -0,0 +1,34 @@ +import { Action } from 'redux'; +import AnyIterator from './AnyIterator'; + +/** The action type a saga declares when it genuinely needs to react to every action. */ +export const ANY_ACTION = '*'; + +/** + * Marks a saga produced by `combineSagas`. + * + * A combined saga could always expose its children as one flattened iterator, and it still does for + * anything that treats it as a plain `ISaga`. But flattening forces every child's yields through a + * single queue, which `createModelSaga` then drains one at a time — so a slow `await` in one saga + * delays every other saga's `put()`. Recognising the group lets `createModelSaga` drive each matching + * child independently instead. + */ +export const MATCH_SAGAS: unique symbol = Symbol.for('aperol.matchSagas') as any; + +export interface IMatchedSaga { + /** Path of the saga within the combined tree, e.g. `math.sum`. Empty for a non-combined saga. */ + key: string; + iterator: AnyIterator; +} + +export interface ISagaGroup { + /** + * Appends one entry per child that reacts to `action`, recursing into nested groups so that each + * entry is a leaf saga rather than another multiplexer. + */ + [MATCH_SAGAS](model: TModel, action: Action, keyPrefix: string, matched: IMatchedSaga[]): void; +} + +export function isSagaGroup(saga: unknown): saga is ISagaGroup { + return typeof (saga as ISagaGroup | undefined)?.[MATCH_SAGAS] === 'function'; +} diff --git a/src/combineSagas.ts b/src/combineSagas.ts index e98f624..274187a 100644 --- a/src/combineSagas.ts +++ b/src/combineSagas.ts @@ -1,9 +1,9 @@ - -import { combineReducers, ReducersMapObject } from 'redux'; -import ISaga from './ISaga'; import { Action } from 'redux'; +import AnyIterator from './AnyIterator'; +import ISaga from './ISaga'; import IUpdaterYield from './IUpdaterYield'; import { createDeferred, IDeferred } from './Promise/deferred'; +import { ANY_ACTION, IMatchedSaga, isSagaGroup, ISagaGroup, MATCH_SAGAS } from './SagaGroup'; export interface ISagasMapObject { [key: string]: ISaga; @@ -13,100 +13,229 @@ export interface ICombinedModel { [key: string]: any; } +const DONE_RESULT: IteratorResult = { value: undefined, done: true }; + +/** Shared, stateless iterator for actions no saga reacts to, so they allocate nothing. */ +const DONE_ITERATOR: AsyncIterator = { + next() { + return Promise.resolve(DONE_RESULT); + } +}; + +/** + * Drives one child saga's iterator, pushing each yielded value into the multiplexer. + * + * Yielded promises are awaited here before being emitted. That is not an optimisation but a + * behavioural requirement: children used to be wrapped in an `async function*`, and `yield x` inside + * an async generator implicitly performs `Await(x)`. Replicating that await explicitly is what keeps + * `const x = yield somePromise` receiving the resolved value. + */ +async function driveChild( + iterator: AnyIterator, + emit: (value: IUpdaterYield) => void, + isCancelled: () => boolean, +) { + let feedback: IUpdaterYield | undefined = undefined; + while (!isCancelled()) { + const item: IteratorResult = await iterator.next(feedback); + if (item.done) { + return; + } + const value: IUpdaterYield = item.value instanceof Promise ? await item.value : item.value; + if (isCancelled()) { + break; + } + emit(value); + feedback = value; + } + await iterator.return?.(undefined); +} + export default function combineSagas( sagas: ISagasMapObject -): ISaga { +): ISaga & ISagaGroup { const sagaKeys = Object.keys(sagas); - const reducer = combineReducers(sagaKeys.reduce( - (reducers: ReducersMapObject, key: string) => { - const saga = sagas[key]; - reducers[key] = saga.reducer; - return reducers; - }, - {} - )); - const updater = function (model: TModel, action: Action) { - let nextDeferred: IDeferred | undefined = undefined; - let done = false; - let errorsQueue: Error[] = []; - const valuesQueue: unknown[] = []; + for (const key of sagaKeys) { + if (!Array.isArray(sagas[key].actionTypes)) { + throw new Error( + `Saga "${key}" is missing the required "actionTypes" declaration. List every action type ` + + `it reacts to, in either its reducer or its updater, or declare ["${ANY_ACTION}"] if it ` + + `genuinely reacts to every action.` + ); + } + } - function doYield(value: unknown) { - valuesQueue.push(value); - nextDeferred?.resolve(); + const wildcardKeys = sagaKeys.filter((key: string) => sagas[key].actionTypes.indexOf(ANY_ACTION) !== -1); + + // Resolved once at combine time: type -> (sagas declaring it + every wildcard saga). A type that + // is absent from the index can only be matched by wildcards, so the lookup needs no per-action + // concat and no unbounded cache. + const keysByActionType = new Map(); + for (const key of sagaKeys) { + for (const actionType of sagas[key].actionTypes) { + if (actionType === ANY_ACTION) { + continue; + } + const existing = keysByActionType.get(actionType); + keysByActionType.set(actionType, existing ? [...existing, key] : [key]); } - function doneYield() { - done = true; - nextDeferred?.resolve(); + } + for (const actionType of Array.from(keysByActionType.keys())) { + keysByActionType.set(actionType, [...keysByActionType.get(actionType)!, ...wildcardKeys]); + } + + function matchingKeys(actionType: string) { + return keysByActionType.get(actionType) ?? wildcardKeys; + } + + const reducer = (model: TModel | undefined, action: Action): TModel => { + if (model === undefined) { + // Store initialisation: every child must run to contribute its default slice. + const initialModel: ICombinedModel = {}; + for (const key of sagaKeys) { + initialModel[key] = sagas[key].reducer(undefined, action); + } + return initialModel as TModel; } - function errYield(error: Error) { - errorsQueue.push(error); - nextDeferred?.resolve(); + + const keys = matchingKeys(action.type); + if (keys.length === 0) { + return model; } - const combinedIterator: AsyncIterator = { - async next(...args: [] | [undefined]): Promise> { - if (errorsQueue.length > 0) { - throw errorsQueue.shift(); - } - if (valuesQueue.length > 0) { - return { - value: valuesQueue.shift(), - done: false, - }; - } - if (done) { - return { - value: undefined, - done: true, - }; + // Copy lazily: an action that changes no slice returns the *same* model object, so no garbage + // is produced and referential identity survives for downstream memoisation. + let nextModel: TModel | undefined = undefined; + for (const key of keys) { + const previousSlice = model[key]; + const nextSlice = sagas[key].reducer(previousSlice, action); + if (nextSlice !== previousSlice) { + if (!nextModel) { + nextModel = Object.assign({}, model); } - if (!nextDeferred) { - nextDeferred = createDeferred(); - } - await nextDeferred.promise; - nextDeferred = undefined; - return this.next(...args); - }, - }; + (nextModel as ICombinedModel)[key] = nextSlice; + } + } + return nextModel ?? model; + }; - const invoke = async function* (key: string) { + /** + * Collects one entry per reacting leaf saga. `createModelSaga` gives each of them its own drive + * loop, so they run concurrently rather than being serialised behind one another. + */ + const matchSagas = (model: TModel, action: Action, keyPrefix: string, matched: IMatchedSaga[]) => { + for (const key of matchingKeys(action.type)) { const saga = sagas[key]; - const iterator = saga.updater(model[key], action); - let nextResult: undefined; - do { - let item: IteratorResult = await iterator.next(nextResult); - if (item.done) { - break; - } - nextResult = yield item.value; - } while (true); - }; - - Promise.allSettled(sagaKeys.map(async (sagaKey: string) => { - const generator = invoke(sagaKey); - let nextResult: any; - try { - do { - let item: IteratorResult = await generator.next(nextResult); - if (item.done) { - break; - } - nextResult = item.value; - doYield(item.value); - } while (true); - } catch (error) { - errYield(error); - throw error; + const childModel = model[key]; + if (isSagaGroup(saga)) { + saga[MATCH_SAGAS](childModel, action, `${keyPrefix}${key}.`, matched); + } else { + matched.push({ + key: `${keyPrefix}${key}`, + iterator: saga.updater(childModel, action) + }); + } + } + }; + + /** + * The flattened `ISaga` view, kept so a combined saga still satisfies the plain saga contract. + * `createModelSaga` does not use it — it takes the concurrent path above. + */ + const updater = function (model: TModel, action: Action): AnyIterator { + const matched: IMatchedSaga[] = []; + matchSagas(model, action, '', matched); + + if (matched.length === 0) { + return DONE_ITERATOR; + } + if (matched.length === 1) { + return matched[0].iterator; + } + + const valuesQueue: IUpdaterYield[] = []; + const errorsQueue: Error[] = []; + let pending = matched.length; + let cancelled = false; + let nextDeferred: IDeferred | undefined = undefined; + const isCancelled = () => cancelled; + + function wake() { + const deferred = nextDeferred; + nextDeferred = undefined; + deferred?.resolve(); + } + + function emit(value: IUpdaterYield) { + valuesQueue.push(value); + wake(); + } + + // A settle counter, not Promise.allSettled: allSettled retains its result array — and through + // it every child's frame and generator — until the last child settles, so one long-running + // saga pinned the whole fan-out. A counter retains nothing. + function onSettle() { + pending--; + if (pending === 0) { + wake(); } - })) - .finally(() => doneYield()); + } + + for (const child of matched) { + driveChild(child.iterator, emit, isCancelled).then(onSettle, (error: Error) => { + errorsQueue.push(error); + onSettle(); + }); + } + + // Deliberately not an `async` method: at `target: es6` every `async` compiles to an __awaiter + // plus __generator state machine, and this is the hottest function on the flattened path. + function pump(): Promise> { + // Values before errors, so a failing saga does not discard work its siblings already did. + if (valuesQueue.length > 0) { + return Promise.resolve({ value: valuesQueue.shift()!, done: false }); + } + if (errorsQueue.length > 0) { + return Promise.reject(errorsQueue.shift()); + } + if (pending === 0 || cancelled) { + return Promise.resolve(DONE_RESULT); + } + if (!nextDeferred) { + nextDeferred = createDeferred(); + } + return nextDeferred.promise.then(pump); + } - return combinedIterator; + return { + next: pump, + return(value?: any) { + cancelled = true; + wake(); + return Promise.resolve({ value, done: true } as IteratorResult); + }, + throw(error?: any) { + cancelled = true; + wake(); + return Promise.reject(error); + }, + [Symbol.asyncIterator]() { + return this; + } + } as any; }; + + // Union of the children's declarations, so a combined saga nested inside another is itself + // indexable. One wildcard child makes the whole group a wildcard. + const actionTypes = wildcardKeys.length > 0 + ? [ANY_ACTION] + : Array.from(keysByActionType.keys()); + return { reducer, updater, + actionTypes, + [MATCH_SAGAS]: matchSagas }; } diff --git a/src/createModelSaga.ts b/src/createModelSaga.ts index f10f396..f4d300f 100644 --- a/src/createModelSaga.ts +++ b/src/createModelSaga.ts @@ -1,5 +1,4 @@ import { createStore, Store, Middleware, Dispatch, Action, AnyAction } from 'redux'; -import IPromiseAction from './IPromiseAction'; import ISaga from './ISaga'; import IUpdaterYield from './IUpdaterYield'; import ObservableSubscribed from './ObservableSubscribed'; @@ -7,52 +6,78 @@ import ObservableYield from './ObservableYield'; import ActionYield from './ActionYield'; import { IProfiler, IProfilerOptions, startProfiler } from './Profiler/profiler'; import AnyIterator from './AnyIterator'; +import { IMatchedSaga, isSagaGroup, MATCH_SAGAS } from './SagaGroup'; +import { getActionPromise, setActionPromise } from './ActionPromise'; + +type OnError = (error: Error, sourceAction: Action) => void; + +interface IContext { + subscriptions: Set; + liveIterators: Set; + dispatch: Dispatch; + profiler: IProfiler | null; + onError: OnError; + destroyed: boolean; +} async function update( - subscriptions: Subscription[], + context: IContext, iterator: AnyIterator, - dispatch: Dispatch, sourceAction: Action, - profiler: IProfiler | null, + sagaKey: string, ) { - const tracking = profiler?.track(iterator, sourceAction); + const tracking = context.profiler?.track(iterator, sourceAction, sagaKey); + // Registered so that destroy() can actually stop work that is still in flight. Previously an + // in-flight updater kept running after destroy(), dispatching into a store that had been thrown + // away — a per-user leak in exactly the backend shape the README recommends. + context.liveIterators.add(iterator); try { - return await doUpdate(subscriptions, iterator, dispatch, sourceAction, profiler); + return await doUpdate(context, iterator, sourceAction); } finally { + context.liveIterators.delete(iterator); tracking?.stop(); } } async function doUpdate( - subscriptions: Subscription[], + context: IContext, iterator: AnyIterator, - dispatch: Dispatch, sourceAction: Action, - profiler: IProfiler | null, ) { let nextResult: IUpdaterYield | undefined; do { + if (context.destroyed) { + await iterator.return?.(undefined); + break; + } let item: IteratorResult = await iterator.next(nextResult); nextResult = undefined; + // Checked again after the await, not only at the top of the loop. An updater suspended at an + // `await` cannot be stopped mid-statement — it necessarily runs on to its next `yield` — so + // the only thing we can guarantee is that whatever it yields there is not acted upon. + if (context.destroyed) { + await iterator.return?.(undefined); + break; + } if (item.done) { break; } else if (isPromiseIteration(item.value)) { const promiseResult: IUpdaterYield = await item.value; if (isObservableIteration(promiseResult)) { - await handleObservable(dispatch, promiseResult.observable, subscriptions, sourceAction, profiler); + await handleObservable(context, promiseResult.observable, sourceAction); } else if (isActionIteration(promiseResult)) { - await handleAction(dispatch, promiseResult.action); + await handleAction(context.dispatch, promiseResult.action); } else { nextResult = promiseResult; } } else if (isObservableIteration(item.value)) { - await handleObservable(dispatch, item.value.observable, subscriptions, sourceAction, profiler); + await handleObservable(context, item.value.observable, sourceAction); } else if (isActionIteration(item.value)) { - await handleAction(dispatch, item.value.action); + await handleAction(context.dispatch, item.value.action); } else { nextResult = item.value; } @@ -72,81 +97,142 @@ function isActionIteration(value: IUpdaterYield): value is ActionYield { } async function handleObservable( - dispatch: Dispatch, + context: IContext, observable: Observable, - subscriptions: Subscription[], sourceAction: Action, - profiler: IProfiler | null, ) { - const subscription = observable.subscribe(function (observableIterator: AnyIterator) { - update(subscriptions, observableIterator, dispatch, sourceAction, profiler); - }); - subscriptions.push(subscription); - const promiseObservableSubscribed = dispatch({ + const { subscriptions, onError } = context; + let subscription: Subscription | undefined = undefined; + let closed = false; + + // The subscription removes itself once the source ends. The previous implementation swept a plain + // array on a 10s interval using `for...in` + `splice`, which shifts elements down as the + // enumeration advances and therefore skipped entries, so completed subscriptions were never + // reclaimed and each one pinned its observer closure — and through it the source action. + const forget = () => { + closed = true; + if (subscription) { + subscriptions.delete(subscription); + } + }; + + subscription = observable.subscribe( + (observableIterator: AnyIterator) => { + update(context, observableIterator, sourceAction, '') + .catch((error: Error) => onError(error, sourceAction)); + }, + (error: Error) => { + forget(); + onError(error, sourceAction); + }, + () => { + forget(); + } + ); + + // An already-complete source calls `complete` synchronously, before `subscribe` returns, so the + // subscription would otherwise be added to the set after having already removed itself. + if (!closed) { + subscriptions.add(subscription); + } + + const promiseObservableSubscribed = context.dispatch({ type: ObservableSubscribed, observable, subscription, - sourceAction, - } as ObservableSubscribed) as Action as IPromiseAction; - if (promiseObservableSubscribed?.__promise instanceof Promise) { - await promiseObservableSubscribed.__promise; + sourceAction + } as ObservableSubscribed); + const promise = getActionPromise(promiseObservableSubscribed); + if (promise) { + await promise; } } async function handleAction(dispatch: Dispatch, action: Action) { - const promiseAction = dispatch(action) as IPromiseAction; - if (promiseAction?.__promise instanceof Promise) { - await promiseAction.__promise; + const dispatched = dispatch(action); + const promise = getActionPromise(dispatched); + if (promise) { + await promise; } } -function startGarbageCollector(subscriptions: Subscription[]) { - const intervalHandler = setInterval( - () => { - for (let index in subscriptions) { - if (subscriptions[index].closed) { - subscriptions.splice(parseInt(index), 1); - } - } - }, - 10e3, - ); - return { - stop() { - clearInterval(intervalHandler); - }, - }; -} - export interface IOptions { profiler?: IProfilerOptions; + /** + * Called when an updater rejects. Updaters run detached from the dispatch that started them, so + * without a handler their rejection is unhandled — which terminates the process on Node >= 15. + */ + onError?: OnError; } +/** Stands in until the middleware is applied and the real store dispatch becomes available. */ +const dispatchNotReady: Dispatch = (() => { + throw new Error('Aperol: dispatch is not available until the middleware has been applied to a store.'); +}) as Dispatch; + export default function createModelSaga(saga: ISaga, options?: IOptions) { const sagaStore = createStore(saga.reducer); - const subscriptions: Subscription[] = []; - const profiler = options?.profiler ? startProfiler(options.profiler) : null; + const context: IContext = { + subscriptions: new Set(), + liveIterators: new Set(), + dispatch: dispatchNotReady, + profiler: options?.profiler ? startProfiler(options.profiler) : null, + onError: options?.onError ?? ((error: Error) => console.error(error)), + destroyed: false + }; const middleware: Middleware = (store: Store) => (nextDispatch: Dispatch) => (action: Action) => { const result = nextDispatch(action); sagaStore.dispatch(action); const model = sagaStore.getState(); - const iterator = saga.updater(model, action); - const promise = update(subscriptions, iterator, store.dispatch, action, profiler); - Object.defineProperty(result, '__promise', { - enumerable: false, - configurable: false, - writable: false, - value: promise - }); + context.dispatch = store.dispatch; + + // Each reacting saga gets its OWN drive loop. The old design multiplexed every saga's yields + // into one queue that was drained a single yield at a time, and each yielded `put()` awaited + // the entire recursive fan-out of that nested action before the next yield was drained — so + // logically independent sagas were serialised behind each other's slowest await. + const matched: IMatchedSaga[] = []; + if (isSagaGroup(saga)) { + saga[MATCH_SAGAS](model, action, '', matched); + } else { + matched.push({ key: '', iterator: saga.updater(model, action) }); + } + + let promise: Promise; + if (matched.length === 0) { + promise = Promise.resolve(); + } else if (matched.length === 1) { + promise = update(context, matched[0].iterator, action, matched[0].key); + } else { + promise = Promise.all( + matched.map((child: IMatchedSaga) => update(context, child.iterator, action, child.key)) + ).then(() => undefined); + } + + // Registering a rejection handler here still lets callers who await the handle observe the + // rejection themselves; it only stops the detached path becoming an unhandled rejection. + promise.catch((error: Error) => context.onError(error, action)); + setActionPromise(result, promise); return result as any; }; - const garbageCollector = startGarbageCollector(subscriptions); const destroy = () => { - garbageCollector.stop(); - subscriptions.forEach((subscription: Subscription) => subscription.unsubscribe()); + context.destroyed = true; + context.liveIterators.forEach((iterator: AnyIterator) => { + iterator.return?.(undefined); + }); + context.liveIterators.clear(); + context.subscriptions.forEach((subscription: Subscription) => subscription.unsubscribe()); + context.subscriptions.clear(); }; + + /** + * Observable subscriptions currently held open. A count that only ever climbs is the signature of + * a subscription leak, which is otherwise invisible until the process runs out of heap. + */ + const getSubscriptionCount = () => context.subscriptions.size; + return { middleware, - destroy + destroy, + getSubscriptionCount }; } diff --git a/src/index.ts b/src/index.ts index 6382b47..978642f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -9,6 +9,8 @@ export { default as ObservableSubscribed } from './ObservableSubscribed'; export { default as IPromiseAction } from './IPromiseAction'; export { default as ISaga } from './ISaga'; export { default as IUpdaterYield } from './IUpdaterYield'; +export { ANY_ACTION } from './SagaGroup'; +export { getActionPromise } from './ActionPromise'; export function observe | AsyncIterator, Error>>( observable: TObservable, ) { diff --git a/tests/integration/Profiler/profiler.spec.ts b/tests/integration/Profiler/profiler.spec.ts index ee86a2d..66c1c93 100644 --- a/tests/integration/Profiler/profiler.spec.ts +++ b/tests/integration/Profiler/profiler.spec.ts @@ -11,6 +11,7 @@ describe('Profiler.profiler', function () { type A2 = { type: typeof A2 }; const saga1 = { + actionTypes: [A1], reducer: () => null, updater: async function* (_model: null, action: A1) { if (action.type === A1) { @@ -22,6 +23,7 @@ describe('Profiler.profiler', function () { }; const saga2 = { + actionTypes: [A2], reducer: () => null, updater: async function* (_model: null, action: A2) { if (action.type === A2) { @@ -30,9 +32,9 @@ describe('Profiler.profiler', function () { }, }; - let warnings: { message: string; action: Action }[]; - const onWarning = (message: string, action: Action) => { - warnings.push({ message, action }); + let warnings: { message: string; action: Action; sagaKey: string }[]; + const onWarning = (message: string, action: Action, sagaKey: string) => { + warnings.push({ message, action, sagaKey }); }; beforeEach(function () { @@ -41,7 +43,11 @@ describe('Profiler.profiler', function () { it('should log when some saga takes too much time', async function () { const sagas = combineSagas({ - saga1, + parent: combineSagas({ + child: combineSagas({ + saga1, + }), + }), }); const modelSaga = createModelSaga(sagas, { profiler: { thresholdMs: 100, onWarning } }); @@ -50,6 +56,8 @@ describe('Profiler.profiler', function () { await action.__promise; should(warnings).lengthOf(1); should(warnings[0].action).eql({ type: A1 }); + should.strictEqual(warnings[0].sagaKey, 'parent.child.saga1'); + should(warnings[0].message).containEql('saga=parent.child.saga1'); modelSaga.destroy(); }); @@ -69,6 +77,8 @@ describe('Profiler.profiler', function () { should(warnings).lengthOf(2); should(warnings[1].action).eql({ type: A1 }); should(warnings[0].action).eql({ type: A2 }); + should.strictEqual(warnings[1].sagaKey, 'saga1'); + should.strictEqual(warnings[0].sagaKey, 'saga2'); modelSaga.destroy(); }); diff --git a/tests/integration/index.spec.ts b/tests/integration/index.spec.ts index 91904a8..79bd1d1 100644 --- a/tests/integration/index.spec.ts +++ b/tests/integration/index.spec.ts @@ -3,6 +3,8 @@ import * as should from 'should'; import createModelSaga from '../../src/createModelSaga'; import combineSagas from '../../src/combineSagas'; import ObservableSubscribed from '../../src/ObservableSubscribed'; +import { ANY_ACTION } from '../../src/SagaGroup'; +import { getActionPromise } from '../../src/ActionPromise'; describe('index', () => { @@ -11,18 +13,26 @@ describe('index', () => { createModelSaga: actualCreateModelSaga, combineSagas: actualCombineSagas, ObservableSubscribed: actualObservableSubscribed, + ANY_ACTION: actualAnyAction, + getActionPromise: actualGetActionPromise, } = require('../../src/index'); should(actualCreateModelSaga).ok(); should(actualCombineSagas).ok(); should(actualObservableSubscribed).ok(); + should(actualAnyAction).ok(); + should(actualGetActionPromise).ok(); should(actualCreateModelSaga).Function(); should(actualCombineSagas).Function(); should(actualObservableSubscribed).String(); + should(actualAnyAction).String(); + should(actualGetActionPromise).Function(); should(actualCreateModelSaga).equal(createModelSaga); should(actualCombineSagas).equal(combineSagas); should(actualObservableSubscribed).equal(ObservableSubscribed); + should(actualAnyAction).equal(ANY_ACTION); + should(actualGetActionPromise).equal(getActionPromise); }); }); diff --git a/tests/unit/ActionPromise.spec.ts b/tests/unit/ActionPromise.spec.ts new file mode 100644 index 0000000..38cbabb --- /dev/null +++ b/tests/unit/ActionPromise.spec.ts @@ -0,0 +1,55 @@ +import * as should from 'should'; +import { getActionPromise, setActionPromise } from '../../src/ActionPromise'; + +describe('ActionPromise', function () { + + it('should set and get a promise without exposing the legacy property', function () { + const action = { type: 'Test' }; + const promise = Promise.resolve(); + + setActionPromise(action, promise); + + should.strictEqual(getActionPromise(action), promise); + should.deepEqual(Object.keys(action), ['type']); + should.deepEqual(Object.getOwnPropertyDescriptor(action, '__promise'), { + configurable: true, + enumerable: false, + value: promise, + writable: true, + }); + }); + + it('should ignore null, primitives, and non-Promise legacy values', function () { + const promise = Promise.resolve(); + for (const value of [null, undefined, false, 1, 'action']) { + setActionPromise(value, promise); + should.strictEqual(getActionPromise(value), undefined); + } + should.strictEqual(getActionPromise({ type: 'Legacy', __promise: {} }), undefined); + }); + + it('should support frozen and sealed actions through the WeakMap', function () { + const frozen = Object.freeze({ type: 'Frozen' }); + const sealed = Object.seal({ type: 'Sealed' }); + const frozenPromise = Promise.resolve(); + const sealedPromise = Promise.resolve(); + + setActionPromise(frozen, frozenPromise); + setActionPromise(sealed, sealedPromise); + + should.strictEqual(getActionPromise(frozen), frozenPromise); + should.strictEqual(getActionPromise(sealed), sealedPromise); + }); + + it('should replace the promise when an action object is reused', function () { + const action: { type: string; __promise?: Promise } = { type: 'Repeated' }; + const firstPromise = Promise.resolve(); + const secondPromise = Promise.resolve(); + + setActionPromise(action, firstPromise); + setActionPromise(action, secondPromise); + + should.strictEqual(getActionPromise(action), secondPromise); + should.strictEqual(action.__promise, secondPromise); + }); +}); diff --git a/tests/unit/combineSagas.spec.ts b/tests/unit/combineSagas.spec.ts index 1c8fc57..e143095 100644 --- a/tests/unit/combineSagas.spec.ts +++ b/tests/unit/combineSagas.spec.ts @@ -1,6 +1,6 @@ import '../../src/polyfill/observable'; -import { put } from '../../src/index'; +import { ANY_ACTION, getActionPromise, put } from '../../src/index'; import { createStore, applyMiddleware, Action } from 'redux'; import * as should from 'should'; import { @@ -34,6 +34,7 @@ describe('Application.combineSaga', function () { } const warningSaga = { + actionTypes: ['Subtract'], reducer(model: IWarningModel = [], action: ISubtract) { switch (action.type) { case 'Subtract': @@ -102,17 +103,13 @@ describe('Application.combineSaga', function () { await promiseSubtract112Again.__promise; should.deepEqual(removeInternalActions(assertations.reducedActions), [ add113, - added, subtract112, secondSubtract112, - warningShown, ]); should.deepEqual(removeInternalActions(assertations.updatedActions), [ add113, - added, subtract112, secondSubtract112, - warningShown, ]); should.deepEqual(removeInternalActions(assertations.dispatchedActions), [ add113, @@ -122,11 +119,9 @@ describe('Application.combineSaga', function () { warningShown, ]); should.deepEqual(assertations.updatedModels, [ - { sum: 113 }, { sum: 113 }, { sum: 1 }, { sum: -111 }, - { sum: -111 }, ]); should.deepEqual(assertations.addedAmounts, [ 113, @@ -134,4 +129,77 @@ describe('Application.combineSaga', function () { should.strictEqual(shownWarningsCount, 1); modelSaga.destroy(); }); + + it('should require action type declarations', function () { + should.throws( + () => combineSagas({ + invalid: { + reducer: (model: null = null) => model, + *updater() { + return; + }, + } as any, + }), + /missing the required "actionTypes" declaration/ + ); + }); + + it('should route only matching sagas and preserve identity for unmatched actions', async function () { + const reduced: string[] = []; + const updated: string[] = []; + const first = { + actionTypes: ['First'], + reducer(model: number = 0, action: Action) { + reduced.push(`first:${action.type}`); + return action.type === 'First' ? model + 1 : model; + }, + *updater(_model: number, action: Action) { + updated.push(`first:${action.type}`); + }, + }; + const second = { + actionTypes: ['Second'], + reducer(model: number = 0, action: Action) { + reduced.push(`second:${action.type}`); + return action.type === 'Second' ? model + 1 : model; + }, + *updater(_model: number, action: Action) { + updated.push(`second:${action.type}`); + }, + }; + const combined = combineSagas({ first, second }); + const initialModel = combined.reducer(undefined as any, { type: '@@init' }); + reduced.length = 0; + const unmatchedModel = combined.reducer(initialModel, { type: 'Unmatched' }); + should.strictEqual(unmatchedModel, initialModel); + should.deepEqual(reduced, []); + + const modelSaga = createModelSaga(combined); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + reduced.length = 0; + const firstAction = store.dispatch({ type: 'First' }); + await getActionPromise(firstAction); + should.deepEqual(reduced, ['first:First']); + should.deepEqual(updated, ['first:First']); + modelSaga.destroy(); + }); + + it('should route wildcard sagas for every action', async function () { + const updated: string[] = []; + const wildcard = { + actionTypes: [ANY_ACTION], + reducer(model: number = 0, action: Action) { + return action.type === 'Increment' ? model + 1 : model; + }, + *updater(_model: number, action: Action) { + updated.push(action.type); + }, + }; + const modelSaga = createModelSaga(combineSagas({ wildcard })); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + const dispatchedAction = store.dispatch({ type: 'OtherwiseUnmatched' }); + await getActionPromise(dispatchedAction); + should.deepEqual(updated, ['OtherwiseUnmatched']); + modelSaga.destroy(); + }); }); diff --git a/tests/unit/createModelSaga.spec.ts b/tests/unit/createModelSaga.spec.ts index 7f57b67..3be0dc1 100644 --- a/tests/unit/createModelSaga.spec.ts +++ b/tests/unit/createModelSaga.spec.ts @@ -9,12 +9,23 @@ import { sumSaga, sumReducer, IAdd, + ISubtract, IAutoAdding, asyncIteratorSumSaga, } from './sumModelMock'; import createModelSaga from '../../src/createModelSaga'; import IPromiseAction from '../../src/IPromiseAction'; import ObservableSubscribed from '../../src/ObservableSubscribed'; +import combineSagas from '../../src/combineSagas'; +import { getActionPromise, observe, put } from '../../src/index'; +import AnyIterator from '../../src/AnyIterator'; + +function requireDefined(value: T | undefined, name: string): T { + if (value === undefined) { + throw new Error(`${name} was not initialized`); + } + return value; +} describe('Application.craeteModelSaga', function () { @@ -168,4 +179,189 @@ describe('Application.craeteModelSaga', function () { ]); modelSaga.destroy(); }); + + it('should run matching leaf sagas concurrently and resolve after all finish', async function () { + let releaseFirst: (() => void) | undefined; + let releaseSecond: (() => void) | undefined; + const firstGate = new Promise((resolve: () => void) => releaseFirst = resolve); + const secondGate = new Promise((resolve: () => void) => releaseSecond = resolve); + const started: string[] = []; + const finished: string[] = []; + const first = { + actionTypes: ['Run'], + reducer: (model: null = null) => model, + *updater() { + started.push('first'); + yield firstGate; + finished.push('first'); + }, + }; + const second = { + actionTypes: ['Run'], + reducer: (model: null = null) => model, + *updater() { + started.push('second'); + yield secondGate; + finished.push('second'); + }, + }; + const modelSaga = createModelSaga(combineSagas({ first, second })); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + const action = store.dispatch({ type: 'Run' }); + const promise = requireDefined(getActionPromise(action), 'action promise'); + should.deepEqual(started, ['first', 'second']); + + let settled = false; + promise.then(() => settled = true); + requireDefined(releaseFirst, 'first release')(); + await Promise.resolve(); + await Promise.resolve(); + should.strictEqual(settled, false); + requireDefined(releaseSecond, 'second release')(); + await promise; + should.deepEqual(finished, ['first', 'second']); + modelSaga.destroy(); + }); + + it('should support redispatching mutable actions and frozen actions', async function () { + const modelSaga = createModelSaga(sumSaga); + const store = createStore(sumReducer, applyMiddleware(modelSaga.middleware)); + const repeated = { type: 'Subtract', amount: 1 } as ISubtract; + const firstResult = store.dispatch(repeated); + const firstPromise = getActionPromise(firstResult); + await firstPromise; + const secondResult = store.dispatch(repeated); + const secondPromise = getActionPromise(secondResult); + await secondPromise; + should.notStrictEqual(firstPromise, secondPromise); + + const frozen = Object.freeze({ type: 'Subtract', amount: 1 }) as ISubtract; + const frozenResult = store.dispatch(frozen); + const frozenPromise = getActionPromise(frozenResult); + should(frozenPromise).ok(); + await frozenPromise; + modelSaga.destroy(); + }); + + it('should read legacy action promises', async function () { + const promise = Promise.resolve(); + const action = { type: 'Legacy', __promise: promise }; + should.strictEqual(getActionPromise(action), promise); + await getActionPromise(action); + }); + + it('should expose updater rejection and report it through onError', async function () { + const failure = new Error('updater failed'); + const errors: Error[] = []; + const saga = { + actionTypes: ['Fail'], + reducer: (model: null = null) => model, + *updater() { + throw failure; + }, + }; + const modelSaga = createModelSaga(saga, { + onError: (error: Error) => errors.push(error), + }); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + const action = store.dispatch({ type: 'Fail' }); + let rejected: Error | undefined; + try { + await getActionPromise(action); + } catch (error) { + rejected = error; + } + should.strictEqual(rejected, failure); + should.deepEqual(errors, [failure]); + modelSaga.destroy(); + }); + + it('should remove completed observable subscriptions immediately', async function () { + let complete: (() => void) | undefined; + const synchronous = new Observable((observer: SubscriptionObserver) => { + observer.complete(); + return () => undefined; + }); + const completable = new Observable((observer: SubscriptionObserver) => { + complete = () => observer.complete(); + return () => undefined; + }); + const saga = { + actionTypes: ['Observe', 'ObserveSync'], + reducer: (model: null = null) => model, + *updater(_model: null, action: Action) { + if (action.type === 'Observe') { + yield observe(completable); + } else if (action.type === 'ObserveSync') { + yield observe(synchronous); + } + }, + }; + const modelSaga = createModelSaga(saga); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + const openAction = store.dispatch({ type: 'Observe' }); + await getActionPromise(openAction); + should.strictEqual(modelSaga.getSubscriptionCount(), 1); + requireDefined(complete, 'observable completion')(); + should.strictEqual(modelSaga.getSubscriptionCount(), 0); + + const syncAction = store.dispatch({ type: 'ObserveSync' }); + await getActionPromise(syncAction); + should.strictEqual(modelSaga.getSubscriptionCount(), 0); + modelSaga.destroy(); + }); + + it('should route observable errors and forget failed subscriptions', async function () { + const failure = new Error('observable failed'); + const errors: Error[] = []; + const observable = new Observable((observer: SubscriptionObserver) => { + observer.error(failure); + return () => undefined; + }); + const saga = { + actionTypes: ['Observe'], + reducer: (model: null = null) => model, + *updater(_model: null, action: Action) { + if (action.type === 'Observe') { + yield observe(observable); + } + }, + }; + const modelSaga = createModelSaga(saga, { + onError: (error: Error) => errors.push(error), + }); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + const dispatchedAction = store.dispatch({ type: 'Observe' }); + await getActionPromise(dispatchedAction); + should.deepEqual(errors, [failure]); + should.strictEqual(modelSaga.getSubscriptionCount(), 0); + modelSaga.destroy(); + }); + + it('should stop in-flight iterators without dispatching post-await work', async function () { + let release: (() => void) | undefined; + const gate = new Promise((resolve: () => void) => release = resolve); + const dispatched: Action[] = []; + const saga = { + actionTypes: ['Run'], + reducer: (model: null = null) => model, + async *updater() { + await gate; + yield put({ type: 'TooLate' }); + }, + }; + const modelSaga = createModelSaga(saga); + const store = createStore( + (state: null = null, action: Action) => { + dispatched.push(action); + return state; + }, + applyMiddleware(modelSaga.middleware) + ); + const dispatchedAction = store.dispatch({ type: 'Run' }); + modelSaga.destroy(); + requireDefined(release, 'iterator release')(); + await getActionPromise(dispatchedAction); + should.strictEqual(dispatched.some((item: Action) => item.type === 'TooLate'), false); + }); }); diff --git a/tests/unit/sumModelMock.ts b/tests/unit/sumModelMock.ts index 3105a1d..3131e19 100644 --- a/tests/unit/sumModelMock.ts +++ b/tests/unit/sumModelMock.ts @@ -66,6 +66,7 @@ export function addAmount(amount: number) { } export const sumSaga = { + actionTypes: ['Add', 'Subtract', 'AutoAdding'], reducer(model: ISumModel = initialSumModel, action: IAdd | ISubtract) { assertations.reducedActions!.push(action); switch (action.type) { @@ -115,6 +116,7 @@ export const sumSaga = { }; export const asyncIteratorSumSaga = { + actionTypes: ['Add', 'Subtract', 'AutoAdding'], reducer(model: ISumModel = initialSumModel, action: IAdd | ISubtract) { assertations.reducedActions!.push(action); switch (action.type) { From 6e3636161458b53b1cc12c2370b72d9f5173eccd Mon Sep 17 00:00:00 2001 From: Michael Zabka Date: Fri, 17 Jul 2026 12:39:36 +0200 Subject: [PATCH 2/2] Synchronize saga correctness fixes Port follow-up behavior from Box MR 3682, including updater routing, reducer delivery, fan-out settlement, completion callbacks, error normalization, and lifecycle cleanup.\n\nCo-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 10 +- README.md | 15 +- src/ISaga.ts | 8 +- src/combineSagas.ts | 31 ++-- src/createModelSaga.ts | 208 ++++++++++++++++----- tests/unit/combineSagas.spec.ts | 25 ++- tests/unit/createModelSaga.spec.ts | 279 ++++++++++++++++++++++++++++- 7 files changed, 503 insertions(+), 73 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e7b7db7..d0c12d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,10 +6,16 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ## [Unreleased] ### Changed -- Require sagas to declare handled action types and run matching sagas concurrently +- Use saga action type declarations for updater routing while every reducer continues to receive every action +- Run matching saga updaters concurrently and wait for every sibling to settle +- Report updater completion duration for dispatched fan-outs and observable emissions ### Fixed -- Clean up completed iterators and observable subscriptions +- Deduplicate updater routing for repeated and overlapping action type declarations +- Normalize updater rejection values and report every sibling failure without duplicate nested-action reports +- Protect updater completion and error reporting from callback failures +- Clean up completed, failed, and manually unsubscribed observable subscriptions +- Make destruction idempotent, stop iterators between steps, and skip new updaters after destruction ## [2.2.0] ### Added diff --git a/README.md b/README.md index e31c317..45f6c90 100644 --- a/README.md +++ b/README.md @@ -91,7 +91,8 @@ const appSaga = { // ... app code ``` -Every saga must declare the action types it handles. Use `ANY_ACTION` in `actionTypes` when a saga needs to process every action. +Every saga must declare the action types handled by its updater. Reducers still receive every action. Use `ANY_ACTION` in +`actionTypes` when an updater needs to process every action. ### Combining more sagas @@ -126,6 +127,18 @@ const modelSaga = createModelSaga(appSaga, { }); ``` +Use `onUpdateComplete` to observe updater duration. It runs once after each dispatched action fan-out and once after each +observable emission, whether processing succeeds or fails. Dispatched fan-outs use an empty saga key; observable emissions +use the originating leaf key. + +```js +const modelSaga = createModelSaga(appSaga, { + onUpdateComplete: (durationMs, sourceAction, sagaKey) => { + console.log(durationMs, sourceAction.type, sagaKey); + }, +}); +``` + ## ES5 transpiled and bundled If you do not transpile code or not using bundler (like webpack) and need to have pre-transpiled code, import file `aperol/dist/es5` instead. diff --git a/src/ISaga.ts b/src/ISaga.ts index 3982e60..364871b 100644 --- a/src/ISaga.ts +++ b/src/ISaga.ts @@ -6,12 +6,10 @@ interface ISaga { reducer(model: TModel, action: Action): TModel; updater(model: TModel, action: Action): Iterator | AsyncIterator; /** - * Every action type this saga reacts to, in EITHER its reducer or its updater. + * Every action type this saga's updater reacts to. * - * `combineSagas` will not run this saga at all for any other action type, which is what keeps - * dispatch cost proportional to the sagas that actually react rather than to the total number of - * sagas. The declaration must therefore cover the reducer as well: if a saga reduces on a type it - * did not declare, that reduction will not happen. + * `combineSagas` uses this declaration to avoid constructing updaters for unrelated actions. + * Reducers always receive every action and are intentionally independent of this list. * * Declare `[ANY_ACTION]` for a saga that genuinely reacts to every action. */ diff --git a/src/combineSagas.ts b/src/combineSagas.ts index 274187a..a4d8999 100644 --- a/src/combineSagas.ts +++ b/src/combineSagas.ts @@ -60,7 +60,7 @@ export default function combineSagas( if (!Array.isArray(sagas[key].actionTypes)) { throw new Error( `Saga "${key}" is missing the required "actionTypes" declaration. List every action type ` + - `it reacts to, in either its reducer or its updater, or declare ["${ANY_ACTION}"] if it ` + + `its updater reacts to, or declare ["${ANY_ACTION}"] if it ` + `genuinely reacts to every action.` ); } @@ -71,18 +71,30 @@ export default function combineSagas( // Resolved once at combine time: type -> (sagas declaring it + every wildcard saga). A type that // is absent from the index can only be matched by wildcards, so the lookup needs no per-action // concat and no unbounded cache. - const keysByActionType = new Map(); + const keySetsByActionType = new Map>(); for (const key of sagaKeys) { for (const actionType of sagas[key].actionTypes) { if (actionType === ANY_ACTION) { continue; } - const existing = keysByActionType.get(actionType); - keysByActionType.set(actionType, existing ? [...existing, key] : [key]); + let keySet = keySetsByActionType.get(actionType); + if (!keySet) { + keySet = new Set(); + keySetsByActionType.set(actionType, keySet); + } + keySet.add(key); } } - for (const actionType of Array.from(keysByActionType.keys())) { - keysByActionType.set(actionType, [...keysByActionType.get(actionType)!, ...wildcardKeys]); + const keysByActionType = new Map(); + for (const actionType of Array.from(keySetsByActionType.keys())) { + const keySet = keySetsByActionType.get(actionType); + if (!keySet) { + continue; + } + for (const wildcardKey of wildcardKeys) { + keySet.add(wildcardKey); + } + keysByActionType.set(actionType, Array.from(keySet)); } function matchingKeys(actionType: string) { @@ -99,15 +111,10 @@ export default function combineSagas( return initialModel as TModel; } - const keys = matchingKeys(action.type); - if (keys.length === 0) { - return model; - } - // Copy lazily: an action that changes no slice returns the *same* model object, so no garbage // is produced and referential identity survives for downstream memoisation. let nextModel: TModel | undefined = undefined; - for (const key of keys) { + for (const key of sagaKeys) { const previousSlice = model[key]; const nextSlice = sagas[key].reducer(previousSlice, action); if (nextSlice !== previousSlice) { diff --git a/src/createModelSaga.ts b/src/createModelSaga.ts index f4d300f..d9ef0a1 100644 --- a/src/createModelSaga.ts +++ b/src/createModelSaga.ts @@ -6,36 +6,85 @@ import ObservableYield from './ObservableYield'; import ActionYield from './ActionYield'; import { IProfiler, IProfilerOptions, startProfiler } from './Profiler/profiler'; import AnyIterator from './AnyIterator'; -import { IMatchedSaga, isSagaGroup, MATCH_SAGAS } from './SagaGroup'; +import { ANY_ACTION, IMatchedSaga, isSagaGroup, MATCH_SAGAS } from './SagaGroup'; import { getActionPromise, setActionPromise } from './ActionPromise'; type OnError = (error: Error, sourceAction: Action) => void; +type OnUpdateComplete = (durationMs: number, sourceAction: Action, sagaKey: string) => void; interface IContext { subscriptions: Set; - liveIterators: Set; dispatch: Dispatch; profiler: IProfiler | null; onError: OnError; + onUpdateComplete: OnUpdateComplete; destroyed: boolean; } +class PropagatedActionError { + public constructor(public readonly error: Error) {} +} + +function describeRejection(value: unknown): string { + if (typeof value === 'string') { + return value; + } + let serialized: string | undefined; + try { + serialized = JSON.stringify(value); + } catch { + serialized = undefined; + } + if (serialized) { + return serialized; + } + try { + return String(value); + } catch { + return ''; + } +} + +function normalizeError(value: unknown): Error { + if (value instanceof Error) { + return value; + } + return new Error(`Aperol updater rejected with a non-Error value: ${describeRejection(value)}`); +} + +function propagatedActionError(error: Error): PropagatedActionError { + return new PropagatedActionError(error); +} + +function isPropagatedActionError(value: unknown): value is PropagatedActionError { + return value instanceof PropagatedActionError; +} + +function unwrapAndReportError(context: IContext, value: unknown, sourceAction: Action): Error { + if (isPropagatedActionError(value)) { + return value.error; + } + const error = normalizeError(value); + context.onError(error, sourceAction); + return error; +} + async function update( context: IContext, iterator: AnyIterator, sourceAction: Action, sagaKey: string, + reportCompletion: boolean, ) { + const startedAt = reportCompletion ? Date.now() : 0; const tracking = context.profiler?.track(iterator, sourceAction, sagaKey); - // Registered so that destroy() can actually stop work that is still in flight. Previously an - // in-flight updater kept running after destroy(), dispatching into a store that had been thrown - // away — a per-user leak in exactly the backend shape the README recommends. - context.liveIterators.add(iterator); try { - return await doUpdate(context, iterator, sourceAction); + return await doUpdate(context, iterator, sourceAction, sagaKey); } finally { - context.liveIterators.delete(iterator); tracking?.stop(); + if (reportCompletion) { + context.onUpdateComplete(Date.now() - startedAt, sourceAction, sagaKey); + } } } @@ -43,29 +92,25 @@ async function doUpdate( context: IContext, iterator: AnyIterator, sourceAction: Action, + sagaKey: string, ) { let nextResult: IUpdaterYield | undefined; do { if (context.destroyed) { - await iterator.return?.(undefined); - break; + if (iterator.return) { + await iterator.return(undefined); + } + return; } let item: IteratorResult = await iterator.next(nextResult); nextResult = undefined; - // Checked again after the await, not only at the top of the loop. An updater suspended at an - // `await` cannot be stopped mid-statement — it necessarily runs on to its next `yield` — so - // the only thing we can guarantee is that whatever it yields there is not acted upon. - if (context.destroyed) { - await iterator.return?.(undefined); - break; - } if (item.done) { break; } else if (isPromiseIteration(item.value)) { const promiseResult: IUpdaterYield = await item.value; if (isObservableIteration(promiseResult)) { - await handleObservable(context, promiseResult.observable, sourceAction); + await handleObservable(context, promiseResult.observable, sourceAction, sagaKey); } else if (isActionIteration(promiseResult)) { await handleAction(context.dispatch, promiseResult.action); @@ -74,7 +119,7 @@ async function doUpdate( } } else if (isObservableIteration(item.value)) { - await handleObservable(context, item.value.observable, sourceAction); + await handleObservable(context, item.value.observable, sourceAction, sagaKey); } else if (isActionIteration(item.value)) { await handleAction(context.dispatch, item.value.action); @@ -100,6 +145,7 @@ async function handleObservable( context: IContext, observable: Observable, sourceAction: Action, + sagaKey: string, ) { const { subscriptions, onError } = context; let subscription: Subscription | undefined = undefined; @@ -118,22 +164,32 @@ async function handleObservable( subscription = observable.subscribe( (observableIterator: AnyIterator) => { - update(context, observableIterator, sourceAction, '') - .catch((error: Error) => onError(error, sourceAction)); + update(context, observableIterator, sourceAction, sagaKey, true) + .catch((error: unknown) => unwrapAndReportError(context, error, sourceAction)); }, - (error: Error) => { + (error: unknown) => { forget(); - onError(error, sourceAction); + onError(normalizeError(error), sourceAction); }, () => { forget(); } ); - // An already-complete source calls `complete` synchronously, before `subscribe` returns, so the - // subscription would otherwise be added to the set after having already removed itself. - if (!closed) { - subscriptions.add(subscription); + const activeSubscription = subscription; + const innerUnsubscribe = activeSubscription.unsubscribe; + activeSubscription.unsubscribe = () => { + if (closed) { + return; + } + forget(); + innerUnsubscribe.call(activeSubscription); + }; + if (context.destroyed || closed) { + activeSubscription.unsubscribe(); + return; + } else if (!closed) { + subscriptions.add(activeSubscription); } const promiseObservableSubscribed = context.dispatch({ @@ -144,7 +200,11 @@ async function handleObservable( } as ObservableSubscribed); const promise = getActionPromise(promiseObservableSubscribed); if (promise) { - await promise; + try { + await promise; + } catch (error) { + throw propagatedActionError(normalizeError(error)); + } } } @@ -152,15 +212,24 @@ async function handleAction(dispatch: Dispatch, action: Action) { const dispatched = dispatch(action); const promise = getActionPromise(dispatched); if (promise) { - await promise; + try { + await promise; + } catch (error) { + throw propagatedActionError(normalizeError(error)); + } } } export interface IOptions { profiler?: IProfilerOptions; + /** Called once for a dispatched action fan-out and once for each observable emission. */ + onUpdateComplete?: OnUpdateComplete; /** * Called when an updater rejects. Updaters run detached from the dispatch that started them, so * without a handler their rejection is unhandled — which terminates the process on Node >= 15. + * + * Called at most once per error: nested action failures are reported by the updater that + * originally failed rather than at every parent put() boundary. */ onError?: OnError; } @@ -172,12 +241,31 @@ const dispatchNotReady: Dispatch = (() => { export default function createModelSaga(saga: ISaga, options?: IOptions) { const sagaStore = createStore(saga.reducer); + const onError = options?.onError || ((error: Error) => console.error(error)); + const safeOnError = (error: Error, sourceAction: Action) => { + try { + onError(error, sourceAction); + } catch (reportingError) { + console.error( + 'Aperol onError callback failed', + normalizeError(reportingError), + { originalError: error, sourceAction } + ); + } + }; + const onUpdateComplete = options?.onUpdateComplete || (() => undefined); const context: IContext = { subscriptions: new Set(), - liveIterators: new Set(), dispatch: dispatchNotReady, profiler: options?.profiler ? startProfiler(options.profiler) : null, - onError: options?.onError ?? ((error: Error) => console.error(error)), + onError: safeOnError, + onUpdateComplete: (durationMs: number, sourceAction: Action, sagaKey: string) => { + try { + onUpdateComplete(durationMs, sourceAction, sagaKey); + } catch (error) { + safeOnError(normalizeError(error), sourceAction); + } + }, destroyed: false }; const middleware: Middleware = (store: Store) => (nextDispatch: Dispatch) => (action: Action) => { @@ -185,6 +273,10 @@ export default function createModelSaga(saga: ISaga(saga: ISaga(saga)) { saga[MATCH_SAGAS](model, action, '', matched); - } else { + } else if (saga.actionTypes.indexOf(action.type) !== -1 || saga.actionTypes.indexOf(ANY_ACTION) !== -1) { matched.push({ key: '', iterator: saga.updater(model, action) }); } @@ -201,25 +293,53 @@ export default function createModelSaga(saga: ISaga { + throw unwrapAndReportError(context, error, action); + }); } else { - promise = Promise.all( - matched.map((child: IMatchedSaga) => update(context, child.iterator, action, child.key)) - ).then(() => undefined); + promise = new Promise((resolve: () => void, reject: (error: Error) => void) => { + let pending = matched.length; + let firstError: Error | undefined; + let failed = false; + const onSettle = () => { + pending--; + if (pending === 0) { + if (failed && firstError) { + reject(firstError); + } else { + resolve(); + } + } + }; + for (const child of matched) { + update(context, child.iterator, action, child.key, false).then(onSettle, (error: unknown) => { + const reportedError = unwrapAndReportError(context, error, action); + if (!failed) { + failed = true; + firstError = reportedError; + } + onSettle(); + }); + } + }); } - - // Registering a rejection handler here still lets callers who await the handle observe the - // rejection themselves; it only stops the detached path becoming an unhandled rejection. - promise.catch((error: Error) => context.onError(error, action)); + if (matched.length > 0) { + const startedAt = Date.now(); + promise.then( + () => context.onUpdateComplete(Date.now() - startedAt, action, ''), + () => context.onUpdateComplete(Date.now() - startedAt, action, '') + ); + } + promise.catch(() => undefined); setActionPromise(result, promise); return result as any; }; const destroy = () => { + if (context.destroyed) { + return; + } context.destroyed = true; - context.liveIterators.forEach((iterator: AnyIterator) => { - iterator.return?.(undefined); - }); - context.liveIterators.clear(); context.subscriptions.forEach((subscription: Subscription) => subscription.unsubscribe()); context.subscriptions.clear(); }; diff --git a/tests/unit/combineSagas.spec.ts b/tests/unit/combineSagas.spec.ts index e143095..d643b56 100644 --- a/tests/unit/combineSagas.spec.ts +++ b/tests/unit/combineSagas.spec.ts @@ -103,8 +103,10 @@ describe('Application.combineSaga', function () { await promiseSubtract112Again.__promise; should.deepEqual(removeInternalActions(assertations.reducedActions), [ add113, + added, subtract112, secondSubtract112, + warningShown, ]); should.deepEqual(removeInternalActions(assertations.updatedActions), [ add113, @@ -144,7 +146,7 @@ describe('Application.combineSaga', function () { ); }); - it('should route only matching sagas and preserve identity for unmatched actions', async function () { + it('should run every reducer, route only matching updaters, and preserve unchanged identity', async function () { const reduced: string[] = []; const updated: string[] = []; const first = { @@ -172,18 +174,35 @@ describe('Application.combineSaga', function () { reduced.length = 0; const unmatchedModel = combined.reducer(initialModel, { type: 'Unmatched' }); should.strictEqual(unmatchedModel, initialModel); - should.deepEqual(reduced, []); + should.deepEqual(reduced, ['first:Unmatched', 'second:Unmatched']); const modelSaga = createModelSaga(combined); const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); reduced.length = 0; const firstAction = store.dispatch({ type: 'First' }); await getActionPromise(firstAction); - should.deepEqual(reduced, ['first:First']); + should.deepEqual(reduced, ['first:First', 'second:First']); should.deepEqual(updated, ['first:First']); modelSaga.destroy(); }); + it('should invoke an updater once for duplicate and overlapping route declarations', async function () { + const updated: string[] = []; + const saga = { + actionTypes: ['Run', 'Run', ANY_ACTION], + reducer: (model: null = null) => model, + *updater(_model: null, action: Action) { + updated.push(action.type); + }, + }; + const modelSaga = createModelSaga(combineSagas({ saga })); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + const dispatchedAction = store.dispatch({ type: 'Run' }); + await getActionPromise(dispatchedAction); + should.deepEqual(updated, ['Run']); + modelSaga.destroy(); + }); + it('should route wildcard sagas for every action', async function () { const updated: string[] = []; const wildcard = { diff --git a/tests/unit/createModelSaga.spec.ts b/tests/unit/createModelSaga.spec.ts index 3be0dc1..78117f7 100644 --- a/tests/unit/createModelSaga.spec.ts +++ b/tests/unit/createModelSaga.spec.ts @@ -27,6 +27,10 @@ function requireDefined(value: T | undefined, name: string): T { return value; } +function isObservableSubscribedAction(action: Action): action is ObservableSubscribed { + return action.type === ObservableSubscribed; +} + describe('Application.craeteModelSaga', function () { beforeEach(() => { @@ -56,7 +60,6 @@ describe('Application.craeteModelSaga', function () { ]); should.deepEqual(removeInternalActions(assertations.updatedActions), [ add113, - added, ]); should.deepEqual(removeInternalActions(assertations.dispatchedActions), [ add113, @@ -64,7 +67,6 @@ describe('Application.craeteModelSaga', function () { ]); should.deepEqual(assertations.updatedModels, [ { sum: 113 }, - { sum: 113 }, ]); should.deepEqual(assertations.addedAmounts, [ 113, @@ -164,7 +166,6 @@ describe('Application.craeteModelSaga', function () { ]); should.deepEqual(removeInternalActions(assertations.updatedActions), [ add113, - added, ]); should.deepEqual(removeInternalActions(assertations.dispatchedActions), [ add113, @@ -172,7 +173,6 @@ describe('Application.craeteModelSaga', function () { ]); should.deepEqual(assertations.updatedModels, [ { sum: 113 }, - { sum: 113 }, ]); should.deepEqual(assertations.addedAmounts, [ 113, @@ -276,6 +276,221 @@ describe('Application.craeteModelSaga', function () { modelSaga.destroy(); }); + it('should run a root updater only for declared action types', async function () { + const updated: string[] = []; + const saga = { + actionTypes: ['Match'], + reducer: (model: null = null) => model, + *updater(_model: null, action: Action) { + updated.push(action.type); + }, + }; + const modelSaga = createModelSaga(saga); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + await getActionPromise(store.dispatch({ type: 'Ignore' })); + await getActionPromise(store.dispatch({ type: 'Match' })); + should.deepEqual(updated, ['Match']); + modelSaga.destroy(); + }); + + it('should normalize non-Error updater rejections', async function () { + const errors: Error[] = []; + const saga = { + actionTypes: ['Fail'], + reducer: (model: null = null) => model, + *updater() { + yield Promise.reject('broken'); + }, + }; + const modelSaga = createModelSaga(saga, { + onError: (error: Error) => errors.push(error), + }); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + let rejected: Error | undefined; + try { + await getActionPromise(store.dispatch({ type: 'Fail' })); + } catch (error) { + rejected = error; + } + should.strictEqual(rejected && rejected.message, 'Aperol updater rejected with a non-Error value: broken'); + should.deepEqual(errors, [rejected]); + modelSaga.destroy(); + }); + + it('should normalize updater rejections that cannot be serialized', async function () { + const rejection = { + toJSON() { + throw new Error('cannot serialize'); + }, + toString() { + throw new Error('cannot stringify'); + }, + }; + const errors: Error[] = []; + const saga = { + actionTypes: ['Fail'], + reducer: (model: null = null) => model, + *updater() { + yield Promise.reject(rejection); + }, + }; + const modelSaga = createModelSaga(saga, { + onError: (error: Error) => errors.push(error), + }); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + let rejected: Error | undefined; + try { + await getActionPromise(store.dispatch({ type: 'Fail' })); + } catch (error) { + rejected = error; + } + should.strictEqual( + rejected && rejected.message, + 'Aperol updater rejected with a non-Error value: ', + ); + should.deepEqual(errors, [rejected]); + modelSaga.destroy(); + }); + + it('should report a nested action failure only at its original updater boundary', async function () { + const failure = new Error('nested failed'); + const errors: Error[] = []; + const saga = { + actionTypes: ['Parent', 'Child'], + reducer: (model: null = null) => model, + *updater(_model: null, action: Action) { + if (action.type === 'Parent') { + yield put({ type: 'Child' }); + } else { + throw failure; + } + }, + }; + const modelSaga = createModelSaga(saga, { + onError: (error: Error) => errors.push(error), + }); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + let rejected: Error | undefined; + try { + await getActionPromise(store.dispatch({ type: 'Parent' })); + } catch (error) { + rejected = error; + } + should.strictEqual(rejected, failure); + should.deepEqual(errors, [failure]); + modelSaga.destroy(); + }); + + it('should wait for all failing sibling updaters and report each failure', async function () { + let release: (() => void) | undefined; + const gate = new Promise((resolve: () => void) => release = resolve); + const firstFailure = new Error('first failed'); + const secondFailure = new Error('second failed'); + const errors: Error[] = []; + const first = { + actionTypes: ['Fail'], + reducer: (model: null = null) => model, + *updater() { + throw firstFailure; + }, + }; + const second = { + actionTypes: ['Fail'], + reducer: (model: null = null) => model, + *updater() { + yield gate; + throw secondFailure; + }, + }; + const modelSaga = createModelSaga(combineSagas({ first, second }), { + onError: (error: Error) => errors.push(error), + }); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + const promise = requireDefined(getActionPromise(store.dispatch({ type: 'Fail' })), 'action promise'); + let settled = false; + promise.catch(() => settled = true); + await Promise.resolve(); + await Promise.resolve(); + should.strictEqual(settled, false); + requireDefined(release, 'second release')(); + let rejected: Error | undefined; + try { + await promise; + } catch (error) { + rejected = error; + } + should.strictEqual(rejected, firstFailure); + should.deepEqual(errors, [firstFailure, secondFailure]); + modelSaga.destroy(); + }); + + it('should keep updater completion unchanged when callbacks throw', async function () { + const callbackFailure = new Error('callback failed'); + const logged: unknown[][] = []; + const originalConsoleError = console.error; + console.error = (...args: unknown[]) => { + logged.push(args); + }; + const saga = { + actionTypes: ['Run'], + reducer: (model: null = null) => model, + *updater() { + return; + }, + }; + const modelSaga = createModelSaga(saga, { + onUpdateComplete: () => { + throw callbackFailure; + }, + onError: () => { + throw callbackFailure; + }, + }); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + try { + await getActionPromise(store.dispatch({ type: 'Run' })); + should.strictEqual(logged.length, 1); + should.strictEqual(logged[0][0], 'Aperol onError callback failed'); + should.strictEqual(logged[0][1], callbackFailure); + } finally { + console.error = originalConsoleError; + modelSaga.destroy(); + } + }); + + it('should report dispatch and observable update completion', async function () { + let emit: ((iterator: AnyIterator) => void) | undefined; + const completions: Array<{ action: string; key: string }> = []; + const observable = new Observable((observer: SubscriptionObserver) => { + emit = (iterator: AnyIterator) => observer.next(iterator); + return () => undefined; + }); + const leaf = { + actionTypes: ['Observe'], + reducer: (model: null = null) => model, + *updater() { + yield observe(observable); + }, + }; + const modelSaga = createModelSaga(combineSagas({ leaf }), { + onUpdateComplete: (_durationMs: number, action: Action, key: string) => { + completions.push({ action: action.type, key }); + }, + }); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + await getActionPromise(store.dispatch({ type: 'Observe' })); + requireDefined(emit, 'observable emitter')((function* () { + return; + })()); + await Promise.resolve(); + await Promise.resolve(); + should.deepEqual(completions, [ + { action: 'Observe', key: '' }, + { action: 'Observe', key: 'leaf' }, + ]); + modelSaga.destroy(); + }); + it('should remove completed observable subscriptions immediately', async function () { let complete: (() => void) | undefined; const synchronous = new Observable((observer: SubscriptionObserver) => { @@ -338,7 +553,34 @@ describe('Application.craeteModelSaga', function () { modelSaga.destroy(); }); - it('should stop in-flight iterators without dispatching post-await work', async function () { + it('should remove manually unsubscribed observables immediately', async function () { + let subscription: Subscription | undefined; + const observable = new Observable((_observer: SubscriptionObserver) => () => undefined); + const saga = { + actionTypes: ['Observe', ObservableSubscribed], + reducer(model: null = null, action: Action) { + if (isObservableSubscribedAction(action)) { + subscription = action.subscription; + } + return model; + }, + *updater(_model: null, action: Action) { + if (action.type === 'Observe') { + yield observe(observable); + } + }, + }; + const modelSaga = createModelSaga(saga); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + await getActionPromise(store.dispatch({ type: 'Observe' })); + should.strictEqual(modelSaga.getSubscriptionCount(), 1); + requireDefined(subscription, 'observable subscription').unsubscribe(); + should.strictEqual(modelSaga.getSubscriptionCount(), 0); + requireDefined(subscription, 'observable subscription').unsubscribe(); + modelSaga.destroy(); + }); + + it('should let an in-flight iteration finish but not start another after destroy', async function () { let release: (() => void) | undefined; const gate = new Promise((resolve: () => void) => release = resolve); const dispatched: Action[] = []; @@ -348,6 +590,7 @@ describe('Application.craeteModelSaga', function () { async *updater() { await gate; yield put({ type: 'TooLate' }); + yield put({ type: 'NeverDispatched' }); }, }; const modelSaga = createModelSaga(saga); @@ -362,6 +605,30 @@ describe('Application.craeteModelSaga', function () { modelSaga.destroy(); requireDefined(release, 'iterator release')(); await getActionPromise(dispatchedAction); - should.strictEqual(dispatched.some((item: Action) => item.type === 'TooLate'), false); + should.strictEqual(dispatched.some((item: Action) => item.type === 'TooLate'), true); + should.strictEqual(dispatched.some((item: Action) => item.type === 'NeverDispatched'), false); + modelSaga.destroy(); + }); + + it('should keep reducing actions but skip new updaters after idempotent destroy', async function () { + const reduced: string[] = []; + const updated: string[] = []; + const saga = { + actionTypes: ['Run'], + reducer(model: null = null, action: Action) { + reduced.push(action.type); + return model; + }, + *updater(_model: null, action: Action) { + updated.push(action.type); + }, + }; + const modelSaga = createModelSaga(saga); + const store = createStore(() => null, applyMiddleware(modelSaga.middleware)); + modelSaga.destroy(); + modelSaga.destroy(); + await getActionPromise(store.dispatch({ type: 'Run' })); + should.strictEqual(reduced.indexOf('Run') !== -1, true); + should.deepEqual(updated, []); }); });