reifydb_core/interface/catalog/
config.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{fmt, str::FromStr};
5
6use reifydb_value::value::{
7	Value, decimal::Decimal, duration::Duration, int::Int, ordered_f32::OrderedF32, ordered_f64::OrderedF64,
8	uint::Uint, value_type::ValueType,
9};
10
11use crate::common::CommitVersion;
12
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub enum AcceptError {
15	TypeMismatch {
16		expected: Vec<ValueType>,
17		actual: ValueType,
18	},
19
20	InvalidValue(String),
21}
22
23impl fmt::Display for AcceptError {
24	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
25		match self {
26			Self::TypeMismatch {
27				expected,
28				actual,
29			} => {
30				write!(f, "expected one of {:?}, got {:?}", expected, actual)
31			}
32			Self::InvalidValue(reason) => write!(f, "{reason}"),
33		}
34	}
35}
36
37#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
38pub enum ConfigKey {
39	OracleWindowSize,
40	OracleWaterMark,
41	QueryRowBatchSize,
42	RowTtlScanBatchSize,
43	RowTtlScanInterval,
44	OperatorTtlScanBatchSize,
45	OperatorTtlScanInterval,
46	VersionEpochSampleInterval,
47	HistoricalGcBatchSize,
48	HistoricalGcInterval,
49	CdcTtlDuration,
50	CdcTtlScanInterval,
51	CdcTtlScanBatchSize,
52	CdcTtlScanMaxBatchesPerTick,
53	CdcTtlReclaimInterval,
54	CdcCompactInterval,
55	CdcCompactBlockSize,
56	CdcCompactSafetyLag,
57	CdcCompactMaxBlocksPerTick,
58	CdcCompactBlockCacheCapacity,
59	CdcCompactZstdLevel,
60	CdcRecentCacheCapacity,
61	MultiReadBufferPages,
62	MultiReadBufferPageSize,
63	MultiReclaimInterval,
64	FlowTick,
65	CdcWatermarkWaitTimeout,
66	CdcConsumeWaitTimeout,
67	FlowJoinProbeBlockSize,
68	ThreadsAsync,
69	ThreadsSystem,
70	ThreadsQuery,
71	ThreadsCommit,
72	ThreadsBackground,
73	FlowWorkerThreads,
74	SubscriptionWorkerThreads,
75	RuntimeMetricsInterval,
76	MetricFlushInterval,
77	MetricsRuntimeRetention,
78	MetricsProfilerRetention,
79	MetricsProfilerSnapshotInterval,
80}
81
82impl ConfigKey {
83	pub fn all() -> &'static [Self] {
84		&[
85			Self::OracleWindowSize,
86			Self::OracleWaterMark,
87			Self::QueryRowBatchSize,
88			Self::RowTtlScanBatchSize,
89			Self::RowTtlScanInterval,
90			Self::OperatorTtlScanBatchSize,
91			Self::OperatorTtlScanInterval,
92			Self::VersionEpochSampleInterval,
93			Self::HistoricalGcBatchSize,
94			Self::HistoricalGcInterval,
95			Self::CdcTtlDuration,
96			Self::CdcTtlScanInterval,
97			Self::CdcTtlScanBatchSize,
98			Self::CdcTtlScanMaxBatchesPerTick,
99			Self::CdcTtlReclaimInterval,
100			Self::CdcCompactInterval,
101			Self::CdcCompactBlockSize,
102			Self::CdcCompactSafetyLag,
103			Self::CdcCompactMaxBlocksPerTick,
104			Self::CdcCompactBlockCacheCapacity,
105			Self::CdcCompactZstdLevel,
106			Self::CdcRecentCacheCapacity,
107			Self::MultiReadBufferPages,
108			Self::MultiReadBufferPageSize,
109			Self::MultiReclaimInterval,
110			Self::FlowTick,
111			Self::CdcWatermarkWaitTimeout,
112			Self::CdcConsumeWaitTimeout,
113			Self::FlowJoinProbeBlockSize,
114			Self::ThreadsAsync,
115			Self::ThreadsSystem,
116			Self::ThreadsQuery,
117			Self::ThreadsCommit,
118			Self::ThreadsBackground,
119			Self::FlowWorkerThreads,
120			Self::SubscriptionWorkerThreads,
121			Self::RuntimeMetricsInterval,
122			Self::MetricFlushInterval,
123			Self::MetricsRuntimeRetention,
124			Self::MetricsProfilerRetention,
125			Self::MetricsProfilerSnapshotInterval,
126		]
127	}
128
129	pub fn default_value(&self) -> Value {
130		match self {
131			Self::OracleWindowSize => Value::Uint8(500),
132			Self::OracleWaterMark => Value::Uint8(20),
133			Self::QueryRowBatchSize => Value::Uint2(32),
134			Self::RowTtlScanBatchSize => Value::Uint8(10000),
135			Self::RowTtlScanInterval => Value::duration_seconds(60),
136			Self::OperatorTtlScanBatchSize => Value::Uint8(10000),
137			Self::OperatorTtlScanInterval => Value::duration_seconds(60),
138			Self::VersionEpochSampleInterval => Value::duration_seconds(1),
139			Self::HistoricalGcBatchSize => Value::Uint8(50_000),
140			Self::HistoricalGcInterval => Value::duration_seconds(30),
141			Self::CdcTtlDuration => Value::None {
142				inner: ValueType::Duration,
143			},
144			Self::CdcTtlScanInterval => Value::duration_seconds(30),
145			Self::CdcTtlScanBatchSize => Value::Uint8(8192),
146			Self::CdcTtlScanMaxBatchesPerTick => Value::Uint8(32),
147			Self::CdcTtlReclaimInterval => Value::duration_seconds(30),
148			Self::CdcCompactInterval => Value::duration_seconds(60),
149			Self::CdcCompactBlockSize => Value::Uint8(1024),
150			Self::CdcCompactSafetyLag => Value::Uint8(1024),
151			Self::CdcCompactMaxBlocksPerTick => Value::Uint8(16),
152			Self::CdcCompactBlockCacheCapacity => Value::Uint8(8),
153			Self::CdcCompactZstdLevel => Value::Uint1(7),
154			Self::CdcRecentCacheCapacity => Value::Uint8(128),
155			Self::MultiReadBufferPages => Value::Uint8(1024),
156			Self::MultiReadBufferPageSize => Value::Uint8(65536),
157			Self::MultiReclaimInterval => Value::duration_seconds(30),
158			Self::FlowTick => Value::duration_seconds(1),
159			Self::CdcWatermarkWaitTimeout => Value::duration_seconds(1),
160			Self::CdcConsumeWaitTimeout => Value::duration_seconds(30),
161			Self::FlowJoinProbeBlockSize => Value::Uint8(1024),
162			Self::ThreadsAsync => Value::Uint2(1),
163			Self::ThreadsSystem => Value::Uint2(2),
164			Self::ThreadsQuery => Value::Uint2(1),
165			Self::ThreadsCommit => Value::Uint2(2),
166			Self::ThreadsBackground => Value::Uint2(1),
167			Self::FlowWorkerThreads => Value::Uint2(0),
168			Self::SubscriptionWorkerThreads => Value::Uint2(0),
169			Self::RuntimeMetricsInterval => Value::duration_seconds(5),
170			Self::MetricFlushInterval => Value::duration_seconds(10),
171			Self::MetricsRuntimeRetention => Value::duration_seconds(7 * 24 * 3600),
172			Self::MetricsProfilerRetention => Value::duration_seconds(3600),
173			Self::MetricsProfilerSnapshotInterval => Value::None {
174				inner: ValueType::Duration,
175			},
176		}
177	}
178
179	pub fn description(&self) -> &'static str {
180		match self {
181			Self::OracleWindowSize => "Number of transactions per conflict-detection window.",
182			Self::OracleWaterMark => "Number of conflict windows retained before cleanup is triggered.",
183			Self::QueryRowBatchSize => {
184				"Number of rows produced per batch by query / DML pipeline operators."
185			}
186			Self::RowTtlScanBatchSize => "Max rows to examine per batch during a row TTL scan.",
187			Self::RowTtlScanInterval => "How often the row TTL actor should scan for expired rows.",
188			Self::OperatorTtlScanBatchSize => {
189				"Max rows to examine per batch during an operator-state TTL scan."
190			}
191			Self::OperatorTtlScanInterval => {
192				"How often the operator-state TTL actor should scan for expired rows."
193			}
194			Self::VersionEpochSampleInterval => {
195				"How often the version-epoch sampler records a (wall-clock, commit version) sample used to map a TTL duration to a cutoff version."
196			}
197			Self::HistoricalGcBatchSize => {
198				"Max historical (key, version) pairs scanned per shape per historical GC tick."
199			}
200			Self::HistoricalGcInterval => {
201				"How often the historical-version GC actor sweeps __historical for versions older than the oracle read watermark."
202			}
203			Self::CdcTtlDuration => {
204				"Maximum age of CDC entries before eviction. When unset, CDC is retained forever; \
205				 when set, must be > 0 and entries older than this duration are evicted regardless \
206				 of consumer state."
207			}
208			Self::CdcTtlScanInterval => {
209				"How often the CDC producer actor scans for and evicts expired CDC entries."
210			}
211			Self::CdcTtlScanBatchSize => {
212				"Max CDC entries deleted per transaction during a CDC TTL eviction tick."
213			}
214			Self::CdcTtlScanMaxBatchesPerTick => {
215				"Upper bound on delete transactions per CDC TTL eviction tick. Caps how long one tick can run when draining a backlog; remaining work continues on the next tick."
216			}
217			Self::CdcTtlReclaimInterval => {
218				"Minimum interval between CDC free-page reclaims (incremental_vacuum + WAL checkpoint) after eviction. Decoupled from the eviction scan so frequent deletes do not trigger frequent heavyweight checkpoints."
219			}
220			Self::CdcCompactInterval => "How often the CDC compaction actor runs.",
221			Self::CdcCompactBlockSize => "Number of CDC entries packed into one compressed block.",
222			Self::CdcCompactSafetyLag => "Versions newer than (max_version - lag) are never compacted.",
223			Self::CdcCompactMaxBlocksPerTick => {
224				"Upper bound on consecutive blocks produced per actor tick."
225			}
226			Self::CdcCompactBlockCacheCapacity => {
227				"Number of decompressed CDC blocks held in the in-memory LRU cache."
228			}
229			Self::CdcCompactZstdLevel => {
230				"Zstd compression level for CDC blocks. Range 1-22; higher means smaller blocks but \
231				 slower compression. Decompression cost is independent of level."
232			}
233			Self::CdcRecentCacheCapacity => {
234				"Number of most-recent decoded CDC entries held in memory so a caught-up consumer \
235				 is served without re-reading and re-deserializing from the backend."
236			}
237			Self::MultiReadBufferPages => {
238				"Number of pages (contiguous row-number buckets) the multi-version read cache keeps \
239				 resident before eviction. Raising it trades RAM for fewer persistent-tier reads."
240			}
241			Self::MultiReadBufferPageSize => {
242				"Number of rows per cached page (bucket) in the multi-version read cache. Must be a \
243				 power of two; sets the granularity of whole-page read-ahead and completeness tracking."
244			}
245			Self::MultiReclaimInterval => {
246				"How often the multi store reclaims free pages (incremental_vacuum + WAL truncate) on its persistent SQLite tier, returning space to the OS after evictions. Decoupled from the GC/flush delete cadence."
247			}
248			Self::FlowTick => {
249				"How often the deferred and transactional flow tick coordinators wake up to dispatch \
250				 due flows."
251			}
252			Self::CdcWatermarkWaitTimeout => {
253				"Backstop timeout for the CDC consumer's wait for the transaction watermark to reach the \
254				 latest commit before consuming; catch-up is event-driven, so this only bounds a missed \
255				 wakeup. Must be > 0."
256			}
257			Self::CdcConsumeWaitTimeout => {
258				"Backstop timeout for the CDC consumer's wait for a consume reply from the downstream \
259				 consumer. A lost reply would otherwise wedge the poll loop forever; on timeout the batch \
260				 is re-dispatched without advancing the checkpoint. Must be > 0."
261			}
262			Self::FlowJoinProbeBlockSize => {
263				"Number of opposite-side rows a streaming join pulls per block when probing its stored \
264				 state. Bounds resident probe memory without dropping matches; smaller trades fewer \
265				 resident rows for more scan round-trips."
266			}
267			Self::ThreadsAsync => {
268				"Number of worker threads for the async runtime. Must be >= 1. \
269				 Read at boot before the runtime starts; changes require restart."
270			}
271			Self::ThreadsSystem => {
272				"Number of worker threads for the system pool (lightweight actors). \
273				 Must be >= 1. Changes require restart."
274			}
275			Self::ThreadsQuery => {
276				"Number of worker threads for the query pool (execution-heavy actors). \
277				 Must be >= 1. Changes require restart."
278			}
279			Self::ThreadsCommit => {
280				"Number of worker threads for the commit pool (synchronous pre-commit flow execution). \
281				 Must be >= 1. Changes require restart."
282			}
283			Self::ThreadsBackground => {
284				"Number of worker threads for the background pool (non-critical cleanup and metrics actors). \
285				 Must be >= 1. Changes require restart."
286			}
287			Self::FlowWorkerThreads => {
288				"Number of deferred-flow worker actors that maintain deferred views in parallel. \
289				 0 means auto (size to the system thread pool). Higher values raise fan-out parallelism \
290				 for many independent views. Changes require restart."
291			}
292			Self::SubscriptionWorkerThreads => {
293				"Number of subscription worker actors that fan out CDC changes to ephemeral \
294				 subscriptions in parallel. 0 means auto (size to the system thread pool). Higher values \
295				 raise fan-out parallelism for many concurrent subscriptions. Changes require restart."
296			}
297			Self::RuntimeMetricsInterval => {
298				"How often the runtime-metrics sampler records a memory snapshot into \
299				 system::metrics::runtime::memory::snapshots. When unset, the history sampler is \
300				 dormant and only the live ::current view is available; when set, must be > 0."
301			}
302			Self::MetricFlushInterval => {
303				"How often the metric collector flushes accumulated storage and CDC stats into the \
304				 system::metrics views. Must be > 0."
305			}
306			Self::MetricsRuntimeRetention => {
307				"Row TTL applied to the system::metrics::runtime::* snapshot series so old samples are \
308				 evicted. Seeded onto each runtime series at bootstrap only when it has no row settings \
309				 yet; changing it affects series created after the change, not already-seeded ones. \
310				 Must be > 0."
311			}
312			Self::MetricsProfilerRetention => {
313				"Row TTL applied to the system::metrics::profiler::*::snapshots series so old samples are \
314				 evicted. Seeded onto each profiler series at bootstrap only when it has no row settings \
315				 yet; changing it affects series created after the change, not already-seeded ones. \
316				 Must be > 0."
317			}
318			Self::MetricsProfilerSnapshotInterval => {
319				"How often the profiler snapshot actor flushes in-memory aggregates into \
320				 system::metrics::profiler::*::snapshots. Defaults to none, which disables snapshot \
321				 persistence entirely (the actor is never spawned) and leaves only the live ::current \
322				 view available; when set, must be > 0. Read once at subsystem construction, so \
323				 changing it requires a restart."
324			}
325		}
326	}
327
328	pub fn requires_restart(&self) -> bool {
329		match self {
330			Self::OracleWindowSize => false,
331			Self::OracleWaterMark => false,
332			Self::QueryRowBatchSize => false,
333			Self::RowTtlScanBatchSize => false,
334			Self::RowTtlScanInterval => false,
335			Self::OperatorTtlScanBatchSize => false,
336			Self::OperatorTtlScanInterval => false,
337			Self::VersionEpochSampleInterval => false,
338			Self::HistoricalGcBatchSize => false,
339			Self::HistoricalGcInterval => false,
340			Self::CdcTtlDuration => false,
341			Self::CdcTtlScanInterval => true,
342			Self::CdcTtlScanBatchSize => false,
343			Self::CdcTtlScanMaxBatchesPerTick => false,
344			Self::CdcTtlReclaimInterval => false,
345			Self::CdcCompactInterval => false,
346			Self::CdcCompactBlockSize => false,
347			Self::CdcCompactSafetyLag => false,
348			Self::CdcCompactMaxBlocksPerTick => false,
349			Self::CdcCompactBlockCacheCapacity => true,
350			Self::CdcCompactZstdLevel => false,
351			Self::CdcRecentCacheCapacity => true,
352			Self::MultiReadBufferPages => true,
353			Self::MultiReadBufferPageSize => true,
354			Self::MultiReclaimInterval => true,
355			Self::FlowTick => false,
356			Self::CdcWatermarkWaitTimeout => false,
357			Self::CdcConsumeWaitTimeout => false,
358			Self::FlowJoinProbeBlockSize => false,
359			Self::ThreadsAsync => true,
360			Self::ThreadsSystem => true,
361			Self::ThreadsQuery => true,
362			Self::ThreadsCommit => true,
363			Self::ThreadsBackground => true,
364			Self::FlowWorkerThreads => true,
365			Self::SubscriptionWorkerThreads => true,
366			Self::RuntimeMetricsInterval => false,
367			Self::MetricFlushInterval => false,
368			Self::MetricsRuntimeRetention => true,
369			Self::MetricsProfilerRetention => true,
370			Self::MetricsProfilerSnapshotInterval => true,
371		}
372	}
373
374	pub fn expected_types(&self) -> &'static [ValueType] {
375		match self {
376			Self::OracleWindowSize => &[ValueType::Uint8],
377			Self::OracleWaterMark => &[ValueType::Uint8],
378			Self::QueryRowBatchSize => &[ValueType::Uint2],
379			Self::RowTtlScanBatchSize => &[ValueType::Uint8],
380			Self::RowTtlScanInterval => &[ValueType::Duration],
381			Self::OperatorTtlScanBatchSize => &[ValueType::Uint8],
382			Self::OperatorTtlScanInterval => &[ValueType::Duration],
383			Self::VersionEpochSampleInterval => &[ValueType::Duration],
384			Self::HistoricalGcBatchSize => &[ValueType::Uint8],
385			Self::HistoricalGcInterval => &[ValueType::Duration],
386			Self::CdcTtlDuration => &[ValueType::Duration],
387			Self::CdcTtlScanInterval => &[ValueType::Duration],
388			Self::CdcTtlScanBatchSize => &[ValueType::Uint8],
389			Self::CdcTtlScanMaxBatchesPerTick => &[ValueType::Uint8],
390			Self::CdcTtlReclaimInterval => &[ValueType::Duration],
391			Self::CdcCompactInterval => &[ValueType::Duration],
392			Self::CdcCompactBlockSize => &[ValueType::Uint8],
393			Self::CdcCompactSafetyLag => &[ValueType::Uint8],
394			Self::CdcCompactMaxBlocksPerTick => &[ValueType::Uint8],
395			Self::CdcCompactBlockCacheCapacity => &[ValueType::Uint8],
396			Self::CdcCompactZstdLevel => &[ValueType::Uint1],
397			Self::CdcRecentCacheCapacity => &[ValueType::Uint8],
398			Self::MultiReadBufferPages => &[ValueType::Uint8],
399			Self::MultiReadBufferPageSize => &[ValueType::Uint8],
400			Self::MultiReclaimInterval => &[ValueType::Duration],
401			Self::FlowTick => &[ValueType::Duration],
402			Self::CdcWatermarkWaitTimeout => &[ValueType::Duration],
403			Self::CdcConsumeWaitTimeout => &[ValueType::Duration],
404			Self::FlowJoinProbeBlockSize => &[ValueType::Uint8],
405			Self::ThreadsAsync => &[ValueType::Uint2],
406			Self::ThreadsSystem => &[ValueType::Uint2],
407			Self::ThreadsQuery => &[ValueType::Uint2],
408			Self::ThreadsCommit => &[ValueType::Uint2],
409			Self::ThreadsBackground => &[ValueType::Uint2],
410			Self::FlowWorkerThreads => &[ValueType::Uint2],
411			Self::SubscriptionWorkerThreads => &[ValueType::Uint2],
412			Self::RuntimeMetricsInterval => &[ValueType::Duration],
413			Self::MetricFlushInterval => &[ValueType::Duration],
414			Self::MetricsRuntimeRetention => &[ValueType::Duration],
415			Self::MetricsProfilerRetention => &[ValueType::Duration],
416			Self::MetricsProfilerSnapshotInterval => &[ValueType::Duration],
417		}
418	}
419
420	pub fn is_optional(&self) -> bool {
421		match self {
422			Self::OracleWindowSize => false,
423			Self::OracleWaterMark => false,
424			Self::QueryRowBatchSize => false,
425			Self::RowTtlScanBatchSize => false,
426			Self::RowTtlScanInterval => false,
427			Self::OperatorTtlScanBatchSize => false,
428			Self::OperatorTtlScanInterval => false,
429			Self::VersionEpochSampleInterval => false,
430			Self::HistoricalGcBatchSize => false,
431			Self::HistoricalGcInterval => false,
432			Self::CdcTtlDuration => true,
433			Self::CdcTtlScanInterval => false,
434			Self::CdcTtlScanBatchSize => false,
435			Self::CdcTtlScanMaxBatchesPerTick => false,
436			Self::CdcTtlReclaimInterval => false,
437			Self::CdcCompactInterval => false,
438			Self::CdcCompactBlockSize => false,
439			Self::CdcCompactSafetyLag => false,
440			Self::CdcCompactMaxBlocksPerTick => false,
441			Self::CdcCompactBlockCacheCapacity => false,
442			Self::CdcCompactZstdLevel => false,
443			Self::CdcRecentCacheCapacity => false,
444			Self::MultiReadBufferPages => false,
445			Self::MultiReadBufferPageSize => false,
446			Self::MultiReclaimInterval => false,
447			Self::FlowTick => false,
448			Self::CdcWatermarkWaitTimeout => false,
449			Self::CdcConsumeWaitTimeout => false,
450			Self::FlowJoinProbeBlockSize => false,
451			Self::ThreadsAsync => false,
452			Self::ThreadsSystem => false,
453			Self::ThreadsQuery => false,
454			Self::ThreadsCommit => false,
455			Self::ThreadsBackground => false,
456			Self::FlowWorkerThreads => false,
457			Self::SubscriptionWorkerThreads => false,
458			Self::RuntimeMetricsInterval => true,
459			Self::MetricFlushInterval => false,
460			Self::MetricsRuntimeRetention => false,
461			Self::MetricsProfilerRetention => false,
462			Self::MetricsProfilerSnapshotInterval => true,
463		}
464	}
465
466	fn validate_canonical(&self, value: &Value) -> Result<(), String> {
467		match self {
468			Self::CdcTtlDuration => match value {
469				Value::None {
470					..
471				} => Ok(()),
472				Value::Duration(d) => {
473					if d.is_positive() {
474						Ok(())
475					} else {
476						Err("CDC_TTL_DURATION must be greater than zero".to_string())
477					}
478				}
479				_ => Ok(()),
480			},
481			Self::CdcCompactInterval => match value {
482				Value::Duration(d) => {
483					if d.is_positive() {
484						Ok(())
485					} else {
486						Err("CDC_COMPACT_INTERVAL must be greater than zero".to_string())
487					}
488				}
489				_ => Ok(()),
490			},
491			Self::CdcCompactBlockSize => match value {
492				Value::Uint8(0) => Err("CDC_COMPACT_BLOCK_SIZE must be greater than zero".to_string()),
493				_ => Ok(()),
494			},
495			Self::QueryRowBatchSize => match value {
496				Value::Uint2(0) => Err("QUERY_ROW_BATCH_SIZE must be greater than zero".to_string()),
497				_ => Ok(()),
498			},
499			Self::CdcCompactBlockCacheCapacity => match value {
500				Value::Uint8(0) => {
501					Err("CDC_COMPACT_BLOCK_CACHE_CAPACITY must be greater than zero".to_string())
502				}
503				_ => Ok(()),
504			},
505			Self::MultiReadBufferPages => match value {
506				Value::Uint8(0) => Err("MULTI_READ_BUFFER_PAGES must be greater than zero".to_string()),
507				_ => Ok(()),
508			},
509			Self::MultiReadBufferPageSize => match value {
510				Value::Uint8(v) if v.is_power_of_two() => Ok(()),
511				Value::Uint8(_) => {
512					Err("MULTI_READ_BUFFER_PAGE_SIZE must be a power of two".to_string())
513				}
514				_ => Ok(()),
515			},
516			Self::CdcCompactZstdLevel => match value {
517				Value::Uint1(v) if (1..=22).contains(v) => Ok(()),
518				Value::Uint1(_) => Err("CDC_COMPACT_ZSTD_LEVEL must be in [1, 22]".to_string()),
519				_ => Ok(()),
520			},
521			Self::HistoricalGcBatchSize => match value {
522				Value::Uint8(0) => {
523					Err("HISTORICAL_GC_BATCH_SIZE must be greater than zero".to_string())
524				}
525				_ => Ok(()),
526			},
527			Self::HistoricalGcInterval => match value {
528				Value::Duration(d) => {
529					if d.is_positive() {
530						Ok(())
531					} else {
532						Err("HISTORICAL_GC_INTERVAL must be greater than zero".to_string())
533					}
534				}
535				_ => Ok(()),
536			},
537			Self::FlowTick => match value {
538				Value::Duration(d) => {
539					if d.is_positive() {
540						Ok(())
541					} else {
542						Err("FLOW_TICK must be greater than zero".to_string())
543					}
544				}
545				_ => Ok(()),
546			},
547			Self::CdcWatermarkWaitTimeout => match value {
548				Value::Duration(d) => {
549					if d.is_positive() {
550						Ok(())
551					} else {
552						Err("CDC_WATERMARK_WAIT_TIMEOUT must be greater than zero".to_string())
553					}
554				}
555				_ => Ok(()),
556			},
557			Self::CdcConsumeWaitTimeout => match value {
558				Value::Duration(d) => {
559					if d.is_positive() {
560						Ok(())
561					} else {
562						Err("CDC_CONSUME_WAIT_TIMEOUT must be greater than zero".to_string())
563					}
564				}
565				_ => Ok(()),
566			},
567			Self::FlowJoinProbeBlockSize => match value {
568				Value::Uint8(0) => {
569					Err("FLOW_JOIN_PROBE_BLOCK_SIZE must be greater than zero".to_string())
570				}
571				_ => Ok(()),
572			},
573			Self::ThreadsAsync => match value {
574				Value::Uint2(0) => Err("THREADS_ASYNC must be greater than zero".to_string()),
575				_ => Ok(()),
576			},
577			Self::ThreadsSystem => match value {
578				Value::Uint2(0) => Err("THREADS_SYSTEM must be greater than zero".to_string()),
579				_ => Ok(()),
580			},
581			Self::ThreadsQuery => match value {
582				Value::Uint2(0) => Err("THREADS_QUERY must be greater than zero".to_string()),
583				_ => Ok(()),
584			},
585			Self::ThreadsCommit => match value {
586				Value::Uint2(0) => Err("THREADS_COMMIT must be greater than zero".to_string()),
587				_ => Ok(()),
588			},
589			Self::ThreadsBackground => match value {
590				Value::Uint2(0) => Err("THREADS_BACKGROUND must be greater than zero".to_string()),
591				_ => Ok(()),
592			},
593			Self::FlowWorkerThreads => Ok(()),
594			Self::SubscriptionWorkerThreads => Ok(()),
595			Self::RuntimeMetricsInterval => match value {
596				Value::None {
597					..
598				} => Ok(()),
599				Value::Duration(d) => {
600					if d.is_positive() {
601						Ok(())
602					} else {
603						Err("RUNTIME_METRICS_INTERVAL must be greater than zero".to_string())
604					}
605				}
606				_ => Ok(()),
607			},
608			Self::MetricFlushInterval => match value {
609				Value::Duration(d) => {
610					if d.is_positive() {
611						Ok(())
612					} else {
613						Err("METRIC_FLUSH_INTERVAL must be greater than zero".to_string())
614					}
615				}
616				_ => Ok(()),
617			},
618			Self::MetricsRuntimeRetention => match value {
619				Value::Duration(d) => {
620					if d.is_positive() {
621						Ok(())
622					} else {
623						Err("METRICS_RUNTIME_RETENTION must be greater than zero".to_string())
624					}
625				}
626				_ => Ok(()),
627			},
628			Self::MetricsProfilerRetention => match value {
629				Value::Duration(d) => {
630					if d.is_positive() {
631						Ok(())
632					} else {
633						Err("METRICS_PROFILER_RETENTION must be greater than zero".to_string())
634					}
635				}
636				_ => Ok(()),
637			},
638			Self::MetricsProfilerSnapshotInterval => match value {
639				Value::None {
640					..
641				} => Ok(()),
642				Value::Duration(d) => {
643					if d.is_positive() {
644						Ok(())
645					} else {
646						Err("METRICS_PROFILER_SNAPSHOT_INTERVAL must be greater than zero"
647							.to_string())
648					}
649				}
650				_ => Ok(()),
651			},
652			_ => Ok(()),
653		}
654	}
655
656	pub fn accept(&self, value: Value) -> Result<Value, AcceptError> {
657		if let Value::None {
658			inner,
659		} = &value
660		{
661			if self.is_optional() && self.expected_types().contains(inner) {
662				return Ok(value);
663			}
664			return Err(AcceptError::TypeMismatch {
665				expected: self.expected_types().to_vec(),
666				actual: value.get_type(),
667			});
668		}
669
670		let canonical = if self.expected_types().contains(&value.get_type()) {
671			value
672		} else {
673			try_coerce_numeric(&value, self.expected_types()).ok_or_else(|| AcceptError::TypeMismatch {
674				expected: self.expected_types().to_vec(),
675				actual: value.get_type(),
676			})?
677		};
678
679		self.validate_canonical(&canonical).map_err(AcceptError::InvalidValue)?;
680		Ok(canonical)
681	}
682}
683
684fn try_coerce_numeric(value: &Value, expected: &[ValueType]) -> Option<Value> {
685	for target in expected {
686		let coerced = match target {
687			ValueType::Uint1 => {
688				value.to_usize().filter(|&v| v <= u8::MAX as usize).map(|v| Value::Uint1(v as u8))
689			}
690			ValueType::Uint2 => {
691				value.to_usize().filter(|&v| v <= u16::MAX as usize).map(|v| Value::Uint2(v as u16))
692			}
693			ValueType::Uint4 => {
694				value.to_usize().filter(|&v| v <= u32::MAX as usize).map(|v| Value::Uint4(v as u32))
695			}
696			ValueType::Uint8 => {
697				value.to_usize().filter(|&v| v <= u64::MAX as usize).map(|v| Value::Uint8(v as u64))
698			}
699			ValueType::Uint16 => value.to_usize().map(|v| Value::Uint16(v as u128)),
700			ValueType::Int1 => {
701				value.to_usize().filter(|&v| v <= i8::MAX as usize).map(|v| Value::Int1(v as i8))
702			}
703			ValueType::Int2 => {
704				value.to_usize().filter(|&v| v <= i16::MAX as usize).map(|v| Value::Int2(v as i16))
705			}
706			ValueType::Int4 => {
707				value.to_usize().filter(|&v| v <= i32::MAX as usize).map(|v| Value::Int4(v as i32))
708			}
709			ValueType::Int8 => {
710				value.to_usize().filter(|&v| v <= i64::MAX as usize).map(|v| Value::Int8(v as i64))
711			}
712			ValueType::Int16 => {
713				value.to_usize().filter(|&v| v <= i128::MAX as usize).map(|v| Value::Int16(v as i128))
714			}
715			ValueType::Uint => value.to_usize().map(|v| Value::Uint(Uint::from_u64(v as u64))),
716			ValueType::Int => value.to_usize().map(|v| Value::Int(Int::from_i64(v as i64))),
717			ValueType::Decimal => value.to_usize().map(|v| Value::Decimal(Decimal::from_i64(v as i64))),
718			ValueType::Float4 => {
719				value.to_usize().and_then(|v| OrderedF32::try_from(v as f32).ok()).map(Value::Float4)
720			}
721			ValueType::Float8 => {
722				value.to_usize().and_then(|v| OrderedF64::try_from(v as f64).ok()).map(Value::Float8)
723			}
724			ValueType::Duration => value
725				.to_usize()
726				.and_then(|v| Duration::from_seconds(v as i64).ok())
727				.map(Value::Duration),
728			_ => None,
729		};
730		if coerced.is_some() {
731			return coerced;
732		}
733	}
734	None
735}
736
737impl fmt::Display for ConfigKey {
738	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
739		match self {
740			Self::OracleWindowSize => write!(f, "ORACLE_WINDOW_SIZE"),
741			Self::OracleWaterMark => write!(f, "ORACLE_WATER_MARK"),
742			Self::QueryRowBatchSize => write!(f, "QUERY_ROW_BATCH_SIZE"),
743			Self::RowTtlScanBatchSize => write!(f, "ROW_TTL_SCAN_BATCH_SIZE"),
744			Self::RowTtlScanInterval => write!(f, "ROW_TTL_SCAN_INTERVAL"),
745			Self::OperatorTtlScanBatchSize => write!(f, "OPERATOR_TTL_SCAN_BATCH_SIZE"),
746			Self::OperatorTtlScanInterval => write!(f, "OPERATOR_TTL_SCAN_INTERVAL"),
747			Self::VersionEpochSampleInterval => write!(f, "VERSION_EPOCH_SAMPLE_INTERVAL"),
748			Self::HistoricalGcBatchSize => write!(f, "HISTORICAL_GC_BATCH_SIZE"),
749			Self::HistoricalGcInterval => write!(f, "HISTORICAL_GC_INTERVAL"),
750			Self::CdcTtlDuration => write!(f, "CDC_TTL_DURATION"),
751			Self::CdcTtlScanInterval => write!(f, "CDC_TTL_SCAN_INTERVAL"),
752			Self::CdcTtlScanBatchSize => write!(f, "CDC_TTL_SCAN_BATCH_SIZE"),
753			Self::CdcTtlScanMaxBatchesPerTick => write!(f, "CDC_TTL_SCAN_MAX_BATCHES_PER_TICK"),
754			Self::CdcTtlReclaimInterval => write!(f, "CDC_TTL_RECLAIM_INTERVAL"),
755			Self::CdcCompactInterval => write!(f, "CDC_COMPACT_INTERVAL"),
756			Self::CdcCompactBlockSize => write!(f, "CDC_COMPACT_BLOCK_SIZE"),
757			Self::CdcCompactSafetyLag => write!(f, "CDC_COMPACT_SAFETY_LAG"),
758			Self::CdcCompactMaxBlocksPerTick => write!(f, "CDC_COMPACT_MAX_BLOCKS_PER_TICK"),
759			Self::CdcCompactBlockCacheCapacity => write!(f, "CDC_COMPACT_BLOCK_CACHE_CAPACITY"),
760			Self::CdcCompactZstdLevel => write!(f, "CDC_COMPACT_ZSTD_LEVEL"),
761			Self::CdcRecentCacheCapacity => write!(f, "CDC_RECENT_CACHE_CAPACITY"),
762			Self::MultiReadBufferPages => write!(f, "MULTI_READ_BUFFER_PAGES"),
763			Self::MultiReadBufferPageSize => write!(f, "MULTI_READ_BUFFER_PAGE_SIZE"),
764			Self::MultiReclaimInterval => write!(f, "MULTI_RECLAIM_INTERVAL"),
765			Self::FlowTick => write!(f, "FLOW_TICK"),
766			Self::CdcWatermarkWaitTimeout => write!(f, "CDC_WATERMARK_WAIT_TIMEOUT"),
767			Self::CdcConsumeWaitTimeout => write!(f, "CDC_CONSUME_WAIT_TIMEOUT"),
768			Self::FlowJoinProbeBlockSize => write!(f, "FLOW_JOIN_PROBE_BLOCK_SIZE"),
769			Self::ThreadsAsync => write!(f, "THREADS_ASYNC"),
770			Self::ThreadsSystem => write!(f, "THREADS_SYSTEM"),
771			Self::ThreadsQuery => write!(f, "THREADS_QUERY"),
772			Self::ThreadsCommit => write!(f, "THREADS_COMMIT"),
773			Self::ThreadsBackground => write!(f, "THREADS_BACKGROUND"),
774			Self::FlowWorkerThreads => write!(f, "FLOW_WORKER_THREADS"),
775			Self::SubscriptionWorkerThreads => write!(f, "SUBSCRIPTION_WORKER_THREADS"),
776			Self::RuntimeMetricsInterval => write!(f, "RUNTIME_METRICS_INTERVAL"),
777			Self::MetricFlushInterval => write!(f, "METRIC_FLUSH_INTERVAL"),
778			Self::MetricsRuntimeRetention => write!(f, "METRICS_RUNTIME_RETENTION"),
779			Self::MetricsProfilerRetention => write!(f, "METRICS_PROFILER_RETENTION"),
780			Self::MetricsProfilerSnapshotInterval => write!(f, "METRICS_PROFILER_SNAPSHOT_INTERVAL"),
781		}
782	}
783}
784
785impl FromStr for ConfigKey {
786	type Err = String;
787
788	fn from_str(s: &str) -> Result<Self, Self::Err> {
789		match s {
790			"ORACLE_WINDOW_SIZE" => Ok(Self::OracleWindowSize),
791			"ORACLE_WATER_MARK" => Ok(Self::OracleWaterMark),
792			"QUERY_ROW_BATCH_SIZE" => Ok(Self::QueryRowBatchSize),
793			"ROW_TTL_SCAN_BATCH_SIZE" => Ok(Self::RowTtlScanBatchSize),
794			"ROW_TTL_SCAN_INTERVAL" => Ok(Self::RowTtlScanInterval),
795			"OPERATOR_TTL_SCAN_BATCH_SIZE" => Ok(Self::OperatorTtlScanBatchSize),
796			"OPERATOR_TTL_SCAN_INTERVAL" => Ok(Self::OperatorTtlScanInterval),
797			"VERSION_EPOCH_SAMPLE_INTERVAL" => Ok(Self::VersionEpochSampleInterval),
798			"HISTORICAL_GC_BATCH_SIZE" => Ok(Self::HistoricalGcBatchSize),
799			"HISTORICAL_GC_INTERVAL" => Ok(Self::HistoricalGcInterval),
800			"CDC_TTL_DURATION" => Ok(Self::CdcTtlDuration),
801			"CDC_TTL_SCAN_INTERVAL" => Ok(Self::CdcTtlScanInterval),
802			"CDC_TTL_SCAN_BATCH_SIZE" => Ok(Self::CdcTtlScanBatchSize),
803			"CDC_TTL_SCAN_MAX_BATCHES_PER_TICK" => Ok(Self::CdcTtlScanMaxBatchesPerTick),
804			"CDC_TTL_RECLAIM_INTERVAL" => Ok(Self::CdcTtlReclaimInterval),
805			"CDC_COMPACT_INTERVAL" => Ok(Self::CdcCompactInterval),
806			"CDC_COMPACT_BLOCK_SIZE" => Ok(Self::CdcCompactBlockSize),
807			"CDC_COMPACT_SAFETY_LAG" => Ok(Self::CdcCompactSafetyLag),
808			"CDC_COMPACT_MAX_BLOCKS_PER_TICK" => Ok(Self::CdcCompactMaxBlocksPerTick),
809			"CDC_COMPACT_BLOCK_CACHE_CAPACITY" => Ok(Self::CdcCompactBlockCacheCapacity),
810			"CDC_COMPACT_ZSTD_LEVEL" => Ok(Self::CdcCompactZstdLevel),
811			"CDC_RECENT_CACHE_CAPACITY" => Ok(Self::CdcRecentCacheCapacity),
812			"MULTI_READ_BUFFER_PAGES" => Ok(Self::MultiReadBufferPages),
813			"MULTI_READ_BUFFER_PAGE_SIZE" => Ok(Self::MultiReadBufferPageSize),
814			"MULTI_RECLAIM_INTERVAL" => Ok(Self::MultiReclaimInterval),
815			"FLOW_TICK" => Ok(Self::FlowTick),
816			"CDC_WATERMARK_WAIT_TIMEOUT" => Ok(Self::CdcWatermarkWaitTimeout),
817			"CDC_CONSUME_WAIT_TIMEOUT" => Ok(Self::CdcConsumeWaitTimeout),
818			"FLOW_JOIN_PROBE_BLOCK_SIZE" => Ok(Self::FlowJoinProbeBlockSize),
819			"THREADS_ASYNC" => Ok(Self::ThreadsAsync),
820			"THREADS_SYSTEM" => Ok(Self::ThreadsSystem),
821			"THREADS_QUERY" => Ok(Self::ThreadsQuery),
822			"THREADS_COMMIT" => Ok(Self::ThreadsCommit),
823			"THREADS_BACKGROUND" => Ok(Self::ThreadsBackground),
824			"FLOW_WORKER_THREADS" => Ok(Self::FlowWorkerThreads),
825			"SUBSCRIPTION_WORKER_THREADS" => Ok(Self::SubscriptionWorkerThreads),
826			"RUNTIME_METRICS_INTERVAL" => Ok(Self::RuntimeMetricsInterval),
827			"METRIC_FLUSH_INTERVAL" => Ok(Self::MetricFlushInterval),
828			"METRICS_RUNTIME_RETENTION" => Ok(Self::MetricsRuntimeRetention),
829			"METRICS_PROFILER_RETENTION" => Ok(Self::MetricsProfilerRetention),
830			"METRICS_PROFILER_SNAPSHOT_INTERVAL" => Ok(Self::MetricsProfilerSnapshotInterval),
831			_ => Err(format!("Unknown system configuration key: {}", s)),
832		}
833	}
834}
835
836#[derive(Debug, Clone)]
837pub struct Config {
838	pub key: ConfigKey,
839
840	pub value: Value,
841
842	pub default_value: Value,
843
844	pub description: &'static str,
845
846	pub requires_restart: bool,
847}
848
849pub trait GetConfig: Send + Sync {
850	fn get_config(&self, key: ConfigKey) -> Value;
851
852	fn get_config_at(&self, key: ConfigKey, version: CommitVersion) -> Value;
853
854	fn get_config_uint8(&self, key: ConfigKey) -> u64 {
855		let val = self.get_config(key);
856		match val {
857			Value::Uint8(v) => v,
858			v => panic!("config key '{}' expected Uint8, got {:?}", key, v),
859		}
860	}
861
862	fn get_config_uint1(&self, key: ConfigKey) -> u8 {
863		let val = self.get_config(key);
864		match val {
865			Value::Uint1(v) => v,
866			v => panic!("config key '{}' expected Uint1, got {:?}", key, v),
867		}
868	}
869
870	fn get_config_uint2(&self, key: ConfigKey) -> u16 {
871		let val = self.get_config(key);
872		match val {
873			Value::Uint2(v) => v,
874			v => panic!("config key '{}' expected Uint2, got {:?}", key, v),
875		}
876	}
877
878	fn get_config_duration(&self, key: ConfigKey) -> Duration {
879		let val = self.get_config(key);
880		match val {
881			Value::Duration(v) => v,
882			v => panic!("config key '{}' expected Duration, got {:?}", key, v),
883		}
884	}
885
886	fn get_config_duration_opt(&self, key: ConfigKey) -> Option<Duration> {
887		match self.get_config(key) {
888			Value::None {
889				..
890			} => None,
891			Value::Duration(v) => Some(v),
892			v => panic!("config key '{}' expected Duration or None, got {:?}", key, v),
893		}
894	}
895}
896
897#[cfg(test)]
898mod tests {
899	use super::*;
900
901	#[test]
902	fn test_cdc_ttl_default_is_typed_null() {
903		// Defaulting to Value::None means "TTL not configured" - producer skips cleanup.
904		let default = ConfigKey::CdcTtlDuration.default_value();
905		assert!(matches!(
906			default,
907			Value::None {
908				inner: ValueType::Duration
909			}
910		));
911	}
912
913	#[test]
914	fn test_cdc_ttl_accept_passes_typed_null() {
915		let none = Value::None {
916			inner: ValueType::Duration,
917		};
918		let v = ConfigKey::CdcTtlDuration.accept(none.clone()).unwrap();
919		assert_eq!(v, none);
920	}
921
922	#[test]
923	fn test_cdc_ttl_accept_passes_positive_duration() {
924		let one_sec = Value::duration_seconds(1);
925		assert_eq!(ConfigKey::CdcTtlDuration.accept(one_sec.clone()).unwrap(), one_sec);
926
927		let one_hour = Value::duration_seconds(3600);
928		assert_eq!(ConfigKey::CdcTtlDuration.accept(one_hour.clone()).unwrap(), one_hour);
929	}
930
931	#[test]
932	fn test_cdc_ttl_accept_rejects_zero() {
933		let zero = Value::duration_seconds(0);
934		match ConfigKey::CdcTtlDuration.accept(zero).unwrap_err() {
935			AcceptError::InvalidValue(reason) => {
936				assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
937			}
938			other => panic!("expected InvalidValue, got {other:?}"),
939		}
940	}
941
942	#[test]
943	fn test_cdc_ttl_accept_rejects_negative() {
944		let negative = Value::duration_seconds(-5);
945		assert!(matches!(ConfigKey::CdcTtlDuration.accept(negative), Err(AcceptError::InvalidValue(_))));
946	}
947
948	#[test]
949	fn test_other_keys_accept_in_type_values() {
950		// Keys without bespoke validation should accept any in-type value.
951		assert!(ConfigKey::OracleWindowSize.accept(Value::Uint8(0)).is_ok());
952		assert!(ConfigKey::RowTtlScanInterval.accept(Value::duration_seconds(0)).is_ok());
953	}
954
955	#[test]
956	fn test_cdc_ttl_round_trips_through_display_and_from_str() {
957		let key: ConfigKey = "CDC_TTL_DURATION".parse().unwrap();
958		assert_eq!(key, ConfigKey::CdcTtlDuration);
959		assert_eq!(format!("{}", ConfigKey::CdcTtlDuration), "CDC_TTL_DURATION");
960	}
961
962	#[test]
963	fn test_cdc_ttl_in_all() {
964		assert!(ConfigKey::all().contains(&ConfigKey::CdcTtlDuration));
965	}
966
967	#[test]
968	fn test_all_contains_every_compact_key_and_has_expected_len() {
969		let all = ConfigKey::all();
970		assert_eq!(all.len(), 41);
971		assert!(all.contains(&ConfigKey::MetricsRuntimeRetention));
972		assert!(all.contains(&ConfigKey::MetricsProfilerRetention));
973		assert!(all.contains(&ConfigKey::MetricsProfilerSnapshotInterval));
974		assert!(all.contains(&ConfigKey::VersionEpochSampleInterval));
975		assert!(all.contains(&ConfigKey::CdcWatermarkWaitTimeout));
976		assert!(all.contains(&ConfigKey::CdcConsumeWaitTimeout));
977		assert!(all.contains(&ConfigKey::FlowJoinProbeBlockSize));
978		assert!(all.contains(&ConfigKey::MultiReclaimInterval));
979		assert!(all.contains(&ConfigKey::CdcTtlScanInterval));
980		assert!(all.contains(&ConfigKey::CdcTtlScanBatchSize));
981		assert!(all.contains(&ConfigKey::CdcTtlReclaimInterval));
982		assert!(all.contains(&ConfigKey::CdcTtlScanMaxBatchesPerTick));
983		assert!(all.contains(&ConfigKey::CdcCompactInterval));
984		assert!(all.contains(&ConfigKey::CdcCompactBlockSize));
985		assert!(all.contains(&ConfigKey::CdcCompactSafetyLag));
986		assert!(all.contains(&ConfigKey::CdcCompactMaxBlocksPerTick));
987		assert!(all.contains(&ConfigKey::CdcCompactBlockCacheCapacity));
988		assert!(all.contains(&ConfigKey::CdcCompactZstdLevel));
989		assert!(all.contains(&ConfigKey::CdcRecentCacheCapacity));
990		assert!(all.contains(&ConfigKey::MultiReadBufferPages));
991		assert!(all.contains(&ConfigKey::MultiReadBufferPageSize));
992		assert!(all.contains(&ConfigKey::QueryRowBatchSize));
993		assert!(all.contains(&ConfigKey::ThreadsAsync));
994		assert!(all.contains(&ConfigKey::ThreadsSystem));
995		assert!(all.contains(&ConfigKey::ThreadsQuery));
996		assert!(all.contains(&ConfigKey::ThreadsCommit));
997		assert!(all.contains(&ConfigKey::ThreadsBackground));
998		assert!(all.contains(&ConfigKey::RuntimeMetricsInterval));
999		assert!(all.contains(&ConfigKey::MetricFlushInterval));
1000		assert!(all.contains(&ConfigKey::SubscriptionWorkerThreads));
1001	}
1002
1003	#[test]
1004	fn test_runtime_metrics_interval_metadata() {
1005		// Single optional Duration knob: default on (5s), none disables the history sampler.
1006		assert_eq!(ConfigKey::RuntimeMetricsInterval.default_value(), Value::duration_seconds(5));
1007		assert_eq!(ConfigKey::RuntimeMetricsInterval.expected_types(), &[ValueType::Duration]);
1008		assert!(ConfigKey::RuntimeMetricsInterval.is_optional());
1009	}
1010
1011	#[test]
1012	fn test_runtime_metrics_interval_round_trip() {
1013		assert_eq!("RUNTIME_METRICS_INTERVAL".parse::<ConfigKey>().unwrap(), ConfigKey::RuntimeMetricsInterval);
1014		assert_eq!(format!("{}", ConfigKey::RuntimeMetricsInterval), "RUNTIME_METRICS_INTERVAL");
1015	}
1016
1017	#[test]
1018	fn test_runtime_metrics_interval_accepts_none_and_positive_rejects_zero() {
1019		let none = Value::None {
1020			inner: ValueType::Duration,
1021		};
1022		assert_eq!(ConfigKey::RuntimeMetricsInterval.accept(none.clone()).unwrap(), none);
1023
1024		let five = Value::duration_seconds(5);
1025		assert_eq!(ConfigKey::RuntimeMetricsInterval.accept(five.clone()).unwrap(), five);
1026
1027		let zero = Value::duration_seconds(0);
1028		assert!(matches!(ConfigKey::RuntimeMetricsInterval.accept(zero), Err(AcceptError::InvalidValue(_))));
1029	}
1030
1031	#[test]
1032	fn test_metric_flush_interval_metadata() {
1033		// Always-on (non-optional) Duration knob defaulting to the historical 10s flush cadence.
1034		assert_eq!(ConfigKey::MetricFlushInterval.default_value(), Value::duration_seconds(10));
1035		assert_eq!(ConfigKey::MetricFlushInterval.expected_types(), &[ValueType::Duration]);
1036		assert!(!ConfigKey::MetricFlushInterval.is_optional());
1037		assert!(!ConfigKey::MetricFlushInterval.requires_restart());
1038	}
1039
1040	#[test]
1041	fn test_metric_flush_interval_round_trip() {
1042		assert_eq!("METRIC_FLUSH_INTERVAL".parse::<ConfigKey>().unwrap(), ConfigKey::MetricFlushInterval);
1043		assert_eq!(format!("{}", ConfigKey::MetricFlushInterval), "METRIC_FLUSH_INTERVAL");
1044	}
1045
1046	#[test]
1047	fn test_metric_flush_interval_accepts_positive_rejects_zero() {
1048		let ten = Value::duration_seconds(10);
1049		assert_eq!(ConfigKey::MetricFlushInterval.accept(ten.clone()).unwrap(), ten);
1050
1051		let zero = Value::duration_seconds(0);
1052		assert!(matches!(ConfigKey::MetricFlushInterval.accept(zero), Err(AcceptError::InvalidValue(_))));
1053	}
1054
1055	#[test]
1056	fn test_cdc_recent_cache_capacity_round_trip() {
1057		assert_eq!(
1058			"CDC_RECENT_CACHE_CAPACITY".parse::<ConfigKey>().unwrap(),
1059			ConfigKey::CdcRecentCacheCapacity
1060		);
1061		assert_eq!(format!("{}", ConfigKey::CdcRecentCacheCapacity), "CDC_RECENT_CACHE_CAPACITY");
1062	}
1063
1064	#[test]
1065	fn test_cdc_recent_cache_capacity_metadata() {
1066		assert_eq!(ConfigKey::CdcRecentCacheCapacity.default_value(), Value::Uint8(128));
1067		assert_eq!(ConfigKey::CdcRecentCacheCapacity.expected_types(), &[ValueType::Uint8]);
1068		assert!(ConfigKey::CdcRecentCacheCapacity.requires_restart());
1069		assert!(!ConfigKey::CdcRecentCacheCapacity.is_optional());
1070	}
1071
1072	#[test]
1073	fn test_multi_read_buffer_pages_round_trip() {
1074		assert_eq!("MULTI_READ_BUFFER_PAGES".parse::<ConfigKey>().unwrap(), ConfigKey::MultiReadBufferPages);
1075		assert_eq!(format!("{}", ConfigKey::MultiReadBufferPages), "MULTI_READ_BUFFER_PAGES");
1076	}
1077
1078	#[test]
1079	fn test_multi_read_buffer_pages_metadata_and_rejects_zero() {
1080		assert_eq!(ConfigKey::MultiReadBufferPages.default_value(), Value::Uint8(1024));
1081		assert_eq!(ConfigKey::MultiReadBufferPages.expected_types(), &[ValueType::Uint8]);
1082		assert!(ConfigKey::MultiReadBufferPages.requires_restart());
1083		assert!(!ConfigKey::MultiReadBufferPages.is_optional());
1084		match ConfigKey::MultiReadBufferPages.accept(Value::Uint8(0)).unwrap_err() {
1085			AcceptError::InvalidValue(reason) => {
1086				assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1087			}
1088			other => panic!("expected InvalidValue, got {other:?}"),
1089		}
1090	}
1091
1092	#[test]
1093	fn test_multi_read_buffer_page_size_round_trip() {
1094		assert_eq!(
1095			"MULTI_READ_BUFFER_PAGE_SIZE".parse::<ConfigKey>().unwrap(),
1096			ConfigKey::MultiReadBufferPageSize
1097		);
1098		assert_eq!(format!("{}", ConfigKey::MultiReadBufferPageSize), "MULTI_READ_BUFFER_PAGE_SIZE");
1099	}
1100
1101	#[test]
1102	fn test_multi_read_buffer_page_size_metadata_and_rejects_non_power_of_two() {
1103		// Page size must be a power of two because pages are addressed by a row-number bit shift
1104		// (bucket = row >> shift); a non-power-of-two would not map to a single shift.
1105		assert_eq!(ConfigKey::MultiReadBufferPageSize.default_value(), Value::Uint8(65536));
1106		assert_eq!(ConfigKey::MultiReadBufferPageSize.expected_types(), &[ValueType::Uint8]);
1107		assert!(ConfigKey::MultiReadBufferPageSize.requires_restart());
1108		assert!(!ConfigKey::MultiReadBufferPageSize.is_optional());
1109		assert_eq!(
1110			ConfigKey::MultiReadBufferPageSize.accept(Value::Uint8(4096)).unwrap(),
1111			Value::Uint8(4096),
1112			"a power-of-two page size is accepted"
1113		);
1114		match ConfigKey::MultiReadBufferPageSize.accept(Value::Uint8(1000)).unwrap_err() {
1115			AcceptError::InvalidValue(reason) => {
1116				assert!(reason.contains("power of two"), "unexpected reason: {reason}");
1117			}
1118			other => panic!("expected InvalidValue, got {other:?}"),
1119		}
1120	}
1121
1122	#[test]
1123	fn test_threads_keys_round_trip() {
1124		assert_eq!("THREADS_ASYNC".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsAsync);
1125		assert_eq!("THREADS_SYSTEM".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsSystem);
1126		assert_eq!("THREADS_QUERY".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsQuery);
1127		assert_eq!("THREADS_COMMIT".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsCommit);
1128		assert_eq!("THREADS_BACKGROUND".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsBackground);
1129		assert_eq!(format!("{}", ConfigKey::ThreadsAsync), "THREADS_ASYNC");
1130		assert_eq!(format!("{}", ConfigKey::ThreadsSystem), "THREADS_SYSTEM");
1131		assert_eq!(format!("{}", ConfigKey::ThreadsQuery), "THREADS_QUERY");
1132		assert_eq!(format!("{}", ConfigKey::ThreadsCommit), "THREADS_COMMIT");
1133		assert_eq!(format!("{}", ConfigKey::ThreadsBackground), "THREADS_BACKGROUND");
1134	}
1135
1136	#[test]
1137	fn test_threads_defaults() {
1138		assert_eq!(ConfigKey::ThreadsAsync.default_value(), Value::Uint2(1));
1139		assert_eq!(ConfigKey::ThreadsSystem.default_value(), Value::Uint2(2));
1140		assert_eq!(ConfigKey::ThreadsQuery.default_value(), Value::Uint2(1));
1141		assert_eq!(ConfigKey::ThreadsCommit.default_value(), Value::Uint2(2));
1142		assert_eq!(ConfigKey::ThreadsBackground.default_value(), Value::Uint2(1));
1143	}
1144
1145	#[test]
1146	fn test_threads_reject_zero() {
1147		for key in [
1148			ConfigKey::ThreadsAsync,
1149			ConfigKey::ThreadsSystem,
1150			ConfigKey::ThreadsQuery,
1151			ConfigKey::ThreadsCommit,
1152			ConfigKey::ThreadsBackground,
1153		] {
1154			match key.accept(Value::Uint2(0)).unwrap_err() {
1155				AcceptError::InvalidValue(reason) => {
1156					assert!(
1157						reason.contains("greater than zero"),
1158						"{key}: unexpected reason: {reason}"
1159					);
1160				}
1161				other => panic!("{key}: expected InvalidValue, got {other:?}"),
1162			}
1163		}
1164	}
1165
1166	#[test]
1167	fn test_threads_accept_positive() {
1168		assert_eq!(ConfigKey::ThreadsAsync.accept(Value::Uint2(4)).unwrap(), Value::Uint2(4));
1169		assert_eq!(ConfigKey::ThreadsSystem.accept(Value::Uint2(8)).unwrap(), Value::Uint2(8));
1170		assert_eq!(ConfigKey::ThreadsQuery.accept(Value::Uint2(16)).unwrap(), Value::Uint2(16));
1171		assert_eq!(ConfigKey::ThreadsCommit.accept(Value::Uint2(4)).unwrap(), Value::Uint2(4));
1172		assert_eq!(ConfigKey::ThreadsBackground.accept(Value::Uint2(2)).unwrap(), Value::Uint2(2));
1173	}
1174
1175	#[test]
1176	fn test_threads_coerce_int4_to_uint2() {
1177		let v = ConfigKey::ThreadsQuery.accept(Value::Int4(8)).unwrap();
1178		assert_eq!(v, Value::Uint2(8));
1179	}
1180
1181	#[test]
1182	fn test_threads_require_restart() {
1183		assert!(ConfigKey::ThreadsAsync.requires_restart());
1184		assert!(ConfigKey::ThreadsSystem.requires_restart());
1185		assert!(ConfigKey::ThreadsQuery.requires_restart());
1186		assert!(ConfigKey::ThreadsCommit.requires_restart());
1187		assert!(ConfigKey::ThreadsBackground.requires_restart());
1188	}
1189
1190	#[test]
1191	fn test_query_row_batch_size_default_is_uint2_32() {
1192		assert_eq!(ConfigKey::QueryRowBatchSize.default_value(), Value::Uint2(32));
1193	}
1194
1195	#[test]
1196	fn test_query_row_batch_size_round_trips_through_display_and_from_str() {
1197		let key: ConfigKey = "QUERY_ROW_BATCH_SIZE".parse().unwrap();
1198		assert_eq!(key, ConfigKey::QueryRowBatchSize);
1199		assert_eq!(format!("{}", ConfigKey::QueryRowBatchSize), "QUERY_ROW_BATCH_SIZE");
1200	}
1201
1202	#[test]
1203	fn test_query_row_batch_size_accept_rejects_zero() {
1204		match ConfigKey::QueryRowBatchSize.accept(Value::Uint2(0)).unwrap_err() {
1205			AcceptError::InvalidValue(reason) => {
1206				assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1207			}
1208			other => panic!("expected InvalidValue, got {other:?}"),
1209		}
1210	}
1211
1212	#[test]
1213	fn test_query_row_batch_size_accept_passes_positive() {
1214		assert_eq!(ConfigKey::QueryRowBatchSize.accept(Value::Uint2(1)).unwrap(), Value::Uint2(1));
1215		assert_eq!(ConfigKey::QueryRowBatchSize.accept(Value::Uint2(1024)).unwrap(), Value::Uint2(1024));
1216	}
1217
1218	#[test]
1219	fn test_query_row_batch_size_accept_rejects_zero_after_coercion() {
1220		match ConfigKey::QueryRowBatchSize.accept(Value::Int4(0)).unwrap_err() {
1221			AcceptError::InvalidValue(reason) => {
1222				assert!(reason.contains("greater than zero"));
1223			}
1224			other => panic!("expected InvalidValue, got {other:?}"),
1225		}
1226	}
1227
1228	#[test]
1229	fn test_query_row_batch_size_coerces_int4_to_uint2() {
1230		let v = ConfigKey::QueryRowBatchSize.accept(Value::Int4(64)).unwrap();
1231		assert_eq!(v, Value::Uint2(64));
1232	}
1233
1234	#[test]
1235	fn test_cdc_compact_interval_round_trips_through_display_and_from_str() {
1236		let key: ConfigKey = "CDC_COMPACT_INTERVAL".parse().unwrap();
1237		assert_eq!(key, ConfigKey::CdcCompactInterval);
1238		assert_eq!(format!("{}", ConfigKey::CdcCompactInterval), "CDC_COMPACT_INTERVAL");
1239	}
1240
1241	#[test]
1242	fn test_cdc_compact_block_size_round_trips_through_display_and_from_str() {
1243		let key: ConfigKey = "CDC_COMPACT_BLOCK_SIZE".parse().unwrap();
1244		assert_eq!(key, ConfigKey::CdcCompactBlockSize);
1245		assert_eq!(format!("{}", ConfigKey::CdcCompactBlockSize), "CDC_COMPACT_BLOCK_SIZE");
1246	}
1247
1248	#[test]
1249	fn test_cdc_compact_safety_lag_round_trips_through_display_and_from_str() {
1250		let key: ConfigKey = "CDC_COMPACT_SAFETY_LAG".parse().unwrap();
1251		assert_eq!(key, ConfigKey::CdcCompactSafetyLag);
1252		assert_eq!(format!("{}", ConfigKey::CdcCompactSafetyLag), "CDC_COMPACT_SAFETY_LAG");
1253	}
1254
1255	#[test]
1256	fn test_cdc_compact_max_blocks_per_tick_round_trips_through_display_and_from_str() {
1257		let key: ConfigKey = "CDC_COMPACT_MAX_BLOCKS_PER_TICK".parse().unwrap();
1258		assert_eq!(key, ConfigKey::CdcCompactMaxBlocksPerTick);
1259		assert_eq!(format!("{}", ConfigKey::CdcCompactMaxBlocksPerTick), "CDC_COMPACT_MAX_BLOCKS_PER_TICK");
1260	}
1261
1262	#[test]
1263	fn test_cdc_compact_interval_default_is_duration() {
1264		assert!(matches!(ConfigKey::CdcCompactInterval.default_value(), Value::Duration(_)));
1265	}
1266
1267	#[test]
1268	fn test_cdc_compact_block_size_default_is_uint8_1024() {
1269		assert_eq!(ConfigKey::CdcCompactBlockSize.default_value(), Value::Uint8(1024));
1270	}
1271
1272	#[test]
1273	fn test_cdc_compact_safety_lag_default_is_uint8_1024() {
1274		assert_eq!(ConfigKey::CdcCompactSafetyLag.default_value(), Value::Uint8(1024));
1275	}
1276
1277	#[test]
1278	fn test_cdc_compact_max_blocks_per_tick_default_is_uint8_16() {
1279		assert_eq!(ConfigKey::CdcCompactMaxBlocksPerTick.default_value(), Value::Uint8(16));
1280	}
1281
1282	#[test]
1283	fn test_cdc_compact_interval_accept_passes_positive_duration() {
1284		let one_sec = Value::duration_seconds(1);
1285		assert_eq!(ConfigKey::CdcCompactInterval.accept(one_sec.clone()).unwrap(), one_sec);
1286	}
1287
1288	#[test]
1289	fn test_cdc_compact_interval_accept_rejects_zero() {
1290		let zero = Value::duration_seconds(0);
1291		match ConfigKey::CdcCompactInterval.accept(zero).unwrap_err() {
1292			AcceptError::InvalidValue(reason) => {
1293				assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1294			}
1295			other => panic!("expected InvalidValue, got {other:?}"),
1296		}
1297	}
1298
1299	#[test]
1300	fn test_cdc_compact_interval_accept_rejects_negative() {
1301		let negative = Value::duration_seconds(-5);
1302		assert!(matches!(ConfigKey::CdcCompactInterval.accept(negative), Err(AcceptError::InvalidValue(_))));
1303	}
1304
1305	#[test]
1306	fn test_cdc_compact_block_size_accept_rejects_zero() {
1307		match ConfigKey::CdcCompactBlockSize.accept(Value::Uint8(0)).unwrap_err() {
1308			AcceptError::InvalidValue(reason) => {
1309				assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1310			}
1311			other => panic!("expected InvalidValue, got {other:?}"),
1312		}
1313	}
1314
1315	#[test]
1316	fn test_cdc_compact_block_size_accept_passes_positive() {
1317		assert_eq!(ConfigKey::CdcCompactBlockSize.accept(Value::Uint8(1)).unwrap(), Value::Uint8(1));
1318		assert_eq!(ConfigKey::CdcCompactBlockSize.accept(Value::Uint8(1024)).unwrap(), Value::Uint8(1024));
1319	}
1320
1321	#[test]
1322	fn test_cdc_compact_safety_lag_and_max_blocks_accept_zero() {
1323		assert_eq!(ConfigKey::CdcCompactSafetyLag.accept(Value::Uint8(0)).unwrap(), Value::Uint8(0));
1324		assert_eq!(ConfigKey::CdcCompactMaxBlocksPerTick.accept(Value::Uint8(0)).unwrap(), Value::Uint8(0));
1325	}
1326
1327	#[test]
1328	fn test_accept_coerces_int4_to_uint8_for_block_size() {
1329		// SET CONFIG CDC_COMPACT_BLOCK_SIZE = 1024 (parsed as Int4) becomes Uint8(1024).
1330		let v = ConfigKey::CdcCompactBlockSize.accept(Value::Int4(1024)).unwrap();
1331		assert_eq!(v, Value::Uint8(1024));
1332	}
1333
1334	#[test]
1335	fn test_accept_coerces_int8_to_uint8_for_block_size() {
1336		let v = ConfigKey::CdcCompactBlockSize.accept(Value::Int8(2048)).unwrap();
1337		assert_eq!(v, Value::Uint8(2048));
1338	}
1339
1340	#[test]
1341	fn test_accept_rejects_zero_after_coercion() {
1342		// Int4(0) coerces to Uint8(0), then validate_canonical rejects it.
1343		match ConfigKey::CdcCompactBlockSize.accept(Value::Int4(0)).unwrap_err() {
1344			AcceptError::InvalidValue(reason) => {
1345				assert!(reason.contains("greater than zero"));
1346			}
1347			other => panic!("expected InvalidValue, got {other:?}"),
1348		}
1349	}
1350
1351	#[test]
1352	fn test_accept_rejects_negative_int_for_uint8_key() {
1353		// to_usize() returns None for negatives -> all coercion arms fail -> TypeMismatch.
1354		assert!(matches!(
1355			ConfigKey::CdcCompactBlockSize.accept(Value::Int4(-1)),
1356			Err(AcceptError::TypeMismatch { .. })
1357		));
1358	}
1359
1360	#[test]
1361	fn test_accept_coerces_int_to_duration_via_seconds() {
1362		// SET CONFIG CDC_COMPACT_INTERVAL = 60 (Int4) -> Duration(60s).
1363		let v = ConfigKey::CdcCompactInterval.accept(Value::Int4(60)).unwrap();
1364		assert!(matches!(v, Value::Duration(_)));
1365	}
1366
1367	#[test]
1368	fn test_accept_idempotent_on_canonical_uint8() {
1369		let canonical = Value::Uint8(42);
1370		assert_eq!(ConfigKey::OracleWindowSize.accept(canonical.clone()).unwrap(), canonical);
1371	}
1372
1373	#[test]
1374	fn test_accept_idempotent_on_canonical_duration() {
1375		let canonical = Value::duration_seconds(5);
1376		assert_eq!(ConfigKey::CdcCompactInterval.accept(canonical.clone()).unwrap(), canonical);
1377	}
1378
1379	#[test]
1380	fn test_accept_rejects_typed_null_for_non_optional_key() {
1381		let err = ConfigKey::CdcCompactBlockSize
1382			.accept(Value::None {
1383				inner: ValueType::Uint8,
1384			})
1385			.unwrap_err();
1386		assert!(matches!(err, AcceptError::TypeMismatch { .. }));
1387	}
1388
1389	#[test]
1390	fn test_accept_passes_typed_null_for_optional_key() {
1391		let none = Value::None {
1392			inner: ValueType::Duration,
1393		};
1394		assert_eq!(ConfigKey::CdcTtlDuration.accept(none.clone()).unwrap(), none);
1395	}
1396
1397	#[test]
1398	fn test_accept_rejects_wrong_inner_type_typed_null_for_optional_key() {
1399		// Optional key still rejects typed-null whose inner doesn't match expected_types.
1400		let err = ConfigKey::CdcTtlDuration
1401			.accept(Value::None {
1402				inner: ValueType::Uint8,
1403			})
1404			.unwrap_err();
1405		assert!(matches!(err, AcceptError::TypeMismatch { .. }));
1406	}
1407
1408	#[test]
1409	fn test_metrics_retention_round_trip() {
1410		assert_eq!(
1411			"METRICS_RUNTIME_RETENTION".parse::<ConfigKey>().unwrap(),
1412			ConfigKey::MetricsRuntimeRetention
1413		);
1414		assert_eq!(
1415			"METRICS_PROFILER_RETENTION".parse::<ConfigKey>().unwrap(),
1416			ConfigKey::MetricsProfilerRetention
1417		);
1418		assert_eq!(format!("{}", ConfigKey::MetricsRuntimeRetention), "METRICS_RUNTIME_RETENTION");
1419		assert_eq!(format!("{}", ConfigKey::MetricsProfilerRetention), "METRICS_PROFILER_RETENTION");
1420	}
1421
1422	#[test]
1423	fn test_metrics_retention_defaults_are_7d_and_1h() {
1424		// Runtime snapshots are sampled every few seconds, so a week is the cap before eviction;
1425		// profiler aggregates are far noisier, so they default to a single hour.
1426		assert_eq!(ConfigKey::MetricsRuntimeRetention.default_value(), Value::duration_seconds(7 * 24 * 3600));
1427		assert_eq!(ConfigKey::MetricsProfilerRetention.default_value(), Value::duration_seconds(3600));
1428	}
1429
1430	#[test]
1431	fn test_metrics_retention_metadata() {
1432		for key in [ConfigKey::MetricsRuntimeRetention, ConfigKey::MetricsProfilerRetention] {
1433			assert_eq!(key.expected_types(), &[ValueType::Duration], "{key}");
1434			assert!(!key.is_optional(), "{key} is always defaulted, never unset");
1435		}
1436	}
1437
1438	#[test]
1439	fn test_metrics_retention_rejects_zero() {
1440		// Zero retention would map the eviction cutoff to "now" and wipe every snapshot on the next
1441		// scan, so it must be rejected like the other positive-duration knobs.
1442		for key in [ConfigKey::MetricsRuntimeRetention, ConfigKey::MetricsProfilerRetention] {
1443			match key.accept(Value::duration_seconds(0)).unwrap_err() {
1444				AcceptError::InvalidValue(reason) => {
1445					assert!(
1446						reason.contains("greater than zero"),
1447						"{key}: unexpected reason: {reason}"
1448					);
1449				}
1450				other => panic!("{key}: expected InvalidValue, got {other:?}"),
1451			}
1452		}
1453	}
1454
1455	#[test]
1456	fn test_metrics_profiler_snapshot_interval_default_is_none() {
1457		// Snapshot persistence is opt-in: leaving this key untouched must not spawn the
1458		// ProfilerSnapshotActor or grow system::metrics::profiler::*::snapshots. A consumer
1459		// that wants persisted profiler snapshots sets this explicitly to a positive duration.
1460		assert_eq!(
1461			ConfigKey::MetricsProfilerSnapshotInterval.default_value(),
1462			Value::None {
1463				inner: ValueType::Duration,
1464			}
1465		);
1466	}
1467
1468	#[test]
1469	fn test_metrics_profiler_snapshot_interval_accepts_none_to_disable_persistence() {
1470		// None is the mechanism a consumer (e.g. raptor, which only ever reads the live
1471		// in-memory accumulator and never queries the persisted ::snapshots series) uses to
1472		// stop ProfilerSnapshotActor from being spawned at all, eliminating unbounded
1473		// system::metrics::profiler::*::snapshots disk growth.
1474		let none = Value::None {
1475			inner: ValueType::Duration,
1476		};
1477		assert_eq!(ConfigKey::MetricsProfilerSnapshotInterval.accept(none.clone()).unwrap(), none);
1478	}
1479
1480	#[test]
1481	fn test_metrics_profiler_snapshot_interval_rejects_zero_and_negative() {
1482		// A zero or negative tick interval would either busy-loop the snapshot actor or fail
1483		// to schedule its timer, so it must be rejected like every other positive-duration
1484		// knob rather than silently misbehaving at runtime.
1485		match ConfigKey::MetricsProfilerSnapshotInterval.accept(Value::duration_seconds(0)).unwrap_err() {
1486			AcceptError::InvalidValue(reason) => {
1487				assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1488			}
1489			other => panic!("expected InvalidValue, got {other:?}"),
1490		}
1491		assert!(matches!(
1492			ConfigKey::MetricsProfilerSnapshotInterval.accept(Value::duration_seconds(-5)),
1493			Err(AcceptError::InvalidValue(_))
1494		));
1495	}
1496
1497	#[test]
1498	fn test_metrics_profiler_snapshot_interval_requires_restart() {
1499		// ProfilerSnapshotActor::init() arms a single fixed-period ctx.schedule_tick(...) at
1500		// actor start and never re-reads this config live, so a change without a restart would
1501		// silently have no effect. This test exists so a future change that makes the actor
1502		// live-reconfigurable doesn't forget to flip this bit back to false.
1503		assert!(ConfigKey::MetricsProfilerSnapshotInterval.requires_restart());
1504	}
1505
1506	#[test]
1507	fn test_metrics_profiler_snapshot_interval_round_trips_through_display_and_from_str() {
1508		assert_eq!(
1509			"METRICS_PROFILER_SNAPSHOT_INTERVAL".parse::<ConfigKey>().unwrap(),
1510			ConfigKey::MetricsProfilerSnapshotInterval
1511		);
1512		assert_eq!(
1513			format!("{}", ConfigKey::MetricsProfilerSnapshotInterval),
1514			"METRICS_PROFILER_SNAPSHOT_INTERVAL"
1515		);
1516	}
1517
1518	#[test]
1519	fn test_metrics_profiler_snapshot_interval_in_all() {
1520		assert!(ConfigKey::all().contains(&ConfigKey::MetricsProfilerSnapshotInterval));
1521	}
1522
1523	#[test]
1524	fn test_historical_gc_keys_round_trip() {
1525		assert_eq!("HISTORICAL_GC_BATCH_SIZE".parse::<ConfigKey>().unwrap(), ConfigKey::HistoricalGcBatchSize);
1526		assert_eq!("HISTORICAL_GC_INTERVAL".parse::<ConfigKey>().unwrap(), ConfigKey::HistoricalGcInterval);
1527		assert_eq!(format!("{}", ConfigKey::HistoricalGcBatchSize), "HISTORICAL_GC_BATCH_SIZE");
1528		assert_eq!(format!("{}", ConfigKey::HistoricalGcInterval), "HISTORICAL_GC_INTERVAL");
1529	}
1530
1531	#[test]
1532	fn test_historical_gc_defaults() {
1533		assert_eq!(ConfigKey::HistoricalGcBatchSize.default_value(), Value::Uint8(50_000));
1534		assert!(matches!(ConfigKey::HistoricalGcInterval.default_value(), Value::Duration(_)));
1535	}
1536
1537	#[test]
1538	fn test_historical_gc_batch_size_rejects_zero() {
1539		match ConfigKey::HistoricalGcBatchSize.accept(Value::Uint8(0)).unwrap_err() {
1540			AcceptError::InvalidValue(reason) => {
1541				assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1542			}
1543			other => panic!("expected InvalidValue, got {other:?}"),
1544		}
1545	}
1546
1547	#[test]
1548	fn test_historical_gc_interval_rejects_zero() {
1549		let zero = Value::duration_seconds(0);
1550		match ConfigKey::HistoricalGcInterval.accept(zero).unwrap_err() {
1551			AcceptError::InvalidValue(reason) => {
1552				assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1553			}
1554			other => panic!("expected InvalidValue, got {other:?}"),
1555		}
1556	}
1557}
reifydb_core/interface/catalog/config.rs

reifydb_core/interface/catalog/
config.rs
