Skip to content

feat(engine-process): support caller-defined engine instance paths #5511

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
NathanFlurry wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(engine-process): support caller-defined engine instance paths #5511

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions engine/packages/cli/src/commands/dev.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -510,6 +510,7 @@ mod tests {
endpoint: DEFAULT_ENGINE_ENDPOINT.to_string(),
bind_host: bind_host.to_string(),
public_url: public_url.map(str::to_string),
instance_path: None,
}
}

Expand Down
86 changes: 64 additions & 22 deletions rivetkit-rust/packages/engine-process/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,11 @@
pub struct EngineResolverConfig {
pub endpoint: String,
pub explicit_binary_path: Option<PathBuf>,
/// Directory for mutable state owned by this managed engine instance.
///
/// Embedders can set this to isolate multiple managed engines. When unset,
/// the legacy shared engine paths are used.
pub instance_path: Option<PathBuf>,
pub bind_host: Option<String>,
pub bind_port: Option<u16>,
pub public_url: Option<String>,
Expand All @@ -50,6 +55,7 @@
Self {
endpoint: endpoint.to_owned(),
explicit_binary_path,
instance_path: None,
bind_host,
bind_port,
public_url: None,
Expand Down Expand Up @@ -77,6 +83,8 @@
pub endpoint: String,
pub bind_host: String,
pub public_url: Option<String>,
#[serde(default)]
pub instance_path: Option<PathBuf>,
}

impl EngineRuntimeStamp {
Expand Down Expand Up @@ -155,7 +163,7 @@
return Ok(Self {
watcher: None,
startup: EngineStartup::Reused {
stamp: read_engine_stamp(),
stamp: read_engine_stamp(config),
},
});
}
Expand All @@ -174,7 +182,7 @@
return Ok(Self {
watcher: None,
startup: EngineStartup::Reused {
stamp: read_engine_stamp(),
stamp: read_engine_stamp(config),
},
});
}
Expand All @@ -188,11 +196,8 @@

let env = engine_env(config)?;
let config_path = write_engine_config(config)?;
let db_path = engine_db_path()?;
let logs_dir = storage_root()?
.join("var")
.join("logs")
.join("rivet-engine");
let db_path = engine_db_path(config)?;
let logs_dir = engine_logs_dir(config)?;
ensure_dir(&db_path).context("create engine db directory")?;
ensure_dir(&logs_dir).context("create engine logs directory")?;

Expand Down Expand Up @@ -284,8 +289,9 @@
endpoint: endpoint.clone(),
bind_host: resolve_bind_host(config)?,
public_url: config.public_url.clone(),
instance_path: config.instance_path.clone(),
};
if let Err(error) = write_engine_stamp(&stamp) {
if let Err(error) = write_engine_stamp(config, &stamp) {
tracing::warn!(?error, "failed to write engine runtime stamp");
}

Expand All @@ -296,16 +302,35 @@
}
}

/// Path to the rivet-engine database directory under the shared storage root.
pub fn engine_db_path() -> Result<PathBuf> {
Ok(storage_root()?.join("var").join("engine").join("db"))
/// Path to the rivet-engine database directory.
pub fn engine_db_path(config: &EngineResolverConfig) -> Result<PathBuf> {
Ok(engine_instance_path(config)?.join("db"))
}

fn engine_stamp_path() -> Result<PathBuf> {
Ok(storage_root()?
.join("var")
.join("engine")
.join("runtime.json"))
/// Default directory for mutable state owned by a managed engine instance.
pub fn default_engine_instance_path() -> Result<PathBuf> {
Ok(storage_root()?.join("var").join("engine"))
}

fn engine_instance_path(config: &EngineResolverConfig) -> Result<PathBuf> {
match &config.instance_path {
Some(path) => Ok(path.clone()),
None => default_engine_instance_path(),
}
}

fn engine_stamp_path(config: &EngineResolverConfig) -> Result<PathBuf> {
Ok(engine_instance_path(config)?.join("runtime.json"))
}

fn engine_logs_dir(config: &EngineResolverConfig) -> Result<PathBuf> {
match &config.instance_path {
Some(path) => Ok(path.join("logs")),
None => Ok(storage_root()?
.join("var")
.join("logs")
.join("rivet-engine")),
}
}

/// The address the engine actually binds: the explicit `bind_host` override, or
Expand All @@ -323,8 +348,8 @@
.to_owned())
}

fn write_engine_stamp(stamp: &EngineRuntimeStamp) -> Result<()> {
let path = engine_stamp_path()?;
fn write_engine_stamp(config: &EngineResolverConfig, stamp: &EngineRuntimeStamp) -> Result<()> {
let path = engine_stamp_path(config)?;
if let Some(dir) = path.parent() {
ensure_dir(dir)?;
}
Expand All @@ -339,8 +364,8 @@
/// callers treat the binding as unknown rather than trusting old data. On
/// platforms without a liveness check (see `pid_is_alive`) the stamp is always
/// treated as unknown for the same reason.
fn read_engine_stamp() -> Option<EngineRuntimeStamp> {
let path = engine_stamp_path().ok()?;
fn read_engine_stamp(config: &EngineResolverConfig) -> Option<EngineRuntimeStamp> {
let path = engine_stamp_path(config).ok()?;
let contents = std::fs::read(&path).ok()?;
let stamp: EngineRuntimeStamp = serde_json::from_slice(&contents).ok()?;
pid_is_alive(stamp.pid).then_some(stamp)
Expand Down Expand Up @@ -384,7 +409,7 @@
.checked_add(10)
.ok_or_else(|| invalid_endpoint(endpoint, "port is too large"))?;

let db_path = engine_db_path()?;
let db_path = engine_db_path(config)?;

Ok(vec![
("RIVET__GUARD__HOST".to_owned(), guard_host.clone()),
Expand All @@ -411,7 +436,7 @@
let public_url = Url::parse(public_url)
.with_context(|| format!("parse engine public URL `{public_url}`"))?;
let peer_url = peer_url_for_public_url(&public_url)?;
let dir = storage_root()?.join("var").join("engine");
let dir = engine_instance_path(config)?;
ensure_dir(&dir)?;
let path = dir.join("config.json");
let config = serde_json::json!({
Expand Down Expand Up @@ -950,6 +975,7 @@
EngineResolverConfig {
endpoint: "http://127.0.0.1:1".to_owned(),
explicit_binary_path: None,
instance_path: None,
bind_host: None,
bind_port: None,
public_url: None,
Expand All @@ -965,9 +991,25 @@
endpoint: "http://127.0.0.1:6420".to_owned(),
bind_host: bind_host.to_owned(),
public_url: None,
instance_path: None,
}
}

#[test]
fn caller_defined_instance_path_scopes_mutable_state() {
let temp = tempfile::tempdir().expect("create temp dir");
let instance_path = temp.path().join("engine-6421");
let mut config = test_config(String::new(), false);
config.instance_path = Some(instance_path.clone());

assert_eq!(engine_db_path(&config).unwrap(), instance_path.join("db"));
assert_eq!(
engine_stamp_path(&config).unwrap(),

Check warning on line 1007 in rivetkit-rust/packages/engine-process/src/lib.rs

View workflow job for this annotation

GitHub Actions / Rustfmt 

Diff in /home/runner/work/rivet/rivet/rivetkit-rust/packages/engine-process/src/lib.rs
instance_path.join("runtime.json")
);
assert_eq!(engine_logs_dir(&config).unwrap(), instance_path.join("logs"));
}

#[test]
fn binds_loopback_only_detects_loopback_addresses() {
assert!(stamp_with_bind("127.0.0.1").binds_loopback_only());
Expand Down
Loading