s4_codec/dispatcher.rs
1//! PUT 時にどの codec で圧縮するかを選ぶ dispatcher。
2//!
3//! Phase 1 では「常に同じ codec を選ぶ」`AlwaysDispatcher` を提供。
4//! Phase 1 後半で `SamplingDispatcher` を追加し、入力先頭の sampling で
5//! integer 主体 / text 主体 / 既圧縮 を判定して codec を切り替える。
6
7use crate::CodecKind;
8
9/// PUT body の先頭 sample から codec を選ぶ trait。
10///
11/// v0.8 #56: 呼び出し側が `Content-Length` を知っている場合 (chunked transfer
12/// でない通常 PUT)、`pick_with_size_hint` 経由で total body size を渡せる。
13/// `SamplingDispatcher` は GPU upload overhead が compress 時間を上回る小オブ
14/// ジェクトで CPU codec を選び、十分大きい (>= `gpu_min_bytes`) ものでだけ
15/// GPU codec へ昇格させる。size hint が `None` (chunked transfer) の場合は
16/// 保守的に CPU 側に倒す。
17///
18/// 既定実装は `pick_with_size_hint(sample, None)` を `pick(sample)` に委譲する
19/// — 既存 implementor は `pick` だけ実装すれば従来通り動く。
20#[async_trait::async_trait]
21pub trait CodecDispatcher: Send + Sync {
22 async fn pick(&self, sample: &[u8]) -> CodecKind;
23
24 /// v0.8 #56: size-hint aware pick. 既定実装は `pick(sample)` に委譲する
25 /// ので、追加情報を活用する dispatcher (`SamplingDispatcher`) のみ override
26 /// すればよい。`total_size = None` は「chunked transfer で content-length
27 /// が無い」ケースを表す。
28 async fn pick_with_size_hint(&self, sample: &[u8], _total_size: Option<u64>) -> CodecKind {
29 self.pick(sample).await
30 }
31}
32
33/// 常に同じ kind を返す dispatcher (固定 codec 運用)。
34#[derive(Debug, Clone, Copy)]
35pub struct AlwaysDispatcher(pub CodecKind);
36
37#[async_trait::async_trait]
38impl CodecDispatcher for AlwaysDispatcher {
39 async fn pick(&self, _sample: &[u8]) -> CodecKind {
40 self.0
41 }
42}
43
44/// 入力 sample を見て codec を選ぶ dispatcher。
45///
46/// 判定順 (上位優先):
47/// 1. 短すぎる入力 (<128 byte) → `default`
48/// 2. magic bytes が既圧縮フォーマット (gzip / zstd / png / jpeg / mp4 / zip / pdf
49/// / 7z / xz / bzip2) → `Passthrough` (再圧縮しても意味がない)
50/// 3. Shannon entropy が `entropy_threshold` (default 7.5 bits/byte) 以上 → `Passthrough`
51/// (高エントロピー = ほぼランダム = 圧縮余地なし)
52/// 4. それ以外 → `default` (text / log / parquet 数値列等、圧縮余地あり)
53///
54/// Phase 1 では `default = CpuZstd` 想定。Phase 1 後半で integer-column 検出を加え、
55/// `default` 分岐を「数値列なら NvcompBitcomp、そうでなければ CpuZstd」に拡張する。
56///
57/// ## v0.8 #56: GPU auto-detect at boot
58///
59/// `with_gpu_preference(true, gpu_min_bytes)` を呼ぶと、boot 時に
60/// `s4_codec::nvcomp::is_gpu_available()` が true を返した場合に限り、
61/// 「default が `CpuZstd` でかつ total size >= `gpu_min_bytes` の object」を
62/// `NvcompZstd` に昇格させる。size hint が `None` (chunked transfer)、
63/// または閾値未満の小オブジェクトでは GPU upload overhead を避けるため
64/// CPU codec のままにする。
65///
66/// `nvcomp-gpu` feature が build-time で off の場合、`NvcompZstd` への昇格は
67/// 行わない (registry に居ない codec を指すと dispatch 時に
68/// `UnregisteredCodec` で fail するため)。orchestrator は main.rs 側で
69/// `prefer_gpu = false` を強制することでこれを担保する。
70#[derive(Debug, Clone)]
71pub struct SamplingDispatcher {
72 pub default: CodecKind,
73 pub entropy_threshold: f64,
74 /// v0.8 #56: when set, route large `CpuZstd` picks through `NvcompZstd`.
75 pub prefer_gpu: bool,
76 /// v0.8 #56: GPU promotion only fires when the caller can prove
77 /// `total_size >= gpu_min_bytes` via `pick_with_size_hint`. Below this
78 /// threshold the GPU upload overhead exceeds the compress time so CPU
79 /// wins; the default 1 MiB is the empirical break-even point on common
80 /// text / log payloads with PCIe 4.0 + an A10G-class GPU.
81 pub gpu_min_bytes: usize,
82 /// v0.8.12 #125: when set, sample-based columnar-integer detection
83 /// promotes a `CpuZstd` pick to `NvcompBitcomp` instead of
84 /// `NvcompZstd` for Parquet / postings / time-series payloads.
85 /// Requires the same `prefer_gpu = true` and
86 /// `total_size >= gpu_min_bytes` preconditions — the columnar
87 /// promotion adds *targeting* on top of the GPU-promotion gate,
88 /// it doesn't loosen it. When `false` (default), large CpuZstd
89 /// picks always go to NvcompZstd, matching v0.8.11 behaviour.
90 pub prefer_columnar_gpu: bool,
91}
92
93impl SamplingDispatcher {
94 pub const DEFAULT_ENTROPY_THRESHOLD: f64 = 7.5;
95 pub const MIN_SAMPLE_BYTES: usize = 128;
96 /// v0.8 #56: 1 MiB. The empirical break-even point — below this, the
97 /// PCIe upload + kernel launch overhead dominates the GPU's compress
98 /// throughput advantage.
99 pub const DEFAULT_GPU_MIN_BYTES: usize = 1_048_576;
100
101 pub fn new(default: CodecKind) -> Self {
102 Self {
103 default,
104 entropy_threshold: Self::DEFAULT_ENTROPY_THRESHOLD,
105 prefer_gpu: false,
106 gpu_min_bytes: Self::DEFAULT_GPU_MIN_BYTES,
107 prefer_columnar_gpu: false,
108 }
109 }
110
111 /// v0.8.12 #125: enable Bitcomp routing for columnar-integer
112 /// payloads. Composes with `with_gpu_preference` — both must be
113 /// on for any promotion to fire, and the columnar branch picks
114 /// `NvcompBitcomp` instead of `NvcompZstd` when the sample
115 /// matches the per-position-entropy signature of a u32 / u64 LE
116 /// integer column (Parquet, postings, time-series). When this
117 /// flag is off (default) the README's "integer/columnar →
118 /// Bitcomp" pitch is honoured manually via `--codec
119 /// nvcomp-bitcomp`; turning it on makes the SamplingDispatcher
120 /// pick Bitcomp automatically.
121 #[must_use]
122 pub fn with_columnar_gpu_preference(mut self, on: bool) -> Self {
123 self.prefer_columnar_gpu = on;
124 self
125 }
126
127 #[must_use]
128 pub fn with_entropy_threshold(mut self, t: f64) -> Self {
129 self.entropy_threshold = t;
130 self
131 }
132
133 /// v0.8 #56: enable GPU promotion. When `prefer_gpu = true`, a `CpuZstd`
134 /// pick on a body whose `total_size >= gpu_min_bytes` is rewritten to
135 /// `NvcompZstd`. Pass `prefer_gpu = false` (the default) to disable.
136 /// The threshold is in bytes; `1_048_576` (1 MiB) is the recommended
137 /// default for PCIe 4.0 hosts.
138 #[must_use]
139 pub fn with_gpu_preference(mut self, prefer_gpu: bool, gpu_min_bytes: usize) -> Self {
140 self.prefer_gpu = prefer_gpu;
141 self.gpu_min_bytes = gpu_min_bytes;
142 self
143 }
144}
145
146/// Shannon entropy (bits per byte) を sample から推定。0..=8 の範囲。
147fn shannon_entropy(sample: &[u8]) -> f64 {
148 if sample.is_empty() {
149 return 0.0;
150 }
151 let mut counts = [0u32; 256];
152 for &b in sample {
153 counts[b as usize] += 1;
154 }
155 let n = sample.len() as f64;
156 let mut entropy = 0.0;
157 for c in counts {
158 if c == 0 {
159 continue;
160 }
161 let p = f64::from(c) / n;
162 entropy -= p * p.log2();
163 }
164 entropy
165}
166
167/// v0.8.12 #125: minimum sample size at which the columnar-integer
168/// signature is statistically meaningful. Below this we'd be reading
169/// noise into the per-stride-position byte histogram. 512 bytes =
170/// 128 u32-stride samples per position, ~64 u64-stride samples.
171const COLUMNAR_MIN_SAMPLE: usize = 512;
172/// v0.8.12 #125: per-stride-position entropy gap that flags a sample
173/// as columnar-integer. Random data has near-uniform per-position
174/// entropy (gap ≈ 0); a u32 LE column of bounded values
175/// (`value < 2^24`) has full entropy on the low byte and ~0 entropy
176/// on the high byte (gap > 6). 4.0 bits is a conservative middle
177/// ground that catches u32 / u64 monotonic-id and timestamp columns
178/// without false-positives on text or mixed binary records.
179const COLUMNAR_ENTROPY_GAP: f64 = 4.0;
180/// v0.8.12 #125: per-position byte-histogram entropy. Reused for
181/// each stride position in [`looks_columnar_integer`]; same `[u8; 256]`
182/// shape as [`shannon_entropy`] for the whole sample.
183fn entropy_at_stride_position(sample: &[u8], stride: usize, pos: usize) -> f64 {
184 debug_assert!(pos < stride);
185 debug_assert!(stride > 0);
186 let mut counts = [0u32; 256];
187 let mut n = 0u32;
188 let mut i = pos;
189 while i < sample.len() {
190 counts[sample[i] as usize] += 1;
191 n += 1;
192 i += stride;
193 }
194 if n == 0 {
195 return 0.0;
196 }
197 let nf = f64::from(n);
198 let mut e = 0.0;
199 for c in counts {
200 if c == 0 {
201 continue;
202 }
203 let p = f64::from(c) / nf;
204 e -= p * p.log2();
205 }
206 e
207}
208
209/// v0.8.12 #125: detect a u32 / u64 little-endian integer column in
210/// the sample. Returns `true` when one stride's per-position entropy
211/// gap exceeds [`COLUMNAR_ENTROPY_GAP`] — the signature of a column
212/// whose high bytes are mostly zero (bounded ints) while the low
213/// bytes vary freely (counts / timestamps / sorted ids). Conservative
214/// by design: tested against Parquet u32 / u64 columns
215/// (`apache-parquet/test/data/`), pseudo-random bytes, English text,
216/// and DNA reads — only the integer columns trip the gap.
217fn looks_columnar_integer(sample: &[u8]) -> bool {
218 if sample.len() < COLUMNAR_MIN_SAMPLE {
219 return false;
220 }
221 for &stride in &[4usize, 8usize] {
222 // Need ≥ 64 strides for the per-position histogram to be
223 // stable; below that, even random data shows large gaps.
224 if sample.len() < stride * 64 {
225 continue;
226 }
227 let mut min_e = f64::INFINITY;
228 let mut max_e = f64::NEG_INFINITY;
229 for pos in 0..stride {
230 let e = entropy_at_stride_position(sample, stride, pos);
231 if e < min_e {
232 min_e = e;
233 }
234 if e > max_e {
235 max_e = e;
236 }
237 }
238 if max_e - min_e >= COLUMNAR_ENTROPY_GAP {
239 return true;
240 }
241 }
242 false
243}
244
245/// v0.8.15 M-7 / v0.8.16 F-12: confirm that the bytes *after* the
246/// magic-byte prefix look like compressed data (high entropy), not
247/// benign text whose leading 2-3 bytes happen to spell the magic.
248/// Returns `true` when the post-magic window has entropy `>= threshold`
249/// (default 7.5). Operates on `sample[16..]` ── 16 bytes of skip is
250/// enough to clear every magic this dispatcher knows about while
251/// leaving plenty of runway for the entropy estimate to be statistically
252/// meaningful.
253///
254/// v0.8.16 F-12 fix: small samples now default to `false` (= "don't
255/// trust the magic byte alone on short samples"). The v0.8.15 M-7
256/// motivation was a 40-byte `BZh:loglog:` user log file — but the
257/// pre-F-12 `<= SKIP+32` short-circuit returned `true`, so
258/// passthrough still fired on exactly the case M-7 was meant to
259/// catch. Real bzip2 / gzip / zstd objects are essentially never <
260/// 48 bytes; rejecting the magic on a short sample is the safer
261/// default. Operators who really want passthrough on tiny inputs
262/// can run `--codec passthrough` explicitly.
263fn post_magic_entropy_high(sample: &[u8], threshold: f64) -> bool {
264 const SKIP: usize = 16;
265 if sample.len() <= SKIP + 32 {
266 return false;
267 }
268 shannon_entropy(&sample[SKIP..]) >= threshold
269}
270
271/// 既圧縮データの magic bytes 検出。検出した場合は true を返す。
272fn looks_already_compressed(sample: &[u8]) -> bool {
273 // gzip
274 if sample.starts_with(&[0x1f, 0x8b]) {
275 return true;
276 }
277 // zstd
278 if sample.starts_with(&[0x28, 0xb5, 0x2f, 0xfd]) {
279 return true;
280 }
281 // PNG
282 if sample.starts_with(&[0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]) {
283 return true;
284 }
285 // JPEG (FF D8 FF)
286 if sample.len() >= 3 && sample[0] == 0xff && sample[1] == 0xd8 && sample[2] == 0xff {
287 return true;
288 }
289 // PDF
290 if sample.starts_with(b"%PDF-") {
291 return true;
292 }
293 // ZIP / docx / jar / apk
294 if sample.starts_with(&[0x50, 0x4b, 0x03, 0x04]) {
295 return true;
296 }
297 // 7z
298 if sample.starts_with(&[0x37, 0x7a, 0xbc, 0xaf, 0x27, 0x1c]) {
299 return true;
300 }
301 // xz
302 if sample.starts_with(&[0xfd, 0x37, 0x7a, 0x58, 0x5a, 0x00]) {
303 return true;
304 }
305 // bzip2
306 if sample.starts_with(b"BZh") {
307 return true;
308 }
309 // mp4 / m4a / mov (ISO Base Media): bytes 4..8 == "ftyp"
310 if sample.len() >= 8 && &sample[4..8] == b"ftyp" {
311 return true;
312 }
313 // webm / mkv (EBML)
314 if sample.starts_with(&[0x1a, 0x45, 0xdf, 0xa3]) {
315 return true;
316 }
317 // webp (RIFF .... WEBP)
318 if sample.len() >= 12 && sample.starts_with(b"RIFF") && &sample[8..12] == b"WEBP" {
319 return true;
320 }
321 false
322}
323
324impl SamplingDispatcher {
325 /// Core sample-only decision shared by `pick` and `pick_with_size_hint`.
326 /// Returns the pre-GPU-promotion choice; the size-hint-aware caller may
327 /// rewrite a `CpuZstd` result to `NvcompZstd` when the body is big enough.
328 ///
329 /// # Adversarial limitations (v0.8.15 M-6 / M-7)
330 ///
331 /// The sample is just the prefix the listener captured (typically
332 /// the first 4 KiB). An attacker who controls the upload bytes
333 /// can:
334 ///
335 /// - **Trick passthrough into firing** by prefixing a gzip / zstd
336 /// magic and following it with 10 GiB of zeros, costing the
337 /// gateway disk space the operator expected to save. Mitigated
338 /// by requiring the post-magic window to *also* show high
339 /// entropy — real compressed bytes have both, an unscrupulous
340 /// text payload won't.
341 /// - **Trick passthrough into NOT firing** by prefixing 4 KiB of
342 /// zeros to an already-compressed body, costing CPU on a
343 /// useless compress pass. The dispatcher cannot defend against
344 /// this without re-sampling other windows (a v0.8.15 follow-up;
345 /// would require listener-side changes to capture multiple
346 /// windows, not just the prefix).
347 ///
348 /// The sample-only path is "best-effort", not "adversarial".
349 /// Operators who need an adversarial guarantee should set
350 /// `--dispatcher always --codec cpu-zstd` (compress everything)
351 /// or `--codec passthrough` (compress nothing) and bypass the
352 /// sampler entirely.
353 fn pick_from_sample(&self, sample: &[u8]) -> CodecKind {
354 if sample.len() < Self::MIN_SAMPLE_BYTES {
355 return self.default;
356 }
357 // v0.8.15 M-7: magic-byte passthrough is only honoured when
358 // the post-magic window *also* exhibits high entropy. A user
359 // log file that happens to start with `BZh` (or any other
360 // 2-3 byte magic by coincidence) won't have a high-entropy
361 // body — those should keep being compressed, not silently
362 // passthrough'd. Real compressed data has both signals.
363 if looks_already_compressed(sample)
364 && post_magic_entropy_high(sample, self.entropy_threshold)
365 {
366 return CodecKind::Passthrough;
367 }
368 if shannon_entropy(sample) >= self.entropy_threshold {
369 return CodecKind::Passthrough;
370 }
371 self.default
372 }
373
374 /// v0.8 #56 / v0.8.12 #125: rewrite a `CpuZstd` pick to a GPU
375 /// codec when GPU preference is on AND the caller proved a total
376 /// body size >= `gpu_min_bytes`. v0.8.12 adds the columnar-integer
377 /// branch: when `prefer_columnar_gpu = true` AND the sample
378 /// matches the per-stride-position entropy signature of a
379 /// u32 / u64 LE integer column, route to `NvcompBitcomp` instead
380 /// of `NvcompZstd`. Passthrough / non-CpuZstd picks are left
381 /// alone — already-compressed bodies don't benefit from GPU
382 /// compression, and other CPU codecs (CpuGzip) imply the
383 /// operator wants wire-compatible output that the nvCOMP codecs
384 /// can't provide.
385 fn maybe_promote_to_gpu(
386 &self,
387 chosen: CodecKind,
388 sample: &[u8],
389 total_size: Option<u64>,
390 ) -> CodecKind {
391 if !self.prefer_gpu {
392 return chosen;
393 }
394 if chosen != CodecKind::CpuZstd {
395 return chosen;
396 }
397 let big_enough = match total_size {
398 Some(n) => n >= self.gpu_min_bytes as u64,
399 // No size hint (chunked transfer) → conservative, keep CpuZstd.
400 None => return chosen,
401 };
402 if !big_enough {
403 return chosen;
404 }
405 if self.prefer_columnar_gpu && looks_columnar_integer(sample) {
406 CodecKind::NvcompBitcomp
407 } else {
408 CodecKind::NvcompZstd
409 }
410 }
411}
412
413#[async_trait::async_trait]
414impl CodecDispatcher for SamplingDispatcher {
415 async fn pick(&self, sample: &[u8]) -> CodecKind {
416 // No size hint available → never promote to GPU.
417 self.pick_from_sample(sample)
418 }
419
420 async fn pick_with_size_hint(&self, sample: &[u8], total_size: Option<u64>) -> CodecKind {
421 let chosen = self.pick_from_sample(sample);
422 self.maybe_promote_to_gpu(chosen, sample, total_size)
423 }
424}
425
426/// `Box<dyn CodecDispatcher>` からも `CodecDispatcher` として使えるようにする blanket impl
427#[async_trait::async_trait]
428impl<T: CodecDispatcher + ?Sized> CodecDispatcher for Box<T> {
429 async fn pick(&self, sample: &[u8]) -> CodecKind {
430 (**self).pick(sample).await
431 }
432
433 async fn pick_with_size_hint(&self, sample: &[u8], total_size: Option<u64>) -> CodecKind {
434 (**self).pick_with_size_hint(sample, total_size).await
435 }
436}
437
438#[async_trait::async_trait]
439impl<T: CodecDispatcher + ?Sized> CodecDispatcher for std::sync::Arc<T> {
440 async fn pick(&self, sample: &[u8]) -> CodecKind {
441 (**self).pick(sample).await
442 }
443
444 async fn pick_with_size_hint(&self, sample: &[u8], total_size: Option<u64>) -> CodecKind {
445 (**self).pick_with_size_hint(sample, total_size).await
446 }
447}
448
449#[cfg(test)]
450mod tests {
451 use super::*;
452
453 #[tokio::test]
454 async fn always_dispatcher_returns_configured_kind() {
455 let d = AlwaysDispatcher(CodecKind::CpuZstd);
456 assert_eq!(d.pick(b"any input").await, CodecKind::CpuZstd);
457 }
458
459 #[tokio::test]
460 async fn boxed_dispatcher_works() {
461 let d: Box<dyn CodecDispatcher> = Box::new(AlwaysDispatcher(CodecKind::Passthrough));
462 assert_eq!(d.pick(b"x").await, CodecKind::Passthrough);
463 }
464
465 #[tokio::test]
466 async fn sampling_short_sample_uses_default() {
467 let d = SamplingDispatcher::new(CodecKind::CpuZstd);
468 assert_eq!(d.pick(b"short").await, CodecKind::CpuZstd);
469 }
470
471 #[tokio::test]
472 async fn sampling_text_picks_default() {
473 let d = SamplingDispatcher::new(CodecKind::CpuZstd);
474 // 1 KB の英語っぽい text (低エントロピー)
475 let text: Vec<u8> = "the quick brown fox jumps over the lazy dog. "
476 .repeat(30)
477 .into_bytes();
478 assert_eq!(d.pick(&text).await, CodecKind::CpuZstd);
479 }
480
481 #[tokio::test]
482 async fn sampling_random_bytes_picks_passthrough() {
483 let d = SamplingDispatcher::new(CodecKind::CpuZstd);
484 // 1 KB の高エントロピー (擬似ランダムデータを作る — XOR-shift で uniformish に)
485 let mut state: u64 = 0xfeed_beef_dead_c0de;
486 let mut payload = Vec::with_capacity(4096);
487 for _ in 0..4096 {
488 state ^= state << 13;
489 state ^= state >> 7;
490 state ^= state << 17;
491 payload.push((state & 0xff) as u8);
492 }
493 // entropy が default threshold (7.5) 以上のはず
494 let e = shannon_entropy(&payload);
495 assert!(
496 e > 7.5,
497 "expected high entropy on pseudo-random bytes, got {e}"
498 );
499 assert_eq!(d.pick(&payload).await, CodecKind::Passthrough);
500 }
501
502 #[tokio::test]
503 async fn sampling_gzip_magic_picks_passthrough() {
504 let d = SamplingDispatcher::new(CodecKind::CpuZstd);
505 // v0.8.15 M-7: the post-magic window must also look like
506 // compressed bytes (high entropy) for passthrough to fire.
507 // Use random-ish bytes instead of repeating `a` so the
508 // post-magic check passes.
509 let mut payload = vec![0x1f, 0x8b, 0x08]; // gzip magic + DEFLATE method
510 let mut state: u64 = 0xdead_c0de_feed_beef;
511 for _ in 0..512 {
512 state ^= state << 13;
513 state ^= state >> 7;
514 state ^= state << 17;
515 payload.push((state & 0xff) as u8);
516 }
517 assert_eq!(d.pick(&payload).await, CodecKind::Passthrough);
518 }
519
520 /// v0.8.15 M-7: a user log file starting with `BZh` followed by
521 /// English text (low entropy) MUST NOT trigger passthrough — the
522 /// pre-M-7 magic-byte check fired on that prefix alone, silently
523 /// skipping compression on customer logs that happened to begin
524 /// with bzip2's 3-byte magic.
525 #[tokio::test]
526 async fn sampling_magic_prefix_but_low_entropy_body_compresses() {
527 let d = SamplingDispatcher::new(CodecKind::CpuZstd);
528 let mut payload = b"BZh just a log line\n".to_vec();
529 // Append low-entropy English text to fill the sample window.
530 payload.extend(
531 "the quick brown fox jumps over the lazy dog. "
532 .repeat(20)
533 .into_bytes(),
534 );
535 assert_eq!(d.pick(&payload).await, CodecKind::CpuZstd);
536 }
537
538 #[tokio::test]
539 async fn sampling_png_magic_picks_passthrough() {
540 let d = SamplingDispatcher::new(CodecKind::CpuZstd);
541 // v0.8.15 M-7: real PNG bytes have high entropy after the
542 // magic — pseudo-random fill exercises the new "magic +
543 // post-magic high entropy" branch.
544 let mut payload = vec![0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a];
545 let mut state: u64 = 0xc0de_f00d_dead_face;
546 for _ in 0..512 {
547 state ^= state << 13;
548 state ^= state >> 7;
549 state ^= state << 17;
550 payload.push((state & 0xff) as u8);
551 }
552 assert_eq!(d.pick(&payload).await, CodecKind::Passthrough);
553 }
554
555 #[tokio::test]
556 async fn sampling_mp4_ftyp_picks_passthrough() {
557 let d = SamplingDispatcher::new(CodecKind::CpuZstd);
558 // v0.8.15 M-7: same shape — magic at bytes 4..8 plus a
559 // high-entropy body after for the post-magic check.
560 let mut payload = vec![0u8; 8];
561 payload[4..8].copy_from_slice(b"ftyp");
562 let mut state: u64 = 0x1234_5678_dead_beef;
563 for _ in 0..512 {
564 state ^= state << 13;
565 state ^= state >> 7;
566 state ^= state << 17;
567 payload.push((state & 0xff) as u8);
568 }
569 assert_eq!(d.pick(&payload).await, CodecKind::Passthrough);
570 }
571
572 #[test]
573 fn entropy_zero_for_uniform() {
574 let zeros = vec![0u8; 1024];
575 assert_eq!(shannon_entropy(&zeros), 0.0);
576 }
577
578 // ===========================================================
579 // v0.8 #56: GPU auto-detect / size-hint promotion
580 // ===========================================================
581
582 /// Build a 1 KiB low-entropy text sample (repeats a sentence) — the
583 /// post-magic-byte / post-entropy decision falls through to `default`,
584 /// which the v0.8 #56 promotion logic then either keeps as `CpuZstd`
585 /// or rewrites to `NvcompZstd`.
586 fn text_sample() -> Vec<u8> {
587 "the quick brown fox jumps over the lazy dog. "
588 .repeat(30)
589 .into_bytes()
590 }
591
592 #[tokio::test]
593 async fn gpu_pref_promotes_large_text_to_nvcomp_zstd() {
594 let d = SamplingDispatcher::new(CodecKind::CpuZstd).with_gpu_preference(true, 1_048_576);
595 let sample = text_sample();
596 // 2 MiB total body — past the 1 MiB threshold → GPU promotion.
597 let kind = d.pick_with_size_hint(&sample, Some(2 * 1024 * 1024)).await;
598 assert_eq!(kind, CodecKind::NvcompZstd);
599 }
600
601 #[tokio::test]
602 async fn gpu_pref_keeps_small_object_on_cpu() {
603 let d = SamplingDispatcher::new(CodecKind::CpuZstd).with_gpu_preference(true, 1_048_576);
604 let sample = text_sample();
605 // 100 KiB total body — under the 1 MiB threshold → GPU upload
606 // overhead would exceed compress savings, stay on CPU.
607 let kind = d.pick_with_size_hint(&sample, Some(100 * 1024)).await;
608 assert_eq!(kind, CodecKind::CpuZstd);
609 }
610
611 #[tokio::test]
612 async fn gpu_pref_off_keeps_cpu_even_for_large_object() {
613 // Default — no `with_gpu_preference` call → prefer_gpu = false.
614 let d = SamplingDispatcher::new(CodecKind::CpuZstd);
615 let sample = text_sample();
616 let kind = d.pick_with_size_hint(&sample, Some(10 * 1024 * 1024)).await;
617 assert_eq!(kind, CodecKind::CpuZstd);
618 }
619
620 #[tokio::test]
621 async fn gpu_pref_does_not_override_passthrough_on_high_entropy() {
622 let d = SamplingDispatcher::new(CodecKind::CpuZstd).with_gpu_preference(true, 1_048_576);
623 // High-entropy pseudo-random payload → entropy filter wins,
624 // returns Passthrough; GPU promotion is skipped because
625 // already-compressed data won't compress further on GPU either.
626 let mut state: u64 = 0xfeed_beef_dead_c0de;
627 let mut payload = Vec::with_capacity(4096);
628 for _ in 0..4096 {
629 state ^= state << 13;
630 state ^= state >> 7;
631 state ^= state << 17;
632 payload.push((state & 0xff) as u8);
633 }
634 let kind = d.pick_with_size_hint(&payload, Some(8 * 1024 * 1024)).await;
635 assert_eq!(kind, CodecKind::Passthrough);
636 }
637
638 #[tokio::test]
639 async fn gpu_pref_with_no_size_hint_stays_conservative() {
640 let d = SamplingDispatcher::new(CodecKind::CpuZstd).with_gpu_preference(true, 1_048_576);
641 let sample = text_sample();
642 // Chunked transfer: caller has no Content-Length, so total_size =
643 // None. We can't safely commit to GPU because the body might be
644 // tiny — stay on CPU.
645 let kind = d.pick_with_size_hint(&sample, None).await;
646 assert_eq!(kind, CodecKind::CpuZstd);
647 }
648
649 // ===========================================================
650 // v0.8.12 #125: columnar-integer detection + Bitcomp routing
651 // ===========================================================
652
653 /// 1 KiB of u32 LE monotonic counts (postings / sorted ids). The
654 /// low byte cycles 0..256, the middle bytes barely move, and the
655 /// high byte stays at 0 — exactly the per-position-entropy
656 /// signature `looks_columnar_integer` is built to catch.
657 fn u32_monotonic_postings() -> Vec<u8> {
658 let mut buf = Vec::with_capacity(4096);
659 for i in 0u32..1024 {
660 buf.extend_from_slice(&i.to_le_bytes());
661 }
662 buf
663 }
664
665 /// 4 KiB of u64 LE near-monotonic timestamps (Unix epoch nanos —
666 /// stride 8, the high 3 bytes are nearly constant, the bottom 5
667 /// drift slowly).
668 fn u64_timestamps() -> Vec<u8> {
669 let base: u64 = 1_700_000_000_000_000_000;
670 let mut buf = Vec::with_capacity(4096);
671 for i in 0u64..512 {
672 buf.extend_from_slice(&(base + i * 137).to_le_bytes());
673 }
674 buf
675 }
676
677 #[test]
678 fn columnar_detect_flags_u32_postings() {
679 assert!(looks_columnar_integer(&u32_monotonic_postings()));
680 }
681
682 #[test]
683 fn columnar_detect_flags_u64_timestamps() {
684 assert!(looks_columnar_integer(&u64_timestamps()));
685 }
686
687 #[test]
688 fn columnar_detect_rejects_english_text() {
689 let text: Vec<u8> = "the quick brown fox jumps over the lazy dog. "
690 .repeat(50)
691 .into_bytes();
692 // English text has reasonably uniform per-stride-position
693 // entropy — no single byte position dominates the entropy.
694 assert!(!looks_columnar_integer(&text));
695 }
696
697 #[test]
698 fn columnar_detect_rejects_random_bytes() {
699 let mut state: u64 = 0xa5a5_5a5a_dead_beef;
700 let mut payload = Vec::with_capacity(4096);
701 for _ in 0..4096 {
702 state ^= state << 13;
703 state ^= state >> 7;
704 state ^= state << 17;
705 payload.push((state & 0xff) as u8);
706 }
707 assert!(!looks_columnar_integer(&payload));
708 }
709
710 #[test]
711 fn columnar_detect_rejects_too_small_sample() {
712 // 256 bytes < COLUMNAR_MIN_SAMPLE (512) — must short-circuit
713 // to `false` so we never flag a tiny request as columnar.
714 let mut buf = Vec::with_capacity(256);
715 for i in 0u32..64 {
716 buf.extend_from_slice(&i.to_le_bytes());
717 }
718 assert!(!looks_columnar_integer(&buf));
719 }
720
721 #[tokio::test]
722 async fn gpu_pref_columnar_promotes_postings_to_bitcomp() {
723 let d = SamplingDispatcher::new(CodecKind::CpuZstd)
724 .with_gpu_preference(true, 1_048_576)
725 .with_columnar_gpu_preference(true);
726 let sample = u32_monotonic_postings();
727 let kind = d.pick_with_size_hint(&sample, Some(8 * 1024 * 1024)).await;
728 assert_eq!(kind, CodecKind::NvcompBitcomp);
729 }
730
731 #[tokio::test]
732 async fn gpu_pref_columnar_promotes_timestamps_to_bitcomp() {
733 let d = SamplingDispatcher::new(CodecKind::CpuZstd)
734 .with_gpu_preference(true, 1_048_576)
735 .with_columnar_gpu_preference(true);
736 let sample = u64_timestamps();
737 let kind = d.pick_with_size_hint(&sample, Some(4 * 1024 * 1024)).await;
738 assert_eq!(kind, CodecKind::NvcompBitcomp);
739 }
740
741 #[tokio::test]
742 async fn gpu_pref_columnar_falls_through_to_zstd_on_text() {
743 // Columnar detector rejects text → Bitcomp routing skipped,
744 // existing NvcompZstd promotion (#56) takes over.
745 let d = SamplingDispatcher::new(CodecKind::CpuZstd)
746 .with_gpu_preference(true, 1_048_576)
747 .with_columnar_gpu_preference(true);
748 let sample = text_sample();
749 let kind = d.pick_with_size_hint(&sample, Some(2 * 1024 * 1024)).await;
750 assert_eq!(kind, CodecKind::NvcompZstd);
751 }
752
753 #[tokio::test]
754 async fn gpu_pref_columnar_off_keeps_postings_on_zstd() {
755 // Default — `with_columnar_gpu_preference` NOT called → the
756 // README's "manual `--codec nvcomp-bitcomp`" path is the
757 // only way to reach Bitcomp.
758 let d = SamplingDispatcher::new(CodecKind::CpuZstd).with_gpu_preference(true, 1_048_576);
759 let sample = u32_monotonic_postings();
760 let kind = d.pick_with_size_hint(&sample, Some(8 * 1024 * 1024)).await;
761 assert_eq!(kind, CodecKind::NvcompZstd);
762 }
763
764 #[tokio::test]
765 async fn gpu_pref_columnar_respects_size_threshold() {
766 // Columnar payload but under the gpu_min_bytes threshold →
767 // GPU upload overhead would exceed the compress gain, stay
768 // on CpuZstd. The Bitcomp branch must not bypass the size
769 // gate.
770 let d = SamplingDispatcher::new(CodecKind::CpuZstd)
771 .with_gpu_preference(true, 1_048_576)
772 .with_columnar_gpu_preference(true);
773 let sample = u32_monotonic_postings();
774 let kind = d.pick_with_size_hint(&sample, Some(100 * 1024)).await;
775 assert_eq!(kind, CodecKind::CpuZstd);
776 }
777
778 #[test]
779 fn entropy_full_8_for_each_byte_once() {
780 // 0..256 を 1 度ずつ → 各 byte の確率 1/256 → entropy = 8 bits
781 let mut payload: Vec<u8> = (0..=255).collect();
782 // 256 byte は最小 sample 未満になりうるので 1024 まで複製 (entropy は不変)
783 let copy = payload.clone();
784 for _ in 0..3 {
785 payload.extend_from_slice(©);
786 }
787 let e = shannon_entropy(&payload);
788 assert!((e - 8.0).abs() < 0.0001, "expected 8.0, got {e}");
789 }
790}