Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
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
6 changes: 6 additions & 0 deletions rivetkit-rust/packages/rivetkit-core/src/actor/context.rs
Original file line number Diff line number Diff line change
Expand Up @@ -953,6 +953,12 @@ impl ActorContext {
&self.0.actor_id
}

/// Returns the envoy generation this context was started for, or `None` before an envoy is
/// configured.
pub fn generation(&self) -> Option<u32> {
self.sleep_generation()
}

pub fn name(&self) -> &str {
&self.0.name
}
Expand Down
16 changes: 15 additions & 1 deletion rivetkit-rust/packages/rivetkit-core/src/testing.rs
Original file line number Diff line number Diff line change
Expand Up @@ -71,9 +71,23 @@ impl ActorContextHarness {
key: ActorKey,
region: impl Into<String>,
config: ActorConfig,
) -> ActorContext {
self.context_with_generation(actor_id, name, key, region, config, 1)
}

/// Constructs a context for a specific actor generation, for tests where generations of the
/// same actor coexist.
pub fn context_with_generation(
&self,
actor_id: impl Into<String>,
name: impl Into<String>,
key: ActorKey,
region: impl Into<String>,
config: ActorConfig,
generation: u32,
) -> ActorContext {
let actor_id = actor_id.into();
let generation = Some(1);
let generation = Some(generation);
let sql = SqliteDb::new_with_remote_sqlite(
self.handle.clone(),
actor_id.clone(),
Expand Down
20 changes: 15 additions & 5 deletions rivetkit-typescript/packages/rivetkit-napi/src/actor_context.rs
Original file line number Diff line number Diff line change
Expand Up @@ -42,9 +42,14 @@
ThreadsafeFunction<DisconnectPredicatePayload, ErrorStrategy::CalleeHandled>;
type RunRestartHook = Arc<dyn Fn(Option<(i64, u64)>) -> anyhow::Result<()> + Send + Sync + 'static>;
pub(crate) type RegisteredTask = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;

Check warning on line 45 in rivetkit-typescript/packages/rivetkit-napi/src/actor_context.rs

View workflow job for this annotation

GitHub Actions / Rustfmt

Diff in /home/runner/work/rivet/rivet/rivetkit-typescript/packages/rivetkit-napi/src/actor_context.rs
static ACTOR_CONTEXT_SHARED: LazyLock<SccHashMap<String, Weak<ActorContextShared>>> =
LazyLock::new(SccHashMap::new);
// Keyed by generation so a newer generation of the same actor never shares, resets, or clears the
// JS runtime state of an older generation that is still shutting down on this host.
Comment on lines 45 to +47

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 Medium · Format the shared-state declaration so Rustfmt passes

The required Rustfmt check fails on this declaration at this head. Apply the formatter’s single-line generic layout (with the initializer on the next line) so the Rust CI gate can pass.

static ACTOR_CONTEXT_SHARED: LazyLock<
SccHashMap<ActorContextSharedKey, Weak<ActorContextShared>>,
> = LazyLock::new(SccHashMap::new);

type ActorContextSharedKey = (String, Option<u32>);

/// N-API wrapper around `rivetkit-core::ActorContext`.
#[derive(Clone)]
Expand Down Expand Up @@ -193,10 +198,12 @@
impl ActorContext {
pub(crate) fn new(inner: CoreActorContext) -> Self {
let actor_id = inner.actor_id().to_owned();
let shared = actor_context_shared(&actor_id);
let generation = inner.generation();
let shared = actor_context_shared(&actor_id, generation);
tracing::debug!(
class = "ActorContext",
%actor_id,
?generation,
shared_strong_count = Arc::strong_count(&shared),
"constructed napi class"
);
Expand Down Expand Up @@ -994,14 +1001,15 @@
}
}

fn actor_context_shared(actor_id: &str) -> Arc<ActorContextShared> {
fn actor_context_shared(actor_id: &str, generation: Option<u32>) -> Arc<ActorContextShared> {
ACTOR_CONTEXT_SHARED.retain_sync(|_, shared| shared.strong_count() > 0);

match ACTOR_CONTEXT_SHARED.entry_sync(actor_id.to_owned()) {
match ACTOR_CONTEXT_SHARED.entry_sync((actor_id.to_owned(), generation)) {
scc::hash_map::Entry::Occupied(mut entry) => {
if let Some(shared) = entry.get().upgrade() {
tracing::debug!(
%actor_id,
?generation,
outcome = "hit",
strong_count = Arc::strong_count(&shared),
"actor context shared-state cache lookup"
Expand All @@ -1013,6 +1021,7 @@
*entry.get_mut() = Arc::downgrade(&shared);
tracing::debug!(
%actor_id,
?generation,
outcome = "stale",
"actor context shared-state cache lookup"
);
Expand All @@ -1023,6 +1032,7 @@
entry.insert_entry(Arc::downgrade(&shared));
tracing::debug!(
%actor_id,
?generation,
outcome = "miss",
"actor context shared-state cache lookup"
);
Expand Down
28 changes: 28 additions & 0 deletions rivetkit-typescript/packages/rivetkit-napi/tests/actor_context.rs
Original file line number Diff line number Diff line change
Expand Up @@ -18,4 +18,32 @@ mod moved_tests {
assert!(!shared.has_end_reason());
assert!(ctx.started());
}

#[test]
fn generations_of_the_same_actor_do_not_share_runtime_state() {
let harness = rivetkit_core::testing::ActorContextHarness::new();
let context_for = |generation| {
ActorContext::new(harness.context_with_generation(
"actor-generations",
"actor",
Vec::new(),
"local",
rivetkit_core::ActorConfig::default(),
generation,
))
};

let old = context_for(1);
let old_again = context_for(1);
let new = context_for(2);

assert!(Arc::ptr_eq(&old.shared, &old_again.shared));
assert!(!Arc::ptr_eq(&old.shared, &new.shared));

// An older generation that finishes shutting down after a newer generation started must
// not reset the newer generation's runtime state.
new.shared.set_end_reason(EndReason::Sleep);
old.shared.reset_runtime_state();
assert!(new.shared.has_end_reason());
}
}
Loading
Loading