rivetkit_core/actor/
config.rs

1use std::fmt;
2use std::sync::Arc;
3use std::time::Duration;
4
5use rivet_envoy_client::config::HttpRequest;
6
7use crate::inspector::InspectorTabEntry;
8
9const DEFAULT_STATE_SAVE_INTERVAL: Duration = Duration::from_secs(1);
10const DEFAULT_CREATE_VARS_TIMEOUT: Duration = Duration::from_secs(5);
11const DEFAULT_CREATE_CONN_STATE_TIMEOUT: Duration = Duration::from_secs(5);
12const DEFAULT_ON_BEFORE_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
13const DEFAULT_ON_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
14const DEFAULT_ON_MIGRATE_TIMEOUT: Duration = Duration::from_secs(30);
15const DEFAULT_ACTION_TIMEOUT: Duration = Duration::from_secs(60);
16const DEFAULT_SLEEP_TIMEOUT: Duration = Duration::from_secs(30);
17const DEFAULT_SLEEP_GRACE_PERIOD: Duration = Duration::from_secs(15);
18const DEFAULT_CONNECTION_LIVENESS_TIMEOUT: Duration = Duration::from_millis(2500);
19const DEFAULT_CONNECTION_LIVENESS_INTERVAL: Duration = Duration::from_secs(5);
20const DEFAULT_MAX_QUEUE_SIZE: u32 = 1000;
21const DEFAULT_MAX_QUEUE_MESSAGE_SIZE: u32 = 65_536;
22const DEFAULT_MAX_INCOMING_MESSAGE_SIZE: u32 = 65_536;
23const DEFAULT_MAX_OUTGOING_MESSAGE_SIZE: u32 = 1_048_576;
24
25#[derive(Clone)]
26pub enum CanHibernateWebSocket {
27	Bool(bool),
28	Callback(Arc<dyn Fn(&HttpRequest) -> bool + Send + Sync>),
29}
30
31impl fmt::Debug for CanHibernateWebSocket {
32	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
33		match self {
34			Self::Bool(value) => f.debug_tuple("Bool").field(value).finish(),
35			Self::Callback(_) => f.write_str("Callback(..)"),
36		}
37	}
38}
39
40impl Default for CanHibernateWebSocket {
41	fn default() -> Self {
42		Self::Bool(false)
43	}
44}
45
46#[derive(Clone, Debug, Default)]
47pub struct ActorConfigOverrides {
48	pub sleep_grace_period: Option<Duration>,
49}
50
51#[derive(Clone, Debug)]
52pub struct ActionDefinition {
53	pub name: String,
54}
55
56#[derive(Clone, Debug)]
57pub struct ActorConfig {
58	pub name: Option<String>,
59	pub icon: Option<String>,
60	/// Whether the user declared a SQLite database for this actor (`db({...})`
61	/// on the TS side). Gates the inspector database tab.
62	pub has_database: bool,
63	pub remote_sqlite: bool,
64	/// Whether the user declared actor state (`state: ...` or `createState`).
65	/// Gates the inspector state tab and state-subscription messages.
66	pub has_state: bool,
67	pub can_hibernate_websocket: CanHibernateWebSocket,
68	pub state_save_interval: Duration,
69	pub create_vars_timeout: Duration,
70	pub create_conn_state_timeout: Duration,
71	pub on_before_connect_timeout: Duration,
72	pub on_connect_timeout: Duration,
73	pub on_migrate_timeout: Duration,
74	pub action_timeout: Duration,
75	pub sleep_timeout: Duration,
76	pub no_sleep: bool,
77	pub sleep_grace_period: Duration,
78	pub sleep_grace_period_overridden: bool,
79	pub connection_liveness_timeout: Duration,
80	pub connection_liveness_interval: Duration,
81	pub max_queue_size: u32,
82	pub max_queue_message_size: u32,
83	pub max_incoming_message_size: u32,
84	pub max_outgoing_message_size: u32,
85	pub preload_max_workflow_bytes: Option<u64>,
86	pub preload_max_connections_bytes: Option<u64>,
87	pub overrides: Option<ActorConfigOverrides>,
88	pub actions: Vec<ActionDefinition>,
89	/// Author-declared inspector tab entries (custom tabs + built-in
90	/// hides). Validated upstream (Zod / builder).
91	pub inspector_tabs: Vec<InspectorTabEntry>,
92}
93
94/// 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().
95#[derive(Clone, Debug, Default)]
96pub struct ActorConfigInput {
97	pub name: Option<String>,
98	pub icon: Option<String>,
99	pub has_database: Option<bool>,
100	pub remote_sqlite: Option<bool>,
101	pub has_state: Option<bool>,
102	pub can_hibernate_websocket: Option<bool>,
103	pub state_save_interval_ms: Option<u32>,
104	pub create_vars_timeout_ms: Option<u32>,
105	pub create_conn_state_timeout_ms: Option<u32>,
106	pub on_before_connect_timeout_ms: Option<u32>,
107	pub on_connect_timeout_ms: Option<u32>,
108	pub on_migrate_timeout_ms: Option<u32>,
109	pub action_timeout_ms: Option<u32>,
110	pub sleep_timeout_ms: Option<u32>,
111	pub no_sleep: Option<bool>,
112	pub sleep_grace_period_ms: Option<u32>,
113	pub connection_liveness_timeout_ms: Option<u32>,
114	pub connection_liveness_interval_ms: Option<u32>,
115	pub max_queue_size: Option<u32>,
116	pub max_queue_message_size: Option<u32>,
117	pub max_incoming_message_size: Option<u32>,
118	pub max_outgoing_message_size: Option<u32>,
119	pub preload_max_workflow_bytes: Option<f64>,
120	pub preload_max_connections_bytes: Option<f64>,
121	pub actions: Option<Vec<ActionDefinition>>,
122	pub inspector_tabs: Option<Vec<InspectorTabEntry>>,
123}
124
125impl ActorConfig {
126	pub fn from_input(config: ActorConfigInput) -> Self {
127		let mut actor_config = Self {
128			name: config.name,
129			icon: config.icon,
130			has_database: config.has_database.unwrap_or(false),
131			remote_sqlite: config.remote_sqlite.unwrap_or(false),
132			has_state: config.has_state.unwrap_or(false),
133			..Self::default()
134		};
135		if let Some(can_hibernate_websocket) = config.can_hibernate_websocket {
136			actor_config.can_hibernate_websocket =
137				CanHibernateWebSocket::Bool(can_hibernate_websocket);
138		}
139		if let Some(value) = config.state_save_interval_ms {
140			actor_config.state_save_interval = duration_ms(value);
141		}
142		if let Some(value) = config.create_vars_timeout_ms {
143			actor_config.create_vars_timeout = duration_ms(value);
144		}
145		if let Some(value) = config.create_conn_state_timeout_ms {
146			actor_config.create_conn_state_timeout = duration_ms(value);
147		}
148		if let Some(value) = config.on_before_connect_timeout_ms {
149			actor_config.on_before_connect_timeout = duration_ms(value);
150		}
151		if let Some(value) = config.on_connect_timeout_ms {
152			actor_config.on_connect_timeout = duration_ms(value);
153		}
154		if let Some(value) = config.on_migrate_timeout_ms {
155			actor_config.on_migrate_timeout = duration_ms(value);
156		}
157		if let Some(value) = config.action_timeout_ms {
158			actor_config.action_timeout = duration_ms(value);
159		}
160		if let Some(value) = config.sleep_timeout_ms {
161			actor_config.sleep_timeout = duration_ms(value);
162		}
163		if let Some(value) = config.no_sleep {
164			actor_config.no_sleep = value;
165		}
166		if let Some(value) = config.sleep_grace_period_ms {
167			actor_config.sleep_grace_period = duration_ms(value);
168			actor_config.sleep_grace_period_overridden = true;
169		}
170		if let Some(value) = config.connection_liveness_timeout_ms {
171			actor_config.connection_liveness_timeout = duration_ms(value);
172		}
173		if let Some(value) = config.connection_liveness_interval_ms {
174			actor_config.connection_liveness_interval = duration_ms(value);
175		}
176		if let Some(value) = config.max_queue_size {
177			actor_config.max_queue_size = value;
178		}
179		if let Some(value) = config.max_queue_message_size {
180			actor_config.max_queue_message_size = value;
181		}
182		if let Some(value) = config.max_incoming_message_size {
183			actor_config.max_incoming_message_size = value;
184		}
185		if let Some(value) = config.max_outgoing_message_size {
186			actor_config.max_outgoing_message_size = value;
187		}
188		actor_config.preload_max_workflow_bytes =
189			config.preload_max_workflow_bytes.map(|value| value as u64);
190		actor_config.preload_max_connections_bytes = config
191			.preload_max_connections_bytes
192			.map(|value| value as u64);
193		if let Some(actions) = config.actions {
194			actor_config.actions = actions;
195		}
196		if let Some(tabs) = config.inspector_tabs {
197			actor_config.inspector_tabs = tabs;
198		}
199
200		actor_config
201	}
202
203	pub fn effective_sleep_grace_period(&self) -> Duration {
204		cap_duration(
205			self.sleep_grace_period,
206			self.overrides
207				.as_ref()
208				.and_then(|overrides| overrides.sleep_grace_period),
209		)
210	}
211
212	/// Runtime authority for rejecting malformed config that bypassed the
213	/// TypeScript Zod layer (direct Rust builders, NAPI shim corruption,
214	/// etc.). Call this before constructing an `ActorFactory` from this
215	/// config so the actor never starts with garbage state.
216	pub fn validate(&self) -> anyhow::Result<()> {
217		crate::inspector::validate_inspector_tabs(&self.inspector_tabs)?;
218		Ok(())
219	}
220}
221
222impl Default for ActorConfig {
223	fn default() -> Self {
224		Self {
225			name: None,
226			icon: None,
227			has_database: false,
228			remote_sqlite: false,
229			has_state: false,
230			can_hibernate_websocket: CanHibernateWebSocket::default(),
231			state_save_interval: DEFAULT_STATE_SAVE_INTERVAL,
232			create_vars_timeout: DEFAULT_CREATE_VARS_TIMEOUT,
233			create_conn_state_timeout: DEFAULT_CREATE_CONN_STATE_TIMEOUT,
234			on_before_connect_timeout: DEFAULT_ON_BEFORE_CONNECT_TIMEOUT,
235			on_connect_timeout: DEFAULT_ON_CONNECT_TIMEOUT,
236			on_migrate_timeout: DEFAULT_ON_MIGRATE_TIMEOUT,
237			action_timeout: DEFAULT_ACTION_TIMEOUT,
238			sleep_timeout: DEFAULT_SLEEP_TIMEOUT,
239			no_sleep: false,
240			sleep_grace_period: DEFAULT_SLEEP_GRACE_PERIOD,
241			sleep_grace_period_overridden: false,
242			connection_liveness_timeout: DEFAULT_CONNECTION_LIVENESS_TIMEOUT,
243			connection_liveness_interval: DEFAULT_CONNECTION_LIVENESS_INTERVAL,
244			max_queue_size: DEFAULT_MAX_QUEUE_SIZE,
245			max_queue_message_size: DEFAULT_MAX_QUEUE_MESSAGE_SIZE,
246			max_incoming_message_size: DEFAULT_MAX_INCOMING_MESSAGE_SIZE,
247			max_outgoing_message_size: DEFAULT_MAX_OUTGOING_MESSAGE_SIZE,
248			preload_max_workflow_bytes: None,
249			preload_max_connections_bytes: None,
250			overrides: None,
251			actions: Vec::new(),
252			inspector_tabs: Vec::new(),
253		}
254	}
255}
256
257fn cap_duration(duration: Duration, override_duration: Option<Duration>) -> Duration {
258	if let Some(override_duration) = override_duration {
259		duration.min(override_duration)
260	} else {
261		duration
262	}
263}
264
265fn duration_ms(value: u32) -> Duration {
266	Duration::from_millis(u64::from(value))
267}
268
269// Test shim keeps moved tests in crate-root tests/ with private-module access.
270#[cfg(test)]
271#[path = "../../tests/config.rs"]
272mod tests;
rivetkit_core/actor/config.rs

rivetkit_core/actor/
config.rs
