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 queries: Vec<String>,
66	pub fingerprints: Vec<String>,
67	pub max_tracked_statement_fingerprints: usize,
68	pub max_tracked_transaction_fingerprints: usize,
69	pub max_prometheus_series: usize,
70	pub max_statements_per_transaction_trace: usize,
71	pub max_get_pages_requests_per_trace: usize,
72	pub max_transaction_name_bytes: usize,
73	pub slow_operation_threshold_ms: u64,
74	pub baseline_sample_rate: f64,
75	pub max_diagnostic_events_per_minute: usize,
76	pub diagnostic_event_queue_capacity: usize,
77}
78
79impl Default for SqliteProfilingConfig {
80	fn default() -> Self {
81		Self {
82			queries: Vec::new(),
83			fingerprints: Vec::new(),
84			max_tracked_statement_fingerprints: 128,
85			max_tracked_transaction_fingerprints: 8,
86			max_prometheus_series: 25_000,
87			max_statements_per_transaction_trace: 32,
88			max_get_pages_requests_per_trace: 16,
89			max_transaction_name_bytes: 128,
90			slow_operation_threshold_ms: 10,
91			baseline_sample_rate: 0.001,
92			max_diagnostic_events_per_minute: 120,
93			diagnostic_event_queue_capacity: 256,
94		}
95	}
96}
97
98/// Sparse experimental SQLite profiling configuration used at runtime
99/// boundaries.
100///
101/// This entire configuration surface, including every field, is subject to
102/// change without notice.
103#[derive(Clone, Debug, Default)]
104pub struct SqliteProfilingConfigInput {
105	pub queries: Option<Vec<String>>,
106	pub fingerprints: Option<Vec<String>>,
107	pub max_tracked_statement_fingerprints: Option<u32>,
108	pub max_tracked_transaction_fingerprints: Option<u32>,
109	pub max_prometheus_series: Option<u32>,
110	pub max_statements_per_transaction_trace: Option<u32>,
111	pub max_get_pages_requests_per_trace: Option<u32>,
112	pub max_transaction_name_bytes: Option<u32>,
113	pub slow_operation_threshold_ms: Option<u32>,
114	pub baseline_sample_rate: Option<f64>,
115	pub max_diagnostic_events_per_minute: Option<u32>,
116	pub diagnostic_event_queue_capacity: Option<u32>,
117}
118
119/// The process override replaces only query selection, retaining diagnostic limits.
120#[derive(Default, serde::Deserialize)]
121#[serde(default, deny_unknown_fields)]
122struct SqliteProfilingQueries {
123	queries: Vec<String>,
124	fingerprints: Vec<String>,
125}
126
127impl SqliteProfilingConfig {
128	/// Process selection overrides actor settings. An empty object clears fingerprint selection.
129	pub(crate) fn with_process_override(self) -> Self {
130		self.with_query_override(std::env::var("RIVETKIT_SQLITE_PROFILING").ok().as_deref())
131	}
132
133	fn with_query_override(mut self, process_override: Option<&str>) -> Self {
134		if let Some(value) = process_override {
135			let selection = match serde_json::from_str::<SqliteProfilingQueries>(value) {
136				Ok(selection) => selection,
137				Err(error) => {
138					tracing::warn!(%error, "invalid RIVETKIT_SQLITE_PROFILING; expected a JSON object with queries or fingerprints arrays; using profiling without fingerprint labels");
139					SqliteProfilingQueries::default()
140				}
141			};
142			self.queries = selection.queries;
143			self.fingerprints = selection.fingerprints;
144		}
145		self
146	}
147
148	fn from_input(input: SqliteProfilingConfigInput) -> Self {
149		let mut config = Self::default();
150		macro_rules! set_usize {
151			($field:ident) => {
152				if let Some(value) = input.$field {
153					config.$field = value as usize;
154				}
155			};
156		}
157		config.queries = input.queries.unwrap_or_default();
158		config.fingerprints = input.fingerprints.unwrap_or_default();
159		set_usize!(max_tracked_statement_fingerprints);
160		set_usize!(max_tracked_transaction_fingerprints);
161		set_usize!(max_prometheus_series);
162		set_usize!(max_statements_per_transaction_trace);
163		set_usize!(max_get_pages_requests_per_trace);
164		set_usize!(max_transaction_name_bytes);
165		if let Some(value) = input.slow_operation_threshold_ms {
166			config.slow_operation_threshold_ms = u64::from(value);
167		}
168		if let Some(value) = input.baseline_sample_rate {
169			config.baseline_sample_rate = value;
170		}
171		set_usize!(max_diagnostic_events_per_minute);
172		set_usize!(diagnostic_event_queue_capacity);
173		config
174	}
175}
176
177/// Per-actor trace sample rates, from 0.0 to 1.0. A rate lowers what is recorded
178/// and never raises it.
179#[derive(Clone, Debug, Default)]
180pub struct ActorTracingConfig {
181	/// `None` defers to the process sampler from `OTEL_TRACES_SAMPLER`.
182	pub sampler: Option<f64>,
183	/// Keyed by flattened action name. Also covers scheduled runs of the action.
184	pub actions: HashMap<String, f64>,
185}
186
187impl ActorTracingConfig {
188	pub(crate) fn sampler_for_action(&self, action_name: &str) -> Option<f64> {
189		self.actions.get(action_name).copied().or(self.sampler)
190	}
191
192	fn validate(&self, actions: &[ActionDefinition]) -> anyhow::Result<()> {
193		for rate in self.sampler.iter().chain(self.actions.values()) {
194			anyhow::ensure!(
195				rate.is_finite() && (0.0..=1.0).contains(rate),
196				"tracing sampler must be between 0 and 1, got {rate}"
197			);
198		}
199		// The Rust SDK never fills `ActorConfig::actions`.
200		if actions.is_empty() {
201			return Ok(());
202		}
203		for action_name in self.actions.keys() {
204			anyhow::ensure!(
205				actions.iter().any(|action| &action.name == action_name),
206				"tracing sampler names unknown action `{action_name}`"
207			);
208		}
209		Ok(())
210	}
211}
212
213#[derive(Clone, Debug)]
214pub struct ActorConfig {
215	pub name: Option<String>,
216	pub icon: Option<String>,
217	/// Whether the user declared a SQLite database for this actor (`db({...})`
218	/// on the TS side). Gates the inspector database tab.
219	pub has_database: bool,
220	pub remote_sqlite: bool,
221	pub sqlite_profiling: SqliteProfilingConfig,
222	pub tracing: Arc<ActorTracingConfig>,
223	/// Enables the experimental Actor Runtime Socket.
224	pub enable_actor_runtime_socket: bool,
225	/// Whether the user declared actor state (`state: ...` or `createState`).
226	/// Gates the inspector state tab and state-subscription messages.
227	pub has_state: bool,
228	pub can_hibernate_websocket: CanHibernateWebSocket,
229	pub state_save_interval: Duration,
230	pub create_vars_timeout: Duration,
231	pub create_conn_state_timeout: Duration,
232	pub on_before_connect_timeout: Duration,
233	pub on_connect_timeout: Duration,
234	pub on_migrate_timeout: Duration,
235	pub action_timeout: Duration,
236	pub sleep_timeout: Duration,
237	pub no_sleep: bool,
238	pub sleep_grace_period: Duration,
239	pub sleep_grace_period_overridden: bool,
240	pub connection_liveness_timeout: Duration,
241	pub connection_liveness_interval: Duration,
242	pub max_queue_size: u32,
243	pub max_schedules: u32,
244	pub max_queue_message_size: u32,
245	pub max_incoming_message_size: u32,
246	pub max_outgoing_message_size: u32,
247	pub overrides: Option<ActorConfigOverrides>,
248	pub actions: Vec<ActionDefinition>,
249	/// Author-declared inspector tab entries (custom tabs + built-in
250	/// hides). Validated upstream (Zod / builder).
251	pub inspector_tabs: Vec<InspectorTabEntry>,
252}
253
254/// 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().
255#[derive(Clone, Debug, Default)]
256pub struct ActorConfigInput {
257	pub name: Option<String>,
258	pub icon: Option<String>,
259	pub has_database: Option<bool>,
260	pub remote_sqlite: Option<bool>,
261	pub sqlite_profiling: Option<SqliteProfilingConfigInput>,
262	pub tracing: Option<ActorTracingConfig>,
263	pub enable_actor_runtime_socket: Option<bool>,
264	pub has_state: Option<bool>,
265	pub can_hibernate_websocket: Option<bool>,
266	pub state_save_interval_ms: Option<u32>,
267	pub create_vars_timeout_ms: Option<u32>,
268	pub create_conn_state_timeout_ms: Option<u32>,
269	pub on_before_connect_timeout_ms: Option<u32>,
270	pub on_connect_timeout_ms: Option<u32>,
271	pub on_migrate_timeout_ms: Option<u32>,
272	pub action_timeout_ms: Option<u32>,
273	pub sleep_timeout_ms: Option<u32>,
274	pub no_sleep: Option<bool>,
275	pub sleep_grace_period_ms: Option<u32>,
276	pub connection_liveness_timeout_ms: Option<u32>,
277	pub connection_liveness_interval_ms: Option<u32>,
278	pub max_queue_size: Option<u32>,
279	pub max_schedules: Option<u32>,
280	pub max_queue_message_size: Option<u32>,
281	pub max_incoming_message_size: Option<u32>,
282	pub max_outgoing_message_size: Option<u32>,
283	pub actions: Option<Vec<ActionDefinition>>,
284	pub inspector_tabs: Option<Vec<InspectorTabEntry>>,
285}
286
287impl ActorConfig {
288	pub fn from_input(config: ActorConfigInput) -> Self {
289		let mut actor_config = Self {
290			name: config.name,
291			icon: config.icon,
292			has_database: config.has_database.unwrap_or(false),
293			remote_sqlite: config.remote_sqlite.unwrap_or(false),
294			sqlite_profiling: config
295				.sqlite_profiling
296				.map(SqliteProfilingConfig::from_input)
297				.unwrap_or_default(),
298			tracing: Arc::new(config.tracing.unwrap_or_default()),
299			enable_actor_runtime_socket: config.enable_actor_runtime_socket.unwrap_or(false),
300			has_state: config.has_state.unwrap_or(false),
301			..Self::default()
302		};
303		if let Some(can_hibernate_websocket) = config.can_hibernate_websocket {
304			actor_config.can_hibernate_websocket =
305				CanHibernateWebSocket::Bool(can_hibernate_websocket);
306		}
307		if let Some(value) = config.state_save_interval_ms {
308			actor_config.state_save_interval = duration_ms(value);
309		}
310		if let Some(value) = config.create_vars_timeout_ms {
311			actor_config.create_vars_timeout = duration_ms(value);
312		}
313		if let Some(value) = config.create_conn_state_timeout_ms {
314			actor_config.create_conn_state_timeout = duration_ms(value);
315		}
316		if let Some(value) = config.on_before_connect_timeout_ms {
317			actor_config.on_before_connect_timeout = duration_ms(value);
318		}
319		if let Some(value) = config.on_connect_timeout_ms {
320			actor_config.on_connect_timeout = duration_ms(value);
321		}
322		if let Some(value) = config.on_migrate_timeout_ms {
323			actor_config.on_migrate_timeout = duration_ms(value);
324		}
325		if let Some(value) = config.action_timeout_ms {
326			actor_config.action_timeout = duration_ms(value);
327		}
328		if let Some(value) = config.sleep_timeout_ms {
329			actor_config.sleep_timeout = duration_ms(value);
330		}
331		if let Some(value) = config.no_sleep {
332			actor_config.no_sleep = value;
333		}
334		if let Some(value) = config.sleep_grace_period_ms {
335			actor_config.sleep_grace_period = duration_ms(value);
336			actor_config.sleep_grace_period_overridden = true;
337		}
338		if let Some(value) = config.connection_liveness_timeout_ms {
339			actor_config.connection_liveness_timeout = duration_ms(value);
340		}
341		if let Some(value) = config.connection_liveness_interval_ms {
342			actor_config.connection_liveness_interval = duration_ms(value);
343		}
344		if let Some(value) = config.max_queue_size {
345			actor_config.max_queue_size = value;
346		}
347		if let Some(value) = config.max_schedules {
348			actor_config.max_schedules = value;
349		}
350		if let Some(value) = config.max_queue_message_size {
351			actor_config.max_queue_message_size = value;
352		}
353		if let Some(value) = config.max_incoming_message_size {
354			actor_config.max_incoming_message_size = value;
355		}
356		if let Some(value) = config.max_outgoing_message_size {
357			actor_config.max_outgoing_message_size = value;
358		}
359		if let Some(actions) = config.actions {
360			actor_config.actions = actions;
361		}
362		if let Some(tabs) = config.inspector_tabs {
363			actor_config.inspector_tabs = tabs;
364		}
365
366		actor_config
367	}
368
369	pub fn effective_sleep_grace_period(&self) -> Duration {
370		cap_duration(
371			self.sleep_grace_period,
372			self.overrides
373				.as_ref()
374				.and_then(|overrides| overrides.sleep_grace_period),
375		)
376	}
377
378	/// Runtime authority for rejecting malformed config that bypassed the
379	/// TypeScript Zod layer (direct Rust builders, NAPI shim corruption,
380	/// etc.). Call this before constructing an `ActorFactory` from this
381	/// config so the actor never starts with garbage state.
382	pub fn validate(&self) -> anyhow::Result<()> {
383		crate::inspector::validate_inspector_tabs(&self.inspector_tabs)?;
384		self.tracing.validate(&self.actions)?;
385		anyhow::ensure!(
386			self.sqlite_profiling.baseline_sample_rate.is_finite()
387				&& (0.0..=1.0).contains(&self.sqlite_profiling.baseline_sample_rate),
388			"SQLite profiling baselineSampleRate must be between 0 and 1"
389		);
390		anyhow::ensure!(
391			self.sqlite_profiling.max_get_pages_requests_per_trace <= 16,
392			"SQLite profiling maxGetPagesRequestsPerTrace must be at most 16"
393		);
394		anyhow::ensure!(
395			self.sqlite_profiling.max_statements_per_transaction_trace
396				<= MAX_SQLITE_TRANSACTION_TRACE_STATEMENTS,
397			"SQLite profiling maxStatementsPerTransactionTrace must be at most 32"
398		);
399		anyhow::ensure!(
400			self.sqlite_profiling.max_transaction_name_bytes > 0,
401			"SQLite profiling maxTransactionNameBytes must be greater than zero"
402		);
403		Ok(())
404	}
405}
406
407impl Default for ActorConfig {
408	fn default() -> Self {
409		Self {
410			name: None,
411			icon: None,
412			has_database: false,
413			remote_sqlite: false,
414			sqlite_profiling: SqliteProfilingConfig::default(),
415			tracing: Arc::default(),
416			enable_actor_runtime_socket: false,
417			has_state: false,
418			can_hibernate_websocket: CanHibernateWebSocket::default(),
419			state_save_interval: DEFAULT_STATE_SAVE_INTERVAL,
420			create_vars_timeout: DEFAULT_CREATE_VARS_TIMEOUT,
421			create_conn_state_timeout: DEFAULT_CREATE_CONN_STATE_TIMEOUT,
422			on_before_connect_timeout: DEFAULT_ON_BEFORE_CONNECT_TIMEOUT,
423			on_connect_timeout: DEFAULT_ON_CONNECT_TIMEOUT,
424			on_migrate_timeout: DEFAULT_ON_MIGRATE_TIMEOUT,
425			action_timeout: DEFAULT_ACTION_TIMEOUT,
426			sleep_timeout: DEFAULT_SLEEP_TIMEOUT,
427			no_sleep: false,
428			sleep_grace_period: DEFAULT_SLEEP_GRACE_PERIOD,
429			sleep_grace_period_overridden: false,
430			connection_liveness_timeout: DEFAULT_CONNECTION_LIVENESS_TIMEOUT,
431			connection_liveness_interval: DEFAULT_CONNECTION_LIVENESS_INTERVAL,
432			max_queue_size: DEFAULT_MAX_QUEUE_SIZE,
433			max_schedules: DEFAULT_MAX_SCHEDULES,
434			max_queue_message_size: DEFAULT_MAX_QUEUE_MESSAGE_SIZE,
435			max_incoming_message_size: DEFAULT_MAX_INCOMING_MESSAGE_SIZE,
436			max_outgoing_message_size: DEFAULT_MAX_OUTGOING_MESSAGE_SIZE,
437			overrides: None,
438			actions: Vec::new(),
439			inspector_tabs: Vec::new(),
440		}
441	}
442}
443
444fn cap_duration(duration: Duration, override_duration: Option<Duration>) -> Duration {
445	if let Some(override_duration) = override_duration {
446		duration.min(override_duration)
447	} else {
448		duration
449	}
450}
451
452fn duration_ms(value: u32) -> Duration {
453	Duration::from_millis(u64::from(value))
454}
455
456// Test shim keeps moved tests in crate-root tests/ with private-module access.
457
rivetkit_core/actor/config.rs

rivetkit_core/actor/
config.rs
