Refactor Stream gRPC handlers to remove exception handling#34
Conversation
…exceptions Changed: Simplified the BatchAppend method by removing the try-catch blocks for IOException, TaskCanceledException, InvalidOperationException, and OperationCanceledException, as they were previously ignored. This streamlines the code and improves readability. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
PR SummaryRefactors gRPC Streams to streamline control flow and modernize files.
Written by Cursor Bugbot for commit 9df52db. This will update automatically on new commits. Configure here. |
WalkthroughStructural refactoring of two gRPC streaming handlers: BatchAppend simplifies control flow by removing exception-swallowing try/catch blocks, relocating type declarations and static helpers; Read adds authorization checks before streaming results and reorganizes error handling with try/catch wrapping. Changes
Sequence DiagramssequenceDiagram
participant Client
participant Server as Server<br/>(Read Handler)
participant AuthProvider as Auth Provider
participant StreamStore as Stream Store
Client->>Server: Request stream read
Server->>Server: Extract user from<br/>HTTP context
Server->>AuthProvider: CheckAccessAsync(user)
alt Authorization denied
AuthProvider-->>Server: Access denied
Server-->>Client: Error response
else Authorization granted
AuthProvider-->>Server: OK
Server->>StreamStore: CreateEnumerator
loop For each event
StreamStore-->>Server: ReadResponse
Server->>Server: TryConvertReadResponse<br/>→ ReadResp
Server-->>Client: Send ReadResp
end
Server->>Server: Register context<br/>cancellation cleanup
Server-->>Client: Stream complete
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Poem
Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
src/EventStore.Core/Services/Transport/Grpc/Streams.Read.cs (1)
23-90: Well-structured authorization and streaming flow.The Read method properly:
- Validates
uuidOptionbefore use- Checks access via
_provider.CheckAccessAsyncbefore creating the enumerator- Uses proper disposal pattern with
await usingHowever, a minor concern with the
DisposeEnumeratorpattern:The
async voidmethod at line 73 could silently swallow exceptions during disposal. While this is a common pattern for cancellation-triggered cleanup, consider logging disposal failures.🔎 Suggested improvement
-async void DisposeEnumerator() => await enumerator.DisposeAsync(); +async void DisposeEnumerator() { + try { + await enumerator.DisposeAsync(); + } catch (Exception ex) { + // Log but don't propagate - this is cleanup during cancellation + Log.Debug(ex, "Exception during enumerator disposal"); + } +}src/EventStore.Core/Services/Transport/Grpc/Streams.BatchAppend.cs (2)
139-148: Authorization check before stream identifier validation.The authorization check at lines 139-148 occurs before validating that
StreamIdentifieris not null (lines 150-158). IfStreamIdentifieris null, the authorization check still runs withrequest.Options.StreamIdentifierpotentially being null.Consider reordering to validate
StreamIdentifierfirst:🔎 Suggested fix
if (request.Options != null) { var timeout = Min(GetRequestedTimeout(request.Options), _writeTimeout); + if (request.Options.StreamIdentifier == null) { + await writer.WriteAsync(new BatchAppendResp { + CorrelationId = request.CorrelationId, + StreamIdentifier = request.Options.StreamIdentifier, + Error = Status.BadRequest( + $"Required field {nameof(request.Options.StreamIdentifier)} not set.") + }, cancellationToken); + continue; + } + if (!await _authorizationProvider.CheckAccessAsync(user, WriteOperation.WithParameter( Plugins.Authorization.Operations.Streams.Parameters.StreamId( request.Options.StreamIdentifier)), cancellationToken)) { await writer.WriteAsync(new BatchAppendResp { CorrelationId = request.CorrelationId, StreamIdentifier = request.Options.StreamIdentifier, Error = Status.AccessDenied }, cancellationToken); continue; } - if (request.Options.StreamIdentifier == null) { - await writer.WriteAsync(new BatchAppendResp { - CorrelationId = request.CorrelationId, - StreamIdentifier = request.Options.StreamIdentifier, - Error = Status.BadRequest( - $"Required field {nameof(request.Options.StreamIdentifier)} not set.") - }, cancellationToken); - continue; - }
326-335: Fire-and-forget timeout task may cause issues.The
Task.Delay(...).ContinueWith(...)pattern at line 334 starts a fire-and-forget timeout. If theClientWriteRequestcompletes before the timeout, the continuation still runs and callsonTimeout(), which attempts to remove frompendingWritesand write a timeout response.The
TryRemovewill returnfalseif already removed, preventing the timeout response — this is correct. However, the timeout task keeps running unnecessarily.Consider using a
CancellationTokenSourcethat can be cancelled when the request completes to avoid unnecessary timer continuations:This is a minor optimization. The current implementation is functionally correct due to the
TryRemoveguard inonTimeout.
📜 Review details
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
src/EventStore.Core/Services/Transport/Grpc/Streams.BatchAppend.cssrc/EventStore.Core/Services/Transport/Grpc/Streams.Read.cs
🧰 Additional context used
🧬 Code graph analysis (1)
src/EventStore.Core/Services/Transport/Grpc/Streams.Read.cs (3)
src/EventStore.Core/Services/Transport/Grpc/ReadReq.cs (6)
Core(28-34)Core(36-42)ReadReq(7-47)Options(9-45)StreamRevision(12-17)StreamRevision(19-24)src/EventStore.Core/Services/Transport/Grpc/ReadResp.cs (2)
ReadResp(4-10)ReadEvent(6-8)src/EventStore.Core/Services/Transport/Grpc/RpcExceptions.cs (11)
RpcExceptions(10-248)Exception(11-11)Exception(19-20)Exception(167-176)Exception(178-186)Exception(188-196)Exception(198-207)Exception(209-217)Exception(219-223)Exception(225-229)Exception(231-235)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (4)
- GitHub Check: Cursor Bugbot
- GitHub Check: Build / noble
- GitHub Check: Build / bookworm-slim
- GitHub Check: Docker Compose Smoke Test
🔇 Additional comments (10)
src/EventStore.Core/Services/Transport/Grpc/Streams.Read.cs (6)
83-89: Verify exception re-throw behavior after duration tracking.The outer catch block at lines 86-89 sets the exception on duration tracking and re-throws. However, the inner catch at lines 83-85 for
ReadResponseExceptioncallsConvertReadResponseExceptionwhich always throws a newRpcException.This means
ReadResponseExceptioninstances will:
- Be caught at line 83
- Converted and thrown as
RpcExceptionat line 84- Caught again at line 86, tracked, and re-thrown
This is correct behavior for duration tracking, but confirm that tracking
RpcException(not the originalReadResponseException) is the intended behavior.
92-239: Large switch expression is well-organized but consider documenting unsupported combinations.The
CreateEnumeratormethod correctly routes to appropriate enumerator implementations based on the combination of stream options, count options, read direction, and filter options.The final catch-all at lines 236-238 throws
RpcExceptions.InvalidCombinationfor unsupported combinations. This is good defensive programming.
241-252: LGTM!The
ConvertToEventFiltermethod cleanly handles both event type and stream identifier filters with prefix and regex options.
254-289: LGTM!The
TryConvertReadResponsepattern correctly returnsfalseforSubscriptionFellBehind(which is not sent to clients) and throws for unknown response types.
291-316: LGTM!The
ConvertReadResponseExceptionmethod provides exhaustive mapping of internal exceptions to gRPC exceptions with a clear fallback for unknown types.
318-364: LGTM!The
ConvertToRecordedEventandConvertToReadEventmethods properly handle nullable positions and the UUID format option.src/EventStore.Core/Services/Transport/Grpc/Streams.BatchAppend.cs (4)
70-74: Channel configuration allows multiple writers but only one is used.The channel is configured with
SingleWriter = false, but theReceivemethod is the only writer (besides the callback inConvertMessage). The callback at line 201 usesTryWrite, which is safe.This configuration is correct given that the
CallbackEnvelopecallbacks can write from different threads.
94-109:async voidhandler swallows some exceptions but propagates others.The
HandleCompletionmethod:
- Catches
OperationCanceledExceptionand sets cancellation on TCS- Catches
IOExceptionand logs + sets exception- Re-throws other exceptions via
TrySetExceptionThis is appropriate for the continuation pattern used here. The exceptions are properly surfaced through the
TaskCompletionSource.
317-345:ClientWriteRequestrecord placement and design.The
ClientWriteRequestis now a private record at class scope (lines 317-345), which improves organization compared to being nested inside the worker.The record uses mutable internal state (
_eventslist,_sizecounter) which is appropriate for the accumulation pattern but note that records typically suggest immutability. The implementation is correct for the use case.
33-41: The code already handles these exceptions appropriately. TheBatchAppendWorker.Work()method catchesOperationCanceledException,IOException, and other exceptions inHandleCompletion()and propagates them via the task completion source, which then propagates to the gRPC framework. The gRPC framework automatically converts these exceptions to appropriate RPC Status codes (e.g.,CANCELLEDforOperationCanceledException,UNAVAILABLE/INTERNALforIOException). No additional upstream exception handling is needed—this is standard gRPC behavior and properly manages client experience.
yordis
changed the title
1 participant
Changed: Simplified the BatchAppend method by removing the try-catch blocks for IOException, TaskCanceledException, InvalidOperationException, and OperationCanceledException, as they were previously ignored. This streamlines the code and improves readability.