1+ use std:: collections:: hash_map:: RandomState ;
2+ use std:: collections:: BTreeMap ;
13use std:: fmt;
4+ use std:: sync:: { Arc , Mutex } ;
5+ use std:: time:: { Duration , SystemTime , UNIX_EPOCH } ;
26
3- use opentelemetry:: metrics:: { Counter , MeterProvider , UpDownCounter } ;
7+ use hyperloglogplus:: { HyperLogLog , HyperLogLogPlus } ;
8+ use opentelemetry:: metrics:: { Counter , MeterProvider , ObservableGauge , UpDownCounter } ;
49use opentelemetry:: KeyValue ;
510use opentelemetry_sdk:: metrics:: SdkMeterProvider ;
11+ use payjoin:: directory:: ShortId ;
612
713pub ( crate ) const TOTAL_CONNECTIONS : & str = "total_connections" ;
814pub ( crate ) const ACTIVE_CONNECTIONS : & str = "active_connections" ;
915pub ( crate ) const HTTP_REQUESTS : & str = "http_request_total" ;
1016pub ( crate ) const DB_ENTRIES : & str = "db_entries_total" ;
17+ pub ( crate ) const UNIQUE_SHORT_IDS : & str = "unique_short_ids" ;
18+
19+ const HLL_PRECISION : u8 = 14 ;
20+ const HOURLY_RETENTION_HOURS : u64 = 168 ; // 7 days
21+ const DAILY_RETENTION_DAYS : u64 = 90 ;
22+
23+ type HllSketch = HyperLogLogPlus < [ u8 ; 8 ] , RandomState > ;
24+
25+ const HOUR : Duration = Duration :: from_secs ( 3600 ) ;
26+ const DAY : Duration = Duration :: from_secs ( 86400 ) ;
27+
28+ /// Convenience helpers for SystemTime to get integer intervals
29+ /// since the UNIX epoch.
30+ trait SystemTimeExt {
31+ fn intervals_since_epoch ( & self , interval : Duration ) -> u64 ;
32+
33+ fn hours_since_epoch ( & self ) -> u64 { self . intervals_since_epoch ( HOUR ) }
34+
35+ fn days_since_epoch ( & self ) -> u64 { self . intervals_since_epoch ( DAY ) }
36+ }
37+
38+ impl SystemTimeExt for SystemTime {
39+ fn intervals_since_epoch ( & self , interval : Duration ) -> u64 {
40+ self . duration_since ( UNIX_EPOCH ) . expect ( "system clock before UNIX epoch" ) . as_secs ( )
41+ / interval. as_secs ( )
42+ }
43+ }
44+
45+ fn new_sketch ( ) -> HllSketch {
46+ HyperLogLogPlus :: new ( HLL_PRECISION , RandomState :: new ( ) ) . expect ( "precision 14 is always valid" )
47+ }
48+
49+ /// Estimates the number of unique ShortIds seen per time window.
50+ /// Two tiers of HLL sketches:
51+ /// - **Hourly** -- one sketch per hour, pruned after 7 days.
52+ /// - **Daily** -- one sketch per day, pruned after 90 days.
53+ struct HllSketches {
54+ hourly : BTreeMap < u64 , HllSketch > ,
55+ daily : BTreeMap < u64 , HllSketch > ,
56+ }
57+
58+ impl HllSketches {
59+ fn new ( ) -> Self { Self { hourly : BTreeMap :: new ( ) , daily : BTreeMap :: new ( ) } }
60+
61+ // NOTE: ShortIds are passed directly to the HyperLogLog sketches held in mailroom's
62+ // RAM; only the resulting cardinality estimates are published to Prometheus. If
63+ // Prometheus ever gains native cardinality-estimation support and raw IDs are
64+ // exposed to it, they should be hashed (e.g. with a keyed PRF) before being
65+ // forwarded to avoid leaking session identifiers through the metrics pipeline.
66+ fn add_id ( & mut self , id : & ShortId ) {
67+ let now = SystemTime :: now ( ) ;
68+ let hour = now. hours_since_epoch ( ) ;
69+ let day = now. days_since_epoch ( ) ;
70+
71+ self . hourly . entry ( hour) . or_insert_with ( new_sketch) . insert ( & id. 0 ) ;
72+ self . daily . entry ( day) . or_insert_with ( new_sketch) . insert ( & id. 0 ) ;
73+
74+ let hourly_cutoff = hour. saturating_sub ( HOURLY_RETENTION_HOURS ) ;
75+ while let Some ( ( & k, _) ) = self . hourly . first_key_value ( ) {
76+ if k < hourly_cutoff {
77+ self . hourly . pop_first ( ) ;
78+ } else {
79+ break ;
80+ }
81+ }
82+ let daily_cutoff = day. saturating_sub ( DAILY_RETENTION_DAYS ) ;
83+ while let Some ( ( & k, _) ) = self . daily . first_key_value ( ) {
84+ if k < daily_cutoff {
85+ self . daily . pop_first ( ) ;
86+ } else {
87+ break ;
88+ }
89+ }
90+ }
91+
92+ fn hourly_count ( & mut self ) -> u64 {
93+ let hour = SystemTime :: now ( ) . hours_since_epoch ( ) ;
94+ self . hourly . get_mut ( & hour) . map ( |hll| hll. count ( ) . trunc ( ) as u64 ) . unwrap_or ( 0 )
95+ }
96+
97+ fn daily_count ( & mut self ) -> u64 {
98+ let day = SystemTime :: now ( ) . days_since_epoch ( ) ;
99+ self . daily . get_mut ( & day) . map ( |hll| hll. count ( ) . trunc ( ) as u64 ) . unwrap_or ( 0 )
100+ }
101+
102+ fn weekly_count ( & mut self ) -> u64 {
103+ let today = SystemTime :: now ( ) . days_since_epoch ( ) ;
104+ self . daily_union_count ( today. saturating_sub ( 6 ) ..=today)
105+ }
106+
107+ fn monthly_count ( & mut self ) -> u64 {
108+ let today = SystemTime :: now ( ) . days_since_epoch ( ) ;
109+ self . daily_union_count ( today. saturating_sub ( 30 ) ..=today)
110+ }
111+
112+ fn daily_union_count < R : std:: ops:: RangeBounds < u64 > > ( & self , range : R ) -> u64 {
113+ let mut union = new_sketch ( ) ;
114+ for sketch in self . daily . range ( range) . map ( |( _, v) | v) {
115+ union. merge ( sketch) . expect ( "same precision" ) ;
116+ }
117+ union. count ( ) . trunc ( ) as u64
118+ }
119+ }
120+
121+ #[ derive( Clone ) ]
122+ pub struct UniqueShortIdTracker {
123+ inner : Arc < Mutex < HllSketches > > ,
124+ }
125+
126+ impl UniqueShortIdTracker {
127+ pub fn new ( ) -> Self { Self { inner : Arc :: new ( Mutex :: new ( HllSketches :: new ( ) ) ) } }
128+
129+ pub fn add_id ( & self , id : & ShortId ) {
130+ self . inner . lock ( ) . expect ( "tracker lock poisoned" ) . add_id ( id) ;
131+ }
132+
133+ pub fn hourly_count ( & self ) -> u64 {
134+ self . inner . lock ( ) . expect ( "tracker lock poisoned" ) . hourly_count ( )
135+ }
136+
137+ pub fn daily_count ( & self ) -> u64 {
138+ self . inner . lock ( ) . expect ( "tracker lock poisoned" ) . daily_count ( )
139+ }
140+
141+ pub fn weekly_count ( & self ) -> u64 {
142+ self . inner . lock ( ) . expect ( "tracker lock poisoned" ) . weekly_count ( )
143+ }
144+
145+ pub fn monthly_count ( & self ) -> u64 {
146+ self . inner . lock ( ) . expect ( "tracker lock poisoned" ) . monthly_count ( )
147+ }
148+ }
149+
150+ impl Default for UniqueShortIdTracker {
151+ fn default ( ) -> Self { Self :: new ( ) }
152+ }
11153
12154#[ derive( Clone ) ]
13155pub struct MetricsService {
@@ -19,6 +161,8 @@ pub struct MetricsService {
19161active_connections : UpDownCounter < i64 > ,
20162 /// Total v1/v2 mailbox entries written, labelled by `version`
21163db_entries_total : Counter < u64 > ,
164+ tracker : UniqueShortIdTracker ,
165+ _unique_ids_gauge : Option < Arc < ObservableGauge < u64 > > > ,
22166}
23167
24168#[ repr( u8 ) ]
@@ -36,6 +180,7 @@ impl fmt::Display for PayjoinVersion {
36180
37181impl MetricsService {
38182 pub fn new ( provider : Option < SdkMeterProvider > ) -> Self {
183+ let has_reader = provider. is_some ( ) ;
39184 let provider = provider. unwrap_or_default ( ) ;
40185 let meter = provider. meter ( "payjoin-mailroom" ) ;
41186
@@ -59,7 +204,46 @@ impl MetricsService {
59204 . with_description ( "Total mailbox entries stored by protocol version" )
60205 . build ( ) ;
61206
62- Self { http_requests_total, total_connections, active_connections, db_entries_total }
207+ let tracker = UniqueShortIdTracker :: new ( ) ;
208+
209+ let unique_ids_gauge = if has_reader {
210+ let gauge_tracker = tracker. clone ( ) ;
211+ Some ( Arc :: new (
212+ meter
213+ . u64_observable_gauge ( UNIQUE_SHORT_IDS )
214+ . with_description ( "Estimated unique short IDs" )
215+ . with_callback ( move |observer| {
216+ observer. observe (
217+ gauge_tracker. hourly_count ( ) ,
218+ & [ KeyValue :: new ( "interval" , "hourly" ) ] ,
219+ ) ;
220+ observer. observe (
221+ gauge_tracker. daily_count ( ) ,
222+ & [ KeyValue :: new ( "interval" , "daily" ) ] ,
223+ ) ;
224+ observer. observe (
225+ gauge_tracker. weekly_count ( ) ,
226+ & [ KeyValue :: new ( "interval" , "weekly" ) ] ,
227+ ) ;
228+ observer. observe (
229+ gauge_tracker. monthly_count ( ) ,
230+ & [ KeyValue :: new ( "interval" , "monthly" ) ] ,
231+ ) ;
232+ } )
233+ . build ( ) ,
234+ ) )
235+ } else {
236+ None
237+ } ;
238+
239+ Self {
240+ http_requests_total,
241+ total_connections,
242+ active_connections,
243+ db_entries_total,
244+ tracker,
245+ _unique_ids_gauge : unique_ids_gauge,
246+ }
63247 }
64248
65249 pub fn record_http_request ( & self , endpoint : & str , method : & str , status_code : u16 ) {
@@ -83,4 +267,6 @@ impl MetricsService {
83267 pub fn record_db_entry ( & self , version : PayjoinVersion ) {
84268 self . db_entries_total . add ( 1 , & [ KeyValue :: new ( "version" , version. to_string ( ) ) ] ) ;
85269 }
270+
271+ pub fn record_short_id ( & self , id : & ShortId ) { self . tracker . add_id ( id) ; }
86272}
0 commit comments