[Feat] Error handler in streams

[sleipnir]

🎯 Context

This PR introduces a new error handling and side-effect model for the GRPC.Stream module, covering both unary and streaming flows.
The implementation improves safety, consistency, and expressiveness across Flow-based pipelines used in gRPC message processing, ensuring that exceptions, errors, and inconsistent return values are handled uniformly and predictably.

---

🚀 New Features

🧩 1. effect/2

A new utility function to safely perform side effects within a stream pipeline without modifying the values being processed.

GRPC.Stream.from(....)
|> GRPC.Stream.effect(fn x -> Logger.debug("Processing #{x}") end)
|> GRPC.Stream.run()

• Ensures exceptions raised inside the effect function do not break the stream.
• Uses the internal safe_invoke/2 mechanism to capture exceptions and thrown values.
• Behaves like a safe tap/2, preserving all original stream data.

---

⚙️ 2. map_error/2

Adds a declarative mechanism for mapping or transforming errors ({:error, reason}) within a stream into structured GRPC.RPCError values.

GRPC.Stream.from([1, 2])
|> GRPC.Stream.map(fn
  2 -> raise "boom"
  x -> x
end)
|> GRPC.Stream.map_error(fn
  {:error, reason} ->
    GRPC.RPCError.exception(message: "Validation or runtime error: #{inspect(reason)}")
  other ->
    other
end)

Behavior:

• {:error, reason} → transformed into %GRPC.RPCError{}
• {:ok, value} → unwrapped to value
• exceptions from previous operators (e.g. map/2) → can be transformed into domain-specific GRPC.RPCError
• normal values are propagated unchanged

This enables localized error recovery and translation, useful for both input validation and unexpected runtime exceptions.

---

🔄 3. Unified Error Matching and Propagation

All stream operators now use the internal safe_invoke/2 wrapper, which standardizes how functions inside the pipeline behave.

Function result	Propagated value	Description
{:ok, value}	value	Normalized success
{:error, reason}	{:error, reason}	Propagated as-is
raise (exception)	{:error, exception}	Caught and wrapped
throw value	{:error, {:throw, value}}	Uniformly captured
raw value	same	Passed through unchanged

This makes all pipelines exception-safe — any operator can raise or throw, but the stream will continue gracefully with well-defined {:error, reason} items.

---

💡 4. Integration with run/1 and run_with/3

Both stream finalizers (run/1, run_with/3) have been updated to:

• Handle {:error, reason} tuples and convert them to %GRPC.RPCError{}.
• Support both unary and streaming flows.
• Ensure that errors in one element do not interrupt subsequent elements in the stream.
• Allow dry_run mode to test flow execution without emitting gRPC responses.

---

✅ 5. Test Coverage

New and extended tests added in GRPC.StreamTest include:

• effect/2: ensures side effects are applied safely and do not affect data flow, even when exceptions occur inside the callback.
• map_error/2: verifies correct error mapping and recovery.
• Validation of error propagation across all operators (map, flat_map, filter, ask, etc.).
• Resilience against process failures (:process_not_alive, :timeout) in ask/3.
• Full pipeline integration tests combining multiple operators and error recovery logic.

---

🧠 Impact

This refactor:

• Unifies and simplifies error handling across the entire streaming subsystem.
• Makes pipelines more robust — no unexpected crashes due to raised errors.
• Improves observability: all errors now flow through %GRPC.RPCError{}.
• Adds new extensibility points for metrics, tracing, and recovery.

🔮 Next Steps

• Integrate tracing hooks to improve runtime observability.
• Update the docs.

[yordis]

A new utility function to safely perform side effects within a stream pipeline without modifying the values being processed.

Other ecosystem call it https://doc.rust-lang.org/std/result/enum.Result.html#method.inspect_err

Eitherway, fine by me

[sleipnir]

A new utility function to safely perform side effects within a stream pipeline without modifying the values being processed.

Other ecosystem call it https://doc.rust-lang.org/std/result/enum.Result.html#method.inspect_err

Eitherway, fine by me

In this case, side effects aren't necessarily linked to an error; side effects in this context are any operation that shouldn't affect the flow. Think of a call to an external system (e.g., sending an email) that shouldn't affect the processing of the main flow, and is therefore a side effect. Then inspect_err would be a misnomer in this context. But I added map_error that can serve this purpose.

[yordis]

🚀 💜

[polvalente]

Suggested change

	GRPC.Stream.unary(request, materializer: materializer)
	|> GRPC.Stream.map(fn %HelloReply{} = reply ->
	request
	|> GRPC.Stream.unary(materializer: materializer)
	|> GRPC.Stream.map(fn %HelloReply{} = reply ->

[polvalente]

Suggested change

	* map_error/2 captures and transforms that error into a structured `GRPC.RPCError` response.
	* `map_error/2` captures and transforms that error into a structured `GRPC.RPCError` response.

[polvalente]

Suggested change

	Applies a **side-effecting function** to each element of the stream **without altering** its values.
	Applies a side-effect function to each element of the stream without altering its values.

Let's remove all of these bold highlights

[polvalente]

This is a bit redundant with the typespec

[polvalente]

stray comment

[polvalente]

let's break this up a bit for readability

[polvalente]

can we extract this to a defp?

[polvalente]

Suggested change

	Stream.cycle(1..5)
	|> Stream.take(5)
	|> GRPC.Stream.from()
	1..5
	|> GRPC.Stream.from()

Doesn't this work? If not, we should make it work in a separate PR. Suggestion for if enumerables aren't allowed directly:

Suggested change

	Stream.cycle(1..5)
	|> Stream.take(5)
	|> GRPC.Stream.from()
	1..5
	|> Stream.take(5)
	|> GRPC.Stream.from()

[polvalente]

Suggested change

	spawn(fn ->
	spawn_link(fn ->

[polvalente]

Suggested change

	# do not send any response
	receive do
	_ -> :ok
	end
	Process.sleep(:infinity)

[polvalente]

Also instead of spawning we could have self() be the pid in ask and maybe even have an assert_receive