objectstore_service/change_stream/
mod.rs

1//! The change stream each storage backend publishes.
2//!
3//! A backend describes its cost-tracking reporting with a [`CostTrackerStreamConfig`];
4//! the service describes where those records go with a [`CostTrackerConfig`], shared by
5//! every backend. [`ChangeStreamFactory`] pairs the two into a [`ChangeStream`].
6//!
7//! Behind the `storage-cogs` feature. Without it every backend gets a [`NoopStream`] and
8//! the transport is left out of the binary.
9
10use std::fmt;
11use std::sync::Arc;
12use std::time::{Duration, SystemTime};
13
14use serde::{Deserialize, Serialize};
15
16use crate::id::ObjectId;
17
18#[cfg(feature = "storage-cogs")]
19mod cost_tracker;
20mod factory;
21
22#[cfg(feature = "storage-cogs")]
23pub use cost_tracker::CostTrackerStream;
24pub use factory::ChangeStreamFactory;
25#[cfg(feature = "storage-cogs")]
26pub use factory::CostTrackerConfig;
27
28#[cfg(all(test, feature = "storage-cogs"))]
29pub(crate) use factory::dummy_factory;
30
31/// How long a backend waits for reported records to be handed off during shutdown.
32pub const FLUSH_TIMEOUT: Duration = Duration::from_secs(2);
33
34/// Scope key holding the Sentry organization ID.
35#[cfg(feature = "storage-cogs")]
36const SCOPE_ORGANIZATION: &str = "org";
37/// Scope key holding the Sentry project ID.
38#[cfg(feature = "storage-cogs")]
39const SCOPE_PROJECT: &str = "project";
40
41/// What a single backend reports for cost tracking, and how much of it.
42///
43/// A backend without one reports nothing.
44///
45/// # Example
46///
47/// ```yaml
48/// storage_cogs:
49///   shared_resource_id: bigtable_objectstore
50///   sample_rate: 1.0
51/// ```
52#[derive(Debug, Clone, Deserialize, Serialize)]
53pub struct CostTrackerStreamConfig {
54    /// Identifies the storage backend resource.
55    ///
56    /// This is meant to correspond to a `shared_resource_id` label on a provisioned
57    /// storage resource so that change stream data can be joined with other data about
58    /// the storage resource.
59    pub shared_resource_id: String,
60
61    /// Proportion of records to report, in `[0, 1]`.
62    ///
63    /// `1.0` reports every change. It can be lowered if the stream is under too much load
64    /// but beware: when the sample rate decreases, records that used to be tracked will
65    /// no longer be tracked. Stream consumers may have inconsistent state for them until
66    /// they expire.
67    #[serde(default = "default_sample_rate")]
68    pub sample_rate: f64,
69}
70
71/// Reports everything by default.
72fn default_sample_rate() -> f64 {
73    1.0
74}
75
76/// Publishes the changes a single backend makes to the objects it stores.
77///
78/// See [module docs](self).
79#[async_trait::async_trait]
80pub trait ChangeStream: fmt::Debug + Send + Sync + 'static {
81    /// Reports that `id` now occupies `size` bytes. Used for new writes and overwrites.
82    fn write(&self, id: &ObjectId, size: u64, expires_at: Option<SystemTime>);
83
84    /// Reports that `id`'s expiration moved, with its stored size unchanged.
85    fn update(&self, id: &ObjectId, expires_at: Option<SystemTime>);
86
87    /// Reports that `id` was deleted explicitly. Does not account for automatic GC.
88    fn delete(&self, id: &ObjectId);
89
90    /// Blocks until reported records have been delivered, or `timeout` elapses.
91    ///
92    /// Call this during shutdown to drain the change stream queue.
93    /// Waits for reported records to be delivered, or until `timeout` elapses.
94    ///
95    /// Awaited from [`Backend::join`](crate::backend::common::Backend::join) so records
96    /// reported just before shutdown are not lost.
97    async fn join(&self, timeout: Duration);
98}
99
100/// Drains `change_stream`, bounded by [`FLUSH_TIMEOUT`].
101///
102/// Backends call this from [`Backend::join`](crate::backend::common::Backend::join) so
103/// records reported just before shutdown are not silently lost.
104pub async fn flush_change_stream(change_stream: &Arc<dyn ChangeStream>) {
105    change_stream.join(FLUSH_TIMEOUT).await;
106}
107
108/// A [`ChangeStream`] that reports nothing.
109#[derive(Clone, Copy, Debug, Default)]
110pub struct NoopStream;
111
112#[async_trait::async_trait]
113impl ChangeStream for NoopStream {
114    fn write(&self, _id: &ObjectId, _size: u64, _expires_at: Option<SystemTime>) {}
115
116    fn update(&self, _id: &ObjectId, _expires_at: Option<SystemTime>) {}
117
118    fn delete(&self, _id: &ObjectId) {}
119
120    async fn join(&self, _timeout: Duration) {}
121}
122
123/// Reads a scope value off `id` as an integer, if present and well-formed.
124#[cfg(feature = "storage-cogs")]
125fn scope_id(id: &ObjectId, scope: &str) -> Option<u64> {
126    id.scopes().get_value(scope)?.parse().ok()
127}
objectstore_service/change_stream/mod.rs

objectstore_service/change_stream/
mod.rs
