rivetkit_core/actor/
config.rs

1use std::collections::HashMap;
2use std::fmt;
3use std::sync::Arc;
4use std::time::Duration;
5
6use rivet_envoy_client::config::HttpRequest;
7
8use crate::inspector::InspectorTabEntry;
9
10const DEFAULT_STATE_SAVE_INTERVAL: Duration = Duration::from_secs(1);
11const DEFAULT_CREATE_VARS_TIMEOUT: Duration = Duration::from_secs(5);
12const DEFAULT_CREATE_CONN_STATE_TIMEOUT: Duration = Duration::from_secs(5);
13const DEFAULT_ON_BEFORE_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
14const DEFAULT_ON_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
15const DEFAULT_ON_MIGRATE_TIMEOUT: Duration = Duration::from_secs(30);
16const DEFAULT_ACTION_TIMEOUT: Duration = Duration::from_secs(60);
17const DEFAULT_SLEEP_TIMEOUT: Duration = Duration::from_secs(30);
18const DEFAULT_SLEEP_GRACE_PERIOD: Duration = Duration::from_secs(15);
19const DEFAULT_CONNECTION_LIVENESS_TIMEOUT: Duration = Duration::from_millis(2500);
20const DEFAULT_CONNECTION_LIVENESS_INTERVAL: Duration = Duration::from_secs(5);
21const DEFAULT_MAX_QUEUE_SIZE: u32 = 1000;
22pub const DEFAULT_MAX_SCHEDULES: u32 = 1_000;
23const DEFAULT_MAX_QUEUE_MESSAGE_SIZE: u32 = 65_536;
24const DEFAULT_MAX_INCOMING_MESSAGE_SIZE: u32 = 65_536;
25const DEFAULT_MAX_OUTGOING_MESSAGE_SIZE: u32 = 1_048_576;
26pub(crate) const MAX_SQLITE_TRANSACTION_TRACE_STATEMENTS: usize = 32;
27
28#[derive(Clone)]
29pub enum CanHibernateWebSocket {
30	Bool(bool),
31	Callback(Arc<dyn Fn(&HttpRequest) -> bool + Send + Sync>),
32}
33
34impl fmt::Debug for CanHibernateWebSocket {
35	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
36		match self {
37			Self::Bool(value) => f.debug_tuple("Bool").field(value).finish(),
38			Self::Callback(_) => f.write_str("Callback(..)"),
39		}
40	}
41}
42
43impl Default for CanHibernateWebSocket {
44	fn default() -> Self {
45		Self::Bool(false)
46	}
47}
48
49#[derive(Clone, Debug, Default)]
50pub struct ActorConfigOverrides {
51	pub sleep_grace_period: Option<Duration>,
52}
53
54#[derive(Clone, Debug)]
55pub struct ActionDefinition {
56	pub name: String,
57}
58
59/// Experimental SQLite profiling configuration.
60///
61/// This entire configuration surface, including every field, is subject to
62/// change without notice.
63#[derive(Clone, Debug)]
64pub struct SqliteProfilingConfig {
65	pub enabled: bool,
66	pub max_tracked_statement_fingerprints: usize,
67	pub max_tracked_transaction_fingerprints: usize,
68	pub max_prometheus_series: usize,
69	pub max_statements_per_transaction_trace: usize,
70	pub max_get_pages_requests_per_trace: usize,
71	pub max_transaction_name_bytes: usize,
72	pub slow_operation_threshold_ms: u64,
73	pub baseline_sample_rate: f64,
74	pub max_diagnostic_events_per_minute: usize,
75	pub diagnostic_event_queue_capacity: usize,
76}
77
78impl Default for SqliteProfilingConfig {
79	fn default() -> Self {
80		Self {
81			enabled: true,
82			max_tracked_statement_fingerprints: 128,
83			max_tracked_transaction_fingerprints: 8,
84			max_prometheus_series: 25_000,
85			max_statements_per_transaction_trace: 32,
86			max_get_pages_requests_per_trace: 16,
87			max_transaction_name_bytes: 128,
88			slow_operation_threshold_ms: 10,
89			baseline_sample_rate: 0.001,
90			max_diagnostic_events_per_minute: 120,
91			diagnostic_event_queue_capacity: 256,
92		}
93	}
94}
95
96/// Sparse experimental SQLite profiling configuration used at runtime
97/// boundaries.
98///
99/// This entire configuration surface, including every field, is subject to
100/// change without notice.
101#[derive(Clone, Debug, Default)]
102pub struct SqliteProfilingConfigInput {
103	pub enabled: Option<bool>,
104	pub max_tracked_statement_fingerprints: Option<u32>,
105	pub max_tracked_transaction_fingerprints: Option<u32>,
106	pub max_prometheus_series: Option<u32>,
107	pub max_statements_per_transaction_trace: Option<u32>,
108	pub max_get_pages_requests_per_trace: Option<u32>,
109	pub max_transaction_name_bytes: Option<u32>,
110	pub slow_operation_threshold_ms: Option<u32>,
111	pub baseline_sample_rate: Option<f64>,
112	pub max_diagnostic_events_per_minute: Option<u32>,
113	pub diagnostic_event_queue_capacity: Option<u32>,
114}
115
116impl SqliteProfilingConfig {
117	fn from_input(input: SqliteProfilingConfigInput) -> Self {
118		let mut config = Self::default();
119		macro_rules! set_usize {
120			($field:ident) => {
121				if let Some(value) = input.$field {
122					config.$field = value as usize;
123				}
124			};
125		}
126		if let Some(value) = input.enabled {
127			config.enabled = value;
128		}
129		set_usize!(max_tracked_statement_fingerprints);
130		set_usize!(max_tracked_transaction_fingerprints);
131		set_usize!(max_prometheus_series);
132		set_usize!(max_statements_per_transaction_trace);
133		set_usize!(max_get_pages_requests_per_trace);
134		set_usize!(max_transaction_name_bytes);
135		if let Some(value) = input.slow_operation_threshold_ms {
136			config.slow_operation_threshold_ms = u64::from(value);
137		}
138		if let Some(value) = input.baseline_sample_rate {
139			config.baseline_sample_rate = value;
140		}
141		set_usize!(max_diagnostic_events_per_minute);
142		set_usize!(diagnostic_event_queue_capacity);
143		config
144	}
145}
146
147/// Per-actor trace sample rates, from 0.0 to 1.0. A rate lowers what is recorded
148/// and never raises it.
149#[derive(Clone, Debug, Default)]
150pub struct ActorTracingConfig {
151	/// `None` defers to the process sampler from `OTEL_TRACES_SAMPLER`.
152	pub sampler: Option<f64>,
153	/// Keyed by flattened action name. Also covers scheduled runs of the action.
154	pub actions: HashMap<String, f64>,
155}
156
157impl ActorTracingConfig {
158	pub(crate) fn sampler_for_action(&self, action_name: &str) -> Option<f64> {
159		self.actions.get(action_name).copied().or(self.sampler)
160	}
161
162	fn validate(&self, actions: &[ActionDefinition]) -> anyhow::Result<()> {
163		for rate in self.sampler.iter().chain(self.actions.values()) {
164			anyhow::ensure!(
165				rate.is_finite() && (0.0..=1.0).contains(rate),
166				"tracing sampler must be between 0 and 1, got {rate}"
167			);
168		}
169		// The Rust SDK never fills `ActorConfig::actions`.
170		if actions.is_empty() {
171			return Ok(());
172		}
173		for action_name in self.actions.keys() {
174			anyhow::ensure!(
175				actions.iter().any(|action| &action.name == action_name),
176				"tracing sampler names unknown action `{action_name}`"
177			);
178		}
179		Ok(())
180	}
181}
182
183#[derive(Clone, Debug)]
184pub struct ActorConfig {
185	pub name: Option<String>,
186	pub icon: Option<String>,
187	/// Whether the user declared a SQLite database for this actor (`db({...})`
188	/// on the TS side). Gates the inspector database tab.
189	pub has_database: bool,
190	pub remote_sqlite: bool,
191	pub sqlite_profiling: SqliteProfilingConfig,
192	pub tracing: Arc<ActorTracingConfig>,
193	/// Enables the experimental Actor Runtime Socket.
194	pub enable_actor_runtime_socket: bool,
195	/// Whether the user declared actor state (`state: ...` or `createState`).
196	/// Gates the inspector state tab and state-subscription messages.
197	pub has_state: bool,
198	pub can_hibernate_websocket: CanHibernateWebSocket,
199	pub state_save_interval: Duration,
200	pub create_vars_timeout: Duration,
201	pub create_conn_state_timeout: Duration,
202	pub on_before_connect_timeout: Duration,
203	pub on_connect_timeout: Duration,
204	pub on_migrate_timeout: Duration,
205	pub action_timeout: Duration,
206	pub sleep_timeout: Duration,
207	pub no_sleep: bool,
208	pub sleep_grace_period: Duration,
209	pub sleep_grace_period_overridden: bool,
210	pub connection_liveness_timeout: Duration,
211	pub connection_liveness_interval: Duration,
212	pub max_queue_size: u32,
213	pub max_schedules: u32,
214	pub max_queue_message_size: u32,
215	pub max_incoming_message_size: u32,
216	pub max_outgoing_message_size: u32,
217	pub overrides: Option<ActorConfigOverrides>,
218	pub actions: Vec<ActionDefinition>,
219	/// Author-declared inspector tab entries (custom tabs + built-in
220	/// hides). Validated upstream (Zod / builder).
221	pub inspector_tabs: Vec<InspectorTabEntry>,
222}
223
224/// Sparse, serialization-friendly actor configuration. All fields are optional with millisecond integers instead of Duration. Used at runtime boundaries (NAPI, config files). Convert to ActorConfig via ActorConfig::from_input().
225#[derive(Clone, Debug, Default)]
226pub struct ActorConfigInput {
227	pub name: Option<String>,
228	pub icon: Option<String>,
229	pub has_database: Option<bool>,
230	pub remote_sqlite: Option<bool>,
231	pub sqlite_profiling: Option<SqliteProfilingConfigInput>,
232	pub tracing: Option<ActorTracingConfig>,
233	pub enable_actor_runtime_socket: Option<bool>,
234	pub has_state: Option<bool>,
235	pub can_hibernate_websocket: Option<bool>,
236	pub state_save_interval_ms: Option<u32>,
237	pub create_vars_timeout_ms: Option<u32>,
238	pub create_conn_state_timeout_ms: Option<u32>,
239	pub on_before_connect_timeout_ms: Option<u32>,
240	pub on_connect_timeout_ms: Option<u32>,
241	pub on_migrate_timeout_ms: Option<u32>,
242	pub action_timeout_ms: Option<u32>,
243	pub sleep_timeout_ms: Option<u32>,
244	pub no_sleep: Option<bool>,
245	pub sleep_grace_period_ms: Option<u32>,
246	pub connection_liveness_timeout_ms: Option<u32>,
247	pub connection_liveness_interval_ms: Option<u32>,
248	pub max_queue_size: Option<u32>,
249	pub max_schedules: Option<u32>,
250	pub max_queue_message_size: Option<u32>,
251	pub max_incoming_message_size: Option<u32>,
252	pub max_outgoing_message_size: Option<u32>,
253	pub actions: Option<Vec<ActionDefinition>>,
254	pub inspector_tabs: Option<Vec<InspectorTabEntry>>,
255}
256
257impl ActorConfig {
258	pub fn from_input(config: ActorConfigInput) -> Self {
259		let mut actor_config = Self {
260			name: config.name,
261			icon: config.icon,
262			has_database: config.has_database.unwrap_or(false),
263			remote_sqlite: config.remote_sqlite.unwrap_or(false),
264			sqlite_profiling: config
265				.sqlite_profiling
266				.map(SqliteProfilingConfig::from_input)
267				.unwrap_or_default(),
268			tracing: Arc::new(config.tracing.unwrap_or_default()),
269			enable_actor_runtime_socket: config.enable_actor_runtime_socket.unwrap_or(false),
270			has_state: config.has_state.unwrap_or(false),
271			..Self::default()
272		};
273		if let Some(can_hibernate_websocket) = config.can_hibernate_websocket {
274			actor_config.can_hibernate_websocket =
275				CanHibernateWebSocket::Bool(can_hibernate_websocket);
276		}
277		if let Some(value) = config.state_save_interval_ms {
278			actor_config.state_save_interval = duration_ms(value);
279		}
280		if let Some(value) = config.create_vars_timeout_ms {
281			actor_config.create_vars_timeout = duration_ms(value);
282		}
283		if let Some(value) = config.create_conn_state_timeout_ms {
284			actor_config.create_conn_state_timeout = duration_ms(value);
285		}
286		if let Some(value) = config.on_before_connect_timeout_ms {
287			actor_config.on_before_connect_timeout = duration_ms(value);
288		}
289		if let Some(value) = config.on_connect_timeout_ms {
290			actor_config.on_connect_timeout = duration_ms(value);
291		}
292		if let Some(value) = config.on_migrate_timeout_ms {
293			actor_config.on_migrate_timeout = duration_ms(value);
294		}
295		if let Some(value) = config.action_timeout_ms {
296			actor_config.action_timeout = duration_ms(value);
297		}
298		if let Some(value) = config.sleep_timeout_ms {
299			actor_config.sleep_timeout = duration_ms(value);
300		}
301		if let Some(value) = config.no_sleep {
302			actor_config.no_sleep = value;
303		}
304		if let Some(value) = config.sleep_grace_period_ms {
305			actor_config.sleep_grace_period = duration_ms(value);
306			actor_config.sleep_grace_period_overridden = true;
307		}
308		if let Some(value) = config.connection_liveness_timeout_ms {
309			actor_config.connection_liveness_timeout = duration_ms(value);
310		}
311		if let Some(value) = config.connection_liveness_interval_ms {
312			actor_config.connection_liveness_interval = duration_ms(value);
313		}
314		if let Some(value) = config.max_queue_size {
315			actor_config.max_queue_size = value;
316		}
317		if let Some(value) = config.max_schedules {
318			actor_config.max_schedules = value;
319		}
320		if let Some(value) = config.max_queue_message_size {
321			actor_config.max_queue_message_size = value;
322		}
323		if let Some(value) = config.max_incoming_message_size {
324			actor_config.max_incoming_message_size = value;
325		}
326		if let Some(value) = config.max_outgoing_message_size {
327			actor_config.max_outgoing_message_size = value;
328		}
329		if let Some(actions) = config.actions {
330			actor_config.actions = actions;
331		}
332		if let Some(tabs) = config.inspector_tabs {
333			actor_config.inspector_tabs = tabs;
334		}
335
336		actor_config
337	}
338
339	pub fn effective_sleep_grace_period(&self) -> Duration {
340		cap_duration(
341			self.sleep_grace_period,
342			self.overrides
343				.as_ref()
344				.and_then(|overrides| overrides.sleep_grace_period),
345		)
346	}
347
348	/// Runtime authority for rejecting malformed config that bypassed the
349	/// TypeScript Zod layer (direct Rust builders, NAPI shim corruption,
350	/// etc.). Call this before constructing an `ActorFactory` from this
351	/// config so the actor never starts with garbage state.
352	pub fn validate(&self) -> anyhow::Result<()> {
353		crate::inspector::validate_inspector_tabs(&self.inspector_tabs)?;
354		self.tracing.validate(&self.actions)?;
355		anyhow::ensure!(
356			self.sqlite_profiling.baseline_sample_rate.is_finite()
357				&& (0.0..=1.0).contains(&self.sqlite_profiling.baseline_sample_rate),
358			"SQLite profiling baselineSampleRate must be between 0 and 1"
359		);
360		anyhow::ensure!(
361			self.sqlite_profiling.max_get_pages_requests_per_trace <= 16,
362			"SQLite profiling maxGetPagesRequestsPerTrace must be at most 16"
363		);
364		anyhow::ensure!(
365			self.sqlite_profiling.max_statements_per_transaction_trace
366				<= MAX_SQLITE_TRANSACTION_TRACE_STATEMENTS,
367			"SQLite profiling maxStatementsPerTransactionTrace must be at most 32"
368		);
369		anyhow::ensure!(
370			self.sqlite_profiling.max_transaction_name_bytes > 0,
371			"SQLite profiling maxTransactionNameBytes must be greater than zero"
372		);
373		Ok(())
374	}
375}
376
377impl Default for ActorConfig {
378	fn default() -> Self {
379		Self {
380			name: None,
381			icon: None,
382			has_database: false,
383			remote_sqlite: false,
384			sqlite_profiling: SqliteProfilingConfig::default(),
385			tracing: Arc::default(),
386			enable_actor_runtime_socket: false,
387			has_state: false,
388			can_hibernate_websocket: CanHibernateWebSocket::default(),
389			state_save_interval: DEFAULT_STATE_SAVE_INTERVAL,
390			create_vars_timeout: DEFAULT_CREATE_VARS_TIMEOUT,
391			create_conn_state_timeout: DEFAULT_CREATE_CONN_STATE_TIMEOUT,
392			on_before_connect_timeout: DEFAULT_ON_BEFORE_CONNECT_TIMEOUT,
393			on_connect_timeout: DEFAULT_ON_CONNECT_TIMEOUT,
394			on_migrate_timeout: DEFAULT_ON_MIGRATE_TIMEOUT,
395			action_timeout: DEFAULT_ACTION_TIMEOUT,
396			sleep_timeout: DEFAULT_SLEEP_TIMEOUT,
397			no_sleep: false,
398			sleep_grace_period: DEFAULT_SLEEP_GRACE_PERIOD,
399			sleep_grace_period_overridden: false,
400			connection_liveness_timeout: DEFAULT_CONNECTION_LIVENESS_TIMEOUT,
401			connection_liveness_interval: DEFAULT_CONNECTION_LIVENESS_INTERVAL,
402			max_queue_size: DEFAULT_MAX_QUEUE_SIZE,
403			max_schedules: DEFAULT_MAX_SCHEDULES,
404			max_queue_message_size: DEFAULT_MAX_QUEUE_MESSAGE_SIZE,
405			max_incoming_message_size: DEFAULT_MAX_INCOMING_MESSAGE_SIZE,
406			max_outgoing_message_size: DEFAULT_MAX_OUTGOING_MESSAGE_SIZE,
407			overrides: None,
408			actions: Vec::new(),
409			inspector_tabs: Vec::new(),
410		}
411	}
412}
413
414fn cap_duration(duration: Duration, override_duration: Option<Duration>) -> Duration {
415	if let Some(override_duration) = override_duration {
416		duration.min(override_duration)
417	} else {
418		duration
419	}
420}
421
422fn duration_ms(value: u32) -> Duration {
423	Duration::from_millis(u64::from(value))
424}
425
426// Test shim keeps moved tests in crate-root tests/ with private-module access.
427#[cfg(test)]
428#[path = "../../tests/config.rs"]
429mod tests;
rivetkit_core/actor/config.rs

rivetkit_core/actor/
config.rs
