1use std::{fmt, str::FromStr};
5
6use reifydb_runtime::version_epoch::BUCKET_WIDTH;
7use reifydb_value::value::{Value, duration::Duration, value_type::ValueType};
8
9use crate::{common::CommitVersion, default};
10
11#[derive(Debug, Clone, PartialEq, Eq)]
12pub enum AcceptError {
13 TypeMismatch {
14 expected: Vec<ValueType>,
15 actual: ValueType,
16 },
17
18 InvalidValue(String),
19}
20
21impl fmt::Display for AcceptError {
22 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
23 match self {
24 Self::TypeMismatch {
25 expected,
26 actual,
27 } => {
28 write!(f, "expected one of {:?}, got {:?}", expected, actual)
29 }
30 Self::InvalidValue(reason) => write!(f, "{reason}"),
31 }
32 }
33}
34
35#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
36pub enum ConfigProfile {
37 Production,
38 Testing,
39}
40
41pub fn active_config_profile() -> ConfigProfile {
42 if cfg!(feature = "testing") {
43 ConfigProfile::Testing
44 } else {
45 ConfigProfile::Production
46 }
47}
48
49#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
50pub enum ConfigKey {
51 OracleWindowSize,
52 QueryRowBatchSize,
53 QueryMemoryLimit,
54 RetentionEvictInterval,
55 RetentionEvictBatchSize,
56 RetentionEvictMaxBatchesPerTick,
57 EpochBucketInterval,
58 RetentionStartupGrace,
59 MaxRetentionHorizonFloor,
60 HistoricalGcBatchSize,
61 HistoricalGcInterval,
62 CdcTtlDuration,
63 CdcTtlScanInterval,
64 CdcTtlScanBatchSize,
65 CdcWalAutocheckpoint,
66 CdcCommitBufferBytes,
67 CdcBlockCutBytes,
68 CdcReadBufferBytes,
69 MultiPointBufferShardBytes,
70 MultiRangeBufferShardBytes,
71 OperatorRangeTierBytes,
72 MultiPointBufferShards,
73 MultiRangeBufferShards,
74 MultiFlushInterval,
75 MultiFlushBudgetBytes,
76 MultiWalAutocheckpoint,
77 OperatorResidentBudget,
78 OperatorDirtyBudget,
79 OperatorFlushSlice,
80 OperatorFlushInterval,
81 OperatorWalAutocheckpoint,
82 FlowTick,
83 FlowSampleInterval,
84 FlowWakeCoalesce,
85 FlowLinger,
86 FlowActorBatchSize,
87 FlowCommitterShards,
88 FlowBacklogMemoryLimit,
89 FlowPullBatchBytes,
90 FlowLoadBatchBytes,
91 CdcConsumeWaitTimeout,
92 FlowJoinProbeBlockSize,
93 ThreadsAsync,
94 ThreadsCoordination,
95 ThreadsFlow,
96 ThreadsTask,
97 ThreadsCompute,
98 ThreadsMaintenance,
99 SubscriptionWorkerThreads,
100 MetricsFlushInterval,
101 MetricsSampleInterval,
102 MetricsSnapshotInterval,
103 QueueLeaseReapInterval,
104 QueueLeaseReapBatchSize,
105 QueueRetentionInterval,
106 QueueRetentionBatchSize,
107 SyncInterval,
108 CommitMergeInterval,
109}
110
111impl ConfigKey {
112 pub fn all() -> &'static [Self] {
113 &[
114 Self::OracleWindowSize,
115 Self::QueryRowBatchSize,
116 Self::QueryMemoryLimit,
117 Self::RetentionEvictInterval,
118 Self::RetentionEvictBatchSize,
119 Self::RetentionEvictMaxBatchesPerTick,
120 Self::EpochBucketInterval,
121 Self::RetentionStartupGrace,
122 Self::MaxRetentionHorizonFloor,
123 Self::HistoricalGcBatchSize,
124 Self::HistoricalGcInterval,
125 Self::CdcTtlDuration,
126 Self::CdcTtlScanInterval,
127 Self::CdcTtlScanBatchSize,
128 Self::CdcWalAutocheckpoint,
129 Self::CdcCommitBufferBytes,
130 Self::CdcBlockCutBytes,
131 Self::CdcReadBufferBytes,
132 Self::MultiPointBufferShardBytes,
133 Self::MultiRangeBufferShardBytes,
134 Self::OperatorRangeTierBytes,
135 Self::MultiPointBufferShards,
136 Self::MultiRangeBufferShards,
137 Self::MultiFlushInterval,
138 Self::MultiFlushBudgetBytes,
139 Self::MultiWalAutocheckpoint,
140 Self::OperatorResidentBudget,
141 Self::OperatorDirtyBudget,
142 Self::OperatorFlushSlice,
143 Self::OperatorFlushInterval,
144 Self::OperatorWalAutocheckpoint,
145 Self::FlowTick,
146 Self::FlowSampleInterval,
147 Self::FlowWakeCoalesce,
148 Self::FlowLinger,
149 Self::FlowActorBatchSize,
150 Self::FlowCommitterShards,
151 Self::FlowBacklogMemoryLimit,
152 Self::FlowPullBatchBytes,
153 Self::FlowLoadBatchBytes,
154 Self::CdcConsumeWaitTimeout,
155 Self::FlowJoinProbeBlockSize,
156 Self::ThreadsAsync,
157 Self::ThreadsCoordination,
158 Self::ThreadsFlow,
159 Self::ThreadsTask,
160 Self::ThreadsCompute,
161 Self::ThreadsMaintenance,
162 Self::SubscriptionWorkerThreads,
163 Self::MetricsFlushInterval,
164 Self::MetricsSampleInterval,
165 Self::MetricsSnapshotInterval,
166 Self::QueueLeaseReapInterval,
167 Self::QueueLeaseReapBatchSize,
168 Self::QueueRetentionInterval,
169 Self::QueueRetentionBatchSize,
170 Self::SyncInterval,
171 Self::CommitMergeInterval,
172 ]
173 }
174
175 fn duration_or_none(duration: Option<Duration>) -> Value {
176 match duration {
177 Some(duration) => Value::Duration(duration),
178 None => Value::None {
179 inner: ValueType::Duration,
180 },
181 }
182 }
183
184 pub fn default_value(&self) -> Value {
185 match active_config_profile() {
186 ConfigProfile::Production => self.production_value(),
187 ConfigProfile::Testing => self.testing_value(),
188 }
189 }
190
191 pub fn production_value(&self) -> Value {
192 match self {
193 Self::OracleWindowSize => Value::Uint8(default::query::ORACLE_WINDOW_SIZE),
194 Self::QueryRowBatchSize => Value::Uint2(default::query::ROW_BATCH_SIZE),
195 Self::QueryMemoryLimit => Value::Uint8(default::query::MEMORY_LIMIT.as_bytes()),
196 Self::RetentionEvictInterval => Value::Duration(default::retention::EVICT_INTERVAL),
197 Self::RetentionEvictBatchSize => Value::Uint8(default::retention::EVICT_BATCH_SIZE),
198 Self::RetentionEvictMaxBatchesPerTick => {
199 Value::Uint8(default::retention::EVICT_MAX_BATCHES_PER_TICK)
200 }
201 Self::EpochBucketInterval => Value::Duration(default::retention::EPOCH_BUCKET_INTERVAL),
202 Self::RetentionStartupGrace => Value::Duration(default::retention::STARTUP_GRACE),
203 Self::MaxRetentionHorizonFloor => Value::Duration(default::retention::MAX_HORIZON_FLOOR),
204 Self::HistoricalGcBatchSize => Value::Uint8(default::retention::HISTORICAL_GC_BATCH_SIZE),
205 Self::HistoricalGcInterval => Value::Duration(default::retention::HISTORICAL_GC_INTERVAL),
206 Self::CdcTtlDuration => Self::duration_or_none(default::cdc::TTL),
207 Self::CdcTtlScanInterval => Value::Duration(default::cdc::TTL_SCAN_INTERVAL),
208 Self::CdcTtlScanBatchSize => Value::Uint8(default::cdc::TTL_SCAN_BATCH_SIZE),
209 Self::CdcWalAutocheckpoint => Value::Uint8(default::cdc::WAL_AUTOCHECKPOINT_PAGES),
210 Self::CdcCommitBufferBytes => Value::Uint8(default::cdc::COMMIT_BUFFER.as_bytes()),
211 Self::CdcBlockCutBytes => Value::Uint8(default::cdc::BLOCK_CUT.as_bytes()),
212 Self::CdcReadBufferBytes => Value::Uint8(default::cdc::READ_BUFFER.as_bytes()),
213 Self::MultiPointBufferShardBytes => {
214 Value::Uint8(default::store::MULTI_POINT_BUFFER_SHARD.as_bytes())
215 }
216 Self::MultiRangeBufferShardBytes => {
217 Value::Uint8(default::store::MULTI_RANGE_BUFFER_SHARD.as_bytes())
218 }
219 Self::OperatorRangeTierBytes => Value::Uint8(default::store::OPERATOR_RANGE_TIER.as_bytes()),
220 Self::MultiPointBufferShards => Value::Uint2(default::store::MULTI_POINT_BUFFER_SHARDS),
221 Self::MultiRangeBufferShards => Value::Uint2(default::store::MULTI_RANGE_BUFFER_SHARDS),
222 Self::MultiFlushInterval => Value::Duration(default::store::MULTI_FLUSH_INTERVAL),
223 Self::MultiFlushBudgetBytes => Value::Uint8(default::store::MULTI_FLUSH_BUDGET.as_bytes()),
224 Self::MultiWalAutocheckpoint => Value::Uint8(default::store::MULTI_WAL_AUTOCHECKPOINT_PAGES),
225 Self::OperatorResidentBudget => {
226 Value::Uint8(default::store::OPERATOR_RESIDENT_BUDGET.as_bytes())
227 }
228 Self::OperatorDirtyBudget => Value::Uint8(default::store::OPERATOR_DIRTY_BUDGET.as_bytes()),
229 Self::OperatorFlushSlice => Value::Uint8(default::store::OPERATOR_FLUSH_SLICE.as_bytes()),
230 Self::OperatorFlushInterval => Value::Duration(default::store::OPERATOR_FLUSH_INTERVAL),
231 Self::OperatorWalAutocheckpoint => {
232 Value::Uint8(default::store::OPERATOR_WAL_AUTOCHECKPOINT_PAGES)
233 }
234 Self::FlowTick => Value::Duration(default::flow::TICK),
235 Self::FlowSampleInterval => Value::Duration(default::flow::SAMPLE_INTERVAL),
236 Self::FlowWakeCoalesce => Value::Duration(default::flow::WAKE_COALESCE),
237 Self::FlowLinger => Value::Duration(default::flow::LINGER),
238 Self::FlowActorBatchSize => Value::Uint2(default::flow::ACTOR_BATCH_SIZE),
239 Self::FlowCommitterShards => Value::Uint2(default::flow::COMMITTER_SHARDS),
240 Self::FlowBacklogMemoryLimit => Value::Uint8(default::flow::BACKLOG_MEMORY_LIMIT.as_bytes()),
241 Self::FlowPullBatchBytes => Value::Uint8(default::flow::PULL_BATCH.as_bytes()),
242 Self::FlowLoadBatchBytes => Value::Uint8(default::flow::LOAD_BATCH.as_bytes()),
243 Self::CdcConsumeWaitTimeout => Value::Duration(default::cdc::CONSUME_WAIT_TIMEOUT),
244 Self::FlowJoinProbeBlockSize => Value::Uint8(default::flow::JOIN_PROBE_BLOCK_SIZE),
245 Self::ThreadsAsync => Value::Uint2(default::threads::ASYNC),
246 Self::ThreadsCoordination => Value::Uint2(default::threads::COORDINATION),
247 Self::ThreadsFlow => Value::Uint2(default::threads::FLOW),
248 Self::ThreadsTask => Value::Uint2(default::threads::TASK),
249 Self::ThreadsCompute => Value::Uint2(default::threads::COMPUTE),
250 Self::ThreadsMaintenance => Value::Uint2(default::threads::MAINTENANCE),
251 Self::SubscriptionWorkerThreads => Value::Uint2(default::threads::SUBSCRIPTION_WORKER),
252 Self::MetricsFlushInterval => Value::Duration(default::metrics::FLUSH_INTERVAL),
253 Self::MetricsSampleInterval => Value::Duration(default::metrics::SAMPLE_INTERVAL),
254 Self::MetricsSnapshotInterval => Self::duration_or_none(default::metrics::SNAPSHOT_INTERVAL),
255 Self::QueueLeaseReapInterval => Value::Duration(default::queue::LEASE_REAP_INTERVAL),
256 Self::QueueLeaseReapBatchSize => Value::Uint8(default::queue::LEASE_REAP_BATCH_SIZE),
257 Self::QueueRetentionInterval => Value::Duration(default::queue::RETENTION_INTERVAL),
258 Self::QueueRetentionBatchSize => Value::Uint8(default::queue::RETENTION_BATCH_SIZE),
259 Self::SyncInterval => Value::Duration(default::store::SYNC_INTERVAL),
260 Self::CommitMergeInterval => Value::Duration(default::store::COMMIT_MERGE_INTERVAL),
261 }
262 }
263
264 pub fn testing_value(&self) -> Value {
265 match self {
266 Self::OracleWindowSize => Value::Uint8(default::query::ORACLE_WINDOW_SIZE_TESTING),
267 Self::QueryRowBatchSize => Value::Uint2(default::query::ROW_BATCH_SIZE_TESTING),
268 Self::QueryMemoryLimit => Value::Uint8(default::query::MEMORY_LIMIT_TESTING.as_bytes()),
269 Self::RetentionEvictInterval => Value::Duration(default::retention::EVICT_INTERVAL_TESTING),
270 Self::RetentionEvictBatchSize => Value::Uint8(default::retention::EVICT_BATCH_SIZE_TESTING),
271 Self::RetentionEvictMaxBatchesPerTick => {
272 Value::Uint8(default::retention::EVICT_MAX_BATCHES_PER_TICK_TESTING)
273 }
274 Self::EpochBucketInterval => Value::Duration(default::retention::EPOCH_BUCKET_INTERVAL_TESTING),
275 Self::RetentionStartupGrace => Value::Duration(default::retention::STARTUP_GRACE_TESTING),
276 Self::MaxRetentionHorizonFloor => {
277 Value::Duration(default::retention::MAX_HORIZON_FLOOR_TESTING)
278 }
279 Self::HistoricalGcBatchSize => {
280 Value::Uint8(default::retention::HISTORICAL_GC_BATCH_SIZE_TESTING)
281 }
282 Self::HistoricalGcInterval => {
283 Value::Duration(default::retention::HISTORICAL_GC_INTERVAL_TESTING)
284 }
285 Self::CdcTtlDuration => Self::duration_or_none(default::cdc::TTL_TESTING),
286 Self::CdcTtlScanInterval => Value::Duration(default::cdc::TTL_SCAN_INTERVAL_TESTING),
287 Self::CdcTtlScanBatchSize => Value::Uint8(default::cdc::TTL_SCAN_BATCH_SIZE_TESTING),
288 Self::CdcWalAutocheckpoint => Value::Uint8(default::cdc::WAL_AUTOCHECKPOINT_PAGES_TESTING),
289 Self::CdcCommitBufferBytes => Value::Uint8(default::cdc::COMMIT_BUFFER_TESTING.as_bytes()),
290 Self::CdcBlockCutBytes => Value::Uint8(default::cdc::BLOCK_CUT_TESTING.as_bytes()),
291 Self::CdcReadBufferBytes => Value::Uint8(default::cdc::READ_BUFFER_TESTING.as_bytes()),
292 Self::MultiPointBufferShardBytes => {
293 Value::Uint8(default::store::MULTI_POINT_BUFFER_SHARD_TESTING.as_bytes())
294 }
295 Self::MultiRangeBufferShardBytes => {
296 Value::Uint8(default::store::MULTI_RANGE_BUFFER_SHARD_TESTING.as_bytes())
297 }
298 Self::OperatorRangeTierBytes => {
299 Value::Uint8(default::store::OPERATOR_RANGE_TIER_TESTING.as_bytes())
300 }
301 Self::MultiPointBufferShards => Value::Uint2(default::store::MULTI_POINT_BUFFER_SHARDS_TESTING),
302 Self::MultiRangeBufferShards => Value::Uint2(default::store::MULTI_RANGE_BUFFER_SHARDS_TESTING),
303 Self::MultiFlushInterval => Value::Duration(default::store::MULTI_FLUSH_INTERVAL_TESTING),
304 Self::MultiFlushBudgetBytes => {
305 Value::Uint8(default::store::MULTI_FLUSH_BUDGET_TESTING.as_bytes())
306 }
307 Self::MultiWalAutocheckpoint => {
308 Value::Uint8(default::store::MULTI_WAL_AUTOCHECKPOINT_PAGES_TESTING)
309 }
310 Self::OperatorResidentBudget => {
311 Value::Uint8(default::store::OPERATOR_RESIDENT_BUDGET_TESTING.as_bytes())
312 }
313 Self::OperatorDirtyBudget => {
314 Value::Uint8(default::store::OPERATOR_DIRTY_BUDGET_TESTING.as_bytes())
315 }
316 Self::OperatorFlushSlice => {
317 Value::Uint8(default::store::OPERATOR_FLUSH_SLICE_TESTING.as_bytes())
318 }
319 Self::OperatorFlushInterval => Value::Duration(default::store::OPERATOR_FLUSH_INTERVAL_TESTING),
320 Self::OperatorWalAutocheckpoint => {
321 Value::Uint8(default::store::OPERATOR_WAL_AUTOCHECKPOINT_PAGES_TESTING)
322 }
323 Self::FlowTick => Value::Duration(default::flow::TICK_TESTING),
324 Self::FlowSampleInterval => Value::Duration(default::flow::SAMPLE_INTERVAL_TESTING),
325 Self::FlowWakeCoalesce => Value::Duration(default::flow::WAKE_COALESCE_TESTING),
326 Self::FlowLinger => Value::Duration(default::flow::LINGER_TESTING),
327 Self::FlowActorBatchSize => Value::Uint2(default::flow::ACTOR_BATCH_SIZE_TESTING),
328 Self::FlowCommitterShards => Value::Uint2(default::flow::COMMITTER_SHARDS_TESTING),
329 Self::FlowBacklogMemoryLimit => {
330 Value::Uint8(default::flow::BACKLOG_MEMORY_LIMIT_TESTING.as_bytes())
331 }
332 Self::FlowPullBatchBytes => Value::Uint8(default::flow::PULL_BATCH_TESTING.as_bytes()),
333 Self::FlowLoadBatchBytes => Value::Uint8(default::flow::LOAD_BATCH_TESTING.as_bytes()),
334 Self::CdcConsumeWaitTimeout => Value::Duration(default::cdc::CONSUME_WAIT_TIMEOUT_TESTING),
335 Self::FlowJoinProbeBlockSize => Value::Uint8(default::flow::JOIN_PROBE_BLOCK_SIZE_TESTING),
336 Self::ThreadsAsync => Value::Uint2(default::threads::ASYNC_TESTING),
337 Self::ThreadsCoordination => Value::Uint2(default::threads::COORDINATION_TESTING),
338 Self::ThreadsFlow => Value::Uint2(default::threads::FLOW_TESTING),
339 Self::ThreadsTask => Value::Uint2(default::threads::TASK_TESTING),
340 Self::ThreadsCompute => Value::Uint2(default::threads::COMPUTE_TESTING),
341 Self::ThreadsMaintenance => Value::Uint2(default::threads::MAINTENANCE_TESTING),
342 Self::SubscriptionWorkerThreads => Value::Uint2(default::threads::SUBSCRIPTION_WORKER_TESTING),
343 Self::MetricsFlushInterval => Value::Duration(default::metrics::FLUSH_INTERVAL_TESTING),
344 Self::MetricsSampleInterval => Value::Duration(default::metrics::SAMPLE_INTERVAL_TESTING),
345 Self::MetricsSnapshotInterval => {
346 Self::duration_or_none(default::metrics::SNAPSHOT_INTERVAL_TESTING)
347 }
348 Self::QueueLeaseReapInterval => Value::Duration(default::queue::LEASE_REAP_INTERVAL_TESTING),
349 Self::QueueLeaseReapBatchSize => Value::Uint8(default::queue::LEASE_REAP_BATCH_SIZE_TESTING),
350 Self::QueueRetentionInterval => Value::Duration(default::queue::RETENTION_INTERVAL_TESTING),
351 Self::QueueRetentionBatchSize => Value::Uint8(default::queue::RETENTION_BATCH_SIZE_TESTING),
352 Self::SyncInterval => Value::Duration(default::store::SYNC_INTERVAL_TESTING),
353 Self::CommitMergeInterval => Value::Duration(default::store::COMMIT_MERGE_INTERVAL_TESTING),
354 }
355 }
356
357 pub fn description(&self) -> &'static str {
358 match self {
359 Self::OracleWindowSize => "Number of transactions per conflict-detection window.",
360 Self::QueryRowBatchSize => {
361 "Number of rows produced per batch by query / DML pipeline operators."
362 }
363 Self::QueryMemoryLimit => {
364 "Maximum bytes a single query may buffer in memory across its blocking operators (joins, sort, top k, distinct) and its accumulated result. A query that would exceed this fails with QUERY_006 instead of growing without bound. Read fresh for each query, so changes take effect immediately."
365 }
366 Self::RetentionEvictInterval => {
367 "How often the retention evictor scans objects with a row TTL for expired rows."
368 }
369 Self::RetentionEvictBatchSize => {
370 "Max rows examined (and thus evicted) per transaction during a retention eviction tick."
371 }
372 Self::RetentionEvictMaxBatchesPerTick => {
373 "Upper bound on eviction transactions per retention tick. Caps how long one tick can run when draining a backlog; remaining work resumes on the next tick."
374 }
375 Self::EpochBucketInterval => {
376 "Wall-clock width of one durable version-epoch bucket. The epoch log persists at most one \
377 (bucket, commit version) sample per bucket, and those samples are what let TTLs resolve a \
378 cutoff after a restart. Smaller buckets give finer expiry resolution at the cost of more \
379 persisted samples over the retention horizon."
380 }
381 Self::RetentionStartupGrace => {
382 "How long after startup every retention executor computes cutoffs but deletes nothing. A \
383 process restarted after a long downtime wakes with a large expired backlog; the grace \
384 period plus per-class budgets drain it over many ticks instead of one mass eviction."
385 }
386 Self::MaxRetentionHorizonFloor => {
387 "Lower bound on the retained version-epoch horizon. The horizon is the longest declared \
388 TTL in the catalog, never less than this floor; epoch samples older than the horizon are \
389 pruned. A TTL longer than the horizon could not resolve a cutoff, so it is rejected at \
390 declaration time rather than silently never expiring."
391 }
392 Self::HistoricalGcBatchSize => {
393 "Max historical (key, version) pairs scanned per object per historical GC tick."
394 }
395 Self::HistoricalGcInterval => {
396 "How often the historical-version GC actor sweeps __historical for versions older than the oracle read watermark."
397 }
398 Self::CdcTtlDuration => {
399 "Maximum age of CDC entries before eviction. When unset, CDC is retained forever; \
400 when set, must be > 0 and entries older than this duration are evicted regardless \
401 of consumer state."
402 }
403 Self::CdcTtlScanInterval => {
404 "How often the CDC producer actor scans for and evicts expired CDC entries."
405 }
406 Self::CdcTtlScanBatchSize => {
407 "Max CDC entries deleted per transaction during a CDC TTL eviction tick."
408 }
409 Self::CdcWalAutocheckpoint => {
410 "WAL frame threshold (SQLite wal_autocheckpoint PRAGMA) for the CDC log's SQLite tier. \
411 CDC has no explicit checkpoint of its own, so this is the sole control over how often \
412 cdc.db's WAL is checkpointed into the main file. Higher values checkpoint less often with \
413 a larger WAL; since CDC is written on the commit path, this also bounds how often a commit \
414 pays an inline auto-checkpoint. Read once at boot; changing it requires a restart."
415 }
416 Self::CdcCommitBufferBytes => {
417 "Upper bound on unflushed CDC bytes held in the commit buffer. A writer that would push the \
418 buffer past this stalls until the flusher drains it, so this is the back-pressure point \
419 between the commit path and the persistent tier. Read once at boot."
420 }
421 Self::CdcBlockCutBytes => {
422 "Target size of one CDC block. The commit buffer cuts a block once its pending bytes reach \
423 this, and that block is the unit of flush, of persistent storage, and of read-cache \
424 residency. Larger blocks compress better but coarsen retention, which drops whole blocks. \
425 Read once at boot."
426 }
427 Self::CdcReadBufferBytes => {
428 "Resident byte budget for the CDC read cache of decoded blocks, split evenly across its \
429 shards. None disables the cache outright, so every miss below the commit buffer decodes a \
430 block straight from the persistent tier. Read once at boot."
431 }
432 Self::MultiPointBufferShardBytes => {
433 "Resident byte budget for each shard of the multi-version point cache; total cache memory is \
434 this value times the shard count. None disables the cache outright, so \
435 every point read that misses the commit buffer goes to the persistent tier. Read once at boot; changing it \
436 requires a restart."
437 }
438 Self::MultiRangeBufferShardBytes => {
439 "Resident byte budget for each shard of the multi-version range cache; total cache memory is \
440 this value times the shard count. None disables the cache outright, so \
441 every multi-version range scan goes to the persistent tier. Read once at boot; changing it \
442 requires a restart."
443 }
444 Self::OperatorRangeTierBytes => {
445 "Resident byte budget for one tier of the operator-state range cache. Every cached keyspace \
446 carries its own tier, so total cache memory is this value times the number of cached \
447 keyspaces. None disables the cache outright, so every operator range scan goes to the \
448 persistent tier. Read once at boot; changing it requires a restart."
449 }
450 Self::MultiPointBufferShards => {
451 "Number of lock-striped shards in the multi-version point cache. Each shard carries its own \
452 byte budget, so raising this raises total cache memory proportionally rather \
453 than dividing a fixed pot. Must be >= 1. Read once at boot; changing it \
454 requires a restart."
455 }
456 Self::MultiRangeBufferShards => {
457 "Number of lock-striped shards in the multi-version range cache. Each shard carries its own \
458 byte budget, so raising this raises total cache memory proportionally rather \
459 than dividing a fixed pot. Must be >= 1. Read once at boot; changing it \
460 requires a restart."
461 }
462 Self::MultiFlushInterval => {
463 "How often the persistent-flush actor drains the in-memory commit buffer into the multi \
464 store's SQLite tier. Longer intervals coalesce more writes per flush - a larger WAL - at \
465 the cost of more resident commit-buffer memory and a longer window before data is \
466 materialized in the persistent file. Read once at boot; changing it requires a restart."
467 }
468 Self::MultiFlushBudgetBytes => {
469 "Maximum bytes of buffered entries the persistent-flush class moves from the commit \
470 buffer to the SQLite tier in one slice. Bounds how long a single flush holds the lane, \
471 so a large backlog drains across ticks instead of stalling every other retention class \
472 behind it."
473 }
474 Self::MultiWalAutocheckpoint => {
475 "WAL frame threshold for the multi store's SQLite tier: sets the SQLite \
476 wal_autocheckpoint PRAGMA that governs when SQLite folds the WAL back into the main \
477 database. Higher values checkpoint less often with a larger WAL, reducing checkpoint \
478 I/O; lower values keep the WAL small at the cost of more frequent checkpoints. Read once \
479 at boot; changing it requires a restart."
480 }
481 Self::OperatorResidentBudget => {
482 "Byte ceiling on resident operator state. Eviction returns clean state to this limit \
483 once it is exceeded. Dirty state is never evicted, so a tier holding mostly unwritten \
484 state stays above the limit until a flush turns it clean; OPERATOR_DIRTY_BUDGET bounds \
485 that, and OPERATOR_FLUSH_SLICE sizes the individual commits a flush writes. Read once \
486 at boot; changing it requires a restart."
487 }
488 Self::OperatorDirtyBudget => {
489 "Byte ceiling on unwritten operator state. A flush is triggered once dirty resident state \
490 exceeds this, which bounds both how much memory dirty state may hold and how much one \
491 drain has to write. It defaults to the resident budget, so the flush interval is the \
492 normal trigger and this stays a backstop for a workload that dirties state faster than \
493 the interval anticipates. Read once at boot; changing it requires a restart."
494 }
495 Self::OperatorFlushSlice => {
496 "Byte target for a single operator-state flush transaction. The drain stops taking work at \
497 the first group boundary past this value, so one commit can exceed it by the size of that \
498 group. Larger values mean fewer and longer commits, and every operator-state write blocks \
499 for the length of a commit. Read once at boot; changing it requires a restart."
500 }
501 Self::OperatorFlushInterval => {
502 "How often the operator-state flush actor drains dirty resident state into the operator \
503 store's SQLite tier. Operator state stays resident after a flush and is freed \
504 separately by eviction, so memory pressure alone can leave state unflushed \
505 indefinitely, which holds the durable checkpoint back and with it the CDC pinning \
506 watermark. Read once at boot; changing it requires a restart."
507 }
508 Self::OperatorWalAutocheckpoint => {
509 "WAL frame threshold for the operator store's SQLite tier: sets the SQLite \
510 wal_autocheckpoint PRAGMA that governs when SQLite folds the WAL back into the main \
511 database. Higher values checkpoint less often with a larger WAL, reducing checkpoint \
512 I/O; lower values keep the WAL small at the cost of more frequent checkpoints. Read \
513 once at boot; changing it requires a restart."
514 }
515 Self::FlowTick => "How often each deferred flow actor wakes up to dispatch due flows.",
516 Self::FlowSampleInterval => {
517 "How often each flow actor samples its operators' approximate memory into the \
518 system::metrics::runtime::memory samples (scope operator::N). Runs on the operator's \
519 own thread, off the apply path. When none, operator sampling is disabled entirely; when \
520 set, must be > 0."
521 }
522 Self::FlowWakeCoalesce => {
523 "How long a flow that folds its input waits after one wake before it takes the next, so upstream \
524 steps inside the window merge into one drain. Zero wakes on every step. Read once at boot; \
525 changing it requires a restart."
526 }
527 Self::FlowLinger => {
528 "How long a flow that folds its input waits, from the first change after it went idle, before it \
529 pulls, so changes arriving inside the window are taken as one step and one commit. Zero pulls \
530 at once. Read once at boot; changing it requires a restart."
531 }
532 Self::FlowActorBatchSize => {
533 "How many messages a flow actor handles per turn on its worker thread before yielding to the \
534 other actors pinned to that thread. Smaller values let a woken flow run sooner. Must be >= 1. \
535 Read once at boot; changing it requires a restart."
536 }
537 Self::FlowCommitterShards => {
538 "How many committer actors share the flow commits. Each flow always commits through the same \
539 shard, so its own commits stay in order while different flows commit side by side. Must be \
540 >= 1. Read once at boot; changing it requires a restart."
541 }
542 Self::FlowBacklogMemoryLimit => {
543 "Byte ceiling of the shared in-memory backlog of decoded CDC entries that feeds flow \
544 consumers. Producer-fed at commit granularity; entries below every flow's cursor are \
545 dropped eagerly and the lowest versions are evicted first once the ceiling is exceeded, \
546 at which point a flow that far behind reloads from disk through the catch-up loader. \
547 Because payload rows are shared, the tally is an upper bound of unique memory."
548 }
549 Self::FlowPullBatchBytes => {
550 "Byte budget a flow actor applies per pull from the CDC backlog. A flow that has fallen \
551 behind receives up to this many bytes of decoded changes in one slice, so catch-up is \
552 vectorized instead of per-version."
553 }
554 Self::FlowLoadBatchBytes => {
555 "Byte budget of one catch-up loader read from the CDC log on behalf of flows that are \
556 behind the in-memory backlog. Identical concurrent requests share a single read."
557 }
558 Self::CdcConsumeWaitTimeout => {
559 "Backstop timeout for the CDC consumer's wait for a consume reply from the downstream \
560 consumer. A lost reply would otherwise wedge the poll loop forever; on timeout the \
561 process aborts with a report naming the consumer, the pending work and the batch. \
562 Must be > 0."
563 }
564 Self::FlowJoinProbeBlockSize => {
565 "Number of opposite-side rows a streaming join pulls per block when probing its stored \
566 state. Bounds resident probe memory without dropping matches; smaller trades fewer \
567 resident rows for more scan round-trips."
568 }
569 Self::ThreadsAsync => {
570 "Number of worker threads for the async runtime. Must be >= 1. \
571 Read at boot before the runtime starts; changes require restart."
572 }
573 Self::ThreadsCoordination => {
574 "Number of worker threads for the coordination group (long-lived actors with \
575 tiny high-frequency handlers and periodic background actors); pinned dispatch. \
576 Must be >= 1. Changes require restart."
577 }
578 Self::ThreadsFlow => {
579 "Number of worker threads for the flow group (long-lived heavy-handler actors: \
580 materialized-view flow execution); pinned dispatch. \
581 Must be >= 1. Changes require restart."
582 }
583 Self::ThreadsTask => {
584 "Number of worker threads for the task pool (short-lived work: per-request \
585 actors and one-shot jobs). Must be >= 1. Changes require restart."
586 }
587 Self::ThreadsCompute => {
588 "Number of worker threads for the compute pool (data-parallel work via install(), \
589 never actors). Must be >= 1. Changes require restart."
590 }
591 Self::ThreadsMaintenance => {
592 "Number of worker threads for the maintenance actor pool (lifecycle tasks, operator range \
593 eviction, filter rebuilds). A long slice on one actor holds a thread, so a count of 1 lets \
594 the slowest task delay every other one. Must be >= 1. Changes require restart."
595 }
596 Self::SubscriptionWorkerThreads => {
597 "Number of subscription worker actors that fan out CDC changes to ephemeral \
598 subscriptions in parallel. 0 means auto (size to the system thread pool). Higher values \
599 raise fan-out parallelism for many concurrent subscriptions. Changes require restart."
600 }
601 Self::MetricsFlushInterval => {
602 "How often the metric collector flushes accumulated storage and CDC accounting into the \
603 system::metrics KV store that backs the storage and cdc views. Must be > 0."
604 }
605 Self::MetricsSampleInterval => {
606 "How often the metrics sampler polls every domain, rolls the window and publishes the \
607 system::metrics ::current and ::total caches. Always on; there is no off value, only a \
608 cadence. Must be > 0. Read once at boot; changing it requires a restart."
609 }
610 Self::MetricsSnapshotInterval => {
611 "How often the published ::current reading of every domain is appended to its ::snapshots \
612 series. When none, no snapshot is ever written; when set, must be > 0 and not shorter than \
613 METRICS_SAMPLE_INTERVAL. Read once at boot; changing it requires a restart."
614 }
615 Self::QueueLeaseReapInterval => {
616 "How often the queue reaper scans for leases whose deadline has passed. A dead worker's item cannot be redelivered sooner than this, so it should stay well below any declared lease ttl."
617 }
618 Self::QueueLeaseReapBatchSize => {
619 "Max queue item-state records one reap slice may scan. Bounds the slice on a deep backlog; the scan resumes from its cursor on the next slice."
620 }
621 Self::CommitMergeInterval => {
622 "How often the commit-merge task merges closed commit-buffer maps off the commit path. Commits merge inline only when the mergeable backlog outgrows the commit store's merge backlog."
623 }
624 Self::QueueRetentionInterval => {
625 "How often the queue retention sweeper deletes finished items whose terminal attempt is older than the queue's declared retention.done, and deduplication records past their own ttl."
626 }
627 Self::QueueRetentionBatchSize => {
628 "Max records one queue retention slice may scan across its item and deduplication sweeps. Remaining work drains on the next slice."
629 }
630 Self::SyncInterval => {
631 "How often the commit log is flushed to disk for commits that do not wait for their own flush."
632 }
633 }
634 }
635
636 pub fn requires_restart(&self) -> bool {
637 match self {
638 Self::OracleWindowSize => false,
639 Self::QueryRowBatchSize => false,
640 Self::QueryMemoryLimit => false,
641 Self::RetentionEvictInterval => true,
642 Self::RetentionEvictBatchSize => false,
643 Self::RetentionEvictMaxBatchesPerTick => false,
644 Self::EpochBucketInterval => true,
645 Self::RetentionStartupGrace => false,
646 Self::MaxRetentionHorizonFloor => false,
647 Self::HistoricalGcBatchSize => false,
648 Self::HistoricalGcInterval => true,
649 Self::CdcTtlDuration => false,
650 Self::CdcTtlScanInterval => true,
651 Self::CdcTtlScanBatchSize => false,
652 Self::CdcWalAutocheckpoint => true,
653 Self::CdcCommitBufferBytes => true,
654 Self::CdcBlockCutBytes => true,
655 Self::CdcReadBufferBytes => true,
656 Self::MultiPointBufferShardBytes => true,
657 Self::MultiRangeBufferShardBytes => true,
658 Self::OperatorRangeTierBytes => true,
659 Self::MultiPointBufferShards => true,
660 Self::MultiRangeBufferShards => true,
661 Self::MultiFlushInterval => true,
662 Self::MultiFlushBudgetBytes => false,
663 Self::MultiWalAutocheckpoint => true,
664 Self::OperatorResidentBudget => true,
665 Self::OperatorDirtyBudget => true,
666 Self::OperatorFlushSlice => true,
667 Self::OperatorFlushInterval => true,
668 Self::OperatorWalAutocheckpoint => true,
669 Self::FlowTick => false,
670 Self::FlowSampleInterval => false,
671 Self::FlowWakeCoalesce => true,
672 Self::FlowLinger => true,
673 Self::FlowActorBatchSize => true,
674 Self::FlowCommitterShards => true,
675 Self::FlowBacklogMemoryLimit => true,
676 Self::FlowPullBatchBytes => true,
677 Self::FlowLoadBatchBytes => true,
678 Self::CdcConsumeWaitTimeout => false,
679 Self::FlowJoinProbeBlockSize => false,
680 Self::ThreadsAsync => true,
681 Self::ThreadsCoordination => true,
682 Self::ThreadsFlow => true,
683 Self::ThreadsTask => true,
684 Self::ThreadsCompute => true,
685 Self::ThreadsMaintenance => true,
686 Self::SubscriptionWorkerThreads => true,
687 Self::MetricsFlushInterval => false,
688 Self::MetricsSampleInterval => true,
689 Self::MetricsSnapshotInterval => true,
690 Self::QueueLeaseReapInterval => false,
691 Self::QueueLeaseReapBatchSize => false,
692 Self::QueueRetentionInterval => false,
693 Self::QueueRetentionBatchSize => false,
694 Self::SyncInterval => true,
695 Self::CommitMergeInterval => true,
696 }
697 }
698
699 pub fn expected_types(&self) -> &'static [ValueType] {
700 match self {
701 Self::OracleWindowSize => &[ValueType::Uint8],
702 Self::QueryRowBatchSize => &[ValueType::Uint2],
703 Self::QueryMemoryLimit => &[ValueType::Uint8],
704 Self::RetentionEvictInterval => &[ValueType::Duration],
705 Self::RetentionEvictBatchSize => &[ValueType::Uint8],
706 Self::RetentionEvictMaxBatchesPerTick => &[ValueType::Uint8],
707 Self::EpochBucketInterval => &[ValueType::Duration],
708 Self::RetentionStartupGrace => &[ValueType::Duration],
709 Self::MaxRetentionHorizonFloor => &[ValueType::Duration],
710 Self::HistoricalGcBatchSize => &[ValueType::Uint8],
711 Self::HistoricalGcInterval => &[ValueType::Duration],
712 Self::CdcTtlDuration => &[ValueType::Duration],
713 Self::CdcTtlScanInterval => &[ValueType::Duration],
714 Self::CdcTtlScanBatchSize => &[ValueType::Uint8],
715 Self::CdcWalAutocheckpoint => &[ValueType::Uint8],
716 Self::CdcCommitBufferBytes => &[ValueType::Uint8],
717 Self::CdcBlockCutBytes => &[ValueType::Uint8],
718 Self::CdcReadBufferBytes => &[ValueType::Uint8],
719 Self::MultiPointBufferShardBytes => &[ValueType::Uint8],
720 Self::MultiRangeBufferShardBytes => &[ValueType::Uint8],
721 Self::OperatorRangeTierBytes => &[ValueType::Uint8],
722 Self::MultiPointBufferShards => &[ValueType::Uint2],
723 Self::MultiRangeBufferShards => &[ValueType::Uint2],
724 Self::MultiFlushInterval => &[ValueType::Duration],
725 Self::MultiFlushBudgetBytes => &[ValueType::Uint8],
726 Self::MultiWalAutocheckpoint => &[ValueType::Uint8],
727 Self::OperatorResidentBudget => &[ValueType::Uint8],
728 Self::OperatorDirtyBudget => &[ValueType::Uint8],
729 Self::OperatorFlushSlice => &[ValueType::Uint8],
730 Self::OperatorFlushInterval => &[ValueType::Duration],
731 Self::OperatorWalAutocheckpoint => &[ValueType::Uint8],
732 Self::FlowTick => &[ValueType::Duration],
733 Self::FlowSampleInterval => &[ValueType::Duration],
734 Self::FlowWakeCoalesce => &[ValueType::Duration],
735 Self::FlowLinger => &[ValueType::Duration],
736 Self::FlowActorBatchSize => &[ValueType::Uint2],
737 Self::FlowCommitterShards => &[ValueType::Uint2],
738 Self::FlowBacklogMemoryLimit => &[ValueType::Uint8],
739 Self::FlowPullBatchBytes => &[ValueType::Uint8],
740 Self::FlowLoadBatchBytes => &[ValueType::Uint8],
741 Self::CdcConsumeWaitTimeout => &[ValueType::Duration],
742 Self::FlowJoinProbeBlockSize => &[ValueType::Uint8],
743 Self::ThreadsAsync => &[ValueType::Uint2],
744 Self::ThreadsCoordination => &[ValueType::Uint2],
745 Self::ThreadsFlow => &[ValueType::Uint2],
746 Self::ThreadsTask => &[ValueType::Uint2],
747 Self::ThreadsCompute => &[ValueType::Uint2],
748 Self::ThreadsMaintenance => &[ValueType::Uint2],
749 Self::SubscriptionWorkerThreads => &[ValueType::Uint2],
750 Self::MetricsFlushInterval => &[ValueType::Duration],
751 Self::MetricsSampleInterval => &[ValueType::Duration],
752 Self::MetricsSnapshotInterval => &[ValueType::Duration],
753 Self::QueueLeaseReapInterval => &[ValueType::Duration],
754 Self::QueueLeaseReapBatchSize => &[ValueType::Uint8],
755 Self::QueueRetentionInterval => &[ValueType::Duration],
756 Self::QueueRetentionBatchSize => &[ValueType::Uint8],
757 Self::SyncInterval => &[ValueType::Duration],
758 Self::CommitMergeInterval => &[ValueType::Duration],
759 }
760 }
761
762 pub fn is_optional(&self) -> bool {
763 match self {
764 Self::OracleWindowSize => false,
765 Self::QueryRowBatchSize => false,
766 Self::QueryMemoryLimit => false,
767 Self::RetentionEvictInterval => false,
768 Self::RetentionEvictBatchSize => false,
769 Self::RetentionEvictMaxBatchesPerTick => false,
770 Self::EpochBucketInterval => false,
771 Self::RetentionStartupGrace => false,
772 Self::MaxRetentionHorizonFloor => false,
773 Self::HistoricalGcBatchSize => false,
774 Self::HistoricalGcInterval => false,
775 Self::CdcTtlDuration => true,
776 Self::CdcTtlScanInterval => false,
777 Self::CdcTtlScanBatchSize => false,
778 Self::CdcWalAutocheckpoint => false,
779 Self::CdcCommitBufferBytes => false,
780 Self::CdcBlockCutBytes => false,
781 Self::CdcReadBufferBytes => true,
782 Self::MultiPointBufferShardBytes => true,
783 Self::MultiRangeBufferShardBytes => true,
784 Self::OperatorRangeTierBytes => true,
785 Self::MultiPointBufferShards => false,
786 Self::MultiRangeBufferShards => false,
787 Self::MultiFlushInterval => false,
788 Self::MultiFlushBudgetBytes => false,
789 Self::MultiWalAutocheckpoint => false,
790 Self::OperatorResidentBudget => false,
791 Self::OperatorDirtyBudget => false,
792 Self::OperatorFlushSlice => false,
793 Self::OperatorFlushInterval => false,
794 Self::OperatorWalAutocheckpoint => false,
795 Self::FlowTick => false,
796 Self::FlowSampleInterval => true,
797 Self::FlowWakeCoalesce => false,
798 Self::FlowLinger => false,
799 Self::FlowActorBatchSize => false,
800 Self::FlowCommitterShards => false,
801 Self::FlowBacklogMemoryLimit => false,
802 Self::FlowPullBatchBytes => false,
803 Self::FlowLoadBatchBytes => false,
804 Self::CdcConsumeWaitTimeout => false,
805 Self::FlowJoinProbeBlockSize => false,
806 Self::ThreadsAsync => false,
807 Self::ThreadsCoordination => false,
808 Self::ThreadsFlow => false,
809 Self::ThreadsTask => false,
810 Self::ThreadsCompute => false,
811 Self::ThreadsMaintenance => false,
812 Self::SubscriptionWorkerThreads => false,
813 Self::MetricsFlushInterval => false,
814 Self::MetricsSampleInterval => false,
815 Self::MetricsSnapshotInterval => true,
816 Self::QueueLeaseReapInterval => false,
817 Self::QueueLeaseReapBatchSize => false,
818 Self::QueueRetentionInterval => false,
819 Self::QueueRetentionBatchSize => false,
820 Self::SyncInterval => false,
821 Self::CommitMergeInterval => false,
822 }
823 }
824
825 fn validate_canonical(&self, value: &Value) -> Result<(), String> {
826 match self {
827 Self::CdcTtlDuration => match value {
828 Value::None {
829 ..
830 } => Ok(()),
831 Value::Duration(d) => {
832 if d.is_positive() {
833 Ok(())
834 } else {
835 Err("CDC_TTL_DURATION must be greater than zero".to_string())
836 }
837 }
838 _ => Ok(()),
839 },
840 Self::EpochBucketInterval => match value {
841 Value::Duration(d) if !d.is_positive() => {
842 Err("EPOCH_BUCKET_INTERVAL must be greater than zero".to_string())
843 }
844 Value::Duration(d) if d.to_std().as_secs() < BUCKET_WIDTH.seconds() => Err(format!(
845 "EPOCH_BUCKET_INTERVAL must be at least {}s: the version epoch resolves cutoffs at \
846 second granularity, so a shorter bucket truncates to zero and silently disables \
847 coarse compaction",
848 BUCKET_WIDTH.seconds()
849 )),
850 _ => Ok(()),
851 },
852 Self::RetentionStartupGrace => match value {
853 Value::Duration(d) if d.is_negative() => {
854 Err("RETENTION_STARTUP_GRACE must not be negative".to_string())
855 }
856 _ => Ok(()),
857 },
858 Self::MaxRetentionHorizonFloor => match value {
859 Value::Duration(d) if !d.is_positive() => {
860 Err("MAX_RETENTION_HORIZON_FLOOR must be greater than zero".to_string())
861 }
862 _ => Ok(()),
863 },
864 Self::QueryRowBatchSize => match value {
865 Value::Uint2(0) => Err("QUERY_ROW_BATCH_SIZE must be greater than zero".to_string()),
866 _ => Ok(()),
867 },
868 Self::QueryMemoryLimit => match value {
869 Value::Uint8(0) => Err("QUERY_MEMORY_LIMIT must be greater than zero".to_string()),
870 _ => Ok(()),
871 },
872 Self::FlowBacklogMemoryLimit => match value {
873 Value::Uint8(0) => {
874 Err("FLOW_BACKLOG_MEMORY_LIMIT must be greater than zero".to_string())
875 }
876 _ => Ok(()),
877 },
878 Self::FlowPullBatchBytes => match value {
879 Value::Uint8(0) => Err("FLOW_PULL_BATCH_BYTES must be greater than zero".to_string()),
880 _ => Ok(()),
881 },
882 Self::FlowLoadBatchBytes => match value {
883 Value::Uint8(0) => Err("FLOW_LOAD_BATCH_BYTES must be greater than zero".to_string()),
884 _ => Ok(()),
885 },
886 Self::MultiPointBufferShardBytes => match value {
887 Value::Uint8(0) => Err(
888 "MULTI_POINT_BUFFER_SHARD_BYTES must be greater than zero; use none to disable the point cache"
889 .to_string(),
890 ),
891 _ => Ok(()),
892 },
893 Self::MultiRangeBufferShardBytes => match value {
894 Value::Uint8(0) => Err(
895 "MULTI_RANGE_BUFFER_SHARD_BYTES must be greater than zero; use none to disable the range cache"
896 .to_string(),
897 ),
898 _ => Ok(()),
899 },
900 Self::OperatorRangeTierBytes => match value {
901 Value::Uint8(0) => Err(
902 "OPERATOR_RANGE_TIER_BYTES must be greater than zero; use none to disable the range cache"
903 .to_string(),
904 ),
905 _ => Ok(()),
906 },
907 Self::MultiPointBufferShards => match value {
908 Value::Uint2(0) => Err("MULTI_POINT_BUFFER_SHARDS must be greater than zero".to_string()),
909 _ => Ok(()),
910 },
911 Self::MultiRangeBufferShards => match value {
912 Value::Uint2(0) => Err("MULTI_RANGE_BUFFER_SHARDS must be greater than zero".to_string()),
913 _ => Ok(()),
914 },
915 Self::CdcCommitBufferBytes => match value {
916 Value::Uint8(0) => Err("CDC_COMMIT_BUFFER_BYTES must be greater than zero".to_string()),
917 _ => Ok(()),
918 },
919 Self::CdcBlockCutBytes => match value {
920 Value::Uint8(0) => Err("CDC_BLOCK_CUT_BYTES must be greater than zero".to_string()),
921 _ => Ok(()),
922 },
923 Self::CdcReadBufferBytes => match value {
924 Value::Uint8(0) => Err(
925 "CDC_READ_BUFFER_BYTES must be greater than zero; use none to disable the block cache"
926 .to_string(),
927 ),
928 _ => Ok(()),
929 },
930 Self::MultiFlushInterval => match value {
931 Value::Duration(d) if d.is_positive() => Ok(()),
932 Value::Duration(_) => Err("MULTI_FLUSH_INTERVAL must be greater than zero".to_string()),
933 _ => Ok(()),
934 },
935 Self::CommitMergeInterval => match value {
936 Value::Duration(d) if d.is_positive() => Ok(()),
937 Value::Duration(_) => Err("COMMIT_MERGE_INTERVAL must be greater than zero".to_string()),
938 _ => Ok(()),
939 },
940 Self::MultiFlushBudgetBytes => match value {
941 Value::Uint8(n) if *n > 0 => Ok(()),
942 Value::Uint8(_) => Err("MULTI_FLUSH_BUDGET_BYTES must be greater than zero".to_string()),
943 _ => Ok(()),
944 },
945 Self::MultiWalAutocheckpoint => match value {
946 Value::Uint8(0) => {
947 Err("MULTI_WAL_AUTOCHECKPOINT must be greater than zero".to_string())
948 }
949 _ => Ok(()),
950 },
951 Self::OperatorResidentBudget => match value {
952 Value::Uint8(n) if *n > 0 => Ok(()),
953 Value::Uint8(_) => {
954 Err("OPERATOR_RESIDENT_BUDGET must be greater than zero".to_string())
955 }
956 _ => Ok(()),
957 },
958 Self::OperatorDirtyBudget => match value {
959 Value::Uint8(n) if *n > 0 => Ok(()),
960 Value::Uint8(_) => Err("OPERATOR_DIRTY_BUDGET must be greater than zero".to_string()),
961 _ => Ok(()),
962 },
963 Self::OperatorFlushSlice => match value {
964 Value::Uint8(n) if *n > 0 => Ok(()),
965 Value::Uint8(_) => Err("OPERATOR_FLUSH_SLICE must be greater than zero".to_string()),
966 _ => Ok(()),
967 },
968 Self::OperatorFlushInterval => match value {
969 Value::Duration(d) if d.is_positive() => Ok(()),
970 Value::Duration(_) => Err("OPERATOR_FLUSH_INTERVAL must be greater than zero".to_string()),
971 _ => Ok(()),
972 },
973 Self::OperatorWalAutocheckpoint => match value {
974 Value::Uint8(0) => {
975 Err("OPERATOR_WAL_AUTOCHECKPOINT must be greater than zero".to_string())
976 }
977 _ => Ok(()),
978 },
979 Self::CdcWalAutocheckpoint => match value {
980 Value::Uint8(0) => Err("CDC_WAL_AUTOCHECKPOINT must be greater than zero".to_string()),
981 _ => Ok(()),
982 },
983 Self::HistoricalGcBatchSize => match value {
984 Value::Uint8(0) => {
985 Err("HISTORICAL_GC_BATCH_SIZE must be greater than zero".to_string())
986 }
987 _ => Ok(()),
988 },
989 Self::HistoricalGcInterval => match value {
990 Value::Duration(d) => {
991 if d.is_positive() {
992 Ok(())
993 } else {
994 Err("HISTORICAL_GC_INTERVAL must be greater than zero".to_string())
995 }
996 }
997 _ => Ok(()),
998 },
999 Self::FlowTick => match value {
1000 Value::Duration(d) => {
1001 if d.is_positive() {
1002 Ok(())
1003 } else {
1004 Err("FLOW_TICK must be greater than zero".to_string())
1005 }
1006 }
1007 _ => Ok(()),
1008 },
1009 Self::FlowSampleInterval => match value {
1010 Value::None {
1011 ..
1012 } => Ok(()),
1013 Value::Duration(d) => {
1014 if d.is_positive() {
1015 Ok(())
1016 } else {
1017 Err("FLOW_SAMPLE_INTERVAL must be greater than zero".to_string())
1018 }
1019 }
1020 _ => Ok(()),
1021 },
1022 Self::FlowWakeCoalesce => match value {
1023 Value::Duration(d) if d.is_negative() => Err("FLOW_WAKE_COALESCE must not be negative".to_string()),
1024 _ => Ok(()),
1025 },
1026 Self::FlowLinger => match value {
1027 Value::Duration(d) if d.is_negative() => Err("FLOW_LINGER must not be negative".to_string()),
1028 _ => Ok(()),
1029 },
1030 Self::FlowActorBatchSize => match value {
1031 Value::Uint2(0) => Err("FLOW_ACTOR_BATCH_SIZE must be greater than zero".to_string()),
1032 _ => Ok(()),
1033 },
1034 Self::FlowCommitterShards => match value {
1035 Value::Uint2(0) => Err("FLOW_COMMITTER_SHARDS must be greater than zero".to_string()),
1036 _ => Ok(()),
1037 },
1038 Self::CdcConsumeWaitTimeout => match value {
1039 Value::Duration(d) => {
1040 if d.is_positive() {
1041 Ok(())
1042 } else {
1043 Err("CDC_CONSUME_WAIT_TIMEOUT must be greater than zero".to_string())
1044 }
1045 }
1046 _ => Ok(()),
1047 },
1048 Self::FlowJoinProbeBlockSize => match value {
1049 Value::Uint8(0) => {
1050 Err("FLOW_JOIN_PROBE_BLOCK_SIZE must be greater than zero".to_string())
1051 }
1052 _ => Ok(()),
1053 },
1054 Self::ThreadsAsync => match value {
1055 Value::Uint2(0) => Err("THREADS_ASYNC must be greater than zero".to_string()),
1056 _ => Ok(()),
1057 },
1058 Self::ThreadsCoordination => match value {
1059 Value::Uint2(0) => Err("THREADS_COORDINATION must be greater than zero".to_string()),
1060 _ => Ok(()),
1061 },
1062 Self::ThreadsFlow => match value {
1063 Value::Uint2(0) => Err("THREADS_FLOW must be greater than zero".to_string()),
1064 _ => Ok(()),
1065 },
1066 Self::ThreadsTask => match value {
1067 Value::Uint2(0) => Err("THREADS_TASK must be greater than zero".to_string()),
1068 _ => Ok(()),
1069 },
1070 Self::ThreadsCompute => match value {
1071 Value::Uint2(0) => Err("THREADS_COMPUTE must be greater than zero".to_string()),
1072 _ => Ok(()),
1073 },
1074 Self::ThreadsMaintenance => match value {
1075 Value::Uint2(0) => Err("THREADS_MAINTENANCE must be greater than zero".to_string()),
1076 _ => Ok(()),
1077 },
1078 Self::SubscriptionWorkerThreads => Ok(()),
1079 Self::MetricsFlushInterval => match value {
1080 Value::Duration(d) => {
1081 if d.is_positive() {
1082 Ok(())
1083 } else {
1084 Err("METRICS_FLUSH_INTERVAL must be greater than zero".to_string())
1085 }
1086 }
1087 _ => Ok(()),
1088 },
1089 Self::MetricsSampleInterval => match value {
1090 Value::Duration(d) => {
1091 if d.is_positive() {
1092 Ok(())
1093 } else {
1094 Err("METRICS_SAMPLE_INTERVAL must be greater than zero".to_string())
1095 }
1096 }
1097 _ => Ok(()),
1098 },
1099 Self::MetricsSnapshotInterval => match value {
1100 Value::None {
1101 ..
1102 } => Ok(()),
1103 Value::Duration(d) => {
1104 if d.is_positive() {
1105 Ok(())
1106 } else {
1107 Err("METRICS_SNAPSHOT_INTERVAL must be greater than zero".to_string())
1108 }
1109 }
1110 _ => Ok(()),
1111 },
1112 Self::SyncInterval => match value {
1113 Value::Duration(d) if d.is_positive() => Ok(()),
1114 Value::Duration(_) => Err("SYNC_INTERVAL must be greater than zero".to_string()),
1115 _ => Ok(()),
1116 },
1117 _ => Ok(()),
1118 }
1119 }
1120
1121 pub fn accept(&self, value: Value) -> Result<Value, AcceptError> {
1122 if let Value::None {
1123 inner,
1124 } = &value
1125 {
1126 if self.is_optional() && self.expected_types().contains(inner) {
1127 return Ok(value);
1128 }
1129 return Err(AcceptError::TypeMismatch {
1130 expected: self.expected_types().to_vec(),
1131 actual: value.get_type(),
1132 });
1133 }
1134
1135 if !self.expected_types().contains(&value.get_type()) {
1136 return Err(AcceptError::TypeMismatch {
1137 expected: self.expected_types().to_vec(),
1138 actual: value.get_type(),
1139 });
1140 }
1141
1142 self.validate_canonical(&value).map_err(AcceptError::InvalidValue)?;
1143 Ok(value)
1144 }
1145}
1146
1147impl fmt::Display for ConfigKey {
1148 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1149 match self {
1150 Self::OracleWindowSize => write!(f, "ORACLE_WINDOW_SIZE"),
1151 Self::QueryRowBatchSize => write!(f, "QUERY_ROW_BATCH_SIZE"),
1152 Self::QueryMemoryLimit => write!(f, "QUERY_MEMORY_LIMIT"),
1153 Self::RetentionEvictInterval => write!(f, "RETENTION_EVICT_INTERVAL"),
1154 Self::RetentionEvictBatchSize => write!(f, "RETENTION_EVICT_BATCH_SIZE"),
1155 Self::RetentionEvictMaxBatchesPerTick => write!(f, "RETENTION_EVICT_MAX_BATCHES_PER_TICK"),
1156 Self::EpochBucketInterval => write!(f, "EPOCH_BUCKET_INTERVAL"),
1157 Self::RetentionStartupGrace => write!(f, "RETENTION_STARTUP_GRACE"),
1158 Self::MaxRetentionHorizonFloor => write!(f, "MAX_RETENTION_HORIZON_FLOOR"),
1159 Self::HistoricalGcBatchSize => write!(f, "HISTORICAL_GC_BATCH_SIZE"),
1160 Self::HistoricalGcInterval => write!(f, "HISTORICAL_GC_INTERVAL"),
1161 Self::CdcTtlDuration => write!(f, "CDC_TTL_DURATION"),
1162 Self::CdcTtlScanInterval => write!(f, "CDC_TTL_SCAN_INTERVAL"),
1163 Self::CdcTtlScanBatchSize => write!(f, "CDC_TTL_SCAN_BATCH_SIZE"),
1164 Self::CdcWalAutocheckpoint => write!(f, "CDC_WAL_AUTOCHECKPOINT"),
1165 Self::CdcCommitBufferBytes => write!(f, "CDC_COMMIT_BUFFER_BYTES"),
1166 Self::CdcBlockCutBytes => write!(f, "CDC_BLOCK_CUT_BYTES"),
1167 Self::CdcReadBufferBytes => write!(f, "CDC_READ_BUFFER_BYTES"),
1168 Self::MultiPointBufferShardBytes => write!(f, "MULTI_POINT_BUFFER_SHARD_BYTES"),
1169 Self::MultiRangeBufferShardBytes => write!(f, "MULTI_RANGE_BUFFER_SHARD_BYTES"),
1170 Self::OperatorRangeTierBytes => write!(f, "OPERATOR_RANGE_TIER_BYTES"),
1171 Self::MultiPointBufferShards => write!(f, "MULTI_POINT_BUFFER_SHARDS"),
1172 Self::MultiRangeBufferShards => write!(f, "MULTI_RANGE_BUFFER_SHARDS"),
1173 Self::MultiFlushInterval => write!(f, "MULTI_FLUSH_INTERVAL"),
1174 Self::MultiFlushBudgetBytes => write!(f, "MULTI_FLUSH_BUDGET_BYTES"),
1175 Self::MultiWalAutocheckpoint => write!(f, "MULTI_WAL_AUTOCHECKPOINT"),
1176 Self::OperatorResidentBudget => write!(f, "OPERATOR_RESIDENT_BUDGET"),
1177 Self::OperatorDirtyBudget => write!(f, "OPERATOR_DIRTY_BUDGET"),
1178 Self::OperatorFlushSlice => write!(f, "OPERATOR_FLUSH_SLICE"),
1179 Self::OperatorFlushInterval => write!(f, "OPERATOR_FLUSH_INTERVAL"),
1180 Self::OperatorWalAutocheckpoint => write!(f, "OPERATOR_WAL_AUTOCHECKPOINT"),
1181 Self::FlowTick => write!(f, "FLOW_TICK"),
1182 Self::FlowSampleInterval => write!(f, "FLOW_SAMPLE_INTERVAL"),
1183 Self::FlowWakeCoalesce => write!(f, "FLOW_WAKE_COALESCE"),
1184 Self::FlowLinger => write!(f, "FLOW_LINGER"),
1185 Self::FlowActorBatchSize => write!(f, "FLOW_ACTOR_BATCH_SIZE"),
1186 Self::FlowCommitterShards => write!(f, "FLOW_COMMITTER_SHARDS"),
1187 Self::FlowBacklogMemoryLimit => write!(f, "FLOW_BACKLOG_MEMORY_LIMIT"),
1188 Self::FlowPullBatchBytes => write!(f, "FLOW_PULL_BATCH_BYTES"),
1189 Self::FlowLoadBatchBytes => write!(f, "FLOW_LOAD_BATCH_BYTES"),
1190 Self::CdcConsumeWaitTimeout => write!(f, "CDC_CONSUME_WAIT_TIMEOUT"),
1191 Self::FlowJoinProbeBlockSize => write!(f, "FLOW_JOIN_PROBE_BLOCK_SIZE"),
1192 Self::ThreadsAsync => write!(f, "THREADS_ASYNC"),
1193 Self::ThreadsCoordination => write!(f, "THREADS_COORDINATION"),
1194 Self::ThreadsFlow => write!(f, "THREADS_FLOW"),
1195 Self::ThreadsTask => write!(f, "THREADS_TASK"),
1196 Self::ThreadsCompute => write!(f, "THREADS_COMPUTE"),
1197 Self::ThreadsMaintenance => write!(f, "THREADS_MAINTENANCE"),
1198 Self::SubscriptionWorkerThreads => write!(f, "SUBSCRIPTION_WORKER_THREADS"),
1199 Self::MetricsFlushInterval => write!(f, "METRICS_FLUSH_INTERVAL"),
1200 Self::MetricsSampleInterval => write!(f, "METRICS_SAMPLE_INTERVAL"),
1201 Self::MetricsSnapshotInterval => write!(f, "METRICS_SNAPSHOT_INTERVAL"),
1202 Self::QueueLeaseReapInterval => write!(f, "QUEUE_LEASE_REAP_INTERVAL"),
1203 Self::QueueLeaseReapBatchSize => write!(f, "QUEUE_LEASE_REAP_BATCH_SIZE"),
1204 Self::QueueRetentionInterval => write!(f, "QUEUE_RETENTION_INTERVAL"),
1205 Self::QueueRetentionBatchSize => write!(f, "QUEUE_RETENTION_BATCH_SIZE"),
1206 Self::SyncInterval => write!(f, "SYNC_INTERVAL"),
1207 Self::CommitMergeInterval => write!(f, "COMMIT_MERGE_INTERVAL"),
1208 }
1209 }
1210}
1211
1212impl FromStr for ConfigKey {
1213 type Err = String;
1214
1215 fn from_str(s: &str) -> Result<Self, Self::Err> {
1216 match s {
1217 "ORACLE_WINDOW_SIZE" => Ok(Self::OracleWindowSize),
1218 "QUERY_ROW_BATCH_SIZE" => Ok(Self::QueryRowBatchSize),
1219 "QUERY_MEMORY_LIMIT" => Ok(Self::QueryMemoryLimit),
1220 "RETENTION_EVICT_INTERVAL" => Ok(Self::RetentionEvictInterval),
1221 "RETENTION_EVICT_BATCH_SIZE" => Ok(Self::RetentionEvictBatchSize),
1222 "RETENTION_EVICT_MAX_BATCHES_PER_TICK" => Ok(Self::RetentionEvictMaxBatchesPerTick),
1223 "EPOCH_BUCKET_INTERVAL" => Ok(Self::EpochBucketInterval),
1224 "RETENTION_STARTUP_GRACE" => Ok(Self::RetentionStartupGrace),
1225 "MAX_RETENTION_HORIZON_FLOOR" => Ok(Self::MaxRetentionHorizonFloor),
1226 "HISTORICAL_GC_BATCH_SIZE" => Ok(Self::HistoricalGcBatchSize),
1227 "HISTORICAL_GC_INTERVAL" => Ok(Self::HistoricalGcInterval),
1228 "CDC_TTL_DURATION" => Ok(Self::CdcTtlDuration),
1229 "CDC_TTL_SCAN_INTERVAL" => Ok(Self::CdcTtlScanInterval),
1230 "CDC_TTL_SCAN_BATCH_SIZE" => Ok(Self::CdcTtlScanBatchSize),
1231 "CDC_WAL_AUTOCHECKPOINT" => Ok(Self::CdcWalAutocheckpoint),
1232 "CDC_COMMIT_BUFFER_BYTES" => Ok(Self::CdcCommitBufferBytes),
1233 "CDC_BLOCK_CUT_BYTES" => Ok(Self::CdcBlockCutBytes),
1234 "CDC_READ_BUFFER_BYTES" => Ok(Self::CdcReadBufferBytes),
1235 "MULTI_POINT_BUFFER_SHARD_BYTES" => Ok(Self::MultiPointBufferShardBytes),
1236 "MULTI_RANGE_BUFFER_SHARD_BYTES" => Ok(Self::MultiRangeBufferShardBytes),
1237 "OPERATOR_RANGE_TIER_BYTES" => Ok(Self::OperatorRangeTierBytes),
1238 "MULTI_POINT_BUFFER_SHARDS" => Ok(Self::MultiPointBufferShards),
1239 "MULTI_RANGE_BUFFER_SHARDS" => Ok(Self::MultiRangeBufferShards),
1240 "MULTI_FLUSH_INTERVAL" => Ok(Self::MultiFlushInterval),
1241 "MULTI_FLUSH_BUDGET_BYTES" => Ok(Self::MultiFlushBudgetBytes),
1242 "MULTI_WAL_AUTOCHECKPOINT" => Ok(Self::MultiWalAutocheckpoint),
1243 "OPERATOR_RESIDENT_BUDGET" => Ok(Self::OperatorResidentBudget),
1244 "OPERATOR_DIRTY_BUDGET" => Ok(Self::OperatorDirtyBudget),
1245 "OPERATOR_FLUSH_SLICE" => Ok(Self::OperatorFlushSlice),
1246 "OPERATOR_FLUSH_INTERVAL" => Ok(Self::OperatorFlushInterval),
1247 "OPERATOR_WAL_AUTOCHECKPOINT" => Ok(Self::OperatorWalAutocheckpoint),
1248 "FLOW_TICK" => Ok(Self::FlowTick),
1249 "FLOW_SAMPLE_INTERVAL" => Ok(Self::FlowSampleInterval),
1250 "FLOW_WAKE_COALESCE" => Ok(Self::FlowWakeCoalesce),
1251 "FLOW_LINGER" => Ok(Self::FlowLinger),
1252 "FLOW_ACTOR_BATCH_SIZE" => Ok(Self::FlowActorBatchSize),
1253 "FLOW_COMMITTER_SHARDS" => Ok(Self::FlowCommitterShards),
1254 "FLOW_BACKLOG_MEMORY_LIMIT" => Ok(Self::FlowBacklogMemoryLimit),
1255 "FLOW_PULL_BATCH_BYTES" => Ok(Self::FlowPullBatchBytes),
1256 "FLOW_LOAD_BATCH_BYTES" => Ok(Self::FlowLoadBatchBytes),
1257 "CDC_CONSUME_WAIT_TIMEOUT" => Ok(Self::CdcConsumeWaitTimeout),
1258 "FLOW_JOIN_PROBE_BLOCK_SIZE" => Ok(Self::FlowJoinProbeBlockSize),
1259 "THREADS_ASYNC" => Ok(Self::ThreadsAsync),
1260 "THREADS_COORDINATION" => Ok(Self::ThreadsCoordination),
1261 "THREADS_FLOW" => Ok(Self::ThreadsFlow),
1262 "THREADS_TASK" => Ok(Self::ThreadsTask),
1263 "THREADS_COMPUTE" => Ok(Self::ThreadsCompute),
1264 "THREADS_MAINTENANCE" => Ok(Self::ThreadsMaintenance),
1265 "SUBSCRIPTION_WORKER_THREADS" => Ok(Self::SubscriptionWorkerThreads),
1266 "METRICS_FLUSH_INTERVAL" => Ok(Self::MetricsFlushInterval),
1267 "METRICS_SAMPLE_INTERVAL" => Ok(Self::MetricsSampleInterval),
1268 "METRICS_SNAPSHOT_INTERVAL" => Ok(Self::MetricsSnapshotInterval),
1269 "QUEUE_LEASE_REAP_INTERVAL" => Ok(Self::QueueLeaseReapInterval),
1270 "QUEUE_LEASE_REAP_BATCH_SIZE" => Ok(Self::QueueLeaseReapBatchSize),
1271 "QUEUE_RETENTION_INTERVAL" => Ok(Self::QueueRetentionInterval),
1272 "QUEUE_RETENTION_BATCH_SIZE" => Ok(Self::QueueRetentionBatchSize),
1273 "SYNC_INTERVAL" => Ok(Self::SyncInterval),
1274 "COMMIT_MERGE_INTERVAL" => Ok(Self::CommitMergeInterval),
1275 _ => Err(format!("Unknown system configuration key: {}", s)),
1276 }
1277 }
1278}
1279
1280#[derive(Debug, Clone)]
1281pub struct Config {
1282 pub key: ConfigKey,
1283
1284 pub value: Value,
1285
1286 pub default_value: Value,
1287
1288 pub description: &'static str,
1289
1290 pub requires_restart: bool,
1291}
1292
1293pub trait GetConfig: Send + Sync {
1294 fn get_config(&self, key: ConfigKey) -> Value;
1295
1296 fn get_config_at(&self, key: ConfigKey, version: CommitVersion) -> Value;
1297
1298 fn get_config_uint8(&self, key: ConfigKey) -> u64 {
1299 let val = self.get_config(key);
1300 match val {
1301 Value::Uint8(v) => v,
1302 v => panic!("config key '{}' expected Uint8, got {:?}", key, v),
1303 }
1304 }
1305
1306 fn get_config_uint1(&self, key: ConfigKey) -> u8 {
1307 let val = self.get_config(key);
1308 match val {
1309 Value::Uint1(v) => v,
1310 v => panic!("config key '{}' expected Uint1, got {:?}", key, v),
1311 }
1312 }
1313
1314 fn get_config_uint2(&self, key: ConfigKey) -> u16 {
1315 let val = self.get_config(key);
1316 match val {
1317 Value::Uint2(v) => v,
1318 v => panic!("config key '{}' expected Uint2, got {:?}", key, v),
1319 }
1320 }
1321
1322 fn get_config_duration(&self, key: ConfigKey) -> Duration {
1323 let val = self.get_config(key);
1324 match val {
1325 Value::Duration(v) => v,
1326 v => panic!("config key '{}' expected Duration, got {:?}", key, v),
1327 }
1328 }
1329
1330 fn get_config_duration_opt(&self, key: ConfigKey) -> Option<Duration> {
1331 match self.get_config(key) {
1332 Value::None {
1333 ..
1334 } => None,
1335 Value::Duration(v) => Some(v),
1336 v => panic!("config key '{}' expected Duration or None, got {:?}", key, v),
1337 }
1338 }
1339
1340 fn get_config_uint8_opt(&self, key: ConfigKey) -> Option<u64> {
1341 match self.get_config(key) {
1342 Value::None {
1343 ..
1344 } => None,
1345 Value::Uint8(v) => Some(v),
1346 v => panic!("config key '{}' expected Uint8 or None, got {:?}", key, v),
1347 }
1348 }
1349}
1350
1351#[cfg(test)]
1352mod tests {
1353 use super::*;
1354
1355 #[test]
1356 fn test_cdc_ttl_default_is_typed_null() {
1357 let default = ConfigKey::CdcTtlDuration.default_value();
1359 assert!(matches!(
1360 default,
1361 Value::None {
1362 inner: ValueType::Duration
1363 }
1364 ));
1365 }
1366
1367 #[test]
1368 fn test_cdc_ttl_accept_passes_typed_null() {
1369 let none = Value::None {
1370 inner: ValueType::Duration,
1371 };
1372 let v = ConfigKey::CdcTtlDuration.accept(none.clone()).unwrap();
1373 assert_eq!(v, none);
1374 }
1375
1376 #[test]
1377 fn test_cdc_ttl_accept_passes_positive_duration() {
1378 let one_sec = Value::duration_seconds(1);
1379 assert_eq!(ConfigKey::CdcTtlDuration.accept(one_sec.clone()).unwrap(), one_sec);
1380
1381 let one_hour = Value::duration_seconds(3600);
1382 assert_eq!(ConfigKey::CdcTtlDuration.accept(one_hour.clone()).unwrap(), one_hour);
1383 }
1384
1385 #[test]
1386 fn test_cdc_ttl_accept_rejects_zero() {
1387 let zero = Value::duration_seconds(0);
1388 match ConfigKey::CdcTtlDuration.accept(zero).unwrap_err() {
1389 AcceptError::InvalidValue(reason) => {
1390 assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1391 }
1392 other => panic!("expected InvalidValue, got {other:?}"),
1393 }
1394 }
1395
1396 #[test]
1397 fn test_cdc_ttl_accept_rejects_negative() {
1398 let negative = Value::duration_seconds(-5);
1399 assert!(matches!(ConfigKey::CdcTtlDuration.accept(negative), Err(AcceptError::InvalidValue(_))));
1400 }
1401
1402 #[test]
1403 fn test_other_keys_accept_in_type_values() {
1404 assert!(ConfigKey::OracleWindowSize.accept(Value::Uint8(0)).is_ok());
1405 }
1406
1407 #[test]
1408 fn test_cdc_ttl_round_trips_through_display_and_from_str() {
1409 let key: ConfigKey = "CDC_TTL_DURATION".parse().unwrap();
1410 assert_eq!(key, ConfigKey::CdcTtlDuration);
1411 assert_eq!(format!("{}", ConfigKey::CdcTtlDuration), "CDC_TTL_DURATION");
1412 }
1413
1414 #[test]
1415 fn test_cdc_ttl_in_all() {
1416 assert!(ConfigKey::all().contains(&ConfigKey::CdcTtlDuration));
1417 }
1418
1419 #[test]
1420 fn test_query_memory_limit_defaults_and_round_trips() {
1421 assert_eq!(ConfigKey::QueryMemoryLimit.production_value(), Value::Uint8(1024 * 1024 * 1024));
1422 assert_eq!(ConfigKey::QueryMemoryLimit.expected_types(), &[ValueType::Uint8]);
1423 let key: ConfigKey = "QUERY_MEMORY_LIMIT".parse().unwrap();
1424 assert_eq!(key, ConfigKey::QueryMemoryLimit);
1425 assert_eq!(format!("{}", ConfigKey::QueryMemoryLimit), "QUERY_MEMORY_LIMIT");
1426 }
1427
1428 #[test]
1429 fn test_query_memory_limit_rejects_zero() {
1430 assert!(ConfigKey::QueryMemoryLimit.accept(Value::Uint8(0)).is_err());
1432 assert_eq!(ConfigKey::QueryMemoryLimit.accept(Value::Uint8(1)).unwrap(), Value::Uint8(1));
1433 }
1434
1435 #[test]
1436 fn test_query_memory_limit_requires_restart_and_optional() {
1437 assert!(!ConfigKey::QueryMemoryLimit.requires_restart());
1439 assert!(!ConfigKey::QueryMemoryLimit.is_optional());
1441 }
1442
1443 #[test]
1444 fn test_all_contains_every_compact_key_and_has_expected_len() {
1445 let all = ConfigKey::all();
1446 assert_eq!(all.len(), 58);
1447 assert!(all.contains(&ConfigKey::QueryMemoryLimit));
1448 assert!(all.contains(&ConfigKey::RetentionEvictInterval));
1449 assert!(all.contains(&ConfigKey::RetentionEvictBatchSize));
1450 assert!(all.contains(&ConfigKey::RetentionEvictMaxBatchesPerTick));
1451 assert!(all.contains(&ConfigKey::MultiFlushInterval));
1452 assert!(all.contains(&ConfigKey::MultiWalAutocheckpoint));
1453 assert!(all.contains(&ConfigKey::OperatorResidentBudget));
1454 assert!(all.contains(&ConfigKey::OperatorDirtyBudget));
1455 assert!(all.contains(&ConfigKey::OperatorFlushSlice));
1456 assert!(all.contains(&ConfigKey::OperatorFlushInterval));
1457 assert!(all.contains(&ConfigKey::OperatorWalAutocheckpoint));
1458 assert!(all.contains(&ConfigKey::CdcWalAutocheckpoint));
1459 assert!(all.contains(&ConfigKey::CdcConsumeWaitTimeout));
1460 assert!(all.contains(&ConfigKey::FlowJoinProbeBlockSize));
1461 assert!(all.contains(&ConfigKey::CdcTtlScanInterval));
1462 assert!(all.contains(&ConfigKey::CdcTtlScanBatchSize));
1463 assert!(all.contains(&ConfigKey::MaxRetentionHorizonFloor));
1464 assert!(all.contains(&ConfigKey::FlowLoadBatchBytes));
1465 assert!(all.contains(&ConfigKey::CdcCommitBufferBytes));
1466 assert!(all.contains(&ConfigKey::CdcBlockCutBytes));
1467 assert!(all.contains(&ConfigKey::CdcReadBufferBytes));
1468 assert!(all.contains(&ConfigKey::OperatorRangeTierBytes));
1469 assert!(all.contains(&ConfigKey::OperatorDirtyBudget));
1470 assert!(all.contains(&ConfigKey::OperatorFlushSlice));
1471 assert!(all.contains(&ConfigKey::MultiPointBufferShards));
1472 assert!(all.contains(&ConfigKey::MultiRangeBufferShards));
1473 assert!(all.contains(&ConfigKey::FlowBacklogMemoryLimit));
1474 assert!(all.contains(&ConfigKey::FlowPullBatchBytes));
1475 assert!(all.contains(&ConfigKey::FlowLoadBatchBytes));
1476 assert!(all.contains(&ConfigKey::QueryRowBatchSize));
1477 assert!(all.contains(&ConfigKey::ThreadsAsync));
1478 assert!(all.contains(&ConfigKey::ThreadsCoordination));
1479 assert!(all.contains(&ConfigKey::ThreadsFlow));
1480 assert!(all.contains(&ConfigKey::ThreadsTask));
1481 assert!(all.contains(&ConfigKey::ThreadsCompute));
1482 assert!(all.contains(&ConfigKey::ThreadsMaintenance));
1483 assert!(all.contains(&ConfigKey::MetricsFlushInterval));
1484 assert!(all.contains(&ConfigKey::SubscriptionWorkerThreads));
1485 assert!(all.contains(&ConfigKey::FlowSampleInterval));
1486 assert!(all.contains(&ConfigKey::MetricsSampleInterval));
1487 assert!(all.contains(&ConfigKey::MetricsSnapshotInterval));
1488 assert!(all.contains(&ConfigKey::QueueLeaseReapInterval));
1489 assert!(all.contains(&ConfigKey::QueueLeaseReapBatchSize));
1490 assert!(all.contains(&ConfigKey::QueueRetentionInterval));
1491 assert!(all.contains(&ConfigKey::QueueRetentionBatchSize));
1492 assert!(all.contains(&ConfigKey::SyncInterval));
1493 assert!(all.contains(&ConfigKey::CommitMergeInterval));
1494 }
1495
1496 #[test]
1497 fn test_metrics_sample_interval_is_always_on() {
1498 assert_eq!(ConfigKey::MetricsSampleInterval.default_value(), Value::duration_seconds(10));
1501 assert_eq!(ConfigKey::MetricsSampleInterval.expected_types(), &[ValueType::Duration]);
1502 assert!(!ConfigKey::MetricsSampleInterval.is_optional(), "there is no off value, only a cadence");
1503 assert!(ConfigKey::MetricsSampleInterval.requires_restart(), "read once at boot");
1504
1505 let ten = Value::duration_seconds(10);
1506 assert_eq!(ConfigKey::MetricsSampleInterval.accept(ten.clone()).unwrap(), ten);
1507 let zero = Value::duration_seconds(0);
1508 assert!(matches!(ConfigKey::MetricsSampleInterval.accept(zero), Err(AcceptError::InvalidValue(_))));
1509 }
1510
1511 #[test]
1512 fn test_metrics_snapshot_interval_accepts_none_and_positive_rejects_zero() {
1513 assert_eq!(
1515 ConfigKey::MetricsSnapshotInterval.default_value(),
1516 Value::None {
1517 inner: ValueType::Duration
1518 },
1519 "snapshotting must be opt-in"
1520 );
1521 assert!(ConfigKey::MetricsSnapshotInterval.is_optional(), "none must stay accepted to turn it off");
1522 assert!(ConfigKey::MetricsSnapshotInterval.requires_restart(), "read once at boot");
1523
1524 let none = Value::None {
1525 inner: ValueType::Duration,
1526 };
1527 assert_eq!(ConfigKey::MetricsSnapshotInterval.accept(none.clone()).unwrap(), none);
1528
1529 let minute = Value::duration_seconds(60);
1530 assert_eq!(ConfigKey::MetricsSnapshotInterval.accept(minute.clone()).unwrap(), minute);
1531
1532 let zero = Value::duration_seconds(0);
1533 match ConfigKey::MetricsSnapshotInterval.accept(zero).unwrap_err() {
1534 AcceptError::InvalidValue(reason) => {
1535 assert!(reason.contains("must be greater than zero"), "unexpected reason: {reason}");
1536 }
1537 other => panic!("expected InvalidValue, got {other:?}"),
1538 }
1539 }
1540
1541 #[test]
1542 fn test_metrics_sampler_keys_round_trip() {
1543 for (key, name) in [
1544 (ConfigKey::MetricsSampleInterval, "METRICS_SAMPLE_INTERVAL"),
1545 (ConfigKey::MetricsSnapshotInterval, "METRICS_SNAPSHOT_INTERVAL"),
1546 ] {
1547 assert_eq!(format!("{key}"), name);
1548 assert_eq!(name.parse::<ConfigKey>().unwrap(), key);
1549 }
1550 }
1551
1552 #[test]
1553 fn test_flow_sample_interval_metadata() {
1554 assert_eq!(ConfigKey::FlowSampleInterval.default_value(), Value::duration_seconds(60));
1557 assert_eq!(ConfigKey::FlowSampleInterval.expected_types(), &[ValueType::Duration]);
1558 assert!(ConfigKey::FlowSampleInterval.is_optional());
1559 }
1560
1561 #[test]
1562 fn test_flow_sample_interval_round_trip() {
1563 assert_eq!("FLOW_SAMPLE_INTERVAL".parse::<ConfigKey>().unwrap(), ConfigKey::FlowSampleInterval);
1564 assert_eq!(format!("{}", ConfigKey::FlowSampleInterval), "FLOW_SAMPLE_INTERVAL");
1565 }
1566
1567 #[test]
1568 fn test_flow_sample_interval_accepts_none_and_positive_rejects_zero() {
1569 let none = Value::None {
1570 inner: ValueType::Duration,
1571 };
1572 assert_eq!(
1573 ConfigKey::FlowSampleInterval.accept(none.clone()).unwrap(),
1574 none,
1575 "none must be accepted so sampling can be turned off"
1576 );
1577
1578 let minute = Value::duration_seconds(60);
1579 assert_eq!(ConfigKey::FlowSampleInterval.accept(minute.clone()).unwrap(), minute);
1580
1581 let zero = Value::duration_seconds(0);
1582 assert!(matches!(ConfigKey::FlowSampleInterval.accept(zero), Err(AcceptError::InvalidValue(_))));
1583 }
1584
1585 #[test]
1586 fn test_flow_latency_knobs_default_to_the_behaviour_they_replaced() {
1587 assert_eq!(ConfigKey::FlowWakeCoalesce.default_value(), Value::duration_milliseconds(20));
1589 assert_eq!(ConfigKey::FlowActorBatchSize.default_value(), Value::Uint2(8));
1590 assert_eq!(ConfigKey::FlowCommitterShards.default_value(), Value::Uint2(1));
1591 assert_eq!(ConfigKey::FlowWakeCoalesce.expected_types(), &[ValueType::Duration]);
1592 assert_eq!(ConfigKey::FlowActorBatchSize.expected_types(), &[ValueType::Uint2]);
1593 assert_eq!(ConfigKey::FlowCommitterShards.expected_types(), &[ValueType::Uint2]);
1594 for key in [ConfigKey::FlowWakeCoalesce, ConfigKey::FlowActorBatchSize, ConfigKey::FlowCommitterShards]
1595 {
1596 assert!(key.requires_restart(), "{key} is read once when the flow subsystem starts");
1597 assert!(!key.is_optional(), "{key} has no off state");
1598 }
1599 }
1600
1601 #[test]
1602 fn test_flow_latency_knobs_round_trip_their_names() {
1603 for (name, key) in [
1604 ("FLOW_WAKE_COALESCE", ConfigKey::FlowWakeCoalesce),
1605 ("FLOW_ACTOR_BATCH_SIZE", ConfigKey::FlowActorBatchSize),
1606 ("FLOW_COMMITTER_SHARDS", ConfigKey::FlowCommitterShards),
1607 ] {
1608 assert_eq!(name.parse::<ConfigKey>().unwrap(), key);
1609 assert_eq!(format!("{key}"), name);
1610 }
1611 }
1612
1613 #[test]
1614 fn test_flow_latency_knobs_reject_values_that_stall_or_break_the_flows() {
1615 let zero = Value::duration_seconds(0);
1617 assert_eq!(
1618 ConfigKey::FlowWakeCoalesce.accept(zero.clone()).unwrap(),
1619 zero,
1620 "zero turns coalescing off"
1621 );
1622 assert!(matches!(
1623 ConfigKey::FlowWakeCoalesce.accept(Value::duration_milliseconds(-1)),
1624 Err(AcceptError::InvalidValue(_))
1625 ));
1626 assert!(matches!(
1627 ConfigKey::FlowActorBatchSize.accept(Value::Uint2(0)),
1628 Err(AcceptError::InvalidValue(_))
1629 ));
1630 assert!(matches!(
1631 ConfigKey::FlowCommitterShards.accept(Value::Uint2(0)),
1632 Err(AcceptError::InvalidValue(_))
1633 ));
1634 assert_eq!(ConfigKey::FlowActorBatchSize.accept(Value::Uint2(1)).unwrap(), Value::Uint2(1));
1635 assert_eq!(ConfigKey::FlowCommitterShards.accept(Value::Uint2(4)).unwrap(), Value::Uint2(4));
1636 }
1637
1638 #[test]
1639 fn test_metrics_flush_interval_metadata() {
1640 assert_eq!(ConfigKey::MetricsFlushInterval.default_value(), Value::duration_seconds(10));
1642 assert_eq!(ConfigKey::MetricsFlushInterval.expected_types(), &[ValueType::Duration]);
1643 assert!(!ConfigKey::MetricsFlushInterval.is_optional());
1644 assert!(!ConfigKey::MetricsFlushInterval.requires_restart());
1645 }
1646
1647 #[test]
1648 fn test_metrics_flush_interval_round_trip() {
1649 assert_eq!("METRICS_FLUSH_INTERVAL".parse::<ConfigKey>().unwrap(), ConfigKey::MetricsFlushInterval);
1650 assert_eq!(format!("{}", ConfigKey::MetricsFlushInterval), "METRICS_FLUSH_INTERVAL");
1651 }
1652
1653 #[test]
1654 fn test_metrics_flush_interval_accepts_positive_rejects_zero() {
1655 let ten = Value::duration_seconds(10);
1656 assert_eq!(ConfigKey::MetricsFlushInterval.accept(ten.clone()).unwrap(), ten);
1657
1658 let zero = Value::duration_seconds(0);
1659 assert!(matches!(ConfigKey::MetricsFlushInterval.accept(zero), Err(AcceptError::InvalidValue(_))));
1660 }
1661
1662 #[test]
1663 fn test_threads_keys_round_trip() {
1664 assert_eq!("THREADS_ASYNC".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsAsync);
1665 assert_eq!("THREADS_COORDINATION".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsCoordination);
1666 assert_eq!("THREADS_FLOW".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsFlow);
1667 assert_eq!("THREADS_TASK".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsTask);
1668 assert_eq!("THREADS_COMPUTE".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsCompute);
1669 assert_eq!("THREADS_MAINTENANCE".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsMaintenance);
1670 assert_eq!(format!("{}", ConfigKey::ThreadsAsync), "THREADS_ASYNC");
1671 assert_eq!(format!("{}", ConfigKey::ThreadsCoordination), "THREADS_COORDINATION");
1672 assert_eq!(format!("{}", ConfigKey::ThreadsFlow), "THREADS_FLOW");
1673 assert_eq!(format!("{}", ConfigKey::ThreadsTask), "THREADS_TASK");
1674 assert_eq!(format!("{}", ConfigKey::ThreadsCompute), "THREADS_COMPUTE");
1675 assert_eq!(format!("{}", ConfigKey::ThreadsMaintenance), "THREADS_MAINTENANCE");
1676 }
1677
1678 #[test]
1679 fn test_threads_defaults() {
1680 assert_eq!(ConfigKey::ThreadsAsync.production_value(), Value::Uint2(1));
1681 assert_eq!(ConfigKey::ThreadsCoordination.production_value(), Value::Uint2(2));
1682 assert_eq!(ConfigKey::ThreadsFlow.production_value(), Value::Uint2(2));
1683 assert_eq!(ConfigKey::ThreadsTask.production_value(), Value::Uint2(2));
1684 assert_eq!(ConfigKey::ThreadsCompute.production_value(), Value::Uint2(2));
1685 assert_eq!(ConfigKey::ThreadsMaintenance.production_value(), Value::Uint2(1));
1686 }
1687
1688 #[test]
1689 fn test_threads_reject_zero() {
1690 for key in [
1691 ConfigKey::ThreadsAsync,
1692 ConfigKey::ThreadsCoordination,
1693 ConfigKey::ThreadsFlow,
1694 ConfigKey::ThreadsTask,
1695 ConfigKey::ThreadsCompute,
1696 ConfigKey::ThreadsMaintenance,
1697 ] {
1698 match key.accept(Value::Uint2(0)).unwrap_err() {
1699 AcceptError::InvalidValue(reason) => {
1700 assert!(
1701 reason.contains("greater than zero"),
1702 "{key}: unexpected reason: {reason}"
1703 );
1704 }
1705 other => panic!("{key}: expected InvalidValue, got {other:?}"),
1706 }
1707 }
1708 }
1709
1710 #[test]
1711 fn test_threads_accept_positive() {
1712 assert_eq!(ConfigKey::ThreadsAsync.accept(Value::Uint2(4)).unwrap(), Value::Uint2(4));
1713 assert_eq!(ConfigKey::ThreadsCoordination.accept(Value::Uint2(8)).unwrap(), Value::Uint2(8));
1714 assert_eq!(ConfigKey::ThreadsFlow.accept(Value::Uint2(16)).unwrap(), Value::Uint2(16));
1715 assert_eq!(ConfigKey::ThreadsTask.accept(Value::Uint2(4)).unwrap(), Value::Uint2(4));
1716 assert_eq!(ConfigKey::ThreadsCompute.accept(Value::Uint2(2)).unwrap(), Value::Uint2(2));
1717 assert_eq!(ConfigKey::ThreadsMaintenance.accept(Value::Uint2(4)).unwrap(), Value::Uint2(4));
1718 }
1719
1720 #[test]
1721 fn test_threads_reject_int4_for_uint2_key() {
1722 assert!(matches!(ConfigKey::ThreadsTask.accept(Value::Int4(8)), Err(AcceptError::TypeMismatch { .. })));
1724 }
1725
1726 #[test]
1727 fn test_threads_require_restart() {
1728 assert!(ConfigKey::ThreadsAsync.requires_restart());
1729 assert!(ConfigKey::ThreadsCoordination.requires_restart());
1730 assert!(ConfigKey::ThreadsFlow.requires_restart());
1731 assert!(ConfigKey::ThreadsTask.requires_restart());
1732 assert!(ConfigKey::ThreadsCompute.requires_restart());
1733 assert!(ConfigKey::ThreadsMaintenance.requires_restart());
1734 }
1735
1736 #[test]
1737 fn test_query_row_batch_size_default_is_uint2_128() {
1738 assert_eq!(ConfigKey::QueryRowBatchSize.production_value(), Value::Uint2(128));
1739 }
1740
1741 #[test]
1742 fn test_query_row_batch_size_round_trips_through_display_and_from_str() {
1743 let key: ConfigKey = "QUERY_ROW_BATCH_SIZE".parse().unwrap();
1744 assert_eq!(key, ConfigKey::QueryRowBatchSize);
1745 assert_eq!(format!("{}", ConfigKey::QueryRowBatchSize), "QUERY_ROW_BATCH_SIZE");
1746 }
1747
1748 #[test]
1749 fn test_query_row_batch_size_accept_rejects_zero() {
1750 match ConfigKey::QueryRowBatchSize.accept(Value::Uint2(0)).unwrap_err() {
1751 AcceptError::InvalidValue(reason) => {
1752 assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1753 }
1754 other => panic!("expected InvalidValue, got {other:?}"),
1755 }
1756 }
1757
1758 #[test]
1759 fn test_query_row_batch_size_accept_passes_positive() {
1760 assert_eq!(ConfigKey::QueryRowBatchSize.accept(Value::Uint2(1)).unwrap(), Value::Uint2(1));
1761 assert_eq!(ConfigKey::QueryRowBatchSize.accept(Value::Uint2(1024)).unwrap(), Value::Uint2(1024));
1762 }
1763
1764 #[test]
1765 fn test_query_row_batch_size_rejects_mismatched_type() {
1766 assert!(matches!(
1768 ConfigKey::QueryRowBatchSize.accept(Value::Int4(64)),
1769 Err(AcceptError::TypeMismatch { .. })
1770 ));
1771 assert!(matches!(
1772 ConfigKey::QueryRowBatchSize.accept(Value::Int4(0)),
1773 Err(AcceptError::TypeMismatch { .. })
1774 ));
1775 }
1776
1777 #[test]
1778 fn test_accept_rejects_int4_for_uint8_key() {
1779 assert!(matches!(
1781 ConfigKey::FlowLoadBatchBytes.accept(Value::Int4(1024)),
1782 Err(AcceptError::TypeMismatch { .. })
1783 ));
1784 assert!(matches!(
1785 ConfigKey::FlowLoadBatchBytes.accept(Value::Int8(2048)),
1786 Err(AcceptError::TypeMismatch { .. })
1787 ));
1788 }
1789
1790 #[test]
1791 fn test_accept_rejects_zero_of_canonical_type() {
1792 match ConfigKey::FlowLoadBatchBytes.accept(Value::Uint8(0)).unwrap_err() {
1793 AcceptError::InvalidValue(reason) => {
1794 assert!(reason.contains("greater than zero"));
1795 }
1796 other => panic!("expected InvalidValue, got {other:?}"),
1797 }
1798 }
1799
1800 #[test]
1801 fn test_accept_rejects_negative_int_for_uint8_key() {
1802 assert!(matches!(
1805 ConfigKey::FlowLoadBatchBytes.accept(Value::Int4(-1)),
1806 Err(AcceptError::TypeMismatch { .. })
1807 ));
1808 }
1809
1810 #[test]
1811 fn test_accept_rejects_int_for_duration_key() {
1812 assert!(matches!(
1815 ConfigKey::MaxRetentionHorizonFloor.accept(Value::Int4(60)),
1816 Err(AcceptError::TypeMismatch { .. })
1817 ));
1818 }
1819
1820 #[test]
1821 fn test_accept_idempotent_on_canonical_uint8() {
1822 let canonical = Value::Uint8(42);
1823 assert_eq!(ConfigKey::OracleWindowSize.accept(canonical.clone()).unwrap(), canonical);
1824 }
1825
1826 #[test]
1827 fn test_accept_idempotent_on_canonical_duration() {
1828 let canonical = Value::duration_seconds(5);
1829 assert_eq!(ConfigKey::MaxRetentionHorizonFloor.accept(canonical.clone()).unwrap(), canonical);
1830 }
1831
1832 #[test]
1833 fn test_accept_rejects_typed_null_for_non_optional_key() {
1834 let err = ConfigKey::FlowLoadBatchBytes
1835 .accept(Value::None {
1836 inner: ValueType::Uint8,
1837 })
1838 .unwrap_err();
1839 assert!(matches!(err, AcceptError::TypeMismatch { .. }));
1840 }
1841
1842 #[test]
1843 fn test_accept_passes_typed_null_for_optional_key() {
1844 let none = Value::None {
1845 inner: ValueType::Duration,
1846 };
1847 assert_eq!(ConfigKey::CdcTtlDuration.accept(none.clone()).unwrap(), none);
1848 }
1849
1850 #[test]
1851 fn test_accept_rejects_wrong_inner_type_typed_null_for_optional_key() {
1852 let err = ConfigKey::CdcTtlDuration
1854 .accept(Value::None {
1855 inner: ValueType::Uint8,
1856 })
1857 .unwrap_err();
1858 assert!(matches!(err, AcceptError::TypeMismatch { .. }));
1859 }
1860
1861 #[test]
1862 fn test_historical_gc_keys_round_trip() {
1863 assert_eq!("HISTORICAL_GC_BATCH_SIZE".parse::<ConfigKey>().unwrap(), ConfigKey::HistoricalGcBatchSize);
1864 assert_eq!("HISTORICAL_GC_INTERVAL".parse::<ConfigKey>().unwrap(), ConfigKey::HistoricalGcInterval);
1865 assert_eq!(format!("{}", ConfigKey::HistoricalGcBatchSize), "HISTORICAL_GC_BATCH_SIZE");
1866 assert_eq!(format!("{}", ConfigKey::HistoricalGcInterval), "HISTORICAL_GC_INTERVAL");
1867 }
1868
1869 #[test]
1870 fn test_historical_gc_defaults() {
1871 assert_eq!(ConfigKey::HistoricalGcBatchSize.production_value(), Value::Uint8(50_000));
1872 assert!(matches!(ConfigKey::HistoricalGcInterval.production_value(), Value::Duration(_)));
1873 }
1874
1875 #[test]
1876 fn test_historical_gc_batch_size_rejects_zero() {
1877 match ConfigKey::HistoricalGcBatchSize.accept(Value::Uint8(0)).unwrap_err() {
1878 AcceptError::InvalidValue(reason) => {
1879 assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1880 }
1881 other => panic!("expected InvalidValue, got {other:?}"),
1882 }
1883 }
1884
1885 #[test]
1886 fn test_historical_gc_interval_rejects_zero() {
1887 let zero = Value::duration_seconds(0);
1888 match ConfigKey::HistoricalGcInterval.accept(zero).unwrap_err() {
1889 AcceptError::InvalidValue(reason) => {
1890 assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1891 }
1892 other => panic!("expected InvalidValue, got {other:?}"),
1893 }
1894 }
1895
1896 #[test]
1897 fn test_operator_flush_budget_bytes_metadata() {
1898 assert_eq!(ConfigKey::OperatorResidentBudget.production_value(), Value::Uint8(128 * 1024 * 1024));
1899 assert_eq!(ConfigKey::OperatorResidentBudget.expected_types(), &[ValueType::Uint8]);
1900 assert!(!ConfigKey::OperatorResidentBudget.is_optional());
1901 assert!(
1902 ConfigKey::OperatorResidentBudget.requires_restart(),
1903 "the budget sizes a MemoryBudget built once with the commit tier; declaring it live would \
1904 promise a rewrite that no running store can adopt"
1905 );
1906 }
1907
1908 #[test]
1909 fn test_operator_flush_budget_bytes_rejects_zero() {
1910 match ConfigKey::OperatorResidentBudget.accept(Value::Uint8(0)).unwrap_err() {
1913 AcceptError::InvalidValue(reason) => {
1914 assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1915 }
1916 other => panic!("expected InvalidValue, got {other:?}"),
1917 }
1918 assert_eq!(ConfigKey::OperatorResidentBudget.accept(Value::Uint8(1)).unwrap(), Value::Uint8(1));
1919 }
1920
1921 #[test]
1922 fn test_operator_flush_budget_bytes_round_trips_through_display_and_from_str() {
1923 assert_eq!("OPERATOR_RESIDENT_BUDGET".parse::<ConfigKey>().unwrap(), ConfigKey::OperatorResidentBudget);
1924 assert_eq!(format!("{}", ConfigKey::OperatorResidentBudget), "OPERATOR_RESIDENT_BUDGET");
1925 }
1926
1927 #[test]
1928 fn test_operator_wal_autocheckpoint_metadata() {
1929 assert_eq!(ConfigKey::OperatorWalAutocheckpoint.production_value(), Value::Uint8(1000000));
1930 assert_eq!(ConfigKey::OperatorWalAutocheckpoint.expected_types(), &[ValueType::Uint8]);
1931 assert!(!ConfigKey::OperatorWalAutocheckpoint.is_optional());
1932 assert!(ConfigKey::OperatorWalAutocheckpoint.requires_restart());
1933 }
1934
1935 #[test]
1936 fn test_operator_wal_autocheckpoint_rejects_zero() {
1937 match ConfigKey::OperatorWalAutocheckpoint.accept(Value::Uint8(0)).unwrap_err() {
1940 AcceptError::InvalidValue(reason) => {
1941 assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1942 }
1943 other => panic!("expected InvalidValue, got {other:?}"),
1944 }
1945 assert_eq!(ConfigKey::OperatorWalAutocheckpoint.accept(Value::Uint8(1)).unwrap(), Value::Uint8(1));
1946 }
1947
1948 #[test]
1949 fn test_operator_wal_autocheckpoint_round_trips_through_display_and_from_str() {
1950 assert_eq!(
1951 "OPERATOR_WAL_AUTOCHECKPOINT".parse::<ConfigKey>().unwrap(),
1952 ConfigKey::OperatorWalAutocheckpoint
1953 );
1954 assert_eq!(format!("{}", ConfigKey::OperatorWalAutocheckpoint), "OPERATOR_WAL_AUTOCHECKPOINT");
1955 }
1956}