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

mz_persist_client/
cfg.rs
