Scan error improvement

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

docs/reference.md

@@ -247,12 +247,13 @@ Curried: `(fn, opts?) => (items) => Promise<Outcome>`
   - `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**:

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,62 +231,116 @@ 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) {
     onFailure(failure)
   }
 
   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) => {