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