Skip to content

Commit 976d08b

Browse files
committed
fix(rivetkit-napi): key actor runtime state by generation
1 parent a34b431 commit 976d08b

5 files changed

Lines changed: 435 additions & 6 deletions

File tree

‎rivetkit-rust/packages/rivetkit-core/src/actor/context.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -953,6 +953,12 @@ impl ActorContext {
953953
&self.0.actor_id
954954
}
955955

956+
/// Returns the envoy generation this context was started for, or `None` before an envoy is
957+
/// configured.
958+
pub fn generation(&self) -> Option<u32> {
959+
self.sleep_generation()
960+
}
961+
956962
pub fn name(&self) -> &str {
957963
&self.0.name
958964
}

‎rivetkit-rust/packages/rivetkit-core/src/testing.rs‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -71,9 +71,23 @@ impl ActorContextHarness {
7171
key: ActorKey,
7272
region: impl Into<String>,
7373
config: ActorConfig,
74+
) -> ActorContext {
75+
self.context_with_generation(actor_id, name, key, region, config, 1)
76+
}
77+
78+
/// Constructs a context for a specific actor generation, for tests where generations of the
79+
/// same actor coexist.
80+
pub fn context_with_generation(
81+
&self,
82+
actor_id: impl Into<String>,
83+
name: impl Into<String>,
84+
key: ActorKey,
85+
region: impl Into<String>,
86+
config: ActorConfig,
87+
generation: u32,
7488
) -> ActorContext {
7589
let actor_id = actor_id.into();
76-
let generation = Some(1);
90+
let generation = Some(generation);
7791
let sql = SqliteDb::new_with_remote_sqlite(
7892
self.handle.clone(),
7993
actor_id.clone(),

‎rivetkit-typescript/packages/rivetkit-napi/src/actor_context.rs‎

Lines changed: 15 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -43,8 +43,13 @@ type DisconnectPredicateTsfn =
4343
type RunRestartHook = Arc<dyn Fn(Option<(i64, u64)>) -> anyhow::Result<()> + Send + Sync + 'static>;
4444
pub(crate) type RegisteredTask = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
4545

46-
static ACTOR_CONTEXT_SHARED: LazyLock<SccHashMap<String, Weak<ActorContextShared>>> =
47-
LazyLock::new(SccHashMap::new);
46+
// Keyed by generation so a newer generation of the same actor never shares, resets, or clears the
47+
// JS runtime state of an older generation that is still shutting down on this host.
48+
static ACTOR_CONTEXT_SHARED: LazyLock<
49+
SccHashMap<ActorContextSharedKey, Weak<ActorContextShared>>,
50+
> = LazyLock::new(SccHashMap::new);
51+
52+
type ActorContextSharedKey = (String, Option<u32>);
4853

4954
/// N-API wrapper around `rivetkit-core::ActorContext`.
5055
#[derive(Clone)]
@@ -193,10 +198,12 @@ struct DisconnectPredicatePayload {
193198
impl ActorContext {
194199
pub(crate) fn new(inner: CoreActorContext) -> Self {
195200
let actor_id = inner.actor_id().to_owned();
196-
let shared = actor_context_shared(&actor_id);
201+
let generation = inner.generation();
202+
let shared = actor_context_shared(&actor_id, generation);
197203
tracing::debug!(
198204
class = "ActorContext",
199205
%actor_id,
206+
?generation,
200207
shared_strong_count = Arc::strong_count(&shared),
201208
"constructed napi class"
202209
);
@@ -994,14 +1001,15 @@ impl Drop for ActorContextShared {
9941001
}
9951002
}
9961003

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

1000-
match ACTOR_CONTEXT_SHARED.entry_sync(actor_id.to_owned()) {
1007+
match ACTOR_CONTEXT_SHARED.entry_sync((actor_id.to_owned(), generation)) {
10011008
scc::hash_map::Entry::Occupied(mut entry) => {
10021009
if let Some(shared) = entry.get().upgrade() {
10031010
tracing::debug!(
10041011
%actor_id,
1012+
?generation,
10051013
outcome = "hit",
10061014
strong_count = Arc::strong_count(&shared),
10071015
"actor context shared-state cache lookup"
@@ -1013,6 +1021,7 @@ fn actor_context_shared(actor_id: &str) -> Arc<ActorContextShared> {
10131021
*entry.get_mut() = Arc::downgrade(&shared);
10141022
tracing::debug!(
10151023
%actor_id,
1024+
?generation,
10161025
outcome = "stale",
10171026
"actor context shared-state cache lookup"
10181027
);
@@ -1023,6 +1032,7 @@ fn actor_context_shared(actor_id: &str) -> Arc<ActorContextShared> {
10231032
entry.insert_entry(Arc::downgrade(&shared));
10241033
tracing::debug!(
10251034
%actor_id,
1035+
?generation,
10261036
outcome = "miss",
10271037
"actor context shared-state cache lookup"
10281038
);

‎rivetkit-typescript/packages/rivetkit-napi/tests/actor_context.rs‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,4 +18,32 @@ mod moved_tests {
1818
assert!(!shared.has_end_reason());
1919
assert!(ctx.started());
2020
}
21+
22+
#[test]
23+
fn generations_of_the_same_actor_do_not_share_runtime_state() {
24+
let harness = rivetkit_core::testing::ActorContextHarness::new();
25+
let context_for = |generation| {
26+
ActorContext::new(harness.context_with_generation(
27+
"actor-generations",
28+
"actor",
29+
Vec::new(),
30+
"local",
31+
rivetkit_core::ActorConfig::default(),
32+
generation,
33+
))
34+
};
35+
36+
let old = context_for(1);
37+
let old_again = context_for(1);
38+
let new = context_for(2);
39+
40+
assert!(Arc::ptr_eq(&old.shared, &old_again.shared));
41+
assert!(!Arc::ptr_eq(&old.shared, &new.shared));
42+
43+
// An older generation that finishes shutting down after a newer generation started must
44+
// not reset the newer generation's runtime state.
45+
new.shared.set_end_reason(EndReason::Sleep);
46+
old.shared.reset_runtime_state();
47+
assert!(new.shared.has_end_reason());
48+
}
2149
}

0 commit comments

Comments
 (0)