1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
//! The `SSTableReader` OPEN constructors and the single place a failed open is
//! recorded (issues #1037, #1704).
//!
//! Split out of `reader/mod.rs` per the campsite rule (#1116): that file is 1721
//! lines against the ~800-line target, and issue #1704 needed to add a constructor
//! here. Everything below is a VERBATIM move except [`SSTableReader::open_with_reporting`],
//! which is new.
//!
//! # One recording boundary per failed operation
//!
//! Every public constructor funnels into [`SSTableReader::open_with_cache_cancellable`],
//! which records a failed open into `cqlite.errors.total{category, subsystem="reader"}`.
//! That is correct when the open IS the operation. It is WRONG when the open is an
//! inner step of an operation whose own seam also records — the failure is then
//! counted twice for one user-visible failure. This is the same single-boundary rule
//! `compaction::finalize_merge_async` already states for its own inner helpers
//! ("this is an *unrecorded* inner helper ... recording here too would double-count").
//!
//! [`SSTableReader::open_with_reporting`] is the entry point for those inner steps.
use std::path::Path;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use super::sstables_open_count_for;
use super::SSTableReader;
use crate::platform::Platform;
use crate::{Config, Result};
impl SSTableReader {
/// Open an SSTable file for reading.
///
/// Instrumented (epic #1031 / #1034): wraps the open in a
/// `sstable.reader.open` span, increments the [`SSTABLES_OPEN`] gauge on
/// success, and records an error on the `reader` subsystem when open fails.
///
/// [`SSTABLES_OPEN`]: crate::observability::catalog::SSTABLES_OPEN
pub async fn open(path: &Path, config: &Config, platform: Arc<Platform>) -> Result<Self> {
// Back-compat: existing callers and sibling crates get a FRESH per-reader
// decompressed-chunk cache sized from config (issue #1567). Production
// reads route through `SSTableManager`, which calls `open_with_cache` with
// its SHARED instance so all readers of a dataset share one cache.
//
// Honor the `config.memory.block_cache.enabled` toggle (issue #1568): when
// disabled this yields a genuine no-op cache so the direct reader path
// bypasses caching identically to the manager path, instead of the toggle
// being decorative here. Reuses the shared `build_chunk_cache` helper.
let cache = super::super::build_chunk_cache(config);
Self::open_with_cache(path, config, platform, cache).await
}
/// Cancel-aware [`open`](Self::open) (issue #2383 fix C).
///
/// Threads a synchronous [`ScanCancel`](crate::storage::scan_cancel::ScanCancel)
/// into the Index.db partition-index parse so a client-disconnect cancel
/// aborts a 1.58M-entry parse within one poll interval instead of pinning a
/// tokio worker to completion. Non-cancellable callers use [`open`](Self::open)
/// (a default never-cancel flag). Returns [`Error::Cancelled`] on a mid-parse
/// trip.
pub async fn open_cancellable(
path: &Path,
config: &Config,
platform: Arc<Platform>,
cancel: crate::storage::scan_cancel::ScanCancel,
) -> Result<Self> {
let cache = super::super::build_chunk_cache(config);
Self::open_with_cache_cancellable(path, config, platform, cache, cancel).await
}
/// Open an SSTable file for reading, sharing the provided
/// [`DecompressedChunkCache`](crate::storage::cache::DecompressedChunkCache).
///
/// Identical to [`open`](Self::open) except the reader stores `cache` (an
/// `Arc` clone) instead of minting its own, so every reader a manager opens
/// for one dataset consults the same bytes-bounded chunk cache (issue #1567).
pub async fn open_with_cache(
path: &Path,
config: &Config,
platform: Arc<Platform>,
cache: Arc<crate::storage::cache::DecompressedChunkCache>,
) -> Result<Self> {
Self::open_with_cache_cancellable(
path,
config,
platform,
cache,
crate::storage::scan_cancel::ScanCancel::default(),
)
.await
}
/// Cancel-aware [`open_with_cache`](Self::open_with_cache) (issue #2383 fix C):
/// same as it but threads `cancel` into the Index.db parse. See
/// [`open_cancellable`](Self::open_cancellable).
pub async fn open_with_cache_cancellable(
path: &Path,
config: &Config,
platform: Arc<Platform>,
cache: Arc<crate::storage::cache::DecompressedChunkCache>,
cancel: crate::storage::scan_cancel::ScanCancel,
) -> Result<Self> {
Self::open_instrumented(
path,
config,
platform,
cache,
cancel,
OpenErrorReporting::SelfReported,
)
.await
}
/// [`open`](Self::open) with the failed-open reporting mode stated BY THE CALLER
/// (issue #1704).
///
/// # Why this is a parameter and not a second named function
///
/// Both mistakes are silent. [`OpenErrorReporting::SelfReported`] under a caller
/// that also records counts one failure TWICE;
/// [`OpenErrorReporting::DeferredToCaller`] under a caller with no seam records it
/// ZERO times, which is worse and is exactly the regression a universally
/// non-recording open introduced. Only the caller knows which it is, so it must
/// say — an unparameterised `open_unrecorded` encoded a precondition that nothing
/// at the call site could check.
///
/// # Behavioural identity, which is now real rather than claimed
///
/// Exactly ONE action differs from [`open`](Self::open): the `record_error` call on
/// the `Err` arm. Both routes go through [`open_instrumented`](Self::open_instrumented),
/// so the `sstable.reader.open` span, its `file_size`/`sstable_format` fields, the
/// parenting of the nested open-phase spans, and the `SSTABLES_OPEN` gauge are
/// literally the same code. The first version of this function called `open_inner`
/// directly and silently dropped the span while its doc claimed identity — the
/// shared helper is what makes the claim checkable instead of aspirational.
#[cfg(any(feature = "delta-scan", feature = "write-support"))]
pub(crate) async fn open_with_reporting(
path: &Path,
config: &Config,
platform: Arc<Platform>,
reporting: OpenErrorReporting,
) -> Result<Self> {
let cache = super::super::build_chunk_cache(config);
Self::open_instrumented(
path,
config,
platform,
cache,
crate::storage::scan_cancel::ScanCancel::default(),
reporting,
)
.await
}
/// The ONE open implementation: span, success gauge, and — for
/// [`OpenErrorReporting::SelfReported`] only — the failed-open increment.
async fn open_instrumented(
path: &Path,
config: &Config,
platform: Arc<Platform>,
cache: Arc<crate::storage::cache::DecompressedChunkCache>,
cancel: crate::storage::scan_cancel::ScanCancel,
reporting: OpenErrorReporting,
) -> Result<Self> {
use crate::observability::{self as obs, catalog};
use tracing::Instrument as _;
let span = tracing::debug_span!(
"sstable.reader.open",
file_size = tracing::field::Empty,
sstable_format = tracing::field::Empty,
);
// Instrument the future rather than holding an entered guard across the
// `.await`: entering a span guard and then awaiting can attach unrelated
// async work scheduled on this task to the span. `Instrument` enters the
// span only while this specific future is polled.
let result = Self::open_inner(path, config, platform, cache, cancel)
.instrument(span.clone())
.await;
match &result {
Ok(reader) => {
let format = reader.sstable_format_label();
span.record("file_size", reader.stats.file_size);
span.record("sstable_format", format);
// SSTABLES_OPEN is a snapshot gauge of the live reader count;
// record the current PER-FORMAT count after this open succeeds so
// the format-attributed gauge series stays correct under mixed
// BIG/BTI readers. A SUCCESSFUL open is identical in both modes:
// the reporting mode is only ever about who counts a FAILURE.
let now = sstables_open_count_for(format).fetch_add(1, Ordering::Relaxed) + 1;
obs::record_gauge(
catalog::SSTABLES_OPEN,
now,
&[(catalog::attr::SSTABLE_FORMAT, format.into())],
);
}
Err(e) => {
// Both arms run WHILE the open span is current: the instrumented
// future has already completed, so the span is no longer entered and
// `in_scope` re-enters it for the duration of the call.
//
// Only the COUNTER is deferred (issue #1704). The SPAN is this call's
// own and must be marked either way — suppressing all of
// `record_error` also suppressed its `mark_span_error`, so a deferred
// open returned `Err` under a span that still looked successful. Same
// shape as roborev E, where this function dropped the span outright:
// an observability change must not quietly lose span fidelity.
span.in_scope(|| match reporting {
OpenErrorReporting::SelfReported => obs::record_error(e, "reader"),
OpenErrorReporting::DeferredToCaller => obs::mark_span_error(e),
});
}
}
result
}
}
/// Who counts a failed open into `cqlite.errors.total{subsystem="reader"}`
/// (issue #1704). The ONLY thing that varies between the two open routes.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum OpenErrorReporting {
/// This open IS the operation: it records its own failure. The mode every
/// PUBLIC entry point uses, so a caller who does not think about it is never
/// silently un-instrumented.
SelfReported,
/// This open is an inner step; the caller's operation seam records instead.
///
/// Constructed by `scan_delta` (feature `delta-scan`) and by the cross-generation
/// streaming merge (`not(feature = "tombstones")`). A build with `tombstones` on
/// and `delta-scan` off therefore has no constructor for it — the variant is still
/// the correct half of the distinction, so it is allowed rather than cfg'd away
/// into a shape that would differ per feature set.
#[allow(dead_code)]
DeferredToCaller,
}