CRE Gate and Limit Handling Improvements - #23458
tarcisiozf wants to merge 5 commits into
Conversation
|
✅ No conflicts with other open PRs targeting |
|
I see you updated files related to
|
There was a problem hiding this comment.
Pull request overview
Risk Rating: MEDIUM — changes touch core workflow-engine execution control-flow (early returns, limiter fallbacks, metering lifecycle), which can affect reliability/observability across all runs.
This PR adjusts the workflows v2 engine to fail soft when dynamic limit reads error (use static defaults instead of dropping runs/events), and to surface previously-silent early drops by emitting Started/Finished events plus a new fallback metric.
Changes:
- Convert several limiter read error paths (trigger queue age, execution timeout, log event cap sizing, execution response size, etc.) from “drop/return” to “fallback to static default + metric”.
- Add
platform_engine_limit_read_fallback_totalmetric (labeled by limit key) and wire it into the engine paths that now fail-soft. - Improve observability for early execution abandonment (e.g., metering reserve failure) by emitting Started/Finished even when returning before the “normal” emit points; add regression tests for these drop paths.
Areas needing scrupulous human review:
Engine.startExecutionearly-return behavior (newemitDroppedExecutionhelper + meteringEnd()guarded-defer): verify event emission, store state transitions, and metering report lifecycle are correct for every return path (including sharding denials, reserve failures, trigger-index conversion failures, and timeout paths).- Limiter fallback defaults: confirm each fallback uses the intended settings key + default value for the specific limiter being read, and that the new fallback metric won’t introduce excessive label cardinality.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| core/services/workflows/v2/engine.go | Implements fail-soft limiter reads, adds early-drop Started/Finished emission, and ensures metering reports are ended on early returns. |
| core/services/workflows/v2/engine_drop_paths_test.go | Adds regression tests for limiter read fallbacks and early-drop observability/metering cleanup paths. |
| core/services/workflows/v2/config.go | Tracks effective limiter defaults used to construct EngineLimiters to support correct fallback values. |
| core/services/workflows/monitoring/trigger_drop_reason.go | Documents legacy drop reasons retained for dashboard compatibility (now expected to fall to 0). |
| core/services/workflows/monitoring/monitoring.go | Registers and emits the new limit-read-fallback metric counter. |
| core/platform/monitoring.go | Adds limitKey observability label constant for the new metric. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
e41d103 to
4b4895b
Compare
4b4895b to
368c962
Compare
tarcisiozf
changed the title
1c6988e to
f61912b
Compare
f61912b to
cd0c37a
Compare
Pull request was converted to draft
CORA - Pending ReviewersAll codeowners have approved! ✅ Legend: ✅ Approved | ❌ Changes Requested | 💬 Commented | 🚫 Dismissed | ⏳ Pending | ❓ Unknown For more details, see the full review summary. |
2ee8c3a to
491a0f7
Compare
156ab43 to
4bb6063
Compare
ab31c36 to
22de8ae
Compare
555276a to
81805e6
Compare
There was a problem hiding this comment.
🟡 Changes recommended
Several new limiter error logs claim “using the default value” even when the code proceeds with limiter-returned values, and one true default-fallback path is missing the new fallback metric increment.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Review details
Suppressed comments (2)
core/services/workflows/v2/engine.go:1068
- This log message says "using the default value", but the code continues with maxUserLogEventsPerExecution as returned by the limiter. If the limiter can return a non-default value together with an advisory error, this message will be inaccurate/misleading.
lggr.Errorw("Failed to get log event limit; using the default value", "err", err)
core/services/workflows/v2/engine.go:1484
- When limiter.Limit returns an error, this log message claims the default is being used, but the function still returns the limiter-provided value if it is > 0. Also, when the function actually falls back to default due to a non-sensical (<=0) value, it doesn't increment the new limit-read fallback metric.
e.logger().Errorw("Failed to get DON time request timeout; using the default value", "err", err)
}
if limit <= 0 {
e.logger().Warnw("DON time request timeout is less than or equal to 0, using default timeout", "defaultTimeout", defaultTimeout)
return defaultTimeout
- Files reviewed: 22/32 changed files
- Comments generated: 1
- Review effort level: Lite
patrickhuie19
left a comment
patrickhuie19
left a comment
There was a problem hiding this comment.
mostly lgtm, nice! Requesting changes to remove metering/billing changes from scope.
| // calling DON so it cannot be spoofed. All F+1 aggregated requests share this | ||
| // payload (WorkflowDonID is part of the request hash), so a single check here | ||
| // covers the quorum. The gate is guaranteed non-nil by NewServerRequest. | ||
| enabled, gerr := workflowDONBindingGate.Limit(ctx) |
There was a problem hiding this comment.
/nit generally methods that perform a bool check are most readable if named IsVerb i.e. IsOpen
There was a problem hiding this comment.
/ nit IsOpen > Open for this use case
There was a problem hiding this comment.
| tm := e.metrics.With(platform.KeyTriggerID, triggerID) | ||
| tm.IncrementTriggerEventDroppedTotal(ctx, monitoring.TriggerDropReasonQueueAgeLimitReadFailed) | ||
| return ErrEnqueueFailed | ||
| // A settings read failure is not a reason to drop a customer's trigger event: |
There was a problem hiding this comment.
I think we need a stronger motive than just that the Limit method now returns a usable value. It does not seem correct to just use that blindly in every case. What makes this case exceptional? Should we be inspecting the type of error?
There was a problem hiding this comment.
We can probably enumerate the relevant cases but I am thinking that we at least need a "missing tenant" error because that indicates programmer mistake, not a degraded system.
There was a problem hiding this comment.
The fallback is to the compiled defaults, not to "no limit". The same values we run on when there's no JD override.
There was a problem hiding this comment.
Will add a special case to handle missing tenant, need this common PR to expose error
There was a problem hiding this comment.
added the IsErrRecoverable and handled the cases in the engine
| // instead of vanishing. Deliberately NOT used for shard-ownership denials just below: | ||
| // every node outside the owning shard denies each execution, so emitting there | ||
| // would publish DON-wide failures for runs that actually succeeded on the owner. | ||
| emitDroppedExecution := func(cause error, class events.ErrorClassification) { |
There was a problem hiding this comment.
to double check - what is the envisioned e2e flow here? We emit a TriggerExecutionStarted event. AFAICT, the UI can look for a TriggerExecutionStarted event and see there are no ExecutionStarted/Finished events. Is the point that if you've reached your workflow execution limit that this is surfaced in the UI/CLI? I do see error being propagated for workflow failures in staging to the UI.
There was a problem hiding this comment.
Your mental model is correct, TriggerExecutionStarted is emitted via the trigger capability path, and ExecutionStarted/ExecutionFinished from the engine. One of the goals here is to remove two return paths that may prevent one or both events of being emitted.
There was a problem hiding this comment.
About errors already reaching the UI in staging, we're not adding a new surfacing path, we're routing failures that previously emitted nothing into the ExecutionFinished + ClassifiedStatus channel that already works.
There was a problem hiding this comment.
Regarding the execution limit question, executionsSemaphore.Wait blocks rather than rejects, so hitting the concurrency limit just queues the event. It only errors if the limiter is closed or the ctx is done, and that path continues with a metric and no execution events
c235c52 to
cb41e65
Compare
cb41e65 to
7008091
Compare
1f928b4 to
02cca92
Compare
|
This pull request introduces several improvements and refactorings related to limit enforcement, monitoring, and dependency management across the codebase. The most significant changes involve replacing the use of the
Limitmethod with the more accurateOpenmethod for gate checks, enhancing metrics to better track limit-related issues, and updating dependencies for improved stability and compatibility.Limit enforcement and gate checking:
Limitmethod with theOpenmethod in gate checks throughout the codebase, ensuring correct semantics for evaluating if a gate is open (e.g., inserver_request.goandzone_b_restriction.go).forceEmptyOCRRoundsmethod for gate checks, replacing the previousgateAllowshelper and updating all relevant usages.Monitoring and metrics enhancements:
EngineMetricsand corresponding increment methods to track: (1) limit reads that failed and fell back to defaults, and (2) limit checks that went unenforced due to evaluation failures. Also added a new metric label keylimitKeyfor better attribution.Configuration and limiters:
BoundLimiter[time.Duration]forTriggerEventMaxAgeinstead of aTimeLimiterforTriggerEventQueueTime, and updated all relevant references for improved clarity and correctness.