diff --git a/.cargo/config.toml b/.cargo/config.toml index 172a1634..5603a56d 100644 --- a/.cargo/config.toml +++ b/.cargo/config.toml @@ -1,17 +1,8 @@ -# Cargo build configuration for dfe-loader -# -# Target platforms: Linux x86_64 and aarch64 only -# SIMD optimizations enabled for sonic-rs - -# ============================================================================= -# Per-target CPU optimizations -# ============================================================================= -# x86_64: AVX2 + SSE4.2 (available on all modern Intel/AMD since ~2013) -# aarch64: NEON (baseline for all ARM64) +# SIMD for sonic-rs: x86-64-v3 (AVX2) on x86_64, NEON is baseline on aarch64. [target.x86_64-unknown-linux-gnu] -rustflags = ["-C", "target-cpu=x86-64-v3", "-C", "link-arg=-fuse-ld=mold"] # AVX2, BMI1/2, FMA (Haswell+) +rustflags = ["-C", "target-cpu=x86-64-v3", "-C", "link-arg=-fuse-ld=mold"] [target.aarch64-unknown-linux-gnu] linker = "aarch64-linux-gnu-gcc" -rustflags = ["-C", "target-cpu=generic"] # NEON is baseline for ARM64 +rustflags = ["-C", "target-cpu=generic"] diff --git a/.config/nextest.toml b/.config/nextest.toml index 2aa23e91..e3fcf06b 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -17,19 +17,14 @@ not ( or test(/^integration::container_hygiene::(a_started_container_carries_the_name_and_label_then_goes_away|the_shared_container_is_named_and_labelled_for_the_run|the_watcher_removes_the_container_once_its_owner_exits|a_sweep_leaves_another_owners_live_container_and_takes_a_lapsed_one)$/) ) ''' -# No retries, deliberately. Retries here previously masked three real defects: -# an amd64-only Kafka image running under emulation, a broker probed before it -# served consumers, and tests asserting a cold consumer's first poll is clean. -# A retry would have kept all three green and unfixed. +# No retries: a retry keeps a real defect green. retries = 0 # These tests are legitimately slower than unit tests; warn rather than kill, # but do not let a hung container run forever. slow-timeout = { period = "60s", terminate-after = 5 } -# Tests that start a container of their own get their own concurrency budget: -# each waits for a cold server or broker, and letting all of them start at once -# starves each other of CPU and pushes them past the readiness timeout. The -# shared instances start once per run, so tests on them stay fully parallel. +# Tests that start their own container get a concurrency budget, or cold starts +# starve each other past the readiness timeout. [test-groups] own-containers = { max-threads = 2 } diff --git a/.gitleaks.toml b/.gitleaks.toml index ba9c9560..477d84f6 100644 --- a/.gitleaks.toml +++ b/.gitleaks.toml @@ -1,29 +1,4 @@ -# Project-specific gitleaks allowlist -# Extends the CI base configuration (ci/.gitleaks.toml) -# -# The CI runs gitleaks to detect secrets in your code. This file lets you -# exclude false positives specific to your project. -# -# Documentation: https://github.com/gitleaks/gitleaks#configuration +# Project gitleaks allowlist: https://github.com/gitleaks/gitleaks#configuration [extend] useDefault = true - -# Example allowlist - uncomment [[allowlist]] blocks as needed -# See: https://github.com/gitleaks/gitleaks#configuration - -# [[allowlist]] -# description = "Test files with fake tokens" -# paths = ['''tests/.*\.rs''', '''tests/fixtures/.*'''] - -# [[allowlist]] -# description = "Example configuration files" -# paths = ['''.env.example''', '''.env.sample'''] - -# [[allowlist]] -# description = "Known false positive patterns" -# regexes = ['''YOUR_.*_TOKEN''', '''test-token'''] - -# [[allowlist]] -# description = "Common false positive stopwords" -# stopwords = ["example", "sample", "test", "mock", "fake", "dummy", "placeholder"] diff --git a/.hyperi-ci.yaml b/.hyperi-ci.yaml index 8f89a929..9cb093e9 100644 --- a/.hyperi-ci.yaml +++ b/.hyperi-ci.yaml @@ -6,44 +6,28 @@ # Copyright: (c) 2026 HYPERI PTY LIMITED classification: product -# Project language (auto-detected if not specified) -# Values: python, typescript, golang, rust, bash, none language: rust # Quality checks quality: enabled: true - # Advisory ignores fed to cargo-audit and osv-scanner. cargo-deny reads - # deny.toml separately; osv-scanner does NOT, so the paste unmaintained - # advisory must be ignored here too or the release Quality stage fails. + # osv-scanner does not read deny.toml, so its ignores repeat here. ignore: - tool: osv-scanner id: RUSTSEC-2024-0436 reason: "paste (dtolnay) unmaintained, no replacement; transitive dep. Mirrors the deny.toml ignore. Drop when a maintained fork lands." - # GHSA-w9wp-h8wv-79jx / CVE-2026-48504: opentelemetry_sdk 0.31.0 is in the - # lock through scalo, in the affected range (<= 0.32.0, fixed in 0.32.1). - # The flaw is in BaggagePropagator::extract_with_context, which parses an - # inbound W3C Baggage header before enforcing its size limits. It is - # unreachable: no Baggage propagator is registered in dfe-loader or in scalo, - # so the function is never called. Accepted as not reachable in dfe-infra - # docs/THREAT-MODEL.md, accepted risks. scalo cannot move to 0.32 while - # metrics-exporter-opentelemetry 0.2.1 requires opentelemetry ^0.31; - # hyperi-io/scalo-rs#281 tracks the move. A GitHub advisory with no RustSec - # entry, so only osv-scanner sees it and deny.toml needs no matching ignore. + # No RustSec entry, so deny.toml needs no matching ignore. - tool: osv-scanner id: GHSA-w9wp-h8wv-79jx reason: "CVE-2026-48504, opentelemetry_sdk 0.31.0 via scalo: unreachable, no Baggage propagator is registered in dfe-loader or scalo. Accepted risk in dfe-infra docs/THREAT-MODEL.md. Blocked on metrics-exporter-opentelemetry 0.2.1 requiring opentelemetry ^0.31; tracked in hyperi-io/scalo-rs#281." # Rust: test specific feature sets (avoids --all-features conflicts). - # DFE policy 2026-04-17 — jemalloc only; mimalloc removed. rust: features: - - "default" # Default features - - "default,jemalloc" # With jemalloc allocator - - "default,testcontainers" # With testcontainers integration tests - # Required: transport-memory gates MemoryTransportAdapter, so without - # this set the transport-adapter suite is compiled out rather than run. - # helm_contract.rs asserts it stays listed. - - "default,transport-memory" # With the in-memory transport adapter + - "default" + - "default,jemalloc" + - "default,testcontainers" + # Without it the transport-adapter suite is compiled out. helm_contract.rs asserts it stays. + - "default,transport-memory" # Testing test: @@ -56,7 +40,6 @@ test: # matrix and the peer tools. tests/TESTING.md says which tests are which. full: required_for_release: true - # min_coverage: 80 # Fail if coverage below this percentage # Building build: diff --git a/BENCHMARKS.md b/BENCHMARKS.md index 2cd59be0..83449beb 100644 --- a/BENCHMARKS.md +++ b/BENCHMARKS.md @@ -178,7 +178,7 @@ open target/criterion/report/index.html Mison-style structural indexing for schema-guided extraction. Based on the VLDB 2017 paper. -### Current Benchmarks (v0.11.0) +### Benchmarks at v0.11.0 | Benchmark | Mison | Traditional | Ratio | |-----------|-------|-------------|-------| diff --git a/Cargo.toml b/Cargo.toml index b52a99d1..bc05d03d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,10 +17,7 @@ path = "src/lib.rs" name = "dfe-loader" path = "src/main.rs" -# PGO workload driver — Kafka producer that drives realistic load against a -# running dfe-loader during hyperi-ci Tier 2 PGO/BOLT builds. -# `required-features` keeps it out of default builds; hyperi-ci ≥v1.9.2 -# respects this and skips the binary in detect-build-config. +# PGO/BOLT workload driver, kept out of default builds by `required-features`. [[bin]] name = "pgo-driver" path = "src/bin/pgo_driver.rs" @@ -31,24 +28,18 @@ required-features = ["pgo-driver"] tokio = { version = ">=1.54.1, <2", features = ["full"] } tokio-util = { version = ">=0.7.20, <0.8", features = ["rt"] } -# Shared library with transport abstraction. The floor moves with -# `release.helm.library` in .hyperi-ci.yaml: the library renders only the -# contract version its scalo release writes. +# The floor moves with `release.helm.library` in .hyperi-ci.yaml. scalo = { version = ">=2.14.29, <3", features = ["config-schema", "transport-kafka", "transport-grpc", "dlq-kafka", "config", "config-reload", "deployment", "version-check", "scaling", "cli-service", "top", "logger", "metrics", "service-metrics", "expression", "memory", "worker-batch", "governor", "secrets", "sink-stack", "geoip-download"] } # Steps scalo's jittered retry schedule (sink_stack backoff). Range MUST track # scalo's: BackoffBuilder only applies to the ExponentialBuilder scalo returns. backon = { version = ">=1.6.0, <2", default-features = false, features = ["std"] } -# JSON Schema derivation for the reflectable config-schema capability -# (scalo-rs#6). Same version range as scalo so cargo unifies to one schemars. +# Config-schema derivation. Same range as scalo so cargo unifies to one schemars. schemars = { version = ">=1.2.3, <2", default-features = false, features = ["std", "derive", "preserve_order", "indexmap2"] } -# CEL expression evaluation (same crate scalo uses, needed for Program type). -# Range MUST track scalo's, which is `>=0.14.5, <0.15`: we hand cel::Program across -# the scalo boundary, so a wider range here lets cargo resolve two semver- -# incompatible cels and Program stops being the same type. In ignoreDeps so it -# moves only with scalo. +# Range MUST track scalo's: cel::Program crosses the scalo boundary, and two +# semver-incompatible cels make it two types. In ignoreDeps, so it moves with scalo. cel = { version = ">=0.14.5, <0.15", features = ["json"] } # JSON parsing (SIMD-accelerated) @@ -68,9 +59,7 @@ thiserror = ">=2.0.21, <3" tracing = ">=0.1.44, <0.2" metrics = ">=0.24.6, <0.25" -# Memory management - jemalloc allocator (DFE policy 2026-04-17: jemalloc only) -# `stats` builds jemalloc with --enable-stats; without it there is no -# stats.allocated for the memory guard's heap source to read (#162). +# `stats` gives the memory guard's heap source its stats.allocated. tikv-jemalloc-ctl = { version = ">=0.7.0, <0.8", optional = true, features = ["stats"] } tikv-jemallocator = { version = ">=0.7.0, <0.8", optional = true } @@ -106,11 +95,7 @@ parking_lot = ">=0.13.0, <0.14" crossbeam-channel = ">=0.5.17, <0.6" indexmap = { version = ">=2.14.2, <3", features = ["serde"] } -# File watching (config hot-reload) - using polling for S3/NFS/FUSE compatibility -# notify = ">=6.1" # Not needed - polling works on all filesystems - -# PGO workload driver — Kafka producer for hyperi-ci Tier 2 PGO/BOLT builds. -# Optional, gated by the `pgo-driver` feature. Main loader binary doesn't link it. +# The PGO workload driver's producer, behind the `pgo-driver` feature. rdkafka = { version = ">=0.39.0, <0.40", features = ["dynamic-linking"], optional = true } [dev-dependencies] @@ -123,9 +108,7 @@ test-case = ">=3.4.1, <4" # Held on 0.12: 0.13 drops the rustls-tls-native-roots feature. reqwest = { version = ">=0.12, <0.13", default-features = false, features = ["blocking", "rustls-tls-native-roots"] } -# The Kafka broker image pin and the TLS / mTLS / SASL_SSL broker fixture the -# Kafka tests share, so neither is copied here. The production dependency above -# does not carry this feature. +# The Kafka tests' broker image pin and TLS/SASL broker fixture. scalo = { version = ">=2.14.29, <3", features = ["transport-kafka-test-support"] } # The status code a raw Push is refused with, through scalo's generated gRPC @@ -186,37 +169,11 @@ name = "json_depth" harness = false [profile.release] -lto = "thin" # Link-time optimization (use "fat" for max perf, slower compile) -codegen-units = 1 # Single codegen unit for better optimization -panic = "abort" # Smaller binary, no unwinding overhead -strip = true # Remove debug symbols -opt-level = 3 # Maximum optimization (default for release, explicit for clarity) - -# ============================================================================= -# Advanced Optimization Options (Manual) -# ============================================================================= -# -# 1. Profile-Guided Optimization (PGO) - 10-20% improvement: -# cargo install cargo-pgo -# cargo pgo build -# ./target/x86_64-unknown-linux-gnu/release/dfe-loader # Run with real workload -# cargo pgo optimize -# -# 2. BOLT Post-Link Optimization (additional 5-15% on top of PGO): -# cargo pgo bolt build -# ./target/x86_64-unknown-linux-gnu/release/dfe-loader-bolt-instrumented # Run workload -# cargo pgo bolt optimize -# -# 3. Combined PGO + BOLT: -# cargo pgo optimize -# cargo pgo bolt optimize --with-pgo -# -# 4. Fat LTO (slower compile, sometimes faster runtime): -# Change lto = "thin" to lto = "fat" -# -# References: -# - https://github.com/Kobzol/cargo-pgo -# - https://nnethercote.github.io/perf-book/build-configuration.html +lto = "thin" +codegen-units = 1 +panic = "abort" +strip = true +opt-level = 3 [profile.bench] inherits = "release" @@ -229,13 +186,10 @@ strip = false [lints.rust] # Tests require unsafe for env::set_var/remove_var (edition 2024). -# Production code has zero unsafe blocks — the lib target is clean. unsafe_code = "allow" -# `#[cfg(tarpaulin)]` is set by cargo-tarpaulin at coverage time. Declaring -# it here silences `unexpected_cfgs` without a build.rs. +# cargo-tarpaulin sets `cfg(tarpaulin)`. unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tarpaulin)'] } -# Some clippy allows below are 1.95+-only (e.g. `duration_suboptimal_units`). -# 1.94 toolchains would otherwise fail with `unknown lint`. +# Some clippy allows below name lints an older toolchain does not know. unknown_lints = "allow" [lints.clippy] @@ -247,40 +201,27 @@ dbg_macro = "deny" module_name_repetitions = "allow" # #[must_use] goes where dropping the value is a bug, not on every pure function. must_use_candidate = "allow" -# 117 doc comments name ClickHouse types and columns without backticks. +# Doc comments name ClickHouse types and columns without backticks. doc_markdown = "allow" -# 41 public Result fns have no # Errors section, and 4 no # Panics section. missing_errors_doc = "allow" missing_panics_doc = "allow" -# 16 sites. items_after_statements = "allow" -# 13 methods would become associated functions, which rewrites every caller. +# Turning a method into an associated function rewrites every caller. unused_self = "allow" -# 10 sites, and changing a signature rewrites its callers. needless_pass_by_value = "allow" -# 9 sites. unreadable_literal = "allow" -# 8 sites. doc_link_with_quotes = "allow" -# 7 sites. match_same_arms = "allow" -# 5 sites. struct_excessive_bools = "allow" -# 4 functions run long, the largest at 665 lines, and splitting them needs its own change. too_many_lines = "allow" -# 4 sites, and boxing a future changes the hot path, which needs a measurement first. +# Boxing a future changes the hot path, which needs a measurement first. large_futures = "allow" -# 3 sites. similar_names = "allow" -# 3 sites. trivially_copy_pass_by_ref = "allow" -# 2 sites. unused_async = "allow" -# 20 sites, all in tests/. +# The next three are tests only. manual_let_else = "allow" -# 8 sites, all in tests/. default_trait_access = "allow" -# 7 sites, all in tests/. format_push_string = "allow" # Dynamic Map patterns need flexible casting cast_possible_truncation = "allow" @@ -288,8 +229,7 @@ cast_sign_loss = "allow" cast_precision_loss = "allow" cast_lossless = "allow" cast_possible_wrap = "allow" -# Tests use `let mut cfg = X::default(); cfg.field = ...` for readability over -# struct-init. Production code is unaffected. +# Tests set fields after `X::default()` for readability. field_reassign_with_default = "allow" # Test data sometimes uses literal 3.14 etc. for clarity over `std::f64::consts::PI`. approx_constant = "allow" @@ -297,24 +237,19 @@ approx_constant = "allow" too_many_arguments = "allow" # Mostly affects internal types in benches/tests where TupleX is fine. type_complexity = "allow" -# `&Option` appears in a few config setters where call-site ergonomics -# (`set_x(&self.x)`) outweigh the API guideline. +# Config setters take `&Option` so call sites read `set_x(&self.x)`. ref_option = "allow" -# Tests use `Duration::from_millis(30000)` etc. for symmetry with config -# values stored in millis. Style-only. +# Tests write durations in millis, as config stores them. duration_suboptimal_units = "allow" -# `.map(|x| ...).unwrap_or(false)` is clearer than `.is_ok_and()` for some -# readers. Stylistic. map_unwrap_or = "allow" [features] default = [] -# Allocator (DFE policy 2026-04-17: jemalloc at every channel, no mimalloc) +# Allocator jemalloc = ["dep:tikv-jemallocator", "dep:tikv-jemalloc-ctl"] # Transport backends transport-memory = ["scalo/transport-memory"] # Memory transport for unit tests # Testing testcontainers = [] # The tests that run real ClickHouse and Kafka containers (Docker required) -# PGO workload driver (hyperi-ci Tier 2 — see scripts/pgo-workload.sh) -# Pulls in rdkafka for the producer binary; main loader binary is unaffected. +# PGO workload driver (scripts/pgo-workload.sh). Only that binary links rdkafka. pgo-driver = ["dep:rdkafka"] diff --git a/docker-compose.dev.yaml b/docker-compose.dev.yaml index 68ee1a0e..91339d89 100644 --- a/docker-compose.dev.yaml +++ b/docker-compose.dev.yaml @@ -12,8 +12,7 @@ services: # ClickHouse for testing inserts clickhouse: # Matches CLICKHOUSE_TAG in tests/common/containers.rs and dfe-infra's - # services.clickhouse-version. The 26.3 minor was a moving tag: every - # rebuild of it changed the server without changing the pin. + # services.clickhouse-version. image: clickhouse/clickhouse-server:26.3.42.3@sha256:21d572843e59539c7d100286b6f5a6053c341fe4b33c2d7b74b1ff7cb24c5399 container_name: clickhouse-dev ports: @@ -39,8 +38,7 @@ services: # Kafka (Redpanda - Kafka-compatible, lighter weight) redpanda: # Matches KAFKA_TAG in scripts/pgo-workload.sh and dfe-infra's - # services.redpanda-version. Two pins of the same broker on different - # patches makes Renovate track two lines and open a PR per line. + # services.redpanda-version, so Renovate tracks one line. image: redpandadata/redpanda:v26.2.6@sha256:8b1076e0d61bfceb66594dc01536427ce0b529f8ead35fdab9763ab6085891ce container_name: redpanda-dev command: diff --git a/docs/config-schema.json b/docs/config-schema.json index cf9f561e..2fce35a4 100644 --- a/docs/config-schema.json +++ b/docs/config-schema.json @@ -500,7 +500,7 @@ "default": null }, "allow_insecure_transport": { - "description": "Opt in to an unencrypted Kafka transport (`plaintext` / `sasl_plaintext`)\nin production. scalo (>=2.8) rejects unencrypted transports under a\nproduction profile at transport construction unless this is set —\ndata and SASL/PLAIN credentials would otherwise ship in the clear. Leave\n`false` (the default) and configure TLS, or set `true` only when the\nin-cluster traffic is already mesh-encrypted.", + "description": "Opt in to an unencrypted Kafka transport (`plaintext` / `sasl_plaintext`)\nin production. scalo rejects unencrypted transports under a\nproduction profile at transport construction unless this is set —\ndata and SASL/PLAIN credentials would otherwise ship in the clear. Leave\n`false` (the default) and configure TLS, or set `true` only when the\nin-cluster traffic is already mesh-encrypted.", "type": "boolean", "default": false }, diff --git a/docs/config-schema.yaml b/docs/config-schema.yaml index c4a4c05b..8d3c769a 100644 --- a/docs/config-schema.yaml +++ b/docs/config-schema.yaml @@ -437,7 +437,7 @@ $defs: allow_insecure_transport: description: |- Opt in to an unencrypted Kafka transport (`plaintext` / `sasl_plaintext`) - in production. scalo (>=2.8) rejects unencrypted transports under a + in production. scalo rejects unencrypted transports under a production profile at transport construction unless this is set — data and SASL/PLAIN credentials would otherwise ship in the clear. Leave `false` (the default) and configure TLS, or set `true` only when the diff --git a/docs/operations.md b/docs/operations.md index 10f85949..41274133 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -124,7 +124,7 @@ The loader runs under a cgroup-aware memory guard: it checks pressure before each Kafka poll and pauses the consumer when memory is tight, rather than risking an OOM kill mid-batch. -The figure it checks is what the kernel charges the loader. On scalo 2.12.2 and later that is the cgroup's `memory.current` (v1 `memory.usage_in_bytes` on an older host), else `VmRSS` from `/proc/self/status`. The cgroup figure is the one the OOM killer compares against `memory.max`, and it counts what jemalloc never sees: librdkafka's allocations, thread stacks and pages jemalloc keeps after a free. Only where neither can be read, such as a macOS dev build, does a `jemalloc` build register jemalloc's live heap instead, so the guard is not left counting only the batch engine's reservations. The `usage_source` field on the `memory guard initialised` log line names the source in force: `cgroup-v2`, `cgroup-v1` or `proc-status` on Linux, `explicit` only in that fallback. +The figure it checks is what the kernel charges the loader. That is the cgroup's `memory.current` (v1 `memory.usage_in_bytes` on an older host), else `VmRSS` from `/proc/self/status`. The cgroup figure is the one the OOM killer compares against `memory.max`, and it counts what jemalloc never sees: librdkafka's allocations, thread stacks and pages jemalloc keeps after a free. Only where neither can be read, such as a macOS dev build, does a `jemalloc` build register jemalloc's live heap instead, so the guard is not left counting only the batch engine's reservations. The `usage_source` field on the `memory guard initialised` log line names the source in force: `cgroup-v2`, `cgroup-v1` or `proc-status` on Linux, `explicit` only in that fallback. `memory.current` also counts reclaimable page cache, such as the mmap'd MaxMind database or a file DLQ, so the brake can engage a little early. That is the safe direction. diff --git a/rustfmt.toml b/rustfmt.toml index 153e7b2a..a14b64ac 100644 --- a/rustfmt.toml +++ b/rustfmt.toml @@ -2,7 +2,3 @@ edition = "2024" max_width = 100 tab_spaces = 4 reorder_imports = true -# imports_granularity and group_imports require nightly rustfmt -# Uncomment when nightly is available in CI: -# imports_granularity = "Module" -# group_imports = "StdExternalCrate" diff --git a/src/clickhouse/client_http.rs b/src/clickhouse/client_http.rs index 66e731f6..d95b71e3 100644 --- a/src/clickhouse/client_http.rs +++ b/src/clickhouse/client_http.rs @@ -822,8 +822,7 @@ fn parse_db_table(table: &str, default_db: &str) -> (String, String) { /// `[A-Za-z0-9_.,= <>!'%-]`. Rejects: `;`, `(`, `)`, `--`, `/*`, `*/`, `\\`, /// backticks, newlines, tabs, and anything > 256 bytes. /// -/// This is defence-in-depth; the method is not currently called from -/// production paths. Real query parameterisation should be used for any +/// This is defence-in-depth. Real query parameterisation should be used for any /// caller-controlled WHERE content. fn validate_where_clause(s: &str) -> Result<()> { if s.len() > 256 { diff --git a/src/config/kafka.rs b/src/config/kafka.rs index b4418467..5a5fa032 100644 --- a/src/config/kafka.rs +++ b/src/config/kafka.rs @@ -30,7 +30,7 @@ pub struct KafkaConfig { pub tls: Option, /// Opt in to an unencrypted Kafka transport (`plaintext` / `sasl_plaintext`) - /// in production. scalo (>=2.8) rejects unencrypted transports under a + /// in production. scalo rejects unencrypted transports under a /// production profile at transport construction unless this is set — /// data and SASL/PLAIN credentials would otherwise ship in the clear. Leave /// `false` (the default) and configure TLS, or set `true` only when the diff --git a/src/kafka/transport.rs b/src/kafka/transport.rs index e05f642c..c3b37e32 100644 --- a/src/kafka/transport.rs +++ b/src/kafka/transport.rs @@ -281,7 +281,7 @@ impl TransportAdapter { client_id: config.client_id.clone(), auto_discover: topics.is_empty(), topics, - // scalo (>=2.8) rejects an unencrypted transport under a production + // scalo rejects an unencrypted transport under a production // profile at construction unless this is explicitly set. Secure by // default; operators opt in for mesh-encrypted in-cluster traffic. allow_insecure_transport: config.allow_insecure_transport, diff --git a/src/main.rs b/src/main.rs index 940c35b0..8360e2c0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -121,8 +121,7 @@ impl ServiceApp for App { // Share the runtime's single ScalingPressure engine (registered via // the `scaling_components` override below and served at - // /scaling/pressure to KEDA). scalo 2.10 unified scaling onto this one - // engine. If the scaling feature/section is off the runtime hands back + // /scaling/pressure to KEDA). If the scaling feature/section is off the runtime hands back // None; fall back to a standalone engine with the same components. let scaling = runtime .scaling @@ -162,9 +161,7 @@ impl ServiceApp for App { // Inject the runtime's SHARED memory guard (the same one feeding the // governor and worker pool) so in-flight byte accounting drives the // inbound brake — never a stand-alone guard the pipeline ignores. - // scalo 2.10 unified scaling onto ONE ScalingPressure engine (the - // separate per-pod scaling-signal cell was removed): the orchestrator - // pushes Kafka assigned-lag + ClickHouse sink circuit-open directly + // The orchestrator pushes Kafka assigned-lag + ClickHouse sink circuit-open directly // into this `scaling` engine, which is served to KEDA at // /scaling/pressure. let mut orchestrator = Orchestrator::with_metrics(config.clone(), metrics) diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs index 81c5c13f..4d57c50f 100644 --- a/src/metrics/mod.rs +++ b/src/metrics/mod.rs @@ -118,7 +118,7 @@ pub struct Metrics { eps_last_update: Arc>, eps_last_count: Arc, - // ClickHouse sink domain metrics (scaling-signal gap audit, scalo 2.8.10). + // ClickHouse sink domain metrics. // The engine can only correlate signals that EXIST — these surface the // ClickHouse sink's load (latency, errors, throughput, backlog, batch size) // so they can drive a domain scaling pressure or KEDA Prometheus trigger. @@ -292,15 +292,13 @@ impl Metrics { eps_last_update: Arc::new(parking_lot::Mutex::new(Instant::now())), eps_last_count: Arc::new(AtomicU64::new(0)), - // ClickHouse sink domain metrics (scaling-signal gap audit, 2.8.10) + // ClickHouse sink domain metrics ch_insert_duration: manager.histogram( "clickhouse_insert_duration_seconds", "ClickHouse insert batch latency", ), - // scalo 2.10 dropped histogram_count / histogram_with_unit: the - // remaining histogram constructors register the manifest unit as - // seconds. These two are a count (rows) and bytes distribution; the - // emitted metric is unchanged, only the manifest unit metadata. + // The histogram constructors register the manifest unit as seconds, + // though these two are a count (rows) and a bytes distribution. ch_flush_rows: manager.histogram( "clickhouse_flush_rows", "Rows per flush to ClickHouse (flush size)", @@ -436,7 +434,7 @@ impl Metrics { self.dfe.transport_sent(TransportKind::Http, rows as u64); self.dfe.transport_send_duration("clickhouse", latency_secs); - // ClickHouse sink domain metrics (2.8.10 scaling-signal gap audit): + // ClickHouse sink domain metrics: // latency distribution, flush-size (rows), and throughput accounting. self.ch_insert_duration.record(latency_secs); self.ch_flush_rows.record(rows as f64); diff --git a/src/pipeline/orchestrator.rs b/src/pipeline/orchestrator.rs index e36ccd86..9393c22e 100644 --- a/src/pipeline/orchestrator.rs +++ b/src/pipeline/orchestrator.rs @@ -701,9 +701,7 @@ impl Orchestrator { } // Push the per-pod Kafka inbound scaling signal into the - // unified ScalingPressure engine (scalo 2.10 collapsed the old - // separate scaling-signal cell into ONE ScalingPressure served - // to KEDA at /scaling/pressure). assigned_lag() sums lag over + // ScalingPressure engine served to KEDA at /scaling/pressure. assigned_lag() sums lag over // THIS pod's ASSIGNED partitions (scale-invariant). gRPC has no // broker lag -> None -> the kafka_lag term stays 0 (CPU-only). // The flush tick (every flush_age_secs, default 5s) is fresher @@ -1472,7 +1470,7 @@ impl Orchestrator { if let Some(ref m) = self.metrics { m.record_flush(result.inserted, latency); m.record_insert_quantities(batch_bytes, 1); - // ClickHouse flush-size (bytes) distribution (2.8.10 audit). + // ClickHouse flush-size (bytes) distribution. m.record_flush_bytes(batch_bytes); } } @@ -1483,7 +1481,7 @@ impl Orchestrator { self.stats.errors += 1; if let Some(ref m) = self.metrics { m.record_error(); - // ClickHouse-specific terminal insert error (2.8.10 audit). + // ClickHouse-specific terminal insert error. m.record_clickhouse_insert_error(); } if self.holds_answers { @@ -2964,7 +2962,7 @@ const SCHEMA_PAUSE_TABLES_LOGGED: usize = 5; /// /// This also settles a second hazard: `BufferManager::get_ready_for_flush` /// iterates an `FxHashMap`, so batch order is nondeterministic and a low-offset -/// batch committing after a high one used to rewind the watermark. One commit +/// batch committing after a high one would rewind the watermark. One commit /// per cycle cannot rewind. /// /// `withheld` carries every offset whose row is not placed yet: rows in a buffer diff --git a/src/transform/extractor.rs b/src/transform/extractor.rs index 78975b01..13a4a151 100644 --- a/src/transform/extractor.rs +++ b/src/transform/extractor.rs @@ -753,9 +753,8 @@ fn get_split<'a>(parsed: &'a Map, path: &str) -> Option<&'a Value /// /// A null reads as absent, so `first()` tries the next path and `@default` applies. /// -/// A common-header column used to empty the field it read, so a meta column -/// reading the same field found nothing and landed NULL. A path in `contended` -/// is copied instead, which leaves it there for the columns that follow. +/// A path in `contended` is copied rather than taken, so a meta column reading +/// the same field after a common-header column still finds it. /// /// `contended` is empty for every table with no shared source field, so the /// ordinary column costs one `is_empty` check and keeps the move. diff --git a/tests/TESTING.md b/tests/TESTING.md index ee46b60b..57a3ebf5 100644 --- a/tests/TESTING.md +++ b/tests/TESTING.md @@ -25,8 +25,8 @@ Full adds every other real-server test: the inserter, readiness, data type, DLQ, | Service | Image | Container ports | |---|---|---| -| ClickHouse | `clickhouse/clickhouse-server:26.3.42.3@sha256:21d572843e59539c7d100286b6f5a6053c341fe4b33c2d7b74b1ff7cb24c5399` | 8123 HTTP, 9000 native, 8443 HTTPS, 9440 native TLS | -| Kafka | `apache/kafka:4.3.1@sha256:77e3df9054047a88b520d0cc46e16696d3b22022e1d580aeccd2632df6532837` | 9092 PLAINTEXT | +| ClickHouse | the image `tests/common/containers.rs` pins | 8123 HTTP, 9000 native, 8443 HTTPS, 9440 native TLS | +| Kafka | scalo's `KAFKA_IMAGE` | 9092 PLAINTEXT | The ClickHouse pin lives in `tests/common/containers.rs` and follows dfe-infra `versions.yaml`. The Kafka pin is scalo's `transport::kafka::test_support::KAFKA_IMAGE`, the line the DFE stack deploys. diff --git a/tests/common/containers.rs b/tests/common/containers.rs index f26d134d..127f2f67 100644 --- a/tests/common/containers.rs +++ b/tests/common/containers.rs @@ -34,7 +34,7 @@ mod testcontainers_impl { /// ClickHouse to test against -- the server dfe-infra `versions.yaml` /// deploys and `docker-compose.dev.yaml` runs, pinned to the patch and its - /// digest rather than the 26.3 minor, so a rebuild of that tag cannot + /// digest rather than the minor, so a rebuild of that tag cannot /// change what a green run tested against. /// /// Tag and digest are separate consts because the org Renovate regex reads diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 3b9afb51..fa8a43eb 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -470,7 +470,7 @@ pub const TEST_SUITE_LABEL: (&str, &str) = ("io.hyperi.test.suite", "dfe-loader- /// /// Pass `Some(test)` -- the owning test -- for anything a test starts for itself, /// which is everything here. `None` is for a container started once for a whole -/// test binary; nothing does that today, and using it from several tests would +/// test binary, and using it from several tests would /// make them collide on the name rather than share the container. /// /// The name ends in the owning process id, so two runs of one test on one