diff --git a/docs/guide.md b/docs/guide.md index 29fb2c5..4fcd6d3 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -57,6 +57,10 @@ All iteration functions (`series`, `filter`, `scan`, `reduce`) support four erro - Sets `failure: false` - Does NOT call `onFailure` +### Source errors + +Errors thrown by the **iteration itself** (e.g. an async generator dying mid-stream) are treated separately from operation errors: they never reach the error strategies' `onError`, and instead are reported through `onSourceError({error, index})` and the additive `sourceErrors` result field — so partial progress survives a dead source under every strategy except `rethrow`. See the [Source errors section in the reference](reference.md#source-errors) for details. + --- ## Features diff --git a/docs/reference.md b/docs/reference.md index 309045b..798fda8 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -247,12 +247,13 @@ Curried: `(fn, opts?) => (items) => Promise` - `onProgress({item, result, index, total})`: Called after each successful item. NOT called for errors or `undefined` drops. - `onError({item, error, index, total})`: Called for each handled item error. Does not affect control flow. - `onFailure(failure)`: Called when failure is truthy. Receives `{item, error, index}` for `failFast`, `{errors}` for `failLate`. + - `onSourceError({error, index})`: Called when the iteration itself throws (a source error). Never called under `rethrow`; mutually exclusive with `onError`. - `take`: Limit items processed. - `total`: Explicit planned input count for progress/error callbacks. If omitted, `series` uses a cheap known input size when available. If `take` is set, callback `total` is limited to `Math.min(take, knownTotal)`. If no total is known, the `total` key is omitted. - `pause`: Milliseconds between successful items. - `pauseOnErrors`: Whether to also pause after errors (default: `false`). -**Return**: `{results, errors, failure}` where `failure` is `false` on success. Collected `errors` are `{item, error, index}`. +**Return**: `{results, errors, sourceErrors, failure}` where `failure` is `false` on success. Collected `errors` are `{item, error, index}`. `sourceErrors` is an array of source-error contexts, empty when the source completed cleanly (see [Source errors](#source-errors)). **Usage Example**: ```javascript @@ -286,6 +287,46 @@ const { results } = await series(items, pipe( )) ``` +#### Source errors + +A **source error** is an error thrown by the iteration itself — the `for await` step (e.g. an async generator dying mid-stream because a DB connection was lost). This is distinct from an **operation error**, which is your `fn` throwing on an item. + +A dead source is terminal: no more items can arrive, so iteration ends under every strategy. The error never reaches `onError`. + +Every return path gains one additive field: + +- `sourceErrors`: array of `{error, index}` contexts — no `item` key, because a source error has no item. `index` is the iteration index of the failed `.next()` call (a generator that yields 2 items then throws produces `index: 2`). Empty array when the source completed cleanly. + +New option: + +- `onSourceError({error, index})`: telemetry callback with the same contract as `onError`. Never called under `rethrow`. `onError` and `onSourceError` are mutually exclusive per error. + +Per-strategy behavior when the source dies: + +| Strategy | Behavior | +| --- | --- | +| `collect` | Results preserved, context recorded in `sourceErrors`, `failure: false` | +| `failFast` | `failure` = the source context, results cleared, accumulator (`value` in reduce mode) preserved | +| `failLate` | Results preserved, source contexts merged into end-of-run aggregation: `failure = {errors: [...errors, ...sourceErrors]}` | +| `skip` | Source errors ignored entirely: `sourceErrors: []`, results preserved, `failure: false` | +| `rethrow` | Throws immediately; no callbacks called | + +```javascript +async function* pages(db) { + while (true) + yield await db.nextPage() // throws when the connection drops +} + +const result = await series(pages(db), processPage, { + strategy: collect, + onSourceError: ({error, index}) => log.warn('source died', {error, index}), +}) +// result.results holds every page processed before the drop +// result.sourceErrors = [{error, index}] +``` + +Cleanup follows native generator semantics: the generator's own `finally` runs during propagation, early exit via `take` still closes the iterator through `iterator.return()`, and if a `finally` itself throws, that error supersedes the original source error. + --- ### scan @@ -302,8 +343,11 @@ const { results } = await series(items, pipe( **Return Type**: A Promise that resolves to an object containing: - `results`: Array of intermediate results (or `[]` on failFast failure) - `errors`: Array of errors encountered (empty for failFast, skip, throw) -- `failure`: `false` on success; `{item, error, index}` for failFast; `{errors}` for failLate -- `value`: Final accumulated value when `storePartialResults: false` (only on success) +- `sourceErrors`: Array of `{error, index}` contexts for errors thrown by the iteration itself; empty when the source completed cleanly +- `failure`: `false` on success; `{item, error, index}` for failFast; `{errors}` for failLate (source errors are merged into the aggregated errors array) +- `value`: Final accumulated value when `storePartialResults: false`. Preserved under every strategy except `rethrow`, even when the source dies mid-stream + +**Source errors**: when the iterable itself throws mid-stream (e.g. an async generator dying), the accumulator and partial results collected so far are preserved per strategy and the context is reported through `onSourceError` instead of `onError`. See [Source errors](#source-errors) under `series` for the full per-strategy behavior. **Key Characteristics**: - **Stateful**: Each transformation depends on the previous result @@ -343,8 +387,9 @@ const { results, errors } = await scan( - `onFailure`: Called when failure is truthy. **Return Type**: A Promise that resolves to an object containing: -- `value`: The final accumulated value (on success). `undefined` on failFast failure. +- `value`: The final accumulated value. Preserved under every strategy except `rethrow`, even when the source dies mid-stream - `errors`: Array of errors encountered (empty for failFast, skip, throw) +- `sourceErrors`: Array of `{error, index}` contexts for errors thrown by the iteration itself; empty when the source completed cleanly - `failure`: `false` on success; `{item, error, index}` for failFast; `{errors}` for failLate **Key Characteristics**: @@ -699,15 +744,17 @@ want Pipelean's structured error collection. **Available sync variants**: -- `seriesSync` returns `{results, errors, failure}` directly -- `filterSync` returns `{results, errors, failure}` directly +- `seriesSync` returns `{results, errors, sourceErrors, failure}` directly +- `filterSync` returns `{results, errors, sourceErrors, failure}` directly - `findSync` returns `{result, errors, failure}` directly and stops at the first match -- `scanSync` returns `{results, errors, failure}` directly -- `reduceSync` returns `{value, errors, failure}` directly +- `scanSync` returns `{results, errors, sourceErrors, failure}` directly +- `reduceSync` returns `{value, errors, sourceErrors, failure}` directly - `pipeSync` composes synchronous functions left-to-right - `flowSync` returns `{value, errors, failure}` directly and runs a state-enrichment pipeline synchronously - `tryCatchSync` wraps a synchronous function with lifecycle hooks +The sync variants handle **source errors** with the same semantics as their async twins: `seriesSync`, `filterSync`, `scanSync`, and `reduceSync` accept `onSourceError({error, index})` and report errors thrown by the iteration itself in the `sourceErrors` field. See [Source errors](#source-errors) under `series`. + > `scanReduceSync` is kept as an alias of `reduceSync` for backward compatibility. **Not supported in sync variants**: diff --git a/src/functional-sync.js b/src/functional-sync.js index 376c85b..68393cd 100644 --- a/src/functional-sync.js +++ b/src/functional-sync.js @@ -1,4 +1,5 @@ /* eslint-disable max-lines */ +/* eslint-disable max-lines-per-function */ import {getPlannedTotal, withTotal} from './shared.js' import { collect, failFast, normalizeOperationError, where, @@ -33,10 +34,11 @@ export const seriesSync = (...args) => { const run = inputItems => { const { strategy = collect, total, - take, onProgress, onError, onFailure, + take, onProgress, onError, onFailure, onSourceError, } = opts const results = [] const errors = [] + const sourceErrors = [] const strategyName = strategy.name ?? strategy const plannedTotal = getPlannedTotal({items: inputItems, take, total}) @@ -49,51 +51,87 @@ export const seriesSync = (...args) => { let index = 0 - for (const item of inputItems) { - if (take !== undefined && index >= take) - break + try { + for (const item of inputItems) { + if (take !== undefined && index >= take) + break + + try { + const result = runFn(item, index) + if (result !== undefined) { + results.push(result) + } + } catch (error) { + if (strategyName === 'throw') { + throw error + } - try { - const result = runFn(item, index) - if (result !== undefined) { - results.push(result) - } - } catch (error) { - if (strategyName === 'throw') { - throw error - } + const errorContext = {item, error, index} - const errorContext = {item, error, index} + if (onError) + onError(withTotal(errorContext, plannedTotal)) - if (onError) - onError(withTotal(errorContext, plannedTotal)) + if (strategyName === 'failFast') { + if (onFailure) { + onFailure(errorContext) + } + return { + results: [], errors, sourceErrors, failure: errorContext, + } + } - if (strategyName === 'failFast') { - if (onFailure) { - onFailure(errorContext) + if (strategyName === 'skip') { + index++ + continue } - return {results: [], errors, failure: errorContext} + + errors.push(errorContext) } + index++ + } + } catch (error) { + if (strategyName === 'throw') { + throw error + } - if (strategyName === 'skip') { - index++ - continue + const sourceContext = {error, index} + + if (onSourceError) + onSourceError(sourceContext) + + if (strategyName === 'failFast') { + if (onFailure) { + onFailure(sourceContext) } + return { + results: [], + errors, + sourceErrors: [sourceContext], + failure: sourceContext, + } + } - errors.push(errorContext) + if (strategyName === 'skip') { + return { + results, errors, sourceErrors: [], failure: false, + } } - index++ + + sourceErrors.push(sourceContext) } - const failure = strategyName === 'failLate' && errors.length > 0 - ? {errors} + const failure = strategyName === 'failLate' && + (errors.length > 0 || sourceErrors.length > 0) + ? {errors: [...errors, ...sourceErrors]} : false if (failure && onFailure) { onFailure(failure) } - return {results, errors, failure} + return { + results, errors, sourceErrors, failure, + } } return immediate ? run(items) : run @@ -193,53 +231,103 @@ export const findSync = (...args) => { // eslint-disable-next-line complexity, max-statements export const scanSync = (iterable, scanner, initialValue, opts = {}) => { const { - strategy = failFast, onError, onFailure, storePartialResults = true, + strategy = failFast, onError, onFailure, onSourceError, + storePartialResults = true, } = opts const results = [] let acc = initialValue const errors = [] + const sourceErrors = [] const strategyName = strategy.name ?? strategy const plannedTotal = getPlannedTotal({items: iterable}) let index = 0 - for (const item of iterable) { - try { - acc = scanner(acc, item, index) - if (storePartialResults) - results.push(acc) - } catch (error) { - const errorContext = {item, error, index} + try { + for (const item of iterable) { + try { + acc = scanner(acc, item, index) + if (storePartialResults) + results.push(acc) + } catch (error) { + const errorContext = {item, error, index} - if (strategyName === 'throw') { - throw error - } + if (strategyName === 'throw') { + throw error + } - if (onError) { - onError(withTotal(errorContext, plannedTotal)) - } + if (onError) { + onError(withTotal(errorContext, plannedTotal)) + } - if (strategyName === 'failFast') { - if (onFailure) { - onFailure(errorContext) + if (strategyName === 'failFast') { + if (onFailure) { + onFailure(errorContext) + } + return storePartialResults + ? { + results: [], errors, sourceErrors, failure: errorContext, + } + : { + value: acc, errors, sourceErrors, failure: errorContext, + } + } + + if (strategyName === 'skip') { + index++ + continue } - return storePartialResults - ? {results: [], errors, failure: errorContext} - : {value: acc, errors, failure: errorContext} + + errors.push(errorContext) } + index++ + } + } catch (error) { + if (strategyName === 'throw') { + throw error + } - if (strategyName === 'skip') { - index++ - continue + const sourceContext = {error, index} + + if (onSourceError) { + onSourceError(sourceContext) + } + + if (strategyName === 'failFast') { + if (onFailure) { + onFailure(sourceContext) } + return storePartialResults + ? { + results: [], + errors, + sourceErrors: [sourceContext], + failure: sourceContext, + } + : { + value: acc, + errors, + sourceErrors: [sourceContext], + failure: sourceContext, + } + } - errors.push(errorContext) + if (strategyName === 'skip') { + return storePartialResults + ? { + results, errors, sourceErrors: [], failure: false, + } + : { + value: acc, errors, sourceErrors: [], failure: false, + } } - index++ + + sourceErrors.push(sourceContext) } const failure = - strategyName === 'failLate' && errors.length > 0 - ? {errors} + strategyName === 'failLate' && + (errors.length > 0 || sourceErrors.length > 0) + ? {errors: [...errors, ...sourceErrors]} : false if (failure && onFailure) { @@ -247,8 +335,12 @@ export const scanSync = (iterable, scanner, initialValue, opts = {}) => { } return storePartialResults - ? {results, errors, failure} - : {value: acc, errors, failure} + ? { + results, errors, sourceErrors, failure, + } + : { + value: acc, errors, sourceErrors, failure, + } } export const reduceSync = (iterable, scanner, initialValue, opts) => { diff --git a/src/functional.js b/src/functional.js index 32b12da..3198926 100644 --- a/src/functional.js +++ b/src/functional.js @@ -1,4 +1,5 @@ /* eslint-disable max-lines */ +/* eslint-disable max-lines-per-function */ import {getPlannedTotal, withTotal} from './shared.js' export const failFast = Object.freeze({name: 'failFast'}) @@ -62,7 +63,6 @@ export const assign = (property, parse) => state => { return value === undefined ? {} : {[property]: value} } -// eslint-disable-next-line max-lines-per-function export const series = (...args) => { const immediate = typeof args[0] !== 'function' const [items, fn, opts = {}] = immediate ? args : [null, args[0], args[1]] @@ -71,10 +71,12 @@ export const series = (...args) => { const run = async inputItems => { const { strategy = collect, total, - take, onProgress, onError, onFailure, pause, pauseOnErrors = false, + take, onProgress, onError, onFailure, onSourceError, + pause, pauseOnErrors = false, } = opts const results = [] const errors = [] + const sourceErrors = [] const strategyName = strategy.name ?? strategy const plannedTotal = getPlannedTotal({items: inputItems, take, total}) @@ -87,64 +89,100 @@ export const series = (...args) => { let index = 0 - for await (const item of inputItems) { - if (take !== undefined && index >= take) - break + try { + for await (const item of inputItems) { + if (take !== undefined && index >= take) + break + + try { + const result = await runFn(item, index) + // undefined is the sentinel value for "drop this item". + // This enables selection/filtering within pipes and + // is how filter() works internally. + if (result !== undefined) { + results.push(result) + } + // Pause after successful item + if (pause) { + await delay(pause) + } + } catch (error) { + if (strategyName === 'throw') { + throw error + } - try { - const result = await runFn(item, index) - // undefined is the sentinel value for "drop this item". - // This enables selection/filtering within pipes and - // is how filter() works internally. - if (result !== undefined) { - results.push(result) - } - // Pause after successful item - if (pause) { - await delay(pause) - } - } catch (error) { - if (strategyName === 'throw') { - throw error - } + const errorContext = {item, error, index} - const errorContext = {item, error, index} + if (onError) + await onError(withTotal(errorContext, plannedTotal)) - if (onError) - await onError(withTotal(errorContext, plannedTotal)) + if (strategyName === 'failFast') { + if (onFailure) { + onFailure(errorContext) + } + return { + results: [], errors, sourceErrors, failure: errorContext, + } + } - if (strategyName === 'failFast') { - if (onFailure) { - onFailure(errorContext) + if (strategyName === 'skip') { + index++ + if (pause) { + await delay(pause) + } + continue } - return {results: [], errors, failure: errorContext} - } - if (strategyName === 'skip') { - index++ - if (pause) { + errors.push(errorContext) + if (pause && pauseOnErrors) { await delay(pause) } - continue } + index++ + } + } catch (error) { + if (strategyName === 'throw') { + throw error + } - errors.push(errorContext) - if (pause && pauseOnErrors) { - await delay(pause) + const sourceContext = {error, index} + + if (onSourceError) + onSourceError(sourceContext) + + if (strategyName === 'failFast') { + if (onFailure) { + onFailure(sourceContext) + } + return { + results: [], + errors, + sourceErrors: [sourceContext], + failure: sourceContext, } } - index++ + + if (strategyName === 'skip') { + return { + results, errors, sourceErrors: [], failure: false, + } + } + + sourceErrors.push(sourceContext) } - const failure = strategyName === 'failLate' && errors.length > 0 - ? {errors} + const failure = strategyName === 'failLate' && + (errors.length > 0 || sourceErrors.length > 0) + ? {errors: [...errors, ...sourceErrors]} : false if (failure && onFailure) { onFailure(failure) } - return {results, errors, failure} + return { + results, errors, sourceErrors, failure, + } } return immediate ? run(items) : run @@ -192,53 +230,103 @@ export const filter = (...args) => { // eslint-disable-next-line complexity, max-statements export const scan = async (iterable, scanner, initialValue, opts = {}) => { const { - strategy = failFast, onError, onFailure, storePartialResults = true, + strategy = failFast, onError, onFailure, onSourceError, + storePartialResults = true, } = opts const results = [] let acc = initialValue const errors = [] + const sourceErrors = [] const strategyName = strategy.name ?? strategy const plannedTotal = getPlannedTotal({items: iterable}) let index = 0 - for await (const item of iterable) { - try { - acc = await scanner(acc, item, index) - if (storePartialResults) - results.push(acc) - } catch (error) { - const errorContext = {item, error, index} + try { + for await (const item of iterable) { + try { + acc = await scanner(acc, item, index) + if (storePartialResults) + results.push(acc) + } catch (error) { + const errorContext = {item, error, index} - if (strategyName === 'throw') { - throw error - } + if (strategyName === 'throw') { + throw error + } - if (onError) { - await onError(withTotal(errorContext, plannedTotal)) - } + if (onError) { + await onError(withTotal(errorContext, plannedTotal)) + } - if (strategyName === 'failFast') { - if (onFailure) { - onFailure(errorContext) + if (strategyName === 'failFast') { + if (onFailure) { + onFailure(errorContext) + } + return storePartialResults + ? { + results: [], errors, sourceErrors, failure: errorContext, + } + : { + value: acc, errors, sourceErrors, failure: errorContext, + } } - return storePartialResults - ? {results: [], errors, failure: errorContext} - : {value: acc, errors, failure: errorContext} + + if (strategyName === 'skip') { + index++ + continue + } + + errors.push(errorContext) } + index++ + } + } catch (error) { + if (strategyName === 'throw') { + throw error + } - if (strategyName === 'skip') { - index++ - continue + const sourceContext = {error, index} + + if (onSourceError) { + onSourceError(sourceContext) + } + + if (strategyName === 'failFast') { + if (onFailure) { + onFailure(sourceContext) } + return storePartialResults + ? { + results: [], + errors, + sourceErrors: [sourceContext], + failure: sourceContext, + } + : { + value: acc, + errors, + sourceErrors: [sourceContext], + failure: sourceContext, + } + } - errors.push(errorContext) + if (strategyName === 'skip') { + return storePartialResults + ? { + results, errors, sourceErrors: [], failure: false, + } + : { + value: acc, errors, sourceErrors: [], failure: false, + } } - index++ + + sourceErrors.push(sourceContext) } const failure = - strategyName === 'failLate' && errors.length > 0 - ? {errors} + strategyName === 'failLate' && + (errors.length > 0 || sourceErrors.length > 0) + ? {errors: [...errors, ...sourceErrors]} : false if (failure && onFailure) { @@ -246,8 +334,12 @@ export const scan = async (iterable, scanner, initialValue, opts = {}) => { } return storePartialResults - ? {results, errors, failure} - : {value: acc, errors, failure} + ? { + results, errors, sourceErrors, failure, + } + : { + value: acc, errors, sourceErrors, failure, + } } export const reduce = (iterable, scanner, initialValue, opts = {}) => diff --git a/tests/filter-sync.test.js b/tests/filter-sync.test.js index 0de55fd..d5b5694 100644 --- a/tests/filter-sync.test.js +++ b/tests/filter-sync.test.js @@ -3,12 +3,16 @@ import {filterSync} from '$src/index' test('predicate truthy keeps item in results', () => { const result = filterSync([1, 2, 3, 4], x => x > 2) - expect(result).toEqual({results: [3, 4], errors: [], failure: false}) + expect(result).toEqual({ + results: [3, 4], errors: [], sourceErrors: [], failure: false, + }) }) test('predicate falsy excludes item without error', () => { const result = filterSync([1, 2, 3], () => false) - expect(result).toEqual({results: [], errors: [], failure: false}) + expect(result).toEqual({ + results: [], errors: [], sourceErrors: [], failure: false, + }) }) test('predicate throws with failFast stops and populates failure', () => { @@ -46,12 +50,16 @@ test('curried form returns a function', () => { test('curried form executes when called with items', () => { const evens = filterSync(x => x % 2 === 0) const result = evens([1, 2, 3, 4]) - expect(result).toEqual({results: [2, 4], errors: [], failure: false}) + expect(result).toEqual({ + results: [2, 4], errors: [], sourceErrors: [], failure: false, + }) }) test('empty array returns empty result shape', () => { const result = filterSync([], () => true) - expect(result).toEqual({results: [], errors: [], failure: false}) + expect(result).toEqual({ + results: [], errors: [], sourceErrors: [], failure: false, + }) }) test('returns value synchronously not a promise', () => { diff --git a/tests/filter.test.js b/tests/filter.test.js index 78028bd..3f2704a 100644 --- a/tests/filter.test.js +++ b/tests/filter.test.js @@ -3,12 +3,16 @@ import {filter} from '$src/functional' test('predicate truthy keeps item in results', async () => { const result = await filter([1, 2, 3, 4], x => x > 2) - expect(result).toEqual({results: [3, 4], errors: [], failure: false}) + expect(result).toEqual({ + results: [3, 4], errors: [], sourceErrors: [], failure: false, + }) }) test('predicate falsy excludes item without error', async () => { const result = await filter([1, 2, 3], () => false) - expect(result).toEqual({results: [], errors: [], failure: false}) + expect(result).toEqual({ + results: [], errors: [], sourceErrors: [], failure: false, + }) }) test('predicate throws with failFast stops and populates failure', async () => { @@ -51,10 +55,14 @@ test('curried form returns a function', () => { test('curried form executes when called with items', async () => { const evens = filter(x => x % 2 === 0) const result = await evens([1, 2, 3, 4]) - expect(result).toEqual({results: [2, 4], errors: [], failure: false}) + expect(result).toEqual({ + results: [2, 4], errors: [], sourceErrors: [], failure: false, + }) }) test('empty array returns empty result shape', async () => { const result = await filter([], () => true) - expect(result).toEqual({results: [], errors: [], failure: false}) + expect(result).toEqual({ + results: [], errors: [], sourceErrors: [], failure: false, + }) }) diff --git a/tests/series-pipe.test.js b/tests/series-pipe.test.js index 71bb45b..5315b9a 100644 --- a/tests/series-pipe.test.js +++ b/tests/series-pipe.test.js @@ -17,6 +17,7 @@ test('accept a pipe with a single mapping function', async () => { expect(result).toEqual({ results: [2, 4, 6], errors: [], + sourceErrors: [], failure: false, }) }) @@ -31,6 +32,7 @@ test('accept a pipe with multiple mapping functions', async () => { expect(result).toEqual({ results: [3, 5, 7], errors: [], + sourceErrors: [], failure: false, }) }) @@ -45,6 +47,7 @@ test('curried functions inside the pipe', async () => { expect(result).toEqual({ results: [20, 30, 40], // (1+1)*10, etc. errors: [], + sourceErrors: [], failure: false, }) }) @@ -68,6 +71,7 @@ test( expect(result).toEqual({ results: [3, 5, 7, 9, 11, 13], errors: [], + sourceErrors: [], failure: false, }) }, @@ -91,6 +95,7 @@ test('mixed mapping and filtering logic within a pipe', async () => { expect(result).toEqual({ results: [5, 9, 13], // inputs: 2, 4, 6 errors: [], + sourceErrors: [], failure: false, }) }) diff --git a/tests/series-sync.test.js b/tests/series-sync.test.js index a159120..718fb8b 100644 --- a/tests/series-sync.test.js +++ b/tests/series-sync.test.js @@ -3,7 +3,9 @@ import {seriesSync, collect} from '$src/index' test('all items succeed returns results with no errors', () => { const result = seriesSync([1, 2, 3], x => x * 2) - expect(result).toEqual({results: [2, 4, 6], errors: [], failure: false}) + expect(result).toEqual({ + results: [2, 4, 6], errors: [], sourceErrors: [], failure: false, + }) }) test('failFast stops on first error with no partial results', () => { @@ -41,7 +43,9 @@ test('passes index as second arg to fn', () => { test('empty array returns empty result shape', () => { const result = seriesSync([], x => x) - expect(result).toEqual({results: [], errors: [], failure: false}) + expect(result).toEqual({ + results: [], errors: [], sourceErrors: [], failure: false, + }) }) test('curried form returns a function', () => { @@ -52,7 +56,9 @@ test('curried form returns a function', () => { test('curried form executes when called with items', () => { const double = seriesSync(x => x * 2) const result = double([1, 2, 3]) - expect(result).toEqual({results: [2, 4, 6], errors: [], failure: false}) + expect(result).toEqual({ + results: [2, 4, 6], errors: [], sourceErrors: [], failure: false, + }) }) test('calls onProgress after each successful item', () => { diff --git a/tests/series.test.js b/tests/series.test.js index 6704da2..6c33d45 100644 --- a/tests/series.test.js +++ b/tests/series.test.js @@ -4,7 +4,9 @@ import {series, collect} from '$src/functional' test('all items succeed returns results with no errors', async () => { const result = await series([1, 2, 3], x => x * 2) - expect(result).toEqual({results: [2, 4, 6], errors: [], failure: false}) + expect(result).toEqual({ + results: [2, 4, 6], errors: [], sourceErrors: [], failure: false, + }) }) test('failFast stops on first error with no partial results', async () => { @@ -47,7 +49,9 @@ test('passes index as second arg to fn', async () => { test('empty array returns empty result shape', async () => { const result = await series([], x => x) - expect(result).toEqual({results: [], errors: [], failure: false}) + expect(result).toEqual({ + results: [], errors: [], sourceErrors: [], failure: false, + }) }) test('curried form returns a function', () => { @@ -58,7 +62,9 @@ test('curried form returns a function', () => { test('curried form executes when called with items', async () => { const double = series(x => x * 2) const result = await double([1, 2, 3]) - expect(result).toEqual({results: [2, 4, 6], errors: [], failure: false}) + expect(result).toEqual({ + results: [2, 4, 6], errors: [], sourceErrors: [], failure: false, + }) }) test('series with pause waits between successful items', async () => { diff --git a/tests/source-errors-helpers.js b/tests/source-errors-helpers.js new file mode 100644 index 0000000..f2ae342 --- /dev/null +++ b/tests/source-errors-helpers.js @@ -0,0 +1,32 @@ +/* eslint-disable no-unsafe-finally */ +export const fragileSource = (values, error) => (async function * () { + for (const value of values) + yield value + throw error +})() + +export const trackedSource = values => { + const state = {cleanedUp: false, yielded: []} + const gen = async function * () { + try { + for (const value of values) { + state.yielded.push(value) + yield value + } + } finally { + state.cleanedUp = true + } + } + return {gen, state} +} + +export const badCleanupSource = (values, sourceError, cleanupError) => + (async function * () { + try { + for (const value of values) + yield value + throw sourceError + } finally { + throw cleanupError + } + })() diff --git a/tests/source-errors-sync.test.js b/tests/source-errors-sync.test.js new file mode 100644 index 0000000..7952fe9 --- /dev/null +++ b/tests/source-errors-sync.test.js @@ -0,0 +1,273 @@ +/* eslint-disable max-lines */ +/* eslint-disable @stylistic/max-len, require-yield */ +import {test, expect, vi} from 'vitest' +import { + collect, + failFast, + failLate, + skip, + rethrow, + seriesSync, + scanSync, + reduceSync, + filterSync, +} from '$src/index' + +const fragileSource = (values, error) => (function * () { + for (const value of values) + yield value + throw error +})() + +test('collect: source death preserves results and records sourceErrors', () => { + const boom = new Error('sync source died') + const items = fragileSource([1, 2], boom) + + const result = seriesSync(items, x => x * 10, {strategy: collect}) + + expect(result.results).toEqual([10, 20]) + expect(result.errors).toEqual([]) + expect(result.sourceErrors).toEqual([{error: boom, index: 2}]) + expect(result.failure).toBe(false) +}) + +test('failFast: failure is the source context and results are cleared', () => { + const boom = new Error('stop the line') + const items = fragileSource([1, 2], boom) + + const result = seriesSync(items, x => x * 10, {strategy: failFast}) + + expect(result.results).toEqual([]) + expect(result.sourceErrors).toEqual([{error: boom, index: 2}]) + expect(result.failure).toEqual({error: boom, index: 2}) +}) + +test('failLate: source context merges into end-of-run failure with op errors', () => { + const opError = new Error('op failed') + const srcError = new Error('source died') + + const gen = function * () { + yield 1 + yield 2 + throw srcError + } + const fn = x => { + if (x === 1) + throw opError + return x + } + + const result = seriesSync(gen(), fn, {strategy: failLate}) + + expect(result.errors).toHaveLength(1) + expect(result.errors[0].error).toBe(opError) + expect(result.sourceErrors).toEqual([{error: srcError, index: 2}]) + expect(result.failure).toEqual({ + errors: [...result.errors, ...result.sourceErrors], + }) +}) + +test('skip: source errors are ignored entirely', () => { + const boom = new Error('skipped death') + const items = fragileSource([1, 2], boom) + + const result = seriesSync(items, x => x * 10, {strategy: skip}) + + expect(result.results).toEqual([10, 20]) + expect(result.errors).toEqual([]) + expect(result.sourceErrors).toEqual([]) + expect(result.failure).toBe(false) +}) + +test('rethrow: source error propagates and onSourceError is not called', () => { + const boom = new Error('raw sync throw') + const onSourceError = vi.fn() + const items = fragileSource([1], boom) + + expect(() => + seriesSync(items, x => x, {strategy: rethrow, onSourceError})).toThrow(boom) + expect(onSourceError).not.toHaveBeenCalled() +}) + +test('onSourceError is called once and onError never, under collect', () => { + const boom = new Error('telemetry') + const onError = vi.fn() + const onSourceError = vi.fn() + const items = fragileSource(['a', 'b'], boom) + + const result = seriesSync(items, x => x, { + strategy: collect, + onError, + onSourceError, + }) + + expect(onSourceError).toHaveBeenCalledTimes(1) + expect(onSourceError).toHaveBeenCalledWith({error: boom, index: 2}) + expect(onError).not.toHaveBeenCalled() + expect(result.sourceErrors).toEqual([{error: boom, index: 2}]) +}) + +test('onFailure receives the source context under failFast', () => { + const boom = new Error('failed fast sync') + const onFailure = vi.fn() + const items = fragileSource(['only'], boom) + + seriesSync(items, x => x, {strategy: failFast, onFailure}) + + expect(onFailure).toHaveBeenCalledTimes(1) + expect(onFailure).toHaveBeenCalledWith({error: boom, index: 1}) +}) + +test('reduceSync: value carries the last accumulator before death', () => { + const boom = new Error('reducer lost') + + const gen = function * () { + yield 1 + yield 2 + yield 3 + throw boom + } + + const collected = reduceSync(gen(), (acc, x) => acc + x, 0, { + strategy: collect, + }) + expect(collected.value).toBe(6) + expect(collected.sourceErrors).toEqual([{error: boom, index: 3}]) + expect(collected.failure).toBe(false) + + const stopped = reduceSync(fragileSource([1, 2], boom), (acc, x) => acc + x, 0, { + strategy: failFast, + }) + expect(stopped.value).toBe(3) + expect(stopped.failure).toEqual({error: boom, index: 2}) +}) + +test('scanSync mode: partial results preserved per strategy', () => { + const boom = new Error('scan lost') + + const gen = function * () { + yield 1 + yield 2 + throw boom + } + + const collectedResult = scanSync(gen(), (acc, x) => acc + x, 0, { + strategy: collect, + }) + expect(collectedResult.results).toEqual([1, 3]) + expect(collectedResult.sourceErrors).toEqual([{error: boom, index: 2}]) + + const skipResult = scanSync(fragileSource([1, 2], boom), (acc, x) => acc + x, 0, { + strategy: skip, + }) + expect(skipResult.results).toEqual([1, 3]) + expect(skipResult.sourceErrors).toEqual([]) + expect(skipResult.failure).toBe(false) +}) + +test('filterSync: filtered results preserved and sourceErrors captured', () => { + const boom = new Error('filter stream broke') + const items = fragileSource([1, 2, 3, 4], boom) + const keepEvens = filterSync(x => x % 2 === 0, {strategy: collect}) + + const result = keepEvens(items) + + expect(result.results).toEqual([2, 4]) + expect(result.sourceErrors).toEqual([{error: boom, index: 4}]) + expect(result.failure).toBe(false) +}) + +test('immediate failure: source throwing before first yield has index 0', () => { + const boom = new Error('never yielded sync') + + const gen = function * () { + throw boom + } + + const result = seriesSync(gen(), x => x, {strategy: collect}) + + expect(result.results).toEqual([]) + expect(result.sourceErrors).toEqual([{error: boom, index: 0}]) +}) + +test('late failure: throw after the last item processed', () => { + const boom = new Error('one too many sync') + + const gen = function * () { + yield 1 + throw boom + } + + const result = seriesSync(gen(), x => x * 2, {strategy: collect}) + + expect(result.results).toEqual([2]) + expect(result.sourceErrors).toEqual([{error: boom, index: 1}]) +}) + +test('take reached before the throw: no source errors and iterator.return called', () => { + const boom = new Error('never reached') + let returned = false + + const gen = function * () { + try { + yield 1 + yield 2 + yield 3 + throw boom + } finally { + returned = true + } + } + + const fn = vi.fn(x => x * 10) + const result = seriesSync(gen(), fn, {take: 2}) + + expect(result.results).toEqual([10, 20]) + expect(fn).toHaveBeenCalledTimes(2) + expect(result.sourceErrors).toEqual([]) + expect(returned).toBe(true) +}) + +test('generator finally runs when the source dies mid-stream', () => { + const boom = new Error('mid-stream sync death') + let cleanupRan = false + + const gen = function * () { + try { + yield 1 + yield 2 + throw boom + } finally { + cleanupRan = true + } + } + + const result = seriesSync(gen(), x => x, {strategy: collect}) + + expect(result.results).toEqual([1, 2]) + expect(cleanupRan).toBe(true) + expect(result.sourceErrors[0].error).toBe(boom) +}) + +test('bad cleanup: finally throwing supersedes the source error', () => { + const original = new Error('original sync death') + const cleanup = new Error('cleanup exploded sync') + + const gen = function * () { + try { + yield 1 + yield 2 + throw original + } finally { + // eslint-disable-next-line no-unsafe-finally + throw cleanup + } + } + + const result = seriesSync(gen(), x => x * 10, {strategy: collect}) + + expect(result.results).toEqual([10, 20]) + expect(result.sourceErrors).toHaveLength(1) + expect(result.sourceErrors[0].index).toBe(2) + expect(result.sourceErrors[0].error).toBe(cleanup) +}) diff --git a/tests/source-errors.test.js b/tests/source-errors.test.js new file mode 100644 index 0000000..2d4a352 --- /dev/null +++ b/tests/source-errors.test.js @@ -0,0 +1,269 @@ +/* eslint-disable max-lines */ +/* eslint-disable require-await, @stylistic/max-len, require-yield */ +import {test, expect, vi} from 'vitest' +import { + collect, + failFast, + failLate, + skip, + rethrow, + series, + scan, + reduce, + filter, +} from '$src/functional' +import { + fragileSource, + trackedSource, + badCleanupSource, +} from './source-errors-helpers' + +test('collect: source death preserves results and records sourceErrors', async () => { + const boom = new Error('db connection lost') + const items = fragileSource([1, 2], boom) + + const result = await series(items, async x => x * 10, {strategy: collect}) + + expect(result.results).toEqual([10, 20]) + expect(result.errors).toEqual([]) + expect(result.sourceErrors).toEqual([{error: boom, index: 2}]) + expect(result.failure).toBe(false) +}) + +test('failFast: failure is the source context and results are cleared', async () => { + const boom = new Error('stream died') + const items = fragileSource([1, 2], boom) + + const result = await series(items, async x => x * 10, {strategy: failFast}) + + expect(result.results).toEqual([]) + expect(result.sourceErrors).toEqual([{error: boom, index: 2}]) + expect(result.failure).toEqual({error: boom, index: 2}) +}) + +test('failLate: source context merges into end-of-run failure', async () => { + const opError = new Error('op failed') + const srcError = new Error('source died') + + const gen = async function * () { + yield 1 + yield 2 + throw srcError + } + const fn = async x => { + if (x === 1) + throw opError + return x + } + + const result = await series(gen(), fn, {strategy: failLate}) + + expect(result.errors).toHaveLength(1) + expect(result.errors[0].error).toBe(opError) + expect(result.sourceErrors).toEqual([{error: srcError, index: 2}]) + expect(result.failure).toEqual({ + errors: [...result.errors, ...result.sourceErrors], + }) +}) + +test('skip: source errors are ignored entirely', async () => { + const boom = new Error('gone') + const items = fragileSource([1, 2], boom) + + const result = await series(items, async x => x * 10, {strategy: skip}) + + expect(result.results).toEqual([10, 20]) + expect(result.errors).toEqual([]) + expect(result.sourceErrors).toEqual([]) + expect(result.failure).toBe(false) +}) + +test('rethrow: source error propagates and onSourceError is not called', async () => { + const boom = new Error('raw throw') + const onSourceError = vi.fn() + const items = fragileSource([1], boom) + + await expect( + series(items, async x => x, {strategy: rethrow, onSourceError}), + ).rejects.toBe(boom) + expect(onSourceError).not.toHaveBeenCalled() +}) + +test('onSourceError is called once with the context under collect', async () => { + const boom = new Error('telemetry check') + const onError = vi.fn() + const onSourceError = vi.fn() + const items = fragileSource([1, 2], boom) + + const result = await series(items, async x => x, { + strategy: collect, + onError, + onSourceError, + }) + + expect(onSourceError).toHaveBeenCalledTimes(1) + expect(onSourceError).toHaveBeenCalledWith({error: boom, index: 2}) + expect(onError).not.toHaveBeenCalled() + expect(result.sourceErrors).toEqual([{error: boom, index: 2}]) +}) + +test('onFailure receives the source context under failFast', async () => { + const boom = new Error('failed fast') + const onFailure = vi.fn() + const items = fragileSource(['a'], boom) + + await series(items, async x => x, {strategy: failFast, onFailure}) + + expect(onFailure).toHaveBeenCalledTimes(1) + expect(onFailure).toHaveBeenCalledWith({error: boom, index: 1}) +}) + +test('reduce: value carries the last accumulator before death', async () => { + const boom = new Error('reducer stream lost') + + const gen = async function * () { + yield 1 + yield 2 + throw boom + } + + const collected = await reduce(gen(), async (acc, x) => acc + x, 0, { + strategy: collect, + }) + expect(collected.value).toBe(3) + expect(collected.errors).toEqual([]) + expect(collected.sourceErrors).toEqual([{error: boom, index: 2}]) + expect(collected.failure).toBe(false) +}) + +test('reduce failFast: value preserved, failure is the source context', async () => { + const boom = new Error('stop now') + + const gen = async function * () { + yield 1 + yield 2 + throw boom + } + + const result = await reduce(gen(), async (acc, x) => acc + x, 0, { + strategy: failFast, + }) + + expect(result.value).toBe(3) + expect(result.sourceErrors).toEqual([{error: boom, index: 2}]) + expect(result.failure).toEqual({error: boom, index: 2}) +}) + +test('scan mode: partial results preserved per strategy', async () => { + const boom = new Error('scan dies') + + const gen = async function * () { + yield 1 + yield 2 + throw boom + } + + const collectedResult = await scan(gen(), async (acc, x) => acc + x, 0, { + strategy: collect, + }) + expect(collectedResult.results).toEqual([1, 3]) + expect(collectedResult.sourceErrors).toEqual([{error: boom, index: 2}]) + + const skipResult = await scan(fragileSource([1, 2], boom), async (acc, x) => acc + x, 0, { + strategy: skip, + }) + expect(skipResult.results).toEqual([1, 3]) + expect(skipResult.sourceErrors).toEqual([]) + expect(skipResult.failure).toBe(false) +}) + +test('filter: filtered results preserved and sourceErrors captured', async () => { + const boom = new Error('filtered stream broke') + const items = fragileSource([1, 2, 3, 4], boom) + const keepEvens = filter(async x => x % 2 === 0, {strategy: collect}) + + const result = await keepEvens(items) + + expect(result.results).toEqual([2, 4]) + expect(result.sourceErrors).toEqual([{error: boom, index: 4}]) + expect(result.failure).toBe(false) +}) + +test('immediate failure: source throwing before first yield has index 0', async () => { + const boom = new Error('never yielded') + + const gen = async function * () { + throw boom + } + + const result = await series(gen(), async x => x, {strategy: collect}) + + expect(result.results).toEqual([]) + expect(result.sourceErrors).toEqual([{error: boom, index: 0}]) +}) + +test('late failure: throw after the last item processed', async () => { + const boom = new Error('one too many') + + const gen = async function * () { + yield 1 + throw boom + } + + const result = await series(gen(), async x => x * 2, {strategy: collect}) + + expect(result.results).toEqual([2]) + expect(result.sourceErrors).toEqual([{error: boom, index: 1}]) +}) + +test('take reached before the throw: no source errors and generator closed', async () => { + const boom = new Error('never reached') + const {gen, state} = trackedSource([1, 2, 3]) + + const result = await series(gen(), async x => x * 10, {take: 2}) + + expect(result.results).toEqual([10, 20]) + expect(result.sourceErrors).toEqual([]) + expect(result.failure).toBe(false) + expect(state.cleanedUp).toBe(true) + + const dyingSource = fragileSource([1, 2], boom) + const fn = vi.fn(async x => x) + await series(dyingSource, fn, {take: 2}) + expect(fn).toHaveBeenCalledTimes(2) +}) + +test('generator finally runs when the source dies mid-stream', async () => { + const boom = new Error('mid-stream death') + let cleanupRan = false + + const gen = async function * () { + try { + yield 1 + yield 2 + throw boom + } finally { + cleanupRan = true + } + } + + const result = await series(gen(), async x => x, {strategy: collect}) + + expect(result.results).toEqual([1, 2]) + expect(cleanupRan).toBe(true) + expect(result.sourceErrors[0].error).toBe(boom) +}) + +test('bad cleanup: finally throwing supersedes the source error', async () => { + const original = new Error('original death') + const cleanup = new Error('cleanup exploded') + const items = badCleanupSource([1, 2], original, cleanup) + + const result = await series(items, async x => x * 10, {strategy: collect}) + + expect(result.results).toEqual([10, 20]) + expect(result.sourceErrors).toHaveLength(1) + expect(result.sourceErrors[0].index).toBe(2) + expect(result.sourceErrors[0].error).toBe(cleanup) + expect(result.sourceErrors[0].error).not.toBe(original) +})