[DUCT-13790] support listing events #582
Conversation
WalkthroughAdds two new Task oneof variants (ListEventFeeds, ListEvents) with proto messages, generated validation for those messages, task routing updates, and two c1api handlers that call ConnectorClient and finish tasks (ListEvents supports paginated fetching). Changes
Sequence DiagramsequenceDiagram
participant Manager as c1ApiTaskManager
participant Handler as ListEventsHandler / ListEventFeedsHandler
participant Connector as ConnectorClient
participant Helpers as TaskHelpers
Manager->>Handler: Dispatch task (ListEventsType / ListEventFeedsType)
Handler->>Handler: Extract payload & Validate()
alt validation fails
Handler->>Helpers: FinishTask(error, non-retryable)
else validation succeeds
opt ListEvents pagination
loop while hasMore
Handler->>Connector: ListEvents(request{event_feed_id, start_at, page_size, cursor})
Connector-->>Handler: ListEventsResponse{events, cursor, has_more, annotations}
Handler->>Handler: Accumulate events, update cursor
end
end
Handler->>Helpers: FinishTask(response, annotations)
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
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: 2
🧹 Nitpick comments (2)
pb/c1/connectorapi/baton/v1/baton.pb.validate.go (1)
3586-3587: Avoid manual edits in generated validation files; add rules in the.protoinstead.
Since this file is generated, the comment will be overwritten on regeneration; ifevent_feed_idneeds constraints, add(validate.rules)toEventFeedTask.event_feed_idinproto/c1/connectorapi/baton/v1/baton.protoand re-generate.pkg/tasks/c1api/event_feed.go (1)
40-66: Add a safety bound for unbounded event aggregation (large feeds can OOM).
Ifevent_feed_idcan produce large histories, aggregating all events into a single[]*v2.Eventcan spike memory. Consider enforcing a max pages/events limit (and returning a non-retryable “response too large” error) or implementing chunked responses if the protocol allows.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (2)
pb/c1/connectorapi/baton/v1/baton.pb.gois excluded by!**/*.pb.gopb/c1/connectorapi/baton/v1/baton_protoopaque.pb.gois excluded by!**/*.pb.go
📒 Files selected for processing (4)
pb/c1/connectorapi/baton/v1/baton.pb.validate.go(1 hunks)pkg/tasks/c1api/event_feed.go(1 hunks)pkg/tasks/c1api/manager.go(1 hunks)proto/c1/connectorapi/baton/v1/baton.proto(1 hunks)
🧰 Additional context used
🧬 Code graph analysis (1)
pkg/tasks/c1api/manager.go (2)
pkg/types/tasks/tasks.go (1)
ListEventsType(96-96)pkg/tasks/c1api/event_feed.go (1)
NewEventFeedHandler(68-73)
⏰ 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). (2)
- GitHub Check: go-test (1.25.2, ubuntu-latest)
- GitHub Check: go-test (1.25.2, windows-latest)
🔇 Additional comments (2)
pkg/tasks/c1api/manager.go (1)
288-289: Task routing forListEventsTypelooks correct.pkg/tasks/c1api/event_feed.go (1)
68-73: Constructor looks fine.
pkg/tasks/c1api/event_feed.go
Outdated
| func (c *eventFeedHandler) HandleTask(ctx context.Context) error { | ||
| ctx, span := tracer.Start(ctx, "eventFeedHandler.HandleTask") | ||
| defer span.End() | ||
|
|
||
| l := ctxzap.Extract(ctx) | ||
| cc := c.helpers.ConnectorClient() | ||
|
|
||
| t := c.task.GetEventFeed() | ||
| if t == nil { | ||
| l.Error("get event feed task was nil", zap.Any("get_event_feed_task", t)) | ||
| return c.helpers.FinishTask(ctx, nil, nil, errors.Join(errors.New("malformed get event feed task"), ErrTaskNonRetryable)) | ||
| } | ||
|
|
||
| var ( | ||
| pageToken = "" | ||
| events []*v2.Event | ||
| ) | ||
|
|
||
| for { | ||
| feeds, err := cc.ListEvents(ctx, v2.ListEventsRequest_builder{ | ||
| EventFeedId: t.GetEventFeedId(), | ||
| StartAt: t.GetStartAt(), | ||
| Cursor: pageToken, | ||
| }.Build()) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| events = append(events, feeds.GetEvents()...) | ||
| pageToken = feeds.GetCursor() | ||
| if !feeds.GetHasMore() { | ||
| break | ||
| } | ||
| } | ||
|
|
||
| resp := v2.ListEventsResponse_builder{ | ||
| Events: events, | ||
| }.Build() | ||
| return c.helpers.FinishTask(ctx, resp, resp.GetAnnotations(), nil) | ||
| } |
There was a problem hiding this comment.
Critical: HandleTask must finish the task on ListEvents errors (don’t just return err).
Right now, if cc.ListEvents(...) fails, the handler exits without calling FinishTask, which can leave the task un-finalized despite the manager expecting handlers to finalize.
func (c *eventFeedHandler) HandleTask(ctx context.Context) error {
ctx, span := tracer.Start(ctx, "eventFeedHandler.HandleTask")
defer span.End()
@@
for {
feeds, err := cc.ListEvents(ctx, v2.ListEventsRequest_builder{
EventFeedId: t.GetEventFeedId(),
StartAt: t.GetStartAt(),
Cursor: pageToken,
}.Build())
if err != nil {
- return err
+ // Ensure the task is finalized on failure.
+ finishErr := c.helpers.FinishTask(ctx, nil, nil, err)
+ if finishErr != nil {
+ return errors.Join(err, finishErr)
+ }
+ // Task is finished; let processing continue.
+ return nil
}
events = append(events, feeds.GetEvents()...)
+ prevToken := pageToken
pageToken = feeds.GetCursor()
if !feeds.GetHasMore() {
break
}
+ if pageToken == "" || pageToken == prevToken {
+ err := errors.New("pagination cursor did not advance")
+ finishErr := c.helpers.FinishTask(ctx, nil, nil, err)
+ if finishErr != nil {
+ return errors.Join(err, finishErr)
+ }
+ return nil
+ }
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| func (c *eventFeedHandler) HandleTask(ctx context.Context) error { | |
| ctx, span := tracer.Start(ctx, "eventFeedHandler.HandleTask") | |
| defer span.End() | |
| l := ctxzap.Extract(ctx) | |
| cc := c.helpers.ConnectorClient() | |
| t := c.task.GetEventFeed() | |
| if t == nil { | |
| l.Error("get event feed task was nil", zap.Any("get_event_feed_task", t)) | |
| return c.helpers.FinishTask(ctx, nil, nil, errors.Join(errors.New("malformed get event feed task"), ErrTaskNonRetryable)) | |
| } | |
| var ( | |
| pageToken = "" | |
| events []*v2.Event | |
| ) | |
| for { | |
| feeds, err := cc.ListEvents(ctx, v2.ListEventsRequest_builder{ | |
| EventFeedId: t.GetEventFeedId(), | |
| StartAt: t.GetStartAt(), | |
| Cursor: pageToken, | |
| }.Build()) | |
| if err != nil { | |
| return err | |
| } | |
| events = append(events, feeds.GetEvents()...) | |
| pageToken = feeds.GetCursor() | |
| if !feeds.GetHasMore() { | |
| break | |
| } | |
| } | |
| resp := v2.ListEventsResponse_builder{ | |
| Events: events, | |
| }.Build() | |
| return c.helpers.FinishTask(ctx, resp, resp.GetAnnotations(), nil) | |
| } | |
| func (c *eventFeedHandler) HandleTask(ctx context.Context) error { | |
| ctx, span := tracer.Start(ctx, "eventFeedHandler.HandleTask") | |
| defer span.End() | |
| l := ctxzap.Extract(ctx) | |
| cc := c.helpers.ConnectorClient() | |
| t := c.task.GetEventFeed() | |
| if t == nil { | |
| l.Error("get event feed task was nil", zap.Any("get_event_feed_task", t)) | |
| return c.helpers.FinishTask(ctx, nil, nil, errors.Join(errors.New("malformed get event feed task"), ErrTaskNonRetryable)) | |
| } | |
| var ( | |
| pageToken = "" | |
| events []*v2.Event | |
| ) | |
| for { | |
| feeds, err := cc.ListEvents(ctx, v2.ListEventsRequest_builder{ | |
| EventFeedId: t.GetEventFeedId(), | |
| StartAt: t.GetStartAt(), | |
| Cursor: pageToken, | |
| }.Build()) | |
| if err != nil { | |
| // Ensure the task is finalized on failure. | |
| finishErr := c.helpers.FinishTask(ctx, nil, nil, err) | |
| if finishErr != nil { | |
| return errors.Join(err, finishErr) | |
| } | |
| // Task is finished; let processing continue. | |
| return nil | |
| } | |
| events = append(events, feeds.GetEvents()...) | |
| prevToken := pageToken | |
| pageToken = feeds.GetCursor() | |
| if !feeds.GetHasMore() { | |
| break | |
| } | |
| if pageToken == "" || pageToken == prevToken { | |
| err := errors.New("pagination cursor did not advance") | |
| finishErr := c.helpers.FinishTask(ctx, nil, nil, err) | |
| if finishErr != nil { | |
| return errors.Join(err, finishErr) | |
| } | |
| return nil | |
| } | |
| } | |
| resp := v2.ListEventsResponse_builder{ | |
| Events: events, | |
| }.Build() | |
| return c.helpers.FinishTask(ctx, resp, resp.GetAnnotations(), nil) | |
| } |
🤖 Prompt for AI Agents
In pkg/tasks/c1api/event_feed.go around lines 27 to 66, the loop returns the raw
error when cc.ListEvents(...) fails which skips calling FinishTask; change the
error branch to call c.helpers.FinishTask to finalize the task before returning.
Replace the current "return err" with code that logs the ListEvents error, calls
c.helpers.FinishTask(ctx, nil, nil, errors.Join(err, ErrTaskNonRetryable)) (or
otherwise wrap the list error consistently with other failures), and then
returns the FinishTask error (or the original error if FinishTask succeeded) so
the task is always finalized on ListEvents failures.
| message EventFeedTask { | ||
| repeated google.protobuf.Any annotations = 1; | ||
| google.protobuf.Timestamp start_at = 2; | ||
| string event_feed_id = 3; | ||
| } |
There was a problem hiding this comment.
Clarify/validate event_feed_id (required vs optional) to avoid empty-ID semantics.
If event_feed_id is required for correct ListEvents behavior, add a validation rule (e.g., min_len: 1) or make it explicitly required; otherwise, document what an empty string means for connectors/backends.
🤖 Prompt for AI Agents
proto/c1/connectorapi/baton/v1/baton.proto lines 54-58: the field event_feed_id
is currently an unconstrained string which can be empty and cause ambiguous
ListEvents semantics; either make it explicitly required/mandatory or add a
validation rule such as (validate.rules).string.min_len = 1 to enforce non-empty
IDs, and update the proto comment/docs to state the semantics of an empty string
if you intentionally allow it; after changing the proto, regenerate stubs and
update any ListEvents handling to assume non-empty IDs (or explicitly handle the
empty-ID case) and add tests that cover validation and API behavior.
Bencheng21
force-pushed
the
DUCT13790-listing_events
branch
2 times, most recently
from
8c53267 to
7465e00
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (1)
proto/c1/connectorapi/baton/v1/baton.proto (1)
60-65: Clarify and/or validateevent_feed_id/page_sizesemantics inListEventsTask.
ListEventsTaskcurrently leavesevent_feed_idandpage_sizeunconstrained. That’s OK if:
- empty
event_feed_idis a valid “all feeds” sentinel; andpage_size == 0or very large values are intentionally delegated to the backend.If instead
event_feed_idmust be non‑empty orpage_sizeshould be within a bounded range, please add PGV rules on these fields (e.g.min_len: 1, or min/max forpage_size) and regenerate validation, and document the intended empty/zero semantics. This was already raised in an earlier review and still appears open.
📜 Review details
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (2)
pb/c1/connectorapi/baton/v1/baton.pb.gois excluded by!**/*.pb.gopb/c1/connectorapi/baton/v1/baton_protoopaque.pb.gois excluded by!**/*.pb.go
📒 Files selected for processing (5)
pb/c1/connectorapi/baton/v1/baton.pb.validate.go(2 hunks)pkg/tasks/c1api/list_event_feeds.go(1 hunks)pkg/tasks/c1api/list_events.go(1 hunks)pkg/tasks/c1api/manager.go(1 hunks)proto/c1/connectorapi/baton/v1/baton.proto(2 hunks)
🧰 Additional context used
🧬 Code graph analysis (3)
pkg/tasks/c1api/list_event_feeds.go (4)
pkg/types/types.go (1)
ConnectorClient(30-46)pkg/annotations/annotations.go (1)
Annotations(12-12)pb/c1/connectorapi/baton/v1/baton.pb.go (2)
Task(82-115)Task(128-128)pkg/tasks/c1api/manager.go (1)
ErrTaskNonRetryable(43-43)
pkg/tasks/c1api/list_events.go (5)
pkg/types/types.go (1)
ConnectorClient(30-46)pkg/annotations/annotations.go (1)
Annotations(12-12)pb/c1/connectorapi/baton/v1/baton_protoopaque.pb.go (2)
Task(82-90)Task(103-103)pb/c1/connectorapi/baton/v1/baton.pb.go (2)
Task(82-115)Task(128-128)pkg/tasks/c1api/manager.go (1)
ErrTaskNonRetryable(43-43)
pkg/tasks/c1api/manager.go (3)
pkg/types/tasks/tasks.go (2)
ListEventFeedsType(97-97)ListEventsType(96-96)pkg/tasks/c1api/list_event_feeds.go (1)
NewListEventFeedsHandler(51-56)pkg/tasks/c1api/list_events.go (1)
NewListEventsHandler(69-74)
⏰ 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). (1)
- GitHub Check: go-test (1.25.2, windows-latest)
🔇 Additional comments (3)
proto/c1/connectorapi/baton/v1/baton.proto (1)
55-69: Event feed & listing task modeling looks consistent and backwards‑compatible.The new
EventFeedTask,ListEventsTask, andListEventFeedsTaskmessages and their addition to theTask.task_typeoneof with tags121and122follow the existing pattern and avoid field‑number conflicts. No issues from a wire‑compatibility perspective.Also applies to: 189-190
pkg/tasks/c1api/manager.go (1)
289-292: Handler wiring for new event tasks matches existing pattern.Dispatching
ListEventFeedsTypetoNewListEventFeedsHandlerandListEventsTypetoNewListEventsHandlerusing the existingtaskHelpersstruct is consistent with other task types; no issues here.pb/c1/connectorapi/baton/v1/baton.pb.validate.go (1)
927-1007: Generated validation for ListEvents/ListEventFeeds tasks matches proto patterns.The new
Taskswitch branches forTask_ListEventFeedsandTask_ListEvents, plus theTask_ListEventsTask/Task_ListEventFeedsTaskvalidation scaffolding, are consistent with the existing PGV‑generated patterns for other task messages. Any future tightening ofevent_feed_idorpage_sizeconstraints should be done in the.protoand this file regenerated, rather than edited manually.Also applies to: 3782-4085
| func (c *listEventFeedsHandler) HandleTask(ctx context.Context) error { | ||
| ctx, span := tracer.Start(ctx, "listEventFeedsHandler.HandleTask") | ||
| defer span.End() | ||
|
|
||
| l := ctxzap.Extract(ctx) | ||
| cc := c.helpers.ConnectorClient() | ||
|
|
||
| t := c.task.GetEventFeed() | ||
| if t == nil { | ||
| l.Error("get list event feeds task was nil", zap.Any("get_list_event_feeds_task", t)) | ||
| return c.helpers.FinishTask(ctx, nil, nil, errors.Join(errors.New("malformed get list event feeds task"), ErrTaskNonRetryable)) | ||
| } | ||
|
|
||
| feeds, err := cc.ListEventFeeds(ctx, &v2.ListEventFeedsRequest{}) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| resp := v2.ListEventFeedsResponse_builder{ | ||
| List: feeds.GetList(), | ||
| }.Build() | ||
| return c.helpers.FinishTask(ctx, resp, resp.GetAnnotations(), nil) | ||
| } |
There was a problem hiding this comment.
Fix incorrect getter (GetEventFeed vs GetListEventFeeds) and finish tasks on RPC error.
Two issues here:
-
Wrong task accessor:
For
ListEventFeedsTypetasks, the proto defines aListEventFeedsTaskbranch, so the generated accessor onv1.Taskwill beGetListEventFeeds(). UsingGetEventFeed()will returnnilfor those tasks, causing all validListEventFeedstasks to be marked as “malformed” and finished non‑retryably.Suggested change:
- t := c.task.GetEventFeed()
- t := c.task.GetListEventFeeds()
if t == nil {
-
l.Error("get list event feeds task was nil", zap.Any("get_list_event_feeds_task", t)) -
return c.helpers.FinishTask(ctx, nil, nil, errors.Join(errors.New("malformed get list event feeds task"), ErrTaskNonRetryable))
-
l.Error("list event feeds task was nil", zap.Any("list_event_feeds_task", t)) -
}
return c.helpers.FinishTask(ctx, nil, nil, errors.Join(errors.New("malformed list event feeds task"), ErrTaskNonRetryable))
-
ListEventFeeds RPC error path doesn’t call
FinishTask:As with
list_events.go,cc.ListEventFeedserrors currently justreturn err, which likely leaves the task unfinished. Consider:
- feeds, err := cc.ListEventFeeds(ctx, &v2.ListEventFeedsRequest{})
- if err != nil {
-
return err - }
- feeds, err := cc.ListEventFeeds(ctx, &v2.ListEventFeedsRequest{})
- if err != nil {
-
return c.helpers.FinishTask(ctx, nil, nil, err) - }
<!-- suggestion_start -->
<details>
<summary>📝 Committable suggestion</summary>
> ‼️ **IMPORTANT**
> Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
```suggestion
func (c *listEventFeedsHandler) HandleTask(ctx context.Context) error {
ctx, span := tracer.Start(ctx, "listEventFeedsHandler.HandleTask")
defer span.End()
l := ctxzap.Extract(ctx)
cc := c.helpers.ConnectorClient()
t := c.task.GetListEventFeeds()
if t == nil {
l.Error("list event feeds task was nil", zap.Any("list_event_feeds_task", t))
return c.helpers.FinishTask(ctx, nil, nil, errors.Join(errors.New("malformed list event feeds task"), ErrTaskNonRetryable))
}
feeds, err := cc.ListEventFeeds(ctx, &v2.ListEventFeedsRequest{})
if err != nil {
return c.helpers.FinishTask(ctx, nil, nil, err)
}
resp := v2.ListEventFeedsResponse_builder{
List: feeds.GetList(),
}.Build()
return c.helpers.FinishTask(ctx, resp, resp.GetAnnotations(), nil)
}
🤖 Prompt for AI Agents
In pkg/tasks/c1api/list_event_feeds.go around lines 27-49, replace the incorrect
task accessor GetEventFeed() with GetListEventFeeds() so valid ListEventFeeds
tasks are recognized, and change the RPC error path for cc.ListEventFeeds to
finish the task via c.helpers.FinishTask(ctx, nil, nil, err) (and return that)
instead of simply returning the error to ensure the task is completed when the
RPC fails.
| func (c *listEventsHandler) HandleTask(ctx context.Context) error { | ||
| ctx, span := tracer.Start(ctx, "listEventHandler.HandleTask") | ||
| defer span.End() | ||
|
|
||
| l := ctxzap.Extract(ctx) | ||
| cc := c.helpers.ConnectorClient() | ||
|
|
||
| t := c.task.GetListEvents() | ||
| if t == nil { | ||
| l.Error("get list event task was nil", zap.Any("get_list_event_task", t)) | ||
| return c.helpers.FinishTask(ctx, nil, nil, errors.Join(errors.New("malformed get list event task"), ErrTaskNonRetryable)) | ||
| } | ||
|
|
||
| var ( | ||
| pageToken = "" | ||
| events []*v2.Event | ||
| ) | ||
|
|
||
| for { | ||
| feeds, err := cc.ListEvents(ctx, v2.ListEventsRequest_builder{ | ||
| EventFeedId: t.GetEventFeedId(), | ||
| StartAt: t.GetStartAt(), | ||
| Cursor: pageToken, | ||
| PageSize: t.GetPageSize(), | ||
| }.Build()) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| events = append(events, feeds.GetEvents()...) | ||
| pageToken = feeds.GetCursor() | ||
| if !feeds.GetHasMore() { | ||
| break | ||
| } | ||
| } | ||
|
|
||
| resp := v2.ListEventsResponse_builder{ | ||
| Events: events, | ||
| }.Build() | ||
| return c.helpers.FinishTask(ctx, resp, resp.GetAnnotations(), nil) | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, find similar handler files in the same directory
find pkg/tasks/c1api -name "*handler*" -o -name "*list_*.go" | head -20Repository: ConductorOne/baton-sdk
Length of output: 171
🏁 Script executed:
# Check the actual file to confirm the exact code and span name
cat -n pkg/tasks/c1api/list_events.go | head -70Repository: ConductorOne/baton-sdk
Length of output: 2357
🏁 Script executed:
# Search for other handlers in the codebase to understand FinishTask patterns
rg -A 5 "func.*HandleTask" pkg/tasks/ --type go | head -100Repository: ConductorOne/baton-sdk
Length of output: 6230
🏁 Script executed:
# Look for FinishTask calls to see error handling patterns
rg "FinishTask.*err" pkg/tasks/ --type go -B 2 -A 2 | head -80Repository: ConductorOne/baton-sdk
Length of output: 5271
🏁 Script executed:
# Check list_ticket_schemas error handling for comparison (also does pagination)
cat -n pkg/tasks/c1api/list_ticket_schemas.go | sed -n '30,80p'Repository: ConductorOne/baton-sdk
Length of output: 1934
🏁 Script executed:
# Check list_event_feeds error handling
cat -n pkg/tasks/c1api/list_event_feeds.go | sed -n '27,70p'Repository: ConductorOne/baton-sdk
Length of output: 1188
🏁 Script executed:
# Check grant handler error handling for comparison
rg -A 15 "func.*grantTaskHandler.*HandleTask" pkg/tasks/c1api/grant.go | head -40Repository: ConductorOne/baton-sdk
Length of output: 811
🏁 Script executed:
# Check if there's a pattern: are all list/query handlers using bare return on error?
rg "func.*Handler.*HandleTask" pkg/tasks/c1api/ --type go | grep -i "list\|get\|bulk"Repository: ConductorOne/baton-sdk
Length of output: 929
🏁 Script executed:
# Look at one more list handler to confirm pattern
cat -n pkg/tasks/c1api/get_ticket.go | sed -n '27,65p'Repository: ConductorOne/baton-sdk
Length of output: 1558
🏁 Script executed:
# Check list_ticket_schemas full error handling to see if it calls FinishTask on RPC error
cat -n pkg/tasks/c1api/list_ticket_schemas.go | sed -n '50,82p'Repository: ConductorOne/baton-sdk
Length of output: 1151
Ensure RPC errors in the pagination loop result in FinishTask being called.
Error handling is inconsistent across handlers. When cc.ListEvents fails (line 52–53), you return the error directly instead of calling FinishTask. This leaves the task potentially unfinished. Other handlers like getTicketTaskHandler correctly route RPC errors through FinishTask:
if err != nil {
- return err
+ return c.helpers.FinishTask(ctx, nil, nil, err)
}Minor issues:
- Span name mismatch: The span name
"listEventHandler.HandleTask"should be"listEventsHandler.HandleTask"to match the type name. - Annotations from paginated responses: You currently drop annotations from intermediate pages and only pass the final response's annotations. Consider accumulating or merging them if backends rely on this.
🤖 Prompt for AI Agents
In pkg/tasks/c1api/list_events.go around lines 27 to 67, the handler returns RPC
errors directly inside the pagination loop which can leave the task unfinished;
change the error path so any error from cc.ListEvents calls FinishTask with nil
response and the error (and appropriate non-retryable flag if needed) before
returning. Also correct the span name from "listEventHandler.HandleTask" to
"listEventsHandler.HandleTask" to match the type, and while iterating
accumulate/merge annotations from each paginated feeds response into a single
annotations set (rather than only using the final page's annotations) so
backend-provided annotations are preserved when calling FinishTask with the
aggregated response.
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
pkg/tasks/c1api/list_event_feeds.go (1)
40-43: Finish task on RPC error to prevent incomplete task state.When
cc.ListEventFeedsfails, returning the error directly likely leaves the task unfinished in the task management system. The task should be explicitly completed viaFinishTaskso the system knows it's done.Apply this diff:
feeds, err := cc.ListEventFeeds(ctx, &v2.ListEventFeedsRequest{}) if err != nil { - return err + return c.helpers.FinishTask(ctx, nil, nil, err) }
🧹 Nitpick comments (1)
pkg/tasks/c1api/list_event_feeds.go (1)
36-37: Simplify log messages.The log message and field name are verbose. "get list event feeds task" is awkward phrasing.
Apply this diff:
- l.Error("get list event feeds task was nil", zap.Any("get_list_event_feeds_task", t)) - return c.helpers.FinishTask(ctx, nil, nil, errors.Join(errors.New("malformed get list event feeds task"), ErrTaskNonRetryable)) + l.Error("list event feeds task was nil", zap.Any("list_event_feeds_task", t)) + return c.helpers.FinishTask(ctx, nil, nil, errors.Join(errors.New("malformed list event feeds task"), ErrTaskNonRetryable))
📜 Review details
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
pkg/tasks/c1api/list_event_feeds.go(1 hunks)pkg/tasks/tasks.go(1 hunks)
🧰 Additional context used
🧬 Code graph analysis (1)
pkg/tasks/tasks.go (2)
pb/c1/connectorapi/baton/v1/baton.pb.go (2)
Task_ListEventFeeds_case(928-928)Task_ListEvents_case(929-929)pkg/types/tasks/tasks.go (2)
ListEventFeedsType(97-97)ListEventsType(96-96)
⏰ 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). (3)
- GitHub Check: go-lint
- GitHub Check: go-test (1.25.2, ubuntu-latest)
- GitHub Check: go-test (1.25.2, windows-latest)
pkg/tasks/c1api/list_events.go
Outdated
| EventFeedId: t.GetEventFeedId(), | ||
| StartAt: t.GetStartAt(), | ||
| Cursor: pageToken, | ||
| PageSize: t.GetPageSize(), |
There was a problem hiding this comment.
I thought the page size was to limit the number of events to fetch, not an implementation detail of how many events to fetch at a time. I'm pretty sure we want a way to limit the number of events we get back. Also we should think about how we can avoid hitting GRPC message size limits.
I think the simplest thing would be to mirror ListEventsRequest/Response. Allow requesting events with a timestamp or an opaque cursor, and have the response contain a cursor and a has_more bool.
Bencheng21
force-pushed
the
DUCT13790-listing_events
branch
from
c810cdb to
d19529a
Compare
| repeated google.protobuf.Any annotations = 1; | ||
| google.protobuf.Timestamp start_at = 2; | ||
| string event_feed_id = 3; | ||
| uint32 page_size = 4; |
There was a problem hiding this comment.
we don't need a cursor like
?There was a problem hiding this comment.
good catch. thanks
Bencheng21
force-pushed
the
DUCT13790-listing_events
branch
from
d19529a to
5800b4a
Compare
implemented listing events handler in baton-sdk.
Summary by CodeRabbit
New Features
Chores
✏️ Tip: You can customize this high-level summary in your review settings.