mz_persist_client/
cfg.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10#![allow(missing_docs)]
11
12//! The tunable knobs for persist.
13//!
14//! Persist configs are `ParameterScope::Environment` as a class, including the
15//! ones only ever read on `clusterd`. The same client code runs in
16//! `environmentd`, and every copy of it acts on shared durable state, so a
17//! replica-scoped persist config could not reach the `environmentd` client and
18//! a rollout targeting replicas would leave a shard's other writer on the old
19//! value indefinitely. Declare new persist configs `Environment` too.
20
21use std::sync::Arc;
22use std::sync::atomic::AtomicBool;
23use std::time::{Duration, SystemTime, UNIX_EPOCH};
24
25use mz_build_info::BuildInfo;
26use mz_dyncfg::{Config, ConfigDefault, ConfigSet, ConfigUpdates, ParameterScope};
27use mz_ore::instrument;
28use mz_ore::now::NowFn;
29use mz_persist::cfg::BlobKnobs;
30use mz_persist::retry::Retry;
31use mz_persist::s3::{BLOB_HTTP_POOL_MAX_IN_FLIGHT_PER_CLIENT, BlobHttpPool};
32use mz_postgres_client::PostgresClientKnobs;
33use proptest_derive::Arbitrary;
34use semver::Version;
35use serde::{Deserialize, Serialize};
36use tokio::sync::watch;
37
38use crate::async_runtime;
39use crate::internal::machine::{
40    NEXT_LISTEN_BATCH_RETRYER_CLAMP, NEXT_LISTEN_BATCH_RETRYER_INITIAL_BACKOFF,
41    NEXT_LISTEN_BATCH_RETRYER_MULTIPLIER,
42};
43use crate::internal::state::ROLLUP_THRESHOLD;
44use crate::operators::STORAGE_SOURCE_DECODE_FUEL;
45use crate::read::READER_LEASE_DURATION;
46
47// Ignores the patch version
48const SELF_MANAGED_VERSIONS: &[Version; 2] = &[
49    // 25.1
50    Version::new(0, 130, 0),
51    // 25.2
52    Version::new(0, 147, 0),
53];
54
55/// The tunable knobs for persist.
56///
57/// Tuning inputs:
58/// - A larger blob_target_size (capped at KEY_VAL_DATA_MAX_LEN) results in
59///   fewer entries in consensus state. Before we have compaction and/or
60///   incremental state, it is already growing without bound, so this is a
61///   concern. OTOH, for any "reasonable" size (> 100MiB?) of blob_target_size,
62///   it seems we'd end up with a pretty tremendous amount of data in the shard
63///   before this became a real issue.
64/// - A larger blob_target_size will results in fewer s3 operations, which are
65///   charged per operation. (Hmm, maybe not if we're charged per call in a
66///   multipart op. The S3Blob impl already chunks things at 8MiB.)
67/// - A smaller blob_target_size will result in more even memory usage in
68///   readers.
69/// - A larger batch_builder_max_outstanding_parts increases throughput (to a
70///   point).
71/// - A smaller batch_builder_max_outstanding_parts provides a bound on the
72///   amount of memory used by a writer.
73/// - A larger compaction_heuristic_min_inputs means state size is larger.
74/// - A smaller compaction_heuristic_min_inputs means more compactions happen
75///   (higher write amp).
76/// - A larger compaction_heuristic_min_updates means more consolidations are
77///   discovered while reading a snapshot (higher read amp and higher space
78///   amp).
79/// - A smaller compaction_heuristic_min_updates means more compactions happen
80///   (higher write amp).
81///
82/// Tuning logic:
83/// - blob_target_size was initially selected to be an exact multiple of 8MiB
84///   (the s3 multipart size) that was in the same neighborhood as our initial
85///   max throughput (~250MiB).
86/// - batch_builder_max_outstanding_parts was initially selected to be as small
87///   as possible without harming pipelining. 0 means no pipelining, 1 is full
88///   pipelining as long as generating data takes less time than writing to s3
89///   (hopefully a fair assumption), 2 is a little extra slop on top of 1.
90/// - compaction_heuristic_min_inputs was set by running the open-loop benchmark
91///   with batches of size 10,240 bytes (selected to be small but such that the
92///   overhead of our columnar encoding format was less than 10%) and manually
93///   increased until the write amp stopped going down. This becomes much less
94///   important once we have incremental state. The initial value is a
95///   placeholder and should be revisited at some point.
96/// - compaction_heuristic_min_updates was set via a thought experiment. This is
97///   an `O(n*log(n))` upper bound on the number of unconsolidated updates that
98///   would be consolidated if we compacted as the in-mem Spine does. The
99///   initial value is a placeholder and should be revisited at some point.
100///
101/// TODO: Move these tuning notes into SessionVar descriptions once we have
102/// SystemVars for most of these.
103//
104// TODO: The configs left here don't react dynamically to changes. Move as many
105// of them to DynamicConfig as possible.
106#[derive(Debug, Clone)]
107pub struct PersistConfig {
108    /// Info about which version of the code is running.
109    pub build_version: Version,
110    /// An opaque string describing the host of this persist client.
111    /// Stored in state and used for debugging.
112    pub hostname: String,
113    /// Whether this persist instance is running in a "cc" sized cluster.
114    pub is_cc_active: bool,
115    /// Memory limit of the process, if known.
116    pub announce_memory_limit: Option<usize>,
117    /// A clock to use for all leasing and other non-debugging use.
118    pub now: NowFn,
119    /// Persist [Config]s that can change value dynamically within the lifetime
120    /// of a process.
121    ///
122    /// TODO(cfg): Entirely replace dynamic with this.
123    pub configs: Arc<ConfigSet>,
124    /// Indicates whether `configs` has been synced at least once with an
125    /// upstream source.
126    configs_synced_once: Arc<watch::Sender<bool>>,
127    /// Whether to physically and logically compact batches in blob storage.
128    pub compaction_enabled: bool,
129    /// Whether the `Compactor` will process compaction requests, or drop them on the floor.
130    pub compaction_process_requests: Arc<AtomicBool>,
131    /// In Compactor::compact_and_apply_background, the maximum number of concurrent
132    /// compaction requests that can execute for a given shard.
133    pub compaction_concurrency_limit: usize,
134    /// In Compactor::compact_and_apply_background, the maximum number of pending
135    /// compaction requests to queue.
136    pub compaction_queue_size: usize,
137    /// In Compactor::compact_and_apply_background, how many updates to encode or
138    /// decode before voluntarily yielding the task.
139    pub compaction_yield_after_n_updates: usize,
140    /// Length of time after a writer's last operation after which the writer
141    /// may be expired.
142    pub writer_lease_duration: Duration,
143    /// Length of time between critical handles' calls to downgrade since
144    pub critical_downgrade_interval: Duration,
145    /// Number of worker threads to create for the [`crate::IsolatedRuntime`], defaults to the
146    /// number of threads.
147    pub isolated_runtime_worker_threads: usize,
148}
149
150// Impl Deref to ConfigSet for convenience of accessing the dynamic configs.
151impl std::ops::Deref for PersistConfig {
152    type Target = ConfigSet;
153    fn deref(&self) -> &Self::Target {
154        &self.configs
155    }
156}
157
158impl PersistConfig {
159    /// Returns a new instance of [PersistConfig] with default tuning and
160    /// default ConfigSet.
161    pub fn new_default_configs(build_info: &BuildInfo, now: NowFn) -> Self {
162        Self::new(build_info, now, all_dyncfgs(ConfigSet::default()))
163    }
164
165    /// Returns a new instance of [PersistConfig] with default tuning and the
166    /// specified ConfigSet.
167    pub fn new(build_info: &BuildInfo, now: NowFn, configs: ConfigSet) -> Self {
168        // Escape hatch in case we need to disable compaction.
169        let compaction_disabled = mz_ore::env::is_var_truthy("MZ_PERSIST_COMPACTION_DISABLED");
170
171        // We create receivers on demand, so we drop the initial receiver.
172        let (configs_synced_once, _) = watch::channel(false);
173
174        Self {
175            build_version: build_info.semver_version(),
176            is_cc_active: false,
177            announce_memory_limit: None,
178            now,
179            configs: Arc::new(configs),
180            configs_synced_once: Arc::new(configs_synced_once),
181            compaction_enabled: !compaction_disabled,
182            compaction_process_requests: Arc::new(AtomicBool::new(true)),
183            compaction_concurrency_limit: 5,
184            compaction_queue_size: 20,
185            compaction_yield_after_n_updates: 100_000,
186            writer_lease_duration: 60 * Duration::from_secs(60),
187            critical_downgrade_interval: Duration::from_secs(30),
188            isolated_runtime_worker_threads: num_cpus::get(),
189            // TODO: This doesn't work with the process orchestrator. Instead,
190            // separate --log-prefix into --service-name and --enable-log-prefix
191            // options, where the first is always provided and the second is
192            // conditionally enabled by the process orchestrator.
193            hostname: {
194                use std::fmt::Write;
195                let mut name = std::env::var("HOSTNAME").unwrap_or_else(|_| "unknown".to_owned());
196                write!(&mut name, " {}", build_info.version)
197                    .expect("writing to string should not fail");
198                name
199            },
200        }
201    }
202
203    pub(crate) fn set_config<T: ConfigDefault>(&self, cfg: &Config<T>, val: T) {
204        let mut updates = ConfigUpdates::default();
205        updates.add(cfg, val);
206        updates.apply(self)
207    }
208
209    /// Applies the provided updates to this configuration.
210    ///
211    /// You should prefer calling this method over mutating `self.configs`
212    /// directly, so that [`Self::configs_synced_once`] can be properly
213    /// maintained.
214    pub fn apply_from(&self, updates: &ConfigUpdates) {
215        updates.apply(&self.configs);
216        self.configs_synced_once.send_replace(true);
217    }
218
219    /// Resolves when `configs` has been synced at least once with an upstream
220    /// source, i.e., via [`Self::apply_from`].
221    ///
222    /// If `configs` has already been synced once at the time the method is
223    /// called, resolves immediately.
224    ///
225    /// Useful in conjunction with configuration parameters that cannot be
226    /// dynamically updated once set (e.g., PubSub).
227    #[instrument(level = "info")]
228    pub async fn configs_synced_once(&self) {
229        self.configs_synced_once
230            .subscribe()
231            .wait_for(|synced| *synced)
232            .await
233            .expect("we have a borrow on sender so it cannot drop");
234    }
235
236    /// The maximum amount of work to do in the persist_source mfp_and_decode
237    /// operator before yielding.
238    pub fn storage_source_decode_fuel(&self) -> usize {
239        STORAGE_SOURCE_DECODE_FUEL.get(self)
240    }
241
242    /// Overrides the value for "persist_reader_lease_duration".
243    pub fn set_reader_lease_duration(&self, val: Duration) {
244        self.set_config(&READER_LEASE_DURATION, val);
245    }
246
247    /// Overrides the value for "persist_rollup_threshold".
248    pub fn set_rollup_threshold(&self, val: usize) {
249        self.set_config(&ROLLUP_THRESHOLD, val);
250    }
251
252    /// Overrides the value for the "persist_next_listen_batch_retryer_*"
253    /// configs.
254    pub fn set_next_listen_batch_retryer(&self, val: RetryParameters) {
255        self.set_config(
256            &NEXT_LISTEN_BATCH_RETRYER_INITIAL_BACKOFF,
257            val.initial_backoff,
258        );
259        self.set_config(&NEXT_LISTEN_BATCH_RETRYER_MULTIPLIER, val.multiplier);
260        self.set_config(&NEXT_LISTEN_BATCH_RETRYER_CLAMP, val.clamp);
261    }
262
263    pub fn disable_compaction(&self) {
264        tracing::info!("Disabling Persist Compaction");
265        self.compaction_process_requests
266            .store(false, std::sync::atomic::Ordering::Relaxed);
267    }
268
269    pub fn enable_compaction(&self) {
270        tracing::info!("Enabling Persist Compaction");
271        self.compaction_process_requests
272            .store(true, std::sync::atomic::Ordering::Relaxed);
273    }
274
275    /// Returns a new instance of [PersistConfig] for tests.
276    pub fn new_for_tests() -> Self {
277        use mz_build_info::DUMMY_BUILD_INFO;
278        use mz_ore::now::SYSTEM_TIME;
279
280        let mut cfg = Self::new_default_configs(&DUMMY_BUILD_INFO, SYSTEM_TIME.clone());
281        cfg.hostname = "tests".into();
282        cfg.isolated_runtime_worker_threads = async_runtime::TEST_THREADS;
283        cfg
284    }
285}
286
287#[allow(non_upper_case_globals)]
288pub(crate) const MiB: usize = 1024 * 1024;
289
290/// Adds the full set of all persist [Config]s.
291///
292/// Persist configs are [`ParameterScope::Environment`] by default, because a
293/// persist client runs in every `clusterd` process and the per-replica dyncfg
294/// push reaches its `ConfigSet` (the compute worker applies the pushed updates
295/// to `persist_clients.cfg()`). `environmentd`'s own persist clients read the
296/// same configs and see the environment-wide value, which is what "replica-
297/// local" means for a config that both processes run. The exceptions are the
298/// configs read only from `environmentd`'s catalog and expression cache, which
299/// are [`ParameterScope::Environment`].
300///
301/// TODO(cfg): Consider replacing this with a static global registry powered by
302/// something like the `ctor` or `inventory` crate. This would involve managing
303/// the footgun of a Config being linked into one binary but not the other.
304pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
305    mz_persist::cfg::all_dyn_configs(configs)
306        .add(&crate::batch::BATCH_DELETE_ENABLED)
307        .add(&crate::batch::BLOB_TARGET_SIZE)
308        .add(&crate::batch::INLINE_WRITES_TOTAL_MAX_BYTES)
309        .add(&crate::batch::INLINE_WRITES_SINGLE_MAX_BYTES)
310        .add(&crate::batch::ENCODING_ENABLE_DICTIONARY)
311        .add(&crate::batch::ENCODING_COMPRESSION_FORMAT)
312        .add(&crate::batch::STRUCTURED_KEY_LOWER_LEN)
313        .add(&crate::batch::MAX_RUN_LEN)
314        .add(&crate::batch::MAX_RUNS)
315        .add(&BLOB_OPERATION_TIMEOUT)
316        .add(&BLOB_OPERATION_ATTEMPT_TIMEOUT)
317        .add(&BLOB_CONNECT_TIMEOUT)
318        .add(&BLOB_READ_TIMEOUT)
319        .add(&SHARD_METRICS)
320        .add(&PER_SHARD_METRICS_ENABLE_REGEX)
321        .add(&SHARD_INFO_METRIC)
322        .add(&crate::cfg::CONSENSUS_CONNECTION_POOL_MAX_SIZE)
323        .add(&crate::cfg::CONSENSUS_CONNECTION_POOL_MAX_WAIT)
324        .add(&crate::cfg::CONSENSUS_CONNECTION_POOL_TTL_STAGGER)
325        .add(&crate::cfg::CONSENSUS_CONNECTION_POOL_TTL)
326        .add(&crate::cfg::CRDB_CONNECT_TIMEOUT)
327        .add(&crate::cfg::CRDB_TCP_USER_TIMEOUT)
328        .add(&crate::cfg::CRDB_KEEPALIVES_IDLE)
329        .add(&crate::cfg::CRDB_KEEPALIVES_INTERVAL)
330        .add(&crate::cfg::CRDB_KEEPALIVES_RETRIES)
331        .add(&crate::cfg::USE_CRITICAL_SINCE_TXN)
332        .add(&crate::cfg::USE_CRITICAL_SINCE_CATALOG)
333        .add(&crate::cfg::USE_CRITICAL_SINCE_SOURCE)
334        .add(&crate::cfg::USE_CRITICAL_SINCE_SNAPSHOT)
335        .add(&crate::cfg::SOURCE_HYDRATION_FRONTIER_COALESCE_BYTES)
336        .add(&crate::cfg::SOURCE_FETCH_CONCURRENCY)
337        .add(&BATCH_BUILDER_MAX_OUTSTANDING_PARTS)
338        .add(&COMPACTION_HEURISTIC_MIN_INPUTS)
339        .add(&COMPACTION_HEURISTIC_MIN_PARTS)
340        .add(&COMPACTION_HEURISTIC_MIN_UPDATES)
341        .add(&COMPACTION_MEMORY_BOUND_BYTES)
342        .add(&GC_BLOB_DELETE_CONCURRENCY_LIMIT)
343        .add(&GC_REMOVE_FINALIZED_ROLLUPS)
344        .add(&STATE_VERSIONS_RECENT_LIVE_DIFFS_LIMIT)
345        .add(&USAGE_STATE_FETCH_CONCURRENCY_LIMIT)
346        .add(&crate::cache::STATE_UPDATE_LEASE_TIMEOUT)
347        .add(&crate::cli::admin::CATALOG_FORCE_COMPACTION_FUEL)
348        .add(&crate::cli::admin::CATALOG_FORCE_COMPACTION_WAIT)
349        .add(&crate::cli::admin::EXPRESSION_CACHE_FORCE_COMPACTION_FUEL)
350        .add(&crate::cli::admin::EXPRESSION_CACHE_FORCE_COMPACTION_WAIT)
351        .add(&crate::fetch::FETCH_SEMAPHORE_COST_ADJUSTMENT)
352        .add(&crate::fetch::FETCH_SEMAPHORE_PERMIT_ADJUSTMENT)
353        .add(&crate::fetch::VALIDATE_PART_BOUNDS_ON_READ)
354        .add(&crate::fetch::OPTIMIZE_IGNORED_DATA_FETCH)
355        .add(&crate::internal::cache::BLOB_CACHE_MEM_LIMIT_BYTES)
356        .add(&crate::internal::cache::BLOB_CACHE_SCALE_WITH_THREADS)
357        .add(&crate::internal::cache::BLOB_CACHE_SCALE_FACTOR_BYTES)
358        .add(&crate::internal::compact::COMPACTION_MINIMUM_TIMEOUT)
359        .add(&crate::internal::compact::COMPACTION_CHECK_PROCESS_FLAG)
360        .add(&crate::internal::machine::CLAIM_UNCLAIMED_COMPACTIONS)
361        .add(&crate::internal::machine::CLAIM_COMPACTION_PERCENT)
362        .add(&crate::internal::machine::CLAIM_COMPACTION_MIN_VERSION)
363        .add(&crate::internal::machine::NEXT_LISTEN_BATCH_RETRYER_CLAMP)
364        .add(&crate::internal::machine::NEXT_LISTEN_BATCH_RETRYER_FIXED_SLEEP)
365        .add(&crate::internal::machine::NEXT_LISTEN_BATCH_RETRYER_INITIAL_BACKOFF)
366        .add(&crate::internal::machine::NEXT_LISTEN_BATCH_RETRYER_MULTIPLIER)
367        .add(&crate::internal::state::ROLLUP_THRESHOLD)
368        .add(&crate::internal::state::ROLLUP_USE_ACTIVE_ROLLUP)
369        .add(&crate::internal::state::GC_FALLBACK_THRESHOLD_MS)
370        .add(&crate::internal::state::GC_USE_ACTIVE_GC)
371        .add(&crate::internal::state::GC_MIN_VERSIONS)
372        .add(&crate::internal::state::GC_MAX_VERSIONS)
373        .add(&crate::internal::state::ROLLUP_FALLBACK_THRESHOLD_MS)
374        .add(&crate::internal::state::ENABLE_INCREMENTAL_COMPACTION)
375        .add(&crate::operators::STORAGE_SOURCE_DECODE_FUEL)
376        .add(&crate::read::READER_LEASE_DURATION)
377        .add(&crate::rpc::PUBSUB_CLIENT_ENABLED)
378        .add(&crate::rpc::PUBSUB_PUSH_DIFF_ENABLED)
379        .add(&crate::rpc::PUBSUB_SAME_PROCESS_DELEGATE_ENABLED)
380        .add(&crate::rpc::PUBSUB_CONNECT_ATTEMPT_TIMEOUT)
381        .add(&crate::rpc::PUBSUB_REQUEST_TIMEOUT)
382        .add(&crate::rpc::PUBSUB_CONNECT_MAX_BACKOFF)
383        .add(&crate::rpc::PUBSUB_CLIENT_SENDER_CHANNEL_SIZE)
384        .add(&crate::rpc::PUBSUB_CLIENT_RECEIVER_CHANNEL_SIZE)
385        .add(&crate::rpc::PUBSUB_SERVER_CONNECTION_CHANNEL_SIZE)
386        .add(&crate::rpc::PUBSUB_STATE_CACHE_SHARD_REF_CHANNEL_SIZE)
387        .add(&crate::rpc::PUBSUB_RECONNECT_BACKOFF)
388        .add(&crate::stats::STATS_AUDIT_PERCENT)
389        .add(&crate::stats::STATS_AUDIT_PANIC)
390        .add(&crate::stats::STATS_BUDGET_BYTES)
391        .add(&crate::stats::STATS_COLLECTION_ENABLED)
392        .add(&crate::stats::STATS_FILTER_ENABLED)
393        .add(&crate::stats::STATS_UNTRIMMABLE_COLUMNS_EQUALS)
394        .add(&crate::stats::STATS_UNTRIMMABLE_COLUMNS_PREFIX)
395        .add(&crate::stats::STATS_UNTRIMMABLE_COLUMNS_SUFFIX)
396        .add(&crate::fetch::PART_DECODE_FORMAT)
397        .add(&crate::fetch::PART_DECODE_BATCH_ROWS)
398        .add(&crate::write::COMBINE_INLINE_WRITES)
399        .add(&crate::write::VALIDATE_PART_BOUNDS_ON_WRITE)
400}
401
402impl PersistConfig {
403    pub(crate) const DEFAULT_FALLBACK_ROLLUP_THRESHOLD_MULTIPLIER: usize = 3;
404
405    pub fn set_state_versions_recent_live_diffs_limit(&self, val: usize) {
406        self.set_config(&STATE_VERSIONS_RECENT_LIVE_DIFFS_LIMIT, val);
407    }
408}
409
410/// Sets the maximum size of the connection pool that is used by consensus.
411///
412/// Applies to existing pools without a restart. Each pool picks up a changed
413/// value the next time a connection is requested from it.
414pub const CONSENSUS_CONNECTION_POOL_MAX_SIZE: Config<usize> = Config::new(
415    "persist_consensus_connection_pool_max_size",
416    50,
417    "The maximum size the connection pool to Postgres/CRDB will grow to.",
418    ParameterScope::Environment,
419);
420
421/// Sets the maximum amount of time we'll wait to acquire a connection from
422/// the connection pool.
423///
424/// Requires a restart of the process to take effect.
425const CONSENSUS_CONNECTION_POOL_MAX_WAIT: Config<Duration> = Config::new(
426    "persist_consensus_connection_pool_max_wait",
427    Duration::from_secs(60),
428    "The amount of time we'll wait for a connection to become available.",
429    ParameterScope::Environment,
430);
431
432/// The minimum TTL of a connection to Postgres/CRDB before it is proactively
433/// terminated. Connections are routinely culled to balance load against the
434/// downstream database.
435const CONSENSUS_CONNECTION_POOL_TTL: Config<Duration> = Config::new(
436    "persist_consensus_connection_pool_ttl",
437    Duration::from_secs(300),
438    "\
439    The minimum TTL of a Consensus connection to Postgres/CRDB before it is \
440    proactively terminated",
441    ParameterScope::Environment,
442);
443
444/// The minimum time between TTLing connections to Postgres/CRDB. This delay is
445/// used to stagger reconnections to avoid stampedes and high tail latencies.
446/// This value should be much less than `consensus_connection_pool_ttl` so that
447/// reconnections are biased towards terminating the oldest connections first. A
448/// value of `consensus_connection_pool_ttl /
449/// consensus_connection_pool_max_size` is likely a good place to start so that
450/// all connections are rotated when the pool is fully used.
451const CONSENSUS_CONNECTION_POOL_TTL_STAGGER: Config<Duration> = Config::new(
452    "persist_consensus_connection_pool_ttl_stagger",
453    Duration::from_secs(6),
454    "The minimum time between TTLing Consensus connections to Postgres/CRDB.",
455    ParameterScope::Environment,
456);
457
458/// The duration to wait for a Consensus Postgres/CRDB connection to be made
459/// before retrying.
460pub const CRDB_CONNECT_TIMEOUT: Config<Duration> = Config::new(
461    "crdb_connect_timeout",
462    Duration::from_secs(5),
463    "The time to connect to CockroachDB before timing out and retrying.",
464    ParameterScope::Environment,
465);
466
467/// The TCP user timeout for a Consensus Postgres/CRDB connection. Specifies the
468/// amount of time that transmitted data may remain unacknowledged before the
469/// TCP connection is forcibly closed.
470pub const CRDB_TCP_USER_TIMEOUT: Config<Duration> = Config::new(
471    "crdb_tcp_user_timeout",
472    Duration::from_secs(30),
473    "\
474    The TCP timeout for connections to CockroachDB. Specifies the amount of \
475    time that transmitted data may remain unacknowledged before the TCP \
476    connection is forcibly closed.",
477    ParameterScope::Environment,
478);
479
480pub const CRDB_KEEPALIVES_IDLE: Config<Duration> = Config::new(
481    "crdb_keepalives_idle",
482    Duration::from_secs(10),
483    "\
484    The amount of idle time before a TCP keepalive packet is sent on CRDB \
485    connections.",
486    ParameterScope::Environment,
487);
488
489pub const CRDB_KEEPALIVES_INTERVAL: Config<Duration> = Config::new(
490    "crdb_keepalives_interval",
491    Duration::from_secs(5),
492    "The time interval between TCP keepalive probes on CRDB connections.",
493    ParameterScope::Environment,
494);
495
496pub const CRDB_KEEPALIVES_RETRIES: Config<u32> = Config::new(
497    "crdb_keepalives_retries",
498    5,
499    "\
500    The maximum number of TCP keepalive probes that will be sent before \
501    dropping a CRDB connection.",
502    ParameterScope::Environment,
503);
504
505/// Migrate the txns code to use the critical since when opening a new read handle.
506pub const USE_CRITICAL_SINCE_TXN: Config<bool> = Config::new(
507    "persist_use_critical_since_txn",
508    true,
509    "Use the critical since (instead of the overall since) when initializing a subscribe.",
510    ParameterScope::Environment,
511);
512
513/// Migrate the catalog to use the critical since when opening a new read handle.
514pub const USE_CRITICAL_SINCE_CATALOG: Config<bool> = Config::new(
515    "persist_use_critical_since_catalog",
516    false,
517    "Use the critical since (instead of the overall since) for the Persist-backed catalog.",
518    ParameterScope::Environment,
519);
520
521/// Migrate the persist source to use the critical since when opening a new read handle.
522pub const USE_CRITICAL_SINCE_SOURCE: Config<bool> = Config::new(
523    "persist_use_critical_since_source",
524    false,
525    "Use the critical since (instead of the overall since) in the Persist source.",
526    ParameterScope::Environment,
527);
528
529/// While the source is catching up to the shard upper observed at hydration,
530/// coalesce per-batch frontier downgrades until at least this many encoded
531/// bytes have been emitted at the held capability. `0` disables coalescing and
532/// restores the per-batch behavior.
533pub const SOURCE_HYDRATION_FRONTIER_COALESCE_BYTES: Config<usize> = Config::new(
534    "persist_source_hydration_frontier_coalesce_bytes",
535    0,
536    "While catching up to the hydration-time upper, the persist source coalesces \
537     frontier downgrades until this many encoded bytes have been emitted (0 disables).",
538    ParameterScope::Environment,
539);
540
541/// Maximum number of part fetches the persist source issues concurrently, per
542/// worker. Concurrent fetches amortize the blob-store round-trip, which
543/// dominates when there are many small parts (e.g. a fine-grained hydration
544/// replay). `1` keeps the previous serial behavior.
545pub const SOURCE_FETCH_CONCURRENCY: Config<usize> = Config::new(
546    "persist_source_fetch_concurrency",
547    1,
548    "Maximum number of part fetches the persist source issues concurrently per worker \
549     (1 = serial).",
550    ParameterScope::Environment,
551);
552
553/// Migrate snapshots to use the critical since when opening a new read handle.
554pub const USE_CRITICAL_SINCE_SNAPSHOT: Config<bool> = Config::new(
555    "persist_use_critical_since_snapshot",
556    false,
557    "Use the critical since (instead of the overall since) when taking snapshots in the controller or in fast-path peeks.",
558    ParameterScope::Environment,
559);
560
561/// The maximum number of parts (s3 blobs) that [crate::batch::BatchBuilder]
562/// will pipeline before back-pressuring [crate::batch::BatchBuilder::add]
563/// calls on previous ones finishing.
564pub const BATCH_BUILDER_MAX_OUTSTANDING_PARTS: Config<usize> = Config::new(
565    "persist_batch_builder_max_outstanding_parts",
566    2,
567    "The number of writes a batch builder can have outstanding before we slow down the writer.",
568    ParameterScope::Environment,
569);
570
571/// In Compactor::compact_and_apply, we do the compaction (don't skip it)
572/// if the number of inputs is at least this many. Compaction is performed
573/// if any of the heuristic criteria are met (they are OR'd).
574pub const COMPACTION_HEURISTIC_MIN_INPUTS: Config<usize> = Config::new(
575    "persist_compaction_heuristic_min_inputs",
576    8,
577    "Don't skip compaction if we have more than this many hollow batches as input.",
578    ParameterScope::Environment,
579);
580
581/// In Compactor::compact_and_apply, we do the compaction (don't skip it)
582/// if the number of batch parts is at least this many. Compaction is performed
583/// if any of the heuristic criteria are met (they are OR'd).
584pub const COMPACTION_HEURISTIC_MIN_PARTS: Config<usize> = Config::new(
585    "persist_compaction_heuristic_min_parts",
586    8,
587    "Don't skip compaction if we have more than this many parts as input.",
588    ParameterScope::Environment,
589);
590
591/// In Compactor::compact_and_apply, we do the compaction (don't skip it)
592/// if the number of updates is at least this many. Compaction is performed
593/// if any of the heuristic criteria are met (they are OR'd).
594pub const COMPACTION_HEURISTIC_MIN_UPDATES: Config<usize> = Config::new(
595    "persist_compaction_heuristic_min_updates",
596    1024,
597    "Don't skip compaction if we have more than this many updates as input.",
598    ParameterScope::Environment,
599);
600
601/// The upper bound on compaction's memory consumption. The value must be at
602/// least 4*`blob_target_size`. Increasing this value beyond the minimum allows
603/// compaction to merge together more runs at once, providing greater
604/// consolidation of updates, at the cost of greater memory usage.
605pub const COMPACTION_MEMORY_BOUND_BYTES: Config<usize> = Config::new(
606    "persist_compaction_memory_bound_bytes",
607    1024 * MiB,
608    "Attempt to limit compaction to this amount of memory.",
609    ParameterScope::Environment,
610);
611
612/// The maximum number of concurrent blob deletes during garbage collection.
613pub const GC_BLOB_DELETE_CONCURRENCY_LIMIT: Config<usize> = Config::new(
614    "persist_gc_blob_delete_concurrency_limit",
615    32,
616    "Limit the number of concurrent deletes GC can perform to this threshold.",
617    ParameterScope::Environment,
618);
619
620/// Kill switch for GC removing a finalized shard's old rollups. When off, they
621/// stay in state and in blob, and new rollups don't record `finalized`.
622///
623/// NOTE: Off by default until no binary that predates `HollowRollup::finalized`
624/// can read state written with it set. See the note atop `state.proto`.
625pub const GC_REMOVE_FINALIZED_ROLLUPS: Config<bool> = Config::new(
626    "persist_gc_remove_finalized_rollups",
627    false,
628    "Whether GC removes a finalized shard's old rollups from state and blob.",
629    ParameterScope::Environment,
630);
631
632/// The # of diffs to initially scan when fetching the latest consensus state, to
633/// determine which requests go down the fast vs slow path. Should be large enough
634/// to fetch all live diffs in the steady-state, and small enough to query Consensus
635/// at high volume. Steady-state usage should accommodate readers that require
636/// seqno-holds for reasonable amounts of time, which to start we say is 10s of minutes.
637///
638/// This value ought to be defined in terms of `NEED_ROLLUP_THRESHOLD` to approximate
639/// when we expect rollups to be written and therefore when old states will be truncated
640/// by GC.
641pub const STATE_VERSIONS_RECENT_LIVE_DIFFS_LIMIT: Config<usize> = Config::new(
642    "persist_state_versions_recent_live_diffs_limit",
643    30 * 128,
644    "Fetch this many diffs when fetching recent diffs.",
645    ParameterScope::Environment,
646);
647
648/// The maximum number of concurrent state fetches during usage computation.
649pub const USAGE_STATE_FETCH_CONCURRENCY_LIMIT: Config<usize> = Config::new(
650    "persist_usage_state_fetch_concurrency_limit",
651    8,
652    "Limit the concurrency in of fetching in the perioding Persist-storage-usage calculation.",
653    ParameterScope::Environment,
654);
655
656impl PostgresClientKnobs for PersistConfig {
657    fn connection_pool_max_size(&self) -> usize {
658        CONSENSUS_CONNECTION_POOL_MAX_SIZE.get(self)
659    }
660
661    fn connection_pool_max_wait(&self) -> Option<Duration> {
662        Some(CONSENSUS_CONNECTION_POOL_MAX_WAIT.get(self))
663    }
664
665    fn connection_pool_ttl(&self) -> Duration {
666        CONSENSUS_CONNECTION_POOL_TTL.get(self)
667    }
668
669    fn connection_pool_ttl_stagger(&self) -> Duration {
670        CONSENSUS_CONNECTION_POOL_TTL_STAGGER.get(self)
671    }
672
673    fn connect_timeout(&self) -> Duration {
674        CRDB_CONNECT_TIMEOUT.get(self)
675    }
676
677    fn tcp_user_timeout(&self) -> Duration {
678        CRDB_TCP_USER_TIMEOUT.get(self)
679    }
680
681    fn keepalives_idle(&self) -> Duration {
682        CRDB_KEEPALIVES_IDLE.get(self)
683    }
684
685    fn keepalives_interval(&self) -> Duration {
686        CRDB_KEEPALIVES_INTERVAL.get(self)
687    }
688
689    fn keepalives_retries(&self) -> u32 {
690        CRDB_KEEPALIVES_RETRIES.get(self)
691    }
692
693    fn statement_timeout(&self) -> Duration {
694        // Persist consensus does not use a server-side statement timeout.
695        Duration::ZERO
696    }
697}
698
699#[derive(Copy, Clone, Debug, Eq, PartialEq, Arbitrary, Serialize, Deserialize)]
700pub struct RetryParameters {
701    pub fixed_sleep: Duration,
702    pub initial_backoff: Duration,
703    pub multiplier: u32,
704    pub clamp: Duration,
705}
706
707impl RetryParameters {
708    pub fn persist_defaults() -> Self {
709        Self {
710            fixed_sleep: Duration::ZERO,
711            // Chosen to meet the following arbitrary criteria: a power of two
712            // that's close to the AWS Aurora latency of 6ms.
713            initial_backoff: Duration::from_millis(4),
714            multiplier: 2,
715            // Chosen to meet the following arbitrary criteria: between 10s and
716            // 60s.
717            clamp: Duration::from_secs(16),
718        }
719    }
720
721    pub(crate) fn into_retry(self, now: SystemTime) -> Retry {
722        let seed = now
723            .duration_since(UNIX_EPOCH)
724            .map_or(0, |x| u64::from(x.subsec_nanos()));
725        Retry {
726            fixed_sleep: self.fixed_sleep,
727            initial_backoff: self.initial_backoff,
728            multiplier: self.multiplier,
729            clamp_backoff: self.clamp,
730            seed,
731        }
732    }
733}
734
735pub(crate) const BLOB_OPERATION_TIMEOUT: Config<Duration> = Config::new(
736    "persist_blob_operation_timeout",
737    Duration::from_secs(180),
738    "Maximum time allowed for a network call, including retry attempts.",
739    ParameterScope::Environment,
740);
741
742pub(crate) const BLOB_OPERATION_ATTEMPT_TIMEOUT: Config<Duration> = Config::new(
743    "persist_blob_operation_attempt_timeout",
744    Duration::from_secs(90),
745    "Maximum time allowed for a single network call.",
746    ParameterScope::Environment,
747);
748
749pub(crate) const BLOB_CONNECT_TIMEOUT: Config<Duration> = Config::new(
750    "persist_blob_connect_timeout",
751    Duration::from_secs(7),
752    "Maximum time to wait for a socket connection to be made.",
753    ParameterScope::Environment,
754);
755
756pub(crate) const BLOB_READ_TIMEOUT: Config<Duration> = Config::new(
757    "persist_blob_read_timeout",
758    Duration::from_secs(10),
759    "Maximum time to wait to read the first byte of a response, including connection time.",
760    ParameterScope::Environment,
761);
762
763/// Which persist shard metrics this process exports, one of the
764/// [`ShardMetricsExport`] values.
765///
766/// Read on every metrics scrape, so a change takes effect on the next scrape.
767/// A scrape reads it twice, once collecting the aggregates and once filtering
768/// the families, so a change between the two mixes the modes for that scrape.
769pub const SHARD_METRICS: Config<&'static str> = Config::new(
770    "persist_shard_metrics",
771    ShardMetricsExport::default().as_str(),
772    "Which persist shard metrics to export: 'per_shard' (the `mz_persist_shard_*` families \
773    with one series per shard), 'summary' (their bounded `_percentile`, `_sum`, `_topk` and \
774    per-version aggregates), 'both', or 'none'. Shard counts and process-level counters are \
775    always exported.",
776    ParameterScope::Environment,
777);
778
779/// The values of [`SHARD_METRICS`].
780#[derive(Debug, Clone, Copy, PartialEq, Eq)]
781pub enum ShardMetricsExport {
782    None,
783    Summary,
784    PerShard,
785    Both,
786}
787
788impl ShardMetricsExport {
789    pub const ALL: [Self; 4] = [Self::None, Self::Summary, Self::PerShard, Self::Both];
790
791    /// The value [`SHARD_METRICS`] defaults to, whose series count is bounded
792    /// by the per-version aggregates rather than the shard count.
793    pub const fn default() -> Self {
794        Self::Summary
795    }
796
797    pub const fn as_str(self) -> &'static str {
798        match self {
799            Self::None => "none",
800            Self::Summary => "summary",
801            Self::PerShard => "per_shard",
802            Self::Both => "both",
803        }
804    }
805
806    /// Parses a value of [`SHARD_METRICS`], or `None` if it is unrecognized.
807    pub fn parse(s: &str) -> Option<Self> {
808        Self::ALL.into_iter().find(|v| v.as_str() == s)
809    }
810
811    /// Parses a value of [`SHARD_METRICS`], falling back to the default if it
812    /// is unrecognized.
813    ///
814    /// Silent: both callers run per scrape, so a typo would log forever.
815    /// `PerShardMetricsFilter` reports it once per change instead.
816    pub fn from_str_lossy(s: &str) -> Self {
817        Self::parse(s).unwrap_or_else(Self::default)
818    }
819
820    pub fn get(configs: &ConfigSet) -> Self {
821        Self::from_str_lossy(&SHARD_METRICS.get(configs))
822    }
823
824    /// Whether to export the per-shard families.
825    pub fn per_shard(self) -> bool {
826        matches!(self, Self::PerShard | Self::Both)
827    }
828
829    /// Whether to export the bounded aggregates of the per-shard families.
830    pub fn summary(self) -> bool {
831        matches!(self, Self::Summary | Self::Both)
832    }
833}
834
835/// Per-shard series to keep exporting while [`SHARD_METRICS`] leaves the
836/// per-shard families out.
837///
838/// A series is kept if its `shard` or `name` label matches this regex. The
839/// match is unanchored, so `s1` also matches `s10`: use `^...$` for an exact
840/// match. Empty keeps none, as does an invalid regex.
841pub const PER_SHARD_METRICS_ENABLE_REGEX: Config<&'static str> = Config::new(
842    "persist_per_shard_metrics_enable_regex",
843    "",
844    "While `persist_shard_metrics` is 'summary' or 'none', keep exporting the per-shard series \
845    whose `shard` or `name` label matches this regex (unanchored). Empty or invalid keeps \
846    none.",
847    ParameterScope::Environment,
848);
849
850/// Whether to export `mz_persist_shard_info`, one series per shard, so that
851/// dashboards can list the shards when [`SHARD_METRICS`] leaves the per-shard
852/// families out.
853pub const SHARD_INFO_METRIC: Config<bool> = Config::new(
854    "persist_shard_info_metric",
855    false,
856    "Whether to export `mz_persist_shard_info`, which lists this process's shards by `shard` \
857    and `name` with one series each.",
858    ParameterScope::Environment,
859);
860
861impl BlobKnobs for PersistConfig {
862    fn operation_timeout(&self) -> Duration {
863        BLOB_OPERATION_TIMEOUT.get(self)
864    }
865
866    fn operation_attempt_timeout(&self) -> Duration {
867        BLOB_OPERATION_ATTEMPT_TIMEOUT.get(self)
868    }
869
870    fn connect_timeout(&self) -> Duration {
871        BLOB_CONNECT_TIMEOUT.get(self)
872    }
873
874    fn read_timeout(&self) -> Duration {
875        BLOB_READ_TIMEOUT.get(self)
876    }
877
878    fn is_cc_active(&self) -> bool {
879        self.is_cc_active
880    }
881
882    fn http_pool(&self) -> BlobHttpPool {
883        BlobHttpPool::get(self)
884    }
885
886    fn http_pool_max_in_flight_per_client(&self) -> usize {
887        BLOB_HTTP_POOL_MAX_IN_FLIGHT_PER_CLIENT.get(self)
888    }
889}
890
891/// If persist gets some encoded ProtoState from the future (e.g. two versions of
892/// code are running simultaneously against the same shard), it might have a
893/// field that the current code doesn't know about. This would be silently
894/// discarded at proto decode time: our Proto library can't handle unknown fields,
895/// and old versions of code might not be able to respect the semantics of the new
896/// fields even if they did.
897///
898/// [1]: https://developers.google.com/protocol-buffers/docs/proto3#unknowns
899///
900/// To detect the bad situation and disallow it, we tag every version of state
901/// written to consensus with the version of code it's compatible with. Then at
902/// decode time, we're able to compare the current version against any we receive
903/// and assert as necessary. The current version is typically the version of code
904/// used to write the state, but it may be lower when code is intentionally emulating
905/// an older version during eg. a graceful upgrade process.
906///
907/// We could do the same for blob data, but it shouldn't be necessary. Any blob
908/// data we read is going to be because we fetched it using a pointer stored in
909/// some persist state. If we can handle the state, we can handle the blobs it
910/// references, too.
911pub fn code_can_read_data(code_version: &Version, data_version: &Version) -> bool {
912    // For now, Persist can read arbitrarily old state data.
913    // We expect to add a floor to this in future versions.
914    code_version.cmp_precedence(data_version).is_ge()
915}
916
917/// Can the given version of the code generate data that older versions can understand?
918/// Imagine the case of eg. garbage collection after a version upgrade... we may need to read old
919/// diffs to be able to find blobs to delete, even if we no longer have code to generate data in
920/// that format.
921pub fn code_can_write_data(code_version: &Version, data_version: &Version) -> bool {
922    if !code_can_read_data(code_version, data_version) {
923        return false;
924    }
925
926    if code_version.major == 0 && code_version.minor <= SELF_MANAGED_VERSIONS[1].minor {
927        // This code was added well after the last ad-hoc version was released,
928        // so we don't strictly model compatibility with earlier releases.
929        true
930    } else if code_version.major == 0 {
931        // Self-managed versions 25.2+ must be upgradeable from 25.1+.
932        SELF_MANAGED_VERSIONS[0]
933            .cmp_precedence(data_version)
934            .is_le()
935    } else if code_version.major <= 26 {
936        // Versions 26.x must be upgradeable from the last pre-1.0 release.
937        SELF_MANAGED_VERSIONS[1]
938            .cmp_precedence(data_version)
939            .is_le()
940    } else {
941        // Otherwise, the data must be from at earliest the _previous_ major version.
942        code_version.major - 1 <= data_version.major
943    }
944}
mz_persist_client/cfg.rs

mz_persist_client/
cfg.rs
