objectstore_inventory_tracker/
record.rs

1//! The wire format emitted onto the inventory topic.
2//!
3//! These types mirror the `shared-resources-inventory` schema registered in
4//! [sentry-kafka-schemas]. The schema sets `additionalProperties: false`, so adding a
5//! field here without a corresponding schema version bump produces messages that
6//! consumers reject.
7//!
8//! [sentry-kafka-schemas]: https://github.com/getsentry/sentry-kafka-schemas
9
10use std::time::{SystemTime, UNIX_EPOCH};
11
12use serde::{Deserialize, Serialize};
13
14/// The kind of change a record describes. `WRITE`, `UPDATE`, or `DELETE`.
15#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
16#[serde(rename_all = "UPPERCASE")]
17pub enum OpType {
18    /// The record was created, or replaced with new contents.
19    Write,
20    /// An existing record's metadata changed without its stored size changing.
21    ///
22    /// Extending a record's expiration deadline is an example of an update operation.
23    Update,
24    /// The record was removed.
25    Delete,
26}
27
28/// A single inventory change event.
29///
30/// Construct these through [`InventoryTracker`](crate::InventoryTracker) rather than
31/// directly.
32#[derive(Clone, Debug, Deserialize, Serialize)]
33pub struct InventoryRecord {
34    /// Identifies the shared resource this record belongs to.
35    ///
36    /// This is meant to match a label on a provisioned storage backend so that downstream
37    /// consumers can join the change stream dataset with, for instance, billing info.
38    pub shared_resource_id: String,
39
40    /// The product feature this record is attributed to.
41    pub app_feature: String,
42
43    /// The type of operation that occurred.
44    pub op_type: OpType,
45
46    /// Opaque stable identifier, unique within `shared_resource_id`.
47    ///
48    /// A hash of the caller's storage key, derived by
49    /// [`InventoryTracker`](crate::InventoryTracker). The raw key is never emitted.
50    pub record_id: String,
51
52    /// When the operation occurred, as Unix epoch microseconds.
53    pub timestamp: i64,
54
55    /// Fraction of records the producer is emitting for this resource, in `[0, 1]`.
56    ///
57    /// `InventoryTracker` will not emit messages for records that are sampled out so the
58    /// consumer doesn't need to do any filtering. The reason the sample rate is included
59    /// on messages is so that the consumer or downstream pipelines can apply a
60    /// `1 / sample_rate` weight when calculating aggregates to account for changes to the
61    /// sample rate.
62    pub sample_rate: f64,
63
64    /// Stored size in bytes. Always set for [`OpType::Write`].
65    #[serde(skip_serializing_if = "Option::is_none")]
66    pub size: Option<u64>,
67
68    /// When this record is set to expire, in Unix epoch microseconds if known.
69    #[serde(skip_serializing_if = "Option::is_none")]
70    pub expiration_time: Option<i64>,
71
72    /// ID of the organization that owns the record, if known.
73    #[serde(skip_serializing_if = "Option::is_none")]
74    pub organization_id: Option<u64>,
75
76    /// ID of the project that owns the record, if known.
77    #[serde(skip_serializing_if = "Option::is_none")]
78    pub project_id: Option<u64>,
79}
80
81/// Converts a [`SystemTime`] to Unix epoch microseconds.
82pub fn epoch_micros(time: SystemTime) -> i64 {
83    match time.duration_since(UNIX_EPOCH) {
84        Ok(duration) => i64::try_from(duration.as_micros()).unwrap_or(i64::MAX),
85        Err(err) => i64::try_from(err.duration().as_micros()).map_or(i64::MIN, |micros| -micros),
86    }
87}
88
89#[cfg(test)]
90mod tests {
91    use std::time::Duration;
92
93    use serde_json::json;
94
95    use super::*;
96
97    fn record(op_type: OpType) -> InventoryRecord {
98        InventoryRecord {
99            shared_resource_id: "example_resource".into(),
100            app_feature: "example_feature".into(),
101            op_type,
102            record_id: "3f7a1c2e9b4d5a6f8c0e1d2b3a4f5e6c".into(),
103            timestamp: 1_785_283_200_000_000,
104            sample_rate: 1.0,
105            size: None,
106            expiration_time: None,
107            organization_id: None,
108            project_id: None,
109        }
110    }
111
112    #[test]
113    fn write_serializes_to_expected_wire_format() {
114        let mut rec = record(OpType::Write);
115        rec.size = Some(524_288);
116        rec.expiration_time = Some(1_785_888_000_000_000);
117        rec.organization_id = Some(1);
118        rec.project_id = Some(1);
119
120        assert_eq!(
121            serde_json::to_value(&rec).unwrap(),
122            json!({
123                "shared_resource_id": "example_resource",
124                "app_feature": "example_feature",
125                "op_type": "WRITE",
126                "record_id": "3f7a1c2e9b4d5a6f8c0e1d2b3a4f5e6c",
127                "timestamp": 1_785_283_200_000_000_i64,
128                "sample_rate": 1.0,
129                "size": 524_288,
130                "expiration_time": 1_785_888_000_000_000_i64,
131                "organization_id": 1,
132                "project_id": 1,
133            })
134        );
135    }
136
137    #[test]
138    fn absent_optional_fields_are_omitted() {
139        let value = serde_json::to_value(record(OpType::Delete)).unwrap();
140        let object = value.as_object().unwrap();
141
142        assert_eq!(object["op_type"], "DELETE");
143        for absent in ["size", "expiration_time", "organization_id", "project_id"] {
144            assert!(!object.contains_key(absent), "{absent} should be omitted");
145        }
146        // The required fields survive that omission.
147        for present in [
148            "shared_resource_id",
149            "app_feature",
150            "op_type",
151            "record_id",
152            "timestamp",
153            "sample_rate",
154        ] {
155            assert!(object.contains_key(present), "{present} is required");
156        }
157    }
158
159    #[test]
160    fn op_types_use_uppercase_spellings() {
161        for (op_type, expected) in [
162            (OpType::Write, "WRITE"),
163            (OpType::Update, "UPDATE"),
164            (OpType::Delete, "DELETE"),
165        ] {
166            assert_eq!(serde_json::to_value(op_type).unwrap(), json!(expected));
167        }
168    }
169
170    #[test]
171    fn epoch_micros_converts_both_directions() {
172        assert_eq!(epoch_micros(UNIX_EPOCH), 0);
173        assert_eq!(
174            epoch_micros(UNIX_EPOCH + Duration::from_micros(1_785_283_200_000_000)),
175            1_785_283_200_000_000
176        );
177        assert_eq!(
178            epoch_micros(UNIX_EPOCH - Duration::from_micros(1_500)),
179            -1_500
180        );
181    }
182
183    #[test]
184    fn epoch_micros_preserves_sub_second_precision() {
185        let time = UNIX_EPOCH + Duration::new(1_785_283_200, 123_456_000);
186        assert_eq!(epoch_micros(time), 1_785_283_200_123_456);
187    }
188}
objectstore_inventory_tracker/record.rs

objectstore_inventory_tracker/
record.rs
