From 191ff426aa81dca78bc4084bbd7f29b2e8b999e2 Mon Sep 17 00:00:00 2001 From: Derek Date: Sat, 14 Nov 2026 12:00:22 +1100 Subject: [PATCH 1/2] docs(loader): remove stale versions and counts Removes or repoints 20 stale sentences in docs and code comments. They cover scalo release history, repeated image pins and floors, and "today" or "used to" phrasing. Cuts 119 comment lines from config files. Removes lint-allow site counts, template examples and decision history. Keeps a short reason beside a surprising value. --- .cargo/config.toml | 15 +---- .config/nextest.toml | 11 +--- .gitleaks.toml | 27 +------- .hyperi-ci.yaml | 31 +++------- BENCHMARKS.md | 2 +- Cargo.toml | 113 ++++++++-------------------------- docker-compose.dev.yaml | 6 +- docs/operations.md | 2 +- rustfmt.toml | 4 -- src/clickhouse/client_http.rs | 3 +- src/config/kafka.rs | 2 +- src/kafka/transport.rs | 2 +- src/main.rs | 7 +-- src/metrics/mod.rs | 12 ++-- src/pipeline/orchestrator.rs | 10 ++- src/transform/extractor.rs | 5 +- tests/TESTING.md | 4 +- tests/common/containers.rs | 2 +- tests/common/mod.rs | 2 +- 19 files changed, 62 insertions(+), 198 deletions(-) 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/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 From 725b9e7a1890f9245fcecf5eac159b3fae498eff Mon Sep 17 00:00:00 2001 From: Derek Date: Sat, 14 Nov 2026 12:26:27 +1100 Subject: [PATCH 2/2] docs(loader): match config schema to its doc comment Update the generated config schema to match the `allow_insecure_transport` doc comment. --- docs/config-schema.json | 2 +- docs/config-schema.yaml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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