diff --git a/core/capabilities/remote/executable/request/server_request.go b/core/capabilities/remote/executable/request/server_request.go
index e64aa77c425..89d256a5a93 100644
--- a/core/capabilities/remote/executable/request/server_request.go
+++ b/core/capabilities/remote/executable/request/server_request.go
@@ -22,6 +22,7 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/remote"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types"
 	p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types"
+	"github.com/smartcontractkit/chainlink/v2/core/utils/crelimits"
 )
 
 type srMetrics struct {
@@ -396,7 +397,7 @@ func executeCapabilityRequest(ctx context.Context, lggr logger.Logger, capabilit
 	// 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)
+	enabled, gerr := crelimits.GateOpen(ctx, workflowDONBindingGate)
 	if gerr != nil {
 		lggr.Errorw("failed to evaluate workflow DON binding gate", "err", gerr)
 		return nil, errors.New("failed to evaluate workflow DON binding gate")
diff --git a/core/capabilities/vault/jwt_based_auth.go b/core/capabilities/vault/jwt_based_auth.go
index f1aa497fa78..734badb4d1d 100644
--- a/core/capabilities/vault/jwt_based_auth.go
+++ b/core/capabilities/vault/jwt_based_auth.go
@@ -23,6 +23,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings"
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
+	"github.com/smartcontractkit/chainlink/v2/core/utils/crelimits"
 )
 
 var (
@@ -220,7 +221,7 @@ func (v *jwtBasedAuth) close() error {
 
 // AuthorizeRequest verifies JWTBasedAuth state and token claims, and returns a common AuthResult.
 func (v *jwtBasedAuth) AuthorizeRequest(ctx context.Context, req jsonrpc.Request[json.RawMessage]) (*AuthResult, error) {
-	isEnabled, err := v.authEnabledGate.Limit(ctx)
+	isEnabled, err := crelimits.GateOpen(ctx, v.authEnabledGate)
 	if err != nil {
 		v.lggr.Errorw("failed to resolve JWTBasedAuth gate", "method", req.Method, "requestID", req.ID, "error", err)
 		return nil, fmt.Errorf("failed to resolve JWTBasedAuth gate: %w", err)
diff --git a/core/capabilities/vault/zone_b_restriction.go b/core/capabilities/vault/zone_b_restriction.go
index 2faefb919e2..6fb0252a6cb 100644
--- a/core/capabilities/vault/zone_b_restriction.go
+++ b/core/capabilities/vault/zone_b_restriction.go
@@ -12,6 +12,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings"
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
 	"github.com/smartcontractkit/chainlink-common/pkg/types/core"
+	"github.com/smartcontractkit/chainlink/v2/core/utils/crelimits"
 )
 
 // zoneBFamily is the DON family (in the capabilities registry) identifying the
@@ -69,7 +70,7 @@ func newZoneBRestrictor(lggr logger.Logger, limitsFactory limits.Factory, capabi
 // resolves to a zone-b DON. The owner is read from ctx, which must already carry
 // the (normalized) CRE owner via RequestMetadata.ContextWithCRE.
 func (z *zoneBRestrictor) enforce(ctx context.Context, workflowDonID uint32) error {
-	enabled, err := z.restrictEnabled.Limit(ctx)
+	enabled, err := crelimits.GateOpen(ctx, z.restrictEnabled)
 	if err != nil {
 		return fmt.Errorf("could not evaluate zone-b vault read restriction gate: %w", err)
 	}
diff --git a/core/platform/monitoring.go b/core/platform/monitoring.go
index 10324f87843..a37ff4909aa 100644
--- a/core/platform/monitoring.go
+++ b/core/platform/monitoring.go
@@ -10,6 +10,7 @@ const (
 	KeyCapabilityID        = "capabilityID"
 	KeyTriggerID           = "triggerID"
 	KeyTriggerDropReason   = "dropReason"
+	KeyLimitKey            = "limitKey"
 	KeyWorkflowID          = "workflowID"
 	KeyWorkflowExecutionID = "workflowExecutionID"
 	KeyWorkflowName        = "workflowName"
diff --git a/core/services/gateway/handlers/vault/aggregator.go b/core/services/gateway/handlers/vault/aggregator.go
index f12106e4f1d..7ca11f0b4aa 100644
--- a/core/services/gateway/handlers/vault/aggregator.go
+++ b/core/services/gateway/handlers/vault/aggregator.go
@@ -23,6 +23,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/vault/vaulttypes"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/vault/vaultutils"
+	"github.com/smartcontractkit/chainlink/v2/core/utils/crelimits"
 )
 
 var errSignedPayloadRequestIDMismatch = errors.New("signed payload request id mismatch")
@@ -55,12 +56,7 @@ func methodSupportsSignedOCRValidation(method string) bool {
 }
 
 func (a *baseAggregator) signedResponseRequestIDEnabled(ctx context.Context, l logger.Logger) bool {
-	allowed, err := a.signedResponseRequestIDGate.Limit(ctx)
-	if err != nil {
-		l.Errorw("unexpected error evaluating CRE gate", "gate", "VaultSignedResponseRequestIDEnabled", "error", err)
-		return false
-	}
-	return allowed
+	return crelimits.GateAllows(ctx, l, a.signedResponseRequestIDGate, "VaultSignedResponseRequestIDEnabled")
 }
 
 func (a *baseAggregator) Aggregate(ctx context.Context, l logger.Logger, requestID string, resps map[string]jsonrpc.Response[json.RawMessage], currResp *jsonrpc.Response[json.RawMessage]) (*jsonrpc.Response[json.RawMessage], error) {
diff --git a/core/services/ocr2/plugins/vault/plugin_utils.go b/core/services/ocr2/plugins/vault/plugin_utils.go
index 3b83828ecfb..4978036e9b1 100644
--- a/core/services/ocr2/plugins/vault/plugin_utils.go
+++ b/core/services/ocr2/plugins/vault/plugin_utils.go
@@ -15,20 +15,13 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/vault/vaulttypes"
 	"github.com/smartcontractkit/chainlink/v2/core/logger"
+	"github.com/smartcontractkit/chainlink/v2/core/utils/crelimits"
 )
 
 // gateAllows reports whether the given CRE gate allows the gated behavior.
 // When evaluation errors for reasons other than ErrorNotAllowed, it logs an error and returns false.
 func gateAllows(ctx context.Context, lggr logger.Logger, gate limits.GateLimiter, gateName string) bool {
-	err := gate.AllowErr(ctx)
-	if err == nil {
-		return true
-	}
-	if errors.Is(err, limits.ErrorNotAllowed{}) {
-		return false
-	}
-	lggr.Errorw("unexpected error evaluating CRE gate", "gate", gateName, "error", err)
-	return false
+	return crelimits.GateAllows(ctx, lggr, gate, gateName)
 }
 
 // resolveVaultOCRBoundLimitInt builds a short-lived BoundLimiter for an integer-sized CRE setting, reads Limit once, and closes the limiter.
diff --git a/core/services/workflows/monitoring/monitoring.go b/core/services/workflows/monitoring/monitoring.go
index 2ae2b5b0199..1289af6952d 100644
--- a/core/services/workflows/monitoring/monitoring.go
+++ b/core/services/workflows/monitoring/monitoring.go
@@ -84,6 +84,9 @@ type EngineMetrics struct {
 	donTimeErrorsCounter   metric.Int64Counter
 
 	orgIDMissingCounter metric.Int64Counter
+
+	limitReadFallbackTotal    metric.Int64Counter
+	limitCheckUnenforcedTotal metric.Int64Counter
 }
 
 func InitMonitoringResources() (em *EngineMetrics, err error) {
@@ -449,6 +452,22 @@ func InitMonitoringResources() (em *EngineMetrics, err error) {
 		return nil, fmt.Errorf("failed to register org id missing counter: %w", err)
 	}
 
+	em.limitReadFallbackTotal, err = beholder.GetMeter().Int64Counter(
+		"platform_engine_limit_read_fallback_total",
+		metric.WithDescription("Limit reads that failed and fell back to the static default, by limitKey"),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to register limit read fallback counter: %w", err)
+	}
+
+	em.limitCheckUnenforcedTotal, err = beholder.GetMeter().Int64Counter(
+		"platform_engine_limit_check_unenforced_total",
+		metric.WithDescription("Limit checks that could not be evaluated, so the limit went unenforced (failed open), by limitKey"),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to register limit check unenforced counter: %w", err)
+	}
+
 	return em, nil
 }
 
@@ -823,3 +842,24 @@ func (c WorkflowsMetricLabeler) IncrementOrgIDMissingCounter(ctx context.Context
 	otelLabels = append(otelLabels, attribute.String("reason", reason))
 	c.em.orgIDMissingCounter.Add(ctx, 1, metric.WithAttributes(otelLabels...))
 }
+
+// IncrementLimitReadFallbackCounter records one limit read that failed and fell back
+// to a default instead of dropping the execution/event. limitKey should be the
+// canonical settings.Setting.Key for the limit that failed.
+func (c WorkflowsMetricLabeler) IncrementLimitReadFallbackCounter(ctx context.Context, limitKey string) {
+	lc := c.With(platform.KeyLimitKey, limitKey)
+	otelLabels := beholder.OtelAttributes(lc.Labels).AsStringAttributes()
+	lc.em.limitReadFallbackTotal.Add(ctx, 1, metric.WithAttributes(otelLabels...))
+}
+
+// IncrementLimitCheckUnenforcedCounter records one limit Check that could not be
+// evaluated (a settings read failure rather than the bound being exceeded), so the
+// limit was skipped and the operation allowed through. Distinct from
+// IncrementLimitReadFallbackCounter: there is no default to substitute here, the limit
+// simply went unenforced, so a non-zero rate means a limit is not being applied.
+// limitKey should be the canonical settings.Setting.Key for the limit that failed.
+func (c WorkflowsMetricLabeler) IncrementLimitCheckUnenforcedCounter(ctx context.Context, limitKey string) {
+	lc := c.With(platform.KeyLimitKey, limitKey)
+	otelLabels := beholder.OtelAttributes(lc.Labels).AsStringAttributes()
+	lc.em.limitCheckUnenforcedTotal.Add(ctx, 1, metric.WithAttributes(otelLabels...))
+}
diff --git a/core/services/workflows/v2/config.go b/core/services/workflows/v2/config.go
index 454240c0c57..b5e6c31259d 100644
--- a/core/services/workflows/v2/config.go
+++ b/core/services/workflows/v2/config.go
@@ -3,6 +3,7 @@ package v2
 import (
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/jonboulle/clockwork"
 
@@ -87,7 +88,7 @@ type EngineLimiters struct {
 	TriggerRegistrationsTime limits.TimeLimiter
 	TriggerSubscription      limits.BoundLimiter[int]
 	TriggerEventQueue        limits.QueueLimiter[enqueuedTriggerEvent]
-	TriggerEventQueueTime    limits.TimeLimiter
+	TriggerEventMaxAge       limits.BoundLimiter[time.Duration]
 	ExecutionConcurrency     limits.ResourcePoolLimiter[int]
 
 	WASMBinarySize           limits.BoundLimiter[config.Size]
@@ -155,7 +156,7 @@ func (l *EngineLimiters) init(lf limits.Factory, cfgFn func(*cresettings.Workflo
 	if err != nil {
 		return
 	}
-	l.TriggerEventQueueTime, err = lf.MakeTimeLimiter(cfg.TriggerEventQueueTimeout)
+	l.TriggerEventMaxAge, err = limits.MakeUpperBoundLimiter(lf, cfg.TriggerEventQueueTimeout)
 	if err != nil {
 		return
 	}
@@ -292,7 +293,7 @@ func (l *EngineLimiters) EvictWorkflow(workflowID string) error {
 		l.TriggerRegistrationsTime,
 		l.TriggerSubscription,
 		l.TriggerEventQueue,
-		l.TriggerEventQueueTime,
+		l.TriggerEventMaxAge,
 		l.ExecutionConcurrency,
 		l.WASMBinarySize,
 		l.WASMMemorySize,
@@ -336,7 +337,7 @@ func (l *EngineLimiters) Close() error {
 		l.TriggerRegistrationsTime,
 		l.TriggerSubscription,
 		l.TriggerEventQueue,
-		l.TriggerEventQueueTime,
+		l.TriggerEventMaxAge,
 		l.ExecutionConcurrency,
 		l.WASMBinarySize,
 		l.WASMMemorySize,
diff --git a/core/services/workflows/v2/engine.go b/core/services/workflows/v2/engine.go
index 59cc894508c..45889c90f6c 100644
--- a/core/services/workflows/v2/engine.go
+++ b/core/services/workflows/v2/engine.go
@@ -24,6 +24,7 @@ import (
 
 	"github.com/smartcontractkit/chainlink-common/pkg/aggregation"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
+	"github.com/smartcontractkit/chainlink-common/pkg/config"
 	"github.com/smartcontractkit/chainlink-common/pkg/contexts"
 	"github.com/smartcontractkit/chainlink-common/pkg/custmsg"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
@@ -693,17 +694,17 @@ func (e *Engine) handleAllTriggerEvents(ctx context.Context) {
 		eventAge := e.cfg.Clock.Now().Sub(queueHead.timestamp)
 		e.logger().Debugw("Popped a trigger event from the queue", "eventID", eventID, "eventAgeMs", eventAge.Milliseconds())
 		triggerMetricLabels.RecordTriggerEventQueueWaitSeconds(ctx, eventAge.Seconds())
-		triggerEventMaxAge, err := e.cfg.LocalLimiters.TriggerEventQueueTime.Limit(ctx)
-		if err != nil {
-			e.logger().Errorw("Failed to get trigger event queue time limit", "err", err)
-			triggerMetricLabels.IncrementTriggerEventDroppedTotal(ctx, monitoring.TriggerDropReasonQueueAgeLimitReadFailed)
-			continue
-		}
-		if eventAge > triggerEventMaxAge {
-			e.logger().Warnw("Trigger event is too old, skipping execution", "triggerID", queueHead.triggerCapID, "eventID", eventID, "eventAgeMs", eventAge.Milliseconds())
-			triggerMetricLabels.IncrementTriggerEventExpiredCounter(ctx)
-			triggerMetricLabels.IncrementTriggerEventDroppedTotal(ctx, monitoring.TriggerDropReasonExpired)
-			continue
+		if ageErr := e.cfg.LocalLimiters.TriggerEventMaxAge.Check(ctx, eventAge); ageErr != nil {
+			if errBoundLimited, ok := errors.AsType[limits.ErrorBoundLimited[time.Duration]](ageErr); ok {
+				e.logger().Warnw("Trigger event is too old, skipping execution", "triggerID", queueHead.triggerCapID, "eventID", eventID, "eventAgeMs", eventAge.Milliseconds(), "maxAgeMs", errBoundLimited.Limit.Milliseconds())
+				triggerMetricLabels.IncrementTriggerEventExpiredCounter(ctx)
+				triggerMetricLabels.IncrementTriggerEventDroppedTotal(ctx, monitoring.TriggerDropReasonExpired)
+				continue
+			}
+			// A settings read failure is not an expiry: run the execution rather than
+			// dropping a customer's trigger event over a transient config read.
+			e.logger().Errorw("Failed to check trigger event queue age limit; proceeding with execution", "err", ageErr)
+			triggerMetricLabels.IncrementLimitCheckUnenforcedCounter(ctx, cresettings.Default.PerWorkflow.TriggerEventQueueTimeout.Key)
 		}
 		semWaitStart := e.cfg.Clock.Now()
 		free, err := e.executionsSemaphore.Wait(ctx, 1) // block if too many concurrent workflow executions
@@ -799,6 +800,18 @@ func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueue
 		}
 	}()
 
+	// emitDroppedExecution publishes the Started/Finished pair for an execution abandoned
+	// before the normal Started/Finished emit points below, so the failure reaches the UI
+	// 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) {
+		executionStatus = store.StatusErrored
+		_ = events.EmitExecutionStartedEvent(ctx, loggerLabels, triggerEvent.ID, executionID)
+		_ = events.EmitExecutionFinishedEvent(ctx, loggerLabels, store.StatusErrored, executionID, cause, class, lggr)
+		e.metrics.IncrementWorkflowExecutionFinishedCounter(ctx, store.StatusErrored)
+	}
+
 	needShardOwnerCheck := e.cfg.ShardRoutingSteady == nil || !e.cfg.ShardRoutingSteady.SkipCommittedOwnerCheck()
 	if e.cfg.ShardingEnabled && needShardOwnerCheck {
 		var verdict shardownership.Verdict
@@ -868,22 +881,48 @@ func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueue
 	}
 
 	isMetering := meteringErr == nil
+	// meteringEnded tracks whether the inline End() call further below already ran.
+	// If startExecution returns before reaching it (any drop after Start),
+	// this guarded defer ends the report instead, so a redelivered
+	// trigger event for the same executionID doesn't hit ErrReportExists and run unmetered.
+	meteringEnded := false
+	defer func() {
+		if !isMetering || meteringEnded {
+			return
+		}
+		if endErr := e.meterReports.End(ctx, executionID); endErr != nil {
+			lggr.Errorw("could not end metering report after dropped execution", "err", endErr)
+		}
+	}()
 	if isMetering {
 		mrErr := meteringReport.Reserve(ctx)
 		if mrErr != nil {
 			lggr.Errorw("could not reserve metering", "err", mrErr)
 			triggerDrop(monitoring.TriggerDropReasonMeteringReserveFailed)
+			// ErrInsufficientFunding is the customer's own out-of-credits state, not a
+			// platform failure; every other Reserve error (billing transport, etc.) fails
+			// open inside Reserve itself, so ErrInsufficientFunding is the only error
+			// actually reachable here today.
+			cause, class := mrErr, events.ErrorClassificationSystem
+			if errors.Is(mrErr, metering.ErrInsufficientFunding) {
+				cause = fmt.Errorf("insufficient credits to reserve this execution; top up your account balance: %w", mrErr)
+				class = events.ErrorClassificationUser
+			}
+			emitDroppedExecution(cause, class)
 			return
 		}
 
 		e.deductStandardBalances(ctx, meteringReport)
 	}
 
+	// WithTimeout returns a usable ctx/cancel even on a read failure; err is advisory.
 	execCtx, execCancel, err := e.cfg.LocalLimiters.ExecutionTime.WithTimeout(ctx)
 	if err != nil {
-		lggr.Errorw("Failed to get execution time limit", "err", err)
-		triggerDrop(monitoring.TriggerDropReasonExecutionTimeLimitReadFailed)
-		return
+		lggr.Errorw("Failed to get execution time limit; proceeding with the default timeout", "err", err)
+		e.metrics.IncrementLimitReadFallbackCounter(ctx, cresettings.Default.PerWorkflow.ExecutionTimeout.Key)
+		if execCtx == nil { // only nil when there's no tenant to resolve a limit for at all
+			execCtx, execCancel = context.WithTimeout(ctx, cresettings.Default.PerWorkflow.ExecutionTimeout.DefaultValue)
+		}
 	}
 	defer execCancel()
 	triggerCapID := wrappedTriggerEvent.triggerCapID
@@ -897,11 +936,12 @@ func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueue
 	executionLogger := logger.With(lggr, "executionID", executionID, "triggerID", wrappedTriggerEvent.triggerCapID,
 		"triggerIndex", wrappedTriggerEvent.triggerIndex, "eventID", triggerEvent.ID)
 
+	// This is only a peek to size the user-log channel's burst buffer; LogEvent.Check
+	// (called per log line in emitUserLogs) is what actually enforces the cap.
 	maxUserLogEventsPerExecution, err := e.cfg.LocalLimiters.LogEvent.Limit(ctx)
 	if err != nil {
-		lggr.Errorw("Failed to get log event limit", "err", err)
-		triggerDrop(monitoring.TriggerDropReasonLogEventLimitReadFailed)
-		return
+		lggr.Errorw("Failed to get log event limit; using the default value", "err", err)
+		e.metrics.IncrementLimitReadFallbackCounter(ctx, cresettings.Default.PerWorkflow.LogEventLimit.Key)
 	}
 	userLogChan := make(chan *protoevents.LogLine, maxUserLogEventsPerExecution)
 	defer close(userLogChan)
@@ -913,6 +953,7 @@ func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueue
 	if err != nil {
 		executionLogger.Errorw("Failed to convert trigger index to uint64", "err", err)
 		triggerDrop(monitoring.TriggerDropReasonTriggerIndexInvalid)
+		emitDroppedExecution(err, events.ErrorClassificationSystem)
 		return
 	}
 
@@ -983,14 +1024,11 @@ func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueue
 	suspension := &suspensionTracker{}
 	timeProvider = newMeasuredTimeProvider(timeProvider, e.cfg.Clock, suspension)
 
+	// Limit is always usable even on a read failure; err is advisory.
 	moduleExecuteMaxResponseSizeBytes, err := e.cfg.LocalLimiters.ExecutionResponse.Limit(ctx)
 	if err != nil {
-		lggr.Errorw("Failed to get execution response size limit", "err", err)
-		executionStatus = store.StatusErrored
-		execErr = err
-		execErrClass = events.ErrorClassificationSystem
-		triggerDrop(monitoring.TriggerDropReasonExecutionResponseLimitReadFailed)
-		return
+		lggr.Errorw("Failed to get execution response size limit; using the default value", "err", err)
+		e.metrics.IncrementLimitReadFallbackCounter(ctx, cresettings.Default.PerWorkflow.ExecutionResponseLimit.Key)
 	}
 	if moduleExecuteMaxResponseSizeBytes < 0 {
 		execErr = fmt.Errorf("invalid moduleExecuteMaxResponseSizeBytes; must not be negative: %d", moduleExecuteMaxResponseSizeBytes)
@@ -1061,6 +1099,7 @@ func (e *Engine) startExecution(ctx context.Context, wrappedTriggerEvent enqueue
 			lggr.Errorw("could not set metering for compute", "err", mrErr)
 		}
 		mrErr = e.meterReports.End(ctx, executionID)
+		meteringEnded = true
 		if mrErr != nil {
 			lggr.Errorw("could not end metering report", "err", mrErr)
 		}
@@ -1236,10 +1275,11 @@ func (e *Engine) deductStandardBalances(ctx context.Context, meteringReport *met
 	// V2Engine runs the entirety of a module's execution as compute. Ensure that the max execution time can run.
 	// Add an extra second of metering padding for context cancel propagation
 	ctxCancelPadding := (time.Millisecond * 1000).Milliseconds()
+	// Limit is always usable even on a read failure; err is advisory.
 	workflowExecutionTimeout, err := e.cfg.LocalLimiters.ExecutionTime.Limit(ctx)
 	if err != nil {
-		e.logger().Errorw("Failed to get execution time limit", "err", err)
-		return
+		e.logger().Errorw("Failed to get execution time limit; using the default value", "err", err)
+		e.metrics.IncrementLimitReadFallbackCounter(ctx, cresettings.Default.PerWorkflow.ExecutionTimeout.Key)
 	}
 	compMs := decimal.NewFromInt(workflowExecutionTimeout.Milliseconds() + ctxCancelPadding)
 	computeUnit := billing.ResourceType_RESOURCE_TYPE_COMPUTE.String()
@@ -1264,22 +1304,22 @@ func (e *Engine) emitUserLogs(ctx context.Context, userLogChan chan *protoevents
 		if e.cfg.DebugMode {
 			e.logger().Debugf("User log: <<<%s>>>, local node timestamp: %s", logLine.Message, logLine.NodeTimestamp)
 		}
-		err := e.cfg.LocalLimiters.LogEvent.Check(emitCtx, count)
-		if err != nil {
+		if err := e.cfg.LocalLimiters.LogEvent.Check(emitCtx, count); err != nil {
 			if errBoundLimited, ok := errors.AsType[limits.ErrorBoundLimited[int]](err); ok {
 				e.logger().Warnw("Max user log events per execution reached, dropping event", "maxEvents", errBoundLimited.Limit, "err", err)
 				return false
 			}
-			e.logger().Errorw("Failed to get user log event limit", "err", err)
-			return false
-		}
-		maxUserLogLength, err := e.cfg.LocalLimiters.LogLine.Limit(emitCtx)
-		if err != nil {
-			e.logger().Errorw("Failed to get user log line limit", "err", err)
-			return false
+			// A settings read failure should not stop the drain. Fail open instead.
+			e.logger().Errorw("Failed to check user log event limit; emitting anyway", "err", err)
+			e.metrics.IncrementLimitCheckUnenforcedCounter(emitCtx, cresettings.Default.PerWorkflow.LogEventLimit.Key)
 		}
-		if len(logLine.Message) > int(maxUserLogLength) {
-			logLine.Message = logLine.Message[:maxUserLogLength] + " ...(truncated)"
+		if err := e.cfg.LocalLimiters.LogLine.Check(emitCtx, config.Size(len(logLine.Message))); err != nil {
+			if errBoundLimited, ok := errors.AsType[limits.ErrorBoundLimited[config.Size]](err); ok {
+				logLine.Message = logLine.Message[:errBoundLimited.Limit] + " ...(truncated)"
+			} else {
+				e.logger().Errorw("Failed to check user log line limit; emitting untruncated", "err", err)
+				e.metrics.IncrementLimitCheckUnenforcedCounter(emitCtx, cresettings.Default.PerWorkflow.LogLineLimit.Key)
+			}
 		}
 
 		if err := events.EmitUserLogs(emitCtx, executionLabels, []*protoevents.LogLine{logLine}, executionID); err != nil {
@@ -1326,17 +1366,18 @@ func (e *Engine) emitUserLogs(ctx context.Context, userLogChan chan *protoevents
 
 func (e *Engine) donTimeRequestTimeout(ctx context.Context, limiter limits.TimeLimiter) time.Duration {
 	defaultTimeout := cresettings.Default.PerWorkflow.DONTime.RequestTimeout.DefaultValue
-	if limiter != nil {
-		limit, err := limiter.Limit(ctx)
-		if err != nil {
-			e.logger().Errorw("Failed to get DON time request timeout", "err", err)
-			return defaultTimeout
-		}
-		if limit <= 0 {
-			e.logger().Warnw("DON time request timeout is less than or equal to 0, using default timeout", "defaultTimeout", defaultTimeout)
-			return defaultTimeout
-		}
-		return limit
+	if limiter == nil {
+		return defaultTimeout
+	}
+	// Limit is always usable even on a read failure; only a nonsensical (<=0) result
+	// falls back to defaultTimeout.
+	limit, err := limiter.Limit(ctx)
+	if err != nil {
+		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
 	}
-	return defaultTimeout
+	return limit
 }
diff --git a/core/services/workflows/v2/engine_drop_paths_test.go b/core/services/workflows/v2/engine_drop_paths_test.go
new file mode 100644
index 00000000000..3a4cf79fb59
--- /dev/null
+++ b/core/services/workflows/v2/engine_drop_paths_test.go
@@ -0,0 +1,453 @@
+package v2_test
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/emptypb"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder/beholdertest"
+	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
+	"github.com/smartcontractkit/chainlink-common/pkg/config"
+	"github.com/smartcontractkit/chainlink-common/pkg/custmsg"
+	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
+	regmocks "github.com/smartcontractkit/chainlink-common/pkg/types/core/mocks"
+	"github.com/smartcontractkit/chainlink-common/pkg/workflows/wasm/host"
+	modulemocks "github.com/smartcontractkit/chainlink-common/pkg/workflows/wasm/host/mocks"
+	billing "github.com/smartcontractkit/chainlink-protos/billing/go"
+	sdkpb "github.com/smartcontractkit/chainlink-protos/cre/go/sdk"
+	pb "github.com/smartcontractkit/chainlink-protos/workflows/go/events"
+	eventsv2 "github.com/smartcontractkit/chainlink-protos/workflows/go/v2"
+	capmocks "github.com/smartcontractkit/chainlink/v2/core/capabilities/mocks"
+	workflowEvents "github.com/smartcontractkit/chainlink/v2/core/services/workflows/events"
+	metmocks "github.com/smartcontractkit/chainlink/v2/core/services/workflows/metering/mocks"
+	v2 "github.com/smartcontractkit/chainlink/v2/core/services/workflows/v2"
+	"github.com/smartcontractkit/chainlink/v2/core/utils/matches"
+)
+
+const (
+	v2ExecutionStartedEntity  = "workflows.v2." + workflowEvents.WorkflowExecutionStarted
+	v2ExecutionFinishedEntity = "workflows.v2." + workflowEvents.WorkflowExecutionFinished
+	v1UserLogsEntity          = "workflows.v1." + workflowEvents.UserLogs
+)
+
+// failAfterNBoundLimiter succeeds (returning ok, nil) for the first n calls to Limit,
+// then returns (ok, failErr) after that, mirroring the real chainlink-common
+// BoundLimiter. LogEvent and ExecutionResponse are both peeked once during
+// trigger-subscription init and again per execution, so an always-failing fake would
+// break engine initialization before the value-on-error path could be exercised.
+type failAfterNBoundLimiter[N limits.Number] struct {
+	mu      sync.Mutex
+	calls   int
+	n       int
+	ok      N
+	failErr error
+}
+
+func (f *failAfterNBoundLimiter[N]) Limit(context.Context) (N, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.calls++
+	if f.calls > f.n {
+		return f.ok, f.failErr // resolved value, not zero; err is advisory
+	}
+	return f.ok, nil
+}
+
+func (f *failAfterNBoundLimiter[N]) Check(context.Context, N) error { return nil }
+func (f *failAfterNBoundLimiter[N]) Close() error                   { return nil }
+
+// alwaysFailTimeLimiter fails every Limit/WithTimeout call. Safe to use unconditionally
+// for ExecutionTime, which is not read during trigger-subscription init.
+type alwaysFailTimeLimiter struct{ err error }
+
+func (f *alwaysFailTimeLimiter) Limit(context.Context) (time.Duration, error) { return 0, f.err }
+
+func (f *alwaysFailTimeLimiter) WithTimeout(context.Context) (context.Context, func(), error) {
+	return nil, nil, f.err
+}
+
+func (f *alwaysFailTimeLimiter) Close() error { return nil }
+
+// alwaysErrCheckLimiter fails every Check with a plain error, standing in for a settings
+// read failure rather than the bound actually being exceeded (which would be an
+// ErrorBoundLimited). Used to exercise the fail-open branches in processLogLine.
+//
+// Limit deliberately still succeeds: LogEvent.Limit is peeked during trigger-subscription
+// init to size the user-log channel, so failing it too would break engine startup before
+// the Check path under test is ever reached.
+type alwaysErrCheckLimiter[N limits.Number] struct {
+	ok  N
+	err error
+}
+
+func (f *alwaysErrCheckLimiter[N]) Limit(context.Context) (N, error) { return f.ok, nil }
+func (f *alwaysErrCheckLimiter[N]) Check(context.Context, N) error   { return f.err }
+func (f *alwaysErrCheckLimiter[N]) Close() error                     { return nil }
+
+// fakeShardResolver reports a fixed shard-ownership verdict, standing in for a real
+// ring orchestrator lookup.
+type fakeShardResolver struct {
+	shardID uint32
+	found   bool
+	err     error
+}
+
+func (f *fakeShardResolver) ResolveShard(context.Context, string, string) (uint32, bool, error) {
+	return f.shardID, f.found, f.err
+}
+
+func (f *fakeShardResolver) ResolveShards(context.Context, []string, []string) (map[string]uint32, error) {
+	return nil, nil
+}
+
+// latestV2FinishedEvent returns the most recently emitted v2 WorkflowExecutionFinished
+// event, or ok=false if none has been emitted yet.
+func latestV2FinishedEvent(t *testing.T, observer beholdertest.Observer) (evt *eventsv2.WorkflowExecutionFinished, ok bool) {
+	t.Helper()
+	msgs := observer.Messages(t, "beholder_entity", v2ExecutionFinishedEntity)
+	if len(msgs) == 0 {
+		return nil, false
+	}
+	evt = &eventsv2.WorkflowExecutionFinished{}
+	require.NoError(t, proto.Unmarshal(msgs[len(msgs)-1].Body, evt))
+	return evt, true
+}
+
+// userLogLines returns every user log line emitted so far via the v1 UserLogs event.
+func userLogLines(t *testing.T, observer beholdertest.Observer) []string {
+	t.Helper()
+	var lines []string
+	for _, msg := range observer.Messages(t, "beholder_entity", v1UserLogsEntity) {
+		var payload pb.UserLogs
+		if err := proto.Unmarshal(msg.Body, &payload); err != nil {
+			continue
+		}
+		for _, l := range payload.LogLines {
+			lines = append(lines, l.Message)
+		}
+	}
+	return lines
+}
+
+// dropPathHarness is a minimal engine with a single trigger subscription (id_0)
+// registered, ready to have a trigger event pushed through eventCh to drive
+// startExecution. Callers customize cfg.LocalLimiters / cfg.ShardingEnabled / etc.
+// via configure before the harness starts the engine, and set up the second
+// Module.Execute expectation (for the per-execution call, if reached) themselves.
+type dropPathHarness struct {
+	module              *modulemocks.ModuleV2
+	eventCh             chan capabilities.TriggerResponse
+	executionFinishedCh chan string
+	beholderObserver    beholdertest.Observer
+}
+
+func newDropPathHarness(t *testing.T, billingClient *metmocks.BillingClient, configure func(cfg *v2.EngineConfig)) *dropPathHarness {
+	t.Helper()
+
+	module := modulemocks.NewModuleV2(t)
+	module.EXPECT().Start()
+	module.EXPECT().Close()
+	capreg := regmocks.NewCapabilitiesRegistry(t)
+	capreg.EXPECT().LocalNode(matches.AnyContext).Return(newNode(t), nil)
+
+	initDoneCh := make(chan error, 1)
+	subscribedToTriggersCh := make(chan []string, 1)
+	executionFinishedCh := make(chan string, 1)
+
+	cfg := defaultTestConfig(t, nil)
+	cfg.Module = module
+	cfg.CapRegistry = capreg
+	cfg.BillingClient = billingClient
+	cfg.OrgResolver = &mockOrgResolver{orgID: "test-org-123"}
+	cfg.Hooks = v2.LifecycleHooks{
+		OnInitialized: func(err error) {
+			initDoneCh <- err
+		},
+		OnSubscribedToTriggers: func(triggerIDs []string) {
+			subscribedToTriggersCh <- triggerIDs
+		},
+		OnExecutionFinished: func(executionID string, _ string) {
+			executionFinishedCh <- executionID
+		},
+	}
+	beholderObserver := beholdertest.NewObserver(t)
+	cfg.BeholderEmitter = custmsg.NewLabeler()
+
+	if configure != nil {
+		configure(cfg)
+	}
+
+	// Trigger-subscription init always calls Module.Execute once for the Subscribe
+	// request, regardless of what's under test below.
+	module.EXPECT().Execute(matches.AnyContext, mock.Anything, mock.Anything).Return(newTriggerSubs(1), nil).Once()
+	trigger := capmocks.NewTriggerCapability(t)
+	capreg.EXPECT().GetTrigger(matches.AnyContext, "id_0").Return(trigger, nil)
+	eventCh := make(chan capabilities.TriggerResponse)
+	trigger.EXPECT().RegisterTrigger(matches.AnyContext, mock.Anything).Return(eventCh, nil).Once()
+	trigger.EXPECT().UnregisterTrigger(matches.AnyContext, mock.Anything).Return(nil).Once()
+	// Reserve-failure drop paths return before the trigger event is ever
+	// ACKed, so this must not be a hard requirement across all harness users.
+	trigger.EXPECT().AckEvent(matches.AnyContext, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe()
+
+	engine, err := v2.NewEngine(cfg)
+	require.NoError(t, err)
+	require.NoError(t, engine.Start(t.Context()))
+	t.Cleanup(func() { require.NoError(t, engine.Close()) })
+
+	require.NoError(t, <-initDoneCh)
+	require.Equal(t, []string{"id_0"}, <-subscribedToTriggersCh)
+
+	return &dropPathHarness{
+		module:              module,
+		eventCh:             eventCh,
+		executionFinishedCh: executionFinishedCh,
+		beholderObserver:    beholderObserver,
+	}
+}
+
+// TestEngine_LimitReadFallback_ExecutesAnyway covers: an execution proceeds to
+// completion (instead of being silently dropped) when a limit read fails,
+// falling back to the setting's static default.
+func TestEngine_LimitReadFallback_ExecutesAnyway(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap
+	testCases := map[string]func(cfg *v2.EngineConfig){
+		"ExecutionTime": func(cfg *v2.EngineConfig) {
+			cfg.LocalLimiters.ExecutionTime = &alwaysFailTimeLimiter{err: errors.New("limit read boom")}
+		},
+		"LogEvent": func(cfg *v2.EngineConfig) {
+			cfg.LocalLimiters.LogEvent = &failAfterNBoundLimiter[int]{n: 1, ok: 1000, failErr: errors.New("limit read boom")}
+		},
+		"ExecutionResponse": func(cfg *v2.EngineConfig) {
+			cfg.LocalLimiters.ExecutionResponse = &failAfterNBoundLimiter[config.Size]{n: 1, ok: config.Size(10 * 1024 * 1024), failErr: errors.New("limit read boom")}
+		},
+	}
+
+	for name, breakLimiter := range testCases { //nolint:paralleltest // shares the package-level beholder singleton
+		t.Run(name, func(t *testing.T) {
+			billingClient := setupMockBillingClient(t)
+			harness := newDropPathHarness(t, billingClient, breakLimiter)
+
+			harness.module.EXPECT().Execute(matches.AnyContext, mock.Anything, mock.Anything).Return(nil, nil).Once()
+			harness.eventCh <- capabilities.TriggerResponse{
+				Event: capabilities.TriggerEvent{TriggerType: "basic-trigger@1.0.0", ID: "event_" + name},
+			}
+
+			executionID := <-harness.executionFinishedCh
+			require.NotEmpty(t, executionID)
+
+			startedMsgs := harness.beholderObserver.Messages(t, "beholder_entity", v2ExecutionStartedEntity)
+			assert.NotEmpty(t, startedMsgs, "execution should still emit a Started event despite the limit read failure")
+
+			evt, ok := latestV2FinishedEvent(t, harness.beholderObserver)
+			require.True(t, ok, "execution should still emit a Finished event despite the limit read failure")
+			assert.Equal(t, eventsv2.ExecutionStatus_EXECUTION_STATUS_SUCCEEDED, evt.Status)
+		})
+	}
+}
+
+// TestEngine_LimitReadFallback_UsesLimiterValue: on a read failure, the engine must use
+// whatever value the limiter returns, not substitute its own compiled default.
+func TestEngine_LimitReadFallback_UsesLimiterValue(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap
+	const resolvedValue = config.Size(12345)
+
+	harness := newDropPathHarness(t, setupMockBillingClient(t), func(cfg *v2.EngineConfig) {
+		cfg.LocalLimiters.ExecutionResponse = &failAfterNBoundLimiter[config.Size]{n: 1, ok: resolvedValue, failErr: errors.New("limit read boom")}
+	})
+
+	var gotMaxResponseSize uint64
+	harness.module.EXPECT().Execute(matches.AnyContext, mock.Anything, mock.Anything).
+		Run(func(_ context.Context, req *sdkpb.ExecuteRequest, _ host.ExecutionHelper) {
+			gotMaxResponseSize = req.MaxResponseSize
+		}).
+		Return(nil, nil).Once()
+	harness.eventCh <- capabilities.TriggerResponse{
+		Event: capabilities.TriggerEvent{TriggerType: "basic-trigger@1.0.0", ID: "event_limiter_value"},
+	}
+
+	executionID := <-harness.executionFinishedCh
+	require.NotEmpty(t, executionID)
+
+	assert.Equal(t, uint64(resolvedValue), gotMaxResponseSize,
+		"engine must use the value the limiter returned, not fall back to its own compiled default")
+}
+
+// setupInsufficientFundingBilling returns a billing mock whose ReserveCredits denies
+// the reservation (Success: false), the only Reserve failure reachable in practice
+// (every other Reserve error fails open inside Reserve itself). expectReceipt controls
+// whether SubmitWorkflowReceipt (called from Reports.End) is asserted as called.
+func setupInsufficientFundingBilling(t *testing.T, expectReceipt bool) *metmocks.BillingClient {
+	t.Helper()
+	billingClient := metmocks.NewBillingClient(t)
+	billingClient.EXPECT().GetWorkflowExecutionRates(mock.Anything, mock.Anything).Return(&billing.GetWorkflowExecutionRatesResponse{
+		RateCards: []*billing.RateCard{
+			{
+				ResourceType:    billing.ResourceType_RESOURCE_TYPE_COMPUTE,
+				MeasurementUnit: billing.MeasurementUnit_MEASUREMENT_UNIT_MILLISECONDS,
+				UnitsPerCredit:  "0.0001",
+			},
+		},
+	}, nil)
+	billingClient.EXPECT().ReserveCredits(mock.Anything, mock.Anything).Return(&billing.ReserveCreditsResponse{Success: false}, nil)
+	receiptCall := billingClient.EXPECT().SubmitWorkflowReceipt(mock.Anything, mock.Anything).Return(&emptypb.Empty{}, nil)
+	if expectReceipt {
+		receiptCall.Once()
+	} else {
+		receiptCall.Maybe()
+	}
+	return billingClient
+}
+
+// TestEngine_MeteringReserveFailure_SurfacesAsUserError covers: a metering
+// Reserve failure (insufficient funding) used to drop the execution before the
+// Started/Finished emit points were reached, so the failure never reached the UI.
+// It now publishes the Started/Finished pair, attributed to the user (their own
+// out-of-credits state), not the platform.
+func TestEngine_MeteringReserveFailure_SurfacesAsUserError(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap
+	billingClient := setupInsufficientFundingBilling(t, false)
+	harness := newDropPathHarness(t, billingClient, nil)
+
+	harness.eventCh <- capabilities.TriggerResponse{
+		Event: capabilities.TriggerEvent{TriggerType: "basic-trigger@1.0.0", ID: "event_reserve_failure"},
+	}
+
+	require.EventuallyWithT(t, func(c *assert.CollectT) {
+		startedMsgs := harness.beholderObserver.Messages(t, "beholder_entity", v2ExecutionStartedEntity)
+		assert.NotEmpty(c, startedMsgs)
+	}, 5*time.Second, 50*time.Millisecond)
+
+	var evt *eventsv2.WorkflowExecutionFinished
+	require.EventuallyWithT(t, func(c *assert.CollectT) {
+		got, ok := latestV2FinishedEvent(t, harness.beholderObserver)
+		if !assert.True(c, ok) {
+			return
+		}
+		evt = got
+	}, 5*time.Second, 50*time.Millisecond)
+
+	assert.Equal(t, eventsv2.ExecutionStatus_EXECUTION_STATUS_FAILED, evt.Status)
+	assert.NotEmpty(t, evt.Error)
+	assert.Equal(t, eventsv2.ClassifiedExecutionStatus_CLASSIFIED_EXECUTION_STATUS_USER_ERROR, evt.ClassifiedStatus)
+}
+
+// TestEngine_MeteringReserveFailure_ReportReleased covers: before this fix,
+// meterReports.End was only reached on the happy/normal-error path, so a Reserve
+// failure leaked the report; a redelivered trigger event for the same executionID
+// would then hit ErrReportExists and run unmetered. SubmitWorkflowReceipt (only
+// reachable via Reports.End) being called proves the report was released.
+func TestEngine_MeteringReserveFailure_ReportReleased(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap
+	billingClient := setupInsufficientFundingBilling(t, true)
+	harness := newDropPathHarness(t, billingClient, nil)
+
+	harness.eventCh <- capabilities.TriggerResponse{
+		Event: capabilities.TriggerEvent{TriggerType: "basic-trigger@1.0.0", ID: "event_report_released"},
+	}
+
+	require.EventuallyWithT(t, func(c *assert.CollectT) {
+		_, ok := latestV2FinishedEvent(t, harness.beholderObserver)
+		assert.True(c, ok)
+	}, 5*time.Second, 50*time.Millisecond)
+	// billingClient's t.Cleanup (registered by metmocks.NewBillingClient) asserts
+	// SubmitWorkflowReceipt was actually called.
+}
+
+// TestEngine_ShardDenial_StaysSilent is a regression guard: with sharding on, every
+// node outside the owning shard denies every execution, so emitting Started/Finished
+// there (unlike the other drop paths) would publish a DON-wide failure for a run
+// that actually succeeded on the owner. This must stay silent.
+func TestEngine_ShardDenial_StaysSilent(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap
+	billingClient := metmocks.NewBillingClient(t) // no billing calls expected: shard check runs before metering
+
+	harness := newDropPathHarness(t, billingClient, func(cfg *v2.EngineConfig) {
+		cfg.ShardingEnabled = true
+		cfg.MyShardID = 0
+		cfg.ShardResolver = &fakeShardResolver{shardID: 1, found: true} // some other shard owns it
+	})
+
+	harness.eventCh <- capabilities.TriggerResponse{
+		Event: capabilities.TriggerEvent{TriggerType: "basic-trigger@1.0.0", ID: "event_shard_denied"},
+	}
+
+	require.Never(t, func() bool {
+		started := harness.beholderObserver.Messages(t, "beholder_entity", v2ExecutionStartedEntity)
+		finished := harness.beholderObserver.Messages(t, "beholder_entity", v2ExecutionFinishedEntity)
+		return len(started) > 0 || len(finished) > 0
+	}, 500*time.Millisecond, 25*time.Millisecond)
+}
+
+// TestEngine_UserLog_LogLineCheck covers the LogLine Limit()->Check() swap: the bound is
+// now enforced by the limiter (so it records usage/denied), truncation is driven by the
+// bound carried on the returned ErrorBoundLimited, and a settings read failure fails open
+// by emitting the line untruncated instead of falling back to a hardcoded default.
+func TestEngine_UserLog_LogLineCheck(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap
+	t.Run("over-long message truncated to the bound", func(t *testing.T) { //nolint:paralleltest // shares the package-level beholder singleton
+		harness := newDropPathHarness(t, setupMockBillingClient(t), func(cfg *v2.EngineConfig) {
+			cfg.LocalLimiters.LogLine = limits.NewUpperBoundLimiter[config.Size](10)
+		})
+
+		harness.module.EXPECT().Execute(matches.AnyContext, mock.Anything, mock.Anything).
+			Run(func(_ context.Context, _ *sdkpb.ExecuteRequest, executor host.ExecutionHelper) {
+				require.NoError(t, executor.EmitUserLog("0123456789ABCDEF"))
+			}).
+			Return(nil, nil).Once()
+		harness.eventCh <- capabilities.TriggerResponse{
+			Event: capabilities.TriggerEvent{TriggerType: "basic-trigger@1.0.0", ID: "event_truncated"},
+		}
+
+		require.EventuallyWithT(t, func(c *assert.CollectT) {
+			assert.Contains(c, userLogLines(t, harness.beholderObserver), "0123456789 ...(truncated)")
+		}, 5*time.Second, 50*time.Millisecond)
+	})
+
+	t.Run("settings read failure emits untruncated", func(t *testing.T) { //nolint:paralleltest // shares the package-level beholder singleton
+		harness := newDropPathHarness(t, setupMockBillingClient(t), func(cfg *v2.EngineConfig) {
+			cfg.LocalLimiters.LogLine = &alwaysErrCheckLimiter[config.Size]{err: errors.New("settings unavailable")}
+		})
+
+		const message = "this message is not over any real bound"
+		harness.module.EXPECT().Execute(matches.AnyContext, mock.Anything, mock.Anything).
+			Run(func(_ context.Context, _ *sdkpb.ExecuteRequest, executor host.ExecutionHelper) {
+				require.NoError(t, executor.EmitUserLog(message))
+			}).
+			Return(nil, nil).Once()
+		harness.eventCh <- capabilities.TriggerResponse{
+			Event: capabilities.TriggerEvent{TriggerType: "basic-trigger@1.0.0", ID: "event_untruncated"},
+		}
+
+		require.EventuallyWithT(t, func(c *assert.CollectT) {
+			assert.Contains(c, userLogLines(t, harness.beholderObserver), message)
+		}, 5*time.Second, 50*time.Millisecond)
+	})
+}
+
+// TestEngine_UserLog_LogEventCheckFailure_KeepsDraining covers the fail-open fix in the
+// LogEvent.Check branch: a settings read failure (a plain error, not ErrorBoundLimited)
+// used to return false from processLogLine, which ended the emitUserLogs drain goroutine
+// and silently discarded every remaining user log for the execution. Both lines must land.
+func TestEngine_UserLog_LogEventCheckFailure_KeepsDraining(t *testing.T) { //nolint:paralleltest // uses beholdertest.NewObserver, a global singleton swap
+	harness := newDropPathHarness(t, setupMockBillingClient(t), func(cfg *v2.EngineConfig) {
+		cfg.LocalLimiters.LogEvent = &alwaysErrCheckLimiter[int]{ok: 1000, err: errors.New("settings unavailable")}
+	})
+
+	harness.module.EXPECT().Execute(matches.AnyContext, mock.Anything, mock.Anything).
+		Run(func(_ context.Context, _ *sdkpb.ExecuteRequest, executor host.ExecutionHelper) {
+			require.NoError(t, executor.EmitUserLog("first line"))
+			require.NoError(t, executor.EmitUserLog("second line"))
+		}).
+		Return(nil, nil).Once()
+	harness.eventCh <- capabilities.TriggerResponse{
+		Event: capabilities.TriggerEvent{TriggerType: "basic-trigger@1.0.0", ID: "event_log_event_check_fails"},
+	}
+
+	require.EventuallyWithT(t, func(c *assert.CollectT) {
+		lines := userLogLines(t, harness.beholderObserver)
+		assert.Contains(c, lines, "first line")
+		assert.Contains(c, lines, "second line")
+	}, 5*time.Second, 50*time.Millisecond)
+}
diff --git a/core/services/workflows/v2/engine_limit_metrics_test.go b/core/services/workflows/v2/engine_limit_metrics_test.go
new file mode 100644
index 00000000000..e1ab48903ad
--- /dev/null
+++ b/core/services/workflows/v2/engine_limit_metrics_test.go
@@ -0,0 +1,92 @@
+package v2
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/attribute"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/metrics"
+	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/monitoring"
+)
+
+// collectCounterByAttr sums the named counter and returns the values observed for
+// attrKey. Companion to collectCounterValue (engine_org_id_missing_test.go), which
+// reads the "reason" attribute instead.
+func collectCounterByAttr(t *testing.T, reader *sdkmetric.ManualReader, name, attrKey string) (int64, []string) {
+	t.Helper()
+
+	var rm metricdata.ResourceMetrics
+	require.NoError(t, reader.Collect(context.Background(), &rm))
+
+	var total int64
+	var attrs []string
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			if m.Name != name {
+				continue
+			}
+			data, ok := m.Data.(metricdata.Sum[int64])
+			require.True(t, ok, "expected Sum[int64] for %s, got %T", name, m.Data)
+			for _, dp := range data.DataPoints {
+				total += dp.Value
+				if v, ok := dp.Attributes.Value(attribute.Key(attrKey)); ok {
+					attrs = append(attrs, v.AsString())
+				}
+			}
+		}
+	}
+	return total, attrs
+}
+
+// TestEngine_LimitDegradationCounters covers the two ways a CRE settings read failure
+// degrades a limiter at a call site, which are tracked separately on purpose:
+//
+//   - read fallback: the limit could not be read, so the static default was substituted.
+//     The limit is still applied.
+//   - check unenforced: the Check could not be evaluated, so the limit was skipped and
+//     the operation allowed through. The limit is NOT applied, which is the more serious
+//     of the two and warrants its own alert rather than a label on the other counter.
+//
+//nolint:paralleltest // setupTestMeter swaps the global beholder client
+func TestEngine_LimitDegradationCounters(t *testing.T) {
+	const (
+		unenforcedCounter = "platform_engine_limit_check_unenforced_total"
+		fallbackCounter   = "platform_engine_limit_read_fallback_total"
+		limitKeyAttr      = "limitKey"
+	)
+
+	t.Run("check unenforced records the limit key", func(t *testing.T) {
+		reader := setupTestMeter(t)
+		em, err := monitoring.InitMonitoringResources()
+		require.NoError(t, err)
+		labeler := monitoring.NewWorkflowsMetricLabeler(metrics.NewLabeler(), em)
+
+		labeler.IncrementLimitCheckUnenforcedCounter(t.Context(), "PerWorkflow.LogLineLimit")
+		labeler.IncrementLimitCheckUnenforcedCounter(t.Context(), "PerWorkflow.LogEventLimit")
+
+		total, keys := collectCounterByAttr(t, reader, unenforcedCounter, limitKeyAttr)
+		assert.Equal(t, int64(2), total)
+		assert.ElementsMatch(t, []string{"PerWorkflow.LogLineLimit", "PerWorkflow.LogEventLimit"}, keys)
+	})
+
+	t.Run("the two degradation modes are separate series", func(t *testing.T) {
+		reader := setupTestMeter(t)
+		em, err := monitoring.InitMonitoringResources()
+		require.NoError(t, err)
+		labeler := monitoring.NewWorkflowsMetricLabeler(metrics.NewLabeler(), em)
+
+		labeler.IncrementLimitReadFallbackCounter(t.Context(), "PerWorkflow.ExecutionTimeout")
+
+		unenforced, _ := collectCounterByAttr(t, reader, unenforcedCounter, limitKeyAttr)
+		fallback, fallbackKeys := collectCounterByAttr(t, reader, fallbackCounter, limitKeyAttr)
+
+		assert.Equal(t, int64(1), fallback)
+		assert.Equal(t, []string{"PerWorkflow.ExecutionTimeout"}, fallbackKeys)
+		assert.Zero(t, unenforced, "a read fallback must not be counted as an unenforced check")
+	})
+}
diff --git a/core/utils/crelimits/gate.go b/core/utils/crelimits/gate.go
new file mode 100644
index 00000000000..85718a08f3f
--- /dev/null
+++ b/core/utils/crelimits/gate.go
@@ -0,0 +1,46 @@
+package crelimits
+
+import (
+	"context"
+	"errors"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
+)
+
+type ErrorLogger interface {
+	Errorw(msg string, keysAndValues ...any)
+}
+
+// GateOpen reports whether gate is open.
+//
+// Prefer this over calling [limits.GateLimiter.Limit] directly: Limit exists to hand
+// out a raw value and only records the limit gauge, whereas AllowErr (used here) is the
+// enforcement method and records the limiter's usage/denied metrics. GateOpen keeps
+// Limit's ergonomics — a closed gate is (false, nil), so only a genuine evaluation
+// failure returns a non-nil error and callers that fail closed can keep doing so.
+func GateOpen(ctx context.Context, gate limits.GateLimiter) (bool, error) {
+	err := gate.AllowErr(ctx)
+	switch {
+	case err == nil:
+		return true, nil
+	case errors.Is(err, limits.ErrorNotAllowed{}):
+		return false, nil
+	default:
+		return false, err
+	}
+}
+
+// GateAllows is [GateOpen] for callers that treat an unevaluatable gate as closed.
+// The evaluation failure is logged at error level and reported as false.
+//
+// Only use this where "closed" is the safe outcome. If a gate read failure must abort
+// the operation instead (a fail-closed gate whose closed state *permits* something),
+// use GateOpen and handle the error explicitly.
+func GateAllows(ctx context.Context, lggr ErrorLogger, gate limits.GateLimiter, gateName string) bool {
+	open, err := GateOpen(ctx, gate)
+	if err != nil {
+		lggr.Errorw("unexpected error evaluating CRE gate", "gate", gateName, "error", err)
+		return false
+	}
+	return open
+}
diff --git a/core/utils/crelimits/gate_test.go b/core/utils/crelimits/gate_test.go
new file mode 100644
index 00000000000..b807fde3c5c
--- /dev/null
+++ b/core/utils/crelimits/gate_test.go
@@ -0,0 +1,95 @@
+package crelimits_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
+	"github.com/smartcontractkit/chainlink/v2/core/utils/crelimits"
+)
+
+// errGate is a GateLimiter whose AllowErr always fails with a given error, standing in
+// for a settings read failure (as opposed to limits.ErrorNotAllowed, which means the
+// gate evaluated fine and is closed).
+type errGate struct{ err error }
+
+func (g errGate) Limit(context.Context) (bool, error) { return false, g.err }
+func (g errGate) AllowErr(context.Context) error      { return g.err }
+func (g errGate) Close() error                        { return nil }
+
+type recordingLogger struct{ calls int }
+
+func (l *recordingLogger) Errorw(string, ...any) { l.calls++ }
+
+func TestGateOpen(t *testing.T) {
+	t.Parallel()
+
+	readErr := errors.New("settings unavailable")
+
+	testCases := []struct {
+		name     string
+		gate     limits.GateLimiter
+		wantOpen bool
+		wantErr  error
+	}{
+		{name: "open gate", gate: limits.NewGateLimiter(true), wantOpen: true},
+		{name: "closed gate is not an error", gate: limits.NewGateLimiter(false), wantOpen: false},
+		{name: "read failure surfaces", gate: errGate{err: readErr}, wantOpen: false, wantErr: readErr},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			open, err := crelimits.GateOpen(t.Context(), tc.gate)
+			if tc.wantErr != nil {
+				require.ErrorIs(t, err, tc.wantErr)
+			} else {
+				require.NoError(t, err)
+			}
+			assert.Equal(t, tc.wantOpen, open)
+		})
+	}
+}
+
+// TestGateOpen_ClosedIsDistinctFromUnevaluatable is the property the fail-closed call
+// sites depend on: a closed gate must not look like a read failure, and vice versa.
+func TestGateOpen_ClosedIsDistinctFromUnevaluatable(t *testing.T) {
+	t.Parallel()
+
+	closedOpen, closedErr := crelimits.GateOpen(t.Context(), limits.NewGateLimiter(false))
+	require.NoError(t, closedErr, "a closed gate is a normal outcome, not an error")
+	assert.False(t, closedOpen)
+
+	failedOpen, failedErr := crelimits.GateOpen(t.Context(), errGate{err: errors.New("boom")})
+	require.Error(t, failedErr, "an unevaluatable gate must be distinguishable so callers can fail closed")
+	assert.False(t, failedOpen)
+}
+
+func TestGateAllows(t *testing.T) {
+	t.Parallel()
+
+	t.Run("open gate allows without logging", func(t *testing.T) {
+		t.Parallel()
+		lggr := &recordingLogger{}
+		assert.True(t, crelimits.GateAllows(t.Context(), lggr, limits.NewGateLimiter(true), "TestGate"))
+		assert.Zero(t, lggr.calls)
+	})
+
+	t.Run("closed gate denies without logging", func(t *testing.T) {
+		t.Parallel()
+		lggr := &recordingLogger{}
+		assert.False(t, crelimits.GateAllows(t.Context(), lggr, limits.NewGateLimiter(false), "TestGate"))
+		assert.Zero(t, lggr.calls, "a closed gate is expected, not an unexpected error")
+	})
+
+	t.Run("read failure denies and logs", func(t *testing.T) {
+		t.Parallel()
+		lggr := &recordingLogger{}
+		assert.False(t, crelimits.GateAllows(t.Context(), lggr, errGate{err: errors.New("boom")}, "TestGate"))
+		assert.Equal(t, 1, lggr.calls)
+	})
+}