1use std::{fmt, str::FromStr};
5
6use reifydb_runtime::version_epoch::BUCKET_WIDTH;
7use reifydb_value::value::{Value, duration::Duration, value_type::ValueType};
8
9use crate::common::CommitVersion;
10
11#[derive(Debug, Clone, PartialEq, Eq)]
12pub enum AcceptError {
13 TypeMismatch {
14 expected: Vec<ValueType>,
15 actual: ValueType,
16 },
17
18 InvalidValue(String),
19}
20
21impl fmt::Display for AcceptError {
22 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
23 match self {
24 Self::TypeMismatch {
25 expected,
26 actual,
27 } => {
28 write!(f, "expected one of {:?}, got {:?}", expected, actual)
29 }
30 Self::InvalidValue(reason) => write!(f, "{reason}"),
31 }
32 }
33}
34
35#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
36pub enum ConfigKey {
37 OracleWindowSize,
38 QueryRowBatchSize,
39 QueryMemoryLimit,
40 RetentionEvictInterval,
41 RetentionEvictBatchSize,
42 RetentionEvictMaxBatchesPerTick,
43 EpochBucketInterval,
44 RetentionStartupGrace,
45 MaxRetentionHorizonFloor,
46 HistoricalGcBatchSize,
47 HistoricalGcInterval,
48 CdcTtlDuration,
49 CdcTtlScanInterval,
50 CdcTtlScanBatchSize,
51 CdcCompactInterval,
52 CdcCompactBlockSize,
53 CdcCompactSafetyLag,
54 CdcCompactMaxBlocksPerTick,
55 CdcCompactBlockCacheCapacity,
56 CdcCompactZstdLevel,
57 CdcWalAutocheckpoint,
58 MultiReadBufferPages,
59 MultiReadBufferPageSize,
60 MultiReadBufferBytes,
61 MultiFlushInterval,
62 MultiFlushKeyBudget,
63 MultiWalAutocheckpoint,
64 FlowTick,
65 FlowSampleInterval,
66 FlowBacklogMemoryLimit,
67 FlowPullBatchBytes,
68 FlowLoadBatchBytes,
69 CdcConsumeWaitTimeout,
70 FlowJoinProbeBlockSize,
71 ThreadsAsync,
72 ThreadsCoordination,
73 ThreadsFlow,
74 ThreadsTask,
75 ThreadsCompute,
76 SubscriptionWorkerThreads,
77 MetricsFlushInterval,
78 MetricsSampleInterval,
79 MetricsSnapshotInterval,
80 CommitGroupLinger,
81 CommitGroupMaxEntries,
82 TombstoneReapInterval,
83 TombstoneReapBatchSize,
84}
85
86impl ConfigKey {
87 pub fn all() -> &'static [Self] {
88 &[
89 Self::OracleWindowSize,
90 Self::QueryRowBatchSize,
91 Self::QueryMemoryLimit,
92 Self::RetentionEvictInterval,
93 Self::RetentionEvictBatchSize,
94 Self::RetentionEvictMaxBatchesPerTick,
95 Self::EpochBucketInterval,
96 Self::RetentionStartupGrace,
97 Self::MaxRetentionHorizonFloor,
98 Self::HistoricalGcBatchSize,
99 Self::HistoricalGcInterval,
100 Self::CdcTtlDuration,
101 Self::CdcTtlScanInterval,
102 Self::CdcTtlScanBatchSize,
103 Self::CdcCompactInterval,
104 Self::CdcCompactBlockSize,
105 Self::CdcCompactSafetyLag,
106 Self::CdcCompactMaxBlocksPerTick,
107 Self::CdcCompactBlockCacheCapacity,
108 Self::CdcCompactZstdLevel,
109 Self::CdcWalAutocheckpoint,
110 Self::MultiReadBufferPages,
111 Self::MultiReadBufferPageSize,
112 Self::MultiReadBufferBytes,
113 Self::MultiFlushInterval,
114 Self::MultiFlushKeyBudget,
115 Self::MultiWalAutocheckpoint,
116 Self::FlowTick,
117 Self::FlowSampleInterval,
118 Self::FlowBacklogMemoryLimit,
119 Self::FlowPullBatchBytes,
120 Self::FlowLoadBatchBytes,
121 Self::CdcConsumeWaitTimeout,
122 Self::FlowJoinProbeBlockSize,
123 Self::ThreadsAsync,
124 Self::ThreadsCoordination,
125 Self::ThreadsFlow,
126 Self::ThreadsTask,
127 Self::ThreadsCompute,
128 Self::SubscriptionWorkerThreads,
129 Self::MetricsFlushInterval,
130 Self::MetricsSampleInterval,
131 Self::MetricsSnapshotInterval,
132 Self::CommitGroupLinger,
133 Self::CommitGroupMaxEntries,
134 Self::TombstoneReapInterval,
135 Self::TombstoneReapBatchSize,
136 ]
137 }
138
139 pub fn default_value(&self) -> Value {
140 match self {
141 Self::OracleWindowSize => Value::Uint8(500),
142 Self::QueryRowBatchSize => Value::Uint2(128),
143 Self::QueryMemoryLimit => Value::Uint8(1024 * 1024 * 1024),
144 Self::RetentionEvictInterval => Value::duration_seconds(60),
145 Self::RetentionEvictBatchSize => Value::Uint8(1024),
146 Self::RetentionEvictMaxBatchesPerTick => Value::Uint8(8),
147 Self::EpochBucketInterval => Value::duration_seconds(60),
148 Self::RetentionStartupGrace => Value::duration_seconds(300),
149 Self::MaxRetentionHorizonFloor => Value::duration_seconds(7 * 24 * 60 * 60),
150 Self::HistoricalGcBatchSize => Value::Uint8(50_000),
151 Self::HistoricalGcInterval => Value::duration_seconds(30),
152 Self::CdcTtlDuration => Value::None {
153 inner: ValueType::Duration,
154 },
155 Self::CdcTtlScanInterval => Value::duration_seconds(30),
156 Self::CdcTtlScanBatchSize => Value::Uint8(8192),
157 Self::CdcCompactInterval => Value::duration_seconds(60),
158 Self::CdcCompactBlockSize => Value::Uint8(1024),
159 Self::CdcCompactSafetyLag => Value::Uint8(1024),
160 Self::CdcCompactMaxBlocksPerTick => Value::Uint8(16),
161 Self::CdcCompactBlockCacheCapacity => Value::Uint8(8),
162 Self::CdcCompactZstdLevel => Value::Uint1(2),
163 Self::CdcWalAutocheckpoint => Value::Uint8(10000),
164 Self::MultiReadBufferPages => Value::Uint8(1024),
165 Self::MultiReadBufferPageSize => Value::Uint8(65536),
166 Self::MultiReadBufferBytes => Value::Uint8(64 * 1024 * 1024),
167 Self::MultiFlushInterval => Value::duration_seconds(5),
168 Self::MultiFlushKeyBudget => Value::Uint8(2048),
169 Self::MultiWalAutocheckpoint => Value::Uint8(10000),
170 Self::FlowTick => Value::duration_seconds(1),
171 Self::FlowSampleInterval => Value::duration_seconds(60),
172 Self::FlowBacklogMemoryLimit => Value::Uint8(64 * 1024 * 1024),
173 Self::FlowPullBatchBytes => Value::Uint8(8 * 1024 * 1024),
174 Self::FlowLoadBatchBytes => Value::Uint8(8 * 1024 * 1024),
175 Self::CdcConsumeWaitTimeout => Value::duration_seconds(30),
176 Self::FlowJoinProbeBlockSize => Value::Uint8(1024),
177 Self::ThreadsAsync => Value::Uint2(1),
178 Self::ThreadsCoordination => Value::Uint2(2),
179 Self::ThreadsFlow => Value::Uint2(2),
180 Self::ThreadsTask => Value::Uint2(2),
181 Self::ThreadsCompute => Value::Uint2(2),
182 Self::SubscriptionWorkerThreads => Value::Uint2(0),
183 Self::MetricsFlushInterval => Value::duration_seconds(10),
184 Self::MetricsSampleInterval => Value::duration_seconds(10),
185 Self::MetricsSnapshotInterval => Value::None {
186 inner: ValueType::Duration,
187 },
188 Self::CommitGroupLinger => Value::None {
189 inner: ValueType::Duration,
190 },
191 Self::CommitGroupMaxEntries => Value::Uint8(256),
192 Self::TombstoneReapInterval => Value::duration_seconds(1),
193 Self::TombstoneReapBatchSize => Value::Uint8(1024),
194 }
195 }
196
197 pub fn description(&self) -> &'static str {
198 match self {
199 Self::OracleWindowSize => "Number of transactions per conflict-detection window.",
200 Self::QueryRowBatchSize => {
201 "Number of rows produced per batch by query / DML pipeline operators."
202 }
203 Self::QueryMemoryLimit => {
204 "Maximum bytes a single query may buffer in memory across its blocking operators (joins, sort, top k, distinct) and its accumulated result. A query that would exceed this fails with QUERY_006 instead of growing without bound. Read fresh for each query, so changes take effect immediately."
205 }
206 Self::RetentionEvictInterval => {
207 "How often the retention evictor scans objects with a row TTL for expired rows."
208 }
209 Self::RetentionEvictBatchSize => {
210 "Max rows examined (and thus evicted) per transaction during a retention eviction tick."
211 }
212 Self::RetentionEvictMaxBatchesPerTick => {
213 "Upper bound on eviction transactions per retention tick. Caps how long one tick can run when draining a backlog; remaining work resumes on the next tick."
214 }
215 Self::EpochBucketInterval => {
216 "Wall-clock width of one durable version-epoch bucket. The epoch log persists at most one \
217 (bucket, commit version) sample per bucket, and those samples are what let TTLs resolve a \
218 cutoff after a restart. Smaller buckets give finer expiry resolution at the cost of more \
219 persisted samples over the retention horizon."
220 }
221 Self::RetentionStartupGrace => {
222 "How long after startup every retention executor computes cutoffs but deletes nothing. A \
223 process restarted after a long downtime wakes with a large expired backlog; the grace \
224 period plus per-class budgets drain it over many ticks instead of one mass eviction."
225 }
226 Self::MaxRetentionHorizonFloor => {
227 "Lower bound on the retained version-epoch horizon. The horizon is the longest declared \
228 TTL in the catalog, never less than this floor; epoch samples older than the horizon are \
229 pruned. A TTL longer than the horizon could not resolve a cutoff, so it is rejected at \
230 declaration time rather than silently never expiring."
231 }
232 Self::HistoricalGcBatchSize => {
233 "Max historical (key, version) pairs scanned per object per historical GC tick."
234 }
235 Self::HistoricalGcInterval => {
236 "How often the historical-version GC actor sweeps __historical for versions older than the oracle read watermark."
237 }
238 Self::CdcTtlDuration => {
239 "Maximum age of CDC entries before eviction. When unset, CDC is retained forever; \
240 when set, must be > 0 and entries older than this duration are evicted regardless \
241 of consumer state."
242 }
243 Self::CdcTtlScanInterval => {
244 "How often the CDC producer actor scans for and evicts expired CDC entries."
245 }
246 Self::CdcTtlScanBatchSize => {
247 "Max CDC entries deleted per transaction during a CDC TTL eviction tick."
248 }
249 Self::CdcCompactInterval => "How often the CDC compaction actor runs.",
250 Self::CdcCompactBlockSize => "Number of CDC entries packed into one compressed block.",
251 Self::CdcCompactSafetyLag => "Versions newer than (max_version - lag) are never compacted.",
252 Self::CdcCompactMaxBlocksPerTick => {
253 "Upper bound on consecutive blocks produced per actor tick."
254 }
255 Self::CdcCompactBlockCacheCapacity => {
256 "Number of decompressed CDC blocks held in the in-memory LRU cache."
257 }
258 Self::CdcCompactZstdLevel => {
259 "Zstd compression level for CDC blocks. Range 1-22; higher means smaller blocks but \
260 slower compression. Decompression cost is independent of level."
261 }
262 Self::CdcWalAutocheckpoint => {
263 "WAL frame threshold (SQLite wal_autocheckpoint PRAGMA) for the CDC log's SQLite tier. \
264 CDC has no explicit checkpoint of its own, so this is the sole control over how often \
265 cdc.db's WAL is checkpointed into the main file. Higher values checkpoint less often with \
266 a larger WAL; since CDC is written on the commit path, this also bounds how often a commit \
267 pays an inline auto-checkpoint. Read once at boot; changing it requires a restart."
268 }
269 Self::MultiReadBufferPages => {
270 "Number of pages (contiguous row-number buckets) the multi-version read cache keeps \
271 resident before eviction. Raising it trades RAM for fewer persistent-tier reads."
272 }
273 Self::MultiReadBufferPageSize => {
274 "Number of rows per cached page (bucket) in the multi-version read cache. Must be a \
275 power of two; sets the granularity of whole-page read-ahead and completeness tracking."
276 }
277 Self::MultiReadBufferBytes => {
278 "Resident byte budget for the multi-version read cache, split evenly across its shards. \
279 None disables the cache outright: no page is warmed or inserted, so reads always go to \
280 the persistent tier. Read once at boot; changing it requires a restart."
281 }
282 Self::MultiFlushInterval => {
283 "How often the persistent-flush actor drains the in-memory commit buffer into the multi \
284 store's SQLite tier. Longer intervals coalesce more writes per flush - a larger WAL - at \
285 the cost of more resident commit-buffer memory and a longer window before data is \
286 materialized in the persistent file. Read once at boot; changing it requires a restart."
287 }
288 Self::MultiFlushKeyBudget => {
289 "Maximum keys the persistent-flush class moves from the commit buffer to the SQLite tier \
290 in one slice. Bounds how long a single flush holds the lane, so a large backlog drains \
291 across ticks instead of stalling every other retention class behind it."
292 }
293 Self::MultiWalAutocheckpoint => {
294 "WAL frame threshold for the multi store's SQLite tier: sets the SQLite \
295 wal_autocheckpoint PRAGMA that governs when SQLite folds the WAL back into the main \
296 database. Higher values checkpoint less often with a larger WAL, reducing checkpoint \
297 I/O; lower values keep the WAL small at the cost of more frequent checkpoints. Read once \
298 at boot; changing it requires a restart."
299 }
300 Self::FlowTick => {
301 "How often the deferred and transactional flow tick coordinators wake up to dispatch \
302 due flows."
303 }
304 Self::FlowSampleInterval => {
305 "How often each flow actor samples its operators' approximate memory into the \
306 system::metrics::runtime::memory samples (scope operator::N). Runs on the operator's \
307 own thread, off the apply path. When none, operator sampling is disabled entirely; when \
308 set, must be > 0."
309 }
310 Self::FlowBacklogMemoryLimit => {
311 "Byte ceiling of the shared in-memory backlog of decoded CDC entries that feeds flow \
312 consumers. Producer-fed at commit granularity; entries below every flow's cursor are \
313 dropped eagerly and the lowest versions are evicted first once the ceiling is exceeded, \
314 at which point a flow that far behind reloads from disk through the catch-up loader. \
315 Because payload rows are shared, the tally is an upper bound of unique memory."
316 }
317 Self::FlowPullBatchBytes => {
318 "Byte budget a flow actor applies per pull from the CDC backlog. A flow that has fallen \
319 behind receives up to this many bytes of decoded changes in one slice, so catch-up is \
320 vectorized instead of per-version."
321 }
322 Self::FlowLoadBatchBytes => {
323 "Byte budget of one catch-up loader read from the CDC log on behalf of flows that are \
324 behind the in-memory backlog. Identical concurrent requests share a single read."
325 }
326 Self::CdcConsumeWaitTimeout => {
327 "Backstop timeout for the CDC consumer's wait for a consume reply from the downstream \
328 consumer. A lost reply would otherwise wedge the poll loop forever; on timeout the batch \
329 is re-dispatched without advancing the checkpoint. Must be > 0."
330 }
331 Self::FlowJoinProbeBlockSize => {
332 "Number of opposite-side rows a streaming join pulls per block when probing its stored \
333 state. Bounds resident probe memory without dropping matches; smaller trades fewer \
334 resident rows for more scan round-trips."
335 }
336 Self::ThreadsAsync => {
337 "Number of worker threads for the async runtime. Must be >= 1. \
338 Read at boot before the runtime starts; changes require restart."
339 }
340 Self::ThreadsCoordination => {
341 "Number of worker threads for the coordination group (long-lived actors with \
342 tiny high-frequency handlers and periodic background actors); pinned dispatch. \
343 Must be >= 1. Changes require restart."
344 }
345 Self::ThreadsFlow => {
346 "Number of worker threads for the flow group (long-lived heavy-handler actors: \
347 materialized-view flow execution); pinned dispatch. \
348 Must be >= 1. Changes require restart."
349 }
350 Self::ThreadsTask => {
351 "Number of worker threads for the task pool (short-lived work: per-request \
352 actors and one-shot jobs). Must be >= 1. Changes require restart."
353 }
354 Self::ThreadsCompute => {
355 "Number of worker threads for the compute pool (data-parallel work via install(), \
356 never actors). Must be >= 1. Changes require restart."
357 }
358 Self::SubscriptionWorkerThreads => {
359 "Number of subscription worker actors that fan out CDC changes to ephemeral \
360 subscriptions in parallel. 0 means auto (size to the system thread pool). Higher values \
361 raise fan-out parallelism for many concurrent subscriptions. Changes require restart."
362 }
363 Self::MetricsFlushInterval => {
364 "How often the metric collector flushes accumulated storage and CDC accounting into the \
365 system::metrics KV store that backs the storage and cdc views. Must be > 0."
366 }
367 Self::MetricsSampleInterval => {
368 "How often the metrics sampler polls every domain, rolls the window and publishes the \
369 system::metrics ::current and ::total caches. Always on; there is no off value, only a \
370 cadence. Must be > 0. Read once at boot; changing it requires a restart."
371 }
372 Self::MetricsSnapshotInterval => {
373 "How often the published ::current reading of every domain is appended to its ::snapshots \
374 series. When none, no snapshot is ever written; when set, must be > 0 and not shorter than \
375 METRICS_SAMPLE_INTERVAL. Read once at boot; changing it requires a restart."
376 }
377 Self::CommitGroupLinger => {
378 "Maximum time an unchecked commit submitted to the group-commit coordinator waits \
379 for other commits to join its group before the merged transaction is flushed. \
380 Defaults to none, which disables grouping entirely (every submission commits \
381 immediately in its own transaction); when set, must be > 0. Read once at database \
382 construction, so changing it requires a restart."
383 }
384 Self::CommitGroupMaxEntries => {
385 "Upper bound on commits merged into one group-commit flush. A group is flushed as \
386 soon as it reaches this size, even before the linger expires. Must be > 0."
387 }
388 Self::TombstoneReapInterval => {
389 "How often the tombstone reaper scans persistent tables for delete-mode tombstones whose superseding write has flushed."
390 }
391 Self::TombstoneReapBatchSize => {
392 "Max tombstones one reap statement may physically delete per table per slice. Bounds the write-connection hold; remaining tombstones drain on the next slice."
393 }
394 }
395 }
396
397 pub fn requires_restart(&self) -> bool {
398 match self {
399 Self::OracleWindowSize => false,
400 Self::QueryRowBatchSize => false,
401 Self::QueryMemoryLimit => false,
402 Self::RetentionEvictInterval => true,
403 Self::RetentionEvictBatchSize => false,
404 Self::RetentionEvictMaxBatchesPerTick => false,
405 Self::EpochBucketInterval => false,
406 Self::RetentionStartupGrace => false,
407 Self::MaxRetentionHorizonFloor => false,
408 Self::HistoricalGcBatchSize => false,
409 Self::HistoricalGcInterval => false,
410 Self::CdcTtlDuration => false,
411 Self::CdcTtlScanInterval => true,
412 Self::CdcTtlScanBatchSize => false,
413 Self::CdcCompactInterval => false,
414 Self::CdcCompactBlockSize => false,
415 Self::CdcCompactSafetyLag => false,
416 Self::CdcCompactMaxBlocksPerTick => false,
417 Self::CdcCompactBlockCacheCapacity => true,
418 Self::CdcCompactZstdLevel => false,
419 Self::CdcWalAutocheckpoint => true,
420 Self::MultiReadBufferPages => true,
421 Self::MultiReadBufferPageSize => true,
422 Self::MultiReadBufferBytes => true,
423 Self::MultiFlushInterval => true,
424 Self::MultiFlushKeyBudget => false,
425 Self::MultiWalAutocheckpoint => true,
426 Self::FlowTick => false,
427 Self::FlowSampleInterval => false,
428 Self::FlowBacklogMemoryLimit => true,
429 Self::FlowPullBatchBytes => true,
430 Self::FlowLoadBatchBytes => true,
431 Self::CdcConsumeWaitTimeout => false,
432 Self::FlowJoinProbeBlockSize => false,
433 Self::ThreadsAsync => true,
434 Self::ThreadsCoordination => true,
435 Self::ThreadsFlow => true,
436 Self::ThreadsTask => true,
437 Self::ThreadsCompute => true,
438 Self::SubscriptionWorkerThreads => true,
439 Self::MetricsFlushInterval => false,
440 Self::MetricsSampleInterval => true,
441 Self::MetricsSnapshotInterval => true,
442 Self::CommitGroupLinger => true,
443 Self::CommitGroupMaxEntries => true,
444 Self::TombstoneReapInterval => false,
445 Self::TombstoneReapBatchSize => false,
446 }
447 }
448
449 pub fn expected_types(&self) -> &'static [ValueType] {
450 match self {
451 Self::OracleWindowSize => &[ValueType::Uint8],
452 Self::QueryRowBatchSize => &[ValueType::Uint2],
453 Self::QueryMemoryLimit => &[ValueType::Uint8],
454 Self::RetentionEvictInterval => &[ValueType::Duration],
455 Self::RetentionEvictBatchSize => &[ValueType::Uint8],
456 Self::RetentionEvictMaxBatchesPerTick => &[ValueType::Uint8],
457 Self::EpochBucketInterval => &[ValueType::Duration],
458 Self::RetentionStartupGrace => &[ValueType::Duration],
459 Self::MaxRetentionHorizonFloor => &[ValueType::Duration],
460 Self::HistoricalGcBatchSize => &[ValueType::Uint8],
461 Self::HistoricalGcInterval => &[ValueType::Duration],
462 Self::CdcTtlDuration => &[ValueType::Duration],
463 Self::CdcTtlScanInterval => &[ValueType::Duration],
464 Self::CdcTtlScanBatchSize => &[ValueType::Uint8],
465 Self::CdcCompactInterval => &[ValueType::Duration],
466 Self::CdcCompactBlockSize => &[ValueType::Uint8],
467 Self::CdcCompactSafetyLag => &[ValueType::Uint8],
468 Self::CdcCompactMaxBlocksPerTick => &[ValueType::Uint8],
469 Self::CdcCompactBlockCacheCapacity => &[ValueType::Uint8],
470 Self::CdcCompactZstdLevel => &[ValueType::Uint1],
471 Self::CdcWalAutocheckpoint => &[ValueType::Uint8],
472 Self::MultiReadBufferPages => &[ValueType::Uint8],
473 Self::MultiReadBufferPageSize => &[ValueType::Uint8],
474 Self::MultiReadBufferBytes => &[ValueType::Uint8],
475 Self::MultiFlushInterval => &[ValueType::Duration],
476 Self::MultiFlushKeyBudget => &[ValueType::Uint8],
477 Self::MultiWalAutocheckpoint => &[ValueType::Uint8],
478 Self::FlowTick => &[ValueType::Duration],
479 Self::FlowSampleInterval => &[ValueType::Duration],
480 Self::FlowBacklogMemoryLimit => &[ValueType::Uint8],
481 Self::FlowPullBatchBytes => &[ValueType::Uint8],
482 Self::FlowLoadBatchBytes => &[ValueType::Uint8],
483 Self::CdcConsumeWaitTimeout => &[ValueType::Duration],
484 Self::FlowJoinProbeBlockSize => &[ValueType::Uint8],
485 Self::ThreadsAsync => &[ValueType::Uint2],
486 Self::ThreadsCoordination => &[ValueType::Uint2],
487 Self::ThreadsFlow => &[ValueType::Uint2],
488 Self::ThreadsTask => &[ValueType::Uint2],
489 Self::ThreadsCompute => &[ValueType::Uint2],
490 Self::SubscriptionWorkerThreads => &[ValueType::Uint2],
491 Self::MetricsFlushInterval => &[ValueType::Duration],
492 Self::MetricsSampleInterval => &[ValueType::Duration],
493 Self::MetricsSnapshotInterval => &[ValueType::Duration],
494 Self::CommitGroupLinger => &[ValueType::Duration],
495 Self::CommitGroupMaxEntries => &[ValueType::Uint8],
496 Self::TombstoneReapInterval => &[ValueType::Duration],
497 Self::TombstoneReapBatchSize => &[ValueType::Uint8],
498 }
499 }
500
501 pub fn is_optional(&self) -> bool {
502 match self {
503 Self::OracleWindowSize => false,
504 Self::QueryRowBatchSize => false,
505 Self::QueryMemoryLimit => false,
506 Self::RetentionEvictInterval => false,
507 Self::RetentionEvictBatchSize => false,
508 Self::RetentionEvictMaxBatchesPerTick => false,
509 Self::EpochBucketInterval => false,
510 Self::RetentionStartupGrace => false,
511 Self::MaxRetentionHorizonFloor => false,
512 Self::HistoricalGcBatchSize => false,
513 Self::HistoricalGcInterval => false,
514 Self::CdcTtlDuration => true,
515 Self::CdcTtlScanInterval => false,
516 Self::CdcTtlScanBatchSize => false,
517 Self::CdcCompactInterval => false,
518 Self::CdcCompactBlockSize => false,
519 Self::CdcCompactSafetyLag => false,
520 Self::CdcCompactMaxBlocksPerTick => false,
521 Self::CdcCompactBlockCacheCapacity => false,
522 Self::CdcCompactZstdLevel => false,
523 Self::CdcWalAutocheckpoint => false,
524 Self::MultiReadBufferPages => false,
525 Self::MultiReadBufferPageSize => false,
526 Self::MultiReadBufferBytes => true,
527 Self::MultiFlushInterval => false,
528 Self::MultiFlushKeyBudget => false,
529 Self::MultiWalAutocheckpoint => false,
530 Self::FlowTick => false,
531 Self::FlowSampleInterval => true,
532 Self::FlowBacklogMemoryLimit => false,
533 Self::FlowPullBatchBytes => false,
534 Self::FlowLoadBatchBytes => false,
535 Self::CdcConsumeWaitTimeout => false,
536 Self::FlowJoinProbeBlockSize => false,
537 Self::ThreadsAsync => false,
538 Self::ThreadsCoordination => false,
539 Self::ThreadsFlow => false,
540 Self::ThreadsTask => false,
541 Self::ThreadsCompute => false,
542 Self::SubscriptionWorkerThreads => false,
543 Self::MetricsFlushInterval => false,
544 Self::MetricsSampleInterval => false,
545 Self::MetricsSnapshotInterval => true,
546 Self::CommitGroupLinger => true,
547 Self::CommitGroupMaxEntries => false,
548 Self::TombstoneReapInterval => false,
549 Self::TombstoneReapBatchSize => false,
550 }
551 }
552
553 fn validate_canonical(&self, value: &Value) -> Result<(), String> {
554 match self {
555 Self::CdcTtlDuration => match value {
556 Value::None {
557 ..
558 } => Ok(()),
559 Value::Duration(d) => {
560 if d.is_positive() {
561 Ok(())
562 } else {
563 Err("CDC_TTL_DURATION must be greater than zero".to_string())
564 }
565 }
566 _ => Ok(()),
567 },
568 Self::CdcCompactInterval => match value {
569 Value::Duration(d) => {
570 if d.is_positive() {
571 Ok(())
572 } else {
573 Err("CDC_COMPACT_INTERVAL must be greater than zero".to_string())
574 }
575 }
576 _ => Ok(()),
577 },
578 Self::EpochBucketInterval => match value {
579 Value::Duration(d) if !d.is_positive() => {
580 Err("EPOCH_BUCKET_INTERVAL must be greater than zero".to_string())
581 }
582 Value::Duration(d) if d.to_std().as_secs() < BUCKET_WIDTH.seconds() => Err(format!(
583 "EPOCH_BUCKET_INTERVAL must be at least {}s: the version epoch resolves cutoffs at \
584 second granularity, so a shorter bucket truncates to zero and silently disables \
585 coarse compaction",
586 BUCKET_WIDTH.seconds()
587 )),
588 _ => Ok(()),
589 },
590 Self::RetentionStartupGrace => match value {
591 Value::Duration(d) if d.is_negative() => {
592 Err("RETENTION_STARTUP_GRACE must not be negative".to_string())
593 }
594 _ => Ok(()),
595 },
596 Self::MaxRetentionHorizonFloor => match value {
597 Value::Duration(d) if !d.is_positive() => {
598 Err("MAX_RETENTION_HORIZON_FLOOR must be greater than zero".to_string())
599 }
600 _ => Ok(()),
601 },
602 Self::CdcCompactBlockSize => match value {
603 Value::Uint8(0) => Err("CDC_COMPACT_BLOCK_SIZE must be greater than zero".to_string()),
604 _ => Ok(()),
605 },
606 Self::QueryRowBatchSize => match value {
607 Value::Uint2(0) => Err("QUERY_ROW_BATCH_SIZE must be greater than zero".to_string()),
608 _ => Ok(()),
609 },
610 Self::QueryMemoryLimit => match value {
611 Value::Uint8(0) => Err("QUERY_MEMORY_LIMIT must be greater than zero".to_string()),
612 _ => Ok(()),
613 },
614 Self::FlowBacklogMemoryLimit => match value {
615 Value::Uint8(0) => {
616 Err("FLOW_BACKLOG_MEMORY_LIMIT must be greater than zero".to_string())
617 }
618 _ => Ok(()),
619 },
620 Self::FlowPullBatchBytes => match value {
621 Value::Uint8(0) => Err("FLOW_PULL_BATCH_BYTES must be greater than zero".to_string()),
622 _ => Ok(()),
623 },
624 Self::FlowLoadBatchBytes => match value {
625 Value::Uint8(0) => Err("FLOW_LOAD_BATCH_BYTES must be greater than zero".to_string()),
626 _ => Ok(()),
627 },
628 Self::CdcCompactBlockCacheCapacity => match value {
629 Value::Uint8(0) => {
630 Err("CDC_COMPACT_BLOCK_CACHE_CAPACITY must be greater than zero".to_string())
631 }
632 _ => Ok(()),
633 },
634 Self::MultiReadBufferPages => match value {
635 Value::Uint8(0) => Err("MULTI_READ_BUFFER_PAGES must be greater than zero".to_string()),
636 _ => Ok(()),
637 },
638 Self::MultiReadBufferPageSize => match value {
639 Value::Uint8(v) if v.is_power_of_two() => Ok(()),
640 Value::Uint8(_) => {
641 Err("MULTI_READ_BUFFER_PAGE_SIZE must be a power of two".to_string())
642 }
643 _ => Ok(()),
644 },
645 Self::MultiReadBufferBytes => match value {
646 Value::Uint8(0) => Err(
647 "MULTI_READ_BUFFER_BYTES must be greater than zero; use none to disable the read cache"
648 .to_string(),
649 ),
650 _ => Ok(()),
651 },
652 Self::MultiFlushInterval => match value {
653 Value::Duration(d) if d.is_positive() => Ok(()),
654 Value::Duration(_) => Err("MULTI_FLUSH_INTERVAL must be greater than zero".to_string()),
655 _ => Ok(()),
656 },
657 Self::MultiFlushKeyBudget => match value {
658 Value::Uint8(n) if *n > 0 => Ok(()),
659 Value::Uint8(_) => Err("MULTI_FLUSH_KEY_BUDGET must be greater than zero".to_string()),
660 _ => Ok(()),
661 },
662 Self::MultiWalAutocheckpoint => match value {
663 Value::Uint8(0) => {
664 Err("MULTI_WAL_AUTOCHECKPOINT must be greater than zero".to_string())
665 }
666 _ => Ok(()),
667 },
668 Self::CdcWalAutocheckpoint => match value {
669 Value::Uint8(0) => Err("CDC_WAL_AUTOCHECKPOINT must be greater than zero".to_string()),
670 _ => Ok(()),
671 },
672 Self::CdcCompactZstdLevel => match value {
673 Value::Uint1(v) if (1..=22).contains(v) => Ok(()),
674 Value::Uint1(_) => Err("CDC_COMPACT_ZSTD_LEVEL must be in [1, 22]".to_string()),
675 _ => Ok(()),
676 },
677 Self::HistoricalGcBatchSize => match value {
678 Value::Uint8(0) => {
679 Err("HISTORICAL_GC_BATCH_SIZE must be greater than zero".to_string())
680 }
681 _ => Ok(()),
682 },
683 Self::HistoricalGcInterval => match value {
684 Value::Duration(d) => {
685 if d.is_positive() {
686 Ok(())
687 } else {
688 Err("HISTORICAL_GC_INTERVAL must be greater than zero".to_string())
689 }
690 }
691 _ => Ok(()),
692 },
693 Self::FlowTick => match value {
694 Value::Duration(d) => {
695 if d.is_positive() {
696 Ok(())
697 } else {
698 Err("FLOW_TICK must be greater than zero".to_string())
699 }
700 }
701 _ => Ok(()),
702 },
703 Self::FlowSampleInterval => match value {
704 Value::None {
705 ..
706 } => Ok(()),
707 Value::Duration(d) => {
708 if d.is_positive() {
709 Ok(())
710 } else {
711 Err("FLOW_SAMPLE_INTERVAL must be greater than zero".to_string())
712 }
713 }
714 _ => Ok(()),
715 },
716 Self::CdcConsumeWaitTimeout => match value {
717 Value::Duration(d) => {
718 if d.is_positive() {
719 Ok(())
720 } else {
721 Err("CDC_CONSUME_WAIT_TIMEOUT must be greater than zero".to_string())
722 }
723 }
724 _ => Ok(()),
725 },
726 Self::FlowJoinProbeBlockSize => match value {
727 Value::Uint8(0) => {
728 Err("FLOW_JOIN_PROBE_BLOCK_SIZE must be greater than zero".to_string())
729 }
730 _ => Ok(()),
731 },
732 Self::ThreadsAsync => match value {
733 Value::Uint2(0) => Err("THREADS_ASYNC must be greater than zero".to_string()),
734 _ => Ok(()),
735 },
736 Self::ThreadsCoordination => match value {
737 Value::Uint2(0) => Err("THREADS_COORDINATION must be greater than zero".to_string()),
738 _ => Ok(()),
739 },
740 Self::ThreadsFlow => match value {
741 Value::Uint2(0) => Err("THREADS_FLOW must be greater than zero".to_string()),
742 _ => Ok(()),
743 },
744 Self::ThreadsTask => match value {
745 Value::Uint2(0) => Err("THREADS_TASK must be greater than zero".to_string()),
746 _ => Ok(()),
747 },
748 Self::ThreadsCompute => match value {
749 Value::Uint2(0) => Err("THREADS_COMPUTE must be greater than zero".to_string()),
750 _ => Ok(()),
751 },
752 Self::SubscriptionWorkerThreads => Ok(()),
753 Self::MetricsFlushInterval => match value {
754 Value::Duration(d) => {
755 if d.is_positive() {
756 Ok(())
757 } else {
758 Err("METRICS_FLUSH_INTERVAL must be greater than zero".to_string())
759 }
760 }
761 _ => Ok(()),
762 },
763 Self::MetricsSampleInterval => match value {
764 Value::Duration(d) => {
765 if d.is_positive() {
766 Ok(())
767 } else {
768 Err("METRICS_SAMPLE_INTERVAL must be greater than zero".to_string())
769 }
770 }
771 _ => Ok(()),
772 },
773 Self::MetricsSnapshotInterval => match value {
774 Value::None {
775 ..
776 } => Ok(()),
777 Value::Duration(d) => {
778 if d.is_positive() {
779 Ok(())
780 } else {
781 Err("METRICS_SNAPSHOT_INTERVAL must be greater than zero".to_string())
782 }
783 }
784 _ => Ok(()),
785 },
786 Self::CommitGroupLinger => match value {
787 Value::None {
788 ..
789 } => Ok(()),
790 Value::Duration(d) => {
791 if d.is_positive() {
792 Ok(())
793 } else {
794 Err("COMMIT_GROUP_LINGER must be greater than zero".to_string())
795 }
796 }
797 _ => Ok(()),
798 },
799 Self::CommitGroupMaxEntries => match value {
800 Value::Uint8(0) => {
801 Err("COMMIT_GROUP_MAX_ENTRIES must be greater than zero".to_string())
802 }
803 _ => Ok(()),
804 },
805 _ => Ok(()),
806 }
807 }
808
809 pub fn accept(&self, value: Value) -> Result<Value, AcceptError> {
810 if let Value::None {
811 inner,
812 } = &value
813 {
814 if self.is_optional() && self.expected_types().contains(inner) {
815 return Ok(value);
816 }
817 return Err(AcceptError::TypeMismatch {
818 expected: self.expected_types().to_vec(),
819 actual: value.get_type(),
820 });
821 }
822
823 if !self.expected_types().contains(&value.get_type()) {
824 return Err(AcceptError::TypeMismatch {
825 expected: self.expected_types().to_vec(),
826 actual: value.get_type(),
827 });
828 }
829
830 self.validate_canonical(&value).map_err(AcceptError::InvalidValue)?;
831 Ok(value)
832 }
833}
834
835impl fmt::Display for ConfigKey {
836 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
837 match self {
838 Self::OracleWindowSize => write!(f, "ORACLE_WINDOW_SIZE"),
839 Self::QueryRowBatchSize => write!(f, "QUERY_ROW_BATCH_SIZE"),
840 Self::QueryMemoryLimit => write!(f, "QUERY_MEMORY_LIMIT"),
841 Self::RetentionEvictInterval => write!(f, "RETENTION_EVICT_INTERVAL"),
842 Self::RetentionEvictBatchSize => write!(f, "RETENTION_EVICT_BATCH_SIZE"),
843 Self::RetentionEvictMaxBatchesPerTick => write!(f, "RETENTION_EVICT_MAX_BATCHES_PER_TICK"),
844 Self::EpochBucketInterval => write!(f, "EPOCH_BUCKET_INTERVAL"),
845 Self::RetentionStartupGrace => write!(f, "RETENTION_STARTUP_GRACE"),
846 Self::MaxRetentionHorizonFloor => write!(f, "MAX_RETENTION_HORIZON_FLOOR"),
847 Self::HistoricalGcBatchSize => write!(f, "HISTORICAL_GC_BATCH_SIZE"),
848 Self::HistoricalGcInterval => write!(f, "HISTORICAL_GC_INTERVAL"),
849 Self::CdcTtlDuration => write!(f, "CDC_TTL_DURATION"),
850 Self::CdcTtlScanInterval => write!(f, "CDC_TTL_SCAN_INTERVAL"),
851 Self::CdcTtlScanBatchSize => write!(f, "CDC_TTL_SCAN_BATCH_SIZE"),
852 Self::CdcCompactInterval => write!(f, "CDC_COMPACT_INTERVAL"),
853 Self::CdcCompactBlockSize => write!(f, "CDC_COMPACT_BLOCK_SIZE"),
854 Self::CdcCompactSafetyLag => write!(f, "CDC_COMPACT_SAFETY_LAG"),
855 Self::CdcCompactMaxBlocksPerTick => write!(f, "CDC_COMPACT_MAX_BLOCKS_PER_TICK"),
856 Self::CdcCompactBlockCacheCapacity => write!(f, "CDC_COMPACT_BLOCK_CACHE_CAPACITY"),
857 Self::CdcCompactZstdLevel => write!(f, "CDC_COMPACT_ZSTD_LEVEL"),
858 Self::CdcWalAutocheckpoint => write!(f, "CDC_WAL_AUTOCHECKPOINT"),
859 Self::MultiReadBufferPages => write!(f, "MULTI_READ_BUFFER_PAGES"),
860 Self::MultiReadBufferPageSize => write!(f, "MULTI_READ_BUFFER_PAGE_SIZE"),
861 Self::MultiReadBufferBytes => write!(f, "MULTI_READ_BUFFER_BYTES"),
862 Self::MultiFlushInterval => write!(f, "MULTI_FLUSH_INTERVAL"),
863 Self::MultiFlushKeyBudget => write!(f, "MULTI_FLUSH_KEY_BUDGET"),
864 Self::MultiWalAutocheckpoint => write!(f, "MULTI_WAL_AUTOCHECKPOINT"),
865 Self::FlowTick => write!(f, "FLOW_TICK"),
866 Self::FlowSampleInterval => write!(f, "FLOW_SAMPLE_INTERVAL"),
867 Self::FlowBacklogMemoryLimit => write!(f, "FLOW_BACKLOG_MEMORY_LIMIT"),
868 Self::FlowPullBatchBytes => write!(f, "FLOW_PULL_BATCH_BYTES"),
869 Self::FlowLoadBatchBytes => write!(f, "FLOW_LOAD_BATCH_BYTES"),
870 Self::CdcConsumeWaitTimeout => write!(f, "CDC_CONSUME_WAIT_TIMEOUT"),
871 Self::FlowJoinProbeBlockSize => write!(f, "FLOW_JOIN_PROBE_BLOCK_SIZE"),
872 Self::ThreadsAsync => write!(f, "THREADS_ASYNC"),
873 Self::ThreadsCoordination => write!(f, "THREADS_COORDINATION"),
874 Self::ThreadsFlow => write!(f, "THREADS_FLOW"),
875 Self::ThreadsTask => write!(f, "THREADS_TASK"),
876 Self::ThreadsCompute => write!(f, "THREADS_COMPUTE"),
877 Self::SubscriptionWorkerThreads => write!(f, "SUBSCRIPTION_WORKER_THREADS"),
878 Self::MetricsFlushInterval => write!(f, "METRICS_FLUSH_INTERVAL"),
879 Self::MetricsSampleInterval => write!(f, "METRICS_SAMPLE_INTERVAL"),
880 Self::MetricsSnapshotInterval => write!(f, "METRICS_SNAPSHOT_INTERVAL"),
881 Self::CommitGroupLinger => write!(f, "COMMIT_GROUP_LINGER"),
882 Self::CommitGroupMaxEntries => write!(f, "COMMIT_GROUP_MAX_ENTRIES"),
883 Self::TombstoneReapInterval => write!(f, "TOMBSTONE_REAP_INTERVAL"),
884 Self::TombstoneReapBatchSize => write!(f, "TOMBSTONE_REAP_BATCH_SIZE"),
885 }
886 }
887}
888
889impl FromStr for ConfigKey {
890 type Err = String;
891
892 fn from_str(s: &str) -> Result<Self, Self::Err> {
893 match s {
894 "ORACLE_WINDOW_SIZE" => Ok(Self::OracleWindowSize),
895 "QUERY_ROW_BATCH_SIZE" => Ok(Self::QueryRowBatchSize),
896 "QUERY_MEMORY_LIMIT" => Ok(Self::QueryMemoryLimit),
897 "RETENTION_EVICT_INTERVAL" => Ok(Self::RetentionEvictInterval),
898 "RETENTION_EVICT_BATCH_SIZE" => Ok(Self::RetentionEvictBatchSize),
899 "RETENTION_EVICT_MAX_BATCHES_PER_TICK" => Ok(Self::RetentionEvictMaxBatchesPerTick),
900 "EPOCH_BUCKET_INTERVAL" => Ok(Self::EpochBucketInterval),
901 "RETENTION_STARTUP_GRACE" => Ok(Self::RetentionStartupGrace),
902 "MAX_RETENTION_HORIZON_FLOOR" => Ok(Self::MaxRetentionHorizonFloor),
903 "HISTORICAL_GC_BATCH_SIZE" => Ok(Self::HistoricalGcBatchSize),
904 "HISTORICAL_GC_INTERVAL" => Ok(Self::HistoricalGcInterval),
905 "CDC_TTL_DURATION" => Ok(Self::CdcTtlDuration),
906 "CDC_TTL_SCAN_INTERVAL" => Ok(Self::CdcTtlScanInterval),
907 "CDC_TTL_SCAN_BATCH_SIZE" => Ok(Self::CdcTtlScanBatchSize),
908 "CDC_COMPACT_INTERVAL" => Ok(Self::CdcCompactInterval),
909 "CDC_COMPACT_BLOCK_SIZE" => Ok(Self::CdcCompactBlockSize),
910 "CDC_COMPACT_SAFETY_LAG" => Ok(Self::CdcCompactSafetyLag),
911 "CDC_COMPACT_MAX_BLOCKS_PER_TICK" => Ok(Self::CdcCompactMaxBlocksPerTick),
912 "CDC_COMPACT_BLOCK_CACHE_CAPACITY" => Ok(Self::CdcCompactBlockCacheCapacity),
913 "CDC_COMPACT_ZSTD_LEVEL" => Ok(Self::CdcCompactZstdLevel),
914 "CDC_WAL_AUTOCHECKPOINT" => Ok(Self::CdcWalAutocheckpoint),
915 "MULTI_READ_BUFFER_PAGES" => Ok(Self::MultiReadBufferPages),
916 "MULTI_READ_BUFFER_PAGE_SIZE" => Ok(Self::MultiReadBufferPageSize),
917 "MULTI_READ_BUFFER_BYTES" => Ok(Self::MultiReadBufferBytes),
918 "MULTI_FLUSH_INTERVAL" => Ok(Self::MultiFlushInterval),
919 "MULTI_FLUSH_KEY_BUDGET" => Ok(Self::MultiFlushKeyBudget),
920 "MULTI_WAL_AUTOCHECKPOINT" => Ok(Self::MultiWalAutocheckpoint),
921 "FLOW_TICK" => Ok(Self::FlowTick),
922 "FLOW_SAMPLE_INTERVAL" => Ok(Self::FlowSampleInterval),
923 "FLOW_BACKLOG_MEMORY_LIMIT" => Ok(Self::FlowBacklogMemoryLimit),
924 "FLOW_PULL_BATCH_BYTES" => Ok(Self::FlowPullBatchBytes),
925 "FLOW_LOAD_BATCH_BYTES" => Ok(Self::FlowLoadBatchBytes),
926 "CDC_CONSUME_WAIT_TIMEOUT" => Ok(Self::CdcConsumeWaitTimeout),
927 "FLOW_JOIN_PROBE_BLOCK_SIZE" => Ok(Self::FlowJoinProbeBlockSize),
928 "THREADS_ASYNC" => Ok(Self::ThreadsAsync),
929 "THREADS_COORDINATION" => Ok(Self::ThreadsCoordination),
930 "THREADS_FLOW" => Ok(Self::ThreadsFlow),
931 "THREADS_TASK" => Ok(Self::ThreadsTask),
932 "THREADS_COMPUTE" => Ok(Self::ThreadsCompute),
933 "SUBSCRIPTION_WORKER_THREADS" => Ok(Self::SubscriptionWorkerThreads),
934 "METRICS_FLUSH_INTERVAL" => Ok(Self::MetricsFlushInterval),
935 "METRICS_SAMPLE_INTERVAL" => Ok(Self::MetricsSampleInterval),
936 "METRICS_SNAPSHOT_INTERVAL" => Ok(Self::MetricsSnapshotInterval),
937 "COMMIT_GROUP_LINGER" => Ok(Self::CommitGroupLinger),
938 "COMMIT_GROUP_MAX_ENTRIES" => Ok(Self::CommitGroupMaxEntries),
939 "TOMBSTONE_REAP_INTERVAL" => Ok(Self::TombstoneReapInterval),
940 "TOMBSTONE_REAP_BATCH_SIZE" => Ok(Self::TombstoneReapBatchSize),
941 _ => Err(format!("Unknown system configuration key: {}", s)),
942 }
943 }
944}
945
946#[derive(Debug, Clone)]
947pub struct Config {
948 pub key: ConfigKey,
949
950 pub value: Value,
951
952 pub default_value: Value,
953
954 pub description: &'static str,
955
956 pub requires_restart: bool,
957}
958
959pub trait GetConfig: Send + Sync {
960 fn get_config(&self, key: ConfigKey) -> Value;
961
962 fn get_config_at(&self, key: ConfigKey, version: CommitVersion) -> Value;
963
964 fn get_config_uint8(&self, key: ConfigKey) -> u64 {
965 let val = self.get_config(key);
966 match val {
967 Value::Uint8(v) => v,
968 v => panic!("config key '{}' expected Uint8, got {:?}", key, v),
969 }
970 }
971
972 fn get_config_uint1(&self, key: ConfigKey) -> u8 {
973 let val = self.get_config(key);
974 match val {
975 Value::Uint1(v) => v,
976 v => panic!("config key '{}' expected Uint1, got {:?}", key, v),
977 }
978 }
979
980 fn get_config_uint2(&self, key: ConfigKey) -> u16 {
981 let val = self.get_config(key);
982 match val {
983 Value::Uint2(v) => v,
984 v => panic!("config key '{}' expected Uint2, got {:?}", key, v),
985 }
986 }
987
988 fn get_config_duration(&self, key: ConfigKey) -> Duration {
989 let val = self.get_config(key);
990 match val {
991 Value::Duration(v) => v,
992 v => panic!("config key '{}' expected Duration, got {:?}", key, v),
993 }
994 }
995
996 fn get_config_duration_opt(&self, key: ConfigKey) -> Option<Duration> {
997 match self.get_config(key) {
998 Value::None {
999 ..
1000 } => None,
1001 Value::Duration(v) => Some(v),
1002 v => panic!("config key '{}' expected Duration or None, got {:?}", key, v),
1003 }
1004 }
1005
1006 fn get_config_uint8_opt(&self, key: ConfigKey) -> Option<u64> {
1007 match self.get_config(key) {
1008 Value::None {
1009 ..
1010 } => None,
1011 Value::Uint8(v) => Some(v),
1012 v => panic!("config key '{}' expected Uint8 or None, got {:?}", key, v),
1013 }
1014 }
1015}
1016
1017#[cfg(test)]
1018mod tests {
1019 use super::*;
1020
1021 #[test]
1022 fn test_cdc_ttl_default_is_typed_null() {
1023 let default = ConfigKey::CdcTtlDuration.default_value();
1025 assert!(matches!(
1026 default,
1027 Value::None {
1028 inner: ValueType::Duration
1029 }
1030 ));
1031 }
1032
1033 #[test]
1034 fn test_cdc_ttl_accept_passes_typed_null() {
1035 let none = Value::None {
1036 inner: ValueType::Duration,
1037 };
1038 let v = ConfigKey::CdcTtlDuration.accept(none.clone()).unwrap();
1039 assert_eq!(v, none);
1040 }
1041
1042 #[test]
1043 fn test_cdc_ttl_accept_passes_positive_duration() {
1044 let one_sec = Value::duration_seconds(1);
1045 assert_eq!(ConfigKey::CdcTtlDuration.accept(one_sec.clone()).unwrap(), one_sec);
1046
1047 let one_hour = Value::duration_seconds(3600);
1048 assert_eq!(ConfigKey::CdcTtlDuration.accept(one_hour.clone()).unwrap(), one_hour);
1049 }
1050
1051 #[test]
1052 fn test_cdc_ttl_accept_rejects_zero() {
1053 let zero = Value::duration_seconds(0);
1054 match ConfigKey::CdcTtlDuration.accept(zero).unwrap_err() {
1055 AcceptError::InvalidValue(reason) => {
1056 assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1057 }
1058 other => panic!("expected InvalidValue, got {other:?}"),
1059 }
1060 }
1061
1062 #[test]
1063 fn test_cdc_ttl_accept_rejects_negative() {
1064 let negative = Value::duration_seconds(-5);
1065 assert!(matches!(ConfigKey::CdcTtlDuration.accept(negative), Err(AcceptError::InvalidValue(_))));
1066 }
1067
1068 #[test]
1069 fn test_other_keys_accept_in_type_values() {
1070 assert!(ConfigKey::OracleWindowSize.accept(Value::Uint8(0)).is_ok());
1071 }
1072
1073 #[test]
1074 fn test_cdc_ttl_round_trips_through_display_and_from_str() {
1075 let key: ConfigKey = "CDC_TTL_DURATION".parse().unwrap();
1076 assert_eq!(key, ConfigKey::CdcTtlDuration);
1077 assert_eq!(format!("{}", ConfigKey::CdcTtlDuration), "CDC_TTL_DURATION");
1078 }
1079
1080 #[test]
1081 fn test_cdc_ttl_in_all() {
1082 assert!(ConfigKey::all().contains(&ConfigKey::CdcTtlDuration));
1083 }
1084
1085 #[test]
1086 fn test_query_memory_limit_defaults_and_round_trips() {
1087 assert_eq!(ConfigKey::QueryMemoryLimit.default_value(), Value::Uint8(1024 * 1024 * 1024));
1088 assert_eq!(ConfigKey::QueryMemoryLimit.expected_types(), &[ValueType::Uint8]);
1089 let key: ConfigKey = "QUERY_MEMORY_LIMIT".parse().unwrap();
1090 assert_eq!(key, ConfigKey::QueryMemoryLimit);
1091 assert_eq!(format!("{}", ConfigKey::QueryMemoryLimit), "QUERY_MEMORY_LIMIT");
1092 }
1093
1094 #[test]
1095 fn test_query_memory_limit_rejects_zero() {
1096 assert!(ConfigKey::QueryMemoryLimit.accept(Value::Uint8(0)).is_err());
1098 assert_eq!(ConfigKey::QueryMemoryLimit.accept(Value::Uint8(1)).unwrap(), Value::Uint8(1));
1099 }
1100
1101 #[test]
1102 fn test_query_memory_limit_requires_restart_and_optional() {
1103 assert!(!ConfigKey::QueryMemoryLimit.requires_restart());
1105 assert!(!ConfigKey::QueryMemoryLimit.is_optional());
1107 }
1108
1109 #[test]
1110 fn test_all_contains_every_compact_key_and_has_expected_len() {
1111 let all = ConfigKey::all();
1112 assert_eq!(all.len(), 47);
1113 assert!(all.contains(&ConfigKey::QueryMemoryLimit));
1114 assert!(all.contains(&ConfigKey::CommitGroupLinger));
1115 assert!(all.contains(&ConfigKey::CommitGroupMaxEntries));
1116 assert!(all.contains(&ConfigKey::RetentionEvictInterval));
1117 assert!(all.contains(&ConfigKey::RetentionEvictBatchSize));
1118 assert!(all.contains(&ConfigKey::RetentionEvictMaxBatchesPerTick));
1119 assert!(all.contains(&ConfigKey::MultiFlushInterval));
1120 assert!(all.contains(&ConfigKey::MultiWalAutocheckpoint));
1121 assert!(all.contains(&ConfigKey::CdcWalAutocheckpoint));
1122 assert!(all.contains(&ConfigKey::CdcConsumeWaitTimeout));
1123 assert!(all.contains(&ConfigKey::FlowJoinProbeBlockSize));
1124 assert!(all.contains(&ConfigKey::CdcTtlScanInterval));
1125 assert!(all.contains(&ConfigKey::CdcTtlScanBatchSize));
1126 assert!(all.contains(&ConfigKey::CdcCompactInterval));
1127 assert!(all.contains(&ConfigKey::CdcCompactBlockSize));
1128 assert!(all.contains(&ConfigKey::CdcCompactSafetyLag));
1129 assert!(all.contains(&ConfigKey::CdcCompactMaxBlocksPerTick));
1130 assert!(all.contains(&ConfigKey::CdcCompactBlockCacheCapacity));
1131 assert!(all.contains(&ConfigKey::CdcCompactZstdLevel));
1132 assert!(all.contains(&ConfigKey::MultiReadBufferPages));
1133 assert!(all.contains(&ConfigKey::FlowBacklogMemoryLimit));
1134 assert!(all.contains(&ConfigKey::FlowPullBatchBytes));
1135 assert!(all.contains(&ConfigKey::FlowLoadBatchBytes));
1136 assert!(all.contains(&ConfigKey::MultiReadBufferPageSize));
1137 assert!(all.contains(&ConfigKey::QueryRowBatchSize));
1138 assert!(all.contains(&ConfigKey::ThreadsAsync));
1139 assert!(all.contains(&ConfigKey::ThreadsCoordination));
1140 assert!(all.contains(&ConfigKey::ThreadsFlow));
1141 assert!(all.contains(&ConfigKey::ThreadsTask));
1142 assert!(all.contains(&ConfigKey::ThreadsCompute));
1143 assert!(all.contains(&ConfigKey::MetricsFlushInterval));
1144 assert!(all.contains(&ConfigKey::SubscriptionWorkerThreads));
1145 assert!(all.contains(&ConfigKey::FlowSampleInterval));
1146 assert!(all.contains(&ConfigKey::MetricsSampleInterval));
1147 assert!(all.contains(&ConfigKey::MetricsSnapshotInterval));
1148 }
1149
1150 #[test]
1151 fn test_metrics_sample_interval_is_always_on() {
1152 assert_eq!(ConfigKey::MetricsSampleInterval.default_value(), Value::duration_seconds(10));
1155 assert_eq!(ConfigKey::MetricsSampleInterval.expected_types(), &[ValueType::Duration]);
1156 assert!(!ConfigKey::MetricsSampleInterval.is_optional(), "there is no off value, only a cadence");
1157 assert!(ConfigKey::MetricsSampleInterval.requires_restart(), "read once at boot");
1158
1159 let ten = Value::duration_seconds(10);
1160 assert_eq!(ConfigKey::MetricsSampleInterval.accept(ten.clone()).unwrap(), ten);
1161 let zero = Value::duration_seconds(0);
1162 assert!(matches!(ConfigKey::MetricsSampleInterval.accept(zero), Err(AcceptError::InvalidValue(_))));
1163 }
1164
1165 #[test]
1166 fn test_metrics_snapshot_interval_accepts_none_and_positive_rejects_zero() {
1167 assert_eq!(
1169 ConfigKey::MetricsSnapshotInterval.default_value(),
1170 Value::None {
1171 inner: ValueType::Duration
1172 },
1173 "snapshotting must be opt-in"
1174 );
1175 assert!(ConfigKey::MetricsSnapshotInterval.is_optional(), "none must stay accepted to turn it off");
1176 assert!(ConfigKey::MetricsSnapshotInterval.requires_restart(), "read once at boot");
1177
1178 let none = Value::None {
1179 inner: ValueType::Duration,
1180 };
1181 assert_eq!(ConfigKey::MetricsSnapshotInterval.accept(none.clone()).unwrap(), none);
1182
1183 let minute = Value::duration_seconds(60);
1184 assert_eq!(ConfigKey::MetricsSnapshotInterval.accept(minute.clone()).unwrap(), minute);
1185
1186 let zero = Value::duration_seconds(0);
1187 match ConfigKey::MetricsSnapshotInterval.accept(zero).unwrap_err() {
1188 AcceptError::InvalidValue(reason) => {
1189 assert!(reason.contains("must be greater than zero"), "unexpected reason: {reason}");
1190 }
1191 other => panic!("expected InvalidValue, got {other:?}"),
1192 }
1193 }
1194
1195 #[test]
1196 fn test_metrics_sampler_keys_round_trip() {
1197 for (key, name) in [
1198 (ConfigKey::MetricsSampleInterval, "METRICS_SAMPLE_INTERVAL"),
1199 (ConfigKey::MetricsSnapshotInterval, "METRICS_SNAPSHOT_INTERVAL"),
1200 ] {
1201 assert_eq!(format!("{key}"), name);
1202 assert_eq!(name.parse::<ConfigKey>().unwrap(), key);
1203 }
1204 }
1205
1206 #[test]
1207 fn test_flow_sample_interval_metadata() {
1208 assert_eq!(ConfigKey::FlowSampleInterval.default_value(), Value::duration_seconds(60));
1211 assert_eq!(ConfigKey::FlowSampleInterval.expected_types(), &[ValueType::Duration]);
1212 assert!(ConfigKey::FlowSampleInterval.is_optional());
1213 }
1214
1215 #[test]
1216 fn test_flow_sample_interval_round_trip() {
1217 assert_eq!("FLOW_SAMPLE_INTERVAL".parse::<ConfigKey>().unwrap(), ConfigKey::FlowSampleInterval);
1218 assert_eq!(format!("{}", ConfigKey::FlowSampleInterval), "FLOW_SAMPLE_INTERVAL");
1219 }
1220
1221 #[test]
1222 fn test_flow_sample_interval_accepts_none_and_positive_rejects_zero() {
1223 let none = Value::None {
1224 inner: ValueType::Duration,
1225 };
1226 assert_eq!(
1227 ConfigKey::FlowSampleInterval.accept(none.clone()).unwrap(),
1228 none,
1229 "none must be accepted so sampling can be turned off"
1230 );
1231
1232 let minute = Value::duration_seconds(60);
1233 assert_eq!(ConfigKey::FlowSampleInterval.accept(minute.clone()).unwrap(), minute);
1234
1235 let zero = Value::duration_seconds(0);
1236 assert!(matches!(ConfigKey::FlowSampleInterval.accept(zero), Err(AcceptError::InvalidValue(_))));
1237 }
1238
1239 #[test]
1240 fn test_metrics_flush_interval_metadata() {
1241 assert_eq!(ConfigKey::MetricsFlushInterval.default_value(), Value::duration_seconds(10));
1243 assert_eq!(ConfigKey::MetricsFlushInterval.expected_types(), &[ValueType::Duration]);
1244 assert!(!ConfigKey::MetricsFlushInterval.is_optional());
1245 assert!(!ConfigKey::MetricsFlushInterval.requires_restart());
1246 }
1247
1248 #[test]
1249 fn test_metrics_flush_interval_round_trip() {
1250 assert_eq!("METRICS_FLUSH_INTERVAL".parse::<ConfigKey>().unwrap(), ConfigKey::MetricsFlushInterval);
1251 assert_eq!(format!("{}", ConfigKey::MetricsFlushInterval), "METRICS_FLUSH_INTERVAL");
1252 }
1253
1254 #[test]
1255 fn test_metrics_flush_interval_accepts_positive_rejects_zero() {
1256 let ten = Value::duration_seconds(10);
1257 assert_eq!(ConfigKey::MetricsFlushInterval.accept(ten.clone()).unwrap(), ten);
1258
1259 let zero = Value::duration_seconds(0);
1260 assert!(matches!(ConfigKey::MetricsFlushInterval.accept(zero), Err(AcceptError::InvalidValue(_))));
1261 }
1262
1263 #[test]
1264 fn test_multi_read_buffer_pages_round_trip() {
1265 assert_eq!("MULTI_READ_BUFFER_PAGES".parse::<ConfigKey>().unwrap(), ConfigKey::MultiReadBufferPages);
1266 assert_eq!(format!("{}", ConfigKey::MultiReadBufferPages), "MULTI_READ_BUFFER_PAGES");
1267 }
1268
1269 #[test]
1270 fn test_multi_read_buffer_pages_metadata_and_rejects_zero() {
1271 assert_eq!(ConfigKey::MultiReadBufferPages.default_value(), Value::Uint8(1024));
1272 assert_eq!(ConfigKey::MultiReadBufferPages.expected_types(), &[ValueType::Uint8]);
1273 assert!(ConfigKey::MultiReadBufferPages.requires_restart());
1274 assert!(!ConfigKey::MultiReadBufferPages.is_optional());
1275 match ConfigKey::MultiReadBufferPages.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_multi_read_buffer_page_size_round_trip() {
1285 assert_eq!(
1286 "MULTI_READ_BUFFER_PAGE_SIZE".parse::<ConfigKey>().unwrap(),
1287 ConfigKey::MultiReadBufferPageSize
1288 );
1289 assert_eq!(format!("{}", ConfigKey::MultiReadBufferPageSize), "MULTI_READ_BUFFER_PAGE_SIZE");
1290 }
1291
1292 #[test]
1293 fn test_multi_read_buffer_page_size_metadata_and_rejects_non_power_of_two() {
1294 assert_eq!(ConfigKey::MultiReadBufferPageSize.default_value(), Value::Uint8(65536));
1297 assert_eq!(ConfigKey::MultiReadBufferPageSize.expected_types(), &[ValueType::Uint8]);
1298 assert!(ConfigKey::MultiReadBufferPageSize.requires_restart());
1299 assert!(!ConfigKey::MultiReadBufferPageSize.is_optional());
1300 assert_eq!(
1301 ConfigKey::MultiReadBufferPageSize.accept(Value::Uint8(4096)).unwrap(),
1302 Value::Uint8(4096),
1303 "a power-of-two page size is accepted"
1304 );
1305 match ConfigKey::MultiReadBufferPageSize.accept(Value::Uint8(1000)).unwrap_err() {
1306 AcceptError::InvalidValue(reason) => {
1307 assert!(reason.contains("power of two"), "unexpected reason: {reason}");
1308 }
1309 other => panic!("expected InvalidValue, got {other:?}"),
1310 }
1311 }
1312
1313 #[test]
1314 fn test_threads_keys_round_trip() {
1315 assert_eq!("THREADS_ASYNC".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsAsync);
1316 assert_eq!("THREADS_COORDINATION".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsCoordination);
1317 assert_eq!("THREADS_FLOW".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsFlow);
1318 assert_eq!("THREADS_TASK".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsTask);
1319 assert_eq!("THREADS_COMPUTE".parse::<ConfigKey>().unwrap(), ConfigKey::ThreadsCompute);
1320 assert_eq!(format!("{}", ConfigKey::ThreadsAsync), "THREADS_ASYNC");
1321 assert_eq!(format!("{}", ConfigKey::ThreadsCoordination), "THREADS_COORDINATION");
1322 assert_eq!(format!("{}", ConfigKey::ThreadsFlow), "THREADS_FLOW");
1323 assert_eq!(format!("{}", ConfigKey::ThreadsTask), "THREADS_TASK");
1324 assert_eq!(format!("{}", ConfigKey::ThreadsCompute), "THREADS_COMPUTE");
1325 }
1326
1327 #[test]
1328 fn test_threads_defaults() {
1329 assert_eq!(ConfigKey::ThreadsAsync.default_value(), Value::Uint2(1));
1330 assert_eq!(ConfigKey::ThreadsCoordination.default_value(), Value::Uint2(2));
1331 assert_eq!(ConfigKey::ThreadsFlow.default_value(), Value::Uint2(2));
1332 assert_eq!(ConfigKey::ThreadsTask.default_value(), Value::Uint2(2));
1333 assert_eq!(ConfigKey::ThreadsCompute.default_value(), Value::Uint2(2));
1334 }
1335
1336 #[test]
1337 fn test_threads_reject_zero() {
1338 for key in [
1339 ConfigKey::ThreadsAsync,
1340 ConfigKey::ThreadsCoordination,
1341 ConfigKey::ThreadsFlow,
1342 ConfigKey::ThreadsTask,
1343 ConfigKey::ThreadsCompute,
1344 ] {
1345 match key.accept(Value::Uint2(0)).unwrap_err() {
1346 AcceptError::InvalidValue(reason) => {
1347 assert!(
1348 reason.contains("greater than zero"),
1349 "{key}: unexpected reason: {reason}"
1350 );
1351 }
1352 other => panic!("{key}: expected InvalidValue, got {other:?}"),
1353 }
1354 }
1355 }
1356
1357 #[test]
1358 fn test_threads_accept_positive() {
1359 assert_eq!(ConfigKey::ThreadsAsync.accept(Value::Uint2(4)).unwrap(), Value::Uint2(4));
1360 assert_eq!(ConfigKey::ThreadsCoordination.accept(Value::Uint2(8)).unwrap(), Value::Uint2(8));
1361 assert_eq!(ConfigKey::ThreadsFlow.accept(Value::Uint2(16)).unwrap(), Value::Uint2(16));
1362 assert_eq!(ConfigKey::ThreadsTask.accept(Value::Uint2(4)).unwrap(), Value::Uint2(4));
1363 assert_eq!(ConfigKey::ThreadsCompute.accept(Value::Uint2(2)).unwrap(), Value::Uint2(2));
1364 }
1365
1366 #[test]
1367 fn test_threads_reject_int4_for_uint2_key() {
1368 assert!(matches!(ConfigKey::ThreadsTask.accept(Value::Int4(8)), Err(AcceptError::TypeMismatch { .. })));
1370 }
1371
1372 #[test]
1373 fn test_threads_require_restart() {
1374 assert!(ConfigKey::ThreadsAsync.requires_restart());
1375 assert!(ConfigKey::ThreadsCoordination.requires_restart());
1376 assert!(ConfigKey::ThreadsFlow.requires_restart());
1377 assert!(ConfigKey::ThreadsTask.requires_restart());
1378 assert!(ConfigKey::ThreadsCompute.requires_restart());
1379 }
1380
1381 #[test]
1382 fn test_query_row_batch_size_default_is_uint2_128() {
1383 assert_eq!(ConfigKey::QueryRowBatchSize.default_value(), Value::Uint2(128));
1384 }
1385
1386 #[test]
1387 fn test_query_row_batch_size_round_trips_through_display_and_from_str() {
1388 let key: ConfigKey = "QUERY_ROW_BATCH_SIZE".parse().unwrap();
1389 assert_eq!(key, ConfigKey::QueryRowBatchSize);
1390 assert_eq!(format!("{}", ConfigKey::QueryRowBatchSize), "QUERY_ROW_BATCH_SIZE");
1391 }
1392
1393 #[test]
1394 fn test_query_row_batch_size_accept_rejects_zero() {
1395 match ConfigKey::QueryRowBatchSize.accept(Value::Uint2(0)).unwrap_err() {
1396 AcceptError::InvalidValue(reason) => {
1397 assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1398 }
1399 other => panic!("expected InvalidValue, got {other:?}"),
1400 }
1401 }
1402
1403 #[test]
1404 fn test_query_row_batch_size_accept_passes_positive() {
1405 assert_eq!(ConfigKey::QueryRowBatchSize.accept(Value::Uint2(1)).unwrap(), Value::Uint2(1));
1406 assert_eq!(ConfigKey::QueryRowBatchSize.accept(Value::Uint2(1024)).unwrap(), Value::Uint2(1024));
1407 }
1408
1409 #[test]
1410 fn test_query_row_batch_size_rejects_mismatched_type() {
1411 assert!(matches!(
1413 ConfigKey::QueryRowBatchSize.accept(Value::Int4(64)),
1414 Err(AcceptError::TypeMismatch { .. })
1415 ));
1416 assert!(matches!(
1417 ConfigKey::QueryRowBatchSize.accept(Value::Int4(0)),
1418 Err(AcceptError::TypeMismatch { .. })
1419 ));
1420 }
1421
1422 #[test]
1423 fn test_cdc_compact_interval_round_trips_through_display_and_from_str() {
1424 let key: ConfigKey = "CDC_COMPACT_INTERVAL".parse().unwrap();
1425 assert_eq!(key, ConfigKey::CdcCompactInterval);
1426 assert_eq!(format!("{}", ConfigKey::CdcCompactInterval), "CDC_COMPACT_INTERVAL");
1427 }
1428
1429 #[test]
1430 fn test_cdc_compact_block_size_round_trips_through_display_and_from_str() {
1431 let key: ConfigKey = "CDC_COMPACT_BLOCK_SIZE".parse().unwrap();
1432 assert_eq!(key, ConfigKey::CdcCompactBlockSize);
1433 assert_eq!(format!("{}", ConfigKey::CdcCompactBlockSize), "CDC_COMPACT_BLOCK_SIZE");
1434 }
1435
1436 #[test]
1437 fn test_cdc_compact_safety_lag_round_trips_through_display_and_from_str() {
1438 let key: ConfigKey = "CDC_COMPACT_SAFETY_LAG".parse().unwrap();
1439 assert_eq!(key, ConfigKey::CdcCompactSafetyLag);
1440 assert_eq!(format!("{}", ConfigKey::CdcCompactSafetyLag), "CDC_COMPACT_SAFETY_LAG");
1441 }
1442
1443 #[test]
1444 fn test_cdc_compact_max_blocks_per_tick_round_trips_through_display_and_from_str() {
1445 let key: ConfigKey = "CDC_COMPACT_MAX_BLOCKS_PER_TICK".parse().unwrap();
1446 assert_eq!(key, ConfigKey::CdcCompactMaxBlocksPerTick);
1447 assert_eq!(format!("{}", ConfigKey::CdcCompactMaxBlocksPerTick), "CDC_COMPACT_MAX_BLOCKS_PER_TICK");
1448 }
1449
1450 #[test]
1451 fn test_cdc_compact_interval_default_is_duration() {
1452 assert!(matches!(ConfigKey::CdcCompactInterval.default_value(), Value::Duration(_)));
1453 }
1454
1455 #[test]
1456 fn test_cdc_compact_block_size_default_is_uint8_1024() {
1457 assert_eq!(ConfigKey::CdcCompactBlockSize.default_value(), Value::Uint8(1024));
1458 }
1459
1460 #[test]
1461 fn test_cdc_compact_safety_lag_default_is_uint8_1024() {
1462 assert_eq!(ConfigKey::CdcCompactSafetyLag.default_value(), Value::Uint8(1024));
1463 }
1464
1465 #[test]
1466 fn test_cdc_compact_max_blocks_per_tick_default_is_uint8_16() {
1467 assert_eq!(ConfigKey::CdcCompactMaxBlocksPerTick.default_value(), Value::Uint8(16));
1468 }
1469
1470 #[test]
1471 fn test_cdc_compact_interval_accept_passes_positive_duration() {
1472 let one_sec = Value::duration_seconds(1);
1473 assert_eq!(ConfigKey::CdcCompactInterval.accept(one_sec.clone()).unwrap(), one_sec);
1474 }
1475
1476 #[test]
1477 fn test_cdc_compact_interval_accept_rejects_zero() {
1478 let zero = Value::duration_seconds(0);
1479 match ConfigKey::CdcCompactInterval.accept(zero).unwrap_err() {
1480 AcceptError::InvalidValue(reason) => {
1481 assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1482 }
1483 other => panic!("expected InvalidValue, got {other:?}"),
1484 }
1485 }
1486
1487 #[test]
1488 fn test_cdc_compact_interval_accept_rejects_negative() {
1489 let negative = Value::duration_seconds(-5);
1490 assert!(matches!(ConfigKey::CdcCompactInterval.accept(negative), Err(AcceptError::InvalidValue(_))));
1491 }
1492
1493 #[test]
1494 fn test_cdc_compact_block_size_accept_rejects_zero() {
1495 match ConfigKey::CdcCompactBlockSize.accept(Value::Uint8(0)).unwrap_err() {
1496 AcceptError::InvalidValue(reason) => {
1497 assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1498 }
1499 other => panic!("expected InvalidValue, got {other:?}"),
1500 }
1501 }
1502
1503 #[test]
1504 fn test_cdc_compact_block_size_accept_passes_positive() {
1505 assert_eq!(ConfigKey::CdcCompactBlockSize.accept(Value::Uint8(1)).unwrap(), Value::Uint8(1));
1506 assert_eq!(ConfigKey::CdcCompactBlockSize.accept(Value::Uint8(1024)).unwrap(), Value::Uint8(1024));
1507 }
1508
1509 #[test]
1510 fn test_cdc_compact_safety_lag_and_max_blocks_accept_zero() {
1511 assert_eq!(ConfigKey::CdcCompactSafetyLag.accept(Value::Uint8(0)).unwrap(), Value::Uint8(0));
1512 assert_eq!(ConfigKey::CdcCompactMaxBlocksPerTick.accept(Value::Uint8(0)).unwrap(), Value::Uint8(0));
1513 }
1514
1515 #[test]
1516 fn test_accept_rejects_int4_for_uint8_block_size() {
1517 assert!(matches!(
1519 ConfigKey::CdcCompactBlockSize.accept(Value::Int4(1024)),
1520 Err(AcceptError::TypeMismatch { .. })
1521 ));
1522 assert!(matches!(
1523 ConfigKey::CdcCompactBlockSize.accept(Value::Int8(2048)),
1524 Err(AcceptError::TypeMismatch { .. })
1525 ));
1526 }
1527
1528 #[test]
1529 fn test_accept_rejects_zero_of_canonical_type() {
1530 match ConfigKey::CdcCompactBlockSize.accept(Value::Uint8(0)).unwrap_err() {
1531 AcceptError::InvalidValue(reason) => {
1532 assert!(reason.contains("greater than zero"));
1533 }
1534 other => panic!("expected InvalidValue, got {other:?}"),
1535 }
1536 }
1537
1538 #[test]
1539 fn test_accept_rejects_negative_int_for_uint8_key() {
1540 assert!(matches!(
1543 ConfigKey::CdcCompactBlockSize.accept(Value::Int4(-1)),
1544 Err(AcceptError::TypeMismatch { .. })
1545 ));
1546 }
1547
1548 #[test]
1549 fn test_accept_rejects_int_for_duration_key() {
1550 assert!(matches!(
1553 ConfigKey::CdcCompactInterval.accept(Value::Int4(60)),
1554 Err(AcceptError::TypeMismatch { .. })
1555 ));
1556 }
1557
1558 #[test]
1559 fn test_accept_idempotent_on_canonical_uint8() {
1560 let canonical = Value::Uint8(42);
1561 assert_eq!(ConfigKey::OracleWindowSize.accept(canonical.clone()).unwrap(), canonical);
1562 }
1563
1564 #[test]
1565 fn test_accept_idempotent_on_canonical_duration() {
1566 let canonical = Value::duration_seconds(5);
1567 assert_eq!(ConfigKey::CdcCompactInterval.accept(canonical.clone()).unwrap(), canonical);
1568 }
1569
1570 #[test]
1571 fn test_accept_rejects_typed_null_for_non_optional_key() {
1572 let err = ConfigKey::CdcCompactBlockSize
1573 .accept(Value::None {
1574 inner: ValueType::Uint8,
1575 })
1576 .unwrap_err();
1577 assert!(matches!(err, AcceptError::TypeMismatch { .. }));
1578 }
1579
1580 #[test]
1581 fn test_accept_passes_typed_null_for_optional_key() {
1582 let none = Value::None {
1583 inner: ValueType::Duration,
1584 };
1585 assert_eq!(ConfigKey::CdcTtlDuration.accept(none.clone()).unwrap(), none);
1586 }
1587
1588 #[test]
1589 fn test_accept_rejects_wrong_inner_type_typed_null_for_optional_key() {
1590 let err = ConfigKey::CdcTtlDuration
1592 .accept(Value::None {
1593 inner: ValueType::Uint8,
1594 })
1595 .unwrap_err();
1596 assert!(matches!(err, AcceptError::TypeMismatch { .. }));
1597 }
1598
1599 #[test]
1600 fn test_commit_group_linger_default_is_none() {
1601 assert_eq!(
1604 ConfigKey::CommitGroupLinger.default_value(),
1605 Value::None {
1606 inner: ValueType::Duration,
1607 }
1608 );
1609 assert!(ConfigKey::CommitGroupLinger.is_optional());
1610 assert_eq!(ConfigKey::CommitGroupLinger.expected_types(), &[ValueType::Duration]);
1611 }
1612
1613 #[test]
1614 fn test_commit_group_linger_accepts_none_to_disable_grouping() {
1615 let none = Value::None {
1616 inner: ValueType::Duration,
1617 };
1618 assert_eq!(ConfigKey::CommitGroupLinger.accept(none.clone()).unwrap(), none);
1619 }
1620
1621 #[test]
1622 fn test_commit_group_linger_rejects_zero_and_negative() {
1623 match ConfigKey::CommitGroupLinger.accept(Value::duration_seconds(0)).unwrap_err() {
1626 AcceptError::InvalidValue(reason) => {
1627 assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1628 }
1629 other => panic!("expected InvalidValue, got {other:?}"),
1630 }
1631 assert!(matches!(
1632 ConfigKey::CommitGroupLinger.accept(Value::duration_seconds(-5)),
1633 Err(AcceptError::InvalidValue(_))
1634 ));
1635 }
1636
1637 #[test]
1638 fn test_commit_group_linger_requires_restart() {
1639 assert!(ConfigKey::CommitGroupLinger.requires_restart());
1642 }
1643
1644 #[test]
1645 fn test_commit_group_linger_round_trips_through_display_and_from_str() {
1646 assert_eq!("COMMIT_GROUP_LINGER".parse::<ConfigKey>().unwrap(), ConfigKey::CommitGroupLinger);
1647 assert_eq!(format!("{}", ConfigKey::CommitGroupLinger), "COMMIT_GROUP_LINGER");
1648 }
1649
1650 #[test]
1651 fn test_commit_group_max_entries_metadata() {
1652 assert_eq!(ConfigKey::CommitGroupMaxEntries.default_value(), Value::Uint8(256));
1653 assert_eq!(ConfigKey::CommitGroupMaxEntries.expected_types(), &[ValueType::Uint8]);
1654 assert!(!ConfigKey::CommitGroupMaxEntries.is_optional());
1655 assert!(ConfigKey::CommitGroupMaxEntries.requires_restart());
1656 }
1657
1658 #[test]
1659 fn test_commit_group_max_entries_rejects_zero() {
1660 match ConfigKey::CommitGroupMaxEntries.accept(Value::Uint8(0)).unwrap_err() {
1663 AcceptError::InvalidValue(reason) => {
1664 assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1665 }
1666 other => panic!("expected InvalidValue, got {other:?}"),
1667 }
1668 assert_eq!(ConfigKey::CommitGroupMaxEntries.accept(Value::Uint8(1)).unwrap(), Value::Uint8(1));
1669 }
1670
1671 #[test]
1672 fn test_commit_group_max_entries_round_trips_through_display_and_from_str() {
1673 assert_eq!("COMMIT_GROUP_MAX_ENTRIES".parse::<ConfigKey>().unwrap(), ConfigKey::CommitGroupMaxEntries);
1674 assert_eq!(format!("{}", ConfigKey::CommitGroupMaxEntries), "COMMIT_GROUP_MAX_ENTRIES");
1675 }
1676
1677 #[test]
1678 fn test_historical_gc_keys_round_trip() {
1679 assert_eq!("HISTORICAL_GC_BATCH_SIZE".parse::<ConfigKey>().unwrap(), ConfigKey::HistoricalGcBatchSize);
1680 assert_eq!("HISTORICAL_GC_INTERVAL".parse::<ConfigKey>().unwrap(), ConfigKey::HistoricalGcInterval);
1681 assert_eq!(format!("{}", ConfigKey::HistoricalGcBatchSize), "HISTORICAL_GC_BATCH_SIZE");
1682 assert_eq!(format!("{}", ConfigKey::HistoricalGcInterval), "HISTORICAL_GC_INTERVAL");
1683 }
1684
1685 #[test]
1686 fn test_historical_gc_defaults() {
1687 assert_eq!(ConfigKey::HistoricalGcBatchSize.default_value(), Value::Uint8(50_000));
1688 assert!(matches!(ConfigKey::HistoricalGcInterval.default_value(), Value::Duration(_)));
1689 }
1690
1691 #[test]
1692 fn test_historical_gc_batch_size_rejects_zero() {
1693 match ConfigKey::HistoricalGcBatchSize.accept(Value::Uint8(0)).unwrap_err() {
1694 AcceptError::InvalidValue(reason) => {
1695 assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1696 }
1697 other => panic!("expected InvalidValue, got {other:?}"),
1698 }
1699 }
1700
1701 #[test]
1702 fn test_historical_gc_interval_rejects_zero() {
1703 let zero = Value::duration_seconds(0);
1704 match ConfigKey::HistoricalGcInterval.accept(zero).unwrap_err() {
1705 AcceptError::InvalidValue(reason) => {
1706 assert!(reason.contains("greater than zero"), "unexpected reason: {reason}");
1707 }
1708 other => panic!("expected InvalidValue, got {other:?}"),
1709 }
1710 }
1711}