feat(service): Instrument the remaining backends with the change stream

objectstore-server/src/config.rs

@@ -683,6 +683,7 @@ impl Default for Config {
 
             storage: StorageConfig::FileSystem(FileSystemConfig {
                 path: PathBuf::from("data"),
+                cogs: None,
             }),
 
             storage_cogs: None,

objectstore-service/docs/architecture.md

@@ -146,16 +146,11 @@ rate-limiting failures at a higher layer) are not counted.
 
 This is gated behind the `storage_cogs` Cargo feature.
 
-Each backend reports every write/overwrite, TTI bump, and delete it performs on
-stored objects to a [`ChangeStream`](change_stream::ChangeStream). To
-turn this change stream into COGS data, a stream consumer has to merge each
-change event into an external table to update an inventory of objects. The
-inventory table can be queried to break down each backend's storage utilization
-by `app_feature`. Note that [`NoopStream`](change_stream::NoopStream) is used
-unless the backend's config includes a
-[`CostTrackerStreamConfig`](change_stream::CostTrackerStreamConfig) and the
-service has a usable transport for it, and unless the `storage-cogs` feature is
-compiled in at all.
+Storage attribution is derived from the [change stream](#change-streams) each
+backend publishes. To turn a change stream into COGS data, a stream consumer has
+to merge each change event into an external table to update an inventory of
+objects. The inventory table can be queried to break down each backend's storage
+utilization by `app_feature`.
 
 Each row in the inventory table has an anonymized hash of an `ObjectId` as well
 as the row's size, expiry, Sentry org/project, `app_feature`, and relevant
@@ -164,11 +159,10 @@ long-term backend the inventory table will contain _two rows_ for an object: a
 row for the actual object and its size in long-term backend, and a separate row
 for the tombstone and the tombstone's size in the high-volume backend.
 
-`ChangeStream` is not aware of any automatic garbage collection that backends
-may perform. Expired objects must be filtered out when querying the inventory
-table.
+Because the change stream does not observe automatic garbage collection, expired
+objects must be filtered out when querying the inventory table.
 
-Under the hood, `CostTrackerStream` uses
+Under the hood, [`CostTrackerStream`](change_stream::CostTrackerStream) uses
 [`InventoryTracker`](objectstore_inventory_tracker::InventoryTracker) to publish
 change events; it is generic over the transport rather than tied to Kafka. Each
 backend has its own sampling rate to lessen the load put on the stream
@@ -179,6 +173,48 @@ sampling rate. When aggregating, divide each row's value by its `sample_rate`.
 
 See also: [`objectstore_inventory_tracker`] documentation.
 
+# Change Streams
+
+Every backend publishes the changes it makes to the objects it stores as a
+[`ChangeStream`](change_stream::ChangeStream). It is a fire-and-forget,
+per-backend feed of three operations:
+
+- `write(id, size, expires_at)`: `id` now occupies `size` bytes. Used for both
+  new objects and overwrites.
+- `update(id, expires_at)`: `id`'s expiration moved while its stored size is
+  unchanged. In practice this is a TTI bump.
+- `delete(id)`: `id` was deleted explicitly.
+
+The stream describes physical storage per backend. When using
+[`TieredStorage`](backend::tiered::TieredStorage), objects that are stored in
+long-term storage will emit a change record for the actual object in long-term
+storage as well as for the tombstone record in high-volume storage.
+
+`size` is a count of bytes that the backend actually stores for an object. This
+includes object payloads, metadata, and sometimes backend-specific overhead.
+
+Decorators such as [`CountingBackend`](backend::counting::CountingBackend) and
+[`TieredStorage`](backend::tiered::TieredStorage) don't publish change streams
+of their own; only leaf backends that actually own bytes do.
+
+Automatic garbage collection is invisible to the change stream. Downstream
+consumers of the stream need to consider the `expires_at` field on messages.
+
+## `ChangeStream` implementation guidance
+
+While the [`ChangeStream`](change_stream::ChangeStream) trait is abstract, that
+abstraction is not surfaced in service configuration. For instance, the
+[storage COGS change stream](#storage-cogs) is configured with a service-wide
+[`CostTrackerConfig`](change_stream::CostTrackerConfig) and per-backend
+[`CostTrackerStreamConfig`](change_stream::CostTrackerStreamConfig)s. These
+configurations are connected in [`ChangeStreamFactory`](change_stream::ChangeStreamFactory)
+to build a [`CostTrackerStream`](change_stream::CostTrackerStream).
+
+New `ChangeStream` implementations may follow the same pattern:
+- per-backend configuration for per-backend IDs or configuration
+- service-wide configuration for a stream sink
+- glue code in and around `ChangeStreamFactory`
+
 # Metadata and Payload
 
 Every object consists of structured **metadata** and a binary **payload**.

objectstore-service/src/backend/in_memory.rs

@@ -19,6 +19,7 @@ use super::common::{
     DeleteResponse, GetResponse, HighVolumeBackend, MultipartUploadBackend, PutResponse, TieredGet,
     TieredMetadata, TieredWrite, Tombstone,
 };
+use crate::change_stream::{ChangeStream, NoopStream, flush_change_stream};
 use crate::error::{Error, Result};
 use crate::id::ObjectId;
 use crate::multipart::{
@@ -34,6 +35,20 @@ enum StoreEntry {
     Tombstone(Tombstone),
 }
 
+impl StoreEntry {
+    /// Number of bytes an entry occupies: its payload plus its serialized metadata.
+    pub(crate) fn stored_size(&self) -> usize {
+        match self {
+            StoreEntry::Object(metadata, payload) => json_len(metadata) + payload.len(),
+            StoreEntry::Tombstone(tombstone) => {
+                // A tombstone carries no payload, only its redirect and expiry.
+                tombstone.target.as_storage_path().to_string().len()
+                    + json_len(&tombstone.expiration_policy)
+            }
+        }
+    }
+}
+
 type Store = HashMap<ObjectId, StoreEntry>;
 
 #[derive(Clone, Debug)]
@@ -61,6 +76,7 @@ pub struct InMemoryBackend {
     name: &'static str,
     store: Arc<Mutex<Store>>,
     multipart_store: Arc<Mutex<MultipartStore>>,
+    change_stream: Arc<dyn ChangeStream>,
 }
 
 impl InMemoryBackend {
@@ -70,9 +86,16 @@ impl InMemoryBackend {
             name,
             store: Arc::new(Mutex::new(HashMap::new())),
             multipart_store: Arc::new(Mutex::new(HashMap::new())),
+            change_stream: Arc::new(NoopStream),
         }
     }
 
+    /// Publishes this backend's changes to `change_stream`.
+    pub fn with_change_stream(mut self, change_stream: Arc<dyn ChangeStream>) -> Self {
+        self.change_stream = change_stream;
+        self
+    }
+
     /// Returns the stored entry for `id`, for direct inspection in tests.
     pub fn get(&self, id: &ObjectId) -> Entry {
         match self.store.lock().unwrap().get(id).cloned() {
@@ -117,10 +140,11 @@ impl super::common::Backend for InMemoryBackend {
         stream: ClientStream,
     ) -> Result<PutResponse> {
         let bytes: BytesMut = stream.try_collect().await?;
-        self.store.lock().unwrap().insert(
-            id.clone(),
-            StoreEntry::Object(metadata.clone(), bytes.freeze()),
-        );
+        let entry = StoreEntry::Object(metadata.clone(), bytes.freeze());
+        let size = entry.stored_size();
+        self.store.lock().unwrap().insert(id.clone(), entry);
+        self.change_stream
+            .write(id, size as u64, metadata.time_expires);
         Ok(())
     }
 
@@ -153,9 +177,15 @@ impl super::common::Backend for InMemoryBackend {
     }
 
     async fn delete_object(&self, id: &ObjectId) -> Result<DeleteResponse> {
-        self.store.lock().unwrap().remove(id);
+        if self.store.lock().unwrap().remove(id).is_some() {
+            self.change_stream.delete(id);
+        }
         Ok(())
     }
+
+    async fn join(&self) {
+        flush_change_stream(&self.change_stream).await;
+    }
 }
 
 #[async_trait::async_trait]
@@ -173,7 +203,11 @@ impl HighVolumeBackend for InMemoryBackend {
 
         let mut metadata = metadata.clone();
         metadata.size = Some(payload.len());
-        store.insert(id.clone(), StoreEntry::Object(metadata, payload));
+        let expires_at = metadata.time_expires;
+        let entry = StoreEntry::Object(metadata, payload);
+        let size = entry.stored_size();
+        store.insert(id.clone(), entry);
+        self.change_stream.write(id, size as u64, expires_at);
         Ok(None)
     }
 
@@ -220,7 +254,9 @@ impl HighVolumeBackend for InMemoryBackend {
             return Ok(Some(tombstone));
         }
 
-        store.remove(id);
+        if store.remove(id).is_some() {
+            self.change_stream.delete(id);
+        }
         Ok(None)
     }
 
@@ -239,13 +275,26 @@ impl HighVolumeBackend for InMemoryBackend {
         if matches_current {
             match write {
                 TieredWrite::Tombstone(tombstone) => {
-                    store.insert(id.clone(), StoreEntry::Tombstone(tombstone));
+                    let expires_at = tombstone
+                        .expiration_policy
+                        .expires_in()
+                        .map(|ttl| SystemTime::now() + ttl);
+                    let entry = StoreEntry::Tombstone(tombstone);
+                    let size = entry.stored_size();
+                    store.insert(id.clone(), entry);
+                    self.change_stream.write(id, size as u64, expires_at);
                 }
                 TieredWrite::Object(metadata, payload) => {
-                    store.insert(id.clone(), StoreEntry::Object(metadata, payload));
+                    let expires_at = metadata.time_expires;
+                    let entry = StoreEntry::Object(metadata, payload);
+                    let size = entry.stored_size();
+                    store.insert(id.clone(), entry);
+                    self.change_stream.write(id, size as u64, expires_at);
                 }
                 TieredWrite::Delete => {
-                    store.remove(id);
+                    if store.remove(id).is_some() {
+                        self.change_stream.delete(id);
+                    }
                 }
             }
         }
@@ -374,7 +423,7 @@ impl MultipartUploadBackend for InMemoryBackend {
         // Validate and assemble while holding the multipart lock, but don't
         // remove the upload yet — a failed validation must leave it intact so
         // the client can retry.
-        let assembled = {
+        let (metadata, payload) = {
             let store = self.multipart_store.lock().unwrap();
             let upload = store
                 .get(&key)
@@ -416,17 +465,23 @@ impl MultipartUploadBackend for InMemoryBackend {
             (metadata, payload.freeze())
         };
 
-        self.store
-            .lock()
-            .unwrap()
-            .insert(id.clone(), StoreEntry::Object(assembled.0, assembled.1));
+        let expires_at = metadata.time_expires;
+        let entry = StoreEntry::Object(metadata, payload);
+        let size = entry.stored_size();
+        self.store.lock().unwrap().insert(id.clone(), entry);
+        self.change_stream.write(id, size as u64, expires_at);
 
         self.multipart_store.lock().unwrap().remove(&key);
 
         Ok(None)
     }
 }
 
+/// Serialized length of `value`, or `0` if it cannot be serialized.
+fn json_len<T: serde::Serialize>(value: &T) -> usize {
+    serde_json::to_string(value).map_or(0, |json| json.len())
+}
+
 /// Returns `true` if `entry` matches the expected tombstone redirect state.
 ///
 /// - `expected = None`: matches any non-tombstone (absent or inline object).
@@ -498,6 +553,11 @@ mod tests {
     use objectstore_types::metadata::ExpirationPolicy;
     use objectstore_types::scope::{Scope, Scopes};
 
+    #[cfg(feature = "storage-cogs")]
+    use objectstore_inventory_tracker::OpType;
+    #[cfg(feature = "storage-cogs")]
+    use objectstore_inventory_tracker::test_utils::DummyProducer;
+
     use super::*;
     use crate::backend::common::Backend;
     use crate::id::ObjectContext;
@@ -822,4 +882,48 @@ mod tests {
             .unwrap();
         assert!(result.is_none(), "retry with correct part should succeed");
     }
+
+    #[cfg(feature = "storage-cogs")]
+    fn backend_with_change_stream() -> (InMemoryBackend, DummyProducer) {
+        use crate::change_stream::CostTrackerStreamConfig;
+
+        let (streams, producer) = crate::change_stream::dummy_factory();
+        let change_stream = streams.build(Some(&CostTrackerStreamConfig {
+            shared_resource_id: "in_memory_objectstore".into(),
+            sample_rate: 1.0,
+        }));
+        (
+            InMemoryBackend::new("test").with_change_stream(change_stream),
+            producer,
+        )
+    }
+
+    #[cfg(feature = "storage-cogs")]
+    #[tokio::test]
+    async fn change_stream_reports_writes_and_deletes() {
+        let (backend, producer) = backend_with_change_stream();
+        let id = make_id();
+        let metadata = Metadata::default();
+        let payload = b"hello";
+
+        backend
+            .put_object(&id, &metadata, stream::single(payload.to_vec()))
+            .await
+            .unwrap();
+        backend.delete_object(&id).await.unwrap();
+        // The object is already gone, so this reports nothing.
+        backend.delete_object(&id).await.unwrap();
+
+        let records = producer.records();
+        assert_eq!(records.len(), 2);
+        assert_eq!(records[0].op_type, OpType::Write);
+        assert_eq!(records[0].app_feature, "testing");
+        assert_eq!(
+            records[0].size,
+            Some((json_len(&metadata) + payload.len()) as u64),
+            "the reported size covers metadata as well as the payload"
+        );
+        assert!(json_len(&metadata) > 0, "metadata must contribute bytes");
+        assert_eq!(records[1].op_type, OpType::Delete);
+    }
 }