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

reifydb_core/interface/catalog/
config.rs
