1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
//! Configuration for logdb.
//!
//! All configuration is validated at construction time via [`Config::validate`].
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use crate::error::ConfigError;
use crate::KeyRing;
/// Policy when the ring buffer is full.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QueueFullPolicy {
/// Block (spin + backoff) until a slot becomes available.
Block,
/// Immediately return [`AppendError::QueueFull`](crate::AppendError::QueueFull).
Drop,
}
/// Durability mode for the Committer.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DurabilityMode {
/// `fdatasync` after every commit batch.
Sync,
/// `fdatasync` when the batch size or time threshold is met.
Batch,
/// `fdatasync` only on explicit `flush()` or shutdown.
Async,
}
/// I/O backend for the Committer.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IoBackend {
/// Standard `pwrite` + `fdatasync`.
Pwrite,
// IoUring reserved for future use.
}
/// Retention policy for old segments.
#[derive(Debug, Clone)]
pub enum RetentionPolicy {
/// Keep all segments indefinitely.
KeepAll,
/// Keep at most `max_bytes` of segment data.
MaxBytes(u64),
/// Keep segments younger than `max_age`.
MaxAge(Duration),
}
/// Wait strategy for background thread spinning.
#[derive(Debug, Clone, Copy)]
pub struct WaitStrategy {
/// Number of spin-loop iterations before yielding.
pub spin_count: u32,
/// Number of yields before parking.
pub yield_count: u32,
/// Duration to park the thread.
pub park_duration: Duration,
}
impl Default for WaitStrategy {
fn default() -> Self {
Self {
spin_count: 64,
yield_count: 16,
park_duration: Duration::from_micros(500),
}
}
}
/// Commit trigger thresholds.
#[derive(Debug, Clone, Copy)]
pub struct CommitTrigger {
/// Trigger commit when batched bytes reach this threshold.
pub bytes: usize,
/// Trigger commit when batched record count reaches this threshold.
pub records: usize,
/// Trigger commit when this interval has elapsed since the first pending record.
pub interval: Duration,
/// Durability mode.
pub durability: DurabilityMode,
}
impl Default for CommitTrigger {
fn default() -> Self {
Self {
bytes: 256 * 1024, // 256KB
records: 1024,
interval: Duration::from_millis(10),
durability: DurabilityMode::Batch,
}
}
}
/// Complete configuration for a [`LogDb`](crate::LogDb) instance.
#[derive(Debug, Clone)]
pub struct Config {
/// Directory for segment files and metadata.
pub data_dir: PathBuf,
/// Maximum size of a single segment file before rolling. Default: 256MB.
pub segment_size: u64,
/// Number of slots per ring (must be a power of two, >= 16). Default: 8192.
pub ring_size: usize,
/// Number of independent rings (shards). Default: 1.
pub shards: usize,
/// Maximum content size for a single record. Default: 1MB.
pub max_content_size: usize,
/// Enable SHA256 hash chain integrity protection. Default: false.
pub hash_enabled: bool,
/// Enable streaming zstd compression (requires "compression" feature).
pub compression_enabled: bool,
pub encryption_keys: Option<Arc<KeyRing>>, // Requires "encryption" feature
/// Durability mode. Default: Batch.
pub durability_mode: DurabilityMode,
/// I/O backend. Default: Pwrite.
pub io_backend: IoBackend,
/// Policy when the ring buffer is full. Default: Block.
pub queue_full_policy: QueueFullPolicy,
/// Wait strategy for background thread spinning.
pub wait_strategy: WaitStrategy,
/// Sparse-index stride: index one anchor every `index_stride` records per
/// segment. Smaller → faster point reads (shorter scan from anchor) at the
/// cost of a larger `.idx` file; larger → smaller index, longer read scan.
/// Default 1024 (~0.02% of data). Set lower (e.g. 64–256) for latency-
/// sensitive point reads (KV/etcd-style workloads). Only affects raw
/// (non-compressed, non-encrypted) segments.
pub index_stride: u32,
/// Timeout for `flush()` calls. Default: 30s.
pub flush_timeout: Duration,
/// Retention policy for old segments. Default: KeepAll.
pub retention: RetentionPolicy,
/// Remote endpoint URL for optional push. Default: None.
pub remote_endpoint: Option<String>,
/// Batch size for remote push. Default: 1024.
pub push_batch_size: usize,
/// Pusher progress save interval (in batches). Default: 10.
pub push_progress_interval: u32,
/// Max push retries before giving up. 0 = infinite. Default: 0.
pub push_max_retries: u32,
/// Base retry backoff. Default: 1s, capped at 60s.
pub push_retry_base: Duration,
}
impl Default for Config {
fn default() -> Self {
Self {
data_dir: PathBuf::from("./logdb_data"),
segment_size: 256 * 1024 * 1024, // 256MB
ring_size: 8192,
shards: 1,
max_content_size: 1 * 1024 * 1024, // 1MB
hash_enabled: false,
compression_enabled: false,
encryption_keys: None,
durability_mode: DurabilityMode::Batch,
io_backend: IoBackend::Pwrite,
queue_full_policy: QueueFullPolicy::Block,
wait_strategy: WaitStrategy::default(),
index_stride: 1024,
flush_timeout: Duration::from_secs(30),
retention: RetentionPolicy::KeepAll,
remote_endpoint: None,
push_batch_size: 1024,
push_progress_interval: 10,
push_max_retries: 0,
push_retry_base: Duration::from_secs(1),
}
}
}
impl Config {
/// Validate the configuration.
///
/// Returns `Ok(())` if all constraints are satisfied, or a structured
/// [`ConfigError`] describing the first violation. Callers can match on the
/// variant to react to a specific misconfiguration.
pub fn validate(&self) -> Result<(), ConfigError> {
if !self.ring_size.is_power_of_two() || self.ring_size < 16 {
return Err(ConfigError::InvalidRingSize(self.ring_size));
}
if self.shards < 1 || self.shards > 256 {
return Err(ConfigError::InvalidShardCount(self.shards));
}
if self.segment_size < 1 * 1024 * 1024 {
return Err(ConfigError::SegmentTooSmall(self.segment_size));
}
if self.max_content_size > 64 * 1024 * 1024 {
return Err(ConfigError::ContentTooLarge(self.max_content_size));
}
if self.index_stride == 0 {
return Err(ConfigError::ZeroIndexStride);
}
// Per-shard hash chain: each shard independently sealed with its own
// hash_init and last_hash, so multi-shard hash-chain is supported.
// (The pre-v0.4 single-shard constraint has been removed.)
// Note: no arena_size constraint — ContentArena has been eliminated.
// Content lives in Slot (inline or spill), gated by the single
// consume_watermark. This removes the fragile ring_size * max_content_size
// product constraint that caused panics in v1.4.
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn default_config_validates() {
Config::default().validate().unwrap();
}
#[test]
fn rejects_non_power_of_two_ring() {
let mut c = Config::default();
c.ring_size = 100;
assert!(matches!(
c.validate(),
Err(ConfigError::InvalidRingSize(100))
));
}
#[test]
fn rejects_small_ring() {
let mut c = Config::default();
c.ring_size = 8;
assert!(matches!(c.validate(), Err(ConfigError::InvalidRingSize(8))));
}
#[test]
fn rejects_zero_shards() {
let mut c = Config::default();
c.shards = 0;
assert!(matches!(
c.validate(),
Err(ConfigError::InvalidShardCount(0))
));
}
#[test]
fn rejects_too_many_shards() {
let mut c = Config::default();
c.shards = 257;
assert!(matches!(
c.validate(),
Err(ConfigError::InvalidShardCount(257))
));
}
#[test]
fn rejects_small_segment() {
let mut c = Config::default();
c.segment_size = 512 * 1024; // 512KB
assert!(matches!(c.validate(), Err(ConfigError::SegmentTooSmall(_))));
}
// cr-032 Phase 3: hash-chain + multi-key encryption is now SUPPORTED (the
// chain key is a stable secret masked on disk, independent of the active
// key, so rotation no longer severs it). Phase 1 temporarily rejected this.
#[cfg(all(feature = "hash-chain", feature = "encryption"))]
#[test]
fn allows_hash_chain_with_multi_key_encryption() {
let mut c = Config::default();
c.hash_enabled = true;
c.encryption_keys = Some(crate::KeyRing::new([0xAAu8; 32], vec![[0xBBu8; 32]]));
assert!(c.validate().is_ok(), "multi-key + hash-chain must validate (cr-032 Phase 3)");
}
#[cfg(all(feature = "hash-chain", feature = "encryption"))]
#[test]
fn allows_hash_chain_with_single_key_encryption() {
let mut c = Config::default();
c.hash_enabled = true;
c.encryption_keys = Some(crate::KeyRing::single([0xAAu8; 32]));
assert!(c.validate().is_ok());
}
#[test]
fn rejects_huge_content() {
let mut c = Config::default();
c.max_content_size = 128 * 1024 * 1024; // 128MB
assert!(matches!(c.validate(), Err(ConfigError::ContentTooLarge(_))));
}
#[test]
fn rejects_zero_index_stride() {
let mut c = Config::default();
c.index_stride = 0;
assert!(matches!(c.validate(), Err(ConfigError::ZeroIndexStride)));
}
#[cfg(feature = "hash-chain")]
#[test]
fn allows_hash_chain_with_multiple_shards() {
// Per-shard hash chain: multi-shard is now supported.
let mut c = Config::default();
c.hash_enabled = true;
c.shards = 4;
assert!(c.validate().is_ok());
}
#[cfg(feature = "hash-chain")]
#[test]
fn accepts_hash_chain_with_single_shard() {
let mut c = Config::default();
c.hash_enabled = true;
c.shards = 1;
assert!(c.validate().is_ok());
}
#[test]
fn no_arena_constraint() {
// v1.4 required arena_size >= ring_size * max_content_size.
// v1.0 has no such constraint — this should validate fine.
let mut c = Config::default();
c.ring_size = 8192;
c.max_content_size = 1 * 1024 * 1024; // 1MB
// No arena_size field exists — constraint is gone.
c.validate().unwrap();
}
}