Skip to main content

loonfs_objectstore/
metrics.rs

1//! A metrics-recording wrapper around any object store: per-operation
2//! samples classified by object family.
3
4use crate::attempts::counting_attempts;
5use crate::layout::{parse_object_key, DurableObjectFamily};
6use crate::object_store::Result;
7use crate::{
8    ByteRange, ByteStream, MultipartCompletion, MultipartPart, ObjectBody, ObjectMetadata,
9    ObjectStore, ObjectStoreError, PutMode, StoredObjectChecksum,
10};
11use async_trait::async_trait;
12use bytes::Bytes;
13use futures::stream::{BoxStream, TryStreamExt};
14use loonfs_api::StorageChecksum;
15use serde::{Deserialize, Serialize};
16use std::fmt;
17use std::fs::{self, File};
18use std::io::{self, BufWriter, Write};
19use std::path::Path;
20use std::sync::{Arc, Mutex};
21use std::time::{Duration, Instant};
22
23/// One object-store call sample delivered to an object-store metrics recorder.
24///
25/// Samples intentionally classify keys and errors instead of exposing raw object keys or provider
26/// error strings.
27#[derive(Debug, Clone, Serialize, Deserialize)]
28pub struct ObjectStoreMetricSample {
29    /// Contract operation that produced the sample.
30    pub operation: ObjectStoreOperation,
31    /// End-to-end operation latency in microseconds, including provider waits.
32    pub elapsed_micros: u128,
33    /// Provider attempts this one call made, counting the first: `1` is a
34    /// call that never retried.
35    ///
36    /// Retries are what makes an otherwise unexplained `elapsed_micros`
37    /// readable. The count covers the bounded retry loops inside the store
38    /// this wrapper measures; a retry loop that sits *above* the wrapper —
39    /// [`crate::ObjectStore::put_immutable_verified`] is the one — surfaces
40    /// as one sample per attempt instead, because each of its attempts
41    /// really is a separate measured call.
42    pub attempts: u32,
43    /// Cardinality-bounded success or failure classification.
44    pub result: ObjectStoreResultClass,
45    /// Request payload bytes for a put, including failed attempts; otherwise `None`.
46    pub bytes_in: Option<u64>,
47    /// Response payload bytes for a successful get, including zero; otherwise `None`.
48    pub bytes_out: Option<u64>,
49    /// Keys yielded by a listing before completion or drop; otherwise `None`.
50    pub item_count: Option<u64>,
51    /// Durable-family grouping derived without retaining the raw key.
52    pub key_class: KeyClass,
53    /// Read-range or listing shape when applicable; otherwise `None`.
54    pub range_class: Option<RangeClass>,
55    /// Requested conditional-write class for a put; otherwise `None`.
56    pub put_mode: Option<PutModeClass>,
57    /// Deployment-supplied provider label, or `None` when the wrapper was left unlabeled.
58    pub store_kind: Option<String>,
59}
60
61/// Classifies the method measured by one object-store sample.
62#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
63#[serde(rename_all = "snake_case")]
64pub enum ObjectStoreOperation {
65    /// Measures a metadata-only point read.
66    Head,
67    /// Measures a self-consistent full-object read with identity metadata.
68    GetWithMetadata,
69    /// Measures a full or ranged byte read.
70    Get,
71    /// Measures an overwrite or conditional write.
72    Put,
73    /// Measures a write whose payload arrived as a stream.
74    PutStreamed,
75    /// Measures an idempotent object delete.
76    Delete,
77    /// Measures opening a client-driven multipart upload.
78    CreateMultipartUpload,
79    /// Measures asking a provider to assemble a multipart upload.
80    CompleteMultipartUpload,
81    /// Measures abandoning a multipart upload and its parts.
82    AbortMultipartUpload,
83    /// Measures a listing collected to completion.
84    ListPrefix,
85    /// Measures a listing stream until completion or early drop.
86    ListPrefixStream,
87}
88
89impl ObjectStoreOperation {
90    /// The label an aggregating recorder groups by. Identical to the serde
91    /// name: one operation has one spelling wherever it is reported.
92    pub fn as_str(self) -> &'static str {
93        match self {
94            Self::Head => "head",
95            Self::GetWithMetadata => "get_with_metadata",
96            Self::Get => "get",
97            Self::Put => "put",
98            Self::PutStreamed => "put_streamed",
99            Self::Delete => "delete",
100            Self::CreateMultipartUpload => "create_multipart_upload",
101            Self::CompleteMultipartUpload => "complete_multipart_upload",
102            Self::AbortMultipartUpload => "abort_multipart_upload",
103            Self::ListPrefix => "list_prefix",
104            Self::ListPrefixStream => "list_prefix_stream",
105        }
106    }
107}
108
109/// Collapses object-store outcomes into a bounded metrics vocabulary.
110#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
111#[serde(rename_all = "snake_case")]
112pub enum ObjectStoreResultClass {
113    /// Indicates the operation returned its requested value or mutation result.
114    Ok,
115    /// Indicates an optional read observed absence or an explicit lookup returned not found.
116    NotFound,
117    /// Indicates key or prefix validation failed before provider IO.
118    InvalidKey,
119    /// Indicates a content reference could not resolve to an immutable key.
120    InvalidContentRef,
121    /// Indicates requested byte-range bounds were invalid for the object.
122    InvalidRange,
123    /// Indicates a create-if-absent or compare-and-swap condition did not hold.
124    PreconditionFailed,
125    /// Indicates the provider rejected identity or authorization.
126    PermissionDenied,
127    /// Indicates the configured store lacks a required capability.
128    Unsupported,
129    /// Indicates configuration, IO, timeout, protocol, or provider transport failure.
130    Transport,
131    /// Reserves a forward-compatible bucket for errors outside the current registry.
132    OtherError,
133}
134
135impl ObjectStoreResultClass {
136    /// The label an aggregating recorder groups by. Identical to the serde
137    /// name: one outcome has one spelling wherever it is reported.
138    pub fn as_str(self) -> &'static str {
139        match self {
140            Self::Ok => "ok",
141            Self::NotFound => "not_found",
142            Self::InvalidKey => "invalid_key",
143            Self::InvalidContentRef => "invalid_content_ref",
144            Self::InvalidRange => "invalid_range",
145            Self::PreconditionFailed => "precondition_failed",
146            Self::PermissionDenied => "permission_denied",
147            Self::Unsupported => "unsupported",
148            Self::Transport => "transport",
149            Self::OtherError => "other_error",
150        }
151    }
152}
153
154/// Groups durable keys into low-cardinality operational families.
155#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
156#[serde(rename_all = "snake_case")]
157pub enum KeyClass {
158    /// Groups immutable whole-file byte objects.
159    Content,
160    /// Groups small control records not assigned a more specific class.
161    Metadata,
162    /// Groups the authoritative namespace WAL head.
163    NamespaceHead,
164    /// Groups immutable WAL segment payloads.
165    WalSegment,
166    /// Groups namespace manifests and their mutable root pointer.
167    NamespaceManifest,
168    /// Groups immutable metadata SST segments.
169    MetadataSst,
170    /// Groups checkpoint records and retained-history floors consulted by garbage collection.
171    GcControl,
172    /// Groups unrecognized keys and coarse listing prefixes.
173    Unknown,
174}
175
176/// Classifies the shape of bytes requested by a get or listing.
177#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
178#[serde(rename_all = "snake_case")]
179pub enum RangeClass {
180    /// Indicates a get without explicit range bounds.
181    FullObject,
182    /// Indicates bytes starting at zero or a prefix-listing operation.
183    Prefix,
184    /// Indicates a range extending from a nonzero start to the sentinel maximum end.
185    Suffix,
186    /// Indicates a finite nonempty range with both bounds inside the keyspace.
187    Bounded,
188    /// Indicates equal range bounds and therefore a zero-byte read.
189    Empty,
190}
191
192/// Classifies the precondition semantics requested by a put.
193#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
194#[serde(rename_all = "snake_case")]
195pub enum PutModeClass {
196    /// Indicates an unconditional replacement.
197    Overwrite,
198    /// Indicates creation only while the key is absent.
199    CreateIfAbsent,
200    /// Indicates replacement only while an opaque compare token matches.
201    CompareAndSwap,
202}
203
204/// Receives object-store metrics samples from `InstrumentedObjectStore`.
205///
206/// Implementations should aggregate or export samples without blocking the object-store hot path.
207pub trait ObjectStoreMetricsRecorder: Send + Sync + 'static {
208    /// Accepts one completed sample without access to raw keys or provider error text.
209    fn record(&self, sample: ObjectStoreMetricSample);
210}
211
212/// In-memory recorder intended for tests and small local diagnostics.
213#[derive(Default)]
214pub struct VecObjectStoreMetricsRecorder {
215    samples: Mutex<Vec<ObjectStoreMetricSample>>,
216}
217
218impl VecObjectStoreMetricsRecorder {
219    /// Returns a snapshot copy of every sample recorded so far.
220    pub fn samples(&self) -> Vec<ObjectStoreMetricSample> {
221        self.samples
222            .lock()
223            .unwrap_or_else(|poisoned| poisoned.into_inner())
224            .clone()
225    }
226}
227
228impl ObjectStoreMetricsRecorder for VecObjectStoreMetricsRecorder {
229    fn record(&self, sample: ObjectStoreMetricSample) {
230        self.samples
231            .lock()
232            .unwrap_or_else(|poisoned| poisoned.into_inner())
233            .push(sample);
234    }
235}
236
237/// Buffered JSONL recorder for process-level diagnostics and benchmarks.
238///
239/// Recording is best-effort: I/O errors while writing a sample are ignored so instrumentation does
240/// not change object-store behavior.
241pub struct JsonlObjectStoreMetricsRecorder {
242    writer: Mutex<BufWriter<File>>,
243}
244
245impl JsonlObjectStoreMetricsRecorder {
246    /// Creates or truncates a JSONL output file, creating missing parent directories.
247    ///
248    /// The operation fails when directories or the output file cannot be created.
249    pub fn create(path: impl AsRef<Path>) -> io::Result<Self> {
250        let path = path.as_ref();
251        if let Some(parent) = path.parent() {
252            if !parent.as_os_str().is_empty() {
253                fs::create_dir_all(parent)?;
254            }
255        }
256        Ok(Self {
257            writer: Mutex::new(BufWriter::new(File::create(path)?)),
258        })
259    }
260
261    /// Flushes buffered samples to the underlying file.
262    ///
263    /// The operation fails when the filesystem cannot accept pending bytes.
264    pub fn flush(&self) -> io::Result<()> {
265        self.writer
266            .lock()
267            .unwrap_or_else(|poisoned| poisoned.into_inner())
268            .flush()
269    }
270}
271
272impl ObjectStoreMetricsRecorder for JsonlObjectStoreMetricsRecorder {
273    fn record(&self, sample: ObjectStoreMetricSample) {
274        let mut writer = self
275            .writer
276            .lock()
277            .unwrap_or_else(|poisoned| poisoned.into_inner());
278        let _ = serde_json::to_writer(&mut *writer, &sample);
279        let _ = writer.write_all(b"\n");
280    }
281}
282
283/// Records one bounded-cardinality sample around each operation on an inner store.
284///
285/// Results and storage semantics pass through unchanged; streamed listings emit
286/// their sample when the stream is completed or dropped.
287pub struct InstrumentedObjectStore<S> {
288    inner: S,
289    recorder: Arc<dyn ObjectStoreMetricsRecorder>,
290    store_kind: Option<String>,
291}
292
293impl<S: fmt::Debug> fmt::Debug for InstrumentedObjectStore<S> {
294    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
295        f.debug_struct("InstrumentedObjectStore")
296            .field("inner", &self.inner)
297            .field("store_kind", &self.store_kind)
298            .finish_non_exhaustive()
299    }
300}
301
302impl<S> InstrumentedObjectStore<S> {
303    /// Wraps a store with the supplied synchronous recorder and no provider label.
304    pub fn new(inner: S, recorder: Arc<dyn ObjectStoreMetricsRecorder>) -> Self {
305        Self {
306            inner,
307            recorder,
308            store_kind: None,
309        }
310    }
311
312    /// Attaches a low-cardinality provider label copied into subsequent samples.
313    pub fn store_kind(mut self, store_kind: impl Into<String>) -> Self {
314        self.store_kind = Some(store_kind.into());
315        self
316    }
317
318    /// Removes instrumentation and returns ownership of the wrapped store.
319    pub fn into_inner(self) -> S {
320        self.inner
321    }
322}
323
324#[allow(clippy::disallowed_methods)]
325fn sample_clock() -> Instant {
326    // Durations only: sampling reads the monotonic clock at this recorder
327    // boundary, so no wall-clock reading reaches protocol code from here.
328    Instant::now()
329}
330
331#[async_trait]
332impl<S> ObjectStore for InstrumentedObjectStore<S>
333where
334    S: ObjectStore,
335{
336    async fn head(&self, key: &str) -> Result<Option<ObjectMetadata>> {
337        let start = sample_clock();
338        let (result, attempts) = counting_attempts(self.inner.head(key)).await;
339        self.record_head_like(
340            ObjectStoreOperation::Head,
341            key,
342            start.elapsed(),
343            attempts,
344            &result,
345        );
346        result
347    }
348
349    async fn head_stored_checksum(&self, key: &str) -> Result<Option<StoredObjectChecksum>> {
350        let start = sample_clock();
351        let (result, attempts) = counting_attempts(self.inner.head_stored_checksum(key)).await;
352        // One provider metadata request, recorded as the head it is: the
353        // point of this call is that it moves no payload.
354        self.record_head_like(
355            ObjectStoreOperation::Head,
356            key,
357            start.elapsed(),
358            attempts,
359            &result,
360        );
361        result
362    }
363
364    async fn create_multipart_upload(&self, key: &str) -> Result<String> {
365        let start = sample_clock();
366        let (result, attempts) = counting_attempts(self.inner.create_multipart_upload(key)).await;
367        // The multipart control calls move no payload of their own: the
368        // parts travel from the client straight to the provider. Timing them
369        // is the only thing there is to record.
370        self.record_unit(
371            ObjectStoreOperation::CreateMultipartUpload,
372            key,
373            start.elapsed(),
374            attempts,
375            &result,
376        );
377        result
378    }
379
380    async fn complete_multipart_upload(
381        &self,
382        key: &str,
383        provider_upload_id: &str,
384        parts: &[MultipartPart],
385        full_object_checksum: &StorageChecksum,
386    ) -> Result<MultipartCompletion> {
387        let start = sample_clock();
388        let (result, attempts) = counting_attempts(self.inner.complete_multipart_upload(
389            key,
390            provider_upload_id,
391            parts,
392            full_object_checksum,
393        ))
394        .await;
395        self.record_unit(
396            ObjectStoreOperation::CompleteMultipartUpload,
397            key,
398            start.elapsed(),
399            attempts,
400            &result,
401        );
402        result
403    }
404
405    async fn abort_multipart_upload(&self, key: &str, provider_upload_id: &str) -> Result<()> {
406        let start = sample_clock();
407        let (result, attempts) =
408            counting_attempts(self.inner.abort_multipart_upload(key, provider_upload_id)).await;
409        self.record_unit(
410            ObjectStoreOperation::AbortMultipartUpload,
411            key,
412            start.elapsed(),
413            attempts,
414            &result,
415        );
416        result
417    }
418
419    async fn get(&self, key: &str, range: Option<ByteRange>) -> Result<Option<Bytes>> {
420        let start = sample_clock();
421        let (result, attempts) = counting_attempts(self.inner.get(key, range.clone())).await;
422        self.record_get(key, range.as_ref(), start.elapsed(), attempts, &result);
423        result
424    }
425
426    async fn get_with_metadata(&self, key: &str) -> Result<Option<ObjectBody>> {
427        let start = sample_clock();
428        let (result, attempts) = counting_attempts(self.inner.get_with_metadata(key)).await;
429        self.record_get_with_metadata(key, start.elapsed(), attempts, &result);
430        result
431    }
432
433    async fn put(&self, key: &str, bytes: Bytes, mode: PutMode) -> Result<ObjectMetadata> {
434        let start = sample_clock();
435        let bytes_in = bytes.len() as u64;
436        let (result, attempts) = counting_attempts(self.inner.put(key, bytes, mode.clone())).await;
437        self.record_put(key, bytes_in, &mode, start.elapsed(), attempts, &result);
438        result
439    }
440
441    /// Streamed writes get their own operation rather than being folded
442    /// into `put`: their request bytes are only known once the stream ends,
443    /// and a deployment reading these samples needs to see which write path
444    /// its content is taking.
445    async fn put_streamed(&self, key: &str, body: ByteStream, mode: PutMode) -> Result<u64> {
446        let start = sample_clock();
447        let (result, attempts) =
448            counting_attempts(self.inner.put_streamed(key, body, mode.clone())).await;
449        self.record(ObjectStoreMetricSample {
450            operation: ObjectStoreOperation::PutStreamed,
451            elapsed_micros: start.elapsed().as_micros(),
452            attempts,
453            result: classify_result(&result),
454            bytes_in: result.as_ref().ok().copied(),
455            bytes_out: None,
456            item_count: None,
457            key_class: classify_key(key),
458            range_class: None,
459            put_mode: Some(classify_put_mode(&mode)),
460            store_kind: self.store_kind.clone(),
461        });
462        result
463    }
464
465    async fn delete(&self, key: &str) -> Result<()> {
466        let start = sample_clock();
467        let (result, attempts) = counting_attempts(self.inner.delete(key)).await;
468        self.record_unit(
469            ObjectStoreOperation::Delete,
470            key,
471            start.elapsed(),
472            attempts,
473            &result,
474        );
475        result
476    }
477
478    fn list_prefix_stream(&self, prefix: &str) -> BoxStream<'static, Result<String>> {
479        // Streamed listings (WAL replay, GC) must not be invisible in the
480        // metrics. The wrapper records one sample when the stream is
481        // dropped — finished or abandoned — carrying the item count and
482        // the first error's class.
483        Box::pin(RecordedListStream {
484            inner: self.inner.list_prefix_stream(prefix),
485            recorder: Arc::clone(&self.recorder),
486            store_kind: self.store_kind.clone(),
487            key_class: classify_key(prefix),
488            started: sample_clock(),
489            items: 0,
490            first_error: None,
491        })
492    }
493
494    async fn list_prefix(&self, prefix: &str) -> Result<Vec<String>> {
495        let start = sample_clock();
496        let (result, attempts): (Result<Vec<_>>, u32) = counting_attempts(async {
497            self.inner
498                .list_prefix_stream(prefix)
499                .try_collect()
500                .await
501                .map(|mut keys: Vec<String>| {
502                    keys.sort();
503                    keys
504                })
505        })
506        .await;
507        self.record_list(prefix, start.elapsed(), attempts, &result);
508        result
509    }
510}
511
512impl<S> InstrumentedObjectStore<S> {
513    fn record_head_like<T>(
514        &self,
515        operation: ObjectStoreOperation,
516        key: &str,
517        elapsed: Duration,
518        attempts: u32,
519        result: &Result<Option<T>>,
520    ) {
521        self.record(ObjectStoreMetricSample {
522            operation,
523            elapsed_micros: elapsed.as_micros(),
524            attempts,
525            result: classify_optional_result(result),
526            bytes_in: None,
527            bytes_out: None,
528            item_count: None,
529            key_class: classify_key(key),
530            range_class: None,
531            put_mode: None,
532            store_kind: self.store_kind.clone(),
533        });
534    }
535
536    fn record_get(
537        &self,
538        key: &str,
539        range: Option<&ByteRange>,
540        elapsed: Duration,
541        attempts: u32,
542        result: &Result<Option<Bytes>>,
543    ) {
544        self.record(ObjectStoreMetricSample {
545            operation: ObjectStoreOperation::Get,
546            elapsed_micros: elapsed.as_micros(),
547            attempts,
548            result: classify_optional_result(result),
549            bytes_in: None,
550            bytes_out: result
551                .as_ref()
552                .ok()
553                .and_then(|bytes| bytes.as_ref().map(|bytes| bytes.len() as u64)),
554            item_count: None,
555            key_class: classify_key(key),
556            range_class: Some(classify_range(range)),
557            put_mode: None,
558            store_kind: self.store_kind.clone(),
559        });
560    }
561
562    fn record_get_with_metadata(
563        &self,
564        key: &str,
565        elapsed: Duration,
566        attempts: u32,
567        result: &Result<Option<ObjectBody>>,
568    ) {
569        self.record(ObjectStoreMetricSample {
570            operation: ObjectStoreOperation::GetWithMetadata,
571            elapsed_micros: elapsed.as_micros(),
572            attempts,
573            result: classify_optional_result(result),
574            bytes_in: None,
575            bytes_out: result
576                .as_ref()
577                .ok()
578                .and_then(|body| body.as_ref().map(|body| body.bytes.len() as u64)),
579            item_count: None,
580            key_class: classify_key(key),
581            range_class: Some(RangeClass::FullObject),
582            put_mode: None,
583            store_kind: self.store_kind.clone(),
584        });
585    }
586
587    fn record_put(
588        &self,
589        key: &str,
590        bytes_in: u64,
591        mode: &PutMode,
592        elapsed: Duration,
593        attempts: u32,
594        result: &Result<ObjectMetadata>,
595    ) {
596        self.record(ObjectStoreMetricSample {
597            operation: ObjectStoreOperation::Put,
598            elapsed_micros: elapsed.as_micros(),
599            attempts,
600            result: classify_result(result),
601            bytes_in: Some(bytes_in),
602            bytes_out: None,
603            item_count: None,
604            key_class: classify_key(key),
605            range_class: None,
606            put_mode: Some(classify_put_mode(mode)),
607            store_kind: self.store_kind.clone(),
608        });
609    }
610
611    fn record_unit<T>(
612        &self,
613        operation: ObjectStoreOperation,
614        key: &str,
615        elapsed: Duration,
616        attempts: u32,
617        result: &Result<T>,
618    ) {
619        self.record(ObjectStoreMetricSample {
620            operation,
621            elapsed_micros: elapsed.as_micros(),
622            attempts,
623            result: classify_result(result),
624            bytes_in: None,
625            bytes_out: None,
626            item_count: None,
627            key_class: classify_key(key),
628            range_class: None,
629            put_mode: None,
630            store_kind: self.store_kind.clone(),
631        });
632    }
633
634    fn record_list(
635        &self,
636        prefix: &str,
637        elapsed: Duration,
638        attempts: u32,
639        result: &Result<Vec<String>>,
640    ) {
641        self.record(ObjectStoreMetricSample {
642            operation: ObjectStoreOperation::ListPrefix,
643            elapsed_micros: elapsed.as_micros(),
644            attempts,
645            result: classify_result(result),
646            bytes_in: None,
647            bytes_out: None,
648            item_count: result.as_ref().ok().map(|items| items.len() as u64),
649            key_class: classify_key(prefix),
650            range_class: Some(RangeClass::Prefix),
651            put_mode: None,
652            store_kind: self.store_kind.clone(),
653        });
654    }
655
656    fn record(&self, sample: ObjectStoreMetricSample) {
657        self.recorder.record(sample);
658    }
659}
660
661struct RecordedListStream {
662    inner: BoxStream<'static, Result<String>>,
663    recorder: Arc<dyn ObjectStoreMetricsRecorder>,
664    store_kind: Option<String>,
665    key_class: KeyClass,
666    started: Instant,
667    items: u64,
668    first_error: Option<ObjectStoreResultClass>,
669}
670
671impl futures::Stream for RecordedListStream {
672    type Item = Result<String>;
673
674    fn poll_next(
675        mut self: std::pin::Pin<&mut Self>,
676        cx: &mut std::task::Context<'_>,
677    ) -> std::task::Poll<Option<Self::Item>> {
678        let polled = self.inner.as_mut().poll_next(cx);
679        if let std::task::Poll::Ready(Some(item)) = &polled {
680            match item {
681                Ok(_) => self.items += 1,
682                Err(error) => {
683                    if self.first_error.is_none() {
684                        self.first_error = Some(classify_error(error));
685                    }
686                }
687            }
688        }
689        polled
690    }
691}
692
693impl Drop for RecordedListStream {
694    fn drop(&mut self) {
695        self.recorder.record(ObjectStoreMetricSample {
696            operation: ObjectStoreOperation::ListPrefixStream,
697            elapsed_micros: self.started.elapsed().as_micros(),
698            // A listing stream is polled by whoever holds it, across tasks a
699            // tally cannot follow, and no LoonFS-owned retry gate sits on
700            // the listing path: what retries a provider client does there
701            // are its own and were never countable from here.
702            attempts: 1,
703            result: self.first_error.unwrap_or(ObjectStoreResultClass::Ok),
704            bytes_in: None,
705            bytes_out: None,
706            item_count: Some(self.items),
707            key_class: self.key_class,
708            range_class: None,
709            put_mode: None,
710            store_kind: self.store_kind.clone(),
711        });
712    }
713}
714
715fn classify_key(key: &str) -> KeyClass {
716    let Some(parsed) = parse_object_key(key) else {
717        return KeyClass::Unknown;
718    };
719
720    match parsed.family() {
721        DurableObjectFamily::ContentBlob => KeyClass::Content,
722        DurableObjectFamily::WalHead => KeyClass::NamespaceHead,
723        DurableObjectFamily::WalSegment => KeyClass::WalSegment,
724        DurableObjectFamily::MetadataManifest => KeyClass::NamespaceManifest,
725        DurableObjectFamily::MetadataTable => KeyClass::MetadataSst,
726        DurableObjectFamily::CheckpointRecord | DurableObjectFamily::WalFloor => {
727            KeyClass::GcControl
728        }
729        DurableObjectFamily::MetadataRoot => KeyClass::NamespaceManifest,
730        DurableObjectFamily::UploadSession => KeyClass::Metadata,
731    }
732}
733
734fn classify_optional_result<T>(result: &Result<Option<T>>) -> ObjectStoreResultClass {
735    match result {
736        Ok(Some(_)) => ObjectStoreResultClass::Ok,
737        Ok(None) => ObjectStoreResultClass::NotFound,
738        Err(error) => classify_error(error),
739    }
740}
741
742fn classify_result<T>(result: &Result<T>) -> ObjectStoreResultClass {
743    match result {
744        Ok(_) => ObjectStoreResultClass::Ok,
745        Err(error) => classify_error(error),
746    }
747}
748
749fn classify_error(error: &ObjectStoreError) -> ObjectStoreResultClass {
750    match error {
751        ObjectStoreError::NotFound { .. } => ObjectStoreResultClass::NotFound,
752        ObjectStoreError::InvalidKey { .. } => ObjectStoreResultClass::InvalidKey,
753        ObjectStoreError::InvalidContentRef(_) => ObjectStoreResultClass::InvalidContentRef,
754        ObjectStoreError::InvalidRange { .. } => ObjectStoreResultClass::InvalidRange,
755        ObjectStoreError::PreconditionFailed { .. } => ObjectStoreResultClass::PreconditionFailed,
756        ObjectStoreError::PermissionDenied { .. } => ObjectStoreResultClass::PermissionDenied,
757        ObjectStoreError::Unsupported(_) => ObjectStoreResultClass::Unsupported,
758        // Configuration failures happen at store construction, before any
759        // metered operation; classify defensively as transport.
760        ObjectStoreError::Configuration(_) => ObjectStoreResultClass::Transport,
761        ObjectStoreError::Transport { .. } => ObjectStoreResultClass::Transport,
762    }
763}
764
765fn classify_range(range: Option<&ByteRange>) -> RangeClass {
766    let Some(range) = range else {
767        return RangeClass::FullObject;
768    };
769    if range.start_inclusive == range.end_exclusive {
770        RangeClass::Empty
771    } else if range.start_inclusive == 0 {
772        RangeClass::Prefix
773    } else if range.end_exclusive == u64::MAX {
774        RangeClass::Suffix
775    } else {
776        RangeClass::Bounded
777    }
778}
779
780fn classify_put_mode(mode: &PutMode) -> PutModeClass {
781    match mode {
782        PutMode::Overwrite => PutModeClass::Overwrite,
783        PutMode::CreateIfAbsent => PutModeClass::CreateIfAbsent,
784        PutMode::CompareAndSwap { .. } => PutModeClass::CompareAndSwap,
785    }
786}