1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
//! Model Predictive Control (MPC) commit pipeline controller (IMPL-26).
//!
//! Scaffolding-only — this controller is not wired into real commit paths yet.
//! The intent is to pin the API and algebra so downstream beads can replace the
//! heuristic in one place.
//!
//! The dynamics are a simple linearised single-dimensional queue model:
//!
//! ```text
//! x_{k+1} = a * x_k + b * u_k + w_k
//! ```
//!
//! where `x` is observed queue depth (pending commits), `u` is admission rate
//! (commits per tick we are willing to accept), and `w` is an unmodelled
//! disturbance. `a` captures natural drain (e.g. replication, flush), and `b`
//! is the actuator gain — i.e. how strongly admission rate moves the queue.
//!
//! For a one-step horizon LQR with cost `(x_{k+1} - target)^2 + rho * (u - u_prev)^2`
//! the unconstrained optimum is closed-form:
//!
//! ```text
//! u* = (target - a * x) / b
//! ```
//!
//! (Derived from requiring `a*x + b*u = target`; note admission *raises* the
//! queue so u is positive when the queue is below target and negative when
//! above; the `[0, max_rate]` clamp handles the "above target → don't admit,
//! let natural drain `a` do the work" case.)
//!
//! After clamping into `[0, max_rate]` we smooth toward the previous control to
//! avoid chattering:
//!
//! ```text
//! u = u_prev + (1 / (1 + rho)) * (u* - u_prev)
//! ```
//!
//! This is intentionally 20-ish lines of real logic. Replace with a multi-step
//! horizon QP when we start caring about anticipated disturbance profiles.
/// Default control-effort weight. Higher values damp adjustments harder.
const DEFAULT_RHO: f64 = 1.0;
/// Minimum positive actuator gain. Protects against division-by-zero when
/// a caller accidentally plugs in `b = 0`.
const MIN_B: f64 = 1e-9;
/// One-step MPC controller for the commit admission pipeline.
#[derive(Debug, Clone, Copy)]
pub struct MpcCommitController {
/// Desired queue depth (setpoint).
target: f64,
/// State dynamics coefficient. `x_{k+1} = a * x_k + b * u_k`.
/// Values in `[0, 1)` model exponential decay when `u = 0`.
a: f64,
/// Actuator gain — admission-rate to queue-depth sensitivity.
b: f64,
/// Previous admission rate, used for smoothing.
u_prev: f64,
/// Admission-rate cap. The controller never returns more than this.
max_rate: f64,
/// Control-effort weight. Higher `rho` means smaller control steps.
rho: f64,
}
impl MpcCommitController {
/// Build a controller with sensible defaults (`a = 0.9`, `b = 1.0`,
/// `rho = 1.0`, `u_prev = 0.0`).
#[must_use]
pub fn new(target: f64, max_rate: f64) -> Self {
Self::with_params(target, max_rate, 0.9, 1.0, DEFAULT_RHO)
}
/// Build a controller with explicit dynamics and smoothing parameters.
#[must_use]
pub fn with_params(target: f64, max_rate: f64, a: f64, b: f64, rho: f64) -> Self {
let target = if target.is_finite() && target >= 0.0 {
target
} else {
0.0
};
let max_rate = if max_rate.is_finite() && max_rate >= 0.0 {
max_rate
} else {
0.0
};
let a = if a.is_finite() { a } else { 0.9 };
let b = if b.is_finite() && b.abs() > MIN_B {
b
} else {
1.0
};
let rho = if rho.is_finite() && rho >= 0.0 {
rho
} else {
DEFAULT_RHO
};
Self {
target,
a,
b,
u_prev: 0.0,
max_rate,
rho,
}
}
/// Current setpoint.
#[must_use]
pub fn target(&self) -> f64 {
self.target
}
/// Last admission rate this controller emitted.
#[must_use]
pub fn last_rate(&self) -> f64 {
self.u_prev
}
/// Advance the controller by one observation and return the next admission
/// rate. Callers typically interpret this as "commits allowed per tick".
pub fn step(&mut self, measured_queue_depth: f64) -> f64 {
let x = if measured_queue_depth.is_finite() {
measured_queue_depth.max(0.0)
} else {
0.0
};
// Unconstrained one-step LQR optimum: pick u so that the predicted
// next state lands exactly on target, i.e.
// a * x + b * u_star = target
// => u_star = (target - a * x) / b
//
// When the queue is *below* target, `u_star` is positive and we admit
// more to drive the queue up toward the setpoint. When the queue is
// above target the optimum becomes negative and the `[0, max_rate]`
// clamp pins it to zero, letting the natural drain coefficient `a` do
// the work.
let u_star_raw = self.a.mul_add(-x, self.target) / self.b;
let u_star = u_star_raw.clamp(0.0, self.max_rate);
// Smooth toward u_star. The factor 1 / (1 + rho) is the closed-form
// minimiser of the smoothed cost once the unconstrained optimum has
// been picked; `rho = 0` means "jump immediately", larger `rho` means
// slower response.
let alpha = 1.0 / (1.0 + self.rho);
let u = self.u_prev + alpha * (u_star - self.u_prev);
let u = u.clamp(0.0, self.max_rate);
self.u_prev = u;
u
}
}
#[cfg(test)]
mod tests {
use super::*;
/// Simulate the plant `x_{k+1} = a*x_k + b*u_k + w_k` using the controller's
/// own dynamics parameters so the test doesn't have to guess.
fn simulate(
ctrl: &mut MpcCommitController,
x0: f64,
steps: usize,
disturbance: impl Fn(usize) -> f64,
) -> Vec<f64> {
let a = ctrl.a;
let b = ctrl.b;
let mut x = x0;
let mut trace = Vec::with_capacity(steps + 1);
trace.push(x);
for k in 0..steps {
let u = ctrl.step(x);
// Plant update with exogenous disturbance.
x = (b.mul_add(u, a * x) + disturbance(k)).max(0.0);
trace.push(x);
}
trace
}
#[test]
fn mpc_commit_converges_to_target() {
// Start at x=10, target=5, drain a=0.9, gain b=1.0, max_rate=10.
let mut ctrl = MpcCommitController::with_params(5.0, 10.0, 0.9, 1.0, 1.0);
let trace = simulate(&mut ctrl, 10.0, 20, |_| 0.0);
let final_x = *trace.last().expect("non-empty trace");
// After 20 steps we should be comfortably within 10% of target.
assert!(
(final_x - 5.0).abs() < 0.5,
"expected convergence toward 5.0, got {final_x}; trace={trace:?}"
);
// And we should be strictly closer than the starting error.
assert!((final_x - 5.0).abs() < (10.0_f64 - 5.0).abs());
}
#[test]
fn mpc_commit_recovers_from_disturbance() {
let mut ctrl = MpcCommitController::with_params(5.0, 10.0, 0.9, 1.0, 1.0);
// Settle at target.
let _ = simulate(&mut ctrl, 5.0, 40, |_| 0.0);
// Inject a one-shot disturbance of +5 at k=0, then let it decay.
let trace = simulate(&mut ctrl, 5.0, 30, |k| if k == 0 { 5.0 } else { 0.0 });
let final_x = *trace.last().expect("non-empty trace");
// Within 20% of target after recovery window.
assert!(
(final_x - 5.0).abs() < 1.0,
"expected recovery within 20% of target, got {final_x}; trace={trace:?}"
);
}
#[test]
fn mpc_commit_rate_is_clamped() {
let mut ctrl = MpcCommitController::with_params(0.0, 3.0, 0.9, 1.0, 0.0);
// Huge queue with rho=0 would want a massive u; clamp must kick in.
let u = ctrl.step(1_000.0);
assert!((0.0..=3.0).contains(&u), "expected clamp to [0,3], got {u}");
}
#[test]
fn mpc_commit_sanitises_bad_inputs() {
// NaN / infinite / negative inputs all get replaced with safe defaults.
let ctrl = MpcCommitController::with_params(
f64::NAN,
f64::INFINITY,
f64::NAN,
0.0, // below MIN_B -> defaults to 1.0
-1.0,
);
assert_eq!(ctrl.target(), 0.0);
assert_eq!(ctrl.last_rate(), 0.0);
}
}