[CRE-6176] move registration handles ack to dispatcher - #23640
Conversation
agparadiso
changed the title
|
I see you updated files related to
|
|
✅ No conflicts with other open PRs targeting |
d73a447 to
887ba51
Compare
… for all workflows
887ba51 to
5b2a083
Compare
|
bolekk
left a comment
bolekk
left a comment
There was a problem hiding this comment.
Can we avoid re-using the name "Dispatcher" please? It will cause confusion with the don2don Dispatcher
| d.mu.Unlock() | ||
|
|
||
| // Unregister with the capability registry outside the lock. | ||
| ctx := context.Background() |
There was a problem hiding this comment.
Can we pass a context derived from the d.eng.NewCtx()?f
| ctx := context.Background() | |
| ctx, cancel := d.eng.NewCtx() | |
| defer cancel() |
| d.startReader(ctx, wid, idx, subs[idx], triggerEventCh) | ||
| } | ||
| d.lggr.Infow("All triggers registered successfully", "numTriggers", len(subs), "triggerIDs", triggerCapIDs) | ||
| d.metrics.IncrementWorkflowRegisteredCounter(ctx) | ||
| return triggerCapIDs, nil | ||
| } | ||
|
|
||
| // startReader runs one reader goroutine per subscription. It resolves the | ||
| // engine from the registry at delivery time and never holds an engine | ||
| // reference; if the engine is gone the reader exits. | ||
| func (d *triggerDispatcher) startReader(ctx context.Context, wid types.WorkflowID, idx int, sub *sdkpb.TriggerSubscription, triggerEventCh <-chan capabilities.TriggerResponse) { | ||
| d.eng.GoCtx(context.WithoutCancel(ctx), func(ctx context.Context) { |
There was a problem hiding this comment.
WDYT about moving d.eng.GoCtx to the caller, so that this method can be defined more simply and with less indentation?
| // start is a no-op: the dispatcher has no background work of its own — reader | ||
| // goroutines are started per subscription by RegisterTriggers and tracked by | ||
| // the embedded services.Engine. The method exists to satisfy the | ||
| // services.Service Start hook contract. | ||
| func (d *triggerDispatcher) start(context.Context) error { return nil } |
There was a problem hiding this comment.
You can omit this when using services.Engine - it nil checks.
| } | ||
|
|
||
| // register to all triggers concurrently | ||
| regCtx, regCancel, err := d.regTime.WithTimeout(ctx) |
There was a problem hiding this comment.
Sometimes when there are multiple contexts in scope, there is a natural point to split out a helper func so that each independent scope can have one clean context named idiomatically as ctx. What if we introduced a method for this block of logic?
| Payload: handle.payload, | ||
| Method: handle.method, | ||
| }); err != nil { | ||
| d.lggr.Errorw("Failed to unregister trigger", "registrationId", registrationID, "err", err) |
There was a problem hiding this comment.
Are these errors significant? I'm worried it could be overlooked that we only log them. Easy change for readability could be to call it tryUnregisterAll or similar. Or another option could be to collect a batch of errors to return to the caller (maybe more testable this way).
|
+1 to Bolek's comment -- let's find another name :) Router maybe? |
| // ReleaseHandles drops the retained handles for a workflow, after which | ||
| // its in-flight executions can no longer acknowledge. Called by the syncer | ||
| // once the engine has drained and closed. | ||
| ReleaseHandles(workflowID string) error |
There was a problem hiding this comment.
The concept of a handle didn't exist before -- what are we doing under the hood? Releasing the subscription? disabling routing?
There was a problem hiding this comment.
Thinking about this more: why do we need ReleaseHandles? What is it doing that UnregisterTriggers couldn't do for us?
| // starts one reader goroutine per subscription, and returns the registered | ||
| // trigger capability IDs. On any registration failure it rolls back the | ||
| // successful registrations. | ||
| func (d *triggerDispatcher) RegisterTriggers(ctx context.Context, cre contexts.CRE, params RegistrationParams, subs []*sdkpb.TriggerSubscription) ([]string, error) { |
There was a problem hiding this comment.
Nit: I wouldn't pass cre in expilictly, pass it in via the ctx -- since the engine owns the root context here we should be able to ensure that all calls to it have cre context
|
|
||
| // ReleaseHandles drops the retained handles for a workflow, after which its | ||
| // in-flight executions can no longer acknowledge. Safe to call multiple times. | ||
| func (d *triggerDispatcher) ReleaseHandles(workflowID string) error { |
There was a problem hiding this comment.
Can UnregisterTriggers call ReleaseHandles for us?
| // and calling AckEvent on the trigger capability. Engines call this after | ||
| // execution starts, on duplicate executions, and on shard-ownership denials — | ||
| // the point at which the event is fully handled and must not be redelivered. | ||
| func (d *triggerDispatcher) Ack(ctx context.Context, triggerCapID, triggerRegistrationID, eventID string) error { |
There was a problem hiding this comment.
If this is called be the engine, it will know what the wid is, so you can just make that part of the interface rather than having to maintain a registration -> wid index
|
Code location comment: why is this code located in the syncer? Is that the best place for it? |
|
|
||
| // triggerHandle is a registered trigger capability plus the registration | ||
| // payload/method needed to unregister it. | ||
| type triggerHandle struct { |
There was a problem hiding this comment.
I would call this triggerRegistration and give it an Unregister method that adds in your payload and method and anything else you need
There was a problem hiding this comment.
Same thing with AckEvent etc
|
|
||
| // ReportWorkflowLimitPerOwner records a per-owner workflow count limit rejection. | ||
| func (d *triggerDispatcher) ReportWorkflowLimitPerOwner(ctx context.Context) { | ||
| d.metrics.IncrementWorkflowLimitPerOwnerCounter(ctx) |
There was a problem hiding this comment.
I don't understand why these metrics are exposed here -- who calls it? This feels like a separate responsibility not related to triggers
| if err2 != nil { | ||
| return nil, fmt.Errorf("invalid chain selector for ID %s: %w", sub.Id, err2) | ||
| } | ||
| _ = chainSelector // chain access check moves with the limiter split (CRE-6177) |
There was a problem hiding this comment.
What is line 188->194 doing? it looks like we're not using this code at all; we should remove it if we're not using it because we may change our decision later
| for registrationID, handle := range wt.handles { | ||
| if unregErr := handle.UnregisterTrigger(ctx, capabilities.TriggerRegistrationRequest{ | ||
| TriggerID: registrationID, | ||
| Metadata: capabilities.RequestMetadata{ |
There was a problem hiding this comment.
The metadata here is different to what we send with RegisterTrigger; they should be the same though, to keep the calls symmetric. Is this what the current implementation does?
4 participants
Description
This pr is part 1 of CRE-6176
Its basically creating the dispatcher component that holds registration, handles and ACK. I've split the ticket in two tickets to make it easier to review. The code added here is not called anywhere yet, the wiring and clean up of the methods moved from the engine will be tackled on the next pr.
Requires
Supports