Improve saga dispatch concurrency and cleanup

CHANGELOG.md

@@ -4,6 +4,19 @@ 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
+- 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
+- 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
 - Simple profiler options for saga time usage

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,9 @@ const appSaga = {
 // ... app code
 ```
 
+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
 
@@ -122,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.
 

src/ActionPromise.ts

@@ -0,0 +1,57 @@
+import { Action } from 'redux';
+
+interface ILegacyPromiseAction extends Action {
+	__promise?: Promise<void>;
+}
+
+/**
+ * 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<object, Promise<void>>();
+
+/** @deprecated Prefer `getActionPromise(action)`. Kept so existing `action.__promise` code works. */
+function attachLegacyProperty(action: object, promise: Promise<void>) {
+	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<void>) {
+	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<void> | 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 };

src/ISaga.ts

@@ -5,5 +5,14 @@ import IUpdaterYield from './IUpdaterYield';
 interface ISaga<TModel, TReturn, TNext> {
 	reducer(model: TModel, action: Action): TModel;
 	updater(model: TModel, action: Action): Iterator<IUpdaterYield, TReturn, TNext> | AsyncIterator<IUpdaterYield, TReturn, TNext>;
+	/**
+	 * Every action type this saga's updater reacts to.
+	 *
+	 * `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.
+	 */
+	actionTypes: readonly string[];
 }
 export default ISaga;

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,28 +12,53 @@ export interface IProfiler {
 	track(
 		iterator: AnyIterator,
 		sourceAction: Action,
+		sagaKey: string,
 	): ITracking;
 }
 
 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,
 						);
 					}
 				},

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<TModel> {
+	/**
+	 * 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<TModel>(saga: unknown): saga is ISagaGroup<TModel> {
+	return typeof (saga as ISagaGroup<TModel> | undefined)?.[MATCH_SAGAS] === 'function';
+}