feat: Introduce lag emergency option for cost-based autoscaler

[Fly-Style]

Summary

Adds a two-tier "critical lag" fast path to the cost-based Kafka/Kinesis supervisor autoscaler, so an operator-configured SLA threshold can trigger a faster scale-up reaction, when enabled.

CostBasedAutoScalerConfig gains an optional criticalLagThreshold (aggregate/global lag across all partitions, not per-partition). When set, CostMetrics#getAggregateLag() is compared against it on every evaluation:

• Tier 1 (≥75% of threshold): the lag-amplification multiplier used in the cost function's lag-recovery-time term maxes out at 6.0 (vs. the default 0.3), and the scale-up candidate search bypasses its usual boundary cap (useTaskCountBoundariesOnScaleUp), letting the argmin search the full candidate range instead of being capped to +2 steps above the current task count.
• Tier 2 (≥95% of threshold): cost minimization is skipped entirely — the autoscaler jumps straight to the maximum valid task count, regardless of configured weights or idle ratio.

criticalLagThreshold is null by default, which leaves existing behavior unchanged. The 75%/95% fractions and the 6.0 multiplier are fixed algorithm constants (WeightedCostFunction.CRITICAL_LAG_TIER1_FRACTION, CRITICAL_LAG_TIER2_FRACTION, CRITICAL_LAG_AMPLIFICATION_MULTIPLIER), not separately configurable.

Meanwhile, default LAG_AMPLIFICATION_MULTIPLIER was set to 0.0 to prioritize better task utilization whenever possible.

Details

Plot shows how the tier-1 fast path (amp=6.0) responds across a range of operating conditions, holding partitionCount=500, aggregateLag=10M (above the 75% tier-1 mark but below the 95% tier-2 mark, given criticalLagThreshold=11.5M), and currentTaskCount=50 fixed:

• X-axis — lagWeight/idleWeight pairs (the two cost-function weights, always summing to 1.0), swept from lag-averse (0.1/0.9) to lag-aggressive (0.9/0.1).
• Y-axis — the resulting optimal task count chosen by the argmin search for each weight pair.
• Lines — one per (idle, rate) combination: color encodes current idle ratio (red=0.1, green=0.25), line style encodes processing rate (solid=50k/s, dashed=80k/s, dotted=100k/s).

The takeaway: even holding lag constant, the tier-1 boost isn't a flat override — task count still climbs smoothly as lagWeight increases, but low-idle/low-rate combinations (red solid) saturate at the task-count ceiling far earlier (~0.6/0.4) than high-idle/high-rate combinations (green dotted, ~0.9/0.1), because those still have slack the idle-cost term can trade off against. This demonstrates tier 1 amplifies urgency rather than short-circuiting the optimization the way tier 2 does.

This PR has:

• been self-reviewed.

[FrankChen021]

Severity	Findings
P0	0
P1	1
P2	0
P3	0
Total	1

Reviewed 6 of 6 changed files.

Found 1 issue: tier-2 emergency lag can still keep the current task count instead of forcing the configured maximum.

---

This is an automated review by Codex GPT-5.5

[FrankChen021]

[P1] Make emergency lag choose the max unconditionally

The emergency path narrows the loop to only the maximum valid task count, but optimalTaskCount and optimalCost are still initialized from the current task count before this block. If the configured weights make the current-count cost lower than the max-count cost, for example lagWeight=0 and idleWeight=1, the loop evaluates the max candidate but never updates optimalTaskCount, so tier-2 emergency lag returns the current count instead of jumping to the max. Set the emergency result to the max candidate unconditionally, or initialize the optimal candidate from startIndex after the emergency bounds are applied.

[FrankChen021]

Severity	Findings
P0	0
P1	0
P2	1
P3	0
Total	1

Reviewed 6 of 6 changed files.

---

This is an automated review by Codex GPT-5.6-Sol

[FrankChen021]

[P2] Use exact total lag for tier checks

collectMetrics() supplies only lagStats.getAvgLag(), while CostMetrics#getAggregateLag() reconstructs the total as avgLag * partitionCount. DefaultLagAggregator computes that average using integer division, so the reconstructed value can undercount by up to partitionCount - 1; for example, total lag 999 across 1000 partitions becomes aggregate lag 0. A valid criticalLagThreshold may therefore be crossed without engaging tier 1 or tier 2. With the updated normal multiplier set to 0, a missed tier also leaves lag unamplified and scale-up boundaries enabled. Carry lagStats.getTotalLag() into CostMetrics and use the exact total for these comparisons.

[FrankChen021]

I have reviewed the updated code for correctness, edge cases, concurrency, and integration risks; no new issues found.

Reviewed 8 of 8 changed files.

---

This is an automated review by Codex GPT-5.6-Sol

[FrankChen021]

Severity	Findings
P0	0
P1	0
P2	1
P3	0
Total	1

Reviewed 9 of 9 changed files.

Found one merge-resolution regression: the documented autoscaler cooldown defaults disagree with the runtime constants.

---

This is an automated review by Codex GPT-5.6-Sol

[FrankChen021]

I have reviewed the updated code for correctness, edge cases, concurrency, lifecycle, compatibility, and integration risks; no new issues found. The prior cooldown-default finding is addressed: runtime defaults now match the documented 15-minute scale-up and 20-minute scale-down values.

Reviewed 9 of 9 changed files.

---

This is an automated review by Codex GPT-5.6-Sol