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

objectstore_service/change_stream/
mod.rs
