1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
// SPDX-License-Identifier: BUSL-1.1
//! `SyncSession` struct + lifecycle helpers.
use std::collections::HashMap;
use std::time::Instant;
use crate::control::security::identity::AuthenticatedIdentity;
use crate::types::TenantId;
use super::super::dlq::DeviceMetadata;
use super::super::rate_limit::{RateLimitConfig, SyncRateLimiter};
/// State of a single sync session (one WebSocket connection).
pub struct SyncSession {
/// Unique session ID.
pub session_id: String,
/// Authenticated tenant.
pub tenant_id: Option<TenantId>,
/// Authenticated username.
pub username: Option<String>,
/// Full authenticated identity (set after handshake).
pub identity: Option<AuthenticatedIdentity>,
/// Whether the handshake completed successfully.
pub authenticated: bool,
/// Client's vector clock per collection.
pub client_clock: HashMap<String, HashMap<String, u64>>,
/// Server's vector clock per collection (latest LSN).
pub server_clock: HashMap<String, u64>,
/// Subscribed shape IDs.
pub subscribed_shapes: Vec<String>,
/// Mutations processed in this session.
pub mutations_processed: u64,
/// Mutations rejected in this session.
pub mutations_rejected: u64,
/// Mutations silently dropped (security rejections).
pub mutations_silent_dropped: u64,
/// Last activity timestamp.
pub last_activity: Instant,
/// Session creation time.
pub created_at: Instant,
/// Per-session rate limiter.
pub rate_limiter: SyncRateLimiter,
/// Device metadata from handshake (for DLQ entries).
pub device_metadata: DeviceMetadata,
/// Set of `(tenant_id, collection_name)` pairs the client has
/// ever sent a delta or shape subscription for — used by the
/// Origin `CollectionPurged` broadcast to decide which sessions
/// need to be notified when a collection is hard-deleted.
pub tracked_collections: std::collections::HashSet<(u64, String)>,
/// Collections whose descriptor has already been announced to the peer
/// this session via a `CollectionSchema` frame. Enforces the
/// announce-precedes-data guard: a collection's schema is emitted at most
/// once per session, strictly before its first shape snapshot or delta.
pub announced_collections: std::collections::HashSet<String>,
/// Last WAL LSN the client advertised in its vector clock at
/// handshake. Used by offline-client replay to identify
/// `CollectionPurged` events that committed while the client
/// was disconnected.
pub last_seen_lsn: u64,
/// Durable producer id assigned by `SyncProducerRegistry` at handshake.
/// `0` means the session is not a Lite client or the registry was
/// unavailable — legacy / non-Lite connections remain at 0.
pub producer_id: u64,
/// The fencing epoch accepted by `SyncProducerRegistry` at handshake.
/// `0` for non-Lite sessions.
pub accepted_epoch: u64,
}
impl SyncSession {
pub fn new(session_id: String) -> Self {
Self::with_rate_limit(session_id, &RateLimitConfig::default())
}
pub fn with_rate_limit(session_id: String, rate_config: &RateLimitConfig) -> Self {
let now = Instant::now();
Self {
session_id,
tenant_id: None,
username: None,
identity: None,
authenticated: false,
client_clock: HashMap::new(),
server_clock: HashMap::new(),
subscribed_shapes: Vec::new(),
mutations_processed: 0,
mutations_rejected: 0,
mutations_silent_dropped: 0,
last_activity: now,
created_at: now,
rate_limiter: SyncRateLimiter::new(rate_config),
device_metadata: DeviceMetadata::default(),
tracked_collections: std::collections::HashSet::new(),
announced_collections: std::collections::HashSet::new(),
last_seen_lsn: 0,
producer_id: 0,
accepted_epoch: 0,
}
}
/// Session uptime in seconds.
pub fn uptime_secs(&self) -> u64 {
self.created_at.elapsed().as_secs()
}
/// Seconds since last activity.
pub fn idle_secs(&self) -> u64 {
self.last_activity.elapsed().as_secs()
}
/// Record that the client has interacted with this
/// `(tenant, collection)` pair. Called from the delta-push and
/// shape-subscribe paths. Membership here is the subscription
/// state the Origin `CollectionPurged` broadcast filters on.
pub fn track_collection(&mut self, tenant_id: u64, collection: &str) {
self.tracked_collections
.insert((tenant_id, collection.to_string()));
}
}